From 8338f0f90b00b04367d4d32e9af6471e78a4608b Mon Sep 17 00:00:00 2001 From: xmap <16776958+xmap@users.noreply.github.com> Date: Wed, 9 Sep 2026 06:55:48 -0500 Subject: [PATCH 1/6] Let a compute parameter reference a capture or a steering advice RecipeComputeStep.parameters copied verbatim through the Recipe aggregate's own event-payload wire, unlike its sibling RecipeActionStep which already resolves BindingRef per-value. Left as-is, a CaptureRef or SteeringRef written into a compute parameter would round-trip back out of Postgres as a bare {"__capture__": ...} dict instead of the sentinel object: silent corruption, and upstream of anything Operation BC does. Also fixes a latent bug in validate_capture_refs: folding a compute step's own capture_name declaration and its parameters' reference check into one if/elif let a step that both declares and forward-references skip its own consume check. Declare and consume now run independently, consume first, mirroring validate_output_refs's existing ordering. First slice of letting a compute step (e.g. a tomography reconstruction) read a measured value or a steering brain's advice into its own parameters, the way a setpoint already can. Co-Authored-By: Claude Sonnet 5 --- .../src/cora/recipe/aggregates/recipe/body.py | 49 ++++++++--- .../api/tests/unit/recipe/test_recipe_body.py | 87 +++++++++++++++++++ 2 files changed, 124 insertions(+), 12 deletions(-) diff --git a/apps/api/src/cora/recipe/aggregates/recipe/body.py b/apps/api/src/cora/recipe/aggregates/recipe/body.py index 4a01bbc4521..63d1b9e370a 100644 --- a/apps/api/src/cora/recipe/aggregates/recipe/body.py +++ b/apps/api/src/cora/recipe/aggregates/recipe/body.py @@ -214,10 +214,18 @@ class RecipeComputeStep: The recipe-template twin of the Conductor's `ComputeStep`. `output_uri` selects the result arm, mirroring the Conductor: SET means the FILE arm (the job writes an artifact; the conduct surfaces an `ArtifactRef`), None - means the VALUE arm (the conduct surfaces a `Measurement`). The `command` - argv + `parameters` are LITERAL (no `BindingRef` on a compute step yet); - binding a compute parameter is a deferred widening (the first deployment - that needs an operator-tunable compute parameter fires it). + means the VALUE arm (the conduct surfaces a `Measurement`). `command` is + LITERAL (no `BindingRef` on the argv). + + `parameters` values may individually be a literal, a `CaptureRef`, or a + `SteeringRef`: the compute-branch sibling of `RecipeSetpointStep.value`. + A `CaptureRef` value rides through expansion unresolved and the Conductor + resolves it at execute time against a value an earlier step captured; a + `SteeringRef` value is the same except the value is loop-seeded by the + decide loop, letting a brain tune a compute parameter directly rather + than only a motor position. `BindingRef` is NOT supported in `parameters` + (operator-tunable, define-time compute-parameter binding is a separate, + still-deferred widening; the first deployment that needs it fires it). `input_uris` elements are each a literal URI (an authored well-known path an acquisition action body wrote) OR an `OutputRef` naming an EARLIER @@ -247,7 +255,7 @@ class RecipeComputeStep: command: tuple[str, ...] input_uris: tuple[str | OutputRef, ...] = () output_uri: str | None = None - parameters: Mapping[str, Any] = field(default_factory=dict[str, Any]) + parameters: Mapping[str, Any | CaptureRef | SteeringRef] = field(default_factory=dict[str, Any]) capture_name: str | None = None output_ref_name: str | None = None @@ -393,7 +401,7 @@ def _step_to_wire(step: RecipeStep) -> dict[str, Any]: "command": list(step.command), "input_uris": [_input_uri_to_wire(u) for u in step.input_uris], "output_uri": step.output_uri, - "parameters": dict(step.parameters), + "parameters": {key: _value_to_wire(val) for key, val in step.parameters.items()}, "capture_name": step.capture_name, "output_ref_name": step.output_ref_name, } @@ -432,7 +440,9 @@ def _step_from_wire(payload: dict[str, Any]) -> RecipeStep: command=tuple(payload["command"]), input_uris=tuple(_input_uri_from_wire(u) for u in payload.get("input_uris", ())), output_uri=payload.get("output_uri"), - parameters=dict(payload.get("parameters", {})), + parameters={ + key: _value_from_wire(val) for key, val in payload.get("parameters", {}).items() + }, capture_name=payload.get("capture_name"), output_ref_name=payload.get("output_ref_name"), ) @@ -560,19 +570,34 @@ def validate_capture_refs(steps: tuple[RecipeStep, ...]) -> None: `RecipeComputeStep` whose `capture_name` is not None (slice 6c: a compute step deposits its produced value into a slot). A name declared twice by EITHER kind raises `DuplicateRecipeCaptureError` (cross-kind duplicates - included). A `CaptureRef` in a later `RecipeSetpointStep` value must - reference an already-declared name (forward / missing -> - `UnboundRecipeCaptureError`). + included). A `CaptureRef` in a later `RecipeSetpointStep` value, or in + any `RecipeComputeStep.parameters` value, must reference an + already-declared name (forward / missing -> `UnboundRecipeCaptureError`). + `SteeringRef` values are exempt (see its own docstring): only `CaptureRef` + is checked. + + The consume check and the declare step are INDEPENDENT per step, + consume running first: a `RecipeComputeStep` both consumes + (`parameters` values) and may declare (`capture_name`) in the same step, + and folding both into one `if`/`elif` would let a step's own declaration + branch swallow its own consume check, silently skipping the reference + check whenever `capture_name` is also set. Running consume unconditionally + before declare mirrors `validate_output_refs`'s explicit "consume check + runs BEFORE this step's own declaration" ordering, which also rejects a + step referencing its own not-yet-produced output. """ declared: set[str] = set() for step in steps: + if isinstance(step, RecipeSetpointStep): + _check_capture_ref(step.value, declared) + elif isinstance(step, RecipeComputeStep): + for value in step.parameters.values(): + _check_capture_ref(value, declared) declared_name = _declared_capture_name(step) if declared_name is not None: if declared_name in declared: raise DuplicateRecipeCaptureError(declared_name) declared.add(declared_name) - elif isinstance(step, RecipeSetpointStep): - _check_capture_ref(step.value, declared) class UnboundRecipeOutputError(Exception): diff --git a/apps/api/tests/unit/recipe/test_recipe_body.py b/apps/api/tests/unit/recipe/test_recipe_body.py index c0a2820ef15..c6799382d0e 100644 --- a/apps/api/tests/unit/recipe/test_recipe_body.py +++ b/apps/api/tests/unit/recipe/test_recipe_body.py @@ -269,6 +269,65 @@ def test_validate_capture_refs_compute_none_capture_name_declares_nothing() -> N validate_capture_refs(steps) # does not raise +@pytest.mark.unit +def test_validate_capture_refs_compute_step_parameter_accepts_earlier_declared_name() -> None: + """A CaptureRef nested inside a compute step's parameters resolves like a setpoint value.""" + steps = ( + RecipeCaptureStep(address="dev:sample:x", capture_name="offset"), + RecipeComputeStep( + command=("tomopy", "recon"), + parameters={"rotation_center": CaptureRef("offset")}, + ), + ) + validate_capture_refs(steps) # does not raise + + +@pytest.mark.unit +def test_validate_capture_refs_rejects_forward_ref_in_compute_parameter() -> None: + """A CaptureRef in parameters before the declaring step is a forward ref.""" + steps = ( + RecipeComputeStep( + command=("tomopy", "recon"), + parameters={"rotation_center": CaptureRef("offset")}, + ), + RecipeCaptureStep(address="dev:sample:x", capture_name="offset"), + ) + with pytest.raises(UnboundRecipeCaptureError, match="offset"): + validate_capture_refs(steps) + + +@pytest.mark.unit +def test_validate_capture_refs_exempts_steering_ref_in_compute_parameter() -> None: + """A SteeringRef in parameters is exempt: the decide loop is the producer, not a step.""" + steps = ( + RecipeComputeStep( + command=("tomopy", "recon"), + parameters={"seed": SteeringRef("focus")}, + ), + ) + validate_capture_refs(steps) # does not raise + + +@pytest.mark.unit +def test_validate_capture_refs_checks_compute_parameters_even_when_step_also_declares() -> None: + """A compute step that both declares capture_name and consumes a forward ref must still fail. + + Regression pin: folding the consume check into the declare branch's + if/elif would let this step's own `capture_name` declaration swallow its + own parameters check, silently passing a step that references a name + nothing declared yet. + """ + steps = ( + RecipeComputeStep( + command=("tomopy", "recon"), + parameters={"rotation_center": CaptureRef("offset")}, + capture_name="quality", + ), + ) + with pytest.raises(UnboundRecipeCaptureError, match="offset"): + validate_capture_refs(steps) + + @pytest.mark.unit def test_validate_output_refs_accepts_clean_linear_chain() -> None: """Chain pr -> norm(pr) -> recon(norm): every OutputRef references an earlier declarer.""" @@ -392,6 +451,34 @@ def test_to_dict_from_dict_roundtrip_preserves_compute_step_none_capture_name() assert steps_from_dict(steps_to_dict(steps)) == steps +@pytest.mark.unit +def test_to_dict_from_dict_roundtrip_preserves_compute_step_parameter_capture_ref() -> None: + """A CaptureRef nested in parameters must survive the Recipe event-payload round trip. + + Before this fix parameters were copied verbatim (`dict(step.parameters)`), + so a CaptureRef would round-trip back as a bare {"__capture__": name} dict + instead of a CaptureRef object: silent corruption, not a loud failure. + """ + steps = ( + RecipeComputeStep( + command=("tomopy", "recon"), + parameters={"algorithm": "sirt", "rotation_center": CaptureRef("offset")}, + ), + ) + assert steps_from_dict(steps_to_dict(steps)) == steps + + +@pytest.mark.unit +def test_to_dict_from_dict_roundtrip_preserves_compute_step_parameter_steering_ref() -> None: + steps = ( + RecipeComputeStep( + command=("tomopy", "recon"), + parameters={"seed": SteeringRef("focus")}, + ), + ) + assert steps_from_dict(steps_to_dict(steps)) == steps + + @pytest.mark.unit def test_to_dict_from_dict_roundtrip_preserves_action_step_with_mixed_params() -> None: steps = ( From 55d7335ac9c77190cc80f2844bd0a4b491f69a80 Mon Sep 17 00:00:00 2001 From: xmap <16776958+xmap@users.noreply.github.com> Date: Wed, 9 Sep 2026 07:01:13 -0500 Subject: [PATCH 2/6] Hash a compute parameter's capture or steering reference, not just its literal steps_to_wire's ComputeStep arm copied parameters verbatim, which would crash canonical_json_bytes the moment one carried a CaptureRef or SteeringRef (no default= to fall back on). Adds a per-value wire encoder mirroring the existing input_uris/OutputRef element encoder and the SetpointStep.value sentinel shape, so a ref-bearing parameter hashes deterministically instead of crashing register_procedure_from_recipe. expand() itself needed no functional change: a shallow dict copy already preserves a CaptureRef/SteeringRef object unchanged, since resolve_value (which only ever substitutes a BindingRef) was never called on parameters. Only the stale comment claiming parameters were purely literal needed correcting. Second slice of letting a compute step read a measured value or a steering brain's advice into its own parameters. Co-Authored-By: Claude Sonnet 5 --- .../operation/_recipe_expansion/_expand.py | 43 +++++++-- .../test_recipe_expansion_capture.py | 88 +++++++++++++++++-- 2 files changed, 119 insertions(+), 12 deletions(-) diff --git a/apps/api/src/cora/operation/_recipe_expansion/_expand.py b/apps/api/src/cora/operation/_recipe_expansion/_expand.py index 00a254f0d7e..703aed06b9b 100644 --- a/apps/api/src/cora/operation/_recipe_expansion/_expand.py +++ b/apps/api/src/cora/operation/_recipe_expansion/_expand.py @@ -79,6 +79,31 @@ def _input_uri_to_wire(uri: str | OutputRef) -> Any: return uri +def _parameter_value_to_wire(value: Any) -> Any: + """Serialize ONE ComputeStep `parameters` value to the hash wire form. + + Mirrors `_input_uri_to_wire`'s element-wise shape, for the same reason: + `canonical_json_bytes` has no `default=`, so a raw `CaptureRef` / + `SteeringRef` value anywhere in the mapping would crash the encoder. A + `CaptureRef` becomes `{"__capture__": name}`, a `SteeringRef` becomes + `{"__steering__": name}`, identically to `_step_to_wire`'s `SetpointStep` + arm; any other value (a literal) passes through unchanged. Encode-only.""" + if isinstance(value, CaptureRef): + return {_CAPTURE_WIRE_KEY: value.capture_name} + if isinstance(value, SteeringRef): + return {_STEERING_WIRE_KEY: value.steering_axis_name} + return value + + +def _parameters_to_wire(parameters: Mapping[str, Any]) -> dict[str, Any]: + """Serialize a ComputeStep `parameters` mapping to the hash wire form. + + Per-key application of `_parameter_value_to_wire`. A `parameters` dict + with no ref values hashes byte-identical to a plain `dict(parameters)` + copy, so no existing pinned `steps_hash` is invalidated.""" + return {key: _parameter_value_to_wire(val) for key, val in parameters.items()} + + def _criterion_from_wire( payload: Mapping[str, Any], ) -> EqualsCriterion | WithinToleranceCriterion: @@ -117,12 +142,16 @@ def _expand_step(step: RecipeStep, bindings: Mapping[str, Any]) -> Step: if isinstance(step, RecipeCaptureStep): return CaptureStep(address=step.address, capture_name=step.capture_name) if isinstance(step, RecipeComputeStep): - # All fields are LITERAL (no BindingRef on a compute step yet), so they - # pass through verbatim with no resolve_value pass; `capture_name` - # (slice 6c) + `output_ref_name` (compute-branch chaining) are literal - # template fields carried through unresolved, and an `OutputRef` element - # of `input_uris` rides through UNRESOLVED (the Conductor resolves it at - # execute time). Binding a compute parameter is the deferred widening. + # `command` + `output_uri` are LITERAL (no BindingRef on either). Each + # `parameters` value may be a literal, a `CaptureRef`, or a `SteeringRef`; + # none of the three needs a resolve_value pass here (`BindingRef` is the + # only sentinel `resolve_value` substitutes, and it stays unsupported in + # `parameters`), so the mapping passes through verbatim and a CaptureRef/ + # SteeringRef rides UNRESOLVED into the runtime ComputeStep, exactly like + # an `OutputRef` element of `input_uris`; the Conductor resolves it at + # execute time. `capture_name` (slice 6c) + `output_ref_name` + # (compute-branch chaining) are likewise literal template fields carried + # through unresolved. return ComputeStep( command=step.command, input_uris=step.input_uris, @@ -200,7 +229,7 @@ def _step_to_wire(step: Step) -> dict[str, Any]: "command": list(step.command), "input_uris": [_input_uri_to_wire(u) for u in step.input_uris], "output_uri": step.output_uri, - "parameters": dict(step.parameters), + "parameters": _parameters_to_wire(step.parameters), "capture_name": step.capture_name, "output_ref_name": step.output_ref_name, } diff --git a/apps/api/tests/unit/operation/test_recipe_expansion_capture.py b/apps/api/tests/unit/operation/test_recipe_expansion_capture.py index 599ac1fc94f..b3b115d366d 100644 --- a/apps/api/tests/unit/operation/test_recipe_expansion_capture.py +++ b/apps/api/tests/unit/operation/test_recipe_expansion_capture.py @@ -1,10 +1,11 @@ """Unit tests for capture-step / CaptureRef handling in the recipe expander. `expand` bridges Recipe BC templates to Conductor `Step`s. A -`RecipeCaptureStep` becomes a `CaptureStep`; a `CaptureRef` setpoint value -rides through UNRESOLVED (unlike a `BindingRef`, which is substituted at -expansion). `steps_to_wire` must serialize both deterministically so the -determinism hash is stable across re-expansion. +`RecipeCaptureStep` becomes a `CaptureStep`; a `CaptureRef` setpoint value, or +a `CaptureRef`/`SteeringRef` nested in a compute step's `parameters`, rides +through UNRESOLVED (unlike a `BindingRef`, which is substituted at +expansion). `steps_to_wire` must serialize all of these deterministically so +the determinism hash is stable across re-expansion. """ from __future__ import annotations @@ -14,11 +15,13 @@ import pytest from cora.operation._recipe_expansion import canonical_json_bytes, expand, steps_to_wire -from cora.operation.conductor import CaptureStep, SetpointStep +from cora.operation.conductor import CaptureStep, ComputeStep, SetpointStep from cora.recipe.aggregates.recipe.body import ( CaptureRef, RecipeCaptureStep, + RecipeComputeStep, RecipeSetpointStep, + SteeringRef, ) @@ -59,3 +62,78 @@ def test_steps_to_wire_encodes_capture_ref_as_a_sentinel_not_a_bare_value() -> N """The CaptureRef wire form is distinct from any literal, so the hash can't alias.""" wire = steps_to_wire(expand((RecipeSetpointStep(address="d:x", value=CaptureRef("home")),), {})) assert wire[0]["value"] == {"__capture__": "home"} + + +@pytest.mark.unit +def test_expand_passes_capture_ref_compute_parameter_through_unresolved() -> None: + """A CaptureRef nested in a compute step's parameters survives expansion too.""" + steps = ( + RecipeComputeStep( + command=("tomopy", "recon"), + parameters={"rotation_center": CaptureRef("home")}, + ), + ) + expanded = expand(steps, {}) + head = expanded[0] + assert isinstance(head, ComputeStep) + assert head.parameters == {"rotation_center": CaptureRef("home")} + + +@pytest.mark.unit +def test_expand_passes_steering_ref_compute_parameter_through_unresolved() -> None: + """A SteeringRef nested in a compute step's parameters survives expansion too.""" + steps = ( + RecipeComputeStep( + command=("tomopy", "recon"), + parameters={"seed": SteeringRef("focus")}, + ), + ) + expanded = expand(steps, {}) + head = expanded[0] + assert isinstance(head, ComputeStep) + assert head.parameters == {"seed": SteeringRef("focus")} + + +@pytest.mark.unit +def test_steps_to_wire_hash_is_stable_for_compute_parameter_capture_ref() -> None: + """Re-expanding the same compute-parameter recipe yields a byte-identical hash.""" + steps = ( + RecipeCaptureStep(address="dev:sample:x", capture_name="home"), + RecipeComputeStep( + command=("tomopy", "recon"), + parameters={"rotation_center": CaptureRef("home")}, + ), + ) + + def _hash() -> str: + return hashlib.sha256(canonical_json_bytes(steps_to_wire(expand(steps, {})))).hexdigest() + + assert _hash() == _hash() + + +@pytest.mark.unit +def test_steps_to_wire_encodes_compute_parameter_capture_ref_as_a_sentinel() -> None: + """The CaptureRef wire form inside parameters matches the setpoint-value sentinel shape.""" + steps = ( + RecipeComputeStep( + command=("tomopy", "recon"), + parameters={"rotation_center": CaptureRef("home"), "seed": SteeringRef("focus")}, + ), + ) + wire = steps_to_wire(expand(steps, {})) + assert wire[0]["parameters"] == { + "rotation_center": {"__capture__": "home"}, + "seed": {"__steering__": "focus"}, + } + + +@pytest.mark.unit +def test_steps_to_wire_literal_only_compute_parameters_unchanged() -> None: + """A literal-only parameters dict hashes byte-identical to before this feature. + + Backward-compat pin: no existing pinned steps_hash may be invalidated by + the new per-value wire encoder. + """ + steps = (RecipeComputeStep(command=("tomopy", "recon"), parameters={"algorithm": "sirt"}),) + wire = steps_to_wire(expand(steps, {})) + assert wire[0]["parameters"] == {"algorithm": "sirt"} From cfe8c5925b8245ea5d6cad4345ea1aaf0e881534 Mon Sep 17 00:00:00 2001 From: xmap <16776958+xmap@users.noreply.github.com> Date: Wed, 9 Sep 2026 07:14:21 -0500 Subject: [PATCH 3/6] Round-trip a compute parameter's ref through the resume-replay payload too step_to_payload/_step_from_payload is a second, independent encode/decode site from _recipe_expansion._expand's determinism hash: it is what ResolvedStepsRecorded persists for resume replay. Left unfixed after the prior two slices, a pinned compute step carrying a CaptureRef/SteeringRef parameter would crash canonical_json_bytes at conduct time (this path runs on every conduct, not just recipe-driven ones) even though the hash serializer already handled it. Adds the same per-value encode/decode pair here, kept as an independent copy rather than importing _expand's, matching this file's own existing precedent for _input_uri_to_wire/_input_uri_from_wire (duplicated between the two modules already, not shared, despite both living under the single cora.operation tach boundary). Extends the architecture fitness test that exists specifically to catch a new field with no serializer arm, so the feature ships without a hole in its own safety net. Third slice; ComputeStep.parameters can now carry a ref through every encoding path. Runtime resolution against the per-conduct captures dict is the next slice. Co-Authored-By: Claude Sonnet 5 --- apps/api/src/cora/operation/conductor.py | 66 ++++++++++++++++++- .../test_step_serializers_cover_every_arm.py | 46 ++++++++++++- 2 files changed, 108 insertions(+), 4 deletions(-) diff --git a/apps/api/src/cora/operation/conductor.py b/apps/api/src/cora/operation/conductor.py index 6f06576e793..5201c7184f1 100644 --- a/apps/api/src/cora/operation/conductor.py +++ b/apps/api/src/cora/operation/conductor.py @@ -447,6 +447,55 @@ def _input_uri_from_wire(value: Any) -> "str | OutputRef": return cast("str", value) +def _parameter_value_to_wire(value: Any) -> Any: + """Serialize ONE ComputeStep `parameters` value to the pinned conduct payload form. + + Mirrors `_input_uri_to_wire`'s element-wise shape for the same reason: a + raw `CaptureRef` / `SteeringRef` anywhere in the mapping would crash the + JSON / hash encoder. A `CaptureRef` becomes `{"__capture__": name}`, a + `SteeringRef` becomes `{"__steering__": name}`, identical in shape to a + `SetpointStep.value` ref and to `_recipe_expansion._expand`'s hash + encoder (the two must stay byte-identical or a resume could mis-decode a + pinned step); any other value (a literal) passes through unchanged.""" + if isinstance(value, CaptureRef): + return {_CAPTURE_REF_KEY: value.capture_name} + if isinstance(value, SteeringRef): + return {_STEERING_REF_KEY: value.steering_axis_name} + return value + + +def _parameter_value_from_wire(value: Any) -> Any: + """Deserialize ONE wire ComputeStep `parameters` value; reconstruct a ref. + + Inverse of `_parameter_value_to_wire`: a `{"__capture__": name}` dict + becomes a `CaptureRef`, a `{"__steering__": name}` dict becomes a + `SteeringRef`, any other value (a literal) passes through.""" + if isinstance(value, dict): + typed = cast("dict[str, Any]", value) + if set(typed) == {_CAPTURE_REF_KEY}: + return CaptureRef(capture_name=str(typed[_CAPTURE_REF_KEY])) + if set(typed) == {_STEERING_REF_KEY}: + return SteeringRef(steering_axis_name=str(typed[_STEERING_REF_KEY])) + return cast("Any", value) + + +def _parameters_to_wire(parameters: Mapping[str, Any]) -> dict[str, Any]: + """Serialize a ComputeStep `parameters` mapping to the pinned conduct payload form. + + Per-key application of `_parameter_value_to_wire`. A `parameters` dict + with no ref values serializes byte-identical to a plain `dict(parameters)` + copy, so no existing recorded payload or `ResolvedStepsRecorded` entry + changes shape.""" + return {key: _parameter_value_to_wire(val) for key, val in parameters.items()} + + +def _parameters_from_wire(payload: Mapping[str, Any]) -> dict[str, Any]: + """Deserialize a wire ComputeStep `parameters` mapping; reconstruct any refs. + + Inverse of `_parameters_to_wire`, applied per-key via `_parameter_value_from_wire`.""" + return {key: _parameter_value_from_wire(val) for key, val in payload.items()} + + """Closed-set step-kind discriminators from [[project_operation_design]]. The source of truth for the value set is `STEP_KIND_VALUES` on the Procedure aggregate (re-imported above); the architecture fitness @@ -708,6 +757,17 @@ class ComputeStep: deposited loud-fails (`UnresolvedOutputRef`) with NO in-flight marker and nothing submitted, parity with a `SetpointStep` `CaptureRef`. + `parameters` values may individually be a literal, a `CaptureRef`, or a + `SteeringRef`: the compute-branch sibling of `SetpointStep.value`. Each + rides through expansion + the determinism hash as an opaque sentinel; the + Conductor resolves every ref-valued key at execute time (BEFORE building + the JobSpec, same ordering as the `input_uris` resolve) against the + per-conduct `captures` dict, so a measured value or a brain's advised + coordinate can become a compute job's parameter rather than only a motor + position. An unresolved `CaptureRef` or unseeded `SteeringRef` loud-fails + with NO in-flight marker and nothing submitted, parity with the + `input_uris` `OutputRef` case. + `output_ref_name` names the `outputs` slot the produced `ArtifactRef` deposits into (the FILE arm), the artifact-bus chaining twin of `capture_name`. When set, after the file arm fetches the `ArtifactRef` and @@ -731,7 +791,7 @@ class ComputeStep: command: tuple[str, ...] input_uris: tuple[str | OutputRef, ...] = () output_uri: str | None = None - parameters: Mapping[str, Any] = field(default_factory=dict[str, Any]) + parameters: Mapping[str, Any | CaptureRef | SteeringRef] = field(default_factory=dict[str, Any]) capture_name: str | None = None output_ref_name: str | None = None @@ -4899,7 +4959,7 @@ def step_to_payload(step: Step) -> dict[str, Any]: "command": list(step.command), "input_uris": [_input_uri_to_wire(u) for u in step.input_uris], "output_uri": step.output_uri, - "parameters": dict(step.parameters), + "parameters": _parameters_to_wire(step.parameters), "capture_name": step.capture_name, "output_ref_name": step.output_ref_name, } @@ -4961,7 +5021,7 @@ def _step_from_payload(payload: Mapping[str, Any]) -> Step: command=tuple(payload["command"]), input_uris=tuple(_input_uri_from_wire(u) for u in payload.get("input_uris", ())), output_uri=payload.get("output_uri"), - parameters=dict(payload.get("parameters", {})), + parameters=_parameters_from_wire(payload.get("parameters", {})), capture_name=payload.get("capture_name"), output_ref_name=payload.get("output_ref_name"), ) diff --git a/apps/api/tests/architecture/test_step_serializers_cover_every_arm.py b/apps/api/tests/architecture/test_step_serializers_cover_every_arm.py index 4f1fe515b5c..134277006b5 100644 --- a/apps/api/tests/architecture/test_step_serializers_cover_every_arm.py +++ b/apps/api/tests/architecture/test_step_serializers_cover_every_arm.py @@ -35,7 +35,7 @@ step_to_payload, steps_from_payload, ) -from cora.recipe.aggregates.recipe.body import OutputRef +from cora.recipe.aggregates.recipe.body import CaptureRef, OutputRef, SteeringRef # One representative instance per Step arm. The mapping is keyed by the arm # class so a new union arm with no instance here trips the coverage assertion @@ -76,6 +76,22 @@ output_ref_name=None, ) +# A ComputeStep whose `parameters` mixes a literal with a CaptureRef and a +# SteeringRef value. Kept SEPARATE from `_INSTANCES` for the same reason as +# `_COMPUTE_STEP_NO_CAPTURE`: it targets one specific field's ref-bearing +# shape rather than doubling as the arm's baseline representative. A raw +# CaptureRef/SteeringRef anywhere in `parameters` would crash +# `canonical_json_bytes` (no `default=`) if a serializer copied the mapping +# verbatim instead of per-value encoding it. +_COMPUTE_STEP_WITH_PARAMETER_REFS = ComputeStep( + command=("tomopy", "recon"), + parameters={ + "algorithm": "sirt", + "rotation_center": CaptureRef(capture_name="offset"), + "seed": SteeringRef(steering_axis_name="focus"), + }, +) + _CHECK_WIRE_KIND = "check" @@ -235,3 +251,31 @@ def test_compute_step_output_ref_round_trips_both_serializers() -> None: # determinism hash splits an output_ref_name-set step from a None one) (wire,) = steps_to_wire((instance,)) assert wire["output_ref_name"] == instance.output_ref_name + + +@pytest.mark.architecture +def test_compute_step_parameter_refs_round_trip_both_serializers() -> None: + """A CaptureRef/SteeringRef nested in `parameters` survives both serializers. + + Mirrors `test_compute_step_output_ref_round_trips_both_serializers`: a raw + CaptureRef/SteeringRef anywhere in `parameters` would crash + `canonical_json_bytes` (no `default=`) if a serializer copied the mapping + verbatim instead of per-value encoding it. + """ + instance = _COMPUTE_STEP_WITH_PARAMETER_REFS + + # payload serializer round-trip (value-identical, incl. the ref objects, + # not flattened to their sentinel dicts) + (rebuilt,) = steps_from_payload([step_to_payload(instance)]) + assert rebuilt == instance, ( + "ComputeStep parameters did not round-trip through step_to_payload/_step_from_payload." + ) + + # hash/wire serializer encodes each ref as its sentinel shape, identical + # to the SetpointStep.value sentinel form + (wire,) = steps_to_wire((instance,)) + assert wire["parameters"] == { + "algorithm": "sirt", + "rotation_center": {"__capture__": "offset"}, + "seed": {"__steering__": "focus"}, + } From 3e45c71844e176d0c7dfccba5d754de4ce98622e Mon Sep 17 00:00:00 2001 From: xmap <16776958+xmap@users.noreply.github.com> Date: Wed, 9 Sep 2026 07:42:08 -0500 Subject: [PATCH 4/6] Let a compute step read a capture or a steering advice into its parameters The final slice: _run_compute now resolves every CaptureRef/SteeringRef value in ComputeStep.parameters against the per-conduct captures dict, immediately after the existing input_uris/OutputRef resolve and before building the JobSpec. An unresolved capture or unseeded steering axis loud-fails with a recorded entry, no in-flight marker, nothing submitted, parity with the OutputRef case and with _run_setpoint's own CaptureRef/SteeringRef handling. Reuses _ERROR_UNRESOLVED_CAPTURE for both ref kinds rather than adding a new error class, matching _run_setpoint's own precedent (it already uses this label for its SteeringRef arm, not a separate one). Provenance rides a new parameter_refs payload key, emitted only when a parameter actually carries a ref, so a literal-only step's recorded entries stay byte-identical to before this feature. Also fixes the pre-existing OutputRef-unresolved failure body, which copied parameters via a raw dict() even though parameters may still hold an unresolved ref at that point (that failure returns before the new parameter resolution runs); it now goes through the same wire encoder. This closes the loop the first three slices set up: a tomography reconstruction (or any compute job) can now read a value an earlier step measured, or a coordinate a steering brain just advised, the same way a setpoint already could. Compute stops being read-only in the conduct loop. Co-Authored-By: Claude Sonnet 5 --- apps/api/src/cora/operation/conductor.py | 90 +++- .../test_conductor_compute_parameter_refs.py | 412 ++++++++++++++++++ 2 files changed, 496 insertions(+), 6 deletions(-) create mode 100644 apps/api/tests/unit/operation/test_conductor_compute_parameter_refs.py diff --git a/apps/api/src/cora/operation/conductor.py b/apps/api/src/cora/operation/conductor.py index 5201c7184f1..34d2d37f35c 100644 --- a/apps/api/src/cora/operation/conductor.py +++ b/apps/api/src/cora/operation/conductor.py @@ -310,9 +310,13 @@ `_run_compute`.""" _ERROR_UNRESOLVED_CAPTURE = "UnresolvedCaptureRef" -"""error_class for a SetpointStep CaptureRef whose name was never captured -in this conduct (e.g. resumed past the capturing step). Loud-fail label, -not an exception type: the failure is recorded + returned, not raised.""" +"""error_class for a CaptureRef (or a SteeringRef seeded by the decide loop) +whose name was never captured/seeded in this conduct (e.g. resumed past the +capturing step). Loud-fail label, not an exception type: the failure is +recorded + returned, not raised. Shared by SetpointStep.value AND, since the +compute-parameter widening, any ComputeStep.parameters value: one label per +ref kind, not per step kind, matching the disambiguation living in `message` ++ `target` rather than in `error_class`.""" _ERROR_DUPLICATE_CAPTURE = "DuplicateCapture" """error_class for a CaptureStep re-capturing an already-filled name within @@ -4308,7 +4312,7 @@ async def _run_compute( "command": list(step.command), "input_refs": [_input_uri_to_wire(u) for u in step.input_uris], "output_uri": step.output_uri, - "parameters": dict(step.parameters), + "parameters": _parameters_to_wire(step.parameters), }, result=_RESULT_FAILED, error_class=_ERROR_UNRESOLVED_OUTPUT, @@ -4325,18 +4329,92 @@ async def _run_compute( else: resolved_input_uris.append(element) resolved_uris = tuple(resolved_input_uris) + # Resolve every CaptureRef / SteeringRef parameter value against the + # per-conduct `captures` dict BEFORE any effect (parity with the + # OutputRef resolve just above, and with _run_setpoint's single-value + # version): an unresolved/unseeded name loud-fails with a recorded + # entry, NO in-flight marker, nothing submitted. `resolved_parameters` + # is what the JobSpec + ComputePort see; the pre-resolution refs are + # recorded separately under `parameter_refs` for provenance, mirroring + # `input_refs` above. + resolved_parameters: dict[str, Any] = {} + for key, param_value in step.parameters.items(): + if isinstance(param_value, CaptureRef): + if param_value.capture_name not in captures: + msg = ( + f"compute step parameter {key!r} references capture " + f"{param_value.capture_name!r} not captured before this step" + ) + await self._record( + envelope=envelope, + index=index, + step_kind=_STEP_KIND_COMPUTE, + body={ + "command": list(step.command), + "input_uris": list(resolved_uris), + "output_uri": step.output_uri, + "parameters": _parameters_to_wire(step.parameters), + }, + result=_RESULT_FAILED, + error_class=_ERROR_UNRESOLVED_CAPTURE, + message=msg, + ) + return ConductorFailure( + step_index=index, + source_kind=_STEP_KIND_COMPUTE, + target=" ".join(step.command), + error_class=_ERROR_UNRESOLVED_CAPTURE, + message=msg, + ) + resolved_parameters[key] = captures[param_value.capture_name] + elif isinstance(param_value, SteeringRef): + if param_value.steering_axis_name not in captures: + msg = ( + f"compute step parameter {key!r} references steering axis " + f"{param_value.steering_axis_name!r} not seeded before this step" + ) + await self._record( + envelope=envelope, + index=index, + step_kind=_STEP_KIND_COMPUTE, + body={ + "command": list(step.command), + "input_uris": list(resolved_uris), + "output_uri": step.output_uri, + "parameters": _parameters_to_wire(step.parameters), + }, + result=_RESULT_FAILED, + error_class=_ERROR_UNRESOLVED_CAPTURE, + message=msg, + ) + return ConductorFailure( + step_index=index, + source_kind=_STEP_KIND_COMPUTE, + target=" ".join(step.command), + error_class=_ERROR_UNRESOLVED_CAPTURE, + message=msg, + ) + resolved_parameters[key] = captures[param_value.steering_axis_name] + else: + resolved_parameters[key] = param_value job_spec = JobSpec( command=step.command, input_uris=resolved_uris, output_uri=step.output_uri, - parameters=step.parameters, + parameters=resolved_parameters, ) payload_body: dict[str, Any] = { "command": list(step.command), "input_uris": list(resolved_uris), "output_uri": step.output_uri, - "parameters": dict(step.parameters), + "parameters": resolved_parameters, } + # Provenance: record the pre-resolution parameter refs (sentinel dicts + # for any CaptureRef/SteeringRef value) beside the resolved parameters + # only when the step carried one, mirroring `input_refs` above and + # _run_setpoint recording value + capture_ref/steering_ref. + if any(isinstance(v, CaptureRef | SteeringRef) for v in step.parameters.values()): + payload_body["parameter_refs"] = _parameters_to_wire(step.parameters) # Provenance: record the pre-resolution refs (sentinel dicts for any # OutputRef element) beside the resolved URIs only when the step carried # a ref, mirroring _run_setpoint recording value + capture_ref. diff --git a/apps/api/tests/unit/operation/test_conductor_compute_parameter_refs.py b/apps/api/tests/unit/operation/test_conductor_compute_parameter_refs.py new file mode 100644 index 00000000000..9d092cd35b5 --- /dev/null +++ b/apps/api/tests/unit/operation/test_conductor_compute_parameter_refs.py @@ -0,0 +1,412 @@ +"""Behavioural tests for CaptureRef/SteeringRef values in ComputeStep.parameters. + +Coverage for the runtime resolution added to `_run_compute` (the fourth and +final slice of the compute-parameter-refs feature; the prior three slices +covered the Recipe event-payload wire, the determinism-hash wire, and the +`ResolvedStepsRecorded` payload wire, all encode/decode only): + + Resolve (the actual behavior): + - a CaptureRef parameter value resolves against the per-conduct `captures` + dict BEFORE the JobSpec is built, so a measured value can become a + compute job's parameter + - a SteeringRef parameter value resolves the same way, so a steering brain + can tune a compute parameter directly rather than only a motor position + - a literal value alongside ref values resolves together in one JobSpec + + Loud-fails (each -> recorded `failed` entry + ConductorFailure halt, NO + in-flight marker, NOTHING submitted, parity with the ComputeStep OutputRef + case and with _run_setpoint's CaptureRef/SteeringRef): + - a CaptureRef to a name never captured + - a SteeringRef to an axis never seeded + + Provenance: + - a ref-bearing ComputeStep's recorded entries carry a `parameter_refs` key + (the pre-resolution sentinel shapes); a literal-only step's entries carry + none, so no existing recorded payload changes shape (backward-compat). + + Interaction with the OutputRef resolve (input_uris) that runs first: + - an unresolved OutputRef input on a step that ALSO carries a parameter ref + must not crash the failure body, since parameters resolution never runs + (the OutputRef failure returns first). + +The unit tier uses `InMemoryComputePort` (records what it received) + the +shared fake append-step handler, mirroring `test_conductor_compute_output.py`. +""" + +from dataclasses import dataclass, field +from datetime import UTC, datetime +from uuid import UUID, uuid4 + +import pytest + +from cora.infrastructure.ports.clock import FakeClock +from cora.infrastructure.routing import NIL_SENTINEL_ID +from cora.operation.adapters.in_memory_compute_port import InMemoryComputePort +from cora.operation.adapters.in_memory_control_port import InMemoryControlPort +from cora.operation.conductor import ComputeStep, Conductor +from cora.operation.features.append_activities.command import AppendProcedureActivities +from cora.operation.ports.compute_port import JobId, JobSpec +from cora.operation.ports.measurement import Measurement +from cora.recipe.aggregates.recipe.body import CaptureRef, OutputRef, SteeringRef + +_FIXED_NOW = datetime(2026, 6, 24, 9, 0, 0, tzinfo=UTC) + + +class _RecordingComputePort(InMemoryComputePort): + """`InMemoryComputePort` that records every submitted `JobSpec`. + + Lets a test assert what the Conductor actually submitted (the RESOLVED + parameters) without reaching into the fake's private job map, and count + submits to prove the unresolved-ref path submits NOTHING for the failing + step.""" + + def __init__(self) -> None: + super().__init__() + self.submitted_specs: list[JobSpec] = [] + + async def submit(self, job_spec: JobSpec) -> JobId: + self.submitted_specs.append(job_spec) + return await super().submit(job_spec) + + @property + def submit_count(self) -> int: + return len(self.submitted_specs) + + +@dataclass +class _AppendCall: + command: AppendProcedureActivities + + +@dataclass +class _FakeAppendStep: + """Fake `Handler` for the append_activities slice; records every call.""" + + calls: list[_AppendCall] = field(default_factory=list[_AppendCall]) + + async def __call__( + self, + command: AppendProcedureActivities, + *, + principal_id: UUID, + correlation_id: UUID, + causation_id: UUID | None = None, + surface_id: UUID = NIL_SENTINEL_ID, + ) -> int: + self.calls.append(_AppendCall(command=command)) + return len(command.entries) + + +def _conductor(appender: _FakeAppendStep, *, compute_port: _RecordingComputePort) -> Conductor: + return Conductor( + control_port=InMemoryControlPort(), + append_step=appender, + clock=FakeClock(_FIXED_NOW), + id_generator=_NilIdGenerator(), + compute_port=compute_port, + ) + + +@dataclass +class _NilIdGenerator: + def new_id(self) -> UUID: + return uuid4() + + +def _entries(appender: _FakeAppendStep) -> list[dict[str, object]]: + return [call.command.entries[0].payload for call in appender.calls] + + +_RECON_URI = "file:///data/2bm/recon.h5" + + +def _offset_measurement(value: object, *, name: str = "offset") -> Measurement: + return Measurement( + value=value, + kind="Scalar", + quality="Good", # type: ignore[arg-type] + produced_at=_FIXED_NOW, + name=name, + units="pixel", + ) + + +@pytest.mark.unit +async def test_capture_ref_parameter_resolves_into_job_spec() -> None: + """A CaptureRef parameter value resolves against externally-seeded captures.""" + appender = _FakeAppendStep() + port = _RecordingComputePort() + conductor = _conductor(appender, compute_port=port) + + step = ComputeStep( + command=("tomopy", "recon"), + output_uri=_RECON_URI, + parameters={"rotation_center": CaptureRef("offset")}, + ) + + result = await conductor.execute( + procedure_id=uuid4(), + principal_id=uuid4(), + correlation_id=uuid4(), + steps=(step,), + captures={"offset": 12.5}, + ) + + assert result.succeeded is True + assert port.submitted_specs[-1].parameters == {"rotation_center": 12.5} + + +@pytest.mark.unit +async def test_capture_ref_parameter_resolves_value_deposited_earlier_in_same_pass() -> None: + """A value-arm ComputeStep's capture_name deposit feeds a LATER compute parameter. + + Mirrors `test_conductor_compute_capture.py`'s same-pass deposit idiom: the + first ComputeStep's produced Measurement deposits into `captures` via + `capture_name`; the second ComputeStep's CaptureRef parameter reads it. + """ + appender = _FakeAppendStep() + port = _RecordingComputePort() + port.set_next_measurements((_offset_measurement(7.5),)) + conductor = _conductor(appender, compute_port=port) + + find_offset = ComputeStep( + command=("tomopy", "find_center"), + input_uris=("file:///a.h5",), + output_uri=None, + parameters={"algorithm": "vo"}, + capture_name="offset", + ) + reconstruct = ComputeStep( + command=("tomopy", "recon"), + output_uri=_RECON_URI, + parameters={"rotation_center": CaptureRef("offset")}, + ) + + result = await conductor.execute( + procedure_id=uuid4(), + principal_id=uuid4(), + correlation_id=uuid4(), + steps=(find_offset, reconstruct), + ) + + assert result.succeeded is True + assert port.submitted_specs[-1].parameters == {"rotation_center": 7.5} + + +@pytest.mark.unit +async def test_steering_ref_parameter_resolves_into_job_spec() -> None: + """A SteeringRef parameter value resolves against externally-seeded captures. + + Mirrors how the decide loop seeds a steering axis before a pass; a unit + test seeds it directly via `execute(..., captures=...)`. + """ + appender = _FakeAppendStep() + port = _RecordingComputePort() + conductor = _conductor(appender, compute_port=port) + + step = ComputeStep( + command=("tomopy", "recon"), + output_uri=_RECON_URI, + parameters={"seed": SteeringRef("focus")}, + ) + + result = await conductor.execute( + procedure_id=uuid4(), + principal_id=uuid4(), + correlation_id=uuid4(), + steps=(step,), + captures={"focus": 0.42}, + ) + + assert result.succeeded is True + assert port.submitted_specs[-1].parameters == {"seed": 0.42} + + +@pytest.mark.unit +async def test_mixed_literal_and_ref_parameters_resolve_together() -> None: + """A literal, a CaptureRef, and a SteeringRef in one parameters dict all resolve.""" + appender = _FakeAppendStep() + port = _RecordingComputePort() + conductor = _conductor(appender, compute_port=port) + + step = ComputeStep( + command=("tomopy", "recon"), + output_uri=_RECON_URI, + parameters={ + "algorithm": "sirt", + "rotation_center": CaptureRef("offset"), + "seed": SteeringRef("focus"), + }, + ) + + result = await conductor.execute( + procedure_id=uuid4(), + principal_id=uuid4(), + correlation_id=uuid4(), + steps=(step,), + captures={"offset": 12.5, "focus": 0.42}, + ) + + assert result.succeeded is True + assert port.submitted_specs[-1].parameters == { + "algorithm": "sirt", + "rotation_center": 12.5, + "seed": 0.42, + } + + +@pytest.mark.unit +async def test_unresolved_capture_ref_parameter_records_failure_with_no_marker_and_no_submit() -> ( + None +): + """A CaptureRef to a name never captured loud-fails BEFORE the marker + submit. + + Parity with the ComputeStep OutputRef case and with _run_setpoint's + UnresolvedCaptureRef: a single FAILED entry, no in-flight marker, and + NOTHING submitted to the compute substrate. + """ + appender = _FakeAppendStep() + port = _RecordingComputePort() + conductor = _conductor(appender, compute_port=port) + + step = ComputeStep( + command=("tomopy", "recon"), + output_uri=_RECON_URI, + parameters={"rotation_center": CaptureRef("missing")}, + ) + + submits_before = port.submit_count + result = await conductor.execute( + procedure_id=uuid4(), + principal_id=uuid4(), + correlation_id=uuid4(), + steps=(step,), + ) + + assert result.succeeded is False + assert result.failure is not None + assert result.failure.source_kind == "compute" + assert result.failure.error_class == "UnresolvedCaptureRef" + assert port.submit_count == submits_before + entries = _entries(appender) + assert len(entries) == 1 + assert entries[0]["result"] == "failed" + assert entries[0]["error_class"] == "UnresolvedCaptureRef" + + +@pytest.mark.unit +async def test_unseeded_steering_ref_parameter_records_failure_with_no_marker_and_no_submit() -> ( + None +): + """A SteeringRef to an axis never seeded loud-fails the same way as an unresolved CaptureRef.""" + appender = _FakeAppendStep() + port = _RecordingComputePort() + conductor = _conductor(appender, compute_port=port) + + step = ComputeStep( + command=("tomopy", "recon"), + output_uri=_RECON_URI, + parameters={"seed": SteeringRef("missing")}, + ) + + submits_before = port.submit_count + result = await conductor.execute( + procedure_id=uuid4(), + principal_id=uuid4(), + correlation_id=uuid4(), + steps=(step,), + ) + + assert result.succeeded is False + assert result.failure is not None + assert result.failure.error_class == "UnresolvedCaptureRef" + assert port.submit_count == submits_before + entries = _entries(appender) + assert len(entries) == 1 + assert entries[0]["result"] == "failed" + + +@pytest.mark.unit +async def test_parameter_refs_recorded_only_when_present() -> None: + """A literal-only ComputeStep's recorded entries carry no parameter_refs key. + + Backward-compat pin: the new provenance field must not appear on a step + that carries no ref, so no existing recorded payload changes shape. A + ref-bearing step's entries DO carry it, with the sentinel shape. + """ + literal_appender = _FakeAppendStep() + literal_conductor = _conductor(literal_appender, compute_port=_RecordingComputePort()) + literal_step = ComputeStep( + command=("tomopy", "recon"), + output_uri=_RECON_URI, + parameters={"algorithm": "sirt"}, + ) + + literal_result = await literal_conductor.execute( + procedure_id=uuid4(), + principal_id=uuid4(), + correlation_id=uuid4(), + steps=(literal_step,), + ) + + assert literal_result.succeeded is True + for entry in _entries(literal_appender): + assert "parameter_refs" not in entry + + ref_appender = _FakeAppendStep() + ref_conductor = _conductor(ref_appender, compute_port=_RecordingComputePort()) + ref_step = ComputeStep( + command=("tomopy", "recon"), + output_uri=_RECON_URI, + parameters={"rotation_center": CaptureRef("offset")}, + ) + + ref_result = await ref_conductor.execute( + procedure_id=uuid4(), + principal_id=uuid4(), + correlation_id=uuid4(), + steps=(ref_step,), + captures={"offset": 12.5}, + ) + + assert ref_result.succeeded is True + marker = next(e for e in _entries(ref_appender) if e["result"] == "in_flight") + assert marker["parameter_refs"] == {"rotation_center": {"__capture__": "offset"}} + + +@pytest.mark.unit +async def test_unresolved_output_ref_does_not_crash_when_parameters_also_carry_a_ref() -> None: + """An unresolved OutputRef input records its failure body even with a parameter ref present. + + The OutputRef resolve (input_uris) runs BEFORE the new parameters resolve, + so a step failing there never reaches parameter resolution. Its failure + body must still serialize `parameters` safely (via the wire encoder, not + a raw `dict()` copy) even though `parameters` still holds an unresolved + CaptureRef at that point. + """ + appender = _FakeAppendStep() + port = _RecordingComputePort() + conductor = _conductor(appender, compute_port=port) + + step = ComputeStep( + command=("tomopy", "recon"), + input_uris=(OutputRef("missing"),), + output_uri=_RECON_URI, + parameters={"rotation_center": CaptureRef("offset")}, + ) + + submits_before = port.submit_count + result = await conductor.execute( + procedure_id=uuid4(), + principal_id=uuid4(), + correlation_id=uuid4(), + steps=(step,), + ) + + assert result.succeeded is False + assert result.failure is not None + assert result.failure.error_class == "UnresolvedOutputRef" + assert port.submit_count == submits_before + entries = _entries(appender) + assert len(entries) == 1 + assert entries[0]["parameters"] == {"rotation_center": {"__capture__": "offset"}} From 7cb93397f690b50b9bdd1cd482681cb79113df03 Mon Sep 17 00:00:00 2001 From: xmap <16776958+xmap@users.noreply.github.com> Date: Wed, 9 Sep 2026 08:46:54 -0500 Subject: [PATCH 5/6] Say what a step does, and stop the module docstring naming three of five The Operation docstring still enumerated the runtime Step union as Setpoint | Action | Check in four places, two step kinds out of date since capture and compute landed. Replaced the enumerations with a pointer to STEP_KIND_VALUES and the fitness test that pins the arm set, rather than a fresh hand-list that goes stale the next time an arm lands. The module doc gains the split the five kinds already have but that no single page stated: capture and check observe, setpoint and action act, compute does both, and deciding is not a step kind at all because the brain runs between passes rather than inside one. That last part is the load-bearing half. Reading the kinds as though they ought to partition into observe, think and act compresses two altitudes into one, which is the reading the prose now heads off. Co-Authored-By: Claude Opus 5 --- apps/api/src/cora/operation/__init__.py | 32 ++++++++++++-------- docs/architecture/modules/operation/index.md | 2 ++ 2 files changed, 21 insertions(+), 13 deletions(-) diff --git a/apps/api/src/cora/operation/__init__.py b/apps/api/src/cora/operation/__init__.py index 52ecbadbc2d..aba71d58e57 100644 --- a/apps/api/src/cora/operation/__init__.py +++ b/apps/api/src/cora/operation/__init__.py @@ -5,9 +5,12 @@ - `Procedure` aggregate: one execution of an episodic operational task — bakeout, characterization, optical alignment, beam-mode change, recovery procedure, ID maintenance, KB switching. Each - Procedure has sequenced steps; each step has a Setpoint / Action - / Check triplet (CORA's rename of ISA-106's canonical - Command/Perform/Verify to avoid catastrophic CQRS collision). + Procedure has sequenced steps; the Setpoint / Action / Check core + is CORA's rename of ISA-106's canonical Command/Perform/Verify to + avoid catastrophic CQRS collision, and Capture (a runtime value + read) plus Compute (a ComputePort job submission) extend it for + the conduct-path runtimes. `STEP_KIND_VALUES` is the single source + of truth for the current set. Track B BC (ISA-106 lens). Independent of Track A (Recipe / Subject / Data). Distinct from ISA-88 batch operations which the Run BC @@ -21,8 +24,8 @@ Slices: `register_procedure` (genesis -> Defined), `start_procedure` / `complete_procedure` / `abort_procedure` / `truncate_procedure` -(FSM transitions), `append_activities` (per-step logbook with -Setpoint/Action/Check rows mirroring Run BC's Observation channel), +(FSM transitions), `append_activities` (per-step logbook, one row +per step kind, mirroring Run BC's Observation channel), `get_procedure` (fold-on-read), `list_procedures` (projection-backed). ## Step vs Activity altitude split @@ -31,18 +34,21 @@ deliberately: - **Runtime `Step` union** (`conductor.py`): the discriminated - union `SetpointStep | ActionStep | CheckStep` — the IN-FLIGHT - spec the Conductor walks during a Procedure execution. Each - variant is what the conductor IS TOLD TO DO at one step. + union of per-kind step variants — the IN-FLIGHT spec the + Conductor walks during a Procedure execution. Each variant is + what the conductor IS TOLD TO DO at one step. The arm set is + pinned against `STEP_KIND_VALUES` by + `test_conductor_step_kinds_match_procedure`; read the union in + `conductor.py` rather than an enumeration here, which goes stale + every time an arm lands. - **Persisted `Activity` entry** (`aggregates/procedure/entries.py`): one row per executed step, capturing WHAT HAPPENED (the step - that ran, with its result). Path C polymorphic table with - `step_kind` discriminator carrying the runtime variant's name - (`setpoint` / `action` / `check`). + that ran, with its result). Path C polymorphic table with a + `step_kind` discriminator carrying the runtime variant's name. The relationship: each runtime `Step` execution writes one -`Activity` entry through the `ActivityStore` port. Conductor's -`SetpointStep | ActionStep | CheckStep` are the SPEC; the +`Activity` entry through the `ActivityStore` port. The Conductor's +`Step` arms are the SPEC; the `entries_operation_procedure_activities` rows are the LOG. The 2026-06-09 logbook-entry rename (project_logbook_entry_storage diff --git a/docs/architecture/modules/operation/index.md b/docs/architecture/modules/operation/index.md index bbe33485bbb..54a8d8a5dca 100644 --- a/docs/architecture/modules/operation/index.md +++ b/docs/architecture/modules/operation/index.md @@ -14,6 +14,8 @@ A Procedure is distinct from a Run by what it leaves of record: a Run exists to **Steering.** A convergence loop follows a fixed recipe; a steered loop asks a brain where to look next. `conduct_until_advised` is the decide-axis twin of the convergence loop: after each pass it hands the accumulated evidence to a `DecidePort` and seeds the next pass from the point the brain advises, until the brain advises stop. The seam is deliberately optimizer- and action-neutral, expressed as six nouns: an objective (what "good" means, named by a measurement), a search space of axes, the evidence so far, one observation, the advice returned, and a budget. Optimizer internals (kernel, acquisition, surrogate) never cross the seam; a next point is coordinates keyed by axis name, never a command, so translating a point into steps stays the loop's job. The brain is handed the full history every call and is assumed to hold no memory between calls, which keeps a pure-function brain and a stateful one behind one surface. Shipped brains: an in-memory fake, a deterministic `grid_walk` sweep, a `sobol` low-discrepancy initial-design seeder, a `botorch` Gaussian-process Bayesian-optimization brain, and a `staged` composite that seeds with Sobol then hands off to the GP once enough successful observations exist. The GP brain and Sobol seeder need the optional `bo` dependency group (BoTorch, on PyTorch); the base install stays lean. When a learning brain decides a pass, the fitted model's summary scalars (per-axis lengthscales, observation noise, acquisition value) are captured for audit as one row per iteration in a per-Procedure diagnostics logbook (a `Diagnostic` entry kind alongside the `Activity` step log; a side table that does not fold into aggregate state), so a reviewer can later reconstruct why the brain advised a given point. Because a Gaussian-process fit is not bit-reproducible across environments even with a fixed seed, a GP-steered run cannot be reconstructed by re-asking the brain. Instead the loop records the coordinate the brain advised each pass (`advised_next_point` on the iteration event), surfaced alongside the verdict and deciding brain in the per-iteration read model, so a finished GP-steered run is reconstructable by reading the recorded decision trail. The trail also records how long the brain took to answer each pass, on the faulted path as well as the advised one. Time is the one cost a non-LLM brain spends: a Gaussian process draws neither tokens nor money, so the spend limits that bound an LLM-backed agent are blind to it, while at a beamline the scarce resource is the beam itself. A caller may declare a `SteeringBudget` (a pass count, a wall-clock allowance, or both) and the loop enforces it between passes, ending the Procedure the same way a brain-advised stop does, with a recorded reason naming which dimension ran out. The budget is per call, restated on every resume rather than accumulated: only the pass count is backstopped across an unbounded chain of resumes, by the same absolute ceiling that guards a brain which never advises stop; wall-clock time is not. Which layer should own a cap that DOES survive a resume, the brain, the Procedure, or the Allocation, is still open. The run stays classified "forward-only" (meaning not re-ask-reproducible, distinct from not-recorded and, now, distinct from not-resumable). That trail records what the brain DID; `SteeringDesignRecorded` records what it was ASKED, pinning the objective, the search space, the budget and the brain configuration once per conduct segment, before the first pass of that segment runs. Both sides are needed for a different and weaker criterion than replay: a reader who wants to draw an inference from a steered run has to know the rule that chose the points, and a rule that was never written down is not thereby harmless, it is unverifiable. A resume may change the budget, the brain, the bounds or the objective and those changes are recorded rather than refused; what it may not do is hand the brain a space that cannot express a coordinate already measured, which is an axis dropped, an axis added, or a categorical axis that has lost a value already tried. A crashed GP-steered run resumes through `conduct_until_advised_from`: it rebuilds the brain's history from the record, the measured value of each closed pass from a per-Procedure outcome logbook (an `Outcome` entry kind, sibling to the diagnostics log) and the advised coordinate from the iteration trail, then consults the brain only at the open frontier. Resume neither re-asks the brain for a pass already done nor re-drives hardware to a past position: the next move is an absolute one issued from wherever the instrument now sits, which reaches the same commanded point regardless of the current position, and direction-of-approach consistency (anti-backlash) is a per-move concern of the motion controller, not of resume. True sample path-dependence (thermal or radiation history, mechanical creep) is shaped by the whole trajectory rather than any single point, so it belongs in provenance, not in a resume-time re-drive; re-establishing one historical setpoint would neither be necessary nor sufficient. See [the recording spine and the optional execution edge](../../standards.md#the-recording-spine-and-the-optional-execution-edge). +**What a step does, and where deciding happens.** The step kinds divide by what they do with a value, and the division is worth stating because it is otherwise only legible one docstring at a time. `CaptureStep` and `CheckStep` observe: a capture is a pure read that stores the readback, a check is a read that also gates. `SetpointStep` and `ActionStep` act: a setpoint commands a value, an action performs a named operation. `ComputeStep` does both in one step, submitting a job and then reading back what it produced, which is the same pairing a `SetpointStep` makes optional through `verify`. Deciding is deliberately not a step kind. A brain runs between passes, not inside one: the loop walks a pass, hands the accumulated evidence to a `DecidePort`, and seeds the next pass from the advised point. That places observing and acting inside the step walk and deciding one layer above it. Reading the step kinds as though they should partition into observe, think and act would compress two altitudes into one: the brain's turn is a separate consult carrying its own record, not a step in the list. +
Out of scope From 4414bb69e76b4467b8009034aeea766fb64f19c0 Mon Sep 17 00:00:00 2001 From: xmap <16776958+xmap@users.noreply.github.com> Date: Wed, 9 Sep 2026 09:39:37 -0500 Subject: [PATCH 6/6] Correct the claim that deciding never happens inside a step The paragraph shipped a day ago said deciding is deliberately not a step kind. A CheckStep evaluates a criterion and gates the walk on the result, and BrainKind names RULE as a real brain, so a deterministic brain already runs inside the step list. The accurate statement is that deciding happens at two altitudes: the cheap deterministic kind fits in a step, and the loop brain does not, because its verdict can complete the whole Procedure and it reads across every pass rather than one. The reasons the loop brain sits outside the list were written down nowhere, so a reader had to reconstruct them from the DecidePort docstring and the conductor's decide pseudo-kind. Co-Authored-By: Claude Opus 5 (1M context) --- docs/architecture/modules/operation/index.md | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/docs/architecture/modules/operation/index.md b/docs/architecture/modules/operation/index.md index 54a8d8a5dca..9aae5837659 100644 --- a/docs/architecture/modules/operation/index.md +++ b/docs/architecture/modules/operation/index.md @@ -14,7 +14,7 @@ A Procedure is distinct from a Run by what it leaves of record: a Run exists to **Steering.** A convergence loop follows a fixed recipe; a steered loop asks a brain where to look next. `conduct_until_advised` is the decide-axis twin of the convergence loop: after each pass it hands the accumulated evidence to a `DecidePort` and seeds the next pass from the point the brain advises, until the brain advises stop. The seam is deliberately optimizer- and action-neutral, expressed as six nouns: an objective (what "good" means, named by a measurement), a search space of axes, the evidence so far, one observation, the advice returned, and a budget. Optimizer internals (kernel, acquisition, surrogate) never cross the seam; a next point is coordinates keyed by axis name, never a command, so translating a point into steps stays the loop's job. The brain is handed the full history every call and is assumed to hold no memory between calls, which keeps a pure-function brain and a stateful one behind one surface. Shipped brains: an in-memory fake, a deterministic `grid_walk` sweep, a `sobol` low-discrepancy initial-design seeder, a `botorch` Gaussian-process Bayesian-optimization brain, and a `staged` composite that seeds with Sobol then hands off to the GP once enough successful observations exist. The GP brain and Sobol seeder need the optional `bo` dependency group (BoTorch, on PyTorch); the base install stays lean. When a learning brain decides a pass, the fitted model's summary scalars (per-axis lengthscales, observation noise, acquisition value) are captured for audit as one row per iteration in a per-Procedure diagnostics logbook (a `Diagnostic` entry kind alongside the `Activity` step log; a side table that does not fold into aggregate state), so a reviewer can later reconstruct why the brain advised a given point. Because a Gaussian-process fit is not bit-reproducible across environments even with a fixed seed, a GP-steered run cannot be reconstructed by re-asking the brain. Instead the loop records the coordinate the brain advised each pass (`advised_next_point` on the iteration event), surfaced alongside the verdict and deciding brain in the per-iteration read model, so a finished GP-steered run is reconstructable by reading the recorded decision trail. The trail also records how long the brain took to answer each pass, on the faulted path as well as the advised one. Time is the one cost a non-LLM brain spends: a Gaussian process draws neither tokens nor money, so the spend limits that bound an LLM-backed agent are blind to it, while at a beamline the scarce resource is the beam itself. A caller may declare a `SteeringBudget` (a pass count, a wall-clock allowance, or both) and the loop enforces it between passes, ending the Procedure the same way a brain-advised stop does, with a recorded reason naming which dimension ran out. The budget is per call, restated on every resume rather than accumulated: only the pass count is backstopped across an unbounded chain of resumes, by the same absolute ceiling that guards a brain which never advises stop; wall-clock time is not. Which layer should own a cap that DOES survive a resume, the brain, the Procedure, or the Allocation, is still open. The run stays classified "forward-only" (meaning not re-ask-reproducible, distinct from not-recorded and, now, distinct from not-resumable). That trail records what the brain DID; `SteeringDesignRecorded` records what it was ASKED, pinning the objective, the search space, the budget and the brain configuration once per conduct segment, before the first pass of that segment runs. Both sides are needed for a different and weaker criterion than replay: a reader who wants to draw an inference from a steered run has to know the rule that chose the points, and a rule that was never written down is not thereby harmless, it is unverifiable. A resume may change the budget, the brain, the bounds or the objective and those changes are recorded rather than refused; what it may not do is hand the brain a space that cannot express a coordinate already measured, which is an axis dropped, an axis added, or a categorical axis that has lost a value already tried. A crashed GP-steered run resumes through `conduct_until_advised_from`: it rebuilds the brain's history from the record, the measured value of each closed pass from a per-Procedure outcome logbook (an `Outcome` entry kind, sibling to the diagnostics log) and the advised coordinate from the iteration trail, then consults the brain only at the open frontier. Resume neither re-asks the brain for a pass already done nor re-drives hardware to a past position: the next move is an absolute one issued from wherever the instrument now sits, which reaches the same commanded point regardless of the current position, and direction-of-approach consistency (anti-backlash) is a per-move concern of the motion controller, not of resume. True sample path-dependence (thermal or radiation history, mechanical creep) is shaped by the whole trajectory rather than any single point, so it belongs in provenance, not in a resume-time re-drive; re-establishing one historical setpoint would neither be necessary nor sufficient. See [the recording spine and the optional execution edge](../../standards.md#the-recording-spine-and-the-optional-execution-edge). -**What a step does, and where deciding happens.** The step kinds divide by what they do with a value, and the division is worth stating because it is otherwise only legible one docstring at a time. `CaptureStep` and `CheckStep` observe: a capture is a pure read that stores the readback, a check is a read that also gates. `SetpointStep` and `ActionStep` act: a setpoint commands a value, an action performs a named operation. `ComputeStep` does both in one step, submitting a job and then reading back what it produced, which is the same pairing a `SetpointStep` makes optional through `verify`. Deciding is deliberately not a step kind. A brain runs between passes, not inside one: the loop walks a pass, hands the accumulated evidence to a `DecidePort`, and seeds the next pass from the advised point. That places observing and acting inside the step walk and deciding one layer above it. Reading the step kinds as though they should partition into observe, think and act would compress two altitudes into one: the brain's turn is a separate consult carrying its own record, not a step in the list. +**What a step does, and where deciding happens.** The step kinds divide by what they do with a value, and the division is worth stating because it is otherwise only legible one docstring at a time. `CaptureStep` and `CheckStep` observe: a capture is a pure read that stores the readback, a check is a read that also gates. `SetpointStep` and `ActionStep` act: a setpoint commands a value, an action performs a named operation. `ComputeStep` does both in one step, submitting a job and then reading back what it produced, which is the same pairing a `SetpointStep` makes optional through `verify`. Deciding is present at two altitudes, which is why it is not one step kind. A `CheckStep` decides in the small: it evaluates a criterion and gates the walk on the result, a deterministic rule brain running inside the list. The brain behind `DecidePort` decides in the large and sits outside the list, for two reasons a reader should not have to reconstruct. Its verdict can complete the whole Procedure, and no step outcome can, since a step succeeds or fails and neither of those means finish, successfully, now. And it is handed the full history across every pass, where a step reads only the current pass's captures bus. The loop brain's answer is about the sequence; a step's answer is about a point in it. So reading the five kinds as a clean partition into observe, think and act flattens two altitudes into one: thinking appears at both, and only its cheap deterministic form fits inside a step.
@@ -22,6 +22,7 @@ Out of scope {: .cora-kicker } - **Cross-aggregate resume cascade.** Pause and resume ship on the Procedure itself (see above), but resuming a parent Run does NOT cascade into its held phase Procedures; the off-diagonal guard only refuses the illegal direction. Coordinating both is a Layer-3 saga, deferred. +- **Steering within a pass.** A brain advises between passes, so a block's later steps cannot be chosen from its own earlier captures. Mechanically this would be a decide-shaped step over the captures bus that already exists; no pilot workload has asked for it. - **Verifying as a first-class FSM state.** Per-step Check happens inside Running synchronously; the standards corpus does not bless a separate Verifying state. - **Per-kind payload validation at the API.** The step `payload` body is `dict[str, Any]` today; per-kind Pydantic models land once pilot vocabulary settles. - **Asset-existence verification at register time.** `target_asset_ids` is taken at face value; existence and decommission-state gating runs at start-procedure time.