From 9e5ec358eb63548c87d3a424057224ce2a7593cc Mon Sep 17 00:00:00 2001
From: Sam Gu <127461594+SamGu-NRX@users.noreply.github.com>
Date: Fri, 16 Oct 2026 17:57:29 +0000
Subject: [PATCH 1/5] feat(orchestrator): per-stage real selection with impl
 markers

HSS_ORCH_REAL_STAGES names stages whose real implementations run even with
stage stubs on; the cpu alias covers every weightless stage and objects is
refused without HSS_OBJECTS_DEVICE=cpu. run_stage publishes impl/<stage>.json
under the pinned hss-stage-impl/v1 contract, and the result stage's inputs
carry impl/. Docs in backend/docs/real-cpu-stages.md.
---
 backend/docs/real-cpu-stages.md               |  70 +++++
 backend/server/orchestrator/graph.py          |   4 +
 backend/server/orchestrator/runner.py         |  44 ++-
 .../server/orchestrator/stage_selection.py    | 119 +++++++++
 .../tests/test_stage_selection.py             | 250 ++++++++++++++++++
 5 files changed, 486 insertions(+), 1 deletion(-)
 create mode 100644 backend/docs/real-cpu-stages.md
 create mode 100644 backend/server/orchestrator/stage_selection.py
 create mode 100644 backend/server/orchestrator/tests/test_stage_selection.py

diff --git a/backend/docs/real-cpu-stages.md b/backend/docs/real-cpu-stages.md
new file mode 100644
index 000000000..d8c77cc39
--- /dev/null
+++ b/backend/docs/real-cpu-stages.md
@@ -0,0 +1,70 @@
+# 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 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` is 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`, `dense`, `coverage`, `scene`,
+`objects_3d`, `placement`, `criteria`, `result`
+
+`objects` is the one stage the alias must not pull in: its real path 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.
+
+## 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=1` — a remote run already runs every stage for real;
+- `SMP_ENVIRONMENT` is anything other than `dev`;
+- `objects` is listed without `HSS_OBJECTS_DEVICE=cpu`.
+
+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.
diff --git a/backend/server/orchestrator/graph.py b/backend/server/orchestrator/graph.py
index f269b53dd..c14277808 100644
--- a/backend/server/orchestrator/graph.py
+++ b/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/",
     ),
 }
 
