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

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."""
@@ -370,27 +400,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,6 +207,71 @@ def _serialise_value(value: Any) -> Optional[dict]:
     return {"text": str(value)}
 
 
+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[tuple[Path, Optional[Dict[str, Any]]]]:
+    """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 ``(path, payload)`` pairs in replay order.  A file whose payload
+    cannot be parsed sorts last, by name, and is returned with ``None`` so
+    the caller reports it 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, None))
+            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.
+                _sort_number(payload.get("seq")),
+                path.name,
+            ),
+            path,
+            payload,
+        ))
+    entries.sort(key=lambda entry: entry[0])
+    return [(path, payload) for _key, path, payload in entries]
+
+
 def recover_pending_to_db(session_db=None, *, session_resolver=None) -> int:
     """Replay flush-dir ``*.json`` files via ``SessionDB.append_message``, deleting each on success.
 
@@ -216,28 +282,35 @@ def recover_pending_to_db(session_db=None, *, session_resolver=None) -> int:
     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.
     """
-    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:
+        for path, payload in flush_files:
             # One unparseable payload or rejected append must only skip THIS file: the file is
             # never unlinked, so aborting the pass would re-poison every later boot.
             # utf-8-sig: our BOM-tolerant read fix for flush files.
             try:
-                payload = json.loads(path.read_text(encoding="utf-8-sig"))
+                if payload is None:
+                    # Unparseable on the ordering pass: re-read so the failure is raised and
+                    # reported here exactly as it was before.
+                    payload = json.loads(path.read_text(encoding="utf-8-sig"))
                 # Agent-history snapshots use a different schema (reason +
                 # messages list) and are meant for manual operator recovery,
                 # not automatic DB insertion. Skip them silently.
                 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 +322,16 @@ 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()))
     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 +343,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", "")

tests/gateway/test_shutdown_flush_recovery.py

