diff --git a/gateway/run.py b/gateway/run.py index b97b74b928a..c77ca7ec848 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -6939,7 +6939,27 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew _strip(_session_messages) except Exception: pass - _flush(_session_messages) + try: + _flush(_session_messages) + except Exception as _flush_err: + # The in-memory transcript could not be persisted + # (e.g. FTS/SQLite index corruption — #72680). A plain + # debug log loses the conversation permanently when the + # process exits. Dump the live agent history to an + # external JSON recovery snapshot so an operator can + # salvage it after repairing state.db. The flush is + # non-fatal; shutdown must never block on a best-effort + # backup. + logger.warning( + "Shutdown transcript flush failed (%s); preserving " + "%d in-memory message(s) to recovery snapshot", + _flush_err, + len(_session_messages), + ) + self._preserve_agent_history_on_shutdown( + getattr(agent, "session_id", None), + _session_messages, + ) except Exception as _e: logger.debug("Shutdown transcript flush failed: %s", _e) try: @@ -6957,6 +6977,66 @@ 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/tests/gateway/test_session_messages_shutdown_preserve.py b/tests/gateway/test_session_messages_shutdown_preserve.py new file mode 100644 index 00000000000..03edf75ba42 --- /dev/null +++ b/tests/gateway/test_session_messages_shutdown_preserve.py @@ -0,0 +1,108 @@ +"""Regression tests for #72680 (retargeted). + +The earlier attempt (#73171) snapshotted GatewayRunner._pending_messages, which +on current main has no writers — the live container is the per-agent +``agent._session_messages`` flushed via ``_flush_messages_to_session_db``. +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``. +""" +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" + + +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 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"}] + ) + # 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) + + files = list((tmp_path / "shutdown-recovery").glob("agent_history_*.json")) + assert files, "expected recovery snapshot" + data = json.loads(files[0].read_text(encoding="utf-8")) + assert data["issue"] == "#72680" + assert data["session_id"] == "sess:abc123" + assert data["count"] == 1 + 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_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 = 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