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

gateway/run.py

@@ -1662,8 +1662,8 @@ def _recover_pending_flushes(runner) -> int:
     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)
+    resolver, held_back = runner.session_store.resolve_session_id_for_key, runner.session_store.spooled_drop_sessions()
+    recovered = recover_pending_to_db(session_resolver=resolver, held_back=held_back)
     if not getattr(runner.config, "multiplex_profiles", False):
         return recovered
     launch_home = Path(get_hermes_home()).resolve()
@@ -1672,7 +1672,7 @@ def _recover_pending_flushes(runner) -> int:
             continue
         token = set_hermes_home_override(str(home))
         try:
-            recovered += recover_pending_to_db(session_resolver=resolver)
+            recovered += recover_pending_to_db(session_resolver=resolver, held_back=held_back)
         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:
@@ -5968,13 +5968,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

@@ -1507,7 +1507,8 @@ 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.run import (
+            _planned_restart_notification_pending, _recover_pending_flushes, _restart_notification_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,12 @@ 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"):
+            if recovered := _recover_pending_flushes(self):
+                logger.info("Recovered %d pending message(s) from shutdown flush", recovered)
         # 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

@@ -43,6 +43,36 @@ def _spool_dropped(session_id: str, message: Dict[str, Any]):
 )
 
 
+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."""
+    is_assistant = message.get("role") == "assistant"
+    timestamp = message.get("timestamp")
+    return {
+        "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"),
+    }
+
+
 class SessionTranscriptMixin:
     """SessionStore transcript I/O: SQLite append with a per-session retry queue,
     compression-reroute following, FTS corruption recovery, rewrite/rewind/load."""
@@ -244,10 +274,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 +324,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:
@@ -341,26 +378,43 @@ def _spool_stalled_backlog(self, session_id: str, queue_session_id: str) -> int:
                     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)
+                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 spooled_drop_sessions(self) -> set:
+        """The live set of sessions whose next transcript write drains the spool first. Boot recovery
+        adds the sessions it held back, so their next live row cannot land ahead of those files."""
+        return self._lazy("_spooled_drop_sessions", set)
 
     def _append_transcript_message(self, session_id: str, message: Dict[str, Any]) -> None:
         """Write one transcript row. Caller handles retry queuing."""
@@ -370,27 +424,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:

gateway/shutdown_flush.py

@@ -15,6 +15,7 @@
 import itertools
 import json
 import logging
+import math
 import os
 import time
 import uuid
@@ -206,7 +207,74 @@ def _serialise_value(value: Any) -> Optional[dict]:
     return {"text": str(value)}
 
 
-def recover_pending_to_db(session_db=None, *, session_resolver=None) -> int:
+def _sort_number(value: Any) -> float:
+    """Coerce a spool ordering field to a float; unusable values sort first.
+
+    Ordering fields are read back from JSON on disk and may be missing or
+    corrupt.  A sort key that mixes ``str`` and ``int`` raises ``TypeError``
+    and would abort the entire recovery pass, so anything non-numeric is
+    normalised to ``0.0``.  So is a JSON integer too large for a float
+    (``float(10**400)`` raises ``OverflowError``) and a NaN/Infinity literal
+    (NaN compares false both ways and would scramble the sort).
+    """
+    if isinstance(value, bool) or not isinstance(value, (int, float)):
+        return 0.0
+    try:
+        number = float(value)
+    except OverflowError:
+        return 0.0
+    return number if math.isfinite(number) else 0.0
+
+
+def _order_flush_files(paths) -> list[Path]:
+    """Order recovery payloads by drop order — ``(ts, seq, filename)``.
+
+    Recovery used to walk ``sorted(glob("*.json"))``, but spool files are
+    named ``pending-<uuid4>.json`` (see :func:`_write_payload`), so filename
+    order is effectively random.  ``SessionDB`` restores a conversation by
+    AUTOINCREMENT id — true insertion order, never timestamp — so replaying
+    in filename order permanently scrambles the recovered transcript and can
+    separate an assistant tool call from its result.
+
+    The payloads already carry ``ts`` and ``seq``
+    (:func:`spool_dropped_transcript_message`), so this mirrors
+    :func:`drain_transcript_spool`'s ``sorted(entries, key=lambda e: e[:3])``
+    and the live drain and the cross-restart drain agree on replay order.
+
+    Returns paths in replay order, retaining only ordering metadata, not decoded
+    payloads. Unparseable files sort last by name. The caller re-reads each
+    payload for replay and reports failures through its own error handling.
+    """
+    entries = []
+    for path in paths:
+        try:
+            payload = json.loads(path.read_text(encoding="utf-8-sig"))
+        # OSError: unreadable; ValueError: bad JSON or bytes; RecursionError: nesting too deep.
+        except (OSError, ValueError, RecursionError):
+            payload = None
+        if not isinstance(payload, dict):
+            entries.append(((1, 0.0, 0.0, path.name), path))
+            continue
+        entries.append((
+            (
+                0,
+                _sort_number(payload.get("ts")),
+                # seq is per process (_TRANSCRIPT_SPOOL_SEQ) and ts has one-second resolution, so
+                # files two processes spool in the same second can interleave; within one process
+                # the order is exact.
+                # Slot heads have no seq; overflow tails start at zero. Interpret legacy
+                # heads as -1 too, so files already on disk recover before their tail.
+                _sort_number(payload.get("seq", -1)),
+                path.name,
+            ),
+            path,
+        ))
+    entries.sort(key=lambda entry: entry[0])
+    return [path for _key, path in entries]
+
+
+def recover_pending_to_db(session_db=None, *, session_resolver=None,
+                          held_back: Optional[set] = None) -> int:
     """Replay flush-dir ``*.json`` files via ``SessionDB.append_message``, deleting each on success.
 
     ``session_db=None`` opens (and afterwards releases) the shared default ``state.db``.
