From 468e9b1ad98f4bf48799c8eaf1d3053bd229e84d Mon Sep 17 00:00:00 2001 From: clawdbot Date: Tue, 13 Oct 2026 09:25:13 +0000 Subject: [PATCH 1/5] fix(ai): reapply Kiro content-filter stream safeguards Usage accounting and unknown stop reasons are not confirmation boundaries, and incomplete outer frames must not resync into nested metadata. Keep published partials stable and stop accepting generation frames after successful completion. Tested: bun --cwd=packages/ai run check Tested: bun test packages/ai/test/kiro-api-key-content-filter.test.ts packages/ai/test/kiro-api-key.test.ts Not-tested: live Kiro API stream Confidence: high Scope-risk: medium Reversibility: easy --- .../changelog.d/6150-kiro-content-filtered.md | 7 + packages/ai/src/providers/kiro-api-key.ts | 304 +++++-- packages/ai/src/types.ts | 1 + .../test/kiro-api-key-content-filter.test.ts | 794 +++++++++++++++++- packages/ai/test/kiro-api-key.test.ts | 8 +- 5 files changed, 1036 insertions(+), 78 deletions(-) create mode 100644 packages/ai/changelog.d/6150-kiro-content-filtered.md diff --git a/packages/ai/changelog.d/6150-kiro-content-filtered.md b/packages/ai/changelog.d/6150-kiro-content-filtered.md new file mode 100644 index 00000000000..91703554b7f --- /dev/null +++ b/packages/ai/changelog.d/6150-kiro-content-filtered.md @@ -0,0 +1,7 @@ +### Fixed + +- Kiro streams now surface explicit content-filter refusal messages with category and explanation instead of the generic "no tokens" error when a refusal is detected. Both bearer token (via `messageMetadataEvent`) and API-key (`ksk_`, via `metadataEvent`) paths are supported, with proper provider safety-stop minting. Text is streamed incrementally unless a thinking block precedes it, in which case the entire answer (after thinking) is buffered until the stream end; tool calls are emitted only at stream end (#6150). +- Detail-free `stopReason: "CONTENT_FILTERED"` (without `stopDetails.refusal` object) is now treated as a refusal and suppresses unconfirmed content. +- API-key text is released only after `COMPLETED`/`TOOL_USE` metadata or clean EOF; usage and unknown stop reasons do not confirm it. Successful-terminal text events are emitted before later transport failures. +- Parser no longer resyncs into nested objects when an incomplete outer JSON frame is encountered. Incomplete frames are retained as remainder; EOF with incomplete frame throws an error. +- Server error events now emit `text_end` for confirmed text before throwing the error, ensuring complete text lifecycle. diff --git a/packages/ai/src/providers/kiro-api-key.ts b/packages/ai/src/providers/kiro-api-key.ts index 1ac392eda68..95e7f783353 100644 --- a/packages/ai/src/providers/kiro-api-key.ts +++ b/packages/ai/src/providers/kiro-api-key.ts @@ -444,6 +444,7 @@ type KiroStreamEvent = | { type: "toolUseInput"; data: { input: string } } | { type: "toolUseStop"; data: { stop: boolean } } | { type: "usage"; data: { inputTokens?: number; outputTokens?: number } } + | { type: "completion"; data: { stopReason: "COMPLETED" | "TOOL_USE" } } | { type: "refusal"; data: { stopReason?: string; stopDetails?: { refusal?: { category?: string; explanation?: string } } }; @@ -497,8 +498,14 @@ export function parseKiroApiEvents(buffer: string): { events: KiroStreamEvent[]; const end = findJsonEnd(buffer, start); if (end < 0) { // Unclosed brace at 'start'. Could be incomplete JSON waiting for more data, or junk/corruption. - // Resync by looking for valid JSON objects within MAX_RESCAN_DISTANCE that parse successfully. const strayBraceEnd = Math.min(start + MAX_RESCAN_DISTANCE, buffer.length); + const firstPayloadCharacter = buffer.slice(start + 1).trimStart()[0]; + if (firstPayloadCharacter === undefined || firstPayloadCharacter === '"' || firstPayloadCharacter === "}") { + // A JSON object can begin only with a quoted key or close immediately. Preserve plausible + // incomplete envelopes; a complete nested child is not evidence that the outer frame is junk. + return { events, remaining: buffer.slice(start, strayBraceEnd) }; + } + // The opener cannot begin a JSON object. Resync only after establishing invalid framing. let resyncPos = start + 1; let found = false; while (resyncPos < strayBraceEnd) { @@ -525,7 +532,7 @@ export function parseKiroApiEvents(buffer: string): { events: KiroStreamEvent[]; if (!found) { // No valid JSON found within MAX_RESCAN_DISTANCE. Retain the stray brace and data up to the cap // for the next parse call. Enforce the MAX_RESCAN_DISTANCE cap to prevent unbounded buffering. - return { events, remaining: buffer.slice(start, start + MAX_RESCAN_DISTANCE) }; + return { events, remaining: buffer.slice(start, strayBraceEnd) }; } // Loop will continue with the resynced pos continue; @@ -567,12 +574,14 @@ export function parseKiroApiEvents(buffer: string): { events: KiroStreamEvent[]; // Check for refusal after usage so that usage is processed first and recorded before refusal terminates // This check is now order-independent: stopDetails can appear at any position in the JSON object + // Also handle detail-free CONTENT_FILTERED as a refusal (no stopDetails.refusal needed) if ( - typeof parsed.stopDetails === "object" && - parsed.stopDetails !== null && - (parsed.stopDetails as Record).refusal + (typeof parsed.stopDetails === "object" && + parsed.stopDetails !== null && + (parsed.stopDetails as Record).refusal) || + parsed.stopReason === "CONTENT_FILTERED" ) { - // Emit refusal event if stopDetails contains actual refusal data + // Emit refusal event if stopDetails contains actual refusal data or stopReason is CONTENT_FILTERED events.push({ type: "refusal", data: { @@ -580,7 +589,6 @@ export function parseKiroApiEvents(buffer: string): { events: KiroStreamEvent[]; stopDetails: parsed.stopDetails as { refusal?: { category?: string; explanation?: string } }, }, }); - // Normal terminal metadata (stopReason: "COMPLETED", etc.) without refusal data is ignored } else if (parsed.error || parsed.Error) { events.push({ type: "error", @@ -589,6 +597,14 @@ export function parseKiroApiEvents(buffer: string): { events: KiroStreamEvent[]; message: (parsed.message || parsed.Message) as string | undefined, }, }); + } else if (parsed.stopReason === "COMPLETED" || parsed.stopReason === "TOOL_USE") { + // Only accepted successful terminal reasons confirm buffered text. + events.push({ + type: "completion", + data: { + stopReason: parsed.stopReason, + }, + }); } } catch { // JSON parsing failed for a balanced frame: could be a junk wrapper like '{junk{...}}' @@ -801,12 +817,24 @@ export const streamKiroApiKey: StreamFunction<"kiro-codewhisperer-stream"> = ( timestamp: Date.now(), }; const blocks = output.content as Block[]; + const snapshotOutput = (): AssistantMessage => ({ + ...output, + content: output.content.map(block => ({ ...block })), + usage: { ...output.usage, cost: { ...output.usage.cost } }, + }); // Variables for streaming tool calls (initialized before try block so they're accessible in catch) let currentTool: { id: string; name: string; input: string } | undefined; let toolComplete = false; // Track whether the current tool has been completed (has stop flag) let toolcallIndex: number | undefined; const pendingToolCalls: Array<{ input: string; toolCall: ToolCall; index: number }> = []; + // Only an accepted successful completion confirms buffered text before clean EOF. + let completionSeen = false; + let textIndex: number | undefined; // Track text block index (accessible in catch) + let textContentConfirmed = false; // Track whether text has been committed to blocks (accessible in catch) + let textStartDeferred = false; // Track if text_start is deferred (accessible in catch) + let pendingTextContent = ""; // Text buffer (accessible in catch) + const pendingTextEvents: Array<{ type: "start" | "delta" | "end"; delta?: string; content?: string }> = []; // Deferred text events (accessible in catch) try { // Use validated identity snapshot from trust check if available (no second read of getters), @@ -873,13 +901,14 @@ export const streamKiroApiKey: StreamFunction<"kiro-codewhisperer-stream"> = ( } if (!response.body) throw new Error("Kiro API key response has no body"); - stream.push({ type: "start", partial: output }); + stream.push({ type: "start", partial: snapshotOutput() }); const reader = response.body.getReader(); const decoder = new TextDecoder(); let buffer = ""; let lastContent = ""; let thinkingIndex: number | undefined; - let textIndex: number | undefined; + // Note: textIndex, textContentConfirmed, textStartDeferred, pendingTextContent, pendingTextEvents + // are declared before the try block so they're accessible in the catch block // Add tool to blocks without emitting events (deferred until stream end) const addToolToBlocks = () => { @@ -938,8 +967,6 @@ export const streamKiroApiKey: StreamFunction<"kiro-codewhisperer-stream"> = ( toolComplete = false; // Reset tool completion flag }; - let textStartDeferred = false; // Track if text_start has been deferred pending thinking - const appendText = (delta: string) => { // Set firstTokenTime on first real (non-thinking) text delta if (!firstTokenEmitted) { @@ -951,51 +978,70 @@ export const streamKiroApiKey: StreamFunction<"kiro-codewhisperer-stream"> = ( } if (textIndex === undefined) { textIndex = blocks.length; + // Create block with empty text (content will be added only after confirmation) blocks.push({ type: "text", text: "", index: textIndex }); - // Defer text_start emission if thinking might come before text. - // thinkingAccumulated is non-empty means we've seen thinking in this or prior content events. - // We don't know yet if more thinking will come, so defer text_start until stream end. - if (thinkingAccumulated.length === 0) { - // No thinking yet; safe to emit text_start now - stream.push({ type: "text_start", contentIndex: textIndex, partial: output }); - } else { - // Thinking exists; defer text_start until we know thinking position - textStartDeferred = true; - } - } - const block = blocks[textIndex] as TextContent; - block.text += delta; - // Only emit text_delta if text_start was already emitted - if (!textStartDeferred) { - stream.push({ type: "text_delta", contentIndex: textIndex, delta, partial: output }); + // ALWAYS defer text_start until we have terminal metadata or reach EOF without error. + // This ensures unconfirmed text is not exposed if an error occurs. + textStartDeferred = true; } + // Buffer text content separately (don't add to blocks yet) + pendingTextContent += delta; + // Don't emit text_delta yet; defer until stream end or terminal metadata is received. + // Queue delta for later emission when confirmed. + pendingTextEvents.push({ type: "delta", delta }); + }; + + // Commit pending text to blocks (only safe at stream end when refusal is ruled out) + const commitPendingText = () => { + if (textContentConfirmed || pendingTextContent.length === 0) return; // Already committed or no text + textContentConfirmed = true; + // Note: we don't populate block.text here to avoid it appearing in previously emitted events + // The text will be populated after text_start is emitted in emitDeferredTextEvents }; // Emit deferred text events if text_start was not yet sent const emitDeferredTextEvents = () => { if (textStartDeferred && textIndex !== undefined && textIndex < blocks.length) { const block = blocks[textIndex]; - if (block && block.type === "text") { - stream.push({ type: "text_start", contentIndex: textIndex, partial: output }); - if (block.text.length > 0) { - stream.push({ - type: "text_delta", - contentIndex: textIndex, - delta: block.text, - partial: output, - }); + // Only emit text events if we have pending text content and it's been confirmed + if (block && block.type === "text" && pendingTextContent.length > 0 && textContentConfirmed) { + // Emit text_start WITHOUT populating block.text yet + // This ensures previously emitted events (like start) see the block as empty + stream.push({ type: "text_start", contentIndex: textIndex, partial: snapshotOutput() }); + // Emit all pending text_delta events WITHOUT populating block.text yet + // (The text will be populated later when text_end is emitted or at EOF) + for (const evt of pendingTextEvents) { + if (evt.type === "delta" && evt.delta) { + stream.push({ + type: "text_delta", + contentIndex: textIndex, + delta: evt.delta, + partial: snapshotOutput(), + }); + } } - textStartDeferred = false; } + textStartDeferred = false; } }; // Emit text_end if text block was opened const closeTextBlock = () => { - if (textIndex !== undefined) { + if (textIndex !== undefined && textContentConfirmed) { const block = blocks[textIndex] as TextContent; - stream.push({ type: "text_end", contentIndex: textIndex, content: block.text, partial: output }); + // Ensure text is populated before emitting text_end + if (block.text.length === 0 && pendingTextContent.length > 0) { + block.text = pendingTextContent; + } + stream.push({ + type: "text_end", + contentIndex: textIndex, + content: block.text, + partial: snapshotOutput(), + }); textIndex = undefined; + pendingTextContent = ""; + pendingTextEvents.length = 0; } }; @@ -1098,6 +1144,15 @@ export const streamKiroApiKey: StreamFunction<"kiro-codewhisperer-stream"> = ( const hasTerminalEvent = events.some(e => e.type === "refusal"); for (const event of events) { + if ( + completionSeen && + (event.type === "content" || + event.type === "toolUse" || + event.type === "toolUseInput" || + event.type === "toolUseStop") + ) { + continue; + } if (event.type === "content") { if (!hasTerminalEvent) { if (event.data === lastContent) continue; @@ -1110,7 +1165,12 @@ export const streamKiroApiKey: StreamFunction<"kiro-codewhisperer-stream"> = ( firstTokenEmitted = true; firstTokenTime = Date.now(); } - if (currentTool && currentTool.id !== event.data.toolUseId) addToolToBlocks(); + if (currentTool && currentTool.id !== event.data.toolUseId) { + // Only emit previous tool if it was completed + if (toolComplete) addToolToBlocks(); + currentTool = undefined; // Clear the incomplete tool + toolComplete = false; // Reset for new tool + } if (!currentTool) { currentTool = { id: event.data.toolUseId, name: event.data.name, input: event.data.input }; toolComplete = false; // New tool starts as incomplete @@ -1135,19 +1195,30 @@ export const streamKiroApiKey: StreamFunction<"kiro-codewhisperer-stream"> = ( if (event.data.inputTokens !== undefined) output.usage.input = event.data.inputTokens; if (event.data.outputTokens !== undefined) output.usage.output = event.data.outputTokens; output.usage.totalTokens = output.usage.input + output.usage.output; + } else if (event.type === "completion") { + // Accepted successful terminal metadata confirms text, unlike usage accounting. + completionSeen = true; + // Finding #2: Immediately commit and emit all pending text events on completion + commitPendingText(); + emitThinking(); + emitDeferredTextEvents(); + closeTextBlock(); } else if (event.type === "refusal") { const refusalData = event.data as { stopReason?: string; stopDetails?: { refusal?: { category?: string; explanation?: string } }; }; - if (refusalData.stopDetails?.refusal) { - // Handle refusal: clear blocks and emit error (don't emit any text/tool that came before) + if (refusalData.stopDetails?.refusal || refusalData.stopReason === "CONTENT_FILTERED") { + // Handle refusal: clear pending content and emit error (don't emit any text/tool that came before) consumeContent(""); // Flush any pending thinking clearPendingToolCalls(); // DROP any pending tool call without emitting events + // Clear pending text (it was never committed to blocks, so it's safe to discard) + pendingTextContent = ""; + pendingTextEvents.length = 0; - const refusal = refusalData.stopDetails.refusal; - const category = refusal.category; - const explanation = refusal.explanation?.trim(); + const refusal = refusalData.stopDetails?.refusal; + const category = refusal?.category; + const explanation = refusal?.explanation?.trim(); const label = category ? `Kiro refused the request (${category})` : "Kiro refused the request"; output.stopReason = "error"; @@ -1177,22 +1248,70 @@ export const streamKiroApiKey: StreamFunction<"kiro-codewhisperer-stream"> = ( output.duration = Date.now() - startTime; if (firstTokenTime) output.ttft = firstTokenTime - startTime; - stream.push({ type: "error", reason: "error", error: output }); + stream.push({ type: "error", reason: "error", error: output, partial: output }); stream.end(); return; } } else if (event.type === "error") { - // On ordinary errors, flush pending COMPLETED tool events before the error terminal + // On ordinary server errors (error event in stream), emit deferred text and tool events before error // (refusals drop them, incomplete tools must not be emitted) + + // For server-sent errors, emit the deferred text first (it was received before the error) + // Usage is accounting metadata, not a successful completion boundary. + if (textContentConfirmed === false && completionSeen) { + // Commit pending text to blocks + if (textIndex !== undefined && textIndex < blocks.length && pendingTextContent.length > 0) { + textContentConfirmed = true; + } + // Emit deferred text start if not yet emitted + if (textStartDeferred && textIndex !== undefined && textIndex < blocks.length) { + const block = blocks[textIndex]; + if (block && block.type === "text" && pendingTextContent.length > 0) { + stream.push({ type: "text_start", contentIndex: textIndex, partial: snapshotOutput() }); + // Emit all pending text_delta events + for (const evt of pendingTextEvents) { + if (evt.type === "delta" && evt.delta) { + stream.push({ + type: "text_delta", + contentIndex: textIndex, + delta: evt.delta, + partial: snapshotOutput(), + }); + } + } + // Emit text_end for confirmed text before error. + block.text = pendingTextContent; + closeTextBlock(); + } + textStartDeferred = false; + } + } + // Only emit the tool if it was explicitly completed (has stop flag) if (toolComplete) addToolToBlocks(); emitPendingToolCalls(); - // Preserve any already-accumulated text in the error context - const accumulatedText = blocks - .filter((b): b is TextContent => b.type === "text") - .map(b => b.text) - .join(""); + // Remove unconfirmed text blocks from the error event + if (!textContentConfirmed && textIndex !== undefined) { + // Remove the unconfirmed text block from blocks + blocks.splice(textIndex, 1); + // Update indices of blocks after the removed one + for (let i = textIndex; i < blocks.length; i++) { + const block = blocks[i] as Block; + if (block && typeof block === "object" && "index" in block) { + block.index = i; + } + } + textIndex = undefined; + } + + // Preserve text in the error context only after successful completion. + const accumulatedText = textContentConfirmed + ? blocks + .filter((b): b is TextContent => b.type === "text") + .map(b => b.text) + .join("") + : ""; // Discard unconfirmed pending text before error const errorMsg = sanitizeKiroError(`${event.data.error}: ${event.data.message ?? ""}`, apiKey); const fullErrorMsg = accumulatedText ? `${errorMsg}\n\nPartial output: ${accumulatedText}` : errorMsg; throw new Error(fullErrorMsg); @@ -1200,8 +1319,21 @@ export const streamKiroApiKey: StreamFunction<"kiro-codewhisperer-stream"> = ( } } - // Stream is done - add any pending tool, emit accumulated thinking, then deferred text, then close text block - addToolToBlocks(); + // Check for incomplete JSON frame in leftover buffer at EOF + if (buffer.trim().length > 0) { + // Non-whitespace content remains in buffer + const openBraceIdx = buffer.indexOf("{"); + if (openBraceIdx >= 0 && findJsonEnd(buffer, openBraceIdx) < 0) { + // Incomplete JSON frame at EOF (opening brace with no matching close) + throw new Error(`Kiro API key stream truncated: incomplete JSON frame at EOF`); + } + } + + // Stream is done - commit pending text (safe since no refusal occurred), add pending tool, emit accumulated thinking, etc. + // First commit pending text to blocks since stream ended without refusal + commitPendingText(); + // Emit tool only if explicitly completed (has stop flag); incomplete tools are discarded at EOF + if (currentTool && toolComplete) addToolToBlocks(); emitThinking(); emitDeferredTextEvents(); // Now emit all tool call events in order (safe since no refusal occurred) @@ -1214,7 +1346,7 @@ export const streamKiroApiKey: StreamFunction<"kiro-codewhisperer-stream"> = ( output.errorMessage = "Kiro API key stream returned no tokens"; output.duration = Date.now() - startTime; if (firstTokenTime) output.ttft = firstTokenTime - startTime; - stream.push({ type: "error", reason: "error", error: output }); + stream.push({ type: "error", reason: "error", error: output, partial: output }); stream.end(); return; } @@ -1237,10 +1369,48 @@ export const streamKiroApiKey: StreamFunction<"kiro-codewhisperer-stream"> = ( stream.push({ type: "done", reason: output.stopReason as "stop" | "toolUse", message: output }); stream.end(); } catch (error) { - // On ordinary errors (including reader.read() throws), emit only COMPLETED tool events - // before the error terminal (same semantics as the ordinary-error flush in the event loop). - // Incomplete currentTool is NOT finalized/emitted (only completed tools in pendingToolCalls flush). - // Refusals drop them via clearPendingToolCalls, but ordinary errors preserve content consistency. + // On ordinary errors (including reader.read() throws): + // - Emit COMPLETED tool events (same semantics as in the event processing loop) + // - Emit confirmed text only after an accepted successful completion + // - Incomplete currentTool is NOT finalized/emitted (only completed tools in pendingToolCalls flush). + // - Unconfirmed text is suppressed (usage alone does not confirm it) + + // Emit confirmed text if we received an accepted successful completion. + if (completionSeen && textContentConfirmed === false) { + // Commit pending text to blocks + if (textIndex !== undefined && textIndex < blocks.length && pendingTextContent.length > 0) { + textContentConfirmed = true; + } + // Emit deferred text start if not yet emitted + if (textStartDeferred && textIndex !== undefined && textIndex < blocks.length) { + const block = blocks[textIndex]; + if (block && block.type === "text" && pendingTextContent.length > 0) { + stream.push({ type: "text_start", contentIndex: textIndex, partial: snapshotOutput() }); + // Emit all pending text_delta events + for (const evt of pendingTextEvents) { + if (evt.type === "delta" && evt.delta) { + stream.push({ + type: "text_delta", + contentIndex: textIndex, + delta: evt.delta, + partial: snapshotOutput(), + }); + } + } + block.text = pendingTextContent; + stream.push({ + type: "text_end", + contentIndex: textIndex, + content: block.text, + partial: snapshotOutput(), + }); + textIndex = undefined; + pendingTextContent = ""; + pendingTextEvents.length = 0; + } + textStartDeferred = false; + } + } // Emit all pending tool call events (inline of emitPendingToolCalls logic) for (const { input, toolCall, index } of pendingToolCalls) { @@ -1260,11 +1430,25 @@ export const streamKiroApiKey: StreamFunction<"kiro-codewhisperer-stream"> = ( } pendingToolCalls.length = 0; + // Remove unconfirmed text blocks from the error event + if (!textContentConfirmed && textIndex !== undefined) { + // Remove the unconfirmed text block from blocks + blocks.splice(textIndex, 1); + // Update indices of blocks after the removed one + for (let i = textIndex; i < blocks.length; i++) { + const block = blocks[i] as Block; + if (block && typeof block === "object" && "index" in block) { + block.index = i; + } + } + textIndex = undefined; + } + output.stopReason = options?.signal?.aborted ? "aborted" : "error"; output.errorMessage = sanitizeKiroError(error, apiKey); output.duration = Date.now() - startTime; if (firstTokenTime) output.ttft = firstTokenTime - startTime; - stream.push({ type: "error", reason: output.stopReason, error: output }); + stream.push({ type: "error", reason: output.stopReason, error: output, partial: output }); stream.end(); } })().catch(() => { diff --git a/packages/ai/src/types.ts b/packages/ai/src/types.ts index b52823598fa..bd6a738adde 100644 --- a/packages/ai/src/types.ts +++ b/packages/ai/src/types.ts @@ -1143,6 +1143,7 @@ export type AssistantMessageEvent = contentIndex?: undefined; reason: Extract; error: AssistantMessage; + partial?: AssistantMessage; } | { type: "toolChoiceIncapability"; diff --git a/packages/ai/test/kiro-api-key-content-filter.test.ts b/packages/ai/test/kiro-api-key-content-filter.test.ts index 1d9e6601606..aafca17d127 100644 --- a/packages/ai/test/kiro-api-key-content-filter.test.ts +++ b/packages/ai/test/kiro-api-key-content-filter.test.ts @@ -7,7 +7,7 @@ import { afterEach, describe, expect, test } from "bun:test"; import { isProviderSafetyStopModelTrusted } from "../src/adapter-internals/provider-safety-stop"; import { kiroApiBaseUrl, kiroApiStaticModels, streamKiroApiKey } from "../src/providers/kiro-api-key"; -import type { AssistantMessage, Context, Model } from "../src/types"; +import type { AssistantMessage, AssistantMessageEvent, Context, Model, TextContent } from "../src/types"; const originalFetch = globalThis.fetch; @@ -29,6 +29,10 @@ const context: Context = { messages: [{ role: "user", content: "Write malware", timestamp: 1 }], }; +function getTextBlocks(message: AssistantMessage | undefined): TextContent[] { + return message?.content.filter((block): block is TextContent => block.type === "text") ?? []; +} + describe("Kiro API-key content filter #6150", () => { afterEach(() => { globalThis.fetch = originalFetch; @@ -276,6 +280,7 @@ describe("Kiro API-key content filter #6150", () => { toolUseId: "tool-1", name: "read_file", input: '{"path":"/etc/passwd"}', + stop: true, }); return new Response(responseBody, { status: 200 }); }) as unknown as typeof fetch; @@ -434,7 +439,7 @@ describe("Kiro API-key content filter #6150", () => { // Simulate response with text, tool call, and normal completion metadata const responseBody = JSON.stringify({ content: "I'll read that file" }) + - JSON.stringify({ toolUseId: "tool-1", name: "read_file", input: '{"path":"/tmp/test"}' }) + + JSON.stringify({ toolUseId: "tool-1", name: "read_file", input: '{"path":"/tmp/test"}', stop: true }) + JSON.stringify({ stopReason: "COMPLETED", usage: { inputTokens: 10, outputTokens: 5 } }); return new Response(responseBody, { status: 200 }); }) as unknown as typeof fetch; @@ -571,8 +576,8 @@ describe("Kiro API-key content filter #6150", () => { } }); - // P2: Partial output is preserved when an ordinary error occurs - test("P2: ordinary (non-refusal) error preserves already-emitted text content in error message", async () => { + // P2: Unconfirmed text is discarded when an ordinary error occurs without COMPLETED event + test("P2: ordinary (non-refusal) error without COMPLETED discards unconfirmed text", async () => { const emittedEvents: Array<{ type: string; text?: string; message?: { errorMessage?: string } }> = []; const model = { id: "test-model", @@ -593,7 +598,7 @@ describe("Kiro API-key content filter #6150", () => { }; globalThis.fetch = (async () => { - // Response with partial content followed by an ordinary error (not a refusal) + // Response with partial content followed by an ordinary error (not a refusal), no COMPLETED event const responseBody = JSON.stringify({ content: "Hello, this is partial" }) + JSON.stringify({ error: "rate_limit_exceeded", message: "Too many requests" }); @@ -615,20 +620,20 @@ describe("Kiro API-key content filter #6150", () => { // Errors may be thrown; events are captured above } - // Should have emitted text_delta events before the error + // Should NOT have emitted text_delta events (no COMPLETED event, so text is unconfirmed) const textDeltaEvents = emittedEvents.filter(e => e.type === "text_delta"); - expect(textDeltaEvents.length).toBeGreaterThan(0); + expect(textDeltaEvents.length).toBe(0); - // Error event should include the partial content in the message + // Error event should NOT include the partial content in the message (unconfirmed text is discarded) const errorEvent = emittedEvents.find(e => e.type === "error"); expect(errorEvent).toBeDefined(); expect(errorEvent?.message?.errorMessage).toContain("rate_limit_exceeded"); expect(errorEvent?.message?.errorMessage).toContain("Too many requests"); - expect(errorEvent?.message?.errorMessage).toContain("Partial output"); - expect(errorEvent?.message?.errorMessage).toContain("Hello, this is partial"); + expect(errorEvent?.message?.errorMessage).not.toContain("Partial output"); + expect(errorEvent?.message?.errorMessage).not.toContain("Hello, this is partial"); }); - test("P2: tool call partial output is preserved when ordinary error occurs", async () => { + test("P2: tool call partial output is discarded when ordinary error occurs without COMPLETED event", async () => { const emittedEvents: Array<{ type: string; message?: { errorMessage?: string } }> = []; const model = { id: "test-model", @@ -649,7 +654,7 @@ describe("Kiro API-key content filter #6150", () => { }; globalThis.fetch = (async () => { - // Response with text, tool call start, and then an ordinary error + // Response with text, tool call start, and then an ordinary error (no COMPLETED event) const responseBody = JSON.stringify({ content: "I'll read that file" }) + JSON.stringify({ toolUseId: "tool-123", name: "read_file", input: "{" }) + @@ -670,17 +675,21 @@ describe("Kiro API-key content filter #6150", () => { // Errors may be thrown; events are captured above } - // Error should be reported but should mention accumulated partial output (text and tool) + // Error should be reported but should NOT mention accumulated partial output without COMPLETED const errorEvent = emittedEvents.find(e => e.type === "error"); expect(errorEvent).toBeDefined(); expect(errorEvent?.message?.errorMessage).toContain("connection_timeout"); expect(errorEvent?.message?.errorMessage).toContain("Connection lost"); - // Should mention partial output was accumulated - expect(errorEvent?.message?.errorMessage).toContain("Partial output"); + // Should NOT mention partial output (unconfirmed text is discarded) + expect(errorEvent?.message?.errorMessage).not.toContain("Partial output"); }); }); describe("reasoning-before-answer contentIndex invariant #6151", () => { + afterEach(() => { + globalThis.fetch = originalFetch; + }); + test("thinking then text — every contentIndex matches block final position", async () => { const emittedEvents: Array<{ type: string; @@ -778,7 +787,8 @@ describe("reasoning-before-answer contentIndex invariant #6151", () => { const responseBody = JSON.stringify({ content: "I need to read a fileLet me read that file for you.", - }) + JSON.stringify({ toolUseId: "tool-123", name: "read_file", input: '{"path":"/tmp/test.txt"}' }); + }) + + JSON.stringify({ toolUseId: "tool-123", name: "read_file", input: '{"path":"/tmp/test.txt"}', stop: true }); return new Response(responseBody, { status: 200 }); }) as unknown as typeof fetch; @@ -916,7 +926,7 @@ describe("reasoning-before-answer contentIndex invariant #6151", () => { for await (const event of stream) { emittedEvents.push({ type: event.type, - error: event.type === "error" && "error" in event ? (event.error as any).errorMessage : undefined, + error: event.type === "error" ? event.error.errorMessage : undefined, }); } } catch { @@ -977,6 +987,10 @@ describe("reasoning-before-answer contentIndex invariant #6151", () => { }); describe("reader.read() error handling with pending tools #6151", () => { + afterEach(() => { + globalThis.fetch = originalFetch; + }); + test("emits completed tool events before error when reader.read() throws", async () => { const emittedEventTypes: string[] = []; @@ -1083,6 +1097,10 @@ describe("reader.read() error handling with pending tools #6151", () => { }); describe("Regression tests for #6151 issues", () => { + afterEach(() => { + globalThis.fetch = originalFetch; + }); + test("#4199034431: incomplete JSON frame without valid nested objects preserves data for next call", async () => { const { parseKiroApiEvents } = await import("../src/providers/kiro-api-key"); @@ -1247,3 +1265,745 @@ describe("Regression tests for #6151 issues", () => { expect(errorIdx).toBeGreaterThan(-1); }); }); + +describe("P1 Regression: incomplete tool emission at tool-ID rollover", () => { + afterEach(() => { + globalThis.fetch = originalFetch; + }); + + test("P1: incomplete tool is not emitted when tool ID changes", async () => { + const emittedEventTypes: string[] = []; + + globalThis.fetch = (async () => { + // First tool incomplete (no stop), then new tool ID changes + const tool1 = JSON.stringify({ + toolUseId: "tool-1", + name: "read_file", + input: '{"path": "/file1"}', + // NOTE: no "stop": true — tool is incomplete + }); + const tool2 = JSON.stringify({ + toolUseId: "tool-2", // Different ID triggers tool-ID transition + name: "write_file", + input: '{"path": "/file2"}', + stop: true, // This tool IS complete + }); + const responseBody = tool1 + tool2; + return new Response(responseBody, { status: 200 }); + }) as unknown as typeof fetch; + + try { + const stream = streamKiroApiKey(model, context, { apiKey: "ksk_test-secret", region: "us-east-1" }); + for await (const event of stream) { + emittedEventTypes.push(event.type); + } + } catch { + // Errors captured + } + + // Count toolcall events + // tool-1 was incomplete, so should NOT be emitted (both start AND end should be absent) + // tool-2 was complete (stop: true), so SHOULD be emitted + const toolcallStarts = emittedEventTypes.filter(e => e === "toolcall_start").length; + const toolcallEnds = emittedEventTypes.filter(e => e === "toolcall_end").length; + + expect(toolcallStarts).toBe(1); // Only tool-2 should have toolcall_start + expect(toolcallEnds).toBe(1); // Only tool-2 should have toolcall_end + }); + + test("P1: incomplete tool is not emitted at stream end", async () => { + const emittedEventTypes: string[] = []; + + globalThis.fetch = (async () => { + // Tool without stop flag at stream end, with text so response is not empty + const text = JSON.stringify({ + content: "I cannot complete this request.", + }); + const tool = JSON.stringify({ + toolUseId: "tool-incomplete", + name: "read_file", + input: '{"path": "/tmp/test"}', + // NOTE: no "stop": true — tool is incomplete + }); + const completion = JSON.stringify({ + stopReason: "COMPLETED", + usage: { inputTokens: 5, outputTokens: 2 }, + }); + const responseBody = text + tool + completion; + return new Response(responseBody, { status: 200 }); + }) as unknown as typeof fetch; + + try { + const stream = streamKiroApiKey(model, context, { apiKey: "ksk_test-secret", region: "us-east-1" }); + for await (const event of stream) { + emittedEventTypes.push(event.type); + } + } catch { + // Errors captured + } + + // With the P1 fix: incomplete tool should NOT emit toolcall_start/end + // (Without fix, incomplete tool would be forced into blocks at stream end) + const toolcallStart = emittedEventTypes.find(e => e === "toolcall_start"); + const toolcallEnd = emittedEventTypes.find(e => e === "toolcall_end"); + const done = emittedEventTypes.find(e => e === "done"); + + // Stream completes normally with done event + expect(done).toBeDefined(); + // But incomplete tool is NOT emitted (without explicit stop flag) + expect(toolcallStart).toBeUndefined(); + expect(toolcallEnd).toBeUndefined(); + }); + + test("incomplete tool-only response errors at clean EOF without emitting a tool", async () => { + const events: AssistantMessageEvent[] = []; + + globalThis.fetch = (async () => { + const incompleteTool = JSON.stringify({ + toolUseId: "tool-incomplete", + name: "read_file", + input: '{"path":"/tmp/test"}', + }); + return new Response(incompleteTool, { status: 200 }); + }) as unknown as typeof fetch; + + const stream = streamKiroApiKey(model, context, { apiKey: "ksk_test-secret", region: "us-east-1" }); + for await (const event of stream) { + events.push(event); + } + + const errorEvent = events.find( + (event): event is Extract => event.type === "error", + ); + expect(errorEvent?.error.errorMessage).toBe("Kiro API key stream returned no tokens"); + expect(events.some(event => event.type === "done")).toBe(false); + expect(events.some(event => event.type === "toolcall_start" || event.type === "toolcall_end")).toBe(false); + }); +}); + +describe("P2 Probepark findings: unconfirmed text on read rejection", () => { + afterEach(() => { + globalThis.fetch = originalFetch; + }); + + test("P2: unconfirmed text should NOT be emitted when read error occurs", async () => { + const emittedEvents: string[] = []; + + globalThis.fetch = (async () => { + // Simulate a response that delivers content but then errors + const body = new ReadableStream({ + start(controller) { + // Send only content, NO terminal metadata + const encoder = new TextEncoder(); + const chunk = encoder.encode(JSON.stringify({ content: "secret_data" })); + controller.enqueue(chunk); + }, + pull(controller) { + // Simulate stream error on second read + controller.error(new Error("Stream error")); + }, + }); + return new Response(body, { status: 200 }); + }) as unknown as typeof fetch; + + try { + const stream = streamKiroApiKey(model, context, { apiKey: "ksk_test-secret", region: "us-east-1" }); + for await (const event of stream) { + emittedEvents.push(event.type); + } + } catch { + // Errors captured in stream + } + + // Content without terminal metadata should NOT be emitted when error occurs + const textDeltas = emittedEvents.filter(e => e === "text_delta"); + expect(textDeltas).toHaveLength(0); + + // Error event should be emitted + const errorIdx = emittedEvents.indexOf("error"); + expect(errorIdx).toBeGreaterThan(-1); + }); + + test("usage metadata does not confirm pending text before a later read error", async () => { + const emittedEvents: AssistantMessageEvent[] = []; + + globalThis.fetch = (async () => { + // Simulate a response with content AND terminal metadata, then error + const body = new ReadableStream({ + start(controller) { + const encoder = new TextEncoder(); + // Content + terminal metadata in first pull + const content = encoder.encode(JSON.stringify({ content: "confirmed" })); + const usage = encoder.encode(JSON.stringify({ usage: { inputTokens: 1, outputTokens: 1 } })); + controller.enqueue(content); + controller.enqueue(usage); + }, + pull(controller) { + // Simulate error on second read (after confirmed metadata) + controller.error(new Error("Network error")); + }, + }); + return new Response(body, { status: 200 }); + }) as unknown as typeof fetch; + + try { + const stream = streamKiroApiKey(model, context, { apiKey: "ksk_test-secret", region: "us-east-1" }); + for await (const event of stream) { + emittedEvents.push(event); + } + } catch { + // Errors captured in stream + } + + const textEvents = emittedEvents.filter( + event => event.type === "text_start" || event.type === "text_delta" || event.type === "text_end", + ); + expect(textEvents).toHaveLength(0); + const errorEvent = emittedEvents.find( + (event): event is Extract => event.type === "error", + ); + expect(errorEvent?.error.errorMessage).toContain("Network error"); + expect(getTextBlocks(errorEvent?.error)).toHaveLength(0); + }); + + test("unknown stopReason does not confirm pending text before a later read error", async () => { + const emittedEvents: AssistantMessageEvent[] = []; + + globalThis.fetch = (async () => { + const eventSequence = [ + JSON.stringify({ content: "unconfirmed" }), + JSON.stringify({ stopReason: "FUTURE_STOP_REASON" }), + ]; + let emitted = 0; + const body = new ReadableStream({ + pull(controller) { + if (emitted < eventSequence.length) { + controller.enqueue(new TextEncoder().encode(eventSequence[emitted])); + emitted++; + } else { + controller.error(new Error("Network error after unknown stop reason")); + } + }, + }); + return new Response(body, { status: 200 }); + }) as unknown as typeof fetch; + + const stream = streamKiroApiKey(model, context, { apiKey: "ksk_test-secret", region: "us-east-1" }); + for await (const event of stream) { + emittedEvents.push(event); + } + + const errorEvent = emittedEvents.find( + (event): event is Extract => event.type === "error", + ); + expect(errorEvent?.error.errorMessage).toContain("Network error after unknown stop reason"); + expect(getTextBlocks(errorEvent?.error)).toHaveLength(0); + expect(emittedEvents.some(event => event.type === "text_delta")).toBe(false); + }); +}); + +describe("P1 Regression: content leaking across batches", () => { + afterEach(() => { + globalThis.fetch = originalFetch; + }); + + test("P1: content and refusal in same batch suppresses text", async () => { + const emittedEvents: Array<{ type: string }> = []; + + globalThis.fetch = (async () => { + // Content and refusal in same JSON batch (same read) + const responseBody = + JSON.stringify({ content: "This is harmful content" }) + + JSON.stringify({ + stopReason: "CONTENT_FILTERED", + stopDetails: { + refusal: { + category: "VIOLENCE", + explanation: "Violent content not allowed", + }, + }, + }); + return new Response(responseBody, { status: 200 }); + }) as unknown as typeof fetch; + + try { + const stream = streamKiroApiKey(model, context, { apiKey: "ksk_test-secret", region: "us-east-1" }); + for await (const event of stream) { + emittedEvents.push({ type: event.type }); + } + } catch { + // Errors captured + } + + // Refusal in same batch should suppress content + const errorIndex = emittedEvents.findIndex(e => e.type === "error"); + expect(errorIndex).toBeGreaterThan(-1); + + const textBeforeError = emittedEvents + .slice(0, errorIndex) + .filter(e => e.type === "text_delta" || e.type === "text_start"); + + // Content should be suppressed because refusal was in same batch + expect(textBeforeError).toHaveLength(0); + }); + + test("P2: unconfirmed text not in published partials when refusal follows", async () => { + const events: AssistantMessageEvent[] = []; + + globalThis.fetch = (async () => { + const eventSequence = [ + JSON.stringify({ content: "Harmful text here" }), + JSON.stringify({ + stopReason: "CONTENT_FILTERED", + stopDetails: { + refusal: { + category: "VIOLENCE", + explanation: "Violent content not allowed", + }, + }, + }), + ]; + let emitted = 0; + const body = new ReadableStream({ + pull(controller) { + if (emitted < eventSequence.length) { + controller.enqueue(new TextEncoder().encode(eventSequence[emitted])); + emitted++; + } else { + controller.close(); + } + }, + }); + return new Response(body, { status: 200 }); + }) as unknown as typeof fetch; + + const stream = streamKiroApiKey(model, context, { apiKey: "ksk_test-secret", region: "us-east-1" }); + for await (const event of stream) { + events.push(event); + } + + const escapedTextEvents = events.filter( + event => event.type === "text_start" || event.type === "text_delta" || event.type === "text_end", + ); + expect(escapedTextEvents).toHaveLength(0); + const startEvent = events.find( + (event): event is Extract => event.type === "start", + ); + expect(startEvent?.partial.content).toHaveLength(0); + const refusalEvent = events.find( + (event): event is Extract => event.type === "error", + ); + expect(refusalEvent?.error.errorMessage).toContain("Kiro refused the request"); + expect(refusalEvent?.error.content).toHaveLength(0); + }); + + test("text delta partials stay empty until text_end commits content", async () => { + const events: AssistantMessageEvent[] = []; + + globalThis.fetch = (async () => { + const eventSequence = [ + JSON.stringify({ content: "Test content" }), + JSON.stringify({ stopReason: "COMPLETED" }), + ]; + let emitted = 0; + const body = new ReadableStream({ + pull(controller) { + if (emitted < eventSequence.length) { + controller.enqueue(new TextEncoder().encode(eventSequence[emitted])); + emitted++; + } else { + controller.close(); + } + }, + }); + return new Response(body, { status: 200 }); + }) as unknown as typeof fetch; + + const stream = streamKiroApiKey(model, context, { apiKey: "ksk_test-secret", region: "us-east-1" }); + for await (const event of stream) { + events.push(event); + } + + const startEvent = events.find( + (event): event is Extract => event.type === "start", + ); + expect(startEvent?.partial.content).toHaveLength(0); + const textDeltaEvents = events.filter( + (event): event is Extract => event.type === "text_delta", + ); + expect(textDeltaEvents).toHaveLength(1); + expect(getTextBlocks(textDeltaEvents[0]?.partial).map(block => block.text)).toEqual([""]); + const doneEvent = events.find( + (event): event is Extract => event.type === "done", + ); + expect(getTextBlocks(doneEvent?.message).map(block => block.text)).toEqual(["Test content"]); + }); +}); + +describe("Completion event handling (probepark fix for #6451)", () => { + afterEach(() => { + globalThis.fetch = originalFetch; + }); + + test("accepted completion stopReason emits a completion event", async () => { + const emittedEventTypes: string[] = []; + + globalThis.fetch = (async () => { + const responseBody = JSON.stringify({ stopReason: "COMPLETED" }); + return new Response(responseBody, { status: 200 }); + }) as unknown as typeof fetch; + + try { + const stream = streamKiroApiKey(model, context, { apiKey: "ksk_test-secret", region: "us-east-1" }); + for await (const event of stream) { + // This test just verifies the stream completes + emittedEventTypes.push(event.type); + } + } catch { + // Error handling + } + + // Verify we get either done or error event (not hanging) + const done = emittedEventTypes.includes("done"); + const error = emittedEventTypes.includes("error"); + expect(done || error).toBe(true); + }); + + test("content + COMPLETED, then stream error: text block in error terminal", async () => { + // When stream contains content followed by COMPLETED event, then reader.read() throws, + // the error terminal should include the confirmed text block. + const emittedEvents: AssistantMessageEvent[] = []; + + globalThis.fetch = (async () => { + // Create a stream that delivers content + COMPLETED, then errors on next read + const eventSequence = [JSON.stringify({ content: "Hello" }), JSON.stringify({ stopReason: "COMPLETED" })]; + let emitted = 0; + + const body = new ReadableStream({ + pull(controller) { + if (emitted < eventSequence.length) { + const chunk = new TextEncoder().encode(eventSequence[emitted]); + controller.enqueue(chunk); + emitted++; + } else if (emitted === eventSequence.length) { + // After all events, throw an error on next read + emitted++; + controller.error(new Error("Network error after completion")); + } + }, + }); + + return new Response(body, { status: 200 }); + }) as unknown as typeof fetch; + + try { + const stream = streamKiroApiKey(model, context, { apiKey: "ksk_test-secret", region: "us-east-1" }); + for await (const event of stream) { + emittedEvents.push(event); + } + } catch { + // Stream error handling + } + + const errorEvent = emittedEvents.find( + (event): event is Extract => event.type === "error", + ); + expect(errorEvent?.error.errorMessage).toContain("Network error after completion"); + expect(getTextBlocks(errorEvent?.error).map(block => block.text)).toEqual(["Hello"]); + }); + + test("content without COMPLETED, then stream error: no text block in error terminal", async () => { + // When stream contains content but NO COMPLETED event (stream is incomplete), + // then reader.read() throws, the error terminal should NOT include unconfirmed text. + // This preserves existing P2 behavior (commit 9b23344 semantics). + const emittedEvents: AssistantMessageEvent[] = []; + + globalThis.fetch = (async () => { + // Create a stream that delivers content only, then errors + const eventSequence = [JSON.stringify({ content: "Hello" })]; + let emitted = 0; + + const body = new ReadableStream({ + pull(controller) { + if (emitted < eventSequence.length) { + const chunk = new TextEncoder().encode(eventSequence[emitted]); + controller.enqueue(chunk); + emitted++; + } else if (emitted === eventSequence.length) { + // After content, throw an error (NO COMPLETED event) + emitted++; + controller.error(new Error("Network error without completion")); + } + }, + }); + + return new Response(body, { status: 200 }); + }) as unknown as typeof fetch; + + try { + const stream = streamKiroApiKey(model, context, { apiKey: "ksk_test-secret", region: "us-east-1" }); + for await (const event of stream) { + emittedEvents.push(event); + } + } catch { + // Stream error handling + } + + // With no COMPLETED event, the stream should produce error event + const errorEvent = emittedEvents.find( + (event): event is Extract => event.type === "error", + ); + expect(errorEvent).toBeDefined(); + + // Since completionSeen is false, text content should not be in the error terminal + // (preserves P2 behavior: unconfirmed text is not published) + expect(getTextBlocks(errorEvent?.error)).toHaveLength(0); + }); + + test("stopReason CONTENT_FILTERED without stopDetails.refusal: treated as refusal", async () => { + // Finding #1: detail-free CONTENT_FILTERED should trigger refusal handling + const emittedEvents: AssistantMessageEvent[] = []; + + globalThis.fetch = (async () => { + // Stream with content followed by CONTENT_FILTERED (no stopDetails.refusal) + const eventSequence = [ + JSON.stringify({ content: "Malicious content" }), + JSON.stringify({ stopReason: "CONTENT_FILTERED" }), + ]; + let emitted = 0; + + const body = new ReadableStream({ + pull(controller) { + if (emitted < eventSequence.length) { + const chunk = new TextEncoder().encode(eventSequence[emitted]); + controller.enqueue(chunk); + emitted++; + } else { + controller.close(); + } + }, + }); + + return new Response(body, { status: 200 }); + }) as unknown as typeof fetch; + + try { + const stream = streamKiroApiKey(model, context, { apiKey: "ksk_test-secret", region: "us-east-1" }); + for await (const event of stream) { + emittedEvents.push(event); + } + } catch { + // Stream error handling + } + + // Verify that CONTENT_FILTERED triggers an error (refusal), not a done + const errorEvent = emittedEvents.find( + (event): event is Extract => event.type === "error", + ); + expect(errorEvent?.error.errorMessage).toMatch(/refuse/i); + expect(emittedEvents.some(event => event.type === "done")).toBe(false); + + // Verify that malicious content is not in the error output + expect(getTextBlocks(errorEvent?.error)).toHaveLength(0); + }); + + test("completion event emits all pending text events immediately (Finding #2)", async () => { + // When completion event arrives, all pending text events should be emitted immediately, + // not deferred until EOF. This includes text_start, text_delta, and text_end. + const emittedEvents: AssistantMessageEvent[] = []; + + globalThis.fetch = (async () => { + // Stream: content event, then completion event, then more data (which should be ignored) + const eventSequence = [ + JSON.stringify({ content: "Hello world" }), + JSON.stringify({ stopReason: "COMPLETED" }), + JSON.stringify({ content: "This should not appear" }), // Extra event after completion + ]; + let emitted = 0; + + const body = new ReadableStream({ + pull(controller) { + if (emitted < eventSequence.length) { + const chunk = new TextEncoder().encode(eventSequence[emitted]); + controller.enqueue(chunk); + emitted++; + } else { + controller.close(); + } + }, + }); + + return new Response(body, { status: 200 }); + }) as unknown as typeof fetch; + + try { + const stream = streamKiroApiKey(model, context, { apiKey: "ksk_test-secret", region: "us-east-1" }); + for await (const event of stream) { + emittedEvents.push(event); + } + } catch { + // Stream error handling + } + + // Verify text_start, text_delta, and text_end are emitted + const eventTypes = emittedEvents.map(e => e.type); + const hasTextStart = eventTypes.includes("text_start"); + const hasTextDelta = eventTypes.includes("text_delta"); + const hasTextEnd = eventTypes.includes("text_end"); + const hasDone = eventTypes.includes("done"); + + expect(hasTextStart).toBe(true); // text_start must be emitted + expect(hasTextDelta).toBe(true); // text_delta must be emitted + expect(hasTextEnd).toBe(true); // text_end must be emitted after completion + expect(hasDone).toBe(true); // done event must follow + + // Verify the order: text events should come before done + const textStartIdx = eventTypes.indexOf("text_start"); + const textEndIdx = eventTypes.indexOf("text_end"); + const doneIdx = eventTypes.indexOf("done"); + expect(textStartIdx).toBeLessThan(textEndIdx); + expect(textEndIdx).toBeLessThan(doneIdx); + const textDeltas = emittedEvents.filter( + (event): event is Extract => event.type === "text_delta", + ); + expect(textDeltas.map(event => event.delta).join("")).toBe("Hello world"); + const doneEvent = emittedEvents.find( + (event): event is Extract => event.type === "done", + ); + expect(getTextBlocks(doneEvent?.message).map(block => block.text)).toEqual(["Hello world"]); + }); + + test("incomplete JSON frame with nested object: no resync into nested (Finding #3)", async () => { + // Finding #3: Parser should not resync into nested objects within an incomplete outer frame. + // Stream incomplete frame like {"stopReason":"COMPLETED","usage": followed by EOF + // should error, not resync to find {inside the outer frame. + const emittedEvents: Array<{ type: string }> = []; + let errorThrown = false; + + globalThis.fetch = (async () => { + // Incomplete outer frame: stopReason complete but usage object starts but never closes + const incompleteFrame = `${JSON.stringify({ stopReason: "COMPLETED" }).slice(0, -1)},"usage":{"inputTokens":100`; // Incomplete, never closes + + const body = new ReadableStream({ + pull(controller) { + // Deliver the incomplete frame and close (EOF with incomplete data) + const chunk = new TextEncoder().encode(incompleteFrame); + controller.enqueue(chunk); + controller.close(); // EOF without closing the frame + }, + }); + + return new Response(body, { status: 200 }); + }) as unknown as typeof fetch; + + try { + const stream = streamKiroApiKey(model, context, { apiKey: "ksk_test-secret", region: "us-east-1" }); + for await (const event of stream) { + emittedEvents.push({ type: event.type }); + } + } catch (err) { + // Expect error due to incomplete JSON frame at EOF + errorThrown = true; + if (err instanceof Error) { + expect(err.message).toMatch(/truncated|incomplete/i); + } + } + + // Verify that an error was thrown (not successful completion) + const errorEvent = emittedEvents.find(e => e.type === "error"); + expect(errorThrown || errorEvent).toBeTruthy(); + }); + + test("incomplete refusal envelope with complete nested object is rejected at EOF", async () => { + const events: AssistantMessageEvent[] = []; + const eventSequence = [ + JSON.stringify({ content: "unconfirmed answer" }), + JSON.stringify({ + stopReason: "CONTENT_FILTERED", + stopDetails: { refusal: { category: "VIOLENCE", explanation: "Do not show answer" } }, + }).slice(0, -1), + ]; + let emitted = 0; + + globalThis.fetch = (async () => { + const body = new ReadableStream({ + pull(controller) { + if (emitted < eventSequence.length) { + controller.enqueue(new TextEncoder().encode(eventSequence[emitted])); + emitted++; + } else { + controller.close(); + } + }, + }); + return new Response(body, { status: 200 }); + }) as unknown as typeof fetch; + + const stream = streamKiroApiKey(model, context, { apiKey: "ksk_test-secret", region: "us-east-1" }); + for await (const event of stream) { + events.push(event); + } + + const errorEvent = events.find( + (event): event is Extract => event.type === "error", + ); + expect(errorEvent?.error.errorMessage).toContain("incomplete JSON frame at EOF"); + expect(getTextBlocks(errorEvent?.error)).toHaveLength(0); + expect( + events.some(event => event.type === "text_start" || event.type === "text_delta" || event.type === "text_end"), + ).toBe(false); + }); + + test("content + COMPLETED, then error event: text_end emitted before error (Finding #4)", async () => { + // Finding #4: When a server error event is received after COMPLETED, + // all text events including text_end should be emitted before the error. + const emittedEvents: Array<{ type: string }> = []; + + globalThis.fetch = (async () => { + const eventSequence = [ + JSON.stringify({ content: "Hello" }), + JSON.stringify({ stopReason: "COMPLETED" }), + JSON.stringify({ error: "Server error", message: "Internal server error" }), + ]; + let emitted = 0; + + const body = new ReadableStream({ + pull(controller) { + if (emitted < eventSequence.length) { + const chunk = new TextEncoder().encode(eventSequence[emitted]); + controller.enqueue(chunk); + emitted++; + } else { + controller.close(); + } + }, + }); + + return new Response(body, { status: 200 }); + }) as unknown as typeof fetch; + + try { + const stream = streamKiroApiKey(model, context, { apiKey: "ksk_test-secret", region: "us-east-1" }); + for await (const event of stream) { + emittedEvents.push({ type: event.type }); + } + } catch { + // Stream error handling + } + + // Verify event sequence includes text_end before error + const eventTypes = emittedEvents.map(e => e.type); + const hasTextStart = eventTypes.includes("text_start"); + const hasTextEnd = eventTypes.includes("text_end"); + const hasError = eventTypes.includes("error"); + + expect(hasTextStart).toBe(true); // text_start should be emitted + expect(hasTextEnd).toBe(true); // text_end should be emitted before error + expect(hasError).toBe(true); // error event follows + + // Verify order: text_end comes before error + const textEndIdx = eventTypes.indexOf("text_end"); + const errorIdx = eventTypes.indexOf("error"); + expect(textEndIdx).toBeLessThan(errorIdx); + }); +}); diff --git a/packages/ai/test/kiro-api-key.test.ts b/packages/ai/test/kiro-api-key.test.ts index fd3e4294c05..bf0c8ac5a46 100644 --- a/packages/ai/test/kiro-api-key.test.ts +++ b/packages/ai/test/kiro-api-key.test.ts @@ -138,17 +138,23 @@ describe("parseKiroApiEvents", () => { test("parses combined usage and stopReason:COMPLETED in the same metadata object", () => { // Metadata object with usage and normal completion (no refusal) const { events } = parseKiroApiEvents('{"stopReason":"COMPLETED","usage":{"inputTokens":20,"outputTokens":5}}'); - // Should only emit usage event (normal completion without refusal is ignored) + // Usage is retained for accounting and the accepted completion confirms the stream. const usageEvents = events.filter(e => e.type === "usage"); expect(usageEvents).toHaveLength(1); if (usageEvents[0]?.type === "usage") { expect(usageEvents[0].data.inputTokens).toBe(20); expect(usageEvents[0].data.outputTokens).toBe(5); } + const completionEvent = events.find(e => e.type === "completion"); + expect(completionEvent).toEqual({ type: "completion", data: { stopReason: "COMPLETED" } }); // Should not emit any refusal events for COMPLETED const refusalEvents = events.filter(e => e.type === "refusal"); expect(refusalEvents).toHaveLength(0); }); + test("does not treat an unknown stop reason as successful completion", () => { + const { events } = parseKiroApiEvents('{"stopReason":"FUTURE_STOP_REASON"}'); + expect(events).toHaveLength(0); + }); test("detects refusal even when stopReason/stopDetails is not the first property (order-independent parsing)", () => { // P1 fix (#6150): refusal must be detected regardless of property order // Previously, pattern-based matching would miss this because stopReason is not first From fae409f2778f71b05d097ee8e1b12dc58b2b70da Mon Sep 17 00:00:00 2001 From: clawdbot Date: Tue, 13 Oct 2026 09:51:24 +0000 Subject: [PATCH 2/5] fix(ai): prioritize Kiro refusals and preserve tool indices A completion frame in the same batch as refusal must not release quarantined text. On error paths, remove unconfirmed text and reindex pending tool events before emission so their contentIndex still points to retained tool blocks. Tested: bun --cwd=packages/ai run check Tested: bun test packages/ai/test/kiro-api-key-content-filter.test.ts packages/ai/test/kiro-api-key.test.ts packages/ai/test/provider-safety-stop.test.ts packages/ai/test/provider-safety-stop-late-refusal-ordering.test.ts Not-tested: live Kiro API stream Confidence: high Scope-risk: medium Reversibility: easy --- .../changelog.d/6150-kiro-content-filtered.md | 2 +- packages/ai/src/providers/kiro-api-key.ts | 56 +++----- .../test/kiro-api-key-content-filter.test.ts | 131 ++++++++++++++++++ 3 files changed, 154 insertions(+), 35 deletions(-) diff --git a/packages/ai/changelog.d/6150-kiro-content-filtered.md b/packages/ai/changelog.d/6150-kiro-content-filtered.md index 91703554b7f..f880102e9f8 100644 --- a/packages/ai/changelog.d/6150-kiro-content-filtered.md +++ b/packages/ai/changelog.d/6150-kiro-content-filtered.md @@ -1,6 +1,6 @@ ### Fixed -- Kiro streams now surface explicit content-filter refusal messages with category and explanation instead of the generic "no tokens" error when a refusal is detected. Both bearer token (via `messageMetadataEvent`) and API-key (`ksk_`, via `metadataEvent`) paths are supported, with proper provider safety-stop minting. Text is streamed incrementally unless a thinking block precedes it, in which case the entire answer (after thinking) is buffered until the stream end; tool calls are emitted only at stream end (#6150). +- Kiro streams now surface explicit content-filter refusal messages with category and explanation instead of the generic "no tokens" error when a refusal is detected. Both bearer token (via `messageMetadataEvent`) and API-key (`ksk_`, via `metadataEvent`) paths are supported, with proper provider safety-stop minting. Bearer-token text streams incrementally unless a thinking block precedes it; API-key text remains quarantined until successful terminal metadata or clean EOF. Tool calls are emitted only at stream end (#6150). - Detail-free `stopReason: "CONTENT_FILTERED"` (without `stopDetails.refusal` object) is now treated as a refusal and suppresses unconfirmed content. - API-key text is released only after `COMPLETED`/`TOOL_USE` metadata or clean EOF; usage and unknown stop reasons do not confirm it. Successful-terminal text events are emitted before later transport failures. - Parser no longer resyncs into nested objects when an incomplete outer JSON frame is encountered. Incomplete frames are retained as remainder; EOF with incomplete frame throws an error. diff --git a/packages/ai/src/providers/kiro-api-key.ts b/packages/ai/src/providers/kiro-api-key.ts index 95e7f783353..f2dfe17b5d7 100644 --- a/packages/ai/src/providers/kiro-api-key.ts +++ b/packages/ai/src/providers/kiro-api-key.ts @@ -826,7 +826,6 @@ export const streamKiroApiKey: StreamFunction<"kiro-codewhisperer-stream"> = ( // Variables for streaming tool calls (initialized before try block so they're accessible in catch) let currentTool: { id: string; name: string; input: string } | undefined; let toolComplete = false; // Track whether the current tool has been completed (has stop flag) - let toolcallIndex: number | undefined; const pendingToolCalls: Array<{ input: string; toolCall: ToolCall; index: number }> = []; // Only an accepted successful completion confirms buffered text before clean EOF. let completionSeen = false; @@ -835,6 +834,24 @@ export const streamKiroApiKey: StreamFunction<"kiro-codewhisperer-stream"> = ( let textStartDeferred = false; // Track if text_start is deferred (accessible in catch) let pendingTextContent = ""; // Text buffer (accessible in catch) const pendingTextEvents: Array<{ type: "start" | "delta" | "end"; delta?: string; content?: string }> = []; // Deferred text events (accessible in catch) + const removeUnconfirmedTextBlock = () => { + if (textContentConfirmed || textIndex === undefined) return; + + const removedIndex = textIndex; + blocks.splice(removedIndex, 1); + for (let i = removedIndex; i < blocks.length; i++) { + const block = blocks[i] as Block; + if (block && typeof block === "object" && "index" in block) { + block.index = i; + } + } + for (const pending of pendingToolCalls) { + if (pending.index > removedIndex) pending.index--; + } + textIndex = undefined; + pendingTextContent = ""; + pendingTextEvents.length = 0; + }; try { // Use validated identity snapshot from trust check if available (no second read of getters), @@ -928,9 +945,6 @@ export const streamKiroApiKey: StreamFunction<"kiro-codewhisperer-stream"> = ( }; const index = blocks.length; blocks.push({ ...toolCall, index }); - if (toolcallIndex === undefined) { - toolcallIndex = index; - } const toolArgs = typeof parsed === "object" && parsed !== null && !Array.isArray(parsed) ? (parsed as Record) @@ -1109,7 +1123,6 @@ export const streamKiroApiKey: StreamFunction<"kiro-codewhisperer-stream"> = ( } // Update state tracking variables for shifted blocks if (textIndex !== undefined) textIndex += 1; - if (toolcallIndex !== undefined) toolcallIndex += 1; // Update indices in pending tool calls since all blocks shifted by 1 for (const pending of pendingToolCalls) { pending.index += 1; @@ -1195,7 +1208,7 @@ export const streamKiroApiKey: StreamFunction<"kiro-codewhisperer-stream"> = ( if (event.data.inputTokens !== undefined) output.usage.input = event.data.inputTokens; if (event.data.outputTokens !== undefined) output.usage.output = event.data.outputTokens; output.usage.totalTokens = output.usage.input + output.usage.output; - } else if (event.type === "completion") { + } else if (event.type === "completion" && !hasTerminalEvent) { // Accepted successful terminal metadata confirms text, unlike usage accounting. completionSeen = true; // Finding #2: Immediately commit and emit all pending text events on completion @@ -1289,22 +1302,9 @@ export const streamKiroApiKey: StreamFunction<"kiro-codewhisperer-stream"> = ( // Only emit the tool if it was explicitly completed (has stop flag) if (toolComplete) addToolToBlocks(); + removeUnconfirmedTextBlock(); emitPendingToolCalls(); - // Remove unconfirmed text blocks from the error event - if (!textContentConfirmed && textIndex !== undefined) { - // Remove the unconfirmed text block from blocks - blocks.splice(textIndex, 1); - // Update indices of blocks after the removed one - for (let i = textIndex; i < blocks.length; i++) { - const block = blocks[i] as Block; - if (block && typeof block === "object" && "index" in block) { - block.index = i; - } - } - textIndex = undefined; - } - // Preserve text in the error context only after successful completion. const accumulatedText = textContentConfirmed ? blocks @@ -1412,6 +1412,8 @@ export const streamKiroApiKey: StreamFunction<"kiro-codewhisperer-stream"> = ( } } + removeUnconfirmedTextBlock(); + // Emit all pending tool call events (inline of emitPendingToolCalls logic) for (const { input, toolCall, index } of pendingToolCalls) { stream.push({ type: "toolcall_start", contentIndex: index, partial: output }); @@ -1430,20 +1432,6 @@ export const streamKiroApiKey: StreamFunction<"kiro-codewhisperer-stream"> = ( } pendingToolCalls.length = 0; - // Remove unconfirmed text blocks from the error event - if (!textContentConfirmed && textIndex !== undefined) { - // Remove the unconfirmed text block from blocks - blocks.splice(textIndex, 1); - // Update indices of blocks after the removed one - for (let i = textIndex; i < blocks.length; i++) { - const block = blocks[i] as Block; - if (block && typeof block === "object" && "index" in block) { - block.index = i; - } - } - textIndex = undefined; - } - output.stopReason = options?.signal?.aborted ? "aborted" : "error"; output.errorMessage = sanitizeKiroError(error, apiKey); output.duration = Date.now() - startTime; diff --git a/packages/ai/test/kiro-api-key-content-filter.test.ts b/packages/ai/test/kiro-api-key-content-filter.test.ts index aafca17d127..37b69a339f9 100644 --- a/packages/ai/test/kiro-api-key-content-filter.test.ts +++ b/packages/ai/test/kiro-api-key-content-filter.test.ts @@ -33,6 +33,23 @@ function getTextBlocks(message: AssistantMessage | undefined): TextContent[] { return message?.content.filter((block): block is TextContent => block.type === "text") ?? []; } +function expectToolEventsReferenceToolBlocks(events: AssistantMessageEvent[]): void { + const toolEvents = events.filter( + ( + event, + ): event is Extract => + event.type === "toolcall_start" || event.type === "toolcall_delta" || event.type === "toolcall_end", + ); + expect(toolEvents.length).toBeGreaterThan(0); + for (const event of toolEvents) { + const block = event.partial.content[event.contentIndex]; + expect(block?.type).toBe("toolCall"); + if (event.type === "toolcall_end" && block?.type === "toolCall") { + expect(block.id).toBe(event.toolCall.id); + } + } +} + describe("Kiro API-key content filter #6150", () => { afterEach(() => { globalThis.fetch = originalFetch; @@ -1500,6 +1517,79 @@ describe("P2 Probepark findings: unconfirmed text on read rejection", () => { expect(getTextBlocks(errorEvent?.error)).toHaveLength(0); expect(emittedEvents.some(event => event.type === "text_delta")).toBe(false); }); + + test("reindexes completed-tool lifecycle events before an ordinary error", async () => { + const events: AssistantMessageEvent[] = []; + const eventSequence = [ + JSON.stringify({ content: "unconfirmed text" }), + JSON.stringify({ toolUseId: "tool-1", name: "read_file", input: '{"path":"/tmp/test"}', stop: true }), + JSON.stringify({ error: "server_error", message: "Temporary failure" }), + ]; + let emitted = 0; + + globalThis.fetch = (async () => { + const body = new ReadableStream({ + pull(controller) { + if (emitted < eventSequence.length) { + controller.enqueue(new TextEncoder().encode(eventSequence[emitted])); + emitted++; + } else { + controller.close(); + } + }, + }); + return new Response(body, { status: 200 }); + }) as unknown as typeof fetch; + + const stream = streamKiroApiKey(model, context, { apiKey: "ksk_test-secret", region: "us-east-1" }); + for await (const event of stream) { + events.push(event); + } + + const errorEvent = events.find( + (event): event is Extract => event.type === "error", + ); + expect(errorEvent?.error.errorMessage).toContain("server_error"); + expect(getTextBlocks(errorEvent?.error)).toHaveLength(0); + expect(errorEvent?.error.content[0]?.type).toBe("toolCall"); + expectToolEventsReferenceToolBlocks(events); + }); + + test("reindexes completed-tool lifecycle events before a reader rejection", async () => { + const events: AssistantMessageEvent[] = []; + const eventSequence = [ + JSON.stringify({ content: "unconfirmed text" }), + JSON.stringify({ toolUseId: "tool-1", name: "read_file", input: '{"path":"/tmp/test"}', stop: true }), + ]; + let emitted = 0; + + globalThis.fetch = (async () => { + const body = new ReadableStream({ + pull(controller) { + if (emitted < eventSequence.length) { + controller.enqueue(new TextEncoder().encode(eventSequence[emitted])); + emitted++; + } else { + controller.error(new Error("Network error after completed tool")); + } + }, + }); + return new Response(body, { status: 200 }); + }) as unknown as typeof fetch; + + const stream = streamKiroApiKey(model, context, { apiKey: "ksk_test-secret", region: "us-east-1" }); + for await (const event of stream) { + events.push(event); + } + + const errorEvent = events.find( + (event): event is Extract => event.type === "error", + ); + expect(errorEvent?.error.errorMessage).toContain("Network error after completed tool"); + expect(getTextBlocks(errorEvent?.error)).toHaveLength(0); + expect(errorEvent?.error.content[0]?.type).toBe("toolCall"); + expectToolEventsReferenceToolBlocks(events); + }); }); describe("P1 Regression: content leaking across batches", () => { @@ -1547,6 +1637,47 @@ describe("P1 Regression: content leaking across batches", () => { expect(textBeforeError).toHaveLength(0); }); + test("batched refusal overrides completion and suppresses quarantined text", async () => { + const events: AssistantMessageEvent[] = []; + const eventSequence = [ + JSON.stringify({ content: "quarantined answer" }), + JSON.stringify({ stopReason: "COMPLETED" }) + + JSON.stringify({ + stopReason: "CONTENT_FILTERED", + stopDetails: { refusal: { category: "VIOLENCE", explanation: "Do not show answer" } }, + }), + ]; + let emitted = 0; + + globalThis.fetch = (async () => { + const body = new ReadableStream({ + pull(controller) { + if (emitted < eventSequence.length) { + controller.enqueue(new TextEncoder().encode(eventSequence[emitted])); + emitted++; + } else { + controller.close(); + } + }, + }); + return new Response(body, { status: 200 }); + }) as unknown as typeof fetch; + + const stream = streamKiroApiKey(model, context, { apiKey: "ksk_test-secret", region: "us-east-1" }); + for await (const event of stream) { + events.push(event); + } + + const errorEvent = events.find( + (event): event is Extract => event.type === "error", + ); + expect(errorEvent?.error.errorMessage).toContain("Kiro refused the request (VIOLENCE)"); + expect(errorEvent?.error.content).toHaveLength(0); + expect( + events.some(event => event.type === "text_start" || event.type === "text_delta" || event.type === "text_end"), + ).toBe(false); + }); + test("P2: unconfirmed text not in published partials when refusal follows", async () => { const events: AssistantMessageEvent[] = []; From 9efba79fd62c645e2b8f6d18d8d4016539a8b464 Mon Sep 17 00:00:00 2001 From: clawdbot Date: Tue, 13 Oct 2026 10:43:07 +0000 Subject: [PATCH 3/5] fix(ai): close Kiro content-filter review findings Duplicate error partials bypassed gateway redaction, and terminal metadata fenced fragments of an active tool. Keep refusal precedence while preventing the leak across public SSE boundaries. Tested: bun --cwd=packages/ai run check Tested: bun test packages/ai/test/kiro-api-key-content-filter.test.ts packages/ai/test/kiro-api-key.test.ts packages/ai/test/provider-safety-stop.test.ts packages/ai/test/provider-safety-stop-late-refusal-ordering.test.ts packages/ai/test/auth-gateway-pi-native.test.ts Tested: git diff --check Not-tested: live Kiro API stream Confidence: high Scope-risk: medium Reversibility: easy --- .../changelog.d/6150-kiro-content-filtered.md | 2 + packages/ai/src/auth-gateway/server.ts | 5 +- packages/ai/src/providers/kiro-api-key.ts | 21 ++-- packages/ai/src/providers/pi-native-server.ts | 2 +- .../ai/test/auth-gateway-pi-native.test.ts | 27 ++++- .../test/kiro-api-key-content-filter.test.ts | 106 ++++++++++++++++-- 6 files changed, 143 insertions(+), 20 deletions(-) diff --git a/packages/ai/changelog.d/6150-kiro-content-filtered.md b/packages/ai/changelog.d/6150-kiro-content-filtered.md index f880102e9f8..021d484d76b 100644 --- a/packages/ai/changelog.d/6150-kiro-content-filtered.md +++ b/packages/ai/changelog.d/6150-kiro-content-filtered.md @@ -3,5 +3,7 @@ - Kiro streams now surface explicit content-filter refusal messages with category and explanation instead of the generic "no tokens" error when a refusal is detected. Both bearer token (via `messageMetadataEvent`) and API-key (`ksk_`, via `metadataEvent`) paths are supported, with proper provider safety-stop minting. Bearer-token text streams incrementally unless a thinking block precedes it; API-key text remains quarantined until successful terminal metadata or clean EOF. Tool calls are emitted only at stream end (#6150). - Detail-free `stopReason: "CONTENT_FILTERED"` (without `stopDetails.refusal` object) is now treated as a refusal and suppresses unconfirmed content. - API-key text is released only after `COMPLETED`/`TOOL_USE` metadata or clean EOF; usage and unknown stop reasons do not confirm it. Successful-terminal text events are emitted before later transport failures. +- A refusal takes precedence over ordinary errors in the same batch. An active tool may finish its input after `TOOL_USE`, while later content and new tool generations remain fenced. +- Error events omit duplicate partial messages so credential-shaped upstream text cannot bypass gateway redaction or leak through SSE encoders. - Parser no longer resyncs into nested objects when an incomplete outer JSON frame is encountered. Incomplete frames are retained as remainder; EOF with incomplete frame throws an error. - Server error events now emit `text_end` for confirmed text before throwing the error, ensuring complete text lifecycle. diff --git a/packages/ai/src/auth-gateway/server.ts b/packages/ai/src/auth-gateway/server.ts index 99d92f91123..8e935f3cf09 100644 --- a/packages/ai/src/auth-gateway/server.ts +++ b/packages/ai/src/auth-gateway/server.ts @@ -589,8 +589,9 @@ export function redactGatewayStream(events: AssistantMessageEventStream): Assist async function* redactedEvents(): AsyncGenerator { try { for await (const event of events) { - if (event.type === "error") yield { ...event, error: redactGatewayMessage(event.error) }; - else yield event; + if (event.type === "error") { + yield { type: "error", reason: event.reason, error: redactGatewayMessage(event.error) }; + } else yield event; } } catch (error) { throw redactGatewayError(error); diff --git a/packages/ai/src/providers/kiro-api-key.ts b/packages/ai/src/providers/kiro-api-key.ts index f2dfe17b5d7..ec89935ce86 100644 --- a/packages/ai/src/providers/kiro-api-key.ts +++ b/packages/ai/src/providers/kiro-api-key.ts @@ -1157,12 +1157,17 @@ export const streamKiroApiKey: StreamFunction<"kiro-codewhisperer-stream"> = ( const hasTerminalEvent = events.some(e => e.type === "refusal"); for (const event of events) { + const isActiveToolContinuation = + currentTool !== undefined && + !toolComplete && + (event.type === "toolUseInput" || + event.type === "toolUseStop" || + (event.type === "toolUse" && event.data.toolUseId === currentTool.id)); if ( completionSeen && (event.type === "content" || - event.type === "toolUse" || - event.type === "toolUseInput" || - event.type === "toolUseStop") + ((event.type === "toolUse" || event.type === "toolUseInput" || event.type === "toolUseStop") && + !isActiveToolContinuation)) ) { continue; } @@ -1201,7 +1206,7 @@ export const streamKiroApiKey: StreamFunction<"kiro-codewhisperer-stream"> = ( firstTokenTime = Date.now(); } currentTool.input += event.data.input; - } else if (event.type === "toolUseStop" && event.data.stop) { + } else if (event.type === "toolUseStop" && event.data.stop && !hasTerminalEvent) { toolComplete = true; // Mark tool as complete when toolUseStop is received addToolToBlocks(); } else if (event.type === "usage") { @@ -1261,11 +1266,11 @@ export const streamKiroApiKey: StreamFunction<"kiro-codewhisperer-stream"> = ( output.duration = Date.now() - startTime; if (firstTokenTime) output.ttft = firstTokenTime - startTime; - stream.push({ type: "error", reason: "error", error: output, partial: output }); + stream.push({ type: "error", reason: "error", error: output }); stream.end(); return; } - } else if (event.type === "error") { + } else if (event.type === "error" && !hasTerminalEvent) { // On ordinary server errors (error event in stream), emit deferred text and tool events before error // (refusals drop them, incomplete tools must not be emitted) @@ -1346,7 +1351,7 @@ export const streamKiroApiKey: StreamFunction<"kiro-codewhisperer-stream"> = ( output.errorMessage = "Kiro API key stream returned no tokens"; output.duration = Date.now() - startTime; if (firstTokenTime) output.ttft = firstTokenTime - startTime; - stream.push({ type: "error", reason: "error", error: output, partial: output }); + stream.push({ type: "error", reason: "error", error: output }); stream.end(); return; } @@ -1436,7 +1441,7 @@ export const streamKiroApiKey: StreamFunction<"kiro-codewhisperer-stream"> = ( output.errorMessage = sanitizeKiroError(error, apiKey); output.duration = Date.now() - startTime; if (firstTokenTime) output.ttft = firstTokenTime - startTime; - stream.push({ type: "error", reason: output.stopReason, error: output, partial: output }); + stream.push({ type: "error", reason: output.stopReason, error: output }); stream.end(); } })().catch(() => { diff --git a/packages/ai/src/providers/pi-native-server.ts b/packages/ai/src/providers/pi-native-server.ts index ed812f45452..99bcd42bac9 100644 --- a/packages/ai/src/providers/pi-native-server.ts +++ b/packages/ai/src/providers/pi-native-server.ts @@ -326,7 +326,7 @@ function sanitizeEvent(event: AssistantMessageEvent): AssistantMessageEvent | un case "done": return { ...event, message: sanitizeMessage(event.message) }; case "error": - return { ...event, error: sanitizeMessage(event.error) }; + return { type: "error", reason: event.reason, error: sanitizeMessage(event.error) }; case "toolChoiceIncapability": return event; case "thinking_start": diff --git a/packages/ai/test/auth-gateway-pi-native.test.ts b/packages/ai/test/auth-gateway-pi-native.test.ts index 37b3465f72d..8ef2093eed7 100644 --- a/packages/ai/test/auth-gateway-pi-native.test.ts +++ b/packages/ai/test/auth-gateway-pi-native.test.ts @@ -3,7 +3,7 @@ import * as fs from "node:fs/promises"; import * as os from "node:os"; import * as path from "node:path"; import { registerCustomApi, unregisterCustomApis } from "../src/api-registry"; -import { startAuthGateway } from "../src/auth-gateway/server"; +import { redactGatewayStream, startAuthGateway } from "../src/auth-gateway/server"; import { AuthStorage, SqliteAuthCredentialStore } from "../src/auth-storage"; import { Effort } from "../src/model-thinking"; import { encodeStream, formatError, parseRequest } from "../src/providers/pi-native-server"; @@ -314,6 +314,31 @@ describe("pi-native encodeStream", () => { expect(parsed[1]).toBe("[DONE]"); }); + it("drops credential-bearing error partials before direct and gateway SSE", async () => { + const credential = "ksk_upstream-secret-value"; + const errored = baseAssistant({ stopReason: "error", errorMessage: "Upstream request failed" }); + const partial = baseAssistant({ errorMessage: `authorization=Bearer ${credential}` }); + const events: AssistantMessageEvent[] = [{ type: "error", reason: "error", error: errored, partial }]; + + const directFrames = (await collectSse(encodeStream(makeEventStream(events, errored)))).map(parseSseLine); + expect(JSON.stringify(directFrames)).not.toContain(credential); + expect(directFrames[0]).not.toHaveProperty("partial"); + + const redactedEvents: AssistantMessageEvent[] = []; + for await (const event of redactGatewayStream(makeEventStream(events, errored))) { + redactedEvents.push(event); + } + const redactedError = redactedEvents[0]; + expect(redactedError?.type).toBe("error"); + if (redactedError?.type === "error") expect(redactedError.partial).toBeUndefined(); + + const gatewayFrames = (await collectSse(encodeStream(redactGatewayStream(makeEventStream(events, errored))))).map( + parseSseLine, + ); + expect(JSON.stringify(gatewayFrames)).not.toContain(credential); + expect(gatewayFrames[0]).not.toHaveProperty("partial"); + }); + it("replays mixed Responses reasoning safely across start, partial, done, and error envelopes", async () => { const startPartial = reasoningMessage("mixed", `${SUMMARY_SENTINEL} start`, "rs-start"); const deltaPartial = reasoningMessage("mixed", `${SUMMARY_SENTINEL} delta`, "rs-delta"); diff --git a/packages/ai/test/kiro-api-key-content-filter.test.ts b/packages/ai/test/kiro-api-key-content-filter.test.ts index 37b69a339f9..951e7b182a2 100644 --- a/packages/ai/test/kiro-api-key-content-filter.test.ts +++ b/packages/ai/test/kiro-api-key-content-filter.test.ts @@ -5,8 +5,17 @@ * "Kiro API key stream returned no tokens". */ import { afterEach, describe, expect, test } from "bun:test"; -import { isProviderSafetyStopModelTrusted } from "../src/adapter-internals/provider-safety-stop"; -import { kiroApiBaseUrl, kiroApiStaticModels, streamKiroApiKey } from "../src/providers/kiro-api-key"; +import { + isProviderSafetyStopAuthenticated, + isProviderSafetyStopModelTrusted, + withProviderSafetyStopAdapterInvocation, +} from "../src/adapter-internals/provider-safety-stop"; +import { + kiroApiBaseUrl, + kiroApiStaticModels, + parseKiroApiEvents, + streamKiroApiKey, +} from "../src/providers/kiro-api-key"; import type { AssistantMessage, AssistantMessageEvent, Context, Model, TextContent } from "../src/types"; const originalFetch = globalThis.fetch; @@ -1119,8 +1128,6 @@ describe("Regression tests for #6151 issues", () => { }); test("#4199034431: incomplete JSON frame without valid nested objects preserves data for next call", async () => { - const { parseKiroApiEvents } = await import("../src/providers/kiro-api-key"); - // Simulate a malformed chunk with a stray opening brace (no complete nested JSON) // This tests that when there's incomplete JSON with no valid nested objects to resync to, // the data is preserved in remaining for the next parse call @@ -1200,8 +1207,6 @@ describe("Regression tests for #6151 issues", () => { }); test("#4199034444: 64KiB retention cap enforced for stray brace with no valid JSON", async () => { - const { parseKiroApiEvents } = await import("../src/providers/kiro-api-key"); - // Create a buffer with a stray opening brace followed by 100KB of garbage // This should be truncated to 64KB max when returned as remaining const strayBrace = "{"; @@ -1221,8 +1226,6 @@ describe("Regression tests for #6151 issues", () => { }); test("#4199282140: balanced junk wrapper around valid refusal event is rescanned", async () => { - const { parseKiroApiEvents } = await import("../src/providers/kiro-api-key"); - // Codex finding: balanced junk wrapper around a valid event // e.g. '{junk{"stopReason":"CONTENT_FILTERED",...}}' should not be discarded const balancedJunkWrapper = @@ -1396,6 +1399,46 @@ describe("P1 Regression: incomplete tool emission at tool-ID rollover", () => { expect(events.some(event => event.type === "done")).toBe(false); expect(events.some(event => event.type === "toolcall_start" || event.type === "toolcall_end")).toBe(false); }); + + test("TOOL_USE completion allows the active tool to finish but fences later content", async () => { + const events: AssistantMessageEvent[] = []; + const eventSequence = [ + JSON.stringify({ stopReason: "TOOL_USE", toolUseId: "tool-continued", name: "search", input: '{"query":' }), + JSON.stringify({ input: '"kiro"}' }), + JSON.stringify({ stop: true }), + JSON.stringify({ content: "post-terminal text" }), + ]; + let emitted = 0; + + globalThis.fetch = (async () => { + const body = new ReadableStream({ + pull(controller) { + if (emitted < eventSequence.length) { + controller.enqueue(new TextEncoder().encode(eventSequence[emitted])); + emitted++; + } else { + controller.close(); + } + }, + }); + return new Response(body, { status: 200 }); + }) as unknown as typeof fetch; + + const stream = streamKiroApiKey(model, context, { apiKey: "ksk_test-secret", region: "us-east-1" }); + for await (const event of stream) { + events.push(event); + } + + const doneEvent = events.find( + (event): event is Extract => event.type === "done", + ); + expect(doneEvent?.reason).toBe("toolUse"); + const toolCall = doneEvent?.message.content.find(block => block.type === "toolCall"); + expect(toolCall?.type).toBe("toolCall"); + if (toolCall?.type === "toolCall") expect(toolCall.arguments).toEqual({ query: "kiro" }); + expect(events.some(event => event.type === "text_delta" && event.delta === "post-terminal text")).toBe(false); + expectToolEventsReferenceToolBlocks(events); + }); }); describe("P2 Probepark findings: unconfirmed text on read rejection", () => { @@ -1515,6 +1558,7 @@ describe("P2 Probepark findings: unconfirmed text on read rejection", () => { ); expect(errorEvent?.error.errorMessage).toContain("Network error after unknown stop reason"); expect(getTextBlocks(errorEvent?.error)).toHaveLength(0); + expect(errorEvent?.partial).toBeUndefined(); expect(emittedEvents.some(event => event.type === "text_delta")).toBe(false); }); @@ -1678,6 +1722,52 @@ describe("P1 Regression: content leaking across batches", () => { ).toBe(false); }); + test("batched refusal takes precedence over an earlier ordinary error", async () => { + const events: AssistantMessageEvent[] = []; + const trustedModel = kiroApiStaticModels()[0]; + if (!trustedModel) throw new Error("Expected a registered Kiro API-key model"); + const eventSequence = [ + JSON.stringify({ content: "quarantined answer" }), + JSON.stringify({ error: "ServerError", message: "ordinary error must not win" }) + + JSON.stringify({ + stopReason: "CONTENT_FILTERED", + stopDetails: { refusal: { category: "MALWARE", explanation: "Refusal must remain authoritative" } }, + }), + ]; + let emitted = 0; + + globalThis.fetch = (async () => { + const body = new ReadableStream({ + pull(controller) { + if (emitted < eventSequence.length) { + controller.enqueue(new TextEncoder().encode(eventSequence[emitted])); + emitted++; + } else { + controller.close(); + } + }, + }); + return new Response(body, { status: 200 }); + }) as unknown as typeof fetch; + + const options = withProviderSafetyStopAdapterInvocation({ apiKey: "ksk_test-secret", region: "us-east-1" }); + const stream = streamKiroApiKey(trustedModel, context, options); + for await (const event of stream) { + events.push(event); + } + + const errorEvent = events.find( + (event): event is Extract => event.type === "error", + ); + expect(errorEvent?.error.errorMessage).toContain("Kiro refused the request (MALWARE)"); + expect(errorEvent?.error.errorMessage).toContain("Refusal must remain authoritative"); + expect(errorEvent?.error.errorMessage).not.toContain("ordinary error must not win"); + expect(errorEvent?.error.errorKind).toBe("provider_safety_stop"); + expect(isProviderSafetyStopAuthenticated(errorEvent?.error)).toBe(true); + expect(errorEvent?.error.content).toHaveLength(0); + expect(events.some(event => event.type === "text_start" || event.type === "text_delta")).toBe(false); + }); + test("P2: unconfirmed text not in published partials when refusal follows", async () => { const events: AssistantMessageEvent[] = []; From 96e896186fd128c3307900128c54da6e47d179e5 Mon Sep 17 00:00:00 2001 From: clawdbot Date: Tue, 13 Oct 2026 10:49:50 +0000 Subject: [PATCH 4/5] fix(ai): close confirmed Kiro text before errors Ordinary errors after accepted completion must preserve text lifecycle invariants. Strengthen the post-terminal content and fragmented TOOL_USE regressions. Tested: bun --cwd=packages/ai run check Tested: bun test packages/ai/test/kiro-api-key-content-filter.test.ts packages/ai/test/kiro-api-key.test.ts packages/ai/test/provider-safety-stop.test.ts packages/ai/test/provider-safety-stop-late-refusal-ordering.test.ts packages/ai/test/auth-gateway-pi-native.test.ts Tested: git diff --check Not-tested: live Kiro API stream Confidence: high Scope-risk: medium Reversibility: easy --- packages/ai/src/providers/kiro-api-key.ts | 26 ++----------- .../test/kiro-api-key-content-filter.test.ts | 38 ++++++++++++++++++- 2 files changed, 41 insertions(+), 23 deletions(-) diff --git a/packages/ai/src/providers/kiro-api-key.ts b/packages/ai/src/providers/kiro-api-key.ts index ec89935ce86..86b77f7d4ae 100644 --- a/packages/ai/src/providers/kiro-api-key.ts +++ b/packages/ai/src/providers/kiro-api-key.ts @@ -1281,28 +1281,10 @@ export const streamKiroApiKey: StreamFunction<"kiro-codewhisperer-stream"> = ( if (textIndex !== undefined && textIndex < blocks.length && pendingTextContent.length > 0) { textContentConfirmed = true; } - // Emit deferred text start if not yet emitted - if (textStartDeferred && textIndex !== undefined && textIndex < blocks.length) { - const block = blocks[textIndex]; - if (block && block.type === "text" && pendingTextContent.length > 0) { - stream.push({ type: "text_start", contentIndex: textIndex, partial: snapshotOutput() }); - // Emit all pending text_delta events - for (const evt of pendingTextEvents) { - if (evt.type === "delta" && evt.delta) { - stream.push({ - type: "text_delta", - contentIndex: textIndex, - delta: evt.delta, - partial: snapshotOutput(), - }); - } - } - // Emit text_end for confirmed text before error. - block.text = pendingTextContent; - closeTextBlock(); - } - textStartDeferred = false; - } + } + if (textContentConfirmed) { + emitDeferredTextEvents(); + closeTextBlock(); } // Only emit the tool if it was explicitly completed (has stop flag) diff --git a/packages/ai/test/kiro-api-key-content-filter.test.ts b/packages/ai/test/kiro-api-key-content-filter.test.ts index 951e7b182a2..228e78e58ec 100644 --- a/packages/ai/test/kiro-api-key-content-filter.test.ts +++ b/packages/ai/test/kiro-api-key-content-filter.test.ts @@ -1436,7 +1436,10 @@ describe("P1 Regression: incomplete tool emission at tool-ID rollover", () => { const toolCall = doneEvent?.message.content.find(block => block.type === "toolCall"); expect(toolCall?.type).toBe("toolCall"); if (toolCall?.type === "toolCall") expect(toolCall.arguments).toEqual({ query: "kiro" }); - expect(events.some(event => event.type === "text_delta" && event.delta === "post-terminal text")).toBe(false); + expect(doneEvent?.message.content.filter(block => block.type === "text")).toHaveLength(0); + expect( + events.some(event => event.type === "text_start" || event.type === "text_delta" || event.type === "text_end"), + ).toBe(false); expectToolEventsReferenceToolBlocks(events); }); }); @@ -1562,6 +1565,39 @@ describe("P2 Probepark findings: unconfirmed text on read rejection", () => { expect(emittedEvents.some(event => event.type === "text_delta")).toBe(false); }); + test("confirmed text lifecycle closes before an ordinary error in the same batch", async () => { + const events: AssistantMessageEvent[] = []; + const responseBody = + JSON.stringify({ content: "confirmed before error" }) + + JSON.stringify({ stopReason: "COMPLETED" }) + + JSON.stringify({ error: "ServerError", message: "failure after completion" }); + + globalThis.fetch = (async () => { + const body = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(responseBody)); + controller.close(); + }, + }); + return new Response(body, { status: 200 }); + }) as unknown as typeof fetch; + + const stream = streamKiroApiKey(model, context, { apiKey: "ksk_test-secret", region: "us-east-1" }); + for await (const event of stream) { + events.push(event); + } + + const errorIndex = events.findIndex(event => event.type === "error"); + const textLifecycle = events + .slice(0, errorIndex) + .filter(event => event.type === "text_start" || event.type === "text_delta" || event.type === "text_end"); + expect(errorIndex).toBeGreaterThan(-1); + expect(textLifecycle.map(event => event.type)).toEqual(["text_start", "text_delta", "text_end"]); + const textEnd = textLifecycle.find(event => event.type === "text_end"); + expect(textEnd?.type === "text_end" ? textEnd.content : undefined).toBe("confirmed before error"); + expect(events.some(event => event.type === "done")).toBe(false); + }); + test("reindexes completed-tool lifecycle events before an ordinary error", async () => { const events: AssistantMessageEvent[] = []; const eventSequence = [ From a93f2a30eab08444b4eef419d1ca925251220bd6 Mon Sep 17 00:00:00 2001 From: clawdbot Date: Tue, 13 Oct 2026 11:26:58 +0000 Subject: [PATCH 5/5] fix(ai): make Kiro terminal handling chunk-safe Completion metadata cannot rule out a later refusal across reader chunks. Keep text private until clean EOF, restrict active tool continuation to TOOL_USE, and resync only after a JSON prefix is syntactically impossible. Tested: bun --cwd=packages/ai run check Tested: bun test packages/ai/test/kiro-api-key-content-filter.test.ts packages/ai/test/kiro-api-key.test.ts packages/ai/test/provider-safety-stop.test.ts packages/ai/test/provider-safety-stop-late-refusal-ordering.test.ts packages/ai/test/auth-gateway-pi-native.test.ts Tested: git diff --check Not-tested: live Kiro API stream Confidence: high Scope-risk: medium Reversibility: easy --- .../changelog.d/6150-kiro-content-filtered.md | 8 +- packages/ai/src/providers/kiro-api-key.ts | 286 ++++++++++++++---- .../test/kiro-api-key-content-filter.test.ts | 168 +++++++--- 3 files changed, 346 insertions(+), 116 deletions(-) diff --git a/packages/ai/changelog.d/6150-kiro-content-filtered.md b/packages/ai/changelog.d/6150-kiro-content-filtered.md index 021d484d76b..034f6eb7948 100644 --- a/packages/ai/changelog.d/6150-kiro-content-filtered.md +++ b/packages/ai/changelog.d/6150-kiro-content-filtered.md @@ -1,9 +1,9 @@ ### Fixed -- Kiro streams now surface explicit content-filter refusal messages with category and explanation instead of the generic "no tokens" error when a refusal is detected. Both bearer token (via `messageMetadataEvent`) and API-key (`ksk_`, via `metadataEvent`) paths are supported, with proper provider safety-stop minting. Bearer-token text streams incrementally unless a thinking block precedes it; API-key text remains quarantined until successful terminal metadata or clean EOF. Tool calls are emitted only at stream end (#6150). +- Kiro streams now surface explicit content-filter refusal messages with category and explanation instead of the generic "no tokens" error when a refusal is detected. Both bearer token (via `messageMetadataEvent`) and API-key (`ksk_`, via `metadataEvent`) paths are supported, with proper provider safety-stop minting. Bearer-token text streams incrementally unless a thinking block precedes it; API-key text remains quarantined until clean EOF. Tool calls are emitted only at stream end (#6150). - Detail-free `stopReason: "CONTENT_FILTERED"` (without `stopDetails.refusal` object) is now treated as a refusal and suppresses unconfirmed content. -- API-key text is released only after `COMPLETED`/`TOOL_USE` metadata or clean EOF; usage and unknown stop reasons do not confirm it. Successful-terminal text events are emitted before later transport failures. -- A refusal takes precedence over ordinary errors in the same batch. An active tool may finish its input after `TOOL_USE`, while later content and new tool generations remain fenced. +- API-key text remains quarantined through `COMPLETED`/`TOOL_USE` metadata and is released only at clean EOF, so a later refusal or read failure cannot leak content across HTTP chunk boundaries. +- A refusal takes precedence over ordinary errors or successful completion in the same batch or a later read. An active tool may finish its input after `TOOL_USE`, while `COMPLETED` fences tool fragments and later content. - Error events omit duplicate partial messages so credential-shaped upstream text cannot bypass gateway redaction or leak through SSE encoders. - Parser no longer resyncs into nested objects when an incomplete outer JSON frame is encountered. Incomplete frames are retained as remainder; EOF with incomplete frame throws an error. -- Server error events now emit `text_end` for confirmed text before throwing the error, ensuring complete text lifecycle. +- Errors before clean EOF discard quarantined text and retain only explicitly completed tool calls. diff --git a/packages/ai/src/providers/kiro-api-key.ts b/packages/ai/src/providers/kiro-api-key.ts index 86b77f7d4ae..a155fcb041a 100644 --- a/packages/ai/src/providers/kiro-api-key.ts +++ b/packages/ai/src/providers/kiro-api-key.ts @@ -483,6 +483,188 @@ function findJsonEnd(text: string, start: number): number { return -1; } +type JsonPrefixResult = "complete" | "incomplete" | "invalid"; + +/** Return whether an unfinished object is still a valid JSON prefix. */ +function isValidJsonPrefix(text: string): boolean { + let position = 0; + const maxDepth = 128; + + function skipWhitespace(): void { + while (position < text.length) { + const char = text[position]; + if (char !== " " && char !== "\t" && char !== "\n" && char !== "\r") break; + position++; + } + } + + function isDigit(char: string | undefined): boolean { + return char !== undefined && char >= "0" && char <= "9"; + } + + function isHexDigit(char: string | undefined): boolean { + return ( + char !== undefined && + ((char >= "0" && char <= "9") || (char >= "a" && char <= "f") || (char >= "A" && char <= "F")) + ); + } + + function parseString(): JsonPrefixResult { + if (text[position] !== '"') return "invalid"; + position++; + while (position < text.length) { + const char = text[position]; + if (char === '"') { + position++; + return "complete"; + } + if (char === "\\") { + position++; + if (position >= text.length) return "incomplete"; + const escaped = text[position]; + if (escaped === "u") { + position++; + for (let i = 0; i < 4; i++) { + if (position >= text.length) return "incomplete"; + if (!isHexDigit(text[position])) return "invalid"; + position++; + } + continue; + } + if (escaped !== '"' && escaped !== "\\" && escaped !== "/" && !"bfnrt".includes(escaped ?? "")) { + return "invalid"; + } + position++; + continue; + } + if (char !== undefined && char.charCodeAt(0) < 0x20) return "invalid"; + position++; + } + return "incomplete"; + } + + function parseLiteral(literal: string): JsonPrefixResult { + for (const char of literal) { + if (position >= text.length) return "incomplete"; + if (text[position] !== char) return "invalid"; + position++; + } + return "complete"; + } + + function parseNumber(): JsonPrefixResult { + if (text[position] === "-") { + position++; + if (position >= text.length) return "incomplete"; + } + + if (text[position] === "0") { + position++; + if (isDigit(text[position])) return "invalid"; + } else if (isDigit(text[position]) && text[position] !== "0") { + while (isDigit(text[position])) position++; + } else { + return "invalid"; + } + + if (text[position] === ".") { + position++; + if (position >= text.length) return "incomplete"; + if (!isDigit(text[position])) return "invalid"; + while (isDigit(text[position])) position++; + } + + if (text[position] === "e" || text[position] === "E") { + position++; + if (position >= text.length) return "incomplete"; + if (text[position] === "+" || text[position] === "-") position++; + if (position >= text.length) return "incomplete"; + if (!isDigit(text[position])) return "invalid"; + while (isDigit(text[position])) position++; + } + return "complete"; + } + + function parseValue(depth: number): JsonPrefixResult { + if (depth > maxDepth) return "incomplete"; + skipWhitespace(); + if (position >= text.length) return "incomplete"; + + const char = text[position]; + if (char === "{") return parseObject(depth); + if (char === "[") return parseArray(depth); + if (char === '"') return parseString(); + if (char === "t") return parseLiteral("true"); + if (char === "f") return parseLiteral("false"); + if (char === "n") return parseLiteral("null"); + if (char === "-" || isDigit(char)) return parseNumber(); + return "invalid"; + } + + function parseObject(depth: number): JsonPrefixResult { + if (depth > maxDepth) return "incomplete"; + if (text[position] !== "{") return "invalid"; + position++; + skipWhitespace(); + if (position >= text.length) return "incomplete"; + if (text[position] === "}") { + position++; + return "complete"; + } + + while (position < text.length) { + const key = parseString(); + if (key !== "complete") return key; + skipWhitespace(); + if (position >= text.length) return "incomplete"; + if (text[position] !== ":") return "invalid"; + position++; + + const value = parseValue(depth + 1); + if (value !== "complete") return value; + skipWhitespace(); + if (position >= text.length) return "incomplete"; + if (text[position] === "}") { + position++; + return "complete"; + } + if (text[position] !== ",") return "invalid"; + position++; + skipWhitespace(); + } + return "incomplete"; + } + + function parseArray(depth: number): JsonPrefixResult { + if (depth > maxDepth) return "incomplete"; + if (text[position] !== "[") return "invalid"; + position++; + skipWhitespace(); + if (position >= text.length) return "incomplete"; + if (text[position] === "]") { + position++; + return "complete"; + } + + while (position < text.length) { + const value = parseValue(depth + 1); + if (value !== "complete") return value; + skipWhitespace(); + if (position >= text.length) return "incomplete"; + if (text[position] === "]") { + position++; + return "complete"; + } + if (text[position] !== ",") return "invalid"; + position++; + skipWhitespace(); + } + return "incomplete"; + } + + return parseObject(0) !== "invalid"; +} + export function parseKiroApiEvents(buffer: string): { events: KiroStreamEvent[]; remaining: string } { const events: KiroStreamEvent[] = []; let pos = 0; @@ -499,13 +681,13 @@ export function parseKiroApiEvents(buffer: string): { events: KiroStreamEvent[]; if (end < 0) { // Unclosed brace at 'start'. Could be incomplete JSON waiting for more data, or junk/corruption. const strayBraceEnd = Math.min(start + MAX_RESCAN_DISTANCE, buffer.length); - const firstPayloadCharacter = buffer.slice(start + 1).trimStart()[0]; - if (firstPayloadCharacter === undefined || firstPayloadCharacter === '"' || firstPayloadCharacter === "}") { - // A JSON object can begin only with a quoted key or close immediately. Preserve plausible - // incomplete envelopes; a complete nested child is not evidence that the outer frame is junk. - return { events, remaining: buffer.slice(start, strayBraceEnd) }; + const incompleteCandidate = buffer.slice(start, strayBraceEnd); + if (isValidJsonPrefix(incompleteCandidate)) { + // Preserve valid incomplete envelopes; a complete nested child is not evidence that the + // outer frame is junk until its own JSON prefix is syntactically impossible. + return { events, remaining: incompleteCandidate }; } - // The opener cannot begin a JSON object. Resync only after establishing invalid framing. + // The prefix is syntactically impossible. Resync only after establishing invalid framing. let resyncPos = start + 1; let found = false; while (resyncPos < strayBraceEnd) { @@ -827,10 +1009,11 @@ export const streamKiroApiKey: StreamFunction<"kiro-codewhisperer-stream"> = ( let currentTool: { id: string; name: string; input: string } | undefined; let toolComplete = false; // Track whether the current tool has been completed (has stop flag) const pendingToolCalls: Array<{ input: string; toolCall: ToolCall; index: number }> = []; - // Only an accepted successful completion confirms buffered text before clean EOF. + // Successful metadata fences later generation but text remains quarantined until clean EOF. let completionSeen = false; + let completionStopReason: "COMPLETED" | "TOOL_USE" | undefined; let textIndex: number | undefined; // Track text block index (accessible in catch) - let textContentConfirmed = false; // Track whether text has been committed to blocks (accessible in catch) + let textContentConfirmed = false; // Track whether clean EOF committed text to blocks (accessible in catch) let textStartDeferred = false; // Track if text_start is deferred (accessible in catch) let pendingTextContent = ""; // Text buffer (accessible in catch) const pendingTextEvents: Array<{ type: "start" | "delta" | "end"; delta?: string; content?: string }> = []; // Deferred text events (accessible in catch) @@ -994,8 +1177,8 @@ export const streamKiroApiKey: StreamFunction<"kiro-codewhisperer-stream"> = ( textIndex = blocks.length; // Create block with empty text (content will be added only after confirmation) blocks.push({ type: "text", text: "", index: textIndex }); - // ALWAYS defer text_start until we have terminal metadata or reach EOF without error. - // This ensures unconfirmed text is not exposed if an error occurs. + // ALWAYS defer text_start until clean EOF. Terminal metadata may be followed + // by a refusal in a later read, so it cannot safely publish text by itself. textStartDeferred = true; } // Buffer text content separately (don't add to blocks yet) @@ -1005,7 +1188,7 @@ export const streamKiroApiKey: StreamFunction<"kiro-codewhisperer-stream"> = ( pendingTextEvents.push({ type: "delta", delta }); }; - // Commit pending text to blocks (only safe at stream end when refusal is ruled out) + // Commit pending text to blocks only at clean EOF, when a later refusal is ruled out. const commitPendingText = () => { if (textContentConfirmed || pendingTextContent.length === 0) return; // Already committed or no text textContentConfirmed = true; @@ -1159,6 +1342,7 @@ export const streamKiroApiKey: StreamFunction<"kiro-codewhisperer-stream"> = ( for (const event of events) { const isActiveToolContinuation = currentTool !== undefined && + completionStopReason === "TOOL_USE" && !toolComplete && (event.type === "toolUseInput" || event.type === "toolUseStop" || @@ -1214,13 +1398,10 @@ export const streamKiroApiKey: StreamFunction<"kiro-codewhisperer-stream"> = ( if (event.data.outputTokens !== undefined) output.usage.output = event.data.outputTokens; output.usage.totalTokens = output.usage.input + output.usage.output; } else if (event.type === "completion" && !hasTerminalEvent) { - // Accepted successful terminal metadata confirms text, unlike usage accounting. + // Successful terminal metadata fences later generation. Keep text quarantined + // until clean EOF so a later refusal cannot depend on HTTP chunk boundaries. completionSeen = true; - // Finding #2: Immediately commit and emit all pending text events on completion - commitPendingText(); - emitThinking(); - emitDeferredTextEvents(); - closeTextBlock(); + completionStopReason = event.data.stopReason; } else if (event.type === "refusal") { const refusalData = event.data as { stopReason?: string; @@ -1271,37 +1452,15 @@ export const streamKiroApiKey: StreamFunction<"kiro-codewhisperer-stream"> = ( return; } } else if (event.type === "error" && !hasTerminalEvent) { - // On ordinary server errors (error event in stream), emit deferred text and tool events before error - // (refusals drop them, incomplete tools must not be emitted) - - // For server-sent errors, emit the deferred text first (it was received before the error) - // Usage is accounting metadata, not a successful completion boundary. - if (textContentConfirmed === false && completionSeen) { - // Commit pending text to blocks - if (textIndex !== undefined && textIndex < blocks.length && pendingTextContent.length > 0) { - textContentConfirmed = true; - } - } - if (textContentConfirmed) { - emitDeferredTextEvents(); - closeTextBlock(); - } + // An ordinary error is not clean EOF, so keep quarantined text private. - // Only emit the tool if it was explicitly completed (has stop flag) + // Preserve only tools that were explicitly completed (have a stop flag). if (toolComplete) addToolToBlocks(); removeUnconfirmedTextBlock(); emitPendingToolCalls(); - // Preserve text in the error context only after successful completion. - const accumulatedText = textContentConfirmed - ? blocks - .filter((b): b is TextContent => b.type === "text") - .map(b => b.text) - .join("") - : ""; // Discard unconfirmed pending text before error const errorMsg = sanitizeKiroError(`${event.data.error}: ${event.data.message ?? ""}`, apiKey); - const fullErrorMsg = accumulatedText ? `${errorMsg}\n\nPartial output: ${accumulatedText}` : errorMsg; - throw new Error(fullErrorMsg); + throw new Error(errorMsg); } } } @@ -1316,8 +1475,7 @@ export const streamKiroApiKey: StreamFunction<"kiro-codewhisperer-stream"> = ( } } - // Stream is done - commit pending text (safe since no refusal occurred), add pending tool, emit accumulated thinking, etc. - // First commit pending text to blocks since stream ended without refusal + // Clean EOF rules out a later refusal, so pending text can now be published. commitPendingText(); // Emit tool only if explicitly completed (has stop flag); incomplete tools are discarded at EOF if (currentTool && toolComplete) addToolToBlocks(); @@ -1358,20 +1516,14 @@ export const streamKiroApiKey: StreamFunction<"kiro-codewhisperer-stream"> = ( } catch (error) { // On ordinary errors (including reader.read() throws): // - Emit COMPLETED tool events (same semantics as in the event processing loop) - // - Emit confirmed text only after an accepted successful completion + // - Emit text only if clean EOF already confirmed it // - Incomplete currentTool is NOT finalized/emitted (only completed tools in pendingToolCalls flush). - // - Unconfirmed text is suppressed (usage alone does not confirm it) + // - Unconfirmed text is suppressed (terminal metadata alone does not rule out a later refusal) - // Emit confirmed text if we received an accepted successful completion. - if (completionSeen && textContentConfirmed === false) { - // Commit pending text to blocks - if (textIndex !== undefined && textIndex < blocks.length && pendingTextContent.length > 0) { - textContentConfirmed = true; - } - // Emit deferred text start if not yet emitted - if (textStartDeferred && textIndex !== undefined && textIndex < blocks.length) { - const block = blocks[textIndex]; - if (block && block.type === "text" && pendingTextContent.length > 0) { + if (textContentConfirmed && textIndex !== undefined && textIndex < blocks.length) { + const block = blocks[textIndex]; + if (block && block.type === "text") { + if (textStartDeferred && pendingTextContent.length > 0) { stream.push({ type: "text_start", contentIndex: textIndex, partial: snapshotOutput() }); // Emit all pending text_delta events for (const evt of pendingTextEvents) { @@ -1384,19 +1536,19 @@ export const streamKiroApiKey: StreamFunction<"kiro-codewhisperer-stream"> = ( }); } } - block.text = pendingTextContent; - stream.push({ - type: "text_end", - contentIndex: textIndex, - content: block.text, - partial: snapshotOutput(), - }); - textIndex = undefined; - pendingTextContent = ""; - pendingTextEvents.length = 0; } - textStartDeferred = false; + if (block.text.length === 0 && pendingTextContent.length > 0) block.text = pendingTextContent; + stream.push({ + type: "text_end", + contentIndex: textIndex, + content: block.text, + partial: snapshotOutput(), + }); } + textIndex = undefined; + pendingTextContent = ""; + pendingTextEvents.length = 0; + textStartDeferred = false; } removeUnconfirmedTextBlock(); diff --git a/packages/ai/test/kiro-api-key-content-filter.test.ts b/packages/ai/test/kiro-api-key-content-filter.test.ts index 228e78e58ec..1a51e1aea4f 100644 --- a/packages/ai/test/kiro-api-key-content-filter.test.ts +++ b/packages/ai/test/kiro-api-key-content-filter.test.ts @@ -1152,6 +1152,15 @@ describe("Regression tests for #6151 issues", () => { expect(contentEvent?.type).toBe("content"); }); + test("malformed quoted envelopes resync to a valid nested event", () => { + const result = parseKiroApiEvents('{"bad":nope{"content":"recoverable content"}'); + + expect(result.events.map(event => event.type)).toEqual(["content"]); + const contentEvent = result.events[0]; + expect(contentEvent?.type === "content" ? contentEvent.data : undefined).toBe("recoverable content"); + expect(result.remaining).toBe(""); + }); + test("#4199034439: incomplete tool should NOT be emitted when reader.read() throws mid-tool", async () => { const emittedEventTypes: string[] = []; @@ -1442,6 +1451,40 @@ describe("P1 Regression: incomplete tool emission at tool-ID rollover", () => { ).toBe(false); expectToolEventsReferenceToolBlocks(events); }); + + test("COMPLETED fences unfinished tool fragments", async () => { + const events: AssistantMessageEvent[] = []; + const eventSequence = [ + JSON.stringify({ toolUseId: "tool-incomplete", name: "search", input: '{"query":' }), + JSON.stringify({ stopReason: "COMPLETED" }), + JSON.stringify({ input: '"kiro"}' }), + JSON.stringify({ stop: true }), + ]; + let emitted = 0; + + globalThis.fetch = (async () => { + const body = new ReadableStream({ + pull(controller) { + if (emitted < eventSequence.length) { + controller.enqueue(new TextEncoder().encode(eventSequence[emitted])); + emitted++; + } else { + controller.close(); + } + }, + }); + return new Response(body, { status: 200 }); + }) as unknown as typeof fetch; + + const stream = streamKiroApiKey(model, context, { apiKey: "ksk_test-secret", region: "us-east-1" }); + for await (const event of stream) { + events.push(event); + } + + const errorEvent = events.find(event => event.type === "error"); + expect(errorEvent).toBeDefined(); + expect(events.some(event => event.type === "toolcall_start" || event.type === "toolcall_end")).toBe(false); + }); }); describe("P2 Probepark findings: unconfirmed text on read rejection", () => { @@ -1565,7 +1608,7 @@ describe("P2 Probepark findings: unconfirmed text on read rejection", () => { expect(emittedEvents.some(event => event.type === "text_delta")).toBe(false); }); - test("confirmed text lifecycle closes before an ordinary error in the same batch", async () => { + test("completed text remains quarantined on an ordinary error before clean EOF", async () => { const events: AssistantMessageEvent[] = []; const responseBody = JSON.stringify({ content: "confirmed before error" }) + @@ -1587,14 +1630,14 @@ describe("P2 Probepark findings: unconfirmed text on read rejection", () => { events.push(event); } - const errorIndex = events.findIndex(event => event.type === "error"); - const textLifecycle = events - .slice(0, errorIndex) - .filter(event => event.type === "text_start" || event.type === "text_delta" || event.type === "text_end"); - expect(errorIndex).toBeGreaterThan(-1); - expect(textLifecycle.map(event => event.type)).toEqual(["text_start", "text_delta", "text_end"]); - const textEnd = textLifecycle.find(event => event.type === "text_end"); - expect(textEnd?.type === "text_end" ? textEnd.content : undefined).toBe("confirmed before error"); + const errorEvent = events.find( + (event): event is Extract => event.type === "error", + ); + expect(errorEvent?.error.errorMessage).toContain("failure after completion"); + expect(getTextBlocks(errorEvent?.error)).toHaveLength(0); + expect( + events.some(event => event.type === "text_start" || event.type === "text_delta" || event.type === "text_end"), + ).toBe(false); expect(events.some(event => event.type === "done")).toBe(false); }); @@ -1758,6 +1801,47 @@ describe("P1 Regression: content leaking across batches", () => { ).toBe(false); }); + test("a refusal in a later read suppresses text after COMPLETED", async () => { + const events: AssistantMessageEvent[] = []; + const eventSequence = [ + JSON.stringify({ content: "filtered across reads" }), + JSON.stringify({ stopReason: "COMPLETED" }), + JSON.stringify({ + stopReason: "CONTENT_FILTERED", + stopDetails: { refusal: { category: "VIOLENCE", explanation: "Do not show answer" } }, + }), + ]; + let emitted = 0; + + globalThis.fetch = (async () => { + const body = new ReadableStream({ + pull(controller) { + if (emitted < eventSequence.length) { + controller.enqueue(new TextEncoder().encode(eventSequence[emitted])); + emitted++; + } else { + controller.close(); + } + }, + }); + return new Response(body, { status: 200 }); + }) as unknown as typeof fetch; + + const stream = streamKiroApiKey(model, context, { apiKey: "ksk_test-secret", region: "us-east-1" }); + for await (const event of stream) { + events.push(event); + } + + const errorEvent = events.find( + (event): event is Extract => event.type === "error", + ); + expect(errorEvent?.error.errorMessage).toContain("Kiro refused the request (VIOLENCE)"); + expect(errorEvent?.error.content).toHaveLength(0); + expect( + events.some(event => event.type === "text_start" || event.type === "text_delta" || event.type === "text_end"), + ).toBe(false); + }); + test("batched refusal takes precedence over an earlier ordinary error", async () => { const events: AssistantMessageEvent[] = []; const trustedModel = kiroApiStaticModels()[0]; @@ -1926,9 +2010,8 @@ describe("Completion event handling (probepark fix for #6451)", () => { expect(done || error).toBe(true); }); - test("content + COMPLETED, then stream error: text block in error terminal", async () => { - // When stream contains content followed by COMPLETED event, then reader.read() throws, - // the error terminal should include the confirmed text block. + test("content + COMPLETED, then stream error: quarantined text stays private", async () => { + // Completion metadata is not enough to publish text when the response fails before clean EOF. const emittedEvents: AssistantMessageEvent[] = []; globalThis.fetch = (async () => { @@ -1966,7 +2049,12 @@ describe("Completion event handling (probepark fix for #6451)", () => { (event): event is Extract => event.type === "error", ); expect(errorEvent?.error.errorMessage).toContain("Network error after completion"); - expect(getTextBlocks(errorEvent?.error).map(block => block.text)).toEqual(["Hello"]); + expect(getTextBlocks(errorEvent?.error)).toHaveLength(0); + expect( + emittedEvents.some( + event => event.type === "text_start" || event.type === "text_delta" || event.type === "text_end", + ), + ).toBe(false); }); test("content without COMPLETED, then stream error: no text block in error terminal", async () => { @@ -2064,9 +2152,8 @@ describe("Completion event handling (probepark fix for #6451)", () => { expect(getTextBlocks(errorEvent?.error)).toHaveLength(0); }); - test("completion event emits all pending text events immediately (Finding #2)", async () => { - // When completion event arrives, all pending text events should be emitted immediately, - // not deferred until EOF. This includes text_start, text_delta, and text_end. + test("clean EOF publishes completion-confirmed text and fences later content", async () => { + // Completion metadata fences later content, but text is not published until EOF rules out a later refusal. const emittedEvents: AssistantMessageEvent[] = []; globalThis.fetch = (async () => { @@ -2102,17 +2189,13 @@ describe("Completion event handling (probepark fix for #6451)", () => { // Stream error handling } - // Verify text_start, text_delta, and text_end are emitted + // Clean EOF publishes the confirmed text as a complete lifecycle. const eventTypes = emittedEvents.map(e => e.type); - const hasTextStart = eventTypes.includes("text_start"); - const hasTextDelta = eventTypes.includes("text_delta"); - const hasTextEnd = eventTypes.includes("text_end"); - const hasDone = eventTypes.includes("done"); - - expect(hasTextStart).toBe(true); // text_start must be emitted - expect(hasTextDelta).toBe(true); // text_delta must be emitted - expect(hasTextEnd).toBe(true); // text_end must be emitted after completion - expect(hasDone).toBe(true); // done event must follow + const textLifecycle = emittedEvents.filter( + event => event.type === "text_start" || event.type === "text_delta" || event.type === "text_end", + ); + expect(textLifecycle.map(event => event.type)).toEqual(["text_start", "text_delta", "text_end"]); + expect(eventTypes.includes("done")).toBe(true); // Verify the order: text events should come before done const textStartIdx = eventTypes.indexOf("text_start"); @@ -2128,6 +2211,7 @@ describe("Completion event handling (probepark fix for #6451)", () => { (event): event is Extract => event.type === "done", ); expect(getTextBlocks(doneEvent?.message).map(block => block.text)).toEqual(["Hello world"]); + expect(doneEvent?.message.content.filter(block => block.type === "text")).toHaveLength(1); }); test("incomplete JSON frame with nested object: no resync into nested (Finding #3)", async () => { @@ -2211,10 +2295,8 @@ describe("Completion event handling (probepark fix for #6451)", () => { ).toBe(false); }); - test("content + COMPLETED, then error event: text_end emitted before error (Finding #4)", async () => { - // Finding #4: When a server error event is received after COMPLETED, - // all text events including text_end should be emitted before the error. - const emittedEvents: Array<{ type: string }> = []; + test("post-completion error in a later read still suppresses text before clean EOF", async () => { + const emittedEvents: AssistantMessageEvent[] = []; globalThis.fetch = (async () => { const eventSequence = [ @@ -2242,25 +2324,21 @@ describe("Completion event handling (probepark fix for #6451)", () => { try { const stream = streamKiroApiKey(model, context, { apiKey: "ksk_test-secret", region: "us-east-1" }); for await (const event of stream) { - emittedEvents.push({ type: event.type }); + emittedEvents.push(event); } } catch { // Stream error handling } - // Verify event sequence includes text_end before error - const eventTypes = emittedEvents.map(e => e.type); - const hasTextStart = eventTypes.includes("text_start"); - const hasTextEnd = eventTypes.includes("text_end"); - const hasError = eventTypes.includes("error"); - - expect(hasTextStart).toBe(true); // text_start should be emitted - expect(hasTextEnd).toBe(true); // text_end should be emitted before error - expect(hasError).toBe(true); // error event follows - - // Verify order: text_end comes before error - const textEndIdx = eventTypes.indexOf("text_end"); - const errorIdx = eventTypes.indexOf("error"); - expect(textEndIdx).toBeLessThan(errorIdx); + const errorEvent = emittedEvents.find( + (event): event is Extract => event.type === "error", + ); + expect(errorEvent?.error.errorMessage).toContain("Internal server error"); + expect(getTextBlocks(errorEvent?.error)).toHaveLength(0); + expect( + emittedEvents.some( + event => event.type === "text_start" || event.type === "text_delta" || event.type === "text_end", + ), + ).toBe(false); }); });