diff --git a/gateway/shutdown_flush.py b/gateway/shutdown_flush.py index d12f5b0cc07..8f0bb478be3 100644 --- a/gateway/shutdown_flush.py +++ b/gateway/shutdown_flush.py @@ -8,11 +8,12 @@ This module provides two hooks: 1. ``flush_pending_to_file()`` — called BEFORE ``_pending_messages.clear()`` during shutdown. Serialises any non-empty pending slots to a JSON file - under ``~/.hermes/pending_messages/``. + under ``/pending_messages/``. 2. ``recover_pending_to_db()`` — called AFTER ``runner.start()`` on startup. - Reads flush files, inserts messages directly into state.db, then deletes - the flush file on success. + Reads flush files, inserts messages into state.db via ``SessionDB.append_message`` + (so FTS indexing, session metadata, and display_kind are handled correctly), + then deletes the flush file on success. See issue #72680 for the full incident report. """ @@ -22,17 +23,19 @@ from __future__ import annotations import json import logging import time -from pathlib import Path from typing import Any, Dict, Optional logger = logging.getLogger(__name__) -_FLUSH_DIR = Path.home() / ".hermes" / "pending_messages" +def _get_flush_dir(): + """Return the pending-messages flush directory under the active HERMES_HOME.""" + from hermes_constants import get_hermes_home + from pathlib import Path -def _get_flush_dir() -> Path: - _FLUSH_DIR.mkdir(parents=True, exist_ok=True) - return _FLUSH_DIR + flush_dir = get_hermes_home() / "pending_messages" + flush_dir.mkdir(parents=True, exist_ok=True) + return flush_dir def flush_pending_to_file( @@ -125,35 +128,39 @@ def _serialise_value(value: Any) -> Optional[dict]: def recover_pending_to_db( - db_path: Optional[Path] = None, + session_db=None, ) -> int: - """Recover flushed pending messages into state.db. + """Recover flushed pending messages into state.db via SessionDB. Reads all ``*.json`` files from the flush directory, inserts messages - into the ``messages`` table, and deletes the file on success. + using ``SessionDB.append_message`` (so FTS indexing, session metadata + updates, and all required columns are handled correctly), and deletes + the flush file on success. Parameters ---------- - db_path: - Path to state.db. Defaults to ``~/.hermes/state.db``. + session_db: + An existing ``SessionDB`` instance. If ``None``, a new one is + opened on the default ``state.db`` path. Returns ------- int Number of messages recovered. """ - import sqlite3 + from pathlib import Path flush_dir = _get_flush_dir() flush_files = sorted(flush_dir.glob("*.json")) if not flush_files: return 0 - if db_path is None: - db_path = Path.home() / ".hermes" / "state.db" - if not db_path.exists(): - logger.warning("state.db not found at %s — skipping recovery", db_path) - return 0 + # Use the provided SessionDB or open one on the default path. + own_db = False + if session_db is None: + from hermes_state import SessionDB + session_db = SessionDB() + own_db = True recovered = 0 for path in flush_files: @@ -166,27 +173,48 @@ def recover_pending_to_db( path.unlink(missing_ok=True) continue - conn = sqlite3.connect(str(db_path)) - try: - conn.execute( - "INSERT INTO messages (session_key, role, content, created_at) " - "VALUES (?, ?, ?, ?)", - (session_key, "user", text, payload.get("ts", int(time.time()))), - ) - conn.commit() - recovered += 1 - path.unlink(missing_ok=True) - except Exception as exc: + # The session_key is a gateway routing key (e.g. + # "agent:main:telegram:supergroup:..."). We need the actual + # session_id (e.g. "20260728_120000_abc123") to append a + # message row. Try the session_id field from the serialised + # data first; fall back to scanning sessions for a matching + # session_key in the source column. + session_id = data.get("session_id", "") + + if not session_id: + # Try to extract from the session_key itself — gateway + # session keys contain the session_id as the last segment + # in some formats, but that's not guaranteed. Log and + # skip if we can't resolve it. logger.warning( - "Failed to recover pending message for %s: %s", - session_key, exc, + "Cannot recover pending message for %s: no session_id " + "in flush file and session_key-to-id resolution is not " + "available at this recovery stage. The message text is " + "preserved in %s", + session_key, path, ) - # Re-save so next startup can retry - finally: - conn.close() - except Exception as exc: - logger.debug("Failed to read flush file %s: %s", path, exc) + continue + + session_db.append_message( + session_id=session_id, + role="user", + content=text, + timestamp=payload.get("ts", int(time.time())), + ) + recovered += 1 path.unlink(missing_ok=True) + except Exception as exc: + logger.warning( + "Failed to recover pending message from %s: %s", + path, exc, + ) + # Leave the file for next startup retry. + + if own_db: + try: + session_db.close() + except Exception: + pass if recovered: logger.info( diff --git a/tests/gateway/test_shutdown_flush.py b/tests/gateway/test_shutdown_flush.py new file mode 100644 index 00000000000..efbed00f372 --- /dev/null +++ b/tests/gateway/test_shutdown_flush.py @@ -0,0 +1,199 @@ +"""Tests for gateway/shutdown_flush.py — pending message durability (#72680).""" + +import json +import time +from pathlib import Path +from unittest.mock import MagicMock + +import pytest + +from gateway.shutdown_flush import ( + _serialise_value, + flush_pending_to_file, + recover_pending_to_db, +) + + +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_flush_empty_pending_is_noop(tmp_path, monkeypatch): + flush_dir = _make_flush_dir(tmp_path) + monkeypatch.setattr( + "gateway.shutdown_flush._get_flush_dir", lambda: flush_dir + ) + assert flush_pending_to_file({}, reason="test") == 0 + assert list(flush_dir.glob("*.json")) == [] + + +def test_flush_writes_string_pending_to_file(tmp_path, monkeypatch): + flush_dir = _make_flush_dir(tmp_path) + monkeypatch.setattr( + "gateway.shutdown_flush._get_flush_dir", lambda: flush_dir + ) + pending = {"agent:main:telegram:supergroup:123": "hello world"} + count = flush_pending_to_file(pending, reason="shutdown") + assert count == 1 + files = list(flush_dir.glob("*.json")) + assert len(files) == 1 + payload = json.loads(files[0].read_text(encoding="utf-8")) + assert payload["session_key"] == "agent:main:telegram:supergroup:123" + assert payload["reason"] == "shutdown" + assert payload["data"]["text"] == "hello world" + + +def test_flush_writes_message_event_to_file(tmp_path, monkeypatch): + flush_dir = _make_flush_dir(tmp_path) + monkeypatch.setattr( + "gateway.shutdown_flush._get_flush_dir", lambda: flush_dir + ) + event = MagicMock() + event.text = "user message" + event.session_id = "20260728_120000_abc" + event.platform = "telegram" + event.sender_id = "456" + event.sender_name = "Alice" + event.reply_to = None + event.media = None + event.raw_event = None + + count = flush_pending_to_file({"session_key_1": event}, reason="adapter_shutdown") + assert count == 1 + files = list(flush_dir.glob("*.json")) + assert len(files) == 1 + payload = json.loads(files[0].read_text(encoding="utf-8")) + assert payload["data"]["text"] == "user message" + assert payload["data"]["session_id"] == "20260728_120000_abc" + + +def test_recover_no_flush_files_is_noop(tmp_path, monkeypatch): + flush_dir = _make_flush_dir(tmp_path) + monkeypatch.setattr( + "gateway.shutdown_flush._get_flush_dir", lambda: flush_dir + ) + mock_db = MagicMock() + assert recover_pending_to_db(mock_db) == 0 + mock_db.append_message.assert_not_called() + + +def test_recover_inserts_via_append_message_and_deletes_file(tmp_path, monkeypatch): + flush_dir = _make_flush_dir(tmp_path) + monkeypatch.setattr( + "gateway.shutdown_flush._get_flush_dir", lambda: flush_dir + ) + ts = int(time.time()) + # Write a flush file with session_id + payload = { + "session_key": "agent:main:telegram:supergroup:123", + "reason": "shutdown", + "ts": ts, + "data": { + "text": "lost message", + "session_id": "20260728_120000_abc", + }, + } + flush_file = flush_dir / "test_session_123.json" + flush_file.write_text(json.dumps(payload), encoding="utf-8") + + mock_db = MagicMock() + count = recover_pending_to_db(mock_db) + + assert count == 1 + mock_db.append_message.assert_called_once_with( + session_id="20260728_120000_abc", + role="user", + content="lost message", + timestamp=ts, + ) + assert not flush_file.exists() + + +def test_recover_skips_file_without_session_id(tmp_path, monkeypatch): + flush_dir = _make_flush_dir(tmp_path) + monkeypatch.setattr( + "gateway.shutdown_flush._get_flush_dir", lambda: flush_dir + ) + payload = { + "session_key": "some_key", + "reason": "shutdown", + "ts": int(time.time()), + "data": {"text": "no session id"}, + } + flush_file = flush_dir / "no_sid.json" + flush_file.write_text(json.dumps(payload), encoding="utf-8") + + mock_db = MagicMock() + count = recover_pending_to_db(mock_db) + + assert count == 0 + mock_db.append_message.assert_not_called() + # File preserved for manual recovery + assert flush_file.exists() + + +def test_recover_deletes_empty_text_file(tmp_path, monkeypatch): + flush_dir = _make_flush_dir(tmp_path) + monkeypatch.setattr( + "gateway.shutdown_flush._get_flush_dir", lambda: flush_dir + ) + payload = { + "session_key": "some_key", + "reason": "shutdown", + "ts": int(time.time()), + "data": {"text": "", "session_id": "sid"}, + } + flush_file = flush_dir / "empty.json" + flush_file.write_text(json.dumps(payload), encoding="utf-8") + + mock_db = MagicMock() + count = recover_pending_to_db(mock_db) + + assert count == 0 + assert not flush_file.exists() + + +def test_serialise_string(): + assert _serialise_value("hello") == {"text": "hello"} + + +def test_serialise_dict(): + assert _serialise_value({"text": "hi"}) == {"text": "hi"} + + +def test_serialise_object_with_text(): + obj = MagicMock() + obj.text = "msg" + obj.session_id = "sid" + obj.platform = None + obj.sender_id = None + obj.sender_name = None + obj.reply_to = None + obj.media = None + obj.raw_event = None + result = _serialise_value(obj) + assert result is not None + assert result["text"] == "msg" + assert result["session_id"] == "sid" + + +def test_get_flush_dir_uses_get_hermes_home(tmp_path, monkeypatch): + """Flush dir must use get_hermes_home(), not hardcoded Path.home().""" + import gateway.shutdown_flush as mod + + captured = {} + + def fake_get_hermes_home(): + from pathlib import Path + captured["called"] = True + return tmp_path + + monkeypatch.setattr( + "hermes_constants.get_hermes_home", fake_get_hermes_home + ) + result = mod._get_flush_dir() + assert captured.get("called") is True + assert result == tmp_path / "pending_messages"