From aca6e162541af14600742055d95ffdb13c5a2d75 Mon Sep 17 00:00:00 2001 From: Eva Date: Thu, 16 Jul 2026 13:02:38 +0700 Subject: [PATCH 1/2] qa: walk gate tri-state verdicts (GREEN/RED/ERROR), provenance stamps, door-cross camera re-assert --- qa/room_pipeline.py | 8 ++ qa/test_walk_test.py | 108 +++++++++++++++++++++++ qa/walk_test.py | 205 ++++++++++++++++++++++++++++++++++++++----- 3 files changed, 301 insertions(+), 20 deletions(-) diff --git a/qa/room_pipeline.py b/qa/room_pipeline.py index 8d50be86..dbb3aa80 100644 --- a/qa/room_pipeline.py +++ b/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}} diff --git a/qa/test_walk_test.py b/qa/test_walk_test.py index 0780dd10..9510bfda 100644 --- a/qa/test_walk_test.py +++ b/qa/test_walk_test.py @@ -151,3 +151,111 @@ 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) == ([], []) diff --git a/qa/walk_test.py b/qa/walk_test.py index cdcf59c7..cd57eb98 100644 --- a/qa/walk_test.py +++ b/qa/walk_test.py @@ -242,6 +242,110 @@ def path_violations(path, mask: dict) -> list: return bad +# --- tri-state classification (PURE; unit-tested) -------------------------------------------------- +# A harness/infrastructure defect must NEVER read as a walkability verdict, in either direction. We +# split every failure into a WALKABILITY class (a real room defect the gate exists to catch) vs a +# HARNESS class (player/engine unreachable, /debug missing the camera fields, /shot capture failure, +# drive-error exceptions). Ambiguous → classify as the side that can never certify a broken room green. +def is_drive_error(landed) -> bool: + """A _drive_and_check result whose `landed` is a 'drive-error:' sentinel = a HARNESS error + (click/engine POST threw), NOT the room refusing/mis-resolving a move. Timeouts are NOT drive + errors — they stay walkability failures (they guard vacuous greens).""" + return isinstance(landed, str) and landed.startswith("drive-error:") + + +def classify_camera_fails(dbg: dict, cam_fails: list) -> tuple: + """Split check_camera_pose failures → (walkability_fails, harness_errors). A wholly-absent camera + extension (camOrtho None) or an unreachable /debug (dbg carries `_error`) is a HARNESS error; any + other camera failure is a real pose MISMATCH — wrong ortho/rotation/aim — which is the 2026-07-15 + root-cause class and a genuine walkability RED.""" + if not cam_fails: + return [], [] + if dbg.get("_error") is not None or dbg.get("camOrtho") is None: + return [], list(cam_fails) + return list(cam_fails), [] + + +def classify_pose_observation(dbg: dict, ortho) -> tuple: + """Classify a door-cross camera re-assert observation → (walkability_fails, harness_errors). + /debug unreachable = HARNESS; a real ortho/aim/rotation mismatch = walkability RED (the camera- + contract regression the gate exists to catch); no pinned ortho for the room = neither (skip).""" + if dbg.get("_error") is not None: + return [], [f"door-cross /debug unreachable: {dbg['_error']}"] + if ortho is None: + return [], [] + return classify_camera_fails(dbg, check_camera_pose(dbg, ortho)) + + +def classify_verdict(report: dict) -> tuple: + """Pure verdict/exit-code decision → (verdict, exit_code). A WALKABILITY failure (reachable/ + impassable/path/door/visual/orphan, OR a camera-pose mismatch, OR a door-cross pose mismatch) + → RED/1 — a real room fail WINS even when harness errors are also present. No walkability failure + but harness_errors non-empty → ERROR/2 (a harness/infra defect, never a room verdict). Clean → + GREEN/0. The impassable timeout-fail and visual zero-measurable-case fails are walkability fails + by construction (they live in the fail counters), so they correctly win over harness.""" + walk = bool( + report.get("camera", {}).get("pose_mismatch") + or report["reachable"]["fail"] + or report["impassable"]["fail"] + or report["doors"]["fail"] + or report.get("orphans") + or report["path"]["fail"] + or report.get("visual", {}).get("fail") + or report.get("door_pose_fail") + ) + if walk: + return "RED", 1 + if report.get("harness_errors"): + return "ERROR", 2 + return "GREEN", 0 + + +# --- provenance stamps (self-describing report; closes the #1607 cert traceability loop) ----------- +def _repo_sha(): + import subprocess # noqa: PLC0415 + try: + r = subprocess.run(["git", "-C", str(REPO), "rev-parse", "--short", "HEAD"], + capture_output=True, text=True, timeout=10) + return r.stdout.strip() or None + except Exception: # noqa: BLE001 + return None + + +def _manifest_sha256(): + import hashlib # noqa: PLC0415 + try: + return hashlib.sha256(MANIFEST.read_bytes()).hexdigest() + except Exception: # noqa: BLE001 + return None + + +def _utc_now_iso() -> str: + from datetime import datetime, timezone # noqa: PLC0415 + return datetime.now(timezone.utc).isoformat() + + +def _init_report(room: str, ortho: float, scene: str, mask: dict, engine: str, qa: str) -> dict: + """Build the base walk_report dict — provenance stamps + zeroed sub-structures. Factored out so + the provenance keys are unit-testable without a live player.""" + return { + "schema_version": 1, + "repo_sha": _repo_sha(), + "manifest_sha256": _manifest_sha256(), + "ts": _utc_now_iso(), + "engine_url": engine, "qa_url": qa, + "room": room, "ortho": ortho, "sceneId": scene, + "grid": {"cols": mask["cols"], "rows": mask["rows"]}, + "camera": {}, "orphans": [], "path": {"pass": 0, "fail": 0, "violations": []}, + "reachable": {"pass": 0, "fail": 0, "cases": []}, + "impassable": {"pass": 0, "fail": 0, "cases": []}, + "doors": {"pass": 0, "fail": 0, "cases": []}, + "visual": {"pass": 0, "fail": 0, "cases": []}, + "harness_errors": [], "door_pose_fail": [], + "shots": [], "verdict": "PENDING", + } + + # --- live I/O helpers ------------------------------------------------------------------------------ def _get(url: str, timeout: float = 5.0): with urllib.request.urlopen(url, timeout=timeout) as r: @@ -261,9 +365,23 @@ def _location(surf: dict): return loc.get("id") if isinstance(loc, dict) else loc -def _check_door_cross(qa: str, engine: str, cell, target, home, settle: float, timeout: float): +def _debug_or_error(qa: str) -> dict: + """POST /debug, returning the JSON or {'_error': str} — never raises (so a pose re-assert on an + unreachable player is classified as a HARNESS error, not a walkability RED).""" + try: + return _post(f"{qa}/debug", {}) + except Exception as e: # noqa: BLE001 + return {"_error": str(e)} + + +def _check_door_cross(qa: str, engine: str, cell, target, home, settle: float, timeout: float, + *, dest_ortho=None, home_ortho=None): """Click a door cell → assert the party CROSSES to `target` → return to `home` via the reciprocal - door. Verifies cross_door works both ways without stranding the party in the wrong room.""" + door. Verifies cross_door works both ways without stranding the party in the wrong room. After the + cross-arrival is confirmed, re-fetch /debug and record the camera pose against the DESTINATION + room's pinned ortho; after the return-home leg, record it against the HOME room's ortho — a pose + mismatch there is the door-cross camera-contract regression (a RED); /debug unreachable is harness. + The raw pose observations ride `detail['pose']`; run_gate classifies them (keeps this fn I/O-only).""" c, r = cell try: _post(f"{qa}/click", {"c": c, "r": r}) @@ -281,8 +399,11 @@ def _check_door_cross(qa: str, engine: str, cell, target, home, settle: float, t crossed = loc break ok = (crossed == target) if target else bool(crossed) - # return home via the target room's door back to `home` + pose: dict = {} + # re-assert the DESTINATION room's camera contract on arrival if crossed: + pose["dest"] = {"ortho": dest_ortho, "dbg": _debug_or_error(qa)} + # return home via the target room's door back to `home` try: back = next((d for d in _get(f"{engine}/combat-surface").get("doors", []) if d.get("to") == home), None) @@ -293,9 +414,14 @@ def _check_door_cross(qa: str, engine: str, cell, target, home, settle: float, t time.sleep(settle) if _location(_get(f"{engine}/combat-surface")) == home: break + # re-assert the HOME room's camera contract after the return leg + pose["home"] = {"ortho": home_ortho, "dbg": _debug_or_error(qa)} except Exception: # noqa: BLE001 pass - return ok, {"crossed_to": crossed, "target": target} + detail = {"crossed_to": crossed, "target": target} + if pose: + detail["pose"] = pose + return ok, detail def _token_cell(surf: dict): @@ -317,6 +443,17 @@ def _room_ortho(room: str) -> float: return float(entry["cameraPin"]["ortho"]) +def _room_ortho_opt(room): + """Pinned ortho for a room, or None if the room isn't in the manifest (door targets can point at + rooms with no pinned plate — a missing ortho means the door-cross pose check is skipped, not RED).""" + if not room: + return None + try: + return _room_ortho(room) + except SystemExit: + return None + + def _visual_registration(qa: str, engine: str, mask: dict, ortho: float, cells: list, out: Path, settle: float, move_timeout: float) -> dict: """MEASURE the actor's rendered screen position with NO client instrumentation (sidecar synthesis, @@ -390,20 +527,19 @@ def run_gate(room: str, engine: str, qa: str, *, stride: int, out: Path, print(f"[walk_test] WARNING: live sceneId '{scene}' does not name room '{room}' — the player " f"may be in a different room; cross a door first or pass the matching --room.") mask = walkmask_from_surface(surf) - report = {"room": room, "ortho": ortho, "sceneId": scene, - "grid": {"cols": mask["cols"], "rows": mask["rows"]}, - "camera": {}, "orphans": [], "path": {"pass": 0, "fail": 0, "violations": []}, - "reachable": {"pass": 0, "fail": 0, "cases": []}, - "impassable": {"pass": 0, "fail": 0, "cases": []}, - "doors": {"pass": 0, "fail": 0, "cases": []}, "shots": [], "verdict": "PENDING"} - - # 1) CAMERA POSE — the root-cause gate. + report = _init_report(room, ortho, scene, mask, engine, qa) + + # 1) CAMERA POSE — the root-cause gate. A pose MISMATCH (wrong ortho/rotation/aim) is a + # walkability RED; a wholly-absent camera extension or an unreachable /debug is a HARNESS error. try: dbg = _post(f"{qa}/debug", {}) except Exception as e: # noqa: BLE001 dbg = {"_error": str(e)} cam_fails = check_camera_pose(dbg, ortho) - report["camera"] = {"dbg": dbg, "fails": cam_fails, "ok": not cam_fails} + cam_walk, cam_harness = classify_camera_fails(dbg, cam_fails) + report["camera"] = {"dbg": dbg, "fails": cam_fails, "ok": not cam_fails, + "pose_mismatch": bool(cam_walk)} + report["harness_errors"].extend(cam_harness) # 1b) REACHABILITY / ORPHAN check (pure engine-truth, no driving): every walkable cell must be # BFS-reachable from the party's current cell. An orphan pocket = a seed defect or paint-invented @@ -425,6 +561,10 @@ def run_gate(room: str, engine: str, qa: str, *, stride: int, out: Path, for (c, r) in _sample(interior, stride): ok, landed, path = _drive_and_check(qa, engine, c, r, settle, move_timeout, expect_move=True) report["reachable"]["cases"].append({"cell": [c, r], "landed": landed, "ok": ok}) + # a drive-error exception is a HARNESS error, not the room mis-resolving a reachable move + if is_drive_error(landed): + report["harness_errors"].append(f"reachable ({c},{r}): {landed}") + continue report["reachable"]["pass" if ok else "fail"] += 1 # audit only THIS move's route: the path's endpoint must be the clicked cell (a stale # lastWalkPath from a previous move is skipped, never mis-attributed to this click) @@ -440,17 +580,39 @@ def run_gate(room: str, engine: str, qa: str, *, stride: int, out: Path, for (c, r) in _sample(blocked, max(1, stride)): ok, landed, _p = _drive_and_check(qa, engine, c, r, settle, move_timeout, expect_move=False) report["impassable"]["cases"].append({"cell": [c, r], "landed": landed, "ok": ok}) + # a drive-error exception is HARNESS; a timeout (token never became (c,r)) stays a walkability + # PASS and a token that DID move onto the blocked cell stays a walkability FAIL (unchanged). + if is_drive_error(landed): + report["harness_errors"].append(f"impassable ({c},{r}): {landed}") + continue report["impassable"]["pass" if ok else "fail"] += 1 # 4) DOORS — clicking a door CROSSES to the linked room (cross_door), it does NOT leave the token # on the door cell. Assert the cross lands in the door's `to` target, then return home. home = _location(surf) or room + home_ortho = _room_ortho_opt(home) for d in surf.get("doors", []): cell = tuple(d["cell"]) target = d.get("to") - ok, detail = _check_door_cross(qa, engine, cell, target, home, settle, move_timeout) + ok, detail = _check_door_cross(qa, engine, cell, target, home, settle, move_timeout, + dest_ortho=_room_ortho_opt(target), home_ortho=home_ortho) report["doors"]["cases"].append({"cell": list(cell), "to": target, "ok": ok, "detail": detail}) + # a door click that threw = HARNESS (player unreachable); a genuine cross pass/fail counts as + # before (existing counting preserved). + if isinstance(detail.get("error"), str) and detail["error"].startswith("click:"): + report["harness_errors"].append(f"door {list(cell)}: {detail['error']}") + continue report["doors"]["pass" if ok else "fail"] += 1 + # classify the door-cross camera re-assert: a pose mismatch = RED (walkability), /debug + # unreachable = harness. + for leg in ("dest", "home"): + obs = detail.get("pose", {}).get(leg) + if not obs: + continue + pose_walk, pose_harness = classify_pose_observation(obs["dbg"], obs["ortho"]) + if pose_walk: + report["door_pose_fail"].append({"door": list(cell), "leg": leg, "fails": pose_walk}) + report["harness_errors"].extend(f"door {list(cell)} {leg}: {m}" for m in pose_harness) # 5) VISUAL REGISTRATION (optional, --visual N): pixel-diff actor localization at the LIVE window # aspect — the client-instrumentation-free measurement of "does the actor RENDER where the plate @@ -485,10 +647,7 @@ def run_gate(room: str, engine: str, qa: str, *, stride: int, out: Path, if start and _location(_get(f"{engine}/combat-surface")) == (_location(surf) or room): _drive_and_check(qa, engine, start[0], start[1], settle, move_timeout, expect_move=True) - hard_fail = bool(cam_fails) or report["reachable"]["fail"] or report["impassable"]["fail"] \ - or report["doors"]["fail"] or report["orphans"] or report["path"]["fail"] \ - or report["visual"]["fail"] - report["verdict"] = "RED" if hard_fail else "GREEN" + report["verdict"], _ = classify_verdict(report) return report @@ -572,8 +731,9 @@ def main(argv=None) -> int: out.mkdir(parents=True, exist_ok=True) (out / "walk_report.json").write_text(json.dumps(report, indent=2) + "\n") + verdict, exit_code = classify_verdict(report) cam = report["camera"] - print(f"\n=== WALK_TEST {args.room} — {report['verdict']} ===") + print(f"\n=== WALK_TEST {args.room} — {verdict} ===") print(f"camera: {'OK' if cam['ok'] else 'FAIL'}" + ("" if cam["ok"] else "".join(f"\n - {m}" for m in cam["fails"]))) for k in ("reachable", "impassable", "doors", "path", "visual"): @@ -581,8 +741,13 @@ def main(argv=None) -> int: print(f"{k:11s}: {s['pass']} pass / {s['fail']} fail") print(f"orphans : {len(report['orphans'])}" + (f" — {report['orphans'][:6]}" if report["orphans"] else "")) + if report.get("door_pose_fail"): + print(f"door pose : {len(report['door_pose_fail'])} mismatch — {report['door_pose_fail'][:3]}") + if report.get("harness_errors"): + print(f"HARNESS ({len(report['harness_errors'])}) — NOT a walkability verdict:" + + "".join(f"\n - {m}" for m in report["harness_errors"][:8])) print(f"report: {out / 'walk_report.json'}") - return 0 if report["verdict"] == "GREEN" else 1 + return exit_code if __name__ == "__main__": From 92edf8d78cb2a1ebf5c63346b3290f19292f3350 Mon Sep 17 00:00:00 2001 From: Eva Date: Thu, 16 Jul 2026 13:24:30 +0700 Subject: [PATCH 2/2] =?UTF-8?q?qa:=20address=20review=20round=20=E2=80=94?= =?UTF-8?q?=20SKIP-can't-certify,=20confirmed-arrival=20door=20pose,=20pol?= =?UTF-8?q?l-outage=20harness,=20.get()-safe=20verdict,=20fire-VFX=20blob?= =?UTF-8?q?=20masking?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- qa/room_pipeline.py | 21 ++++- qa/test_room_certification.py | 47 ++++++++++ qa/test_walk_test.py | 157 ++++++++++++++++++++++++++++++++++ qa/walk_test.py | 141 ++++++++++++++++++++++++++---- 4 files changed, 345 insertions(+), 21 deletions(-) diff --git a/qa/room_pipeline.py b/qa/room_pipeline.py index dbb3aa80..551bd2bb 100644 --- a/qa/room_pipeline.py +++ b/qa/room_pipeline.py @@ -168,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" @@ -194,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() @@ -207,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} diff --git a/qa/test_room_certification.py b/qa/test_room_certification.py index ae67ac64..1ad48a95 100644 --- a/qa/test_room_certification.py +++ b/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 diff --git a/qa/test_walk_test.py b/qa/test_walk_test.py index 9510bfda..22024de6 100644 --- a/qa/test_walk_test.py +++ b/qa/test_walk_test.py @@ -259,3 +259,160 @@ def test_classify_pose_observation_no_ortho_skips(): 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 diff --git a/qa/walk_test.py b/qa/walk_test.py index cd57eb98..519717a3 100644 --- a/qa/walk_test.py +++ b/qa/walk_test.py @@ -49,6 +49,7 @@ HERE = Path(__file__).resolve().parent REPO = HERE.parent MANIFEST = REPO / "extensions" / "renderers" / "unity" / "plates_manifest.json" +GEO_DIR = HERE / "room_geometries" # --- the camera contract (MIRROR qa/greybox_render_headless — build_room_unified's rig) ------------ PITCH_DEG, YAW_DEG, PULLBACK = 30.0, 45.0, 80.0 @@ -221,6 +222,63 @@ def nearest_blob_distance(blobs: list, expected_px: tuple) -> float: return best +# --- animated-fire-VFX masking (#1525): braziers/hearths spawn flame VFX whose frame-to-frame flicker +# produces a large diff blob that can WIN the nearest-neighbour race against the actor sprite, reading +# a walkable cell as a false visual RED. These helpers only ever REMOVE candidate blobs / deprioritize +# fire-adjacent sample cells — they can never invent a pass (a case that masks to zero blobs stays a +# loud fail, same as a case with no measurable actor blob). +FIRE_KINDS = frozenset({"brazier", "campfire", "hearth"}) + + +def fire_anchor_cells(source: dict) -> set: + """Cells occupied by animated-fire props — read from a room-geometry (or surface) `props` list + whose entries carry {kind, cells:[[c,r],...]}. Empty set if the source has no such props.""" + out: set = set() + for p in (source or {}).get("props", []): + if p.get("kind") in FIRE_KINDS: + for cr in p.get("cells", []): + out.add((int(cr[0]), int(cr[1]))) + return out + + +def mask_fire_blobs(blobs: list, fire_px: list, radius_px: float) -> list: + """Drop any diff blob whose CENTROID lies within `radius_px` of a fire-anchor screen position. + Additive-only: returns a (possibly empty) subset of `blobs`, never a new blob.""" + if not fire_px: + return list(blobs) + kept = [] + for bl in blobs: + cx, cy = bl["cx"], bl["cy"] + if any(((cx - fx) ** 2 + (cy - fy) ** 2) ** 0.5 <= radius_px for fx, fy in fire_px): + continue + kept.append(bl) + return kept + + +def _cheby_far(cell, fire_cells, min_cheby: int) -> bool: + """True iff `cell` is ≥ min_cheby chebyshev-distance from EVERY fire-anchor cell (vacuously true + when there are no fire cells).""" + return all(max(abs(cell[0] - fc[0]), abs(cell[1] - fc[1])) >= min_cheby for fc in fire_cells) + + +def select_visual_cells(pool: list, n: int, fire_cells, *, min_cheby: int = 2) -> list: + """Pick up to `n` visual-registration cells, PREFERRING cells ≥ min_cheby from any fire anchor + (their pixel-diff is polluted by brazier/hearth flicker). Falls back to fire-adjacent cells only + to top the sample up to `n` when the far pool is too small. Deterministic (strided sample of the + far tier, then in-order fill). `pool` must already exclude the current/zero-hop cell.""" + if not pool or n <= 0: + return [] + far = [c for c in pool if _cheby_far(c, fire_cells, min_cheby)] + ordered = _sample(far, max(1, len(far) // max(1, n))) if far else [] + for src in (far, pool): # top up: remaining far first, then fire-adjacent as a last resort + for c in src: + if len(ordered) >= n: + break + if c not in ordered: + ordered.append(c) + return ordered[:n] + + def _path_cell_cr(cell) -> list: """Normalize a path cell ([c,r] list or {c,r} dict — the two shapes path_violations accepts).""" if isinstance(cell, dict): @@ -286,11 +344,11 @@ def classify_verdict(report: dict) -> tuple: by construction (they live in the fail counters), so they correctly win over harness.""" walk = bool( report.get("camera", {}).get("pose_mismatch") - or report["reachable"]["fail"] - or report["impassable"]["fail"] - or report["doors"]["fail"] + or report.get("reachable", {}).get("fail") + or report.get("impassable", {}).get("fail") + or report.get("doors", {}).get("fail") or report.get("orphans") - or report["path"]["fail"] + or report.get("path", {}).get("fail") or report.get("visual", {}).get("fail") or report.get("door_pose_fail") ) @@ -400,9 +458,13 @@ def _check_door_cross(qa: str, engine: str, cell, target, home, settle: float, t break ok = (crossed == target) if target else bool(crossed) pose: dict = {} - # re-assert the DESTINATION room's camera contract on arrival + # re-assert the DESTINATION room's camera contract — ONLY when arrival at `target` is CONFIRMED + # (crossed != home only proves we left, not that we reached `target`) and the target room has a + # pinned ortho. An unconfirmed cross is already counted by `ok`; do not double-report as a pose + # fail, and never /debug an unpinned-ortho leg (there is nothing to assert it against). if crossed: - pose["dest"] = {"ortho": dest_ortho, "dbg": _debug_or_error(qa)} + if target and crossed == target and dest_ortho is not None: + pose["dest"] = {"ortho": dest_ortho, "dbg": _debug_or_error(qa)} # return home via the target room's door back to `home` try: back = next((d for d in _get(f"{engine}/combat-surface").get("doors", []) @@ -410,12 +472,17 @@ def _check_door_cross(qa: str, engine: str, cell, target, home, settle: float, t if back: _post(f"{qa}/click", {"c": back["cell"][0], "r": back["cell"][1]}) bdl = time.time() + timeout + returned_home = False while time.time() < bdl: time.sleep(settle) if _location(_get(f"{engine}/combat-surface")) == home: + returned_home = True break - # re-assert the HOME room's camera contract after the return leg - pose["home"] = {"ortho": home_ortho, "dbg": _debug_or_error(qa)} + # re-assert the HOME room's camera contract — ONLY when the return CONFIRMED home (a + # timed-out return strands the party in the target room; asserting HOME's ortho there + # compares the wrong room = false RED) and home has a pinned ortho. + if returned_home and home_ortho is not None: + pose["home"] = {"ortho": home_ortho, "dbg": _debug_or_error(qa)} except Exception: # noqa: BLE001 pass detail = {"crossed_to": crossed, "target": target} @@ -454,8 +521,31 @@ def _room_ortho_opt(room): return None +def _load_room_geometry(room: str) -> dict: + """Best-effort load of a room's geometry JSON (for fire-anchor masking, #1525). Returns {} when no + geometry file is found — fire masking is additive, never fatal. Resolves via walk_static's + GEOMETRY_OF map, then a `{room}_geometry.json` fallback (generated rooms land there).""" + candidates = [] + try: + sys.path.insert(0, str(HERE)) + import walk_static as WS # noqa: PLC0415 + g = WS.GEOMETRY_OF.get(room) + if g: + candidates.append(GEO_DIR / g) + except Exception: # noqa: BLE001 + pass + candidates.append(GEO_DIR / f"{room}_geometry.json") + for c in candidates: + try: + if c.is_file(): + return json.loads(c.read_text()) + except Exception: # noqa: BLE001 + continue + return {} + + def _visual_registration(qa: str, engine: str, mask: dict, ortho: float, cells: list, - out: Path, settle: float, move_timeout: float) -> dict: + out: Path, settle: float, move_timeout: float, fire_cells=frozenset()) -> dict: """MEASURE the actor's rendered screen position with NO client instrumentation (sidecar synthesis, adopted): walk a chain of cells; pixel-diff consecutive /shots; the diff blobs are the departure + arrival actor sprites. Assert each expected cell projection (world_to_window_px at the LIVE window @@ -479,6 +569,12 @@ def _proj(cell): wz = ((rows - 1) / 2.0 - cell[1]) * 2.0 return world_to_window_px(wx, 0.6, wz, ortho, w, h) # 0.6 up = lower-torso/feet band + # brazier/hearth flame VFX flicker sits ON the fire cell — mask any diff blob within ~1.5 cells of + # a fire-anchor screen position so it cannot win the nearest-neighbour race against the actor. + fire_px = [_proj(fc) for fc in fire_cells] + fire_radius = 1.5 * cell_px(ortho, h) + res["fire_cells"] = [list(fc) for fc in sorted(fire_cells)] + prev = _token_cell(_get(f"{engine}/combat-surface")) # Chain the sample nearest-neighbour from the current cell: SHORT hops. The engine token flips # immediately but the CLIENT GLIDES the sprite over ~0.3-0.4s/cell — a /shot fired at @@ -502,12 +598,14 @@ def _proj(cell): shot_new = _capture_shot(qa, out, f"vis_{i}_c{cell[0]}r{cell[1]}") case = {"cell": list(cell), "moved": ok_move, "dist_px": None, "tol_px": round(tol, 1), "ok": False} if ok_move and shot_prev and shot_new: - blobs = diff_blobs(Image.open(shot_prev).convert("RGB"), Image.open(shot_new).convert("RGB")) - d_new = nearest_blob_distance(blobs, _proj(cell)) + raw = diff_blobs(Image.open(shot_prev).convert("RGB"), Image.open(shot_new).convert("RGB")) + blobs = mask_fire_blobs(raw, fire_px, fire_radius) # drop brazier-flicker blobs first + d_new = nearest_blob_distance(blobs, _proj(cell)) # inf when all blobs were fire-masked d_prev = nearest_blob_distance(blobs, _proj(prev)) if prev else 0.0 case["dist_px"] = round(d_new, 1) case["dist_prev_px"] = round(d_prev, 1) case["n_blobs"] = len(blobs) + case["n_blobs_masked"] = len(raw) - len(blobs) case["ok"] = d_new <= tol res["cases"].append(case) res["pass" if case["ok"] else "fail"] += 1 @@ -527,6 +625,9 @@ def run_gate(room: str, engine: str, qa: str, *, stride: int, out: Path, print(f"[walk_test] WARNING: live sceneId '{scene}' does not name room '{room}' — the player " f"may be in a different room; cross a door first or pass the matching --room.") mask = walkmask_from_surface(surf) + # fire-anchor cells (brazier/hearth VFX) from the room geometry and/or the surface props — used to + # mask animated-flame diff blobs in visual mode and to prefer fire-distant visual sample cells. + fire_cells = fire_anchor_cells(_load_room_geometry(room)) | fire_anchor_cells(surf) report = _init_report(room, ortho, scene, mask, engine, qa) # 1) CAMERA POSE — the root-cause gate. A pose MISMATCH (wrong ortho/rotation/aim) is a @@ -624,14 +725,11 @@ def run_gate(room: str, engine: str, qa: str, *, stride: int, out: Path, # the gate always measures the requested count when enough cells exist. cur = _token_cell(_get(f"{engine}/combat-surface")) pool = [c for c in interior if c != cur] - vis_cells = _sample(pool, max(1, len(pool) // max(1, visual)))[:visual] - for c in pool: - if len(vis_cells) >= visual: - break - if c not in vis_cells: - vis_cells.append(c) + # prefer cells ≥2 chebyshev from any brazier/hearth (their diff is polluted by flame VFX + # flicker); fall back to fire-adjacent cells only to fill the requested N. + vis_cells = select_visual_cells(pool, visual, fire_cells) report["visual"] = _visual_registration(qa, engine, mask, ortho, vis_cells, out, - settle, move_timeout) + settle, move_timeout, fire_cells=fire_cells) # a requested visual gate that measured NOTHING must fail loud, never read as a vacuous GREEN if not report["visual"]["cases"]: report["visual"]["fail"] += 1 @@ -662,12 +760,14 @@ def _drive_and_check(qa: str, engine: str, c: int, r: int, settle: float, timeou return False, f"drive-error:{e}", None deadline = time.time() + timeout landed, path = before, None + any_surface = False # did ANY poll after the click actually read the surface? while time.time() < deadline: time.sleep(settle) try: surf = _get(f"{engine}/combat-surface") except Exception: # noqa: BLE001 continue + any_surface = True landed = _token_cell(surf) # combat moves ride lastPath; rest walks ride the additive lastWalkPath (#1582 — before it, # rest-mode path audits were silently vacuous because the surface's lastPath is combat-only) @@ -676,6 +776,11 @@ def _drive_and_check(qa: str, engine: str, c: int, r: int, settle: float, timeou return True, list(landed), path if not expect_move and landed == (c, r): return False, list(landed), path # moved onto an impassable cell — a real failure + # the engine died mid-probe (ZERO successful surface reads after the click) → a HARNESS error, not + # a verdict: an impassable check would otherwise false-PASS on the stale `before` cell and a + # reachable check would false-RED. Partial outages (≥1 good poll) keep the normal timeout semantics. + if not any_surface: + return False, f"drive-error:engine surface unreachable after click ({c},{r})", None if expect_move: return (landed == (c, r)), (list(landed) if landed else None), path # impassable: pass iff the token never became (c,r)