feat: Unify account refresh and random retry runtime behavior

app/control/account/quota_defaults.py

@@ -2,10 +2,10 @@
 
 Canonical quota totals per pool type (from upstream rate-limits API):
 
-              auto    fast    expert    heavy
-  basic         20      60       8        —      window: 72000 / 36000 s
-  super         50     140      50        —      window: 7200 s
-  heavy        150     400     150       20      window: 7200 s
+              auto    fast    expert    heavy    grok_4_3
+  basic         20      60       8        —         —        window: 72000 / 36000 s
+  super         50     140      50        —        50        window: 7200 s
+  heavy        150     400     150       20       150        window: 7200 s
 
 Pool inference uses ``auto.total`` as the primary signal — the three values
 (20 / 50 / 150) are mutually exclusive across pool types.

app/control/account/refresh.py

@@ -57,9 +57,9 @@ class AccountRefreshService:
     """Fetches real quota data from the upstream usage API and persists it.
 
     Triggers:
-      1. Import   — super accounts: fetch all 3 modes.
-      2. Call     — super accounts: fetch the called mode (async, non-blocking).
-      3. Schedule — super: fetch all 3 modes; basic: static window reset check.
+      1. Import   — fetch all modes supported by the account's pool.
+      2. Call     — fetch the called mode only (async, non-blocking).
+      3. Schedule — refresh one pool per loop using that pool's supported modes.
     """
 
     def __init__(self, repository: "AccountRepository") -> None:
@@ -75,7 +75,13 @@ def __init__(self, repository: "AccountRepository") -> None:
     async def _fetch_all_quotas(
         self, token: str, pool: str
     ) -> dict[int, QuotaWindow] | None:
-        """Fetch quota windows for modes supported by *pool*. Returns {mode_id: window}."""
+        """Fetch quota windows for every mode supported by *pool*.
+
+        Examples:
+          - basic -> auto / fast / expert
+          - super -> auto / fast / expert / grok_4_3
+          - heavy -> auto / fast / expert / heavy / grok_4_3
+        """
         try:
             from app.dataplane.reverse.protocol.xai_usage import fetch_all_quotas
 
@@ -142,7 +148,7 @@ async def refresh_on_import(self, tokens: list[str]) -> RefreshResult:
         return agg
 
     async def refresh_call_async(self, token: str, mode_id: int) -> None:
-        """Fire-and-forget quota sync after a successful call (all accounts)."""
+        """Fire-and-forget single-mode quota sync after a successful call."""
         record = (await self._repo.get_accounts([token]) or [None])[0]
         if record is None or record.is_deleted():
             return
@@ -219,7 +225,7 @@ async def _refresh_one(
         *,
         apply_fallback: bool = False,
     ) -> RefreshResult:
-        """Fetch all 3 modes from the usage API and persist real quota data.
+        """Fetch all pool-supported modes from the usage API and persist them.
 
         apply_fallback=True  — used by scheduled/import paths: when API fails,
                                decrement REAL quotas or reset expired DEFAULT windows.

app/control/account/runtime.py

@@ -1,11 +1,19 @@
-"""Account runtime singletons exposed without importing app.main."""
+"""Account runtime singletons and hot-apply helpers.
 
-from typing import TYPE_CHECKING
+These helpers expose the process-local account refresh runtime without making
+callers import ``app.main``. Admin handlers use them to reconcile strategy and
+scheduler state after hot config updates.
+"""
+
+from typing import TYPE_CHECKING, Literal
 
 if TYPE_CHECKING:
     from .refresh import AccountRefreshService
+    from .scheduler import AccountRefreshScheduler
 
 _refresh_service: "AccountRefreshService | None" = None
+_refresh_scheduler: "AccountRefreshScheduler | None" = None
+_refresh_scheduler_leader = False
 
 
 def set_refresh_service(service: "AccountRefreshService | None") -> None:
@@ -19,4 +27,76 @@ def get_refresh_service() -> "AccountRefreshService | None":
     return _refresh_service
 
 
-__all__ = ["get_refresh_service", "set_refresh_service"]
+def set_refresh_scheduler(scheduler: "AccountRefreshScheduler | None") -> None:
+    """Register the process-global account refresh scheduler."""
+    global _refresh_scheduler
+    _refresh_scheduler = scheduler
+
+
+def get_refresh_scheduler() -> "AccountRefreshScheduler | None":
+    """Return the registered account refresh scheduler, if any."""
+    return _refresh_scheduler
+
+
+def set_refresh_scheduler_leader(is_leader: bool) -> None:
+    """Record whether this worker currently owns the refresh scheduler lock."""
+    global _refresh_scheduler_leader
+    _refresh_scheduler_leader = bool(is_leader)
+
+
+def is_refresh_scheduler_leader() -> bool:
+    """Return True when this worker is the active refresh-scheduler leader."""
+    return _refresh_scheduler_leader
+
+
+def reconcile_refresh_runtime(
+    enabled: bool | None = None,
+) -> Literal["quota", "random"]:
+    """Hot-apply refresh strategy and scheduler state for the current worker."""
+    from app.dataplane.account.selector import current_strategy, set_strategy
+    from app.platform.config.snapshot import config
+    from app.platform.logging.logger import logger
+
+    refresh_enabled = (
+        config.get_bool("account.refresh.enabled", False)
+        if enabled is None
+        else bool(enabled)
+    )
+    target_strategy: Literal["quota", "random"] = (
+        "quota" if refresh_enabled else "random"
+    )
+    previous_strategy = current_strategy()
+    if previous_strategy != target_strategy:
+        set_strategy(target_strategy)
+
+    scheduler_action = "unchanged"
+    scheduler = _refresh_scheduler
+    if scheduler is not None and _refresh_scheduler_leader:
+        if refresh_enabled:
+            if not scheduler.is_running():
+                scheduler.start()
+                scheduler_action = "started"
+        elif scheduler.is_running():
+            scheduler.stop()
+            scheduler_action = "stopped"
+
+    if previous_strategy != target_strategy or scheduler_action != "unchanged":
+        logger.info(
+            "account refresh runtime reconciled: previous_strategy={} strategy={} leader={} scheduler_action={}",
+            previous_strategy,
+            target_strategy,
+            _refresh_scheduler_leader,
+            scheduler_action,
+        )
+    return target_strategy
+
+
+__all__ = [
+    "get_refresh_service",
+    "set_refresh_service",
+    "get_refresh_scheduler",
+    "set_refresh_scheduler",
+    "is_refresh_scheduler_leader",
+    "set_refresh_scheduler_leader",
+    "reconcile_refresh_runtime",
+]

