From 55e9861f98def92c256f578253b499c08fe00c61 Mon Sep 17 00:00:00 2001 From: Alex Southwell <4596216+saphid@users.noreply.github.com> Date: Wed, 4 Nov 2026 18:18:27 +1100 Subject: [PATCH 1/4] fix(server): Stop on a stalled run closes its approval and question cards Co-Authored-By: Claude Opus 5.5 --- .../BackgroundWorkStop.integration.test.ts | 85 +++++++++++++++++++ .../src/orchestration-v2/Orchestrator.ts | 17 +++- 2 files changed, 98 insertions(+), 4 deletions(-) diff --git a/apps/server/src/orchestration-v2/BackgroundWorkStop.integration.test.ts b/apps/server/src/orchestration-v2/BackgroundWorkStop.integration.test.ts index ffe8203c8cbbf..934edb1eaecef 100644 --- a/apps/server/src/orchestration-v2/BackgroundWorkStop.integration.test.ts +++ b/apps/server/src/orchestration-v2/BackgroundWorkStop.integration.test.ts @@ -11,6 +11,7 @@ import { ProviderTurnId, RunAttemptId, RunId, + RuntimeRequestId, ThreadId, TurnItemId, type OrchestrationV2DomainEvent, @@ -378,6 +379,79 @@ const stopEarlierBackgroundWork = ({ }, ], }); + // Shaped like CodexAdapterV2's buildApprovalRequestArtifacts output. + const approvalRequestId = RuntimeRequestId.make("request:stalled-approval"); + const approvalNodeId = NodeId.make("node:stalled-approval"); + const approvalItemId = TurnItemId.make("turn-item:stalled-approval"); + if (stalledRun === "returned-interrupt") { + yield* sink.write({ + events: [ + { + id: EventId.make("stalled-approval-node"), + type: "node.updated", + threadId, + runId: run.id, + occurredAt: now, + payload: { + ...node, + id: approvalNodeId, + parentNodeId: node.id, + kind: "approval_request", + status: "waiting", + countsForRun: false, + runtimeRequestId: approvalRequestId, + }, + }, + { + id: EventId.make("stalled-approval-request"), + type: "runtime-request.updated", + threadId, + occurredAt: now, + payload: { + id: approvalRequestId, + nodeId: approvalNodeId, + providerTurnId: codexTurn.id, + nativeRequestRef: { driver, nativeId: "approval", strength: "strong" }, + kind: "command", + status: "pending", + responseCapability: { + type: "live", + providerSessionId: first.providerThread.providerSessionId!, + }, + createdAt: now, + resolvedAt: null, + }, + }, + { + id: EventId.make("stalled-approval-item"), + type: "turn-item.updated", + threadId, + runId: run.id, + occurredAt: now, + payload: { + id: approvalItemId, + threadId, + runId: run.id, + nodeId: approvalNodeId, + providerThreadId: codexTurn.providerThreadId, + providerTurnId: codexTurn.id, + nativeItemRef: null, + parentItemId: null, + ordinal: 101, + status: "waiting", + title: null, + startedAt: now, + completedAt: null, + updatedAt: now, + type: "approval_request", + requestId: approvalRequestId, + requestKind: "command", + prompt: "rm -rf build", + }, + }, + ], + }); + } if (stalledRun === "superseded-attempt") { yield* sink.write({ events: [ @@ -439,6 +513,17 @@ const stopEarlierBackgroundWork = ({ interrupted ? 1 : 0, ); assert.isEmpty(after.runs.filter((candidate) => candidate.status === "waiting")); + if (stalledRun === "returned-interrupt") { + assert.deepEqual( + [ + after.runtimeRequests.find((candidate) => candidate.id === approvalRequestId) + ?.status, + after.nodes.find((candidate) => candidate.id === approvalNodeId)?.status, + after.turnItems.find((candidate) => candidate.id === approvalItemId)?.status, + ], + ["cancelled", "interrupted", "interrupted"], + ); + } return; } const settled = yield* watch( diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index b4bdbd705c290..73f9357dfe7f8 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -10666,6 +10666,14 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio .filter((subagent) => subagent.origin === "app_owned") .map((subagent) => subagent.id), ); + // Callers load only tool and message items; the request cards close with the requests below. + const requestCards = yield* projectionStore + .getThreadRecords(run.threadId, ["turnItems"], { + turnItemTypes: ["approval_request", "user_input_request"], + turnItemRunIds: [run.id], + turnItemStatuses: ["waiting"], + }) + .pipe(mapDispatchError(input.command)); const cascaded = yield* RunExecutionService.cascadeTerminalizeRunOwnedSubagents({ run, status: "interrupted", @@ -10688,13 +10696,14 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ), turnItems: new Map(), childTurnItems: new Map( - input.projection.turnItems - .filter( + [ + ...input.projection.turnItems.filter( (item) => item.runId === run.id && (item.type === "assistant_message" || item.type === "reasoning"), - ) - .map((item) => [item.id, item]), + ), + ...requestCards.turnItems, + ].map((item) => [item.id, item]), ), linkedChildThreadIds: new Set([run.threadId]), }, From 7f538a0d3b5578a9b80943f77886b615a0efa5a3 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Wed, 4 Nov 2026 08:17:35 +0000 Subject: [PATCH 2/4] fix(server): preserve stalled codex turns until force stop --- .../Adapters/CodexAdapterV2.test.ts | 26 ++++++++++++++++--- .../Adapters/CodexAdapterV2.ts | 4 ++- 2 files changed, 25 insertions(+), 5 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 9482bb5a3fe25..4f018cd152d70 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -1720,8 +1720,10 @@ describe("CodexAdapterV2 post-settle continuation", () => { }).pipe(Effect.scoped), ); - it.effect("waits for native start before interrupting an acknowledged queued turn", () => + it.effect.each([false, true])("waits for native start before Stop (cancelled: %s)", (cancelled) => Effect.gen(function* () { + const interruptAwaitingResponse = yield* Deferred.make(); + const releaseInterrupt = yield* Deferred.make(); const nativeThreadId = "early-stop-thread"; const nativeTurnId = "early-stop-turn"; const prompt = "Run a command."; @@ -1772,7 +1774,10 @@ describe("CodexAdapterV2 post-settle continuation", () => { method: "turn/completed", params: { threadId: nativeThreadId, - turn: makeCodexReplayTurn({ id: nativeTurnId, status: "interrupted" }), + turn: makeCodexReplayTurn({ + id: nativeTurnId, + status: cancelled ? "completed" : "interrupted", + }), }, }, }, @@ -1786,6 +1791,13 @@ describe("CodexAdapterV2 post-settle continuation", () => { Effect.sync(() => { if (method === "turn/interrupt") interruptSent = true; }), + undefined, + (entry) => + cancelled && entry.label === "turn/interrupt" + ? Deferred.succeed(interruptAwaitingResponse, undefined).pipe( + Effect.andThen(Deferred.await(releaseInterrupt)), + ) + : Effect.void, ); yield* harness.runtime.startTurn( makeCodexTestTurnInput({ @@ -1806,9 +1818,15 @@ describe("CodexAdapterV2 post-settle continuation", () => { yield* TestClock.adjust("500 millis"); assert.isFalse(interruptSent, "Stop must not reach Codex before native turn/started"); yield* TestClock.adjust("500 millis"); - yield* Fiber.join(interrupt); + if (cancelled) { + yield* Deferred.await(interruptAwaitingResponse); + yield* Fiber.interrupt(interrupt); + yield* Deferred.succeed(releaseInterrupt, undefined); + } else { + yield* Fiber.join(interrupt); + } yield* harness.firstTerminal; - assert.equal(harness.terminalEvents()[0]?.status, "interrupted"); + assert.equal(harness.terminalEvents()[0]?.status, cancelled ? "completed" : "interrupted"); assert.lengthOf(harness.terminalEvents(), 1); assert.isFalse(yield* harness.hasPendingBackgroundWork); }).pipe( diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 0f8e0ef1f3366..73d62b5d63686 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -7713,7 +7713,9 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( }), ); }).pipe( - Effect.onError(() => finalizeRemainingInterruptLineage), + Effect.onError((cause) => + Cause.hasInterruptsOnly(cause) ? Effect.void : finalizeRemainingInterruptLineage, + ), Effect.ensuring(cleanupInterruptState), ); }, From 21f213ce48e1138d3f5de9e9e95835a84732b16c Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Wed, 4 Nov 2026 08:33:44 +0000 Subject: [PATCH 3/4] fix(server): retain commands during cancelled stop attempts --- .../Adapters/CodexAdapterV2.test.ts | 248 ++++++++++-------- .../Adapters/CodexAdapterV2.ts | 7 +- 2 files changed, 145 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 4f018cd152d70..26eb2a3969023 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -1720,126 +1720,156 @@ describe("CodexAdapterV2 post-settle continuation", () => { }).pipe(Effect.scoped), ); - it.effect.each([false, true])("waits for native start before Stop (cancelled: %s)", (cancelled) => - Effect.gen(function* () { - const interruptAwaitingResponse = yield* Deferred.make(); - const releaseInterrupt = yield* Deferred.make(); - const nativeThreadId = "early-stop-thread"; - const nativeTurnId = "early-stop-turn"; - const prompt = "Run a command."; - const preamble = codexReplayPreamble({ nativeThreadId, nativeTurnId, prompt }); - const transcript = makeCodexReplayTranscript({ - scenario: "early-stop-await-native-start", - entries: [ - ...preamble.slice(0, -2), - { - type: "emit_inbound", - label: "turn/start/queued", - frame: { - id: 3, - result: { - turn: { - ...makeCodexReplayTurn({ id: nativeTurnId, status: "inProgress" }), - startedAt: null, + it.effect.each(["responsive", "cancelled-completed", "cancelled-interrupted"] as const)( + "waits for native start before Stop (%s)", + (outcome) => + Effect.gen(function* () { + const cancelled = outcome !== "responsive"; + const status = outcome === "cancelled-completed" ? "completed" : "interrupted"; + const interruptAwaitingResponse = yield* Deferred.make(); + const releaseInterrupt = yield* Deferred.make(); + const nativeThreadId = "early-stop-thread"; + const nativeTurnId = "early-stop-turn"; + const prompt = "Run a command."; + const preamble = codexReplayPreamble({ nativeThreadId, nativeTurnId, prompt }); + const transcript = makeCodexReplayTranscript({ + scenario: "early-stop-await-native-start", + entries: [ + ...preamble.slice(0, -2), + { + type: "emit_inbound", + label: "turn/start/queued", + frame: { + id: 3, + result: { + turn: { + ...makeCodexReplayTurn({ id: nativeTurnId, status: "inProgress" }), + startedAt: null, + }, }, }, }, - }, - { - type: "emit_inbound", - label: "turn/started", - afterMs: 1000, - frame: { - method: "turn/started", - params: { - threadId: nativeThreadId, - turn: makeCodexReplayTurn({ id: nativeTurnId, status: "inProgress" }), + { + type: "emit_inbound", + label: "turn/started", + afterMs: 1000, + frame: { + method: "turn/started", + params: { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: nativeTurnId, status: "inProgress" }), + }, }, }, - }, - { - type: "expect_outbound", - label: "turn/interrupt", - frame: { - id: 4, - method: "turn/interrupt", - params: { threadId: nativeThreadId, turnId: nativeTurnId }, + { + 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: "turn/completed", - frame: { - method: "turn/completed", - params: { - threadId: nativeThreadId, - turn: makeCodexReplayTurn({ - id: nativeTurnId, - status: cancelled ? "completed" : "interrupted", - }), + ...(outcome === "cancelled-interrupted" + ? [ + { + type: "emit_inbound" as const, + label: "item/started/unfinished-command", + frame: { + method: "item/started", + params: { + threadId: nativeThreadId, + turnId: nativeTurnId, + item: { ...backgroundCommandItem("inProgress"), processId: null }, + }, + }, + }, + ] + : []), + { type: "emit_inbound", label: "turn/interrupt", frame: { id: 4, result: {} } }, + { + type: "emit_inbound", + label: "turn/completed", + frame: { + method: "turn/completed", + params: { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ + id: nativeTurnId, + status, + }), + }, }, }, - }, - ], - }); - let interruptSent = false; - const harness = yield* makeCodexReplayHarness( - transcript, - () => Effect.void, - (method) => - Effect.sync(() => { - if (method === "turn/interrupt") interruptSent = true; + ], + }); + let interruptSent = false; + const harness = yield* makeCodexReplayHarness( + transcript, + () => Effect.void, + (method) => + Effect.sync(() => { + if (method === "turn/interrupt") interruptSent = true; + }), + undefined, + (entry) => + cancelled && entry.label === "turn/interrupt" + ? Deferred.succeed(interruptAwaitingResponse, undefined).pipe( + Effect.andThen(Deferred.await(releaseInterrupt)), + ) + : Effect.void, + ); + yield* harness.runtime.startTurn( + makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("early-stop-attempt"), + text: prompt, }), - undefined, - (entry) => - cancelled && entry.label === "turn/interrupt" - ? Deferred.succeed(interruptAwaitingResponse, undefined).pipe( - Effect.andThen(Deferred.await(releaseInterrupt)), - ) - : Effect.void, - ); - yield* harness.runtime.startTurn( - makeCodexTestTurnInput({ - threadId: harness.threadId, - providerThread: harness.providerThread, - now: yield* DateTime.now, - attemptId: RunAttemptId.make("early-stop-attempt"), - text: prompt, - }), - ); - 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.forkScoped); - yield* TestClock.adjust("500 millis"); - assert.isFalse(interruptSent, "Stop must not reach Codex before native turn/started"); - yield* TestClock.adjust("500 millis"); - if (cancelled) { - yield* Deferred.await(interruptAwaitingResponse); - yield* Fiber.interrupt(interrupt); - yield* Deferred.succeed(releaseInterrupt, undefined); - } else { - yield* Fiber.join(interrupt); - } - yield* harness.firstTerminal; - assert.equal(harness.terminalEvents()[0]?.status, cancelled ? "completed" : "interrupted"); - assert.lengthOf(harness.terminalEvents(), 1); - assert.isFalse(yield* harness.hasPendingBackgroundWork); - }).pipe( - Effect.scoped, - Effect.provide( - Layer.mergeAll( - IdAllocator.layer, - McpProviderSessions.layer, - NodeServices.layer, - NodeCrypto.layer, + ); + 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.forkScoped); + yield* TestClock.adjust("500 millis"); + assert.isFalse(interruptSent, "Stop must not reach Codex before native turn/started"); + yield* TestClock.adjust("500 millis"); + if (cancelled) { + yield* Deferred.await(interruptAwaitingResponse); + yield* Fiber.interrupt(interrupt); + yield* Deferred.succeed(releaseInterrupt, undefined); + } else { + yield* Fiber.join(interrupt); + } + yield* harness.firstTerminal; + assert.equal(harness.terminalEvents()[0]?.status, status); + assert.lengthOf(harness.terminalEvents(), 1); + if (outcome === "cancelled-interrupted") { + assert.deepEqual( + harness.events.flatMap((event) => + event.type === "turn_item.updated" && event.turnItem.type === "command_execution" + ? [event.turnItem.status] + : [], + ), + ["running", "interrupted"], + ); + } + assert.isFalse(yield* harness.hasPendingBackgroundWork); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), ), ), - ), ); const assistantMessages = (events: ReadonlyArray) => diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 73d62b5d63686..eb4ff42d5543a 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -7425,10 +7425,15 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( } return updated; }); + const remainingActiveTurns = 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; + if ( + settledTurnContexts.includes(target.context) || + remainingActiveTurns.get(target.context.nativeTurnId) === target.context + ) + continue; updated.delete(target.context.nativeTurnId); } return updated; From a70ffec9c507525f316fcdd11967c66c0fe87d44 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Sat, 14 Nov 2026 14:50:57 +1100 Subject: [PATCH 4/4] fix(server): keep Codex Stop settlement from main #36331 replaced the cancelled-stop command retention in this branch. The merge keeps that settlement and the native-start Stop test. --- .../Adapters/CodexAdapterV2.test.ts | 233 +++++++----------- .../Adapters/CodexAdapterV2.ts | 7 +- 2 files changed, 95 insertions(+), 145 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 2265d1390d21a..df827e3e74fef 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -1933,156 +1933,111 @@ describe("CodexAdapterV2 post-settle continuation", () => { }), ); - it.effect.each(["responsive", "cancelled-completed", "cancelled-interrupted"] as const)( - "waits for native start before Stop (%s)", - (outcome) => - Effect.gen(function* () { - const cancelled = outcome !== "responsive"; - const status = outcome === "cancelled-completed" ? "completed" : "interrupted"; - const interruptAwaitingResponse = yield* Deferred.make(); - const releaseInterrupt = yield* Deferred.make(); - const nativeThreadId = "early-stop-thread"; - const nativeTurnId = "early-stop-turn"; - const prompt = "Run a command."; - const preamble = codexReplayPreamble({ nativeThreadId, nativeTurnId, prompt }); - const transcript = makeCodexReplayTranscript({ - scenario: "early-stop-await-native-start", - entries: [ - ...preamble.slice(0, -2), - { - type: "emit_inbound", - label: "turn/start/queued", - frame: { - id: 3, - result: { - turn: { - ...makeCodexReplayTurn({ id: nativeTurnId, status: "inProgress" }), - startedAt: null, - }, + it.effect("waits for native start before Stop", () => + Effect.gen(function* () { + const nativeThreadId = "early-stop-thread"; + const nativeTurnId = "early-stop-turn"; + const prompt = "Run a command."; + const preamble = codexReplayPreamble({ nativeThreadId, nativeTurnId, prompt }); + const transcript = makeCodexReplayTranscript({ + scenario: "early-stop-await-native-start", + entries: [ + ...preamble.slice(0, -2), + { + type: "emit_inbound", + label: "turn/start/queued", + frame: { + id: 3, + result: { + turn: { + ...makeCodexReplayTurn({ id: nativeTurnId, status: "inProgress" }), + startedAt: null, }, }, }, - { - type: "emit_inbound", - label: "turn/started", - afterMs: 1000, - frame: { - method: "turn/started", - params: { - threadId: nativeThreadId, - turn: makeCodexReplayTurn({ id: nativeTurnId, status: "inProgress" }), - }, + }, + { + type: "emit_inbound", + label: "turn/started", + afterMs: 1000, + frame: { + method: "turn/started", + params: { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: nativeTurnId, status: "inProgress" }), }, }, - { - type: "expect_outbound", - label: "turn/interrupt", - frame: { - id: 4, - method: "turn/interrupt", - params: { threadId: nativeThreadId, turnId: nativeTurnId }, - }, + }, + { + type: "expect_outbound", + label: "turn/interrupt", + frame: { + id: 4, + method: "turn/interrupt", + params: { threadId: nativeThreadId, turnId: nativeTurnId }, }, - ...(outcome === "cancelled-interrupted" - ? [ - { - type: "emit_inbound" as const, - label: "item/started/unfinished-command", - frame: { - method: "item/started", - params: { - threadId: nativeThreadId, - turnId: nativeTurnId, - item: { ...backgroundCommandItem("inProgress"), processId: null }, - }, - }, - }, - ] - : []), - { type: "emit_inbound", label: "turn/interrupt", frame: { id: 4, result: {} } }, - { - type: "emit_inbound", - label: "turn/completed", - frame: { - method: "turn/completed", - params: { - threadId: nativeThreadId, - turn: makeCodexReplayTurn({ - id: nativeTurnId, - status, - }), - }, + }, + { type: "emit_inbound", label: "turn/interrupt", frame: { id: 4, result: {} } }, + { + type: "emit_inbound", + label: "turn/completed", + frame: { + method: "turn/completed", + params: { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ + id: nativeTurnId, + status: "interrupted", + }), }, }, - ], - }); - let interruptSent = false; - const harness = yield* makeCodexReplayHarness( - transcript, - () => Effect.void, - (method) => - Effect.sync(() => { - if (method === "turn/interrupt") interruptSent = true; - }), - undefined, - (entry) => - cancelled && entry.label === "turn/interrupt" - ? Deferred.succeed(interruptAwaitingResponse, undefined).pipe( - Effect.andThen(Deferred.await(releaseInterrupt)), - ) - : Effect.void, - ); - yield* harness.runtime.startTurn( - makeCodexTestTurnInput({ - threadId: harness.threadId, - providerThread: harness.providerThread, - now: yield* DateTime.now, - attemptId: RunAttemptId.make("early-stop-attempt"), - text: prompt, + }, + ], + }); + let interruptSent = false; + const harness = yield* makeCodexReplayHarness( + transcript, + () => Effect.void, + (method) => + Effect.sync(() => { + if (method === "turn/interrupt") interruptSent = true; }), - ); - 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.forkScoped); - yield* TestClock.adjust("500 millis"); - assert.isFalse(interruptSent, "Stop must not reach Codex before native turn/started"); - yield* TestClock.adjust("500 millis"); - if (cancelled) { - yield* Deferred.await(interruptAwaitingResponse); - yield* Fiber.interrupt(interrupt); - yield* Deferred.succeed(releaseInterrupt, undefined); - } else { - yield* Fiber.join(interrupt); - } - yield* harness.firstTerminal; - assert.equal(harness.terminalEvents()[0]?.status, status); - assert.lengthOf(harness.terminalEvents(), 1); - if (outcome === "cancelled-interrupted") { - assert.deepEqual( - harness.events.flatMap((event) => - event.type === "turn_item.updated" && event.turnItem.type === "command_execution" - ? [event.turnItem.status] - : [], - ), - ["running", "interrupted"], - ); - } - assert.isFalse(yield* harness.hasPendingBackgroundWork); - }).pipe( - Effect.scoped, - Effect.provide( - Layer.mergeAll( - IdAllocator.layer, - McpProviderSessions.layer, - NodeServices.layer, - NodeCrypto.layer, - ), + ); + yield* harness.runtime.startTurn( + makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("early-stop-attempt"), + text: prompt, + }), + ); + 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.forkScoped); + yield* TestClock.adjust("500 millis"); + assert.isFalse(interruptSent, "Stop must not reach Codex before native turn/started"); + yield* TestClock.adjust("500 millis"); + yield* Fiber.join(interrupt); + yield* harness.firstTerminal; + assert.equal(harness.terminalEvents()[0]?.status, "interrupted"); + assert.lengthOf(harness.terminalEvents(), 1); + assert.isFalse(yield* harness.hasPendingBackgroundWork); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, ), ), + ), ); it.effect.each(["subagent", "remote", "toggled-off"] as const)( diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 7a98ab217149b..e26b3dc5860c2 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -8056,15 +8056,10 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( } return updated; }); - const remainingActiveTurns = yield* Ref.get(activeTurns); yield* Ref.update(runningCommandItemsByTurn, (current) => { const updated = new Map(current); for (const target of interruptTargets) { - if ( - settledTurnContexts.includes(target.context) || - remainingActiveTurns.get(target.context.nativeTurnId) === target.context - ) - continue; + if (settledTurnContexts.includes(target.context)) continue; updated.delete(target.context.nativeTurnId); } return updated;