fix(gateway): harden restart resume recovery

gateway/run.py

@@ -1744,6 +1744,57 @@ def _last_transcript_timestamp(history: Optional[List[Dict[str, Any]]]) -> Any:
     return None
 
 
+def _last_usable_transcript_row(
+    history: Optional[List[Dict[str, Any]]],
+) -> Optional[Dict[str, Any]]:
+    """Return the last transcript row that would be replayed to the agent."""
+    if not history:
+        return None
+    for msg in reversed(history):
+        if not isinstance(msg, dict):
+            continue
+        role = msg.get("role")
+        if not role or role in {"session_meta", "system"}:
+            continue
+        return msg
+    return None
+
+
+def _last_user_message_id(
+    history: Optional[List[Dict[str, Any]]],
+) -> Optional[str]:
+    """Return the platform message id for the last persisted user turn."""
+    if not history:
+        return None
+    for msg in reversed(history):
+        if not isinstance(msg, dict) or msg.get("role") != "user":
+            continue
+        message_id = msg.get("message_id")
+        if message_id is None:
+            return None
+        text = str(message_id).strip()
+        return text or None
+    return None
+
+
+def _transcript_tail_is_completed_assistant(
+    history: Optional[List[Dict[str, Any]]],
+) -> bool:
+    """Return whether replayable history already ends in a final answer."""
+    msg = _last_usable_transcript_row(history)
+    if not msg or msg.get("role") != "assistant":
+        return False
+    if msg.get("tool_calls") or msg.get("function_call"):
+        return False
+    finish_reason = msg.get("finish_reason")
+    if finish_reason not in {"stop", "end_turn", "complete", "completed"}:
+        return False
+    content = msg.get("content")
+    if isinstance(content, str):
+        return bool(content.strip())
+    return bool(content)
+
+
 # Tool results can contain literal MEDIA: examples in docs, logs, or other
 # ordinary outputs. Only tools that intentionally create deliverable media
 # artifacts should be eligible for automatic append when the model omits them
@@ -10846,9 +10897,11 @@ async def _notify_active_sessions_of_shutdown(self) -> None:
 
         notified: set[tuple[str, str, Optional[str]]] = set()
         for session_key in active:
-            source = None
+            # The live cache carries the current inbound reply anchor; the
+            # persisted origin is intentionally long-lived and can be stale.
+            source = self._get_cached_session_source(session_key)
             try:
-                if getattr(self, "session_store", None) is not None:
+                if source is None and getattr(self, "session_store", None) is not None:
                     await self.async_session_store._ensure_loaded()
                     entry = self.session_store._entries.get(session_key)
                     source = getattr(entry, "origin", None) if entry else None
@@ -10859,9 +10912,6 @@ async def _notify_active_sessions_of_shutdown(self) -> None:
                     e,
                 )
 
-            if source is None:
-                source = self._get_cached_session_source(session_key)
-
             if source is not None:
                 platform_str = source.platform.value
                 chat_id = str(source.chat_id)
@@ -12037,7 +12087,10 @@ def _log_background_boot_send_result(task: "asyncio.Task") -> None:
             task, "background boot-path send failed after gate release: see traceback"
         )
 
-    async def _claim_pending_obligations(self) -> list:
+    async def _claim_pending_obligations(
+        self,
+        platform: Optional[Platform] = None,
+    ) -> list:
         """Claim recoverable delivery-ledger rows and clear their
         ``resume_pending`` flags. Pure DB work — no network sends.
 
@@ -12066,8 +12119,12 @@ async def _claim_pending_obligations(self) -> list:
             # holds a platform only after its connect() succeeded, and each
             # claim spends one of the row's three redelivery attempts.
             _deliverable = {
-                getattr(p, "value", str(p)) for p in self.adapters
+                getattr(p, "value", str(p))
+                for p in self.adapters
+                if platform is None or p == platform
             }
+            if not _deliverable:
+                return 0
             claimed = await asyncio.to_thread(
                 sweep_recoverable, None, deliverable_platforms=_deliverable
             )
@@ -12168,7 +12225,10 @@ async def _redeliver_claimed_obligations(self, claimed: list) -> int:
                 logger.debug("delivery ledger update failed", exc_info=True)
         return redelivered
 
-    async def _redeliver_pending_obligations(self) -> int:
+    async def _redeliver_pending_obligations(
+        self,
+        platform: Optional[Platform] = None,
+    ) -> int:
         """Claim + redeliver in one call — composition of
         :meth:`_claim_pending_obligations` and
         :meth:`_redeliver_claimed_obligations`.
@@ -12178,10 +12238,10 @@ async def _redeliver_pending_obligations(self) -> int:
         so the DB half can run inline before the abandonable send task.
         """
         return await self._redeliver_claimed_obligations(
-            await self._claim_pending_obligations()
+            await self._claim_pending_obligations(platform=platform)
         )
 
-    def _schedule_resume_pending_sessions(self, platform=None) -> int:
+    async def _schedule_resume_pending_sessions(self, platform=None) -> int:
         """Auto-continue fresh restart-interrupted sessions after startup.
 
         ``resume_pending`` already preserves the transcript AND the existing
@@ -12206,16 +12266,13 @@ def _schedule_resume_pending_sessions(self, platform=None) -> int:
         """
         window = _auto_continue_freshness_window()
         try:
-            with self.session_store._lock:  # noqa: SLF001 — snapshot under lock
-                self.session_store._ensure_loaded_locked()  # noqa: SLF001
-                candidates = [
-                    entry for entry in self.session_store._entries.values()  # noqa: SLF001
-                    if entry.resume_pending
-                    and not entry.suspended
-                    and entry.origin is not None
-                    and entry.resume_reason in self._AUTO_RESUME_REASONS
-                    and (platform is None or entry.origin.platform == platform)
-                ]
+            candidates = [
+                entry
+                for entry in await self.async_session_store.list_resume_pending(
+                    platform=platform
+                )
+                if entry.resume_reason in self._AUTO_RESUME_REASONS
+            ]
         except Exception as exc:
             logger.warning("Failed to enumerate resume-pending sessions: %s", exc)
             return 0
@@ -12245,16 +12302,58 @@ def _schedule_resume_pending_sessions(self, platform=None) -> int:
         now = datetime.now()
         scheduled = 0
         for entry in candidates:
-            marker = entry.last_resume_marked_at or entry.updated_at
-            if marker is not None and (now - marker).total_seconds() > window:
+            effective_session_id = await self._effective_resume_session_id(entry)
+            history = None
+            try:
+                history = await self.async_session_store.load_transcript(
+                    effective_session_id
+                )
+                if not isinstance(history, list):
+                    history = None
+            except Exception:
+                logger.debug(
+                    "Failed to load transcript while checking auto-resume freshness for %s",
+                    entry.session_key,
+                    exc_info=True,
+                )
+
+            if _transcript_tail_is_completed_assistant(history):
+                try:
+                    await self.async_session_store.clear_resume_pending(
+                        entry.session_key
+                    )
+                except Exception:
+                    logger.debug(
+                        "clear stale resume_pending failed for %s",
+                        entry.session_key,
+                        exc_info=True,
+                    )
                 continue
 
+            transcript_ts = _last_transcript_timestamp(history)
+            if history:
+                if not _is_fresh_gateway_interruption(
+                    transcript_ts,
+                    now=now.timestamp(),
+                    window_secs=window,
+                ):
+                    continue
+            else:
+                marker = entry.last_resume_marked_at or entry.updated_at
+                if marker is not None and (now - marker).total_seconds() > window:
+                    continue
+
             # Already being resumed (e.g. scheduled at startup and still
             # in-flight) — don't synthesize a second continuation turn.
             if self._is_session_running(entry.session_key):
                 continue
 
             source = entry.origin
+            resume_message_id = _last_user_message_id(history)
+            try:
+                source = dataclasses.replace(source, message_id=resume_message_id)
+            except Exception:
+                pass
             adapter = self._adapter_for_source(source)
             if adapter is None:
                 logger.debug(
@@ -12302,6 +12401,8 @@ def _schedule_resume_pending_sessions(self, platform=None) -> int:
                 text="",
                 message_type=MessageType.TEXT,
                 source=source,
+                message_id=resume_message_id,
+                reply_to_message_id=resume_message_id,
                 internal=True,
             )
             task = asyncio.create_task(
@@ -12323,6 +12424,42 @@ def _schedule_resume_pending_sessions(self, platform=None) -> int:
             )
         return scheduled
 
+    async def _effective_resume_session_id(self, entry) -> str:
+        """Return the transcript id an auto-resume would actually consume."""
+        session_id = str(getattr(entry, "session_id", "") or "")
+        source = getattr(entry, "origin", None)
+        session_db = getattr(self, "_session_db", None)
+        if (
+            source is None
+            or session_db is None
+            or getattr(source, "platform", None) != Platform.TELEGRAM
+            or getattr(source, "chat_type", None) != "dm"
+            or not getattr(source, "chat_id", None)
+            or not getattr(source, "thread_id", None)
+        ):
+            return session_id
+        try:
+            binding = await session_db.get_telegram_topic_binding(
+                chat_id=str(source.chat_id),
+                thread_id=str(source.thread_id),
+            )
+            bound_session_id = str((binding or {}).get("session_id") or "")
+            if not bound_session_id:
+                return session_id
+            try:
+                tip = await session_db.get_compression_tip(bound_session_id)
+            except Exception:
+                logger.debug(
+                    "Failed to resolve compression tip for Telegram topic resume binding %s",
+                    bound_session_id,
+                    exc_info=True,
+                )
+                tip = None
+            return str(tip or bound_session_id)
+        except Exception:
+            logger.debug("Failed to resolve Telegram topic resume binding", exc_info=True)
+            return session_id
+
     def _startup_should_abort(self) -> bool:
         return (
             self._restart_requested
@@ -13318,7 +13455,7 @@ async def _connect_one_startup(p, p_cfg, adp):
         # a session whose final response was generated but never
         # confirmed-delivered has its answer in the ledger — redelivering it
         # is strictly cheaper and more correct than re-running the whole turn.
-        self._schedule_resume_pending_sessions()
+        await self._schedule_resume_pending_sessions()
         await self._finish_startup_restore()
 
         # Surface state.db init failures to the user's messaging platforms
@@ -14597,14 +14734,25 @@ async def _platform_reconnect_watcher(self) -> None:
                         except Exception:
                             pass
 
-                        # A platform that was offline at gateway startup never
-                        # got its restart-interrupted sessions auto-resumed —
-                        # the startup pass skips sessions whose adapter isn't
-                        # connected yet. Now that it's back, retry the
-                        # auto-resume scoped to this platform so recovery
-                        # doesn't silently wait for a manual user message.
+                        # A platform that was offline at startup may have both
+                        # completed responses waiting in the durable ledger and
+                        # interrupted turns waiting to resume. Deliver stored
+                        # responses first so their resume markers are cleared
+                        # before considering another model turn.
+                        try:
+                            await self._redeliver_pending_obligations(
+                                platform=platform
+                            )
+                        except Exception:
+                            logger.debug(
+                                "pending delivery redelivery after %s reconnect failed",
+                                platform.value,
+                                exc_info=True,
+                            )
                         try:
-                            self._schedule_resume_pending_sessions(platform=platform)
+                            await self._schedule_resume_pending_sessions(
+                                platform=platform
+                            )
                         except Exception:
                             logger.debug(
                                 "resume-pending reschedule after %s reconnect failed",

gateway/session.py

@@ -1438,6 +1438,24 @@ def _ensure_loaded(self) -> None:
         with self._lock:
             self._ensure_loaded_locked()
 
+    def list_resume_pending(
+        self, platform: Optional[Platform] = None
+    ) -> List[SessionEntry]:
+        """Return a stable snapshot of restart-interrupted routing entries."""
+        with self._lock:
+            self._ensure_loaded_locked()
+            return [
+                replace(entry)
+                for entry in self._entries.values()
+                if entry.resume_pending
+                and not entry.suspended
+                and entry.origin is not None
+                and (
+                    platform is None
+                    or entry.origin.platform == platform
+                )
+            ]
+
     def _routing_scope(self) -> str:
         """Namespace for this store's rows in the gateway_routing table.
 

