fix(ci): bound the identity-ladder harness so the fast lane cannot time out

.github/workflows/ci.yml

@@ -566,8 +566,16 @@ jobs:
       # same browser/OCR corpus again on macOS exceeded the 900-second test
       # budget after all other macOS tests passed, without adding OS-specific
       # coverage. Keep every other macOS test and deselect only that exact node.
+      # OPENADAPT_IDENTITY_LADDER_EXHAUSTIVE opts this slow lane into the
+      # EXHAUSTIVE identity-ladder corpus (every collapse pair); the fast PR
+      # `test` job runs the bounded class-covering subset instead, because the
+      # full sweep's runtime scales with runner speed and intermittently blew
+      # its 900s budget on shared PR runners. Pinned by
+      # tests/test_ci_workflow_contract.py.
       - name: Test (full suite incl. e2e, canonical Ubuntu)
         if: runner.os == 'Linux'
+        env:
+          OPENADAPT_IDENTITY_LADDER_EXHAUSTIVE: "1"
         run: |
           mkdir -p runs
           pytest -q --basetemp=runs/ci

openadapt_flow/validation/identity_ladder.py

@@ -100,6 +100,27 @@
 # assigned round-robin from the pool below (so adding pairs never truncates).
 _COLLAPSE_PAIRS = list(COLLAPSE_PAIRS)
 
