From 6e4e0638e83bf35cab66d692c22cf801f73d7845 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=B5=B5=E6=A1=82=E9=9B=84?= Date: Sat, 26 Sep 2026 14:26:57 +0800 Subject: [PATCH] fix(agent): heal a session row deleted under a live agent (#123583) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `_ensure_db_session` trusts the cached `_session_db_created` flag as proof the row exists, and the flush path only retries row creation while that flag is False. Any store-side removal of the row under a live agent — `hermes sessions delete`, the Desktop/web delete, bulk prune, a profile-repair move, an in-place store rebuild — leaves the flag stale, so every later turn's append fails the FK and is dropped with one WARNING per turn. The agent keeps answering; the durable transcript silently stops growing, and because the failed transaction leaves no rows behind there is no post-hoc trace in the store. Classification (`hermes_state_errors.py`): add the `session_row_missing` cause — matched by `SQLITE_CONSTRAINT_FOREIGNKEY` (787) when the code survives, else the RPC-wrapped phrase — so the turn-end explanation names the real failure instead of "unknown". Heal (`agent/session_persistence.py::_db_flush_failed`): on that cause, drop the stale flag, reset the flush markers, call `_ensure_db_session()`, and replay once within the flush's existing `_adoption_budget`. Because the deletion erased the session's message rows too, the replay clears the per-message persisted markers so the FULL in-memory transcript lands on the recreated row, not just the current tail (mirrors `_db_flush_adopt_compression_tip`). If row creation also fails, return False without appending into a guaranteed rollback — fail-open, batch stays unmarked for the next flush. No new fail-closed path. (cherry picked from commit ad97047da0f9801a4fdca4478bfd68085f02c3a8) --- agent/session_persistence.py | 32 ++++- hermes_state_errors.py | 4 +- ...st_session_row_under_live_agent_persist.py | 122 ++++++++++++++++++ 3 files changed, 155 insertions(+), 3 deletions(-) create mode 100644 tests/agent/test_session_row_under_live_agent_persist.py 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