fix(state): reactions resolve rows across the compression lineage, not the tip alone

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

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

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]) == []