+
+def bounded_pairs() -> list:
+    """The first pair of every (glyph_class, flank) equivalence class.
+
+    The CI fast lane runs the harness on this bounded, class-covering subset
+    (every collapse mechanism -- O/0 and l/1, digit-/alpha-flanked and purely
+    numeric -- is still measured through the full production tier stack); the
+    nightly/dispatch full matrix runs the exhaustive corpus. Adding pairs to
+    COLLAPSE_PAIRS never grows this subset unless it introduces a NEW class,
+    which keeps the fast lane's runtime bounded by construction.
+    """
+    seen: set[tuple[str, str]] = set()
+    subset = []
+    for p in _COLLAPSE_PAIRS:
+        key = (p.glyph_class, p.flank)
+        if key not in seen:
+            seen.add(key)
+            subset.append(p)
+    return subset
+
+
 # A shared name+DOB pool (so the ONLY discriminator is the collapsible MRN --
 # the exact wrong-patient shape). Names are fake.
 _SHARED = [
@@ -270,18 +291,30 @@ def _make_backend(viewport, live_png, structured_live):
 
 
 def _anchor(
-    rec: Rendered, *, with_structured: bool, with_crop: bool, bundle_dir: Optional[Path]
+    rec: Rendered,
+    *,
+    with_structured: bool,
+    with_crop: bool,
+    bundle_dir: Optional[Path],
+    crop_name: str = "idcrop.png",
+    ocr_lines=None,
 ) -> Anchor:
     """Build the recorded anchor exactly as the compiler would for a click on
     the Open button: OCR context band (name+DOB+MRN), optional DOM structured
-    identity, optional identifier crop (the MRN cell)."""
+    identity, optional identifier crop (the MRN cell).
+
+    ``ocr_lines`` lets the caller supply the recorded frame's OCR result (the
+    full-frame OCR is the harness's dominant cost and depends only on the
+    frame); ``crop_name`` must be unique per recorded frame so anchors cached
+    across configs never read another pair's crop from the shared bundle dir.
+    """
     import openadapt_flow.vision as vision
     from openadapt_flow.runtime.identity import band_region, context_from_lines
 
     frame_bgr = cv2.imdecode(np.frombuffer(rec.png, np.uint8), cv2.IMREAD_COLOR)
     click = rec.open_point
     crop_region = _discriminative_crop_region(frame_bgr, click)
-    lines = vision.ocr(rec.png)
+    lines = vision.ocr(rec.png) if ocr_lines is None else ocr_lines
     from datetime import date
 
     context_text = context_from_lines(
@@ -300,8 +333,8 @@ def _anchor(
         crop = frame_bgr[y : y + h, x : x + w]
         ok, buf = cv2.imencode(".png", crop)
         assert ok
-        (bundle_dir / "idcrop.png").write_bytes(buf.tobytes())
-        identifier_crop = "idcrop.png"
+        (bundle_dir / crop_name).write_bytes(buf.tobytes())
+        identifier_crop = crop_name
         identifier_region = rec.mrn_region
 
     return Anchor(
@@ -316,20 +349,23 @@ def _anchor(
 
 
 def _verdict(
-    rec: Rendered,
     live: Rendered,
     *,
+    anchor: Anchor,
     with_structured: bool,
-    with_crop: bool,
     vlm,
     bundle_dir: Optional[Path],
 ) -> I.IdentityCheck:
-    """Drive the REAL Replayer._verify_identity for this substrate config."""
+    """Drive the REAL Replayer._verify_identity for this substrate config.
+
+    ``anchor`` is the recorded anchor for this (pair, substrate) -- built once
+    per combination by ``run`` (the anchor depends only on the RECORDED frame
+    and the substrate flags, so rebuilding it per verdict re-ran the recorded
+    frame's full-frame OCR ~22x per pair for the same result: the runtime
+    nondeterminism that intermittently blew the CI timeout).
+    """
     import openadapt_flow.vision as vision
 
-    anchor = _anchor(
-        rec, with_structured=with_structured, with_crop=with_crop, bundle_dir=bundle_dir
-    )
     step = Step(
         id="open",
         intent="open patient chart",
@@ -372,8 +408,17 @@ def _measure(name: str, cases: list[dict]) -> dict:
     }
 
 
-def run(out_dir: Path) -> dict:
-    pairs = [(p, _SHARED[i % len(_SHARED)]) for i, p in enumerate(_COLLAPSE_PAIRS)]
+def run(out_dir: Path, pair_subset=None) -> dict:
+    """Measure every config over ``pair_subset`` (default: the full corpus).
+
+    ``pair_subset`` bounds RUNTIME only -- which homonym pairs are measured --
+    never the tier stack, the configs, or the verdict logic: every pair still
+    runs through the real ``Replayer._verify_identity`` under all five
+    substrate configs. The fast CI lane passes :func:`bounded_pairs`; the
+    nightly full matrix and the CLI run the exhaustive corpus.
+    """
+    corpus = _COLLAPSE_PAIRS if pair_subset is None else list(pair_subset)
+    pairs = [(p, _SHARED[i % len(_SHARED)]) for i, p in enumerate(corpus)]
     # Pre-render every (pair, condition) frame once.
     rec: dict[str, Rendered] = {}
     stable_t: dict[str, Rendered] = {}
@@ -405,25 +450,52 @@ def run(out_dir: Path) -> dict:
     results: dict[str, dict] = {}
     tmp = Path(tempfile.mkdtemp(prefix="idladder_"))
 
+    # The recorded anchor depends only on the RECORDED frame and the substrate
+    # flags -- never on the live frame or the VLM -- so build each variant once
+    # and reuse it across every verdict/config. This removes the ~22x-per-pair
+    # redundant full-frame OCR of the same recorded frame (the dominant,
+    # runner-speed-scaled cost that intermittently blew the CI timeout). The
+    # recorded frame's OCR lines are likewise shared across the variants.
+    _rec_lines: dict[str, list] = {}
+    _anchors: dict[tuple[str, bool, bool], Anchor] = {}
+
+    def recorded_anchor(label: str, *, with_structured: bool, with_crop: bool):
+        key = (label, with_structured, with_crop)
+        if key not in _anchors:
+            if label not in _rec_lines:
+                import openadapt_flow.vision as vision
+
+                _rec_lines[label] = vision.ocr(rec[label].png)
+            _anchors[key] = _anchor(
+                rec[label],
+                with_structured=with_structured,
+                with_crop=with_crop,
+                bundle_dir=tmp if with_crop else None,
+                crop_name=f"idcrop_{label}.png",
+                ocr_lines=_rec_lines[label],
+            )
+        return _anchors[key]
+
     def config(name, *, cond_kind, with_structured, with_crop, vlm_on):
         cases = []
         for p, _ in pairs:
             bd = tmp if with_crop else None
+            anchor = recorded_anchor(
+                p.label, with_structured=with_structured, with_crop=with_crop
+            )
             if cond_kind == "stable":
                 live_c, live_w = stable_t[p.label], stable_s[p.label]
                 chk_c = _verdict(
-                    rec[p.label],
                     live_c,
+                    anchor=anchor,
                     with_structured=with_structured,
-                    with_crop=with_crop,
                     vlm=(ProbeFaithfulVLM(True) if vlm_on else None),
                     bundle_dir=bd,
                 )
                 chk_w = _verdict(
-                    rec[p.label],
                     live_w,
+                    anchor=anchor,
                     with_structured=with_structured,
-                    with_crop=with_crop,
                     vlm=(ProbeFaithfulVLM(False) if vlm_on else None),
                     bundle_dir=bd,
                 )
@@ -452,18 +524,16 @@ def config(name, *, cond_kind, with_structured, with_crop, vlm_on):
                     live_c = drift_t[(p.label, d.name)]
                     live_w = drift_s[(p.label, d.name)]
                     chk_c = _verdict(
-                        rec[p.label],
                         live_c,
+                        anchor=anchor,
                         with_structured=with_structured,
-                        with_crop=with_crop,
                         vlm=(ProbeFaithfulVLM(True) if vlm_on else None),
                         bundle_dir=bd,
                     )
                     chk_w = _verdict(
-                        rec[p.label],
                         live_w,
+                        anchor=anchor,
                         with_structured=with_structured,
-                        with_crop=with_crop,
                         vlm=(ProbeFaithfulVLM(False) if vlm_on else None),
                         bundle_dir=bd,
                     )

public-artifacts.json

@@ -89,7 +89,7 @@
     },
     {
       "path": ".github/workflows/ci.yml",
-      "sha256": "7752239ab2541c70b24e86e16706b50093f972e8c794d6e6fe385ea6db1f3739"
+      "sha256": "34f47a2752dda1e3502e90d32f443f07d27c5452af8e89e78931b507323e9bf4"
     },
     {
       "path": ".github/workflows/citrix-workspace-standin.yml",

tests/test_ci_workflow_contract.py

@@ -96,6 +96,34 @@ def test_macos_deselects_only_redundant_heavy_identity_harness() -> None:
     assert workflow.count(f"--deselect={node}") == 1
 
 
+def test_exhaustive_identity_ladder_corpus_runs_in_the_slow_lane_only() -> None:
+    """The exhaustive identity-ladder sweep stays OUT of the fast PR lane.
+
+    The fast `test` job runs the bounded class-covering harness; the
+    nightly/dispatch canonical-Ubuntu matrix leg opts into the exhaustive
+    corpus via OPENADAPT_IDENTITY_LADDER_EXHAUSTIVE. This pins both sides so
+    the exhaustive sweep can neither silently stop running anywhere nor creep
+    back into the fast lane whose 900s budget it intermittently exceeded.
+    """
+    workflow = CI.read_text(encoding="utf-8")
+    flag = "OPENADAPT_IDENTITY_LADDER_EXHAUSTIVE"
+
+    linux_start = workflow.index(
+        "- name: Test (full suite incl. e2e, canonical Ubuntu)"
+    )
+    macos_start = workflow.index(
+        "- name: Test (full suite incl. e2e, macOS platform coverage)"
+    )
+    linux_step = workflow[linux_start:macos_start]
+    assert f'{flag}: "1"' in linux_step
+
+    fast_start = workflow.index("- name: Test (fast unit suite)")
+    fast_end = workflow.index("- name: Coverage (whole-package visibility)")
+    assert flag not in workflow[fast_start:fast_end]
+    # exactly one opt-in: the canonical Ubuntu matrix leg
+    assert workflow.count(f'{flag}: "1"') == 1
+
+
 def test_clean_machine_lifecycle_declares_utf8_on_every_os() -> None:
     workflow = QUICKSTART.read_text(encoding="utf-8")
     lifecycle_start = workflow.index("  lifecycle:")

tests/test_identity_ladder.py

@@ -455,12 +455,7 @@ class _NoStruct(_Backend):
 # ---------------------------------------------------------------------------
 
 
-@pytest.mark.timeout(900)  # heavy browser+OCR integration; ~350s local, slower CI
-def test_harness_zero_false_accept_all_configs(tmp_path) -> None:
-    pytest.importorskip("playwright")
-    from openadapt_flow.validation import identity_ladder as H
-
-    summary = H.run(tmp_path)
+def _assert_zero_false_accept(summary: dict) -> None:
     # THE safety invariant, measured on the REAL Replayer._verify_identity
     # production tier stack (the OCR tier the replayer always appends is in the
     # stack for every config): 0 false-accept everywhere, incl. the homonym.
@@ -473,3 +468,48 @@ def test_harness_zero_false_accept_all_configs(tmp_path) -> None:
     # the OCR-only-confusable config the flawed harness never measured now
     # shows HIGH over-halt (OCR alone cannot verify a collapsible MRN).
     assert cfgs["ocr_only_confusable"]["over_halt_rate"] == 1.0
+
+
+@pytest.mark.timeout(900)  # bounded corpus: ~1/3 the exhaustive sweep's work
+def test_harness_zero_false_accept_bounded_configs(tmp_path) -> None:
+    """The fast-lane harness run: a BOUNDED, class-covering pair subset.
+
+    Runs on every PR. ``bounded_pairs`` keeps one homonym pair per
+    (glyph_class, flank) collapse class, so every collapse MECHANISM still
+    goes through the real production tier stack under all five substrate
+    configs while the runtime stays bounded by construction -- growing
+    COLLAPSE_PAIRS can never grow this test's work unless a genuinely new
+    collapse class appears. The exhaustive corpus runs nightly (see
+    ``test_harness_zero_false_accept_all_configs``)."""
+    pytest.importorskip("playwright")
+    from openadapt_flow.validation import identity_ladder as H
+
+    subset = H.bounded_pairs()
+    # every collapse class in the corpus is represented, none dropped
+    assert {(p.glyph_class, p.flank) for p in subset} == {
+        (p.glyph_class, p.flank) for p in H._COLLAPSE_PAIRS
+    }
+    _assert_zero_false_accept(H.run(tmp_path, pair_subset=subset))
+
+
+@pytest.mark.timeout(900)  # heavy browser+OCR integration, exhaustive corpus
+def test_harness_zero_false_accept_all_configs(tmp_path) -> None:
+    """The EXHAUSTIVE harness run: every collapse pair, every config.
+
+    The full 14-pair sweep's runtime scales with runner speed and blew the
+    900s budget intermittently on shared PR runners, so it runs where the
+    slow lane already runs -- the nightly/dispatch full matrix, which sets
+    OPENADAPT_IDENTITY_LADDER_EXHAUSTIVE=1 (pinned by
+    tests/test_ci_workflow_contract.py). The invariant itself is unchanged
+    and is also asserted on every PR by the bounded run above."""
+    import os
+
+    if not os.environ.get("OPENADAPT_IDENTITY_LADDER_EXHAUSTIVE"):
+        pytest.skip(
+            "exhaustive corpus runs in the nightly/dispatch full matrix "
+            "(set OPENADAPT_IDENTITY_LADDER_EXHAUSTIVE=1 to run it here)"
+        )
+    pytest.importorskip("playwright")
+    from openadapt_flow.validation import identity_ladder as H
+
+    _assert_zero_false_accept(H.run(tmp_path))