From f014905ed1c73600310accc6d2bc378c9d5c7840 Mon Sep 17 00:00:00 2001 From: nullstack65 Date: Sat, 19 Sep 2026 20:51:39 +0000 Subject: [PATCH] fix(server): opt-in fail-closed policy for codex thread resume A requested provider-thread resume currently falls back to a fresh thread on any recoverable error, so a failed resume is accepted as successful continuity. Add an opt-in 'fail-closed' Codex resumeFailurePolicy (per provider instance) that propagates the resume failure instead of starting a replacement thread. Default behavior is unchanged. --- .../src/provider/Layers/CodexAdapter.ts | 3 + .../Layers/CodexSessionRuntime.test.ts | 154 ++++++++++++++++++ .../provider/Layers/CodexSessionRuntime.ts | 41 ++++- .../ProviderInstanceRegistryLive.test.ts | 1 + apps/server/src/serverSettings.test.ts | 2 + packages/contracts/src/settings.ts | 12 ++ 6 files changed, 206 insertions(+), 7 deletions(-) diff --git a/apps/server/src/provider/Layers/CodexAdapter.ts b/apps/server/src/provider/Layers/CodexAdapter.ts index b43755736ca3..56bdcb69a425 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.ts @@ -2266,6 +2266,9 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* ( ...(isCodexResumeCursorSchema(input.resumeCursor) ? { resumeCursor: input.resumeCursor } : {}), + ...(codexConfig.resumeFailurePolicy === "fail-closed" + ? { resumeFailurePolicy: codexConfig.resumeFailurePolicy } + : {}), runtimeMode: input.runtimeMode, ...(input.modelSelection?.instanceId === boundInstanceId ? { model: input.modelSelection.model } diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts index c6257221c737..b44a128e1b61 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts @@ -1039,4 +1039,158 @@ describe("openCodexThread", () => { NodeAssert.equal(error.errorMessage, "timed out waiting for server"); }), ); + + it.effect("starts a fresh thread when no resume is requested", () => + Effect.gen(function* () { + const calls: Array = []; + const started = makeThreadOpenResponse("fresh-thread"); + const opened = yield* openCodexThread({ + client: { + raw: { + request: () => Effect.die("No resume must not call thread/resume"), + }, + request: (method) => { + calls.push(method); + return Effect.succeed(started); + }, + }, + threadId: ThreadId.make("thread-1"), + runtimeMode: "full-access", + cwd: "/tmp/project", + requestedModel: "gpt-5.3-codex", + serviceTier: undefined, + resumeThreadId: undefined, + }); + + NodeAssert.equal(opened.thread.id, "fresh-thread"); + NodeAssert.deepStrictEqual(calls, ["thread/start"]); + }), + ); + + it.effect("fail-closed refuses a fresh start after a recoverable resume failure", () => + Effect.gen(function* () { + const calls: Array = []; + const client = { + raw: { + request: (method: "thread/resume") => { + calls.push(method); + return Effect.fail( + new CodexErrors.CodexAppServerRequestError({ + code: -32603, + errorMessage: "thread not found", + }), + ); + }, + }, + request: () => Effect.die("fail-closed must not start a fresh thread"), + }; + + const error = yield* openCodexThread({ + client, + threadId: ThreadId.make("thread-1"), + runtimeMode: "full-access", + cwd: "/tmp/project", + requestedModel: "gpt-5.3-codex", + serviceTier: undefined, + resumeThreadId: "stale-thread", + resumeFailurePolicy: "fail-closed", + }).pipe(Effect.flip); + + NodeAssert.ok(isCodexAppServerRequestError(error)); + NodeAssert.equal(error.errorMessage, "thread not found"); + NodeAssert.deepStrictEqual(calls, ["thread/resume"]); + }), + ); + + it.effect("fail-closed still resumes a valid thread without starting a fresh one", () => + Effect.gen(function* () { + const calls: Array = []; + const opened = yield* openCodexThread({ + client: { + request: () => Effect.die("A valid resumed thread must not start fresh"), + raw: { + request: (method: "thread/resume") => { + calls.push(method); + return Effect.succeed(makeThreadOpenResponse("saved-thread")); + }, + }, + }, + threadId: ThreadId.make("thread-1"), + runtimeMode: "auto", + cwd: "/tmp/project", + requestedModel: "gpt-5.3-codex", + serviceTier: undefined, + resumeThreadId: "saved-thread", + resumeFailurePolicy: "fail-closed", + }); + + NodeAssert.equal(opened.thread.id, "saved-thread"); + NodeAssert.deepStrictEqual(calls, ["thread/resume"]); + }), + ); + + it.effect("fail-closed still starts a fresh thread when no resume is requested", () => + Effect.gen(function* () { + const calls: Array = []; + const started = makeThreadOpenResponse("fresh-thread"); + const opened = yield* openCodexThread({ + client: { + raw: { + request: () => Effect.die("No resume must not call thread/resume"), + }, + request: (method) => { + calls.push(method); + return Effect.succeed(started); + }, + }, + threadId: ThreadId.make("thread-1"), + runtimeMode: "full-access", + cwd: "/tmp/project", + requestedModel: "gpt-5.3-codex", + serviceTier: undefined, + resumeThreadId: undefined, + resumeFailurePolicy: "fail-closed", + }); + + NodeAssert.equal(opened.thread.id, "fresh-thread"); + NodeAssert.deepStrictEqual(calls, ["thread/start"]); + }), + ); + + it.effect("explicit fallback-to-new-thread matches the stock default", () => + Effect.gen(function* () { + const calls: Array = []; + const started = makeThreadOpenResponse("fresh-thread"); + const opened = yield* openCodexThread({ + client: { + raw: { + request: (method: "thread/resume") => { + calls.push(method); + return Effect.fail( + new CodexErrors.CodexAppServerRequestError({ + code: -32603, + errorMessage: "thread not found", + }), + ); + }, + }, + request: (method) => { + calls.push(method); + return Effect.succeed(started); + }, + }, + threadId: ThreadId.make("thread-1"), + runtimeMode: "full-access", + cwd: "/tmp/project", + requestedModel: "gpt-5.3-codex", + serviceTier: undefined, + resumeThreadId: "stale-thread", + resumeFailurePolicy: "fallback-to-new-thread", + }); + + NodeAssert.notEqual(opened.thread.id, "stale-thread"); + NodeAssert.equal(opened.thread.id, "fresh-thread"); + NodeAssert.deepStrictEqual(calls, ["thread/resume", "thread/start"]); + }), + ); }); diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index a04db1405912..65a92834b243 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -65,6 +65,16 @@ const RECOVERABLE_THREAD_RESUME_ERROR_SNIPPETS = [ "no rollout found", ]; +/** + * What to do when a requested provider thread cannot be resumed. + * + * `fallback-to-new-thread` is the stock behavior: a recoverable resume failure + * silently starts a fresh provider thread. `fail-closed` propagates the resume + * failure instead, so a caller that requires continuity can never accept a + * replacement thread as the resumed one. + */ +export type CodexThreadResumeFailurePolicy = "fallback-to-new-thread" | "fail-closed"; + export function hasConfiguredMcpServer(appServerArgs: ReadonlyArray | undefined): boolean { return appServerArgs?.some((argument) => argument.includes("mcp_servers.")) === true; } @@ -178,6 +188,7 @@ export interface CodexSessionRuntimeOptions { readonly model?: string; readonly serviceTier?: CodexServiceTier | undefined; readonly resumeCursor?: CodexResumeCursor; + readonly resumeFailurePolicy?: CodexThreadResumeFailurePolicy; readonly appServerArgs?: ReadonlyArray; /** Capabilities the session's `t3-code` MCP credential grants; drives the prompt blocks. */ readonly mcpCapabilities?: ReadonlySet; @@ -724,8 +735,10 @@ export const openCodexThread = (input: { readonly requestedModel: string | undefined; readonly serviceTier: CodexServiceTier | undefined; readonly resumeThreadId: string | undefined; + readonly resumeFailurePolicy?: CodexThreadResumeFailurePolicy; }): Effect.Effect => { const resumeThreadId = input.resumeThreadId; + const resumeFailurePolicy = input.resumeFailurePolicy ?? "fallback-to-new-thread"; const startParams = buildThreadStartParams({ cwd: input.cwd, runtimeMode: input.runtimeMode, @@ -759,13 +772,24 @@ export const openCodexThread = (input: { ), ), Effect.catchIf(isRecoverableThreadResumeError, (error) => - 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))), + resumeFailurePolicy === "fail-closed" + ? Effect.logError( + "codex app-server thread resume failed; fail-closed policy refuses a fresh start", + { + threadId: input.threadId, + requestedRuntimeMode: input.runtimeMode, + resumeThreadId, + recoverable: true, + cause: error, + }, + ).pipe(Effect.andThen(Effect.fail(error))) + : 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))), ), ); }; @@ -2377,6 +2401,9 @@ export const makeCodexSessionRuntime = ( requestedModel, serviceTier: options.serviceTier, resumeThreadId: readResumeCursorThreadId(options.resumeCursor), + ...(options.resumeFailurePolicy !== undefined + ? { resumeFailurePolicy: options.resumeFailurePolicy } + : {}), }); const providerThreadId = opened.thread.id; diff --git a/apps/server/src/provider/Layers/ProviderInstanceRegistryLive.test.ts b/apps/server/src/provider/Layers/ProviderInstanceRegistryLive.test.ts index 25dafa5ba040..a44d1d671fc4 100644 --- a/apps/server/src/provider/Layers/ProviderInstanceRegistryLive.test.ts +++ b/apps/server/src/provider/Layers/ProviderInstanceRegistryLive.test.ts @@ -104,6 +104,7 @@ const makeCodexConfig = (overrides: Partial): CodexSettings => ({ homePath: "", shadowHomePath: "", launchArgs: "", + resumeFailurePolicy: "fallback-to-new-thread", customModels: [], ...overrides, }); diff --git a/apps/server/src/serverSettings.test.ts b/apps/server/src/serverSettings.test.ts index 208d75fb6517..8f3082afd11c 100644 --- a/apps/server/src/serverSettings.test.ts +++ b/apps/server/src/serverSettings.test.ts @@ -230,6 +230,7 @@ it.layer(NodeServices.layer)("server settings", (it) => { homePath: "/Users/julius/.codex", shadowHomePath: "", launchArgs: "", + resumeFailurePolicy: "fallback-to-new-thread", customModels: [], }); assert.deepEqual(next.providers.claudeAgent, { @@ -939,6 +940,7 @@ it.layer(NodeServices.layer)("server settings", (it) => { homePath: "", shadowHomePath: "", launchArgs: "", + resumeFailurePolicy: "fallback-to-new-thread", customModels: [], }); assert.deepEqual(next.providers.claudeAgent, { diff --git a/packages/contracts/src/settings.ts b/packages/contracts/src/settings.ts index 1c53b36037ba..de59677484d0 100644 --- a/packages/contracts/src/settings.ts +++ b/packages/contracts/src/settings.ts @@ -597,6 +597,15 @@ export const CodexSettings = makeProviderSettingsSchema( description: "Additional CLI arguments passed to codex app-server on session start.", }), ), + resumeFailurePolicy: Schema.Literals(["fallback-to-new-thread", "fail-closed"]).pipe( + Schema.withDecodingDefault(Effect.succeed("fallback-to-new-thread")), + Schema.annotateKey({ + title: "Thread resume failure policy", + description: + "What to do when a requested provider thread cannot be resumed. 'fail-closed' refuses to start a replacement thread so a failed resume is never reported as continuity.", + providerSettingsForm: { hidden: true }, + }), + ), customModels: Schema.Array(CustomModelSetting).pipe( Schema.withDecodingDefault(Effect.succeed([])), Schema.annotateKey({ providerSettingsForm: { hidden: true } }), @@ -1299,6 +1308,9 @@ const CodexSettingsPatch = Schema.Struct({ homePath: Schema.optionalKey(TrimmedString), shadowHomePath: Schema.optionalKey(TrimmedString), launchArgs: Schema.optionalKey(TrimmedString), + resumeFailurePolicy: Schema.optionalKey( + Schema.Literals(["fallback-to-new-thread", "fail-closed"]), + ), customModels: Schema.optionalKey(Schema.Array(CustomModelSetting)), });