diff --git a/hermes_state.py b/hermes_state.py
index 9b8215390d1e6..38b90ed7e4fd3 100644
--- a/hermes_state.py
+++ b/hermes_state.py
@@ -5478,6 +5478,80 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
         rowcount = self._execute_write(_do)
         return rowcount > 0
 
+    def set_session_read(self, session_id: str, read: bool = True) -> bool:
+        """Mark a session read or unread (and its whole compression lineage).
+
+        Read state is a watermark, not a flag: ``last_read_at`` records when
+        the conversation was last read, and it counts as unread when activity
+        postdates that watermark (the derived ``unread`` key on
+        :meth:`list_sessions_rich` rows). New messages therefore flip a read
+        conversation back to unread without any write on the message path.
+        Three states:
+
+        * NULL — never tracked (every pre-feature row): treated as read, so
+          shipping the column doesn't badge a user's entire history at once.
+        * 0 — explicitly marked unread: any activity postdates it.
+        * timestamp — read up to that moment.
+
+        Like :meth:`set_session_archived` / :meth:`set_session_pinned`, the
+        whole compression chain is stamped as a unit, so reading the surfaced
+        tip clears the root (and vice-versa) no matter which id the caller
+        holds. Returns True when at least one row changed.
+        """
+        def _do(conn):
+            cursor = conn.execute(
+                """
+                WITH RECURSIVE
+                  ancestors(id) AS (
+                    SELECT ?
+                    UNION
+                    SELECT parent.id
+                    FROM ancestors a
+                    JOIN sessions child ON child.id = a.id
+                    JOIN sessions parent ON parent.id = child.parent_session_id
+                    WHERE parent.end_reason = 'compression'
+                  ),
+                  descendants(id) AS (
+                    SELECT ?
+                    UNION
+                    SELECT child.id
+                    FROM descendants d
+                    JOIN sessions parent ON parent.id = d.id
+                    JOIN sessions child ON child.parent_session_id = parent.id
+                    WHERE parent.end_reason = 'compression'
+                  ),
+                  lineage(id) AS (
+                    SELECT id FROM ancestors
+                    UNION
+                    SELECT id FROM descendants
+                  )
+                UPDATE sessions
+                SET last_read_at = ?
+                WHERE id IN (SELECT id FROM lineage)
+                """,
+                (session_id, session_id, time.time() if read else 0.0),
+            )
+            rowcount = cursor.rowcount
+            if rowcount is None or rowcount < 0:
+                rowcount = conn.execute("SELECT changes()").fetchone()[0]
+            return rowcount
+        rowcount = self._execute_write(_do)
+        return rowcount > 0
+
+    @staticmethod
+    def session_unread(session_row: Dict[str, Any]) -> bool:
+        """Derive unread from a session row's watermark and activity.
+
+        Shared by ``list_sessions_rich`` and any future surface that holds a
+        row (or projected row) with ``last_read_at`` and ``last_active``.
+        NULL watermark = never tracked = read.
+        """
+        last_read = session_row.get("last_read_at")
+        if last_read is None:
+            return False
+        last_active = session_row.get("last_active") or session_row.get("started_at")
+        return float(last_active or 0) > float(last_read)
+
     def get_session_by_title(self, title: str) -> Optional[Dict[str, Any]]:
         """Look up a session by exact title. Returns session dict or None."""
         with self._read_ctx() as conn:
@@ -5994,6 +6068,13 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
                 projected.append(merged)
             sessions = projected
 
+        # Derive read state per surfaced conversation. ``last_read_at`` is
+        # lineage-stamped by set_session_read, so a projected row's root
+        # watermark and its tip's are the same value — comparing it against
+        # the tip's last_active is correct either way.
+        for s in sessions:
+            s["unread"] = self.session_unread(s)
+
         return sessions
 
     # =========================================================================
diff --git a/hermes_state_common.py b/hermes_state_common.py
index c520f1c51df63..4e66115f81780 100644
--- a/hermes_state_common.py
+++ b/hermes_state_common.py
@@ -245,6 +245,7 @@ CREATE TABLE IF NOT EXISTS sessions (
     rewind_count INTEGER NOT NULL DEFAULT 0,
     archived INTEGER NOT NULL DEFAULT 0,
     pinned INTEGER NOT NULL DEFAULT 0,
+    last_read_at REAL,
     FOREIGN KEY (parent_session_id) REFERENCES sessions(id),
     FOREIGN KEY (system_prompt_hash) REFERENCES system_prompts(hash)
 );
