diff --git a/apps/mobile/src/features/usage/UsageLimitsSection.tsx b/apps/mobile/src/features/usage/UsageLimitsSection.tsx new file mode 100644 index 000000000000..345ad52ca1e4 --- /dev/null +++ b/apps/mobile/src/features/usage/UsageLimitsSection.tsx @@ -0,0 +1,183 @@ +import { useAtomValue } from "@effect/atom-react"; +import type { + ServerProvider, + ServerProviderUsageWindow, + UsageLimitSourceAccount, +} from "@t3tools/contracts"; +import { + collectLimitSources, + collectLimitsGroups, + elapsedShare, + formatResetsIn, + limitsNotice, + paceOf, + providerLimitsLabel, +} from "@t3tools/shared/usageLimits"; +import { useState } from "react"; +import { View } from "react-native"; + +import { AppText as Text } from "../../components/AppText"; +import { environmentPresentations } from "../../state/presentation"; +import { SettingsSection } from "../settings/components/SettingsSection"; + +const PACE_LABEL = { ahead: "ahead of pace", on: "on pace", under: "under pace" } as const; + +/** + * One window as a bar spanning its whole duration: the fill is quota spent, + * the hairline is how far into the window the clock is. + */ +function WindowBar(props: { readonly window: ServerProviderUsageWindow; readonly now: number }) { + const { window, now } = props; + const used = Math.round(Math.max(0, Math.min(100, window.usedPercent))); + const elapsed = elapsedShare(window, now); + const pace = paceOf(window, now); + const resetsIn = formatResetsIn(window, now); + const detail = [pace ? PACE_LABEL[pace] : null, resetsIn].filter(Boolean).join(" · "); + return ( + + + {window.label} + {used}% used + + + + = 90 + ? "h-full rounded-full bg-destructive" + : used >= 70 + ? "h-full rounded-full bg-warning" + : "h-full rounded-full bg-foreground" + } + style={{ flex: used }} + /> + + + {elapsed !== null ? ( + + ) : null} + + {detail ? {detail} : null} + + ); +} + +function AccountLimits(props: { + readonly label: string; + readonly detail: string | undefined; + readonly limits: ServerProvider["usageLimits"]; + readonly now: number; + readonly first: boolean; +}) { + const { limits, now } = props; + if (!limits) return null; + const notice = limitsNotice(limits); + return ( + + + {props.label} + {props.detail ? ( + {props.detail} + ) : null} + + {notice ? ( + {notice} + ) : ( + limits.windows.map((window) => ) + )} + + ); +} + +function ProviderLimits(props: { + readonly provider: ServerProvider; + readonly now: number; + readonly first: boolean; +}) { + const { provider } = props; + return ( + undefined)} + detail={provider.auth.label} + limits={provider.usageLimits} + now={props.now} + first={props.first} + /> + ); +} + +const DRIVER_LABEL: Partial> = { codex: "Codex", claudeAgent: "Claude" }; + +/** Emails stay off the phone screen; the plan and driver identify the row. */ +function SourceAccountLimits(props: { + readonly account: UsageLimitSourceAccount; + readonly now: number; + readonly first: boolean; +}) { + const { account } = props; + return ( + + ); +} + +/** + * Subscription quota windows from every connected environment's providers, + * read from the config each environment already streams. Countdowns anchor to + * render time rather than ticking. + */ +export function UsageLimitsSection() { + const presentations = useAtomValue(environmentPresentations.presentationsAtom); + const groups = collectLimitsGroups(presentations); + const sources = collectLimitSources(presentations); + // Anchored once per mount on purpose: countdowns must not tick. + const [now] = useState(() => Date.now()); + if (groups.length === 0 && sources.length === 0) return null; + + return ( + <> + {sources.map((source) => ( + + {source.error ? ( + {source.error} + ) : source.accounts.length === 0 ? ( + No accounts reported. + ) : ( + source.accounts.map((account, index) => ( + + )) + )} + + ))} + {groups.map((group) => ( + + {group.providers.map((provider, index) => ( + + ))} + + ))} + + ); +} diff --git a/apps/mobile/src/features/usage/UsageRouteScreen.tsx b/apps/mobile/src/features/usage/UsageRouteScreen.tsx index 817e6d7f9543..6e913b4999d8 100644 --- a/apps/mobile/src/features/usage/UsageRouteScreen.tsx +++ b/apps/mobile/src/features/usage/UsageRouteScreen.tsx @@ -21,6 +21,7 @@ import { NativeStackScreenOptions } from "../../native/StackHeader"; import { useUsage, type EnvironmentUsageStatus } from "../../state/usage"; import { SettingsSection } from "../settings/components/SettingsSection"; import { UsageDailyChart } from "./UsageDailyChart"; +import { UsageLimitsSection } from "./UsageLimitsSection"; import type { UsageChartMetric } from "./usageChartData"; import { PROVIDER_LABEL, useProviderColors } from "./usageProviders"; @@ -139,6 +140,7 @@ export function UsageRouteScreen() { timeZone={window.timeZone} /> + diff --git a/apps/mobile/src/state/server.ts b/apps/mobile/src/state/server.ts index 1b7060571a5b..2157c72e13ef 100644 --- a/apps/mobile/src/state/server.ts +++ b/apps/mobile/src/state/server.ts @@ -7,6 +7,7 @@ import { environmentSession } from "./session"; export const serverEnvironment = createServerEnvironmentAtoms(connectionAtomRuntime, { initialConfigValueAtom: environmentSession.initialConfigValueAtom, + usageLimitSources: true, }); export const environmentServerConfigsAtom = createEnvironmentServerConfigsAtom({ catalogValueAtom: environmentCatalog.catalogValueAtom, diff --git a/apps/server/src/environment/ServerEnvironment.ts b/apps/server/src/environment/ServerEnvironment.ts index 1010011e90cd..d050bde88565 100644 --- a/apps/server/src/environment/ServerEnvironment.ts +++ b/apps/server/src/environment/ServerEnvironment.ts @@ -221,6 +221,7 @@ export const make = Effect.gen(function* () { threadAutoSettlement: true, threadSnooze: true, environmentThemes: true, + usageLimitSources: true, threadPinning: true, threadPinReorder: true, threadTitleRegeneration: true, diff --git a/apps/server/src/provider/Drivers/ClaudeDriver.ts b/apps/server/src/provider/Drivers/ClaudeDriver.ts index d47d9c062ae2..efcf19e9d6b1 100644 --- a/apps/server/src/provider/Drivers/ClaudeDriver.ts +++ b/apps/server/src/provider/Drivers/ClaudeDriver.ts @@ -29,6 +29,7 @@ import { ServerConfig } from "../../config.ts"; import { ServerSettingsService } from "../../serverSettings.ts"; import { ProviderDriverError } from "../Errors.ts"; import { makeClaudeAdapter } from "../Layers/ClaudeAdapter.ts"; +import { makeClaudeScopedLimitNames } from "../Layers/claudeUsageLimits.ts"; import { checkClaudeProviderStatus, makePendingClaudeProvider, @@ -134,10 +135,14 @@ export const ClaudeDriver: ProviderDriver = { continuationGroupKey, }); + // One per instance: the status probe writes the model-scoped bucket + // names it saw, the adapter reads them to place turn-driven events. + const scopedLimitNames = yield* makeClaudeScopedLimitNames; const adapterOptions = { instanceId, environment: processEnv, modelCatalog, + scopedLimitNames, ...(eventLoggers.native ? { nativeEventLogger: eventLoggers.native } : {}), }; const adapter = yield* makeClaudeAdapter(effectiveConfig, adapterOptions); @@ -171,6 +176,7 @@ export const ClaudeDriver: ProviderDriver = { processEnv, cwd, resolveClaudeModelCatalog(manifest), + scopedLimitNames, ), ), Effect.map(stampIdentity), diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts index 3b5c0f8586a7..eb8f131009b4 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts @@ -28,6 +28,7 @@ import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Random from "effect/Random"; +import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import * as TestClock from "effect/testing/TestClock"; @@ -44,6 +45,7 @@ import { } from "../ClaudeModelCatalog.testFixtures.ts"; import { ProviderAdapterProcessError, ProviderAdapterValidationError } from "../Errors.ts"; import type { ClaudeAdapterShape } from "../Services/ClaudeAdapter.ts"; +import type { ClaudeScopedLimitNames } from "./claudeUsageLimits.ts"; import { makeClaudeAdapter, type ClaudeAdapterLiveOptions } from "./ClaudeAdapter.ts"; const decodeClaudeSettings = Schema.decodeSync(ClaudeSettings); const encodeUnknownJsonString = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown)); @@ -163,6 +165,7 @@ function makeHarness(config?: { readonly baseDir?: string; readonly claudeConfig?: Partial; readonly instanceId?: ProviderInstanceId; + readonly scopedLimitNames?: ClaudeAdapterLiveOptions["scopedLimitNames"]; }) { const query = new FakeClaudeQuery(); let createInput: @@ -174,6 +177,7 @@ function makeHarness(config?: { const adapterOptions: ClaudeAdapterLiveOptions = { ...(config?.instanceId ? { instanceId: config.instanceId } : {}), + ...(config?.scopedLimitNames ? { scopedLimitNames: config.scopedLimitNames } : {}), modelCatalog: Effect.succeed(SYNTHETIC_CLAUDE_MODEL_CATALOG), createQuery: (input) => { createInput = input; @@ -1239,6 +1243,84 @@ describe("ClaudeAdapterLive", () => { ); }); + it.effect("places overage-included rate-limit events on the bucket the probe named", () => { + const scopedLimitNames = Ref.makeUnsafe({ overageIncluded: undefined }); + const harness = makeHarness({ scopedLimitNames }); + const rateLimitEvent = (utilization: number): SDKMessage => + ({ + type: "rate_limit_event", + rate_limit_info: { + status: "allowed", + rateLimitType: "seven_day_overage_included", + utilization, + }, + uuid: `rate-limit-${utilization}`, + session_id: "sdk-session-1", + }) as unknown as SDKMessage; + const resultMessage = (uuid: string): SDKMessage => + ({ + type: "result", + subtype: "success", + is_error: false, + errors: [], + num_turns: 1, + session_id: "sdk-session-1", + uuid, + }) as unknown as SDKMessage; + const limitsUpdates = (events: Iterable) => + Array.from(events).flatMap((event) => + event.type === "account.rate-limits.updated" ? [event.payload.limits] : [], + ); + return Effect.gen(function* () { + const adapter = yield* ClaudeAdapter; + const session = yield* adapter.startSession({ + threadId: THREAD_ID, + provider: ProviderDriverKind.make("claudeAgent"), + runtimeMode: "full-access", + }); + + // Before any probe names the bucket the event has nowhere to land. + // Collecting through the turn's completion proves the SDK message was + // handled, not merely still queued. + const firstTurnFiber = yield* adapter.streamEvents.pipe( + Stream.takeUntil((event) => event.type === "turn.completed"), + Stream.runCollect, + Effect.forkChild, + ); + yield* adapter.sendTurn({ threadId: session.threadId, input: "hello", attachments: [] }); + harness.query.emit(rateLimitEvent(0.2)); + harness.query.emit(resultMessage("result-1")); + assert.deepStrictEqual(limitsUpdates(yield* Fiber.join(firstTurnFiber)), []); + + // The status probe reads `get_usage` and records the model it saw. + yield* Ref.set(scopedLimitNames, { overageIncluded: "Fable" }); + const secondTurnFiber = yield* adapter.streamEvents.pipe( + Stream.takeUntil((event) => event.type === "turn.completed"), + Stream.runCollect, + Effect.forkChild, + ); + yield* adapter.sendTurn({ threadId: session.threadId, input: "again", attachments: [] }); + harness.query.emit(rateLimitEvent(0.4)); + harness.query.emit(resultMessage("result-2")); + assert.deepStrictEqual(limitsUpdates(yield* Fiber.join(secondTurnFiber)), [ + { + windows: [ + { + id: "seven_day_fable", + kind: "weekly", + label: "Weekly · Fable", + usedPercent: 40, + windowDurationMins: 10_080, + }, + ], + }, + ]); + }).pipe( + Effect.provideService(Random.Random, makeDeterministicRandomService()), + Effect.provide(harness.layer), + ); + }); + it.effect("does not emit turn.completed for a result with no active turn", () => { const harness = makeHarness(); return Effect.gen(function* () { diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.ts b/apps/server/src/provider/Layers/ClaudeAdapter.ts index 62763f947c7d..24a7fb28fd6a 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.ts @@ -21,6 +21,7 @@ import { } from "@anthropic-ai/claude-agent-sdk"; import { parseCliArgs } from "@t3tools/shared/cliArgs"; import { isWorkspaceImagePreviewPath } from "@t3tools/shared/filePreview"; +import { type ClaudeScopedLimitNames, claudeRateLimitEventToUpdate } from "./claudeUsageLimits.ts"; import { ApprovalRequestId, type CanonicalItemType, @@ -342,6 +343,8 @@ export interface ClaudeAdapterLiveOptions { readonly nativeEventLogPath?: string; readonly nativeEventLogger?: EventNdjsonLogger; readonly modelCatalog?: Effect.Effect; + /** Scoped-bucket names the driver's status probe last saw; see `claudeUsageLimits`. */ + readonly scopedLimitNames?: Ref.Ref; } function isUuid(value: string): boolean { @@ -3595,12 +3598,15 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( } if (message.type === "rate_limit_event") { + const names = options?.scopedLimitNames + ? yield* Ref.get(options.scopedLimitNames) + : { overageIncluded: undefined }; + const limits = claudeRateLimitEventToUpdate(message.rate_limit_info, names); + if (!limits) return; yield* offerRuntimeEvent({ ...base, type: "account.rate-limits.updated", - payload: { - rateLimits: message, - }, + payload: { limits }, }); return; } diff --git a/apps/server/src/provider/Layers/ClaudeCapabilitiesProbe.test.ts b/apps/server/src/provider/Layers/ClaudeCapabilitiesProbe.test.ts index 2f842bf581f7..253118299820 100644 --- a/apps/server/src/provider/Layers/ClaudeCapabilitiesProbe.test.ts +++ b/apps/server/src/provider/Layers/ClaudeCapabilitiesProbe.test.ts @@ -77,22 +77,31 @@ it.layer(NodeServices.layer)("Claude capability probe SDK boundary", (it) => { "const lines = createInterface({ input: process.stdin });", 'lines.on("line", (line) => {', " const message = JSON.parse(line);", - ' if (message.type !== "control_request" || message.request?.subtype !== "initialize") return;', - " process.stdout.write(JSON.stringify({", + ' if (message.type !== "control_request") return;', + " const reply = (response) => process.stdout.write(JSON.stringify({", ' type: "control_response",', - " response: {", - ' subtype: "success",', - " request_id: message.request_id,", - " response: {", - ' commands: [{ name: "review", description: "Review changes", argumentHint: "[path]" }],', - " agents: [],", - ' output_style: "default",', - ' available_output_styles: ["default"],', - " models: [],", - ' account: { email: "dev@example.com", subscriptionType: "pro", tokenSource: "oauth" },', - " },", - " },", + ' response: { subtype: "success", request_id: message.request_id, response },', ' }) + "\\n");', + ' if (message.request?.subtype === "initialize") {', + " reply({", + ' commands: [{ name: "review", description: "Review changes", argumentHint: "[path]" }],', + " agents: [],", + ' output_style: "default",', + ' available_output_styles: ["default"],', + " models: [],", + ' account: { email: "dev@example.com", subscriptionType: "pro", tokenSource: "oauth" },', + " });", + " }", + " // The probe follows initialize with get_usage on the same process.", + ' if (message.request?.subtype === "get_usage") {', + " reply({", + " session: {},", + ' subscription_type: "pro",', + " rate_limits_available: true,", + ' rate_limits: { five_hour: { utilization: 12, resets_at: "2026-07-18T14:39:00Z" } },', + " behaviors: null,", + " });", + " }", "});", "setInterval(() => {}, 1_000);", "", @@ -122,6 +131,10 @@ it.layer(NodeServices.layer)("Claude capability probe SDK boundary", (it) => { input: { hint: "[path]" }, }, ], + usage: { + rate_limits_available: true, + rate_limits: { five_hour: { utilization: 12, resets_at: "2026-07-18T14:39:00Z" } }, + }, }); // @effect-diagnostics-next-line preferSchemaOverJson:off diff --git a/apps/server/src/provider/Layers/ClaudeProvider.ts b/apps/server/src/provider/Layers/ClaudeProvider.ts index bf41046f61e8..bb60327ada0e 100644 --- a/apps/server/src/provider/Layers/ClaudeProvider.ts +++ b/apps/server/src/provider/Layers/ClaudeProvider.ts @@ -8,6 +8,7 @@ import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; import * as Option from "effect/Option"; import * as Path from "effect/Path"; +import * as Ref from "effect/Ref"; import * as Result from "effect/Result"; import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"; import { createModelCapabilities } from "@t3tools/shared/model"; @@ -16,6 +17,7 @@ import { query as claudeQuery, type Options as ClaudeQueryOptions, type SlashCommand as ClaudeSlashCommand, + type SDKControlGetUsageResponse, type SDKUserMessage, type SettingSource, } from "@anthropic-ai/claude-agent-sdk"; @@ -32,6 +34,12 @@ import { import { resolveClaudeSdkExecutablePath } from "../Drivers/ClaudeExecutable.ts"; import { makeClaudeEnvironment } from "../Drivers/ClaudeHome.ts"; import { discoverClaudeSkills } from "../Drivers/ClaudeSkills.ts"; +import { makeUnavailableUsageLimits } from "../providerUsageLimits.ts"; +import { + type ClaudeScopedLimitNames, + claudeUsageResponseToLimits, + recordClaudeUsageResponse, +} from "./claudeUsageLimits.ts"; import { BUNDLED_CLAUDE_MODEL_CATALOG, type ClaudeModelCatalog, @@ -225,6 +233,12 @@ type ClaudeCapabilitiesProbe = { */ readonly apiProvider: string | undefined; readonly slashCommands: ReadonlyArray; + /** + * Subscription windows from the SDK's `get_usage` control request, or + * `undefined` when the request itself failed. Absent windows on an + * otherwise successful response mean the account has none (API key). + */ + readonly usage?: Pick; }; function parseClaudeInitializationCommands( @@ -340,6 +354,15 @@ const probeClaudeCapabilities = ( }), }); const init = await q.initializationResult(); + // Usage is a second control round trip on the same process; a failure + // there must not cost the slash commands and account we already have. + const usage = await q.usage_EXPERIMENTAL_MAY_CHANGE_DO_NOT_RELY_ON_THIS_API_YET().then( + (response) => ({ + rate_limits_available: response.rate_limits_available, + rate_limits: response.rate_limits, + }), + () => undefined, + ); const account = init.account as | { readonly email?: string; @@ -354,6 +377,7 @@ const probeClaudeCapabilities = ( tokenSource: account?.tokenSource, apiProvider: account?.apiProvider, slashCommands: parseClaudeInitializationCommands(init.commands), + ...(usage ? { usage } : {}), } satisfies ClaudeCapabilitiesProbe; }); }).pipe( @@ -395,6 +419,8 @@ export const checkClaudeProviderStatus = Effect.fn("checkClaudeProviderStatus")( environment?: NodeJS.ProcessEnv, cwd?: string, modelCatalog: ClaudeModelCatalog = BUNDLED_CLAUDE_MODEL_CATALOG, + /** Shared with the adapter so turn events reuse the scoped-bucket names this probe saw. */ + scopedLimitNames?: Ref.Ref, ): Effect.fn.Return< ServerProviderDraft, never, @@ -535,6 +561,14 @@ export const checkClaudeProviderStatus = Effect.fn("checkClaudeProviderStatus")( subscriptionType: capabilities.subscriptionType, authMethod: capabilities.tokenSource, }) ?? apiProviderAuthMetadata(capabilities.apiProvider); + const usageLimits = !capabilities.usage + ? makeUnavailableUsageLimits({ checkedAt, reason: "probeFailed" }) + : scopedLimitNames + ? yield* recordClaudeUsageResponse(scopedLimitNames, { + response: capabilities.usage, + checkedAt, + }) + : claudeUsageResponseToLimits({ response: capabilities.usage, checkedAt }).limits; return buildServerProvider({ presentation: CLAUDE_PRESENTATION, enabled: claudeSettings.enabled, @@ -552,6 +586,7 @@ export const checkClaudeProviderStatus = Effect.fn("checkClaudeProviderStatus")( ...(authMetadata ? authMetadata : {}), }, ...(versionUpgradeMessage ? { message: versionUpgradeMessage } : {}), + usageLimits, }, }); }); diff --git a/apps/server/src/provider/Layers/CodexAdapter.ts b/apps/server/src/provider/Layers/CodexAdapter.ts index fa8511ee09a1..c66f2457c9fd 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.ts @@ -69,6 +69,7 @@ import { } from "./CodexSessionRuntime.ts"; import { type EventNdjsonLogger, makeEventNdjsonLogger } from "./EventNdjsonLogger.ts"; import { resolveCodexLaunchArgs } from "./codexLaunchArgs.ts"; +import { codexRateLimitsToUpdate } from "./codexUsageLimits.ts"; const isCodexAppServerProcessExitedError = Schema.is(CodexErrors.CodexAppServerProcessExitedError); const isCodexAppServerTransportError = Schema.is(CodexErrors.CodexAppServerTransportError); const isCodexSessionRuntimeThreadIdMissingError = Schema.is( @@ -1725,16 +1726,19 @@ function mapToRuntimeEvents( } if (event.method === "account/rateLimits/updated") { - if (!readPayload(EffectCodexSchema.V2AccountRateLimitsUpdatedNotification, event.payload)) { + const payload = readPayload( + EffectCodexSchema.V2AccountRateLimitsUpdatedNotification, + event.payload, + ); + const limits = payload ? codexRateLimitsToUpdate(payload.rateLimits) : undefined; + if (!limits) { return []; } return [ { type: "account.rate-limits.updated", ...runtimeEventBase(event, canonicalThreadId), - payload: { - rateLimits: event.payload ?? {}, - }, + payload: { limits }, }, ]; } diff --git a/apps/server/src/provider/Layers/CodexProvider.ts b/apps/server/src/provider/Layers/CodexProvider.ts index 4d7efbe2106b..9ffd6ccff9ec 100644 --- a/apps/server/src/provider/Layers/CodexProvider.ts +++ b/apps/server/src/provider/Layers/CodexProvider.ts @@ -33,8 +33,19 @@ import { type ServerProviderDraft, } from "../providerSnapshot.ts"; import { expandHomePath } from "../../pathExpansion.ts"; +import { makeUnavailableUsageLimits } from "../providerUsageLimits.ts"; +import { + codexRateLimitsFailureMessage, + codexRateLimitsToLimits, + type CodexRateLimitSnapshot, +} from "./codexUsageLimits.ts"; import packageJson from "../../../package.json" with { type: "json" }; const isCodexAppServerSpawnError = Schema.is(CodexErrors.CodexAppServerSpawnError); +const RATE_LIMITS_PROBE_TIMEOUT_MS = 3_000; + +type CodexRateLimitsProbe = + | { readonly snapshot: CodexRateLimitSnapshot } + | { readonly failure: string }; const CODEX_APP_SERVER_PROBE_FORCE_KILL_AFTER = "2 seconds" as const; @@ -45,6 +56,7 @@ const CODEX_PRESENTATION = { export interface CodexAppServerProviderSnapshot { readonly account: CodexSchema.V2GetAccountResponse; + readonly rateLimits?: CodexRateLimitsProbe; readonly version: string | undefined; readonly models: ReadonlyArray; readonly skills: ReadonlyArray; @@ -72,8 +84,12 @@ function codexAccountAuthLabel(account: CodexSchema.V2GetAccountResponse["accoun if (account.type === "apiKey") return "OpenAI API Key"; if (account.type === "amazonBedrock") return "Amazon Bedrock"; if (account.type !== "chatgpt") return undefined; + return codexPlanLabel(account.planType); +} - switch (account.planType) { +/** Shared with usage-limit sources, which report the same `planType` slugs. */ +export function codexPlanLabel(planType: string | null | undefined): string | undefined { + switch (planType) { case "free": return "ChatGPT Free Subscription"; case "go": @@ -102,7 +118,6 @@ function codexAccountAuthLabel(account: CodexSchema.V2GetAccountResponse["accoun case "unknown": return "ChatGPT Subscription"; default: - account.planType satisfies never; return undefined; } } @@ -394,18 +409,35 @@ const probeCodexAppServerProvider = Effect.fn("probeCodexAppServerProvider")(fun } satisfies CodexAppServerProviderSnapshot; } - const [skillsResponse, models] = yield* Effect.all( + const [skillsResponse, models, rateLimits] = yield* Effect.all( [ client.request("skills/list", { cwds: [input.cwd], }), requestAllCodexModels(client), + // Usage is an enrichment: a failure or a slow answer degrades to "no + // usage this probe" rather than costing the account and models. + client.request("account/rateLimits/read", undefined).pipe( + Effect.map((response): CodexRateLimitsProbe => ({ snapshot: response.rateLimits })), + Effect.timeoutOption(Duration.millis(RATE_LIMITS_PROBE_TIMEOUT_MS)), + Effect.map( + Option.getOrElse((): CodexRateLimitsProbe => ({ + failure: "Codex did not answer the usage request.", + })), + ), + Effect.catch((error) => + Effect.logDebug("Codex rate-limit read failed.", { cause: error }).pipe( + Effect.as({ failure: codexRateLimitsFailureMessage(error) }), + ), + ), + ), ], { concurrency: "unbounded" }, ); return { account: accountResponse, + rateLimits, version, models: applyPreferredCodexDefaultModel( appendCustomCodexModels(models, input.customModels ?? []), @@ -640,6 +672,16 @@ export const checkCodexProviderStatus = Effect.fn("checkCodexProviderStatus")(fu const snapshot = probeResult.success.value; const accountStatus = accountProbeStatus(snapshot.account); + const usageLimits = + snapshot.account.account?.type === "apiKey" + ? makeUnavailableUsageLimits({ checkedAt, reason: "unsupported" }) + : snapshot.rateLimits === undefined || "failure" in snapshot.rateLimits + ? makeUnavailableUsageLimits({ + checkedAt, + reason: "probeFailed", + ...(snapshot.rateLimits ? { message: snapshot.rateLimits.failure } : {}), + }) + : codexRateLimitsToLimits({ snapshot: snapshot.rateLimits.snapshot, checkedAt }); return buildServerProvider({ presentation: CODEX_PRESENTATION, @@ -660,6 +702,7 @@ export const checkCodexProviderStatus = Effect.fn("checkCodexProviderStatus")(fu status: accountStatus.status, auth: accountStatus.auth, ...(accountStatus.message ? { message: accountStatus.message } : {}), + usageLimits, }, }); }); diff --git a/apps/server/src/provider/Layers/ProviderAdapterRegistry.test.ts b/apps/server/src/provider/Layers/ProviderAdapterRegistry.test.ts index 2336184adaaa..18d94ebd0e18 100644 --- a/apps/server/src/provider/Layers/ProviderAdapterRegistry.test.ts +++ b/apps/server/src/provider/Layers/ProviderAdapterRegistry.test.ts @@ -118,6 +118,7 @@ const makeFakeInstance = ( getSnapshot: Effect.succeed({} as unknown as ServerProvider), refresh: Effect.succeed({} as unknown as ServerProvider), streamChanges: Stream.empty, + applyUsageLimits: () => Effect.void, }, adapter, textGeneration: {} as unknown as TextGeneration.TextGeneration["Service"], diff --git a/apps/server/src/provider/Layers/ProviderRegistry.test.ts b/apps/server/src/provider/Layers/ProviderRegistry.test.ts index 6942d7f9dd52..6816df464fb8 100644 --- a/apps/server/src/provider/Layers/ProviderRegistry.test.ts +++ b/apps/server/src/provider/Layers/ProviderRegistry.test.ts @@ -1098,6 +1098,7 @@ it.layer(Layer.mergeAll(NodeServices.layer, ServerSettingsModule.layerTest(), Te Effect.andThen(Effect.never), ), streamChanges: Stream.empty, + applyUsageLimits: () => Effect.void, }, adapter: {} as ProviderInstance["adapter"], textGeneration: {} as ProviderInstance["textGeneration"], @@ -1187,6 +1188,7 @@ it.layer(Layer.mergeAll(NodeServices.layer, ServerSettingsModule.layerTest(), Te getSnapshot: Effect.succeed(provider), refresh: Effect.succeed(provider), streamChanges: Stream.empty, + applyUsageLimits: () => Effect.void, }, snapshotForCwd, adapter: {} as ProviderInstance["adapter"], @@ -1377,6 +1379,7 @@ it.layer(Layer.mergeAll(NodeServices.layer, ServerSettingsModule.layerTest(), Te Effect.as(codexProvider), ), streamChanges: Stream.empty, + applyUsageLimits: () => Effect.void, }, adapter: {} as ProviderInstance["adapter"], textGeneration: {} as ProviderInstance["textGeneration"], @@ -1400,6 +1403,7 @@ it.layer(Layer.mergeAll(NodeServices.layer, ServerSettingsModule.layerTest(), Te Effect.andThen(Ref.get(catalogSnapshot)), ), streamChanges: Stream.empty, + applyUsageLimits: () => Effect.void, }, adapter: {} as ProviderInstance["adapter"], textGeneration: {} as ProviderInstance["textGeneration"], @@ -1521,6 +1525,7 @@ it.layer(Layer.mergeAll(NodeServices.layer, ServerSettingsModule.layerTest(), Te getSnapshot: Effect.succeed(initialProvider), refresh: Effect.succeed(refreshedProvider), streamChanges: Stream.fromPubSub(changes), + applyUsageLimits: () => Effect.void, }, adapter: {} as ProviderInstance["adapter"], textGeneration: {} as ProviderInstance["textGeneration"], @@ -1650,6 +1655,7 @@ it.layer(Layer.mergeAll(NodeServices.layer, ServerSettingsModule.layerTest(), Te getSnapshot: Effect.succeed(initialProvider), refresh: Effect.succeed(authoritativeProvider), streamChanges: Stream.fromPubSub(changes), + applyUsageLimits: () => Effect.void, }, adapter: {} as ProviderInstance["adapter"], textGeneration: {} as ProviderInstance["textGeneration"], @@ -1757,6 +1763,7 @@ it.layer(Layer.mergeAll(NodeServices.layer, ServerSettingsModule.layerTest(), Te getSnapshot: Effect.succeed(cachedProvider), refresh: Effect.die(new Error("simulated refresh failure")), streamChanges: Stream.empty, + applyUsageLimits: () => Effect.void, }, adapter: {} as ProviderInstance["adapter"], textGeneration: {} as ProviderInstance["textGeneration"], @@ -1850,6 +1857,7 @@ it.layer(Layer.mergeAll(NodeServices.layer, ServerSettingsModule.layerTest(), Te getSnapshot: Effect.succeed(provider), refresh: Effect.succeed(provider), streamChanges: Stream.empty, + applyUsageLimits: () => Effect.void, }, adapter: {} as ProviderInstance["adapter"], textGeneration: {} as ProviderInstance["textGeneration"], diff --git a/apps/server/src/provider/Layers/ProviderUsageLimitsIngestion.ts b/apps/server/src/provider/Layers/ProviderUsageLimitsIngestion.ts new file mode 100644 index 000000000000..fd63ff8a8c68 --- /dev/null +++ b/apps/server/src/provider/Layers/ProviderUsageLimitsIngestion.ts @@ -0,0 +1,44 @@ +/** + * ProviderUsageLimitsIngestionLive — folds `account.rate-limits.updated` + * runtime events into the owning instance's published snapshot. + * + * Adapters normalise their native payloads before emitting, so this layer + * never sees a driver shape: it routes the typed update to the instance and + * lets `ServerProviderShape.applyUsageLimits` merge and republish on the + * instance's own change stream, which `ProviderRegistry` already aggregates. + * + * @module provider/Layers/ProviderUsageLimitsIngestion + */ +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Stream from "effect/Stream"; + +import { ProviderInstanceRegistry } from "../Services/ProviderInstanceRegistry.ts"; +import { ProviderService } from "../Services/ProviderService.ts"; + +export const ProviderUsageLimitsIngestionLive = Layer.effectDiscard( + Effect.gen(function* () { + const providerService = yield* ProviderService; + const instanceRegistry = yield* ProviderInstanceRegistry; + + yield* providerService.streamEvents.pipe( + Stream.filter((event) => event.type === "account.rate-limits.updated"), + Stream.runForEach((event) => + Effect.gen(function* () { + if (!event.providerInstanceId) { + return; + } + const instance = yield* instanceRegistry.getInstance(event.providerInstanceId); + if (!instance) { + return; + } + const checkedAt = DateTime.formatIso(yield* DateTime.now); + yield* instance.snapshot.applyUsageLimits({ ...event.payload.limits, checkedAt }); + // One bad event must not end the subscriber for every later one. + }).pipe(Effect.ignoreCause({ log: true })), + ), + Effect.forkScoped, + ); + }), +); diff --git a/apps/server/src/provider/Layers/claudeUsageLimits.test.ts b/apps/server/src/provider/Layers/claudeUsageLimits.test.ts new file mode 100644 index 000000000000..a3d326a7dcb2 --- /dev/null +++ b/apps/server/src/provider/Layers/claudeUsageLimits.test.ts @@ -0,0 +1,180 @@ +import { describe, expect, it } from "vite-plus/test"; + +import { claudeRateLimitEventToUpdate, claudeUsageResponseToLimits } from "./claudeUsageLimits.ts"; + +const checkedAt = "2026-07-18T10:00:00.000Z"; +const noNames = { overageIncluded: undefined } as const; + +describe("claudeUsageResponseToLimits", () => { + it("maps the session, weekly, and model-scoped weekly windows", () => { + expect( + claudeUsageResponseToLimits({ + checkedAt, + response: { + rate_limits_available: true, + rate_limits: { + five_hour: { utilization: 54, resets_at: "2026-07-18T14:39:00Z" }, + seven_day: { utilization: 18.4, resets_at: "2026-07-24T08:59:00+00:00" }, + seven_day_opus: { utilization: 3, resets_at: null }, + // Newer CLIs add this on top of the typed keys; the pinned SDK + // typings do not know it yet. + ...({ + model_scoped: [ + { display_name: "Fable", utilization: 73, resets_at: "2026-07-24T08:59:00Z" }, + { display_name: "Ghost", utilization: null, resets_at: null }, + ], + } as object), + extra_usage: { + is_enabled: false, + monthly_limit: null, + used_credits: null, + utilization: null, + }, + }, + }, + }), + ).toEqual({ + names: { overageIncluded: "Fable" }, + limits: { + checkedAt, + windows: [ + { + id: "five_hour", + kind: "session", + label: "Session", + usedPercent: 54, + windowDurationMins: 300, + resetsAt: "2026-07-18T14:39:00.000Z", + }, + { + id: "seven_day", + kind: "weekly", + label: "Weekly", + usedPercent: 18.4, + windowDurationMins: 10080, + resetsAt: "2026-07-24T08:59:00.000Z", + }, + { + id: "seven_day_fable", + kind: "weekly", + label: "Weekly · Fable", + usedPercent: 73, + windowDurationMins: 10080, + resetsAt: "2026-07-24T08:59:00.000Z", + }, + ], + }, + }); + }); + + it("names the overage-included bucket only from a scoped entry that drew a row", () => { + expect( + claudeUsageResponseToLimits({ + checkedAt, + response: { + rate_limits_available: true, + rate_limits: { + ...({ + model_scoped: [ + { display_name: "Ghost", utilization: null, resets_at: null }, + { display_name: "Fable", utilization: 5, resets_at: null }, + ], + } as object), + }, + }, + }).names, + ).toEqual({ overageIncluded: "Fable" }); + }); + + it("reports API key and Bedrock accounts as unsupported", () => { + expect( + claudeUsageResponseToLimits({ + checkedAt, + response: { rate_limits_available: false, rate_limits: null }, + }).limits, + ).toEqual({ checkedAt, windows: [], unavailable: { reason: "unsupported" } }); + }); + + it("skips a window the endpoint reports without a utilization", () => { + expect( + claudeUsageResponseToLimits({ + checkedAt, + response: { + rate_limits_available: true, + rate_limits: { + five_hour: { utilization: null, resets_at: null }, + seven_day: { utilization: 250, resets_at: null }, + }, + }, + }).limits.windows, + ).toEqual([ + { + id: "seven_day", + kind: "weekly", + label: "Weekly", + usedPercent: 100, + windowDurationMins: 10080, + }, + ]); + }); +}); + +describe("claudeRateLimitEventToUpdate", () => { + it("scales the 0–1 utilization and epoch-second reset onto the probe's window id", () => { + expect( + claudeRateLimitEventToUpdate( + { + status: "allowed_warning", + rateLimitType: "seven_day", + utilization: 0.85, + resetsAt: 1_784_000_000, + }, + noNames, + ), + ).toEqual({ + windows: [ + { + id: "seven_day", + kind: "weekly", + label: "Weekly", + usedPercent: 85, + windowDurationMins: 10080, + resetsAt: "2026-07-14T03:33:20.000Z", + }, + ], + }); + }); + + it("lands the streamed overage-included bucket on the row the probe named", () => { + const event = { + status: "allowed", + rateLimitType: "seven_day_overage_included" as never, + utilization: 0.4, + } as const; + // No probe has named the bucket yet: guessing would open a stray row. + expect(claudeRateLimitEventToUpdate(event, noNames)).toBeUndefined(); + expect(claudeRateLimitEventToUpdate(event, { overageIncluded: "Fable" })).toEqual({ + windows: [ + { + id: "seven_day_fable", + kind: "weekly", + label: "Weekly · Fable", + usedPercent: 40, + windowDurationMins: 10080, + }, + ], + }); + }); + + it("ignores windows the page does not render and events without a utilization", () => { + expect( + claudeRateLimitEventToUpdate( + { status: "allowed", rateLimitType: "seven_day_opus", utilization: 0.1 }, + noNames, + ), + ).toBeUndefined(); + expect( + claudeRateLimitEventToUpdate({ status: "rejected", rateLimitType: "five_hour" }, noNames), + ).toBeUndefined(); + }); +}); diff --git a/apps/server/src/provider/Layers/claudeUsageLimits.ts b/apps/server/src/provider/Layers/claudeUsageLimits.ts new file mode 100644 index 000000000000..b67645102e5a --- /dev/null +++ b/apps/server/src/provider/Layers/claudeUsageLimits.ts @@ -0,0 +1,199 @@ +/** + * Claude Code subscription usage. Both sources produce windows with the same + * ids so a turn-driven `rate_limit_event` lands on the row the SDK's + * `get_usage` read established: + * + * - `get_usage` (on demand, during the capabilities probe) reports every + * window at once as 0–100 percentages with ISO reset times. + * - `rate_limit_event` (streamed during a turn) names one window at a time + * with a 0–1 utilization fraction and an epoch-seconds reset. + * + * @module provider/Layers/claudeUsageLimits + */ +import type { SDKControlGetUsageResponse, SDKRateLimitInfo } from "@anthropic-ai/claude-agent-sdk"; +import type { + ProviderUsageLimitsUpdate, + ServerProviderUsageLimits, + ServerProviderUsageWindow, +} from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Option from "effect/Option"; +import * as Ref from "effect/Ref"; + +import { + clampPercent, + makeUnavailableUsageLimits, + makeUsageLimits, +} from "../providerUsageLimits.ts"; + +const SESSION_MINS = 5 * 60; +const WEEK_MINS = 7 * 24 * 60; + +/** + * The account-wide windows, keyed by the SDK's `rateLimitType`. Model-scoped + * weeklies are additive on top of these: the CLI reports them under + * `rate_limits.model_scoped[]` on `get_usage` and streams the overage-included + * model bucket (Fable today) as `seven_day_overage_included`. + */ +const WINDOWS: Readonly< + Record> +> = { + five_hour: { kind: "session", label: "Session", windowDurationMins: SESSION_MINS }, + seven_day: { kind: "weekly", label: "Weekly", windowDurationMins: WEEK_MINS }, +}; + +/** + * The streamed event names the overage-included bucket by type + * (`seven_day_overage_included`), while `get_usage` names it by the model's + * `display_name`. Which model that is changes over time, so the probe records + * the name it saw and the event mapper reuses it; the mid-turn update then + * lands on the row the probe drew instead of opening a second one. + */ +const OVERAGE_INCLUDED_EVENT_TYPE = "seven_day_overage_included"; + +export interface ClaudeScopedLimitNames { + readonly overageIncluded: string | undefined; +} + +export const makeClaudeScopedLimitNames = Ref.make({ + overageIncluded: undefined, +}); + +function scopedWindowId(displayName: string): string { + return `seven_day_${displayName.toLowerCase().replace(/[^a-z0-9]+/g, "_")}`; +} + +function scopedWindow( + displayName: string, + usedPercent: number, + resetsAt: string | undefined, +): ServerProviderUsageWindow { + return { + id: scopedWindowId(displayName), + kind: "weekly", + label: `Weekly · ${displayName}`, + windowDurationMins: WEEK_MINS, + usedPercent: clampPercent(usedPercent), + ...(resetsAt ? { resetsAt } : {}), + }; +} + +/** + * `model_scoped` shipped in the CLI after the SDK typings we pin, so it is + * read structurally until the `.d.ts` catches up. + */ +interface ModelScopedWindow { + readonly display_name: string; + readonly utilization: number | null; + readonly resets_at: string | null; +} + +function readModelScoped(rateLimits: object): ReadonlyArray { + const raw = (rateLimits as { readonly model_scoped?: unknown }).model_scoped; + if (!Array.isArray(raw)) return []; + return raw.filter( + (entry): entry is ModelScopedWindow => + typeof entry === "object" && + entry !== null && + typeof (entry as ModelScopedWindow).display_name === "string", + ); +} + +function isoFromEpochSeconds(value: number | undefined): string | undefined { + if (value === undefined || !Number.isFinite(value) || value <= 0) return undefined; + const dt = DateTime.make(value * 1000); + return Option.isSome(dt) ? DateTime.formatIso(dt.value) : undefined; +} + +function isoFromString(value: string | null | undefined): string | undefined { + if (!value) return undefined; + const dt = DateTime.make(value); + return Option.isSome(dt) ? DateTime.formatIso(dt.value) : undefined; +} + +function makeWindow( + id: keyof typeof WINDOWS & string, + usedPercent: number, + resetsAt: string | undefined, +): ServerProviderUsageWindow { + const window = WINDOWS[id]!; + return { + id, + ...window, + usedPercent: clampPercent(usedPercent), + ...(resetsAt ? { resetsAt } : {}), + }; +} + +/** + * Utilization is a 0–1 fraction on the streamed event. An overage-included + * event before any probe has named the bucket is dropped: guessing a name + * would draw a row the next probe cannot reconcile. + */ +export function claudeRateLimitEventToUpdate( + info: SDKRateLimitInfo, + names: ClaudeScopedLimitNames, +): ProviderUsageLimitsUpdate | undefined { + const type: string | undefined = info.rateLimitType; + if (!type || typeof info.utilization !== "number") { + return undefined; + } + const usedPercent = info.utilization * 100; + const resetsAt = isoFromEpochSeconds(info.resetsAt); + if (type in WINDOWS) { + return { windows: [makeWindow(type, usedPercent, resetsAt)] }; + } + if (type === OVERAGE_INCLUDED_EVENT_TYPE && names.overageIncluded) { + return { windows: [scopedWindow(names.overageIncluded, usedPercent, resetsAt)] }; + } + return undefined; +} + +/** + * Percentages on the `get_usage` response are already 0–100. Also yields the + * scoped-bucket names the response carried, for the event mapper to reuse. + */ +export function claudeUsageResponseToLimits(input: { + readonly response: Pick; + readonly checkedAt: string; +}): { readonly limits: ServerProviderUsageLimits; readonly names: ClaudeScopedLimitNames } { + const { response, checkedAt } = input; + if (!response.rate_limits_available || !response.rate_limits) { + return { + limits: makeUnavailableUsageLimits({ checkedAt, reason: "unsupported" }), + names: { overageIncluded: undefined }, + }; + } + const windows: ServerProviderUsageWindow[] = []; + for (const id of Object.keys(WINDOWS)) { + const window = response.rate_limits[id as "five_hour" | "seven_day"]; + if (!window || typeof window.utilization !== "number") continue; + windows.push(makeWindow(id, window.utilization, isoFromString(window.resets_at))); + } + // The CLI filters `model_scoped` to the overage-included allowlist, which + // today holds one model; the first entry is the one the event refers to. + let overageIncluded: string | undefined; + for (const entry of readModelScoped(response.rate_limits)) { + if (typeof entry.utilization !== "number") continue; + windows.push( + scopedWindow(entry.display_name, entry.utilization, isoFromString(entry.resets_at)), + ); + // Only a bucket that drew a row may receive events; naming one that was + // skipped would let a mid-turn event open a row the probe never showed. + overageIncluded ??= entry.display_name; + } + return { + limits: makeUsageLimits({ checkedAt, windows }), + names: { overageIncluded }, + }; +} + +/** Probe-side helper: map the response and remember the scoped names for events. */ +export const recordClaudeUsageResponse = ( + namesRef: Ref.Ref, + input: Parameters[0], +): Effect.Effect => { + const { limits, names } = claudeUsageResponseToLimits(input); + return Ref.set(namesRef, names).pipe(Effect.as(limits)); +}; diff --git a/apps/server/src/provider/Layers/codexUsageLimits.test.ts b/apps/server/src/provider/Layers/codexUsageLimits.test.ts new file mode 100644 index 000000000000..62cfa0552aab --- /dev/null +++ b/apps/server/src/provider/Layers/codexUsageLimits.test.ts @@ -0,0 +1,103 @@ +import * as CodexErrors from "effect-codex-app-server/errors"; +import { describe, expect, it } from "vite-plus/test"; + +import { + codexRateLimitsFailureMessage, + codexRateLimitsToLimits, + codexRateLimitsToUpdate, +} from "./codexUsageLimits.ts"; + +const checkedAt = "2026-07-18T10:00:00.000Z"; + +describe("codexRateLimitsToLimits", () => { + it("maps primary and secondary onto the session and weekly windows", () => { + expect( + codexRateLimitsToLimits({ + checkedAt, + snapshot: { + planType: "plus", + primary: { usedPercent: 12, resetsAt: 1_784_000_000, windowDurationMins: 300 }, + secondary: { usedPercent: 47, resetsAt: 1_784_500_000, windowDurationMins: 10080 }, + }, + }), + ).toEqual({ + checkedAt, + windows: [ + { + id: "primary", + kind: "session", + label: "Session", + usedPercent: 12, + windowDurationMins: 300, + resetsAt: "2026-07-14T03:33:20.000Z", + }, + { + id: "secondary", + kind: "weekly", + label: "Weekly", + usedPercent: 47, + windowDurationMins: 10080, + resetsAt: "2026-07-19T22:26:40.000Z", + }, + ], + }); + }); + + it("treats a lone duration-less primary as monthly on Free and Go", () => { + expect( + codexRateLimitsToLimits({ + checkedAt, + snapshot: { planType: "free", primary: { usedPercent: 80, resetsAt: null } }, + }).windows, + ).toEqual([ + { + id: "primary", + kind: "monthly", + label: "Monthly", + usedPercent: 80, + windowDurationMins: 43_200, + }, + ]); + }); +}); + +describe("codexRateLimitsToUpdate", () => { + it("carries only the windows the notification names", () => { + expect( + codexRateLimitsToUpdate({ + secondary: { usedPercent: 51, windowDurationMins: 10080 }, + }), + ).toEqual({ + windows: [ + { + id: "secondary", + kind: "weekly", + label: "Weekly", + usedPercent: 51, + windowDurationMins: 10080, + }, + ], + }); + expect(codexRateLimitsToUpdate({ planType: "plus" })).toBeUndefined(); + }); +}); + +describe("codexRateLimitsFailureMessage", () => { + it("keeps the JSON-RPC code and nothing else from a request failure", () => { + expect( + codexRateLimitsFailureMessage( + new CodexErrors.CodexAppServerRequestError({ + code: -32603, + errorMessage: + "failed to fetch codex rate limits: GET https://chatgpt.com/backend-api/wham/usage failed: 401 Unauthorized", + }), + ), + ).toBe("Codex could not read usage (JSON-RPC -32603)."); + }); + + it("phrases a dead process differently from a bad answer", () => { + expect( + codexRateLimitsFailureMessage(new CodexErrors.CodexAppServerProcessExitedError({ code: 1 })), + ).toBe("Codex exited before it could report usage."); + }); +}); diff --git a/apps/server/src/provider/Layers/codexUsageLimits.ts b/apps/server/src/provider/Layers/codexUsageLimits.ts new file mode 100644 index 000000000000..1b6148f0da82 --- /dev/null +++ b/apps/server/src/provider/Layers/codexUsageLimits.ts @@ -0,0 +1,118 @@ +/** + * Codex subscription usage. The `account/rateLimits/read` response and the + * `account/rateLimits/updated` notification carry the same snapshot shape, so + * one mapper serves the status probe and the turn-driven update; both emit + * windows with the same ids so they merge onto the same rows. + * + * @module provider/Layers/codexUsageLimits + */ +import type { + ProviderUsageLimitsUpdate, + ServerProviderUsageLimits, + ServerProviderUsageWindow, +} from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +import * as Option from "effect/Option"; +import type * as CodexErrors from "effect-codex-app-server/errors"; + +import { clampPercent, makeUsageLimits } from "../providerUsageLimits.ts"; + +interface CodexRateLimitWindow { + readonly usedPercent: number; + readonly resetsAt?: number | null; + readonly windowDurationMins?: number | null; +} + +/** Structural view of the generated `RateLimitSnapshot`; both messages satisfy it. */ +export interface CodexRateLimitSnapshot { + readonly planType?: string | null; + readonly primary?: CodexRateLimitWindow | null; + readonly secondary?: CodexRateLimitWindow | null; +} + +const SESSION_MINS = 5 * 60; +const WEEK_MINS = 7 * 24 * 60; +const MONTH_MINS = 30 * 24 * 60; + +function isoFromEpochSeconds(value: number | null | undefined): string | undefined { + if (typeof value !== "number" || !Number.isFinite(value) || value <= 0) return undefined; + const dt = DateTime.make(value * 1000); + return Option.isSome(dt) ? DateTime.formatIso(dt.value) : undefined; +} + +function kindForDuration(mins: number): ServerProviderUsageWindow["kind"] { + if (mins >= MONTH_MINS) return "monthly"; + if (mins >= WEEK_MINS) return "weekly"; + return "session"; +} + +function labelForKind(kind: ServerProviderUsageWindow["kind"]): string { + return kind === "session" ? "Session" : kind === "weekly" ? "Weekly" : "Monthly"; +} + +/** + * `primary` / `secondary` are positions, not durations. Codex usually sends + * `windowDurationMins`; when it does not, paid plans expose the 5-hour and + * weekly pair and Free/Go expose one monthly allowance. + */ +export function codexRateLimitsToWindows( + snapshot: CodexRateLimitSnapshot, +): ReadonlyArray { + const isMonthlyPlan = snapshot.planType === "free" || snapshot.planType === "go"; + const positions = [ + ["primary", snapshot.primary, isMonthlyPlan ? MONTH_MINS : SESSION_MINS], + ["secondary", snapshot.secondary, WEEK_MINS], + ] as const; + const windows: ServerProviderUsageWindow[] = []; + for (const [id, window, fallbackMins] of positions) { + if (!window || !Number.isFinite(window.usedPercent)) continue; + const windowDurationMins = + typeof window.windowDurationMins === "number" ? window.windowDurationMins : fallbackMins; + const kind = kindForDuration(windowDurationMins); + const resetsAt = isoFromEpochSeconds(window.resetsAt); + windows.push({ + id, + kind, + label: labelForKind(kind), + usedPercent: clampPercent(window.usedPercent), + windowDurationMins, + ...(resetsAt ? { resetsAt } : {}), + }); + } + return windows; +} + +export function codexRateLimitsToLimits(input: { + readonly snapshot: CodexRateLimitSnapshot; + readonly checkedAt: string; +}): ServerProviderUsageLimits { + return makeUsageLimits({ + checkedAt: input.checkedAt, + windows: codexRateLimitsToWindows(input.snapshot), + }); +} + +export function codexRateLimitsToUpdate( + snapshot: CodexRateLimitSnapshot, +): ProviderUsageLimitsUpdate | undefined { + const windows = codexRateLimitsToWindows(snapshot); + return windows.length > 0 ? { windows } : undefined; +} + +/** + * A bounded, client-safe reason for a failed `account/rateLimits/read`. The + * raw error is for the log; only the category and, for a JSON-RPC failure, + * the code reach the Limits view. + */ +export function codexRateLimitsFailureMessage(error: CodexErrors.CodexAppServerError): string { + switch (error._tag) { + case "CodexAppServerRequestError": + return `Codex could not read usage (JSON-RPC ${error.code}).`; + case "CodexAppServerSpawnError": + return "Codex could not be started to read usage."; + case "CodexAppServerProcessExitedError": + return "Codex exited before it could report usage."; + default: + return "Codex did not answer the usage request."; + } +} diff --git a/apps/server/src/provider/Services/ServerProvider.ts b/apps/server/src/provider/Services/ServerProvider.ts index 121625129270..5d0486d87ca7 100644 --- a/apps/server/src/provider/Services/ServerProvider.ts +++ b/apps/server/src/provider/Services/ServerProvider.ts @@ -1,4 +1,4 @@ -import type { ServerProvider } from "@t3tools/contracts"; +import type { ProviderUsageLimitsUpdate, ServerProvider } from "@t3tools/contracts"; import type * as Effect from "effect/Effect"; import type * as Stream from "effect/Stream"; import type { ProviderMaintenanceCapabilities } from "../providerMaintenance.ts"; @@ -8,4 +8,12 @@ export interface ServerProviderShape { readonly getSnapshot: Effect.Effect; readonly refresh: Effect.Effect; readonly streamChanges: Stream.Stream; + /** + * Fold a runtime rate-limit update into the published snapshot without + * waiting for the next status probe. Sparse: windows merge by id and an + * update with no usable window leaves the snapshot untouched. + */ + readonly applyUsageLimits: ( + update: ProviderUsageLimitsUpdate & { readonly checkedAt: string }, + ) => Effect.Effect; } diff --git a/apps/server/src/provider/makeManagedServerProvider.test.ts b/apps/server/src/provider/makeManagedServerProvider.test.ts index fd50fa13eb08..aa0828e048c9 100644 --- a/apps/server/src/provider/makeManagedServerProvider.test.ts +++ b/apps/server/src/provider/makeManagedServerProvider.test.ts @@ -522,4 +522,137 @@ describe("makeManagedServerProvider", () => { }), ).pipe(Effect.provide(AlwaysRunTestLayer)), ); + + it.effect("applies runtime usage updates onto the published snapshot", () => + Effect.scoped( + Effect.gen(function* () { + const provider = yield* makeManagedServerProvider({ + maintenanceCapabilities, + getSettings: Effect.succeed({ enabled: true }), + streamSettings: Stream.empty, + haveSettingsChanged: (previous, next) => previous.enabled !== next.enabled, + initialSnapshot: () => Effect.succeed(initialSnapshot), + checkProvider: Effect.succeed({ + ...refreshedSnapshot, + usageLimits: { + checkedAt: "2026-04-10T00:00:01.000Z", + windows: [ + { id: "five_hour", kind: "session", label: "Session", usedPercent: 10 }, + { + id: "seven_day", + kind: "weekly", + label: "Weekly", + usedPercent: 20, + resetsAt: "2026-04-17T00:00:00.000Z", + }, + ], + }, + } satisfies ServerProvider), + refreshInterval: "1 hour", + }); + yield* Stream.take(provider.streamChanges, 1).pipe(Stream.runDrain); + + const updatesFiber = yield* Stream.take(provider.streamChanges, 1).pipe( + Stream.runCollect, + Effect.forkChild, + ); + yield* Effect.yieldNow; + + // Percent-only weekly update: keeps the probe's reset time. + yield* provider.applyUsageLimits({ + checkedAt: "2026-04-10T00:05:00.000Z", + windows: [{ id: "seven_day", kind: "weekly", label: "Weekly", usedPercent: 25 }], + }); + // No windows: nothing to merge, nothing published. + yield* provider.applyUsageLimits({ checkedAt: "2026-04-10T00:06:00.000Z", windows: [] }); + + const [update] = Array.from(yield* Fiber.join(updatesFiber)); + assert.deepStrictEqual(update?.usageLimits, { + checkedAt: "2026-04-10T00:05:00.000Z", + windows: [ + { id: "five_hour", kind: "session", label: "Session", usedPercent: 10 }, + { + id: "seven_day", + kind: "weekly", + label: "Weekly", + usedPercent: 25, + resetsAt: "2026-04-17T00:00:00.000Z", + }, + ], + }); + assert.deepStrictEqual(yield* provider.getSnapshot, update); + }), + ).pipe(Effect.provide(AlwaysRunTestLayer)), + ); + + it.effect("keeps live usage windows across a failed probe and a stale enrichment", () => + Effect.scoped( + Effect.gen(function* () { + const releaseEnrichment = yield* Deferred.make(); + const refreshCount = yield* Ref.make(0); + const probedLimits = { + checkedAt: "2026-04-10T00:00:01.000Z", + windows: [{ id: "primary", kind: "session", label: "Session", usedPercent: 10 }], + } as const; + const provider = yield* makeManagedServerProvider({ + maintenanceCapabilities, + getSettings: Effect.succeed({ enabled: true }), + streamSettings: Stream.empty, + haveSettingsChanged: (previous, next) => previous.enabled !== next.enabled, + initialSnapshot: () => Effect.succeed(initialSnapshot), + checkProvider: Ref.updateAndGet(refreshCount, (count) => count + 1).pipe( + Effect.map((count) => + count === 1 + ? { ...refreshedSnapshot, usageLimits: probedLimits } + : { + ...refreshedSnapshotSecond, + usageLimits: { + checkedAt: "2026-04-10T00:00:03.000Z", + windows: [], + unavailable: { reason: "probeFailed" }, + }, + }, + ), + ), + enrichSnapshot: ({ snapshot, publishSnapshot }) => + Deferred.await(releaseEnrichment).pipe( + Effect.flatMap(() => + publishSnapshot({ + ...enrichedSnapshot, + ...snapshot, + models: enrichedSnapshot.models, + }), + ), + ), + refreshInterval: "1 hour", + }); + yield* Stream.take(provider.streamChanges, 1).pipe(Stream.runDrain); + + const liveWindow = { + id: "primary", + kind: "session", + label: "Session", + usedPercent: 60, + } as const; + yield* provider.applyUsageLimits({ + checkedAt: "2026-04-10T00:00:02.000Z", + windows: [liveWindow], + }); + + // Enrichment computed from the pre-update snapshot lands afterwards. + yield* Deferred.succeed(releaseEnrichment, undefined); + const enriched = yield* Stream.take(provider.streamChanges, 1).pipe( + Stream.runCollect, + Effect.map((chunk) => Array.from(chunk)[0]!), + ); + assert.deepStrictEqual(enriched.models, enrichedSnapshot.models); + assert.deepStrictEqual(enriched.usageLimits?.windows, [liveWindow]); + + // A probe that could not read usage keeps the last good windows. + const refreshed = yield* provider.refresh; + assert.strictEqual(refreshed.message, refreshedSnapshotSecond.message); + assert.deepStrictEqual(refreshed.usageLimits?.windows, [liveWindow]); + }), + ).pipe(Effect.provide(AlwaysRunTestLayer)), + ); }); diff --git a/apps/server/src/provider/makeManagedServerProvider.ts b/apps/server/src/provider/makeManagedServerProvider.ts index a009157144c7..ec3d26e6c87e 100644 --- a/apps/server/src/provider/makeManagedServerProvider.ts +++ b/apps/server/src/provider/makeManagedServerProvider.ts @@ -17,6 +17,7 @@ import * as Semaphore from "effect/Semaphore"; import * as BackgroundPolicy from "../background/BackgroundPolicy.ts"; import { ServerSettingsService } from "../serverSettings.ts"; +import { applyUsageLimitsUpdate, resolveUsageLimitsAfterProbe } from "./providerUsageLimits.ts"; import type { ServerProviderShape } from "./Services/ServerProvider.ts"; interface ProviderSnapshotState { @@ -24,6 +25,17 @@ interface ProviderSnapshotState { readonly enrichmentGeneration: number; } +function withUsageLimits( + snapshot: ServerProvider, + usageLimits: ServerProvider["usageLimits"], +): ServerProvider { + if (snapshot.usageLimits === usageLimits) { + return snapshot; + } + const { usageLimits: _previous, ...rest } = snapshot; + return usageLimits ? { ...rest, usageLimits } : rest; +} + export const makeManagedServerProvider = Effect.fn("makeManagedServerProvider")(function* < Settings, >(input: { @@ -69,16 +81,16 @@ export const makeManagedServerProvider = Effect.fn("makeManagedServerProvider")( nextSnapshot: ServerProvider, ) { const snapshotToPublish = yield* Ref.modify(snapshotStateRef, (state) => { - if (state.enrichmentGeneration !== generation || Equal.equals(state.snapshot, nextSnapshot)) { + if (state.enrichmentGeneration !== generation) { return [null, state] as const; } - return [ - nextSnapshot, - { - ...state, - snapshot: nextSnapshot, - }, - ] as const; + // Enrichment derives from the snapshot it was handed; a runtime usage + // update that landed since must not be reverted by it. + const merged = withUsageLimits(nextSnapshot, state.snapshot.usageLimits); + if (Equal.equals(state.snapshot, merged)) { + return [null, state] as const; + } + return [merged, { ...state, snapshot: merged }] as const; }); if (snapshotToPublish === null) { return; @@ -138,19 +150,26 @@ export const makeManagedServerProvider = Effect.fn("makeManagedServerProvider")( return state.snapshot; } - const nextSnapshot = yield* input.checkProvider; - const nextGeneration = yield* Ref.modify(snapshotStateRef, (state) => { - const generation = input.enrichSnapshot - ? state.enrichmentGeneration + 1 - : state.enrichmentGeneration; - return [ - generation, - { - snapshot: nextSnapshot, - enrichmentGeneration: generation, - }, - ] as const; - }); + const probedSnapshot = yield* input.checkProvider; + const { snapshot: nextSnapshot, generation: nextGeneration } = yield* Ref.modify( + snapshotStateRef, + (state) => { + const generation = input.enrichSnapshot + ? state.enrichmentGeneration + 1 + : state.enrichmentGeneration; + const snapshot = withUsageLimits( + probedSnapshot, + resolveUsageLimitsAfterProbe({ + published: state.snapshot.usageLimits, + probed: probedSnapshot.usageLimits, + }), + ); + return [ + { snapshot, generation }, + { snapshot, enrichmentGeneration: generation }, + ] as const; + }, + ); yield* Ref.set(settingsRef, nextSettings); yield* PubSub.publish(changesPubSub, nextSnapshot); yield* restartSnapshotEnrichment(nextSettings, nextSnapshot, nextGeneration); @@ -159,6 +178,32 @@ export const makeManagedServerProvider = Effect.fn("makeManagedServerProvider")( const applySnapshot = (nextSettings: Settings, options?: { readonly forceRefresh?: boolean }) => refreshSemaphore.withPermits(1)(applySnapshotBase(nextSettings, options)); + /** + * Runtime usage updates arrive between probes. They patch only + * `usageLimits` on whatever snapshot is published and leave the enrichment + * generation alone, so an in-flight enrichment still lands. + */ + const applyUsageLimits: ServerProviderShape["applyUsageLimits"] = (update) => + Effect.gen(function* () { + const snapshotToPublish = yield* Ref.modify(snapshotStateRef, (state) => { + const usageLimits = applyUsageLimitsUpdate({ + previous: state.snapshot.usageLimits, + update, + checkedAt: update.checkedAt, + }); + // `applyUsageLimitsUpdate` hands back the same object when nothing + // moved, which is the common case for Codex's per-tick notification. + if (usageLimits === state.snapshot.usageLimits) { + return [null, state] as const; + } + const snapshot = withUsageLimits(state.snapshot, usageLimits); + return [snapshot, { ...state, snapshot }] as const; + }); + if (snapshotToPublish !== null) { + yield* PubSub.publish(changesPubSub, snapshotToPublish); + } + }); + const refreshSnapshot = Effect.fn("refreshSnapshot")(function* () { const nextSettings = yield* input.getSettings; return yield* applySnapshot(nextSettings, { forceRefresh: true }); @@ -241,6 +286,7 @@ export const makeManagedServerProvider = Effect.fn("makeManagedServerProvider")( maintenanceCapabilities: input.maintenanceCapabilities, getSnapshot: Ref.get(snapshotStateRef).pipe(Effect.map((state) => state.snapshot)), refresh: refreshSnapshot().pipe(Effect.tapError(Effect.logError), Effect.orDie), + applyUsageLimits, get streamChanges() { return Stream.fromPubSub(changesPubSub); }, diff --git a/apps/server/src/provider/providerSnapshot.ts b/apps/server/src/provider/providerSnapshot.ts index adbe110d9408..ff98ba8a00d5 100644 --- a/apps/server/src/provider/providerSnapshot.ts +++ b/apps/server/src/provider/providerSnapshot.ts @@ -7,6 +7,7 @@ import type { ServerProviderSlashCommand, ServerProviderModel, ServerProviderState, + ServerProviderUsageLimits, } from "@t3tools/contracts"; import * as Effect from "effect/Effect"; import * as PlatformError from "effect/PlatformError"; @@ -50,6 +51,7 @@ export interface ProviderProbeResult { readonly status: Exclude; readonly auth: ServerProviderAuth; readonly message?: string; + readonly usageLimits?: ServerProviderUsageLimits; } export interface ServerProviderPresentation { @@ -249,6 +251,7 @@ export function buildServerProvider(input: { models: input.models, slashCommands: [...(input.slashCommands ?? [])], skills: [...(input.skills ?? [])], + ...(input.probe.usageLimits ? { usageLimits: input.probe.usageLimits } : {}), ...(versionAdvisory ? { versionAdvisory } : {}), }; } diff --git a/apps/server/src/provider/providerUsageLimits.test.ts b/apps/server/src/provider/providerUsageLimits.test.ts new file mode 100644 index 000000000000..1fc4c4959a44 --- /dev/null +++ b/apps/server/src/provider/providerUsageLimits.test.ts @@ -0,0 +1,74 @@ +import { describe, expect, it } from "vite-plus/test"; + +import { applyUsageLimitsUpdate, resolveUsageLimitsAfterProbe } from "./providerUsageLimits.ts"; + +const checkedAt = "2026-09-03T12:00:00.000Z"; +const session = { + id: "five_hour", + kind: "session", + label: "Session", + usedPercent: 40, + windowDurationMins: 300, + resetsAt: "2026-09-03T14:00:00.000Z", +} as const; +const weekly = { + id: "seven_day", + kind: "weekly", + label: "Weekly", + usedPercent: 20, + windowDurationMins: 10_080, +} as const; +const published = { checkedAt, windows: [session, weekly] }; + +describe("applyUsageLimitsUpdate", () => { + it("returns the published object itself when no window moved", () => { + // Codex repeats the same numbers beside every token-usage tick; the + // ingestion path relies on identity to skip the publish. + const next = applyUsageLimitsUpdate({ + previous: published, + checkedAt: "2026-09-03T12:00:05.000Z", + update: { + windows: [ + { ...weekly }, + { id: "five_hour", kind: "session", label: "Session", usedPercent: 40 }, + ], + }, + }); + expect(next).toBe(published); + }); + + it("upserts by id and keeps the reset a percent-only update omits", () => { + const next = applyUsageLimitsUpdate({ + previous: published, + checkedAt: "2026-09-03T12:00:05.000Z", + update: { + windows: [{ id: "five_hour", kind: "session", label: "Session", usedPercent: 55 }], + }, + }); + expect(next).not.toBe(published); + expect(next).toEqual({ + checkedAt: "2026-09-03T12:00:05.000Z", + windows: [{ ...session, usedPercent: 55 }, weekly], + }); + }); + + it("leaves an unsupported account and an empty update alone", () => { + const unsupported = { checkedAt, windows: [], unavailable: { reason: "unsupported" as const } }; + expect( + applyUsageLimitsUpdate({ previous: unsupported, checkedAt, update: { windows: [session] } }), + ).toBe(unsupported); + expect( + applyUsageLimitsUpdate({ previous: published, checkedAt, update: { windows: [] } }), + ).toBe(published); + }); +}); + +describe("resolveUsageLimitsAfterProbe", () => { + it("keeps the last good windows through a failed probe but not an unsupported one", () => { + const failed = { checkedAt, windows: [], unavailable: { reason: "probeFailed" as const } }; + const unsupported = { checkedAt, windows: [], unavailable: { reason: "unsupported" as const } }; + expect(resolveUsageLimitsAfterProbe({ published, probed: failed })).toBe(published); + expect(resolveUsageLimitsAfterProbe({ published, probed: unsupported })).toBe(unsupported); + expect(resolveUsageLimitsAfterProbe({ published: undefined, probed: failed })).toBe(failed); + }); +}); diff --git a/apps/server/src/provider/providerUsageLimits.ts b/apps/server/src/provider/providerUsageLimits.ts new file mode 100644 index 000000000000..706825579f0a --- /dev/null +++ b/apps/server/src/provider/providerUsageLimits.ts @@ -0,0 +1,131 @@ +import type { + ProviderUsageLimitsUpdate, + ServerProviderUsageLimits, + ServerProviderUsageWindow, +} from "@t3tools/contracts"; + +const WINDOW_KIND_ORDER: Record = { + session: 0, + weekly: 1, + monthly: 2, + other: 3, +}; + +export function clampPercent(value: number): number { + return Number.isFinite(value) ? Math.max(0, Math.min(100, value)) : 0; +} + +function sortWindows( + windows: Iterable, +): ReadonlyArray { + return [...windows].toSorted( + (left, right) => + WINDOW_KIND_ORDER[left.kind] - WINDOW_KIND_ORDER[right.kind] || + left.id.localeCompare(right.id), + ); +} + +export function makeUsageLimits(input: { + readonly checkedAt: string; + readonly windows: Iterable; +}): ServerProviderUsageLimits { + return { checkedAt: input.checkedAt, windows: sortWindows(input.windows) }; +} + +export function makeUnavailableUsageLimits(input: { + readonly checkedAt: string; + readonly reason: "unsupported" | "probeFailed"; + readonly message?: string; +}): ServerProviderUsageLimits { + return { + checkedAt: input.checkedAt, + windows: [], + unavailable: { + reason: input.reason, + ...(input.message ? { message: input.message } : {}), + }, + }; +} + +/** + * Fold a sparse runtime update into the limits a provider currently + * publishes. Windows upsert by `id`; a window the update omits keeps its + * previous values, and a window that arrives without `resetsAt` or + * `windowDurationMins` keeps whatever the last probe resolved for it. An + * update with no windows leaves `previous` untouched. + * + * An `unsupported` snapshot stays unsupported: an account that cannot have + * subscription windows will not start reporting them mid-turn. + */ +export function applyUsageLimitsUpdate(input: { + readonly previous: ServerProviderUsageLimits | undefined; + readonly update: ProviderUsageLimitsUpdate; + readonly checkedAt: string; +}): ServerProviderUsageLimits | undefined { + const { previous, update } = input; + if (update.windows.length === 0 || previous?.unavailable?.reason === "unsupported") { + return previous; + } + const merged = new Map(previous?.windows.map((window) => [window.id, window] as const)); + // Codex sends this notification beside every token-usage tick, almost + // always with unchanged numbers. Decide "nothing changed" per window on + // the way through so the no-op case never allocates a new snapshot. + let changed = false; + for (const window of update.windows) { + const existing = merged.get(window.id); + const next: ServerProviderUsageWindow = { + ...window, + usedPercent: clampPercent(window.usedPercent), + ...(window.resetsAt === undefined && existing?.resetsAt !== undefined + ? { resetsAt: existing.resetsAt } + : {}), + ...(window.windowDurationMins === undefined && existing?.windowDurationMins !== undefined + ? { windowDurationMins: existing.windowDurationMins } + : {}), + }; + if (existing === undefined || !usageWindowEquals(existing, next)) { + merged.set(window.id, next); + changed = true; + } + } + if (!changed && previous !== undefined && previous.unavailable === undefined) { + return previous; + } + return makeUsageLimits({ checkedAt: input.checkedAt, windows: merged.values() }); +} + +function usageWindowEquals(a: ServerProviderUsageWindow, b: ServerProviderUsageWindow): boolean { + return ( + a.id === b.id && + a.kind === b.kind && + a.label === b.label && + a.usedPercent === b.usedPercent && + a.resetsAt === b.resetsAt && + a.windowDurationMins === b.windowDurationMins + ); +} + +/** + * Choose what to publish after a status probe finishes. A probe that failed + * this time must not wipe bars a previous probe or a turn already + * established, so the last good snapshot stays; `unsupported` is + * authoritative and replaces them. + * + * A successful probe replaces the published windows outright, including any + * runtime update that landed while it was running. That is a deliberate + * trade-off: the Codex and Claude reads take a few seconds at most, the + * probe is the fresher full read in every case except that window, and the + * per-window epoch bookkeeping needed to reconcile the two was more code + * than the sub-second regression it prevented. The next runtime event + * corrects it. + */ +export function resolveUsageLimitsAfterProbe(input: { + readonly published: ServerProviderUsageLimits | undefined; + readonly probed: ServerProviderUsageLimits | undefined; +}): ServerProviderUsageLimits | undefined { + const { published, probed } = input; + if (probed?.unavailable?.reason === "probeFailed" && published && !published.unavailable) { + return published; + } + return probed; +} diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index d972e2e00803..29338f0b73ad 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -101,6 +101,7 @@ import { import * as CheckpointDiffQuery from "./checkpointing/CheckpointDiffQuery.ts"; import * as GitManager from "./git/GitManager.ts"; import * as EnvironmentTheme from "./environmentTheme.ts"; +import * as UsageLimitSources from "./usage/UsageLimitSources.ts"; import * as Keybindings from "./keybindings.ts"; import * as ExternalLauncher from "./process/externalLauncher.ts"; import * as RemoteOpenTargets from "./environment/RemoteOpenTargets.ts"; @@ -740,6 +741,11 @@ const buildAppUnderTest = (options?: { streamChanges: Stream.empty, ...options?.layers?.environmentTheme, }), + Layer.mock(UsageLimitSources.UsageLimitSources)({ + current: Effect.succeed([]), + streamChanges: Stream.empty, + refresh: Effect.void, + }), ), ), Layer.provide( diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 3a93adc6d761..ee0d6936f394 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -41,6 +41,7 @@ import { AntigravityInstallation } from "./provider/AntigravityInstallation.ts"; import { ProviderInstanceRegistry } from "./provider/Services/ProviderInstanceRegistry.ts"; import { ProviderRegistry } from "./provider/Services/ProviderRegistry.ts"; import { ProviderSessionReaperLive } from "./provider/Layers/ProviderSessionReaper.ts"; +import { ProviderUsageLimitsIngestionLive } from "./provider/Layers/ProviderUsageLimitsIngestion.ts"; import * as OpenCodeRuntime from "./provider/opencodeRuntime.ts"; import * as CheckpointDiffQuery from "./checkpointing/CheckpointDiffQuery.ts"; import * as CheckpointStore from "./checkpointing/CheckpointStore.ts"; @@ -118,6 +119,7 @@ import * as NativeTelemetryClient from "./resourceTelemetry/NativeTelemetryClien import * as ResourceAttribution from "./resourceTelemetry/ResourceAttribution.ts"; import * as ResourceMonitorBinary from "./resourceTelemetry/ResourceMonitorBinary.ts"; import * as ResourceTelemetry from "./resourceTelemetry/ResourceTelemetry.ts"; +import * as UsageLimitSources from "./usage/UsageLimitSources.ts"; import * as UsageService from "./usage/UsageService.ts"; import { OrchestrationLayerLive } from "./orchestration/runtimeLayer.ts"; import { @@ -407,6 +409,9 @@ const CloudManagedEndpointRuntimeLive = Layer.mergeAll( ); const ProviderRuntimeLayerLive = ProviderSessionReaperLive.pipe( + // Subscribes to `account.rate-limits.updated` so usage bars track live + // telemetry instead of waiting for the next status probe. + Layer.provideMerge(ProviderUsageLimitsIngestionLive), Layer.provideMerge(ProviderLayerLive), Layer.provideMerge(OrchestrationLayerLive), ); @@ -454,7 +459,9 @@ const RuntimeCoreDependenciesLive = ReactorLayerLive.pipe( Layer.provideMerge(PersistenceLayerLive), // Both read a user-owned file out of the state directory and stream changes // to clients; neither depends on the other. - Layer.provideMerge(Layer.mergeAll(Keybindings.layer, EnvironmentTheme.layer)), + Layer.provideMerge( + Layer.mergeAll(Keybindings.layer, EnvironmentTheme.layer, UsageLimitSources.layer), + ), Layer.provideMerge(ProviderRegistryLive), // The instance registry is the new routing keystone — text generation, // adapter lookup, and runtime ingestion all resolve `ProviderInstanceId` diff --git a/apps/server/src/serverSettings.ts b/apps/server/src/serverSettings.ts index 5a8650b7e405..5f2550534883 100644 --- a/apps/server/src/serverSettings.ts +++ b/apps/server/src/serverSettings.ts @@ -18,6 +18,7 @@ import { type ModelSelection, type ProviderInstanceConfig, type ProviderInstanceEnvironmentVariable, + type UsageLimitSourceConfig, ProviderDriverKind, ProviderInstanceId, ServerSettings, @@ -133,6 +134,17 @@ function providerEnvironmentSecretName(input: { return `provider-env-${Buffer.from(input.instanceId, "utf8").toString("base64url")}-${Buffer.from(input.name, "utf8").toString("base64url")}`; } +/** + * On disk the hub key is replaced by this marker and the real value lives in + * the secret store, mirroring provider environment secrets. A client that + * sends the marker back means "keep what you have". + */ +const USAGE_LIMIT_SOURCE_KEY_REDACTED = "\u2022\u2022\u2022\u2022\u2022\u2022"; + +export function usageLimitSourceSecretName(sourceId: string): string { + return `usage-limit-source-${Buffer.from(sourceId, "utf8").toString("base64url")}`; +} + function redactProviderEnvironmentVariable( variable: ProviderInstanceEnvironmentVariable, ): ProviderInstanceEnvironmentVariable { @@ -159,7 +171,17 @@ export function redactServerSettingsForClient(settings: ServerSettings): ServerS : instance, ]), ); - return { ...settings, providerInstances }; + // The hub key is a bearer secret; clients only need to know one is set. + const usageLimitSources = Object.fromEntries( + Object.entries(settings.usageLimitSources).map(([id, source]) => [ + id, + { + ...source, + managementKey: source.managementKey.length > 0 ? USAGE_LIMIT_SOURCE_KEY_REDACTED : "", + }, + ]), + ); + return { ...settings, providerInstances, usageLimitSources }; } export class ServerSettingsService extends Context.Service< @@ -512,9 +534,28 @@ const make = Effect.gen(function* () { environment, } satisfies ProviderInstanceConfig; } + const usageLimitSources: Record = {}; + for (const [sourceId, source] of Object.entries(settings.usageLimitSources)) { + if (source.managementKey !== USAGE_LIMIT_SOURCE_KEY_REDACTED) { + usageLimitSources[sourceId] = source; + continue; + } + const secret = yield* secretStore + .get(usageLimitSourceSecretName(sourceId)) + .pipe( + Effect.mapError( + (cause) => new ServerSettingsError({ settingsPath, operation: "read-secret", cause }), + ), + ); + usageLimitSources[sourceId] = { + ...source, + managementKey: Option.isSome(secret) ? textDecoder.decode(secret.value) : "", + }; + } return { ...settings, providerInstances: providerInstances as ServerSettings["providerInstances"], + usageLimitSources: usageLimitSources as ServerSettings["usageLimitSources"], }; }); @@ -630,9 +671,52 @@ const make = Effect.gen(function* () { } } + const usageLimitSources: Record = {}; + for (const [sourceId, source] of Object.entries(next.usageLimitSources)) { + const secretName = usageLimitSourceSecretName(sourceId); + if (source.managementKey === USAGE_LIMIT_SOURCE_KEY_REDACTED) { + // Unchanged from the client's point of view; the store already has it. + usageLimitSources[sourceId] = source; + continue; + } + if (source.managementKey.length === 0) { + yield* secretStore + .remove(secretName) + .pipe( + Effect.mapError( + (cause) => + new ServerSettingsError({ settingsPath, operation: "remove-secret", cause }), + ), + ); + usageLimitSources[sourceId] = source; + continue; + } + yield* secretStore + .set(secretName, textEncoder.encode(source.managementKey)) + .pipe( + Effect.mapError( + (cause) => + new ServerSettingsError({ settingsPath, operation: "write-secret", cause }), + ), + ); + usageLimitSources[sourceId] = { ...source, managementKey: USAGE_LIMIT_SOURCE_KEY_REDACTED }; + } + for (const sourceId of Object.keys(current.usageLimitSources)) { + if (sourceId in next.usageLimitSources) continue; + yield* secretStore + .remove(usageLimitSourceSecretName(sourceId)) + .pipe( + Effect.mapError( + (cause) => + new ServerSettingsError({ settingsPath, operation: "remove-stale-secret", cause }), + ), + ); + } + return { ...next, providerInstances: providerInstances as ServerSettings["providerInstances"], + usageLimitSources: usageLimitSources as ServerSettings["usageLimitSources"], }; }); diff --git a/apps/server/src/usage/UsageLimitSources.ts b/apps/server/src/usage/UsageLimitSources.ts new file mode 100644 index 000000000000..abe7f8e64999 --- /dev/null +++ b/apps/server/src/usage/UsageLimitSources.ts @@ -0,0 +1,202 @@ +/** + * UsageLimitSources — quota from places this environment cannot run turns + * on, today a CLIProxyAPI hub pooling several subscription accounts. + * + * Each configured `settings.usageLimitSources` entry is polled on the + * provider health-check interval and on every settings change, then + * published as one snapshot per source over `subscribeServerConfig`. A source + * that fails keeps its row with `error` set so the user can see it is + * configured but unreachable. Nothing is persisted: like provider status, + * this is live state that re-derives on boot. + * + * @module usage/UsageLimitSources + */ +import { + DEFAULT_PROVIDER_HEALTH_REFRESH_INTERVAL, + type ServerSettings, + type UsageLimitSourceConfig, + type UsageLimitSourceId, + type UsageLimitSourceSnapshot, +} from "@t3tools/contracts"; +import { resolveServerBackgroundActivitySettings } from "@t3tools/shared/backgroundActivitySettings"; +import type * as Cause from "effect/Cause"; +import * as Context from "effect/Context"; +import * as Data from "effect/Data"; +import * as DateTime from "effect/DateTime"; +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as Equal from "effect/Equal"; +import * as Layer from "effect/Layer"; +import * as PubSub from "effect/PubSub"; +import * as Ref from "effect/Ref"; +import * as Semaphore from "effect/Semaphore"; +import type * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; +import { HttpClient, type HttpClientError, HttpClientResponse } from "effect/unstable/http"; + +import * as BackgroundPolicy from "../background/BackgroundPolicy.ts"; +import { ServerSettingsService } from "../serverSettings.ts"; +import { cliproxyStatusToAccounts, decodeCliproxyQuotaStatus } from "./cliproxyUsageLimits.ts"; + +const FETCH_TIMEOUT = "10 seconds"; +const QUOTA_STATUS_PATH = "/v0/management/quota-scheduler/status"; + +export class UsageLimitSources extends Context.Service< + UsageLimitSources, + { + readonly current: Effect.Effect>; + /** The current set followed by every change, with repeats dropped. */ + readonly streamChanges: Stream.Stream>; + /** Re-read every source now. Never fails; failures land on the snapshot. */ + readonly refresh: Effect.Effect; + } +>()("t3/usage/UsageLimitSources") {} + +/** + * A bounded, client-safe reason for a failed hub read. The exact failure + * (which can carry the request URL and response body) goes to the log. + */ +function readFailureMessage( + error: HttpClientError.HttpClientError | Schema.SchemaError | Cause.TimeoutError | InvalidUrl, +): string { + switch (error._tag) { + case "InvalidUrl": + return "The hub URL is not valid."; + case "TimeoutError": + return "The hub did not answer in time."; + case "SchemaError": + return "The hub answered with an unexpected shape."; + case "HttpClientError": + return error.reason._tag === "StatusCodeError" + ? `The hub refused the request (HTTP ${error.reason.response.status}).` + : "The hub could not be reached."; + } +} + +class InvalidUrl extends Data.TaggedError("InvalidUrl")<{ + readonly url: string; + readonly cause: unknown; +}> {} + +function sourceLabel(id: string, config: UsageLimitSourceConfig): string { + if (config.label) return config.label; + try { + return new URL(config.url).host; + } catch { + return id; + } +} + +export const make = Effect.gen(function* () { + const httpClient = yield* HttpClient.HttpClient; + const settingsService = yield* ServerSettingsService; + const backgroundPolicy = yield* BackgroundPolicy.BackgroundPolicy; + const stateRef = yield* Ref.make>([]); + const changes = yield* Effect.acquireRelease( + PubSub.unbounded>(), + PubSub.shutdown, + ); + + const readSource = Effect.fn("UsageLimitSources.readSource")(function* ( + id: UsageLimitSourceId, + config: UsageLimitSourceConfig, + ) { + const checkedAt = DateTime.formatIso(yield* DateTime.now); + const base = { id, kind: config.kind, label: sourceLabel(id, config), checkedAt } as const; + if (config.managementKey.length === 0) { + return { ...base, accounts: [], error: "No management key configured." }; + } + const accounts = yield* Effect.try({ + try: () => new URL(QUOTA_STATUS_PATH, config.url).toString(), + catch: (cause) => new InvalidUrl({ url: config.url, cause }), + }).pipe( + Effect.flatMap((url) => + httpClient.get(url, { headers: { Authorization: `Bearer ${config.managementKey}` } }), + ), + Effect.flatMap(HttpClientResponse.filterStatusOk), + Effect.flatMap((response) => response.json), + Effect.flatMap(decodeCliproxyQuotaStatus), + Effect.map((status) => cliproxyStatusToAccounts(status, checkedAt)), + Effect.timeout(FETCH_TIMEOUT), + Effect.result, + ); + if (accounts._tag === "Failure") { + yield* Effect.logDebug("usage limit source read failed", { id, cause: accounts.failure }); + return { ...base, accounts: [], error: readFailureMessage(accounts.failure) }; + } + return { ...base, accounts: accounts.success }; + }); + + const publish = (next: ReadonlyArray) => + Effect.gen(function* () { + const changed = yield* Ref.modify(stateRef, (previous) => + Equal.equals(previous, next) ? [false, previous] : [true, next], + ); + if (changed) yield* PubSub.publish(changes, next); + }); + + // One refresh at a time: a slow hub read started before a settings change + // must not publish after the change's own refresh and resurrect a removed + // source. Callers queue behind the in-flight run and see current settings. + const refreshLock = yield* Semaphore.make(1); + const refresh = Effect.gen(function* () { + const settings = yield* settingsService.getSettings.pipe( + Effect.orElseSucceed((): ServerSettings | null => null), + ); + const entries = Object.entries(settings?.usageLimitSources ?? {}).filter( + ([, config]) => config.enabled, + ); + const snapshots = yield* Effect.forEach( + entries, + ([id, config]) => readSource(id as UsageLimitSourceId, config), + { concurrency: 4 }, + ); + yield* publish(snapshots); + }).pipe(refreshLock.withPermits(1), Effect.ignoreCause({ log: true })); + + // Settings edits re-read straight away so a new hub shows up without + // waiting for the interval, and a removed one leaves the list. + yield* settingsService.streamChanges.pipe( + Stream.map((settings) => settings.usageLimitSources), + Stream.changes, + Stream.runForEach(() => refresh), + Effect.forkScoped, + ); + + const interval = settingsService.getSettings.pipe( + Effect.map( + (settings) => resolveServerBackgroundActivitySettings(settings).providerHealthRefreshInterval, + ), + Effect.orElseSucceed(() => DEFAULT_PROVIDER_HEALTH_REFRESH_INTERVAL), + ); + yield* Effect.forever( + interval.pipe( + Effect.flatMap((wait) => + Effect.sleep(Duration.toMillis(Duration.fromInputUnsafe(wait)) <= 0 ? "60 seconds" : wait), + ), + Effect.andThen(backgroundPolicy.shouldRunScopeWork({ type: "provider-status" })), + Effect.flatMap((shouldRun) => (shouldRun ? refresh : Effect.void)), + Effect.ignoreCause({ log: true }), + ), + ).pipe(Effect.forkScoped); + + yield* refresh.pipe(Effect.forkScoped); + + return { + current: Ref.get(stateRef), + refresh, + get streamChanges() { + return Stream.unwrap( + Effect.gen(function* () { + const subscription = yield* PubSub.subscribe(changes); + const snapshot = yield* Ref.get(stateRef); + return Stream.concat(Stream.make(snapshot), Stream.fromSubscription(subscription)).pipe( + Stream.changes, + ); + }), + ); + }, + } satisfies UsageLimitSources["Service"]; +}); + +export const layer = Layer.effect(UsageLimitSources, make); diff --git a/apps/server/src/usage/cliproxyUsageLimits.test.ts b/apps/server/src/usage/cliproxyUsageLimits.test.ts new file mode 100644 index 000000000000..19767f3a9270 --- /dev/null +++ b/apps/server/src/usage/cliproxyUsageLimits.test.ts @@ -0,0 +1,118 @@ +import { describe, expect, it } from "vite-plus/test"; + +import { accountEmailFromAuthFile, cliproxyStatusToAccounts } from "./cliproxyUsageLimits.ts"; + +const checkedAt = "2026-09-03T22:00:00.000Z"; + +describe("cliproxyStatusToAccounts", () => { + // Trimmed from a live `quota-scheduler/status`: one Claude account with a + // hard-limited Fable bucket, one Codex account whose 5h window is unknown. + it("maps each pooled account onto the windows the provider drivers use", () => { + const accounts = cliproxyStatusToAccounts( + { + accounts: { + "claude-jmarminge@gmail.com.json": { + provider: "claude", + fetched_at: "2026-09-03T15:06:07-07:00", + five_hour: { hard_limited: false, known: true, used_percent: 0 }, + seven_day: { + hard_limited: false, + known: true, + reset_at: "2026-09-07T07:59:59Z", + used_percent: 51, + }, + fable: { + hard_limited: true, + known: true, + reset_at: "2026-09-07T07:59:59Z", + used_percent: 100, + }, + }, + "codex-7f42123a-jmarminge@gmail.com-pro.json": { + provider: "codex", + plan: "pro", + fetched_at: "2026-09-03T15:07:07-07:00", + five_hour: { hard_limited: false, known: false, used_percent: 0 }, + weekly: { + hard_limited: false, + known: true, + reset_at: "2026-09-06T19:52:53-07:00", + used_percent: 12, + }, + }, + "xai-someone@example.com.json": { provider: "xai" }, + }, + }, + checkedAt, + ); + + expect(accounts).toEqual([ + { + id: "claude-jmarminge@gmail.com.json", + driver: "claudeAgent", + email: "jmarminge@gmail.com", + plan: "Claude Subscription", + usageLimits: { + checkedAt: "2026-09-03T22:06:07.000Z", + windows: [ + { + id: "five_hour", + kind: "session", + label: "Session", + usedPercent: 0, + windowDurationMins: 300, + }, + { + id: "seven_day", + kind: "weekly", + label: "Weekly", + usedPercent: 51, + windowDurationMins: 10080, + resetsAt: "2026-09-07T07:59:59.000Z", + }, + { + id: "seven_day_fable", + kind: "weekly", + label: "Weekly · Fable", + usedPercent: 100, + windowDurationMins: 10080, + resetsAt: "2026-09-07T07:59:59.000Z", + }, + ], + }, + }, + { + id: "codex-7f42123a-jmarminge@gmail.com-pro.json", + driver: "codex", + email: "jmarminge@gmail.com", + plan: "ChatGPT Pro 20x Subscription", + usageLimits: { + checkedAt: "2026-09-03T22:07:07.000Z", + windows: [ + { + id: "secondary", + kind: "weekly", + label: "Weekly", + usedPercent: 12, + windowDurationMins: 10080, + resetsAt: "2026-09-07T02:52:53.000Z", + }, + ], + }, + }, + ]); + }); +}); + +describe("accountEmailFromAuthFile", () => { + it("pulls the email out of the hub's auth file names", () => { + expect(accountEmailFromAuthFile("claude-julius@ping.gg.json")).toBe("julius@ping.gg"); + expect(accountEmailFromAuthFile("codex-e413dce6-julius@ping.gg-pro.json")).toBe( + "julius@ping.gg", + ); + expect(accountEmailFromAuthFile("claude-first-last@example.com.json")).toBe( + "first-last@example.com", + ); + expect(accountEmailFromAuthFile("mystery.json")).toBeUndefined(); + }); +}); diff --git a/apps/server/src/usage/cliproxyUsageLimits.ts b/apps/server/src/usage/cliproxyUsageLimits.ts new file mode 100644 index 000000000000..cd2b1e277da6 --- /dev/null +++ b/apps/server/src/usage/cliproxyUsageLimits.ts @@ -0,0 +1,160 @@ +/** + * Maps a CLIProxyAPI hub's `quota-scheduler/status` response onto the usage + * limit windows the Limits view renders, one account per pooled auth file. + * + * The hub already normalises each upstream: Claude accounts carry + * `five_hour` / `seven_day` / `fable`, Codex accounts `five_hour` / `weekly`. + * Every window is `{ used_percent, reset_at?, known, hard_limited }`. + * + * @module usage/cliproxyUsageLimits + */ +import { + ProviderDriverKind, + type ServerProviderUsageLimits, + type ServerProviderUsageWindow, + type UsageLimitSourceAccount, +} from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +import * as Option from "effect/Option"; +import * as Schema from "effect/Schema"; + +import { codexPlanLabel } from "../provider/Layers/CodexProvider.ts"; +import { clampPercent, makeUsageLimits } from "../provider/providerUsageLimits.ts"; + +const QuotaWindow = Schema.Struct({ + used_percent: Schema.Number, + reset_at: Schema.optional(Schema.String), + known: Schema.optional(Schema.Boolean), + hard_limited: Schema.optional(Schema.Boolean), +}); + +const QuotaAccount = Schema.Struct({ + provider: Schema.String, + plan: Schema.optional(Schema.String), + fetched_at: Schema.optional(Schema.String), + five_hour: Schema.optional(QuotaWindow), + seven_day: Schema.optional(QuotaWindow), + weekly: Schema.optional(QuotaWindow), + fable: Schema.optional(QuotaWindow), +}); + +export const CliproxyQuotaStatus = Schema.Struct({ + accounts: Schema.Record(Schema.String, QuotaAccount), +}); +export type CliproxyQuotaStatus = typeof CliproxyQuotaStatus.Type; +export const decodeCliproxyQuotaStatus = Schema.decodeUnknownEffect(CliproxyQuotaStatus); + +const SESSION_MINS = 5 * 60; +const WEEK_MINS = 7 * 24 * 60; + +/** Window ids match what the provider drivers emit, so rows read the same across sources. */ +const WINDOWS: ReadonlyArray<{ + readonly key: keyof Omit; + readonly id: string; + readonly kind: ServerProviderUsageWindow["kind"]; + readonly label: string; + readonly windowDurationMins: number; +}> = [ + { + key: "five_hour", + id: "five_hour", + kind: "session", + label: "Session", + windowDurationMins: SESSION_MINS, + }, + { + key: "seven_day", + id: "seven_day", + kind: "weekly", + label: "Weekly", + windowDurationMins: WEEK_MINS, + }, + { + key: "weekly", + id: "secondary", + kind: "weekly", + label: "Weekly", + windowDurationMins: WEEK_MINS, + }, + { + key: "fable", + id: "seven_day_fable", + kind: "weekly", + label: "Weekly · Fable", + windowDurationMins: WEEK_MINS, + }, +]; + +const DRIVER_BY_HUB_PROVIDER: Readonly> = { + claude: ProviderDriverKind.make("claudeAgent"), + codex: ProviderDriverKind.make("codex"), +}; + +function isoFromHub(value: string | undefined): string | undefined { + if (!value) return undefined; + const dt = DateTime.make(value); + return Option.isSome(dt) ? DateTime.formatIso(dt.value) : undefined; +} + +/** `claude-julius@ping.gg.json` → `julius@ping.gg`; `codex--x@y-pro.json` → `x@y`. */ +export function accountEmailFromAuthFile(fileName: string): string | undefined { + const stem = fileName.replace(/\.json$/i, ""); + // Strip the provider prefix (and Codex's hash) rather than splitting on + // `-`, so a hyphenated local part such as `first-last@` survives. + return stem.match(/^(?:claude-|codex-[a-z0-9]+-)?([^\s/]+@[^\s/]+?)(?:-[a-z0-9]+)?$/i)?.[1]; +} + +/** + * The hub only reports a plan slug for Codex. Claude accounts carry no tier + * in the scheduler status, so the row says what it is: a Claude subscription. + */ +function planLabel(account: typeof QuotaAccount.Type): string | undefined { + if (account.provider === "codex") return codexPlanLabel(account.plan); + if (account.provider === "claude") return "Claude Subscription"; + return undefined; +} + +export function cliproxyAccountToUsageLimits( + account: typeof QuotaAccount.Type, + checkedAt: string, +): ServerProviderUsageLimits { + const windows: ServerProviderUsageWindow[] = []; + for (const spec of WINDOWS) { + const window = account[spec.key]; + if (!window || window.known === false) continue; + const resetsAt = isoFromHub(window.reset_at); + windows.push({ + id: spec.id, + kind: spec.kind, + label: spec.label, + windowDurationMins: spec.windowDurationMins, + // The hub flags a window it has seen a 429 on; the percent may lag. + usedPercent: window.hard_limited ? 100 : clampPercent(window.used_percent), + ...(resetsAt ? { resetsAt } : {}), + }); + } + return makeUsageLimits({ checkedAt: isoFromHub(account.fetched_at) ?? checkedAt, windows }); +} + +export function cliproxyStatusToAccounts( + status: CliproxyQuotaStatus, + checkedAt: string, +): ReadonlyArray { + const accounts: UsageLimitSourceAccount[] = []; + for (const [fileName, account] of Object.entries(status.accounts)) { + const driver = DRIVER_BY_HUB_PROVIDER[account.provider]; + if (!driver) continue; + const email = accountEmailFromAuthFile(fileName); + const plan = planLabel(account); + accounts.push({ + id: fileName, + driver, + ...(email ? { email } : {}), + ...(plan ? { plan } : {}), + usageLimits: cliproxyAccountToUsageLimits(account, checkedAt), + }); + } + return accounts.toSorted( + (left, right) => left.driver.localeCompare(right.driver) || left.id.localeCompare(right.id), + ); +} diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 839937cf2ea7..c8dd103bcaea 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -127,6 +127,7 @@ import * as ProcessDiagnostics from "./diagnostics/ProcessDiagnostics.ts"; import * as ProcessResourceMonitor from "./diagnostics/ProcessResourceMonitor.ts"; import * as ResourceTelemetry from "./resourceTelemetry/ResourceTelemetry.ts"; import * as AnalyticsService from "./telemetry/AnalyticsService.ts"; +import * as UsageLimitSources from "./usage/UsageLimitSources.ts"; import * as UsageService from "./usage/UsageService.ts"; import * as TraceDiagnostics from "./diagnostics/TraceDiagnostics.ts"; import * as PullRequestService from "./pullRequest/PullRequestService.ts"; @@ -501,6 +502,7 @@ const makeWsRpcLayer = ( const checkpointDiffQuery = yield* CheckpointDiffQuery.CheckpointDiffQuery; const keybindings = yield* Keybindings.Keybindings; const environmentTheme = yield* EnvironmentTheme.EnvironmentThemeService; + const usageLimitSources = yield* UsageLimitSources.UsageLimitSources; const externalLauncher = yield* ExternalLauncher.ExternalLauncher; const remoteOpenTargets = yield* RemoteOpenTargets.RemoteOpenTargets; const gitWorkflow = yield* GitWorkflowService.GitWorkflowService; @@ -1674,6 +1676,13 @@ const makeWsRpcLayer = ( observeRpcEffect( WS_METHODS.serverRefreshProviders, Effect.gen(function* () { + // An untargeted refresh is "re-read everything's status", which + // includes quota from configured usage-limit sources. Awaited, + // not forked: the RPC scope closes on return and would + // interrupt a fork before the hub answered. + if (input.instanceId === undefined) { + yield* usageLimitSources.refresh; + } let providers = yield* input.cwd !== undefined && input.instanceId !== undefined ? providerRegistry.refreshWorkspaceSnapshot({ instanceId: input.instanceId, @@ -2571,6 +2580,17 @@ const makeWsRpcLayer = ( })), ) : Stream.empty; + // Same gate as themes: an older client dies on an unknown event. + const usageLimitSourceUpdates = + input.usageLimitSources === true + ? usageLimitSources.streamChanges.pipe( + Stream.map((sources) => ({ + version: 1 as const, + type: "usageLimitSourcesUpdated" as const, + payload: { sources }, + })), + ) + : Stream.empty; const settingsUpdates = serverSettings.streamChanges.pipe( Stream.map((settings) => ServerSettings.redactServerSettingsForClient(settings)), Stream.map((settings) => ({ @@ -2588,7 +2608,10 @@ const makeWsRpcLayer = ( keybindingsUpdates, Stream.merge( providerStatuses, - Stream.merge(settingsUpdates, environmentThemeUpdates), + Stream.merge( + settingsUpdates, + Stream.merge(environmentThemeUpdates, usageLimitSourceUpdates), + ), ), ); diff --git a/apps/web/src/components/usage/AddUsageLimitSourceDialog.tsx b/apps/web/src/components/usage/AddUsageLimitSourceDialog.tsx new file mode 100644 index 000000000000..5ce580969d6e --- /dev/null +++ b/apps/web/src/components/usage/AddUsageLimitSourceDialog.tsx @@ -0,0 +1,153 @@ +import { UsageLimitSourceId } from "@t3tools/contracts"; +import { useState } from "react"; + +import { useUpdatePrimarySettings } from "../../hooks/useSettings"; +import { Button } from "../ui/button"; +import { + Dialog, + DialogDescription, + DialogFooter, + DialogHeader, + DialogPanel, + DialogPopup, + DialogTitle, +} from "../ui/dialog"; +import { Input } from "../ui/input"; +import { Label } from "../ui/label"; + +/** + * Stable per hub and readable in settings.json. Dots and dashes in the host + * are kept so `foo-bar.com` and `foo.bar.com` do not collide; anything else + * (a port's colon, a path) is folded to a dash. + */ +function sourceIdFromUrl(url: string): UsageLimitSourceId { + let host = url; + try { + host = new URL(url).host; + } catch { + // Keep the raw text; the server reports the bad URL on its row. + } + const slug = host + .toLowerCase() + .replace(/[^a-z0-9.-]+/g, "-") + .replace(/^-+|-+$/g, ""); + return UsageLimitSourceId.make(`cliproxy-${slug || "hub"}`); +} + +/** + * Adds a CLIProxyAPI hub as a usage-limit source on the primary environment. + * The management key is sent once and kept in the server's secret store; + * settings only ever carry a redaction marker for it afterwards. + */ +export function AddUsageLimitSourceDialog({ + open, + onOpenChange, +}: { + readonly open: boolean; + readonly onOpenChange: (open: boolean) => void; +}) { + const updateSettings = useUpdatePrimarySettings(); + const [label, setLabel] = useState(""); + const [url, setUrl] = useState(""); + const [managementKey, setManagementKey] = useState(""); + const trimmedUrl = url.trim(); + const canSave = trimmedUrl.length > 0 && managementKey.trim().length > 0; + + const reset = () => { + setLabel(""); + setUrl(""); + setManagementKey(""); + }; + + const save = () => { + if (!canSave) return; + const id = sourceIdFromUrl(trimmedUrl); + // The patch names only this entry; the server merges it into its map. + updateSettings({ + usageLimitSources: { + [id]: { + kind: "cliproxy", + ...(label.trim() ? { label: label.trim() } : {}), + url: trimmedUrl, + managementKey: managementKey.trim(), + enabled: true, + }, + }, + }); + reset(); + onOpenChange(false); + }; + + return ( + { + if (!next) reset(); + onOpenChange(next); + }} + > + + + Add a CLIProxyAPI hub + + Show the quota of every account the hub pools, next to the providers on this machine. + The key stays on the server. + + + +
{ + event.preventDefault(); + save(); + }} + > +
+ + setUrl(event.target.value)} + autoFocus + /> +
+
+ + setManagementKey(event.target.value)} + /> +
+
+ + setLabel(event.target.value)} + /> +
+
+ + + + + + + + ); +} diff --git a/apps/web/src/components/usage/UsageLimits.tsx b/apps/web/src/components/usage/UsageLimits.tsx new file mode 100644 index 000000000000..3b36f50948d4 --- /dev/null +++ b/apps/web/src/components/usage/UsageLimits.tsx @@ -0,0 +1,458 @@ +import type { + ServerProvider, + ServerProviderUsageWindow, + UsageLimitSourceAccount, + UsageLimitSourceId, + UsageLimitSourceSnapshot, + UsageProviderKind, +} from "@t3tools/contracts"; +import { useAtomValue } from "@effect/atom-react"; +import { + collectLimitSources, + collectLimitsGroups, + elapsedShare, + formatResetsIn, + limitsNotice, + type LimitPace, + paceOf, + providerLimitsLabel, +} from "@t3tools/shared/usageLimits"; +import { GaugeIcon, PlusIcon, TrendingDownIcon, TrendingUpIcon } from "lucide-react"; +import { Fragment, useState } from "react"; + +import { + usePrimarySettings, + usePrimarySettingsAvailable, + useUpdatePrimarySettings, +} from "../../hooks/useSettings"; +import { usePrimaryEnvironmentId } from "../../state/environments"; +import { environmentPresentations } from "../../state/presentation"; +import { formatUpcomingTimestamp } from "../../timestampFormat"; +import { ProviderInstanceIcon } from "../chat/ProviderInstanceIcon"; +import { getDriverOption } from "../settings/providerDriverMeta"; +import { RedactedSensitiveText } from "../settings/RedactedSensitiveText"; +import { + AlertDialog, + AlertDialogClose, + AlertDialogDescription, + AlertDialogFooter, + AlertDialogHeader, + AlertDialogPopup, + AlertDialogTitle, +} from "../ui/alert-dialog"; +import { Badge } from "../ui/badge"; +import { Button } from "../ui/button"; +import { Tooltip, TooltipPopup, TooltipTrigger } from "../ui/tooltip"; +import { AddUsageLimitSourceDialog } from "./AddUsageLimitSourceDialog"; +import { PROVIDER_PRESENTATION } from "./usageProviders"; + +const PACE: Record = { + ahead: { label: "Ahead of pace: spending faster than the window elapses", icon: TrendingUpIcon }, + on: { label: "On pace with the window", icon: GaugeIcon }, + under: { label: "Under pace: headroom left for the rest of the window", icon: TrendingDownIcon }, +}; + +/** The series colour the cost chart uses for this driver, so the two views read as one. */ +function barColor(driver: ServerProvider["driver"]): string { + const kind: UsageProviderKind | undefined = + driver === "codex" ? "codex" : driver === "claudeAgent" ? "claude" : undefined; + return kind ? PROVIDER_PRESENTATION[kind].color : "var(--foreground)"; +} + +/** Pace as a glyph with the words on hover. */ +function PaceIcon({ pace }: { readonly pace: LimitPace }) { + const Icon = PACE[pace].icon; + return ( + + + } + > + + + {PACE[pace].label} + + ); +} + +/** + * One window as a full-width bar from the moment it opened to its reset. + * The fill is the share of quota spent; the hairline is how far into the + * window the clock is, which is also where even spending would have put the + * fill. Hover for the exact figures and reset time. + */ +function WindowBar({ + color, + window, + now, +}: { + readonly color: string; + readonly window: ServerProviderUsageWindow; + readonly now: number; +}) { + const timestampFormat = usePrimarySettings((settings) => settings.timestampFormat); + const used = Math.max(0, Math.min(100, window.usedPercent)); + const elapsed = elapsedShare(window, now); + const resetsIn = formatResetsIn(window, now); + const resetsAt = window.resetsAt + ? formatUpcomingTimestamp(window.resetsAt, timestampFormat, now) + : null; + const summary = `${window.label}: ${Math.round(used)}% used${ + elapsed === null ? "" : `, ${Math.round(elapsed * 100)}% of the window elapsed` + }${resetsIn ? `, ${resetsIn}` : ""}`; + + return ( + + + } + > +
+ {used > 0 ? ( +
+ ) : null} + {elapsed !== null ? ( + + ) : null} + + +
+ + {Math.round(used)}% used + {elapsed !== null ? ` · ${Math.round(elapsed * 100)}% of the window elapsed` : ""} + + {elapsed !== null ? ( + The line is where even spending would be. + ) : null} + {resetsAt ? ( + + Resets {resetsAt} + {resetsIn ? ` · ${resetsIn}` : ""} + + ) : null} +
+ + + ); +} + +/** One account's windows as rows: label and percent, bar, pace and countdown. */ +function LimitWindows({ + driver, + windows, + now, +}: { + readonly driver: ServerProvider["driver"]; + readonly windows: ReadonlyArray; + readonly now: number; +}) { + const color = barColor(driver); + return ( +
+ {windows.map((window, index) => { + // Windows that reset together show the countdown once. + const previous = windows[index - 1]; + const sharesReset = + previous?.resetsAt !== undefined && previous.resetsAt === window.resetsAt; + const pace = paceOf(window, now); + const resetsIn = formatResetsIn(window, now); + return ( + + + {window.label} + + {Math.round(window.usedPercent)}% + + + + + {pace ? : null} + {sharesReset ? "" : (resetsIn ?? "")} + + + ); + })} +
+ ); +} + +/** + * Heading shared by local providers and source accounts: icon, name, plan, + * and the signed-in email blurred until clicked, as provider settings do. + */ +function AccountHeading({ + driver, + label, + plan, + email, + accentColor, + badge, +}: { + readonly driver: ServerProvider["driver"]; + readonly label: string; + readonly plan: string | undefined; + readonly email: string | undefined; + readonly accentColor?: string | undefined; + readonly badge?: string | undefined; +}) { + return ( +

+ + {label} + {plan ? · {plan} : null} + {email ? ( + + ) : null} + {badge ? ( + + {badge} + + ) : null} +

+ ); +} + +function ProviderLimits({ + provider, + now, +}: { + readonly provider: ServerProvider; + readonly now: number; +}) { + const limits = provider.usageLimits; + if (!limits) return null; + const notice = limitsNotice(limits); + return ( +
+ getDriverOption(driver)?.label)} + plan={provider.auth.label} + email={provider.auth.email} + accentColor={provider.accentColor} + /> + {notice ? ( + {notice} + ) : ( + + )} +
+ ); +} + +/** One account pooled by a usage-limit source, drawn like a provider row. */ +function SourceAccountLimits({ + account, + sourceKind, + now, +}: { + readonly account: UsageLimitSourceAccount; + readonly sourceKind: string; + readonly now: number; +}) { + const notice = limitsNotice(account.usageLimits); + return ( +
+ + {notice ? ( + {notice} + ) : ( + + )} +
+ ); +} + +/** + * Accounts a configured source (a CLIProxyAPI hub) pools, grouped under the + * source's name. Unlike provider rows these are read-only: nothing on this + * environment can run a turn against them. + */ +const SOURCE_KIND_LABEL: Record = { + cliproxy: "CLIProxyAPI", +}; + +/** + * Removing a hub also deletes its management key from the server, so it + * asks first and says so. A bare icon that acted on click was too easy to + * hit while reaching for the row beside it. + */ +function RemoveSourceButton({ + source, + onConfirm, +}: { + readonly source: UsageLimitSourceSnapshot; + readonly onConfirm: () => void; +}) { + const [open, setOpen] = useState(false); + return ( + <> + + + + + Remove {source.label}? + + The hub's management key is deleted from this server. Its accounts leave the Limits + view; the hub itself is untouched. Add it again with the URL and key to bring them + back. + + + + }>Cancel + + + + + + ); +} + +function SourceLimits({ + source, + now, + onRemove, +}: { + readonly source: UsageLimitSourceSnapshot; + readonly now: number; + readonly onRemove: (() => void) | null; +}) { + const kind = SOURCE_KIND_LABEL[source.kind]; + return ( +
+
+

+ {source.label} · {kind} +

+ {onRemove ? : null} +
+ {source.error ? ( + {source.error} + ) : source.accounts.length === 0 ? ( + No accounts reported. + ) : ( + source.accounts.map((account) => ( + + )) + )} +
+ ); +} + +/** + * Subscription quota windows from every connected environment's providers. + * Countdowns anchor to render time rather than ticking: a live clock would + * repaint the page every minute for no decision-changing gain. + */ +export function UsageLimitsSection() { + const presentations = useAtomValue(environmentPresentations.presentationsAtom); + const groups = collectLimitsGroups(presentations); + const sources = collectLimitSources(presentations); + const configuredSources = usePrimarySettings((settings) => settings.usageLimitSources); + const canEditSources = usePrimarySettingsAvailable(); + const primaryEnvironmentId = usePrimaryEnvironmentId(); + const updateSettings = useUpdatePrimarySettings(); + const [adding, setAdding] = useState(false); + // Anchored once per mount on purpose: countdowns must not tick (see below). + const [now] = useState(() => Date.now()); + + // The patch names only this entry, so two edits in flight cannot clobber + // each other's map. + const removeSource = (id: UsageLimitSourceId) => { + updateSettings({ usageLimitSources: { [id]: null } }); + }; + + const addHubButton = canEditSources ? ( + + ) : null; + + return ( +
+ {groups.length === 0 && sources.length === 0 ? ( +

+ No provider on a connected environment reports subscription limits. +

+ ) : null} + {sources.map((source) => ( + removeSource(source.id) + : null + } + /> + ))} + {groups.map((group) => ( +
+ {group.environmentLabel ? ( +

+ {group.environmentLabel} +

+ ) : null} + {group.providers.map((provider) => ( + + ))} +
+ ))} + {addHubButton ?
{addHubButton}
: null} + +
+ ); +} diff --git a/apps/web/src/components/usage/UsagePage.tsx b/apps/web/src/components/usage/UsagePage.tsx index 9e4838c0de8b..4b62df374122 100644 --- a/apps/web/src/components/usage/UsagePage.tsx +++ b/apps/web/src/components/usage/UsagePage.tsx @@ -6,7 +6,10 @@ import type { DailyTotals, HourlyTotals } from "@t3tools/shared/usageMerge"; import { isElectron } from "../../env"; import { cn } from "../../lib/utils"; +import { usePrimaryEnvironmentId } from "../../state/environments"; +import { serverEnvironment } from "../../state/server"; import { useUsage, type EnvironmentUsageStatus } from "../../state/usage"; +import { useAtomCommand } from "../../state/use-atom-command"; import { enumerateDays, enumerateHourStarts, @@ -32,9 +35,21 @@ import { } from "../WorkspaceBreadcrumb"; import { WorkspacePageContainer } from "../WorkspacePageContainer"; import { WorkspacePageHeader } from "../WorkspacePageHeader"; +import { UsageLimitsSection } from "./UsageLimits"; import { UsageProviderChart, type UsageChartMetric } from "./UsageProviderChart"; import { PROVIDER_ORDER, PROVIDER_PRESENTATION, providersWithUsage } from "./usageProviders"; +type UsageMetric = UsageChartMetric | "limits"; +const METRIC_OPTIONS = [ + { value: "cost", label: "Cost" }, + { value: "tokens", label: "Tokens" }, + { value: "limits", label: "Limits" }, +] as const satisfies readonly { value: UsageMetric; label: string }[]; + +function isUsageMetric(value: string | null | undefined): value is UsageMetric { + return METRIC_OPTIONS.some((option) => option.value === value); +} + const WINDOW_OPTIONS = [ { days: 1, label: "Past 24h" }, { days: 7, label: "7 days" }, @@ -47,11 +62,16 @@ export function UsagePage() { days: 30, window: makeWindow(30), })); - const [metric, setMetric] = useState("cost"); + const [metric, setMetric] = useState("cost"); + const showingLimits = metric === "limits"; const [breakdown, setBreakdown] = useState<"model" | "time">("model"); const { days: windowDays, window } = windowSelection; const isPast24Hours = windowDays === 1; const { merged, environments, isPending, isPartial, refresh } = useUsage(window); + const primaryEnvironmentId = usePrimaryEnvironmentId(); + const refreshProviders = useAtomCommand(serverEnvironment.refreshProviders, { + reportFailure: false, + }); // Hold the content until every environment is terminal. Rendering merged // totals while devices are still answering makes every number on the page @@ -94,6 +114,15 @@ export function UsagePage() { }); }; const refreshWindow = () => { + // On Limits the button re-probes every provider (and usage-limit source) + // on the primary environment; the live snapshots then flow in over the + // config stream, so nothing else needs to move. + if (showingLimits) { + if (primaryEnvironmentId) { + void refreshProviders({ environmentId: primaryEnvironmentId, input: {} }); + } + return; + } const nextWindow = makeWindow(windowDays, undefined, isPast24Hours ? "hour" : "day"); if ( nextWindow.sinceDay === window.sinceDay && @@ -116,10 +145,14 @@ export function UsagePage() {

