[obvious-fleet] Backend: the result names which stages ran as stand-ins

backend/server/orchestrator/graph.py

@@ -263,6 +263,7 @@ def _gpu(
         "verdicts/",
         "reads/reads.json",
         "scale/scale.json",
+        "impl/",
     ),
 }
 

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/<stage>.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/<stage>.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,
     )
 
 

backend/server/result/review/template.html

@@ -10,6 +10,9 @@
   <h1>Capture {{ capture_id }}</h1>
   <p>Run {{ run_id }} · device profile {{ profile }}</p>
   <p>True decision: {{ true_decision }}</p>
+  {% if stand_ins %}
+  <p>Stand-in steps: {{ stand_ins | join(", ") }}. Their output is not from this scan.</p>
+  {% endif %}
   <p>Failure code: {{ failure_code or "none" }} · degraded: {{ degraded | join(", ") or "none" }}</p>
   {% if scale %}
   <p>Scale {{ scale.source }} / {{ scale.status }} · rel bound {{ scale.relBound95 }}</p>

backend/server/result/tests/test_stand_ins.py

@@ -0,0 +1,127 @@
+"""The result names which stages ran as stand-ins.
+
+The orchestration leaves one impl/<stage>.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)