diff --git a/gateway/run.py b/gateway/run.py index c77ca7ec848..78ca83129e5 100644 --- a/gateway/run.py +++ b/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], diff --git a/gateway/shutdown_flush.py b/gateway/shutdown_flush.py index e144c6ef43e..7b5c57be9e7 100644 --- a/gateway/shutdown_flush.py +++ b/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, + ) diff --git a/tests/gateway/test_session_messages_shutdown_preserve.py b/tests/gateway/test_session_messages_shutdown_preserve.py index 03edf75ba42..a9367dcbaee 100644 --- a/tests/gateway/test_session_messages_shutdown_preserve.py +++ b/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 "" + + 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] == ""