From f20a73a8bed5568f9254a8ed06c43c38f471dcd7 Mon Sep 17 00:00:00 2001 From: d3oxy Date: Thu, 27 Aug 2026 06:39:31 +0530 Subject: [PATCH 1/3] fix(server): Codex threads survive protocol drift on resume Codex CLI 0.150.0 added a fourth subAgentActivity kind ("completed"). The generated app-server bindings are pinned to a July revision that knows three, so decoding a thread/resume response containing that kind failed, and because a schema error is not a recoverable resume error the thread could never be opened again. thread/start and thread/resume now decode only the fields the session runtime consumes, so history this build cannot name no longer fails the request. A response that still cannot be decoded is treated as recoverable and falls back to a fresh Codex thread, and notifications we cannot decode are logged instead of silently dropped. Fixes #8322 --- .../Layers/CodexSessionRuntime.test.ts | 180 +++++++++++++----- .../provider/Layers/CodexSessionRuntime.ts | 54 ++++-- docs/internals/providers.md | 21 ++ .../src/client.test.ts | 37 ++++ .../effect-codex-app-server/src/client.ts | 50 +++-- .../fixtures/codex-app-server-mock-peer.ts | 42 ++++ 6 files changed, 316 insertions(+), 68 deletions(-) diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts index 6a6cec5b1e61..f5c694283c80 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts @@ -6,7 +6,6 @@ import * as Schema from "effect/Schema"; import { describe } from "vite-plus/test"; import { DEFAULT_MODEL, ThreadId } from "@t3tools/contracts"; import * as CodexErrors from "effect-codex-app-server/errors"; -import * as CodexRpc from "effect-codex-app-server/rpc"; import * as EffectCodexSchema from "effect-codex-app-server/schema"; import { @@ -43,9 +42,12 @@ 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", @@ -57,13 +59,48 @@ function makeThreadOpenResponse( id: threadId, createdAt: "2026-04-18T00:00:00.000Z", source: { session: "cli" }, - turns: [], + turns: items.length === 0 ? [] : [{ id: "turn-1", status: "completed", items }], status: { state: "idle", activeFlags: [], }, }, - } as unknown as CodexRpc.ClientRequestResponsesByMethod["thread/start"]; + }; +} + +/** + * Mirrors the real client: params are ignored, and the raw payload is decoded + * with whichever response schema the caller supplied, surfacing failures the + * same way (`operation: "decode-payload"`). + */ +function makeThreadOpenClient( + respond: ( + method: "thread/start" | "thread/resume", + ) => Effect.Effect, +) { + return { + request: ( + method: "thread/start" | "thread/resume", + _payload: unknown, + responseSchema: Schema.Codec, + ) => + respond(method).pipe( + Effect.flatMap((raw) => + Schema.decodeUnknownEffect(responseSchema)(raw).pipe( + Effect.mapError( + (cause) => + new CodexErrors.CodexAppServerRequestError({ + code: -32603, + errorMessage: `Invalid payload for method '${method}' during 'decode-payload'`, + method, + operation: "decode-payload", + cause, + }), + ), + ), + ), + ), + }; } describe("buildTurnStartParams", () => { @@ -752,6 +789,20 @@ describe("isRecoverableThreadResumeError", () => { ); }); + it("matches responses this build cannot decode", () => { + NodeAssert.equal( + isRecoverableThreadResumeError( + new CodexErrors.CodexAppServerRequestError({ + code: -32603, + 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 +826,55 @@ 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]); + // Codex CLI 0.150.0 grew a fourth subAgentActivity kind. Opening the + // session must not depend on history this build cannot name (#8322). + const resumed = makeThreadOpenResponse("resumed-thread", [ + { + id: "item-18", + type: "subAgentActivity", + agentPath: "/root/child", + agentThreadId: "child-thread", + kind: "completed", }, - }; + ]); + 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 +887,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..96410d608e88 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.optional(Schema.String), + model: Schema.optional(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), + ), + ), ), ); }; @@ -2046,8 +2076,8 @@ export const makeCodexSessionRuntime = ( const session = { ...(yield* Ref.get(sessionRef)), status: "ready", - cwd: opened.cwd, - model: opened.model, + cwd: opened.cwd ?? options.cwd, + model: opened.model ?? requestedModel, resumeCursor: { threadId: providerThreadId }, updatedAt: yield* nowIso, } satisfies ProviderSession; diff --git a/docs/internals/providers.md b/docs/internals/providers.md index 7d68dc4efe17..34ade78dbc69 100644 --- a/docs/internals/providers.md +++ b/docs/internals/providers.md @@ -39,6 +39,27 @@ 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. + +Regenerating is not a substitute for either rule: because upstream also adds _required_ fields, +pinning forward can break users still on an older CLI. Bindings track the protocol; the rules above +absorb the gap between releases. + ## 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..a49ec0fa18aa 100644 --- a/packages/effect-codex-app-server/src/client.test.ts +++ b/packages/effect-codex-app-server/src/client.test.ts @@ -9,7 +9,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 +128,38 @@ 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("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..661e99618929 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], @@ -150,6 +164,15 @@ 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) => + 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 +217,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 never), + raw, + ), ), ); @@ -227,7 +251,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..074cb7547e3d 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 @@ -81,6 +81,48 @@ const handleMethod = (message: Record) => { }); return; } + // Replays a resume payload whose history carries a subAgentActivity kind + // newer than the generated bindings, the shape that made resuming a thread + // fail outright before responses could be decoded narrowly. + 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: "completed", + }, + ], + }, + ], + }, + }); + return; + } case "skills/list": { pendingSkillsListRequestId = message.id as number | string; pendingUserInputRequestId = sendRequest("item/tool/requestUserInput", { From 85ebca546a586e1c7abccf250ae0424575ce4f09 Mon Sep 17 00:00:00 2001 From: d3oxy Date: Thu, 27 Aug 2026 06:49:07 +0530 Subject: [PATCH 2/3] refactor(server): tighten Codex resume fix after review Require cwd and model in the narrow thread-open schema rather than inventing fallbacks: both are required in every Codex release we support, and defaulting to the requested model could record a model Codex never honored. Warn once per method for undecodable notifications instead of once per message, so drift in a high-frequency stream such as agent message deltas cannot warn per token, and cover that with a test. Make the thread-open test double faithful to the real client: generated param types, and the same invalidPayload error the client raises. --- .../Layers/CodexSessionRuntime.test.ts | 50 ++++++++--------- .../provider/Layers/CodexSessionRuntime.ts | 8 +-- .../src/client.test.ts | 54 +++++++++++++++++++ .../effect-codex-app-server/src/client.ts | 28 ++++++---- .../fixtures/codex-app-server-mock-peer.ts | 16 ++++++ 5 files changed, 119 insertions(+), 37 deletions(-) diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts index f5c694283c80..dd2387e3ba14 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts @@ -6,6 +6,7 @@ import * as Schema from "effect/Schema"; import { describe } from "vite-plus/test"; import { DEFAULT_MODEL, ThreadId } from "@t3tools/contracts"; import * as CodexErrors from "effect-codex-app-server/errors"; +import * as CodexRpc from "effect-codex-app-server/rpc"; import * as EffectCodexSchema from "effect-codex-app-server/schema"; import { @@ -54,24 +55,28 @@ function makeThreadOpenResponse(threadId: string, items: ReadonlyArray 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" }, + 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 }], - status: { - state: "idle", - activeFlags: [], - }, }, }; } /** - * Mirrors the real client: params are ignored, and the raw payload is decoded - * with whichever response schema the caller supplied, surfacing failures the - * same way (`operation: "decode-payload"`). + * 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: ( @@ -79,23 +84,20 @@ function makeThreadOpenClient( ) => Effect.Effect, ) { return { - request: ( - method: "thread/start" | "thread/resume", - _payload: unknown, - responseSchema: Schema.Codec, + request: ( + method: M, + _payload: CodexRpc.ClientRequestParamsByMethod[M], + responseSchema: Schema.Codec, ) => respond(method).pipe( Effect.flatMap((raw) => Schema.decodeUnknownEffect(responseSchema)(raw).pipe( - Effect.mapError( - (cause) => - new CodexErrors.CodexAppServerRequestError({ - code: -32603, - errorMessage: `Invalid payload for method '${method}' during 'decode-payload'`, - method, - operation: "decode-payload", - cause, - }), + Effect.mapError((cause) => + CodexErrors.CodexAppServerRequestError.invalidPayload( + method, + "decode-payload", + cause, + ), ), ), ), @@ -793,7 +795,7 @@ describe("isRecoverableThreadResumeError", () => { NodeAssert.equal( isRecoverableThreadResumeError( new CodexErrors.CodexAppServerRequestError({ - code: -32603, + code: -32602, errorMessage: "Invalid payload for method 'thread/resume' during 'decode-payload'", method: "thread/resume", operation: "decode-payload", diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index 96410d608e88..5ccce4198caa 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -688,8 +688,8 @@ export function isRecoverableThreadResumeError(error: unknown): boolean { */ const CodexThreadOpenResponse = Schema.Struct({ thread: Schema.Struct({ id: Schema.String }), - cwd: Schema.optional(Schema.String), - model: Schema.optional(Schema.String), + cwd: Schema.String, + model: Schema.String, }); type CodexThreadOpenResponse = typeof CodexThreadOpenResponse.Type; @@ -2076,8 +2076,8 @@ export const makeCodexSessionRuntime = ( const session = { ...(yield* Ref.get(sessionRef)), status: "ready", - cwd: opened.cwd ?? options.cwd, - model: opened.model ?? requestedModel, + cwd: opened.cwd, + model: opened.model, resumeCursor: { threadId: providerThreadId }, updatedAt: yield* nowIso, } satisfies ProviderSession; diff --git a/packages/effect-codex-app-server/src/client.test.ts b/packages/effect-codex-app-server/src/client.test.ts index a49ec0fa18aa..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"; @@ -160,6 +161,59 @@ it.layer(NodeServices.layer)("effect-codex-app-server client", (it) => { }), ); + 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 661e99618929..79f9c4eb5bf2 100644 --- a/packages/effect-codex-app-server/src/client.ts +++ b/packages/effect-codex-app-server/src/client.ts @@ -51,10 +51,10 @@ export class CodexAppServerClient extends Context.Service< method: M, payload: CodexRpc.ClientRequestParamsByMethod[M], ): Effect.Effect; - ( + ( method: M, payload: CodexRpc.ClientRequestParamsByMethod[M], - responseSchema: Schema.Codec, + responseSchema: Schema.Codec, ): Effect.Effect; }; readonly notify: ( @@ -105,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; @@ -168,10 +172,16 @@ export const make = Effect.fn("effect-codex-app-server/CodexAppServerClient.make // 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) => - Effect.logWarning("codex app-server notification dropped", { - method: notification.method, - cause: 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 }), @@ -217,10 +227,10 @@ export const make = Effect.fn("effect-codex-app-server/CodexAppServerClient.make onRequest: dispatchRequest, }); - const request = ( + const request = ( method: M, payload: CodexRpc.ClientRequestParamsByMethod[M], - responseSchema?: Schema.Codec, + responseSchema?: Schema.Codec, ): Effect.Effect => encodeOptionalPayload(method, getClientRequestParamSchema(method), payload).pipe( Effect.flatMap((encoded) => transport.request(method, encoded)), @@ -228,7 +238,7 @@ export const make = Effect.fn("effect-codex-app-server/CodexAppServerClient.make (raw): Effect.Effect => decodeOptionalPayload( method, - responseSchema ?? (getClientRequestResponseSchema(method) as never), + responseSchema ?? (getClientRequestResponseSchema(method) as Schema.Codec), raw, ), ), 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 074cb7547e3d..7f54f8b074ee 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: { From 2147190576fd0cdb7ac5d5a7f533ea23dc2230c0 Mon Sep 17 00:00:00 2001 From: d3oxy Date: Thu, 27 Aug 2026 10:49:12 +0530 Subject: [PATCH 3/3] test(server): exercise Codex drift with a value no build names #8346 taught the generated bindings the Codex 0.150 enum values, so asserting that they reject "completed" no longer tests anything. Point the fixtures at a kind the bindings do not name instead, which keeps exercising drift after they catch up with any given release. Also record the generator's definition overrides alongside the two drift rules in the providers doc: overrides fix the variants we already know about, the rules cover the ones we do not. --- .../src/provider/Layers/CodexSessionRuntime.test.ts | 7 ++++--- docs/internals/providers.md | 11 ++++++++--- .../test/fixtures/codex-app-server-mock-peer.ts | 8 +++++--- 3 files changed, 17 insertions(+), 9 deletions(-) diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts index dd2387e3ba14..a8cf90c241dd 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts @@ -830,15 +830,16 @@ describe("isRecoverableThreadResumeError", () => { describe("openCodexThread", () => { it.effect("resumes a thread whose history uses a newer protocol variant", () => Effect.gen(function* () { - // Codex CLI 0.150.0 grew a fourth subAgentActivity kind. Opening the - // session must not depend on history this build cannot name (#8322). + // 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: "completed", + kind: "escalated", }, ]); const calls: Array = []; diff --git a/docs/internals/providers.md b/docs/internals/providers.md index 34ade78dbc69..e28a757c7dbd 100644 --- a/docs/internals/providers.md +++ b/docs/internals/providers.md @@ -56,9 +56,14 @@ Two rules keep that drift from breaking sessions: 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. -Regenerating is not a substitute for either rule: because upstream also adds _required_ fields, -pinning forward can break users still on an older CLI. Bindings track the protocol; the rules above -absorb the gap between releases. +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 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 7f54f8b074ee..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 @@ -98,8 +98,10 @@ const handleMethod = (message: Record) => { return; } // Replays a resume payload whose history carries a subAgentActivity kind - // newer than the generated bindings, the shape that made resuming a thread - // fail outright before responses could be decoded narrowly. + // 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(), @@ -130,7 +132,7 @@ const handleMethod = (message: Record) => { type: "subAgentActivity", agentPath: "/root/child", agentThreadId: "child-thread", - kind: "completed", + kind: "escalated", }, ], },