From 895f1fef42ccdc5fc885fa6a6339ed5fc68f4472 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 13 Nov 2026 16:39:30 +0000 Subject: [PATCH 1/4] fix(server): proxied usage limits take their reset time from the CLIProxyAPI hub Through a CLIProxyAPI hub the Claude CLI drops the hub's Retry-After and the hub filters Codex's rate-limit snapshot, so usage_limit failures were stored with resetAt null and auto-resume and snooze-until-reset never armed. Providers now report the origin of their custom endpoint (ANTHROPIC_BASE_URL for Claude, the custom route for Codex), and each usage limit source reports its hub origin. When a usage_limit failure arrives without a reset time, the ingestor asks the hub at the instance's endpoint: if every account of that driver is spent, the reset is the earliest refill among them. An account that could not be read, or still has room, leaves the reset unknown. A Claude model-scoped weekly window counts only for its own model. Fixes pingdotgg/t3code#26574 Co-Authored-By: Claude Opus 5.5 --- .../ProviderEventIngestor.test.ts | 32 +++++- .../orchestration-v2/ProviderEventIngestor.ts | 34 +++++- apps/server/src/provider/ClaudeProvider.ts | 5 +- .../server/src/provider/CodexProvider.test.ts | 9 ++ apps/server/src/provider/CodexProvider.ts | 33 ++++-- apps/server/src/provider/claudeUsageLimits.ts | 14 ++- apps/server/src/server.ts | 1 + .../src/usage/UsageLimitSources.test.ts | 107 +++++++++++++++++- apps/server/src/usage/UsageLimitSources.ts | 66 ++++++++++- packages/contracts/src/providerUsageLimits.ts | 2 + packages/contracts/src/server.ts | 2 + 11 files changed, 288 insertions(+), 17 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts b/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts index f69444d23571..ebe5fba1efb0 100644 --- a/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts +++ b/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts @@ -49,8 +49,10 @@ import * as ProviderEventIngestor from "./ProviderEventIngestor.ts"; import * as ProviderUsageAnalytics from "../telemetry/ProviderUsageAnalytics.ts"; import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts"; import * as ProviderInstanceRegistry from "../provider/ProviderInstanceRegistry.ts"; +import { UsageLimitResetFallback } from "../usage/UsageLimitSources.ts"; import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; import type { ProviderInstance } from "@t3tools/provider-core/server/driver"; +import type { ServerProvider } from "@t3tools/contracts"; import { makeProviderFailure } from "@t3tools/provider-core/server/failure"; import { makeProviderEventRoutingState, @@ -230,7 +232,7 @@ function threadCreatedEvent( const layer = it.layer(layerTest); -it.effect("refreshes the failing instance snapshot after a committed usage_limit terminal", () => +it.effect("takes a hub's reset for an unreset usage_limit, then refreshes the snapshot", () => Effect.gen(function* () { const refreshed = yield* Deferred.make(); const adapter = yield* ProviderAdapter.ProviderAdapterV2.pipe( @@ -261,7 +263,9 @@ it.effect("refreshes the failing instance snapshot after a committed usage_limit }, snapshot: { refreshUsage: Deferred.succeed(refreshed, undefined).pipe(Effect.asVoid), - getSnapshot: Effect.die("unused"), + getSnapshot: Effect.succeed({ + auth: { status: "authenticated", endpoint: "http://hub.test:8317" }, + } as ServerProvider), refresh: Effect.die("unused"), resolveMaintenance: () => Effect.die("unused"), applyUsageLimits: () => Effect.die("unused"), @@ -269,6 +273,7 @@ it.effect("refreshes the failing instance snapshot after a committed usage_limit }, }; const lookups: ProviderInstanceId[] = []; + const hubLookups: Array<[string, string, string | null]> = []; const instances = Layer.mock(ProviderInstanceRegistry.ProviderInstanceRegistry)({ getInstance: (id) => Effect.sync(() => { @@ -317,10 +322,29 @@ it.effect("refreshes the failing instance snapshot after a committed usage_limit }); assert.equal(stored.length, 1); if (failureClass === "validation_error") assert.deepEqual(lookups, []); + else { + const event = stored[0]?.event; + assert.equal( + event?.type === "turn-item.updated" && event.payload.type === "error" + ? event.payload.failure.resetAt + : undefined, + "2026-11-13T18:00:00.000Z", + ); + } } - assert.deepEqual(lookups, [instance.instanceId]); + assert.deepEqual(lookups, [instance.instanceId, instance.instanceId]); + assert.deepEqual(hubLookups, [["cursor", "http://hub.test:8317", "gpt-5.4"]]); yield* Deferred.await(refreshed); - }).pipe(Effect.provide(makeLayerTest(layerTestEventSink, instances))); + }).pipe( + Effect.provide(makeLayerTest(layerTestEventSink, instances)), + Effect.provideService(UsageLimitResetFallback, { + resetAt: (driver, endpoint, model) => + Effect.sync(() => { + hubLookups.push([driver, endpoint, model]); + return "2026-11-13T18:00:00.000Z"; + }), + }), + ); }), ); diff --git a/apps/server/src/orchestration-v2/ProviderEventIngestor.ts b/apps/server/src/orchestration-v2/ProviderEventIngestor.ts index 6fd6e7bc3129..fddffcd6ee19 100644 --- a/apps/server/src/orchestration-v2/ProviderEventIngestor.ts +++ b/apps/server/src/orchestration-v2/ProviderEventIngestor.ts @@ -32,6 +32,7 @@ import { } from "../telemetry/ProviderUsageAnalytics.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"; @@ -254,6 +255,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(); @@ -656,9 +658,39 @@ 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; + // A proxied CLI can drop the reset time; the hub still knows when its pool refills. + const endpoint = + reported.class === "usage_limit" && reported.resetAt == null + ? (yield* Effect.flatMap( + instances.getInstance(input.providerInstanceId), + (instance) => instance?.snapshot.getSnapshot ?? Effect.undefined, + ))?.auth.endpoint + : undefined; + const hubResetAt = + endpoint === undefined + ? null + : yield* Effect.gen(function* () { + // The failing run's model, which may differ from a selection changed mid-turn. + const { runs } = + input.runId === undefined + ? { runs: [] } + : yield* projections.getThreadRecords(input.threadId, ["runs"], { + runIds: [input.runId], + }); + return ( + runs[0]?.modelSelection ?? + (yield* projections.getThread(input.threadId)).modelSelection + ).model; + }).pipe( + Effect.orElseSucceed(() => null), + Effect.flatMap((model) => + usageLimitResets.resetAt(input.event.driver, endpoint, model), + ), + ); + const failure = hubResetAt === null ? reported : { ...reported, resetAt: hubResetAt }; return [ ...dismissed, yield* makeDomainEvent(input, { diff --git a/apps/server/src/provider/ClaudeProvider.ts b/apps/server/src/provider/ClaudeProvider.ts index f969544fad06..18885f0417eb 100644 --- a/apps/server/src/provider/ClaudeProvider.ts +++ b/apps/server/src/provider/ClaudeProvider.ts @@ -776,6 +776,7 @@ export const checkClaudeProviderStatus = Effect.fn("checkClaudeProviderStatus")( (yield* readClaudeSettingsEnv(claudeSettings, cwd, resolvedEnvironment)).ANTHROPIC_BASE_URL ?? resolvedEnvironment.ANTHROPIC_BASE_URL; const customApiEndpoint = customClaudeApiEndpointHost(baseUrl) !== undefined; + const endpoint = customApiEndpoint ? URL.parse(baseUrl?.trim() ?? "")?.origin : undefined; if (!customApiEndpoint && claudeAuthStatus(capabilities) === "unauthenticated") { const path = yield* Path.Path; const configDir = (yield* makeClaudeEnvironment(claudeSettings, resolvedEnvironment)) @@ -804,7 +805,9 @@ export const checkClaudeProviderStatus = Effect.fn("checkClaudeProviderStatus")( // A proxy's ANTHROPIC_AUTH_TOKEN reads as an API key; the endpoint is what the token is for. const authMetadata = apiProviderAuthMetadata(capabilities.apiProvider) ?? - (customApiEndpoint ? { type: "customEndpoint", label: "Custom API endpoint" } : undefined) ?? + (customApiEndpoint + ? { type: "customEndpoint", label: "Custom API endpoint", ...(endpoint ? { endpoint } : {}) } + : undefined) ?? claudeAuthMetadata({ subscriptionType: capabilities.subscriptionType, authMethod: capabilities.tokenSource, diff --git a/apps/server/src/provider/CodexProvider.test.ts b/apps/server/src/provider/CodexProvider.test.ts index 7eadf550eab2..ad15bf0896eb 100644 --- a/apps/server/src/provider/CodexProvider.test.ts +++ b/apps/server/src/provider/CodexProvider.test.ts @@ -2,6 +2,7 @@ import { assert, it } from "@effect/vitest"; import { applyPreferredCodexDefaultModel, + codexCustomEndpoint, codexRoutesSpendNativeQuota, mapCodexModelCapabilities, } from "./CodexProvider.ts"; @@ -192,4 +193,12 @@ it("reports native quota only for an OpenAI login on a native route", () => { name, ); } + // The custom endpoint is what a usage-limit hub is matched against. + assert.strictEqual(codexCustomEndpoint(hub), "https://hub.example.com"); + assert.strictEqual( + codexCustomEndpoint({ openai_base_url: "http://127.0.0.1:8317/v1" }), + "http://127.0.0.1:8317", + ); + assert.strictEqual(codexCustomEndpoint({ openai_base_url: "https://chatgpt.com/x" }), undefined); + assert.strictEqual(codexCustomEndpoint({}), undefined); }); diff --git a/apps/server/src/provider/CodexProvider.ts b/apps/server/src/provider/CodexProvider.ts index f3bbf514a83e..df105d3ba20f 100644 --- a/apps/server/src/provider/CodexProvider.ts +++ b/apps/server/src/provider/CodexProvider.ts @@ -94,17 +94,21 @@ const CodexCyberAccessConfig = Schema.Struct({ }); const decodeCodexCyberAccessConfig = Schema.decodeUnknownOption(CodexCyberAccessConfig); -function routesToNativeOpenai(config: unknown): boolean { +// Null is Codex's auth-dependent OpenAI default; undefined is a route that cannot be read. +function codexRouteBaseUrl(config: unknown): string | null | undefined { const decoded = decodeCodexRoutingConfig(config); - if (Option.isNone(decoded)) return false; + if (Option.isNone(decoded)) return undefined; const routing = decoded.value; const providerId = routing.model_provider ?? "openai"; + if (providerId === "openai") return routing.openai_base_url || null; const provider = routing.model_providers?.[providerId]; - if (providerId !== "openai" && !provider) return false; - const baseUrl = - providerId === "openai" ? routing.openai_base_url || undefined : provider?.base_url; - // A provider without an override uses Codex's auth-dependent OpenAI default. - if (baseUrl == null) return true; + return provider ? (provider.base_url ?? null) : undefined; +} + +function routesToNativeOpenai(config: unknown): boolean { + const baseUrl = codexRouteBaseUrl(config); + if (baseUrl === undefined) return false; + if (baseUrl === null) return true; if (!URL.canParse(baseUrl)) return false; const url = new URL(baseUrl); return ( @@ -113,6 +117,15 @@ function routesToNativeOpenai(config: unknown): boolean { ); } +/** The origin of the custom endpoint Codex sends model requests to, if it uses one. */ +export function codexCustomEndpoint(config: unknown): string | undefined { + const baseUrl = codexRouteBaseUrl(config); + const url = baseUrl ? URL.parse(baseUrl) : null; + return url && /^https?:$/.test(url.protocol) && !routesToNativeOpenai(config) + ? url.origin + : undefined; +} + // A stored login does not mean this provider's traffic spends that login's quota. export const codexRoutesSpendNativeQuota = ( account: CodexSchema.V2GetAccountResponse, @@ -140,6 +153,7 @@ const CODEX_PRESENTATION = { export interface CodexAppServerProviderSnapshot { readonly account: CodexSchema.V2GetAccountResponse; + readonly endpoint?: string | undefined; readonly nativeUsage?: boolean; readonly rateLimits?: CodexRateLimitsProbe; readonly version: string | undefined; @@ -757,6 +771,7 @@ const probeCodexAppServerProvider = Effect.fn("probeCodexAppServerProvider")(fun const cyberAccess = Option.getOrUndefined(decodeCodexCyberAccessConfig(config)); return { account: accountResponse, + endpoint: config ? codexCustomEndpoint(config) : undefined, nativeUsage, ...(rateLimits ? { rateLimits } : {}), version, @@ -1075,7 +1090,9 @@ export const checkCodexProviderStatus = Effect.fn("checkCodexProviderStatus")(fu installed: true, version: snapshot.version ?? null, status: snapshot.models === undefined ? "warning" : accountStatus.status, - auth: accountStatus.auth, + auth: snapshot.endpoint + ? { ...accountStatus.auth, endpoint: snapshot.endpoint } + : accountStatus.auth, ...(accountStatus.message ? { message: accountStatus.message } : snapshot.models === undefined diff --git a/apps/server/src/provider/claudeUsageLimits.ts b/apps/server/src/provider/claudeUsageLimits.ts index d0acf43bb496..08da2812a642 100644 --- a/apps/server/src/provider/claudeUsageLimits.ts +++ b/apps/server/src/provider/claudeUsageLimits.ts @@ -47,8 +47,20 @@ export const makeClaudeScopedLimitNames = Ref.make({ overageIncluded: undefined, }); +const SCOPED_WINDOW_PREFIX = "seven_day_"; +const scopeToken = (name: string) => name.toLowerCase().replace(/[^a-z0-9]+/g, "_"); + function scopedWindowId(displayName: string): string { - return `seven_day_${displayName.toLowerCase().replace(/[^a-z0-9]+/g, "_")}`; + return `${SCOPED_WINDOW_PREFIX}${scopeToken(displayName)}`; +} + +/** A model-scoped weekly allowance blocks only its own model; an unknown model is not blocked. */ +export function windowLimitsModel( + window: Pick, + model: string | null, +): boolean { + if (!window.id.startsWith(SCOPED_WINDOW_PREFIX)) return true; + return model !== null && scopeToken(model).includes(window.id.slice(SCOPED_WINDOW_PREFIX.length)); } function scopedWindow( diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index dae95787afe8..9353fa2f508a 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -574,6 +574,7 @@ const layerScheduledTaskWebhookOrigin = Layer.effect( ); const layerOrchestrationV2Runtime = RuntimeLayer.layerProduction.pipe( + Layer.provide(UsageLimitSources.layerResetFallback), Layer.provide(layerScheduledTaskWebhookOrigin), Layer.provide(ProviderUsageAnalytics.layer), Layer.provide(layerCheckpointStore), diff --git a/apps/server/src/usage/UsageLimitSources.test.ts b/apps/server/src/usage/UsageLimitSources.test.ts index ae3e468d099e..6724d6cf5699 100644 --- a/apps/server/src/usage/UsageLimitSources.test.ts +++ b/apps/server/src/usage/UsageLimitSources.test.ts @@ -1,7 +1,15 @@ import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; import { expect, it } from "@effect/vitest"; -import { SecretProviderId, ServerConfigStreamEvent, UsageLimitSourceId } from "@t3tools/contracts"; +import { + ProviderDriverKind, + SecretProviderId, + ServerConfigStreamEvent, + UsageLimitSourceId, + type ServerProviderUsageWindow, + type UsageLimitSourceSnapshot, +} from "@t3tools/contracts"; import * as SecretResolver from "@t3tools/secret-provider-core/server/SecretResolver"; +import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Schema from "effect/Schema"; @@ -26,6 +34,14 @@ it.effect("keeps malformed source URLs encodable for config subscribers", () => "hub.test:8317", "Usage limit source", ]); + // Only an http(s) hub URL names an endpoint an agent can be sent to. + expect(snapshots.map((source) => source.origin)).toEqual([ + undefined, + undefined, + undefined, + "https://hub.test:8317", + undefined, + ]); expect(() => encodeConfigEvent({ version: 1, @@ -155,3 +171,92 @@ it.effect("reads a management key from its secret provider", () => { ]), ); }); + +const NOW = Date.parse("2026-11-13T12:00:00.000Z"); +const at = (hours: number) => DateTime.formatIso(DateTime.makeUnsafe(NOW + hours * 3_600_000)); +const spent = (id: string, hours: number): ServerProviderUsageWindow => ({ + id, + kind: "weekly", + label: id, + usedPercent: 100, + resetsAt: at(hours), +}); +const hub = ( + accounts: ReadonlyArray<{ + driver?: string; + windows?: ReadonlyArray; + unavailable?: true; + }>, +): UsageLimitSourceSnapshot => ({ + id: UsageLimitSourceId.make("hub"), + kind: "cliproxy", + label: "hub", + checkedAt: at(0), + accounts: accounts.map((account, index) => ({ + id: `account-${index}`, + driver: ProviderDriverKind.make(account.driver ?? "claudeAgent"), + usageLimits: { + checkedAt: at(0), + windows: account.windows ?? [], + ...(account.unavailable ? { unavailable: { reason: "probeFailed" as const } } : {}), + }, + })), +}); + +it.each([ + { + name: "earliest refill when every account is spent, each at its latest spent window", + accounts: [ + { windows: [spent("five_hour", 1), spent("seven_day", 3)] }, + { windows: [spent("five_hour", 2)] }, + { driver: "codex", windows: [] }, + ], + model: "claude-haiku-5-5", + expected: at(2), + }, + { + name: "unknown while an account still has room", + accounts: [{ windows: [spent("five_hour", 1)] }, { windows: [] }], + model: "claude-haiku-5-5", + expected: null, + }, + { + name: "unknown while an account could not be read", + accounts: [{ windows: [spent("five_hour", 1)] }, { unavailable: true as const }], + model: "claude-haiku-5-5", + expected: null, + }, + { + name: "unknown when a spent window has no reset or already reset", + accounts: [{ windows: [{ ...spent("five_hour", 1), resetsAt: undefined }] }], + model: "claude-haiku-5-5", + expected: null, + }, + { + name: "a model-scoped weekly window blocks only its own model", + accounts: [{ windows: [spent("seven_day_fable", 5)] }], + model: "claude-haiku-5-5", + expected: null, + }, + { + name: "a model-scoped weekly window blocks its own model", + accounts: [{ windows: [spent("seven_day_fable", 5)] }], + model: "claude-fable-5-2", + expected: at(5), + }, + { + name: "no account of the driver", + accounts: [{ driver: "codex", windows: [spent("five_hour", 1)] }], + model: "claude-haiku-5-5", + expected: null, + }, +])("hub pool reset: $name", ({ accounts, model, expected }) => { + expect( + UsageLimitSources.hubPoolResetAt( + [hub(accounts)], + ProviderDriverKind.make("claudeAgent"), + model, + NOW, + ), + ).toBe(expected); +}); diff --git a/apps/server/src/usage/UsageLimitSources.ts b/apps/server/src/usage/UsageLimitSources.ts index 7d40b0fb9ece..96bcedaa0968 100644 --- a/apps/server/src/usage/UsageLimitSources.ts +++ b/apps/server/src/usage/UsageLimitSources.ts @@ -3,6 +3,7 @@ import { UsageLimitSourceError, type UsageLimitSourceConsumeResetCreditInput, type ProviderConsumeResetCreditResult, + type ProviderDriverKind, type ServerSettings, UsageLimitSourceConfig, settingsSecretLiteral, @@ -26,6 +27,7 @@ import * as SecretResolver from "@t3tools/secret-provider-core/server/SecretReso import * as SecretResolverLayer from "../secrets/SecretResolver.ts"; import { resolveSettingsSecrets } from "../secrets/settingsSecretValue.ts"; import * as Settings from "../serverSettings.ts"; +import { windowLimitsModel } from "../provider/claudeUsageLimits.ts"; import { makeCliproxyApi } from "./cliproxyApi.ts"; export class UsageLimitSources extends Context.Service< @@ -79,7 +81,14 @@ export const make = Effect.gen(function* () { config: UsageLimitSourceConfig, ) { const checkedAt = DateTime.formatIso(yield* DateTime.now); - const base = { id, kind: config.kind, label: sourceLabel(id, config), checkedAt } as const; + const url = URL.parse(config.url); + const base = { + id, + kind: config.kind, + label: sourceLabel(id, config), + ...(url && /^https?:$/.test(url.protocol) ? { origin: url.origin } : {}), + checkedAt, + } as const; const managementKey = yield* readManagementKey(config); if (managementKey.length === 0) { return { ...base, accounts: [], error: missingManagementKey(config) }; @@ -198,3 +207,58 @@ export const make = Effect.gen(function* () { export const layer = Layer.effect(UsageLimitSources, make).pipe( Layer.provide(SecretResolverLayer.layer), ); + +/** A hub answers with a usage limit only when every account it routes the driver to is spent. */ +export function hubPoolResetAt( + sources: ReadonlyArray, + driver: ProviderDriverKind, + model: string | null, + now: number, +): string | null { + let earliest: number | null = null; + for (const source of sources) { + for (const account of source.accounts) { + if (account.driver !== driver) continue; + // An unread account may be the one still blocked, so its refill is unknown. + if (account.usageLimits.unavailable) return null; + const spent = account.usageLimits.windows.filter( + (window) => window.usedPercent >= 100 && windowLimitsModel(window, model), + ); + if (spent.length === 0) return null; + const resets = spent.map((window) => Date.parse(window.resetsAt ?? "")); + if (resets.some((reset) => !Number.isFinite(reset))) return null; + const recoversAt = Math.max(...resets); + if (recoversAt <= now) return null; + earliest = earliest === null ? recoversAt : Math.min(earliest, recoversAt); + } + } + return earliest === null ? null : DateTime.formatIso(DateTime.makeUnsafe(earliest)); +} + +export class UsageLimitResetFallback extends Context.Reference<{ + readonly resetAt: ( + driver: ProviderDriverKind, + endpoint: string, + model: string | null, + ) => Effect.Effect; +}>("t3/usage/UsageLimitResetFallback", { + defaultValue: () => ({ resetAt: () => Effect.succeed(null) }), +}) {} + +export const layerResetFallback = Layer.effect( + UsageLimitResetFallback, + Effect.gen(function* () { + const sources = yield* UsageLimitSources; + return { + resetAt: Effect.fn("UsageLimitResetFallback.resetAt")(function* (driver, endpoint, model) { + // Only the hub at the instance's endpoint serves its turns. + return hubPoolResetAt( + (yield* sources.current).filter((source) => source.origin === endpoint), + driver, + model, + DateTime.toEpochMillis(yield* DateTime.now), + ); + }), + }; + }), +); diff --git a/packages/contracts/src/providerUsageLimits.ts b/packages/contracts/src/providerUsageLimits.ts index 851de4c72ffe..901b096635d7 100644 --- a/packages/contracts/src/providerUsageLimits.ts +++ b/packages/contracts/src/providerUsageLimits.ts @@ -84,6 +84,8 @@ export const UsageLimitSourceSnapshot = Schema.Struct({ id: UsageLimitSourceId, kind: Schema.Literal("cliproxy"), label: TrimmedNonEmptyString, + /** Origin of the hub, which also serves the agents sent through it. */ + origin: Schema.optional(TrimmedNonEmptyString), checkedAt: IsoDateTime, accounts: ForwardCompatibleArray(UsageLimitSourceAccount), error: Schema.optional(TrimmedNonEmptyString), diff --git a/packages/contracts/src/server.ts b/packages/contracts/src/server.ts index a983e7c0826f..e8bcda3a76f5 100644 --- a/packages/contracts/src/server.ts +++ b/packages/contracts/src/server.ts @@ -70,6 +70,8 @@ export const ServerProviderAuth = Schema.Struct({ subscriptionSharing: Schema.optional(Schema.Boolean), profileId: Schema.optional(TrimmedNonEmptyString), workspaceId: Schema.optional(TrimmedNonEmptyString), + /** Origin of the custom endpoint the instance sends its requests to. */ + endpoint: Schema.optional(TrimmedNonEmptyString), }); export type ServerProviderAuth = typeof ServerProviderAuth.Type; From e42fc7e05932775d9ed8ac82cb20657dd40d33db Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 13 Nov 2026 16:52:22 +0000 Subject: [PATCH 2/4] fix(server): refresh the hub before reading a usage limit's reset The pool usually runs out after the last hub poll, so the cached accounts still showed room and the fallback stored no reset. The fallback now refreshes the hub at the instance's endpoint first, giving up after five seconds so a stalled hub cannot hold the failed turn. Co-Authored-By: Claude Opus 5.5 --- .../src/usage/UsageLimitSources.test.ts | 45 +++++++++++++++++++ apps/server/src/usage/UsageLimitSources.ts | 8 +++- 2 files changed, 52 insertions(+), 1 deletion(-) diff --git a/apps/server/src/usage/UsageLimitSources.test.ts b/apps/server/src/usage/UsageLimitSources.test.ts index 6724d6cf5699..d7e73a1da96c 100644 --- a/apps/server/src/usage/UsageLimitSources.test.ts +++ b/apps/server/src/usage/UsageLimitSources.test.ts @@ -11,9 +11,11 @@ import { import * as SecretResolver from "@t3tools/secret-provider-core/server/SecretResolver"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; import { HttpClient, HttpClientResponse } from "effect/http"; import * as BackgroundPolicy from "../background/BackgroundPolicy.ts"; @@ -260,3 +262,46 @@ it.each([ ), ).toBe(expected); }); + +it.effect("the reset fallback refreshes its hub, and a stalled hub cannot hold the failure", () => + Effect.gen(function* () { + yield* TestClock.setTime(NOW); + const roomLeft = { ...hub([{ windows: [spent("five_hour", 2)] }]), origin: "http://hub.test" }; + const cached = { + ...roomLeft, + accounts: roomLeft.accounts.map((account) => ({ + ...account, + usageLimits: { + ...account.usageLimits, + windows: account.usageLimits.windows.map((window) => ({ ...window, usedPercent: 99 })), + }, + })), + }; + let current: ReadonlyArray = [cached]; + let refresh: Effect.Effect = Effect.sync(() => { + current = [roomLeft]; + }); + const lookup = UsageLimitSources.UsageLimitResetFallback.pipe( + Effect.flatMap((fallback) => + fallback.resetAt(ProviderDriverKind.make("claudeAgent"), "http://hub.test", null), + ), + Effect.provide( + UsageLimitSources.layerResetFallback.pipe( + Layer.provide( + Layer.mock(UsageLimitSources.UsageLimitSources)({ + current: Effect.sync(() => current), + refresh: Effect.suspend(() => refresh), + }), + ), + ), + ), + ); + // The last poll saw room; the account was spent after it. + expect(yield* lookup).toBe(at(2)); + current = [cached]; + refresh = Effect.never; + const stalled = yield* Effect.forkChild(lookup); + yield* TestClock.adjust("5 seconds"); + expect(yield* Fiber.join(stalled)).toBeNull(); + }), +); diff --git a/apps/server/src/usage/UsageLimitSources.ts b/apps/server/src/usage/UsageLimitSources.ts index 96bcedaa0968..c59193184341 100644 --- a/apps/server/src/usage/UsageLimitSources.ts +++ b/apps/server/src/usage/UsageLimitSources.ts @@ -252,8 +252,14 @@ export const layerResetFallback = Layer.effect( return { resetAt: Effect.fn("UsageLimitResetFallback.resetAt")(function* (driver, endpoint, model) { // Only the hub at the instance's endpoint serves its turns. + const atEndpoint = Effect.map(sources.current, (current) => + current.filter((source) => source.origin === endpoint), + ); + if ((yield* atEndpoint).length === 0) return null; + // The pool usually ran out after the last poll, so the cached accounts still show room. + yield* sources.refresh.pipe(Effect.timeout("5 seconds"), Effect.ignore); return hubPoolResetAt( - (yield* sources.current).filter((source) => source.origin === endpoint), + yield* atEndpoint, driver, model, DateTime.toEpochMillis(yield* DateTime.now), From f16818094a879c7c79d5839283522820c81de015 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 13 Nov 2026 17:20:58 +0000 Subject: [PATCH 3/4] fix(server): a thread held behind a usage limit snoozes with its queued messages A usage limit holds the thread's queue, so its queued user messages cannot start until the user acts. Snooze still refused the thread, so a limit with no known reset left no way to set it aside without dropping the queue. Snooze now accepts held user work when the latest run failed on a usage limit. Unheld user work, and work held for any other reason, still blocks it. Co-Authored-By: Claude Opus 5.5 --- .../src/orchestration-v2/Orchestrator.ts | 16 +++++- .../src/orchestration-v2/runtimeLayer.test.ts | 51 +++++++++++++++++-- 2 files changed, 61 insertions(+), 6 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 26482feb129f..5aa574abcbe3 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -4438,7 +4438,21 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ) .map((message) => message.id), ); - if (queuedRuns.some((run) => !automaticMessageIds.has(run.userMessageId))) { + const userRuns = queuedRuns.filter((run) => !automaticMessageIds.has(run.userMessageId)); + const latest = latestExecutedRun(projection.runs); + // User work held behind a usage limit waits for the reset either way, so it may sleep through it. + const heldBehindLimit = + userRuns.length > 0 && + userRuns.every((run) => run.queueHeld === true) && + latest?.status === "failed" && + latestRootProviderFailure( + latest, + (yield* loadProjectionForCommand(command, ["turnItems"], { + turnItemTypes: ["error"], + turnItemRunIds: [latest.id], + })).turnItems, + )?.class === "usage_limit"; + if (userRuns.length > 0 && !heldBehindLimit) { return yield* new OrchestratorDispatchError({ commandId: command.commandId, commandType: command.type, diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index e62cd8725de9..014b4fad0f6f 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -9276,8 +9276,8 @@ it.layer(layerSharedApplicationDataPlaneTest)("pending provider interruption", ( }); it.layer(layerTest)("snooze queued runs", (it) => { - it.effect.each(["notification", "delegated-completion", "user"] as const)( - "only user messages block snooze with a held %s run", + it.effect.each(["notification", "delegated-completion", "user", "limited-user"] as const)( + "only user messages block snooze with a held %s run, unless a usage limit holds them", (kind) => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; @@ -9346,8 +9346,8 @@ it.layer(layerTest)("snooze queued runs", (it) => { attachments: [], modelSelection, dispatchMode: { type: "queue_after_active" }, - createdBy: kind === "user" ? "user" : "agent", - creationSource: kind === "user" ? "web" : "server", + createdBy: kind === "user" || kind === "limited-user" ? "user" : "agent", + creationSource: kind === "user" || kind === "limited-user" ? "web" : "server", ...(kind === "notification" ? { notification: { @@ -9381,10 +9381,51 @@ it.layer(layerTest)("snooze queued runs", (it) => { payload: run.status === "queued" ? { ...run, queueHeld: true } - : { ...run, status: "cancelled" as const, completedAt: now }, + : { + ...run, + status: kind === "limited-user" ? ("failed" as const) : ("cancelled" as const), + completedAt: now, + }, })), effects: [], }); + if (kind === "limited-user") { + yield* eventSink.write({ + events: [ + { + id: EventId.make(`limited:${threadId}`), + type: "turn-item.updated", + threadId, + occurredAt: now, + payload: { + id: TurnItemId.make(`limited:${threadId}`), + type: "error", + threadId, + runId: activeRun.id, + nodeId: activeRun.rootNodeId, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 2, + status: "failed", + title: "Usage limit reached", + startedAt: now, + completedAt: now, + updatedAt: now, + // A proxied provider can report no reset time. + failure: { + class: "usage_limit", + message: "Claude API rate limit reached. Try again later.", + code: "api_error_429", + retryable: true, + resetAt: null, + }, + }, + }, + ], + }); + } const beforeSnooze = yield* orchestrator.getThreadProjection(threadId); assert.isTrue(beforeSnooze.runs.some((run) => run.status === "queued" && run.queueHeld)); const snooze = orchestrator.dispatch({ From 47a32f68eecfc93759c87f4b4db70a4536a609dc Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 13 Nov 2026 17:44:51 +0000 Subject: [PATCH 4/4] fix(server): snooze accepts queued messages that a usage limit keeps waiting A usage limit stops the queue from starting without marking its runs held, so the previous queueHeld check never matched the reported case and an ordinary snooze still refused the thread. Snooze now asks the same rule the queue starter uses: queued user work is allowed when a usage limit blocks every queued run's provider instance. Co-Authored-By: Claude Opus 5.5 --- .../src/orchestration-v2/Orchestrator.ts | 23 ++++---- .../src/orchestration-v2/runtimeLayer.test.ts | 57 +++++++++++-------- 2 files changed, 44 insertions(+), 36 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 5aa574abcbe3..3e6a09c1e58a 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -4439,19 +4439,20 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio .map((message) => message.id), ); const userRuns = queuedRuns.filter((run) => !automaticMessageIds.has(run.userMessageId)); + // User work waiting behind a usage limit cannot start until the reset or the user acts. const latest = latestExecutedRun(projection.runs); - // User work held behind a usage limit waits for the reset either way, so it may sleep through it. + const limitRecords = + userRuns.length === 0 || latest?.status !== "failed" + ? null + : yield* loadProjectionForCommand(command, ["turnItems", "providerSessions"], { + turnItemTypes: ["error"], + turnItemRunIds: [latest.id], + }); const heldBehindLimit = - userRuns.length > 0 && - userRuns.every((run) => run.queueHeld === true) && - latest?.status === "failed" && - latestRootProviderFailure( - latest, - (yield* loadProjectionForCommand(command, ["turnItems"], { - turnItemTypes: ["error"], - turnItemRunIds: [latest.id], - })).turnItems, - )?.class === "usage_limit"; + limitRecords !== null && + userRuns.every((run) => + usageLimitHoldsQueue({ ...limitRecords, runs: projection.runs }, run.providerInstanceId), + ); if (userRuns.length > 0 && !heldBehindLimit) { return yield* new OrchestratorDispatchError({ commandId: command.commandId, diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 014b4fad0f6f..698c80813cdd 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -9277,7 +9277,7 @@ it.layer(layerSharedApplicationDataPlaneTest)("pending provider interruption", ( it.layer(layerTest)("snooze queued runs", (it) => { it.effect.each(["notification", "delegated-completion", "user", "limited-user"] as const)( - "only user messages block snooze with a held %s run, unless a usage limit holds them", + "only user messages block snooze with a queued %s run, unless a usage limit holds them", (kind) => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; @@ -9366,29 +9366,6 @@ it.layer(layerTest)("snooze queued runs", (it) => { } : {}), }); - const beforeRecovery = yield* orchestrator.getThreadProjection(threadId); - yield* eventSink.commitCommand({ - commandId: CommandId.make(`recovery:${threadId}`), - threadId, - commandType: "provider-runtime.reconcile", - acceptedAt: now, - events: beforeRecovery.runs.map((run) => ({ - id: EventId.make(`recovery:${run.id}`), - type: "run.updated" as const, - threadId, - runId: run.id, - occurredAt: now, - payload: - run.status === "queued" - ? { ...run, queueHeld: true } - : { - ...run, - status: kind === "limited-user" ? ("failed" as const) : ("cancelled" as const), - completedAt: now, - }, - })), - effects: [], - }); if (kind === "limited-user") { yield* eventSink.write({ events: [ @@ -9426,8 +9403,38 @@ it.layer(layerTest)("snooze queued runs", (it) => { ], }); } + const beforeRecovery = yield* orchestrator.getThreadProjection(threadId); + yield* eventSink.commitCommand({ + commandId: CommandId.make(`recovery:${threadId}`), + threadId, + commandType: "provider-runtime.reconcile", + acceptedAt: now, + events: beforeRecovery.runs.map((run) => ({ + id: EventId.make(`recovery:${run.id}`), + type: "run.updated" as const, + threadId, + runId: run.id, + occurredAt: now, + payload: + run.status === "queued" + ? // A usage limit stops the queue from starting without marking it held. + kind === "limited-user" + ? run + : { ...run, queueHeld: true } + : { + ...run, + status: kind === "limited-user" ? ("failed" as const) : ("cancelled" as const), + completedAt: now, + }, + })), + effects: [], + }); const beforeSnooze = yield* orchestrator.getThreadProjection(threadId); - assert.isTrue(beforeSnooze.runs.some((run) => run.status === "queued" && run.queueHeld)); + assert.isTrue( + beforeSnooze.runs.some( + (run) => run.status === "queued" && (kind === "limited-user" || run.queueHeld), + ), + ); const snooze = orchestrator.dispatch({ type: "thread.snooze", commandId: CommandId.make(`snooze:${threadId}`),