diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 09ec3034a9bc9..0fc1b9a2bf82d 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -9523,6 +9523,187 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); + it.effect.each([ + "Codex completes it later", + "its thread is unloaded", + "Codex completes it after the session's Stop deadline", + "Codex completes it after a second Stop fails", + ] as const)("keeps a turn open after an unconfirmed Stop until %s", (ending) => + Effect.gen(function* () { + const nativeThreadId = "unconfirmed-stop-thread"; + const nativeTurnId = "unconfirmed-stop-turn"; + const usage = { + totalTokens: 15, + inputTokens: 10, + cachedInputTokens: 2, + outputTokens: 5, + reasoningOutputTokens: 1, + }; + const pastDeadline = ending === "Codex completes it after the session's Stop deadline"; + const harness = yield* makeCodexReplayHarness( + makeCodexReplayTranscript({ + scenario: `unconfirmed-stop-${ending.length}`, + entries: [ + ...codexReplayPreamble({ nativeThreadId, nativeTurnId, prompt: "work" }), + { + type: "expect_outbound", + label: "turn/interrupt", + frame: { + id: 4, + method: "turn/interrupt", + params: { threadId: nativeThreadId, turnId: nativeTurnId }, + }, + }, + { + type: "emit_inbound", + label: "still working", + frame: { + method: "thread/tokenUsage/updated", + params: { + threadId: nativeThreadId, + turnId: nativeTurnId, + tokenUsage: { total: usage, last: usage, modelContextWindow: 200_000 }, + }, + }, + }, + { + type: "emit_inbound", + label: "turn/interrupt", + // A slow acknowledgement leaves the adapter's own wait running past the deadline. + ...(pastDeadline ? { afterMs: 6_000 } : {}), + frame: { id: 4, result: {} }, + }, + ...(ending === "its thread is unloaded" + ? ([ + { + type: "expect_outbound", + label: "thread/unsubscribe", + frame: { + id: 5, + method: "thread/unsubscribe", + params: { threadId: nativeThreadId }, + }, + }, + { + type: "emit_inbound", + label: "thread/unsubscribe", + frame: { id: 5, result: { status: "unsubscribed" } }, + }, + ] satisfies Array) + : ([ + ...(ending === "Codex completes it after a second Stop fails" + ? ([ + { + type: "expect_outbound", + label: "turn/interrupt/second", + frame: { + id: 5, + method: "turn/interrupt", + params: { threadId: nativeThreadId, turnId: nativeTurnId }, + }, + }, + { + type: "emit_inbound", + label: "turn/interrupt/second", + frame: { id: 5, error: { code: -32603, message: "internal error" } }, + }, + ] satisfies Array) + : []), + { + type: "emit_inbound", + label: "turn/completed", + afterMs: 20_000, + frame: { + method: "turn/completed", + params: { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: nativeTurnId, status: "completed" }), + }, + }, + }, + ] satisfies Array)), + ], + }), + ); + yield* harness.runtime.startTurn( + makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("attempt-unconfirmed-stop"), + text: "work", + }), + ); + const stopping = yield* harness.runtime + .interruptTurn({ + providerThread: harness.providerThread, + providerTurnId: (yield* IdAllocator.IdAllocatorV2).derive.providerTurn({ + driver: CodexAdapterV2.CODEX_DRIVER_KIND, + nativeTurnId, + }), + }) + .pipe( + // The session manager bounds every Stop the same way. + Effect.timeoutOrElse({ + duration: "15 seconds", + orElse: () => Effect.fail({ _tag: "SessionStopDeadline" } as const), + }), + Effect.result, + Effect.forkScoped, + ); + // Codex received the interrupt and kept reporting work on the turn. + yield* harness.until(() => + harness.events.some( + (event) => + event.type === "provider_turn.updated" && + event.providerTurn.tokenUsage?.usedTokens === usage.totalTokens, + ), + ); + yield* TestClock.adjust(pastDeadline ? "15 seconds" : "10 seconds"); + + const stopped = yield* Fiber.join(stopping); + assert.equal( + stopped._tag === "Failure" ? stopped.failure._tag : "confirmed", + pastDeadline ? "SessionStopDeadline" : "ProviderAdapterInterruptTimeoutError", + ); + assert.deepEqual(harness.terminalEvents(), []); + + if (ending === "Codex completes it after a second Stop fails") { + const second = yield* harness.runtime + .interruptTurn({ + providerThread: harness.providerThread, + providerTurnId: (yield* IdAllocator.IdAllocatorV2).derive.providerTurn({ + driver: CodexAdapterV2.CODEX_DRIVER_KIND, + nativeTurnId, + }), + }) + .pipe(Effect.result); + assert.equal(second._tag, "Failure"); + assert.deepEqual(harness.terminalEvents(), []); + } + if (ending === "its thread is unloaded") { + yield* harness.runtime.unloadThread!({ providerThread: harness.providerThread }); + } else { + yield* TestClock.adjust("20 seconds"); + } + yield* harness.awaitTerminal(1); + assert.deepEqual( + harness.terminalEvents().map((event) => event.status), + ["interrupted"], + ); + }).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 efa5d0a9c28f1..e26b3dc5860c2 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -89,6 +89,7 @@ import * as Digest from "effect/crypto/Digest"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; import * as Fiber from "effect/Fiber"; import * as FileSystem from "effect/FileSystem"; import * as Function from "effect/Function"; @@ -159,6 +160,8 @@ import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter" const CODEX_PROVIDER = ProviderDriverKind.make("codex"); const CODEX_CAPACITY_MAX_RETRIES = 5; +// How long Stop waits for Codex to end the turn before the user is offered Force stop. +const CODEX_STOP_CONFIRM_TIMEOUT_MS = 10_000; // Codex can announce a retry and never resume. Recovered retries resume model output within ~75s. const CODEX_PROVIDER_RETRY_STALL_TIMEOUT = "10 minutes"; export const CODEX_DRIVER_KIND = CODEX_PROVIDER; @@ -7896,13 +7899,30 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( // unarchives it on the next turn. if (unloadInput.archive === true) { yield* client.request("thread/archive", { threadId: nativeThreadId }); - return; + } else { + // A resume may already wait on the pending entry; replacing it would strand that wait. + if (!threadClosings.has(nativeThreadId)) { + threadClosings.set(nativeThreadId, yield* Deferred.make()); + } + yield* client.request("thread/unsubscribe", { threadId: nativeThreadId }); } - // A resume may already wait on the pending entry; replacing it would strand that wait. - if (!threadClosings.has(nativeThreadId)) { - threadClosings.set(nativeThreadId, yield* Deferred.make()); + // No turn/completed arrives for an unloaded thread, so a Stop Codex never + // confirmed would otherwise keep its turn open for the life of the session. + const interrupting = yield* Ref.get(interruptingNativeTurns); + const completedAt = yield* DateTime.now; + for (const context of (yield* Ref.get(activeTurns)).values()) { + if ( + context.providerThread.id !== unloadInput.providerThread.id || + !interrupting.has(context.nativeTurnId) + ) + continue; + yield* finalizeCodexTurn({ + context, + nativeTurnId: context.nativeTurnId, + status: "interrupted", + completedAt, + }); } - yield* client.request("thread/unsubscribe", { threadId: nativeThreadId }); }, (effect, unloadInput) => effect.pipe( @@ -7990,23 +8010,30 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( readonly context: ActiveCodexTurnContext; readonly completion: Deferred.Deferred; }> = []; + const trackedInterruptNativeTurnIds = new Set(); + let joinedPendingStop = false; + // Untraced: runs once per lineage turn inside the interruptTurn span. + const trackInterruptTarget = Effect.fnUntraced(function* ( + context: ActiveCodexTurnContext, + ) { + const fresh = yield* Deferred.make(); + // An earlier Stop may still wait on this turn; replacing its waiter would strand it. + const completion = yield* Ref.modify(turnWaiters, (current) => { + const pending = current.get(context.nativeTurnId); + return pending === undefined + ? [fresh, new Map(current).set(context.nativeTurnId, fresh)] + : [pending, current]; + }); + joinedPendingStop ||= completion !== fresh; + yield* Ref.update(interruptingNativeTurns, (current) => + new Set(current).add(context.nativeTurnId), + ); + interruptTargets.push({ context, completion }); + trackedInterruptNativeTurnIds.add(context.nativeTurnId); + }); for (const context of interruptTargetContexts) { - interruptTargets.push({ context, completion: yield* Deferred.make() }); + yield* trackInterruptTarget(context); } - yield* Ref.update(turnWaiters, (current) => { - const updated = new Map(current); - for (const target of interruptTargets) { - updated.set(target.context.nativeTurnId, target.completion); - } - return updated; - }); - yield* Ref.update(interruptingNativeTurns, (current) => { - const updated = new Set(current); - for (const target of interruptTargets) { - updated.add(target.context.nativeTurnId); - } - return updated; - }); const registeredActiveTurns = yield* Ref.get(activeTurns); for (const target of interruptTargets) { if (registeredActiveTurns.get(target.context.nativeTurnId) !== target.context) { @@ -8039,9 +8066,6 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( }); }); - const trackedInterruptNativeTurnIds = new Set( - interruptTargets.map((target) => target.context.nativeTurnId), - ); const finalizeRemainingInterruptLineage = Effect.gen(function* () { while (true) { const newlyDiscovered: Array = []; @@ -8054,20 +8078,8 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( if (trackedInterruptNativeTurnIds.has(context.nativeTurnId)) { continue; } - const completion = yield* Deferred.make(); - interruptTargets.push({ context, completion }); - trackedInterruptNativeTurnIds.add(context.nativeTurnId); + yield* trackInterruptTarget(context); newlyDiscovered.push(context); - yield* Ref.update(turnWaiters, (current) => { - const updated = new Map(current); - updated.set(context.nativeTurnId, completion); - return updated; - }); - yield* Ref.update(interruptingNativeTurns, (current) => { - const updated = new Set(current); - updated.add(context.nativeTurnId); - return updated; - }); } return lineage; }), @@ -8130,19 +8142,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( !trackedInterruptNativeTurnIds.has(context.nativeTurnId), ); for (const context of lineage) { - const completion = yield* Deferred.make(); - interruptTargets.push({ context, completion }); - trackedInterruptNativeTurnIds.add(context.nativeTurnId); - yield* Ref.update(turnWaiters, (current) => { - const updated = new Map(current); - updated.set(context.nativeTurnId, completion); - return updated; - }); - yield* Ref.update(interruptingNativeTurns, (current) => { - const updated = new Set(current); - updated.add(context.nativeTurnId); - return updated; - }); + yield* trackInterruptTarget(context); } return lineage; }), @@ -8169,114 +8169,55 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( } }); - yield* Effect.gen(function* () { - for (const target of interruptTargets) { - const context = target.context; - if (context.capacityRetryPending !== undefined) { - yield* finalizeCodexTurn({ - context, - nativeTurnId: context.nativeTurnId, - status: "interrupted", - completedAt: yield* DateTime.now, - }); - continue; - } - // Null start timestamps ACK queued turns; interruption must wait for turn/started. - if (context.nativeStartReady !== undefined) { - const ready = yield* Deferred.await(context.nativeStartReady).pipe( - Effect.timeoutOption("10 seconds"), - ); - if (Option.isNone(ready)) { - return yield* toProtocolError( - "Codex did not start the queued turn within 10 seconds; Stop could not be delivered.", - ); - } - } - if ((yield* Ref.get(activeTurns)).get(context.nativeTurnId) !== context) { - continue; - } - yield* client.request("turn/interrupt", { - threadId: yield* getNativeThreadId(context.providerThread), - turnId: context.nativeTurnId, - }); - } - const containedTerminalKeys = new Set(); - const attemptedTerminalKeys = new Set(); - const collectTrackedTerminals = Effect.fn("CodexAdapterV2.collectTrackedTerminals")( - function* (targets: typeof interruptTargets) { - const trackedTerminals = new Map< - string, - { readonly nativeThreadId: string; readonly processId: string } - >(); - const runningItems = yield* Ref.get(runningCommandItemsByTurn); - for (const target of targets) { - const nativeThreadId = yield* getNativeThreadId( - target.context.providerThread, - ); - const items = runningItems.get(target.context.nativeTurnId); - for (const item of items?.values() ?? []) { - if (item.processId !== undefined) { - trackedTerminals.set(`${nativeThreadId}:${item.processId}`, { - nativeThreadId, - processId: item.processId, - }); - } + const containedTerminalKeys = new Set(); + const attemptedTerminalKeys = new Set(); + const collectTrackedTerminals = Effect.fn("CodexAdapterV2.collectTrackedTerminals")( + function* (targets: typeof interruptTargets) { + const trackedTerminals = new Map< + string, + { readonly nativeThreadId: string; readonly processId: string } + >(); + const runningItems = yield* Ref.get(runningCommandItemsByTurn); + for (const target of targets) { + const nativeThreadId = yield* getNativeThreadId(target.context.providerThread); + const items = runningItems.get(target.context.nativeTurnId); + for (const item of items?.values() ?? []) { + if (item.processId !== undefined) { + trackedTerminals.set(`${nativeThreadId}:${item.processId}`, { + nativeThreadId, + processId: item.processId, + }); } } - return trackedTerminals; - }, - ); - const terminateTrackedTerminals = Effect.fn( - "CodexAdapterV2.terminateTrackedTerminals", - )(function* (targets: typeof interruptTargets) { - const trackedTerminals = yield* collectTrackedTerminals(targets); - const pendingTerminals = Array.from(trackedTerminals.entries()) - .filter(([key]) => !containedTerminalKeys.has(key)) - .sort( - ([leftKey], [rightKey]) => - Number(attemptedTerminalKeys.has(leftKey)) - - Number(attemptedTerminalKeys.has(rightKey)), - ); - return yield* Effect.forEach( - pendingTerminals, - ([key, { nativeThreadId, processId }]) => - Effect.gen(function* () { - attemptedTerminalKeys.add(key); - yield* terminateBackgroundTerminal(nativeThreadId, processId); - containedTerminalKeys.add(key); - }).pipe( - Effect.as({ success: true as const }), - Effect.catch((error) => Effect.succeed({ success: false as const, error })), - ), + } + return trackedTerminals; + }, + ); + const terminateTrackedTerminals = Effect.fn( + "CodexAdapterV2.terminateTrackedTerminals", + )(function* (targets: typeof interruptTargets) { + const trackedTerminals = yield* collectTrackedTerminals(targets); + const pendingTerminals = Array.from(trackedTerminals.entries()) + .filter(([key]) => !containedTerminalKeys.has(key)) + .sort( + ([leftKey], [rightKey]) => + Number(attemptedTerminalKeys.has(leftKey)) - + Number(attemptedTerminalKeys.has(rightKey)), ); - }); - - const [, completed] = yield* Effect.all( - [ - terminateTrackedTerminals(interruptTargets), - Effect.forEach( - interruptTargets, - (target) => Deferred.await(target.completion), - { concurrency: "unbounded", discard: true }, - ).pipe(Effect.timeoutOption("10 seconds")), - ], - { concurrency: "unbounded" }, + return yield* Effect.forEach( + pendingTerminals, + ([key, { nativeThreadId, processId }]) => + Effect.gen(function* () { + attemptedTerminalKeys.add(key); + yield* terminateBackgroundTerminal(nativeThreadId, processId); + containedTerminalKeys.add(key); + }).pipe( + Effect.as({ success: true as const }), + Effect.catch((error) => Effect.succeed({ success: false as const, error })), + ), ); - if (Option.isNone(completed)) { - for (const target of interruptTargets) { - const context = target.context; - if ((yield* Ref.get(activeTurns)).get(context.nativeTurnId) !== context) { - continue; - } - yield* Effect.logWarning("orchestration-v2.codex-interrupt-timeout", { - providerSessionId: input.providerSessionId, - providerThreadId: context.providerThread.id, - providerTurnId: context.providerTurnId, - nativeTurnId: context.nativeTurnId, - }); - } - } - + }); + const settleInterruptedLineage = Effect.gen(function* () { yield* interruptLateDescendants; yield* finalizeRemainingInterruptLineage; @@ -8316,10 +8257,114 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( } }), ); + }); + // Codex may still be working, so a root stays open and marked interrupting. + // Whatever ends it later (Codex, Force stop) settles the rest of this Stop. + const deferToActiveRoots = Effect.gen(function* () { + const registered = yield* Ref.get(activeTurns); + const roots = interruptTargets.filter( + ({ context }) => + lineageRoots.includes(context) && + registered.get(context.nativeTurnId) === context, + ); + if (roots.length === 0) return false; + for (const { context } of roots) { + yield* Effect.logWarning("orchestration-v2.codex-interrupt-timeout", { + providerSessionId: input.providerSessionId, + providerThreadId: context.providerThread.id, + providerTurnId: context.providerTurnId, + nativeTurnId: context.nativeTurnId, + }); + } + yield* Effect.forEach(roots, (target) => Deferred.await(target.completion), { + concurrency: "unbounded", + discard: true, + }).pipe( + Effect.andThen(settleInterruptedLineage), + Effect.catchCause((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.void + : Effect.logWarning("orchestration-v2.codex-late-stop-settle-failed", { + providerSessionId: input.providerSessionId, + cause, + }), + ), + Effect.ensuring(cleanupInterruptState), + Effect.forkIn(scope), + ); + return true; + }); + + const unconfirmed = yield* Effect.gen(function* () { + for (const target of interruptTargets) { + const context = target.context; + if (context.capacityRetryPending !== undefined) { + yield* finalizeCodexTurn({ + context, + nativeTurnId: context.nativeTurnId, + status: "interrupted", + completedAt: yield* DateTime.now, + }); + continue; + } + // Null start timestamps ACK queued turns; interruption must wait for turn/started. + if (context.nativeStartReady !== undefined) { + const ready = yield* Deferred.await(context.nativeStartReady).pipe( + Effect.timeoutOption("10 seconds"), + ); + if (Option.isNone(ready)) { + return yield* toProtocolError( + "Codex did not start the queued turn within 10 seconds; Stop could not be delivered.", + ); + } + } + if ((yield* Ref.get(activeTurns)).get(context.nativeTurnId) !== context) { + continue; + } + yield* client.request("turn/interrupt", { + threadId: yield* getNativeThreadId(context.providerThread), + turnId: context.nativeTurnId, + }); + } + const [, completed] = yield* Effect.all( + [ + terminateTrackedTerminals(interruptTargets), + Effect.forEach( + interruptTargets, + (target) => Deferred.await(target.completion), + { concurrency: "unbounded", discard: true }, + ).pipe(Effect.timeoutOption(CODEX_STOP_CONFIRM_TIMEOUT_MS)), + ], + { concurrency: "unbounded" }, + ); + if (Option.isSome(completed) || !(yield* deferToActiveRoots)) { + yield* settleInterruptedLineage; + return false; + } + return true; }).pipe( - Effect.onError(() => finalizeRemainingInterruptLineage), - Effect.ensuring(cleanupInterruptState), + Effect.onExit((exit) => { + if (Exit.isSuccess(exit)) return exit.value ? Effect.void : cleanupInterruptState; + const abandon = finalizeRemainingInterruptLineage.pipe( + Effect.ensuring(cleanupInterruptState), + ); + // Neither a caller's deadline nor a failed retry of a pending Stop confirms + // that the turn ended. + return Cause.hasInterruptsOnly(exit.cause) || joinedPendingStop + ? Effect.flatMap(deferToActiveRoots, (deferred) => + deferred ? Effect.void : abandon, + ) + : abandon; + }), ); + if (unconfirmed) { + return yield* new ProviderAdapter.ProviderAdapterInterruptTimeoutError({ + driver: CODEX_PROVIDER, + providerThreadId: turnInput.providerThread.id, + providerTurnId: turnInput.providerTurnId, + timeoutMs: CODEX_STOP_CONFIRM_TIMEOUT_MS, + }); + } }, (effect, requestedInput) => Effect.acquireUseRelease( @@ -8338,14 +8383,15 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( else goalStops.delete(nativeThreadId); }), ).pipe( - Effect.mapError( - (cause) => - new ProviderAdapter.ProviderAdapterInterruptError({ - driver: CODEX_PROVIDER, - providerThreadId: requestedInput.providerThread.id, - providerTurnId: requestedInput.providerTurnId, - cause, - }), + Effect.mapError((cause) => + cause._tag === "ProviderAdapterInterruptTimeoutError" + ? cause + : new ProviderAdapter.ProviderAdapterInterruptError({ + driver: CODEX_PROVIDER, + providerThreadId: requestedInput.providerThread.id, + providerTurnId: requestedInput.providerTurnId, + cause, + }), ), ), ),