Give recipes a second, terminal-only closing-step list

apps/api/openapi.json

@@ -3874,6 +3874,14 @@
             ],
             "title": "Actuation Kind"
           },
+          "closing_failures": {
+            "description": "Every closing step that failed. Always empty on `acquisition_halt=True`. Never flips `succeeded`.",
+            "items": {
+              "$ref": "#/components/schemas/ConductorFailureResponse"
+            },
+            "title": "Closing Failures",
+            "type": "array"
+          },
           "completed_count": {
             "title": "Completed Count",
             "type": "integer"
@@ -3986,6 +3994,14 @@
             ],
             "title": "Actuation Kind"
           },
+          "closing_failures": {
+            "description": "Every closing step that failed. Always empty on `held=True`: closing runs only on a real terminal, never on Held. Never flips `succeeded`.",
+            "items": {
+              "$ref": "#/components/schemas/ConductorFailureResponse"
+            },
+            "title": "Closing Failures",
+            "type": "array"
+          },
           "completed_count": {
             "title": "Completed Count",
             "type": "integer"
@@ -4097,6 +4113,14 @@
             ],
             "title": "Actuation Kind"
           },
+          "closing_failures": {
+            "description": "Every closing step that failed. Isolated from `failure`: a closing failure never flips `succeeded`, and one closing step failing does not stop the rest of the closing walk. Empty when the recipe has no closing steps, or all ran clean.",
+            "items": {
+              "$ref": "#/components/schemas/ConductorFailureResponse"
+            },
+            "title": "Closing Failures",
+            "type": "array"
+          },
           "completed_count": {
             "title": "Completed Count",
             "type": "integer"
@@ -4137,7 +4161,7 @@
                 }
               ]
             },
-            "description": "Every control address this conduct wrote, in first-write order, carrying the last value written to each. CORA does not restore what it sets, and a halt returns at the failing step without running the recipe's remaining steps, so on a failed conduct this is what was left set with nothing having put it back. Reports what was WRITTEN, not what changed: a write whose value already matched the address still appears.",
+            "description": "Every control address this conduct wrote, in first-write order, carrying the last value written to each. Includes the recipe's closing steps, which run on a real terminal (Completed or Aborted). Reports what was WRITTEN, not what changed: a write whose value already matched the address still appears.",
             "title": "Substrate Writes",
             "type": "object"
           },
@@ -6645,6 +6669,12 @@
             "title": "Capability Id",
             "type": "string"
           },
