fix(gateway): re-attach after session.replay_gap instead of failing the turn

On overflow the owner retires a viewer's subscription and sends one
session.replay_gap while the turn runs on. Every Python subscriber treated
that frame as a failed turn: the classic CLI reported "outcome unknown" and
then hung on every later turn, GatewayChatView (stream-json, -z,
--safe-mode) failed, and ACP stopped its event pump, failing the waiting
prompts of every session. #140292 made gaps much rarer by raising the frame
cap, but kept the overflow path.

A gap now means "this subscription ended; re-attach". The client resumes
that session only, adopts the new subscription, wakes whatever waits on its
pending rows or controls, and answers the controls the snapshot lists. The
snapshot is taken under the event-stream lock: a queued or started
admission it lists publishes its completion on the new subscription; an
unknown one owes none and settles as unknown now, in every client; one it
no longer lists ended in the hole and its receipt settles it, read once and
only against a snapshot that covers it (one whose resume left after the
submit). ACP's pump is the one recovery owner, so a reply is delivered at
most once. After a gap a control blocks only an admission the snapshot
still lists.

Live deltas stop at the gap and the settled reply shows once (the rest of
an exact prefix, else "[output resumed]" and the whole reply), to the
terminal and to the voice stream separately. An interim stays whole, as it
can be the reply's only copy. stream-json emits no further text and flags
its result with replay_gap. After a gap or from a receipt, a reused reply
is suppressed only when all of it was shown; without a gap main's rule
holds. A refused re-attach still ends the wait with today's error; an
unreadable receipt fails only that admission. A CLI turn owns its turn
token and its session from the pre-submit re-attach through _result and
painting, bounded by the interrupt settle deadline, and any received
boundary closes the reply segment.

The real-rig tests share one real owner that pins the fanout frame cap at
256, as #140292's own overflow probe does, so the paused viewer still
overflows.

Refs #134889
Fixes #140388

(cherry picked from commit 0d51f10640c560e0cb0b2205e2dbad18d06c7c5c)

acp_adapter/gateway_reply.py

