qa: walk gate tri-state verdicts (GREEN/RED/ERROR), provenance stamps, door-cross camera re-assert

qa/room_pipeline.py

@@ -87,6 +87,14 @@ def stage_walk(room: str, out: Path, engine: str, qa: str, stride: int) -> dict:
     except Exception as e:  # noqa: BLE001
         return {"status": "SKIP", "detail": f"walk_test could not run (player up on {qa}?): {e}"}
     (out / "walk_report.json").write_text(json.dumps(report, indent=2))
+    # ERROR verdict = a harness/infrastructure defect (player/engine/debug/shot unreachable), NOT a
+    # walkability verdict. Map it to SKIP (which already blocks shippable) so a broken harness never
+    # reads as a RED room defect AND never certifies the room — it must be retried/investigated.
+    if report["verdict"] == "ERROR":
+        errs = report.get("harness_errors", [])
+        first = "; ".join(errs[:3]) if errs else "(unspecified)"
+        return {"status": "SKIP",
+                "detail": f"harness error — retry/investigate (never a verdict): {first}"}
     cam = report["camera"]["ok"]
     counts = {k: report[k] for k in ("reachable", "impassable", "doors")}
     return {"status": report["verdict"], "detail": {"camera_ok": cam, **counts}}
@@ -160,6 +168,18 @@ def verify_certification(room: str) -> list:
     return fails
 
 
+def _is_shippable(results: dict, gate_stages: list) -> bool:
+    """Pure ship gate. A room ships ONLY when: no RED gate, no pending MANUAL stage, at least one
+    GREEN gate, AND — whenever a walk gate ran — its verdict is GREEN. A walk=SKIP (harness ERROR /
+    player down) or coherence-alone GREEN must NEVER certify: a coherent paint says nothing about
+    whether the room is walkable, which is the entire point of the walk gate."""
+    reds = [n for n in gate_stages if results[n]["status"] == "RED"]
+    manuals = [n for (n, r) in results.items() if r["status"] == "MANUAL"]
+    walk_ok = ("walk" not in gate_stages) or (results.get("walk", {}).get("status") == "GREEN")
+    return bool(not reds and not manuals and walk_ok
+                and any(results[n]["status"] == "GREEN" for n in gate_stages))
+
+
 def run(room: str, mode: str, out: Path, engine: str, qa: str, stride: int, resume: bool) -> dict:
     out.mkdir(parents=True, exist_ok=True)
     marker = out / "stages.json"
