diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index e20edf4c5e723..9f51465454ac1 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -7702,6 +7702,121 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); + it.effect("unarchives an unloaded Codex thread before rolling it back", () => + Effect.gen(function* () { + const nativeThreadId = "archived-rollback-thread"; + const resumeParams = { + threadId: nativeThreadId, + excludeTurns: true, + cwd: CODEX_TEST_RUNTIME_POLICY.cwd, + model: CODEX_TEST_MODEL_SELECTION.model, + config: CodexAdapterV2.CODEX_THREAD_CONFIG, + }; + const thread = codexReplayThreadResult({ nativeThreadId, forkedFromId: null }); + const exchange = ( + id: number, + method: string, + params: unknown, + reply: { readonly result: unknown } | { readonly error: unknown }, + ): Array => [ + { type: "expect_outbound", label: method, frame: { id, method, params } }, + { type: "emit_inbound", label: method, frame: { id, ...reply } }, + ]; + const transcript = makeCodexReplayTranscript({ + scenario: "codex-archived-rollback", + entries: [ + ...codexReplayPreamble({ + nativeThreadId, + nativeTurnId: "archived-rollback-turn", + prompt: "unused", + }).slice(0, 5), + ...exchange( + 3, + "thread/read", + { threadId: nativeThreadId, includeTurns: false }, + { + result: { + thread: { + id: nativeThreadId, + historyMode: "paginated", + status: { type: "notLoaded" }, + }, + }, + }, + ), + ...exchange(4, "thread/resume", resumeParams, { + error: { code: -32600, message: `session ${nativeThreadId} is archived` }, + }), + ...exchange(5, "thread/unarchive", { threadId: nativeThreadId }, { result: thread }), + ...exchange(6, "thread/resume", resumeParams, { result: thread }), + ...exchange( + 7, + "thread/turns/list", + { + threadId: nativeThreadId, + cursor: null, + limit: 1, + sortDirection: "desc", + itemsView: "summary", + }, + { + result: { + data: [{ id: "native-turn-second", items: [], status: "completed", error: null }], + nextCursor: null, + }, + }, + ), + ...exchange( + 8, + "thread/revert", + { threadId: nativeThreadId, beforeTurnId: "native-turn-second" }, + { result: thread }, + ), + ], + }); + const harness = yield* makeCodexReplayHarness(transcript); + const now = yield* DateTime.now; + const firstTurn = codexReplaySourceTurn({ + id: "provider-turn-first", + ordinal: 1, + nativeId: "native-turn-first", + providerThreadId: harness.providerThread.id, + now, + }); + const secondTurn = codexReplaySourceTurn({ + id: "provider-turn-second", + ordinal: 2, + nativeId: "native-turn-second", + providerThreadId: harness.providerThread.id, + now, + }); + + const snapshot = yield* harness.runtime.rollbackThread({ + providerThread: harness.providerThread, + target: { + type: "provider_turn", + checkpointId: CheckpointId.make("checkpoint-archived-rollback"), + appRunOrdinal: 1, + providerTurn: firstTurn, + }, + providerThreadTurns: [firstTurn, secondTurn], + }); + + assert.equal(snapshot.providerThread.nativeThreadRef?.nativeId, nativeThreadId); + assert.deepEqual(snapshot.providerThread.nativeConversationHeadRef, firstTurn.nativeTurnRef); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ); + it.effect( "falls back to fork-local thread/revert on paginated history when the source turn lacks a native reference", () => diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 277232830676d..b0018be2d0f16 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -6696,6 +6696,24 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( ); }); + // Codex refuses to resume a session archived elsewhere until it is unarchived. + const resumeNativeThread = Effect.fn("CodexAdapterV2.resumeNativeThread")(function* ( + threadId: string, + params: Record, + ) { + const resume = client.raw.request("thread/resume", { threadId, ...params }); + return yield* resume.pipe( + Effect.catchTags({ + CodexAppServerRequestError: (cause) => + /\bsession \S+ is archived\b|\bcodex unarchive\b/i.test(cause.errorMessage) + ? client.raw + .request("thread/unarchive", { threadId }) + .pipe(Effect.andThen(resume)) + : Effect.fail(cause), + }), + ); + }); + const startNativeTurn = Effect.fn("CodexAdapterV2.startNativeTurn")( function* ( turnInput: ProviderAdapter.ProviderAdapterV2TurnInput, @@ -7048,8 +7066,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( function* (threadInput) { const nativeThreadId = yield* getNativeThreadId(threadInput.providerThread); const appThreadId = threadInput.threadId ?? threadInput.providerThread.appThreadId; - const resume = client.raw.request("thread/resume", { - threadId: nativeThreadId, + const resume = resumeNativeThread(nativeThreadId, { excludeTurns: true, ...codexThreadRuntimeParams({ threadId: appThreadId, @@ -7063,24 +7080,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( }), }); const response = yield* ensureInitialized.pipe( - Effect.andThen( - resume.pipe( - Effect.catchTags({ - CodexAppServerRequestError: (cause) => { - if ( - !/\bsession \S+ is archived\b|\bcodex unarchive\b/i.test( - cause.errorMessage, - ) - ) { - return Effect.fail(cause); - } - return client.raw - .request("thread/unarchive", { threadId: nativeThreadId }) - .pipe(Effect.andThen(resume)); - }, - }), - ), - ), + Effect.andThen(resume), Effect.flatMap(decodeCodexResumeMetadata), ); return { @@ -7834,8 +7834,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( } // thread/revert requires the thread loaded in this app-server; resume after restart or idle release. if (!loaded) { - yield* client.raw.request("thread/resume", { - threadId, + yield* resumeNativeThread(threadId, { excludeTurns: true, ...codexThreadRuntimeParams({ threadId: threadInput.providerThread.appThreadId,