From 3a592c0e1d8ac158eb51aedd334c243016262988 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 13 Nov 2026 19:52:37 +0000 Subject: [PATCH 1/4] fix(server): a Codex Stop that never completes offers Force stop When Codex acknowledged turn/interrupt but never sent turn/completed, the adapter waited ten seconds, reported the still-running turn as interrupted and returned success. Stop looked finished while Codex could still be working, and the Stop banner never offered Force stop. interruptTurn now fails with the interrupt-timeout error when the turn it was asked to stop is still active after the wait, as the Claude adapter does, and leaves that turn open. The turn keeps its interrupting mark, so a completion that arrives later is still reported as interrupted and a late capacity failure is not retried; the rest of the Stop (descendants, tracked terminals) runs once the turn ends. Unloading the thread, which is what Force stop does on a shared app-server, ends such a turn, because no completion can arrive for it afterwards. A second Stop on a turn shares the first one's waiter instead of replacing it. A restart still fails on this error before any replacement starts. Fixes pingdotgg/t3code#30985 Co-Authored-By: Claude Fable 5.2 --- .../Adapters/CodexAdapterV2.test.ts | 132 +++++++++ .../Adapters/CodexAdapterV2.ts | 255 ++++++++++-------- 2 files changed, 277 insertions(+), 110 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 09ec3034a9bc9..41daafdb65a51 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -9523,6 +9523,138 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); + it.effect.each(["Codex completes it later", "its thread is unloaded"] 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 harness = yield* makeCodexReplayHarness( + makeCodexReplayTranscript({ + scenario: `unconfirmed-stop-${ending === "its thread is unloaded" ? "unload" : "late"}`, + 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: "turn/interrupt", frame: { id: 4, result: {} } }, + { + type: "emit_inbound", + label: "still working", + frame: { + method: "thread/tokenUsage/updated", + params: { + threadId: nativeThreadId, + turnId: nativeTurnId, + tokenUsage: { total: usage, last: usage, modelContextWindow: 200_000 }, + }, + }, + }, + ...(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) + : ([ + { + 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(Effect.result, Effect.forkScoped); + // Codex acknowledged 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("10 seconds"); + + const stopped = yield* Fiber.join(stopping); + assert.equal( + stopped._tag === "Failure" ? stopped.failure._tag : "confirmed", + "ProviderAdapterInterruptTimeoutError", + ); + assert.deepEqual(harness.terminalEvents(), []); + + if (ending === "its thread is unloaded") { + yield* harness.runtime.unloadThread!({ providerThread: harness.providerThread }); + } else { + yield* TestClock.adjust("10 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..abe511dbad5aa 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,27 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( readonly context: ActiveCodexTurnContext; readonly completion: Deferred.Deferred; }> = []; + const trackedInterruptNativeTurnIds = new Set(); + 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]; + }); + 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 +8063,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 +8075,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 +8139,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,7 +8166,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( } }); - yield* Effect.gen(function* () { + const unconfirmed = yield* Effect.gen(function* () { for (const target of interruptTargets) { const context = target.context; if (context.capacityRetryPending !== undefined) { @@ -8258,68 +8255,105 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( interruptTargets, (target) => Deferred.await(target.completion), { concurrency: "unbounded", discard: true }, - ).pipe(Effect.timeoutOption("10 seconds")), + ).pipe(Effect.timeoutOption(CODEX_STOP_CONFIRM_TIMEOUT_MS)), ], { concurrency: "unbounded" }, ); - 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; + + const terminationResults = yield* terminateTrackedTerminals(interruptTargets); + const failedTermination = terminationResults.find((result) => !result.success); + if (failedTermination !== undefined && !failedTermination.success) { + return yield* failedTermination.error; } + yield* turnTerminalizationPermit.withPermits(1)( + Effect.gen(function* () { + const completedAt = yield* DateTime.now; + for (const target of interruptTargets) { + const context = target.context; + if ((yield* Ref.get(settledTurns)).get(context.nativeTurnId) !== context) + continue; + yield* terminalizeRunningCommandItems( + context, + context.nativeTurnId, + "interrupted", + completedAt, + ); + yield* terminalizeRunningDynamicTools( + context, + context.nativeTurnId, + "interrupted", + completedAt, + true, + ); + yield* terminalizeMcpAppCaptures(context, context.nativeTurnId); + yield* Ref.update(runningCommandItemsByTurn, (current) => { + const updated = new Map(current); + updated.delete(context.nativeTurnId); + return updated; + }); + yield* clearImageWork(context); + yield* releaseSettledTurnIfIdle(context.nativeTurnId); + } + }), + ); + }); + const registered = yield* Ref.get(activeTurns); + const unconfirmedRoots = Option.isSome(completed) + ? [] + : interruptTargets.filter( + ({ context }) => + lineageRoots.includes(context) && + registered.get(context.nativeTurnId) === context, + ); + if (unconfirmedRoots.length === 0) { + yield* settleInterruptedLineage; + return false; } - - yield* interruptLateDescendants; - yield* finalizeRemainingInterruptLineage; - - const terminationResults = yield* terminateTrackedTerminals(interruptTargets); - const failedTermination = terminationResults.find((result) => !result.success); - if (failedTermination !== undefined && !failedTermination.success) { - return yield* failedTermination.error; + for (const { context } of unconfirmedRoots) { + yield* Effect.logWarning("orchestration-v2.codex-interrupt-timeout", { + providerSessionId: input.providerSessionId, + providerThreadId: context.providerThread.id, + providerTurnId: context.providerTurnId, + nativeTurnId: context.nativeTurnId, + }); } - yield* turnTerminalizationPermit.withPermits(1)( - Effect.gen(function* () { - const completedAt = yield* DateTime.now; - for (const target of interruptTargets) { - const context = target.context; - if ((yield* Ref.get(settledTurns)).get(context.nativeTurnId) !== context) - continue; - yield* terminalizeRunningCommandItems( - context, - context.nativeTurnId, - "interrupted", - completedAt, - ); - yield* terminalizeRunningDynamicTools( - context, - context.nativeTurnId, - "interrupted", - completedAt, - true, - ); - yield* terminalizeMcpAppCaptures(context, context.nativeTurnId); - yield* Ref.update(runningCommandItemsByTurn, (current) => { - const updated = new Map(current); - updated.delete(context.nativeTurnId); - return updated; - }); - yield* clearImageWork(context); - yield* releaseSettledTurnIfIdle(context.nativeTurnId); - } - }), + // Codex may still be working, so the turn stays open and marked interrupting. + // Whatever ends it later (Codex, Force stop) settles the rest of this Stop. + yield* Effect.forEach( + unconfirmedRoots, + (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; }).pipe( Effect.onError(() => finalizeRemainingInterruptLineage), - Effect.ensuring(cleanupInterruptState), + Effect.onExit((exit) => + Exit.isSuccess(exit) && exit.value ? Effect.void : cleanupInterruptState, + ), ); + 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 +8372,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, + }), ), ), ), From ac741a9765490c7625a9ebd851b9b9d7f37035d6 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 13 Nov 2026 20:03:57 +0000 Subject: [PATCH 2/4] chore(server): say why Stop target registration is untraced Co-Authored-By: Claude Fable 5.2 --- apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index abe511dbad5aa..3f32f29f9d229 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -8011,6 +8011,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( readonly completion: Deferred.Deferred; }> = []; const trackedInterruptNativeTurnIds = new Set(); + // Untraced: runs once per lineage turn inside the interruptTurn span. const trackInterruptTarget = Effect.fnUntraced(function* ( context: ActiveCodexTurnContext, ) { From f6a92f02823a9b42feaf6beb37596db44f6f18d4 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 13 Nov 2026 20:22:35 +0000 Subject: [PATCH 3/4] fix(server): a Codex Stop cut short by the session deadline stays unconfirmed The session manager bounds every Stop at fifteen seconds. When that deadline interrupted the adapter, for example because Codex took more than five seconds to acknowledge the interrupt, the adapter still reported the running turn as interrupted. The stored terminal then let a retried restart skip the interrupt and start its replacement. An interrupted Stop now hands a still-active turn to the same pending settle as the adapter's own timeout: the turn stays open until Codex ends it or its thread is unloaded. Co-Authored-By: Claude Fable 5.2 --- .../Adapters/CodexAdapterV2.test.ts | 251 ++++++++-------- .../Adapters/CodexAdapterV2.ts | 271 +++++++++--------- 2 files changed, 273 insertions(+), 249 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 41daafdb65a51..a350ea13f1f72 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -9523,136 +9523,153 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); - it.effect.each(["Codex completes it later", "its thread is unloaded"] 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 harness = yield* makeCodexReplayHarness( - makeCodexReplayTranscript({ - scenario: `unconfirmed-stop-${ending === "its thread is unloaded" ? "unload" : "late"}`, - entries: [ - ...codexReplayPreamble({ nativeThreadId, nativeTurnId, prompt: "work" }), - { - type: "expect_outbound", - label: "turn/interrupt", - frame: { - id: 4, - method: "turn/interrupt", - params: { threadId: nativeThreadId, turnId: nativeTurnId }, - }, + it.effect.each([ + "Codex completes it later", + "its thread is unloaded", + "Codex completes it after the session's Stop deadline", + ] 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: "turn/interrupt", frame: { id: 4, result: {} } }, - { - 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: "still working", + frame: { + method: "thread/tokenUsage/updated", + params: { + threadId: nativeThreadId, + turnId: nativeTurnId, + tokenUsage: { total: usage, last: usage, modelContextWindow: 200_000 }, }, }, - ...(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" } }, + }, + { + 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 }, }, - ] satisfies Array) - : ([ - { - type: "emit_inbound", - label: "turn/completed", - afterMs: 20_000, - frame: { - method: "turn/completed", - params: { - threadId: nativeThreadId, - turn: makeCodexReplayTurn({ id: nativeTurnId, status: "completed" }), - }, + }, + { + type: "emit_inbound", + label: "thread/unsubscribe", + frame: { id: 5, result: { status: "unsubscribed" } }, + }, + ] 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)), - ], + }, + ] 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, }), - ); - yield* harness.runtime.startTurn( - makeCodexTestTurnInput({ - threadId: harness.threadId, - providerThread: harness.providerThread, - now: yield* DateTime.now, - attemptId: RunAttemptId.make("attempt-unconfirmed-stop"), - text: "work", + }) + .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, ); - const stopping = yield* harness.runtime - .interruptTurn({ - providerThread: harness.providerThread, - providerTurnId: (yield* IdAllocator.IdAllocatorV2).derive.providerTurn({ - driver: CodexAdapterV2.CODEX_DRIVER_KIND, - nativeTurnId, - }), - }) - .pipe(Effect.result, Effect.forkScoped); - // Codex acknowledged 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("10 seconds"); + // 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", - "ProviderAdapterInterruptTimeoutError", - ); - assert.deepEqual(harness.terminalEvents(), []); + const stopped = yield* Fiber.join(stopping); + assert.equal( + stopped._tag === "Failure" ? stopped.failure._tag : "confirmed", + pastDeadline ? "SessionStopDeadline" : "ProviderAdapterInterruptTimeoutError", + ); + assert.deepEqual(harness.terminalEvents(), []); - if (ending === "its thread is unloaded") { - yield* harness.runtime.unloadThread!({ providerThread: harness.providerThread }); - } else { - yield* TestClock.adjust("10 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, - ), + 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", () => { diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 3f32f29f9d229..044375972f149 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -8167,6 +8167,132 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( } }); + 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 })), + ), + ); + }); + const settleInterruptedLineage = Effect.gen(function* () { + yield* interruptLateDescendants; + yield* finalizeRemainingInterruptLineage; + + const terminationResults = yield* terminateTrackedTerminals(interruptTargets); + const failedTermination = terminationResults.find((result) => !result.success); + if (failedTermination !== undefined && !failedTermination.success) { + return yield* failedTermination.error; + } + yield* turnTerminalizationPermit.withPermits(1)( + Effect.gen(function* () { + const completedAt = yield* DateTime.now; + for (const target of interruptTargets) { + const context = target.context; + if ((yield* Ref.get(settledTurns)).get(context.nativeTurnId) !== context) + continue; + yield* terminalizeRunningCommandItems( + context, + context.nativeTurnId, + "interrupted", + completedAt, + ); + yield* terminalizeRunningDynamicTools( + context, + context.nativeTurnId, + "interrupted", + completedAt, + true, + ); + yield* terminalizeMcpAppCaptures(context, context.nativeTurnId); + yield* Ref.update(runningCommandItemsByTurn, (current) => { + const updated = new Map(current); + updated.delete(context.nativeTurnId); + return updated; + }); + yield* clearImageWork(context); + yield* releaseSettledTurnIfIdle(context.nativeTurnId); + } + }), + ); + }); + // 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; @@ -8198,57 +8324,6 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( 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, - }); - } - } - } - 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 })), - ), - ); - }); - const [, completed] = yield* Effect.all( [ terminateTrackedTerminals(interruptTargets), @@ -8260,92 +8335,24 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( ], { concurrency: "unbounded" }, ); - const settleInterruptedLineage = Effect.gen(function* () { - yield* interruptLateDescendants; - yield* finalizeRemainingInterruptLineage; - - const terminationResults = yield* terminateTrackedTerminals(interruptTargets); - const failedTermination = terminationResults.find((result) => !result.success); - if (failedTermination !== undefined && !failedTermination.success) { - return yield* failedTermination.error; - } - yield* turnTerminalizationPermit.withPermits(1)( - Effect.gen(function* () { - const completedAt = yield* DateTime.now; - for (const target of interruptTargets) { - const context = target.context; - if ((yield* Ref.get(settledTurns)).get(context.nativeTurnId) !== context) - continue; - yield* terminalizeRunningCommandItems( - context, - context.nativeTurnId, - "interrupted", - completedAt, - ); - yield* terminalizeRunningDynamicTools( - context, - context.nativeTurnId, - "interrupted", - completedAt, - true, - ); - yield* terminalizeMcpAppCaptures(context, context.nativeTurnId); - yield* Ref.update(runningCommandItemsByTurn, (current) => { - const updated = new Map(current); - updated.delete(context.nativeTurnId); - return updated; - }); - yield* clearImageWork(context); - yield* releaseSettledTurnIfIdle(context.nativeTurnId); - } - }), - ); - }); - const registered = yield* Ref.get(activeTurns); - const unconfirmedRoots = Option.isSome(completed) - ? [] - : interruptTargets.filter( - ({ context }) => - lineageRoots.includes(context) && - registered.get(context.nativeTurnId) === context, - ); - if (unconfirmedRoots.length === 0) { + if (Option.isSome(completed) || !(yield* deferToActiveRoots)) { yield* settleInterruptedLineage; return false; } - for (const { context } of unconfirmedRoots) { - yield* Effect.logWarning("orchestration-v2.codex-interrupt-timeout", { - providerSessionId: input.providerSessionId, - providerThreadId: context.providerThread.id, - providerTurnId: context.providerTurnId, - nativeTurnId: context.nativeTurnId, - }); - } - // Codex may still be working, so the turn stays open and marked interrupting. - // Whatever ends it later (Codex, Force stop) settles the rest of this Stop. - yield* Effect.forEach( - unconfirmedRoots, - (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; }).pipe( - Effect.onError(() => finalizeRemainingInterruptLineage), - Effect.onExit((exit) => - Exit.isSuccess(exit) && exit.value ? Effect.void : cleanupInterruptState, - ), + Effect.onExit((exit) => { + if (Exit.isSuccess(exit)) return exit.value ? Effect.void : cleanupInterruptState; + const abandon = finalizeRemainingInterruptLineage.pipe( + Effect.ensuring(cleanupInterruptState), + ); + // A caller's deadline cutting this short confirms nothing about the turn either. + return Cause.hasInterruptsOnly(exit.cause) + ? Effect.flatMap(deferToActiveRoots, (deferred) => + deferred ? Effect.void : abandon, + ) + : abandon; + }), ); if (unconfirmed) { return yield* new ProviderAdapter.ProviderAdapterInterruptTimeoutError({ From 7c7339c473ac7dd2c6cb843ba73daff7ccfaaee5 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 13 Nov 2026 20:38:24 +0000 Subject: [PATCH 4/4] fix(server): a failed retry of a pending Codex Stop keeps its turn open A second Stop on a turn whose first Stop was still unconfirmed could fail, for example on an RPC error, and then reported the running turn as interrupted. It now leaves the turn to the pending settle, as the first Stop did. Co-Authored-By: Claude Fable 5.2 --- .../Adapters/CodexAdapterV2.test.ts | 32 +++++++++++++++++++ .../Adapters/CodexAdapterV2.ts | 7 ++-- 2 files changed, 37 insertions(+), 2 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index a350ea13f1f72..0fc1b9a2bf82d 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -9527,6 +9527,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { "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"; @@ -9590,6 +9591,24 @@ describe("CodexAdapterV2 post-settle continuation", () => { }, ] 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", @@ -9649,6 +9668,19 @@ describe("CodexAdapterV2 post-settle continuation", () => { ); 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 { diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 044375972f149..e26b3dc5860c2 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -8011,6 +8011,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( 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, @@ -8023,6 +8024,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( ? [fresh, new Map(current).set(context.nativeTurnId, fresh)] : [pending, current]; }); + joinedPendingStop ||= completion !== fresh; yield* Ref.update(interruptingNativeTurns, (current) => new Set(current).add(context.nativeTurnId), ); @@ -8346,8 +8348,9 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( const abandon = finalizeRemainingInterruptLineage.pipe( Effect.ensuring(cleanupInterruptState), ); - // A caller's deadline cutting this short confirms nothing about the turn either. - return Cause.hasInterruptsOnly(exit.cause) + // 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, )