@@ -1,6 +1,6 @@
 """How a gateway turn's reply reaches the editor: ``messageId`` grouping, exactly-once final text, tool and
 permission bubbles closed as the in-process agent closed them. Mixed into ``GatewayACPAgent``, whose state
-(``_conn``, ``_streamed``, ``_segments``, ``_tool_args``, ``_message_ids``) it reads and updates."""
+(``_conn``, ``_streamed``, ``_segments``, ``_tool_args``, ``_message_ids``, ``_cut``) it reads and updates."""
 from __future__ import annotations
 
 import logging
@@ -104,13 +104,22 @@ class GatewayReplyMixin:
         text = _text(payload)
         streamed = self._streamed.pop(aid, "")
         shown = self._segments.pop(aid, [])
+        cut = self._cut.pop(aid, None)
         # Local interrupt status is metadata; ACP carries it in stop_reason.
         if not text or (payload.get("outcome") == "cancelled"
                         and text.startswith(INTERRUPT_WAITING_FOR_MODEL_PREFIX)):
             return ""
-        if payload.get("response_reused") is True and (streamed or shown):
-            # The owner names the final as the reply already painted (never inferred from text).
+        segments = {s.strip() for s in shown}
+        if payload.get("response_reused") is True and (
+                text.strip() in segments | {streamed.strip()} if cut is not None or payload.get("receipt")
+                else (streamed or shown)):
+            # The owner names the final as a reply already painted (never inferred from text); after a
+            # replay gap or from a receipt, only when this editor showed all of it.
             return ""
+        if cut is not None:  # a replay gap stopped its live text: the rest once, or marked in full
+            from hermes_cli.gateway_replay_gap import owed_reply
+            # An interim after the gap closed the cut segment and may already have shown the whole reply.
+            return "" if not cut and text.strip() in segments else owed_reply(cut, text)
         if text.startswith(streamed):
             remainder = text[len(streamed):]
             # An unstreamed final the owner already published as an interim segment.

acp_adapter/gateway_server.py

@@ -96,6 +96,10 @@ class GatewayACPAgent(GatewayReplyMixin, acp.Agent):
         self._settling = set()
         # Admissions a terminal receipt already delivered: their late message.complete adds nothing.
         self._receipted = {}
+        # ``gateway_replay_gap``: admission -> the reply segment shown when a gap cut it; own prompts awaiting
+        # their terminal, in admission order; those whose reply the hole lost for good; per session, the gap
+        # re-attaches sent and adopted (a snapshot covers the submits sent before its resume).
+        self._cut, self._inflight, self._lost, self._resumes, self._adopted = {}, {}, {}, {}, {}
         self._pending_cancels = set()
         # (session, tool_call_id) -> (tool name, args, admission) of a call still open on the editor.
         self._tool_args = {}
@@ -232,6 +236,7 @@ class GatewayACPAgent(GatewayReplyMixin, acp.Agent):
         self._streamed.clear()
         self._segments.clear()
         self._message_ids.clear()
+        self._cut.clear()
         async with self._changed:
             # Every waiter on the old transport wakes to its ambiguous failure, even one that had
             # not yet observed ``_failure`` before it is cleared here.
@@ -587,7 +592,7 @@ class GatewayACPAgent(GatewayReplyMixin, acp.Agent):
         try:
             # Inside the try: a failed reconnect is as ambiguous for a retained identity as a lost ack.
             client = await self._client()
-            link = self._link
+            link, resumes = self._link, self._resumes.get(session_id, 0)
             receipt = await client.rpc("prompt.submit", session_id=session_id, input_id=input_id, **submit)
         except BaseException as exc:
             from hermes_cli.gateway_client import GatewayRPCError
@@ -610,13 +615,15 @@ class GatewayACPAgent(GatewayReplyMixin, acp.Agent):
             # A retry of an un-acked submit the owner lost in a restart: never a second turn.
             self._admissions.pop(session_id, None)
             raise GatewayClientError(self._unknown_recovery(session_id, admission_id))
+        self._inflight[(session_id, admission_id)] = None
         try:
+            await self._after_ack(session_id, admission_id, link, resumes)
             if session_id in self._pending_cancels:
                 self._pending_cancels.discard(session_id)
                 await self._cancel_admission(session_id, admission_id)
             async with self._changed:
                 await self._changed.wait_for(lambda: admission_id in self._terminals or self._failure is not None
-                                            or self._link != link
+                                            or self._link != link or admission_id in self._lost
                                             or self._blocked_admission(session_id, admission_id))
                 # A successor's unknown state cannot replace our committed terminal result.
                 if admission_id not in self._terminals:
@@ -626,13 +633,15 @@ class GatewayACPAgent(GatewayReplyMixin, acp.Agent):
                         raise self._lost_failure
                     if self._failure:
                         raise self._failure
-                    raise GatewayClientError(self._unknown_recovery(session_id, admission_id))
+                    raise self._lost.pop(admission_id, None) or GatewayClientError(
+                        self._unknown_recovery(session_id, admission_id))
                 terminal = self._terminals.pop(admission_id)
         finally:
             # Do not let one prompt tear down a newer mapping if the session has
             # already advanced while this coroutine unwinds.
             if self._admissions.get(session_id) == admission_id:
                 self._admissions.pop(session_id, None)
+            self._inflight.pop((session_id, admission_id), None)
         outcome = terminal.get("outcome")
         if outcome == "unknown":
             # Our own lost turn: the same refusal whether session.info or this completion woke us.
@@ -688,10 +697,50 @@ class GatewayACPAgent(GatewayReplyMixin, acp.Agent):
             return None
         return retry_key, uuid.uuid4().hex, staged
 
-    async def _settled_receipt(self, client, held_id, session_id, receipt):
+    async def _after_ack(self, session_id, admission_id, link, resumes):
+        """A gap re-attach whose resume left after this submit covers it. The pump owns recovery: it
+        recovers this admission against that snapshot behind the ones it already holds (FIFO, once)."""
+        if self._resumes.get(session_id, 0) == resumes:
+            return
+        async with self._changed:
+            await self._changed.wait_for(lambda: self._adopted.get(session_id, 0) > resumes
+                                         or self._failure is not None or self._link != link)
+        if self._adopted.get(session_id, 0) > resumes:
+            with suppress(asyncio.QueueFull):  # full: the reader's own overflow ends this link
+                self._gateway.events.put_nowait((session_id, admission_id))
+
+    async def _recover(self, session_id, admission_id):
+        """Queued: it lost nothing. Running: cut until its live completion. Unknown: the waiter refuses it.
+        Unlisted: it ended in the hole and its receipt settles it (unreadable: that prompt alone fails).
+        Terminal evidence received, being delivered or already delivered always wins."""
+        from hermes_cli.gateway_replay_gap import pending_status, settled
+        status = pending_status(self._snapshots[session_id].get("pending"), admission_id)
+        if status in ("queued", "unknown") or {admission_id} & (
+                self._terminals.keys() | self._settling | self._receipted.keys()):
+            return
+        self._cut.setdefault(admission_id, self._streamed.pop(admission_id, ""))
+        if status:
+            return
+        try:
+            receipt = await self._gateway.rpc("prompt.receipt", session_id=session_id, admission_id=admission_id,
+                                              include_result=True)
+        except (GatewayClientError, OSError, TimeoutError):
+            receipt = {"status": "terminal"}  # unreadable
+        verdict = settled(receipt)
+        if verdict is not False:  # its one recovery read: a later entry (a post-ack request) joins it
+            self._receipted.setdefault(admission_id, None)
+        if verdict:
+            await self._settled_receipt(self._gateway, self._editor_id(session_id), session_id, receipt, receipt)
+        elif verdict is None and admission_id not in self._terminals and admission_id not in self._settling:
+            async with self._changed:
+                self._lost[admission_id] = GatewayClientError("session_replay_gap")
+                self._changed.notify_all()
+
+    async def _settled_receipt(self, client, held_id, session_id, receipt, saved=None):
+        """``saved``: the receipt already fetched with its result (a replay gap's recovery read)."""
         admission_id = receipt["admission_id"]
-        saved = await client.rpc("prompt.receipt", session_id=session_id, admission_id=admission_id,
-                                 include_result=True)
+        saved = saved or await client.rpc("prompt.receipt", session_id=session_id, admission_id=admission_id,
+                                          include_result=True)
         if admission_id in self._terminals or admission_id in self._settling:
             return  # this connection's pump carried the completion itself
         self._settling.add(admission_id)  # claimed before the first await; the pump yields to it
@@ -700,8 +749,10 @@ class GatewayACPAgent(GatewayReplyMixin, acp.Agent):
             self._receipted.pop(next(iter(self._receipted)))
         try:
             outcome = {"interrupted": "cancelled"}.get(receipt.get("outcome"), receipt.get("outcome"))
-            text = (saved.get("result") or {}).get("final_response") or ""
-            await self._finish_reply(session_id, held_id, admission_id, {"text": text, "outcome": outcome})
+            result = saved.get("result") or {}
+            await self._finish_reply(session_id, held_id, admission_id, {
+                "text": result.get("final_response") or "", "outcome": outcome,
+                "response_reused": result.get("response_reused"), "receipt": True})
             async with self._changed:
                 self._terminals[admission_id] = {"outcome": outcome}
                 self._changed.notify_all()
@@ -786,6 +837,9 @@ class GatewayACPAgent(GatewayReplyMixin, acp.Agent):
                 frame = await gateway.events.get()
                 if isinstance(frame, Exception):
                     raise frame
+                if isinstance(frame, tuple):  # a post-ack recovery (``_after_ack``), behind every frame so far
+                    await self._recover(*frame)
+                    continue
                 event = frame.get("params", {})
                 sid = event.get("session_id")
                 snapshot = self._snapshots.get(sid)
@@ -814,8 +868,8 @@ class GatewayACPAgent(GatewayReplyMixin, acp.Agent):
         if kind in {"session.info", "session.updated", "session.usage", "session.tools_changed"}:
             await self._project_session_state(kind, sid, editor_id, payload, event)
             return
-        if kind == "session.replay_gap":
-            raise GatewayClientError("session_replay_gap")
+        if await self._gap_guard(kind, sid, aid):
+            return
         if kind in {"approval.request", "clarify.request"}:
             self._permission(sid, payload)
             return
@@ -863,6 +917,32 @@ class GatewayACPAgent(GatewayReplyMixin, acp.Agent):
             finally:
                 self._settling.discard(aid)
 
+    async def _gap_guard(self, kind, sid, aid):
+        """What a replay gap changes in the projection; True when the frame is consumed here."""
+        if kind == "session.replay_gap":
+            # Re-attach that session alone (a refusal still fails the pump: the next call redials).
+            params = {"editor": self._editor_mcp[sid]} if sid in self._editor_mcp else {}
+            self._resumes[sid] = self._resumes.get(sid, 0) + 1
+            snapshot = await self._gateway.rpc("session.resume", session_id=sid, **params)
+            async with self._changed:  # its pending rows and controls may settle or block a waiter
+                self._snapshots[sid], self._adopted[sid] = snapshot, self._resumes[sid]
+                self._changed.notify_all()
+            for prompt in snapshot.get("prompts", []):
+                self._permission(sid, prompt)
+            started = [row["admission_id"] for row in snapshot.get("pending", []) if row.get("status") == "started"]
+            for admission_id in dict.fromkeys([*started, *(a for held, a in self._inflight if held == sid)]):
+                await self._recover(sid, admission_id)
+            return True
+        if aid not in self._cut:
+            return False
+        if kind in {"tool.start", "message.interim"}:  # a new segment: what was shown before it is closed
+            if self._cut[aid].strip():
+                self._segments.setdefault(aid, []).append(self._cut[aid])
+            self._cut[aid] = ""
+            self._end_message(sid)
+        # Live deltas stopped at the gap; an interim is whole (it may be the only copy of the reply).
+        return kind == "message.delta"
+
     async def _project_session_state(self, kind, sid, editor_id, payload, event):
         if kind == "session.usage":
             await self._project_usage(sid, editor_id, payload)

hermes_cli/gateway_agent.py

@@ -18,7 +18,7 @@ import time
 import uuid
 
 from hermes_cli.gateway_client import GatewayClientError, GatewayRPCError
-from hermes_cli.gateway_connection import DISCONNECTED, lost_before
+from hermes_cli.gateway_connection import DISCONNECTED, CallStopped, lost_before
 
 logger = logging.getLogger(__name__)
 
@@ -64,6 +64,17 @@ class _TodoView:
         return list(self.items)
 
 
+class _Turn:
+    """One ``run_conversation``; what each consumer got of the reply's current segment (``shown``: the
+    terminal, ``spoken``: ``stream_callback``) and the closed segments and interims (``seen``), for the
+    settlement after a replay gap."""
+
+    def __init__(self, session_id):
+        self.session_id, self.admission = session_id, None
+        self.shown = self.spoken = ""
+        self.seen, self.answered, self.gapped = set(), set(), False
+
+
 class GatewayAgent:
     """Gateway-backed stand-in for ``run_agent.AIAgent`` (constructed at the same call site)."""
 
@@ -152,6 +163,7 @@ class GatewayAgent:
         self._todo_store = _TodoView()
         self._generation = None
         self._admission = None
+        self._turn = None  # a turn a later one replaced stops taking frames
         self._epoch = None  # the connection epoch this turn was submitted on
         self._interrupt_sent_at = None
         self._rate_limit_state = None
@@ -568,6 +580,7 @@ class GatewayAgent:
 
     def run_conversation(self, user_message, conversation_history=None, stream_callback=None,
                          task_id=None, persist_user_message=None, turn_author=None, **_kw):
+        turn = self._turn = _Turn(self.session_id)  # first: a turn main gave up on stops consuming
         history = list(conversation_history or [])
         self._turn_ask_index = len(history)  # the staged ask's row in this turn's transcript
         self._turn_api_calls = None
@@ -584,8 +597,15 @@ class GatewayAgent:
         else:
             self._drain_stale()
             try:
-                receipt = self.connection.call("prompt.submit", **self._submit_params(
-                    user_message, persist_user_message, turn_author))
+                state = self._state()
+                if state.gaps != state.attached_gaps:  # a replay gap between turns retired the subscription
+                    self.connection.reattach(turn.session_id, abandon=self._stop(turn))
+                params = self._submit_params(user_message, persist_user_message, turn_author)
+                if self._stop(turn)():
+                    raise CallStopped("prompt.submit")
+                receipt = self.connection.call("prompt.submit", **{**params, "session_id": turn.session_id})
+            except CallStopped:  # main moved on (a later turn, /new, Ctrl-C's bound) before the submit
+                return self._failed(history, user_message, "Stopped before the prompt was sent; nothing was sent.")
             except GatewayRPCError as exc:
                 # The owner answered and refused: not a lost connection. Only its own site code
                 # (session_busy: send again) is a verdict; any other refusal stays unclassified.
@@ -595,18 +615,24 @@ class GatewayAgent:
             except GatewayClientError as exc:
                 return self._failed(history, user_message, str(exc), failure_reason="gateway_connection")
         self._epoch = getattr(self.connection, "epoch", None)
-        self._admission = (receipt or {}).get("admission_id")
+        turn.admission = self._admission = (receipt or {}).get("admission_id")
         if receipt and receipt.get("execution_generation") is not None:
             self._generation = receipt["execution_generation"]
         starts: dict = {}
         try:
-            outcome = self._pump(stream_callback, starts)
+            outcome = self._pump(stream_callback, starts, turn)
         finally:
-            self._generation_done()
-        if isinstance(outcome, Exception):
+            if self._owns(turn):
+                self._generation_done()
+        if isinstance(outcome, Exception) and self._owns(turn):
             return self._failed(history, user_message, str(outcome) or DISCONNECTED,
                                 failure_reason="gateway_connection")
-        return self._result(history, user_message, outcome)
+        result = self._result(history, user_message, outcome, turn) if self._owns(turn) else None
+        if not self._owns(turn):  # a later turn or a session switch took over: it reads and paints nothing
+            return self._failed(history, user_message, "A later turn took over this terminal.")
+        if turn.gapped:
+            self._paint_owed(turn, result, outcome, stream_callback)
+        return result
 
     def _drain_stale(self):
         """Frames left from an earlier turn of this viewer (a late settlement) are not this turn's."""
@@ -624,18 +650,31 @@ class GatewayAgent:
     def _generation_done(self):
         self._generation = None
 
-    def _pump(self, stream_callback, starts):
+    def _expired(self):
+        return self._interrupt_sent_at is not None and time.monotonic() - self._interrupt_sent_at > INTERRUPT_SETTLE_SECONDS
+
+    def _owns(self, turn):
+        return self._turn is turn and self.session_id == turn.session_id
+
+    def _stop(self, turn):
+        """The ``abandon`` of a turn's blocking reads: it lost the terminal, or Ctrl-C's settle bound passed."""
+        return lambda: not self._owns(turn) or self._expired()
+
+    def _pump(self, stream_callback, starts, turn):
         """Dispatch frames until this admission's ``message.complete``; returns its payload, or the
         transport error (disconnect)."""
         events = self.connection.events
-        while True:
-            if (self._interrupt_sent_at is not None
-                    and time.monotonic() - self._interrupt_sent_at > INTERRUPT_SETTLE_SECONDS):
+        while self._owns(turn):
+            if self._expired():
                 return {"status": "interrupted", "outcome": "cancelled", "_settle_timeout": True}
             try:
                 frame = events.get(timeout=_POLL)
             except queue.Empty:
                 continue
+            if not self._owns(turn):
+                with events.mutex:  # the later turn's frame goes back where it was
+                    events.queue.appendleft(frame)
+                break
             if isinstance(frame, Exception):
                 if lost_before(frame, self._epoch):
                     continue  # the previous connection's loss, already reported; this turn is live
@@ -656,11 +695,45 @@ class GatewayAgent:
             if kind == "message.complete":
                 return payload
             if kind == "session.replay_gap":
-                return GatewayClientError("session_replay_gap: " + DISCONNECTED)
+                outcome = self._recover(turn)
+                if outcome is not None:
+                    return outcome
+                continue
             try:
                 self._dispatch(kind, payload, stream_callback, starts)
             except Exception:
                 logger.debug("event %s dispatch failed", kind, exc_info=True)
+        return None
+
+    def _recover(self, turn):
+        """``gateway_replay_gap``: re-attach, answer the controls the hole hid, stop painting live text. A
+        turn the snapshot no longer lists settles from its receipt; None waits for the live completion."""
+        from hermes_cli.gateway_replay_gap import settled
+        lost, stop = GatewayClientError("session_replay_gap: " + DISCONNECTED), self._stop(turn)
+        try:
+            snapshot = self.connection.reattach(turn.session_id, abandon=stop)
+            row = next((r for r in snapshot.get("pending") or [] if r.get("admission_id") == turn.admission), {})
+            if row.get("status") == "unknown":
+                return lost  # its completion was the one the hole lost: no other is owed
+            turn.gapped |= row.get("status") != "queued"  # one still queued lost nothing
+            for prompt in snapshot.get("prompts") or []:
+                if row.get("status") == "started" and prompt.get("execution_generation") == row.get("execution_generation") \
+                        and prompt.get("prompt_id") not in turn.answered and not stop():
+                    turn.answered.add(prompt.get("prompt_id"))
+                    try:
+                        self._dispatch(f"{prompt.get('kind')}.request", prompt, None, {})
+                    except Exception:  # a broken modal is a skipped answer, never a failed turn
+                        logger.debug("control %s dispatch failed", prompt.get("prompt_id"), exc_info=True)
+            if row or not turn.admission or stop():
+                return None
+            receipt = self.connection.call("prompt.receipt", session_id=turn.session_id, admission_id=turn.admission,
+                                           include_result=True, abandon=stop)
+        except CallStopped:
+            return None
+        except GatewayClientError:
+            return lost
+        verdict = settled(receipt or {})
+        return {"_receipt": receipt} if verdict else lost if verdict is None else None
 
     def _dispatch(self, kind, payload, stream_callback, starts):
         handler = _HANDLERS.get(kind)
@@ -669,14 +742,30 @@ class GatewayAgent:
 
     # event handlers (agent thread) ------------------------------------------------------------
     def _on_delta(self, payload, stream_callback, starts):
-        text = payload.get("text")
+        text, turn = payload.get("text"), self._turn
+        if not isinstance(text, str):
+            return self._close_segment(starts, separator=True)
+        if turn.gapped:
+            return  # live text stopped at the replay gap; the settlement shows the reply once
         if self.stream_delta_callback is not None:
             self.stream_delta_callback(text)
-        if stream_callback is not None and isinstance(text, str):
+            turn.shown += text
+        if stream_callback is not None:
             stream_callback(text)
+            turn.spoken += text
+
+    def _close_segment(self, starts, separator=False):
+        """A segment boundary: the ``text: None`` separator, or a later boundary when it fell in a gap."""
+        turn = self._turn
+        if self.stream_delta_callback is not None and (separator or turn.shown):
+            self.stream_delta_callback(None)
+        turn.seen |= {turn.shown.strip(), turn.spoken.strip()}
+        turn.shown = turn.spoken = ""
 
     def _on_interim(self, payload, stream_callback, starts):
         text = payload.get("text")
+        self._close_segment(starts)
+        self._turn.seen.add(str(text or "").strip())
         if self.interim_assistant_callback is not None:
             self.interim_assistant_callback(text, already_streamed=bool(payload.get("already_streamed")))
         elif text and str(text).strip():
@@ -687,6 +776,7 @@ class GatewayAgent:
             self.tool_gen_callback(payload.get("name"))
 
     def _on_tool_start(self, payload, stream_callback, starts):
+        self._close_segment(starts)
         name = payload.get("name") or payload.get("tool_name") or "tool"
         args = payload.get("args") if isinstance(payload.get("args"), dict) else {}
         call_id = payload.get("tool_id") or payload.get("tool_call_id") or ""
@@ -908,12 +998,14 @@ class GatewayAgent:
             out.update(failure_reason=failure_reason, failure_retryable=retryable)
         return out
 
-    def _result(self, history, user_message, complete):
-        receipt = {}
-        if self._admission:
+    def _result(self, history, user_message, complete, turn):
+        receipt = (complete or {}).get("_receipt") or {}
+        if receipt:  # settled from it after a replay gap
+            complete = {"outcome": receipt.get("outcome")}
+        elif turn.admission:
             try:
-                receipt = self.connection.call("prompt.receipt", session_id=self.session_id,
-                                               admission_id=self._admission, include_result=True) or {}
+                receipt = self.connection.call("prompt.receipt", session_id=turn.session_id, abandon=self._stop(turn),
+                                               admission_id=turn.admission, include_result=True) or {}
             except GatewayClientError as exc:
                 logger.debug("prompt.receipt failed: %s", exc)
         stored = dict(receipt.get("result") or {})
@@ -935,12 +1027,29 @@ class GatewayAgent:
         if not suffix and out.get("final_response") and not out.get("failed"):
             out["messages"].append({"role": "assistant", "content": out["final_response"]})
         out["session_id"] = self.session_id
+        if not self._owns(turn):  # taken over during that read: the newer turn's state stays untouched
+            return out
         self._session_messages = out["messages"]
         calls = stored.get("api_calls")
         self._turn_api_calls = calls if isinstance(calls, int) and not isinstance(calls, bool) else None
-        self._apply_usage(stored, usage)
+        self._apply_usage(stored, usage, turn)
         return out
 
+    def _paint_owed(self, turn, out, complete, stream_callback):
+        """After a replay gap, the reply once (``owed_reply``): a reuse this terminal showed whole stays
+        the only copy; otherwise main's panel shows what streaming did not."""
+        from hermes_cli.gateway_replay_gap import owed_reply
+        reply = out.get("pre_transform_response") if out.get("response_transformed") else None
+        reply = reply if isinstance(reply, str) else out.get("final_response")
+        if out.get("failed") or not isinstance(reply, str) or not reply.strip():
+            return
+        reused = out.get("response_previewed") or out.get("response_reused") or complete.get("response_reused")
+        out["response_previewed"] = bool(reused) and reply.strip() in turn.seen | {turn.shown.strip(), turn.spoken.strip()}
+        for consumer, got in ((self.stream_delta_callback, turn.shown), (stream_callback, turn.spoken)):
+            owed = owed_reply(got, reply) if consumer is not None and not out["response_previewed"] else ""
+            if owed:  # each consumer completes what it got itself, never twice
+                consumer(owed)
+
     def _anchor_display(self):
         """Main's bar shows the anchored ``context_tokens`` by re-deriving it from
         ``_turn_base_usage_anchor`` over ``_session_messages`` (``anchored_context_tokens``), and marks
@@ -979,23 +1088,25 @@ class GatewayAgent:
         estimate = estimate_messages_tokens_rough(delta, charge_stale_thinking=False) if delta else 0
         self._turn_base_usage_anchor = capture_usage_anchor(max(1, tokens - estimate), 0, messages[:base])
 
-    def _refresh_stats(self):
+    def _refresh_stats(self, turn=None):
         """``session.stats`` → the same attributes ``session.usage`` sets; False when unavailable."""
         try:
-            stats = self.connection.call("session.stats", session_id=self.session_id, timeout=10)
+            stats = self.connection.call("session.stats", session_id=self.session_id, timeout=10,
+                                         abandon=turn and self._stop(turn))
         except GatewayClientError as exc:
             logger.debug("session.stats unavailable: %s", exc)
             return False
         stats = stats.get("usage") if isinstance(stats, dict) and isinstance(stats.get("usage"), dict) else stats
         if not (isinstance(stats, dict) and isinstance(stats.get("api_calls"), int)):
             return False
-        self._state().apply_usage(stats)
+        if turn is None or self._owns(turn):
+            self._state().apply_usage(stats)
         return True
 
-    def _apply_usage(self, stored, receipt_usage):
+    def _apply_usage(self, stored, receipt_usage, turn):
         """Turn end: the session-cumulative ``session.stats`` (the same dict ``session.usage`` carries);
         a gateway without the verb leaves the per-turn receipt figures as before."""
-        if self._refresh_stats():
+        if self._refresh_stats(turn) or not self._owns(turn):
             return
         # Fallback: the classic-CLI result carries the gateway agent's session_* counters
         # (turn_finalizer _SESSION_TOKEN_KEYS); per-turn receipt usage only accumulates what is missing.

hermes_cli/gateway_chat_view.py

@@ -26,6 +26,9 @@ def _pending_preview(text):
     return f": {line}" if line else ""
 
 
+_UNKNOWN_OUTCOME = ("Execution outcome is unknown. Resume this session to inspect the lost turn "
+                    "and /discard it; do not resend the input.")
+
 _BLOCKING_CONTROL_EVENTS = frozenset({
     "approval.request", "approval.settled", "clarify.request", "clarify.settled",
     "session.info",
@@ -53,6 +56,8 @@ class GatewayChatView:
         self._finite_events = []
         self.streams = {}
         self.completions = {}
+        # ``gateway_replay_gap``: admission -> the reply segment shown when a gap cut it; own unsettled turns.
+        self.cut, self.mine, self.gaps, self._attaching = {}, {}, 0, asyncio.Lock()
         self.changed = asyncio.Event()
         self.failure = None
         # Set once with ``failure``: the interactive composer races its line read against it so a
@@ -103,8 +108,10 @@ class GatewayChatView:
             kind, payload = params.get("type"), params.get("payload", {})
             self.generation = params.get("execution_generation", self.generation)
             if kind == "session.replay_gap":
-                self._fail(GatewayClientError("session_replay_gap"))
-                return
+                await self._reattach()
+                if self.failure:
+                    return
+                continue
             admission = params.get("admission_id") or payload.get("admission_id")
             if self.finite and kind in _BLOCKING_CONTROL_EVENTS:
                 # Gates and unknown executions block the session FIFO, not merely one
@@ -129,6 +136,55 @@ class GatewayChatView:
         self.failed.set()
         self.changed.set()
 
+    async def _reattach(self):
+        """The owner retired this view's subscription; its turns run on. Re-attach (a refusal fails the
+        view as before), show the controls the hole hid, and recover the turns in admission order."""
+        sid = self.session_id
+        async with self._attaching:  # a submit acked meanwhile recovers against this snapshot, which covers it
+            self.gaps += 1
+            try:
+                snapshot = await self.client.rpc("session.resume", session_id=sid)
+            except Exception:  # health: allow BLE001 -- any failed re-attach ends the wait, never hangs it
+                if self.session_id == sid:
+                    self._fail(GatewayClientError("session_replay_gap"))
+                return
+            if self.session_id != sid:
+                return  # /new or /branch moved this view meanwhile
+            self.generation = snapshot.get("execution_generation", self.generation)
+            self.pending, prompts = snapshot.get("pending", self.pending), {p["prompt_id"]: p for p in snapshot.get("prompts", [])}
+            for prompt_id in prompts.keys() - self.prompts.keys() if not self.finite else ():
+                self.show_prompt(prompts[prompt_id])
+            self.prompts = prompts
+            for admission in dict.fromkeys([*self.streams, *self.mine]):  # the running turn first: one FIFO
+                await self._recover(admission, sid)
+            self.changed.set()  # after its own turn settled: a control it lists may end a one-shot's wait
+
+    async def _recover(self, admission, sid):
+        """Queued: it lost nothing. Running: cut until its live completion. Unknown: its completion was the
+        one lost, so an own turn settles as unknown now. Unlisted: it ended in the hole."""
+        from hermes_cli.gateway_replay_gap import pending_status, settled
+        status = pending_status(self.pending, admission)
+        if status == "unknown" and admission in self.mine:
+            self._complete(admission, {"outcome": "unknown", "text": _UNKNOWN_OUTCOME})
+            self.changed.set()
+        if not admission or status in ("queued", "unknown") or admission in self.completions:
+            return
+        self.cut.setdefault(admission, self.streams.pop(admission, ""))
+        if status or any(kind == "message.complete" and own == admission for kind, own, _ in self._finite_events):
+            return  # (its completion already arrived, held for the submit ack)
+        try:
+            receipt = await self.client.rpc("prompt.receipt", session_id=sid, admission_id=admission, include_result=True)
+        except Exception:  # health: allow BLE001 -- unreadable, like a receipt without its reply
+            receipt = {"status": "terminal"}
+        verdict = settled(receipt) if self.session_id == sid and admission not in self.completions else False
+        if verdict:
+            outcome = {"interrupted": "cancelled"}.get(receipt.get("outcome"), receipt.get("outcome"))
+            self._complete(admission, {"outcome": outcome, "_receipt": receipt,
+                                       "text": (receipt.get("result") or {}).get("final_response") or ""})
+            self.changed.set()
+        elif verdict is None and self.finite:  # unreadable: fails as before (interactive: the next gap retries)
+            self._fail(GatewayClientError("session_replay_gap"))
+
     async def _read_line(self, prompt, symbol):
         """The next composer line, or ``""`` once the event stream failed first (``self.failure``
         is then set). The pending read is cancelled (prompt_toolkit restores the terminal) so the
