diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts index d7346a0e0db..17daddb53dc 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts @@ -17,6 +17,7 @@ import { codexSessionAppServerArgs } from "./codexLaunchArgs.ts"; import { buildTurnStartParams, hasConfiguredMcpServer, + isArchivedThreadResumeError, isRecoverableThreadResumeError, openCodexThread, } from "./CodexSessionRuntime.ts"; @@ -346,6 +347,28 @@ describe("codexSessionAppServerArgs", () => { }); describe("isRecoverableThreadResumeError", () => { + it("matches archived Codex session resume errors separately", () => { + NodeAssert.equal( + isArchivedThreadResumeError( + new CodexErrors.CodexAppServerRequestError({ + code: -32603, + errorMessage: + "session 019f4b39-5e16-7661-a5eb-8e482f309226 is archived. Run `codex unarchive 019f4b39-5e16-7661-a5eb-8e482f309226` to unarchive it first.", + }), + ), + true, + ); + NodeAssert.equal( + isArchivedThreadResumeError( + new CodexErrors.CodexAppServerRequestError({ + code: -32603, + errorMessage: "Archived thread metadata could not be decoded", + }), + ), + false, + ); + }); + it("matches missing thread errors", () => { NodeAssert.equal( isRecoverableThreadResumeError( @@ -393,12 +416,166 @@ describe("isRecoverableThreadResumeError", () => { }); describe("openCodexThread", () => { + it.effect("unarchives an archived thread before retrying resume", () => + Effect.gen(function* () { + type Method = "thread/start" | "thread/resume" | "thread/unarchive"; + const calls: Array<{ method: Method; payload: unknown }> = []; + const resumed = makeThreadOpenResponse("archived-thread"); + let resumeAttempts = 0; + const client = { + request: ( + method: M, + payload: CodexRpc.ClientRequestParamsByMethod[M], + ) => { + calls.push({ method, payload }); + if (method === "thread/resume" && resumeAttempts++ === 0) { + return Effect.fail( + new CodexErrors.CodexAppServerRequestError({ + code: -32603, + errorMessage: + "session archived-thread is archived. Run `codex unarchive archived-thread` to unarchive it first.", + }), + ); + } + return Effect.succeed(resumed 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: "archived-thread", + }); + + NodeAssert.equal(opened.thread.id, "archived-thread"); + NodeAssert.deepStrictEqual( + calls.map((call) => call.method), + ["thread/resume", "thread/unarchive", "thread/resume"], + ); + NodeAssert.deepStrictEqual(calls[1]?.payload, { + threadId: "archived-thread", + }); + NodeAssert.ok(calls[0]); + NodeAssert.ok(calls[2]); + NodeAssert.equal( + (calls[0].payload as { readonly threadId?: string }).threadId, + "archived-thread", + ); + NodeAssert.equal( + (calls[2].payload as { readonly threadId?: string }).threadId, + "archived-thread", + ); + }), + ); + + it.effect("propagates an unarchive failure instead of starting a new thread", () => + Effect.gen(function* () { + type Method = "thread/start" | "thread/resume" | "thread/unarchive"; + const calls: Array = []; + const client = { + request: ( + method: M, + _payload: CodexRpc.ClientRequestParamsByMethod[M], + ) => { + calls.push(method); + if (method === "thread/resume") { + return Effect.fail( + new CodexErrors.CodexAppServerRequestError({ + code: -32603, + errorMessage: "thread archived-thread is archived", + }), + ); + } + if (method === "thread/unarchive") { + return Effect.fail( + new CodexErrors.CodexAppServerRequestError({ + code: -32603, + errorMessage: "thread not found while unarchiving archived-thread", + }), + ); + } + return Effect.succeed( + makeThreadOpenResponse("fresh-thread") as CodexRpc.ClientRequestResponsesByMethod[M], + ); + }, + }; + + const error = yield* openCodexThread({ + client, + threadId: ThreadId.make("thread-1"), + runtimeMode: "full-access", + cwd: "/tmp/project", + requestedModel: "gpt-5.3-codex", + serviceTier: undefined, + resumeThreadId: "archived-thread", + }).pipe(Effect.flip); + + NodeAssert.ok(isCodexAppServerRequestError(error)); + NodeAssert.equal(error.errorMessage, "thread not found while unarchiving archived-thread"); + NodeAssert.deepStrictEqual(calls, ["thread/resume", "thread/unarchive"]); + }), + ); + + it.effect("propagates a failed resume retry after unarchiving", () => + Effect.gen(function* () { + type Method = "thread/start" | "thread/resume" | "thread/unarchive"; + const calls: Array = []; + let resumeAttempts = 0; + const client = { + request: ( + method: M, + _payload: CodexRpc.ClientRequestParamsByMethod[M], + ) => { + calls.push(method); + if (method === "thread/resume" && resumeAttempts++ === 0) { + return Effect.fail( + new CodexErrors.CodexAppServerRequestError({ + code: -32603, + errorMessage: "thread archived-thread is archived", + }), + ); + } + if (method === "thread/resume") { + return Effect.fail( + new CodexErrors.CodexAppServerRequestError({ + code: -32603, + errorMessage: "thread not found while retrying archived-thread", + }), + ); + } + return Effect.succeed( + makeThreadOpenResponse("archived-thread") as CodexRpc.ClientRequestResponsesByMethod[M], + ); + }, + }; + + const error = yield* openCodexThread({ + client, + threadId: ThreadId.make("thread-1"), + runtimeMode: "full-access", + cwd: "/tmp/project", + requestedModel: "gpt-5.3-codex", + serviceTier: undefined, + resumeThreadId: "archived-thread", + }).pipe(Effect.flip); + + NodeAssert.ok(isCodexAppServerRequestError(error)); + NodeAssert.equal(error.errorMessage, "thread not found while retrying archived-thread"); + NodeAssert.deepStrictEqual(calls, ["thread/resume", "thread/unarchive", "thread/resume"]); + }), + ); + it.effect("falls back to thread/start when resume fails recoverably", () => Effect.gen(function* () { - const calls: Array<{ method: "thread/start" | "thread/resume"; payload: unknown }> = []; + type Method = "thread/start" | "thread/resume" | "thread/unarchive"; + const calls: Array<{ method: Method; payload: unknown }> = []; const started = makeThreadOpenResponse("fresh-thread"); const client = { - request: ( + request: ( method: M, payload: CodexRpc.ClientRequestParamsByMethod[M], ) => { @@ -435,8 +612,9 @@ describe("openCodexThread", () => { it.effect("propagates non-recoverable resume failures", () => Effect.gen(function* () { + type Method = "thread/start" | "thread/resume" | "thread/unarchive"; 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 67108dd4dbb..62f8fe8dfad 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -59,6 +59,7 @@ const RECOVERABLE_THREAD_RESUME_ERROR_SNIPPETS = [ "unknown thread", "does not exist", ]; +const ARCHIVED_THREAD_RESUME_ERROR_REGEX = /\b(?:session|thread)\s+\S+\s+is archived\b/i; export function hasConfiguredMcpServer(appServerArgs: ReadonlyArray | undefined): boolean { return appServerArgs?.some((argument) => argument.includes("mcp_servers.")) === true; @@ -441,11 +442,16 @@ export function isRecoverableThreadResumeError(error: unknown): boolean { return RECOVERABLE_THREAD_RESUME_ERROR_SNIPPETS.some((snippet) => message.includes(snippet)); } +export function isArchivedThreadResumeError(error: unknown): boolean { + const message = error instanceof Error ? error.message : String(error); + return ARCHIVED_THREAD_RESUME_ERROR_REGEX.test(message); +} + type CodexThreadOpenResponse = | CodexRpc.ClientRequestResponsesByMethod["thread/start"] | CodexRpc.ClientRequestResponsesByMethod["thread/resume"]; -type CodexThreadOpenMethod = "thread/start" | "thread/resume"; +type CodexThreadOpenMethod = "thread/start" | "thread/resume" | "thread/unarchive"; interface CodexThreadOpenClient { readonly request: ( @@ -475,22 +481,46 @@ export const openCodexThread = (input: { return input.client.request("thread/start", startParams); } - return input.client - .request("thread/resume", { + const resumeThread = () => + input.client.request("thread/resume", { threadId: resumeThreadId, ...startParams, - }) - .pipe( - Effect.catchIf(isRecoverableThreadResumeError, (error) => - Effect.logWarning("codex app-server thread resume fell back to fresh start", { + }); + + return resumeThread().pipe( + Effect.catch((error) => { + if (isArchivedThreadResumeError(error)) { + return Effect.gen(function* () { + yield* Effect.logWarning( + "codex app-server thread resume is unarchiving the persisted session", + { + threadId: input.threadId, + requestedRuntimeMode: input.runtimeMode, + resumeThreadId, + recoverable: true, + cause: error, + }, + ); + yield* input.client.request("thread/unarchive", { + threadId: resumeThreadId, + }); + return yield* resumeThread(); + }); + } + + if (isRecoverableThreadResumeError(error)) { + return Effect.logWarning("codex app-server thread resume fell back to fresh start", { threadId: input.threadId, requestedRuntimeMode: input.runtimeMode, resumeThreadId, recoverable: true, cause: error, - }).pipe(Effect.andThen(input.client.request("thread/start", startParams))), - ), - ); + }).pipe(Effect.andThen(input.client.request("thread/start", startParams))); + } + + return Effect.fail(error); + }), + ); }; function readNotificationThreadId(notification: CodexServerNotification): string | undefined {