diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index ea5ee249a787..5c227ee90a39 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -23,6 +23,7 @@ import type { TaskUpdateInput, TaskUpdateOutput, } from "@anthropic-ai/claude-agent-sdk/sdk-tools"; +import * as UsageLimitSources from "../../usage/UsageLimitSources.ts"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { ChatAttachmentId, @@ -5750,6 +5751,158 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); + it.effect.each([ + "environment", + "settings", + "lenient settings", + "root local", + "process cwd", + "launch inline", + "launch file", + "stalled lookup", + "reused endpoint", + "reused model", + ])("uses the matching proxy reset from %s when Claude omits a reset time", (source) => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const configDir = yield* fs.makeTempDirectoryScoped({ prefix: "t3-proxy-reset-" }); + if (source === "settings") + yield* fs.writeFileString( + path.join(configDir, "settings.json"), + JSON.stringify({ env: { ANTHROPIC_BASE_URL: "https://hub.test/v1" } }), + ); + const cwd = path.join(configDir, "repo", "nested"); + yield* fs.makeDirectory(path.join(configDir, "repo", ".git"), { recursive: true }); + yield* fs.makeDirectory(path.join(cwd, ".claude"), { recursive: true }); + if (source === "lenient settings") + yield* fs.writeFileString( + path.join(configDir, "settings.json"), + '{ // Claude settings allow comments and trailing commas\n "env": { "ANTHROPIC_BASE_URL": "https://hub.test/v1", }, }', + ); + if (source === "process cwd") + yield* fs.writeFileString( + path.join(cwd, ".claude", "settings.json"), + JSON.stringify({ env: { ANTHROPIC_BASE_URL: "https://hub.test/v1" } }), + ); + if (source === "root local") { + yield* fs.writeFileString( + path.join(cwd, ".claude", "settings.local.json"), + JSON.stringify({ env: { ANTHROPIC_BASE_URL: "https://wrong.test" } }), + ); + yield* fs.makeDirectory(path.join(configDir, "repo", ".claude")); + yield* fs.writeFileString( + path.join(configDir, "repo", ".claude", "settings.local.json"), + JSON.stringify({ env: { ANTHROPIC_BASE_URL: "https://hub.test/v1" } }), + ); + } + const launchSettings = JSON.stringify({ env: { ANTHROPIC_BASE_URL: "https://hub.test/v1" } }); + if (source === "launch file") + yield* fs.writeFileString(path.join(cwd, "launch.json"), launchSettings); + const lookupStarted = yield* Deferred.make(); + const harness = yield* makeWakeHarnessWithOptions({ + settings: { + ...DEFAULT_CLAUDE_SETTINGS, + launchArgs: + source === "launch inline" + ? `--settings '${launchSettings}'` + : source === "launch file" + ? "--settings launch.json" + : "", + }, + environment: { + CLAUDE_CONFIG_DIR: configDir, + ANTHROPIC_BASE_URL: + source === "environment" || source.startsWith("reused") || source === "stalled lookup" + ? "https://hub.test/v1" + : "https://wrong.test", + }, + }).pipe( + Effect.provideService(HostProcess.WorkingDirectory, cwd), + Effect.provide( + Layer.mock(UsageLimitSources.UsageLimitSources)({ + resolveResetAt: (input) => { + assert.equal(input.endpoint, "https://hub.test/v1"); + assert.equal(input.model, CLAUDE_TEST_MODEL_SELECTION.model); + if (source === "stalled lookup") + return Deferred.succeed(lookupStarted, undefined).pipe( + Effect.andThen(Effect.never), + ); + return Effect.succeed("2099-01-01T02:00:00.000Z"); + }, + }), + ), + ); + const now = yield* DateTime.now; + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-proxy-reset"), + text: "Continue.", + attachments: [], + runtimePolicy: { + ...CLAUDE_TEST_RUNTIME_POLICY, + cwd: source === "process cwd" || source === "stalled lookup" ? null : cwd, + }, + }), + ); + if (source.startsWith("reused")) { + yield* Queue.offer(harness.sdkMessages, wakeTaskStarted); + yield* Queue.offer(harness.sdkMessages, turnOneResult); + assert.equal((yield* Queue.take(harness.terminalReceipts)).status, "completed"); + yield* harness.offerAndWait(wakeNotification); + yield* harness.offerAndWait(wakeTurnInit); + if (source === "reused endpoint") + yield* fs.writeFileString( + path.join(configDir, "settings.json"), + JSON.stringify({ env: { ANTHROPIC_BASE_URL: "https://wrong.test" } }), + ); + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-proxy-continuation"), + providerTurnOrdinal: 2, + messageCreatedBy: "agent", + messageCreationSource: "provider", + modelSelection: CLAUDE_TEST_RESTART_MODEL_SELECTION, + runtimePolicy: { ...CLAUDE_TEST_RUNTIME_POLICY, cwd: configDir }, + text: "Background task completed.", + attachments: [], + }), + ); + } + yield* Queue.offerAll(harness.sdkMessages, [ + makeAssistantErrorFrame({ + uuid: "00000000-0000-4000-8000-000000000652", + error: "rate_limit", + }), + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000653", + result: "Usage limit reached.", + terminalReason: "blocking_limit", + isError: true, + apiErrorStatus: 429, + }), + ]); + if (source === "stalled lookup") { + yield* Deferred.await(lookupStarted); + yield* TestClock.adjust("15 seconds"); + } + const terminal = yield* Queue.take(harness.terminalReceipts); + assert.equal(terminal.status, "failed"); + if (terminal.status !== "failed") return; + assert.equal(terminal.failure.class, "usage_limit"); + assert.equal( + terminal.failure.resetAt, + source === "stalled lookup" ? null : "2099-01-01T02:00:00.000Z", + ); + }).pipe(Effect.provide([IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer])), + ); + it.effect.each([ { status: 503, cooldown: "4m 6s", seconds: 246 }, { status: 503, cooldown: "1h 2m 3s", seconds: 3723 }, diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index b1273cee7206..9419ea817dab 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -140,6 +140,7 @@ import { compileClaudeModelSelection, type CompiledClaudeModelSelection, } from "../../claudeModelOptions.ts"; +import * as UsageLimitSources from "../../usage/UsageLimitSources.ts"; import * as ServerConfig from "../../config.ts"; import * as PortScanner from "../../preview/PortScanner.ts"; import * as ServerSettings from "../../serverSettings.ts"; @@ -2596,6 +2597,14 @@ const ClaudeBypassSettings = Schema.StructWithRest( }), [Schema.Record(Schema.String, Schema.Unknown)], ); +const decodeClaudeEndpointSettings = Schema.decodeUnknownEffect( + fromLenientJson( + Schema.Struct({ + env: Schema.optional(Schema.Struct({ ANTHROPIC_BASE_URL: Schema.optional(Schema.String) })), + }), + ), +); + const decodeClaudeBypassSettings = Schema.decodeUnknownEffect( fromLenientJson(ClaudeBypassSettings), ); @@ -3955,6 +3964,7 @@ interface ClaudeLiveQueryContext { readonly nativeThreadId: string; readonly query: ClaudeAgentSdkQuerySession; readonly queryPolicyKey: string; + readonly proxyEndpoint: string | null; selection: CompiledClaudeModelSelection; readonly closed: Deferred.Deferred; lastTurn: ActiveClaudeTurnContext | null; @@ -4367,7 +4377,48 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( const mcpSessions = yield* McpProviderSessions.McpProviderSessions; const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; const serverSettings = yield* Effect.serviceOption(ServerSettings.ServerSettingsService); + const usageLimitSources = yield* Effect.serviceOption(UsageLimitSources.UsageLimitSources); + const workingDirectory = yield* HostProcess.WorkingDirectory; const { attachmentsDir, fileSystem, path, crypto, idAllocator, queryRunner } = adapterOptions; + const resolveProxyEndpoint = Effect.fn("ClaudeAdapterV2.resolveProxyEndpoint")( + function* (queryCwd: string | null, launchSettings: ClaudeSdkSettings | undefined) { + if (Option.isNone(usageLimitSources)) return null; + const cwd = queryCwd ?? workingDirectory; + const localDirectory = yield* localSettingsDirectory(cwd); + let endpoint = adapterOptions.environment.ANTHROPIC_BASE_URL; + const settingsPaths = [ + path.join(homePath, "settings.json"), + path.join(cwd, ".claude", "settings.json"), + path.join(cwd, ".claude", "settings.local.json"), + ...(localDirectory === cwd + ? [] + : [path.join(localDirectory, ".claude", "settings.local.json")]), + ]; + for (const settingsPath of settingsPaths) { + const settings = yield* fileSystem.readFileString(settingsPath).pipe( + Effect.flatMap(decodeClaudeEndpointSettings), + Effect.orElseSucceed(() => undefined), + ); + endpoint = settings?.env?.ANTHROPIC_BASE_URL ?? endpoint; + } + return launchSettings?.env?.ANTHROPIC_BASE_URL ?? endpoint ?? null; + }, + Effect.timeout("15 seconds"), + Effect.orElseSucceed(() => null), + ); + + const proxyResetAt = Effect.fn("ClaudeAdapterV2.proxyResetAt")( + function* (liveQuery: ClaudeLiveQueryContext) { + if (Option.isNone(usageLimitSources) || !liveQuery.proxyEndpoint) return null; + return yield* usageLimitSources.value.resolveResetAt({ + endpoint: liveQuery.proxyEndpoint, + model: liveQuery.selection.apiModelId, + }); + }, + Effect.timeout("15 seconds"), + Effect.orElseSucceed(() => null), + ); + const subagentModelCatalog = scopeClaudeModelCatalog( BUNDLED_CLAUDE_MODEL_CATALOG, adapterOptions.settings.customModels, @@ -9217,9 +9268,15 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( code: "compaction_failed", }) : providerFailureFromResult(message, completedAt, failureHint, usageLimited); + const fallbackResetAt = + resultFailure?.class === "usage_limit" && + resetAt === null && + resultFailure.resetAt == null + ? yield* proxyResetAt(liveQuery) + : null; const terminalFailure = resultFailure?.class === "usage_limit" - ? { ...resultFailure, resetAt: resetAt ?? resultFailure.resetAt ?? null } + ? { ...resultFailure, resetAt: resetAt ?? resultFailure.resetAt ?? fallbackResetAt } : resultFailure; yield* finalizeActiveTurn({ context, @@ -9945,6 +10002,10 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( ...(appendSystemPromptFileText === undefined ? {} : { appendSystemPromptFileText }), ...(toolSearchOverride === undefined ? {} : { toolSearchOverride }), }); + const proxyEndpoint = yield* resolveProxyEndpoint( + turnInput.runtimePolicy.cwd, + launchArgSettings, + ); const querySession = yield* queryRunner .open({ threadId: turnInput.threadId, @@ -9982,6 +10043,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( query: querySession, queryPolicyKey, selection: compiledSelection, + proxyEndpoint, closed, lastTurn: null, promptEchoMode: "unknown", diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 741fc73103b7..e98dbfa16173 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -634,11 +634,10 @@ const layerRuntimeCoreDependenciesBase = Layer.mergeAll( Layer.provideMerge(layerVcs), Layer.provideMerge(Layer.mergeAll(layerTerminal, layerPreview, layerDevice)), Layer.provideMerge(layerPersistence), - Layer.provideMerge( - Layer.mergeAll(Keybindings.layer, EnvironmentTheme.layer, UsageLimitSources.layer), - ), + Layer.provideMerge(Layer.mergeAll(Keybindings.layer, EnvironmentTheme.layer)), Layer.provideMerge(ProviderRegistry.layer), Layer.provideMerge(ProviderInstanceRegistryHydration.layer), + Layer.provideMerge(UsageLimitSources.layer), // Provider hydration builds the Claude runner, which reports agents to the same scanner the // terminal and preview layers share. Layer.provideMerge(layerPortScanner), diff --git a/apps/server/src/usage/UsageLimitSources.test.ts b/apps/server/src/usage/UsageLimitSources.test.ts index b8976fca81e7..6e766c869072 100644 --- a/apps/server/src/usage/UsageLimitSources.test.ts +++ b/apps/server/src/usage/UsageLimitSources.test.ts @@ -4,7 +4,7 @@ import { ServerConfigStreamEvent, UsageLimitSourceId } from "@t3tools/contracts" import * as Effect from "effect/Effect"; 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 +83,137 @@ it.effect("keeps malformed source URLs encodable for config subscribers", () => Effect.scoped, ), ); + +it.effect.each([ + { scenario: "multiple windows", expected: "2099-01-01T02:00:00.000Z" }, + { scenario: "base model for 1m", expected: "2099-01-01T02:00:00.000Z" }, + { scenario: "suffixed model", expected: "2099-01-01T02:00:00.000Z" }, + { scenario: "unknown suffix", expected: null }, + { scenario: "model scopes", expected: "2099-01-01T04:00:00.000Z" }, + { scenario: "disabled account", expected: "2099-01-01T03:00:00.000Z" }, + { scenario: "model list unavailable", expected: "2099-01-01T03:00:00.000Z" }, + { scenario: "missing resets", expected: null }, + { scenario: "hub unavailable", expected: null }, + { scenario: "disabled source", expected: null }, +])("resolves only the matching hub and model: $scenario", ({ scenario, expected }) => + Effect.gen(function* () { + const sources = yield* UsageLimitSources.make; + const unrelated = yield* sources.resolveResetAt({ + endpoint: "https://other.test", + model: "claude-opus-4-6", + }); + expect(unrelated).toBeNull(); + const reset = yield* sources.resolveResetAt({ + endpoint: "https://hub.test/v1", + model: + scenario === "base model for 1m" || scenario === "suffixed model" + ? "claude-opus-4-6[1m]" + : scenario === "unknown suffix" + ? "claude-opus-4-6[2m]" + : "claude-opus-4-6", + }); + expect(reset).toBe(expected); + const unlisted = yield* sources.resolveResetAt({ + endpoint: "https://hub.test/v1", + model: "unlisted", + }); + expect(unlisted).toBeNull(); + }).pipe( + Effect.provide([ + ServerSettings.layerTest({ + usageLimitSources: { + [UsageLimitSourceId.make("hub")]: { + kind: "cliproxy", + url: "https://hub.test", + managementKey: "test", + enabled: scenario !== "disabled source", + }, + }, + }), + Layer.mock(BackgroundPolicy.BackgroundPolicy)({ + shouldRunScopeWork: () => Effect.succeed(false), + }), + Layer.succeed( + HttpClient.HttpClient, + HttpClient.make((request) => + Effect.sync(() => { + const url = new URL(request.url); + if ( + scenario === "hub unavailable" || + (scenario === "model list unavailable" && url.searchParams.get("name") === "fast") + ) { + return HttpClientResponse.fromWeb(request, new Response(null, { status: 503 })); + } + const files = ["slow", "fast", "wrong-model"]; + const body = url.pathname.endsWith("/auth-files") + ? { + files: files.map((id) => ({ + id, + auth_index: id, + provider: "claude", + disabled: scenario === "disabled account" && id === "fast", + })), + } + : url.pathname.endsWith("/auth-files/models") + ? { + models: [ + { + id: + url.searchParams.get("name") === "wrong-model" + ? "claude-sonnet-4-6" + : scenario === "suffixed model" + ? "claude-opus-4-6[1m]" + : "claude-opus-4-6", + }, + ], + } + : (() => { + const input = JSON.parse( + request.body._tag === "Uint8Array" + ? new TextDecoder().decode(request.body.body) + : "{}", + ); + const hour = + input.auth_index === "slow" + ? "03" + : input.auth_index === "fast" + ? "02" + : "01"; + return { + status_code: 200, + body: JSON.stringify({ + five_hour: { + utilization: 100, + resets_at: scenario === "missing resets" ? null : "2099-01-01T01:00:00Z", + }, + seven_day: { utilization: 100, resets_at: `2099-01-01T${hour}:00:00Z` }, + ...(scenario === "model scopes" + ? { + limits: [ + { + kind: "weekly_scoped", + percent: 100, + resets_at: "2099-01-01T04:00:00Z", + scope: { model: { display_name: "Opus 4.6" } }, + }, + { + kind: "weekly_scoped", + percent: 100, + resets_at: "2099-01-01T05:00:00Z", + scope: { model: { display_name: "Sonnet" } }, + }, + ], + } + : {}), + }), + }; + })(); + return HttpClientResponse.fromWeb(request, Response.json(body)); + }), + ), + ), + NodeCrypto.layer, + ]), + Effect.scoped, + ), +); diff --git a/apps/server/src/usage/UsageLimitSources.ts b/apps/server/src/usage/UsageLimitSources.ts index 7a068cde85f6..f18117a5c134 100644 --- a/apps/server/src/usage/UsageLimitSources.ts +++ b/apps/server/src/usage/UsageLimitSources.ts @@ -30,6 +30,10 @@ export class UsageLimitSources extends Context.Service< readonly current: Effect.Effect>; readonly streamChanges: Stream.Stream>; readonly refresh: Effect.Effect; + readonly resolveResetAt: (input: { + readonly endpoint: string; + readonly model: string; + }) => Effect.Effect; readonly consumeResetCredit: ( input: UsageLimitSourceConsumeResetCreditInput, ) => Effect.Effect; @@ -149,8 +153,60 @@ export const make = Effect.gen(function* () { yield* refresh.pipe(Effect.forkScoped); + const resolveResetAt = Effect.fn("UsageLimitSources.resolveResetAt")( + function* (input: { readonly endpoint: string; readonly model: string }) { + const endpoint = URL.parse(input.endpoint); + if (!endpoint || !["http:", "https:"].includes(endpoint.protocol)) return null; + const settings = yield* settingsService.getSettings; + const entries = Object.entries(settings.usageLimitSources).filter(([, config]) => { + const hub = URL.parse(config.url); + if (!config.enabled || !config.managementKey || !hub || hub.origin !== endpoint.origin) + return false; + const basePath = hub.pathname.replace(/\/$/, ""); + return endpoint.pathname === basePath || endpoint.pathname.startsWith(`${basePath}/`); + }); + if (entries.length !== 1) return null; + const [id, config] = entries[0]!; + const snapshot = yield* readSource(id as UsageLimitSourceId, config); + const now = DateTime.toEpochMillis(yield* DateTime.now); + const resets = yield* Effect.forEach( + snapshot.accounts.filter((account) => account.driver === "claudeAgent"), + (account) => + Effect.gen(function* () { + const models = yield* api.accountModels(config, account.id); + const modelId = models.includes(input.model) + ? input.model + : input.model.replace(/\[1m\]$/, ""); + if (!models.includes(modelId) || account.usageLimits.unavailable) return null; + const model = modelId.replace(/^claude-/, "").replace(/[^a-z0-9]+/g, "_"); + const blocked = account.usageLimits.windows.filter((window) => { + if (window.usedPercent < 100) return false; + if (window.id === "five_hour" || window.id === "seven_day") return true; + const scope = window.id.replace(/^seven_day_/, "").replace(/^claude_/, ""); + if (!/^(opus|sonnet|haiku)(_|$)/.test(scope)) return true; + return model === scope || model.startsWith(`${scope}_`); + }); + if ( + blocked.length === 0 || + blocked.some((window) => !window.resetsAt || Date.parse(window.resetsAt) <= now) + ) + return null; + return Math.max(...blocked.map((window) => Date.parse(window.resetsAt!))); + }).pipe(Effect.orElseSucceed(() => null)), + { concurrency: 4 }, + ); + const known = resets.filter((reset) => reset !== null); + return known.length === 0 + ? null + : DateTime.formatIso(DateTime.makeUnsafe(Math.min(...known))); + }, + Effect.timeout("15 seconds"), + Effect.orElseSucceed(() => null), + ); + return { current: Ref.get(stateRef), + resolveResetAt, consumeResetCredit, refresh, get streamChanges() { diff --git a/apps/server/src/usage/cliproxyApi.ts b/apps/server/src/usage/cliproxyApi.ts index 91504d800229..65bba7ef662e 100644 --- a/apps/server/src/usage/cliproxyApi.ts +++ b/apps/server/src/usage/cliproxyApi.ts @@ -83,6 +83,11 @@ const CreditList = Schema.Struct({ ), }); +const decodeAccountModels = Schema.decodeUnknownEffect( + Schema.Struct({ + models: Schema.Array(Schema.Struct({ id: Schema.String })), + }), +); const decodeAuthFiles = Schema.decodeUnknownEffect(AuthFiles); const encodeJson = Schema.encodeEffect(Schema.fromJsonString(Schema.Unknown)); const decodeApiResponse = Schema.decodeUnknownEffect(ApiResponse); @@ -423,5 +428,15 @@ export const makeCliproxyApi = Effect.gen(function* () { ), ); }); - return { readAccounts, consume }; + const accountModels = Effect.fn("CliproxyApi.accountModels")(function* ( + config: UsageLimitSourceConfig, + accountId: string, + ) { + const response = yield* management( + config, + `auth-files/models?name=${encodeURIComponent(accountId)}`, + ); + return (yield* decodeAccountModels(response)).models.map((model) => model.id); + }); + return { readAccounts, consume, accountModels }; });