diff --git a/backend/scripts/capture_flow_e2e.py b/backend/scripts/capture_flow_e2e.py index b8d4b381b..3240b1bdd 100644 --- a/backend/scripts/capture_flow_e2e.py +++ b/backend/scripts/capture_flow_e2e.py @@ -191,6 +191,10 @@ def record(stage: str, **fields) -> None: # Register, PUT to signed URLs, commit. files = _sealed_files(fixture_root, packet) + # The replay anchor: a saved receipt carries the exact bodies the device + # sent, so a later replay can prove the same hashes still commit. + receipt["idempotencyKeys"] = {"create": packet["packetId"], "finalize": packet["packetId"]} + receipt["files"] = [dict(item) for item in files] receipt["fileCounts"]["expected"] = len(files) registered = journey.client.post(f"/v1/captures/{capture_id}/files", json={"files": files}) record("register", httpStatus=registered.status_code, files=len(files)) @@ -243,6 +247,7 @@ def record(stage: str, **fields) -> None: # Finalize packet.json. raw = (fixture_root / "packet.json").read_bytes() + receipt["packetSha256"] = hashlib.sha256(raw).hexdigest() accepted = journey.client.post( f"/v1/captures/{capture_id}/finalize", content=raw, diff --git a/backend/scripts/replay_capture.py b/backend/scripts/replay_capture.py new file mode 100644 index 000000000..65fae0dc1 --- /dev/null +++ b/backend/scripts/replay_capture.py @@ -0,0 +1,550 @@ +#!/usr/bin/env python3 +"""Replay a saved capture-flow receipt against the real capture API, no cloud. + +One repeatable command for reviewers: + + cd backend && uv run --extra api python scripts/replay_capture.py \ + --receipt /receipt.json + +The input is the JSON receipt ``scripts/capture_flow_e2e.py`` writes: it names +the packet, the idempotency keys, the sealed per-file bodies (path, bytes, +sha256) and the packet digest. The replay re-sends that journey — create, +register, PUT, commit, finalize — to a fresh in-process instance of the real +app entry point ``create_app()`` (memory state, local volume storage, the +documented dev handoff ``HSS_INGEST_HOOKS=auto`` with orchestrator stage +stubs, no GCP, no Modal, no network, no credentials), then drains the finalize +work and reads the terminal capture status and result back through the device +routes. It asserts four things the raw HTTP log of a run cannot prove: + +- the saved idempotency keys are reused verbatim on create and finalize; +- a request that "times out" after the server processed it replays to the + same capture id (a 200 on the stored body, never a second 201 capture); +- the committed file hashes are the receipt's hashes, on the first attempt + and on a resumed commit after the files were already committed; +- the run lands on one terminal status, served once by ``GET /result``, with + exactly one terminal verdict event and one run id. + +A scripted, deterministic ``FailureInjector`` simulates lost responses: the +request really executes against the app, then the injector raises +``SimulatedTimeout`` so the caller never sees the response — the situation a +device hits when a cell handoff eats an HTTP reply. Each fault fires at most +once per scripted occurrence and every step retries at most ``MAX_RETRIES`` +times; a fault scripted beyond that budget fails the replay loudly instead of +retrying forever. + +It writes a JSON report (default ``/replay-receipt.json``) and exits +0 when every assertion held, 1 otherwise. It never changes route contracts and +never touches iOS. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import sys +import tempfile +from datetime import UTC, datetime +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from fastapi.testclient import TestClient # noqa: E402 + +from server.api.app import create_app # noqa: E402 +from server.api.deps import drain_finalize # noqa: E402 +from server.common.config import Settings # noqa: E402 + +ROOT = Path(__file__).resolve().parents[1] +DEFAULT_PACKET = ROOT / "data/fixtures/packets/mini-arkit" + +# Same static test token the API test suite uses (server/api/tests/conftest.py). +TOKEN = "d" * 40 + +# A lost response is retried once with the same idempotency key. A second +# failure is a finding, not something to hide behind an unbounded retry loop. +MAX_RETRIES = 1 + +# Bounded, deterministic readiness polling: drain_finalize joins the finalize +# thread, so a drained app is already terminal and one read normally settles +# it. The loop exists so a slow app fails loudly after N reads, never hangs. +MAX_POLLS = 5 + +TERMINAL_STATUSES = {"complete", "failed", "needs_views", "manual_review"} +TERMINAL_EVENTS = {"verdict_ready", "retake_request", "failed"} + + +class SimulatedTimeout(Exception): + """A response the server produced but the client never received.""" + + +class FailureInjector: + """Scripted lost-response faults: ``{"create": 1}`` loses one create reply. + + Deterministic by construction — no clocks, no randomness. ``fire`` is + called after a request has actually executed against the app, so the + server-side effect exists before the caller is told the response was lost, + which is exactly what makes the idempotent replay meaningful. + """ + + def __init__(self, script: dict[str, int] | None = None) -> None: + self.script = dict(script or {}) + self.fired: list[str] = [] + + def fire(self, step: str) -> None: + remaining = self.script.get(step, 0) + if remaining <= 0: + return + self.script[step] = remaining - 1 + self.fired.append(step) + raise SimulatedTimeout(step) + + +def receipt_problem(receipt: object) -> str | None: + """A named reason the saved receipt cannot be replayed, or None. + + The replay block the runner writes (files, idempotencyKeys, packetSha256) + is the replay's ground truth; a receipt without it was written by a + pre-replay runner and must be rejected, never replayed with guessed + hashes. + """ + if not isinstance(receipt, dict) or receipt.get("runner") != "capture_flow_e2e": + return "not_a_capture_flow_e2e_receipt" + missing = [key for key in ("files", "idempotencyKeys", "packetSha256") if not receipt.get(key)] + if missing: + return f"missing_replay_block:{','.join(missing)}" + return None + + +def check_fixture(packet_dir: Path, receipt: dict, errors: list[str]) -> dict | None: + """The replayed bytes must still hash to the saved receipt's hashes.""" + packet_json = packet_dir / "packet.json" + if not packet_json.is_file(): + errors.append(f"packet_json_missing:{packet_dir}") + return None + packet = json.loads(packet_json.read_text()) + saved_files = receipt.get("files") or [] + saved_hashes = {item["path"]: item["sha256"] for item in saved_files} + drift: list[str] = [] + for entry in packet["files"]: + path = entry["path"] + data = (packet_dir / path).read_bytes() + digest = hashlib.sha256(data).hexdigest() + if path in saved_hashes and saved_hashes[path] != digest: + drift.append(path) + if entry.get("sha256") and entry["sha256"] != digest: + drift.append(f"{path}:packet-manifest") + manifest_paths = {entry["path"] for entry in packet["files"]} + for path in saved_hashes: + if path not in manifest_paths: + drift.append(f"{path}:not-in-manifest") + saved_sha = receipt.get("packetSha256") + actual_sha = hashlib.sha256(packet_json.read_bytes()).hexdigest() + if saved_sha and saved_sha != actual_sha: + drift.append("packet.json") + if drift: + errors.append(f"fixture_drift:{','.join(drift)}") + return {"packet": packet, "packetSha256": actual_sha, "drift": drift} + + +class _ReplayApp: + """The real app under replay: dev handoff on, everything else local.""" + + def __init__(self, workdir: Path) -> None: + # The orchestrator hook binds services from the environment; it needs + # the same volume root the API storage uses or the bind no-ops. + self._env = {key: os.environ.get(key) for key in ("VOLUME_ROOT", "HSS_ORCH_STAGE_STUBS", "STATE_BACKEND")} + os.environ["VOLUME_ROOT"] = str(workdir / "vol") + os.environ["HSS_ORCH_STAGE_STUBS"] = "1" + os.environ.setdefault("STATE_BACKEND", "memory") + settings = Settings( + SMP_ENVIRONMENT="dev", + HSS_AUTH_MODE="static", + CAPTURE_API_TOKENS=f"device:{TOKEN}", + UPLOAD_URL_HMAC_SECRET="h" * 32, + STORAGE_BACKEND="volume", + STATE_BACKEND="memory", + VOLUME_ROOT=str(workdir / "vol"), + CAPTURE_API_PUBLIC_URL="http://testserver", + HSS_INGEST_HOOKS="auto", + VISION_EXTERNAL_CALLS="off", + GIT_SHA="0123456789abcdef", + SMP_SERVICE_NAME="capture-api", + ) + self.app = create_app(settings=settings) + self.client = TestClient(self.app) + self.client.headers["Authorization"] = f"Bearer {TOKEN}" + + def close(self) -> None: + for key, value in self._env.items(): + if value is None: + os.environ.pop(key, None) + else: + os.environ[key] = value + + def drain(self) -> None: + drain_finalize(self.app.state.deps) + + +def _send(step: str, request, injector: FailureInjector, report: dict, errors: list[str]) -> object: + """Run ``request``; replay the identical request after a lost response. + + The fault fires only after the request executed, so a retried create is a + genuine duplicate-delivery: the server must answer from its idempotency + store, not open a second capture. + """ + attempts: list[dict] = [] + report.setdefault("attempts", {})[step] = attempts + attempt = 0 + while True: + response = request() # the request really went out and was processed + try: + injector.fire(step) + except SimulatedTimeout: + attempts.append({"attempt": attempt, "outcome": "simulated_timeout", "httpStatus": response.status_code}) + attempt += 1 + if attempt > MAX_RETRIES: + errors.append(f"{step}_retry_budget_exhausted") + return response + continue # same bytes, same idempotency key + attempts.append({"attempt": attempt, "outcome": "delivered", "httpStatus": response.status_code}) + return response + + +def run_replay( + receipt_path: Path, + workdir: Path, + *, + packet_dir: Path | None = None, + faults: dict[str, int] | None = None, +) -> dict: + """Replay the saved receipt and return the report. Never raises: failures + land in ``report["errorCodes"]`` with ``report["ok"] is False`` — including + receipt problems and any unexpected exception.""" + started_at = datetime.now(UTC).isoformat() + report: dict = { + "replay": "replay_capture", + "startedAt": started_at, + "receiptPath": str(receipt_path), + "packetDir": None, + "workdir": str(workdir), + "faults": dict(faults or {}), + "idempotencyKeys": {}, + "captureId": None, + "runId": None, + "steps": [], + "fileHashesVerified": 0, + "finalCaptureStatus": None, + "resultOutcomeKind": None, + "terminalEvents": [], + "errorCodes": [], + "ok": False, + } + errors = report["errorCodes"] + injector = FailureInjector(faults) + + def record(step: str, **fields) -> None: + report["steps"].append({"step": step, **fields}) + + app: _ReplayApp | None = None + try: + receipt = json.loads(Path(receipt_path).read_text()) + problem = receipt_problem(receipt) + if problem: + errors.append(problem) + return _finish(report, errors, f"{receipt_path}: {problem} — written by a runner without the replay block") + saved_dir = receipt.get("packetDir") + packet_dir = packet_dir or (Path(saved_dir) if saved_dir and Path(saved_dir).is_dir() else DEFAULT_PACKET) + report["packetDir"] = str(packet_dir) + report["idempotencyKeys"] = receipt.get("idempotencyKeys") or {} + fixture = check_fixture(packet_dir, receipt, errors) + record( + "fixture", files=len((fixture or {}).get("packet", {}).get("files", [])), drift=(fixture or {}).get("drift") + ) + if fixture is None or errors: + return _finish(report, errors, "the saved receipt no longer matches the packet fixture") + packet = fixture["packet"] + packet_sha = fixture["packetSha256"] + saved_files = receipt["files"] + report["fileHashesVerified"] = len(saved_files) + keys = receipt.get("idempotencyKeys") or {} + create_key = keys.get("create") or packet["packetId"] + finalize_key = keys.get("finalize") or packet["packetId"] + report["idempotencyKeys"] = {"create": create_key, "finalize": finalize_key} + + workdir.mkdir(parents=True, exist_ok=True) + app = _ReplayApp(workdir) + + # Create the capture. A lost first response must replay to the same + # capture id from the idempotency store: a 200, never a second 201. + body = { + "packetId": packet["packetId"], + "formatVersion": packet["formatVersion"], + "tier": packet["session"]["tier"], + "flow": "guided", + "device": { + "model": packet["session"]["deviceModel"], + "systemVersion": packet["session"]["systemVersion"], + "appVersion": packet["session"]["appVersion"], + }, + } + + def _create(): + return app.client.post("/v1/captures", json=body, headers={"Idempotency-Key": create_key}) + + created = _send("create", _create, injector, report, errors) + created_body = created.json() if created.status_code in (200, 201) else {} + capture_id = created_body.get("captureId") + report["captureId"] = capture_id + first_status = report["attempts"]["create"][0]["httpStatus"] + record( + "create", + idempotencyKey=create_key, + firstHttpStatus=first_status, + httpStatus=created.status_code, + captureId=capture_id, + ) + if created.status_code not in (200, 201): + errors.append(f"create_http_{created.status_code}") + return _finish(report, errors, f"create failed: {created.text[:300]}") + if not capture_id: + return _finish(report, errors, "no captureId in create response") + + def _create_again(): + return app.client.post("/v1/captures", json=body, headers={"Idempotency-Key": create_key}) + + again = _create_again() + stable = again.status_code == 200 and again.json().get("captureId") == capture_id + record("create-stable", httpStatus=again.status_code, captureId=again.json().get("captureId"), stable=stable) + if not stable: + errors.append("capture_id_not_stable") + + # Register, PUT the receipt's bytes, commit. The PUT bodies are hashed + # again here: the replay uploads exactly what the receipt says it did. + registered = app.client.post(f"/v1/captures/{capture_id}/files", json={"files": saved_files}) + record("register", httpStatus=registered.status_code, files=len(saved_files)) + if registered.status_code != 200: + errors.append(f"register_http_{registered.status_code}") + return _finish(report, errors, f"register failed: {registered.text[:300]}") + uploaded: list[str] = [] + for saved, target in zip(saved_files, registered.json()["files"], strict=True): + data = (packet_dir / saved["path"]).read_bytes() + if hashlib.sha256(data).hexdigest() != saved["sha256"]: + errors.append(f"upload_hash_drift:{saved['path']}") + if target.get("state") == "committed": + continue # already on the server: resume needs no second PUT + put = TestClient(app.app).put( + target["upload"]["url"], + content=data, + headers=target["upload"].get("headers") or {}, + ) + if put.status_code == 200: + uploaded.append(target["path"]) + else: + errors.append(f"upload_http_{put.status_code}:{target['path']}") + record("upload", uploaded=len(uploaded), skippedCommitted=len(saved_files) - len(uploaded)) + app.drain() + + def _commit(): + return app.client.post( + f"/v1/captures/{capture_id}/files:commit", + json={"files": [{"path": item["path"], "sha256": item["sha256"]} for item in saved_files]}, + ) + + commit = _send("commit", _commit, injector, report, errors) + commit_body = commit.json() if commit.status_code == 200 else {} + record( + "commit", + httpStatus=commit.status_code, + committed=commit_body.get("committed", []), + notFound=commit_body.get("notFound", []), + mismatch=commit_body.get("mismatch", []), + ) + if commit.status_code != 200 or commit_body.get("notFound") or commit_body.get("mismatch"): + errors.append(f"commit_failed_http_{commit.status_code}") + if sorted(commit_body.get("committed", [])) != sorted(item["path"] for item in saved_files): + errors.append("commit_incomplete") + app.drain() + + # A lost commit reply must resume without re-uploading: register + # reports the files committed from server state, and the same commit + # body commits cleanly again with unchanged hashes. + if injector.fired.count("commit") or report["attempts"]["commit"][-1]["outcome"] == "simulated_timeout": + resumed = app.client.post(f"/v1/captures/{capture_id}/files", json={"files": saved_files}) + states = ( + {item["path"]: item["state"] for item in resumed.json().get("files", [])} + if resumed.status_code == 200 + else {} + ) + recommit = _commit() + recommit_body = recommit.json() if recommit.status_code == 200 else {} + record( + "resume-commit", + registerHttpStatus=resumed.status_code, + serverStates=states, + recommitHttpStatus=recommit.status_code, + recomputed=recommit_body.get("committed", []), + ) + if not states or not all(state == "committed" for state in states.values()): + errors.append("resume_register_not_committed") + if recommit.status_code != 200 or sorted(recommit_body.get("committed", [])) != sorted(states): + errors.append("resume_commit_failed") + + # Finalize with the saved key and packet digest. A lost finalize reply + # replays to the same run id: the run is derived from the capture and + # the packet bytes, so a retry can never start a second run. + raw = (packet_dir / "packet.json").read_bytes() + + def _finalize(): + return app.client.post( + f"/v1/captures/{capture_id}/finalize", + content=raw, + headers={ + "Idempotency-Key": finalize_key, + "Content-Type": "application/json", + "X-Packet-Sha256": packet_sha, + }, + ) + + accepted = _send("finalize", _finalize, injector, report, errors) + accepted_body = accepted.json() if accepted.status_code == 202 else {} + run_id = accepted_body.get("runId") + report["runId"] = run_id + record("finalize", idempotencyKey=finalize_key, httpStatus=accepted.status_code, runId=run_id) + if accepted.status_code != 202: + errors.append(f"finalize_http_{accepted.status_code}") + return _finish(report, errors, f"finalize failed: {accepted.text[:300]}") + expected_run_id = "run_" + hashlib.sha256(capture_id.encode() + b"\0" + raw).hexdigest()[:16] + if run_id != expected_run_id: + errors.append("run_id_not_deterministic") + if accepted_body.get("missing"): + errors.append("finalize_missing_files") + + def _finalize_again(): + return _finalize() + + replayed = _finalize_again() + replayed_body = replayed.json() if replayed.status_code == 202 else {} + record("finalize-replay", httpStatus=replayed.status_code, runId=replayed_body.get("runId")) + if replayed.status_code != 202 or replayed_body.get("runId") != run_id: + errors.append("finalize_replay_run_id_changed") + app.drain() + + # Terminal status and result, read the way a device reads them, with a + # bounded poll: a drained app is already terminal, so this settles in + # one read and any missed terminal state fails the replay loudly. + status_payload = None + for _ in range(MAX_POLLS): + status_payload = app.client.get(f"/v1/captures/{capture_id}").json() + if status_payload.get("status") in TERMINAL_STATUSES: + break + terminal = (status_payload or {}).get("status") + report["finalCaptureStatus"] = terminal + record("status", finalCaptureStatus=terminal, filesCommitted=(status_payload or {}).get("filesCommitted")) + if terminal not in TERMINAL_STATUSES: + errors.append(f"not_terminal:{terminal}") + + result_payload = None + for _ in range(MAX_POLLS): + got = app.client.get(f"/v1/captures/{capture_id}/result") + if got.status_code == 200 and got.json().get("outcome"): + result_payload = got.json() + break + result_payload = got.json() if got.status_code == 200 else None + report["resultOutcomeKind"] = (result_payload or {}).get("outcome", {}).get("kind") if result_payload else None + record( + "result", + httpStatus=200 if result_payload else None, + outcomeKind=report["resultOutcomeKind"], + runId=(result_payload or {}).get("runId"), + ) + if result_payload is None: + errors.append("result_not_served") + else: + if result_payload.get("runId") != run_id: + errors.append("result_run_id_mismatch") + if report["resultOutcomeKind"] != terminal: + errors.append("result_outcome_status_mismatch") + + # The terminal state must not duplicate anything: a final create with + # the saved key still maps to the one capture, and the event feed ends + # with exactly one terminal verdict for the one run. + final_create = app.client.post("/v1/captures", json=body, headers={"Idempotency-Key": create_key}) + record("create-terminal", httpStatus=final_create.status_code, captureId=final_create.json().get("captureId")) + if final_create.status_code != 200 or final_create.json().get("captureId") != capture_id: + errors.append("terminal_create_duplicated_capture") + + events: list[dict] = [] + after = 0 + while True: + page = app.client.get(f"/v1/captures/{capture_id}/events", params={"after": after, "wait": 0}).json() + events.extend(page.get("events", [])) + if not page.get("events") or page.get("next") in (None, after): + break + after = page["next"] + terminals = [event["type"] for event in events if event["type"] in TERMINAL_EVENTS] + report["terminalEvents"] = terminals + record("events", count=len(events), terminalEvents=terminals) + if terminals != ["verdict_ready"]: + errors.append(f"terminal_events:{terminals}") + except Exception as exc: # noqa: BLE001 - the report must survive any failure + errors.append(f"exception:{type(exc).__name__}") + return _finish(report, errors, f"exception: {type(exc).__name__}: {exc}") + finally: + if app is not None: + app.close() + return _finish(report, errors, None) + + +def _finish(report: dict, errors: list[str], detail: str | None) -> dict: + if detail: + report["errorDetail"] = detail + report["finishedAt"] = datetime.now(UTC).isoformat() + report["ok"] = not errors + return report + + +def write_report(report: dict, path: Path) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(report, indent=2) + "\n") + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument("--receipt", type=Path, required=True, help="saved capture_flow_e2e receipt to replay") + parser.add_argument("--workdir", type=Path, help="run directory (default: a fresh temp dir)") + parser.add_argument( + "--packet", type=Path, help="packet fixture (default: the receipt's packet dir, else the committed mini-arkit)" + ) + parser.add_argument("--report", type=Path, help="report path (default: /replay-receipt.json)") + parser.add_argument( + "--fail-at", + default="", + help="comma list of steps that lose one response: create,commit,finalize", + ) + args = parser.parse_args(argv) + faults = {step.strip(): 1 for step in args.fail_at.split(",") if step.strip()} + owned = args.workdir is None + workdir = args.workdir or Path(tempfile.mkdtemp(prefix="capture-replay-")) + report = run_replay(args.receipt, workdir, packet_dir=args.packet, faults=faults) + report_path = args.report or workdir / "replay-receipt.json" + write_report(report, report_path) + print( + json.dumps( + { + "ok": report["ok"], + "report": str(report_path), + "captureId": report["captureId"], + "finalCaptureStatus": report["finalCaptureStatus"], + **({"errorDetail": report["errorDetail"]} if "errorDetail" in report else {}), + } + ) + ) + if owned: + print(f"workdir: {workdir}") + return 0 if report["ok"] else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/backend/tests/test_replay_capture.py b/backend/tests/test_replay_capture.py new file mode 100644 index 000000000..15f9e3360 --- /dev/null +++ b/backend/tests/test_replay_capture.py @@ -0,0 +1,238 @@ +"""Checks for ``scripts/replay_capture.py``: the saved-receipt replay. + +Each test first runs the real runner (``scripts/capture_flow_e2e.py``) to +produce a saved receipt, then replays that receipt against a fresh instance of +the real app and inspects the replay report. The failure paths use the +deterministic ``FailureInjector`` — a scripted lost response that fires after +the request actually executed — never clocks, randomness, or unbounded +retries. Everything runs in-process over the local memory/volume backends with +orchestrator stage stubs; no network, no Modal, no credentials. +""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parents[1] + + +@pytest.fixture +def runner(): + import scripts.capture_flow_e2e as mod + + return mod + + +@pytest.fixture +def replay(): + import scripts.replay_capture as mod + + return mod + + +@pytest.fixture +def saved_receipt(runner, tmp_path) -> Path: + """A receipt the runner wrote to disk, the way a prior session leaves it.""" + workdir = tmp_path / "original" + receipt = runner.run_capture_flow(workdir) + assert receipt["ok"] is True, receipt.get("errorDetail") + path = workdir / "receipt.json" + runner.write_receipt(receipt, path) + return path + + +def _by_step(report: dict) -> dict: + return {step["step"]: step for step in report["steps"]} + + +def test_replay_of_saved_receipt_reaches_the_terminal_result(replay, saved_receipt, tmp_path) -> None: + report = replay.run_replay(saved_receipt, tmp_path / "replay") + report_path = tmp_path / "replay" / "replay-receipt.json" + replay.write_report(report, report_path) + + assert report["ok"] is True, report.get("errorDetail") + assert report["errorCodes"] == [] + assert report["finalCaptureStatus"] in replay.TERMINAL_STATUSES + assert report["resultOutcomeKind"] == report["finalCaptureStatus"] + assert report["fileHashesVerified"] > 0 + assert report["terminalEvents"] == ["verdict_ready"] + + steps = _by_step(report) + assert steps["create"]["idempotencyKey"] == "mini-arkit" + assert steps["finalize"]["idempotencyKey"] == "mini-arkit" + assert steps["fixture"]["drift"] == [] + assert steps["result"]["runId"] == report["runId"] == steps["finalize"]["runId"] + + stored = json.loads(report_path.read_text()) + assert stored["ok"] is True and stored["captureId"] == report["captureId"] + + +def test_retry_after_simulated_timeout_keeps_one_capture_id(replay, saved_receipt, tmp_path) -> None: + report = replay.run_replay(saved_receipt, tmp_path / "replay", faults={"create": 1}) + + assert report["ok"] is True, report.get("errorDetail") + attempts = report["attempts"]["create"] + # The first request really executed (201); its response was lost, and the + # identical retry came back from the idempotency store: 200, same capture. + assert [attempt["outcome"] for attempt in attempts] == ["simulated_timeout", "delivered"] + assert attempts[0]["httpStatus"] == 201 + assert attempts[1]["httpStatus"] == 200 + + steps = _by_step(report) + assert steps["create"]["httpStatus"] == 200 + assert steps["create-stable"]["httpStatus"] == 200 + assert steps["create-stable"]["stable"] is True + assert steps["create-terminal"]["captureId"] == report["captureId"] + + +def test_resume_after_files_already_committed_reuses_the_server_state(replay, saved_receipt, tmp_path) -> None: + report = replay.run_replay(saved_receipt, tmp_path / "replay", faults={"commit": 1}) + + assert report["ok"] is True, report.get("errorDetail") + steps = _by_step(report) + assert steps["commit"]["notFound"] == [] and steps["commit"]["mismatch"] == [] + + # The replay learned what it may skip from the server, not from itself: + # re-register reports every file committed, so no bytes were PUT twice. + resume = steps["resume-commit"] + assert resume["registerHttpStatus"] == 200 + assert resume["serverStates"] and set(resume["serverStates"].values()) == {"committed"} + assert resume["recommitHttpStatus"] == 200 + assert sorted(resume["recomputed"]) == sorted(resume["serverStates"]) + assert steps["upload"]["skippedCommitted"] == 0 # nothing was skipped pre-commit + + # The resumed commit committed the same hashes: the terminal journey holds. + assert report["finalCaptureStatus"] in replay.TERMINAL_STATUSES + assert report["terminalEvents"] == ["verdict_ready"] + + +def test_retry_budget_is_bounded_and_fails_loudly(replay, saved_receipt, tmp_path) -> None: + # Three lost finalize responses against a one-retry budget: the replay + # must give up with a named error, not retry forever. + report = replay.run_replay(saved_receipt, tmp_path / "replay", faults={"finalize": 3}) + + assert report["ok"] is False + assert "finalize_retry_budget_exhausted" in report["errorCodes"] + assert len(report["attempts"]["finalize"]) == replay.MAX_RETRIES + 1 + + +def test_replay_fails_loudly_when_the_fixture_drifts_from_the_receipt(replay, saved_receipt, tmp_path) -> None: + receipt = json.loads(saved_receipt.read_text()) + receipt["files"][0]["sha256"] = "0" * 64 + drifted = tmp_path / "drifted.json" + drifted.write_text(json.dumps(receipt)) + + report = replay.run_replay(drifted, tmp_path / "replay") + + assert report["ok"] is False + assert any(code.startswith("fixture_drift:") for code in report["errorCodes"]) + assert report["captureId"] is None # nothing was sent to the API + + +def test_cli_replays_a_saved_receipt_and_exits_zero(replay, saved_receipt, tmp_path) -> None: + env = os.environ.copy() + env["PYTHONPATH"] = str(ROOT) + completed = subprocess.run( + [ + sys.executable, + "scripts/replay_capture.py", + "--receipt", + str(saved_receipt), + "--workdir", + str(tmp_path / "cli"), + ], + cwd=ROOT, + check=False, + capture_output=True, + text=True, + env=env, + ) + assert completed.returncode == 0, completed.stdout + completed.stderr + stored = json.loads((tmp_path / "cli" / "replay-receipt.json").read_text()) + assert stored["ok"] is True + assert stored["finalCaptureStatus"] in replay.TERMINAL_STATUSES + + +def test_replay_rejects_a_pre_replay_receipt_missing_the_replay_block(replay, saved_receipt, tmp_path) -> None: + """A receipt written before the replay block exists must fail loudly. + + The runner's replay block (files, idempotencyKeys, packetSha256) is what + makes the receipt replayable; a pre-PR receipt has none of it and must be + rejected with a named error, not replayed with guessed hashes — and the + CLI must exit non-zero instead of reporting a green run. + """ + receipt = json.loads(saved_receipt.read_text()) + for key in ("files", "idempotencyKeys", "packetSha256"): + receipt.pop(key, None) + stripped = tmp_path / "stripped" / "receipt.json" + stripped.parent.mkdir(parents=True) + stripped.write_text(json.dumps(receipt)) + + report = replay.run_replay(stripped, tmp_path / "replay") + + assert report["ok"] is False, report + assert any(code.startswith("missing_replay_block") for code in report["errorCodes"]), report["errorCodes"] + assert report["captureId"] is None + + env = {**os.environ, "PYTHONPATH": str(ROOT)} + completed = subprocess.run( + [ + sys.executable, + str(ROOT / "scripts" / "replay_capture.py"), + "--receipt", + str(stripped), + "--workdir", + str(tmp_path / "cli"), + ], + capture_output=True, + text=True, + env=env, + ) + assert completed.returncode == 1, completed.stdout + completed.stderr + stored = json.loads((tmp_path / "cli" / "replay-receipt.json").read_text()) + assert stored["ok"] is False + + +def test_a_crash_mid_replay_reports_ok_false(replay, saved_receipt, tmp_path) -> None: + """No exception path may leave the report green: a crash is a failure. + + The receipt's files block is malformed (an entry without path/sha256), so + the replay dies before the first error code would be appended. The report + must still come back ok=False with an exception error code — and the CLI + must exit 1. + """ + receipt = json.loads(saved_receipt.read_text()) + receipt["files"] = [{"path": "x.bin", "sha256": "0" * 64}, {"broken": True}] + broken = tmp_path / "broken" / "receipt.json" + broken.parent.mkdir(parents=True) + broken.write_text(json.dumps(receipt)) + + report = replay.run_replay(broken, tmp_path / "replay") + + assert report["ok"] is False, report + assert any(code.startswith("exception:") for code in report["errorCodes"]), report["errorCodes"] + assert report["captureId"] is None + + +def test_fixture_check_flags_receipt_files_outside_the_manifest(replay) -> None: + """A receipt claiming files the packet fixture never had is drift. + + The old comparison only looked at paths present in both sets, so an empty + or disjoint receipt files list passed vacuously. Every receipt path must + exist in the committed manifest. + """ + errors: list[str] = [] + + fixture = replay.check_fixture( + replay.DEFAULT_PACKET, {"files": [{"path": "not/in/this-packet.bin", "sha256": "0" * 64}]}, errors + ) + + assert fixture is not None and fixture["drift"], fixture + assert errors and errors[0].startswith("fixture_drift:"), errors + assert "not/in/this-packet.bin:not-in-manifest" in fixture["drift"]