From d9154276d75d5240bc3b118cd899a227ffb44056 Mon Sep 17 00:00:00 2001 From: SamGu-NRX Date: Thu, 22 Oct 2026 12:11:03 -0500 Subject: [PATCH] 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) --- 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 24e9ed2b..d38b9ca1 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/.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 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 d8c77cc3..c0b5029d 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 b2ace188..3833bedc 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 c065af86..b0130f9b 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: