From 4a7c290bf50c911a67bf7979b25f86e3e3e04ce7 Mon Sep 17 00:00:00 2001 From: Alex Southwell <4596216+saphid@users.noreply.github.com> Date: Tue, 3 Nov 2026 08:01:29 +1100 Subject: [PATCH 1/3] fix(codex): restoring a checkpoint works after the Codex session was archived Co-Authored-By: Claude Opus 5.5 --- .../Adapters/CodexAdapterV2.test.ts | 110 ++++++++++++++++++ .../Adapters/CodexAdapterV2.ts | 40 +++---- 2 files changed, 128 insertions(+), 22 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 2d14fa9518de6..2b7f60be55106 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -7466,6 +7466,116 @@ 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), + ), + ), + ); + 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 f81a3809b35fe..85200ca79de9e 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -6696,6 +6696,21 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( ); }); + // Codex refuses to resume a session archived elsewhere until it is unarchived. + const resumeNativeThread = (threadId: string, params: Record) => { + const resume = client.raw.request("thread/resume", { threadId, ...params }); + return 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 +7063,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 +7077,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 +7831,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, From f550fb86c18176db79cfbab554cafbc2c77600da Mon Sep 17 00:00:00 2001 From: Alex Southwell <4596216+saphid@users.noreply.github.com> Date: Tue, 3 Nov 2026 08:59:53 +1100 Subject: [PATCH 2/3] refactor(codex): name the resume span Co-Authored-By: Claude Opus 5.5 --- .../src/orchestration-v2/Adapters/CodexAdapterV2.ts | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 85200ca79de9e..170d4319f5f97 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -6697,9 +6697,12 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( }); // Codex refuses to resume a session archived elsewhere until it is unarchived. - const resumeNativeThread = (threadId: string, params: Record) => { + const resumeNativeThread = Effect.fn("CodexAdapterV2.resumeNativeThread")(function* ( + threadId: string, + params: Record, + ) { const resume = client.raw.request("thread/resume", { threadId, ...params }); - return resume.pipe( + return yield* resume.pipe( Effect.catchTags({ CodexAppServerRequestError: (cause) => /\bsession \S+ is archived\b|\bcodex unarchive\b/i.test(cause.errorMessage) @@ -6709,7 +6712,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( : Effect.fail(cause), }), ); - }; + }); const startNativeTurn = Effect.fn("CodexAdapterV2.startNativeTurn")( function* ( From 51c61c42226cd3f0f903276810206c9c6560959b Mon Sep 17 00:00:00 2001 From: maria-rcks <254055478+maria-rcks@users.noreply.github.com> Date: Tue, 3 Nov 2026 00:41:18 +0000 Subject: [PATCH 3/3] test(codex): provide NodeCrypto to the rollback unarchive test after the Effect 4.1.2 upgrade --- .../src/orchestration-v2/Adapters/CodexAdapterV2.test.ts | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 65667ce466b5a..9f51465454ac1 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -7807,7 +7807,12 @@ describe("CodexAdapterV2 post-settle continuation", () => { }).pipe( Effect.scoped, Effect.provide( - Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), ), ), );