From 54f1f90187e3589d1d288830d385a60d91cf5365 Mon Sep 17 00:00:00 2001 From: Byntham <135178921+Byntham@users.noreply.github.com> Date: Tue, 6 Oct 2026 22:45:22 -0700 Subject: [PATCH 1/3] fix(server): tool calls cut off by a lost provider stream no longer wait forever When a run's provider event stream is lost mid tool call, for example when t3_worktree_handoff detaches the session running it, the run was marked failed but its open tool call stayed running. Later runs inherited it as background work, so the thread showed "Waiting on background task" after every later run until a server restart. The lost-stream fallback now cancels the run's open command and dynamic tool items with the run, as restart recovery does for a dead process. Provider-reported terminals are unchanged: their process is alive, so open items still pass to the next run. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../BackgroundWorkStop.integration.test.ts | 315 +++++++++++++----- .../RunExecutionService.test.ts | 51 +++ .../orchestration-v2/RunExecutionService.ts | 101 +++--- 3 files changed, 350 insertions(+), 117 deletions(-) diff --git a/apps/server/src/orchestration-v2/BackgroundWorkStop.integration.test.ts b/apps/server/src/orchestration-v2/BackgroundWorkStop.integration.test.ts index 6d5e0b92d87f..794cf3827d4f 100644 --- a/apps/server/src/orchestration-v2/BackgroundWorkStop.integration.test.ts +++ b/apps/server/src/orchestration-v2/BackgroundWorkStop.integration.test.ts @@ -14,12 +14,15 @@ import { ThreadId, TurnItemId, type OrchestrationV2DomainEvent, + type OrchestrationV2ProviderCapabilities, } from "@t3tools/contracts"; +import { derivePendingBackgroundWork } from "@t3tools/shared/orchestrationV2PendingBackgroundWork"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; import * as Queue from "effect/Queue"; import * as Stream from "effect/Stream"; +import { ClaudeProviderCapabilitiesV2 } from "./Adapters/ClaudeAdapterV2.ts"; import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; import * as EffectWorker from "./EffectWorker.ts"; import * as EventSink from "./EventSink.ts"; @@ -34,6 +37,94 @@ const driver = ProviderDriverKind.make("codex"); const instanceId = ProviderInstanceId.make("codex"); const modelSelection = { instanceId, model: "test-model" }; +// A provider whose turns start running, then report only what a test offers +// on `events`. +const scriptedAdapter = (input: { + readonly cwd: string; + readonly events: Queue.Queue; + readonly started: ProviderAdapter.ProviderAdapterV2TurnInput[]; + readonly capabilities: OrchestrationV2ProviderCapabilities; + readonly interrupts?: ProviderAdapter.ProviderAdapterV2InterruptInput[]; +}): ProviderAdapter.ProviderAdapterV2["Service"] => ({ + instanceId, + driver, + getCapabilities: () => Effect.succeed(input.capabilities), + planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }), + openSession: (session) => + Effect.gen(function* () { + const now = yield* DateTime.now; + return { + instanceId, + driver, + providerSessionId: session.providerSessionId, + providerSession: { + id: session.providerSessionId, + driver, + providerInstanceId: instanceId, + status: "ready", + cwd: input.cwd, + model: modelSelection.model, + capabilities: input.capabilities, + createdAt: now, + updatedAt: now, + lastError: null, + }, + events: Stream.fromQueue(input.events), + ensureThread: ({ threadId }) => + Effect.succeed({ + id: ProviderThreadId.make(`provider-thread:codex:${threadId}`), + driver, + providerInstanceId: instanceId, + providerSessionId: session.providerSessionId, + appThreadId: threadId, + ownerNodeId: null, + nativeThreadRef: { driver, nativeId: "native-thread", strength: "strong" }, + nativeConversationHeadRef: null, + status: "idle", + firstRunOrdinal: null, + lastRunOrdinal: null, + handoffIds: [], + forkedFrom: null, + createdAt: now, + updatedAt: now, + }), + resumeThread: ({ providerThread }) => Effect.succeed(providerThread), + startTurn: (turn) => + Effect.gen(function* () { + input.started.push(turn); + yield* Queue.offer(input.events, { + type: "provider_turn.updated", + driver, + providerTurn: { + id: ProviderTurnId.make(`provider-turn:${turn.attemptId}`), + providerThreadId: turn.providerThread.id, + nodeId: turn.rootNodeId, + runAttemptId: turn.attemptId, + nativeTurnRef: { + driver, + nativeId: `native:${turn.attemptId}`, + strength: "strong", + }, + ordinal: turn.providerTurnOrdinal, + status: "running", + startedAt: now, + completedAt: null, + }, + }); + }), + steerTurn: () => Effect.die("unused"), + interruptTurn: (interrupt) => + Effect.sync(() => { + input.interrupts?.push(interrupt); + }), + respondToRuntimeRequest: () => Effect.die("unused"), + readThreadSnapshot: () => Effect.die("unused"), + rollbackThread: () => Effect.die("unused"), + forkThread: () => Effect.die("unused"), + }; + }), +}); + // Codex turns leave commands running, then the thread moves to another // provider thread (a provider switch). Stop must end all of the Codex work, // including when it selects an older resumed run on the other provider thread. @@ -59,85 +150,13 @@ const stopEarlierBackgroundWork = ({ const events = yield* Queue.unbounded(); const started: ProviderAdapter.ProviderAdapterV2TurnInput[] = []; const interrupts: ProviderAdapter.ProviderAdapterV2InterruptInput[] = []; - const adapter: ProviderAdapter.ProviderAdapterV2["Service"] = { - instanceId, - driver, - getCapabilities: () => Effect.succeed(CodexProviderCapabilitiesV2), - planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }), - openSession: (input) => - Effect.gen(function* () { - const now = yield* DateTime.now; - return { - instanceId, - driver, - providerSessionId: input.providerSessionId, - providerSession: { - id: input.providerSessionId, - driver, - providerInstanceId: instanceId, - status: "ready", - cwd, - model: modelSelection.model, - capabilities: CodexProviderCapabilitiesV2, - createdAt: now, - updatedAt: now, - lastError: null, - }, - events: Stream.fromQueue(events), - ensureThread: ({ threadId }) => - Effect.succeed({ - id: ProviderThreadId.make(`provider-thread:codex:${threadId}`), - driver, - providerInstanceId: instanceId, - providerSessionId: input.providerSessionId, - appThreadId: threadId, - ownerNodeId: null, - nativeThreadRef: { driver, nativeId: "native-thread", strength: "strong" }, - nativeConversationHeadRef: null, - status: "idle", - firstRunOrdinal: null, - lastRunOrdinal: null, - handoffIds: [], - forkedFrom: null, - createdAt: now, - updatedAt: now, - }), - resumeThread: ({ providerThread }) => Effect.succeed(providerThread), - startTurn: (turn) => - Effect.gen(function* () { - started.push(turn); - yield* Queue.offer(events, { - type: "provider_turn.updated", - driver, - providerTurn: { - id: ProviderTurnId.make(`provider-turn:${turn.attemptId}`), - providerThreadId: turn.providerThread.id, - nodeId: turn.rootNodeId, - runAttemptId: turn.attemptId, - nativeTurnRef: { - driver, - nativeId: `native:${turn.attemptId}`, - strength: "strong", - }, - ordinal: turn.providerTurnOrdinal, - status: "running", - startedAt: now, - completedAt: null, - }, - }); - }), - steerTurn: () => Effect.die("unused"), - interruptTurn: (interrupt) => - Effect.sync(() => { - interrupts.push(interrupt); - }), - respondToRuntimeRequest: () => Effect.die("unused"), - readThreadSnapshot: () => Effect.die("unused"), - rollbackThread: () => Effect.die("unused"), - forkThread: () => Effect.die("unused"), - }; - }), - }; + const adapter = scriptedAdapter({ + cwd, + events, + started, + capabilities: CodexProviderCapabilitiesV2, + interrupts, + }); yield* Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; @@ -759,6 +778,144 @@ const stopEarlierBackgroundWork = ({ }), ); +// `t3_worktree_handoff` moves the calling thread, so the workspace change +// detaches the session that is running that very tool call. Nothing will report +// on the call again, so it is cancelled with the run instead of waiting as background work. +it.effect("cancels a tool call cut off when a workspace change detaches its session", () => + Effect.scoped( + Effect.gen(function* () { + const cwd = yield* checkpointWorkspace("workspace-detach-tool-call"); + const events = yield* Queue.unbounded(); + const started: ProviderAdapter.ProviderAdapterV2TurnInput[] = []; + const adapter = scriptedAdapter({ + cwd, + events, + started, + capabilities: ClaudeProviderCapabilitiesV2, + }); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const threadId = ThreadId.make("thread:workspace-detach-tool-call"); + const watch = (predicate: (event: OrchestrationV2DomainEvent) => boolean) => + orchestrator.streamDomainEvents.pipe( + Stream.filter(predicate), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make("create"), + threadId, + projectId: ProjectId.make("project:workspace-detach-tool-call"), + title: "Workspace detach", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: cwd, + createdBy: "user", + creationSource: "web", + }); + const running = yield* watch( + (event) => event.type === "provider-turn.updated" && event.payload.status === "running", + ); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("hand-off"), + threadId, + messageId: MessageId.make("message:hand-off"), + text: "Move this work to a worktree", + attachments: [], + dispatchMode: { type: "start_immediately" }, + createdBy: "user", + creationSource: "web", + }); + yield* worker.drain(); + yield* Fiber.join(running); + const turn = started[0]!; + const providerTurn = (yield* orchestrator.getThreadProjection(threadId)).providerTurns[0]!; + const toolCallId = TurnItemId.make("turn-item:worktree-handoff"); + const toolCallStarted = yield* watch( + (event) => event.type === "turn-item.updated" && event.payload.id === toolCallId, + ); + const now = yield* DateTime.now; + yield* Queue.offer(events, { + type: "turn_item.updated", + driver, + turnItem: { + id: toolCallId, + threadId, + runId: turn.runId, + nodeId: turn.rootNodeId, + providerThreadId: providerTurn.providerThreadId, + providerTurnId: providerTurn.id, + nativeItemRef: null, + parentItemId: null, + ordinal: 100, + status: "running", + title: null, + startedAt: now, + completedAt: null, + updatedAt: now, + type: "dynamic_tool", + toolName: "mcp__t3-code__t3_worktree_handoff", + input: { branch: "t3code/handoff" }, + }, + }); + yield* Fiber.join(toolCallStarted); + + // Replays from here, so the failure commit cannot slip past the watch. + const sink = yield* EventSink.EventSinkV2; + const failed = yield* sink + .stream({ threadId, afterSequence: yield* sink.latestSequence() }) + .pipe( + Stream.filter( + ({ event }) => + event.type === "run.updated" && + event.payload.id === turn.runId && + event.payload.status === "failed", + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + // What the workspace change's detach effect runs. A Claude session hosts + // one provider thread, so detaching it releases the process. + const sessions = yield* ProviderSessionManager.ProviderSessionManagerV2; + yield* sessions.detach({ + providerSessionId: turn.providerThread.providerSessionId!, + threadId, + detail: "Workspace changed.", + }); + yield* Fiber.join(failed); + + const after = yield* orchestrator.getThreadProjection(threadId); + assert.equal(after.turnItems.find((item) => item.id === toolCallId)?.status, "cancelled"); + assert.deepEqual( + derivePendingBackgroundWork({ + latestRun: after.runs.at(-1), + providerThreads: after.providerThreads, + turnItems: after.turnItems, + activeProviderThreadId: after.thread.activeProviderThreadId, + runs: after.runs, + }), + [], + ); + }).pipe( + Effect.provide( + ProviderReplayHarness.layerWithRegistry( + { name: "workspace-detach-tool-call" }, + ProviderAdapterRegistry.layerSingle(adapter), + { runEffectWorker: false }, + ), + ), + ); + }), + ), +); + it.effect("Stop reaches background work an earlier provider thread still runs", () => stopEarlierBackgroundWork({}), ); diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index 708205663787..5050556b9481 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -3450,6 +3450,57 @@ it.effect("refreshes pull requests after a provider stream exits with an error", }), ); +it.effect("cancels the run's open tool calls when its provider stream is lost", () => + Effect.gen(function* () { + const finishedItemId = TurnItemId.make("turn-item:lost-stream:finished"); + const { written } = yield* captureRootRunTermination({ + key: "lost-stream-open-tool", + shouldFinalizeRun: () => Effect.succeed(true), + events: (ids) => + Stream.concat( + Stream.make( + backgroundTurnItemEvent(ids, "dynamic_tool", "running", 1), + backgroundTurnItemEvent(ids, "command_execution", "running", 2, finishedItemId), + backgroundTurnItemEvent(ids, "command_execution", "completed", 3, finishedItemId), + ), + // A workspace change detaches the session while the tool call runs. + Stream.fail( + new ProviderAdapter.ProviderAdapterEventStreamError({ + driver, + providerSessionId: ProviderSessionId.make("session:detached"), + cause: "Workspace changed.", + }), + ), + ), + }); + assert.deepEqual( + written.map((item) => [item.type, item.status]), + [ + ["dynamic_tool", "cancelled"], + ["error", "failed"], + ], + ); + }), +); + +it.effect("leaves open tool calls to the next run when the provider reports the terminal", () => + Effect.gen(function* () { + const { written } = yield* captureRootRunTermination({ + key: "reported-terminal-open-tool", + shouldFinalizeRun: () => Effect.succeed(true), + events: (ids) => + Stream.make( + backgroundTurnItemEvent(ids, "command_execution", "running", 1), + rootTerminalEvent(ids, "interrupted"), + ), + }); + assert.deepEqual( + written.map((item) => item.type), + ["run_interrupt_result"], + ); + }), +); + it.effect("refreshes pull requests only once when startup failure closes its event stream", () => Effect.gen(function* () { const ingestionStarted = yield* Deferred.make(); diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index a316ce0ccd62..0b575ea624f2 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -566,6 +566,13 @@ export const layer: Layer.Layer< readonly shouldFinalizeRun?: () => Effect.Effect; readonly hasUnpairedRunInterruptRequest?: () => Effect.Effect; readonly openRunOwnedSubagents?: OpenRunOwnedSubagentProjection; + /** + * Open tool and command items of a run whose provider stream was lost. + * Nothing will report on them again, so they are cancelled with the run, + * as restart recovery does, instead of passing to the next run as + * background work. + */ + readonly orphanedTurnItems?: ReadonlyArray; readonly terminal: ProviderTerminalEvent; readonly failureItemPersisted: boolean; readonly refreshAfterTurn: Effect.Effect; @@ -636,6 +643,19 @@ export const layer: Layer.Layer< allocateEventId, }) : []; + const orphanedTurnItemEvents: Array = []; + for (const item of input.orphanedTurnItems ?? []) { + orphanedTurnItemEvents.push({ + id: yield* allocateEventId(), + type: "turn-item.updated", + threadId: item.threadId, + runId: input.run.id, + ...(item.nodeId === null ? {} : { nodeId: item.nodeId }), + providerInstanceId: input.run.providerInstanceId, + occurredAt: completedAt, + payload: { ...item, status: "cancelled", completedAt, updatedAt: completedAt }, + }); + } const persistedStatus = input.terminal.status === "completed" ? "waiting" : input.terminal.status; // Completion cohorts are advanced by Orchestrator while a provider @@ -690,6 +710,7 @@ export const layer: Layer.Layer< // Terminalize open run-owned subagent rows before the root run // settles so projections never keep a forever-running subagent card. ...cascadedSubagentEvents, + ...orphanedTurnItemEvents, ...(finalizedAttempt === null ? [] : [ @@ -955,9 +976,11 @@ export const layer: Layer.Layer< const providerThreadOwnerLost = yield* Ref.make(false); const activeChildProviderTurns = yield* Ref.make>(new Set()); const activeChildSubagents = yield* Ref.make>(new Set()); + // Open background items with their latest update, if one arrived on + // this stream. A lost stream cancels this run's tool and command items. const activeBackgroundTurnItems = yield* Ref.make< - ReadonlySet - >(new Set(inheritedBackgroundTurnItemsById.keys())); + ReadonlyMap + >(new Map(Array.from(inheritedBackgroundTurnItemsById.keys(), (id) => [id, undefined]))); const openRunOwnedSubagents = yield* Ref.make(emptyOpenRunOwnedSubagentProjection()); const finalizeRootRun = (terminal: ProviderTerminalEvent) => Effect.gen(function* () { @@ -1092,11 +1115,11 @@ export const layer: Layer.Layer< belongsToInheritedBackgroundItem) ) { yield* Ref.update(activeBackgroundTurnItems, (current) => { - const next = new Set(current); + const next = new Map(current); if (isSettledTurnItemStatus(event.turnItem.status)) { next.delete(event.turnItem.id); } else { - next.add(event.turnItem.id); + next.set(event.turnItem.id, event.turnItem); } return next; }); @@ -1291,41 +1314,43 @@ export const layer: Layer.Layer< Effect.andThen( finalized ? Effect.void - : Ref.get(latestProviderThread).pipe( - Effect.flatMap((providerThread) => - Ref.get(latestTurnItemOrdinal).pipe( - Effect.flatMap((latestItemOrdinal) => - Ref.get(openRunOwnedSubagents).pipe( - Effect.flatMap((openSubagents) => - writeFinalRunEvents({ - run: input.run, - rootNode: input.rootNode, - checkpointScope: input.checkpointScope, - providerThread, - attempt: input.attempt, - // The failure may be the ownership - // read itself, so check in the write. - writeIfRunCurrent: { - activeAttemptId: input.attempt.id, - expectedStatus: "running", - }, - openRunOwnedSubagents: openSubagents, - terminal: makeFailedTerminalEvent( - makeProviderFailure({ - cause: Cause.squash(cause), - class: "unknown", - }), - latestItemOrdinal + 1, - ), - failureItemPersisted: false, - refreshAfterTurn, - }), - ), - ), - ), + : Effect.gen(function* () { + const latestItemOrdinal = yield* Ref.get(latestTurnItemOrdinal); + yield* writeFinalRunEvents({ + run: input.run, + rootNode: input.rootNode, + checkpointScope: input.checkpointScope, + providerThread: yield* Ref.get(latestProviderThread), + attempt: input.attempt, + // The failure may be the ownership + // read itself, so check in the write. + writeIfRunCurrent: { + activeAttemptId: input.attempt.id, + expectedStatus: "running", + }, + openRunOwnedSubagents: yield* Ref.get(openRunOwnedSubagents), + // The stream that would finish them is gone, for + // example when a workspace change detaches the + // session mid tool call. + orphanedTurnItems: Array.from( + (yield* Ref.get(activeBackgroundTurnItems)).values(), + ).filter( + (item): item is OrchestrationV2TurnItem => + item?.runId === input.run.id && + (item.type === "command_execution" || + item.type === "dynamic_tool"), ), - ), - ), + terminal: makeFailedTerminalEvent( + makeProviderFailure({ + cause: Cause.squash(cause), + class: "unknown", + }), + latestItemOrdinal + 1, + ), + failureItemPersisted: false, + refreshAfterTurn, + }); + }), ), Effect.mapError( (writeCause) => From 5d0b5e1d6662324236fba8493c9df72446830e57 Mon Sep 17 00:00:00 2001 From: Byntham <135178921+Byntham@users.noreply.github.com> Date: Wed, 7 Oct 2026 00:06:12 -0700 Subject: [PATCH 2/3] fix(server): only a failed provider stream cancels a run's open tool calls A failure to store an event also ends a run's ingestion, but the provider can still be running the tool. Cancelling it then would hide live work from Stop and the next run. The cleanup now applies only when the provider's own stream fails, as when its session is released. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../RunExecutionService.test.ts | 31 ++++++++++++++++++- .../orchestration-v2/RunExecutionService.ts | 22 ++++++++----- 2 files changed, 44 insertions(+), 9 deletions(-) diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index 5050556b9481..025517023513 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -3483,6 +3483,34 @@ it.effect("cancels the run's open tool calls when its provider stream is lost", }), ); +it.effect("leaves open tool calls to the next run when only storing an event fails", () => + Effect.gen(function* () { + const { written } = yield* captureRootRunTermination({ + key: "ingest-failure-open-tool", + shouldFinalizeRun: () => Effect.succeed(true), + events: (ids) => + Stream.make( + backgroundTurnItemEvent(ids, "dynamic_tool", "running", 1), + backgroundTurnItemEvent(ids, "dynamic_tool", "running", 2), + ), + // The provider is still running the tool; only this server failed to store an update. + ingestNormalized: (ingest) => + ingest.event.type === "turn_item.updated" && ingest.event.turnItem.ordinal === 2 + ? Effect.fail( + new ProviderEventIngestor.ProviderEventPublishError({ + providerSessionId: ingest.providerSessionId, + eventCount: 1, + }), + ) + : Effect.succeed([]), + }); + assert.deepEqual( + written.map((item) => [item.type, item.status]), + [["error", "failed"]], + ); + }), +); + it.effect("leaves open tool calls to the next run when the provider reports the terminal", () => Effect.gen(function* () { const { written } = yield* captureRootRunTermination({ @@ -3557,6 +3585,7 @@ function captureRootRunTermination(input: { >; readonly startTurn?: ProviderAdapter.ProviderAdapterV2SessionRuntime["startTurn"]; readonly refreshAfterTurn?: Effect.Effect; + readonly ingestNormalized?: ProviderEventIngestor.ProviderEventIngestorV2["Service"]["ingestNormalized"]; }) { return Effect.gen(function* () { const ids = backgroundScenarioIds(input.key); @@ -3622,7 +3651,7 @@ function captureRootRunTermination(input: { }), IdAllocator.layer, Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({ - ingestNormalized: () => Effect.succeed([]), + ingestNormalized: input.ingestNormalized ?? (() => Effect.succeed([])), }), ServerSettings.layerTest(), Layer.succeed(RunFinalizationService.RunFinalizationObserver, { diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index 0b575ea624f2..37b7cffe1ce2 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -973,6 +973,9 @@ export const layer: Layer.Layer< ); const rootTerminalSeen = yield* Ref.make(false); const rootRunFinalized = yield* Ref.make(false); + // Set when the provider's stream itself fails, as when its session is + // released. A failure to store an event leaves the provider running. + const providerStreamLost = yield* Ref.make(false); const providerThreadOwnerLost = yield* Ref.make(false); const activeChildProviderTurns = yield* Ref.make>(new Set()); const activeChildSubagents = yield* Ref.make>(new Set()); @@ -1208,6 +1211,7 @@ export const layer: Layer.Layer< }); const filterAssistantEvent = makeAssistantStreamingFilter(responseStreamingMode); const providerEventFiber = yield* eventSubscription.events.pipe( + Stream.tapCause(() => Ref.set(providerStreamLost, true)), Stream.filterEffect((event) => Ref.modify(eventRouting, (state) => routeProviderEvent(event, routeIdentity, state)), ), @@ -1332,14 +1336,16 @@ export const layer: Layer.Layer< // The stream that would finish them is gone, for // example when a workspace change detaches the // session mid tool call. - orphanedTurnItems: Array.from( - (yield* Ref.get(activeBackgroundTurnItems)).values(), - ).filter( - (item): item is OrchestrationV2TurnItem => - item?.runId === input.run.id && - (item.type === "command_execution" || - item.type === "dynamic_tool"), - ), + orphanedTurnItems: (yield* Ref.get(providerStreamLost)) + ? Array.from( + (yield* Ref.get(activeBackgroundTurnItems)).values(), + ).filter( + (item): item is OrchestrationV2TurnItem => + item?.runId === input.run.id && + (item.type === "command_execution" || + item.type === "dynamic_tool"), + ) + : [], terminal: makeFailedTerminalEvent( makeProviderFailure({ cause: Cause.squash(cause), From 2f44ec9cff6372bd1e1bf03f69ef194319bb0d16 Mon Sep 17 00:00:00 2001 From: Byntham <135178921+Byntham@users.noreply.github.com> Date: Wed, 7 Oct 2026 11:33:23 -0700 Subject: [PATCH 3/3] fix(server): cancelled tool calls drop unserved image bytes like other updates The cancellation written for a lost stream's open tool calls came from the adapter's raw item, so it skipped the image stripping the ingestor applies to every stored update. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../RunExecutionService.test.ts | 43 +++++++++++++++++++ .../orchestration-v2/RunExecutionService.ts | 10 ++++- 2 files changed, 52 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index 025517023513..762d971b9f82 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -3483,6 +3483,49 @@ it.effect("cancels the run's open tool calls when its provider stream is lost", }), ); +it.effect("strips unserved image bytes from a tool call it cancels", () => + Effect.gen(function* () { + const imageBase64 = Buffer.alloc(3_000, 7).toString("base64"); + const { written } = yield* captureRootRunTermination({ + key: "lost-stream-tool-image", + shouldFinalizeRun: () => Effect.succeed(true), + events: (ids) => + Stream.concat( + Stream.make({ + type: "turn_item.updated", + driver, + turnItem: { + id: ids.itemId, + threadId: ids.threadId, + runId: ids.runId, + providerTurnId: ids.rootProviderTurnId, + ordinal: 1, + type: "dynamic_tool", + status: "running", + toolName: "Read", + output: { + type: "image", + file: { base64: imageBase64, type: "image/png", originalSize: 3_000 }, + }, + }, + } as ProviderAdapter.ProviderAdapterV2Event), + Stream.fail( + new ProviderAdapter.ProviderAdapterEventStreamError({ + driver, + providerSessionId: ProviderSessionId.make("session:detached"), + cause: "Workspace changed.", + }), + ), + ), + }); + const cancelled = written.find((item) => item.type === "dynamic_tool"); + assert.deepEqual(cancelled?.type === "dynamic_tool" ? cancelled.output : null, { + type: "image", + file: { type: "image/png", originalSize: 3_000, sizeBytes: 3_000 }, + }); + }), +); + it.effect("leaves open tool calls to the next run when only storing an event fails", () => Effect.gen(function* () { const { written } = yield* captureRootRunTermination({ diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index 37b7cffe1ce2..34a6fb549556 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -49,6 +49,7 @@ import { makeProviderFailureTurnItem, } from "@t3tools/provider-core/server/failure"; import * as RunFinalizationService from "./RunFinalizationService.ts"; +import { stripUnservedToolOutputImageBytes } from "./toolOutputImageBytes.ts"; export interface ProviderEventRoutingState { readonly ownedThreadIds: ReadonlySet; @@ -653,7 +654,14 @@ export const layer: Layer.Layer< ...(item.nodeId === null ? {} : { nodeId: item.nodeId }), providerInstanceId: input.run.providerInstanceId, occurredAt: completedAt, - payload: { ...item, status: "cancelled", completedAt, updatedAt: completedAt }, + // The snapshot is the adapter's item, so it gets the same image + // stripping the ingestor gives every stored update. + payload: stripUnservedToolOutputImageBytes({ + ...item, + status: "cancelled", + completedAt, + updatedAt: completedAt, + }), }); } const persistedStatus =