fix(backend): review-page links, packet still paths, C2 join agreement, and hydrated reads/scale

backend/server/criteria/c2/veto.py

@@ -64,7 +64,8 @@ def join(runs: list[dict], spots: list[dict], sm, max_dist_m: float = 0.35, widt
             agreed_positions += 1
     return {
         "vetoes": vetoes,
-        "agreement_with_c1": round(agreed_positions / positions, 3) if positions else 1.0,
+        "positions_compared": positions,
+        "agreement_with_c1": round(agreed_positions / positions, 3) if positions else None,
     }
 
 

backend/server/criteria/run.py

@@ -317,7 +317,7 @@ 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, "vetoes": 0}
+    c2_report = {"status": "off", "agreement_with_c1": None, "positions_compared": 0, "vetoes": 0}
     c2_by_profile: dict[str, dict] = {}
     verdicts = {}
     t_all = time.time()
@@ -378,18 +378,28 @@ def evaluate_scene(
             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, "vetoes": 0}
+                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, "vetoes": 0}
+                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, "vetoes": 0}
+                c2_report = {"status": "skipped", "agreement_with_c1": None, "positions_compared": 0, "vetoes": 0}
             else:
                 stamped = c2_join(
                     runs,
@@ -400,6 +410,7 @@ def evaluate_scene(
                 c2_report = {
                     "status": "ok",
                     "agreement_with_c1": stamped["agreement_with_c1"],
+                    "positions_compared": stamped["positions_compared"],
                     "vetoes": stamped["vetoes"],
                 }
                 c2_by_profile[name] = {

backend/server/criteria/tests/test_criteria.py

@@ -1155,6 +1155,35 @@ def run(outcome, u, checks, failing):
     assert blocked["cand"].checks[0].outcome == "unsure"
 
 
+def test_join_without_matches_reports_no_comparison():
+    """No C2 spot in range: the cross-check is unavailable, not perfect."""
+    from server.criteria.c1.engine3d import Check
+    from server.criteria.c2.veto import join
+
+    run_row = {
+        "wall": "w1",
+        "u0": 0.0,
+        "u1": 0.0,
+        "outcome": "pass",
+        "failing": [],
+        "unsure": [],
+        "cand": type("C", (), {"checks": [Check("wall_backing", "pass")]})(),
+    }
+    spots = [
+        {
+            "outcome": "pass",
+            "wall_id": "w2",
+            "s_m": 0.0,
+            "rear_mid_xy": [0, 0],
+            "checks": [{"id": "wall_backing", "outcome": "pass"}],
+        }
+    ]
+    stamped = join([run_row], spots, {"w1": (0.0, 1)})
+    assert stamped["positions_compared"] == 0
+    assert stamped["agreement_with_c1"] is None
+    assert run_row["outcome"] == "pass"
+
+
 def test_placement_box_is_metres_at_the_short_route_spot():
     """The AR box is d/2 + 0.02 m off the face, at the shortest spot inside the run."""
     from shapely.geometry import Polygon

backend/server/reads/aggregate.py

@@ -302,6 +302,10 @@ def build_reads(
             path, sha = _still_bytes(still, files)
             if not path or not sha:
                 continue
+            # Packet paths are relative (`stills/s1.jpg`); promotion stored the bytes
+            # under `captures/{capture_id}/` (the server.api.keys.final_key layout),
+            # and uploads/ expires. Not imported: server.api pulls in the FastAPI app.
+            path = f"captures/{capture_id}/{path}"
             read_still(
                 StillRef(
                     capture_id=capture_id,

backend/server/reads/tests/test_reads.py

@@ -603,6 +603,19 @@ def _wrap(criteria: list[dict], views: list[dict]) -> dict:
     }
 
 
+def test_build_reads_resolves_packet_paths_against_promoted_files() -> None:
+    """Packet stills name packet-relative paths; the bytes live under captures/ after promotion."""
+    storage = MemoryStorage()
+    still = _still()
+    storage.put_bytes(f"captures/cap_test/{Path(still.path).name}", _jpeg(), "image/jpeg")
+    packet = {
+        "stills": [{"id": still.still_id, "purpose": still.purpose, "img": "s1.jpg", "sha256": still.sha256}],
+        "files": [{"path": "s1.jpg", "sha256": still.sha256}],
+    }
+    reads = build_reads("cap_test", storage, _cfg(external_vision=False), packet)
+    assert reads.timings_s, "the packet still was never read"
+
+
 def test_worker_disabled_opens_no_socket(monkeypatch: pytest.MonkeyPatch) -> None:
     def boom(*_args: object, **_kwargs: object) -> None:
         raise AssertionError("socket opened")
@@ -906,7 +919,8 @@ def test_empty_close_keeps_meter_identity() -> None:
 def test_packet_uses_img_and_file_digest() -> None:
     storage = MemoryStorage()
     path = "keyframes/k00003.jpg"
-    storage.put_bytes(path, _jpeg(), "image/jpeg")
+    # Promotion stores packet files under the capture's final prefix; uploads expire.
+    storage.put_bytes(f"captures/cap_test/{path}", _jpeg(), "image/jpeg")
     sha = "b" * 64
     packet = {
         "stills": [{"id": "s1", "purpose": "meter_close", "img": path}],

backend/server/result/compose.py

@@ -99,6 +99,8 @@ def compose_result(
     catalogue: Catalogue,
     config: ResultConfig = _DEFAULT_CONFIG,
     epochs: list[str] | None = None,
+    storage: object = None,
+    settings: object = None,
 ) -> ComposeResult:
     warnings: list[str] = []
     if catalogue.source == "fallback":
@@ -225,7 +227,20 @@ def compose_result(
     (out_dir / "result.json").write_text(_dumps(payload))
     (out_dir / "result.internal.json").write_text(_dumps(internal))
     render_review(
-        _review_context(capture, run, verdicts, profile, report, scene, response, internal, meter, reads_doc),
+        _review_context(
+            capture,
+            run,
+            verdicts,
+            profile,
+            report,
+            scene,
+            response,
+            internal,
+            meter,
+            reads_doc,
+            storage=storage,
+            settings=settings,
+        ),
         out_dir / "review.html",
         three_url=config.threejs_url,
     )
@@ -251,12 +266,22 @@ def emit_events(result: ComposeResult, state) -> list[int]:
     return seqs
 
 
+def _stage_input(ctx: StageContext, short: str, key: str) -> Path | None:
+    """The short input key wins; orchestrated runs hydrate `work_dir/<key>` instead."""
+
+    raw = ctx.inputs.get(short)
+    if raw:
+        return Path(raw)
+    path = Path(ctx.work_dir) / key
+    return path if path.is_file() else None
+
+
 def run(ctx: StageContext) -> StageResult:
     config = load_config(ctx.settings)
     scene_dir = Path(ctx.inputs.get("scene") or Path(ctx.work_dir) / "scene")
     verdict_dir = Path(ctx.inputs.get("verdicts") or Path(ctx.work_dir) / "verdicts")
-    reads = Path(ctx.inputs["reads"]) if ctx.inputs.get("reads") else None
-    scale = Path(ctx.inputs["scale"]) if ctx.inputs.get("scale") else None
+    reads = _stage_input(ctx, "reads", "reads/reads.json")
+    scale = _stage_input(ctx, "scale", "scale/scale.json")
     out_dir = Path(ctx.inputs.get("out") or Path(ctx.work_dir) / "result")
     capture = ctx.state.get_capture(ctx.capture_id)
     run_rec = ctx.state.get_run(ctx.run_id)
@@ -276,6 +301,8 @@ def run(ctx: StageContext) -> StageResult:
             catalogue=catalogue,
             config=config,
             epochs=epochs,
+            storage=ctx.storage,
+            settings=ctx.settings,
         )
     except ResultError:
         raise
@@ -783,7 +810,33 @@ def _judge_flags(report: dict | None, verdict_dir: Path) -> list:
     return list(flags)
 
 
-def _review_context(capture, run, verdicts, profile, report, scene, response, internal, meter: MeterBlock, reads):
+def _signed_url(storage: object, settings: object, key: str) -> str | None:
+    """A time-limited download URL, or None when signing is unavailable (offline tests, CLI)."""
+
+    sign = getattr(storage, "signed_get_url", None)
+    if not callable(sign):
+        return None
+    ttl = getattr(settings, "HSS_INGEST_RESULT_URL_TTL_S", 900)
+    try:
+        return str(sign(key, int(ttl) if ttl else 900))
+    except Exception:
+        return None
+
+
+def _review_context(
+    capture,
+    run,
+    verdicts,
+    profile,
+    report,
+    scene,
+    response,
+    internal,
+    meter: MeterBlock,
+    reads,
+    storage=None,
+    settings=None,
+):
     profiles = []
     for name in ("A", "B", "C", "S"):
         doc = verdicts.get(name)
@@ -801,10 +854,19 @@ def _review_context(capture, run, verdicts, profile, report, scene, response, in
     if reads and reads.get("confirm") and reads["confirm"].get("still_id"):
         meter_ids.add(reads["confirm"]["still_id"])
     for kf in scene.get("keyframes") or []:
+        # Keyframe image names are packet-relative; the bytes a reviewer can open live
+        # under the capture's promoted prefix. Sign what exists; keep the raw name
+        # otherwise so an offline render still names the file.
+        image = kf.get("image")
+        url = None
+        if image:
+            photo_key = f"captures/{capture.capture_id}/{image}"
+            if storage is not None and getattr(storage, "exists", lambda _k: False)(photo_key):
+                url = _signed_url(storage, settings, photo_key)
         photos.append(
             {
                 "id": kf["id"],
-                "url": kf.get("image") or kf["id"],
+                "url": url or image or kf["id"],
                 "purpose": "meter close-up" if kf["id"] in meter_ids else "keyframe",
             }
         )
@@ -827,11 +889,34 @@ def _review_context(capture, run, verdicts, profile, report, scene, response, in
         "views": [view.model_dump(mode="json") for view in response.outcome.viewsNeeded] if response.outcome else [],
         "meter": meter.model_dump(mode="json"),
         "profile": profile,
-        "links": [f"verdict.{name}.json" for name in ("A", "B", "C", "S") if name in verdicts]
-        + ["criteria_report.json", "result.json", "result.internal.json"],
+        "links": _review_links(capture, run, verdicts, storage, settings),
     }
 
 
+def _review_links(capture, run, verdicts, storage, settings) -> list[str]:
+    """Reviewer links. Bare names offline; signed download URLs where storage can sign.
+
+    The page is served at /v1/captures/{id}/review, and the API has no route for
+    relative artifact names, so a bare name is a dead link whenever storage is
+    available. result.json and result.internal.json are written by this stage but
+    published with the run, which must land before anyone can open this page.
+    """
+
+    names = [f"verdict.{name}.json" for name in ("A", "B", "C", "S") if name in verdicts]
+    names += ["criteria_report.json", "result.json", "result.internal.json"]
+    prefix = f"captures/{capture.capture_id}/runs/{run.run_id}"
+    keys = [f"{prefix}/verdicts/verdict.{name}.json" for name in ("A", "B", "C", "S") if name in verdicts]
+    keys += [
+        f"{prefix}/verdicts/criteria_report.json",
+        f"{prefix}/result/result.json",
+        f"{prefix}/result/result.internal.json",
+    ]
+    links = []
+    for name, key in zip(names, keys, strict=False):
+        links.append(_signed_url(storage, settings, key) or name)
+    return links
+
+
 def _wall_strips(scene: dict, verdict: dict) -> list[dict]:
     walls = []
     by_id = {wall["id"]: wall for wall in scene.get("walls") or []}