feat(reset-credits): add banked rate-limit reset credits

app/core/clients/headers.py

@@ -0,0 +1,29 @@
+from __future__ import annotations
+
+from app.core.utils.request_id import get_request_id
+
+
+def build_chatgpt_auth_headers(
+    access_token: str,
+    account_id: str | None,
+    *,
+    extra: dict[str, str] | None = None,
+) -> dict[str, str]:
+    """Build the headers required to call ChatGPT ``backend-api`` endpoints.
+
+    Includes the OAuth bearer token and the ``chatgpt-account-id`` header. The
+    account-id header is omitted when the id is a synthetic ``email_``/``local_``
+    prefix, matching upstream behavior. An active request id (if any) is attached.
+    """
+    headers: dict[str, str] = {
+        "Authorization": f"Bearer {access_token}",
+        "Accept": "application/json",
+    }
+    request_id = get_request_id()
+    if request_id:
+        headers["x-request-id"] = request_id
+    if account_id and not account_id.startswith(("email_", "local_")):
+        headers["chatgpt-account-id"] = account_id
+    if extra:
+        headers.update(extra)
+    return headers

app/core/clients/rate_limit_reset_credits.py

@@ -0,0 +1,284 @@
+from __future__ import annotations
+
+import asyncio
+import logging
+import uuid
+from datetime import datetime
+
+import aiohttp
+from aiohttp_retry import ExponentialRetry, RetryClient
+from pydantic import BaseModel, ConfigDict, Field, ValidationError
+
+from app.core.clients.headers import build_chatgpt_auth_headers
+from app.core.clients.http import lease_retry_client
+from app.core.config.settings import get_settings
+from app.core.types import JsonObject
+from app.core.utils.request_id import get_request_id
+
+RETRYABLE_STATUS = {408, 429, 500, 502, 503, 504}
+RETRY_START_TIMEOUT = 0.5
+RETRY_MAX_TIMEOUT = 2.0
+
+logger = logging.getLogger(__name__)
+
+
+class ResetCreditFetchError(Exception):
+    def __init__(self, status_code: int, message: str, code: str | None = None) -> None:
+        super().__init__(message)
+        self.status_code = status_code
+        self.message = message
+        self.code = code
+
+
+class ConsumeResetCreditError(Exception):
+    def __init__(self, status_code: int, message: str, code: str | None = None) -> None:
+        super().__init__(message)
+        self.status_code = status_code
+        self.message = message
+        self.code = code
+
+
+class ResetCreditItem(BaseModel):
+    model_config = ConfigDict(extra="ignore")
+
+    id: str
+    reset_type: str | None = None
+    status: str | None = None
+    granted_at: datetime | None = None
+    expires_at: datetime | None = None
+    title: str | None = None
+    description: str | None = None
+    redeem_started_at: datetime | None = None
+    redeemed_at: datetime | None = None
+
+
+class ResetCreditsResponse(BaseModel):
+    model_config = ConfigDict(extra="ignore")
+
+    credits: list[ResetCreditItem]
+    available_count: int
+
+
+class ConsumeResetCreditCredit(BaseModel):
+    model_config = ConfigDict(extra="ignore")
+
+    id: str | None = None
+    reset_type: str | None = None
+    status: str | None = None
+    redeemed_at: datetime | None = None
+
+
+class ConsumeResetCreditResponse(BaseModel):
+    model_config = ConfigDict(extra="ignore")
+
+    code: str
+    credit: ConsumeResetCreditCredit
+    windows_reset: int
+
+
+class RateLimitResetCreditsSnapshot(BaseModel):
+    """In-memory snapshot of a single account's banked reset credits.
+
+    Carries the upstream ``available_count``, the soonest expiry among the
+    available credits, and the full credit list so dashboard endpoints can
+    render details and the consume path can re-select at click time.
+    """
+
+    model_config = ConfigDict(extra="ignore")
+
+    available_count: int = 0
+    nearest_expires_at: datetime | None = None
+    credits: list[ResetCreditItem] = Field(default_factory=list)
+
+
+async def fetch_reset_credits(
+    access_token: str,
+    account_id: str | None,
+    *,
+    base_url: str | None = None,
+    timeout_seconds: float | None = None,
+    max_retries: int | None = None,
+    client: RetryClient | None = None,
+) -> ResetCreditsResponse:
+    settings = get_settings()
+    usage_base = base_url or settings.upstream_base_url
+    url = _reset_credits_url(usage_base)
+    timeout = aiohttp.ClientTimeout(total=timeout_seconds or settings.usage_fetch_timeout_seconds)
+    retries = max_retries if max_retries is not None else settings.usage_fetch_max_retries
+    headers = build_chatgpt_auth_headers(access_token, account_id)
+    retry_options = _retry_options(retries + 1)
+
+    try:
+        async with lease_retry_client(client) as retry_client:
+            async with retry_client.request(
+                "GET",
+                url,
+                headers=headers,
+                timeout=timeout,
+                retry_options=retry_options,
+            ) as resp:
+                data = await _safe_json(resp)
+                if resp.status >= 400:
+                    code = _extract_error_code(data)
+                    message = _extract_error_message(data) or f"Reset credits fetch failed ({resp.status})"
+                    logger.warning(
+                        "Reset credits fetch failed request_id=%s status=%s code=%s message=%s",
+                        get_request_id(),
+                        resp.status,
+                        code,
+                        message,
+                    )
+                    raise ResetCreditFetchError(resp.status, message, code=code)
+                try:
+                    return ResetCreditsResponse.model_validate(_success_payload(data))
+                except (ValueError, ValidationError) as exc:
+                    logger.warning("Reset credits fetch invalid payload request_id=%s", get_request_id())
+                    raise ResetCreditFetchError(502, "Invalid reset credits payload") from exc
+    except (aiohttp.ClientError, asyncio.TimeoutError) as exc:
+        logger.warning("Reset credits fetch error request_id=%s error=%s", get_request_id(), exc)
+        raise ResetCreditFetchError(0, f"Reset credits fetch failed: {exc}") from exc
+
+
+async def consume_reset_credit(
+    access_token: str,
+    account_id: str | None,
+    credit_id: str,
+    *,
+    base_url: str | None = None,
+    timeout_seconds: float | None = None,
+    max_retries: int | None = None,
+    client: RetryClient | None = None,
+) -> ConsumeResetCreditResponse:
+    settings = get_settings()
+    usage_base = base_url or settings.upstream_base_url
+    url = _consume_url(usage_base)
+    timeout = aiohttp.ClientTimeout(total=timeout_seconds or settings.usage_fetch_timeout_seconds)
+    retries = max_retries if max_retries is not None else settings.usage_fetch_max_retries
+    headers = build_chatgpt_auth_headers(
+        access_token,
+        account_id,
+        extra={"Content-Type": "application/json"},
+    )
+    redeem_request_id = str(uuid.uuid4())
+    body = {"credit_id": credit_id, "redeem_request_id": redeem_request_id}
+    retry_options = _retry_options(retries + 1)
+
+    try:
+        async with lease_retry_client(client) as retry_client:
+            async with retry_client.request(
+                "POST",
+                url,
+                headers=headers,
+                json=body,
+                timeout=timeout,
+                retry_options=retry_options,
+            ) as resp:
+                data = await _safe_json(resp)
+                if resp.status >= 400:
+                    code = _extract_error_code(data)
+                    message = _extract_error_message(data) or f"Reset credits consume failed ({resp.status})"
+                    logger.warning(
+                        "Reset credits consume failed request_id=%s status=%s code=%s message=%s",
+                        get_request_id(),
+                        resp.status,
+                        code,
+                        message,
+                    )
+                    raise ConsumeResetCreditError(resp.status, message, code=code)
+                try:
+                    return ConsumeResetCreditResponse.model_validate(_success_payload(data))
+                except (ValueError, ValidationError) as exc:
+                    logger.warning("Reset credits consume invalid payload request_id=%s", get_request_id())
+                    raise ConsumeResetCreditError(502, "Invalid reset credits consume payload") from exc
+    except (aiohttp.ClientError, asyncio.TimeoutError) as exc:
+        logger.warning("Reset credits consume error request_id=%s error=%s", get_request_id(), exc)
+        raise ConsumeResetCreditError(0, f"Reset credits consume failed: {exc}") from exc
+
+
+def build_snapshot(response: ResetCreditsResponse) -> RateLimitResetCreditsSnapshot:
+    """Project an upstream list response into the cached snapshot shape."""
+    nearest = _nearest_available_expires_at(response.credits)
+    return RateLimitResetCreditsSnapshot(
+        available_count=response.available_count,
+        nearest_expires_at=nearest,
+        credits=list(response.credits),
+    )
+
+
+def _nearest_available_expires_at(credits: list[ResetCreditItem]) -> datetime | None:
+    candidates = [
+        credit.expires_at for credit in credits if credit.status == "available" and credit.expires_at is not None
+    ]
+    return min(candidates) if candidates else None
+
+
+def _reset_credits_url(base_url: str) -> str:
+    normalized = base_url.rstrip("/")
+    if "/backend-api" not in normalized:
+        normalized = f"{normalized}/backend-api"
+    return f"{normalized}/wham/rate-limit-reset-credits"
+
+
+def _consume_url(base_url: str) -> str:
+    return f"{_reset_credits_url(base_url)}/consume"
+
+
+async def _safe_json(resp: aiohttp.ClientResponse) -> JsonObject:
+    try:
+        data = await resp.json(content_type=None)
+    except Exception:
+        text = await resp.text()
+        return {"error": {"message": text.strip()}}
+    return data if isinstance(data, dict) else {"error": {"message": str(data)}}
+
+
+def _success_payload(payload: JsonObject) -> JsonObject:
+    if "error" in payload:
+        raise ValueError("success response carried error payload")
+    return payload
+
+
+class _ErrorEnvelope(BaseModel):
+    model_config = ConfigDict(extra="ignore")
+
+    error: dict[str, str | None] | str | None = None
+    error_description: str | None = None
+    message: str | None = None
+
+
+def _extract_error_message(payload: JsonObject) -> str | None:
+    envelope = _ErrorEnvelope.model_validate(payload)
+    error = envelope.error
+    if isinstance(error, dict):
+        message = error.get("message")
+        if isinstance(message, str) and message:
+            return message
+        description = error.get("error_description")
+        if isinstance(description, str) and description:
+            return description
+    if isinstance(error, str) and error:
+        return envelope.error_description or error
+    return envelope.message
+
+
+def _extract_error_code(payload: JsonObject) -> str | None:
+    envelope = _ErrorEnvelope.model_validate(payload)
+    error = envelope.error
+    if isinstance(error, dict):
+        code = error.get("code")
+        if isinstance(code, str):
+            normalized = code.strip().lower()
+            return normalized or None
+    return None
+
+
+def _retry_options(attempts: int) -> ExponentialRetry:
+    return ExponentialRetry(
+        attempts=attempts,
+        start_timeout=RETRY_START_TIMEOUT,
+        max_timeout=RETRY_MAX_TIMEOUT,
+        factor=2.0,
+        statuses=RETRYABLE_STATUS,
+        exceptions={aiohttp.ClientError, asyncio.TimeoutError},
+        retry_all_server_errors=False,
+    )

