From a43c69bf4c2beb50119c7cc2577e02cc86d6f2a3 Mon Sep 17 00:00:00 2001 From: Pixel Perfect Date: Fri, 10 Jul 2026 11:56:31 -0700 Subject: [PATCH 1/5] fix(orchestrator): keep stop and steer responsive --- .../Adapters/CodexAdapterV2.ts | 58 ++++++++++++++++--- apps/web/src/components/ChatView.tsx | 1 + .../chat/ComposerPrimaryActions.test.ts | 4 +- .../chat/ComposerPrimaryActions.tsx | 41 +++++++------ .../src/operations/commands.test.ts | 53 ++++++++++++++++- .../client-runtime/src/operations/commands.ts | 28 ++++++--- .../src/state/threadCommands.test.ts | 22 +++++++ .../src/state/threadCommands.ts | 23 ++++++-- 8 files changed, 189 insertions(+), 41 deletions(-) create mode 100644 packages/client-runtime/src/state/threadCommands.test.ts diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index c77d7b484c3..6ad42461ab4 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -35,6 +35,7 @@ import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; import * as Path from "effect/Path"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; @@ -1283,7 +1284,9 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi const events = yield* Queue.unbounded(); const activeTurns = yield* Ref.make(new Map()); const pendingRootTurns = yield* Ref.make(new Map()); - const turnWaiters = yield* Ref.make(new Map>()); + const turnWaiters = yield* Ref.make( + new Map>>(), + ); const subagentThreads = yield* Ref.make(new Map()); const pendingSubagentTurns = yield* Ref.make( new Map>(), @@ -3514,10 +3517,15 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi }, ); } - const waiter = (yield* Ref.get(turnWaiters)).get(payload.turn.id); - if (waiter !== undefined) { - yield* Deferred.succeed(waiter, undefined); - } + const waiters = yield* Ref.modify(turnWaiters, (current) => { + const matching = current.get(payload.turn.id) ?? new Set(); + const updated = new Map(current); + updated.delete(payload.turn.id); + return [matching, updated] as const; + }); + yield* Effect.forEach(waiters, (waiter) => Deferred.succeed(waiter, undefined), { + discard: true, + }); yield* Ref.update(activeTurns, (current) => { const updated = new Map(current); updated.delete(payload.turn.id); @@ -3694,10 +3702,44 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi `Provider turn ${turnInput.providerTurnId} is not active and cannot be interrupted.`, ); } - yield* client.request("turn/interrupt", { - threadId, - turnId: activeTurn.nativeTurnId, + const terminal = yield* Deferred.make(); + yield* Ref.update(turnWaiters, (current) => { + const updated = new Map(current); + const waiters = new Set(updated.get(activeTurn.nativeTurnId)); + waiters.add(terminal); + updated.set(activeTurn.nativeTurnId, waiters); + return updated; }); + const unregisterTerminalWaiter = Ref.update(turnWaiters, (current) => { + const existing = current.get(activeTurn.nativeTurnId); + if (existing === undefined || !existing.has(terminal)) { + return current; + } + const updated = new Map(current); + const waiters = new Set(existing); + waiters.delete(terminal); + if (waiters.size === 0) { + updated.delete(activeTurn.nativeTurnId); + } else { + updated.set(activeTurn.nativeTurnId, waiters); + } + return updated; + }); + const stopped = yield* client + .request("turn/interrupt", { + threadId, + turnId: activeTurn.nativeTurnId, + }) + .pipe( + Effect.andThen(Deferred.await(terminal)), + Effect.timeoutOption("10 seconds"), + Effect.ensuring(unregisterTerminalWaiter), + ); + if (Option.isNone(stopped)) { + return yield* toProtocolError( + `Provider turn ${turnInput.providerTurnId} did not reach a terminal state before the interrupt timeout.`, + ); + } }).pipe( Effect.mapError( (cause) => diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 8b1d2c672ba..8d7eec1d8fa 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -4393,6 +4393,7 @@ function ChatViewContent(props: ChatViewProps) { environmentId, input: { threadId: activeThread.id, + ...(activeRuntime?.activeRunId ? { runId: activeRuntime.activeRunId } : {}), }, }); if (result._tag === "Failure" && !isAtomCommandInterrupted(result)) { diff --git a/apps/web/src/components/chat/ComposerPrimaryActions.test.ts b/apps/web/src/components/chat/ComposerPrimaryActions.test.ts index 655e83bc062..0ca500125bd 100644 --- a/apps/web/src/components/chat/ComposerPrimaryActions.test.ts +++ b/apps/web/src/components/chat/ComposerPrimaryActions.test.ts @@ -123,7 +123,7 @@ describe("active-turn primary action", () => { expect(markup).not.toContain("steer active turn"); }); - it("replaces stop with send while the active composer has content", () => { + it("keeps stop available alongside send while the active composer has content", () => { const markup = renderToStaticMarkup( createElement(ComposerPrimaryActions, { ...activeTurnProps, @@ -133,6 +133,6 @@ describe("active-turn primary action", () => { ); expect(markup).toContain('aria-label="Send message to steer active turn"'); - expect(markup).not.toContain('aria-label="Stop generation"'); + expect(markup).toContain('aria-label="Stop generation"'); }); }); diff --git a/apps/web/src/components/chat/ComposerPrimaryActions.tsx b/apps/web/src/components/chat/ComposerPrimaryActions.tsx index 2e9aed5aef3..15695f6a17c 100644 --- a/apps/web/src/components/chat/ComposerPrimaryActions.tsx +++ b/apps/web/src/components/chat/ComposerPrimaryActions.tsx @@ -124,21 +124,21 @@ export const ComposerPrimaryActions = memo(function ComposerPrimaryActions({ ); } - if (isRunning && !hasSendableContent) { - return ( - - ); - } + const stopButton = ( + + ); + + if (isRunning && !hasSendableContent) return stopButton; if (showPlanFollowUpPrompt) { if (promptHasText) { @@ -233,9 +233,12 @@ export const ComposerPrimaryActions = memo(function ComposerPrimaryActions({ if (!isRunning) return sendButton; return ( - - - Send now to steer the active turn - +
+ {stopButton} + + + Send now to steer the active turn + +
); }); diff --git a/packages/client-runtime/src/operations/commands.test.ts b/packages/client-runtime/src/operations/commands.test.ts index be5950e427c..0052a104f59 100644 --- a/packages/client-runtime/src/operations/commands.test.ts +++ b/packages/client-runtime/src/operations/commands.test.ts @@ -39,11 +39,13 @@ import { archiveThread, createProject, forkThreadFromRun, + interruptThreadTurn, mergeThreadBack, promoteQueuedRun, reorderQueuedRun, revertThreadCheckpoint, startThreadTurn, + ThreadTurnNotInterruptibleError, updateThreadMetadata, } from "./commands.ts"; @@ -67,6 +69,7 @@ const makeSupervisor = Effect.fn("TestEnvironmentCommands.makeSupervisor")(funct readonly projects: ProjectMutation[]; readonly launches?: OrchestrationV2ThreadLaunchInput[]; readonly projection?: OrchestrationV2ThreadProjection; + readonly projectionReads?: { count: number }; }) { const client = { [ORCHESTRATION_V2_WS_METHODS.dispatchCommand]: (command: OrchestrationV2Command) => @@ -75,7 +78,10 @@ const makeSupervisor = Effect.fn("TestEnvironmentCommands.makeSupervisor")(funct return { sequence: input.commands.length }; }), [ORCHESTRATION_V2_WS_METHODS.getThreadProjection]: () => - Effect.succeed(input.projection ?? v2Projection), + Effect.sync(() => { + if (input.projectionReads) input.projectionReads.count += 1; + return input.projection ?? v2Projection; + }), [ORCHESTRATION_V2_WS_METHODS.launchThread]: (launchInput: OrchestrationV2ThreadLaunchInput) => Effect.sync(() => { input.launches?.push(launchInput); @@ -163,6 +169,51 @@ describe("V2 environment commands", () => { }).pipe(Effect.provide(TEST_CRYPTO_LAYER)), ); + it.effect("interrupts an explicitly targeted run without a projection round trip", () => + Effect.gen(function* () { + const commands: OrchestrationV2Command[] = []; + const projectionReads = { count: 0 }; + const supervisor = yield* makeSupervisor({ commands, projects: [], projectionReads }); + + yield* interruptThreadTurn({ + commandId: CommandId.make("interrupt-explicit-run"), + threadId: v2ThreadId, + runId: RunId.make("run-explicit"), + }).pipe(Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor)); + + expect(projectionReads.count).toBe(0); + expect(commands).toEqual([ + { + type: "run.interrupt", + commandId: "interrupt-explicit-run", + threadId: v2ThreadId, + runId: "run-explicit", + }, + ]); + }).pipe(Effect.provide(TEST_CRYPTO_LAYER)), + ); + + it.effect("fails instead of reporting success when no active run can be interrupted", () => + Effect.gen(function* () { + const commands: OrchestrationV2Command[] = []; + const supervisor = yield* makeSupervisor({ + commands, + projects: [], + projection: { ...v2Projection, runs: [] }, + }); + + const error = yield* Effect.flip( + interruptThreadTurn({ + commandId: CommandId.make("interrupt-missing-run"), + threadId: v2ThreadId, + }).pipe(Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor)), + ); + + expect(error).toBeInstanceOf(ThreadTurnNotInterruptibleError); + expect(commands).toEqual([]); + }).pipe(Effect.provide(TEST_CRYPTO_LAYER)), + ); + it.effect("resolves run ordinal zero to the persisted thread-start checkpoint", () => Effect.gen(function* () { const scopeId = CheckpointScopeId.make("checkpoint-scope-root"); diff --git a/packages/client-runtime/src/operations/commands.ts b/packages/client-runtime/src/operations/commands.ts index fda9980ac55..add0818863a 100644 --- a/packages/client-runtime/src/operations/commands.ts +++ b/packages/client-runtime/src/operations/commands.ts @@ -17,11 +17,12 @@ import { type RunId, type RuntimeMode, type RuntimeRequestId, - type ThreadId, + ThreadId, type UploadChatAttachment, } from "@t3tools/contracts"; import * as Crypto from "effect/Crypto"; import * as Effect from "effect/Effect"; +import * as Schema from "effect/Schema"; import { request } from "../rpc/client.ts"; @@ -129,6 +130,17 @@ export interface InterruptThreadTurnInput extends ThreadCommandInput { readonly turnId?: string; } +export class ThreadTurnNotInterruptibleError extends Schema.TaggedErrorClass()( + "ThreadTurnNotInterruptibleError", + { + threadId: ThreadId, + }, +) { + override get message(): string { + return `Thread ${this.threadId} has no active run to interrupt.`; + } +} + export interface RespondToThreadApprovalInput extends ThreadCommandInput { readonly requestId: RuntimeRequestId; readonly decision: ProviderApprovalDecision; @@ -503,18 +515,20 @@ export const startThreadTurn = Effect.fn("EnvironmentCommands.startThreadTurn")( export const interruptThreadTurn = Effect.fn("EnvironmentCommands.interruptThreadTurn")(function* ( input: InterruptThreadTurnInput, ) { - const projection = yield* getProjection(input.threadId); - const runId = - input.runId ?? - (input.turnId as RunId | undefined) ?? - projection.runs.findLast( + let runId = input.runId ?? (input.turnId as RunId | undefined); + if (runId === undefined) { + const projection = yield* getProjection(input.threadId); + runId = projection.runs.findLast( (run) => run.status === "preparing" || run.status === "starting" || run.status === "running" || run.status === "waiting", )?.id; - if (runId === undefined) return { sequence: 0 }; + } + if (runId === undefined) { + return yield* new ThreadTurnNotInterruptibleError({ threadId: input.threadId }); + } return yield* dispatch({ type: "run.interrupt", commandId: yield* allocateCommandId(input), diff --git a/packages/client-runtime/src/state/threadCommands.test.ts b/packages/client-runtime/src/state/threadCommands.test.ts new file mode 100644 index 00000000000..382478e8bd0 --- /dev/null +++ b/packages/client-runtime/src/state/threadCommands.test.ts @@ -0,0 +1,22 @@ +import { describe, expect, it } from "vite-plus/test"; + +import { threadCommandConcurrencyKey } from "./threadCommands.ts"; + +describe("thread command concurrency", () => { + const input = { + environmentId: "environment-1", + input: { threadId: "thread-1" }, + }; + + it("keeps control commands off the mutation lane", () => { + expect(threadCommandConcurrencyKey("control", input)).not.toBe( + threadCommandConcurrencyKey("mutation", input), + ); + }); + + it("serializes repeated control commands for the same thread", () => { + expect(threadCommandConcurrencyKey("control", input)).toBe( + threadCommandConcurrencyKey("control", input), + ); + }); +}); diff --git a/packages/client-runtime/src/state/threadCommands.ts b/packages/client-runtime/src/state/threadCommands.ts index 93cb6334bc6..c9b16c0365e 100644 --- a/packages/client-runtime/src/state/threadCommands.ts +++ b/packages/client-runtime/src/state/threadCommands.ts @@ -40,6 +40,15 @@ import { } from "../operations/commands.ts"; import type { EnvironmentRegistry } from "../connection/registry.ts"; +export type ThreadCommandLane = "control" | "mutation"; + +export function threadCommandConcurrencyKey( + lane: ThreadCommandLane, + { environmentId, input }: { environmentId: string; input: { threadId: string } }, +): string { + return JSON.stringify([environmentId, input.threadId, lane]); +} + export type { ArchiveThreadInput, CreateThreadInput, @@ -60,6 +69,7 @@ export type { UnarchiveThreadInput, UpdateThreadMetadataInput, } from "../operations/commands.ts"; +export { ThreadTurnNotInterruptibleError } from "../operations/commands.ts"; export function createThreadEnvironmentAtoms( runtime: Atom.AtomRuntime, @@ -67,8 +77,13 @@ export function createThreadEnvironmentAtoms( const scheduler = createAtomCommandScheduler(); const concurrency = { mode: "serial" as const, - key: ({ environmentId, input }: { environmentId: string; input: { threadId: string } }) => - JSON.stringify([environmentId, input.threadId]), + key: (input: { environmentId: string; input: { threadId: string } }) => + threadCommandConcurrencyKey("mutation", input), + }; + const controlConcurrency = { + mode: "serial" as const, + key: (input: { environmentId: string; input: { threadId: string } }) => + threadCommandConcurrencyKey("control", input), }; return { create: createEnvironmentCommand(runtime, { @@ -123,7 +138,7 @@ export function createThreadEnvironmentAtoms( label: "environment-data:commands:thread:interrupt-turn", execute: (input: InterruptThreadTurnInput) => interruptThreadTurn(input), scheduler, - concurrency, + concurrency: controlConcurrency, }), respondToApproval: createEnvironmentCommand(runtime, { label: "environment-data:commands:thread:respond-to-approval", @@ -147,7 +162,7 @@ export function createThreadEnvironmentAtoms( label: "environment-data:commands:thread:stop-session", execute: (input: StopThreadSessionInput) => stopThreadSession(input), scheduler, - concurrency, + concurrency: controlConcurrency, }), forkFromRun: createEnvironmentCommand(runtime, { label: "environment-data:commands:thread:fork-from-run", From e6c6392974a6b65a219b2d8674435649a6d1ccea Mon Sep 17 00:00:00 2001 From: Pixel Perfect Date: Fri, 10 Jul 2026 12:34:34 -0700 Subject: [PATCH 2/5] fix(orchestrator): close interrupt registration races Interrupting a Codex turn now registers its terminal waiter before re-checking the active set, and turn completion retires the active turn before draining waiters, so a completion landing between the interrupt's lookup and registration resolves immediately instead of timing out. On the client, an interrupt that finds no active run while a start for the same thread is still in flight waits for that start to settle and retries once, so a quick Stop after Send targets the run the send creates while staying on the fast control lane. --- .../Adapters/CodexAdapterV2.test.ts | 100 +++++++++++++ .../Adapters/CodexAdapterV2.ts | 133 +++++++++++++----- .../src/state/threadCommands.test.ts | 104 +++++++++++++- .../src/state/threadCommands.ts | 126 +++++++++++++++-- 4 files changed, 411 insertions(+), 52 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index e7e33e85449..888dedec49e 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -15,6 +15,7 @@ import { assert, describe, it } from "@effect/vitest"; import { HostProcessEnvironment, HostProcessPlatform } from "@t3tools/shared/hostProcess"; import { SpawnExecutableResolution } from "@t3tools/shared/shell"; import * as DateTime from "effect/DateTime"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Ref from "effect/Ref"; import { TestClock } from "effect/testing"; @@ -31,7 +32,9 @@ import { makeCodexAppServerProtocolLogger, makeCodexAppServerSpawnCommand, projectCodexDynamicToolItem, + registerCodexTurnTerminalWaiter, resolveCodexRollbackTurnCount, + settleCodexTurnTerminal, } from "./CodexAdapterV2.ts"; describe("CodexAdapterV2 assistant message streaming", () => { @@ -434,6 +437,103 @@ describe("CodexAdapterV2 native protocol logging", () => { }); }); +describe("CodexAdapterV2 interrupt terminal tracking", () => { + const makeTracking = Effect.gen(function* () { + const activeTurns = yield* Ref.make(new Map([["turn-1", "context"]])); + const turnWaiters = yield* Ref.make( + new Map>>(), + ); + return { activeTurns, turnWaiters }; + }); + + it.effect("signals a first-time terminal waiter once the turn settles", () => + Effect.gen(function* () { + const { activeTurns, turnWaiters } = yield* makeTracking; + + const waiter = yield* registerCodexTurnTerminalWaiter({ + activeTurns, + turnWaiters, + nativeTurnId: "turn-1", + }); + + assert.isFalse(waiter.alreadyTerminal); + assert.equal((yield* Ref.get(turnWaiters)).get("turn-1")?.size, 1); + + yield* settleCodexTurnTerminal({ activeTurns, turnWaiters, nativeTurnId: "turn-1" }); + yield* waiter.awaitTerminal; + + assert.isFalse((yield* Ref.get(activeTurns)).has("turn-1")); + assert.isUndefined((yield* Ref.get(turnWaiters)).get("turn-1")); + }), + ); + + it.effect("reports already-terminal when completion lands before registration", () => + Effect.gen(function* () { + const { activeTurns, turnWaiters } = yield* makeTracking; + + // turn/completed can run on another fiber between interruptTurn's + // active-turn lookup and its waiter registration; the registration + // re-check must detect the settled turn instead of waiting forever. + yield* settleCodexTurnTerminal({ activeTurns, turnWaiters, nativeTurnId: "turn-1" }); + const waiter = yield* registerCodexTurnTerminalWaiter({ + activeTurns, + turnWaiters, + nativeTurnId: "turn-1", + }); + + assert.isTrue(waiter.alreadyTerminal); + yield* waiter.unregister; + assert.isUndefined((yield* Ref.get(turnWaiters)).get("turn-1")); + }), + ); + + it.effect("signals every concurrently registered waiter for the same turn", () => + Effect.gen(function* () { + const { activeTurns, turnWaiters } = yield* makeTracking; + + const first = yield* registerCodexTurnTerminalWaiter({ + activeTurns, + turnWaiters, + nativeTurnId: "turn-1", + }); + const second = yield* registerCodexTurnTerminalWaiter({ + activeTurns, + turnWaiters, + nativeTurnId: "turn-1", + }); + + assert.equal((yield* Ref.get(turnWaiters)).get("turn-1")?.size, 2); + + yield* settleCodexTurnTerminal({ activeTurns, turnWaiters, nativeTurnId: "turn-1" }); + yield* first.awaitTerminal; + yield* second.awaitTerminal; + }), + ); + + it.effect("unregistering one waiter keeps the remaining waiters registered", () => + Effect.gen(function* () { + const { activeTurns, turnWaiters } = yield* makeTracking; + + const abandoned = yield* registerCodexTurnTerminalWaiter({ + activeTurns, + turnWaiters, + nativeTurnId: "turn-1", + }); + const kept = yield* registerCodexTurnTerminalWaiter({ + activeTurns, + turnWaiters, + nativeTurnId: "turn-1", + }); + + yield* abandoned.unregister; + assert.equal((yield* Ref.get(turnWaiters)).get("turn-1")?.size, 1); + + yield* settleCodexTurnTerminal({ activeTurns, turnWaiters, nativeTurnId: "turn-1" }); + yield* kept.awaitTerminal; + }), + ); +}); + describe("CodexAdapterV2 rollback mapping", () => { it.effect("derives native rollback count from durable provider turns", () => Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 6ad42461ab4..df4fc7fd031 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -1231,6 +1231,87 @@ export const layer: Layer.Layer< }), ); +export interface CodexTurnTerminalTrackingOptions { + readonly activeTurns: Ref.Ref>; + readonly turnWaiters: Ref.Ref>>>; + readonly nativeTurnId: string; +} + +export interface CodexTurnTerminalWaiter { + /** Resolves once the turn reaches a terminal state. */ + readonly awaitTerminal: Effect.Effect; + readonly unregister: Effect.Effect; + /** + * The turn was already removed from the active set when the waiter was + * registered, so no terminal signal will arrive for it. + */ + readonly alreadyTerminal: boolean; +} + +/** + * Registers a terminal waiter for a native turn, then re-checks the active + * set. Because {@link settleCodexTurnTerminal} removes the active turn before + * draining waiters, a turn still active after registration is guaranteed to + * signal this waiter; a missing turn already reached a terminal state. + */ +export const registerCodexTurnTerminalWaiter = ( + options: CodexTurnTerminalTrackingOptions, +): Effect.Effect => + Effect.gen(function* () { + const terminal = yield* Deferred.make(); + yield* Ref.update(options.turnWaiters, (current) => { + const updated = new Map(current); + const waiters = new Set(updated.get(options.nativeTurnId) ?? []); + waiters.add(terminal); + updated.set(options.nativeTurnId, waiters); + return updated; + }); + const unregister = Ref.update(options.turnWaiters, (current) => { + const existing = current.get(options.nativeTurnId); + if (existing === undefined || !existing.has(terminal)) { + return current; + } + const updated = new Map(current); + const waiters = new Set(existing); + waiters.delete(terminal); + if (waiters.size === 0) { + updated.delete(options.nativeTurnId); + } else { + updated.set(options.nativeTurnId, waiters); + } + return updated; + }); + const alreadyTerminal = !(yield* Ref.get(options.activeTurns)).has(options.nativeTurnId); + return { awaitTerminal: Deferred.await(terminal), unregister, alreadyTerminal }; + }); + +/** + * Marks a native turn terminal: removes it from the active set first, then + * drains and signals its waiters. The ordering pairs with + * {@link registerCodexTurnTerminalWaiter} so a completion racing an interrupt + * can never strand a waiter registered after the drain. + */ +export const settleCodexTurnTerminal = ( + options: CodexTurnTerminalTrackingOptions, +): Effect.Effect => + Effect.gen(function* () { + yield* Ref.update(options.activeTurns, (current) => { + const updated = new Map(current); + updated.delete(options.nativeTurnId); + return updated; + }); + const waiters = yield* Ref.modify(options.turnWaiters, (current) => { + const matching = + current.get(options.nativeTurnId) ?? new Set>(); + const updated = new Map(current); + updated.delete(options.nativeTurnId); + return [matching, updated] as const; + }); + yield* Effect.forEach(waiters, (waiter) => Deferred.succeed(waiter, undefined), { + discard: true, + }); + }); + export interface CodexAdapterV2Options { readonly instanceId: ProviderInstanceId; readonly settings: CodexSettings; @@ -3517,19 +3598,10 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi }, ); } - const waiters = yield* Ref.modify(turnWaiters, (current) => { - const matching = current.get(payload.turn.id) ?? new Set(); - const updated = new Map(current); - updated.delete(payload.turn.id); - return [matching, updated] as const; - }); - yield* Effect.forEach(waiters, (waiter) => Deferred.succeed(waiter, undefined), { - discard: true, - }); - yield* Ref.update(activeTurns, (current) => { - const updated = new Map(current); - updated.delete(payload.turn.id); - return updated; + yield* settleCodexTurnTerminal({ + activeTurns, + turnWaiters, + nativeTurnId: payload.turn.id, }); }), ); @@ -3702,38 +3774,25 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi `Provider turn ${turnInput.providerTurnId} is not active and cannot be interrupted.`, ); } - const terminal = yield* Deferred.make(); - yield* Ref.update(turnWaiters, (current) => { - const updated = new Map(current); - const waiters = new Set(updated.get(activeTurn.nativeTurnId)); - waiters.add(terminal); - updated.set(activeTurn.nativeTurnId, waiters); - return updated; - }); - const unregisterTerminalWaiter = Ref.update(turnWaiters, (current) => { - const existing = current.get(activeTurn.nativeTurnId); - if (existing === undefined || !existing.has(terminal)) { - return current; - } - const updated = new Map(current); - const waiters = new Set(existing); - waiters.delete(terminal); - if (waiters.size === 0) { - updated.delete(activeTurn.nativeTurnId); - } else { - updated.set(activeTurn.nativeTurnId, waiters); - } - return updated; + const waiter = yield* registerCodexTurnTerminalWaiter({ + activeTurns, + turnWaiters, + nativeTurnId: activeTurn.nativeTurnId, }); + if (waiter.alreadyTerminal) { + // The turn settled between the active-turn lookup and waiter + // registration; there is nothing left to interrupt. + return yield* waiter.unregister; + } const stopped = yield* client .request("turn/interrupt", { threadId, turnId: activeTurn.nativeTurnId, }) .pipe( - Effect.andThen(Deferred.await(terminal)), + Effect.andThen(waiter.awaitTerminal), Effect.timeoutOption("10 seconds"), - Effect.ensuring(unregisterTerminalWaiter), + Effect.ensuring(waiter.unregister), ); if (Option.isNone(stopped)) { return yield* toProtocolError( diff --git a/packages/client-runtime/src/state/threadCommands.test.ts b/packages/client-runtime/src/state/threadCommands.test.ts index 382478e8bd0..623981edb47 100644 --- a/packages/client-runtime/src/state/threadCommands.test.ts +++ b/packages/client-runtime/src/state/threadCommands.test.ts @@ -1,6 +1,13 @@ +import { ThreadId } from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; +import { AsyncResult, type AtomRegistry } from "effect/unstable/reactivity"; import { describe, expect, it } from "vite-plus/test"; -import { threadCommandConcurrencyKey } from "./threadCommands.ts"; +import { + coordinateInterruptWithPendingStarts, + threadCommandConcurrencyKey, + ThreadTurnNotInterruptibleError, +} from "./threadCommands.ts"; describe("thread command concurrency", () => { const input = { @@ -20,3 +27,98 @@ describe("thread command concurrency", () => { ); }); }); + +describe("interrupt coordination with pending starts", () => { + const registry = {} as AtomRegistry.AtomRegistry; + const threadId = ThreadId.make("thread-1"); + const target = { environmentId: "environment-1", input: { threadId } }; + const notInterruptible = () => + AsyncResult.failure(Cause.fail(new ThreadTurnNotInterruptibleError({ threadId }))); + + it("retries an interrupt that raced an in-flight start for the same thread", async () => { + let releaseStart = () => {}; + const startGate = new Promise((resolve) => { + releaseStart = resolve; + }); + let startSettled = false; + const interruptAttempts: Array<"no-active-run" | "dispatched"> = []; + const commands = coordinateInterruptWithPendingStarts({ + startTurn: { + label: "start-turn", + run: async () => { + await startGate; + startSettled = true; + return AsyncResult.success("started"); + }, + }, + interruptTurn: { + label: "interrupt-turn", + run: async () => { + if (!startSettled) { + interruptAttempts.push("no-active-run"); + return notInterruptible(); + } + interruptAttempts.push("dispatched"); + return AsyncResult.success("interrupted"); + }, + }, + }); + + const started = commands.startTurn.run(registry, target); + const interrupted = commands.interruptTurn.run(registry, target); + releaseStart(); + await started; + const result = await interrupted; + + expect(interruptAttempts).toEqual(["no-active-run", "dispatched"]); + expect(result._tag).toBe("Success"); + }); + + it("surfaces the failure without retrying when no start is in flight", async () => { + let interruptAttempts = 0; + const commands = coordinateInterruptWithPendingStarts({ + startTurn: { + label: "start-turn", + run: async () => AsyncResult.success("started"), + }, + interruptTurn: { + label: "interrupt-turn", + run: async () => { + interruptAttempts += 1; + return notInterruptible(); + }, + }, + }); + + const result = await commands.interruptTurn.run(registry, target); + + expect(interruptAttempts).toBe(1); + expect(result._tag).toBe("Failure"); + }); + + it("ignores starts for other threads when deciding whether to retry", async () => { + let interruptAttempts = 0; + const commands = coordinateInterruptWithPendingStarts({ + startTurn: { + label: "start-turn", + run: () => new Promise(() => {}), + }, + interruptTurn: { + label: "interrupt-turn", + run: async () => { + interruptAttempts += 1; + return notInterruptible(); + }, + }, + }); + + void commands.startTurn.run(registry, { + environmentId: "environment-1", + input: { threadId: ThreadId.make("thread-other") }, + }); + const result = await commands.interruptTurn.run(registry, target); + + expect(interruptAttempts).toBe(1); + expect(result._tag).toBe("Failure"); + }); +}); diff --git a/packages/client-runtime/src/state/threadCommands.ts b/packages/client-runtime/src/state/threadCommands.ts index c9b16c0365e..49044b56316 100644 --- a/packages/client-runtime/src/state/threadCommands.ts +++ b/packages/client-runtime/src/state/threadCommands.ts @@ -1,7 +1,14 @@ +import * as Cause from "effect/Cause"; import * as Crypto from "effect/Crypto"; -import { Atom } from "effect/unstable/reactivity"; +import * as Schema from "effect/Schema"; +import { Atom, type AtomRegistry } from "effect/unstable/reactivity"; -import { createAtomCommandScheduler, createEnvironmentCommand } from "./runtime.ts"; +import { + type AtomCommand, + type AtomCommandResult, + createAtomCommandScheduler, + createEnvironmentCommand, +} from "./runtime.ts"; import { type ArchiveThreadInput, type CreateThreadInput, @@ -35,6 +42,7 @@ import { setThreadRuntimeMode, startThreadTurn, stopThreadSession, + ThreadTurnNotInterruptibleError, unarchiveThread, updateThreadMetadata, } from "../operations/commands.ts"; @@ -71,6 +79,92 @@ export type { } from "../operations/commands.ts"; export { ThreadTurnNotInterruptibleError } from "../operations/commands.ts"; +interface ThreadCommandTarget { + readonly environmentId: string; + readonly input: { readonly threadId: string }; +} + +const isThreadTurnNotInterruptibleError = Schema.is(ThreadTurnNotInterruptibleError); + +function isThreadTurnNotInterruptibleFailure(result: AtomCommandResult): boolean { + return result._tag === "Failure" && isThreadTurnNotInterruptibleError(Cause.squash(result.cause)); +} + +/** + * Keeps `interruptTurn` on its own fast control lane while closing the + * Send-then-quick-Stop race: when an interrupt finds no active run but a + * `startTurn` for the same thread was still in flight when the interrupt was + * issued, the interrupt waits for that specific start to settle and retries + * once, so it targets the run the pending send creates instead of failing + * against a stale projection. + */ +export function coordinateInterruptWithPendingStarts< + StartTarget extends ThreadCommandTarget, + InterruptTarget extends ThreadCommandTarget, + SA, + SE, + IA, + IE, +>(commands: { + readonly startTurn: AtomCommand; + readonly interruptTurn: AtomCommand; +}): { + readonly startTurn: AtomCommand; + readonly interruptTurn: AtomCommand; +} { + const pendingStarts = new WeakMap< + AtomRegistry.AtomRegistry, + Map>> + >(); + const pendingKey = (target: ThreadCommandTarget) => + JSON.stringify([target.environmentId, target.input.threadId]); + + return { + startTurn: { + label: commands.startTurn.label, + run: (registry, input) => { + let byThread = pendingStarts.get(registry); + if (byThread === undefined) { + byThread = new Map(); + pendingStarts.set(registry, byThread); + } + const key = pendingKey(input); + let inFlight = byThread.get(key); + if (inFlight === undefined) { + inFlight = new Set(); + byThread.set(key, inFlight); + } + const active = inFlight; + const threads = byThread; + const result = commands.startTurn.run(registry, input); + active.add(result); + const settle = () => { + active.delete(result); + if (active.size === 0 && threads.get(key) === active) { + threads.delete(key); + } + }; + void result.then(settle, settle); + return result; + }, + }, + interruptTurn: { + label: commands.interruptTurn.label, + run: async (registry, input) => { + const startsAtDispatch = Array.from( + pendingStarts.get(registry)?.get(pendingKey(input)) ?? [], + ); + const result = await commands.interruptTurn.run(registry, input); + if (startsAtDispatch.length === 0 || !isThreadTurnNotInterruptibleFailure(result)) { + return result; + } + await Promise.allSettled(startsAtDispatch); + return commands.interruptTurn.run(registry, input); + }, + }, + }; +} + export function createThreadEnvironmentAtoms( runtime: Atom.AtomRuntime, ) { @@ -85,6 +179,20 @@ export function createThreadEnvironmentAtoms( key: (input: { environmentId: string; input: { threadId: string } }) => threadCommandConcurrencyKey("control", input), }; + const turnLifecycle = coordinateInterruptWithPendingStarts({ + startTurn: createEnvironmentCommand(runtime, { + label: "environment-data:commands:thread:start-turn", + execute: (input: StartThreadTurnInput) => startThreadTurn(input), + scheduler, + concurrency, + }), + interruptTurn: createEnvironmentCommand(runtime, { + label: "environment-data:commands:thread:interrupt-turn", + execute: (input: InterruptThreadTurnInput) => interruptThreadTurn(input), + scheduler, + concurrency: controlConcurrency, + }), + }); return { create: createEnvironmentCommand(runtime, { label: "environment-data:commands:thread:create", @@ -128,18 +236,8 @@ export function createThreadEnvironmentAtoms( scheduler, concurrency, }), - startTurn: createEnvironmentCommand(runtime, { - label: "environment-data:commands:thread:start-turn", - execute: (input: StartThreadTurnInput) => startThreadTurn(input), - scheduler, - concurrency, - }), - interruptTurn: createEnvironmentCommand(runtime, { - label: "environment-data:commands:thread:interrupt-turn", - execute: (input: InterruptThreadTurnInput) => interruptThreadTurn(input), - scheduler, - concurrency: controlConcurrency, - }), + startTurn: turnLifecycle.startTurn, + interruptTurn: turnLifecycle.interruptTurn, respondToApproval: createEnvironmentCommand(runtime, { label: "environment-data:commands:thread:respond-to-approval", execute: (input: RespondToThreadApprovalInput) => respondToThreadApproval(input), From 2d5e5edc2589303031746a312a49cf7c26d08b4c Mon Sep 17 00:00:00 2001 From: Pixel Perfect Date: Fri, 10 Jul 2026 12:39:50 -0700 Subject: [PATCH 3/5] chore: retrigger checks From 368b9232124ac7be60299fafe5bf1787c1180cea Mon Sep 17 00:00:00 2001 From: Pixel Perfect Date: Fri, 10 Jul 2026 12:45:21 -0700 Subject: [PATCH 4/5] fix(client): retry interrupt after the in-flight start only The mutation lane is serial per thread, so only the earliest pending start is executing. Waiting for every captured start made Stop drain the whole queued send backlog before retrying; retry now fires as soon as the first start settles. --- .../src/state/threadCommands.test.ts | 47 +++++++++++++++++++ .../src/state/threadCommands.ts | 12 ++++- 2 files changed, 58 insertions(+), 1 deletion(-) diff --git a/packages/client-runtime/src/state/threadCommands.test.ts b/packages/client-runtime/src/state/threadCommands.test.ts index 623981edb47..32a7cb0d5a7 100644 --- a/packages/client-runtime/src/state/threadCommands.test.ts +++ b/packages/client-runtime/src/state/threadCommands.test.ts @@ -74,6 +74,53 @@ describe("interrupt coordination with pending starts", () => { expect(result._tag).toBe("Success"); }); + it("retries as soon as the in-flight start settles without draining queued sends", async () => { + let releaseFirstStart = () => {}; + const firstStartGate = new Promise((resolve) => { + releaseFirstStart = resolve; + }); + let startsBegun = 0; + let firstStartSettled = false; + const interruptAttempts: Array<"no-active-run" | "dispatched"> = []; + const commands = coordinateInterruptWithPendingStarts({ + startTurn: { + label: "start-turn", + run: async () => { + startsBegun += 1; + if (startsBegun === 1) { + await firstStartGate; + firstStartSettled = true; + return AsyncResult.success("started"); + } + // A queued send that never settles: the interrupt retry must not + // wait for it. + return new Promise(() => {}); + }, + }, + interruptTurn: { + label: "interrupt-turn", + run: async () => { + if (!firstStartSettled) { + interruptAttempts.push("no-active-run"); + return notInterruptible(); + } + interruptAttempts.push("dispatched"); + return AsyncResult.success("interrupted"); + }, + }, + }); + + const firstStart = commands.startTurn.run(registry, target); + void commands.startTurn.run(registry, target); + const interrupted = commands.interruptTurn.run(registry, target); + releaseFirstStart(); + await firstStart; + const result = await interrupted; + + expect(interruptAttempts).toEqual(["no-active-run", "dispatched"]); + expect(result._tag).toBe("Success"); + }); + it("surfaces the failure without retrying when no start is in flight", async () => { let interruptAttempts = 0; const commands = coordinateInterruptWithPendingStarts({ diff --git a/packages/client-runtime/src/state/threadCommands.ts b/packages/client-runtime/src/state/threadCommands.ts index 49044b56316..02ebf1ca850 100644 --- a/packages/client-runtime/src/state/threadCommands.ts +++ b/packages/client-runtime/src/state/threadCommands.ts @@ -158,7 +158,17 @@ export function coordinateInterruptWithPendingStarts< if (startsAtDispatch.length === 0 || !isThreadTurnNotInterruptibleFailure(result)) { return result; } - await Promise.allSettled(startsAtDispatch); + // The mutation lane is serial per thread, so only the earliest + // pending start is in flight; retry as soon as any start settles + // instead of waiting behind the whole queued send backlog. + await Promise.race( + startsAtDispatch.map((start) => + start.then( + () => undefined, + () => undefined, + ), + ), + ); return commands.interruptTurn.run(registry, input); }, }, From ccd1303f57dcfcd18ba1f947e9441b7880d5d5d2 Mon Sep 17 00:00:00 2001 From: Pixel Perfect Date: Fri, 10 Jul 2026 12:58:45 -0700 Subject: [PATCH 5/5] fix(client): ignore starts settled before an interrupt The dispatch snapshot could capture a start whose settlement cleanup was still queued, making a Stop with nothing in flight retry against a later, unrelated start. Snapshot after one microtask yield so earlier settlements drain first and only genuinely in-flight starts count. --- .../src/state/threadCommands.test.ts | 27 +++++++++++++++++++ .../src/state/threadCommands.ts | 7 +++++ 2 files changed, 34 insertions(+) diff --git a/packages/client-runtime/src/state/threadCommands.test.ts b/packages/client-runtime/src/state/threadCommands.test.ts index 32a7cb0d5a7..9db7e503134 100644 --- a/packages/client-runtime/src/state/threadCommands.test.ts +++ b/packages/client-runtime/src/state/threadCommands.test.ts @@ -121,6 +121,33 @@ describe("interrupt coordination with pending starts", () => { expect(result._tag).toBe("Success"); }); + it("does not retry when the only pending start had already settled at dispatch", async () => { + let interruptAttempts = 0; + const commands = coordinateInterruptWithPendingStarts({ + startTurn: { + label: "start-turn", + run: async () => AsyncResult.success("started"), + }, + interruptTurn: { + label: "interrupt-turn", + run: async () => { + interruptAttempts += 1; + return notInterruptible(); + }, + }, + }); + + // Dispatch a start that settles immediately, then dispatch the interrupt + // in the same tick — before the start's cleanup microtask has removed it + // from the pending set. + const started = commands.startTurn.run(registry, target); + const result = await commands.interruptTurn.run(registry, target); + await started; + + expect(interruptAttempts).toBe(1); + expect(result._tag).toBe("Failure"); + }); + it("surfaces the failure without retrying when no start is in flight", async () => { let interruptAttempts = 0; const commands = coordinateInterruptWithPendingStarts({ diff --git a/packages/client-runtime/src/state/threadCommands.ts b/packages/client-runtime/src/state/threadCommands.ts index 02ebf1ca850..dccd2311e7e 100644 --- a/packages/client-runtime/src/state/threadCommands.ts +++ b/packages/client-runtime/src/state/threadCommands.ts @@ -151,6 +151,13 @@ export function coordinateInterruptWithPendingStarts< interruptTurn: { label: commands.interruptTurn.label, run: async (registry, input) => { + // A start that settled before this interrupt was issued may still + // have its cleanup queued as a microtask. Yield once so those + // cleanups drain (they were queued earlier, so they run first) and + // the snapshot only contains starts genuinely in flight — a stale + // entry must not trigger a retry that could interrupt a later, + // unrelated start. + await Promise.resolve(); const startsAtDispatch = Array.from( pendingStarts.get(registry)?.get(pendingKey(input)) ?? [], );