@@ -186,9 +206,12 @@ def run(room: str, mode: str, out: Path, engine: str, qa: str, stride: int, resu
 
     results = {}
     for name, fn in stages:
-        if resume and done.get(name, {}).get("status") in ("GREEN", "SKIP"):
+        # Only a cached GREEN is terminal on --resume. A cached SKIP must RE-RUN: an ERROR-mapped walk
+        # SKIP is a harness error to retry, and re-running a deliberate MANUAL/SKIP stage is cheap and
+        # safe. (Reusing SKIP as terminal could carry a harness outage forward as if it were settled.)
+        if resume and done.get(name, {}).get("status") == "GREEN":
             results[name] = done[name]
-            _log(f"{name}: (cached {done[name]['status']})")
+            _log(f"{name}: (cached GREEN)")
             continue
         _log(f"{name}: running…")
         res = fn()
@@ -199,7 +222,7 @@ def run(room: str, mode: str, out: Path, engine: str, qa: str, stride: int, resu
     gate_stages = [n for n in ("coherence", "walk") if n in results]
     reds = [n for n in gate_stages if results[n]["status"] == "RED"]
     manuals = [n for (n, r) in results.items() if r["status"] == "MANUAL"]
-    shippable = not reds and not manuals and any(results[n]["status"] == "GREEN" for n in gate_stages)
+    shippable = _is_shippable(results, gate_stages)
     report = {"room": room, "mode": mode, "stages": results,
               "gate_stages": gate_stages, "reds": reds, "pending_manual": manuals,
               "shippable": shippable, "ts": None}

qa/test_room_certification.py

@@ -38,3 +38,50 @@ def test_sha_drift_reads_stale(tmp_path, monkeypatch):
     monkeypatch.setattr(RP, "CERT_DIR", tmp_cert)
     fails = RP.verify_certification("shop")
     assert any("drifted" in f for f in fails)
+
+
+# --- Fix 1: a SKIP (harness/ERROR) walk gate must NEVER certify a room ------------------------------
+def test_is_shippable_walk_skip_never_ships():
+    """walk=SKIP + coherence=GREEN must NOT ship: a coherent paint says nothing about walkability."""
+    results = {"coherence": {"status": "GREEN", "detail": ""}, "walk": {"status": "SKIP", "detail": ""}}
+    assert RP._is_shippable(results, ["coherence", "walk"]) is False
+
+
+def test_is_shippable_requires_green_walk():
+    results = {"coherence": {"status": "GREEN"}, "walk": {"status": "GREEN"}}
+    assert RP._is_shippable(results, ["coherence", "walk"]) is True
+
+
+def test_is_shippable_coherence_only_still_ships():
+    """When no walk gate ran (walk not in gate_stages), a GREEN coherence still ships — the new gate
+    only closes the walk=SKIP hole, it does not change the walk-absent path."""
+    assert RP._is_shippable({"coherence": {"status": "GREEN"}}, ["coherence"]) is True
+
+
+def test_run_full_walk_skip_is_not_certified(tmp_path, monkeypatch):
+    """End-to-end: walk=SKIP + coherence=GREEN (MANUAL stages stubbed GREEN) → not shippable, and
+    write_certification is NEVER called."""
+    monkeypatch.setattr(RP, "stage_coherence", lambda room, out: {"status": "GREEN", "detail": "ok"})
+    monkeypatch.setattr(RP, "stage_walk",
+                        lambda room, out, engine, qa, stride: {"status": "SKIP", "detail": "harness"})
+    monkeypatch.setattr(RP, "stage_manual", lambda name, cmd, needs: {"status": "GREEN", "detail": "stub"})
+    cert_calls = []
+    monkeypatch.setattr(RP, "write_certification", lambda *a, **k: cert_calls.append(1))
+    rep = RP.run("crypt", "full", tmp_path, "e", "q", 3, resume=False)
+    assert rep["shippable"] is False
+    assert "certification" not in rep and cert_calls == []
+
+
+def test_resume_reexecutes_cached_skip(tmp_path, monkeypatch):
+    """A cached SKIP is NOT terminal on --resume — an ERROR-mapped harness SKIP must be RETRIED."""
+    calls = {"n": 0}
+
+    def _walk(room, out, engine, qa, stride):
+        calls["n"] += 1
+        return {"status": "SKIP", "detail": "harness"}
+
+    monkeypatch.setattr(RP, "stage_walk", _walk)
+    RP.run("crypt", "verify", tmp_path, "e", "q", 3, resume=False)
+    assert calls["n"] == 1
+    RP.run("crypt", "verify", tmp_path, "e", "q", 3, resume=True)   # cached SKIP must re-run
+    assert calls["n"] == 2

qa/test_walk_test.py

@@ -151,3 +151,268 @@ def test_path_cell_cr_normalizes_both_shapes():
     assert _path_cell_cr([3, 4]) == [3, 4]
     assert _path_cell_cr((3, 4)) == [3, 4]
     assert _path_cell_cr({"c": 3, "r": 4}) == [3, 4]
+
+
+# --- tri-state verdict classification (GREEN / RED / ERROR) -----------------------------------------
+def _base_report(**over):
+    """A clean report skeleton carrying only the fields classify_verdict reads."""
+    r = {"camera": {"pose_mismatch": False}, "reachable": {"fail": 0}, "impassable": {"fail": 0},
+         "doors": {"fail": 0}, "orphans": [], "path": {"fail": 0}, "visual": {"fail": 0},
+         "door_pose_fail": [], "harness_errors": []}
+    r.update(over)
+    return r
+
+
+def test_classify_verdict_clean_is_green():
+    assert W.classify_verdict(_base_report()) == ("GREEN", 0)
+
+
+def test_classify_verdict_harness_only_is_error():
+    """A harness/infra defect with NO walkability failure = ERROR/2 — never a room verdict."""
+    r = _base_report(harness_errors=["reachable (3,4): drive-error:conn refused"])
+    assert W.classify_verdict(r) == ("ERROR", 2)
+
+
+def test_classify_verdict_walkability_fail_is_red():
+    assert W.classify_verdict(_base_report(reachable={"fail": 1})) == ("RED", 1)
+
+
+def test_classify_verdict_real_fail_wins_over_harness():
+    """A genuine walkability failure present ALONGSIDE harness errors → RED/1 (real fail wins), so a
+    partial harness outage can never downgrade a proven-broken room to a mere ERROR."""
+    r = _base_report(impassable={"fail": 2}, harness_errors=["door [4,0]: click:timeout"])
+    assert W.classify_verdict(r) == ("RED", 1)
+
+
+def test_classify_verdict_camera_pose_mismatch_is_red():
+    assert W.classify_verdict(_base_report(camera={"pose_mismatch": True})) == ("RED", 1)
+
+
+def test_classify_verdict_door_pose_mismatch_is_red():
+    r = _base_report(door_pose_fail=[{"door": [1, 2], "leg": "dest", "fails": ["camOrtho ..."]}])
+    assert W.classify_verdict(r) == ("RED", 1)
+
+
+def test_classify_verdict_orphans_and_visual_are_walkability():
+    """The visual zero-measurable-case fail and orphan pockets are walkability RED (guard vacuous
+    greens), never reclassified as harness."""
+    assert W.classify_verdict(_base_report(orphans=[[7, 3]])) == ("RED", 1)
+    assert W.classify_verdict(_base_report(visual={"fail": 1})) == ("RED", 1)
+
+
+def test_is_drive_error_only_matches_the_sentinel():
+    """A drive-error string is harness; a settled cell list or a plain None (timeout) is not."""
+    assert W.is_drive_error("drive-error:HTTPError") is True
+    assert W.is_drive_error([4, 5]) is False
+    assert W.is_drive_error(None) is False
+    assert W.is_drive_error("cell(4,5)") is False
+
+
+# --- provenance stamps -----------------------------------------------------------------------------
+def test_init_report_carries_provenance_stamps():
+    """The report self-describes its provenance (closes the #1607 cert traceability loop)."""
+    r = W._init_report("crypt", CRYPT_ORTHO, "crypt_scene",
+                       {"cols": 5, "rows": 4}, "http://engine:8766", "http://qa:8971")
+    assert r["schema_version"] == 1
+    assert "T" in r["ts"]                       # ISO8601 UTC
+    assert r["engine_url"] == "http://engine:8766" and r["qa_url"] == "http://qa:8971"
+    assert "repo_sha" in r and "manifest_sha256" in r   # present (value may be None if unavailable)
+    assert r["manifest_sha256"] and len(r["manifest_sha256"]) == 64   # real sha of the on-disk manifest
+    assert r["harness_errors"] == [] and r["door_pose_fail"] == []
+    assert r["verdict"] == "PENDING" and r["ortho"] == CRYPT_ORTHO
+
+
+# --- camera-fail + door-cross pose classification (walkability RED vs harness) ----------------------
+def test_classify_camera_fails_missing_extension_is_harness():
+    """A player build with NO /debug camera fields → harness (never a walkability verdict)."""
+    dbg = {"ok": True, "enq": 3}                 # no camOrtho
+    fails = W.check_camera_pose(dbg, CRYPT_ORTHO)
+    walk, harness = W.classify_camera_fails(dbg, fails)
+    assert walk == [] and harness and "unavailable" in harness[0]
+
+
+def test_classify_camera_fails_pose_mismatch_is_walkability():
+    """Wrong ortho/rotation/aim (fields present) = the 2026-07-15 root-cause class = walkability RED."""
+    snap = _contract_snapshot()
+    snap["originVX"], snap["originVY"] = 0.72, 0.34
+    fails = W.check_camera_pose(snap, CRYPT_ORTHO)
+    walk, harness = W.classify_camera_fails(snap, fails)
+    assert walk and harness == []
+
+
+def test_classify_pose_observation_debug_unreachable_is_harness():
+    walk, harness = W.classify_pose_observation({"_error": "connection refused"}, CRYPT_ORTHO)
+    assert walk == [] and harness and "unreachable" in harness[0]
+
+
+def test_classify_pose_observation_mismatch_is_walkability_red():
+    snap = _contract_snapshot()
+    snap["camRy"] = 60.0                          # destination camera off the frozen dimetric contract
+    walk, harness = W.classify_pose_observation(snap, CRYPT_ORTHO)
+    assert walk and harness == []
+
+
+def test_classify_pose_observation_no_ortho_skips():
+    """A door target with no pinned ortho asserts nothing — not RED, not harness."""
+    assert W.classify_pose_observation(_contract_snapshot(), None) == ([], [])
+
+
+def test_classify_pose_observation_contract_pose_is_clean():
+    assert W.classify_pose_observation(_contract_snapshot(), CRYPT_ORTHO) == ([], [])
+
+
+# --- Fix 3: poll-time engine outage in _drive_and_check → harness, not a false verdict --------------
+def test_drive_and_check_total_poll_outage_is_harness(monkeypatch):
+    """Engine alive at click, then dies for EVERY poll → harness sentinel (an impassable check would
+    otherwise false-PASS on the stale `before` cell; a reachable check would false-RED)."""
+    calls = {"n": 0}
+
+    def _get_stub(url, timeout=5.0):
+        calls["n"] += 1
+        if calls["n"] == 1:
+            return {"tokens": [{"x": 0, "y": 0}]}          # the pre-click `before` read succeeds
+        raise ConnectionError("engine died mid-probe")       # every poll after the click fails
+
+    monkeypatch.setattr(W, "_get", _get_stub)
+    monkeypatch.setattr(W, "_post", lambda url, body, timeout=5.0: {"ok": True})
+    monkeypatch.setattr(W.time, "sleep", lambda s: None)
+    ok, landed, path = W._drive_and_check("q", "e", 3, 4, 0.001, 0.03, expect_move=False)
+    assert ok is False and W.is_drive_error(landed)          # NOT a false impassable PASS
+
+
+def test_drive_and_check_partial_outage_keeps_verdict(monkeypatch):
+    """A single failed poll among good ones keeps normal timeout semantics — only a TOTAL outage is
+    harness."""
+    seq = [{"tokens": [{"x": 0, "y": 0}]}, ConnectionError("blip"), {"tokens": [{"x": 3, "y": 4}]}]
+
+    def _get_stub(url, timeout=5.0):
+        v = seq.pop(0) if seq else {"tokens": [{"x": 3, "y": 4}]}
+        if isinstance(v, Exception):
+            raise v
+        return v
+
+    monkeypatch.setattr(W, "_get", _get_stub)
+    monkeypatch.setattr(W, "_post", lambda url, body, timeout=5.0: {"ok": True})
+    monkeypatch.setattr(W.time, "sleep", lambda s: None)
+    ok, landed, path = W._drive_and_check("q", "e", 3, 4, 0.001, 5.0, expect_move=True)
+    assert ok is True and landed == [3, 4]                   # a good poll saw arrival → real PASS
+
+
+# --- Fix 2 / 4b: door-cross pose is captured only on a CONFIRMED leg, and never on an unpinned ortho -
+class _FakeWorld:
+    """A minimal live-player+engine fake: click 1 crosses to `crossed_to`; click 2 returns home iff
+    `return_home`. /combat-surface reports the current location + a back-door to home; /debug returns
+    a contract snapshot."""
+    def __init__(self, home, target, crossed_to, return_home):
+        self.home, self.target, self.crossed_to, self.return_home = home, target, crossed_to, return_home
+        self.loc, self.clicks, self.debug_calls = home, 0, 0
+
+    def get(self, url, timeout=5.0):
+        return {"location": self.loc, "doors": [{"cell": [1, 1], "to": self.home}]}
+
+    def post(self, url, body=None, timeout=5.0):
+        if url.endswith("/click"):
+            self.clicks += 1
+            if self.clicks == 1:
+                self.loc = self.crossed_to
+            elif self.clicks == 2 and self.return_home:
+                self.loc = self.home
+            return {"ok": True}
+        if url.endswith("/debug"):
+            self.debug_calls += 1
+            return dict(_contract_snapshot())
+        return {}
+
+
+def _wire(monkeypatch, fw):
+    monkeypatch.setattr(W, "_get", fw.get)
+    monkeypatch.setattr(W, "_post", fw.post)
+    monkeypatch.setattr(W.time, "sleep", lambda s: None)
+
+
+def test_door_home_pose_skipped_on_timed_out_return(monkeypatch):
+    """The party crosses to target but the return leg TIMES OUT (still in target room). The home-leg
+    pose must NOT be recorded — asserting HOME's ortho against the target room = a false RED."""
+    fw = _FakeWorld("home_room", "target_room", crossed_to="target_room", return_home=False)
+    _wire(monkeypatch, fw)
+    ok, detail = W._check_door_cross("q", "e", (2, 0), "target_room", "home_room", 0.001, 0.03,
+                                     dest_ortho=CRYPT_ORTHO, home_ortho=CRYPT_ORTHO)
+    assert "dest" in detail["pose"]                # arrival at target confirmed → dest pose recorded
+    assert "home" not in detail.get("pose", {})    # unconfirmed return → NO home pose
+
+
+def test_door_home_pose_recorded_on_confirmed_return(monkeypatch):
+    fw = _FakeWorld("home_room", "target_room", crossed_to="target_room", return_home=True)
+    _wire(monkeypatch, fw)
+    ok, detail = W._check_door_cross("q", "e", (2, 0), "target_room", "home_room", 0.001, 0.5,
+                                     dest_ortho=CRYPT_ORTHO, home_ortho=CRYPT_ORTHO)
+    assert "dest" in detail["pose"] and "home" in detail["pose"]
+
+
+def test_door_dest_pose_skipped_when_not_arrived_at_target(monkeypatch):
+    """`crossed` != home only proves we LEFT; it does not prove we reached `target`. A cross to the
+    wrong room must NOT capture a dest pose (it would assert the wrong room's ortho)."""
+    fw = _FakeWorld("home_room", "roomB", crossed_to="roomX", return_home=False)
+    _wire(monkeypatch, fw)
+    ok, detail = W._check_door_cross("q", "e", (2, 0), "roomB", "home_room", 0.001, 0.03,
+                                     dest_ortho=CRYPT_ORTHO, home_ortho=CRYPT_ORTHO)
+    assert ok is False                             # crossed to the wrong room
+    assert "dest" not in detail.get("pose", {})
+
+
+def test_door_unpinned_ortho_skips_debug_fetch(monkeypatch):
+    """codex-P2: when a leg's room has no pinned ortho, skip the /debug fetch ENTIRELY — no pose work,
+    and no spurious harness error from a /debug that we would never assert against."""
+    fw = _FakeWorld("home_room", "target_room", crossed_to="target_room", return_home=True)
+    _wire(monkeypatch, fw)
+    ok, detail = W._check_door_cross("q", "e", (2, 0), "target_room", "home_room", 0.001, 0.5,
+                                     dest_ortho=None, home_ortho=None)
+    assert fw.debug_calls == 0                      # no /debug fetched for an unpinned leg
+    assert detail.get("pose", {}) == {}
+
+
+# --- Addendum: animated-fire-VFX masking (#1525) ---------------------------------------------------
+def test_fire_anchor_cells_from_geometry():
+    geo = {"props": [{"kind": "brazier", "cells": [[5, 1]]},
+                     {"kind": "wall_run", "cells": [[0, 0]]},
+                     {"kind": "hearth", "cells": [[10, 2], [10, 3]]}]}
+    assert W.fire_anchor_cells(geo) == {(5, 1), (10, 2), (10, 3)}
+    assert W.fire_anchor_cells({}) == set()
+
+
+def test_fire_mask_removes_flicker_blob_and_selects_actor():
+    """A brazier-flicker blob nearer to the actor's expected cell is masked, so the nearest-neighbour
+    selection resolves to the ACTOR blob instead of losing the race to the flame VFX."""
+    fire_blob = {"cx": 100.0, "cy": 100.0, "bottom": (100, 110), "area": 5000}
+    actor_blob = {"cx": 300.0, "cy": 300.0, "bottom": (300, 320), "area": 900}
+    kept = W.mask_fire_blobs([fire_blob, actor_blob], [(100.0, 100.0)], radius_px=30.0)
+    assert kept == [actor_blob]
+    assert W.nearest_blob_distance(kept, (300, 320)) < 25
+
+
+def test_fire_mask_all_excluded_is_empty_not_a_pass():
+    """If every blob is fire-masked the case has ZERO measurable blobs → the caller fails it loud
+    (nearest_blob_distance is inf). Masking never invents a pass."""
+    blobs = [{"cx": 100.0, "cy": 100.0, "bottom": (100, 110), "area": 5000}]
+    assert W.mask_fire_blobs(blobs, [(100.0, 100.0)], 30.0) == []
+    assert W.nearest_blob_distance([], (300, 320)) == float("inf")
+
+
+def test_fire_mask_noop_without_fire():
+    blobs = [{"cx": 1.0, "cy": 1.0, "bottom": (1, 1), "area": 1}]
+    assert W.mask_fire_blobs(blobs, [], 30.0) == blobs
+
+
+def test_select_visual_cells_deprioritizes_fire_but_fills_to_n():
+    fire = {(5, 5)}
+    pool = [(5, 5), (5, 6), (6, 5), (1, 1), (2, 2), (3, 3), (8, 8)]   # first three are fire-adjacent
+    picked = W.select_visual_cells(pool, 4, fire, min_cheby=2)
+    assert len(picked) == 4
+    assert set(picked) <= {(1, 1), (2, 2), (3, 3), (8, 8)}           # no fire-adjacent cell chosen
+
+
+def test_select_visual_cells_falls_back_to_fire_adjacent_when_needed():
+    fire = {(5, 5)}
+    pool = [(5, 5), (5, 6), (1, 1)]                                  # only ONE far cell; need 3
+    picked = W.select_visual_cells(pool, 3, fire, min_cheby=2)
+    assert len(picked) == 3 and (1, 1) in picked                     # fills to N incl. fire-adjacent