Sync: a flag set on the server left the thread counters stale

Reported from the running app: a message showed as unread in the list — bold,
blue dot — while the reading pane showed it as read, and only a manual "mark as
read" reconciled them.

The two views read different things. The pane reads the message's own flags; the
list reads `threads.unread_count`, denormalized from them. `apply_delta` updated
message flags with a bare UPDATE and never recomputed the thread, so a flag set
anywhere but in this app — read on a phone, filed by a server-side rule — moved
one and not the other. Auto-mark-on-read could not fix it either: it sees a
message that is already \Seen and correctly does nothing, leaving the counter
wrong forever.

Expunges had the same hole, which also left `message_count` overstated.

`apply_delta` now collects the affected threads and re-denormalizes them after
the deletes, so a thread emptied by an expunge is removed rather than recounted.

Migration 0018 repairs rows that already drifted — recomputed from the messages,
so it is a no-op on a healthy database. One thread was stale here; installs that
have been syncing longer will have more.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018SqJYt7jD7oUEkoY68sBZo

backend/app/sync/engine.py

@@ -23,6 +23,7 @@ from app.providers.imap_smtp import ImapSmtpProvider
 from app.search.factory import get_search_backend
 from app.sync.folder_names import infer_special_use
 from app.sync.provider_factory import build_provider
+from app.threading_.threader import denormalize_thread
 
 logger = logging.getLogger(__name__)
 