Usage

 - - - {windowLabel} - + {showingLimits ? null : ( + <> + + + {windowLabel} + + + )}
{ const value = next[0]; - if (value === "cost" || value === "tokens") setMetric(value); + if (isUsageMetric(value)) setMetric(value); }} > - {(["cost", "tokens"] as const).map((option) => ( - - {option === "cost" ? "Cost" : "Tokens"} + {METRIC_OPTIONS.map((option) => ( + + {option.label} ))} + {/* The period does not apply to Limits, so it stays in place but + disabled; unmounting it shifted the metric toggle ~300px. */} { const value = next[0]; if (value) selectWindow(Number(value)); @@ -152,7 +188,12 @@ export function UsagePage() { ))} -
@@ -160,7 +201,7 @@ export function UsagePage() { - selectWindow(Number(value))} + > -
@@ -209,7 +264,9 @@ export function UsagePage() { - {settling ? ( + {showingLimits ? ( + + ) : settling ? ( <> {environments.length > 1 ? : null} diff --git a/apps/web/src/state/server.ts b/apps/web/src/state/server.ts index 991abc3bc2e2..0eacc933da49 100644 --- a/apps/web/src/state/server.ts +++ b/apps/web/src/state/server.ts @@ -28,6 +28,7 @@ import { environmentSession } from "./session"; export const serverEnvironment = createServerEnvironmentAtoms(connectionAtomRuntime, { initialConfigValueAtom: environmentSession.initialConfigValueAtom, environmentThemes: true, + usageLimitSources: true, }); export const environmentServerConfigsAtom = createEnvironmentServerConfigsAtom({ catalogValueAtom: environmentCatalog.catalogValueAtom, diff --git a/apps/web/src/timestampFormat.ts b/apps/web/src/timestampFormat.ts index 72ba711a1ade..c6a9bdd29e10 100644 --- a/apps/web/src/timestampFormat.ts +++ b/apps/web/src/timestampFormat.ts @@ -159,6 +159,32 @@ export function formatDayAwareTimestamp( return `${dateFormatter.format(date)} ${time}`; } +/** + * The forward-looking counterpart of {@link formatDayAwareTimestamp} for an + * instant that has not happened yet (a usage-limit reset): today `12:34 PM`, + * tomorrow `tomorrow at 12:34 PM`, later `8/13 12:34 PM`. + */ +export function formatUpcomingTimestamp( + isoDate: string, + timestampFormat: TimestampFormat, + nowMs: number = Date.now(), +): string { + const date = parseTimestampDate(isoDate); + if (!date) return ""; + const time = getTimestampFormatter(timestampFormat, false).format(date); + + const now = new Date(nowMs); + const startOfToday = new Date(now.getFullYear(), now.getMonth(), now.getDate()).getTime(); + const startOfTargetDay = new Date(date.getFullYear(), date.getMonth(), date.getDate()).getTime(); + const dayDiff = Math.round((startOfTargetDay - startOfToday) / 86_400_000); + + if (dayDiff <= 0) return time; + if (dayDiff === 1) return `tomorrow at ${time}`; + const dateFormatter = + date.getFullYear() === now.getFullYear() ? numericDateFormatter : numericDateWithYearFormatter; + return `${dateFormatter.format(date)} ${time}`; +} + /** * Format a relative time string from an ISO date. * Returns `{ value: "20s", suffix: "ago" }` or `{ value: "just now", suffix: null }` diff --git a/docs/internals/glossary.md b/docs/internals/glossary.md index e4b70d9185d8..5b58414bea40 100644 --- a/docs/internals/glossary.md +++ b/docs/internals/glossary.md @@ -117,6 +117,14 @@ Controls how assistant text reaches the thread timeline. In [the contracts][1], A point-in-time view of state. The word is used in multiple layers, including orchestration, provider, and checkpointing. See [ProjectionSnapshotQuery.ts][10], [ProviderAdapter.ts][15], and [CheckpointStore.ts][19]. +#### Usage limits + +The rolling subscription quota windows a provider reports for its signed-in account, such as Claude's five-hour and weekly windows or Codex's primary and secondary allowances. Each driver decides in its own `checkProvider` whether it has any and returns them on the snapshot as `usageLimits`; drivers with no notion of subscription usage leave the field absent. Adapters that receive rate-limit telemetry during a turn normalise it into a `ProviderUsageLimitsUpdate` at the boundary, and `ProviderUsageLimitsIngestion` folds it onto the owning instance's snapshot through `ServerProviderShape.applyUsageLimits`, so no central service needs to know a driver kind. See [providerUsageLimits.ts](../../packages/contracts/src/providerUsageLimits.ts) and [makeManagedServerProvider.ts](../../apps/server/src/provider/makeManagedServerProvider.ts). + +#### Usage limit source + +A read-only quota feed outside this environment's provider CLIs, configured under `settings.usageLimitSources`. The only kind today is a CLIProxyAPI hub, whose `quota-scheduler/status` reports the windows of every pooled account. `UsageLimitSources` polls each source on the provider health interval and publishes `UsageLimitSourceSnapshot`s over the config stream as `usageLimitSourcesUpdated`, gated by a client capability flag the way environment themes are. The management key round-trips through the secret store with a redaction marker on disk. See [UsageLimitSources.ts](../../apps/server/src/usage/UsageLimitSources.ts). + #### Model manifest The per-driver list of current model slugs that decides which models land in the model picker's legacy section. Bundled at `apps/server/src/provider/model-manifest.json` and refreshed at runtime from the same file on `main`, so classification updates ship as commits instead of releases. See the [provider architecture][16] model manifest section. diff --git a/docs/user/usage.md b/docs/user/usage.md index 24e2d6a575cf..5b9e474c6aad 100644 --- a/docs/user/usage.md +++ b/docs/user/usage.md @@ -8,6 +8,21 @@ separate from the raw token cost shown here. Grok Build totals come from persisted session updates. Interactive turns that never wrote a completed-turn record will not appear. +The **Limits** view shows how much of each subscription window you have used on Codex and Claude +Code, per connected environment: the session and weekly windows, plus a per-model weekly window +such as Fable when your plan has one. Each window is a bar from the moment it opened to its reset, +filled by the share of quota spent; a thin line marks how far into the window you are, which is +also where even spending would have put the fill, and the icon beside the label says whether you +are ahead of, on, or under that pace. Hover a bar for the exact reset time. Limits refresh on the +provider health-check interval and update live while a turn runs. API-key accounts have no +subscription windows and say so; that includes a Claude Code that reaches Anthropic through a proxy +via `ANTHROPIC_AUTH_TOKEN`, since the CLI then treats itself as an API-key client. + +If you pool accounts behind a CLIProxyAPI hub, **Add CLIProxyAPI hub** on the Limits view shows +every account the hub manages, each marked _via CLIProxyAPI_ so it is not mistaken for the provider +signed in on this machine. Enter the hub's URL and management key; the key is stored on the server +and never sent back to a client. Emails are blurred until clicked, as in provider settings. + Use **Past 24h** for an hourly chart covering the exact rolling 24-hour period. The **7 days**, **30 days**, and **90 days** ranges use daily resolution. Cost and token toggles update both the headline and chart. Refreshing rescans every connected environment and refetches model pricing on diff --git a/packages/client-runtime/src/rpc/session.test.ts b/packages/client-runtime/src/rpc/session.test.ts index f7e945197021..f8f940bc551f 100644 --- a/packages/client-runtime/src/rpc/session.test.ts +++ b/packages/client-runtime/src/rpc/session.test.ts @@ -714,7 +714,9 @@ describe("RpcSessionFactory", () => { clearVcsRefs: () => Effect.void, clear: () => Effect.void, }); - const configState = yield* makeEnvironmentServerConfigState(true).pipe( + const configState = yield* makeEnvironmentServerConfigState({ + environmentThemes: true, + }).pipe( Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), Effect.provideService(Persistence.EnvironmentCacheStore, cache), ); diff --git a/packages/client-runtime/src/state/server.test.ts b/packages/client-runtime/src/state/server.test.ts index 0e6ffb24e790..6567726a4809 100644 --- a/packages/client-runtime/src/state/server.test.ts +++ b/packages/client-runtime/src/state/server.test.ts @@ -590,7 +590,7 @@ describe("server state projection", () => { yield* Effect.scoped( Effect.gen(function* () { - const state = yield* makeEnvironmentServerConfigState().pipe( + const state = yield* makeEnvironmentServerConfigState({}).pipe( Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), Effect.provideService(Persistence.EnvironmentCacheStore, cache), ); @@ -650,7 +650,7 @@ describe("server state projection", () => { }); yield* Effect.scoped( - makeEnvironmentServerConfigState().pipe( + makeEnvironmentServerConfigState({}).pipe( Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), Effect.provideService(Persistence.EnvironmentCacheStore, cache), ), diff --git a/packages/client-runtime/src/state/server.ts b/packages/client-runtime/src/state/server.ts index 120432e8190e..851e67dfa82b 100644 --- a/packages/client-runtime/src/state/server.ts +++ b/packages/client-runtime/src/state/server.ts @@ -353,8 +353,13 @@ const cachedConfigSnapshotEvent = (config: ServerConfig): ServerConfigStreamEven config, }); +export interface ServerConfigSubscriptionOptions { + readonly environmentThemes?: boolean; + readonly usageLimitSources?: boolean; +} + export const makeEnvironmentServerConfigState = Effect.fn("EnvironmentServerConfigState.make")( - function* (environmentThemes?: boolean) { + function* (subscription: ServerConfigSubscriptionOptions) { const supervisor = yield* EnvironmentSupervisor; const cache = yield* EnvironmentCacheStore; const environmentId = supervisor.target.environmentId; @@ -415,10 +420,10 @@ export const makeEnvironmentServerConfigState = Effect.fn("EnvironmentServerConf Effect.forkScoped, ); - yield* subscribe( - WS_METHODS.subscribeServerConfig, - environmentThemes === true ? { environmentThemes: true } : {}, - ).pipe( + yield* subscribe(WS_METHODS.subscribeServerConfig, { + ...(subscription.environmentThemes === true ? { environmentThemes: true } : {}), + ...(subscription.usageLimitSources === true ? { usageLimitSources: true } : {}), + }).pipe( Stream.runForEach((event) => Effect.gen(function* () { const next = applyServerConfigProjection(yield* SubscriptionRef.get(state), event); @@ -450,12 +455,12 @@ export const makeEnvironmentServerConfigState = Effect.fn("EnvironmentServerConf export function serverConfigStateChanges( environmentId: EnvironmentId, - environmentThemes?: boolean, + subscription: ServerConfigSubscriptionOptions, ) { return followStreamInEnvironment( environmentId, Stream.unwrap( - makeEnvironmentServerConfigState(environmentThemes).pipe( + makeEnvironmentServerConfigState(subscription).pipe( Effect.map((state) => SubscriptionRef.changes(state).pipe( Stream.filterMap((projection) => @@ -514,6 +519,8 @@ export function createServerEnvironmentAtoms( * receives the payload. */ readonly environmentThemes?: boolean; + /** Whether this surface renders quota from configured usage-limit sources. */ + readonly usageLimitSources?: boolean; }, ) { const configScheduler = createAtomCommandScheduler(); @@ -525,7 +532,12 @@ export function createServerEnvironmentAtoms( }; const configProjectionFamily = Atom.family((environmentId: EnvironmentId) => runtime - .atom(serverConfigStateChanges(environmentId, options.environmentThemes)) + .atom( + serverConfigStateChanges(environmentId, { + ...(options.environmentThemes === true ? { environmentThemes: true } : {}), + ...(options.usageLimitSources === true ? { usageLimitSources: true } : {}), + }), + ) .pipe( Atom.setIdleTTL(5 * 60_000), Atom.withLabel(`environment-data:server:config-projection:${environmentId}`), diff --git a/packages/client-runtime/src/state/serverConfigProjection.ts b/packages/client-runtime/src/state/serverConfigProjection.ts index 6f4a812cf7e9..77883980332a 100644 --- a/packages/client-runtime/src/state/serverConfigProjection.ts +++ b/packages/client-runtime/src/state/serverConfigProjection.ts @@ -9,12 +9,15 @@ export interface ServerConfigProjection { /** * Cached config keeps the provider and model catalog available across reconnects. - * Published themes are current machine state, so a cache could restore themes - * that the machine no longer publishes. Replay sends themes as a separate event. + * Published themes and usage-limit sources are current machine state, so a + * cache could restore a set the machine no longer reports. Replay sends both + * as separate events. */ export function withoutEnvironmentThemes(config: ServerConfig): ServerConfig { - if (config.environmentThemes === undefined) return config; - const { environmentThemes: _ephemeral, ...rest } = config; + if (config.environmentThemes === undefined && config.usageLimitSources === undefined) { + return config; + } + const { environmentThemes: _themes, usageLimitSources: _sources, ...rest } = config; return rest; } @@ -27,13 +30,21 @@ export function applyServerConfigProjection( // Wire snapshots never contain published themes. Keep the previous set // until a capable server sends its authoritative theme event. A legacy // server cannot send a later removal, so a downgrade must clear the set. - const carried = - event.config.environment.capabilities.environmentThemes === true && Option.isSome(current) + const capabilities = event.config.environment.capabilities; + const carriedThemes = + capabilities.environmentThemes === true && Option.isSome(current) ? current.value.config.environmentThemes : undefined; + const carriedSources = + capabilities.usageLimitSources === true && Option.isSome(current) + ? current.value.config.usageLimitSources + : undefined; return Option.some({ - config: - carried === undefined ? event.config : { ...event.config, environmentThemes: carried }, + config: { + ...event.config, + ...(carriedThemes === undefined ? {} : { environmentThemes: carriedThemes }), + ...(carriedSources === undefined ? {} : { usageLimitSources: carriedSources }), + }, latestEvent: event, source: "live" as const, }); @@ -75,5 +86,14 @@ export function applyServerConfigProjection( latestEvent: event, source: "live", })); + case "usageLimitSourcesUpdated": + return Option.map(current, (projection) => ({ + config: { + ...projection.config, + usageLimitSources: event.payload.sources.length > 0 ? event.payload.sources : undefined, + }, + latestEvent: event, + source: "live", + })); } } diff --git a/packages/contracts/src/environment.ts b/packages/contracts/src/environment.ts index e9a3f1a1bb65..8d03a73b0249 100644 --- a/packages/contracts/src/environment.ts +++ b/packages/contracts/src/environment.ts @@ -102,6 +102,9 @@ export const ExecutionEnvironmentCapabilities = Schema.Struct({ client reconnecting to one must drop published themes rather than keep showing a set nothing will ever update. */ environmentThemes: Schema.optionalKey(Schema.Boolean), + /** Server streams quota from configured usage-limit sources. Same + version-skew contract as environmentThemes. */ + usageLimitSources: Schema.optionalKey(Schema.Boolean), /** Server understands thread.pin / thread.unpin commands. Same version-skew contract as threadSettlement. */ threadPinning: Schema.optionalKey(Schema.Boolean), diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index eb2a476f9e23..cc0cc0d197aa 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -14,6 +14,8 @@ export * from "./provider.ts"; export * from "./providerInstance.ts"; export * from "./providerSetup.ts"; export * from "./providerRuntime.ts"; +export * from "./providerUsageLimits.ts"; +export * from "./usageLimitSourceId.ts"; export * from "./model.ts"; export * from "./keybindings.ts"; export * from "./server.ts"; diff --git a/packages/contracts/src/providerRuntime.ts b/packages/contracts/src/providerRuntime.ts index f2ca5118fe07..28d9d00f64b2 100644 --- a/packages/contracts/src/providerRuntime.ts +++ b/packages/contracts/src/providerRuntime.ts @@ -14,6 +14,7 @@ import { TurnId, } from "./baseSchemas.ts"; import { ProviderInstanceId, ProviderDriverKind } from "./providerInstance.ts"; +import { ProviderUsageLimitsUpdate } from "./providerUsageLimits.ts"; import { ProviderApprovalOption } from "./orchestration.ts"; const TrimmedNonEmptyStringSchema = TrimmedNonEmptyString; @@ -749,8 +750,12 @@ const AccountUpdatedPayload = Schema.Struct({ }); export type AccountUpdatedPayload = typeof AccountUpdatedPayload.Type; +/** + * Adapters normalise their native rate-limit payload at the boundary so the + * consumer that folds it into the provider snapshot never sees driver shapes. + */ const AccountRateLimitsUpdatedPayload = Schema.Struct({ - rateLimits: Schema.Unknown, + limits: ProviderUsageLimitsUpdate, }); export type AccountRateLimitsUpdatedPayload = typeof AccountRateLimitsUpdatedPayload.Type; diff --git a/packages/contracts/src/providerUsageLimits.ts b/packages/contracts/src/providerUsageLimits.ts new file mode 100644 index 000000000000..05a54ea94574 --- /dev/null +++ b/packages/contracts/src/providerUsageLimits.ts @@ -0,0 +1,92 @@ +import * as Schema from "effect/Schema"; + +import { + ForwardCompatibleArray, + IsoDateTime, + NonNegativeInt, + TrimmedNonEmptyString, +} from "./baseSchemas.ts"; +import { ProviderDriverKind } from "./providerInstance.ts"; +import { UsageLimitSourceId } from "./usageLimitSourceId.ts"; + +/** + * One rolling quota window a subscription provider reports for the signed-in + * account, e.g. Claude's five-hour session or Codex's weekly allowance. + * + * `id` is stable per provider (`five_hour`, `seven_day_opus`, `primary`) so a + * sparse turn-driven update lands on the same row a full probe produced. + * `kind` only orders and labels the bar. + */ +export const ServerProviderUsageWindow = Schema.Struct({ + id: TrimmedNonEmptyString, + kind: Schema.Literals(["session", "weekly", "monthly", "other"]), + label: TrimmedNonEmptyString, + usedPercent: Schema.Number.check(Schema.isBetween({ minimum: 0, maximum: 100 })), + resetsAt: Schema.optional(IsoDateTime), + windowDurationMins: Schema.optional(NonNegativeInt), +}); +export type ServerProviderUsageWindow = typeof ServerProviderUsageWindow.Type; + +/** + * Subscription usage the provider knows about the signed-in account. + * + * `unavailable` distinguishes an account that can never report windows (API + * key, Bedrock) from a probe that failed this time, so clients can keep the + * last good bars for the latter and clear them for the former. + */ +export const ServerProviderUsageLimits = Schema.Struct({ + checkedAt: IsoDateTime, + windows: ForwardCompatibleArray(ServerProviderUsageWindow), + unavailable: Schema.optional( + Schema.Struct({ + reason: Schema.Literals(["unsupported", "probeFailed"]), + message: Schema.optional(TrimmedNonEmptyString), + }), + ), +}); +export type ServerProviderUsageLimits = typeof ServerProviderUsageLimits.Type; + +/** + * What an adapter reports when its runtime pushes a rate-limit update during + * a turn. Sparse by contract: Claude's `rate_limit_event` names one window at + * a time and Codex documents its notification as a partial. Windows merge by + * `id` onto the instance's published snapshot; omitted windows are unchanged. + */ +export const ProviderUsageLimitsUpdate = Schema.Struct({ + windows: Schema.Array(ServerProviderUsageWindow), +}); +export type ProviderUsageLimitsUpdate = typeof ProviderUsageLimitsUpdate.Type; + +/** + * One account a usage-limit source reports on. `driver` is the provider the + * account belongs to, for the icon and colour clients already have; the + * account itself is not something this environment can run turns on. + */ +export const UsageLimitSourceAccount = Schema.Struct({ + id: TrimmedNonEmptyString, + driver: ProviderDriverKind, + /** The signed-in address, when the source names one; clients blur it like provider auth. */ + email: Schema.optional(TrimmedNonEmptyString), + /** Plan as the matching provider would label it (`ChatGPT Pro 20x Subscription`). */ + plan: Schema.optional(TrimmedNonEmptyString), + usageLimits: ServerProviderUsageLimits, +}); +export type UsageLimitSourceAccount = typeof UsageLimitSourceAccount.Type; + +/** + * The published state of one configured `usageLimitSources` entry. A source + * that could not be read keeps `error` beside an empty account list rather + * than vanishing, so the user can see it is configured but failing. + */ +export const UsageLimitSourceSnapshot = Schema.Struct({ + id: UsageLimitSourceId, + kind: Schema.Literal("cliproxy"), + label: TrimmedNonEmptyString, + checkedAt: IsoDateTime, + accounts: ForwardCompatibleArray(UsageLimitSourceAccount), + error: Schema.optional(TrimmedNonEmptyString), +}); +export type UsageLimitSourceSnapshot = typeof UsageLimitSourceSnapshot.Type; + +export const UsageLimitSourceSnapshots = ForwardCompatibleArray(UsageLimitSourceSnapshot); +export type UsageLimitSourceSnapshots = typeof UsageLimitSourceSnapshots.Type; diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index 9b2953a6009d..84634664fa37 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -1116,6 +1116,8 @@ export const WsSubscribeServerConfigRpc = Rpc.make(WS_METHODS.subscribeServerCon * dropped by old servers. */ environmentThemes: Schema.optional(Schema.Boolean), + /** Whether this client understands `usageLimitSourcesUpdated` events. */ + usageLimitSources: Schema.optional(Schema.Boolean), }), success: ServerConfigStreamEvent, error: Schema.Union([KeybindingsConfigError, ServerSettingsError, EnvironmentAuthorizationError]), diff --git a/packages/contracts/src/server.test.ts b/packages/contracts/src/server.test.ts index 47f4145207cc..00bb5daa5b86 100644 --- a/packages/contracts/src/server.test.ts +++ b/packages/contracts/src/server.test.ts @@ -157,6 +157,24 @@ describe("server config forward compatibility", () => { expect(parsed).toEqual([decodedBase]); }); + + it("drops usage windows this build cannot decode instead of failing the provider", () => { + const parsed = decodeServerProvider({ + ...baseProviderSnapshot, + usageLimits: { + checkedAt: "2026-04-10T00:00:00.000Z", + windows: [ + { id: "primary", kind: "session", label: "Session", usedPercent: 12 }, + { id: "future", kind: "some-future-kind", label: "Future", usedPercent: 1 }, + { id: "bad", kind: "weekly", label: "Weekly", usedPercent: 120 }, + ], + }, + }); + + expect(parsed.usageLimits?.windows).toEqual([ + { id: "primary", kind: "session", label: "Session", usedPercent: 12 }, + ]); + }); }); describe("resolveEnvironmentMachineKind", () => { diff --git a/packages/contracts/src/server.ts b/packages/contracts/src/server.ts index 5f9e6628b73a..ba0d6679cfc2 100644 --- a/packages/contracts/src/server.ts +++ b/packages/contracts/src/server.ts @@ -24,6 +24,7 @@ import { import { EditorId, FileManagerRevealKind, RemoteOpenTarget } from "./editor.ts"; import { ModelCapabilities } from "./model.ts"; import { ProviderDriverKind, ProviderInstanceId } from "./providerInstance.ts"; +import { ServerProviderUsageLimits, UsageLimitSourceSnapshots } from "./providerUsageLimits.ts"; import { ServerSettings } from "./settings.ts"; const KeybindingsMalformedConfigIssue = Schema.Struct({ @@ -227,6 +228,8 @@ export const ServerProvider = Schema.Struct({ ), skills: Schema.Array(ServerProviderSkill).pipe(Schema.withDecodingDefault(Effect.succeed([]))), workspaceSnapshots: Schema.optionalKey(Schema.Array(ServerProviderWorkspaceSnapshot)), + // Absent when the driver has no notion of subscription usage. + usageLimits: Schema.optional(ServerProviderUsageLimits), versionAdvisory: Schema.optionalKey(ServerProviderVersionAdvisory), updateState: Schema.optionalKey(ServerProviderUpdateState), }); @@ -581,6 +584,12 @@ export const ServerConfig = Schema.Struct({ * and it stays absent for subscribers that did not opt in. */ environmentThemes: Schema.optional(Schema.Array(EnvironmentTheme)), + /** + * Quota reported by configured `usageLimitSources`. Like themes, never in + * a snapshot: the source stream emits the current set on subscribe, and it + * stays absent for subscribers that did not opt in. + */ + usageLimitSources: Schema.optional(UsageLimitSourceSnapshots), }); export type ServerConfig = typeof ServerConfig.Type; @@ -692,12 +701,28 @@ export const ServerConfigStreamEnvironmentThemesUpdatedEvent = Schema.Struct({ export type ServerConfigStreamEnvironmentThemesUpdatedEvent = typeof ServerConfigStreamEnvironmentThemesUpdatedEvent.Type; +export const ServerConfigUsageLimitSourcesUpdatedPayload = Schema.Struct({ + /** The full set; empty once no source is configured. */ + sources: UsageLimitSourceSnapshots, +}); +export type ServerConfigUsageLimitSourcesUpdatedPayload = + typeof ServerConfigUsageLimitSourcesUpdatedPayload.Type; + +export const ServerConfigStreamUsageLimitSourcesUpdatedEvent = Schema.Struct({ + version: Schema.Literal(1), + type: Schema.Literal("usageLimitSourcesUpdated"), + payload: ServerConfigUsageLimitSourcesUpdatedPayload, +}); +export type ServerConfigStreamUsageLimitSourcesUpdatedEvent = + typeof ServerConfigStreamUsageLimitSourcesUpdatedEvent.Type; + export const ServerConfigStreamEvent = Schema.Union([ ServerConfigStreamSnapshotEvent, ServerConfigStreamKeybindingsUpdatedEvent, ServerConfigStreamProviderStatusesEvent, ServerConfigStreamSettingsUpdatedEvent, ServerConfigStreamEnvironmentThemesUpdatedEvent, + ServerConfigStreamUsageLimitSourcesUpdatedEvent, ]); export type ServerConfigStreamEvent = typeof ServerConfigStreamEvent.Type; diff --git a/packages/contracts/src/settings.ts b/packages/contracts/src/settings.ts index 9f1f0c846ac3..275c0328b7f5 100644 --- a/packages/contracts/src/settings.ts +++ b/packages/contracts/src/settings.ts @@ -3,6 +3,7 @@ import * as Duration from "effect/Duration"; import * as Schema from "effect/Schema"; import * as SchemaTransformation from "effect/SchemaTransformation"; import { ForwardCompatibleNullable, TrimmedNonEmptyString, TrimmedString } from "./baseSchemas.ts"; +import { UsageLimitSourceId } from "./usageLimitSourceId.ts"; import { EnvironmentMachineKind, ThreadEnvMode } from "./environment.ts"; import { DEFAULT_TEXT_GENERATION_MODEL, @@ -727,6 +728,21 @@ export const OpenCodeSettings = makeProviderSettingsSchema( ); export type OpenCodeSettings = typeof OpenCodeSettings.Type; +/** + * A read-only quota source outside this environment's provider CLIs. The + * only kind today is a CLIProxyAPI hub, whose management API reports the + * windows of every pooled account. The key travels in settings for now, like + * provider environment secrets; it is redacted before reaching a client. + */ +export const UsageLimitSourceConfig = Schema.Struct({ + kind: Schema.Literal("cliproxy"), + label: Schema.optional(TrimmedNonEmptyString), + url: TrimmedNonEmptyString, + managementKey: TrimmedString.pipe(Schema.withDecodingDefault(Effect.succeed(""))), + enabled: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(true))), +}); +export type UsageLimitSourceConfig = typeof UsageLimitSourceConfig.Type; + export const ObservabilitySettings = Schema.Struct({ otlpTracesUrl: TrimmedString.pipe(Schema.withDecodingDefault(Effect.succeed(""))), otlpMetricsUrl: TrimmedString.pipe(Schema.withDecodingDefault(Effect.succeed(""))), @@ -905,6 +921,11 @@ export const ServerSettings = Schema.Struct({ Schema.withDecodingDefault(Effect.succeed({})), ), observability: ObservabilitySettings.pipe(Schema.withDecodingDefault(Effect.succeed({}))), + // Keyed by a user-chosen id so a source keeps its rows across edits. Entries + // this build cannot decode round-trip untouched, as provider instances do. + usageLimitSources: Schema.Record(UsageLimitSourceId, UsageLimitSourceConfig).pipe( + Schema.withDecodingDefault(Effect.succeed({})), + ), }); export type ServerSettings = typeof ServerSettings.Type; @@ -1110,6 +1131,12 @@ export const ServerSettingsPatch = Schema.Struct({ // patches risk leaving driver-specific config in a half-merged state. // The web UI sends a fully-formed map every time it edits this field. providerInstances: Schema.optionalKey(Schema.Record(ProviderInstanceId, ProviderInstanceConfig)), + // Per-entry, unlike `providerInstances`: a client only ever adds or removes + // one source, and sending the whole map races another edit that has not + // echoed back yet. `null` removes; the server merges into its current map. + usageLimitSources: Schema.optionalKey( + Schema.Record(UsageLimitSourceId, Schema.NullOr(UsageLimitSourceConfig)), + ), }); export type ServerSettingsPatch = typeof ServerSettingsPatch.Type; diff --git a/packages/contracts/src/usageLimitSourceId.ts b/packages/contracts/src/usageLimitSourceId.ts new file mode 100644 index 000000000000..1b323d311814 --- /dev/null +++ b/packages/contracts/src/usageLimitSourceId.ts @@ -0,0 +1,9 @@ +import * as Schema from "effect/Schema"; + +/** + * Key of one `settings.usageLimitSources` entry. Lives in its own module so + * both the settings and the usage-limit contracts can import it without + * importing each other. + */ +export const UsageLimitSourceId = Schema.String.pipe(Schema.brand("UsageLimitSourceId")); +export type UsageLimitSourceId = typeof UsageLimitSourceId.Type; diff --git a/packages/shared/package.json b/packages/shared/package.json index f6d17fa209cd..d9233a2c54c4 100644 --- a/packages/shared/package.json +++ b/packages/shared/package.json @@ -243,6 +243,10 @@ "types": "./src/usageFormat.ts", "import": "./src/usageFormat.ts" }, + "./usageLimits": { + "types": "./src/usageLimits.ts", + "import": "./src/usageLimits.ts" + }, "./desktopAppControl": { "types": "./src/desktopAppControl.ts", "import": "./src/desktopAppControl.ts" diff --git a/packages/shared/src/serverSettings.test.ts b/packages/shared/src/serverSettings.test.ts index baa84a4e1aa8..64b0e1b8337b 100644 --- a/packages/shared/src/serverSettings.test.ts +++ b/packages/shared/src/serverSettings.test.ts @@ -2,6 +2,7 @@ import { DEFAULT_SERVER_SETTINGS, ProviderDriverKind, ProviderInstanceId, + UsageLimitSourceId, type ServerProvider, } from "@t3tools/contracts"; import * as Duration from "effect/Duration"; @@ -298,6 +299,29 @@ describe("serverSettings helpers", () => { }); }); + it("upserts and removes usageLimitSources per entry so concurrent edits cannot clobber", () => { + const hubA = UsageLimitSourceId.make("cliproxy-a"); + const hubB = UsageLimitSourceId.make("cliproxy-b"); + const source = (url: string) => ({ + kind: "cliproxy" as const, + url, + managementKey: "secret", + enabled: true, + }); + const current = { + ...DEFAULT_SERVER_SETTINGS, + usageLimitSources: { [hubA]: source("http://a:8318") }, + }; + + const added = applyServerSettingsPatch(current, { + usageLimitSources: { [hubB]: source("http://b:8318") }, + }); + expect(Object.keys(added.usageLimitSources)).toEqual([hubA, hubB]); + + const removed = applyServerSettingsPatch(added, { usageLimitSources: { [hubA]: null } }); + expect(Object.keys(removed.usageLimitSources)).toEqual([hubB]); + }); + it("stores background activity profiles as a versioned object and syncs legacy aliases", () => { const next = applyServerSettingsPatch(DEFAULT_SERVER_SETTINGS, { backgroundActivity: { diff --git a/packages/shared/src/serverSettings.ts b/packages/shared/src/serverSettings.ts index 69fc9eaacbcc..67f73f84063a 100644 --- a/packages/shared/src/serverSettings.ts +++ b/packages/shared/src/serverSettings.ts @@ -122,6 +122,22 @@ function mergeModelSelectionOptionsById(input: { return [...merged.entries()].map(([id, value]) => ({ id, value })); } +/** Upsert each patched entry; `null` removes it. Entries the patch omits are untouched. */ +function mergeUsageLimitSources( + current: ServerSettings["usageLimitSources"], + patch: NonNullable, +): ServerSettings["usageLimitSources"] { + const next = new Map(Object.entries(current)); + for (const [id, config] of Object.entries(patch)) { + if (config === null) { + next.delete(id); + } else { + next.set(id, config); + } + } + return Object.fromEntries(next) as ServerSettings["usageLimitSources"]; +} + export function applyServerSettingsPatch( current: ServerSettings, patch: ServerSettingsPatch, @@ -132,6 +148,8 @@ export function applyServerSettingsPatch( providerHealthRefreshInterval, backgroundActivityProfile, backgroundActivity, + // Merged per entry below; its `null` removals must not reach deepMerge. + usageLimitSources: usageLimitSourcesPatch, ...patchForMerge } = patch; const currentBackgroundActivity = normalizeServerBackgroundActivitySettings(current); @@ -188,6 +206,14 @@ export function applyServerSettingsPatch( ...(patch.providerInstances !== undefined ? { providerInstances: patch.providerInstances } : {}), + ...(usageLimitSourcesPatch !== undefined + ? { + usageLimitSources: mergeUsageLimitSources( + current.usageLimitSources, + usageLimitSourcesPatch, + ), + } + : {}), ...(patch.sourceControlWriterModelSelection !== undefined ? { sourceControlWriterModelSelection: patch.sourceControlWriterModelSelection } : {}), diff --git a/packages/shared/src/usageLimits.test.ts b/packages/shared/src/usageLimits.test.ts new file mode 100644 index 000000000000..33f180ac9d6c --- /dev/null +++ b/packages/shared/src/usageLimits.test.ts @@ -0,0 +1,181 @@ +import { ProviderDriverKind, ProviderInstanceId, type ServerProvider } from "@t3tools/contracts"; +import { describe, expect, it } from "vite-plus/test"; + +import { + collectLimitSources, + collectLimitsGroups, + elapsedShare, + formatResetsIn, + limitsNotice, + paceOf, + providersWithLimits, +} from "./usageLimits.ts"; + +const now = Date.parse("2026-09-03T12:00:00.000Z"); + +const window = { + id: "five_hour", + kind: "session", + label: "Session", + usedPercent: 40, + windowDurationMins: 300, + resetsAt: "2026-09-03T14:00:00.000Z", +} as const; + +function provider(overrides: Partial): ServerProvider { + return { + instanceId: ProviderInstanceId.make("codex"), + driver: ProviderDriverKind.make("codex"), + enabled: true, + installed: true, + version: null, + status: "ready", + auth: { status: "authenticated" }, + checkedAt: "2026-09-03T11:00:00.000Z", + models: [], + slashCommands: [], + skills: [], + ...overrides, + }; +} + +describe("pace", () => { + it("places the clock three fifths through a five-hour window with two hours left", () => { + expect(elapsedShare(window, now)).toBeCloseTo(0.6); + expect(paceOf(window, now)).toBe("under"); + expect(paceOf({ ...window, usedPercent: 62 }, now)).toBe("on"); + expect(paceOf({ ...window, usedPercent: 80 }, now)).toBe("ahead"); + }); + + it("has no pace without a reset or a duration", () => { + expect(paceOf({ ...window, resetsAt: undefined }, now)).toBeNull(); + expect(paceOf({ ...window, windowDurationMins: undefined }, now)).toBeNull(); + expect(formatResetsIn({ ...window, resetsAt: undefined }, now)).toBeNull(); + }); + + it("phrases the reset as a countdown", () => { + expect(formatResetsIn(window, now)).toBe("resets in 2h 0m"); + expect(formatResetsIn({ ...window, resetsAt: "2026-09-06T15:30:00.000Z" }, now)).toBe( + "resets in 3d 3h", + ); + expect(formatResetsIn({ ...window, resetsAt: "2026-09-03T11:00:00.000Z" }, now)).toBe( + "resets now", + ); + }); +}); + +describe("limitsNotice", () => { + it("explains empty bars and passes provider messages through", () => { + const checkedAt = "2026-09-03T11:00:00.000Z"; + expect(limitsNotice({ checkedAt, windows: [window] })).toBeNull(); + expect(limitsNotice({ checkedAt, windows: [] })).toBe("No limits reported."); + expect(limitsNotice({ checkedAt, windows: [], unavailable: { reason: "unsupported" } })).toBe( + "This account has no subscription limits.", + ); + expect( + limitsNotice({ + checkedAt, + windows: [], + unavailable: { reason: "probeFailed", message: "Codex timed out." }, + }), + ).toBe("Codex timed out."); + }); +}); + +describe("providersWithLimits", () => { + it("keeps only usable providers whose driver reports limits at all", () => { + const limits = { checkedAt: "2026-09-03T11:00:00.000Z", windows: [window] }; + const codex = provider({ usageLimits: limits }); + expect( + providersWithLimits([ + codex, + provider({ + instanceId: ProviderInstanceId.make("cursor"), + driver: ProviderDriverKind.make("cursor"), + }), + provider({ + instanceId: ProviderInstanceId.make("off"), + enabled: false, + usageLimits: limits, + }), + provider({ + instanceId: ProviderInstanceId.make("gone"), + installed: false, + usageLimits: limits, + }), + provider({ + instanceId: ProviderInstanceId.make("shadow"), + availability: "unavailable", + usageLimits: limits, + }), + ]), + ).toEqual([codex]); + }); +}); + +describe("collectLimitsGroups", () => { + it("labels environments only when more than one reports limits", () => { + const limits = { checkedAt: "2026-09-03T11:00:00.000Z", windows: [window] }; + const codex = provider({ usageLimits: limits }); + const one = new Map([ + ["env-a", { entry: { target: { label: "Laptop" } }, serverConfig: { providers: [codex] } }], + [ + "env-b", + { entry: { target: { label: "Desktop" } }, serverConfig: { providers: [provider({})] } }, + ], + ] as const); + expect(collectLimitsGroups(one as never).map((group) => group.environmentLabel)).toEqual([ + null, + ]); + + const two = new Map([ + ["env-a", { entry: { target: { label: "Laptop" } }, serverConfig: { providers: [codex] } }], + ["env-b", { entry: { target: { label: "Desktop" } }, serverConfig: { providers: [codex] } }], + ] as const); + expect(collectLimitsGroups(two as never).map((group) => group.environmentLabel)).toEqual([ + "Laptop", + "Desktop", + ]); + }); +}); + +describe("collectLimitSources", () => { + const source = { + id: "cliproxy-hub" as never, + kind: "cliproxy" as const, + label: "hub", + checkedAt: "2026-09-03T11:00:00.000Z", + accounts: [], + }; + + it("keys sources per environment and names the environment only when several have some", () => { + const one = new Map([ + [ + "env-a", + { entry: { target: { label: "Laptop" } }, serverConfig: { usageLimitSources: [source] } }, + ], + [ + "env-b", + { entry: { target: { label: "Desktop" } }, serverConfig: { usageLimitSources: [] } }, + ], + ] as const); + expect(collectLimitSources(one as never).map((entry) => [entry.key, entry.label])).toEqual([ + ["env-a:cliproxy-hub", "hub"], + ]); + + const two = new Map([ + [ + "env-a", + { entry: { target: { label: "Laptop" } }, serverConfig: { usageLimitSources: [source] } }, + ], + [ + "env-b", + { entry: { target: { label: "Desktop" } }, serverConfig: { usageLimitSources: [source] } }, + ], + ] as const); + expect(collectLimitSources(two as never).map((entry) => entry.label)).toEqual([ + "Laptop · hub", + "Desktop · hub", + ]); + }); +}); diff --git a/packages/shared/src/usageLimits.ts b/packages/shared/src/usageLimits.ts new file mode 100644 index 000000000000..3c5b39ea06a7 --- /dev/null +++ b/packages/shared/src/usageLimits.ts @@ -0,0 +1,178 @@ +/** + * Selection and pace maths for the provider limits view, shared by web and + * mobile so both agree on which providers show, what "ahead of pace" means, + * and how a reset is phrased. + * + * @module usageLimits + */ +import { + type EnvironmentId, + isProviderAvailable, + type ServerProvider, + type ServerProviderUsageLimits, + type ServerProviderUsageWindow, + type UsageLimitSourceSnapshot, + type UsageLimitSourceSnapshots, +} from "@t3tools/contracts"; + +const MINUTE = 60_000; +const HOUR = 60 * MINUTE; +const DAY = 24 * HOUR; + +/** + * Providers that belong on the Limits view: enabled, installed, and one whose + * driver reports subscription usage at all. A driver with no notion of usage + * never sets `usageLimits`, so it has no row rather than an empty one. + */ +export function providersWithLimits( + providers: readonly ServerProvider[], +): readonly ServerProvider[] { + return providers.filter( + (provider) => + provider.enabled && + provider.installed && + isProviderAvailable(provider) && + provider.usageLimits !== undefined, + ); +} + +export interface LimitsGroup { + readonly environmentId: EnvironmentId; + /** Null while only one environment is connected; there is nothing to tell apart. */ + readonly environmentLabel: string | null; + readonly providers: readonly ServerProvider[]; +} + +/** + * One group per connected environment with a provider reporting limits. + * Provider snapshots come from the config stream every client already holds, + * so opening the view costs no extra request. + */ +export function collectLimitsGroups( + presentations: ReadonlyMap< + EnvironmentId, + { + readonly entry: { readonly target: { readonly label: string } }; + readonly serverConfig: { readonly providers: readonly ServerProvider[] } | null; + } + >, +): readonly LimitsGroup[] { + const groups: LimitsGroup[] = []; + for (const [environmentId, presentation] of presentations) { + const providers = providersWithLimits(presentation.serverConfig?.providers ?? []); + if (providers.length === 0) continue; + groups.push({ environmentId, environmentLabel: presentation.entry.target.label, providers }); + } + return groups.length > 1 ? groups : groups.map((group) => ({ ...group, environmentLabel: null })); +} + +/** + * Every usage-limit source across connected environments, keyed so two + * environments pointing at the same hub still get their own rows. The label + * carries the environment only when more than one environment has sources. + */ +export function collectLimitSources( + presentations: ReadonlyMap< + EnvironmentId, + { + readonly entry: { readonly target: { readonly label: string } }; + readonly serverConfig: { + readonly usageLimitSources?: UsageLimitSourceSnapshots | undefined; + } | null; + } + >, +): ReadonlyArray< + UsageLimitSourceSnapshot & { readonly key: string; readonly environmentId: EnvironmentId } +> { + const perEnvironment: Array<{ + readonly environmentId: EnvironmentId; + readonly environmentLabel: string; + readonly sources: UsageLimitSourceSnapshots; + }> = []; + for (const [environmentId, presentation] of presentations) { + const sources = presentation.serverConfig?.usageLimitSources ?? []; + if (sources.length === 0) continue; + perEnvironment.push({ + environmentId, + environmentLabel: presentation.entry.target.label, + sources, + }); + } + const labelEnvironment = perEnvironment.length > 1; + return perEnvironment.flatMap(({ environmentId, environmentLabel, sources }) => + sources.map((source) => ({ + ...source, + environmentId, + key: `${environmentId}:${source.id}`, + label: labelEnvironment ? `${environmentLabel} · ${source.label}` : source.label, + })), + ); +} + +/** The instance's configured name, else the driver's, else its raw kind. */ +export function providerLimitsLabel( + provider: ServerProvider, + driverLabel: (driver: ServerProvider["driver"]) => string | undefined, +): string { + return provider.displayName?.trim() || driverLabel(provider.driver) || String(provider.driver); +} + +/** The one-line status under a provider heading when there are no bars to draw. */ +export function limitsNotice(limits: ServerProviderUsageLimits): string | null { + if (limits.unavailable?.reason === "unsupported") { + return limits.unavailable.message ?? "This account has no subscription limits."; + } + if (limits.unavailable?.reason === "probeFailed") { + return limits.unavailable.message ?? "Could not read limits."; + } + return limits.windows.length === 0 ? "No limits reported." : null; +} + +export function resetMillis(window: ServerProviderUsageWindow): number | null { + if (window.resetsAt === undefined) return null; + const at = Date.parse(window.resetsAt); + return Number.isFinite(at) ? at : null; +} + +/** Elapsed share of the window, 0..1, or null when its length or reset is unknown. */ +export function elapsedShare(window: ServerProviderUsageWindow, now: number): number | null { + const resetsAt = resetMillis(window); + if (resetsAt === null || window.windowDurationMins === undefined) return null; + const length = window.windowDurationMins * MINUTE; + if (length <= 0) return null; + return Math.max(0, Math.min(1, (length - (resetsAt - now)) / length)); +} + +export type LimitPace = "ahead" | "on" | "under"; + +/** + * Usage against the clock. The bar is the whole window, so the elapsed share + * is also where even spending would have put the fill; within five points of + * it counts as on pace. + */ +export function paceOf(window: ServerProviderUsageWindow, now: number): LimitPace | null { + const elapsed = elapsedShare(window, now); + if (elapsed === null) return null; + const gap = window.usedPercent - elapsed * 100; + if (gap > 5) return "ahead"; + if (gap < -5) return "under"; + return "on"; +} + +/** `2h 13m`, `3d 4h`, `12m`. */ +export function formatDuration(ms: number): string { + const remaining = Math.max(0, ms); + const days = Math.floor(remaining / DAY); + const hours = Math.floor((remaining % DAY) / HOUR); + const minutes = Math.floor((remaining % HOUR) / MINUTE); + if (days > 0) return `${days}d ${hours}h`; + if (hours > 0) return `${hours}h ${minutes}m`; + return `${minutes}m`; +} + +/** `resets in 2h 13m`, or null when the window has no reset. */ +export function formatResetsIn(window: ServerProviderUsageWindow, now: number): string | null { + const resetsAt = resetMillis(window); + if (resetsAt === null) return null; + return resetsAt <= now ? "resets now" : `resets in ${formatDuration(resetsAt - now)}`; +}