diff --git a/backend/server/orchestrator/runner.py b/backend/server/orchestrator/runner.py
index f864bc179..fc7d2fdb6 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/<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}"
diff --git a/backend/server/orchestrator/stage_selection.py b/backend/server/orchestrator/stage_selection.py
new file mode 100644
index 000000000..b2ace1888
--- /dev/null
+++ b/backend/server/orchestrator/stage_selection.py
@@ -0,0 +1,119 @@
+"""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/<stage>.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. ``objects`` is the one stage ``cpu`` must not pull in: its real path loads
+OWLv2/SigLIP weights (``server/objects/run.py``), so listing it needs
+``HSS_OBJECTS_DEVICE=cpu`` as well.
+
+The variable is dev-only by construction: it is refused together with
+``HSS_ORCH_MODAL_REMOTE=1`` (a remote run already runs everything real) and
+outside ``SMP_ENVIRONMENT=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.
+"""
+
+from __future__ import annotations
+
+import os
+from collections.abc import Mapping
+
+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"
+
+#: Stages whose real path is deterministic CPU work: no GPU, no local weights,
+#: no paid or networked model call. ``objects`` is the deliberate omission —
+#: its real path loads OWLv2/SigLIP weights, so it never rides the alias.
+CPU_STAGES: tuple[str, ...] = (
+    "validate",
+    "poses",
+    "reads",
+    "scale",
+    "dense",
+    "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)
+_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()
+    if (env.get(REMOTE_VAR) or "").strip() == "1":
+        raise ValueError(
+            f"{ENV_VAR}={raw!r} selects real stages for a local run, but {REMOTE_VAR}=1 already runs "
+            "every stage for real; set one or the other, never both."
+        )
+    environment = (env.get(ENVIRONMENT_VAR) or "dev").strip() or "dev"
+    if environment != "dev":
+        raise ValueError(
+            f"{ENV_VAR}={raw!r} is a dev-only selection; {ENVIRONMENT_VAR}={environment!r} runs the real "
+            "orchestration, where the variable has nothing to select. Unset one of the two."
+        )
+    known = frozenset(STAGE_NAMES)
+    selected: set[str] = set()
+    for part in raw.split(","):
+        name = part.strip()
+        if not name:
+            continue
+        if name == CPU_ALIAS:
+            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/tests/test_stage_selection.py b/backend/server/orchestrator/tests/test_stage_selection.py
new file mode 100644
index 000000000..31bd63a78
--- /dev/null
+++ b/backend/server/orchestrator/tests/test_stage_selection.py
@@ -0,0 +1,250 @@
+"""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 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
+from server.orchestrator.executor import InlineExecutor
+from server.orchestrator.graph import STAGE_NAMES, select_stage_artifacts
+from server.orchestrator.runner import stage_callable
+from server.orchestrator.services import Services, bind, packet_sha256
+
+FIXTURE = Path(__file__).resolve().parents[4] / "data" / "fixtures" / "packets" / "mini-arkit"
+PACKET_RAW = (FIXTURE / "packet.json").read_bytes()
+PACKET = json.loads(PACKET_RAW)
+
+IMPL_SCHEMA = "hss-stage-impl/v1"
+
+
+def _env(**overrides: str) -> dict[str, str]:
+    env = {stage_selection.ENV_VAR: "validate"}
+    env.update(overrides)
+    return env
+
+
+# --- 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({stage_selection.ENV_VAR: " validate , scale ,"}) == frozenset(
+        {"validate", "scale"}
+    )
+
+
+def test_cpu_alias_covers_every_weightless_stage() -> None:
+    selected = stage_selection.parse_real_stages({stage_selection.ENV_VAR: "cpu"})
+    assert selected == frozenset(stage_selection.CPU_STAGES)
+    assert "objects" not in selected
+    assert selected | {"objects"} == frozenset(STAGE_NAMES)
+
+
+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({stage_selection.ENV_VAR: "poses,wat"})
+    assert "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)
+
+
+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)
+
+
+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({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({stage_selection.ENV_VAR: "cpu,objects"})
+
+
+def test_objects_allowed_with_cpu_device() -> None:
+    selected = stage_selection.parse_real_stages(
+        {stage_selection.ENV_VAR: "objects", stage_selection.OBJECTS_DEVICE_VAR: "cpu"}
+    )
+    assert selected == frozenset({"objects"})
+
+
+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 -------------------------------------------------------
+
+
+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"
+
+
+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"
+
+
+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.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:
+    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's hydration carries the whole impl/ directory.
+    selected = select_stage_artifacts("result", run.artifacts)
+    for stage in STAGE_NAMES:
+        assert f"impl/{stage}.json" in selected
+
+
+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.run"
+    assert _impl_body(services, run, "result")["implementation"] == "stand-in"

From 1c73313a1f67e76ea157523686f9f31b04091055 Mon Sep 17 00:00:00 2001
From: Sam Gu <127461594+SamGu-NRX@users.noreply.github.com>
Date: Fri, 16 Oct 2026 18:00:20 +0000
Subject: [PATCH 2/5] test: correct fixture path and placement dotted callable
 in expectations

---
 backend/server/orchestrator/tests/test_stage_selection.py | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/backend/server/orchestrator/tests/test_stage_selection.py b/backend/server/orchestrator/tests/test_stage_selection.py
index 31bd63a78..c065af862 100644
--- a/backend/server/orchestrator/tests/test_stage_selection.py
+++ b/backend/server/orchestrator/tests/test_stage_selection.py
@@ -28,7 +28,7 @@
 from server.orchestrator.runner import stage_callable
 from server.orchestrator.services import Services, bind, packet_sha256
 
-FIXTURE = Path(__file__).resolve().parents[4] / "data" / "fixtures" / "packets" / "mini-arkit"
+FIXTURE = Path(__file__).resolve().parents[3] / "data" / "fixtures" / "packets" / "mini-arkit"
 PACKET_RAW = (FIXTURE / "packet.json").read_bytes()
 PACKET = json.loads(PACKET_RAW)
 
@@ -118,7 +118,7 @@ def test_selected_stage_runs_real_under_stubs(monkeypatch: pytest.MonkeyPatch) -
     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"
+    assert placement.__module__ == "server.placement.adapter"
 
 
 def test_unselected_stages_keep_the_stub(monkeypatch: pytest.MonkeyPatch) -> None:
@@ -246,5 +246,5 @@ def test_unset_selection_records_today_implementations(tmp_path: Path, monkeypat
     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.run"
+    assert _impl_body(services, run, "placement")["callable"] == "server.placement.adapter.run"
     assert _impl_body(services, run, "result")["implementation"] == "stand-in"

From 6829f0975532b8de614881954c5d71bfbc31f789 Mon Sep 17 00:00:00 2001
From: Sam Gu <127461594+SamGu-NRX@users.noreply.github.com>
Date: Fri, 16 Oct 2026 18:23:29 +0000
Subject: [PATCH 3/5] docs: PR description for the per-stage real-selection
 packet

---
 PR.md | 59 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 59 insertions(+)
 create mode 100644 PR.md

diff --git a/PR.md b/PR.md
new file mode 100644
index 000000000..f32640c76
--- /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/<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 dense coverage scene objects_3d placement criteria result`. `objects` is the one deliberate exclusion — its real path loads OWLv2/SigLIP weights — and listing it is refused unless `HSS_OBJECTS_DEVICE=cpu` opts into local weights.
+
+## 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 `HSS_ORCH_MODAL_REMOTE=1` (a remote run already runs everything real) and outside `SMP_ENVIRONMENT=dev` — 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` — recorded in a comment below (baseline on `alpha` carries pre-existing SSR-hygiene and eval-gates failures).
+- 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)

From 098d8fdfc780adbbb66df02a3b9bd956b4b53f06 Mon Sep 17 00:00:00 2001
From: Sam Gu <127461594+SamGu-NRX@users.noreply.github.com>
Date: Fri, 16 Oct 2026 18:23:50 +0000
Subject: [PATCH 4/5] docs: record the verified full-suite result in the PR
 description

---
 PR.md | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/PR.md b/PR.md
index f32640c76..24e9ed2b1 100644
--- a/PR.md
+++ b/PR.md
@@ -42,7 +42,7 @@ N/A — orchestration backend only; no client-visible surface changes.
 - `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` — recorded in a comment below (baseline on `alpha` carries pre-existing SSR-hygiene and eval-gates failures).
+- 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

From d9154276d75d5240bc3b118cd899a227ffb44056 Mon Sep 17 00:00:00 2001
From: SamGu-NRX <sgu07966@gmail.com>
Date: Thu, 22 Oct 2026 12:11:03 -0500
Subject: [PATCH 5/5] fix(orchestrator): stage selection fails closed outside
 an explicit dev

HSS_ORCH_REAL_STAGES treated a missing or blank SMP_ENVIRONMENT as dev,
while Settings reads the same values as prod. An unconfigured production
process could therefore mix real and stand-in stages. Only an exact "dev"
now passes, matching Settings.

The remote refusal compared HSS_ORCH_MODAL_REMOTE to "1" only; load_orch
also accepts true/yes/on in any case. It now uses the same _flag parser.

The cpu alias no longer includes dense, which falls back to model weights
on photo-only packets, and is refused while VISION_EXTERNAL_CALLS is on,
since reads and criteria would then make paid calls.

The journey test now checks the markers the result stage receives in its
work dir, rather than filtering run.artifacts after the run.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 PR.md                                         |   4 +-
 backend/docs/real-cpu-stages.md               |  37 +++--
 .../server/orchestrator/stage_selection.py    |  62 +++++---
 .../tests/test_stage_selection.py             | 133 ++++++++++++++++--
 4 files changed, 190 insertions(+), 46 deletions(-)

diff --git a/PR.md b/PR.md
index 24e9ed2b1..d38b9ca16 100644
--- a/PR.md
+++ b/PR.md
@@ -8,11 +8,11 @@ Until now, `HSS_ORCH_STAGE_STUBS=1` was all or nothing: a local run answered eve
 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 dense coverage scene objects_3d placement criteria result`. `objects` is the one deliberate exclusion — its real path loads OWLv2/SigLIP weights — and listing it is refused unless `HSS_OBJECTS_DEVICE=cpu` opts into local weights.
+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 `HSS_ORCH_MODAL_REMOTE=1` (a remote run already runs everything real) and outside `SMP_ENVIRONMENT=dev` — 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.
+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.
 
diff --git a/backend/docs/real-cpu-stages.md b/backend/docs/real-cpu-stages.md
index d8c77cc39..c0b5029d1 100644
--- a/backend/docs/real-cpu-stages.md
+++ b/backend/docs/real-cpu-stages.md
@@ -7,7 +7,7 @@ scan. `HSS_ORCH_REAL_STAGES` adds the middle setting: it names the stages whose
 exercise one real stage at a time against stand-in upstreams.
 
 ```sh
-export HSS_ORCH_STAGE_STUBS=1 HSS_ORCH_REAL_STAGES=validate
+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
 ```
@@ -16,18 +16,28 @@ 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` is excluded
+## 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`, `dense`, `coverage`, `scene`,
-`objects_3d`, `placement`, `criteria`, `result`
+`validate`, `poses`, `reads`, `scale`, `coverage`, `scene`, `objects_3d`,
+`placement`, `criteria`, `result`
 
-`objects` is the one stage the alias must not pull in: its real path 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.
+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
 
@@ -36,9 +46,14 @@ 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=1` — a remote run already runs every stage for real;
-- `SMP_ENVIRONMENT` is anything other than `dev`;
-- `objects` is listed without `HSS_OBJECTS_DEVICE=cpu`.
+- `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
diff --git a/backend/server/orchestrator/stage_selection.py b/backend/server/orchestrator/stage_selection.py
index b2ace1888..3833bedc7 100644
--- a/backend/server/orchestrator/stage_selection.py
+++ b/backend/server/orchestrator/stage_selection.py
@@ -9,19 +9,27 @@
 
 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. ``objects`` is the one stage ``cpu`` must not pull in: its real path loads
-OWLv2/SigLIP weights (``server/objects/run.py``), so listing it needs
-``HSS_OBJECTS_DEVICE=cpu`` as well.
+call. Two stages stay out of ``cpu``:
 
-The variable is dev-only by construction: it is refused together with
-``HSS_ORCH_MODAL_REMOTE=1`` (a remote run already runs everything real) and
-outside ``SMP_ENVIRONMENT=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.
+- ``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.
+wave. It does reuse that file's boolean parser, so a value means the same thing
+in both places.
 """
 
 from __future__ import annotations
@@ -29,6 +37,7 @@
 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"
@@ -36,16 +45,17 @@
 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. ``objects`` is the deliberate omission —
-#: its real path loads OWLv2/SigLIP weights, so it never rides the alias.
+#: 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",
-    "dense",
     "coverage",
     "scene",
     "objects_3d",
@@ -56,7 +66,7 @@
 
 # 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)
+_CACHE_KEYS = (ENV_VAR, REMOTE_VAR, ENVIRONMENT_VAR, OBJECTS_DEVICE_VAR, VISION_VAR)
 _cache: tuple[tuple[tuple[str, str], ...], frozenset[str]] | None = None
 
 
@@ -67,16 +77,23 @@ def parse_real_stages(environ: Mapping[str, str] | None = None) -> frozenset[str
     raw = (env.get(ENV_VAR) or "").strip()
     if not raw:
         return frozenset()
-    if (env.get(REMOTE_VAR) or "").strip() == "1":
+    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}=1 already runs "
+            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."
         )
