diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts index a80ef2cf56a69..e01c1ad89fd09 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts @@ -19,6 +19,7 @@ import { buildTurnStartParams, hasConfiguredMcpServer, isRecoverableThreadResumeError, + isThreadActiveWriterError, makeMemoryConsolidationNotificationFilter, openCodexThread, } from "./CodexSessionRuntime.ts"; @@ -570,13 +571,93 @@ describe("isRecoverableThreadResumeError", () => { }); }); +describe("isThreadActiveWriterError", () => { + it("matches Codex thread writer conflicts", () => { + NodeAssert.equal( + isThreadActiveWriterError( + new CodexErrors.CodexAppServerRequestError({ + code: -32603, + errorMessage: "thread provider-thread already has an active writer", + }), + ), + true, + ); + }); + + it("ignores unrelated writer errors", () => { + NodeAssert.equal( + isThreadActiveWriterError( + new CodexErrors.CodexAppServerRequestError({ + code: -32603, + errorMessage: "log file already has an active writer", + }), + ), + false, + ); + }); +}); + describe("openCodexThread", () => { + it.effect("forks the thread when another Codex process owns its writer", () => + Effect.gen(function* () { + const calls: Array<{ + method: "thread/start" | "thread/resume" | "thread/fork"; + payload: unknown; + }> = []; + const forked = makeThreadOpenResponse("forked-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 provider-thread already has an active writer", + }), + ); + } + return Effect.succeed(forked as CodexRpc.ClientRequestResponsesByMethod[M]); + }, + }; + + const opened = yield* openCodexThread({ + client, + threadId: ThreadId.make("thread-1"), + runtimeMode: "full-access", + cwd: "/tmp/project", + requestedModel: "gpt-5.3-codex", + serviceTier: undefined, + resumeThreadId: "provider-thread", + }); + + NodeAssert.equal(opened.thread.id, "forked-thread"); + NodeAssert.deepStrictEqual( + calls.map((call) => call.method), + ["thread/resume", "thread/fork"], + ); + NodeAssert.deepStrictEqual(calls[1]?.payload, { + threadId: "provider-thread", + cwd: "/tmp/project", + approvalPolicy: "never", + sandbox: "danger-full-access", + approvalsReviewer: "user", + model: "gpt-5.3-codex", + }); + }), + ); + it.effect("falls back to thread/start when resume fails recoverably", () => Effect.gen(function* () { - const calls: Array<{ method: "thread/start" | "thread/resume"; payload: unknown }> = []; + const calls: Array<{ + method: "thread/start" | "thread/resume" | "thread/fork"; + payload: unknown; + }> = []; const started = makeThreadOpenResponse("fresh-thread"); const client = { - request: ( + request: ( method: M, payload: CodexRpc.ClientRequestParamsByMethod[M], ) => { @@ -614,7 +695,7 @@ describe("openCodexThread", () => { it.effect("propagates non-recoverable resume failures", () => Effect.gen(function* () { const client = { - request: ( + request: ( method: M, _payload: CodexRpc.ClientRequestParamsByMethod[M], ) => { diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index 29bb992611c14..4f09d61664ca1 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -447,11 +447,17 @@ export function isRecoverableThreadResumeError(error: unknown): boolean { return RECOVERABLE_THREAD_RESUME_ERROR_SNIPPETS.some((snippet) => message.includes(snippet)); } +export function isThreadActiveWriterError(error: unknown): boolean { + const message = (error instanceof Error ? error.message : String(error)).toLowerCase(); + return message.includes("thread") && message.includes("already has an active writer"); +} + type CodexThreadOpenResponse = | CodexRpc.ClientRequestResponsesByMethod["thread/start"] - | CodexRpc.ClientRequestResponsesByMethod["thread/resume"]; + | CodexRpc.ClientRequestResponsesByMethod["thread/resume"] + | CodexRpc.ClientRequestResponsesByMethod["thread/fork"]; -type CodexThreadOpenMethod = "thread/start" | "thread/resume"; +type CodexThreadOpenMethod = "thread/start" | "thread/resume" | "thread/fork"; interface CodexThreadOpenClient { readonly request: ( @@ -487,6 +493,29 @@ export const openCodexThread = (input: { ...startParams, }) .pipe( + Effect.catchIf(isThreadActiveWriterError, (error) => + Effect.logWarning("codex app-server thread resume forked from active writer", { + threadId: input.threadId, + requestedRuntimeMode: input.runtimeMode, + resumeThreadId, + cause: error, + }).pipe( + Effect.andThen( + Effect.suspend(() => { + const config = runtimeModeToThreadConfig(input.runtimeMode); + return input.client.request("thread/fork", { + threadId: resumeThreadId, + cwd: input.cwd, + approvalPolicy: config.approvalPolicy, + approvalsReviewer: config.approvalsReviewer, + sandbox: config.sandbox, + ...(input.requestedModel ? { model: input.requestedModel } : {}), + ...(input.serviceTier ? { serviceTier: input.serviceTier } : {}), + }); + }), + ), + ), + ), Effect.catchIf(isRecoverableThreadResumeError, (error) => Effect.logWarning("codex app-server thread resume fell back to fresh start", { threadId: input.threadId,