fix(gateway): replay the transcript spool in drop order with full fidelity on restart (salvage #122718)

gateway/run.py

@@ -1651,35 +1651,6 @@ def _multiplex_profile_homes(config: object) -> list[tuple[str, "Path"]]:
     return list(profiles_to_serve(multiplex=True))
 
 
-def _recover_pending_flushes(runner) -> int:
-    """Replay every ``pending_messages`` spool this gateway owns into state.db; return the count.
-
-    ``_get_flush_dir`` follows the active HERMES_HOME, so a routed turn on a multiplexed gateway spools
-    its stalled transcript backlog under ``profiles/<name>/`` and the runtime drain forgets it on
-    restart. After the launch home, replay each served profile inside its own home so the default
-    store ``recover_pending_to_db`` opens is that profile's state.db (#123584).
-    """
-    from gateway.shutdown_flush import recover_pending_to_db
-    from hermes_constants import get_hermes_home, reset_hermes_home_override, set_hermes_home_override
-
-    resolver = runner.session_store.resolve_session_id_for_key
-    recovered = recover_pending_to_db(session_resolver=resolver)
-    if not getattr(runner.config, "multiplex_profiles", False):
-        return recovered
-    launch_home = Path(get_hermes_home()).resolve()
-    for name, home in _multiplex_profile_homes(runner.config):
-        if Path(home).resolve() == launch_home or not (Path(home) / "pending_messages").is_dir():
-            continue
-        token = set_hermes_home_override(str(home))
-        try:
-            recovered += recover_pending_to_db(session_resolver=resolver)
-        except Exception:  # one profile's unreadable spool must not strand the others'
-            logger.warning("Pending-message recovery failed for profile %s", name, exc_info=True)
-        finally:
-            reset_hermes_home_override(token)
-    return recovered
-
-
 def _cron_tick_profile_homes(config: object) -> list[tuple[str, "Path"]]:
     """Profile homes the in-process ticker visits: the served set PLUS the process-active
     profile: ``profiles_to_serve`` lists default + every live named profile, but a ``--profile
@@ -5968,13 +5939,6 @@ def _start_keepalive() -> None:
     if not success:
         _shutdown_gateway_health_export(runner)
         return False
-
-    def _recover_pending() -> None:
-        recovered = _recover_pending_flushes(runner)
-        if recovered:
-            logger.info("Recovered %d pending message(s) from shutdown flush", recovered)
-
-    _best_effort(_recover_pending)
     if runner.should_exit_cleanly:
         _shutdown_gateway_health_export(runner)
         if runner.exit_reason:

gateway/run_startup.py

@@ -1508,6 +1508,7 @@ async def _start_post_connect_services(self, connected_count: int) -> None:
     async def _start_finish_wiring(self, connected_count: int) -> None:
         """Post-connect wiring: services, boot notifications, startup restore, recovered watchers."""
         from gateway.run import _planned_restart_notification_pending, _restart_notification_pending
+        from gateway.shutdown_flush import recover_gateway_pending
         await self._start_post_connect_services(connected_count)
         # Let fresh adapters settle before lifecycle sends (helps Discord thread deliveries).
         if connected_count > 0:
@@ -1523,6 +1524,11 @@ async def _start_finish_wiring(self, connected_count: int) -> None:
         await self._await_startup_boot_sends(
             planned_restart_notification_pending=_planned_restart_notification_pending(),
         )
+        # Replay the previous run's pending spool before resume turns and queued inbound write live
+        # rows: only this pass tells the store which sessions have spooled rows, and a live row
+        # written first would get a lower row id than them for good.
+        with _log_suppressed(logging.WARNING, "Pending-message recovery failed: %s"):
+            recover_gateway_pending(self)
         # Auto-resume restart-interrupted sessions (ledger-answered ones were cleared above); a failed
         # auto-resume stays visible on the next user message.
         self._schedule_resume_pending_sessions()

gateway/session_transcript.py

@@ -5,6 +5,7 @@
 from __future__ import annotations
 
 import contextlib
+import json
 import logging
 import threading
 import time
@@ -41,6 +42,55 @@ def _spool_dropped(session_id: str, message: Dict[str, Any]):
     "reasoning", "reasoning_content", "reasoning_details", "codex_reasoning_items",
     "codex_message_items",
 )
+# Message keys bound straight to a TEXT column. A spooled message is JSON from disk, and a dict or list
+# here makes sqlite refuse the whole row.
+_BOUND_TEXT_KEYS = ("role", "tool_name", "tool_call_id", "reasoning", "reasoning_content", "platform_message_id")
+
+
+def _bindable_text(value: Any) -> Any:
+    """*value* when sqlite can bind it as is, else its JSON text (``str`` if that fails too). An int
+    outside sqlite's signed 64-bit range raises OverflowError even for a TEXT column."""
+    if value is None or isinstance(value, (str, float)) or (
+            isinstance(value, int) and -2**63 <= value < 2**63):
+        return value
+    try:
+        return json.dumps(value, default=str)
+    except (TypeError, ValueError):
+        return str(value)
+
+
+def transcript_append_kwargs(session_id: str, message: Dict[str, Any], *, fallback_ts: Any = None) -> Dict[str, Any]:
+    """``SessionDB.append_message`` kwargs for one transcript row. The live writer and the restart
+    replay of the transcript spool (gateway/shutdown_flush.py) both build rows here, so a replayed
+    message lands as the row the live drain would have written. Fields are listed rather than
+    splatted: a spooled message is arbitrary JSON from disk, and an unexpected key would raise and
+    abort the replay. *fallback_ts* applies only to a message with no timestamp at all, because a
+    truthiness test would rewrite epoch 0. Values bound straight to a TEXT column are made bindable
+    (:func:`_bindable_text`), so one malformed field cannot make the row unwritable."""
+    is_assistant = message.get("role") == "assistant"
+    timestamp = message.get("timestamp")
+    kwargs = {
+        "session_id": session_id,
+        "role": message.get("role", "unknown"),
+        "content": message.get("content"),
+        "tool_name": message.get("tool_name"),
+        "tool_calls": message.get("tool_calls"),
+        "tool_call_id": message.get("tool_call_id"),
+        **{k: message.get(k) if is_assistant else None for k in _ASSISTANT_ONLY_KEYS},
+        "platform_message_id": message.get("platform_message_id") or message.get("message_id"),
+        "observed": bool(message.get("observed")),
+        "timestamp": fallback_ts if timestamp is None else timestamp,
+        # Exact bytes sent to the API (prompt-cache-stable replay); must survive every
+        # persistence path or the next replay diverges.
+        "api_content": extract_api_content_sidecar(message),
+        # Presentation typing ("internal_notification" for self-injected async-delegation/background
+        # notification turns, #82888). DB-only; stripped from provider-bound payloads.
+        "display_kind": message.get("display_kind"),
+        "display_metadata": message.get("display_metadata"),
+    }
+    for key in _BOUND_TEXT_KEYS:
+        kwargs[key] = _bindable_text(kwargs[key])
+    return kwargs
 
 
 class SessionTranscriptMixin:
@@ -244,10 +294,15 @@ def _ack_head() -> bool:
 
         # DB write outside the retry lock so other sessions can append.
         while True:
-            # Spooled backlog (cap eviction or a stalled session) is older than ``msg``: replay it
-            # first so recovery keeps transcript order; a still-dead DB just fails both.
-            self._drain_spooled_drops(session_id)
+            spool_exc = None
             try:
+                # Spooled backlog (cap eviction, a stalled session, a boot hold-back) is older than
+                # ``msg``: replay it first. While any of it stays on disk ``msg`` stays queued, since
+                # writing it now would give it a lower row id than that backlog for good. The replay's
+                # own error is raised so the repair/divert handling below still classifies it.
+                spool_exc = self._drain_spooled_drops(session_id)
+                if spool_exc is not None:
+                    raise spool_exc
                 self._append_transcript_message(session_id, msg)
             except Exception as exc:
                 from hermes_state import StateDbCorruptError, StateDbReplacedError
@@ -289,6 +344,8 @@ def _ack_head() -> bool:
                             "no unique live child; not retrying", session_id)
                         return
                 if self._is_fts_corruption_error(exc) and self._rebuild_fts_once():
