diff --git a/hermes_state.py b/hermes_state.py index a6a0dc7441..420c840014 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -776,6 +776,14 @@ def _strip_background_review_harness( # Matches a bare protocol/tool-name marker such as "[memory]" or "[skill_manage]". _STALE_TOOL_CALL_MARKER_RE = re.compile(r"^\[[A-Za-z_][A-Za-z0-9_.-]*\]$") +# Intrinsic persistence marker stamped on message dicts that are known-durable. +# MUST stay in sync with run_agent._DB_PERSISTED_MARKER and +# agent.context_compressor._DB_PERSISTED_MARKER (same literal; defined locally +# because hermes_state must not import run_agent — circular import — and +# importing the agent layer for one constant inverts the state/agent layering). +# Drift is guarded by test_marker_constant_in_sync (#92231). +_DB_PERSISTED_MARKER_KEY = "_db_persisted" + def _is_stale_tool_call_marker_message(msg: Dict[str, Any]) -> bool: """True when ``msg`` is a persisted assistant turn whose content is a bare @@ -11090,6 +11098,18 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin) if row["role"] in {"user", "assistant"} and isinstance(content, str): content = sanitize_context(content).strip() msg = {"role": row["role"], "content": content} + # Born durable (#92231): this dict is materialized FROM a durable + # row, so stamp the persistence marker at the source instead of + # relying on every restore caller to thread the loaded list back + # through a flush as ``conversation_history=`` — any + # identity-losing handoff (compression's durable-snapshot + # adoption, incremental persists with no history arg) would + # otherwise re-append the ENTIRE transcript on flush. + # Underscore-prefixed like ``_row_id``: every transport strips it + # before the wire, and compression's assembly copies deliberately + # strip it so rotated child handoffs still flush (see + # _fresh_compaction_message_copy). + msg[_DB_PERSISTED_MARKER_KEY] = True # Durable per-message identity for surfaces that need to address a # specific row later (desktop reactions). OPT-IN: only the gateway # asks for it — every other consumer (ACP restore, export, diff --git a/tests/acp/test_session.py b/tests/acp/test_session.py index 3d8bc19089..7922846725 100644 --- a/tests/acp/test_session.py +++ b/tests/acp/test_session.py @@ -326,6 +326,9 @@ class TestPersistence: assert restored is not None msg = restored.history[0] assert isinstance(msg.pop("timestamp", None), (int, float)) + # Load-time durability stamp (#92231): rows materialized from the DB + # are marked persisted so a later flush can't re-append them. + assert msg.pop("_db_persisted", None) is True assert restored.history == [{ "role": "assistant", "content": "hello", diff --git a/tests/agent/test_load_time_durability_stamp_92231.py b/tests/agent/test_load_time_durability_stamp_92231.py new file mode 100644 index 0000000000..32803071c2 --- /dev/null +++ b/tests/agent/test_load_time_durability_stamp_92231.py @@ -0,0 +1,147 @@ +"""Regression (#92231): resumed transcripts must not re-append to state.db. + +Root cause: ``get_messages_as_conversation`` / ``get_resume_conversations`` +returned plain dicts WITHOUT ``_DB_PERSISTED_MARKER``. Any flush that received +that loaded history without a matching ``conversation_history=`` identity +boundary (compression durable-snapshot adoption, incremental tool-call +persists, rotation preflight on cold resume) treated every loaded row as new +and re-appended the ENTIRE transcript. Compression cycles then doubled the +copies: the incident session grew 998 → 1995 → 3990 → 7981 rows across three +aborted rotations (15,962 active rows / 472 distinct contents). + +Fix: rows are stamped durable at materialization time in +``SessionDB._rows_to_conversation`` — a dict built FROM a durable row is +persisted by construction, no matter which caller loads it or how it is later +handed to a flush. +""" + +from __future__ import annotations + +from pathlib import Path +from types import SimpleNamespace + +from hermes_state import SessionDB, _DB_PERSISTED_MARKER_KEY +from run_agent import AIAgent + + +def _make_flush_agent(db: SessionDB, session_id: str): + """Minimal agent shell that owns the real flush implementation.""" + agent = SimpleNamespace( + _session_db=db, + _session_db_created=True, + _persist_disabled=False, + session_id=session_id, + _session_persist_lock=None, + _flushed_db_message_ids=set(), + _flushed_db_message_session_id=None, + _last_flushed_db_idx=0, + _persist_user_message_idx=None, + _persist_user_message_override=None, + _persist_user_message_timestamp=None, + _pending_cli_user_message=None, + ) + agent._ensure_db_session = lambda: None + agent._flush_messages_to_session_db = ( + AIAgent._flush_messages_to_session_db.__get__(agent, AIAgent) + ) + agent._flush_messages_to_session_db_unlocked = ( + AIAgent._flush_messages_to_session_db_unlocked.__get__(agent, AIAgent) + ) + return agent + + +def _seed_session(db: SessionDB, sid: str, turns: int = 3) -> None: + db.create_session(sid, source="cli") + for i in range(turns): + db.append_message(sid, "user", f"question {i}") + db.append_message(sid, "assistant", f"answer {i}") + + +def test_marker_constant_in_sync() -> None: + """hermes_state cannot import run_agent (circular) — literals must match.""" + import agent.context_compressor as cc + import run_agent + + assert _DB_PERSISTED_MARKER_KEY == run_agent._DB_PERSISTED_MARKER + assert _DB_PERSISTED_MARKER_KEY == cc._DB_PERSISTED_MARKER + + +def test_loaded_rows_are_stamped_durable(tmp_path: Path) -> None: + db = SessionDB(db_path=tmp_path / "state.db") + _seed_session(db, "S1") + + loaded = db.get_messages_as_conversation("S1") + assert loaded + assert all(m.get(_DB_PERSISTED_MARKER_KEY) is True for m in loaded) + + model_history, display_history = db.get_resume_conversations("S1") + assert model_history and display_history + assert all(m.get(_DB_PERSISTED_MARKER_KEY) is True for m in model_history) + assert all(m.get(_DB_PERSISTED_MARKER_KEY) is True for m in display_history) + + +def test_repeated_identityless_flushes_do_not_amplify(tmp_path: Path) -> None: + """The #92231 shape: reload + flush cycles must keep the row count flat. + + Each cycle simulates what compression's durable-snapshot adoption (or any + identity-losing handoff) used to do: load the transcript fresh from the DB + and flush it with NO ``conversation_history`` boundary. Pre-fix each cycle + doubled the row count (998 → 1995 → 3990 → 7981 in the incident session). + """ + db = SessionDB(db_path=tmp_path / "state.db") + _seed_session(db, "S2", turns=4) + baseline = len(db.get_messages("S2")) + + for _ in range(3): + loaded = db.get_messages_as_conversation("S2") + agent = _make_flush_agent(db, "S2") # fresh agent: no identity state + agent._flush_messages_to_session_db(loaded) + + assert len(db.get_messages("S2")) == baseline + + +def test_new_tail_after_loaded_history_still_flushes(tmp_path: Path) -> None: + """Guard against over-skipping: only loaded rows are exempt, new turns write.""" + db = SessionDB(db_path=tmp_path / "state.db") + _seed_session(db, "S3", turns=1) + + loaded = db.get_messages_as_conversation("S3") + live = [ + *loaded, + {"role": "user", "content": "new question"}, + {"role": "assistant", "content": "new answer"}, + ] + agent = _make_flush_agent(db, "S3") + agent._flush_messages_to_session_db(live) + + contents = [m.get("content") for m in db.get_messages("S3")] + assert contents == ["question 0", "answer 0", "new question", "new answer"] + + +def test_compaction_copy_strips_stamp_so_child_flush_writes(tmp_path: Path) -> None: + """Rotation handoff: compression copies must stay flushable to the child. + + ``_fresh_compaction_message_copy`` / ``_strip_persistence_markers`` remove + the marker from assembled compaction output so the rotation flush WRITES + the compacted transcript to the child session (#57491). Load-stamping must + not defeat that: a loaded row that goes through the compaction copy is + written to the child exactly once. + """ + from agent.context_compressor import _fresh_compaction_message_copy + + db = SessionDB(db_path=tmp_path / "state.db") + _seed_session(db, "PARENT", turns=2) + db.create_session("CHILD", source="cli", parent_session_id="PARENT") + + loaded = db.get_messages_as_conversation("PARENT") + compacted = [_fresh_compaction_message_copy(m) for m in loaded] + assert all(_DB_PERSISTED_MARKER_KEY not in m for m in compacted) + + agent = _make_flush_agent(db, "CHILD") + agent._flush_messages_to_session_db(compacted) + child_rows = db.get_messages("CHILD") + assert len(child_rows) == len(loaded) + + # Idempotent thereafter: the flush stamped the copies on write. + agent._flush_messages_to_session_db(compacted) + assert len(db.get_messages("CHILD")) == len(loaded) diff --git a/tests/agent/test_session_rotation_flush_cold_resume_68454.py b/tests/agent/test_session_rotation_flush_cold_resume_68454.py index c7650fa631..48c29e0178 100644 --- a/tests/agent/test_session_rotation_flush_cold_resume_68454.py +++ b/tests/agent/test_session_rotation_flush_cold_resume_68454.py @@ -50,8 +50,15 @@ def _contents(rows): return [r.get("content") for r in rows] -def test_rotation_flush_without_history_boundary_duplicates(tmp_path: Path) -> None: - """Control: bare flush of unstamped cold-resume rows double-writes (#68454).""" +def test_rotation_flush_without_history_boundary_is_safe(tmp_path: Path) -> None: + """Bare flush of cold-resumed rows must NOT double-write (#68454 → #92231). + + Historically this was a control test asserting the double-write: loaded + rows carried no ``_DB_PERSISTED_MARKER``, so a flush without + ``conversation_history=`` re-appended the whole transcript. Since #92231 + the loaders stamp the marker at row-materialization time, so even the + "wrong" call shape (no history boundary) is idempotent. + """ db = SessionDB(db_path=tmp_path / "state.db") sid = "COLD_ROTATE_DUP" db.create_session(sid, source="cli") @@ -63,7 +70,7 @@ def test_rotation_flush_without_history_boundary_duplicates(tmp_path: Path) -> N agent._flush_messages_to_session_db(loaded) # missing conversation_history= rows = db.get_messages_as_conversation(sid, include_inactive=True) - assert _contents(rows).count("persisted question") == 2 + assert _contents(rows) == ["persisted question", "persisted answer"] def test_rotation_flush_with_history_boundary_is_noop(tmp_path: Path) -> None: diff --git a/tests/test_message_reactions.py b/tests/test_message_reactions.py index acc4aaac03..309693929a 100644 --- a/tests/test_message_reactions.py +++ b/tests/test_message_reactions.py @@ -164,12 +164,20 @@ def test_row_id_is_opt_in_and_never_reaches_the_provider(session, db): """Only include_row_ids=True consumers see _row_id — and it's underscore- prefixed so transports strip it before the wire even for them. Default consumers (ACP restore, export) get the transcript in its historical shape. + + ``_db_persisted`` is the other sanctioned underscore key: stamped on every + loaded row (#92231) so a flush can never re-append a resumed transcript. + Like ``_row_id`` it is stripped before the wire by every transport. """ key, _rows = session for message in db.get_messages_as_conversation(key): assert "_row_id" not in message + assert message.get("_db_persisted") is True for message in db.get_messages_as_conversation(key, include_row_ids=True): assert "_row_id" in message - assert all(not k.startswith("_") or k == "_row_id" for k in message) + assert all( + not k.startswith("_") or k in {"_row_id", "_db_persisted"} + for k in message + )