@@ -371,22 +372,42 @@ async def apply_delta(
             if r is not None:
                 results.append(r)
             max_uid = max(max_uid, remote.uid)
+        # Threads whose denormalized counters this batch invalidates. A flag set
+        # on the server (read on a phone, filed by a server-side rule) and an
+        # expunge both change what `threads.unread_count` should say, and neither
+        # goes through apply_flags_to_messages — so without this the row keeps its
+        # old count and the list shows a message as unread that the reading pane
+        # shows as read.
+        touched_threads: set[uuid.UUID] = set()
+
         for change in delta.flag_changes:
             await session.execute(
                 update(Message)
                 .where(Message.mailbox_id == mailbox.id, Message.uid == change.uid)
                 .values(flags=sorted(change.flags))
             )
-        if delta.expunged_uids:
+        if delta.flag_changes:
-            # Server-side expunges (QRESYNC/VANISHED, SPEC §6.3). Capture the row ids
+            touched_threads.update(
-            # first so we can drop them from the search index after the delete.
-            gone_ids = list(
                 await session.scalars(
-                    select(Message.id).where(
+                    select(Message.thread_id).where(
-                        Message.mailbox_id == mailbox.id, Message.uid.in_(delta.expunged_uids)
+                        Message.mailbox_id == mailbox.id,
+                        Message.uid.in_([c.uid for c in delta.flag_changes]),
+                        Message.thread_id.is_not(None),
                     )
                 )
             )
+        if delta.expunged_uids:
+            # Server-side expunges (QRESYNC/VANISHED, SPEC §6.3). Capture the row ids
+            # first so we can drop them from the search index after the delete.
+            gone = (
+                await session.execute(
+                    select(Message.id, Message.thread_id).where(
+                        Message.mailbox_id == mailbox.id, Message.uid.in_(delta.expunged_uids)
+                    )
+                )
+            ).all()
+            gone_ids = [row[0] for row in gone]
+            touched_threads.update(row[1] for row in gone if row[1] is not None)
             await session.execute(
                 delete(Message).where(
                     Message.mailbox_id == mailbox.id, Message.uid.in_(delta.expunged_uids)
@@ -398,6 +419,11 @@ async def apply_delta(
                 .where(Draft.account_id == account.id, Draft.imap_uid.in_(delta.expunged_uids))
                 .values(imap_uid=None)
             )
+        # After the deletes, so a thread left with no messages is removed rather
+        # than recounted (denormalize_thread handles that case).
+        for tid in touched_threads:
+            await denormalize_thread(session, tid)
+
         updates = {"last_seen_uid": max_uid}
         if delta.new_highestmodseq is not None:
             updates["highestmodseq"] = delta.new_highestmodseq

backend/migrations/versions/0018_repair_thread_counters.py

@@ -0,0 +1,55 @@
+"""repair thread counters left stale by server-side flag changes and expunges
+
+`threads.unread_count`, `pinned` and `message_count` are denormalized from the
+thread's messages. Two sync paths updated messages without recomputing them: a
+flag change reported by the server (mail read on a phone, or filed by a
+server-side rule) and a QRESYNC expunge. The row then disagreed with its own
+messages — the message list showing a message as unread that the reading pane
+showed as read, and only a manual "mark as read" reconciling it.
+
+The code paths are fixed; this repairs rows that already drifted. Recomputes from
+the messages themselves, so it is safe to run on a healthy database: threads that
+already agree are left untouched.
+
+Revision ID: 0018
+Revises: 0017
+Create Date: 2026-09-18
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+from alembic import op
+
+revision: str = "0018"
+down_revision: str | None = "0017"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.execute(
+        r"""
+        UPDATE threads t
+           SET unread_count  = s.unread,
+            pinned        = s.pinned,
+            message_count = s.total
+        FROM (
+            SELECT thread_id,
+                   count(*)                                        AS total,
+                   count(*) FILTER (WHERE NOT ('\Seen' = ANY(flags)))    AS unread,
+                   bool_or('\Flagged' = ANY(flags))                AS pinned
+            FROM messages
+            WHERE thread_id IS NOT NULL
+            GROUP BY thread_id
+        ) AS s
+        WHERE t.id = s.thread_id
+          AND (t.unread_count, t.pinned, t.message_count) IS DISTINCT FROM
+              (s.unread, s.pinned, s.total)
+        """
+    )
+
+
+def downgrade() -> None:
+    """Nothing to undo: this only makes rows agree with the messages they describe."""

backend/tests/sync/test_secondary_poll.py

@@ -241,3 +241,79 @@ async def test_apply_delta_expunges_vanished_uids() -> None:  # type: ignore[no-
             await cleanup.execute(delete(User).where(User.id == user.id))
         await session.close()
         await engine.dispose()
+
+
+async def test_server_side_flag_change_updates_the_thread_counters() -> None:
+    """A flag set elsewhere (phone, webmail, server rule) must re-denormalize.
+
+    `threads.unread_count` is what the message list renders, while the reading
+    pane renders the message's own flags. When a delta updated flags without
+    recomputing the thread, the two disagreed forever: the list kept showing a
+    message as unread that the pane showed as read, and only a manual "mark as
+    read" — which goes through a different path — reconciled them.
+    """
+    from app.models import Thread
+    from app.threading_.threader import denormalize_thread
+
+    engine = create_async_engine(get_settings().database_url, poolclass=NullPool)
+    sm = async_sessionmaker(engine, expire_on_commit=False)
+    session = sm()
+    user = User(handle=f"fc{uuid.uuid4().hex[:8]}@test.local", password_hash=hash_password("pw"))
+    session.add(user)
+    await session.flush()
+    account = Account(
+        user_id=user.id,
+        email="me@example.com",
+        imap_config={},
+        smtp_config={},
+        secret_ct=b"",
+        discovery_src="manual",
+    )
+    session.add(account)
+    await session.flush()
+    inbox = Mailbox(
+        account_id=account.id, user_id=user.id, path="INBOX", special_use="inbox", last_seen_uid=5
+    )
+    thread = Thread(user_id=user.id, subject_norm="t", last_message_at=datetime.now(UTC))
+    session.add_all([inbox, thread])
+    await session.flush()
+    message = Message(
+        user_id=user.id,
+        account_id=account.id,
+        mailbox_id=inbox.id,
+        thread_id=thread.id,
+        uid=5,
+        internal_date=datetime.now(UTC),
+        blob_sha256="0" * 64,
+        flags=[],
+    )
+    session.add(message)
+    await session.flush()
+    await denormalize_thread(session, thread.id)
+    await session.commit()
+    assert (await session.get(Thread, thread.id)).unread_count == 1
+
+    class FakeProvider:
+        qresync_enabled = True
+
+        async def close(self) -> None:
+            return None
+
+    class _FlagChange:
+        uid = 5
+        flags = ["\\Seen"]
+
+    delta = MailboxDelta([], [_FlagChange()], [], None, None)
+    await sync_engine.apply_delta(FakeProvider(), account, inbox, delta)  # type: ignore[arg-type]
+
+    try:
+        session.expunge_all()
+        refreshed = await session.get(Thread, thread.id)
+        assert refreshed.unread_count == 0, (
+            "the server said the message was read; the thread counter did not follow"
+        )
+    finally:
+        async with session_scope() as cleanup:
+            await cleanup.execute(delete(User).where(User.id == user.id))
+        await session.close()
+        await engine.dispose()