@@ -0,0 +1,162 @@
+"""Cross-restart recovery of cap-dropped transcript spool files (#78182).
+
+``recover_pending_to_db`` is the restart-time consumer of the same spool
+``drain_transcript_spool`` drains during live operation.  These tests pin the
+properties the live drain already guarantees for that spool.
+"""
+
+import json
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pytest
+
+from gateway.shutdown_flush import (
+    TRANSCRIPT_CAP_DROP_REASON,
+    recover_pending_to_db,
+)
+
+
+@pytest.fixture
+def flush_dir(tmp_path, monkeypatch):
+    """A temp spool directory wired into the module under test."""
+    directory = tmp_path / "pending_messages"
+    directory.mkdir(parents=True, exist_ok=True)
+    monkeypatch.setattr(
+        "gateway.shutdown_flush._get_flush_dir", lambda: directory
+    )
+    return directory
+
+
+def _write_spool(
+    flush_dir: Path,
+    name: str,
+    session_id: str,
+    message: dict,
+    *,
+    ts: int,
+    seq: int,
+) -> Path:
+    """Write one cap-drop spool payload under an explicit file name.
+
+    Production names these ``pending-<uuid4>.json``; the tests choose the
+    names so that filename order and drop order can be made to disagree.
+    """
+    path = flush_dir / name
+    path.write_text(
+        json.dumps(
+            {
+                "session_key": session_id,
+                "reason": TRANSCRIPT_CAP_DROP_REASON,
+                "ts": ts,
+                "seq": seq,
+                "data": {"session_id": session_id, "message": message},
+            }
+        ),
+        encoding="utf-8",
+    )
+    return path
+
+
+def _contents(mock_db) -> list:
+    return [c.kwargs["content"] for c in mock_db.append_message.call_args_list]
+
+
+def test_replays_in_drop_order_not_file_name_order(flush_dir):
+    """SessionDB restores by AUTOINCREMENT id, so append order IS the order the user sees
+    after recovery. Production names are ``pending-<uuid4>.json``; these names sort opposite
+    to drop order, and a burst inside one second shares ``ts``, so ``seq`` breaks the tie."""
+    _write_spool(flush_dir, "pending-zzz.json", "sess-1",
+                 {"role": "user", "content": "first"}, ts=100, seq=0)
+    _write_spool(flush_dir, "pending-mmm.json", "sess-1",
+                 {"role": "assistant", "content": "second"}, ts=100, seq=1)
+    _write_spool(flush_dir, "pending-aaa.json", "sess-1",
+                 {"role": "user", "content": "third"}, ts=101, seq=2)
+
+    mock_db = MagicMock()
+    assert recover_pending_to_db(mock_db) == 3
+    assert _contents(mock_db) == ["first", "second", "third"]
+
+
+def test_an_out_of_range_ordering_field_does_not_abort_recovery(flush_dir):
+    """Ordering runs before the per-file error handling, so one file whose ``ts`` is a JSON
+    integer no float can hold must not stop every other file from being recovered."""
+    _write_spool(flush_dir, "pending-bad.json", "sess-1",
+                 {"role": "user", "content": "huge ts"}, ts=10**400, seq=0)
+    _write_spool(flush_dir, "pending-ok.json", "sess-2",
+                 {"role": "user", "content": "valid"}, ts=100, seq=1)
+
+    mock_db = MagicMock()
+    assert recover_pending_to_db(mock_db) == 2
+    assert sorted(_contents(mock_db)) == ["huge ts", "valid"]
+
+
+def test_a_replayed_row_is_the_row_the_live_writer_writes(flush_dir):
+    """Every field the live drain persists survives the restart round trip: losing
+    tool_call_id orphans a tool result, losing api_content makes the next replay diverge,
+    and reasoning stays assistant-only. A missing timestamp falls back to the payload
+    clock, but epoch 0 is a real timestamp."""
+    tool_calls = [{"id": "call-1", "type": "function",
+                   "function": {"name": "send_payment", "arguments": "{}"}}]
+    _write_spool(flush_dir, "pending-ccc.json", "sess-1", {
+        "role": "assistant", "content": None, "tool_calls": tool_calls,
+        "reasoning": "deliberating", "reasoning_content": "chain",
+        "reasoning_details": [{"type": "text"}], "codex_reasoning_items": [{"id": "r1"}],
+        "codex_message_items": [{"id": "m1"}], "platform_message_id": "tg-42",
+        "observed": True, "timestamp": 0, "api_content": "exact bytes sent to the API",
+        "display_kind": "internal_notification",
+    }, ts=100, seq=0)
+    _write_spool(flush_dir, "pending-bbb.json", "sess-1", {
+        "role": "tool", "content": "receipt-1", "tool_call_id": "call-1", "tool_name": "send_payment",
+    }, ts=100, seq=1)
+    _write_spool(flush_dir, "pending-aaa.json", "sess-1", {
+        "role": "user", "content": "hi", "reasoning": "leaked", "message_id": "tg-7",
+    }, ts=999, seq=2)
+
+    mock_db = MagicMock()
+    assert recover_pending_to_db(mock_db) == 3
+    assistant, tool, user = (c.kwargs for c in mock_db.append_message.call_args_list)
+
+    assert assistant["content"] is None  # a tool-call row legitimately has no content
+    assert assistant["tool_calls"] == tool_calls
+    assert (assistant["reasoning"], assistant["reasoning_content"]) == ("deliberating", "chain")
+    assert assistant["reasoning_details"] == [{"type": "text"}]
+    assert assistant["codex_reasoning_items"] == [{"id": "r1"}]
+    assert assistant["codex_message_items"] == [{"id": "m1"}]
+    assert (assistant["platform_message_id"], assistant["observed"]) == ("tg-42", True)
+    assert assistant["api_content"] == "exact bytes sent to the API"
+    assert assistant["display_kind"] == "internal_notification"
+    assert assistant["timestamp"] == 0
+    assert (tool["tool_call_id"], tool["tool_name"]) == ("call-1", "send_payment")
+    assert user["reasoning"] is None
+    assert user["platform_message_id"] == "tg-7"
+    assert user["timestamp"] == 999
+
+
+def test_a_failed_replay_holds_back_that_sessions_later_messages_only(flush_dir, caplog):
+    """Writing "second" after "first" failed would give it a lower row id than "first" once
+    "first" is retried on a later start: the inversion lands on disk for good. Another
+    session is unaffected, and the pass says what it held back."""
+    first = _write_spool(flush_dir, "pending-aaa.json", "sess-1",
+                         {"role": "user", "content": "first"}, ts=100, seq=0)
+    second = _write_spool(flush_dir, "pending-bbb.json", "sess-1",
+                          {"role": "user", "content": "second"}, ts=101, seq=1)
+    other = _write_spool(flush_dir, "pending-ccc.json", "sess-2",
+                         {"role": "user", "content": "other-session"}, ts=102, seq=2)
+
+    mock_db = MagicMock()
+
+    def append_message(**kwargs):
+        if kwargs["content"] == "first":
+            raise RuntimeError("controlled database outage")
+        return 1
+
+    mock_db.append_message.side_effect = append_message
+
+    with caplog.at_level("INFO", logger="gateway.shutdown_flush"):
+        assert recover_pending_to_db(mock_db) == 1
+
+    assert _contents(mock_db) == ["first", "other-session"]
+    assert first.exists() and second.exists()
+    assert not other.exists()
+    assert "Held back 2 spooled transcript file(s)" in caplog.text and "sess-1 (2)" in caplog.text