From 49e6d661a02076d9e46f2e61390605e558d20b41 Mon Sep 17 00:00:00 2001 From: Xipong <217837358+Xipong@users.noreply.github.com> Date: Wed, 9 Sep 2026 01:36:15 +0300 Subject: [PATCH] fix: bound compacted display history paging --- hermes_state_common.py | 19 +++- hermes_state_messages.py | 103 +++++++++++++++--- .../test_get_messages_include_compacted.py | 66 +++++++++++ 3 files changed, 172 insertions(+), 16 deletions(-) diff --git a/hermes_state_common.py b/hermes_state_common.py index 3868fb1979..935640dad4 100644 --- a/hermes_state_common.py +++ b/hermes_state_common.py @@ -363,7 +363,8 @@ CREATE TABLE IF NOT EXISTS messages ( compacted INTEGER NOT NULL DEFAULT 0, api_content TEXT, display_kind TEXT, - display_metadata TEXT + display_metadata TEXT, + display_order INTEGER ); CREATE TABLE IF NOT EXISTS session_model_usage ( @@ -513,6 +514,22 @@ CREATE INDEX IF NOT EXISTS idx_async_delegations_delivery DEFERRED_INDEX_SQL = """ CREATE INDEX IF NOT EXISTS idx_messages_session_active ON messages(session_id, active, timestamp); +CREATE INDEX IF NOT EXISTS idx_messages_display_page + ON messages(session_id, display_order, active DESC, id DESC) + WHERE active = 1 OR compacted = 1; +CREATE INDEX IF NOT EXISTS idx_messages_display_backfill + ON messages(session_id) WHERE display_order IS NULL AND (active = 1 OR compacted = 1); +CREATE TRIGGER IF NOT EXISTS messages_display_order_insert +AFTER INSERT ON messages WHEN new.display_order IS NULL +BEGIN + UPDATE messages SET display_order = COALESCE(( + SELECT MIN(COALESCE(display_order, id)) FROM messages + WHERE session_id = new.session_id AND id <> new.id + AND (active = 1 OR compacted = 1) + AND role IS new.role AND content IS new.content AND timestamp IS new.timestamp + AND tool_call_id IS new.tool_call_id AND tool_calls IS new.tool_calls AND tool_name IS new.tool_name + ), new.id) WHERE id = new.id; +END; CREATE INDEX IF NOT EXISTS idx_messages_active_null ON messages(active) WHERE active IS NULL; CREATE INDEX IF NOT EXISTS idx_sessions_session_key diff --git a/hermes_state_messages.py b/hermes_state_messages.py index c099bbc763..d3767d8ca5 100644 --- a/hermes_state_messages.py +++ b/hermes_state_messages.py @@ -249,6 +249,23 @@ class SessionMessagesMixin: _str_or_none(msg.get("api_content")), _str_or_none(msg.get("display_kind")), self._encode_display_metadata(msg.get("display_metadata"))) + def _inherit_composite_display_order(self, conn, row_id: int, session_id: str, msg: Dict[str, Any], + message_timestamp: float, tool_calls: Any) -> None: + handoff, live_view = split_user_originated_turn(msg) + if handoff is None or live_view is None: + return + encoded_content = self._encode_content(live_view.get("content")) + encoded_tool_calls = json.dumps(tool_calls) if tool_calls else None + row = conn.execute( + "SELECT MIN(COALESCE(display_order, id)) FROM messages " + "WHERE session_id = ? AND id <> ? AND (active = 1 OR compacted = 1) " + "AND role = 'user' AND content IS ? AND timestamp IS ? AND tool_call_id IS ? " + "AND tool_calls IS ? AND tool_name IS ?", + (session_id, row_id, encoded_content, message_timestamp, msg.get("tool_call_id"), + encoded_tool_calls, msg.get("tool_name"))).fetchone() + if row is not None and row[0] is not None: + conn.execute("UPDATE messages SET display_order = ? WHERE id = ?", (row[0], row_id)) + @staticmethod def _bump_session_counters(conn, session_id: str, inserted: int, tool_calls: int, *, unit: bool) -> None: """Bump sessions.* counters after an insert; *unit* bakes the ``+ 1`` literal into the SQL.""" @@ -278,12 +295,16 @@ class SessionMessagesMixin: # Encode outside the write txn (display metadata first: log-order parity). msg["display_metadata"] = self._encode_display_metadata(display_metadata) tool_calls = _parse_tool_calls(tool_calls) + message_timestamp = _coerce_timestamp(timestamp, time.time()) params = self._message_row_params( - session_id, role, msg, tool_calls, _coerce_timestamp(timestamp, time.time()), keep_reasoning=True) + session_id, role, msg, tool_calls, message_timestamp, keep_reasoning=True) def _do(conn): self._check_transcript_write_guards(conn, session_id, compression_lock_holder, turn_lease_holder=turn_lease_holder, turn_lease_ttl_seconds=turn_lease_ttl_seconds) msg_id = conn.execute(_INSERT_MESSAGE_SQL, params).lastrowid + if msg_id is not None: + self._inherit_composite_display_order( + conn, int(msg_id), session_id, msg, message_timestamp, tool_calls) self._bump_session_counters(conn, session_id, 1, _tool_calls_count(tool_calls), unit=True) return msg_id # THE critical write (failure aborts the turn): long patience so a sibling legitimately @@ -431,6 +452,8 @@ class SessionMessagesMixin: session_id, role, msg, tool_calls, message_timestamp, keep_reasoning=role == "assistant")) if cur.lastrowid is not None: msg["_row_id"] = cur.lastrowid + self._inherit_composite_display_order( + conn, int(cur.lastrowid), session_id, msg, message_timestamp, tool_calls) inserted += 1 tool_calls_total += _tool_calls_count(tool_calls) now_ts = max(now_ts, message_timestamp) + 1e-6 @@ -623,6 +646,19 @@ class SessionMessagesMixin: "AND role = 'user' AND active = 1 AND content IS ?", (_scrub_surrogates(api_content), row_id, session_id, self._encode_content(content))) + def _display_dedupe_key(self, row) -> Tuple[Any, ...]: + """Historical display identity, including normalized live content from user handoff carriers.""" + dedupe_content = row["content"] + if row["role"] == "user": + handoff, live_view = split_user_originated_turn({ + "role": "user", "content": self._decode_content(row["content"]), + "display_kind": row["display_kind"], + "display_metadata": self._decode_display_metadata(row["display_metadata"])}) + if handoff is not None and live_view is not None: + dedupe_content = self._encode_content(live_view.get("content")) + return (row["role"], dedupe_content, row["timestamp"], + row["tool_call_id"], row["tool_calls"], row["tool_name"]) + def _dedupe_display_generations(self, rows): """Collapse compaction generations so each logical message appears once (the protected tail is copied into each generation: same role/content/timestamp, different ``active``/id); prefer the live row, then @@ -630,18 +666,7 @@ class SessionMessagesMixin: seen: Dict[Tuple[Any, ...], Any] = {} first_id: Dict[Tuple[Any, ...], int] = {} for row in rows: - dedupe_content = row["content"] - if row["role"] == "user": - handoff, live_view = split_user_originated_turn({ - "role": "user", "content": self._decode_content(row["content"]), - "display_kind": row["display_kind"], - "display_metadata": self._decode_display_metadata(row["display_metadata"])}) - if handoff is not None and live_view is not None: - dedupe_content = self._encode_content(live_view.get("content")) - # Tool fields key too: identical tool messages collapse, distinct calls with equal - # role/content/timestamp never merge. - key = (row["role"], dedupe_content, row["timestamp"], - row["tool_call_id"], row["tool_calls"], row["tool_name"]) + key = self._display_dedupe_key(row) cur = seen.get(key) if cur is None or (row["active"], row["id"]) > (cur["active"], cur["id"]): seen[key] = row @@ -650,6 +675,36 @@ class SessionMessagesMixin: # copy in a newer generation has a higher id than messages emitted after the original. return [seen[key] for key in sorted(seen, key=first_id.__getitem__)] + def _ensure_display_order(self, session_id: str) -> bool: + """Backfill one legacy session once, preserving the pre-index display identity exactly.""" + missing_sql = ( + "SELECT 1 FROM messages WHERE session_id = ? AND (active = 1 OR compacted = 1) " + "AND display_order IS NULL LIMIT 1") + if self._read_one(missing_sql, (session_id,)) is None: + return True + if getattr(self, "read_only", False): + return False + + def _do(conn): + missing = conn.execute(missing_sql, (session_id,)).fetchone() + if missing is None: + return True + rows = conn.execute( + "SELECT * FROM messages WHERE session_id = ? AND (active = 1 OR compacted = 1) ORDER BY id", + (session_id,)).fetchall() + first_id: Dict[Tuple[Any, ...], int] = {} + keyed_rows = [] + for row in rows: + key = self._display_dedupe_key(row) + first_id[key] = min(first_id.get(key, row["id"]), row["id"]) + keyed_rows.append((row["id"], key)) + conn.executemany( + "UPDATE messages SET display_order = ? WHERE id = ?", + [(first_id[key], row_id) for row_id, key in keyed_rows]) + return True + + return bool(self._execute_write(_do)) + def _row_to_message_dict(self, row, *, warn_context: str, summary_flag: bool) -> Dict[str, Any]: """``dict(row)`` with content/tool_calls/display_metadata decoded; *summary_flag* keeps ``_compressed_summary`` only as ``True``.""" @@ -680,8 +735,26 @@ class SessionMessagesMixin: if after_id is not None and include_compacted: raise ValueError("after_id is incompatible with include_compacted (deduped display reads use offset paging)") active_clause = self._active_clause(include_inactive, include_compacted) - if include_compacted: - # Full display set (the UI row cap lives in the endpoint), dedupe, then page ([:None] is a no-op). + if include_compacted and not include_inactive and self._ensure_display_order(session_id): + direction = "DESC" if latest else "ASC" + sql = f"""WITH page AS ( + SELECT display_order FROM messages + WHERE session_id = ? AND (active = 1 OR compacted = 1) + GROUP BY display_order ORDER BY display_order {direction} + LIMIT ? OFFSET ? + ) + SELECT chosen.* FROM page + JOIN messages AS chosen ON chosen.id = ( + 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) + ORDER BY candidate.active DESC, candidate.id DESC LIMIT 1 + ) + ORDER BY page.display_order ASC""" + rows = self._read_all(sql, [session_id, -1 if limit is None else limit, offset, session_id]) + elif include_compacted: + # Read-only legacy stores cannot persist display identities; retain the exact old projection. rows = self._dedupe_display_generations(self._read_all( "SELECT * FROM messages WHERE session_id = ?" + active_clause + " ORDER BY id ASC", [session_id])) rows = rows[::-1][offset:][:limit][::-1] if latest else rows[offset:][:limit] diff --git a/tests/hermes_state/test_get_messages_include_compacted.py b/tests/hermes_state/test_get_messages_include_compacted.py index c7386ee35a..5e41f4a7b6 100644 --- a/tests/hermes_state/test_get_messages_include_compacted.py +++ b/tests/hermes_state/test_get_messages_include_compacted.py @@ -14,6 +14,13 @@ job of ``include_inactive`` (audit / debug reads). import pytest +from agent.context_compressor import ( + HISTORICAL_TASK_HEADING, + SUMMARY_PREFIX, + _MERGED_PRIOR_CONTEXT_HEADER, + _MERGED_SUMMARY_DELIMITER, + _SUMMARY_END_MARKER, +) from hermes_state import SessionDB @@ -140,6 +147,65 @@ class TestDisplayDedupe: db._execute_write(_do) + def test_latest_bounded_page_has_bounded_database_work(self, db): + for sid, count in (("small", 2_000), ("large", 20_000)): + db.create_session(sid, source="desktop") + db.append_messages_batch( + sid, + [{"role": "assistant", "content": f"row-{index}"} for index in range(count)], + chunk_rows=500, + ) + + db._wal_active = False + + def progress_steps(sid): + callbacks = 0 + + def progress(): + nonlocal callbacks + callbacks += 1 + return 0 + + db._conn.set_progress_handler(progress, 100) + try: + page = db.get_messages( + sid, include_compacted=True, latest=True, limit=120) + finally: + db._conn.set_progress_handler(None, 0) + assert len(page) == 120 + return callbacks + + small_steps = progress_steps("small") + large_steps = progress_steps("large") + assert large_steps < small_steps * 3 + + def test_composite_handoff_keeps_live_turn_identity_and_first_position(self, db): + sid = "composite" + db.create_session(sid, source="desktop") + original_id = db.append_message(sid, role="user", content="live ask", timestamp=100.0) + later_id = db.append_message(sid, role="assistant", content="later answer", timestamp=200.0) + db._execute_write(lambda conn: conn.execute( + "UPDATE messages SET active = 0, compacted = 1 WHERE session_id = ?", (sid,))) + carrier = ( + f"{_MERGED_PRIOR_CONTEXT_HEADER}\n" + "live ask\n\n" + f"{_MERGED_SUMMARY_DELIMITER}\n\n" + f"{SUMMARY_PREFIX}\n\n" + f"{HISTORICAL_TASK_HEADING}\nold work\n\n" + f"{_SUMMARY_END_MARKER}" + ) + carrier_id = db.append_message(sid, role="user", content=carrier, timestamp=100.0) + # Simulate rows written before display_order existed; lazy backfill must use + # the same normalized user-turn key as the historical Python projection. + db._execute_write(lambda conn: conn.execute( + "UPDATE messages SET display_order = NULL WHERE session_id = ?", (sid,))) + + messages = db.get_messages(sid, include_compacted=True) + + assert [message["id"] for message in messages] == [carrier_id, later_id] + assert messages[0]["content"] == carrier + assert original_id not in [message["id"] for message in messages] + def test_copied_protected_tail_is_surfaced_once(self, db): """A message copied across compaction epochs appears exactly once.""" sid = "s1"