fix(tui): drop cached unread rows the server no longer lists

CHANGELOG.md

@@ -30,6 +30,9 @@ Upgrade notes:
   message listed under a superseded room and its successor counts once). The request is made only
   when a room moved. The vendored API snapshot is refreshed with it (#147).
 
+- The rail drops cached unread rows for topics the server's latest `GET /api/v1/groups/unread`
+  answer no longer lists, instead of keeping them until a local read passes them. The rail shows
+  dots only; `unread_total` is not displayed (#155).
 - A server that answers 404 to `GET /api/v1/groups/unread` is left alone for ten minutes instead of
   being asked on every poll, and the room list's ETag is kept so unchanged polls are 304 again
   (#154).

src/pepper_cli/tui/app.py

@@ -760,7 +760,12 @@ def _on_rail(self, update: poller.RailUpdate) -> None:
     def _apply_groups(self, groups: GroupList, update: poller.RailUpdate) -> None:
         had = self.marks.mentioned_topics(self.rooms) if self._bell else set()
         self.rooms = groups.groups
-        self.marks.merge_unread(update.unread, update.server_marks, total=update.unread_total)
+        self.marks.merge_unread(
+            update.unread,
+            update.server_marks,
+            total=update.unread_total,
+            snapshot=update.unread_answered,
+        )
         self.marks.settle(self.rooms)
         if self._bell and (self.marks.mentioned_topics(self.rooms) - had - {self.current.topic_id}):
             self.bell()

src/pepper_cli/tui/poller.py

@@ -59,6 +59,10 @@ class RailUpdate:
     unread: list[GroupUnreadRow] = field(default_factory=list)
     #: The server's deduped total, to show as it is; never a sum of ``unread``.
     unread_total: int | None = None
+    #: The server answered the unread request, so ``unread`` is its whole snapshot: a topic it
+    #: no longer lists is no longer unread. ``False`` when none was made or it failed, where an
+    #: empty ``unread`` means "not asked", not "nothing unread".
+    unread_answered: bool = False
     etag: str | None = None
     seen: dict[str, _Signature] | None = None
     #: The newest :meth:`RailPoller.refetch_all` request this step answered by reading every
@@ -142,6 +146,7 @@ def step(self) -> RailUpdate:
             server_marks=fetched.marks if fetched else {},
             unread=fetched.unread if fetched else [],
             unread_total=fetched.unread_total if fetched else None,
+            unread_answered=fetched is not None,
             etag=etag if complete else None,
             seen=seen,
             full=asked if full and complete else 0,

src/pepper_cli/tui/state.py

@@ -176,21 +176,32 @@ def merge_unread(
         marks: Mapping[str, str],
         *,
         total: int | None = None,
+        snapshot: bool = False,
     ) -> None:
         """Take the server's unread rows (and the cursors answered with them).
 
+        With ``snapshot`` the rows are the server's whole answer, so a topic it no longer lists
+        loses its cached row (otherwise rows only accumulate and a topic the server dropped
+        stays as it was last counted). Leave it off for a partial or absent answer.
+
         Every row is kept, even one counted from a cursor older than one held here (an answer
         that crossed a read on this device or another): it may count a message newer than that
         read, and dropping it would hide the message until the room moved again. Each row
         remembers the cursor it was counted from, and :meth:`settle` applies only a cursor that
         moved after it.
         """
         self.merge_server(marks)
+        rows = list(rows)
+        if snapshot:
+            listed = {row.topic_id for row in rows}
+            for topic_id in self.unread.keys() - listed:
+                del self.unread[topic_id]
+                self.counted_from.pop(topic_id, None)
         for row in rows:
             self.unread[row.topic_id] = row
             self.counted_from[row.topic_id] = parse_ts(marks.get(row.topic_id))
-        if total is not None:
-            self.unread_total = total
+        if total is not None or snapshot:
+            self.unread_total = total  # a snapshot without a total leaves none, not a stale one
 
     def merge_advanced(self, payload: object) -> str | None:
         """Fold a ``read_state.advanced`` event payload (``{topic_id, group_id, last_read_at}``)

tests/test_tui_poller.py

@@ -164,6 +164,21 @@ def test_a_failed_unread_request_is_asked_again_on_the_next_poll() -> None:
     assert _unread_reads(server) == 2
 
 
+def test_the_update_says_whether_the_unread_snapshot_was_answered() -> None:
+    server = ts.standard_server()
+    with server.client() as api:
+        rail = poller.RailPoller(api)
+        first = rail.step()
+        assert first.unread_answered is True
+        rail.commit(first)
+        server.bump()
+        unmoved = rail.step()  # a new ETag but no room moved: no unread request
+        assert unmoved.groups is not None and unmoved.unread_answered is False
+    server.fail["/api/v1/groups/unread"] = 502
+    with server.client() as api:
+        assert poller.RailPoller(api).step().unread_answered is False
+
+
 def test_a_404_on_the_unread_route_backs_off_and_keeps_the_etag(monkeypatch) -> None:
     server = ts.standard_server()
     server.fail["/api/v1/groups/unread"] = 404

tests/test_tui_state.py

@@ -116,6 +116,24 @@ def test_a_cursor_the_server_counted_from_does_not_settle_its_answer() -> None:
     assert not marks.is_unread(topic)
 
 
+def test_a_snapshot_drops_rows_the_server_no_longer_lists() -> None:
+    marks = state.ReadMarks()
+    marks.merge_unread([_row("t1", 2), _row("t2", 1)], {"t1": "2026-01-01T00:00:00Z"})
+    marks.merge_unread([_row("t2", 1)], {}, snapshot=True)
+    assert set(marks.unread) == {"t2"}
+    assert "t1" not in marks.counted_from
+    marks.merge_unread([_row("t2", 1)], {}, total=1, snapshot=True)
+    marks.merge_unread([_row("t2", 1)], {}, snapshot=True)
+    assert marks.unread_total is None  # the snapshot did not carry one: no stale total
+
+
+def test_a_partial_answer_keeps_rows_it_does_not_list() -> None:
+    marks = state.ReadMarks()
+    marks.merge_unread([_row("t1", 2), _row("t2", 1)], {})
+    marks.merge_unread([], {})  # nothing asked: not "nothing unread"
+    assert set(marks.unread) == {"t1", "t2"}
+
+
 def test_the_unread_total_is_the_servers_and_a_settled_row_does_not_change_it() -> None:
     (room,) = _rooms(ts.room(ts.ROOM_A, "A", [ts.topic("t1", "One", "2026-02-01T00:00:00Z")]))
     marks = state.ReadMarks()