diff --git a/tests/hermes_state/test_session_read_state.py b/tests/hermes_state/test_session_read_state.py
new file mode 100644
index 0000000000000..94f229584cf44
--- /dev/null
+++ b/tests/hermes_state/test_session_read_state.py
@@ -0,0 +1,105 @@
+import time
+
+import pytest
+
+from hermes_state import SessionDB
+
+
+@pytest.fixture
+def db(tmp_path):
+    database = SessionDB(tmp_path / "state.db")
+    try:
+        yield database
+    finally:
+        database.close()
+
+
+def _last_read(db, sid):
+    row = db._conn.execute(
+        "SELECT last_read_at FROM sessions WHERE id = ?", (sid,)
+    ).fetchone()
+    return row["last_read_at"] if row is not None else None
+
+
+def _row(db, sid):
+    rows = db.list_sessions_rich(include_archived=True)
+    return next(s for s in rows if s["id"] == sid)
+
+
+def test_untracked_sessions_are_read(db):
+    """NULL watermark = never tracked = read, so shipping the column doesn't
+    badge a user's entire pre-feature history at once."""
+    db.create_session(session_id="s1", source="cli")
+    db.append_message(session_id="s1", role="user", content="hi")
+
+    assert _last_read(db, "s1") is None
+    assert _row(db, "s1")["unread"] is False
+
+
+def test_mark_read_then_new_activity_flips_back_to_unread(db):
+    db.create_session(session_id="s1", source="cli")
+    db.append_message(session_id="s1", role="user", content="hi")
+
+    assert db.set_session_read("s1") is True
+    assert _row(db, "s1")["unread"] is False
+
+    # New activity postdating the watermark makes it unread again without
+    # any write on the message path.
+    time.sleep(0.01)
+    db.append_message(session_id="s1", role="assistant", content="reply")
+    assert _row(db, "s1")["unread"] is True
+
+
+def test_mark_unread_explicitly(db):
+    db.create_session(session_id="s1", source="cli")
+    db.append_message(session_id="s1", role="user", content="hi")
+    db.set_session_read("s1")
+
+    assert db.set_session_read("s1", read=False) is True
+    assert _last_read(db, "s1") == 0.0
+    assert _row(db, "s1")["unread"] is True
+
+
+def test_missing_session_returns_false(db):
+    assert db.set_session_read("nope") is False
+
+
+def _compression_pair(db: SessionDB):
+    base = time.time() - 100
+    db.create_session("root", source="cli")
+    db.create_session("tip", source="cli", parent_session_id="root")
+    db._conn.execute(
+        "UPDATE sessions SET started_at = ?, ended_at = ?, end_reason = 'compression', message_count = 1 WHERE id = 'root'",
+        (base, base + 10),
+    )
+    db._conn.execute(
+        "UPDATE sessions SET started_at = ?, message_count = 1 WHERE id = 'tip'",
+        (base + 20,),
+    )
+    db._conn.commit()
+
+
+def test_reading_compression_tip_stamps_whole_lineage(db):
+    _compression_pair(db)
+
+    assert db.set_session_read("tip") is True
+
+    root_read = _last_read(db, "root")
+    assert root_read is not None and root_read > 0
+    assert root_read == _last_read(db, "tip")
+
+    # The projected conversation row (root surfaced as tip) derives read.
+    rows = db.list_sessions_rich(order_by_last_active=True)
+    assert [s["id"] for s in rows] == ["tip"]
+    assert rows[0]["unread"] is False
+
+
+def test_marking_root_unread_marks_projected_conversation(db):
+    _compression_pair(db)
+    db.set_session_read("tip")
+
+    assert db.set_session_read("root", read=False) is True
+
+    rows = db.list_sessions_rich(order_by_last_active=True)
+    assert [s["id"] for s in rows] == ["tip"]
+    assert rows[0]["unread"] is True