feat(ci): run L0/L1 offline checks, mutation suites and yamllint in bounded parallel pools

- conformance/lib/run_l1_offline.py: checks run in a thread pool (--jobs / CONF_OFFLINE_JOBS,
  default min(8, usable CPUs)); results consumed in submission order, so log lines, JUnit,
  step summary and the 0/1/3 aggregation are unchanged. Exports a smaller MUTATION_JOBS to
  the children when parallel, so nested pools do not oversubscribe.
- keycloak-{passkey,client-secret}-reconciler/mutations.py: anchors checked first (serial),
  then the --selftest subprocesses run in a pool (MUTATION_JOBS, default min(8, CPUs));
  verdicts reported in MUTATIONS order. Output byte-identical to the serial loop.
- ci/lint.sh: yamllint over LINT_JOBS contiguous slices concurrently, per-slice output
  buffered and printed in order; stage rc = highest slice rc.

Co-Authored-By: Claude Opus 5.5 <[email protected]>

ci/lint.sh

@@ -25,8 +25,44 @@ if [ "${#files[@]}" -eq 0 ]; then
   exit 0
 fi
 
-yamllint -c "${CI_DIR}/yamllint.yaml" -f "${YAMLLINT_FORMAT:-standard}" "${files[@]}"
-ok "yaml/lint passed (${#files[@]} files)"
+# yamllint is single-threaded, so the file list is split into LINT_JOBS CONTIGUOUS slices of the
+# sorted list, one yamllint per slice, run concurrently. Each slice's output goes to its own file
+# and the files are printed in slice order afterwards: the log is the same text, in the same
+# file order, as one serial `yamllint "${files[@]}"`, and a finding is never interleaved with
+# another process's output (yamllint's standard format puts the FILE NAME on its own line above
+# that file's `line:col` findings, so interleaving would misattribute them). The stage's exit
+# code is the highest any slice returned (yamllint: 1 = error, 2 = warning under strict), so a
+# single failing slice fails the stage exactly as the serial call did. LINT_JOBS=1 is serial.
+lint_jobs="${LINT_JOBS:-$(n="$(nproc 2>/dev/null || echo 1)"; [ "$n" -gt 8 ] && n=8; echo "$n")}"
+case "$lint_jobs" in ''|*[!0-9]*|0) err "LINT_JOBS must be a positive integer, got '${lint_jobs}'"; exit 2 ;; esac
+[ "$lint_jobs" -gt "${#files[@]}" ] && lint_jobs="${#files[@]}"
+lint_tmp="$(mktemp -d)"
+trap 'rm -rf "$lint_tmp"' EXIT
+per=$(( (${#files[@]} + lint_jobs - 1) / lint_jobs ))
+pids=()
+for (( i = 0; i < lint_jobs; i++ )); do
+  slice=("${files[@]:$(( i * per )):${per}}")
+  [ "${#slice[@]}" -gt 0 ] || break
+  ( rc=0
+    yamllint -c "${CI_DIR}/yamllint.yaml" -f "${YAMLLINT_FORMAT:-standard}" "${slice[@]}" \
+      >"${lint_tmp}/${i}.out" 2>&1 || rc=$?
+    echo "$rc" >"${lint_tmp}/${i}.rc" ) &
+  pids+=("$!")
+done
+wait "${pids[@]}" || true
+lint_rc=0
+for (( i = 0; i < ${#pids[@]}; i++ )); do
+  cat "${lint_tmp}/${i}.out"
+  # A slice that left no rc file died before yamllint returned: that is a failure, not a pass.
+  slice_rc="$(cat "${lint_tmp}/${i}.rc" 2>/dev/null || true)"
+  case "$slice_rc" in ''|*[!0-9]*) slice_rc=1 ;; esac
+  [ "$slice_rc" -gt "$lint_rc" ] && lint_rc="$slice_rc"
+done
+if [ "$lint_rc" -ne 0 ]; then
+  err "yaml/lint failed (yamllint exit ${lint_rc}; ${#files[@]} files in ${#pids[@]} parallel slice(s))"
+  exit "$lint_rc"
+fi
+ok "yaml/lint passed (${#files[@]} files, ${#pids[@]} parallel slice(s))"
 
 # A workflow's `configurationFile:` must name a file that exists. Renovate's did not after #79
 # renamed renovate.json -> renovate.json5, and every scheduled run failed with ENOENT for 11 days

conformance/lib/run_l1_offline.py

@@ -43,13 +43,35 @@ Exit codes: 0 all selected checks passed / 1 at least one FAILED / 3 nothing fai
 scope-declared check was UNWIRED. conformance/run.sh propagates 3 verbatim; ci/l0.sh turns it
 into the gate's own rc 3. This is the ONE place the "unwired, not green" rule lives now,
 instead of once per hand-written stage.
+
+PARALLELISM (--jobs, env CONF_OFFLINE_JOBS)
+-------------------------------------------
+The checks are independent processes, so they run in a bounded thread pool (one `bash` child
+per worker). Default: min(8, usable CPUs); `--jobs 1` is the old serial runner exactly. Each
+check's stdout+stderr is CAPTURED and only reported after every check has finished, in the
+ORDER THE DIRS WERE GIVEN, so the log, the JUnit and the step summary are byte-for-byte
+independent of which check finished first (only the trailing "[N.Ns]" wall-time on each
+status line varies). The 0/1/3 aggregation runs over the same ordered case list as before.
+
+Parallel-safety is a property of the CHECKS, not of this runner: every check must keep its
+scratch state in a private mktemp/mkdtemp dir and must not write into the tree or a fixed
+path. The audit that established that for the current set is recorded in
+decisions/2026-11-08-parallel-l0-checks.md; a new check that needs shared state must say so.
+
+Nested pools: the keycloak reconciler checks run their own mutation suites in a pool
+(mutations.py, env MUTATION_JOBS). When this runner is itself parallel it exports a SMALLER
+MUTATION_JOBS (unless the caller already set one), so the two suites share the CPUs the
+other jobs-2 checks are not using instead of each taking min(8, CPUs). See inner_jobs().
 """
 from __future__ import annotations
 
 import argparse
+import os
 import re
 import subprocess
 import sys
+import time
+from concurrent.futures import ThreadPoolExecutor
 import xml.etree.ElementTree as ET
 from pathlib import Path
 from xml.dom import minidom
@@ -97,6 +119,46 @@ def collect_meta(test_yaml: Path, prefix: str):
     return name, reqs
 
 
+def usable_cpus() -> int:
+    """CPUs this process may run on (the cgroup/affinity view, not the host's count)."""
+    try:
+        return max(1, len(os.sched_getaffinity(0)))
+    except (AttributeError, OSError):
+        return max(1, os.cpu_count() or 1)
+
+
+def positive_int(text: str) -> int:
+    n = int(text)
+    if n < 1:
+        raise argparse.ArgumentTypeError(f"must be >= 1, got {n}")
+    return n
+
+
+def inner_jobs(cpus: int, outer: int) -> int:
+    """MUTATION_JOBS to hand the checks when this runner runs `outer` of them at once.
+
+    Two checks (the keycloak client-secret and passkey reconcilers) fan out internally; the
+    other outer-2 workers are single processes. Give the two suites an equal share of the CPUs
+    those others leave free, never fewer than 2 each (a 1-wide pool would make the 40-mutation
+    passkey suite the critical path of the whole stage), never more than 8. Peak process count
+    is then about (outer-2) + 2*inner, i.e. <= cpus on any machine with cpus >= outer + 2, and
+    at worst ~1.25-1.5x cpus on a small runner — CPU-bound oversubscription of that size costs
+    latency per process, not throughput.
+    """
+    if outer <= 1:
+        return min(8, cpus)
+    return max(2, min(8, (cpus - outer + 2) // 2))
+
+
+def run_check(check: Path, env: dict) -> tuple[int, str, float]:
+    t0 = time.monotonic()
+    proc = subprocess.run(
+        ["bash", str(check)], capture_output=True, text=True,
+        stdin=subprocess.DEVNULL, env=env,
+    )
+    return proc.returncode, (proc.stdout + proc.stderr).strip(), time.monotonic() - t0
+
+
 def main() -> int:
     ap = argparse.ArgumentParser(description=__doc__)
     ap.add_argument("--junit", type=Path, required=True)
@@ -108,26 +170,54 @@ def main() -> int:
                          "(makes this runner exit 3); repeatable")
     ap.add_argument("--summary-out", type=Path, default=None,
                     help="write one Markdown line per case here, for the CI step summary")
+    ap.add_argument("--jobs", type=positive_int, default=None,
+                    help="checks to run at once (default: env CONF_OFFLINE_JOBS, else "
+                         "min(8, usable CPUs)); 1 = serial")
     ap.add_argument("dirs", nargs="+", type=Path)
     args = ap.parse_args()
 
     strict_scopes = set(args.unwired_scope)
 
-    cases = []  # (name, reqs, dir, rc, output, scope)
+    cpus = usable_cpus()
+    if args.jobs is not None:
+        jobs = args.jobs
+    elif os.environ.get("CONF_OFFLINE_JOBS", "").strip():
+        try:
+            jobs = positive_int(os.environ["CONF_OFFLINE_JOBS"].strip())
+        except (ValueError, argparse.ArgumentTypeError) as e:
+            ap.error(f"CONF_OFFLINE_JOBS={os.environ['CONF_OFFLINE_JOBS']!r}: {e}")
+    else:
+        jobs = min(8, cpus)
+    child_env = dict(os.environ)
+    if not child_env.get("MUTATION_JOBS", "").strip():
+        child_env["MUTATION_JOBS"] = str(inner_jobs(cpus, jobs))
+    print(f"[offline] running {len(args.dirs)} check(s), jobs={jobs} "
+          f"(cpus={cpus}, MUTATION_JOBS={child_env['MUTATION_JOBS']})", file=sys.stderr)
+
+    # Metadata first, in the given order; then the checks themselves through the pool. The
+    # results are consumed in SUBMISSION order, so `cases` is identical to the serial runner's.
+    planned = []  # (name, reqs, dir, scope, check-or-None)
     for d in args.dirs:
         check = next((d / s for s in CHECK_SCRIPTS if (d / s).exists()), None)
         name, reqs = collect_meta(d / "chainsaw-test.yaml", args.label_prefix)
         scope = scope_of(d / "chainsaw-test.yaml", args.scope_annotation)
-        if check is None:
+        planned.append((name, reqs, d, scope, check))
+
+    cases = []  # (name, reqs, dir, rc, output, scope)
+    elapsed = []  # wall seconds per case, same order (None: nothing ran)
+    with ThreadPoolExecutor(max_workers=jobs) as pool:
+        futures = [pool.submit(run_check, check, child_env) if check is not None else None
+                   for (_n, _r, _d, _s, check) in planned]
+        for (name, reqs, d, scope, check), fut in zip(planned, futures):
+            if fut is None:
                 cases.append((name, reqs, d, 3,
                               f"no offline check script ({' or '.join(CHECK_SCRIPTS)}) in {d}",
                               scope))
+                elapsed.append(None)
                 continue
-        proc = subprocess.run(
-            ["bash", str(check)], capture_output=True, text=True
-        )
-        out = (proc.stdout + proc.stderr).strip()
-        cases.append((name, reqs, d, proc.returncode, out, scope))
+            rc, out, secs = fut.result()
+            cases.append((name, reqs, d, rc, out, scope))
+            elapsed.append(secs)
 
     failures = sum(1 for c in cases if c[3] not in (0, 3))
     skipped = sum(1 for c in cases if c[3] == 3)
@@ -145,7 +235,7 @@ def main() -> int:
         },
     )
     summary_lines = []
-    for name, reqs, d, rc, out, scope in cases:
+    for (name, reqs, d, rc, out, scope), secs in zip(cases, elapsed):
         tc = ET.SubElement(
             suite,
             "testcase",
@@ -164,8 +254,9 @@ def main() -> int:
             fail.text = out
         status = "PASS" if rc == 0 else ("SKIP" if rc == 3 else "FAIL")
         mark = f" scope={scope}" if scope else ""
+        took = f" [{secs:.1f}s]" if secs is not None else ""
         print(f"[offline] {layer_of(d):3s} {status:4s} {name} "
-              f"({','.join(reqs) or 'untagged'}){mark}",
+              f"({','.join(reqs) or 'untagged'}){mark}{took}",
               file=sys.stderr)
         if rc not in (0, 3):
             print(out, file=sys.stderr)

conformance/tests/l0-static/keycloak-client-secret-reconciler/mutations.py

@@ -25,15 +25,19 @@
 #
 #   python3 mutations.py <path-to-extracted-reconciler.py>
 #
-# Runtime is ~20s (one subprocess per mutation, each running the full selftest
-# against an in-process stub Keycloak; no network, no cluster).
+# Runtime is ~11s serial (one subprocess per mutation, each running the full selftest
+# against an in-process stub Keycloak; no network, no cluster). The selftests run in a
+# bounded pool (env MUTATION_JOBS, default min(8, CPUs); 1 = serial); output order and
+# verdicts are independent of the pool size.
 # =============================================================================
 from __future__ import annotations
 
+import os
 import re
 import subprocess
 import sys
 import tempfile
+from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
 
 
@@ -136,32 +140,84 @@ MUTATIONS: list[tuple[str, str, str, str]] = [
 ]
 
 
+def mutation_jobs() -> int:
+    """Selftests to run at once: env MUTATION_JOBS, else min(8, usable CPUs).
+
+    conformance/lib/run_l1_offline.py exports a smaller MUTATION_JOBS when it is itself
+    running several checks in parallel, so the nested pools do not oversubscribe the host.
+    MUTATION_JOBS=1 reproduces the old serial loop exactly.
+    """
+    raw = os.environ.get("MUTATION_JOBS", "").strip()
+    if raw:
+        try:
+            n = int(raw)
+        except ValueError:
+            n = 0
+        if n < 1:
+            raise ValueError("MUTATION_JOBS must be a positive integer, got %r" % raw)
+        return n
+    try:
+        cpus = len(os.sched_getaffinity(0))
+    except (AttributeError, OSError):
+        cpus = os.cpu_count() or 1
+    return max(1, min(8, cpus))
+
+
+def run_selftest(mutated_src: str) -> subprocess.CompletedProcess:
+    with tempfile.NamedTemporaryFile("w", suffix=".py", delete=False) as fh:
+        fh.write(mutated_src)
+        path = fh.name
+    try:
+        return subprocess.run([sys.executable, path, "--selftest"],
+                              capture_output=True, text=True, timeout=240,
+                              stdin=subprocess.DEVNULL)
+    finally:
+        Path(path).unlink(missing_ok=True)
+
+
 def main() -> int:
     if len(sys.argv) != 2:
         print("usage: mutations.py <extracted-reconciler.py>", file=sys.stderr)
         return 2
     src = Path(sys.argv[1]).read_text()
 
-    problems: list[str] = []
-    killed = 0
+    try:
+        jobs = mutation_jobs()
+    except ValueError as e:
+        print("mutations.py: %s" % e, file=sys.stderr)
+        return 2
+
+    # Pass 1, serial and cheap: every anchor is checked BEFORE anything runs, exactly as the
+    # serial loop did, and each mutation's source is built here. None = anchor failure.
+    planned: list[tuple[str, str, str | None, str | None]] = []
     for name, why, old, new in MUTATIONS:
         old, new = dedent4(old), dedent4(new)
         # An anchor that no longer matches is a FAILURE, not a skip: it means the code
         # moved and this mutation silently stopped proving anything.
         n = src.count(old)
         if n != 1:
-            problems.append("%s: anchor matches %d times (expected exactly 1) -- the "
+            planned.append((name, why,
+                            "%s: anchor matches %d times (expected exactly 1) -- the "
                             "code moved and this mutation no longer proves anything. "
-                            "Re-anchor it against the current source." % (name, n))
+                            "Re-anchor it against the current source." % (name, n), None))
             continue
-        with tempfile.NamedTemporaryFile("w", suffix=".py", delete=False) as fh:
-            fh.write(src.replace(old, new, 1))
-            path = fh.name
+        planned.append((name, why, None, src.replace(old, new, 1)))
+
+    # Pass 2: the selftests, `jobs` at a time. Each child gets its own temp file, its own
+    # mkdtemp dirs and its own 127.0.0.1:0 stub port, so they share nothing. Results are
+    # consumed in MUTATIONS order, so every line below is printed in the same order, with the
+    # same verdict, as the serial loop produced -- whichever child finishes first.
+    problems: list[str] = []
+    killed = 0
+    pool = ThreadPoolExecutor(max_workers=jobs)
     try:
-            proc = subprocess.run([sys.executable, path, "--selftest"],
-                                  capture_output=True, text=True, timeout=240)
-        finally:
-            Path(path).unlink(missing_ok=True)
+        futures = [pool.submit(run_selftest, mutated) if mutated is not None else None
+                   for (_n, _w, _a, mutated) in planned]
+        for (name, why, anchor_problem, _m), fut in zip(planned, futures):
+            if anchor_problem is not None:
+                problems.append(anchor_problem)
+                continue
+            proc = fut.result()
             if proc.returncode == 0:
                 problems.append("%s SURVIVED -- --selftest still passed with this regression "
                                 "applied, so nothing in the suite actually checks it. The "
@@ -173,6 +229,10 @@ def main() -> int:
             how = ("%s assertion(s) caught it" % m.group(1) if m
                    else "the selftest aborted (caught, less cleanly)")
             print("  killed  %-42s %s" % (name, how))
+    finally:
+        # A timeout (or any other exception) still propagates and fails the check, as it did
+        # serially; queued mutations are cancelled rather than run for nothing.
+        pool.shutdown(wait=True, cancel_futures=True)
 
     print()
     if problems:

conformance/tests/l0-static/keycloak-passkey-reconciler/mutations.py

@@ -40,15 +40,19 @@
 #
 #   python3 mutations.py <path-to-extracted-reconciler.py>
 #
-# Runtime is ~60s (one subprocess per mutation, each running the full selftest
-# against an in-process stub Keycloak; no network, no cluster).
+# Runtime is ~85s serial (one subprocess per mutation, each running the full selftest
+# against an in-process stub Keycloak; no network, no cluster). The selftests run in a
+# bounded pool (env MUTATION_JOBS, default min(8, CPUs); 1 = serial); output order and
+# verdicts are independent of the pool size.
 # =============================================================================
 from __future__ import annotations
 
+import os
 import re
 import subprocess
 import sys
 import tempfile
+from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
 
 # (name, the mistake being modelled, anchor, replacement)
@@ -403,31 +407,83 @@ MUTATIONS: list[tuple[str, str, str, str]] = [
 ]
 
 
+def mutation_jobs() -> int:
+    """Selftests to run at once: env MUTATION_JOBS, else min(8, usable CPUs).
+
+    conformance/lib/run_l1_offline.py exports a smaller MUTATION_JOBS when it is itself
+    running several checks in parallel, so the nested pools do not oversubscribe the host.
+    MUTATION_JOBS=1 reproduces the old serial loop exactly.
+    """
+    raw = os.environ.get("MUTATION_JOBS", "").strip()
+    if raw:
+        try:
+            n = int(raw)
+        except ValueError:
+            n = 0
+        if n < 1:
+            raise ValueError("MUTATION_JOBS must be a positive integer, got %r" % raw)
+        return n
+    try:
+        cpus = len(os.sched_getaffinity(0))
+    except (AttributeError, OSError):
+        cpus = os.cpu_count() or 1
+    return max(1, min(8, cpus))
+
+
+def run_selftest(mutated_src: str) -> subprocess.CompletedProcess:
+    with tempfile.NamedTemporaryFile("w", suffix=".py", delete=False) as fh:
+        fh.write(mutated_src)
+        path = fh.name
+    try:
+        return subprocess.run([sys.executable, path, "--selftest"],
+                              capture_output=True, text=True, timeout=300,
+                              stdin=subprocess.DEVNULL)
+    finally:
+        Path(path).unlink(missing_ok=True)
+
+
 def main() -> int:
     if len(sys.argv) != 2:
         print("usage: mutations.py <extracted-reconciler.py>", file=sys.stderr)
         return 2
     src = Path(sys.argv[1]).read_text()
 
-    problems: list[str] = []
-    killed = 0
+    try:
+        jobs = mutation_jobs()
+    except ValueError as e:
+        print("mutations.py: %s" % e, file=sys.stderr)
+        return 2
+
+    # Pass 1, serial and cheap: every anchor is checked BEFORE anything runs, exactly as the
+    # serial loop did, and each mutation's source is built here. None = anchor failure.
+    planned: list[tuple[str, str, str | None, str | None]] = []
     for name, why, old, new in MUTATIONS:
         # An anchor that no longer matches is a FAILURE, not a skip: it means the code
         # moved and this mutation silently stopped proving anything.
         n = src.count(old)
         if n != 1:
-            problems.append("%s: anchor matches %d times (expected exactly 1) -- the "
+            planned.append((name, why,
+                            "%s: anchor matches %d times (expected exactly 1) -- the "
                             "code moved and this mutation no longer proves anything. "
-                            "Re-anchor it against the current source." % (name, n))
+                            "Re-anchor it against the current source." % (name, n), None))
             continue
-        with tempfile.NamedTemporaryFile("w", suffix=".py", delete=False) as fh:
-            fh.write(src.replace(old, new, 1))
-            path = fh.name
+        planned.append((name, why, None, src.replace(old, new, 1)))
+
+    # Pass 2: the selftests, `jobs` at a time. Each child gets its own temp file, its own
+    # mkdtemp dirs and its own 127.0.0.1:0 stub port, so they share nothing. Results are
+    # consumed in MUTATIONS order, so every line below is printed in the same order, with the
+    # same verdict, as the serial loop produced -- whichever child finishes first.
+    problems: list[str] = []
+    killed = 0
+    pool = ThreadPoolExecutor(max_workers=jobs)
     try:
-            proc = subprocess.run([sys.executable, path, "--selftest"],
-                                  capture_output=True, text=True, timeout=300)
-        finally:
-            Path(path).unlink(missing_ok=True)
+        futures = [pool.submit(run_selftest, mutated) if mutated is not None else None
+                   for (_n, _w, _a, mutated) in planned]
+        for (name, why, anchor_problem, _m), fut in zip(planned, futures):
+            if anchor_problem is not None:
+                problems.append(anchor_problem)
+                continue
+            proc = fut.result()
             if proc.returncode == 0:
                 problems.append("%s SURVIVED -- --selftest still passed with this regression "
                                 "applied, so nothing in the suite actually checks it. The "
@@ -439,6 +495,10 @@ def main() -> int:
             how = ("%s assertion(s) caught it" % m.group(1) if m
                    else "the selftest aborted (caught, less cleanly)")
             print("  killed  %-54s %s" % (name, how))
+    finally:
+        # A timeout (or any other exception) still propagates and fails the check, as it did
+        # serially; queued mutations are cancelled rather than run for nothing.
+        pool.shutdown(wait=True, cancel_futures=True)
 
     print()
     if problems: