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

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

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]:

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
 
 

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

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()