diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 6e5fd3516e0b1..cc2042554f3b8 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -9595,6 +9595,72 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); + it.effect("/goal resume tracks the goal turn Codex already started", () => + Effect.gen(function* () { + const requests: string[] = []; + const transcript = makeCodexReplayTranscript({ + scenario: "goal-resume-running", + entries: [ + ...sessionStart, + // Codex restores an active goal on thread/resume and starts its turn on its own. + notification("goal restored", "thread/goal/updated", { + threadId: nativeThreadId, + goal: codexGoal("active", 0), + }), + notification("native goal turn", "turn/started", { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: "native-goal-turn", status: "inProgress" }), + }), + ...request( + 3, + "thread/goal/get", + { threadId: nativeThreadId }, + { goal: codexGoal("active", 0) }, + ), + notification("goal complete", "thread/goal/updated", { + threadId: nativeThreadId, + turnId: "native-goal-turn", + goal: codexGoal("complete", 10), + }), + notification("native goal turn done", "turn/completed", { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: "native-goal-turn", status: "completed" }), + }), + ], + }); + const harness = yield* makeCodexReplayHarness( + transcript, + () => Effect.void, + (method) => + Effect.sync(() => { + requests.push(method); + }), + ); + const start = yield* harness.runtime + .startTurn(yield* goalTurnInput(harness, "/goal resume")) + .pipe(Effect.result); + + // Codex answers an empty turn/start on a running turn with EmptyInput. + assert.notInclude(requests, "turn/start"); + assert.equal(start._tag, "Success"); + yield* harness.firstTerminal; + assert.deepEqual( + harness.terminalEvents().map((event) => event.status), + ["completed"], + ); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ); + it.effect("rejects an oversized /goal objective before any goal request", () => Effect.gen(function* () { const oversized = "x".repeat(4001); diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 4e8fd1ac5f30b..992d8078b3b4c 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -2050,6 +2050,11 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( const goalStops = new Map(); // Goals paused because a delegated task will wake the thread; the next ordinary turn resumes them. const goalsPausedForWake = new Set(); + // Running turns Codex started with no owning run, such as an active goal's turn on thread/resume. + const unownedTurns = new Map< + string, + { readonly nativeTurnId: string; readonly startedAt: DateTime.Utc } + >(); const latestRunTurnId = (providerTurnId: ProviderTurnId) => continuedRuns.get(providerTurnId)?.at(-1)?.providerTurnId ?? providerTurnId; const isContinuedRunDescendant = ( @@ -4934,6 +4939,10 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( ); return; } + unownedTurns.set(payload.threadId, { + nativeTurnId: payload.turn.id, + startedAt: codexTimestamp(payload.turn.startedAt), + }); yield* rememberSubagentTurnStarted({ nativeThreadId: payload.threadId, nativeTurnId: payload.turn.id, @@ -7009,7 +7018,15 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( yield* restoreAdditionalContext(payload.threadId).pipe( Effect.forkIn(scope, { startImmediately: true }), ); - const context = (yield* Ref.get(activeTurns)).get(payload.turn.id); + // Serialized with /goal resume adopting the turn, so it cannot miss this completion. + const context = yield* turnTerminalizationPermit.withPermits(1)( + Effect.suspend(() => { + if (unownedTurns.get(payload.threadId)?.nativeTurnId === payload.turn.id) { + unownedTurns.delete(payload.threadId); + } + return Ref.get(activeTurns).pipe(Effect.map((turns) => turns.get(payload.turn.id))); + }), + ); if (context === undefined) { return; } @@ -7233,6 +7250,17 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( ); }); + const clearPendingRootTurn = ( + threadId: string, + turnInput: ProviderAdapter.ProviderAdapterV2TurnInput, + ) => + Ref.update(pendingRootTurns, (current) => { + if (current.get(threadId) !== turnInput) return current; + const updated = new Map(current); + updated.delete(threadId); + return updated; + }); + const startNativeTurn = Effect.fn("CodexAdapterV2.startNativeTurn")( function* ( turnInput: ProviderAdapter.ProviderAdapterV2TurnInput, @@ -7304,12 +7332,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( effect.pipe( Effect.ensuring( Effect.flatMap(getNativeThreadId(turnInput.providerThread), (threadId) => - Ref.update(pendingRootTurns, (current) => { - if (current.get(threadId) !== turnInput) return current; - const updated = new Map(current); - updated.delete(threadId); - return updated; - }), + clearPendingRootTurn(threadId, turnInput), ).pipe(Effect.ignore), ), ), @@ -7369,6 +7392,12 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( if (command.type === "set") yield* validateCodexGoalObjective(command.objective); const threadId = yield* getNativeThreadId(turnInput.providerThread); rootProviderThreads.set(threadId, turnInput.providerThread); + // Codex starts an active goal's turn on its own (on thread/resume); bind it to this run. + if (command.type === "resume") { + yield* Ref.update(pendingRootTurns, (current) => + current.has(threadId) ? current : new Map(current).set(threadId, turnInput), + ); + } const readGoal = client .request("thread/goal/get", { threadId }) .pipe( @@ -7412,6 +7441,24 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( }); goalsByNativeThread.set(threadId, providerGoalFromCodex(goal)); } + // Codex rejects an empty turn/start while its own goal turn runs, so own that turn instead. + if (command.type === "resume" && current?.status === "active") { + const adopted = yield* turnTerminalizationPermit.withPermits(1)( + Effect.gen(function* () { + yield* clearPendingRootTurn(threadId, turnInput); + const owned = Array.from((yield* Ref.get(activeTurns)).values()).some( + (context) => context.input.attemptId === turnInput.attemptId, + ); + if (owned) return true; + const running = unownedTurns.get(threadId); + if (running === undefined) return false; + unownedTurns.delete(threadId); + yield* registerRootTurn({ turnInput, ...running }); + return true; + }), + ); + if (adopted) return; + } // A bare objective gets echoed back, and "start this goal" gets a // create_goal call that Codex rejects because the goal already exists. yield* activateGoal( @@ -7433,9 +7480,12 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( Effect.ensuring( Effect.suspend(() => { const threadId = turnInput.providerThread.nativeThreadRef?.nativeId; - return threadId == null || !goalUpdatesHeld.has(threadId) - ? Effect.void - : releaseGoalUpdates(threadId); + if (threadId == null) return Effect.void; + return clearPendingRootTurn(threadId, turnInput).pipe( + Effect.andThen( + goalUpdatesHeld.has(threadId) ? releaseGoalUpdates(threadId) : Effect.void, + ), + ); }), ), Effect.mapError(