diff --git a/agent/session_persistence.py b/agent/session_persistence.py index 46674dcd75..6279f4e3d5 100644 --- a/agent/session_persistence.py +++ b/agent/session_persistence.py @@ -5,6 +5,7 @@ import hashlib import logging import re +import sqlite3 from contextlib import nullcontext from typing import Any, Dict, List, Optional, Tuple @@ -296,7 +297,8 @@ 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 @@ -304,6 +306,32 @@ def _db_flush_failed(agent, e: Exception, batch_rows: List[Dict[str, Any]], adop 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 @@ class SessionPersistenceMixin: 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 diff --git a/hermes_state_errors.py b/hermes_state_errors.py index 4528a31755..09c667c9db 100644 --- a/hermes_state_errors.py +++ b/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 @@ _PERSISTENCE_CAUSE_BY_PHRASE = ( (("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"), ) diff --git a/tests/agent/test_session_row_under_live_agent_persist.py b/tests/agent/test_session_row_under_live_agent_persist.py new file mode 100644 index 0000000000..2de689f9f7 --- /dev/null +++ b/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