From 92394f5c07a9fdf7cebbebf9d5fc21348ec09ea2 Mon Sep 17 00:00:00 2001 From: can1357 Date: Wed, 1 Jul 2026 00:54:01 +0200 Subject: [PATCH] feat(ai): enabled streaming tool call argument deltas and validated identifier formats - Streamed parameter chunk deltas as `toolArgDelta` events during execution for Anthropic and DeepSeek dialects. - Integrated streaming tool call tracking in `InbandStreamProjector` to preserve native tool identifiers and partial JSON arguments. - Added message validation functions to detect and drop malformed tool calls with empty or whitespace-only IDs. - Introduced comprehensive suite of integration and unit tests validating argument delta streaming and tool-call sanitization. --- packages/ai/CHANGELOG.md | 10 ++ packages/ai/src/dialect/anthropic.ts | 26 ++- packages/ai/src/dialect/deepseek.ts | 24 ++- packages/ai/src/dialect/owned-stream.ts | 75 ++++++-- .../ai/src/providers/transform-messages.ts | 39 +++-- packages/ai/test/inband-tools.test.ts | 93 ++++++++++ .../test/owned-stream-native-toolcall.test.ts | 163 ++++++++++++++++++ ...form-messages-malformed-tool-calls.test.ts | 40 +++++ 8 files changed, 428 insertions(+), 42 deletions(-) diff --git a/packages/ai/CHANGELOG.md b/packages/ai/CHANGELOG.md index 1df8f2fdd..d62cba166 100644 --- a/packages/ai/CHANGELOG.md +++ b/packages/ai/CHANGELOG.md @@ -2,8 +2,18 @@ ## [Unreleased] +### Added + +- Added streaming support for keyed parameter argument deltas in XML-family in-band tool call scanners (Anthropic, DeepSeek, XML, Minimax) + +### Changed + +- Improved native tool-call passthrough in `wrapInbandToolStream` to accurately mirror live streaming IDs, arguments, and partial JSON states from the underlying provider + ### Fixed +- Fixed a bug where tool calls with empty or missing IDs were not detected as malformed, causing API validation failures (e.g., 400 errors with Anthropic) on subsequent requests + - Raised Gemini header runaway threshold to prevent premature interruption of complex reasoning loops - Fixed leaked ` ```thinking ` fences with nested language-tagged Markdown code blocks so inner fences remain inside structured thinking instead of leaking as visible reply text. diff --git a/packages/ai/src/dialect/anthropic.ts b/packages/ai/src/dialect/anthropic.ts index 2cae53bbe..195a02068 100644 --- a/packages/ai/src/dialect/anthropic.ts +++ b/packages/ai/src/dialect/anthropic.ts @@ -132,7 +132,7 @@ export class AnthropicInbandScanner implements InbandScanner { progressed = this.#consumeInvoke(final, events); break; case "parameter": - progressed = this.#consumeParameter(final); + progressed = this.#consumeParameter(final, events); break; case "thinking": progressed = this.#consumeThinking(final, events); @@ -272,7 +272,7 @@ export class AnthropicInbandScanner implements InbandScanner { return true; } - #consumeParameter(final: boolean): boolean { + #consumeParameter(final: boolean, events: InbandScanEvent[]): boolean { const tagStart = this.#buffer.indexOf("<"); if (tagStart === -1) { if (final) { @@ -280,14 +280,14 @@ export class AnthropicInbandScanner implements InbandScanner { this.#buffer = ""; return false; } - this.#appendParameterValue(this.#buffer); + this.#appendParameterValue(this.#buffer, events); this.#rawBlock += this.#buffer; this.#buffer = ""; return false; } if (tagStart > 0) { const consumed = this.#buffer.slice(0, tagStart); - this.#appendParameterValue(consumed); + this.#appendParameterValue(consumed, events); this.#rawBlock += consumed; this.#buffer = this.#buffer.slice(tagStart); return true; @@ -307,7 +307,7 @@ export class AnthropicInbandScanner implements InbandScanner { return false; } const consumed = this.#buffer[0]!; - this.#appendParameterValue(consumed); + this.#appendParameterValue(consumed, events); this.#rawBlock += consumed; this.#buffer = this.#buffer.slice(1); return true; @@ -378,10 +378,22 @@ export class AnthropicInbandScanner implements InbandScanner { this.#state = "parameter"; } - #appendParameterValue(delta: string): void { + #appendParameterValue(delta: string, events: InbandScanEvent[]): void { if (delta.length === 0) return; const remaining = MAX_PARAMETER_VALUE_LENGTH - this.#paramValue.length; - if (remaining > 0) this.#paramValue += delta.slice(0, remaining); + const accepted = remaining > 0 ? delta.slice(0, remaining) : ""; + if (accepted.length > 0) { + this.#paramValue += accepted; + if (this.#started && this.#paramName.length > 0) { + events.push({ + type: "toolArgDelta", + id: this.#id, + name: this.#name, + key: this.#paramName, + delta: accepted, + }); + } + } if (delta.length > remaining) this.#paramTruncated = true; } diff --git a/packages/ai/src/dialect/deepseek.ts b/packages/ai/src/dialect/deepseek.ts index 82f0581b5..6ea5b3fe3 100644 --- a/packages/ai/src/dialect/deepseek.ts +++ b/packages/ai/src/dialect/deepseek.ts @@ -99,6 +99,7 @@ export class DeepSeekInbandScanner implements InbandScanner { #dsmlArgs: Record = {}; #dsmlParamName = ""; #dsmlParamIsString = true; + #dsmlParamRaw = ""; #rawBlock = ""; #stripLeadingWhitespace = false; @@ -153,7 +154,7 @@ export class DeepSeekInbandScanner implements InbandScanner { if (!this.#consumeDsmlInvoke(final, events)) break; continue; } - if (!this.#consumeDsmlParam(final)) break; + if (!this.#consumeDsmlParam(final, events)) break; } if (final && this.#state === "thinking") this.#endThinking(events); if (final && this.#buffer.length === 0 && this.#rawBlock.length > 0) this.#rawBlock = ""; @@ -396,18 +397,23 @@ export class DeepSeekInbandScanner implements InbandScanner { return final; } - #consumeDsmlParam(final: boolean): boolean { + #consumeDsmlParam(final: boolean, events: InbandScanEvent[]): boolean { const close = findEarliestToken(this.#buffer, DSML_PARAMETER_CLOSE_TOKENS); if (!close) { + const hold = final ? 0 : partialSuffixOverlapAny(this.#buffer, DSML_PARAMETER_CLOSE_TOKENS); + const chunk = this.#buffer.slice(0, this.#buffer.length - hold); + this.#streamDsmlParam(chunk, events); + this.#buffer = this.#buffer.slice(this.#buffer.length - hold); if (final) this.#resetDsmlTool(); return false; } - const rawValue = this.#buffer.slice(0, close.index); - this.#dsmlArgs[this.#dsmlParamName] = coerceDsmlValue(rawValue, this.#dsmlParamIsString); - this.#rawBlock += rawValue + close.token; + this.#streamDsmlParam(this.#buffer.slice(0, close.index), events); + this.#dsmlArgs[this.#dsmlParamName] = coerceDsmlValue(this.#dsmlParamRaw, this.#dsmlParamIsString); + this.#rawBlock += close.token; this.#buffer = this.#buffer.slice(close.index + close.token.length); this.#dsmlParamName = ""; this.#dsmlParamIsString = true; + this.#dsmlParamRaw = ""; this.#state = "dsmlInvoke"; return true; } @@ -418,6 +424,13 @@ export class DeepSeekInbandScanner implements InbandScanner { events.push({ type: "toolStart", id: this.#id, name: this.#name }); } + #streamDsmlParam(chunk: string, events: InbandScanEvent[]): void { + if (chunk.length === 0) return; + this.#dsmlParamRaw += chunk; + this.#rawBlock += chunk; + events.push({ type: "toolArgDelta", id: this.#id, name: this.#name, key: this.#dsmlParamName, delta: chunk }); + } + #emitThinking(delta: string, events: InbandScanEvent[]): void { if (delta.length === 0) return; if (this.#parseThinking) { @@ -509,6 +522,7 @@ export class DeepSeekInbandScanner implements InbandScanner { this.#dsmlArgs = {}; this.#dsmlParamName = ""; this.#dsmlParamIsString = true; + this.#dsmlParamRaw = ""; this.#rawBlock = ""; } } diff --git a/packages/ai/src/dialect/owned-stream.ts b/packages/ai/src/dialect/owned-stream.ts index 73bc7d667..31c5ddd70 100644 --- a/packages/ai/src/dialect/owned-stream.ts +++ b/packages/ai/src/dialect/owned-stream.ts @@ -5,6 +5,12 @@ import type { ThinkingContent, ToolCall, } from "../types"; +import { + clearStreamingPartialJson, + getStreamingPartialJson, + type StreamingPartialJsonCarrier, + setStreamingPartialJson, +} from "../utils/block-symbols"; import { AssistantMessageEventStream } from "../utils/event-stream"; import { buildStringArgsResolver } from "./coercion"; import { createInbandScanner } from "./factory"; @@ -36,6 +42,31 @@ function firstTokenIndex(text: string, tokens: readonly string[]): number { type OpenText = { index: number } | undefined; type OpenThinking = { index: number; text: string } | undefined; +type StreamingToolCall = ToolCall & StreamingPartialJsonCarrier; + +function cloneToolCall(source: StreamingToolCall): StreamingToolCall { + const block: StreamingToolCall = { + type: "toolCall", + id: source.id, + name: source.name, + arguments: source.arguments, + ...(source.rawBlock !== undefined ? { rawBlock: source.rawBlock } : {}), + }; + const partialJson = getStreamingPartialJson(source); + if (partialJson !== undefined) setStreamingPartialJson(block, partialJson); + return block; +} + +function syncToolCall(target: StreamingToolCall, source: StreamingToolCall): void { + target.id = source.id; + target.name = source.name; + target.arguments = source.arguments; + target.rawBlock = source.rawBlock; + const partialJson = getStreamingPartialJson(source); + if (partialJson === undefined) clearStreamingPartialJson(target); + else setStreamingPartialJson(target, partialJson); +} + export function parseInbandToolMessage( message: AssistantMessage, dialect: Dialect, @@ -95,12 +126,18 @@ export function wrapInbandToolStream( // projector ignores nameless "ghost" parts and de-conflicts with the // in-band channel. const src = event.partial.content[event.contentIndex]; - projector?.nativeToolStart(event.contentIndex, src?.type === "toolCall" ? src.name : ""); + projector?.nativeToolStart(event.contentIndex, src?.type === "toolCall" ? src : undefined); break; } - case "toolcall_delta": - projector?.nativeToolDelta(event.contentIndex, event.delta); + case "toolcall_delta": { + const src = event.partial.content[event.contentIndex]; + projector?.nativeToolDelta( + event.contentIndex, + event.delta, + src?.type === "toolCall" ? src : undefined, + ); break; + } case "toolcall_end": projector?.nativeToolEnd(event.contentIndex, event.toolCall); break; @@ -138,7 +175,7 @@ class InbandStreamProjector { // `contentIndex`. `#toolChannel` records which channel produced the turn's // first real call so the other is dropped — no double-dispatch, and no // guessing from emptiness. Nameless "ghost" parts never lock a channel. - #nativeBlocks = new Map(); + #nativeBlocks = new Map(); #toolChannel: "native" | "inband" | undefined; constructor( @@ -167,27 +204,33 @@ class InbandStreamProjector { this.#partial.content.push(block); } - // Forward a native tool call's lifecycle live. `name` comes from the inner - // stream's partial (set at start for well-behaved providers). Empty `name` - // means a not-yet-identified or "ghost" call — skip until `nativeToolEnd` - // can confirm. Once the in-band channel owns the turn, native calls are - // dropped to avoid double-dispatch. - nativeToolStart(srcIndex: number, name: string): void { - if (this.#stopped || !name || this.#toolChannel === "inband") return; + // Forward a native tool call's lifecycle live. `source` comes from the inner + // stream's current partial block. When owned mode wraps a provider that still + // emits native tool calls, the projected block must mirror the provider's live + // id / args / partial-json state rather than inventing `{ id: "", arguments: + // {} }` placeholders — otherwise the UI loses streaming args and can mis-key + // the call until `toolcall_end`. + nativeToolStart(srcIndex: number, source: StreamingToolCall | undefined): void { + if (this.#stopped || !source?.name || this.#toolChannel === "inband") return; this.#toolChannel = "native"; this.#closeText(); this.#closeThinking(); - const block: ToolCall = { type: "toolCall", id: "", name, arguments: {} }; + const block = cloneToolCall(source); this.#partial.content.push(block); const index = this.#partial.content.length - 1; this.#nativeBlocks.set(srcIndex, { index, block }); if (this.#emitEvents) this.#out.push({ type: "toolcall_start", contentIndex: index, partial: this.#partial }); } - nativeToolDelta(srcIndex: number, delta: string): void { + nativeToolDelta(srcIndex: number, delta: string, source: StreamingToolCall | undefined): void { if (this.#stopped) return; - const entry = this.#nativeBlocks.get(srcIndex); + let entry = this.#nativeBlocks.get(srcIndex); + if (!entry && source?.name && this.#toolChannel !== "inband") { + this.nativeToolStart(srcIndex, source); + entry = this.#nativeBlocks.get(srcIndex); + } if (!entry) return; + if (source) syncToolCall(entry.block, source); if (this.#emitEvents) this.#out.push({ type: "toolcall_delta", contentIndex: entry.index, delta, partial: this.#partial }); } @@ -196,7 +239,7 @@ class InbandStreamProjector { if (this.#stopped) return; const entry = this.#nativeBlocks.get(srcIndex); if (entry) { - Object.assign(entry.block, toolCall); + syncToolCall(entry.block, toolCall); if (this.#emitEvents) this.#out.push({ type: "toolcall_end", @@ -214,7 +257,7 @@ class InbandStreamProjector { this.#toolChannel = "native"; this.#closeText(); this.#closeThinking(); - const block: ToolCall = { ...toolCall }; + const block = cloneToolCall(toolCall); this.#partial.content.push(block); const index = this.#partial.content.length - 1; if (this.#emitEvents) { diff --git a/packages/ai/src/providers/transform-messages.ts b/packages/ai/src/providers/transform-messages.ts index 49b623480..4ebac332a 100644 --- a/packages/ai/src/providers/transform-messages.ts +++ b/packages/ai/src/providers/transform-messages.ts @@ -126,18 +126,20 @@ function deduplicateToolCallIds( } /** - * Drop assistant `toolCall` blocks whose `name` is empty or whitespace-only, + * Drop assistant `toolCall` blocks whose `id` or `name` is empty / whitespace-only, * the `toolResult` messages they point at, and any assistant turn that has no * replayable content left. * - * Models occasionally emit `{ "name": "", "arguments": "{}" }` (observed: - * GLM-5.2 + thinking on long turns, #3458). The agent loop rejects the call - * at execution time with `Tool not found`, but the malformed block and its - * error tool-result stay in `currentContext.messages`, so every subsequent - * request replays them. Every provider validates the function name — - * Anthropic 400s on `tool_use.name` (alongside an orphan `tool_result`), - * OpenAI Chat Completions 400s on `tool_calls[i].function.name` — wedging the - * session in a 400 loop until manual `/clear`. + * Models occasionally emit malformed calls such as `{ "name": "", "arguments": "{}" }` + * (observed: GLM-5.2 + thinking on long turns, #3458) or a structurally valid + * `toolCall` whose provider/native passthrough id never materialized (`id: ""`). + * The agent loop rejects or skips these at execution time, but the malformed block + * and its error tool-result can stay in `currentContext.messages`, so every + * subsequent request replays them. Every provider validates the call shape — + * Anthropic 400s on `tool_use.name` / `tool_use.id` (alongside an orphan + * `tool_result`), OpenAI Chat Completions 400s on malformed + * `tool_calls[i].function.*` — wedging the session in a 400 loop until manual + * `/clear`. * * Run before any other transform so the rest of the pipeline never sees a * malformed call. Idempotent: a re-run on an already-sanitized list returns @@ -147,13 +149,21 @@ function isMalformedToolCallName(name: string | undefined): boolean { return !name || name.trim().length === 0; } +function isMalformedToolCallId(id: string | undefined): boolean { + return !id || id.trim().length === 0; +} + +function isMalformedToolCall(block: { id: string; name: string }): boolean { + return isMalformedToolCallId(block.id) || isMalformedToolCallName(block.name); +} + function sanitizeMalformedToolCalls(messages: Message[]): Message[] { // Fast path: skip the rewrite entirely when nothing is malformed. let hasMalformed = false; outer: for (const msg of messages) { if (msg.role !== "assistant") continue; for (const block of msg.content) { - if (block.type === "toolCall" && isMalformedToolCallName(block.name)) { + if (block.type === "toolCall" && isMalformedToolCall(block)) { hasMalformed = true; break outer; } @@ -179,7 +189,7 @@ function sanitizeMalformedToolCalls(messages: Message[]): Message[] { const filtered: AssistantMessage["content"] = []; for (const block of msg.content) { if (block.type === "toolCall") { - const malformed = isMalformedToolCallName(block.name); + const malformed = isMalformedToolCall(block); const queue = dropQueues.get(block.id); if (queue) queue.push(malformed); else dropQueues.set(block.id, [malformed]); @@ -283,9 +293,10 @@ export function transformMessages( duplicateToolCallIdSuffixPrefix = "_dup", targetCompat: Model["compat"] = model.compat, ): Message[] { - // Drop assistant `toolCall` blocks with empty/whitespace `name` (and their - // matched `toolResult` messages) before anything else looks at the history. - // Replays of these would 400 every provider — see `sanitizeMalformedToolCalls`. + // Drop assistant `toolCall` blocks with empty/whitespace `id` or `name` + // (and their matched `toolResult` messages) before anything else looks at + // the history. Replays of these would 400 every provider — see + // `sanitizeMalformedToolCalls`. messages = sanitizeMalformedToolCalls(messages); // Build a map of original tool call IDs to normalized IDs diff --git a/packages/ai/test/inband-tools.test.ts b/packages/ai/test/inband-tools.test.ts index ddb3d71c7..117d18661 100644 --- a/packages/ai/test/inband-tools.test.ts +++ b/packages/ai/test/inband-tools.test.ts @@ -94,6 +94,61 @@ function expectRawBlock(dialect: Dialect, text: string, expected: string): void expect(firstRawBlock(dialect, text), dialect).toBe(expected); } +function parameterDeltaEvents( + events: readonly InbandScanEvent[], +): Extract[] { + return events.filter((event): event is Extract => { + return event.type === "toolArgDelta"; + }); +} + +const XML_PARAMETER_STREAMS: readonly { dialect: Dialect; chunks: readonly string[] }[] = [ + { + dialect: "anthropic", + chunks: [ + '\n', + "src/", + "a.ts', + "2\n", + ], + }, + { + dialect: "xml", + chunks: [ + '\n', + "src/", + "a.ts', + "2\n", + ], + }, + { + dialect: "minimax", + chunks: [ + '\n', + "src/", + "a.ts', + "2\n", + ], + }, + { + dialect: "deepseek", + chunks: [ + '<｜DSML｜tool_calls>\n<｜DSML｜invoke name="read"><｜DSML｜parameter name="path" string="true">', + "src/", + "a.ts<｜DSML｜parameter name="count" string="false">', + "2\n", + ], + }, +]; + describe("in-band tool dialects", () => { it("renders a tool prompt for every dialect", () => { for (const dialect of DIALECTS) { @@ -122,6 +177,44 @@ describe("in-band tool dialects", () => { } }); + it("streams keyed parameter argument deltas before the final XML-family tool end", () => { + for (const { dialect, chunks } of XML_PARAMETER_STREAMS) { + const scanner = createInbandScanner(dialect, { tools: TOOLS, parseThinking: true }); + const perFeedEvents = chunks.map(chunk => scanner.feed(chunk)); + const events = perFeedEvents.flat(); + events.push(...scanner.flush()); + const starts = events.filter((event): event is Extract => { + return event.type === "toolStart"; + }); + expect(starts, dialect).toHaveLength(1); + + const callId = starts[0]!.id; + expect(starts[0], dialect).toMatchObject({ id: callId, name: "read" }); + expect(parameterDeltaEvents(perFeedEvents[1]!), dialect).toEqual([ + { type: "toolArgDelta", id: callId, name: "read", key: "path", delta: "src/" }, + ]); + expect(parameterDeltaEvents(perFeedEvents[2]!), dialect).toEqual([ + { type: "toolArgDelta", id: callId, name: "read", key: "path", delta: "a.ts" }, + ]); + expect(toolEnds(perFeedEvents[2]!), dialect).toHaveLength(0); + expect(parameterDeltaEvents(perFeedEvents[4]!), dialect).toEqual([ + { type: "toolArgDelta", id: callId, name: "read", key: "count", delta: "2" }, + ]); + + const calls = toolEnds(events); + expect(calls, dialect).toHaveLength(1); + expect(calls[0], dialect).toMatchObject({ + id: callId, + name: "read", + arguments: { path: "src/a.ts", count: 2 }, + }); + const finalIndex = events.findIndex(event => event.type === "toolEnd"); + const lastDeltaIndex = events.findLastIndex(event => event.type === "toolArgDelta"); + expect(lastDeltaIndex, dialect).toBeGreaterThan(-1); + expect(finalIndex, dialect).toBeGreaterThan(lastDeltaIndex); + } + }); + it("captures exact raw tool call blocks for debugging", () => { expectRawBlock( "glm", diff --git a/packages/ai/test/owned-stream-native-toolcall.test.ts b/packages/ai/test/owned-stream-native-toolcall.test.ts index 25ff62ab4..ec0b7f357 100644 --- a/packages/ai/test/owned-stream-native-toolcall.test.ts +++ b/packages/ai/test/owned-stream-native-toolcall.test.ts @@ -1,6 +1,7 @@ import { describe, expect, it } from "bun:test"; import { wrapInbandToolStream } from "../src/dialect/owned-stream"; import type { AssistantMessage, AssistantMessageEvent, ThinkingContent, ToolCall, Usage } from "../src/types"; +import { getStreamingPartialJson, setStreamingPartialJson } from "../src/utils/block-symbols"; import { AssistantMessageEventStream } from "../src/utils/event-stream"; const TOOLS = [ @@ -70,6 +71,44 @@ function geminiNativeOnly(): AssistantMessageEventStream { }); } +function controlledNativeToolArgGrowth(): { + stream: AssistantMessageEventStream; + pushStart: () => void; + pushFirstDelta: () => void; + pushSecondDelta: () => void; + pushEnd: () => void; + finish: () => void; +} { + const stream = new AssistantMessageEventStream(); + const out = makeAssistant([]); + const block: ToolCall = { type: "toolCall", id: "tool_todo_streaming", name: "todo", arguments: {} }; + stream.push({ type: "start", partial: out }); + + return { + stream, + pushStart: () => { + out.content.push(block); + stream.push({ type: "toolcall_start", contentIndex: 0, partial: out }); + }, + pushFirstDelta: () => { + setStreamingPartialJson(block, '{"ops":['); + stream.push({ type: "toolcall_delta", contentIndex: 0, delta: '{"ops":[', partial: out }); + }, + pushSecondDelta: () => { + block.arguments = { ops: [{ op: "view" }] }; + setStreamingPartialJson(block, '{"ops":[{"op":"view"}]}'); + stream.push({ type: "toolcall_delta", contentIndex: 0, delta: '{"op":"view"}]}', partial: out }); + }, + pushEnd: () => { + stream.push({ type: "toolcall_end", contentIndex: 0, toolCall: block, partial: out }); + }, + finish: () => { + stream.push({ type: "done", reason: "toolUse", message: out }); + stream.end(out); + }, + }; +} + // A nameless native "ghost" part (Gemini emits these beside a real call) must be // dropped, while the real native call is still forwarded. function ghostThenRealNative(): AssistantMessageEventStream { @@ -111,6 +150,60 @@ function inbandPlusNative(): AssistantMessageEventStream { }); } +function cloneArgs(args: Record): Record { + return JSON.parse(JSON.stringify(args)) as Record; +} + +type ToolCallSnapshot = { + type: "toolcall_start" | "toolcall_delta" | "toolcall_end"; + delta?: string; + id: string; + name: string; + arguments: Record; + partialJson: string | undefined; + endToolCall?: { + id: string; + name: string; + arguments: Record; + partialJson: string | undefined; + }; +}; + +function snapshotToolCallEvent(event: AssistantMessageEvent): ToolCallSnapshot | undefined { + if (event.type !== "toolcall_start" && event.type !== "toolcall_delta" && event.type !== "toolcall_end") { + return undefined; + } + const block = event.partial.content[event.contentIndex]; + if (block?.type !== "toolCall") return undefined; + return { + type: event.type, + ...(event.type === "toolcall_delta" ? { delta: event.delta } : {}), + id: block.id, + name: block.name, + arguments: cloneArgs(block.arguments), + partialJson: getStreamingPartialJson(block), + ...(event.type === "toolcall_end" + ? { + endToolCall: { + id: event.toolCall.id, + name: event.toolCall.name, + arguments: cloneArgs(event.toolCall.arguments), + partialJson: getStreamingPartialJson(event.toolCall), + }, + } + : {}), + }; +} + +async function nextToolCallSnapshot(iterator: AsyncIterator): Promise { + for (;;) { + const next = await iterator.next(); + if (next.done) throw new Error("stream ended before the next tool-call event"); + const snapshot = snapshotToolCallEvent(next.value); + if (snapshot) return snapshot; + } +} + async function collect(stream: AssistantMessageEventStream): Promise<{ message: AssistantMessage; events: string[] }> { const events: string[] = []; for await (const event of stream) events.push(event.type); @@ -118,6 +211,76 @@ async function collect(stream: AssistantMessageEventStream): Promise<{ message: } describe("wrapInbandToolStream native tool-call passthrough", () => { + it("preserves the provider call id and streamed argument state across native deltas", async () => { + const controlled = controlledNativeToolArgGrowth(); + const stream = wrapInbandToolStream(controlled.stream, TOOLS, "gemini"); + const iterator = stream[Symbol.asyncIterator](); + + const startPromise = nextToolCallSnapshot(iterator); + controlled.pushStart(); + const start = await startPromise; + + const firstDeltaPromise = nextToolCallSnapshot(iterator); + controlled.pushFirstDelta(); + const firstDelta = await firstDeltaPromise; + + const secondDeltaPromise = nextToolCallSnapshot(iterator); + controlled.pushSecondDelta(); + const secondDelta = await secondDeltaPromise; + + const endPromise = nextToolCallSnapshot(iterator); + controlled.pushEnd(); + const end = await endPromise; + + controlled.finish(); + const message = await stream.result(); + + expect([start, firstDelta, secondDelta, end]).toEqual([ + { + type: "toolcall_start", + id: "tool_todo_streaming", + name: "todo", + arguments: {}, + partialJson: undefined, + }, + { + type: "toolcall_delta", + delta: '{"ops":[', + id: "tool_todo_streaming", + name: "todo", + arguments: {}, + partialJson: '{"ops":[', + }, + { + type: "toolcall_delta", + delta: '{"op":"view"}]}', + id: "tool_todo_streaming", + name: "todo", + arguments: { ops: [{ op: "view" }] }, + partialJson: '{"ops":[{"op":"view"}]}', + }, + { + type: "toolcall_end", + id: "tool_todo_streaming", + name: "todo", + arguments: { ops: [{ op: "view" }] }, + partialJson: '{"ops":[{"op":"view"}]}', + endToolCall: { + id: "tool_todo_streaming", + name: "todo", + arguments: { ops: [{ op: "view" }] }, + partialJson: '{"ops":[{"op":"view"}]}', + }, + }, + ]); + + const calls = message.content.filter((b): b is ToolCall => b.type === "toolCall"); + expect(calls).toHaveLength(1); + expect(calls[0]!.id).toBe("tool_todo_streaming"); + expect(calls[0]!.name).toBe("todo"); + expect(calls[0]!.arguments).toEqual({ ops: [{ op: "view" }] }); + }); + it("streams a provider-native tool call that arrives without in-band text", async () => { const { message, events } = await collect(wrapInbandToolStream(geminiNativeOnly(), TOOLS, "gemini")); diff --git a/packages/ai/test/transform-messages-malformed-tool-calls.test.ts b/packages/ai/test/transform-messages-malformed-tool-calls.test.ts index f1d549bd8..5b8e867c1 100644 --- a/packages/ai/test/transform-messages-malformed-tool-calls.test.ts +++ b/packages/ai/test/transform-messages-malformed-tool-calls.test.ts @@ -103,6 +103,46 @@ describe("transformMessages drops malformed (empty-name) tool calls", () => { expect(continuation).toMatchObject({ role: "user", content: "continue" }); }); + it("treats an empty tool-call id as malformed and preserves surrounding replayable content", () => { + const messages: Message[] = [ + { role: "user", content: "Read both files", timestamp: 1 }, + assistant( + [ + { type: "text", text: "Reading the files." }, + { type: "toolCall", id: "", name: "read", arguments: { path: "bad.ts" } }, + { type: "toolCall", id: "call_keep", name: "read", arguments: { path: "good.ts" } }, + ], + 2, + ), + toolResult("", "empty-id result that Anthropic cannot match", 3), + toolResult("call_keep", "good file contents", 4), + { role: "user", content: "continue", timestamp: 5 }, + ]; + + const transformed = transformMessages(messages, model); + + const survivingCalls = getToolCalls(transformed); + expect(survivingCalls).toHaveLength(1); + expect(survivingCalls[0]).toMatchObject({ + id: "call_keep", + name: "read", + arguments: { path: "good.ts" }, + }); + + const assistantOut = transformed.find((m): m is AssistantMessage => m.role === "assistant"); + expect(assistantOut?.content).toEqual([ + { type: "text", text: "Reading the files." }, + { type: "toolCall", id: "call_keep", name: "read", arguments: { path: "good.ts" } }, + ]); + + const toolResults = transformed.filter((m): m is ToolResultMessage => m.role === "toolResult"); + expect(toolResults).toHaveLength(1); + expect(toolResults[0]).toMatchObject({ toolCallId: "call_keep", toolName: "read" }); + expect(toolResults[0]?.content).toEqual([{ type: "text", text: "good file contents" }]); + expect(transformed.find(m => m.role === "toolResult" && m.toolCallId === "")).toBeUndefined(); + expect(transformed.findLast(m => m.role === "user")).toMatchObject({ role: "user", content: "continue" }); + }); + it("drops the whole assistant turn when only the malformed tool call was emitted", () => { const messages: Message[] = [ { role: "user", content: "Help me out", timestamp: 1 },