refactor(agents): share Responses tool-call tracking (#108721)

2026-07-25 14:21:12 +00:00 · 2026-07-16 00:41:30 -07:00
parent 527c8cee0c
commit 7c19c3c068
4 changed files with 207 additions and 279 deletions
--- a/packages/ai/src/internal/openai.ts
+++ b/packages/ai/src/internal/openai.ts
@@ -7,6 +7,7 @@ export * from "../providers/openai-prompt-cache.js";
 export * from "../providers/openai-reasoning-effort.js";
 export * from "../providers/openai-responses.js";
 export * from "../providers/openai-responses-stream-compat.js";
+export * from "../providers/openai-responses-tool-call-tracker.js";
 export * from "../providers/openai-stop-reason.js";
 export * from "../providers/openai-tool-projection.js";
 export * from "../providers/openai-tool-schema.js";
--- a/packages/ai/src/providers/openai-responses-shared.ts
+++ b/packages/ai/src/providers/openai-responses-shared.ts
@@ -57,6 +57,11 @@ import {
  isResponsesTextContentPartType,
  resolveResponsesMessageSnapshotCollapse,
 } from "./openai-responses-stream-compat.js";
+import {
+  createResponsesToolCallTracker,
+  readResponsesToolCallItemIdentity,
+  type ResponsesToolCallState,
+} from "./openai-responses-tool-call-tracker.js";
 import { convertResponsesToolPayload } from "./openai-responses-tools.js";
 import { describeToolResultMediaPlaceholder, extractToolResultText } from "./tool-result-text.js";
 import { transformMessages } from "./transform-messages.js";
@@ -692,14 +697,11 @@ export async function processResponsesStream<TApi extends Api>(
  let currentBlock: ThinkingContent | TextContent | (ToolCall & { partialJson: string }) | null =
    null;
  type StreamingToolCallBlock = ToolCall & { partialJson: string };
-  type StreamingToolCallIdentity = { itemId?: string; callId?: string };
-  type StreamingToolCallState = StreamingToolCallIdentity & {
+  type StreamingToolCallState = ResponsesToolCallState & {
    block: StreamingToolCallBlock;
    contentIndex: number;
-    argumentStreamReliable: boolean;
  };
-  const toolCallsByOutputIndex = new Map<number, StreamingToolCallState>();
-  const unindexedToolCalls = new Set<StreamingToolCallState>();
+  const streamingToolCalls = createResponsesToolCallTracker<StreamingToolCallState>();
  let lastTextBlock: {
    block: TextContent;
    index: number;
@@ -711,119 +713,10 @@ export async function processResponsesStream<TApi extends Api>(
  let pendingMessageText: string | null = null;
  const blocks = output.content;
  const blockIndex = () => blocks.length - 1;
-  const readOutputIndex = (event: { output_index?: unknown }): number | undefined =>
-    typeof event.output_index === "number" &&
-    Number.isInteger(event.output_index) &&
-    event.output_index >= 0
-      ? event.output_index
-      : undefined;
  const readIdentityValue = (value: unknown): string | undefined => {
    const identity = typeof value === "string" ? value.trim() : "";
    return identity || undefined;
  };
-  const readEventToolCallIdentity = (event: { item_id?: unknown }): StreamingToolCallIdentity => ({
-    itemId: readIdentityValue(event.item_id),
-  });
-  const readItemToolCallIdentity = (item: {
-    id?: unknown;
-    call_id?: unknown;
-  }): StreamingToolCallIdentity => ({
-    itemId: readIdentityValue(item.id),
-    callId: readIdentityValue(item.call_id),
-  });
-  const identitiesConflict = (
-    state: StreamingToolCallState,
-    identity: StreamingToolCallIdentity,
-  ): boolean =>
-    Boolean(
-      (state.itemId && identity.itemId && state.itemId !== identity.itemId) ||
-      (state.callId && identity.callId && state.callId !== identity.callId),
-    );
-  const sharesIdentity = (
-    state: StreamingToolCallState,
-    identity: StreamingToolCallIdentity,
-  ): boolean =>
-    Boolean(
-      (state.itemId && identity.itemId && state.itemId === identity.itemId) ||
-      (state.callId && identity.callId && state.callId === identity.callId),
-    );
-  const adoptToolCallIdentity = (
-    state: StreamingToolCallState,
-    identity: StreamingToolCallIdentity,
-  ): StreamingToolCallState => {
-    state.itemId ??= identity.itemId;
-    state.callId ??= identity.callId;
-    return state;
-  };
-  const resolveCompatibleToolCall = (
-    candidates: Iterable<StreamingToolCallState>,
-    identity: StreamingToolCallIdentity,
-  ): StreamingToolCallState | undefined => {
-    const uniqueCandidates = [...new Set(candidates)];
-    if (!identity.itemId && !identity.callId) {
-      return uniqueCandidates.length === 1 ? uniqueCandidates.at(0) : undefined;
-    }
-    const compatible = uniqueCandidates.filter((state) => !identitiesConflict(state, identity));
-    const matches = compatible.filter((state) => sharesIdentity(state, identity));
-    const matched = matches.length === 1 ? matches.at(0) : undefined;
-    if (matched) {
-      return adoptToolCallIdentity(matched, identity);
-    }
-    // Only a sole active call may adopt an identity it did not already know.
-    // Parallel calls require a positive match so missing indices stay fail-closed.
-    const soleCompatible =
-      uniqueCandidates.length === 1 && compatible.length === 1 && matches.length === 0
-        ? compatible.at(0)
-        : undefined;
-    return soleCompatible ? adoptToolCallIdentity(soleCompatible, identity) : undefined;
-  };
-  const resolveStreamingToolCall = (
-    event: { output_index?: unknown; item_id?: unknown },
-    identity: StreamingToolCallIdentity = readEventToolCallIdentity(event),
-  ): StreamingToolCallState | undefined => {
-    const outputIndex = readOutputIndex(event);
-    if (outputIndex !== undefined) {
-      const indexed = toolCallsByOutputIndex.get(outputIndex);
-      if (indexed) {
-        if (indexed.callId && identity.callId && indexed.callId !== identity.callId) {
-          return undefined;
-        }
-        // output_index owns routing once registered, but call_id stays stable;
-        // compatible providers may rotate item_id for the same output item.
-        return adoptToolCallIdentity(indexed, identity);
-      }
-      // A compatibility stream may add calls without indices, then start
-      // including them. Bind only the one identity-matched (or sole) candidate.
-      const unindexed = resolveCompatibleToolCall(unindexedToolCalls, identity);
-      if (unindexed) {
-        unindexedToolCalls.delete(unindexed);
-        toolCallsByOutputIndex.set(outputIndex, unindexed);
-      }
-      return unindexed;
-    }
-
-    return resolveCompatibleToolCall(
-      [...toolCallsByOutputIndex.values(), ...unindexedToolCalls],
-      identity,
-    );
-  };
-  const forgetStreamingToolCall = (toolCall: StreamingToolCallState) => {
-    for (const [trackedIndex, tracked] of toolCallsByOutputIndex) {
-      if (tracked === toolCall) {
-        toolCallsByOutputIndex.delete(trackedIndex);
-      }
-    }
-    unindexedToolCalls.delete(toolCall);
-  };
-  const markActiveToolCallArgumentsUnreliable = () => {
-    // An unrouteable argument event may belong to any active call. Only an
-    // authoritative full argument snapshot can recover that call.
-    for (const toolCall of new Set([...toolCallsByOutputIndex.values(), ...unindexedToolCalls])) {
-      toolCall.argumentStreamReliable = false;
-    }
-  };
-  const hasActiveStreamingToolCall = () =>
-    toolCallsByOutputIndex.size > 0 || unindexedToolCalls.size > 0;
  // Opening fragments may carry the only function name. A conflicting
  // completion must never retarget an already-started call.
  const resolveCompletedToolCallName = (
@@ -910,31 +803,24 @@ export async function processResponsesStream<TApi extends Api>(
          stream.push({ type: "text_start", contentIndex: blockIndex(), partial: output });
        }
      } else if (item.type === "function_call") {
-        const outputIndex = readOutputIndex(event);
-        if (outputIndex !== undefined && toolCallsByOutputIndex.has(outputIndex)) {
-          throw new Error(`Responses stream reused active tool-call output index ${outputIndex}`);
-        }
-        currentItem = item;
-        currentBlock = {
+        const toolCallBlock: StreamingToolCallBlock = {
          type: "toolCall",
          id: resolveResponsesToolCallId(item),
          name: readIdentityValue(item.name) ?? "",
          arguments: {},
          partialJson: item.arguments || "",
        };
-        output.content.push(currentBlock);
-        const contentIndex = blockIndex();
-        const toolCallState = {
-          block: currentBlock,
+        const contentIndex = output.content.length;
+        const toolCallState: StreamingToolCallState = {
+          block: toolCallBlock,
          contentIndex,
          argumentStreamReliable: true,
-          ...readItemToolCallIdentity(item),
+          ...readResponsesToolCallItemIdentity(item),
        };
-        if (outputIndex !== undefined) {
-          toolCallsByOutputIndex.set(outputIndex, toolCallState);
-        } else {
-          unindexedToolCalls.add(toolCallState);
-        }
+        streamingToolCalls.register(event, toolCallState);
+        currentItem = item;
+        currentBlock = toolCallBlock;
+        output.content.push(toolCallBlock);
        stream.push({ type: "toolcall_start", contentIndex, partial: output });
      }
    } else if (event.type === "response.reasoning_summary_part.added") {
@@ -1057,7 +943,7 @@ export async function processResponsesStream<TApi extends Api>(
        }
      }
    } else if (event.type === "response.function_call_arguments.delta") {
-      const toolCall = resolveStreamingToolCall(event);
+      const toolCall = streamingToolCalls.resolve(event);
      if (toolCall) {
        toolCall.block.partialJson += event.delta;
        toolCall.block.arguments = parseStreamingJson(toolCall.block.partialJson);
@@ -1067,11 +953,11 @@ export async function processResponsesStream<TApi extends Api>(
          delta: event.delta,
          partial: output,
        });
-      } else if (hasActiveStreamingToolCall()) {
-        markActiveToolCallArgumentsUnreliable();
+      } else if (streamingToolCalls.hasActive()) {
+        streamingToolCalls.markArgumentsUnreliable();
      }
    } else if (event.type === "response.function_call_arguments.done") {
-      const toolCall = resolveStreamingToolCall(event);
+      const toolCall = streamingToolCalls.resolve(event);
      if (toolCall) {
        const previousPartialJson = toolCall.block.partialJson;
        const doneArguments = typeof event.arguments === "string" ? event.arguments : undefined;
@@ -1096,8 +982,8 @@ export async function processResponsesStream<TApi extends Api>(
            });
          }
        }
-      } else if (hasActiveStreamingToolCall()) {
-        markActiveToolCallArgumentsUnreliable();
+      } else if (streamingToolCalls.hasActive()) {
+        streamingToolCalls.markArgumentsUnreliable();
      }
    } else if (event.type === "response.output_item.done") {
      const item = event.item;
@@ -1176,10 +1062,13 @@ export async function processResponsesStream<TApi extends Api>(
        }
        currentBlock = null;
      } else if (item.type === "function_call") {
-        const streamingToolCall = resolveStreamingToolCall(event, readItemToolCallIdentity(item));
+        const streamingToolCall = streamingToolCalls.resolve(
+          event,
+          readResponsesToolCallItemIdentity(item),
+        );
        // Do not turn an unresolved completion into a second public call while
        // an indexed call is still open. Its identity or index must match.
-        if (!streamingToolCall && hasActiveStreamingToolCall()) {
+        if (!streamingToolCall && streamingToolCalls.hasActive()) {
          continue;
        }
        const completedName = resolveCompletedToolCallName(streamingToolCall, item.name);
@@ -1224,7 +1113,7 @@ export async function processResponsesStream<TApi extends Api>(
        }

        if (streamingToolCall) {
-          forgetStreamingToolCall(streamingToolCall);
+          streamingToolCalls.forget(streamingToolCall);
        }
        if (currentBlock === toolCall) {
          currentBlock = null;
@@ -1238,7 +1127,7 @@ export async function processResponsesStream<TApi extends Api>(
        });
      }
    } else if (event.type === "response.completed") {
-      if (hasActiveStreamingToolCall()) {
+      if (streamingToolCalls.hasActive()) {
        throw new Error("Responses stream completed with unresolved tool calls");
      }
      const response = event.response;
@@ -1289,7 +1178,7 @@ export async function processResponsesStream<TApi extends Api>(
      throw new Error(msg);
    }
  }
-  if (hasActiveStreamingToolCall()) {
+  if (streamingToolCalls.hasActive()) {
    throw new Error("Responses stream ended with unresolved tool calls");
  }
 }
--- a/packages/ai/src/providers/openai-responses-tool-call-tracker.ts
+++ b/packages/ai/src/providers/openai-responses-tool-call-tracker.ts
@@ -0,0 +1,148 @@
+export type ResponsesToolCallIdentity = { itemId?: string; callId?: string };
+
+export type ResponsesToolCallState = ResponsesToolCallIdentity & {
+  argumentStreamReliable: boolean;
+};
+
+type ResponsesToolCallEvent = {
+  output_index?: unknown;
+  item_id?: unknown;
+};
+
+function readIdentityValue(value: unknown): string | undefined {
+  const identity = typeof value === "string" ? value.trim() : "";
+  return identity || undefined;
+}
+
+function readOutputIndex(event: ResponsesToolCallEvent): number | undefined {
+  return typeof event.output_index === "number" &&
+    Number.isInteger(event.output_index) &&
+    event.output_index >= 0
+    ? event.output_index
+    : undefined;
+}
+
+function readEventIdentity(event: ResponsesToolCallEvent): ResponsesToolCallIdentity {
+  return { itemId: readIdentityValue(event.item_id) };
+}
+
+export function readResponsesToolCallItemIdentity(item: {
+  id?: unknown;
+  call_id?: unknown;
+}): ResponsesToolCallIdentity {
+  return {
+    itemId: readIdentityValue(item.id),
+    callId: readIdentityValue(item.call_id),
+  };
+}
+
+export function createResponsesToolCallTracker<TState extends ResponsesToolCallState>() {
+  const indexedCalls = new Map<number, TState>();
+  const unindexedCalls = new Set<TState>();
+
+  const identitiesConflict = (state: TState, identity: ResponsesToolCallIdentity): boolean =>
+    Boolean(
+      (state.itemId && identity.itemId && state.itemId !== identity.itemId) ||
+      (state.callId && identity.callId && state.callId !== identity.callId),
+    );
+
+  const sharesIdentity = (state: TState, identity: ResponsesToolCallIdentity): boolean =>
+    Boolean(
+      (state.itemId && identity.itemId && state.itemId === identity.itemId) ||
+      (state.callId && identity.callId && state.callId === identity.callId),
+    );
+
+  const adoptIdentity = (state: TState, identity: ResponsesToolCallIdentity): TState => {
+    state.itemId ??= identity.itemId;
+    state.callId ??= identity.callId;
+    return state;
+  };
+
+  const resolveCompatible = (
+    candidates: Iterable<TState>,
+    identity: ResponsesToolCallIdentity,
+  ): TState | undefined => {
+    const uniqueCandidates = [...new Set(candidates)];
+    if (!identity.itemId && !identity.callId) {
+      return uniqueCandidates.length === 1 ? uniqueCandidates.at(0) : undefined;
+    }
+    const compatible = uniqueCandidates.filter((state) => !identitiesConflict(state, identity));
+    const matches = compatible.filter((state) => sharesIdentity(state, identity));
+    const matched = matches.length === 1 ? matches.at(0) : undefined;
+    if (matched) {
+      return adoptIdentity(matched, identity);
+    }
+
+    // Only a sole active call may adopt an identity it did not already know.
+    // Parallel calls require a positive match so missing indices stay fail-closed.
+    const soleCompatible =
+      uniqueCandidates.length === 1 && compatible.length === 1 && matches.length === 0
+        ? compatible.at(0)
+        : undefined;
+    return soleCompatible ? adoptIdentity(soleCompatible, identity) : undefined;
+  };
+
+  return {
+    register(event: ResponsesToolCallEvent, state: TState): void {
+      const outputIndex = readOutputIndex(event);
+      if (outputIndex === undefined) {
+        unindexedCalls.add(state);
+        return;
+      }
+      if (indexedCalls.has(outputIndex)) {
+        throw new Error(`Responses stream reused active tool-call output index ${outputIndex}`);
+      }
+      indexedCalls.set(outputIndex, state);
+    },
+
+    resolve(
+      event: ResponsesToolCallEvent,
+      identity: ResponsesToolCallIdentity = readEventIdentity(event),
+    ): TState | undefined {
+      const outputIndex = readOutputIndex(event);
+      if (outputIndex !== undefined) {
+        const indexed = indexedCalls.get(outputIndex);
+        if (indexed) {
+          if (indexed.callId && identity.callId && indexed.callId !== identity.callId) {
+            return undefined;
+          }
+          // output_index owns routing once registered, but call_id stays stable;
+          // compatible providers may rotate item_id for the same output item.
+          return adoptIdentity(indexed, identity);
+        }
+
+        // A compatibility stream may add calls without indices, then start
+        // including them. Bind only the one identity-matched (or sole) candidate.
+        const unindexed = resolveCompatible(unindexedCalls, identity);
+        if (unindexed) {
+          unindexedCalls.delete(unindexed);
+          indexedCalls.set(outputIndex, unindexed);
+        }
+        return unindexed;
+      }
+
+      return resolveCompatible([...indexedCalls.values(), ...unindexedCalls], identity);
+    },
+
+    forget(toolCall: TState): void {
+      for (const [outputIndex, tracked] of indexedCalls) {
+        if (tracked === toolCall) {
+          indexedCalls.delete(outputIndex);
+        }
+      }
+      unindexedCalls.delete(toolCall);
+    },
+
+    markArgumentsUnreliable(): void {
+      // An unrouteable argument event may belong to any active call. Only an
+      // authoritative full argument snapshot can recover that call.
+      for (const toolCall of new Set([...indexedCalls.values(), ...unindexedCalls])) {
+        toolCall.argumentStreamReliable = false;
+      }
+    },
+
+    hasActive(): boolean {
+      return indexedCalls.size > 0 || unindexedCalls.size > 0;
+    },
+  };
+}
--- a/src/agents/openai-responses-transport.ts
+++ b/src/agents/openai-responses-transport.ts
@@ -3,12 +3,14 @@
 */
 import { randomUUID } from "node:crypto";
 import {
+  createResponsesToolCallTracker,
  isOpenAICompatibleAzureResponsesBaseUrl,
  isResponsesTextContentPartType,
  isResponsesTextDeltaEventType,
  normalizeOpenAIReasoningEffort,
  normalizeOpenAIStrictToolParameters,
  projectOpenAITools,
+  readResponsesToolCallItemIdentity,
  reconcileOpenAIResponsesToolChoice,
  resolveAzureDeploymentNameFromMap,
  resolveOpenAIReasoningEffortForModel,
@@ -16,6 +18,7 @@ import {
  type OpenAIApiReasoningEffort,
  type OpenAIReasoningEffort,
  type OpenAIToolProjection,
+  type ResponsesToolCallState,
 } from "@openclaw/ai/internal/openai";
 import {
  calculateCost,
@@ -1226,14 +1229,11 @@ async function processResponsesStream(
  };
  let currentItem: Record<string, unknown> | null = null;
  let currentBlock: Record<string, unknown> | null = null;
-  type StreamingToolCallIdentity = { itemId?: string; callId?: string };
-  type StreamingToolCallState = StreamingToolCallIdentity & {
+  type StreamingToolCallState = ResponsesToolCallState & {
    block: Record<string, unknown>;
    contentIndex: number;
-    argumentStreamReliable: boolean;
  };
-  const toolCallsByOutputIndex = new Map<number, StreamingToolCallState>();
-  const unindexedToolCalls = new Set<StreamingToolCallState>();
+  const streamingToolCalls = createResponsesToolCallTracker<StreamingToolCallState>();
  let lastTextBlock: {
    block: Record<string, unknown>;
    index: number;
@@ -1248,116 +1248,10 @@ async function processResponsesStream(
  const eventTypes = new Map<string, number>();
  const sseDebugMode = resolveModelSseDebugMode();
  const blockIndex = () => output.content.length - 1;
-  const readOutputIndex = (event: Record<string, unknown>): number | undefined =>
-    typeof event.output_index === "number" &&
-    Number.isInteger(event.output_index) &&
-    event.output_index >= 0
-      ? event.output_index
-      : undefined;
  const readIdentityValue = (value: unknown): string | undefined => {
    const identity = typeof value === "string" ? value.trim() : "";
    return identity || undefined;
  };
-  const readEventToolCallIdentity = (
-    event: Record<string, unknown>,
-  ): StreamingToolCallIdentity => ({ itemId: readIdentityValue(event.item_id) });
-  const readItemToolCallIdentity = (item: Record<string, unknown>): StreamingToolCallIdentity => ({
-    itemId: readIdentityValue(item.id),
-    callId: readIdentityValue(item.call_id),
-  });
-  const identitiesConflict = (
-    state: StreamingToolCallState,
-    identity: StreamingToolCallIdentity,
-  ): boolean =>
-    Boolean(
-      (state.itemId && identity.itemId && state.itemId !== identity.itemId) ||
-      (state.callId && identity.callId && state.callId !== identity.callId),
-    );
-  const sharesIdentity = (
-    state: StreamingToolCallState,
-    identity: StreamingToolCallIdentity,
-  ): boolean =>
-    Boolean(
-      (state.itemId && identity.itemId && state.itemId === identity.itemId) ||
-      (state.callId && identity.callId && state.callId === identity.callId),
-    );
-  const adoptToolCallIdentity = (
-    state: StreamingToolCallState,
-    identity: StreamingToolCallIdentity,
-  ): StreamingToolCallState => {
-    state.itemId ??= identity.itemId;
-    state.callId ??= identity.callId;
-    return state;
-  };
-  const resolveCompatibleToolCall = (
-    candidates: Iterable<StreamingToolCallState>,
-    identity: StreamingToolCallIdentity,
-  ): StreamingToolCallState | undefined => {
-    const uniqueCandidates = [...new Set(candidates)];
-    if (!identity.itemId && !identity.callId) {
-      return uniqueCandidates.length === 1 ? uniqueCandidates[0] : undefined;
-    }
-    const compatible = uniqueCandidates.filter((state) => !identitiesConflict(state, identity));
-    const matches = compatible.filter((state) => sharesIdentity(state, identity));
-    if (matches.length === 1) {
-      const match = matches.at(0);
-      return match ? adoptToolCallIdentity(match, identity) : undefined;
-    }
-    // Only a sole active call may adopt an identity it did not already know.
-    // Parallel calls require a positive match so missing indices stay fail-closed.
-    if (uniqueCandidates.length !== 1 || compatible.length !== 1 || matches.length !== 0) {
-      return undefined;
-    }
-    const candidate = compatible.at(0);
-    return candidate ? adoptToolCallIdentity(candidate, identity) : undefined;
-  };
-  const resolveStreamingToolCall = (
-    event: Record<string, unknown>,
-    identity: StreamingToolCallIdentity = readEventToolCallIdentity(event),
-  ): StreamingToolCallState | undefined => {
-    const outputIndex = readOutputIndex(event);
-    if (outputIndex !== undefined) {
-      const indexed = toolCallsByOutputIndex.get(outputIndex);
-      if (indexed) {
-        if (indexed.callId && identity.callId && indexed.callId !== identity.callId) {
-          return undefined;
-        }
-        // output_index owns routing once registered, but call_id stays stable;
-        // compatible providers may rotate item_id for the same output item.
-        return adoptToolCallIdentity(indexed, identity);
-      }
-      // A compatibility stream may add calls without indices, then start
-      // including them. Bind only the one identity-matched (or sole) candidate.
-      const unindexed = resolveCompatibleToolCall(unindexedToolCalls, identity);
-      if (unindexed) {
-        unindexedToolCalls.delete(unindexed);
-        toolCallsByOutputIndex.set(outputIndex, unindexed);
-      }
-      return unindexed;
-    }
-
-    return resolveCompatibleToolCall(
-      [...toolCallsByOutputIndex.values(), ...unindexedToolCalls],
-      identity,
-    );
-  };
-  const forgetStreamingToolCall = (toolCall: StreamingToolCallState) => {
-    for (const [trackedIndex, tracked] of toolCallsByOutputIndex) {
-      if (tracked === toolCall) {
-        toolCallsByOutputIndex.delete(trackedIndex);
-      }
-    }
-    unindexedToolCalls.delete(toolCall);
-  };
-  const markActiveToolCallArgumentsUnreliable = () => {
-    // An unrouteable argument event may belong to any active call. Only an
-    // authoritative full argument snapshot can recover that call.
-    for (const toolCall of new Set([...toolCallsByOutputIndex.values(), ...unindexedToolCalls])) {
-      toolCall.argumentStreamReliable = false;
-    }
-  };
-  const hasActiveStreamingToolCall = () =>
-    toolCallsByOutputIndex.size > 0 || unindexedToolCalls.size > 0;
  // Opening fragments may carry the only function name. A conflicting
  // completion must never retarget an already-started call.
  const resolveCompletedToolCallName = (
@@ -1553,31 +1447,24 @@ async function processResponsesStream(
          stream.push({ type: "text_start", contentIndex: blockIndex(), partial: output });
        }
      } else if (item.type === "function_call") {
-        const outputIndex = readOutputIndex(event);
-        if (outputIndex !== undefined && toolCallsByOutputIndex.has(outputIndex)) {
-          throw new Error(`Responses stream reused active tool-call output index ${outputIndex}`);
-        }
-        currentItem = item;
-        currentBlock = {
+        const toolCallBlock: Record<string, unknown> = {
          type: "toolCall",
          id: resolveToolCallId(item),
          name: readIdentityValue(item.name) ?? "",
          arguments: {},
          partialJson: stringifyJsonLike(item.arguments),
        };
-        output.content.push(currentBlock);
-        const contentIndex = blockIndex();
-        const toolCallState = {
-          block: currentBlock,
+        const contentIndex = output.content.length;
+        const toolCallState: StreamingToolCallState = {
+          block: toolCallBlock,
          contentIndex,
          argumentStreamReliable: true,
-          ...readItemToolCallIdentity(item),
+          ...readResponsesToolCallItemIdentity(item),
        };
-        if (outputIndex !== undefined) {
-          toolCallsByOutputIndex.set(outputIndex, toolCallState);
-        } else {
-          unindexedToolCalls.add(toolCallState);
-        }
+        streamingToolCalls.register(event, toolCallState);
+        currentItem = item;
+        currentBlock = toolCallBlock;
+        output.content.push(toolCallBlock);
        stream.push({ type: "toolcall_start", contentIndex, partial: output });
      }
    } else if (type === "response.reasoning_summary_text.delta") {
@@ -1604,7 +1491,7 @@ async function processResponsesStream(
        }
      }
    } else if (type === "response.function_call_arguments.delta") {
-      const toolCall = resolveStreamingToolCall(event);
+      const toolCall = streamingToolCalls.resolve(event);
      if (toolCall) {
        toolCall.block.partialJson = `${stringifyJsonLike(toolCall.block.partialJson)}${stringifyJsonLike(event.delta)}`;
        toolCall.block.arguments = parseStreamingJson(
@@ -1616,11 +1503,11 @@ async function processResponsesStream(
          delta: stringifyJsonLike(event.delta),
          partial: output,
        });
-      } else if (hasActiveStreamingToolCall()) {
-        markActiveToolCallArgumentsUnreliable();
+      } else if (streamingToolCalls.hasActive()) {
+        streamingToolCalls.markArgumentsUnreliable();
      }
    } else if (type === "response.function_call_arguments.done") {
-      const toolCall = resolveStreamingToolCall(event);
+      const toolCall = streamingToolCalls.resolve(event);
      if (toolCall) {
        const previousPartialJson = stringifyJsonLike(toolCall.block.partialJson);
        const doneArguments = typeof event.arguments === "string" ? event.arguments : undefined;
@@ -1643,8 +1530,8 @@ async function processResponsesStream(
            });
          }
        }
-      } else if (hasActiveStreamingToolCall()) {
-        markActiveToolCallArgumentsUnreliable();
+      } else if (streamingToolCalls.hasActive()) {
+        streamingToolCalls.markArgumentsUnreliable();
      }
    } else if (type === "response.output_item.done") {
      const item = event.item as Record<string, unknown>;
@@ -1745,10 +1632,13 @@ async function processResponsesStream(
        }
        currentBlock = null;
      } else if (item.type === "function_call") {
-        const streamingToolCall = resolveStreamingToolCall(event, readItemToolCallIdentity(item));
+        const streamingToolCall = streamingToolCalls.resolve(
+          event,
+          readResponsesToolCallItemIdentity(item),
+        );
        // Do not turn an unresolved completion into a second public call while
        // an indexed call is still open. Its identity or index must match.
-        if (!streamingToolCall && hasActiveStreamingToolCall()) {
+        if (!streamingToolCall && streamingToolCalls.hasActive()) {
          await cooperativeScheduler.afterEvent();
          continue;
        }
@@ -1802,7 +1692,7 @@ async function processResponsesStream(
          partial: output,
        });
        if (streamingToolCall) {
-          forgetStreamingToolCall(streamingToolCall);
+          streamingToolCalls.forget(streamingToolCall);
        }
        if (currentBlock === toolCallBlock) {
          currentBlock = null;
@@ -1810,7 +1700,7 @@ async function processResponsesStream(
        }
      }
    } else if (type === "response.completed") {
-      if (hasActiveStreamingToolCall()) {
+      if (streamingToolCalls.hasActive()) {
        throw new Error("Responses stream completed with unresolved tool calls");
      }
      const response = event.response as Record<string, unknown> | undefined;
@@ -1879,7 +1769,7 @@ async function processResponsesStream(
    }
    await cooperativeScheduler.afterEvent();
  }
-  if (hasActiveStreamingToolCall()) {
+  if (streamingToolCalls.hasActive()) {
    throw new Error("Responses stream ended with unresolved tool calls");
  }
  const eventTypeSummary = [...eventTypes.entries()]