collapse cross-account dupes, Lato UI font, mail-list selection & density tuning

Backend:
- Collapse identical cross-account deliveries in a thread: new messages.delivery_id
  (migration 0011) preserves the real Message-ID even when message_id_hdr is
  synthesized (§7.1); get_thread groups by it and reports also_in_account_ids; the
  list count counts distinct deliveries. Backfill script for existing rows.

Frontend:
- Bundle Lato (SIL OFL) as the UI font; unread mail-list rows use Black weight.
- Mail-list selection: hover-avatar checkbox, tri-state "select all", bulk
  Archive/Pin/Delete (Restore/Delete in Archive/Trash/Spam) with a count in the
  reading pane; unread blue dot; "also delivered to" note in the thread.
- Spacing/size tuning: sender-column width, icon gaps, avatar size, bar heights,
  and a round of -1px font sizes across the sidebar, mail list, and reading pane.

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

backend/app/api/schemas.py

@@ -56,6 +56,8 @@ class MessageEnvelope(BaseModel):
     has_remote_images: bool
     size_bytes: int
     snippet: str
+    # Other accounts this same delivery also landed in (collapsed duplicates).
+    also_in_account_ids: list[uuid.UUID] = []
 
 
 class ThreadDetail(BaseModel):

backend/app/api/threads.py

@@ -6,7 +6,7 @@ import uuid
 from typing import Annotated
 
 from fastapi import APIRouter, Depends, Query