+                    if spool_exc is not None:
+                        continue  # repaired: drain the older spool again before ``msg``
                     try:
                         self._append_transcript_message(session_id, msg)
                     except Exception as retry_exc:
@@ -334,33 +391,54 @@ def _spool_stalled_backlog(self, session_id: str, queue_session_id: str) -> int:
                 break
             spooled += 1
         if spooled:
-            self._lazy("_spooled_drop_sessions", set).add(session_id)
             with self._transcript_retry_lock:
+                self._lazy("_spooled_drop_sessions", set).add(session_id)
                 del pending[:spooled]
                 if not pending:
                     self._dirty_transcripts.pop(queue_session_id, None)
         return spooled
 
-    def _drain_spooled_drops(self, session_id: str) -> None:
-        """Replay cap-dropped spooled transcript messages after DB recovery. Best-effort: replay
-        failures keep the spool files for the next successful flush; nothing here may raise."""
+    def _drain_spooled_drops(self, session_id: str) -> Exception | None:
+        """Replay cap-dropped spooled transcript messages after DB recovery; return the replay
+        failure while some of them are still on disk, else None. Best-effort: replay failures keep
+        the spool files for the next successful flush; nothing here may raise."""
         spooled_sessions = getattr(self, "_spooled_drop_sessions", None)
         if not spooled_sessions or session_id not in spooled_sessions:
-            return
+            return None
+        failures: list[Exception] = []
+
+        def replay(message: Dict[str, Any]) -> None:
+            try:
+                self._append_transcript_message(session_id, message)
+            except Exception as exc:
+                failures.append(exc)
+                raise
+
         try:
             from gateway.shutdown_flush import drain_transcript_spool
             # Inside an outage the append that follows logs/escalates the same failure; the
             # replay attempt is only the order-preserving probe, so its failure stays at DEBUG.
             with self._transcript_retry_lock:
                 known_failing = bool(self._transcript_append_failures.get(session_id))
             _replayed, remaining = drain_transcript_spool(
-                session_id, lambda message: self._append_transcript_message(session_id, message),
-                db_known_failing=known_failing,
-            )
+                session_id, replay, db_known_failing=known_failing)
             if not remaining:
-                spooled_sessions.discard(session_id)
+                with self._transcript_retry_lock:
+                    spooled_sessions.discard(session_id)
+                return None
+            return failures[-1] if failures else RuntimeError(
+                f"older spooled transcript rows for {session_id} still pending")
         except Exception as exc:
             logger.warning("Failed to drain transcript spool for %s: %s", session_id, exc)
+            return None
+
+    def mark_spooled_drop_sessions(self, session_ids) -> None:
+        """Have each session's next transcript write drain the on-disk spool first. Boot recovery marks
+        the sessions it held back, so their next live row cannot land ahead of those files."""
+        if not session_ids:
+            return
+        with self._transcript_retry_lock:
+            self._lazy("_spooled_drop_sessions", set).update(session_ids)
 
     def _append_transcript_message(self, session_id: str, message: Dict[str, Any]) -> None:
         """Write one transcript row. Caller handles retry queuing."""
@@ -370,27 +448,7 @@ def _append_transcript_message(self, session_id: str, message: Dict[str, Any]) -
             # into the ambient store.
             raise RuntimeError(
                 f"no owning session store for {session_id}; deferring transcript write")
-        is_assistant = message.get("role") == "assistant"
-        _db.append_message(
-            session_id=session_id,
-            role=message.get("role", "unknown"),
-            content=message.get("content"),
-            tool_name=message.get("tool_name"),
-            tool_calls=message.get("tool_calls"),
-            tool_call_id=message.get("tool_call_id"),
-            **{k: message.get(k) if is_assistant else None for k in _ASSISTANT_ONLY_KEYS},
-            platform_message_id=(message.get("platform_message_id") or message.get("message_id")),
-            observed=bool(message.get("observed")),
-            timestamp=message.get("timestamp"),
-            # Exact bytes sent to the API (prompt-cache-stable replay); must survive every
-            # persistence path or the next replay diverges.
-            api_content=extract_api_content_sidecar(message),
-            # Presentation typing (e.g. "internal_notification"); DB-only.
-            # "internal_notification" for self-injected async-delegation/background notification turns,
-            # #82888). DB-only; stripped from provider-bound payloads.
-            display_kind=message.get("display_kind"),
-            display_metadata=message.get("display_metadata"),
-        )
+        _db.append_message(**transcript_append_kwargs(session_id, message))
 
     @staticmethod
     def _is_fts_corruption_error(exc: Exception) -> bool: