Backend: compose and serve a real result from hydrated artifacts

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 []}

backend/server/result/tests/test_compose.py

@@ -364,6 +364,107 @@ class _Settings:
     assert (tmp_path / "out" / "review.html").is_file()
 
 
+def test_run_resolves_reads_and_scale_from_work_dir(tmp_path, monkeypatch):
+    """Orchestrated runs hydrate reads/reads.json and scale/scale.json into work_dir without short keys."""
+
+    class _Settings:
+        RULES_PROFILE_DEFAULT = "C"
+        SHOW_REJECT_TO_HOMEOWNER = False
+        HSS_RESULT_MAX_VIEWS = 6
+        HSS_RESULT_SIGNED_URL_TTL_S = 900
+        HSS_RESULT_PREVIEW_MAX_POINTS = 200_000
+        HSS_RESULT_METER_AUTOFILL = "barcode"
+        HSS_RESULT_REVIEW_THREEJS_URL = "https://unpkg.com/three@0.160.0/build/three.module.js"
+        HSS_RESULT_PROMPTS_PATH = str(DATA / "prompts.en-US.json")
+        BASE_WEBHOOK_URL = None
+
+    for name in ("WEBHOOK_HMAC_SECRET", "BASE_WEBHOOK_URL", "SHOW_REJECT_TO_HOMEOWNER", "HSS_RESULT_PROMPTS_PATH"):
+        monkeypatch.delenv(name, raising=False)
+    capture, run_rec, epochs = _capture_run()
+    state = MemoryState()
+    state.put_capture(capture)
+    state.put_run(run_rec)
+    (tmp_path / "reads").mkdir(parents=True)
+    (tmp_path / "reads" / "reads.json").write_bytes((DATA / "reads.auto_fill.json").read_bytes())
+    (tmp_path / "scale").mkdir()
+    (tmp_path / "scale" / "scale.json").write_text(
+        json.dumps({"source": "lidar", "status": "verified", "rel_bound_95": 0.1})
+    )
+
+    ctx = StageContext(
+        capture_id=capture.capture_id,
+        run_id=run_rec.run_id,
+        stage="result",
+        attempt=1,
+        branch="lidar",
+        flow="guided",
+        tier="arkit",
+        work_dir=tmp_path,
+        inputs={
+            "scene": str(DATA / "synthetic"),
+            "verdicts": str(DATA / "synthetic" / "verdicts"),
+            "out": str(tmp_path / "out"),
+            "epochs": ",".join(epochs),
+        },
+        deadline_at="2026-09-27T00:00:00Z",
+        heartbeat=lambda _step: None,
+        vlm_budget_usd=0,
+        storage=None,
+        state=state,
+        settings=_Settings(),
+    )
+    run(ctx)
+    body = json.loads((tmp_path / "out" / "result.json").read_text())
+    internal = json.loads((tmp_path / "out" / "result.internal.json").read_text())
+    assert "reads_missing" not in internal["warnings"]
+    assert body["scale"]["source"] == "lidar"
+    assert body["meter"]["status"] == "auto_filled"
+
+
+class _SigningStorage:
+    """Storage double that signs any key, the way the deployed backends do."""
+
+    def __init__(self) -> None:
+        self.objects: dict[str, bytes] = {}
+
+    def exists(self, key: str) -> bool:
+        return key in self.objects
+
+    def get_bytes(self, key: str) -> bytes:
+        return self.objects[key]
+
+    def put_bytes(self, key: str, data: bytes, content_type: str) -> None:
+        del content_type
+        self.objects[key] = data
+
+    def signed_get_url(self, key: str, ttl_s: int) -> str:
+        return f"https://signed.test/{key}?ttl={ttl_s}"
+
+
+def test_review_page_signs_photos_and_raw_links(tmp_path):
+    """The review page is served under /v1/captures/{id}/review, so bare names are dead links."""
+    capture, run_rec, _epochs = _capture_run()
+    storage = _SigningStorage()
+    storage.put_bytes(f"captures/{capture.capture_id}/k01.jpg", b"jpeg-bytes", "image/jpeg")
+    compose_result(
+        capture,
+        run_rec,
+        DATA / "photo",
+        DATA / "photo" / "verdicts",
+        tmp_path,
+        reads=DATA / "reads.auto_fill.json",
+        scale=None,
+        catalogue=_catalogue(),
+        config=ResultConfig(catalogue_path=DATA / "prompts.en-US.json"),
+        storage=storage,
+    )
+    html = (tmp_path / "review.html").read_text()
+    assert f"https://signed.test/captures/{capture.capture_id}/k01.jpg" in html
+    assert f"https://signed.test/captures/{capture.capture_id}/runs/{run_rec.run_id}/verdicts/verdict.C.json" in html
+    assert f"https://signed.test/captures/{capture.capture_id}/runs/{run_rec.run_id}/result/result.json" in html
+    assert 'href="verdict.C.json"' not in html
+
+
 def test_deliver_signs_current_time_and_warns_on_failure(monkeypatch):
     stamped: dict[str, int] = {}