Fix #72680 (retargeted): preserve agent._session_messages on shutdown flush failure

The previous attempt (#73171) snapshotted GatewayRunner._pending_messages,
which on current main has no writers (commit f6736ced8 removed its write
path; interrupt delivery uses adapter._pending_messages instead). 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 disk=0/memory=N state from #72680), the in-memory transcript is lost when
the process exits.

Retarget the preservation to the real path: in _finalize_shutdown_agents, wrap
the _flush call; on exception, dump agent._session_messages to an external JSON
recovery snapshot under $HERMES_HOME/shutdown-recovery/ (tagged issue=#72680)
so an operator can salvage it after repairing state.db. The dump is fully
guarded (non-fatal) so shutdown never blocks on a best-effort backup.

This directly addresses the reviewer note on #73171: retarget to the actual
cached-agent history (agent._session_messages) and prove a stale DB + shutdown
leaves a recoverable transcript.

Regression tests: tests/gateway/test_session_messages_shutdown_preserve.py
- flush raises -> recovery file written with session_id + messages
- healthy flush -> no recovery file
- write error -> non-fatal, no raise

Fixes #72680
This commit is contained in:
Baophan00 2026-07-28 20:54:19 +07:00 • committed by kshitij
parent 015718066a
commit 40837e2dd0
2 changed files with 189 additions and 1 deletions

View file

@ -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],

View file

@ -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