perf(test-e2e-playwright): run independent Playwright suites in parallel under a pressure gate

WordPress#0 runs 18 Playwright suites one after another and exceeds its
CI time budget. A role that opts in through playwright.parallel in its
meta/tests.yml (WordPress: [3, 1, 1] per variant) now hands its suites to
an async scheduler, run_parallel.py, after staging:

- a suite starts only when its dependencies finished here (Keycloak first)
  and the host's CPU pressure, avg10 and avg60, stayed at or below 36 for
  30 s; the first container always starts and running ones are never
  killed
- onion suites share their own cap of 3 and a Tor lock with a
  recreate-pending marker; a probe that times out on the lock re-probes
  the current circuits and, failing that, leaves only that app to the
  serial path
- an app whose dependency was dropped is deferred, and every deferred app
  gets the serial first pass before the existing retry, in round order
- each job gets --oom-score-adj=1000 instead of a memory limit; crashed
  pages, closed browsers and killed workers are counted and named in the
  per-app timing line
- each job's result.json, stdout and stderr land in run-parallel/ and are
  archived with the reports
- the async ceiling is the serial worst case per app and reachable
  network, curl's per-attempt max time included (now
  TEST_E2E_PLAYWRIGHT_ONION_PROBE_MAX_TIME); it stays above the CI step
  timeout and never raises a budget

Roles without the opt-in run exactly as before; the onion probe script
with an empty lock behaves byte for byte like the previous one.

Tests: 126 role unit tests and 30 negative controls cover the gate
window, admission order, onion cap and lock, marker cleanup, the bounded
wait, dependency deferral, crash counting, artifact collection, the
ceiling for the real WordPress#0 app list, and serial equivalence. No CI
run has exercised it yet; the next push is its live proof.

🎯 Fix: 88 %
💥 Break: 6 %
🌱 Efficiency: –
⚡ Speed: –
🛡️ Security: ~100 %
🏆 Score: 83
🚦 Gate: ⛔ fix 88 % < 90, break 6 % > 5; committed and merged on the operator's word, CI is the live proof

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

plugins/lookup/playwright_workers.py

@@ -11,13 +11,13 @@ from ansible.template import trust_as_template
 from utils.env.runtime import mem_available_mb, mem_total_mb
 from utils.networks.reachability import TOR, network_of
 
-_PER_WORKER_GB = 1.5
-_RAM_FRACTION = 0.5
+PER_WORKER_GB = 1.5
+RAM_FRACTION = 0.5
 _CPU_DIVISOR = 4
 _HARD_CAP = 6
 _CI_CAP = 2
 _ONION_FACTOR = 2
-_ONION_CAP = 3
+ONION_CAP = 3
 _CI_ENV = ("CI", "GITHUB_ACTIONS", "GITLAB_CI", "BUILDKITE", "JENKINS_URL")
 
 