-    environment = (env.get(ENVIRONMENT_VAR) or "dev").strip() or "dev"
+    # 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; {ENVIRONMENT_VAR}={environment!r} runs the real "
-            "orchestration, where the variable has nothing to select. Unset one of the two."
+            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()
@@ -85,6 +102,15 @@ def parse_real_stages(environ: Mapping[str, str] | None = None) -> frozenset[str
         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:
diff --git a/backend/server/orchestrator/tests/test_stage_selection.py b/backend/server/orchestrator/tests/test_stage_selection.py
index c065af862..b0130f9b2 100644
--- a/backend/server/orchestrator/tests/test_stage_selection.py
+++ b/backend/server/orchestrator/tests/test_stage_selection.py
@@ -9,6 +9,7 @@
 
 from __future__ import annotations
 
+import functools
 import importlib
 import json
 from pathlib import Path
@@ -22,25 +23,34 @@
 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
+from server.orchestrator.config import OrchConfig, load_orch
 from server.orchestrator.executor import InlineExecutor
-from server.orchestrator.graph import STAGE_NAMES, select_stage_artifacts
+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"}
+    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 -----------------------------------------------------------------
 
 
@@ -51,22 +61,52 @@ def test_unset_selects_nothing() -> None:
 
 def test_single_stage_and_whitespace() -> None:
     assert stage_selection.parse_real_stages(_env()) == frozenset({"validate"})
-    assert stage_selection.parse_real_stages({stage_selection.ENV_VAR: " validate , scale ,"}) == frozenset(
+    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({stage_selection.ENV_VAR: "cpu"})
+    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 selected | {"objects"} == frozenset(STAGE_NAMES)
+    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({stage_selection.ENV_VAR: "poses,wat"})
-    assert "wat" in str(excinfo.value)
+        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:
@@ -75,27 +115,76 @@ def test_remote_refusal_names_both_variables() -> None:
     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({stage_selection.ENV_VAR: "objects"})
+        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({stage_selection.ENV_VAR: "cpu,objects"})
+        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(
-        {stage_selection.ENV_VAR: "objects", stage_selection.OBJECTS_DEVICE_VAR: "cpu"}
+        _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
@@ -109,6 +198,7 @@ def test_selection_cache_follows_the_environment(monkeypatch: pytest.MonkeyPatch
 # --- 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)
@@ -128,6 +218,7 @@ def test_unselected_stages_keep_the_stub(monkeypatch: pytest.MonkeyPatch) -> Non
     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."""
@@ -193,6 +284,7 @@ def _run_journey(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, selection: str
     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):
@@ -211,6 +303,18 @@ def _assert_same_terminal_state(run, capture) -> None:
 
 
 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)
 
@@ -232,10 +336,9 @@ def test_real_validate_journey_marks_mixed_implementations(tmp_path: Path, monke
     assert _impl_body(services, run, "placement")["implementation"] == "real"
     assert _impl_body(services, run, "result")["implementation"] == "stand-in"
 
-    # The result stage's hydration carries the whole impl/ directory.
-    selected = select_stage_artifacts("result", run.artifacts)
-    for stage in STAGE_NAMES:
-        assert f"impl/{stage}.json" in selected
+    # 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: