diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 92f0e4205b639..e6fcbd06da210 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -14168,6 +14168,423 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); + // Claude opens a wake turn and asks for a tool before T3 has started its continuation run. + const EARLY_WAKE_TOOL_USE_ID = "toolu_01EarlyWakeApprovalBash"; + const earlyWakeApprovalPolicy = ProviderAdapter.ProviderAdapterV2RuntimePolicy.make({ + runtimeMode: "approval-required", + interactionMode: "default", + cwd: "/workspace", + }); + const askBeforeWakeContinuation = Effect.gen(function* () { + const harness = yield* makeWakeHarness; + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("attempt-claude-early-wake-approval-1"), + text: "Run the build in the background.", + attachments: [], + runtimePolicy: earlyWakeApprovalPolicy, + }), + ); + yield* Queue.offer(harness.sdkMessages, wakeTaskStarted); + yield* Queue.offer(harness.sdkMessages, turnOneResult); + yield* harness.awaitTerminal(1); + + yield* harness.offerAndWait(wakeNotification); + yield* harness.offerAndWait(wakeTurnInit); + assert.lengthOf(harness.continuationRequests, 1); + yield* harness.offerAndWait( + claudeSdkFrame({ + type: "assistant", + message: { + model: "claude-sonnet-4-6", + id: "msg_early_wake_bash", + type: "message", + role: "assistant", + content: [ + { + type: "tool_use", + id: EARLY_WAKE_TOOL_USE_ID, + name: "Bash", + input: { command: "ls" }, + }, + ], + }, + parent_tool_use_id: null, + uuid: "00000000-0000-4000-8000-000000000795", + session_id: WAKE_NATIVE_SESSION, + }), + ); + // The callback listens for Claude's abort only once it waits for the continuation run. + // @effect-diagnostics-next-line abortControllerInEffect:off - Claude, not the test fiber, owns the request. + const withdrawal = new AbortController(); + const listening = Promise.withResolvers(); + const addEventListener = withdrawal.signal.addEventListener.bind(withdrawal.signal); + withdrawal.signal.addEventListener = (...args: Parameters) => { + addEventListener(...args); + listening.resolve(); + }; + const decision = yield* Effect.promise(() => + harness.getOpenedOptions()!.canUseTool!( + "Bash", + { command: "ls" }, + { + signal: withdrawal.signal, + toolUseID: EARLY_WAKE_TOOL_USE_ID, + requestId: "request-early-wake-bash", + }, + ), + ).pipe(Effect.forkScoped); + const waiting = Effect.promise(() => listening.promise); + return { harness, decision, waiting, withdrawal }; + }); + + it.effect("asks for an approval a wake turn raises before its continuation run starts", () => + Effect.gen(function* () { + const { harness, decision, waiting } = yield* askBeforeWakeContinuation; + const wakeAttempt = RunAttemptId.make("attempt-claude-early-wake-approval-2"); + const early = yield* Effect.raceFirst( + Effect.as(waiting, { type: "waiting" } as const), + Fiber.join(decision).pipe(Effect.map((result) => ({ type: "decision", result }) as const)), + ); + assert.equal( + early.type, + "waiting", + `Expected the request to wait for the continuation run, got ${JSON.stringify(early.type === "decision" ? early.result : null)}`, + ); + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: wakeAttempt, + text: "Background task completed.", + attachments: [], + providerTurnOrdinal: 2, + messageCreatedBy: "agent", + messageCreationSource: "provider", + runtimePolicy: earlyWakeApprovalPolicy, + }), + ); + const first = yield* Effect.raceFirst( + Queue.take(harness.runtimeRequestReceipts).pipe( + Effect.map((event) => ({ type: "request", event }) as const), + ), + Fiber.join(decision).pipe(Effect.map((result) => ({ type: "decision", result }) as const)), + ); + assert.equal( + first.type, + "request", + `Expected an approval request, got ${JSON.stringify(first.type === "decision" ? first.result : null)}`, + ); + if (first.type !== "request") return; + const bashRuns = harness.events.flatMap((event) => + event.type === "turn_item.updated" && + event.turnItem.nativeItemRef?.nativeId === EARLY_WAKE_TOOL_USE_ID + ? [event.turnItem.runId] + : [], + ); + assert.isNotEmpty(bashRuns); + assert.isTrue(bashRuns.every((runId) => runId === RunId.make(`run-${wakeAttempt}`))); + yield* harness.runtime.respondToRuntimeRequest({ + requestId: first.event.runtimeRequest.id, + decision: "accept", + }); + assert.equal((yield* Fiber.join(decision))?.behavior, "allow"); + + yield* harness.offerAndWait(wakeResult); + yield* harness.awaitTerminal(2); + assert.equal(harness.terminalEvents()[1]?.status, "completed"); + }).pipe( + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ); + + it.effect( + "denies a wake turn's tool request when its continuation run is cancelled unstarted", + () => + Effect.gen(function* () { + const { harness, decision, waiting } = yield* askBeforeWakeContinuation; + yield* waiting; + yield* harness.continuationRequests[0]!.onRunEndedWithoutStarting!(); + const result = yield* Fiber.join(decision); + assert.equal(result?.behavior, "deny"); + assert.isFalse(harness.events.some((event) => event.type === "runtime_request.updated")); + }).pipe( + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ); + + it.effect( + "denies a wake turn's tool request when Claude withdraws it before its continuation run starts", + () => + Effect.gen(function* () { + const { harness, decision, waiting, withdrawal } = yield* askBeforeWakeContinuation; + yield* waiting; + withdrawal.abort(); + const result = yield* Fiber.join(decision); + assert.equal(result?.behavior, "deny"); + assert.isFalse(harness.events.some((event) => event.type === "runtime_request.updated")); + }).pipe( + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ); + + it.effect( + "denies a wake turn's tool request when a user turn attaches before its continuation run", + () => + Effect.gen(function* () { + const { harness, decision, waiting } = yield* askBeforeWakeContinuation; + yield* waiting; + // The wake's buffered tool_use replays in its continuation, so a user turn must not own it. + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("attempt-claude-early-wake-user-turn"), + text: "Status?", + attachments: [], + providerTurnOrdinal: 2, + runtimePolicy: earlyWakeApprovalPolicy, + }), + ); + const result = yield* Fiber.join(decision); + assert.equal(result?.behavior, "deny"); + assert.isFalse(harness.events.some((event) => event.type === "runtime_request.updated")); + }).pipe( + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ); + + it.effect( + "keeps a later wake's tool request out of a Monitor continuation that leaves it buffered", + () => + Effect.gen(function* () { + const harness = yield* makeWakeHarness; + const now = yield* DateTime.now; + const taskId = "bmonitorsplit"; + const monitorToolUseId = "toolu_monitor_split"; + const bashToolUseId = "toolu_01SplitWakeBash"; + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-monitor-split-1"), + text: "Watch the run.", + attachments: [], + runtimePolicy: earlyWakeApprovalPolicy, + }), + ); + for (const frame of [ + claudeSdkFrame({ + type: "assistant", + message: { + model: "claude-sonnet-4-6", + id: "msg_monitor_split", + type: "message", + role: "assistant", + content: [ + { + type: "tool_use", + id: monitorToolUseId, + name: "Monitor", + input: { description: "line monitor", command: "./watch.sh" }, + }, + ], + stop_reason: null, + stop_sequence: null, + usage: { input_tokens: 1, output_tokens: 1 }, + }, + parent_tool_use_id: null, + uuid: "00000000-0000-4000-8000-000000000a01", + session_id: WAKE_NATIVE_SESSION, + }), + claudeSdkFrame({ + type: "system", + subtype: "task_started", + task_id: taskId, + tool_use_id: monitorToolUseId, + description: "line monitor", + is_backgrounded: true, + task_type: "local_bash", + uuid: "00000000-0000-4000-8000-000000000a02", + session_id: WAKE_NATIVE_SESSION, + }), + claudeSdkFrame({ + type: "user", + message: { + role: "user", + content: [ + { + type: "tool_result", + tool_use_id: monitorToolUseId, + content: `Monitor started (task ${taskId}). You will be notified on each event.`, + }, + ], + }, + parent_tool_use_id: null, + uuid: "00000000-0000-4000-8000-000000000a03", + session_id: WAKE_NATIVE_SESSION, + tool_use_result: { taskId, persistent: false }, + }), + makeResultFrame({ uuid: "00000000-0000-4000-8000-000000000a04", result: "STARTED" }), + ]) { + yield* Queue.offer(harness.sdkMessages, frame); + } + yield* harness.awaitTerminal(1); + + // A Monitor event wakes Claude and ends with its own result; its continuation is queued. + yield* harness.offerAndWait(wakeTurnInit); + yield* harness.offerAndWait( + makeAssistantTextFrame({ uuid: "00000000-0000-4000-8000-000000000a05", text: "Noted." }), + ); + yield* harness.awaitContinuation(1); + yield* harness.offerAndWait( + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000a06", + result: "Noted.", + origin: { kind: "task-notification" }, + }), + ); + // A later wake asks for a tool while still buffered behind the Monitor continuation. + yield* harness.offerAndWait( + claudeSdkFrame({ + type: "system", + subtype: "init", + uuid: "00000000-0000-4000-8000-000000000a07", + session_id: WAKE_NATIVE_SESSION, + }), + ); + yield* harness.offerAndWait( + claudeSdkFrame({ + type: "assistant", + message: { + model: "claude-sonnet-4-6", + id: "msg_split_wake_bash", + type: "message", + role: "assistant", + content: [ + { type: "tool_use", id: bashToolUseId, name: "Bash", input: { command: "ls" } }, + ], + }, + parent_tool_use_id: null, + uuid: "00000000-0000-4000-8000-000000000a08", + session_id: WAKE_NATIVE_SESSION, + }), + ); + // @effect-diagnostics-next-line abortControllerInEffect:off - Claude, not the test fiber, owns the request. + const withdrawal = new AbortController(); + const listening = Promise.withResolvers(); + const addEventListener = withdrawal.signal.addEventListener.bind(withdrawal.signal); + withdrawal.signal.addEventListener = ( + ...args: Parameters + ) => { + addEventListener(...args); + listening.resolve(); + }; + const decision = yield* Effect.promise(() => + harness.getOpenedOptions()!.canUseTool!( + "Bash", + { command: "ls" }, + { + signal: withdrawal.signal, + toolUseID: bashToolUseId, + requestId: "request-split-bash", + }, + ), + ).pipe(Effect.forkScoped); + yield* Effect.promise(() => listening.promise); + + // The Monitor continuation drains only the Monitor wake; the later wake keeps its tool_use. + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-monitor-split-2"), + text: "Background task completed.", + attachments: [], + providerTurnOrdinal: 2, + messageCreatedBy: "agent", + messageCreationSource: "provider", + runtimePolicy: earlyWakeApprovalPolicy, + }), + ); + assert.equal((yield* Fiber.join(decision))?.behavior, "deny"); + assert.isFalse(harness.events.some((event) => event.type === "runtime_request.updated")); + const bashRuns = () => + harness.events.flatMap((event) => + event.type === "turn_item.updated" && + event.turnItem.nativeItemRef?.nativeId === bashToolUseId + ? [event.turnItem.runId] + : [], + ); + // The Monitor continuation settles without the later wake's tool, which gets its own run. + yield* harness.awaitTerminal(2); + assert.isEmpty(bashRuns()); + yield* harness.awaitContinuation(2); + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-monitor-split-3"), + text: "Background task completed.", + attachments: [], + providerTurnOrdinal: 3, + messageCreatedBy: "agent", + messageCreationSource: "provider", + runtimePolicy: earlyWakeApprovalPolicy, + }), + ); + yield* harness.until(() => bashRuns().length > 0); + assert.isTrue( + bashRuns().every((runId) => runId === RunId.make("run-attempt-claude-monitor-split-3")), + ); + }).pipe( + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ); + it.effect.each([ { card: "an approval", toolName: "Bash", toolInput: { command: "echo probe" } }, { diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 4ea7b6f8008e4..09097785463db 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -4248,6 +4248,14 @@ function toSessionPermissionUpdates( ]; } +const awaitAbortSignal = (signal: AbortSignal) => + Effect.callback((resume) => { + const abort = () => resume(Effect.void); + signal.addEventListener("abort", abort, { once: true }); + if (signal.aborted) abort(); + return Effect.sync(() => signal.removeEventListener("abort", abort)); + }); + export const awaitClaudeApprovalDecision = Effect.fn("awaitClaudeApprovalDecision")(function* ( decision: Deferred.Deferred, signal: AbortSignal, @@ -6059,6 +6067,14 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( >(), ); const absorbedWakes = yield* Ref.make(new Set()); + // Completed and replaced when a turn attaches or a wake latch settles; tool requests from a wake wait on it. + const wakeTurnChange = yield* Ref.make(yield* Deferred.make()); + const signalWakeTurnChange = Effect.gen(function* () { + const previous = yield* Ref.getAndSet(wakeTurnChange, yield* Deferred.make()); + yield* Deferred.succeed(previous, undefined); + }); + // The wake a continuation turn drained; its latch is gone once the turn attaches. + const wakeTokenByTurn = new WeakMap(); const heldProposedPlansByToolUseId = new Map(); // /clear moves a live process to a new native session; persist that one so a reopen resumes it. const clearedNativeSessions = new Map(); @@ -6582,6 +6598,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( updated.delete(nativeThreadId); return updated; }); + yield* signalWakeTurnChange; yield* forgetAbsorbedWake(nativeThreadId); }); @@ -6591,7 +6608,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( const updated = new Map(current); updated.delete(nativeThreadId); return updated; - }); + }).pipe(Effect.andThen(signalWakeTurnChange)); // Claude runs one root turn at a time, so a user turn's result also ends a wake turn it // folded in. The continuation replays that wake's buffered frames but gets no result of its own. @@ -9781,7 +9798,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( return latch === undefined || latch.retired === retired ? current : new Map(current).set(nativeThreadId, { ...latch, retired }); - }); + }).pipe(Effect.andThen(signalWakeTurnChange)); // Compare-and-set: a process reset clears the buffer without the SDK message lock, and must not be undone. const replaceWakeBuffer = Effect.fnUntraced(function* ( @@ -11737,13 +11754,77 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( yield* handleRoutedSdkMessage(input); }); + // Claude can ask for a tool in a turn it started itself before T3 starts the continuation run + // offered for it. Wait for that run; a request whose wake has no run coming is still denied. + const isToolUseBuffered = (nativeThreadId: string, toolUseId: string) => + Effect.map(Ref.get(wakeBuffers), (buffers) => + (buffers.get(nativeThreadId)?.messages ?? []).some( + (message) => + message.type === "assistant" && + message.message.content.some( + (block) => block.type === "tool_use" && block.id === toolUseId, + ), + ), + ); + + const awaitWakeContinuationTurn = Effect.fnUntraced(function* ( + signal: AbortSignal, + toolUseId: string, + ) { + const liveQuery = yield* Ref.get(queryContext); + if (liveQuery === null || liveQuery.stopping) return null; + const nativeThreadId = liveQuery.nativeThreadId; + const token = (yield* Ref.get(requestedContinuations)).get(nativeThreadId)?.token; + const ended = Effect.raceFirst( + awaitAbortSignal(signal), + Deferred.await(liveQuery.closed), + ); + const isOwnQuery = Effect.map( + Ref.get(queryContext), + (current) => current === liveQuery && !liveQuery.stopping && !signal.aborted, + ); + const isPendingWake = Effect.map(Ref.get(requestedContinuations), (latches) => { + const latch = latches.get(nativeThreadId); + return latch !== undefined && latch.token === token && latch.retired === null; + }); + while (yield* isOwnQuery) { + const changed = yield* Ref.get(wakeTurnChange); + if ((yield* Ref.get(activeTurn)) !== null) { + // A continuation replays the wake's buffered frames under this lock; answer after them. + const context = yield* Effect.raceFirst( + sdkMessageLock.withPermits(1)(Ref.get(activeTurn)), + Effect.as(ended, null), + ); + // Only the continuation that drained this wake's tool_use owns the tool. A user turn that + // attaches first, or a split drain that leaves a later wake buffered, would replay that + // tool_use under a second run. + return context !== null && + context.nativeThreadId === nativeThreadId && + wakeTokenByTurn.get(context) === token && + !(yield* isToolUseBuffered(nativeThreadId, toolUseId)) && + (yield* isOwnQuery) + ? context + : null; + } + if (!(yield* isPendingWake)) return null; + const woke = yield* Effect.raceFirst( + Effect.as(Deferred.await(changed), true), + Effect.as(ended, false), + ); + if (!woke) return null; + } + return null; + }); + const canUseToolEffect = Effect.fn("ClaudeAdapterV2.canUseTool")(function* ( toolName: Parameters[0], toolInput: Parameters[1], callbackOptions: Parameters[2], options?: { readonly startToolCall?: boolean }, ) { - const context = yield* Ref.get(activeTurn); + const context = + (yield* Ref.get(activeTurn)) ?? + (yield* awaitWakeContinuationTurn(callbackOptions.signal, callbackOptions.toolUseID)); if (context === null) { return { behavior: "deny", @@ -12832,6 +12913,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( detail: CLAUDE_QUERY_ENDED_BEFORE_TURN_DETAIL, }); } + yield* signalWakeTurnChange; if ( !isClaudeProviderContinuationTurn(turnInput) && permissionModeForClaudeRuntimePolicy(turnInput.runtimePolicy) === @@ -12922,11 +13004,12 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( ownEndIndex !== -1 && (yield* holdsBufferedWake(nativeThreadId, buffered.slice(ownEndIndex + 1))); const drained = leavesLaterWake ? buffered.slice(0, ownEndIndex + 1) : buffered; - yield* Ref.update(requestedContinuations, (current) => { + const consumedWake = yield* Ref.modify(requestedContinuations, (current) => { const updated = new Map(current); updated.delete(nativeThreadId); - return updated; + return [current.get(nativeThreadId), updated] as const; }); + if (consumedWake !== undefined) wakeTokenByTurn.set(context, consumedWake.token); // A root frame clears the absorbed marker, so behind a split it is the later wake's // and stays; the split wake was absorbed when its end is not a result. const wakeAbsorbed = leavesLaterWake diff --git a/apps/server/src/orchestration-v2/ProviderContinuationService.test.ts b/apps/server/src/orchestration-v2/ProviderContinuationService.test.ts index 1827700ca16c5..903724fa6e7ed 100644 --- a/apps/server/src/orchestration-v2/ProviderContinuationService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderContinuationService.test.ts @@ -11,6 +11,7 @@ import { type OrchestrationV2ThreadProjection, type OrchestrationV2Notification, } from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; @@ -19,6 +20,7 @@ import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; import * as Semaphore from "effect/Semaphore"; +import * as Stream from "effect/Stream"; import * as TestClock from "effect/testing/TestClock"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; @@ -530,6 +532,58 @@ describe("ProviderContinuationService", () => { }); }); + it.effect("releases a wake once its continuation run starts on another provider thread", () => { + return Effect.gen(function* () { + const messageId = MessageId.make("message-provider-continuation-moved"); + const runId = RunId.make("run-provider-continuation-moved"); + const released = yield* Deferred.make(); + const layerThreads = Layer.mock(ThreadManagementService.ThreadManagementService)({ + getThreadRecords: () => Effect.succeed(projection), + dispatch: () => + Effect.succeed({ + sequence: 1, + storedEvents: [ + { event: { type: "run.created", payload: { id: runId, userMessageId: messageId } } }, + ], + } as never), + // The thread moved to another session, so the run started there and is still running. + streamStoredEventsFrom: () => + Stream.make({ + event: { + type: "run.updated", + payload: { + id: runId, + status: "running", + startedAt: DateTime.makeUnsafe(0), + providerThreadId: ProviderThreadId.make("provider-thread-continuation-other"), + }, + }, + } as never), + }); + yield* Effect.gen(function* () { + const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests; + yield* requests.offer({ + ...request(), + messageId, + onRunEndedWithoutStarting: () => Deferred.succeed(released, undefined), + }); + yield* Deferred.await(released); + }).pipe( + Effect.provide( + Layer.merge( + ProviderContinuationRequests.layer, + ProviderContinuationService.layer.pipe( + Layer.provide( + Layer.mergeAll(IdAllocator.layer, ProviderContinuationRequests.layer, layerThreads), + ), + ), + ), + ), + Effect.scoped, + ); + }); + }); + it.effect("dispatches a current request exactly once", () => { return Effect.gen(function* () { const dispatched = yield* Queue.unbounded(); diff --git a/apps/server/src/orchestration-v2/ProviderContinuationService.ts b/apps/server/src/orchestration-v2/ProviderContinuationService.ts index 01f6e0acb989d..e1113d06149f4 100644 --- a/apps/server/src/orchestration-v2/ProviderContinuationService.ts +++ b/apps/server/src/orchestration-v2/ProviderContinuationService.ts @@ -95,6 +95,14 @@ export const layer = Layer.effectDiscard( : [], )[0]; if (runId === undefined) return; + // The thread moved to another provider thread, so this run can never show the request's output. + const startedOnAnotherProviderThread = (run: { + readonly startedAt: unknown; + readonly providerThreadId: string | null; + }) => + run.startedAt !== null && + run.providerThreadId !== null && + run.providerThreadId !== request.providerThreadId; yield* threads .streamStoredEventsFrom({ threadId: request.threadId, @@ -106,13 +114,15 @@ export const layer = Layer.effectDiscard( ({ event }) => event.type === "run.updated" && event.payload.id === runId && - ThreadManagementService.isTerminalRunStatus(event.payload.status), + (ThreadManagementService.isTerminalRunStatus(event.payload.status) || + startedOnAnotherProviderThread(event.payload)), ), Stream.runHead, Effect.flatMap((ended) => Option.isSome(ended) && ended.value.event.type === "run.updated" && - ended.value.event.payload.startedAt === null + (ended.value.event.payload.startedAt === null || + startedOnAnotherProviderThread(ended.value.event.payload)) ? onRunEndedWithoutStarting() : Effect.void, ), diff --git a/packages/provider-core/src/server/ProviderContinuationRequests.ts b/packages/provider-core/src/server/ProviderContinuationRequests.ts index 576edcc221a11..d60d05526ffde 100644 --- a/packages/provider-core/src/server/ProviderContinuationRequests.ts +++ b/packages/provider-core/src/server/ProviderContinuationRequests.ts @@ -30,7 +30,7 @@ export interface ProviderContinuationRequest { effect: Effect.Effect, ) => Effect.Effect, E, R>; readonly clearIfCurrent?: () => Effect.Effect; - /** Called when the run this request dispatched ends without starting, such as a cancelled queued run. */ + /** Called when the run this request dispatched ends without starting (such as a cancelled queued run) or ran on another provider thread. */ readonly onRunEndedWithoutStarting?: () => Effect.Effect; }