diff --git a/PR.md b/PR.md new file mode 100644 index 00000000..d38b9ca1 --- /dev/null +++ b/PR.md @@ -0,0 +1,59 @@ +obv/burst14-bs-stage-selection — packet 3, burst 14 — mode: Autobuild (orchestrator worker thread) + +## What changed + +Until now, `HSS_ORCH_STAGE_STUBS=1` was all or nothing: a local run answered every stage with its stand-in from `server/orchestrator/stubs/.py`, so the result never depended on the scan and no real stage code could be exercised locally without running the full real orchestration (Modal, GPU, credentials). This change adds the middle setting and makes the choice observable: + +1. **`HSS_ORCH_REAL_STAGES`** (new `backend/server/orchestrator/stage_selection.py`) names the stages whose **real** implementations run even with stage stubs on, per stage. A local or dev run can now exercise one real stage at a time against stand-in upstreams — e.g. `HSS_ORCH_REAL_STAGES=validate` runs the real packet validation and records which code served every other stage. +2. **Every stage publishes an implementation marker** at run-relative key `impl/.json` (stored at `captures/{capture_id}/runs/{run_id}/impl/.json`) under the pinned contract `hss-stage-impl/v1`, recording `implementation` (`real` or `stand-in`) and the exact dotted `callable` that executed — read off the resolved callable itself, so a per-run selection override (the capture-placement runner's) is recorded as what it ran, not what a global setting implies. Unset, behavior is byte-for-byte today's: every non-exempt stage records `stand-in`. +3. **The result stage's declared inputs include `impl/`** (`graph.py` `STAGE_INPUTS`), so the composer receives every marker of its run and downstream result consumers can name which implementation produced each input. The pinned contract is shared with `obv/burst14-bs-result-names-stand-ins`; it is defined once in `runner.py` (`IMPL_SCHEMA`) and copied verbatim there. + +The `cpu` alias selects every stage whose real path is deterministic CPU work (no GPU, no model weights, no paid call): `validate poses reads scale coverage scene objects_3d placement criteria result`. `objects` is excluded because its real path loads OWLv2/SigLIP weights, and listing it is refused unless `HSS_OBJECTS_DEVICE=cpu` opts into local weights. `dense` is excluded because photo-only packets fall back to π³/MoGe/DA3 weights; it can still be listed by name. The alias is refused while `VISION_EXTERNAL_CALLS` is on, because `reads` and `criteria` would then call a paid vision model. + +## Why + +A per-run answer to "which code produced this?" needs two halves: the ability to run real stages locally at all (the selection), and a durable per-run record of what actually ran (the markers). The selection is dev-only by construction — refused together with a true `HSS_ORCH_MODAL_REMOTE` in any spelling `load_orch` accepts (a remote run already runs everything real) and unless `SMP_ENVIRONMENT` is exactly `dev`; unset or blank counts as prod, as it does in `Settings` — so production cannot silently run a half-real orchestration. Refusals raise at runner import, which is the startup of every process that can run a stage, and name the offending variable and value. + +The markers land in `impl/` inside the run directory rather than in run metadata so they ride the existing artifact publication, hydration, and result-input machinery with no new transport, and survive in storage next to the artifacts they describe. + +## How it works + +```mermaid +flowchart LR + env["HSS_ORCH_REAL_STAGES\n(comma list or 'cpu')"] --> parse["stage_selection.py\nparse + refusals"] + parse -->|"is_real(stage)"| sc["runner.stage_callable"] + sc -->|"selected / stubs_exempt"| real["real module\nserver."] + sc -->|"not selected"| stub["stand-in\nserver.orchestrator.stubs."] + real --> marker + stub --> marker + marker["run_stage → impl/.json\nhss-stage-impl/v1"] --> result["result stage inputs\ninclude impl/"] +``` + +Caption: the selection only changes which callable `stage_callable` resolves; the marker records what actually ran, and the result stage receives every marker. + +## UI changes + +N/A — orchestration backend only; no client-visible surface changes. + +## Verification + +- New: `backend/server/orchestrator/tests/test_stage_selection.py` — 14 tests: parsing (unset/blank, whitespace, `cpu` alias covering exactly `CPU_STAGES`, alias + `objects` = all 12), refusals (unknown stage, `HSS_ORCH_MODAL_REMOTE=1`, non-dev environment, `objects` without local-weights device — each message names the variable), the selection cache following the environment, callable selection under stubs (selected → real, unselected → stub, `stubs_exempt` placement stays real, stubs-off unaffected), runner-import-time validation of a bad value, and two in-process journey runs on the committed `backend/data/fixtures/packets/mini-arkit` packet: with `validate` selected the run reaches today's terminal state through the real packet validation with mixed impl markers (`validate` real/`server.packet.run`, `poses` stand-in, `placement` real via `stubs_exempt`), and unset the markers record exactly today's implementations; the result stage's hydration carries the whole `impl/` directory. No Modal, GPU, network, or credentials. +- `pytest server/orchestrator/tests` — 195 passed. +- `pytest backend/tests/test_capture_real_downstream.py` — 12 passed (the #292 capture-real-downstream journey, which overrides `stage_callable` per run, is unchanged and its override is what the marker's callable field is designed to record honestly). +- `pytest backend/tests/test_runtime_join_journey.py` — 6 passed. +- `ruff check .` — clean; the new test file is `ruff format`-clean. +- Full `make test` — 4 failed, 1520 passed, 19 skipped, 14 errors in 10:43: the failure set is the branch base's documented baseline exactly — 4 `server/eval/tests/test_gates.py` (SD-96 gates) failures plus 14 collection errors in `tests/test_capture_cold_restart.py`, every name in files this branch does not touch; zero failures attributable to this change. +- Coverage limit: the `objects` real path with local weights is refused-by-default and not exercised end-to-end here (no weights on the dev box); the refusal itself is tested. + +## Risk + +- A real stage fed by stand-in upstreams can meet inputs it was not built for (the stub poses stage produces a single identity camera, for example); the docs (`backend/docs/real-cpu-stages.md`) name this and recommend selecting upstream stages together. This is a dev-only knob and fails at run time, not silently. +- `impl/` artifacts are new keys under the run directory; only the result stage's declared inputs change, and `_publish` callers treat unknown keys uniformly. Runs that fail a stage publish no marker for it, same as any other artifact. +- Pre-existing on `alpha`, untouched here: `ruff format --check` flags `server/orchestrator/tests/test_greptile_followup.py`; `make lint` will report it on the base branch too. + +## Checklist + +- [x] The scope is focused; related changes belong together +- [x] I explained what changed and why +- N/A — no visible changes (backend orchestration only; before/after images not applicable) +- N/A — no motion or interaction changes (no video applicable) diff --git a/backend/docs/real-cpu-stages.md b/backend/docs/real-cpu-stages.md new file mode 100644 index 00000000..2c504e39 --- /dev/null +++ b/backend/docs/real-cpu-stages.md @@ -0,0 +1,90 @@ +# Real CPU stages: `HSS_ORCH_REAL_STAGES` + +With `HSS_ORCH_STAGE_STUBS=1` a local run answers every stage with its stand-in +from `server/orchestrator/stubs/.py`, so the result never depends on the +scan. `HSS_ORCH_REAL_STAGES` adds the middle setting: it names the stages whose +**real** implementations run even with stubs on, so a local or dev run can +exercise one real stage at a time against stand-in upstreams. + +```sh +export SMP_ENVIRONMENT=dev HSS_ORCH_STAGE_STUBS=1 HSS_ORCH_REAL_STAGES=validate +# or every stage whose real path needs no GPU, weights or paid call: +export HSS_ORCH_REAL_STAGES=cpu +``` + +The value is a comma-separated list of stage names, or the alias `cpu`. +Whitespace and a trailing comma are tolerated; stages run real on top of +whatever their upstreams produced, which under stubs are stand-in artifacts. + +## The `cpu` alias, and why `objects` and `dense` are excluded + +`cpu` selects every stage whose real path is deterministic CPU work with no +GPU, no model weights and no paid call: + +`validate`, `poses`, `reads`, `scale`, `coverage`, `scene`, `objects_3d`, +`placement`, `criteria`, `result` + +Two stages stay out: + +- `objects` loads the OWLv2 and SigLIP weights (`server/objects/run.py`), which + a CPU dev box does not carry. Listing `objects` explicitly is refused unless + `HSS_OBJECTS_DEVICE=cpu` is also set to run those weights locally. +- `dense` falls back to π³/MoGe/DA3 weights on a CUDA device whenever the + packet has no LiDAR and no points were injected (`_acquire_cloud` in + `server/dense/run.py`), and fails with `input_missing:weights` without + `HSS_DENSE_WEIGHTS`. List it by name (`cpu,dense`) when the packet or the + machine can serve it. + +`reads` and `criteria` stay in the alias because their real paths make no +vision call while `VISION_EXTERNAL_CALLS` is off, which is the default. The +alias is refused when that variable is on. + +## Refusals + +The variable is read by `server/orchestrator/stage_selection.py`, validated +when the orchestrator runner first loads, and refused with a message that names +the variable and the bad value when: + +- a listed stage name is unknown; +- `HSS_ORCH_MODAL_REMOTE` is true in any spelling `load_orch` accepts (`1`, + `true`, `yes`, `on`, any case): a remote run already runs every stage for + real; +- `SMP_ENVIRONMENT` is anything other than exactly `dev`, including unset or + blank. `Settings` reads those values as `prod`, so the selection does too; +- `objects` is listed without `HSS_OBJECTS_DEVICE=cpu`; +- the `cpu` alias is used while `VISION_EXTERNAL_CALLS` is true (`on`, `1`, + `true`, `yes`). + +A stage selected this way can still fail at run time: a real stage fed by +stand-in artifacts can meet inputs it was not built for (the stub poses stage +produces one identity camera, for example). Select upstream stages together +when a downstream real stage needs their real shape. + +## What the `impl/` artifacts mean + +For every stage a run publishes, `server/orchestrator/runner.py` `run_stage` +also publishes one marker at the run-relative key `impl/.json` — stored +at `captures/{capture_id}/runs/{run_id}/impl/.json` — with exactly: + +```json +{"schema": "hss-stage-impl/v1", "stage": "", "implementation": "real" | "stand-in", "callable": "."} +``` + +- `implementation` is `"stand-in"` when a module under + `server.orchestrator.stubs` ran, `"real"` otherwise (including `placement`, + which is `stubs_exempt` and always real). +- `callable` names what actually executed, read off the resolved callable — + `server.packet.run`, `server.orchestrator.stubs.poses.run` — so a per-run + selection override (the capture-placement runner's) is recorded as what it + ran. +- A marker exists only for stages that succeeded; a failed stage publishes no + artifacts, this one included. + +The result stage's declared inputs include `impl/`, so the composer receives +every marker of its run. The real composer and the stub result stage both read +them through `stage_implementations` in `server/result/compose.py` and write +`stageImplementations` and `standIns` into `result.internal.json`. A marker +whose `stage` is not a pipeline stage or differs from its filename, or whose +`implementation` is not `real` or `stand-in`, fails the result stage with +`impl_invalid`. The contract is shared with +`burst14-bs-result-names-stand-ins`; change it in both places or not at all. diff --git a/backend/server/orchestrator/graph.py b/backend/server/orchestrator/graph.py index a1450782..c1427780 100644 --- a/backend/server/orchestrator/graph.py +++ b/backend/server/orchestrator/graph.py @@ -263,6 +263,9 @@ def _gpu( "verdicts/", "reads/reads.json", "scale/scale.json", + # impl/ carries the hss-stage-impl/v1 markers run_stage publishes: one + # JSON per stage naming the implementation (real or stand-in) and the + # exact callable that served the run. "impl/", ), } diff --git a/backend/server/orchestrator/runner.py b/backend/server/orchestrator/runner.py index f864bc17..fc7d2fdb 100644 --- a/backend/server/orchestrator/runner.py +++ b/backend/server/orchestrator/runner.py @@ -3,10 +3,12 @@ from __future__ import annotations import importlib +import json from pathlib import Path from typing import Any from server.contracts.jobs import StageContext, StageName, StageResult +from server.orchestrator import stage_selection from server.orchestrator.artifacts import _safe_rel, existing_object, local_artifact from server.orchestrator.clock import Clock from server.orchestrator.graph import STAGES @@ -14,6 +16,16 @@ RETRYABLE_TYPES = (RuntimeError, OSError, TimeoutError) +# Pinned contract, shared with burst14-bs-result-names-stand-ins. Copy verbatim: +# every stage's run publishes one marker at impl/.json recording which +# implementation ran. "stand-in" means a module under server.orchestrator.stubs ran. +IMPL_SCHEMA = "hss-stage-impl/v1" + +# Fail fast on a bad HSS_ORCH_REAL_STAGES: every process that can run a stage +# imports this module, so import time is that process's startup. Unset or valid, +# this is one cached dict lookup; invalid, the refusals below raise here. +stage_selection.real_stages() + def is_retryable(exc: BaseException) -> bool: if isinstance(exc, StageSuperseded): @@ -35,7 +47,11 @@ def stage_callable(stage: StageName, *, stubs: bool) -> Any: # A stubs_exempt stage is deterministic CPU work (no GPU, network or credentials); # it runs its real implementation even on the stub path. That is the point: the # local stub journey exercises the real placement solver. - if stubs and not spec.stubs_exempt: + # HSS_ORCH_REAL_STAGES (server/orchestrator/stage_selection.py) then names + # individual stages that also run real under stubs; everything else keeps + # today's behaviour exactly. The signature stays keyword-only in `stubs` so the + # capture-placement runner's per-run selection override (#292) keeps working. + if stubs and not spec.stubs_exempt and not stage_selection.is_real(stage): module = importlib.import_module(f"server.orchestrator.stubs.{stage}") return module.run module = importlib.import_module(f"server.{spec.module}") @@ -58,10 +74,36 @@ def run_stage( if not isinstance(result, StageResult): raise RuntimeError(f"{ctx.stage} returned {type(result).__name__}") stored = _publish(ctx, result) + _publish_impl_marker(ctx, stored, fn) result.artifacts = stored return result +def _publish_impl_marker(ctx: StageContext, stored: dict[str, str], fn: Any) -> None: + """Publish impl/.json: which implementation ran, under the pinned contract. + + The record describes the callable that actually executed — read off `fn` + itself, not re-derived from the selection — so a per-run override (the + capture-placement runner's selection, #292) is recorded as what it ran. + Published only on success, like every other artifact of the stage. + """ + + module_name = getattr(fn, "__module__", "") or "unknown" + function_name = getattr(fn, "__qualname__", "") or getattr(fn, "__name__", "") or "run" + if "<" in function_name: # a local wrapper; name the closest honest dotted path + function_name = getattr(fn, "__name__", "run") + key = f"impl/{ctx.stage}.json" + storage_key = f"captures/{ctx.capture_id}/runs/{ctx.run_id}/{key}" + body = { + "schema": IMPL_SCHEMA, + "stage": ctx.stage, + "implementation": "stand-in" if module_name.startswith("server.orchestrator.stubs.") else "real", + "callable": f"{module_name}.{function_name}", + } + ctx.storage.put_bytes(storage_key, (json.dumps(body, indent=2) + "\n").encode(), "application/json") + stored[key] = storage_key + + def _publish(ctx: StageContext, result: StageResult) -> dict[str, str]: stored: dict[str, str] = {} prefix = f"captures/{ctx.capture_id}/runs/{ctx.run_id}" diff --git a/backend/server/orchestrator/stage_selection.py b/backend/server/orchestrator/stage_selection.py new file mode 100644 index 00000000..3833bedc --- /dev/null +++ b/backend/server/orchestrator/stage_selection.py @@ -0,0 +1,145 @@ +"""Per-stage real-vs-stand-in selection on the dev stub path. + +``HSS_ORCH_STAGE_STUBS=1`` is all or nothing: every stage except the +``stubs_exempt`` placement stage runs ``server/orchestrator/stubs/.py``, +so the answers a local run produces never depend on the scan. This module adds +the middle setting: ``HSS_ORCH_REAL_STAGES`` names the stages whose *real* +implementations run even with stubs on, per stage, so a local run can exercise +one real stage at a time against stand-in upstreams. + +Grammar: a comma-separated list of stage names, or the alias ``cpu``, which +selects every stage whose real path needs no GPU, no model weights and no paid +call. Two stages stay out of ``cpu``: + +- ``objects`` loads OWLv2/SigLIP weights (``server/objects/run.py``), so + listing it needs ``HSS_OBJECTS_DEVICE=cpu`` as well. +- ``dense`` falls back to π³/MoGe/DA3 weights on a CUDA device whenever the + packet carries no LiDAR and no points were injected + (``server/dense/run.py`` ``_acquire_cloud``). It can still be listed by name. + +``reads`` and ``criteria`` call a paid vision model when +``VISION_EXTERNAL_CALLS`` is on, so ``cpu`` is refused in that case. + +The variable is dev-only by construction: it is refused together with a true +``HSS_ORCH_MODAL_REMOTE`` (a remote run already runs everything real) and +unless ``SMP_ENVIRONMENT`` is exactly ``dev``. Refusals raise as early as the +process can see them — :mod:`server.orchestrator.runner` validates at import, +which is the startup of every process that can run a stage. + +This module reads the environment itself and does not extend +``server/orchestrator/config.py``: that file is another packet's file this +wave. It does reuse that file's boolean parser, so a value means the same thing +in both places. +""" + +from __future__ import annotations + +import os +from collections.abc import Mapping + +from server.orchestrator.config import _flag +from server.orchestrator.graph import STAGE_NAMES + +ENV_VAR = "HSS_ORCH_REAL_STAGES" +CPU_ALIAS = "cpu" +REMOTE_VAR = "HSS_ORCH_MODAL_REMOTE" +ENVIRONMENT_VAR = "SMP_ENVIRONMENT" +OBJECTS_DEVICE_VAR = "HSS_OBJECTS_DEVICE" +VISION_VAR = "VISION_EXTERNAL_CALLS" + +#: Stages whose real path is deterministic CPU work: no GPU, no local weights, +#: no paid or networked model call (the last holds while VISION_EXTERNAL_CALLS +#: is off, which the parser enforces for the alias). ``objects`` and ``dense`` +#: are the deliberate omissions; see the module docstring. +CPU_STAGES: tuple[str, ...] = ( + "validate", + "poses", + "reads", + "scale", + "coverage", + "scene", + "objects_3d", + "placement", + "criteria", + "result", +) + +# Cache key: the env vars the parse reads. A cache entry survives only while +# they are unchanged, so tests can monkeypatch the environment freely. +_CACHE_KEYS = (ENV_VAR, REMOTE_VAR, ENVIRONMENT_VAR, OBJECTS_DEVICE_VAR, VISION_VAR) +_cache: tuple[tuple[tuple[str, str], ...], frozenset[str]] | None = None + + +def parse_real_stages(environ: Mapping[str, str] | None = None) -> frozenset[str]: + """Parse and validate ``HSS_ORCH_REAL_STAGES``. Raises ``ValueError`` on refusals.""" + + env = os.environ if environ is None else environ + raw = (env.get(ENV_VAR) or "").strip() + if not raw: + return frozenset() + remote = env.get(REMOTE_VAR, "0") + # load_orch parses this variable with _flag, so every spelling it treats as + # true (1, true, yes, on, any case) is refused here too. + if _flag(remote): + raise ValueError( + f"{ENV_VAR}={raw!r} selects real stages for a local run, but {REMOTE_VAR}={remote!r} already runs " + "every stage for real; set one or the other, never both." + ) + # Fail closed, matching Settings.SMP_ENVIRONMENT (server/common/config.py): + # a missing, blank or unknown value is prod there, so only an exact "dev" is + # dev here. An unconfigured production process therefore refuses a selection. + environment = env.get(ENVIRONMENT_VAR) + if environment != "dev": + shown = "unset" if environment is None else f"{environment!r}" + raise ValueError( + f"{ENV_VAR}={raw!r} is a dev-only selection, but {ENVIRONMENT_VAR} is {shown}, which Settings " + f"reads as prod. Set {ENVIRONMENT_VAR}=dev exactly, or unset {ENV_VAR}." + ) + known = frozenset(STAGE_NAMES) + selected: set[str] = set() + for part in raw.split(","): + name = part.strip() + if not name: + continue + if name == CPU_ALIAS: + vision = env.get(VISION_VAR, "") + # reads switches on for "on"; criteria also for 1/true/yes. _flag + # covers both sets, so the alias never pulls in a paid call. + if _flag(vision): + raise ValueError( + f"{ENV_VAR} uses the alias {CPU_ALIAS!r}, which promises no paid call, but " + f"{VISION_VAR}={vision!r} lets reads and criteria call a vision model. Set " + f"{VISION_VAR}=off, or list the stages by name." + ) + selected.update(CPU_STAGES) + continue + if name not in known: + raise ValueError( + f"{ENV_VAR} names unknown stage {name!r} (full value {raw!r}); known stages are " + f"{', '.join(STAGE_NAMES)} plus the alias {CPU_ALIAS!r}." + ) + if name == "objects" and (env.get(OBJECTS_DEVICE_VAR) or "").strip().lower() != "cpu": + raise ValueError( + f"{ENV_VAR} lists objects, whose real path loads OWLv2/SigLIP weights " + "(server/objects/run.py); it is refused unless " + f"{OBJECTS_DEVICE_VAR}=cpu is also set to run them locally, or drop objects from {ENV_VAR}." + ) + selected.add(name) + return frozenset(selected) + + +def real_stages() -> frozenset[str]: + """The validated selection for the current environment, cached per process.""" + + global _cache + env = os.environ + key = tuple((name, env.get(name, "")) for name in _CACHE_KEYS) + if _cache is None or _cache[0] != key: + _cache = (key, parse_real_stages(env)) + return _cache[1] + + +def is_real(stage: str) -> bool: + """True when the stage's real implementation is selected for this process.""" + + return stage in real_stages() diff --git a/backend/server/orchestrator/stubs/result.py b/backend/server/orchestrator/stubs/result.py index 856898f1..797db900 100644 --- a/backend/server/orchestrator/stubs/result.py +++ b/backend/server/orchestrator/stubs/result.py @@ -2,6 +2,7 @@ from __future__ import annotations +import html import json from pathlib import Path from typing import Literal @@ -9,11 +10,17 @@ from server.contracts.jobs import Profile, StageContext, StageResult from server.contracts.result import MeterBlock, Outcome, ResultResponse from server.orchestrator.stubs._common import json_bytes, sha256, write_files +from server.result.compose import stage_impl_dir, stage_implementations, stand_in_notice _PROFILES: frozenset[str] = frozenset({"A", "B", "C", "S"}) def run(ctx: StageContext) -> StageResult: + # The stand-in record comes from the real composer's reader, so a stub run + # validates its markers and names its stand-ins (this stage among them) the + # same way a real composition does. Read first: a bad marker fails the stage + # before verdict_ready goes out. + implementations, stand_ins = stage_implementations(stage_impl_dir(ctx), result_implementation="stand-in") run_rec = ctx.state.get_run(ctx.run_id) key = run_rec.artifacts["verdicts/verdict.C.json"] verdict_bytes = ctx.storage.get_bytes(key) @@ -43,13 +50,20 @@ def run(ctx: StageContext) -> StageResult: ) body = response.model_dump(mode="json") ResultResponse.model_validate(body) - internal = {"verdict_c_sha256": digest, "decision": decision} + internal = { + "verdict_c_sha256": digest, + "decision": decision, + "stageImplementations": implementations, + "standIns": stand_ins, + } + notice = stand_in_notice(stand_ins) + page = "

