fix(server): a session released mid-turn no longer leaves tool calls running

apps/server/src/orchestration-v2/RunExecutionService.test.ts

@@ -3450,6 +3450,61 @@ it.effect("refreshes pull requests after a provider stream exits with an error",
   }),
 );
 
+it.effect("closes the run's open tool calls when its provider stream is lost", () =>
+  Effect.gen(function* () {
+    const key = "lost-stream-open-tools";
+    const toolItemId = TurnItemId.make(`turn-item:${key}`);
+    const finishedItemId = TurnItemId.make(`turn-item:${key}:finished`);
+    const commandItemId = TurnItemId.make(`turn-item:${key}:command`);
+    const imageData = "iVBORw0KGgo".repeat(16);
+    const unpairedNodeId = NodeId.make(`node:${key}:unpaired`);
+    const { written, observed } = yield* captureRootRunTermination({
+      key,
+      shouldFinalizeRun: () => Effect.succeed(true),
+      events: (ids) =>
+        Stream.concat(
+          Stream.fromIterable([
+            backgroundTurnItemEvent(ids, "dynamic_tool", "running", 1, finishedItemId),
+            backgroundTurnItemEvent(ids, "dynamic_tool", "completed", 2, finishedItemId),
+            toolCallNodeEvent(ids, "running"),
+            withToolOutput(
+              withToolCallNode(ids, backgroundTurnItemEvent(ids, "dynamic_tool", "running", 3)),
+              { source: { type: "base64", media_type: "image/png", data: imageData } },
+            ),
+            backgroundTurnItemEvent(ids, "command_execution", "running", 4, commandItemId),
+            // The stream ends before this node's item arrives.
+            toolCallNodeEvent(ids, "running", unpairedNodeId),
+          ]),
+          // A provider switch releases the session, which fails its event stream.
+          Stream.fail(
+            new ProviderAdapter.ProviderAdapterEventStreamError({
+              driver,
+              providerSessionId: ProviderSessionId.make(`session:${key}`),
+              cause: "Provider session released: provider_switch.",
+            }),
+          ),
+        ),
+    });
+    assert.deepEqual(observed, [
+      `node:${toolItemId}:node:interrupted`,
+      `node:${unpairedNodeId}:interrupted`,
+      "run:failed",
+      "pull-requests-refreshed",
+    ]);
+    assert.deepEqual(
+      written.map((item) => [item.type, item.type === "error" ? null : item.id, item.status]),
+      [
+        ["dynamic_tool", toolItemId, "interrupted"],
+        ["command_execution", commandItemId, "interrupted"],
+        ["error", null, "failed"],
+      ],
+    );
+    // Closed as the ingestor stored it: without the unserved image bytes.
+    const closedTool = written.find((item) => item.id === toolItemId);
+    assert.isFalse(JSON.stringify(closedTool).includes(imageData));
+  }),
+);
+
 it.effect("refreshes pull requests only once when startup failure closes its event stream", () =>
   Effect.gen(function* () {
     const ingestionStarted = yield* Deferred.make<void>();
@@ -3530,6 +3585,12 @@ function captureRootRunTermination(input: {
           if (event.type === "turn-item.updated") {
             yield* captureTurnItem(event.payload);
           }
+          if (event.type === "node.updated" && event.payload.kind === "tool_call") {
+            yield* Ref.update(observed, (current) => [
+              ...current,
+              `node:${event.payload.id}:${event.payload.status}`,
+            ]);
+          }
           if (event.type === "run.updated") {
             yield* Ref.update(observed, (current) => [...current, `run:${event.payload.status}`]);
           }
@@ -3772,6 +3833,47 @@ function backgroundTurnItemEvent(
   } as ProviderAdapter.ProviderAdapterV2Event;
 }
 
+function toolCallNodeEvent(
+  ids: BackgroundScenarioIds,
+  status: "running" | "completed",
+  nodeId: NodeId = NodeId.make(`${ids.itemId}:node`),
+): ProviderAdapter.ProviderAdapterV2Event {
+  return {
+    type: "node.updated",
+    driver,
+    node: {
+      id: nodeId,
+      threadId: ids.threadId,
+      runId: ids.runId,
+      kind: "tool_call",
+      status,
+    },
+  } as ProviderAdapter.ProviderAdapterV2Event;
+}
+
+function withToolCallNode(
+  ids: BackgroundScenarioIds,
+  event: ProviderAdapter.ProviderAdapterV2Event,
+): ProviderAdapter.ProviderAdapterV2Event {
+  if (event.type !== "turn_item.updated") {
+    return event;
+  }
+  return { ...event, turnItem: { ...event.turnItem, nodeId: NodeId.make(`${ids.itemId}:node`) } };
+}
+
+function withToolOutput(
+  event: ProviderAdapter.ProviderAdapterV2Event,
+  output: unknown,
+): ProviderAdapter.ProviderAdapterV2Event {
+  if (event.type !== "turn_item.updated") {
+    return event;
+  }
+  return {
+    ...event,
+    turnItem: { ...event.turnItem, output },
+  } as ProviderAdapter.ProviderAdapterV2Event;
+}
+
 function backgroundTurnItemEventForRun(
   ids: BackgroundScenarioIds,
   runId: RunId,

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<ThreadId>;
@@ -318,6 +319,71 @@ export function cascadeTerminalizeRunOwnedSubagents(input: {
   });
 }
 
+/**
+ * A root run's tool calls and their nodes that have not ended yet. Items are
+ * kept as the ingestor persists them, without unserved image bytes.
+ */
+type OpenRunToolCalls = {
+  readonly turnItems: ReadonlyMap<TurnItemId, OrchestrationV2TurnItem>;
+  readonly nodes: ReadonlyMap<NodeId, OrchestrationV2ExecutionNode>;
+};
+
+function emptyOpenRunToolCalls(): OpenRunToolCalls {
+  return { turnItems: new Map(), nodes: new Map() };
+}
+
+/**
+ * Ends the tool calls a run left open when its provider event stream was lost,
+ * such as when a provider switch released the session mid-turn. No process is
+ * left to report how they ended, so they would otherwise stay running. They
+ * were cut short, not failed: a call such as t3_thread_configure may have done
+ * its work before the session went away.
+ */
+function interruptOpenRunToolCalls(input: {
+  readonly run: OrchestrationV2Run;
+  readonly open: OpenRunToolCalls;
+  readonly completedAt: DateTime.Utc;
+  readonly allocateEventId: () => Effect.Effect<EventId, IdAllocator.IdAllocatorV2AllocationError>;
+}): Effect.Effect<
+  ReadonlyArray<OrchestrationV2DomainEvent>,
+  IdAllocator.IdAllocatorV2AllocationError
+> {
+  return Effect.gen(function* () {
+    const events: Array<OrchestrationV2DomainEvent> = [];
+    // A node can arrive without its item when the stream ends between the two.
+    for (const node of input.open.nodes.values()) {
+      events.push({
+        id: yield* input.allocateEventId(),
+        type: "node.updated",
+        threadId: node.threadId,
+        runId: input.run.id,
+        nodeId: node.id,
+        providerInstanceId: input.run.providerInstanceId,
+        occurredAt: input.completedAt,
+        payload: { ...node, status: "interrupted", completedAt: input.completedAt },
+      });
+    }
+    for (const turnItem of input.open.turnItems.values()) {
+      events.push({
+        id: yield* input.allocateEventId(),
+        type: "turn-item.updated",
+        threadId: turnItem.threadId,
+        runId: input.run.id,
+        ...(turnItem.nodeId === null ? {} : { nodeId: turnItem.nodeId }),
+        providerInstanceId: input.run.providerInstanceId,
+        occurredAt: input.completedAt,
+        payload: {
+          ...turnItem,
+          status: "interrupted",
+          completedAt: input.completedAt,
+          updatedAt: input.completedAt,
+        },
+      });
+    }
+    return events;
+  });
+}
+
 export function finalProviderThreadStatus(
   disposition: ProviderTerminalEvent["threadDisposition"],
 ): OrchestrationV2ProviderThread["status"] {
@@ -566,6 +632,8 @@ export const layer: Layer.Layer<
       readonly shouldFinalizeRun?: () => Effect.Effect<boolean, ProjectionStoreV2Error>;
       readonly hasUnpairedRunInterruptRequest?: () => Effect.Effect<boolean, never>;
       readonly openRunOwnedSubagents?: OpenRunOwnedSubagentProjection;
+      /** Set only when the provider stream was lost before the turn ended. */
+      readonly abandonedToolCalls?: OpenRunToolCalls;
       readonly terminal: ProviderTerminalEvent;
       readonly failureItemPersisted: boolean;
       readonly refreshAfterTurn: Effect.Effect<void>;
@@ -636,6 +704,15 @@ export const layer: Layer.Layer<
                 allocateEventId,
               })
             : [];
+        const abandonedToolCallEvents =
+          input.abandonedToolCalls === undefined
+            ? []
+            : yield* interruptOpenRunToolCalls({
+                run: input.run,
+                open: input.abandonedToolCalls,
+                completedAt,
+                allocateEventId,
+              });
         const persistedStatus =
           input.terminal.status === "completed" ? "waiting" : input.terminal.status;
         // Completion cohorts are advanced by Orchestrator while a provider
@@ -690,6 +767,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,
+            ...abandonedToolCallEvents,
             ...(finalizedAttempt === null
               ? []
               : [
@@ -959,6 +1037,7 @@ export const layer: Layer.Layer<
             ReadonlySet<OrchestrationV2TurnItem["id"]>
           >(new Set(inheritedBackgroundTurnItemsById.keys()));
           const openRunOwnedSubagents = yield* Ref.make(emptyOpenRunOwnedSubagentProjection());
+          const openToolCalls = yield* Ref.make(emptyOpenRunToolCalls());
           const finalizeRootRun = (terminal: ProviderTerminalEvent) =>
             Effect.gen(function* () {
               if (yield* Ref.get(rootRunFinalized)) {
@@ -1057,6 +1136,23 @@ export const layer: Layer.Layer<
                   });
                 }
               }
+              if (
+                event.type === "node.updated" &&
+                event.node.kind === "tool_call" &&
+                event.node.threadId === input.run.threadId &&
+                event.node.runId === input.run.id
+              ) {
+                const node = event.node;
+                yield* Ref.update(openToolCalls, (current) => {
+                  const nodes = new Map(current.nodes);
+                  if (isOpenExecutionNodeStatus(node.status)) {
+                    nodes.set(node.id, node);
+                  } else {
+                    nodes.delete(node.id);
+                  }
+                  return { ...current, nodes };
+                });
+              }
               if (event.type === "node.updated") {
                 const belongsToRootSubagent =
                   event.node.kind === "subagent" && event.node.runId === input.run.id;
@@ -1112,6 +1208,23 @@ export const layer: Layer.Layer<
                     return { ...current, childTurnItems };
                   });
                 }
+                if (
+                  belongsToRootRun &&
+                  event.turnItem.threadId === input.run.threadId &&
+                  (event.turnItem.type === "command_execution" ||
+                    event.turnItem.type === "dynamic_tool")
+                ) {
+                  const toolItem = event.turnItem;
+                  yield* Ref.update(openToolCalls, (current) => {
+                    const turnItems = new Map(current.turnItems);
+                    if (isSettledTurnItemStatus(toolItem.status)) {
+                      turnItems.delete(toolItem.id);
+                    } else {
+                      turnItems.set(toolItem.id, stripUnservedToolOutputImageBytes(toolItem));
+                    }
+                    return { ...current, turnItems };
+                  });
+                }
                 if (belongsToRootRun && event.turnItem.type === "subagent") {
                   const subagentItem = event.turnItem;
                   yield* Ref.update(openRunOwnedSubagents, (current) => {
@@ -1295,8 +1408,11 @@ export const layer: Layer.Layer<
                             Effect.flatMap((providerThread) =>
                               Ref.get(latestTurnItemOrdinal).pipe(
                                 Effect.flatMap((latestItemOrdinal) =>
-                                  Ref.get(openRunOwnedSubagents).pipe(
-                                    Effect.flatMap((openSubagents) =>
+                                  Effect.all([
+                                    Ref.get(openRunOwnedSubagents),
+                                    Ref.get(openToolCalls),
+                                  ]).pipe(
+                                    Effect.flatMap(([openSubagents, abandonedToolCalls]) =>
                                       writeFinalRunEvents({
                                         run: input.run,
                                         rootNode: input.rootNode,
@@ -1310,6 +1426,7 @@ export const layer: Layer.Layer<
                                           expectedStatus: "running",
                                         },
                                         openRunOwnedSubagents: openSubagents,
+                                        abandonedToolCalls,
                                         terminal: makeFailedTerminalEvent(
                                           makeProviderFailure({
                                             cause: Cause.squash(cause),