From b01f501c5e46bf57c43bfd9ca5275bc18a464da6 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 4 Nov 2026 16:56:29 +0000 Subject: [PATCH 1/9] test(server): reproduce #30985 Codex Stop that never completes is reported as interrupted When Codex acknowledges turn/interrupt but never sends turn/completed, the adapter synthesizes an interrupted terminal and interruptTurn succeeds, so Force stop is never offered. This test expects the turn to stay open and interruptTurn to fail with ProviderAdapterInterruptTimeoutError. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/CodexAdapterV2.test.ts | 81 +++++++++++++++++++ 1 file changed, 81 insertions(+) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 9482bb5a3fe2..f2accd40d537 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -1824,6 +1824,87 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); + it.effect( + "leaves the turn open and reports a timeout when Codex never completes an interrupted turn", + () => + Effect.gen(function* () { + const nativeThreadId = "unconfirmed-stop-thread"; + const nativeTurnId = "unconfirmed-stop-turn"; + const prompt = "Keep working."; + const interruptSent = yield* Deferred.make(); + const harness = yield* makeCodexReplayHarness( + makeCodexReplayTranscript({ + scenario: "unconfirmed-stop", + entries: [ + ...codexReplayPreamble({ nativeThreadId, nativeTurnId, prompt }), + { + type: "expect_outbound", + label: "turn/interrupt", + frame: { + id: 4, + method: "turn/interrupt", + params: { threadId: nativeThreadId, turnId: nativeTurnId }, + }, + }, + { + type: "emit_inbound", + label: "turn/interrupt", + afterMs: 1000, + frame: { id: 4, result: {} }, + }, + ], + }), + () => Effect.void, + (method) => + method === "turn/interrupt" ? Deferred.succeed(interruptSent, undefined) : Effect.void, + ); + yield* harness.runtime.startTurn( + makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("unconfirmed-stop-attempt"), + text: prompt, + }), + ); + yield* harness.until(() => + harness.events.some((event) => event.type === "provider_turn.updated"), + ); + const providerTurnId = (yield* IdAllocator.IdAllocatorV2).derive.providerTurn({ + driver: CodexAdapterV2.CODEX_DRIVER_KIND, + nativeTurnId, + }); + const interrupt = yield* harness.runtime + .interruptTurn({ providerThread: harness.providerThread, providerTurnId }) + .pipe(Effect.exit, Effect.forkScoped); + yield* Deferred.await(interruptSent); + yield* TestClock.adjust("1 second"); + yield* TestClock.adjust("10 seconds"); + const exit = yield* Fiber.join(interrupt); + + assert.deepEqual( + harness.terminalEvents().map((event) => event.status), + [], + "an unconfirmed Stop must not synthesize an interrupted terminal", + ); + assert.isTrue(Exit.isFailure(exit), "an unconfirmed Stop must fail"); + if (!Exit.isFailure(exit)) return; + const error = Cause.squash(exit.cause); + assert.instanceOf(error, ProviderAdapter.ProviderAdapterInterruptTimeoutError); + assert.equal(error.driver, CodexAdapterV2.CODEX_DRIVER_KIND); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ); + const assistantMessages = (events: ReadonlyArray) => events.filter( ( From c3cd4ca5c6619bbcc6968bb1935ca1b75f5b43c3 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 4 Nov 2026 17:08:49 +0000 Subject: [PATCH 2/9] fix(server): Codex Stop that never completes offers Force stop When Codex acknowledged turn/interrupt but never sent turn/completed within the 10 second wait, the adapter finalized the turn as interrupted anyway, so the thread looked stopped while Codex kept working and Force stop was never offered. An unconfirmed root now leaves its lineage open and fails with ProviderAdapterInterruptTimeoutError, which the outer mapError passes through so Stop is marked unresponsive. Interrupt cleanup keeps the running command items of turns that are still active, so a later turn/completed from Codex finalizes the turn and its commands through the normal notification path. AI-assisted; verified with the tests and tools listed above. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/CodexAdapterV2.test.ts | 119 ++++++++++++++++++ .../Adapters/CodexAdapterV2.ts | 41 ++++-- 2 files changed, 150 insertions(+), 10 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index f2accd40d537..68980629d7e7 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -1905,6 +1905,125 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); + it.effect( + "finalizes an unconfirmed Stop and its running commands when Codex completes later", + () => + Effect.gen(function* () { + const nativeThreadId = "late-stop-thread"; + const nativeTurnId = "late-stop-turn"; + const prompt = "Keep working."; + const commandItem = { + type: "commandExecution", + id: "late-stop-command", + command: "sleep 60", + cwd: "/workspace", + source: "unifiedExecStartup", + commandActions: [{ type: "unknown", command: "sleep 60" }], + aggregatedOutput: null, + exitCode: null, + durationMs: null, + }; + const interruptSent = yield* Deferred.make(); + const harness = yield* makeCodexReplayHarness( + makeCodexReplayTranscript({ + scenario: "late-stop", + entries: [ + ...codexReplayPreamble({ nativeThreadId, nativeTurnId, prompt }), + { + type: "emit_inbound", + label: "item/started", + frame: { + method: "item/started", + params: { + threadId: nativeThreadId, + turnId: nativeTurnId, + startedAtMs: 1782622445000, + item: { ...commandItem, status: "inProgress" }, + }, + }, + }, + { + type: "expect_outbound", + label: "turn/interrupt", + frame: { + id: 4, + method: "turn/interrupt", + params: { threadId: nativeThreadId, turnId: nativeTurnId }, + }, + }, + { + type: "emit_inbound", + label: "turn/interrupt", + afterMs: 1000, + frame: { id: 4, result: {} }, + }, + { + type: "emit_inbound", + label: "turn/completed", + afterMs: 15_000, + frame: { + method: "turn/completed", + params: { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: nativeTurnId, status: "interrupted" }), + }, + }, + }, + ], + }), + () => Effect.void, + (method) => + method === "turn/interrupt" ? Deferred.succeed(interruptSent, undefined) : Effect.void, + ); + yield* harness.runtime.startTurn( + makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("late-stop-attempt"), + text: prompt, + }), + ); + const commandUpdates = () => + harness.events.flatMap((event) => + event.type === "turn_item.updated" && event.turnItem.type === "command_execution" + ? [event.turnItem.status] + : [], + ); + yield* harness.until(() => commandUpdates().length > 0); + const providerTurnId = (yield* IdAllocator.IdAllocatorV2).derive.providerTurn({ + driver: CodexAdapterV2.CODEX_DRIVER_KIND, + nativeTurnId, + }); + const interrupt = yield* harness.runtime + .interruptTurn({ providerThread: harness.providerThread, providerTurnId }) + .pipe(Effect.exit, Effect.forkScoped); + yield* Deferred.await(interruptSent); + yield* TestClock.adjust("1 second"); + yield* TestClock.adjust("10 seconds"); + assert.isTrue(Exit.isFailure(yield* Fiber.join(interrupt))); + assert.deepEqual(harness.terminalEvents(), []); + + yield* TestClock.adjust("5 seconds"); + yield* harness.until(() => harness.terminalEvents().length > 0); + assert.deepEqual( + harness.terminalEvents().map((event) => event.status), + ["interrupted"], + ); + assert.equal(commandUpdates().at(-1), "interrupted"); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ); + const assistantMessages = (events: ReadonlyArray) => events.filter( ( diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 0f8e0ef1f336..c47b3d6762f4 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -157,6 +157,8 @@ import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter" const CODEX_PROVIDER = ProviderDriverKind.make("codex"); const CODEX_CAPACITY_MAX_RETRIES = 5; +// How long a Stop waits for turn/completed before the user is offered Force stop. +const CODEX_STOP_CONFIRM_TIMEOUT_MS = 10_000; export const CODEX_DRIVER_KIND = CODEX_PROVIDER; export const CODEX_DEFAULT_INSTANCE_ID = defaultInstanceIdForDriver(CODEX_DRIVER_KIND); @@ -7425,10 +7427,14 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( } return updated; }); + const stillActiveTurns = yield* Ref.get(activeTurns); yield* Ref.update(runningCommandItemsByTurn, (current) => { const updated = new Map(current); for (const target of interruptTargets) { if (settledTurnContexts.includes(target.context)) continue; + // An unconfirmed turn keeps its commands for its own later completion. + if (stillActiveTurns.get(target.context.nativeTurnId) === target.context) + continue; updated.delete(target.context.nativeTurnId); } return updated; @@ -7565,7 +7571,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( } }); - yield* Effect.gen(function* () { + const rootConfirmed = yield* Effect.gen(function* () { for (const target of interruptTargets) { const context = target.context; if (context.capacityRetryPending !== undefined) { @@ -7654,16 +7660,18 @@ 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)) { + let rootStillActive = false; for (const target of interruptTargets) { const context = target.context; if ((yield* Ref.get(activeTurns)).get(context.nativeTurnId) !== context) { continue; } + if (lineageRoots.includes(context)) rootStillActive = true; yield* Effect.logWarning("orchestration-v2.codex-interrupt-timeout", { providerSessionId: input.providerSessionId, providerThreadId: context.providerThread.id, @@ -7671,6 +7679,9 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( nativeTurnId: context.nativeTurnId, }); } + // The lineage stays open: Codex may still be working, and the user decides + // whether to Force stop it. A later turn/completed still finalizes it. + if (rootStillActive) return false; } yield* interruptLateDescendants; @@ -7712,10 +7723,19 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( } }), ); + return true; }).pipe( Effect.onError(() => finalizeRemainingInterruptLineage), Effect.ensuring(cleanupInterruptState), ); + if (!rootConfirmed) { + return yield* new ProviderAdapter.ProviderAdapterInterruptTimeoutError({ + driver: CODEX_PROVIDER, + providerThreadId: requestedInput.providerThread.id, + providerTurnId: requestedInput.providerTurnId, + timeoutMs: CODEX_STOP_CONFIRM_TIMEOUT_MS, + }); + } }, (effect, requestedInput) => Effect.acquireUseRelease( @@ -7734,14 +7754,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 b8b861fcf00ad390a7e36b91d324109d46a7d318 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 4 Nov 2026 17:27:33 +0000 Subject: [PATCH 3/9] fix(server): a Codex Stop that times out stays stopped until the turn ends When Codex acknowledged turn/interrupt but never sent turn/completed, the timed-out Stop cleared its interrupting mark. A late serverOverloaded failure then scheduled a capacity retry and continued the stopped run, and a late native completion read as completed. The timeout path also skipped the second terminal sweep, so a descendant command started after the first sweep was never terminated. The interrupting mark now stays on turns Codex has not finished and is cleared when the turn finalizes. Command tracking is kept only while a Stop is still waiting. The timeout path still runs the second terminal sweep and the settled-target cleanup, and reports the timeout over a termination failure, since Force stop covers both. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/CodexAdapterV2.test.ts | 226 +++++++++++++++++- .../Adapters/CodexAdapterV2.ts | 35 ++- 2 files changed, 248 insertions(+), 13 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 68980629d7e7..048920bafafe 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -1965,7 +1965,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { method: "turn/completed", params: { threadId: nativeThreadId, - turn: makeCodexReplayTurn({ id: nativeTurnId, status: "interrupted" }), + turn: makeCodexReplayTurn({ id: nativeTurnId, status: "completed" }), }, }, }, @@ -2023,6 +2023,104 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ), ); + it.effect("does not retry capacity after a Stop Codex never confirmed", () => + Effect.gen(function* () { + const nativeThreadId = "capacity-after-stop-thread"; + const nativeTurnId = "capacity-after-stop-turn"; + const prompt = "Keep working."; + const interruptSent = yield* Deferred.make(); + let starts = 0; + const harness = yield* makeCodexReplayHarness( + makeCodexReplayTranscript({ + scenario: "capacity-after-stop", + entries: [ + ...codexReplayPreamble({ nativeThreadId, nativeTurnId, prompt }), + { + type: "expect_outbound", + label: "turn/interrupt", + frame: { + id: 4, + method: "turn/interrupt", + params: { threadId: nativeThreadId, turnId: nativeTurnId }, + }, + }, + { type: "emit_inbound", afterMs: 1000, frame: { id: 4, result: {} } }, + { + type: "emit_inbound", + label: "turn/completed", + afterMs: 15_000, + frame: { + method: "turn/completed", + params: { + threadId: nativeThreadId, + turn: { + ...makeCodexReplayTurn({ id: nativeTurnId, status: "failed" }), + error: { message: "At capacity", codexErrorInfo: "serverOverloaded" }, + }, + }, + }, + }, + ], + }), + () => Effect.void, + (method) => + Effect.gen(function* () { + if (method === "turn/start") starts++; + if (method === "turn/interrupt") yield* Deferred.succeed(interruptSent, undefined); + }), + ); + yield* harness.runtime.startTurn( + makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("capacity-after-stop-attempt"), + text: prompt, + }), + ); + yield* harness.until(() => + harness.events.some((event) => event.type === "provider_turn.updated"), + ); + const providerTurnId = (yield* IdAllocator.IdAllocatorV2).derive.providerTurn({ + driver: CodexAdapterV2.CODEX_DRIVER_KIND, + nativeTurnId, + }); + const interrupt = yield* harness.runtime + .interruptTurn({ providerThread: harness.providerThread, providerTurnId }) + .pipe(Effect.exit, Effect.forkScoped); + yield* Deferred.await(interruptSent); + yield* TestClock.adjust("1 second"); + yield* TestClock.adjust("10 seconds"); + assert.isTrue(Exit.isFailure(yield* Fiber.join(interrupt))); + + yield* TestClock.adjust("5 seconds"); + yield* harness.until( + () => + harness.terminalEvents().length > 0 || + harness.events.some( + (event) => event.type === "turn_item.updated" && event.turnItem.type === "error", + ), + ); + // A scheduled capacity retry would start its turn within its backoff. + yield* TestClock.adjust("30 seconds"); + assert.equal(starts, 1, "a stopped run must not continue on a capacity retry"); + // Matches the same frame inside the Stop wait: only a native completion reads as interrupted. + assert.deepEqual( + harness.terminalEvents().map((event) => event.status), + ["failed"], + ); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ); const assistantMessages = (events: ReadonlyArray) => events.filter( @@ -5630,6 +5728,132 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); + it.effect("contains descendant commands when Codex never confirms the root Stop", () => + Effect.scoped( + Effect.gen(function* () { + const terminated: Array = []; + const harness = yield* makeCodexReplayHarness( + makeCodexReplayTranscript({ + scenario: "codex-interrupt-unconfirmed-root-child-command", + entries: [ + ...interruptSubagentCommandTranscript.entries.slice(0, rootCompletionIndex), + { + type: "emit_inbound", + label: "item/started/child-command-after-sweep", + frame: { + method: "item/started", + params: { + item: { + ...interruptCommandItem("inProgress"), + id: `${INTERRUPT_CHILD_COMMAND_ITEM}-late`, + processId: "57683", + }, + threadId: INTERRUPT_CHILD_NATIVE_THREAD, + turnId: INTERRUPT_CHILD_NATIVE_TURN, + startedAtMs: 1782622442000, + }, + }, + }, + { + type: "emit_inbound", + label: "turn/completed/child", + frame: { + method: "turn/completed", + params: { + threadId: INTERRUPT_CHILD_NATIVE_THREAD, + turn: makeCodexReplayTurn({ + id: INTERRUPT_CHILD_NATIVE_TURN, + status: "interrupted", + }), + }, + }, + }, + { + type: "expect_outbound", + label: "thread/backgroundTerminals/terminate/child-late", + frame: { + id: 7, + method: "thread/backgroundTerminals/terminate", + params: { threadId: INTERRUPT_CHILD_NATIVE_THREAD, processId: "57683" }, + }, + }, + { + type: "emit_inbound", + label: "thread/backgroundTerminals/terminate/child-late", + frame: { id: 7, result: { terminated: true } }, + }, + ], + }), + () => Effect.void, + (method, params) => + Effect.sync(() => { + if (method === "thread/backgroundTerminals/terminate") + terminated.push((params as { readonly processId: string }).processId); + }), + ); + yield* harness.runtime.startTurn( + makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("attempt-codex-interrupt-unconfirmed-root"), + text: INTERRUPT_PROMPT, + }), + ); + yield* harness.until(() => + harness.events.some( + (event) => + event.type === "turn_item.updated" && + event.turnItem.type === "command_execution" && + event.turnItem.nativeItemRef?.nativeId === INTERRUPT_CHILD_COMMAND_ITEM && + event.turnItem.status === "running", + ), + ); + const providerTurnId = harness.events.find( + ( + event, + ): event is Extract< + ProviderAdapter.ProviderAdapterV2Event, + { type: "provider_turn.updated" } + > => event.type === "provider_turn.updated" && event.threadId === harness.threadId, + )?.providerTurn.id; + assert.isDefined(providerTurnId); + + const interrupt = yield* harness.runtime + .interruptTurn({ providerThread: harness.providerThread, providerTurnId }) + .pipe(Effect.exit, Effect.forkScoped); + yield* harness.until(() => + harness.events.some( + (event) => + event.type === "provider_turn.updated" && + event.providerTurn.nativeTurnRef?.nativeId === INTERRUPT_CHILD_NATIVE_TURN && + event.providerTurn.status === "interrupted", + ), + ); + yield* TestClock.adjust("10 seconds"); + const exit = yield* Fiber.join(interrupt); + + assert.isTrue(Exit.isFailure(exit)); + if (!Exit.isFailure(exit)) return; + assert.instanceOf( + Cause.squash(exit.cause), + ProviderAdapter.ProviderAdapterInterruptTimeoutError, + ); + assert.deepEqual(terminated, ["57682", "57683"]); + assert.deepEqual(harness.terminalEvents(), []); + }).pipe( + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ), + ); + const interruptSubagentTurnAfterStopTranscript = makeCodexReplayTranscript({ scenario: "codex-interrupt-subagent-turn-after-stop", entries: [ diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index c47b3d6762f4..3215feba6c76 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -6346,9 +6346,15 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( if (waiter !== undefined) { yield* Deferred.succeed(waiter, undefined); } - const interruptInProgress = (yield* Ref.get(interruptingNativeTurns)).has( - input.nativeTurnId, - ); + // A waiting Stop keeps the turn's commands for its second terminal sweep and + // clears its own mark; a timed-out Stop left the mark for this finalization. + const interruptInProgress = waiter !== undefined; + yield* Ref.update(interruptingNativeTurns, (current) => { + if (interruptInProgress || !current.has(input.nativeTurnId)) return current; + const updated = new Set(current); + updated.delete(input.nativeTurnId); + return updated; + }); const retainSettledContext = input.status === "completed" && (yield* turnHasRetainedBackgroundWork(input.nativeTurnId)); @@ -7420,14 +7426,17 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( } return updated; }); + const stillActiveTurns = yield* Ref.get(activeTurns); + // A turn Codex never confirmed stays stopped until finalizeCodexTurn clears it. yield* Ref.update(interruptingNativeTurns, (current) => { const updated = new Set(current); for (const target of interruptTargets) { + if (stillActiveTurns.get(target.context.nativeTurnId) === target.context) + continue; updated.delete(target.context.nativeTurnId); } return updated; }); - const stillActiveTurns = yield* Ref.get(activeTurns); yield* Ref.update(runningCommandItemsByTurn, (current) => { const updated = new Map(current); for (const target of interruptTargets) { @@ -7664,8 +7673,8 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( ], { concurrency: "unbounded" }, ); + let rootStillActive = false; if (Option.isNone(completed)) { - let rootStillActive = false; for (const target of interruptTargets) { const context = target.context; if ((yield* Ref.get(activeTurns)).get(context.nativeTurnId) !== context) { @@ -7679,17 +7688,19 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( nativeTurnId: context.nativeTurnId, }); } - // The lineage stays open: Codex may still be working, and the user decides - // whether to Force stop it. A later turn/completed still finalizes it. - if (rootStillActive) return false; } - - yield* interruptLateDescendants; - yield* finalizeRemainingInterruptLineage; + // An unconfirmed root keeps its lineage open: Codex may still be working, and + // the user decides whether to Force stop it. A later turn/completed finalizes it. + if (!rootStillActive) { + yield* interruptLateDescendants; + yield* finalizeRemainingInterruptLineage; + } const terminationResults = yield* terminateTrackedTerminals(interruptTargets); const failedTermination = terminationResults.find((result) => !result.success); if (failedTermination !== undefined && !failedTermination.success) { + // Force stop answers both, and the timeout says Codex itself has not stopped. + if (rootStillActive) return false; return yield* failedTermination.error; } yield* turnTerminalizationPermit.withPermits(1)( @@ -7723,7 +7734,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( } }), ); - return true; + return !rootStillActive; }).pipe( Effect.onError(() => finalizeRemainingInterruptLineage), Effect.ensuring(cleanupInterruptState), From 39ab4f1b1b6769dd66eaec30e601f91888f86b7b Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 4 Nov 2026 17:46:57 +0000 Subject: [PATCH 4/9] fix(server): an unconfirmed Codex Stop settles its lineage when Codex completes A Stop that timed out dropped its remaining work: descendants stayed active, commands started after the timeout were never terminated, and subagents spawned during the wait were never interrupted. The session manager's 15s cap also ran the error path on interruption, which synthesized interrupted terminals and hid Force stop. The work after Codex confirms a Stop now runs in a fiber in the session scope. A timed-out or capped Stop hands off to it, and it waits for the roots' own turn/completed. Each pending Stop keeps its own turn waiter, so a second Stop does not strand the first. Force stop's unloadThread finalizes turns that still have a Stop pending. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/CodexAdapterV2.test.ts | 314 +++++++++++++- .../Adapters/CodexAdapterV2.ts | 396 ++++++++++-------- 2 files changed, 529 insertions(+), 181 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 048920bafafe..703aff7f1679 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -1852,6 +1852,20 @@ describe("CodexAdapterV2 post-settle continuation", () => { afterMs: 1000, frame: { id: 4, result: {} }, }, + { + 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" } }, + }, ], }), () => Effect.void, @@ -1892,6 +1906,15 @@ describe("CodexAdapterV2 post-settle continuation", () => { const error = Cause.squash(exit.cause); assert.instanceOf(error, ProviderAdapter.ProviderAdapterInterruptTimeoutError); assert.equal(error.driver, CodexAdapterV2.CODEX_DRIVER_KIND); + + // Force stop on a shared app-server unloads the thread; no completion follows. + yield* harness.runtime.unloadThread!({ providerThread: harness.providerThread }); + yield* harness.awaitTerminal(1); + assert.deepEqual( + harness.terminalEvents().map((event) => event.status), + ["interrupted"], + ); + assert.isFalse(yield* harness.hasPendingBackgroundWork); }).pipe( Effect.scoped, Effect.provide( @@ -1924,6 +1947,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { durationMs: null, }; const interruptSent = yield* Deferred.make(); + const terminateSent = yield* Deferred.make(); const harness = yield* makeCodexReplayHarness( makeCodexReplayTranscript({ scenario: "late-stop", @@ -1957,10 +1981,29 @@ describe("CodexAdapterV2 post-settle continuation", () => { afterMs: 1000, frame: { id: 4, result: {} }, }, + { + type: "emit_inbound", + label: "item/started/after-timeout", + afterMs: 12_000, + frame: { + method: "item/started", + params: { + threadId: nativeThreadId, + turnId: nativeTurnId, + startedAtMs: 1782622446000, + item: { + ...commandItem, + id: "late-stop-command-after-timeout", + status: "inProgress", + processId: "late-stop-process", + }, + }, + }, + }, { type: "emit_inbound", label: "turn/completed", - afterMs: 15_000, + afterMs: 3_000, frame: { method: "turn/completed", params: { @@ -1969,11 +2012,29 @@ describe("CodexAdapterV2 post-settle continuation", () => { }, }, }, + { + type: "expect_outbound", + label: "thread/backgroundTerminals/terminate", + frame: { + id: 5, + method: "thread/backgroundTerminals/terminate", + params: { threadId: nativeThreadId, processId: "late-stop-process" }, + }, + }, + { + type: "emit_inbound", + label: "thread/backgroundTerminals/terminate", + frame: { id: 5, result: { terminated: true } }, + }, ], }), () => Effect.void, (method) => - method === "turn/interrupt" ? Deferred.succeed(interruptSent, undefined) : Effect.void, + method === "turn/interrupt" + ? Deferred.succeed(interruptSent, undefined) + : method === "thread/backgroundTerminals/terminate" + ? Deferred.succeed(terminateSent, undefined) + : Effect.void, ); yield* harness.runtime.startTurn( makeCodexTestTurnInput({ @@ -2011,6 +2072,8 @@ describe("CodexAdapterV2 post-settle continuation", () => { ["interrupted"], ); assert.equal(commandUpdates().at(-1), "interrupted"); + // A command started after the timeout is still the Stop's to contain. + yield* Deferred.await(terminateSent); }).pipe( Effect.scoped, Effect.provide( @@ -2122,6 +2185,127 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); + it.effect("keeps a Stop pending when the session manager's cap interrupts it", () => + Effect.gen(function* () { + const nativeThreadId = "capped-stop-thread"; + const nativeTurnId = "capped-stop-turn"; + const prompt = "Keep working."; + const interruptSent = yield* Deferred.make(); + const harness = yield* makeCodexReplayHarness( + makeCodexReplayTranscript({ + scenario: "capped-stop", + entries: [ + ...codexReplayPreamble({ nativeThreadId, nativeTurnId, prompt }), + { + type: "expect_outbound", + label: "turn/interrupt", + frame: { + id: 4, + method: "turn/interrupt", + params: { threadId: nativeThreadId, turnId: nativeTurnId }, + }, + }, + { type: "emit_inbound", afterMs: 6_000, frame: { id: 4, result: {} } }, + { + type: "emit_inbound", + label: "item/started/after-cap", + afterMs: 10_000, + frame: { + method: "item/started", + params: { + threadId: nativeThreadId, + turnId: nativeTurnId, + startedAtMs: 1782622446000, + item: { + type: "commandExecution", + id: "capped-stop-command", + command: "true", + cwd: "/workspace", + source: "unifiedExecStartup", + commandActions: [{ type: "unknown", command: "true" }], + aggregatedOutput: null, + exitCode: null, + durationMs: null, + status: "inProgress", + }, + }, + }, + }, + { + type: "emit_inbound", + label: "turn/completed", + afterMs: 1_000, + frame: { + method: "turn/completed", + params: { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: nativeTurnId, status: "completed" }), + }, + }, + }, + ], + }), + () => Effect.void, + (method) => + method === "turn/interrupt" ? Deferred.succeed(interruptSent, undefined) : Effect.void, + ); + yield* harness.runtime.startTurn( + makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("capped-stop-attempt"), + text: prompt, + }), + ); + yield* harness.until(() => + harness.events.some((event) => event.type === "provider_turn.updated"), + ); + const providerTurnId = (yield* IdAllocator.IdAllocatorV2).derive.providerTurn({ + driver: CodexAdapterV2.CODEX_DRIVER_KIND, + nativeTurnId, + }); + // Mirrors ProviderSessionManager's PROVIDER_INTERRUPT_TIMEOUT_MS cap on interruptTurn. + const interrupt = yield* harness.runtime + .interruptTurn({ providerThread: harness.providerThread, providerTurnId }) + .pipe( + Effect.timeoutOrElse({ duration: "15 seconds", orElse: () => Effect.fail("capped") }), + Effect.exit, + Effect.forkScoped, + ); + yield* Deferred.await(interruptSent); + yield* TestClock.adjust("6 seconds"); + yield* TestClock.adjust("9 seconds"); + const exit = yield* Fiber.join(interrupt); + assert.isTrue(Exit.isFailure(exit)); + + yield* TestClock.adjust("2 seconds"); + yield* harness.awaitTerminal(1); + const lateItemIndex = harness.events.findIndex( + (event) => + event.type === "turn_item.updated" && + event.turnItem.nativeItemRef?.nativeId === "capped-stop-command", + ); + const terminalIndex = harness.events.findIndex((event) => event.type === "turn.terminal"); + assert.isAtLeast(lateItemIndex, 0, "the turn must stay open after the cap"); + assert.isAbove(terminalIndex, lateItemIndex, "only Codex's completion ends the turn"); + assert.deepEqual( + harness.terminalEvents().map((event) => event.status), + ["interrupted"], + ); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ); + const assistantMessages = (events: ReadonlyArray) => events.filter( ( @@ -5854,6 +6038,132 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); + it.effect("settles the Stop's lineage when Codex completes an unconfirmed root later", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeCodexReplayHarness( + makeCodexReplayTranscript({ + scenario: "codex-interrupt-unconfirmed-root-late-completion", + entries: [ + ...interruptSubagentCommandTranscript.entries.slice(0, rootCompletionIndex), + { + type: "emit_inbound", + label: "turn/started/late-child", + frame: { + method: "turn/started", + params: { + threadId: INTERRUPT_CHILD_NATIVE_THREAD, + turn: makeCodexReplayTurn({ + id: INTERRUPT_LATE_CHILD_NATIVE_TURN, + status: "inProgress", + }), + }, + }, + }, + { + type: "emit_inbound", + label: "turn/completed/root", + afterMs: 15_000, + frame: { + method: "turn/completed", + params: { + threadId: INTERRUPT_NATIVE_THREAD, + turn: makeCodexReplayTurn({ id: INTERRUPT_NATIVE_TURN, status: "interrupted" }), + }, + }, + }, + { + type: "expect_outbound", + label: "turn/interrupt/late-child", + frame: { + id: 7, + method: "turn/interrupt", + params: { + threadId: INTERRUPT_CHILD_NATIVE_THREAD, + turnId: INTERRUPT_LATE_CHILD_NATIVE_TURN, + }, + }, + }, + { + type: "emit_inbound", + label: "turn/interrupt/late-child", + frame: { id: 7, result: {} }, + }, + ], + }), + ); + yield* harness.runtime.startTurn( + makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("attempt-codex-interrupt-unconfirmed-root-late"), + text: INTERRUPT_PROMPT, + }), + ); + yield* harness.until(() => + harness.events.some( + (event) => + event.type === "turn_item.updated" && + event.turnItem.type === "command_execution" && + event.turnItem.nativeItemRef?.nativeId === INTERRUPT_CHILD_COMMAND_ITEM && + event.turnItem.status === "running", + ), + ); + const providerTurnId = harness.events.find( + ( + event, + ): event is Extract< + ProviderAdapter.ProviderAdapterV2Event, + { type: "provider_turn.updated" } + > => event.type === "provider_turn.updated" && event.threadId === harness.threadId, + )?.providerTurn.id; + assert.isDefined(providerTurnId); + + const interrupt = yield* harness.runtime + .interruptTurn({ providerThread: harness.providerThread, providerTurnId }) + .pipe(Effect.exit, Effect.forkScoped); + yield* harness.until(() => + harness.events.some( + (event) => + event.type === "provider_turn.updated" && + event.providerTurn.nativeTurnRef?.nativeId === INTERRUPT_LATE_CHILD_NATIVE_TURN, + ), + ); + yield* TestClock.adjust("10 seconds"); + const exit = yield* Fiber.join(interrupt); + assert.isTrue(Exit.isFailure(exit)); + assert.deepEqual(harness.terminalEvents(), []); + + yield* TestClock.adjust("5 seconds"); + yield* harness.awaitTerminal(1); + assert.equal(harness.terminalEvents()[0]?.status, "interrupted"); + const lateChildUpdates = harness.events.filter( + ( + event, + ): event is Extract< + ProviderAdapter.ProviderAdapterV2Event, + { type: "provider_turn.updated" } + > => + event.type === "provider_turn.updated" && + event.providerTurn.nativeTurnRef?.nativeId === INTERRUPT_LATE_CHILD_NATIVE_TURN, + ); + assert.equal(lateChildUpdates.at(-1)?.providerTurn.status, "interrupted"); + assertChildProviderTerminalBeforeRoot(harness.events, harness.threadId); + assert.isFalse(yield* harness.hasPendingBackgroundWork); + }).pipe( + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ), + ); + const interruptSubagentTurnAfterStopTranscript = makeCodexReplayTranscript({ scenario: "codex-interrupt-subagent-turn-after-stop", entries: [ diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 3215feba6c76..3ae62282ef9f 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -1889,7 +1889,14 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( const pendingRootTurns = yield* Ref.make( new Map(), ); - const turnWaiters = yield* Ref.make(new Map>()); + // Each pending Stop on a native turn holds its own waiter until that Stop settles. + const turnWaiters = yield* Ref.make( + new Map>>(), + ); + const addTurnWaiter = (nativeTurnId: string, waiter: Deferred.Deferred) => + Ref.update(turnWaiters, (current) => + new Map(current).set(nativeTurnId, [...(current.get(nativeTurnId) ?? []), waiter]), + ); const subagentThreads = yield* Ref.make(new Map()); const subagentSelections = new Map>(); // A reroute names one turn; until new settings arrive, the thread's model is unknown. @@ -1987,6 +1994,26 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( const goalActivations = new Map(); // Held runs must not complete while Stop resolves its target. const goalStops = new Map(); + const holdGoalStop = ( + providerThread: ProviderAdapter.ProviderAdapterV2TurnInput["providerThread"], + effect: Effect.Effect, + ) => + Effect.acquireUseRelease( + Effect.sync(() => { + const nativeThreadId = providerThread.nativeThreadRef?.nativeId; + if (nativeThreadId != null) + goalStops.set(nativeThreadId, (goalStops.get(nativeThreadId) ?? 0) + 1); + return nativeThreadId; + }), + () => effect, + (nativeThreadId) => + Effect.sync(() => { + if (nativeThreadId == null) return; + const stops = (goalStops.get(nativeThreadId) ?? 1) - 1; + if (stops > 0) goalStops.set(nativeThreadId, stops); + else goalStops.delete(nativeThreadId); + }), + ); const latestRunTurnId = (providerTurnId: ProviderTurnId) => continuedRuns.get(providerTurnId)?.at(-1)?.providerTurnId ?? providerTurnId; const isContinuedRunDescendant = ( @@ -6342,19 +6369,12 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( ...(holdsForGoal ? { goalHoldTurn: completedTurn } : {}), }); } - const waiter = (yield* Ref.get(turnWaiters)).get(input.nativeTurnId); - if (waiter !== undefined) { + for (const waiter of (yield* Ref.get(turnWaiters)).get(input.nativeTurnId) ?? []) { yield* Deferred.succeed(waiter, undefined); } - // A waiting Stop keeps the turn's commands for its second terminal sweep and - // clears its own mark; a timed-out Stop left the mark for this finalization. - const interruptInProgress = waiter !== undefined; - yield* Ref.update(interruptingNativeTurns, (current) => { - if (interruptInProgress || !current.has(input.nativeTurnId)) return current; - const updated = new Set(current); - updated.delete(input.nativeTurnId); - return updated; - }); + const interruptInProgress = (yield* Ref.get(interruptingNativeTurns)).has( + input.nativeTurnId, + ); const retainSettledContext = input.status === "completed" && (yield* turnHasRetainedBackgroundWork(input.nativeTurnId)); @@ -7307,6 +7327,22 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( const nativeThreadId = yield* getNativeThreadId(unloadInput.providerThread); sessionApprovals.delete(nativeThreadId); yield* client.request("thread/unsubscribe", { threadId: nativeThreadId }); + // No turn/completed arrives after unsubscribing, so a Stop still waiting ends here. + const interrupting = yield* Ref.get(interruptingNativeTurns); + const completedAt = yield* DateTime.now; + for (const context of (yield* Ref.get(activeTurns)).values()) { + if ( + context.providerThread.nativeThreadRef?.nativeId !== nativeThreadId || + !interrupting.has(context.nativeTurnId) + ) + continue; + yield* finalizeCodexTurn({ + context, + nativeTurnId: context.nativeTurnId, + status: "interrupted", + completedAt, + }); + } }, (effect, unloadInput) => effect.pipe( @@ -7397,13 +7433,9 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( for (const context of interruptTargetContexts) { interruptTargets.push({ context, completion: yield* Deferred.make() }); } - yield* Ref.update(turnWaiters, (current) => { - const updated = new Map(current); - for (const target of interruptTargets) { - updated.set(target.context.nativeTurnId, target.completion); - } - return updated; - }); + for (const target of interruptTargets) { + yield* addTurnWaiter(target.context.nativeTurnId, target.completion); + } yield* Ref.update(interruptingNativeTurns, (current) => { const updated = new Set(current); for (const target of interruptTargets) { @@ -7419,31 +7451,32 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( } const cleanupInterruptState = Effect.gen(function* () { - yield* Ref.update(turnWaiters, (current) => { + const waiters = yield* Ref.updateAndGet(turnWaiters, (current) => { const updated = new Map(current); for (const target of interruptTargets) { - updated.delete(target.context.nativeTurnId); + const remaining = (updated.get(target.context.nativeTurnId) ?? []).filter( + (waiter) => waiter !== target.completion, + ); + if (remaining.length > 0) updated.set(target.context.nativeTurnId, remaining); + else updated.delete(target.context.nativeTurnId); } return updated; }); - const stillActiveTurns = yield* Ref.get(activeTurns); - // A turn Codex never confirmed stays stopped until finalizeCodexTurn clears it. + // Another Stop still pending on a turn keeps its mark and commands. + const releasedTargets = interruptTargets.filter( + (target) => !waiters.has(target.context.nativeTurnId), + ); yield* Ref.update(interruptingNativeTurns, (current) => { const updated = new Set(current); - for (const target of interruptTargets) { - if (stillActiveTurns.get(target.context.nativeTurnId) === target.context) - continue; + for (const target of releasedTargets) { updated.delete(target.context.nativeTurnId); } return updated; }); yield* Ref.update(runningCommandItemsByTurn, (current) => { const updated = new Map(current); - for (const target of interruptTargets) { + for (const target of releasedTargets) { if (settledTurnContexts.includes(target.context)) continue; - // An unconfirmed turn keeps its commands for its own later completion. - if (stillActiveTurns.get(target.context.nativeTurnId) === target.context) - continue; updated.delete(target.context.nativeTurnId); } return updated; @@ -7469,11 +7502,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( interruptTargets.push({ context, completion }); trackedInterruptNativeTurnIds.add(context.nativeTurnId); newlyDiscovered.push(context); - yield* Ref.update(turnWaiters, (current) => { - const updated = new Map(current); - updated.set(context.nativeTurnId, completion); - return updated; - }); + yield* addTurnWaiter(context.nativeTurnId, completion); yield* Ref.update(interruptingNativeTurns, (current) => { const updated = new Set(current); updated.add(context.nativeTurnId); @@ -7544,11 +7573,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( 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* addTurnWaiter(context.nativeTurnId, completion); yield* Ref.update(interruptingNativeTurns, (current) => { const updated = new Set(current); updated.add(context.nativeTurnId); @@ -7580,127 +7605,66 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( } }); - const rootConfirmed = 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(CODEX_STOP_CONFIRM_TIMEOUT_MS)), - ], - { 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 })), + ), ); - let rootStillActive = false; - if (Option.isNone(completed)) { - for (const target of interruptTargets) { - const context = target.context; - if ((yield* Ref.get(activeTurns)).get(context.nativeTurnId) !== context) { - continue; - } - if (lineageRoots.includes(context)) rootStillActive = true; - yield* Effect.logWarning("orchestration-v2.codex-interrupt-timeout", { - providerSessionId: input.providerSessionId, - providerThreadId: context.providerThread.id, - providerTurnId: context.providerTurnId, - nativeTurnId: context.nativeTurnId, - }); - } - } - // An unconfirmed root keeps its lineage open: Codex may still be working, and - // the user decides whether to Force stop it. A later turn/completed finalizes it. - if (!rootStillActive) { - yield* interruptLateDescendants; - yield* finalizeRemainingInterruptLineage; - } + }); + const awaitCompletions = (targets: typeof interruptTargets) => + Effect.forEach(targets, (target) => Deferred.await(target.completion), { + concurrency: "unbounded", + discard: true, + }); + const settleStop = Effect.gen(function* () { + yield* interruptLateDescendants; + yield* finalizeRemainingInterruptLineage; const terminationResults = yield* terminateTrackedTerminals(interruptTargets); const failedTermination = terminationResults.find((result) => !result.success); if (failedTermination !== undefined && !failedTermination.success) { - // Force stop answers both, and the timeout says Codex itself has not stopped. - if (rootStillActive) return false; return yield* failedTermination.error; } yield* turnTerminalizationPermit.withPermits(1)( @@ -7734,37 +7698,111 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( } }), ); - return !rootStillActive; }).pipe( - Effect.onError(() => finalizeRemainingInterruptLineage), + // An interrupted Stop (session teardown) must not claim Codex stopped the turn. + Effect.onError((cause) => + Cause.hasInterruptsOnly(cause) ? Effect.void : finalizeRemainingInterruptLineage, + ), Effect.ensuring(cleanupInterruptState), ); - if (!rootConfirmed) { - return yield* new ProviderAdapter.ProviderAdapterInterruptTimeoutError({ - driver: CODEX_PROVIDER, - providerThreadId: requestedInput.providerThread.id, - providerTurnId: requestedInput.providerTurnId, - timeoutMs: CODEX_STOP_CONFIRM_TIMEOUT_MS, - }); - } + // The rest of a Stop runs in the session scope: Codex may still be working while the + // user decides on Force stop, and the session manager's cap may interrupt the caller. + const settleInScope = holdGoalStop( + requestedInput.providerThread, + awaitCompletions( + interruptTargets.filter((target) => lineageRoots.includes(target.context)), + ).pipe(Effect.andThen(settleStop)), + ).pipe( + Effect.tapCause((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.void + : Effect.logWarning("orchestration-v2.codex-interrupt-settle-failed", { + providerSessionId: input.providerSessionId, + providerTurnId: requestedInput.providerTurnId, + cause, + }), + ), + Effect.forkIn(scope, { startImmediately: true }), + ); + + const confirmed = 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), + awaitCompletions(interruptTargets).pipe( + Effect.timeoutOption(CODEX_STOP_CONFIRM_TIMEOUT_MS), + ), + ], + { concurrency: "unbounded" }, + ); + if (Option.isSome(completed)) return true; + let rootStillActive = false; + for (const target of interruptTargets) { + const context = target.context; + if ((yield* Ref.get(activeTurns)).get(context.nativeTurnId) !== context) { + continue; + } + if (lineageRoots.includes(context)) rootStillActive = true; + yield* Effect.logWarning("orchestration-v2.codex-interrupt-timeout", { + providerSessionId: input.providerSessionId, + providerThreadId: context.providerThread.id, + providerTurnId: context.providerTurnId, + nativeTurnId: context.nativeTurnId, + }); + } + if (!rootStillActive) return true; + yield* terminateTrackedTerminals(interruptTargets); + return false; + }).pipe( + Effect.onError((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.asVoid(settleInScope) + : finalizeRemainingInterruptLineage.pipe( + Effect.ensuring(cleanupInterruptState), + ), + ), + ); + if (confirmed) return yield* Fiber.join(yield* settleInScope); + yield* settleInScope; + return yield* new ProviderAdapter.ProviderAdapterInterruptTimeoutError({ + driver: CODEX_PROVIDER, + providerThreadId: requestedInput.providerThread.id, + providerTurnId: requestedInput.providerTurnId, + timeoutMs: CODEX_STOP_CONFIRM_TIMEOUT_MS, + }); }, (effect, requestedInput) => - Effect.acquireUseRelease( - Effect.sync(() => { - const nativeThreadId = requestedInput.providerThread.nativeThreadRef?.nativeId; - if (nativeThreadId != null) - goalStops.set(nativeThreadId, (goalStops.get(nativeThreadId) ?? 0) + 1); - return nativeThreadId; - }), - () => effect, - (nativeThreadId) => - Effect.sync(() => { - if (nativeThreadId == null) return; - const stops = (goalStops.get(nativeThreadId) ?? 1) - 1; - if (stops > 0) goalStops.set(nativeThreadId, stops); - else goalStops.delete(nativeThreadId); - }), - ).pipe( + holdGoalStop(requestedInput.providerThread, effect).pipe( Effect.mapError((cause) => cause._tag === "ProviderAdapterInterruptTimeoutError" ? cause From 4c0a5533e575a3aa1285a9a3a71acbe1ebe9966f Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 4 Nov 2026 18:02:29 +0000 Subject: [PATCH 5/9] fix(server): a Codex Stop that times out contains descendants that settled meanwhile Co-Authored-By: Claude Opus 5.5 --- .../Adapters/CodexAdapterV2.test.ts | 144 ++++++++++++++++++ .../Adapters/CodexAdapterV2.ts | 13 +- 2 files changed, 156 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 703aff7f1679..3ff23ac90a94 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -2306,6 +2306,150 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); + it.effect( + "contains a descendant that settles after a timed-out Stop and before its root completes", + () => + Effect.gen(function* () { + const nativeThreadId = "late-settled-root"; + const nativeTurnId = "late-settled-root-turn"; + const childThreadId = "late-settled-child"; + const childTurnId = "late-settled-child-turn"; + const prompt = "Keep working."; + const interruptSent = yield* Deferred.make(); + const terminated: string[] = []; + const harness = yield* makeCodexReplayHarness( + makeCodexReplayTranscript({ + scenario: "late-settled-settled-descendant", + entries: [ + ...codexReplayPreamble({ nativeThreadId, nativeTurnId, prompt }), + { + type: "expect_outbound", + frame: { + id: 4, + method: "turn/interrupt", + params: { threadId: nativeThreadId, turnId: nativeTurnId }, + }, + }, + { type: "emit_inbound", afterMs: 1_000, frame: { id: 4, result: {} } }, + { + type: "emit_inbound", + afterMs: 12_000, + frame: { + method: "item/completed", + params: { + threadId: nativeThreadId, + turnId: nativeTurnId, + completedAtMs: 1782622441000, + item: { + type: "subAgentActivity", + id: "late-settled-spawn", + kind: "started", + agentThreadId: childThreadId, + agentPath: "/root/late_settled_child", + }, + }, + }, + }, + { + type: "emit_inbound", + frame: { + method: "turn/started", + params: { + threadId: childThreadId, + turn: makeCodexReplayTurn({ id: childTurnId, status: "inProgress" }), + }, + }, + }, + { + type: "emit_inbound", + frame: { + method: "item/started", + params: { + threadId: childThreadId, + turnId: childTurnId, + startedAtMs: 1782622442000, + item: { + ...interruptCommandItem("inProgress"), + id: "late-settled-child-command", + processId: "late-settled-child-process", + }, + }, + }, + }, + { + type: "emit_inbound", + frame: { + method: "turn/completed", + params: { + threadId: childThreadId, + turn: makeCodexReplayTurn({ id: childTurnId, status: "completed" }), + }, + }, + }, + { + type: "emit_inbound", + afterMs: 3_000, + frame: { + method: "turn/completed", + params: { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: nativeTurnId, status: "interrupted" }), + }, + }, + }, + ], + }), + () => Effect.void, + (method, params) => + Effect.gen(function* () { + if (method === "turn/interrupt") yield* Deferred.succeed(interruptSent, undefined); + if (method === "thread/backgroundTerminals/terminate") + terminated.push((params as { processId: string }).processId); + }), + ); + yield* harness.runtime.startTurn( + makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("late-settled-attempt"), + text: prompt, + }), + ); + yield* harness.until(() => + harness.events.some((event) => event.type === "provider_turn.updated"), + ); + const providerTurnId = (yield* IdAllocator.IdAllocatorV2).derive.providerTurn({ + driver: CodexAdapterV2.CODEX_DRIVER_KIND, + nativeTurnId, + }); + const interrupt = yield* harness.runtime + .interruptTurn({ providerThread: harness.providerThread, providerTurnId }) + .pipe(Effect.exit, Effect.forkScoped); + yield* Deferred.await(interruptSent); + yield* TestClock.adjust("1 second"); + yield* TestClock.adjust("10 seconds"); + assert.isTrue(Exit.isFailure(yield* Fiber.join(interrupt))); + yield* TestClock.adjust("5 seconds"); + yield* harness.awaitTerminal(1); + const pending = yield* harness.hasPendingBackgroundWork; + assert.deepEqual( + { terminated, pending }, + { terminated: ["late-settled-child-process"], pending: false }, + ); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ); + const assistantMessages = (events: ReadonlyArray) => events.filter( ( diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 3ae62282ef9f..7cfd16089a3f 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -7569,8 +7569,19 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( inLineage(context) && !trackedInterruptNativeTurnIds.has(context.nativeTurnId), ); - for (const context of lineage) { + // Descendants that already settled need no interrupt, only containment. + const settledLineage = Array.from( + (yield* Ref.get(settledTurns)).values(), + ).filter( + (context) => + inLineage(context) && + !trackedInterruptNativeTurnIds.has(context.nativeTurnId), + ); + for (const context of [...lineage, ...settledLineage]) { const completion = yield* Deferred.make(); + if (settledLineage.includes(context)) { + yield* Deferred.succeed(completion, undefined); + } interruptTargets.push({ context, completion }); trackedInterruptNativeTurnIds.add(context.nativeTurnId); yield* addTurnWaiter(context.nativeTurnId, completion); From 6e62881f1d848a4a37b4d83e5b1e407ff86ea905 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 4 Nov 2026 18:11:06 +0000 Subject: [PATCH 6/9] fix(server): a Codex Stop keeps a late settled descendant's commands when containment fails Co-Authored-By: Claude Opus 5.5 --- .../Adapters/CodexAdapterV2.test.ts | 163 ++++++++++++++++++ .../Adapters/CodexAdapterV2.ts | 5 +- 2 files changed, 167 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 3ff23ac90a94..a0727ceefe60 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -2397,6 +2397,15 @@ describe("CodexAdapterV2 post-settle continuation", () => { }, }, }, + { + type: "expect_outbound", + frame: { + id: 5, + method: "thread/backgroundTerminals/terminate", + params: { threadId: childThreadId, processId: "late-settled-child-process" }, + }, + }, + { type: "emit_inbound", frame: { id: 5, result: { terminated: true } } }, ], }), () => Effect.void, @@ -2450,6 +2459,160 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); + it.effect("keeps tracking a late settled descendant's command when containing it fails", () => + Effect.gen(function* () { + const nativeThreadId = "late-settled-root"; + const nativeTurnId = "late-settled-root-turn"; + const childThreadId = "late-settled-child"; + const childTurnId = "late-settled-child-turn"; + const prompt = "Keep working."; + const interruptSent = yield* Deferred.make(); + const terminated: string[] = []; + const harness = yield* makeCodexReplayHarness( + makeCodexReplayTranscript({ + scenario: "late-settled-descendant-terminate-fails", + entries: [ + ...codexReplayPreamble({ nativeThreadId, nativeTurnId, prompt }), + { + type: "expect_outbound", + frame: { + id: 4, + method: "turn/interrupt", + params: { threadId: nativeThreadId, turnId: nativeTurnId }, + }, + }, + { type: "emit_inbound", afterMs: 1_000, frame: { id: 4, result: {} } }, + { + type: "emit_inbound", + afterMs: 12_000, + frame: { + method: "item/completed", + params: { + threadId: nativeThreadId, + turnId: nativeTurnId, + completedAtMs: 1782622441000, + item: { + type: "subAgentActivity", + id: "late-settled-spawn", + kind: "started", + agentThreadId: childThreadId, + agentPath: "/root/late_settled_child", + }, + }, + }, + }, + { + type: "emit_inbound", + frame: { + method: "turn/started", + params: { + threadId: childThreadId, + turn: makeCodexReplayTurn({ id: childTurnId, status: "inProgress" }), + }, + }, + }, + { + type: "emit_inbound", + frame: { + method: "item/started", + params: { + threadId: childThreadId, + turnId: childTurnId, + startedAtMs: 1782622442000, + item: { + ...interruptCommandItem("inProgress"), + id: "late-settled-child-command", + processId: "late-settled-child-process", + }, + }, + }, + }, + { + type: "emit_inbound", + frame: { + method: "turn/completed", + params: { + threadId: childThreadId, + turn: makeCodexReplayTurn({ id: childTurnId, status: "completed" }), + }, + }, + }, + { + type: "emit_inbound", + afterMs: 3_000, + frame: { + method: "turn/completed", + params: { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: nativeTurnId, status: "interrupted" }), + }, + }, + }, + { + type: "expect_outbound", + frame: { + id: 5, + method: "thread/backgroundTerminals/terminate", + params: { threadId: childThreadId, processId: "late-settled-child-process" }, + }, + }, + { + type: "emit_inbound", + frame: { id: 5, error: { code: -32603, message: "terminate failed" } }, + }, + ], + }), + () => Effect.void, + (method, params) => + Effect.gen(function* () { + if (method === "turn/interrupt") yield* Deferred.succeed(interruptSent, undefined); + if (method === "thread/backgroundTerminals/terminate") + terminated.push((params as { processId: string }).processId); + }), + ); + yield* harness.runtime.startTurn( + makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("late-settled-fail-attempt"), + text: prompt, + }), + ); + yield* harness.until(() => + harness.events.some((event) => event.type === "provider_turn.updated"), + ); + const providerTurnId = (yield* IdAllocator.IdAllocatorV2).derive.providerTurn({ + driver: CodexAdapterV2.CODEX_DRIVER_KIND, + nativeTurnId, + }); + const interrupt = yield* harness.runtime + .interruptTurn({ providerThread: harness.providerThread, providerTurnId }) + .pipe(Effect.exit, Effect.forkScoped); + yield* Deferred.await(interruptSent); + yield* TestClock.adjust("1 second"); + yield* TestClock.adjust("10 seconds"); + assert.isTrue(Exit.isFailure(yield* Fiber.join(interrupt))); + yield* TestClock.adjust("5 seconds"); + yield* harness.awaitTerminal(1); + const pending = yield* harness.hasPendingBackgroundWork; + assert.deepEqual( + { terminated, pending }, + { terminated: ["late-settled-child-process"], pending: true }, + ); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ); + const assistantMessages = (events: ReadonlyArray) => events.filter( ( diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 7cfd16089a3f..8fd2bedc716e 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -7450,6 +7450,8 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( } } + // Settled targets keep their commands unless settleStop contained them. + const settledTargets = new Set(settledTurnContexts); const cleanupInterruptState = Effect.gen(function* () { const waiters = yield* Ref.updateAndGet(turnWaiters, (current) => { const updated = new Map(current); @@ -7476,7 +7478,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( yield* Ref.update(runningCommandItemsByTurn, (current) => { const updated = new Map(current); for (const target of releasedTargets) { - if (settledTurnContexts.includes(target.context)) continue; + if (settledTargets.has(target.context)) continue; updated.delete(target.context.nativeTurnId); } return updated; @@ -7577,6 +7579,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( inLineage(context) && !trackedInterruptNativeTurnIds.has(context.nativeTurnId), ); + for (const context of settledLineage) settledTargets.add(context); for (const context of [...lineage, ...settledLineage]) { const completion = yield* Deferred.make(); if (settledLineage.includes(context)) { From 9d1ec54d62de070fb5c6c2bbdb2a24cb1572b6e4 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 11 Nov 2026 00:26:42 +0000 Subject: [PATCH 7/9] fix(server): a provider switch proceeds when the old Codex turn never stops A mid-turn provider switch (restart with a detach or replace session transition) now treats the interrupt timeout as done: the detach that follows unloads the thread, which ends the turn Codex never completed. An in-session restart still fails on the timeout. AI-assisted; verified with the tests and tools listed above. --- .../src/orchestration-v2/EffectWorker.ts | 1 + .../ProviderTurnControlService.test.ts | 70 +++++++++++++++---- .../ProviderTurnControlService.ts | 19 +++-- 3 files changed, 72 insertions(+), 18 deletions(-) diff --git a/apps/server/src/orchestration-v2/EffectWorker.ts b/apps/server/src/orchestration-v2/EffectWorker.ts index c5b5ea9b383a..1868fc2b4016 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.ts @@ -347,6 +347,7 @@ export const layerExecutor: Layer.Layer< providerThreadId: effect.request.providerThreadId, providerTurnId: effect.request.providerTurnId, interruptedAttemptId: effect.request.interruptedAttemptId, + detachesSession: effect.request.sessionTransition !== undefined, ...(effect.request.sessionTransition?.type === "replace" ? { replacementProviderSessionId: diff --git a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts index 8b30a961c099..7899e77f536f 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts @@ -24,7 +24,7 @@ import * as Stream from "effect/Stream"; import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; -import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; +import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; import * as ProviderSessionManager from "./ProviderSessionManager.ts"; import * as ProviderTurnControlService from "./ProviderTurnControlService.ts"; @@ -163,6 +163,7 @@ it.effect( makeProjection({ now, threadId, providerThread, providerTurnId, attemptId }), ); const interruptedThread = yield* Ref.make(null); + const interruptTimesOut = yield* Ref.make(false); const providerSession = { id: oldSessionId, driver, @@ -186,19 +187,32 @@ it.effect( startTurn: () => Effect.die("unused startTurn"), steerTurn: () => Effect.die("unused steerTurn"), interruptTurn: ({ providerThread: target }) => - Effect.all( - [ - Ref.set(interruptedThread, target), - Ref.update(projection, (current) => ({ - ...current, - providerTurns: current.providerTurns.map((turn) => - turn.id === providerTurnId - ? { ...turn, status: "interrupted" as const, completedAt: now } - : turn, - ), - })), - ], - { discard: true }, + Effect.gen(function* () { + if (yield* Ref.get(interruptTimesOut)) { + return yield* new ProviderAdapter.ProviderAdapterInterruptTimeoutError({ + driver, + providerThreadId: target.id, + providerTurnId, + timeoutMs: 10_000, + }); + } + }).pipe( + Effect.andThen( + Effect.all( + [ + Ref.set(interruptedThread, target), + Ref.update(projection, (current) => ({ + ...current, + providerTurns: current.providerTurns.map((turn) => + turn.id === providerTurnId + ? { ...turn, status: "interrupted" as const, completedAt: now } + : turn, + ), + })), + ], + { discard: true }, + ), + ), ), respondToRuntimeRequest: () => Effect.die("unused respondToRuntimeRequest"), readThreadSnapshot: () => Effect.die("unused readThreadSnapshot"), @@ -345,5 +359,33 @@ it.effect( assert.equal(interrupted?.providerSessionId, oldSessionId); assert.equal(interrupted?.id, providerThreadId); assert.equal(interrupted?.nativeThreadRef?.nativeId, "native-thread:restart-session"); + + // A provider that never completes the interrupted turn blocks an in-session restart, + // while a restart that detaches the session proceeds and lets the detach end the turn. + yield* Ref.set(interruptTimesOut, true); + yield* Ref.update(projection, (current) => ({ + ...current, + providerTurns: current.providerTurns.map((turn) => + turn.id === providerTurnId ? { ...turn, status: "running" as const } : turn, + ), + })); + const [inSession, detaching] = yield* Effect.gen(function* () { + const control = yield* ProviderTurnControlService.ProviderTurnControlServiceV2; + const restart = (detachesSession: boolean) => + Effect.exit( + control.interruptAndAwaitTerminal({ + threadId, + providerSessionId: oldSessionId, + replacementProviderSessionId: replacementSessionId, + providerThreadId, + providerTurnId, + interruptedAttemptId: attemptId, + detachesSession, + }), + ); + return [yield* restart(false), yield* restart(true)] as const; + }).pipe(Effect.provide(layerControl)); + assert.isTrue(Exit.isFailure(inSession)); + assert.isTrue(Exit.isSuccess(detaching)); }), ); diff --git a/apps/server/src/orchestration-v2/ProviderTurnControlService.ts b/apps/server/src/orchestration-v2/ProviderTurnControlService.ts index bd225bb41773..ddb82aa3a42f 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnControlService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnControlService.ts @@ -75,6 +75,7 @@ export interface ProviderTurnControlServiceV2Shape { readonly providerThreadId: ProviderThreadId; readonly providerTurnId: ProviderTurnId; readonly interruptedAttemptId: RunAttemptId; + readonly detachesSession?: boolean; }) => Effect.Effect; } @@ -290,10 +291,20 @@ export const layer: Layer.Layer< return; } - yield* loaded.session.value.interruptTurn({ - providerThread: loaded.providerThread, - providerTurnId: loaded.providerTurn.id, - }); + const interrupted = yield* loaded.session.value + .interruptTurn({ + providerThread: loaded.providerThread, + providerTurnId: loaded.providerTurn.id, + }) + .pipe( + Effect.as(true), + // A detach unloads the thread, which ends a turn the provider never completed. + Effect.catchTags({ + ProviderAdapterInterruptTimeoutError: (cause) => + input.detachesSession === true ? Effect.succeed(false) : Effect.fail(cause), + }), + ); + if (!interrupted) return; for (let remaining = 1_000; remaining > 0; remaining -= 1) { const { providerTurn, attempt } = yield* projections.getProviderControlContext( From 2d6bdb9c72f3915922830496d57929cd4eb0708b Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Nov 2026 11:27:26 +0000 Subject: [PATCH 8/9] fix(server): a timed-out Codex Stop still stops children that start late Settling a Stop waits for the root's turn/completed, so after the confirmation timeout a descendant that started late kept running. At the timeout, interrupt the descendants found so far, and interrupt any child turn that starts under a root whose Stop is still unconfirmed. Drops the settled-descendant containment that only existed for children that could now run past the timeout. AI-assisted; verified with the tests and tools listed above. --- .../Adapters/CodexAdapterV2.test.ts | 48 ++++++------------- .../Adapters/CodexAdapterV2.ts | 30 +++++------- 2 files changed, 27 insertions(+), 51 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index b15c264c8284..3e505bc07705 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -2351,7 +2351,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { ); it.effect( - "contains a descendant that settles after a timed-out Stop and before its root completes", + "interrupts a descendant that starts after a timed-out Stop and before its root completes", () => Effect.gen(function* () { const nativeThreadId = "late-settled-root"; @@ -2360,7 +2360,6 @@ describe("CodexAdapterV2 post-settle continuation", () => { const childTurnId = "late-settled-child-turn"; const prompt = "Keep working."; const interruptSent = yield* Deferred.make(); - const terminated: string[] = []; const harness = yield* makeCodexReplayHarness( makeCodexReplayTranscript({ scenario: "late-settled-settled-descendant", @@ -2405,28 +2404,21 @@ describe("CodexAdapterV2 post-settle continuation", () => { }, }, { - type: "emit_inbound", + type: "expect_outbound", frame: { - method: "item/started", - params: { - threadId: childThreadId, - turnId: childTurnId, - startedAtMs: 1782622442000, - item: { - ...interruptCommandItem("inProgress"), - id: "late-settled-child-command", - processId: "late-settled-child-process", - }, - }, + id: 5, + method: "turn/interrupt", + params: { threadId: childThreadId, turnId: childTurnId }, }, }, + { type: "emit_inbound", frame: { id: 5, result: {} } }, { type: "emit_inbound", frame: { method: "turn/completed", params: { threadId: childThreadId, - turn: makeCodexReplayTurn({ id: childTurnId, status: "completed" }), + turn: makeCodexReplayTurn({ id: childTurnId, status: "interrupted" }), }, }, }, @@ -2441,24 +2433,11 @@ describe("CodexAdapterV2 post-settle continuation", () => { }, }, }, - { - type: "expect_outbound", - frame: { - id: 5, - method: "thread/backgroundTerminals/terminate", - params: { threadId: childThreadId, processId: "late-settled-child-process" }, - }, - }, - { type: "emit_inbound", frame: { id: 5, result: { terminated: true } } }, ], }), () => Effect.void, - (method, params) => - Effect.gen(function* () { - if (method === "turn/interrupt") yield* Deferred.succeed(interruptSent, undefined); - if (method === "thread/backgroundTerminals/terminate") - terminated.push((params as { processId: string }).processId); - }), + (method) => + method === "turn/interrupt" ? Deferred.succeed(interruptSent, undefined) : Effect.void, ); yield* harness.runtime.startTurn( makeCodexTestTurnInput({ @@ -2485,11 +2464,12 @@ describe("CodexAdapterV2 post-settle continuation", () => { assert.isTrue(Exit.isFailure(yield* Fiber.join(interrupt))); yield* TestClock.adjust("5 seconds"); yield* harness.awaitTerminal(1); - const pending = yield* harness.hasPendingBackgroundWork; - assert.deepEqual( - { terminated, pending }, - { terminated: ["late-settled-child-process"], pending: false }, + const childRunning = harness.events.some( + (event) => + event.type === "provider_turn.updated" && + event.providerTurn.nativeTurnRef?.nativeId === childTurnId, ); + assert.isFalse(childRunning); }).pipe( Effect.scoped, Effect.provide( diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index b4b417293dd9..862b4497804b 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -2057,6 +2057,8 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( const goalActivations = new Map(); // Held runs must not complete while Stop resolves its target. const goalStops = new Map(); + // Native turns whose Stop timed out unconfirmed; children they start are stopped on sight. + const unconfirmedStopTurns = new Set(); const holdGoalStop = ( providerThread: ProviderAdapter.ProviderAdapterV2TurnInput["providerThread"], effect: Effect.Effect, @@ -2860,7 +2862,10 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( const terminalizedNativeTurns = yield* Ref.get(terminalizedNonCompletedNativeTurns); let ancestor: ActiveCodexTurnContext | undefined = subagent.parentContext; while (ancestor !== undefined) { - if (terminalizedNativeTurns.has(ancestor.nativeTurnId)) { + if ( + terminalizedNativeTurns.has(ancestor.nativeTurnId) || + unconfirmedStopTurns.has(ancestor.nativeTurnId) + ) { const nativeThreadId = yield* getNativeThreadId(subagent.providerThread); // Fork requests off the notification reader so it can read their responses. yield* client @@ -7963,8 +7968,6 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( } } - // Settled targets keep their commands unless settleStop contained them. - const settledTargets = new Set(settledTurnContexts); const cleanupInterruptState = Effect.gen(function* () { const waiters = yield* Ref.updateAndGet(turnWaiters, (current) => { const updated = new Map(current); @@ -7988,10 +7991,13 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( } return updated; }); + for (const target of releasedTargets) { + unconfirmedStopTurns.delete(target.context.nativeTurnId); + } yield* Ref.update(runningCommandItemsByTurn, (current) => { const updated = new Map(current); for (const target of releasedTargets) { - if (settledTargets.has(target.context)) continue; + if (settledTurnContexts.includes(target.context)) continue; updated.delete(target.context.nativeTurnId); } return updated; @@ -8084,20 +8090,8 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( inLineage(context) && !trackedInterruptNativeTurnIds.has(context.nativeTurnId), ); - // Descendants that already settled need no interrupt, only containment. - const settledLineage = Array.from( - (yield* Ref.get(settledTurns)).values(), - ).filter( - (context) => - inLineage(context) && - !trackedInterruptNativeTurnIds.has(context.nativeTurnId), - ); - for (const context of settledLineage) settledTargets.add(context); - for (const context of [...lineage, ...settledLineage]) { + for (const context of lineage) { const completion = yield* Deferred.make(); - if (settledLineage.includes(context)) { - yield* Deferred.succeed(completion, undefined); - } interruptTargets.push({ context, completion }); trackedInterruptNativeTurnIds.add(context.nativeTurnId); yield* addTurnWaiter(context.nativeTurnId, completion); @@ -8308,6 +8302,8 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( }); } if (!rootStillActive) return true; + for (const root of lineageRoots) unconfirmedStopTurns.add(root.nativeTurnId); + yield* interruptLateDescendants; yield* terminateTrackedTerminals(interruptTargets); return false; }).pipe( From 1cfb434b557fe57cd5b43699e942b01d60268d77 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Nov 2026 11:44:20 +0000 Subject: [PATCH 9/9] fix(server): a capped Codex Stop still stops children that start late When the session manager's cap interrupts a Stop before its confirmation timeout, the settlement fork waited for the root without marking it unconfirmed, so a child starting afterwards ran unstopped. The capped path now marks the roots and interrupts late descendants, like the timeout path. AI-assisted; verified with the tests and tools listed above. --- .../Adapters/CodexAdapterV2.test.ts | 45 +++++++++++++++++++ .../Adapters/CodexAdapterV2.ts | 12 +++-- 2 files changed, 54 insertions(+), 3 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 3e505bc07705..54f7180c074d 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -2275,6 +2275,43 @@ describe("CodexAdapterV2 post-settle continuation", () => { }, }, }, + { + type: "emit_inbound", + frame: { + method: "item/completed", + params: { + threadId: nativeThreadId, + turnId: nativeTurnId, + completedAtMs: 1782622446000, + item: { + type: "subAgentActivity", + id: "capped-stop-spawn", + kind: "started", + agentThreadId: "capped-stop-child", + agentPath: "/root/capped_stop_child", + }, + }, + }, + }, + { + type: "emit_inbound", + frame: { + method: "turn/started", + params: { + threadId: "capped-stop-child", + turn: makeCodexReplayTurn({ id: "capped-stop-child-turn", status: "inProgress" }), + }, + }, + }, + { + type: "expect_outbound", + frame: { + id: 5, + method: "turn/interrupt", + params: { threadId: "capped-stop-child", turnId: "capped-stop-child-turn" }, + }, + }, + { type: "emit_inbound", frame: { id: 5, result: {} } }, { type: "emit_inbound", label: "turn/completed", @@ -2337,6 +2374,14 @@ describe("CodexAdapterV2 post-settle continuation", () => { harness.terminalEvents().map((event) => event.status), ["interrupted"], ); + assert.isFalse( + harness.events.some( + (event) => + event.type === "provider_turn.updated" && + event.providerTurn.nativeTurnRef?.nativeId === "capped-stop-child-turn", + ), + "a child starting after the cap is stopped, not run", + ); }).pipe( Effect.scoped, Effect.provide( diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 862b4497804b..1e355e172364 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -8246,6 +8246,10 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( Effect.forkIn(scope, { startImmediately: true }), ); + // Unconfirmed (timed out or capped): children starting under these roots are stopped on sight. + const stopLateDescendants = Effect.sync(() => { + for (const root of lineageRoots) unconfirmedStopTurns.add(root.nativeTurnId); + }).pipe(Effect.andThen(interruptLateDescendants)); const confirmed = yield* Effect.gen(function* () { for (const target of interruptTargets) { const context = target.context; @@ -8302,14 +8306,16 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( }); } if (!rootStillActive) return true; - for (const root of lineageRoots) unconfirmedStopTurns.add(root.nativeTurnId); - yield* interruptLateDescendants; + yield* stopLateDescendants; yield* terminateTrackedTerminals(interruptTargets); return false; }).pipe( Effect.onError((cause) => Cause.hasInterruptsOnly(cause) - ? Effect.asVoid(settleInScope) + ? Effect.forkIn(stopLateDescendants, scope).pipe( + Effect.andThen(settleInScope), + Effect.asVoid, + ) : finalizeRemainingInterruptLineage.pipe( Effect.ensuring(cleanupInterruptState), ),