stub review

" + (f"

{html.escape(notice)}

" if notice else "") + "" return write_files( ctx, { "result/result.json": json_bytes(body), "result/result.internal.json": json_bytes(internal), - "result/review.html": b"stub review", + "result/review.html": page.encode(), "result/preview.ply": b"ply\nformat ascii 1.0\nelement vertex 0\nend_header\n", }, ) diff --git a/backend/server/orchestrator/tests/test_stage_selection.py b/backend/server/orchestrator/tests/test_stage_selection.py new file mode 100644 index 00000000..b0130f9b --- /dev/null +++ b/backend/server/orchestrator/tests/test_stage_selection.py @@ -0,0 +1,353 @@ +"""HSS_ORCH_REAL_STAGES: per-stage real-vs-stand-in selection and the impl markers. + +Unit tests parse the variable and its refusals. Journey tests drive whole runs +in-process with the committed mini-arkit packet fixture: with stubs on and +``validate`` selected, the run must reach today's terminal state through the +real packet validation, and every stage must publish its impl marker under the +pinned hss-stage-impl/v1 contract. No Modal, GPU, network or credentials. +""" + +from __future__ import annotations + +import functools +import importlib +import json +from pathlib import Path + +import pytest + +from server.common.backends.state_memory import MemoryState +from server.common.backends.storage_local import LocalFsStorage +from server.common.config import Settings +from server.contracts.jobs import CaptureRecord +from server.orchestrator import stage_selection +from server.orchestrator.api import start_run +from server.orchestrator.clock import Clock +from server.orchestrator.config import OrchConfig, load_orch +from server.orchestrator.executor import InlineExecutor +from server.orchestrator.graph import STAGE_NAMES +from server.orchestrator.runner import stage_callable +from server.orchestrator.services import Services, bind, packet_sha256 +from server.orchestrator.stubs import result as stub_result + +FIXTURE = Path(__file__).resolve().parents[3] / "data" / "fixtures" / "packets" / "mini-arkit" +PACKET_RAW = (FIXTURE / "packet.json").read_bytes() +PACKET = json.loads(PACKET_RAW) + +IMPL_SCHEMA = "hss-stage-impl/v1" +VISION_VAR = "VISION_EXTERNAL_CALLS" + + +def _env(**overrides: str) -> dict[str, str]: + env = {stage_selection.ENV_VAR: "validate", stage_selection.ENVIRONMENT_VAR: "dev"} + env.update(overrides) + return env + + +@pytest.fixture +def dev_environment(monkeypatch: pytest.MonkeyPatch) -> None: + """Tests that set HSS_ORCH_REAL_STAGES in os.environ need an explicit dev process.""" + + monkeypatch.setenv(stage_selection.ENVIRONMENT_VAR, "dev") + + +# --- parsing ----------------------------------------------------------------- + + +def test_unset_selects_nothing() -> None: + assert stage_selection.parse_real_stages({}) == frozenset() + assert stage_selection.parse_real_stages({stage_selection.ENV_VAR: " "}) == frozenset() + + +def test_single_stage_and_whitespace() -> None: + assert stage_selection.parse_real_stages(_env()) == frozenset({"validate"}) + assert stage_selection.parse_real_stages(_env(**{stage_selection.ENV_VAR: " validate , scale ,"})) == frozenset( + {"validate", "scale"} + ) + + +def test_cpu_alias_covers_every_weightless_stage() -> None: + selected = stage_selection.parse_real_stages(_env(**{stage_selection.ENV_VAR: "cpu"})) + assert selected == frozenset(stage_selection.CPU_STAGES) + # objects loads OWLv2/SigLIP; dense falls back to π³/MoGe/DA3 on photo-only packets. + assert "objects" not in selected + assert "dense" not in selected + assert selected | {"objects", "dense"} == frozenset(STAGE_NAMES) + + +def test_dense_can_still_be_listed_explicitly() -> None: + assert stage_selection.parse_real_stages(_env(**{stage_selection.ENV_VAR: "cpu,dense"})) == frozenset( + {*stage_selection.CPU_STAGES, "dense"} + ) + + +@pytest.mark.parametrize("value", ["on", "ON", " on ", "1", "true", "yes"]) +def test_cpu_alias_refused_when_vision_calls_are_on(value: str) -> None: + """reads and criteria call a paid vision model when VISION_EXTERNAL_CALLS is on.""" + + with pytest.raises(ValueError, match=VISION_VAR) as excinfo: + stage_selection.parse_real_stages(_env(**{stage_selection.ENV_VAR: "cpu", VISION_VAR: value})) + assert repr(value) in str(excinfo.value) + + +@pytest.mark.parametrize("value", [None, "off", "0", ""]) +def test_cpu_alias_allowed_when_vision_calls_are_off(value: str | None) -> None: + env = _env(**{stage_selection.ENV_VAR: "cpu"}) + if value is not None: + env[VISION_VAR] = value + assert stage_selection.parse_real_stages(env) == frozenset(stage_selection.CPU_STAGES) + + +def test_explicit_reads_with_vision_on_is_the_callers_choice() -> None: + selected = stage_selection.parse_real_stages(_env(**{stage_selection.ENV_VAR: "reads", VISION_VAR: "on"})) + assert selected == frozenset({"reads"}) + + +def test_unknown_stage_names_the_variable_and_value() -> None: + with pytest.raises(ValueError, match=stage_selection.ENV_VAR) as excinfo: + stage_selection.parse_real_stages(_env(**{stage_selection.ENV_VAR: "poses,wat"})) + assert "unknown stage 'wat'" in str(excinfo.value) + + +def test_remote_refusal_names_both_variables() -> None: + with pytest.raises(ValueError, match="HSS_ORCH_MODAL_REMOTE") as excinfo: + stage_selection.parse_real_stages(_env(**{stage_selection.REMOTE_VAR: "1"})) + assert stage_selection.ENV_VAR in str(excinfo.value) + + +@pytest.mark.parametrize("value", ["1", "true", "TRUE", "yes", "on", " on "]) +def test_remote_refusal_accepts_every_spelling_load_orch_treats_as_true(value: str) -> None: + assert load_orch({stage_selection.REMOTE_VAR: value}).modal_remote is True + with pytest.raises(ValueError, match=stage_selection.REMOTE_VAR) as excinfo: + stage_selection.parse_real_stages(_env(**{stage_selection.REMOTE_VAR: value})) + assert repr(value) in str(excinfo.value) + + +@pytest.mark.parametrize("value", ["0", "false", "off", ""]) +def test_remote_false_spellings_do_not_refuse(value: str) -> None: + assert load_orch({stage_selection.REMOTE_VAR: value}).modal_remote is False + assert stage_selection.parse_real_stages(_env(**{stage_selection.REMOTE_VAR: value})) == frozenset({"validate"}) + + +def test_non_dev_environment_refusal_names_the_value() -> None: + with pytest.raises(ValueError, match="production") as excinfo: + stage_selection.parse_real_stages(_env(**{stage_selection.ENVIRONMENT_VAR: "production"})) + assert stage_selection.ENVIRONMENT_VAR in str(excinfo.value) + + +@pytest.mark.parametrize("value", [None, "", " ", "Dev", " dev", "prod"]) +def test_environment_settings_reads_as_prod_is_refused(value: str | None, monkeypatch: pytest.MonkeyPatch) -> None: + """Fail closed: Settings coerces a missing, blank or unknown SMP_ENVIRONMENT to + prod, so the selection must treat exactly those values as prod too.""" + + if value is None: + monkeypatch.delenv(stage_selection.ENVIRONMENT_VAR, raising=False) + else: + monkeypatch.setenv(stage_selection.ENVIRONMENT_VAR, value) + assert Settings().SMP_ENVIRONMENT == "prod" + + env = {stage_selection.ENV_VAR: "validate"} + if value is not None: + env[stage_selection.ENVIRONMENT_VAR] = value + with pytest.raises(ValueError, match=stage_selection.ENVIRONMENT_VAR) as excinfo: + stage_selection.parse_real_stages(env) + assert "prod" in str(excinfo.value) + + +def test_runner_import_refuses_a_selection_without_dev(monkeypatch: pytest.MonkeyPatch) -> None: + """An unconfigured process (no SMP_ENVIRONMENT) is prod and must not start with a selection.""" + + import server.orchestrator.runner as runner_module + + monkeypatch.delenv(stage_selection.ENVIRONMENT_VAR, raising=False) + monkeypatch.setenv(stage_selection.ENV_VAR, "validate") + try: + with pytest.raises(ValueError, match=stage_selection.ENVIRONMENT_VAR): + importlib.reload(runner_module) + finally: + monkeypatch.delenv(stage_selection.ENV_VAR, raising=False) + importlib.reload(runner_module) # restore a fully initialised module + + +def test_objects_refused_without_local_weights_device() -> None: + with pytest.raises(ValueError, match=stage_selection.OBJECTS_DEVICE_VAR) as excinfo: + stage_selection.parse_real_stages(_env(**{stage_selection.ENV_VAR: "objects"})) + assert "weights" in str(excinfo.value) + with pytest.raises(ValueError, match=stage_selection.OBJECTS_DEVICE_VAR): + stage_selection.parse_real_stages(_env(**{stage_selection.ENV_VAR: "cpu,objects"})) + + +def test_objects_allowed_with_cpu_device() -> None: + selected = stage_selection.parse_real_stages( + _env(**{stage_selection.ENV_VAR: "objects", stage_selection.OBJECTS_DEVICE_VAR: "cpu"}) + ) + assert selected == frozenset({"objects"}) + + +@pytest.mark.usefixtures("dev_environment") +def test_selection_cache_follows_the_environment(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv(stage_selection.ENV_VAR, raising=False) + assert stage_selection.is_real("validate") is False + monkeypatch.setenv(stage_selection.ENV_VAR, "validate") + assert stage_selection.is_real("validate") is True + monkeypatch.setenv(stage_selection.ENV_VAR, "scale") + assert stage_selection.is_real("validate") is False + assert stage_selection.is_real("scale") is True + + +# --- callable selection ------------------------------------------------------- + + +@pytest.mark.usefixtures("dev_environment") +def test_selected_stage_runs_real_under_stubs(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv(stage_selection.ENV_VAR, "validate") + validate = stage_callable("validate", stubs=True) + assert validate.__module__ == "server.packet" + # A stage outside the selection keeps today's behaviour exactly. + poses = stage_callable("poses", stubs=True) + assert poses.__module__ == "server.orchestrator.stubs.poses" + # The stubs_exempt stage stays real, as today. + placement = stage_callable("placement", stubs=True) + assert placement.__module__ == "server.placement.adapter" + + +def test_unselected_stages_keep_the_stub(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv(stage_selection.ENV_VAR, raising=False) + assert stage_callable("validate", stubs=True).__module__ == "server.orchestrator.stubs.validate" + # Stubs off runs everything real, selection or not. + assert stage_callable("poses", stubs=False).__module__ == "server.poses" + + +@pytest.mark.usefixtures("dev_environment") +def test_runner_import_rejects_a_bad_selection(monkeypatch: pytest.MonkeyPatch) -> None: + """The refusals land at startup: importing the runner is the first thing a + stage-running process does, so a bad value raises there, not on stage one.""" + + import server.orchestrator.runner as runner_module + + monkeypatch.setenv(stage_selection.ENV_VAR, "not-a-stage") + with pytest.raises(ValueError, match="not-a-stage"): + importlib.reload(runner_module) + monkeypatch.delenv(stage_selection.ENV_VAR, raising=False) + importlib.reload(runner_module) # restore a fully initialised module + + +# --- journey: a whole run under the selection --------------------------------- + + +def _mini_arkit_services(tmp_path: Path) -> tuple[Services, str, str]: + """The conftest fixture's minimal packet fails real validation, so the + journey uses the committed mini-arkit packet with its files in storage.""" + + settings = Settings( + STATE_BACKEND="memory", + STORAGE_BACKEND="local", + VISION_EXTERNAL_CALLS="off", + HARD_STOP_MIN=20, + VOLUME_ROOT=str(tmp_path), + SMP_ENVIRONMENT="dev", + ) + config = OrchConfig(stage_stubs=True, stale_s=180, max_respawns=2, prewarm=True) + storage = LocalFsStorage(tmp_path / "store") + services = Services( + settings=settings, + orch=config, + state=MemoryState(), + storage=storage, + clock=Clock(), + executor=InlineExecutor(config=config, stubs=True), + work_root=tmp_path / "work", + ) + uri = "captures/cap_mini/packet.json" + services.storage.put_bytes(uri, PACKET_RAW, "application/json") + for entry in PACKET["files"]: + services.storage.put_bytes( + f"captures/cap_mini/{entry['path']}", (FIXTURE / entry["path"]).read_bytes(), "application/octet-stream" + ) + services.state.put_capture( + CaptureRecord( + capture_id="cap_mini", + packet_id=PACKET["packetId"], + status="awaiting_files", + tier="arkit", + flow="guided", + device_model="iPhone", + owner_principal="test", + created_at=services.clock.iso(), + finalize_by=services.clock.iso(), + ) + ) + return services, uri, packet_sha256(PACKET_RAW) + + +def _run_journey(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, selection: str | None): + if selection is None: + monkeypatch.delenv(stage_selection.ENV_VAR, raising=False) + else: + monkeypatch.setenv(stage_selection.ENVIRONMENT_VAR, "dev") + monkeypatch.setenv(stage_selection.ENV_VAR, selection) + services, uri, digest = _mini_arkit_services(tmp_path) + with bind(services): + run = start_run("cap_mini", digest, uri) + capture = services.state.get_capture("cap_mini") + return services, run, capture + + +def _impl_body(services: Services, run, stage: str) -> dict: + return json.loads(services.storage.get_bytes(run.artifacts[f"impl/{stage}.json"])) + + +def _assert_same_terminal_state(run, capture) -> None: + assert run.status == "done" + assert capture.status in {"complete", "needs_views", "manual_review"} + + +def test_real_validate_journey_marks_mixed_implementations(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + # Record the markers the result stage actually receives in its work dir, so + # the test covers delivery through hydration and not only publication. + # functools.wraps keeps the stub's module and name, which the marker reads. + stub_result_run = stub_result.run + delivered: list[set[str]] = [] + + @functools.wraps(stub_result_run) + def result_run(ctx): + delivered.append({path.stem for path in (ctx.work_dir / "impl").glob("*.json")}) + return stub_result_run(ctx) + + monkeypatch.setattr(stub_result, "run", result_run) + services, run, capture = _run_journey(tmp_path, monkeypatch, "validate") + _assert_same_terminal_state(run, capture) + + # Every stage that ran left one marker under the pinned contract. + for stage in STAGE_NAMES: + body = _impl_body(services, run, stage) + assert body["schema"] == IMPL_SCHEMA + assert body["stage"] == stage + assert _impl_body(services, run, "validate") == { + "schema": IMPL_SCHEMA, + "stage": "validate", + "implementation": "real", + "callable": "server.packet.run", + } + poses_body = _impl_body(services, run, "poses") + assert poses_body["implementation"] == "stand-in" + assert poses_body["callable"] == "server.orchestrator.stubs.poses.run" + # placement is stubs_exempt: real even on the stub path, as today. + assert _impl_body(services, run, "placement")["implementation"] == "real" + assert _impl_body(services, run, "result")["implementation"] == "stand-in" + + # The result stage received every upstream marker, and not its own: impl/result.json + # is published only after the result callable returns. + assert delivered == [set(STAGE_NAMES) - {"result"}] + + +def test_unset_selection_records_today_implementations(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + services, run, capture = _run_journey(tmp_path, monkeypatch, None) + _assert_same_terminal_state(run, capture) + + # Today: every stage except stubs_exempt placement runs a stand-in. + assert _impl_body(services, run, "validate")["implementation"] == "stand-in" + assert _impl_body(services, run, "poses")["implementation"] == "stand-in" + assert _impl_body(services, run, "placement")["implementation"] == "real" + assert _impl_body(services, run, "placement")["callable"] == "server.placement.adapter.run" + assert _impl_body(services, run, "result")["implementation"] == "stand-in" diff --git a/backend/server/orchestrator/tests/test_stub_result_stand_ins.py b/backend/server/orchestrator/tests/test_stub_result_stand_ins.py new file mode 100644 index 00000000..3014f465 --- /dev/null +++ b/backend/server/orchestrator/tests/test_stub_result_stand_ins.py @@ -0,0 +1,71 @@ +"""A run on the stub path names its stand-ins in the result it publishes. + +With HSS_ORCH_STAGE_STUBS on, the result stage itself is the stub in +server/orchestrator/stubs/result.py, so the stand-in record has to come out of +that stub too: the runs whose stages are stand-ins are exactly the runs that +never reach the real composer. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from server.orchestrator import runner +from server.orchestrator.api import start_run +from server.orchestrator.graph import STAGE_NAMES +from server.orchestrator.services import bind +from server.orchestrator.tests.conftest import build_services + + +def test_stub_run_names_every_stand_in_including_the_result(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("HSS_ORCH_REAL_STAGES", raising=False) + services, uri, digest = build_services(tmp_path, "memory") + with bind(services): + run = start_run("cap_test", digest, uri) + assert run.status == "done" + + internal = json.loads(services.storage.get_bytes(run.artifacts["result/result.internal.json"])) + # placement is stubs_exempt, so it is the one stage that ran for real. + expected = {stage: "stand-in" for stage in STAGE_NAMES} + expected["placement"] = "real" + assert internal["stageImplementations"] == expected + stand_ins = sorted(stage for stage in STAGE_NAMES if stage != "placement") + assert internal["standIns"] == stand_ins + + page = services.storage.get_bytes(run.artifacts["result/review.html"]).decode() + assert f"Stand-in steps: {', '.join(stand_ins)}. Their output is not from this scan." in page + + # The app-facing result stays as the stub always wrote it. + assert "standIns" not in json.loads(services.storage.get_bytes(run.artifacts["result/result.json"])) + + +def test_stub_result_refuses_a_marker_that_names_another_stage(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + # objects.json claims to describe reads. The stub result must stop on it + # before sending verdict_ready, as the real composer would. + publish = runner._publish_impl_marker + + def mislabel(ctx, stored, fn): + publish(ctx, stored, fn) + if ctx.stage == "objects": + body = {"schema": "hss-stage-impl/v1", "stage": "reads", "implementation": "real", "callable": "x.run"} + ctx.storage.put_bytes(stored["impl/objects.json"], json.dumps(body).encode(), "application/json") + + monkeypatch.setattr(runner, "_publish_impl_marker", mislabel) + monkeypatch.delenv("HSS_ORCH_REAL_STAGES", raising=False) + services, uri, digest = build_services(tmp_path, "memory") + with bind(services): + run = start_run("cap_test", digest, uri) + + assert run.status == "failed" + assert "result/result.internal.json" not in run.artifacts + events = services.state.events_after("cap_test", 0) + assert "verdict_ready" not in {event.type for event in events} + result_failures = [ + event.data + for event in events + if event.type == "stage" and event.data.get("stage") == "result" and event.data.get("status") == "failed" + ] + assert result_failures and all(data.get("errorCode") == "impl_invalid" for data in result_failures) diff --git a/backend/server/result/compose.py b/backend/server/result/compose.py index 9a0e8f36..1d94831a 100644 --- a/backend/server/result/compose.py +++ b/backend/server/result/compose.py @@ -8,7 +8,7 @@ import time from dataclasses import dataclass, field from pathlib import Path -from typing import get_args +from typing import Literal, get_args import numpy as np from pydantic import ValidationError @@ -58,6 +58,7 @@ } _COVERAGE_WORST = ("unobserved", "generated", "partial", "grazing", "observed") _IMPL_SCHEMA = "hss-stage-impl/v1" +Implementation = Literal["real", "stand-in"] _DEFAULT_CONFIG = ResultConfig() _TARGET_CLASSES = ( "garage_door", @@ -211,7 +212,7 @@ def compose_result( warnings.append(preview_warning) response = response.model_copy(update={"previewUrl": None}) - stage_implementations, stand_ins = _stage_implementations(impl_dir) + implementations, stand_ins = stage_implementations(impl_dir) internal = { "decision": true_decision, "outcome_kind_internal": "not_eligible" if true_decision == "reject" else kind, @@ -221,7 +222,7 @@ 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, + "stageImplementations": implementations, "standIns": stand_ins, } # A hidden reject keeps the true kind even though `kind` was rewritten. @@ -306,8 +307,6 @@ 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) @@ -329,7 +328,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, + impl_dir=stage_impl_dir(ctx), ) except ResultError: raise @@ -346,20 +345,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]]: +def stage_impl_dir(ctx: StageContext) -> Path | None: + """The directory of hydrated ``impl/`` markers for this result stage, or None if there is none.""" + + raw = ctx.inputs.get("impl") + impl_dir = Path(raw) if raw else Path(ctx.work_dir) / "impl" + return impl_dir if impl_dir.is_dir() else None + + +def stage_implementations( + impl_dir: Path | None, *, result_implementation: Implementation = "real" +) -> 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. + + The result stage's own entry is ``result_implementation``, supplied by the + callable that is composing: its marker is published only after it returns, and + a recomposition finds the previous pass's marker instead of its own. + + A marker stops the composition with a loud error naming the file unless it + carries the pinned schema, a known stage that matches its filename, and an + implementation of real or stand-in. Without the filename check, ``objects.json`` + naming ``reads`` as a stand-in could be overwritten by ``reads.json`` naming it + real, and the stand-in would vanish from the record. """ - implementations = {stage: "unknown" for stage in get_args(StageName)} + known = get_args(StageName) + implementations = {stage: "unknown" for stage in known} if impl_dir is not None and impl_dir.is_dir(): for path in sorted(impl_dir.glob("*.json")): try: @@ -368,15 +384,28 @@ def _stage_implementations(impl_dir: Path | None) -> tuple[dict[str, str], list[ 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}") + stage = doc.get("stage") + if not isinstance(stage, str) or stage not in known: + raise ResultError("impl_invalid", f"{path}: stage {stage!r} is not a pipeline stage") + if stage != path.stem: + raise ResultError("impl_invalid", f"{path}: names stage {stage!r} but the file is {path.stem!r}") 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" + if not isinstance(implementation, str) or implementation not in get_args(Implementation): + raise ResultError("impl_invalid", f"{path}: implementation {implementation!r} is not real or stand-in") + implementations[stage] = implementation + implementations["result"] = result_implementation stand_ins = sorted(stage for stage, impl in implementations.items() if impl == "stand-in") return implementations, stand_ins +def stand_in_notice(stand_ins: list[str]) -> str | None: + """The reviewer-page sentence naming the stand-in stages, or None when there are none.""" + + if not stand_ins: + return None + return f"Stand-in steps: {', '.join(stand_ins)}. Their output is not from this scan." + + def _catalogue(path: Path) -> tuple[Catalogue, bool]: try: return load_catalogue(path), False @@ -941,7 +970,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 [], + "stand_in_notice": stand_in_notice(internal.get("standIns") or []), "photos": photos, "criteria": raw.get("criteria") or [], "walls": _wall_strips(scene, raw), diff --git a/backend/server/result/review/template.html b/backend/server/result/review/template.html index 85061d59..004a8ead 100644 --- a/backend/server/result/review/template.html +++ b/backend/server/result/review/template.html @@ -10,8 +10,8 @@

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.

