From f0df5cd9dc2fb52c0bb53b53b9ec0480967bf74c Mon Sep 17 00:00:00 2001 From: Sam Gu <127461594+SamGu-NRX@users.noreply.github.com> Date: Thu, 15 Oct 2026 10:23:47 +0000 Subject: [PATCH 1/2] test(orchestrator): prove the deterministic local stub path on a committed packet MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit One committed synthetic packet (tests/data/packets/synthetic-meter.tar) drives run-local — the supported local entry point — with all eleven stage stubs, no Modal, GPU, network or credentials. Asserts the run, eleven done stage records, one terminal verdict event, the API-consumed result artifacts, byte-identical determinism across two runs, and that an exhausted stage lands as recorded failure data (manual result published, failed event emitted) instead of a coordinator crash. --- .../tests/data/packets/synthetic-meter.tar | Bin 0 -> 10240 bytes .../tests/test_local_fixture_proof.py | 197 ++++++++++++++++++ 2 files changed, 197 insertions(+) create mode 100644 backend/server/orchestrator/tests/data/packets/synthetic-meter.tar create mode 100644 backend/server/orchestrator/tests/test_local_fixture_proof.py diff --git a/backend/server/orchestrator/tests/data/packets/synthetic-meter.tar b/backend/server/orchestrator/tests/data/packets/synthetic-meter.tar new file mode 100644 index 0000000000000000000000000000000000000000..9c6b4bc4f17825e5b5be6335caa1b030a7b6588c GIT binary patch literal 10240 zcmeH}-EP`26vw&dDNwFQu@gHVeS=+0lQwBLQ-vJo7*Ya>;-D5yeb4P_I@v$73MxW#9B9^Hj>UdeoTkw-_L(O=E`tYmk`js{I7&uKy!1i%he^_FY>WT zeIoxU3!~kcu+9I#X94E_MR4Ef^lLa8c~Vr(x`xl0ck3!sl|R~sX<5s;_sP1byZ~*? zKTTycvb}8a(oN%cM)5VPFlX49kShBs@8Q?YPYZ4LuM)qTiwsr;%!9r$|i;amawvG@?778IUV1&u^VdmG<;*$ z{XgCN=l=`5SCV<={3nFc{`t>*g6IF0@ShyUzW-dzPkNyd2_OL^fCP{L5.py``). + +Two runs of the same packet must agree on stage outcomes and on the +deterministic artifact bytes, and an injected stage failure must land as +recorded failure data, not as an uncaught coordinator crash. +""" + +from __future__ import annotations + +import hashlib +import json +from collections.abc import Callable +from pathlib import Path +from typing import Any + +import pytest + +from server.contracts.result import ResultResponse +from server.orchestrator import runner +from server.orchestrator.cli import main +from server.orchestrator.graph import STAGE_NAMES + +FIXTURE = Path(__file__).parent / "data" / "packets" / "synthetic-meter.tar" +CAPTURE_ID = "cap_fixture_orch" + +# The documented local contract (orchestrator README "Run locally"). ``run-local`` +# pins these in its own Settings too; the pin here keeps the test honest if that +# ever drifts. +LOCAL_ENV = { + "STATE_BACKEND": "memory", + "STORAGE_BACKEND": "local", + "HSS_ORCH_STAGE_STUBS": "1", + "VISION_EXTERNAL_CALLS": "off", + "SMP_ENVIRONMENT": "dev", +} + + +@pytest.fixture +def no_modal(monkeypatch: pytest.MonkeyPatch) -> list[str]: + """Fail the run if anything reaches for Modal.""" + touched: list[str] = [] + + def _forbidden(*args: Any, **kwargs: Any) -> bool: + touched.append("modal_available") + return False + + monkeypatch.setattr("server.orchestrator.cli.modal_available", _forbidden) + return touched + + +@pytest.fixture +def env(monkeypatch: pytest.MonkeyPatch) -> None: + for key, value in LOCAL_ENV.items(): + monkeypatch.setenv(key, value) + + +@pytest.fixture +def stub_trace(monkeypatch: pytest.MonkeyPatch) -> list[tuple[str, bool]]: + """Record (stage, stubs?) for every stage the coordinator dispatches.""" + calls: list[tuple[str, bool]] = [] + real = runner.stage_callable + + def spy(stage: str, *, stubs: bool) -> Callable[..., Any]: + calls.append((str(stage), stubs)) + return real(stage, stubs=stubs) # type: ignore[arg-type,return-value] + + monkeypatch.setattr(runner, "stage_callable", spy) + return calls + + +def _events(out: Path) -> list[dict]: + return [json.loads(line) for line in (out / "events.jsonl").read_text().splitlines()] + + +def _latest_stage_status(events: list[dict]) -> dict[str, dict]: + latest: dict[str, dict] = {} + for event in events: + if event["type"] != "stage": + continue + data = event.get("data") or {} + if data.get("stage"): + latest[str(data["stage"])] = data + return latest + + +def _terminal(events: list[dict]) -> list[str]: + return [event["type"] for event in events if event["type"] in {"verdict_ready", "retake_request", "failed"}] + + +def _stored(out: Path, key: str) -> bytes: + return (out / "store" / key).read_bytes() + + +def _run(out: Path) -> dict: + return json.loads((out / "run.json").read_text()) + + +def _run_local(out: Path, monkeypatch: pytest.MonkeyPatch, no_modal: list[str]) -> int: + assert not no_modal, "Modal was reached during run-local" + return main(["run-local", "--tar", str(FIXTURE), "--variant", "arkit-nonlidar", "--out", str(out)]) + + +def test_committed_packet_runs_to_a_terminal_verdict( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, no_modal: list[str], env: None, stub_trace: list[tuple[str, bool]] +) -> None: + out = tmp_path / "out" + code = _run_local(out, monkeypatch, no_modal) + assert code == 0 + assert not no_modal, "run-local reached for Modal" + + run = _run(out) + assert run["status"] == "done" + assert run["outcome_kind"] == "manual_review" # stub verdicts are manual_review, views_needed [] + assert run["failure_code"] is None + assert run["packet_sha256"] == hashlib.sha256(_stored(out, f"captures/{CAPTURE_ID}/packet.json")).hexdigest() + + # Every stubbed stage, named from the execution trace: all eleven, stubs on. + traced = {stage for stage, stubs in stub_trace if stubs} + assert traced == set(STAGE_NAMES), f"stubbed stages traced: {sorted(traced)}" + + # Stage records: all eleven reach done. (The graph still *plans* GPU names on + # stage events; the stubs bypass GPU execution — proven by the stub trace above.) + latest = _latest_stage_status(_events(out)) + assert set(latest) == set(STAGE_NAMES) + for stage in STAGE_NAMES: + assert latest[stage]["status"] == "done", stage + + # Exactly one terminal outcome, and it is the verdict. + terminals = _terminal(_events(out)) + assert terminals == ["verdict_ready"] + + # The artifacts the API consumes (server/api/routes/result.py reads + # run.artifacts["result/result.json"]): present, stored, and schema-valid. + assert run["artifacts"]["result/result.json"].startswith(f"captures/{CAPTURE_ID}/runs/{run['run_id']}/") + result = ResultResponse.model_validate(json.loads(_stored(out, run["artifacts"]["result/result.json"]))) + assert result.status == "manual_review" + assert result.outcome.kind == "manual_review" + for key in ("result/review.html", "result/preview.ply", "verdicts/verdict.C.json"): + assert run["artifacts"][key] + assert len(_stored(out, run["artifacts"][key])) > 0 + + +def test_same_packet_twice_is_deterministic( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, no_modal: list[str], env: None +) -> None: + """Stub outputs are pure: same packet bytes, same artifact bytes, same stage outcomes.""" + outs = [tmp_path / "first", tmp_path / "second"] + digests: list[dict[str, str]] = [] + signatures: list[list[tuple[str, str]]] = [] + for out in outs: + assert _run_local(out, monkeypatch, no_modal) == 0 + run = _run(out) + digests.append( + { + key: hashlib.sha256(_stored(out, stored)).hexdigest() + for key, stored in run["artifacts"].items() + if key in {"verdicts/verdict.C.json", "result/result.internal.json", "scene/scene.json"} + } + ) + latest = _latest_stage_status(_events(out)) + signatures.append([(stage, latest[stage]["status"]) for stage in STAGE_NAMES]) + assert digests[0] == digests[1] + assert signatures[0] == signatures[1] + + +def test_stage_failure_is_recorded_data_not_a_crash( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, no_modal: list[str], env: None +) -> None: + """``scale`` exhausted its attempts: the run ends as recorded failure data.""" + monkeypatch.setenv("HSS_ORCH_TEST_FAIL_STAGE", "scale:compute_error:2") + out = tmp_path / "out" + assert _run_local(out, monkeypatch, no_modal) == 0 # no uncaught coordinator crash + + run = _run(out) + assert run["status"] == "failed" + assert run["failure_code"] == "stage_exhausted:scale:compute_error" + assert run["outcome_kind"] == "manual_review" # row 3: exhausted required stage -> manual review + + latest = _latest_stage_status(_events(out)) + assert latest["scale"]["status"] == "failed" + assert latest["scale"]["attempt"] >= 2 # retried before being recorded failed + + terminals = _terminal(_events(out)) + assert terminals == ["failed"] + + # Even the failed run publishes the API's result document (manual result). + result = ResultResponse.model_validate(json.loads(_stored(out, run["artifacts"]["result/result.json"]))) + assert result.status == "manual_review" From adf8cc877bb445ed037b614b7c68f5f46a210ae5 Mon Sep 17 00:00:00 2001 From: Sam Gu <127461594+SamGu-NRX@users.noreply.github.com> Date: Thu, 15 Oct 2026 10:23:58 +0000 Subject: [PATCH 2/2] chore: add the required co-author trailer to the previous commit Previous commit f0df5cd9 (test(orchestrator): prove the deterministic local stub path on a committed packet) is this same change. Co-authored-by: obvious-autobuild[bot] <262744130+obvious-autobuild[bot]@users.noreply.github.com>