@@ -171,6 +227,8 @@ class GatewayChatView:
     def _tool_start(self, admission, payload):
         # Deltas before a tool call are interim commentary the final reply does not repeat;
         # close that segment so `_complete` measures only the final's own stream.
+        if admission in self.cut:
+            self.cut[admission] = ""
         if self.streams.pop(admission, None) and not self.quiet:
             print(flush=True)
         if self.emitter is not None:
@@ -191,8 +249,8 @@ class GatewayChatView:
 
     def _delta(self, admission, payload):
         text = payload.get("text") or payload.get("delta") or payload.get("content") or ""
-        if not isinstance(text, str):
-            return
+        if not isinstance(text, str) or admission in self.cut:
+            return  # cut: live text stopped at the replay gap; the settlement shows the reply once
         if self.emitter is not None:
             self.emitter.on_text_delta(text)
         elif not self.quiet:
@@ -204,8 +262,14 @@ class GatewayChatView:
 
     def _complete(self, admission, payload):
         text = payload.get("text") or payload.get("content") or ""
-        streamed = self.streams.pop(admission, "")
-        if not self.quiet:
+        streamed, cut = self.streams.pop(admission, ""), self.cut.pop(admission, None)
+        self.mine.pop(admission, None)
+        if cut is not None:
+            from hermes_cli.gateway_replay_gap import owed_reply
+            payload = {**payload, "replay_gap": True}
+            if not self.quiet and text:
+                print(owed_reply(cut, text), flush=True)
+        elif not self.quiet:
             # The final's deltas carry the agent's segment break (leading blank lines after a
             # tool call) that the settled text has trimmed; a match modulo that edge whitespace
             # is the same reply already on screen.
