fix(ci): stream lane-shas progress + distinguish timeout from content drift

lane-shas.yml has intermittently failed under Act-runner contention with
truncated/ambiguous logs (see docs/plans/adr-0241-0242-session-completion-
summary-2026-07-15.md #38 postmortem: run #167 failed at ~22m39s, but local
verify_lane_shas.py showed 9/9 pins matching — the failure was never
conclusively attributed to a timeout vs an incomplete log, because the
script buffers all per-lane output until every lane finishes).

Two changes, no lane logic touched:

1. Stream progress per lane (`-> lane_id ... done (Ns)`, flush=True) instead
   of buffering until the end, so a hang shows exactly which lane and for
   how long even if the job is killed before the final summary prints.
2. Make the per-lane subprocess timeout configurable via
   CORE_LANE_VERIFY_TIMEOUT_S (default unchanged at 900s — same knob shape
   as CORE_SHOWCASE_HARD_BUDGET), and report a TimeoutExpired distinctly
   from a genuine SHA mismatch so the remediation text stops suggesting
   `--update` (re-pinning) for what is actually runner contention.

[Verification]: tests/test_lane_sha_verifier.py 6 passed; ruff clean;
scripts/verify_lane_shas.py run locally with the new streaming output
confirmed correct per-lane timing (e.g. demo_composition 154s) end to end.

scripts/verify_lane_shas.py

@@ -21,9 +21,11 @@ from __future__ import annotations
 import argparse
 import hashlib
 import json
+import os
 import subprocess
 import sys
 import tempfile
+import time
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any
@@ -31,6 +33,13 @@ from typing import Any
 
 REPO_ROOT = Path(__file__).resolve().parent.parent
 
