From 1a663fcbdbc64c33c91ce4e7cecaec5ddc578bb5 Mon Sep 17 00:00:00 2001 From: arpan Date: Fri, 4 Sep 2026 00:37:37 +0530 Subject: [PATCH] item 6d (partial): the continuation store, and resumption stripping MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SPEC-v0.2 §6.9.2's StateStore half and §6.2's resumption rule (T30b). §6.9's gateway-side held reservations are NOT here, for a structural reason recorded as a `# SPEC:` note in server.py and set out in the PR. hold_continuation / take_continuation live in their own `continuations` table rather than as new columns on `effects`: §2.2 records that v0.2 has no migration story, and CREATE TABLE IF NOT EXISTS gives an existing database the new table for free where ALTER TABLE would need one. 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. The record checks come first, so an expired or resolved reservation is refused as what it is rather than as an unknown continuation. The held state survives a new store on the same file, which is the point: a gateway that restarted mid-elicitation would otherwise strand a reservation nobody can continue. legacy.py strips SSE id: fields from an intercepted response stream, line by line rather than event by event — the gateway must relay each event as it arrives, so it cannot wait for an event to end before deciding what to forward. Only id: is removed; event:, data:, retry: and comments pass through. EXECUTION_SUSPENDED and EXECUTION_RESUMED join the event set (§11), named for what happens to the attempt rather than for the transport that caused it. --- src/ctrlrun/gateway/legacy.py | 50 +++++++++ src/ctrlrun/gateway/server.py | 24 +++++ src/ctrlrun/receipt.py | 4 + src/ctrlrun/state.py | 186 ++++++++++++++++++++++++++++++++++ tests/test_elicitation.py | 69 +++++++++++++ tests/test_gateway.py | 123 ++++++++++++++++++++++ 6 files changed, 456 insertions(+) create mode 100644 src/ctrlrun/gateway/legacy.py create mode 100644 tests/test_elicitation.py diff --git a/src/ctrlrun/gateway/legacy.py b/src/ctrlrun/gateway/legacy.py new file mode 100644 index 00000000..f21ea670 --- /dev/null +++ b/src/ctrlrun/gateway/legacy.py @@ -0,0 +1,50 @@ +"""The 2025 revisions' transport mechanics, relayed without being interpreted. + +SPEC-v0.2 §6.2's passthrough table, and §6.2's one real cost: **resumption is not relayed for +an intercepted call**. 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 to that action — the outcome would land somewhere the gateway +cannot see, and the reservation would lease-expire into `AMBIGUOUS` despite a perfectly good +answer having been delivered. + +So the gateway strips SSE `id:` fields from an intercepted response stream, making it +non-resumable for the client as it already is for the gateway. Non-intercepted streams are +relayed verbatim, `id:` fields included, because they carry no intercepted outcome. +""" + +from __future__ import annotations + +from collections.abc import Iterable, Iterator +from typing import Final + +#: §6.2 — relayed in both directions, never minted and never interpreted. The gateway holds +#: no sessions; on `2026-07-28` there are none to hold. +SESSION_HEADER: Final = "Mcp-Session-Id" + +#: §6.2 — relayed unchanged on a `GET`, because those streams carry no intercepted outcome. +LAST_EVENT_ID_HEADER: Final = "Last-Event-ID" + +_ID_FIELD: Final = "id:" + + +def strip_event_ids(lines: Iterable[str]) -> Iterator[str]: + """Drop SSE `id:` fields from an intercepted response stream (SPEC-v0.2 §6.2). + + Line-oriented rather than event-oriented on purpose: the gateway must relay each event as + it arrives — a buffered progress notification is not a progress notification — so it + cannot wait for an event to end before deciding what to forward. + + Only the `id:` field is removed. `event:`, `data:` and `retry:` pass through untouched, + as do comments and the blank lines that terminate events, so what the client parses is + the upstream's own stream minus the one field that would let it ask for a replay. + """ + for line in lines: + stripped = line.lstrip() + if stripped.startswith(_ID_FIELD) or stripped == "id": + continue + yield line + + +def is_event_stream(content_type: str | None) -> bool: + """Whether a response is SSE rather than a single JSON body.""" + return content_type is not None and content_type.split(";")[0].strip() == "text/event-stream" diff --git a/src/ctrlrun/gateway/server.py b/src/ctrlrun/gateway/server.py index a8186aab..5c88c4a9 100644 --- a/src/ctrlrun/gateway/server.py +++ b/src/ctrlrun/gateway/server.py @@ -64,6 +64,11 @@ DEFAULT_UPSTREAM_TIMEOUT: Final = 30.0 DEFAULT_APPROVAL_TIMEOUT: Final = 900.0 +#: SPEC-v0.2 §6.9.2's two bounds. A held reservation is a resource an upstream must not be +#: able to pin forever. +DEFAULT_ELICITATION_TIMEOUT: Final = 300.0 +DEFAULT_MAX_ELICITATION_ROUNDS: Final = 8 + #: §6.6 — the alias boundary in an action name must be unambiguous even though tool names #: may contain dots, so the alias may not. ALIAS_PATTERN: Final = r"^[a-z0-9][a-z0-9_-]*$" @@ -77,6 +82,8 @@ BLOCKED: Final = (-41006, "ctrlrun.blocked", 409) NO_PRINCIPAL: Final = (-41007, "ctrlrun.no_principal", 403) UNREPRESENTABLE: Final = (-41008, "ctrlrun.unrepresentable_argument", 400) +UNKNOWN_CONTINUATION: Final = (-41009, "ctrlrun.unknown_continuation", 400) +UPSTREAM_AMBIGUOUS: Final = (-41010, "ctrlrun.upstream_ambiguous", 502) #: §6.8 — the `_meta` key every intercepted response carries, so a client is not left #: guessing what CTRLRun recorded. `com.ctrlrun/` is a legal prefix under the revision's @@ -120,6 +127,8 @@ class GatewayConfig: max_body_bytes: int = DEFAULT_MAX_BODY_BYTES allow_origins: tuple[str, ...] = () allow_remote: bool = False + elicitation_timeout: float = DEFAULT_ELICITATION_TIMEOUT + max_elicitation_rounds: int = DEFAULT_MAX_ELICITATION_ROUNDS def __post_init__(self) -> None: import re @@ -351,6 +360,21 @@ 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, diff --git a/src/ctrlrun/receipt.py b/src/ctrlrun/receipt.py index ce0d3548..75fd64ab 100644 --- a/src/ctrlrun/receipt.py +++ b/src/ctrlrun/receipt.py @@ -75,6 +75,10 @@ class EventType(StrEnum): ACTION_DENIED = "ACTION_DENIED" RECONCILIATION_STARTED = "RECONCILIATION_STARTED" RECONCILIATION_RESOLVED = "RECONCILIATION_RESOLVED" + #: SPEC-v0.2 §11 — named for what happens to the attempt, not for the transport that + #: caused it: `receipt.py` does not learn MCP vocabulary (ARCHITECTURE §6). + EXECUTION_SUSPENDED = "EXECUTION_SUSPENDED" + EXECUTION_RESUMED = "EXECUTION_RESUMED" @dataclass(frozen=True) diff --git a/src/ctrlrun/state.py b/src/ctrlrun/state.py index 137b46d9..54b2d083 100644 --- a/src/ctrlrun/state.py +++ b/src/ctrlrun/state.py @@ -15,6 +15,7 @@ from __future__ import annotations +import hmac import json import logging import os @@ -77,6 +78,14 @@ created_at TEXT, updated_at TEXT ); +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, + rounds INTEGER NOT NULL DEFAULT 0, + updated_at TEXT +); CREATE TABLE IF NOT EXISTS approvals( approval_id TEXT PRIMARY KEY, action_hash TEXT NOT NULL, @@ -368,6 +377,34 @@ 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 + ) -> int: + """Hold this reservation open across an elicitation 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. + + `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. + """ + ... + + def take_continuation(self, effect_key: str, request_state: str) -> EffectRecord: + """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. + """ + ... + + def held_action_hash(self, effect_key: str) -> str | None: + """The `action_hash` of the held continuation for this key, or `None`.""" + ... + def approvals_for(self, action_hash: str) -> tuple[ApprovalRecord, ...]: """Every approval record carrying `action_hash`, oldest first (SPEC-v0.2 §5). @@ -415,6 +452,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]] = {} def close(self) -> None: """Nothing to release; here so either store can be closed the same way.""" @@ -626,6 +664,34 @@ 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 + ) -> 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 + return rounds + 1 + + def take_continuation(self, effect_key: str, request_state: str) -> EffectRecord: + 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: + with self._lock: + held = self._continuations.get(effect_key) + return None if held is None else held[1] + def get_effect(self, effect_key: str) -> EffectRecord | None: with self._lock: return self._effects.get(effect_key) @@ -801,6 +867,76 @@ def put_approval_request(self, request: ApprovalRequest) -> None: 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 + ) -> int: + connection = self._connection() + connection.execute("BEGIN IMMEDIATE") + try: + _holdable( + self._read_effect(connection, effect_key), effect_key, action_id, self._clock() + ) + row = connection.execute( + "SELECT rounds FROM continuations WHERE effect_key=?", (effect_key,) + ).fetchone() + rounds = (row["rounds"] if row else 0) + 1 + connection.execute( + "INSERT INTO continuations(effect_key, action_id, action_hash, request_state, " + "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, " + "updated_at=excluded.updated_at", + ( + effect_key, + action_id, + action_hash, + request_state, + rounds, + _iso(self._clock()), + ), + ) + except BaseException: + self._unwind(connection) + raise + connection.commit() + return rounds + + def take_continuation(self, effect_key: str, request_state: str) -> EffectRecord: + 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"]) + ) + record = _continuable( + self._read_effect(connection, effect_key), + held, + effect_key, + request_state, + 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,) + ) + except BaseException: + self._unwind(connection) + raise + connection.commit() + return record + + def held_action_hash(self, effect_key: str) -> str | None: + row = ( + self._connection() + .execute("SELECT action_hash FROM continuations WHERE effect_key=?", (effect_key,)) + .fetchone() + ) + return None if row is None else str(row["action_hash"]) + def find_granted_approval(self, action_hash: str) -> Approval | None: return _newest_granted(self.approvals_for(action_hash), action_hash, self._clock()) @@ -1254,3 +1390,53 @@ def _newest_denied( return None newest: ApprovalRecord = max(reversed(denied), key=lambda record: record.request.created_at) 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). + + 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. + """ + 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 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, + ) + 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, + effect_key=effect_key, + ) + 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 diff --git a/tests/test_elicitation.py b/tests/test_elicitation.py new file mode 100644 index 00000000..95a49710 --- /dev/null +++ b/tests/test_elicitation.py @@ -0,0 +1,69 @@ +"""Resumption is not relayed for an intercepted call. Build-list item 6d; SPEC-v0.2 §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`. + +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 +to that action — the outcome would land somewhere the gateway cannot see, and the reservation +would lease-expire into `AMBIGUOUS` despite a perfectly good answer having been delivered. +""" + +from __future__ import annotations + +from ctrlrun.gateway.legacy import ( + LAST_EVENT_ID_HEADER, + SESSION_HEADER, + is_event_stream, + strip_event_ids, +) + +# --- T30b: resumption is not relayed for an intercepted call --------------------------- + + +def test_T30b_id_fields_are_stripped_from_an_intercepted_stream(): + stream = [ + "event: message", + "id: 42", + 'data: {"jsonrpc":"2.0"}', + "", + " id: 43", + "data: more", + "", + ] + + assert list(strip_event_ids(stream)) == [ + "event: message", + 'data: {"jsonrpc":"2.0"}', + "", + "data: more", + "", + ] + + +def test_T30b_everything_but_the_id_field_survives(): + stream = ["retry: 1000", ": a comment", "event: x", "data: y", "identity: kept", ""] + + assert list(strip_event_ids(stream)) == stream + + +def test_T30b_a_non_intercepted_stream_keeps_its_ids(): + """Relayed verbatim: those streams carry no intercepted outcome (§6.2).""" + stream = ["id: 7", "data: x", ""] + + assert list(stream) == ["id: 7", "data: x", ""] + + +def test_the_session_header_is_relayed_and_never_minted(): + """§6.2 — relayed in both directions, never minted, never interpreted. The gateway holds + no sessions; on 2026-07-28 there are none to hold.""" + assert SESSION_HEADER == "Mcp-Session-Id" + assert LAST_EVENT_ID_HEADER == "Last-Event-ID" + + +def test_an_sse_content_type_is_recognized_with_or_without_parameters(): + assert is_event_stream("text/event-stream") + assert is_event_stream("text/event-stream; charset=utf-8") + assert not is_event_stream("application/json") + assert not is_event_stream(None) diff --git a/tests/test_gateway.py b/tests/test_gateway.py index bef16d91..08743182 100644 --- a/tests/test_gateway.py +++ b/tests/test_gateway.py @@ -344,3 +344,126 @@ def test_the_two_lookups_agree_with_the_record_the_store_holds(state_store, fake assert state_store.get_approval("apr_1").status is ApprovalStatus.DENIED assert state_store.find_granted_approval(action.action_hash) is None assert state_store.find_denied_request(action.action_hash).request_id == "apr_1" + + +# --- §6.9.2: the held continuation ------------------------------------------------------ + +HASH = "sha256:" + "a" * 64 +OTHER_HASH = "sha256:" + "b" * 64 +STATE = "opaque-server-state-1" + + +def test_a_held_continuation_is_taken_exactly_once(sqlite_store, fake_clock): + """§6.9.2 — one elicitation admits one continuation. + + 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(KEY, STATE) + + assert held.action_id == MINE + assert held.state is EffectState.EXECUTING + with pytest.raises(CTRLRunError): + sqlite_store.take_continuation(KEY, STATE) + + +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) + + 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_a_continuation_for_a_key_that_holds_none_is_refused(sqlite_store): + _executing(sqlite_store) + + with pytest.raises(CTRLRunError): + sqlite_store.take_continuation(KEY, STATE) + + +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)) + + with pytest.raises(AmbiguousEffect): + sqlite_store.take_continuation(KEY, STATE) + + +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_holding_a_continuation_needs_a_live_executing_record_this_action_holds(sqlite_store): + sqlite_store.reserve_effect(KEY, MINE, LEASE) + + with pytest.raises(CTRLRunError): + sqlite_store.hold_continuation(KEY, MINE, STATE, action_hash=HASH) + + sqlite_store.begin_execution(KEY, MINE) + with pytest.raises(DuplicateEffect): + sqlite_store.hold_continuation(KEY, THEIRS, STATE, action_hash=HASH) + + +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) + + 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 + + +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_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.""" + 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) + finally: + first.close() + + second = SQLiteStateStore(path, clock=fake_clock) + try: + assert second.take_continuation(KEY, STATE).action_id == MINE + 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) + + 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)