@@ -231,10 +295,19 @@ class GatewayChatView:
 
     async def submit(self, text):
         self.mutations.retire(self.session_id)
-        return await self.client.rpc("prompt.submit", session_id=self.session_id,
-                                     input_id=uuid.uuid4().hex, text=text,
-                                     **({"finite": True} if self.finite else {}),
-                                     **({"unattended": True} if self.finite and self.unattended else {}))
+        sid, gaps = self.session_id, self.gaps
+        receipt = await self.client.rpc("prompt.submit", session_id=self.session_id,
+                                        input_id=uuid.uuid4().hex, text=text,
+                                        **({"finite": True} if self.finite else {}),
+                                        **({"unattended": True} if self.finite and self.unattended else {}))
+        admission = (receipt or {}).get("admission_id")
+        if admission and admission not in self.completions and self.session_id == sid:
+            self.mine[admission] = None
+            if self.gaps != gaps:  # a re-attach left after this submit: its snapshot covers it
+                async with self._attaching:
+                    if admission not in self.cut and not self.failure:
+                        await self._recover(admission, sid)
+        return receipt
 
     async def command(self, text):
         command, _, rest = text.partition(" ")
@@ -353,14 +426,12 @@ class GatewayChatView:
         only a still-queued input behind someone else's unknown turn is retained and detached."""
         own = next((row["status"] for row in self.pending if row["admission_id"] == admission), None)
         if own == "unknown":
-            self.completions[admission] = {"outcome": "unknown", "text": (
-                "Execution outcome is unknown. Resume this session to inspect the lost turn "
-                "and /discard it; do not resend the input.")}
+            self.completions[admission] = {"outcome": "unknown", "text": _UNKNOWN_OUTCOME}
             return None
         if own == "queued" and self.unknown_admissions():
             return ("Unknown execution blocks this session; your accepted input is retained. "
                     "Resume interactively to resolve the lost turn; do not resend the input.")
