diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 4471be814bf39..c3efaa7ce8af9 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -8111,6 +8111,84 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); + it.effect("unarchives an archived Codex source thread before forking it", () => + Effect.gen(function* () { + const nativeThreadId = "archived-fork-source"; + const forkThreadId = "archived-fork-target"; + const forkParams = { + threadId: nativeThreadId, + lastTurnId: "native-turn-first", + config: CodexAdapterV2.CODEX_THREAD_CONFIG, + }; + 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-fork", + entries: [ + ...codexReplayPreamble({ + nativeThreadId, + nativeTurnId: "archived-fork-turn", + prompt: "unused", + }).slice(0, 5), + ...exchange(3, "thread/fork", forkParams, { + error: { + code: -32600, + message: `session ${nativeThreadId} is archived. Run \`codex unarchive ${nativeThreadId}\` to unarchive it first.`, + }, + }), + ...exchange( + 4, + "thread/unarchive", + { threadId: nativeThreadId }, + { result: codexReplayThreadResult({ nativeThreadId, forkedFromId: null }) }, + ), + ...exchange(5, "thread/fork", forkParams, { + result: codexReplayThreadResult({ + nativeThreadId: forkThreadId, + forkedFromId: nativeThreadId, + }), + }), + ], + }); + 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 forkedProviderThread = yield* harness.runtime.forkThread({ + sourceProviderThread: harness.providerThread, + sourceProviderTurns: [firstTurn], + providerTurnId: firstTurn.id, + targetThreadId: ThreadId.make("thread-archived-fork-target"), + }); + + assert.equal(forkedProviderThread.nativeThreadRef?.nativeId, forkThreadId); + assert.equal(forkedProviderThread.forkedFrom?.providerTurnId, firstTurn.id); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ); + describe("native goals", () => { const nativeThreadId = "goal-thread"; const objective = "Ship the feature"; diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 688de092c03d0..43e33d1d6692c 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -7909,24 +7909,32 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( function* (threadInput) { const threadId = yield* getNativeThreadId(threadInput.sourceProviderThread); const boundary = yield* resolveCodexForkBoundary(threadInput); + const fork = client.request("thread/fork", { + threadId, + ...(boundary.lastTurnId === undefined ? {} : { lastTurnId: boundary.lastTurnId }), + ...codexThreadRuntimeParams({ + threadId: threadInput.targetThreadId, + mcpSession: yield* mcpSessions.read(threadInput.targetThreadId), + ...(threadInput.modelSelection === undefined + ? {} + : { modelSelection: threadInput.modelSelection }), + ...(threadInput.runtimePolicy === undefined + ? {} + : { runtimePolicy: threadInput.runtimePolicy }), + }), + }); const response = yield* ensureInitialized.pipe( Effect.andThen( - client.request("thread/fork", { - threadId, - ...(boundary.lastTurnId === undefined - ? {} - : { lastTurnId: boundary.lastTurnId }), - ...codexThreadRuntimeParams({ - threadId: threadInput.targetThreadId, - mcpSession: yield* mcpSessions.read(threadInput.targetThreadId), - ...(threadInput.modelSelection === undefined - ? {} - : { modelSelection: threadInput.modelSelection }), - ...(threadInput.runtimePolicy === undefined - ? {} - : { runtimePolicy: threadInput.runtimePolicy }), + fork.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(fork)) + : Effect.fail(cause), }), - }), + ), ), Effect.mapError( (cause) =>