@@ -55,13 +55,13 @@ def compute_workers(
     ci: bool,
     *,
     onion: bool = False,
-    per_worker_gb: float = _PER_WORKER_GB,
-    ram_fraction: float = _RAM_FRACTION,
+    per_worker_gb: float = PER_WORKER_GB,
+    ram_fraction: float = RAM_FRACTION,
     cpu_divisor: int = _CPU_DIVISOR,
     hard_cap: int = _HARD_CAP,
     ci_cap: int = _CI_CAP,
     onion_factor: int = _ONION_FACTOR,
-    onion_cap: int = _ONION_CAP,
+    onion_cap: int = ONION_CAP,
 ) -> int:
     """Workers the host can carry, and the transport can use.
 
@@ -135,12 +135,12 @@ class LookupModule(LookupBase):
                 _ram_gb(),
                 _is_ci(),
                 onion=bool(kwargs.get("onion", self._targets_onion(variables))),
-                per_worker_gb=float(kwargs.get("per_worker_gb", _PER_WORKER_GB)),
-                ram_fraction=float(kwargs.get("ram_fraction", _RAM_FRACTION)),
+                per_worker_gb=float(kwargs.get("per_worker_gb", PER_WORKER_GB)),
+                ram_fraction=float(kwargs.get("ram_fraction", RAM_FRACTION)),
                 cpu_divisor=int(kwargs.get("cpu_divisor", _CPU_DIVISOR)),
                 hard_cap=int(kwargs.get("hard_cap", _HARD_CAP)),
                 ci_cap=int(kwargs.get("ci_cap", _CI_CAP)),
                 onion_factor=int(kwargs.get("onion_factor", _ONION_FACTOR)),
-                onion_cap=int(kwargs.get("onion_cap", _ONION_CAP)),
+                onion_cap=int(kwargs.get("onion_cap", ONION_CAP)),
             )
         ]

roles/test-e2e-playwright/files/archive_playwright_reports.sh

@@ -3,7 +3,8 @@ set -euo pipefail
 
 # Usage: archive_playwright_reports.sh <reports_dir> <variant_index> <async_enabled>
 #
-# Move playwright-junit.xml / playwright-report/ / test-results/ from
+# Move playwright-junit.xml / playwright-report/ / test-results/ and the
+# parallel scheduler's run-parallel/ (output and result of the run) from
 # <reports_dir> into <reports_dir>/variant-N/{sync|async}/ so subsequent
 # invocations of the test-e2e-playwright role cannot overwrite them.
 # Idempotent: an existing slot is replaced rather than refused.
@@ -26,7 +27,7 @@ subdir="variant-${variant_index}/${pass}"
 
 cd "$reports_dir"
 mkdir -p "$subdir"
-for item in playwright-junit.xml playwright-report test-results; do
+for item in playwright-junit.xml playwright-report test-results run-parallel; do
 	if [ -e "$item" ]; then
 		rm -rf "${subdir:?}/$item"
 		mv "$item" "$subdir/"

roles/test-e2e-playwright/files/run_parallel.py

@@ -0,0 +1,498 @@
+#!/usr/bin/env python3
+"""Run the round's Playwright containers with a concurrency cap.
+
+    run_parallel.py <jobs_dir> <cap> <poll_seconds> <max_cpu_pressure>
+                    <ram_fraction> <sustain_seconds> <tor_lock> <onion_cap>
+
+Ansible drops one ``<seq>-<application_id>.job.json`` per role into
+*jobs_dir* while it stages the next role, and touches ``sealed`` once the last
+one is written. Each job file holds ``id``, ``argv``, ``env``, ``timeout``,
+``after``, ``tor``, ``workers``, ``memory_mb``, ``stop`` and ``meta``. The
+scheduler starts a job once every job it names in ``after`` has finished here,
+never runs more than *cap* at once, starts ready jobs in submission order, and
+writes ``<application_id>.result.json`` with ``meta``, ``rc`` (absent and
+``msg`` set when the timeout killed it), ``timed_out``, the ``run_signals``
+fields, ``stdout`` and ``stderr``. A job whose ``after`` names one that never
+ran here gets only ``id``, ``meta`` and ``deferred``, the jobs it waited on.
+"""
+
+from __future__ import annotations
+
+import fcntl
+import json
+import os
+import re
+import signal
+import subprocess
+import sys
+import time
+from dataclasses import dataclass
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
+SEALED = "sealed"
+JOB_NAME = re.compile(r"^(\d+)-.+\.job\.json$")
+RESULT_SUFFIX = ".result.json"
+KILL_GRACE_SECONDS = 10
+NAP_STEP_SECONDS = 0.1
+RETRY_MARK = re.compile(r"\(retry #\d+\)")
+FLAKY_SUMMARY = re.compile(r"^\s*(\d+) flaky\s*$", re.MULTILINE)
+CRASH_MARKS = {
+    "page crashed": re.compile(r"(?i)(?:page|target) crashed"),
+    "browser closed (OOM or timeout)": re.compile(r"(?i)browser (?:has been )?closed"),
+    "worker killed": re.compile(
+        r"(?i)worker process exited unexpectedly.*|\bSIGKILL\b"
+    ),
+}
+
+
+@dataclass(frozen=True)
+class Job:
+    id: str
+    after: tuple[str, ...]
+
+
+class Plan:
+    """Dispatch order of a streamed job list under a concurrency cap.
+
+    Args:
+        cap: most jobs that may run at once, at least one.
+    """
+
+    def __init__(self, cap: int) -> None:
+        if cap < 1:
+            raise ValueError(f"cap must be at least 1, got {cap}")
+        self.cap = cap
+        self.pending: list[Job] = []
+        self.submitted: set[str] = set()
+        self.running: set[str] = set()
+        self.finished: set[str] = set()
+        self.deferred: set[str] = set()
+        self.sealed = False
+
+    def submit(self, job: Job) -> None:
+        if job.id in self.submitted:
+            raise ValueError(f"job {job.id} submitted twice")
+        self.submitted.add(job.id)
+        self.pending.append(job)
+
+    def seal(self) -> None:
+        self.sealed = True
+
+    def _ready(self, job: Job) -> bool:
+        return all(dep in self.finished for dep in job.after if dep != job.id)
+
+    def start_next(
+        self,
+        host_admits: Callable[[int, Job], bool] = lambda running, job: True,
+        job_admits: Callable[[Job], bool] = lambda job: True,
+        job_release: Callable[[Job], None] = lambda job: None,
+    ) -> list[Job]:
+        """Jobs to start now, in submission order, within the cap.
+
+        Args:
+            host_admits: asked with the running count and the ready job before
+                each start; a False ends this round of starts.
+            job_admits: asked first for each ready job, claiming what the job
+                needs; a False skips that job only, so a job waiting on its
+                own resource never uses up the host's admission.
+            job_release: hands back what *job_admits* claimed when the host
+                then refuses the job.
+
+        Returns:
+            The started jobs, already counted as running.
+        """
+        started: list[Job] = []
+        for job in list(self.pending):
+            if len(self.running) >= self.cap:
+                break
+            if not self._ready(job):
+                continue
+            if not job_admits(job):
+                continue
+            if not host_admits(len(self.running), job):
+                job_release(job)
+                break
+            self.pending.remove(job)
+            self.running.add(job.id)
+            started.append(job)
+        return started
+
+    def defer_orphans(self) -> list[Job]:
+        """Pull the jobs of a sealed list whose ``after`` can never finish here.
+
+        Returns:
+            Jobs that wait on one never submitted, such as an app staging
+            dropped, or on one deferred, so they run serially after it instead.
+
+        Raises:
+            RuntimeError: nothing runs and the remaining jobs wait on each other.
+        """
+        deferred: list[Job] = []
+        while self.sealed and (orphans := [j for j in self.pending if self._orphan(j)]):
+            for job in orphans:
+                self.pending.remove(job)
+                self.deferred.add(job.id)
+            deferred += orphans
+        stuck = self.sealed and self.pending and not self.running
+        if stuck and not any(self._ready(job) for job in self.pending):
+            raise RuntimeError(f"unsatisfiable after: {self.pending}")
+        return deferred
+
+    def _orphan(self, job: Job) -> bool:
+        return bool(set(job.after) - {job.id} - (self.submitted - self.deferred))
+
+    def finish(self, job_id: str) -> None:
+        self.running.remove(job_id)
+        self.finished.add(job_id)
+
+    def done(self) -> bool:
+        return self.sealed and not self.pending and not self.running
+
+
+class HostGate:
+    """Admit one more container only while the host has room for it.
+
+    The first container always starts, so the round never runs slower than
+    the serial runner.
+
+    The scheduler samples the host on every pass, whether or not a job waits.
+    A further container needs CPU pressure at or below the ceiling in both
+    the 10 s and the 60 s average on every sample of a sustained window that
+    began after the last start; one sample above the ceiling restarts the
+    window. Running containers are never stopped, so their results stay real.
+
+    Args:
+        max_cpu_pressure: highest CPU ``some`` pressure, in percent, in both
+            ``avg10`` and ``avg60``. WordPress#0's serial run peaked at 35.9 %
+            ``some avg60`` (run 41926377524), so no browser is added while the
+            host stalls more than it already did serially.
+        ram_fraction: share of MemAvailable one container's memory estimate
+            may claim, the fraction ``playwright_workers`` sizes browsers by.
+        sustain_seconds: how long every sample must stay calm before another
+            container starts.
+        pressure_path: the kernel's CPU pressure file.
+        meminfo_path: the kernel's meminfo file.
+        clock: monotonic clock.
+    """
+
+    def __init__(
+        self,
+        max_cpu_pressure: float,
+        ram_fraction: float,
+        sustain_seconds: float,
+        pressure_path: Path,
+        meminfo_path: Path,
+        clock: Callable[[], float],
+    ) -> None:
+        self.max_cpu_pressure = max_cpu_pressure
+        self.ram_fraction = ram_fraction
+        self.sustain_seconds = sustain_seconds
+        self.pressure_path = pressure_path
+        self.meminfo_path = meminfo_path
+        self.clock = clock
+        self.calm_since: float | None = None
+        self.available_mb: float | None = None
+
+    def cpu_pressure(self) -> tuple[float, float] | None:
+        try:
+            for line in self.pressure_path.read_text(encoding="utf-8").splitlines():
+                if line.startswith("some "):
+                    fields = dict(f.split("=") for f in line.split()[1:])
+                    return float(fields["avg10"]), float(fields["avg60"])
+        except (OSError, KeyError, ValueError):
+            return None
+        return None
+
+    def mem_available_mb(self) -> float | None:
+        try:
+            for line in self.meminfo_path.read_text(encoding="utf-8").splitlines():
+                if line.startswith("MemAvailable:"):
+                    return int(line.split()[1]) / 1024
+        except (OSError, ValueError, IndexError):
+            return None
+        return None
+
+    def observe(self) -> None:
+        """Take one sample and extend or restart the calm window."""
+        pressure = self.cpu_pressure()
+        self.available_mb = self.mem_available_mb()
+        calm = (
+            pressure is not None
+            and self.available_mb is not None
+            and max(pressure) <= self.max_cpu_pressure
+        )
+        if not calm:
+            self.calm_since = None
+        elif self.calm_since is None:
+            self.calm_since = self.clock()
+
+    def admits(self, running: int, memory_mb: float) -> bool:
+        """Whether one more container may start now; a yes restarts the window.
+
+        Args:
+            running: containers already running.
+            memory_mb: the memory estimate of the container that would start.
+
+        Returns:
+            True for the first container, otherwise only after a full calm
+            window and when the estimate fits the RAM fraction of what the
+            last sample found available.
+        """
+        room = running == 0 or (
+            self.calm_since is not None
+            and self.clock() - self.calm_since >= self.sustain_seconds
+            and self.available_mb is not None
+            and memory_mb <= self.available_mb * self.ram_fraction
+        )
+        if room:
+            self.calm_since = None
+        return room
+
+
+class TorReaders:
+    """Shared locks the running onion jobs hold on the Tor lock file.
+
+    ``wait_onion_reachable.sh`` takes the exclusive lock before it recreates
+    Tor, so a recreate waits until no onion job runs, and no onion job starts
+    while a recreate holds it. Linux ``flock`` does not prefer a waiting
+    writer, so the script announces itself with ``<path>.recreate-pending``
+    and no new shared lock is taken while that file exists. A probe killed
+    before its exit trap ran leaves that file behind, so ``main`` removes it
+    at startup, before staging can probe Tor for this round. The browsers of
+    all running onion jobs together stay within *onion_cap*, the Tor ceiling
+    ``playwright_workers`` sizes one suite by, since every browser holds its
+    own circuits on the one shared daemon.
+
+    Args:
+        path: the lock file both sides open.
+        onion_cap: most browser workers that may run over Tor at once.
+    """
+
+    def __init__(self, path: Path, onion_cap: int) -> None:
+        self.path = path
+        self.pending = path.with_name(path.name + ".recreate-pending")
+        self.onion_cap = onion_cap
+        self.fds: dict[str, int] = {}
+        self.workers: dict[str, int] = {}
+
+    def acquire(self, job_id: str, workers: int) -> bool:
+        if self.workers and sum(self.workers.values()) + workers > self.onion_cap:
+            return False
+        if self.pending.exists():
+            return False
+        fd = os.open(self.path, os.O_RDONLY | os.O_CREAT, 0o644)
+        try:
+            fcntl.flock(fd, fcntl.LOCK_SH | fcntl.LOCK_NB)
+        except BlockingIOError:
+            os.close(fd)
+            return False
+        self.fds[job_id] = fd
+        self.workers[job_id] = workers
+        return True
+
+    def release(self, job_id: str) -> None:
+        self.workers.pop(job_id, None)
+        fd = self.fds.pop(job_id, None)
+        if fd is not None:
+            os.close(fd)
+
+
+def _load_jobs(jobs_dir: Path, seen: set[str]) -> list[dict]:
+    named = []
+    for path in jobs_dir.iterdir():
+        match = JOB_NAME.match(path.name)
+        if match and path.name not in seen:
+            named.append((int(match.group(1)), path))
+    seen.update(path.name for _, path in named)
+    return [json.loads(path.read_text(encoding="utf-8")) for _, path in sorted(named)]
+
+
+def _write_result(jobs_dir: Path, job_id: str, result: dict) -> None:
+    tmp = jobs_dir / f".{job_id}.result.tmp"
+    tmp.write_text(json.dumps(result), encoding="utf-8")
+    tmp.replace(jobs_dir / f"{job_id}{RESULT_SUFFIX}")
+
+
+def _stop(proc: subprocess.Popen, stop_argv: list[str]) -> None:
+    for sig in (signal.SIGTERM, signal.SIGKILL):
+        try:
+            os.killpg(proc.pid, sig)
+        except ProcessLookupError:
+            break
+        try:
+            proc.wait(timeout=KILL_GRACE_SECONDS)
+            break
+        except subprocess.TimeoutExpired:
+            continue
+    subprocess.run(
+        stop_argv, stdin=subprocess.DEVNULL, capture_output=True, check=False
+    )
+
+
+def _nap(procs: list[subprocess.Popen], seconds: float) -> None:
+    """Sleep up to *seconds*, waking as soon as one of *procs* exits."""
+    deadline = time.monotonic() + seconds
+    while time.monotonic() < deadline:
+        if any(proc.poll() is not None for proc in procs):
+            return
+        time.sleep(min(NAP_STEP_SECONDS, max(0.0, deadline - time.monotonic())))
+
+
+def run_signals(output: str) -> dict[str, int]:
+    """What a run's output tells beyond its exit code.
+
+    Args:
+        output: the container's stdout and stderr.
+
+    Returns:
+        ``retries``: attempts Playwright repeated, from its ``(retry #N)``
+        marks; ``flaky``: tests that passed only on a retry, from its summary;
+        ``crashes``: crashed pages, closed browsers and killed test workers,
+        the shapes an OOM kill takes, even when a retry later passed;
+        ``crash_kinds``: those counts by kind, for example
+        ``page crashed 1, worker killed 2``, empty when there were none.
+    """
+    flaky = FLAKY_SUMMARY.findall(output)
+    kinds = {kind: len(mark.findall(output)) for kind, mark in CRASH_MARKS.items()}
+    return {
+        "retries": len(RETRY_MARK.findall(output)),
+        "flaky": int(flaky[-1]) if flaky else 0,
+        "crashes": sum(kinds.values()),
+        "crash_kinds": ", ".join(
+            f"{kind} {count}" for kind, count in kinds.items() if count
+        ),
+    }
+
+
+def _raise_on_signal(signum, _frame):
+    raise SystemExit(128 + signum)
+
+
+def run(jobs_dir: Path, cap: int, poll: float, gate: HostGate, tor: TorReaders) -> int:
+    """Drive the jobs in *jobs_dir* until the list is sealed and drained.
+
+    Args:
+        jobs_dir: directory Ansible streams job files into.
+        cap: most containers running at once.
+        poll: seconds between scans of *jobs_dir* and the running jobs.
+        gate: decides whether the host has room for one more container.
+        tor: shared Tor locks for onion jobs.
+
+    Returns:
+        0 once every job has a result file; a job's own failure lives in its
+        result, so it never aborts the others. On any exit, crash or signal
+        included, every container still running is stopped.
+    """
+    plan = Plan(cap)
+    seen: set[str] = set()
+    specs: dict[str, dict] = {}
+    running: dict[str, tuple[subprocess.Popen, float, float, Path, Path]] = {}
+
+    def job_admits(job: Job) -> bool:
+        spec = specs[job.id]
+        return not spec["tor"] or tor.acquire(job.id, int(spec["workers"]))
+
+    try:
+        while True:
+            for job_id, (proc, started, deadline, out_path, err_path) in list(
+                running.items()
+            ):
+                rc = proc.poll()
+                timed_out = rc is None and time.monotonic() > deadline
+                if rc is None and not timed_out:
+                    continue
+                result = {
+                    "meta": specs[job_id]["meta"],
+                    "timed_out": timed_out,
+                    "seconds": round(time.monotonic() - started, 1),
+                    "ended_at": round(time.time(), 1),
+                }
+                if timed_out:
+                    _stop(proc, specs[job_id]["stop"])
+                    result["msg"] = (
+                        f"killed after its {specs[job_id]['timeout']}s timeout"
+                    )
+                else:
+                    result["rc"] = rc
+                result["stdout"] = out_path.read_text(
+                    encoding="utf-8", errors="replace"
+                )
+                result["stderr"] = err_path.read_text(
+                    encoding="utf-8", errors="replace"
+                ) + (f"\n[run_parallel] {result['msg']}\n" if timed_out else "")
+                result.update(run_signals(result["stdout"] + result["stderr"]))
+                _write_result(jobs_dir, job_id, result)
+                del running[job_id]
+                tor.release(job_id)
+                plan.finish(job_id)
+            sealed_now = (jobs_dir / SEALED).exists()
+            for spec in _load_jobs(jobs_dir, seen):
+                specs[spec["id"]] = spec
+                plan.submit(Job(spec["id"], tuple(spec["after"])))
+            if sealed_now:
+                plan.seal()
+            gate.observe()
+            for job in plan.start_next(
+                lambda count, ready: gate.admits(
+                    count, float(specs[ready.id]["memory_mb"])
+                ),
+                job_admits,
+                lambda refused: tor.release(refused.id),
+            ):
+                spec = specs[job.id]
+                out_path = jobs_dir / f"{job.id}.stdout"
+                err_path = jobs_dir / f"{job.id}.stderr"
+                with out_path.open("wb") as out, err_path.open("wb") as err:
+                    proc = subprocess.Popen(
+                        spec["argv"],
+                        env={**os.environ, **spec["env"]},
+                        stdin=subprocess.DEVNULL,
+                        stdout=out,
+                        stderr=err,
+                        start_new_session=True,
+                    )
+                started = time.monotonic()
+                running[job.id] = (
+                    proc,
+                    started,
+                    started + float(spec["timeout"]),
+                    out_path,
+                    err_path,
+                )
+            for job in plan.defer_orphans():
+                waits = [dep for dep in job.after if dep not in plan.finished]
+                meta = specs[job.id]["meta"]
+                result = {"id": job.id, "meta": meta, "deferred": waits}
+                _write_result(jobs_dir, job.id, result)
+            if plan.done():
+                return 0
+            _nap([entry[0] for entry in running.values()], poll)
+    finally:
+        for job_id, (proc, *_rest) in running.items():
+            _stop(proc, specs[job_id]["stop"])
+            tor.release(job_id)
+
+
+def main(argv: list[str]) -> int:
+    signal.signal(signal.SIGTERM, _raise_on_signal)
+    signal.signal(signal.SIGHUP, _raise_on_signal)
+    jobs_dir, cap, poll, max_cpu, ram_fraction, sustain, tor_lock, onion_cap = argv
+    gate = HostGate(
+        float(max_cpu),
+        float(ram_fraction),
+        float(sustain),
+        Path("/proc/pressure/cpu"),
+        Path("/proc/meminfo"),
+        time.monotonic,
+    )
+    tor = TorReaders(Path(tor_lock), int(onion_cap))
+    tor.pending.unlink(missing_ok=True)
+    return run(Path(jobs_dir), int(cap), float(poll), gate, tor)
+
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv[1:]))

roles/test-e2e-playwright/files/wait_onion_reachable.sh

@@ -10,21 +10,31 @@
 #   $2 domain    -- onion domain (plaintext http; Tor provides the encryption)
 #   $3 attempts  -- probe attempts per pass
 #   $4 delay     -- seconds between attempts
-#   $5 tor_dir   -- svc-net-tor compose instance dir (for --force-recreate)
+#   $5 max_time  -- seconds one probe attempt may take
+#   $6 tor_dir   -- svc-net-tor compose instance dir (for --force-recreate)
+#   $7 tor_lock  -- empty in the serial run, which recreates at once; in the
+#                   parallel run the lock file whose exclusive lock the
+#                   recreate holds. The scheduler holds a shared lock per
+#                   running onion run and starts no new one while
+#                   <tor_lock>.recreate-pending exists. When onion runs still
+#                   hold it after attempts x delay, the recreate is skipped
+#                   and the current circuits get one more probe pass.
 set -euo pipefail
 
 socks="${1:?socks host:port required}"
 domain="${2:?onion domain required}"
 attempts="${3:?attempts required}"
 delay="${4:?delay seconds required}"
-tor_dir="${5:?tor compose dir required}"
+max_time="${5:?max time per attempt required}"
+tor_dir="${6:?tor compose dir required}"
+tor_lock="${7?tor lock file or empty required}"
 
 url="http://${domain}/"
 
 probe() {
   local i code
   for ((i = 1; i <= attempts; i++)); do
-    if code="$(curl --silent --show-error --max-time 120 --socks5-hostname "${socks}" \
+    if code="$(curl --silent --show-error --max-time "${max_time}" --socks5-hostname "${socks}" \
         -o /dev/null -w '%{http_code}' "${url}")" && [ "${code}" -lt 500 ]; then
       return 0
     fi
@@ -37,6 +47,21 @@ if probe; then
   exit 0
 fi
 
+if [ -n "${tor_lock}" ]; then
+	pending="${tor_lock}.recreate-pending"
+	wait_seconds=$((attempts * delay))
+	echo ">>> ${domain} did not answer in ${attempts} attempts; waiting up to ${wait_seconds}s until no onion run uses Tor"
+	touch "${pending}"
+	trap 'rm -f "${pending}"' EXIT
+	exec 9>>"${tor_lock}"
+	if ! flock -x -w "${wait_seconds}" 9; then
+		rm -f "${pending}"
+		echo ">>> skipped recreating Tor for ${domain}: onion Playwright runs held ${tor_lock} for ${wait_seconds}s; probing the current circuits once more" >&2
+		probe
+		exit 0
+	fi
+fi
+
 echo ">>> ${domain} did not answer in ${attempts} attempts; recreating Tor for fresh circuits"
 if docker service inspect tor_tor >/dev/null 2>&1; then
 	docker service update --force tor_tor --detach

roles/test-e2e-playwright/lookup_plugins/playwright_parallel.py

@@ -0,0 +1,144 @@
+"""Lookup `playwright_parallel`: how the round runs its Playwright containers.
+
+    lookup('playwright_parallel')                         -> the round's cap
+    lookup('playwright_parallel', 'memory_mb', workers=N) -> one container's limit
+    lookup('playwright_parallel', 'ram_fraction')         -> share of free RAM it may claim
+    lookup('playwright_parallel', 'onion_cap')            -> browsers that may share Tor
+
+A primary app opts its rows in through ``playwright.parallel`` in its
+``meta/tests.yml``: a list with one entry per ``meta/variants.yml`` variant,
+each the cap for the row that deploys that variant. The round's cap is the
+smallest cap among ``PRIMARY_APPS`` at ``VARIANT_INDEX``, so a round runs in
+parallel only when every app it was asked for opts in. A round without
+primary apps, or with a primary app that declares no list, runs one container
+at a time as it always did, and so does a round whose cap cannot be read: the
+lookup decides the serial loop's ``when`` too, so it warns instead of raising.
+The memory estimate, the RAM fraction and the Tor cap come from the sizing
+``playwright_workers`` applies to every browser.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from ansible.errors import AnsibleError
+from ansible.plugins.loader import lookup_loader
+from ansible.plugins.lookup import LookupBase
+from ansible.utils.display import Display
+
+from plugins.lookup.playwright_workers import ONION_CAP, PER_WORKER_GB, RAM_FRACTION
+
+PARALLEL = "playwright.parallel"
+
+
+def primary_apps(value: Any) -> list[str]:
+    """``PRIMARY_APPS`` as a list of app ids.
+
+    Both CI drivers pass it as ``-e PRIMARY_APPS=["web-app-x"]``, which
+    Ansible's key=value parser hands over as a string.
+
+    Args:
+        value: the templated ``PRIMARY_APPS``, a list or its JSON text.
+
+    Returns:
+        The app ids.
+
+    Raises:
+        AnsibleError: *value* is neither a list of app ids nor JSON for one.
+    """
+    if isinstance(value, str):
+        try:
+            value = json.loads(value)
+        except json.JSONDecodeError as exc:
+            raise AnsibleError(f"PRIMARY_APPS is no JSON list: {value!r}") from exc
+    if not isinstance(value, list) or not all(
+        isinstance(app, str) and app for app in value
+    ):
+        raise AnsibleError(f"PRIMARY_APPS must be a list of app ids, got {value!r}.")
+    return value
+
+
+def validate_caps(app: str, caps: Any) -> list[int]:
+    """*caps* as declared, once it is a list of positive integers.
+
+    Args:
+        app: the primary app whose ``meta/tests.yml`` declared *caps*.
+        caps: the rendered ``playwright.parallel``.
+
+    Returns:
+        *caps*.
+
+    Raises:
+        AnsibleError: *caps* is not a list of positive integers.
+    """
+    if not isinstance(caps, list) or not all(
+        isinstance(cap, int) and not isinstance(cap, bool) and cap >= 1 for cap in caps
+    ):
+        raise AnsibleError(
+            f"{app}: {PARALLEL} in meta/tests.yml must be a list of positive "
+            f"integers, one per variant, got {caps!r}."
+        )
+    return caps
+
+
+def row_cap(app: str, caps: Any, variant: int) -> int:
+    """Cap *app* declares for its row at *variant*.
+
+    ``VARIANT_INDEX`` is the round index. The planner runs an app with fewer
+    variants than the round on variant 0, or on a pinned variant, so that
+    round is not the row the cap was declared for and runs serially.
+
+    Args:
+        app: the primary app whose ``meta/tests.yml`` declared *caps*.
+        caps: the rendered ``playwright.parallel``, ``None`` when absent.
+        variant: the round's ``VARIANT_INDEX``.
+
+    Returns:
+        The declared cap, or 1 when *app* declares none or has no entry for
+        *variant*.
+
+    Raises:
+        AnsibleError: *caps* is not a list of positive integers.
+    """
+    if caps is None:
+        return 1
+    caps = validate_caps(app, caps)
+    return caps[variant] if 0 <= variant < len(caps) else 1
+
+
+class LookupModule(LookupBase):
+    def run(
+        self, terms, variables: dict[str, Any] | None = None, **kwargs
+    ) -> list[Any]:
+        if terms == ["memory_mb"]:
+            return [int(PER_WORKER_GB * 1024 * int(kwargs["workers"]))]
+        if terms == ["ram_fraction"]:
+            return [RAM_FRACTION]
+        if terms == ["onion_cap"]:
+            return [ONION_CAP]
+        if terms:
+            raise AnsibleError(
+                f"lookup('playwright_parallel'): unknown terms {terms!r}."
+            )
+        variables = variables or {}
+        try:
+            apps = primary_apps(self._templar.template(variables["PRIMARY_APPS"]))
+            variant = int(self._templar.template(variables["VARIANT_INDEX"]))
+            role_tests = lookup_loader.get(
+                "role_tests", loader=self._loader, templar=self._templar
+            )
+            caps = [
+                row_cap(
+                    app,
+                    role_tests.run([app, PARALLEL], variables=variables)[0],
+                    variant,
+                )
+                for app in apps
+            ]
+        except (AnsibleError, KeyError, TypeError, ValueError) as exc:
+            Display().warning(
+                f"lookup('playwright_parallel'): {exc}; the round runs serially."
+            )
+            return [1]
+        return [min(caps) if caps else 1]

roles/test-e2e-playwright/tasks/01_setup/orchestrate.yml

@@ -25,10 +25,26 @@
   ansible.builtin.include_tasks: 02_run_one.yml
   vars:
     application_id: "{{ playwright_role_id }}"
+    _playwright_dispatch: false
   loop: "{{ TEST_E2E_PLAYWRIGHT_APPS }}"
   loop_control:
     loop_var: playwright_role_id
     label: "{{ playwright_role_id }}"
+  when: TEST_E2E_PLAYWRIGHT_PARALLEL | int == 1
+
+- name: 🔀 Run the round's Playwright containers in parallel
+  ansible.builtin.include_tasks: 01_setup/parallel.yml
+  when: TEST_E2E_PLAYWRIGHT_PARALLEL | int > 1
+
+- name: ▶️ Run the roles left to the serial runner, in round order, before the retry
+  ansible.builtin.include_tasks: 02_run_one.yml
+  vars:
+    application_id: "{{ playwright_role_id }}"
+    _playwright_dispatch: false
+  loop: "{{ TEST_E2E_PLAYWRIGHT_APPS | select('in', _playwright_serial_roles | default([])) | list }}"
+  loop_control:
+    loop_var: playwright_role_id
+    label: "{{ playwright_role_id }}"
 
 - name: 🔁 Stash first-pass failures for one flake retry
   ansible.builtin.set_fact:
@@ -42,6 +58,7 @@
   ansible.builtin.include_tasks: 02_run_one.yml
   vars:
     application_id: "{{ playwright_role_id }}"
+    _playwright_dispatch: false
   loop: "{{ _playwright_retry_roles | default([]) }}"
   loop_control:
     loop_var: playwright_role_id

roles/test-e2e-playwright/tasks/01_setup/parallel.yml

@@ -0,0 +1,121 @@
+---
+- name: 📁 Recreate the parallel jobs directory
+  ansible.builtin.file:
+    path: "{{ TEST_E2E_PLAYWRIGHT_PARALLEL_JOBS_DIR }}"
+    state: "{{ playwright_jobs_dir_state }}"
+    mode: "0755"
+  loop: [absent, directory]
+  loop_control:
+    loop_var: playwright_jobs_dir_state
+
+- name: 🧹 Remove parallel Playwright containers a killed scheduler left behind
+  ansible.builtin.command:
+    argv: "{{ ['container', 'rm', '--force'] + (TEST_E2E_PLAYWRIGHT_APPS | map('regex_replace', '^', TEST_E2E_PLAYWRIGHT_PARALLEL_CONTAINER_PREFIX) | list) }}"
+  changed_when: false
+  failed_when: false
+
+- name: 📝 Start with no run handed to the scheduler or left to the serial runner
+  ansible.builtin.set_fact:
+    _playwright_submitted: []
+    _playwright_serial_roles: []
+
+- name: 📋 Ship the parallel scheduler
+  ansible.builtin.copy:
+    src: "{{ role_path }}/files/run_parallel.py"
+    dest: "{{ TEST_E2E_PLAYWRIGHT_PARALLEL_SCHEDULER }}"
+    mode: "0755"
+
+- name: 🚦 Start the scheduler that runs the staged containers
+  ansible.builtin.command:
+    argv:
+      - python3
+      - "{{ TEST_E2E_PLAYWRIGHT_PARALLEL_SCHEDULER }}"
+      - "{{ TEST_E2E_PLAYWRIGHT_PARALLEL_JOBS_DIR }}"
+      - "{{ TEST_E2E_PLAYWRIGHT_PARALLEL | int }}"
+      - "{{ TEST_E2E_PLAYWRIGHT_PARALLEL_POLL_SECONDS }}"
+      - "{{ TEST_E2E_PLAYWRIGHT_PARALLEL_MAX_CPU_PRESSURE }}"
+      - "{{ lookup('playwright_parallel', 'ram_fraction') }}"
+      - "{{ TEST_E2E_PLAYWRIGHT_PARALLEL_SUSTAIN_SECONDS }}"
+      - "{{ TEST_E2E_PLAYWRIGHT_TOR_LOCK }}"
+      - "{{ lookup('playwright_parallel', 'onion_cap') }}"
+  # rationale: async; runs on the stack host it was started on, and the always section seals, waits for and reaps it before the play goes on, so no container outlives this file
+  async: "{{ TEST_E2E_PLAYWRIGHT_PARALLEL_BUDGET | int }}"
+  poll: 0
+  register: _playwright_scheduler
+  changed_when: false
+
+- name: 📦 Stage each role and hand its run to the scheduler
+  block:
+    - name: ▶️ Stage Playwright for each detected role
+      ansible.builtin.include_tasks: 02_run_one.yml
+      vars:
+        application_id: "{{ playwright_role_id }}"
+        _playwright_dispatch: true
+      loop: "{{ TEST_E2E_PLAYWRIGHT_APPS }}"
+      loop_control:
+        loop_var: playwright_role_id
+        label: "{{ playwright_role_id }}"
+  always:
+    - name: 🔏 Seal the job list so the scheduler drains and exits
+      ansible.builtin.file:
+        path: "{{ TEST_E2E_PLAYWRIGHT_PARALLEL_JOBS_DIR }}/sealed"
+        state: touch
+        mode: "0644"
+        modification_time: preserve
+        access_time: preserve
+
+    - name: ⏳ Wait for the scheduler to drain the job list
+      ansible.builtin.async_status:
+        jid: "{{ _playwright_scheduler.ansible_job_id }}"
+      register: _playwright_scheduler_status
+      until: _playwright_scheduler_status.finished | bool
+      retries: "{{ ((TEST_E2E_PLAYWRIGHT_PARALLEL_BUDGET | int) / (TEST_E2E_PLAYWRIGHT_PARALLEL_WAIT_DELAY | int)) | round(0, 'ceil') | int }}"
+      delay: "{{ TEST_E2E_PLAYWRIGHT_PARALLEL_WAIT_DELAY }}"
+      failed_when: false
+
+    - name: 🧹 Remove any handed-over container the scheduler did not stop
+      ansible.builtin.command:
+        argv: "{{ ['container', 'rm', '--force'] + (_playwright_submitted | map(attribute='id') | map('regex_replace', '^', TEST_E2E_PLAYWRIGHT_PARALLEL_CONTAINER_PREFIX) | list) }}"
+      changed_when: false
+      failed_when: false
+      when: _playwright_submitted | length > 0
+
+    - name: 📥 Read each handed-over run's job and result
+      ansible.builtin.slurp:
+        src: "{{ TEST_E2E_PLAYWRIGHT_PARALLEL_JOBS_DIR }}/{{ playwright_read }}"
+      register: _playwright_reads
+      failed_when: false
+      loop: "{{ _playwright_submitted | map(attribute='job') | zip(_playwright_submitted | map(attribute='id') | map('regex_replace', '$', '.result.json')) | flatten }}"
+      loop_control:
+        loop_var: playwright_read
+
+    - name: 🧭 Leave the runs whose dependency never ran here to the serial runner
+      ansible.builtin.set_fact:
+        _playwright_serial_roles: >-
+          {{ _playwright_serial_roles
+             + (_playwright_reads.results | selectattr('content', 'defined')
+                | map(attribute='content') | map('b64decode') | map('from_json')
+                | selectattr('deferred', 'defined') | map(attribute='id') | list) }}
+
+    - name: 📝 Report and archive each handed-over run
+      ansible.builtin.include_tasks: 04_collect.yml
+      vars:
+        application_id: "{{ playwright_submitted.id }}"
+        _playwright_job_read: "{{ _playwright_reads.results[2 * playwright_index] }}"
+        _playwright_result_read: "{{ _playwright_reads.results[2 * playwright_index + 1] }}"
+        _playwright_pass_reports_dir: "{{ (_playwright_job_read.content | b64decode | from_json).meta.reports_dir }}"
+        _playwright_run: >-
+          {{ (_playwright_result_read.content | b64decode | from_json)
+             if _playwright_result_read.content is defined
+             else {'stderr': 'no result: the scheduler ended before this run finished'} }}
+      loop: "{{ _playwright_submitted }}"
+      loop_control:
+        loop_var: playwright_submitted
+        index_var: playwright_index
+        label: "{{ playwright_submitted.id }}"
+      when: _playwright_run.deferred is not defined
+
+    - name: 🛑 Fail when the scheduler itself did not finish cleanly
+      ansible.builtin.fail:
+        msg: "The parallel Playwright scheduler ended abnormally: {{ _playwright_scheduler_status.stderr | default(_playwright_scheduler_status.msg) }}"
+      when: not (_playwright_scheduler_status.finished | bool) or 'rc' not in _playwright_scheduler_status or _playwright_scheduler_status.rc != 0

roles/test-e2e-playwright/tasks/03_run_pass.yml

@@ -59,86 +59,109 @@
     {{ _playwright_pass_domain | quote }}
     {{ TEST_E2E_PLAYWRIGHT_ONION_PROBE_RETRIES | int }}
     {{ TEST_E2E_PLAYWRIGHT_ONION_PROBE_DELAY | int }}
+    {{ TEST_E2E_PLAYWRIGHT_ONION_PROBE_MAX_TIME | int }}
     {{ lookup('container', 'svc-net-tor', 'directories.instance') | quote }}
+    {{ (TEST_E2E_PLAYWRIGHT_TOR_LOCK if (_playwright_dispatch | bool) else '') | quote }}
+  register: _playwright_probe
   changed_when: false
+  failed_when: _playwright_probe.rc != 0 and not (_playwright_dispatch | bool)
   become: true
   when: TEST_E2E_PLAYWRIGHT_IS_ONION | bool
 
-- name: ▶️ Run Playwright container
-  ansible.builtin.command:
-    argv: >-
-      {{
-        ['container', 'run', '--rm', '--ipc=host', '--shm-size=1g']
-        + (
-            ['--network', 'host']
-            if IS_SWARM_MODE
-            else (
-              ['--add-host=host.docker.internal:host-gateway']
-              + (
-                  ['--add-host=' ~ lookup('tls', 'web-app-keycloak', 'domain') ~ ':host-gateway']
-                  if 'web-app-keycloak' in group_names
-                  else []
-                )
-            )
-          )
-        + ['--env-file', _playwright_stage_dir ~ '/.env']
-        + ['-e', 'INFINITO_PLAYWRIGHT_KEEP=' ~ test_e2e_playwright_keep]
-        + ['-e', 'INFINITO_PLAYWRIGHT_GLOBAL_TIMEOUT_MS=' ~ TEST_E2E_PLAYWRIGHT_GLOBAL_TIMEOUT_MS]
-        + ['-e', 'PLAYWRIGHT_PROXY=' ~ TEST_E2E_PLAYWRIGHT_PROXY]
-        + ['-e', 'PLAYWRIGHT_WORKERS=' ~ test_e2e_playwright_workers]
-        + ['-e', 'PLAYWRIGHT_FULLY_PARALLEL=' ~ TEST_E2E_PLAYWRIGHT_FULLY_PARALLEL]
-        + ['-e', 'PLAYWRIGHT_TIMEOUT_FACTOR=' ~ TIMEOUT_FACTOR]
-        + (['-e', 'PLAYWRIGHT_ONION_TIMEOUT_MULTIPLIER=' ~ TEST_E2E_PLAYWRIGHT_ONION_TIMEOUT_MULTIPLIER] if (TEST_E2E_PLAYWRIGHT_IS_ONION | bool) else [])
-        + ['-v', _playwright_stage_dir ~ ':/e2e']
-        + ['-v', _playwright_stage_dir ~ '/volume:/volume']
-        + ['-v', _playwright_pass_reports_dir ~ ':/reports']
-        + ['-v', TEST_E2E_PLAYWRIGHT_SHARED_NODE_MODULES ~ ':/e2e/node_modules:ro']
-        + ['-v', TEST_E2E_PLAYWRIGHT_SHARED_PERSONAS ~ ':/e2e/tests/personas:ro']
-        + ['-w', '/e2e']
-        + [TEST_E2E_PLAYWRIGHT_IMAGE]
-        + ['/bin/bash', '-lc', TEST_E2E_PLAYWRIGHT_COMMAND]
-      }}
-  timeout: "{{ lookup('timeout', TEST_E2E_PLAYWRIGHT_RUN_TIMEOUT_EFFECTIVE) | int }}"
-  environment: >-
-    {{
-      {
-        'CA_TRUST_CERT_HOST': _playwright_ca_cert_host,
-        'CA_TRUST_WRAPPER_HOST': _playwright_ca_wrapper_host,
-        'CA_TRUST_NAME': _playwright_ca_trust_name
-      } if (_playwright_ca_required | bool) else {}
-    }}
-  register: _playwright_run
-  changed_when: false
-  failed_when: false
-
-- name: 📝 Report and record the failed Playwright run
-  when: (_playwright_run.rc | default(1)) != 0
+- name: 🏃 Run the Playwright container in this play and report it
+  when: not (_playwright_dispatch | bool)
   block:
-    - name: 💬 Print Playwright output (on failure)
-      ansible.builtin.debug:
-        msg: |
-          rc: {{ _playwright_run.rc | default('<absent: run was aborted>') }}
-
-          stdout:
-          {{ _playwright_run.stdout | default('') }}
-
-          stderr:
-          {{ _playwright_run.stderr | default(_playwright_run.msg | default('')) }}
+    - name: ▶️ Run Playwright container
+      ansible.builtin.command:
+        argv: &playwright_argv >-
+          {{
+            ['container', 'run', '--rm', '--ipc=host', '--shm-size=1g']
+            + (
+                ['--network', 'host']
+                if IS_SWARM_MODE
+                else (
+                  ['--add-host=host.docker.internal:host-gateway']
+                  + (
+                      ['--add-host=' ~ lookup('tls', 'web-app-keycloak', 'domain') ~ ':host-gateway']
+                      if 'web-app-keycloak' in group_names
+                      else []
+                    )
+                )
+              )
+            + ['--env-file', _playwright_stage_dir ~ '/.env']
+            + ['-e', 'INFINITO_PLAYWRIGHT_KEEP=' ~ test_e2e_playwright_keep]
+            + ['-e', 'INFINITO_PLAYWRIGHT_GLOBAL_TIMEOUT_MS=' ~ TEST_E2E_PLAYWRIGHT_GLOBAL_TIMEOUT_MS]
+            + ['-e', 'PLAYWRIGHT_PROXY=' ~ TEST_E2E_PLAYWRIGHT_PROXY]
+            + ['-e', 'PLAYWRIGHT_WORKERS=' ~ test_e2e_playwright_workers]
+            + ['-e', 'PLAYWRIGHT_FULLY_PARALLEL=' ~ TEST_E2E_PLAYWRIGHT_FULLY_PARALLEL]
+            + ['-e', 'PLAYWRIGHT_TIMEOUT_FACTOR=' ~ TIMEOUT_FACTOR]
+            + (['-e', 'PLAYWRIGHT_ONION_TIMEOUT_MULTIPLIER=' ~ TEST_E2E_PLAYWRIGHT_ONION_TIMEOUT_MULTIPLIER] if (TEST_E2E_PLAYWRIGHT_IS_ONION | bool) else [])
+            + ['-v', _playwright_stage_dir ~ ':/e2e']
+            + ['-v', _playwright_stage_dir ~ '/volume:/volume']
+            + ['-v', _playwright_pass_reports_dir ~ ':/reports']
+            + ['-v', TEST_E2E_PLAYWRIGHT_SHARED_NODE_MODULES ~ ':/e2e/node_modules:ro']
+            + ['-v', TEST_E2E_PLAYWRIGHT_SHARED_PERSONAS ~ ':/e2e/tests/personas:ro']
+            + ['-w', '/e2e']
+            + [TEST_E2E_PLAYWRIGHT_IMAGE]
+            + ['/bin/bash', '-lc', TEST_E2E_PLAYWRIGHT_COMMAND]
+          }}
+      timeout: "{{ lookup('timeout', TEST_E2E_PLAYWRIGHT_RUN_TIMEOUT_EFFECTIVE) | int }}"
+      environment: &playwright_environment >-
+        {{
+          {
+            'CA_TRUST_CERT_HOST': _playwright_ca_cert_host,
+            'CA_TRUST_WRAPPER_HOST': _playwright_ca_wrapper_host,
+            'CA_TRUST_NAME': _playwright_ca_trust_name
+          } if (_playwright_ca_required | bool) else {}
+        }}
+      register: _playwright_run
       changed_when: false
+      failed_when: false
 
-    - name: 📝 Record failure
+    - name: 📝 Report and archive the Playwright run
+      ansible.builtin.include_tasks: 04_collect.yml
+
+- name: 📬 Queue the Playwright run for the parallel scheduler
+  when: _playwright_dispatch | bool
+  block:
+    - name: 📮 Hand the Playwright run to the parallel scheduler
+      ansible.builtin.copy:
+        dest: "{{ TEST_E2E_PLAYWRIGHT_PARALLEL_JOBS_DIR }}/{{ TEST_E2E_PLAYWRIGHT_APPS.index(application_id) }}-{{ application_id }}.job.json"
+        mode: "0644"
+        content: >-
+          {{
+            {
+              'id': application_id,
+              'argv': _playwright_argv[:2] + _playwright_parallel_flags + _playwright_argv[2:],
+              'env': _playwright_environment,
+              'timeout': lookup('timeout', TEST_E2E_PLAYWRIGHT_RUN_TIMEOUT_EFFECTIVE) | int,
+              'after': ([TEST_E2E_PLAYWRIGHT_APPS | first] + lookup('role_tests', application_id, 'playwright.run_after', default=[]))
+                       | intersect(TEST_E2E_PLAYWRIGHT_APPS) | difference([application_id]),
+              'tor': TEST_E2E_PLAYWRIGHT_IS_ONION | bool,
+              'workers': _playwright_workers | int,
+              'memory_mb': lookup('playwright_parallel', 'memory_mb', workers=_playwright_workers) | int,
+              'stop': ['container', 'rm', '--force', TEST_E2E_PLAYWRIGHT_PARALLEL_CONTAINER_PREFIX ~ application_id],
+              'meta': {'reports_dir': _playwright_pass_reports_dir}
+            } | to_json
+          }}
+      vars:
+        _playwright_argv: *playwright_argv
+        _playwright_environment: *playwright_environment
+        _playwright_workers: "{{ (_playwright_argv | select('match', '^PLAYWRIGHT_WORKERS=') | first).split('=', 1)[1] }}"
+        _playwright_parallel_flags:
+          - "--name={{ TEST_E2E_PLAYWRIGHT_PARALLEL_CONTAINER_PREFIX ~ application_id }}"
+          - "--oom-score-adj={{ TEST_E2E_PLAYWRIGHT_PARALLEL_OOM_SCORE_ADJ }}"
+      register: _playwright_job_written
+      when: _playwright_probe is skipped or (_playwright_probe.rc | default(1)) == 0
+
+    - name: 📝 Remember the run the scheduler now owns
       ansible.builtin.set_fact:
-        _playwright_failures: "{{ (_playwright_failures | default([])) + [application_id] }}"
+        _playwright_submitted: >-
+          {{ _playwright_submitted
+             + [{'id': application_id, 'job': _playwright_job_written.dest | basename}] }}
+      when: _playwright_job_written is not skipped
 
-    - name: 📝 Record a suite-budget timeout, which no retry can change
+    - name: 📝 Leave the run to the serial runner, its onion never answered over Tor
       ansible.builtin.set_fact:
-        _playwright_budget_timeouts: "{{ (_playwright_budget_timeouts | default([])) + [application_id] }}"
-      when: "'for the test suite to run' in (_playwright_run.stdout | default(''))"
-
-- name: 🗄️ Archive Playwright reports for this variant + pass
-  ansible.builtin.script: >-
-    {{ role_path }}/files/archive_playwright_reports.sh
-    {{ _playwright_pass_reports_dir | quote }}
-    {{ VARIANT_INDEX | int }}
-    {{ ASYNC_ENABLED | bool | string | lower }}
-  changed_when: true
+        _playwright_serial_roles: "{{ _playwright_serial_roles + [application_id] }}"
+      when: _playwright_job_written is skipped

roles/test-e2e-playwright/tasks/04_collect.yml

@@ -0,0 +1,54 @@
+---
+- name: 🧾 Account for a run the parallel scheduler finished
+  when: _playwright_run.ended_at is defined
+  block:
+    - name: ⏱️ Show when the parallel run ended, how long it took and what it retried
+      ansible.builtin.debug:
+        msg: >-
+          {{ application_id }}: {{ _playwright_run.seconds }}s, ended at {{ _playwright_run.ended_at }},
+          retries {{ _playwright_run.retries }}, flaky {{ _playwright_run.flaky }},
+          browser crashes {{ _playwright_run.crashes }}{{ (' (' ~ _playwright_run.crash_kinds ~ '; an OOM kill surfaces as one of these, even when a retry passed)') if (_playwright_run.crashes | int) > 0 else '' }}
+      changed_when: false
+
+    - name: 📋 Keep the parallel run's output and result with its reports
+      ansible.builtin.copy:
+        src: "{{ TEST_E2E_PLAYWRIGHT_PARALLEL_JOBS_DIR }}/{{ application_id }}.{{ playwright_output }}"
+        dest: "{{ _playwright_pass_reports_dir }}/run-parallel/"
+        remote_src: true
+        mode: "0644"
+      become: true
+      loop: [result.json, stdout, stderr]
+      loop_control:
+        loop_var: playwright_output
+
+- name: 📝 Report and record the failed Playwright run
+  when: (_playwright_run.rc | default(1)) != 0
+  block:
+    - name: 💬 Print Playwright output (on failure)
+      ansible.builtin.debug:
+        msg: |
+          rc: {{ _playwright_run.rc | default('<absent: run was aborted>') }}
+
+          stdout:
+          {{ _playwright_run.stdout | default('') }}
+
+          stderr:
+          {{ _playwright_run.stderr | default(_playwright_run.msg | default('')) }}
+      changed_when: false
+
+    - name: 📝 Record failure
+      ansible.builtin.set_fact:
+        _playwright_failures: "{{ (_playwright_failures | default([])) + [application_id] }}"
+
+    - name: 📝 Record a suite-budget timeout, which no retry can change
+      ansible.builtin.set_fact:
+        _playwright_budget_timeouts: "{{ (_playwright_budget_timeouts | default([])) + [application_id] }}"
+      when: "'for the test suite to run' in (_playwright_run.stdout | default(''))"
+
+- name: 🗄️ Archive Playwright reports for this variant + pass
+  ansible.builtin.script: >-
+    {{ role_path }}/files/archive_playwright_reports.sh
+    {{ _playwright_pass_reports_dir | quote }}
+    {{ VARIANT_INDEX | int }}
+    {{ ASYNC_ENABLED | bool | string | lower }}
+  changed_when: true

roles/test-e2e-playwright/vars/main.yml

@@ -31,6 +31,37 @@ TEST_E2E_PLAYWRIGHT_HELPERS_DIR: "{{ lookup('path_absolute', 'roles/test-e2e-pla
 TEST_E2E_PLAYWRIGHT_HELPERS_ARCHIVE: "/tmp/infinito-playwright-helpers.tar"
 TEST_E2E_PLAYWRIGHT_SHARED_STAMP: "{{ TEST_E2E_PLAYWRIGHT_SHARED_DIR }}/.installed-version"
 
+TEST_E2E_PLAYWRIGHT_PARALLEL: "{{ lookup('playwright_parallel') }}"
+TEST_E2E_PLAYWRIGHT_PARALLEL_JOBS_DIR: "{{ TEST_E2E_PLAYWRIGHT_STAGE_BASE_DIR }}/_jobs"
+TEST_E2E_PLAYWRIGHT_PARALLEL_SCHEDULER: "{{ TEST_E2E_PLAYWRIGHT_SHARED_DIR }}/run_parallel.py"
+TEST_E2E_PLAYWRIGHT_PARALLEL_POLL_SECONDS: 1
+TEST_E2E_PLAYWRIGHT_PARALLEL_WAIT_DELAY: 10
+TEST_E2E_PLAYWRIGHT_PARALLEL_MAX_CPU_PRESSURE: 36
+TEST_E2E_PLAYWRIGHT_PARALLEL_SUSTAIN_SECONDS: 30
+TEST_E2E_PLAYWRIGHT_PARALLEL_OOM_SCORE_ADJ: 1000
+TEST_E2E_PLAYWRIGHT_PARALLEL_CONTAINER_PREFIX: "infinito-playwright-"
+TEST_E2E_PLAYWRIGHT_TOR_LOCK: "{{ TEST_E2E_PLAYWRIGHT_STAGE_BASE_DIR }}/tor.lock"
+TEST_E2E_PLAYWRIGHT_PARALLEL_BUDGET: >-
+  {%- set retries = TEST_E2E_PLAYWRIGHT_ONION_PROBE_RETRIES | int -%}
+  {%- set delay = TEST_E2E_PLAYWRIGHT_ONION_PROBE_DELAY | int -%}
+  {%- set onion_wait = 2 * retries * ((TEST_E2E_PLAYWRIGHT_ONION_PROBE_MAX_TIME | int) + delay) + retries * delay -%}
+  {%- set clearnet_wait = (test_e2e_playwright_wait_retries | int) * (test_e2e_playwright_wait_delay | int) -%}
+  {%- set ns = namespace(total=0) -%}
+  {%- for app in TEST_E2E_PLAYWRIGHT_APPS -%}
+  {%- set role_timeout = lookup('role_tests', app, 'playwright.timeout', default='') | string | trim -%}
+  {%- set picks = [lookup('domain', app)] if lookup('config', app, 'networks.reachability.handover', False)
+        else lookup('network_siblings', lookup('domain', app)) -%}
+  {%- set worst = namespace(seconds=0) -%}
+  {%- for network in picks | map('network_of') | unique -%}
+  {%- set onion = network == 'tor' -%}
+  {%- set run_timeout = (role_timeout | int) if (role_timeout | length > 0)
+        else (TEST_E2E_PLAYWRIGHT_ONION_RUN_TIMEOUT if onion else test_e2e_playwright_run_timeout) -%}
+  {%- set worst.seconds = [worst.seconds, (lookup('timeout', run_timeout) | int) + (onion_wait if onion else clearnet_wait)] | max -%}
+  {%- endfor -%}
+  {%- set ns.total = ns.total + worst.seconds -%}
+  {%- endfor -%}
+  {{ ns.total }}
+
 TEST_E2E_PLAYWRIGHT_IS_ONION: "{{ (_playwright_pass_domain | network_of) == 'tor' }}"
 TEST_E2E_PLAYWRIGHT_PROXY: "{{ ('socks5://' ~ lookup('tor_socks', '127.0.0.1' if IS_SWARM_MODE else 'host.docker.internal')) if (TEST_E2E_PLAYWRIGHT_IS_ONION | bool)
   else '' }}"
