diff --git a/agent/agent_runtime_helpers.py b/agent/agent_runtime_helpers.py index 72e714a0f6..5db6e7c449 100644 --- a/agent/agent_runtime_helpers.py +++ b/agent/agent_runtime_helpers.py @@ -593,6 +593,7 @@ def _prune_unanswered_tool_calls(messages: List[Dict]) -> Tuple[List[Dict], int] def _merge_consecutive_users(messages: List[Dict]) -> Tuple[List[Dict], int]: """Pass 3: merge consecutive plain-text user messages (no user input lost).""" from agent.context_compressor import _DB_PERSISTED_MARKER, split_user_originated_turn + from agent.conversation_compression_archive import MERGED_DURABLE_ROWS from hermes_state import SessionDB def _plain_text(content: Any) -> bool: @@ -620,6 +621,12 @@ def _merge_consecutive_users(messages: List[Dict]) -> Tuple[List[Dict], int]: (prev_content + "\n\n" + new_content) if prev_content and new_content else (prev_content or new_content) ) had_api_sidecar = "api_content" in prev + # Read before the marker is popped below. An unpersisted turn folded in ends the claim: + # the dict no longer stands for durable rows only. + if (prev.get(_DB_PERSISTED_MARKER) or prev.get(MERGED_DURABLE_ROWS)) and msg.get(_DB_PERSISTED_MARKER): + prev[MERGED_DURABLE_ROWS] = int(prev.get(MERGED_DURABLE_ROWS) or 1) + 1 + else: + prev.pop(MERGED_DURABLE_ROWS, None) prev["content"] = merged_content # The clean-text persist override must replace only the absorbed turn, never the # unanswered text before it; kept across replay passes (an empty turn absorbs too). diff --git a/agent/conversation_compression_archive.py b/agent/conversation_compression_archive.py index b73d624b2a..0b17f73c4f 100644 --- a/agent/conversation_compression_archive.py +++ b/agent/conversation_compression_archive.py @@ -11,6 +11,9 @@ from __future__ import annotations from typing import Any, Dict, List, Optional, Sequence, Tuple ABSORBED_ROW_IDS = "_absorbed_row_ids" +# How many durable rows an alternation repair folded into one user dict. A reload without row ids +# has no other way to tell that dict from a prompt that was never persisted. +MERGED_DURABLE_ROWS = "_merged_durable_rows" def _positive_id(value: Any) -> Optional[int]: diff --git a/gateway/run.py b/gateway/run.py index 016e1d6762..56d1c2299e 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -1156,6 +1156,9 @@ def _build_replay_entry( # flushes skip rows already in state.db (#121462/#123462). if msg.get("_db_persisted"): entry["_db_persisted"] = True + # A merged user dict lost that stamp; this one is what ties it to its durable rows at compaction. + if msg.get("_merged_durable_rows"): + entry["_merged_durable_rows"] = msg["_merged_durable_rows"] return entry diff --git a/hermes_state_messages.py b/hermes_state_messages.py index 50461e5403..f08228a26b 100644 --- a/hermes_state_messages.py +++ b/hermes_state_messages.py @@ -1000,9 +1000,13 @@ class SessionMessagesMixin: A reload without row ids turns a durable ``user;user`` pair (a prompt that never got its reply) into one dict equal to neither row. Read as an unpersisted turn, its rows would be re-sequenced after the compacted set like concurrent appends, behind the turn that is running. + Only a dict the repair stamped: a prompt that was never persisted can carry the same text as + rows another surface appended, and those were never held. """ - content = message.get("content") - if message.get("role") != "user" or not isinstance(content, str) or "\n\n" not in content: + from agent.conversation_compression_archive import MERGED_DURABLE_ROWS + + content, width = message.get("content"), message.get(MERGED_DURABLE_ROWS) + if message.get("role") != "user" or not isinstance(content, str) or type(width) is not int or width < 2: return [] rows = [ (int(row["id"]), row["role"], self._loaded_view_content(row["role"], self._decode_content(row["content"]))) @@ -1019,8 +1023,9 @@ class SessionMessagesMixin: merged = f"{merged}\n\n{part}" if merged and part else (merged or part) if not content.startswith(merged): break - if end > start and merged == content: - runs.append([rows[i][0] for i in range(start, end + 1)]) + if end - start + 1 == width: + if merged == content: + runs.append([rows[i][0] for i in range(start, end + 1)]) break if len(runs) > 1: return None diff --git a/tests/hermes_state/test_archive_merged_user_rows.py b/tests/hermes_state/test_archive_merged_user_rows.py new file mode 100644 index 0000000000..e634b77d6b --- /dev/null +++ b/tests/hermes_state/test_archive_merged_user_rows.py @@ -0,0 +1,25 @@ +"""A held dict may only name rows the compressor could have been handed. + +The commit resolves a merged user dict to the durable ``user;user`` run behind it. A prompt that was +never persisted can carry the same text as rows another surface appended after the snapshot. +""" + +from hermes_state import SessionDB + + +def test_unpersisted_prompt_leaves_rows_appended_after_the_snapshot_live(tmp_path): + db = SessionDB(tmp_path / "state.db") + db.create_session("sid", source="test") + held = [db.append_message("sid", "user", "U1"), db.append_message("sid", "assistant", "A1")] + watermark = db.get_active_message_watermark("sid") + db.append_message("sid", "user", "same first paragraph") + db.append_message("sid", "user", "same second paragraph") + prompt = {"role": "user", "content": "same first paragraph\n\nsame second paragraph"} + + db.archive_and_compact( + "sid", [{"role": "user", "content": "summary"}, {"role": "assistant", "content": "ok"}, prompt], + watermark=watermark, covered_ids=held, unresolved_held=[prompt]) + + assert [row["content"] for row in db.get_messages("sid")] == [ + "summary", "ok", prompt["content"], "same first paragraph", "same second paragraph"] + db.close()