sync-v2 step 2: extract AccountRunner

Move the per-account INBOX-IDLE + secondary-poll loops out of the supervisor
into a self-contained AccountRunner (start/stop/teardown). The startup supervisor
now launches one runner per live account. Pure refactor, no behavior change.

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

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,17 +1,14 @@
-"""Live-sync supervisor (SPEC §6.3).
+"""Live-sync supervisor (refactor §4, transitional).
 
-Runs inside the arq worker: on startup it launches one IDLE/poll loop per live
+Runs inside the sync worker: on startup it launches one :class:`AccountRunner`
-account so new mail and flag changes flow to the UI over WebSocket. Loops
+per live account. This step (v2 §9.2) only extracts the per-account loops into a
-self-heal (reconnect with backoff) and are cancelled on worker shutdown.
+runner; account discovery is still startup-only and is replaced by the lease
-
+claimer in a later step.
-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
@@ -20,58 +17,30 @@ from sqlalchemy import select
 
 from app.db import session_scope
 from app.models import Account
-from app.sync.engine import live_loop, secondary_poll_loop
+from app.sync.runner import AccountRunner
 
 logger = logging.getLogger(__name__)
 
-_RECONNECT_BACKOFF = 30.0
 
-
+async def _live_account_ids() -> list[uuid.UUID]:
-async def _run_account(account_id: uuid.UUID, stop: asyncio.Event) -> None:
+    async with session_scope() as session:
-    while not stop.is_set():
+        return list(await session.scalars(select(Account.id).where(Account.state == "live")))
-        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)."""
+    """Launch an AccountRunner per live account (called from the worker on_startup)."""
-    async with session_scope() as session:
+    runners: dict[uuid.UUID, AccountRunner] = {}
-        account_ids = list(await session.scalars(select(Account.id).where(Account.state == "live")))
+    for account_id in await _live_account_ids():
-    stop = asyncio.Event()
+        runner = AccountRunner(account_id)
-    ctx["supervisor_stop"] = stop
+        runner.start()
-    tasks: list[asyncio.Task[None]] = []
+        runners[account_id] = runner
-    for aid in account_ids:
+    ctx["runners"] = runners
-        tasks.append(asyncio.create_task(_run_account(aid, stop)))  # INBOX IDLE
+    logger.info("live supervisor started for %d account(s)", len(runners))
-        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)."""
+    """Stop and tear down all account runners (called from the worker on_shutdown)."""
-    stop = ctx.get("supervisor_stop")
+    runners: dict[uuid.UUID, AccountRunner] = ctx.get("runners", {})
-    if stop is not None:
+    if runners:
-        stop.set()
+        await asyncio.gather(*(r.stop() for r in runners.values()), return_exceptions=True)
-    tasks = ctx.get("supervisor_tasks", [])
+    ctx["runners"] = {}
-    for task in tasks:
-        task.cancel()
-    if tasks:
-        await asyncio.gather(*tasks, return_exceptions=True)