From 911efd6d8eba29ed79434bd70583cca7d80529b9 Mon Sep 17 00:00:00 2001 From: Sam Gu <127461594+SamGu-NRX@users.noreply.github.com> Date: Thu, 15 Oct 2026 10:32:00 +0000 Subject: [PATCH 1/2] feat(backend): one repeatable no-cloud capture flow runner with a JSON receipt scripts/capture_flow_e2e.py drives create_app() in-process over the committed mini-arkit packet: create, register, signed-URL PUTs, commit, finalize, events. It writes a receipt (request IDs, per-stage HTTP statuses, file counts, final capture status, error codes), exits nonzero on any unexpected result, and supports --skip-upload to reach the awaiting_files path on purpose. A test covers the happy-path receipt, the nonzero failure exit, and the CLI from a clean directory. --- backend/scripts/capture_flow_e2e.py | 345 ++++++++++++++++++ backend/tests/test_capture_flow_e2e_runner.py | 95 +++++ 2 files changed, 440 insertions(+) create mode 100644 backend/scripts/capture_flow_e2e.py create mode 100644 backend/tests/test_capture_flow_e2e_runner.py diff --git a/backend/scripts/capture_flow_e2e.py b/backend/scripts/capture_flow_e2e.py new file mode 100644 index 000000000..b8d4b381b --- /dev/null +++ b/backend/scripts/capture_flow_e2e.py @@ -0,0 +1,345 @@ +#!/usr/bin/env python3 +"""Drive the whole capture journey over the committed mini-arkit packet, no cloud. + +One repeatable command for reviewers: + + cd backend && uv run --extra api python scripts/capture_flow_e2e.py + +It copies the committed ``data/fixtures/packets/mini-arkit`` packet into a clean +temporary directory, starts the real app entry point ``create_app()`` in-process +(memory state, local volume storage, static auth, a test HMAC secret — no GCP, +no Modal, no secrets), and sends the app's own sequence: create the capture, +register files, PUT bytes to the signed upload URLs, commit, finalize +``packet.json``, and read events. It executes the HTTP API and the finalize +handoff only; it does not run the reconstruction/orchestrator pipeline, does not +touch the network beyond the in-process ASGI transport, and never deploys. + +It writes a small JSON receipt (default ``/receipt.json``) with request +IDs, per-stage HTTP statuses, file counts, the final capture status and error +codes, then exits 0 when the run matched the expected outcome and 1 otherwise. +The receipt is the replay anchor: rerunning the command reproduces the same +journey, and a failed run leaves the exact captureId, runId and failing stage +behind so a later session can replay or diff against it. + +``--skip-upload PATH`` stops one file short on purpose, which drives the +capture into ``awaiting_files`` — combine with ``--expect-status awaiting_files`` +to check that path instead. +""" + +from __future__ import annotations + +import argparse +import base64 +import hashlib +import json +import shutil +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 + + +def _content_type(role: str) -> str: + if role == "stream": + return "application/gzip" + if role in {"features", "depth", "confidence", "mesh"}: + return "application/octet-stream" + return "image/jpeg" + + +def _sealed_files(root: Path, packet: dict) -> list[dict]: + """Register bodies with the hashes the app computes from the sealed files.""" + files = [] + for entry in packet["files"]: + data = (root / entry["path"]).read_bytes() + role = entry.get("role", "still") + files.append( + { + "path": entry["path"], + "bytes": len(data), + "sha256": hashlib.sha256(data).hexdigest(), + "md5": base64.b64encode(hashlib.md5(data).digest()).decode(), + "role": role, + "contentType": _content_type(role), + } + ) + return files + + +class _Journey: + """The app under test plus a private copy of the committed packet fixture.""" + + def __init__(self, workdir: Path) -> None: + self.workdir = workdir + self.root = workdir / "packet" + self.packet = json.loads((self.root / "packet.json").read_text()) + 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="null", + VISION_EXTERNAL_CALLS="on", + 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 drain(self) -> None: + """Join finalize background work so the receipt is deterministic.""" + drain_finalize(self.app.state.deps) + + +def run_capture_flow( + workdir: Path, + *, + packet_dir: Path | None = None, + skip_upload: str | None = None, + expect_status: str = "processing", +) -> dict: + """Run the journey and return the receipt dict. Never raises: failures land + in ``receipt["errorCodes"]`` with ``receipt["ok"] is False``.""" + started_at = datetime.now(UTC).isoformat() + packet_dir = packet_dir or DEFAULT_PACKET + receipt: dict = { + "runner": "capture_flow_e2e", + "startedAt": started_at, + "packetDir": str(packet_dir), + "workdir": str(workdir), + "expectStatus": expect_status, + "skipUpload": skip_upload, + "captureId": None, + "runId": None, + "stages": [], + "fileCounts": {"registered": 0, "uploaded": 0, "committed": 0, "expected": 0}, + "finalCaptureStatus": None, + "errorCodes": [], + "ok": False, + } + errors = receipt["errorCodes"] + stages = receipt["stages"] + + def record(stage: str, **fields) -> None: + stages.append({"stage": stage, **fields}) + + try: + workdir.mkdir(parents=True, exist_ok=True) + fixture_root = workdir / "packet" + shutil.copytree(packet_dir, fixture_root) + journey = _Journey(workdir) + packet = journey.packet + receipt["packet"] = { + "packetId": packet["packetId"], + "formatVersion": packet["formatVersion"], + "tier": packet["session"]["tier"], + "files": len(packet["files"]), + } + + # Create the capture. + session = packet["session"] + created = journey.client.post( + "/v1/captures", + json={ + "packetId": packet["packetId"], + "formatVersion": packet["formatVersion"], + "tier": session["tier"], + "flow": "guided", + "device": { + "model": session["deviceModel"], + "systemVersion": session["systemVersion"], + "appVersion": session["appVersion"], + }, + }, + headers={"Idempotency-Key": packet["packetId"]}, + ) + body = created.json() if created.status_code == 201 else {} + capture_id = body.get("captureId") + receipt["captureId"] = capture_id + record( + "create", + httpStatus=created.status_code, + captureId=capture_id, + idempotencyKey=packet["packetId"], + status=body.get("status"), + ) + if created.status_code != 201: + errors.append(f"create_http_{created.status_code}") + if body.get("status") != "uploading": + errors.append("create_status_not_uploading") + if not capture_id: + return _finish(receipt, errors, f"no captureId; body={created.text[:300]}") + + # Register, PUT to signed URLs, commit. + files = _sealed_files(fixture_root, packet) + 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)) + if registered.status_code != 200: + errors.append(f"register_http_{registered.status_code}") + return _finish(receipt, errors, f"register failed: {registered.text[:300]}") + receipt["fileCounts"]["registered"] = len(registered.json().get("files", [])) + + uploaded: list[str] = [] + for item, target in zip(files, registered.json()["files"], strict=True): + if target["path"] == skip_upload: + continue + put = TestClient(journey.app).put( + target["upload"]["url"], + content=(fixture_root / item["path"]).read_bytes(), + 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']}") + journey.drain() + receipt["fileCounts"]["uploaded"] = len(uploaded) + record( + "upload", + uploaded=len(uploaded), + skipped=[skip_upload] if skip_upload else [], + ) + + commit = journey.client.post( + f"/v1/captures/{capture_id}/files:commit", + json={"files": [{"path": item["path"], "sha256": item["sha256"]} for item in files]}, + ) + commit_body = commit.json() if commit.status_code == 200 else {} + receipt["fileCounts"]["committed"] = len(commit_body.get("committed", [])) + 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: + errors.append(f"commit_http_{commit.status_code}") + if commit_body.get("notFound"): + errors.append("commit_not_found") + if commit_body.get("mismatch"): + errors.append("commit_sha_mismatch") + journey.drain() + + # Finalize packet.json. + raw = (fixture_root / "packet.json").read_bytes() + accepted = journey.client.post( + f"/v1/captures/{capture_id}/finalize", + content=raw, + headers={ + "Idempotency-Key": packet["packetId"], + "Content-Type": "application/json", + "X-Packet-Sha256": hashlib.sha256(raw).hexdigest(), + }, + ) + payload = accepted.json() if accepted.status_code == 202 else {} + run_id = payload.get("runId") + receipt["runId"] = run_id + record( + "finalize", + httpStatus=accepted.status_code, + status=payload.get("status"), + runId=run_id, + missing=payload.get("missing", []), + ) + if accepted.status_code != 202: + errors.append(f"finalize_http_{accepted.status_code}") + if payload.get("missing"): + errors.append("finalize_missing_files") + journey.drain() + + # Final status and the event log the device observes. + status = journey.client.get(f"/v1/captures/{capture_id}").json() + receipt["finalCaptureStatus"] = status.get("status") + record( + "status", + finalCaptureStatus=status.get("status"), + filesCommitted=status.get("filesCommitted"), + ) + if status.get("status") != expect_status: + errors.append("unexpected_final_status") + + events = journey.client.get(f"/v1/captures/{capture_id}/events").json() + types = [event["type"] for event in events.get("events", [])] + validate_ok = any( + (event.get("data") or {}).get("stage") == "validate" and (event.get("data") or {}).get("runId") == run_id + for event in events.get("events", []) + ) + record("events", count=len(events.get("events", [])), types=types, validateStageForRun=validate_ok) + if events.get("status") != status.get("status"): + errors.append("events_status_mismatch") + if expect_status == "processing" and not validate_ok: + errors.append("events_missing_validate_stage") + except Exception as exc: # noqa: BLE001 - the receipt must survive any failure + return _finish(receipt, errors, f"exception: {type(exc).__name__}: {exc}") + return _finish(receipt, errors, None) + + +def _finish(receipt: dict, errors: list[str], detail: str | None) -> dict: + if detail: + receipt["errorDetail"] = detail + receipt["finishedAt"] = datetime.now(UTC).isoformat() + receipt["ok"] = not errors + return receipt + + +def write_receipt(receipt: dict, path: Path) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(receipt, indent=2) + "\n") + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument("--workdir", type=Path, help="run directory (default: a fresh temp dir)") + parser.add_argument("--packet", type=Path, default=DEFAULT_PACKET, help="committed packet fixture") + parser.add_argument("--receipt", type=Path, help="receipt path (default: /receipt.json)") + parser.add_argument("--skip-upload", help="register but never upload this file (fault injection)") + parser.add_argument("--expect-status", default="processing", help="final capture status the run must end in") + args = parser.parse_args(argv) + owned = args.workdir is None + workdir = args.workdir or Path(tempfile.mkdtemp(prefix="capture-flow-e2e-")) + receipt = run_capture_flow( + workdir, + packet_dir=args.packet, + skip_upload=args.skip_upload, + expect_status=args.expect_status, + ) + receipt_path = args.receipt or workdir / "receipt.json" + write_receipt(receipt, receipt_path) + print( + json.dumps( + { + "ok": receipt["ok"], + "receipt": str(receipt_path), + **({"errorDetail": receipt["errorDetail"]} if "errorDetail" in receipt else {}), + } + ) + ) + if owned: + # Keep the default-run evidence for inspection; the caller can delete it. + print(f"workdir: {workdir}") + return 0 if receipt["ok"] else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/backend/tests/test_capture_flow_e2e_runner.py b/backend/tests/test_capture_flow_e2e_runner.py new file mode 100644 index 000000000..830752963 --- /dev/null +++ b/backend/tests/test_capture_flow_e2e_runner.py @@ -0,0 +1,95 @@ +"""Checks for ``scripts/capture_flow_e2e.py``: the receipt and the failure exit. + +The happy-path run replays the fixture-test journey through the runner and +inspects the receipt it writes; the failure path injects one never-uploaded +file and asserts the receipt records the error codes and ``main`` exits +nonzero. The CLI itself is exercised the way a reviewer would run it, as a +subprocess from ``backend/``. +""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parents[1] +FIRST_FILE = "keyframes/k00001.jpg" + + +def _read_receipt(path: Path) -> dict: + return json.loads(path.read_text()) + + +@pytest.fixture +def runner(): + import scripts.capture_flow_e2e as mod + + return mod + + +def test_happy_path_receipt_records_ids_statuses_counts(runner, tmp_path) -> None: + receipt = runner.run_capture_flow(tmp_path) + receipt_path = tmp_path / "receipt.json" + runner.write_receipt(receipt, receipt_path) + stored = _read_receipt(receipt_path) + + assert receipt["ok"] is True, receipt.get("errorDetail") + assert receipt["errorCodes"] == [] + assert receipt["captureId"] and receipt["runId"] + assert stored["captureId"] == receipt["captureId"] and stored["runId"] == receipt["runId"] + + by_stage = {stage["stage"]: stage for stage in receipt["stages"]} + assert by_stage["create"]["httpStatus"] == 201 + assert by_stage["create"]["idempotencyKey"] + assert by_stage["register"]["httpStatus"] == 200 + assert by_stage["commit"]["httpStatus"] == 200 + assert by_stage["commit"]["notFound"] == [] and by_stage["commit"]["mismatch"] == [] + assert by_stage["finalize"]["httpStatus"] == 202 + assert by_stage["finalize"]["missing"] == [] + assert by_stage["events"]["count"] > 0 and "files_committed" in by_stage["events"]["types"] + assert by_stage["events"]["validateStageForRun"] is True + + counts = receipt["fileCounts"] + assert counts["expected"] == counts["registered"] == counts["uploaded"] == counts["committed"] + assert receipt["finalCaptureStatus"] == "processing" + assert by_stage["status"]["filesCommitted"] == counts["expected"] + + +def test_missing_upload_fails_with_error_codes_and_nonzero_exit(runner, tmp_path) -> None: + receipt = runner.run_capture_flow(tmp_path, skip_upload=FIRST_FILE) + + assert receipt["ok"] is False + assert receipt["finalCaptureStatus"] == "awaiting_files" + assert "finalize_missing_files" in receipt["errorCodes"] + assert "unexpected_final_status" in receipt["errorCodes"] + assert receipt["fileCounts"]["uploaded"] == receipt["fileCounts"]["expected"] - 1 + finalize = next(stage for stage in receipt["stages"] if stage["stage"] == "finalize") + assert finalize["missing"] == [FIRST_FILE] + + # The same fault through the CLI exits nonzero and still writes the receipt. + code = runner.main(["--workdir", str(tmp_path / "cli"), "--skip-upload", FIRST_FILE]) + assert code == 1 + stored = _read_receipt(tmp_path / "cli" / "receipt.json") + assert stored["ok"] is False and stored["errorCodes"] + + +def test_cli_runs_from_a_clean_directory_and_exits_zero(runner, tmp_path) -> None: + workdir = tmp_path / "clean" + env = os.environ.copy() + env["PYTHONPATH"] = str(ROOT) + completed = subprocess.run( + [sys.executable, "scripts/capture_flow_e2e.py", "--workdir", str(workdir)], + cwd=ROOT, + check=False, + capture_output=True, + text=True, + env=env, + ) + assert completed.returncode == 0, completed.stdout + completed.stderr + receipt = _read_receipt(workdir / "receipt.json") + assert receipt["ok"] is True and receipt["finalCaptureStatus"] == "processing" From da0b63801d54762f6f0b3140ffbe2bc5accd5ff7 Mon Sep 17 00:00:00 2001 From: Sam Gu <127461594+SamGu-NRX@users.noreply.github.com> Date: Thu, 15 Oct 2026 19:57:10 +0000 Subject: [PATCH 2/2] fix(backend): fail closed when the capture-flow runner crashes before a stage A crashed run reported ok:true and exited 0: _finish computed success from errorCodes alone, so an exception with no stage code yet (rerunning into the receipt's recorded workdir hits copytree FileExistsError; a missing --packet hits FileNotFoundError) wrote ok:true beside the errorDetail and main exited 0. The except handler now records a runner_exception code and _finish treats any recorded detail as failure (ok = not errors and not detail). Two new tests cover the rerun-into-used-workdir and missing-packet crashes; both failed against the previous behavior. --- backend/scripts/capture_flow_e2e.py | 8 ++++-- backend/tests/test_capture_flow_e2e_runner.py | 27 +++++++++++++++++++ 2 files changed, 33 insertions(+), 2 deletions(-) diff --git a/backend/scripts/capture_flow_e2e.py b/backend/scripts/capture_flow_e2e.py index b8d4b381b..ab36e8f48 100644 --- a/backend/scripts/capture_flow_e2e.py +++ b/backend/scripts/capture_flow_e2e.py @@ -118,7 +118,8 @@ def run_capture_flow( expect_status: str = "processing", ) -> dict: """Run the journey and return the receipt dict. Never raises: failures land - in ``receipt["errorCodes"]`` with ``receipt["ok"] is False``.""" + in ``receipt["errorCodes"]`` with ``receipt["ok"] is False`` — including an + exception before any stage, which becomes the ``runner_exception`` code.""" started_at = datetime.now(UTC).isoformat() packet_dir = packet_dir or DEFAULT_PACKET receipt: dict = { @@ -291,6 +292,7 @@ def record(stage: str, **fields) -> None: if expect_status == "processing" and not validate_ok: errors.append("events_missing_validate_stage") except Exception as exc: # noqa: BLE001 - the receipt must survive any failure + errors.append("runner_exception") return _finish(receipt, errors, f"exception: {type(exc).__name__}: {exc}") return _finish(receipt, errors, None) @@ -299,7 +301,9 @@ def _finish(receipt: dict, errors: list[str], detail: str | None) -> dict: if detail: receipt["errorDetail"] = detail receipt["finishedAt"] = datetime.now(UTC).isoformat() - receipt["ok"] = not errors + # A recorded detail without an error code is still a failure: a run that + # crashed before any stage could record a code must never report ok. + receipt["ok"] = not errors and not detail return receipt diff --git a/backend/tests/test_capture_flow_e2e_runner.py b/backend/tests/test_capture_flow_e2e_runner.py index 830752963..10ef1bab3 100644 --- a/backend/tests/test_capture_flow_e2e_runner.py +++ b/backend/tests/test_capture_flow_e2e_runner.py @@ -93,3 +93,30 @@ def test_cli_runs_from_a_clean_directory_and_exits_zero(runner, tmp_path) -> Non assert completed.returncode == 0, completed.stdout + completed.stderr receipt = _read_receipt(workdir / "receipt.json") assert receipt["ok"] is True and receipt["finalCaptureStatus"] == "processing" + + +def test_rerun_into_a_used_workdir_fails_closed(runner, tmp_path) -> None: + """The replay action the receipt invites — rerun with its recorded --workdir — + hits the already-present packet copy, so the crash must exit 1, not report + success with the errorDetail tucked beside ok:true.""" + used = tmp_path / "replay" + assert runner.run_capture_flow(used)["ok"] is True + + receipt = runner.run_capture_flow(used) + assert receipt["ok"] is False + assert "runner_exception" in receipt["errorCodes"] + assert receipt["errorDetail"] + + code = runner.main(["--workdir", str(used)]) + assert code == 1 + stored = _read_receipt(used / "receipt.json") + assert stored["ok"] is False and "runner_exception" in stored["errorCodes"] + + +def test_missing_packet_dir_fails_closed(runner, tmp_path) -> None: + code = runner.main(["--packet", str(tmp_path / "no-such-packet"), "--workdir", str(tmp_path / "w")]) + assert code == 1 + stored = _read_receipt(tmp_path / "w" / "receipt.json") + assert stored["ok"] is False + assert "runner_exception" in stored["errorCodes"] + assert stored["errorDetail"]