fix: consolidate agent-history preservation into shutdown_flush module

Follow-up for salvaged PRs #73400 + #73372:
- Move _preserve_agent_history_on_shutdown logic into
  gateway/shutdown_flush.py as flush_agent_history_to_file()
- Reuse the hardened _write_payload (atomic writes, fsync, private
  permissions, UUID filenames) from #73372 instead of the plain
  open()/write() from #73400
- Replace os.environ.get('HERMES_HOME') with get_hermes_home() for
  profile-safe path resolution
- Update tests to test the real function directly
- Net: removes 37 lines from gateway/run.py, unifies both fix paths
  in a single module with consistent atomic-write guarantees

gateway/run.py

@@ -6956,7 +6956,8 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                             _flush_err,
                             len(_session_messages),
                         )
-                        self._preserve_agent_history_on_shutdown(
+                        from gateway.shutdown_flush import flush_agent_history_to_file
+                        flush_agent_history_to_file(
                             getattr(agent, "session_id", None),
                             _session_messages,
                         )
@@ -6977,66 +6978,6 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                 agent, context="shutdown finalize"
             )
 
-    def _preserve_agent_history_on_shutdown(
-        self, session_id: Optional[str], history: list
-    ) -> None:
-        """Best-effort dump of an agent's in-memory transcript before teardown.
-
-        Used when ``_flush_messages_to_session_db`` raises (e.g. FTS/SQLite
-        index corruption, #72680): the live ``agent._session_messages`` could
-        not be written to disk, and a plain debug log would lose it permanently
-        when the process exits. Serialize to an external JSON file outside the
-        broken DB so an operator can salvage the conversation after repairing
-        state.db.
-
-        Failures are swallowed — shutdown must never block on a best-effort
-        backup.
-        """
-        if not history:
-            return
-        try:
-            import json
-            import os
-            from datetime import datetime, timezone
-
-            hermes_home = os.environ.get("HERMES_HOME", os.path.expanduser("~/.hermes"))
-            out_dir = os.path.join(hermes_home, "shutdown-recovery")
-            os.makedirs(out_dir, exist_ok=True)
-            stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
-            safe_sid = (session_id or "unknown").replace("/", "_").replace("\\", "_")
-            out_path = os.path.join(out_dir, f"agent_history_{safe_sid}_{stamp}.json")
-            snapshot = []
-            for _m in history:
-                try:
-                    snapshot.append(
-                        _m if isinstance(_m, (dict, list, str, int, float, bool, type(None)))
-                        else str(_m)
-                    )
-                except Exception:
-                    continue
-            with open(out_path, "w", encoding="utf-8") as _fh:
-                json.dump(
-                    {
-                        "reason": "shutdown-with-unpersisted-agent-history",
-                        "issue": "#72680",
-                        "session_id": session_id,
-                        "count": len(snapshot),
-                        "messages": snapshot,
-                    },
-                    _fh,
-                    ensure_ascii=False,
-                    indent=2,
-                )
-            logger.warning(
-                "Preserved %d in-memory message(s) for session %s to %s "
-                "(possible FTS corruption — recover after repairing state.db)",
-                len(snapshot),
-                session_id,
-                out_path,
-            )
-        except Exception as _e:
-            logger.debug("Agent-history shutdown preservation skipped: %s", _e)
-
     def _should_emit_long_running_notification(
         self,
         session_key: Optional[str],

gateway/shutdown_flush.py

@@ -1,10 +1,11 @@
-"""Flush pending messages to disk before shutdown to prevent data loss.
+"""Flush pending messages and agent transcripts to disk before shutdown to prevent data loss.
 
 When FTS5 index corruption prevents ``INSERT INTO messages``, the gateway
-accumulates messages in ``_pending_messages`` (memory-only).  On shutdown,
-``.clear()`` discards the only surviving copy — permanent user data loss.
+accumulates messages in ``_pending_messages`` (memory-only) and the live
+``agent._session_messages`` cannot be flushed via ``_flush_messages_to_session_db``.
+On shutdown, ``.clear()`` discards the only surviving copy — permanent user data loss.
 
-This module provides two hooks:
+This module provides three hooks:
 
 1. ``flush_pending_to_file()`` — called BEFORE ``_pending_messages.clear()``
    during shutdown.  Serialises any non-empty pending slots to a JSON file
@@ -15,6 +16,10 @@ This module provides two hooks:
    (so FTS indexing, session metadata, and display_kind are handled correctly),
    then deletes the flush file on success.
 
+3. ``flush_agent_history_to_file()`` — called from ``_finalize_shutdown_agents``
+   when ``_flush_messages_to_session_db`` raises.  Dumps the live
+   ``agent._session_messages`` to the same atomic JSON recovery directory.
+
 See issue #72680 for the full incident report.
 """
 
@@ -198,6 +203,11 @@ def recover_pending_to_db(
     for path in flush_files:
         try:
             payload = json.loads(path.read_text(encoding="utf-8"))
+            # 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
             session_key = payload.get("session_key", "")
             data = payload.get("data", {})
             text = data.get("text", "")
@@ -257,3 +267,55 @@ def recover_pending_to_db(
             "Recovered %d pending message(s) from shutdown flush", recovered,
         )
     return recovered
+
+
+def flush_agent_history_to_file(
+    session_id: Optional[str],
+    history: list,
+) -> None:
+    """Best-effort dump of an agent's in-memory transcript before teardown.
+
+    Used when ``_flush_messages_to_session_db`` raises (e.g. FTS/SQLite
+    index corruption, #72680): the live ``agent._session_messages`` could
+    not be written to disk, and a plain debug log would lose it permanently
+    when the process exits. Serialize to an atomic JSON file outside the
+    broken DB so an operator can salvage the conversation after repairing
+    state.db.
+
+    Failures are swallowed — shutdown must never block on a best-effort
+    backup.
+    """
+    if not history:
+        return
+    try:
+        flush_dir = _get_flush_dir()
+        snapshot = []
+        for _m in history:
+            try:
+                snapshot.append(
+                    _m if isinstance(_m, (dict, list, str, int, float, bool, type(None)))
+                    else str(_m)
+                )
+            except Exception:
+                continue
+        _write_payload(
+            flush_dir,
+            {
+                "reason": "shutdown-with-unpersisted-agent-history",
+                "issue": "#72680",
+                "session_id": session_id,
+                "count": len(snapshot),
+                "messages": snapshot,
+            },
+        )
+        logger.warning(
+            "Preserved %d in-memory message(s) for session %s "
+            "(possible FTS corruption — recover after repairing state.db)",
+            len(snapshot),
+            session_id,
+        )
+    except Exception as _e:
+        logger.warning(
+            "Agent-history shutdown preservation failed for session %s: %s",
+            session_id, _e,
+        )

tests/gateway/test_session_messages_shutdown_preserve.py

@@ -7,59 +7,37 @@ When that flush raises (FTS/SQLite corruption) the in-memory transcript must
 be dumped to a recovery snapshot instead of lost.
 
 These tests exercise the real preservation path:
-``_finalize_shutdown_agents`` -> flush raises -> ``_preserve_agent_history_on_shutdown``.
+``_finalize_shutdown_agents`` -> flush raises -> ``_preserve_agent_history_on_shutdown``
+-> ``flush_agent_history_to_file``.
 """
 from __future__ import annotations
 
-import importlib.util
 import json
-import os
-import sys
-import types
 from pathlib import Path
 
 import pytest
 
-_REPO = Path(__file__).resolve().parents[2]
-_GATEWAY_RUN = _REPO / "gateway" / "run.py"
+from gateway.shutdown_flush import (
+    flush_agent_history_to_file,
+)
 
 
-def _make_runner_with_agent(mod, *, flush_raises=False, history=None):
-    """Build a minimal object graph exercising the real method chain."""
-    runner = types.SimpleNamespace()
-    runner._pending_messages = {}  # runner dict (unused by live path, kept for parity)
-    runner._preserve_agent_history_on_shutdown = (
-        mod.GatewayRunner._preserve_agent_history_on_shutdown.__get__(runner, mod.GatewayRunner)
-    )
-
-    class FakeAgent:
-        session_id = "sess:abc123"
-        _session_messages = history or [{"role": "user", "content": "hi"}]
-
-        def _flush_messages_to_session_db(self, messages, conversation_history=None):
-            if flush_raises:
-                raise RuntimeError("database disk image is malformed")
-            # healthy path: nothing to dump
-            self._flushed = True
-
-    agent = FakeAgent()
-    return runner, agent
+def _make_flush_dir(tmp_path: Path) -> Path:
+    """Create a temp flush dir and monkeypatch _get_flush_dir to use it."""
+    flush_dir = tmp_path / "pending_messages"
+    flush_dir.mkdir(parents=True, exist_ok=True)
+    return flush_dir
 
 
 def test_preserves_agent_history_when_flush_raises(tmp_path, monkeypatch):
-    mod = _load_gateway_run()
-    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
-    runner, agent = _make_runner_with_agent(
-        mod, flush_raises=True, history=[{"role": "user", "content": "lost msg"}]
+    flush_dir = _make_flush_dir(tmp_path)
+    monkeypatch.setattr(
+        "gateway.shutdown_flush._get_flush_dir", lambda: flush_dir
     )
-    # Simulate the relevant slice of _finalize_shutdown_agents.
-    _flush = getattr(agent, "_flush_messages_to_session_db")
-    try:
-        _flush(agent._session_messages)
-    except Exception as _flush_err:
-        runner._preserve_agent_history_on_shutdown(agent.session_id, agent._session_messages)
+    history = [{"role": "user", "content": "lost msg"}]
+    flush_agent_history_to_file("sess:abc123", history)
 
-    files = list((tmp_path / "shutdown-recovery").glob("agent_history_*.json"))
+    files = list(flush_dir.glob("*.json"))
     assert files, "expected recovery snapshot"
     data = json.loads(files[0].read_text(encoding="utf-8"))
     assert data["issue"] == "#72680"
@@ -68,41 +46,56 @@ def test_preserves_agent_history_when_flush_raises(tmp_path, monkeypatch):
     assert data["messages"][0]["content"] == "lost msg"
 
 
-def test_no_recovery_file_on_healthy_flush(tmp_path, monkeypatch):
-    mod = _load_gateway_run()
-    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
-    runner, agent = _make_runner_with_agent(mod, flush_raises=False)
-    _flush = getattr(agent, "_flush_messages_to_session_db")
-    try:
-        _flush(agent._session_messages)
-    except Exception as _flush_err:
-        runner._preserve_agent_history_on_shutdown(agent.session_id, agent._session_messages)
-    assert not list((tmp_path / "shutdown-recovery").glob("*.json"))
+def test_no_recovery_file_on_empty_history(tmp_path, monkeypatch):
+    flush_dir = _make_flush_dir(tmp_path)
+    monkeypatch.setattr(
+        "gateway.shutdown_flush._get_flush_dir", lambda: flush_dir
+    )
+    flush_agent_history_to_file("sess:abc123", [])
+    assert not list(flush_dir.glob("*.json"))
+
+
+def test_no_recovery_file_on_none_history(tmp_path, monkeypatch):
+    flush_dir = _make_flush_dir(tmp_path)
+    monkeypatch.setattr(
+        "gateway.shutdown_flush._get_flush_dir", lambda: flush_dir
+    )
+    flush_agent_history_to_file("sess:abc123", None)  # type: ignore[arg-type]
+    assert not list(flush_dir.glob("*.json"))
 
 
 def test_non_fatal_on_write_error(tmp_path, monkeypatch):
-    mod = _load_gateway_run()
-    bad = tmp_path / "file"
-    bad.write_text("x")
-    monkeypatch.setenv("HERMES_HOME", str(bad))
-    runner, agent = _make_runner_with_agent(
-        mod, flush_raises=True, history=[{"role": "user", "content": "x"}]
+    flush_dir = _make_flush_dir(tmp_path)
+    monkeypatch.setattr(
+        "gateway.shutdown_flush._get_flush_dir", lambda: flush_dir
+    )
+
+    def fail_write(*args, **kwargs):
+        raise OSError("simulated write failure")
+
+    monkeypatch.setattr("gateway.shutdown_flush._write_payload", fail_write)
+
+    # Must not raise even though the dump target is broken.
+    flush_agent_history_to_file(
+        "sess:abc123", [{"role": "user", "content": "x"}]
     )
-    _flush = getattr(agent, "_flush_messages_to_session_db")
-    try:
-        _flush(agent._session_messages)
-    except Exception as _flush_err:
-        # Must not raise even though the dump target is invalid.
-        runner._preserve_agent_history_on_shutdown(agent.session_id, agent._session_messages)
 
 
-def _load_gateway_run():
-    spec = importlib.util.spec_from_file_location("gateway_run_72680b", _GATEWAY_RUN)
-    mod = importlib.util.module_from_spec(spec)
-    mod.logger = types.SimpleNamespace(debug=lambda *a, **k: None, warning=lambda *a, **k: None)
-    sys.modules["gateway_run_72680b"] = mod
-    try:
-        spec.loader.exec_module(mod)
-    except Exception:
-        pass
-    return mod
+def test_preserves_non_serializable_as_string(tmp_path, monkeypatch):
+    flush_dir = _make_flush_dir(tmp_path)
+    monkeypatch.setattr(
+        "gateway.shutdown_flush._get_flush_dir", lambda: flush_dir
+    )
+
+    class Weird:
+        def __str__(self):
+            return "<weird>"
+
+    history = [{"role": "user", "content": "ok"}, Weird()]
+    flush_agent_history_to_file("sess:abc", history)
+
+    files = list(flush_dir.glob("*.json"))
+    assert len(files) == 1
+    data = json.loads(files[0].read_text(encoding="utf-8"))
+    assert data["count"] == 2
+    assert data["messages"][1] == "<weird>"