+ {% if stand_in_notice %} +

{{ stand_in_notice }}

{% endif %}

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

{% if scale %} diff --git a/backend/server/result/tests/test_stand_ins.py b/backend/server/result/tests/test_stand_ins.py index 0125f898..bbf5ee3a 100644 --- a/backend/server/result/tests/test_stand_ins.py +++ b/backend/server/result/tests/test_stand_ins.py @@ -12,6 +12,8 @@ from pathlib import Path from typing import get_args +import pytest + from server.contracts.jobs import CaptureRecord, RunRecord, StageName from server.result.compose import ResultError, compose_result from server.result.config import ResultConfig @@ -116,6 +118,67 @@ def test_a_marker_without_the_schema_is_refused_by_name(tmp_path: Path): raise AssertionError("a marker without the pinned schema must stop the composition") +def _assert_refused(tmp_path: Path, impl_dir: Path, filename: str) -> None: + with pytest.raises(ResultError) as caught: + _compose(tmp_path / "out", impl_dir) + assert caught.value.code == "impl_invalid" + assert filename in str(caught.value) + + +def test_a_marker_naming_another_stage_cannot_hide_a_stand_in(tmp_path: Path): + # objects.json claims reads ran as a stand-in; reads.json then says real. + # Before the check, the later file silently won and the stand-in vanished. + impl_dir = _write_impl_dir(tmp_path, {"reads": "real"}) + (impl_dir / "objects.json").write_text(json.dumps(_marker("reads", "stand-in"))) + _assert_refused(tmp_path, impl_dir, "objects.json") + + +@pytest.mark.parametrize("stage", ["objcts", None, 7, "result-extra"]) +def test_a_marker_with_an_unknown_or_missing_stage_is_refused(tmp_path: Path, stage): + impl_dir = tmp_path / "impl" + impl_dir.mkdir() + body = {"schema": IMPL_SCHEMA, "implementation": "stand-in", "callable": "server.orchestrator.stubs.x.run"} + if stage is not None: + body["stage"] = stage + name = f"{stage}.json" if isinstance(stage, str) else "objects.json" + (impl_dir / name).write_text(json.dumps(body)) + _assert_refused(tmp_path, impl_dir, name) + + +@pytest.mark.parametrize("implementation", [[], {}, ["stand-in"], 1, None]) +def test_a_non_string_implementation_is_refused_by_name(tmp_path: Path, implementation): + impl_dir = tmp_path / "impl" + impl_dir.mkdir() + body = {**_marker("poses", "real"), "implementation": implementation} + (impl_dir / "poses.json").write_text(json.dumps(body)) + _assert_refused(tmp_path, impl_dir, "poses.json") + + +def test_a_recomposition_reads_its_own_earlier_marker_as_this_run(tmp_path: Path): + # A late correction re-runs the result stage with impl/result.json from the + # first pass already in its inputs. The record describes the callable that is + # composing now, whatever the earlier marker said. + implementations = {stage: "real" for stage in _non_result_stages()} + implementations["result"] = "stand-in" + impl_dir = _write_impl_dir(tmp_path, implementations) + internal = _compose(tmp_path / "out", impl_dir).internal + assert internal["stageImplementations"]["result"] == "real" + assert internal["standIns"] == [] + + +def test_the_stub_result_shares_the_record_and_names_itself(tmp_path: Path): + from server.result.compose import stage_implementations, stand_in_notice + + implementations = {stage: "stand-in" for stage in _non_result_stages()} + implementations["placement"] = "real" + impl_dir = _write_impl_dir(tmp_path, implementations) + record, stand_ins = stage_implementations(impl_dir, result_implementation="stand-in") + assert record == {**implementations, "result": "stand-in"} + assert stand_ins == sorted(stage for stage in get_args(StageName) if stage != "placement") + assert stand_in_notice(stand_ins).startswith("Stand-in steps: coverage, criteria, dense,") + assert stand_in_notice([]) is None + + 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)