diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts index 8697505ef24..7261dc7db48 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts @@ -1995,6 +1995,64 @@ describe("ClaudeAdapterLive", () => { ); }); + it.effect("normalizes Claude rate limit events onto canonical usage windows", () => { + const harness = makeHarness(); + return Effect.gen(function* () { + const adapter = yield* ClaudeAdapter; + const runtimeEvents: Array = []; + const runtimeEventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) => + Effect.sync(() => runtimeEvents.push(event)), + ).pipe(Effect.forkChild); + + yield* adapter.startSession({ + threadId: THREAD_ID, + provider: ProviderDriverKind.make("claudeAgent"), + runtimeMode: "full-access", + }); + + harness.query.emit({ + type: "rate_limit_event", + rate_limit_info: { + status: "allowed_warning", + rateLimitType: "five_hour", + utilization: 82, + resetsAt: 1_800_000_000, + }, + session_id: "session", + uuid: "rate-limit-1", + } as unknown as SDKMessage); + // A bare status carries no usage figures and must not invent a window. + harness.query.emit({ + type: "rate_limit_event", + rate_limit_info: { status: "allowed" }, + session_id: "session", + uuid: "rate-limit-2", + } as unknown as SDKMessage); + yield* Effect.yieldNow; + yield* Effect.yieldNow; + + const rateLimitEvents = runtimeEvents.filter( + (event) => event.type === "account.rate-limits.updated", + ); + assert.deepEqual( + rateLimitEvents.map((event) => + event.type === "account.rate-limits.updated" ? event.payload : undefined, + ), + [ + { + status: "warning", + windows: [{ kind: "five_hour", usedPercent: 82, resetsAt: 1_800_000_000 }], + }, + { status: "allowed", windows: [] }, + ], + ); + runtimeEventsFiber.interruptUnsafe(); + }).pipe( + Effect.provideService(Random.Random, makeDeterministicRandomService()), + Effect.provide(harness.layer), + ); + }); + it.effect("consumes undeclared and UX-internal system subtypes without warning rows", () => { 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 812a7310928..8f08ab4d13f 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.ts @@ -15,6 +15,7 @@ import { type PermissionUpdate, type SDKMessage, type SDKControlGetContextUsageResponse, + type SDKRateLimitInfo, type SDKResultMessage, type SettingSource, type SDKUserMessage, @@ -22,6 +23,7 @@ import { } from "@anthropic-ai/claude-agent-sdk"; import { parseCliArgs } from "@t3tools/shared/cliArgs"; import { + type AccountRateLimitsUpdatedPayload, ApprovalRequestId, type CanonicalItemType, type CanonicalRequestType, @@ -1500,6 +1502,35 @@ function sdkMessageSubtype(value: unknown): string | undefined { return typeof record.subtype === "string" ? record.subtype : undefined; } +const CLAUDE_RATE_LIMIT_STATUS = { + allowed: "allowed", + allowed_warning: "warning", + rejected: "rejected", +} as const satisfies Record; + +/** + * Normalizes the SDK rate-limit snapshot onto the canonical payload. Claude + * reports the one window currently governing the account rather than a full + * set, so `windows` holds at most one entry, and none at all when the SDK + * sends a bare status. `utilization` is a 0-100 percentage, matching the + * `rate_limits` windows the SDK documents on its usage response. + */ +function rateLimitsPayloadFromSdk(info: SDKRateLimitInfo): AccountRateLimitsUpdatedPayload { + const window = + info.rateLimitType !== undefined && info.utilization !== undefined + ? { + kind: info.rateLimitType, + usedPercent: info.utilization, + ...(info.resetsAt !== undefined ? { resetsAt: info.resetsAt } : {}), + } + : undefined; + + return { + status: CLAUDE_RATE_LIMIT_STATUS[info.status], + windows: window ? [window] : [], + }; +} + function sdkNativeMethod(message: SDKMessage): string { const subtype = sdkMessageSubtype(message); if (subtype) { @@ -3446,9 +3477,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( yield* offerRuntimeEvent({ ...base, type: "account.rate-limits.updated", - payload: { - rateLimits: message, - }, + payload: rateLimitsPayloadFromSdk(message.rate_limit_info), }); return; } diff --git a/apps/server/src/provider/Layers/CodexAdapter.test.ts b/apps/server/src/provider/Layers/CodexAdapter.test.ts index 7b8fbec5666..c557f1ad90d 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.test.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.test.ts @@ -723,6 +723,78 @@ lifecycleLayer("CodexAdapterLive lifecycle", (it) => { }), ); + it.effect("normalizes Codex rate limit notifications onto canonical usage windows", () => + Effect.gen(function* () { + const { adapter, runtime } = yield* startLifecycleRuntime(); + const firstEventFiber = yield* Stream.runHead(adapter.streamEvents).pipe(Effect.forkChild); + + yield* runtime.emit({ + id: asEventId("evt-rate-limits"), + kind: "notification", + provider: ProviderDriverKind.make("codex"), + threadId: asThreadId("thread-1"), + createdAt: "2026-01-01T00:00:00.000Z", + method: "account/rateLimits/updated", + payload: { + rateLimits: { + primary: { usedPercent: 40, resetsAt: 1_800_000_000, windowDurationMins: 300 }, + secondary: { usedPercent: 12 }, + }, + }, + } satisfies ProviderEvent); + + const firstEvent = yield* Fiber.join(firstEventFiber); + + NodeAssert.equal(firstEvent._tag, "Some"); + if (firstEvent._tag !== "Some") { + return; + } + NodeAssert.equal(firstEvent.value.type, "account.rate-limits.updated"); + if (firstEvent.value.type !== "account.rate-limits.updated") { + return; + } + // No exhaustion signal in the snapshot: status stays absent rather than + // asserting an `allowed` Codex never reported. + NodeAssert.deepEqual(firstEvent.value.payload, { + windows: [ + { kind: "primary", usedPercent: 40, resetsAt: 1_800_000_000, windowDurationMins: 300 }, + { kind: "secondary", usedPercent: 12 }, + ], + }); + }), + ); + + it.effect("treats Codex spend-control exhaustion as a rejected account", () => + Effect.gen(function* () { + const { adapter, runtime } = yield* startLifecycleRuntime(); + const firstEventFiber = yield* Stream.runHead(adapter.streamEvents).pipe(Effect.forkChild); + + yield* runtime.emit({ + id: asEventId("evt-spend-control"), + kind: "notification", + provider: ProviderDriverKind.make("codex"), + threadId: asThreadId("thread-1"), + createdAt: "2026-01-01T00:00:00.000Z", + method: "account/rateLimits/updated", + payload: { + rateLimits: { spendControlReached: true }, + }, + } satisfies ProviderEvent); + + const firstEvent = yield* Fiber.join(firstEventFiber); + + NodeAssert.equal(firstEvent._tag, "Some"); + if (firstEvent._tag !== "Some") { + return; + } + NodeAssert.equal(firstEvent.value.type, "account.rate-limits.updated"); + if (firstEvent.value.type !== "account.rate-limits.updated") { + return; + } + NodeAssert.deepEqual(firstEvent.value.payload, { status: "rejected", windows: [] }); + }), + ); + it.effect("maps retryable Codex error notifications to runtime.warning", () => Effect.gen(function* () { const { adapter, runtime } = yield* startLifecycleRuntime(); diff --git a/apps/server/src/provider/Layers/CodexAdapter.ts b/apps/server/src/provider/Layers/CodexAdapter.ts index 6b99bf52b1e..094dc4ea68a 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.ts @@ -8,6 +8,7 @@ * @module CodexAdapterLive */ import { + type AccountRateLimitsUpdatedPayload, type CanonicalItemType, type CanonicalRequestType, type CodexSettings, @@ -436,6 +437,47 @@ function providerRefsFromEvent( return Object.keys(refs).length > 0 ? (refs as ProviderRuntimeEvent["providerRefs"]) : undefined; } +function rateLimitWindow( + kind: string, + window: EffectCodexSchema.V2AccountRateLimitsUpdatedNotification__RateLimitWindow | null, +): AccountRateLimitsUpdatedPayload["windows"][number] | undefined { + if (!window) { + return undefined; + } + return { + kind, + usedPercent: window.usedPercent, + ...(window.resetsAt != null ? { resetsAt: window.resetsAt } : {}), + ...(window.windowDurationMins != null ? { windowDurationMins: window.windowDurationMins } : {}), + }; +} + +/** + * Normalizes a Codex rate-limit snapshot onto the canonical payload. Codex + * reports two windows at once and signals exhaustion out of band, through + * `rateLimitReachedType` and `spendControlReached`, rather than as a status on + * the windows themselves. + * + * These notifications are sparse: an omitted field means "unknown", not + * "recovered" — the generated schema says as much on `spendControlReached`. + * So a status is claimed only on positive evidence of exhaustion, and left + * absent otherwise rather than asserting `allowed` the snapshot never stated. + */ +function rateLimitsPayloadFromNotification( + snapshot: EffectCodexSchema.V2AccountRateLimitsUpdatedNotification__RateLimitSnapshot, +): AccountRateLimitsUpdatedPayload { + const windows = [ + rateLimitWindow("primary", snapshot.primary ?? null), + rateLimitWindow("secondary", snapshot.secondary ?? null), + ].filter((window) => window !== undefined); + const exhausted = snapshot.rateLimitReachedType != null || snapshot.spendControlReached === true; + + return { + ...(exhausted ? { status: "rejected" as const } : {}), + windows, + }; +} + function runtimeEventBase( event: ProviderEvent, canonicalThreadId: ThreadId, @@ -1393,16 +1435,18 @@ function mapToRuntimeEvents( } if (event.method === "account/rateLimits/updated") { - if (!readPayload(EffectCodexSchema.V2AccountRateLimitsUpdatedNotification, event.payload)) { + const payload = readPayload( + EffectCodexSchema.V2AccountRateLimitsUpdatedNotification, + event.payload, + ); + if (!payload) { return []; } return [ { type: "account.rate-limits.updated", ...runtimeEventBase(event, canonicalThreadId), - payload: { - rateLimits: event.payload ?? {}, - }, + payload: rateLimitsPayloadFromNotification(payload.rateLimits), }, ]; } diff --git a/packages/contracts/src/providerRuntime.ts b/packages/contracts/src/providerRuntime.ts index bd525e6542e..ab7f7e797d9 100644 --- a/packages/contracts/src/providerRuntime.ts +++ b/packages/contracts/src/providerRuntime.ts @@ -699,8 +699,40 @@ const AccountUpdatedPayload = Schema.Struct({ }); export type AccountUpdatedPayload = typeof AccountUpdatedPayload.Type; +/** + * Whether the account may currently send work. `warning` means the provider + * accepted this turn but flagged the window as close to exhausted. Absent + * means the update did not report one — providers send sparse snapshots, and + * silence is not recovery. + */ +const RateLimitStatus = Schema.Literals(["allowed", "warning", "rejected"]); +export type RateLimitStatus = typeof RateLimitStatus.Type; + +/** + * One rolling usage window on a provider account. Providers disagree on how + * many they report — Claude sends the single governing window, Codex sends a + * primary/secondary pair — so consumers read the array, not fixed fields. + */ +const RateLimitWindow = Schema.Struct({ + /** Provider-native window name, e.g. `five_hour`, `seven_day`, `primary`. */ + kind: TrimmedNonEmptyStringSchema, + /** Share of the window consumed, 0-100. */ + usedPercent: Schema.Number, + /** Unix epoch seconds at which the window resets. */ + resetsAt: Schema.optional(Schema.Number), + /** Window length in minutes, when the provider reports one. */ + windowDurationMins: Schema.optional(Schema.Number), +}); +export type RateLimitWindow = typeof RateLimitWindow.Type; + const AccountRateLimitsUpdatedPayload = Schema.Struct({ - rateLimits: Schema.Unknown, + status: Schema.optional(RateLimitStatus), + /** + * The windows this update reported, empty when it carried none. Updates are + * sparse, so a window missing here is unchanged rather than cleared — merge + * by `kind` instead of replacing wholesale. + */ + windows: Schema.Array(RateLimitWindow), }); export type AccountRateLimitsUpdatedPayload = typeof AccountRateLimitsUpdatedPayload.Type;