From 42ea2615efc8e0ddd185b5a9def6f0dc8ce91153 Mon Sep 17 00:00:00 2001 From: arpan Date: Fri, 4 Sep 2026 00:53:41 +0530 Subject: [PATCH 1/2] item 6d: Control grows a resume path, and the gateway holds elicitations MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SPEC-v0.2 §6.9 amended, §11 and ARCHITECTURE §6 with it. A held reservation spans two HTTP requests and therefore two Control calls, and Control.execute writes a terminal outcome for any executor exception — so the gateway could not express "suspend, record nothing" through it. The kernel grows one signal and one method rather than the gateway growing a second reservation path. Suspension is modelled the way v0.1 models "nothing happened": an explicit opt-in an executor raises, never a default and never inferred. Suspended is handled above the generic branch, alongside NotExecuted. The record stays EXECUTING, the lease is extended, the continuation is held in the same transaction, EXECUTION_SUSPENDED is appended, no receipt is written — the same rule as an action awaiting a human — and the signal propagates so the gateway can relay the elicitation. Suspended with no effect key holds nothing and the event says so. Suspension without an effect key gives no protection at all, exactly as retries without one give none; it is logged rather than silently pretended. Control.resume consumes the continuation atomically, rehydrates the action from the store — which is what lets a process that restarted mid-round still finish one — and runs the executor through the SAME outcome mapping execute uses. That path is now one private function called by both. A resumption that decided outcomes differently would be a second implementation of the only question this library exists to answer. resume re-evaluates the policy for the receipt, not to re-decide: the action was decided and reserved on the first leg, and refusing at resume would strand a reservation the remote may already be acting on. The gateway maps input_required to Suspended(requestState), a follow-up carrying requestState to Control.resume, and an elicitation with no requestState to §6.9.3's floor. Both of §6.9.2's bounds are enforced: the lease via suspend_timeout, the rounds via continuation_rounds. Control remains the only module that composes the others; the gateway calls it. ARCHITECTURE §6 says so explicitly now. --- docs/ARCHITECTURE.md | 8 +- docs/SPEC-v0.2.md | 41 +++- src/ctrlrun/__init__.py | 2 + src/ctrlrun/control.py | 112 +++++++++++ src/ctrlrun/errors.py | 29 +++ src/ctrlrun/gateway/server.py | 139 ++++++++++++-- src/ctrlrun/state.py | 223 ++++++++++++---------- tests/test_elicitation.py | 327 +++++++++++++++++++++++++++++++- tests/test_gateway.py | 190 +++++++++++-------- tests/test_resume.py | 343 ++++++++++++++++++++++++++++++++++ 10 files changed, 1209 insertions(+), 205 deletions(-) create mode 100644 tests/test_resume.py diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 4e660dc1..d61cd47b 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -139,12 +139,18 @@ Pragmas: `journal_mode=WAL`, `busy_timeout=5000`, `synchronous=NORMAL`. | `approval.py` | request/grant/consume, providers | executors | | `effect.py` | key templating, state enum, transition rules | SQLite | | `state.py` | `StateStore` protocol + SQLite/in-memory impls | policy, decorator, sinks | -| `control.py` | `Control` orchestration, decorator, context | CLI | +| `control.py` | `Control` orchestration, decorator, context, suspend/resume | CLI | | `receipt.py` | Receipt/Event models, `EventSink`, JSONL sink | everything else | | `cli/` | click commands, demo | internals beyond `Control` | Dependencies point downward only. `Control` is the only module that composes the others. +The gateway (v0.2) does not change this. It builds an Action and calls `Control` — including +`Control.resume` for an elicitation's second leg — rather than reserving and committing for +itself. A gateway that owned the reservation would be a second module composing the others, +and a second implementation of SPEC-v0.1 §5.5's asymmetry, which is the one rule in this +codebase that must not drift. + One exception, added in v0.2 and worth stating rather than discovering: `policy.py` imports the template grammar (`template_placeholders`) from `effect.py`, because SPEC-v0.2 §3.1 requires an `effect:` / `resource:` template to be validated when the policy loads and the grammar is diff --git a/docs/SPEC-v0.2.md b/docs/SPEC-v0.2.md index e3eabe09..2779769a 100644 --- a/docs/SPEC-v0.2.md +++ b/docs/SPEC-v0.2.md @@ -77,6 +77,9 @@ anything it cannot parse, and it never lets a transport error look like a defini ### 2.1 The hook ```python +Suspended(continuation: bytes | str) # raised by an executor; §6.9 +Control.resume(continuation: str, executor) -> Receipt + ReconcileOutcome = Literal["committed", "not_executed", "unknown"] protect(..., reconcile: Callable[[str], ReconcileOutcome] | None = None, @@ -785,6 +788,37 @@ that exchange. #### 6.9.2 Held reservation, where the upstream made it possible +**How this reaches the kernel.** A held reservation spans two HTTP requests and therefore two +`Control` calls, and `Control.execute` writes a terminal outcome for any executor exception +(v0.1 §5.5) — so the gateway cannot express "suspend, record nothing" through it. The kernel +grows one signal and one method, and suspension is modelled the way v0.1 models *nothing +happened*: an explicit opt-in an executor raises, never a default and never inferred. + +- `Suspended(continuation)`, raised by an executor, is handled in `Control.execute` **before** + the generic exception branch, alongside `NotExecuted`. The record stays `EXECUTING`, the + lease is extended by `Control(suspend_timeout=...)`, the continuation is held via + `hold_continuation` **in the same transaction**, `EXECUTION_SUSPENDED` is appended, **no + receipt is written** — the same rule as an action awaiting a human (v0.1 §6.1) — and the + signal propagates so the caller can relay the elicitation. +- `Suspended` on an action with **no effect key** holds nothing: there is no reservation to + hold. `EXECUTION_SUSPENDED` records `held: false` and the signal still propagates. + Suspension without an effect key gives no protection at all — a concurrent caller can start + the same work — exactly as retries without one give none (v0.1 §5.1). It is logged, not + silently pretended. +- `Control.resume(continuation, executor)` consumes the continuation atomically, rehydrates + the action and effect key from the store, checks the record is still `EXECUTING` under a + live lease, appends `EXECUTION_RESUMED`, and runs the executor through **the same** outcome + mapping `execute` uses. That path is one private function called by both: a resumption that + decided outcomes differently would be a second implementation of the only question this + library exists to answer. A resumed executor may raise `Suspended` again; each round + re-holds. +- `resume` re-evaluates the policy for the **receipt**, not to re-decide. The action was + decided and reserved on the first leg, and refusing at resume would strand a reservation + the remote may already be acting on. + +`Control` remains the only module that composes the others (ARCHITECTURE §6); the gateway +calls it. + When an intercepted `tools/call` with an effect key returns `input_required` **with a `requestState`**: @@ -1286,7 +1320,7 @@ claim (asserted against a word list including "compliant", "compliance", "certif from .effect import ReconcileOutcome from .receipt import EventSink, JSONLEventSink from .approval import WebhookApprovalProvider -from .errors import MissingDependency +from .errors import MissingDependency, Suspended # lazily importable, not re-exported at package import: # ctrlrun.otel.OTelEventSink (ctrlrun[otel]) # ctrlrun.gateway.serve (ctrlrun[gateway]) @@ -1314,8 +1348,9 @@ StateStore.find_granted_approval(action_hash: str) -> Approval | None StateStore.find_denied_request(action_hash: str) -> ApprovalRequest | None StateStore.resolve_effect(effect_key, state, *, resolved_by: str) -> None StateStore.extend_lease(effect_key: str, action_id: str, until: datetime) -> None -StateStore.hold_continuation(effect_key, action_id, request_state: str) -> None -StateStore.take_continuation(effect_key, request_state: str) -> EffectRecord +StateStore.hold_continuation(action, effect_key, continuation: str, until: datetime) -> int +StateStore.take_continuation(continuation: str) -> HeldContinuation +StateStore.continuation_rounds(effect_key: str) -> int ``` **Removed:** `SQLiteStateStore.journal`, and the `EventLog` behind it — `JSONLEventSink` is diff --git a/src/ctrlrun/__init__.py b/src/ctrlrun/__init__.py index 0e671b9c..1f204959 100644 --- a/src/ctrlrun/__init__.py +++ b/src/ctrlrun/__init__.py @@ -26,6 +26,7 @@ MissingDependency, NotExecuted, PolicyError, + Suspended, ) from .policy import Decision, Policy from .receipt import Event, EventSink, JSONLEventSink, Receipt @@ -64,6 +65,7 @@ "SQLiteStateStore", "ScriptedApprovalProvider", "StateStore", + "Suspended", "action_hash", "canonicalize", "context", diff --git a/src/ctrlrun/control.py b/src/ctrlrun/control.py index d750afca..e935c74c 100644 --- a/src/ctrlrun/control.py +++ b/src/ctrlrun/control.py @@ -49,6 +49,7 @@ EffectKeyError, InvalidArgument, NotExecuted, + Suspended, ) from .policy import Decision, Evaluation, Policy, discover_policy_path from .receipt import ( @@ -70,6 +71,10 @@ DEFAULT_ENVIRONMENT: Final = "production" +#: SPEC-v0.2 §6.9.2 — how long a reservation is held open across one round trip. A client +#: that never returns lets this lapse, and the record becomes AMBIGUOUS by v0.1 §5.3 E3. +DEFAULT_SUSPEND_TIMEOUT: Final = timedelta(minutes=5) + #: Denial reason when a protected function is called outside `ctrlrun.context()`. NO_PRINCIPAL: Final = "no_principal" @@ -209,6 +214,7 @@ def __init__( approval_ttl: timedelta = DEFAULT_APPROVAL_TTL, lease: timedelta = DEFAULT_LEASE, sinks: Sequence[EventSink] = (), + suspend_timeout: timedelta = DEFAULT_SUSPEND_TIMEOUT, ) -> None: self._policy = policy self._store = store @@ -219,6 +225,7 @@ def __init__( self._approval_ttl = approval_ttl self._lease = _checked_lease(lease, "Control(lease=...)") self._sinks = tuple(sinks) + self._suspend_timeout = _checked_lease(suspend_timeout, "Control(suspend_timeout=...)") @classmethod def from_file(cls, path: str | os.PathLike[str] | None = None) -> Control: @@ -331,8 +338,74 @@ def execute( ) raise self._append(EventType.EXECUTION_STARTED, action, {}, effect_key, approval=approval) + return self._outcome( + action, evaluation, executor, effect_key, attempt, approval, started_at, reconciler + ) + + def resume(self, continuation: str, executor: Callable[[], Any]) -> Receipt: + """Finish an action a `Suspended` executor left open (SPEC-v0.2 §6.9). + + The continuation is taken and consumed atomically, so one suspension admits exactly + one resumption; the action and its effect key are rehydrated from the store, which is + what lets a process that restarted mid-round still finish one. The record must still + be `EXECUTING` under a live lease — a lease that lapsed while the client was + answering is `AMBIGUOUS` by the ordinary path of v0.1 §5.3 E3. + + The executor then runs through the **same** outcome mapping `execute` uses. There is + one implementation of v0.1 §5.5 in this codebase, and a resumed call gets that one: + a resumption that decided outcomes differently would be a second answer to the only + question this library exists to answer. + """ + held = self._store.take_continuation(continuation) + started_at = self._clock() + action = held.action + self._append( + EventType.EXECUTION_RESUMED, + action, + {"round": held.rounds}, + held.effect_key, + ) + # SPEC: §6.9.2 — the policy is evaluated again for the *receipt*, not to re-decide. + # The action was decided and reserved on the first leg; refusing here would strand a + # reservation the remote may already be acting on, which is the one thing a held + # reservation exists to avoid. + evaluation = self._policy.evaluate(action) + return self._outcome( + action, + evaluation, + executor, + held.effect_key, + held.record.attempt, + None, + started_at, + _Reconciler(None, False), + ) + + def _outcome( + self, + action: Action, + evaluation: Evaluation, + executor: Callable[[], Any], + effect_key: str | None, + attempt: int, + approval: Approval | None, + started_at: datetime, + reconciler: _Reconciler, + ) -> Receipt: + """Run the executor and record what happened (SPEC-v0.1 §5.5). + + The single implementation of the asymmetry: `NotExecuted` is the only outcome that + maps to `FAILED`, `Suspended` records no outcome at all, and everything else is + `AMBIGUOUS`. `execute` and `resume` both come through here, so the two cannot drift. + """ try: result = executor() + except Suspended as suspension: + # SPEC-v0.2 §6.9 — no outcome, no receipt: the remote has not said what happened + # and this attempt is not finished. Handled above the generic branch precisely so + # it cannot be mistaken for one. + self._suspend(action, effect_key, suspension, approval) + raise except NotExecuted as exc: if effect_key is not None: # SPEC §5.5 — the executor asserted the remote side did nothing, so this is @@ -431,6 +504,45 @@ def execute( attempt=attempt, ) + def _suspend( + self, + action: Action, + effect_key: str | None, + suspension: Suspended, + approval: Approval | None, + ) -> None: + """Hold the reservation open and record that it is held (SPEC-v0.2 §6.9.2). + + With no effect key there is nothing to hold, and the event says so. Suspension + without an effect key gives no protection at all — a second caller can start the same + work while the first is waiting — exactly as retries without one give none (v0.1 + §5.1). It is permitted, logged, and recorded as unheld rather than silently pretended. + """ + if effect_key is None: + _LOG.warning( + "%s: suspended with no effect key, so nothing is held; a concurrent caller " + "can start the same work. Declare effect= to make a suspension mean something", + action.name, + ) + self._append( + EventType.EXECUTION_SUSPENDED, + action, + {"held": False, "round": 1}, + None, + approval=approval, + ) + return + rounds = self._store.hold_continuation( + action, effect_key, suspension.continuation, self._clock() + self._suspend_timeout + ) + self._append( + EventType.EXECUTION_SUSPENDED, + action, + {"held": True, "round": rounds}, + effect_key, + approval=approval, + ) + # --- the effect key (SPEC-v0.1 §5.1) ------------------------------------------------ def _resolve_effect(self, action: Action, template: str | None) -> str | None: diff --git a/src/ctrlrun/errors.py b/src/ctrlrun/errors.py index 7f8d95b7..b34938d6 100644 --- a/src/ctrlrun/errors.py +++ b/src/ctrlrun/errors.py @@ -114,6 +114,35 @@ class NotExecuted(CTRLRunError): """ +class Suspended(CTRLRunError): + """Raised by an executor: the remote asked for something before it will finish. + + SPEC-v0.2 §6.9 in the kernel's own terms, and modelled the way v0.1 §5.5 models "nothing + happened" — an explicit opt-in signal, never a default and never inferred. There is no + outcome to record: the effect record stays `EXECUTING`, its lease is extended, the + continuation is held, and the caller gets this back to relay. + + `continuation` is whatever the remote said to present again. It is opaque here — CTRLRun + never parses it, and only ever compares it with `hmac.compare_digest`. + """ + + def __init__(self, continuation: bytes | str) -> None: + if isinstance(continuation, bytes): + try: + continuation = continuation.decode("utf-8") + except UnicodeDecodeError as exc: + raise InvalidArgument( + "Suspended(continuation=...) must be UTF-8 decodable" + ) from exc + if not continuation: + raise InvalidArgument( + "Suspended(continuation=...) must be non-empty; a continuation nobody can " + "present again is a suspension nobody can end" + ) + super().__init__(f"suspended awaiting a continuation ({len(continuation)} bytes)") + self.continuation = continuation + + class MissingDependency(CTRLRunError): """An optional extra is not installed (SPEC-v0.2 §1.1, §11). diff --git a/src/ctrlrun/gateway/server.py b/src/ctrlrun/gateway/server.py index 5c88c4a9..bbac1886 100644 --- a/src/ctrlrun/gateway/server.py +++ b/src/ctrlrun/gateway/server.py @@ -34,6 +34,7 @@ EffectKeyError, InvalidArgument, NotExecuted, + Suspended, ) from ..receipt import Receipt from .mcp import DEFAULT_MAX_BODY_BYTES, ParsedRequest, Refusal, parse_request @@ -93,6 +94,27 @@ #: §6.6 — the decision reason recorded when an argument cannot be represented. UNREPRESENTABLE_REASON: Final = "unrepresentable_argument" +#: §6.9.3 — an upstream that elicits without a `requestState` cannot be fronted for a +#: consequential tool without a `ctrlrun resolve` per call. Stated as a cost, not a feature: +#: the protocol minted no identity for that exchange, and the only thing distinguishing a +#: continuation from a fresh duplicate call is a field the agent controls completely. +_STATELESS_ELICITATION: Final = GatewayOutcome( + EffectState.AMBIGUOUS, code=-41010, token="ctrlrun.upstream_ambiguous", http_status=502 +) + + +def _request_state(payload: bytes | None) -> str | None: + """The `requestState` an `input_required` result carried, if it carried one (§6.9.1).""" + if payload is None: + return None + try: + document = json.loads(payload) + except (ValueError, UnicodeDecodeError): + return None + result = document.get("result") if isinstance(document, dict) else None + state = result.get("requestState") if isinstance(result, Mapping) else None + return state if isinstance(state, str) and state else None + class UpstreamAmbiguous(CTRLRunError): """What the executor raises where §6.8 says the outcome is unknown. @@ -360,21 +382,6 @@ def _record_unrepresentable( error=f"{pointer} is a float, which an Action cannot represent", ) - # SPEC: v0.2 §6.9 — held reservations across an elicitation are NOT implemented, and - # the reason is structural rather than an omission. §6.9.2 requires the effect record to - # stay `EXECUTING` with no outcome written while the client answers, which spans two HTTP - # requests and therefore two `Control.execute` calls. `Control.execute` writes a terminal - # outcome for any executor exception (v0.1 §5.5), and there is no way for an executor to - # say "suspend, record nothing" — so an `input_required` currently reaches §6.8's - # unrecognized-resultType row and is recorded `AMBIGUOUS`, which is §6.9.3's fail-closed - # floor applied to every elicitation rather than only to the ones with no `requestState`. - # - # Closing it needs one of two decisions, and both are wider than §6.9's own text: either - # `Control` grows a resume path, or the gateway owns reserve/begin/commit for intercepted - # calls — which would make it the second module that composes the others, against - # ARCHITECTURE §6. The StateStore half (`hold_continuation`, `take_continuation`) is - # built and tested; this is the composition question above it. - def _execute( self, action: Action, @@ -395,6 +402,8 @@ def _execute( options = self._control.policy.mcp_options(action.name) held: dict[str, Any] = {} + presented = parsed.document.get("params", {}) + presented = presented.get("requestState") if isinstance(presented, Mapping) else None def executor() -> Any: # §6.7 — the request the gateway sends is built from the action's *canonical* @@ -413,14 +422,65 @@ def executor() -> Any: held["status"] = status held["headers"] = response_headers held["outcome"] = outcome + if outcome.elicitation: + # §6.9 — neither leg is an outcome. Where the upstream gave a `requestState` + # the reservation is held across the round trip; where it did not, the + # protocol minted no identity for the exchange and §6.9.3's floor applies. + state = _request_state(payload) + if state is None: + raise UpstreamAmbiguous(_STATELESS_ELICITATION) + raise Suspended(state) if outcome.effect is EffectState.COMMITTED: return payload if outcome.effect is EffectState.FAILED: raise NotExecuted(str(outcome.token or observed)) raise UpstreamAmbiguous(outcome) + if isinstance(presented, str) and presented: + return self._continue(action, executor, held, presented, request_id) return self._through_control(action, executor, effect_key, held, request_id) + def _continue( + self, + action: Action, + executor: Callable[[], Any], + held: dict[str, Any], + presented: str, + request_id: JsonRpcId, + ) -> _Response: + """A continuation: the same action, the same reservation (SPEC-v0.2 §6.9.2). + + `Control.resume` consumes the continuation atomically and rehydrates the action from + the store, so a gateway that restarted mid-round can still finish one. The hash check + is this layer's: a continuation must be the same action, and `resume` cannot know what + the client just sent. + """ + code, token, status = UNKNOWN_CONTINUATION + try: + receipt = self._control.resume(presented, executor) + except Suspended: + key = self._control.policy.effect_template(action.name) + resolved = None if key is None else resolve_effect_key(key, action) + over = self._over_the_round_bound(action, resolved, request_id) + return over or self._upstream_response(action, held, resolved, suspended=True) + except InvalidArgument as refused: + _LOG.warning("refused a continuation for %s: %s", action.name, refused) + return _json( + status, + json_rpc_error(request_id, code, token, str(refused), tool=action.name), + ) + except AmbiguousEffect as refused: + code, token, status = AMBIGUOUS_EFFECT + return _json( + status, + json_rpc_error( + request_id, code, token, str(refused), effect_key=refused.effect_key + ), + ) + except (NotExecuted, UpstreamAmbiguous): + return self._upstream_response(action, held, None) + return self._upstream_response(action, held, receipt.effect_key, receipt) + def _through_control( self, action: Action, @@ -486,6 +546,11 @@ def _through_control( status, json_rpc_error(request_id, code, token, str(refused), reason=refused.reason), ) + except Suspended: + # §6.9.2 step 5 — the InputRequiredResult is relayed unchanged, plus the receipt + # meta. No outcome was written and no receipt exists; the reservation is held. + over = self._over_the_round_bound(action, effect_key, request_id) + return over or self._upstream_response(action, held, effect_key, suspended=True) except (NotExecuted, UpstreamAmbiguous): return self._upstream_response(action, held, effect_key) return self._upstream_response(action, held, effect_key, receipt) @@ -551,6 +616,40 @@ def _before_asking_a_human( ) return None + def _over_the_round_bound( + self, action: Action, effect_key: str | None, request_id: JsonRpcId + ) -> _Response | None: + """§6.9.2's first bound. A held reservation is a resource an upstream must not be able + to pin forever, and the lease bound only stops a client that stops answering. + + Exceeding it records the effect `AMBIGUOUS` and returns `-41010`: the gateway has sent + that many tool calls and knows the outcome of none of them. + """ + if effect_key is None: + return None + rounds = self._control.store.continuation_rounds(effect_key) + if rounds <= self._config.max_elicitation_rounds: + return None + record = self._control.store.get_effect(effect_key) + if record is not None: + self._control.store.mark_ambiguous( + effect_key, record.action_id, f"elicitation exceeded {rounds - 1} rounds" + ) + _LOG.warning("%s: elicitation exceeded the round bound at %s", action.name, rounds) + code, token, status = UPSTREAM_AMBIGUOUS + return _json( + status, + json_rpc_error( + request_id, + code, + token, + f"the upstream elicited more than {self._config.max_elicitation_rounds} times " + "and the outcome of none of those calls is known", + effect_key=effect_key, + action_id=action.action_id, + ), + ) + def _awaiting(self, action: Action, pending: ApprovalRequired, request_id: Any) -> _Response: """§6.10 — "no" is an answer, and re-asking is not free.""" record = self._control.store.get_approval(pending.request_id) @@ -566,6 +665,8 @@ def _upstream_response( held: dict[str, Any], effect_key: str | None, receipt: Receipt | None = None, + *, + suspended: bool = False, ) -> _Response: """Relay the upstream's own answer, or synthesize one where none arrived (§6.8).""" outcome = held.get("outcome") @@ -573,7 +674,13 @@ def _upstream_response( "action_id": action.action_id, "receipt_id": receipt.receipt_id if receipt is not None else None, "effect_key": effect_key, - "result": str(outcome.effect) if outcome and outcome.effect else "ambiguous", + "result": ( + "suspended" + if suspended + else str(outcome.effect) + if outcome and outcome.effect + else "ambiguous" + ), "attempt": receipt.attempt if receipt is not None else 1, } if outcome is not None and outcome.relay and held.get("payload") is not None: diff --git a/src/ctrlrun/state.py b/src/ctrlrun/state.py index 54b2d083..034b3a1b 100644 --- a/src/ctrlrun/state.py +++ b/src/ctrlrun/state.py @@ -22,7 +22,7 @@ import sqlite3 import threading from collections.abc import Callable, Iterable -from dataclasses import replace +from dataclasses import dataclass, replace from datetime import UTC, datetime, timedelta from pathlib import Path from typing import Any, Final, Protocol, TypeVar @@ -81,11 +81,13 @@ CREATE TABLE IF NOT EXISTS continuations( effect_key TEXT PRIMARY KEY, action_id TEXT NOT NULL, - action_hash TEXT NOT NULL, - request_state TEXT NOT NULL, + action_json TEXT NOT NULL, + continuation TEXT NOT NULL, rounds INTEGER NOT NULL DEFAULT 0, updated_at TEXT ); +CREATE UNIQUE INDEX IF NOT EXISTS continuations_by_value + ON continuations(continuation) WHERE continuation <> ''; CREATE TABLE IF NOT EXISTS approvals( approval_id TEXT PRIMARY KEY, action_hash TEXT NOT NULL, @@ -296,6 +298,16 @@ def _reserved( ) +@dataclass(frozen=True) +class HeldContinuation: + """What `take_continuation` hands back: the same action, still reserved (§6.9.2).""" + + action: Action + effect_key: str + record: EffectRecord + rounds: int + + class StateStore(ApprovalStore, Protocol): """Durable state behind a `Control` (SPEC-v0.1 §5.3): approvals, effects, evidence.""" @@ -378,31 +390,36 @@ def find_denied_request(self, action_hash: str) -> ApprovalRequest | None: ... def hold_continuation( - self, effect_key: str, action_id: str, request_state: str, *, action_hash: str + self, action: Action, effect_key: str, continuation: str, until: datetime ) -> int: - """Hold this reservation open across an elicitation round trip (SPEC-v0.2 §6.9.2). + """Hold this reservation open across a round trip (SPEC-v0.2 §6.9.2). - Refused unless the record is `EXECUTING`, held by this `action_id`, with a live - lease — the same conditions `extend_lease` requires, for the same reason. Returns the - round number, which is what `--max-elicitation-rounds` is counted against. + One transaction: the lease is extended to `until`, the continuation is stored, and + the round counter is incremented. Refused unless the record is `EXECUTING`, held by + this action, with a live lease — the conditions `extend_lease` requires, for the same + reason. Returns the round number. - `action_hash` travels with it because the continuation is *the same action*, and the - gateway has to be able to prove that before admitting one. + The whole `Action` travels with it because a resumption is *the same action*, and + rehydrating it from the store is the only way a gateway that restarted mid-round can + still finish one. """ ... - def take_continuation(self, effect_key: str, request_state: str) -> EffectRecord: + def take_continuation(self, continuation: str) -> HeldContinuation: """Consume a held continuation, or refuse (SPEC-v0.2 §6.9.2). - The `request_state` is compared with `hmac.compare_digest` and consumed in the same - transaction that admits the continuation, so one elicitation admits exactly one. Two - continuations racing on one held key is the duplicate execution this product exists - to prevent, wearing an `inputResponses` field as a disguise. + Compared with `hmac.compare_digest` and consumed in the same transaction that admits + it, so one suspension admits exactly one resumption. """ ... - def held_action_hash(self, effect_key: str) -> str | None: - """The `action_hash` of the held continuation for this key, or `None`.""" + def continuation_rounds(self, effect_key: str) -> int: + """How many times this effect has been suspended, or 0 (SPEC-v0.2 §6.9.2). + + A read, so a caller can enforce its own round bound: a held reservation is a resource + an upstream must not be able to pin forever, and the lease bound alone only stops a + client that stops answering, not an upstream that elicits forever. + """ ... def approvals_for(self, action_hash: str) -> tuple[ApprovalRecord, ...]: @@ -452,7 +469,7 @@ def __init__(self, *, clock: Callable[[], datetime] = _utc_now) -> None: self._receipts: list[Receipt] = [] self._approvals: dict[str, ApprovalRecord] = {} self._effects: dict[str, EffectRecord] = {} - self._continuations: dict[str, tuple[str, str, int]] = {} + self._continuations: dict[str, tuple[str, Action, int]] = {} def close(self) -> None: """Nothing to release; here so either store can be closed the same way.""" @@ -665,32 +682,38 @@ def extend_lease(self, effect_key: str, action_id: str, until: datetime) -> None self._effects[effect_key] = extended def hold_continuation( - self, effect_key: str, action_id: str, request_state: str, *, action_hash: str + self, action: Action, effect_key: str, continuation: str, until: datetime ) -> int: with self._lock: - record = _holdable(self._effects.get(effect_key), effect_key, action_id, self._clock()) - _, _, rounds = self._continuations.get(effect_key, ("", "", 0)) - self._continuations[effect_key] = (request_state, action_hash, rounds + 1) - assert record is not None + extended = plan_lease_extension( + self._effects.get(effect_key), effect_key, action.action_id, until, self._clock() + ) + self._reject_duplicate_continuation(effect_key, continuation) + _, _, rounds = self._continuations.get(effect_key, ("", action, 0)) + self._effects[effect_key] = extended + self._continuations[effect_key] = (continuation, action, rounds + 1) return rounds + 1 - def take_continuation(self, effect_key: str, request_state: str) -> EffectRecord: + def take_continuation(self, continuation: str) -> HeldContinuation: with self._lock: - record = _continuable( - self._effects.get(effect_key), - self._continuations.get(effect_key), - effect_key, - request_state, - self._clock(), - ) - held = self._continuations[effect_key] - self._continuations[effect_key] = ("", held[1], held[2]) - return record - - def held_action_hash(self, effect_key: str) -> str | None: + for effect_key, (held, action, rounds) in self._continuations.items(): + if held and hmac.compare_digest(held, continuation): + record = _continuable(self._effects.get(effect_key), effect_key, self._clock()) + self._continuations[effect_key] = ("", action, rounds) + return HeldContinuation(action, effect_key, record, rounds) + raise InvalidArgument(_NO_SUCH_CONTINUATION) + + def continuation_rounds(self, effect_key: str) -> int: with self._lock: held = self._continuations.get(effect_key) - return None if held is None else held[1] + return 0 if held is None else held[2] + + def _reject_duplicate_continuation(self, effect_key: str, continuation: str) -> None: + for other, (held, _, _) in self._continuations.items(): + if other != effect_key and held and hmac.compare_digest(held, continuation): + raise InvalidArgument( + f"continuation is already held for {other!r}; two effects cannot share one" + ) def get_effect(self, effect_key: str) -> EffectRecord | None: with self._lock: @@ -868,74 +891,94 @@ def get_approval(self, approval_id: str) -> ApprovalRecord | None: return self._read_approval(self._connection(), approval_id) def hold_continuation( - self, effect_key: str, action_id: str, request_state: str, *, action_hash: str + self, action: Action, effect_key: str, continuation: str, until: datetime ) -> int: connection = self._connection() connection.execute("BEGIN IMMEDIATE") try: - _holdable( - self._read_effect(connection, effect_key), effect_key, action_id, self._clock() + extended = plan_lease_extension( + self._read_effect(connection, effect_key), + effect_key, + action.action_id, + until, + self._clock(), ) row = connection.execute( "SELECT rounds FROM continuations WHERE effect_key=?", (effect_key,) ).fetchone() rounds = (row["rounds"] if row else 0) + 1 + self._write_effect(connection, extended) connection.execute( - "INSERT INTO continuations(effect_key, action_id, action_hash, request_state, " + "INSERT INTO continuations(effect_key, action_id, action_json, continuation, " "rounds, updated_at) VALUES(?,?,?,?,?,?) ON CONFLICT(effect_key) DO UPDATE SET " - "action_id=excluded.action_id, action_hash=excluded.action_hash, " - "request_state=excluded.request_state, rounds=excluded.rounds, " + "action_id=excluded.action_id, action_json=excluded.action_json, " + "continuation=excluded.continuation, rounds=excluded.rounds, " "updated_at=excluded.updated_at", ( effect_key, - action_id, - action_hash, - request_state, + action.action_id, + _action_json(action), + continuation, rounds, _iso(self._clock()), ), ) + except sqlite3.IntegrityError as exc: + self._unwind(connection) + raise InvalidArgument( + "continuation is already held for another effect; two effects cannot share one" + ) from exc except BaseException: self._unwind(connection) raise connection.commit() return rounds - def take_continuation(self, effect_key: str, request_state: str) -> EffectRecord: + def take_continuation(self, continuation: str) -> HeldContinuation: connection = self._connection() connection.execute("BEGIN IMMEDIATE") try: - row = connection.execute( - "SELECT request_state, action_hash, rounds FROM continuations WHERE effect_key=?", - (effect_key,), - ).fetchone() - held = ( - None if row is None else (row["request_state"], row["action_hash"], row["rounds"]) + rows = connection.execute( + "SELECT effect_key, action_json, continuation, rounds FROM continuations " + "WHERE continuation <> ''" + ).fetchall() + found = next( + ( + row + for row in rows + if hmac.compare_digest(str(row["continuation"]), continuation) + ), + None, ) + if found is None: + raise InvalidArgument(_NO_SUCH_CONTINUATION) + effect_key = str(found["effect_key"]) record = _continuable( - self._read_effect(connection, effect_key), - held, - effect_key, - request_state, - self._clock(), + self._read_effect(connection, effect_key), effect_key, self._clock() ) # Consumed in the same transaction that admits it (§6.9.2). connection.execute( - "UPDATE continuations SET request_state='' WHERE effect_key=?", (effect_key,) + "UPDATE continuations SET continuation='' WHERE effect_key=?", (effect_key,) + ) + held = HeldContinuation( + _action_from_json(str(found["action_json"])), + effect_key, + record, + int(found["rounds"]), ) except BaseException: self._unwind(connection) raise connection.commit() - return record + return held - def held_action_hash(self, effect_key: str) -> str | None: + def continuation_rounds(self, effect_key: str) -> int: row = ( self._connection() - .execute("SELECT action_hash FROM continuations WHERE effect_key=?", (effect_key,)) + .execute("SELECT rounds FROM continuations WHERE effect_key=?", (effect_key,)) .fetchone() ) - return None if row is None else str(row["action_hash"]) + return 0 if row is None else int(row["rounds"]) def find_granted_approval(self, action_hash: str) -> Approval | None: return _newest_granted(self.approvals_for(action_hash), action_hash, self._clock()) @@ -1392,51 +1435,31 @@ def _newest_denied( return newest.request -def _holdable( - record: EffectRecord | None, effect_key: str, action_id: str, now: datetime -) -> EffectRecord: - """Whether this reservation may be held open across a round trip (SPEC-v0.2 §6.9.2). +def _continuable(record: EffectRecord | None, effect_key: str, now: datetime) -> EffectRecord: + """Whether a resumption may be admitted (SPEC-v0.2 §6.9.2). - The same conditions `extend_lease` requires, and for the same reason: a record that is - not this attempt's, live and executing is not this attempt's to suspend either. + The record checks are the ones a lease extension makes, and for the same reason: a record + that has been resolved, or whose lease lapsed while the client was answering, is not this + attempt's to finish. An expired one is `AMBIGUOUS` by the ordinary path of v0.1 §5.3 E3 — + the executor may have died mid-flight, and that is exactly what has happened. """ - if record is None or record.state is not EffectState.EXECUTING or not record.lease_is_live(now): + if record is None or record.state is not EffectState.EXECUTING: raise AmbiguousEffect( - f"effect {effect_key!r} is not executing under a live lease and cannot be held", + f"effect {effect_key!r} is no longer executing and cannot be resumed", effect_key=effect_key, action_id=None if record is None else record.action_id, ) - if record.action_id != action_id: - raise DuplicateEffect( - f"effect {effect_key!r} is held by {record.action_id}, not {action_id}", - state=IN_PROGRESS_EFFECT, + if not record.lease_is_live(now): + raise AmbiguousEffect( + f"effect {effect_key!r} was left executing by {record.action_id} and its lease " + "expired while the continuation was outstanding", effect_key=effect_key, + action_id=record.action_id, ) return record -def _continuable( - record: EffectRecord | None, - held: tuple[str, str, int] | None, - effect_key: str, - request_state: str, - now: datetime, -) -> EffectRecord: - """Whether a continuation may be admitted (SPEC-v0.2 §6.9.2). - - The record checks come first, so an expired or resolved reservation is refused as what it - is rather than as an unknown continuation. `hmac.compare_digest` because the presented - value is agent-supplied and comparing it byte by byte leaks its prefix. - """ - if record is None or record.state is not EffectState.EXECUTING or not record.lease_is_live(now): - raise AmbiguousEffect( - f"effect {effect_key!r} is no longer executing under a live lease", - effect_key=effect_key, - action_id=None if record is None else record.action_id, - ) - stored = "" if held is None else held[0] - if not stored or not hmac.compare_digest(stored, request_state): - raise InvalidArgument( - f"no held elicitation on {effect_key!r} matches the presented requestState" - ) - return record +#: What a continuation nobody holds is refused with. Deliberately identical whether the value +#: was forged, already consumed, or belongs to a suspension that has since moved on: telling +#: an agent which of those it hit is telling it how to search. +_NO_SUCH_CONTINUATION: Final = "no held suspension matches the presented continuation" diff --git a/tests/test_elicitation.py b/tests/test_elicitation.py index 95a49710..6011f35f 100644 --- a/tests/test_elicitation.py +++ b/tests/test_elicitation.py @@ -1,8 +1,12 @@ -"""Resumption is not relayed for an intercepted call. Build-list item 6d; SPEC-v0.2 §6.2. +"""Held reservations across an elicitation. Build-list item 6d; SPEC-v0.2 §6.9, §6.2. -Acceptance test T30b. §6.9's held reservations are the other half of this item and are not -here: see the PR, and the `# SPEC:` note in `server.py`. The store layer they need is in -`tests/test_gateway.py`. +Acceptance tests T26 and T30b. The kernel half — `Suspended` and `Control.resume` — is in +`tests/test_resume.py`; this is the MCP half on top of it. + +Both naive readings are wrong. Treating the first leg as a completed action records an +outcome the upstream never reported; refusing the continuation makes every eliciting tool +unusable. The gateway holds the reservation open instead: the effect stays `EXECUTING`, +concurrent duplicates stay blocked throughout, and only the final result is mapped by §6.8. A legacy stream may be resumed by a `GET` carrying `Last-Event-ID`, which would let the final response to an intercepted `tools/call` arrive on a connection the gateway is not attributing @@ -12,12 +16,327 @@ from __future__ import annotations +import json +import threading +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +import pytest + +from ctrlrun import Control, EffectState, Policy, SQLiteStateStore from ctrlrun.gateway.legacy import ( LAST_EVENT_ID_HEADER, SESSION_HEADER, is_event_stream, strip_event_ids, ) +from ctrlrun.gateway.server import Gateway, GatewayConfig, build_server, httpx_forwarder +from ctrlrun.receipt import EventType + +httpx = pytest.importorskip("httpx", reason="the gateway extra is not installed") + +POLICY = """ +schema: ctrlrun.policy/v2 +actions: + mcp.acme.create_refund: + effect: "refund:{payment_id}" + resource: "payment:{payment_id}" + decision: allow + mcp.acme.read_only: + decision: allow +""" + +CURRENT = "2026-07-28" +STATE = "opaque-state-from-the-server" + + +def _call(tool="create_refund", *, arguments=None, request_id=1, extra=None): + params = { + "name": tool, + "arguments": {"payment_id": "txn_1", "amount": 200} if arguments is None else arguments, + } + params.update(extra or {}) + return {"jsonrpc": "2.0", "id": request_id, "method": "tools/call", "params": params} + + +def _headers(body=None): + call = body or _call() + return { + "MCP-Protocol-Version": CURRENT, + "Mcp-Method": "tools/call", + "Mcp-Name": call["params"]["name"], + "X-Agent": "refund-agent", + } + + +class Upstream: + def __init__(self) -> None: + self.calls: list[dict] = [] + self.replies: list[dict] = [] + + def next_reply(self) -> dict: + return self.replies.pop(0) if self.replies else {"resultType": "complete", "content": []} + + def elicits(self, *, request_state=STATE, rounds=1): + for index in range(rounds): + result = {"resultType": "input_required", "inputRequests": {"q1": "elicitation"}} + if request_state is not None: + result["requestState"] = f"{request_state}-{index}" if rounds > 1 else request_state + self.replies.append(result) + + +@pytest.fixture +def upstream(): + state = Upstream() + + class Handler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def do_POST(self): + length = int(self.headers.get("Content-Length") or 0) + state.calls.append(json.loads(self.rfile.read(length))) + payload = json.dumps({"jsonrpc": "2.0", "id": 1, "result": state.next_reply()}).encode() + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(payload))) + self.end_headers() + self.wfile.write(payload) + + def log_message(self, *args): + pass + + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + threading.Thread(target=server.serve_forever, daemon=True).start() + state.url = f"http://127.0.0.1:{server.server_address[1]}/mcp" + yield state + server.shutdown() + server.server_close() + + +@pytest.fixture +def store(tmp_path): + store = SQLiteStateStore(tmp_path / "state.db") + yield store + store.close() + + +def _gateway(upstream, store, **overrides): + config = GatewayConfig( + upstream=upstream.url, alias="acme", principal_header="X-Agent", port=0, **overrides + ) + forwarder = httpx_forwarder(config) + return Gateway(config, Control(Policy.from_yaml(POLICY), store), forwarder), forwarder + + +@pytest.fixture +def client(upstream, store): + gateway, forwarder = _gateway(upstream, store) + server = build_server(gateway) + threading.Thread(target=server.serve_forever, daemon=True).start() + with httpx.Client( + base_url=f"http://127.0.0.1:{server.server_address[1]}", timeout=10 + ) as opened: + yield opened + server.shutdown() + server.server_close() + forwarder.close() + + +def _post(client, body=None): + body = body or _call() + return client.post("/mcp", content=json.dumps(body).encode(), headers=_headers(body)) + + +def _continuation(request_state=STATE, request_id=2, arguments=None): + return _call( + request_id=request_id, + arguments=arguments, + extra={"requestState": request_state, "inputResponses": {"q1": "yes"}}, + ) + + +def _error(response): + return response.json()["error"] + + +# --- T26: with a requestState ---------------------------------------------------------- + + +def test_T26_an_input_required_result_leaves_the_effect_executing(client, upstream, store): + upstream.elicits() + + response = _post(client) + + assert store.get_effect("refund:txn_1").state is EffectState.EXECUTING + assert response.json()["result"]["resultType"] == "input_required" + assert store.receipts() == () + + +def test_T26_the_response_meta_says_suspended(client, upstream, store): + upstream.elicits() + + meta = _post(client).json()["_meta"]["com.ctrlrun/receipt"] + + assert meta["result"] == "suspended" + assert meta["effect_key"] == "refund:txn_1" + + +def test_T26_execution_suspended_is_appended_with_the_round(client, upstream, store): + upstream.elicits() + + _post(client) + + suspended = [e for e in store.events() if e.type is EventType.EXECUTION_SUSPENDED] + assert len(suspended) == 1 + assert suspended[0].data == {"held": True, "round": 1} + + +def test_T26_a_concurrent_duplicate_is_refused_while_the_elicitation_is_outstanding( + client, upstream, store +): + """The guarantee that actually matters: the reservation is still held.""" + upstream.elicits() + _post(client) + + duplicate = _post(client, _call(request_id=99)) + + assert duplicate.status_code == 409 + assert _error(duplicate)["code"] == -41004 + assert _error(duplicate)["data"]["state"] == "in_progress" + + +def test_T26_the_continuation_is_admitted_and_produces_one_receipt(client, upstream, store): + upstream.elicits() + _post(client) + + final = _post(client, _continuation()) + + assert final.status_code == 200 + assert final.json()["result"]["resultType"] == "complete" + assert store.get_effect("refund:txn_1").state is EffectState.COMMITTED + assert len(store.receipts()) == 1 + assert store.receipts()[0].attempt == 1 + + +def test_T26_the_continuation_is_the_same_action(client, upstream, store): + """§6.9.2 — same action_id, same action_hash, same reservation, same attempt.""" + upstream.elicits() + _post(client) + proposed = store.events()[0] + + _post(client, _continuation()) + + assert store.receipts()[0].action_id == proposed.action_id + assert store.receipts()[0].action_hash == proposed.data["action_hash"] + + +def test_T26_suspended_precedes_resumed(client, upstream, store): + upstream.elicits() + _post(client) + _post(client, _continuation()) + + types = [e.type for e in store.events()] + assert types.index(EventType.EXECUTION_SUSPENDED) < types.index(EventType.EXECUTION_RESUMED) + + +# --- T26: the refusals, each leaving the record EXECUTING ------------------------------ + + +def test_T26_a_continuation_with_the_wrong_request_state_is_refused_with_41009( + client, upstream, store +): + upstream.elicits() + _post(client) + before = len(upstream.calls) + + refused = _post(client, _continuation(request_state="guessed")) + + assert refused.status_code == 400 + assert _error(refused)["code"] == -41009 + assert store.get_effect("refund:txn_1").state is EffectState.EXECUTING + assert len(upstream.calls) == before + + +def test_T26_a_second_continuation_on_a_consumed_state_is_refused(client, upstream, store): + """One elicitation admits one continuation.""" + upstream.elicits() + _post(client) + _post(client, _continuation()) + before = len(upstream.calls) + + replayed = _post(client, _continuation(request_id=3)) + + assert _error(replayed)["code"] == -41009 + assert len(upstream.calls) == before + + +def test_T26_a_continuation_matching_no_held_elicitation_is_refused_with_41009(client, upstream): + """A forgery, or a gateway that restarted before the record was written.""" + refused = _post(client, _continuation()) + + assert _error(refused)["code"] == -41009 + assert upstream.calls == [] + + +# --- T26: the bounds ------------------------------------------------------------------- + + +def test_T26_exceeding_the_round_bound_records_ambiguous_and_returns_41010(upstream, store): + """A held reservation is a resource an upstream must not be able to pin forever: the + gateway has sent that many tool calls and knows the outcome of none.""" + gateway, forwarder = _gateway(upstream, store, max_elicitation_rounds=1) + server = build_server(gateway) + threading.Thread(target=server.serve_forever, daemon=True).start() + upstream.elicits(rounds=4) + try: + with httpx.Client( + base_url=f"http://127.0.0.1:{server.server_address[1]}", timeout=10 + ) as client: + _post(client) + last = _post(client, _continuation(request_state=f"{STATE}-0")) + finally: + server.shutdown() + server.server_close() + forwarder.close() + + assert _error(last)["code"] == -41010 + assert store.get_effect("refund:txn_1").state is EffectState.AMBIGUOUS + + +# --- T26: without a requestState (§6.9.3) ---------------------------------------------- + + +def test_T26_an_elicitation_without_a_request_state_records_ambiguous(client, upstream, store): + """The fail-closed floor, stated as a cost. The protocol minted no identity for that + exchange, so the only thing distinguishing a continuation from a fresh duplicate call is + a field the agent controls completely.""" + upstream.elicits(request_state=None) + + _post(client) + + assert store.get_effect("refund:txn_1").state is EffectState.AMBIGUOUS + + +def test_T26_a_continuation_after_a_stateless_elicitation_is_refused_with_41005( + client, upstream, store +): + upstream.elicits(request_state=None) + _post(client) + + refused = _post(client, _call(request_id=2)) + + assert _error(refused)["code"] == -41005 + + +def test_T26_a_tool_with_no_effect_key_is_unaffected(client, upstream, store): + """No reservation to hold, so an eliciting *read* costs a log line either way.""" + upstream.elicits() + body = _call(tool="read_only") + + response = client.post("/mcp", content=json.dumps(body).encode(), headers=_headers(body)) + + assert response.json()["result"]["resultType"] == "input_required" + assert store.list_effects() == () + # --- T30b: resumption is not relayed for an intercepted call --------------------------- diff --git a/tests/test_gateway.py b/tests/test_gateway.py index 08743182..8bf62d28 100644 --- a/tests/test_gateway.py +++ b/tests/test_gateway.py @@ -239,7 +239,7 @@ def _request(store, request_id: str, action: Action, clock, ttl=DEFAULT_APPROVAL return request -def _action(amount: int = 2000) -> Action: +def _approvable(amount: int = 2000) -> Action: return Action( name="mcp.acme.refund", arguments={"payment_id": "txn_1", "amount": amount}, @@ -248,7 +248,7 @@ def _action(amount: int = 2000) -> Action: def test_find_granted_approval_returns_the_newest_granted_one(state_store, fake_clock): - action = _action() + action = _approvable() _request(state_store, "apr_1", action, fake_clock) state_store.grant_approval("apr_1", "human:ops") fake_clock.advance(timedelta(minutes=1)) @@ -263,7 +263,7 @@ def test_find_granted_approval_returns_the_newest_granted_one(state_store, fake_ def test_find_granted_approval_breaks_a_tie_toward_the_later_record(state_store, fake_clock): """Two grants in one clock tick tie on `granted_at`; the later-stored one is newer.""" - action = _action() + action = _approvable() _request(state_store, "apr_1", action, fake_clock) _request(state_store, "apr_2", action, fake_clock) state_store.grant_approval("apr_1", "human:ops") @@ -273,14 +273,14 @@ def test_find_granted_approval_breaks_a_tie_toward_the_later_record(state_store, def test_find_granted_approval_ignores_a_pending_one(state_store, fake_clock): - action = _action() + action = _approvable() _request(state_store, "apr_1", action, fake_clock) assert state_store.find_granted_approval(action.action_hash) is None def test_find_granted_approval_ignores_a_consumed_one(state_store, fake_clock): - action = _action() + action = _approvable() _request(state_store, "apr_1", action, fake_clock) state_store.grant_approval("apr_1", "human:ops") state_store.consume_approval("apr_1", action.action_hash) @@ -289,7 +289,7 @@ def test_find_granted_approval_ignores_a_consumed_one(state_store, fake_clock): def test_find_granted_approval_ignores_an_expired_one(state_store, fake_clock): - action = _action() + action = _approvable() _request(state_store, "apr_1", action, fake_clock, ttl=timedelta(minutes=5)) state_store.grant_approval("apr_1", "human:ops") fake_clock.advance(timedelta(minutes=6)) @@ -298,16 +298,16 @@ def test_find_granted_approval_ignores_an_expired_one(state_store, fake_clock): def test_find_granted_approval_ignores_another_actions_hash(state_store, fake_clock): - action = _action() + action = _approvable() _request(state_store, "apr_1", action, fake_clock) state_store.grant_approval("apr_1", "human:ops") - assert state_store.find_granted_approval(_action(amount=9999).action_hash) is None + assert state_store.find_granted_approval(_approvable(amount=9999).action_hash) is None def test_find_denied_request_returns_an_unexpired_denial(state_store, fake_clock): """§6.10 — "no" is an answer, and re-asking is not free.""" - action = _action() + action = _approvable() _request(state_store, "apr_1", action, fake_clock) state_store.deny_approval("apr_1", "human:ops") @@ -320,7 +320,7 @@ def test_find_denied_request_returns_an_unexpired_denial(state_store, fake_clock def test_find_denied_request_ignores_an_expired_denial(state_store, fake_clock): """A denial holds for the life of the request that carried it; after that, asking again is legitimate (§6.10).""" - action = _action() + action = _approvable() _request(state_store, "apr_1", action, fake_clock, ttl=timedelta(minutes=5)) state_store.deny_approval("apr_1", "human:ops") fake_clock.advance(timedelta(minutes=6)) @@ -329,7 +329,7 @@ def test_find_denied_request_ignores_an_expired_denial(state_store, fake_clock): def test_find_denied_request_ignores_a_granted_one(state_store, fake_clock): - action = _action() + action = _approvable() _request(state_store, "apr_1", action, fake_clock) state_store.grant_approval("apr_1", "human:ops") @@ -337,7 +337,7 @@ def test_find_denied_request_ignores_a_granted_one(state_store, fake_clock): def test_the_two_lookups_agree_with_the_record_the_store_holds(state_store, fake_clock): - action = _action() + action = _approvable() _request(state_store, "apr_1", action, fake_clock) state_store.deny_approval("apr_1", "human:ops") @@ -346,124 +346,152 @@ def test_the_two_lookups_agree_with_the_record_the_store_holds(state_store, fake assert state_store.find_denied_request(action.action_hash).request_id == "apr_1" -# --- §6.9.2: the held continuation ------------------------------------------------------ +# --- §6.9.2: the held continuation, at the store layer --------------------------------- -HASH = "sha256:" + "a" * 64 -OTHER_HASH = "sha256:" + "b" * 64 -STATE = "opaque-server-state-1" +CONTINUATION = "opaque-server-state-1" + + +def _action() -> Action: + return Action( + name="mcp.acme.refund", + arguments={"payment_id": "txn_1", "amount": 2000}, + principal=Principal(agent="refund-agent"), + ) + + +def _held(store, fake_clock, action=None): + action = action or _action() + store.reserve_effect(KEY, action.action_id, LEASE) + store.begin_execution(KEY, action.action_id) + return action, store.hold_continuation( + action, KEY, CONTINUATION, fake_clock.now + timedelta(minutes=30) + ) def test_a_held_continuation_is_taken_exactly_once(sqlite_store, fake_clock): - """§6.9.2 — one elicitation admits one continuation. + """§6.9.2 — one suspension admits one resumption. Two racing on one held key is the + duplicate execution this product exists to prevent, wearing a continuation as a + disguise.""" + action, rounds = _held(sqlite_store, fake_clock) - The spec puts single-use on the party that needs it, and the party holding a reservation - needs it: two continuations racing on one held key is the duplicate execution this - product exists to prevent, wearing an `inputResponses` field as a disguise. - """ - _executing(sqlite_store) - sqlite_store.hold_continuation(KEY, MINE, STATE, action_hash=HASH) + held = sqlite_store.take_continuation(CONTINUATION) - held = sqlite_store.take_continuation(KEY, STATE) + assert rounds == 1 + assert held.action.action_id == action.action_id + assert held.effect_key == KEY + with pytest.raises(InvalidArgument): + sqlite_store.take_continuation(CONTINUATION) - assert held.action_id == MINE - assert held.state is EffectState.EXECUTING - with pytest.raises(CTRLRunError): - sqlite_store.take_continuation(KEY, STATE) +def test_holding_a_continuation_extends_the_lease_in_the_same_transaction(sqlite_store, fake_clock): + """§6.9.2 steps 2 and 3 are one write: a lease extended without the continuation stored + would hold a reservation nobody can ever continue.""" + _held(sqlite_store, fake_clock) -def test_a_continuation_with_the_wrong_request_state_is_refused(sqlite_store): - _executing(sqlite_store) - sqlite_store.hold_continuation(KEY, MINE, STATE, action_hash=HASH) + assert sqlite_store.get_effect(KEY).lease_expires_at == fake_clock.now + timedelta(minutes=30) - with pytest.raises(CTRLRunError): - sqlite_store.take_continuation(KEY, "guessed") - # Refused without consuming: the real continuation can still arrive. - assert sqlite_store.take_continuation(KEY, STATE).action_id == MINE +def test_the_whole_action_comes_back_with_the_continuation(sqlite_store, fake_clock): + """A resumption is the *same action*, and rehydrating it from the store is the only way a + process that restarted mid-round can still finish one.""" + action, _ = _held(sqlite_store, fake_clock) + held = sqlite_store.take_continuation(CONTINUATION) -def test_a_continuation_for_a_key_that_holds_none_is_refused(sqlite_store): - _executing(sqlite_store) + assert held.action.action_hash == action.action_hash + assert held.action.name == "mcp.acme.refund" + assert held.action.canonical_arguments == action.canonical_arguments - with pytest.raises(CTRLRunError): - sqlite_store.take_continuation(KEY, STATE) + +def test_a_forged_continuation_is_refused(sqlite_store, fake_clock): + _held(sqlite_store, fake_clock) + + with pytest.raises(InvalidArgument): + sqlite_store.take_continuation("guessed") + + # Refused without consuming: the real continuation can still arrive. + assert sqlite_store.take_continuation(CONTINUATION).effect_key == KEY def test_a_continuation_is_refused_once_the_lease_has_expired(sqlite_store, fake_clock): - """§6.9.2's second bound: a client that never returns lets the lease lapse, and the - record becomes AMBIGUOUS by the ordinary path of v0.1 §5.3 E3.""" - _executing(sqlite_store) - sqlite_store.hold_continuation(KEY, MINE, STATE, action_hash=HASH) - fake_clock.advance(LEASE + timedelta(seconds=1)) + _held(sqlite_store, fake_clock) + fake_clock.advance(timedelta(minutes=31)) with pytest.raises(AmbiguousEffect): - sqlite_store.take_continuation(KEY, STATE) + sqlite_store.take_continuation(CONTINUATION) -def test_a_continuation_is_refused_once_the_record_is_ambiguous(sqlite_store): - _executing(sqlite_store) - sqlite_store.hold_continuation(KEY, MINE, STATE, action_hash=HASH) - sqlite_store.mark_ambiguous(KEY, MINE, "lost") +def test_a_continuation_is_refused_once_the_record_is_resolved(sqlite_store, fake_clock): + action, _ = _held(sqlite_store, fake_clock) + sqlite_store.mark_ambiguous(KEY, action.action_id, "lost") with pytest.raises(AmbiguousEffect): - sqlite_store.take_continuation(KEY, STATE) + sqlite_store.take_continuation(CONTINUATION) -def test_holding_a_continuation_needs_a_live_executing_record_this_action_holds(sqlite_store): - sqlite_store.reserve_effect(KEY, MINE, LEASE) +def test_holding_needs_a_live_executing_record_this_action_holds(sqlite_store, fake_clock): + action = _action() + sqlite_store.reserve_effect(KEY, action.action_id, LEASE) + until = fake_clock.now + timedelta(minutes=30) with pytest.raises(CTRLRunError): - sqlite_store.hold_continuation(KEY, MINE, STATE, action_hash=HASH) + sqlite_store.hold_continuation(action, KEY, CONTINUATION, until) - sqlite_store.begin_execution(KEY, MINE) + sqlite_store.begin_execution(KEY, action.action_id) with pytest.raises(DuplicateEffect): - sqlite_store.hold_continuation(KEY, THEIRS, STATE, action_hash=HASH) + sqlite_store.hold_continuation(_action(), KEY, CONTINUATION, until) -def test_each_hold_increments_the_round_counter(sqlite_store): - """§6.9.2 — the bound an upstream must not be able to pin a reservation past.""" - _executing(sqlite_store) +def test_each_hold_increments_the_round_counter(sqlite_store, fake_clock): + """The bound an upstream must not be able to pin a reservation past (§6.9.2).""" + action, first = _held(sqlite_store, fake_clock) + sqlite_store.take_continuation(CONTINUATION) + until = fake_clock.now + timedelta(minutes=30) - assert sqlite_store.hold_continuation(KEY, MINE, STATE, action_hash=HASH) == 1 - sqlite_store.take_continuation(KEY, STATE) - assert sqlite_store.hold_continuation(KEY, MINE, "state-2", action_hash=HASH) == 2 - sqlite_store.take_continuation(KEY, "state-2") - assert sqlite_store.hold_continuation(KEY, MINE, "state-3", action_hash=HASH) == 3 + second = sqlite_store.hold_continuation(action, KEY, "state-2", until) + sqlite_store.take_continuation("state-2") + third = sqlite_store.hold_continuation(action, KEY, "state-3", until) + assert (first, second, third) == (1, 2, 3) -def test_the_held_action_hash_comes_back_with_the_record(sqlite_store): - """The continuation is the *same action*, so the gateway has to be able to prove it.""" - _executing(sqlite_store) - sqlite_store.hold_continuation(KEY, MINE, STATE, action_hash=HASH) - assert sqlite_store.held_action_hash(KEY) == HASH +def test_two_effects_cannot_share_one_continuation(sqlite_store, fake_clock): + """Fail loudly rather than resolve to somebody else's action: a continuation is the only + thing identifying which suspension a resumption ends.""" + _held(sqlite_store, fake_clock) + other = _action() + sqlite_store.reserve_effect("refund:txn_2", other.action_id, LEASE) + sqlite_store.begin_execution("refund:txn_2", other.action_id) + + with pytest.raises(InvalidArgument): + sqlite_store.hold_continuation( + other, "refund:txn_2", CONTINUATION, fake_clock.now + timedelta(minutes=30) + ) def test_the_held_state_survives_a_new_store_on_the_same_file(tmp_path, fake_clock): - """§6.9.2 — the held state lives in the StateStore, not in gateway memory: a gateway - that restarted mid-elicitation would otherwise strand a reservation nobody can - continue, which is a self-inflicted AMBIGUOUS on every deploy.""" + """§6.9.2 — the held state lives in the StateStore, not in gateway memory: a gateway that + restarted mid-elicitation would otherwise strand a reservation nobody can continue, which + is a self-inflicted AMBIGUOUS on every deploy.""" path = tmp_path / "state.db" first = SQLiteStateStore(path, clock=fake_clock) try: - first.reserve_effect(KEY, MINE, LEASE) - first.begin_execution(KEY, MINE) - first.hold_continuation(KEY, MINE, STATE, action_hash=HASH) + action, _ = _held(first, fake_clock) finally: first.close() second = SQLiteStateStore(path, clock=fake_clock) try: - assert second.take_continuation(KEY, STATE).action_id == MINE + assert second.take_continuation(CONTINUATION).action.action_id == action.action_id finally: second.close() def test_the_in_memory_store_holds_continuations_the_same_way(state_store, fake_clock): - state_store.reserve_effect(KEY, MINE, LEASE) - state_store.begin_execution(KEY, MINE) + """A double that admits what SQLite refuses invalidates every test that uses it.""" + action, rounds = _held(state_store, fake_clock) - assert state_store.hold_continuation(KEY, MINE, STATE, action_hash=HASH) == 1 - assert state_store.take_continuation(KEY, STATE).action_id == MINE - with pytest.raises(CTRLRunError): - state_store.take_continuation(KEY, STATE) + assert rounds == 1 + assert state_store.take_continuation(CONTINUATION).action.action_id == action.action_id + with pytest.raises(InvalidArgument): + state_store.take_continuation(CONTINUATION) diff --git a/tests/test_resume.py b/tests/test_resume.py new file mode 100644 index 00000000..cd5220c1 --- /dev/null +++ b/tests/test_resume.py @@ -0,0 +1,343 @@ +"""Suspension and resumption. Build-list item 6d; SPEC-v0.2 §6.9 via the kernel. + +Suspension is modelled the way v0.1 models "nothing happened": an explicit opt-in signal an +executor raises, never a default and never inferred. `Suspended` says the remote asked for +something before it will finish — so there is no outcome to record, the reservation stays +exactly where it is, and the caller gets the signal back to relay. + +`Control.resume` is the other half, and it runs the executor through the *same* outcome +mapping `execute` uses. There is one implementation of v0.1 §5.5 in this codebase and this +item does not add a second. +""" + +from __future__ import annotations + +import contextlib +from datetime import timedelta + +import pytest + +from ctrlrun import ( + AmbiguousEffect, + Control, + DuplicateEffect, + EffectState, + InvalidArgument, + NotExecuted, + Policy, + Suspended, + context, + protect, +) +from ctrlrun.receipt import EventType, ReceiptResult + +POLICY = """ +schema: ctrlrun.policy/v1 +actions: + stripe.refund: + decision: allow + customer.read: + decision: allow +""" + +CONTINUATION = "opaque-state-from-the-server" + + +@pytest.fixture +def control(state_store, fake_clock): + return Control(Policy.from_yaml(POLICY), state_store, clock=fake_clock) + + +def _suspending(control, continuation=CONTINUATION, effect="refund:{payment_id}"): + @protect("stripe.refund", effect=effect, control=control) + def refund(payment_id: str, amount: int) -> str: + raise Suspended(continuation) + + return refund + + +def _types(store): + return [event.type for event in store.events()] + + +def _suspend(control, payment_id="txn_1"): + refund = _suspending(control) + with context(agent="refund-agent"), pytest.raises(Suspended): + refund(payment_id=payment_id, amount=200) + + +# --- suspension holds the record, and records nothing else ----------------------------- + + +def test_a_suspension_leaves_the_record_executing(control, state_store): + _suspend(control) + + record = state_store.get_effect("refund:txn_1") + assert record.state is EffectState.EXECUTING + + +def test_a_suspension_writes_no_receipt(control, state_store): + """The same rule as an action awaiting a human: it has not reached a terminal state.""" + _suspend(control) + + assert state_store.receipts() == () + + +def test_a_suspension_appends_execution_suspended(control, state_store): + _suspend(control) + + suspended = [e for e in state_store.events() if e.type is EventType.EXECUTION_SUSPENDED] + assert len(suspended) == 1 + assert suspended[0].effect_key == "refund:txn_1" + assert suspended[0].data["round"] == 1 + assert suspended[0].data["held"] is True + + +def test_the_lease_is_extended_by_the_suspend_timeout(state_store, fake_clock): + control = Control( + Policy.from_yaml(POLICY), + state_store, + clock=fake_clock, + lease=timedelta(minutes=1), + suspend_timeout=timedelta(minutes=30), + ) + _suspend(control) + + record = state_store.get_effect("refund:txn_1") + assert record.lease_expires_at == fake_clock.now + timedelta(minutes=30) + + +def test_the_signal_propagates_to_the_caller(control): + """The gateway needs it back to relay the elicitation to the client.""" + refund = _suspending(control) + + with context(agent="refund-agent"), pytest.raises(Suspended) as raised: + refund(payment_id="txn_1", amount=200) + + assert raised.value.continuation == CONTINUATION + + +def test_a_concurrent_duplicate_is_refused_while_suspended(control, state_store): + """The guarantee that actually matters across a round trip: the key is still held.""" + _suspend(control) + other = _suspending(control, continuation="another") + + with context(agent="refund-agent-b"), pytest.raises(DuplicateEffect) as refused: + other(payment_id="txn_1", amount=200) + + assert refused.value.state == "in_progress" + + +# --- resumption runs the same outcome mapping ------------------------------------------ + + +def test_resume_commits_and_writes_one_receipt(control, state_store): + _suspend(control) + + receipt = control.resume(CONTINUATION, lambda: "re_1") + + assert receipt.result is ReceiptResult.COMMITTED + assert state_store.get_effect("refund:txn_1").state is EffectState.COMMITTED + assert len(state_store.receipts()) == 1 + assert receipt.attempt == 1 + + +def test_resume_keeps_the_original_action_id_and_hash(control, state_store): + """§6.9.2 — the continuation is the same action: same id, same hash, one receipt.""" + _suspend(control) + proposed = state_store.events()[0] + + receipt = control.resume(CONTINUATION, lambda: "re_1") + + assert receipt.action_id == proposed.action_id + assert receipt.action_hash == proposed.data["action_hash"] + + +def test_resume_appends_execution_resumed_after_suspended(control, state_store): + _suspend(control) + control.resume(CONTINUATION, lambda: "re_1") + + types = _types(state_store) + assert types.index(EventType.EXECUTION_SUSPENDED) < types.index(EventType.EXECUTION_RESUMED) + + +@pytest.mark.parametrize( + ("executor", "state", "result"), + [ + (lambda: "re_1", EffectState.COMMITTED, ReceiptResult.COMMITTED), + ( + lambda: (_ for _ in ()).throw(NotExecuted("rejected before any side effect")), + EffectState.FAILED, + ReceiptResult.FAILED, + ), + ( + lambda: (_ for _ in ()).throw(TimeoutError("response lost")), + EffectState.AMBIGUOUS, + ReceiptResult.AMBIGUOUS, + ), + ], +) +def test_resume_maps_outcomes_exactly_as_execute_does( + control, state_store, executor, state, result +): + """The same cases `execute` is held to, run through `resume` (v0.1 §5.5). + + Parametrized against the same three outcomes deliberately: if these two paths ever + disagree, the asymmetry this library exists for has two implementations and one of them + is wrong. + """ + _suspend(control) + + with contextlib.suppress(NotExecuted, TimeoutError): + control.resume(CONTINUATION, executor) + + assert state_store.get_effect("refund:txn_1").state is state + assert state_store.receipts()[-1].result is result + + +def test_a_failed_resume_permits_a_retry(control, state_store): + """`NotExecuted` is still the only outcome that leaves the key retryable (v0.1 §5.4).""" + _suspend(control) + with pytest.raises(NotExecuted): + control.resume(CONTINUATION, lambda: (_ for _ in ()).throw(NotExecuted("nothing ran"))) + + refund = _suspending(control, continuation="round-2") + with context(agent="refund-agent"), pytest.raises(Suspended): + refund(payment_id="txn_1", amount=200) + + assert state_store.get_effect("refund:txn_1").attempt == 2 + + +# --- what resumption refuses ----------------------------------------------------------- + + +def test_a_continuation_cannot_be_consumed_twice(control): + """One suspension admits one resumption. Two racing on one held key is the duplicate + execution this product exists to prevent, wearing a continuation as a disguise.""" + _suspend(control) + control.resume(CONTINUATION, lambda: "re_1") + + with pytest.raises(InvalidArgument): + control.resume(CONTINUATION, lambda: "re_2") + + +def test_a_forged_continuation_is_refused(control, state_store): + _suspend(control) + + with pytest.raises(InvalidArgument): + control.resume("guessed", lambda: "re_1") + + assert state_store.get_effect("refund:txn_1").state is EffectState.EXECUTING + + +def test_a_continuation_nobody_holds_is_refused(control): + with pytest.raises(InvalidArgument): + control.resume(CONTINUATION, lambda: "re_1") + + +def test_resume_after_the_lease_expired_is_ambiguous(control, state_store, fake_clock): + """The only way an elicitation ends ambiguous: a client that never returns.""" + _suspend(control) + fake_clock.advance(timedelta(minutes=10)) + + with pytest.raises(AmbiguousEffect): + control.resume(CONTINUATION, lambda: "re_1") + + +def test_resume_after_another_attempt_ambiguated_the_record_is_refused(control, state_store): + _suspend(control) + record = state_store.get_effect("refund:txn_1") + state_store.mark_ambiguous("refund:txn_1", record.action_id, "lost") + + with pytest.raises(AmbiguousEffect): + control.resume(CONTINUATION, lambda: "re_1") + + +def test_the_executor_never_runs_when_the_continuation_is_refused(control): + calls: list[int] = [] + + with pytest.raises(InvalidArgument): + control.resume("guessed", lambda: calls.append(1)) + + assert calls == [] + + +# --- multi-round, and the keyless case ------------------------------------------------- + + +def test_a_resumed_executor_may_suspend_again(control, state_store): + _suspend(control) + + def elicits_again() -> str: + raise Suspended("round-2") + + with pytest.raises(Suspended): + control.resume(CONTINUATION, elicits_again) + + assert state_store.get_effect("refund:txn_1").state is EffectState.EXECUTING + rounds = [ + event.data["round"] + for event in state_store.events() + if event.type is EventType.EXECUTION_SUSPENDED + ] + assert rounds == [1, 2] + assert control.resume("round-2", lambda: "re_1").result is ReceiptResult.COMMITTED + + +def test_the_first_continuation_is_dead_once_the_second_is_held(control): + _suspend(control) + with pytest.raises(Suspended): + control.resume(CONTINUATION, lambda: (_ for _ in ()).throw(Suspended("round-2"))) + + with pytest.raises(InvalidArgument): + control.resume(CONTINUATION, lambda: "re_1") + + +def test_suspending_without_an_effect_key_holds_nothing_and_says_so(control, state_store): + """Documented, not silent: suspension without an effect key gives no protection, exactly + as retries without one give none (v0.1 §5.1).""" + + @protect("customer.read", control=control) + def read(customer_id: str) -> str: + raise Suspended(CONTINUATION) + + with context(agent="support-agent"), pytest.raises(Suspended): + read(customer_id="cus_1") + + suspended = [e for e in state_store.events() if e.type is EventType.EXECUTION_SUSPENDED] + assert suspended[0].data["held"] is False + assert state_store.list_effects() == () + with pytest.raises(InvalidArgument): + control.resume(CONTINUATION, lambda: "ok") + + +def test_a_keyless_suspension_writes_no_receipt_either(control, state_store): + @protect("customer.read", control=control) + def read(customer_id: str) -> str: + raise Suspended(CONTINUATION) + + with context(agent="support-agent"), pytest.raises(Suspended): + read(customer_id="cus_1") + + assert state_store.receipts() == () + + +# --- the signal itself ----------------------------------------------------------------- + + +def test_Suspended_accepts_bytes_and_str(): + assert Suspended(b"abc").continuation == "abc" + assert Suspended("abc").continuation == "abc" + + +def test_Suspended_refuses_an_empty_continuation(): + """A continuation nobody can present again is a suspension nobody can end.""" + for empty in ("", b""): + with pytest.raises(InvalidArgument): + Suspended(empty) + + +def test_Suspended_is_a_ctrlrun_error(): + from ctrlrun import CTRLRunError + + assert issubclass(Suspended, CTRLRunError) From 10c82dbce6996a570e15e88647437419340e3919 Mon Sep 17 00:00:00 2001 From: arpan Date: Fri, 4 Sep 2026 01:00:47 +0530 Subject: [PATCH 2/2] fix: two guards in the resume path that could only fire together MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Mutation-testing §6.9 found the same pattern this codebase keeps producing: a resolved record and a lapsed lease both refused a resumption with the same exception and wrote nothing, so removing either changed nothing a caller or a test could see. One branch now, naming the state it found; the tests distinguish the two by message. The harness had the ambiguous-anchor bug again — `except sqlite3.IntegrityError` appears in put_approval_request too, and replace(..., 1) hit that one. And compare_digest vs == is an equivalent mutant for correctness, so M21 now mutates the matching itself rather than the comparison primitive. --- src/ctrlrun/state.py | 16 +++++++++------- tests/test_gateway.py | 8 ++++++-- 2 files changed, 15 insertions(+), 9 deletions(-) diff --git a/src/ctrlrun/state.py b/src/ctrlrun/state.py index 034b3a1b..556d7c70 100644 --- a/src/ctrlrun/state.py +++ b/src/ctrlrun/state.py @@ -1443,16 +1443,18 @@ def _continuable(record: EffectRecord | None, effect_key: str, now: datetime) -> attempt's to finish. An expired one is `AMBIGUOUS` by the ordinary path of v0.1 §5.3 E3 — the executor may have died mid-flight, and that is exactly what has happened. """ - if record is None or record.state is not EffectState.EXECUTING: + if record is None: raise AmbiguousEffect( - f"effect {effect_key!r} is no longer executing and cannot be resumed", - effect_key=effect_key, - action_id=None if record is None else record.action_id, + f"effect {effect_key!r} has no record to resume", effect_key=effect_key ) - if not record.lease_is_live(now): + if not (record.state is EffectState.EXECUTING and record.lease_is_live(now)): + # One branch, not two: a resolved record and a lapsed lease both refuse with the same + # exception and write nothing, so separate guards for each would be one defence + # wearing two hats — indistinguishable to a caller and to a test. The message names + # the state instead. raise AmbiguousEffect( - f"effect {effect_key!r} was left executing by {record.action_id} and its lease " - "expired while the continuation was outstanding", + f"effect {effect_key!r} is {record.state} under {record.action_id} with no live " + "lease and cannot be resumed", effect_key=effect_key, action_id=record.action_id, ) diff --git a/tests/test_gateway.py b/tests/test_gateway.py index 8bf62d28..98b4e192 100644 --- a/tests/test_gateway.py +++ b/tests/test_gateway.py @@ -417,17 +417,21 @@ def test_a_continuation_is_refused_once_the_lease_has_expired(sqlite_store, fake _held(sqlite_store, fake_clock) fake_clock.advance(timedelta(minutes=31)) - with pytest.raises(AmbiguousEffect): + with pytest.raises(AmbiguousEffect) as refused: sqlite_store.take_continuation(CONTINUATION) + assert "executing" in str(refused.value) + def test_a_continuation_is_refused_once_the_record_is_resolved(sqlite_store, fake_clock): action, _ = _held(sqlite_store, fake_clock) sqlite_store.mark_ambiguous(KEY, action.action_id, "lost") - with pytest.raises(AmbiguousEffect): + with pytest.raises(AmbiguousEffect) as refused: sqlite_store.take_continuation(CONTINUATION) + assert "ambiguous" in str(refused.value) + def test_holding_needs_a_live_executing_record_this_action_holds(sqlite_store, fake_clock): action = _action()