diff --git a/changelog.d/fixes/9314-kiro-nested-tool-call-validation.md b/changelog.d/fixes/9314-kiro-nested-tool-call-validation.md new file mode 100644 index 0000000000..22c6ef5665 --- /dev/null +++ b/changelog.d/fixes/9314-kiro-nested-tool-call-validation.md @@ -0,0 +1 @@ +- **fix(kiro):** validate completed nested tool-call payloads before forwarding them. (thanks @SemonCat) diff --git a/open-sse/executors/kiro.ts b/open-sse/executors/kiro.ts index eb0a93eacc..65c4e6186a 100644 --- a/open-sse/executors/kiro.ts +++ b/open-sse/executors/kiro.ts @@ -21,6 +21,18 @@ import { } from "./kiroThinking.ts"; import { ByteQueue, TEXT_ENCODER, parseEventFrame } from "./kiro/eventstream.ts"; import { kiroRuntimeHost, resolveKiroRuntimeRegion } from "../services/kiroRegion.ts"; +import { + KIRO_TOOL_CALL_WRAPPER, + appendBufferedKiroToolInput, + encodeSse, + getBufferedKiroToolInput, + validateKiroToolCallWrapperInput, + validateKiroToolName, + validateKiroToolUse, + type PendingKiroWrapperToolCall, +} from "./kiroToolCallValidation.ts"; + +export { validateKiroToolUse } from "./kiroToolCallValidation.ts"; type JsonRecord = Record; @@ -42,6 +54,9 @@ type KiroStreamState = { seenToolIds: Map; toolArgsEmitted: Map; toolArgsBuffered: Map; + generatedToolIdCounter: number; + pendingWrapperToolCalls: Map; + invalidToolCall?: boolean; totalContentLength?: number; contextUsagePercentage?: number; hasContextUsage?: boolean; @@ -396,11 +411,116 @@ export class KiroExecutor extends BaseExecutor { seenToolIds: new Map(), toolArgsEmitted: new Map(), toolArgsBuffered: new Map(), + generatedToolIdCounter: 0, + pendingWrapperToolCalls: new Map(), hasReasoningContent: false, reasoningChunkCount: 0, thinking: thinkingExpected ? { thinkingMode: false, pendingTag: "" } : undefined, }; + const getToolCallId = (toolUse: JsonRecord): string => { + if (typeof toolUse.toolUseId === "string" && toolUse.toolUseId) { + return toolUse.toolUseId; + } + state.generatedToolIdCounter += 1; + return `call_${created}_${state.generatedToolIdCounter}`; + }; + + const emitToolCallStart = ( + controller: TransformStreamDefaultController, + toolCallId: string, + toolName: string, + toolIndex: number + ) => { + const startChunk: JsonRecord = { + id: responseId, + object: "chat.completion.chunk", + created, + model, + choices: [ + { + index: 0, + delta: { + ...(chunkIndex === 0 ? { role: "assistant" } : {}), + tool_calls: [ + { + index: toolIndex, + id: toolCallId, + type: "function", + function: { name: toolName, arguments: "" }, + }, + ], + }, + finish_reason: null, + }, + ], + }; + chunkIndex += 1; + controller.enqueue(encodeSse(`data: ${JSON.stringify(startChunk)}\n\n`)); + }; + + const emitToolCallArguments = ( + controller: TransformStreamDefaultController, + toolIndex: number, + argumentsStr: string + ) => { + const argsChunk: JsonRecord = { + id: responseId, + object: "chat.completion.chunk", + created, + model, + choices: [ + { + index: 0, + delta: { + tool_calls: [{ index: toolIndex, function: { arguments: argumentsStr } }], + }, + finish_reason: null, + }, + ], + }; + chunkIndex += 1; + controller.enqueue(encodeSse(`data: ${JSON.stringify(argsChunk)}\n\n`)); + }; + + const failInvalidToolCall = (controller: TransformStreamDefaultController, message: string) => { + const error = { + error: { + message, + type: "invalid_request_error", + code: "invalid_kiro_tool_call", + }, + }; + state.invalidToolCall = true; + state.finishEmitted = true; + controller.enqueue(encodeSse(`data: ${JSON.stringify(error)}\n\n`)); + controller.enqueue(encodeSse("data: [DONE]\n\n")); + controller.terminate(); + }; + + const flushPendingWrapperToolCalls = ( + controller: TransformStreamDefaultController + ): boolean => { + for (const toolCall of state.pendingWrapperToolCalls.values()) { + const toolInput = getBufferedKiroToolInput(toolCall); + try { + validateKiroToolCallWrapperInput(toolInput); + } catch (error) { + failInvalidToolCall(controller, error instanceof Error ? error.message : String(error)); + return false; + } + + const toolIndex = state.toolCallIndex++; + state.seenToolIds.set(toolCall.toolCallId, toolIndex); + emitToolCallStart(controller, toolCall.toolCallId, toolCall.toolName, toolIndex); + const argumentsStr = + typeof toolInput === "string" ? toolInput : JSON.stringify(toolInput ?? {}); + if (argumentsStr) emitToolCallArguments(controller, toolIndex, argumentsStr); + } + state.pendingWrapperToolCalls.clear(); + return true; + }; + const transformStream = new TransformStream( { async transform(chunk, controller) { @@ -618,50 +738,64 @@ export class KiroExecutor extends BaseExecutor { const toolUse = event.payload; const toolUses = Array.isArray(toolUse) ? toolUse : [toolUse]; - for (const singleToolUse of toolUses) { - const toolCallId = singleToolUse.toolUseId || `call_${Date.now()}`; - const toolName = singleToolUse.name || ""; + for (const rawToolUse of toolUses) { + const singleToolUse = rawToolUse as JsonRecord; + let toolName: string; + try { + toolName = validateKiroToolName(singleToolUse); + } catch (error) { + failInvalidToolCall( + controller, + error instanceof Error ? error.message : String(error) + ); + return; + } + + const toolCallId = getToolCallId(singleToolUse); const toolInput = singleToolUse.input; + if (toolName === KIRO_TOOL_CALL_WRAPPER) { + let pending = state.pendingWrapperToolCalls.get(toolCallId); + if (!pending) { + if (state.seenToolIds.has(toolCallId)) { + failInvalidToolCall( + controller, + "Invalid Kiro tool_call payload: duplicate toolUseId reused by wrapper" + ); + return; + } + pending = { toolCallId, toolName }; + state.pendingWrapperToolCalls.set(toolCallId, pending); + } + try { + appendBufferedKiroToolInput(pending, toolInput); + } catch (error) { + failInvalidToolCall( + controller, + error instanceof Error ? error.message : String(error) + ); + return; + } + continue; + } + + if (state.pendingWrapperToolCalls.has(toolCallId)) { + failInvalidToolCall( + controller, + "Invalid Kiro tool_call payload: mixed wrapper and direct tool fragments" + ); + return; + } + let toolIndex; const isNewTool = !state.seenToolIds.has(toolCallId); if (isNewTool) { toolIndex = state.toolCallIndex++; state.seenToolIds.set(toolCallId, toolIndex); - - const startChunk = { - id: responseId, - object: "chat.completion.chunk", - created, - model, - choices: [ - { - index: 0, - delta: { - ...(chunkIndex === 0 ? { role: "assistant" } : {}), - tool_calls: [ - { - index: toolIndex, - id: toolCallId, - type: "function", - function: { - name: toolName, - arguments: "", - }, - }, - ], - }, - finish_reason: null, - }, - ], - }; - chunkIndex++; - controller.enqueue( - TEXT_ENCODER.encode(`data: ${JSON.stringify(startChunk)}\n\n`) - ); + emitToolCallStart(controller, toolCallId, toolName, toolIndex); } else { - toolIndex = state.seenToolIds.get(toolCallId); + toolIndex = state.seenToolIds.get(toolCallId) as number; } if (toolInput !== undefined) { @@ -714,6 +848,7 @@ export class KiroExecutor extends BaseExecutor { // Handle messageStopEvent if (eventType === "messageStopEvent") { + if (!flushPendingWrapperToolCalls(controller)) return; flushBufferedToolArgs(state, controller, { responseId, created, model }); state.stopSeen = true; } @@ -819,6 +954,8 @@ export class KiroExecutor extends BaseExecutor { }, flush(controller) { + if (!flushPendingWrapperToolCalls(controller)) return; + if (state.invalidToolCall) return; // Flush any buffered tool arguments (partial-object payloads) before finishing — // idempotent against toolArgsEmitted if messageStopEvent already flushed them. flushBufferedToolArgs(state, controller, { responseId, created, model }); diff --git a/open-sse/executors/kiroToolCallValidation.ts b/open-sse/executors/kiroToolCallValidation.ts new file mode 100644 index 0000000000..d447a427f0 --- /dev/null +++ b/open-sse/executors/kiroToolCallValidation.ts @@ -0,0 +1,94 @@ +import { TEXT_ENCODER } from "./kiro/eventstream.ts"; + +/** + * Validation + buffering helpers for Kiro's nested `tool_call` wrapper payloads. + * + * Extracted from kiro.ts (file-size gate, #9314) — pure functions, no dependency on + * KiroExecutor instance state. + */ + +export type JsonRecord = Record; + +export const KIRO_TOOL_CALL_WRAPPER = "tool_call"; + +export type PendingKiroWrapperToolCall = { + toolCallId: string; + toolName: string; + inputKind?: "string" | "object"; + inputText?: string; + inputObject?: Record; +}; + +export function parseKiroToolInput(toolInput: unknown): unknown { + if (typeof toolInput !== "string") return toolInput; + try { + return JSON.parse(toolInput); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + throw new Error(`Invalid Kiro tool_call payload: input must be valid JSON (${message})`); + } +} + +export function validateKiroToolName(toolUse: JsonRecord): string { + const toolName = typeof toolUse.name === "string" ? toolUse.name.trim() : ""; + if (!toolName) throw new Error("Invalid Kiro toolUseEvent: missing tool name"); + return toolName; +} + +export function validateKiroToolCallWrapperInput(toolInput: unknown): void { + if (toolInput === undefined) { + throw new Error("Invalid Kiro tool_call payload: missing input"); + } + const input = parseKiroToolInput(toolInput); + if (!input || typeof input !== "object" || Array.isArray(input)) { + throw new Error( + "Invalid Kiro tool_call payload: input must be an object with name and arguments" + ); + } + const record = input as JsonRecord; + if (typeof record.name !== "string" || !record.name.trim()) { + throw new Error("Invalid Kiro tool_call payload: missing nested MCP tool name at input.name"); + } + if (!Object.prototype.hasOwnProperty.call(record, "arguments")) { + throw new Error( + "Invalid Kiro tool_call payload: missing nested MCP tool arguments at input.arguments" + ); + } +} + +export function validateKiroToolUse(toolUse: JsonRecord): void { + const toolName = validateKiroToolName(toolUse); + if (toolName === KIRO_TOOL_CALL_WRAPPER) { + validateKiroToolCallWrapperInput(toolUse.input); + } +} + +export function appendBufferedKiroToolInput( + toolCall: PendingKiroWrapperToolCall, + toolInput: unknown +): void { + if (toolInput === undefined) return; + if (typeof toolInput === "string") { + if (toolCall.inputKind && toolCall.inputKind !== "string") { + throw new Error("Invalid Kiro tool_call payload: mixed input fragment types"); + } + toolCall.inputKind = "string"; + toolCall.inputText = `${toolCall.inputText || ""}${toolInput}`; + return; + } + if (toolInput && typeof toolInput === "object" && !Array.isArray(toolInput)) { + if (toolCall.inputKind && toolCall.inputKind !== "object") { + throw new Error("Invalid Kiro tool_call payload: mixed input fragment types"); + } + toolCall.inputKind = "object"; + toolCall.inputObject = toolInput as Record; + } +} + +export function getBufferedKiroToolInput(toolCall: PendingKiroWrapperToolCall): unknown { + return toolCall.inputKind === "string" ? toolCall.inputText || "" : toolCall.inputObject; +} + +export function encodeSse(value: string): Uint8Array { + return TEXT_ENCODER.encode(value); +} diff --git a/open-sse/utils/stream.ts b/open-sse/utils/stream.ts index 27182d6887..45b0d92c0c 100644 --- a/open-sse/utils/stream.ts +++ b/open-sse/utils/stream.ts @@ -41,6 +41,11 @@ import { } from "./responsesCommentaryDrop.ts"; import { buildErrorBody } from "./error.ts"; import { parseTextualToolCallCandidate, isValidToolCallHeaderPrefix } from "./textualToolCall.ts"; +import { + formatTranslatedStreamError, + normalizeStreamFailurePayload, + type StreamFailurePayload, +} from "./streamErrorFormat.ts"; import { recordToolLatency } from "../services/toolLatencyTracker.ts"; import { extractToolSchemaMap } from "../translator/response/openai-responses/toolSchemas.ts"; import { @@ -118,13 +123,6 @@ type StreamCompletePayload = { ttft?: number | null; }; -type StreamFailurePayload = { - status: number; - message: string; - code?: string; - type?: string; -}; - type StreamOptions = { mode?: string; targetFormat?: string; @@ -404,63 +402,6 @@ function toResponsesCompletedWithToolCalls(parsed: JsonRecord, toolCalls: ToolCa }; } -function toStreamFailureStatus(value: unknown): number | null { - if (typeof value === "number" && Number.isInteger(value) && value >= 400 && value <= 599) { - return value; - } - if (typeof value === "string" && /^\d{3}$/.test(value.trim())) { - const parsed = Number(value.trim()); - return parsed >= 400 && parsed <= 599 ? parsed : null; - } - return null; -} - -function looksLikeStreamRateLimit(code: string, type: string, message: string): boolean { - const haystack = `${code} ${type} ${message}`.toLowerCase(); - return ( - haystack.includes("usage_limit_reached") || - haystack.includes("rate_limit") || - haystack.includes("rate limit") || - haystack.includes("quota") || - haystack.includes("too many requests") || - haystack.includes("limit reached") || - haystack.includes("limit has been reached") - ); -} - -function normalizeStreamFailurePayload(payload: unknown): StreamFailurePayload | null { - const record = payload && typeof payload === "object" ? (payload as JsonRecord) : {}; - const response = asRecord(record.response); - const error = Object.keys(asRecord(response.error)).length - ? asRecord(response.error) - : Object.keys(asRecord(record.error)).length - ? asRecord(record.error) - : record; - const code = typeof error.code === "string" ? error.code : "upstream_error"; - const type = typeof error.type === "string" ? error.type : undefined; - const message = - typeof error.message === "string" && error.message.trim() - ? error.message - : typeof record.message === "string" && record.message.trim() - ? record.message - : "Upstream failure"; - const status = - toStreamFailureStatus(error.status_code) ?? - toStreamFailureStatus(error.status) ?? - toStreamFailureStatus(response.status_code) ?? - toStreamFailureStatus(response.status) ?? - toStreamFailureStatus(record.status_code) ?? - toStreamFailureStatus(record.status) ?? - (looksLikeStreamRateLimit(code, type || "", message) ? 429 : 502); - - return { - status, - message, - code, - ...(type ? { type } : {}), - }; -} - type ClaudeEmptyResponseLifecycle = { hasMessageStart: boolean; hasContentBlock: boolean; @@ -831,6 +772,7 @@ export function createSSEStream(options: StreamOptions = {}) { // Guard against duplicate [DONE] events — ensures exactly one per stream let doneSent = false; + let upstreamErrorForwarded = false; const providerPayloadCollector = createStructuredSSECollector({ stage: "provider_response", }); @@ -1993,6 +1935,17 @@ export function createSSEStream(options: StreamOptions = {}) { const parsed = parseSSELine(trimmed); if (!parsed) continue; + if (upstreamErrorForwarded) continue; + + if (parsed.error) { + const output = formatTranslatedStreamError(parsed, sourceFormat); + reqLogger?.appendConvertedChunk?.(output); + controller.enqueue(encoder.encode(output)); + upstreamErrorForwarded = true; + doneSent = true; + continue; + } + // #5786 — drop replayed Responses-API events (identical/lower sequence_number // re-sent on an upstream reconnect) so their deltas are not glued twice into // the translated client stream. @@ -2184,6 +2137,10 @@ export function createSSEStream(options: StreamOptions = {}) { if (streamTimedOut) { return; } + if (upstreamErrorForwarded) { + clearPendingRequestFromStream(); + return; + } try { const remaining = decoder.decode(); if (remaining) buffer += remaining; diff --git a/open-sse/utils/streamErrorFormat.ts b/open-sse/utils/streamErrorFormat.ts new file mode 100644 index 0000000000..56b747f4e4 --- /dev/null +++ b/open-sse/utils/streamErrorFormat.ts @@ -0,0 +1,115 @@ +import { FORMATS } from "../translator/formats.ts"; +import { buildErrorBody } from "./error.ts"; + +/** + * Upstream stream-failure normalization + client-format error framing. + * + * Extracted from stream.ts (file-size gate, #9314) — pure functions operating only + * on plain payload objects, no dependency on the SSE stream/controller state. + */ + +type JsonRecord = Record; + +export type StreamFailurePayload = { + status: number; + message: string; + code?: string; + type?: string; +}; + +function asRecord(value: unknown): JsonRecord { + return value && typeof value === "object" && !Array.isArray(value) ? (value as JsonRecord) : {}; +} + +function toStreamFailureStatus(value: unknown): number | null { + if (typeof value === "number" && Number.isInteger(value) && value >= 400 && value <= 599) { + return value; + } + if (typeof value === "string" && /^\d{3}$/.test(value.trim())) { + const parsed = Number(value.trim()); + return parsed >= 400 && parsed <= 599 ? parsed : null; + } + return null; +} + +function looksLikeStreamRateLimit(code: string, type: string, message: string): boolean { + const haystack = `${code} ${type} ${message}`.toLowerCase(); + return ( + haystack.includes("usage_limit_reached") || + haystack.includes("rate_limit") || + haystack.includes("rate limit") || + haystack.includes("quota") || + haystack.includes("too many requests") || + haystack.includes("limit reached") || + haystack.includes("limit has been reached") + ); +} + +export function normalizeStreamFailurePayload(payload: unknown): StreamFailurePayload | null { + const record = payload && typeof payload === "object" ? (payload as JsonRecord) : {}; + const response = asRecord(record.response); + const error = Object.keys(asRecord(response.error)).length + ? asRecord(response.error) + : Object.keys(asRecord(record.error)).length + ? asRecord(record.error) + : record; + const code = typeof error.code === "string" ? error.code : "upstream_error"; + const type = typeof error.type === "string" ? error.type : undefined; + const message = + typeof error.message === "string" && error.message.trim() + ? error.message + : typeof record.message === "string" && record.message.trim() + ? record.message + : "Upstream failure"; + const status = + toStreamFailureStatus(error.status_code) ?? + toStreamFailureStatus(error.status) ?? + toStreamFailureStatus(response.status_code) ?? + toStreamFailureStatus(response.status) ?? + toStreamFailureStatus(record.status_code) ?? + toStreamFailureStatus(record.status) ?? + (looksLikeStreamRateLimit(code, type || "", message) ? 429 : 502); + + return { + status, + message, + code, + ...(type ? { type } : {}), + }; +} + +export function formatTranslatedStreamError(payload: unknown, sourceFormat?: string): string { + const failure = normalizeStreamFailurePayload(payload) ?? { + status: 502, + message: "Upstream stream error", + code: "stream_error", + type: "server_error", + }; + const errorBody = buildErrorBody(failure.status, failure.message, undefined, { + type: failure.type ?? "server_error", + code: failure.code ?? "stream_error", + }); + + if (sourceFormat === FORMATS.OPENAI_RESPONSES) { + const failed = { + type: "response.failed", + response: { + id: `resp_error_${Date.now()}`, + object: "response", + created_at: Math.floor(Date.now() / 1000), + status: "failed", + background: false, + error: errorBody.error, + output: [], + }, + sequence_number: 0, + }; + return `event: response.failed\ndata: ${JSON.stringify(failed)}\n\n`; + } + + if (sourceFormat === FORMATS.CLAUDE) { + return `event: error\ndata: ${JSON.stringify({ type: "error", error: errorBody.error })}\n\n`; + } + + return `data: ${JSON.stringify(errorBody)}\n\ndata: [DONE]\n\n`; +} diff --git a/tests/unit/kiro-tool-call-validation.test.ts b/tests/unit/kiro-tool-call-validation.test.ts new file mode 100644 index 0000000000..f0e7c47f7f --- /dev/null +++ b/tests/unit/kiro-tool-call-validation.test.ts @@ -0,0 +1,260 @@ +import test from "node:test"; +import assert from "node:assert/strict"; + +import { KiroExecutor } from "../../open-sse/executors/kiro.ts"; +import { FORMATS } from "../../open-sse/translator/formats.ts"; +import { createSSETransformStreamWithLogger } from "../../open-sse/utils/stream.ts"; + +const textEncoder = new TextEncoder(); + +function crc32(bytes: Uint8Array): number { + const table = new Uint32Array(256); + for (let i = 0; i < 256; i++) { + let value = i; + for (let bit = 0; bit < 8; bit++) { + value = value & 1 ? 0xedb88320 ^ (value >>> 1) : value >>> 1; + } + table[i] = value >>> 0; + } + + let value = 0xffffffff; + for (const byte of bytes) { + value = table[(value ^ byte) & 0xff] ^ (value >>> 8); + } + return (value ^ 0xffffffff) >>> 0; +} + +function encodeHeader(name: string, value: string): Uint8Array { + const nameBytes = textEncoder.encode(name); + const valueBytes = textEncoder.encode(value); + const header = new Uint8Array(1 + nameBytes.length + 1 + 2 + valueBytes.length); + let offset = 0; + header[offset++] = nameBytes.length; + header.set(nameBytes, offset); + offset += nameBytes.length; + header[offset++] = 7; + header[offset++] = (valueBytes.length >> 8) & 0xff; + header[offset++] = valueBytes.length & 0xff; + header.set(valueBytes, offset); + return header; +} + +function encodeEventFrame(eventType: string, payload: Record): Uint8Array { + const headers = encodeHeader(":event-type", eventType); + const payloadBytes = textEncoder.encode(JSON.stringify(payload)); + const totalLength = 12 + headers.length + payloadBytes.length + 4; + const frame = new Uint8Array(totalLength); + const view = new DataView(frame.buffer); + view.setUint32(0, totalLength, false); + view.setUint32(4, headers.length, false); + view.setUint32(8, crc32(frame.slice(0, 8)), false); + frame.set(headers, 12); + frame.set(payloadBytes, 12 + headers.length); + view.setUint32(totalLength - 4, crc32(frame.slice(0, totalLength - 4)), false); + return frame; +} + +function buildEventStreamResponse(frames: Uint8Array[]): Response { + return new Response( + new ReadableStream({ + start(controller) { + for (const frame of frames) controller.enqueue(frame); + controller.close(); + }, + }), + { status: 200 } + ); +} + +function collectSseJson(text: string): Array> { + return text + .split("\n") + .filter((line) => line.startsWith("data: ")) + .map((line) => line.slice(6).trim()) + .filter((line) => line && line !== "[DONE]") + .map((line) => JSON.parse(line) as Record); +} + +function toolDeltas(text: string): Array> { + return collectSseJson(text).flatMap((chunk) => { + const choices = Array.isArray(chunk.choices) ? chunk.choices : []; + const choice = choices[0] as Record | undefined; + const delta = choice?.delta as Record | undefined; + return Array.isArray(delta?.tool_calls) + ? (delta.tool_calls as Array>) + : []; + }); +} + +test("Kiro rejects a completed malformed tool_call wrapper without emitting a fake function", async () => { + const executor = new KiroExecutor(); + const response = buildEventStreamResponse([ + encodeEventFrame("toolUseEvent", { + toolUseId: "wrapper_1", + name: "tool_call", + input: { arguments: { query: "router" } }, + }), + encodeEventFrame("messageStopEvent", {}), + ]); + + const text = await executor.transformEventStreamToSSE(response, "kiro-model").text(); + + assert.match(text, /invalid_kiro_tool_call/); + assert.match(text, /missing nested MCP tool name/); + assert.doesNotMatch(text, /"name":"tool_call"/); + assert.match(text, /data: \[DONE\]/); +}); + +test("Kiro validates a wrapper only after string fragments are complete", async () => { + const executor = new KiroExecutor(); + const response = buildEventStreamResponse([ + encodeEventFrame("toolUseEvent", { toolUseId: "wrapper_1", name: "tool_call" }), + encodeEventFrame("toolUseEvent", { + toolUseId: "wrapper_1", + name: "tool_call", + input: '{"name":"mcp_search",', + }), + encodeEventFrame("toolUseEvent", { + toolUseId: "wrapper_1", + name: "tool_call", + input: '"arguments":{"query":"router"}}', + }), + encodeEventFrame("messageStopEvent", {}), + ]); + + const text = await executor.transformEventStreamToSSE(response, "kiro-model").text(); + const deltas = toolDeltas(text); + const args = deltas + .map((delta) => { + const fn = delta.function as Record | undefined; + return typeof fn?.arguments === "string" ? fn.arguments : ""; + }) + .join(""); + + assert.doesNotMatch(text, /invalid_kiro_tool_call/); + assert.equal((deltas[0].function as Record).name, "tool_call"); + assert.deepEqual(JSON.parse(args), { name: "mcp_search", arguments: { query: "router" } }); +}); + +test("Kiro waits for the final growing object before validating a wrapper", async () => { + const executor = new KiroExecutor(); + const response = buildEventStreamResponse([ + encodeEventFrame("toolUseEvent", { + toolUseId: "wrapper_1", + name: "tool_call", + input: { arguments: { query: "router" } }, + }), + encodeEventFrame("toolUseEvent", { + toolUseId: "wrapper_1", + name: "tool_call", + input: { name: "mcp_search", arguments: { query: "router" } }, + }), + encodeEventFrame("messageStopEvent", {}), + ]); + + const text = await executor.transformEventStreamToSSE(response, "kiro-model").text(); + const deltas = toolDeltas(text); + const args = deltas + .map((delta) => { + const fn = delta.function as Record | undefined; + return typeof fn?.arguments === "string" ? fn.arguments : ""; + }) + .join(""); + + assert.doesNotMatch(text, /invalid_kiro_tool_call/); + assert.deepEqual(JSON.parse(args), { name: "mcp_search", arguments: { query: "router" } }); +}); + +test("Kiro assigns direct tools before buffered wrappers when interleaved", async () => { + const executor = new KiroExecutor(); + const response = buildEventStreamResponse([ + encodeEventFrame("toolUseEvent", { + toolUseId: "wrapper_1", + name: "tool_call", + input: { name: "mcp_search", arguments: { query: "router" } }, + }), + encodeEventFrame("toolUseEvent", { + toolUseId: "direct_1", + name: "read_file", + input: { path: "README.md" }, + }), + encodeEventFrame("messageStopEvent", {}), + ]); + + const text = await executor.transformEventStreamToSSE(response, "kiro-model").text(); + const starts = toolDeltas(text).filter((delta) => typeof delta.id === "string"); + + assert.deepEqual( + starts.map((delta) => (delta.function as Record).name), + ["read_file", "tool_call"] + ); + assert.deepEqual( + starts.map((delta) => delta.index), + [0, 1] + ); +}); + +test("Kiro cancels the upstream body after an invalid wrapper", async () => { + const executor = new KiroExecutor(); + let cancelled = false; + let resolveCancelled: (() => void) | undefined; + const cancelledPromise = new Promise((resolve) => { + resolveCancelled = resolve; + }); + const response = new Response( + new ReadableStream({ + start(controller) { + controller.enqueue( + encodeEventFrame("toolUseEvent", { + toolUseId: "wrapper_1", + name: "tool_call", + input: { arguments: { query: "router" } }, + }) + ); + controller.enqueue(encodeEventFrame("messageStopEvent", {})); + }, + cancel() { + cancelled = true; + resolveCancelled?.(); + }, + }) + ); + + const textPromise = executor.transformEventStreamToSSE(response, "kiro-model").text(); + await Promise.race([cancelledPromise, new Promise((resolve) => setTimeout(resolve, 250))]); + await textPromise; + + assert.equal(cancelled, true); +}); + +test("Kiro stream errors become Responses response.failed events", async () => { + const transform = createSSETransformStreamWithLogger( + FORMATS.KIRO, + FORMATS.OPENAI_RESPONSES, + "kiro", + null, + null, + "kiro-model" + ); + const writer = transform.writable.getWriter(); + const responseText = new Response(transform.readable).text(); + + await writer.write( + textEncoder.encode( + `data: ${JSON.stringify({ + error: { + message: "Invalid Kiro tool_call payload: missing nested MCP tool name at input.name", + type: "invalid_request_error", + code: "invalid_kiro_tool_call", + }, + })}\n\n` + ) + ); + await writer.close(); + const text = await responseText; + + assert.match(text, /event: response\.failed/); + assert.match(text, /invalid_kiro_tool_call/); + assert.match(text, /missing nested MCP tool name/); + assert.doesNotMatch(text, /response\.output_item\.added/); +});