@@ -215,15 +283,21 @@ def recover_pending_to_db(session_db=None, *, session_resolver=None) -> int:
     ``MessageEvent`` objects carry no ``session_id``, so without it every recovery lands in the skip
     branch. A returned ``db`` routes the append to the profile store owning the key (multiplexed
     gateways); ``None`` falls back to ``session_db``. Returns the number of messages recovered.
+    ``held_back`` (optional set) receives the session ids whose spool files this pass held back
+    after a failed replay, so the caller can have the live writer drain them before that session's
+    next row.
     """
-    flush_files = sorted(_get_flush_dir().glob("*.json"))
+    flush_files = _order_flush_files(_get_flush_dir().glob("*.json"))
     if not flush_files:
         return 0
     own_db = session_db is None
     if own_db:
         from hermes_state_registry import acquire
         session_db = acquire()
     recovered = 0
+    # Sessions whose spool replay already failed this pass -> files held back for them. Ordering is a
+    # per-session property, so one unhealthy session must not hold back the others.
+    blocked_sessions: Dict[str, int] = {}
     try:
         for path in flush_files:
             # One unparseable payload or rejected append must only skip THIS file: the file is
@@ -237,7 +311,8 @@ def recover_pending_to_db(session_db=None, *, session_resolver=None) -> int:
                 if payload.get("reason") == "shutdown-with-unpersisted-agent-history":
                     continue
                 if _recover_one_payload(session_db, path, payload,
-                                        session_resolver=session_resolver):
+                                        session_resolver=session_resolver,
+                                        blocked_sessions=blocked_sessions):
                     recovered += 1
                     path.unlink(missing_ok=True)
             except Exception as exc:
@@ -249,11 +324,18 @@ def recover_pending_to_db(session_db=None, *, session_resolver=None) -> int:
                 release_or_close(session_db)
     if recovered:
         logger.info("Recovered %d pending message(s) from shutdown flush", recovered)
+    if blocked_sessions:
+        # Otherwise a spool dir that never empties gives an operator no reason why.
+        logger.info("Held back %d spooled transcript file(s) for the next start after a failed replay: %s",
+                    sum(blocked_sessions.values()),
+                    ", ".join(f"{sid} ({count})" for sid, count in blocked_sessions.items()))
+    if held_back is not None:
+        held_back.update(blocked_sessions)
     return recovered
 
 
 def _recover_one_payload(session_db, path: Path, payload: Dict[str, Any], *,
-                         session_resolver=None) -> bool:
+                         session_resolver=None, blocked_sessions: Optional[Dict[str, int]] = None) -> bool:
     """Append one flush payload to ``session_db``; False (file kept) when structurally invalid."""
     # Cap-dropped transcript payloads carry the full message dict keyed by session_id — replay directly
     # (#78182). This handles spool files that were never drained before a restart.
@@ -265,9 +347,21 @@ def _recover_one_payload(session_db, path: Path, payload: Dict[str, Any], *,
             logger.warning("Cannot recover structurally invalid transcript spool "
                            "file %s; preserved for manual inspection", path)
             return False
-        session_db.append_message(session_id=spooled_sid, role=message.get("role", "unknown"),
-                                  content=message.get("content") or "",
-                                  timestamp=message.get("timestamp") or payload.get("ts"))
+        if blocked_sessions is not None and spooled_sid in blocked_sessions:
+            # An older message for this session could not be replayed. Writing this one now would give
+            # it a lower row id than the message it follows, permanently inverting the transcript, so
+            # leave it for the next start.
+            blocked_sessions[spooled_sid] += 1
+            return False
+        try:
+            from gateway.session_transcript import transcript_append_kwargs
+            session_db.append_message(**transcript_append_kwargs(spooled_sid, message, fallback_ts=payload.get("ts")))
+        except Exception:
+            # Same contract as drain_transcript_spool: stop this session's replay on the first failure
+            # and keep the remaining spool files for the next attempt.
+            if blocked_sessions is not None:
+                blocked_sessions[spooled_sid] = 1
+            raise
         return True
     session_key, data = payload.get("session_key", ""), payload.get("data", {})
     text = data.get("text", "")