@@ -45,6 +76,7 @@ TEST_E2E_PLAYWRIGHT_ONION_TIMEOUT_MULTIPLIER: "{{ lookup('config', 'svc-net-tor'
 TEST_E2E_PLAYWRIGHT_WAIT_ENABLED: true
 TEST_E2E_PLAYWRIGHT_ONION_PROBE_RETRIES: 40
 TEST_E2E_PLAYWRIGHT_ONION_PROBE_DELAY: 15
+TEST_E2E_PLAYWRIGHT_ONION_PROBE_MAX_TIME: 120
 TEST_E2E_PLAYWRIGHT_NAVIGATION_TIMEOUT: 60000
 TEST_E2E_PLAYWRIGHT_ONION_NAVIGATION_TIMEOUT: 180000
 TEST_E2E_PLAYWRIGHT_EXPECT_TIMEOUT: 10000

roles/web-app-wordpress/meta/tests.yml

@@ -1,3 +1,5 @@
 ---
 cli:
   timeout: 900
+playwright:
+  parallel: [3, 1, 1]

tests/unit/python/roles/test-e2e-playwright/__init__.py

@@ -1,3 +1,30 @@
 from pathlib import Path
 
+from utils.cache.yaml import load_yaml_any
+
 PROJECT_ROOT: Path = Path(__file__).resolve().parents[5]
+ROLE: Path = PROJECT_ROOT / "roles" / "test-e2e-playwright"
+
+
+def role_task(task_file: str, prefix: str) -> dict:
+    """Find a task of the Playwright role by the start of its name.
+
+    Args:
+        task_file: path of the task file below the role's ``tasks`` directory.
+        prefix: the start of the task name, usually its emoji.
+
+    Returns:
+        The first matching task, searched depth first through ``block``,
+        ``rescue`` and ``always``.
+    """
+    pending = list(
+        load_yaml_any(str(ROLE / "tasks" / task_file), default_if_missing=[])
+    )
+    while pending:
+        task = pending.pop(0)
+        if task.get("name", "").startswith(prefix):
+            return task
+        pending[:0] = (
+            task.get("block", []) + task.get("rescue", []) + task.get("always", [])
+        )
+    raise LookupError(f"{task_file}: no task named {prefix}")

tests/unit/python/roles/test-e2e-playwright/files/__init__.py

@@ -0,0 +1,5 @@
+from pathlib import Path
+
+PROJECT_ROOT: Path = Path(__file__).resolve().parents[6]
+
+__all__ = ["PROJECT_ROOT"]

tests/unit/python/roles/test-e2e-playwright/files/test_run_parallel.py

@@ -0,0 +1,745 @@
+import fcntl
+import importlib.util
+import itertools
+import json
+import os
+import signal
+import subprocess
+import sys
+import tempfile
+import threading
+import time
+import unittest
+from pathlib import Path
+
+from . import PROJECT_ROOT
+
+
+def _load_module():
+    path = PROJECT_ROOT / "roles" / "test-e2e-playwright" / "files" / "run_parallel.py"
+    spec = importlib.util.spec_from_file_location("playwright_run_parallel", path)
+    module = importlib.util.module_from_spec(spec)
+    assert spec and spec.loader
+    sys.modules["playwright_run_parallel"] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+rp = _load_module()
+
+ROW = [
+    ("web-app-keycloak", 118),
+    ("web-app-hermes", 50),
+    ("web-app-matomo", 165),
+    ("web-app-openclaw", 29),
+    ("web-app-weblate", 84),
+    ("web-svc-cdn", 8),
+    ("web-svc-design", 85),
+    ("web-svc-logout", 76),
+    ("web-svc-mirror", 7),
+    ("web-svc-simpleicons", 8),
+    ("web-svc-translate", 20),
+    ("web-app-dashboard", 247),
+    ("web-app-discourse", 224),
+    ("web-app-stalwart", 162),
+    ("web-app-jitsi", 65),
+    ("web-app-matrix", 449),
+    ("web-app-prometheus", 39),
+    ("web-app-wordpress", 182),
+]
+
+
+def simulate(jobs, cap, release=None):
+    """Drive a Plan on a virtual clock.
+
+    Args:
+        jobs: ``(Job, seconds)`` in submission order.
+        cap: the plan's concurrency cap.
+        release: per job id, the virtual second its file appears; 0 when absent.
+
+    Returns:
+        ``(starts, ends, peak)``: per job id its start and end second, and the
+        most jobs that ever ran at once.
+    """
+    release = release or {}
+    plan = rp.Plan(cap)
+    durations = {job.id: seconds for job, seconds in jobs}
+    queue = list(jobs)
+    starts, ends, running = {}, {}, {}
+    now, peak = 0.0, 0
+    while True:
+        while queue and release.get(queue[0][0].id, 0) <= now:
+            plan.submit(queue.pop(0)[0])
+        if not queue:
+            plan.seal()
+        for job in plan.start_next():
+            starts[job.id] = now
+            running[job.id] = now + durations[job.id]
+        peak = max(peak, len(running))
+        if plan.done():
+            return starts, ends, peak
+        events = list(running.values()) + (
+            [release.get(queue[0][0].id, 0)] if queue else []
+        )
+        now = min(event for event in events if event >= now)
+        for job_id, end in list(running.items()):
+            if end <= now:
+                del running[job_id]
+                ends[job_id] = end
+                plan.finish(job_id)
+
+
+def row_jobs():
+    head = ROW[0][0]
+    return [
+        (rp.Job(app, () if app == head else (head,)), seconds) for app, seconds in ROW
+    ]
+
+
+class TestPlan(unittest.TestCase):
+    def test_the_cap_is_never_exceeded(self):
+        for cap in (1, 2, 3, 4):
+            with self.subTest(cap=cap):
+                _, _, peak = simulate(row_jobs(), cap)
+                self.assertLessEqual(peak, cap)
+
+    def test_cap_one_reproduces_the_serial_order(self):
+        starts, ends, _ = simulate(row_jobs(), 1)
+        order = sorted(starts, key=starts.get)
+        self.assertEqual(order, [app for app, _ in ROW])
+        for previous, following in itertools.pairwise(order):
+            self.assertLessEqual(ends[previous], starts[following])
+
+    def test_no_job_starts_before_the_jobs_it_runs_after(self):
+        jobs = row_jobs()
+        jobs[13] = (
+            rp.Job("web-app-stalwart", ("web-app-keycloak", "web-app-matomo")),
+            162,
+        )
+        starts, ends, _ = simulate(jobs, 3)
+        for job, _ in jobs:
+            for dep in job.after:
+                self.assertGreaterEqual(
+                    starts[job.id], ends[dep], f"{job.id} started before {dep} ended"
+                )
+
+    def test_the_identity_provider_runs_alone_first(self):
+        starts, ends, _ = simulate(row_jobs(), 3)
+        head = ROW[0][0]
+        self.assertEqual(starts[head], 0)
+        self.assertTrue(all(starts[app] >= ends[head] for app, _ in ROW[1:]))
+
+    def test_every_job_runs_exactly_once(self):
+        starts, ends, _ = simulate(row_jobs(), 3)
+        self.assertEqual(set(starts), {app for app, _ in ROW})
+        self.assertEqual(set(ends), set(starts))
+
+    def test_a_dependency_that_is_not_submitted_yet_holds_the_job(self):
+        plan = rp.Plan(3)
+        plan.submit(rp.Job("b", ("a",)))
+        self.assertEqual(plan.start_next(), [])
+        plan.submit(rp.Job("a", ()))
+        self.assertEqual([job.id for job in plan.start_next()], ["a"])
+
+    def test_a_dependency_that_never_ran_defers_its_dependents_transitively(self):
+        plan = rp.Plan(3)
+        plan.submit(rp.Job("b", ("dropped",)))
+        plan.submit(rp.Job("c", ("b",)))
+        plan.submit(rp.Job("d", ()))
+        plan.seal()
+        self.assertEqual([job.id for job in plan.start_next()], ["d"])
+        self.assertEqual([job.id for job in plan.defer_orphans()], ["b", "c"])
+        plan.finish("d")
+        self.assertTrue(plan.done())
+
+    def test_a_ready_job_overtakes_a_blocked_one_but_keeps_submission_order(self):
+        plan = rp.Plan(2)
+        plan.submit(rp.Job("a", ()))
+        plan.submit(rp.Job("b", ("a",)))
+        plan.submit(rp.Job("c", ()))
+        plan.submit(rp.Job("d", ()))
+        self.assertEqual([job.id for job in plan.start_next()], ["a", "c"])
+
+    def test_an_unsatisfiable_sealed_list_raises(self):
+        plan = rp.Plan(2)
+        plan.submit(rp.Job("a", ("b",)))
+        plan.submit(rp.Job("b", ("a",)))
+        plan.seal()
+        self.assertEqual(plan.start_next(), [])
+        with self.assertRaises(RuntimeError):
+            plan.defer_orphans()
+
+    def test_a_duplicate_submission_raises(self):
+        plan = rp.Plan(2)
+        plan.submit(rp.Job("a", ()))
+        with self.assertRaises(ValueError):
+            plan.submit(rp.Job("a", ()))
+
+    def test_a_cap_below_one_raises(self):
+        with self.assertRaises(ValueError):
+            rp.Plan(0)
+
+    def test_the_measured_row_drains_well_inside_the_serial_time(self):
+        setup = 27
+        release = {app: setup * (index + 1) for index, (app, _) in enumerate(ROW)}
+        _, ends, _ = simulate(row_jobs(), 3, release)
+        serial = sum(seconds + setup for _, seconds in ROW)
+        self.assertEqual(serial, 2504)
+        self.assertEqual(max(ends.values()), 1020)
+
+
+class FakeHost:
+    """Pressure and meminfo files a test rewrites at will.
+
+    Args:
+        root: directory for the two files.
+    """
+
+    def __init__(self, root: Path) -> None:
+        self.pressure = root / "pressure"
+        self.meminfo = root / "meminfo"
+        self.now = 0.0
+        self.set(cpu_avg10=5.0, mem_available_mb=8192)
+
+    def set(
+        self, *, cpu_avg10: float, mem_available_mb: int, cpu_avg60: float = 1.0
+    ) -> None:
+        self.pressure.write_text(
+            f"some avg10={cpu_avg10:.2f} avg60={cpu_avg60:.2f} avg300=1.00 total=1\n"
+            "full avg10=0.00 avg60=0.00 avg300=0.00 total=0\n"
+        )
+        self.meminfo.write_text(
+            f"MemTotal: 16000000 kB\nMemAvailable: {mem_available_mb * 1024} kB\n"
+        )
+
+    def gate(self, sustain: float = 0.0, clock=time.monotonic) -> "rp.HostGate":
+        return rp.HostGate(36.0, 0.5, sustain, self.pressure, self.meminfo, clock)
+
+
+class TestHostGate(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory(prefix="host_gate_")
+        self.addCleanup(self.tmp.cleanup)
+        self.host = FakeHost(Path(self.tmp.name))
+        self.clock = [0.0]
+        self.gate = self.host.gate(sustain=30.0, clock=lambda: self.clock[0])
+
+    def _samples(self, seconds, **load):
+        self.host.set(**load)
+        for _ in range(seconds):
+            self.gate.observe()
+            self.clock[0] += 1.0
+
+    def test_the_first_container_always_starts(self):
+        self.host.set(cpu_avg10=99.0, cpu_avg60=99.0, mem_available_mb=10)
+        self.gate.observe()
+        self.assertTrue(self.gate.admits(0, 1536))
+
+    def test_a_further_container_needs_a_full_calm_window(self):
+        self._samples(29, cpu_avg10=5.0, mem_available_mb=8192)
+        self.assertFalse(self.gate.admits(1, 1536))
+        self._samples(2, cpu_avg10=5.0, mem_available_mb=8192)
+        self.assertTrue(self.gate.admits(1, 1536))
+
+    def test_a_single_dip_in_avg10_does_not_admit(self):
+        self._samples(60, cpu_avg10=50.0, cpu_avg60=50.0, mem_available_mb=8192)
+        self._samples(1, cpu_avg10=10.0, cpu_avg60=50.0, mem_available_mb=8192)
+        self.assertFalse(self.gate.admits(1, 1536))
+        self._samples(40, cpu_avg10=10.0, cpu_avg60=50.0, mem_available_mb=8192)
+        self.assertFalse(self.gate.admits(1, 1536))
+
+    def test_one_hot_sample_restarts_the_window(self):
+        self._samples(25, cpu_avg10=5.0, mem_available_mb=8192)
+        self._samples(1, cpu_avg10=36.5, mem_available_mb=8192)
+        self._samples(25, cpu_avg10=5.0, mem_available_mb=8192)
+        self.assertFalse(self.gate.admits(1, 1536))
+        self._samples(6, cpu_avg10=5.0, mem_available_mb=8192)
+        self.assertTrue(self.gate.admits(1, 1536))
+
+    def test_a_start_restarts_the_window(self):
+        self._samples(31, cpu_avg10=5.0, mem_available_mb=8192)
+        self.assertTrue(self.gate.admits(1, 1536))
+        self.assertFalse(self.gate.admits(2, 1536))
+        self._samples(31, cpu_avg10=5.0, mem_available_mb=8192)
+        self.assertTrue(self.gate.admits(2, 1536))
+
+    def test_an_estimate_above_half_the_free_memory_holds_the_next_container(self):
+        self._samples(31, cpu_avg10=5.0, mem_available_mb=3071)
+        self.assertFalse(self.gate.admits(1, 1536))
+        self._samples(1, cpu_avg10=5.0, mem_available_mb=3072)
+        self.assertTrue(self.gate.admits(1, 1536))
+
+    def test_an_unreadable_host_keeps_the_round_serial(self):
+        self.host.pressure.unlink()
+        for _ in range(60):
+            self.gate.observe()
+            self.clock[0] += 1.0
+        self.assertFalse(self.gate.admits(1, 1536))
+        self.assertTrue(self.gate.admits(0, 1536))
+
+
+class TestRun(unittest.TestCase):
+    SCHEDULER = (
+        PROJECT_ROOT / "roles" / "test-e2e-playwright" / "files" / "run_parallel.py"
+    )
+
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory(prefix="run_parallel_")
+        self.addCleanup(self.tmp.cleanup)
+        self.root = Path(self.tmp.name)
+        self.jobs = self.root / "jobs"
+        self.jobs.mkdir()
+        self.trace = self.root / "trace.log"
+        self.host = FakeHost(self.root)
+        self.tor_lock = self.root / "tor.lock"
+
+    def _job(
+        self,
+        index,
+        job_id,
+        seconds,
+        *,
+        rc=0,
+        after=(),
+        timeout=60,
+        tor=False,
+        name=None,
+    ):
+        script = (
+            "import sys,time;"
+            f"open({str(self.trace)!r},'a').write('start {job_id} %f\\n' % time.time());"
+            f"time.sleep({seconds});"
+            f"open({str(self.trace)!r},'a').write('end {job_id} %f\\n' % time.time());"
+            f"print('out-{job_id}');print('err-{job_id}',file=sys.stderr);sys.exit({rc})"
+        )
+        spec = {
+            "id": job_id,
+            "argv": [sys.executable, "-c", script],
+            "env": {"PLAYWRIGHT_TEST_MARKER": job_id},
+            "timeout": timeout,
+            "after": list(after),
+            "tor": tor,
+            "workers": 1,
+            "memory_mb": 1536,
+            "stop": [
+                sys.executable,
+                "-c",
+                f"open({str(self.root / ('stopped-' + job_id))!r},'w').write('x')",
+            ],
+            "meta": {"reports_dir": f"/reports/{job_id}"},
+        }
+        (self.jobs / (name or f"{index}-{job_id}.job.json")).write_text(
+            json.dumps(spec)
+        )
+
+    def _run(self, cap, poll=0.05, onion_cap=3):
+        return rp.run(
+            self.jobs,
+            cap,
+            poll,
+            self.host.gate(),
+            rp.TorReaders(self.tor_lock, onion_cap),
+        )
+
+    def _result(self, job_id):
+        text = (self.jobs / f"{job_id}.result.json").read_text()  # nocheck: cache-read
+        return json.loads(text)
+
+    def _trace(self):
+        text = self.trace.read_text()  # nocheck: cache-read
+        return [line.split() for line in text.splitlines()]
+
+    def _stamps(self, kind, job_id):
+        return [
+            float(stamp)
+            for event, name, stamp in self._trace()
+            if [event, name] == [kind, job_id]
+        ]
+
+    def _peak(self):
+        live, peak = 0, 0
+        events = sorted(
+            (float(stamp), 1 if kind == "start" else -1)
+            for kind, _, stamp in self._trace()
+        )
+        for _, delta in events:
+            live += delta
+            peak = max(peak, live)
+        return peak
+
+    def test_jobs_run_concurrently_within_the_cap_and_keep_their_own_results(self):
+        self._job(0, "head", 0.3)
+        for index, name in enumerate(["a", "b", "c", "d"], start=1):
+            self._job(index, name, 0.6, rc=1 if name == "b" else 0, after=["head"])
+        (self.jobs / rp.SEALED).touch()
+        began = time.monotonic()
+        self.assertEqual(self._run(2), 0)
+        elapsed = time.monotonic() - began
+        self.assertEqual(self._peak(), 2)
+        self.assertLess(elapsed, 0.3 + 4 * 0.6)
+        self.assertEqual(self._result("b")["rc"], 1)
+        for name in ("head", "a", "c", "d"):
+            self.assertEqual(self._result(name)["rc"], 0)
+            self.assertEqual(self._result(name)["stdout"].strip(), f"out-{name}")
+            self.assertEqual(self._result(name)["stderr"].strip(), f"err-{name}")
+            self.assertEqual(
+                self._result(name)["meta"], {"reports_dir": f"/reports/{name}"}
+            )
+        head_end = self._stamps("end", "head")[0]
+        for name in ("a", "b", "c", "d"):
+            self.assertGreaterEqual(self._stamps("start", name)[0], head_end)
+
+    def test_a_hot_host_runs_the_round_serially(self):
+        self.host.set(cpu_avg10=80.0, mem_available_mb=8192)
+        for index, name in enumerate(["a", "b", "c"]):
+            self._job(index, name, 0.2)
+        (self.jobs / rp.SEALED).touch()
+        self.assertEqual(self._run(3), 0)
+        self.assertEqual(self._peak(), 1)
+
+    def test_a_timeout_kills_only_its_own_job_and_says_so_in_stderr(self):
+        self._job(0, "slow", 30, timeout=0.5)
+        self._job(1, "fast", 0.2)
+        (self.jobs / rp.SEALED).touch()
+        self.assertEqual(self._run(2), 0)
+        slow = self._result("slow")
+        self.assertTrue(slow["timed_out"])
+        self.assertNotIn("rc", slow)
+        self.assertIn("timeout", slow["msg"])
+        self.assertIn(slow["msg"], slow["stderr"])
+        self.assertTrue((self.root / "stopped-slow").exists())
+        self.assertEqual(self._result("fast")["rc"], 0)
+
+    def test_jobs_streamed_before_the_seal_are_all_run(self):
+        self._job(0, "first", 0.2)
+        runner = threading.Thread(target=self._run, args=(3,))
+        runner.start()
+        time.sleep(0.4)
+        self._job(1, "second", 0.1)
+        (self.jobs / rp.SEALED).touch()
+        runner.join(timeout=10)
+        self.assertFalse(runner.is_alive())
+        self.assertEqual(self._result("first")["rc"], 0)
+        self.assertEqual(self._result("second")["rc"], 0)
+
+    def test_the_temp_file_ansible_copy_writes_is_ignored(self):
+        self._job(0, "a", 0.1)
+        self._job(1, "b", 0.1, name=".ansible_tmpk3j2x11-b.job.json")
+        (self.jobs / ".ansible_tmpq9z").write_text("{")
+        (self.jobs / rp.SEALED).touch()
+        self.assertEqual(self._run(2), 0)
+        self.assertEqual(self._result("a")["rc"], 0)
+        self.assertFalse((self.jobs / "b.result.json").exists())
+
+    def test_a_freed_slot_is_refilled_without_waiting_a_poll(self):
+        self._job(0, "a", 0.2)
+        self._job(1, "b", 0.1, after=["a"])
+        (self.jobs / rp.SEALED).touch()
+        self.assertEqual(self._run(1, poll=1.0), 0)
+        self.assertLess(
+            self._stamps("start", "b")[0] - self._stamps("end", "a")[0], 0.6
+        )
+
+    def test_a_scheduler_crash_stops_every_running_container(self):
+        self._job(0, "long", 30)
+        runner_error = []
+
+        def drive():
+            try:
+                self._run(2)
+            except Exception as exc:
+                runner_error.append(exc)
+
+        runner = threading.Thread(target=drive)
+        runner.start()
+        deadline = time.monotonic() + 5
+        while not self.trace.exists() and time.monotonic() < deadline:
+            time.sleep(0.05)
+        (self.jobs / "1-broken.job.json").write_text("{not json")
+        runner.join(timeout=30)
+        self.assertFalse(runner.is_alive())
+        self.assertEqual(len(runner_error), 1)
+        self.assertTrue((self.root / "stopped-long").exists())
+        self.assertEqual(self._stamps("end", "long"), [])
+
+    def _scheduler(self) -> subprocess.Popen:
+        return subprocess.Popen(
+            [
+                sys.executable,
+                str(self.SCHEDULER),
+                str(self.jobs),
+                "2",
+                "0.05",
+                "36",
+                "0.5",
+                "0",
+                str(self.tor_lock),
+                "3",
+            ],
+            env=os.environ.copy(),
+        )
+
+    def test_a_marker_a_killed_probe_left_in_an_earlier_round_does_not_hold_onion_runs(
+        self,
+    ):
+        self.tor_lock.with_name(self.tor_lock.name + ".recreate-pending").touch()
+        self._job(0, "onion", 0.1, tor=True)
+        (self.jobs / rp.SEALED).touch()
+        proc = self._scheduler()
+        try:
+            self.assertEqual(proc.wait(timeout=10), 0)
+        finally:
+            proc.kill()
+        self.assertEqual(self._result("onion")["rc"], 0)
+
+    def test_sigterm_to_the_scheduler_stops_its_containers(self):
+        self._job(0, "long", 30)
+        proc = self._scheduler()
+        deadline = time.monotonic() + 10
+        while not self.trace.exists() and time.monotonic() < deadline:
+            time.sleep(0.05)
+        proc.send_signal(signal.SIGTERM)
+        self.assertEqual(proc.wait(timeout=30), 128 + signal.SIGTERM)
+        self.assertTrue((self.root / "stopped-long").exists())
+
+    def test_an_onion_job_waits_while_tor_is_being_recreated(self):
+        self._job(0, "onion", 0.1, tor=True)
+        self._job(1, "clear", 0.1)
+        writer = os.open(self.tor_lock, os.O_RDWR | os.O_CREAT, 0o644)
+        fcntl.flock(writer, fcntl.LOCK_EX)
+        (self.jobs / rp.SEALED).touch()
+        runner = threading.Thread(target=self._run, args=(2,))
+        runner.start()
+        time.sleep(0.6)
+        self.assertEqual(self._stamps("start", "onion"), [])
+        self.assertEqual(len(self._stamps("end", "clear")), 1)
+        released = time.time()
+        fcntl.flock(writer, fcntl.LOCK_UN)
+        os.close(writer)
+        runner.join(timeout=10)
+        self.assertGreaterEqual(self._stamps("start", "onion")[0], released)
+
+    def test_onion_jobs_share_tor_with_each_other(self):
+        self._job(0, "onion-a", 0.5, tor=True)
+        self._job(1, "onion-b", 0.5, tor=True)
+        (self.jobs / rp.SEALED).touch()
+        self.assertEqual(self._run(2), 0)
+        self.assertEqual(self._peak(), 2)
+
+    def test_onion_jobs_stay_within_the_tor_cap(self):
+        for index, name in enumerate(["onion-a", "onion-b", "onion-c"]):
+            self._job(index, name, 0.4, tor=True)
+        self._job(3, "clear", 0.4)
+        (self.jobs / rp.SEALED).touch()
+        self.assertEqual(self._run(4, onion_cap=1), 0)
+        onion = sorted(
+            (self._stamps("start", name)[0], self._stamps("end", name)[0])
+            for name in ("onion-a", "onion-b", "onion-c")
+        )
+        for (_, previous_end), (next_start, _) in itertools.pairwise(onion):
+            self.assertGreaterEqual(next_start, previous_end)
+        self.assertLess(self._stamps("start", "clear")[0], onion[0][1])
+
+    def test_a_dropped_identity_provider_runs_none_of_its_dependents_here(self):
+        head = "web-app-keycloak"
+        self._job(1, "a", 0.1, after=[head])
+        self._job(2, "b", 0.1, after=[head])
+        self._job(3, "c", 0.1, after=[head, "b"])
+        (self.jobs / rp.SEALED).touch()
+        self.assertEqual(self._run(3), 0)
+        self.assertFalse(self.trace.exists())
+        self.assertEqual(self._result("a")["deferred"], [head])
+        self.assertEqual(self._result("c")["deferred"], [head, "b"])
+        self.assertNotIn("rc", self._result("b"))
+
+    def test_no_onion_job_starts_while_a_recreate_is_pending(self):
+        self._job(0, "onion", 0.1, tor=True)
+        pending = self.tor_lock.with_name(self.tor_lock.name + ".recreate-pending")
+        pending.touch()
+        (self.jobs / rp.SEALED).touch()
+        runner = threading.Thread(target=self._run, args=(2,))
+        runner.start()
+        time.sleep(0.5)
+        self.assertFalse(self.trace.exists())
+        cleared = time.time()
+        pending.unlink()
+        runner.join(timeout=10)
+        self.assertGreaterEqual(self._stamps("start", "onion")[0], cleared)
+
+
+class TestRunSignals(unittest.TestCase):
+    def test_retries_flaky_tests_and_crashed_pages_are_counted(self):
+        output = (
+            "  ✘  3 tests/a.js:1:1 › login (retry #1) (2.1m)\n"
+            "  ✓  4 tests/a.js:1:1 › login (retry #2) (1.0m)\n"
+            "Error: page.goto: Page crashed\n"
+            "  1 flaky\n"
+            "  12 passed (9.0m)\n"
+        )
+        self.assertEqual(
+            rp.run_signals(output),
+            {"retries": 2, "flaky": 1, "crashes": 1, "crash_kinds": "page crashed 1"},
+        )
+
+    def test_a_killed_browser_or_worker_that_a_retry_hid_is_still_named(self):
+        output = (
+            "  ✘  1 tests/a.js:1:1 › login (2.1m)\n"
+            "Error: browserType.launch: Browser closed.\n"
+            "Error: page.click: Target page, context or browser has been closed\n"
+            "Error: worker process exited unexpectedly (code=null, signal=SIGKILL)\n"
+            "  ✓  2 tests/a.js:1:1 › login (retry #1) (1.0m)\n"
+            "  1 flaky\n"
+        )
+        self.assertEqual(
+            rp.run_signals(output),
+            {
+                "retries": 1,
+                "flaky": 1,
+                "crashes": 3,
+                "crash_kinds": "browser closed (OOM or timeout) 2, worker killed 1",
+            },
+        )
+
+    def test_a_clean_run_reports_nothing(self):
+        self.assertEqual(
+            rp.run_signals("  12 passed (9.0m)\n"),
+            {"retries": 0, "flaky": 0, "crashes": 0, "crash_kinds": ""},
+        )
+
+
+class TestTorRecreateWaitsForOnionRuns(unittest.TestCase):
+    """The probe script recreates Tor only once no onion run holds the lock."""
+
+    SCRIPT = (
+        PROJECT_ROOT
+        / "roles"
+        / "test-e2e-playwright"
+        / "files"
+        / "wait_onion_reachable.sh"
+    )
+
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory(prefix="tor_lock_")
+        self.addCleanup(self.tmp.cleanup)
+        self.root = Path(self.tmp.name)
+        self.bin = self.root / "bin"
+        self.bin.mkdir()
+        self.marker = self.root / "recreated"
+        self.answers = self.root / "onion-answers"
+        self.lock = self.root / "tor.lock"
+        self.pending = self.root / "tor.lock.recreate-pending"
+        (self.bin / "curl").write_text(
+            f"#!/bin/sh\nif [ -e {self.answers} ]; then printf 200; exit 0; fi\nexit 7\n"
+        )
+        (self.bin / "docker").write_text(
+            "#!/bin/sh\n"
+            f"if [ \"$2\" = update ]; then python3 -c \"import time; open('{self.marker}','w').write(str(time.time()))\"; fi\n"
+            "exit 0\n"
+        )
+        for stub in ("curl", "docker"):
+            (self.bin / stub).chmod(0o755)
+
+    def start(self, script: Path, lock: str) -> subprocess.Popen:
+        """Start *script* with one attempt, a 2 s delay and a 1 s max time.
+
+        Args:
+            script: the probe script under test.
+            lock: the Tor lock argument, empty for the serial run.
+
+        Returns:
+            The running script.
+        """
+        return subprocess.Popen(
+            [
+                "bash",
+                str(script),
+                "127.0.0.1:9050",
+                "x.onion",
+                "1",
+                "2",
+                "1",
+                str(self.root),
+                lock,
+            ],
+            env={**os.environ, "PATH": f"{self.bin}:{os.environ['PATH']}"},
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            text=True,
+        )
+
+    def run_script(
+        self, script: Path, hold: float, answer_at: float | None = None
+    ) -> tuple[float | None, int, str, bool]:
+        """Run *script* in the parallel mode while a reader holds the Tor lock.
+
+        Args:
+            script: the probe script under test.
+            hold: seconds the shared lock is held after the script starts.
+            answer_at: second at which the onion starts to answer, None for never.
+
+        Returns:
+            Recreate time minus release time (None when it never recreated),
+            the exit code, stderr, and whether the pending marker existed
+            while the script waited.
+        """
+        reader = os.open(self.lock, os.O_RDONLY | os.O_CREAT, 0o644)
+        fcntl.flock(reader, fcntl.LOCK_SH)
+        proc = self.start(script, str(self.lock))
+        began = time.monotonic()
+        announced, released = False, None
+        events = [(min(hold, 2.5), "look"), (hold, "release")]
+        if answer_at is not None:
+            events.append((answer_at, "answer"))
+        for at, event in sorted(events):
+            time.sleep(max(0.0, at - (time.monotonic() - began)))
+            if event == "look":
+                announced = self.pending.exists()
+            elif event == "answer":
+                self.answers.touch()
+            else:
+                released = time.time()
+                fcntl.flock(reader, fcntl.LOCK_UN)
+                os.close(reader)
+        _, stderr = proc.communicate(timeout=20)
+        recreated = None
+        if self.marker.exists():
+            recreated = float(self.marker.read_text()) - released  # nocheck: cache-read
+        self.assertFalse(self.pending.exists())
+        return recreated, proc.returncode, stderr, announced
+
+    def test_the_recreate_waits_until_the_onion_run_releases_tor(self):
+        recreated, rc, _, announced = self.run_script(self.SCRIPT, hold=3.0)
+        self.assertIsNotNone(recreated)
+        self.assertGreaterEqual(recreated, 0)
+        self.assertTrue(announced)
+        self.assertEqual(rc, 1)
+
+    def test_onion_runs_outlasting_the_wait_skip_the_recreate_and_a_recovered_onion_passes(
+        self,
+    ):
+        recreated, rc, stderr, _ = self.run_script(self.SCRIPT, hold=6.0, answer_at=3.0)
+        self.assertIsNone(recreated)
+        self.assertEqual(rc, 0)
+        self.assertIn(
+            "skipped recreating Tor for x.onion: onion Playwright runs held", stderr
+        )
+
+    def test_onion_runs_outlasting_the_wait_fail_only_this_probe_when_the_onion_stays_silent(
+        self,
+    ):
+        recreated, rc, stderr, _ = self.run_script(self.SCRIPT, hold=6.0)
+        self.assertIsNone(recreated)
+        self.assertEqual(rc, 1)
+        self.assertIn("skipped recreating Tor", stderr)
+
+    def test_the_serial_run_recreates_at_once_without_touching_the_lock(self):
+        began = time.time()
+        proc = self.start(self.SCRIPT, "")
+        stdout, _ = proc.communicate(timeout=20)
+        self.assertEqual(proc.returncode, 1)
+        recreated = float(self.marker.read_text())  # nocheck: cache-read
+        self.assertLess(recreated - began, 2 + 1.5)
+        self.assertIn("recreating Tor for fresh circuits", stdout)
+        self.assertNotIn("waiting up to", stdout)
+        self.assertFalse(self.lock.exists())
+        self.assertFalse(self.pending.exists())

tests/unit/python/roles/test-e2e-playwright/lookup_plugins/test_playwright_parallel.py

@@ -0,0 +1,180 @@
+import importlib.util
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from ansible.errors import AnsibleError
+from ansible.parsing.dataloader import DataLoader
+from ansible.template import Templar
+
+from plugins.lookup.role_tests import LookupModule as RoleTestsLookup
+from utils.cache.yaml import dump_yaml, load_yaml_any
+from utils.roles.mapping import ROLE_FILE_META_TESTS, ROLE_FILE_META_VARIANTS
+
+from . import PROJECT_ROOT
+
+
+def _load_module():
+    path = (
+        PROJECT_ROOT
+        / "roles"
+        / "test-e2e-playwright"
+        / "lookup_plugins"
+        / "playwright_parallel.py"
+    )
+    spec = importlib.util.spec_from_file_location("playwright_parallel_lookup", path)
+    module = importlib.util.module_from_spec(spec)
+    assert spec and spec.loader
+    sys.modules["playwright_parallel_lookup"] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+_plugin = _load_module()
+
+
+def declared_caps() -> dict:
+    """Collect every role's ``playwright.parallel`` entry.
+
+    Returns:
+        The declared value per role name, for the roles that declare one.
+    """
+    found = {}
+    for role in sorted(
+        path for path in (PROJECT_ROOT / "roles").iterdir() if path.is_dir()
+    ):
+        playwright = (
+            load_yaml_any(str(role / ROLE_FILE_META_TESTS), default_if_missing={}) or {}
+        ).get("playwright") or {}
+        if "parallel" in playwright:
+            found[role.name] = playwright["parallel"]
+    return found
+
+
+class TestRowCap(unittest.TestCase):
+    def test_an_app_without_the_list_runs_one_container(self):
+        self.assertEqual(_plugin.row_cap("web-app-x", None, 0), 1)
+
+    def test_the_entry_of_the_round_variant_is_the_cap(self):
+        self.assertEqual(_plugin.row_cap("web-app-x", [3, 1, 2], 2), 2)
+
+    def test_a_round_beyond_the_declared_variants_runs_serially(self):
+        self.assertEqual(_plugin.row_cap("web-app-x", [3], 1), 1)
+        self.assertEqual(_plugin.row_cap("web-app-x", [3], 3), 1)
+
+    def test_a_non_positive_or_non_integer_entry_is_rejected(self):
+        for caps in ([0], [-1], ["3"], [True], 3, {"0": 3}):
+            with self.subTest(caps=caps), self.assertRaises(AnsibleError):
+                _plugin.validate_caps("web-app-x", caps)
+
+
+class TestPlaywrightParallelLookup(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory(prefix="playwright_parallel_")
+        self.addCleanup(self.tmp.cleanup)
+        self.playbook_dir = Path(self.tmp.name)
+        for role in ("web-app-a", "web-app-b", "web-app-c"):
+            (self.playbook_dir / "roles" / role / "meta").mkdir(parents=True)
+        dump_yaml(
+            self.playbook_dir / "roles" / "web-app-a" / ROLE_FILE_META_TESTS,
+            {"playwright": {"parallel": [3, 1]}},
+        )
+        dump_yaml(
+            self.playbook_dir / "roles" / "web-app-b" / ROLE_FILE_META_TESTS,
+            {"playwright": {"parallel": [2, 4]}},
+        )
+
+    def _cap(self, primary_apps, variant, playbook_dir=None) -> int:
+        variables = {
+            "playbook_dir": str(playbook_dir or self.playbook_dir),
+            "PRIMARY_APPS": primary_apps,
+            "VARIANT_INDEX": variant,
+        }
+        templar = Templar(loader=DataLoader(), variables=variables)
+        lookup = _plugin.LookupModule(loader=DataLoader(), templar=templar)
+        role_tests = RoleTestsLookup(loader=DataLoader(), templar=templar)
+        with patch.object(_plugin, "lookup_loader") as loader:
+            loader.get.return_value = role_tests
+            return lookup.run([], variables=variables)[0]
+
+    def test_a_round_without_primary_apps_runs_one_container(self):
+        self.assertEqual(self._cap([], 0), 1)
+
+    def test_the_primary_app_row_sets_the_cap(self):
+        self.assertEqual(self._cap(["web-app-a"], 0), 3)
+        self.assertEqual(self._cap(["web-app-a"], 1), 1)
+
+    def test_a_primary_app_that_did_not_opt_in_keeps_the_round_serial(self):
+        self.assertEqual(self._cap(["web-app-a", "web-app-c"], 0), 1)
+
+    def test_several_primary_apps_take_the_smallest_cap(self):
+        self.assertEqual(self._cap(["web-app-a", "web-app-b"], 0), 2)
+        self.assertEqual(self._cap(["web-app-a", "web-app-b"], 1), 1)
+
+    def test_the_cli_string_form_of_primary_apps_opts_the_row_in(self):
+        self.assertEqual(self._cap('["web-app-a"]', "0"), 3)
+        self.assertEqual(self._cap('["web-app-wordpress"]', "0", PROJECT_ROOT), 3)
+        self.assertEqual(self._cap("[]", "0"), 1)
+
+    def test_a_primary_apps_value_that_is_no_list_warns_and_keeps_the_round_serial(
+        self,
+    ):
+        for value in ("web-app-a", '"web-app-a"', '{"a": 1}', "[1]"):
+            with self.subTest(value=value), patch.object(_plugin, "Display") as display:
+                self.assertEqual(self._cap(value, "0"), 1)
+                display.return_value.warning.assert_called_once()
+
+    def test_a_multi_app_round_beyond_a_primary_variants_runs_serially(self):
+        for variant in ("1", "2", "3"):
+            with self.subTest(variant=variant):
+                self.assertEqual(
+                    self._cap(
+                        '["web-app-wordpress","web-app-nextcloud"]',
+                        variant,
+                        PROJECT_ROOT,
+                    ),
+                    1,
+                )
+        self.assertEqual(self._cap('["web-app-a","web-app-b"]', "5"), 1)
+
+    def test_a_malformed_cap_list_warns_and_keeps_the_round_serial(self):
+        dump_yaml(
+            self.playbook_dir / "roles" / "web-app-c" / ROLE_FILE_META_TESTS,
+            {"playwright": {"parallel": "3"}},
+        )
+        with patch.object(_plugin, "Display") as display:
+            self.assertEqual(self._cap(["web-app-c"], 0), 1)
+            display.return_value.warning.assert_called_once()
+
+    def test_every_declared_cap_list_is_valid_and_covers_every_variant(self):
+        for role, caps in declared_caps().items():
+            with self.subTest(role=role):
+                _plugin.validate_caps(role, caps)
+                variants = load_yaml_any(
+                    str(PROJECT_ROOT / "roles" / role / ROLE_FILE_META_VARIANTS),
+                    default_if_missing=[{}],
+                )
+                self.assertEqual(len(caps), len(variants))
+
+    def test_the_sizing_terms_follow_playwright_workers(self):
+        lookup = _plugin.LookupModule(
+            loader=DataLoader(), templar=Templar(loader=DataLoader())
+        )
+        self.assertEqual(lookup.run(["memory_mb"], workers="1"), [1536])
+        self.assertEqual(lookup.run(["memory_mb"], workers=2), [3072])
+        self.assertEqual(lookup.run(["ram_fraction"]), [0.5])
+        self.assertEqual(lookup.run(["onion_cap"]), [3])
+
+    def test_wordpress_opts_in_its_first_row_only(self):
+        self.assertEqual(self._cap(["web-app-wordpress"], 0, PROJECT_ROOT), 3)
+        self.assertEqual(self._cap(["web-app-wordpress"], 1, PROJECT_ROOT), 1)
+        self.assertEqual(self._cap(["web-app-wordpress"], 2, PROJECT_ROOT), 1)
+
+    def test_every_other_role_keeps_the_serial_runner(self):
+        self.assertEqual(sorted(declared_caps()), ["web-app-wordpress"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/python/roles/test-e2e-playwright/test_parallel_orchestration.py

@@ -0,0 +1,514 @@
+import base64
+import importlib.util
+import json
+import re
+import shlex
+import sys
+import tempfile
+import time
+import unittest
+from pathlib import Path
+
+import jinja2
+from ansible.parsing.dataloader import DataLoader
+from ansible.plugins.filter.core import FilterModule
+from ansible.plugins.test.core import TestModule
+from ansible.template import Templar
+from jinja2.nativetypes import NativeEnvironment
+
+from plugins.lookup.role_tests import LookupModule as RoleTestsLookup
+from plugins.lookup.timeout import scaled_timeout
+from utils.cache.files import read_text
+from utils.cache.yaml import load_yaml_any
+from utils.roles.mapping import (
+    ROLE_FILE_DEFAULTS_MAIN,
+    ROLE_FILE_META_NETWORKS,
+    ROLE_FILE_VARS_MAIN,
+)
+
+from . import PROJECT_ROOT, ROLE, role_task
+
+AFTER_STAGING = (
+    "🔏 Seal the job list so the scheduler drains and exits",
+    "⏳ Wait for the scheduler to drain the job list",
+    "🧹 Remove any handed-over container the scheduler did not stop",
+    "📥 Read each handed-over run's job and result",
+    "🧭 Leave the runs whose dependency never ran here to the serial runner",
+    "📝 Report and archive each handed-over run",
+    "🛑 Fail when the scheduler itself did not finish cleanly",
+)
+BUDGET_INPUTS = (
+    "TEST_E2E_PLAYWRIGHT_ONION_RUN_TIMEOUT",
+    "test_e2e_playwright_run_timeout",
+    "TEST_E2E_PLAYWRIGHT_ONION_PROBE_RETRIES",
+    "TEST_E2E_PLAYWRIGHT_ONION_PROBE_DELAY",
+    "TEST_E2E_PLAYWRIGHT_ONION_PROBE_MAX_TIME",
+    "test_e2e_playwright_wait_retries",
+    "test_e2e_playwright_wait_delay",
+)
+FIXTURE_INPUTS = dict(zip(BUDGET_INPUTS, (5400, 3600, 40, 15, 120, 30, 5), strict=True))
+WORDPRESS0_APPS = (
+    "web-app-keycloak",
+    "web-app-hermes",
+    "web-app-matomo",
+    "web-app-openclaw",
+    "web-app-weblate",
+    "web-svc-cdn",
+    "web-svc-design",
+    "web-svc-logout",
+    "web-svc-mirror",
+    "web-svc-simpleicons",
+    "web-svc-translate",
+    "web-app-dashboard",
+    "web-app-discourse",
+    "web-app-stalwart",
+    "web-app-jitsi",
+    "web-app-matrix",
+    "web-app-prometheus",
+    "web-app-wordpress",
+)
+
+
+def budget_expression() -> str:
+    """Read ``TEST_E2E_PLAYWRIGHT_PARALLEL_BUDGET`` as the role writes it.
+
+    Returns:
+        The Jinja expression.
+    """
+    return load_yaml_any(str(ROLE / ROLE_FILE_VARS_MAIN), default_if_missing={})[
+        "TEST_E2E_PLAYWRIGHT_PARALLEL_BUDGET"
+    ]
+
+
+def render_budget(
+    expression, apps, factor, inputs, role_timeout, networks, handover
+) -> int:
+    """Render the scheduler's async ceiling for one round.
+
+    Args:
+        expression: the ceiling expression to render.
+        apps: ``TEST_E2E_PLAYWRIGHT_APPS`` of the round.
+        factor: the ``TIMEOUT_FACTOR`` the ``timeout`` lookup scales by.
+        inputs: the role variables the expression reads, by name.
+        role_timeout: an app's ``playwright.timeout``, ``''`` when unset.
+        networks: the networks an app's domain has siblings on.
+        handover: whether an app is pinned to its own domain.
+
+    Returns:
+        The ceiling in seconds.
+    """
+
+    def lookup(name, *terms, **kwargs):
+        if name == "role_tests":
+            return role_timeout(terms[0])
+        if name == "timeout":
+            return scaled_timeout(terms[0], factor)
+        if name == "domain":
+            return f"clearnet:{terms[0]}"
+        if name == "config":
+            return handover(terms[0])
+        if name == "network_siblings":
+            app = terms[0].split(":", 1)[1]
+            return [f"{network}:{app}" for network in networks(app)]
+        raise AssertionError(name)
+
+    env = jinja2.Environment(
+        autoescape=jinja2.select_autoescape(default_for_string=False)
+    )
+    env.globals["lookup"] = lookup
+    env.filters["network_of"] = lambda domain: domain.split(":", 1)[0]
+    return int(
+        env.from_string(expression).render(
+            TEST_E2E_PLAYWRIGHT_APPS=list(apps), **inputs
+        )
+    )
+
+
+def conditional(expression: str, **variables) -> bool:
+    """Evaluate a task condition the way Ansible does.
+
+    Args:
+        expression: the ``when`` or ``failed_when`` text.
+        **variables: the variables in scope.
+
+    Returns:
+        The condition's truth.
+    """
+    env = jinja2.Environment(
+        autoescape=jinja2.select_autoescape(default_for_string=False)
+    )
+    env.filters.update(FilterModule().filters())
+    env.tests.update(TestModule().tests())
+    return (
+        env.from_string("{{ (" + expression + ") | bool }}").render(**variables)
+        == "True"
+    )
+
+
+def native(expression: str, **variables):
+    """Evaluate a task value the way Ansible does, keeping its native type.
+
+    Args:
+        expression: the templated value.
+        **variables: the variables in scope.
+
+    Returns:
+        The rendered value.
+    """
+    env = NativeEnvironment()
+    env.filters.update(FilterModule().filters())
+    env.tests.update(TestModule().tests())
+    return env.from_string(expression).render(**variables)
+
+
+def scheduler_module():
+    """Load ``run_parallel.py``, which the role ships as a plain file.
+
+    Returns:
+        The scheduler module.
+    """
+    spec = importlib.util.spec_from_file_location(
+        "playwright_run_parallel_flow", ROLE / "files" / "run_parallel.py"
+    )
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[spec.name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+rp = scheduler_module()
+
+
+def encoded(path: Path) -> str:
+    """Return a file the way ``ansible.builtin.slurp`` hands it back.
+
+    Args:
+        path: a file the test wrote.
+
+    Returns:
+        The base64 content.
+    """
+    data = path.read_bytes()  # nocheck: cache-read
+    return base64.b64encode(data).decode()
+
+
+class TestParallelOrchestration(unittest.TestCase):
+    def test_a_dropped_identity_provider_sends_its_dependents_to_a_serial_first_pass(
+        self,
+    ):
+        head = "web-app-keycloak"
+        apps = [head, "web-app-a", "web-app-b", "web-app-c"]
+        with tempfile.TemporaryDirectory() as tmp:
+            jobs, trace = Path(tmp) / "jobs", Path(tmp) / "trace"
+            jobs.mkdir()
+            names = []
+            for index, app in enumerate(apps[1:], start=1):
+                names.append(f"{index}-{app}.job.json")
+                spec = {
+                    "id": app,
+                    "argv": [sys.executable, "-c", f"open({str(trace)!r}, 'a')"],
+                    "env": {},
+                    "timeout": 30,
+                    "after": [head],
+                    "tor": False,
+                    "workers": 1,
+                    "memory_mb": 1,
+                    "stop": ["true"],
+                    "meta": {"reports_dir": tmp},
+                }
+                (jobs / names[-1]).write_text(json.dumps(spec))
+            (jobs / "sealed").touch()
+            gate = rp.HostGate(36.0, 0.5, 0.0, jobs / "-", jobs / "-", time.monotonic)
+            tor = rp.TorReaders(Path(tmp) / "tor.lock", 3)
+            self.assertEqual(rp.run(jobs, 3, 0.05, gate, tor), 0)
+            self.assertFalse(trace.exists())
+            reads = [
+                {"playwright_read": name, "content": encoded(jobs / name)}
+                for app, job in zip(apps[1:], names, strict=True)
+                for name in (job, f"{app}.result.json")
+            ]
+        leave = role_task("03_run_pass.yml", "📝 Leave")["ansible.builtin.set_fact"]
+        serial = native(
+            leave["_playwright_serial_roles"],
+            _playwright_serial_roles=[],
+            application_id=head,
+        )
+        defer = role_task("01_setup/parallel.yml", "🧭")["ansible.builtin.set_fact"]
+        serial = native(
+            defer["_playwright_serial_roles"],
+            _playwright_serial_roles=serial,
+            _playwright_reads={"results": reads},
+        )
+        report = role_task("01_setup/parallel.yml", "📝 Report")
+        for read in reads[1::2]:
+            run = json.loads(base64.b64decode(read["content"]))
+            self.assertFalse(conditional(report["when"], _playwright_run=run))
+        first_pass = role_task("01_setup/orchestrate.yml", "▶️ Run the roles left")
+        self.assertEqual(
+            native(
+                first_pass["loop"],
+                TEST_E2E_PLAYWRIGHT_APPS=apps,
+                _playwright_serial_roles=serial,
+            ),
+            apps,
+        )
+        self.assertEqual(native(first_pass["loop"], TEST_E2E_PLAYWRIGHT_APPS=apps), [])
+        self.assertFalse(first_pass["vars"]["_playwright_dispatch"])
+        orchestrate = [
+            task["name"]
+            for task in load_yaml_any(
+                str(ROLE / "tasks" / "01_setup" / "orchestrate.yml"),
+                default_if_missing=[],
+            )
+        ]
+        parallel_at = next(i for i, n in enumerate(orchestrate) if n.startswith("🔀"))
+        self.assertEqual(orchestrate[parallel_at + 1], first_pass["name"])
+        self.assertEqual(
+            [n.split()[1] for n in orchestrate[parallel_at + 2 : parallel_at + 4]],
+            ["Stash", "Retry"],
+        )
+
+    def test_a_failed_staging_still_waits_reaps_and_reports_the_handed_over_runs(self):
+        staging = role_task("01_setup/parallel.yml", "📦")
+        self.assertEqual([t["name"] for t in staging["always"]], list(AFTER_STAGING))
+        include = role_task("01_setup/orchestrate.yml", "🔀")
+        self.assertEqual(
+            include["ansible.builtin.include_tasks"], "01_setup/parallel.yml"
+        )
+        self.assertEqual(include["when"], "TEST_E2E_PLAYWRIGHT_PARALLEL | int > 1")
+
+    def test_the_ceiling_counts_probe_passes_only_where_an_onion_can_be_picked(self):
+        timeouts = {"web-app-docs": 9000}
+        networks = {
+            "web-app-docs": ["clearnet", "tor"],
+            "web-app-x": ["clearnet"],
+            "web-app-h": ["clearnet", "tor"],
+        }
+        onion_wait = 2 * 40 * (120 + 15) + 40 * 15
+        for factor in (1, 2):
+            with self.subTest(factor=factor):
+                self.assertEqual(
+                    render_budget(
+                        budget_expression(),
+                        ["web-app-docs", "web-app-x", "web-app-h"],
+                        factor,
+                        FIXTURE_INPUTS,
+                        lambda app: timeouts.get(app, ""),
+                        networks.get,
+                        lambda app: app == "web-app-h",
+                    ),
+                    (factor * 9000 + onion_wait) + 2 * (factor * 3600 + 30 * 5),
+                )
+
+    def test_the_real_wordpress0_ceiling_never_undercuts_the_ci_step_timeout(self):
+        role_vars = {
+            **load_yaml_any(str(ROLE / ROLE_FILE_DEFAULTS_MAIN), default_if_missing={}),
+            **load_yaml_any(str(ROLE / ROLE_FILE_VARS_MAIN), default_if_missing={}),
+        }
+        variables = {"playbook_dir": str(PROJECT_ROOT)}
+        role_tests = RoleTestsLookup(
+            loader=DataLoader(),
+            templar=Templar(loader=DataLoader(), variables=variables),
+        )
+        factor = load_yaml_any(
+            str(PROJECT_ROOT / "group_vars" / "all" / "17_resource.yml"),
+            default_if_missing={},
+        )["TIMEOUT_FACTOR"]
+        ceiling = render_budget(
+            budget_expression(),
+            WORDPRESS0_APPS,
+            factor,
+            {name: role_vars[name] for name in BUDGET_INPUTS},
+            lambda app: role_tests.run(
+                [app, "playwright.timeout"], variables=variables, default=""
+            )[0],
+            lambda app: ["clearnet", "tor"],
+            lambda app: bool(
+                (
+                    load_yaml_any(
+                        str(PROJECT_ROOT / "roles" / app / ROLE_FILE_META_NETWORKS),
+                        default_if_missing={},
+                    )
+                    or {}
+                )
+                .get("reachability", {})
+                .get("handover", False)
+            ),
+        )
+        self.assertEqual(ceiling, 18 * (5400 + 2 * 40 * (120 + 15) + 40 * 15))
+        workflow = load_yaml_any(
+            str(PROJECT_ROOT / ".github" / "workflows" / "call-test-deploy.yml"),
+            default_if_missing={},
+        )
+        drill = next(
+            step
+            for job in workflow["jobs"].values()
+            for step in job.get("steps", [])
+            if str(step.get("name", "")).startswith("🚀 Drill")
+        )
+        self.assertGreater(ceiling, int(drill["timeout-minutes"]) * 60)
+
+    def test_each_parallel_run_keeps_output_result_and_retry_count_with_its_reports(
+        self,
+    ):
+        line = role_task("04_collect.yml", "⏱️")
+        for signal in ("retries", "flaky", "crashes", "crash_kinds"):
+            self.assertIn(
+                f"_playwright_run.{signal}", line["ansible.builtin.debug"]["msg"]
+            )
+        keep = role_task("04_collect.yml", "📋")
+        self.assertEqual(keep["loop"], ["result.json", "stdout", "stderr"])
+        self.assertTrue(keep["ansible.builtin.copy"]["dest"].endswith("/run-parallel/"))
+        tasks = load_yaml_any(
+            str(ROLE / "tasks" / "04_collect.yml"), default_if_missing=[]
+        )
+        holder = next(
+            i for i, t in enumerate(tasks) if keep in [t, *t.get("block", [])]
+        )
+        archive = next(i for i, t in enumerate(tasks) if t["name"].startswith("🗄️"))
+        self.assertLess(holder, archive)
+        script = read_text(str(ROLE / "files" / "archive_playwright_reports.sh"))
+        self.assertIn(
+            "for item in playwright-junit.xml playwright-report test-results run-parallel; do",
+            script,
+        )
+
+    def test_only_the_parallel_run_hands_the_probe_its_tor_lock(self):
+        probe = role_task("03_run_pass.yml", "⏳ Wait until the onion")
+        self.assertEqual(probe["when"], "TEST_E2E_PLAYWRIGHT_IS_ONION | bool")
+        env = jinja2.Environment(
+            autoescape=jinja2.select_autoescape(default_for_string=False)
+        )
+        env.filters.update(FilterModule().filters())
+        env.globals["lookup"] = lambda name, *terms: {
+            "tor_socks": "127.0.0.1:9050",
+            "container": "/opt/tor",
+        }[name]
+        for dispatch, lock in ((False, ""), (True, "/stage/tor.lock")):
+            with self.subTest(dispatch=dispatch):
+                argv = shlex.split(
+                    env.from_string(probe["ansible.builtin.script"]).render(
+                        role_path="/role",
+                        _playwright_pass_domain="x.onion",
+                        _playwright_dispatch=dispatch,
+                        TEST_E2E_PLAYWRIGHT_ONION_PROBE_RETRIES=40,
+                        TEST_E2E_PLAYWRIGHT_ONION_PROBE_DELAY=15,
+                        TEST_E2E_PLAYWRIGHT_ONION_PROBE_MAX_TIME=120,
+                        TEST_E2E_PLAYWRIGHT_TOR_LOCK="/stage/tor.lock",
+                    )
+                )
+                self.assertEqual(
+                    argv[1:],
+                    ["127.0.0.1:9050", "x.onion", "40", "15", "120", "/opt/tor", lock],
+                )
+
+    def test_an_unreachable_onion_fails_the_serial_run_but_only_its_own_app_in_parallel(
+        self,
+    ):
+        probe = role_task("03_run_pass.yml", "⏳ Wait until the onion")
+        queue = role_task("03_run_pass.yml", "📮")
+        remember = role_task("03_run_pass.yml", "📝 Remember")
+        leave = role_task("03_run_pass.yml", "📝 Leave")
+        skipped = {"skipped": True, "changed": False}
+        self.assertTrue(
+            conditional(
+                probe["failed_when"],
+                _playwright_probe={"rc": 1},
+                _playwright_dispatch=False,
+            )
+        )
+        self.assertFalse(
+            conditional(
+                probe["failed_when"],
+                _playwright_probe={"rc": 0},
+                _playwright_dispatch=False,
+            )
+        )
+        self.assertFalse(
+            conditional(
+                probe["failed_when"],
+                _playwright_probe={"rc": 1},
+                _playwright_dispatch=True,
+            )
+        )
+        for result, queued in (
+            (skipped, True),
+            ({"rc": 0}, True),
+            ({"rc": 1}, False),
+            ({"msg": "transfer failed"}, False),
+        ):
+            with self.subTest(result=result):
+                self.assertEqual(
+                    conditional(queue["when"], _playwright_probe=result), queued
+                )
+                written = {"dest": "/jobs/3-x.job.json"} if queued else skipped
+                self.assertEqual(
+                    conditional(remember["when"], _playwright_job_written=written),
+                    queued,
+                )
+                self.assertEqual(
+                    conditional(leave["when"], _playwright_job_written=written),
+                    not queued,
+                )
+        self.assertIn("_playwright_serial_roles", leave["ansible.builtin.set_fact"])
+
+    def test_the_seal_keeps_its_times_and_the_async_states_why_it_is_safe(self):
+        seal = role_task("01_setup/parallel.yml", "🔏")["ansible.builtin.file"]
+        self.assertEqual(
+            (seal["modification_time"], seal["access_time"]), ("preserve", "preserve")
+        )
+        self.assertRegex(
+            read_text(str(ROLE / "tasks" / "01_setup" / "parallel.yml")),
+            r"#\s*rationale:\s*async;\s*\S",
+        )
+
+
+class TestParallelContainerFlags(unittest.TestCase):
+    """The flags only the parallel run adds must not shift the image argument."""
+
+    def _container_module(self):
+        path = (
+            PROJECT_ROOT
+            / "roles"
+            / "sys-svc-container"
+            / "files"
+            / "python"
+            / "container.py"
+        )
+        spec = importlib.util.spec_from_file_location("container_wrapper", path)
+        module = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(module)
+        return module
+
+    def _flags(self):
+        return role_task("03_run_pass.yml", "📮")["vars"]["_playwright_parallel_flags"]
+
+    def test_every_parallel_flag_carries_its_value_in_one_token(self):
+        flags = self._flags()
+        self.assertEqual(len(flags), 2)
+        for flag in flags:
+            self.assertRegex(re.sub(r"\{\{.*?\}\}", "X", flag), r"^--[a-z-]+=\S+$")
+
+    def test_no_job_gets_a_memory_cap_that_could_kill_it_running_alone(self):
+        self.assertFalse(
+            any(flag.startswith(("--memory", "-m")) for flag in self._flags())
+        )
+        self.assertTrue(
+            any(flag.startswith("--oom-score-adj=") for flag in self._flags())
+        )
+
+    def test_the_wrapper_still_finds_the_image(self):
+        container = self._container_module()
+        argv = [
+            "--rm",
+            "--name=infinito-playwright-x",
+            "--oom-score-adj=1000",
+            "image:tag",
+            "/bin/bash",
+        ]
+        _, image_and_args = container.split_docker_run_argv(argv)
+        self.assertEqual(image_and_args[0], "image:tag")
+
+
+if __name__ == "__main__":
+    unittest.main()