sync-v2 step 4: event-driven account lifecycle (§3)

- Coordinator subscribes to account.state_changed over Redis pub/sub: going
  live marks the account claimable at once; disabled/deleted/error tears the
  owning runner down immediately (≤30s) instead of waiting for the lease TTL.
- delete_account now publishes account.state_changed:deleted after dropping the
  account from sync:live, so teardown is event-driven, not just fencing.
- AccountRunner gains a verify loop (defense in depth): it periodically confirms
  the account is still live and flags itself dead if not; the coordinator reaps
  dead runners and releases their lease. Zombie loops are structurally impossible.
- New core event user.active (SPEC §8.3), published on login and on every WS
  connect; consumed for tier promotion in step 5.
- Acceptance: added event-driven teardown test (isolated from fencing/verify).

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

backend/app/sync/coordinator.py

@@ -16,6 +16,7 @@ account because claims are ``SET NX`` and renew/release are owner-checked.
 from __future__ import annotations
 
 import asyncio
+import contextlib
 import json
 import logging
 import os
@@ -29,6 +30,7 @@ from sqlalchemy import select
 
 from app.config import get_settings
 from app.db import session_scope
+from app.events import channel_for
 from app.models import Account
 from app.redis_client import aw, get_redis
 from app.sync import leases
@@ -78,6 +80,7 @@ class SyncCoordinator:
             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)),
+            asyncio.create_task(self._event_subscriber()),
         ]
         logger.info("sync coordinator %s started", self.worker_id)
 
@@ -132,6 +135,14 @@ class SyncCoordinator:
 
     async def _renew_once(self) -> None:
         for account_id in list(self._runners):
+            runner = self._runners.get(account_id)
+            if runner is None:
+                continue
+            # Reap a runner that self-verified it is no longer live (§3).
+            if runner.dead.is_set():
+                logger.info("reaping dead runner for account %s", account_id)
+                await self._release(account_id)
+                continue
             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)
@@ -203,3 +214,39 @@ class SyncCoordinator:
         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)
+
+    # --- event-driven lifecycle (§3) -----------------------------------------
+
+    async def _event_subscriber(self) -> None:
+        """React to ``account.state_changed`` across processes (Redis pub/sub).
+
+        Going live makes an account claimable at once; going away tears the
+        runner down immediately (≤30s from the API call) instead of waiting for
+        the lease to lapse.
+        """
+        pubsub = self._redis.pubsub()
+        await pubsub.subscribe(channel_for("account.state_changed"))
+        try:
+            async for message in pubsub.listen():
+                if self._stop.is_set():
+                    break
+                if message.get("type") != "message":
+                    continue
+                try:
+                    payload = json.loads(message["data"]).get("payload", {})
+                except (ValueError, TypeError):
+                    continue
+                await self._on_state_changed(payload)
+        finally:
+            with contextlib.suppress(Exception):
+                await pubsub.aclose()  # type: ignore[no-untyped-call]
+
+    async def _on_state_changed(self, payload: dict[str, object]) -> None:
+        raw = payload.get("account_id")
+        if not isinstance(raw, str):
+            return
+        account_id = uuid.UUID(raw)
+        if payload.get("state") == "live":
+            await leases.mark_live(self._redis, account_id)  # claimable next cycle
+        elif account_id in self._runners:
+            await self._release(account_id)  # disabled/deleted/error -> stop now

backend/app/sync/runner.py

@@ -14,11 +14,16 @@ import logging
 import uuid
 from collections.abc import Awaitable, Callable
 
+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
 
 logger = logging.getLogger(__name__)
 
 _RECONNECT_BACKOFF = 30.0
+_VERIFY_INTERVAL = 60.0
 
 
 class AccountRunner:
@@ -28,6 +33,10 @@ class AccountRunner:
         self.account_id = account_id
         self._stop = asyncio.Event()
         self._tasks: list[asyncio.Task[None]] = []
+        # Set when the account is found no longer live (§3 defense in depth); the
+        # coordinator reaps such runners and releases their lease.
+        self.dead = asyncio.Event()
+        self.verify_interval = _VERIFY_INTERVAL
 
     def start(self) -> None:
         """Launch the account's sync loops (idempotent-ish: call once)."""
@@ -35,6 +44,7 @@ class AccountRunner:
         self._tasks = [
             asyncio.create_task(self._supervise(self._run_live, "live")),
             asyncio.create_task(self._supervise(self._run_secondary, "secondary")),
+            asyncio.create_task(self._verify_loop()),
         ]
 
     def running(self) -> bool:
@@ -55,6 +65,28 @@ class AccountRunner:
     async def _run_secondary(self) -> None:
         await secondary_poll_loop(self.account_id, self._stop)
 
+    async def _verify_loop(self) -> None:
+        """Periodically confirm the account is still live; flag ``dead`` if not.
+
+        This makes zombie loops structurally impossible even if a lifecycle event
+        is lost: an account that was disabled/deleted stops being synced within a
+        verify cycle regardless of the lease/event paths.
+        """
+        while not self._stop.is_set():
+            with contextlib.suppress(TimeoutError):
+                await asyncio.wait_for(self._stop.wait(), timeout=self.verify_interval)
+            if self._stop.is_set():
+                return
+            if not await self._account_is_live():
+                logger.info("account %s no longer live; flagging runner dead", self.account_id)
+                self.dead.set()
+                return
+
+    async def _account_is_live(self) -> bool:
+        async with session_scope() as session:
+            state = await session.scalar(select(Account.state).where(Account.id == self.account_id))
+        return state == "live"
+
     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():