diff --git a/apps/server/src/binMain.ts b/apps/server/src/binMain.ts index 0dabd73873440..22052cf29654c 100644 --- a/apps/server/src/binMain.ts +++ b/apps/server/src/binMain.ts @@ -19,9 +19,6 @@ export async function runCliCommand() { // Exit 2 identifies document rejections; OS and process failures remain retryable. process.exitCode = 2; } - } else if (command === "__managed-chatgpt-token") { - const { runManagedChatGptTokenCli } = await import("./cli/managedChatGptToken.ts"); - await runManagedChatGptTokenCli(process.argv.slice(3)); } else if (command === "acp-mcp-bridge" || command === "acp-mcp-call") { const { runAcpMcpCliFastPath } = await import("./mcp/AcpMcpStdioBridge.ts"); await runAcpMcpCliFastPath(command, process.argv.slice(3)); diff --git a/apps/server/src/cli/managedChatGptToken.ts b/apps/server/src/cli/managedChatGptToken.ts deleted file mode 100644 index 84ed18fb5228a..0000000000000 --- a/apps/server/src/cli/managedChatGptToken.ts +++ /dev/null @@ -1,48 +0,0 @@ -// @effect-diagnostics nodeBuiltinImport:off - Lean self-invoked local IPC client; no auth/store graph. -import * as NodeNet from "node:net"; -import * as Effect from "effect/Effect"; -import * as Exit from "effect/Exit"; -import * as Schema from "effect/Schema"; -import * as Protocol from "../provider/CodexManagedTokenProtocol.ts"; - -const decodeArguments = Schema.decodeUnknownEffect( - Schema.Tuple([ - Schema.NonEmptyString, - Protocol.Request.fields.instanceId, - Protocol.Request.fields.environmentId, - Protocol.Request.fields.profileKey, - ]), -); -const encodeRequest = Schema.encodeSync(Schema.fromJsonString(Protocol.Request)); -const decodeResponse = Schema.decodeUnknownEffect(Schema.fromJsonString(Protocol.Response), { - onExcessProperty: "error", -}); - -/** Codex's auth.command protocol is a bare token on stdout, never a CLI diagnostic. */ -export async function runManagedChatGptTokenCli(args: ReadonlyArray): Promise { - const program = Effect.gen(function* () { - const [endpoint, instanceId, environmentId, profileKey] = yield* decodeArguments(args); - const socket = yield* Effect.acquireRelease( - Effect.sync(() => NodeNet.createConnection(endpoint)), - (socket) => Effect.sync(() => socket.destroy()), - ); - socket.on("error", () => {}); - socket.once("connect", () => - socket.write(encodeRequest({ version: 1, instanceId, environmentId, profileKey }) + "\n"), - ); - return yield* Protocol.readFrame( - socket, - Protocol.RESPONSE_BYTES, - Protocol.COMMAND_TIMEOUT_MS, - ).pipe(Effect.flatMap(decodeResponse)); - }).pipe(Effect.scoped); - const result = await Effect.runPromiseExit(program); - if (Exit.isFailure(result)) { - process.stderr.write( - "Could not access the managed ChatGPT connection. Reconnect in T3 Code.\n", - ); - process.exitCode = 1; - return; - } - process.stdout.write(result.value.token); -} diff --git a/apps/server/src/provider/CodexChatGptAuth.test.ts b/apps/server/src/provider/CodexChatGptAuth.test.ts index 45ea557b00b5a..3e22a0a6e0d5b 100644 --- a/apps/server/src/provider/CodexChatGptAuth.test.ts +++ b/apps/server/src/provider/CodexChatGptAuth.test.ts @@ -1,9 +1,4 @@ // @effect-diagnostics nodeBuiltinImport:off globalFetchInEffect:off - Local mock OAuth server validates the browser callback boundary. -import * as NodeNet from "node:net"; -import * as Context from "effect/Context"; -import * as Scope from "effect/Scope"; -import * as ServerConfig from "../config.ts"; -import * as CodexManagedTokenBroker from "./CodexManagedTokenBroker.ts"; import * as NodeHttp from "node:http"; import * as NodeCrypto from "node:crypto"; import * as NodePlatformCrypto from "@effect/platform-node/NodeCrypto"; @@ -18,7 +13,7 @@ import * as TestClock from "effect/testing/TestClock"; import * as Fiber from "effect/Fiber"; import * as Deferred from "effect/Deferred"; import { subscribeChatGptHandoff } from "./CodexChatGptHandoff.ts"; -import { FetchHttpClient, HttpClient, HttpClientResponse } from "effect/http"; +import { FetchHttpClient } from "effect/http"; import * as ServerSecretStore from "../auth/ServerSecretStore.ts"; import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; import * as ProviderCredentialStore from "./ProviderCredentialStore.ts"; @@ -1769,171 +1764,25 @@ it.effect("a running session renews its bound profile but refuses a replacement ), ); -const makeRenewalBroker = Effect.gen(function* () { - const oauthEntered = yield* Deferred.make(); - const oauthRelease = yield* Deferred.make(); - let pauseOAuth = false; - const refreshTokens: string[] = []; - const credentials = { - clientId: "oaiapp_broker", - subject: "broker-user", - idToken: "broker-sign-in", - email: null, - accessToken: "expired-access", - refreshToken: "single-use-refresh", - expiresAt: 0, - earliestRefreshAt: null, - scopes: ["chatgpt.tokens.use.direct"], - }; - const environmentId = EnvironmentId.make("00000000-0000-4000-8000-000000000001"); - const http = HttpClient.make((request) => - Effect.gen(function* () { - if (request.url.endsWith("openid-configuration")) { - return HttpClientResponse.fromWeb( - request, - Response.json({ - issuer: "https://auth.openai.com", - authorization_endpoint: "https://auth.openai.com/authorize", - token_endpoint: "https://auth.openai.com/token", - jwks_uri: "https://auth.openai.com/jwks", - }), - ); - } - assert.equal(request.url, "https://auth.openai.com/token"); - assert.equal(request.body._tag, "Uint8Array"); - if (request.body._tag !== "Uint8Array") return yield* Effect.die("Missing refresh body"); - const body = new URLSearchParams(new TextDecoder().decode(request.body.body)); - refreshTokens.push(body.get("refresh_token")!); - yield* Deferred.succeed(oauthEntered, undefined); - if (pauseOAuth) yield* Deferred.await(oauthRelease); - return HttpClientResponse.fromWeb( - request, - Response.json({ - access_token: "renewed-access", - refresh_token: "rotated-refresh", - expires_in: 3600, - token_type: "Bearer", - scope: "chatgpt.tokens.use.direct", - }), - ); - }), - ); - return yield* Effect.gen(function* () { - const store = yield* ProviderCredentialStore.make("codex-chatgpt", instanceId); - yield* store.set(new TextEncoder().encode(JSON.stringify(credentials))); - const auth = yield* makeCodexChatGptAuth({ instanceId }); - const profileKey = yield* auth.profileKey(credentials); - const scope = yield* Scope.make(); - yield* Effect.addFinalizer((exit) => Scope.close(scope, exit)); - const context = yield* Layer.buildWithScope(CodexManagedTokenBroker.layer, scope); - const broker = Context.get(context, CodexManagedTokenBroker.CodexManagedTokenBroker); - const endpoint = yield* broker.endpoint; - yield* Effect.addFinalizer(() => Deferred.succeed(oauthRelease, undefined)); - return { - auth, - store, - broker, - endpoint, - refreshTokens, - request: { version: 1, instanceId, environmentId, profileKey }, - oauthEntered, - oauthRelease, - pause: () => { - pauseOAuth = true; - }, - }; - }).pipe( - Effect.provide(ServerSecretStore.layer), - Effect.provideService(HttpClient.HttpClient, http), - Effect.provideService(ServerEnvironment.ServerEnvironmentIdentity, { - getEnvironmentId: Effect.succeed(environmentId), - }), - ); -}); - -const brokerClient = (endpoint: string, request: unknown) => - Effect.acquireRelease( - Effect.callback<{ socket: NodeNet.Socket; response: Promise }, Error>((resume) => { - const socket = NodeNet.createConnection(endpoint); - let text = ""; - socket.setEncoding("utf8"); - socket.on("data", (chunk) => { - text += chunk; - }); - const response = new Promise((resolve) => socket.once("close", () => resolve(text))); - socket.on("error", () => {}); - socket.once("error", (error) => resume(Effect.fail(error))); - socket.once("connect", () => { - socket.write(JSON.stringify(request) + "\n"); - resume(Effect.succeed({ socket, response })); - }); - return Effect.sync(() => { - if (socket.connecting) socket.destroy(); - }); - }), - ({ socket }) => Effect.sync(() => socket.destroy()), - ); -const provideBroker = (effect: Effect.Effect) => - effect.pipe( - Effect.scoped, - Effect.provide( - ServerConfig.layerTest(process.cwd(), { prefix: "t3-renewal-test-" }).pipe( - Layer.provideMerge(Layer.mergeAll(NodeServices.layer, NodePlatformCrypto.layer)), - ), - ), - ); - -// regression: #16132 a renewal must outlive the Codex helper that asked for it, because the -// refresh token it spends is single-use. -it.effect("server-owned renewal persists rotated credentials after the helper disconnects", () => - provideBroker( - Effect.gen(function* () { - const h = yield* makeRenewalBroker; - h.pause(); - const client = yield* brokerClient(h.endpoint, h.request); - yield* Deferred.await(h.oauthEntered); - client.socket.destroy(); - yield* Effect.promise(() => client.response); - assert.equal(Option.getOrThrow(yield* h.auth.read).refreshToken, "single-use-refresh"); - yield* Deferred.succeed(h.oauthRelease, undefined); - yield* h.broker.awaitIdle; - const persisted = Option.getOrThrow(yield* h.auth.read); - assert.equal(persisted.refreshToken, "rotated-refresh"); - assert.equal((yield* h.auth.access).accessToken, "renewed-access"); - assert.deepEqual(h.refreshTokens, ["single-use-refresh"]); - }), - ), -); - -it.live("the private broker refuses another environment, instance, or sign-in profile", () => - provideBroker( +// regression: #29927 the API can reject a token before its locally stated expiry. +it.effect("renews a locally unexpired rejected token once across bound sessions", () => + provision( Effect.gen(function* () { - const h = yield* makeRenewalBroker; - yield* Effect.forEach( - [ - { - ...h.request, - environmentId: EnvironmentId.make("00000000-0000-4000-8000-000000000002"), - }, - { ...h.request, instanceId: ProviderInstanceId.make("another-instance") }, - { ...h.request, profileKey: "0".repeat(64) }, - ], - (request) => - Effect.scoped( - Effect.gen(function* () { - const client = yield* brokerClient(h.endpoint, request); - assert.equal(yield* Effect.promise(() => client.response), ""); - }), - ), + const h = yield* makeHarness; + yield* h.signIn; + yield* h.phase("succeeded"); + const saved = Option.getOrThrow(yield* h.auth.read); + const profileKey = yield* h.auth.profileKey(saved); + const first = yield* h.bindProfile(profileKey); + const second = yield* h.bindProfile(profileKey); + const renewed = yield* Effect.all( + [first.renew(saved.accessToken), second.renew(saved.accessToken)], + { concurrency: "unbounded" }, ); - yield* h.broker.awaitIdle; - assert.deepEqual(h.refreshTokens, []); - const client = yield* brokerClient(h.endpoint, h.request); - assert.deepEqual(JSON.parse(yield* Effect.promise(() => client.response)), { - token: "renewed-access", - }); - yield* h.broker.awaitIdle; - assert.deepEqual(h.refreshTokens, ["single-use-refresh"]); + assert.equal(h.refreshes(), 1); + assert.equal(renewed[0].accessToken, "access-1"); + assert.deepEqual(renewed[0], renewed[1]); + assert.equal(yield* h.auth.profileKey(renewed[0]), profileKey); }), ), ); diff --git a/apps/server/src/provider/CodexChatGptAuth.ts b/apps/server/src/provider/CodexChatGptAuth.ts index 5be14a8a1a08f..fa5197cafdaa3 100644 --- a/apps/server/src/provider/CodexChatGptAuth.ts +++ b/apps/server/src/provider/CodexChatGptAuth.ts @@ -818,69 +818,82 @@ export const makeCodexChatGptAuth = Effect.fn("makeCodexChatGptAuth")(function* ), ) .pipe(Effect.map(Hex.encode), Effect.orDie); - const access = Effect.gen(function* () { - const saved = yield* read; - if (Option.isNone(saved)) - return yield* failure("access", "Sign in with ChatGPT to use managed Codex."); - let record = saved.value; - if ( - options.expectedProfileKey !== undefined && - (yield* profileKey(record)) !== options.expectedProfileKey - ) - return yield* failure("access", "The ChatGPT connection changed. Start a new session."); - if (!record.scopes.includes(REQUIRED_SCOPE)) - return yield* failure( - "sharing", - "Token sharing is disabled. Sign in with ChatGPT and enable token sharing.", - ); - const now = yield* Clock.currentTimeMillis; - if (record.expiresAt - now > 60_000) return record; - if (record.earliestRefreshAt !== null && record.earliestRefreshAt > now) { - if (record.expiresAt > now) return record; - return yield* failure( - "refresh", - "ChatGPT cannot renew this connection yet. Try again shortly.", - ); - } - if (!record.refreshToken) { - yield* remove; - return yield* failure("refresh", "Your ChatGPT connection expired. Sign in again."); - } - const tokens = yield* exchange( - new URLSearchParams({ - grant_type: "refresh_token", - client_id: record.clientId, - refresh_token: record.refreshToken, - resource, - }), - ).pipe( - Effect.interruptible, - Effect.timeout("20 seconds"), - Effect.mapError((error) => - error._tag === "ProviderSetupError" && error.operation === "client" - ? error - : failure("refresh", "Could not renew the ChatGPT connection. Retry, or sign in again."), - ), - ); - if (!tokens.refresh_token || tokens.token_type.toLowerCase() !== "bearer") { - return yield* failure("refresh", "ChatGPT returned an invalid renewal. Sign in again."); - } - record = { - ...record, - accessToken: tokens.access_token, - refreshToken: tokens.refresh_token, - expiresAt: (yield* Clock.currentTimeMillis) + tokens.expires_in * 1000, - earliestRefreshAt: earliest(tokens.earliest_refresh_at), - scopes: tokens.scope.split(/\s+/).filter(Boolean), - }; - yield* save(record); - if (!record.scopes.includes(REQUIRED_SCOPE)) - return yield* failure( - "sharing", - "ChatGPT token sharing is no longer enabled. Sign in again.", + const resolveAccess = Effect.fn("CodexChatGptAuth.access")( + function* (rejectedAccessToken: string | undefined) { + const saved = yield* read; + if (Option.isNone(saved)) + return yield* failure("access", "Sign in with ChatGPT to use managed Codex."); + let record = saved.value; + if ( + options.expectedProfileKey !== undefined && + (yield* profileKey(record)) !== options.expectedProfileKey + ) + return yield* failure("access", "The ChatGPT connection changed. Start a new session."); + if (!record.scopes.includes(REQUIRED_SCOPE)) + return yield* failure( + "sharing", + "Token sharing is disabled. Sign in with ChatGPT and enable token sharing.", + ); + const now = yield* Clock.currentTimeMillis; + if (record.accessToken === rejectedAccessToken) { + record = { ...record, expiresAt: Math.min(record.expiresAt, now) }; + yield* save(record); + } + if (record.expiresAt - now > 60_000) return record; + if (record.earliestRefreshAt !== null && record.earliestRefreshAt > now) { + if (record.expiresAt > now) return record; + return yield* failure( + "refresh", + "ChatGPT cannot renew this connection yet. Try again shortly.", + ); + } + if (!record.refreshToken) { + yield* remove; + return yield* failure("refresh", "Your ChatGPT connection expired. Sign in again."); + } + const tokens = yield* exchange( + new URLSearchParams({ + grant_type: "refresh_token", + client_id: record.clientId, + refresh_token: record.refreshToken, + resource, + }), + ).pipe( + Effect.interruptible, + Effect.timeout("20 seconds"), + Effect.mapError((error) => + error._tag === "ProviderSetupError" && error.operation === "client" + ? error + : failure( + "refresh", + "Could not renew the ChatGPT connection. Retry, or sign in again.", + ), + ), ); - return record; - }).pipe(Effect.uninterruptible, withSessionLock, lock.withPermit); + if (!tokens.refresh_token || tokens.token_type.toLowerCase() !== "bearer") { + return yield* failure("refresh", "ChatGPT returned an invalid renewal. Sign in again."); + } + record = { + ...record, + accessToken: tokens.access_token, + refreshToken: tokens.refresh_token, + expiresAt: (yield* Clock.currentTimeMillis) + tokens.expires_in * 1000, + earliestRefreshAt: earliest(tokens.earliest_refresh_at), + scopes: tokens.scope.split(/\s+/).filter(Boolean), + }; + yield* save(record); + if (!record.scopes.includes(REQUIRED_SCOPE)) + return yield* failure( + "sharing", + "ChatGPT token sharing is no longer enabled. Sign in again.", + ); + return record; + }, + Effect.uninterruptible, + withSessionLock, + lock.withPermit, + ); + const access = resolveAccess(undefined); const logout = Effect.gen(function* () { const credentials = Option.getOrUndefined(yield* read); let confirmed = !credentials?.refreshToken; @@ -1117,6 +1130,7 @@ export const makeCodexChatGptAuth = Effect.fn("makeCodexChatGptAuth")(function* controller: { ...controller, reconnectProfile, importProfile }, read, access, + renew: (rejectedAccessToken: string) => resolveAccess(rejectedAccessToken), profileKey, exportProfile, revoke: lock.withPermit(withSessionLock(remove)), diff --git a/apps/server/src/provider/CodexManagedResponses.test.ts b/apps/server/src/provider/CodexManagedResponses.test.ts new file mode 100644 index 0000000000000..c9418afa9cb74 --- /dev/null +++ b/apps/server/src/provider/CodexManagedResponses.test.ts @@ -0,0 +1,344 @@ +import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, it } from "@effect/vitest"; +import { EnvironmentId, ProviderInstanceId } from "@t3tools/contracts"; +import * as Clock from "effect/Clock"; +import * as Deferred from "effect/Deferred"; +import * as Digest from "effect/crypto/Digest"; +import * as Fiber from "effect/Fiber"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import { FetchHttpClient, HttpClient, HttpClientRequest, HttpClientResponse } from "effect/http"; +import * as ServerSecretStore from "../auth/ServerSecretStore.ts"; +import * as ServerConfig from "../config.ts"; +import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; +import { makeCodexChatGptAuth } from "./CodexChatGptAuth.ts"; +import * as CodexManagedResponses from "./CodexManagedResponses.ts"; +import * as ProviderCredentialStore from "./ProviderCredentialStore.ts"; + +const instanceId = ProviderInstanceId.make("codex"); +const expired = { error: { code: "token_expired", message: "expired" } }; +const makeHarness = Effect.fnUntraced(function* ( + options: { + error?: unknown; + retryStatus?: number; + renewalFails?: boolean; + concurrent?: boolean; + concurrentRegistration?: boolean; + pauseRenewal?: boolean; + constraint?: "replacement" | "earliestRefreshAt"; + } = {}, +) { + const store = yield* ProviderCredentialStore.make("codex-chatgpt", instanceId); + const credentials = { + clientId: "oaiapp_bridge", + subject: "bridge-user", + idToken: "sign-in", + email: null, + accessToken: "stale-access", + refreshToken: "single-use-refresh", + expiresAt: (yield* Clock.currentTimeMillis) + 3_600_000, + earliestRefreshAt: null as number | null, + scopes: ["chatgpt.tokens.use.direct"], + }; + const save = (record: typeof credentials) => + store.set(new TextEncoder().encode(JSON.stringify(record))); + yield* save(credentials); + const requests: { token: string | undefined; body: string; url: string }[] = []; + const refreshTokens: string[] = []; + const bothRejected = yield* Deferred.make(); + const renewalEntered = yield* Deferred.make(); + const renewalRelease = yield* Deferred.make(); + yield* Effect.addFinalizer(() => Deferred.succeed(renewalRelease, undefined)); + const http = HttpClient.make((request) => + Effect.gen(function* () { + if (request.url.endsWith("openid-configuration")) + return HttpClientResponse.fromWeb( + request, + Response.json({ + issuer: "https://auth.openai.com", + authorization_endpoint: "https://auth.openai.com/authorize", + token_endpoint: "https://auth.openai.com/token", + jwks_uri: "https://auth.openai.com/jwks", + }), + ); + if (request.body._tag !== "Uint8Array") return yield* Effect.die("Missing request body"); + const body = new TextDecoder().decode(request.body.body); + if (request.url === "https://auth.openai.com/token") { + refreshTokens.push(new URLSearchParams(body).get("refresh_token")!); + yield* Deferred.succeed(renewalEntered, undefined); + if (options.pauseRenewal) yield* Deferred.await(renewalRelease); + return HttpClientResponse.fromWeb( + request, + options.renewalFails + ? Response.json({ error: "temporarily_unavailable" }, { status: 503 }) + : Response.json({ + access_token: "fresh-access", + refresh_token: "rotated-refresh", + expires_in: 3600, + token_type: "Bearer", + scope: "chatgpt.tokens.use.direct", + }), + ); + } + requests.push({ token: request.headers.authorization, body, url: request.url }); + if (request.headers.authorization === "Bearer stale-access") { + if (options.constraint) + yield* save({ + ...credentials, + ...(options.constraint === "replacement" + ? { idToken: "new-sign-in" } + : { earliestRefreshAt: credentials.expiresAt }), + }).pipe(Effect.orDie); + if (options.concurrent) { + if (requests.length === 2) yield* Deferred.succeed(bothRejected, undefined); + yield* Deferred.await(bothRejected); + } + return HttpClientResponse.fromWeb( + request, + Response.json(options.error ?? expired, { status: 401 }), + ); + } + return HttpClientResponse.fromWeb( + request, + options.retryStatus + ? Response.json(expired, { status: options.retryStatus }) + : new Response('data: {"output":"OK"}\n\n', { + headers: { "content-type": "text/event-stream" }, + }), + ); + }), + ); + const auth = yield* makeCodexChatGptAuth({ instanceId }).pipe( + Effect.provideService(HttpClient.HttpClient, http), + ); + const digest = yield* Digest.Digest; + const secondRegistering = yield* Deferred.make(); + const firstRegistered = yield* Deferred.make(); + let registrations = 0; + const bridge = yield* CodexManagedResponses.make().pipe( + Effect.provideService(HttpClient.HttpClient, http), + Effect.provideService(Digest.Digest, { + ...digest, + digest: (algorithm, input) => + Effect.gen(function* () { + // Pause credential-store construction after both registrations miss the grant lookup. + if ( + options.concurrentRegistration && + new TextDecoder().decode(input) === + `${"codex-chatgpt".length}:codex-chatgpt${instanceId}` + ) { + const registration = ++registrations; + if (registration === 1) yield* Deferred.await(secondRegistering); + if (registration === 2) { + yield* Deferred.succeed(secondRegistering, undefined); + yield* Deferred.await(firstRegistered); + } + } + return yield* digest.digest(algorithm, input); + }), + }), + ); + const profileKey = yield* auth.profileKey(credentials); + const register = bridge.register({ instanceId, profileKey }); + const tokens = options.concurrentRegistration + ? yield* Effect.gen(function* () { + const first = yield* register.pipe(Effect.forkChild); + const second = yield* register.pipe(Effect.forkChild); + const token = yield* Fiber.join(first); + yield* Deferred.succeed(firstRegistered, undefined); + return [token, yield* Fiber.join(second)]; + }) + : [yield* register]; + const token = tokens[0]!; + const body = JSON.stringify({ + model: "gpt-test", + input: "same turn", + stream: true, + metadata: { thread: "thread-test" }, + }); + const send = (bearer = token, path = "/responses") => + HttpClientRequest.post(`${bridge.baseUrl}${path}`).pipe( + HttpClientRequest.bearerToken(bearer), + HttpClientRequest.setHeader("x-client-request-id", "same-request"), + HttpClientRequest.bodyText(body, "application/json"), + HttpClient.execute, + Effect.flatMap((response) => + Effect.map(response.text, (text) => ({ status: response.status, text })), + ), + Effect.scoped, + Effect.provide(FetchHttpClient.layer), + ); + return { + renewalEntered, + renewalRelease, + send, + tokens, + register, + auth, + credentials, + requests, + refreshTokens, + body, + expire: () => save({ ...credentials, expiresAt: 0 }), + }; +}); +const provide = (effect: Effect.Effect) => + effect.pipe( + Effect.scoped, + Effect.provide(ServerSecretStore.layer), + Effect.provideService(ServerEnvironment.ServerEnvironmentIdentity, { + getEnvironmentId: Effect.succeed(EnvironmentId.make("00000000-0000-4000-8000-000000000001")), + }), + Effect.provide( + ServerConfig.layerTest(process.cwd(), { prefix: "managed-responses-" }).pipe( + Layer.provideMerge(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer)), + ), + ), + ); + +it.live("keeps concurrent first registrations valid and reuses the matching grant", () => + provide( + Effect.gen(function* () { + const h = yield* makeHarness({ concurrentRegistration: true }); + const results = yield* Effect.forEach(h.tokens, (token) => h.send(token)); + assert.deepEqual( + results.map((result) => result.status), + [200, 200], + ); + assert.isTrue(h.tokens[0] === h.tokens[1]); + assert.isTrue((yield* h.register) === h.tokens[0]); + assert.equal(h.requests.length, 3); + assert.deepEqual(h.refreshTokens, ["single-use-refresh"]); + }), + ), +); + +it.live.each([false, true])( + "renews locally unexpired tokens and replays the same request once (concurrent=%s)", + (concurrent) => + provide( + Effect.gen(function* () { + const h = yield* makeHarness({ concurrent }); + const count = concurrent ? 2 : 1; + const results = yield* Effect.all( + Array.from({ length: count }, () => h.send()), + { concurrency: "unbounded" }, + ); + assert.deepEqual( + results, + Array.from({ length: count }, () => ({ status: 200, text: 'data: {"output":"OK"}\n\n' })), + ); + assert.deepEqual(h.refreshTokens, ["single-use-refresh"]); + assert.equal(h.requests.length, count * 2); + for (const request of h.requests) { + assert.equal(request.body, h.body); + assert.equal(request.url, "https://api.openai.com/v1/responses"); + } + assert.equal(h.requests.at(-1)!.token, "Bearer fresh-access"); + const saved = Option.getOrThrow(yield* h.auth.read); + assert.equal(saved.refreshToken, "rotated-refresh"); + assert.equal(saved.subject, h.credentials.subject); + }), + ), +); + +it.live.each([ + { + name: "other 401", + error: { error: { code: "invalid_api_key", message: "stale-access single-use-refresh" } }, + renewals: 0, + attempts: 1, + }, + { name: "second 401", retryStatus: 401, renewals: 1, attempts: 2 }, + { name: "renewal failure", renewalFails: true, renewals: 1, attempts: 1 }, +])("terminates on $name", (options) => + provide( + Effect.gen(function* () { + const h = yield* makeHarness(options); + const result = yield* h.send(); + assert.equal(result.status, 400); + assert.include(result.text, "t3_managed_auth"); + for (const secret of [h.credentials.accessToken, h.credentials.refreshToken]) + assert.notInclude(result.text, secret); + assert.equal(h.refreshTokens.length, options.renewals); + assert.equal(h.requests.length, options.attempts); + }), + ), +); + +it.live.each(["replacement", "earliestRefreshAt"] as const)( + "respects %s while recovering a rejected token", + (constraint) => + provide( + Effect.gen(function* () { + const h = yield* makeHarness({ constraint }); + const result = yield* h.send(); + assert.equal(result.status, 400); + assert.deepEqual(h.refreshTokens, []); + assert.equal(h.requests.length, 1); + for (const secret of [h.credentials.accessToken, h.credentials.refreshToken]) + assert.notInclude(result.text, secret); + if (constraint === "earliestRefreshAt") + assert.isAtMost( + Option.getOrThrow(yield* h.auth.read).expiresAt, + yield* Clock.currentTimeMillis, + ); + }), + ), +); + +it.live("refuses unauthenticated callers and non-Responses destinations", () => + provide( + Effect.gen(function* () { + const h = yield* makeHarness(); + assert.equal((yield* h.send("wrong-token")).status, 401); + assert.equal((yield* h.send(undefined, "/untrusted")).status, 404); + assert.deepEqual(h.requests, []); + assert.deepEqual(h.refreshTokens, []); + }), + ), +); + +it.live("failed proactive renewal is terminal without an upstream Responses request", () => + provide( + Effect.gen(function* () { + const h = yield* makeHarness({ renewalFails: true }); + yield* h.expire(); + const result = yield* h.send(); + assert.equal(result.status, 400); + assert.include(result.text, "t3_managed_auth"); + assert.equal(h.refreshTokens.length, 1); + assert.deepEqual(h.requests, []); + }), + ), +); + +it.live.each([400, 403, 429, 503])("preserves upstream HTTP %s after renewal", (retryStatus) => + provide( + Effect.gen(function* () { + const h = yield* makeHarness({ retryStatus }); + const result = yield* h.send(); + assert.equal(result.status, retryStatus); + assert.deepEqual(JSON.parse(result.text), expired); + assert.equal(h.requests.length, 2); + assert.equal(h.refreshTokens.length, 1); + }), + ), +); + +it.live("persists a single-use renewal after the Responses caller disconnects", () => + provide( + Effect.gen(function* () { + const h = yield* makeHarness({ pauseRenewal: true }); + const caller = yield* h.send().pipe(Effect.forkChild); + yield* Deferred.await(h.renewalEntered); + yield* Fiber.interrupt(caller); + yield* Deferred.succeed(h.renewalRelease, undefined); + const saved = yield* h.auth.access; + assert.equal(saved.refreshToken, "rotated-refresh"); + assert.deepEqual(h.refreshTokens, ["single-use-refresh"]); + }), + ), +); diff --git a/apps/server/src/provider/CodexManagedResponses.ts b/apps/server/src/provider/CodexManagedResponses.ts new file mode 100644 index 0000000000000..31112584e1e92 --- /dev/null +++ b/apps/server/src/provider/CodexManagedResponses.ts @@ -0,0 +1,193 @@ +// @effect-diagnostics nodeBuiltinImport:off - A server-owned loopback bridge keeps OAuth recovery at the HTTP boundary. +import * as NodeHttp from "node:http"; +import * as SecureRandom from "effect/crypto/SecureRandom"; +import * as Hex from "effect/encoding/Hex"; +import * as NodeHttpServer from "@effect/platform-node/NodeHttpServer"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as Fiber from "effect/Fiber"; +import * as FiberSet from "effect/FiberSet"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Schema from "effect/Schema"; +import * as Scope from "effect/Scope"; +import { HttpClient, HttpClientRequest, HttpServerRequest, HttpServerResponse } from "effect/http"; +import { type ProviderInstanceId, type ProviderSetupError } from "@t3tools/contracts"; +import { makeCodexChatGptAuth, type CodexChatGptCredentials } from "./CodexChatGptAuth.ts"; + +const ExpiredToken = Schema.Struct({ + error: Schema.Struct({ code: Schema.Literal("token_expired") }), +}); +const decodeExpiredToken = Schema.decodeUnknownOption(Schema.fromJsonString(ExpiredToken)); +const droppedHeaders = new Set([ + "authorization", + "host", + "connection", + "keep-alive", + "proxy-authenticate", + "proxy-authorization", + "te", + "trailer", + "transfer-encoding", + "upgrade", + "content-length", + "accept-encoding", + "cookie", + "set-cookie", +]); +const forwardHeaders = (headers: Readonly>, decoded = false) => { + const connectionHeaders = new Set(headers.connection?.toLowerCase().split(/\s*,\s*/u)); + return Object.fromEntries( + Object.entries(headers).filter( + ([name, value]) => + value !== undefined && + !droppedHeaders.has(name) && + !connectionHeaders.has(name) && + !(decoded && name === "content-encoding"), + ), + ); +}; + +class StartError extends Schema.TaggedError()("ManagedResponsesStartError", { + cause: Schema.Defect(), +}) { + override get message() { + return "Could not start the managed Codex loopback listener."; + } +} +export class CodexManagedResponses extends Context.Service< + CodexManagedResponses, + { readonly bridge: Effect.Effect>, StartError> } +>()("t3/provider/CodexManagedResponses") {} + +// Codex 0.166 maps HTTP 400 to non-retryable InvalidRequest; 401 enters its stream retry loop. +const terminalAuth = (message: string) => + HttpServerResponse.jsonUnsafe({ error: { code: "t3_managed_auth", message } }, { status: 400 }); + +export const make = Effect.fn("CodexManagedResponses.make")(function* () { + const http = HttpClient.withScope(yield* HttpClient.HttpClient); + const random = yield* SecureRandom.SecureRandom; + const services = + yield* Effect.context>>(); + const renewals = yield* FiberSet.make(); + // A spent refresh token must be saved even if Codex disconnects during the exchange. + yield* Effect.addFinalizer(() => FiberSet.awaitEmpty(renewals)); + const grants = new Map< + string, + { + instanceId: ProviderInstanceId; + profileKey: string; + auth: Effect.Success>; + } + >(); + const register = Effect.fn("CodexManagedResponses.register")(function* (options: { + readonly instanceId: ProviderInstanceId; + readonly profileKey: string; + }) { + for (const [token, grant] of grants) { + if (grant.instanceId === options.instanceId && grant.profileKey === options.profileKey) { + yield* grant.auth.access; + return token; + } + } + const auth = yield* makeCodexChatGptAuth({ + instanceId: options.instanceId, + expectedProfileKey: options.profileKey, + }); + yield* auth.access; + const token = Hex.encode(yield* random.randomBytes(32)); + for (const [existingToken, grant] of grants) { + if (grant.instanceId !== options.instanceId) continue; + if (grant.profileKey === options.profileKey) return existingToken; + grants.delete(existingToken); + } + grants.set(token, { ...options, auth }); + return token; + }, Effect.provideContext(services)); + const forward = Effect.fn("CodexManagedResponses.forward")(function* ( + request: HttpServerRequest.HttpServerRequest, + ) { + const grant = grants.get(request.headers.authorization?.replace(/^Bearer /u, "") ?? ""); + if (!grant) return HttpServerResponse.empty({ status: 401 }); + const path = request.url.split("?", 1)[0] ?? ""; + const inference = + request.method === "POST" && + ["/v1/responses", "/v1/responses/compact", "/v1/responses/input_tokens"].includes(path); + if (!inference && !(request.method === "GET" && path === "/v1/models")) + return HttpServerResponse.empty({ status: 404 }); + // Replay only the request body; successful SSE responses continue streaming with backpressure. + const body = inference ? new Uint8Array(yield* request.arrayBuffer) : undefined; + const access = yield* FiberSet.run(renewals, grant.auth.access); + let credentials = yield* Fiber.join(access); + const send = () => + http.execute( + HttpClientRequest.make(inference ? "POST" : "GET")( + `https://api.openai.com${request.url}`, + ).pipe( + HttpClientRequest.setHeaders(forwardHeaders(request.headers)), + HttpClientRequest.bearerToken(credentials.accessToken), + body === undefined + ? (self) => self + : HttpClientRequest.bodyUint8Array(body, request.headers["content-type"]), + ), + ); + let response = yield* send(); + if (inference && response.status === 401) { + const expired = Option.isSome(decodeExpiredToken(yield* response.text)); + if (!expired) + return terminalAuth("ChatGPT rejected the managed credential (HTTP 401). Sign in again."); + const renewed = yield* FiberSet.run(renewals, grant.auth.renew(credentials.accessToken)).pipe( + Effect.flatMap(Fiber.join), + Effect.option, + ); + if (Option.isNone(renewed)) + return terminalAuth( + "The managed ChatGPT token expired and could not be renewed. Retry or sign in again.", + ); + credentials = renewed.value; + response = yield* send(); + if (response.status === 401) + return terminalAuth( + "ChatGPT rejected the renewed managed credential (HTTP 401). Sign in again.", + ); + } + return HttpServerResponse.stream(response.stream, { + status: response.status, + headers: forwardHeaders(response.headers, true), + ...(response.headers["content-type"] + ? { contentType: response.headers["content-type"] } + : {}), + }); + }); + const server = yield* NodeHttpServer.make(NodeHttp.createServer, { host: "127.0.0.1", port: 0 }); + yield* server.serve( + Effect.flatMap(HttpServerRequest.HttpServerRequest, forward).pipe( + Effect.catchTags({ + ProviderSetupError: (error) => Effect.succeed(terminalAuth(error.detail)), + }), + Effect.orElseSucceed(() => HttpServerResponse.empty({ status: 502 })), + ), + ); + if (server.address._tag === "UnixPathAddress") + return yield* Effect.die("Expected loopback TCP listener"); + return { baseUrl: `http://127.0.0.1:${server.address.port}/v1`, register }; +}); + +// A failed listener only disables managed Codex sessions, never server startup. +export const layer = Layer.effect( + CodexManagedResponses, + Effect.gen(function* () { + const scope = yield* Scope.fork(yield* Effect.scope); + const started = yield* make().pipe( + Scope.provide(scope), + Effect.mapError((cause) => new StartError({ cause })), + Effect.result, + ); + if (started._tag === "Failure") { + yield* Scope.close(scope, Exit.fail(started.failure)); + return CodexManagedResponses.of({ bridge: Effect.fail(started.failure) }); + } + return CodexManagedResponses.of({ bridge: Effect.succeed(started.success) }); + }), +); diff --git a/apps/server/src/provider/CodexManagedRuntime.test.ts b/apps/server/src/provider/CodexManagedRuntime.test.ts index 2c697f8fa8531..5b56e974c8a05 100644 --- a/apps/server/src/provider/CodexManagedRuntime.test.ts +++ b/apps/server/src/provider/CodexManagedRuntime.test.ts @@ -1,13 +1,10 @@ // @effect-diagnostics nodeBuiltinImport:off - checks the host's default Codex directory without writing to it. -import * as NodeURL from "node:url"; -import * as HostProcess from "@t3tools/shared/HostProcess"; -import * as Stream from "effect/Stream"; -import { ChildProcess, ChildProcessSpawner } from "effect/process"; -import * as CodexManagedTokenBroker from "./CodexManagedTokenBroker.ts"; +import * as CodexManagedResponses from "./CodexManagedResponses.ts"; import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, it } from "@effect/vitest"; import { CodexSettings, EnvironmentId, ProviderInstanceId } from "@t3tools/contracts"; +import * as HostProcess from "@t3tools/shared/HostProcess"; import * as Clock from "effect/Clock"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; @@ -26,9 +23,6 @@ import { codexAppServerArgs } from "./codexLaunchArgs.ts"; const encodeJson = Schema.encodeEffect(Schema.fromJsonString(Schema.Unknown)); const decodeSettings = Schema.decodeSync(CodexSettings); -const decodeCommand = Schema.decodeUnknownSync( - Schema.fromJsonString(Schema.Tuple([Schema.String, Schema.Array(Schema.String)])), -); it.live.each( (["managed", "local"] as const).flatMap((source) => (["primary", "additional"] as const).map((account) => ({ source, account })), @@ -144,58 +138,13 @@ it.live.each( const provider = args.find((arg) => arg.startsWith("model_providers.openai_token_sharing="), )!; - assert.include(provider, 'base_url="https://api.openai.com/v1"'); - assert.include(provider, 'model_catalog_url="https://api.openai.com/v1/models"'); - assert.notInclude(provider, "env_key"); - assert.include(provider, "refresh_interval_ms=0"); - // regression: #16132 Codex fetches each token through this helper, so a running - // session picks up renewals instead of keeping the token it launched with. - if (source === "managed" && account === "primary") { - const commandMatch = /auth=\{command=(.*),args=(\[.*\]),timeout_ms=/u.exec(provider); - assert.isNotNull(commandMatch); - const [command, helperArgs] = decodeCommand( - `[${commandMatch![1]},${commandMatch![2]}]`, - ); - const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; - const runHelper = Effect.gen(function* () { - const child = yield* spawner.spawn( - ChildProcess.make(command, helperArgs, { - extendEnv: false, - env: effective.environment, - }), - ); - const [stdout, stderr, code] = yield* Effect.all( - [ - child.stdout.pipe(Stream.decodeText(), Stream.mkString), - child.stderr.pipe(Stream.decodeText(), Stream.mkString), - child.exitCode, - ], - { concurrency: "unbounded" }, - ); - return { stdout, stderr, code: Number(code) }; - }).pipe(Effect.scoped); - assert.deepEqual(yield* runHelper, { - stdout: "dummy-owned-access", - stderr: "", - code: 0, - }); - const saved = Option.getOrThrow(yield* runtime.auth.read); - const expired = yield* encodeJson({ ...saved, expiresAt: 0 }); - yield* store.set(new TextEncoder().encode(expired)); - assert.deepEqual(yield* runHelper, { stdout: "renewed-access", stderr: "", code: 0 }); - assert.equal(renewals, 1); - assert.equal( - Option.getOrThrow(yield* runtime.auth.read).refreshToken, - "renewed-refresh", - ); - assert.equal((yield* runtime.resolve).revision, effective.revision); - const replaced = yield* encodeJson({ ...saved, subject: "replacement-user" }); - yield* store.set(new TextEncoder().encode(replaced)); - const refused = yield* runHelper; - assert.equal(refused.code, 1); - assert.equal(refused.stdout, ""); - assert.isFalse(yield* fs.exists(ambient.T3CODE_HOME)); - } + assert.include(provider, 'base_url="http://127.0.0.1:'); + assert.include(provider, 'model_catalog_url="http://127.0.0.1:'); + assert.include(provider, 'env_key="T3_CODEX_MANAGED_TOKEN"'); + assert.isString(effective.environment.T3_CODEX_MANAGED_TOKEN); + assert.notInclude(provider, "auth={"); + assert.equal((yield* runtime.resolve).revision, effective.revision); + assert.equal(renewals, 0); assert.include(args, "features.api_key_model_discovery=true"); assert.include(args, "features.multi_agent=false"); assert.include(args, "features.multi_agent_v2=false"); @@ -216,11 +165,7 @@ it.live.each( "native-auth-unchanged", ); }).pipe( - Effect.provide(CodexManagedTokenBroker.layer), - Effect.provideService(HostProcess.Arguments, [ - process.execPath, - NodeURL.fileURLToPath(new URL("../bin.ts", import.meta.url)), - ]), + Effect.provide(CodexManagedResponses.layer), Effect.provideService(ServerSecretStore.ServerSecretStore, secrets), Effect.provideService(ServerEnvironment.ServerEnvironmentIdentity, { getEnvironmentId: Effect.succeed( diff --git a/apps/server/src/provider/CodexManagedRuntime.ts b/apps/server/src/provider/CodexManagedRuntime.ts index 3704c919c2791..e59102877191d 100644 --- a/apps/server/src/provider/CodexManagedRuntime.ts +++ b/apps/server/src/provider/CodexManagedRuntime.ts @@ -5,12 +5,9 @@ import * as FileSystem from "effect/FileSystem"; import * as Path from "effect/Path"; import * as Schema from "effect/Schema"; import * as ServerConfig from "../config.ts"; -import * as CodexManagedTokenBroker from "./CodexManagedTokenBroker.ts"; -import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; -import { resolveSelfInvocation, selfInvocationArgs } from "@t3tools/shared/nodeRuntime"; +import * as CodexManagedResponses from "./CodexManagedResponses.ts"; import * as CodexInstallation from "./CodexInstallation.ts"; import { makeCodexChatGptAuth } from "./CodexChatGptAuth.ts"; -import { COMMAND_TIMEOUT_MS } from "./CodexManagedTokenProtocol.ts"; import { materializeCodexShadowHome } from "./Drivers/CodexHomeLayout.ts"; export interface CodexEffectiveRuntime { @@ -19,11 +16,11 @@ export interface CodexEffectiveRuntime { readonly revision: string; } const decodeSettings = Schema.decodeSync(CodexSettings); -const managedCodexLaunchArgs = (command: string, args: ReadonlyArray) => +const managedCodexLaunchArgs = (baseUrl: string) => [ 'model_provider="openai_token_sharing"', // Codex persists this provider ID in rollouts and needs it unchanged to resume. - `model_providers.openai_token_sharing={name="OpenAI Token Sharing",base_url="https://api.openai.com/v1",model_catalog_url="https://api.openai.com/v1/models",wire_api="responses",requires_openai_auth=false,supports_websockets=false,auth={command=${JSON.stringify(command)},args=${JSON.stringify(args)},timeout_ms=${COMMAND_TIMEOUT_MS},refresh_interval_ms=0}}`, + `model_providers.openai_token_sharing={name="OpenAI Token Sharing",base_url=${JSON.stringify(baseUrl)},model_catalog_url=${JSON.stringify(`${baseUrl}/models`)},wire_api="responses",requires_openai_auth=false,supports_websockets=false,env_key="T3_CODEX_MANAGED_TOKEN"}`, "features.api_key_model_discovery=true", "features.multi_agent=false", "features.multi_agent_v2=false", @@ -41,10 +38,7 @@ export const makeCodexManagedRuntime = Effect.fn("makeCodexManagedRuntime")(func const config = yield* ServerConfig.ServerConfig; const fs = yield* FileSystem.FileSystem; const path = yield* Path.Path; - const broker = yield* CodexManagedTokenBroker.CodexManagedTokenBroker; - const environmentIdentity = yield* ServerEnvironment.ServerEnvironmentIdentity; - const environmentId = yield* environmentIdentity.getEnvironmentId; - const invocation = yield* resolveSelfInvocation(); + const managedResponses = yield* CodexManagedResponses.CodexManagedResponses; const auth = yield* makeCodexChatGptAuth({ instanceId: options.instanceId, defaultReturnUrl: new URL( @@ -59,15 +53,14 @@ export const makeCodexManagedRuntime = Effect.fn("makeCodexManagedRuntime")(func ); const homePath = homeLayout.effectiveHomePath ?? homeLayout.sharedHomePath; const resolve = Effect.gen(function* () { - // Establish renewal availability before auth.access can consume a refresh token. - const endpoint = yield* broker.endpoint.pipe( + // Establish the HTTP boundary before auth.access can consume a refresh token. + const responses = yield* managedResponses.bridge.pipe( Effect.mapError( (cause) => new ProviderSetupError({ instanceId: options.instanceId, operation: "runtime", - detail: - "Could not prepare the private managed Codex token listener. Restart T3 and retry.", + detail: "Could not prepare the private managed Codex listener. Restart T3 and retry.", cause, }), ), @@ -84,13 +77,7 @@ export const makeCodexManagedRuntime = Effect.fn("makeCodexManagedRuntime")(func ); const credentials = yield* auth.access; const profileKey = yield* auth.profileKey(credentials); - const authArgs = selfInvocationArgs(invocation, [ - "__managed-chatgpt-token", - endpoint, - options.instanceId, - environmentId, - profileKey, - ]); + const token = yield* responses.register({ instanceId: options.instanceId, profileKey }); yield* materializeCodexShadowHome(homeLayout).pipe( Effect.provideService(FileSystem.FileSystem, fs), Effect.provideService(Path.Path, path), @@ -117,6 +104,7 @@ export const makeCodexManagedRuntime = Effect.fn("makeCodexManagedRuntime")(func const environment: NodeJS.ProcessEnv = { ...options.environment, CODEX_HOME: homePath, + T3_CODEX_MANAGED_TOKEN: token, }; delete environment.ACCESS_TOKEN; delete environment.T3CODE_CODEX_LAUNCH_ARGS; @@ -128,7 +116,7 @@ export const makeCodexManagedRuntime = Effect.fn("makeCodexManagedRuntime")(func setupMode: "managed", binaryPath: executable.executablePath, homePath, - launchArgs: managedCodexLaunchArgs(invocation.command, authArgs), + launchArgs: managedCodexLaunchArgs(responses.baseUrl), }), environment, revision: profileKey, diff --git a/apps/server/src/provider/CodexManagedTokenBroker.ts b/apps/server/src/provider/CodexManagedTokenBroker.ts deleted file mode 100644 index 4a1e861827142..0000000000000 --- a/apps/server/src/provider/CodexManagedTokenBroker.ts +++ /dev/null @@ -1,225 +0,0 @@ -// @effect-diagnostics nodeBuiltinImport:off - Server-owned private pipe/Unix socket, never a public listener. -import * as NodeFSP from "node:fs/promises"; -import * as NodeNet from "node:net"; -import * as NodeOS from "node:os"; -import * as NodePath from "node:path"; -import * as NodeCrypto from "node:crypto"; -import { EnvironmentId, ProviderInstanceId } from "@t3tools/contracts"; -import * as HostProcess from "@t3tools/shared/HostProcess"; -import * as Context from "effect/Context"; -import * as Effect from "effect/Effect"; -import * as FiberSet from "effect/FiberSet"; -import * as Layer from "effect/Layer"; -import * as Exit from "effect/Exit"; -import * as Scope from "effect/Scope"; -import * as Schema from "effect/Schema"; -import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; -import { makeCodexChatGptAuth } from "./CodexChatGptAuth.ts"; -import * as Protocol from "./CodexManagedTokenProtocol.ts"; - -const MAX_CONNECTIONS = 32; -const MAX_JOBS = 16; -// Stay below macOS's 104-byte and Linux's 108-byte sockaddr_un limits, including NUL. -const UNIX_SOCKET_PATH_BYTES = 100; -const encodeResponse = Schema.encodeSync(Schema.fromJsonString(Protocol.Response)); -class BrokerStartError extends Schema.TaggedError()( - "ManagedTokenBrokerStartError", - { - cause: Schema.Defect(), - }, -) { - override get message() { - return "Could not start the private managed token listener."; - } -} -export class CodexManagedTokenBroker extends Context.Service< - CodexManagedTokenBroker, - { - readonly endpoint: Effect.Effect; - readonly awaitIdle: Effect.Effect; - } ->()("t3/provider/CodexManagedTokenBroker") {} - -const io = (f: () => Promise) => - Effect.tryPromise({ try: f, catch: (cause) => new BrokerStartError({ cause }) }); -const acquireListener = Effect.gen(function* () { - const identity = yield* ServerEnvironment.ServerEnvironmentIdentity; - const environmentId = yield* identity.getEnvironmentId; - const platform = yield* HostProcess.Platform; - const uid = yield* HostProcess.UserId; - const owned = yield* Effect.acquireRelease( - io(async () => { - if (platform === "win32") - return { endpoint: `\\\\.\\pipe\\t3-token-${NodeCrypto.randomUUID()}` }; - if (uid === undefined) throw new Error("Missing IPC owner"); - const acquireDirectory = async (candidate: string) => { - const parent = await NodeFSP.realpath(candidate); - const stat = await NodeFSP.lstat(parent); - if ( - !stat.isDirectory() || - stat.isSymbolicLink() || - !( - (stat.uid === uid && (stat.mode & 0o022) === 0) || - (stat.uid === 0 && ((stat.mode & 0o022) === 0 || (stat.mode & 0o1000) !== 0)) - ) - ) - throw new Error("Untrusted IPC parent"); - const directory = await NodeFSP.mkdtemp(NodePath.join(parent, "t3-token-")); - try { - await NodeFSP.chmod(directory, 0o700); - const created = await NodeFSP.lstat(directory); - if ( - !created.isDirectory() || - created.isSymbolicLink() || - created.uid !== uid || - (created.mode & 0o777) !== 0o700 - ) - throw new Error("Unprotected IPC directory"); - const endpoint = NodePath.join(directory, "s"); - if (Buffer.byteLength(endpoint) > UNIX_SOCKET_PATH_BYTES) - throw new Error("IPC path too long"); - return { endpoint, directory, created }; - } catch (error) { - await NodeFSP.rmdir(directory); - throw error; - } - }; - return acquireDirectory(NodeOS.tmpdir()).catch(() => acquireDirectory("/tmp")); - }), - (owned) => - Effect.promise(async () => { - if (!("directory" in owned)) return; - // Never recursively remove a replaced directory or another process's files. - const current = await NodeFSP.lstat(owned.directory).catch(() => undefined); - if ( - current && - current.dev === owned.created.dev && - current.ino === owned.created.ino && - !current.isSymbolicLink() - ) - await NodeFSP.rmdir(owned.directory).catch(() => undefined); - }), - ); - const jobs = yield* FiberSet.make(); - const runJob = - yield* FiberSet.runtime(jobs)< - Exclude< - Effect.Services>, - import("effect/Scope").Scope - > - >(); - const connections = yield* FiberSet.make(); - const runConnection = yield* FiberSet.runtime(connections)(); - const sockets = new Set(); - let closing = false; - let admitted = 0; - // Keep branded validation here so the helper need not load the contracts graph. - const decode = Schema.decodeUnknownEffect( - Schema.fromJsonString( - Schema.Struct({ - ...Protocol.Request.fields, - instanceId: ProviderInstanceId, - environmentId: EnvironmentId, - }), - ), - { - onExcessProperty: "error", - }, - ); - const server = NodeNet.createServer((socket) => { - socket.on("error", () => {}); - if (closing || sockets.size >= MAX_CONNECTIONS) { - socket.destroy(); - return; - } - sockets.add(socket); - socket.once("close", () => sockets.delete(socket)); - let transferred = false; - runConnection( - Effect.gen(function* () { - // Windows pipes permit read-only peers by default; send no bytes until a complete request. - const request = yield* Protocol.readFrame( - socket, - Protocol.REQUEST_BYTES, - Protocol.SOCKET_IDLE_MS, - ).pipe(Effect.flatMap(decode)); - if (closing || admitted >= MAX_JOBS || request.environmentId !== environmentId) return; - admitted++; - transferred = true; - runJob( - Effect.scoped( - Effect.gen(function* () { - const auth = yield* makeCodexChatGptAuth({ - instanceId: request.instanceId, - expectedProfileKey: request.profileKey, - }); - // Retire abandoned permit waits; access saves a started rotation before honoring interruption. - const credentials = yield* auth.access.pipe( - Effect.timeout(Protocol.COMMAND_TIMEOUT_MS), - ); - yield* Protocol.writeFrame( - socket, - encodeResponse({ token: credentials.accessToken }), - Protocol.RESPONSE_BYTES, - ); - }), - ).pipe( - Effect.ignoreCause, - Effect.ensuring( - Effect.sync(() => { - admitted--; - socket.destroy(); - }), - ), - ), - ); - }).pipe( - Effect.ignoreCause, - Effect.ensuring( - Effect.sync(() => { - if (!transferred) socket.destroy(); - }), - ), - ), - ); - }); - server.on("error", () => {}); - // Drain accepted rotations before the FiberSet finalizers interrupt their fibers. - yield* Effect.addFinalizer(() => - Effect.gen(function* () { - closing = true; - for (const socket of sockets) socket.destroy(); - yield* Effect.promise(() => new Promise((resolve) => server.close(() => resolve()))); - yield* FiberSet.awaitEmpty(connections); - yield* FiberSet.awaitEmpty(jobs); - }), - ); - yield* Effect.callback((resume) => { - const onError = (cause: Error) => resume(Effect.fail(new BrokerStartError({ cause }))); - server.once("error", onError); - try { - server.listen(owned.endpoint, () => resume(Effect.void)); - } catch (cause) { - resume(Effect.fail(new BrokerStartError({ cause }))); - } - return Effect.sync(() => server.off("error", onError)); - }); - if (platform !== "win32") yield* io(() => NodeFSP.chmod(owned.endpoint, 0o600)); - return { endpoint: owned.endpoint, awaitIdle: FiberSet.awaitEmpty(jobs) }; -}); -const make = Effect.gen(function* () { - const scope = yield* Scope.fork(yield* Effect.scope); - const acquired = yield* acquireListener.pipe(Scope.provide(scope), Effect.result); - if (acquired._tag === "Failure") { - yield* Scope.close(scope, Exit.fail(acquired.failure)); - return CodexManagedTokenBroker.of({ - endpoint: Effect.fail(acquired.failure), - awaitIdle: Effect.void, - }); - } - return CodexManagedTokenBroker.of({ - endpoint: Effect.succeed(acquired.success.endpoint), - awaitIdle: acquired.success.awaitIdle, - }); -}); -export const layer = Layer.effect(CodexManagedTokenBroker, make); diff --git a/apps/server/src/provider/CodexManagedTokenProtocol.ts b/apps/server/src/provider/CodexManagedTokenProtocol.ts deleted file mode 100644 index d5f9e610d2188..0000000000000 --- a/apps/server/src/provider/CodexManagedTokenProtocol.ts +++ /dev/null @@ -1,72 +0,0 @@ -// @effect-diagnostics nodeBuiltinImport:off - Private local IPC framing, shared by server and lean CLI. -import type * as NodeNet from "node:net"; -import * as Effect from "effect/Effect"; -import * as Schema from "effect/Schema"; - -export const Request = Schema.Struct({ - version: Schema.Literal(1), - instanceId: Schema.NonEmptyString, - environmentId: Schema.NonEmptyString, - profileKey: Schema.String.check(Schema.isPattern(/^[a-f0-9]{64}$/u)), -}); -export const Response = Schema.Struct({ token: Schema.NonEmptyString }); -// Selectors are small; tokens can be larger JWTs. Never truncate either frame -// into something that could be accepted. -export const REQUEST_BYTES = 1024; -export const RESPONSE_BYTES = 64 * 1024; -// Idle/incomplete peers and blocked response consumers must not retain sockets. -export const SOCKET_IDLE_MS = 5_000; -// Existing lock retry (40s), OAuth (20s), plus startup/store headroom (20s). -export const COMMAND_TIMEOUT_MS = 80_000; -class ProtocolError extends Schema.TaggedError()("ManagedTokenProtocolError", {}) { - override get message(): string { - return "The managed token frame was invalid or incomplete."; - } -} - -export const readFrame = (socket: NodeNet.Socket, maxBytes: number, timeoutMs: number) => - Effect.callback((resume) => { - let bytes = Buffer.alloc(0); - const fail = () => resume(Effect.fail(new ProtocolError())); - const onData = (chunk: Buffer) => { - if (bytes.length + chunk.length > maxBytes) return fail(); - bytes = Buffer.concat([bytes, chunk]); - const end = bytes.indexOf(10); - if (end < 0) return; - if (end !== bytes.length - 1) return fail(); - resume(Effect.succeed(bytes.subarray(0, end).toString("utf8"))); - }; - socket.on("data", onData); - socket.once("error", fail); - socket.once("close", fail); - socket.once("end", fail); - return Effect.sync(() => { - socket.off("data", onData); - socket.off("error", fail); - socket.off("close", fail); - socket.off("end", fail); - }); - }).pipe( - Effect.timeoutOrElse({ duration: timeoutMs, orElse: () => Effect.fail(new ProtocolError()) }), - ); - -export const writeFrame = (socket: NodeNet.Socket, frame: string, maxBytes: number) => - Effect.callback((resume) => { - if (socket.destroyed || Buffer.byteLength(frame) + 1 > maxBytes) { - resume(Effect.fail(new ProtocolError())); - return; - } - const fail = () => resume(Effect.fail(new ProtocolError())); - socket.once("error", fail); - socket.once("close", fail); - socket.end(`${frame}\n`, () => resume(Effect.void)); - return Effect.sync(() => { - socket.off("error", fail); - socket.off("close", fail); - }); - }).pipe( - Effect.timeoutOrElse({ - duration: SOCKET_IDLE_MS, - orElse: () => Effect.fail(new ProtocolError()), - }), - ); diff --git a/apps/server/src/provider/Drivers/CodexDriver.test.ts b/apps/server/src/provider/Drivers/CodexDriver.test.ts index 597be85623115..6063eab11811d 100644 --- a/apps/server/src/provider/Drivers/CodexDriver.test.ts +++ b/apps/server/src/provider/Drivers/CodexDriver.test.ts @@ -1,5 +1,5 @@ import * as CodexInstallation from "../CodexInstallation.ts"; -import * as CodexManagedTokenBroker from "../CodexManagedTokenBroker.ts"; +import * as CodexManagedResponses from "../CodexManagedResponses.ts"; import * as ServerSecretStore from "../../auth/ServerSecretStore.ts"; import * as ServerEnvironment from "../../environment/ServerEnvironment.ts"; // @effect-diagnostics nodeBuiltinImport:off @@ -63,9 +63,11 @@ const layerDeps = ServerConfig.layerTest(process.cwd(), { ), Layer.provideMerge(Layer.mock(ServerSecretStore.ServerSecretStore)({})), Layer.provideMerge( - Layer.mock(CodexManagedTokenBroker.CodexManagedTokenBroker)({ - endpoint: Effect.succeed("unused-private-pipe"), - awaitIdle: Effect.void, + Layer.mock(CodexManagedResponses.CodexManagedResponses)({ + bridge: Effect.succeed({ + baseUrl: "http://127.0.0.1:1234/v1", + register: () => Effect.succeed("dummy-local-capability"), + }), }), ), Layer.provideMerge( @@ -285,7 +287,7 @@ it.layer(layerTest)("CodexDriver", (it) => { expect(launch.settings.binaryPath).toBe(executable.executablePath); expect(launch.settings.launchArgs).toContain("openai_token_sharing"); expect(launch.environment.ACCESS_TOKEN).toBeUndefined(); - expect(launch.settings.launchArgs).toContain("__managed-chatgpt-token"); + expect(launch.settings.launchArgs).toContain("T3_CODEX_MANAGED_TOKEN"); expect(launch.settings.launchArgs).not.toContain("dummy-owned-access"); expect(launch.environment.OPENAI_API_KEY).toBeUndefined(); expect(launch.environment.CODEX_HOME).toBe(launch.settings.homePath); diff --git a/apps/server/src/provider/Drivers/CodexDriver.ts b/apps/server/src/provider/Drivers/CodexDriver.ts index 50fbbc7024b99..b592311c775b2 100644 --- a/apps/server/src/provider/Drivers/CodexDriver.ts +++ b/apps/server/src/provider/Drivers/CodexDriver.ts @@ -58,7 +58,7 @@ import { import { makeManagedCodexProvider } from "./CodexManagedProvider.ts"; import { codexUsageReader } from "./codexUsage.ts"; import * as CodexInstallation from "../CodexInstallation.ts"; -import * as CodexManagedTokenBroker from "../CodexManagedTokenBroker.ts"; +import * as CodexManagedResponses from "../CodexManagedResponses.ts"; import * as ServerSecretStore from "../../auth/ServerSecretStore.ts"; import * as ServerEnvironment from "../../environment/ServerEnvironment.ts"; import * as HostProcess from "@t3tools/shared/HostProcess"; @@ -143,7 +143,7 @@ export type CodexDriverEnv = | ServerSecretStore.ServerSecretStore | ServerEnvironment.ServerEnvironmentIdentity | CodexInstallation.CodexInstallation - | CodexManagedTokenBroker.CodexManagedTokenBroker; + | CodexManagedResponses.CodexManagedResponses; export const CodexDriver: ProviderDriver< CodexSettings, diff --git a/apps/server/src/provider/Drivers/CodexManagedProvider.ts b/apps/server/src/provider/Drivers/CodexManagedProvider.ts index f182245f8840d..970f164cb7b66 100644 --- a/apps/server/src/provider/Drivers/CodexManagedProvider.ts +++ b/apps/server/src/provider/Drivers/CodexManagedProvider.ts @@ -240,7 +240,7 @@ export const makeManagedCodexProvider = Effect.fn("makeManagedCodexProvider")(fu Effect.forkScoped, ); const resolveRuntime = runtime.auth.controller.withAccess!(runtime.resolve); - // Launch settings bind the refresh command to the signed-in profile. The registry + // Launch settings bind the loopback credential to the signed-in profile. The registry // already wraps openSession in withAccess, so resolve without re-entering it. const models = snapshot.getSnapshot.pipe(Effect.map((value) => value.models)); const orchestrationAdapter = yield* createCodexAdapterV2(input, { diff --git a/apps/server/src/provider/ProviderInstanceRegistry.test.ts b/apps/server/src/provider/ProviderInstanceRegistry.test.ts index ab35fe4d9ae59..e539e811a3b0d 100644 --- a/apps/server/src/provider/ProviderInstanceRegistry.test.ts +++ b/apps/server/src/provider/ProviderInstanceRegistry.test.ts @@ -1,6 +1,6 @@ import * as ServerSecretStore from "../auth/ServerSecretStore.ts"; import * as CodexInstallation from "./CodexInstallation.ts"; -import * as CodexManagedTokenBroker from "./CodexManagedTokenBroker.ts"; +import * as CodexManagedResponses from "./CodexManagedResponses.ts"; import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; import * as ClaudeSdk from "@anthropic-ai/claude-agent-sdk"; import { describe, expect, it } from "@effect/vitest"; @@ -174,11 +174,7 @@ describe("ProviderInstanceRegistry — multi-instance codex slice", () => { const layerBaseDeps = ServerConfig.layerTest(process.cwd(), { prefix: "provider-instance-registry-test", }).pipe( - Layer.provideMerge( - Layer.mock(CodexManagedTokenBroker.CodexManagedTokenBroker)({ - endpoint: Effect.succeed("unused-private-pipe"), - }), - ), + Layer.provideMerge(Layer.mock(CodexManagedResponses.CodexManagedResponses)({})), Layer.provideMerge(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer)), Layer.provideMerge( Layer.mock(CodexInstallation.CodexInstallation)({ diff --git a/apps/server/src/provider/ProviderRegistry.test.ts b/apps/server/src/provider/ProviderRegistry.test.ts index c0f60b9d77e66..6f5837386855b 100644 --- a/apps/server/src/provider/ProviderRegistry.test.ts +++ b/apps/server/src/provider/ProviderRegistry.test.ts @@ -2,7 +2,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 CodexInstallation from "./CodexInstallation.ts"; -import * as CodexManagedTokenBroker from "./CodexManagedTokenBroker.ts"; +import * as CodexManagedResponses from "./CodexManagedResponses.ts"; import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; import * as NodeServices from "@effect/platform-node/NodeServices"; @@ -424,9 +424,7 @@ const layerTestNodeServices = Layer.mergeAll( ProviderLatestVersions.layer, NodeServices.layer, NodeCrypto.layer, - Layer.mock(CodexManagedTokenBroker.CodexManagedTokenBroker)({ - endpoint: Effect.succeed("unused-private-pipe"), - }), + Layer.mock(CodexManagedResponses.CodexManagedResponses)({}), Layer.mock(CodexInstallation.CodexInstallation)({ managedDirectory: "unused-managed-installation", }), diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 2510c86fd0f8b..a5430685e632c 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -150,7 +150,7 @@ import * as RemoteOpenTargets from "./environment/RemoteOpenTargets.ts"; import * as AuthHttp from "./auth/http.ts"; import * as ReplayMarkers from "./auth/replayMarkers.ts"; import * as ServerSecretStore from "./auth/ServerSecretStore.ts"; -import * as CodexManagedTokenBroker from "./provider/CodexManagedTokenBroker.ts"; +import * as CodexManagedResponses from "./provider/CodexManagedResponses.ts"; import * as WebhookRoute from "./scheduledTasks/webhookRoute.ts"; import * as RelayDeliveryProof from "./scheduledTasks/RelayDeliveryProof.ts"; import * as HeldHooksWaker from "./relay/HeldHooksWaker.ts"; @@ -755,7 +755,7 @@ const layerRuntimeCoreDependencies = layerRuntimeCoreDependenciesBase.pipe( Layer.provideMerge(ProjectEnrichmentService.layer), Layer.provideMerge(Layer.mergeAll(NativeAppIconResolver.layer, layerProjectFaviconResolver)), Layer.provideMerge(layerRepositoryIdentityResolver), - Layer.provideMerge(CodexManagedTokenBroker.layer), + Layer.provideMerge(CodexManagedResponses.layer), Layer.provideMerge(layerServerEnvironment), Layer.provideMerge(layerAuth), Layer.provideMerge(Layer.mergeAll(AssetSigner.layer, EnvironmentKeys.layer)),