diff --git a/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts b/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts index 7f1006f1c5ec5..5116590c07322 100644 --- a/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts +++ b/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts @@ -47,6 +47,7 @@ import * as ProjectionStore from "./ProjectionStore.ts"; import * as ProviderEventIngestor from "./ProviderEventIngestor.ts"; import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts"; import * as ProviderInstanceRegistry from "../provider/ProviderInstanceRegistry.ts"; +import * as UsageLimitSources from "../usage/UsageLimitSources.ts"; import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; import type { ProviderInstance } from "@t3tools/provider-core/server/driver"; import { makeProviderFailure } from "@t3tools/provider-core/server/failure"; @@ -233,6 +234,17 @@ it.effect("refreshes the failing instance snapshot after a committed usage_limit Effect.scoped( Effect.gen(function* () { const refreshed = yield* Deferred.make(); + const hubResetAt = "2026-10-30T05:00:00.000Z"; + const nativeResetAt = "2026-10-30T03:00:00.000Z"; + const fallbackLookups: Array<{ instanceId: ProviderInstanceId; driver: ProviderDriverKind }> = + []; + const layerResetFallback = Layer.succeed(UsageLimitSources.UsageLimitResetFallback, { + resetAt: (lookup) => + Effect.sync(() => { + fallbackLookups.push(lookup); + return hubResetAt; + }), + }); const adapter = yield* ProviderAdapter.ProviderAdapterV2.pipe( Effect.provide( Layer.mock(ProviderAdapter.ProviderAdapterV2)({ @@ -290,7 +302,11 @@ it.effect("refreshes the failing instance snapshot after a committed usage_limit providerInstanceId: instance.instanceId, threadId: threadEvent.threadId, }; - for (const failureClass of ["validation_error", "usage_limit"] as const) { + for (const [failureClass, resetAt, expectedResetAt] of [ + ["validation_error", undefined, undefined], + ["usage_limit", undefined, hubResetAt], + ["usage_limit", nativeResetAt, nativeResetAt], + ] as const) { const stored = yield* ingestor.ingestNormalized({ ...input, event: { @@ -302,7 +318,7 @@ it.effect("refreshes the failing instance snapshot after a committed usage_limit }), providerTurnId: idAllocator.derive.providerTurn({ driver: instance.driverKind, - nativeTurnId: failureClass, + nativeTurnId: `${failureClass}:${resetAt ?? "unknown"}`, }), runOrdinal: 1, failureItemOrdinal: 1, @@ -310,16 +326,27 @@ it.effect("refreshes the failing instance snapshot after a committed usage_limit failure: makeProviderFailure({ class: failureClass, message: "Provider rejected turn.", + ...(resetAt === undefined ? {} : { resetAt }), }), threadDisposition: "reusable", }, }); assert.equal(stored.length, 1); + const event = stored[0]?.event; + const item = event?.type === "turn-item.updated" ? event.payload : undefined; + assert.equal(item?.type === "error" ? item.failure.resetAt : null, expectedResetAt); if (failureClass === "validation_error") assert.deepEqual(lookups, []); } - assert.deepEqual(lookups, [instance.instanceId]); + assert.deepEqual(lookups, [instance.instanceId, instance.instanceId]); + assert.deepEqual(fallbackLookups, [ + { instanceId: instance.instanceId, driver: instance.driverKind }, + ]); yield* Deferred.await(refreshed); - }).pipe(Effect.provide(makeLayerTest(layerTestEventSink, instances))); + }).pipe( + Effect.provide( + makeLayerTest(layerTestEventSink, instances).pipe(Layer.provide(layerResetFallback)), + ), + ); }), ), ); diff --git a/apps/server/src/orchestration-v2/ProviderEventIngestor.ts b/apps/server/src/orchestration-v2/ProviderEventIngestor.ts index 5a2aa39faece4..e77b1889610fe 100644 --- a/apps/server/src/orchestration-v2/ProviderEventIngestor.ts +++ b/apps/server/src/orchestration-v2/ProviderEventIngestor.ts @@ -35,6 +35,7 @@ import { getModelSelectionStringOptionValue, modelSelectionsEqual } from "@t3too import * as AnalyticsService from "../telemetry/AnalyticsService.ts"; import * as GitVcsDriver from "../vcs/GitVcsDriver.ts"; import * as ProviderInstanceRegistry from "../provider/ProviderInstanceRegistry.ts"; +import { UsageLimitResetFallback } from "../usage/UsageLimitSources.ts"; import * as EventSink from "./EventSink.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; @@ -339,6 +340,7 @@ export const layer: Layer.Layer< const analytics = yield* ProviderTurnAnalytics; const agentScope = yield* AgentScope; const instances = yield* ProviderInstanceRegistry.ProviderInstanceRegistry; + const usageLimitResets = yield* UsageLimitResetFallback; const scope = yield* Effect.scope; const completedTurnAnalytics = new Set(); @@ -711,9 +713,18 @@ export const layer: Layer.Layer< return dismissed; } const occurredAt = yield* DateTime.now; - const failure = (yield* agentScope.oomKilled(input.threadId)) + const reported = (yield* agentScope.oomKilled(input.threadId)) ? { ...input.event.failure, message: "Killed: out of memory.", code: "oom_kill" } : input.event.failure; + const fallbackResetAt = + reported.class === "usage_limit" && reported.resetAt == null + ? yield* usageLimitResets.resetAt({ + instanceId: input.providerInstanceId, + driver: input.event.driver, + }) + : null; + const failure = + fallbackResetAt === null ? reported : { ...reported, resetAt: fallbackResetAt }; return [ ...dismissed, yield* makeDomainEvent(input, { diff --git a/apps/server/src/provider/claudeUsageLimits.ts b/apps/server/src/provider/claudeUsageLimits.ts index d0acf43bb4968..0196cbe9ad7ca 100644 --- a/apps/server/src/provider/claudeUsageLimits.ts +++ b/apps/server/src/provider/claudeUsageLimits.ts @@ -47,8 +47,15 @@ export const makeClaudeScopedLimitNames = Ref.make({ overageIncluded: undefined, }); +const SCOPED_WINDOW_PREFIX = "seven_day_"; + function scopedWindowId(displayName: string): string { - return `seven_day_${displayName.toLowerCase().replace(/[^a-z0-9]+/g, "_")}`; + return `${SCOPED_WINDOW_PREFIX}${displayName.toLowerCase().replace(/[^a-z0-9]+/g, "_")}`; +} + +/** Model-scoped allowances leave the account usable for other models. */ +export function isClaudeModelScopedWindow(window: Pick): boolean { + return window.id.startsWith(SCOPED_WINDOW_PREFIX); } function scopedWindow( diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 452addfb4a667..8fa6f22065674 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -521,6 +521,7 @@ const layerScheduledTaskWebhookOrigin = Layer.effect( const layerOrchestrationV2Runtime = RuntimeLayer.layerProduction.pipe( Layer.provide(layerScheduledTaskWebhookOrigin), Layer.provide(ProviderEventIngestor.layerAnalytics), + Layer.provide(UsageLimitSources.layerResetFallback), Layer.provide(layerCheckpointStore), Layer.provide(layerGitWorkflow), Layer.provide( diff --git a/apps/server/src/usage/UsageLimitSources.test.ts b/apps/server/src/usage/UsageLimitSources.test.ts index b8976fca81e7d..0783a2ff063c8 100644 --- a/apps/server/src/usage/UsageLimitSources.test.ts +++ b/apps/server/src/usage/UsageLimitSources.test.ts @@ -1,10 +1,21 @@ import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; -import { expect, it } from "@effect/vitest"; -import { ServerConfigStreamEvent, UsageLimitSourceId } from "@t3tools/contracts"; +import { describe, expect, it } from "@effect/vitest"; +import { + ProviderDriverKind, + ProviderInstanceId, + ServerConfigStreamEvent, + UsageLimitSourceId, + type ServerProviderUsageWindow, + type UsageLimitSourceSnapshot, +} from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; +import * as Deferred from "effect/Deferred"; +import * as Fiber from "effect/Fiber"; +import * as TestClock from "effect/testing/TestClock"; import * as Layer from "effect/Layer"; import * as Schema from "effect/Schema"; -import { HttpClient } from "effect/http"; +import { HttpClient, HttpClientResponse } from "effect/http"; import * as BackgroundPolicy from "../background/BackgroundPolicy.ts"; import * as ServerSettings from "../serverSettings.ts"; @@ -83,3 +94,205 @@ it.effect("keeps malformed source URLs encodable for config subscribers", () => Effect.scoped, ), ); + +it.effect("reads the selected hub while an unrelated refresh is still waiting", () => + Effect.gen(function* () { + const stalled = yield* Deferred.make(); + const release = yield* Deferred.make(); + const http = HttpClient.make((request) => + Effect.gen(function* () { + if (new URL(request.url).hostname === "slow.test") { + yield* Deferred.succeed(stalled, undefined); + yield* Deferred.await(release); + } + const body = request.url.endsWith("/auth-files") + ? { files: [{ id: "account", auth_index: "one", provider: "claude" }] } + : { + status_code: 200, + body: JSON.stringify({ + five_hour: { utilization: 100, resets_at: "2099-01-01T00:00:00.000Z" }, + }), + }; + return HttpClientResponse.fromWeb(request, Response.json(body)); + }), + ); + const sources = yield* UsageLimitSources.make.pipe( + Effect.provideService(HttpClient.HttpClient, http), + ); + yield* Deferred.await(stalled); + const fallback = yield* UsageLimitSources.UsageLimitResetFallback.pipe( + Effect.provide(UsageLimitSources.layerResetFallback), + Effect.provideService(UsageLimitSources.UsageLimitSources, sources), + ); + const result = yield* fallback + .resetAt({ + instanceId: ProviderInstanceId.make("proxy"), + driver: ProviderDriverKind.make("claudeAgent"), + }) + .pipe(Effect.forkChild); + yield* TestClock.adjust("5 seconds"); + expect(yield* Fiber.join(result)).toBe("2099-01-01T00:00:00.000Z"); + expect(yield* Deferred.isDone(release)).toBe(false); + }).pipe( + Effect.provide([ + ServerSettings.layerTest({ + providerInstances: { + [ProviderInstanceId.make("proxy")]: { + driver: ProviderDriverKind.make("claudeAgent"), + environment: [ + { name: "ANTHROPIC_BASE_URL", value: "https://fast.test", sensitive: false }, + ], + }, + }, + usageLimitSources: Object.fromEntries( + ["fast", "slow"].map((id) => [ + id, + { + kind: "cliproxy" as const, + url: `https://${id}.test`, + managementKey: "key", + enabled: true, + }, + ]), + ), + }), + Layer.mock(BackgroundPolicy.BackgroundPolicy)({ + shouldRunScopeWork: () => Effect.succeed(false), + }), + NodeCrypto.layer, + ]), + Effect.scoped, + ), +); + +describe("hubPoolResetAt", () => { + const now = Date.parse("2026-10-30T00:00:00.000Z"); + const hours = (count: number) => DateTime.formatIso(DateTime.makeUnsafe(now + count * 3_600_000)); + const window = ( + usedPercent: number, + resetsAt?: string, + id = resetsAt ?? `window-${usedPercent}`, + ): ServerProviderUsageWindow => ({ + id, + kind: "session", + label: "5h", + usedPercent, + ...(resetsAt === undefined ? {} : { resetsAt }), + }); + const hub = ( + ...accounts: ReadonlyArray]> + ): UsageLimitSourceSnapshot => ({ + id: UsageLimitSourceId.make("hub"), + kind: "cliproxy", + label: "hub", + checkedAt: hours(0), + accounts: accounts.map(([driver, windows], index) => ({ + id: `account-${index}`, + driver: ProviderDriverKind.make(driver), + usageLimits: { checkedAt: hours(0), windows }, + })), + }); + + it.each([ + { + name: "the first spent account to recover, after all its spent windows reset", + sources: [ + hub( + ["codex", [window(100, hours(2))]], + ["codex", [window(100, hours(1)), window(100, hours(3))]], + ), + hub(["claudeAgent", [window(100, hours(0.5))]]), + ], + expected: hours(2), + }, + { + name: "nothing while an account still has quota", + sources: [hub(["codex", [window(100, hours(2))]], ["codex", [window(40, hours(1))]])], + expected: null, + }, + { + name: "nothing once a spent window's reset has passed", + sources: [hub(["codex", [window(100, hours(2))]], ["codex", [window(100, hours(-1))]])], + expected: null, + }, + { + name: "a recovery time while skipping accounts without readings or reset times", + sources: [hub(["codex", []], ["codex", [window(100)]], ["codex", [window(100, hours(4))]])], + expected: hours(4), + }, + { + name: "a recovery time that ignores spent model-scoped allowances", + driver: "claudeAgent", + sources: [ + hub([ + "claudeAgent", + [window(100, hours(2), "five_hour"), window(100, hours(90), "seven_day_fable")], + ]), + ], + expected: hours(2), + }, + { + name: "nothing when the hub pools none of the driver's accounts", + sources: [hub(["claudeAgent", [window(100, hours(1))]])], + expected: null, + }, + ])("returns $name", ({ sources, expected, driver = "codex" }) => { + expect(UsageLimitSources.hubPoolResetAt(sources, ProviderDriverKind.make(driver), now)).toBe( + expected, + ); + }); +}); + +describe("instanceHubIds", () => { + const source = (url: string, enabled = true) => ({ + kind: "cliproxy" as const, + url, + managementKey: "key", + enabled, + }); + const settings = { + usageLimitSources: { + [UsageLimitSourceId.make("fleet")]: source("https://hub.test:8443/"), + [UsageLimitSourceId.make("local")]: source("http://127.0.0.1:8317"), + [UsageLimitSourceId.make("off")]: source("https://off.test", false), + [UsageLimitSourceId.make("plain")]: source("https://plain.test"), + }, + providerInstances: { + [ProviderInstanceId.make("claudeProxy")]: { + driver: ProviderDriverKind.make("claudeAgent"), + environment: [ + { name: "ANTHROPIC_BASE_URL", value: "https://HUB.test:8443", sensitive: false }, + ], + }, + [ProviderInstanceId.make("codexProxy")]: { + driver: ProviderDriverKind.make("codex"), + config: { + launchArgs: `-c 'model_providers.p={base_url="http://127.0.0.1:8317/v1"}' https://off.test`, + }, + }, + [ProviderInstanceId.make("defaultPort")]: { + driver: ProviderDriverKind.make("claudeAgent"), + environment: [ + { name: "ANTHROPIC_BASE_URL", value: "https://plain.test:443/v1", sensitive: false }, + { name: "OTHER_URL", value: "http://hub.test:8443", sensitive: false }, + ], + }, + [ProviderInstanceId.make("otherPort")]: { + driver: ProviderDriverKind.make("codex"), + config: { launchArgs: `-c 'model_providers.p={base_url="http://127.0.0.1:83170/v1"}'` }, + }, + }, + }; + + it.each([ + ["claudeProxy", ["fleet"]], + ["codexProxy", ["local"]], + ["defaultPort", ["plain"]], + ["otherPort", []], + ["missing", []], + ])("matches %s to the enabled hubs its settings name", (instanceId, expected) => { + expect([ + ...UsageLimitSources.instanceHubIds(settings, ProviderInstanceId.make(instanceId)), + ]).toEqual(expected); + }); +}); diff --git a/apps/server/src/usage/UsageLimitSources.ts b/apps/server/src/usage/UsageLimitSources.ts index 7a068cde85f6c..5e5d8a3dd6805 100644 --- a/apps/server/src/usage/UsageLimitSources.ts +++ b/apps/server/src/usage/UsageLimitSources.ts @@ -3,6 +3,8 @@ import { UsageLimitSourceError, type UsageLimitSourceConsumeResetCreditInput, type ProviderConsumeResetCreditResult, + type ProviderDriverKind, + type ProviderInstanceId, type ServerSettings, type UsageLimitSourceConfig, type UsageLimitSourceId, @@ -21,6 +23,7 @@ import * as Semaphore from "effect/Semaphore"; import * as Stream from "effect/Stream"; import * as BackgroundPolicy from "../background/BackgroundPolicy.ts"; +import { isClaudeModelScopedWindow } from "../provider/claudeUsageLimits.ts"; import * as Settings from "../serverSettings.ts"; import { makeCliproxyApi } from "./cliproxyApi.ts"; @@ -30,6 +33,9 @@ export class UsageLimitSources extends Context.Service< readonly current: Effect.Effect>; readonly streamChanges: Stream.Stream>; readonly refresh: Effect.Effect; + readonly read: ( + sourceIds: ReadonlySet, + ) => Effect.Effect>; readonly consumeResetCredit: ( input: UsageLimitSourceConsumeResetCreditInput, ) => Effect.Effect; @@ -100,6 +106,19 @@ export const make = Effect.gen(function* () { yield* publish(snapshots); }).pipe(refreshLock.withPermits(1), Effect.ignoreCause({ log: true })); + // Targeted reads do not publish, so unrelated refreshes and redemptions keep cache ownership. + const read = Effect.fn("UsageLimitSources.read")(function* (sourceIds: ReadonlySet) { + const settings = yield* settingsService.getSettings.pipe(Effect.option); + const entries = Object.entries( + settings._tag === "Some" ? settings.value.usageLimitSources : {}, + ).filter(([id, config]) => sourceIds.has(id) && config.enabled); + return yield* Effect.forEach( + entries, + ([id, config]) => readSource(id as UsageLimitSourceId, config), + { concurrency: 4 }, + ); + }); + // Shares the refresh lock so a stale in-flight read cannot overwrite a redemption. const consumeResetCredit = Effect.fn("UsageLimitSources.consumeResetCredit")(function* ( input: UsageLimitSourceConsumeResetCreditInput, @@ -151,6 +170,7 @@ export const make = Effect.gen(function* () { return { current: Ref.get(stateRef), + read, consumeResetCredit, refresh, get streamChanges() { @@ -168,3 +188,93 @@ export const make = Effect.gen(function* () { }); export const layer = Layer.effect(UsageLimitSources, make); + +/** + * A hub answers for a 429 only when every account it pools for the driver is spent; + * the pool recovers when its first account does. + */ +export function hubPoolResetAt( + sources: ReadonlyArray, + driver: ProviderDriverKind, + nowMs: number, +): string | null { + let earliest: number | null = null; + for (const account of sources.flatMap((source) => source.accounts)) { + if (account.driver !== driver) continue; + const windows = account.usageLimits.windows.filter( + (window) => !isClaudeModelScopedWindow(window), + ); + const exhausted = windows.filter((window) => window.usedPercent >= 100); + if (exhausted.length === 0) { + if (windows.length > 0) return null; + continue; + } + const resets = exhausted.map((window) => Date.parse(window.resetsAt ?? "")); + if (resets.some((reset) => !Number.isFinite(reset))) continue; + const recoversAt = Math.max(...resets); + if (recoversAt <= nowMs) return null; + earliest = earliest === null ? recoversAt : Math.min(earliest, recoversAt); + } + return earliest === null ? null : DateTime.formatIso(DateTime.makeUnsafe(earliest)); +} + +/** CLIProxyAPI serves agents and management on one address, so an instance uses the hubs it names. */ +export function instanceHubIds( + settings: Pick, + instanceId: ProviderInstanceId, +): ReadonlySet { + const instance = settings.providerInstances[instanceId]; + if (!instance) return new Set(); + const origins = new Set( + ( + JSON.stringify([instance.environment ?? [], instance.config ?? null]).match( + /https?:\/\/[^\s"'\\]+/gi, + ) ?? [] + ).flatMap((url) => (URL.canParse(url) ? [new URL(url).origin] : [])), + ); + return new Set( + Object.entries(settings.usageLimitSources).flatMap(([id, config]) => + config.enabled && URL.canParse(config.url) && origins.has(new URL(config.url).origin) + ? [id] + : [], + ), + ); +} + +/** Supplies a reset time for usage limits a provider reports without one. */ +export class UsageLimitResetFallback extends Context.Reference<{ + readonly resetAt: (input: { + readonly instanceId: ProviderInstanceId; + readonly driver: ProviderDriverKind; + }) => Effect.Effect; +}>("t3/usage/UsageLimitResetFallback", { + defaultValue: () => ({ resetAt: () => Effect.succeed(null) }), +}) {} + +export const layerResetFallback = Layer.effect( + UsageLimitResetFallback, + Effect.gen(function* () { + const sources = yield* UsageLimitSources; + const settingsService = yield* Settings.ServerSettingsService; + return { + resetAt: Effect.fn("UsageLimitResetFallback.resetAt")(function* (input: { + readonly instanceId: ProviderInstanceId; + readonly driver: ProviderDriverKind; + }) { + const settings = yield* settingsService.getSettings.pipe(Effect.option); + if (settings._tag === "None") return null; + const hubIds = instanceHubIds(settings.value, input.instanceId); + if (hubIds.size === 0) return null; + // The terminal waits on this read, so a slow hub falls back to its last snapshot. + const refreshed = yield* sources.read(hubIds).pipe(Effect.timeoutOption("5 seconds")); + const snapshots = refreshed._tag === "Some" ? refreshed.value : yield* sources.current; + const now = yield* DateTime.now; + return hubPoolResetAt( + snapshots.filter((source) => hubIds.has(source.id)), + input.driver, + DateTime.toEpochMillis(now), + ); + }), + }; + }), +); diff --git a/docs/user/usage.md b/docs/user/usage.md index 447924a072eff..615ec85bed06e 100644 --- a/docs/user/usage.md +++ b/docs/user/usage.md @@ -157,6 +157,13 @@ This connection supplies usage information; configure the provider separately to send agent requests through the hub. Remove the hub from the same settings section when you no longer need it. +CLIProxyAPI does not pass its reset time through Claude Code or Codex. When a Claude or Codex +account sends requests through the hub, its environment variables or launch arguments name the +hub's address. If such an account stops on a usage limit without a reset time, and every account +the hub pools for that provider is spent, the thread resets when the first of those accounts +refills. **Resume at reset**, automatic +resume, and snooze then work as they do for direct subscriptions. + ## Subscription usage widget Add **Subscription usage** from your iOS or Android widget gallery to see remaining Codex and