diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 6e5fd3516e0b..54f7180c074d 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -1966,6 +1966,568 @@ 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: {} }, + }, + { + 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, + (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); + + // 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( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ); + + 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 terminateSent = 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: "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: 3_000, + frame: { + method: "turn/completed", + params: { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: nativeTurnId, status: "completed" }), + }, + }, + }, + { + 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) + : method === "thread/backgroundTerminals/terminate" + ? Deferred.succeed(terminateSent, 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"); + // A command started after the timeout is still the Stop's to contain. + yield* Deferred.await(terminateSent); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ); + + 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", + 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", + 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"], + ); + 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( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ); + + it.effect( + "interrupts a descendant that starts 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 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: "expect_outbound", + frame: { + 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: "interrupted" }), + }, + }, + }, + { + type: "emit_inbound", + afterMs: 3_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-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 childRunning = harness.events.some( + (event) => + event.type === "provider_turn.updated" && + event.providerTurn.nativeTurnRef?.nativeId === childTurnId, + ); + assert.isFalse(childRunning); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ); + it.effect.each(["subagent", "remote", "toggled-off"] as const)( "never samples a suggestion for a %s thread", (delegation) => diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 4e8fd1ac5f30..1e355e172364 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -158,6 +158,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; // Codex can announce a retry and never resume. Recovered retries resume model output within ~75s. const CODEX_PROVIDER_RETRY_STALL_TIMEOUT = "10 minutes"; export const CODEX_DRIVER_KIND = CODEX_PROVIDER; @@ -1942,7 +1944,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. @@ -2048,6 +2057,28 @@ 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, + ) => + 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); + }), + ); // Goals paused because a delegated task will wake the thread; the next ordinary turn resumes them. const goalsPausedForWake = new Set(); const latestRunTurnId = (providerTurnId: ProviderTurnId) => @@ -2831,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 @@ -6733,8 +6767,7 @@ 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); } const interruptInProgress = (yield* Ref.get(interruptingNativeTurns)).has( @@ -7812,6 +7845,22 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( threadClosings.set(nativeThreadId, yield* Deferred.make()); } 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( @@ -7902,13 +7951,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) { @@ -7924,23 +7969,34 @@ 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; }); + // 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) { + for (const target of releasedTargets) { updated.delete(target.context.nativeTurnId); } 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 interruptTargets) { + for (const target of releasedTargets) { if (settledTurnContexts.includes(target.context)) continue; updated.delete(target.context.nativeTurnId); } @@ -7967,11 +8023,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); @@ -8042,11 +8094,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); @@ -8078,117 +8126,63 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( } }); - yield* Effect.gen(function* () { - for (const target of interruptTargets) { - const context = target.context; - if (context.capacityRetryPending !== undefined) { - yield* finalizeCodexTurn({ - context, - nativeTurnId: context.nativeTurnId, - status: "interrupted", - completedAt: yield* DateTime.now, - }); - continue; - } - // Null start timestamps ACK queued turns; interruption must wait for turn/started. - if (context.nativeStartReady !== undefined) { - const ready = yield* Deferred.await(context.nativeStartReady).pipe( - Effect.timeoutOption("10 seconds"), - ); - if (Option.isNone(ready)) { - return yield* toProtocolError( - "Codex did not start the queued turn within 10 seconds; Stop could not be delivered.", - ); - } - } - if ((yield* Ref.get(activeTurns)).get(context.nativeTurnId) !== context) { - continue; - } - yield* client.request("turn/interrupt", { - threadId: yield* getNativeThreadId(context.providerThread), - turnId: context.nativeTurnId, - }); - } - const containedTerminalKeys = new Set(); - const attemptedTerminalKeys = new Set(); - const collectTrackedTerminals = Effect.fn("CodexAdapterV2.collectTrackedTerminals")( - function* (targets: typeof interruptTargets) { - const trackedTerminals = new Map< - string, - { readonly nativeThreadId: string; readonly processId: string } - >(); - const runningItems = yield* Ref.get(runningCommandItemsByTurn); - for (const target of targets) { - const nativeThreadId = yield* getNativeThreadId( - target.context.providerThread, - ); - const items = runningItems.get(target.context.nativeTurnId); - for (const item of items?.values() ?? []) { - if (item.processId !== undefined) { - trackedTerminals.set(`${nativeThreadId}:${item.processId}`, { - nativeThreadId, - processId: item.processId, - }); - } + const containedTerminalKeys = new Set(); + const attemptedTerminalKeys = new Set(); + const collectTrackedTerminals = Effect.fn("CodexAdapterV2.collectTrackedTerminals")( + function* (targets: typeof interruptTargets) { + const trackedTerminals = new Map< + string, + { readonly nativeThreadId: string; readonly processId: string } + >(); + const runningItems = yield* Ref.get(runningCommandItemsByTurn); + for (const target of targets) { + const nativeThreadId = yield* getNativeThreadId(target.context.providerThread); + const items = runningItems.get(target.context.nativeTurnId); + for (const item of items?.values() ?? []) { + if (item.processId !== undefined) { + trackedTerminals.set(`${nativeThreadId}:${item.processId}`, { + nativeThreadId, + processId: item.processId, + }); } } - return trackedTerminals; - }, - ); - const terminateTrackedTerminals = Effect.fn( - "CodexAdapterV2.terminateTrackedTerminals", - )(function* (targets: typeof interruptTargets) { - const trackedTerminals = yield* collectTrackedTerminals(targets); - const pendingTerminals = Array.from(trackedTerminals.entries()) - .filter(([key]) => !containedTerminalKeys.has(key)) - .sort( - ([leftKey], [rightKey]) => - Number(attemptedTerminalKeys.has(leftKey)) - - Number(attemptedTerminalKeys.has(rightKey)), - ); - return yield* Effect.forEach( - pendingTerminals, - ([key, { nativeThreadId, processId }]) => - Effect.gen(function* () { - attemptedTerminalKeys.add(key); - yield* terminateBackgroundTerminal(nativeThreadId, processId); - containedTerminalKeys.add(key); - }).pipe( - Effect.as({ success: true as const }), - Effect.catch((error) => Effect.succeed({ success: false as const, error })), - ), + } + return trackedTerminals; + }, + ); + const terminateTrackedTerminals = Effect.fn( + "CodexAdapterV2.terminateTrackedTerminals", + )(function* (targets: typeof interruptTargets) { + const trackedTerminals = yield* collectTrackedTerminals(targets); + const pendingTerminals = Array.from(trackedTerminals.entries()) + .filter(([key]) => !containedTerminalKeys.has(key)) + .sort( + ([leftKey], [rightKey]) => + Number(attemptedTerminalKeys.has(leftKey)) - + Number(attemptedTerminalKeys.has(rightKey)), ); - }); - - const [, completed] = yield* Effect.all( - [ - terminateTrackedTerminals(interruptTargets), - Effect.forEach( - interruptTargets, - (target) => Deferred.await(target.completion), - { concurrency: "unbounded", discard: true }, - ).pipe(Effect.timeoutOption("10 seconds")), - ], - { concurrency: "unbounded" }, + return yield* Effect.forEach( + pendingTerminals, + ([key, { nativeThreadId, processId }]) => + Effect.gen(function* () { + attemptedTerminalKeys.add(key); + yield* terminateBackgroundTerminal(nativeThreadId, processId); + containedTerminalKeys.add(key); + }).pipe( + Effect.as({ success: true as const }), + Effect.catch((error) => Effect.succeed({ success: false as const, error })), + ), ); - if (Option.isNone(completed)) { - for (const target of interruptTargets) { - const context = target.context; - if ((yield* Ref.get(activeTurns)).get(context.nativeTurnId) !== context) { - continue; - } - yield* Effect.logWarning("orchestration-v2.codex-interrupt-timeout", { - providerSessionId: input.providerSessionId, - providerThreadId: context.providerThread.id, - providerTurnId: context.providerTurnId, - nativeTurnId: context.nativeTurnId, - }); - } - } + }); + const 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) { @@ -8226,35 +8220,127 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( }), ); }).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), ); + // 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 }), + ); + + // 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; + 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* stopLateDescendants; + yield* terminateTrackedTerminals(interruptTargets); + return false; + }).pipe( + Effect.onError((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.forkIn(stopLateDescendants, scope).pipe( + Effect.andThen(settleInScope), + Effect.asVoid, + ) + : 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( - Effect.mapError( - (cause) => - new ProviderAdapter.ProviderAdapterInterruptError({ - driver: CODEX_PROVIDER, - providerThreadId: requestedInput.providerThread.id, - providerTurnId: requestedInput.providerTurnId, - cause, - }), + holdGoalStop(requestedInput.providerThread, effect).pipe( + Effect.mapError((cause) => + cause._tag === "ProviderAdapterInterruptTimeoutError" + ? cause + : new ProviderAdapter.ProviderAdapterInterruptError({ + driver: CODEX_PROVIDER, + providerThreadId: requestedInput.providerThread.id, + providerTurnId: requestedInput.providerTurnId, + cause, + }), ), ), ), diff --git a/apps/server/src/orchestration-v2/EffectWorker.ts b/apps/server/src/orchestration-v2/EffectWorker.ts index 9dd507980574..b882e04f6430 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.ts @@ -385,6 +385,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 34ff5a763110..74fc46ba82e4 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"), @@ -348,5 +362,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 0973be037e39..ea98ad7cf20f 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnControlService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnControlService.ts @@ -90,6 +90,7 @@ export interface ProviderTurnControlServiceV2Shape { readonly providerThreadId: ProviderThreadId; readonly providerTurnId: ProviderTurnId; readonly interruptedAttemptId: RunAttemptId; + readonly detachesSession?: boolean; }) => Effect.Effect; } @@ -325,10 +326,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(