tests/gateway/restart_test_helpers.py

@@ -158,6 +158,24 @@ def make_restart_runner(
     runner.pairing_store = MagicMock()
     runner.session_store = MagicMock()
     runner.session_store._entries = {}
+    runner.session_store._ensure_loaded = MagicMock()
+    runner.session_store.load_transcript = MagicMock(return_value=[])
+    runner._async_session_store = MagicMock()
+    runner._async_session_store._store = runner.session_store
+    runner._async_session_store._ensure_loaded = AsyncMock()
+    runner._async_session_store.list_resume_pending = AsyncMock(
+        side_effect=lambda platform=None: [
+            entry
+            for entry in runner.session_store._entries.values()
+            if entry.resume_pending
+            and not entry.suspended
+            and entry.origin is not None
+            and (platform is None or entry.origin.platform == platform)
+        ]
+    )
+    runner._async_session_store.load_transcript = AsyncMock(return_value=[])
+    runner._async_session_store.clear_resume_pending = AsyncMock()
+    runner._session_db = None
     runner.delivery_router = MagicMock()
 
     platform_adapter = adapter or RestartTestAdapter()

tests/gateway/test_platform_reconnect.py

@@ -227,7 +227,13 @@ async def test_reconnect_retries_resume_pending_for_platform(self):
         """
         runner = _make_runner()
         runner._sync_voice_mode_state_to_adapter = MagicMock()
-        runner._schedule_resume_pending_sessions = MagicMock(return_value=1)
+        order = []
+        runner._redeliver_pending_obligations = AsyncMock(
+            side_effect=lambda **_: order.append("ledger") or 1
+        )
+        runner._schedule_resume_pending_sessions = AsyncMock(
+            side_effect=lambda **_: order.append("resume") or 1
+        )
 
         platform_config = PlatformConfig(enabled=True, token="test")
         runner._failed_platforms[Platform.TELEGRAM] = {
@@ -258,9 +264,13 @@ async def fake_sleep(n):
                 await run_one_iteration()
 
         assert Platform.TELEGRAM in runner.adapters
-        runner._schedule_resume_pending_sessions.assert_called_once_with(
+        runner._redeliver_pending_obligations.assert_awaited_once_with(
+            platform=Platform.TELEGRAM
+        )
+        runner._schedule_resume_pending_sessions.assert_awaited_once_with(
             platform=Platform.TELEGRAM
         )
+        assert order == ["ledger", "resume"]
 
 
     @pytest.mark.asyncio