From 072d4ae0fd1062fc1db32ee50ec9b18d9f6fe034 Mon Sep 17 00:00:00 2001 From: DIodide Date: Sun, 21 Jun 2026 18:51:38 -0400 Subject: [PATCH] feat(usage): real Claude subscription usage bars (5h + weekly) --- apps/web/src/components/usage-display.test.ts | 73 +++++++- apps/web/src/components/usage-display.tsx | 158 ++++++++++++++---- .../app/services/agents/session_manager.py | 60 ++++++- packages/fastapi/app/services/usage.py | 82 +++++++++ .../fastapi/tests/test_subscription_usage.py | 112 +++++++++++++ 5 files changed, 452 insertions(+), 33 deletions(-) create mode 100644 packages/fastapi/tests/test_subscription_usage.py diff --git a/apps/web/src/components/usage-display.test.ts b/apps/web/src/components/usage-display.test.ts index d72d8a2..c0c2d91 100644 --- a/apps/web/src/components/usage-display.test.ts +++ b/apps/web/src/components/usage-display.test.ts @@ -1,5 +1,9 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; -import { accountUsageFromRateLimit, formatResetTime } from "./usage-display"; +import { + accountUsageFromRateLimit, + accountUsagesFromRateLimit, + formatResetTime, +} from "./usage-display"; describe("formatResetTime", () => { beforeEach(() => { @@ -118,3 +122,70 @@ describe("accountUsageFromRateLimit (SDK rate_limit_info shape)", () => { expect(a?.status).toBe("rejected"); }); }); + +describe("accountUsagesFromRateLimit (multi-window buckets shape)", () => { + const futureSec = Math.floor(Date.now() / 1000) + 3600; + const pastSec = Math.floor(Date.now() / 1000) - 3600; + + it("parses 5h + weekly windows, ordered, normalizing 0–1 to 0–100", () => { + const windows = accountUsagesFromRateLimit({ + buckets: { + // out of order on purpose — should sort five_hour first + seven_day: { + utilization: 0.46, + status: "allowed", + resetsAt: futureSec, + }, + five_hour: { + utilization: 0.14, + status: "allowed", + resetsAt: futureSec, + }, + seven_day_sonnet: { + utilization: 0.07, + status: "allowed", + resetsAt: futureSec, + }, + }, + }); + expect(windows.map((w) => w.label)).toEqual([ + "Current session", + "Current week", + "Current week (Sonnet)", + ]); + expect(windows[0].utilization).toBeCloseTo(14); + expect(windows[1].utilization).toBeCloseTo(46); + expect(windows[2].utilization).toBeCloseTo(7); + }); + + it("keeps a rejected window and drops one whose reset has passed", () => { + const windows = accountUsagesFromRateLimit({ + buckets: { + five_hour: { utilization: 1, status: "rejected", resetsAt: futureSec }, + seven_day: { utilization: 0.9, status: "allowed", resetsAt: pastSec }, + }, + }); + expect(windows).toHaveLength(1); + expect(windows[0].label).toBe("Current session"); + expect(windows[0].status).toBe("rejected"); + expect(windows[0].utilization).toBeCloseTo(100); + }); + + it("falls back to the legacy flat single-window shape", () => { + const windows = accountUsagesFromRateLimit({ + rateLimitType: "seven_day", + status: "allowed", + utilization: 73.7, + resetsAt: futureSec, + }); + expect(windows).toHaveLength(1); + expect(windows[0].label).toBe("Current week"); + expect(windows[0].utilization).toBeCloseTo(73.7); + }); + + it("returns [] for an empty/unknown snapshot", () => { + expect(accountUsagesFromRateLimit(null)).toEqual([]); + expect(accountUsagesFromRateLimit({})).toEqual([]); + expect(accountUsagesFromRateLimit({ buckets: {} })).toEqual([]); + }); +}); diff --git a/apps/web/src/components/usage-display.tsx b/apps/web/src/components/usage-display.tsx index 375470c..d839c41 100644 --- a/apps/web/src/components/usage-display.tsx +++ b/apps/web/src/components/usage-display.tsx @@ -85,19 +85,102 @@ export function accountUsageFromRateLimit( }; } -/** The freshest credential's account-limit snapshot (rows already carry the - * account-level rateLimit; pick the most recently active credential). */ -function latestAccountUsage( +// Window ordering for the bars: short rolling window first, then weekly caps. +const WINDOW_ORDER = [ + "five_hour", + "seven_day", + "seven_day_opus", + "seven_day_sonnet", + "overage", +]; + +/** Parse one window of the multi-window `buckets` snapshot (the live 5h/weekly + * utilization Harness fetches from Anthropic's rate-limit headers). */ +function parseWindow(type: string, w: unknown): AccountUsage | null { + if (!w || typeof w !== "object") return null; + const o = w as Record; + const rawStatus = typeof o.status === "string" ? o.status : "allowed"; + const status: AccountUsage["status"] = + rawStatus === "rejected" + ? "rejected" + : rawStatus.includes("warning") + ? "warning" + : "allowed"; + let utilization: number | undefined; + if (typeof o.utilization === "number" && Number.isFinite(o.utilization)) { + // Anthropic's unified headers report a 0–1 fraction; legacy flat + // snapshots used 0–100. Normalize both to a 0–100 percentage. + const u = o.utilization <= 1 ? o.utilization * 100 : o.utilization; + utilization = Math.max(0, Math.min(100, u)); + } + const resetsAtMs = toResetMs(o.resetsAt); + // A window whose reset already passed carries a stale number — drop it. + if (resetsAtMs !== undefined && resetsAtMs <= Date.now()) return null; + if ( + status === "allowed" && + utilization === undefined && + resetsAtMs === undefined + ) { + return null; + } + return { + label: RATE_LIMIT_LABELS[type] ?? "Claude account", + status, + utilization, + resetsAtMs, + }; +} + +/** All renderable windows from a credential's snapshot. Handles the multi-window + * `{ buckets: { five_hour, seven_day, … } }` shape (subscription usage) and the + * legacy flat single-window shape. */ +export function accountUsagesFromRateLimit(rateLimit: unknown): AccountUsage[] { + if (!rateLimit || typeof rateLimit !== "object") return []; + const rl = rateLimit as Record; + const buckets = rl.buckets; + if (buckets && typeof buckets === "object") { + return Object.entries(buckets as Record) + .map(([type, w]) => ({ type, a: parseWindow(type, w) })) + .filter((x): x is { type: string; a: AccountUsage } => x.a !== null) + .sort((x, y) => { + const ix = WINDOW_ORDER.indexOf(x.type); + const iy = WINDOW_ORDER.indexOf(y.type); + return (ix < 0 ? 99 : ix) - (iy < 0 ? 99 : iy); + }) + .map((x) => x.a); + } + const a = accountUsageFromRateLimit(rl); + return a ? [a] : []; +} + +/** The freshest credential's window list (rows carry the account-level snapshot; + * pick the most recently active credential that has one). */ +function latestAccountUsages( rows: AgentUsageRow[] | undefined, -): AccountUsage | null { +): AccountUsage[] { const byRecency = [...(rows ?? [])].sort( (a, b) => (b.lastTurnAt ?? 0) - (a.lastTurnAt ?? 0), ); for (const r of byRecency) { - const a = accountUsageFromRateLimit(r.rateLimit); - if (a) return a; + const windows = accountUsagesFromRateLimit(r.rateLimit); + if (windows.length) return windows; } - return null; + return []; +} + +/** The single most-restrictive window (a hit limit, else the highest %) — used + * for the compact badge color. */ +function latestAccountUsage( + rows: AgentUsageRow[] | undefined, +): AccountUsage | null { + const windows = latestAccountUsages(rows); + if (!windows.length) return null; + const rejected = windows.find((w) => w.status === "rejected"); + if (rejected) return rejected; + return windows.reduce( + (m, w) => ((w.utilization ?? 0) > (m.utilization ?? 0) ? w : m), + windows[0], + ); } /** "in 5h 12m" until a reset timestamp (ms), or undefined when past/absent. */ @@ -467,11 +550,8 @@ export function UsageDisplay() { */ function AccountLimitsSection() { const { data } = useQuery(convexQuery(api.agentUsage.getMyAgentUsage, {})); - const acct = latestAccountUsage(data as AgentUsageRow[] | undefined); - if (!acct) return null; - - const resetsIn = resetsInLabel(acct.resetsAtMs); - const resetSub = resetsIn ? `Resets in ${resetsIn}` : undefined; + const windows = latestAccountUsages(data as AgentUsageRow[] | undefined); + if (windows.length === 0) return null; return (
@@ -483,25 +563,41 @@ function AccountLimitsSection() { your subscription
- {acct.status === "rejected" ? ( -
- {acct.label} limit reached - {resetsIn ? ` · resets in ${resetsIn}` : ""}. Agent turns are paused - until it resets. -
- ) : acct.utilization !== undefined ? ( - - ) : ( -

- {acct.label} - {acct.status === "warning" ? " · approaching limit" : ""} - {resetsIn ? ` · resets in ${resetsIn}` : ""} -

- )} +
+ {windows.map((w) => { + const resetsIn = resetsInLabel(w.resetsAtMs); + const resetSub = resetsIn ? `Resets in ${resetsIn}` : undefined; + if (w.status === "rejected") { + return ( +
+ {w.label} limit reached + {resetsIn ? ` · resets in ${resetsIn}` : ""}. Agent turns are + paused until it resets. +
+ ); + } + if (w.utilization !== undefined) { + return ( + + ); + } + return ( +

+ {w.label} + {w.status === "warning" ? " · approaching limit" : ""} + {resetsIn ? ` · resets in ${resetsIn}` : ""} +

+ ); + })} +
); } diff --git a/packages/fastapi/app/services/agents/session_manager.py b/packages/fastapi/app/services/agents/session_manager.py index e9bfaf2..6f7fff1 100644 --- a/packages/fastapi/app/services/agents/session_manager.py +++ b/packages/fastapi/app/services/agents/session_manager.py @@ -1025,6 +1025,10 @@ def __init__(self): # Fire-and-forget usage-recording tasks; held so they aren't GC'd # mid-flight, discarded on completion. self._usage_tasks: set[asyncio.Task] = set() + # Per-credential debounce for the subscription-usage ping (monotonic ts + # of the last fetch) — the windows move slowly, so once a minute is plenty + # and keeps the extra inference calls negligible. + self._sub_usage_fetched_at: dict[str, float] = {} @staticmethod def _runtime_key( @@ -1043,6 +1047,55 @@ def _http_client(self) -> httpx.AsyncClient: self._http = httpx.AsyncClient(timeout=30.0) return self._http + def _maybe_fetch_subscription_usage( + self, session: AgentSession, cred_id: str + ) -> None: + """Kick off a debounced subscription-usage fetch for this credential.""" + now = time.monotonic() + if now - self._sub_usage_fetched_at.get(cred_id, 0.0) < 60.0: + return + # Set before firing so the many agent_usage events in one turn don't all + # queue a fetch; a failure simply waits out the TTL before retrying. + self._sub_usage_fetched_at[cred_id] = now + task = asyncio.create_task( + self._fetch_subscription_usage(session.user_id, cred_id) + ) + self._usage_tasks.add(task) + task.add_done_callback(self._usage_tasks.discard) + + async def _fetch_subscription_usage(self, user_id: str, cred_id: str) -> None: + """Resolve the credential's OAuth token, read the live 5h + weekly + windows off Anthropic's rate-limit headers, and persist them onto the + credential. Only OAuth (subscription) credentials carry these windows; + api-key credentials are skipped. Best-effort.""" + try: + from app.services.agents.credentials import resolve_agent_credentials + from app.services.usage import ( + fetch_subscription_usage, + record_agent_rate_limit, + ) + + creds = await resolve_agent_credentials( + self._http_client(), "claude-code", user_id, cred_id + ) + token = (creds.env or {}).get("CLAUDE_CODE_OAUTH_TOKEN") + if not token: + return # api-key credential — no subscription windows to show + usage = await fetch_subscription_usage(self._http_client(), token) + if usage: + await record_agent_rate_limit( + self._http_client(), + user_id=user_id, + agent_credential_id=cred_id, + rate_limit=usage, + ) + except Exception: + logger.warning( + "Subscription usage fetch failed for credential '%s'", + cred_id, + exc_info=True, + ) + async def _inject_workspace_env( self, creds, harness: HarnessConfig, user_id: str, ) -> str: @@ -2108,10 +2161,10 @@ async def _process_session_update(self, session: AgentSession, params: dict) -> # see the reset time. if event["event"] == "agent_usage": rl = (update.get("_meta") or {}).get("_claude/rateLimit") + cred_id = session.harness.agent_credential_id if rl is not None: changed = rl != session.last_rate_limit session.last_rate_limit = rl - cred_id = session.harness.agent_credential_id # Only persist when the snapshot actually changed — avoids a # credential-row write on every usage_update that merely # repeats the same rate-limit state. @@ -2128,6 +2181,11 @@ async def _process_session_update(self, session: AgentSession, params: dict) -> ) self._usage_tasks.add(rl_task) rl_task.add_done_callback(self._usage_tasks.discard) + elif cred_id: + # The Claude Agent SDK doesn't surface subscription usage in + # the stream (the _meta snapshot is empty), so fetch the 5h + + # weekly windows ourselves — debounced per credential. + self._maybe_fetch_subscription_usage(session, cred_id) # Persist per-credential agent usage on the terminal usage_update # (the only one carrying cost). Thin: no cache tokens, cache-excluded # cost — for claude-code the SDK result message upgrades this row in diff --git a/packages/fastapi/app/services/usage.py b/packages/fastapi/app/services/usage.py index 6fbe1eb..0624a0c 100644 --- a/packages/fastapi/app/services/usage.py +++ b/packages/fastapi/app/services/usage.py @@ -288,3 +288,85 @@ async def record_agent_rate_limit( logger.exception( "Failed to record rate limit for credential '%s'", agent_credential_id ) + + +# Any model the OAuth subscription can call works — the unified windows are +# account-level, not model-specific (the per-model 7d_sonnet cap aside). Bumped +# only if Anthropic retires it; a failed ping just yields no bars (best-effort). +SUBSCRIPTION_USAGE_PING_MODEL = "claude-sonnet-4-5-20250929" + +# Maps the Anthropic `anthropic-ratelimit-unified--*` header group to the +# window name the frontend renders (RATE_LIMIT_LABELS in usage-display.tsx). +_UNIFIED_WINDOWS = ( + ("five_hour", "5h"), + ("seven_day", "7d"), + ("seven_day_sonnet", "7d_sonnet"), +) + + +def _parse_unified_rate_limit_headers(headers: httpx.Headers) -> dict | None: + """Build a multi-window snapshot from Anthropic's unified rate-limit headers. + + Returns the `{"buckets": {...}}` shape the credential stores and the panel + renders, or None when no unified window is present (e.g. an api-key + credential, which has no subscription windows). Utilization is the raw 0–1 + fraction; reset is the Unix-seconds the headers carry. + """ + buckets: dict[str, dict] = {} + for name, key in _UNIFIED_WINDOWS: + util = headers.get(f"anthropic-ratelimit-unified-{key}-utilization") + if util is None: + continue + try: + window: dict = {"utilization": float(util)} + except ValueError: + continue + status = headers.get(f"anthropic-ratelimit-unified-{key}-status") + if status: + window["status"] = status + reset = headers.get(f"anthropic-ratelimit-unified-{key}-reset") + if reset is not None: + try: + window["resetsAt"] = int(reset) + except ValueError: + pass + buckets[name] = window + return {"buckets": buckets} if buckets else None + + +async def fetch_subscription_usage( + http_client: httpx.AsyncClient, oauth_token: str +) -> dict | None: + """Read the user's Claude subscription usage (5h + weekly windows). + + The Agent SDK / ACP does NOT surface these (the `_meta._claude/rateLimit` + snapshot is empty), and the `/api/oauth/usage` endpoint needs a `user:profile` + scope agent tokens lack — but the unified rate-limit headers ride on every + `/v1/messages` response, which is exactly what Claude Code's own `/usage` + reads. So make the cheapest possible inference call and read the headers off + it. Best-effort: returns None on any failure rather than disrupting a turn. + """ + try: + resp = await http_client.post( + "https://api.anthropic.com/v1/messages", + headers={ + "Authorization": f"Bearer {oauth_token}", + "anthropic-version": "2023-06-01", + "anthropic-beta": "oauth-2025-04-20", + "User-Agent": "claude-code/2.1.0", + "Content-Type": "application/json", + }, + json={ + "model": SUBSCRIPTION_USAGE_PING_MODEL, + "max_tokens": 1, + "system": "You are Claude Code, Anthropic's official CLI for Claude.", + "messages": [{"role": "user", "content": "."}], + }, + timeout=20.0, + ) + except Exception: + logger.warning("Subscription usage ping failed", exc_info=True) + return None + # Read headers regardless of status: a 429 still reports the (maxed) windows, + # which is exactly when the user needs to see them. + return _parse_unified_rate_limit_headers(resp.headers) diff --git a/packages/fastapi/tests/test_subscription_usage.py b/packages/fastapi/tests/test_subscription_usage.py new file mode 100644 index 0000000..a5ea3d3 --- /dev/null +++ b/packages/fastapi/tests/test_subscription_usage.py @@ -0,0 +1,112 @@ +"""Subscription usage: parse Anthropic's unified rate-limit headers into the +multi-window snapshot the panel renders, and read them off a /v1/messages ping.""" + +import httpx +import respx + +from app.services.usage import ( + _parse_unified_rate_limit_headers, + fetch_subscription_usage, +) + + +def _headers(**overrides: str) -> httpx.Headers: + base = { + "anthropic-ratelimit-unified-5h-utilization": "0.06", + "anthropic-ratelimit-unified-5h-status": "allowed", + "anthropic-ratelimit-unified-5h-reset": "1782095400", + "anthropic-ratelimit-unified-7d-utilization": "0.44", + "anthropic-ratelimit-unified-7d-status": "allowed", + "anthropic-ratelimit-unified-7d-reset": "1782226800", + "anthropic-ratelimit-unified-7d_sonnet-utilization": "0.07", + "anthropic-ratelimit-unified-7d_sonnet-status": "allowed", + "anthropic-ratelimit-unified-7d_sonnet-reset": "1782226800", + } + base.update(overrides) + return httpx.Headers(base) + + +def test_parses_all_three_windows(): + out = _parse_unified_rate_limit_headers(_headers()) + assert out == { + "buckets": { + "five_hour": { + "utilization": 0.06, + "status": "allowed", + "resetsAt": 1782095400, + }, + "seven_day": { + "utilization": 0.44, + "status": "allowed", + "resetsAt": 1782226800, + }, + "seven_day_sonnet": { + "utilization": 0.07, + "status": "allowed", + "resetsAt": 1782226800, + }, + } + } + + +def test_no_unified_headers_returns_none(): + # An api-key credential's response has standard rate-limit headers but no + # subscription windows — nothing to show. + assert _parse_unified_rate_limit_headers(httpx.Headers({})) is None + assert ( + _parse_unified_rate_limit_headers( + httpx.Headers({"anthropic-ratelimit-requests-remaining": "100"}) + ) + is None + ) + + +def test_rejected_window_is_kept(): + out = _parse_unified_rate_limit_headers( + _headers(**{"anthropic-ratelimit-unified-7d-status": "rejected"}) + ) + assert out["buckets"]["seven_day"]["status"] == "rejected" + + +def test_non_numeric_utilization_skipped(): + out = _parse_unified_rate_limit_headers( + _headers(**{"anthropic-ratelimit-unified-5h-utilization": "n/a"}) + ) + assert "five_hour" not in out["buckets"] + assert "seven_day" in out["buckets"] + + +@respx.mock +async def test_fetch_reads_headers_off_the_ping(): + respx.post("https://api.anthropic.com/v1/messages").mock( + return_value=httpx.Response(200, headers=dict(_headers()), json={"id": "x"}) + ) + async with httpx.AsyncClient() as client: + out = await fetch_subscription_usage(client, "sk-ant-oat01-fake") + assert out["buckets"]["five_hour"]["utilization"] == 0.06 + assert out["buckets"]["seven_day"]["utilization"] == 0.44 + + +@respx.mock +async def test_fetch_reads_headers_even_on_429(): + # A maxed-out subscription returns 429 but still reports the windows. + respx.post("https://api.anthropic.com/v1/messages").mock( + return_value=httpx.Response( + 429, + headers=dict( + _headers(**{"anthropic-ratelimit-unified-5h-status": "rejected"}) + ), + ) + ) + async with httpx.AsyncClient() as client: + out = await fetch_subscription_usage(client, "sk-ant-oat01-fake") + assert out["buckets"]["five_hour"]["status"] == "rejected" + + +async def test_fetch_returns_none_on_network_error(): + # No respx route registered + a real client → connection error → None, never + # raises into the turn. + async with httpx.AsyncClient(transport=httpx.MockTransport( + lambda req: (_ for _ in ()).throw(httpx.ConnectError("boom")) + )) as client: + assert await fetch_subscription_usage(client, "tok") is None