diff --git a/apps/server/src/auth/RpcAuthorization.test.ts b/apps/server/src/auth/RpcAuthorization.test.ts index 8044740a7e2e..84995e1eeb87 100644 --- a/apps/server/src/auth/RpcAuthorization.test.ts +++ b/apps/server/src/auth/RpcAuthorization.test.ts @@ -68,6 +68,19 @@ describe("RPC authorization scopes", () => { } }); + it("keeps proxy accounts, which read and spend pooled subscriptions, behind provider management", () => { + for (const method of [ + WS_METHODS.providerProxyGetState, + WS_METHODS.providerProxyStartLogin, + WS_METHODS.providerProxyGetLoginStatus, + WS_METHODS.providerProxyCompleteLogin, + WS_METHODS.providerProxySetAccountDisabled, + WS_METHODS.providerProxyRemoveAccount, + ]) { + expect(requiredScopeForRpcMethod(method)).toBe(AuthProvidersManageScope); + } + }); + it("rejects unknown RPC method names", () => { for (const method of ["server.notRegistered", "toString", "constructor"]) { expect(() => requiredScopeForRpcMethod(method)).toThrow( diff --git a/apps/server/src/auth/RpcAuthorization.ts b/apps/server/src/auth/RpcAuthorization.ts index 6049121b41a5..b6b9d36c066d 100644 --- a/apps/server/src/auth/RpcAuthorization.ts +++ b/apps/server/src/auth/RpcAuthorization.ts @@ -66,6 +66,9 @@ export const RPC_REQUIRED_SCOPES = { [WS_METHODS.serverUpdateProvider]: AuthProvidersManageScope, [WS_METHODS.providerAuthStart]: AuthProvidersManageScope, [WS_METHODS.providerConsumeResetCredit]: AuthProvidersManageScope, + // Account emails and login status come from the management key. + [WS_METHODS.providerProxyGetState]: AuthProvidersManageScope, + [WS_METHODS.providerProxyGetLoginStatus]: AuthProvidersManageScope, [WS_METHODS.providerAuthComplete]: AuthProvidersManageScope, [WS_METHODS.chatGptReconnectProfile]: AuthProvidersManageScope, [WS_METHODS.chatGptImportProfile]: AuthProvidersManageScope, diff --git a/apps/server/src/observability/RpcInstrumentation.ts b/apps/server/src/observability/RpcInstrumentation.ts index 1e3388e4136a..a154c12839dd 100644 --- a/apps/server/src/observability/RpcInstrumentation.ts +++ b/apps/server/src/observability/RpcInstrumentation.ts @@ -38,6 +38,12 @@ const RPC_AGGREGATES = { [WS_METHODS.serverUpdateProvider]: "server", [WS_METHODS.providerAuthStart]: "provider", [WS_METHODS.providerConsumeResetCredit]: "provider", + [WS_METHODS.providerProxyGetState]: "provider", + [WS_METHODS.providerProxyStartLogin]: "provider", + [WS_METHODS.providerProxyGetLoginStatus]: "provider", + [WS_METHODS.providerProxyCompleteLogin]: "provider", + [WS_METHODS.providerProxySetAccountDisabled]: "provider", + [WS_METHODS.providerProxyRemoveAccount]: "provider", [WS_METHODS.providerAuthComplete]: "provider", [WS_METHODS.chatGptReconnectProfile]: "provider", [WS_METHODS.chatGptImportProfile]: "provider", diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 356f4cbd1efc..0063f3c8414f 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -1006,12 +1006,17 @@ describe("ClaudeAdapterV2 Auto-accept edits", () => { }); describe("ClaudeAdapterV2 append-system-prompt-file launch argument", () => { - const startTurnWithLaunchArgs = Effect.fnUntraced(function* (launchArgs: string, cwd: string) { + const startTurnWithLaunchArgs = Effect.fnUntraced(function* ( + launchArgs: string, + cwd: string, + proxy?: { readonly url: string; readonly apiKey: string }, + ) { const fileSystem = yield* FileSystem.FileSystem; const opened: Array = []; const adapter = yield* ClaudeAdapterV2.makeClaudeAdapterV2({ instanceId: ClaudeAdapterV2.CLAUDE_DEFAULT_INSTANCE_ID, settings: { ...DEFAULT_CLAUDE_SETTINGS, launchArgs }, + ...(proxy === undefined ? {} : { proxy }), environment: {}, attachmentsDir: yield* fileSystem.makeTempDirectoryScoped({ prefix: "t3-claude-append-" }), fileSystem, @@ -1106,7 +1111,7 @@ describe("ClaudeAdapterV2 append-system-prompt-file launch argument", () => { // regression: #15073 the SDK replaced the --settings launch argument with T3's settings it.effect.each([ - `--settings '{"skillOverrides":{"lint":"off"},"showThinkingSummaries":false}'`, + `--settings '{"skillOverrides":{"lint":"off"},"showThinkingSummaries":false,"env":{"OTHER":"kept","ANTHROPIC_BASE_URL":"https://old.example.test"}}'`, "--settings user-settings.json", ])("merges %s beneath T3's settings", (launchArgs) => Effect.scoped( @@ -1115,7 +1120,7 @@ describe("ClaudeAdapterV2 append-system-prompt-file launch argument", () => { const cwd = yield* fileSystem.makeTempDirectoryScoped({ prefix: "t3-claude-settings-" }); yield* fileSystem.writeFileString( NodePath.join(cwd, "user-settings.json"), - '{"skillOverrides":{"lint":"off"},"showThinkingSummaries":false}', + '{"skillOverrides":{"lint":"off"},"showThinkingSummaries":false,"env":{"OTHER":"kept","ANTHROPIC_BASE_URL":"https://old.example.test"}}', ); const { exit, opened } = yield* startTurnWithLaunchArgs(launchArgs, cwd); assert.isTrue(Exit.isSuccess(exit)); @@ -1124,6 +1129,22 @@ describe("ClaudeAdapterV2 append-system-prompt-file launch argument", () => { showThinkingSummaries: true, }); assert.isUndefined(opened[0]!.extraArgs?.["settings"]); + const proxy = { url: "http://proxy.test:8317", apiKey: "proxy-test-key" }; + const routed = yield* startTurnWithLaunchArgs(launchArgs, cwd, proxy); + assert.isTrue(Exit.isSuccess(routed.exit)); + assert.deepInclude(routed.opened[0]!.settings, { + skillOverrides: { lint: "off" }, + env: { + OTHER: "kept", + ANTHROPIC_BASE_URL: proxy.url, + ANTHROPIC_AUTH_TOKEN: proxy.apiKey, + ANTHROPIC_API_KEY: "", + CLAUDE_CODE_OAUTH_TOKEN: "", + CLAUDE_CODE_USE_BEDROCK: "", + CLAUDE_CODE_USE_VERTEX: "", + CLAUDE_CODE_USE_FOUNDRY: "", + }, + }); }).pipe( Effect.provide( Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 06b645bb7c05..84ad90a06e22 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -153,6 +153,8 @@ import { expandHomePath } from "@t3tools/provider-core/server/pathExpansion"; import { spawnAndCollect } from "@t3tools/provider-core/server/snapshotProbe"; import { signalProcessGroup } from "@t3tools/provider-core/server/processGroup"; import { + applyClaudeProxyEnvironment, + claudeProxySettingsEnvironment, claudeSignedOutMessage, makeClaudeEnvironment, resolveClaudeHomePath, @@ -4482,6 +4484,7 @@ function claudeProposedPlan(input: unknown): string | null { export interface ClaudeAdapterV2Options { readonly instanceId: ProviderInstanceId; + readonly proxy?: ProviderAdapterDriverCreateInput["proxy"]; readonly settings: ClaudeSettings; readonly environment: NodeJS.ProcessEnv; readonly attachmentsDir: string; @@ -10213,12 +10216,22 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( fileSystem, path, }); - const launchArgSettings = yield* readClaudeLaunchArgSettings({ + const userLaunchArgSettings = yield* readClaudeLaunchArgSettings({ settings: adapterOptions.settings, cwd: turnInput.runtimePolicy.cwd, fileSystem, path, }); + const launchArgSettings = + adapterOptions.proxy === undefined + ? userLaunchArgSettings + : { + ...userLaunchArgSettings, + env: { + ...userLaunchArgSettings?.env, + ...claudeProxySettingsEnvironment(adapterOptions.proxy), + }, + }; const closedExistingNativeThreadId = existing !== null ? existing.nativeThreadId : null; if (existing !== null) { yield* existing.query.close.pipe(Effect.ignore); @@ -11535,14 +11548,17 @@ export const createClaudeAdapterV2 = Effect.fn("ClaudeAdapterV2Driver.create")( input: ProviderAdapterDriverCreateInput, hooks: Pick = {}, ) { - const { instanceId, environment, enabled, config } = input; + const { instanceId, environment, enabled, config, proxy } = input; const fileSystem = yield* FileSystem.FileSystem; const hostEnvironment = yield* HostProcess.Environment; const idAllocator = yield* IdAllocator.IdAllocatorV2; const queryRunner = yield* ClaudeAgentSdkQueryRunner; const serverConfig = yield* ServerConfig.ServerConfig; const continuationRequests = yield* ProviderContinuationRequests.ProviderContinuationRequests; - const baseEnvironment = yield* mergeProviderInstanceEnvironment(environment, hostEnvironment); + const baseEnvironment = applyClaudeProxyEnvironment( + yield* mergeProviderInstanceEnvironment(environment, hostEnvironment), + proxy, + ); const claudeEnvironment = yield* makeClaudeEnvironment(config, baseEnvironment); const path = yield* Path.Path; const crypto = yield* Crypto.Crypto; @@ -11553,6 +11569,7 @@ export const createClaudeAdapterV2 = Effect.fn("ClaudeAdapterV2Driver.create")( return yield* makeClaudeAdapterV2({ instanceId, settings: { ...config, enabled, binaryPath }, + ...(proxy === undefined ? {} : { proxy }), environment: claudeEnvironment, attachmentsDir: serverConfig.attachmentsDir, fileSystem, diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 90ee44f3d5b6..7a47806744ba 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -68,6 +68,7 @@ import * as Orchestrator from "../Orchestrator.ts"; import * as ProviderReplayHarness from "../testkit/ProviderReplayHarness.ts"; import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; import type * as ProviderContinuationRequests from "@t3tools/provider-core/server/ProviderContinuationRequests"; +import * as CodexLaunchArgs from "../../provider/codexLaunchArgs.ts"; import * as ModelManifest from "../../provider/ModelManifest.ts"; import { resolveProviderCompatibility } from "../../provider/providerCompatibility.ts"; import * as CodexAdapterV2 from "./CodexAdapterV2.ts"; @@ -1142,6 +1143,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { onRequest: (method: string, params: unknown) => Effect.Effect = () => Effect.void, readChildMetadata?: Parameters[2], beforeEmitInbound?: CodexReplay.CodexAppServerReplayDriver["beforeEmitInbound"], + modelProvider?: string, ) => Effect.gen(function* () { const fileSystem = yield* FileSystem.FileSystem; @@ -1191,6 +1193,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { instanceId: CodexAdapterV2.CODEX_DEFAULT_INSTANCE_ID, settings: DEFAULT_CODEX_SETTINGS, environment: {}, + modelProvider, clientFactory, crypto: yield* Crypto.Crypto, fileSystem, @@ -2418,6 +2421,80 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); + it.effect("pins thread start and resume to the proxy provider when no model is resent", () => + Effect.scoped( + Effect.gen(function* () { + const nativeThreadId = "proxied-thread"; + const modelProvider = CodexLaunchArgs.CODEX_PROXY_MODEL_PROVIDER; + const preamble = codexReplayPreamble({ + nativeThreadId, + nativeTurnId: "unused-turn", + prompt: "unused-prompt", + }) + .slice(0, 5) + .map((entry) => + entry.type === "expect_outbound" && entry.label === "thread/start" + ? { + ...entry, + frame: { + id: 2, + method: "thread/start", + params: { modelProvider, config: CodexAdapterV2.CODEX_THREAD_CONFIG }, + }, + } + : entry, + ); + const harness = yield* makeCodexReplayHarness( + makeCodexReplayTranscript({ + scenario: "proxy-resume", + entries: [ + ...preamble, + { + type: "expect_outbound", + label: "resume through the proxy", + frame: { + id: 3, + method: "thread/resume", + params: { + threadId: nativeThreadId, + excludeTurns: true, + modelProvider, + config: CodexAdapterV2.CODEX_THREAD_CONFIG, + }, + }, + }, + { + type: "emit_inbound", + label: "resume result", + frame: { + id: 3, + result: { + thread: { id: nativeThreadId, updatedAt: 1782622450, turns: [] }, + }, + }, + }, + ], + }), + undefined, + undefined, + undefined, + undefined, + modelProvider, + ); + + const resumed = yield* harness.runtime.resumeThread({ + providerThread: harness.providerThread, + }); + + assert.equal(resumed.nativeThreadRef?.nativeId, nativeThreadId); + }).pipe( + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ), + ); + // regression: #15389 archived native Codex sessions resume via thread/unarchive it.effect.each( ( diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 821cf051c323..3b94b8394642 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -118,7 +118,12 @@ import { shouldPersistProviderEvent, } from "../../provider/EventNdjsonLogger.ts"; import * as ProviderEventLoggers from "@t3tools/provider-core/server/ProviderEventLoggers"; -import { codexAppServerArgs, resolveCodexLaunchArgs } from "../../provider/codexLaunchArgs.ts"; +import { + applyCodexProxyRuntime, + CODEX_PROXY_MODEL_PROVIDER, + codexAppServerArgs, + resolveCodexLaunchArgs, +} from "../../provider/codexLaunchArgs.ts"; import { mergeProviderInstanceEnvironment } from "@t3tools/provider-core/server/instanceEnvironment"; import * as McpProviderSession from "@t3tools/provider-core/server/mcpSession"; import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions"; @@ -1337,15 +1342,19 @@ function codexThreadRuntimeParams(input: { readonly mcpSession: McpProviderSession.McpProviderSessionConfig | undefined; readonly modelSelection?: { readonly model: string }; readonly runtimePolicy?: ProviderAdapter.ProviderAdapterV2RuntimePolicy; + readonly modelProvider?: string | undefined; }): { readonly cwd?: string; readonly model?: string; + readonly modelProvider?: string; readonly config: Readonly>; } { const { mcpSession } = input; return { ...(input.runtimePolicy?.cwd == null ? {} : { cwd: input.runtimePolicy.cwd }), ...(input.modelSelection === undefined ? {} : { model: input.modelSelection.model }), + // Without it, a resume reuses the provider stored in the rollout and bypasses the proxy. + ...(input.modelProvider === undefined ? {} : { modelProvider: input.modelProvider }), config: { ...CODEX_THREAD_CONFIG, ...(mcpSession === undefined @@ -1562,7 +1571,13 @@ export type CodexAdapterV2DriverEnv = | ServerConfig; export const createCodexAdapterV2 = Effect.fn("CodexAdapterV2.createCodexAdapterV2")(function* ( - { instanceId, environment, enabled, config }: ProviderAdapterDriverCreateInput, + { + instanceId, + environment, + enabled, + config, + proxy, + }: ProviderAdapterDriverCreateInput, hooks: Pick = {}, ) { const clientFactory = yield* CodexAppServerClientFactory; @@ -1586,8 +1601,13 @@ export const createCodexAdapterV2 = Effect.fn("CodexAdapterV2.createCodexAdapter ), ); + const routed = applyCodexProxyRuntime( + config, + yield* mergeProviderInstanceEnvironment(environment, hostEnvironment), + proxy, + ); const settings = { - ...config, + ...routed.config, enabled, binaryPath: expandHomePath(config.binaryPath, yield* HostProcess.HomeDirectory), homePath: homeLayout.effectiveHomePath ?? "", @@ -1596,7 +1616,8 @@ export const createCodexAdapterV2 = Effect.fn("CodexAdapterV2.createCodexAdapter return yield* makeCodexAdapterV2({ instanceId, settings, - environment: yield* mergeProviderInstanceEnvironment(environment, hostEnvironment), + environment: routed.environment, + ...(proxy === undefined ? {} : { modelProvider: CODEX_PROXY_MODEL_PROVIDER }), clientFactory, crypto, fileSystem, @@ -1613,6 +1634,8 @@ export interface CodexAdapterV2Options { readonly environment: NodeJS.ProcessEnv; readonly clientFactory: CodexAppServerClientFactoryShape; readonly onUsageLimits?: ManagedServerProvider["applyUsageLimits"]; + /** Pins thread start and resume to this Codex model provider. */ + readonly modelProvider?: string | undefined; readonly resolveRuntime?: Effect.Effect; readonly compatibility?: ( version: string, @@ -3045,6 +3068,9 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( .request("thread/resume", { threadId: input.nativeThreadId, excludeTurns: true, + ...(adapterOptions.modelProvider === undefined + ? {} + : { modelProvider: adapterOptions.modelProvider }), }) .pipe(Effect.flatMap(decodeCodexChildModel)), ), @@ -6957,6 +6983,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( client.request( "thread/start", codexThreadRuntimeParams({ + modelProvider: adapterOptions.modelProvider, mcpSession, modelSelection: threadInput.modelSelection, runtimePolicy: threadInput.runtimePolicy, @@ -6989,6 +7016,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( threadId: nativeThreadId, excludeTurns: true, ...codexThreadRuntimeParams({ + modelProvider: adapterOptions.modelProvider, mcpSession: yield* readMcpSession( threadInput.threadId ?? threadInput.providerThread.appThreadId, ), @@ -7776,6 +7804,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( threadId, excludeTurns: true, ...codexThreadRuntimeParams({ + modelProvider: adapterOptions.modelProvider, mcpSession: yield* readMcpSession(threadInput.providerThread.appThreadId), modelSelection: input.modelSelection, runtimePolicy: input.runtimePolicy, @@ -7828,6 +7857,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( ? {} : { lastTurnId: boundary.lastTurnId }), ...codexThreadRuntimeParams({ + modelProvider: adapterOptions.modelProvider, mcpSession: yield* mcpSessions.read(threadInput.targetThreadId), ...(threadInput.modelSelection === undefined ? {} diff --git a/apps/server/src/orchestration-v2/ProviderAdapterRegistry.ts b/apps/server/src/orchestration-v2/ProviderAdapterRegistry.ts index a04c1df88727..44c8ea2c330e 100644 --- a/apps/server/src/orchestration-v2/ProviderAdapterRegistry.ts +++ b/apps/server/src/orchestration-v2/ProviderAdapterRegistry.ts @@ -42,6 +42,9 @@ export interface ProviderAdapterRegistryV2Shape { readonly get: ( instanceId: ProviderInstanceId, ) => Effect.Effect; + readonly getRuntimeIdentity?: ( + instanceId: ProviderInstanceId, + ) => Effect.Effect; readonly list: () => Effect.Effect>; readonly getMetadata?: (instanceId: ProviderInstanceId) => Effect.Effect< { @@ -68,6 +71,11 @@ export const layerFromProviderInstanceRegistry: Layer.Layer< Effect.gen(function* () { const instances = yield* ProviderInstanceRegistry.ProviderInstanceRegistry; return ProviderAdapterRegistryV2.of({ + getRuntimeIdentity: Effect.fn("ProviderAdapterRegistryV2.getRuntimeIdentity")((instanceId) => + instances + .getInstance(instanceId) + .pipe(Effect.map((instance) => instance?.orchestrationAdapter)), + ), get: Effect.fn("ProviderAdapterRegistryV2.get")(function (instanceId) { return instances.getInstance(instanceId).pipe( Effect.flatMap((instance) => { diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index 5bbb3926e2b6..a295e7f761a7 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -464,6 +464,7 @@ function layerTest(input: { readonly state: Ref.Ref; readonly idleTimeoutMs: number; readonly maxIdlePinMs?: number; + readonly runtimeIdentity?: Ref.Ref; readonly failEventStream?: boolean; readonly capabilities?: OrchestrationV2ProviderCapabilities; readonly mcpConfigs?: Ref.Ref< @@ -528,7 +529,17 @@ function layerTest(input: { ...(input.scopeCloseReached === undefined ? {} : { scopeCloseReached: input.scopeCloseReached }), - }).pipe(Effect.map(ProviderAdapterRegistry.layerSingle)), + }).pipe( + Effect.map((adapter) => + input.runtimeIdentity === undefined + ? ProviderAdapterRegistry.layerSingle(adapter) + : Layer.succeed(ProviderAdapterRegistry.ProviderAdapterRegistryV2, { + get: () => Effect.succeed(adapter), + list: () => Effect.succeed([adapter.instanceId]), + getRuntimeIdentity: () => Ref.get(input.runtimeIdentity!), + }), + ), + ), ); const layerConfiguredMcpRegistry = input.pauseResolve === undefined @@ -3411,6 +3422,51 @@ it.effect( }), ); +it.effect( + "ProviderSessionManagerV2 replaces a cached session when its provider route changes", + () => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + const runtimeIdentity = yield* Ref.make({}); + const pendingWork = yield* Ref.make(false); + yield* Effect.gen(function* () { + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + const eventSink = yield* EventSink.EventSinkV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const threadId = ThreadId.make("proxy-route-change"); + const now = yield* DateTime.now; + yield* eventSink.write({ + events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })], + }); + const providerSessionId = idAllocator.derive.providerSession({ + providerInstanceId: modelSelection.instanceId, + }); + const input = { threadId, providerSessionId, modelSelection, runtimePolicy }; + const direct = yield* manager.open(input); + assert.strictEqual(yield* manager.open(input), direct); + yield* Ref.set(runtimeIdentity, {}); + yield* Ref.set(pendingWork, true); + assert.isTrue(Exit.isFailure(yield* Effect.exit(manager.open(input)))); + assert.equal((yield* Ref.get(state)).closeCount, 0); + yield* Ref.set(pendingWork, false); + const proxied = yield* manager.open(input); + assert.notStrictEqual(proxied, direct); + assert.equal((yield* Ref.get(state)).openCount, 2); + assert.equal((yield* Ref.get(state)).closeCount, 1); + assert.strictEqual(yield* manager.open(input), proxied); + }).pipe( + Effect.provide( + layerTest({ + state, + runtimeIdentity, + idleTimeoutMs: 60000, + hasPendingBackgroundWork: Ref.get(pendingWork), + }), + ), + ); + }), +); + it.effect( "ProviderSessionManagerV2 invalidates only errored threads and keeps reusable session errors cached", () => diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index 03a7278774a6..852a03c3b28d 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -216,6 +216,7 @@ export class ProviderSessionManagerV2 extends Context.Service< >()("t3/orchestration-v2/ProviderSessionManager/ProviderSessionManagerV2") {} interface LiveSessionEntry { + readonly runtimeIdentity: object | undefined; readonly attachedThreadIds: ReadonlySet; readonly loadedProviderThreadKeyByThread: ReadonlyMap; /** Scope revocation to recorded credential ids so superseded sessions cannot revoke replacements opened before detach. */ @@ -2135,7 +2136,44 @@ export const layerWithOptions = ( } } const key = sessionKey(input.providerSessionId); - const existing = (yield* Ref.get(sessions)).get(key); + const runtimeIdentity = + registry.getRuntimeIdentity === undefined + ? undefined + : yield* registry.getRuntimeIdentity(input.modelSelection.instanceId); + let existing = (yield* Ref.get(sessions)).get(key); + if (existing !== undefined && existing.runtimeIdentity !== runtimeIdentity) { + const changedSettings = () => + new ProviderSessionOpenError({ + instanceId: input.modelSelection.instanceId, + providerSessionId: input.providerSessionId, + cause: + "Provider settings changed. Wait for its active work to finish before starting another turn.", + }); + const pending = + existing.runtime.hasPendingBackgroundWork === undefined + ? false + : yield* existing.runtime.hasPendingBackgroundWork; + if (existing.busyTurns.size > 0 || pending) { + return yield* changedSettings(); + } + yield* releaseEntry({ + providerSessionId: input.providerSessionId, + reason: "manual_shutdown", + detail: "Provider settings changed.", + onlyIfIdleGeneration: existing.idleGeneration, + }).pipe( + Effect.mapError( + (cause) => + new ProviderSessionOpenError({ + instanceId: input.modelSelection.instanceId, + providerSessionId: input.providerSessionId, + cause, + }), + ), + ); + if ((yield* Ref.get(sessions)).has(key)) return yield* changedSettings(); + existing = undefined; + } if (existing !== undefined) { const attached = existing.attachedThreadIds.has(input.threadId); if (!attached && !existing.supportsMultipleProviderThreads) { @@ -2234,6 +2272,7 @@ export const layerWithOptions = ( >(new Map()); const exposedRuntime = decorateRuntime(runtime, eventSubscribers); const entry: LiveSessionEntry = { + runtimeIdentity, attachedThreadIds: new Set([input.threadId]), loadedProviderThreadKeyByThread: new Map(), mcpCredentialIdByThread: diff --git a/apps/server/src/provider/ClaudeProvider.ts b/apps/server/src/provider/ClaudeProvider.ts index 95f5c6709a4e..055b62281e6f 100644 --- a/apps/server/src/provider/ClaudeProvider.ts +++ b/apps/server/src/provider/ClaudeProvider.ts @@ -40,6 +40,10 @@ import { hasClaudeManagedMcpConfig, makeClaudeEnvironment } from "./Drivers/Clau import { discoverClaudeSkills, readClaudeSettingsBaseUrl } from "./Drivers/ClaudeSkills.ts"; import type { ProviderWorkspaceSnapshot } from "@t3tools/provider-core/server/driver"; import { makeUnavailableUsageLimits } from "@t3tools/provider-core/server/usageLimits"; +import { + PROXY_AUTH_METADATA, + PROXY_LIMITS_MESSAGE, +} from "@t3tools/provider-core/server/providerProxy"; import { type ClaudeScopedLimitNames, claudeUsageResponseToLimits, @@ -459,6 +463,7 @@ export const checkClaudeProviderStatus = Effect.fn("checkClaudeProviderStatus")( /** Shared with the adapter so turn events reuse the scoped-bucket names this probe saw. */ scopedLimitNames?: Ref.Ref, resolveResetCredits?: (version: string) => Effect.Effect, + routedThroughProxy = false, ): Effect.fn.Return< ServerProviderDraft, never, @@ -628,26 +633,32 @@ 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) ?? - claudeAuthMetadata({ - subscriptionType: capabilities.subscriptionType, - authMethod: capabilities.tokenSource, - }); + const authMetadata = routedThroughProxy + ? PROXY_AUTH_METADATA + : (apiProviderAuthMetadata(capabilities.apiProvider) ?? + (customApiEndpoint ? { type: "customEndpoint", label: "Custom API endpoint" } : undefined) ?? + claudeAuthMetadata({ + subscriptionType: capabilities.subscriptionType, + authMethod: capabilities.tokenSource, + })); const usageCheckedAt = capabilities.usageCheckedAt ?? checkedAt; - const usageLimits = !capabilities.usage - ? makeUnavailableUsageLimits({ checkedAt, reason: "probeFailed" }) - : scopedLimitNames - ? yield* recordClaudeUsageResponse(scopedLimitNames, { - response: capabilities.usage, - checkedAt: usageCheckedAt, - }) - : claudeUsageResponseToLimits({ - response: capabilities.usage, - checkedAt: usageCheckedAt, - }).limits; + const usageLimits = routedThroughProxy + ? makeUnavailableUsageLimits({ + checkedAt, + reason: "unsupported", + message: PROXY_LIMITS_MESSAGE, + }) + : !capabilities.usage + ? makeUnavailableUsageLimits({ checkedAt, reason: "probeFailed" }) + : scopedLimitNames + ? yield* recordClaudeUsageResponse(scopedLimitNames, { + response: capabilities.usage, + checkedAt: usageCheckedAt, + }) + : claudeUsageResponseToLimits({ + response: capabilities.usage, + checkedAt: usageCheckedAt, + }).limits; const resetCredits = resolveResetCredits && capabilities.subscriptionType && diff --git a/apps/server/src/provider/CodexProvider.ts b/apps/server/src/provider/CodexProvider.ts index ed9e3d0cde68..e02f3fad8fc9 100644 --- a/apps/server/src/provider/CodexProvider.ts +++ b/apps/server/src/provider/CodexProvider.ts @@ -46,6 +46,10 @@ import { } from "@t3tools/provider-core/server/snapshotProbe"; import { expandHomePath } from "@t3tools/provider-core/server/pathExpansion"; import { makeUnavailableUsageLimits } from "@t3tools/provider-core/server/usageLimits"; +import { + PROXY_AUTH_METADATA, + PROXY_LIMITS_MESSAGE, +} from "@t3tools/provider-core/server/providerProxy"; import { codexRateLimitsFailureMessage, codexRateLimitsToLimits, @@ -749,6 +753,7 @@ export const checkCodexProviderStatus = Effect.fn("checkCodexProviderStatus")(fu > = probeCodexAppServerProvider, environment?: NodeJS.ProcessEnv, managedAuth?: ServerProvider["auth"], + routedThroughProxy = false, ): Effect.fn.Return< ServerProviderDraft, ServerSettingsError, @@ -781,7 +786,7 @@ export const checkCodexProviderStatus = Effect.fn("checkCodexProviderStatus")(fu launchArgs: resolveCodexLaunchArgs(codexSettings.launchArgs, resolvedEnvironment), cwd: process.cwd(), environment: resolvedEnvironment, - ...(managedAuth ? { skipNativeUsage: true } : {}), + ...(managedAuth || routedThroughProxy ? { skipNativeUsage: true } : {}), }).pipe( // Process teardown can outlast the work budget without discarding a completed snapshot. Effect.timeoutOption(Duration.millis(CODEX_AUTH_PROBE_TIMEOUT_MS)), @@ -841,9 +846,20 @@ export const checkCodexProviderStatus = Effect.fn("checkCodexProviderStatus")(fu const snapshot = probeResult.success.value; const accountStatus = managedAuth ? { status: "ready" as const, auth: managedAuth, message: undefined } - : accountProbeStatus(snapshot); - const usageLimits = - snapshot.account.account?.type === "apiKey" + : routedThroughProxy + ? { + status: "ready" as const, + auth: { status: "authenticated" as const, ...PROXY_AUTH_METADATA }, + message: undefined, + } + : accountProbeStatus(snapshot); + const usageLimits = routedThroughProxy + ? makeUnavailableUsageLimits({ + checkedAt, + reason: "unsupported", + message: PROXY_LIMITS_MESSAGE, + }) + : snapshot.account.account?.type === "apiKey" ? makeUnavailableUsageLimits({ checkedAt, reason: "unsupported" }) : snapshot.rateLimits === undefined || "failure" in snapshot.rateLimits ? makeUnavailableUsageLimits({ diff --git a/apps/server/src/provider/Drivers/ClaudeDriver.ts b/apps/server/src/provider/Drivers/ClaudeDriver.ts index e1fbcd30080f..e8df26c4fc78 100644 --- a/apps/server/src/provider/Drivers/ClaudeDriver.ts +++ b/apps/server/src/provider/Drivers/ClaudeDriver.ts @@ -56,6 +56,7 @@ import { type ProviderSnapshotSettings, } from "@t3tools/provider-core/server/snapshotSettings"; import { + applyClaudeProxyEnvironment, makeClaudeCapabilitiesCacheKey, makeClaudeContinuationGroupKey, resolveClaudeHomePath, @@ -120,6 +121,7 @@ export const ClaudeDriver: ProviderDriver snapshot.applyUsageLimits(update) }, ).pipe( @@ -195,6 +201,7 @@ export const ClaudeDriver: ProviderDriver(undefined); const capabilitiesProbeCache = yield* Cache.makeWith( () => - probeClaudeCapabilities(effectiveConfig, processEnv, stateDir).pipe( + probeClaudeCapabilities(effectiveConfig, processEnv, stateDir, proxy === undefined).pipe( Effect.flatMap((capabilities) => ClaudeResetCredits.readClaudeOrganizationId(accountConfigPath).pipe( Effect.map((workspaceId) => ({ @@ -272,6 +279,7 @@ export const ClaudeDriver: ProviderDriver diff --git a/apps/server/src/provider/Drivers/ClaudeHome.test.ts b/apps/server/src/provider/Drivers/ClaudeHome.test.ts index b9eb3b806ce0..cf850d499122 100644 --- a/apps/server/src/provider/Drivers/ClaudeHome.test.ts +++ b/apps/server/src/provider/Drivers/ClaudeHome.test.ts @@ -6,12 +6,40 @@ import * as Effect from "effect/Effect"; import * as Path from "effect/Path"; import { + applyClaudeProxyEnvironment, makeClaudeCapabilitiesCacheKey, makeClaudeContinuationGroupKey, makeClaudeEnvironment, resolveClaudeHomePath, } from "./ClaudeHome.ts"; +describe("applyClaudeProxyEnvironment", () => { + it("replaces every ambient route with the proxy", () => { + expect( + applyClaudeProxyEnvironment( + { + PATH: "/bin", + ANTHROPIC_API_KEY: "sk-ant-direct", + ANTHROPIC_BASE_URL: "https://api.anthropic.com", + CLAUDE_CODE_OAUTH_TOKEN: "oauth", + CLAUDE_CODE_USE_BEDROCK: "1", + }, + { url: "http://proxy:8317", apiKey: "sk-proxy" }, + ), + ).toEqual({ + PATH: "/bin", + ANTHROPIC_API_KEY: "", + ANTHROPIC_BASE_URL: "http://proxy:8317", + ANTHROPIC_AUTH_TOKEN: "sk-proxy", + }); + // Claude Code reports "Not logged in" without contacting a proxy when no credential is set. + expect( + applyClaudeProxyEnvironment({}, { url: "http://proxy:8317", apiKey: "" }) + .ANTHROPIC_AUTH_TOKEN, + ).toBe("t3-proxy"); + }); +}); + it.layer(NodeServices.layer)("ClaudeHome", (it) => { describe("Claude home resolution", () => { it.effect("resolves configured Claude HOME and stamps continuation/cache keys with it", () => diff --git a/apps/server/src/provider/Drivers/ClaudeHome.ts b/apps/server/src/provider/Drivers/ClaudeHome.ts index bd81aa10bce7..a6affb50d5be 100644 --- a/apps/server/src/provider/Drivers/ClaudeHome.ts +++ b/apps/server/src/provider/Drivers/ClaudeHome.ts @@ -5,6 +5,10 @@ import * as Path from "effect/Path"; import * as Schema from "effect/Schema"; import { expandHomePath } from "@t3tools/provider-core/server/pathExpansion"; +import { + providerProxyToken, + type ProviderProxyRoute, +} from "@t3tools/provider-core/server/providerProxy"; import * as HostProcess from "@t3tools/shared/HostProcess"; const quotePath = Schema.encodeSync(Schema.fromJsonString(Schema.String)); @@ -42,6 +46,37 @@ export const makeClaudeEnvironment = Effect.fn("makeClaudeEnvironment")(function }; }); +// Ambient Bedrock, Vertex, or API-key settings would send requests around the proxy. +const CLAUDE_ROUTE_OVERRIDE_VARIABLES = [ + "CLAUDE_CODE_OAUTH_TOKEN", + "CLAUDE_CODE_USE_BEDROCK", + "CLAUDE_CODE_USE_VERTEX", + "CLAUDE_CODE_USE_FOUNDRY", +] as const; + +export const claudeProxySettingsEnvironment = (proxy: ProviderProxyRoute) => ({ + ANTHROPIC_BASE_URL: proxy.url, + ANTHROPIC_AUTH_TOKEN: providerProxyToken(proxy), + ANTHROPIC_API_KEY: "", + ...Object.fromEntries(CLAUDE_ROUTE_OVERRIDE_VARIABLES.map((name) => [name, ""])), +}); + +export const applyClaudeProxyEnvironment = ( + environment: NodeJS.ProcessEnv, + proxy: ProviderProxyRoute | undefined, +): NodeJS.ProcessEnv => { + if (proxy === undefined) return environment; + const next: NodeJS.ProcessEnv = { + ...environment, + ANTHROPIC_BASE_URL: proxy.url, + ANTHROPIC_AUTH_TOKEN: providerProxyToken(proxy), + // Empty, not removed: Claude Code otherwise falls back to a saved API key. + ANTHROPIC_API_KEY: "", + }; + for (const name of CLAUDE_ROUTE_OVERRIDE_VARIABLES) delete next[name]; + return next; +}; + /** managed-mcp.json excludes other MCP config and forbids --strict-mcp-config; omit the flag when present. */ export const hasClaudeManagedMcpConfig = Effect.fn("hasClaudeManagedMcpConfig")(function* () { const fileSystem = yield* FileSystem.FileSystem; diff --git a/apps/server/src/provider/Drivers/CodexDriver.ts b/apps/server/src/provider/Drivers/CodexDriver.ts index 14cb4b6d25af..d1981c7b2ed2 100644 --- a/apps/server/src/provider/Drivers/CodexDriver.ts +++ b/apps/server/src/provider/Drivers/CodexDriver.ts @@ -26,7 +26,7 @@ import { probeCodexSkillsForCwd, withCodexAppServerClient, } from "../CodexProvider.ts"; -import { resolveCodexLaunchArgs } from "../codexLaunchArgs.ts"; +import { applyCodexProxyRuntime, resolveCodexLaunchArgs } from "../codexLaunchArgs.ts"; import { makeManagedServerProvider } from "@t3tools/provider-core/server/managedProvider"; import * as ModelCatalog from "@t3tools/provider-core/server/ModelCatalog"; import { applyCodexModelCatalog } from "../codexModelCatalog.ts"; @@ -131,7 +131,15 @@ export const CodexDriver: ProviderDriver< environment, enabled, config, + proxy, }) { + if (config.setupMode === "managed" && proxy !== undefined) { + return yield* new ProviderDriverError({ + driver: DRIVER_KIND, + instanceId, + detail: "T3-managed ChatGPT sign-in cannot route through a proxy. Turn off Use proxy.", + }); + } if (config.setupMode === "managed") return yield* makeManagedCodexProvider({ instanceId, @@ -150,10 +158,12 @@ export const CodexDriver: ProviderDriver< const installation = yield* CodexInstallation.CodexInstallation; const modelCatalog = yield* ModelCatalog.ModelCatalog; const currentCatalog = modelCatalog.current(DRIVER_KIND); - const processEnv = yield* mergeProviderInstanceEnvironment( - environment, - yield* HostProcess.Environment, + const routed = applyCodexProxyRuntime( + config, + yield* mergeProviderInstanceEnvironment(environment, yield* HostProcess.Environment), + proxy, ); + const processEnv = routed.environment; const homeLayout = yield* resolveCodexHomeLayout(config); const continuationIdentity = codexContinuationIdentity(homeLayout); const stampIdentity = withInstanceIdentity({ @@ -175,7 +185,7 @@ export const CodexDriver: ProviderDriver< ), ); const effectiveConfig = { - ...config, + ...routed.config, enabled, binaryPath: expandHomePath(config.binaryPath, yield* HostProcess.HomeDirectory), homePath: homeLayout.effectiveHomePath ?? "", @@ -202,6 +212,7 @@ export const CodexDriver: ProviderDriver< environment, enabled, config, + proxy, }, { onUsageLimits: (update) => snapshot.applyUsageLimits(update), @@ -224,7 +235,13 @@ export const CodexDriver: ProviderDriver< const checkProvider = modelCatalog.refreshInBackground.pipe( Effect.andThen( Effect.zipWith( - checkCodexProviderStatus(effectiveConfig, undefined, processEnv), + checkCodexProviderStatus( + effectiveConfig, + undefined, + processEnv, + undefined, + proxy !== undefined, + ), currentCatalog, (draft, catalog) => stampIdentity(applyCodexModelCatalog(draft, catalog)), { concurrent: true }, diff --git a/apps/server/src/provider/ProviderInstanceRegistry.test.ts b/apps/server/src/provider/ProviderInstanceRegistry.test.ts index cf9a6c4f8e17..19dec29cbeaf 100644 --- a/apps/server/src/provider/ProviderInstanceRegistry.test.ts +++ b/apps/server/src/provider/ProviderInstanceRegistry.test.ts @@ -480,6 +480,7 @@ describe("ProviderInstanceRegistry — multi-instance codex slice", () => { const claudeId = ProviderInstanceId.make("claude_probe_recovery"); let probeCount = 0; let successfulProbeCount = 0; + let usageProbeCount = 0; let failProbe = false; const query = vi.spyOn(ClaudeSdk, "query").mockImplementation( () => @@ -504,12 +505,15 @@ describe("ProviderInstanceRegistry — multi-instance codex slice", () => { }, }); }, - usage_EXPERIMENTAL_MAY_CHANGE_DO_NOT_RELY_ON_THIS_API_YET: async () => ({ - rate_limits_available: true, - rate_limits: { - five_hour: { utilization: 12, resets_at: "2026-07-18T14:39:00Z" }, - }, - }), + usage_EXPERIMENTAL_MAY_CHANGE_DO_NOT_RELY_ON_THIS_API_YET: async () => { + usageProbeCount++; + return { + rate_limits_available: true, + rate_limits: { + five_hour: { utilization: 12, resets_at: "2026-07-18T14:39:00Z" }, + }, + }; + }, }) as ReturnType, ); yield* Effect.addFinalizer(() => Effect.sync(() => query.mockRestore())); @@ -597,6 +601,30 @@ describe("ProviderInstanceRegistry — multi-instance codex slice", () => { expect.arrayContaining([expect.objectContaining({ name: "recovered" })]), ); expect(probeCount).toBeGreaterThan(sameAccountFailureProbeCount); + expect(usageProbeCount).toBeGreaterThan(0); + failProbe = false; + const beforeProxy = usageProbeCount; + const { registry: proxiedRegistry } = yield* makeProviderInstanceRegistry({ + drivers: [ClaudeDriver], + configMap: { + [claudeId]: { + driver: ProviderDriverKind.make("claudeAgent"), + enabled: true, + useProxy: true, + proxy: { url: "http://proxy.test:8317", apiKey: "proxy-test-key" }, + config: makeClaudeConfig({ + enabled: true, + binaryPath: fixtures.claudeBinaryPath, + homePath: fixtures.claudeHomePath, + }), + }, + }, + }); + const proxied = yield* proxiedRegistry.getInstance(claudeId); + expect(proxied).toBeDefined(); + const proxySnapshot = yield* proxied!.snapshot.refresh; + expect(proxySnapshot.auth.type).toBe("proxy"); + expect(usageProbeCount).toBe(beforeProxy); }).pipe(Effect.provide(layerTest)), ); @@ -804,6 +832,39 @@ describe("ProviderInstanceRegistry — multi-instance codex slice", () => { expect(ghost.unavailableReason).toMatch(/ghostDriver/); }).pipe(Effect.provide(layerTest)), ); + + it.live( + "rebuilds a proxied instance when the proxy changes and refuses one without a proxy", + () => + Effect.gen(function* () { + const codexId = ProviderInstanceId.make("codex_proxied"); + const entry = { + driver: ProviderDriverKind.make("codex"), + useProxy: true, + config: makeCodexConfig({ enabled: false }), + }; + const { registry, mutator } = yield* makeProviderInstanceRegistry({ + drivers: [CodexDriver], + configMap: { [codexId]: { ...entry, proxy: { url: "http://a:8317", apiKey: "k" } } }, + }); + const first = yield* registry.getInstance(codexId); + expect(first).toBeDefined(); + + yield* mutator.reconcile({ + [codexId]: { ...entry, proxy: { url: "http://b:8317", apiKey: "k" } }, + }); + const second = yield* registry.getInstance(codexId); + expect(second).toBeDefined(); + expect(second).not.toBe(first); + + // Running on the instance's own login instead would spend a different account. + yield* mutator.reconcile({ [codexId]: entry }); + expect(yield* registry.getInstance(codexId)).toBeUndefined(); + const [unavailable] = yield* registry.listUnavailable; + expect(unavailable?.instanceId).toBe(codexId); + expect(unavailable?.unavailableReason).toMatch(/no proxy/); + }).pipe(Effect.provide(layerTest)), + ); }); const DisabledByDefaultConfig = Schema.Struct({ diff --git a/apps/server/src/provider/ProviderInstanceRegistry.ts b/apps/server/src/provider/ProviderInstanceRegistry.ts index 57a218a0dafe..36afc4d6bc2c 100644 --- a/apps/server/src/provider/ProviderInstanceRegistry.ts +++ b/apps/server/src/provider/ProviderInstanceRegistry.ts @@ -1,8 +1,8 @@ import { + PROXY_ROUTED_DRIVERS, providerInstanceConfigEnabledFlag, ProviderInstanceId, type ProviderInstanceConfig, - type ProviderInstanceConfigMap, type ProviderInstanceEnvironment, type ProviderDriverKind, type ResolvedProviderInstanceEnvironment, @@ -51,7 +51,7 @@ interface InstanceSecrets { interface LiveEntry { readonly instance: ProviderInstance; readonly scope: Scope.Closeable; - readonly entry: ProviderInstanceConfig; + readonly entry: ProviderInstanceRegistryMutator.ProviderInstanceRuntimeEntry; readonly secrets: InstanceSecrets; } @@ -61,8 +61,24 @@ interface RegistryState { readonly changes: PubSub.PubSub; } -const entryEqual = (a: ProviderInstanceConfig, b: ProviderInstanceConfig): boolean => - Equal.equals(a, b); +const entryEqual = ( + a: ProviderInstanceRegistryMutator.ProviderInstanceRuntimeEntry, + b: ProviderInstanceRegistryMutator.ProviderInstanceRuntimeEntry, +): boolean => Equal.equals(a, b); + +const proxyUnavailableReason = ( + entry: ProviderInstanceRegistryMutator.ProviderInstanceRuntimeEntry, +): string | undefined => { + if (entry.useProxy !== true) return undefined; + if (!PROXY_ROUTED_DRIVERS.has(entry.driver)) { + return `Driver '${entry.driver}' cannot route through a proxy. Turn off Use proxy for this instance.`; + } + // Falling back to the instance's own login would silently spend a different account. + if (entry.proxy === undefined) { + return "Use proxy is on, but this environment has no proxy. Choose one in Settings → Providers → Proxy, or turn off Use proxy."; + } + return undefined; +}; const secretReferencesIn = (environment: ProviderInstanceEnvironment): Array => environment.flatMap((variable) => (typeof variable.value === "string" ? [] : [variable.value])); @@ -149,7 +165,7 @@ const buildEntry = Effect.fn("ProviderInstanceRegistry.buildEntry")(function* (input: { readonly blockedReason: BlockedReason; readonly parentScope: Scope.Scope; }): (( - configMap: ProviderInstanceConfigMap, + configMap: ProviderInstanceRegistryMutator.ProviderInstanceRuntimeMap, rebuild?: ReadonlySet, deferSecrets?: boolean, ) => Effect.Effect) => { @@ -419,7 +450,7 @@ export const makeProviderInstanceRegistry = Effect.fn( "ProviderInstanceRegistry.makeProviderInstanceRegistry", )(function* (input: { readonly drivers: ReadonlyArray>; - readonly configMap: ProviderInstanceConfigMap; + readonly configMap: ProviderInstanceRegistryMutator.ProviderInstanceRuntimeMap; /** A reason to show instead of running a driver, such as an administrator's policy. */ readonly blockedReason?: BlockedReason | undefined; }): Effect.fn.Return< @@ -559,7 +590,7 @@ export const makeProviderInstanceRegistry = Effect.fn( export const layer = (input: { readonly drivers: ReadonlyArray>; - readonly configMap: ProviderInstanceConfigMap; + readonly configMap: ProviderInstanceRegistryMutator.ProviderInstanceRuntimeMap; readonly blockedReason?: BlockedReason | undefined; }): Layer.Layer< ProviderInstanceRegistry | ProviderInstanceRegistryMutator.ProviderInstanceRegistryMutator, diff --git a/apps/server/src/provider/ProviderInstanceRegistryHydration.test.ts b/apps/server/src/provider/ProviderInstanceRegistryHydration.test.ts index c497e00ac17e..c5fddd35e30f 100644 --- a/apps/server/src/provider/ProviderInstanceRegistryHydration.test.ts +++ b/apps/server/src/provider/ProviderInstanceRegistryHydration.test.ts @@ -77,6 +77,49 @@ describe("deriveProviderInstanceConfigMap", () => { expect(merged[instanceId]).toBeUndefined(); }); + it("attaches the environment's proxy only to instances that use it", () => { + const claudeId = ProviderInstanceId.make("claude_direct"); + const settings = settingsWith({ + providerProxy: { url: " http://proxy:8317/v1/ ", apiKey: "sk-proxy", managementKey: "" }, + providerInstances: { + [instanceId]: { driver: ProviderDriverKind.make("codex"), useProxy: true }, + [claudeId]: { driver: ProviderDriverKind.make("claudeAgent") }, + }, + }); + + const merged = deriveProviderInstanceConfigMap(settings); + + expect(merged[instanceId]?.proxy).toEqual({ url: "http://proxy:8317", apiKey: "sk-proxy" }); + expect(merged[claudeId]).toBe(settings.providerInstances[claudeId]); + expect( + deriveProviderInstanceConfigMap({ + ...settings, + providerProxy: { ...settings.providerProxy, url: "" }, + })[instanceId]?.proxy, + ).toBeUndefined(); + + // The host decides which proxy applies; a managed one carries its own loopback route and key. + const managed = { + url: "http://127.0.0.1:41234", + apiKey: "sk-managed", + managementKey: "mgmt-managed", + }; + const withHost = (host: "off" | "managed" | "external") => ({ + ...settings, + providerProxy: { ...settings.providerProxy, host }, + }); + expect( + deriveProviderInstanceConfigMap(withHost("managed"), managed)[instanceId]?.proxy, + ).toEqual({ url: "http://127.0.0.1:41234", apiKey: "sk-managed" }); + expect(deriveProviderInstanceConfigMap(withHost("managed"))[instanceId]?.proxy).toBeUndefined(); + expect( + deriveProviderInstanceConfigMap(withHost("off"), managed)[instanceId]?.proxy, + ).toBeUndefined(); + expect( + deriveProviderInstanceConfigMap(withHost("external"), managed)[instanceId]?.proxy, + ).toEqual({ url: "http://proxy:8317", apiKey: "sk-proxy" }); + }); + it("ignores a project override that disables an instance, leaving the machine's config in charge", () => { const settings = settingsWith({ providerInstances: { diff --git a/apps/server/src/provider/ProviderInstanceRegistryHydration.ts b/apps/server/src/provider/ProviderInstanceRegistryHydration.ts index 9dd93bd3f585..6d984dfc029b 100644 --- a/apps/server/src/provider/ProviderInstanceRegistryHydration.ts +++ b/apps/server/src/provider/ProviderInstanceRegistryHydration.ts @@ -5,7 +5,6 @@ import { providerInstanceConfigEnabledFlag, resolveProviderInstanceEnabled, type ProviderInstanceConfig, - type ProviderInstanceConfigMap, ServerSettings, } from "@t3tools/contracts"; import * as HostProcess from "@t3tools/shared/HostProcess"; @@ -18,6 +17,11 @@ import * as Stream from "effect/Stream"; import * as ManagedProviders from "../policy/managedProviders.ts"; import * as Settings from "../serverSettings.ts"; +import { + resolveProviderProxyRoute, + type ManagedProxyEndpoint, +} from "../providerProxy/effectiveProxy.ts"; +import * as ManagedProviderProxy from "../providerProxy/ManagedProviderProxy.ts"; import * as SecretResolver from "../secrets/SecretResolver.ts"; import { BUILT_IN_DRIVERS, type BuiltInDriversEnv } from "./builtInDrivers.ts"; import * as ProviderInstanceRegistry from "./ProviderInstanceRegistry.ts"; @@ -38,6 +42,7 @@ type ProviderInstanceRegistryHydrationEnv = | ProviderHost.ProviderHost > | Settings.ServerSettingsService + | ManagedProviderProxy.ManagedProviderProxy | BackgroundPolicy.BackgroundPolicy | ServerConfig.ServerConfig | Path.Path @@ -49,7 +54,8 @@ type ProviderInstanceRegistryHydrationEnv = export const deriveProviderInstanceConfigMap = ( settings: ServerSettings, -): ProviderInstanceConfigMap => { + managedProxy?: ManagedProxyEndpoint, +): ProviderInstanceRegistryMutator.ProviderInstanceRuntimeMap => { const merged: Record = { ...settings.providerInstances }; for (const driver of BUILT_IN_DRIVERS) { @@ -78,7 +84,14 @@ export const deriveProviderInstanceConfigMap = ( } } - return merged as ProviderInstanceConfigMap; + const proxy = resolveProviderProxyRoute(settings, managedProxy); + if (proxy === undefined) return merged; + return Object.fromEntries( + Object.entries(merged).map(([id, entry]) => [ + id, + entry.useProxy === true ? { ...entry, proxy } : entry, + ]), + ); }; export const DEV_REPLAY_INSTANCE_ID = ProviderInstanceId.make("devReplay"); @@ -127,13 +140,16 @@ const layerSettingsWatcher = (settingsChanges: Stream.Stream) => Effect.gen(function* () { const mutator = yield* ProviderInstanceRegistryMutator.ProviderInstanceRegistryMutator; const serverSettings = yield* Settings.ServerSettingsService; - yield* settingsChanges.pipe( + const managedProxy = yield* ManagedProviderProxy.ManagedProviderProxy; + // The managed proxy's endpoint appears once it first starts and moves if its port is taken. + // Its stream replays the current value; reconciling unchanged entries rebuilds nothing. + yield* Stream.merge(settingsChanges, managedProxy.endpointChanges).pipe( Stream.runForEach(() => // The subscription predates the seed read, so a queued change can be // older than the seed. A failed current read must leave the registry intact. - serverSettings.getSettings.pipe( - Effect.flatMap((current) => - mutator.reconcile(deriveProviderInstanceConfigMap(current)), + Effect.all([serverSettings.getSettings, managedProxy.endpoint]).pipe( + Effect.flatMap(([current, endpoint]) => + mutator.reconcile(deriveProviderInstanceConfigMap(current, endpoint)), ), Effect.catchCause((cause) => Effect.logError("ProviderInstanceRegistry reconcile failed", cause), @@ -161,10 +177,11 @@ export const layer: Layer.Layer< const initialSettings: ServerSettings | undefined = yield* serverSettings.getSettings.pipe( Effect.orElseSucceed(() => undefined), ); + const managedProxy = yield* ManagedProviderProxy.ManagedProviderProxy; const initialConfigMap = initialSettings === undefined - ? ({} as ProviderInstanceConfigMap) - : deriveProviderInstanceConfigMap(initialSettings); + ? ({} as ProviderInstanceRegistryMutator.ProviderInstanceRuntimeMap) + : deriveProviderInstanceConfigMap(initialSettings, yield* managedProxy.endpoint); // Drivers launch from the host environment, minus launch overrides a policy ignores. const hostEnvironment = ManagedProviders.providerHostEnvironment( diff --git a/apps/server/src/provider/ProviderInstanceRegistryMutator.ts b/apps/server/src/provider/ProviderInstanceRegistryMutator.ts index 35eb52d3f290..d9c3ab123aca 100644 --- a/apps/server/src/provider/ProviderInstanceRegistryMutator.ts +++ b/apps/server/src/provider/ProviderInstanceRegistryMutator.ts @@ -1,10 +1,17 @@ /** Reconcile closes removed/replaced scopes before creating replacements: at most one live instance per id. */ -import type { ProviderInstanceConfigMap, ProviderInstanceId } from "@t3tools/contracts"; +import type { ProviderInstanceConfig, ProviderInstanceId } from "@t3tools/contracts"; +import type { ProviderProxyRoute } from "@t3tools/provider-core/server/providerProxy"; import * as Context from "effect/Context"; import type * as Effect from "effect/Effect"; +/** Runtime-only inputs ride on the entry, so a change to them rebuilds the instance. */ +export type ProviderInstanceRuntimeEntry = ProviderInstanceConfig & { + readonly proxy?: ProviderProxyRoute; +}; +export type ProviderInstanceRuntimeMap = Readonly>; + export interface ProviderInstanceRegistryMutatorShape { - readonly reconcile: (configMap: ProviderInstanceConfigMap) => Effect.Effect; + readonly reconcile: (configMap: ProviderInstanceRuntimeMap) => Effect.Effect; /** Re-reads the secrets of every instance, or one, and rebuilds those whose secrets changed. */ readonly reloadSecrets: (instanceId?: ProviderInstanceId) => Effect.Effect; } diff --git a/apps/server/src/provider/ProviderRegistry.test.ts b/apps/server/src/provider/ProviderRegistry.test.ts index 5e1bc9e962b5..bf3ee2ad7e85 100644 --- a/apps/server/src/provider/ProviderRegistry.test.ts +++ b/apps/server/src/provider/ProviderRegistry.test.ts @@ -1,6 +1,7 @@ import * as CodexErrors from "effect-codex-app-server/errors"; import { vi } from "vite-plus/test"; import * as ServerSecretStore from "../auth/ServerSecretStore.ts"; +import * as ManagedProviderProxy from "../providerProxy/ManagedProviderProxy.ts"; import * as CodexInstallation from "./CodexInstallation.ts"; import * as CodexManagedTokenBroker from "./CodexManagedTokenBroker.ts"; import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; @@ -337,6 +338,7 @@ const layerTestNodeServices = Layer.mergeAll( managedDirectory: "unused-managed-installation", }), Layer.mock(ServerSecretStore.ServerSecretStore)({}), + ManagedProviderProxy.ManagedProviderProxy.layerTest(), Layer.succeed(ServerEnvironment.ServerEnvironmentIdentity, { getEnvironmentId: Effect.succeed(EnvironmentId.make("00000000-0000-4000-8000-000000000001")), }), diff --git a/apps/server/src/provider/codexLaunchArgs.test.ts b/apps/server/src/provider/codexLaunchArgs.test.ts index ce18c8f3f1bd..cb62bceedfa7 100644 --- a/apps/server/src/provider/codexLaunchArgs.test.ts +++ b/apps/server/src/provider/codexLaunchArgs.test.ts @@ -2,7 +2,11 @@ import * as NodeAssert from "node:assert/strict"; import { describe, it } from "vite-plus/test"; -import { codexAppServerArgs, codexExecLaunchArgs } from "./codexLaunchArgs.ts"; +import { + applyCodexProxyRuntime, + codexAppServerArgs, + codexExecLaunchArgs, +} from "./codexLaunchArgs.ts"; describe("codexAppServerArgs", () => { it("preserves escaped TOML quotes in config values", () => { @@ -27,3 +31,30 @@ describe("codexExecLaunchArgs", () => { ]); }); }); + +describe("applyCodexProxyRuntime", () => { + const proxyProvider = + 'model_providers.t3_proxy={name="T3 proxy",base_url="http://proxy:8317/v1",env_key="T3_PROXY_KEY",wire_api="responses",requires_openai_auth=false,supports_websockets=false}'; + + it("adds the proxy provider after env launch args and keeps the key out of argv", () => { + const { config, environment } = applyCodexProxyRuntime( + { launchArgs: "--ignored" }, + { T3CODE_CODEX_LAUNCH_ARGS: "--strict-config", PATH: "/bin" }, + { url: "http://proxy:8317", apiKey: "sk-proxy" }, + ); + + NodeAssert.deepStrictEqual(codexAppServerArgs(config.launchArgs), [ + "app-server", + "--strict-config", + "-c", + 'model_provider="t3_proxy"', + "-c", + proxyProvider, + ]); + NodeAssert.deepStrictEqual(codexExecLaunchArgs(config.launchArgs).slice(-2), [ + "-c", + proxyProvider, + ]); + NodeAssert.deepStrictEqual(environment, { PATH: "/bin", T3_PROXY_KEY: "sk-proxy" }); + }); +}); diff --git a/apps/server/src/provider/codexLaunchArgs.ts b/apps/server/src/provider/codexLaunchArgs.ts index 33def5ce9a61..7a182197bef1 100644 --- a/apps/server/src/provider/codexLaunchArgs.ts +++ b/apps/server/src/provider/codexLaunchArgs.ts @@ -1,3 +1,8 @@ +import { + providerProxyApiUrl, + providerProxyToken, + type ProviderProxyRoute, +} from "@t3tools/provider-core/server/providerProxy"; import { tokenizeCliArgs } from "@t3tools/shared/cliArgs"; const T3CODE_CODEX_LAUNCH_ARGS_ENV = "T3CODE_CODEX_LAUNCH_ARGS"; @@ -7,6 +12,36 @@ export const resolveCodexLaunchArgs = ( environment: NodeJS.ProcessEnv = process.env, ) => environment[T3CODE_CODEX_LAUNCH_ARGS_ENV]?.trim() || launchArgs?.trim() || ""; +// Codex stores this ID in each rollout and resumes through it; renaming it strands proxied threads. +export const CODEX_PROXY_MODEL_PROVIDER = "t3_proxy"; +const CODEX_PROXY_KEY_ENV = "T3_PROXY_KEY"; + +/** Appends the proxy provider after any configured launch arguments; the key stays out of argv. */ +export const applyCodexProxyRuntime = ( + config: Config, + environment: NodeJS.ProcessEnv, + proxy: ProviderProxyRoute | undefined, +): { readonly config: Config; readonly environment: NodeJS.ProcessEnv } => { + if (proxy === undefined) return { config, environment }; + const proxyArgs = [ + `model_provider="${CODEX_PROXY_MODEL_PROVIDER}"`, + `model_providers.${CODEX_PROXY_MODEL_PROVIDER}={name="T3 proxy",base_url=${JSON.stringify(providerProxyApiUrl(proxy))},env_key="${CODEX_PROXY_KEY_ENV}",wire_api="responses",requires_openai_auth=false,supports_websockets=false}`, + ] + .map((value) => `-c ${JSON.stringify(value)}`) + .join(" "); + const launchArgs = resolveCodexLaunchArgs(config.launchArgs, environment); + const next: NodeJS.ProcessEnv = { + ...environment, + [CODEX_PROXY_KEY_ENV]: providerProxyToken(proxy), + }; + // Folded into launchArgs above; left in place it would replace the proxy arguments. + delete next[T3CODE_CODEX_LAUNCH_ARGS_ENV]; + return { + config: { ...config, launchArgs: launchArgs === "" ? proxyArgs : `${launchArgs} ${proxyArgs}` }, + environment: next, + }; +}; + const codexLaunchArgv = (launchArgs?: string): ReadonlyArray => tokenizeCliArgs(launchArgs); export const codexAppServerArgs = (launchArgs?: string) => [ diff --git a/apps/server/src/providerProxy/CliProxyApiBinary.ts b/apps/server/src/providerProxy/CliProxyApiBinary.ts new file mode 100644 index 000000000000..5810d0985047 --- /dev/null +++ b/apps/server/src/providerProxy/CliProxyApiBinary.ts @@ -0,0 +1,303 @@ +/** The pinned CLIProxyAPI release T3 downloads into T3 home, following the cloudflared pattern. */ +import * as HostProcess from "@t3tools/shared/HostProcess"; +import { isWindowsRenameLock } from "@t3tools/shared/atomicWrite"; +import * as Context from "effect/Context"; +import * as Crypto from "effect/Crypto"; +import * as Effect from "effect/Effect"; +import * as Hex from "effect/encoding/Hex"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Path from "effect/Path"; +import * as Schedule from "effect/Schedule"; +import * as Schema from "effect/Schema"; +import * as Semaphore from "effect/Semaphore"; +import { HttpClient, HttpClientRequest, HttpClientResponse } from "effect/http"; +import { ChildProcess, ChildProcessSpawner } from "effect/process"; + +import * as ServerConfig from "../config.ts"; + +const CLIPROXYAPI_VERSION = "8.0.47"; +const CLIPROXYAPI_PATH_ENV_NAME = "T3CODE_CLIPROXYAPI_PATH"; +const DOWNLOAD_TIMEOUT = "10 minutes"; +const VERSION_PROBE_TIMEOUT = "10 seconds"; +const ACTIVATE_RETRY_COUNT = 40; +const ACTIVATE_RETRY_DELAY = "250 millis"; + +interface ReleaseAsset { + readonly url: string; + readonly sha256: string; + readonly archive: "tgz" | "zip"; +} + +const releaseAsset = (name: string, sha256: string): ReleaseAsset => ({ + url: `https://github.com/router-for-me/CLIProxyAPI/releases/download/v${CLIPROXYAPI_VERSION}/CLIProxyAPI_${CLIPROXYAPI_VERSION}_${name}`, + sha256, + archive: name.endsWith(".zip") ? "zip" : "tgz", +}); + +// From the release's checksums.txt, which matches GitHub's asset digests. +const RELEASE_ASSETS: Readonly>> = { + "darwin-arm64": releaseAsset( + "darwin_aarch64.tar.gz", + "a5b260ec0f843d2291488acd76e5595232da8f3b8ec2d334154f1570f5f3ca08", + ), + "darwin-x64": releaseAsset( + "darwin_amd64.tar.gz", + "2e72cb7b28905b9135563e4fc01af00009ce72a444729eb95bdb634543ce7d41", + ), + "linux-arm64": releaseAsset( + "linux_aarch64.tar.gz", + "da1326ed4434fb6c9609927f317c093f9a4bb482a61e5451e9bea99a0dc7c4bf", + ), + "linux-x64": releaseAsset( + "linux_amd64.tar.gz", + "2b5111dc6a0dfe0d607457787f794330d1b477bc9daa6337c26b0cd032a7827e", + ), + "win32-arm64": releaseAsset( + "windows_aarch64.zip", + "9a947eec7234e5a2e2ce57a9fe55c82328fc823ffd2f41cca85f9da41d11eae0", + ), + "win32-x64": releaseAsset( + "windows_amd64.zip", + "8e23cbb2f9f695411b7df46320855820b352fb886c3c3bdf45ab4634f2f14632", + ), +}; + +export type CliProxyApiBinaryStatus = + | { + readonly status: "available"; + readonly executablePath: string; + readonly source: "override" | "managed"; + readonly version: string; + } + | { readonly status: "missing" } + | { readonly status: "unsupported"; readonly platform: string; readonly arch: string }; +export type AvailableCliProxyApiBinary = Extract< + CliProxyApiBinaryStatus, + { readonly status: "available" } +>; + +export class CliProxyApiInstallError extends Schema.TaggedError()( + "CliProxyApiInstallError", + { + reason: Schema.Literals([ + "download_failed", + "invalid_checksum", + "override_invalid", + "unsupported_platform", + "validation_failed", + "write_failed", + ]), + cause: Schema.optional(Schema.Defect()), + }, +) { + override get message(): string { + switch (this.reason) { + case "download_failed": + return "Could not download CLIProxyAPI."; + case "invalid_checksum": + return "The CLIProxyAPI download did not match the pinned checksum."; + case "override_invalid": + return `${CLIPROXYAPI_PATH_ENV_NAME} does not point to a working CLIProxyAPI.`; + case "unsupported_platform": + return "CLIProxyAPI has no build for this platform."; + case "validation_failed": + return `The downloaded CLIProxyAPI did not report version ${CLIPROXYAPI_VERSION}.`; + case "write_failed": + return "Could not install CLIProxyAPI into T3 home."; + } + } +} + +/** `-help` prints `CLIProxyAPI Version: 8.0.47, Commit: ...` before the usage text. */ +export function parseCliProxyApiVersionOutput(output: string): string | null { + return /\bCLIProxyAPI Version: v?(\d+\.\d+\.\d+)\b/u.exec(output)?.[1] ?? null; +} + +/** T3 writes v8's config layout; v7 and older read top-level host, port, and auth fields. */ +export const isCompatibleCliProxyApiVersion = (version: string): boolean => + version.split(".")[0] === CLIPROXYAPI_VERSION.split(".")[0]; + +export class CliProxyApiBinary extends Context.Service< + CliProxyApiBinary, + { + readonly resolve: Effect.Effect; + /** Downloads the pinned release. Callers own consent: only a user choosing a managed proxy. */ + readonly install: Effect.Effect; + } +>()("t3/providerProxy/CliProxyApiBinary") {} + +const make = Effect.gen(function* () { + const crypto = yield* Crypto.Crypto; + const fileSystem = yield* FileSystem.FileSystem; + const httpClient = yield* HttpClient.HttpClient; + const path = yield* Path.Path; + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const config = yield* ServerConfig.ServerConfig; + const platform = yield* HostProcess.Platform; + const arch = yield* HostProcess.Architecture; + const installSemaphore = yield* Semaphore.make(1); + const asset = RELEASE_ASSETS[`${platform}-${arch}`]; + const executableName = platform === "win32" ? "cli-proxy-api.exe" : "cli-proxy-api"; + const managedPath = path.join( + config.baseDir, + "tools", + "cliproxyapi", + CLIPROXYAPI_VERSION, + `${platform}-${arch}`, + executableName, + ); + + const probeVersion = Effect.fn("CliProxyApiBinary.probeVersion")(function* ( + executablePath: string, + ) { + const info = yield* fileSystem.stat(executablePath).pipe(Effect.option); + if (Option.isNone(info) || info.value.type !== "File") return null; + if (platform !== "win32" && (info.value.mode & 0o111) === 0) return null; + return yield* spawner + .string(ChildProcess.make(executablePath, ["-help"], { stdin: "ignore", stderr: "ignore" })) + .pipe( + Effect.map(parseCliProxyApiVersionOutput), + Effect.timeoutOption(VERSION_PROBE_TIMEOUT), + Effect.map(Option.getOrNull), + Effect.orElseSucceed(() => null), + ); + }); + + const override = Effect.gen(function* () { + const value = (yield* HostProcess.Environment)[CLIPROXYAPI_PATH_ENV_NAME]?.trim(); + return value === undefined || value === "" ? null : value; + }); + + const resolve: CliProxyApiBinary["Service"]["resolve"] = Effect.gen(function* () { + const overridePath = yield* override; + if (overridePath !== null) { + const version = yield* probeVersion(overridePath); + if (version !== null && isCompatibleCliProxyApiVersion(version)) { + return { + status: "available", + executablePath: overridePath, + source: "override", + version, + } satisfies CliProxyApiBinaryStatus; + } + yield* Effect.logWarning( + version === null + ? `${CLIPROXYAPI_PATH_ENV_NAME} does not point to CLIProxyAPI` + : `${CLIPROXYAPI_PATH_ENV_NAME} points to CLIProxyAPI ${version}; T3 Code needs ${CLIPROXYAPI_VERSION.split(".")[0]}.x`, + ); + return { status: "missing" } satisfies CliProxyApiBinaryStatus; + } + if ((yield* probeVersion(managedPath)) === CLIPROXYAPI_VERSION) { + return { + status: "available", + executablePath: managedPath, + source: "managed", + version: CLIPROXYAPI_VERSION, + } satisfies CliProxyApiBinaryStatus; + } + return asset === undefined + ? ({ status: "unsupported", platform, arch } satisfies CliProxyApiBinaryStatus) + : ({ status: "missing" } satisfies CliProxyApiBinaryStatus); + }).pipe(Effect.withSpan("CliProxyApiBinary.resolve")); + + const failWith = + (reason: CliProxyApiInstallError["reason"]) => + (effect: Effect.Effect) => + effect.pipe(Effect.mapError((cause) => new CliProxyApiInstallError({ reason, cause }))); + + const download = Effect.fn("CliProxyApiBinary.download")(function* (release: ReleaseAsset) { + const bytes = yield* httpClient.execute(HttpClientRequest.get(release.url)).pipe( + Effect.flatMap(HttpClientResponse.filterStatusOk), + Effect.flatMap((response) => response.arrayBuffer), + Effect.timeout(DOWNLOAD_TIMEOUT), + failWith("download_failed"), + ); + const data = new Uint8Array(bytes); + const digest = yield* crypto.digest("SHA-256", data).pipe(failWith("validation_failed")); + if (Hex.encode(digest) !== release.sha256) { + return yield* new CliProxyApiInstallError({ reason: "invalid_checksum" }); + } + return data; + }); + + const extract = Effect.fn("CliProxyApiBinary.extract")(function* ( + archivePath: string, + directory: string, + archive: ReleaseAsset["archive"], + ) { + // bsdtar, which Windows ships, also reads zip archives. + const child = yield* spawner.spawn( + ChildProcess.make( + "tar", + [archive === "tgz" ? "-xzf" : "-xf", archivePath, "-C", directory, executableName], + { shell: false, stdin: "ignore", stdout: "ignore", stderr: "ignore" }, + ), + ); + const exitCode = Number(yield* child.exitCode); + if (exitCode !== 0) { + return yield* new CliProxyApiInstallError({ reason: "write_failed" }); + } + }); + + const renameWhenUnlocked = (from: string, to: string) => + fileSystem.rename(from, to).pipe( + Effect.retry({ + times: ACTIVATE_RETRY_COUNT, + schedule: Schedule.spaced(ACTIVATE_RETRY_DELAY), + while: (error) => platform === "win32" && isWindowsRenameLock(error), + }), + ); + + const installUnlocked = Effect.gen(function* () { + const current = yield* resolve; + if (current.status === "available") return current; + if ((yield* override) !== null) { + return yield* new CliProxyApiInstallError({ reason: "override_invalid" }); + } + if (asset === undefined) { + return yield* new CliProxyApiInstallError({ reason: "unsupported_platform" }); + } + const managedDirectory = path.dirname(managedPath); + yield* fileSystem + .makeDirectory(managedDirectory, { recursive: true }) + .pipe(failWith("write_failed")); + const staging = yield* fileSystem + .makeTempDirectoryScoped({ directory: managedDirectory, prefix: ".install-" }) + .pipe(failWith("write_failed")); + const archivePath = path.join(staging, asset.archive === "tgz" ? "release.tgz" : "release.zip"); + yield* Effect.logInfo("Downloading CLIProxyAPI", { version: CLIPROXYAPI_VERSION }); + const bytes = yield* download(asset); + yield* fileSystem.writeFile(archivePath, bytes).pipe(failWith("write_failed")); + yield* extract(archivePath, staging, asset.archive).pipe( + Effect.catchTags({ + PlatformError: (cause) => + Effect.fail(new CliProxyApiInstallError({ reason: "write_failed", cause })), + }), + ); + const stagedExecutable = path.join(staging, executableName); + if (platform !== "win32") { + yield* fileSystem.chmod(stagedExecutable, 0o755).pipe(failWith("write_failed")); + } + // A mislabeled asset would otherwise be redownloaded on every start. + if ((yield* probeVersion(stagedExecutable)) !== CLIPROXYAPI_VERSION) { + return yield* new CliProxyApiInstallError({ reason: "validation_failed" }); + } + yield* renameWhenUnlocked(stagedExecutable, managedPath).pipe(failWith("write_failed")); + return { + status: "available", + executablePath: managedPath, + source: "managed", + version: CLIPROXYAPI_VERSION, + } satisfies AvailableCliProxyApiBinary; + }).pipe(Effect.scoped); + + const install: CliProxyApiBinary["Service"]["install"] = installSemaphore + .withPermit(installUnlocked) + .pipe(Effect.withSpan("CliProxyApiBinary.install")); + + return CliProxyApiBinary.of({ resolve, install }); +}); + +export const layer = Layer.effect(CliProxyApiBinary, make); diff --git a/apps/server/src/providerProxy/ManagedProviderProxy.test.ts b/apps/server/src/providerProxy/ManagedProviderProxy.test.ts new file mode 100644 index 000000000000..a009128cbe73 --- /dev/null +++ b/apps/server/src/providerProxy/ManagedProviderProxy.test.ts @@ -0,0 +1,51 @@ +import { describe, expect, it } from "vite-plus/test"; +import { parse as parseYaml } from "yaml"; + +import { + isCompatibleCliProxyApiVersion, + parseCliProxyApiVersionOutput, +} from "./CliProxyApiBinary.ts"; +import { renderCliProxyApiConfig } from "./ManagedProviderProxy.ts"; + +describe("renderCliProxyApiConfig", () => { + it("writes CLIProxyAPI's v8 layout: loopback, one client key, a management key, sticky sessions", () => { + const config = parseYaml( + renderCliProxyApiConfig({ + port: 41234, + apiKey: 'client"key', + managementKey: "mgmt\\key", + authDir: "C:\\Users\\me\\.t3\\userdata\\cliproxyapi\\auths", + }), + ); + + expect(config).toEqual({ + "config-version": 8, + server: { host: "127.0.0.1", port: 41234 }, + management: { + "allow-remote": false, + "secret-key": "mgmt\\key", + "disable-control-panel": true, + }, + access: { "api-keys": ['client"key'] }, + routing: { "session-affinity": true }, + oauth: { "auth-dir": "C:\\Users\\me\\.t3\\userdata\\cliproxyapi\\auths" }, + }); + }); +}); + +describe("parseCliProxyApiVersionOutput", () => { + it("reads the version CLIProxyAPI prints before its usage text", () => { + expect( + parseCliProxyApiVersionOutput( + "CLIProxyAPI Version: 8.0.47, Commit: 90c447d3, BuiltAt: 2026-10-29T00:27:53Z\nUsage of cli-proxy-api\n", + ), + ).toBe("8.0.47"); + expect(parseCliProxyApiVersionOutput("Usage of cli-proxy-api\n -config string\n")).toBeNull(); + // A v7 override would ignore the v8 listener and management settings T3 writes. + expect(["8.0.47", "8.1.0", "7.2.81"].map(isCompatibleCliProxyApiVersion)).toEqual([ + true, + true, + false, + ]); + }); +}); diff --git a/apps/server/src/providerProxy/ManagedProviderProxy.ts b/apps/server/src/providerProxy/ManagedProviderProxy.ts new file mode 100644 index 000000000000..a1763f3e2578 --- /dev/null +++ b/apps/server/src/providerProxy/ManagedProviderProxy.ts @@ -0,0 +1,311 @@ +/** Runs CLIProxyAPI on loopback while `providerProxy.host` is `managed`, restarting it with backoff. */ +import { + providerProxyHost, + type ProviderProxyRuntimeStatus, + type ServerSettings as ServerSettingsSchema, +} from "@t3tools/contracts"; +import * as NetService from "@t3tools/shared/Net"; +import * as Clock from "effect/Clock"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as FiberHandle from "effect/FiberHandle"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Path from "effect/Path"; +import * as Semaphore from "effect/Semaphore"; +import * as Stream from "effect/Stream"; +import * as SubscriptionRef from "effect/SubscriptionRef"; +import { ChildProcess, ChildProcessSpawner } from "effect/process"; + +import * as ServerSecretStore from "../auth/ServerSecretStore.ts"; +import * as ServerConfig from "../config.ts"; +import * as ServerSettings from "../serverSettings.ts"; +import * as CliProxyApiBinary from "./CliProxyApiBinary.ts"; +import type { ManagedProxyEndpoint } from "./effectiveProxy.ts"; + +const API_KEY_SECRET = "provider-proxy-managed-api-key"; +const MANAGEMENT_KEY_SECRET = "provider-proxy-managed-management-key"; +const PORT_SECRET = "provider-proxy-managed-port"; +const LOOPBACK_HOST = "127.0.0.1"; +const READY_LINE = /\bAPI server started successfully\b/u; +// An instantly failing proxy would otherwise respawn in a tight loop. +const RESTART_STABLE_UPTIME_MS = 30_000; +const RESTART_BACKOFF_BASE_MS = 1_000; +const RESTART_BACKOFF_MAX_MS = 60_000; +// A failed download is retried no sooner than this, so an offline host does not redownload in a loop. +const INSTALL_RETRY_MS = 10 * 60_000; + +/** CLIProxyAPI's v8 config layout. T3 rewrites it before every start. */ +export function renderCliProxyApiConfig(input: { + readonly port: number; + readonly apiKey: string; + readonly managementKey: string; + readonly authDir: string; +}): string { + // JSON strings are valid YAML double-quoted scalars, including escaped Windows paths. + const quote = JSON.stringify; + return [ + "# Written by T3 Code before each start; edits are replaced.", + "config-version: 8", + "server:", + ` host: ${quote(LOOPBACK_HOST)}`, + ` port: ${input.port}`, + "management:", + " allow-remote: false", + ` secret-key: ${quote(input.managementKey)}`, + " disable-control-panel: true", + "access:", + " api-keys:", + ` - ${quote(input.apiKey)}`, + "routing:", + // A thread stays on one account, which also keeps prompt caching working. + " session-affinity: true", + "oauth:", + ` auth-dir: ${quote(input.authDir)}`, + "", + ].join("\n"); +} + +export class ManagedProviderProxy extends Context.Service< + ManagedProviderProxy, + { + readonly status: Effect.Effect; + /** Emits the current status, then each change. */ + readonly statusChanges: Stream.Stream; + /** Stable across restarts; `undefined` until the host was first set to `managed`. */ + readonly endpoint: Effect.Effect; + /** Emits the current endpoint, then each change. */ + readonly endpointChanges: Stream.Stream; + } +>()("t3/providerProxy/ManagedProviderProxy") { + /** No proxy process; `endpoint` stands in for one that already started. */ + static readonly layerTest = (endpoint?: ManagedProxyEndpoint) => + Layer.succeed(ManagedProviderProxy, { + status: Effect.succeed({ + status: endpoint === undefined ? "stopped" : "running", + }), + statusChanges: Stream.empty, + endpoint: Effect.succeed(endpoint), + endpointChanges: Stream.make(endpoint), + }); +} + +const make = Effect.gen(function* () { + const settings = yield* ServerSettings.ServerSettingsService; + const secrets = yield* ServerSecretStore.ServerSecretStore; + const binary = yield* CliProxyApiBinary.CliProxyApiBinary; + const net = yield* NetService.NetService; + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const config = yield* ServerConfig.ServerConfig; + const dataDir = path.join(config.stateDir, "cliproxyapi"); + const authDir = path.join(dataDir, "auths"); + const configPath = path.join(dataDir, "config.yaml"); + + const statusRef = yield* SubscriptionRef.make({ status: "stopped" }); + const endpointRef = yield* SubscriptionRef.make(undefined); + const setStatus = (status: ProviderProxyRuntimeStatus) => SubscriptionRef.set(statusRef, status); + + const text = (bytes: Uint8Array) => new TextDecoder().decode(bytes); + const randomKey = (name: string) => + secrets + .getOrCreateRandom(name, 32) + .pipe(Effect.map((bytes) => Buffer.from(bytes).toString("base64url"))); + const savedPort = secrets + .get(PORT_SECRET) + .pipe( + Effect.map(Option.map((bytes) => Number(text(bytes)))), + Effect.map(Option.filter((port) => Number.isInteger(port) && port > 0 && port < 65_536)), + ); + const toEndpoint = (port: number, apiKey: string, managementKey: string) => ({ + url: `http://${LOOPBACK_HOST}:${port}`, + apiKey, + managementKey, + }); + + const loadEndpoint = Effect.gen(function* () { + const port = yield* savedPort; + if (Option.isNone(port)) return undefined; + const apiKey = yield* secrets.get(API_KEY_SECRET); + const managementKey = yield* secrets.get(MANAGEMENT_KEY_SECRET); + if (Option.isNone(apiKey) || Option.isNone(managementKey)) return undefined; + return toEndpoint( + port.value, + Buffer.from(apiKey.value).toString("base64url"), + Buffer.from(managementKey.value).toString("base64url"), + ); + }); + + // Keeps the saved port while it is free, so proxied instances do not rebuild on every restart. + const prepareEndpoint = Effect.fn("ManagedProviderProxy.prepareEndpoint")(function* () { + const apiKey = yield* randomKey(API_KEY_SECRET); + const managementKey = yield* randomKey(MANAGEMENT_KEY_SECRET); + const saved = yield* savedPort; + const port = + Option.isSome(saved) && (yield* net.canListenOnHost(saved.value, LOOPBACK_HOST)) + ? saved.value + : yield* net.reserveLoopbackPort(LOOPBACK_HOST); + if (Option.getOrUndefined(saved) !== port) { + yield* secrets.set(PORT_SECRET, new TextEncoder().encode(String(port))); + } + const endpoint = toEndpoint(port, apiKey, managementKey); + const previous = yield* SubscriptionRef.get(endpointRef); + if ( + previous?.url !== endpoint.url || + previous.apiKey !== endpoint.apiKey || + previous.managementKey !== endpoint.managementKey + ) { + yield* SubscriptionRef.set(endpointRef, endpoint); + } + return { port, apiKey, managementKey }; + }); + + const writeConfig = Effect.fn("ManagedProviderProxy.writeConfig")(function* (input: { + readonly port: number; + readonly apiKey: string; + readonly managementKey: string; + }) { + yield* fileSystem.makeDirectory(authDir, { recursive: true }); + yield* fileSystem.chmod(dataDir, 0o700); + // The file holds the keys until CLIProxyAPI hashes the management key on startup. + yield* fileSystem.writeFileString(configPath, renderCliProxyApiConfig({ ...input, authDir }), { + mode: 0o600, + }); + yield* fileSystem.chmod(configPath, 0o600); + }); + + const ensureBinary = Effect.gen(function* () { + const resolved = yield* binary.resolve; + if (resolved.status === "available") return resolved; + if (resolved.status === "unsupported") { + return yield* Effect.fail( + `CLIProxyAPI has no build for ${resolved.platform}-${resolved.arch}.` as const, + ); + } + yield* setStatus({ status: "installing" }); + return yield* binary.install.pipe(Effect.mapError((error) => error.message)); + }); + + /** One proxy process, from start to exit. Fails with a message the status can show. */ + const runOnce = Effect.fn("ManagedProviderProxy.runOnce")(function* ( + executable: CliProxyApiBinary.AvailableCliProxyApiBinary, + ) { + const keys = yield* prepareEndpoint(); + yield* writeConfig(keys); + yield* setStatus({ status: "starting", version: executable.version }); + const child = yield* spawner.spawn( + ChildProcess.make(executable.executablePath, ["-config", configPath], { + cwd: dataDir, + shell: false, + detached: false, + stdin: "ignore", + stdout: "pipe", + stderr: "pipe", + }), + ); + yield* Effect.logInfo("CLIProxyAPI started", { + pid: Number(child.pid), + version: executable.version, + source: executable.source, + }); + // Output lines are per-event work for the life of the process; keep them untraced. + yield* child.all.pipe( + Stream.decodeText(), + Stream.splitLines, + Stream.runForEach( + Effect.fnUntraced(function* (line) { + if (READY_LINE.test(line)) { + yield* setStatus({ status: "running", version: executable.version }); + } + const output = line + .replaceAll(keys.apiKey, "") + .replaceAll(keys.managementKey, ""); + if (/\[(?:warn|warning|error|fatal|panic)\]/iu.test(line)) { + yield* Effect.logWarning("CLIProxyAPI output", { output }); + } else { + yield* Effect.logDebug("CLIProxyAPI output", { output }); + } + }), + ), + Effect.ignore, + Effect.forkScoped, + ); + const exitCode = Number(yield* child.exitCode); + return yield* Effect.fail(`CLIProxyAPI exited with code ${exitCode}.`); + }, Effect.scoped); + + const supervise = Effect.gen(function* () { + let restartDelay = 0; + while (true) { + const executable = yield* Effect.result(ensureBinary); + if (executable._tag === "Failure") { + yield* setStatus({ status: "error", message: executable.failure }); + yield* Effect.logWarning("Could not prepare CLIProxyAPI", { reason: executable.failure }); + yield* Effect.sleep(INSTALL_RETRY_MS); + continue; + } + const startedAt = yield* Clock.currentTimeMillis; + const result = yield* runOnce(executable.success).pipe( + Effect.catch((error) => + typeof error === "string" + ? Effect.succeed(error) + : Effect.logWarning("Could not start CLIProxyAPI", { cause: error }).pipe( + Effect.as("Could not start CLIProxyAPI."), + ), + ), + ); + const uptime = (yield* Clock.currentTimeMillis) - startedAt; + restartDelay = + uptime >= RESTART_STABLE_UPTIME_MS + ? RESTART_BACKOFF_BASE_MS + : Math.min(Math.max(restartDelay * 2, RESTART_BACKOFF_BASE_MS), RESTART_BACKOFF_MAX_MS); + yield* setStatus({ status: "error", message: result }); + yield* Effect.logWarning("CLIProxyAPI stopped; restarting", { reason: result, restartDelay }); + yield* Effect.sleep(restartDelay); + } + }); + + const supervisor = yield* FiberHandle.make(); + const applySemaphore = yield* Semaphore.make(1); + let appliedManaged = false; + const apply = Effect.fn("ManagedProviderProxy.apply")(function* (managed: boolean) { + if (managed === appliedManaged) return; + appliedManaged = managed; + yield* FiberHandle.clear(supervisor); + if (!managed) { + yield* setStatus({ status: "stopped" }); + return; + } + yield* FiberHandle.run(supervisor, supervise); + }, applySemaphore.withPermits(1)); + + const initialEndpoint = yield* loadEndpoint.pipe(Effect.orElseSucceed(() => undefined)); + if (initialEndpoint !== undefined) yield* SubscriptionRef.set(endpointRef, initialEndpoint); + + const isManaged = (current: ServerSettingsSchema) => + providerProxyHost(current.providerProxy) === "managed"; + // Subscribe before the seed read so a save in between is not lost. + const changes = yield* settings.subscribeChanges; + const initialManaged = yield* settings.getSettings.pipe( + Effect.map(isManaged), + Effect.orElseSucceed(() => false), + ); + yield* apply(initialManaged); + yield* changes.pipe( + Stream.map(isManaged), + Stream.changes, + Stream.runForEach(apply), + Effect.forkScoped, + ); + + return ManagedProviderProxy.of({ + status: SubscriptionRef.get(statusRef), + statusChanges: SubscriptionRef.changes(statusRef), + endpoint: SubscriptionRef.get(endpointRef), + endpointChanges: SubscriptionRef.changes(endpointRef), + }); +}); + +export const layer = Layer.effect(ManagedProviderProxy, make); diff --git a/apps/server/src/providerProxy/ProxyAccounts.test.ts b/apps/server/src/providerProxy/ProxyAccounts.test.ts new file mode 100644 index 000000000000..98df5a6ddd4a --- /dev/null +++ b/apps/server/src/providerProxy/ProxyAccounts.test.ts @@ -0,0 +1,258 @@ +import { it as effectIt } from "@effect/vitest"; +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 TestClock from "effect/testing/TestClock"; +import { HttpClient, type HttpClientRequest, HttpClientResponse, UrlParams } from "effect/http"; +import { describe, expect, it } from "vite-plus/test"; + +import * as ServerSettings from "../serverSettings.ts"; +import * as ManagedProviderProxy from "./ManagedProviderProxy.ts"; +import * as ProxyAccounts from "./ProxyAccounts.ts"; + +const now = DateTime.makeUnsafe("2026-10-30T00:00:00Z"); + +describe("toProviderProxyAccount", () => { + it.each([ + [ + "an active account", + { name: "claude-a.json", provider: "claude", email: "a@example.test", status: "active" }, + { name: "claude-a.json", provider: "claude", email: "a@example.test", state: "active" }, + ], + [ + "a disabled account", + { name: "codex-b.json", type: "codex", disabled: true, status: "disabled" }, + { name: "codex-b.json", provider: "codex", state: "disabled" }, + ], + [ + "a cooling-down account", + { + name: "codex-c.json", + provider: "codex", + account: "c@example.test", + status: "error", + status_message: "quota exhausted", + unavailable: true, + next_retry_after: "2026-10-30T02:00:00Z", + }, + { + name: "codex-c.json", + provider: "codex", + email: "c@example.test", + state: "cooldown", + statusMessage: "quota exhausted", + retryAt: "2026-10-30T02:00:00.000Z", + }, + ], + [ + "a failing account whose retry time passed", + { + name: "claude-d.json", + provider: "claude", + status: "error", + status_message: "token expired", + unavailable: true, + next_retry_after: "2026-10-29T23:00:00Z", + }, + { name: "claude-d.json", provider: "claude", state: "error", statusMessage: "token expired" }, + ], + ])("maps %s", (_label, file, account) => { + expect(ProxyAccounts.toProviderProxyAccount(file, now)).toEqual(account); + }); + + it("skips config-defined API keys, which have no credential file to manage", () => { + expect( + ProxyAccounts.toProviderProxyAccount( + { name: "gemini-api-key", provider: "gemini", runtime_only: true }, + now, + ), + ).toBeUndefined(); + }); +}); + +const fullUrl = (request: HttpClientRequest.HttpClientRequest) => { + const query = UrlParams.toString(request.urlParams); + return query === "" ? request.url : `${request.url}?${query}`; +}; +const bodyText = (request: HttpClientRequest.HttpClientRequest) => + request.body._tag === "Uint8Array" ? new TextDecoder().decode(request.body.body) : ""; + +interface Recorded { + readonly method: string; + readonly url: string; + readonly authorization: string | undefined; + readonly body: string; +} + +/** A fake management API answering by `METHOD path`; unlisted routes answer 404. */ +const managementApi = ( + routes: Record, + recorded: Array, +) => + HttpClient.make((request) => { + const url = fullUrl(request); + recorded.push({ + method: request.method, + url, + authorization: request.headers.authorization, + body: bodyText(request), + }); + const path = url.replace("http://proxy.test:8317/v0/management/", ""); + const route = routes[`${request.method} ${path}`]; + return Effect.succeed( + HttpClientResponse.fromWeb( + request, + route === undefined + ? Response.json({ error: "not found" }, { status: 404 }) + : Response.json(route.body, { status: route.status ?? 200 }), + ), + ); + }); + +const withProxy = (client: HttpClient.HttpClient) => + Effect.provide( + ProxyAccounts.layer.pipe( + Layer.provide( + Layer.mergeAll( + ServerSettings.layerTest({ + providerProxy: { + host: "external", + url: "http://proxy.test:8317/v1", + apiKey: "", + managementKey: "mgmt-key", + }, + }), + ManagedProviderProxy.ManagedProviderProxy.layerTest(), + Layer.succeed(HttpClient.HttpClient, client), + ), + ), + ), + ); + +describe("ProxyAccounts", () => { + effectIt.effect("times out a proxy that sends headers and then stalls the body", () => + Effect.gen(function* () { + const accounts = yield* ProxyAccounts.ProxyAccounts; + const removal = yield* accounts + .removeAccount("claude-a.json") + .pipe(Effect.flip, Effect.forkChild); + yield* TestClock.adjust("15 seconds"); + expect((yield* Fiber.join(removal)).detail).toBe( + "The proxy did not answer within 15 seconds.", + ); + }).pipe( + withProxy( + HttpClient.make((request) => + Effect.succeed( + HttpClientResponse.fromWeb(request, new Response(new ReadableStream({ start() {} }))), + ), + ), + ), + ), + ); + + effectIt.effect("lists, disables, and removes accounts with the management key", () => { + const recorded: Array = []; + return Effect.gen(function* () { + const accounts = yield* ProxyAccounts.ProxyAccounts; + const state = yield* accounts.getState; + expect(state.accounts).toEqual([ + { name: "claude-a.json", provider: "claude", email: "a@example.test", state: "active" }, + ]); + yield* accounts.setAccountDisabled({ name: "claude-a.json", disabled: true }); + yield* accounts.removeAccount("claude a.json"); + const missing = yield* accounts.removeAccount("gone.json").pipe(Effect.flip); + expect(missing.detail).toBe("The proxy answered with HTTP 404: auth file not found."); + + expect(recorded.map(({ method, url }) => `${method} ${url}`)).toEqual([ + "GET http://proxy.test:8317/v0/management/auth-files", + "PATCH http://proxy.test:8317/v0/management/auth-files/status", + "DELETE http://proxy.test:8317/v0/management/auth-files?name=claude%20a.json", + "DELETE http://proxy.test:8317/v0/management/auth-files?name=gone.json", + ]); + expect(recorded.every((request) => request.authorization === "Bearer mgmt-key")).toBe(true); + expect(JSON.parse(recorded[1]!.body)).toEqual({ name: "claude-a.json", disabled: true }); + }).pipe( + withProxy( + managementApi( + { + "GET auth-files": { + body: { + files: [ + { name: "claude-a.json", provider: "claude", email: "a@example.test" }, + { name: "gemini-key", provider: "gemini", runtime_only: true }, + ], + }, + }, + "PATCH auth-files/status": { body: { status: "ok" } }, + "DELETE auth-files?name=claude%20a.json": { body: { status: "ok" } }, + "DELETE auth-files?name=gone.json": { + status: 404, + body: { error: "auth file not found" }, + }, + }, + recorded, + ), + ), + ); + }); + + effectIt.effect( + "signs in with paste-back when the proxy cannot catch the redirect itself", + () => { + const recorded: Array = []; + return Effect.gen(function* () { + const accounts = yield* ProxyAccounts.ProxyAccounts; + const login = yield* accounts.startLogin("claude"); + expect(login).toEqual({ url: "https://claude.ai/oauth/authorize?x=1", state: "abc123" }); + // Clients open this URL, so a proxy can't hand them a non-https link. + const unsafe = yield* accounts.startLogin("codex").pipe(Effect.flip); + expect(unsafe._tag).toBe("ProviderProxyError"); + expect(yield* accounts.getLoginStatus("abc123")).toEqual({ status: "wait" }); + + const wrongLogin = yield* accounts + .completeLogin({ + state: "abc123", + redirectUrl: "http://localhost:54545/callback?code=c&state=other", + }) + .pipe(Effect.flip); + expect(wrongLogin._tag).toBe("ProviderProxyError"); + yield* accounts.completeLogin({ + state: "abc123", + redirectUrl: "http://localhost:54545/callback?code=c&state=abc123", + }); + + expect(recorded.map(({ method, url }) => `${method} ${url}`)).toEqual([ + "GET http://proxy.test:8317/v0/management/anthropic-auth-url", + "GET http://proxy.test:8317/v0/management/codex-auth-url", + "GET http://proxy.test:8317/v0/management/get-auth-status?state=abc123", + "POST http://proxy.test:8317/v0/management/oauth-callback", + ]); + expect(JSON.parse(recorded[3]!.body)).toEqual({ + redirect_url: "http://localhost:54545/callback?code=c&state=abc123", + }); + }).pipe( + withProxy( + managementApi( + { + "GET anthropic-auth-url": { + body: { + status: "ok", + url: "https://claude.ai/oauth/authorize?x=1", + state: "abc123", + }, + }, + "GET codex-auth-url": { + body: { status: "ok", url: "file:///etc/passwd", state: "def456" }, + }, + "GET get-auth-status?state=abc123": { body: { status: "wait" } }, + "POST oauth-callback": { body: { status: "ok" } }, + }, + recorded, + ), + ), + ); + }, + ); +}); diff --git a/apps/server/src/providerProxy/ProxyAccounts.ts b/apps/server/src/providerProxy/ProxyAccounts.ts new file mode 100644 index 000000000000..48f292d8ab61 --- /dev/null +++ b/apps/server/src/providerProxy/ProxyAccounts.ts @@ -0,0 +1,323 @@ +/** The proxy's pooled accounts, through CLIProxyAPI's `/v0/management` API. */ +import { + providerProxyHost, + ProviderProxyError, + type ProviderProxyAccount, + type ProviderProxyLogin, + type ProviderProxyLoginProvider, + type ProviderProxyLoginStatus, + type ProviderProxyRuntimeStatus, + type ProviderProxyState, +} from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; +import * as Context from "effect/Context"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Schema from "effect/Schema"; +import { HttpClient, HttpClientRequest } from "effect/http"; + +import * as ServerSettings from "../serverSettings.ts"; +import { resolveProxyManagement, type ProxyManagementEndpoint } from "./effectiveProxy.ts"; +import * as ManagedProviderProxy from "./ManagedProviderProxy.ts"; + +const OptionalText = Schema.optional(Schema.NullOr(Schema.String)); +const AuthFile = Schema.Struct({ + name: Schema.String, + provider: OptionalText, + type: OptionalText, + email: OptionalText, + account: OptionalText, + label: OptionalText, + status: OptionalText, + status_message: OptionalText, + disabled: Schema.optional(Schema.NullOr(Schema.Boolean)), + unavailable: Schema.optional(Schema.NullOr(Schema.Boolean)), + runtime_only: Schema.optional(Schema.NullOr(Schema.Boolean)), + next_retry_after: OptionalText, +}); +export type CliProxyAuthFile = typeof AuthFile.Type; +const decodeAuthFiles = Schema.decodeUnknownEffect( + Schema.Struct({ files: Schema.Array(AuthFile) }), +); +const decodeLogin = Schema.decodeUnknownEffect( + Schema.Struct({ url: Schema.String, state: Schema.String }), +); +const decodeLoginStatus = Schema.decodeUnknownEffect( + Schema.Struct({ + status: Schema.Literals(["ok", "wait", "error"]), + error: OptionalText, + }), +); +const decodeErrorBody = Schema.decodeUnknownOption(Schema.Struct({ error: Schema.String })); + +const REQUEST_TIMEOUT = "15 seconds"; +const MAX_PROXY_MESSAGE_LENGTH = 200; +// CLIProxyAPI names the Claude login after its upstream. +const LOGIN_PATHS: Record = { + claude: "anthropic-auth-url", + codex: "codex-auth-url", +}; + +const nonEmpty = (value: string | null | undefined) => { + const trimmed = value?.trim(); + return trimmed === undefined || trimmed === "" ? undefined : trimmed; +}; +const bounded = (value: string) => + value.length > MAX_PROXY_MESSAGE_LENGTH ? `${value.slice(0, MAX_PROXY_MESSAGE_LENGTH)}…` : value; + +/** Maps one `auth-files` entry; config-defined API keys have no file to manage and are skipped. */ +export function toProviderProxyAccount( + file: CliProxyAuthFile, + now: DateTime.Utc, +): ProviderProxyAccount | undefined { + const name = nonEmpty(file.name); + const provider = nonEmpty(file.provider) ?? nonEmpty(file.type); + if (name === undefined || provider === undefined || file.runtime_only === true) return undefined; + const email = nonEmpty(file.email) ?? nonEmpty(file.account) ?? nonEmpty(file.label); + const retryAt = Option.flatMap(Option.fromNullishOr(nonEmpty(file.next_retry_after)), (value) => + Option.filter(DateTime.make(value), (at) => DateTime.isGreaterThan(at, now)), + ); + const state: ProviderProxyAccount["state"] = + file.disabled === true || file.status === "disabled" + ? "disabled" + : Option.isSome(retryAt) + ? "cooldown" + : file.status === "error" || file.unavailable === true + ? "error" + : "active"; + const statusMessage = state === "active" ? undefined : nonEmpty(file.status_message); + return { + name, + provider, + ...(email === undefined ? {} : { email }), + state, + ...(statusMessage === undefined ? {} : { statusMessage: bounded(statusMessage) }), + ...(state === "cooldown" && Option.isSome(retryAt) + ? { retryAt: DateTime.formatIso(retryAt.value) } + : {}), + }; +} + +/** The pasted URL must belong to the sign-in it completes. */ +function redirectUrlMatchesLogin(redirectUrl: string, state: string): boolean { + try { + return new URL(redirectUrl).searchParams.get("state") === state; + } catch { + return false; + } +} + +export class ProxyAccounts extends Context.Service< + ProxyAccounts, + { + readonly getState: Effect.Effect; + readonly startLogin: ( + provider: ProviderProxyLoginProvider, + ) => Effect.Effect; + readonly getLoginStatus: ( + state: string, + ) => Effect.Effect; + readonly completeLogin: (input: { + readonly state: string; + readonly redirectUrl: string; + }) => Effect.Effect; + readonly setAccountDisabled: (input: { + readonly name: string; + readonly disabled: boolean; + }) => Effect.Effect; + readonly removeAccount: (name: string) => Effect.Effect; + } +>()("t3/providerProxy/ProxyAccounts") {} + +const make = Effect.gen(function* () { + const client = yield* HttpClient.HttpClient; + const settingsService = yield* ServerSettings.ServerSettingsService; + const managed = yield* ManagedProviderProxy.ManagedProviderProxy; + + const context = Effect.gen(function* () { + const settings = yield* settingsService.getSettings.pipe( + Effect.mapError(() => new ProviderProxyError({ detail: "Could not read proxy settings." })), + ); + const host = providerProxyHost(settings.providerProxy); + const runtime: ProviderProxyRuntimeStatus = + host === "managed" ? yield* managed.status : { status: "stopped" }; + const management = resolveProxyManagement(settings, yield* managed.endpoint); + return { host, runtime, management }; + }); + + const requireManagement = context.pipe( + Effect.flatMap(({ management }) => + management === undefined + ? Effect.fail( + new ProviderProxyError({ + detail: + "This environment cannot manage the proxy's accounts. Run the proxy here, or add the external proxy's management key.", + }), + ) + : Effect.succeed(management), + ), + ); + + const request = Effect.fn("ProxyAccounts.request")(function* ( + management: ProxyManagementEndpoint, + build: (url: string) => HttpClientRequest.HttpClientRequest, + path: string, + ) { + // The deadline covers the body too: a proxy can send headers and then stall. + const { response, body } = yield* client + .execute( + build(`${management.url}/v0/management/${path}`).pipe( + HttpClientRequest.bearerToken(management.managementKey), + ), + ) + .pipe( + Effect.flatMap((response) => + response.json.pipe( + Effect.orElseSucceed((): unknown => null), + Effect.map((body) => ({ response, body })), + ), + ), + Effect.timeout(REQUEST_TIMEOUT), + Effect.mapError( + (error) => + new ProviderProxyError({ + detail: Cause.isTimeoutError(error) + ? "The proxy did not answer within 15 seconds." + : "Could not reach the proxy.", + }), + ), + ); + if (response.status >= 200 && response.status < 300) return body; + // CLIProxyAPI's own error strings, such as "auth file not found", carry no credentials. + const proxyMessage = Option.match(decodeErrorBody(body), { + onNone: () => "", + onSome: ({ error }) => `: ${bounded(error)}`, + }); + return yield* new ProviderProxyError({ + detail: `The proxy answered with HTTP ${response.status}${proxyMessage}.`, + }); + }); + + const malformed = () => + new ProviderProxyError({ detail: "The proxy sent a response T3 Code does not understand." }); + + const getState: ProxyAccounts["Service"]["getState"] = Effect.gen(function* () { + const { host, runtime, management } = yield* context; + // A managed proxy that is not up yet has no accounts to read. + if (management === undefined || (host === "managed" && runtime.status !== "running")) { + return { runtime, accounts: null } satisfies ProviderProxyState; + } + return yield* request(management, HttpClientRequest.get, "auth-files").pipe( + Effect.flatMap((body) => decodeAuthFiles(body).pipe(Effect.mapError(malformed))), + Effect.flatMap(({ files }) => + DateTime.now.pipe( + Effect.map((now) => files.flatMap((file) => toProviderProxyAccount(file, now) ?? [])), + ), + ), + Effect.map((accounts): ProviderProxyState => ({ runtime, accounts })), + Effect.catchTags({ + ProviderProxyError: (error) => + Effect.succeed({ + runtime, + accounts: [], + accountsError: error.detail, + } satisfies ProviderProxyState), + }), + ); + }).pipe( + Effect.catchTags({ + ProviderProxyError: () => + Effect.succeed({ + runtime: { status: "stopped" }, + accounts: null, + } satisfies ProviderProxyState), + }), + Effect.withSpan("ProxyAccounts.getState"), + ); + + const startLogin = Effect.fn("ProxyAccounts.startLogin")(function* ( + provider: ProviderProxyLoginProvider, + ) { + const management = yield* requireManagement; + const path = LOGIN_PATHS[provider]; + // Not `is_webui`: its callback forwarder binds 0.0.0.0, so sign-in completes by paste-back. + const body = yield* request(management, HttpClientRequest.get, path); + const login = yield* decodeLogin(body).pipe(Effect.mapError(malformed)); + // Clients open this URL; an external proxy must not hand them a file: or app link. + if (URL.parse(login.url)?.protocol !== "https:") return yield* malformed(); + return login; + }); + + const getLoginStatus = Effect.fn("ProxyAccounts.getLoginStatus")(function* (state: string) { + const management = yield* requireManagement; + const body = yield* request( + management, + HttpClientRequest.get, + `get-auth-status?state=${encodeURIComponent(state)}`, + ); + const status = yield* decodeLoginStatus(body).pipe(Effect.mapError(malformed)); + const message = nonEmpty(status.error); + return { + status: status.status, + ...(message === undefined ? {} : { message: bounded(message) }), + } satisfies ProviderProxyLoginStatus; + }); + + const completeLogin = Effect.fn("ProxyAccounts.completeLogin")(function* (input: { + readonly state: string; + readonly redirectUrl: string; + }) { + if (!redirectUrlMatchesLogin(input.redirectUrl, input.state)) { + return yield* new ProviderProxyError({ + detail: + "Paste the full address from this sign-in's browser tab, including ?code=…&state=….", + }); + } + const management = yield* requireManagement; + yield* request( + management, + (url) => + HttpClientRequest.post(url).pipe( + HttpClientRequest.bodyJsonUnsafe({ redirect_url: input.redirectUrl }), + ), + "oauth-callback", + ); + }); + + const setAccountDisabled = Effect.fn("ProxyAccounts.setAccountDisabled")(function* (input: { + readonly name: string; + readonly disabled: boolean; + }) { + const management = yield* requireManagement; + yield* request( + management, + (url) => + HttpClientRequest.patch(url).pipe( + HttpClientRequest.bodyJsonUnsafe({ name: input.name, disabled: input.disabled }), + ), + "auth-files/status", + ); + }); + + const removeAccount = Effect.fn("ProxyAccounts.removeAccount")(function* (name: string) { + const management = yield* requireManagement; + yield* request( + management, + HttpClientRequest.delete, + `auth-files?name=${encodeURIComponent(name)}`, + ); + }); + + return ProxyAccounts.of({ + getState, + startLogin, + getLoginStatus, + completeLogin, + setAccountDisabled, + removeAccount, + }); +}); + +export const layer = Layer.effect(ProxyAccounts, make); diff --git a/apps/server/src/providerProxy/effectiveProxy.ts b/apps/server/src/providerProxy/effectiveProxy.ts new file mode 100644 index 000000000000..8ff4c73f2187 --- /dev/null +++ b/apps/server/src/providerProxy/effectiveProxy.ts @@ -0,0 +1,54 @@ +import { providerProxyHost, type ServerSettings } from "@t3tools/contracts"; +import { + normalizeProviderProxyUrl, + type ProviderProxyRoute, +} from "@t3tools/provider-core/server/providerProxy"; + +/** The loopback CLIProxyAPI this environment runs; its keys live in the secret store. */ +export interface ManagedProxyEndpoint { + readonly url: string; + readonly apiKey: string; + readonly managementKey: string; +} + +export interface ProxyManagementEndpoint { + readonly url: string; + readonly managementKey: string; +} + +/** Where `useProxy` instances send model traffic, or `undefined` when there is no proxy. */ +export function resolveProviderProxyRoute( + settings: ServerSettings, + managed: ManagedProxyEndpoint | undefined, +): ProviderProxyRoute | undefined { + switch (providerProxyHost(settings.providerProxy)) { + case "off": + return undefined; + case "managed": + return managed === undefined ? undefined : { url: managed.url, apiKey: managed.apiKey }; + case "external": { + const url = normalizeProviderProxyUrl(settings.providerProxy.url); + return url === "" ? undefined : { url, apiKey: settings.providerProxy.apiKey }; + } + } +} + +/** The proxy whose management API this environment may call: accounts and usage limits. */ +export function resolveProxyManagement( + settings: ServerSettings, + managed: ManagedProxyEndpoint | undefined, +): ProxyManagementEndpoint | undefined { + switch (providerProxyHost(settings.providerProxy)) { + case "off": + return undefined; + case "managed": + return managed === undefined + ? undefined + : { url: managed.url, managementKey: managed.managementKey }; + case "external": { + const url = normalizeProviderProxyUrl(settings.providerProxy.url); + const { managementKey } = settings.providerProxy; + return url === "" || managementKey === "" ? undefined : { url, managementKey }; + } + } +} diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index d0d51a0b241b..49ffb4cb4ee2 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -105,6 +105,9 @@ import * as ProviderAdapterRegistry from "./orchestration-v2/ProviderAdapterRegi import * as ProviderRegistry from "./provider/ProviderRegistry.ts"; import * as ProviderUsageLimitsIngestion from "./provider/ProviderUsageLimitsIngestion.ts"; import * as UsageLimitSources from "./usage/UsageLimitSources.ts"; +import * as CliProxyApiBinary from "./providerProxy/CliProxyApiBinary.ts"; +import * as ManagedProviderProxy from "./providerProxy/ManagedProviderProxy.ts"; +import * as ProxyAccounts from "./providerProxy/ProxyAccounts.ts"; import * as ProjectFaviconResolver from "./project/ProjectFaviconResolver.ts"; import * as T3ProjectFileLoader from "./project/T3ProjectFileLoader.ts"; import * as RepositoryIdentityResolver from "./project/RepositoryIdentityResolver.ts"; @@ -525,6 +528,13 @@ const layerAuth = EnvironmentAuth.layer.pipe( Layer.provide(ServerSecretStore.layer), ); +const layerProviderProxy = ProxyAccounts.layer.pipe( + Layer.provideMerge(ManagedProviderProxy.layer), + Layer.provide(CliProxyApiBinary.layer), + Layer.provide(ServerSecretStore.layer), + Layer.provide(NetService.layer), +); + const layerCloudManagedEndpointRuntime = Layer.mergeAll( layerRelayClient, CloudManagedEndpointRuntime.layer.pipe( @@ -679,6 +689,7 @@ const layerRuntimeCoreDependenciesBase = Layer.mergeAll( ), Layer.provideMerge(ProviderRegistry.layer), Layer.provideMerge(ProviderInstanceRegistryHydration.layer), + Layer.provideMerge(layerProviderProxy), // 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/serverSettings.test.ts b/apps/server/src/serverSettings.test.ts index 281bab5e4227..5c196617d326 100644 --- a/apps/server/src/serverSettings.test.ts +++ b/apps/server/src/serverSettings.test.ts @@ -1277,7 +1277,7 @@ it.layer(NodeServices.layer)("server settings", (it) => { ); it.effect( - "uses the same secret lifecycle for provider config, environment, host tokens, and quota keys", + "uses the same secret lifecycle for provider config, environment, host tokens, quota and proxy keys", () => Effect.gen(function* () { const service = yield* ServerSettingsModule.ServerSettingsService; @@ -1309,6 +1309,11 @@ it.layer(NodeServices.layer)("server settings", (it) => { enabled: true, }, }, + providerProxy: { + url: "http://proxy.example.test:8317", + apiKey: "proxy-client-secret", + managementKey: "proxy-management-secret", + }, easDeviceHosts: [ { kind: "eas", @@ -1328,6 +1333,8 @@ it.layer(NodeServices.layer)("server settings", (it) => { "bitbucket-secret", "gitcafe-secret", "quota-secret", + "proxy-client-secret", + "proxy-management-secret", "expo-secret", ]; const client = ServerSettingsModule.redactServerSettingsForClient(saved); @@ -1343,6 +1350,7 @@ it.layer(NodeServices.layer)("server settings", (it) => { providerInstances: client.providerInstances, sourceControlHosts: client.sourceControlHosts, usageLimitSources: client.usageLimitSources, + providerProxy: client.providerProxy, easDeviceHosts: client.easDeviceHosts, }); for (const value of values) assert.include(JSON.stringify(echoed), value); @@ -1393,6 +1401,8 @@ it.layer(NodeServices.layer)("server settings", (it) => { }), sourceControlHostSecretName({ kind: "gitcafe", field: "token", serverHost: null }), "usage-limit-source-aHVi", + "provider-proxy-YXBpS2V5", + "provider-proxy-bWFuYWdlbWVudEtleQ", ]; for (const name of secretNames) { assert.isTrue(Option.isSome(yield* store.get(name)), name); @@ -1426,6 +1436,7 @@ it.layer(NodeServices.layer)("server settings", (it) => { gitcafe: { token: "" }, }, usageLimitSources: { [UsageLimitSourceId.make("hub")]: null }, + providerProxy: { apiKey: "", managementKey: "" }, }); for (const name of secretNames) { assert.isTrue(Option.isNone(yield* store.get(name))); diff --git a/apps/server/src/settingsSecrets.ts b/apps/server/src/settingsSecrets.ts index 3d4daa659e58..947d3368f516 100644 --- a/apps/server/src/settingsSecrets.ts +++ b/apps/server/src/settingsSecrets.ts @@ -3,6 +3,7 @@ import { ClaudeSettings, CodexSettings, EasDeviceHostConfig, + ProviderProxySettings, type ServerSettings, SimlockDeviceHostConfig, UsageLimitSourceConfig, @@ -34,6 +35,7 @@ const providerSecretFields = new Map([ ), ]); const usageLimitSecretFields = secretSettingsFields(UsageLimitSourceConfig); +const providerProxySecretFields = secretSettingsFields(ProviderProxySettings); const easDeviceHostSecretFields = secretSettingsFields(EasDeviceHostConfig); const simlockDeviceHostSecretFields = secretSettingsFields(SimlockDeviceHostConfig); const encode = (value: string) => Buffer.from(value, "utf8").toString("base64url"); @@ -150,6 +152,13 @@ export function mapSettingsSecrets( ]), ) as ServerSettings["usageLimitSources"]; + const providerProxy = mapFields( + settings.providerProxy, + providerProxySecretFields, + (field, serverHost) => + `provider-proxy-${encode(field)}` + (serverHost === null ? "" : `-${encode(serverHost)}`), + ) as ServerSettings["providerProxy"]; + const easDeviceHosts = settings.easDeviceHosts.map( (host) => mapFields( @@ -178,6 +187,7 @@ export function mapSettingsSecrets( providerInstances, sourceControlHosts, usageLimitSources, + providerProxy, easDeviceHosts, simlockDeviceHosts, }; diff --git a/apps/server/src/textGeneration/ClaudeTextGeneration.test.ts b/apps/server/src/textGeneration/ClaudeTextGeneration.test.ts index 3ea0502f83b3..f8dd41aeaa2c 100644 --- a/apps/server/src/textGeneration/ClaudeTextGeneration.test.ts +++ b/apps/server/src/textGeneration/ClaudeTextGeneration.test.ts @@ -66,14 +66,22 @@ function withFakeClaudeEnv( configDirMustBe?: string; cwdMustNotBe?: string; claudeConfig?: Partial; + proxy?: { readonly url: string; readonly apiKey: string }; }, - effectFn: (textGeneration: TextGeneration.TextGeneration["Service"]) => Effect.Effect, + effectFn: ( + textGeneration: TextGeneration.TextGeneration["Service"], + tempDir: string, + ) => Effect.Effect, ) { return Effect.gen(function* () { const fs = yield* FileSystem.FileSystem; const tempDir = yield* fs.makeTempDirectoryScoped({ prefix: "t3code-claude-text-" }); const binDir = yield* makeFakeClaudeBinary(tempDir, input.lifecyclePort); const pathDelimiter = (yield* HostProcess.isWindows) ? ";" : ":"; + const path = yield* Path.Path; + const platform = yield* HostProcess.Platform; + const previousSettingsExpected = process.env.T3_FAKE_CLAUDE_SETTINGS_EXPECTED; + const previousSettingsReceipt = process.env.T3_FAKE_CLAUDE_SETTINGS_RECEIPT; const previousPath = process.env.PATH; const previousOutput = process.env.T3_FAKE_CLAUDE_OUTPUT; const previousExitCode = process.env.T3_FAKE_CLAUDE_EXIT_CODE; @@ -86,6 +94,16 @@ function withFakeClaudeEnv( yield* Effect.acquireRelease( Effect.sync(() => { + if (input.proxy === undefined) { + delete process.env.T3_FAKE_CLAUDE_SETTINGS_EXPECTED; + delete process.env.T3_FAKE_CLAUDE_SETTINGS_RECEIPT; + } else { + process.env.T3_FAKE_CLAUDE_SETTINGS_EXPECTED = JSON.stringify({ + ...input.proxy, + platform, + }); + process.env.T3_FAKE_CLAUDE_SETTINGS_RECEIPT = path.join(tempDir, "settings-path.txt"); + } process.env.PATH = `${binDir}${pathDelimiter}${previousPath ?? ""}`; process.env.T3_FAKE_CLAUDE_OUTPUT = input.output; @@ -133,6 +151,12 @@ function withFakeClaudeEnv( }), () => Effect.sync(() => { + if (previousSettingsExpected === undefined) + delete process.env.T3_FAKE_CLAUDE_SETTINGS_EXPECTED; + else process.env.T3_FAKE_CLAUDE_SETTINGS_EXPECTED = previousSettingsExpected; + if (previousSettingsReceipt === undefined) + delete process.env.T3_FAKE_CLAUDE_SETTINGS_RECEIPT; + else process.env.T3_FAKE_CLAUDE_SETTINGS_RECEIPT = previousSettingsReceipt; process.env.PATH = previousPath; if (previousOutput === undefined) { @@ -190,15 +214,16 @@ function withFakeClaudeEnv( config, undefined, Effect.succeed(SYNTHETIC_CLAUDE_MODEL_CATALOG), + input.proxy, ); - return yield* effectFn(textGeneration); + return yield* effectFn(textGeneration, tempDir); }).pipe(Effect.scoped); } it.layer(layerClaudeTextGenerationTest)("ClaudeTextGeneration", (it) => { - it.effect( - "keeps canonical built-in capabilities when a custom model collides with its alias", - () => + it.effect.each([false, true])( + "keeps canonical built-in capabilities when a custom model collides with its alias (proxy: %s)", + (proxied) => withFakeClaudeEnv( { output: JSON.stringify({ @@ -207,10 +232,16 @@ it.layer(layerClaudeTextGenerationTest)("ClaudeTextGeneration", (it) => { body: "Body", }, }), - argsMustContain: `--model ${SYNTHETIC_CLAUDE_CAPABLE_MODEL}[expanded] --effort max --settings {"disableAllHooks":true,"fastMode":true}`, + ...(proxied ? { argsMustNotContain: "proxy-test-key" } : {}), + argsMustContain: proxied + ? "--effort max --settings" + : `--model ${SYNTHETIC_CLAUDE_CAPABLE_MODEL}[expanded] --effort max --settings {"disableAllHooks":true,"fastMode":true}`, + ...(proxied + ? { proxy: { url: "http://proxy.test:8317", apiKey: "proxy-test-key" } } + : {}), claudeConfig: { customModels: [SYNTHETIC_CLAUDE_COLLIDING_ALIAS] }, }, - (textGeneration) => + (textGeneration, tempDir) => Effect.gen(function* () { const generated = yield* textGeneration.generatePrContent({ cwd: process.cwd(), @@ -232,6 +263,14 @@ it.layer(layerClaudeTextGenerationTest)("ClaudeTextGeneration", (it) => { }); expect(generated.title).toBe("Improve orchestration flow"); + if (proxied) { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const settingsPath = yield* fs.readFileString( + path.join(tempDir, "settings-path.txt"), + ); + expect(yield* fs.exists(settingsPath)).toBe(false); + } }), ), ); diff --git a/apps/server/src/textGeneration/ClaudeTextGeneration.ts b/apps/server/src/textGeneration/ClaudeTextGeneration.ts index 8fa9ccb1c809..3db69dcd1f87 100644 --- a/apps/server/src/textGeneration/ClaudeTextGeneration.ts +++ b/apps/server/src/textGeneration/ClaudeTextGeneration.ts @@ -1,12 +1,14 @@ 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 Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import { ChildProcess, ChildProcessSpawner } from "effect/process"; import { type ClaudeSettings, TextGenerationError } from "@t3tools/contracts"; import { resolveSpawnCommand } from "@t3tools/shared/shell"; +import type { ProviderProxyRoute } from "@t3tools/provider-core/server/providerProxy"; import * as TextGenerationOperations from "@t3tools/provider-core/server/textGenerationOperations"; import { @@ -31,6 +33,7 @@ import { } from "../provider/ClaudeModelCatalog.ts"; import { hasClaudeManagedMcpConfig, + claudeProxySettingsEnvironment, makeClaudeEnvironment, } from "../provider/Drivers/ClaudeHome.ts"; @@ -81,9 +84,11 @@ export const makeClaudeTextGeneration = Effect.fn("makeClaudeTextGeneration")(fu claudeSettings: ClaudeSettings, environment?: NodeJS.ProcessEnv, modelCatalog: Effect.Effect = Effect.succeed(BUNDLED_CLAUDE_MODEL_CATALOG), + proxy?: ProviderProxyRoute, ) { const commandSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; const claudeEnvironment = yield* makeClaudeEnvironment(claudeSettings, environment); const scopedModelCatalog = modelCatalog.pipe( Effect.map((catalog) => scopeClaudeModelCatalog(catalog, claudeSettings.customModels)), @@ -171,6 +176,7 @@ export const makeClaudeTextGeneration = Effect.fn("makeClaudeTextGeneration")(fu fastModeDescriptor?.type === "boolean" ? fastModeDescriptor.currentValue : undefined; const settings = { disableAllHooks: true, + ...(proxy === undefined ? {} : { env: claudeProxySettingsEnvironment(proxy) }), ...(typeof thinking === "boolean" ? { alwaysThinkingEnabled: thinking } : {}), ...(fastMode ? { fastMode: true } : {}), ...(ultracode ? { ultracode: true } : {}), @@ -192,6 +198,27 @@ export const makeClaudeTextGeneration = Effect.fn("makeClaudeTextGeneration")(fu ), ) : cwd; + const settingsArgument = + proxy === undefined + ? settingsJson + : yield* Effect.gen(function* () { + const directory = yield* fileSystem.makeTempDirectoryScoped({ + prefix: "t3code-claude-settings-", + }); + yield* fileSystem.chmod(directory, 0o700); + const settingsPath = path.join(directory, "settings.json"); + yield* fileSystem.writeFileString(settingsPath, settingsJson, { mode: 0o600 }); + return settingsPath; + }).pipe( + Effect.mapError((cause) => + normalizeCliError( + "claude", + operation, + cause, + "Failed to prepare Claude CLI settings", + ), + ), + ); const managedMcpConfig = yield* hasClaudeManagedMcpConfig(); const spawnCommand = yield* resolveSpawnCommand( claudeSettings.binaryPath || "claude", @@ -205,7 +232,7 @@ export const makeClaudeTextGeneration = Effect.fn("makeClaudeTextGeneration")(fu resolveClaudeCatalogApiModelId(catalog, resolvedModelSelection), ...(cliEffort ? ["--effort", cliEffort] : []), "--settings", - settingsJson, + settingsArgument, ...(investigate ? [ "--tools", diff --git a/apps/server/src/textGeneration/testing/ClaudeTextGeneration.fixture.mjs b/apps/server/src/textGeneration/testing/ClaudeTextGeneration.fixture.mjs index d5ca2f185eb8..d9c99b6c303d 100644 --- a/apps/server/src/textGeneration/testing/ClaudeTextGeneration.fixture.mjs +++ b/apps/server/src/textGeneration/testing/ClaudeTextGeneration.fixture.mjs @@ -1,4 +1,5 @@ import * as NodeFS from "node:fs"; +import * as NodePath from "node:path"; const argv = process.argv.slice(2); const args = argv.join(" "); @@ -26,7 +27,30 @@ if (!argv.includes("--strict-mcp-config")) { fail("text generation must not load configured MCP servers", 9); } const settingsIndex = argv.indexOf("--settings"); -if (settingsIndex === -1 || JSON.parse(argv[settingsIndex + 1]).disableAllHooks !== true) { +const settingsArgument = argv[settingsIndex + 1]; +const expectedProxy = process.env.T3_FAKE_CLAUDE_SETTINGS_EXPECTED; +if (expectedProxy) { + const { url, apiKey, platform } = JSON.parse(expectedProxy); + if (args.includes(apiKey)) fail("proxy settings exposed in argv", 13); + const settings = JSON.parse(NodeFS.readFileSync(settingsArgument, "utf8")); + if (settings.env?.ANTHROPIC_BASE_URL !== url || settings.env?.ANTHROPIC_AUTH_TOKEN !== apiKey) { + fail("settings file missing proxy route", 14); + } + if ( + platform !== "win32" && + ((NodeFS.statSync(settingsArgument).mode & 0o777) !== 0o600 || + (NodeFS.statSync(NodePath.dirname(settingsArgument)).mode & 0o777) !== 0o700) + ) { + fail("settings file permissions expose credentials", 15); + } + NodeFS.writeFileSync(process.env.T3_FAKE_CLAUDE_SETTINGS_RECEIPT, settingsArgument); +} +const settings = JSON.parse( + settingsArgument.startsWith("{") + ? settingsArgument + : NodeFS.readFileSync(settingsArgument, "utf8"), +); +if (settingsIndex === -1 || settings.disableAllHooks !== true) { fail("text generation must disable hooks", 10); } const cwdMustNotBe = process.env.T3_FAKE_CLAUDE_CWD_MUST_NOT_BE; diff --git a/apps/server/src/usage/UsageLimitSources.test.ts b/apps/server/src/usage/UsageLimitSources.test.ts index b8976fca81e7..fb3b41617788 100644 --- a/apps/server/src/usage/UsageLimitSources.test.ts +++ b/apps/server/src/usage/UsageLimitSources.test.ts @@ -1,12 +1,21 @@ import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; import { expect, it } from "@effect/vitest"; -import { ServerConfigStreamEvent, UsageLimitSourceId } from "@t3tools/contracts"; +import { + DEFAULT_SERVER_SETTINGS, + ServerConfigStreamEvent, + UsageLimitSourceId, +} from "@t3tools/contracts"; import * as Effect from "effect/Effect"; +import * as Deferred from "effect/Deferred"; +import * as Queue from "effect/Queue"; +import * as Ref from "effect/Ref"; +import * as Stream from "effect/Stream"; 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 ManagedProviderProxy from "../providerProxy/ManagedProviderProxy.ts"; import * as ServerSettings from "../serverSettings.ts"; import * as UsageLimitSources from "./UsageLimitSources.ts"; @@ -74,6 +83,7 @@ it.effect("keeps malformed source URLs encodable for config subscribers", () => Layer.mock(BackgroundPolicy.BackgroundPolicy)({ shouldRunScopeWork: () => Effect.succeed(false), }), + ManagedProviderProxy.ManagedProviderProxy.layerTest(), Layer.succeed( HttpClient.HttpClient, HttpClient.make(() => Effect.die("Unexpected network request")), @@ -83,3 +93,128 @@ it.effect("keeps malformed source URLs encodable for config subscribers", () => Effect.scoped, ), ); + +it("reports the routing proxy's accounts unless a hub already points at it or owns its id", () => { + const hub = { + kind: "cliproxy" as const, + url: "http://proxy:8317/", + managementKey: "mgmt", + enabled: true, + }; + const settings = { + ...DEFAULT_SERVER_SETTINGS, + providerProxy: { url: "http://proxy:8317/v1", apiKey: "sk", managementKey: "mgmt" }, + }; + + expect(UsageLimitSources.effectiveUsageLimitSources(settings)).toEqual({ + [UsageLimitSources.PROVIDER_PROXY_SOURCE_ID]: { + kind: "cliproxy", + label: "Proxy", + url: "http://proxy:8317", + managementKey: "mgmt", + enabled: true, + }, + }); + for (const unusable of [ + { ...hub, managementKey: "" }, + { ...hub, managementKey: "rotated-old-key" }, + { ...hub, enabled: false }, + ]) { + expect( + UsageLimitSources.effectiveUsageLimitSources({ + ...settings, + usageLimitSources: { [UsageLimitSourceId.make("hub")]: unusable }, + })[UsageLimitSources.PROVIDER_PROXY_SOURCE_ID]?.managementKey, + ).toBe("mgmt"); + } + for (const listed of [ + { [UsageLimitSourceId.make("hub")]: hub }, + { + [UsageLimitSources.PROVIDER_PROXY_SOURCE_ID]: { + ...hub, + url: "https://other.example.test", + }, + }, + ]) { + expect( + UsageLimitSources.effectiveUsageLimitSources({ ...settings, usageLimitSources: listed }), + ).toBe(listed); + } + + // A managed proxy reports through its own loopback management key; turning the host off hides it. + const managed = { url: "http://127.0.0.1:41234", apiKey: "sk", managementKey: "managed-mgmt" }; + const managedSettings = { + ...settings, + providerProxy: { ...settings.providerProxy, host: "managed" as const }, + }; + expect( + UsageLimitSources.effectiveUsageLimitSources(managedSettings, managed)[ + UsageLimitSources.PROVIDER_PROXY_SOURCE_ID + ], + ).toMatchObject({ url: "http://127.0.0.1:41234", managementKey: "managed-mgmt" }); + expect(UsageLimitSources.effectiveUsageLimitSources(managedSettings)).toEqual({}); + expect( + UsageLimitSources.effectiveUsageLimitSources( + { ...settings, providerProxy: { ...settings.providerProxy, host: "off" } }, + managed, + ), + ).toEqual({}); +}); + +it.effect("refreshes once at startup and again when managed proxy status changes", () => + Effect.gen(function* () { + const refreshes = yield* Ref.make(0); + const initialStatusRead = yield* Deferred.make(); + const changedStatusRead = yield* Deferred.make(); + const statuses = yield* Queue.unbounded<{ readonly status: "running" }>(); + const baseline = yield* ManagedProviderProxy.ManagedProviderProxy.pipe( + Effect.provide(ManagedProviderProxy.ManagedProviderProxy.layerTest()), + ); + const managed = Layer.succeed(ManagedProviderProxy.ManagedProviderProxy, { + ...baseline, + statusChanges: Stream.concat( + Stream.make({ status: "stopped" } as const), + Stream.concat( + Stream.fromEffect(Deferred.succeed(initialStatusRead, undefined)).pipe(Stream.drain), + Stream.fromQueue(statuses), + ), + ), + }); + const http = HttpClient.make((request) => + Ref.updateAndGet(refreshes, (count) => count + 1).pipe( + Effect.tap((count) => + count === 3 ? Deferred.succeed(changedStatusRead, undefined) : Effect.void, + ), + Effect.map(() => HttpClientResponse.fromWeb(request, new Response("{}", { status: 401 }))), + ), + ); + yield* Effect.gen(function* () { + const sources = yield* UsageLimitSources.make; + yield* Deferred.await(initialStatusRead); + yield* sources.refresh; + expect(yield* Ref.get(refreshes)).toBe(2); + yield* Queue.offer(statuses, { status: "running" }); + yield* Deferred.await(changedStatusRead); + yield* sources.refresh; + expect(yield* Ref.get(refreshes)).toBe(4); + }).pipe(Effect.provide([managed, Layer.succeed(HttpClient.HttpClient, http)])); + }).pipe( + Effect.provide([ + ServerSettings.layerTest({ + usageLimitSources: { + [UsageLimitSourceId.make("hub")]: { + kind: "cliproxy", + url: "http://hub.test", + managementKey: "mgmt", + enabled: true, + }, + }, + }), + Layer.mock(BackgroundPolicy.BackgroundPolicy)({ + shouldRunScopeWork: () => Effect.succeed(false), + }), + NodeCrypto.layer, + ]), + Effect.scoped, + ), +); diff --git a/apps/server/src/usage/UsageLimitSources.ts b/apps/server/src/usage/UsageLimitSources.ts index 7a068cde85f6..fc0cbf9fa60e 100644 --- a/apps/server/src/usage/UsageLimitSources.ts +++ b/apps/server/src/usage/UsageLimitSources.ts @@ -8,6 +8,12 @@ import { type UsageLimitSourceId, type UsageLimitSourceSnapshot, } from "@t3tools/contracts"; +import { normalizeProviderProxyUrl } from "@t3tools/provider-core/server/providerProxy"; +import { + resolveProxyManagement, + type ManagedProxyEndpoint, +} from "../providerProxy/effectiveProxy.ts"; +import * as ManagedProviderProxy from "../providerProxy/ManagedProviderProxy.ts"; import { resolveServerBackgroundActivitySettings } from "@t3tools/shared/backgroundActivitySettings"; import * as Context from "effect/Context"; import * as DateTime from "effect/DateTime"; @@ -46,10 +52,43 @@ function sourceLabel(id: string, config: UsageLimitSourceConfig): string { } } +export const PROVIDER_PROXY_SOURCE_ID = "provider-proxy" as UsageLimitSourceId; + +/** The routing proxy reports its accounts without a separate hub entry, unless one already points at it. */ +export function effectiveUsageLimitSources( + settings: ServerSettings, + managedProxy?: ManagedProxyEndpoint, +): Readonly> { + const management = resolveProxyManagement(settings, managedProxy); + if (management === undefined) return settings.usageLimitSources; + const { url, managementKey } = management; + const alreadyListed = Object.values(settings.usageLimitSources).some( + (source) => + source.enabled && + source.managementKey === managementKey && + normalizeProviderProxyUrl(source.url) === url, + ); + // A hand-written source may already use the synthetic id; it keeps the slot. + if (alreadyListed || PROVIDER_PROXY_SOURCE_ID in settings.usageLimitSources) { + return settings.usageLimitSources; + } + return { + ...settings.usageLimitSources, + [PROVIDER_PROXY_SOURCE_ID]: { + kind: "cliproxy", + label: "Proxy", + url, + managementKey, + enabled: true, + }, + }; +} + /** @public */ export const make = Effect.gen(function* () { const api = yield* makeCliproxyApi; const settingsService = yield* Settings.ServerSettingsService; + const managedProxy = yield* ManagedProviderProxy.ManagedProviderProxy; const backgroundPolicy = yield* BackgroundPolicy.BackgroundPolicy; const stateRef = yield* Ref.make>([]); const changes = yield* Effect.acquireRelease( @@ -89,9 +128,10 @@ export const make = 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 endpoint = yield* managedProxy.endpoint; + const entries = Object.entries( + settings === null ? {} : effectiveUsageLimitSources(settings, endpoint), + ).filter(([, config]) => config.enabled); const snapshots = yield* Effect.forEach( entries, ([id, config]) => readSource(id as UsageLimitSourceId, config), @@ -107,7 +147,9 @@ export const make = Effect.gen(function* () { const settings = yield* settingsService.getSettings.pipe( Effect.mapError(() => new UsageLimitSourceError({ detail: "Could not read hub settings." })), ); - const config = settings.usageLimitSources[input.sourceId]; + const config = effectiveUsageLimitSources(settings, yield* managedProxy.endpoint)[ + input.sourceId + ]; if (!config?.enabled || !config.managementKey) { return yield* new UsageLimitSourceError({ detail: "The usage limit source is missing or disabled.", @@ -120,9 +162,14 @@ export const make = Effect.gen(function* () { return result; }, refreshLock.withPermits(1)); - yield* settingsService.streamChanges.pipe( - Stream.map((settings) => settings.usageLimitSources), - Stream.changes, + yield* Stream.merge( + settingsService.streamChanges.pipe( + Stream.map((settings) => effectiveUsageLimitSources(settings)), + Stream.changes, + ), + // Starting makes the managed proxy answer; stopping must drop its accounts. + managedProxy.statusChanges.pipe(Stream.drop(1)), + ).pipe( Stream.runForEach(() => refresh), Effect.forkScoped, ); diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index 09db12ea35b7..25017c7f7fc0 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -97,6 +97,7 @@ import * as TerminalManager from "./terminal/Manager.ts"; import * as ThreadLaunchService from "./orchestration-v2/ThreadLaunchService.ts"; import * as ThreadSearch from "./orchestration-v2/ThreadSearch.ts"; import * as UsageLimitSources from "./usage/UsageLimitSources.ts"; +import * as ProxyAccounts from "./providerProxy/ProxyAccounts.ts"; import * as UsageService from "./usage/UsageService.ts"; import * as VcsProvisioningService from "./vcs/VcsProvisioningService.ts"; import * as VcsStatusBroadcaster from "./vcs/VcsStatusBroadcaster.ts"; @@ -594,6 +595,7 @@ const layerUnusedConfigRpcServices = Layer.mergeAll( Layer.mock(ThreadManagement.ThreadManagementService)({}), Layer.mock(ThreadSearch.ThreadSearch)({}), Layer.mock(UsageService.UsageService)({}), + Layer.mock(ProxyAccounts.ProxyAccounts)({}), Layer.mock(VcsProvisioningService.VcsProvisioningService)({}), Layer.mock(VcsStatusBroadcaster.VcsStatusBroadcaster)({}), Layer.mock(WorkspaceEntries.WorkspaceEntries)({}), diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 12436650b073..4da7a2c71173 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -270,6 +270,7 @@ import { import * as AgentSessionScanner from "./project/AgentSessionScanner.ts"; import * as AgentSessionImporter from "./project/AgentSessionImporter.ts"; import * as UsageLimitSources from "./usage/UsageLimitSources.ts"; +import * as ProxyAccounts from "./providerProxy/ProxyAccounts.ts"; const CONFIG_DISCOVERY_TIMEOUT = Duration.seconds(5); const isProviderUploadFeedbackError = Schema.is(ProviderUploadFeedbackError); @@ -1342,6 +1343,7 @@ export const layerWsRpc = ( ); const usage = yield* UsageService.UsageService; const usageLimitSources = yield* UsageLimitSources.UsageLimitSources; + const proxyAccounts = yield* ProxyAccounts.ProxyAccounts; const worktreeSetupTracker = yield* WorktreeSetupTracker.WorktreeSetupTracker; const projectCloneTracker = yield* ProjectCloneTracker.ProjectCloneTracker; const repositoryIdentityResolver = @@ -2478,6 +2480,14 @@ export const layerWsRpc = ( ), [WS_METHODS.serverUpdateProvider]: (input) => providerMaintenanceRunner.updateProvider(input), + [WS_METHODS.providerProxyGetState]: () => proxyAccounts.getState, + [WS_METHODS.providerProxyStartLogin]: ({ provider }) => proxyAccounts.startLogin(provider), + [WS_METHODS.providerProxyGetLoginStatus]: ({ state }) => + proxyAccounts.getLoginStatus(state), + [WS_METHODS.providerProxyCompleteLogin]: (input) => proxyAccounts.completeLogin(input), + [WS_METHODS.providerProxySetAccountDisabled]: (input) => + proxyAccounts.setAccountDisabled(input), + [WS_METHODS.providerProxyRemoveAccount]: ({ name }) => proxyAccounts.removeAccount(name), [WS_METHODS.providerConsumeResetCredit]: /* Request instrumentation already owns the span; keep this helper in that context. */ Effect.fnUntraced(function* (input) { diff --git a/apps/web/src/components/settings/ProviderInstanceCard.tsx b/apps/web/src/components/settings/ProviderInstanceCard.tsx index ca9db4b968d1..66439d850e6e 100644 --- a/apps/web/src/components/settings/ProviderInstanceCard.tsx +++ b/apps/web/src/components/settings/ProviderInstanceCard.tsx @@ -18,6 +18,7 @@ import * as Result from "effect/Result"; import { Fragment, useState, type ReactElement, type ReactNode } from "react"; import { isProviderDriverKind, + PROXY_ROUTED_DRIVERS, resolveProviderInstanceEnabled, type ProviderInstanceConfig, type ProviderInstanceEnvironmentVariable, @@ -259,6 +260,51 @@ export function nextProviderEnvironmentWithFieldValue( return next; } +function proxyBlockedReason(instance: ProviderInstanceConfig): string | undefined { + if (instance.driver === "codex" && readCodexSetupMode(instance.config) === "managed") { + return "T3-managed ChatGPT sign-in can't use a proxy."; + } + const serverUrl = (instance.config as { readonly serverUrl?: unknown } | undefined)?.serverUrl; + if (instance.driver === "opencode" && typeof serverUrl === "string" && serverUrl.trim() !== "") { + return "An external OpenCode server can't use this environment's proxy."; + } + return undefined; +} + +function ProviderProxyRow(props: { + readonly instanceId: ProviderInstanceId; + readonly instance: ProviderInstanceConfig; + readonly proxyConfigured: boolean; + readonly onUpdate: (nextInstance: ProviderInstanceConfig) => void; +}) { + const enabled = props.instance.useProxy === true; + const blocked = proxyBlockedReason(props.instance); + const switchId = `provider-instance-${props.instanceId}-use-proxy`; + return ( + Use proxy} + description={ + blocked ?? + (props.proxyConfigured + ? "Send this provider's requests through the proxy and its accounts." + : "Choose where the proxy runs under Proxy in these settings first.") + } + control={ + { + const { useProxy: _previous, ...rest } = props.instance; + props.onUpdate(useProxy ? { ...rest, useProxy: true } : rest); + }} + /> + } + /> + ); +} + export function nextProviderEnvironmentWithFieldSecret( environment: ReadonlyArray | undefined, field: ProviderEnvironmentField, @@ -674,6 +720,8 @@ interface ProviderInstanceCardProps { readonly environmentId?: EnvironmentId | undefined; readonly secretSources?: ReadonlyArray; readonly projectOverride?: ProjectOverrideControl | undefined; + /** Whether this environment has a proxy for `useProxy` to route through. */ + readonly proxyConfigured?: boolean | undefined; readonly acpProjects?: | ReadonlyArray<{ readonly id: ProjectId; @@ -692,6 +740,7 @@ const NO_CUSTOM_MODELS: ReadonlyArray = []; export function ProviderInstanceCard({ instanceId, instance, + proxyConfigured = false, driverOption, liveProvider, mode, @@ -1334,6 +1383,22 @@ export function ProviderInstanceCard({ ) : null} + {PROXY_ROUTED_DRIVERS.has(instance.driver) ? ( + + + + ) : null} + = { + off: "Off", + managed: "This environment", + external: "External URL", +}; + +const HOST_DESCRIPTIONS: Record = { + off: "Send Claude, Codex, and OpenCode requests through a CLIProxyAPI server, which spreads them across several accounts.", + managed: + "T3 Code downloads CLIProxyAPI from GitHub and runs it on this machine. Turn on Use proxy for a Claude, Codex, or OpenCode provider to send its requests here.", + external: + "A CLIProxyAPI server you run. Turn on Use proxy for a Claude, Codex, or OpenCode provider to send its requests here.", +}; + +const PROVIDER_LABELS: Readonly> = { claude: "Claude", codex: "Codex" }; +const LOGIN_PROVIDERS: ReadonlyArray = ["claude", "codex"]; + +const failureToast = (title: string, description: string) => + toastManager.add(stackedThreadToast({ type: "error", title, description })); + +function runtimeDescription(runtime: ProviderProxyRuntimeStatus): string { + switch (runtime.status) { + case "stopped": + return "Stopped."; + case "installing": + return "Downloading CLIProxyAPI…"; + case "starting": + return "Starting…"; + case "running": + return runtime.version ? `Running CLIProxyAPI ${runtime.version}.` : "Running."; + case "error": + return runtime.message ?? "CLIProxyAPI stopped. Restarting…"; + } +} + +function accountStateDescription(account: ProviderProxyAccount): string { + switch (account.state) { + case "active": + return "Active"; + case "disabled": + return "Disabled"; + case "cooldown": + return account.retryAt + ? `Cooling down until ${new Date(account.retryAt).toLocaleTimeString([], { hour: "numeric", minute: "2-digit" })}` + : "Cooling down"; + case "error": + return account.statusMessage ? `Error: ${account.statusMessage}` : "Error"; + } +} + +/** Secrets arrive redacted: any non-empty value means one is stored. */ +export function ProviderProxySettings({ + environmentId, + proxy, + readOnly, +}: { + readonly environmentId: EnvironmentId; + readonly proxy: UnifiedSettings["providerProxy"]; + readonly readOnly: boolean; +}) { + const updateSettings = useUpdateEnvironmentSettings(environmentId); + const update = (patch: Partial) => + updateSettings({ providerProxy: patch }); + const host = providerProxyHost(proxy); + // The server reads proxy state only for sessions that can manage providers. + const canManageProviders = useAtomValue( + providerProxyEnvironment.startLogin.permissionAtom(environmentId), + ); + const managesAccounts = + canManageProviders && + (host === "managed" || (host === "external" && proxy.managementKey !== "")); + + return ( + + + + + } + /> + {host === "external" ? ( + + ) : null} + {managesAccounts ? ( + + ) : null} + + ); +} + +function ExternalProxyRows({ + proxy, + readOnly, + update, +}: { + readonly proxy: UnifiedSettings["providerProxy"]; + readonly readOnly: boolean; + readonly update: (patch: Partial) => void; +}) { + return ( + <> + Proxy URL} + control={ + update({ url: url.trim() })} + /> + } + /> + update({ apiKey })} + /> + update({ managementKey })} + /> + + ); +} + +function SecretRow({ + id, + title, + description, + configured, + readOnly, + onCommit, +}: { + readonly id: string; + readonly title: string; + readonly description: string; + readonly configured: boolean; + readonly readOnly: boolean; + readonly onCommit: (value: string) => void; +}) { + return ( + {title}} + description={description} + control={ +
+ onCommit(value.trim())} + /> + {configured && !readOnly ? ( + + ) : null} +
+ } + /> + ); +} + +function ProxyAccountRows({ + environmentId, + showRuntime, +}: { + readonly environmentId: EnvironmentId; + readonly showRuntime: boolean; +}) { + const state = useEnvironmentQuery(providerProxyEnvironment.state({ environmentId, input: {} })); + const canAdd = useAtomValue(providerProxyEnvironment.startLogin.permissionAtom(environmentId)); + const startLogin = useAtomCommand(providerProxyEnvironment.startLogin, { + label: "start proxy sign-in", + }); + const [login, setLogin] = useState< + (ProviderProxyLogin & { readonly provider: ProviderProxyLoginProvider }) | null + >(null); + const [starting, setStarting] = useState(false); + + const add = async (provider: ProviderProxyLoginProvider) => { + setStarting(true); + const result = await startLogin({ environmentId, input: { provider } }); + setStarting(false); + if (result._tag === "Success") setLogin({ ...result.value, provider }); + else if (!isAtomCommandInterrupted(result)) { + failureToast("Could not start sign-in", String(squashAtomCommandFailure(result))); + } + }; + + if (state.error) { + return ; + } + const data = state.data; + if (!data) return ; + const accounts = data.accounts ?? []; + const providers = [ + ...new Set([...LOGIN_PROVIDERS, ...accounts.map((account) => account.provider)]), + ]; + + return ( + <> + {showRuntime ? ( + + ) : null} + {data.accountsError ? ( + + ) : null} + {data.accounts === null ? null : ( + + {LOGIN_PROVIDERS.map((provider) => ( + + ))} + + ) : null + } + /> + )} + {login === null ? null : ( + setLogin(null)} /> + )} + {providers.flatMap((provider) => + accounts + .filter((account) => account.provider === provider) + .map((account) => ( + + )), + )} + + ); +} + +function ProxyLoginRow({ + environmentId, + login, + onDone, +}: { + readonly environmentId: EnvironmentId; + readonly login: ProviderProxyLogin & { readonly provider: ProviderProxyLoginProvider }; + readonly onDone: () => void; +}) { + const status = useEnvironmentQuery( + providerProxyEnvironment.loginStatus({ environmentId, input: { state: login.state } }), + ); + const completeLogin = useAtomCommand(providerProxyEnvironment.completeLogin, { + label: "finish proxy sign-in", + }); + const [redirectUrl, setRedirectUrl] = useState(""); + const [submitting, setSubmitting] = useState(false); + const label = PROVIDER_LABELS[login.provider] ?? login.provider; + const outcome = status.data?.status; + + const open = () => { + void ensureLocalApi() + .shell.openExternal(login.url) + .catch(() => + failureToast("Could not open the sign-in page", "Copy the link and open it yourself."), + ); + }; + const submit = async () => { + setSubmitting(true); + const result = await completeLogin({ + environmentId, + input: { state: login.state, redirectUrl: redirectUrl.trim() }, + }); + setSubmitting(false); + if (result._tag === "Failure" && !isAtomCommandInterrupted(result)) { + failureToast("Could not finish sign-in", String(squashAtomCommandFailure(result))); + } + }; + + const description = + outcome === "ok" + ? `${label} account added.` + : outcome === "error" + ? (status.data?.message ?? "Sign-in failed.") + : "Open the sign-in page. When the browser lands on a localhost address that does not load, copy that address and paste it here."; + + return ( +
+
+
+
+

Add a {label} account

+

+ {description} +

+
+
+ {outcome === "ok" || outcome === "error" ? ( + + ) : ( + <> + + + + )} +
+
+ {outcome === "ok" || outcome === "error" ? null : ( +
{ + event.preventDefault(); + if (redirectUrl.trim() !== "") void submit(); + }} + > + setRedirectUrl(event.target.value)} + /> + +
+ )} +
+
+ ); +} + +function ProxyAccountRow({ + environmentId, + account, + providerLabel, +}: { + readonly environmentId: EnvironmentId; + readonly account: ProviderProxyAccount; + readonly providerLabel: string; +}) { + const setDisabled = useAtomCommand(providerProxyEnvironment.setAccountDisabled, { + label: "change proxy account", + }); + const remove = useAtomCommand(providerProxyEnvironment.removeAccount, { + label: "remove proxy account", + }); + const canChange = useAtomValue( + providerProxyEnvironment.setAccountDisabled.permissionAtom(environmentId), + ); + const canRemove = useAtomValue( + providerProxyEnvironment.removeAccount.permissionAtom(environmentId), + ); + const [busy, setBusy] = useState(false); + const title = account.email ?? account.name; + const disabled = account.state === "disabled"; + + const toggle = async () => { + setBusy(true); + const result = await setDisabled({ + environmentId, + input: { name: account.name, disabled: !disabled }, + }); + setBusy(false); + if (result._tag === "Failure" && !isAtomCommandInterrupted(result)) { + failureToast( + disabled ? "Could not enable the account" : "Could not disable the account", + String(squashAtomCommandFailure(result)), + ); + } + }; + const removeAccount = async () => { + const confirmed = await (requestConfirmDialog( + `Remove ${title} from the proxy? Its sign-in is deleted; add it again to use it.`, + { variant: "destructive" }, + ) ?? Promise.resolve(false)); + if (!confirmed) return; + setBusy(true); + const result = await remove({ environmentId, input: { name: account.name } }); + setBusy(false); + if (result._tag === "Failure" && !isAtomCommandInterrupted(result)) { + failureToast("Could not remove the account", String(squashAtomCommandFailure(result))); + } + }; + + return ( +
+
+
+

{title}

+

+ {providerLabel} · {accountStateDescription(account)} +

+
+
+ {canChange ? ( + + ) : null} + {canRemove ? ( + + ) : null} +
+
+
+ ); +} diff --git a/apps/web/src/components/settings/ProviderSettingsPanel.tsx b/apps/web/src/components/settings/ProviderSettingsPanel.tsx index e0cdea543216..49dfb203037e 100644 --- a/apps/web/src/components/settings/ProviderSettingsPanel.tsx +++ b/apps/web/src/components/settings/ProviderSettingsPanel.tsx @@ -22,6 +22,7 @@ import { type ProjectId, type ProviderInstanceConfig, type ProviderInstanceId, + providerProxyConfigured, type ServerProvider, resolveEnvironmentMachineKind, resolveProjectProviderInstanceEnabled, @@ -99,6 +100,7 @@ import { AddProviderInstanceDialog } from "./AddProviderInstanceDialog"; import { useManagedPolicy } from "~/state/managedPolicy"; import { ExpandableText } from "./ExpandableText"; import { ProviderInstanceCard } from "./ProviderInstanceCard"; +import { ProviderProxySettings } from "./ProviderProxySettings"; import { useSecretSources } from "./SecretSourceInput"; import { UsageProviderSettings } from "./UsageProviderSettings"; import { ProviderSetupSection, readAntigravityAuthMethod } from "./ProviderSetupSection"; @@ -358,7 +360,8 @@ function ProviderSettingsPanelContent(target: ProviderSettingsTarget) { } else if ( !target.scoped && (searchTargetId === searchableSetting("provider-health-check-interval").id || - searchTargetId === searchableSetting("usage-providers").id) && + searchTargetId === searchableSetting("usage-providers").id || + searchTargetId === searchableSetting("provider-proxy").id) && !selectedEnvironmentCanRenderSettings && searchableEnvironmentId !== undefined ) { @@ -937,6 +940,7 @@ export function EnvironmentProviderSettings({ canWriteSettings={canWriteSettings} liveProvider={serverProviders.find((candidate) => candidate.instanceId === row.instanceId)} isUpdateDispatching={updatingProviderInstanceIds.has(row.instanceId)} + proxyConfigured={providerProxyConfigured(settings.providerProxy)} showProjectOverride={showProjectOverride} projectOverrideValue={ showProjectOverride @@ -1075,6 +1079,13 @@ export function EnvironmentProviderSettings({ + +