app/control/account/scheduler.py

@@ -39,8 +39,16 @@ def __init__(self, refresh_service: AccountRefreshService) -> None:
         self._tasks:  list[asyncio.Task] = []
         self._stop    = asyncio.Event()
 
+    def bind_service(self, refresh_service: AccountRefreshService) -> None:
+        """Update the refresh service used by the singleton scheduler."""
+        self._service = refresh_service
+
+    def is_running(self) -> bool:
+        """Return True while any pool refresh loop is still active."""
+        return any(not task.done() for task in self._tasks)
+
     def start(self) -> None:
-        if self._tasks and not all(t.done() for t in self._tasks):
+        if self.is_running():
             return
         self._stop.clear()
         self._tasks = [
@@ -54,11 +62,14 @@ def start(self) -> None:
         )
 
     def stop(self) -> None:
+        was_running = self.is_running()
         self._stop.set()
         for t in self._tasks:
             if not t.done():
                 t.cancel()
-        logger.info("account refresh scheduler stopped")
+        self._tasks = []
+        if was_running:
+            logger.info("account refresh scheduler stopped")
 
     async def _loop(self, pool: str) -> None:
         while not self._stop.is_set():
@@ -102,6 +113,8 @@ def get_account_refresh_scheduler(
     global _scheduler
     if _scheduler is None:
         _scheduler = AccountRefreshScheduler(refresh_service)
+    else:
+        _scheduler.bind_service(refresh_service)
     return _scheduler
 
 

app/dataplane/account/__init__.py

@@ -8,16 +8,17 @@
 import asyncio
 from typing import TYPE_CHECKING
 
+from app.platform.config.snapshot import get_config
 from app.platform.logging.logger import logger
 from app.platform.runtime.clock import now_s
 from app.control.account.repository import AccountRepository
 from app.control.account.enums import FeedbackKind
 from .table import AccountRuntimeTable
 from .lease import AccountLease, new_lease
-from .selector import select, select_any
+from .selector import current_strategy, select, select_any
 from .sync import bootstrap as _bootstrap, apply_changes
 from . import feedback as fb
-from ..shared.enums import StatusId
+from ..shared.enums import POOL_ID_TO_STR, StatusId
 
 if TYPE_CHECKING:
     pass
@@ -250,12 +251,22 @@ async def feedback(
 
         ts = now_s_val if now_s_val is not None else now_s()
 
+        strategy = current_strategy()
+
         async with self._lock:
             if kind == FeedbackKind.SUCCESS:
-                fb.apply_success(table, idx, mode_id)
+                if strategy == "random":
+                    fb.apply_success_random(table, idx)
+                else:
+                    fb.apply_success_quota(table, idx, mode_id)
 
             elif kind == FeedbackKind.RATE_LIMITED:
-                fb.apply_rate_limited(table, idx, mode_id)
+                if strategy == "random":
+                    pool_id = int(table.pool_by_idx[idx])
+                    cooling_sec = _pool_cooling_sec(pool_id)
+                    fb.apply_rate_limited_random(table, idx, cooling_sec=cooling_sec)
+                else:
+                    fb.apply_rate_limited_quota(table, idx, mode_id)
                 fb.update_last_fail(table, idx, ts)
 
             elif kind == FeedbackKind.UNAUTHORIZED:
@@ -271,8 +282,9 @@ async def feedback(
                 fb.apply_server_error(table, idx)
                 fb.update_last_fail(table, idx, ts)
 
-            # Apply authoritative quota data from upstream response headers.
-            if remaining is not None and reset_at_ms is not None:
+            # Quota strategy may receive authoritative quota data from upstream
+            # response headers; the random strategy ignores this entirely.
+            if strategy == "quota" and remaining is not None and reset_at_ms is not None:
                 reset_s = int(reset_at_ms // 1000)
                 fb.apply_quota_update(table, idx, mode_id, remaining, reset_s)
 
@@ -289,6 +301,27 @@ def revision(self) -> int:
         return self._table.revision if self._table else 0
 
 
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+_POOL_INTERVAL_CONFIG: dict[str, tuple[str, int]] = {
+    "basic": ("account.refresh.basic_interval_sec", 36_000),
+    "super": ("account.refresh.super_interval_sec", 7_200),
+    "heavy": ("account.refresh.heavy_interval_sec", 7_200),
+}
+
+
+def _pool_cooling_sec(pool_id: int) -> int:
+    """Cooling seconds for a 429 on a given pool (random strategy only)."""
+    pool_str = POOL_ID_TO_STR.get(pool_id, "basic")
+    interval_key, default_interval = _POOL_INTERVAL_CONFIG.get(
+        pool_str, _POOL_INTERVAL_CONFIG["basic"]
+    )
+    return max(0, int(get_config(interval_key, default_interval)))
+
+
 # ---------------------------------------------------------------------------
 # Module-level singleton
 # ---------------------------------------------------------------------------