Backend: run real CPU criteria on published capture artifacts

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,55 +392,41 @@ 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,
                     [s for s in c2_spots if s.get("wall_dist_m", 99) <= 1.2],
                     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)

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

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",
+]

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-<variant>/``):
+
+    scene-<variant>/scene.json     base-scene/v0, asset sha256 stamps filled
+    scene-<variant>/laser.ply      the generator's binary laser point cloud
+    scene-<variant>/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-<variant>`` directories under ``dest``."""
+    return {name: write_variant(Path(dest) / f"scene-{name}", variant=name) for name in VARIANTS}