diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 3145e33dabd1a650a67c6c45754af604f9a41a14..85748fa056e77181bcc18a4ddb57700829a4e84a 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -2597,6 +2597,161 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); + it.effect.each( + ( + [ + ["recovers archived-session errors", "session saved-thread is archived", 0, ""], + [ + "recovers unarchive hints", + "Run `codex unarchive saved-thread` to unarchive it first.", + 0, + "", + ], + ["preserves missing-thread errors", "thread not found", 3, "thread not found"], + ["preserves missing-rollout errors", "no rollout found", 3, "no rollout found"], + ["preserves authentication errors", "authentication failed", 3, "authentication failed"], + [ + "preserves archived-workspace errors", + "workspace is archived", + 3, + "workspace is archived", + ], + [ + "preserves unrelated archive-path errors", + "permission denied reading archived_sessions/saved-thread", + 3, + "permission denied reading archived_sessions/saved-thread", + ], + [ + "propagates unarchive missing-thread errors", + "session saved-thread is archived", + 4, + "thread not found", + ], + [ + "propagates unarchive archived errors", + "session saved-thread is archived", + 4, + "session saved-thread is archived", + ], + [ + "propagates retry missing-thread errors", + "session saved-thread is archived", + 5, + "thread not found", + ], + [ + "does not retry archived errors twice", + "session saved-thread is archived", + 5, + "session saved-thread is archived", + ], + ] as const + ).map(([name, resumeError, failAt, finalError]) => ({ name, resumeError, failAt, finalError })), + )("$name", ({ name, resumeError, failAt, finalError }) => + Effect.scoped( + Effect.gen(function* () { + const nativeThreadId = "saved-thread"; + const params = { + threadId: nativeThreadId, + excludeTurns: true, + cwd: CODEX_TEST_RUNTIME_POLICY.cwd, + model: CODEX_TEST_MODEL_SELECTION.model, + config: CodexAdapterV2.CODEX_THREAD_CONFIG, + }; + const entries: Array = [ + ...codexReplayPreamble({ + nativeThreadId, + nativeTurnId: "unused-turn", + prompt: "unused-prompt", + }).slice(0, 5), + { + type: "expect_outbound", + label: "resume archived thread", + frame: { id: 3, method: "thread/resume", params }, + }, + { + type: "emit_inbound", + label: "resume error", + frame: { id: 3, error: { code: -32600, message: resumeError } }, + }, + ]; + if (failAt !== 3) { + entries.push( + { + type: "expect_outbound", + label: "unarchive same thread", + frame: { id: 4, method: "thread/unarchive", params: { threadId: nativeThreadId } }, + }, + { + type: "emit_inbound", + label: "unarchive result", + frame: + failAt === 4 + ? { id: 4, error: { code: -32600, message: finalError } } + : { id: 4, result: { thread: { turns: [{ type: "unknown-history-item" }] } } }, + }, + ); + } + if (failAt === 0 || failAt === 5) { + entries.push( + { + type: "expect_outbound", + label: "retry identical resume", + frame: { id: 5, method: "thread/resume", params }, + }, + { + type: "emit_inbound", + label: "retry result", + frame: + failAt === 5 + ? { id: 5, error: { code: -32600, message: finalError } } + : { + id: 5, + result: { + thread: { + id: nativeThreadId, + updatedAt: 1782622450, + turns: [{ type: "unknown-history-item" }], + }, + }, + }, + }, + ); + } + const harness = yield* makeCodexReplayHarness( + makeCodexReplayTranscript({ scenario: name, entries }), + ); + const resume = harness.runtime.resumeThread({ + providerThread: harness.providerThread, + modelSelection: CODEX_TEST_MODEL_SELECTION, + runtimePolicy: CODEX_TEST_RUNTIME_POLICY, + }); + if (failAt !== 0) { + const error = yield* Effect.flip(resume); + assert.equal(error._tag, "ProviderAdapterResumeThreadError"); + assert.nestedPropertyVal(error, "cause.errorMessage", finalError); + assert.nestedPropertyVal( + error, + "cause.method", + failAt === 4 ? "thread/unarchive" : "thread/resume", + ); + assert.nestedPropertyVal(error, "cause.requestId", String(failAt)); + return; + } + const resumed = yield* resume; + assert.equal(resumed.id, harness.providerThread.id); + assert.equal(resumed.nativeThreadRef?.nativeId, nativeThreadId); + assert.deepEqual( + resumed.nativeConversationHeadRef, + harness.providerThread.nativeConversationHeadRef, + ); + assert.equal(resumed.status, "idle"); + assert.equal(DateTime.toEpochMillis(resumed.updatedAt), 1782622450000); + }).pipe(Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))), + ), + ); + it.effect("continues an interrupted native thread with empty input and reasoning summaries", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 19c3486f5af783dad1df32b1f14c443d287ee55e..8da6ffe4357dd5ea0334d1ee13d84ea7c244d736 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -5429,23 +5429,39 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi resumeThread: (threadInput) => Effect.gen(function* () { const nativeThreadId = yield* getNativeThreadId(threadInput.providerThread); - + // excludeTurns is not in the generated request schema yet. + const resume = client.raw.request("thread/resume", { + threadId: nativeThreadId, + excludeTurns: true, + ...codexThreadRuntimeParams({ + threadId: threadInput.threadId ?? threadInput.providerThread.appThreadId, + ...(threadInput.modelSelection === undefined + ? {} + : { modelSelection: threadInput.modelSelection }), + ...(threadInput.runtimePolicy === undefined + ? {} + : { runtimePolicy: threadInput.runtimePolicy }), + }), + }); const response = yield* ensureInitialized.pipe( Effect.andThen( - // excludeTurns is not in the generated request schema yet. - client.raw.request("thread/resume", { - threadId: nativeThreadId, - excludeTurns: true, - ...codexThreadRuntimeParams({ - threadId: threadInput.threadId ?? threadInput.providerThread.appThreadId, - ...(threadInput.modelSelection === undefined - ? {} - : { modelSelection: threadInput.modelSelection }), - ...(threadInput.runtimePolicy === undefined - ? {} - : { runtimePolicy: threadInput.runtimePolicy }), + resume.pipe( + Effect.catchTags({ + CodexAppServerRequestError: (cause) => { + if ( + !/\bsession \S+ is archived\b|\bcodex unarchive\b/i.test( + cause.errorMessage, + ) + ) { + return Effect.fail(cause); + } + // Keep the session's history without decoding the unarchive response. + return client.raw + .request("thread/unarchive", { threadId: nativeThreadId }) + .pipe(Effect.andThen(resume)); + }, }), - }), + ), ), Effect.flatMap(decodeCodexResumeMetadata), );