fix: rewrite recover_pending_to_db to use SessionDB.append_message

Critical fixes to salvaged PR #73020:
- Use SessionDB.append_message instead of raw INSERT INTO messages.
  The original used wrong column names (session_key/created_at vs
  session_id/timestamp) and bypassed FTS indexing, session metadata
  updates, display_kind, and all other columns append_message handles.
- Use get_hermes_home() instead of hardcoded Path.home()/'.hermes'.
  Profile-aware path resolution under HERMES_HOME override and active
  profile isolation.
- Add 11 tests covering flush, recovery, serialisation, edge cases.

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 ``<hermes_home>/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(

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"