From 92e47e7adbdb64d21a36830d29c9a31eb40ef973 Mon Sep 17 00:00:00 2001 From: Sam Gu <127461594+SamGu-NRX@users.noreply.github.com> Date: Fri, 16 Oct 2026 17:54:09 +0000 Subject: [PATCH 1/3] feat(result): name which stages ran as stand-ins The result composer reads every impl/.json marker the orchestration publishes into the result stage's inputs, refuses a marker without the pinned hss-stage-impl/v1 schema, and records stageImplementations plus a sorted standIns list in the internal record. The reviewer page names the stand-in steps next to the true decision. result.json is unchanged: the iOS decoder pins it. result/'s STAGE_INPUTS gains impl/ so the markers hydrate. --- backend/server/orchestrator/graph.py | 1 + backend/server/result/compose.py | 45 +++++++++++++++++++++- backend/server/result/review/template.html | 3 ++ 3 files changed, 48 insertions(+), 1 deletion(-) diff --git a/backend/server/orchestrator/graph.py b/backend/server/orchestrator/graph.py index f269b53dd..a14507825 100644 --- a/backend/server/orchestrator/graph.py +++ b/backend/server/orchestrator/graph.py @@ -263,6 +263,7 @@ def _gpu( "verdicts/", "reads/reads.json", "scale/scale.json", + "impl/", ), } diff --git a/backend/server/result/compose.py b/backend/server/result/compose.py index 7b403263e..9a0e8f368 100644 --- a/backend/server/result/compose.py +++ b/backend/server/result/compose.py @@ -8,11 +8,12 @@ import time from dataclasses import dataclass, field from pathlib import Path +from typing import get_args import numpy as np from pydantic import ValidationError -from server.contracts.jobs import CaptureRecord, EventRecord, RunRecord, StageContext, StageResult +from server.contracts.jobs import CaptureRecord, EventRecord, RunRecord, StageContext, StageName, StageResult from server.contracts.result import ( BoxArkitWorld, BoxSceneM, @@ -56,6 +57,7 @@ "S": (0.7874, 1.0033, 0.5588), } _COVERAGE_WORST = ("unobserved", "generated", "partial", "grazing", "observed") +_IMPL_SCHEMA = "hss-stage-impl/v1" _DEFAULT_CONFIG = ResultConfig() _TARGET_CLASSES = ( "garage_door", @@ -101,6 +103,7 @@ def compose_result( epochs: list[str] | None = None, storage: object = None, settings: object = None, + impl_dir: Path | None = None, ) -> ComposeResult: warnings: list[str] = [] if catalogue.source == "fallback": @@ -208,6 +211,7 @@ def compose_result( warnings.append(preview_warning) response = response.model_copy(update={"previewUrl": None}) + stage_implementations, stand_ins = _stage_implementations(impl_dir) internal = { "decision": true_decision, "outcome_kind_internal": "not_eligible" if true_decision == "reject" else kind, @@ -217,6 +221,8 @@ def compose_result( "c2_vetoes": (report or {}).get("c2", {}).get("vetoes") if report else None, "decidability": (report or {}).get("decidability") if report else None, "warnings": warnings, + "stageImplementations": stage_implementations, + "standIns": stand_ins, } # A hidden reject keeps the true kind even though `kind` was rewritten. if true_decision == "reject" and not config.show_reject: @@ -300,6 +306,8 @@ def run(ctx: StageContext) -> StageResult: verdict_dir = Path(ctx.inputs.get("verdicts") or Path(ctx.work_dir) / "verdicts") reads = _stage_input(ctx, "reads", "reads/reads.json") scale = _stage_input(ctx, "scale", "scale/scale.json") + impl_raw = ctx.inputs.get("impl") + impl_dir = Path(impl_raw) if impl_raw else Path(ctx.work_dir) / "impl" 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) @@ -321,6 +329,7 @@ def run(ctx: StageContext) -> StageResult: epochs=epochs, storage=ctx.storage, settings=ctx.settings, + impl_dir=impl_dir if impl_dir.is_dir() else None, ) except ResultError: raise @@ -337,6 +346,37 @@ def run(ctx: StageContext) -> StageResult: return StageResult(artifacts=artifacts, metrics={"events": len(seqs)}, warnings=list(composed.warnings)) +def _stage_implementations(impl_dir: Path | None) -> tuple[dict[str, str], list[str]]: + """Stage -> real | stand-in | unknown, from the impl files the runner publishes. + + The orchestration leaves one ``impl/.json`` per run stage in the result + stage's inputs; ``stand-in`` means the stage ran a module under + server.orchestrator.stubs, so its output is not real measurement work. A stage + with no file is ``unknown`` — as for runs composed before this record existed. + The result stage always reads as ``real``: this function runs inside it. A + file that does not carry the pinned schema stops the composition with a loud + error naming the file, because a mis-published marker would otherwise read as + a trustworthy name. + """ + + implementations = {stage: "unknown" for stage in get_args(StageName)} + if impl_dir is not None and impl_dir.is_dir(): + for path in sorted(impl_dir.glob("*.json")): + try: + doc = json.loads(path.read_text()) + except (OSError, ValueError) as exc: + raise ResultError("impl_invalid", f"{path}: unreadable impl file ({exc})") from exc + if not isinstance(doc, dict) or doc.get("schema") != _IMPL_SCHEMA: + raise ResultError("impl_invalid", f"{path}: schema is not {_IMPL_SCHEMA}") + implementation = doc.get("implementation") + if implementation not in {"real", "stand-in"}: + raise ResultError("impl_invalid", f"{path}: implementation is neither real nor stand-in") + implementations[str(doc.get("stage") or path.stem)] = implementation + implementations["result"] = "real" + stand_ins = sorted(stage for stage, impl in implementations.items() if impl == "stand-in") + return implementations, stand_ins + + def _catalogue(path: Path) -> tuple[Catalogue, bool]: try: return load_catalogue(path), False @@ -901,6 +941,7 @@ def _review_context( "timings": (report or {}).get("timings_s") or {}, "vlm_cost": ((report or {}).get("c3") or {}).get("cost_usd"), "decidability": internal.get("decidability"), + "stand_ins": internal.get("standIns") or [], "photos": photos, "criteria": raw.get("criteria") or [], "walls": _wall_strips(scene, raw), @@ -982,6 +1023,7 @@ def main(argv: list[str] | None = None) -> None: parser.add_argument("--catalogue", type=Path) parser.add_argument("--profile", default=None) parser.add_argument("--show-reject", action="store_true") + parser.add_argument("--impl", type=Path, help="directory of impl/.json markers the orchestration published") args = parser.parse_args(argv) config = load_config() if args.profile: @@ -1006,6 +1048,7 @@ def main(argv: list[str] | None = None) -> None: catalogue=catalogue, config=config, epochs=epochs, + impl_dir=args.impl, ) diff --git a/backend/server/result/review/template.html b/backend/server/result/review/template.html index f62bdb660..85061d598 100644 --- a/backend/server/result/review/template.html +++ b/backend/server/result/review/template.html @@ -10,6 +10,9 @@

Capture {{ capture_id }}

Run {{ run_id }} · device profile {{ profile }}

True decision: {{ true_decision }}

+ {% if stand_ins %} +

Stand-in steps: {{ stand_ins | join(", ") }}. Their output is not from this scan.

+ {% endif %}

Failure code: {{ failure_code or "none" }} · degraded: {{ degraded | join(", ") or "none" }}

{% if scale %}

Scale {{ scale.source }} / {{ scale.status }} · rel bound {{ scale.relBound95 }}

From 02d2ea7857959299a2ea7f82fc6fbbd1486a383d Mon Sep 17 00:00:00 2001 From: Sam Gu <127461594+SamGu-NRX@users.noreply.github.com> Date: Fri, 16 Oct 2026 17:55:54 +0000 Subject: [PATCH 2/3] test(result): pin the stand-in naming on the synthetic fixture All real, two stand-ins, no impl dir, a marker without the pinned schema refused, and result.json byte-identical with and without impl files. The five fail against the pre-change composer. --- backend/server/result/tests/test_stand_ins.py | 119 ++++++++++++++++++ 1 file changed, 119 insertions(+) create mode 100644 backend/server/result/tests/test_stand_ins.py diff --git a/backend/server/result/tests/test_stand_ins.py b/backend/server/result/tests/test_stand_ins.py new file mode 100644 index 000000000..8a5e0c8a4 --- /dev/null +++ b/backend/server/result/tests/test_stand_ins.py @@ -0,0 +1,119 @@ +"""The result names which stages ran as stand-ins. + +The orchestration leaves one impl/.json per run stage in the result +stage's inputs (schema hss-stage-impl/v1). The composer turns those markers +into a stageImplementations record and a sorted standIns list, refuses a +marker that does not carry the schema, and leaves result.json untouched. +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import get_args + +from server.contracts.jobs import CaptureRecord, RunRecord, StageName +from server.result.compose import ResultError, compose_result +from server.result.config import ResultConfig +from server.result.copy import load_catalogue +from server.result.tests.support import DATA + +IMPL_SCHEMA = "hss-stage-impl/v1" + + +def _capture_run(): + capture = json.loads((DATA / "capture.sample.json").read_text()) + epochs = list(capture.pop("epochs")) + return capture, json.loads((DATA / "run.ok.json").read_text()), epochs + + +def _compose(out_dir: Path, impl_dir: Path | None): + capture_raw, run_raw, epochs = _capture_run() + return compose_result( + CaptureRecord.model_validate(capture_raw), + RunRecord.model_validate(run_raw), + DATA / "synthetic", + DATA / "synthetic" / "verdicts", + out_dir, + reads=DATA / "reads.auto_fill.json", + scale=None, + catalogue=load_catalogue(DATA / "prompts.en-US.json"), + config=ResultConfig(catalogue_path=DATA / "prompts.en-US.json"), + epochs=epochs, + impl_dir=impl_dir, + ) + + +def _marker(stage: str, implementation: str) -> dict: + callable_ = "server.orchestrator.stubs" if implementation == "stand-in" else "server" + return {"schema": IMPL_SCHEMA, "stage": stage, "implementation": implementation, "callable": f"{callable_}.{stage}.run"} + + +def _write_impl_dir(root: Path, implementations: dict[str, str]) -> Path: + impl_dir = root / "impl" + impl_dir.mkdir(parents=True) + for stage, implementation in implementations.items(): + (impl_dir / f"{stage}.json").write_text(json.dumps(_marker(stage, implementation)) + "\n") + return impl_dir + + +def _non_result_stages() -> list[str]: + return [stage for stage in get_args(StageName) if stage != "result"] + + +def test_every_stage_real_when_all_markers_say_so(tmp_path: Path): + impl_dir = _write_impl_dir(tmp_path, {stage: "real" for stage in _non_result_stages()}) + composed = _compose(tmp_path / "out", impl_dir) + internal = composed.internal + assert internal["stageImplementations"] == {stage: "real" for stage in get_args(StageName)} + assert internal["standIns"] == [] + + +def test_two_stand_ins_are_named_in_the_record_and_the_page(tmp_path: Path): + implementations = {stage: "real" for stage in _non_result_stages()} + implementations["objects"] = "stand-in" + implementations["reads"] = "stand-in" + impl_dir = _write_impl_dir(tmp_path, implementations) + composed = _compose(tmp_path / "out", impl_dir) + internal = composed.internal + assert internal["standIns"] == ["objects", "reads"] + assert internal["stageImplementations"]["objects"] == "stand-in" + assert internal["stageImplementations"]["reads"] == "stand-in" + # The result stage counts itself as real, marker or no marker. + assert internal["stageImplementations"]["result"] == "real" + page = (tmp_path / "out" / "review.html").read_text() + assert "Stand-in steps: objects, reads. Their output is not from this scan." in page + + +def test_no_impl_dir_leaves_every_other_stage_unknown(tmp_path: Path): + composed = _compose(tmp_path / "out", None) + internal = composed.internal + assert internal["stageImplementations"] == {**{stage: "unknown" for stage in _non_result_stages()}, "result": "real"} + assert internal["standIns"] == [] + # The reviewer page stays silent when nothing ran as a stand-in. + page = (tmp_path / "out" / "review.html").read_text() + assert "Stand-in steps" not in page + + +def test_a_marker_without_the_schema_is_refused_by_name(tmp_path: Path): + impl_dir = tmp_path / "impl" + impl_dir.mkdir() + (impl_dir / "poses.json").write_text(json.dumps({"schema": "hss-stage-impl/v2", "stage": "poses"})) + try: + _compose(tmp_path / "out", impl_dir) + except ResultError as exc: + assert exc.code == "impl_invalid" + assert "poses.json" in str(exc) + else: + raise AssertionError("a marker without the pinned schema must stop the composition") + + +def test_result_json_is_byte_identical_with_and_without_impl_files(tmp_path: Path): + impl_dir = _write_impl_dir(tmp_path, {stage: "real" for stage in _non_result_stages()}) + _compose(tmp_path / "with", impl_dir) + _compose(tmp_path / "without", None) + with_bytes = (tmp_path / "with" / "result.json").read_bytes() + without_bytes = (tmp_path / "without" / "result.json").read_bytes() + assert with_bytes == without_bytes + # The app-facing result carries no impl fields; only the internal record does. + assert "stageImplementations" not in json.loads(with_bytes) From 8057732e68f1245d5b0b2f18dfee52e3a8c3187e Mon Sep 17 00:00:00 2001 From: Sam Gu <127461594+SamGu-NRX@users.noreply.github.com> Date: Fri, 16 Oct 2026 17:56:05 +0000 Subject: [PATCH 3/3] style(result): ruff-format the stand-in tests --- backend/server/result/tests/test_stand_ins.py | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/backend/server/result/tests/test_stand_ins.py b/backend/server/result/tests/test_stand_ins.py index 8a5e0c8a4..0125f8987 100644 --- a/backend/server/result/tests/test_stand_ins.py +++ b/backend/server/result/tests/test_stand_ins.py @@ -46,7 +46,12 @@ def _compose(out_dir: Path, impl_dir: Path | None): def _marker(stage: str, implementation: str) -> dict: callable_ = "server.orchestrator.stubs" if implementation == "stand-in" else "server" - return {"schema": IMPL_SCHEMA, "stage": stage, "implementation": implementation, "callable": f"{callable_}.{stage}.run"} + return { + "schema": IMPL_SCHEMA, + "stage": stage, + "implementation": implementation, + "callable": f"{callable_}.{stage}.run", + } def _write_impl_dir(root: Path, implementations: dict[str, str]) -> Path: @@ -88,7 +93,10 @@ def test_two_stand_ins_are_named_in_the_record_and_the_page(tmp_path: Path): def test_no_impl_dir_leaves_every_other_stage_unknown(tmp_path: Path): composed = _compose(tmp_path / "out", None) internal = composed.internal - assert internal["stageImplementations"] == {**{stage: "unknown" for stage in _non_result_stages()}, "result": "real"} + assert internal["stageImplementations"] == { + **{stage: "unknown" for stage in _non_result_stages()}, + "result": "real", + } assert internal["standIns"] == [] # The reviewer page stays silent when nothing ran as a stand-in. page = (tmp_path / "out" / "review.html").read_text()