app/core/clients/usage.py

@@ -13,6 +13,7 @@
     create_codex_session,
     require_route_or_direct_egress_opt_in,
 )
+from app.core.clients.headers import build_chatgpt_auth_headers
 from app.core.clients.http import lease_retry_client
 from app.core.config.settings import get_settings
 from app.core.types import JsonObject
@@ -220,13 +221,7 @@ def _usage_url(base_url: str) -> str:
 
 
 def _usage_headers(access_token: str, account_id: str | None) -> dict[str, str]:
-    headers = {"Authorization": f"Bearer {access_token}", "Accept": "application/json"}
-    request_id = get_request_id()
-    if request_id:
-        headers["x-request-id"] = request_id
-    if account_id and not account_id.startswith(("email_", "local_")):
-        headers["chatgpt-account-id"] = account_id
-    return headers
+    return build_chatgpt_auth_headers(access_token, account_id)
 
 
 async def _safe_json(resp: aiohttp.ClientResponse) -> JsonObject:

app/core/config/settings.py

@@ -193,6 +193,7 @@ class Settings(BaseSettings):
     usage_fetch_max_retries: int = 2
     usage_refresh_enabled: bool = True
     usage_refresh_interval_seconds: int = Field(default=60, gt=0)
+    rate_limit_reset_credits_refresh_interval_seconds: int = Field(default=60, gt=0)
     openai_cache_affinity_max_age_seconds: int = Field(default=1800, gt=0)
     warmup_model: str = "gpt-5.4-mini"
     openai_prompt_cache_key_derivation_enabled: bool = True

app/core/exceptions.py

@@ -82,3 +82,8 @@ class DashboardRateLimitError(AppError):
     def __init__(self, message: str, *, retry_after: int, code: str | None = None) -> None:
         self.retry_after = retry_after
         super().__init__(message, code=code)
+
+
+class DashboardServiceUnavailableError(AppError):
+    status_code = 503
+    code = "service_unavailable"

app/core/handlers/exceptions.py

@@ -20,6 +20,7 @@
     DashboardNotFoundError,
     DashboardPermissionError,
     DashboardRateLimitError,
+    DashboardServiceUnavailableError,
     DashboardValidationError,
     ProxyAuthError,
     ProxyModelNotAllowed,
@@ -45,6 +46,7 @@
     DashboardBadRequestError,
     DashboardValidationError,
     DashboardRateLimitError,
+    DashboardServiceUnavailableError,
 )