From c173952b984c8a4c12d72d7393d82bc7a877affc Mon Sep 17 00:00:00 2001 From: xmap <16776958+xmap@users.noreply.github.com> Date: Thu, 10 Sep 2026 07:29:39 -0500 Subject: [PATCH 1/4] Let a Procedure record which concerns are holding it, not just that one is `ProcedureStatus.HELD` is one bit. That was adequate while a hold had a single author, and became a fault once four independent concerns could each park the same conduct: two operator surfaces, a recoverable step failure, and a steering driver going non-ACTIVE. A second holder arriving at an already-Held Procedure is refused and records nothing, and the first holder's resume then restarts the conduct with the second's cause unenforced. Which concern survives depends only on arrival order. Run hit this exact fault, fixed it with cause-scoped claims, and wrote it up in project_hold_claim_ordering_fault. This ports the aggregate half of that design rather than inventing a second one: `ProcedureHeld` carries `claim_id` + a closed `cause`, `ProcedureResumed` carries `released_claim_id`, an audit-only `ProcedureHoldClaimReleased` discharges a claim without resuming, and claims fold onto `Procedure.hold_claims`. Causes are the three concerns that actually park a conduct today, not Run's four: `operator`, `step-fault`, `driver-stand-down`. The last two are separate because they are discharged by different things, the equipment recovering versus the driver being reinstated, so they must be able to hold at once. Nothing changes behaviour yet. No decider enforces resume-only-when-last and no writer passes a cause, so every hold still folds to the single legacy claim and every resume still clears it. That is deliberate: this slice makes the aggregate able to represent the truth, and the enforcement follows in its own slice with its own review. Legacy streams replay to exactly their old one-bit meaning. A claimless hold folds to LEGACY_CLAIM_ID, repeated claimless holds collapse rather than accumulate, and a bare resume clears everything. Two guards earned their keep. The evolver carry-forward fitness test caught the new state field immediately, which is the drift catcher working as designed; its registry is updated with the arms that write the field. And the generated record-export disposition table needed regenerating, so the new keys are dispositioned exactly as Run's are rather than diverging. The release arm uses `replace` rather than a hand-listed constructor, deliberately: re-listing seventeen fields to change one is how a field gets silently dropped, which is the bug class this slice's own field nearly fell into. Verified by mutation: making a release clear every claim fails three of the new tests, including the one pinning that a Procedure stays Held while another concern holds it. Co-Authored-By: Claude Opus 5 (1M context) --- .../record_export/_dispositions.py | 9 + .../aggregates/procedure/__init__.py | 14 + .../operation/aggregates/procedure/events.py | 127 +++++++++ .../operation/aggregates/procedure/evolver.py | 69 ++++- .../operation/aggregates/procedure/state.py | 12 + .../test_procedure_evolver_carry_forward.py | 14 + .../operation/test_hold_procedure_handler.py | 4 + .../unit/operation/test_procedure_events.py | 3 + .../operation/test_procedure_hold_claims.py | 246 ++++++++++++++++++ .../test_resume_procedure_handler.py | 4 + 10 files changed, 500 insertions(+), 2 deletions(-) create mode 100644 apps/api/tests/unit/operation/test_procedure_hold_claims.py diff --git a/apps/api/src/cora/infrastructure/record_export/_dispositions.py b/apps/api/src/cora/infrastructure/record_export/_dispositions.py index b40cb2b1787..26205e7e937 100644 --- a/apps/api/src/cora/infrastructure/record_export/_dispositions.py +++ b/apps/api/src/cora/infrastructure/record_export/_dispositions.py @@ -1432,11 +1432,19 @@ }, "ProcedureHeld": { "actuation_kind": "drop:text", + "cause": "drop:text", + "claim_id": "token:uuid", "decided_by_decision_id": "token:uuid", "occurred_at": "keep:time", "procedure_id": "token:uuid", "reason": "drop:text", }, + "ProcedureHoldClaimReleased": { + "cause": "drop:text", + "claim_id": "token:uuid", + "occurred_at": "keep:time", + "procedure_id": "token:uuid", + }, "ProcedureIterationEnded": { "advice_latency_ms": "keep:number", "advised_next_point": "drop:opaque", @@ -1486,6 +1494,7 @@ "occurred_at": "keep:time", "procedure_id": "token:uuid", "re_establishment_boundary": "keep:number", + "released_claim_id": "token:uuid", }, "ProcedureStarted": { "beam_requirement": "keep:enum:BeamRequirement", diff --git a/apps/api/src/cora/operation/aggregates/procedure/__init__.py b/apps/api/src/cora/operation/aggregates/procedure/__init__.py index f35d12069a3..2a5146981f0 100644 --- a/apps/api/src/cora/operation/aggregates/procedure/__init__.py +++ b/apps/api/src/cora/operation/aggregates/procedure/__init__.py @@ -25,12 +25,19 @@ PostgresOutcomeStore, ) from cora.operation.aggregates.procedure.events import ( + HOLD_CAUSE_DRIVER_STAND_DOWN, + HOLD_CAUSE_OPERATOR, + HOLD_CAUSE_STEP_FAULT, + HOLD_CAUSES, + LEGACY_CAUSE, + LEGACY_CLAIM_ID, ProcedureAborted, ProcedureActivitiesLogbookOpened, ProcedureCompleted, ProcedureDiagnosticLogbookOpened, ProcedureEvent, ProcedureHeld, + ProcedureHoldClaimReleased, ProcedureIterationEnded, ProcedureIterationStarted, ProcedureOutcomeLogbookOpened, @@ -114,6 +121,12 @@ __all__ = [ "DIAGNOSTIC_LOGBOOK_SCHEMA", + "HOLD_CAUSES", + "HOLD_CAUSE_DRIVER_STAND_DOWN", + "HOLD_CAUSE_OPERATOR", + "HOLD_CAUSE_STEP_FAULT", + "LEGACY_CAUSE", + "LEGACY_CLAIM_ID", "LOGBOOK_KIND_ACTIVITY", "LOGBOOK_KIND_DIAGNOSTIC", "LOGBOOK_KIND_OUTCOME", @@ -167,6 +180,7 @@ "ProcedureEnclosureCoverageMismatchError", "ProcedureEvent", "ProcedureHeld", + "ProcedureHoldClaimReleased", "ProcedureHoldReason", "ProcedureIterationEnded", "ProcedureIterationLimitReachedError", diff --git a/apps/api/src/cora/operation/aggregates/procedure/events.py b/apps/api/src/cora/operation/aggregates/procedure/events.py index 1e350823112..7d366d8343d 100644 --- a/apps/api/src/cora/operation/aggregates/procedure/events.py +++ b/apps/api/src/cora/operation/aggregates/procedure/events.py @@ -441,6 +441,43 @@ class ProcedureAborted: actuation_kind: str | None = None +HOLD_CAUSE_OPERATOR = "operator" +"""A person asked for the pause, through the REST route or the MCP tool.""" + +HOLD_CAUSE_STEP_FAULT = "step-fault" +"""The Conductor parked the conduct because a step failed recoverably +(`conduct_or_hold`). Released by whatever re-establishes the step.""" + +HOLD_CAUSE_DRIVER_STAND_DOWN = "driver-stand-down" +"""The Conductor parked a steered loop because its steering driver went +non-ACTIVE (`_hold_driver_stood_down`). A DIFFERENT concern from a step +fault: it is discharged by the driver being reinstated, not by the +equipment recovering, so the two must be able to hold at the same time.""" + +HOLD_CAUSES: frozenset[str] = frozenset( + { + HOLD_CAUSE_OPERATOR, + HOLD_CAUSE_STEP_FAULT, + HOLD_CAUSE_DRIVER_STAND_DOWN, + } +) +"""The closed set of concerns that may hold a Procedure. Coarse on purpose: +a cause names WHICH concern is holding, not why in prose. The prose stays on +`ProcedureHeld.reason`. Mirrors `HOLD_CAUSES` on Run.""" + +LEGACY_CLAIM_ID = UUID("01900000-0000-7000-8000-00000001dead") +"""The single claim a pre-claim `ProcedureHeld` folds to, so a legacy stream +replays to exactly its old one-bit meaning. Mirrors Run's sentinel.""" + +LEGACY_CAUSE = "legacy-unscoped" +"""The cause a pre-claim `ProcedureHeld` folds to.""" + + +def _optional_uuid(raw: object) -> UUID | None: + """Parse an optional UUID payload key, absent on pre-claim streams.""" + return UUID(str(raw)) if raw is not None else None + + @dataclass(frozen=True) class ProcedureHeld: """A Procedure conduct was operator-paused (Running -> Held). @@ -482,6 +519,17 @@ class ProcedureHeld: occurred_at: datetime decided_by_decision_id: UUID | None = None actuation_kind: str | None = None + claim_id: UUID | None = None + """Which claim this hold places. `Held` alone is one bit and cannot say + who is holding or whether anyone else still is, which is a safety fault + once independent concerns can each park the same Procedure: a second + holder arriving at an already-Held Procedure cannot record its intent, + and the FIRST holder's resume then restarts the conduct with the + second's cause unenforced. None on a legacy stream, folding to + `LEGACY_CLAIM_ID`.""" + cause: str | None = None + """Which concern is holding, from `HOLD_CAUSES`. None on a legacy + stream, folding to `LEGACY_CAUSE`.""" @dataclass(frozen=True) @@ -513,6 +561,41 @@ class ProcedureResumed: re_establishment_boundary: int occurred_at: datetime decided_by_decision_id: UUID | None = None + released_claim_id: UUID | None = None + """The claim this resume discharges. Legal ONLY when it is the last + active claim, which is what makes RUNNING mean "no concern is holding + this" rather than "whoever spoke last is done". None on a legacy + stream, where a bare resume clears every claim and so replays the old + one-bit behaviour exactly.""" + + +@dataclass(frozen=True) +class ProcedureHoldClaimReleased: + """One hold claim was discharged while other claims remain active. + + Audit-only: the evolver returns prior state with the claim removed and + the status untouched, so the Procedure stays `Held`. + + This is the event that makes the hold algebra compositional. Without it + a concern has exactly two ways to stop holding, both wrong when it is + not the only holder: append `ProcedureResumed` and restart a conduct + other concerns still want parked, or append nothing and hold forever. + A releaser picks by folding the active claims first: + + - own claim is the ONLY active one -> `ProcedureResumed(released_claim_id)` + - other claims remain -> `ProcedureHoldClaimReleased(claim_id)` + + `claim_id` must name an active claim; releasing an unknown or + already-released claim is a no-op at the fold and is rejected by the + deciders rather than silently absorbed. Mirrors Run's `HoldClaimReleased`, + named with the `Procedure` prefix because this BC's event names are + aggregate-qualified. + """ + + procedure_id: UUID + claim_id: UUID + cause: str + occurred_at: datetime @dataclass(frozen=True) @@ -816,6 +899,7 @@ class SteeringDesignRecorded: | ProcedureTruncated | ProcedureHeld | ProcedureResumed + | ProcedureHoldClaimReleased | ProcedureActivitiesLogbookOpened | ProcedureDiagnosticLogbookOpened | ProcedureOutcomeLogbookOpened @@ -942,6 +1026,8 @@ def to_payload(event: ProcedureEvent) -> dict[str, Any]: occurred_at=occurred_at, decided_by_decision_id=decided_by_decision_id, actuation_kind=actuation_kind, + claim_id=claim_id, + cause=cause, ): return { "procedure_id": str(procedure_id), @@ -951,12 +1037,15 @@ def to_payload(event: ProcedureEvent) -> dict[str, Any]: ), "occurred_at": occurred_at.isoformat(), "actuation_kind": actuation_kind, + "claim_id": str(claim_id) if claim_id is not None else None, + "cause": cause, } case ProcedureResumed( procedure_id=procedure_id, re_establishment_boundary=re_establishment_boundary, occurred_at=occurred_at, decided_by_decision_id=decided_by_decision_id, + released_claim_id=released_claim_id, ): return { "procedure_id": str(procedure_id), @@ -965,6 +1054,21 @@ def to_payload(event: ProcedureEvent) -> dict[str, Any]: str(decided_by_decision_id) if decided_by_decision_id is not None else None ), "occurred_at": occurred_at.isoformat(), + "released_claim_id": ( + str(released_claim_id) if released_claim_id is not None else None + ), + } + case ProcedureHoldClaimReleased( + procedure_id=procedure_id, + claim_id=claim_id, + cause=cause, + occurred_at=occurred_at, + ): + return { + "procedure_id": str(procedure_id), + "claim_id": str(claim_id), + "cause": cause, + "occurred_at": occurred_at.isoformat(), } case ProcedureActivitiesLogbookOpened( procedure_id=procedure_id, @@ -1338,6 +1442,10 @@ def _build_held() -> ProcedureHeld: occurred_at=datetime.fromisoformat(payload["occurred_at"]), # Additive: pre-activation streams omit the key -> None. actuation_kind=payload.get("actuation_kind"), + # Additive: pre-claim streams omit both -> None, which the + # fold reads as the single LEGACY_CLAIM_ID claim. + claim_id=_optional_uuid(payload.get("claim_id")), + cause=payload.get("cause"), ) return deserialize_or_raise("ProcedureHeld", _build_held) @@ -1352,9 +1460,21 @@ def _build_resumed() -> ProcedureResumed: UUID(raw_decided_by) if raw_decided_by is not None else None ), occurred_at=datetime.fromisoformat(payload["occurred_at"]), + # Additive: a bare legacy resume clears every claim. + released_claim_id=_optional_uuid(payload.get("released_claim_id")), ) return deserialize_or_raise("ProcedureResumed", _build_resumed) + case "ProcedureHoldClaimReleased": + return deserialize_or_raise( + "ProcedureHoldClaimReleased", + lambda: ProcedureHoldClaimReleased( + procedure_id=UUID(payload["procedure_id"]), + claim_id=UUID(payload["claim_id"]), + cause=payload["cause"], + occurred_at=datetime.fromisoformat(payload["occurred_at"]), + ), + ) case "ProcedureActivitiesLogbookOpened": return deserialize_or_raise( "ProcedureActivitiesLogbookOpened", @@ -1490,12 +1610,19 @@ def _build_resumed() -> ProcedureResumed: __all__ = [ + "HOLD_CAUSES", + "HOLD_CAUSE_DRIVER_STAND_DOWN", + "HOLD_CAUSE_OPERATOR", + "HOLD_CAUSE_STEP_FAULT", + "LEGACY_CAUSE", + "LEGACY_CLAIM_ID", "ProcedureAborted", "ProcedureActivitiesLogbookOpened", "ProcedureCompleted", "ProcedureDiagnosticLogbookOpened", "ProcedureEvent", "ProcedureHeld", + "ProcedureHoldClaimReleased", "ProcedureIterationEnded", "ProcedureIterationStarted", "ProcedureOutcomeLogbookOpened", diff --git a/apps/api/src/cora/operation/aggregates/procedure/evolver.py b/apps/api/src/cora/operation/aggregates/procedure/evolver.py index fe519d82fe8..d6e86aa6382 100644 --- a/apps/api/src/cora/operation/aggregates/procedure/evolver.py +++ b/apps/api/src/cora/operation/aggregates/procedure/evolver.py @@ -61,16 +61,21 @@ """ from collections.abc import Sequence +from dataclasses import replace from typing import assert_never +from uuid import UUID from cora.infrastructure.evolver import require_state from cora.operation.aggregates.procedure.events import ( + LEGACY_CAUSE, + LEGACY_CLAIM_ID, ProcedureAborted, ProcedureActivitiesLogbookOpened, ProcedureCompleted, ProcedureDiagnosticLogbookOpened, ProcedureEvent, ProcedureHeld, + ProcedureHoldClaimReleased, ProcedureIterationEnded, ProcedureIterationStarted, ProcedureOutcomeLogbookOpened, @@ -90,6 +95,39 @@ ) +def _with_claim( + claims: tuple[tuple[UUID, str], ...], + claim_id: UUID | None, + cause: str | None, +) -> tuple[tuple[UUID, str], ...]: + """Add one hold claim, idempotent on an already-active claim id. + + A `ProcedureHeld` written before holds were cause-scoped carries no + `claim_id`; it folds to the single `LEGACY_CLAIM_ID` claim so repeated + legacy holds collapse to one rather than accumulating, which is what makes + a legacy stream replay to its original one-bit meaning. + """ + key = claim_id if claim_id is not None else LEGACY_CLAIM_ID + if any(existing == key for existing, _ in claims): + return claims + return (*claims, (key, cause if cause is not None else LEGACY_CAUSE)) + + +def _without_claim( + claims: tuple[tuple[UUID, str], ...], + claim_id: UUID | None, +) -> tuple[tuple[UUID, str], ...]: + """Drop one hold claim; `None` clears every claim (legacy bare resume). + + Dropping a claim that is not active is a no-op rather than an error: the + evolver folds whatever the stream says and leaves rejection to the deciders, + which is what keeps replay total over any historical stream. + """ + if claim_id is None: + return () + return tuple((cid, cause) for cid, cause in claims if cid != claim_id) + + def evolve(state: Procedure | None, event: ProcedureEvent) -> Procedure: """Apply one event to the current state.""" match event: @@ -117,6 +155,7 @@ def evolve(state: Procedure | None, event: ProcedureEvent) -> Procedure: outcome_logbook_id=None, capability_id=capability_id, recipe_id=recipe_id, + hold_claims=(), current_iteration_index=None, iteration_count=0, consecutive_unconverged_iterations=0, @@ -137,6 +176,7 @@ def evolve(state: Procedure | None, event: ProcedureEvent) -> Procedure: outcome_logbook_id=prior.outcome_logbook_id, capability_id=prior.capability_id, recipe_id=prior.recipe_id, + hold_claims=prior.hold_claims, current_iteration_index=prior.current_iteration_index, iteration_count=prior.iteration_count, consecutive_unconverged_iterations=prior.consecutive_unconverged_iterations, @@ -160,6 +200,7 @@ def evolve(state: Procedure | None, event: ProcedureEvent) -> Procedure: outcome_logbook_id=prior.outcome_logbook_id, capability_id=prior.capability_id, recipe_id=prior.recipe_id, + hold_claims=(), current_iteration_index=prior.current_iteration_index, iteration_count=prior.iteration_count, consecutive_unconverged_iterations=prior.consecutive_unconverged_iterations, @@ -184,6 +225,7 @@ def evolve(state: Procedure | None, event: ProcedureEvent) -> Procedure: outcome_logbook_id=prior.outcome_logbook_id, capability_id=prior.capability_id, recipe_id=prior.recipe_id, + hold_claims=(), current_iteration_index=prior.current_iteration_index, iteration_count=prior.iteration_count, consecutive_unconverged_iterations=prior.consecutive_unconverged_iterations, @@ -208,6 +250,7 @@ def evolve(state: Procedure | None, event: ProcedureEvent) -> Procedure: outcome_logbook_id=prior.outcome_logbook_id, capability_id=prior.capability_id, recipe_id=prior.recipe_id, + hold_claims=(), current_iteration_index=prior.current_iteration_index, iteration_count=prior.iteration_count, consecutive_unconverged_iterations=prior.consecutive_unconverged_iterations, @@ -217,7 +260,11 @@ def evolve(state: Procedure | None, event: ProcedureEvent) -> Procedure: beam_requirement=prior.beam_requirement, actuation_kind=prior.actuation_kind, ) - case ProcedureHeld(actuation_kind=held_actuation_kind): + case ProcedureHeld( + actuation_kind=held_actuation_kind, + claim_id=held_claim_id, + cause=held_cause, + ): # Operator-pause transition (Running -> Held). Status-only change; # every non-status field carries verbatim from prior (especially # the iteration denorms). Mirrors RunHeld. EXCEPT actuation_kind: @@ -240,6 +287,7 @@ def evolve(state: Procedure | None, event: ProcedureEvent) -> Procedure: outcome_logbook_id=prior.outcome_logbook_id, capability_id=prior.capability_id, recipe_id=prior.recipe_id, + hold_claims=_with_claim(prior.hold_claims, held_claim_id, held_cause), current_iteration_index=prior.current_iteration_index, iteration_count=prior.iteration_count, consecutive_unconverged_iterations=prior.consecutive_unconverged_iterations, @@ -249,7 +297,7 @@ def evolve(state: Procedure | None, event: ProcedureEvent) -> Procedure: beam_requirement=prior.beam_requirement, actuation_kind=merge_actuation_kinds(prior.actuation_kind, held_actuation_kind), ) - case ProcedureResumed(): + case ProcedureResumed(released_claim_id=released_claim_id): # Resume transition (Held -> Running). Status-only change; every # non-status field carries verbatim from prior. The # re_establishment_boundary rides the event for the Conductor's @@ -267,6 +315,7 @@ def evolve(state: Procedure | None, event: ProcedureEvent) -> Procedure: outcome_logbook_id=prior.outcome_logbook_id, capability_id=prior.capability_id, recipe_id=prior.recipe_id, + hold_claims=_without_claim(prior.hold_claims, released_claim_id), current_iteration_index=prior.current_iteration_index, iteration_count=prior.iteration_count, consecutive_unconverged_iterations=prior.consecutive_unconverged_iterations, @@ -276,6 +325,17 @@ def evolve(state: Procedure | None, event: ProcedureEvent) -> Procedure: beam_requirement=prior.beam_requirement, actuation_kind=prior.actuation_kind, ) + case ProcedureHoldClaimReleased(claim_id=released_claim_id): + # Audit-only: one concern discharged its claim while others still + # hold, so the Procedure STAYS Held. `replace` rather than a + # hand-listed constructor precisely because this arm changes one + # field: re-listing seventeen is how a field gets silently + # dropped, which is the bug class this whole slice exists to fix. + prior = require_state(state, "ProcedureHoldClaimReleased") + return replace( + prior, + hold_claims=_without_claim(prior.hold_claims, released_claim_id), + ) case ProcedureActivitiesLogbookOpened(logbook_id=logbook_id): # Lazy open-on-first-write: preserve all # prior state, set activity_logbook_id. Status NOT touched -- the @@ -294,6 +354,7 @@ def evolve(state: Procedure | None, event: ProcedureEvent) -> Procedure: outcome_logbook_id=prior.outcome_logbook_id, capability_id=prior.capability_id, recipe_id=prior.recipe_id, + hold_claims=prior.hold_claims, current_iteration_index=prior.current_iteration_index, iteration_count=prior.iteration_count, consecutive_unconverged_iterations=prior.consecutive_unconverged_iterations, @@ -321,6 +382,7 @@ def evolve(state: Procedure | None, event: ProcedureEvent) -> Procedure: outcome_logbook_id=prior.outcome_logbook_id, capability_id=prior.capability_id, recipe_id=prior.recipe_id, + hold_claims=prior.hold_claims, current_iteration_index=prior.current_iteration_index, iteration_count=prior.iteration_count, consecutive_unconverged_iterations=prior.consecutive_unconverged_iterations, @@ -348,6 +410,7 @@ def evolve(state: Procedure | None, event: ProcedureEvent) -> Procedure: outcome_logbook_id=logbook_id, capability_id=prior.capability_id, recipe_id=prior.recipe_id, + hold_claims=prior.hold_claims, current_iteration_index=prior.current_iteration_index, iteration_count=prior.iteration_count, consecutive_unconverged_iterations=prior.consecutive_unconverged_iterations, @@ -397,6 +460,7 @@ def evolve(state: Procedure | None, event: ProcedureEvent) -> Procedure: outcome_logbook_id=prior.outcome_logbook_id, capability_id=prior.capability_id, recipe_id=prior.recipe_id, + hold_claims=prior.hold_claims, current_iteration_index=iteration_index, iteration_count=prior.iteration_count + 1, consecutive_unconverged_iterations=prior.consecutive_unconverged_iterations, @@ -427,6 +491,7 @@ def evolve(state: Procedure | None, event: ProcedureEvent) -> Procedure: outcome_logbook_id=prior.outcome_logbook_id, capability_id=prior.capability_id, recipe_id=prior.recipe_id, + hold_claims=prior.hold_claims, current_iteration_index=None, iteration_count=prior.iteration_count, consecutive_unconverged_iterations=consecutive_unconverged, diff --git a/apps/api/src/cora/operation/aggregates/procedure/state.py b/apps/api/src/cora/operation/aggregates/procedure/state.py index 1726477b6aa..ac6ed30652c 100644 --- a/apps/api/src/cora/operation/aggregates/procedure/state.py +++ b/apps/api/src/cora/operation/aggregates/procedure/state.py @@ -1557,6 +1557,18 @@ class Procedure: a denorm for audit-by-Capability read paths without requiring a Recipe join. Both fields are set by `register_procedure_from_recipe` to the same logical binding.""" + hold_claims: tuple[tuple[UUID, str], ...] = () + """Which concerns are currently holding this Procedure, oldest first. + + `status` alone answers "is this Held"; it cannot answer "by whom, and is + anyone else still holding it". That was adequate while a hold had a single + author and became a fault once independent concerns could each park the + same conduct: a second holder could not record its intent, and the first + holder's resume then restarted the conduct with the second's cause + unenforced. Terminal arms clear it, since a finished Procedure holds + nothing. Defaults to empty so pre-claim streams fold cleanly. Mirrors + `Run.hold_claims`.""" + current_iteration_index: int | None = field(default=None) """The convergence-loop iteration currently open, or None. diff --git a/apps/api/tests/architecture/test_procedure_evolver_carry_forward.py b/apps/api/tests/architecture/test_procedure_evolver_carry_forward.py index a46b96604ec..d10493fbd04 100644 --- a/apps/api/tests/architecture/test_procedure_evolver_carry_forward.py +++ b/apps/api/tests/architecture/test_procedure_evolver_carry_forward.py @@ -79,6 +79,20 @@ # Declared at genesis and never rewritten: a Procedure's beam need is # a property of the task, not of any transition it makes. "beam_requirement": frozenset(), + # The hold-claim set: placed by ProcedureHeld, discharged by + # ProcedureResumed (which is legal only for the LAST claim) and by the + # audit-only ProcedureHoldClaimReleased (which leaves the status alone). + # The three terminals clear it, since a finished Procedure holds nothing. + "hold_claims": frozenset( + { + "ProcedureHeld", + "ProcedureResumed", + "ProcedureHoldClaimReleased", + "ProcedureCompleted", + "ProcedureAborted", + "ProcedureTruncated", + } + ), } #: Fields every arm sets structurally rather than carrying forward: diff --git a/apps/api/tests/unit/operation/test_hold_procedure_handler.py b/apps/api/tests/unit/operation/test_hold_procedure_handler.py index 1378ea10256..00d7c96ee8b 100644 --- a/apps/api/tests/unit/operation/test_hold_procedure_handler.py +++ b/apps/api/tests/unit/operation/test_hold_procedure_handler.py @@ -63,6 +63,10 @@ async def test_handler_appends_procedure_held_event_with_trimmed_reason() -> Non "occurred_at": _NOW.isoformat(), # Operator hold (no conduct observer) leaves actuation_kind None. "actuation_kind": None, + # No cause supplied yet: the writers start passing one in a later + # slice, and until then a hold folds to the legacy single claim. + "claim_id": None, + "cause": None, } diff --git a/apps/api/tests/unit/operation/test_procedure_events.py b/apps/api/tests/unit/operation/test_procedure_events.py index b6987e51fcd..1a2a1803bf1 100644 --- a/apps/api/tests/unit/operation/test_procedure_events.py +++ b/apps/api/tests/unit/operation/test_procedure_events.py @@ -1189,6 +1189,8 @@ def test_to_payload_serializes_procedure_held() -> None: "decided_by_decision_id": str(decision_id), "occurred_at": _NOW.isoformat(), "actuation_kind": "Simulated", + "claim_id": None, + "cause": None, } @@ -1203,6 +1205,7 @@ def test_to_payload_serializes_procedure_resumed_with_null_decision() -> None: "re_establishment_boundary": 5, "decided_by_decision_id": None, "occurred_at": _NOW.isoformat(), + "released_claim_id": None, } diff --git a/apps/api/tests/unit/operation/test_procedure_hold_claims.py b/apps/api/tests/unit/operation/test_procedure_hold_claims.py new file mode 100644 index 00000000000..16d864f4527 --- /dev/null +++ b/apps/api/tests/unit/operation/test_procedure_hold_claims.py @@ -0,0 +1,246 @@ +"""Procedure hold-claim folding: the set of concerns currently holding a conduct. + +`status` alone is one bit, so a second holder arriving at an already-Held +Procedure could not record its intent and the first holder's resume restarted +the conduct with the second's cause unenforced. These tests pin the fold that +replaces the bit, and the legacy replay that keeps historical streams meaning +exactly what they meant before claims existed. +""" + +from datetime import UTC, datetime +from uuid import UUID, uuid4 + +import pytest + +from cora.operation.aggregates.procedure import ( + HOLD_CAUSE_DRIVER_STAND_DOWN, + HOLD_CAUSE_OPERATOR, + HOLD_CAUSE_STEP_FAULT, + LEGACY_CAUSE, + LEGACY_CLAIM_ID, + Procedure, + ProcedureAborted, + ProcedureCompleted, + ProcedureEvent, + ProcedureHeld, + ProcedureHoldClaimReleased, + ProcedureRegistered, + ProcedureResumed, + ProcedureStarted, + ProcedureStatus, + ProcedureTruncated, + fold, +) + +pytestmark = pytest.mark.unit + +_NOW = datetime(2026, 9, 10, 12, 0, tzinfo=UTC) +_PID = UUID("01900000-0000-7000-8000-0000000c1a10") +_OPERATOR_CLAIM = UUID("01900000-0000-7000-8000-00000000aa01") +_FAULT_CLAIM = UUID("01900000-0000-7000-8000-00000000bb02") + + +def _fold(events: list[ProcedureEvent]) -> Procedure: + """`fold` is Optional; every test here folds a genesis-first stream.""" + state = fold(events) + assert state is not None + return state + + +def _running() -> list[ProcedureEvent]: + events: list[ProcedureEvent] = [ + ProcedureRegistered( + procedure_id=_PID, + name="alignment", + kind="alignment", + target_asset_ids=(), + parent_run_id=None, + occurred_at=_NOW, + ), + ProcedureStarted(procedure_id=_PID, occurred_at=_NOW), + ] + return events + + +def _held(claim_id: UUID | None, cause: str | None) -> ProcedureHeld: + return ProcedureHeld( + procedure_id=_PID, + reason="parked", + occurred_at=_NOW, + claim_id=claim_id, + cause=cause, + ) + + +def test_two_concerns_hold_the_same_procedure_and_both_are_recorded() -> None: + state = _fold( + [ + *_running(), + _held(_FAULT_CLAIM, HOLD_CAUSE_STEP_FAULT), + _held(_OPERATOR_CLAIM, HOLD_CAUSE_OPERATOR), + ] + ) + assert state.status is ProcedureStatus.HELD + assert state.hold_claims == ( + (_FAULT_CLAIM, HOLD_CAUSE_STEP_FAULT), + (_OPERATOR_CLAIM, HOLD_CAUSE_OPERATOR), + ) + + +def test_releasing_one_of_two_claims_leaves_the_procedure_held() -> None: + """The fault this whole slice exists to fix: the first holder discharging + must NOT restart a conduct the second holder still wants parked.""" + state = _fold( + [ + *_running(), + _held(_FAULT_CLAIM, HOLD_CAUSE_STEP_FAULT), + _held(_OPERATOR_CLAIM, HOLD_CAUSE_OPERATOR), + ProcedureHoldClaimReleased( + procedure_id=_PID, + claim_id=_FAULT_CLAIM, + cause=HOLD_CAUSE_STEP_FAULT, + occurred_at=_NOW, + ), + ] + ) + assert state.status is ProcedureStatus.HELD + assert state.hold_claims == ((_OPERATOR_CLAIM, HOLD_CAUSE_OPERATOR),) + + +def test_resuming_the_last_claim_clears_it_and_runs() -> None: + state = _fold( + [ + *_running(), + _held(_OPERATOR_CLAIM, HOLD_CAUSE_OPERATOR), + ProcedureResumed( + procedure_id=_PID, + re_establishment_boundary=0, + occurred_at=_NOW, + released_claim_id=_OPERATOR_CLAIM, + ), + ] + ) + assert state.status is ProcedureStatus.RUNNING + assert state.hold_claims == () + + +def test_re_holding_under_a_live_claim_id_is_idempotent() -> None: + """A re-delivered hold from the same concern re-derives the same claim id, + so it must fold to one claim rather than stacking a second.""" + state = _fold( + [ + *_running(), + _held(_FAULT_CLAIM, HOLD_CAUSE_STEP_FAULT), + _held(_FAULT_CLAIM, HOLD_CAUSE_STEP_FAULT), + ] + ) + assert state.hold_claims == ((_FAULT_CLAIM, HOLD_CAUSE_STEP_FAULT),) + + +def test_a_distinct_conductor_concern_holds_alongside_a_step_fault() -> None: + """A stood-down steering driver is discharged by the driver being + reinstated, not by the equipment recovering, so the two coexist.""" + driver_claim = uuid4() + state = _fold( + [ + *_running(), + _held(_FAULT_CLAIM, HOLD_CAUSE_STEP_FAULT), + _held(driver_claim, HOLD_CAUSE_DRIVER_STAND_DOWN), + ] + ) + assert dict(state.hold_claims) == { + _FAULT_CLAIM: HOLD_CAUSE_STEP_FAULT, + driver_claim: HOLD_CAUSE_DRIVER_STAND_DOWN, + } + + +def test_legacy_claimless_hold_folds_to_the_single_legacy_claim() -> None: + state = _fold([*_running(), _held(None, None)]) + assert state.hold_claims == ((LEGACY_CLAIM_ID, LEGACY_CAUSE),) + + +def test_repeated_legacy_holds_collapse_to_one_claim() -> None: + """Pre-claim streams replay to their original one-bit meaning, so repeated + claimless holds must not accumulate.""" + state = _fold([*_running(), _held(None, None), _held(None, None)]) + assert state.hold_claims == ((LEGACY_CLAIM_ID, LEGACY_CAUSE),) + + +def test_legacy_bare_resume_clears_every_claim() -> None: + """A resume with no released_claim_id is the old one-bit semantics, which + cleared the hold outright.""" + state = _fold( + [ + *_running(), + _held(_FAULT_CLAIM, HOLD_CAUSE_STEP_FAULT), + _held(_OPERATOR_CLAIM, HOLD_CAUSE_OPERATOR), + ProcedureResumed(procedure_id=_PID, re_establishment_boundary=0, occurred_at=_NOW), + ] + ) + assert state.status is ProcedureStatus.RUNNING + assert state.hold_claims == () + + +def test_releasing_an_inactive_claim_is_a_fold_level_no_op() -> None: + """The evolver folds whatever the stream says and leaves rejection to the + deciders, which is what keeps replay total over any historical stream.""" + state = _fold( + [ + *_running(), + _held(_OPERATOR_CLAIM, HOLD_CAUSE_OPERATOR), + ProcedureHoldClaimReleased( + procedure_id=_PID, + claim_id=uuid4(), + cause=HOLD_CAUSE_STEP_FAULT, + occurred_at=_NOW, + ), + ] + ) + assert state.hold_claims == ((_OPERATOR_CLAIM, HOLD_CAUSE_OPERATOR),) + + +@pytest.mark.parametrize( + "terminal", + [ + ProcedureCompleted(procedure_id=_PID, occurred_at=_NOW), + ProcedureAborted(procedure_id=_PID, reason="gave up", occurred_at=_NOW), + ProcedureTruncated( + procedure_id=_PID, + reason="cut short", + interrupted_at=_NOW, + occurred_at=_NOW, + ), + ], + ids=["completed", "aborted", "truncated"], +) +def test_a_terminal_clears_every_claim(terminal: ProcedureEvent) -> None: + """A finished Procedure holds nothing, however many concerns were holding.""" + state = _fold( + [ + *_running(), + _held(_FAULT_CLAIM, HOLD_CAUSE_STEP_FAULT), + _held(_OPERATOR_CLAIM, HOLD_CAUSE_OPERATOR), + terminal, + ] + ) + assert state.hold_claims == () + + +def test_a_release_preserves_every_other_field() -> None: + """The audit-only arm changes one field. Pinned because the evolver's other + arms hand-list every field, which is how one gets silently dropped.""" + before = _fold([*_running(), _held(_OPERATOR_CLAIM, HOLD_CAUSE_OPERATOR)]) + after = _fold( + [ + *_running(), + _held(_OPERATOR_CLAIM, HOLD_CAUSE_OPERATOR), + _held(_FAULT_CLAIM, HOLD_CAUSE_STEP_FAULT), + ProcedureHoldClaimReleased( + procedure_id=_PID, + claim_id=_FAULT_CLAIM, + cause=HOLD_CAUSE_STEP_FAULT, + occurred_at=_NOW, + ), + ] + ) + assert after == before diff --git a/apps/api/tests/unit/operation/test_resume_procedure_handler.py b/apps/api/tests/unit/operation/test_resume_procedure_handler.py index 626ff4135e6..988c208edbb 100644 --- a/apps/api/tests/unit/operation/test_resume_procedure_handler.py +++ b/apps/api/tests/unit/operation/test_resume_procedure_handler.py @@ -142,6 +142,10 @@ async def test_handler_appends_procedure_resumed_event() -> None: "re_establishment_boundary": 2, "decided_by_decision_id": None, "occurred_at": _NOW.isoformat(), + # No claim released yet: the decider starts choosing between a resume + # and an audit-only release in a later slice. A bare resume clears + # every claim, which is the pre-claim behaviour this replays. + "released_claim_id": None, } From e9e45fe3efd78b2cea933f8d8c22330b91cbd561 Mon Sep 17 00:00:00 2001 From: xmap <16776958+xmap@users.noreply.github.com> Date: Thu, 10 Sep 2026 08:19:35 -0500 Subject: [PATCH 2/4] Refuse a resume that would restart work another concern is still holding S1 let a Procedure record which concerns hold it. This makes the deciders act on that record, which is the half that changes behaviour. Holding: `Held` becomes a legal starting status, so a second concern can record its intent on an already-parked conduct. That was previously refused outright, which is the fault: the second holder's cause went unrecorded and the first holder's resume then restarted the conduct with it unenforced. The guard moves from "is this Procedure un-held" to "is THIS CONCERN already holding it", so the alternation the old rule protected survives, scoped to the concern rather than the Procedure. Resuming: the caller discharges its own claim and the decider picks. Last active claim, emit ProcedureResumed and run. Others still holding, emit the audit-only ProcedureHoldClaimReleased and stay Held. No claim of its own, raise ProcedureHoldClaimsRemainError naming who is still holding, so the caller learns which concern to address rather than only that it was refused. Causes are not reachable from the wire. `cause` defaults to `operator` and neither the REST route nor the MCP tool exposes it, so a caller cannot label a machine-parked conduct an operator pause and inherit an operator's authority to clear it. The Conductor will set its own causes when the writers are wired; that is the next slice. Legacy holds stay resumable. A hold placed before claims existed has no owner, so no derived id matches it and every Procedure held at that moment would be stuck forever. An operator resume owns the legacy claim in addition to its own, since the operator is the authority that could always clear such a hold; a machine concern deliberately does not, so a step fault clearing cannot silently adopt a hold nobody can attribute. THREE TESTS CHANGED THAT PINNED THE OLD RULE, and they are the ones worth looking at rather than the payload churn: test_decide_rejects_non_running_status asserted "re-holding a Held one too" raises, and the two property tests derived their permitted set from `_HOLDABLE_SOURCES = (RUNNING,)`. All three encoded exactly the behaviour this commit deliberately changes, the same way seven Run tests did before its hold-claim fix. The replacement protection is the per-claim guard, covered directly. Two fitness tests earned their keep again: the decider-docstring invariants block caught both new raises going undocumented, and the generated disposition table needed regenerating for the release event's decision link. Verified by mutation: making a resume ignore other holders fails the test pinning that a discharge leaves the conduct Held. Co-Authored-By: Claude Opus 5 (1M context) --- .../record_export/_dispositions.py | 1 + .../aggregates/procedure/__init__.py | 4 + .../operation/aggregates/procedure/events.py | 9 + .../aggregates/procedure/hold_claims.py | 43 ++++ .../operation/aggregates/procedure/state.py | 55 ++++- .../features/hold_procedure/command.py | 11 + .../features/hold_procedure/decider.py | 45 +++- .../features/resume_procedure/command.py | 11 + .../features/resume_procedure/decider.py | 76 ++++++- apps/api/src/cora/operation/routes.py | 9 +- .../operation/test_hold_procedure_decider.py | 10 +- .../test_hold_procedure_decider_properties.py | 20 +- .../operation/test_hold_procedure_handler.py | 10 +- .../test_procedure_hold_claim_enforcement.py | 199 ++++++++++++++++++ .../test_resume_procedure_decider.py | 3 + .../test_resume_procedure_handler.py | 9 +- 16 files changed, 472 insertions(+), 43 deletions(-) create mode 100644 apps/api/src/cora/operation/aggregates/procedure/hold_claims.py create mode 100644 apps/api/tests/unit/operation/test_procedure_hold_claim_enforcement.py diff --git a/apps/api/src/cora/infrastructure/record_export/_dispositions.py b/apps/api/src/cora/infrastructure/record_export/_dispositions.py index 26205e7e937..df76b42248c 100644 --- a/apps/api/src/cora/infrastructure/record_export/_dispositions.py +++ b/apps/api/src/cora/infrastructure/record_export/_dispositions.py @@ -1442,6 +1442,7 @@ "ProcedureHoldClaimReleased": { "cause": "drop:text", "claim_id": "token:uuid", + "decided_by_decision_id": "token:uuid", "occurred_at": "keep:time", "procedure_id": "token:uuid", }, diff --git a/apps/api/src/cora/operation/aggregates/procedure/__init__.py b/apps/api/src/cora/operation/aggregates/procedure/__init__.py index 2a5146981f0..83ec16c9690 100644 --- a/apps/api/src/cora/operation/aggregates/procedure/__init__.py +++ b/apps/api/src/cora/operation/aggregates/procedure/__init__.py @@ -53,6 +53,7 @@ to_payload, ) from cora.operation.aggregates.procedure.evolver import evolve, fold +from cora.operation.aggregates.procedure.hold_claims import derive_claim_id from cora.operation.aggregates.procedure.read import ( load_procedure, load_procedure_with_events, @@ -94,6 +95,7 @@ ProcedureCannotTruncateError, ProcedureCapabilityExecutorMismatchError, ProcedureEnclosureCoverageMismatchError, + ProcedureHoldClaimsRemainError, ProcedureHoldReason, ProcedureIterationLimitReachedError, ProcedureName, @@ -181,6 +183,7 @@ "ProcedureEvent", "ProcedureHeld", "ProcedureHoldClaimReleased", + "ProcedureHoldClaimsRemainError", "ProcedureHoldReason", "ProcedureIterationEnded", "ProcedureIterationLimitReachedError", @@ -213,6 +216,7 @@ "ResolvedStepsRecorded", "SteeringDesignRecorded", "StepKind", + "derive_claim_id", "event_type_name", "evolve", "fold", diff --git a/apps/api/src/cora/operation/aggregates/procedure/events.py b/apps/api/src/cora/operation/aggregates/procedure/events.py index 7d366d8343d..549968647be 100644 --- a/apps/api/src/cora/operation/aggregates/procedure/events.py +++ b/apps/api/src/cora/operation/aggregates/procedure/events.py @@ -596,6 +596,10 @@ class ProcedureHoldClaimReleased: claim_id: UUID cause: str occurred_at: datetime + decided_by_decision_id: UUID | None = None + """Optional Decision-causation link, mirroring `ProcedureHeld`: None for + an operator-routed release, set when an in-process runtime discharges a + claim it placed.""" @dataclass(frozen=True) @@ -1063,12 +1067,16 @@ def to_payload(event: ProcedureEvent) -> dict[str, Any]: claim_id=claim_id, cause=cause, occurred_at=occurred_at, + decided_by_decision_id=decided_by_decision_id, ): return { "procedure_id": str(procedure_id), "claim_id": str(claim_id), "cause": cause, "occurred_at": occurred_at.isoformat(), + "decided_by_decision_id": ( + str(decided_by_decision_id) if decided_by_decision_id is not None else None + ), } case ProcedureActivitiesLogbookOpened( procedure_id=procedure_id, @@ -1473,6 +1481,7 @@ def _build_resumed() -> ProcedureResumed: claim_id=UUID(payload["claim_id"]), cause=payload["cause"], occurred_at=datetime.fromisoformat(payload["occurred_at"]), + decided_by_decision_id=_optional_uuid(payload.get("decided_by_decision_id")), ), ) case "ProcedureActivitiesLogbookOpened": diff --git a/apps/api/src/cora/operation/aggregates/procedure/hold_claims.py b/apps/api/src/cora/operation/aggregates/procedure/hold_claims.py new file mode 100644 index 00000000000..9b6527e597a --- /dev/null +++ b/apps/api/src/cora/operation/aggregates/procedure/hold_claims.py @@ -0,0 +1,43 @@ +"""Deriving the claim id a concern holds a Procedure under. + +The claim SET itself is folded onto `Procedure.hold_claims` by the evolver, so +a decider that already has folded state needs no second read. This module holds +only the derivation, which both a holder and a releaser must agree on without +either storing the id. + +Deliberately smaller than Run's sibling module, which also folds claims straight +from the event store. Run needs that because concerns OUTSIDE its BC hold Runs +and cannot fold the aggregate. Every Procedure holder today is the Conductor or +an operator surface, all of which go through a decider that receives folded +state. Add the event-store fold when an out-of-BC holder actually appears. +""" + +from __future__ import annotations + +from uuid import UUID, uuid5 + +# Stable namespace for deriving a concern's claim id on a Procedure. Distinct +# from Run's namespace so the two aggregates cannot collide, following the +# fixed-uuid5-namespace-per-derivation-family convention. +_CLAIM_NAMESPACE = UUID("01900000-0000-7000-8000-00000000c1a2") + + +def derive_claim_id(procedure_id: UUID, cause: str) -> UUID: + """The claim id a concern holds a given Procedure under: one per cause. + + Deterministic so a holder and a releaser agree without either storing it, + and so a re-delivered hold re-derives the same claim and folds idempotently + rather than stacking a second one. + + ONE claim per (Procedure, cause) is a deliberate coarsening, and it is what + the code already did: because every holder guarded `status is RUNNING`, a + second request from the SAME concern on an already-held Procedure was + refused outright. Keeping that collapse means this change fixes the + cross-cause fault without silently altering same-cause behaviour. Scoping a + claim more finely (per failing step, say) is a further refinement and would + need the release path to know which of them are still outstanding. + """ + return uuid5(_CLAIM_NAMESPACE, f"{procedure_id}|{cause}") + + +__all__ = ["derive_claim_id"] diff --git a/apps/api/src/cora/operation/aggregates/procedure/state.py b/apps/api/src/cora/operation/aggregates/procedure/state.py index ac6ed30652c..776afa1afa6 100644 --- a/apps/api/src/cora/operation/aggregates/procedure/state.py +++ b/apps/api/src/cora/operation/aggregates/procedure/state.py @@ -980,26 +980,59 @@ def __init__(self, procedure_id: UUID, current_status: "ProcedureStatus") -> Non class ProcedureCannotHoldError(Exception): - """Attempted to hold a Procedure not in `Running`. - - Single-source guard: `hold_procedure` accepts only `Running`. - Re-holding an already-`Held` Procedure raises (strict-not- - idempotent); holding a `Defined` or terminal Procedure raises. - Mirrors `RunCannotHoldError`. Hold <-> Resume is bidirectional and - unlimited-cycle: an operator can hold -> resume -> hold repeatedly - within one conduct, each hold requiring an intervening resume. - Mapped to HTTP 409. + """Attempted to hold a Procedure that this concern cannot hold. + + Two cases. The status is neither `Running` nor `Held` (a `Defined` or + terminal Procedure cannot be parked at all), or THIS concern already + holds an active claim. + + `Held` became a legal starting status when holds gained claims. It was + not before, and that was right while a hold had one author: it became a + fault once independent concerns could each park the same conduct, since + a second concern arriving at an already-Held Procedure could not record + its intent and the first concern's resume then restarted the conduct + with the second's cause unenforced. The guard moved from "is this + Procedure un-held" to "is THIS CONCERN already holding it", so + alternation is still enforced per claim while two DIFFERENT concerns + may now hold at once. Mirrors `RunCannotHoldError`. Mapped to HTTP 409. """ def __init__(self, procedure_id: UUID, current_status: "ProcedureStatus") -> None: super().__init__( - f"Procedure {procedure_id} cannot be held: currently in status " - f"{current_status.value}, hold requires {ProcedureStatus.RUNNING.value}" + f"Procedure {procedure_id} cannot be held by this concern: currently in " + f"status {current_status.value}, and a hold requires " + f"{ProcedureStatus.RUNNING.value} or {ProcedureStatus.HELD.value} with no " + f"active claim for this cause" ) self.procedure_id = procedure_id self.current_status = current_status +class ProcedureHoldClaimsRemainError(Exception): + """Attempted to resume a Held Procedure OTHER concerns are still holding. + + The Procedure is `Held` and the caller holds no active claim of its own, + so resuming would clear a hold the caller never placed. That is the fault + this class exists to make impossible: an operator must not restart a + conduct the Conductor parked on a failed setpoint, and a re-established + step must not restart one whose steering driver is still stood down. + + `blocking_causes` names the concerns still holding, so the caller learns + which one to address rather than only that it was refused. Clearing + another concern's claim is that concern's business. Mirrors + `RunHoldClaimsRemainError`. Mapped to HTTP 409. + """ + + def __init__(self, procedure_id: UUID, blocking_causes: tuple[str, ...]) -> None: + causes = ", ".join(blocking_causes) if blocking_causes else "unknown" + super().__init__( + f"Procedure {procedure_id} cannot be resumed: still held by {causes}. " + f"Each concern discharges its own claim." + ) + self.blocking_causes = blocking_causes + self.procedure_id = procedure_id + + class ProcedureCannotResumeError(Exception): """Attempted to resume a Procedure that cannot be resumed. diff --git a/apps/api/src/cora/operation/features/hold_procedure/command.py b/apps/api/src/cora/operation/features/hold_procedure/command.py index 0346338eea1..974d37da2c4 100644 --- a/apps/api/src/cora/operation/features/hold_procedure/command.py +++ b/apps/api/src/cora/operation/features/hold_procedure/command.py @@ -19,6 +19,8 @@ from dataclasses import dataclass from uuid import UUID +from cora.operation.aggregates.procedure import HOLD_CAUSE_OPERATOR + @dataclass(frozen=True) class HoldProcedure: @@ -26,6 +28,15 @@ class HoldProcedure: procedure_id: UUID reason: str + cause: str = HOLD_CAUSE_OPERATOR + """Which concern is placing or discharging the hold, from `HOLD_CAUSES`. + + Defaults to `operator` so the REST route and the MCP tool, which do NOT + expose this field, always speak for an operator. A caller able to choose + its own cause could label a machine-parked conduct as an operator pause; + the in-process Conductor sets its cause explicitly instead. The claim id + is NOT a command field: it is derived from (procedure_id, cause), so a + holder and a releaser agree on it without either storing it.""" decided_by_decision_id: UUID | None = None actuation_kind: str | None = None """The raw `ActuationKind` value the Conductor observed in the conduct up diff --git a/apps/api/src/cora/operation/features/hold_procedure/decider.py b/apps/api/src/cora/operation/features/hold_procedure/decider.py index 41b249faf4d..8b182341309 100644 --- a/apps/api/src/cora/operation/features/hold_procedure/decider.py +++ b/apps/api/src/cora/operation/features/hold_procedure/decider.py @@ -1,12 +1,20 @@ """Pure decider for the `HoldProcedure` command. -Single-source pause transition: `Running -> Held`. Re-holding an -already-`Held` Procedure raises (strict-not-idempotent); holding a -`Defined` or terminal Procedure raises. Mirrors `hold_run`. +Pause transition: `Running | Held -> Held`, placing ONE hold claim. -Hold <-> Resume is bidirectional and unlimited-cycle: an operator can -hold -> resume -> hold repeatedly within one conduct, each hold -requiring an intervening resume. +`Held` is a legal starting status. It was not, and re-holding raised on +the PackML and Bluesky precedent that hold and resume alternate. That was +right while a hold had one author and became a safety fault once +independent concerns could each park the same conduct: a second concern +arriving at an already-Held Procedure could not record its intent at all, +so the FIRST concern's release restarted the conduct with the second's +cause unenforced. + +So the guard moved from "is this Procedure un-held" to "is THIS CONCERN +already holding it". Alternation is still enforced per claim, which is +what the original precedent was protecting. What is newly admitted is two +DIFFERENT concerns holding at once, because that is the situation that +actually arises. Mirrors `hold_run`. `reason` validation goes through the `ProcedureHoldReason` VO (which calls the shared `validate_bounded_text` helper). The on-the-wire @@ -16,23 +24,31 @@ - State must not be None -> ProcedureNotFoundError - command.reason must be 1-500 chars after trimming -> InvalidProcedureHoldReasonError - - State.status must be in {Running} + - State.status must be in {Running, Held} + -> ProcedureCannotHoldError(current_status=...) + - command.cause must be in HOLD_CAUSES -> ValueError + - This cause's claim must not already be active -> ProcedureCannotHoldError(current_status=...) """ from datetime import datetime from cora.operation.aggregates.procedure import ( + HOLD_CAUSES, Procedure, ProcedureCannotHoldError, ProcedureHeld, ProcedureHoldReason, ProcedureNotFoundError, ProcedureStatus, + derive_claim_id, ) from cora.operation.features.hold_procedure.command import HoldProcedure -_HOLDABLE_STATUSES: tuple[ProcedureStatus, ...] = (ProcedureStatus.RUNNING,) +_HOLDABLE_STATUSES: tuple[ProcedureStatus, ...] = ( + ProcedureStatus.RUNNING, + ProcedureStatus.HELD, +) def decide( @@ -47,6 +63,17 @@ def decide( reason = ProcedureHoldReason(command.reason) if state.status not in _HOLDABLE_STATUSES: raise ProcedureCannotHoldError(state.id, current_status=state.status) + if command.cause not in HOLD_CAUSES: + raise ValueError( + f"Unknown hold cause {command.cause!r}; expected one of {sorted(HOLD_CAUSES)}" + ) + claim_id = derive_claim_id(state.id, command.cause) + # Per-claim alternation: this concern must discharge before holding again, + # which is what the original strict-not-idempotent rule was protecting. Two + # DIFFERENT concerns holding at once is what this decider now admits, and is + # the whole point of the change. + if any(active_id == claim_id for active_id, _ in state.hold_claims): + raise ProcedureCannotHoldError(state.id, current_status=state.status) return [ ProcedureHeld( procedure_id=state.id, @@ -54,5 +81,7 @@ def decide( decided_by_decision_id=command.decided_by_decision_id, occurred_at=now, actuation_kind=command.actuation_kind, + claim_id=claim_id, + cause=command.cause, ) ] diff --git a/apps/api/src/cora/operation/features/resume_procedure/command.py b/apps/api/src/cora/operation/features/resume_procedure/command.py index a65c7139114..dd366fb7efc 100644 --- a/apps/api/src/cora/operation/features/resume_procedure/command.py +++ b/apps/api/src/cora/operation/features/resume_procedure/command.py @@ -17,6 +17,8 @@ from dataclasses import dataclass from uuid import UUID +from cora.operation.aggregates.procedure import HOLD_CAUSE_OPERATOR + @dataclass(frozen=True) class ResumeProcedure: @@ -24,4 +26,13 @@ class ResumeProcedure: procedure_id: UUID re_establishment_boundary: int + cause: str = HOLD_CAUSE_OPERATOR + """Which concern is placing or discharging the hold, from `HOLD_CAUSES`. + + Defaults to `operator` so the REST route and the MCP tool, which do NOT + expose this field, always speak for an operator. A caller able to choose + its own cause could label a machine-parked conduct as an operator pause; + the in-process Conductor sets its cause explicitly instead. The claim id + is NOT a command field: it is derived from (procedure_id, cause), so a + holder and a releaser agree on it without either storing it.""" decided_by_decision_id: UUID | None = None diff --git a/apps/api/src/cora/operation/features/resume_procedure/decider.py b/apps/api/src/cora/operation/features/resume_procedure/decider.py index f958845b6dc..74718492f40 100644 --- a/apps/api/src/cora/operation/features/resume_procedure/decider.py +++ b/apps/api/src/cora/operation/features/resume_procedure/decider.py @@ -15,6 +15,9 @@ [[project_resumable_conduct_design]]. Invariants: + - command.cause must be in HOLD_CAUSES -> ValueError + - The caller must hold an active claim, else other concerns block + -> ProcedureHoldClaimsRemainError(blocking_causes=...) - State must not be None -> ProcedureNotFoundError - command.re_establishment_boundary must be >= 0 -> InvalidProcedureReEstablishmentBoundaryError @@ -25,14 +28,21 @@ """ from datetime import datetime +from uuid import UUID from cora.operation.aggregates.procedure import ( + HOLD_CAUSE_OPERATOR, + HOLD_CAUSES, + LEGACY_CLAIM_ID, InvalidProcedureReEstablishmentBoundaryError, Procedure, ProcedureCannotResumeError, + ProcedureHoldClaimReleased, + ProcedureHoldClaimsRemainError, ProcedureNotFoundError, ProcedureResumed, ProcedureStatus, + derive_claim_id, ) from cora.operation.features.resume_procedure.command import ResumeProcedure @@ -45,7 +55,7 @@ def decide( *, parent_run_held: bool = False, now: datetime, -) -> list[ProcedureResumed]: +) -> list[ProcedureResumed | ProcedureHoldClaimReleased]: """Decide the events produced by resuming a held Procedure. `parent_run_held` is the handler-derived fact that this Procedure's @@ -62,11 +72,61 @@ def decide( raise ProcedureCannotResumeError( state.id, current_status=state.status, parent_run_held=True ) - return [ - ProcedureResumed( - procedure_id=state.id, - re_establishment_boundary=command.re_establishment_boundary, - decided_by_decision_id=command.decided_by_decision_id, - occurred_at=now, + if command.cause not in HOLD_CAUSES: + raise ValueError( + f"Unknown hold cause {command.cause!r}; expected one of {sorted(HOLD_CAUSES)}" ) - ] + + def _resumed( + released_claim_id: UUID | None, + ) -> list[ProcedureResumed | ProcedureHoldClaimReleased]: + return [ + ProcedureResumed( + procedure_id=state.id, + re_establishment_boundary=command.re_establishment_boundary, + decided_by_decision_id=command.decided_by_decision_id, + occurred_at=now, + released_claim_id=released_claim_id, + ) + ] + + claim_id = derive_claim_id(state.id, command.cause) + active = tuple(active_id for active_id, _ in state.hold_claims) + if not active: + # Held with no active claim. Unreachable from a well-formed stream (a + # ProcedureHeld always yields at least the legacy claim), but if it + # happens the safety property already holds (no concern is holding + # this), so resume rather than wedge the conduct shut. + return _resumed(None) + # A hold placed BEFORE holds carried claims has no recorded owner, so no + # derived id matches it and it would otherwise be unresumable: every + # Procedure held at the moment this shipped. The operator is the authority + # that could always clear such a hold, so an operator resume owns the + # legacy claim in addition to its own. + owned = {claim_id} + if command.cause == HOLD_CAUSE_OPERATOR: + owned.add(LEGACY_CLAIM_ID) + held_by_caller = tuple(cid for cid in active if cid in owned) + if not held_by_caller: + # Held, but not by us. Refuse, and name who is holding so the caller + # learns which concern to address rather than only that it was refused. + raise ProcedureHoldClaimsRemainError( + state.id, + blocking_causes=tuple(cause for _, cause in state.hold_claims), + ) + if set(active) <= {LEGACY_CLAIM_ID}: + # A legacy one-bit hold: clearing it means clearing the hold outright, + # which is exactly what a bare ProcedureResumed does at the fold. + return _resumed(None) + if len(active) > 1: + # Others still hold it: discharge our claim without moving the status. + return [ + ProcedureHoldClaimReleased( + procedure_id=state.id, + claim_id=held_by_caller[0], + cause=command.cause, + decided_by_decision_id=command.decided_by_decision_id, + occurred_at=now, + ) + ] + return _resumed(held_by_caller[0]) diff --git a/apps/api/src/cora/operation/routes.py b/apps/api/src/cora/operation/routes.py index 638aef52bff..18a93bddd98 100644 --- a/apps/api/src/cora/operation/routes.py +++ b/apps/api/src/cora/operation/routes.py @@ -59,6 +59,7 @@ ProcedureCannotTruncateError, ProcedureCapabilityExecutorMismatchError, ProcedureEnclosureCoverageMismatchError, + ProcedureHoldClaimsRemainError, ProcedureIterationLimitReachedError, ProcedureNotFoundError, ProcedurePlanAssetDecommissionedError, @@ -296,10 +297,14 @@ def register_operation_routes(app: FastAPI) -> None: ProcedureCannotCompleteError, ProcedureCannotAbortError, ProcedureCannotTruncateError, - # resumable-conduct pause/resume guards (Running->Held->Running): - # holding a non-Running procedure, or resuming a non-Held one. + # resumable-conduct pause/resume guards (Running|Held->Held->Running): + # holding a Defined or terminal procedure, holding one this concern + # already holds, or resuming a non-Held one. ProcedureCannotHoldError, ProcedureCannotResumeError, + # resuming a Held procedure OTHER concerns are still holding: doing so + # would clear a hold the caller never placed. + ProcedureHoldClaimsRemainError, # iteration boundary guards (start/end): not-Running, no/already-open # iteration, and non-sequential / mismatched operator-supplied index. ProcedureCannotStartIterationError, diff --git a/apps/api/tests/unit/operation/test_hold_procedure_decider.py b/apps/api/tests/unit/operation/test_hold_procedure_decider.py index bd89465080f..ab9f8691b4a 100644 --- a/apps/api/tests/unit/operation/test_hold_procedure_decider.py +++ b/apps/api/tests/unit/operation/test_hold_procedure_decider.py @@ -123,14 +123,18 @@ def test_decide_rejects_too_long_reason() -> None: "status", [ ProcedureStatus.DEFINED, - ProcedureStatus.HELD, ProcedureStatus.COMPLETED, ProcedureStatus.ABORTED, ProcedureStatus.TRUNCATED, ], ) -def test_decide_rejects_non_running_status(status: ProcedureStatus) -> None: - """Holding a non-Running procedure raises (re-holding a Held one too).""" +def test_decide_rejects_non_holdable_status(status: ProcedureStatus) -> None: + """A Defined or terminal Procedure cannot be parked at all. + + `Held` is deliberately absent: a SECOND concern must be able to record a + hold on an already-held conduct, which is the fault cause-scoped claims + exist to fix. Re-holding under the SAME cause is still refused, by the + per-claim guard rather than by this status guard.""" proc = _procedure(status=status) with pytest.raises(ProcedureCannotHoldError) as exc: hold_procedure.decide( diff --git a/apps/api/tests/unit/operation/test_hold_procedure_decider_properties.py b/apps/api/tests/unit/operation/test_hold_procedure_decider_properties.py index 45a8d726e1b..13a187ddd93 100644 --- a/apps/api/tests/unit/operation/test_hold_procedure_decider_properties.py +++ b/apps/api/tests/unit/operation/test_hold_procedure_decider_properties.py @@ -29,12 +29,14 @@ from hypothesis import strategies as st from cora.operation.aggregates.procedure import ( + HOLD_CAUSE_OPERATOR, Procedure, ProcedureCannotHoldError, ProcedureHeld, ProcedureName, ProcedureNotFoundError, ProcedureStatus, + derive_claim_id, ) from cora.operation.features import hold_procedure from cora.operation.features.hold_procedure import HoldProcedure @@ -46,7 +48,10 @@ _REASON = printable_ascii_text(min_size=1, max_size=500) -_HOLDABLE_SOURCES = (ProcedureStatus.RUNNING,) +# `Held` is holdable because a second CONCERN may hold an already-held +# conduct; the same concern re-holding is refused by the per-claim guard, +# which `test_hold_procedure_decider` covers directly. +_HOLDABLE_SOURCES = (ProcedureStatus.RUNNING, ProcedureStatus.HELD) _DISALLOWED_SOURCES = tuple(s for s in ProcedureStatus if s not in frozenset(_HOLDABLE_SOURCES)) @@ -91,13 +96,22 @@ def test_hold_from_permitted_source_emits_single_event( reason: str, now: datetime, ) -> None: - """Running emits one ProcedureHeld with the threaded reason.""" + """A holdable source emits one ProcedureHeld with the threaded reason, + carrying the claim this concern holds it under.""" events = hold_procedure.decide( state=_procedure(procedure_id=procedure_id, status=source), command=HoldProcedure(procedure_id=procedure_id, reason=reason), now=now, ) - assert events == [ProcedureHeld(procedure_id=procedure_id, reason=reason, occurred_at=now)] + assert events == [ + ProcedureHeld( + procedure_id=procedure_id, + reason=reason, + occurred_at=now, + claim_id=derive_claim_id(procedure_id, HOLD_CAUSE_OPERATOR), + cause=HOLD_CAUSE_OPERATOR, + ) + ] @pytest.mark.unit diff --git a/apps/api/tests/unit/operation/test_hold_procedure_handler.py b/apps/api/tests/unit/operation/test_hold_procedure_handler.py index 00d7c96ee8b..b1b99d16774 100644 --- a/apps/api/tests/unit/operation/test_hold_procedure_handler.py +++ b/apps/api/tests/unit/operation/test_hold_procedure_handler.py @@ -12,9 +12,11 @@ from cora.infrastructure.adapters.in_memory_event_store import InMemoryEventStore from cora.operation.aggregates.procedure import ( + HOLD_CAUSE_OPERATOR, InvalidProcedureHoldReasonError, ProcedureCannotHoldError, ProcedureNotFoundError, + derive_claim_id, ) from cora.operation.errors import UnauthorizedError from cora.operation.features import hold_procedure @@ -63,10 +65,10 @@ async def test_handler_appends_procedure_held_event_with_trimmed_reason() -> Non "occurred_at": _NOW.isoformat(), # Operator hold (no conduct observer) leaves actuation_kind None. "actuation_kind": None, - # No cause supplied yet: the writers start passing one in a later - # slice, and until then a hold folds to the legacy single claim. - "claim_id": None, - "cause": None, + # The route and tool do not expose `cause`, so an operator hold is + # what a wire caller always gets, under a claim derived from it. + "claim_id": str(derive_claim_id(_PROCEDURE_ID, HOLD_CAUSE_OPERATOR)), + "cause": HOLD_CAUSE_OPERATOR, } diff --git a/apps/api/tests/unit/operation/test_procedure_hold_claim_enforcement.py b/apps/api/tests/unit/operation/test_procedure_hold_claim_enforcement.py new file mode 100644 index 00000000000..b0eb378bb05 --- /dev/null +++ b/apps/api/tests/unit/operation/test_procedure_hold_claim_enforcement.py @@ -0,0 +1,199 @@ +"""Hold and resume deciders enforcing cause-scoped claims. + +The fold in `test_procedure_hold_claims` proves a Procedure can REPRESENT two +concerns holding it. These prove the deciders act on that: a second concern can +record its hold, and a resume that would restart work another concern still +wants parked is refused rather than silently granted. +""" + +from datetime import UTC, datetime +from uuid import UUID, uuid4 + +import pytest + +from cora.operation.aggregates.procedure import ( + HOLD_CAUSE_DRIVER_STAND_DOWN, + HOLD_CAUSE_OPERATOR, + HOLD_CAUSE_STEP_FAULT, + LEGACY_CAUSE, + LEGACY_CLAIM_ID, + Procedure, + ProcedureCannotHoldError, + ProcedureHeld, + ProcedureHoldClaimReleased, + ProcedureHoldClaimsRemainError, + ProcedureName, + ProcedureResumed, + ProcedureStatus, + derive_claim_id, +) +from cora.operation.features import hold_procedure, resume_procedure +from cora.operation.features.hold_procedure import HoldProcedure +from cora.operation.features.resume_procedure import ResumeProcedure + +pytestmark = pytest.mark.unit + +_NOW = datetime(2026, 9, 10, 12, 0, tzinfo=UTC) +_PID = UUID("01900000-0000-7000-8000-0000000c1a20") + + +def _procedure( + *, + status: ProcedureStatus = ProcedureStatus.RUNNING, + hold_claims: tuple[tuple[UUID, str], ...] = (), +) -> Procedure: + return Procedure( + id=_PID, + name=ProcedureName("alignment"), + kind="alignment", + target_asset_ids=frozenset(), + status=status, + parent_run_id=None, + hold_claims=hold_claims, + ) + + +def _claim(cause: str) -> UUID: + return derive_claim_id(_PID, cause) + + +def _held_by(*causes: str) -> Procedure: + return _procedure( + status=ProcedureStatus.HELD, + hold_claims=tuple((_claim(cause), cause) for cause in causes), + ) + + +def _resume(cause: str) -> ResumeProcedure: + return ResumeProcedure(procedure_id=_PID, re_establishment_boundary=0, cause=cause) + + +def test_a_second_concern_can_hold_an_already_held_procedure() -> None: + """The fault this slice fixes: previously this was refused outright and the + second concern's intent went unrecorded.""" + events = hold_procedure.decide( + state=_held_by(HOLD_CAUSE_STEP_FAULT), + command=HoldProcedure( + procedure_id=_PID, reason="swapping the sample", cause=HOLD_CAUSE_OPERATOR + ), + now=_NOW, + ) + assert [type(e) for e in events] == [ProcedureHeld] + assert isinstance(events[0], ProcedureHeld) + assert events[0].cause == HOLD_CAUSE_OPERATOR + assert events[0].claim_id == _claim(HOLD_CAUSE_OPERATOR) + + +def test_the_same_concern_cannot_hold_twice_without_discharging() -> None: + """Per-claim alternation survives: what the old status guard protected is + still protected, just scoped to the concern instead of the Procedure.""" + with pytest.raises(ProcedureCannotHoldError): + hold_procedure.decide( + state=_held_by(HOLD_CAUSE_STEP_FAULT), + command=HoldProcedure(procedure_id=_PID, reason="again", cause=HOLD_CAUSE_STEP_FAULT), + now=_NOW, + ) + + +def test_an_unknown_cause_is_refused() -> None: + with pytest.raises(ValueError, match="Unknown hold cause"): + hold_procedure.decide( + state=_procedure(), + command=HoldProcedure(procedure_id=_PID, reason="x", cause="whatever"), + now=_NOW, + ) + + +def test_resuming_while_another_concern_holds_discharges_without_restarting() -> None: + """The safety property. The step fault clearing must NOT restart a conduct + the operator is still holding.""" + events = resume_procedure.decide( + state=_held_by(HOLD_CAUSE_STEP_FAULT, HOLD_CAUSE_OPERATOR), + command=_resume(HOLD_CAUSE_STEP_FAULT), + now=_NOW, + ) + assert [type(e) for e in events] == [ProcedureHoldClaimReleased] + assert isinstance(events[0], ProcedureHoldClaimReleased) + assert events[0].claim_id == _claim(HOLD_CAUSE_STEP_FAULT) + + +def test_resuming_the_last_claim_restarts_the_conduct() -> None: + events = resume_procedure.decide( + state=_held_by(HOLD_CAUSE_OPERATOR), + command=_resume(HOLD_CAUSE_OPERATOR), + now=_NOW, + ) + assert [type(e) for e in events] == [ProcedureResumed] + assert isinstance(events[0], ProcedureResumed) + assert events[0].released_claim_id == _claim(HOLD_CAUSE_OPERATOR) + + +def test_resuming_without_a_claim_is_refused_and_names_the_holders() -> None: + """Refusing is half the job: the caller must learn WHICH concern to address + rather than only that it was refused.""" + with pytest.raises(ProcedureHoldClaimsRemainError) as exc: + resume_procedure.decide( + state=_held_by(HOLD_CAUSE_STEP_FAULT, HOLD_CAUSE_DRIVER_STAND_DOWN), + command=_resume(HOLD_CAUSE_OPERATOR), + now=_NOW, + ) + assert exc.value.blocking_causes == ( + HOLD_CAUSE_STEP_FAULT, + HOLD_CAUSE_DRIVER_STAND_DOWN, + ) + assert HOLD_CAUSE_STEP_FAULT in str(exc.value) + + +def test_an_operator_resume_owns_a_legacy_claim() -> None: + """A hold placed before claims existed has no owner, so no derived id + matches it and it would be unresumable forever. The operator is the + authority that could always clear such a hold.""" + events = resume_procedure.decide( + state=_procedure( + status=ProcedureStatus.HELD, + hold_claims=((LEGACY_CLAIM_ID, LEGACY_CAUSE),), + ), + command=_resume(HOLD_CAUSE_OPERATOR), + now=_NOW, + ) + assert [type(e) for e in events] == [ProcedureResumed] + assert isinstance(events[0], ProcedureResumed) + # Bare resume: at the fold that clears every claim, which is exactly the + # one-bit behaviour a pre-claim stream had. + assert events[0].released_claim_id is None + + +def test_a_machine_concern_does_not_own_a_legacy_claim() -> None: + """Only the operator inherits an unowned hold. A step fault clearing must + not silently adopt and discharge a hold nobody can attribute.""" + with pytest.raises(ProcedureHoldClaimsRemainError): + resume_procedure.decide( + state=_procedure( + status=ProcedureStatus.HELD, + hold_claims=((LEGACY_CLAIM_ID, LEGACY_CAUSE),), + ), + command=_resume(HOLD_CAUSE_STEP_FAULT), + now=_NOW, + ) + + +def test_a_held_procedure_with_no_claims_resumes_rather_than_wedging() -> None: + """Unreachable from a well-formed stream, but if it happens the safety + property already holds, so do not lock the conduct shut.""" + events = resume_procedure.decide( + state=_procedure(status=ProcedureStatus.HELD), + command=_resume(HOLD_CAUSE_OPERATOR), + now=_NOW, + ) + assert [type(e) for e in events] == [ProcedureResumed] + + +def test_the_wire_surfaces_cannot_choose_a_cause() -> None: + """A caller able to pick its own cause could label a machine-parked conduct + an operator pause. The command defaults instead, and the route and tool do + not expose the field.""" + assert HoldProcedure(procedure_id=uuid4(), reason="x").cause == HOLD_CAUSE_OPERATOR + assert ( + ResumeProcedure(procedure_id=uuid4(), re_establishment_boundary=0).cause + == HOLD_CAUSE_OPERATOR + ) diff --git a/apps/api/tests/unit/operation/test_resume_procedure_decider.py b/apps/api/tests/unit/operation/test_resume_procedure_decider.py index 5ac93884bc0..f5676cb2a4d 100644 --- a/apps/api/tests/unit/operation/test_resume_procedure_decider.py +++ b/apps/api/tests/unit/operation/test_resume_procedure_decider.py @@ -130,6 +130,9 @@ def test_decide_accepts_zero_boundary() -> None: command=ResumeProcedure(procedure_id=proc.id, re_establishment_boundary=0), now=_NOW, ) + # The decider now returns a union: a resume, or an audit-only release + # when other concerns still hold. This path is the resume. + assert isinstance(events[0], ProcedureResumed) assert events[0].re_establishment_boundary == 0 diff --git a/apps/api/tests/unit/operation/test_resume_procedure_handler.py b/apps/api/tests/unit/operation/test_resume_procedure_handler.py index 988c208edbb..b5c068659c4 100644 --- a/apps/api/tests/unit/operation/test_resume_procedure_handler.py +++ b/apps/api/tests/unit/operation/test_resume_procedure_handler.py @@ -14,10 +14,12 @@ from cora.infrastructure.adapters.in_memory_event_store import InMemoryEventStore from cora.infrastructure.event_envelope import to_new_event from cora.operation.aggregates.procedure import ( + HOLD_CAUSE_OPERATOR, ProcedureCannotResumeError, ProcedureNotFoundError, ProcedureRegistered, ProcedureStarted, + derive_claim_id, event_type_name, to_payload, ) @@ -142,10 +144,9 @@ async def test_handler_appends_procedure_resumed_event() -> None: "re_establishment_boundary": 2, "decided_by_decision_id": None, "occurred_at": _NOW.isoformat(), - # No claim released yet: the decider starts choosing between a resume - # and an audit-only release in a later slice. A bare resume clears - # every claim, which is the pre-claim behaviour this replays. - "released_claim_id": None, + # The hold above placed an operator claim, and it is the only one, so + # the resume discharges it by name rather than clearing the world. + "released_claim_id": str(derive_claim_id(_PROCEDURE_ID, HOLD_CAUSE_OPERATOR)), } From 18f993c82b4176bd0a7af7ce1cd0bee35a1f0dce Mon Sep 17 00:00:00 2001 From: xmap <16776958+xmap@users.noreply.github.com> Date: Thu, 10 Sep 2026 10:27:48 -0500 Subject: [PATCH 3/4] Let the Conductor say which concern parked a conduct, and let an operator clear it S1 gave a Procedure somewhere to record which concerns hold it, and S2 made the deciders act on that record. Every hold still folded to a single claim because no writer passed a cause. This wires the Conductor's two. Passing the causes alone would have wedged the system, which the change had to answer before it could ship. `conduct_or_hold` parks a conduct on a recoverable step failure; `_hold_driver_stood_down` parks a steered loop whose driver was switched off. Under S2's rule that each concern discharges its own claim, neither hold could ever be discharged: nothing in CORA re-establishes a faulted step or reinstates a driver, and an operator's resume derives a different claim id. Every fault-parked conduct would have become permanently unresumable. Reproduced at the decider before writing the fix: WEDGED: ProcedureHoldClaimsRemainError ... still held by step-fault. The wedge exposes a distinction the claim algebra had been carrying implicitly. A hold claim is one of two things. An AUTHORITY claim is held on behalf of a rule the holder enforces, and letting anyone else clear it is the bypass it exists to prevent: Run's ratification and kill-switch claims are these, which is why `resume_run` refuses an operator outright. An ATTENTION claim is the opposite. The Conductor is reporting a conduct it cannot itself un-stick, the claim's job is to say what an operator would be resuming past, and an operator resuming IS the discharge. So `ATTENTION_HOLD_CAUSES` records which Procedure causes are which, and an operator's resume clears its own claim, a legacy claim, and every attention claim. The set is declared rather than assumed, with `test_every_hold_cause_is_classified` failing until a newly added cause is put on one side or the other. "May an operator clear this" is the question a new concern exists to raise, and inheriting an answer is how it would go unasked. Clearing several claims at once needed the emission to change. The caller's own claim rides the `ProcedureResumed`; each claim it clears on another concern's behalf gets its own `ProcedureHoldClaimReleased` first. A resume that answers three concerns then records three discharges rather than one status change that silently absorbed two of them. WHAT DOES NOT CHANGE TODAY: every resume in the system is an operator resume, since `cause` defaults to `operator` and no wire surface exposes it, and an operator may clear every Procedure cause. So `ProcedureHoldClaimsRemainError` does not fire in production. It is the guard that keeps a future machine resumer from clearing a hold it never placed. What changes now is that a machine hold is RECORDED at all: before this, a conduct faulting on an already-held Procedure filed its fault under the operator's claim, was refused as a duplicate, and the cause went nowhere. A GUARD, because the widening rests on something outside the algebra. `operator` is the DEFAULT cause, so it means "the caller named no other concern", and nothing checks what kind of thing the caller is. Every unmarked resume traces to a person today only because no agent is granted `ResumeProcedure`; grant one tomorrow and it defaults into `operator` and inherits authority over every machine hold with nobody having decided that. `test_no_agent_is_granted_resume_procedure` puts that where it can be seen, in the grant table rather than in a docstring, and says what to do instead if an agent ever needs to resume. The same pressure is why a machine is already an operator in one place: `_experiment_coordinator` has an in-process `HoldProcedure` grant and sets no cause, so an agent's own decision is filed as "a person asked". A DELIBERATE DIVERGENCE FROM RUN, since the two now differ and a reader will ask. `resume_run` exposes `cause` on both the REST route and the MCP tool, added because a kill-switch hold otherwise had no exit but abort_run and its destroyed beamtime; on a Run, naming the cause is how a person says which claim they are clearing. `resume_procedure` does not expose it, and an operator's ownership of attention claims does that job instead. Both let a person un-stick a machine-parked thing. Run's records the intent more precisely, this one takes one call. The difference is defensible because no Procedure claim is an authority claim, and it stops being defensible the moment one is, which is what the classification test exists to catch. One interaction is deliberately left alone. `_experiment_coordinator` still issues a follow-on hold under the default operator cause, so on the Held path it now records a mislabelled operator claim where it previously raised. That whole call is removed by 187c55455cc on `fix-coordinator-holds-aborted-procedure`, which found it invalid on every path that reaches it; wiring a cause into code being deleted would only conflict with that fix, and `steer_experiment` has no production caller. THREE INTEGRATION TESTS CHANGED, and one of them was already broken. The two `conduct_from` soft-IOC tests pin the exact event stream, which now carries the release between the hold and the resume; that path parking and resuming against a real IOC and Postgres is the end-to-end proof the wedge is gone. The third, `test_a_person_and_an_agent_hold_through_one_gate_and_ the_record_agrees`, compares two Procedures' ProcedureHeld payloads field by field after stripping what must differ. S2 began writing a claim_id derived from the procedure id, so it differs for the same reason procedure_id does, and nothing stripped it: that test has been red since 479b9ae12d9 and was reported green on a test count that did not cover it. Fixed by asserting the field rather than dropping it, so both paths still have to file a hold under the same cause, differing only in which Procedure the claim is on. Verified by mutation, five separately: dropping the operator's ownership of attention claims, dropping either Conductor cause, dropping the extra release events, granting an agent `ResumeProcedure`, and deriving a claim from a fixed id instead of the Procedure's each fail tests that name the behaviour rather than the shape. Co-Authored-By: Claude Opus 5 (1M context) --- .../aggregates/procedure/__init__.py | 2 + .../operation/aggregates/procedure/events.py | 61 ++++++++-- apps/api/src/cora/operation/conductor.py | 15 ++- .../features/resume_procedure/decider.py | 108 +++++++++++++----- ...t_conduct_from_against_softioc_postgres.py | 10 ++ .../test_paired_hold_authz_postgres.py | 20 +++- .../tests/unit/api/test_in_process_grants.py | 35 +++++- apps/api/tests/unit/operation/_helpers.py | 2 + .../tests/unit/operation/test_conductor.py | 5 + ...st_conductor_steering_driver_stand_down.py | 5 + .../test_procedure_hold_claim_enforcement.py | 68 ++++++++++- 11 files changed, 282 insertions(+), 49 deletions(-) diff --git a/apps/api/src/cora/operation/aggregates/procedure/__init__.py b/apps/api/src/cora/operation/aggregates/procedure/__init__.py index 83ec16c9690..ed9372c68ff 100644 --- a/apps/api/src/cora/operation/aggregates/procedure/__init__.py +++ b/apps/api/src/cora/operation/aggregates/procedure/__init__.py @@ -25,6 +25,7 @@ PostgresOutcomeStore, ) from cora.operation.aggregates.procedure.events import ( + ATTENTION_HOLD_CAUSES, HOLD_CAUSE_DRIVER_STAND_DOWN, HOLD_CAUSE_OPERATOR, HOLD_CAUSE_STEP_FAULT, @@ -122,6 +123,7 @@ ) __all__ = [ + "ATTENTION_HOLD_CAUSES", "DIAGNOSTIC_LOGBOOK_SCHEMA", "HOLD_CAUSES", "HOLD_CAUSE_DRIVER_STAND_DOWN", diff --git a/apps/api/src/cora/operation/aggregates/procedure/events.py b/apps/api/src/cora/operation/aggregates/procedure/events.py index 549968647be..35c5e94fb73 100644 --- a/apps/api/src/cora/operation/aggregates/procedure/events.py +++ b/apps/api/src/cora/operation/aggregates/procedure/events.py @@ -446,13 +446,15 @@ class ProcedureAborted: HOLD_CAUSE_STEP_FAULT = "step-fault" """The Conductor parked the conduct because a step failed recoverably -(`conduct_or_hold`). Released by whatever re-establishes the step.""" +(`conduct_or_hold`). An attention claim: nothing in CORA re-establishes a +faulted step, so an operator discharges it by resuming.""" HOLD_CAUSE_DRIVER_STAND_DOWN = "driver-stand-down" """The Conductor parked a steered loop because its steering driver went non-ACTIVE (`_hold_driver_stood_down`). A DIFFERENT concern from a step -fault: it is discharged by the driver being reinstated, not by the -equipment recovering, so the two must be able to hold at the same time.""" +fault: it is answered by the driver being reinstated, not by the equipment +recovering, so the two must be able to hold at the same time. Also an +attention claim.""" HOLD_CAUSES: frozenset[str] = frozenset( { @@ -465,6 +467,30 @@ class ProcedureAborted: a cause names WHICH concern is holding, not why in prose. The prose stays on `ProcedureHeld.reason`. Mirrors `HOLD_CAUSES` on Run.""" +ATTENTION_HOLD_CAUSES: frozenset[str] = frozenset( + { + HOLD_CAUSE_STEP_FAULT, + HOLD_CAUSE_DRIVER_STAND_DOWN, + } +) +"""The causes whose claim asks for an operator rather than asserting authority. + +A hold claim is one of two things, and which one decides who may discharge it. +An AUTHORITY claim is held on behalf of a rule the holder enforces: Run's +ratification and kill-switch claims are authority claims, and letting anyone +else clear them would be the bypass they exist to prevent. An ATTENTION claim +is the opposite. The Conductor parks a conduct it cannot itself un-stick and +records why; nothing in CORA will ever discharge that claim, so an operator's +resume does, and the claim's job was to say what they are resuming past. + +Every non-operator Procedure cause is an attention claim today, which is what +makes an operator's resume able to clear a fault-parked conduct in one act. +That is a property of these particular causes and NOT of Procedures, so the set +is declared rather than assumed: `test_every_hold_cause_is_classified` fails +until a newly added cause is put on one side or the other, which is the point +at which "may an operator clear this" has to be answered rather than inherited. +""" + LEGACY_CLAIM_ID = UUID("01900000-0000-7000-8000-00000001dead") """The single claim a pre-claim `ProcedureHeld` folds to, so a legacy stream replays to exactly its old one-bit meaning. Mirrors Run's sentinel.""" @@ -562,19 +588,28 @@ class ProcedureResumed: occurred_at: datetime decided_by_decision_id: UUID | None = None released_claim_id: UUID | None = None - """The claim this resume discharges. Legal ONLY when it is the last - active claim, which is what makes RUNNING mean "no concern is holding - this" rather than "whoever spoke last is done". None on a legacy - stream, where a bare resume clears every claim and so replays the old - one-bit behaviour exactly.""" + """The resumer's OWN claim, discharged by this resume. + + A `ProcedureResumed` is legal only once no claim would remain active, which + is what makes RUNNING mean "no concern is holding this" rather than + "whoever spoke last is done". Any OTHER claim the resumer was entitled to + clear (an attention claim, per `ATTENTION_HOLD_CAUSES`) is discharged by + its own `ProcedureHoldClaimReleased` first, so a resume that answers three + concerns records three discharges rather than collapsing them into this one + field. + + None when the resumer held no claim of its own, and on a legacy stream, + where a bare resume clears every claim and so replays the old one-bit + behaviour exactly.""" @dataclass(frozen=True) class ProcedureHoldClaimReleased: - """One hold claim was discharged while other claims remain active. + """One hold claim was discharged. - Audit-only: the evolver returns prior state with the claim removed and - the status untouched, so the Procedure stays `Held`. + Status-neutral: the evolver returns prior state with the claim removed and + the status untouched. Whether the Procedure goes on to run again is the + business of a `ProcedureResumed` in the same append, not of this event. This is the event that makes the hold algebra compositional. Without it a concern has exactly two ways to stop holding, both wrong when it is @@ -584,6 +619,9 @@ class ProcedureHoldClaimReleased: - own claim is the ONLY active one -> `ProcedureResumed(released_claim_id)` - other claims remain -> `ProcedureHoldClaimReleased(claim_id)` + - an operator clears attention claims alongside its own + -> one of these per extra claim, + then the `ProcedureResumed` `claim_id` must name an active claim; releasing an unknown or already-released claim is a no-op at the fold and is rejected by the @@ -1619,6 +1657,7 @@ def _build_resumed() -> ProcedureResumed: __all__ = [ + "ATTENTION_HOLD_CAUSES", "HOLD_CAUSES", "HOLD_CAUSE_DRIVER_STAND_DOWN", "HOLD_CAUSE_OPERATOR", diff --git a/apps/api/src/cora/operation/conductor.py b/apps/api/src/cora/operation/conductor.py index 34d2d37f35c..fba96c913e7 100644 --- a/apps/api/src/cora/operation/conductor.py +++ b/apps/api/src/cora/operation/conductor.py @@ -132,6 +132,8 @@ from cora.infrastructure.routing import NIL_SENTINEL_ID from cora.operation._control_dispatch_context import with_dispatch_correlation_id from cora.operation.aggregates.procedure import ( + HOLD_CAUSE_DRIVER_STAND_DOWN, + HOLD_CAUSE_STEP_FAULT, ProcedureIterationLimitReachedError, ProcedureNotFoundError, ProcedureTerminationReason, @@ -2044,6 +2046,12 @@ async def conduct_or_hold( # Carry the observed-so-far kind so a later conduct_from # folds the pre-hold provenance with the replay tail. actuation_kind=actuation_kind, + # A hold of this concern's OWN, not an operator pause: + # without it the default cause would file this under the + # operator's claim, so an operator pausing the same + # conduct would find its hold refused as a duplicate and + # the fault would go unrecorded. + cause=HOLD_CAUSE_STEP_FAULT, ), **envelope_kwargs, ) @@ -2613,7 +2621,12 @@ async def _hold_driver_stood_down( held_ok = False with contextlib.suppress(Exception): await self._hold_procedure( # type: ignore[misc] - HoldProcedure(procedure_id=procedure_id, reason=reason, actuation_kind=folded_kind), + HoldProcedure( + procedure_id=procedure_id, + reason=reason, + actuation_kind=folded_kind, + cause=HOLD_CAUSE_DRIVER_STAND_DOWN, + ), **envelope_kwargs, ) held_ok = True diff --git a/apps/api/src/cora/operation/features/resume_procedure/decider.py b/apps/api/src/cora/operation/features/resume_procedure/decider.py index 74718492f40..679e95d3b6c 100644 --- a/apps/api/src/cora/operation/features/resume_procedure/decider.py +++ b/apps/api/src/cora/operation/features/resume_procedure/decider.py @@ -14,9 +14,25 @@ standalone Procedure (no parent Run). See [[project_resumable_conduct_design]]. +## Who may clear which claim + +A resume discharges claims, and only then, if none remain, moves the status. +A machine concern discharges only the claim it placed. An operator discharges +its own, any legacy claim, and every ATTENTION claim, because an attention +claim is the Conductor reporting a conduct it cannot un-stick and no runtime +will ever come back to clear it. Without that widening, wiring the Conductor's +causes would leave every fault-parked conduct permanently unresumable. See +`ATTENTION_HOLD_CAUSES` for why the same widening would be a bypass on a Run. + +Today every resume in the system is an operator resume: `cause` defaults to +`operator` and no wire surface exposes it. So the refusal below does not fire +in production yet; it is the guard that keeps a future machine resumer from +clearing a hold it never placed, and the behaviour that changes NOW is that a +machine hold is recorded at all. + Invariants: - command.cause must be in HOLD_CAUSES -> ValueError - - The caller must hold an active claim, else other concerns block + - The caller must be entitled to clear at least one active claim -> ProcedureHoldClaimsRemainError(blocking_causes=...) - State must not be None -> ProcedureNotFoundError - command.re_establishment_boundary must be >= 0 @@ -31,6 +47,7 @@ from uuid import UUID from cora.operation.aggregates.procedure import ( + ATTENTION_HOLD_CAUSES, HOLD_CAUSE_OPERATOR, HOLD_CAUSES, LEGACY_CLAIM_ID, @@ -49,6 +66,43 @@ _RESUMABLE_STATUSES: tuple[ProcedureStatus, ...] = (ProcedureStatus.HELD,) +def _clearable_by(procedure_id: UUID, cause: str, own_claim_id: UUID) -> set[UUID]: + """The claims a resumer with this cause is entitled to discharge. + + A machine concern clears only its own. An operator clears its own, plus + every attention claim, plus a legacy claim, because those are precisely the + claims nothing else in CORA will ever clear: + + - attention claims (`ATTENTION_HOLD_CAUSES`) are the Conductor saying it + parked a conduct it cannot un-stick. No runtime discharges them, so + without this an operator resume would raise + `ProcedureHoldClaimsRemainError` on every fault-parked conduct and the + hold would be permanent. + - a hold placed BEFORE holds carried claims has no recorded owner, so no + derived id matches it: every Procedure held at the moment claims + shipped. Clearing one was always the operator's to do. + + `operator` is a CONCERN, not an identity: it is the default cause, so it + means "the caller named no other concern", and nothing here checks what + kind of thing the caller is. Every unmarked resume traces to a person + today only because no agent is granted `ResumeProcedure`, which is what + `test_no_agent_is_granted_resume_procedure` keeps true. + + Widening this for the operator is safe in a way it would NOT be on a Run, + where the machine causes are authority claims a co-signature or a + kill-switch holds deliberately against an operator, and where the wire + deliberately DOES expose `cause` so a person can name the claim they are + clearing. See `ATTENTION_HOLD_CAUSES` for the distinction. + """ + if cause != HOLD_CAUSE_OPERATOR: + return {own_claim_id} + return { + own_claim_id, + LEGACY_CLAIM_ID, + *(derive_claim_id(procedure_id, attention) for attention in ATTENTION_HOLD_CAUSES), + } + + def decide( state: Procedure | None, command: ResumeProcedure, @@ -90,7 +144,16 @@ def _resumed( ) ] - claim_id = derive_claim_id(state.id, command.cause) + def _released(claim_id: UUID, cause: str) -> ProcedureHoldClaimReleased: + return ProcedureHoldClaimReleased( + procedure_id=state.id, + claim_id=claim_id, + cause=cause, + decided_by_decision_id=command.decided_by_decision_id, + occurred_at=now, + ) + + own_claim_id = derive_claim_id(state.id, command.cause) active = tuple(active_id for active_id, _ in state.hold_claims) if not active: # Held with no active claim. Unreachable from a well-formed stream (a @@ -98,35 +161,28 @@ def _resumed( # happens the safety property already holds (no concern is holding # this), so resume rather than wedge the conduct shut. return _resumed(None) - # A hold placed BEFORE holds carried claims has no recorded owner, so no - # derived id matches it and it would otherwise be unresumable: every - # Procedure held at the moment this shipped. The operator is the authority - # that could always clear such a hold, so an operator resume owns the - # legacy claim in addition to its own. - owned = {claim_id} - if command.cause == HOLD_CAUSE_OPERATOR: - owned.add(LEGACY_CLAIM_ID) - held_by_caller = tuple(cid for cid in active if cid in owned) - if not held_by_caller: - # Held, but not by us. Refuse, and name who is holding so the caller - # learns which concern to address rather than only that it was refused. + owned = _clearable_by(state.id, command.cause, own_claim_id) + discharged = tuple((cid, cause) for cid, cause in state.hold_claims if cid in owned) + if not discharged: + # Held, and by nothing this caller may clear. Refuse, and name who is + # holding so the caller learns which concern to address rather than + # only that it was refused. raise ProcedureHoldClaimsRemainError( state.id, blocking_causes=tuple(cause for _, cause in state.hold_claims), ) + if any(cid not in owned for cid in active): + # Something this caller may not clear still holds it: discharge what we + # can and leave the status where it is. + return [_released(cid, cause) for cid, cause in discharged] if set(active) <= {LEGACY_CLAIM_ID}: # A legacy one-bit hold: clearing it means clearing the hold outright, # which is exactly what a bare ProcedureResumed does at the fold. return _resumed(None) - if len(active) > 1: - # Others still hold it: discharge our claim without moving the status. - return [ - ProcedureHoldClaimReleased( - procedure_id=state.id, - claim_id=held_by_caller[0], - cause=command.cause, - decided_by_decision_id=command.decided_by_decision_id, - occurred_at=now, - ) - ] - return _resumed(held_by_caller[0]) + # Nothing will remain, so the conduct runs again. The caller's own claim + # rides the resume; every claim it clears on another concern's behalf gets + # its own event, so a resume that answers three concerns records three + # discharges instead of one status change that silently absorbed them. + extras = [_released(cid, cause) for cid, cause in discharged if cid != own_claim_id] + own = own_claim_id if any(cid == own_claim_id for cid, _ in discharged) else None + return [*extras, *_resumed(own)] diff --git a/apps/api/tests/integration/test_conduct_from_against_softioc_postgres.py b/apps/api/tests/integration/test_conduct_from_against_softioc_postgres.py index 0b32c96021d..7cc186ffb02 100644 --- a/apps/api/tests/integration/test_conduct_from_against_softioc_postgres.py +++ b/apps/api/tests/integration/test_conduct_from_against_softioc_postgres.py @@ -232,6 +232,11 @@ async def test_conduct_or_hold_parks_at_held_then_conduct_from_replays_the_tail( "ProcedureStarted", "ProcedureActivitiesLogbookOpened", "ProcedureHeld", + # The conduct parks under its own `step-fault` claim, so the operator's + # resume discharges that claim by name before the status moves. Two + # events, because "the fault was cleared" and "the conduct runs again" + # became separate facts once more than one concern could hold it. + "ProcedureHoldClaimReleased", "ProcedureResumed", "ProcedureCompleted", ] @@ -317,6 +322,11 @@ async def test_conduct_from_aborts_when_the_replayed_check_still_fails( "ProcedureStarted", "ProcedureActivitiesLogbookOpened", "ProcedureHeld", + # The conduct parks under its own `step-fault` claim, so the operator's + # resume discharges that claim by name before the status moves. Two + # events, because "the fault was cleared" and "the conduct runs again" + # became separate facts once more than one concern could hold it. + "ProcedureHoldClaimReleased", "ProcedureResumed", "ProcedureAborted", ] diff --git a/apps/api/tests/integration/test_paired_hold_authz_postgres.py b/apps/api/tests/integration/test_paired_hold_authz_postgres.py index 05612220df7..7292f8e3a08 100644 --- a/apps/api/tests/integration/test_paired_hold_authz_postgres.py +++ b/apps/api/tests/integration/test_paired_hold_authz_postgres.py @@ -72,7 +72,9 @@ from cora.infrastructure.ports import Allow, Conjunct, Deny from cora.infrastructure.routing import NIL_SENTINEL_ID, SYSTEM_HTTP_SURFACE_ID from cora.operation.aggregates.procedure import ( + HOLD_CAUSE_OPERATOR, ProcedureRegistered, + derive_claim_id, event_type_name, to_payload, ) @@ -293,13 +295,15 @@ async def verdicts_for(command: str) -> dict[str, Allow | Deny]: # equal. A future field written on one path and not the other fails # here without anyone remembering to add an assertion for it. # - # Three fields are stripped and each for its own reason. `stream_id` + # Four fields are stripped and each for its own reason. `stream_id` # and the payload's `procedure_id` name the two different Procedures. - # `principal_id` is the one the claim is about, and it is asserted - # explicitly below rather than merely dropped. + # `claim_id` is DERIVED from that same procedure_id, so it differs for + # exactly the same reason. `principal_id` is the one the claim is about. + # The last two are asserted explicitly below rather than merely dropped. def comparable(row: dict[str, Any]) -> dict[str, Any]: payload = dict(row["payload"]) payload.pop("procedure_id", None) + payload.pop("claim_id", None) return { **{k: v for k, v in row.items() if k not in ("stream_id", "principal_id", "payload")}, "payload": payload, @@ -310,6 +314,16 @@ def comparable(row: dict[str, Any]) -> dict[str, Any]: # The comparison is only worth something if there was something to # compare: an empty dict on both sides would satisfy it silently. assert human_row["payload"], human_row + + # The stripped claim id, checked rather than ignored. Both paths must place + # the hold under the SAME cause, differing only by which Procedure the + # claim is on. An agent whose hold was filed under a different cause than a + # person's would be the two principals shaped differently, which is the + # thing this test exists to refuse. + for proc in (human_proc, agent_proc): + payload = by_stream[proc]["payload"] + assert payload["cause"] == HOLD_CAUSE_OPERATOR, proc + assert payload["claim_id"] == str(derive_claim_id(proc, HOLD_CAUSE_OPERATOR)), proc assert human_row["metadata"], human_row assert by_stream[human_proc]["principal_id"] == human_id assert by_stream[agent_proc]["principal_id"] == agent_id diff --git a/apps/api/tests/unit/api/test_in_process_grants.py b/apps/api/tests/unit/api/test_in_process_grants.py index e2498acc052..3060c9451c8 100644 --- a/apps/api/tests/unit/api/test_in_process_grants.py +++ b/apps/api/tests/unit/api/test_in_process_grants.py @@ -2,8 +2,10 @@ `IN_PROCESS_GRANTS` is inert data: nothing in the running app reads it (only the architecture fitness test and `tools/gen_policy_grants.py` -do), so these tests are a light sanity check on the table's own shape -rather than a behavioral test of anything it drives. +do), so most of these are a light sanity check on the table's own shape +rather than a behavioral test of anything it drives. The exception is +`test_no_agent_is_granted_resume_procedure`, which guards a safety +property the table can silently break. """ from uuid import UUID @@ -35,6 +37,35 @@ def test_no_two_principal_ids_collide() -> None: assert len(principal_ids) == len(set(principal_ids)) +@pytest.mark.unit +def test_no_agent_is_granted_resume_procedure() -> None: + """A machine resumer would inherit an operator's reach without asking. + + `ResumeProcedure.cause` defaults to `operator`, and an operator's resume + clears every ATTENTION claim on a Procedure: the whole point, since nothing + else discharges a fault-parked conduct. Nothing checks the caller's + species, so `operator` means only "the caller named no other concern". An + agent added here would default into it and gain the authority to restart a + conduct a Conductor parked, or to clear a person's deliberate pause, with + no one having decided that. + + That the widening is safe today rests on this table and not on the claim + algebra, which is why the table is where the guard belongs. + + If an agent genuinely needs to resume, the fix is not to delete this test. + Give the concern its own cause in `HOLD_CAUSES`, decide whether it belongs + in `ATTENTION_HOLD_CAUSES`, and set it explicitly at the call site the way + `_run_supervisor` sets `HOLD_CAUSE_SUPERVISOR` on a Run. Then update this + test to require that agent sets a cause rather than to forbid the grant. + """ + holders = [ + principal_id + for principal_id, command_names in IN_PROCESS_GRANTS.items() + if "ResumeProcedure" in command_names + ] + assert holders == [] + + @pytest.mark.unit def test_table_is_read_only() -> None: """`MappingProxyType` refuses mutation; a plain dict here would let a diff --git a/apps/api/tests/unit/operation/_helpers.py b/apps/api/tests/unit/operation/_helpers.py index 528b35bbf99..f1750b15266 100644 --- a/apps/api/tests/unit/operation/_helpers.py +++ b/apps/api/tests/unit/operation/_helpers.py @@ -252,6 +252,7 @@ class Transcript: default_factory=list[dict[str, object]] ) resume_boundaries: list[int] = field(default_factory=list[int]) + hold_causes: list[str] = field(default_factory=list[str]) complete_termination_reasons: list[object] = field(default_factory=list[object]) @@ -300,6 +301,7 @@ async def end_iteration(command: EndProcedureIteration, **_: object) -> None: async def hold_procedure(command: HoldProcedure, **_: object) -> None: transcript.events.append(f"hold_procedure[{command.reason}]") + transcript.hold_causes.append(command.cause) return { "hold_procedure": hold_procedure, diff --git a/apps/api/tests/unit/operation/test_conductor.py b/apps/api/tests/unit/operation/test_conductor.py index 082f18502e1..f418c41b54e 100644 --- a/apps/api/tests/unit/operation/test_conductor.py +++ b/apps/api/tests/unit/operation/test_conductor.py @@ -77,6 +77,7 @@ from cora.infrastructure.routing import NIL_SENTINEL_ID from cora.operation.adapters.control_port_registry import ControlPortRegistry from cora.operation.adapters.in_memory_control_port import InMemoryControlPort +from cora.operation.aggregates.procedure import HOLD_CAUSE_STEP_FAULT from cora.operation.conductor import ( ActionContext, ActionStep, @@ -2616,6 +2617,10 @@ async def test_conduct_or_hold_held_procedure_does_not_run_closing_steps() -> No assert result.held is True assert len(hold.calls) == 1 assert dict(result.substrate_writes) == {"2bma:shutter": 1} # NOT "2bma:closing" + # The conduct holds under its OWN claim. On the command's default cause the + # fault would be filed as an operator pause, so an operator holding the same + # conduct would find their hold refused as a duplicate of it. + assert hold.calls[0].command.cause == HOLD_CAUSE_STEP_FAULT @pytest.mark.unit diff --git a/apps/api/tests/unit/operation/test_conductor_steering_driver_stand_down.py b/apps/api/tests/unit/operation/test_conductor_steering_driver_stand_down.py index 3dd514371b9..be6e44411f8 100644 --- a/apps/api/tests/unit/operation/test_conductor_steering_driver_stand_down.py +++ b/apps/api/tests/unit/operation/test_conductor_steering_driver_stand_down.py @@ -23,6 +23,7 @@ from cora.operation.adapters.in_memory_compute_port import InMemoryComputePort from cora.operation.adapters.in_memory_control_port import InMemoryControlPort from cora.operation.adapters.in_memory_decide_port import InMemoryDecidePort +from cora.operation.aggregates.procedure import HOLD_CAUSE_DRIVER_STAND_DOWN from cora.operation.conductor import Conductor, ConductorResult from cora.operation.ports.decide_port import ( SteeringAdvice, @@ -160,6 +161,10 @@ async def test_loop_holds_at_the_boundary_where_the_driver_was_stood_down() -> N assert "hold_procedure" in " ".join(transcript.events) assert "abort_procedure" not in transcript.events assert "complete_procedure" not in transcript.events + # Its own claim, distinct from a step fault: a stood-down driver is answered + # by reinstating the driver, not by the equipment recovering, so the two + # must be able to hold the same conduct at once. + assert transcript.hold_causes == [HOLD_CAUSE_DRIVER_STAND_DOWN] @pytest.mark.unit diff --git a/apps/api/tests/unit/operation/test_procedure_hold_claim_enforcement.py b/apps/api/tests/unit/operation/test_procedure_hold_claim_enforcement.py index b0eb378bb05..8c8f4f95994 100644 --- a/apps/api/tests/unit/operation/test_procedure_hold_claim_enforcement.py +++ b/apps/api/tests/unit/operation/test_procedure_hold_claim_enforcement.py @@ -12,9 +12,11 @@ import pytest from cora.operation.aggregates.procedure import ( + ATTENTION_HOLD_CAUSES, HOLD_CAUSE_DRIVER_STAND_DOWN, HOLD_CAUSE_OPERATOR, HOLD_CAUSE_STEP_FAULT, + HOLD_CAUSES, LEGACY_CAUSE, LEGACY_CLAIM_ID, Procedure, @@ -26,6 +28,7 @@ ProcedureResumed, ProcedureStatus, derive_claim_id, + evolve, ) from cora.operation.features import hold_procedure, resume_procedure from cora.operation.features.hold_procedure import HoldProcedure @@ -133,15 +136,68 @@ def test_resuming_without_a_claim_is_refused_and_names_the_holders() -> None: rather than only that it was refused.""" with pytest.raises(ProcedureHoldClaimsRemainError) as exc: resume_procedure.decide( - state=_held_by(HOLD_CAUSE_STEP_FAULT, HOLD_CAUSE_DRIVER_STAND_DOWN), - command=_resume(HOLD_CAUSE_OPERATOR), + state=_held_by(HOLD_CAUSE_OPERATOR), + command=_resume(HOLD_CAUSE_STEP_FAULT), now=_NOW, ) - assert exc.value.blocking_causes == ( - HOLD_CAUSE_STEP_FAULT, - HOLD_CAUSE_DRIVER_STAND_DOWN, + assert exc.value.blocking_causes == (HOLD_CAUSE_OPERATOR,) + assert HOLD_CAUSE_OPERATOR in str(exc.value) + + +def test_an_operator_resumes_a_fault_parked_conduct_in_one_act() -> None: + """Wiring the Conductor's causes without this raised + ProcedureHoldClaimsRemainError here, leaving every fault-parked conduct + permanently unresumable: nothing in CORA discharges a step-fault claim.""" + events = resume_procedure.decide( + state=_held_by(HOLD_CAUSE_STEP_FAULT), + command=_resume(HOLD_CAUSE_OPERATOR), + now=_NOW, + ) + assert [type(e) for e in events] == [ProcedureHoldClaimReleased, ProcedureResumed] + assert isinstance(events[0], ProcedureHoldClaimReleased) + assert events[0].cause == HOLD_CAUSE_STEP_FAULT + + +def test_an_operator_resume_records_every_claim_it_clears() -> None: + """One resume, three concerns answered, three discharges in the record. The + status change must not silently absorb the two it did not place.""" + events = resume_procedure.decide( + state=_held_by(HOLD_CAUSE_OPERATOR, HOLD_CAUSE_STEP_FAULT, HOLD_CAUSE_DRIVER_STAND_DOWN), + command=_resume(HOLD_CAUSE_OPERATOR), + now=_NOW, + ) + assert [type(e) for e in events] == [ + ProcedureHoldClaimReleased, + ProcedureHoldClaimReleased, + ProcedureResumed, + ] + released = [e.cause for e in events if isinstance(e, ProcedureHoldClaimReleased)] + assert released == [HOLD_CAUSE_STEP_FAULT, HOLD_CAUSE_DRIVER_STAND_DOWN] + assert isinstance(events[2], ProcedureResumed) + assert events[2].released_claim_id == _claim(HOLD_CAUSE_OPERATOR) + + +def test_the_operator_claim_rides_the_resume_and_the_rest_get_their_own_event() -> None: + """The fold, not just the payload: releases come FIRST, so the claim the + resume does not name is gone by the time the status moves.""" + events = resume_procedure.decide( + state=_held_by(HOLD_CAUSE_OPERATOR, HOLD_CAUSE_STEP_FAULT), + command=_resume(HOLD_CAUSE_OPERATOR), + now=_NOW, ) - assert HOLD_CAUSE_STEP_FAULT in str(exc.value) + state = _held_by(HOLD_CAUSE_OPERATOR, HOLD_CAUSE_STEP_FAULT) + for event in events: + state = evolve(state, event) + assert state.status is ProcedureStatus.RUNNING + assert state.hold_claims == () + + +def test_every_hold_cause_is_classified() -> None: + """A new cause must be put on one side or the other. Defaulting it into + either would answer "may an operator clear this" by inheritance, which is + exactly the question a new concern exists to raise.""" + assert ATTENTION_HOLD_CAUSES | {HOLD_CAUSE_OPERATOR} == HOLD_CAUSES + assert HOLD_CAUSE_OPERATOR not in ATTENTION_HOLD_CAUSES def test_an_operator_resume_owns_a_legacy_claim() -> None: From 028d2491bd78ea400ff0e2bbec22b738a8702fa7 Mon Sep 17 00:00:00 2001 From: xmap <16776958+xmap@users.noreply.github.com> Date: Thu, 10 Sep 2026 12:54:04 -0500 Subject: [PATCH 4/4] Tell a paused conduct apart from a stuck one, instead of waiting a week for both The ProcedureWatcher clocked every Held procedure against one week-long window because `status` is one bit and cannot say WHICH concern is holding. That window exists for a real case: a deliberate operator pause (a bakeout, waiting on beam, waiting on a collaborator) legitimately runs for days, and an hour-long window false-flagged every one of them. But `Held` never meant "deliberate pause". The Conductor parks a conduct on a recoverable step fault or a stood-down steering driver, and that is the opposite case: nothing in CORA will move it until a person comes, so a week of silence is precisely wrong for it. The watcher could not tell the two apart, so it gave every hold the benefit of the doubt and the case most worth surfacing was the one it hid longest. Now that a hold records its cause, it does not have to. The long window is not retired, it is narrowed to what its own config comment always described. What changed is the question that selects it. THE RULE IS POSITIVE EVIDENCE, not absence of a fault. The long window applies only when `hold_causes` is non-empty AND every cause in it is `operator`. An empty set does not qualify, and neither does `legacy-unscoped`: a Held row this watcher cannot vouch for takes the short window, because treating "no evidence" as "deliberate" is exactly how the case worth catching would go on hiding. Being wrong costs one advisory Decision a person can ignore, which is the cheap direction for a flag-only watcher. Two consequences worth stating: a deployment carrying pre-claim holds gets one flag per procedure, which is the right answer to "nobody can say why this is held"; and an operator pausing a conduct does not vouch for a fault that arrived alongside it, so one non-operator cause loses the long window. The causes reach the watcher through the read model. `hold_causes text[]` is denormed onto `proj_operation_procedure_summary`, maintained by the Held (append) / HoldClaimReleased (remove) / Resumed and three terminal (clear) arms. Denorming the CAUSES rather than a precomputed "needs attention" flag keeps the classification in `ATTENTION_HOLD_CAUSES`: baking it into the projection would leave old rows silently wrong the day that set changes, and it makes "held by what" answerable on a plain list read. The alternative considered was folding the aggregate per candidate, which `_run_supervisor` already does for Runs; it was rejected for costing a full stream fold per held procedure per tick and for leaving the fact invisible to every other reader. The append is deduplicated by cause rather than a bare `array_append`, since a re-delivered `ProcedureHeld` must not stack a second copy. That mirrors the aggregate, where `_with_claim` is idempotent on an already-active claim. A pre-claim hold carries no cause and folds to `LEGACY_CAUSE` there, so it lands as that string here too rather than as a Held row claiming nothing holds it. The migration backfills currently-Held rows the same way. THE UNIT TESTS COULD NOT HAVE CAUGHT A BROKEN ARRAY UPDATE: the projection's tests mock the connection, so this SQL had never executed. The Postgres integration test now drives the whole chain instead, a conduct parking itself, a person pausing the same conduct, causes accumulating oldest-first, and one resume draining both. ONE MISS WORTH RECORDING. The architecture suite passed 51,665 tests with `EXPECTED_SCHEMA_VERSION` stale, because `test_schema_version_pin` reads `tracked_migration_files()` and the new migration was still untracked. The integration gate, which reads the directory, failed 15 tests and errored 8 on one root cause. The pin's guard was never wrong; running it before staging was. Confirmed by reverting the pin with the file tracked, which fails it immediately. PRE-COMMIT REJECTED THE FIRST ATTEMPT, on a boundary worth recording rather than working around. `cora.api` may not import `cora.operation.aggregates`, and the watcher was reaching in for `HOLD_CAUSE_OPERATOR` to restate the rule locally. The rule is domain knowledge about causes, so it moved to `is_deliberate_pause` beside the causes it classifies, where whoever adds the next cause will see it; the watcher applies it rather than repeating it. That still needs the dependency, which is now declared with its reason alongside the six `*.aggregates` edges `cora.api` already carries, including the `cora.access.aggregates` one this same runtime uses for `Actor.active`. Stating the rule as positive evidence rather than as "no attention cause is present" also settles a case that does not exist yet: a cause that is neither `operator` nor an attention claim, which is what an authority claim on a Procedure would be. Such a hold is not somebody taking a break, and under the negative phrasing it would have inherited a pause's latitude by being unclassified. Verified by mutation: treating an empty cause set as a deliberate pause, and selecting the window by status again, each fail tests that name the behaviour. Co-Authored-By: Claude Opus 5 (1M context) --- apps/api/src/cora/api/_procedure_watcher.py | 58 ++++++-- apps/api/src/cora/infrastructure/config.py | 15 +- .../src/cora/infrastructure/schema_version.py | 2 +- .../aggregates/procedure/__init__.py | 2 + .../operation/aggregates/procedure/events.py | 28 +++- .../features/list_procedures/handler.py | 13 +- .../cora/operation/projections/procedure.py | 55 +++++++- apps/api/tach.toml | 8 ++ .../test_held_status_projection_postgres.py | 65 ++++++++- .../tests/unit/api/test_procedure_watcher.py | 129 +++++++++++++++++- apps/api/tests/unit/api/test_status_push.py | 1 + .../test_procedure_hold_claim_enforcement.py | 18 +++ .../test_procedure_summary_projection.py | 101 +++++++++++++- ...3339_procedure_summary_add_hold_causes.sql | 34 +++++ infra/atlas/migrations/atlas.sum | 3 +- 15 files changed, 498 insertions(+), 34 deletions(-) create mode 100644 infra/atlas/migrations/20260911193339_procedure_summary_add_hold_causes.sql diff --git a/apps/api/src/cora/api/_procedure_watcher.py b/apps/api/src/cora/api/_procedure_watcher.py index d6642c1dfb2..6896e051b02 100644 --- a/apps/api/src/cora/api/_procedure_watcher.py +++ b/apps/api/src/cora/api/_procedure_watcher.py @@ -23,16 +23,29 @@ `stalled_seconds = now - last_progress_at`. -`Running` and `Held` are clocked against SEPARATE operator-config windows -(`procedure_watcher_stale_after_seconds` and -`procedure_watcher_held_stale_after_seconds`), because the two statuses mean -different things: a `Running` procedure sitting idle for an hour is a plausible -stall, but a `Held` procedure sitting for an hour is routinely a deliberate -operator pause (a bakeout, waiting on beam, waiting on a collaborator) that can -legitimately run for days. One shared window would either false-flag an -ordinary overnight hold or blind the watcher to a genuinely stuck `Running` -conduct; this mirrors `_campaign_watcher`'s week-long default for its own -`Held` window. +Two operator-config windows (`procedure_watcher_stale_after_seconds` and +`procedure_watcher_held_stale_after_seconds`) are selected per procedure, and +what selects them is NOT the status. A `Running` procedure idle for an hour is +a plausible stall; a `Held` one is routinely a deliberate operator pause (a +bakeout, waiting on beam, waiting on a collaborator) that legitimately runs for +days, which is why the long window exists and mirrors `_campaign_watcher`'s. + +But `Held` alone never meant "deliberate pause". The Conductor parks a conduct +on a recoverable step fault or a stood-down steering driver, and that is the +OPPOSITE case: nothing in CORA will move it until a person comes, so a week of +silence is precisely wrong for it. While a hold was one bit, the watcher could +not tell the two apart and gave every hold the benefit of the doubt. Now that a +hold records its cause, it does not have to. + +So the long window applies only on POSITIVE evidence of a deliberate pause: +`hold_causes` is non-empty and every cause in it is `operator`. Everything else +gets the short one. That is deliberately the loud direction on missing or +unrecognized evidence, because the whole point is that a machine-parked conduct +must not hide, and being wrong costs one advisory Decision a person can ignore. +A hold placed before causes were recorded folds to `legacy-unscoped`, which is +not evidence of anything and so gets the short window; on a deployment carrying +such holds that is a one-time flag per procedure, which is the correct answer +to "nobody can say why this is held". For `Held` the conduct is paused and accepts no activity, so `last_status_changed_at` (the time it was held, on the list projection) is the @@ -84,6 +97,7 @@ from cora.operation.adapters.postgres_procedure_activity_lookup import ( PostgresProcedureActivityLookup, ) +from cora.operation.aggregates.procedure import is_deliberate_pause from cora.operation.errors import UnauthorizedError from cora.operation.features.list_procedures import ListProcedures from cora.operation.ports import InMemoryProcedureActivityLookup @@ -136,6 +150,7 @@ async def _record_decision( *, procedure_id: UUID, status: str, + hold_causes: tuple[str, ...], last_progress_at: datetime, now: datetime, ) -> None: @@ -152,13 +167,17 @@ async def _record_decision( entity_id=procedure_id, now=now, reasoning=( - f"Procedure has been {status} for {stalled_seconds}s without progressing " - "(past the staleness window, no recent activity); surfaced for operator " - "follow-up." + f"Procedure has been {status}{_held_by_clause(hold_causes)} for " + f"{stalled_seconds}s without progressing (past the staleness window, no " + "recent activity); surfaced for operator follow-up." ), inputs={ "procedure_id": str(procedure_id), "status": status, + # The causes ride the record because they are what chose the + # window: without them the Decision cannot be read back to see + # whether the short window was applied for the right reason. + "hold_causes": ",".join(hold_causes), "last_progress_at": last_progress_at.isoformat(), "stalled_seconds": str(stalled_seconds), "occurred_at": now.isoformat(), @@ -223,7 +242,10 @@ async def _watch_tick( if base is None: # No status-change timestamp recorded: cannot evaluate; defer. continue - stale_after = stale_after_held if item.status == _STATUS_HELD else stale_after_running + hold_causes = tuple(item.hold_causes) + stale_after = stale_after_running + if item.status == _STATUS_HELD and is_deliberate_pause(hold_causes): + stale_after = stale_after_held if not is_stalled(base, now, stale_after): # Fresh by its status's own window. For Running a later activity only # makes it fresher, so skipping here cannot hide a stall. @@ -244,6 +266,7 @@ async def _watch_tick( deps, procedure_id=item.procedure_id, status=item.status, + hold_causes=hold_causes, last_progress_at=last_progress_at, now=now, ) @@ -289,4 +312,11 @@ async def startup_probe() -> None: yield +def _held_by_clause(hold_causes: tuple[str, ...]) -> str: + """The ` (held by x, y)` fragment, empty when nothing is recorded.""" + if not hold_causes: + return "" + return f" (held by {', '.join(hold_causes)})" + + __all__ = ["is_stalled", "procedure_watcher_lifespan"] diff --git a/apps/api/src/cora/infrastructure/config.py b/apps/api/src/cora/infrastructure/config.py index 8c0791c950d..963b41e8cc6 100644 --- a/apps/api/src/cora/infrastructure/config.py +++ b/apps/api/src/cora/infrastructure/config.py @@ -529,12 +529,15 @@ class Settings(BaseSettings): # sit without progressing before it is flagged; live conduct is far # shorter-lived than a clearance or calibration, so the default is an hour # (off by default; an operator sets the real window on enable). - # `procedure_watcher_held_stale_after_seconds` is the separate window for a - # Held procedure: a hold is commonly a deliberate operator pause (a bakeout, - # waiting on beam, waiting on a collaborator) that legitimately runs far - # longer than an hour, and a Held conduct logs no activity to fold in as a - # second chance, so it needs its own, much longer default; matches - # `campaign_watcher_stale_after_seconds`'s Held precedent, a week. + # `procedure_watcher_held_stale_after_seconds` is the window for a DELIBERATE + # operator pause (a bakeout, waiting on beam, waiting on a collaborator), + # which legitimately runs far longer than an hour and logs no activity to + # fold in as a second chance; matches `campaign_watcher_stale_after_seconds`'s + # Held precedent, a week. It is selected by the hold's recorded CAUSE, not by + # the Held status: a conduct the Conductor parked on a fault will not move + # until a person comes, so it takes the shorter window above. Anything the + # record cannot positively show to be an operator pause takes the short one + # too. procedure_watcher_enabled: bool = False procedure_watcher_tick_seconds: float = 300.0 procedure_watcher_stale_after_seconds: float = 3600.0 diff --git a/apps/api/src/cora/infrastructure/schema_version.py b/apps/api/src/cora/infrastructure/schema_version.py index 62c3f357887..7bcc6e71a9e 100644 --- a/apps/api/src/cora/infrastructure/schema_version.py +++ b/apps/api/src/cora/infrastructure/schema_version.py @@ -74,7 +74,7 @@ class SchemaCheck: expected: str -EXPECTED_SCHEMA_VERSION: Final = "20260910222120" +EXPECTED_SCHEMA_VERSION: Final = "20260911193339" """The newest migration this build was written against. Hand-maintained, and deliberately not derived at runtime: the image does diff --git a/apps/api/src/cora/operation/aggregates/procedure/__init__.py b/apps/api/src/cora/operation/aggregates/procedure/__init__.py index ed9372c68ff..851f4167240 100644 --- a/apps/api/src/cora/operation/aggregates/procedure/__init__.py +++ b/apps/api/src/cora/operation/aggregates/procedure/__init__.py @@ -51,6 +51,7 @@ SteeringDesignRecorded, event_type_name, from_stored, + is_deliberate_pause, to_payload, ) from cora.operation.aggregates.procedure.evolver import evolve, fold @@ -223,6 +224,7 @@ "evolve", "fold", "from_stored", + "is_deliberate_pause", "load_procedure", "load_procedure_with_events", "merge_actuation_kinds", diff --git a/apps/api/src/cora/operation/aggregates/procedure/events.py b/apps/api/src/cora/operation/aggregates/procedure/events.py index 35c5e94fb73..0155e6c0cd1 100644 --- a/apps/api/src/cora/operation/aggregates/procedure/events.py +++ b/apps/api/src/cora/operation/aggregates/procedure/events.py @@ -64,7 +64,7 @@ """ import json -from collections.abc import Mapping +from collections.abc import Iterable, Mapping from dataclasses import dataclass from datetime import datetime from typing import Any, assert_never @@ -491,6 +491,31 @@ class ProcedureAborted: at which "may an operator clear this" has to be answered rather than inherited. """ + +def is_deliberate_pause(hold_causes: Iterable[str]) -> bool: + """Did a person choose to pause this, as opposed to something getting stuck? + + True only when at least one cause is recorded and every one of them is + `operator`. Two things follow from stating it that way rather than as "no + attention cause is present". + + An empty set is NOT a deliberate pause. A Held Procedure recording nothing + that holds it is one nothing can vouch for, including the `LEGACY_CAUSE` + holds placed before causes existed, and reading "no evidence" as "a person + meant this" is how a stuck conduct would go unnoticed. + + Nor is a cause that is neither `operator` nor an attention claim, which is + what a future authority claim on a Procedure would be. Such a hold is not + someone taking a break either, so it does not inherit a deliberate pause's + latitude by being unclassified. + + Its caller is the ProcedureWatcher, which grants a much longer staleness + window to a pause than to a conduct nobody is coming back to. + """ + causes = tuple(hold_causes) + return bool(causes) and all(cause == HOLD_CAUSE_OPERATOR for cause in causes) + + LEGACY_CLAIM_ID = UUID("01900000-0000-7000-8000-00000001dead") """The single claim a pre-claim `ProcedureHeld` folds to, so a legacy stream replays to exactly its old one-bit meaning. Mirrors Run's sentinel.""" @@ -1683,5 +1708,6 @@ def _build_resumed() -> ProcedureResumed: "SteeringDesignRecorded", "event_type_name", "from_stored", + "is_deliberate_pause", "to_payload", ] diff --git a/apps/api/src/cora/operation/features/list_procedures/handler.py b/apps/api/src/cora/operation/features/list_procedures/handler.py index 0704c9706e8..4b4bfe8e66f 100644 --- a/apps/api/src/cora/operation/features/list_procedures/handler.py +++ b/apps/api/src/cora/operation/features/list_procedures/handler.py @@ -48,6 +48,15 @@ class ProcedureSummaryItem: last_status_reason: str | None interrupted_at: datetime | None iteration_count: int + hold_causes: list[str] + """Which concerns are holding this Procedure, oldest first; empty unless Held. + + `status` says a conduct is paused, never by whom, and the two holds that + reach a Procedure want opposite responses: an operator pause is deliberate + and can legitimately run for days, while a conduct the Conductor parked on + a fault will not move until a person comes. Classified by + `ATTENTION_HOLD_CAUSES`, which is deliberately NOT applied here: the causes + ride the row so the rule stays in one place.""" @dataclass(frozen=True) @@ -74,7 +83,8 @@ async def __call__( _SELECT_COLUMNS = ( "procedure_id, name, kind, target_asset_ids, parent_run_id, status, " "activity_logbook_id, registered_at, " - "last_status_changed_at, last_status_reason, interrupted_at, iteration_count" + "last_status_changed_at, last_status_reason, interrupted_at, iteration_count, " + "hold_causes" ) @@ -94,6 +104,7 @@ def _row_to_item(row: Any) -> ProcedureSummaryItem: ), interrupted_at=row["interrupted_at"], iteration_count=int(row["iteration_count"]), + hold_causes=[str(cause) for cause in (row["hold_causes"] or ())], ) diff --git a/apps/api/src/cora/operation/projections/procedure.py b/apps/api/src/cora/operation/projections/procedure.py index fd2700243a9..92ea71608f1 100644 --- a/apps/api/src/cora/operation/projections/procedure.py +++ b/apps/api/src/cora/operation/projections/procedure.py @@ -14,9 +14,15 @@ + interrupted_at - ProcedureHeld -> UPDATE status='Held' + status-change ts + last_status_reason + + hold_causes (append) + - ProcedureHoldClaimReleased -> UPDATE hold_causes (remove); status NOT + touched, since one concern letting + go does not decide whether the + Procedure runs again - ProcedureResumed -> UPDATE status='Running' + status-change ts (clears last_status_reason: Running is not reason-bearing) + + hold_causes = {} - ProcedureActivitiesLogbookOpened -> UPDATE activity_logbook_id (status NOT touched; logbook is orthogonal to lifecycle) @@ -40,6 +46,23 @@ conditional columns and read worse than the explicit constants. Revisit only if a future arm restores uniformity. +## hold_causes + +Which concerns are holding, oldest first, denormed so a reader can ask "held +by what" without folding the aggregate. Its consumer is the ProcedureWatcher, +which used to clock every hold against one week-long window because `status` +could not tell a deliberate operator pause from a conduct the Conductor parked +on a fault. The CAUSES are denormed rather than a precomputed +"needs attention" flag: the classification lives in `ATTENTION_HOLD_CAUSES`, +and baking it in here would leave old rows silently wrong the day it changes. + +Append is deduplicated by claim CAUSE rather than by `array_append` alone, +because a re-delivered `ProcedureHeld` must not stack a second copy. That +mirrors the aggregate, where `_with_claim` is idempotent on an already-active +claim id. A pre-claim `ProcedureHeld` carries no cause and folds to +`LEGACY_CAUSE` in the aggregate, so it lands as that string here too rather +than as an empty append. + All branches idempotent. The status CHECK was widened to admit 'Held' in migration `20260621060000_proc_summary_status_admit_held` (Resumed maps back to 'Running', so 'Held' is the only new persisted value). See @@ -53,14 +76,15 @@ from cora.infrastructure.ports.event_store import StoredEvent from cora.infrastructure.projection.handler import ConnectionLike +from cora.operation.aggregates.procedure import LEGACY_CAUSE _INSERT_PROCEDURE_SQL = """ INSERT INTO proj_operation_procedure_summary (procedure_id, name, kind, target_asset_ids, parent_run_id, status, activity_logbook_id, registered_at, last_status_changed_at, last_status_reason, interrupted_at, - recipe_id, iteration_count) -VALUES ($1, $2, $3, $4::uuid[], $5, 'Defined', NULL, $6, NULL, NULL, NULL, $7, 0) + recipe_id, iteration_count, hold_causes) +VALUES ($1, $2, $3, $4::uuid[], $5, 'Defined', NULL, $6, NULL, NULL, NULL, $7, 0, '{}') ON CONFLICT (procedure_id) DO NOTHING """ @@ -76,6 +100,7 @@ UPDATE proj_operation_procedure_summary SET status = 'Completed', last_status_changed_at = $2, + hold_causes = '{}', updated_at = now() WHERE procedure_id = $1 """ @@ -85,6 +110,7 @@ SET status = 'Aborted', last_status_changed_at = $2, last_status_reason = $3, + hold_causes = '{}', updated_at = now() WHERE procedure_id = $1 """ @@ -95,6 +121,7 @@ last_status_changed_at = $2, last_status_reason = $3, interrupted_at = $4, + hold_causes = '{}', updated_at = now() WHERE procedure_id = $1 """ @@ -104,6 +131,16 @@ SET status = 'Held', last_status_changed_at = $2, last_status_reason = $3, + hold_causes = CASE WHEN $4 = ANY(hold_causes) + THEN hold_causes + ELSE array_append(hold_causes, $4::text) END, + updated_at = now() +WHERE procedure_id = $1 +""" + +_UPDATE_HOLD_CLAIM_RELEASED_SQL = """ +UPDATE proj_operation_procedure_summary +SET hold_causes = array_remove(hold_causes, $2::text), updated_at = now() WHERE procedure_id = $1 """ @@ -113,6 +150,7 @@ SET status = 'Running', last_status_changed_at = $2, last_status_reason = NULL, + hold_causes = '{}', updated_at = now() WHERE procedure_id = $1 """ @@ -144,6 +182,7 @@ class ProcedureSummaryProjection: "ProcedureAborted", "ProcedureTruncated", "ProcedureHeld", + "ProcedureHoldClaimReleased", "ProcedureResumed", "ProcedureActivitiesLogbookOpened", "ProcedureIterationStarted", @@ -220,11 +259,23 @@ async def apply( return if event.event_type == "ProcedureHeld": + # A pre-claim hold carries no cause; the aggregate folds it to + # LEGACY_CAUSE, so the denorm says the same rather than recording + # a Held row that nothing appears to hold. await conn.execute( _UPDATE_HELD_SQL, UUID(event.payload["procedure_id"]), datetime.fromisoformat(event.payload["occurred_at"]), event.payload["reason"], + event.payload.get("cause") or LEGACY_CAUSE, + ) + return + + if event.event_type == "ProcedureHoldClaimReleased": + await conn.execute( + _UPDATE_HOLD_CLAIM_RELEASED_SQL, + UUID(event.payload["procedure_id"]), + event.payload["cause"], ) return diff --git a/apps/api/tach.toml b/apps/api/tach.toml index 525c1e64cc6..dc6c3c3ba4e 100644 --- a/apps/api/tach.toml +++ b/apps/api/tach.toml @@ -452,6 +452,14 @@ depends_on = [ "cora.equipment.aggregates", "cora.federation", "cora.operation", + # ProcedureWatcher asks the Operation BC whether a Held procedure's recorded + # hold causes amount to a deliberate operator pause (is_deliberate_pause), + # which decides the staleness window it clocks that procedure against. The + # rule lives beside the causes it classifies rather than being restated at + # the composition root. Blessed cross-BC aggregate read, same pattern as + # cora.access.aggregates, which this same runtime already uses for + # Actor.active. + "cora.operation.aggregates", "cora.recipe", # Two composition-root consumers read recipe aggregates: the RunSupervisor # gated-resume re-check (load Plan/Practice/Method, see cora.equipment.aggregates diff --git a/apps/api/tests/integration/test_held_status_projection_postgres.py b/apps/api/tests/integration/test_held_status_projection_postgres.py index 71fc5d8be6b..17e530494eb 100644 --- a/apps/api/tests/integration/test_held_status_projection_postgres.py +++ b/apps/api/tests/integration/test_held_status_projection_postgres.py @@ -12,6 +12,9 @@ - ProcedureResumed folds back to status='Running' and clears last_status_reason (Running is not reason-bearing). - The list_procedures read path surfaces + filters on status='Held'. + - `hold_causes` accumulates per concern and drains per discharge. The + unit tests mock the connection, so the array SQL (a deduplicated + append, an array_remove) executes for the first time here. """ # pyright: reportUnknownMemberType=false, reportUnknownVariableType=false, reportUnknownArgumentType=false @@ -25,6 +28,10 @@ from cora.infrastructure.kernel import Kernel from cora.infrastructure.projection import ProjectionRegistry, drain_projections from cora.operation._projections import register_operation_projections +from cora.operation.aggregates.procedure import ( + HOLD_CAUSE_OPERATOR, + HOLD_CAUSE_STEP_FAULT, +) from cora.operation.features.hold_procedure import HoldProcedure from cora.operation.features.hold_procedure import bind as bind_hold from cora.operation.features.list_procedures import ListProcedures @@ -56,7 +63,8 @@ async def _drain(db_pool: asyncpg.Pool) -> None: async def _status_row(db_pool: asyncpg.Pool, proc_id: UUID) -> asyncpg.Record: async with db_pool.acquire() as conn: row = await conn.fetchrow( - "SELECT status, last_status_reason FROM proj_operation_procedure_summary " + "SELECT status, last_status_reason, hold_causes " + "FROM proj_operation_procedure_summary " "WHERE procedure_id = $1", proc_id, ) @@ -91,6 +99,7 @@ async def test_hold_then_resume_drives_status_in_read_model(db_pool: asyncpg.Poo held = await _status_row(db_pool, proc_id) assert held["status"] == "Held" assert held["last_status_reason"] == "beam dropped" + assert held["hold_causes"] == ["operator"] # The list read path surfaces + filters on the new status. page = await bind_list(deps)( @@ -112,3 +121,57 @@ async def test_hold_then_resume_drives_status_in_read_model(db_pool: asyncpg.Poo resumed = await _status_row(db_pool, proc_id) assert resumed["status"] == "Running" assert resumed["last_status_reason"] is None + assert resumed["hold_causes"] == [] + + +@pytest.mark.integration +async def test_two_concerns_accumulate_and_drain_in_the_read_model( + db_pool: asyncpg.Pool, +) -> None: + """The whole chain in the read model: a conduct parks itself, a person + pauses the same conduct, and the person's resume discharges both. + + What this catches that the mocked unit tests cannot is the array SQL + itself, including that a re-delivered hold does not stack a second copy + of a cause it already carries. + """ + proc_id = uuid4() + deps = _build_deps(db_pool, [proc_id, *[uuid4() for _ in range(8)]]) + + await bind_register(deps)( + RegisterProcedure(name="2-BM rotation alignment", kind="center_alignment"), + principal_id=_PRINCIPAL_ID, + correlation_id=_CORRELATION_ID, + ) + await bind_start(deps)( + StartProcedure(procedure_id=proc_id), + principal_id=_PRINCIPAL_ID, + correlation_id=_CORRELATION_ID, + ) + + for cause, reason in ( + (HOLD_CAUSE_STEP_FAULT, "check on 2bma:rot:rbv did not settle"), + (HOLD_CAUSE_OPERATOR, "swapping the sample"), + ): + await bind_hold(deps)( + HoldProcedure(procedure_id=proc_id, reason=reason, cause=cause), + principal_id=_PRINCIPAL_ID, + correlation_id=_CORRELATION_ID, + ) + await _drain(db_pool) + both = await _status_row(db_pool, proc_id) + assert both["status"] == "Held" + # Oldest first, mirroring Procedure.hold_claims. + assert both["hold_causes"] == [HOLD_CAUSE_STEP_FAULT, HOLD_CAUSE_OPERATOR] + + # One resume, both concerns answered: the operator's claim rides the + # ProcedureResumed and the fault's gets its own release event. + await bind_resume(deps)( + ResumeProcedure(procedure_id=proc_id, re_establishment_boundary=0), + principal_id=_PRINCIPAL_ID, + correlation_id=_CORRELATION_ID, + ) + await _drain(db_pool) + cleared = await _status_row(db_pool, proc_id) + assert cleared["status"] == "Running" + assert cleared["hold_causes"] == [] diff --git a/apps/api/tests/unit/api/test_procedure_watcher.py b/apps/api/tests/unit/api/test_procedure_watcher.py index a460f0cef99..d118986bfa2 100644 --- a/apps/api/tests/unit/api/test_procedure_watcher.py +++ b/apps/api/tests/unit/api/test_procedure_watcher.py @@ -104,6 +104,7 @@ def _item( *, status: str = "Running", last_status_changed_at: datetime | None, + hold_causes: list[str] | None = None, ) -> ProcedureSummaryItem: return ProcedureSummaryItem( procedure_id=procedure_id, @@ -118,6 +119,7 @@ def _item( last_status_reason=None, interrupted_at=None, iteration_count=0, + hold_causes=hold_causes if hold_causes is not None else [], ) @@ -172,7 +174,7 @@ async def test_tick_flags_stale_held_without_folding_activity() -> None: await seed_procedure_watcher_agent(kernel) pid = uuid4() list_procedures = _make_list_procedures( - [_item(pid, status="Held", last_status_changed_at=_OLD)] + [_item(pid, status="Held", last_status_changed_at=_OLD, hold_causes=["operator"])] ) lookup = InMemoryProcedureActivityLookup() lookup.register(procedure_id=pid, recorded_at=_RECENT) # must NOT rescue a Held @@ -187,14 +189,21 @@ async def test_tick_does_not_flag_held_overnight_pause() -> None: """Regression test: Held used to share Running's 1-hour window, so any deliberate operator pause longer than an hour (a bakeout, waiting on beam, waiting on a collaborator) raised a false Stall. Under the real default - (a week), a 12-hour hold is not stalled.""" + (a week), a 12-hour OPERATOR hold is not stalled.""" from cora.api._procedure_watcher import _watch_tick kernel = _kernel() # held_stale_after defaults to the real week-long window await seed_procedure_watcher_agent(kernel) pid = uuid4() list_procedures = _make_list_procedures( - [_item(pid, status="Held", last_status_changed_at=_OVERNIGHT_HELD)] + [ + _item( + pid, + status="Held", + last_status_changed_at=_OVERNIGHT_HELD, + hold_causes=["operator"], + ) + ] ) await _watch_tick( @@ -207,6 +216,106 @@ async def test_tick_does_not_flag_held_overnight_pause() -> None: assert await load_decision(kernel.event_store, decision_id) is None +@pytest.mark.unit +async def test_tick_flags_a_machine_parked_conduct_long_before_the_held_window() -> None: + """The point of the slice. The same 12 hours that is an ordinary pause for + an operator is a conduct nobody has come back to: the Conductor cannot + un-stick a faulted step, so waiting out a week-long window would hide + exactly the case worth surfacing.""" + from cora.api._procedure_watcher import _watch_tick + + kernel = _kernel() # the real week-long Held window + await seed_procedure_watcher_agent(kernel) + pid = uuid4() + list_procedures = _make_list_procedures( + [ + _item( + pid, + status="Held", + last_status_changed_at=_OVERNIGHT_HELD, + hold_causes=["step-fault"], + ) + ] + ) + + await _watch_tick( + deps=kernel, + list_procedures=list_procedures, + activity_lookup=InMemoryProcedureActivityLookup(), + ) + + decision = await load_decision(kernel.event_store, _derive_decision_id(pid, _OVERNIGHT_HELD)) + assert decision is not None + # The causes ride the Decision: a reader can see WHY the short window + # applied rather than having to re-derive it. + assert decision.reasoning is not None + assert "step-fault" in decision.reasoning + + +@pytest.mark.unit +async def test_tick_flags_a_pause_an_operator_shares_with_a_fault() -> None: + """A person pausing a conduct does not vouch for a fault that arrived + alongside it. One non-operator cause is enough to lose the long window.""" + from cora.api._procedure_watcher import _watch_tick + + kernel = _kernel() + await seed_procedure_watcher_agent(kernel) + pid = uuid4() + list_procedures = _make_list_procedures( + [ + _item( + pid, + status="Held", + last_status_changed_at=_OVERNIGHT_HELD, + hold_causes=["operator", "step-fault"], + ) + ] + ) + + await _watch_tick( + deps=kernel, + list_procedures=list_procedures, + activity_lookup=InMemoryProcedureActivityLookup(), + ) + + assert await load_decision(kernel.event_store, _derive_decision_id(pid, _OVERNIGHT_HELD)) + + +@pytest.mark.unit +async def test_tick_flags_a_held_procedure_recording_no_cause_at_all() -> None: + """No evidence is not evidence of a deliberate pause. A Held row holding + nothing is one this watcher cannot vouch for, including the + `legacy-unscoped` holds placed before causes were recorded, so it takes the + short window and costs one advisory a person can ignore.""" + from cora.api._procedure_watcher import _watch_tick + + kernel = _kernel() + await seed_procedure_watcher_agent(kernel) + unrecorded, legacy = uuid4(), uuid4() + list_procedures = _make_list_procedures( + [ + _item(unrecorded, status="Held", last_status_changed_at=_OVERNIGHT_HELD), + _item( + legacy, + status="Held", + last_status_changed_at=_OVERNIGHT_HELD, + hold_causes=["legacy-unscoped"], + ), + ] + ) + + await _watch_tick( + deps=kernel, + list_procedures=list_procedures, + activity_lookup=InMemoryProcedureActivityLookup(), + ) + + for pid in (unrecorded, legacy): + assert await load_decision(kernel.event_store, _derive_decision_id(pid, _OVERNIGHT_HELD)), ( + pid + ) + + @pytest.mark.unit async def test_tick_does_not_flag_running_with_recent_activity() -> None: """The anti-false-flag fold: a Running procedure that looks stale by its @@ -330,10 +439,20 @@ async def test_record_decision_is_idempotent_on_repeated_episode() -> None: kernel = _kernel() pid = uuid4() await _record_decision( - kernel, procedure_id=pid, status="Running", last_progress_at=_OLD, now=_NOW + kernel, + procedure_id=pid, + status="Running", + hold_causes=(), + last_progress_at=_OLD, + now=_NOW, ) await _record_decision( - kernel, procedure_id=pid, status="Running", last_progress_at=_OLD, now=_NOW + kernel, + procedure_id=pid, + status="Running", + hold_causes=(), + last_progress_at=_OLD, + now=_NOW, ) assert await load_decision(kernel.event_store, _derive_decision_id(pid, _OLD)) is not None diff --git a/apps/api/tests/unit/api/test_status_push.py b/apps/api/tests/unit/api/test_status_push.py index a00da6f6853..d13462d63d9 100644 --- a/apps/api/tests/unit/api/test_status_push.py +++ b/apps/api/tests/unit/api/test_status_push.py @@ -1680,6 +1680,7 @@ def _procedure_item( last_status_reason="operator said something private", interrupted_at=None, iteration_count=3, + hold_causes=[], ) diff --git a/apps/api/tests/unit/operation/test_procedure_hold_claim_enforcement.py b/apps/api/tests/unit/operation/test_procedure_hold_claim_enforcement.py index 8c8f4f95994..299801191fc 100644 --- a/apps/api/tests/unit/operation/test_procedure_hold_claim_enforcement.py +++ b/apps/api/tests/unit/operation/test_procedure_hold_claim_enforcement.py @@ -29,6 +29,7 @@ ProcedureStatus, derive_claim_id, evolve, + is_deliberate_pause, ) from cora.operation.features import hold_procedure, resume_procedure from cora.operation.features.hold_procedure import HoldProcedure @@ -192,6 +193,23 @@ def test_the_operator_claim_rides_the_resume_and_the_rest_get_their_own_event() assert state.hold_claims == () +def test_only_an_all_operator_hold_is_a_deliberate_pause() -> None: + """Stated as positive evidence, so the two cases that would otherwise + inherit a pause's latitude do not. + + An empty set is a Held Procedure recording nothing that holds it, which is + the shape a pre-claim stream and a corrupt row both take. A cause that is + neither `operator` nor an attention claim is what a future authority claim + would be, and that is not someone taking a break either. + """ + assert is_deliberate_pause([HOLD_CAUSE_OPERATOR]) + assert not is_deliberate_pause([]) + assert not is_deliberate_pause([LEGACY_CAUSE]) + assert not is_deliberate_pause([HOLD_CAUSE_STEP_FAULT]) + assert not is_deliberate_pause([HOLD_CAUSE_OPERATOR, HOLD_CAUSE_DRIVER_STAND_DOWN]) + assert not is_deliberate_pause([HOLD_CAUSE_OPERATOR, "some-future-authority-claim"]) + + def test_every_hold_cause_is_classified() -> None: """A new cause must be put on one side or the other. Defaulting it into either would answer "may an operator clear this" by inheritance, which is diff --git a/apps/api/tests/unit/operation/test_procedure_summary_projection.py b/apps/api/tests/unit/operation/test_procedure_summary_projection.py index 884b83b9ea5..ad7f70c8a14 100644 --- a/apps/api/tests/unit/operation/test_procedure_summary_projection.py +++ b/apps/api/tests/unit/operation/test_procedure_summary_projection.py @@ -13,6 +13,7 @@ import pytest from cora.infrastructure.ports.event_store import StoredEvent +from cora.operation.aggregates.procedure import LEGACY_CAUSE from cora.operation.projections import ProcedureSummaryProjection _PROCEDURE_ID = uuid4() @@ -50,6 +51,7 @@ def test_projection_metadata() -> None: "ProcedureAborted", "ProcedureTruncated", "ProcedureHeld", + "ProcedureHoldClaimReleased", "ProcedureResumed", "ProcedureActivitiesLogbookOpened", "ProcedureIterationStarted", @@ -247,6 +249,99 @@ async def test_procedure_held_updates_status_and_reason() -> None: assert conn.execute.call_args.args[3] == "beam dropped" +@pytest.mark.unit +async def test_procedure_held_appends_the_holding_cause() -> None: + """`status` says a conduct is paused, never by whom, and the watcher's + window now turns on which concern it is.""" + proj = ProcedureSummaryProjection() + conn = AsyncMock() + event = _stored( + "ProcedureHeld", + { + "procedure_id": str(_PROCEDURE_ID), + "reason": "check did not settle", + "cause": "step-fault", + "occurred_at": _NOW.isoformat(), + }, + ) + await proj.apply(event, conn) + sql = conn.execute.call_args.args[0] + # Deduplicated rather than a bare array_append: a re-delivered hold must + # not stack a second copy, mirroring the aggregate's idempotent + # `_with_claim`. + assert "array_append(hold_causes" in sql + assert "= ANY(hold_causes)" in sql + assert conn.execute.call_args.args[4] == "step-fault" + + +@pytest.mark.unit +async def test_procedure_held_without_a_cause_records_the_legacy_one() -> None: + """A pre-claim hold carries no cause and the aggregate folds it to + LEGACY_CAUSE. Writing nothing instead would leave a Held row claiming + that nothing holds it, which reads as a deliberate pause to the watcher.""" + proj = ProcedureSummaryProjection() + conn = AsyncMock() + event = _stored( + "ProcedureHeld", + { + "procedure_id": str(_PROCEDURE_ID), + "reason": "beam dropped", + "occurred_at": _NOW.isoformat(), + }, + ) + await proj.apply(event, conn) + assert conn.execute.call_args.args[4] == LEGACY_CAUSE + + +@pytest.mark.unit +async def test_hold_claim_released_drops_the_cause_and_leaves_the_status() -> None: + """One concern letting go does not decide whether the Procedure runs + again; that is the ProcedureResumed's business in the same append.""" + proj = ProcedureSummaryProjection() + conn = AsyncMock() + event = _stored( + "ProcedureHoldClaimReleased", + { + "procedure_id": str(_PROCEDURE_ID), + "claim_id": str(uuid4()), + "cause": "step-fault", + "occurred_at": _NOW.isoformat(), + }, + ) + await proj.apply(event, conn) + sql = conn.execute.call_args.args[0] + assert "array_remove(hold_causes" in sql + assert "SET status" not in sql + assert conn.execute.call_args.args[1] == _PROCEDURE_ID + assert conn.execute.call_args.args[2] == "step-fault" + + +@pytest.mark.unit +async def test_resume_and_every_terminal_clear_the_holding_causes() -> None: + """A Running or finished Procedure holds nothing, so no arm may leave a + cause behind for the watcher to read as a live hold.""" + proj = ProcedureSummaryProjection() + for event_type, extra in ( + ("ProcedureResumed", {"re_establishment_boundary": 0}), + ("ProcedureCompleted", {}), + ("ProcedureAborted", {"reason": "vacuum loss"}), + ("ProcedureTruncated", {"reason": "power loss", "interrupted_at": _NOW.isoformat()}), + ): + conn = AsyncMock() + await proj.apply( + _stored( + event_type, + { + "procedure_id": str(_PROCEDURE_ID), + "occurred_at": _NOW.isoformat(), + **extra, + }, + ), + conn, + ) + assert "hold_causes = '{}'" in conn.execute.call_args.args[0], event_type + + @pytest.mark.unit async def test_procedure_resumed_updates_status_to_running_and_clears_reason() -> None: proj = ProcedureSummaryProjection() @@ -316,8 +411,10 @@ async def test_procedure_registered_seeds_iteration_count_to_zero() -> None: await proj.apply(event, conn) sql = conn.execute.call_args.args[0] assert "iteration_count" in sql - # iteration_count is seeded with the literal 0 (no positional arg). - assert ", 0)" in sql.replace("\n", " ").replace(" ", " ") + # iteration_count and hold_causes are both seeded with literals (no + # positional arg): a fresh Procedure has run no iterations and is Defined, + # so nothing holds it. + assert ", 0, '{}')" in sql.replace("\n", " ").replace(" ", " ") @pytest.mark.unit diff --git a/infra/atlas/migrations/20260911193339_procedure_summary_add_hold_causes.sql b/infra/atlas/migrations/20260911193339_procedure_summary_add_hold_causes.sql new file mode 100644 index 00000000000..2326efdfb4a --- /dev/null +++ b/infra/atlas/migrations/20260911193339_procedure_summary_add_hold_causes.sql @@ -0,0 +1,34 @@ +-- Procedure summary projection: additive hold_causes column. +-- +-- `status = 'Held'` is one bit and cannot say WHICH concern is holding, so +-- every hold was clocked against one week-long staleness window on the +-- assumption that a hold is a deliberate operator pause. A conduct the +-- Conductor parked on a step fault is the opposite: nothing will move it +-- without a person, and a week of silence is exactly wrong for it. +-- +-- The Procedure aggregate now records its hold claims, so this denorms the +-- CAUSES onto the read model and the ProcedureWatcher selects its window per +-- row instead of per status. Causes rather than a precomputed +-- "needs attention" flag: the classification lives in +-- `ATTENTION_HOLD_CAUSES`, and baking it into the projection would leave old +-- rows silently wrong the day that set changes. +-- +-- Ordered oldest-first, mirroring `Procedure.hold_claims`. Maintained by the +-- Held (append, deduplicated) / HoldClaimReleased (remove) / Resumed and the +-- three terminal arms (clear) of ProcedureSummaryProjection. +-- +-- Backfill: a row Held right now was held before causes were recorded, which +-- is what the aggregate folds to `LEGACY_CAUSE`, so the denorm says the same +-- rather than claiming (with an empty array) that nothing holds it. A +-- projection rebuild replaces these with the real causes where the stream has +-- them. +-- +-- Mutable read model. cora_app keeps its existing DML grants on +-- proj_operation_procedure_summary. + +ALTER TABLE proj_operation_procedure_summary + ADD COLUMN hold_causes TEXT[] NOT NULL DEFAULT '{}'; + +UPDATE proj_operation_procedure_summary +SET hold_causes = ARRAY['legacy-unscoped'] +WHERE status = 'Held'; diff --git a/infra/atlas/migrations/atlas.sum b/infra/atlas/migrations/atlas.sum index 87731a6618a..3078daa0c4f 100644 --- a/infra/atlas/migrations/atlas.sum +++ b/infra/atlas/migrations/atlas.sum @@ -1,4 +1,4 @@ -h1:m9WL1j5CeVOYtyYi9kxsQ8HLMh1mLF6x/t2NdHDAJIg= +h1:ee+R17aFDQ0Wh/A9p10Scg6W48nOPjigE0bK1fubYl0= 20260509120000_init_events.sql h1:GmgCZKfaqXu1m96/cKAks2vhaLWTdEaHTLkFtUo9FXg= 20260509170000_init_idempotency.sql h1:Nbu8DIE4Sv1WiHw3G22+tYffPhKc5Jryw3PMK8wB2zY= 20260510010000_add_event_id.sql h1:RbtYP6uMnOB20zhJ9dNXUi4YVqbmlEzf562pmygnRW8= @@ -177,3 +177,4 @@ h1:m9WL1j5CeVOYtyYi9kxsQ8HLMh1mLF6x/t2NdHDAJIg= 20260831150000_seed_in_process_surface.sql h1:QVyvB18DHwE6gIagfeu5DuC2P8uqqzz8dD65uSd32AU= 20260904120000_add_proj_operation_procedure_iterations_advice_latency.sql h1:MR1m7EoZ1Bf51f+A6qwf7M/WCpv2SusKVs8schf79Ko= 20260910222120_init_proj_data_shortfall_summary.sql h1:+3EbIqdQY5J76TtmUGW+AgbK+qA1lsajN5GEeb6HT8s= +20260911193339_procedure_summary_add_hold_causes.sql h1:uX4s6sKZomEnEVh3LsRWvTAZAaotKvWMdp0IwsaFR+A=