+# Per-lane subprocess wall-clock budget. Overridable so CI can raise it under
+# known cold/contended-runner conditions without a code change — same knob
+# shape as CORE_SHOWCASE_HARD_BUDGET / CORE_SHOWCASE_SKIP_BUDGET elsewhere in
+# this repo. The job-level workflow timeout (lane-shas.yml) is a separate,
+# larger ceiling; this is what actually fires first when one lane hangs.
+LANE_TIMEOUT_S = int(os.environ.get("CORE_LANE_VERIFY_TIMEOUT_S", "900"))
+
 
 PINNED_SHAS: dict[str, str] = {
     "reviewer_registry": "681a2aab5aa4ffd58cd837ce5673c8b2a9545b570117aec3c02726a12f6876e6",
@@ -125,8 +134,6 @@ LANE_SPECS: tuple[LaneSpec, ...] = (
 
 
 def _invoke_runner(spec: LaneSpec, *, target_path: Path | None = None) -> Path:
-    import os
-
     env = {"PYTHONPATH": str(REPO_ROOT), **os.environ}
     # Force hermetic engine_state for every lane so local lived state and CI
     # workspaces cannot change demo / showcase report bytes.
@@ -147,7 +154,7 @@ def _invoke_runner(spec: LaneSpec, *, target_path: Path | None = None) -> Path:
         env=env,
         capture_output=True,
         text=True,
-        timeout=900,
+        timeout=LANE_TIMEOUT_S,
     )
     if result.returncode != 0:
         raise RuntimeError(
@@ -178,6 +185,7 @@ class LaneVerification:
     matched: bool
     report_path: str
     error: str | None = None
+    timed_out: bool = False
 
     def as_dict(self) -> dict[str, Any]:
         return {
@@ -187,13 +195,17 @@ class LaneVerification:
             "matched": self.matched,
             "report_path": self.report_path,
             "error": self.error,
+            "timed_out": self.timed_out,
         }
 
 
-def verify_all(*, ephemeral: bool = True) -> list[LaneVerification]:
+def verify_all(*, ephemeral: bool = True, stream: bool = False) -> list[LaneVerification]:
     results: list[LaneVerification] = []
     for spec in LANE_SPECS:
         pinned = PINNED_SHAS.get(spec.lane_id, "")
+        if stream:
+            print(f"  -> {spec.lane_id} ...", end="", flush=True)
+        started = time.monotonic()
         try:
             if ephemeral:
                 with tempfile.TemporaryDirectory(prefix=f"lane_{spec.lane_id}_") as d:
@@ -203,7 +215,30 @@ def verify_all(*, ephemeral: bool = True) -> list[LaneVerification]:
             else:
                 report_path = _invoke_runner(spec)
                 actual = _sha_of(report_path)
+        except subprocess.TimeoutExpired as exc:
+            elapsed = time.monotonic() - started
+            if stream:
+                print(f" TIMEOUT after {elapsed:.0f}s (budget {LANE_TIMEOUT_S}s)", flush=True)
+            results.append(
+                LaneVerification(
+                    lane_id=spec.lane_id,
+                    pinned_sha=pinned,
+                    actual_sha="",
+                    matched=False,
+                    report_path=str(spec.canonical_report),
+                    error=(
+                        f"TimeoutExpired: lane exceeded {LANE_TIMEOUT_S}s "
+                        f"(CORE_LANE_VERIFY_TIMEOUT_S) — likely runner "
+                        f"contention/cold-start, not a content change: {exc}"
+                    ),
+                    timed_out=True,
+                )
+            )
+            continue
         except Exception as exc:
+            elapsed = time.monotonic() - started
+            if stream:
+                print(f" ERROR after {elapsed:.0f}s", flush=True)
             results.append(
                 LaneVerification(
                     lane_id=spec.lane_id,
@@ -215,6 +250,9 @@ def verify_all(*, ephemeral: bool = True) -> list[LaneVerification]:
                 )
             )
             continue
+        if stream:
+            elapsed = time.monotonic() - started
+            print(f" done ({elapsed:.0f}s)", flush=True)
         results.append(
             LaneVerification(
                 lane_id=spec.lane_id,
@@ -251,7 +289,7 @@ def main(argv: list[str] | None = None) -> int:
     args = parser.parse_args(argv)
 
     if args.update:
-        results = verify_all(ephemeral=False)
+        results = verify_all(ephemeral=False, stream=not args.json)
         new_pins = {r.lane_id: r.actual_sha for r in results if not r.error}
         _rewrite_pins(new_pins)
         if args.json:
@@ -262,7 +300,7 @@ def main(argv: list[str] | None = None) -> int:
                 print(f"  {lane_id:>32}: {sha}")
         return 0
 
-    results = verify_all()
+    results = verify_all(stream=not args.json)
     if args.json:
         payload = {
             "total": len(results),
@@ -285,10 +323,21 @@ def main(argv: list[str] | None = None) -> int:
                 print()
         total = len(results)
         matched = sum(1 for r in results if r.matched)
+        timed_out = [r.lane_id for r in results if r.timed_out]
         print(f"\nlanes: {matched}/{total} match pinned SHAs")
-        if matched < total:
+        if timed_out:
             print(
-                "\nremediation:\n"
+                "\nremediation (timeout, not a content mismatch):\n"
+                f"  lane(s) {', '.join(timed_out)} exceeded the "
+                f"{LANE_TIMEOUT_S}s per-lane budget (CORE_LANE_VERIFY_TIMEOUT_S).\n"
+                "  Do NOT re-pin on a timeout alone — that only masks runner\n"
+                "  contention/cold-start under load. Re-run first; if it recurs, raise\n"
+                "  CORE_LANE_VERIFY_TIMEOUT_S for this job or investigate Act runner load\n"
+                "  (see docs/ci-optimization.md)."
+            )
+        if matched < total and len(timed_out) < (total - matched):
+            print(
+                "\nremediation (content drift):\n"
                 "  if the drift is intentional (e.g. you touched core/cognition/result.py,\n"
                 "  chat/runtime.py, generate/realizer.py, capability registries, or other\n"
                 "  lane-affecting code), re-pin with:\n"