diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index f8c63587fec6f..d446637cbadd7 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -9093,6 +9093,500 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); + describe("/goal resume", () => { + const turnStarted = (id: string) => + notification(`${id} started`, "turn/started", { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id, status: "inProgress" }), + }); + const turnCompleted = (id: string) => + notification(`${id} completed`, "turn/completed", { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id, status: "completed" }), + }); + const goalUpdated = (status: string, turnId: string | null) => + notification(`goal ${status}`, "thread/goal/updated", { + threadId: nativeThreadId, + turnId, + goal: codexGoal(status, 10), + }); + const resumeRequests = (goalBefore: string, goalAfter: string) => [ + ...request( + 3, + "thread/goal/get", + { threadId: nativeThreadId }, + { goal: codexGoal(goalBefore, 10) }, + ), + ...request( + 4, + "thread/settings/update", + { + threadId: nativeThreadId, + cwd: "/workspace", + model: "gpt-5.4", + approvalPolicy: "never", + approvalsReviewer: "user", + sandboxPolicy: { type: "dangerFullAccess" }, + summary: "detailed", + collaborationMode: { mode: "default", settings: { model: "gpt-5.4" } }, + }, + {}, + ), + ...request( + 5, + "thread/goal/set", + { threadId: nativeThreadId, status: "active" }, + { goal: codexGoal(goalAfter, 10) }, + ), + ]; + const runningTurns = (events: ReadonlyArray) => + events.flatMap((event) => + event.type === "provider_turn.updated" && event.providerTurn.status === "running" + ? [[event.providerTurn.nativeTurnRef?.nativeId ?? null, event.providerTurn.ordinal]] + : [], + ); + + const runResume = (scenario: string, entries: Array) => + Effect.gen(function* () { + const requests: Array = []; + const harness = yield* makeCodexReplayHarness( + makeCodexReplayTranscript({ scenario, entries: [...sessionStart, ...entries] }), + () => Effect.void, + (method) => + Effect.sync(() => { + requests.push(method); + }), + ); + yield* harness.runtime.startTurn(yield* goalTurnInput(harness, "/goal resume")); + yield* harness.firstTerminal; + return { harness, requests }; + }); + const provideServices = Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ); + + it.effect("attaches the run to the turn Codex already started", () => + Effect.gen(function* () { + const { harness, requests } = yield* runResume("goal-resume-started-turn", [ + turnStarted("resumed-turn"), + ...resumeRequests("blocked", "active"), + goalUpdated("active", "resumed-turn"), + turnCompleted("resumed-turn"), + goalUpdated("complete", null), + ]); + + assert.notInclude(requests, "turn/start"); + assert.deepEqual(runningTurns(harness.events), [["resumed-turn", 1]]); + const providerTurnId = (yield* IdAllocator.IdAllocatorV2).derive.providerTurn({ + driver: CodexAdapterV2.CODEX_DRIVER_KIND, + nativeTurnId: "resumed-turn", + }); + assert.deepEqual( + harness.terminalEvents().map((event) => [event.providerTurnId, event.status]), + [[providerTurnId, "completed"]], + ); + assert.include(providerGoals(harness.events), "active"); + }).pipe(Effect.scoped, provideServices), + ); + + it.effect("adopts a native goal turn started while goal/get is pending", () => + Effect.gen(function* () { + const getPending = yield* Deferred.make(); + const releaseGet = yield* Deferred.make(); + const startObserved = yield* Deferred.make(); + const requests: Array = []; + const resume = resumeRequests("active", "active"); + const harness = yield* makeCodexReplayHarness( + makeCodexReplayTranscript({ + scenario: "goal-resume-pending-get", + entries: [ + ...sessionStart, + ...resume.slice(0, 1), + goalUpdated("active", null), + turnStarted("pending-get-turn"), + notification("native start observed", "thread/goal/updated", { + threadId: nativeThreadId, + turnId: "pending-get-turn", + goal: codexGoal("active", 11), + }), + ...resume.slice(1), + goalUpdated("complete", "pending-get-turn"), + turnCompleted("pending-get-turn"), + ], + }), + (event) => + event.type === "provider_thread.updated" && + event.providerThread.goal?.tokensUsed === 11 + ? Deferred.succeed(startObserved, undefined) + : Effect.void, + (method) => + Effect.sync(() => { + requests.push(method); + }), + undefined, + (entry) => + entry.label === "thread/goal/get" + ? Deferred.succeed(getPending, undefined).pipe( + Effect.andThen(Deferred.await(releaseGet)), + ) + : Effect.void, + ); + const start = yield* harness.runtime + .startTurn(yield* goalTurnInput(harness, "/goal resume")) + .pipe(Effect.forkChild({ startImmediately: true })); + yield* Deferred.await(getPending); + yield* Deferred.await(startObserved); + assert.notInclude(requests, "thread/goal/set"); + assert.notInclude(requests, "turn/interrupt"); + yield* Deferred.succeed(releaseGet, undefined); + yield* Fiber.join(start); + yield* harness.firstTerminal; + assert.deepEqual(runningTurns(harness.events), [["pending-get-turn", 1]]); + assert.notInclude(requests, "turn/start"); + assert.notInclude(requests, "turn/interrupt"); + assert.deepEqual( + harness.terminalEvents().map((event) => event.status), + ["completed"], + ); + }).pipe(Effect.scoped, provideServices), + ); + + it.effect("cancels a pending resume on Stop before claiming native work", () => + Effect.gen(function* () { + const getPending = yield* Deferred.make(); + const requests: Array = []; + const resume = resumeRequests("active", "active"); + const harness = yield* makeCodexReplayHarness( + makeCodexReplayTranscript({ + scenario: "goal-resume-pending-stop", + entries: [ + ...sessionStart, + ...resume.slice(0, 1), + goalUpdated("active", null), + ...request( + 4, + "thread/goal/set", + { threadId: nativeThreadId, status: "paused" }, + { goal: codexGoal("paused", 10) }, + ), + ...resume.slice(1, 2), + ], + }), + (event) => + event.type === "provider_thread.updated" && + event.providerThread.goal?.status === "active" + ? Deferred.succeed(getPending, undefined) + : Effect.void, + (method) => + Effect.sync(() => { + requests.push(method); + }), + ); + const start = yield* harness.runtime + .startTurn(yield* goalTurnInput(harness, "/goal resume")) + .pipe(Effect.forkChild({ startImmediately: true })); + yield* Deferred.await(getPending); + yield* harness.runtime.interruptTurn({ + providerThread: harness.providerThread, + providerTurnId: ProviderTurnId.make("pending-goal-turn"), + }); + yield* Fiber.join(start); + yield* harness.firstTerminal; + assert.deepEqual(requests.slice(2), ["thread/goal/get", "thread/goal/set"]); + assert.deepEqual( + harness.terminalEvents().map((event) => event.status), + ["interrupted"], + ); + }).pipe(Effect.scoped, provideServices), + ); + + it.effect("cleans up resume failures without success or orphaned native work", () => + Effect.gen(function* () { + for (const { stage, ownership, cleanup } of [ + { stage: "settings", ownership: "unclaimed", cleanup: "ok" }, + { stage: "read", ownership: "unclaimed", cleanup: "pause-error" }, + { stage: "activation", ownership: "held", cleanup: "ok" }, + { stage: "activation", ownership: "adopted", cleanup: "interrupt-error" }, + { stage: "post-stop-pause", ownership: "held", cleanup: "ok" }, + { stage: "stopped", ownership: "held", cleanup: "ok" }, + { stage: "stopped", ownership: "adopted", cleanup: "continues-during-pause" }, + { stage: "pending-stop", ownership: "unclaimed", cleanup: "ok" }, + { stage: "activated-stop", ownership: "adopted", cleanup: "ok" }, + { stage: "activated-stop", ownership: "continued", cleanup: "ok" }, + ]) { + yield* Effect.gen(function* () { + const requests: Array = []; + const activationPending = yield* Deferred.make(); + const stopping = + stage === "post-stop-pause" || + stage === "stopped" || + stage === "pending-stop" || + stage === "activated-stop"; + const rejected = (id: number, message = "Resume rejected") => + ({ + type: "emit_inbound", + label: message, + frame: { id, error: { code: -32603, message } }, + }) satisfies CodexReplay.CodexAppServerReplayEntry; + const resume = resumeRequests("paused", "active"); + const failureId = + stage === "read" || stage === "pending-stop" ? 3 : stage === "settings" ? 4 : 5; + const pauseId = failureId + 1; + const nativeId = "failed-resume-turn"; + const nativeRunning = ownership !== "held"; + const interruptNativeId = + cleanup === "continues-during-pause" ? `${nativeId}-next` : nativeId; + const interruptsNative = nativeRunning && cleanup !== "completes-during-pause"; + const postStopPauseId = interruptsNative ? 8 : 7; + const interruptId = pauseId + 1; + const interrupt = interruptsNative + ? [ + ...request( + interruptId, + "turn/interrupt", + { threadId: nativeThreadId, turnId: interruptNativeId }, + {}, + ).slice(0, 1), + ...(cleanup === "interrupt-error" + ? [rejected(interruptId, "Interrupt rejected")] + : [ + ...request(interruptId, "turn/interrupt", {}, {}).slice(1), + turnCompleted(interruptNativeId), + ]), + ] + : []; + const pause = request( + pauseId, + "thread/goal/set", + { threadId: nativeThreadId, status: "paused" }, + { goal: codexGoal("paused", 10) }, + ); + const entries = [...sessionStart]; + if (ownership === "unclaimed" || ownership === "adopted") { + entries.push(goalUpdated("active", nativeId), turnStarted(nativeId)); + } + if (stage === "activated-stop") entries.push(...resume); + else entries.push(...resume.slice(0, (failureId - 3) * 2 + 1)); + if (ownership === "continued") { + entries.push(goalUpdated("active", nativeId), turnStarted(nativeId)); + } + if (stopping) { + if (stage !== "activated-stop") + entries.push( + notification("activation pending", "thread/goal/updated", { + threadId: nativeThreadId, + turnId: null, + goal: codexGoal("active", 11), + }), + ); + entries.push(...pause.slice(0, 1)); + if (cleanup === "completes-during-pause" || cleanup === "continues-during-pause") { + entries.push(goalUpdated("paused", nativeId), turnCompleted(nativeId)); + } + if (cleanup === "continues-during-pause") + entries.push(turnStarted(interruptNativeId)); + entries.push(...pause.slice(1), ...interrupt); + if (stage === "pending-stop") { + entries.push(...resume.slice(1, 2)); + } else if (stage !== "activated-stop") { + entries.push( + ...resume.slice(5), + ...withReplayRequestId(pause.slice(0, 1), postStopPauseId), + ...(stage === "stopped" + ? withReplayRequestId(pause.slice(1), postStopPauseId) + : [rejected(postStopPauseId)]), + ); + if (stage === "post-stop-pause") + entries.push(...withReplayRequestId(pause, postStopPauseId + 1)); + } + } else { + entries.push(rejected(failureId)); + entries.push(...pause.slice(0, 1)); + if (cleanup === "completes-during-pause") { + entries.push(goalUpdated("paused", nativeId), turnCompleted(nativeId)); + } + entries.push( + ...(cleanup === "pause-error" + ? [rejected(pauseId, "Pause rejected")] + : pause.slice(1)), + ...interrupt, + ); + } + const harness = yield* makeCodexReplayHarness( + makeCodexReplayTranscript({ + scenario: `goal-resume-failed-${stage}-${ownership}-${cleanup}`, + entries, + }), + (event) => + stopping && + event.type === "provider_thread.updated" && + event.providerThread.goal?.tokensUsed === 11 + ? Deferred.succeed(activationPending, undefined) + : Effect.void, + (method) => + Effect.sync(() => { + requests.push(method); + }), + ); + const turnInput = yield* goalTurnInput(harness, "/goal resume"); + const start = yield* harness.runtime + .startTurn(turnInput) + .pipe(Effect.result, Effect.forkChild({ startImmediately: true })); + if (stopping) { + // Leave ingestion out of target selection: production can still name + // earlier background work after startTurn has successfully returned. + if (stage === "activated-stop") yield* Fiber.join(start); + else yield* Deferred.await(activationPending); + yield* harness.runtime.interruptTurn({ + providerThread: harness.providerThread, + providerTurnId: + stage !== "post-stop-pause" + ? ProviderTurnId.make("earlier-goal-turn") + : (yield* IdAllocator.IdAllocatorV2).derive.providerTurn({ + driver: CodexAdapterV2.CODEX_DRIVER_KIND, + nativeTurnId: `goal-command:${turnInput.attemptId}`, + }), + requestRuntimeRestart: true, + }); + if (interruptsNative) { + assert.include(requests, "turn/interrupt", `${stage}-${ownership}`); + } + } + const result = yield* Fiber.join(start); + assert.equal( + result._tag, + stopping && stage !== "post-stop-pause" ? "Success" : "Failure", + stage, + ); + if (result._tag === "Failure") { + const error = result.failure; + assert.instanceOf(error, ProviderAdapter.ProviderAdapterTurnStartError, stage); + assert.include(errorCauseChainText(error), "Resume rejected", stage); + } + yield* TestClock.adjust("6 seconds"); + assert.deepEqual( + harness.terminalEvents().map((event) => event.status), + ownership === "unclaimed" && !stopping ? [] : ["interrupted"], + stage, + ); + if (ownership === "unclaimed" && !stopping) { + assert.deepEqual(runningTurns(harness.events), [], stage); + } else { + for (const event of harness.events) { + if ( + event.type !== "provider_turn.updated" || + event.providerTurn.status !== "running" + ) + continue; + assert.isTrue( + harness.events.some( + (candidate) => + candidate.type === "provider_turn.updated" && + candidate.providerTurn.id === event.providerTurn.id && + candidate.providerTurn.status !== "running", + ), + stage, + ); + } + } + assert.equal( + requests.filter((method) => method === "turn/interrupt").length, + interruptsNative ? 1 : 0, + `${stage}-${ownership}-${cleanup}`, + ); + assert.notInclude(requests, "turn/start", stage); + }).pipe(Effect.scoped); + } + }).pipe(provideServices), + ); + + it.effect("settles the run when Codex leaves the goal inactive", () => + Effect.gen(function* () { + const { harness, requests } = yield* runResume("goal-resume-inactive", [ + ...resumeRequests("budgetLimited", "budgetLimited"), + ]); + + assert.notInclude(requests, "turn/start"); + assert.deepEqual( + harness.terminalEvents().map((event) => event.status), + ["completed"], + ); + assert.deepEqual(runningTurns(harness.events), [[null, 1]]); + }).pipe(Effect.scoped, provideServices), + ); + + it.effect("waits through slow activation and adopts the next native goal turn", () => + Effect.gen(function* () { + const activationRequested = yield* Deferred.make(); + const activate = yield* Deferred.make(); + const requests: Array = []; + const harness = yield* makeCodexReplayHarness( + makeCodexReplayTranscript({ + scenario: "goal-resume-slow-activation", + entries: [ + ...sessionStart, + turnStarted("finished-turn"), + turnCompleted("finished-turn"), + ...resumeRequests("paused", "active").slice(0, 4), + goalUpdated("paused", null), + ...resumeRequests("paused", "active").slice(4), + turnStarted("continuation"), + goalUpdated("active", "continuation"), + turnCompleted("continuation"), + goalUpdated("complete", null), + ], + }), + () => Effect.void, + (method) => + Effect.sync(() => { + requests.push(method); + }).pipe( + Effect.andThen( + method === "thread/goal/set" + ? Deferred.succeed(activationRequested, undefined).pipe( + Effect.andThen(Deferred.await(activate)), + ) + : Effect.void, + ), + ), + ); + const start = yield* harness.runtime + .startTurn(yield* goalTurnInput(harness, "/goal resume")) + .pipe(Effect.forkChild({ startImmediately: true })); + yield* Deferred.await(activationRequested); + + // The continuation grace is for Codex to start a turn once the goal is + // active, so it must not run out while the activation is in flight. + yield* TestClock.adjust("6 seconds"); + assert.lengthOf(harness.terminalEvents(), 0); + + yield* Deferred.succeed(activate, undefined); + yield* Fiber.join(start); + yield* harness.firstTerminal; + + const providerTurnId = (yield* IdAllocator.IdAllocatorV2).derive.providerTurn({ + driver: CodexAdapterV2.CODEX_DRIVER_KIND, + nativeTurnId: "continuation", + }); + assert.deepEqual( + harness.terminalEvents().map((event) => [event.providerTurnId, event.status]), + [[providerTurnId, "completed"]], + ); + assert.notInclude(requests, "turn/start"); + assert.notInclude(requests, "turn/interrupt"); + assert.deepEqual(runningTurns(harness.events), [ + [null, 1], + ["continuation", 2], + ]); + }).pipe(Effect.scoped, provideServices), + ); + }); + it.effect("settles a run held for the next goal turn when Stop arrives in between", () => Effect.gen(function* () { const prompt = "Keep going"; diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 6953636a8dcf5..f57ddcb42251d 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -2011,9 +2011,39 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( string, { readonly context: ActiveCodexTurnContext; readonly retry: ActiveCodexProviderRetry } >(); - const goalActivations = new Map(); + const goalActivations = new Map< + string, + { + stopped: boolean; + readonly resumeInput: ProviderAdapter.ProviderAdapterV2TurnInput | undefined; + // Stop can name earlier work before the claimed turn is projected. + resumeTurnId: ProviderTurnId | undefined; + } + >(); + // Codex can start a turn on its own while no run exists, such as when it + // resumes a goal with the thread. `/goal resume` adopts it. + const unownedRootTurns = new Map< + string, + { readonly nativeTurnId: string; readonly startedAt: DateTime.Utc } + >(); // Held runs must not complete while Stop resolves its target. const goalStops = new Map(); + const pauseGoal = (threadId: string) => + client.request("thread/goal/set", { threadId, status: "paused" }).pipe( + Effect.tap(({ goal }) => + Effect.sync(() => goalsByNativeThread.set(threadId, providerGoalFromCodex(goal))), + ), + Effect.catch((cause) => + Effect.logWarning("orchestration-v2.codex-goal-pause-failed", { cause }), + ), + ); + const stopGoalTurnWithoutRun = (threadId: string, turnId: string) => + pauseGoal(threadId).pipe( + Effect.andThen(client.request("turn/interrupt", { threadId, turnId })), + Effect.catch((cause) => + Effect.logWarning("orchestration-v2.codex-goal-turn-stop-failed", { cause }), + ), + ); const latestRunTurnId = (providerTurnId: ProviderTurnId) => continuedRuns.get(providerTurnId)?.at(-1)?.providerTurnId ?? providerTurnId; const isContinuedRunDescendant = ( @@ -4780,38 +4810,29 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( return; } // Pause orphan goal turns so they cannot work without an owning run. + const activation = goalActivations.get(payload.threadId); if ( rootProviderThreads.has(payload.threadId) && - (goalsByNativeThread.get(payload.threadId)?.status === "active" || - goalStops.has(payload.threadId)) + (goalStops.has(payload.threadId) || + activation?.stopped === true || + ((activation?.resumeInput === undefined || activation.resumeTurnId !== undefined) && + goalsByNativeThread.get(payload.threadId)?.status === "active")) ) { yield* Effect.logWarning("orchestration-v2.codex-goal-turn-without-run", { nativeThreadId: payload.threadId, nativeTurnId: payload.turn.id, }); - yield* client - .request("thread/goal/set", { threadId: payload.threadId, status: "paused" }) - .pipe( - Effect.catch((cause) => - Effect.logWarning("orchestration-v2.codex-goal-pause-failed", { cause }), - ), - Effect.andThen( - client.request("turn/interrupt", { - threadId: payload.threadId, - turnId: payload.turn.id, - }), - ), - Effect.catch((cause) => - Effect.logWarning("orchestration-v2.codex-goal-turn-stop-failed", { cause }), - ), - Effect.forkIn(scope), - ); + yield* stopGoalTurnWithoutRun(payload.threadId, payload.turn.id).pipe( + Effect.forkIn(scope), + ); return; } + const startedAt = codexTimestamp(payload.turn.startedAt); + unownedRootTurns.set(payload.threadId, { nativeTurnId: payload.turn.id, startedAt }); yield* rememberSubagentTurnStarted({ nativeThreadId: payload.threadId, nativeTurnId: payload.turn.id, - startedAt: codexTimestamp(payload.turn.startedAt), + startedAt, }); }).pipe(Effect.orDie, turnTerminalizationPermit.withPermits(1)), ); @@ -6066,6 +6087,16 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( } }); + const startGoalHoldGrace = (nativeThreadId: string, event: CodexRootTerminalEvent) => + Effect.sleep(CODEX_GOAL_CONTINUATION_GRACE).pipe( + Effect.andThen(pauseUncontinuedGoal(nativeThreadId, event)), + Effect.andThen( + turnTerminalizationPermit.withPermits(1)(releaseGoalHold(nativeThreadId, event)), + ), + // The grace window starts when the turn completes, not when the fiber is scheduled. + Effect.forkIn(scope, { startImmediately: true }), + ); + const emitRootTerminal = Effect.fnUntraced(function* ( context: ActiveCodexTurnContext, event: CodexRootTerminalEvent, @@ -6146,14 +6177,11 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( completedTurn: input.goalHoldTurn, next: yield* Deferred.make(), }); - yield* Effect.sleep(CODEX_GOAL_CONTINUATION_GRACE).pipe( - Effect.andThen(pauseUncontinuedGoal(nativeThreadId, event)), - Effect.andThen( - turnTerminalizationPermit.withPermits(1)(releaseGoalHold(nativeThreadId, event)), - ), - // The grace window starts when the turn completes, not when the fiber is scheduled. - Effect.forkIn(scope, { startImmediately: true }), - ); + // The grace is for Codex to start a turn once the goal is active, + // so an activation in flight starts it when it finishes. + if (!goalActivations.has(nativeThreadId)) { + yield* startGoalHoldGrace(nativeThreadId, event); + } return; } yield* emitRootTerminal(input.context, event); @@ -6233,7 +6261,12 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( expected?: CodexRootTerminalEvent, ) { const hold = goalHolds.get(nativeThreadId); - if (hold === undefined || goalStops.has(nativeThreadId)) return; + if ( + hold === undefined || + goalStops.has(nativeThreadId) || + goalActivations.has(nativeThreadId) + ) + return; if (expected !== undefined && hold.event !== expected) return; goalHolds.delete(nativeThreadId); yield* settleGoalHold(hold, "completed"); @@ -6387,12 +6420,15 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( const nativeThreadId = input.context.providerThread.nativeThreadRef?.nativeId; const activation = nativeThreadId == null ? undefined : goalActivations.get(nativeThreadId); + // A pause can arrive before Stop selects the completed turn. const holdsForGoal = input.context.subagent === null && input.status === "completed" && nativeThreadId != null && (goalsByNativeThread.get(nativeThreadId)?.status === "active" || - (activation !== undefined && !activation.stopped)); + activation !== undefined || + (goalsByNativeThread.get(nativeThreadId) != null && + goalStops.has(nativeThreadId))); const completedTurn: OrchestrationV2ProviderTurn = { id: input.context.providerTurnId, providerThreadId: input.context.providerThread.id, @@ -6773,6 +6809,9 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( yield* client.handleServerNotification("turn/completed", (payload) => Effect.gen(function* () { + if (unownedRootTurns.get(payload.threadId)?.nativeTurnId === payload.turn.id) { + unownedRootTurns.delete(payload.threadId); + } const context = (yield* Ref.get(activeTurns)).get(payload.turn.id); if (context === undefined) { return; @@ -6801,10 +6840,15 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( }), ); - /** Settles a /goal turn that never reaches Codex. State changes reply through the goal row. */ + /** + * Settles a /goal turn that never reaches Codex. State changes reply through the goal row. + * With `holdForGoalTurn`, call with the terminalization permit: the run + * stays open for the turn Codex starts once the goal is active. + */ const completeGoalCommandTurn = Effect.fnUntraced(function* ( turnInput: ProviderAdapter.ProviderAdapterV2TurnInput, reply: string | null, + holdForGoalTurn = false, ) { const now = yield* DateTime.now; const nativeId = `goal-command:${turnInput.attemptId}`; @@ -6870,11 +6914,21 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( turnItem: artifacts.turnItem, }); } + const completedTurn = { ...providerTurn, status: "completed", completedAt: now } as const; + if (holdForGoalTurn) { + yield* emitOrDeferRootTerminal({ + context, + nativeTurnId: nativeId, + status: "completed", + goalHoldTurn: completedTurn, + }); + return context.providerTurnId; + } yield* emitProviderEvent({ type: "provider_turn.updated", driver: CODEX_PROVIDER, threadId: turnInput.threadId, - providerTurn: { ...providerTurn, status: "completed", completedAt: now }, + providerTurn: completedTurn, }); yield* emitRootTerminal(context, { type: "turn.terminal", @@ -6886,6 +6940,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( failure: null, threadDisposition: "reusable", }); + return context.providerTurnId; }); const heldGoalTurn = ( @@ -6910,34 +6965,51 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( goalStops.set(goalThreadId, (goalStops.get(goalThreadId) ?? 0) + 1); return yield* Effect.gen(function* () { // The run can settle while the pause is in flight, which drops its record. - const requestedTurnId = latestRunTurnId(turnInput.providerTurnId); + const activation = goalActivations.get(goalThreadId); + if (activation !== undefined) activation.stopped = true; + const resumeInput = activation?.resumeInput; + // Thread-wide Stop can name earlier background work before ingestion + // catches up, including after activation has finished. + const currentGoalTurn = + turnInput.requestRuntimeRestart === true && + goalsByNativeThread.get(goalThreadId) != null + ? (goalHolds.get(goalThreadId)?.context ?? + (yield* findActiveTurnByNativeThreadId(goalThreadId))) + : undefined; + const requestedTurnId = + resumeInput === undefined + ? latestRunTurnId( + currentGoalTurn?.subagent === null && + currentGoalTurn.providerThread.id === turnInput.providerThread.id + ? currentGoalTurn.providerTurnId + : turnInput.providerTurnId, + ) + : ((yield* claimGoalTurn(resumeInput, true)) ?? turnInput.providerTurnId); const heldBefore = goalHolds.get(goalThreadId)?.context; const turnsBefore = [ + ...(continuedRuns.get(turnInput.providerTurnId) ?? []), ...(continuedRuns.get(requestedTurnId) ?? []), ...Array.from((yield* Ref.get(activeTurns)).values()).filter( - (context) => context.providerTurnId === requestedTurnId, + (context) => + context.providerTurnId === requestedTurnId || + context.providerTurnId === turnInput.providerTurnId, ), + ...(turnInput.requestRuntimeRestart === true + ? Array.from((yield* Ref.get(settledTurns)).values()).filter( + (context) => context.providerTurnId === turnInput.providerTurnId, + ) + : []), ...(heldBefore?.providerTurnId === requestedTurnId ? [heldBefore] : []), ]; - const activation = goalActivations.get(goalThreadId); - if (activation !== undefined) activation.stopped = true; - if (goalsByNativeThread.get(goalThreadId)?.status === "active") { - yield* client - .request("thread/goal/set", { threadId: goalThreadId, status: "paused" }) - .pipe( - Effect.tap(({ goal }) => - Effect.sync(() => - goalsByNativeThread.set(goalThreadId, providerGoalFromCodex(goal)), - ), - ), - Effect.catch((cause) => - Effect.logWarning("orchestration-v2.codex-goal-pause-failed", { cause }), - ), - ); + if ( + activation !== undefined || + goalsByNativeThread.get(goalThreadId)?.status === "active" + ) { + yield* pauseGoal(goalThreadId); } return yield* turnTerminalizationPermit.withPermits(1)( Effect.gen(function* () { - const providerTurnId = latestRunTurnId(turnInput.providerTurnId); + const providerTurnId = latestRunTurnId(requestedTurnId); const hold = goalHolds.get(goalThreadId); const held = hold !== undefined && hold.context.providerTurnId === providerTurnId; const goalTurns: ReadonlyArray = Array.from( @@ -6967,6 +7039,25 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( ); }); + const prepareNativeTurn = Effect.fn("CodexAdapterV2.prepareNativeTurn")(function* ( + turnInput: ProviderAdapter.ProviderAdapterV2TurnInput, + codexInput: ReadonlyArray, + ) { + const threadId = yield* getNativeThreadId(turnInput.providerThread); + const mcpSession = yield* mcpSessions.read(turnInput.threadId); + return yield* buildCodexTurnStartParams({ + nativeThreadId: threadId, + codexInput, + runtimePolicy: turnInput.runtimePolicy, + modelSelection: turnInput.modelSelection, + hasT3Mcp: mcpSession !== undefined, + browserToolsAvailable: mcpSession?.browserToolsAvailable ?? true, + deviceToolsAvailable: mcpSession?.capabilities?.has("device") ?? false, + omitServiceTier: adapterOptions.resolveRuntime !== undefined, + ...(turnInput.appContext === undefined ? {} : { appContext: turnInput.appContext }), + }); + }); + // Codex refuses to resume a session archived elsewhere until it is unarchived. const resumeNativeThread = Effect.fn("CodexAdapterV2.resumeNativeThread")(function* ( threadId: string, @@ -6991,18 +7082,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( codexInput: ReadonlyArray, ) { const threadId = yield* getNativeThreadId(turnInput.providerThread); - const mcpSession = yield* mcpSessions.read(turnInput.threadId); - const turnStartParams = yield* buildCodexTurnStartParams({ - nativeThreadId: threadId, - codexInput, - runtimePolicy: turnInput.runtimePolicy, - modelSelection: turnInput.modelSelection, - hasT3Mcp: mcpSession !== undefined, - browserToolsAvailable: mcpSession?.browserToolsAvailable ?? true, - deviceToolsAvailable: mcpSession?.capabilities?.has("device") ?? false, - omitServiceTier: adapterOptions.resolveRuntime !== undefined, - ...(turnInput.appContext === undefined ? {} : { appContext: turnInput.appContext }), - }); + const turnStartParams = yield* prepareNativeTurn(turnInput, codexInput); yield* Ref.update(pendingRootTurns, (current) => { const updated = new Map(current); updated.set(threadId, turnInput); @@ -7059,9 +7139,34 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( ); /** - * Setting or resuming a goal starts the first goal turn with this run's - * turn configuration (approvals, sandbox, model, plan mode), since a - * turn Codex starts on its own reuses the last turn's settings. + * A resumed thread can already be running a goal turn, which rejects an + * empty turn/start. Adopt that turn or hold for native goal activation. + */ + const claimGoalTurn = Effect.fn("CodexAdapterV2.claimGoalTurn")( + function* (turnInput: ProviderAdapter.ProviderAdapterV2TurnInput, stopping = false) { + const threadId = yield* getNativeThreadId(turnInput.providerThread); + const activation = goalActivations.get(threadId); + if (activation?.resumeInput !== turnInput || (activation.stopped && !stopping)) return; + if (activation.resumeTurnId !== undefined) { + return latestRunTurnId(activation.resumeTurnId); + } + const started = unownedRootTurns.get(threadId); + if (started === undefined) { + const providerTurnId = yield* completeGoalCommandTurn(turnInput, null, true); + activation.resumeTurnId = providerTurnId; + return providerTurnId; + } + const context = yield* registerRootTurn({ turnInput, ...started }); + activation.resumeTurnId = context.providerTurnId; + unownedRootTurns.delete(threadId); + return context.providerTurnId; + }, + (effect) => turnTerminalizationPermit.withPermits(1)(effect), + ); + + /** + * Setting a goal sends its objective as input; resuming lets Codex + * continue the existing objective without another user turn. */ const runGoalCommand = Effect.fn("CodexAdapterV2.runGoalCommand")( function* ( @@ -7071,66 +7176,85 @@ 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); - const readGoal = client - .request("thread/goal/get", { threadId }) - .pipe( - Effect.map((response) => - response.goal == null ? null : providerGoalFromCodex(response.goal), - ), - ); - const current = yield* readGoal; - goalsByNativeThread.set(threadId, current); - if (command.type === "show") { - return yield* completeGoalCommandTurn( - turnInput, - current === null ? "No goal is set." : describeCodexGoal(current), - ); - } - if (command.type === "clear") { - const { cleared } = yield* client.request("thread/goal/clear", { threadId }); - goalsByNativeThread.set(threadId, null); - return yield* completeGoalCommandTurn(turnInput, cleared ? null : "No goal is set."); - } - if (current === null && command.type !== "set") { - return yield* completeGoalCommandTurn(turnInput, "No goal is set."); - } - if (command.type === "pause") { - const { goal } = yield* client.request("thread/goal/set", { - threadId, - status: "paused", - }); - goalsByNativeThread.set(threadId, providerGoalFromCodex(goal)); - return yield* completeGoalCommandTurn(turnInput, null); - } - if (command.type === "set") { - goalUpdatesHeld.add(threadId); - if (current !== null) yield* client.request("thread/goal/clear", { threadId }); - // Keep paused until our turn starts so Codex cannot start one first. - const { goal } = yield* client.request("thread/goal/set", { - threadId, - objective: command.objective, - status: "paused", - }); - goalsByNativeThread.set(threadId, providerGoalFromCodex(goal)); - } - // Hold fast first turns until activation; Stop cancels activation. - const activation = { stopped: false }; - goalActivations.set(threadId, activation); + const activation = { + stopped: false, + resumeInput: command.type === "resume" ? turnInput : undefined, + resumeTurnId: undefined as ProviderTurnId | undefined, + }; + if (command.type === "resume") goalActivations.set(threadId, activation); yield* Effect.gen(function* () { - // A bare objective gets echoed back, and "start this goal" gets a - // create_goal call that Codex rejects because the goal already exists. - yield* startNativeTurn( - turnInput, - command.type === "set" - ? yield* toCodexInput({ - ...turnInput, - message: { - ...turnInput.message, - text: `This thread's goal is already set. Do not call create_goal. Start working on it now: ${command.objective}`, - }, - }) - : [], - ); + const readGoal = client + .request("thread/goal/get", { threadId }) + .pipe( + Effect.map((response) => + response.goal == null ? null : providerGoalFromCodex(response.goal), + ), + ); + const current = yield* readGoal; + if (activation.stopped) return; + goalsByNativeThread.set(threadId, current); + if (command.type === "show") { + return yield* completeGoalCommandTurn( + turnInput, + current === null ? "No goal is set." : describeCodexGoal(current), + ); + } + if (command.type === "clear") { + const { cleared } = yield* client.request("thread/goal/clear", { threadId }); + goalsByNativeThread.set(threadId, null); + return yield* completeGoalCommandTurn( + turnInput, + cleared ? null : "No goal is set.", + ); + } + if (current === null && command.type !== "set") { + return yield* completeGoalCommandTurn(turnInput, "No goal is set."); + } + if (command.type === "pause") { + const { goal } = yield* client.request("thread/goal/set", { + threadId, + status: "paused", + }); + goalsByNativeThread.set(threadId, providerGoalFromCodex(goal)); + return yield* completeGoalCommandTurn(turnInput, null); + } + if (command.type === "set") { + goalUpdatesHeld.add(threadId); + if (current !== null) yield* client.request("thread/goal/clear", { threadId }); + // Keep paused until our turn starts so Codex cannot start one first. + const { goal } = yield* client.request("thread/goal/set", { + threadId, + objective: command.objective, + status: "paused", + }); + goalsByNativeThread.set(threadId, providerGoalFromCodex(goal)); + } + // Hold fast first turns until activation; Stop cancels activation. + goalActivations.set(threadId, activation); + if (command.type === "set") { + // A bare objective gets echoed back, and "start this goal" gets a + // create_goal call that Codex rejects because the goal already exists. + yield* startNativeTurn( + turnInput, + yield* toCodexInput({ + ...turnInput, + message: { + ...turnInput.message, + text: `This thread's goal is already set. Do not call create_goal. Start working on it now: ${command.objective}`, + }, + }), + ); + } else { + const { + input: _input, + additionalContext: _context, + ...settings + } = yield* prepareNativeTurn(turnInput, []); + // The generated client lacks thread/settings/update. It updates + // future turns; a running turn keeps its native settings. + yield* client.raw.request("thread/settings/update", settings); + yield* claimGoalTurn(turnInput); + } if (activation.stopped) return; const { goal } = yield* client.request("thread/goal/set", { threadId, @@ -7145,14 +7269,65 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( goalsByNativeThread.set(threadId, providerGoalFromCodex(paused.goal)); } }).pipe( + Effect.onError(() => + Effect.gen(function* () { + if (command.type !== "resume" || goalActivations.get(threadId) !== activation) + return; + activation.stopped = true; + const started = unownedRootTurns.get(threadId); + unownedRootTurns.delete(threadId); + const context = yield* turnTerminalizationPermit.withPermits(1)( + Effect.gen(function* () { + const owned = [ + goalHolds.get(threadId)?.context, + ...(yield* Ref.get(activeTurns)).values(), + ]; + return owned.find( + (turn) => + turn?.subagent === null && turn.input.attemptId === turnInput.attemptId, + ); + }), + ); + if (context !== undefined) { + yield* runtime.interruptTurn({ + providerThread: turnInput.providerThread, + providerTurnId: context.providerTurnId, + }); + } else if (started !== undefined) { + yield* stopGoalTurnWithoutRun(threadId, started.nativeTurnId); + } else { + yield* pauseGoal(threadId); + } + }).pipe( + Effect.catchCause((cause) => + Effect.logWarning("orchestration-v2.codex-goal-resume-cleanup-failed", { + cause, + }), + ), + ), + ), Effect.ensuring( turnTerminalizationPermit.withPermits(1)( Effect.gen(function* () { if (goalActivations.get(threadId) === activation) { goalActivations.delete(threadId); } + if (activation.stopped) { + const hold = goalHolds.get(threadId); + if ( + command.type === "resume" && + hold?.context.input.attemptId === turnInput.attemptId + ) { + goalHolds.delete(threadId); + yield* settleGoalHold(hold, "interrupted"); + } + return; + } if (goalsByNativeThread.get(threadId)?.status !== "active") { yield* releaseGoalHold(threadId); + } else if (!goalActivations.has(threadId)) { + const hold = goalHolds.get(threadId); + if (hold !== undefined) yield* startGoalHoldGrace(threadId, hold.event); } }), ),