From 3154ccfe76e7ee08037ce3102fe161058b1e000e Mon Sep 17 00:00:00 2001 From: John Paul Soliva Date: Mon, 28 Sep 2026 14:31:18 +0900 Subject: [PATCH] fix(compression): a merged user dict names rows by provenance, not by text The commit resolved a marker-less user dict to a run of active rows whose joined text equals it. A prompt that was never persisted can carry the same text as a user;user pair another surface appended after the snapshot, so those unseen rows were archived as summarized instead of being cloned after the compacted set. The alternation repair now stamps the dict with how many durable rows it folded, the gateway replay entry carries that stamp, and the commit resolves a run only for a stamped dict of that exact width. Folding an unpersisted turn into the dict drops the stamp. (cherry picked from commit 13bbb4d1c4c13874bd825d212503f9d9539060f0) (cherry picked from commit df3fca28cba8a2758c5c711cc0ee582a8e374d51) --- agent/agent_runtime_helpers.py | 7 ++++++ agent/conversation_compression_archive.py | 3 +++ gateway/run.py | 3 +++ hermes_state_messages.py | 13 +++++++--- .../test_archive_merged_user_rows.py | 25 +++++++++++++++++++ 5 files changed, 47 insertions(+), 4 deletions(-) create mode 100644 tests/hermes_state/test_archive_merged_user_rows.py 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()