Gateway follow-ups queued during a busy/compression window no longer vanish; overflow survives shutdown (#99882, salvage #99912)

gateway/run.py

@@ -9765,6 +9765,72 @@ def _queue_depth(self, session_key: str, *, adapter: Any = None) -> int:
             depth += 1
         return depth
 
+    def _rescue_orphaned_overflow(
+        self, session_key: str, adapter: Any
+    ) -> Optional["MessageEvent"]:
+        """Pop the oldest orphaned FIFO overflow event for an idle session (#99882).
+
+        The FIFO overflow (``queued_events``) drains only at the post-turn
+        promotion site (``_promote_queued_event`` inside the ``_run_agent``
+        drain).  When a busy window ends without that drain running — the
+        #99882 shape: a follow-up queued during compression-in-flight lands
+        in overflow, compression finishes, the slot event's turn runs, but
+        the drain recursion exits before promoting (or the busy window ends
+        through an exception / interrupt / generation-bump exit that never
+        reaches the promotion site) — the overflow entries are silently
+        orphaned: never dispatched, never persisted, never logged.
+
+        This rescue runs at the point where a NEW event arrives for a
+        session that is NOT busy (the idle entry in
+        ``_process_message_priority``).  If the session went idle with a
+        populated overflow, the oldest orphan is returned so the caller runs
+        it as THIS turn, and the next orphan (if any) is staged into the
+        slot so the post-turn drain continues the chain in arrival order
+        (#28503).  The caller then enqueues the incoming event behind the
+        chain via ``_enqueue_fifo``.
+
+        The returned event is REMOVED from both stores: leaving it in the
+        slot while it also runs as the current turn would make the post-turn
+        ``_dequeue_pending_event`` run it a second time.
+
+        Returns the orphaned event to run now, or ``None`` when there is
+        nothing to rescue (no overflow, slot occupied, or no slot storage).
+        """
+        try:
+            _q_state = self._peek_session_state(session_key)
+            overflow = _q_state.conversation.queued_events if _q_state else None
+            if not overflow:
+                return None
+            pending_slot = getattr(adapter, "_pending_messages", None)
+            if not isinstance(pending_slot, dict) or pending_slot.get(session_key):
+                # Slot occupied (busy) or no slot storage — promotion owns
+                # this; do not fight it from the idle path.
+                return None
+            head = overflow.pop(0)
+            # Keep the slot occupied for the rest of the chain so the drain
+            # promotes in order and any mid-chain arrival routes to overflow
+            # instead of jumping the queue (same invariant as the drain's
+            # own _promote_queued_event).  Only ONE event fits the slot.
+            if overflow:
+                pending_slot[session_key] = overflow.pop(0)
+            logger.warning(
+                "Rescued orphaned FIFO overflow event for idle session "
+                "%s — it was queued during a busy window but the post-turn "
+                "drain never promoted it (#99882)",
+                session_key,
+            )
+            if overflow:
+                logger.warning(
+                    "%d overflow event(s) still queued for session %s after "
+                    "rescue staging (will drain via normal promotion)",
+                    len(overflow),
+                    session_key,
+                )
+            return head
+        except Exception:
+            logger.debug("FIFO overflow rescue failed for %s", session_key, exc_info=True)
+            return None
+
     @staticmethod
     def _is_goal_continuation_event(event_or_text: Any) -> bool:
         """Return True for synthetic /goal continuation turns.
@@ -16457,6 +16523,21 @@ def _phase_elapsed() -> float:
                 flush_pending_to_file(dict(self._pending_messages), reason="shutdown")
             except Exception:
                 pass
+            # The FIFO tail lives in SessionState.conversation.queued_events,
+            # not in the slot dict above — flush it too or every follow-up
+            # parked in overflow at restart time is lost (#99882).
+            try:
+                from gateway.shutdown_flush import flush_overflow_to_file
+                flush_overflow_to_file(
+                    {
+                        _k: list(_v)
+                        for _k, _v in dict(getattr(self, "_queued_events", None) or {}).items()
+                        if _v
+                    },
+                    reason="shutdown",
+                )
+            except Exception:
+                pass
             # On the real runner these are live SessionState views whose
             # clear() resets one field per session — never a wholesale dict
             # swap, so a concurrent writer on another session can't lose its
@@ -19712,6 +19793,50 @@ async def _do_undo():
                 _quick_key,
             )
             return _limit_message
+
+        # ── FIFO orphan rescue (#99882) ────────────────────────────────
+        # If this session went idle with a populated overflow (queued
+        # during a busy window whose post-turn drain never promoted —
+        # e.g. a compression-demoted follow-up after the compression
+        # window ended through an exit that skipped the promotion site),
+        # those events were silently orphaned.  We are starting the next
+        # turn for this session NOW: re-stage the orphans in FIFO order
+        # and enqueue the incoming event behind them, so arrival order
+        # (#28503) holds: oldest orphan runs as this turn, the rest drain
+        # in order, the new message last.  Skipped for control commands
+        # (/stop etc. own their own semantics) and internal events.
+        try:
+            _orphan_adapter = self._adapter_for_source(source)
+            if (
+                _orphan_adapter is not None
+                and not bool(getattr(event, "internal", False))
+                and not event.get_command()
+            ):
+                _rescued = self._rescue_orphaned_overflow(
+                    _quick_key, _orphan_adapter
+                )
+                if _rescued is not None:
+                    # The oldest orphan runs as THIS turn.  Park the
+                    # incoming event behind the rest of the chain: into the
+                    # slot when the chain was a single orphan (so the
+                    # post-turn drain picks it up), otherwise into overflow
+                    # behind the already-staged next orphan (FIFO).
+                    self._enqueue_fifo(_quick_key, event, _orphan_adapter)
+                    event = _rescued
+                    # Same session key by construction; carry the orphan's
+                    # own source so reply anchors / thread metadata point
+                    # at the message that is actually being answered.
+                    _rescued_source = getattr(_rescued, "source", None)
+                    if _rescued_source is not None:
+                        source = _rescued_source
+                    is_internal = bool(getattr(_rescued, "internal", False))
+        except Exception:
+            logger.debug(
+                "FIFO orphan rescue pre-claim failed for %s",
+                _quick_key,
+                exc_info=True,
+            )
+
         _claim_state = self._session_state(_quick_key)
         if _active_session_lease is not None:
             _claim_state.turn.lease = _active_session_lease

gateway/shutdown_flush.py

@@ -142,6 +142,66 @@ def flush_pending_to_file(
     return flushed
 
 
+def flush_overflow_to_file(
+    overflow_by_session: Dict[str, Any],
+    *,
+    reason: str = "shutdown",
+) -> int:
+    """Serialise the FIFO overflow tails (``queued_events``) to disk.
+
+    Sibling of :func:`flush_pending_to_file` for the second half of the
+    gateway FIFO (#99882): the adapter slot holds the queue head, and the
+    per-session ``SessionState.conversation.queued_events`` list holds the
+    tail.  Shutdown flushed only the slot, so every follow-up parked in
+    overflow at restart time vanished with the process.  Each overflow
+    event is written as its own payload in the same shape as a slot flush
+    so ``recover_pending_to_db`` replays them unchanged; a ``seq`` field
+    preserves arrival order within a session.
+
+    Returns the number of events flushed.
+    """
+    if not overflow_by_session:
+        return 0
+
+    flush_dir = _get_flush_dir()
+    ts = int(time.time())
+    flushed = 0
+
+    for session_key, events in list(overflow_by_session.items()):
+        if not session_key or not events:
+            continue
+        for seq, value in enumerate(list(events)):
+            if value is None:
+                continue
+            try:
+                serialised = _serialise_value(value)
+                if serialised is None:
+                    continue
+                _write_payload(
+                    flush_dir,
+                    {
+                        "session_key": session_key,
+                        "reason": reason,
+                        "ts": ts,
+                        "seq": seq,
+                        "data": serialised,
+                    },
+                )
+                flushed += 1
+            except Exception as exc:
+                logger.debug(
+                    "Failed to flush overflow message for %s: %s",
+                    session_key, exc,
+                )
+
+    if flushed:
+        logger.info(
+            "Flushed %d queued overflow message(s) to %s (reason=%s)",
+            flushed, flush_dir, reason,
+        )
+    return flushed
+
+
 # Reason tag for transcript messages dropped by the in-memory pending cap
 # during live operation (#78182). These payloads carry the full transcript
 # message dict so they can be replayed verbatim once the DB recovers.

tests/gateway/test_fifo_overflow_rescue.py

@@ -0,0 +1,159 @@
+"""Regression tests for #99882: FIFO overflow orphan rescue.
+
+When a follow-up is demoted to /queue during compression-in-flight,
+it lands in SessionState.conversation.queued_events (overflow) with
+the current turn's event occupying adapter._pending_messages[session_key]
+(slot).  After the slot's turn completes, _promote_queued_event moves
+the overflow head into the slot.  When that drain never runs — the
+busy window ended through an exit that skipped the promotion site
+(/stop, turn exception, generation bump) — the overflow is silently
+orphaned: never dispatched, never persisted, never logged.
+
+The rescue in GatewayRunner._rescue_orphaned_overflow pops the oldest
+orphan for the caller to run as the current turn and stages the next
+orphan in the slot, so FIFO order (#28503) holds and nothing runs twice.
+"""
+
+from unittest.mock import MagicMock
+
+from gateway.platforms.base import (
+    BasePlatformAdapter,
+    MessageEvent,
+    MessageType,
+    Platform,
+    PlatformConfig,
+)
+from gateway.run import GatewayRunner
+
+
+class _StubAdapter(BasePlatformAdapter):
+    def __init__(self):
+        super().__init__(PlatformConfig(enabled=True, token="test"), Platform.TELEGRAM)
+
+    async def connect(self, *, is_reconnect: bool = False) -> bool:
+        return True
+
+    async def disconnect(self) -> None:
+        self._mark_disconnected()
+
+    async def send(self, chat_id, content, reply_to=None, metadata=None):
+        from gateway.platforms.base import SendResult
+
+        return SendResult(success=True, message_id="msg-1")
+
+    async def get_chat_info(self, chat_id):
+        return {"id": chat_id, "type": "dm"}
+
+
+def _text_event(text: str, msg_id: str) -> MessageEvent:
+    return MessageEvent(
+        text=text,
+        message_type=MessageType.TEXT,
+        source=MagicMock(chat_id="123", platform=Platform.TELEGRAM, profile=None),
+        message_id=msg_id,
+    )
+
+
+def _runner() -> GatewayRunner:
+    runner = GatewayRunner.__new__(GatewayRunner)
+    runner._queued_events = {}
+    return runner
+
+
+class TestRescueOrphanedOverflow:
+    def test_single_orphan_is_returned_and_removed_from_both_stores(self):
+        runner = _runner()
+        adapter = _StubAdapter()
+        session_key = "telegram:user:1"
+        runner._session_state(session_key).conversation.queued_events.append(
+            _text_event("orphan-1", "o1")
+        )
+        assert session_key not in adapter._pending_messages
+
+        rescued = runner._rescue_orphaned_overflow(session_key, adapter)
+
+        assert rescued is not None and rescued.text == "orphan-1"
+        # The rescued event runs as the current turn, so it must NOT also
+        # sit in the slot — the post-turn drain would run it a second time.
+        assert session_key not in adapter._pending_messages
+        assert runner._session_state(session_key).conversation.queued_events == []
+
+    def test_two_orphans_return_oldest_and_stage_next_in_slot(self):
+        runner = _runner()
+        adapter = _StubAdapter()
+        session_key = "telegram:user:1b"
+        runner._session_state(session_key).conversation.queued_events.extend(
+            [_text_event("orphan-1", "o1"), _text_event("orphan-2", "o2")]
+        )
+
+        rescued = runner._rescue_orphaned_overflow(session_key, adapter)
+
+        assert rescued is not None and rescued.text == "orphan-1"
+        # Slot now holds the NEXT orphan so the drain continues the chain.
+        assert adapter._pending_messages[session_key].text == "orphan-2"
+        assert runner._session_state(session_key).conversation.queued_events == []
+
+    def test_noop_when_slot_occupied(self):
+        runner = _runner()
+        adapter = _StubAdapter()
+        session_key = "telegram:user:2"
+        runner._session_state(session_key).conversation.queued_events.append(
+            _text_event("orphan", "o1")
+        )
+        adapter._pending_messages[session_key] = _text_event("busy-slot", "slot")
+
+        rescued = runner._rescue_orphaned_overflow(session_key, adapter)
+
+        assert rescued is None
+        assert adapter._pending_messages[session_key].text == "busy-slot"
+        assert len(runner._session_state(session_key).conversation.queued_events) == 1
+
+    def test_noop_when_no_overflow(self):
+        runner = _runner()
+        adapter = _StubAdapter()
+        session_key = "telegram:user:3"
+
+        rescued = runner._rescue_orphaned_overflow(session_key, adapter)
+
+        assert rescued is None
+        assert session_key not in adapter._pending_messages
+
+    def test_fifo_order_preserved_across_rescue_and_new_message(self):
+        """Oldest orphan runs first, new arrival last — FIFO (#28503).
+
+        Mirrors the idle-arrival call site: rescue → _enqueue_fifo(new).
+        """
+        runner = _runner()
+        adapter = _StubAdapter()
+        session_key = "telegram:user:4"
+        runner._session_state(session_key).conversation.queued_events.extend(
+            [_text_event("orphan-1", "o1"), _text_event("orphan-2", "o2")]
+        )
+
+        rescued = runner._rescue_orphaned_overflow(session_key, adapter)
+        assert rescued is not None and rescued.text == "orphan-1"
+        runner._enqueue_fifo(session_key, _text_event("new-msg", "new1"), adapter)
+
+        # Drain order after this turn: slot (orphan-2), then overflow (new-msg)
+        assert adapter._pending_messages[session_key].text == "orphan-2"
+        overflow_texts = [
+            e.text for e in runner._session_state(session_key).conversation.queued_events
+        ]
+        assert overflow_texts == ["new-msg"]
+
+    def test_single_orphan_then_new_message_lands_in_slot(self):
+        """With one orphan the slot is free after rescue, so the incoming
+        message must go to the slot (not overflow) or the drain never sees it."""
+        runner = _runner()
+        adapter = _StubAdapter()
+        session_key = "telegram:user:5"
+        runner._session_state(session_key).conversation.queued_events.append(
+            _text_event("orphan-1", "o1")
+        )
+
+        rescued = runner._rescue_orphaned_overflow(session_key, adapter)
+        assert rescued is not None and rescued.text == "orphan-1"
+        runner._enqueue_fifo(session_key, _text_event("new-msg", "new1"), adapter)
+
+        assert adapter._pending_messages[session_key].text == "new-msg"
+        assert runner._session_state(session_key).conversation.queued_events == []