-        if self.prompts:
+        if self.prompts and (own or not self.gaps):  # a control blocks only a turn still pending
             return "Input required; detached without cancelling. Resume this session interactively."
         return None
 
@@ -432,7 +503,7 @@ class GatewayChatView:
                 admission = receipt["admission_id"]
                 self.finite_admission = admission
                 for kind, event_admission, payload in self._finite_events:
-                    if event_admission == admission:
+                    if event_admission == admission and admission not in self.completions:
                         self._dispatch_event(kind, event_admission, payload)
                 self._finite_events.clear()
                 self.changed.set()
@@ -453,7 +524,8 @@ class GatewayChatView:
                 # in-process exit contracts: `-z` 0/2/130 (1 = no text), `chat -q`/`-Q` 0/1/130.
                 from hermes_cli.oneshot import _oneshot_exit_code
                 from hermes_cli.turn_exit import turn_exit_code
-                result = await self._settled_result(admission)
+                recovered = terminal.get("_receipt")
+                result = dict(recovered.get("result") or {}) if recovered else await self._settled_result(admission)
                 result.update(failed=bool(result.get("failed")) or outcome not in ("completed", "cancelled"),
                               interrupted=bool(result.get("interrupted")) or outcome == "cancelled")
                 exit_code = (_oneshot_exit_code(text, result) if self.unattended
@@ -461,7 +533,8 @@ class GatewayChatView:
                 if self.usage_file:
                     self._write_usage_file(result, outcome)
                 if self.emitter is not None:
-                    return self.emitter.emit_result({**result, "final_response": text},
+                    return self.emitter.emit_result({**result, "final_response": text,
+                                                     "replay_gap": bool(terminal.get("replay_gap"))},
                                                     session_id=self.session_id, exit_code=exit_code)
                 self._report_silent_interrupt(text, result)
                 print(text, flush=True)

hermes_cli/gateway_connection.py

@@ -61,6 +61,9 @@ class SessionState:
         self.request_listeners: list = []
         # A gateway-admitted notification turn the idle CLI queued a header for (gateway_tools_notify).
         self.admitted_turn: dict | None = None
+        # ``session.replay_gap``s seen, and how many a re-attach covered: the owner retired this viewer's
+        # subscription on overflow, so its frames stop until ``GatewayConnection.reattach``.
+        self.gaps = self.attached_gaps = 0
 
     def adopt_snapshot(self, snapshot):
         self.info = dict(snapshot.get("info") or {})
@@ -87,6 +90,8 @@ class SessionState:
         elif kind == "request.cancel" and isinstance(payload, dict):
             for listener in list(self.request_listeners):
                 listener(payload)
+        elif kind == "session.replay_gap":
+            self.gaps += 1
 
     def apply_usage(self, usage):
         """One session-cumulative usage dict (``session.usage`` payload or ``session.stats`` result)."""
@@ -309,8 +314,17 @@ class GatewayConnection:
             return False
         return True
 
-    def call(self, method, *, timeout=None, **params):
-        return self.call_tagged(method, timeout=timeout, **params)[0]
+    def call(self, method, *, timeout=None, abandon=None, **params):
+        return self.call_tagged(method, timeout=timeout, abandon=abandon, **params)[0]
+
+    def reattach(self, session_id, abandon=None):
+        """``session.resume`` after a replay gap; its snapshot names the subscription to detach."""
+        state = self.state(session_id)
+        gaps = state.gaps
+        snapshot = self.call("session.resume", session_id=session_id, abandon=abandon)
+        state.adopt_snapshot(snapshot)
+        state.attached_gaps = gaps  # a gap during this resume (or a wait given up) still owes one
+        return snapshot
 
     def _send(self, method, budget, params):
         """``(future, epoch)`` of one request sent on the live socket (see :meth:`call_tagged`)."""
@@ -383,13 +397,21 @@ class GatewayConnection:
             wake.set()  # ends the watcher above
             await asyncio.gather(stop, return_exceptions=True)
 
-    def call_tagged(self, method, *, timeout=None, **params):
+    def call_tagged(self, method, *, timeout=None, abandon=None, **params):
         """``(result, epoch)``: the epoch of the socket that carried the request, so work it admitted
-        is tied to that gateway life even if another caller redials before this one reads ``epoch``."""
+        is tied to that gateway life even if another caller redials before this one reads ``epoch``.
+        ``abandon`` (polled) true stops the wait with :class:`CallStopped`; the answer is dropped."""
+        from concurrent.futures import wait
         budget = DEFAULT_RPC_TIMEOUT if timeout is None else timeout
+        if abandon is not None and abandon():
+            raise CallStopped(method)
         future, epoch = self._send(method, budget, params)
+        deadline = time.monotonic() + budget + 5
         try:
-            return future.result(budget + 5), epoch
+            while abandon is not None and not wait([future], 0.1).done and time.monotonic() < deadline:
+                if abandon():
+                    raise CallStopped(method)
+            return future.result(max(deadline - time.monotonic(), 0)), epoch
         except TimeoutError as exc:
             future.cancel()
             raise GatewayClientError(f"Gateway did not answer {method}") from exc

hermes_cli/gateway_replay_gap.py

@@ -0,0 +1,32 @@
+"""After ``session.replay_gap``: re-attach, settle once, never splice the reply across the hole.
+
+On overflow the owner retires a viewer's subscription and sends one ``session.replay_gap``; the turns run
+on. The viewer re-attaches with ``session.resume``. Its snapshot is taken under the event-stream lock, so
+an admission it still lists publishes its completion on the new subscription, while one it no longer
+lists ended in the hole and only its receipt holds the reply. Live text stops at the gap.
+"""
+from __future__ import annotations
+
+RESUMED = "[output resumed]"
+
+
+def pending_status(pending, admission):
+    """``queued``/``started`` (its completion follows live), or None (it ended in the hole)."""
+    return next((row.get("status") for row in pending or () if row.get("admission_id") == admission), None)
+
+
+def settled(receipt):
+    """True: terminal with its reply (cancelled/failed need none); False: still running; None: unreadable."""
+    if receipt.get("status") != "terminal":
+        return False
+    return True if receipt.get("outcome") in {"cancelled", "interrupted", "failed"} or "result" in receipt else None
+
+
+def owed_reply(shown, final):
+    """What is owed of ``final`` after ``shown`` of its segment was painted: the rest of an exact prefix
+    (ignoring the owner's leading ``"\\n\\n"`` and trailing whitespace), else a marker and the whole."""
+    head = shown.removeprefix("\n\n")
+    for prefix in (head, head.rstrip()):
+        if final.startswith(prefix):
+            return final[len(prefix):]
+    return f"\n{RESUMED}\n{final}"

hermes_cli/gateway_session_store.py

@@ -290,6 +290,8 @@ class GatewaySessionStore:
             try:
                 # Hung up: redial first, so the detach names the subscription the reconnect resumed.
                 self.connection.start()
+                if state.gaps != state.attached_gaps:  # a replay gap retired it: re-attach, so the detach lands
+                    self.connection.reattach(session_id)
             except GatewayClientError as exc:
                 logger.debug("redial for session.detach failed: %s", exc)
                 reachable = False

hermes_cli/stream_json.py

@@ -96,6 +96,9 @@ class StreamJsonEmitter:
                    "duration_ms": int((time.time() - self._start) * 1000)}
         if data.get("error"):
             payload["error"] = str(data["error"])
+        if data.get("replay_gap"):
+            # The viewer fell behind and re-attached: ``text`` events stopped there; ``text`` is the whole reply.
+            payload["replay_gap"] = True
         self._emit(payload)
         print(f"\nsession_id: {session_id or self._session_id}", file=sys.stderr)  # same stderr contract as -Q
         return exit_code

tests/acp_adapter/test_gateway_replay_gap.py

@@ -0,0 +1,205 @@
+"""ACP across a ``session.replay_gap``: the pump re-attaches that session and survives; a cut prompt settles once,
+live or from one receipt read the pump owns; terminal evidence beats a recovery error; an interim stays whole;
+a reused reply is hidden after a gap or a receipt only when all of it was shown. (Refused re-attach:
+test_gateway_terminal_projection.)"""
+import asyncio
+from unittest.mock import AsyncMock
+
+import pytest
+import pytest_asyncio
+from acp.schema import TextContentBlock
+
+from acp_adapter.gateway_server import GatewayACPAgent
+from hermes_cli.gateway_client import GatewayClientError
+from tests.gateway.fixtures.replay_gap_owner import FINAL, Owner, frame_cap  # frame_cap: autouse
+from tests.waiting import async_wait_until as until
+
+def resumed(*pending):
+    return {"replay_epoch": "e2", "last_sequence": 0, "pending": list(pending), "prompts": [], "resumed": True}
+
+
+def shown(agent, session="s"):
+    return "".join(c.kwargs["update"].content.text for c in agent._conn.session_update.await_args_list
+                   if c.kwargs["session_id"] == session and hasattr(getattr(c.kwargs["update"], "content", None), "text"))
+
+
+def frame(kind, payload, *, session="s", admission=None, seq=None):
+    params = {"type": kind, "session_id": session, "payload": payload, "admission_id": admission}
+    return {"method": "event", "params": {**params, **({"replay_epoch": "e", "seq": seq} if seq else {})}}
+
+
+def terminal(admission, reply, **result):
+    return {"admission_id": admission, "status": "terminal", "outcome": "completed",
+            "result": {"final_response": reply, "completed": True, **result}}
+
+
+@pytest_asyncio.fixture
+async def rig():
+    """``rig(answer)``: an agent whose gateway ``rpc`` is ``answer``, with its pump running; torn down after."""
+    agents = []
+
+    def make(answer):
+        agent = GatewayACPAgent()
+        agent._snapshots.update(s={"replay_epoch": "e", "last_sequence": 0}, t={"replay_epoch": "e", "last_sequence": 0})
+        agent._gateway, agent._conn, agent.asked = AsyncMock(events=asyncio.Queue()), AsyncMock(), []
+        agent._gateway.rpc.side_effect = answer
+        agent._event_task = asyncio.create_task(agent._events())
+        agents.append(agent)
+        return agent
+    yield make
+    tasks = [task for agent in agents for task in (*agent.asked, agent._event_task)]
+    for task in tasks:
+        task.cancel()
+    await asyncio.gather(*tasks, return_exceptions=True)
+
+
+def ask(agent, session="s"):
+    agent.asked.append(asyncio.create_task(agent.prompt([TextContentBlock(type="text", text="q")], session)))
+    return agent.asked[-1]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("mode", ["ended", "live"])
+async def test_pump_survives_and_the_editor_sees_the_reply_once(tmp_path, mode):
+    owner = Owner(tmp_path / "state.db", pause=mode)
+    agent = GatewayACPAgent()
+    agent._gateway, agent._conn, agent.asked = owner, AsyncMock(), []
+    agent._snapshots["s"] = await owner.conn.resume(owner.ref, {})
+    pump = agent._event_task = asyncio.create_task(agent._events())
+    try:
+        response = await asyncio.wait_for(ask(agent), 30)
+        await asyncio.to_thread(owner.turns[-1].join, 10)
+        await until(lambda: owner.events.empty())
+        assert owner.gaps, "probe did not overflow the fanout (vacuous)"
+        assert response.stop_reason == "end_turn" and not pump.done() and agent._failure is None
+        assert shown(agent) == FINAL, (len(shown(agent)), len(FINAL))
+    finally:
+        for task in (*agent.asked, pump):
+            task.cancel()
+        await asyncio.gather(*agent.asked, pump, return_exceptions=True)
+        await asyncio.to_thread(owner.finish)
+
+
+@pytest.mark.asyncio
+async def test_completion_in_delivery_beats_a_recovery_error(rig):
+    delivering, release = asyncio.Event(), asyncio.Event()
+
+    async def answer(method, **params):
+        if method == "prompt.submit":
+            agent._gateway.events.put_nowait(frame("session.replay_gap", {}))
+            await until(lambda: agent._snapshots["s"].get("resumed"))
+            # Admitted after the snapshot: its completion is live, and in delivery when the ack lands.
+            agent._gateway.events.put_nowait(frame("message.complete", {"text": "done", "outcome": "completed"},
+                                                   admission="a", seq=1))
+            await delivering.wait()
+            return {"admission_id": "a", "status": "queued"}
+        if method == "session.resume":
+            return resumed()
+        raise GatewayClientError("read_failed")
+
+    agent = rig(answer)
+
+    async def update(**kwargs):
+        delivering.set()
+        await release.wait()
+
+    agent._conn.session_update.side_effect = update
+    task = ask(agent)
+    await asyncio.wait_for(delivering.wait(), 5)
+    await until(lambda: task.done() or agent._admissions.get("s") == "a")
+    release.set()
+    assert (await asyncio.wait_for(task, 2)).stop_reason == "end_turn"
+
+
+@pytest.mark.asyncio
+async def test_one_recovery_read_settles_a_cancellation_and_the_pump_survives(rig):
+    """Acked during the re-attach, a turn that ended in the hole is read once (the pump owns its recovery) and
+    that fetched cancellation is final (a second read never answers). The other session streams on."""
+    attaching, reads = asyncio.Event(), []
+
+    async def answer(method, **params):
+        if method == "prompt.submit" and params["session_id"] == "t":
+            return {"admission_id": "t-a", "status": "queued"}
+        if method == "prompt.submit":
+            agent._gateway.events.put_nowait(frame("session.replay_gap", {}))
+            await attaching.wait()
+            return {"admission_id": "a", "status": "queued"}
+        if method == "session.resume":
+            attaching.set()
+            await until(lambda: agent._admissions.get("s") == "a")
+            return resumed()
+        reads.append(params["admission_id"])
+        if len(reads) > 1:
+            await asyncio.Event().wait()
+        return {"admission_id": "a", "status": "terminal", "outcome": "cancelled"}
+
+    agent = rig(answer)
+    other = ask(agent, "t")
+    await until(lambda: agent._admissions.get("t") == "t-a")
+    first = ask(agent)
+    assert (await asyncio.wait_for(asyncio.shield(first), 2)).stop_reason == "cancelled"
+    agent._gateway.events.put_nowait(frame("message.complete", {"text": "other", "outcome": "completed"},
+                                           session="t", admission="t-a", seq=1))
+    assert (await asyncio.wait_for(other, 2)).stop_reason == "end_turn"
+    assert reads == ["a"] and agent._failure is None
+
+
+@pytest.mark.asyncio
+async def test_retry_after_a_lost_ack_completes_a_partial_reused_reply(rig):
+    submissions = []
+
+    async def answer(method, **params):
+        if method == "prompt.submit":
+            submissions.append(params["input_id"])
+            if len(submissions) == 1:
+                agent._gateway.events.put_nowait(frame("message.delta", {"text": "Hel"}, admission="a", seq=1))
+                agent._gateway.events.put_nowait(frame("session.replay_gap", {}))
+                raise GatewayClientError("submit timed out")
+            return {"admission_id": "a", "status": "terminal", "outcome": "completed"}
+        return resumed() if method == "session.resume" else terminal("a", "Hello", response_reused=True)
+
+    agent = rig(answer)
+    with pytest.raises(GatewayClientError):
+        await ask(agent)
+    await until(lambda: agent._snapshots["s"].get("resumed"))
+    assert (await asyncio.wait_for(ask(agent), 2)).stop_reason == "end_turn"
+    assert submissions[0] == submissions[1] and shown(agent) == "Hello"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("completion", ["", "ANSWER"])
+async def test_interim_after_the_gap_shows_the_reply_once(rig, completion):
+    """An unstreamed interim may be the reply's only copy (a codex app-server final; the completion is empty)."""
+    async def answer(method, **params):
+        if method == "prompt.submit":
+            return {"admission_id": "a", "status": "queued"}
+        return resumed({"admission_id": "a", "status": "started"})
+
+    agent = rig(answer)
+    task = ask(agent)
+    await until(lambda: agent._admissions.get("s") == "a")
+    agent._gateway.events.put_nowait(frame("message.delta", {"text": "Let me "}, admission="a", seq=1))
+    agent._gateway.events.put_nowait(frame("session.replay_gap", {}))
+    await until(lambda: agent._snapshots["s"].get("resumed"))
+    for seq, (kind, text) in enumerate([("message.interim", "ANSWER"), ("message.complete", completion)], 1):
+        agent._gateway.events.put_nowait(frame(kind, {"text": text, "already_streamed": False,
+                                                      "outcome": "completed"}, admission="a", seq=seq))
+    assert (await asyncio.wait_for(task, 2)).stop_reason == "end_turn"
+    assert shown(agent).count("ANSWER") == 1, shown(agent)
+
+
+@pytest.mark.asyncio
+async def test_without_a_gap_a_streamed_reused_reply_is_not_repainted(rig):
+    """The owner may append to a reused final (auto media tags); without a gap the streamed copy stays the one."""
+    agent = rig(AsyncMock(return_value={"admission_id": "a", "status": "queued"}))
+    task = ask(agent)
+    await until(lambda: agent._admissions.get("s") == "a")
+    for seq, (kind, payload) in enumerate([
+            ("message.delta", {"text": "FINAL"}), ("message.delta", {"text": None}),
+            ("tool.start", {"tool_call_id": "t1", "tool_name": "memory", "args": {}}),
+            ("tool.complete", {"tool_call_id": "t1", "tool_name": "memory", "result": "ok"}),
+            ("message.complete", {"text": "FINAL\nMEDIA:/tmp/x.png", "outcome": "completed",
+                                  "response_reused": True})], 1):
+        agent._gateway.events.put_nowait(frame(kind, payload, admission="a", seq=seq))
+    assert (await asyncio.wait_for(task, 2)).stop_reason == "end_turn"
+    assert shown(agent).count("FINAL") == 1, shown(agent)

tests/acp_adapter/test_gateway_terminal_projection.py

@@ -45,13 +45,18 @@ async def test_cancelled_admission_suppresses_only_interrupt_metadata(text, visi
 
 
 @pytest.mark.asyncio
-async def test_replay_gap_is_a_recoverable_projection_failure():
+async def test_refused_reattach_after_a_replay_gap_still_fails_the_pump():
+    """A gap re-attaches that session (test_gateway_replay_gap); a refused re-attach ends the wait as
+    before: the pump fails, the next call redials."""
+    from hermes_cli.gateway_client import GatewayRPCError
     agent = agent_with_admission()
-    with pytest.raises(GatewayClientError, match='session_replay_gap'):
-        await agent._project({
-            'session_id': 's', 'type': 'session.replay_gap',
-            'payload': {'reason': 'subscriber_overflow'},
-        })
+    agent._snapshots['s'] = {'replay_epoch': 'e', 'last_sequence': 9}
+    agent._gateway.events = asyncio.Queue()
+    agent._gateway.events.put_nowait({'method': 'event', 'params': {
+        'session_id': 's', 'type': 'session.replay_gap', 'payload': {'reason': 'subscriber_overflow'}}})
+    agent._gateway.rpc.side_effect = GatewayRPCError('not_found')
+    await asyncio.wait_for(agent._events(), 2)
+    assert isinstance(agent._failure, GatewayRPCError) and str(agent._failure) == 'not_found'
 
 
 @pytest.mark.asyncio

tests/gateway/fixtures/replay_gap_owner.py

@@ -0,0 +1,86 @@
+"""A real owner (SessionAuthority, AuthorityConnection, FanoutTransport overflow, the drain's publish sequence)
+for one session ``s`` and one viewer whose socket stops reading while paused (Ctrl-Z). #140292 raised the frame
+cap to 16384 and kept the overflow path; ``frame_cap`` pins the old 256, as its own overflow probe does."""
+import asyncio
+import threading
+import time
+from types import SimpleNamespace
+
+import pytest
+
+DELTAS = [f"{i:07d}|" for i in range(600)]
+FINAL = "".join(DELTAS)
+
+
+@pytest.fixture(autouse=True)
+def frame_cap(monkeypatch):
+    from tui_gateway.transport import FanoutTransport
+    monkeypatch.setattr(FanoutTransport, "_MAX_PENDING_FRAMES", 256)
+
+
+class Owner:
+    """``pause``: the next turn pauses the viewer throughout (``"ended"``) or until delta 400 (``"live"``)."""
+
+    def __init__(self, db_path, pause="ended"):
+        from gateway.session_authority import LiveSession, SessionAuthority
+        from gateway.session_contract import SessionRef
+        from gateway.session_controls import AuthorityConnection
+        from hermes_state import SessionDB
+        from hermes_state_runtime import begin_runtime_epoch
+        self.loop, self.events, self.running = asyncio.get_running_loop(), asyncio.Queue(), threading.Event()
+        self.pause, self.gaps, self.turns, self.ref = pause, [], [], SessionRef("owned", "s")
+        self.running.set()
+        self.db = SessionDB(db_path=db_path)
+        self.db.create_session("s", source="test")
+        self.epoch = begin_runtime_epoch(self.db, instance_id="owner")
+        self.authority = SessionAuthority(SimpleNamespace(_draining=False), profile_id="owned",
+                                          instance_id="owner", db=self.db, epoch=self.epoch)
+        self.live = self.authority.sessions["s"] = LiveSession(None, "route")
+        retire = self.authority._retire_overflowed
+        self.authority._retire_overflowed = lambda sid, peer: (self.gaps.append(sid), retire(sid, peer))
+        self.conn = AuthorityConnection(self.authority, self, {"user_id": "viewer"})
+
+    def write(self, frame):  # the viewer's socket
+        self.running.wait(30)
+        self.loop.call_soon_threadsafe(self.events.put_nowait, frame)
+        return True
+
+    def close(self):
+        return None
+
+    async def rpc(self, method, _timeout=None, **params):
+        from hermes_state_runtime import admit_session_input, claim_session_input
+        if method == "prompt.submit":
+            admit_session_input(self.db, epoch=self.epoch, principal_id=self.conn.actor.subject, session_id="s",
+                                request_id=params.get("input_id") or params.get("submission_id"),
+                                payload={"text": params["text"]})
+            row = claim_session_input(self.db, epoch=self.epoch, session_id="s")
+            self.turns.append(threading.Thread(target=self._turn, args=(row, self.pause), daemon=True))
+            self.pause = None
+            self.turns[-1].start()
+            return {"admission_id": row["admission_id"], "status": "queued"}
+        verb = {"session.resume": "resume", "prompt.receipt": "receipt", "session.detach": "detach"}.get(method)
+        return await getattr(self.conn, verb)(self.ref, {} if verb == "resume" else params) if verb else {}
+
+    def _turn(self, row, pause):
+        from gateway.session_settlement_recovery import commit_and_publish
+        self.authority._stamp_claim(self.ref, self.live, row)
+        if pause:
+            self.running.clear()  # Ctrl-Z mid-turn
+        for index, delta in enumerate(DELTAS):
+            if pause == "live" and index == 400:
+                self.running.set()  # fg mid-turn
+                deadline = time.monotonic() + 5
+                while not self.live.subscribers and time.monotonic() < deadline:
+                    time.sleep(0.01)  # health: allow HX013 -- the owner thread waits for the re-attach
+            self.authority.publish_execution("s", row["generation"], "message.delta", {"text": delta})
+        with self.live.event_stream.lock:
+            self.authority.executing.pop("s", None)
+        commit_and_publish(self.authority, self.live, self.ref, row, FINAL, "completed", None, {})
+        self.running.set()
+
+    def finish(self):
+        self.running.set()
+        for turn in self.turns:
+            turn.join(10)
+        self.db.close()

tests/hermes_cli/test_gateway_agent_replay_gap.py

@@ -0,0 +1,128 @@
+"""Classic CLI (`hermes`, `hermes chat` -> GatewayAgent) across a ``session.replay_gap``: it re-attaches, shows
+the settled reply once and never leaves a later turn waiting on a retired subscription. A recovery belongs
+to its turn: a session switch, a later turn or Ctrl-C's settle bound ends it before it sends or paints."""
+import asyncio
+import contextlib
+import threading
+
+import pytest
+
+from tests.gateway.fixtures.replay_gap_owner import FINAL, Owner, frame_cap  # frame_cap: autouse
+from tests.hermes_cli.test_gateway_agent import SID, FakeClient, connect_with, frame
+
+GAP = frame("session.replay_gap", {}, admission=None)
+
+
+def run_turn(agent, text, timeout):
+    out = {}
+    worker = threading.Thread(target=lambda: out.setdefault("result", agent.run_conversation(text)), daemon=True)
+    worker.start()
+    worker.join(timeout)
+    return out.get("result")
+
+
+def test_turn_ended_in_the_hole_reports_the_answer_and_the_next_turn_streams(tmp_path):
+    from hermes_cli.gateway_agent import GatewayAgent
+    from hermes_cli.gateway_connection import GatewayConnection
+    owners, streamed = [], []
+
+    @contextlib.asynccontextmanager
+    async def connect():
+        owners.append(Owner(tmp_path / "state.db"))
+        yield owners[-1]
+
+    connection = GatewayConnection(connect=connect)
+    connection.start(timeout=30)
+    try:
+        connection.state("s").adopt_snapshot(connection.call("session.resume", session_id="s"))
+        agent = GatewayAgent(connection=connection, session_id="s")
+        agent.stream_delta_callback = streamed.append
+        first = run_turn(agent, "write a long answer", 30)
+        assert owners[0].gaps, "probe did not overflow the fanout (vacuous)"
+        assert first is not None and not first["failed"] and first["final_response"] == FINAL, first
+        assert "".join(t for t in streamed if t) == FINAL, "hole or duplicate"
+        streamed.clear()
+        second = run_turn(agent, "and a short one", 15)
+        assert second is not None, "turn 2 hangs: the gap detached the viewer and nothing re-attached"
+        assert not second["failed"] and "".join(t for t in streamed if t) == FINAL
+    finally:
+        owners[0].finish()
+        connection.close()
+
+
+@pytest.fixture
+def agents():
+    from hermes_cli.gateway_agent import GatewayAgent
+    from hermes_cli.gateway_connection import GatewayConnection
+    made = []
+
+    def build(script, **kw):
+        client = FakeClient(script, result=kw.pop("result", None))
+        made.append(GatewayConnection(connect=connect_with(client)))
+        made[-1].start(5)
+        made[-1].state(SID)  # gaps are counted per attached session
+        return GatewayAgent(connection=made[-1], session_id=SID, **kw), client
+    yield build
+    for connection in made:
+        connection.close()
+
+
+@pytest.mark.parametrize("change", ["session", "interrupt"])
+def test_idle_gap_reattach_cannot_submit_abandoned_input(agents, monkeypatch, change):
+    entered, release = threading.Event(), threading.Event()
+    agent, client = agents([])
+    agent.connection.state(SID).gaps = 1  # a gap between turns retired the subscription
+    original = client.rpc
+
+    async def rpc(method, **params):
+        if method == "session.resume" and params["session_id"] == SID:
+            entered.set()
+            await asyncio.to_thread(release.wait, 5)
+            return {"pending": []}
+        return await original(method, **params)
+
+    monkeypatch.setattr(client, "rpc", rpc)
+    worker = threading.Thread(target=agent.run_conversation, args=("old question",), daemon=True)
+    worker.start()
+    try:
+        assert entered.wait(2)
+        if change == "session":
+            agent.session_id = "new-session"
+            agent.reset_session_state()
+        else:
+            monkeypatch.setattr("hermes_cli.gateway_agent.INTERRUPT_SETTLE_SECONDS", 0)
+            agent.interrupt()
+    finally:
+        release.set()
+        worker.join(2)
+    assert not worker.is_alive()
+    assert not [params for method, params in client.calls if method == "prompt.submit"]
+
+
+def test_tool_start_closes_a_prefix_whose_separator_fell_in_the_hole(agents, monkeypatch):
+    output, final = [], "The answer is 42."
+    agent, client = agents([frame("message.delta", {"text": "The answer"}), GAP],
+                           result={"final_response": final, "completed": True},
+                           stream_delta_callback=output.append,
+                           tool_progress_callback=lambda *a, **k: output.append("<tool>"))
+    original = client.rpc
+
+    async def rpc(method, **params):
+        if method == "session.resume":
+            await client.events.put(frame("tool.start", {"name": "read_file", "tool_id": "t", "args": {}}))
+            await client.events.put(frame("message.complete", {"text": final, "outcome": "completed"}))
+            return {"pending": [{"admission_id": "adm-1", "status": "started"}]}
+        return await original(method, **params)
+
+    monkeypatch.setattr(client, "rpc", rpc)
+    assert run_turn(agent, "q", 5) is not None
+    assert "".join(x for x in output[output.index("<tool>") + 1:] if isinstance(x, str)) == final
+
+
+def test_gap_completes_the_voice_stream_without_terminal_streaming(agents):
+    spoken, final = [], "First sentence. Second sentence."
+    agent, _ = agents([frame("message.delta", {"text": "First sentence. "}), GAP],
+                      result={"final_response": final, "completed": True})
+    result = agent.run_conversation("q", stream_callback=spoken.append)
+    assert not result["failed"] and result["final_response"] == final
+    assert "".join(spoken) == final

tests/hermes_cli/test_gateway_chat_view_replay_gap.py

@@ -0,0 +1,137 @@
+"""GatewayChatView (`chat -q`, stream-json, `-z`, the thin chat) across a ``session.replay_gap``: it re-attaches;
+a one-shot prints the reply once; stream-json emits no ``text`` after the gap and its ``result`` carries the
+whole reply with ``replay_gap: true``. Adoption wakes a waiting one-shot; a control blocks only an admission
+the snapshot still lists; an ack received mid-resume settles against that snapshot; ``unknown`` settles."""
+import asyncio
+import json
+
+import pytest
+
+from tests.gateway.fixtures.replay_gap_owner import FINAL, Owner, frame_cap  # frame_cap: autouse
+from tests.waiting import async_wait_until as until
+
+GAP = {"method": "event", "params": {"type": "session.replay_gap", "session_id": "s", "payload": {}}}
+
+
+async def real_oneshot(tmp_path, mode, **view_kwargs):
+    from hermes_cli.gateway_chat_view import GatewayChatView
+    owner = Owner(tmp_path / "state.db", pause=mode)
+    try:
+        view = GatewayChatView(owner, await owner.conn.resume(owner.ref, {}), **view_kwargs)
+        code = await asyncio.wait_for(view.run("write a long answer", oneshot=True), 30)
+    finally:
+        await asyncio.to_thread(owner.finish)
+    assert owner.gaps, "probe did not overflow the fanout (vacuous)"
+    return code
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("mode", ["ended", "live"])
+async def test_stream_json_result_is_authoritative_and_no_text_follows_the_gap(tmp_path, capsys, mode):
+    from hermes_cli.stream_json import StreamJsonEmitter
+    assert await real_oneshot(tmp_path, mode, emitter=StreamJsonEmitter(model="m", session_id="s")) == 0
+    records = [json.loads(line) for line in capsys.readouterr().out.splitlines() if line.strip()]
+    text = "".join(r.get("text", "") for r in records if r["type"] == "text")
+    assert FINAL.startswith(text) and len(text) < len(FINAL), "text events after the gap (a hole or a splice)"
+    assert records[-1]["type"] == "result" and records[-1]["text"] == FINAL and records[-1]["replay_gap"] is True
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("mode", ["ended", "live"])
+async def test_quiet_oneshot_prints_the_answer_once(tmp_path, capsys, mode):
+    assert await real_oneshot(tmp_path, mode, quiet=True) == 0
+    assert capsys.readouterr().out.count(FINAL) == 1
+
+
+class Peer:
+    def __init__(self, answer):
+        self.events, self.answer, self.acked = asyncio.Queue(), answer, False
+
+    async def rpc(self, method, **params):
+        result = await self.answer(method, params)
+        self.acked |= method == "prompt.submit"
+        return result
+
+
+async def gapped(answer, run):
+    """``run(view)`` on a view of ``s`` whose admission ``a`` is acked, then cut by a gap."""
+    from hermes_cli.gateway_chat_view import GatewayChatView
+    view = GatewayChatView(Peer(answer), {"stored_session_id": "s"})
+    task = asyncio.create_task(run(view))
+    try:
+        await until(lambda: view.finite_admission or (view.client.acked and not view.finite) or task.done())
+        view.client.events.put_nowait(GAP)
+        return view, await asyncio.wait_for(asyncio.shield(task), 2)
+    finally:
+        task.cancel()
+        await asyncio.gather(task, return_exceptions=True)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("own, code", [("queued", 3), (None, 0)])
+async def test_adopted_approval_ends_the_oneshot_only_while_the_snapshot_lists_it(own, code):
+    """Queued behind another turn's approval: adoption wakes the one-shot, which detaches (3). Ended in the hole
+    while a successor waits for approval: its own receipt settles it (0)."""
+    async def answer(method, params):
+        if method == "prompt.submit":
+            return {"admission_id": "a", "status": "queued"}
+        if method == "session.resume":
+            return {"pending": [{"admission_id": "b", "status": "started"}, *([{"admission_id": "a", "status": own}]
+                                                                             if own else [])],
+                    "prompts": [{"prompt_id": "p", "kind": "approval", "execution_generation": 8}]}
+        return {"admission_id": "a", "status": "terminal", "outcome": "completed",
+                "result": {"final_response": "DONE", "completed": True}}
+    view, result = await gapped(answer, lambda view: view.run("q", oneshot=True))
+    assert result == code and (own or view.completions["a"]["text"] == "DONE")
+
+
+@pytest.mark.asyncio
+async def test_unknown_row_settles_the_interactive_turn(capsys):
+    async def answer(method, params):
+        if method == "prompt.submit":
+            return {"admission_id": "a", "status": "queued"}
+        return {"pending": [{"admission_id": "a", "status": "unknown"}], "prompts": []}
+
+    async def interactive(view):
+        renderer = asyncio.create_task(view.render())
+        await view.submit("q")
+        await until(lambda: "a" in view.completions or view.failure is not None)
+        renderer.cancel()
+        return view.failure
+    view, failure = await gapped(answer, interactive)
+    assert failure is None and view.completions["a"]["outcome"] == "unknown"
+    assert "outcome is unknown" in capsys.readouterr().out
+
+
+@pytest.mark.asyncio
+async def test_ack_during_the_reattach_settles_against_its_snapshot(capsys):
+    from hermes_cli.gateway_chat_view import GatewayChatView
+    entered, release = asyncio.Event(), asyncio.Event()
+
+    def event(kind, payload):
+        return {"method": "event", "params": {"type": kind, "session_id": "s", "admission_id": "a", "payload": payload}}
+
+    async def answer(method, params):
+        if method == "prompt.submit":
+            peer.events.put_nowait(GAP)
+            await entered.wait()
+            return {"admission_id": "a"}
+        entered.set()  # the snapshot is captured while a runs; its completion follows live
+        peer.events.put_nowait(event("message.delta", {"text": "Hello"}))
+        peer.events.put_nowait(event("message.complete", {"text": "Hello", "outcome": "completed"}))
+        await release.wait()
+        return {"pending": [{"admission_id": "a", "status": "started"}], "prompts": []}
+
+    peer = Peer(answer)
+    view = GatewayChatView(peer, {"stored_session_id": "s"})
+    renderer = asyncio.create_task(view.render())
+    try:
+        submitted = asyncio.create_task(view.submit("q"))
+        await asyncio.wait_for(entered.wait(), 5)
+        release.set()
+        await asyncio.wait_for(submitted, 5)
+        await until(lambda: peer.events.empty() and "a" in view.completions)
+    finally:
+        renderer.cancel()
+        await asyncio.gather(renderer, return_exceptions=True)
+    assert capsys.readouterr().out == "Hello\n"

website/docs/reference/cli-commands.md

@@ -181,7 +181,7 @@ Every event carries `timestamp` (Unix epoch milliseconds).
 | `text` | `text` — a streamed assistant text delta |
 | `tool_use` | `name`; `input` when the tool arguments are available |
 | `tool_result` | `name`, `output` (capped at 5000 chars), `duration_ms`, `is_error` |
-| `result` | `session_id`, `exit_code`, `text`, `tokens` (`input`, `output`, `total`, `cache_read`, `cache_write`), `duration_ms`; `error` when the turn failed |
+| `result` | `session_id`, `exit_code`, `text`, `tokens` (`input`, `output`, `total`, `cache_read`, `cache_write`), `duration_ms`; `error` when the turn failed; `replay_gap: true` when the run fell behind the gateway's stream and re-attached, so its `text` events stopped early (`text` here is still the whole reply) |
 
 Once a conversation starts, its terminal record is always `result` — including
 `exit_code: 130` when it is interrupted with Ctrl-C. Treat that record as the