diff --git a/backend/server/criteria/run.py b/backend/server/criteria/run.py index 5ca864742..a4890293c 100644 --- a/backend/server/criteria/run.py +++ b/backend/server/criteria/run.py @@ -224,12 +224,30 @@ def downgrade_c1_passes_without_c2(runs: list[dict]) -> None: run["causes"] = causes -def c2_sweep_document(by_profile: dict[str, dict], default_profile: str) -> dict: - """Top-level counts are the default profile, not whichever profile ran last.""" +def c2_default_document(by_profile: dict[str, dict], default_profile: str) -> dict: + """The report's whole C2 document: the default profile's comparison, nobody else's. + + A default profile with no C2 entry keeps the named ``off`` status with no + fabricated agreement; other profiles' evidence stays in ``by_profile``. + """ chosen = by_profile.get(default_profile) - if chosen is None and by_profile: - chosen = next(iter(by_profile.values())) - chosen = chosen or {"status": "off", "n_pass": 0, "n_spots": 0} + if chosen is None: + return {"status": "off", "agreement_with_c1": None, "positions_compared": 0, "vetoes": 0} + return { + "status": chosen["status"], + "agreement_with_c1": chosen["agreement_with_c1"], + "positions_compared": chosen["positions_compared"], + "vetoes": chosen["vetoes"], + } + + +def c2_sweep_document(by_profile: dict[str, dict], default_profile: str) -> dict: + """Top-level counts are the default profile, not whichever profile ran last. + + A default profile with no C2 entry reports the off counts; another profile's + numbers are never substituted (its evidence stays in ``by_profile``). + """ + chosen = by_profile.get(default_profile) or {"status": "off", "n_pass": 0, "n_spots": 0} return { "status": chosen["status"], "n_pass": chosen["n_pass"], @@ -317,7 +335,6 @@ def evaluate_scene( spend = HouseSpend(float(getattr(cfg, "c3_budget_usd", 0.60))) judge_by_profile: list[dict] = [] judge_hits: list[str | None] = [] - c2_report = {"status": "off", "agreement_with_c1": None, "positions_compared": 0, "vetoes": 0} c2_by_profile: dict[str, dict] = {} verdicts = {} t_all = time.time() @@ -375,31 +392,23 @@ def evaluate_scene( except Exception: log.exception("c2 failed") c2_spots, status = [], "skipped" + # Comparison evidence is recorded per profile. The top-level report is + # selected from the default profile after the loop, never overwritten here. + agreement: float | None = None + compared = 0 + vetoes = 0 if status == "no_coverage_grid": degraded.append("no_coverage_grid") downgrade_c1_passes_without_c2(runs) - c2_report = { - "status": "no_coverage_grid", - "agreement_with_c1": None, - "positions_compared": 0, - "vetoes": 0, - } elif status == "no_observed_ground": degraded.append("no_observed_ground") for run in runs: run["outcome"] = "unsure" - c2_report = { - "status": "no_observed_ground", - "agreement_with_c1": None, - "positions_compared": 0, - "vetoes": 0, - } elif status == "skipped": for run in runs: if run["outcome"] == "pass": run["outcome"] = "unsure" run["unsure"] = sorted(set(run["unsure"]) | {"c2_unavailable"}) - c2_report = {"status": "skipped", "agreement_with_c1": None, "positions_compared": 0, "vetoes": 0} else: stamped = c2_join( runs, @@ -407,23 +416,17 @@ def evaluate_scene( sm, width_ft=profile["battery"]["width_ft"], ) - c2_report = { - "status": "ok", - "agreement_with_c1": stamped["agreement_with_c1"], - "positions_compared": stamped["positions_compared"], - "vetoes": stamped["vetoes"], - } - c2_by_profile[name] = { - "status": "ok", - "n_pass": sum(1 for s in c2_spots if s.get("outcome") == "pass"), - "n_spots": len(c2_spots), - } - if name not in c2_by_profile: - c2_by_profile[name] = { - "status": c2_report["status"], - "n_pass": 0, - "n_spots": len(c2_spots), - } + agreement = stamped["agreement_with_c1"] + compared = stamped["positions_compared"] + vetoes = stamped["vetoes"] + c2_by_profile[name] = { + "status": status, + "agreement_with_c1": agreement, + "positions_compared": compared, + "vetoes": vetoes, + "n_pass": sum(1 for s in c2_spots if s.get("outcome") == "pass") if status == "ok" else 0, + "n_spots": len(c2_spots), + } _apply_scale(runs, scale, degraded) chosen = choose(runs) view_items = list((read_doc or {}).get("views_needed") or []) @@ -525,8 +528,10 @@ def evaluate_scene( verdicts[name] = verdict timings[name] = round(time.time() - t0, 3) sweep = c2_sweep_document(c2_by_profile, cfg.default_profile) - if cfg.default_profile in c2_by_profile: - c2_report = {**c2_report, "status": sweep["status"]} + # The whole top-level document is the default profile's comparison — status, + # agreement, positions and vetoes together. A default profile that never ran + # C2 keeps the named off document; another profile's numbers are never promoted. + c2_report = c2_default_document(c2_by_profile, cfg.default_profile) fact_rows, fact_record = evaluate_facts(scene, cfg, scene_dir=scene_dir, spend=spend) for name, verdict in verdicts.items(): attach_c3_facts(verdict, fact_rows) diff --git a/backend/server/criteria/tests/test_criteria.py b/backend/server/criteria/tests/test_criteria.py index d53a19967..d49a6ccf4 100644 --- a/backend/server/criteria/tests/test_criteria.py +++ b/backend/server/criteria/tests/test_criteria.py @@ -1930,6 +1930,70 @@ def test_c2_sweep_counts_the_default_profile(): assert doc["by_profile"]["S"]["n_pass"] == 0 +def test_c2_sweep_document_never_substitutes_another_profile(): + from server.criteria.run import c2_sweep_document + + doc = c2_sweep_document( + { + "S": { + "status": "ok", + "agreement_with_c1": 0.9, + "positions_compared": 30, + "vetoes": 2, + "n_pass": 9, + "n_spots": 30, + } + }, + "C", + ) + assert doc["status"] == "off" + assert doc["n_pass"] == 0 + assert doc["n_spots"] == 0 + assert doc["by_profile"]["S"]["n_pass"] == 9 # the evidence is kept, never promoted + + +def test_c2_default_document_selects_the_whole_default_comparison(): + from server.criteria.run import c2_default_document + + by_profile = { + "S": { + "status": "ok", + "agreement_with_c1": 0.9, + "positions_compared": 30, + "vetoes": 2, + "n_pass": 9, + "n_spots": 30, + }, + # #255: a zero-comparison sweep is ok with no agreement, never perfect agreement. + "C": { + "status": "ok", + "agreement_with_c1": None, + "positions_compared": 0, + "vetoes": 0, + "n_pass": 0, + "n_spots": 12, + }, + } + assert c2_default_document(by_profile, "C") == { + "status": "ok", + "agreement_with_c1": None, + "positions_compared": 0, + "vetoes": 0, + } + assert c2_default_document(by_profile, "S") == { + "status": "ok", + "agreement_with_c1": 0.9, + "positions_compared": 30, + "vetoes": 2, + } + assert c2_default_document({}, "C") == { + "status": "off", + "agreement_with_c1": None, + "positions_compared": 0, + "vetoes": 0, + } + + def test_placement_margin_over_uncertainty(): from server.criteria.c1.engine3d import Check from server.criteria.candidates import _margin_over_uncertainty diff --git a/backend/tests/fixtures/cpu_downstream/__init__.py b/backend/tests/fixtures/cpu_downstream/__init__.py new file mode 100644 index 000000000..0c73ea9d9 --- /dev/null +++ b/backend/tests/fixtures/cpu_downstream/__init__.py @@ -0,0 +1,29 @@ +"""Fixture helpers for the real CPU downstream stages (criteria, result). + +``scenes.supported_wall_variants(dest)`` writes two supported single-wall source +scenes — the synthetic flat-wall generator's geometry unchanged, plus one variant +carrying an observed gas-meter hazard at known dimensions. See ``scenes.py`` for +the callable and the relative artifact layout. +""" + +from .scenes import ( + HAZARD_BOTTOM_M, + HAZARD_S_M, + HAZARD_TOP_M, + VARIANTS, + hazard_object, + original_scene, + supported_wall_variants, + write_variant, +) + +__all__ = [ + "HAZARD_BOTTOM_M", + "HAZARD_S_M", + "HAZARD_TOP_M", + "VARIANTS", + "hazard_object", + "original_scene", + "supported_wall_variants", + "write_variant", +] diff --git a/backend/tests/fixtures/cpu_downstream/scenes.py b/backend/tests/fixtures/cpu_downstream/scenes.py new file mode 100644 index 000000000..aed856bbe --- /dev/null +++ b/backend/tests/fixtures/cpu_downstream/scenes.py @@ -0,0 +1,117 @@ +"""Two supported single-wall source scenes for the real CPU downstream stages. + +Callable (import by path — ``backend/tests`` has no package init): + + import importlib.util + from pathlib import Path + + spec = importlib.util.spec_from_file_location( + "cpu_downstream_scenes", + Path(__file__).parent / "fixtures" / "cpu_downstream" / "scenes.py", + ) + scenes = importlib.util.module_from_spec(spec) + spec.loader.exec_module(scenes) + dirs = scenes.supported_wall_variants(tmp_path) + +Relative artifact layout written per variant (``scene-/``): + + scene-/scene.json base-scene/v0, asset sha256 stamps filled + scene-/laser.ply the generator's binary laser point cloud + scene-/coverage.npz state/origin/voxel/views occupancy grid + +Both variants reuse the synthetic flat-wall generator's geometry unchanged +(``server.criteria.tests.data.synthetic_flatwall.make``): a 12 m house wall, +observed concrete ground, 24 synthetic cameras, and the generator's own PLY and +coverage grid. ``provenance.data`` stays ``"synthetic"`` — nothing here is real +site data. + +- ``clean`` — the generator's own positive-pass scene. +- ``hazard`` — the same geometry plus one observed gas-meter hazard at a known + span (s 3.0–3.4 m, 0.5–1.0 m high) derived from the wall's known dimensions. + +The placement bridge (``server.placement.scene_bridge``) refuses grid-bearing +coverage and keyframes by contract; these scenes carry the synthetic coverage +grid so the C2 sweep can run. ``original_scene`` hands back the generator's +verbatim scene dict for that boundary's diagnostic — never a reason to weaken +the bridge. +""" + +from __future__ import annotations + +import hashlib +import json +from pathlib import Path + +import numpy as np + +from server.criteria.tests.data.synthetic_flatwall import make as _flatwall + +VARIANTS = ("clean", "hazard") + +# Known dimensions of the observed hazard, in metres along w1 from the meter. +HAZARD_S_M = (3.0, 3.4) +HAZARD_BOTTOM_M = 0.5 +HAZARD_TOP_M = 1.0 + + +def hazard_object() -> dict: + """The observed gas-meter hazard the ``hazard`` variant adds to the wall.""" + s0, s1 = HAZARD_S_M + return { + "id": "gas-hazard", + "class": "gas_meter", + "label": "gas meter", + "geometry": { + "kind": "wall_rect", + "wall_id": "w1", + "s_m": [s0, s1], + "bottom_m": HAZARD_BOTTOM_M, + "top_m": HAZARD_TOP_M, + }, + "wall_id": "w1", + "confidence": 0.95, + "plus_minus_m": 0.05, + "source": {"kind": "annotation"}, + "evidence": "observed", + } + + +def original_scene() -> dict: + """The synthetic flat-wall generator's scene dict, completely unchanged.""" + return _flatwall.scene_dict() + + +def write_variant(dest: Path, *, variant: str = "clean") -> Path: + """Write one variant's scene.json, laser.ply and coverage.npz into ``dest``.""" + if variant not in VARIANTS: + raise ValueError(f"unknown variant {variant!r}; expected one of {VARIANTS}") + scene = _flatwall.scene_dict() + scene["scene_id"] = f"synthetic-flatwall-{variant}" + scene["provenance"]["site"] = scene["scene_id"] + if variant == "hazard": + scene["objects"].append(hazard_object()) + dest = Path(dest) + dest.mkdir(parents=True, exist_ok=True) + from server.criteria.harness import load + + h = load() + pts = _flatwall._points() + h.sceneio.write_ply(dest / "laser.ply", pts) + state, views = _flatwall._grid() + np.savez_compressed( + dest / "coverage.npz", + state=state, + origin=np.array([-7.0, -0.4, -0.4]), + voxel=np.float64(0.1), + views=views, + ) + scene["assets"][0]["sha256"] = hashlib.sha256((dest / "laser.ply").read_bytes()).hexdigest() + scene["assets"][0]["count"] = int(len(pts)) + scene["assets"][1]["sha256"] = hashlib.sha256((dest / "coverage.npz").read_bytes()).hexdigest() + (dest / "scene.json").write_text(json.dumps(scene, indent=1) + "\n") + return dest + + +def supported_wall_variants(dest: Path) -> dict[str, Path]: + """Write both variants into sibling ``scene-`` directories under ``dest``.""" + return {name: write_variant(Path(dest) / f"scene-{name}", variant=name) for name in VARIANTS} diff --git a/backend/tests/test_criteria_published_inputs.py b/backend/tests/test_criteria_published_inputs.py new file mode 100644 index 000000000..11ba55dbb --- /dev/null +++ b/backend/tests/test_criteria_published_inputs.py @@ -0,0 +1,571 @@ +"""The real CPU criteria stage over published capture artifacts. + +Five claims, each against real bytes through the real seam: + +- Upstream publication. Each scene's genuine nonempty laser PLY, coverage grid and + declared assets are published under the same ``captures//runs//`` + map an upstream scene/coverage stage supplies (``graph.STAGE_INPUTS["criteria"]`` + keys), and the stage work dir is hydrated by ``Services.stage_context`` — the test + never pre-seeds it. +- Real stage. ``runner.run_stage(ctx, stubs=False)`` executes the §3.9 criteria stage + entry (``server/criteria/__init__.py::run``), which evaluates the scene and publishes + fresh verdicts for every profile plus ``criteria_report.json`` — never + ``evaluate_scene`` alone, never a stub. +- Declared bytes. The published PLY and grid hash to the scene's declared asset + ``sha256`` stamps, and each verdict carries the hash of the scene it consumed. +- Honest C2. The report names the sweep's actual bounded status; a skipped or + bounded-out sweep downgrades C1 passes instead of standing in for agreement, and a + zero-comparison sweep reports no agreement (the #255 contract, regression pinned in + ``server/criteria/tests``). +- Explicit failure. A missing, empty or truncated declared PLY fails the stage with + the typed, non-retryable ``occluders_missing`` code — never a clean verdict. + +Synthetic geometry only: ``provenance.data`` is ``"synthetic"``. The placement bridge +(``scene_bridge.convert_scene``) refuses these grid-bearing, keyframe-carrying scenes +by contract; the diagnostic below records that refusal on the generator's unchanged +geometry instead of weakening the bridge. Two distinct capture/run identities run the +stage and never share files. +""" + +from __future__ import annotations + +import hashlib +import importlib.util +import json +from pathlib import Path + +import pytest + +from server.common.backends.state_memory import MemoryState +from server.common.backends.storage_local import LocalFsStorage +from server.common.config import Settings +from server.contracts.jobs import CaptureRecord, RunRecord, StageRecord +from server.contracts.verdict import Verdict +from server.criteria.occluders import ply_vertex_count +from server.eval.sceneio import verdict_contract_errors +from server.orchestrator.clock import Clock +from server.orchestrator.config import OrchConfig +from server.orchestrator.executor import InlineExecutor +from server.orchestrator.graph import STAGE_INPUTS +from server.orchestrator.runner import error_parts, is_retryable, run_stage +from server.orchestrator.services import Services, bind + +FIXTURES = Path(__file__).resolve().parent / "fixtures" / "cpu_downstream" + +VERDICT_KEYS = ( + "verdicts/verdict.A.json", + "verdicts/verdict.B.json", + "verdicts/verdict.C.json", + "verdicts/verdict.S.json", + "verdicts/criteria_report.json", + "verdicts/placement.C.json", + "verdicts/c2.sweep.json", + "verdicts/c3.facts.json", +) + + +def _load_helper(): + spec = importlib.util.spec_from_file_location("cpu_downstream_scenes", FIXTURES / "scenes.py") + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +@pytest.fixture +def stage_env(monkeypatch): + """Offline, conservative criteria settings. C2 runs for real with a workable budget.""" + monkeypatch.setenv("VISION_EXTERNAL_CALLS", "off") + monkeypatch.setenv("HSS_CRITERIA_C3_JUDGE", "off") + monkeypatch.setenv("HSS_CRITERIA_ABSENT_FACTS", "review") + monkeypatch.setenv("HSS_CRITERIA_C2_BUDGET_S", "120") + # Explicit static test auth and HMAC test secret. Nothing external is contacted. + monkeypatch.setenv("CAPTURE_API_TOKENS", "static-test-token") + monkeypatch.setenv("UPLOAD_URL_HMAC_SECRET", "static-test-hmac-secret") + monkeypatch.setenv("WEBHOOK_HMAC_SECRET", "static-test-webhook-hmac") + + +PACKET = { + "keyframes": [{"id": "k0"}], + "stills": [], + "taps": [{"id": "meter", "element": "meter"}], + "scaleReference": {"ext": {"doorOnMeterSide": False, "doorViews": []}}, +} + +READS = { + "format": "base-reads/v0", + "capture_id": "", + "facts": [ + { + "id": "panel_recall", + "scope": "site", + "outcome": "unsure", + "unsure_cause": "not_evaluated", + "reason": "static test fact", + "measured": None, + "threshold": None, + } + ], +} + + +def _services(tmp_path: Path) -> Services: + """Memory state, local volume storage, no prewarm, no stubs — the real path.""" + settings = Settings( + STATE_BACKEND="memory", + STORAGE_BACKEND="local", + VISION_EXTERNAL_CALLS="off", + HARD_STOP_MIN=20, + VOLUME_ROOT=str(tmp_path / "volume"), + SMP_ENVIRONMENT="dev", + ) + orch = OrchConfig(stage_stubs=False, stale_s=180, max_respawns=2, prewarm=False) + return Services( + settings=settings, + orch=orch, + state=MemoryState(), + storage=LocalFsStorage(tmp_path / "store"), + clock=Clock(), + executor=InlineExecutor(config=orch, stubs=False), + work_root=tmp_path / "work", + ) + + +def _bind_capture(services: Services, capture_id: str) -> bytes: + packet = json.dumps(PACKET).encode() + services.storage.put_bytes(f"captures/{capture_id}/packet.json", packet, "application/json") + services.state.put_capture( + CaptureRecord( + capture_id=capture_id, + packet_id=f"packet-{capture_id}", + status="processing", + tier="arkit", + flow="photo", + device_model="iPhone", + owner_principal="test", + created_at=services.clock.iso(), + finalize_by=services.clock.iso(), + packet_sha256=hashlib.sha256(packet).hexdigest(), + ) + ) + return packet + + +def _publish_inputs(services: Services, capture_id: str, run_id: str, files: dict[str, bytes]) -> dict[str, str]: + """The same storage/artifact map runner._publish uses for upstream stages.""" + prefix = f"captures/{capture_id}/runs/{run_id}" + artifacts: dict[str, str] = {} + for key, blob in files.items(): + storage_key = f"{prefix}/{key}" + ctype = "application/json" if key.endswith(".json") else "application/octet-stream" + services.storage.put_bytes(storage_key, blob, ctype) + artifacts[key] = storage_key + return artifacts + + +def _hydrated_context(services: Services, capture_id: str, run_id: str, packet: bytes, artifacts: dict[str, str]): + run = RunRecord( + run_id=run_id, + capture_id=capture_id, + packet_sha256=hashlib.sha256(packet).hexdigest(), + status="running", + branch="image", + flow="photo", + tier="arkit", + profile_default="C", + created_at=services.clock.iso(), + deadline_at=services.clock.iso(), + artifacts=artifacts, + ) + services.state.put_run(run) + rec = StageRecord(run_id=run_id, stage="criteria", status="running", attempt=1, heartbeat=services.clock.iso()) + services.state.upsert_stage(rec) + ctx = services.stage_context(run, rec, "call-burst9") + assert set(ctx.inputs) == {"packet_dir", "run_control"} # inputs ride the work dir, not ctx.inputs + return ctx + + +def _scene_files(scene_dir: Path, capture_id: str, laser: bytes | None) -> dict[str, bytes]: + scene_bytes = (scene_dir / "scene.json").read_bytes() + scene = json.loads(scene_bytes) + meter = next(o for o in scene["objects"] if o["class"] == "meter") + files = { + "scene/scene.json": scene_bytes, + "coverage/coverage.npz": (scene_dir / "coverage.npz").read_bytes(), + "coverage/bands.json": (json.dumps({"bands": scene["coverage"]["bands"]}) + "\n").encode(), + "coverage/coverage_report.json": b'{"views_needed": []}\n', + "objects/objects_3d.json": (json.dumps({"format": "hss-objects-3d/v0", "objects": [meter]}) + "\n").encode(), + "reads/reads.json": (json.dumps({**READS, "capture_id": capture_id}) + "\n").encode(), + "scale/scale.json": b'{"rel_bound_95": 0.01}\n', + } + if laser is not None: + files["scene/laser.ply"] = laser + return files + + +def _assert_c2_honest(report: dict, sweep: dict, verdict_c: dict) -> None: + """A skipped or bounded-out sweep is a downgrade, never a pass.""" + status = report["c2"]["status"] + assert status == sweep["status"] + if status == "ok": + compared = report["c2"]["positions_compared"] + assert isinstance(compared, int) and compared >= 0 + if compared == 0: + assert report["c2"]["agreement_with_c1"] is None # #255: no comparison is not perfect agreement + assert sweep["by_profile"]["C"]["status"] == "ok" + return + rows = verdict_c["criteria"] + assert not any(row["outcome"] == "pass" for row in rows), status + assert verdict_c["site"]["decision"] != "pass" + + +def test_published_inputs_feed_the_real_criteria_stage(stage_env, tmp_path): + scenes = _load_helper().supported_wall_variants(tmp_path) + services = _services(tmp_path) + runs: dict[str, tuple] = {} + with bind(services): + for variant, scene_dir in scenes.items(): + capture_id, run_id = f"cap-criteria-{variant}", f"run-criteria-{variant}" + packet = _bind_capture(services, capture_id) + files = _scene_files(scene_dir, capture_id, laser=(scene_dir / "laser.ply").read_bytes()) + for key in files: + assert STAGE_INPUTS["criteria"] and any( + (key.startswith(p.rstrip("/")) if p.endswith("/") else key == p) for p in STAGE_INPUTS["criteria"] + ), key + artifacts = _publish_inputs(services, capture_id, run_id, files) + ctx = _hydrated_context(services, capture_id, run_id, packet, artifacts) + work = Path(ctx.work_dir) + # Hydrated from storage by the service path, never pre-seeded by the test. + assert (work / "scene" / "scene.json").is_file() + assert (work / "scene" / "laser.ply").is_file() + assert (work / "coverage" / "coverage.npz").is_file() + assert not (work / "verdicts").exists() + result = run_stage(ctx, stubs=False) + assert result.metrics["verdicts"] == 4.0 + runs[variant] = (capture_id, run_id, ctx, result) + + storage = services.storage + docs: dict[str, dict] = {} + for variant, (capture_id, run_id, ctx, result) in runs.items(): + prefix = f"captures/{capture_id}/runs/{run_id}" + for key in VERDICT_KEYS: + assert key in result.artifacts, key + blob = storage.get_bytes(result.artifacts[key]) + assert blob, key + json.loads(blob) + for key, storage_key in result.artifacts.items(): + assert storage_key.startswith(prefix + "/"), (key, storage_key) + + report = json.loads(storage.get_bytes(result.artifacts["verdicts/criteria_report.json"])) + assert report["profiles"] == ["A", "B", "C", "S"] + assert report["provisional"] is False + + verdict_c = json.loads(storage.get_bytes(result.artifacts["verdicts/verdict.C.json"])) + assert Verdict.model_validate(verdict_c) + assert verdict_contract_errors(verdict_c) == [] + + # Published bytes hash to the scene's declared asset stamps. Inputs are read + # from the run's own storage prefix — result.artifacts carries stage outputs. + scene = json.loads(storage.get_bytes(f"{prefix}/scene/scene.json")) + for asset in scene["assets"]: + artifact_key = f"{prefix}/" + ("scene/" if asset["id"] == "laser" else "coverage/") + asset["path"] + blob = storage.get_bytes(artifact_key) + assert hashlib.sha256(blob).hexdigest() == asset["sha256"], asset["id"] + assert ply_vertex_count(Path(ctx.work_dir) / "scene" / "laser.ply") > 0 + + # The verdict carries exactly the scene bytes the stage consumed. + consumed = json.dumps(scene) + "\n" + assert verdict_c["scene"]["sha256"] == hashlib.sha256(consumed.encode()).hexdigest() + assert verdict_c["scene"]["data"] == "synthetic" + assert verdict_c["scene"]["scene_id"] == f"synthetic-flatwall-{variant}" + + sweep = json.loads(storage.get_bytes(result.artifacts["verdicts/c2.sweep.json"])) + _assert_c2_honest(report, sweep, verdict_c) + docs[variant] = { + "report": report, + "verdict": verdict_c, + "raw": json.loads(storage.get_bytes(result.artifacts["verdicts/c1.C.raw.json"])), + "sweep": sweep, + } + + # Scene-dependent outcomes. The hazard is real to the engine: candidate runs near + # its known span fail or go unsure on gas_clearance in one contiguous block, the + # evaluated check names the hazard as its subject, and the sweep compares more + # ground positions. The chosen candidate routes around the hazard, so the final + # criteria rows still pass gas clearance — the effect lives in the runs and sweep. + clean, hazard = docs["clean"], docs["hazard"] + assert json.dumps(clean["verdict"]["criteria"], sort_keys=True) != json.dumps( + hazard["verdict"]["criteria"], sort_keys=True + ) + clean_fail = [r for r in clean["raw"]["runs"] if "gas_clearance" in r.get("failing", [])] + hazard_fail = [r for r in hazard["raw"]["runs"] if "gas_clearance" in r.get("failing", [])] + hazard_unsure = [r for r in hazard["raw"]["runs"] if "gas_clearance" in r.get("unsure", [])] + assert not clean_fail + assert len(hazard_fail) >= 20 + fail_u0 = [r["u"][0] for r in hazard_fail] + assert max(fail_u0) - min(fail_u0) < 10.0 # one clearance block, not scattered noise + unsure_u0 = [r["u"][0] for r in hazard_unsure] + assert min(unsure_u0) < min(fail_u0) and max(fail_u0) < max(unsure_u0) # unsure ring wraps the block + best_gas = next(c for c in hazard["raw"]["best"]["checks"] if c["id"] == "gas_clearance") + assert best_gas["subject"] == "gas-hazard" + assert best_gas["outcome"] == "pass" and best_gas["measured"] >= best_gas["threshold"] + assert hazard["sweep"]["by_profile"]["C"]["n_spots"] >= 0 + assert hazard["report"]["c2"]["positions_compared"] > clean["report"]["c2"]["positions_compared"] + + # Two capture/run identities never share files. + clean_blob = storage.get_bytes(runs["clean"][3].artifacts["verdicts/verdict.C.json"]) + hazard_blob = storage.get_bytes(runs["hazard"][3].artifacts["verdicts/verdict.C.json"]) + assert clean_blob != hazard_blob + clean_prefix = "captures/cap-criteria-clean/runs/run-criteria-clean" + for name in storage.list_prefix(clean_prefix + "/"): + assert "cap-criteria-hazard" not in str(name) + clean_work, hazard_work = Path(runs["clean"][2].work_dir), Path(runs["hazard"][2].work_dir) + assert clean_work != hazard_work + assert "run-criteria-clean" in str(clean_work) and "run-criteria-hazard" in str(hazard_work) + + +@pytest.mark.parametrize("mode", ["missing", "empty", "truncated"]) +def test_broken_declared_ply_fails_explicitly(stage_env, tmp_path, mode): + scene_dir = _load_helper().supported_wall_variants(tmp_path)["clean"] + services = _services(tmp_path) + capture_id, run_id = "cap-broken-ply", "run-broken-ply" + with bind(services): + packet = _bind_capture(services, capture_id) + laser = (scene_dir / "laser.ply").read_bytes() + broken = { + "missing": None, + "empty": b"", + "truncated": laser[:64], # header cut before end_header: zero declared vertices + }[mode] + files = _scene_files(scene_dir, capture_id, laser=broken) + artifacts = _publish_inputs(services, capture_id, run_id, files) + ctx = _hydrated_context(services, capture_id, run_id, packet, artifacts) + with pytest.raises(Exception) as excinfo: + run_stage(ctx, stubs=False) + code, text = error_parts(excinfo.value) + assert code == "occluders_missing", (code, text) + assert not is_retryable(excinfo.value) + prefix = "captures/cap-broken-ply/runs/run-broken-ply/verdicts" + assert services.storage.list_prefix(prefix + "/") == [] + + +def _run_stage_once(monkeypatch, services, scene_dir, capture_id, run_id, env, files=None): + """One real publication → hydration → run_stage(stubs=False) pass over committed inputs.""" + packet = _bind_capture(services, capture_id) + if files is None: + files = _scene_files(scene_dir, capture_id, laser=(scene_dir / "laser.ply").read_bytes()) + artifacts = _publish_inputs(services, capture_id, run_id, files) + ctx = _hydrated_context(services, capture_id, run_id, packet, artifacts) + for name, value in env.items(): + monkeypatch.setenv(name, value) + result = run_stage(ctx, stubs=False) + storage = services.storage + report = json.loads(storage.get_bytes(result.artifacts["verdicts/criteria_report.json"])) + sweep = json.loads(storage.get_bytes(result.artifacts["verdicts/c2.sweep.json"])) + return report, sweep, result + + +OFF_DOCUMENT = {"status": "off", "agreement_with_c1": None, "positions_compared": 0, "vetoes": 0} + + +def test_c2_report_evidence_follows_the_default_profile_not_the_iteration_order(stage_env, tmp_path, monkeypatch): + """Regression for the two review findings (Greptile r4290120811, Codex r4290131173). + + With HSS_CRITERIA_C2_PROFILES=all the old code overwrote the top-level C2 report on + every profile iteration and corrected only the status afterwards, so the report's count + and agreement described the last profile (S) while the status described the default (C). + The real stage must keep every profile's comparison evidence and select the WHOLE + top-level document from the configured default — for C and for a non-C default, + independent of profile iteration order. + """ + scenes = _load_helper().supported_wall_variants(tmp_path) + services = _services(tmp_path) + docs: dict[tuple[str, str], tuple[dict, dict, object]] = {} + with bind(services): + for variant, scene_dir in scenes.items(): + for label, profiles, default in ( + ("s_last", "A,B,C,S", "C"), # S runs last: where the old overwrite bit + ("c_last", "S,B,A,C", "S"), # reordered iteration and a non-C default + ): + docs[(variant, label)] = _run_stage_once( + monkeypatch, + services, + scene_dir, + f"cap-c2sel-{variant}-{label}", + f"run-c2sel-{variant}-{label}", + env={ + "HSS_CRITERIA_C2_PROFILES": "all", + "HSS_CRITERIA_PROFILES": profiles, + "RULES_PROFILE_DEFAULT": default, + }, + ) + + for variant in scenes: + for label, default in (("s_last", "C"), ("c_last", "S")): + report, sweep, _result = docs[(variant, label)] + by_profile = sweep["by_profile"] + assert set(by_profile) == {"A", "B", "C", "S"}, by_profile # evidence kept, not overwritten + entry = by_profile[default] + assert report["c2"]["status"] == entry["status"] == sweep["status"] + assert report["c2"]["agreement_with_c1"] == entry["agreement_with_c1"] + assert report["c2"]["positions_compared"] == entry["positions_compared"] + assert report["c2"]["vetoes"] == entry["vetoes"] + assert sweep["n_pass"] == entry["n_pass"] + assert sweep["n_spots"] == entry["n_spots"] + for record in by_profile.values(): + if record["status"] == "ok" and record["positions_compared"] == 0: + assert record["agreement_with_c1"] is None # #255: no comparison is not perfect agreement + + # The reviewers' scenario, measured on the hazard variant: C and S sweep the same + # 16 positions, but S's comparison agrees 0.938 with one veto where C's agrees + # 1.0 with none. With profiles ordered A,B,C,S (S last) and default C, the old + # last-profile overwrite published S's agreement and vetoes under C's status; the + # report must carry the default's own comparison values. + s_last_report, s_last_sweep, _ = docs[("hazard", "s_last")] + assert s_last_sweep["by_profile"]["C"] == { + "status": "ok", + "agreement_with_c1": 1.0, + "positions_compared": 16, + "vetoes": 0, + "n_pass": 26, + "n_spots": 120, + } + assert s_last_sweep["by_profile"]["S"]["agreement_with_c1"] == 0.938 + assert s_last_sweep["by_profile"]["S"]["vetoes"] == 1 + assert s_last_report["c2"]["agreement_with_c1"] == 1.0 + assert s_last_report["c2"]["vetoes"] == 0 + # Order independence: the same scene+profile yields the same evidence in both runs. + kept = docs[(variant, "s_last")][1]["by_profile"] + assert docs[(variant, "c_last")][1]["by_profile"]["C"] == kept["C"] + + +def test_c2_report_names_off_absent_default_skipped_and_no_coverage_without_fabricated_agreement( + stage_env, tmp_path, monkeypatch +): + """Named statuses through the real stage: an off, an unavailable default, a skipped + sweep and a grid-less scene each keep agreement None and never borrow numbers.""" + scenes = _load_helper().supported_wall_variants(tmp_path) + clean = scenes["clean"] + services = _services(tmp_path) + + # (a) C2 disabled: no profile runs it; the report keeps the named off document. + report, sweep, _ = _run_stage_once( + monkeypatch, + services, + clean, + "cap-c2-off", + "run-c2-off", + env={ + "HSS_CRITERIA_C2": "off", + "HSS_CRITERIA_C2_PROFILES": "all", + "HSS_CRITERIA_PROFILES": "A,B,C,S", + "RULES_PROFILE_DEFAULT": "C", + "HSS_CRITERIA_C2_BUDGET_S": "120", + }, + ) + assert report["c2"] == OFF_DOCUMENT + assert sweep["status"] == "off" and sweep["n_pass"] == 0 and sweep["n_spots"] == 0 + assert sweep["by_profile"] == {} + + # (b) Other profiles really ran C2 but the default has no entry: no substitution. + report, sweep, _ = _run_stage_once( + monkeypatch, + services, + clean, + "cap-c2-absent", + "run-c2-absent", + env={ + "HSS_CRITERIA_C2": "on", + "HSS_CRITERIA_C2_PROFILES": "all", + "HSS_CRITERIA_PROFILES": "A,S", + "RULES_PROFILE_DEFAULT": "C", + "HSS_CRITERIA_C2_BUDGET_S": "120", + }, + ) + assert set(sweep["by_profile"]) == {"A", "S"} + assert all(record["n_spots"] > 0 for record in sweep["by_profile"].values()) # their evidence is real + assert report["c2"] == OFF_DOCUMENT + assert sweep["status"] == "off" and sweep["n_pass"] == 0 and sweep["n_spots"] == 0 + + # (c) A budget that cannot finish the first sweep names every profile skipped. + report, sweep, result = _run_stage_once( + monkeypatch, + services, + clean, + "cap-c2-budget", + "run-c2-budget", + env={ + "HSS_CRITERIA_C2": "on", + "HSS_CRITERIA_C2_PROFILES": "all", + "HSS_CRITERIA_PROFILES": "A,B,C,S", + "RULES_PROFILE_DEFAULT": "C", + "HSS_CRITERIA_C2_BUDGET_S": "0.5", + }, + ) + assert report["c2"] == {"status": "skipped", "agreement_with_c1": None, "positions_compared": 0, "vetoes": 0} + assert set(sweep["by_profile"]) == {"A", "B", "C", "S"} + assert all( + record["status"] == "skipped" and record["agreement_with_c1"] is None and record["positions_compared"] == 0 + for record in sweep["by_profile"].values() + ) + verdict_c = json.loads(services.storage.get_bytes(result.artifacts["verdicts/verdict.C.json"])) + assert verdict_c["site"]["decision"] != "pass" # a skipped sweep never yields a pass + + # (d) A scene that declares no coverage grid names no_coverage_grid and downgrades. + # The standard committed file set, with only the scene's inline coverage grid removed. + files = _scene_files(clean, "cap-c2-nogrid", laser=(clean / "laser.ply").read_bytes()) + scene = json.loads(files["scene/scene.json"]) + scene["coverage"]["grid"] = None + files["scene/scene.json"] = json.dumps(scene).encode() + b"\n" + report, sweep, result = _run_stage_once( + monkeypatch, + services, + clean, + "cap-c2-nogrid", + "run-c2-nogrid", + env={ + "HSS_CRITERIA_C2": "on", + "HSS_CRITERIA_C2_PROFILES": "all", + "HSS_CRITERIA_PROFILES": "A,B,C,S", + "RULES_PROFILE_DEFAULT": "C", + "HSS_CRITERIA_C2_BUDGET_S": "120", + }, + files=files, + ) + assert report["c2"] == { + "status": "no_coverage_grid", + "agreement_with_c1": None, + "positions_compared": 0, + "vetoes": 0, + } + assert set(sweep["by_profile"]) == {"A", "B", "C", "S"} + assert all( + record["status"] == "no_coverage_grid" and record["agreement_with_c1"] is None + for record in sweep["by_profile"].values() + ) + assert "no_coverage_grid" in report["degraded"] + verdict_c = json.loads(services.storage.get_bytes(result.artifacts["verdicts/verdict.C.json"])) + assert not any(row["outcome"] == "pass" for row in verdict_c["criteria"]) + assert verdict_c["site"]["decision"] != "pass" + + +def test_bridge_boundary_diagnostic_on_unchanged_generator_geometry(tmp_path): + from server.criteria.tests.data.synthetic_flatwall.make import scene_dict + from server.placement.scene_bridge import UNSUPPORTED, SceneBridgeError, convert_scene + + scenes = _load_helper() + original = scenes.original_scene() + assert original == scene_dict() # the generator's geometry, byte-for-byte unchanged + with pytest.raises(SceneBridgeError) as first: + convert_scene(original) + assert first.value.code == UNSUPPORTED + # The documented refusals, in the bridge's own order: keyframes, then the grid. + without_keyframes = dict(original, keyframes=[]) + with pytest.raises(SceneBridgeError) as second: + convert_scene(without_keyframes) + assert second.value.code == UNSUPPORTED + assert second.value.pointer == "/coverage/grid" + for scene_dir in scenes.supported_wall_variants(tmp_path).values(): + document = json.loads((scene_dir / "scene.json").read_text()) + with pytest.raises(SceneBridgeError) as variant_refusal: + convert_scene(document) + assert variant_refusal.value.code == UNSUPPORTED