fix(gateway): keep live transcript rows queued while older spool files fail to drain

The live writer replayed a session's spool before each write but ignored a failed replay and wrote the new row anyway, so a second failure (boot hold-back, then the pre-write drain) gave the live row a lower row id than the older spooled rows. _drain_spooled_drops now reports files left on disk, and the write takes the existing retry path instead, leaving the row queued in memory behind them.

Review finding by @ehz0ah.

gateway/session_transcript.py

@@ -274,10 +274,12 @@ class SessionTranscriptMixin:
 
         # 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)
             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.
+                if self._drain_spooled_drops(session_id):
+                    raise RuntimeError(f"older spooled transcript rows for {session_id} still pending")
                 self._append_transcript_message(session_id, msg)
             except Exception as exc:
                 from hermes_state import StateDbCorruptError, StateDbReplacedError
@@ -371,12 +373,13 @@ class SessionTranscriptMixin:
                     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) -> bool:
+        """Replay cap-dropped spooled transcript messages after DB recovery; True while some of them
+        are still on disk. 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 False
         try:
             from gateway.shutdown_flush import drain_transcript_spool
             # Inside an outage the append that follows logs/escalates the same failure; the
@@ -389,8 +392,10 @@ class SessionTranscriptMixin:
             )
             if not remaining:
                 spooled_sessions.discard(session_id)
+            return bool(remaining)
         except Exception as exc:
             logger.warning("Failed to drain transcript spool for %s: %s", session_id, exc)
+            return False
 
     def spooled_drop_sessions(self) -> set:
         """The live set of sessions whose next transcript write drains the spool first. Boot recovery

tests/gateway/test_shutdown_flush_recovery.py

@@ -193,3 +193,38 @@ def test_held_back_files_drain_before_that_sessions_next_live_write(flush_dir, m
 
     assert _contents(mock_db) == ["old0", "old0", "old1", "new-live"]
     assert not list(flush_dir.glob("*.json"))
+
+
+def test_live_write_waits_while_held_back_files_still_fail_to_drain(flush_dir, monkeypatch):
+    """If the pre-write drain fails again, the live row must stay queued in memory: writing it
+    while the older files are still on disk gives it a lower row id than they will get."""
+    import hermes_state_registry
+    from gateway.run import _recover_pending_flushes
+    from gateway.session import SessionStore
+
+    _write_spool(flush_dir, "pending-bbb.json", "sess-1",
+                 {"role": "user", "content": "old0"}, ts=100, seq=0)
+    _write_spool(flush_dir, "pending-aaa.json", "sess-1",
+                 {"role": "user", "content": "old1"}, ts=100, seq=1)
+    mock_db = MagicMock()
+    mock_db.append_message.side_effect = [
+        RuntimeError("still locked at boot"), RuntimeError("still locked at drain"), 1, 1, 1, 1]
+    monkeypatch.setattr(hermes_state_registry, "acquire", lambda: mock_db)
+    monkeypatch.setattr(hermes_state_registry, "release_or_close", lambda _db: None)
+    store = object.__new__(SessionStore)
+    store._db = mock_db
+    store._transcript_retry_lock = threading.Lock()
+    store._dirty_transcripts = {}
+    store._transcript_append_failures = {}
+    store._fts_rebuild_last_attempt_at = time.monotonic()
+
+    runner = SimpleNamespace(config=SimpleNamespace(multiplex_profiles=False), session_store=store)
+    assert _recover_pending_flushes(runner) == 0
+    store.append_to_transcript("sess-1", {"role": "user", "content": "new-live"})
+    assert _contents(mock_db) == ["old0", "old0"]
+    assert len(list(flush_dir.glob("*.json"))) == 2
+
+    store.append_to_transcript("sess-1", {"role": "user", "content": "new-live-2"})
+    assert _contents(mock_db) == ["old0", "old0", "old0", "old1", "new-live", "new-live-2"]
+    assert not list(flush_dir.glob("*.json"))
+    assert not store._dirty_transcripts