feat(fleet): add refresh endpoint

app/main.py

@@ -51,6 +51,7 @@
 from app.modules.dashboard import api as dashboard_api
 from app.modules.dashboard_auth import api as dashboard_auth_api
 from app.modules.firewall import api as firewall_api
+from app.modules.fleet import api as fleet_api
 from app.modules.health import api as health_api
 from app.modules.oauth import api as oauth_api
 from app.modules.proxy import api as proxy_api
@@ -404,6 +405,7 @@ def create_app() -> FastAPI:
     app.include_router(dashboard_auth_api.router)
     app.include_router(settings_api.router)
     app.include_router(firewall_api.router)
+    app.include_router(fleet_api.router)
     app.include_router(sticky_sessions_api.router)
     app.include_router(api_keys_api.router)
     app.include_router(health_api.router)

app/modules/accounts/repository.py

@@ -68,6 +68,15 @@ async def list_accounts(self, *, refresh_existing: bool = False) -> list[Account
         result = await self._session.execute(stmt)
         return list(result.scalars().all())
 
+    async def list_accounts_by_ids(self, account_ids: list[str], *, refresh_existing: bool = False) -> list[Account]:
+        if not account_ids:
+            return []
+        stmt = select(Account).where(Account.id.in_(account_ids)).order_by(Account.email)
+        if refresh_existing:
+            stmt = stmt.execution_options(populate_existing=True)
+        result = await self._session.execute(stmt)
+        return list(result.scalars().all())
+
     async def list_request_usage_summary_by_account(
         self,
         account_ids: list[str] | None = None,

app/modules/accounts/service.py

@@ -128,18 +128,35 @@ def __init__(
         self._encryptor = TokenEncryptor()
         self._auth_manager = auth_manager
 
-    async def list_accounts(self) -> list[AccountSummary]:
-        accounts = await self._repo.list_accounts()
+    async def list_accounts(self, *, account_ids: list[str] | None = None) -> list[AccountSummary]:
+        accounts = (
+            await self._repo.list_accounts_by_ids(account_ids)
+            if account_ids is not None
+            else await self._repo.list_accounts()
+        )
         if not accounts:
             return []
-        account_ids = [account.id for account in accounts]
-        account_id_set = set(account_ids)
-        primary_usage = await self._usage_repo.latest_by_account(window="primary") if self._usage_repo else {}
-        secondary_usage = await self._usage_repo.latest_by_account(window="secondary") if self._usage_repo else {}
-        monthly_usage = await self._usage_repo.latest_by_account(window="monthly") if self._usage_repo else {}
-        request_usage_rows = await self._repo.list_request_usage_summary_by_account(account_ids)
+        visible_account_ids = [account.id for account in accounts]
+        account_id_set = set(visible_account_ids)
+        usage_account_ids = visible_account_ids if account_ids is not None else None
+        primary_usage = (
+            await self._usage_repo.latest_by_account(window="primary", account_ids=usage_account_ids)
+            if self._usage_repo
+            else {}
+        )
+        secondary_usage = (
+            await self._usage_repo.latest_by_account(window="secondary", account_ids=usage_account_ids)
+            if self._usage_repo
+            else {}
+        )
+        monthly_usage = (
+            await self._usage_repo.latest_by_account(window="monthly", account_ids=usage_account_ids)
+            if self._usage_repo
+            else {}
+        )
+        request_usage_rows = await self._repo.list_request_usage_summary_by_account(visible_account_ids)
         limit_warmups_by_account = (
-            await self._limit_warmup_repo.latest_by_account(account_ids) if self._limit_warmup_repo else {}
+            await self._limit_warmup_repo.latest_by_account(visible_account_ids) if self._limit_warmup_repo else {}
         )
         request_usage_by_account = {
             account_id: AccountRequestUsage(
@@ -154,10 +171,18 @@ async def list_accounts(self) -> list[AccountSummary]:
         additional_usage_repo = cast(AdditionalUsageRepository | None, self._additional_usage_repo)
         if additional_usage_repo:
             additional_quota_routing_overrides = await self._repo.additional_quota_routing_policy_overrides()
-            quota_keys = await additional_usage_repo.list_quota_keys(account_ids=account_ids)
+            quota_keys = await additional_usage_repo.list_quota_keys(account_ids=visible_account_ids)
             for quota_key in quota_keys:
-                primary_entries = await additional_usage_repo.latest_by_account(quota_key, "primary")
-                secondary_entries = await additional_usage_repo.latest_by_account(quota_key, "secondary")
+                primary_entries = await additional_usage_repo.latest_by_account(
+                    quota_key,
+                    "primary",
+                    account_ids=visible_account_ids,
+                )
+                secondary_entries = await additional_usage_repo.latest_by_account(
+                    quota_key,
+                    "secondary",
+                    account_ids=visible_account_ids,
+                )
                 for account_id in (set(primary_entries) | set(secondary_entries)) & account_id_set:
                     primary_entry = primary_entries.get(account_id)
                     secondary_entry = secondary_entries.get(account_id)

app/modules/fleet/__init__.py

@@ -0,0 +1 @@
+"""Fleet summary endpoints for external local dashboards."""

app/modules/fleet/api.py

@@ -0,0 +1,134 @@
+from __future__ import annotations
+
+import asyncio
+import logging
+
+from fastapi import APIRouter, Depends, Security
+
+from app.core.auth.dependencies import set_dashboard_error_format, validate_usage_api_key
+from app.core.config.settings_cache import get_settings_cache
+from app.core.utils.time import utcnow
+from app.db.models import AccountStatus
+from app.db.session import get_background_session
+from app.dependencies import AccountsContext, get_accounts_context
+from app.modules.accounts.repository import AccountsRepository
+from app.modules.api_keys.service import ApiKeyData
+from app.modules.fleet.mappers import build_fleet_account_summaries
+from app.modules.fleet.schemas import FleetRefreshResponse, FleetSummaryResponse
+from app.modules.proxy.account_cache import get_account_selection_cache
+from app.modules.proxy.rate_limit_cache import get_rate_limit_headers_cache
+from app.modules.usage.repository import AdditionalUsageRepository, UsageRepository
+from app.modules.usage.updater import UsageUpdater
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(
+    prefix="/api/fleet",
+    tags=["fleet"],
+    dependencies=[Depends(set_dashboard_error_format)],
+)
+
+_REFRESH_SKIP_STATUSES = {AccountStatus.PAUSED, AccountStatus.REAUTH_REQUIRED, AccountStatus.DEACTIVATED}
+
+
+def _visible_account_ids(api_key: ApiKeyData) -> list[str] | None:
+    return list(api_key.assigned_account_ids) if api_key.account_assignment_scope_enabled else None
+
+
+def _usage_sections(raw: str) -> set[str]:
+    if not raw or not raw.strip():
+        return set()
+    return {section.strip() for section in raw.split(",") if section.strip()}
+
+
+async def _can_view_fleet_usage(api_key: ApiKeyData) -> bool:
+    if "account_pool_usage" not in _usage_sections(api_key.usage_sections):
+        return False
+    settings = await get_settings_cache().get()
+    return not bool(getattr(settings, "hide_upstream_quota_from_api_keys", False))
+
+
+@router.get("/summary", response_model=FleetSummaryResponse)
+async def get_fleet_summary(
+    context: AccountsContext = Depends(get_accounts_context),
+    api_key: ApiKeyData = Security(validate_usage_api_key),
+) -> FleetSummaryResponse:
+    """Read-only, minimal per-account capacity summary for fleet consumers."""
+
+    visible_account_ids = _visible_account_ids(api_key)
+    include_usage = await _can_view_fleet_usage(api_key)
+    accounts = await context.service.list_accounts(account_ids=visible_account_ids)
+    persisted_status_by_account_id: dict[str, str] | None = None
+    if not include_usage:
+        persisted_accounts = (
+            await context.repository.list_accounts_by_ids(visible_account_ids, refresh_existing=True)
+            if visible_account_ids is not None
+            else await context.repository.list_accounts(refresh_existing=True)
+        )
+        persisted_status_by_account_id = {account.id: account.status.value for account in persisted_accounts}
+    return FleetSummaryResponse(
+        accounts=build_fleet_account_summaries(
+            accounts,
+            include_usage=include_usage,
+            persisted_status_by_account_id=persisted_status_by_account_id,
+        )
+    )
+
+
+@router.post("/refresh", response_model=FleetRefreshResponse)
+async def refresh_fleet_usage(
+    api_key: ApiKeyData = Security(validate_usage_api_key),
+) -> FleetRefreshResponse:
+    """Request a bounded usage refresh using codex-lb's normal refresh rules."""
+
+    task = asyncio.create_task(
+        _refresh_fleet_usage_with_owned_session(_visible_account_ids(api_key)),
+        name="fleet-usage-refresh",
+    )
+    try:
+        return await asyncio.shield(task)
+    except asyncio.CancelledError:
+        task.add_done_callback(_log_cancelled_refresh_task_exception)
+        raise
+
+
+async def _refresh_fleet_usage_with_owned_session(visible_account_ids: list[str] | None) -> FleetRefreshResponse:
+    async with get_background_session() as session:
+        accounts_repo = AccountsRepository(session)
+        usage_repo = UsageRepository(session)
+        additional_usage_repo = AdditionalUsageRepository(session)
+        accounts = (
+            await accounts_repo.list_accounts_by_ids(visible_account_ids, refresh_existing=True)
+            if visible_account_ids is not None
+            else await accounts_repo.list_accounts(refresh_existing=True)
+        )
+        eligible_accounts = [account for account in accounts if account.status not in _REFRESH_SKIP_STATUSES]
+        latest_primary = await usage_repo.latest_by_account(window="primary", account_ids=visible_account_ids)
+        usage_written = await UsageUpdater(
+            usage_repo,
+            accounts_repo,
+            additional_usage_repo,
+        ).refresh_accounts(eligible_accounts, latest_primary, own_singleflight_sessions=True)
+        if usage_written:
+            await get_rate_limit_headers_cache().invalidate()
+            get_account_selection_cache().invalidate()
+        return FleetRefreshResponse(
+            usage_written=usage_written,
+            account_count=len(accounts),
+            attempted_count=len(eligible_accounts),
+            generated_at=utcnow(),
+        )
+
+
+def _log_cancelled_refresh_task_exception(task: asyncio.Task[FleetRefreshResponse]) -> None:
+    if task.cancelled():
+        return
+    try:
+        exc = task.exception()
+    except asyncio.CancelledError:
+        return
+    if exc is not None:
+        logger.error(
+            "Fleet usage refresh failed after request cancellation",
+            exc_info=(type(exc), exc, exc.__traceback__),
+        )

app/modules/fleet/mappers.py

@@ -0,0 +1,55 @@
+from __future__ import annotations
+
+from app.modules.accounts.schemas import AccountSummary
+from app.modules.fleet.schemas import FleetAccountSummary, FleetWindowSummary
+
+
+def fleet_account_summary_from_account(
+    account: AccountSummary,
+    *,
+    include_usage: bool = True,
+    persisted_status_by_account_id: dict[str, str] | None = None,
+) -> FleetAccountSummary:
+    """Project a dashboard account into the minimal fleet payload."""
+
+    usage = account.usage
+    if include_usage:
+        status = account.status
+    elif persisted_status_by_account_id is None:
+        status = "unknown"
+    else:
+        status = persisted_status_by_account_id.get(account.account_id, "unknown")
+    return FleetAccountSummary(
+        account_id=account.account_id,
+        display_name=account.display_name,
+        email=account.email,
+        status=status,
+        plan_type=account.plan_type,
+        primary=FleetWindowSummary(
+            remaining_percent=usage.primary_remaining_percent if include_usage and usage is not None else None,
+            reset_at=account.reset_at_primary if include_usage else None,
+            window_minutes=account.window_minutes_primary if include_usage else None,
+        ),
+        secondary=FleetWindowSummary(
+            remaining_percent=usage.secondary_remaining_percent if include_usage and usage is not None else None,
+            reset_at=account.reset_at_secondary if include_usage else None,
+            window_minutes=account.window_minutes_secondary if include_usage else None,
+        ),
+        last_refresh_at=account.last_refresh_at if include_usage else None,
+    )
+
+
+def build_fleet_account_summaries(
+    accounts: list[AccountSummary],
+    *,
+    include_usage: bool = True,
+    persisted_status_by_account_id: dict[str, str] | None = None,
+) -> list[FleetAccountSummary]:
+    return [
+        fleet_account_summary_from_account(
+            account,
+            include_usage=include_usage,
+            persisted_status_by_account_id=persisted_status_by_account_id,
+        )
+        for account in accounts
+    ]

app/modules/fleet/schemas.py

@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+from datetime import datetime
+
+from pydantic import Field
+
+from app.modules.shared.schemas import DashboardModel
+
+
+class FleetWindowSummary(DashboardModel):
+    """One minimal capacity window for a single account."""
+
+    remaining_percent: float | None = None
+    reset_at: datetime | None = None
+    window_minutes: int | None = None
+
+
+class FleetAccountSummary(DashboardModel):
+    """Non-sensitive capacity projection for fleet consumers."""
+
+    account_id: str
+    display_name: str
+    email: str
+    status: str
+    plan_type: str
+    primary: FleetWindowSummary
+    secondary: FleetWindowSummary
+    last_refresh_at: datetime | None = None
+
+
+class FleetSummaryResponse(DashboardModel):
+    accounts: list[FleetAccountSummary] = Field(default_factory=list)
+
+
+class FleetRefreshResponse(DashboardModel):
+    ok: bool = True
+    usage_written: bool
+    account_count: int
+    attempted_count: int
+    generated_at: datetime