[obvious-fleet] Backend: per-stage real selection with implementation markers

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/<stage>.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/<stage>.json` (stored at `captures/{capture_id}/runs/{run_id}/impl/<stage>.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.<stage>"]
+    sc -->|"not selected"| stub["stand-in\nserver.orchestrator.stubs.<stage>"]
+    real --> marker
+    stub --> marker
+    marker["run_stage → impl/<stage>.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)

backend/docs/real-cpu-stages.md

@@ -0,0 +1,85 @@
+# 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/<stage>.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/<stage>.json` — stored
+at `captures/{capture_id}/runs/{run_id}/impl/<stage>.json` — with exactly:
+
+```json
+{"schema": "hss-stage-impl/v1", "stage": "<stage name>", "implementation": "real" | "stand-in", "callable": "<dotted module>.<function>"}
+```
+
+- `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 contract is shared with
+`burst14-bs-result-names-stand-ins`; change it in both places or not at all.

backend/server/orchestrator/graph.py

@@ -263,6 +263,10 @@ 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/",
     ),
 }
 

backend/server/orchestrator/runner.py

@@ -3,17 +3,29 @@
 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
 from server.orchestrator.heartbeat import Heartbeat, StageSuperseded
 
 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/<stage>.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/<stage>.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}"