hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 32 Packages Projects Releases Wiki Activity

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

Browse Source 
`_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)
This commit is contained in:
赵桂雄
2026-09-26 14:26:57 +08:00
committed by kshitij
parent 41303b379e
commit 6e4e0638e8
3 changed files with 155 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
agent/session_persistence.py
View File

@@ -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

4
hermes_state_errors.py
View File

@@ -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"),
)

122
tests/agent/test_session_row_under_live_agent_persist.py Normal file
View File

@@ -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