+          "closing_steps": {
+            "additionalProperties": true,
+            "description": "Optional wire-format closing-step sequence, same shape as `steps`: `{steps: [...]}`. The Conductor walks these after `steps` ends on a real terminal (Completed or Aborted), never on a Held pause. Empty by default; most recipes have none.",
+            "title": "Closing Steps",
+            "type": "object"
+          },
           "name": {
             "description": "Display name for the new Recipe.",
             "maxLength": 200,
@@ -11846,13 +11876,18 @@
         "type": "string"
       },
       "RecipeResponse": {
-        "description": "Read-side DTO at the API boundary.\n\nCarries primitives, not domain VOs. `status` is the StrEnum's\nstring value (Defined / Versioned / Deprecated). `version` is\nthe operator-supplied label of the most recent `version_recipe`\ncall (null until first version). `steps` is the wire-format\ndict (BindingRef sentinels serialize as `{__binding__: name}`).\n`replaced_by_recipe_id` is null on Defined / Versioned /\nDeprecated-without-replacement; populated when a deprecation\nsupplied a successor pointer. `created_at` / `versioned_at` /\n`deprecated_at` are projection-sourced lifecycle timestamps\n(Path C); see module docstring for null semantics.",
+        "description": "Read-side DTO at the API boundary.\n\nCarries primitives, not domain VOs. `status` is the StrEnum's\nstring value (Defined / Versioned / Deprecated). `version` is\nthe operator-supplied label of the most recent `version_recipe`\ncall (null until first version). `steps` is the wire-format\ndict (BindingRef sentinels serialize as `{__binding__: name}`).\n`replaced_by_recipe_id` is null on Defined / Versioned /\nDeprecated-without-replacement; populated when a deprecation\nsupplied a successor pointer. `created_at` / `versioned_at` /\n`deprecated_at` are projection-sourced lifecycle timestamps\n(Path C); see module docstring for null semantics. `closing_steps`\nmirrors `steps`' wire shape; empty for the common case of a recipe\nwith no closing steps.",
         "properties": {
           "capability_id": {
             "format": "uuid",
             "title": "Capability Id",
             "type": "string"
           },
+          "closing_steps": {
+            "additionalProperties": true,
+            "title": "Closing Steps",
+            "type": "object"
+          },
           "created_at": {
             "anyOf": [
               {
@@ -11939,6 +11974,7 @@
           "status",
           "version",
           "steps",
+          "closing_steps",
           "replaced_by_recipe_id"
         ],
         "title": "RecipeResponse",
@@ -17230,6 +17266,12 @@
       "VersionRecipeRequest": {
         "description": "Body for `POST /recipes/{recipe_id}/version`.",
         "properties": {
+          "closing_steps": {
+            "additionalProperties": true,
+            "description": "Replacement closing-step sequence, same shape as `steps` (wholesale replace alongside it). Empty by default.",
+            "title": "Closing Steps",
+            "type": "object"
+          },
           "steps": {
             "additionalProperties": true,
             "description": "Replacement step sequence for the new version (wholesale replace; the prior steps are dropped). BindingRef sentinels are re-validated against the CURRENT Capability.parameters_schema.",
@@ -40947,7 +40989,7 @@
             "description": "Procedure is not in `Held` status, OR its parent Run is itself `Held` (off-diagonal guard)."
           },
           "422": {
-            "description": "Path parameter or request body failed schema validation."
+            "description": "Path parameter or request body failed schema validation, OR a closing step's CaptureRef names a capture only a pre-boundary main step declares (ClosingCaptureBeforeBoundaryError)."
           },
           "500": {
             "content": {

apps/api/src/cora/infrastructure/record_export/_dispositions.py

@@ -1493,6 +1493,21 @@
     },
     "RecipeDefined": {
         "capability_id": "token:uuid",
+        "closing": {
+            "address": "drop:text",
+            "capture_name": "drop:text",
+            "command": "drop:text",
+            "criterion": "drop:opaque",
+            "input_uris": "by-value",
+            "name": "drop:text",
+            "output_ref_name": "drop:text",
+            "output_uri": "drop:text",
+            "parameters": "drop:opaque",
+            "params": "drop:opaque",
+            "timeout_s": "keep:number",
+            "value": "by-value",
+            "verify": "keep:number",
+        },
         "name": "drop:text",
         "occurred_at": "keep:time",
         "recipe_id": "token:uuid",
@@ -1532,6 +1547,21 @@
         "steps_hash": "drop:text",
     },
     "RecipeVersioned": {
+        "closing": {
+            "address": "drop:text",
+            "capture_name": "drop:text",
+            "command": "drop:text",
+            "criterion": "drop:opaque",
+            "input_uris": "by-value",
+            "name": "drop:text",
+            "output_ref_name": "drop:text",
+            "output_uri": "drop:text",
+            "parameters": "drop:opaque",
+            "params": "drop:opaque",
+            "timeout_s": "keep:number",
+            "value": "by-value",
+            "verify": "keep:number",
+        },
         "occurred_at": "keep:time",
         "recipe_id": "token:uuid",
         "steps": {
@@ -1554,6 +1584,7 @@
     "ResolvedStepsRecorded": {
         "occurred_at": "keep:time",
         "procedure_id": "token:uuid",
+        "resolved_closing_steps": "drop:opaque",
         "resolved_steps": "drop:opaque",
         "step_count": "keep:number",
     },

apps/api/src/cora/operation/_conduct_preparation.py

@@ -66,6 +66,7 @@ def decide_resolved_steps_recorded(
     resolved_steps: Sequence[Mapping[str, Any]],
     *,
     now: datetime,
+    resolved_closing_steps: Sequence[Mapping[str, Any]] = (),
 ) -> list[ResolvedStepsRecorded]:
     """Pin the resolved step list iff the Procedure is pre-conduct (Defined).
 
@@ -77,16 +78,22 @@ def decide_resolved_steps_recorded(
     normal lifecycle failure, preserving the conduct route's failures-in-body
     contract instead of raising a fresh HTTP error here. Kept as a pure
     function so the decision is unit-testable without an event store.
+
+    `resolved_closing_steps` is the SAME resolution applied to the Recipe's
+    closing steps, pinned separately (not flattened into `resolved_steps`)
+    so a resume boundary can never land inside the closing region.
     """
     if state is None or state.status is not ProcedureStatus.DEFINED:
         return []
     steps = tuple(dict(step) for step in resolved_steps)
+    closing_steps = tuple(dict(step) for step in resolved_closing_steps)
     return [
         ResolvedStepsRecorded(
             procedure_id=state.id,
             resolved_steps=steps,
-            step_count=len(steps),
+            step_count=len(steps) + len(closing_steps),
             occurred_at=now,
+            resolved_closing_steps=closing_steps,
         )
     ]
 
@@ -102,16 +109,22 @@ async def resolve_and_pin_conduct_steps(
     principal_id: UUID,
     correlation_id: UUID,
     causation_id: UUID | None,
-) -> tuple[Step, ...]:
+) -> tuple[tuple[Step, ...], tuple[Step, ...]]:
     """Resolve the final conduct step list + pin it as `ResolvedStepsRecorded`.
 
     The shared pre-Conductor work for `conduct` / `conduct_or_hold`: recipe
     re-expansion (recipe-driven Procedures) -> pseudoaxis constituent
-    expansion (Run-phase Procedures) -> pin. Returns the resolved steps to
-    hand to the Conductor. `command_name` rides the pinned event's metadata.
+    expansion (Run-phase Procedures) -> pin. Returns `(steps, closing_steps)`
+    to hand to the Conductor. `command_name` rides the pinned event's
+    metadata.
+
+    A legacy (non-recipe-driven) Procedure has no closing steps: `caller_steps`
+    is an inline list with no separate closing half, so `closing_steps` is
+    always `()` on that path.
     """
+    closing_steps: tuple[Step, ...] = ()
     if procedure.recipe_id is not None:
-        steps = await _re_expand_steps(
+        steps, closing_steps = await _re_expand_steps(
             procedure_id=procedure.id,
             recipe_id=procedure.recipe_id,
             caller_steps=caller_steps,
@@ -146,13 +159,21 @@ def _resolve_constituents(asset_id: UUID) -> tuple[UUID, ...]:
     # Pre-Conductor PseudoAxis expansion: rewrite any virtual-axis SetpointStep
     # into N sequential constituent SetpointSteps so the Conductor's dispatch
     # loop walks the constituents in declared order. ActionStep / CheckStep
-    # pass through unchanged ([[project-pseudoaxis-design]] v3).
+    # pass through unchanged ([[project-pseudoaxis-design]] v3). Closing steps
+    # get the SAME expansion, or a pseudoaxis closing setpoint would reach the
+    # Conductor unresolved.
     steps = await expansion_port.expand_pseudoaxis(
         steps,
         event_store=deps.event_store,
         correlation_id=correlation_id,
         constituent_resolver=constituent_resolver,
     )
+    closing_steps = await expansion_port.expand_pseudoaxis(
+        closing_steps,
+        event_store=deps.event_store,
+        correlation_id=correlation_id,
+        constituent_resolver=constituent_resolver,
+    )
 
     # Pin the resolved step list (after recipe + pseudoaxis expansion) BEFORE
     # conducting, so a future resume replays this exact list. The helper emits
@@ -163,6 +184,7 @@ def _resolve_constituents(asset_id: UUID) -> tuple[UUID, ...]:
         procedure,
         tuple(step_to_payload(step) for step in steps),
         now=deps.clock.now(),
+        resolved_closing_steps=tuple(step_to_payload(step) for step in closing_steps),
     )
     if resolved_steps_events:
         _, current_version = await deps.event_store.load(
@@ -187,7 +209,7 @@ def _resolve_constituents(asset_id: UUID) -> tuple[UUID, ...]:
             ],
         )
 
-    return steps
+    return steps, closing_steps
 
 
 async def _re_expand_steps(
@@ -198,7 +220,7 @@ async def _re_expand_steps(
     stored_events: list[StoredEvent],
     event_store: EventStore,
     expansion_port: RecipeExpander,
-) -> tuple[Step, ...]:
+) -> tuple[tuple[Step, ...], tuple[Step, ...]]:
     """Run the recipe-replay gate per [[project-run-procedure-replay-design]].
 
     Six steps: reject non-empty caller steps -> find_recipe_expansion_record
@@ -209,7 +231,8 @@ async def _re_expand_steps(
     propagates from helper) -> load_capability + reject Deprecated
     (raise ProcedureBoundCapabilityDeprecatedError, symmetric to
     start_run's RunBoundPlanDeprecatedError) -> verify_bindings_hash ->
-    expand -> verify_steps_hash -> return the re-expanded tuple.
+    expand (both `recipe.steps` and `recipe.closing_steps`) ->
+    verify_steps_hash (one combined pin) -> return `(steps, closing_steps)`.
     """
     if list(caller_steps):
         raise ProcedureStepsForbiddenForRecipeDrivenError(procedure_id)
@@ -244,6 +267,8 @@ async def _re_expand_steps(
         raise ProcedureBoundCapabilityDeprecatedError(procedure_id, recipe.capability_id)
 
     verify_bindings_hash(procedure_id, pins)
-    expanded = expansion_port.expand(recipe.steps, dict(pins.bindings))
-    verify_steps_hash(procedure_id, expanded, pins)
-    return expanded
+    bindings_dict = dict(pins.bindings)
+    expanded = expansion_port.expand(recipe.steps, bindings_dict)
+    expanded_closing = expansion_port.expand(recipe.closing_steps, bindings_dict)
+    verify_steps_hash(procedure_id, expanded, pins, closing_steps=expanded_closing)
+    return expanded, expanded_closing

apps/api/src/cora/operation/_conduct_wire.py

@@ -22,7 +22,7 @@
 tuple for the in-process Conductor.
 """
 
-from collections.abc import Mapping
+from collections.abc import Mapping, Sequence
 from typing import Annotated, Any, Literal, cast
 
 from pydantic import BaseModel, Field
@@ -192,3 +192,15 @@ def substrate_writes_to_wire(
     `result_to_wire` (conduct / conduct_or_hold / conduct_from) now that a
     third call site would otherwise repeat it."""
     return {k: list(v) if isinstance(v, tuple) else v for k, v in substrate_writes.items()}
+
+
+def closing_failures_to_wire(
+    closing_failures: Sequence[ConductorFailure],
+) -> list[ConductorFailureResponse]:
+    """Project `ConductorResult.closing_failures` onto its JSON wire shape.
+
+    Reuses `failure_to_wire` per entry -- a closing failure has the exact
+    same shape as the main-walk `failure`, just isolated into its own list
+    rather than halting the walk. Empty by default: most conducts have no
+    closing steps at all, or every closing step ran clean."""
+    return [failure_to_wire(f) for f in closing_failures]

apps/api/src/cora/operation/_recipe_expansion/__init__.py

@@ -7,7 +7,8 @@
 that replay / verify that resolution:
 
   - `_expand`: the pure `expand(steps, bindings) -> Step list` substitution kernel
-    (+ `steps_to_wire` / `canonical_json_bytes` for provenance hashing).
+    (+ `steps_to_wire` / `steps_to_wire_with_closing` / `canonical_json_bytes`
+    for provenance hashing).
   - `_replay`: locate + verify the genesis `RecipeExpansionRecorded` provenance
     on the `conduct_procedure` path (re-expand and compare pinned hashes).
   - `_resolved_steps_replay`: locate the pinned `ResolvedStepsRecorded` provenance
@@ -21,6 +22,7 @@
     canonical_json_bytes,
     expand,
     steps_to_wire,
+    steps_to_wire_with_closing,
 )
 from cora.operation._recipe_expansion._replay import (
     MismatchField,
@@ -43,6 +45,7 @@
     "find_resolved_steps_record",
     "pins_from_payload",
     "steps_to_wire",
+    "steps_to_wire_with_closing",
     "verify_bindings_hash",
     "verify_steps_hash",
 ]

apps/api/src/cora/operation/_recipe_expansion/_expand.py

@@ -229,4 +229,27 @@ def steps_to_wire(steps: tuple[Step, ...]) -> list[dict[str, Any]]:
     return [_step_to_wire(step) for step in steps]
 
 
-__all__ = ["canonical_json_bytes", "expand", "steps_to_wire"]
+def steps_to_wire_with_closing(
+    steps: tuple[Step, ...], closing_steps: tuple[Step, ...]
+) -> list[dict[str, Any]]:
+    """Canonical hash form for main + closing steps COMBINED, one pin.
+
+    Closing entries are tagged `"closing": true` by THIS composing
+    function, not inside `_step_to_wire` (which receives a bare `Step`
+    with no partition context). An empty `closing_steps` returns
+    EXACTLY `steps_to_wire(steps)` with nothing appended, so every
+    recipe that has never used closing steps keeps its existing
+    `steps_hash` byte-identical -- same precedent as `CheckStep.timeout_s`
+    being emitted only when set (see
+    `test_a_deadlineless_check_hashes_the_same_as_it_always_did`).
+    """
+    main = steps_to_wire(steps)
+    if not closing_steps:
+        return main
+    closing = steps_to_wire(closing_steps)
+    for entry in closing:
+        entry["closing"] = True
+    return main + closing
+
+
+__all__ = ["canonical_json_bytes", "expand", "steps_to_wire", "steps_to_wire_with_closing"]