diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts index 6a6cec5b1e61..a8cf90c241dd 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts @@ -43,27 +43,66 @@ describe("CodexSessionRuntimeIdentifierGenerationError", () => { }); }); -function makeThreadOpenResponse( - threadId: string, -): CodexRpc.ClientRequestResponsesByMethod["thread/start"] { +/** + * Raw `thread/start` / `thread/resume` payload as Codex puts it on the wire. + * `items` accepts arbitrary history entries so tests can replay shapes newer + * than the generated bindings. + */ +function makeThreadOpenResponse(threadId: string, items: ReadonlyArray = []): unknown { return { cwd: "/tmp/project", model: "gpt-5.3-codex", modelProvider: "openai", approvalPolicy: "never", approvalsReviewer: "user", - sandbox: { type: "danger-full-access" }, + sandbox: { type: "dangerFullAccess" }, thread: { id: threadId, - createdAt: "2026-04-18T00:00:00.000Z", - source: { session: "cli" }, - turns: [], - status: { - state: "idle", - activeFlags: [], - }, + cliVersion: "0.150.0", + createdAt: 0, + updatedAt: 0, + cwd: "/tmp/project", + ephemeral: false, + modelProvider: "openai", + preview: "", + sessionId: "session-1", + source: "cli", + status: { type: "idle" }, + turns: items.length === 0 ? [] : [{ id: "turn-1", status: "completed", items }], }, - } as unknown as CodexRpc.ClientRequestResponsesByMethod["thread/start"]; + }; +} + +/** + * Mirrors the real client: params keep their generated types, and the raw + * payload is decoded with whichever response schema the caller supplied, + * failing exactly as the client would. + */ +function makeThreadOpenClient( + respond: ( + method: "thread/start" | "thread/resume", + ) => Effect.Effect, +) { + return { + request: ( + method: M, + _payload: CodexRpc.ClientRequestParamsByMethod[M], + responseSchema: Schema.Codec, + ) => + respond(method).pipe( + Effect.flatMap((raw) => + Schema.decodeUnknownEffect(responseSchema)(raw).pipe( + Effect.mapError((cause) => + CodexErrors.CodexAppServerRequestError.invalidPayload( + method, + "decode-payload", + cause, + ), + ), + ), + ), + ), + }; } describe("buildTurnStartParams", () => { @@ -752,6 +791,20 @@ describe("isRecoverableThreadResumeError", () => { ); }); + it("matches responses this build cannot decode", () => { + NodeAssert.equal( + isRecoverableThreadResumeError( + new CodexErrors.CodexAppServerRequestError({ + code: -32602, + errorMessage: "Invalid payload for method 'thread/resume' during 'decode-payload'", + method: "thread/resume", + operation: "decode-payload", + }), + ), + true, + ); + }); + it("ignores unrelated missing-resource errors that do not mention threads", () => { NodeAssert.equal( isRecoverableThreadResumeError( @@ -775,27 +828,56 @@ describe("isRecoverableThreadResumeError", () => { }); describe("openCodexThread", () => { - it.effect("falls back to thread/start when resume fails recoverably", () => + it.effect("resumes a thread whose history uses a newer protocol variant", () => Effect.gen(function* () { - const calls: Array<{ method: "thread/start" | "thread/resume"; payload: unknown }> = []; - const started = makeThreadOpenResponse("fresh-thread"); - const client = { - request: ( - method: M, - payload: CodexRpc.ClientRequestParamsByMethod[M], - ) => { - calls.push({ method, payload }); - if (method === "thread/resume") { - return Effect.fail( - new CodexErrors.CodexAppServerRequestError({ - code: -32603, - errorMessage: "thread not found", - }), - ); - } - return Effect.succeed(started as CodexRpc.ClientRequestResponsesByMethod[M]); + // Opening a session must not depend on history this build cannot name + // (#8322). The kind is deliberately fictional so the test keeps + // exercising drift after the bindings learn today's values. + const resumed = makeThreadOpenResponse("resumed-thread", [ + { + id: "item-18", + type: "subAgentActivity", + agentPath: "/root/child", + agentThreadId: "child-thread", + kind: "escalated", }, - }; + ]); + const calls: Array = []; + const client = makeThreadOpenClient((method) => { + calls.push(method); + return Effect.succeed(resumed); + }); + + const opened = yield* openCodexThread({ + client, + threadId: ThreadId.make("thread-1"), + runtimeMode: "full-access", + cwd: "/tmp/project", + requestedModel: "gpt-5.3-codex", + serviceTier: undefined, + resumeThreadId: "resumed-thread", + }); + + NodeAssert.equal(opened.thread.id, "resumed-thread"); + NodeAssert.deepStrictEqual(calls, ["thread/resume"]); + }), + ); + + it.effect("falls back to thread/start when resume fails recoverably", () => + Effect.gen(function* () { + const calls: Array = []; + const client = makeThreadOpenClient((method) => { + calls.push(method); + if (method === "thread/resume") { + return Effect.fail( + new CodexErrors.CodexAppServerRequestError({ + code: -32603, + errorMessage: "thread not found", + }), + ); + } + return Effect.succeed(makeThreadOpenResponse("fresh-thread")); + }); const opened = yield* openCodexThread({ client, @@ -808,33 +890,48 @@ describe("openCodexThread", () => { }); NodeAssert.equal(opened.thread.id, "fresh-thread"); - NodeAssert.deepStrictEqual( - calls.map((call) => call.method), - ["thread/resume", "thread/start"], - ); + NodeAssert.deepStrictEqual(calls, ["thread/resume", "thread/start"]); + }), + ); + + it.effect("falls back to thread/start when the resume response cannot be decoded", () => + Effect.gen(function* () { + const calls: Array = []; + const client = makeThreadOpenClient((method) => { + calls.push(method); + if (method === "thread/resume") { + return Effect.succeed({ thread: {} }); + } + return Effect.succeed(makeThreadOpenResponse("fresh-thread")); + }); + + const opened = yield* openCodexThread({ + client, + threadId: ThreadId.make("thread-1"), + runtimeMode: "full-access", + cwd: "/tmp/project", + requestedModel: "gpt-5.3-codex", + serviceTier: undefined, + resumeThreadId: "stale-thread", + }); + + NodeAssert.equal(opened.thread.id, "fresh-thread"); + NodeAssert.deepStrictEqual(calls, ["thread/resume", "thread/start"]); }), ); it.effect("propagates non-recoverable resume failures", () => Effect.gen(function* () { - const client = { - request: ( - method: M, - _payload: CodexRpc.ClientRequestParamsByMethod[M], - ) => { - if (method === "thread/resume") { - return Effect.fail( + const client = makeThreadOpenClient((method) => + method === "thread/resume" + ? Effect.fail( new CodexErrors.CodexAppServerRequestError({ code: -32603, errorMessage: "timed out waiting for server", }), - ); - } - return Effect.succeed( - makeThreadOpenResponse("fresh-thread") as CodexRpc.ClientRequestResponsesByMethod[M], - ); - }, - }; + ) + : Effect.succeed(makeThreadOpenResponse("fresh-thread")), + ); const error = yield* openCodexThread({ client, diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index b34067b7fb90..5ccce4198caa 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -661,7 +661,16 @@ function classifyCodexStderrLine(rawLine: string): { readonly message: string } return { message: line }; } +const isCodexAppServerRequestError = Schema.is(CodexErrors.CodexAppServerRequestError); + export function isRecoverableThreadResumeError(error: unknown): boolean { + // A response we cannot decode means Codex resumed the thread but described it + // in a protocol shape this build does not know. Retrying will never help, so + // treat it as recoverable and open a fresh Codex thread instead of leaving + // the T3 thread permanently unusable. + if (isCodexAppServerRequestError(error) && error.operation === "decode-payload") { + return true; + } const message = (error instanceof Error ? error.message : String(error)).toLowerCase(); if (!message.includes("thread")) { return false; @@ -669,9 +678,21 @@ export function isRecoverableThreadResumeError(error: unknown): boolean { return RECOVERABLE_THREAD_RESUME_ERROR_SNIPPETS.some((snippet) => message.includes(snippet)); } -type CodexThreadOpenResponse = - | CodexRpc.ClientRequestResponsesByMethod["thread/start"] - | CodexRpc.ClientRequestResponsesByMethod["thread/resume"]; +/** + * The only parts of a `thread/start` or `thread/resume` response this runtime + * consumes. Codex replays the whole thread history in those responses, and + * decoding it against generated bindings makes opening a session fail whenever + * upstream adds a protocol variant we have not regenerated yet (see #8322). + * Session state is rebuilt from notifications anyway, so read the handful of + * fields we need and let the rest pass through undecoded. + */ +const CodexThreadOpenResponse = Schema.Struct({ + thread: Schema.Struct({ id: Schema.String }), + cwd: Schema.String, + model: Schema.String, +}); + +type CodexThreadOpenResponse = typeof CodexThreadOpenResponse.Type; type CodexThreadOpenMethod = "thread/start" | "thread/resume"; @@ -679,7 +700,8 @@ interface CodexThreadOpenClient { readonly request: ( method: M, payload: CodexRpc.ClientRequestParamsByMethod[M], - ) => Effect.Effect; + responseSchema: typeof CodexThreadOpenResponse, + ) => Effect.Effect; } export const openCodexThread = (input: { @@ -700,14 +722,18 @@ export const openCodexThread = (input: { }); if (resumeThreadId === undefined) { - return input.client.request("thread/start", startParams); + return input.client.request("thread/start", startParams, CodexThreadOpenResponse); } return input.client - .request("thread/resume", { - threadId: resumeThreadId, - ...startParams, - }) + .request( + "thread/resume", + { + threadId: resumeThreadId, + ...startParams, + }, + CodexThreadOpenResponse, + ) .pipe( Effect.catchIf(isRecoverableThreadResumeError, (error) => Effect.logWarning("codex app-server thread resume fell back to fresh start", { @@ -716,7 +742,11 @@ export const openCodexThread = (input: { resumeThreadId, recoverable: true, cause: error, - }).pipe(Effect.andThen(input.client.request("thread/start", startParams))), + }).pipe( + Effect.andThen( + input.client.request("thread/start", startParams, CodexThreadOpenResponse), + ), + ), ), ); }; diff --git a/docs/internals/providers.md b/docs/internals/providers.md index 7d68dc4efe17..e28a757c7dbd 100644 --- a/docs/internals/providers.md +++ b/docs/internals/providers.md @@ -39,6 +39,32 @@ directory to route session and turn operations for a thread, so callers name a t Adding a driver means writing the driver plus adapter and adding it to `BUILT_IN_DRIVERS`. No orchestration, contract, or client change is required for the common case. +## Generated Codex bindings + +`packages/effect-codex-app-server` holds Effect/Schema bindings generated from a pinned +`openai/codex` revision (`scripts/generate.ts`). Codex ships far more often than we regenerate, so +at any moment an installed CLI may describe itself with protocol variants those bindings do not +name — a new enum member, a new item type, a newly required field. + +Two rules keep that drift from breaking sessions: + +- **Decode only what you consume.** `thread/start` and `thread/resume` replay an entire thread + history, and the session runtime needs three fields from it. Pass a narrow response schema as the + third argument to `client.request` rather than accepting the generated one; anything the runtime + does not read cannot then fail the request. +- **Never treat drift as fatal.** A response we cannot decode counts as a recoverable resume error, + so the thread falls back to a fresh Codex session instead of becoming permanently unopenable. + Notifications we cannot decode are dropped with a warning, never silently. + +When a live event carries a value the bindings reject, teach them that one value: the generator's +definition overrides (`Codex0150DefinitionSchemas` in `scripts/generate.ts`) widen a named +definition without moving the pin. Prefer that to a full refresh, which drags in unrelated changes +and can break older CLIs — upstream adds _required_ fields too, so pinning forward breaks anyone +who has not upgraded. + +None of that replaces the two rules. Overrides fix the variants we already know about; the rules +are what keep the ones we do not know about yet from being fatal. + ## Model manifest The model picker's legacy section is driven by `apps/server/src/provider/model-manifest.json`, which diff --git a/packages/effect-codex-app-server/src/client.test.ts b/packages/effect-codex-app-server/src/client.test.ts index 3830c5fc5f6f..c5d11836277f 100644 --- a/packages/effect-codex-app-server/src/client.test.ts +++ b/packages/effect-codex-app-server/src/client.test.ts @@ -1,5 +1,6 @@ import * as Exit from "effect/Exit"; import * as Layer from "effect/Layer"; +import * as Logger from "effect/Logger"; import * as Path from "effect/Path"; import * as Effect from "effect/Effect"; import * as Ref from "effect/Ref"; @@ -9,7 +10,12 @@ import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, it } from "@effect/vitest"; +import * as Schema from "effect/Schema"; + import * as CodexClient from "./client.ts"; +import * as CodexErrors from "./errors.ts"; + +const isRequestError = Schema.is(CodexErrors.CodexAppServerRequestError); const mockPeerPath = Effect.map(Effect.service(Path.Path), (path) => path.join(import.meta.dirname, "../test/fixtures/codex-app-server-mock-peer.ts"), @@ -123,6 +129,91 @@ it.layer(NodeServices.layer)("effect-codex-app-server client", (it) => { ]); }), ); + it.effect("decodes a response with the caller's schema instead of the generated one", () => + Effect.gen(function* () { + const handle = yield* makeHandle(); + const scope = yield* Scope.make(); + const clientLayer = CodexClient.layerChildProcess(handle); + const context = yield* Layer.buildWithScope(clientLayer, scope); + + const result = yield* Effect.gen(function* () { + const client = yield* CodexClient.CodexAppServerClient; + const params = { threadId: "resumed-thread" }; + + const generated = yield* client.request("thread/resume", params).pipe(Effect.flip); + const narrow = yield* client.request( + "thread/resume", + params, + Schema.Struct({ thread: Schema.Struct({ id: Schema.String }) }), + ); + + return { generated, narrow }; + }).pipe(Effect.provide(context), Effect.ensuring(Scope.close(scope, Exit.void))); + + // The generated bindings reject history they do not recognize... + assert.equal(isRequestError(result.generated), true); + assert.equal( + isRequestError(result.generated) ? result.generated.operation : undefined, + "decode-payload", + ); + // ...while a caller that reads only what it uses is unaffected. + assert.equal(result.narrow.thread.id, "resumed-thread"); + }), + ); + + it.effect("drops undecodable notifications, warning once per method", () => { + const warnings: Array = []; + const logger = Logger.make(({ message }) => { + warnings.push(String(message)); + }); + + return Effect.gen(function* () { + const deltas = yield* Ref.make>([]); + const handle = yield* makeHandle({ CODEX_APP_SERVER_TEST_DRIFT: "1" }); + const scope = yield* Scope.make(); + const clientLayer = CodexClient.layerChildProcess(handle); + const context = yield* Layer.buildWithScope(clientLayer, scope); + + yield* Effect.gen(function* () { + const client = yield* CodexClient.CodexAppServerClient; + yield* client.handleServerNotification("item/agentMessage/delta", (payload) => + Ref.update(deltas, (current) => [...current, payload]), + ); + yield* client.request("initialize", { + clientInfo: { + name: "effect-codex-app-server-test", + title: "Effect Codex App Server Test", + version: "0.0.0", + }, + capabilities: { + experimentalApi: true, + optOutNotificationMethods: null, + }, + }); + yield* client.notify("initialized", undefined); + // A round trip on the same stdio proves the peer's notifications have + // all been read before the assertions below. + yield* client.request("account/read", {}); + }).pipe(Effect.provide(context), Effect.ensuring(Scope.close(scope, Exit.void))); + + // The two undecodable deltas never reach the handler... + assert.deepEqual(yield* Ref.get(deltas), [ + { + delta: "Mock server is ready.", + itemId: "item-1", + threadId: "thread-1", + turnId: "turn-1", + }, + ]); + // ...and drift is reported once for the method, not once per message. + assert.deepEqual( + warnings.filter((message) => message.includes("codex app-server notification dropped")) + .length, + 1, + ); + }).pipe(Effect.provide(Logger.layer([logger], { mergeWithExisting: false }))); + }); + it.effect("drains child stderr so large diagnostics cannot block protocol responses", () => Effect.gen(function* () { const handle = yield* makeHandle({ diff --git a/packages/effect-codex-app-server/src/client.ts b/packages/effect-codex-app-server/src/client.ts index c0cb5b1dc23a..79f9c4eb5bf2 100644 --- a/packages/effect-codex-app-server/src/client.ts +++ b/packages/effect-codex-app-server/src/client.ts @@ -39,10 +39,24 @@ export class CodexAppServerClient extends Context.Service< CodexAppServerClient, { readonly raw: CodexAppServerClientRaw; - readonly request: ( - method: M, - payload: CodexRpc.ClientRequestParamsByMethod[M], - ) => Effect.Effect; + /** + * Sends a client request. Params are always encoded with the generated + * schema; the response is decoded with the generated schema unless a + * narrower `responseSchema` is supplied. Pass one when the caller only + * consumes part of a large payload, so an unrecognized protocol variant + * elsewhere in that payload cannot fail the request. + */ + readonly request: { + ( + method: M, + payload: CodexRpc.ClientRequestParamsByMethod[M], + ): Effect.Effect; + ( + method: M, + payload: CodexRpc.ClientRequestParamsByMethod[M], + responseSchema: Schema.Codec, + ): Effect.Effect; + }; readonly notify: ( method: M, payload: CodexRpc.ClientNotificationParamsByMethod[M], @@ -91,6 +105,10 @@ export const make = Effect.fn("effect-codex-app-server/CodexAppServerClient.make ): Effect.fn.Return { const requestHandlers = new Map(); const notificationHandlers = new Map>(); + // Methods already reported as undecodable. Drift is a property of the method, + // not of each message, and high-frequency streams such as agent message + // deltas would otherwise warn once per token. + const warnedUndecodableMethods = new Set(); let unknownRequestHandler: | ((method: string, params: unknown) => Effect.Effect) | undefined; @@ -150,6 +168,21 @@ export const make = Effect.fn("effect-codex-app-server/CodexAppServerClient.make if (schema) { return decodeNotificationPayload(notification.method, schema, notification.params).pipe( + // A notification we cannot decode is dropped rather than failing the + // session, but it is still protocol drift we want to hear about: + // discarding it silently hides the loss of a real lifecycle event. + Effect.tapError((error) => + warnedUndecodableMethods.has(notification.method) + ? Effect.void + : Effect.sync(() => warnedUndecodableMethods.add(notification.method)).pipe( + Effect.andThen( + Effect.logWarning("codex app-server notification dropped", { + method: notification.method, + cause: error, + }), + ), + ), + ), Effect.flatMap((decoded) => Effect.forEach(handlers, (handler) => handler(decoded), { discard: true }), ), @@ -194,19 +227,20 @@ export const make = Effect.fn("effect-codex-app-server/CodexAppServerClient.make onRequest: dispatchRequest, }); - const request = ( + const request = ( method: M, payload: CodexRpc.ClientRequestParamsByMethod[M], - ): Effect.Effect => + responseSchema?: Schema.Codec, + ): Effect.Effect => encodeOptionalPayload(method, getClientRequestParamSchema(method), payload).pipe( Effect.flatMap((encoded) => transport.request(method, encoded)), Effect.flatMap( - ( - raw, - ): Effect.Effect< - CodexRpc.ClientRequestResponsesByMethod[M], - CodexError.CodexAppServerError - > => decodeOptionalPayload(method, getClientRequestResponseSchema(method), raw), + (raw): Effect.Effect => + decodeOptionalPayload( + method, + responseSchema ?? (getClientRequestResponseSchema(method) as Schema.Codec), + raw, + ), ), ); @@ -227,7 +261,7 @@ export const make = Effect.fn("effect-codex-app-server/CodexAppServerClient.make respond: transport.respond, respondError: transport.respondError, }, - request, + request: request as CodexAppServerClient["Service"]["request"], notify, handleServerRequest: (method, handler) => Effect.sync(() => { diff --git a/packages/effect-codex-app-server/test/fixtures/codex-app-server-mock-peer.ts b/packages/effect-codex-app-server/test/fixtures/codex-app-server-mock-peer.ts index 3f2a213d38c7..ded2fe6098bf 100644 --- a/packages/effect-codex-app-server/test/fixtures/codex-app-server-mock-peer.ts +++ b/packages/effect-codex-app-server/test/fixtures/codex-app-server-mock-peer.ts @@ -59,6 +59,22 @@ const handleMethod = (message: Record) => { return; } case "initialized": { + // oxlint-disable-next-line t3code/no-global-process-runtime -- Standalone mock peer process has no Effect runtime. + if (process.env.CODEX_APP_SERVER_TEST_DRIFT === "1") { + // Two notifications the generated bindings cannot decode, standing in + // for a Codex release whose payloads outgrew them. + for (let index = 0; index < 2; index += 1) { + writeMessage({ + method: "item/agentMessage/delta", + params: { + delta: index, + itemId: "item-1", + threadId: "thread-1", + turnId: "turn-1", + }, + }); + } + } writeMessage({ method: "item/agentMessage/delta", params: { @@ -81,6 +97,50 @@ const handleMethod = (message: Record) => { }); return; } + // Replays a resume payload whose history carries a subAgentActivity kind + // the generated bindings do not name, the shape that made resuming a + // thread fail outright before responses could be decoded narrowly. The + // value is deliberately fictional: naming a real one would stop testing + // drift as soon as the bindings caught up with it. + case "thread/resume": { + respond(message.id as number | string, { + cwd: process.cwd(), + model: "gpt-5.3-codex", + modelProvider: "openai", + approvalPolicy: "never", + approvalsReviewer: "user", + sandbox: { type: "dangerFullAccess" }, + thread: { + id: "resumed-thread", + cliVersion: "0.150.0", + createdAt: 0, + updatedAt: 0, + cwd: process.cwd(), + ephemeral: false, + modelProvider: "openai", + preview: "", + sessionId: "session-1", + source: "cli", + status: { type: "idle" }, + turns: [ + { + id: "turn-1", + status: "completed", + items: [ + { + id: "item-18", + type: "subAgentActivity", + agentPath: "/root/child", + agentThreadId: "child-thread", + kind: "escalated", + }, + ], + }, + ], + }, + }); + return; + } case "skills/list": { pendingSkillsListRequestId = message.id as number | string; pendingUserInputRequestId = sendRequest("item/tool/requestUserInput", {