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 c77d7b484c3..df4fc7fd031 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"; @@ -1230,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; @@ -1283,7 +1365,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,14 +3598,10 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi }, ); } - const waiter = (yield* Ref.get(turnWaiters)).get(payload.turn.id); - if (waiter !== undefined) { - yield* Deferred.succeed(waiter, undefined); - } - 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, }); }), ); @@ -3694,10 +3774,31 @@ 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 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(waiter.awaitTerminal), + Effect.timeoutOption("10 seconds"), + Effect.ensuring(waiter.unregister), + ); + 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..9db7e503134 --- /dev/null +++ b/packages/client-runtime/src/state/threadCommands.test.ts @@ -0,0 +1,198 @@ +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 { + coordinateInterruptWithPendingStarts, + threadCommandConcurrencyKey, + ThreadTurnNotInterruptibleError, +} 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), + ); + }); +}); + +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("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("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({ + 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 93cb6334bc6..dccd2311e7e 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,11 +42,21 @@ import { setThreadRuntimeMode, startThreadTurn, stopThreadSession, + ThreadTurnNotInterruptibleError, unarchiveThread, updateThreadMetadata, } 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 +77,110 @@ export type { UnarchiveThreadInput, UpdateThreadMetadataInput, } 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) => { + // 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)) ?? [], + ); + const result = await commands.interruptTurn.run(registry, input); + if (startsAtDispatch.length === 0 || !isThreadTurnNotInterruptibleFailure(result)) { + return result; + } + // 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); + }, + }, + }; +} export function createThreadEnvironmentAtoms( runtime: Atom.AtomRuntime, @@ -67,9 +188,28 @@ 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), + }; + 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", @@ -113,18 +253,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, - }), + startTurn: turnLifecycle.startTurn, + interruptTurn: turnLifecycle.interruptTurn, respondToApproval: createEnvironmentCommand(runtime, { label: "environment-data:commands:thread:respond-to-approval", execute: (input: RespondToThreadApprovalInput) => respondToThreadApproval(input), @@ -147,7 +277,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",