From 5f9e29576556f635757eea0605d5a23d29ce1cc5 Mon Sep 17 00:00:00 2001 From: Sam Gu <127461594+SamGu-NRX@users.noreply.github.com> Date: Thu, 15 Oct 2026 22:01:42 +0000 Subject: [PATCH] Enforce every reported acceptance check in the placement replay runner The runner previously trusted aggregate verdicts it could not have earned. It now binds each solver response's stats.input_sha256 to the exact bytes each path sent, compares the converted scene field-by-field against independently authored wire fixtures, checks a metre-to-feet sentinel, served-result HTTP/outcome/run identity, finalize-replay idempotence, and requires the positive cases to carry distinct normalized geometry (identity stamps are not evidence: one geometry under two case ids is a blocker). Positive refusals and negative solutions are blockers either way, and case selection cannot pass vacuously. The committed manifest's positive sources refuse conversion at the bridge (stand-off meter anchor), so the success path is exercised through a temporary manifest whose sources, wire scenes, and pinned expectations are authored by hand in the tests - the same geometry in metres and in feet, never produced by running the converter. Fault injections prove the checks are load-bearing: a corrupted converter, a forged input hash in both replies, a lying served result, a non-idempotent replay, a declared negative that solves, and a substituted boundary scene all fail the suite. --- backend/scripts/capture_placement_cases.py | 393 ++++++++- backend/tests/test_capture_placement_cases.py | 819 +++++++++++++++++- 2 files changed, 1141 insertions(+), 71 deletions(-) diff --git a/backend/scripts/capture_placement_cases.py b/backend/scripts/capture_placement_cases.py index 1418752d1..bc9506f49 100644 --- a/backend/scripts/capture_placement_cases.py +++ b/backend/scripts/capture_placement_cases.py @@ -34,15 +34,34 @@ Checks per case (any mismatch fails the suite with a nonzero exit): - The merged scene in the run's artifacts equals the independently computed - stamp + merge of that case's source — a substituted reference scene, another + stamp + merge of that case's source — the merge contract is re-derived here + (same-id precedence asserted explicitly, never delegated to the production + ``merge_object_lists`` helper) — so a substituted reference scene, another case's source, or a second merge is caught here. - The placement stage's typed refusal (code, pointer) for source cases the - bridge refuses; no placement artifact on refusal; the served device result - stays conservative (``manual_review``) with exactly one terminal event. -- When a case converts: the stage response equals a direct ``POST /v1/placements`` - on exactly the captured converted bytes, and the decision semantics match the - independently authored wire case's expectations (input hashes differ by - identity stamp, so hashes are never compared across the pair). + bridge refuses; no placement artifact on refusal; a positive record must + solve (``status == done`` plus a real placement artifact — a declared refusal + is still a blocker); a negative case that somehow solves is a blocker, never + a success; the served device result stays conservative (``manual_review``, + HTTP 200, the journey's own run id) with exactly one terminal event. +- When a case converts: the captured converted bytes are compared field by + field against the independently authored wire scene (identity stripped, + floats canonicalized) — covering the tap hazard, units, evidence-derived + coverage and chain ends — and the decision semantics match the wire case's + expectations. Both solver responses' ``stats.input_sha256`` must match their + own exact request bytes (the stage's against the captured converted bytes, + the direct response's against the bytes posted), so forging the same wrong + hash in both replies is caught even though they then agree with each other. + +Aggregate checks: ``crossCapture`` compares the run's two positive cases and +requires distinct capture ids with distinct *normalized* geometry — identity +fields stripped and the rest canonicalized, so different identity stamps or +stamped scene hashes alone never pass. ``finalizeReplay`` must be an idempotent +success on the same run id. A full run (no ``--cases`` selection) requires both +positive cases present and solved; a subset diagnostic run marks the aggregate +checks it cannot make ``not_applicable``; an empty or unknown selection cannot +pass. Every failed check contributes a blocker, and the receipt is ``ok`` only +when there are none. Findings this run pins (the reason the suite exits 1 on the committed fixtures): @@ -66,6 +85,7 @@ import copy import hashlib import json +import math import os import shutil import subprocess @@ -88,7 +108,6 @@ from server.orchestrator.stateio import get_json # noqa: E402 from server.orchestrator.stubs._common import json_bytes, write_files # noqa: E402 from server.placement import adapter, case_suite # noqa: E402 -from server.scene.merge import merge_object_lists # noqa: E402 RECEIPT_SCHEMA_VERSION = "hss-capture-placement-receipt/v0" TERMINAL_EVENTS = {"verdict_ready", "retake_request", "failed"} @@ -433,9 +452,137 @@ def _run_artifacts(harness: _Harness, run_id: str) -> dict[str, str]: return artifacts +# The merge contract, re-derived here without the production +# ``server.scene.merge.merge_object_lists`` helper: incoming objects append, and a +# detector or VLM scene object leaves only when an incoming object takes its id or +# its place (same class, wall_rect on the same wall, or the same centre within the +# merge tolerance). Re-deriving keeps the oracle independent: a coordinator bug and +# an oracle bug can no longer be the same bug. +_MERGE_SAME_PLACE_TOL_M = 0.3 + + +def _object_center_m(obj: dict[str, Any]) -> tuple[float, float, float] | None: + geometry = obj.get("geometry") or {} + if geometry.get("kind") == "point": + x, y, z = (float(v) for v in geometry["pos_m"]) + return x, y, z + if geometry.get("kind") == "wall_rect": + span = geometry.get("s_m") or [0.0, 0.0] + return ( + (float(span[0]) + float(span[1])) / 2, + (float(geometry.get("bottom_m", 0.0)) + float(geometry.get("top_m", 0.0))) / 2, + 0.0, + ) + return None + + +def _same_place(a: dict[str, Any], b: dict[str, Any]) -> bool: + if a.get("class") != b.get("class"): + return False + center_a, center_b = _object_center_m(a), _object_center_m(b) + if center_a is None or center_b is None: + return False + geometry_a, geometry_b = a.get("geometry") or {}, b.get("geometry") or {} + if geometry_a.get("kind") == "wall_rect" and geometry_b.get("kind") == "wall_rect": + if geometry_a.get("wall_id") != geometry_b.get("wall_id"): + return False + return abs(center_a[0] - center_b[0]) <= _MERGE_SAME_PLACE_TOL_M and abs(center_a[1] - center_b[1]) <= ( + _MERGE_SAME_PLACE_TOL_M + ) + return math.dist(center_a, center_b) <= _MERGE_SAME_PLACE_TOL_M + + +def _replaced_independently(prev: dict[str, Any], incoming: list[dict[str, Any]]) -> bool: + kind = (prev.get("source") or {}).get("kind") + if kind not in {"detector", "vlm"}: + return False + for obj in incoming: + if prev.get("id") and prev.get("id") == obj.get("id"): + return True + if _same_place(prev, obj): + return True + return False + + +def _assert_merge_precedence(merged: dict[str, Any], journey: dict[str, Any], mismatches: list[str]) -> None: + """Same-id precedence, asserted explicitly: every injected object appears in the + merged scene exactly once and the merged copy *is* the injected object.""" + entry = _REGISTRY[journey["captureId"]] + merged_objects = merged.get("objects") or [] + for incoming in entry["objects_3d"]: + object_id = incoming.get("id") + matches = [obj for obj in merged_objects if obj.get("id") == object_id] + if len(matches) != 1: + mismatches.append( + f"injected object {object_id!r} appears {len(matches)} times after the merge, not exactly once" + ) + elif matches[0] != incoming: + mismatches.append( + f"injected object {object_id!r} lost same-id precedence: the merged copy is not the injected object" + ) + + +def _canonical(value: Any) -> Any: + if value is None or isinstance(value, (bool, str)): + return value + if isinstance(value, (int, float)): + return round(float(value), 6) + if isinstance(value, list): + return [_canonical(item) for item in value] + if isinstance(value, dict): + return {key: _canonical(item) for key, item in sorted(value.items())} + return value + + +def _normalized_geometry(document: dict[str, Any]) -> dict[str, Any]: + """Identity-stripped, canonicalized geometry. Hashes of normalized geometry are + equal only when the geometry itself is equal: different identity stamps or + different stamped scene hashes prove nothing here.""" + stripped = { + key: value for key, value in document.items() if key not in {"scene_id", "provenance", "schema_version"} + } + return _canonical(stripped) + + +def _geometry_diffs(converted: Any, expected: Any, path: str = "", limit: int = 5) -> list[str]: + """The first structural differences between the converted scene and the + independently authored wire scene, as readable paths.""" + if isinstance(converted, dict) and isinstance(expected, dict): + diffs: list[str] = [] + for key in sorted(set(converted) | set(expected)): + if len(diffs) >= limit: + return diffs + if key not in converted: + diffs.append(f"{path}/{key}: missing from the converted scene") + elif key not in expected: + diffs.append(f"{path}/{key}: not in the wire scene") + else: + diffs.extend(_geometry_diffs(converted[key], expected[key], f"{path}/{key}", limit - len(diffs))) + return diffs + if isinstance(converted, list) and isinstance(expected, list): + diffs = [] + if len(converted) != len(expected): + diffs.append(f"{path}: {len(converted)} entries, wire scene has {len(expected)}") + else: + for index, (a, b) in enumerate(zip(converted, expected, strict=True)): + if len(diffs) >= limit: + return diffs + diffs.extend(_geometry_diffs(a, b, f"{path}/{index}", limit - len(diffs))) + return diffs + if _canonical(converted) != _canonical(expected): + return [f"{path}: {converted!r} != wire {expected!r}"] + return [] + + +# One metre in feet, by definition: every converted length is the source metres +# divided by exactly 0.3048. The unit sentinel re-derives one length end to end. +M_PER_FT = 0.3048 + + def _expected_merged_scene(journey: dict[str, Any]) -> dict[str, Any]: """The merged scene this case must produce, computed independently of the - pipeline: the stamped source plus the injected objects, merged once.""" + pipeline: the stamped source plus the injected objects, merged once under the + contract re-derived above.""" entry = _REGISTRY[journey["captureId"]] stamped = _stamped_scene( entry["record"], @@ -444,7 +591,77 @@ def _expected_merged_scene(journey: dict[str, Any]) -> dict[str, Any]: capture_id=journey["captureId"], run_id=journey["runId"], ) - return merge_object_lists(stamped, copy.deepcopy(entry["objects_3d"])) + incoming = copy.deepcopy(entry["objects_3d"]) + kept = [obj for obj in stamped.get("objects") or [] if not _replaced_independently(obj, incoming)] + expected = dict(stamped) + expected["objects"] = kept + incoming + return expected + + +def _check_converted_geometry( + case: dict[str, Any], + record: case_suite.CaseRecord, + *, + converted_bytes: bytes, + merged_bytes: bytes | None, + mismatches: list[str], +) -> None: + """The captured converted scene against independently derived facts: the unit + sentinel, the preserved chain ends, the tap hazard's once-only conversion, and + a field-by-field comparison with the authored wire scene.""" + converted = json.loads(converted_bytes) + merged = json.loads(merged_bytes) if merged_bytes else {} + + source_height_m = (merged.get("walls") or [{}])[0].get("height_m") + converted_height_ft = (converted.get("walls") or [{}])[0].get("height_ft") + units_ok = ( + source_height_m is not None + and converted_height_ft is not None + and abs(float(converted_height_ft) - float(source_height_m) / M_PER_FT) <= 1e-6 + ) + case["unitsSentinel"] = "matched" if units_ok else "mismatched" + if not units_ok: + mismatches.append("converted wall height is not the merged source height divided by 0.3048 (units sentinel)") + + chain_ends = (merged.get("chain") or {}).get("ends") or {} + converted_ends = (converted.get("coverage") or {}).get("ends") or {} + for side, end in sorted(chain_ends.items()): + converted_kind = (converted_ends.get(side) or {}).get("kind") + if converted_kind != end.get("kind"): + mismatches.append( + f"converted coverage end {side!r} is {converted_kind!r}, not the source chain's {end.get('kind')!r}" + ) + + if record.id == TAP_CASE_ID: + expected_span_ft = [value / M_PER_FT for value in TAP_HAZARD["geometry"]["s_m"]] + taps = [ + obj + for obj in converted.get("objects") or [] + if obj.get("type") == TAP_HAZARD["class"] + and obj.get("source") == TAP_HAZARD["source"]["kind"] + and len(obj.get("span_ft") or []) == 2 + and all(abs(a - b) <= 1e-6 for a, b in zip(obj["span_ft"], expected_span_ft, strict=True)) + ] + case["tapHazardConvertedCount"] = len(taps) + if len(taps) != 1: + mismatches.append( + f"the observed tap hazard appears {len(taps)} times in the converted scene at its expected feet " + "span, not exactly once" + ) + + if record.wire_path is None: + case["geometryComparison"] = "not_applicable" + return + wire = json.loads(record.wire_path.read_bytes()) + normalized = _normalized_geometry(converted) + diffs = _geometry_diffs(normalized, _normalized_geometry(wire)) + case["geometryComparison"] = "matched" if not diffs else "mismatched" + case["geometryDiffs"] = diffs + # Canonical normalized-geometry hash, identity stamps stripped: the + # cross-capture comparison fails same-geometry-different-identity runs. + case["mergedGeometrySha256"] = _sha256(json.dumps(normalized, sort_keys=True).encode()) + if diffs: + mismatches.append("converted scene differs from the independently authored wire scene: " + "; ".join(diffs[:3])) def _decision_bytes(payload: dict[str, Any]) -> str: @@ -513,27 +730,63 @@ def _collect_case(harness: _Harness, record: case_suite.CaseRecord, journey: dic case["hasPlacementArtifact"] = bool(placement_artifact_key) and deps.storage.exists(placement_artifact_key) if stage_status == "done" and case["hasPlacementArtifact"]: + if record.kind != case_suite.KIND_POSITIVE: + case["mismatches"].append( + f"{record.kind!r} case produced a successful placement; only a positive case may solve" + ) case["status"] = "solved" + converted_bytes = deps.storage.get_bytes(converted_key) if converted_key else None placement = json.loads(deps.storage.get_bytes(placement_artifact_key)) - case["convertedSha256"] = _sha256(deps.storage.get_bytes(converted_key)) + case["convertedSha256"] = _sha256(converted_bytes) if converted_bytes else None case["resultSha256"] = _sha256(deps.storage.get_bytes(placement_artifact_key)) case["solverPolicy"] = placement.get("policy") - direct = TestClient(adapter.load_api()["api"].app).post( - "/v1/placements", - content=deps.storage.get_bytes(converted_key), - headers={"Content-Type": "application/json"}, + + # Both solver responses must bind their own input hash to their own exact + # request bytes. Forging the same wrong hash in both replies leaves them in + # agreement with each other and still fails here. + stage_input_sha = (placement.get("stats") or {}).get("input_sha256") + case["stageInputShaMatchesCapturedBytes"] = stage_input_sha == case["convertedSha256"] + if not case["stageInputShaMatchesCapturedBytes"]: + case["mismatches"].append( + "the stage response's stats.input_sha256 does not match the sha256 of the captured converted bytes it solved" + ) + + _check_converted_geometry( + case, + record, + converted_bytes=converted_bytes or b"{}", + merged_bytes=merged_bytes, + mismatches=case["mismatches"], ) - if direct.status_code != 200: - case["errors"].append(f"direct_placements_http_{direct.status_code}") + + if converted_bytes is None: + case["errors"].append("the placement stage solved without publishing placement/scene.json") case["directComparison"] = "error" else: - case["directComparison"] = ( - "matched" if _decision_bytes(direct.json()) == _decision_bytes(placement) else "mismatched" + direct = TestClient(adapter.load_api()["api"].app).post( + "/v1/placements", + content=converted_bytes, + headers={"Content-Type": "application/json"}, ) - if case["directComparison"] == "mismatched": - case["mismatches"].append( - "direct POST /v1/placements on the captured bytes disagrees with the stage response" + if direct.status_code != 200: + case["errors"].append(f"direct_placements_http_{direct.status_code}") + case["directComparison"] = "error" + else: + direct_body = direct.json() + direct_sha = (direct_body.get("stats") or {}).get("input_sha256") + case["directInputShaMatchesPostedBytes"] = direct_sha == case["convertedSha256"] + if not case["directInputShaMatchesPostedBytes"]: + case["mismatches"].append( + "the direct solver response's stats.input_sha256 does not match the sha256 of the exact " + "request bytes" + ) + case["directComparison"] = ( + "matched" if _decision_bytes(direct_body) == _decision_bytes(placement) else "mismatched" ) + if case["directComparison"] == "mismatched": + case["mismatches"].append( + "direct POST /v1/placements on the captured bytes disagrees with the stage response" + ) if record.wire_path is not None: wire_mismatches = case_suite._evaluate_expectations(record.expectations, placement) case["wireSemanticComparison"] = "matched" if not wire_mismatches else "mismatched" @@ -562,10 +815,16 @@ def _collect_case(harness: _Harness, record: case_suite.CaseRecord, journey: dic f"refusal pointer {declared.get('pointer')!r} not found in the refusal message" ) elif record.kind == case_suite.KIND_POSITIVE: - case["mismatches"].append( - "positive case refused conversion, but the manifest declares no expected error for it " - "(the source does not pass the scene bridge as authored)" - ) + if declared: + case["mismatches"].append( + "positive case refused conversion; even with a declared error a positive record must solve " + "and publish a placement artifact" + ) + else: + case["mismatches"].append( + "positive case refused conversion, but the manifest declares no expected error for it " + "(the source does not pass the scene bridge as authored)" + ) case["convertedSha256"] = None case["resultSha256"] = None case["solverPolicy"] = None @@ -577,14 +836,18 @@ def _collect_case(harness: _Harness, record: case_suite.CaseRecord, journey: dic f"placement stage {stage_status!r} with error {stage_error!r}: neither solved nor refused" ) - # The device boundary: conservative outcome, one terminal event. + # The device boundary: conservative outcome, the journey's own identity, one + # terminal event. Criteria and the result stay separately stubbed, so this pins + # the served boundary, never a claim that the decision caused the device state. client = harness.client capture = client.get(f"/v1/captures/{journey['captureId']}").json() served = client.get(f"/v1/captures/{journey['captureId']}/result") + served_body = served.json() if served.status_code == 200 else {} case["deviceOutcome"] = { "captureStatus": capture.get("status"), "servedResultHttpStatus": served.status_code, - "servedResultKind": served.json().get("outcome", {}).get("kind") if served.status_code == 200 else None, + "servedResultKind": (served_body.get("outcome") or {}).get("kind"), + "servedResultRunId": served_body.get("runId"), } terminals = [event for event in journey["events"] if event["type"] in TERMINAL_EVENTS] case["terminalEventCount"] = len(terminals) @@ -594,6 +857,17 @@ def _collect_case(harness: _Harness, record: case_suite.CaseRecord, journey: dic case["mismatches"].append( f"capture status {case['deviceOutcome']['captureStatus']!r} is not the conservative manual_review" ) + if served.status_code != 200: + case["mismatches"].append(f"GET /result returned HTTP {served.status_code}, not 200") + elif case["deviceOutcome"]["servedResultKind"] != "manual_review": + case["mismatches"].append( + f"served result outcome {case['deviceOutcome']['servedResultKind']!r} is not the conservative manual_review" + ) + if case["deviceOutcome"]["servedResultRunId"] != journey.get("runId"): + case["mismatches"].append( + f"served result run id {case['deviceOutcome']['servedResultRunId']!r} does not match this journey's " + f"run id {journey.get('runId')!r}" + ) return case @@ -606,7 +880,10 @@ def run_cases( """Run the manifest's cases through the front door and return the receipt.""" _apply_dev_env(workdir / "vol") workdir.mkdir(parents=True, exist_ok=True) + if case_ids is not None and not case_ids: + raise SystemExit("empty case selection: name at least one case id, or run every case") records = case_suite.load_cases(manifest_path) + full_run = case_ids is None if case_ids: wanted = set(case_ids) records = [record for record in records if record.id in wanted] @@ -631,15 +908,40 @@ def run_cases( journeys.append(_walk_front_door(harness, record, packet_id=f"placement-case-{index + 1:02d}")) harness.drain() cases = [_collect_case(harness, record, journey) for record, journey in zip(records, journeys, strict=True)] - replay = _replay_finalize(harness, journeys[0]) if journeys else None + if journeys and journeys[0].get("captureId") is not None: + replay = _replay_finalize(harness, journeys[0]) + replay["applicable"] = True + elif journeys: + replay = { + "applicable": False, + "status": "not_applicable", + "reason": "the first journey never created a capture", + "ok": None, + } + else: + replay = {"applicable": False, "status": "not_applicable", "reason": "no case ran", "ok": None} finally: _remove_boundaries() - # Cross-capture: distinct captures in one app must carry distinct scenes. - merged = [case.get("mergedSceneSha256") for case in cases[:2]] + # Cross-capture: distinct captures in one app must carry distinct scenes — and + # distinct *normalized* geometry, not merely distinct identity stamps or + # stamped scene hashes. Only the two positive cases are compared. + positive_cases = [case for case in cases if case["kind"] == case_suite.KIND_POSITIVE] + pair = positive_cases[:2] + raw_hashes = [case.get("mergedSceneSha256") for case in pair] + geometry_hashes = [case.get("mergedGeometrySha256") for case in pair] + capture_ids = [case.get("captureId") for case in pair] + applicable = len(pair) == 2 and all(raw_hashes) and all(geometry_hashes) and all(capture_ids) cross_capture = { - "captureIds": [case.get("captureId") for case in cases[:2]], - "distinctScenes": len(merged) == 2 and merged[0] != merged[1], + "applicable": applicable, + "status": "evaluated" if applicable else "not_applicable", + "reason": None + if applicable + else "not_applicable: this run does not include two positive cases that each carried a merged scene", + "captureIds": capture_ids, + "distinctScenes": (raw_hashes[0] != raw_hashes[1]) if applicable else None, + "distinctNormalizedGeometry": (geometry_hashes[0] != geometry_hashes[1]) if applicable else None, + "mergedGeometrySha256": dict(zip([case["id"] for case in pair], geometry_hashes, strict=False)) if pair else {}, } blockers = [] @@ -658,6 +960,31 @@ def run_cases( blockers.append(f"{case['id']}: " + "; ".join(case["mismatches"])) if case["errors"]: blockers.append(f"{case['id']}: " + "; ".join(case["errors"])) + if cross_capture["applicable"]: + if not ( + cross_capture["distinctScenes"] + and cross_capture["distinctNormalizedGeometry"] + and len(set(cross_capture["captureIds"])) == 2 + ): + blockers.append( + "crossCapture: the two positive cases do not carry distinct capture ids with distinct normalized " + f"geometry (distinctScenes={cross_capture['distinctScenes']}, " + f"distinctNormalizedGeometry={cross_capture['distinctNormalizedGeometry']})" + ) + if replay is not None and replay.get("applicable", True) and replay.get("ok") is not True: + blockers.append( + "finalizeReplay: the finalize replay is not an idempotent success (HTTP 202, same run id, one terminal " + f"event); got {json.dumps(replay)}" + ) + if full_run: + if len(positive_cases) < 2: + blockers.append(f"full run requires both positive cases; this manifest ran {len(positive_cases)}") + unsolved = [case["id"] for case in positive_cases if case["status"] != "solved"] + if unsolved: + blockers.append( + "full run requires both positive cases to solve and publish placement artifacts; not solved: " + + ", ".join(unsolved) + ) statuses = [case["status"] for case in cases] receipt: dict[str, Any] = { diff --git a/backend/tests/test_capture_placement_cases.py b/backend/tests/test_capture_placement_cases.py index 2d109b024..28760438c 100644 --- a/backend/tests/test_capture_placement_cases.py +++ b/backend/tests/test_capture_placement_cases.py @@ -1,41 +1,21 @@ -"""Focused tests for the capture placement case runner. - -The runner (`scripts/capture_placement_cases.py`) replays every hash-pinned case -in `server/placement/tests/data/cases/manifest.json` through the real capture -front door — create, register, signed upload, commit, finalize, events, result — -over the documented dev settings (SD-73), injecting each case's synthetic source -scene **only** at the upstream `scene` stage boundary and an explicit object list -at the `objects_3d` boundary, with the actual capture/run/packet identity stamped -in the contract's identity fields. - -These tests pin the three claims that make the runner worth trusting: - -1. The journey evidence: the coordinator's merged scene equals the case's - stamped source merged once with the injected objects (never a reference - scene, another capture's scene, or a second merge); the observed tap hazard - lands exactly once; refusals are typed, leave no placement artifact, and end - in the conservative device outcome with exactly one terminal event; two - captures in one app carry distinct scenes; a finalize replay is idempotent. -2. The suite fails loudly on tampering: substituting the old default stub's - reference scene at the boundary, or removing the real placement stage, both - produce mismatches and a nonzero exit — the boundary checks are load-bearing, - not decorative. -3. The committed fixtures' pinned state (2026-10-15): both positive cases refuse - conversion at `/objects/0/geometry/standoff_m` (a stand-off meter anchor the - bridge does not synthesise, then the footprint hazard kind) with no expected - error declared in the manifest, and `generated-hazard-unsupported` refuses at - `/generated_regions` rather than its declared `/objects/1/evidence`. The - suite reports these as blockers and exits nonzero: a producer-contract - mismatch, recorded here as evidence rather than papered over. +"""Acceptance tests for the capture placement replay runner. + +The committed manifest's positive sources refuse conversion (the pinned blocker), +so the success path is exercised through a temporary manifest whose positive +sources, wire scenes, and expectations are authored by hand in this module — the +same geometry in metres and in feet, never produced by running the converter. """ from __future__ import annotations +import hashlib import importlib import json +import subprocess import sys from pathlib import Path +import httpx import pytest BACKEND = Path(__file__).resolve().parents[1] @@ -52,6 +32,18 @@ POSITIVE_IDS = ("hazard-right-ground-gap", "hazard-left-ground-gap") NEGATIVE_IDS = ("multiwall-source-unsupported", "generated-hazard-unsupported") +M_PER_FT = 0.3048 + + +def _ft(metres: float) -> float: + """Feet for a source metres value — the conversion the bridge must apply, + used here to author the wire scenes by hand, not by running the converter.""" + return metres / M_PER_FT + + +def _sha256_file(path: Path) -> str: + return hashlib.sha256(path.read_bytes()).hexdigest() + @pytest.fixture def receipt(tmp_path, monkeypatch): @@ -64,6 +56,507 @@ def _case(receipt: dict, case_id: str) -> dict: return next(case for case in receipt["cases"] if case["id"] == case_id) +# --------------------------------------------------------------------------- +# Hand-authored supported sources, wire scenes, and pinned expectations +# --------------------------------------------------------------------------- + + +def _supported_source_a() -> dict: + """A hand-authored positive source: one wall, a meter anchor point, a window, + a tap gas hazard, an observed ground pad, overhead, facing gap, coverage. + The solver has plausible spots but unsure evidence forces manual review.""" + return { + "format": "base-scene/v0", + "scene_id": "burst8-supported-a", + "provenance": { + "data": "synthetic", + "site": "burst8-supported-a", + "producer": {"path": "capture-placement-cases-test"}, + "scale_source": "native_metric", + "gravity_source": "ground_truth", + }, + "frame": { + "units": "m", + "convention": "meter_anchored/v0", + "anchor": {"element_id": "meter", "plus_minus_m": 0.01}, + }, + "walls": [ + { + "id": "side", + "role": "house", + "baseline_m": [[-3.0, 0.0], [5.0, 0.0]], + "height_m": 3.0, + "plus_minus_m": 0.05, + "evidence": "observed", + } + ], + "chain": { + "wall_ids": ["side"], + "ends": {"left": {"kind": "unexplored"}, "right": {"kind": "limit", "note": "fence line"}}, + }, + "ground": { + "surface": {"kind": "plane", "normal": [0, 1, 0], "offset_m": 0}, + "plus_minus_m": 0.05, + "regions": [ + { + "id": "pad", + "material": "concrete", + "polygon_m": [[0.2, 0.5], [2.0, 0.5], [2.0, 3.0], [0.2, 3.0]], + "evidence": "observed", + "plus_minus_m": 0.1, + } + ], + }, + "objects": [ + { + "id": "meter", + "class": "meter", + "geometry": {"kind": "point", "pos_m": [0.0, 1.5, 0.02]}, + "confidence": 1, + "plus_minus_m": 0.02, + "source": {"kind": "reconstruction"}, + "evidence": "observed", + }, + { + "id": "win", + "class": "window", + "geometry": {"kind": "wall_rect", "wall_id": "side", "s_m": [1.0, 2.0], "bottom_m": 1.0, "top_m": 2.2}, + "confidence": 0.9, + "plus_minus_m": 0.08, + "source": {"kind": "vlm"}, + "evidence": "observed", + }, + { + "id": "gas", + "class": "gas_meter", + "geometry": { + "kind": "wall_rect", + "wall_id": "side", + "s_m": [-0.6, -0.3], + "bottom_m": 0.3, + "top_m": 0.9, + }, + "confidence": 0.95, + "plus_minus_m": 0.05, + "source": {"kind": "tap"}, + "evidence": "observed", + "attrs": {"operable": True}, + }, + ], + "coverage": { + "method": "capture-placement-cases-test", + "sensitivity": [ + { + "surface": "wall", + "max_incidence_deg": 75, + "bands": [{"band": "wall", "s_m": [-2.0, 4.0], "out_m": 2.13}], + } + ], + }, + "overheads": [ + {"wall_id": "side", "s_m": [1.0, 2.0], "clearance_m": 2.1, "plus_minus_m": 0.1, "evidence": "observed"} + ], + "facing": [ + {"wall_id": "side", "s_m": [-1.0, 1.0], "depth_m": 8.0, "plus_minus_m": 0.3, "evidence": "observed"} + ], + } + + +def _supported_source_b() -> dict: + """A second, mirrored positive source: the gas hazard right of the meter, the + window left, observed ground on the left, the left end a real limit and the + right end unexplored — the opposite of source A on every one of those.""" + return { + "format": "base-scene/v0", + "scene_id": "burst8-supported-b", + "provenance": { + "data": "synthetic", + "site": "burst8-supported-b", + "producer": {"path": "capture-placement-cases-test"}, + "scale_source": "native_metric", + "gravity_source": "ground_truth", + }, + "frame": { + "units": "m", + "convention": "meter_anchored/v0", + "anchor": {"element_id": "meter", "plus_minus_m": 0.01}, + }, + "walls": [ + { + "id": "side", + "role": "house", + "baseline_m": [[-5.0, 0.0], [3.0, 0.0]], + "height_m": 2.8, + "plus_minus_m": 0.06, + "evidence": "observed", + } + ], + "chain": { + "wall_ids": ["side"], + "ends": {"left": {"kind": "limit", "note": "hedge line"}, "right": {"kind": "unexplored"}}, + }, + "ground": { + "surface": {"kind": "plane", "normal": [0, 1, 0], "offset_m": 0}, + "plus_minus_m": 0.05, + "regions": [ + { + "id": "pad", + "material": "gravel", + "polygon_m": [[-2.0, 0.5], [-0.6, 0.5], [-0.6, 3.0], [-2.0, 3.0]], + "evidence": "observed", + "plus_minus_m": 0.08, + } + ], + }, + "objects": [ + { + "id": "meter", + "class": "meter", + "geometry": {"kind": "point", "pos_m": [0.0, 1.4, 0.03]}, + "confidence": 1, + "plus_minus_m": 0.03, + "source": {"kind": "reconstruction"}, + "evidence": "observed", + }, + { + "id": "win", + "class": "window", + "geometry": { + "kind": "wall_rect", + "wall_id": "side", + "s_m": [-2.4, -1.4], + "bottom_m": 1.0, + "top_m": 2.0, + }, + "confidence": 0.9, + "plus_minus_m": 0.08, + "source": {"kind": "vlm"}, + "evidence": "observed", + }, + { + "id": "gas", + "class": "gas_meter", + "geometry": { + "kind": "wall_rect", + "wall_id": "side", + "s_m": [0.8, 1.3], + "bottom_m": 0.25, + "top_m": 0.85, + }, + "confidence": 0.95, + "plus_minus_m": 0.05, + "source": {"kind": "tap"}, + "evidence": "observed", + }, + ], + "coverage": { + "method": "capture-placement-cases-test", + "sensitivity": [ + { + "surface": "wall", + "max_incidence_deg": 75, + "bands": [{"band": "wall", "s_m": [-4.0, 2.0], "out_m": 2.5}], + } + ], + }, + "overheads": [ + {"wall_id": "side", "s_m": [-2.4, -1.4], "clearance_m": 2.0, "plus_minus_m": 0.12, "evidence": "observed"} + ], + "facing": [ + {"wall_id": "side", "s_m": [-1.5, 0.5], "depth_m": 7.0, "plus_minus_m": 0.25, "evidence": "observed"} + ], + } + + +def _wire_a(*, with_tap: bool = True) -> dict: + """The wire scene a converter must produce for source A, authored from the + source geometry with metres divided by 0.3048. The observed tap hazard is + injected by the runner for the reserved positive id, so the wire scene + includes it as a third object unless a caller excludes it.""" + wire = { + "meter": {"pos": [0.0, _ft(1.5), _ft(0.02)], "wall_id": "side", "plus_minus_ft": _ft(0.02)}, + "walls": [ + { + "id": "side", + "baseline": [[_ft(-3.0), 0.0], [_ft(5.0), 0.0]], + "height_ft": _ft(3.0), + "plus_minus_ft": _ft(0.05), + } + ], + "objects": [ + { + "type": "window", + "wall_id": "side", + "span_ft": [_ft(1.0), _ft(2.0)], + "bottom_ft": _ft(1.0), + "top_ft": _ft(2.2), + "conf": 0.9, + "plus_minus_ft": _ft(0.08), + "source": "vlm", + }, + { + "type": "gas_meter", + "wall_id": "side", + "span_ft": [_ft(-0.6), _ft(-0.3)], + "bottom_ft": _ft(0.3), + "top_ft": _ft(0.9), + "conf": 0.95, + "plus_minus_ft": _ft(0.05), + "source": "tap", + "attrs": {"operable": True}, + }, + ], + "ground": [ + { + "type": "concrete", + "polygon": [ + [_ft(0.2), _ft(0.5)], + [_ft(2.0), _ft(0.5)], + [_ft(2.0), _ft(3.0)], + [_ft(0.2), _ft(3.0)], + ], + "plus_minus_ft": _ft(0.1), + } + ], + "overheads": [ + {"wall_id": "side", "span_ft": [_ft(1.0), _ft(2.0)], "clearance_ft": _ft(2.1), "plus_minus_ft": _ft(0.1)} + ], + "facing": [ + {"wall_id": "side", "span_ft": [_ft(-1.0), _ft(1.0)], "depth_ft": _ft(8.0), "plus_minus_ft": _ft(0.3)} + ], + "coverage": { + "ends": {"left": {"kind": "unexplored"}, "right": {"kind": "limit", "note": "fence line"}}, + "observed": [{"band": "wall", "span_ft": [_ft(-2.0), _ft(4.0)], "out_ft": _ft(2.13)}], + }, + } + if with_tap: + wire["objects"].append( + { + "type": "vent", + "wall_id": "side", + "span_ft": [_ft(1.6), _ft(1.8)], + "bottom_ft": _ft(0.5), + "top_ft": _ft(2.0), + "conf": 0.9, + "plus_minus_ft": _ft(0.05), + "source": "tap", + } + ) + return wire + + +def _wire_b() -> dict: + """The wire scene for source B, authored the same way — from the metres.""" + return { + "meter": {"pos": [0.0, _ft(1.4), _ft(0.03)], "wall_id": "side", "plus_minus_ft": _ft(0.03)}, + "walls": [ + { + "id": "side", + "baseline": [[_ft(-5.0), 0.0], [_ft(3.0), 0.0]], + "height_ft": _ft(2.8), + "plus_minus_ft": _ft(0.06), + } + ], + "objects": [ + { + "type": "window", + "wall_id": "side", + "span_ft": [_ft(-2.4), _ft(-1.4)], + "bottom_ft": _ft(1.0), + "top_ft": _ft(2.0), + "conf": 0.9, + "plus_minus_ft": _ft(0.08), + "source": "vlm", + }, + { + "type": "gas_meter", + "wall_id": "side", + "span_ft": [_ft(0.8), _ft(1.3)], + "bottom_ft": _ft(0.25), + "top_ft": _ft(0.85), + "conf": 0.95, + "plus_minus_ft": _ft(0.05), + "source": "tap", + }, + ], + "ground": [ + { + "type": "gravel", + "polygon": [ + [_ft(-2.0), _ft(0.5)], + [_ft(-0.6), _ft(0.5)], + [_ft(-0.6), _ft(3.0)], + [_ft(-2.0), _ft(3.0)], + ], + "plus_minus_ft": _ft(0.08), + } + ], + "overheads": [ + {"wall_id": "side", "span_ft": [_ft(-2.4), _ft(-1.4)], "clearance_ft": _ft(2.0), "plus_minus_ft": _ft(0.12)} + ], + "facing": [ + {"wall_id": "side", "span_ft": [_ft(-1.5), _ft(0.5)], "depth_ft": _ft(7.0), "plus_minus_ft": _ft(0.25)} + ], + "coverage": { + "ends": {"left": {"kind": "limit", "note": "hedge line"}, "right": {"kind": "unexplored"}}, + "observed": [{"band": "wall", "span_ft": [_ft(-4.0), _ft(2.0)], "out_ft": _ft(2.5)}], + }, + } + + +def _expected_a() -> dict: + """Geometry facts, not policy: pinned from the solver's deterministic output + on exactly the merged scene this manifest produces (tap hazard included).""" + return { + "decision": "manual_review", + "spot": { + "wall_id": "side", + "segment": 0, + "span_ft": [9.56168, 12.145013], + "outcome": "unsure", + }, + "checks": { + "gas_clearance": {"outcome": "unsure", "unsure_cause": "unobserved"}, + "ground_surface": {"outcome": "unsure", "unsure_cause": "unobserved"}, + "opening_clearance": {"outcome": "unsure", "unsure_cause": "margin"}, + "wall_equipment_above": {"outcome": "pass", "unsure_cause": None}, + "route_length": {"outcome": "pass", "unsure_cause": None}, + }, + "missing_evidence": [ + {"kind": "band", "band": "ground", "span_ft": [-0.614621, 22.321313], "out_ft": 12.009634}, + {"kind": "past_end", "side": "left", "span_ft": [-9.84252, -9.84252]}, + ], + "ends": {"left": "unexplored", "right": "limit"}, + } + + +def _expected_b() -> dict: + return { + "decision": "manual_review", + "spot": { + "wall_id": "side", + "segment": 0, + "span_ft": [-13.457349, -10.874016], + "outcome": "unsure", + }, + "checks": { + "ground_surface": {"outcome": "unsure", "unsure_cause": "unobserved"}, + "gas_clearance": {"outcome": "unsure", "unsure_cause": "unobserved"}, + "opening_clearance": {"outcome": "unsure", "unsure_cause": "margin"}, + "meter_working_space": {"outcome": "pass", "unsure_cause": None}, + "wall_equipment_above": {"outcome": "unsure", "unsure_cause": "unobserved"}, + "route_length": {"outcome": "pass", "unsure_cause": None}, + }, + "missing_evidence": [ + {"kind": "band", "band": "ground", "span_ft": [-23.666497, -0.664867], "out_ft": 12.042482}, + {"kind": "past_end", "side": "right", "span_ft": [9.84252, 9.84252]}, + ], + "ends": {"left": "limit", "right": "unexplored"}, + } + + +def _write_supported_manifest(tmp_path: Path) -> Path: + """A temp manifest: two hand-authored positive cases (source, wire scene and + expectations written and sha-pinned like the committed fixtures) plus one + committed negative case referenced by absolute path. This manifest is the + one command's success path: both positives must solve and publish.""" + fixtures = tmp_path / "fixtures" + fixtures.mkdir(parents=True) + cases = [] + for case_id, source, wire, expected in ( + (runner.TAP_CASE_ID, _supported_source_a(), _wire_a(), _expected_a()), + ("hazard-left-wall-gap", _supported_source_b(), _wire_b(), _expected_b()), + ): + source_path = fixtures / f"{case_id}.source.json" + source_path.write_text(json.dumps(source, indent=1) + "\n") + wire_path = fixtures / f"{case_id}.wire.json" + wire_path.write_text(json.dumps(wire, indent=1, sort_keys=True) + "\n") + cases.append( + { + "id": case_id, + "kind": "positive", + "source": {"path": str(source_path), "sha256": _sha256_file(source_path)}, + "wire": {"path": str(wire_path), "sha256": _sha256_file(wire_path)}, + "expectations": expected, + "note": "Hand-authored pair: the same geometry in metres and in feet.", + } + ) + cases.append( + { + "id": "multiwall-source-unsupported", + "kind": "negative_source", + "source": { + "path": str(BACKEND / "server/placement/tests/data/cases/multiwall-source-unsupported.source.json"), + "sha256": _sha256_file( + BACKEND / "server/placement/tests/data/cases/multiwall-source-unsupported.source.json" + ), + }, + "wire": None, + "expected_error": {"code": "placement_scene_unsupported", "pointer": "/walls"}, + } + ) + manifest_path = tmp_path / "manifest.json" + manifest_path.write_text( + json.dumps( + { + "schema_version": "placement-case-manifest/v0", + "description": ["Temporary supported-source manifest for the runner acceptance tests."], + "cases": cases, + }, + indent=1, + ) + + "\n" + ) + return manifest_path + + +def _write_twin_manifest(tmp_path: Path) -> Path: + """Two positive cases sharing one source and one wire scene byte for byte: + different identity stamps only. Cross-capture must fail them on geometry.""" + fixtures = tmp_path / "fixtures" + fixtures.mkdir(parents=True) + source_path = fixtures / "twin.source.json" + source_path.write_text(json.dumps(_supported_source_a(), indent=1) + "\n") + wire_path = fixtures / "twin.wire.json" + wire_path.write_text(json.dumps(_wire_a(with_tap=False), indent=1, sort_keys=True) + "\n") + source_ref = {"path": str(source_path), "sha256": _sha256_file(source_path)} + wire_ref = {"path": str(wire_path), "sha256": _sha256_file(wire_path)} + cases = [ + { + "id": "twin-one", + "kind": "positive", + "source": source_ref, + "wire": wire_ref, + "expectations": {"decision": "manual_review"}, + }, + { + "id": "twin-two", + "kind": "positive", + "source": source_ref, + "wire": wire_ref, + "expectations": {"decision": "manual_review"}, + }, + ] + manifest_path = tmp_path / "manifest.json" + manifest_path.write_text( + json.dumps( + { + "schema_version": "placement-case-manifest/v0", + "description": ["Twin-identity manifest: one geometry, two identities."], + "cases": cases, + }, + indent=1, + ) + + "\n" + ) + return manifest_path + + +# --------------------------------------------------------------------------- +# The committed manifest: refusal state, pinned and loud +# --------------------------------------------------------------------------- + + def test_every_case_walks_the_real_front_door_with_case_true_boundaries(receipt): """Each case's journey: injected boundaries hold, refusal is typed and leaves no placement artifact, the device stays conservative, one terminal event.""" @@ -82,22 +575,31 @@ def test_every_case_walks_the_real_front_door_with_case_true_boundaries(receipt) assert case["hasPlacementArtifact"] is False, "a refused placement must not publish an artifact" assert case["status"] == "refused" assert case["refusal"]["code"] == "placement_scene_unsupported" - assert case["deviceOutcome"] == { - "captureStatus": "manual_review", - "servedResultHttpStatus": 200, - "servedResultKind": "manual_review", - } + assert case["deviceOutcome"]["captureStatus"] == "manual_review" + assert case["deviceOutcome"]["servedResultHttpStatus"] == 200 + assert case["deviceOutcome"]["servedResultKind"] == "manual_review" + assert case["deviceOutcome"]["servedResultRunId"] == case["runId"], ( + "the served result must carry this journey's own run id" + ) assert case["terminalEventCount"] == 1 # The one nonempty objects_3d case: the observed tap hazard, merged exactly once. assert _case(receipt, runner.TAP_CASE_ID)["tapHazardMergeCount"] == 1 - # Two distinct captures in one app: no cross-capture scene reuse. - assert receipt["crossCapture"]["distinctScenes"] is True + # Both positives refused: no merged scenes, so the cross-capture comparison + # cannot run and must say so instead of passing vacuously. + cross = receipt["crossCapture"] + assert cross["status"] == "not_applicable" + assert cross["distinctScenes"] is None and cross["distinctNormalizedGeometry"] is None + assert cross["captureIds"] == [ + _case(receipt, "hazard-right-ground-gap")["captureId"], + _case(receipt, "hazard-left-ground-gap")["captureId"], + ] # A finalize replay is idempotent: same run id, still one terminal event. assert receipt["finalizeReplay"]["ok"] is True assert receipt["finalizeReplay"]["terminalEventCountAfter"] == 1 + assert receipt["finalizeReplay"]["applicable"] is True # The receipt never carries the token or any signed upload capability. assert runner.TOKEN not in json.dumps(receipt) @@ -121,7 +623,9 @@ def test_declared_refusals_match_their_declarations(receipt): def test_committed_positive_sources_do_not_convert(receipt): """The positive pairs' sources are refused by the bridge as authored — the - pinned blocker this suite reports instead of papering over.""" + pinned blocker this suite reports instead of papering over. A declared or + undeclared positive refusal is a blocker either way: only a positive case + that solves counts as success.""" for case_id in POSITIVE_IDS: refusal = _case(receipt, case_id)["refusal"] assert refusal["declared"] is False @@ -129,7 +633,9 @@ def test_committed_positive_sources_do_not_convert(receipt): "the meter anchor's stand-off is the first refusal the bridge raises" ) assert _case(receipt, case_id)["mismatches"], "an undeclared positive refusal must be a suite blocker" - assert receipt["blockers"], "the fixture state above must make the suite fail loudly" + blockers = "\n".join(receipt["blockers"]) + assert "full run requires both positive cases" in blockers + assert "hazard-right-ground-gap" in blockers and "hazard-left-ground-gap" in blockers assert receipt["ok"] is False @@ -202,3 +708,240 @@ def test_cli_runs_from_a_temporary_directory_and_exits_nonzero_while_blocked(tmp assert runner.TOKEN not in output.read_text() blockers = "\n".join(receipt["blockers"]) assert "standoff_m" in blockers and "/objects/1/evidence" in blockers + + +# --------------------------------------------------------------------------- +# The supported manifest: the success path and the faults that break it +# --------------------------------------------------------------------------- + + +def test_supported_sources_solve_end_to_end(tmp_path): + """Hand-authored positive sources convert, solve, and publish: the converted + bytes match the wire scene field by field, both solver responses bind their + input hash to their own exact request bytes, the semantics match the pinned + expectations, the device boundary stays conservative on the journey's own run + id, and the two positives carry distinct normalized geometry.""" + manifest = _write_supported_manifest(tmp_path) + receipt = runner.run_cases(manifest, tmp_path / "run") + assert receipt["ok"] is True, receipt["blockers"] + + positives = [case for case in receipt["cases"] if case["kind"] == "positive"] + assert len(positives) == 2 + for case in positives: + assert case["status"] == "solved" + assert case["sceneBoundaryMatchesCase"] is True + assert case["hasPlacementArtifact"] is True + assert case["stageInputShaMatchesCapturedBytes"] is True + assert case["directInputShaMatchesPostedBytes"] is True + assert case["directComparison"] == "matched" + assert case["wireSemanticComparison"] == "matched", case["wireSemanticMismatches"] + assert case["geometryComparison"] == "matched", case.get("geometryDiffs") + assert case["unitsSentinel"] == "matched" + assert case["deviceOutcome"]["servedResultHttpStatus"] == 200 + assert case["deviceOutcome"]["servedResultKind"] == "manual_review" + assert case["deviceOutcome"]["servedResultRunId"] == case["runId"] + assert case["terminalEventCount"] == 1 + assert case["mismatches"] == [] + + # The tap hazard converted exactly once, at its expected feet span. + assert _case(receipt, runner.TAP_CASE_ID)["tapHazardConvertedCount"] == 1 + + cross = receipt["crossCapture"] + assert cross["status"] == "evaluated" + assert cross["distinctScenes"] is True + assert cross["distinctNormalizedGeometry"] is True + assert len(set(cross["captureIds"])) == 2 + assert receipt["finalizeReplay"]["ok"] is True + + negative = _case(receipt, "multiwall-source-unsupported") + assert negative["status"] == "refused" and negative["mismatches"] == [] + + +def test_cli_on_the_supported_manifest_exits_zero(tmp_path): + """The one-command entry point on a manifest that can actually succeed.""" + manifest = _write_supported_manifest(tmp_path) + output = tmp_path / "receipt.json" + proc = subprocess.run( + [ + sys.executable, + str(BACKEND / "scripts" / "capture_placement_cases.py"), + "--manifest", + str(manifest), + "--output", + str(output), + "--workdir", + str(tmp_path / "run"), + ], + capture_output=True, + text=True, + cwd=tmp_path, + timeout=600, + ) + assert proc.returncode == 0, proc.stderr[-2000:] + receipt = json.loads(output.read_text()) + assert receipt["ok"] is True + assert receipt["crossCapture"]["distinctNormalizedGeometry"] is True + + +def test_cli_on_the_committed_manifest_exits_nonzero_in_a_subprocess(tmp_path): + """The failure path through the real argv/exit-code boundary too.""" + output = tmp_path / "receipt.json" + proc = subprocess.run( + [ + sys.executable, + str(BACKEND / "scripts" / "capture_placement_cases.py"), + "--output", + str(output), + "--workdir", + str(tmp_path / "run"), + ], + capture_output=True, + text=True, + cwd=tmp_path, + timeout=600, + ) + assert proc.returncode == 1 + assert "standoff_m" in proc.stderr + + +def test_case_selection_cannot_pass_silently(tmp_path): + """An empty selection and an unknown selection both refuse to run.""" + manifest = _write_supported_manifest(tmp_path) + with pytest.raises(SystemExit): + runner.run_cases(manifest, tmp_path / "run", case_ids=[]) + with pytest.raises(SystemExit): + runner.run_cases(manifest, tmp_path / "run", case_ids=["no-such-case"]) + + +def test_corrupting_the_converted_scene_fails_the_suite(tmp_path, monkeypatch): + """A converter that silently moves the gas hazard is caught by the + field-by-field comparison against the hand-authored wire scene — the staged + response and the direct POST both agree with each other and both are wrong.""" + + placement_adapter = importlib.import_module("server.placement.adapter") + real_convert = placement_adapter.convert_scene + + def corrupting_convert(document): + converted = real_convert(document) + for obj in converted.get("objects", []): + if obj.get("type") == "gas_meter": + obj["span_ft"] = [obj["span_ft"][0] + 0.5, obj["span_ft"][1] + 0.5] + return converted + + monkeypatch.setattr(placement_adapter, "convert_scene", corrupting_convert) + receipt = runner.run_cases(_write_supported_manifest(tmp_path), tmp_path / "run", case_ids=[runner.TAP_CASE_ID]) + case = receipt["cases"][0] + assert case["status"] == "solved" + assert case["geometryComparison"] == "mismatched" + assert any("wire scene" in mismatch for mismatch in case["mismatches"]) + assert receipt["ok"] is False + + +def test_forging_the_solver_input_hash_fails_the_suite(tmp_path, monkeypatch): + """A solver that stamps the same forged hash into both replies keeps them in + agreement with each other — and fails both hash bindings against the exact + request bytes.""" + + placement_adapter = importlib.import_module("server.placement.adapter") + modules = placement_adapter.load_api() + solver, api = modules["solver"], modules["api"] + real_solve = solver.solve + + def forging_solve(parsed, rules): + result = real_solve(parsed, rules) + result["stats"]["input_sha256"] = "f" * 64 + return result + + # The stage resolves solver.solve dynamically; the API route binds the name + # at import (`from solver import solve`). Patch both so both replies carry + # the same forged hash and still agree with each other. + monkeypatch.setattr(solver, "solve", forging_solve) + monkeypatch.setattr(api, "solve", forging_solve) + receipt = runner.run_cases(_write_supported_manifest(tmp_path), tmp_path / "run", case_ids=[runner.TAP_CASE_ID]) + case = receipt["cases"][0] + assert case["status"] == "solved" + assert case["stageInputShaMatchesCapturedBytes"] is False + assert case["directInputShaMatchesPostedBytes"] is False + assert case["directComparison"] == "matched", "the two forged responses must still agree with each other" + assert any("input_sha256" in mismatch for mismatch in case["mismatches"]) + assert receipt["ok"] is False + + +@pytest.mark.parametrize("mutation,field", [("kind", "servedResultKind"), ("runId", "servedResultRunId")]) +def test_a_served_result_that_lies_fails_the_suite(tmp_path, monkeypatch, mutation, field): + """The device boundary is checked against what is actually served: a result + endpoint that flips the outcome kind or the run id fails the suite.""" + + real_client = runner.TestClient + + class LyingClient(real_client): + def get(self, url, **kwargs): + response = super().get(url, **kwargs) + if str(url).endswith("/result") and response.status_code == 200: + body = response.json() + if mutation == "kind": + body["outcome"] = {"kind": "proposed", "reason_codes": ["forged"]} + else: + body["runId"] = "run_forged" + return httpx.Response(200, json=body) + return response + + monkeypatch.setattr(runner, "TestClient", LyingClient) + receipt = runner.run_cases(_write_supported_manifest(tmp_path), tmp_path / "run", case_ids=[runner.TAP_CASE_ID]) + case = receipt["cases"][0] + assert case["deviceOutcome"][field] == ("proposed" if mutation == "kind" else "run_forged") + assert case["mismatches"], "a lying served result must be a case mismatch" + assert receipt["ok"] is False + + +def test_a_failed_finalize_replay_blocks_the_suite(tmp_path, monkeypatch): + """A finalize replay that is no longer idempotent is an aggregate blocker.""" + + def failed_replay(harness, journey): + return { + "caseId": journey["caseId"], + "replayHttpStatus": 202, + "sameRunId": False, + "terminalEventCountAfter": 2, + "ok": False, + "applicable": True, + } + + monkeypatch.setattr(runner, "_replay_finalize", failed_replay) + receipt = runner.run_cases( + _write_supported_manifest(tmp_path), tmp_path / "run", case_ids=["multiwall-source-unsupported"] + ) + assert receipt["finalizeReplay"]["ok"] is False + assert any("finalizeReplay" in blocker for blocker in receipt["blockers"]) + assert receipt["ok"] is False + + +def test_one_geometry_under_two_identities_fails_cross_capture(tmp_path): + """The same source and wire under two case ids produce different identity + stamps (distinct raw merged-scene hashes) but identical geometry — the + cross-capture check must fail on normalized geometry, not pass on stamps.""" + receipt = runner.run_cases(_write_twin_manifest(tmp_path), tmp_path / "run") + cross = receipt["crossCapture"] + assert cross["status"] == "evaluated" + assert cross["distinctScenes"] is True, "identity stamps alone differ" + assert cross["distinctNormalizedGeometry"] is False, "the geometry is byte-identical" + assert any("crossCapture" in blocker for blocker in receipt["blockers"]) + assert receipt["ok"] is False + + +def test_a_declared_negative_that_solves_fails_the_suite(tmp_path, monkeypatch): + """If the converter someday accepts a declared negative's source, the case + must fail loudly — a negative case that solves is never a success.""" + + placement_adapter = importlib.import_module("server.placement.adapter") + real_convert = placement_adapter.convert_scene + valid_wire = real_convert(_supported_source_a()) + + monkeypatch.setattr(placement_adapter, "convert_scene", lambda document: json.loads(json.dumps(valid_wire))) + receipt = runner.run_cases( + _write_supported_manifest(tmp_path), tmp_path / "run", case_ids=["multiwall-source-unsupported"] + ) + case = receipt["cases"][0] + assert case["status"] == "solved" + assert any("produced a successful placement" in mismatch for mismatch in case["mismatches"]) + assert receipt["ok"] is False