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)