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/record_export/_dispositions.py b/apps/api/src/cora/infrastructure/record_export/_dispositions.py index b40cb2b1787..df76b42248c 100644 --- a/apps/api/src/cora/infrastructure/record_export/_dispositions.py +++ b/apps/api/src/cora/infrastructure/record_export/_dispositions.py @@ -1432,11 +1432,20 @@ }, "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", + "decided_by_decision_id": "token:uuid", + "occurred_at": "keep:time", + "procedure_id": "token:uuid", + }, "ProcedureIterationEnded": { "advice_latency_ms": "keep:number", "advised_next_point": "drop:opaque", @@ -1486,6 +1495,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/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 f35d12069a3..851f4167240 100644 --- a/apps/api/src/cora/operation/aggregates/procedure/__init__.py +++ b/apps/api/src/cora/operation/aggregates/procedure/__init__.py @@ -25,12 +25,20 @@ PostgresOutcomeStore, ) from cora.operation.aggregates.procedure.events import ( + ATTENTION_HOLD_CAUSES, + 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, @@ -43,9 +51,11 @@ SteeringDesignRecorded, event_type_name, from_stored, + is_deliberate_pause, 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, @@ -87,6 +97,7 @@ ProcedureCannotTruncateError, ProcedureCapabilityExecutorMismatchError, ProcedureEnclosureCoverageMismatchError, + ProcedureHoldClaimsRemainError, ProcedureHoldReason, ProcedureIterationLimitReachedError, ProcedureName, @@ -113,7 +124,14 @@ ) __all__ = [ + "ATTENTION_HOLD_CAUSES", "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 +185,8 @@ "ProcedureEnclosureCoverageMismatchError", "ProcedureEvent", "ProcedureHeld", + "ProcedureHoldClaimReleased", + "ProcedureHoldClaimsRemainError", "ProcedureHoldReason", "ProcedureIterationEnded", "ProcedureIterationLimitReachedError", @@ -199,10 +219,12 @@ "ResolvedStepsRecorded", "SteeringDesignRecorded", "StepKind", + "derive_claim_id", "event_type_name", "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 1e350823112..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 @@ -441,6 +441,94 @@ 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`). 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 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( + { + 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.""" + +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. +""" + + +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.""" + +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 +570,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 +612,57 @@ class ProcedureResumed: re_establishment_boundary: int occurred_at: datetime decided_by_decision_id: UUID | None = None + released_claim_id: UUID | None = None + """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. + + 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 + 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)` + - 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 + 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 + 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) @@ -816,6 +966,7 @@ class SteeringDesignRecorded: | ProcedureTruncated | ProcedureHeld | ProcedureResumed + | ProcedureHoldClaimReleased | ProcedureActivitiesLogbookOpened | ProcedureDiagnosticLogbookOpened | ProcedureOutcomeLogbookOpened @@ -942,6 +1093,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 +1104,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 +1121,25 @@ 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, + 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, @@ -1338,6 +1513,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 +1531,22 @@ 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"]), + decided_by_decision_id=_optional_uuid(payload.get("decided_by_decision_id")), + ), + ) case "ProcedureActivitiesLogbookOpened": return deserialize_or_raise( "ProcedureActivitiesLogbookOpened", @@ -1490,12 +1682,20 @@ def _build_resumed() -> ProcedureResumed: __all__ = [ + "ATTENTION_HOLD_CAUSES", + "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", @@ -1508,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/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/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 1726477b6aa..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. @@ -1557,6 +1590,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/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/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/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/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..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,7 +14,26 @@ 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 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 -> InvalidProcedureReEstablishmentBoundaryError @@ -25,27 +44,72 @@ """ from datetime import datetime +from uuid import UUID from cora.operation.aggregates.procedure import ( + ATTENTION_HOLD_CAUSES, + 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 _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, *, 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 +126,63 @@ def decide( raise ProcedureCannotResumeError( state.id, current_status=state.status, parent_run_held=True ) - return [ - ProcedureResumed( + 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, + ) + ] + + def _released(claim_id: UUID, cause: str) -> ProcedureHoldClaimReleased: + return ProcedureHoldClaimReleased( procedure_id=state.id, - re_establishment_boundary=command.re_establishment_boundary, + 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 + # 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) + 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) + # 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/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/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/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/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/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_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/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/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/_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_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 1378ea10256..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,6 +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, + # 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_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_claim_enforcement.py b/apps/api/tests/unit/operation/test_procedure_hold_claim_enforcement.py new file mode 100644 index 00000000000..299801191fc --- /dev/null +++ b/apps/api/tests/unit/operation/test_procedure_hold_claim_enforcement.py @@ -0,0 +1,273 @@ +"""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 ( + ATTENTION_HOLD_CAUSES, + HOLD_CAUSE_DRIVER_STAND_DOWN, + HOLD_CAUSE_OPERATOR, + HOLD_CAUSE_STEP_FAULT, + HOLD_CAUSES, + LEGACY_CAUSE, + LEGACY_CLAIM_ID, + Procedure, + ProcedureCannotHoldError, + ProcedureHeld, + ProcedureHoldClaimReleased, + ProcedureHoldClaimsRemainError, + ProcedureName, + ProcedureResumed, + 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 +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_OPERATOR), + command=_resume(HOLD_CAUSE_STEP_FAULT), + now=_NOW, + ) + 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, + ) + 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_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 + 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: + """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_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_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/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 626ff4135e6..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,6 +144,9 @@ async def test_handler_appends_procedure_resumed_event() -> None: "re_establishment_boundary": 2, "decided_by_decision_id": None, "occurred_at": _NOW.isoformat(), + # 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)), } 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=