diff --git a/backend/server/result/compose.py b/backend/server/result/compose.py index 623354f0e..268830662 100644 --- a/backend/server/result/compose.py +++ b/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/` 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 []} diff --git a/backend/server/result/tests/test_compose.py b/backend/server/result/tests/test_compose.py index e8ebd10ca..28460fe94 100644 --- a/backend/server/result/tests/test_compose.py +++ b/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] = {} diff --git a/backend/tests/test_result_published_inputs.py b/backend/tests/test_result_published_inputs.py new file mode 100644 index 000000000..fbb534fa8 --- /dev/null +++ b/backend/tests/test_result_published_inputs.py @@ -0,0 +1,499 @@ +"""Storage-to-device proof: the real composer publishes the bytes the route serves. + +The device result boundary was only ever proven with seeded ``result/result.json`` +(#256) or a stub composer (#274). Nothing showed a real composed result traveling +the full path this packet owns: upstream stage artifacts in storage → production +hydration into a stage work dir → ``server.result.compose.run`` through +``runner.run_stage`` with stubs disabled → normal coordinator completion → +``GET /v1/captures/{id}/result`` as a device and as a reviewer. + +The upstream scene/verdicts/reads/scale artifacts are **fixture-authored** at +this packet's boundary: they are committed synthetic fixtures published at the +exact storage keys the upstream stages write, registered the way a stage +completion registers them. The final criteria-integration packet replaces this +boundary with real criteria output. ``placement/placement.json`` is published as +opaque bytes and is never read as a criteria verdict — it does not govern +eligibility here. + +Nothing seeds ``result/result.json``, ``result/result.internal.json`` or +``result/review.html``; the route only ever serves bytes the real composer +published. A private canary is written into the seeded verdict blob so the +device-output checks prove internal verdict data never leaks. +""" + +from __future__ import annotations + +import json +from dataclasses import replace +from pathlib import Path + +import pytest +from fastapi.testclient import TestClient + +from server.api.app import create_app +from server.api.deps import drain_finalize +from server.api.replay import run_replay +from server.common.config import Settings +from server.contracts.result import ResultResponse +from server.orchestrator.coordinator import RunCoordinator +from server.orchestrator.runner import run_stage +from server.orchestrator.services import bind, current +from server.orchestrator.stateio import put_json +from server.result.compose import ResultError +from server.result.tests.support import DATA, eligible_verdict, reject_verdict, views_verdict + +ROOT = Path(__file__).resolve().parents[1] +PACKET = ROOT / "data" / "fixtures" / "packets" / "mini-arkit" / "packet.json" + +TOKEN = "d" * 40 +REVIEWER_TOKEN = "f" * 40 +CANARY = "private-canary-burst9-9f3a" + +# Fixture-authored upstream inputs, pinned because the composed result must +# carry them. reads.auto_fill.json is the committed reads fixture; the scale +# document mirrors its committed shape. Without the #255 hydration fallback +# both are dropped (reads_missing warning, scale "none"/"unverified"). +SCALE_FIXTURE = {"source": "lidar", "status": "verified", "rel_bound_95": 0.1} +REPORT_FIXTURE = {"judge_flags": [], "c2": {"vetoes": []}} +PLACEHOLDER_BYTES = json.dumps({"fixture-authored": "burst9-bs-result-published-inputs"}).encode() + +TERMINAL_EVENTS = frozenset({"verdict_ready", "retake_request", "failed"}) + +# Independently declared expectations per authored verdict fixture. +EXPECTED = { + "eligible": {"capture_status": "complete", "event": "verdict_ready", "views": []}, + "views": { + "capture_status": "needs_views", + "event": "retake_request", + "views": ["vn-wall", "vn-ground"], + }, + "reject": {"capture_status": "manual_review", "event": "verdict_ready", "views": []}, +} + + +def _with_canary(builder) -> dict: + doc = builder() + doc["evaluator"]["path"] = f"server/result/tests#{CANARY}" + return doc + + +def _verdict_bytes(doc: dict) -> bytes: + return (json.dumps(doc, indent=1) + "\n").encode() + + +def _settings(tmp_path: Path) -> Settings: + return Settings( + STORAGE_BACKEND="volume", + STATE_BACKEND="memory", + VOLUME_ROOT=str(tmp_path / "vol"), + CAPTURE_API_PUBLIC_URL="http://testserver", + CAPTURE_API_TOKENS=f"device:{TOKEN},reviewer:{REVIEWER_TOKEN}", + UPLOAD_URL_HMAC_SECRET="h" * 32, + HSS_INGEST_HOOKS="auto", + HSS_AUTH_MODE="static", + GIT_SHA="0123456789abcdef", + SMP_SERVICE_NAME="capture-api", + SHOW_REJECT_TO_HOMEOWNER=False, + ) + + +class Journey: + """One capture through the real front door, composed by the real result stage.""" + + def __init__(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, scene_dir: Path, verdict: dict) -> None: + # Prewarm off, webhooks off, result knobs from the committed catalogue. + # Real stages for the finalize thread's coordinator tick: the tick must + # close the run through the early no-VLM gates, which it cannot do on + # stubbed stages. Pin the flag because sibling runners (the placement + # case suite) write HSS_ORCH_STAGE_STUBS=1 straight into os.environ and + # that write outlives their test. + monkeypatch.setenv("HSS_ORCH_STAGE_STUBS", "0") + monkeypatch.setenv("HSS_ORCH_PREWARM", "0") + monkeypatch.delenv("WEBHOOK_HMAC_SECRET", raising=False) + monkeypatch.setenv("BASE_WEBHOOK_URL", "") + monkeypatch.setenv("STORAGE_BACKEND", "volume") + monkeypatch.setenv("STATE_BACKEND", "memory") + monkeypatch.setenv("VOLUME_ROOT", str(tmp_path / "vol")) + monkeypatch.setenv("HSS_RESULT_PROMPTS_PATH", str(DATA / "prompts.en-US.json")) + monkeypatch.setenv("HSS_RESULT_METER_AUTOFILL", "barcode") + + self.app = create_app(settings=_settings(tmp_path)) + self.client = TestClient(self.app) + self.deps = self.app.state.deps + self.services = replace(current(), state=self.deps.state, storage=self.deps.storage) + self.coordinator = RunCoordinator(self.services) + + summary = run_replay( + packet_path=PACKET, + api="http://testserver/v1", + token=TOKEN, + client=self.client, + until="processing", + ) + assert summary["exit"] == 0, summary + self.capture_id = summary["captureId"] + # Join the finalize thread: start_run's writes are deterministic here, + # same as the API client fixture does for its tests. + drain_finalize(self.deps) + self.run_id = self.deps.state.get_capture(self.capture_id).run_id + run_row = self.deps.state.get_run(self.run_id) + assert run_row is not None, run_row + # The finalize thread also drives the real coordinator tick loop. With + # no VLM model configured the early gates close the run (objects/reads + # degrade, capture manual_review, failure_code recorded). The fixture + # boundary below re-registers every upstream stage as done, so reset + # the run to a clean in-flight state before seeding — the composer + # would otherwise refuse eligible for the stale failure code. + self.run = run_row.model_copy(update={"status": "running", "degraded": [], "failure_code": None}) + self.deps.state.put_run(self.run) + + self.seed_upstream(scene_dir, verdict) + self.result_key: str | None = None + + # ------------------------------------------------------------------ seeds + + def seed_stage(self, stage: str, artifacts: dict[str, tuple[bytes, str]]) -> None: + """Publish fixture bytes at the run's artifact keys and register completion. + + This mirrors what a real upstream stage completion writes: bytes in + storage, ``stage_artifacts::`` in the state store, a done + stage row, and the keys merged into the run record's artifact map. + """ + + prefix = f"captures/{self.capture_id}/runs/{self.run_id}" + storage_keys = {} + for key, (data, ctype) in artifacts.items(): + storage_key = f"{prefix}/{key}" + self.deps.storage.put_bytes(storage_key, data, ctype) + storage_keys[key] = storage_key + put_json(self.deps.state, f"stage_artifacts:{self.run_id}:{stage}", storage_keys) + rec = self.deps.state.get_stage(self.run_id, stage) + assert rec is not None + now = self.services.clock.iso() + self.deps.state.upsert_stage( + rec.model_copy( + update={ + "status": "done", + "heartbeat": now, + "finished_at": now, + "artifact_uri": next(iter(storage_keys.values()), None), + "metrics": {"duration_s": 0.5, "failures": 0.0, "respawns": 0.0}, + } + ) + ) + merged = dict(self.run.artifacts) + merged.update(storage_keys) + self.run = self.run.model_copy(update={"artifacts": merged}) + self.deps.state.put_run(self.run) + + def seed_upstream(self, scene_dir: Path, verdict: dict) -> None: + octet = "application/octet-stream" + blob = (PLACEHOLDER_BYTES, "application/json") + self.seed_stage("validate", {"packet/validation.json": blob, "packet/normalized.json": blob}) + self.seed_stage("poses", {"poses/cameras.json": blob, "poses/registration_report.json": blob}) + self.seed_stage("objects", {"objects/objects.json": blob}) + self.seed_stage( + "reads", + {"reads/reads.json": ((DATA / "reads.auto_fill.json").read_bytes(), "application/json")}, + ) + self.seed_stage("scale", {"scale/scale.json": (json.dumps(SCALE_FIXTURE).encode(), "application/json")}) + self.seed_stage("dense", {"dense/dense_report.json": blob}) + self.seed_stage("coverage", {"coverage/coverage_report.json": blob}) + self.seed_stage( + "scene", + { + "scene/scene.json": ((scene_dir / "scene.json").read_bytes(), "application/json"), + "scene/points.ply": ((scene_dir / "points.ply").read_bytes(), octet), + }, + ) + self.seed_stage("objects_3d", {"objects/objects_3d.json": blob}) + # Opaque placement bytes: never a criteria verdict, never eligibility input. + self.seed_stage("placement", {"placement/placement.json": blob}) + self.seed_stage( + "criteria", + { + "verdicts/verdict.C.json": (_verdict_bytes(verdict), "application/json"), + "verdicts/criteria_report.json": (json.dumps(REPORT_FIXTURE).encode(), "application/json"), + }, + ) + + def overwrite_verdict(self, verdict: dict) -> None: + """Replace the published verdict bytes in place; registration is unchanged.""" + + key = self.run.artifacts["verdicts/verdict.C.json"] + self.deps.storage.put_bytes(key, _verdict_bytes(verdict), "application/json") + + # ------------------------------------------------- real result-stage runs + + def _running_rec(self, attempt: int): + state = self.deps.state + rec = state.get_stage(self.run_id, "result") + assert rec is not None + rec = rec.model_copy( + update={ + "status": "running", + "attempt": attempt, + "worker_call_id": f"test-call-{attempt}", + "error": None, + "error_code": None, + "heartbeat": self.services.clock.iso(), + } + ) + state.upsert_stage(rec) + return rec + + def run_result_stage(self, attempt: int = 1) -> None: + """Hydrate through services.stage_context, run the real composer, complete.""" + + state = self.deps.state + rec = self._running_rec(attempt) + with bind(self.services): + context = self.services.stage_context(state.get_run(self.run_id), rec, f"test-call-{attempt}") + # No short input keys exist in orchestrated inputs; the hydrated + # work_dir copies are the only reads/scale the composer can see. + assert "reads" not in context.inputs and "scale" not in context.inputs + result = run_stage(context, stubs=False, call_id=f"test-call-{attempt}") + self.coordinator._mark_done(state.get_run(self.run_id), rec, result) + self.coordinator._maybe_finish(state.get_run(self.run_id)) + self.result_key = self.deps.state.get_run(self.run_id).artifacts["result/result.json"] + + def run_result_stage_failing(self) -> ResultError: + """The real composer on broken upstream bytes, closed through the coordinator.""" + + state = self.deps.state + rec = self._running_rec(1) + with bind(self.services): + context = self.services.stage_context(state.get_run(self.run_id), rec, "test-call") + try: + run_stage(context, stubs=False, call_id="test-call-1") + except ResultError as exc: + self.coordinator._mark_error(state.get_run(self.run_id), rec, exc) + return exc + raise AssertionError("the result stage unexpectedly succeeded on broken inputs") + + # ----------------------------------------------------------------- views + + def get_result(self, token: str = TOKEN): + return self.client.get( + f"/v1/captures/{self.capture_id}/result", + headers={"Authorization": f"Bearer {token}"}, + ) + + def final_events(self) -> list: + # The composer's own terminal event, written exactly once per run. The + # fixture's finalize thread may legitimately leave a `failed` event + # from its no-VLM tick closure; that is run history, not a composed + # outcome, and emit_events dedupes only on its own final types. + events = self.deps.state.events_after(self.capture_id, 0, limit=500) + composed = ("verdict_ready", "retake_request") + return [event for event in events if event.type in composed and event.run_id == self.run_id] + + def internal_document(self) -> dict: + assert self.result_key is not None + internal_key = self.result_key.replace("result/result.json", "result/result.internal.json") + return json.loads(self.deps.storage.get_bytes(internal_key)) + + def assert_served_from_published_bytes(self, response) -> dict: + """The served body equals the composer's stored bytes plus the composed keys.""" + + assert self.result_key is not None + stored = json.loads(self.deps.storage.get_bytes(self.result_key)) + body = response.json() + composed_keys = ("status", "runId", "verdictUrl", "previewUrl", "reviewUrl") + rest_served = {key: value for key, value in body.items() if key not in composed_keys} + rest_stored = {key: value for key, value in stored.items() if key not in composed_keys} + differing = { + key: {"stored": rest_stored.get(key), "served": rest_served.get(key)} + for key in sorted(set(rest_served) | set(rest_stored)) + if rest_served.get(key) != rest_stored.get(key) + } + assert not differing, f"the route served bytes the composer did not publish: {differing}" + return body + + def assert_common_shape(self, body: dict, expectation: dict) -> None: + # Identity: the served body names this run, and the record agrees. + assert body["runId"] == self.run_id + assert self.deps.state.get_capture(self.capture_id).status == expectation["capture_status"] + assert body["status"] == expectation["capture_status"] + # Hydrated reads and scale survive into the served body (the values the + # #255 fallback rescues; dropped without it). + assert body["scale"] == {"source": "lidar", "status": "verified", "relBound95": 0.1} + assert body["meter"]["status"] == "auto_filled" + assert body["memberActions"] == [] + # Views match the authored verdict fixture, not a stub's empty list. + assert [view["id"] for view in body["outcome"]["viewsNeeded"]] == expectation["views"] + assert [view["id"] for view in body["viewsNeeded"]] == expectation["views"] + # The response validates against the wire contract. + ResultResponse.model_validate(body) + # Exactly one terminal event for this run, of the declared type. + finals = self.final_events() + assert len(finals) == 1, [event.type for event in finals] + assert finals[0].type == expectation["event"] + # The internal document recorded no missing reads. + internal = self.internal_document() + assert "reads_missing" not in internal["warnings"] + + +@pytest.fixture +def make_journey(tmp_path, monkeypatch): + def _make(scene_dir: Path, verdict: dict) -> Journey: + return Journey(tmp_path, monkeypatch, scene_dir, verdict) + + return _make + + +def test_eligible_result_served_from_published_bytes(make_journey) -> None: + journey = make_journey(DATA / "synthetic", _with_canary(eligible_verdict)) + journey.run_result_stage() + + response = journey.get_result() + assert response.status_code == 200 + body = journey.assert_served_from_published_bytes(response) + expectation = EXPECTED["eligible"] + journey.assert_common_shape(body, expectation) + # Outcome and placement come from the authored verdict, not a stub. + assert body["outcome"]["kind"] == "eligible" + assert body["site"]["decision"] == "pass" + assert body["outcome"]["recommendedPlacement"] is not None + # The private canary lives in the stored verdict blob and never in the body. + verdict_key = journey.run.artifacts["verdicts/verdict.C.json"] + assert CANARY in journey.deps.storage.get_bytes(verdict_key).decode() + assert CANARY not in response.text + + +def test_reviewer_gets_signed_verdict_and_device_does_not(make_journey) -> None: + journey = make_journey(DATA / "synthetic", _with_canary(eligible_verdict)) + journey.run_result_stage() + + device = journey.get_result() + assert device.status_code == 200 + device_body = device.json() + assert device_body["verdictUrl"] is None + assert CANARY not in device.text + + reviewer = journey.get_result(REVIEWER_TOKEN) + assert reviewer.status_code == 200 + reviewer_body = reviewer.json() + assert reviewer_body["verdictUrl"] is not None + assert reviewer_body["verdictUrl"].startswith("http://testserver/v1/uploads/") + assert CANARY not in reviewer.text + + # Redaction is the only device/reviewer difference in composed fields. + for key in ("status", "runId", "outcome", "site", "scale", "meter", "viewsNeeded"): + assert device_body[key] == reviewer_body[key], key + + +def test_needs_views_result_lists_fixture_views(make_journey) -> None: + journey = make_journey(DATA / "photo", _with_canary(views_verdict)) + journey.run_result_stage() + + response = journey.get_result() + assert response.status_code == 200 + body = journey.assert_served_from_published_bytes(response) + journey.assert_common_shape(body, EXPECTED["views"]) + assert body["outcome"]["kind"] == "needs_more_photos" + assert body["site"]["decision"] == "manual_review" + + +def test_hidden_reject_masks_device_and_keeps_internal_decision(make_journey) -> None: + journey = make_journey(DATA / "photo", _with_canary(reject_verdict)) + journey.run_result_stage() + + response = journey.get_result() + assert response.status_code == 200 + body = journey.assert_served_from_published_bytes(response) + expectation = EXPECTED["reject"] + journey.assert_common_shape(body, expectation) + # Hidden-rejection policy: the device sees manual_review, no reasons, no + # failed-criteria detail, and no verdict link. + assert body["outcome"]["kind"] == "manual_review" + assert body["site"]["decision"] == "manual_review" + assert body["outcome"]["reasons"] == [] + assert "closer to the driveway" not in response.text + assert body["verdictUrl"] is None + # The internal document keeps the true decision. + internal = journey.internal_document() + assert internal["decision"] == "reject" + assert internal["outcome_kind_internal"] == "not_eligible" + assert CANARY not in response.text + + +def test_malformed_verdict_yields_no_successful_result(make_journey) -> None: + journey = make_journey(DATA / "synthetic", _with_canary(eligible_verdict)) + key = journey.run.artifacts["verdicts/verdict.C.json"] + journey.deps.storage.put_bytes(key, b'{"site": not json', "application/json") + + exc = journey.run_result_stage_failing() + assert exc.code == "verdict_invalid" + rec = journey.deps.state.get_stage(journey.run_id, "result") + assert rec is not None and rec.status == "failed" + assert rec.error_code == "verdict_invalid" + + response = journey.get_result() + assert response.status_code == 200 + body = response.json() + # A failed result stage still closes through the coordinator, which + # publishes a manual-review envelope: the device never sees a successful + # result for broken inputs. + assert body["runId"] == journey.run_id + assert body["status"] == "manual_review" + assert body["outcome"]["kind"] == "manual_review" + assert body["viewsNeeded"] == [] + assert body["verdictUrl"] is None + assert journey.deps.state.get_capture(journey.capture_id).status == "manual_review" + assert journey.final_events() == [] # no composed terminal event was ever written + assert journey.result_key is None + + +def test_missing_verdicts_yields_no_successful_result(make_journey) -> None: + journey = make_journey(DATA / "synthetic", _with_canary(eligible_verdict)) + # Re-register the criteria stage with no verdict artifact at all. + put_json(journey.deps.state, f"stage_artifacts:{journey.run_id}:criteria", {}) + merged = {key: value for key, value in journey.run.artifacts.items() if not key.startswith("verdicts/")} + journey.run = journey.run.model_copy(update={"artifacts": merged}) + journey.deps.state.put_run(journey.run) + + exc = journey.run_result_stage_failing() + assert exc.code == "verdict_missing" + rec = journey.deps.state.get_stage(journey.run_id, "result") + assert rec is not None and rec.status == "failed" and rec.error_code == "verdict_missing" + + response = journey.get_result() + assert response.status_code == 200 + body = response.json() + # Same closure contract: manual review, never a successful result. + assert body["runId"] == journey.run_id + assert body["status"] == "manual_review" + assert body["outcome"]["kind"] == "manual_review" + assert journey.final_events() == [] + assert journey.result_key is None + + +def test_controlled_verdict_change_changes_the_served_response(make_journey) -> None: + journey = make_journey(DATA / "synthetic", _with_canary(eligible_verdict)) + journey.run_result_stage() + + first = journey.get_result() + assert first.status_code == 200 + first_body = first.json() + assert first_body["outcome"]["kind"] == "eligible" + assert first_body["viewsNeeded"] == [] + + # Controlled upstream change: swap the published verdict for the views + # fixture and re-run the real composer (a second attempt on the same run). + journey.overwrite_verdict(_with_canary(views_verdict)) + journey.run_result_stage(attempt=2) + + second = journey.get_result() + assert second.status_code == 200 + second_body = second.json() + assert second_body["outcome"]["kind"] == "needs_more_photos" + assert second_body["status"] == "needs_views" + assert [view["id"] for view in second_body["viewsNeeded"]] == EXPECTED["views"]["views"] + assert second_body["runId"] == journey.run_id + # Exactly one terminal event for the run despite the re-compose: the + # event log keeps the first composition's verdict_ready, while the served + # bytes reflect the newer published result. + finals = journey.final_events() + assert len(finals) == 1 and finals[0].type == "verdict_ready" + assert CANARY not in second.text