From fc5424d00e8a4c660fd98c38dcdf4b2c14640bfe Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Fri, 18 Sep 2026 23:41:16 +0530 Subject: [PATCH] fix(state): reactions resolve rows across the compression lineage, not the tip alone MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A display resume materializes the whole compression lineage with row ids (`get_resume_conversations(include_ancestors=True)`), so the desktop shows — and lets the user react to — rows that live in an ended parent segment. The gateway's `session_key` is re-anchored to the continuation after every compaction, and `set_message_reaction` scoped the row by that exact key, so every reaction on a pre-compaction message returned None and the desktop surfaced RPC 4040 "message not found in this session" (#80670: the 802 compacted-row repro; the agent-side `react_to_message` tool hit the same wall in #108633). `set_message_reaction` / `get_message_reactions` / `take_unseen_reactions` now scope by `_resume_lineage_ids(session_id)` — the same set the resume loads, so an explicit /branch copy still owns only its own rows and an unrelated session's row stays foreign. The RPC handler and the tool are unchanged: ownership is decided once, at the row. Lineage ownership was first identified in #108635 by @KoNit-K (tool path); the compacted-row half of the unseen-reaction scan is @Liuzikaii's #108542, cherry-picked ahead of this commit. --- hermes_state_messages.py | 19 +++++++++++----- tests/agent/test_message_reactions.py | 31 +++++++++++++++++++++++++++ 2 files changed, 45 insertions(+), 5 deletions(-) diff --git a/hermes_state_messages.py b/hermes_state_messages.py index 51ff31545d..13458d6acf 100644 --- a/hermes_state_messages.py +++ b/hermes_state_messages.py @@ -36,7 +36,7 @@ _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)" -_DISPLAY_META_ROW_SQL = "SELECT display_metadata FROM messages WHERE id = ? AND session_id = ?" +_DISPLAY_META_ROW_SQL = "SELECT display_metadata FROM messages WHERE id = ? AND session_id IN ({ids})" _ACTIVE_IDS_SQL = "SELECT id FROM messages WHERE session_id = ? AND active = 1 ORDER BY id" _SET_COUNTERS_SQL = "UPDATE sessions SET message_count = ?, tool_call_count = ?" _RESET_COUNTERS_SQL = "UPDATE sessions SET message_count = 0, tool_call_count = 0 WHERE id = ?" @@ -394,8 +394,9 @@ class SessionMessagesMixin: ``None`` for a foreign row.""" if not session_id or message_row_id is None: return None + sql, params = self._reaction_row_query(session_id, message_row_id) def _do(conn): - row = conn.execute(_DISPLAY_META_ROW_SQL, (message_row_id, session_id)).fetchone() + row = conn.execute(sql, params).fetchone() if row is None: return None meta = self._decode_display_metadata(row[0]) or {} @@ -416,21 +417,29 @@ class SessionMessagesMixin: """Reaction list persisted on one message row (never ``None``).""" if not session_id or message_row_id is None: return [] - row = self._read_one(_DISPLAY_META_ROW_SQL, (message_row_id, session_id)) + row = self._read_one(*self._reaction_row_query(session_id, message_row_id)) return self._reaction_list(self._decode_display_metadata(row[0])) if row is not None else [] + def _reaction_row_query(self, session_id: str, message_row_id: int) -> Tuple[str, tuple]: + """A reaction addresses a row the client can SEE, and a display resume materializes the whole + compression lineage with row ids — so a row is "in this session" when its owner is any lineage + segment, not only the tip. Explicit ``/branch`` copies keep their own rows (``_resume_lineage_ids``).""" + lineage = self._resume_lineage_ids(session_id) + return _DISPLAY_META_ROW_SQL.format(ids=_placeholders(lineage)), (message_row_id, *lineage) + def take_unseen_reactions(self, session_id: str, *, author: str = "user") -> List[Dict[str, Any]]: """Return *author*'s not-yet-surfaced reactions and mark them seen. Reactions are announced on the NEXT user turn (never by rewriting the reacted message: cache-safe); ``seen`` makes it exactly once. Include compaction-archived history that remains visible, but exclude rewound/superseded rows.""" if not session_id: return [] + lineage = self._resume_lineage_ids(session_id) def _do(conn): pending = [] for row in conn.execute("SELECT id, role, content, display_metadata FROM messages " - "WHERE session_id = ? AND (active = 1 OR compacted = 1) " + f"WHERE session_id IN ({_placeholders(lineage)}) AND (active = 1 OR compacted = 1) " "AND display_metadata IS NOT NULL ORDER BY id", - (session_id,)).fetchall(): + tuple(lineage)).fetchall(): meta = self._decode_display_metadata(row["display_metadata"]) reactions = meta.get(self.REACTIONS_METADATA_KEY) if meta else None if not isinstance(reactions, list): diff --git a/tests/agent/test_message_reactions.py b/tests/agent/test_message_reactions.py index 6028c521d4..0419c5e419 100644 --- a/tests/agent/test_message_reactions.py +++ b/tests/agent/test_message_reactions.py @@ -187,3 +187,34 @@ def test_row_id_is_opt_in_and_never_reaches_the_provider(session, db): not k.startswith("_") or k in {"_row_id", "_db_persisted"} for k in message ) + + +def test_ancestor_rows_react_through_the_continuation_key(session, db): + """A display resume shows the whole compression lineage with row ids, so a row in an ended parent + segment must react, read back and announce through the CONTINUATION key the client holds (#80670).""" + parent, rows = session + assert db.try_acquire_compression_lock(parent, "w", ttl_seconds=60) + db.publish_compression_child(parent_session_id=parent, child_session_id="react-tip", source="test", + messages=[{"role": "user", "content": "summary"}], compression_lock_holder="w") + display_rows = [m["_row_id"] for m in db.get_resume_conversations("react-tip")[1]] + assert rows[1] in display_rows + + assert db.set_message_reaction("react-tip", rows[1], "👍") == db.get_message_reactions("react-tip", rows[1]) + assert db.get_message_reactions("react-tip", rows[1])[0]["emoji"] == "👍" + assert [p["row_id"] for p in db.take_unseen_reactions("react-tip")] == [rows[1]] + + +def test_lineage_scope_still_rejects_foreign_and_branch_rows(session, db): + """Lineage widening must not turn row ids into a global lookup: an unrelated session's row and a row + of an explicit /branch copy (which keeps its own rows) stay foreign to the continuation.""" + parent, rows = session + assert db.try_acquire_compression_lock(parent, "w", ttl_seconds=60) + db.publish_compression_child(parent_session_id=parent, child_session_id="react-tip", source="test", + messages=[{"role": "user", "content": "summary"}], compression_lock_holder="w") + other = db.create_session("elsewhere", "test") + foreign = db.append_message(other, "user", "other conversation") + branch = db.create_session("branch", "test", parent_session_id=parent, model_config={"_branched_from": parent}) + + assert db.set_message_reaction("react-tip", foreign, "👍") is None + assert db.set_message_reaction(branch, rows[0], "👍") is None + assert db.get_message_reactions(branch, rows[0]) == []