item 6d (partial): the continuation store, and resumption stripping

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"

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,

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)

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