-from sqlalchemy import func, literal, or_, select, tuple_
+from sqlalchemy import Text, func, literal, or_, select, tuple_
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.api.cursors import decode_cursor, encode_cursor
@@ -131,8 +131,11 @@ async def list_threads(
     # keeps the list count in step with a single deleted message.
     counts: dict[uuid.UUID, int] = {}
     if threads:
+        # count DISTINCT deliveries so collapsed cross-account copies count once,
+        # matching what get_thread renders.
+        delivery = func.coalesce(Message.delivery_id, func.cast(Message.id, Text))
         count_stmt = (
-            select(Message.thread_id, func.count())
+            select(Message.thread_id, func.count(func.distinct(delivery)))
             .join(Mailbox, Mailbox.id == Message.mailbox_id)
             .where(Message.user_id == user.id, Message.thread_id.in_([t.id for t in threads]))
             .group_by(Message.thread_id)
@@ -196,11 +199,34 @@ async def get_thread(
             or_(Mailbox.special_use.is_(None), Mailbox.special_use.notin_(_REMOVAL_FOLDERS))
         )
     messages = list(await db.scalars(stmt))
+
+    # Collapse duplicate deliveries: the same email (same delivery_id) that landed
+    # in more than one of the user's accounts shows once, noting the other accounts.
+    # Messages without a real Message-ID (delivery_id is NULL) never collapse.
+    reps: dict[str, Message] = {}
+    also: dict[str, list[uuid.UUID]] = {}
+    ordered: list[Message] = []
+    for m in messages:
+        key = m.delivery_id or f"id:{m.id}"
+        rep = reps.get(key)
+        if rep is None:
+            reps[key] = m
+            also[key] = []
+            ordered.append(m)
+        elif m.account_id != rep.account_id and m.account_id not in also[key]:
+            also[key].append(m.account_id)
+
+    envelopes = []
+    for m in ordered:
+        env = _envelope(m)
+        env.also_in_account_ids = also[m.delivery_id or f"id:{m.id}"]
+        envelopes.append(env)
+
     order = "desc" if sort == "desc" else "asc"
     return ThreadDetail(
         id=thread.id,
         subject=thread.subject_norm or "(no subject)",
         order=order,
         pinned=thread.pinned,
-        messages=[_envelope(m) for m in messages],
+        messages=envelopes,
     )

backend/app/indexer/indexer.py

@@ -161,6 +161,9 @@ async def persist_message(
         mailbox_id=mailbox_id,
         uid=remote.uid,
         message_id_hdr=message_id_hdr,
+        # Keep the real delivered Message-ID (may differ from the stored, possibly
+        # synthesized, message_id_hdr) so cross-account copies collapse in threads.
+        delivery_id=parsed.message_id,
         in_reply_to=parsed.in_reply_to,
         references_hdr=parsed.references,
         subject=parsed.subject,

backend/app/models.py

@@ -157,6 +157,10 @@ class Message(Base):
         UUID(as_uuid=True), ForeignKey("threads.id")
     )
     message_id_hdr: Mapped[str | None] = mapped_column(Text)
+    # The real header Message-ID as delivered (kept even when message_id_hdr is
+    # synthesized for a cross-account duplicate, §7.1). Identifies one delivery, so
+    # the same email fanned out to several accounts collapses to one row in a thread.
+    delivery_id: Mapped[str | None] = mapped_column(Text)
     in_reply_to: Mapped[str | None] = mapped_column(Text)
     references_hdr: Mapped[list[str]] = mapped_column(
         ARRAY(Text), nullable=False, server_default=text("'{}'")

backend/migrations/versions/0011_message_delivery_id.py

@@ -0,0 +1,38 @@
+"""add messages.delivery_id
+
+The real header Message-ID as delivered, kept even when message_id_hdr is
+synthesized for a cross-account duplicate (SPEC §7.1). It identifies a single
+delivery, so the same email delivered to several of the user's accounts can be
+collapsed to one row in the thread view. Backfilled from message_id_hdr for
+non-synthetic ids; synthetic rows are recovered from the raw blob by a script.
+
+Revision ID: 0011
+Revises: 0010
+Create Date: 2026-07-22
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+from alembic import op
+
+revision: str = "0011"
+down_revision: str | None = "0010"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.execute("ALTER TABLE messages ADD COLUMN delivery_id text")
+    # Non-synthetic ids are the real Message-ID already; use them directly.
+    op.execute(
+        "UPDATE messages SET delivery_id = message_id_hdr "
+        "WHERE message_id_hdr IS NOT NULL AND message_id_hdr NOT LIKE '%@synthetic.local'"
+    )
+    op.execute("CREATE INDEX ix_messages_delivery ON messages (user_id, delivery_id)")
+
+
+def downgrade() -> None:
+    op.execute("DROP INDEX IF EXISTS ix_messages_delivery")
+    op.execute("ALTER TABLE messages DROP COLUMN IF EXISTS delivery_id")

backend/scripts/backfill_delivery_id.py

@@ -0,0 +1,35 @@
+"""One-off: recover delivery_id for messages whose stored Message-ID was
+synthesized (§7.1 cross-account duplicate). The real Message-ID is still in the
+raw blob, so re-parse it. Non-synthetic rows were already backfilled by migration
+0011; here we only touch rows left with delivery_id IS NULL that have a real
+Message-ID in their raw bytes."""
+
+import asyncio
+
+from sqlalchemy import select
+
+from app.db import session_scope
+from app.indexer import blobstore
+from app.indexer.mime import parse_message
+from app.models import Message
+
+
+async def main() -> None:
+    async with session_scope() as session:
+        rows = list(
+            await session.scalars(select(Message).where(Message.delivery_id.is_(None)))
+        )
+        fixed = 0
+        for m in rows:
+            try:
+                raw = blobstore.load(m.user_id, m.blob_sha256)
+            except FileNotFoundError:
+                continue
+            parsed = parse_message(raw)
+            if parsed.message_id:
+                m.delivery_id = parsed.message_id
+                fixed += 1
+        print(f"scanned {len(rows)} null rows, set delivery_id on {fixed}")
+
+
+asyncio.run(main())

backend/tests/api/test_api.py

@@ -314,6 +314,44 @@ async def test_folder_counts(api: ApiCtx) -> None:
     assert body["inbox_unread"] == 2
 
 
+async def test_thread_collapses_cross_account_duplicate(api: ApiCtx) -> None:
+    assert api.session is not None
+    # The kickoff thread's first message; give it a real delivery_id.
+    kickoff = next(t for t in (await api.client.get("/api/threads")).json()["threads"]
+                   if t["message_count"] == 2)
+    detail = (await api.client.get(f"/api/threads/{kickoff['id']}")).json()
+    first_id = uuid.UUID(detail["messages"][0]["id"])
+    first = await api.session.get(Message, first_id)
+    assert first is not None
+    first.delivery_id = "<shared@x.com>"
+
+    # A second account receives the SAME email (same delivery_id) into the same thread.
+    acct2 = Account(
+        user_id=api.user_id, email="me2@example.com", imap_config={}, smtp_config={},
+        secret_ct=b"", discovery_src="manual",
+    )
+    api.session.add(acct2)
+    await api.session.flush()
+    inbox2 = Mailbox(account_id=acct2.id, user_id=api.user_id, path="INBOX", special_use="inbox")
+    api.session.add(inbox2)
+    await api.session.flush()
+    api.session.add(Message(
+        user_id=api.user_id, account_id=acct2.id, mailbox_id=inbox2.id, uid=1,
+        thread_id=first.thread_id, message_id_hdr="<synthetic@synthetic.local>",
+        delivery_id="<shared@x.com>", subject=first.subject,
+        internal_date=datetime.now(UTC), blob_sha256="z" * 64,  # later -> first stays the rep
+    ))
+    await api.session.commit()
+
+    body = (await api.client.get(f"/api/threads/{kickoff['id']}")).json()
+    ids = [m["id"] for m in body["messages"]]
+    # Still two rows (kickoff + reply) — the duplicate collapsed into the first, not
+    # a third row — and the first now notes the other account.
+    assert len(ids) == 2
+    rep = next(m for m in body["messages"] if m["id"] == str(first_id))
+    assert rep["also_in_account_ids"] == [str(acct2.id)]
+
+
 async def test_message_html_is_sanitized(api: ApiCtx) -> None:
     # the lunch message has HTML with a script
     resp = await api.client.get("/api/threads")