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..556d7c70 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,33 @@ 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: raise AmbiguousEffect( - f"effect {effect_key!r} is not executing under a live lease and cannot be held", - 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 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.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} is {record.state} under {record.action_id} with no live " + "lease and cannot be resumed", 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..98b4e192 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,156 @@ 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) + 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_ambiguous(sqlite_store): - _executing(sqlite_store) - sqlite_store.hold_continuation(KEY, MINE, STATE, action_hash=HASH) - sqlite_store.mark_ambiguous(KEY, MINE, "lost") - with pytest.raises(AmbiguousEffect): - sqlite_store.take_continuation(KEY, STATE) +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) as refused: + sqlite_store.take_continuation(CONTINUATION) + assert "ambiguous" in str(refused.value) -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)