sync-v2 step 3: lease-owned account claimer (replaces startup supervisor)

Ownership of an account now moves through a Redis lease so any number of sync
workers can run against the same account set with no double-IDLE.

- leases.py: sync:live registry + sync:lease:{id} (SET NX EX) with owner-checked
  Lua renew/release; mark_live / mark_not_live helpers.
- coordinator.py: SyncCoordinator claims from sync:live up to MAX_ACCOUNTS_PER_
  WORKER, renews every LEASE_RENEW (fences the runner on lease loss), reconciles
  sync:live against Postgres, heartbeats a worker key, and rebalances when it
  holds well above a fair share.
- supervisor start/stop now drive a coordinator; engine._set_state and
  delete_account maintain sync:live membership (+ lease drop = fencing teardown).
- config: MAX_ACCOUNTS_PER_WORKER, LEASE_TTL, LEASE_RENEW.
- tests: acceptance 1-3 (single ownership, crash failover, create/delete
  lifecycle) driving the real coordinator + Redis with faked IMAP loops.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Makefile

@@ -9,8 +9,12 @@ reset-db:  ## DESTRUCTIVE: wipe db + search + blobs, re-migrate
 	cd backend && alembic upgrade head
 api:
 	uvicorn app.main:create_app --factory --reload --port 8000 --app-dir backend
-worker:
+worker:      ## run both sync + jobs in one process (dev)
-	cd backend && arq workers.worker.WorkerSettings
+	cd backend && WORKER_ROLE=all python -m workers.run
+worker-sync: ## run only the account-sync supervisor
+	cd backend && WORKER_ROLE=sync python -m workers.run
+worker-jobs: ## run only the arq job consumers (interactive + bulk)
+	cd backend && WORKER_ROLE=jobs python -m workers.run
 web:
 	cd frontend && npm run dev
 test:

backend/Dockerfile

@@ -0,0 +1,16 @@
+# Shared image for the worker roles (refactor §1). Built from backend/.
+FROM python:3.11-slim
+
+WORKDIR /app
+
+# Install deps first for layer caching.
+COPY pyproject.toml ./
+COPY app ./app
+COPY workers ./workers
+COPY migrations ./migrations
+COPY alembic.ini ./alembic.ini
+RUN pip install --no-cache-dir .
+
+# Default to the combined role; docker-compose overrides WORKER_ROLE per service.
+ENV WORKER_ROLE=all
+CMD ["python", "-m", "workers.run"]

backend/app/accounts/service.py

