Let a compute step's parameters carry a capture or steering reference

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

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,
         }

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
@@ -447,6 +451,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 +761,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 +795,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
 
@@ -4248,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,
@@ -4265,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.
@@ -4899,7 +5037,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 +5099,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"),
         )