diff --git a/agent/context_compressor.py b/agent/context_compressor.py index f711b6291c..99add9a9ae 100644 --- a/agent/context_compressor.py +++ b/agent/context_compressor.py @@ -298,6 +298,9 @@ COMPRESSED_SUMMARY_HAS_USER_TURN_KEY = "_compressed_summary_has_user_turn" # Only micro markers may be superseded/defragged/rehydrated: a batch marker's # content is NOT in the rolling micro summary, so rewriting one destroys history. MICRO_COMPACT_MARKER_KEY = "_micro_compact_marker" +# ``display_metadata`` flag on a row the model reads but nobody typed as one message (micro-compaction's +# merge of adjacent user turns). Its source rows stay in display history, so display projections skip it. +MODEL_ONLY_DISPLAY_METADATA_KEY = "model_only" # Intrinsic marker stamped on a message dict once it has been written to the SQLite session store. Used by # ``_flush_messages_to_session_db`` to decide what is already durable. An object-identity (``id(msg)``) # dedup set cannot be trusted across turns: once a flushed message dict is dropped from the live list (e.g. diff --git a/agent/micro_compaction.py b/agent/micro_compaction.py index c815bc14be..718162ec5f 100644 --- a/agent/micro_compaction.py +++ b/agent/micro_compaction.py @@ -426,6 +426,10 @@ class MicroCompactionMixin: if _plain_user(msg) and _plain_user(prev): prev["content"] = "\n\n".join(c for c in (prev["content"], msg["content"]) if c) drop_stale_api_content(prev) # merged content invalidates the api_content sidecar + # The originals stay in display history as compacted rows; showing the join too + # would paint every merged input twice on resume. + prev["display_metadata"] = {**(prev.get("display_metadata") or {}), + _cc().MODEL_ONLY_DISPLAY_METADATA_KEY: True} # The merge rewrites a live dict that may carry _db_persisted: pop the stamp # and flag the finalizer to invalidate the bounded flush-scan cursor, or the # merged text is identity-skipped and never reaches state.db. Same contract diff --git a/hermes_state_messages.py b/hermes_state_messages.py index a1aa9ed637..168923f538 100644 --- a/hermes_state_messages.py +++ b/hermes_state_messages.py @@ -11,8 +11,8 @@ import time from typing import Any, Dict, List, Optional, Tuple from agent.context_compressor import ( - _DB_PERSISTED_MARKER as _DB_PERSISTED_MARKER_KEY, _is_checkpoint_item, _newest_checkpoint_carrier, - split_user_originated_turn) + _DB_PERSISTED_MARKER as _DB_PERSISTED_MARKER_KEY, MODEL_ONLY_DISPLAY_METADATA_KEY, _is_checkpoint_item, + _newest_checkpoint_carrier, split_user_originated_turn) from agent.memory_manager import sanitize_context from agent.message_sanitization import _sanitize_surrogates from hermes_cli.timefmt import coerce_epoch @@ -45,6 +45,10 @@ _BUMP_GENERATION_SQL = """ _TURN_LEASE_ROW_SQL = "SELECT holder, expires_at FROM session_turn_leases WHERE conversation_id = ?" _DELETE_COMPRESSION_LOCK_SQL = "DELETE FROM compression_locks WHERE session_id = ? AND holder = ?" _DISPLAY_ACTIVE_CLAUSE = " AND (active = 1 OR compacted = 1)" +# Model-only rows (see MODEL_ONLY_DISPLAY_METADATA_KEY) never enter a display projection. Unqualified on +# purpose: inside a correlated subquery it binds to the innermost ``messages`` alias. +DISPLAY_VISIBLE_SQL = ( + f" AND COALESCE({_sql_json_extract('display_metadata', '$.' + MODEL_ONLY_DISPLAY_METADATA_KEY)}, 0) = 0") _DISPLAY_META_ROW_SQL = "SELECT display_metadata FROM messages WHERE id = ? AND session_id IN ({ids})" + _DISPLAY_ACTIVE_CLAUSE # A display row is indexed only when both halves are set; the read path backfills before projecting, so # the in-transaction delete fence must refuse (not project) any session this probe still matches. @@ -876,6 +880,10 @@ class SessionMessagesMixin: return (row["role"], dedupe_content, row["timestamp"], row["tool_call_id"], row["tool_calls"], row["tool_name"]) + def _is_model_only_row(self, row) -> bool: + """Python twin of :data:`DISPLAY_VISIBLE_SQL`.""" + return bool((self._decode_display_metadata(row["display_metadata"]) or {}).get(MODEL_ONLY_DISPLAY_METADATA_KEY)) + @staticmethod def _display_identity(key: Tuple[Any, ...]) -> bytes: """Fixed-width durable identity for indexed display-generation lookup.""" @@ -888,6 +896,8 @@ class SessionMessagesMixin: seen: Dict[Tuple[Any, ...], Any] = {} first_id: Dict[Tuple[Any, ...], int] = {} for row in rows: + if self._is_model_only_row(row): + continue key = self._display_dedupe_key(row) cur = seen.get(key) if cur is None or (row["active"], row["id"]) > (cur["active"], cur["id"]): @@ -957,6 +967,8 @@ class SessionMessagesMixin: f"WHERE session_id = ?{active_clause} ORDER BY id ASC", (session_id,)) for row in rows: + if self._is_model_only_row(row): + continue identity = self._display_identity(self._display_dedupe_key(row)) current = representatives.get(identity) candidate = (row["active"], row["id"]) @@ -1011,7 +1023,7 @@ class SessionMessagesMixin: return conn.execute( f"""WITH page AS ( SELECT display_order FROM messages - WHERE session_id = ? AND (active = 1 OR compacted = 1) + WHERE session_id = ? AND (active = 1 OR compacted = 1){DISPLAY_VISIBLE_SQL} GROUP BY display_order ORDER BY display_order {direction} LIMIT ? OFFSET ? ) @@ -1020,7 +1032,7 @@ class SessionMessagesMixin: SELECT candidate.id FROM messages AS candidate WHERE candidate.session_id = ? AND candidate.display_order = page.display_order - AND (candidate.active = 1 OR candidate.compacted = 1) + AND (candidate.active = 1 OR candidate.compacted = 1){DISPLAY_VISIBLE_SQL} ORDER BY candidate.active DESC, candidate.id DESC LIMIT 1 ) ORDER BY page.display_order ASC""", diff --git a/hermes_state_timeline.py b/hermes_state_timeline.py index 272bbf6556..a305cd03e5 100644 --- a/hermes_state_timeline.py +++ b/hermes_state_timeline.py @@ -7,6 +7,7 @@ from contextlib import contextmanager from agent.compaction_display import project_compaction_message_for_display from agent.context_compressor import user_originated_turn_view +from hermes_state_messages import DISPLAY_VISIBLE_SQL _SYNTHETIC_PROMPT = re.compile( @@ -56,10 +57,10 @@ def _display_rows_sql(conn, session_id, *, users_only=False): SQLite; only user content crosses the Python boundary for carrier normalization. Current stores use the durable display index, including protected-tail copies. """ - role = " AND role = 'user'" if users_only else "" + filters = (" AND role = 'user'" if users_only else "") + DISPLAY_VISIBLE_SQL indexed = conn.execute( "SELECT 1 FROM messages WHERE session_id = ? AND (active = 1 OR compacted = 1) " - f"{role} AND (display_order IS NULL OR display_identity IS NULL) LIMIT 1", + f"{filters} AND (display_order IS NULL OR display_identity IS NULL) LIMIT 1", (session_id,), ).fetchone() is None if indexed: @@ -67,16 +68,16 @@ def _display_rows_sql(conn, session_id, *, users_only=False): SELECT (SELECT candidate.id FROM messages candidate WHERE candidate.session_id = :sid AND candidate.display_order = m.display_order - AND (candidate.active = 1 OR candidate.compacted = 1) + AND (candidate.active = 1 OR candidate.compacted = 1){DISPLAY_VISIBLE_SQL} ORDER BY candidate.active DESC, candidate.id DESC LIMIT 1) AS row_id, m.display_order AS sort_id - FROM messages m WHERE session_id = :sid AND (active = 1 OR compacted = 1){role} + FROM messages m WHERE session_id = :sid AND (active = 1 OR compacted = 1){filters} GROUP BY m.display_order )""" return f"""WITH ranked AS ( SELECT id, MIN(id) OVER identity AS sort_id, ROW_NUMBER() OVER (identity ORDER BY active DESC, id DESC) AS preference - FROM messages WHERE session_id = :sid AND (active = 1 OR compacted = 1){role} + FROM messages WHERE session_id = :sid AND (active = 1 OR compacted = 1){filters} WINDOW identity AS (PARTITION BY role, CASE WHEN role = 'user' THEN timeline_identity_content(content, display_kind) ELSE content END, timestamp, tool_call_id, tool_calls, tool_name) diff --git a/tests/agent/test_micro_compaction.py b/tests/agent/test_micro_compaction.py index fa75a1b7a5..ce001a4c30 100644 --- a/tests/agent/test_micro_compaction.py +++ b/tests/agent/test_micro_compaction.py @@ -816,3 +816,46 @@ class TestMergeAdjacentUserTurnsPersistedMarker: assert merged[0]["content"] == "first\n\nsecond" assert _DB_PERSISTED_MARKER not in merged[0] assert cc._flush_scan_cursor_invalidated is True + + +def test_superseding_marker_never_shows_a_user_input_twice_in_display_history(tmp_path): + """A supersede merges the adjacent user turns for the model; the originals stay in display + history as compacted rows, so no display projection (resume, REST page, legacy page, prompt + timeline) may also paint the merged row. The model view still holds each input exactly once.""" + from agent.context_compressor import _DB_PERSISTED_MARKER + from hermes_state import SessionDB + from hermes_state_timeline import get_session_timeline + + db = SessionDB(db_path=tmp_path / "state.db") + db.create_session("s", source="cli") + messages = _conversation(exchanges=8) + for i, msg in enumerate(messages): + msg["timestamp"] = 1000.0 + i + msg["_row_id"] = db.append_message("s", role=msg["role"], content=msg["content"], timestamp=msg["timestamp"]) + msg[_DB_PERSISTED_MARKER] = True + cc = _compressor() + cc._session_db, cc._session_id = db, "s" + for _ in range(4): + messages = cc._micro_compact(messages) + assert len(_summary_markers(messages)) == 1 + assert any("question 0\n\nquestion 1" in str(m.get("content")) for m in messages), "no supersede merge happened" + + def shown(msgs): + return sorted(t for m in msgs if m.get("role") == "user" for t in str(m["content"]).split("\n\n")) + + typed = sorted(m["content"] for m in _conversation(exchanges=8) if m["role"] == "user") + model, display = db.get_resume_conversations("s") + assert shown(display) == typed + assert shown(model) == typed + assert shown(db.get_messages("s", include_compacted=True)) == typed + timeline = [e["preview"] for e in get_session_timeline(db, "s")["entries"]] + assert sorted(timeline) == typed + # Legacy stores (no display index, read-only so no backfill) take the payload-bounded page. + db._execute_write(lambda conn: conn.execute("UPDATE messages SET display_order = NULL")) + legacy = SessionDB(db_path=tmp_path / "state.db", read_only=True) + try: + assert shown(legacy.get_messages("s", include_compacted=True)) == typed + assert sorted(e["preview"] for e in get_session_timeline(legacy, "s")["entries"]) == typed + finally: + legacy.close() + db.close() diff --git a/website/docs/developer-guide/micro-compaction.md b/website/docs/developer-guide/micro-compaction.md index 27a332b687..3c3f319086 100644 --- a/website/docs/developer-guide/micro-compaction.md +++ b/website/docs/developer-guide/micro-compaction.md @@ -128,6 +128,12 @@ matters more than it sounds: leaving them in place stacks near-duplicate copies of the same text, each with its own heading and end-marker scaffolding, and the transcript grows on every turn instead of shrinking. +Dropping a marker leaves the user turns on either side of it adjacent, so they +are joined into one user message for the model. That joined row is flagged +`display_metadata.model_only`: the originals stay in display history as compacted +rows, and every display projection (resume, paged history, prompt timeline) +skips the join, so a resumed session shows each input exactly once. + ### Defrag Merge into a summary often enough and it gets baggy — repetitive, and larger