@@ -80,6 +80,13 @@ async def delete_account(
     await session.execute(delete(Account).where(Account.id == account_id))
     await session.flush()
 
+    # Deregister from sync:live and drop any lease so the owning sync worker
+    # stops the account's runner on its next renewal (refactor §2 fencing).
+    from app.redis_client import get_redis
+    from app.sync import leases
+
+    await leases.mark_not_live(get_redis(), account_id)
+
     # Content-addressed blobs are shared across the user's accounts (dedup): only
     # remove a blob when no remaining message/attachment references it.
     for sha in candidate_shas:

backend/app/api/accounts.py

@@ -30,7 +30,7 @@ from app.api.schemas import AccountOut
 from app.db import get_session
 from app.identity.deps import CurrentUser, require_csrf
 from app.models import Account
-from app.queue import get_arq_pool
+from app.queue import enqueue_job
 from app.redis_client import get_redis
 from app.security.crypto import encrypt_secret
 
@@ -91,9 +91,8 @@ async def list_accounts(user: CurrentUser, db: DbSession) -> list[AccountOut]:
 async def sync_all_accounts(user: CurrentUser, db: DbSession) -> dict[str, str]:
     """Poll every folder of each of the user's accounts (manual Refresh)."""
     account_ids = list(await db.scalars(select(Account.id).where(Account.user_id == user.id)))
-    pool = await get_arq_pool()
     for account_id in account_ids:
-        await pool.enqueue_job("refresh_account_job", str(account_id))
+        await enqueue_job("refresh_account_job", str(account_id))
     return {"status": "queued"}
 
 
@@ -145,8 +144,7 @@ async def create_new_account(
         raise_api_error(409, "account_exists", "an account for this email already exists")
     await db.commit()
 
-    pool = await get_arq_pool()
+    await enqueue_job("sync_bootstrap_account", str(account.id))
-    await pool.enqueue_job("sync_bootstrap_account", str(account.id))
 
     return _detail(account)
 
@@ -227,8 +225,7 @@ async def update_account(
     await db.commit()
 
     if conn_change:
-        pool = await get_arq_pool()
+        await enqueue_job("sync_bootstrap_account", str(account.id))
-        await pool.enqueue_job("sync_bootstrap_account", str(account.id))
 
     return _detail(account)
 

backend/app/api/actions.py

@@ -15,7 +15,7 @@ from app.db import get_session
 from app.events import get_event_bus
 from app.identity.deps import CurrentUser, require_csrf
 from app.models import Mailbox, Message, Thread
-from app.queue import get_arq_pool
+from app.queue import enqueue_job
 from app.sync.actions import apply_flags_to_messages, sanitize_flags
 
 router = APIRouter(tags=["mail"], dependencies=[Depends(require_csrf)])
@@ -26,10 +26,9 @@ DbSession = Annotated[AsyncSession, Depends(get_session)]
 async def _push_and_publish(
     user_id: uuid.UUID, changed: list[Message], add: frozenset[str], remove: frozenset[str]
 ) -> None:
-    pool = await get_arq_pool()
     bus = get_event_bus()
     for message in changed:
-        await pool.enqueue_job("push_flags_job", str(message.id), sorted(add), sorted(remove))
+        await enqueue_job("push_flags_job", str(message.id), sorted(add), sorted(remove))
         await bus.publish(
             "flags.changed",
             {
@@ -95,8 +94,7 @@ async def _move_thread(
     )
     if thread is None:
         raise_api_error(404, "not_found", "thread not found")
-    pool = await get_arq_pool()
+    await enqueue_job("move_thread_to_folder_job", str(thread_id), special)
-    await pool.enqueue_job("move_thread_to_folder_job", str(thread_id), special)
 
 
 @router.post("/threads/{thread_id}/archive")
@@ -126,8 +124,7 @@ async def delete_thread(thread_id: uuid.UUID, user: CurrentUser, db: DbSession)
     )
     if thread is None:
         raise_api_error(404, "not_found", "thread not found")
-    pool = await get_arq_pool()
+    await enqueue_job("delete_thread_job", str(thread_id))
-    await pool.enqueue_job("delete_thread_job", str(thread_id))
     return {"status": "queued"}
 
 
@@ -154,6 +151,5 @@ async def move_message(
     message.mailbox_id = dst.id  # optimistic
     await db.commit()
 
-    pool = await get_arq_pool()
+    await enqueue_job("move_job", str(message.id), str(dst.id), str(src_mailbox_id))
-    await pool.enqueue_job("move_job", str(message.id), str(dst.id), str(src_mailbox_id))
     return {"id": str(message.id), "mailbox_id": str(dst.id)}

backend/app/api/compose.py

@@ -25,7 +25,7 @@ from app.outbox.schemas import (
     SendResponse,
 )
 from app.outbox.service import DraftError, cancel_outbox, create_outbox, upsert_draft
-from app.queue import get_arq_pool
+from app.queue import enqueue_job
 
 router = APIRouter(tags=["compose"], dependencies=[Depends(require_csrf)])
 
@@ -155,9 +155,7 @@ async def list_remote_drafts(user: CurrentUser, db: DbSession) -> list[RemoteDra
 
 
 @router.post("/drafts/adopt/{message_id}", response_model=DraftOut)
-async def adopt_remote_draft(
+async def adopt_remote_draft(message_id: uuid.UUID, user: CurrentUser, db: DbSession) -> DraftOut:
-    message_id: uuid.UUID, user: CurrentUser, db: DbSession
-) -> DraftOut:
     """Turn a Drafts-folder message into an editable local draft (linked by UID)."""
     msg = await db.scalar(
         select(Message)
@@ -235,8 +233,7 @@ async def publish_draft(draft_id: uuid.UUID, user: CurrentUser, db: DbSession) -
     draft = await db.scalar(select(Draft).where(Draft.id == draft_id, Draft.user_id == user.id))
     if draft is None:
         raise_api_error(404, "not_found", "draft not found")
-    pool = await get_arq_pool()
+    await enqueue_job("push_draft_to_imap_job", str(draft_id))
-    await pool.enqueue_job("push_draft_to_imap_job", str(draft_id))
     return {"status": "queued"}
 
 
@@ -250,8 +247,7 @@ async def delete_draft(draft_id: uuid.UUID, user: CurrentUser, db: DbSession) ->
     await db.commit()
     # Remove the mirrored copy from the server Drafts folder too.
     if imap_uid:
-        pool = await get_arq_pool()
+        await enqueue_job("remove_imap_draft_job", str(account_id), imap_uid)
-        await pool.enqueue_job("remove_imap_draft_job", str(account_id), imap_uid)
     return {"status": "deleted"}
 
 
@@ -265,8 +261,7 @@ async def send(body: SendRequest, user: CurrentUser, db: DbSession) -> SendRespo
     outbox = await create_outbox(db, user_id=user.id, draft=draft)
     await db.commit()
 
-    pool = await get_arq_pool()
+    await enqueue_job(
-    await pool.enqueue_job(
         "send_job",
         str(outbox.id),
         _defer_by=timedelta(seconds=get_settings().undo_seconds),

backend/app/config.py

@@ -34,6 +34,17 @@ class Settings(BaseSettings):
     max_attachment_mb: int = Field(default=25, alias="MAX_ATTACHMENT_MB")
     log_level: str = Field(default="INFO", alias="LOG_LEVEL")
 
+    # --- Sync engine v2 (workers layer) --------------------------------------
+    # Worker role: "sync" (lease-owned account runners) | "jobs" (arq consumer)
+    # | "all" (both, single-process dev default).
+    worker_role: str = Field(default="all", alias="WORKER_ROLE")
+    interactive_concurrency: int = Field(default=20, alias="INTERACTIVE_CONCURRENCY")
+    bulk_concurrency: int = Field(default=2, alias="BULK_CONCURRENCY")
+    # Lease protocol (§2): account ownership across sync workers.
+    max_accounts_per_worker: int = Field(default=400, alias="MAX_ACCOUNTS_PER_WORKER")
+    lease_ttl: int = Field(default=60, alias="LEASE_TTL")
+    lease_renew: int = Field(default=20, alias="LEASE_RENEW")
+
     oauth_google_client_id: str = Field(default="", alias="OAUTH_GOOGLE_CLIENT_ID")
     oauth_google_client_secret: str = Field(default="", alias="OAUTH_GOOGLE_CLIENT_SECRET")
     oauth_ms_client_id: str = Field(default="", alias="OAUTH_MS_CLIENT_ID")

backend/app/outbox/draft_sync.py

@@ -34,9 +34,7 @@ async def _ensure_drafts_folder(provider: Any, account: Account, user_id: uuid.U
     uidvalidity = next((r.uidvalidity for r in remote if r.path == path), 0)
     async with session_scope() as session:
         existing = await session.scalar(
-            select(Mailbox).where(
+            select(Mailbox).where(Mailbox.account_id == account.id, Mailbox.special_use == "drafts")
-                Mailbox.account_id == account.id, Mailbox.special_use == "drafts"
-            )
         )
         if existing is not None:
             return existing.path
@@ -72,9 +70,7 @@ async def push_draft_to_imap_job(ctx: dict[str, Any], draft_id: str) -> None:
         attachments = await load_attachments(session, user_id, draft)
         _, sent_raw, _, _ = build_mime(account, draft, reply_to, attachments)
         drafts_mb = await session.scalar(
-            select(Mailbox).where(
+            select(Mailbox).where(Mailbox.account_id == account.id, Mailbox.special_use == "drafts")
-                Mailbox.account_id == account.id, Mailbox.special_use == "drafts"
-            )
         )
         drafts_path = drafts_mb.path if drafts_mb else None
         account_obj = account

backend/app/outbox/jobs.py

@@ -52,9 +52,7 @@ async def _harvest_recipients(session: AsyncSession, user_id: uuid.UUID, emails:
         await session.execute(stmt)
 
 
-async def _persist_sent_copy(
+async def _persist_sent_copy(user_id: uuid.UUID, account: Account, raw: bytes, uid: int) -> None:
-    user_id: uuid.UUID, account: Account, raw: bytes, uid: int
-) -> None:
     """Store the just-sent message in the DB (Sent mailbox) and notify the UI.
 
     No-op without a UIDPLUS append UID (uid <= 0); in that case a future Sent sync
@@ -64,9 +62,7 @@ async def _persist_sent_copy(
         return
     async with session_scope() as session:
         sent_mb = await session.scalar(
-            select(Mailbox).where(
+            select(Mailbox).where(Mailbox.account_id == account.id, Mailbox.special_use == "sent")
-                Mailbox.account_id == account.id, Mailbox.special_use == "sent"
-            )
         )
         if sent_mb is None:
             return
@@ -99,9 +95,7 @@ async def _persist_sent_copy(
         },
     )
     if thread_id is not None:
-        await bus.publish(
+        await bus.publish("thread.updated", {"user_id": str(user_id), "thread_id": str(thread_id)})
-            "thread.updated", {"user_id": str(user_id), "thread_id": str(thread_id)}
-        )
 
 
 async def send_job(ctx: dict[str, Any], outbox_id: str) -> None:

backend/app/queue.py

@@ -1,14 +1,33 @@
-"""arq queue helpers (enqueue side, used by the API)."""
+"""arq queue helpers (enqueue side, used by the API).
+
+Two named queues (refactor §6): ``interactive`` (fast user actions) and ``bulk``
+(backfills). Enqueue via :func:`enqueue_job`, which routes each job to its queue —
+the API never needs to know queue names.
+"""
 
 from __future__ import annotations
 
+from typing import Any
+
 from arq import create_pool
 from arq.connections import ArqRedis, RedisSettings
+from arq.jobs import Job
 
 from app.config import get_settings
 
 _pool: ArqRedis | None = None
 
+QUEUE_INTERACTIVE = "interactive"
+QUEUE_BULK = "bulk"
+
+# Jobs default to the interactive queue; only bulk backfills go to the bulk queue.
+_BULK_JOBS = frozenset({"sync_bootstrap_account"})
+
+
+def queue_for(job_name: str) -> str:
+    """Return the queue a job should run on."""
+    return QUEUE_BULK if job_name in _BULK_JOBS else QUEUE_INTERACTIVE
+
 
 def redis_settings() -> RedisSettings:
     """Return arq RedisSettings derived from ``REDIS_URL``."""
@@ -23,6 +42,13 @@ async def get_arq_pool() -> ArqRedis:
     return _pool
 
 
+async def enqueue_job(name: str, *args: Any, **kwargs: Any) -> Job | None:
+    """Enqueue ``name`` onto its routed queue (interactive/bulk)."""
+    pool = await get_arq_pool()
+    kwargs.setdefault("_queue_name", queue_for(name))
+    return await pool.enqueue_job(name, *args, **kwargs)
+
+
 async def close_arq_pool() -> None:
     """Close the arq pool."""
     global _pool

backend/app/sync/actions.py

@@ -273,9 +273,7 @@ async def delete_thread_job(ctx: dict[str, Any], thread_id: str) -> None:
                 .where(Draft.reply_to_message_id.in_(msg_ids))
                 .values(reply_to_message_id=None)
             )
-        await session.execute(
+        await session.execute(update(Draft).where(Draft.thread_id == tid).values(thread_id=None))
-            update(Draft).where(Draft.thread_id == tid).values(thread_id=None)
-        )
         for m in msgs:
             await session.delete(m)
         thread = await session.get(Thread, tid)

backend/app/sync/coordinator.py

@@ -0,0 +1,205 @@
+"""Sync coordinator: lease-owned account claimer (refactor §2, step 3).
+
+Replaces the startup-only supervisor. Each sync worker runs one coordinator that:
+
+* claims accounts from ``sync:live`` (up to ``MAX_ACCOUNTS_PER_WORKER``),
+  starting an :class:`AccountRunner` for each lease it wins;
+* renews its leases every ``LEASE_RENEW`` seconds and, on losing one (a renewal
+  that returns 0), stops that account's runner immediately (fencing);
+* reconciles ``sync:live`` against Postgres periodically as a safety net;
+* heartbeats a worker key so peers can compute a fair share and rebalance.
+
+All ownership is via :mod:`app.sync.leases`; two workers can never run the same
+account because claims are ``SET NX`` and renew/release are owner-checked.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import os
+import random
+import secrets
+import socket
+import time
+import uuid
+
+from sqlalchemy import select
+
+from app.config import get_settings
+from app.db import session_scope
+from app.models import Account
+from app.redis_client import aw, get_redis
+from app.sync import leases
+from app.sync.runner import AccountRunner
+
+logger = logging.getLogger(__name__)
+
+
+def _new_worker_id() -> str:
+    return f"{socket.gethostname()}:{os.getpid()}:{secrets.token_hex(4)}"
+
+
+class SyncCoordinator:
+    """Owns the leased account runners for a single sync worker."""
+
+    def __init__(self, worker_id: str | None = None) -> None:
+        self.worker_id = worker_id or _new_worker_id()
+        self._redis = get_redis()
+        self._scripts = leases.LeaseScripts(self._redis)
+        self._runners: dict[uuid.UUID, AccountRunner] = {}
+        self._stop = asyncio.Event()
+        self._tasks: list[asyncio.Task[None]] = []
+
+        s = get_settings()
+        self.lease_ttl = s.lease_ttl
+        self.lease_renew = s.lease_renew
+        self.max_accounts = s.max_accounts_per_worker
+        # Operational cadences (not operator-tunable; §2 protocol timings).
+        self.claim_interval = 10.0
+        self.claim_jitter = 5.0
+        self.reconcile_interval = 300.0
+        self.heartbeat_ttl = 30
+        self.heartbeat_refresh = 10.0
+        self.rebalance_interval = 60.0
+
+    # --- lifecycle -----------------------------------------------------------
+
+    async def start(self) -> None:
+        self._stop.clear()
+        await self._reconcile_once()  # seed sync:live so the first claim isn't delayed
+        await self._heartbeat_once()
+        self._tasks = [
+            asyncio.create_task(
+                self._loop(self._claim_once, self.claim_interval, self.claim_jitter)
+            ),
+            asyncio.create_task(self._loop(self._renew_once, self.lease_renew)),
+            asyncio.create_task(self._loop(self._reconcile_once, self.reconcile_interval)),
+            asyncio.create_task(self._loop(self._heartbeat_once, self.heartbeat_refresh)),
+            asyncio.create_task(self._loop(self._rebalance_once, self.rebalance_interval)),
+        ]
+        logger.info("sync coordinator %s started", self.worker_id)
+
+    async def stop(self) -> None:
+        self._stop.set()
+        for task in self._tasks:
+            task.cancel()
+        if self._tasks:
+            await asyncio.gather(*self._tasks, return_exceptions=True)
+        self._tasks = []
+        # Release every lease so peers can pick the accounts up without waiting for TTL.
+        await asyncio.gather(
+            *(self._release(aid) for aid in list(self._runners)), return_exceptions=True
+        )
+        await self._redis.delete(leases.worker_key(self.worker_id))
+        logger.info("sync coordinator %s stopped", self.worker_id)
+
+    @property
+    def owned(self) -> set[uuid.UUID]:
+        return set(self._runners)
+
+    # --- periodic loop driver ------------------------------------------------
+
+    async def _loop(self, step: object, interval: float, jitter: float = 0.0) -> None:
+        while not self._stop.is_set():
+            delay = interval + random.uniform(0.0, jitter)
+            try:
+                await asyncio.wait_for(self._stop.wait(), timeout=delay)
+                return  # stop requested
+            except TimeoutError:
+                pass
+            try:
+                await step()  # type: ignore[operator]
+            except asyncio.CancelledError:
+                raise
+            except Exception:
+                logger.exception("coordinator step failed on worker %s", self.worker_id)
+
+    # --- claim / renew / release ---------------------------------------------
+
+    async def _claim_once(self) -> None:
+        if len(self._runners) >= self.max_accounts:
+            return
+        async for raw in self._redis.sscan_iter(leases.LIVE_SET):
+            if self._stop.is_set() or len(self._runners) >= self.max_accounts:
+                break
+            account_id = uuid.UUID(raw)
+            if account_id in self._runners:
+                continue
+            if await leases.claim(self._redis, account_id, self.worker_id, self.lease_ttl):
+                self._start_runner(account_id)
+
+    async def _renew_once(self) -> None:
+        for account_id in list(self._runners):
+            if not await self._scripts.renew(account_id, self.worker_id, self.lease_ttl):
+                logger.warning("lost lease for account %s; fencing runner", account_id)
+                await self._drop(account_id)
+
+    async def _release(self, account_id: uuid.UUID) -> None:
+        """Graceful hand-off: stop the runner, then delete the lease if still ours."""
+        runner = self._runners.pop(account_id, None)
+        if runner is not None:
+            await runner.stop()
+        await self._scripts.release(account_id, self.worker_id)
+
+    async def _drop(self, account_id: uuid.UUID) -> None:
+        """Fencing: stop the runner without touching the lease (we no longer own it)."""
+        runner = self._runners.pop(account_id, None)
+        if runner is not None:
+            await runner.stop()
+
+    def _start_runner(self, account_id: uuid.UUID) -> None:
+        runner = AccountRunner(account_id)
+        runner.start()
+        self._runners[account_id] = runner
+
+    # --- reconcile / heartbeat / rebalance -----------------------------------
+
+    async def _reconcile_once(self) -> None:
+        """Make ``sync:live`` match the set of live accounts in Postgres.
+
+        Never touches leases — only set membership. Accounts that disappear here
+        stop being claimed; their runners are dropped when their lease is removed.
+        """
+        async with session_scope() as session:
+            live_db = {
+                str(a)
+                for a in await session.scalars(select(Account.id).where(Account.state == "live"))
+            }
+        live_redis = set(await aw(self._redis.smembers(leases.LIVE_SET)))
+        to_add = live_db - live_redis
+        to_remove = live_redis - live_db
+        if to_add:
+            await aw(self._redis.sadd(leases.LIVE_SET, *to_add))
+        if to_remove:
+            await aw(self._redis.srem(leases.LIVE_SET, *to_remove))
+            # Stop any runners for accounts we own that are no longer live.
+            for raw in to_remove:
+                account_id = uuid.UUID(raw)
+                if account_id in self._runners:
+                    await self._release(account_id)
+
+    async def _heartbeat_once(self) -> None:
+        payload = json.dumps({"owned": len(self._runners), "ts": time.time()})
+        await self._redis.set(leases.worker_key(self.worker_id), payload, ex=self.heartbeat_ttl)
+
+    async def _alive_worker_count(self) -> int:
+        count = 0
+        async for _ in self._redis.scan_iter(match=f"{leases.WORKERS_PREFIX}*"):
+            count += 1
+        return max(count, 1)
+
+    async def _rebalance_once(self) -> None:
+        """Shed accounts if we hold well above a fair share of the live total."""
+        total = int(await aw(self._redis.scard(leases.LIVE_SET)))
+        workers = await self._alive_worker_count()
+        target = total / workers
+        threshold = int(target * 1.2) + 5
+        excess = len(self._runners) - threshold
+        if excess <= 0:
+            return
+        # Release the newest claims first (least likely to be mid-backfill on us).
+        for account_id in list(self._runners)[-excess:]:
+            await self._release(account_id)
+        logger.info("worker %s shed %d account(s) to rebalance", self.worker_id, excess)

backend/app/sync/engine.py

@@ -41,6 +41,17 @@ async def _set_state(account_id: uuid.UUID, state: str, detail: str = "") -> Non
         await session.execute(
             update(Account).where(Account.id == account_id).values(state=state, state_detail=detail)
         )
+    # Keep the sync:live registry in step with the account's state so the sync
+    # coordinator claims it once live and drops it otherwise (refactor §2).
+    from app.redis_client import get_redis
+    from app.sync import leases
+
+    redis = get_redis()
+    if state == "live":
+        await leases.mark_live(redis, account_id)
+    else:
+        await leases.mark_not_live(redis, account_id)
+
     bus = get_event_bus()
     account = None
     async with session_scope() as session:

backend/app/sync/leases.py

@@ -0,0 +1,83 @@
+"""Redis lease primitives for account ownership (refactor §2).
+
+A sync worker must hold the lease ``sync:lease:{account_id}`` before it runs an
+account's sync loops, and no two workers may hold the same lease. Claims are
+atomic (``SET NX EX``); renewal and release are owner-checked Lua so a worker can
+never renew or delete a lease that has been taken over by another worker.
+
+The set ``sync:live`` is the registry of accounts that *should* be synced; the
+claimer only ever claims ids drawn from it.
+"""
+
+from __future__ import annotations
+
+import uuid
+
+import redis.asyncio as aioredis
+
+from app.redis_client import aw
+
+LIVE_SET = "sync:live"
+_LEASE_PREFIX = "sync:lease:"
+WORKERS_PREFIX = "sync:workers:"
+
+
+def lease_key(account_id: uuid.UUID) -> str:
+    return f"{_LEASE_PREFIX}{account_id}"
+
+
+def worker_key(worker_id: str) -> str:
+    return f"{WORKERS_PREFIX}{worker_id}"
+
+
+# Renew only if we still own the lease (guards against a stale worker extending a
+# lease another worker has already claimed after expiry). Returns 1 on success.
+_RENEW_LUA = """
+if redis.call('GET', KEYS[1]) == ARGV[1] then
+  return redis.call('EXPIRE', KEYS[1], ARGV[2])
+else
+  return 0
+end
+"""
+
+# Delete only if we still own the lease. Returns 1 if we released it, else 0.
+_RELEASE_LUA = """
+if redis.call('GET', KEYS[1]) == ARGV[1] then
+  return redis.call('DEL', KEYS[1])
+else
+  return 0
+end
+"""
+
+
+class LeaseScripts:
+    """Server-registered Lua for owner-checked renew/release."""
+
+    def __init__(self, redis: aioredis.Redis) -> None:
+        self._renew = redis.register_script(_RENEW_LUA)
+        self._release = redis.register_script(_RELEASE_LUA)
+
+    async def renew(self, account_id: uuid.UUID, worker_id: str, ttl: int) -> bool:
+        result = await self._renew(keys=[lease_key(account_id)], args=[worker_id, ttl])
+        return bool(result)
+
+    async def release(self, account_id: uuid.UUID, worker_id: str) -> bool:
+        result = await self._release(keys=[lease_key(account_id)], args=[worker_id])
+        return bool(result)
+
+
+async def claim(redis: aioredis.Redis, account_id: uuid.UUID, worker_id: str, ttl: int) -> bool:
+    """Atomically claim the lease for ``account_id``; True iff we now own it."""
+    ok = await redis.set(lease_key(account_id), worker_id, nx=True, ex=ttl)
+    return bool(ok)
+
+
+async def mark_live(redis: aioredis.Redis, account_id: uuid.UUID) -> None:
+    """Register an account as one that should be synced (idempotent)."""
+    await aw(redis.sadd(LIVE_SET, str(account_id)))
+
+
+async def mark_not_live(redis: aioredis.Redis, account_id: uuid.UUID) -> None:
+    """Deregister an account and drop any lease so its owner stops it (fencing)."""
+    await aw(redis.srem(LIVE_SET, str(account_id)))
+    await aw(redis.delete(lease_key(account_id)))

backend/app/sync/runner.py

@@ -0,0 +1,68 @@
+"""Per-account sync runner (refactor §4).
+
+An ``AccountRunner`` owns the long-lived sync loops for a single account and
+exposes a small ``start`` / ``stop`` lifecycle. This is a pure extraction of the
+loops that previously lived inline in the supervisor — behavior is unchanged
+(INBOX IDLE + secondary CONDSTORE poll, each self-healing with backoff).
+"""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import logging
+import uuid
+from collections.abc import Awaitable, Callable
+
+from app.sync.engine import live_loop, secondary_poll_loop
+
+logger = logging.getLogger(__name__)
+
+_RECONNECT_BACKOFF = 30.0
+
+
+class AccountRunner:
+    """Owns the sync loops for one account until stopped."""
+
+    def __init__(self, account_id: uuid.UUID) -> None:
+        self.account_id = account_id
+        self._stop = asyncio.Event()
+        self._tasks: list[asyncio.Task[None]] = []
+
+    def start(self) -> None:
+        """Launch the account's sync loops (idempotent-ish: call once)."""
+        self._stop.clear()
+        self._tasks = [
+            asyncio.create_task(self._supervise(self._run_live, "live")),
+            asyncio.create_task(self._supervise(self._run_secondary, "secondary")),
+        ]
+
+    def running(self) -> bool:
+        return any(not t.done() for t in self._tasks)
+
+    async def stop(self) -> None:
+        """Signal and cancel the loops, waiting for teardown."""
+        self._stop.set()
+        for task in self._tasks:
+            task.cancel()
+        if self._tasks:
+            await asyncio.gather(*self._tasks, return_exceptions=True)
+        self._tasks = []
+
+    async def _run_live(self) -> None:
+        await live_loop(self.account_id, "INBOX", self._stop)
+
+    async def _run_secondary(self) -> None:
+        await secondary_poll_loop(self.account_id, self._stop)
+
+    async def _supervise(self, run: Callable[[], Awaitable[None]], label: str) -> None:
+        """Run a loop, restarting it (with backoff) until stopped."""
+        while not self._stop.is_set():
+            try:
+                await run()
+            except asyncio.CancelledError:
+                raise
+            except Exception:
+                logger.exception("%s loop for account %s crashed; retrying", label, self.account_id)
+                with contextlib.suppress(TimeoutError):
+                    await asyncio.wait_for(self._stop.wait(), timeout=_RECONNECT_BACKOFF)

backend/app/sync/supervisor.py

@@ -1,77 +1,29 @@
-"""Live-sync supervisor (SPEC §6.3).
+"""Live-sync entrypoint (refactor §2, step 3).
 
-Runs inside the arq worker: on startup it launches one IDLE/poll loop per live
+The sync worker's ``on_startup`` starts a :class:`SyncCoordinator`, which claims
-account so new mail and flag changes flow to the UI over WebSocket. Loops
+accounts from the ``sync:live`` registry via Redis leases and runs one
-self-heal (reconnect with backoff) and are cancelled on worker shutdown.
+:class:`~app.sync.runner.AccountRunner` per owned account. ``on_shutdown`` stops
-
+it, releasing every lease so peers can take the accounts over immediately.
-NOTE(spec): accounts that become live *after* worker startup are picked up on the
-next worker restart; reacting to ``account.state_changed`` live is a later refinement.
 """
 
 from __future__ import annotations
 
-import asyncio
-import contextlib
 import logging
-import uuid
 from typing import Any
 
-from sqlalchemy import select
+from app.sync.coordinator import SyncCoordinator
-
-from app.db import session_scope
-from app.models import Account
-from app.sync.engine import live_loop, secondary_poll_loop
 
 logger = logging.getLogger(__name__)
 
-_RECONNECT_BACKOFF = 30.0
-
-
-async def _run_account(account_id: uuid.UUID, stop: asyncio.Event) -> None:
-    while not stop.is_set():
-        try:
-            await live_loop(account_id, "INBOX", stop)
-        except asyncio.CancelledError:
-            raise
-        except Exception:
-            logger.exception("live loop for account %s crashed; reconnecting", account_id)
-            with contextlib.suppress(TimeoutError):
-                await asyncio.wait_for(stop.wait(), timeout=_RECONNECT_BACKOFF)
-
-
-async def _run_secondary(account_id: uuid.UUID, stop: asyncio.Event) -> None:
-    while not stop.is_set():
-        try:
-            await secondary_poll_loop(account_id, stop)
-        except asyncio.CancelledError:
-            raise
-        except Exception:
-            logger.exception("secondary poll loop for %s crashed; retrying", account_id)
-            with contextlib.suppress(TimeoutError):
-                await asyncio.wait_for(stop.wait(), timeout=_RECONNECT_BACKOFF)
-
 
 async def start_supervisor(ctx: dict[str, Any]) -> None:
-    """Launch a live loop per live account (called from the worker on_startup)."""
+    coordinator = SyncCoordinator()
-    async with session_scope() as session:
+    await coordinator.start()
-        account_ids = list(await session.scalars(select(Account.id).where(Account.state == "live")))
+    ctx["coordinator"] = coordinator
-    stop = asyncio.Event()
-    ctx["supervisor_stop"] = stop
-    tasks: list[asyncio.Task[None]] = []
-    for aid in account_ids:
-        tasks.append(asyncio.create_task(_run_account(aid, stop)))  # INBOX IDLE
-        tasks.append(asyncio.create_task(_run_secondary(aid, stop)))  # other folders poll
-    ctx["supervisor_tasks"] = tasks
-    logger.info("live supervisor started for %d account(s)", len(account_ids))
 
 
 async def stop_supervisor(ctx: dict[str, Any]) -> None:
-    """Signal and cancel all live loops (called from the worker on_shutdown)."""
+    coordinator: SyncCoordinator | None = ctx.get("coordinator")
-    stop = ctx.get("supervisor_stop")
+    if coordinator is not None:
-    if stop is not None:
+        await coordinator.stop()
-        stop.set()
+    ctx["coordinator"] = None
-    tasks = ctx.get("supervisor_tasks", [])
-    for task in tasks:
-        task.cancel()
-    if tasks:
-        await asyncio.gather(*tasks, return_exceptions=True)

backend/migrations/versions/0006_user_quota.py

@@ -21,9 +21,7 @@ depends_on: str | Sequence[str] | None = None
 
 
 def upgrade() -> None:
-    op.execute(
+    op.execute("ALTER TABLE users ADD COLUMN quota_bytes bigint NOT NULL DEFAULT 5368709120")
-        "ALTER TABLE users ADD COLUMN quota_bytes bigint NOT NULL DEFAULT 5368709120"
-    )
 
 
 def downgrade() -> None:

backend/tests/accounts/test_account_api.py

@@ -13,6 +13,7 @@ from sqlalchemy import delete, select
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
 from sqlalchemy.pool import NullPool
 
+from app import queue as app_queue
 from app.api import accounts as accounts_api
 from app.config import get_settings
 from app.identity.passwords import hash_password
@@ -114,14 +115,14 @@ async def test_create_account_enqueues_sync(ctx: Ctx, monkeypatch) -> None:  # t
     enqueued: list[tuple[str, str]] = []
 
     class FakePool:
-        async def enqueue_job(self, name: str, arg: str) -> None:
+        async def enqueue_job(self, name: str, arg: str, **kwargs: object) -> None:
             enqueued.append((name, arg))
 
     async def fake_pool() -> FakePool:
         return FakePool()
 
     monkeypatch.setattr(accounts_api, "test_account", fake_test)
-    monkeypatch.setattr(accounts_api, "get_arq_pool", fake_pool)
+    monkeypatch.setattr(app_queue, "get_arq_pool", fake_pool)
 
     resp = await ctx.client.post(
         "/api/accounts",
@@ -151,14 +152,14 @@ async def _make_account(ctx: Ctx, monkeypatch) -> str:  # type: ignore[no-untype
         return None
 
     class FakePool:
-        async def enqueue_job(self, name: str, arg: str) -> None:
+        async def enqueue_job(self, name: str, arg: str, **kwargs: object) -> None:
             return None
 
     async def fake_pool() -> FakePool:
         return FakePool()
 
     monkeypatch.setattr(accounts_api, "test_account", fake_test)
-    monkeypatch.setattr(accounts_api, "get_arq_pool", fake_pool)
+    monkeypatch.setattr(app_queue, "get_arq_pool", fake_pool)
     resp = await ctx.client.post(
         "/api/accounts",
         json={**GOOD_CFG, "email": "me@example.com", "display_name": "Me", "password": "secret"},
@@ -174,13 +175,13 @@ async def test_sync_all_enqueues_refresh_per_account(ctx: Ctx, monkeypatch) -> N
     enqueued: list[tuple[str, str]] = []
 
     class FakePool:
-        async def enqueue_job(self, name: str, arg: str) -> None:
+        async def enqueue_job(self, name: str, arg: str, **kwargs: object) -> None:
             enqueued.append((name, arg))
 
     async def fake_pool() -> FakePool:
         return FakePool()
 
-    monkeypatch.setattr(accounts_api, "get_arq_pool", fake_pool)
+    monkeypatch.setattr(app_queue, "get_arq_pool", fake_pool)
     resp = await ctx.client.post("/api/accounts/sync", headers=_hdr())
     assert resp.status_code == 200
     assert enqueued == [("refresh_account_job", aid)]
@@ -221,7 +222,7 @@ async def test_update_password_reencrypts_and_resyncs(ctx: Ctx, monkeypatch) ->
     enqueued: list[tuple[str, str]] = []
 
     class FakePool:
-        async def enqueue_job(self, name: str, arg: str) -> None:
+        async def enqueue_job(self, name: str, arg: str, **kwargs: object) -> None:
             enqueued.append((name, arg))
 
     async def fake_pool() -> FakePool:
@@ -231,7 +232,7 @@ async def test_update_password_reencrypts_and_resyncs(ctx: Ctx, monkeypatch) ->
         return None
 
     monkeypatch.setattr(accounts_api, "test_account", fake_test)
-    monkeypatch.setattr(accounts_api, "get_arq_pool", fake_pool)
+    monkeypatch.setattr(app_queue, "get_arq_pool", fake_pool)
 
     resp = await ctx.client.patch(
         f"/api/accounts/{aid}", json={"password": "newsecret"}, headers=_hdr()
@@ -259,14 +260,14 @@ async def test_delete_account(ctx: Ctx, monkeypatch) -> None:  # type: ignore[no
         return None
 
     class FakePool:
-        async def enqueue_job(self, name: str, arg: str) -> None:
+        async def enqueue_job(self, name: str, arg: str, **kwargs: object) -> None:
             return None
 
     async def fake_pool() -> FakePool:
         return FakePool()
 
     monkeypatch.setattr(accounts_api, "test_account", fake_test)
-    monkeypatch.setattr(accounts_api, "get_arq_pool", fake_pool)
+    monkeypatch.setattr(app_queue, "get_arq_pool", fake_pool)
 
     created = await ctx.client.post(
         "/api/accounts",

backend/tests/mail/test_actions.py

@@ -15,7 +15,7 @@ from sqlalchemy import delete, func, select
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
 from sqlalchemy.pool import NullPool
 
-from app.api import actions as actions_api
+from app import queue as app_queue
 from app.config import get_settings
 from app.db import session_scope
 from app.identity.passwords import hash_password
@@ -112,13 +112,13 @@ async def ctx(monkeypatch) -> AsyncIterator[Ctx]:  # type: ignore[no-untyped-def
 
     # capture enqueues instead of hitting a real arq pool
     class FakePool:
-        async def enqueue_job(self, *args: object) -> None:
+        async def enqueue_job(self, *args: object, **kwargs: object) -> None:
             c.enqueued.append(args)
 
     async def fake_pool() -> FakePool:
         return FakePool()
 
-    monkeypatch.setattr(actions_api, "get_arq_pool", fake_pool)
+    monkeypatch.setattr(app_queue, "get_arq_pool", fake_pool)
 
     token = await create_session(user.id, "pytest")
     c.client = httpx.AsyncClient(

backend/tests/outbox/test_draft_sync.py

@@ -127,9 +127,7 @@ async def test_draft_messages_are_not_threaded(ctx: Ctx) -> None:
     m.set_content("body")
 
     drafts_mb = await ctx.session.scalar(
-        select(Mailbox).where(
+        select(Mailbox).where(Mailbox.account_id == ctx.account_id, Mailbox.special_use == "drafts")
-            Mailbox.account_id == ctx.account_id, Mailbox.special_use == "drafts"
-        )
     )
     assert drafts_mb is not None
     remote = RemoteMessage(

backend/tests/outbox/test_outbox.py

@@ -16,7 +16,7 @@ from sqlalchemy import delete, select
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
 from sqlalchemy.pool import NullPool
 
-from app.api import compose as compose_api
+from app import queue as app_queue
 from app.config import get_settings
 from app.db import session_scope
 from app.identity.passwords import hash_password
@@ -181,7 +181,7 @@ async def ctx(monkeypatch) -> AsyncIterator[Ctx]:  # type: ignore[no-untyped-def
     async def fake_pool() -> FakePool:
         return FakePool()
 
-    monkeypatch.setattr(compose_api, "get_arq_pool", fake_pool)
+    monkeypatch.setattr(app_queue, "get_arq_pool", fake_pool)
 
     token = await create_session(user.id, "pytest")
     c.client = httpx.AsyncClient(

backend/tests/sync/test_acceptance.py

@@ -0,0 +1,226 @@
+"""Sync-engine v2 acceptance tests 1-3 (refactor §10).
+
+These drive the real :class:`SyncCoordinator` against the test Redis with the
+per-account IMAP loops replaced by fakes, so the lease protocol — single
+ownership, crash failover, and create/delete lifecycle — is exercised without a
+live mail server. Intervals are shortened so the tests run in seconds.
+
+  1. Two workers, N accounts: every account owned by exactly one worker, exactly
+     one live (IDLE) connection each.
+  2. Kill a worker: its orphaned accounts are re-owned by the survivor once the
+     leases expire, still one connection each (no double ownership).
+  3. A newly-live account is claimed without a restart; a removed account's
+     connection is torn down (lease fencing), with no further ownership.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import uuid
+from collections import defaultdict
+from collections.abc import AsyncIterator, Callable
+
+import pytest_asyncio
+from sqlalchemy import delete
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+from sqlalchemy.pool import NullPool
+
+from app.config import get_settings
+from app.identity.passwords import hash_password
+from app.models import Account, User
+from app.redis_client import get_redis
+from app.sync import coordinator as coordinator_mod
+from app.sync import leases, runner
+from app.sync.coordinator import SyncCoordinator
+
+# --- fakes: replace the real IMAP loops with connection-counting stand-ins ----
+
+
+@pytest_asyncio.fixture
+async def connections(monkeypatch) -> AsyncIterator[dict[uuid.UUID, int]]:  # type: ignore[no-untyped-def]
+    """Patch the runner's loops; yield a live-connection count per account."""
+    conns: dict[uuid.UUID, int] = defaultdict(int)
+
+    async def fake_live_loop(account_id: uuid.UUID, folder: str, stop: asyncio.Event) -> None:
+        conns[account_id] += 1  # one IDLE connection while the loop is alive
+        try:
+            await stop.wait()
+        finally:
+            conns[account_id] -= 1
+
+    async def fake_secondary(account_id: uuid.UUID, stop: asyncio.Event | None = None) -> None:
+        assert stop is not None
+        await stop.wait()
+
+    monkeypatch.setattr(runner, "live_loop", fake_live_loop)
+    monkeypatch.setattr(runner, "secondary_poll_loop", fake_secondary)
+    yield conns
+
+
+@pytest_asyncio.fixture
+async def make_accounts() -> AsyncIterator[Callable[[int, str], object]]:
+    """Factory creating N live accounts; cleans up its users afterwards."""
+    engine = create_async_engine(get_settings().database_url, poolclass=NullPool)
+    sm = async_sessionmaker(engine, expire_on_commit=False)
+    user_ids: list[uuid.UUID] = []
+
+    async def _make(n: int, state: str = "live") -> list[uuid.UUID]:
+        async with sm() as session:
+            user = User(
+                handle=f"acc{uuid.uuid4().hex[:8]}@test.local", password_hash=hash_password("pw")
+            )
+            session.add(user)
+            await session.flush()
+            user_ids.append(user.id)
+            ids: list[uuid.UUID] = []
+            for _ in range(n):
+                account = Account(
+                    user_id=user.id,
+                    email=f"{uuid.uuid4().hex[:10]}@example.com",
+                    display_name="Acct",
+                    imap_config={},
+                    smtp_config={},
+                    secret_ct=b"",
+                    discovery_src="manual",
+                    state=state,
+                )
+                session.add(account)
+                await session.flush()
+                ids.append(account.id)
+            await session.commit()
+            return ids
+
+    try:
+        yield _make
+    finally:
+        async with sm() as session:
+            for uid in user_ids:
+                await session.execute(delete(User).where(User.id == uid))
+            await session.commit()
+        await engine.dispose()
+
+
+async def _no_reconcile() -> None:
+    """Stub: keep the shared dev DB's real live accounts out of the test."""
+    return None
+
+
+def _fast(c: SyncCoordinator) -> SyncCoordinator:
+    """Compress cadences and drive sync:live explicitly (no DB reconcile).
+
+    The tests share the dev Postgres, which holds real live accounts; reconciling
+    from it would pull them into the test. We seed ``sync:live`` ourselves instead.
+    """
+    c.claim_interval = 0.05
+    c.claim_jitter = 0.0
+    c.lease_renew = 0.3
+    c.lease_ttl = 2  # short so a crashed worker's leases expire quickly
+    c.reconcile_interval = 1000.0
+    c.heartbeat_refresh = 1000.0
+    c.rebalance_interval = 1000.0
+    c._reconcile_once = _no_reconcile  # type: ignore[method-assign]
+    return c
+
+
+async def _seed(ids: set[uuid.UUID]) -> None:
+    for account_id in ids:
+        await leases.mark_live(get_redis(), account_id)
+
+
+async def _until(pred: Callable[[], bool], timeout_s: float = 6.0) -> None:
+    loop = asyncio.get_running_loop()
+    deadline = loop.time() + timeout_s
+    while loop.time() < deadline:
+        if pred():
+            return
+        await asyncio.sleep(0.05)
+    assert pred(), "condition not met within timeout"
+
+
+# --- acceptance 1: single ownership, one connection per account ---------------
+
+
+async def test_acceptance_1_single_ownership(connections, make_accounts) -> None:  # type: ignore[no-untyped-def]
+    ids = set(await make_accounts(20))
+    await _seed(ids)
+    c1 = _fast(SyncCoordinator("w1"))
+    c2 = _fast(SyncCoordinator("w2"))
+    c1.rebalance_interval = c2.rebalance_interval = 0.3  # let the load even out
+    # Start concurrently so the two workers race for the leases.
+    await asyncio.gather(c1.start(), c2.start())
+    try:
+        # Both workers end up holding accounts (via the initial race + rebalance),
+        # together covering all 20 with no account owned twice.
+        await _until(lambda: c1.owned | c2.owned == ids and bool(c1.owned and c2.owned))
+        assert c1.owned & c2.owned == set()
+        # At steady state each account has exactly one live IDLE connection.
+        await _until(lambda: all(connections[a] == 1 for a in ids))
+    finally:
+        await c1.stop()
+        await c2.stop()
+    await _until(lambda: all(connections[a] == 0 for a in ids))
+
+
+# --- acceptance 2: crash failover --------------------------------------------
+
+
+async def test_acceptance_2_failover(connections, make_accounts) -> None:  # type: ignore[no-untyped-def]
+    ids = set(await make_accounts(10))
+    await _seed(ids)
+    c1 = _fast(SyncCoordinator("w1"))
+    await c1.start()
+    try:
+        await _until(lambda: c1.owned == ids)  # sole worker owns everything
+
+        c2 = _fast(SyncCoordinator("w2"))
+        await c2.start()
+
+        # Simulate `kill -9` on w1: connections drop and renewals stop, but the
+        # leases are NOT released — they must expire before w2 can take over.
+        for task in c1._tasks:
+            task.cancel()
+        await asyncio.gather(*c1._tasks, return_exceptions=True)
+        c1._tasks = []
+        for r in list(c1._runners.values()):
+            await r.stop()
+        c1._runners.clear()
+
+        try:
+            # w2 re-owns every orphan once the (short) leases expire.
+            await _until(lambda: c2.owned == ids, timeout_s=8.0)
+            assert all(connections[a] == 1 for a in ids)  # one connection each, no dupes
+        finally:
+            await c2.stop()
+    finally:
+        # w1 already crashed; clear any residual leases it left behind.
+        await get_redis().flushdb()
+    await _until(lambda: all(connections[a] == 0 for a in ids))
+
+
+# --- acceptance 3: create / delete lifecycle without restart ------------------
+
+
+async def test_acceptance_3_create_delete(connections, make_accounts) -> None:  # type: ignore[no-untyped-def]
+    c = _fast(SyncCoordinator("w1"))
+    await c.start()  # starts with no live accounts
+    try:
+        assert c.owned == set()
+
+        # A newly-live account (as the bootstrap job's _set_state('live') would do):
+        # register it in sync:live; the running coordinator must claim it, no restart.
+        (aid,) = await make_accounts(1)
+        await leases.mark_live(get_redis(), aid)
+        await _until(lambda: aid in c.owned)
+        assert connections[aid] == 1
+
+        # Removing it (as delete_account does) fences the runner: lease gone ->
+        # next renewal fails -> connection torn down, no further ownership.
+        await leases.mark_not_live(get_redis(), aid)
+        await _until(lambda: aid not in c.owned)
+        await _until(lambda: connections[aid] == 0)
+    finally:
+        await c.stop()
+
+
+def test_worker_id_is_unique() -> None:
+    assert coordinator_mod._new_worker_id() != coordinator_mod._new_worker_id()

backend/workers/run.py

@@ -0,0 +1,111 @@
+"""Worker entrypoint (refactor §1).
+
+One image, two roles selected by ``WORKER_ROLE``:
+
+- ``jobs`` — arq consumer for the ``interactive`` and ``bulk`` queues (both in one
+  process, per-queue concurrency/timeout).
+- ``sync`` — runs the account-sync supervisor (long-lived per-account loops).
+- ``all``  — both, for single-process development (the default).
+
+Run with: ``python -m workers.run``.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import signal
+from typing import Any
+
+from arq.worker import Worker
+
+from app.config import get_settings
+from app.queue import QUEUE_BULK, QUEUE_INTERACTIVE, redis_settings
+from workers.worker import (
+    BULK_FUNCTIONS,
+    INTERACTIVE_FUNCTIONS,
+    sync_on_shutdown,
+    sync_on_startup,
+)
+
+logger = logging.getLogger(__name__)
+
+# arq requires a non-empty function set; the sync worker consumes a queue nothing
+# is ever enqueued to, so it just keeps the event loop alive for the supervisor.
+_SYNC_CONTROL_QUEUE = "sync:control"
+
+
+async def _noop(ctx: dict[str, Any]) -> None:
+    return None
+
+
+def _worker(
+    functions: list[Any],
+    queue_name: str,
+    max_jobs: int,
+    job_timeout: int,
+    on_startup: Any = None,
+    on_shutdown: Any = None,
+) -> Worker:
+    return Worker(
+        functions=functions,
+        queue_name=queue_name,
+        redis_settings=redis_settings(),
+        max_jobs=max_jobs,
+        job_timeout=job_timeout,
+        on_startup=on_startup,
+        on_shutdown=on_shutdown,
+        handle_signals=False,  # the runner owns signal handling (graceful shutdown)
+    )
+
+
+def build_workers(role: str) -> list[Worker]:
+    """Construct the arq Worker instances for the given role."""
+    s = get_settings()
+    workers: list[Worker] = []
+    if role in ("sync", "all"):
+        workers.append(
+            _worker([_noop], _SYNC_CONTROL_QUEUE, 1, 3600, sync_on_startup, sync_on_shutdown)
+        )
+    if role in ("jobs", "all"):
+        workers.append(
+            _worker(INTERACTIVE_FUNCTIONS, QUEUE_INTERACTIVE, s.interactive_concurrency, 120)
+        )
+        workers.append(_worker(BULK_FUNCTIONS, QUEUE_BULK, s.bulk_concurrency, 3600))
+    return workers
+
+
+async def _run() -> None:
+    logging.basicConfig(level=get_settings().log_level)
+    role = get_settings().worker_role
+    workers = build_workers(role)
+    logger.info("worker starting: role=%s (%d arq worker(s))", role, len(workers))
+
+    tasks = [asyncio.create_task(w.async_run()) for w in workers]
+    stop = asyncio.Event()
+    loop = asyncio.get_running_loop()
+    for sig in (signal.SIGINT, signal.SIGTERM):
+        loop.add_signal_handler(sig, stop.set)
+
+    await stop.wait()
+    logger.info("worker stopping: cancelling %d worker(s)", len(tasks))
+    for task in tasks:
+        task.cancel()  # arq runs each worker's on_shutdown in its finally
+    await asyncio.gather(*tasks, return_exceptions=True)
+
+    from app.db import dispose_engine
+    from app.queue import close_arq_pool
+    from app.redis_client import close_redis
+
+    await dispose_engine()
+    await close_redis()
+    await close_arq_pool()
+    logger.info("worker stopped")
+
+
+def main() -> None:
+    asyncio.run(_run())
+
+
+if __name__ == "__main__":
+    main()

backend/workers/worker.py

@@ -1,6 +1,8 @@
-"""arq worker entrypoint (SPEC §2 ``workers/worker.py``).
+"""Worker job functions and the sync-worker lifecycle (refactor §1, §6).
 
-Run with: ``arq workers.worker.WorkerSettings`` (from ``backend/``).
+The process is launched via ``workers.run`` (``WORKER_ROLE`` decides which arq
+workers run). This module only defines the job callables and the two queue
+function-sets; the runner wires them onto the ``interactive`` / ``bulk`` queues.
 """
 
 from __future__ import annotations
@@ -9,10 +11,8 @@ import logging
 import uuid
 from typing import Any
 
-from app.config import get_settings
 from app.outbox.draft_sync import push_draft_to_imap_job, remove_imap_draft_job
 from app.outbox.jobs import send_job
-from app.queue import redis_settings as arq_redis_settings
 from app.sync.actions import (
     delete_thread_job,
     move_job,
@@ -32,41 +32,33 @@ async def sync_bootstrap_account(ctx: dict[str, Any], account_id: str) -> int:
     return count
 
 
-async def on_startup(ctx: dict[str, Any]) -> None:
+# Fast, interactive user actions (§6): short timeout, high concurrency.
-    logging.basicConfig(level=get_settings().log_level)
+INTERACTIVE_FUNCTIONS = [
+    push_flags_job,
+    move_job,
+    move_thread_to_folder_job,
+    delete_thread_job,
+    send_job,
+    push_draft_to_imap_job,
+    remove_imap_draft_job,
+    refresh_account_job,
+]
+
+# Bulk work (§6): long timeout, low concurrency, isolated from interactive jobs.
+BULK_FUNCTIONS = [sync_bootstrap_account]
+
+
+async def sync_on_startup(ctx: dict[str, Any]) -> None:
+    """Sync-worker startup: launch the account-sync supervisor."""
     from app.sync.supervisor import start_supervisor
 
     await start_supervisor(ctx)
-    logger.info("webmail worker started")
+    logger.info("sync worker: supervisor started")
 
 
-async def on_shutdown(ctx: dict[str, Any]) -> None:
+async def sync_on_shutdown(ctx: dict[str, Any]) -> None:
-    from app.db import dispose_engine
+    """Sync-worker shutdown: stop the supervisor (process cleanup is in the runner)."""
-    from app.redis_client import close_redis
     from app.sync.supervisor import stop_supervisor
 
     await stop_supervisor(ctx)
-    await dispose_engine()
+    logger.info("sync worker: supervisor stopped")
-    await close_redis()
-    logger.info("webmail worker stopped")
-
-
-class WorkerSettings:
-    """arq worker configuration."""
-
-    functions = [
-        sync_bootstrap_account,
-        push_flags_job,
-        move_job,
-        move_thread_to_folder_job,
-        delete_thread_job,
-        send_job,
-        push_draft_to_imap_job,
-        remove_imap_draft_job,
-        refresh_account_job,
-    ]
-    on_startup = on_startup
-    on_shutdown = on_shutdown
-    max_jobs = 10  # global worker cap (SPEC §6.3)
-    job_timeout = 3600  # backfill of a large account can be long
-    redis_settings = arq_redis_settings()

docker/docker-compose.yml

@@ -36,6 +36,36 @@ services:
       timeout: 3s
       retries: 10
 
+  # Sync-engine v2 workers (refactor §1): two roles sharing one image. Scale the
+  # sync role with `--scale worker-sync=N` (lease protocol prevents double-owning).
+  worker-sync:
+    build: { context: ../backend }
+    command: ["python", "-m", "workers.run"]
+    environment: &worker-env
+      WORKER_ROLE: sync
+      APP_DOMAIN: dev.webmail.local
+      DATABASE_URL: postgresql+asyncpg://webmail:webmail_dev@postgres:5432/webmail_dev
+      REDIS_URL: redis://redis:6379/0
+      MASTER_KEY: CHANGE_ME_base64_32_bytes
+      BLOB_ROOT: /var/blobs
+    volumes: [blobs:/var/blobs]
+    ulimits: { nofile: 65536 }
+    depends_on:
+      postgres: { condition: service_healthy }
+      redis: { condition: service_healthy }
+
+  worker-jobs:
+    build: { context: ../backend }
+    command: ["python", "-m", "workers.run"]
+    environment:
+      <<: *worker-env
+      WORKER_ROLE: jobs
+    volumes: [blobs:/var/blobs]
+    ulimits: { nofile: 65536 }
+    depends_on:
+      postgres: { condition: service_healthy }
+      redis: { condition: service_healthy }
+
   # Test mail servers (integration tests + manual dev against a throwaway mailbox)
   greenmail:
     image: greenmail/standalone:2.0.1
@@ -51,3 +81,4 @@ services:
 volumes:
   pgdata:
   meilidata:
+  blobs:

sync-engine-v2-refactor.md

@@ -0,0 +1,119 @@
+# Sync Engine v2 — Refactor Instructions (replaces SPEC §6.3)
+
+This document is a self-contained repair/rebuild task for the implementing agent. It replaces the current single-supervisor sync design with a sharded, lease-owned, tiered sync engine targeting **1000+ accounts**. Read it fully, then execute the migration steps in §9 **in order**.
+
+## 0. Ground rules
+
+- This is a **workers-layer refactor only**. Frozen and untouchable: the `MailProvider` interface (SPEC §5.1), the database schema (no migration is needed or allowed by this task), the REST/WS API surface, invariant **I-1** (API never opens IMAP connections), and the idempotency discipline (`UNIQUE (mailbox_id, uid)` upserts).
+- Existing tests must stay green after every migration step. Each step ends with its own acceptance checks (§10 maps them).
+- All new tunables go through `config.py` (pydantic-settings) — no literals in code. New env vars are listed in §8.
+- Redis is already the queue and event bus; it now also carries leases, the live-account set, tier state, and worker heartbeats. No new infrastructure.
+
+## 1. Process model: two roles
+
+The single worker process splits into two entrypoints (same codebase, `WORKER_ROLE` env):
+
+- **`role=jobs`** — pure arq consumer for the discrete jobs (send, flags, move, drafts, backfill). Stateless, horizontally scalable; run N instances freely.
+- **`role=sync`** — runs the lease claimer + per-account runners (long-lived loops). Scalable via the lease protocol in §2; run 1..N instances.
+
+Default dev/prod-small topology: 2× sync + 2× jobs. docker-compose gets two services (`worker-sync`, `worker-jobs`) sharing the image, each with `ulimits: { nofile: 65536 }`.
+
+## 2. Lease protocol (account ownership)
+
+Exactly one sync worker owns each live account at any moment. Ownership is a Redis lease:
+
+- **Registry:** `sync:live` — a Redis SET of account ids with `state='live'`. Maintained by the API on every account state change AND reconciled against Postgres by each sync worker every 300 s (belt and braces; reconcile only adds/removes ids, never touches leases).
+- **Lease key:** `sync:lease:{account_id}` = `{worker_id}`, TTL `LEASE_TTL` (60 s). `worker_id` = `{hostname}:{pid}:{random8}` generated at startup.
+- **Claim:** `SET sync:lease:{id} {worker_id} NX EX {LEASE_TTL}` — success starts the account runner (§4).
+- **Claimer loop:** every 10 s + random jitter 0–5 s, while `owned < MAX_ACCOUNTS_PER_WORKER`: iterate `sync:live` (SSCAN), attempt claims on ids with no lease. Jitter prevents thundering-herd on mass restart.
+- **Renewal:** every `LEASE_RENEW` (20 s), for each owned account run this Lua atomically (never a plain EXPIRE):
+  ```lua
+  if redis.call('GET', KEYS[1]) == ARGV[1]
+  then return redis.call('EXPIRE', KEYS[1], ARGV[2]) else return 0 end
+  ```
+- **Fencing (hard rule):** if renewal returns 0 for an account (lease lost or taken), the worker cancels that account's runner tasks **immediately, before any further provider or DB operation for it**. In-flight operations are tolerable because all writes are idempotent upserts.
+- **Release:** on graceful stop or account disable, cancel the runner, then delete the lease with the owner-checked Lua (`GET==me → DEL`).
+- **Worker registry & rebalance:** each sync worker heartbeats `sync:workers:{worker_id}` = JSON `{owned, sockets_open, tiers:{hot,warm,cold}, ts}` (TTL 30 s, refresh 10 s). Every 60 s: `target = |sync:live| / |alive workers|`; if `owned > target*1.2 + 5`, release the excess, choosing least-recently-active accounts first — released accounts are claimed by under-loaded workers within one claimer cycle.
+
+**Failover story (emergent, no extra code):** worker crash → leases expire ≤ `LEASE_TTL` → survivors claim orphans on the next claimer tick. Total re-ownership within ~90 s.
+
+## 3. Event-driven account lifecycle (kills startup-only discovery + zombie loops)
+
+Sync workers subscribe to the event bus:
+
+- `account.state_changed → live` (new account finished backfill, or re-enabled): API adds the id to `sync:live`; some worker claims it within one claimer cycle. **No restart, ever.**
+- `account.state_changed → disabled|deleted`: API removes the id from `sync:live`; the owning worker (it can check `GET sync:lease:{id}==me` cheaply, or simply receives the event and checks ownership) stops the runner, closes connections, releases the lease — target ≤ 30 s from the API call.
+- Defense in depth: every runner poll cycle re-verifies the account row exists with `state='live'`; if not, self-terminate and release. This makes zombie loops structurally impossible even if an event is lost.
+- **New core event `user.active`** (add to SPEC §8.3 catalog): published by the API on successful login and on every WS connect, payload `{user_id}`. Consumed for tier promotion (§5).
+
+## 4. Per-account runner (replaces the fixed 2-task model)
+
+One `AccountRunner` per owned account, whose task set depends on the current tier:
+
+- **hot:** task A = INBOX IDLE loop (one persistent connection; re-issue IDLE ≤ 25 min; on wake → `poll()` INBOX). Task B = secondary CONDSTORE poll of Sent/Drafts/Archive/Trash/Junk every `POLL_WARM` s on a transient connection.
+- **warm:** no persistent connection. Single task: transient CONDSTORE poll of INBOX + special folders every `POLL_WARM` s.
+- **cold:** same single task, every `POLL_COLD` s.
+
+Delta application, UIDVALIDITY handling, expunge soft-handling, and outbound flag write-then-enqueue semantics are **unchanged** from the current implementation.
+
+**Connection budgets:**
+- ≤ 2 concurrent IMAP connections per account from the sync side (IDLE + one transient); jobs may add short-lived connections but must acquire the same per-account semaphore (cap 4 total, as today).
+- Per-destination-host semaphore (process-local), cap `PER_HOST_CONN_CAP` (default 50): acquired around every IMAP/SMTP connect, so a worker never opens hundreds of simultaneous sockets to one provider. Sharding by account makes process-local sufficient.
+
+## 5. Activity tiers
+
+Tier state: `sync:tier:{account_id}` = `hot|warm|cold` (TTL none; default `warm` on claim). Evaluated by the owning worker every 120 s and on events:
+
+- **→ hot** when: `user.active` received for the owning user (all their accounts promote within ≤ 5 s, IDLE established), OR the account received new mail in the last `TIER_HOT_ACTIVITY_H` (24 h). Session-based heat persists while WS pings continue + `TIER_SESSION_GRACE_MIN` (30 min) after the last one (API refreshes `sess:active:{user_id}` TTL; workers just read it).
+- **→ warm** when neither hot condition holds and last activity < `TIER_COLD_AFTER_D` (3 d).
+- **→ cold** when no user session and no new mail for ≥ `TIER_COLD_AFTER_D` days. Any new mail found by a cold poll immediately re-promotes to hot (mail arriving = activity).
+- Optional global guard: if a worker's hot count exceeds `HOT_BUDGET_PER_WORKER` (default 250), the least-recently-active hot accounts without live sessions degrade to warm. Session-hot accounts are never degraded by the budget.
+
+Expected steady state at ~3000 accounts: roughly 15–25 % hot → 450–750 persistent sockets across the pool, the rest on cheap transient polls. Correctness is identical in all tiers — only latency differs (seconds / ≤ `POLL_WARM` / ≤ `POLL_COLD`).
+
+## 6. Queue split (interactive vs bulk)
+
+Two arq queues, consumed by `role=jobs` workers:
+
+| Queue | Jobs | `job_timeout` | Concurrency per worker |
+|---|---|---|---|
+| `interactive` | send_job, push_flags_job, move_job, move_thread_to_folder_job, delete_thread_job, push_draft_to_imap_job, remove_imap_draft_job, refresh_account_job | 120 s | 20 |
+| `bulk` | sync_bootstrap_account (and future re-backfills) | 3600 s | `BULK_CONCURRENCY` (default 2) |
+
+Every jobs worker consumes both queues (arq supports it) with the above per-queue caps — a storm of backfills can no longer delay a send. Retry/backoff and failure-revert semantics per job are unchanged.
+
+## 7. Backfill memory bounds
+
+- Global backfill parallelism is bounded by the `bulk` queue concurrency (above) — remove any other ad-hoc caps.
+- Per-process parse semaphore `PARSE_CONCURRENCY` (default 4) around MIME parsing.
+- Streaming threshold: a raw message larger than `STREAM_PARSE_MB` (default 8) is written to the blob store first and parsed from the file handle; smaller messages may parse in memory. Result: peak RAM ≈ `PARSE_CONCURRENCY × STREAM_PARSE_MB` per process, independent of mailbox sizes.
+- `sync.progress` publication cadence unchanged (per batch).
+
+## 8. New configuration (add to SPEC §14; all via `config.py`)
+
+`WORKER_ROLE=jobs|sync` · `MAX_ACCOUNTS_PER_WORKER=400` · `LEASE_TTL=60` · `LEASE_RENEW=20` · `POLL_WARM=180` · `POLL_COLD=900` · `TIER_HOT_ACTIVITY_H=24` · `TIER_SESSION_GRACE_MIN=30` · `TIER_COLD_AFTER_D=3` · `HOT_BUDGET_PER_WORKER=250` · `PER_HOST_CONN_CAP=50` · `BULK_CONCURRENCY=2` · `PARSE_CONCURRENCY=4` · `STREAM_PARSE_MB=8`. Remove `POLL_INTERVAL` (superseded by POLL_WARM/POLL_COLD).
+
+## 9. Migration steps (execute in order; suite green after each)
+
+1. **Roles + queues.** Add `WORKER_ROLE`; move job registration to the two named queues with the §6 routing/timeouts; compose gets `worker-sync` / `worker-jobs` services with `nofile` ulimits. Supervisor behavior untouched.
+2. **Extract `AccountRunner`.** Refactor the current per-account loops into a self-contained runner class (start/stop/teardown API), still launched by the old startup supervisor. Pure refactor, no behavior change.
+3. **Lease manager + claimer.** Implement §2 (worker_id, claim, Lua renewal, fencing, release, heartbeats, rebalance). Replace the startup supervisor with the claimer. Verify single-worker behavior is equivalent, then run two sync workers against the same accounts.
+4. **Event-driven lifecycle.** Implement §3 including the new `user.active` event in the API and the per-cycle state re-verification. Delete the startup-only discovery code path.
+5. **Tiers.** Implement §5 tier storage, evaluation, promotions/demotions, and the tier-shaped task sets in `AccountRunner`.
+6. **Memory bounds.** Implement §7 streaming threshold and semaphores.
+7. **Observability + graceful shutdown.** Structured log fields (`worker_id`, `account_id`, `tier`), heartbeat JSON per §2, SIGTERM handler: stop claimer → cancel runners → release leases → drain (target < 10 s).
+
+## 10. Acceptance criteria (all must pass; add as automated tests where marked)
+
+1. **[test]** Two sync workers, 20 mock accounts: at no point do two workers hold the same lease; each account has exactly one IDLE connection (assert via mock provider connection counter).
+2. **[test]** `kill -9` one of two sync workers: every orphaned account is re-owned and syncing within 90 s; message table has zero duplicate `(mailbox_id, uid)` rows afterwards.
+3. **[test]** Create an account via the API: its live loop starts within 30 s **without any worker restart**. Delete an account: all its connections closed and no further provider calls within 30 s (zombie regression test — mock provider records post-deletion calls, must be zero after the 30 s cutoff).
+4. **[test]** Publish `user.active`: all of that user's accounts reach `hot` (IDLE established on mock) within 5 s. After session grace expiry with no activity, they demote to `warm` on the next evaluation cycle.
+5. **[test]** Queue isolation: enqueue 10 bulk backfills + 20 interactive sends; all sends complete in < 5 s p95 while backfills run.
+6. **[test]** SIGTERM a sync worker: leases released and adopted by the peer within 15 s; no fencing violations logged.
+7. **[load, manual/CI-nightly]** 1000 simulated accounts (mock provider), 20 % with active sessions: steady-state persistent sockets ≤ hot count + 5 %; worker RSS stable over 1 h; oldest-sync-lag < `POLL_COLD` for every account.
+8. Backfilling a mailbox containing a 40 MB message never raises process RSS by more than ~`PARSE_CONCURRENCY × STREAM_PARSE_MB` above baseline.
+
+## 11. Out of scope for this task
+
+Cross-worker (Redis-based) per-host connection accounting, pgbouncer deployment, JMAP provider, multi-region operation, per-user rate plans. Note them in `docs/open-questions.md` if the need is observed.

webmail-client-spec_v2.md

@@ -0,0 +1,744 @@
+# Webmail Client — Implementation Specification (v1.3, agent-ready)
+
+This document is a complete, self-contained build specification. It is written to be handed to an AI coding agent. Read it fully before writing code. The companion document `webmail-client-concept.md` gives background; where they differ, **this spec wins**.
+
+---
+
+## 0. Instructions to the implementing agent
+
+1. **Build milestone by milestone** (§12), in order. Do not start a milestone before the previous one's acceptance criteria pass.
+2. **Do not add features beyond this spec.** Anything not specified here is out of scope; leave `TODO(spec)` comments instead of inventing behavior.
+3. **Definition of done** for every task: code + type hints + unit tests + passing `ruff check`, `ruff format --check`, `mypy --strict` (backend) / `tsc --noEmit`, `eslint` (frontend) + updated docs in `docs/` if an interface changed.
+4. **Stability over novelty:** use exactly the pinned dependencies in §3. Do not swap libraries or add dependencies without a `# DEPENDENCY:` comment explaining why, and never for something the stdlib does adequately.
+5. **All interfaces in §5 are frozen contracts.** Implementations may evolve; signatures may not (without bumping the version comment at the top of the interface file).
+6. **Security rules in §10 are hard requirements**, not suggestions.
+7. Every module publishing or consuming events must use the event names in §8 verbatim.
+8. Commit in small, working increments; every commit must leave the test suite green.
+
+## 1. Architecture invariants (never violate)
+
+- **I-1 Sync-first:** the web UI and REST API never open IMAP connections. Only sync workers talk to mail servers. The API reads exclusively from Postgres/Meilisearch.
+- **I-2 Independent identity:** login uses app accounts (`localpart@our-web-client.example`) with Argon2id hashes. Mailbox credentials are payload data, encrypted at rest, decrypted only inside sync workers.
+- **I-3 One index, many views:** unified inbox, per-account views, virtual mailboxes, and search results are all parameterized queries over the same index. No duplicated storage per view.
+- **I-4 Threading is ours:** JWZ in our backend (§7). Never use the IMAP `THREAD` extension.
+- **I-5 Direction-agnostic thread UI:** thread ordering (oldest-first/newest-first) is a data-ordering concern. One layout, one composer-anchor slot (§9.4).
+- **I-6 Plugins extend, core stays closed:** new features beyond this spec must be expressible through the plugin surfaces (§11).
+- **I-7 Multi-user isolation:** every query (SQL and search) carries a mandatory `user_id` filter. No exceptions, including admin tooling.
+- **I-8 Mail HTML is hostile input:** sanitize with nh3 server-side at index time AND render in a sandboxed iframe. Both, always.
+
+## 2. Repository layout (monorepo)
+
+```
+webmail/
+├── backend/
+│   ├── pyproject.toml
+│   ├── app/
+│   │   ├── main.py                  # FastAPI app factory
+│   │   ├── config.py                # pydantic-settings; all config via env
+│   │   ├── db.py                    # SQLAlchemy async engine/session
+│   │   ├── events.py                # EventBus (Redis pub/sub + arq enqueue)
+│   │   ├── identity/                # §6.1 auth, sessions, handles, TOTP
+│   │   ├── accounts/                # §6.2 wizard, autodiscovery, credential vault
+│   │   ├── providers/
+│   │   │   ├── base.py              # MailProvider interface (§5.1)
+│   │   │   └── imap_smtp/           # first implementation
+│   │   ├── sync/                    # §6.3 backfill, IDLE loops, reconciliation
+│   │   ├── indexer/                 # §6.4 MIME parsing → DB rows + blobs
+│   │   ├── threading_/              # §7 JWZ (trailing _ avoids stdlib clash)
+│   │   ├── search/
+│   │   │   ├── base.py              # SearchBackend interface (§5.2)
+│   │   │   ├── queryparser.py       # operator grammar (§8.2)
+│   │   │   ├── pg.py                # Postgres FTS backend (M0–M1)
+│   │   │   └── meili.py             # Meilisearch backend (M2)
+│   │   ├── outbox/                  # §6.5 drafts, MIME assembly, send queue, undo
+│   │   ├── api/                     # routers, one file per §9.1 group
+│   │   ├── ws.py                    # WebSocket hub (§9.2)
+│   │   └── plugins/                 # §11 loader + plugin packages
+│   ├── workers/worker.py            # arq worker entrypoint
+│   ├── migrations/                  # alembic
+│   └── tests/
+├── frontend/
+│   ├── package.json
+│   └── src/
+│       ├── api/                     # generated client from OpenAPI + WS layer
+│       ├── routes/                  # §9.3
+│       ├── components/
+│       │   ├── thread/ThreadView.tsx    # §9.4 direction-agnostic
+│       │   ├── thread/Composer.tsx
+│       │   ├── list/MessageList.tsx     # virtualized
+│       │   └── slots/               # plugin UI slots (§11)
+│       └── lib/
+├── docker/
+│   ├── docker-compose.yml           # app, worker, postgres:16, redis:7, meilisearch:v1.8, dovecot (test)
+│   └── docker-compose.ci.yml        # + greenmail for integration tests
+└── docs/
+```
+
+## 3. Pinned stack
+
+**Backend (Python 3.11):** fastapi ~0.115, uvicorn ~0.30, sqlalchemy ~2.0 (async, asyncpg), alembic ~1.13, pydantic ~2.8, pydantic-settings, arq ~0.26, redis ~5.0, aioimaplib ~1.1, imapclient ~3.0 (utility/fallback), aiosmtplib ~3.0, argon2-cffi ~23.1, pyotp ~2.9, dnspython ~2.6, httpx ~0.27, nh3 ~0.2, cryptography ~43.0, mail-parser ~4.0 (helpers only; primary parsing via stdlib `email.message.EmailMessage` with `policy.default`).
+
+**Frontend:** TypeScript ~5.5, React ~18.3, Vite ~5, TanStack Query ~5, TanStack Virtual ~3, Radix UI, Tailwind ~3.4, zod (WS payload validation). API client generated from OpenAPI (`openapi-typescript`).
+
+**Infra:** PostgreSQL 16 (with `citext` extension), Redis 7, Meilisearch v1.8 (M2+), MinIO or plain filesystem for the blob store (config-switchable; filesystem default).
+
+## 4. Database schema (Postgres 16, authoritative DDL)
+
+All PKs are `uuid DEFAULT gen_random_uuid()`. All tables carrying user data include `user_id` with an index. Timestamps are `timestamptz`.
+
+```sql
+CREATE EXTENSION IF NOT EXISTS citext;
+
+CREATE TABLE users (
+  id            uuid PRIMARY KEY DEFAULT gen_random_uuid(),
+  handle        citext UNIQUE NOT NULL,        -- full form: localpart@<APP_DOMAIN>
+  password_hash text   NOT NULL,               -- argon2id
+  totp_secret   bytea,                         -- encrypted; NULL = 2FA off
+  recovery_email citext,
+  settings      jsonb  NOT NULL DEFAULT '{}',  -- incl. {"thread_order":"asc"|"desc"}
+  created_at    timestamptz NOT NULL DEFAULT now()
+);
+-- Handle rules (enforce in code, mail-grade from day one):
+--   localpart: 3–30 chars, [a-z0-9._-], no leading/trailing/double dot,
+--   case-insensitive, reserved list (postmaster, abuse, admin, root, noreply, ...).
+
+CREATE TABLE accounts (
+  id             uuid PRIMARY KEY DEFAULT gen_random_uuid(),
+  user_id        uuid NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+  email          citext NOT NULL,
+  display_name   text NOT NULL DEFAULT '',
+  color          text NOT NULL DEFAULT '#888888',
+  imap_config    jsonb NOT NULL,   -- {host,port,tls:"ssl"|"starttls",auth:"password"|"oauth2"}
+  smtp_config    jsonb NOT NULL,   -- same shape
+  secret_ct      bytea NOT NULL,   -- envelope-encrypted password or OAuth token bundle (§10.2)
+  discovery_src  text  NOT NULL,   -- preset|autoconfig|ispdb|srv|probe|manual
+  state          text  NOT NULL DEFAULT 'backfilling',  -- backfilling|live|error|disabled
+  state_detail   text  NOT NULL DEFAULT '',
+  last_sync_at   timestamptz,
+  created_at     timestamptz NOT NULL DEFAULT now(),
+  UNIQUE (user_id, email)
+);
+
+CREATE TABLE mailboxes (
+  id             uuid PRIMARY KEY DEFAULT gen_random_uuid(),
+  account_id     uuid NOT NULL REFERENCES accounts(id) ON DELETE CASCADE,
+  user_id        uuid NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+  path           text NOT NULL,                 -- IMAP mailbox name, UTF-7 decoded
+  special_use    text,                          -- inbox|sent|drafts|trash|junk|archive|NULL
+  uidvalidity    bigint NOT NULL DEFAULT 0,
+  highestmodseq  bigint,                        -- NULL if no CONDSTORE
+  last_seen_uid  bigint NOT NULL DEFAULT 0,
+  UNIQUE (account_id, path)
+);
+
+CREATE TABLE threads (
+  id              uuid PRIMARY KEY DEFAULT gen_random_uuid(),
+  user_id         uuid NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+  subject_norm    text NOT NULL,                -- normalized subject (§7.3)
+  snippet         text NOT NULL DEFAULT '',
+  participants    jsonb NOT NULL DEFAULT '[]',  -- [{name,email}], deduped, ordered by recency
+  account_ids     uuid[] NOT NULL DEFAULT '{}',
+  message_count   int NOT NULL DEFAULT 0,
+  unread_count    int NOT NULL DEFAULT 0,
+  has_attachments boolean NOT NULL DEFAULT false,
+  last_message_at timestamptz NOT NULL
+);
+CREATE INDEX ON threads (user_id, last_message_at DESC);
+
+CREATE TABLE messages (
+  id              uuid PRIMARY KEY DEFAULT gen_random_uuid(),
+  user_id         uuid NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+  account_id      uuid NOT NULL REFERENCES accounts(id) ON DELETE CASCADE,
+  mailbox_id      uuid NOT NULL REFERENCES mailboxes(id) ON DELETE CASCADE,
+  uid             bigint NOT NULL,              -- IMAP UID within mailbox
+  thread_id       uuid REFERENCES threads(id),
+  message_id_hdr  text,                         -- Message-ID, angle brackets stripped
+  in_reply_to     text,
+  references_hdr  text[] NOT NULL DEFAULT '{}',
+  subject         text NOT NULL DEFAULT '',
+  from_addr       jsonb NOT NULL DEFAULT '[]',
+  to_addr         jsonb NOT NULL DEFAULT '[]',
+  cc_addr         jsonb NOT NULL DEFAULT '[]',
+  date_hdr        timestamptz,
+  internal_date   timestamptz NOT NULL,
+  flags           text[] NOT NULL DEFAULT '{}', -- \Seen,\Flagged,\Answered,\Draft,\Deleted
+  size_bytes      int NOT NULL DEFAULT 0,
+  has_attachments boolean NOT NULL DEFAULT false,
+  snippet         text NOT NULL DEFAULT '',     -- first ~160 chars of text body
+  blob_sha256     text NOT NULL,                -- raw RFC822 in blob store (§6.4)
+  body_indexed    tsvector,                     -- Postgres FTS (M0–M1 SearchBackend)
+  UNIQUE (mailbox_id, uid)
+);
+CREATE INDEX ON messages (user_id, thread_id);
+CREATE INDEX ON messages (account_id, message_id_hdr);
+CREATE INDEX ON messages USING gin (body_indexed);
+
+CREATE TABLE body_parts (
+  id          uuid PRIMARY KEY DEFAULT gen_random_uuid(),
+  message_id  uuid NOT NULL REFERENCES messages(id) ON DELETE CASCADE,
+  kind        text NOT NULL,        -- text|html_sanitized
+  content     text NOT NULL
+);
+
+CREATE TABLE attachments (
+  id          uuid PRIMARY KEY DEFAULT gen_random_uuid(),
+  message_id  uuid NOT NULL REFERENCES messages(id) ON DELETE CASCADE,
+  user_id     uuid NOT NULL,
+  filename    text NOT NULL,
+  mime_type   text NOT NULL,
+  size_bytes  int  NOT NULL,
+  content_id  text,                 -- for inline cid: images
+  blob_sha256 text NOT NULL
+);
+
+CREATE TABLE drafts (
+  id          uuid PRIMARY KEY DEFAULT gen_random_uuid(),
+  user_id     uuid NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+  account_id  uuid NOT NULL REFERENCES accounts(id) ON DELETE CASCADE,
+  thread_id   uuid REFERENCES threads(id),      -- NULL = new mail
+  reply_to_message_id uuid REFERENCES messages(id),
+  kind        text NOT NULL DEFAULT 'reply',    -- reply|reply_all|forward|new
+  payload     jsonb NOT NULL,       -- {to,cc,bcc,subject,html,text,attachment_ids}
+  updated_at  timestamptz NOT NULL DEFAULT now()
+);
+
+CREATE TABLE outbox (
+  id          uuid PRIMARY KEY DEFAULT gen_random_uuid(),
+  user_id     uuid NOT NULL,
+  draft_id    uuid NOT NULL,
+  state       text NOT NULL DEFAULT 'undo_window',  -- undo_window|sending|sent|failed|cancelled
+  send_after  timestamptz NOT NULL,                 -- now() + undo window
+  error       text
+);
+
+CREATE TABLE saved_searches (
+  id       uuid PRIMARY KEY DEFAULT gen_random_uuid(),
+  user_id  uuid NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+  name     text NOT NULL,
+  query    text NOT NULL,           -- raw operator-syntax string (§8.2)
+  pinned   boolean NOT NULL DEFAULT true,
+  UNIQUE (user_id, name)
+);
+
+-- Core recipient autocomplete (Gmail-style, derived from mail history).
+-- The Contacts plugin (Appendix B) ENRICHES suggestions; this table works without it.
+CREATE TABLE known_recipients (
+  user_id      uuid NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+  email        citext NOT NULL,
+  name         text NOT NULL DEFAULT '',      -- last seen display name
+  times_used   int NOT NULL DEFAULT 0,        -- weighted: sent-to counts 3, received-from counts 1
+  last_used_at timestamptz NOT NULL,
+  PRIMARY KEY (user_id, email)
+);
+CREATE INDEX ON known_recipients (user_id, email text_pattern_ops);
+```
+
+**Blob store:** raw RFC822 messages and attachment payloads are stored content-addressed at `${BLOB_ROOT}/{user_id}/{sha256[:2]}/{sha256}` (filesystem) or the equivalent object key (MinIO). Postgres holds metadata only. Deleting a user removes their blob subtree.
+
+**Sessions** live in Redis (`sess:{token}` → JSON `{user_id, created, ua}`, TTL 30 days sliding), not Postgres.
+
+## 5. Frozen core interfaces (Python)
+
+### 5.1 `providers/base.py` — MailProvider
+
+```python
+# INTERFACE VERSION: 1
+from dataclasses import dataclass
+from typing import AsyncIterator, Protocol
+from datetime import datetime
+
+@dataclass(frozen=True)
+class RemoteMailbox:
+    path: str
+    special_use: str | None      # "inbox"|"sent"|"drafts"|"trash"|"junk"|"archive"|None
+    uidvalidity: int
+
+@dataclass(frozen=True)
+class RemoteMessage:
+    uid: int
+    raw: bytes                   # full RFC822
+    flags: frozenset[str]
+    internal_date: datetime
+
+@dataclass(frozen=True)
+class FlagChange:
+    uid: int
+    flags: frozenset[str]
+
+@dataclass(frozen=True)
+class MailboxDelta:
+    new: list[RemoteMessage]
+    flag_changes: list[FlagChange]
+    expunged_uids: list[int]
+    new_uidvalidity: int | None  # non-None => UIDVALIDITY changed, caller must full-resync
+    new_highestmodseq: int | None
+
+class MailProvider(Protocol):
+    async def verify(self) -> None: ...
+        # Raises ProviderAuthError / ProviderConnectError with a human-readable .detail
+
+    async def list_mailboxes(self) -> list[RemoteMailbox]: ...
+
+    def backfill(self, mailbox: str, *, newest_first: bool = True,
+                 batch_size: int = 200) -> AsyncIterator[list[RemoteMessage]]: ...
+        # Yields batches until the mailbox is exhausted. Headers+body in one pass.
+
+    async def poll(self, mailbox: str, *, last_seen_uid: int,
+                   highestmodseq: int | None) -> MailboxDelta: ...
+        # One reconciliation pass. Uses QRESYNC/CONDSTORE when highestmodseq is not None,
+        # otherwise UID SEARCH + FETCH FLAGS diff.
+
+    async def idle(self, mailbox: str) -> None: ...
+        # Blocks until the server signals change or 25 min elapse (re-issue IDLE). May raise
+        # ProviderNotSupported => sync loop falls back to poll() every POLL_INTERVAL (default 120 s).
+
+    async def push_flags(self, mailbox: str, uid: int,
+                         add: frozenset[str], remove: frozenset[str]) -> None: ...
+    async def move(self, uid: int, src: str, dst: str) -> int: ...   # returns new UID
+    async def append(self, mailbox: str, raw: bytes,
+                     flags: frozenset[str]) -> int: ...               # Sent/Drafts upload
+    async def send(self, raw: bytes, envelope_from: str,
+                   envelope_to: list[str]) -> None: ...               # SMTP submission
+```
+
+`ImapSmtpProvider` is the only M0–M2 implementation. A future `NativeProvider` (hosted mailboxes) or `JmapProvider` implements the same protocol — nothing above this interface may import IMAP libraries.
+
+### 5.2 `search/base.py` — SearchBackend
+
+```python
+# INTERFACE VERSION: 1
+@dataclass(frozen=True)
+class SearchQuery:
+    text: str | None                      # free-text remainder after operator extraction
+    filters: dict[str, object]            # parsed operators, keys per §8.2
+    user_id: UUID                         # ALWAYS set (invariant I-7)
+
+@dataclass(frozen=True)
+class SearchHit:
+    message_id: UUID
+    thread_id: UUID
+    score: float
+    highlights: dict[str, str]            # field -> HTML-escaped snippet with <mark>
+
+class SearchBackend(Protocol):
+    async def index(self, doc: SearchDoc) -> None: ...
+    async def delete(self, message_id: UUID) -> None: ...
+    async def delete_user(self, user_id: UUID) -> None: ...
+    async def query(self, q: SearchQuery, *, limit: int, cursor: str | None
+                    ) -> tuple[list[SearchHit], str | None]: ...
+```
+
+`SearchDoc` fields: `message_id, thread_id, user_id, account_id, mailbox_special_use, from_ (name+email), to, cc, subject, body_text, attachment_names[], has_attachments, is_unread, is_starred, date (unix), size_bytes`. Both backends (pg.py, meili.py) index the same doc. Meilisearch: one index `messages`, filterable = every non-body field incl. `user_id`, searchable = `subject, body_text, from_, to, cc, attachment_names`, sortable = `date`; ranking rules default + `date:desc` tiebreak.
+
+### 5.3 `events.py` — EventBus
+
+```python
+# INTERFACE VERSION: 1
+class EventBus(Protocol):
+    async def publish(self, event: str, payload: dict) -> None:
+        # Fan-out to Redis pub/sub channel "evt:{event}" AND enqueue arq job
+        # "on_{event}" for durable consumers (indexer, plugins).
+    def subscribe(self, event: str, handler: Handler) -> None: ...
+```
+
+Payloads always include `user_id`. Event names are listed in §8.3 and are the only ones allowed.
+
+## 6. Backend module specifications
+
+### 6.1 Identity (`app/identity/`)
+- Registration: validate localpart per §4 rules against reserved list in `identity/reserved.py`; store full handle `localpart@{APP_DOMAIN}` (env `APP_DOMAIN`).
+- Login: constant-time verify (argon2id, `time_cost=3, memory_cost=64MiB, parallelism=4`); if `totp_secret` set, require valid TOTP (±1 window). Rate limit: 10 attempts / 15 min per handle AND per IP (Redis counters); lockout returns 429 with `Retry-After`.
+- Session cookie: `__Host-session`, httpOnly, Secure, SameSite=Lax. CSRF: double-submit token header `X-CSRF` for all mutating requests.
+- TOTP setup: `POST /auth/totp/setup` returns provisioning URI + QR payload; enable requires one valid code; secret stored envelope-encrypted.
+- Password reset: only if `recovery_email` verified; single-use token, 30 min TTL, Redis-backed.
+
+### 6.2 Accounts & autodiscovery (`app/accounts/`)
+`discover(email) -> list[CandidateConfig]` runs the chain, **returning on first success** but recording every step's outcome in the response (`steps: [{method, status, detail}]`) for wizard UX:
+
+1. **Presets** — `accounts/presets.yaml` (config, not code): match by domain; ships with gmail.com, googlemail.com, outlook.com/hotmail/live, office365, yahoo, icloud/me.com, gmx, wp.pl, onet.pl, o2.pl, interia.pl. Each entry: imap/smtp host+port+tls, auth kind, oauth provider id if applicable.
+2. **Autoconfig** — GET `https://autoconfig.{domain}/mail/config-v1.1.xml?emailaddress={email}` then `https://{domain}/.well-known/autoconfig/mail/config-v1.1.xml`; parse Thunderbird XML (incomingServer type=imap, outgoingServer type=smtp; `%EMAILADDRESS%`/`%EMAILLOCALPART%` substitution).
+3. **ISPDB** — GET `https://autoconfig.thunderbird.net/v1.1/{domain}`; same XML format.
+4. **DNS SRV (RFC 6186)** — `_imaps._tcp.{domain}` (SSL 993), `_imap._tcp` (STARTTLS 143), `_submissions._tcp` (465), `_submission._tcp` (587); lowest priority wins.
+5. **Probe** — try `{imap,mail}.{domain}` × ports 993(ssl)/143(starttls) and `{smtp,mail}.{domain}` × 465(ssl)/587(starttls); TCP connect + TLS handshake + IMAP/SMTP greeting, 5 s timeout each, all concurrently.
+
+All HTTP/DNS lookups: 5 s timeout, no redirects off-domain, HTTPS only. `test(config, credentials)`: full IMAP LOGIN + `LIST`, and SMTP EHLO+AUTH (no message sent); errors mapped to precise user-facing strings (`auth_failed`, `tls_failed`, `connect_timeout`, `unknown_host`). On save: encrypt credentials (§10.2), create account row (`state=backfilling`), enqueue `sync.bootstrap_account`.
+
+### 6.3 Sync engine v2 (`app/sync/`) — sharded, lease-owned, tiered
+(Authoritative detail: `sync-engine-v2-refactor.md`; this section summarizes the binding rules.)
+
+- **Two worker roles** (`WORKER_ROLE`): `jobs` = stateless arq consumer, N instances; `sync` = lease claimer + per-account runners, 1..N instances. Compose services `worker-jobs` / `worker-sync`, `nofile` ulimit 65536.
+- **Lease ownership:** exactly one sync worker owns each live account. Redis SET `sync:live` (API-maintained, DB-reconciled every 300 s); lease `sync:lease:{account_id}` claimed `SET NX EX LEASE_TTL`, renewed every `LEASE_RENEW` via owner-checked Lua; **fencing** — on renewal failure, cancel that account's runner before any further provider/DB op. Worker heartbeats in `sync:workers:{id}`; rebalance releases excess above `target*1.2+5`. Crash failover = lease expiry + reclaim (≤ ~90 s), no coordinator.
+- **Event-driven lifecycle:** `account.state_changed` adds/removes ids from `sync:live` at runtime — accounts go live ≤ 30 s after creation and stop ≤ 30 s after deletion with no restarts; runners re-verify `state='live'` each cycle (zombie-proof). New core event **`user.active`** (API publishes on login/WS connect) drives tier promotion.
+- **Bootstrap (unchanged from v1, now a `bulk` job):** `list_mailboxes()` → upsert `mailboxes` (SPECIAL-USE mapping with name heuristics — Sent/Wysłane, Trash/Kosz, etc. in `sync/folder_names.py`); backfill newest-first per mailbox (INBOX first, then sent, then rest); `sync.progress` published per batch; on completion set `state='live'` → the API adds the id to `sync:live` and a sync worker claims it.
+- **AccountRunner per tier:** *hot* = persistent INBOX IDLE + transient CONDSTORE poll of special folders every `POLL_WARM`; *warm* = transient poll of INBOX+specials every `POLL_WARM`; *cold* = every `POLL_COLD`. Promotion to hot on `user.active` (≤ 5 s) or new mail within `TIER_HOT_ACTIVITY_H`; demotion by inactivity; `HOT_BUDGET_PER_WORKER` degrades least-active non-session accounts. Delta application, UIDVALIDITY wipe-and-rebackfill, expunge soft-handling, and write-then-enqueue outbound flags are unchanged from v1.
+- **Connection budgets:** ≤ 2 sync-side connections per account (4 total incl. jobs, shared semaphore); process-local per-destination-host semaphore cap `PER_HOST_CONN_CAP`.
+- **Queues:** `interactive` (send/flags/move/draft jobs, timeout 120 s, concurrency 20) and `bulk` (backfills, timeout 3600 s, concurrency `BULK_CONCURRENCY`) — bulk work can never delay a send.
+- **Memory bounds:** parse semaphore `PARSE_CONCURRENCY`; raw messages > `STREAM_PARSE_MB` stream to blob and parse from file — peak RAM ≈ `PARSE_CONCURRENCY × STREAM_PARSE_MB` per process regardless of mailbox size.
+- **Shutdown:** SIGTERM → stop claimer → cancel runners → release leases → drain in < 10 s.
+
+### 6.4 Indexer (`app/indexer/`)
+Input: `RemoteMessage`. Steps: store raw blob (sha256, dedupe) → parse with `email` stdlib (`policy.default`) → extract: headers (decode RFC2047), `references_hdr` = parsed `References` + `In-Reply-To` message-ids in order; best text body and best HTML body (multipart/alternative preference: html for display, text for indexing; if only HTML, derive text via nh3 strip); sanitize HTML with nh3 (allowlist: structural tags, inline styles filtered, `cid:` images rewritten to `/api/messages/{id}/cid/{content_id}`, remote `src` rewritten to blocked-placeholder attributes for FR-23) → write `messages`, `body_parts` (kinds `text` and `html_sanitized`), `attachments` → hand to Threader (§7) → `bus.publish("message.new", ...)` (search indexing + thread denormalization are consumers). For non-junk received messages, upsert the `From` address into `known_recipients` (weight 1).
+
+### 6.5 Outbox (`app/outbox/`)
+- Drafts: autosave endpoint upserts `drafts` (client debounces 2 s); a background job mirrors drafts to the account's Drafts folder via `append()` at most every 60 s per draft (replace previous UID).
+- Send: `POST /send {draft_id}` → build MIME (html + auto-generated text alternative via html-to-text; `In-Reply-To`/`References` from `reply_to_message_id`; attachments from blob store) → create `outbox` row with `send_after = now()+UNDO_SECONDS` (env, default 10) → schedule arq job at `send_after`. Undo endpoint flips state to `cancelled` if still in window. The job: SMTP `send()`, then `append()` to Sent (skip for Gmail preset — Gmail self-appends), mark `sent`, publish `message.sent`, and upsert every To/Cc/Bcc address into `known_recipients` (weight 3). Failure → `failed` + WS notification, draft preserved.
+
+## 7. Threading specification (JWZ, `app/threading_/`)
+
+### 7.1 Normalization
+- `message_id_hdr`, `in_reply_to`, each `References` entry: strip angle brackets and whitespace, lowercase the domain part only. Discard obviously invalid ids (no `@`).
+- Missing/duplicate `Message-ID`: synthesize `sha256(raw)[:32]@synthetic.local` and mark it non-authoritative (never used to join other messages to it).
+
+### 7.2 Incremental algorithm (per new message)
+1. Candidate set = all normalized ids in `references_hdr` + `in_reply_to` + own `message_id_hdr`.
+2. `SELECT DISTINCT thread_id FROM messages WHERE user_id=$1 AND message_id_hdr = ANY($2) AND thread_id IS NOT NULL`.
+3. 0 threads → subject fallback (§7.3); still none → create new thread.
+4. 1 thread → attach.
+5. ≥2 threads → **merge**: keep the thread with the oldest `last_message_at` as canonical, repoint messages of the others, delete emptied thread rows, publish `thread.updated` for all involved.
+6. Recompute denormalized thread fields (count, unread, participants, snippet = newest message's snippet, has_attachments, account_ids, last_message_at) in one UPDATE.
+
+Threads are cross-account by design: the candidate lookup is scoped by `user_id`, not `account_id` (invariant I-3). Sent-folder messages join threads exactly like received ones (FR-15).
+
+### 7.3 Subject fallback
+Only when reference joining fails: normalize subject (strip repeated `Re:/Fw:/Fwd:/Odp:/PD:` prefixes case-insensitively, collapse whitespace); join to a thread with the same `subject_norm` that has activity within 7 days AND shares ≥1 participant email; never join two messages that both lack references AND have empty subjects.
+
+### 7.4 Test vectors (must pass in `tests/threading/`)
+Linear chain; chain arriving out of order; reply missing `In-Reply-To` but carrying `References`; thread merge on late-arriving parent; broken client (subject-only) joins; distinct-topic same-subject ("hi") does NOT join; cross-account merge (message in account A references message in account B); duplicate Message-ID; synthetic-ID isolation.
+
+## 8. Search specification
+
+### 8.1 Pipeline
+Consumer of `message.new` / `message.changed` / `message.deleted` → builds `SearchDoc` → active `SearchBackend` (env `SEARCH_BACKEND=pg|meili`). Backfill indexes inline with sync (per batch), never as a separate pass the user waits on (FR-30).
+
+### 8.2 Query grammar (`search/queryparser.py`, engine-independent)
+Tokenization: whitespace-split, respecting `"quoted phrases"`. A token matching `key:value` (or `-key:value`) with a known key becomes a filter; everything else joins free text. `-` prefix negates a term or filter. `OR` (uppercase) groups adjacent terms/filters into a disjunction; default combinator is AND.
+
+| Operator | Filter semantics |
+|---|---|
+| `from:` `to:` `cc:` | substring match on name or email of that field |
+| `subject:` | match constrained to subject |
+| `account:` | account by `display_name` or email substring |
+| `in:` | `inbox\|sent\|drafts\|trash\|junk\|archive` (special_use) or folder path |
+| `is:` | `unread\|read\|starred\|answered` |
+| `has:attachment` | boolean |
+| `filename:` | attachment name substring |
+| `before:` `after:` | `YYYY-MM-DD` |
+| `newer_than:` `older_than:` | `Nd\|Nw\|Nm\|Ny` |
+| `larger:` `smaller:` | `NK\|NM` bytes |
+
+Parser output = `SearchQuery` (§5.2). Unknown `key:` tokens are treated as free text (forward compatibility). Unit tests must cover every operator, negation, quoting, `OR`, and garbage input (parser never raises).
+
+### 8.3 Event catalog (complete)
+`message.new`, `message.changed`, `message.deleted`, `message.sent`, `flags.changed`, `thread.updated`, `sync.progress`, `account.state_changed`, `draft.saved`, `outbox.state_changed`, `user.active` (published by the API on login and WS connect; consumed by sync workers for tier promotion — not forwarded to browsers). Payloads: `{user_id, ...ids, minimal-diff fields}` — documented in `docs/events.md` (agent: generate it from code).
+
+Plugins may additionally publish events namespaced `plugin.{name}.*` (e.g. `plugin.calendar.event_updated`); core code never subscribes to plugin events, but the WS hub forwards them to the owning user like any other event.
+
+## 9. API, realtime, frontend
+
+### 9.1 REST API (prefix `/api`, session-cookie auth, JSON; errors: `{error:{code,message}}`)
+
+**Auth:** `POST /auth/register` · `POST /auth/login` · `POST /auth/logout` · `GET /auth/me` · `GET /auth/sessions` · `DELETE /auth/sessions/{id}` · `POST /auth/totp/setup` · `POST /auth/totp/enable` · `POST /auth/password/reset-request` · `POST /auth/password/reset`
+
+**Accounts:** `POST /accounts/discover {email}` → `{candidates:[{imap,smtp,source}], steps:[...]}` · `POST /accounts/test {config, credentials}` → `{ok}|{error:{code,detail}}` · `POST /accounts` · `GET /accounts` (incl. `state`, sync progress) · `PATCH /accounts/{id}` · `DELETE /accounts/{id}` (cascades rows, blobs, search docs)
+
+**Mail:** `GET /threads?view=unified|account:{id}|mailbox:{id}|search&q=&cursor=&limit=50` → thread summaries (denormalized fields only; cursor = `last_message_at,id` keyset) · `GET /threads/{id}?sort=asc|desc` → ordered full messages (bodies of the N newest expanded per client hint `expand=`; older = envelope only) · `POST /threads/{id}/flags {add,remove}` (whole thread) · `POST /messages/{id}/flags` · `POST /messages/{id}/move {mailbox_id}` · `GET /messages/{id}/html` (sanitized, for iframe `srcdoc`) · `GET /messages/{id}/cid/{content_id}` · `GET /attachments/{id}` (Content-Disposition attachment; images/PDF also inline for preview) · `POST /messages/{id}/allow-remote-images` (per-sender allowlist in user settings)
+
+**Compose:** `POST /drafts` · `PUT /drafts/{id}` · `DELETE /drafts/{id}` · `POST /send {draft_id}` → `{outbox_id, undo_until}` · `POST /send/{outbox_id}/undo` · `GET /recipients/suggest?q=` → top 10 from `known_recipients` (prefix match on email or name, ranked by `times_used` decayed by recency); the composer merges these with any plugin capability providers (§11.3)
+
+**Search:** `GET /search?q=&cursor=` → `{threads:[...], hits_by_thread, highlights}` (grouped by thread, FR-28; `&group=message` for flat) · `GET/POST/DELETE /saved-searches`
+
+**Settings:** `GET/PATCH /settings` (incl. `thread_order: "asc"|"desc"`, `tz` IANA timezone — default from browser at registration, undo window, per-sender image allowlist)
+
+### 9.2 WebSocket `/api/ws`
+Auth via session cookie. Server → client frames: `{event, payload}` for every §8.3 event belonging to the user (that's the whole contract — the UI refetches via TanStack Query invalidation keyed on ids in the payload). Client → server: `{type:"ping"}` only. Zod-validate all frames.
+
+### 9.3 Frontend routes
+`/login`, `/register`, `/mail` (unified inbox), `/mail/a/:accountId`, `/mail/a/:accountId/f/:mailboxId`, `/mail/t/:threadId` (thread pane; on desktop rendered beside the list), `/mail/search?q=`, `/settings/{profile|security|accounts|appearance}`; `/settings/accounts/new` = wizard.
+
+**Account wizard flow:** step 1 email (+ password field revealed unless preset says oauth) → calls `discover` while typing pauses; step 2 shows resolved config with source badge ("found via ISPDB") and a Test button (auto-run); failure expands the manual form pre-filled with the best guess and the precise error; step 3 = created, live backfill progress bar (WS `sync.progress`) with "start reading now" (list fills as backfill proceeds).
+
+### 9.4 ThreadView contract (invariant I-5)
+```
+ThreadView(props: { messages: Message[] /* ASC always */, order: 'asc'|'desc',
+                    composerState, ... })
+```
+- Rendering order = `order==='asc' ? messages : [...messages].reverse()`. No other branching on `order` is allowed in layout code.
+- **Composer anchor** = a slot component rendered adjacent to the newest message: after it when `asc` (bottom of thread), directly after it when `desc` (top of thread, above older messages). Same component, one placement rule: "composer is always attached to the newest message."
+- Initial scroll: to the newest message (bottom for asc, top for desc). Collapse: all read messages except the newest are collapsed; "Expand all" toggles. Unread messages are never collapsed.
+- Keyboard: `j`/`k` move in *visual* order; `o`/`Enter` expand; `r` reply, `a` reply-all, `f` forward (open composer at anchor); `e` archive; `/` focus search; `gi` go to unified inbox; `esc` close composer (draft persists). Toolbar has the per-thread order toggle; it updates local state only, and long-presses/menu offers "make default" (PATCH /settings).
+- Message HTML renders in `<iframe sandbox="" srcdoc={...}>` sized via ResizeObserver postMessage shim; remote images show a "load images / always from this sender" bar when the sanitizer flagged any.
+
+## 10. Security requirements (hard)
+
+1. Argon2id params per §6.1; no other password hashing anywhere.
+2. **Credential vault:** mailbox secrets encrypted with AES-256-GCM under per-account data keys, wrapped by a master key from env `MASTER_KEY` (32 B, base64) — `cryptography` library, key id embedded for future rotation. Decryption code path exists only in `workers/` and `accounts/test`. Secrets never appear in logs, API responses, or exceptions (custom repr).
+3. CSRF double-submit on all mutating routes; `__Host-` cookie prefix; CSP: `default-src 'self'; frame-src 'self'; img-src 'self' data:` (remote mail images are proxied? **No** — v1 loads allowed remote images inside the sandboxed iframe only, which has its own relaxed CSP via srcdoc; document this in code).
+4. nh3 sanitization at index time; re-sanitize on render if sanitizer version changed (store `sanitizer_v` on body_parts).
+5. Rate limits: login (§6.1), `POST /accounts/discover` 10/min/user, `POST /send` 30/h/user (env-tunable).
+6. All outbound autodiscovery HTTP: HTTPS only, 5 s timeout, response size cap 256 KB, no off-domain redirects (SSRF guard: resolve + reject private IP ranges).
+7. Every SQL query and search query includes `user_id` (enforced by repository-layer helpers; raw session use in API code is a review error).
+8. Uploads (attachments to drafts): 25 MB/file, 40 MB/message total, content-type sniffed server-side, stored to blob store, never executed/served with original content type unless in safe list (images/pdf/text) — others force `application/octet-stream` + attachment disposition.
+
+## 11. Plugin API (freeze in M3 — including the heavy-plugin surfaces below)
+
+A plugin = Python package `webmail_plugin_<name>` exposing:
+
+```python
+plugin = Plugin(
+    name, version,
+    hooks       = {event: handler},          # subscribe to §8.3 events
+    router      = APIRouter | None,          # mounted at /api/plugins/{name}
+    ui_manifest = UiManifest | None,         # slots + pages, see below
+    migrations  = "migrations/" | None,      # 11.1
+    jobs        = [JobSpec, ...] | None,     # 11.2
+)
+```
+
+Loader discovers plugins via entry points group `webmail.plugins`, forwards declared events, and serves `GET /api/plugins/ui-manifest` for the frontend loader. Plugins may not import from `app.*` internals other than `app.plugins.sdk` (which exposes: DB session factory scoped to plugin tables, EventBus publish under `plugin.{name}.*`, blob store, outbox draft/send helpers, settings accessor, HTTP client with the §10.6 SSRF guard built in).
+
+Every plugin gets the namespaced KV store (`plugin_kv(plugin, user_id, key)`) for light state. Heavy plugins additionally get:
+
+### 11.1 Plugin-owned tables
+A plugin may ship Alembic migrations in its own version branch (`alembic -n plugin_{name}`). All tables MUST be prefixed `plg_{name}_` and every user-data table MUST carry `user_id` (invariant I-7 applies to plugins). Core migrations never reference plugin tables and vice versa. Uninstalling a plugin offers `--drop-data` to run its downgrade chain.
+
+### 11.2 Plugin background jobs
+`JobSpec(func, schedule: cron_str | None, singleton: bool)` registers tasks on the shared arq worker: cron-scheduled (e.g. a sync loop trigger) or enqueued on demand from hooks/routes. `singleton=True` guarantees at most one running instance per (plugin, user) via a Redis lock. Plugin jobs run in the same worker process pool with the same retry semantics as core jobs and are visible in worker logs under `plg.{name}.*`.
+
+### 11.3 Frontend surfaces
+- **Slots** (lazy-loaded ESM injected into mail UI): `thread.toolbar`, `message.actions`, `sidebar.section`, `settings.page`.
+- **Pages** (`app.page`): a plugin may declare full views: `{route: "/calendar", nav: {icon, label, order}, bundle}`. The shell adds the nav entry, and the route (including sub-routes under it) lazy-loads the bundle into the **entire content area** — no mail list/thread components are rendered. Page bundles receive the SDK context (API client, WS event stream filtered to `plugin.{name}.*` + requested core events, design-system components) and are sandboxed to their route: importing mail-component internals is a review error. Removing the plugin removes the nav entry and route cleanly.
+- **Capabilities**: the manifest may declare named capabilities mapping to plugin endpoints, e.g. `capabilities: {"recipients.suggest": "/api/plugins/contacts/suggest"}`. Core UI features query all providers of a capability they support and merge results client-side (defined capabilities in v1: `recipients.suggest` — used by the composer To/Cc/Bcc fields and by any other plugin needing person-picking, such as calendar attendees). This keeps core backend code plugin-unaware.
+
+Acceptance test for the API: implement **filters** (rules on `message.new`: match → move/flag) and **labels** entirely as plugins using only slots + KV; then implement the **calendar** (Appendix A) using tables + jobs + a page — all with zero core diffs.
+
+## 12. Milestones with acceptance criteria
+
+**M0 — Architecture spike**
+Login/register (no 2FA), one hardcoded-config IMAP account, backfill+IDLE for INBOX, JWZ, read-only unified list + thread view with working asc/desc toggle and correct composer-anchor placeholder, Postgres FTS free-text search.
+✅ Accept: sync a real 10k-message Gmail account via app password; threads match Gmail's grouping on a 50-thread sample ≥ 90%; toggle order and the newest message + anchor swap ends correctly; search returns in < 500 ms.
+
+**M1 — MVP**
+Full identity (2FA, sessions UI, reset), account wizard with full discovery chain, multi-account + unified inbox with account chips, flags/move/archive both directions, inline reply + reply-all/forward + SMTP send + Sent append, recipient autocomplete from `known_recipients`, drafts autosave, sanitized HTML + blocked remote images, operator search (full §8.2 grammar) on PG FTS, WS live updates.
+✅ Accept: add gmail.com, outlook.com, and one autoconfig-only domain through the wizard typing only address+password; reply from a thread lands correctly threaded in the recipient's client; all §8.2 operators pass integration tests; `docker-compose up` from clean checkout works.
+
+**M2 — Hardening / v1.0**
+OAuth2 (Google + Microsoft, XOAUTH2 for IMAP/SMTP, server-side refresh), QRESYNC fast resync, drafts mirrored to server, undo-send, attachment upload/preview, Meilisearch backend + instant search-as-you-type UI + saved searches as virtual mailboxes, session revocation, backfill progress UX, folder-name i18n heuristics.
+✅ Accept: typo query (`invocie`) finds "invoice" mail < 200 ms on a 100k-message corpus; kill -9 the worker mid-backfill and restart resumes without duplicates (idempotency via `(mailbox_id, uid)` unique key); undo-send within window results in no SMTP traffic.
+
+**M3 — Plugin platform**
+Loader, full SDK per §11 **including** plugin migrations (11.1), plugin jobs (11.2), and `app.page` (11.3); filters + labels shipped as plugins; plugin docs.
+✅ Accept: both plugins install/uninstall cleanly with zero core diffs; a third trivial plugin ("Unsubscribe header button" on `message.actions`) can be written from `docs/plugins.md` alone; a skeleton page-plugin (route + nav entry + one plugin table + one cron job) round-trips install → run → uninstall `--drop-data` leaving no residue.
+
+**M4 — Calendar plugin (Appendix A)**
+Phases A1–A2: local calendars + full page views + iMIP invite handling; then A3: CalDAV client sync + ICS subscriptions.
+✅ Accept: criteria in Appendix A §A.7.
+
+**M5 — Contacts plugin (Appendix B)**
+Phases B1–B2: contact books, full page, groups, add-from-mail, `recipients.suggest` capability, vCard import/export; then B3: CardDAV client sync. May be built in parallel with M4 — both depend only on the M3 SDK.
+✅ Accept: criteria in Appendix B §B.7.
+
+**M6 (optional) — Hosted mailboxes**
+Inbound MTA (Stalwart) delivering to a `NativeProvider`; handle becomes receiving address. Specified separately when reached.
+
+## 13. Testing & CI
+
+- **Unit:** pytest + pytest-asyncio; threading vectors (§7.4); query parser (§8.2); autodiscovery resolver with `respx`-mocked HTTP + `dnspython` stub; MIME corpus in `tests/fixtures/mime/` (agent: include ≥ 25 nasty real-world-style fixtures — encoded headers incl. RFC2047 Polish diacritics, nested multiparts, missing Message-ID, malformed dates, HTML-only, inline cid images, 8-bit subjects, winmail.dat placeholder).
+- **Integration:** docker-compose.ci.yml with Greenmail (IMAP+SMTP) and Dovecot; full loop test: append message to Greenmail → appears in API < 10 s → reply via API → arrives in Greenmail SMTP sink with correct References.
+- **Frontend:** vitest + React Testing Library; ThreadView ordering/anchor tests for both directions; Playwright smoke: login → add Greenmail account via wizard → read → inline reply → search.
+- **CI gates:** ruff, mypy --strict, pytest, tsc, eslint, vitest, Playwright smoke; coverage floor 80% on `threading_`, `search/queryparser`, `accounts/autodiscovery`, `indexer`.
+
+## 14. Configuration (env, complete list)
+
+`APP_DOMAIN`, `DATABASE_URL`, `REDIS_URL`, `MASTER_KEY`, `BLOB_ROOT` or `S3_*`, `SEARCH_BACKEND=pg|meili`, `MEILI_URL`, `MEILI_KEY`, `UNDO_SECONDS=10`, `SESSION_TTL_DAYS=30`, `OAUTH_GOOGLE_CLIENT_ID/SECRET`, `OAUTH_MS_CLIENT_ID/SECRET`, `MAX_ATTACHMENT_MB=25`, `LOG_LEVEL`; sync engine v2 (§6.3): `WORKER_ROLE=jobs|sync`, `MAX_ACCOUNTS_PER_WORKER=400`, `LEASE_TTL=60`, `LEASE_RENEW=20`, `POLL_WARM=180`, `POLL_COLD=900`, `TIER_HOT_ACTIVITY_H=24`, `TIER_SESSION_GRACE_MIN=30`, `TIER_COLD_AFTER_D=3`, `HOT_BUDGET_PER_WORKER=250`, `PER_HOST_CONN_CAP=50`, `BULK_CONCURRENCY=2`, `PARSE_CONCURRENCY=4`, `STREAM_PARSE_MB=8`. All read once in `config.py` (pydantic-settings); no `os.environ` elsewhere.
+
+---
+
+## Appendix A — Calendar plugin (`webmail_plugin_calendar`)
+
+The calendar is the reference "heavy plugin": it lives in the same app and repo, uses the §11 SDK only, and must build with **zero core diffs**. It ships as a page (`/calendar`) beside Mail — one product, one login.
+
+### A.1 Scope by phase
+- **A1 — Local calendar:** create/edit/delete events, multiple calendars per user, month/week/day/agenda views, recurrence, reminders (WS notification), timezone-correct display.
+- **A2 — Invites over mail (iMIP):** detect `text/calendar` parts on `message.new`; render an invite card in the thread (`message.actions` + inline summary); Accept/Tentative/Decline sends an iTIP REPLY through the outbox; organizer flow sends REQUEST/CANCEL to attendees; updates (SEQUENCE bumps) modify the stored event.
+- **A3 — External sources:** `CalendarProvider` interface with **CalDAV client** (first implementation; covers iCloud, Fastmail, Nextcloud) and read-only **ICS URL subscriptions** (holidays, team feeds); sync loop as a plugin job.
+- **A4 (deferred, spec later):** Google/Microsoft API providers reusing M2 OAuth machinery; secret **ICS export feed** `GET /api/plugins/calendar/feed/{token}.ics`; CalDAV **server** endpoint for native phone sync — architecturally anticipated, not built now.
+
+### A.2 Dependencies
+`icalendar ~5.0` (parse/serialize RFC 5545), `python-dateutil` (RRULE expansion), `caldav ~1.3` (A3), `tzdata`. No other additions.
+
+### A.3 Data model (plugin migrations, prefix `plg_calendar_`)
+```sql
+CREATE TABLE plg_calendar_calendars (
+  id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
+  user_id uuid NOT NULL,
+  name text NOT NULL,
+  color text NOT NULL DEFAULT '#3b82f6',
+  kind text NOT NULL DEFAULT 'local',     -- local|caldav|ics_subscription
+  source_config jsonb NOT NULL DEFAULT '{}',  -- caldav url/collection, ics url
+  secret_ct bytea,                        -- caldav credentials, §10.2 vault scheme
+  sync_token text,                        -- CalDAV sync-token / ICS etag
+  read_only boolean NOT NULL DEFAULT false,
+  UNIQUE (user_id, name)
+);
+
+CREATE TABLE plg_calendar_events (
+  id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
+  user_id uuid NOT NULL,
+  calendar_id uuid NOT NULL REFERENCES plg_calendar_calendars(id) ON DELETE CASCADE,
+  ical_uid text NOT NULL,                 -- UID from iCalendar; join key for iMIP updates
+  sequence int NOT NULL DEFAULT 0,
+  summary text NOT NULL DEFAULT '',
+  description text NOT NULL DEFAULT '',
+  location text NOT NULL DEFAULT '',
+  dtstart timestamptz NOT NULL,
+  dtend timestamptz,
+  all_day boolean NOT NULL DEFAULT false,
+  tz text NOT NULL DEFAULT 'UTC',         -- original TZID; render in user tz, store UTC
+  rrule text,                             -- raw RRULE string, NULL = single
+  exdates timestamptz[] NOT NULL DEFAULT '{}',
+  recurrence_id timestamptz,              -- non-NULL = detached exception instance
+  organizer jsonb,                        -- {name,email}
+  attendees jsonb NOT NULL DEFAULT '[]',  -- [{name,email,partstat,role}]
+  status text NOT NULL DEFAULT 'confirmed',   -- confirmed|tentative|cancelled
+  my_partstat text,                       -- needs-action|accepted|declined|tentative
+  reminders int[] NOT NULL DEFAULT '{}',  -- minutes before start
+  origin_message_id uuid,                 -- link back to the invite mail thread
+  raw_ical text,                          -- round-trip fidelity for iMIP/CalDAV
+  UNIQUE (calendar_id, ical_uid, recurrence_id)
+);
+CREATE INDEX ON plg_calendar_events (user_id, dtstart);
+```
+
+**Recurrence rule:** store the RRULE string; expand instances at query time with `dateutil.rrule` over the requested view range only (no materialization table). Exceptions: `exdates` remove instances; a row with `recurrence_id` overrides one instance. Range queries must account for events starting before the window but recurring into it (expand from `dtstart`, cap iterations at 1000 per event).
+
+### A.4 CalendarProvider interface (inside the plugin, mirrors §5.1)
+```python
+# PLUGIN INTERFACE VERSION: 1
+class CalendarProvider(Protocol):
+    async def verify(self) -> None: ...
+    async def list_collections(self) -> list[RemoteCalendar]: ...
+    async def pull(self, collection: str, *, sync_token: str | None
+                   ) -> tuple[list[RemoteEvent], list[str], str]: ...
+        # returns (changed_events_as_ical, deleted_uids, new_sync_token)
+    async def push(self, collection: str, ical: str, *, uid: str,
+                   etag: str | None) -> str: ...        # returns new etag
+    async def delete(self, collection: str, uid: str, etag: str | None) -> None: ...
+```
+Implementations in A3: `CaldavProvider`, `IcsSubscriptionProvider` (pull-only, `push`/`delete` raise `ReadOnlyCalendar`). Sync loop = plugin cron job (`schedule="*/5 * * * *"`, `singleton=True` per user) doing pull → upsert by `(calendar_id, ical_uid, recurrence_id)` → local-dirty push with etag conflict handling (conflict ⇒ server wins, local copy saved as a duplicate marked `status='tentative'`, user notified).
+
+### A.5 iMIP flow (A2)
+- **Inbound:** hook on `message.new`; if a `text/calendar` part exists, parse METHOD:
+  `REQUEST` → upsert event (match `ical_uid`; apply only if `SEQUENCE` ≥ stored) into the user's default calendar with `my_partstat='needs-action'`, set `origin_message_id`; `CANCEL` → `status='cancelled'`; `REPLY` → update that attendee's `partstat` on the organizer's copy.
+- **RSVP:** invite card buttons call `POST /api/plugins/calendar/events/{id}/rsvp {partstat}` → update row → build iTIP REPLY ics → send via SDK outbox helper to the organizer (subject `Accepted: {summary}` etc.) → publish `plugin.calendar.event_updated`.
+- **Organizer:** creating an event with attendees sends METHOD:REQUEST to each; edits bump `SEQUENCE` and re-send; deletion sends CANCEL. All outgoing invite mail uses the account whose address matches the organizer identity (picker if ambiguous).
+
+### A.6 UI (page `/calendar`, nav icon between Mail and Settings)
+Routes: `/calendar` (month default), `/calendar/{month|week|day|agenda}/:date`, `/calendar/event/:id`, `/calendar/settings` (calendars CRUD, CalDAV/ICS wizard reusing the account-wizard UX pattern: URL+credentials → test → collection picker). Interactions: click-drag create, drag-move/resize (writes optimistically, reverts on API error), event popover with edit/RSVP, mini-month navigator, calendar visibility toggles (client state), `t` = today, arrows = prev/next period. Invite linkage both ways: invite card in thread links to `/calendar/event/:id`; event view links back to the mail thread when `origin_message_id` is set. Live updates via WS `plugin.calendar.*` events. Timezone: render in user setting `settings.tz` (add to §9.1 settings; default from browser at registration).
+
+### A.7 Acceptance criteria
+1. Plugin installs/uninstalls (`--drop-data`) with zero diffs in `app/` and `frontend/src/` outside `plugins/`/`slots/` loader code.
+2. A weekly-recurring event with two EXDATEs and one moved instance renders correctly in month and week views across a DST boundary (Europe/Warsaw fixtures).
+3. Round-trip with Google Calendar **via iMIP mail only**: receive a Google invite → Accept in our UI → organizer's Google Calendar shows attendee accepted; our organizer invite → Google attendee accepts → our event shows `partstat=accepted`.
+4. CalDAV sync against a Nextcloud/Radicale test container: create/edit/delete propagate both directions in ≤ 5 min (one cron cycle); conflict case produces the A.4 tentative-duplicate, never silent loss.
+5. ICS subscription to a public holiday feed renders read-only events; editing is blocked in UI and API (403 `read_only_calendar`).
+6. Killing the worker mid-sync and restarting causes no duplicate events (upsert key discipline).
+
+---
+
+## Appendix B — Contacts plugin (`webmail_plugin_contacts`)
+
+The second heavy plugin: same rules as the calendar — §11 SDK only, zero core diffs, own page. Division of labor with core: core owns *implicit* recipient history (`known_recipients`, §4/§9.1) so autocomplete works even without this plugin; the Contacts plugin owns *explicit*, user-curated contact records and enriches suggestions via the `recipients.suggest` capability (§11.3).
+
+### B.1 Scope by phase
+- **B1 — Local contacts:** address books, contact CRUD with structured fields, groups, photo, search, dedicated page `/contacts`; "Add sender to contacts" from the thread view; suggestion capability wired into the composer.
+- **B2 — Portability:** vCard 3.0/4.0 import (file, multi-card) and export (single contact / book), duplicate detection + merge tool.
+- **B3 — External sources:** `ContactProvider` interface with **CardDAV client** sync (iCloud, Fastmail, Nextcloud, Radicale); per-book sync job.
+- **B4 (deferred, spec later):** Google People / Microsoft Graph providers (reusing M2 OAuth), CardDAV **server** endpoint for native phone sync, book sharing.
+
+### B.2 Dependencies
+`vobject ~0.9` (vCard parse/serialize). Photos stored via the SDK blob store (JPEG/PNG, re-encoded server-side, ≤ 512×512). Nothing else.
+
+### B.3 Data model (plugin migrations, prefix `plg_contacts_`)
+```sql
+CREATE TABLE plg_contacts_books (
+  id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
+  user_id uuid NOT NULL,
+  name text NOT NULL,
+  kind text NOT NULL DEFAULT 'local',       -- local|carddav
+  source_config jsonb NOT NULL DEFAULT '{}',-- carddav url/collection
+  secret_ct bytea,                          -- carddav credentials, §10.2 vault scheme
+  sync_token text,
+  read_only boolean NOT NULL DEFAULT false,
+  UNIQUE (user_id, name)
+);
+
+CREATE TABLE plg_contacts_contacts (
+  id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
+  user_id uuid NOT NULL,
+  book_id uuid NOT NULL REFERENCES plg_contacts_books(id) ON DELETE CASCADE,
+  vcard_uid text NOT NULL,                  -- UID property; join key for CardDAV
+  etag text,                                -- server etag (carddav books)
+  name_display text NOT NULL DEFAULT '',    -- FN
+  name_parts jsonb NOT NULL DEFAULT '{}',   -- {given,family,middle,prefix,suffix} (N)
+  emails jsonb NOT NULL DEFAULT '[]',       -- [{value,type,pref}]
+  phones jsonb NOT NULL DEFAULT '[]',
+  addresses jsonb NOT NULL DEFAULT '[]',    -- structured ADR
+  org text NOT NULL DEFAULT '',
+  title text NOT NULL DEFAULT '',
+  birthday date,
+  urls jsonb NOT NULL DEFAULT '[]',
+  notes text NOT NULL DEFAULT '',
+  photo_sha256 text,
+  raw_vcard text,                           -- round-trip fidelity: preserve unknown props
+  updated_at timestamptz NOT NULL DEFAULT now(),
+  UNIQUE (book_id, vcard_uid)
+);
+CREATE INDEX ON plg_contacts_contacts (user_id, name_display);
+CREATE INDEX ON plg_contacts_contacts USING gin (emails jsonb_path_ops);
+
+CREATE TABLE plg_contacts_groups (
+  id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
+  user_id uuid NOT NULL,
+  name text NOT NULL,
+  UNIQUE (user_id, name)
+);
+CREATE TABLE plg_contacts_group_members (
+  group_id uuid NOT NULL REFERENCES plg_contacts_groups(id) ON DELETE CASCADE,
+  contact_id uuid NOT NULL REFERENCES plg_contacts_contacts(id) ON DELETE CASCADE,
+  PRIMARY KEY (group_id, contact_id)
+);
+```
+
+vCard mapping discipline: parse known properties into columns, keep the full source in `raw_vcard`, and on export/push regenerate from columns **merged over** `raw_vcard` so properties we don't model (e.g. `X-` extensions) survive round-trips.
+
+### B.4 ContactProvider interface (inside the plugin, mirrors A.4)
+```python
+# PLUGIN INTERFACE VERSION: 1
+class ContactProvider(Protocol):
+    async def verify(self) -> None: ...
+    async def list_books(self) -> list[RemoteBook]: ...
+    async def pull(self, book: str, *, sync_token: str | None
+                   ) -> tuple[list[RemoteVcard], list[str], str]: ...
+        # (changed vcards, deleted uids, new sync_token)
+    async def push(self, book: str, vcard: str, *, uid: str,
+                   etag: str | None) -> str: ...
+    async def delete(self, book: str, uid: str, etag: str | None) -> None: ...
+```
+Sync loop = plugin cron job (`*/5 * * * *`, singleton per user), upsert by `(book_id, vcard_uid)`, etag conflict ⇒ server wins + local copy kept as a duplicate flagged for the merge tool (never silent loss) — same policy as A.4.
+
+### B.5 API & capability (`/api/plugins/contacts/`)
+`GET /contacts?book=&group=&q=&cursor=` · `POST /contacts` · `GET/PUT/DELETE /contacts/{id}` · `POST /contacts/from-message {message_id}` (prefill from From header) · `GET/POST/DELETE /books`, `/groups`, group membership ops · `POST /import` (vCard file → report `{created, updated, skipped}`) · `GET /export?book=|contact=` (`.vcf`) · `GET /duplicates` (candidate pairs: same email, or same normalized name + fuzzy) · `POST /merge {keep_id, remove_id}` (field union, keep wins conflicts, group memberships merged) · **`GET /suggest?q=`** — the `recipients.suggest` capability endpoint: matches name/email prefix and substring, returns `[{name, email, source:"contact", contact_id, photo_url?}]`. Composer merge rule (client-side, defined here so behavior is deterministic): contact hits rank above `known_recipients` hits at equal prefix quality; dedupe by email, contact entry wins and carries the photo/display name.
+
+### B.6 UI
+Page `/contacts` (nav entry after Calendar): three-pane layout — books/groups sidebar, alphabetically grouped virtualized list with search-as-you-type, detail pane with edit-in-place form; import/export and merge review under `/contacts/tools`; CardDAV book wizard reusing the account-wizard UX pattern (URL + credentials → test → book picker). Mail integration via `message.actions` slot: "Add to Contacts" (hidden when the sender already matches a contact; then it becomes "View contact" linking to `/contacts/{id}`). Group compose: selecting a group in the composer expands to its members' preferred emails.
+
+### B.7 Acceptance criteria
+1. Zero core diffs; install/uninstall `--drop-data` round-trip clean.
+2. Import a 500-card vCard file mixing v3.0/v4.0, quoted-printable encoding, and Polish diacritics in names (`Grzegorz Brzęczyszczykiewicz`) — all names, emails, and photos correct; re-export → re-import is lossless including `X-` properties (raw_vcard merge discipline).
+3. Composer typing a 3-char prefix returns merged suggestions in < 100 ms p95 (contacts ranked above history, deduped by email, photos shown for contacts).
+4. CardDAV sync against Radicale: create/edit/delete propagate both directions within one cron cycle; concurrent edit conflict produces a flagged duplicate, never silent loss; worker kill mid-sync ⇒ no duplicate contacts on restart.
+5. "Add sender to contacts" from a thread creates a prefilled contact; the same sender's messages immediately render the contact's display name, and the action flips to "View contact".
+6. Merge tool: merging two contacts unions emails/phones/groups, keeps the chosen primary's conflicting fields, and repoints nothing else in the system (calendar attendees and mail remain keyed by email, by design).
+
+---
+*End of specification. Ambiguities: prefer the simplest behavior consistent with the invariants in §1, leave a `TODO(spec)` comment, and list open questions in `docs/open-questions.md` rather than guessing silently.*