fix(agent): heal a session row deleted under a live agent (#123583)

agent/session_persistence.py

@@ -5,6 +5,7 @@
 
 import logging
 import re
+import sqlite3
 from contextlib import nullcontext
 
 from typing import Any, Dict, List, Optional, Tuple
@@ -296,14 +297,41 @@ def _db_flush_adopt_compression_tip(agent) -> bool:
     return True
 
 
-def _db_flush_failed(agent, e: Exception, batch_rows: List[Dict[str, Any]], adoption_budget: int) -> bool:
+def _db_flush_failed(agent, e: Exception, batch_rows: List[Dict[str, Any]], adoption_budget: int,
+                     messages: Optional[List[Dict]] = None) -> bool:
     """Classify a failed flush; True when the caller should retry once on an adopted compression tip."""
     agent._db_flush_scan_prefix = None  # full re-scan next flush: an exception mid-loop leaves mixed dispositions
     # The only place the SQLite error is visible before it becomes a bare False — classify it so the turn-end
     # explanation can distinguish lock contention from disk-full/read-only.
     from hermes_state import StateDbCorruptError, StateDbReplacedError, classify_persistence_error, divert_session_transcript_jsonl
     from hermes_state_errors import CompressionSessionClosedError
     agent._last_persistence_error_cause = classify_persistence_error(e)
+    if getattr(e, "sqlite_errorcode", None) == getattr(sqlite3, "SQLITE_CONSTRAINT_FOREIGNKEY", 787) \
+            or "foreign key constraint" in str(e).lower():
+        # The session row was removed under this live agent (`hermes sessions delete`, the Desktop/web
+        # delete, bulk prune, a profile-repair move, an in-place store rebuild — none visible to the
+        # cached agent, so the cached `_session_db_created` flag is stale and every later append hits
+        # the FK). The deletion already erased the session's message rows with it, so the durable
+        # transcript is empty: drop the stale flag, reset the flush markers, and replay the FULL
+        # in-memory transcript onto the recreated row — not just the current tail (#123583).
+        if adoption_budget <= 0:
+            return False
+        for msg in messages or ():
+            if isinstance(msg, dict):
+                msg.pop(_DB_PERSISTED_MARKER, None)
+        agent._flushed_db_message_ids = set()
+        agent._last_flushed_db_idx = 0
+        agent._session_db_created = False
+        agent._ensure_db_session()
+        if not agent._session_db_created:
+            # Row creation failed too (transient store trouble): don't append into a guaranteed
+            # rollback — keep the batch unmarked so the next flush retries the whole thing.
+            logger.warning("Session DB row for %s is missing and could not be recreated; will retry next flush",
+                           getattr(agent, "session_id", None))
+            return False
+        logger.warning("Session DB row for %s was removed under the live agent; recreated it and replaying the transcript",
+                       getattr(agent, "session_id", None))
+        return True
     if isinstance(e, (StateDbReplacedError, StateDbCorruptError)):
         # A replaced/quarantined handle will not take this batch again — keep it on disk.
         try:
@@ -415,7 +443,7 @@ def _flush_messages_to_session_db_unlocked(
             self._db_flush_scan_prefix = messages[:]
             return True
         except Exception as e:
-            if _db_flush_failed(self, e, batch_rows, _adoption_budget):
+            if _db_flush_failed(self, e, batch_rows, _adoption_budget, messages):
                 return self._flush_messages_to_session_db_unlocked(messages, conversation_history, _adoption_budget=0)
             return False
 

hermes_state_errors.py

@@ -98,7 +98,7 @@ def is_disk_full_error(exc: BaseException | str | None) -> bool:
 # Every classify_persistence_error bucket; consumers enumerate this tuple.
 PERSISTENCE_ERROR_CAUSES = (
     "locked", "compression", "compression_closed", "turn_lease", "corrupt", "fts_index",
-    "replaced", "deleted_wal", "disk", "unknown",
+    "replaced", "deleted_wal", "disk", "session_row_missing", "unknown",
 )
 
 
@@ -260,6 +260,8 @@ class StateDbCorruptError(sqlite3.DatabaseError):
     (("was replaced underneath",), "replaced"),
     (_DB_CORRUPTION_MARKERS, "corrupt"),
     (("locked", "busy"), "locked"),
+    # A flush rejected by the session-row FK: the row was removed under a live agent (#123583).
+    (("foreign key constraint failed",), "session_row_missing"),
 )
 
 

tests/agent/test_session_row_under_live_agent_persist.py

@@ -0,0 +1,122 @@
+"""A session row deleted under a live agent must heal on the next flush (#123583).
+
+Before the fix, ``_flush_messages_to_session_db`` trusts the cached
+``_session_db_created`` flag: after ``hermes sessions delete`` (or Desktop delete /
+bulk prune / profile-repair move / in-place store rebuild) removes the row, every
+later turn's append fails the FK and is dropped with one WARNING per turn — the
+durable transcript silently stops growing and leaves no trace in the store.
+
+Maintainer triage direction (issue #123583, maintainer pass): classify the FK
+failure as ``session_row_missing``, recreate the row, and replay the FULL
+in-memory transcript — not just the current tail.
+"""
+
+import os
+import tempfile
+from pathlib import Path
+from unittest.mock import patch
+
+
+def _make_agent(session_db, session_id):
+    with patch.dict(os.environ, {"OPENROUTER_API_KEY": "test-key"}):
+        from run_agent import AIAgent
+
+        return AIAgent(
+            api_key="test-key",
+            base_url="https://openrouter.ai/api/v1",
+            model="test/model",
+            quiet_mode=True,
+            session_db=session_db,
+            session_id=session_id,
+            skip_context_files=True,
+            skip_memory=True,
+        )
+
+
+def test_flush_recreates_row_deleted_under_live_agent():
+    from hermes_state import SessionDB
+
+    with tempfile.TemporaryDirectory() as tmpdir:
+        db = SessionDB(db_path=Path(tmpdir) / "test.db")
+        agent = _make_agent(db, "sess-live")
+
+        transcript = [
+            {"role": "user", "content": "turn one"},
+            {"role": "assistant", "content": "answer one"},
+        ]
+        agent._flush_messages_to_session_db(transcript, [])
+        assert len(db.get_messages("sess-live")) == 2
+        assert agent._session_db_created is True
+
+        # Row removed under the live agent: the same store API behind
+        # `hermes sessions delete`, the Desktop/web delete, and bulk prune.
+        assert db.delete_session("sess-live") is True
+
+        # The live transcript keeps growing: two more turns join the same list
+        # (marked dicts from the earlier flush + fresh tail), as in a real session.
+        transcript += [
+            {"role": "user", "content": "turn two"},
+            {"role": "assistant", "content": "answer two"},
+        ]
+        healed = agent._flush_messages_to_session_db(transcript, [])
+
+        assert healed is True
+        assert agent._last_persistence_error_cause == "session_row_missing"
+        rows = db.get_messages("sess-live")
+        assert len(rows) == 4, (
+            "Heal must replay the FULL in-memory transcript onto the recreated "
+            "row (4 messages), not just the current tail; a silent drop here is "
+            "the #123583 transcript-loss bug."
+        )
+        assert agent._session_db_created is True
+        db.close()
+
+
+def test_flush_recovers_when_row_deleted_between_turns_twice():
+    """The healed state is stable: a second deletion keeps healing, not just once."""
+    from hermes_state import SessionDB
+
+    with tempfile.TemporaryDirectory() as tmpdir:
+        db = SessionDB(db_path=Path(tmpdir) / "test.db")
+        agent = _make_agent(db, "sess-live-2")
+
+        agent._flush_messages_to_session_db([{"role": "user", "content": "a"}], [])
+        assert len(db.get_messages("sess-live-2")) == 1
+
+        for round_no in ("x", "y"):
+            assert db.delete_session("sess-live-2") is True
+            healed = agent._flush_messages_to_session_db(
+                [{"role": "user", "content": round_no}], []
+            )
+            assert healed is True
+            rows = db.get_messages("sess-live-2")
+            assert len(rows) == 1 and rows[-1]["role"] == "user"
+        db.close()
+
+
+def test_flush_fails_open_when_row_cannot_be_recreated(monkeypatch):
+    """Scenario B: if row creation fails too, the flush returns False instead of
+    appending into a guaranteed rollback — fail-open, batch stays unmarked."""
+    import sqlite3 as _sqlite3
+
+    from hermes_state import SessionDB
+
+    with tempfile.TemporaryDirectory() as tmpdir:
+        db = SessionDB(db_path=Path(tmpdir) / "test.db")
+        agent = _make_agent(db, "sess-gone")
+
+        agent._flush_messages_to_session_db([{"role": "user", "content": "a"}], [])
+        assert len(db.get_messages("sess-gone")) == 1
+        assert db.delete_session("sess-gone") is True
+
+        # Row creation inside the heal now raises (transient store trouble).
+        def _broken_create(*a, **kw):
+            raise _sqlite3.OperationalError("unable to open database file")
+
+        monkeypatch.setattr(db, "create_session", _broken_create)
+
+        healed = agent._flush_messages_to_session_db(
+            [{"role": "user", "content": "b"}], []
+        )
+        assert healed is False
+        assert agent._session_db_created is False