diff --git a/open-sse/executors/kiro.js b/open-sse/executors/kiro.js
index 556e0c7293..90ec713e6f 100644
--- a/open-sse/executors/kiro.js
+++ b/open-sse/executors/kiro.js
@@ -5,6 +5,207 @@ import { v4 as uuidv4 } from "uuid";
import { refreshKiroToken } from "../services/tokenRefresh.js";
import { SSE_DONE, SSE_HEADERS } from "../utils/sseConstants.js";
import { getCapabilitiesForModel } from "../providers/capabilities.js";
+import { STREAM_FIRST_CHUNK_TIMEOUT_MS } from "../config/runtimeConfig.js";
+
+const KIRO_REPAIR_BUFFER_MAX_BYTES = 8 * 1024 * 1024;
+const KIRO_REPAIR_HEARTBEAT_MS = 10_000;
+const KIRO_SHORT_FINAL_MAX_CHARS = 800;
+const EVENTSTREAM_MAX_MESSAGE_BYTES = 24 * 1024 * 1024;
+const EVENTSTREAM_MAX_HEADERS_BYTES = 128 * 1024;
+const KIRO_EVENT_TYPES = new Set([
+ "assistantResponseEvent",
+ "reasoningContentEvent",
+ "codeEvent",
+ "toolUseEvent",
+ "messageStopEvent",
+ "metadataEvent",
+ "MetadataEvent",
+ "contextUsageEvent",
+ "meteringEvent",
+ "metricsEvent"
+]);
+const encoder = new TextEncoder();
+const decoder = new TextDecoder();
+const CRC32_TABLE = Uint32Array.from({ length: 256 }, (_, index) => {
+ let value = index;
+ for (let bit = 0; bit < 8; bit++) {
+ value = (value >>> 1) ^ ((value & 1) ? 0xedb88320 : 0);
+ }
+ return value >>> 0;
+});
+
+const REPAIR_INSTRUCTIONS = Object.freeze({
+ tool: "Retry the previous response because its Kiro tool_call wrapper was malformed. If you use the wrapper tool named tool_call, its input must contain a non-empty name and an arguments field.",
+ ellipsis: "Retry the previous response because it ended with only an ellipsis. Return the complete final answer, not only ... or ….",
+ short_final: "Retry the previous response because its final only announced a future action. Complete the check now and return the result or a concrete blocker."
+});
+const SHORT_FUTURE_ACTION = /^(?:(?:(?:現在|接著|接下來|下一步)[，,:：\s]*(?:我(?:只)?(?:會|要|將|再)?\s*)?|我只再)(?:補|查|確認|驗證|追(?:查|蹤)?|繼續|檢查|測試)|我(?:會|要|將)(?:再|重新)?(?:補(?:齊|查)?|抓取|查(?:詢)?|確認|驗證|追(?:查|蹤)?|繼續|檢查|測試)|(?:(?:next|now|then)\b[\s,:-]*)?(?:i(?:'ll| will| am going to| need to)|let me)\s+(?:verify|check|confirm|validate|investigate|trace|continue|follow up|test)\b)/iu;
+// Keep this tied to the observed whole-response signature. Broader Chinese
+// result/progress heuristics create false positives for completed findings.
+const OBSERVED_TRAILING_FUTURE_ACTION = /^目前證據顯示[\s\S]{1,700}[。.!?；;]\s*最後補查\s+504\s+access\s+log[，,]\s*確認\s+host[／/]路徑與是否為集中流量[。.!]?$/iu;
+const ENGLISH_FUTURE_ACTION = /^(?:(?:next|now|then)\b[\s,:-]*)?(?:i(?:'ll| will| am going to| need to)|let me)\s+(?:verify|check|confirm|validate|investigate|trace|continue|follow up|test)\b/iu;
+const ENGLISH_RESULT_CLAUSE = /(?:[:;\n]|[.!?]\s+\S|\b(?:status|checksum|response|deployment)\s+(?:is|are|was|were|matches?|equals?|returned)\b)/iu;
+const CHINESE_FUTURE_ACTION = /^(?:(?:現在|接著|接下來|下一步)[，,:：\s]*(?:我(?:只)?(?:會|要|將|再)?\s*)?|我只再|我(?:會|要|將)(?:再|重新)?)(?:補|抓取|查|確認|驗證|追|繼續|檢查|測試)/u;
+const CHINESE_RESULT_CLAUSE = /(?:[。！？]\s*\S|(?:版本|狀態|回應|結果|部署|校驗碼)(?:是|為|等於|顯示))/u;
+const USER_WAIT = /(?:請(?:你|先)|你(?:先|需要|可以|提供|確認|批准|允許)|等待(?:你|使用者)|等你|核准|同意|授權|\b(?:after|when|once)\s+you\b|\byour\s+(?:approval|confirmation|permission|input)\b|\bwait(?:ing)?\s+for\s+you\b|\bplease\s+(?:approve|confirm|provide|send)\b)/iu;
+const COMPLETED_FINAL = /(?:已(?:經)?完成|完成(?:了|驗證|確認)|修復完成|確認無誤|驗證(?:完成|通過)|測試(?:均)?通過|結論|總結|\b(?:done|completed|fixed|verified|confirmed|passed|in conclusion|summary)\b|\b(?:is|are) complete\b)/iu;
+const RESULT_EVIDENCE = /(?:顯示|發現|因此|成功|失敗|正常|無錯誤|沒有錯誤|\b(?:found|shows?|showed|because|therefore|succeeded|failed|healthy|green|no errors?)\b)/iu;
+
+function crc32(bytes) {
+ let crc = 0xffffffff;
+ for (const byte of bytes) crc = CRC32_TABLE[(crc ^ byte) & 0xff] ^ (crc >>> 8);
+ return (crc ^ 0xffffffff) >>> 0;
+}
+
+function envPositiveInt(name, fallback) {
+ const parsed = Number.parseInt(process.env?.[name] || "", 10);
+ return Number.isFinite(parsed) && parsed > 0 ? parsed : fallback;
+}
+
+function concatChunks(chunks, totalBytes) {
+ const output = new Uint8Array(totalBytes);
+ let offset = 0;
+ for (const chunk of chunks) {
+ output.set(chunk, offset);
+ offset += chunk.byteLength;
+ }
+ return output;
+}
+
+function makeAbortError(reason) {
+ const error = new Error(reason?.message || reason || "Request aborted");
+ error.name = "AbortError";
+ return error;
+}
+
+async function readWithTimeout(reader, signal, timeoutMs, message) {
+ if (signal?.aborted) throw makeAbortError(signal.reason);
+ let timeout;
+ let abortHandler;
+ const timeoutPromise = new Promise((_, reject) => {
+ timeout = setTimeout(() => reject(new Error(message)), timeoutMs);
+ });
+ const abortPromise = new Promise((_, reject) => {
+ abortHandler = () => reject(makeAbortError(signal.reason));
+ signal?.addEventListener("abort", abortHandler, { once: true });
+ });
+ try {
+ return await Promise.race([reader.read(), timeoutPromise, abortPromise]);
+ } finally {
+ clearTimeout(timeout);
+ signal?.removeEventListener?.("abort", abortHandler);
+ }
+}
+
+async function readResponsePrefix(response, signal, maxBytes, timeoutMs) {
+ const reader = response?.body?.getReader?.();
+ if (!reader) return "";
+ const chunks = [];
+ let totalBytes = 0;
+ try {
+ while (totalBytes < maxBytes) {
+ const { done, value } = await readWithTimeout(
+ reader,
+ signal,
+ timeoutMs,
+ "Kiro retry error body stalled"
+ );
+ if (done) break;
+ const remaining = maxBytes - totalBytes;
+ const chunk = value.byteLength > remaining ? value.slice(0, remaining) : value;
+ chunks.push(chunk);
+ totalBytes += chunk.byteLength;
+ if (value.byteLength > remaining) break;
+ }
+ } finally {
+ await reader.cancel("bounded Kiro retry error body").catch(() => {});
+ }
+ return decoder.decode(concatChunks(chunks, totalBytes));
+}
+
+function appendRepairInstruction(body, kind) {
+ const repaired = structuredClone(body || {});
+ const instruction = REPAIR_INSTRUCTIONS[kind] || "Retry the previous incomplete Kiro response.";
+ repaired.systemPrompt = repaired.systemPrompt
+ ? `${repaired.systemPrompt}\n\n${instruction}`
+ : instruction;
+ return repaired;
+}
+
+function normalizeStopReason(value) {
+ const reason = String(value || "").trim().replace(/([a-z])([A-Z])/g, "$1_$2").toLowerCase().replace(/[\s-]+/g, "_");
+ if (["endturn", "end_turn", "stop", "stop_sequence"].includes(reason)) return "end_turn";
+ if (["tooluse", "tool_use", "tool_calls"].includes(reason)) return "tool_use";
+ if (["maxtokens", "max_tokens", "max_output_tokens", "length"].includes(reason)) return "max_tokens";
+ return reason || null;
+}
+
+function stopDisposition(stopReason, hasToolCalls) {
+ if (["malformed_model_output", "invalid_model_output"].includes(stopReason)) return "retryable_protocol_failure";
+ if (["cancelled", "pause_turn", "model_context_window_exceeded"].includes(stopReason)) return "terminal_incomplete";
+ if (stopReason === "refusal" || /(?:content.*filter|guardrail|safety|policy|blocked)/u.test(stopReason)) return "terminal_refusal";
+ if (stopReason === "max_tokens") return hasToolCalls ? "terminal_incomplete" : "length";
+ if (stopReason && !["end_turn", "tool_use"].includes(stopReason)) return "unknown_failure";
+ if (hasToolCalls || stopReason === "tool_use") return "tool_use";
+ if (!stopReason || stopReason === "end_turn") return "complete";
+ return "unknown_failure";
+}
+
+function mergeStopReason(current, incoming) {
+ if (!incoming) return current;
+ if (!current) return incoming;
+ const severity = (reason) => {
+ const disposition = stopDisposition(reason, false);
+ if (disposition === "terminal_refusal") return 6;
+ if (disposition === "terminal_incomplete") return 5;
+ if (disposition === "unknown_failure") return 4;
+ if (disposition === "retryable_protocol_failure") return 3;
+ if (disposition === "length") return 2;
+ return 1;
+ };
+ return severity(incoming) > severity(current) ? incoming : current;
+}
+
+function isEllipsisOnly(value) {
+ return ["...", "…"].includes(String(value || "").trim());
+}
+
+function isShortFutureAction(value) {
+ const text = String(value || "").trim().replaceAll("’", "'");
+ if (OBSERVED_TRAILING_FUTURE_ACTION.test(text)) return true;
+ if (ENGLISH_FUTURE_ACTION.test(text) && ENGLISH_RESULT_CLAUSE.test(text)) return false;
+ if (CHINESE_FUTURE_ACTION.test(text) && CHINESE_RESULT_CLAUSE.test(text)) return false;
+ return text.length > 0 && text.length <= KIRO_SHORT_FINAL_MAX_CHARS &&
+ SHORT_FUTURE_ACTION.test(text) && !USER_WAIT.test(text) &&
+ !COMPLETED_FINAL.test(text) && !RESULT_EVIDENCE.test(text);
+}
+
+function encodeSSEError(code, message, details) {
+ return encoder.encode(`data: ${JSON.stringify({ error: {
+ message,
+ type: "upstream_error",
+ code,
+ ...(details ? { details } : {})
+ } })}\n\ndata: [DONE]\n\n`);
+}
+
+function inspectSSEChunk(chunk, state) {
+ for (const line of decoder.decode(chunk).split("\n")) {
+ if (!line.startsWith("data: ")) continue;
+ const data = line.slice(6).trim();
+ if (!data || data === "[DONE]") continue;
+ try {
+ const event = JSON.parse(data);
+ if (event.error) state.error = event.error;
+ for (const choice of event.choices || []) {
+ const delta = choice.delta || {};
+ if (typeof delta.content === "string") state.content += delta.content;
+ if (typeof delta.reasoning_content === "string") state.reasoning += delta.reasoning_content;
+ if (delta.tool_calls?.length) state.hasToolCalls = true;
+ }
+ } catch { /* a malformed SSE line is diagnosed by the transformer */ }
+ }
+}
/**
* KiroExecutor - Executor for Kiro AI (AWS CodeWhisperer)
@@ -111,393 +312,776 @@ export class KiroExecutor extends BaseExecutor {
*/
async execute(args) {
const result = await super.execute(args);
- if (result?.response?.ok) {
- result.response = this.transformEventStreamToSSE(result.response, args.model);
- }
+ if (result?.response?.ok) this.attachIntegrityGate(result, args);
return result;
}
- /**
- * Transform AWS EventStream binary response to SSE text stream
- * Using TransformStream instead of ReadableStream.pull() to avoid Workers timeout
- */
- transformEventStreamToSSE(response, model) {
- let buffer = new Uint8Array(0);
- let chunkIndex = 0;
+ attachIntegrityGate(result, args) {
+ const abortController = new AbortController();
+ const maxBytes = envPositiveInt("KIRO_TOOL_CALL_REPAIR_BUFFER_MAX_BYTES", KIRO_REPAIR_BUFFER_MAX_BYTES);
+ const legacyTimeout = envPositiveInt("KIRO_TOOL_CALL_REPAIR_TIMEOUT_MS", STREAM_FIRST_CHUNK_TIMEOUT_MS);
+ const ttftTimeoutMs = envPositiveInt("KIRO_TOOL_CALL_REPAIR_TTFT_TIMEOUT_MS", legacyTimeout);
+ const stallTimeoutMs = envPositiveInt("KIRO_TOOL_CALL_REPAIR_STALL_TIMEOUT_MS", legacyTimeout);
+ const repairEnabled = args.credentials?.providerSpecificData?.kiroToolCallRepair !== false &&
+ process.env.KIRO_TOOL_CALL_REPAIR !== "false";
+ const forwardAbort = () => abortController.abort(args.signal?.reason);
+ args.signal?.addEventListener("abort", forwardAbort, { once: true });
+ let open = true;
+ let heartbeatTimer;
+
+ const stream = new ReadableStream({
+ start: async (controller) => {
+ const heartbeat = () => {
+ if (!open) return;
+ try {
+ controller.enqueue(encoder.encode(": kiro-validation\n\n"));
+ } catch {
+ open = false;
+ }
+ };
+ heartbeat();
+ heartbeatTimer = setInterval(heartbeat, KIRO_REPAIR_HEARTBEAT_MS);
+
+ try {
+ const bytes = await this.runIntegrityRecovery(result.response, args, {
+ signal: abortController.signal,
+ maxBytes,
+ ttftTimeoutMs,
+ stallTimeoutMs,
+ repairEnabled
+ });
+ if (abortController.signal.aborted) throw makeAbortError(abortController.signal.reason);
+ controller.enqueue(bytes);
+ controller.close();
+ } catch (error) {
+ if (open && error.name === "AbortError") {
+ controller.error(error);
+ } else if (open && error.name !== "AbortError") {
+ controller.enqueue(encodeSSEError(
+ "kiro_integrity_gate_failed",
+ error.message || "Kiro integrity validation failed"
+ ));
+ controller.close();
+ }
+ } finally {
+ open = false;
+ clearInterval(heartbeatTimer);
+ args.signal?.removeEventListener?.("abort", forwardAbort);
+ }
+ },
+ cancel(reason) {
+ open = false;
+ clearInterval(heartbeatTimer);
+ abortController.abort(reason || "client cancelled");
+ }
+ });
+
+ result.response = new Response(stream, {
+ status: result.response.status,
+ statusText: result.response.statusText,
+ headers: { ...SSE_HEADERS }
+ });
+ }
+
+ async runIntegrityRecovery(rawResponse, args, options) {
+ const first = await this.readRecoverableIntegrityAttempt(
+ rawResponse,
+ args.model,
+ options,
+ "initial"
+ );
+ if (first.kind === "complete") return first.bytes;
+ if (first.kind === "terminal_stop" || first.kind === "upstream_error") {
+ return this.integrityFailureSSE(first);
+ }
+ if (first.kind === "invalid_tool" && !options.repairEnabled) {
+ return encodeSSEError("invalid_kiro_tool_call", first.message, first.diagnostics);
+ }
+
+ const repairKind = ["ellipsis", "short_final", "invalid_tool"].includes(first.kind)
+ ? first.kind
+ : null;
+ const repairBody = repairKind
+ ? appendRepairInstruction(args.body, repairKind === "invalid_tool" ? "tool" : repairKind)
+ : structuredClone(args.body || {});
+
+ const retry = await BaseExecutor.prototype.execute.call(this, {
+ ...args,
+ body: repairBody,
+ signal: options.signal
+ });
+ if (!retry?.response?.ok) {
+ let body = "";
+ try {
+ body = await readResponsePrefix(
+ retry?.response,
+ options.signal,
+ Math.min(options.maxBytes, 4096),
+ options.stallTimeoutMs
+ );
+ } catch (error) {
+ if (error.name === "AbortError") throw error;
+ }
+ return encodeSSEError(
+ "kiro_integrity_retry_upstream_error",
+ body || `Kiro integrity retry failed with HTTP ${retry?.response?.status || 502}`,
+ { status: retry?.response?.status || 502 }
+ );
+ }
+
+ const second = await this.readRecoverableIntegrityAttempt(
+ retry.response,
+ args.model,
+ options,
+ "retry"
+ );
+ if (second.kind === "complete") return second.bytes;
+ if (second.kind === "terminal_stop" || second.kind === "upstream_error") {
+ return this.integrityFailureSSE(second);
+ }
+ const code = second.kind === "ellipsis"
+ ? "kiro_ellipsis_retry_failed"
+ : second.kind === "short_final"
+ ? "kiro_short_final_retry_failed"
+ : second.kind === "invalid_tool"
+ ? "kiro_tool_call_repair_retry_failed"
+ : "kiro_missing_terminal_retry_failed";
+ return encodeSSEError(
+ code,
+ `Kiro integrity validation failed after one bounded retry: ${second.message || second.kind}`,
+ { attempts: [first.diagnostics, second.diagnostics].filter(Boolean) }
+ );
+ }
+
+ integrityFailureSSE(attempt) {
+ const disposition = attempt.diagnostics?.stop_disposition;
+ const code = attempt.diagnostics?.terminal_provenance === "integrity_buffer_exceeded"
+ ? "kiro_integrity_buffer_exceeded"
+ : attempt.kind === "upstream_error"
+ ? "kiro_upstream_eventstream_error"
+ : disposition === "terminal_refusal"
+ ? "kiro_terminal_refusal"
+ : disposition === "terminal_incomplete"
+ ? "kiro_terminal_incomplete"
+ : "kiro_unknown_stop_reason";
+ return encodeSSEError(code, attempt.message || "Kiro stream ended with a terminal failure", attempt.diagnostics);
+ }
+
+ async readRecoverableIntegrityAttempt(rawResponse, model, options, attempt) {
+ try {
+ return await this.readIntegrityAttempt(rawResponse, model, options, attempt);
+ } catch (error) {
+ if (error.name === "AbortError") throw error;
+ return {
+ kind: "missing_terminal",
+ message: error.message || "Kiro transport read failed",
+ diagnostics: {
+ attempt,
+ terminal_provenance: "transport_read_error",
+ transport_state: "upstream_error",
+ stop_reason: null,
+ stop_disposition: "terminal_incomplete",
+ response_state: "no_semantic_output",
+ event_counts: {},
+ incomplete_frame_bytes: 0
+ }
+ };
+ }
+ }
+
+ async readIntegrityAttempt(rawResponse, model, options, attempt) {
+ let diagnostics;
+ const transformed = this.transformEventStreamToSSE(rawResponse, model, {
+ maxToolBytes: Math.max(1, Math.floor(options.maxBytes / 2)),
+ onTerminalState: (value) => {
+ diagnostics = value;
+ }
+ });
+ const reader = transformed.body.getReader();
+ const chunks = [];
+ let totalBytes = 0;
+ let sawChunk = false;
+ const output = { content: "", reasoning: "", hasToolCalls: false, error: null };
+
+ try {
+ while (true) {
+ const timeoutMs = sawChunk ? options.stallTimeoutMs : options.ttftTimeoutMs;
+ const phase = sawChunk ? "stalled" : "timed out before first chunk";
+ const { done, value } = await readWithTimeout(
+ reader,
+ options.signal,
+ timeoutMs,
+ `Kiro integrity validation ${phase}`
+ );
+ if (done) break;
+ sawChunk = true;
+ totalBytes += value.byteLength;
+ if (totalBytes > options.maxBytes) {
+ await reader.cancel("kiro_integrity_buffer_exceeded").catch(() => {});
+ return {
+ kind: "terminal_stop",
+ message: `Kiro integrity buffer exceeded ${options.maxBytes} bytes`,
+ diagnostics: { terminal_provenance: "integrity_buffer_exceeded" }
+ };
+ }
+ chunks.push(value);
+ inspectSSEChunk(value, output);
+ }
+ } catch (error) {
+ await reader.cancel(error.message).catch(() => {});
+ throw error;
+ }
+
+ const safeDiagnostics = {
+ attempt,
+ terminal_provenance: diagnostics?.terminal_provenance || "missing_terminal_diagnostics",
+ transport_state: diagnostics?.transport_state || "unknown",
+ stop_reason: diagnostics?.stop_reason || null,
+ stop_disposition: diagnostics?.stop_disposition || "terminal_incomplete",
+ response_state: diagnostics?.response_state || "no_semantic_output",
+ event_counts: diagnostics?.event_counts || {},
+ incomplete_frame_bytes: diagnostics?.incomplete_frame_bytes || 0
+ };
+ if (safeDiagnostics.stop_disposition === "retryable_protocol_failure") {
+ const kind = safeDiagnostics.terminal_provenance === "invalid_tool_call"
+ ? "invalid_tool"
+ : "retryable_stop";
+ return { kind, message: output.error?.message, diagnostics: safeDiagnostics };
+ }
+ if (safeDiagnostics.stop_disposition === "terminal_incomplete" ||
+ safeDiagnostics.stop_disposition === "terminal_refusal" ||
+ safeDiagnostics.stop_disposition === "unknown_failure") {
+ const kind = safeDiagnostics.terminal_provenance === "upstream_eventstream_error"
+ ? "upstream_error"
+ : safeDiagnostics.terminal_provenance === "integrity_buffer_exceeded"
+ ? "terminal_stop"
+ : ["metadata_stop_reason", "message_stop_event"].includes(safeDiagnostics.terminal_provenance)
+ ? "terminal_stop"
+ : "missing_terminal";
+ return { kind, message: output.error?.message, diagnostics: safeDiagnostics };
+ }
+ if (output.error) {
+ return { kind: "missing_terminal", message: output.error.message, diagnostics: safeDiagnostics };
+ }
+ if (!output.hasToolCalls) {
+ if (isEllipsisOnly(output.content) ||
+ (!output.content.trim() && isEllipsisOnly(output.reasoning))) {
+ return { kind: "ellipsis", diagnostics: safeDiagnostics };
+ }
+ if (isShortFutureAction(output.content)) {
+ return { kind: "short_final", diagnostics: safeDiagnostics };
+ }
+ }
+ return { kind: "complete", bytes: concatChunks(chunks, totalBytes), diagnostics: safeDiagnostics };
+ }
+
+ transformEventStreamToSSE(response, model, options = {}) {
const responseId = `chatcmpl-${Date.now()}`;
const created = Math.floor(Date.now() / 1000);
const capabilityModel = resolveKiroModel(model).upstream;
const contextWindow = getCapabilitiesForModel("kiro", capabilityModel).contextWindow || 200000;
+ const eventCounts = {};
const state = {
- endDetected: false,
- finishEmitted: false,
+ buffer: new Uint8Array(0),
+ chunkIndex: 0,
+ toolCounter: 0,
+ tools: new Map(),
+ bufferedToolBytes: 0,
+ hasText: false,
+ hasReasoning: false,
+ hasCode: false,
hasToolCalls: false,
- hasReasoningContent: false,
- reasoningChunkCount: 0,
- toolCallIndex: 0,
- seenToolIds: new Map(),
- inThinking: false
+ sawToolUse: false,
+ explicitStop: false,
+ stopReason: null,
+ terminalProvenance: null,
+ transportState: "consuming_response",
+ totalContentLength: 0,
+ contextUsagePercentage: 0,
+ hasContextUsage: false,
+ hasMetering: false,
+ usage: null,
+ inThinking: false,
+ toolValidationError: null,
+ validatedFrames: 0,
+ finished: false
};
- const transformStream = new TransformStream({
- async transform(chunk, controller) {
- // Track output so we can emit a keepalive if this frame yields no chunk.
- const enqueueCountBefore = chunkIndex;
- // Append to buffer
- const newBuffer = new Uint8Array(buffer.length + chunk.length);
- newBuffer.set(buffer);
- newBuffer.set(chunk, buffer.length);
- buffer = newBuffer;
-
- // Parse events from buffer
- let iterations = 0;
- const maxIterations = 1000;
- while (buffer.length >= 16 && iterations < maxIterations) {
- iterations++;
- const view = new DataView(buffer.buffer, buffer.byteOffset);
- const totalLength = view.getUint32(0, false);
-
- if (totalLength < 16 || totalLength > buffer.length || buffer.length < totalLength) break;
-
- const eventData = buffer.slice(0, totalLength);
- buffer = buffer.slice(totalLength);
-
- const event = parseEventFrame(eventData);
- if (!event) continue;
-
- const eventType = event.headers[":event-type"] || "";
-
- // Track total content length for token estimation
- if (!state.totalContentLength) state.totalContentLength = 0;
- if (!state.contextUsagePercentage) state.contextUsagePercentage = 0;
-
- // Handle assistantResponseEvent
- if (eventType === "assistantResponseEvent" && event.payload?.content) {
- let content = event.payload.content;
-
- // Kiro Claude models can leak blocks into the content stream.
- // We strip these literal tags to prevent duplication, as the reasoning
- // is already routed correctly via reasoningContentEvent.
- if (state.inThinking) {
- if (content.includes("")) {
- state.inThinking = false;
- const after = content.split("").slice(1).join("");
- content = after.startsWith("\n") ? after.substring(1) : after;
- } else {
- content = ""; // Drop entirely while inside thinking block
- }
- } else if (content.includes("")) {
- state.inThinking = true;
- if (content.includes("")) {
- state.inThinking = false;
- const before = content.split("")[0];
- const after = content.split("").slice(1).join("");
- content = before + (after.startsWith("\n") ? after.substring(1) : after);
- } else {
- content = content.split("")[0];
- }
- }
-
- if (!content && state.hasReasoningContent) {
- // If we stripped everything, skip emitting an empty content chunk
- continue;
- }
-
- state.totalContentLength += content.length;
-
- const chunk = {
- id: responseId,
- object: "chat.completion.chunk",
- created,
- model,
- choices: [{
- index: 0,
- delta: chunkIndex === 0
- ? { role: "assistant", content }
- : { content },
- finish_reason: null
- }]
- };
- chunkIndex++;
- controller.enqueue(new TextEncoder().encode(`data: ${JSON.stringify(chunk)}\n\n`));
+ const diagnostics = (overrides = {}) => ({
+ terminal_provenance: state.terminalProvenance || "clean_eventstream_eof",
+ transport_state: state.transportState,
+ stop_reason: state.stopReason,
+ stop_disposition: stopDisposition(state.stopReason, state.hasToolCalls),
+ response_state: state.hasToolCalls
+ ? "valid_tool"
+ : state.hasText || state.hasReasoning || state.hasCode
+ ? "text_reasoning"
+ : state.explicitStop
+ ? "explicit_stop"
+ : "no_semantic_output",
+ event_counts: { ...eventCounts },
+ incomplete_frame_bytes: state.buffer.byteLength,
+ ...overrides
+ });
+ const sseChunk = (delta, finishReason = null, usage) => encoder.encode(`data: ${JSON.stringify({
+ id: responseId,
+ object: "chat.completion.chunk",
+ created,
+ model,
+ choices: [{ index: 0, delta, finish_reason: finishReason }],
+ ...(usage ? { usage } : {})
+ })}\n\n`);
+ const emitDelta = (controller, delta) => {
+ if (state.chunkIndex === 0) delta = { role: "assistant", ...delta };
+ state.chunkIndex++;
+ controller.enqueue(sseChunk(delta));
+ };
+ const fail = (controller, provenance, code, message, extra = {}) => {
+ state.finished = true;
+ state.terminalProvenance = provenance;
+ state.transportState = extra.transport_state || "corrupt_frame";
+ const detail = diagnostics({
+ stop_disposition: extra.stop_disposition || "terminal_incomplete",
+ ...extra
+ });
+ options.onTerminalState?.(detail);
+ controller.enqueue(encodeSSEError(code, message, detail));
+ };
+ const assertToolBufferBound = () => {
+ if (state.bufferedToolBytes <= (options.maxToolBytes || KIRO_REPAIR_BUFFER_MAX_BYTES / 2)) return;
+ const error = new Error("Kiro buffered tool input exceeded the integrity memory bound");
+ error.code = "KIRO_BUFFER_EXCEEDED";
+ throw error;
+ };
+ const appendToolInput = (tool, input) => {
+ if (input === undefined) return;
+ if (typeof input === "string") {
+ if (tool.inputKind && tool.inputKind !== "string") throw new Error("Kiro tool input changed fragment type");
+ tool.inputKind = "string";
+ tool.inputChunks ||= [];
+ tool.inputChunks.push(input);
+ state.bufferedToolBytes += encoder.encode(input).byteLength;
+ } else if (input && typeof input === "object" && !Array.isArray(input)) {
+ if (tool.inputKind && tool.inputKind !== "object") throw new Error("Kiro tool input changed fragment type");
+ tool.inputKind = "object";
+ state.bufferedToolBytes -= tool.inputBytes || 0;
+ tool.inputObject = input;
+ tool.inputBytes = encoder.encode(JSON.stringify(input)).byteLength;
+ state.bufferedToolBytes += tool.inputBytes;
+ } else {
+ throw new Error("Kiro tool input must be a JSON object");
+ }
+ assertToolBufferBound();
+ };
+ const parsedToolInput = (tool) => {
+ if (!tool.inputKind) throw new Error("Kiro tool call is missing input");
+ if (tool.inputKind === "object") return tool.inputObject;
+ try {
+ const input = JSON.parse(tool.inputChunks.join(""));
+ if (!input || typeof input !== "object" || Array.isArray(input)) throw new Error("not an object");
+ return input;
+ } catch (error) {
+ throw new Error(`Kiro tool input must be valid object JSON (${error.message})`);
+ }
+ };
+ const emitTools = (controller) => {
+ for (const tool of state.tools.values()) {
+ const input = parsedToolInput(tool);
+ if (tool.name === "tool_call") {
+ if (typeof input.name !== "string" || !input.name.trim()) {
+ throw new Error("Invalid Kiro tool_call payload: missing nested MCP tool name");
}
-
- // Handle reasoningContentEvent (Kiro thinking / reasoning)
- // Kiro returns reasoning as a separate event when the request system
- // prompt contains enabled. Surface it
- // as OpenAI delta.reasoning_content so downstream translators can map
- // it back to Claude thinking blocks / Anthropic reasoning, etc.
- if (eventType === "reasoningContentEvent") {
- const reasoning = event.payload?.reasoningContentEvent || event.payload || {};
- const reasoningText = (typeof reasoning === "string")
- ? reasoning
- : (reasoning.text || reasoning.content || "");
- if (reasoningText) {
- state.hasReasoningContent = true;
- state.totalContentLength += reasoningText.length;
-
- const reasoningDelta = state.reasoningChunkCount === 0 && chunkIndex === 0
- ? { role: "assistant", reasoning_content: reasoningText }
- : { reasoning_content: reasoningText };
-
- const chunk = {
- id: responseId,
- object: "chat.completion.chunk",
- created,
- model,
- choices: [{
- index: 0,
- delta: reasoningDelta,
- finish_reason: null
- }]
- };
- chunkIndex++;
- state.reasoningChunkCount++;
- controller.enqueue(new TextEncoder().encode(`data: ${JSON.stringify(chunk)}\n\n`));
- }
+ if (!Object.prototype.hasOwnProperty.call(input, "arguments")) {
+ throw new Error("Invalid Kiro tool_call payload: missing nested MCP tool arguments");
}
+ }
+ const index = state.toolCounter++;
+ emitDelta(controller, {
+ tool_calls: [{
+ index,
+ id: tool.id,
+ type: "function",
+ function: { name: tool.name, arguments: "" }
+ }]
+ });
+ emitDelta(controller, {
+ tool_calls: [{ index, function: { arguments: JSON.stringify(input) } }]
+ });
+ state.hasToolCalls = true;
+ }
+ state.tools.clear();
+ state.bufferedToolBytes = 0;
+ if (state.stopReason === "tool_use" && !state.hasToolCalls) {
+ throw new Error("Kiro tool_use stop reason did not include a complete tool call");
+ }
+ };
+ const processEvent = (event, controller) => {
+ const messageType = event.headers[":message-type"];
+ if (messageType === "error" || messageType === "exception") {
+ fail(
+ controller,
+ "upstream_eventstream_error",
+ "kiro_upstream_eventstream_error",
+ event.payload?.message || `Kiro upstream sent an EventStream ${messageType}`,
+ { transport_state: "upstream_error" }
+ );
+ return false;
+ }
- // Handle codeEvent
- if (eventType === "codeEvent" && event.payload?.content) {
- const chunk = {
- id: responseId,
- object: "chat.completion.chunk",
- created,
- model,
- choices: [{
- index: 0,
- delta: { content: event.payload.content },
- finish_reason: null
- }]
- };
- chunkIndex++;
- controller.enqueue(new TextEncoder().encode(`data: ${JSON.stringify(chunk)}\n\n`));
+ const eventType = event.headers[":event-type"] || "";
+ const eventCountKey = KIRO_EVENT_TYPES.has(eventType) ? eventType : "other";
+ eventCounts[eventCountKey] = (eventCounts[eventCountKey] || 0) + 1;
+ if (eventType === "assistantResponseEvent" && typeof event.payload?.content === "string") {
+ let content = event.payload.content;
+ if (state.inThinking) {
+ const end = content.indexOf("");
+ if (end < 0) content = "";
+ else {
+ state.inThinking = false;
+ content = content.slice(end + 11).replace(/^\n/u, "");
}
-
- // Handle toolUseEvent
- if (eventType === "toolUseEvent" && event.payload) {
- state.hasToolCalls = true;
- const toolUse = event.payload;
- const toolUses = Array.isArray(toolUse) ? toolUse : [toolUse];
-
- for (const singleToolUse of toolUses) {
- const toolCallId = singleToolUse.toolUseId || `call_${Date.now()}`;
- const toolName = singleToolUse.name || "";
- const toolInput = singleToolUse.input;
-
- let toolIndex;
- const isNewTool = !state.seenToolIds.has(toolCallId);
-
- if (isNewTool) {
- toolIndex = state.toolCallIndex++;
- state.seenToolIds.set(toolCallId, toolIndex);
-
- const startChunk = {
- id: responseId,
- object: "chat.completion.chunk",
- created,
- model,
- choices: [{
- index: 0,
- delta: {
- ...(chunkIndex === 0 ? { role: "assistant" } : {}),
- tool_calls: [{
- index: toolIndex,
- id: toolCallId,
- type: "function",
- function: {
- name: toolName,
- arguments: ""
- }
- }]
- },
- finish_reason: null
- }]
- };
- chunkIndex++;
- controller.enqueue(new TextEncoder().encode(`data: ${JSON.stringify(startChunk)}\n\n`));
- } else {
- toolIndex = state.seenToolIds.get(toolCallId);
- }
-
- if (toolInput !== undefined) {
- let argumentsStr;
-
- if (typeof toolInput === 'string') {
- argumentsStr = toolInput;
- } else if (typeof toolInput === 'object') {
- argumentsStr = JSON.stringify(toolInput);
- } else {
- continue;
- }
-
- const argsChunk = {
- id: responseId,
- object: "chat.completion.chunk",
- created,
- model,
- choices: [{
- index: 0,
- delta: {
- tool_calls: [{
- index: toolIndex,
- function: {
- arguments: argumentsStr
- }
- }]
- },
- finish_reason: null
- }]
- };
- chunkIndex++;
- controller.enqueue(new TextEncoder().encode(`data: ${JSON.stringify(argsChunk)}\n\n`));
- }
+ } else {
+ const start = content.indexOf("");
+ if (start >= 0) {
+ const end = content.indexOf("", start + 10);
+ if (end < 0) {
+ state.inThinking = true;
+ content = content.slice(0, start);
+ } else {
+ content = content.slice(0, start) + content.slice(end + 11).replace(/^\n/u, "");
}
}
-
- // Handle messageStopEvent
- if (eventType === "messageStopEvent") {
- const chunk = {
- id: responseId,
- object: "chat.completion.chunk",
- created,
- model,
- choices: [{
- index: 0,
- delta: {},
- finish_reason: state.hasToolCalls ? "tool_calls" : "stop"
- }]
- };
- state.finishEmitted = true;
- controller.enqueue(new TextEncoder().encode(`data: ${JSON.stringify(chunk)}\n\n`));
- }
-
- // Handle contextUsageEvent to extract contextUsagePercentage
- if (eventType === "contextUsageEvent" && event.payload?.contextUsagePercentage) {
- state.contextUsagePercentage = event.payload.contextUsagePercentage;
- // Mark that we received context usage event
- state.hasContextUsage = true;
+ }
+ if (content || !state.hasReasoning) {
+ state.hasText ||= content.length > 0;
+ state.totalContentLength += content.length;
+ emitDelta(controller, { content });
+ }
+ } else if (eventType === "reasoningContentEvent") {
+ const value = event.payload?.reasoningContentEvent || event.payload || {};
+ const content = typeof value === "string" ? value : value.text || value.content || "";
+ if (content) {
+ state.hasReasoning = true;
+ state.totalContentLength += content.length;
+ emitDelta(controller, { reasoning_content: content });
+ }
+ } else if (eventType === "codeEvent" && typeof event.payload?.content === "string") {
+ state.hasCode = true;
+ state.totalContentLength += event.payload.content.length;
+ emitDelta(controller, { content: event.payload.content });
+ } else if (eventType === "toolUseEvent") {
+ state.sawToolUse = true;
+ if (state.toolValidationError) return true;
+ const values = Array.isArray(event.payload) ? event.payload : [event.payload];
+ if (!values[0]) throw new Error("Kiro toolUseEvent is empty");
+ for (const value of values) {
+ const name = typeof value?.name === "string" ? value.name.trim() : "";
+ if (!name) throw new Error("Kiro toolUseEvent is missing a tool name");
+ let id;
+ if (value.toolUseId == null) {
+ id = `call_${created}_${state.tools.size + 1}`;
+ } else if (typeof value.toolUseId !== "string" || !value.toolUseId.trim()) {
+ throw new Error("Kiro toolUseEvent has an invalid toolUseId");
+ } else {
+ id = value.toolUseId;
}
-
- // Handle meteringEvent - mark that we received it
- if (eventType === "meteringEvent") {
- state.hasMeteringEvent = true;
+ let tool = state.tools.get(id);
+ if (!tool) {
+ tool = { id, name };
+ state.tools.set(id, tool);
+ state.bufferedToolBytes += encoder.encode(id).byteLength + encoder.encode(name).byteLength + 32;
+ assertToolBufferBound();
+ } else if (tool.name !== name) {
+ throw new Error("Kiro tool name changed between fragments");
}
+ appendToolInput(tool, value.input);
+ }
+ } else if (eventType === "messageStopEvent") {
+ state.explicitStop = true;
+ const reason = normalizeStopReason(
+ event.payload?.stopReason ?? event.payload?.stop_reason
+ ) || (state.sawToolUse ? "tool_use" : "end_turn");
+ const merged = mergeStopReason(state.stopReason, reason);
+ if (merged !== state.stopReason) state.terminalProvenance = "message_stop_event";
+ state.stopReason = merged;
+ } else if (eventType === "metadataEvent" || eventType === "MetadataEvent") {
+ const metadata = event.payload?.metadataEvent || event.payload?.metadata || event.payload;
+ const reason = normalizeStopReason(metadata?.stopReason ?? metadata?.stop_reason);
+ if (reason) {
+ state.explicitStop = true;
+ const merged = mergeStopReason(state.stopReason, reason);
+ if (merged !== state.stopReason) state.terminalProvenance = "metadata_stop_reason";
+ state.stopReason = merged;
+ }
+ } else if (eventType === "contextUsageEvent") {
+ const percentage = Number(event.payload?.contextUsagePercentage);
+ if (Number.isFinite(percentage)) {
+ state.contextUsagePercentage = percentage;
+ state.hasContextUsage = true;
+ }
+ } else if (eventType === "meteringEvent") {
+ state.hasMetering = true;
+ const metering = event.payload?.meteringEvent || event.payload || {};
+ const credits = Number(metering.usage);
+ if (Number.isFinite(credits)) {
+ state.usage = {
+ ...(state.usage || {}),
+ kiro_credits: credits,
+ kiro_credit_unit: typeof metering.unit === "string" ? metering.unit : "credit"
+ };
+ }
+ } else if (eventType === "metricsEvent") {
+ const metrics = event.payload?.metricsEvent || event.payload || {};
+ const prompt = Number(metrics.inputTokens) || 0;
+ const completion = Number(metrics.outputTokens) || 0;
+ if (prompt || completion) {
+ state.usage = {
+ ...(state.usage || {}),
+ prompt_tokens: prompt,
+ completion_tokens: completion,
+ total_tokens: prompt + completion
+ };
+ const cacheRead = Number(metrics.cacheReadInputTokens || metrics.cache_read_input_tokens) || 0;
+ const cacheCreate = Number(metrics.cacheCreationInputTokens || metrics.cache_creation_input_tokens) || 0;
+ if (cacheRead) state.usage.cache_read_input_tokens = cacheRead;
+ if (cacheCreate) state.usage.cache_creation_input_tokens = cacheCreate;
+ }
+ }
+ return true;
+ };
+ const processBytes = (chunk, controller) => {
+ const combinedLength = state.buffer.byteLength + chunk.byteLength;
+ if (combinedLength > (options.maxRawBytes || EVENTSTREAM_MAX_MESSAGE_BYTES)) {
+ fail(
+ controller,
+ "corrupt_eventstream_frame",
+ "kiro_missing_terminal",
+ "Kiro EventStream buffered bytes exceed the protocol bound"
+ );
+ return false;
+ }
+ if (state.buffer.byteLength === 0) {
+ state.buffer = chunk;
+ } else {
+ const joined = new Uint8Array(combinedLength);
+ joined.set(state.buffer);
+ joined.set(chunk, state.buffer.byteLength);
+ state.buffer = joined;
+ }
- // Handle metricsEvent for token usage
- if (eventType === "metricsEvent") {
- // Extract usage data from metricsEvent payload
- const metrics = event.payload?.metricsEvent || event.payload;
- if (metrics && typeof metrics === 'object') {
- const inputTokens = metrics.inputTokens || 0;
- const outputTokens = metrics.outputTokens || 0;
- // ponytail: Amazon Q upstream does not expose cache fields today,
- // but pick up cache_read_input_tokens / cache_creation_input_tokens
- // if the event shape grows them so cost tracking stays accurate.
- const cachedTokens = metrics.cacheReadInputTokens || metrics.cache_read_input_tokens || 0;
- const cacheCreationInputTokens = metrics.cacheCreationInputTokens || metrics.cache_creation_input_tokens || 0;
-
- if (inputTokens > 0 || outputTokens > 0) {
- state.usage = {
- prompt_tokens: inputTokens,
- completion_tokens: outputTokens,
- total_tokens: inputTokens + outputTokens
- };
- // Kiro is Claude-backed: inputTokens EXCLUDES cache (Claude convention),
- // not inclusive like OpenAI's cached_tokens. Emit cache_read_input_tokens
- // (not cached_tokens) so canonicalizeUsage takes the Claude fold path and
- // correctly adds cache back into prompt_tokens instead of undercharging.
- if (cachedTokens > 0) state.usage.cache_read_input_tokens = cachedTokens;
- if (cacheCreationInputTokens > 0) state.usage.cache_creation_input_tokens = cacheCreationInputTokens;
- }
- }
+ while (state.buffer.byteLength >= 12) {
+ const view = new DataView(state.buffer.buffer, state.buffer.byteOffset);
+ if (view.getUint32(8, false) !== crc32(state.buffer.subarray(0, 8))) {
+ fail(controller, "corrupt_eventstream_frame", "kiro_missing_terminal", "Kiro EventStream prelude CRC mismatch");
+ return false;
+ }
+ const totalLength = view.getUint32(0, false);
+ const headersLength = view.getUint32(4, false);
+ if (totalLength < 16 || totalLength > EVENTSTREAM_MAX_MESSAGE_BYTES ||
+ headersLength > EVENTSTREAM_MAX_HEADERS_BYTES || headersLength > totalLength - 16) {
+ fail(controller, "corrupt_eventstream_frame", "kiro_missing_terminal", "Kiro EventStream frame bounds are invalid");
+ return false;
+ }
+ if (state.buffer.byteLength < totalLength) break;
+ const frame = state.buffer.slice(0, totalLength);
+ state.buffer = state.buffer.slice(totalLength);
+ let event;
+ try {
+ event = parseEventFrame(frame);
+ } catch (error) {
+ fail(controller, "corrupt_eventstream_frame", "kiro_missing_terminal", error.message);
+ return false;
+ }
+ state.transportState = "valid_complete_frame";
+ state.validatedFrames++;
+ try {
+ if (!processEvent(event, controller)) return false;
+ } catch (error) {
+ const bufferExceeded = error.code === "KIRO_BUFFER_EXCEEDED";
+ if (!bufferExceeded) {
+ state.toolValidationError ||= error.message;
+ state.tools.clear();
+ state.bufferedToolBytes = 0;
+ continue;
}
-
- // Emit final chunk only after receiving BOTH meteringEvent AND contextUsageEvent
- if (state.hasMeteringEvent && state.hasContextUsage && !state.finishEmitted) {
- state.finishEmitted = true;
-
- // Estimate tokens if not available from events
- if (!state.usage) {
- // Estimate output tokens from content length
- const estimatedOutputTokens = state.totalContentLength > 0
- ? Math.max(1, Math.floor(state.totalContentLength / 4))
- : 0;
-
- // Estimate input tokens from contextUsagePercentage
- const estimatedInputTokens = state.contextUsagePercentage > 0
- ? Math.floor(state.contextUsagePercentage * contextWindow / 100)
- : 0;
-
- state.usage = {
- prompt_tokens: estimatedInputTokens,
- completion_tokens: estimatedOutputTokens,
- total_tokens: estimatedInputTokens + estimatedOutputTokens
- };
- }
-
- const finishChunk = {
- id: responseId,
- object: "chat.completion.chunk",
- created,
- model,
- choices: [{
- index: 0,
- delta: {},
- finish_reason: state.hasToolCalls ? "tool_calls" : "stop"
- }]
- };
-
- // Include usage in final chunk if available
- if (state.usage) {
- finishChunk.usage = state.usage;
+ fail(
+ controller,
+ "integrity_buffer_exceeded",
+ "kiro_integrity_buffer_exceeded",
+ error.message,
+ {
+ transport_state: state.transportState,
+ stop_disposition: "terminal_incomplete"
}
-
- controller.enqueue(new TextEncoder().encode(`data: ${JSON.stringify(finishChunk)}\n\n`));
- }
- }
-
- if (iterations >= maxIterations) {
- console.warn("[Kiro] Max iterations reached in event parsing");
+ );
+ return false;
}
+ }
+ return true;
+ };
+ const finish = (controller) => {
+ if (state.finished) return;
+ if (state.buffer.byteLength) {
+ fail(
+ controller,
+ "incomplete_eventstream_frame",
+ "kiro_missing_terminal",
+ "Kiro EventStream ended with a truncated frame",
+ { transport_state: "incomplete_frame" }
+ );
+ return;
+ }
+ state.transportState = "clean_eof";
+ const declaredDisposition = stopDisposition(state.stopReason, state.sawToolUse);
+ if (["retryable_protocol_failure", "terminal_incomplete", "terminal_refusal", "unknown_failure"].includes(declaredDisposition)) {
+ const code = declaredDisposition === "retryable_protocol_failure"
+ ? "kiro_retryable_protocol_failure"
+ : declaredDisposition === "terminal_refusal"
+ ? "kiro_terminal_refusal"
+ : declaredDisposition === "terminal_incomplete"
+ ? "kiro_terminal_incomplete"
+ : "kiro_unknown_stop_reason";
+ fail(
+ controller,
+ state.terminalProvenance || "metadata_stop_reason",
+ code,
+ `Kiro ended with non-success stop reason: ${state.stopReason}`,
+ { transport_state: state.transportState, stop_disposition: declaredDisposition }
+ );
+ return;
+ }
+ if (state.toolValidationError) {
+ fail(
+ controller,
+ "invalid_tool_call",
+ "invalid_kiro_tool_call",
+ state.toolValidationError,
+ { transport_state: state.transportState, stop_disposition: "retryable_protocol_failure" }
+ );
+ return;
+ }
+ try {
+ emitTools(controller);
+ } catch (error) {
+ fail(
+ controller,
+ "invalid_tool_call",
+ "invalid_kiro_tool_call",
+ error.message,
+ { transport_state: state.transportState, stop_disposition: "retryable_protocol_failure" }
+ );
+ return;
+ }
- // No client chunk produced this frame — emit an SSE comment keepalive
- // so the stall watchdog sees upstream activity (ignored by parser/client).
- if (chunkIndex === enqueueCountBefore && !state.finishEmitted) {
- controller.enqueue(new TextEncoder().encode(": ka\n\n"));
- }
- },
+ const hasOutput = state.hasText || state.hasReasoning || state.hasCode || state.hasToolCalls;
+ if (!hasOutput && !state.explicitStop) {
+ fail(
+ controller,
+ "empty_response_eof",
+ "kiro_missing_terminal",
+ "Kiro EventStream ended without model output",
+ { transport_state: state.transportState }
+ );
+ return;
+ }
- flush(controller) {
- // Emit finish chunk if not already sent
- if (!state.finishEmitted) {
- state.finishEmitted = true;
- const finishChunk = {
- id: responseId,
- object: "chat.completion.chunk",
- created,
- model,
- choices: [{
- index: 0,
- delta: {},
- finish_reason: state.hasToolCalls ? "tool_calls" : "stop"
- }]
- };
- controller.enqueue(new TextEncoder().encode(`data: ${JSON.stringify(finishChunk)}\n\n`));
- }
+ const disposition = stopDisposition(state.stopReason, state.hasToolCalls);
+ if (["retryable_protocol_failure", "terminal_incomplete", "terminal_refusal", "unknown_failure"].includes(disposition)) {
+ const code = disposition === "retryable_protocol_failure"
+ ? "kiro_retryable_protocol_failure"
+ : disposition === "terminal_refusal"
+ ? "kiro_terminal_refusal"
+ : disposition === "terminal_incomplete"
+ ? "kiro_terminal_incomplete"
+ : "kiro_unknown_stop_reason";
+ fail(
+ controller,
+ state.terminalProvenance || "metadata_stop_reason",
+ code,
+ `Kiro ended with non-success stop reason: ${state.stopReason}`,
+ { transport_state: state.transportState, stop_disposition: disposition }
+ );
+ return;
+ }
- // Send final done message
- controller.enqueue(new TextEncoder().encode(SSE_DONE));
+ if (state.hasMetering && state.hasContextUsage && !state.usage?.total_tokens) {
+ const completion = state.totalContentLength
+ ? Math.max(1, Math.floor(state.totalContentLength / 4))
+ : 0;
+ const prompt = Math.floor(state.contextUsagePercentage * contextWindow / 100);
+ state.usage = {
+ ...(state.usage || {}),
+ prompt_tokens: prompt,
+ completion_tokens: completion,
+ total_tokens: prompt + completion
+ };
}
- });
+ const finishReason = state.hasToolCalls
+ ? "tool_calls"
+ : disposition === "length"
+ ? "length"
+ : "stop";
+ controller.enqueue(sseChunk({}, finishReason, state.usage));
+ controller.enqueue(encoder.encode(SSE_DONE));
+ state.finished = true;
+ options.onTerminalState?.(diagnostics({
+ terminal_provenance: state.terminalProvenance || "clean_eventstream_eof",
+ transport_state: state.transportState,
+ stop_disposition: disposition
+ }));
+ };
- // Pipe response body through transform stream
if (!response.body) {
- return new Response(SSE_DONE, { status: response.status, headers: { "Content-Type": "text/event-stream" } });
+ const detail = diagnostics({
+ terminal_provenance: "missing_response_body",
+ transport_state: "missing_body",
+ stop_disposition: "terminal_incomplete"
+ });
+ options.onTerminalState?.(detail);
+ return new Response(encodeSSEError(
+ "kiro_missing_terminal",
+ "Kiro response did not include an EventStream body",
+ detail
+ ), { status: response.status, headers: { ...SSE_HEADERS } });
}
- const transformedStream = response.body.pipeThrough(transformStream);
- return new Response(transformedStream, {
+ const reader = response.body.getReader();
+ const stream = new ReadableStream({
+ start: async (controller) => {
+ try {
+ while (!state.finished) {
+ const { done, value } = await reader.read();
+ if (done) break;
+ const chunksBefore = state.chunkIndex;
+ const framesBefore = state.validatedFrames;
+ if (!processBytes(value, controller)) {
+ await reader.cancel("invalid Kiro EventStream").catch(() => {});
+ break;
+ }
+ if (state.validatedFrames > framesBefore && state.chunkIndex === chunksBefore) {
+ controller.enqueue(encoder.encode(": kiro-upstream\n\n"));
+ }
+ }
+ finish(controller);
+ controller.close();
+ } catch (error) {
+ if (!state.finished) {
+ fail(
+ controller,
+ "upstream_read_error",
+ "kiro_missing_terminal",
+ error.message || "Kiro EventStream read failed",
+ { transport_state: "upstream_error" }
+ );
+ }
+ controller.close();
+ }
+ },
+ cancel(reason) {
+ return reader.cancel(reason);
+ }
+ });
+ return new Response(stream, {
status: response.status,
statusText: response.statusText,
headers: { ...SSE_HEADERS }
@@ -527,65 +1111,90 @@ export class KiroExecutor extends BaseExecutor {
/**
* Parse AWS EventStream frame
*/
-function parseEventFrame(data) {
- try {
- const view = new DataView(data.buffer, data.byteOffset);
- const headersLength = view.getUint32(4, false);
-
- // Parse headers
- const headers = {};
- let offset = 12; // After prelude
- const headerEnd = 12 + headersLength;
-
- while (offset < headerEnd && offset < data.length) {
- const nameLen = data[offset];
- offset++;
- if (offset + nameLen > data.length) break;
-
- const name = new TextDecoder().decode(data.slice(offset, offset + nameLen));
- offset += nameLen;
-
- const headerType = data[offset];
- offset++;
- if (headerType === 7) { // String type
- const valueLen = (data[offset] << 8) | data[offset + 1];
- offset += 2;
- if (offset + valueLen > data.length) break;
+function parseEventFrame(data) {
+ if (!(data instanceof Uint8Array) || data.byteLength < 16) {
+ throw new Error("AWS EventStream frame is shorter than 16 bytes");
+ }
+ const view = new DataView(data.buffer, data.byteOffset, data.byteLength);
+ const totalLength = view.getUint32(0, false);
+ const headersLength = view.getUint32(4, false);
+ if (totalLength !== data.byteLength) {
+ throw new Error("AWS EventStream frame length does not match its prelude");
+ }
+ if (totalLength > EVENTSTREAM_MAX_MESSAGE_BYTES ||
+ headersLength > EVENTSTREAM_MAX_HEADERS_BYTES ||
+ headersLength > totalLength - 16) {
+ throw new Error("AWS EventStream frame bounds are invalid");
+ }
+ if (view.getUint32(8, false) !== crc32(data.subarray(0, 8))) {
+ throw new Error("AWS EventStream prelude CRC mismatch");
+ }
+ if (view.getUint32(totalLength - 4, false) !== crc32(data.subarray(0, totalLength - 4))) {
+ throw new Error("AWS EventStream message CRC mismatch");
+ }
- const value = new TextDecoder().decode(data.slice(offset, offset + valueLen));
- offset += valueLen;
- headers[name] = value;
- } else {
- break;
- }
+ const headers = Object.create(null);
+ const names = new Set();
+ let offset = 12;
+ const headerEnd = offset + headersLength;
+ const requireBytes = (count) => {
+ if (offset + count > headerEnd) {
+ throw new Error("AWS EventStream header exceeds its declared bounds");
}
-
- // Parse payload
- const payloadStart = 12 + headersLength;
- const payloadEnd = data.length - 4; // Exclude message CRC
-
- let payload = null;
- if (payloadEnd > payloadStart) {
- const payloadStr = new TextDecoder().decode(data.slice(payloadStart, payloadEnd));
-
- // Skip empty or whitespace-only payloads
- if (!payloadStr || !payloadStr.trim()) {
- return { headers, payload: null };
- }
-
- try {
- payload = JSON.parse(payloadStr);
- } catch (parseError) {
- // Log parse error for debugging
- console.warn(`[Kiro] Failed to parse payload: ${parseError.message} | payload: ${payloadStr.substring(0, 100)}`);
- payload = { raw: payloadStr };
- }
+ };
+
+ while (offset < headerEnd) {
+ requireBytes(1);
+ const nameLength = data[offset++];
+ requireBytes(nameLength + 1);
+ const name = decoder.decode(data.subarray(offset, offset + nameLength));
+ offset += nameLength;
+ if (names.has(name)) throw new Error(`AWS EventStream contains duplicate header: ${name}`);
+ names.add(name);
+ const type = data[offset++];
+
+ if (type === 0 || type === 1) {
+ headers[name] = type === 0;
+ } else if (type === 2) {
+ requireBytes(1);
+ headers[name] = view.getInt8(offset);
+ offset += 1;
+ } else if (type === 3) {
+ requireBytes(2);
+ headers[name] = view.getInt16(offset, false);
+ offset += 2;
+ } else if (type === 4) {
+ requireBytes(4);
+ headers[name] = view.getInt32(offset, false);
+ offset += 4;
+ } else if (type === 5 || type === 8) {
+ requireBytes(8);
+ offset += 8;
+ } else if (type === 6 || type === 7) {
+ requireBytes(2);
+ const valueLength = view.getUint16(offset, false);
+ offset += 2;
+ requireBytes(valueLength);
+ const bytes = data.subarray(offset, offset + valueLength);
+ headers[name] = type === 7 ? decoder.decode(bytes) : bytes;
+ offset += valueLength;
+ } else if (type === 9) {
+ requireBytes(16);
+ offset += 16;
+ } else {
+ throw new Error(`AWS EventStream header ${name} has unknown type ${type}`);
}
+ }
- return { headers, payload };
- } catch {
- return null;
+ const payloadBytes = data.subarray(headerEnd, totalLength - 4);
+ if (payloadBytes.byteLength === 0) return { headers, payload: null };
+ const payloadText = decoder.decode(payloadBytes);
+ if (!payloadText.trim()) return { headers, payload: null };
+ try {
+ return { headers, payload: JSON.parse(payloadText) };
+ } catch (error) {
+ throw new Error(`AWS EventStream payload is not valid JSON (${error.message})`);
}
}
diff --git a/open-sse/handlers/chatCore/sseToJsonHandler.js b/open-sse/handlers/chatCore/sseToJsonHandler.js
index f25986c53b..9638824b6b 100644
--- a/open-sse/handlers/chatCore/sseToJsonHandler.js
+++ b/open-sse/handlers/chatCore/sseToJsonHandler.js
@@ -40,15 +40,21 @@ function pickAssistantMessageForChatCompletion(output) {
*/
export function parseSSEToOpenAIResponse(rawSSE, fallbackModel) {
const chunks = [];
+ let streamError = null;
for (const line of String(rawSSE || "").split("\n")) {
const trimmed = line.trim();
if (!trimmed.startsWith("data:")) continue;
const payload = trimmed.slice(5).trim();
if (!payload || payload === "[DONE]") continue;
- try { chunks.push(JSON.parse(payload)); } catch { /* ignore malformed lines */ }
+ try {
+ const chunk = JSON.parse(payload);
+ if (chunk?.error) streamError = chunk.error;
+ else chunks.push(chunk);
+ } catch { /* ignore malformed lines */ }
}
+ if (streamError) return { error: streamError };
if (chunks.length === 0) return null;
const first = chunks[0];
@@ -196,6 +202,12 @@ export async function handleForcedSSEToJson({ providerResponse, sourceFormat, pr
const sseText = await providerResponse.text();
const parsed = parseSSEToOpenAIResponse(sseText, model);
if (!parsed) return createErrorResult(HTTP_STATUS.BAD_GATEWAY, "Invalid SSE response for non-streaming request");
+ if (parsed.error) {
+ return createErrorResult(
+ HTTP_STATUS.BAD_GATEWAY,
+ parsed.error.message || "Upstream SSE stream failed"
+ );
+ }
if (onRequestSuccess) await onRequestSuccess();
diff --git a/tests/unit/kiro-nonstream-error.test.js b/tests/unit/kiro-nonstream-error.test.js
new file mode 100644
index 0000000000..d36b1c0a14
--- /dev/null
+++ b/tests/unit/kiro-nonstream-error.test.js
@@ -0,0 +1,64 @@
+import { describe, expect, it, vi } from "vitest";
+
+vi.mock("@/lib/usageDb.js", () => ({
+ appendRequestLog: vi.fn(async () => {}),
+ saveRequestDetail: vi.fn(async () => {}),
+ saveRequestUsage: vi.fn(async () => {})
+}));
+
+const { FORMATS } = await import("../../open-sse/translator/formats.js");
+const {
+ handleForcedSSEToJson,
+ parseSSEToOpenAIResponse
+} = await import("../../open-sse/handlers/chatCore/sseToJsonHandler.js");
+
+describe("Kiro non-streaming error propagation", () => {
+ it("prefers a terminal SSE error over earlier semantic chunks", () => {
+ const raw = [
+ 'data: {"choices":[{"delta":{"content":"partial"},"finish_reason":null}]}',
+ 'data: {"error":{"message":"Kiro transport failed","code":"kiro_missing_terminal"}}',
+ "data: [DONE]"
+ ].join("\n\n");
+
+ expect(parseSSEToOpenAIResponse(raw, "kiro")).toEqual({
+ error: {
+ message: "Kiro transport failed",
+ code: "kiro_missing_terminal"
+ }
+ });
+ });
+
+ it("returns 502 instead of collapsing a failed Kiro SSE stream into stop", async () => {
+ const encoder = new TextEncoder();
+ const raw = [
+ 'data: {"choices":[{"delta":{"content":"partial"},"finish_reason":null}]}',
+ 'data: {"error":{"message":"Kiro stream ended incompletely","code":"kiro_missing_terminal"}}',
+ "data: [DONE]",
+ ""
+ ].join("\n\n");
+ const result = await handleForcedSSEToJson({
+ providerResponse: new Response(new ReadableStream({
+ start(controller) {
+ controller.enqueue(encoder.encode(raw));
+ controller.close();
+ }
+ }), { headers: { "content-type": "text/event-stream" } }),
+ sourceFormat: FORMATS.OPENAI,
+ provider: "kiro",
+ model: "kr/claude-opus-4.8",
+ body: { model: "kr/claude-opus-4.8", messages: [] },
+ stream: false,
+ requestStartTime: Date.now(),
+ connectionId: "test-connection",
+ clientRawRequest: { endpoint: "/v1/chat/completions" },
+ trackDone: vi.fn(),
+ appendLog: vi.fn()
+ });
+ const json = await result.response.json();
+
+ expect(result.success).toBe(false);
+ expect(result.response.status).toBe(502);
+ expect(json.error.message).toContain("Kiro stream ended incompletely");
+ expect(json).not.toHaveProperty("choices");
+ });
+});
diff --git a/tests/unit/kiro-terminal-integrity.test.js b/tests/unit/kiro-terminal-integrity.test.js
new file mode 100644
index 0000000000..aaf6620900
--- /dev/null
+++ b/tests/unit/kiro-terminal-integrity.test.js
@@ -0,0 +1,778 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+
+const fetchMock = vi.fn();
+vi.mock("../../open-sse/utils/proxyFetch.js", () => ({
+ proxyAwareFetch: (...args) => fetchMock(...args)
+}));
+
+const { KiroExecutor } = await import("../../open-sse/executors/kiro.js");
+
+const encoder = new TextEncoder();
+const credentials = {
+ accessToken: "test-token",
+ providerSpecificData: { kiroToolCallRepair: true }
+};
+
+function crc32(bytes) {
+ let crc = 0xffffffff;
+ for (const byte of bytes) {
+ crc ^= byte;
+ for (let bit = 0; bit < 8; bit++) {
+ crc = (crc >>> 1) ^ ((crc & 1) ? 0xedb88320 : 0);
+ }
+ }
+ return (crc ^ 0xffffffff) >>> 0;
+}
+
+function encodeHeader(name, value) {
+ const nameBytes = encoder.encode(name);
+ const valueBytes = encoder.encode(value);
+ const bytes = new Uint8Array(1 + nameBytes.length + 3 + valueBytes.length);
+ let offset = 0;
+ bytes[offset++] = nameBytes.length;
+ bytes.set(nameBytes, offset);
+ offset += nameBytes.length;
+ bytes[offset++] = 7;
+ new DataView(bytes.buffer).setUint16(offset, valueBytes.length, false);
+ offset += 2;
+ bytes.set(valueBytes, offset);
+ return bytes;
+}
+
+function concat(chunks) {
+ const output = new Uint8Array(chunks.reduce((size, chunk) => size + chunk.byteLength, 0));
+ let offset = 0;
+ for (const chunk of chunks) {
+ output.set(chunk, offset);
+ offset += chunk.byteLength;
+ }
+ return output;
+}
+
+function frameFromEntries(entries, payload) {
+ const headers = concat(entries.map(([name, value]) => encodeHeader(name, value)));
+ const payloadBytes = encoder.encode(JSON.stringify(payload));
+ const totalLength = 12 + headers.byteLength + payloadBytes.byteLength + 4;
+ const frame = new Uint8Array(totalLength);
+ const view = new DataView(frame.buffer);
+ view.setUint32(0, totalLength, false);
+ view.setUint32(4, headers.byteLength, false);
+ frame.set(headers, 12);
+ frame.set(payloadBytes, 12 + headers.byteLength);
+ return checksum(frame);
+}
+
+function frame(eventType, payload) {
+ return frameFromEntries([[":event-type", eventType]], payload);
+}
+
+function checksum(bytes) {
+ const view = new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength);
+ view.setUint32(8, crc32(bytes.subarray(0, 8)), false);
+ view.setUint32(bytes.byteLength - 4, crc32(bytes.subarray(0, bytes.byteLength - 4)), false);
+ return bytes;
+}
+
+function response(frames, status = 200) {
+ return new Response(new ReadableStream({
+ start(controller) {
+ for (const value of frames) controller.enqueue(value);
+ controller.close();
+ }
+ }), { status, statusText: status === 200 ? "OK" : "Upstream Error" });
+}
+
+function controlledResponse(frames = []) {
+ let controller;
+ const value = new Response(new ReadableStream({
+ start(streamController) {
+ controller = streamController;
+ for (const item of frames) controller.enqueue(item);
+ }
+ }), { status: 200 });
+ return {
+ value,
+ enqueue(item) {
+ controller.enqueue(item);
+ },
+ close() {
+ controller.close();
+ }
+ };
+}
+
+async function text(stream) {
+ const reader = stream.getReader();
+ const decoder = new TextDecoder();
+ let output = "";
+ while (true) {
+ const { done, value } = await reader.read();
+ if (done) return output + decoder.decode();
+ output += decoder.decode(value, { stream: true });
+ }
+}
+
+async function execute(executor = new KiroExecutor(), overrides = {}) {
+ return executor.execute({
+ model: "kr/claude-opus-4.8",
+ body: { systemPrompt: "base", conversationState: {} },
+ stream: true,
+ credentials,
+ ...overrides
+ });
+}
+
+beforeEach(() => {
+ fetchMock.mockReset();
+ delete process.env.KIRO_TOOL_CALL_REPAIR_BUFFER_MAX_BYTES;
+ delete process.env.KIRO_TOOL_CALL_REPAIR_TTFT_TIMEOUT_MS;
+ delete process.env.KIRO_TOOL_CALL_REPAIR_STALL_TIMEOUT_MS;
+});
+
+afterEach(() => {
+ delete process.env.KIRO_TOOL_CALL_REPAIR_BUFFER_MAX_BYTES;
+ delete process.env.KIRO_TOOL_CALL_REPAIR_TTFT_TIMEOUT_MS;
+ delete process.env.KIRO_TOOL_CALL_REPAIR_STALL_TIMEOUT_MS;
+});
+
+describe("Kiro terminal integrity recovery", () => {
+ it("keeps semantic output private behind a heartbeat until clean EOF", async () => {
+ const upstream = controlledResponse([
+ frame("assistantResponseEvent", { content: "private until validated" })
+ ]);
+ fetchMock.mockResolvedValueOnce(upstream.value);
+
+ const result = await execute();
+ const reader = result.response.body.getReader();
+ expect(new TextDecoder().decode((await reader.read()).value)).toBe(": kiro-validation\n\n");
+
+ let settled = false;
+ const semantic = reader.read().then((value) => {
+ settled = true;
+ return value;
+ });
+ await Promise.resolve();
+ expect(settled).toBe(false);
+
+ upstream.close();
+ expect(new TextDecoder().decode((await semantic).value)).toContain("private until validated");
+ await reader.cancel();
+ });
+
+ it("accepts CLI-compatible text and usage frames at clean EOF without messageStop", async () => {
+ fetchMock.mockResolvedValueOnce(response([
+ frame("assistantResponseEvent", { content: "Complete answer." }),
+ frame("meteringEvent", { usage: 2, unit: "credit" }),
+ frame("contextUsageEvent", { contextUsagePercentage: 10 })
+ ]));
+
+ const body = await (await execute()).response.text();
+
+ expect(fetchMock).toHaveBeenCalledTimes(1);
+ expect(body).toContain("Complete answer.");
+ expect(body).toContain('"finish_reason":"stop"');
+ expect(body).toContain('"kiro_credits":2');
+ });
+
+ it("parses frames split across chunks and multiple frames in one chunk", async () => {
+ const first = frame("assistantResponseEvent", { content: "split " });
+ const second = frame("assistantResponseEvent", { content: "boundaries" });
+ const combined = concat([first, second]);
+ fetchMock.mockResolvedValueOnce(new Response(new ReadableStream({
+ start(controller) {
+ controller.enqueue(combined.slice(0, 9));
+ controller.enqueue(combined.slice(9, first.byteLength + 5));
+ controller.enqueue(combined.slice(first.byteLength + 5));
+ controller.close();
+ }
+ })));
+
+ const body = await (await execute()).response.text();
+
+ expect(body).toContain('"content":"split "');
+ expect(body).toContain('"content":"boundaries"');
+ expect(body).toContain('"finish_reason":"stop"');
+ });
+
+ it("accepts messageStop without semantic output as explicit completion", async () => {
+ fetchMock.mockResolvedValueOnce(response([frame("messageStopEvent", {})]));
+
+ const body = await (await execute()).response.text();
+
+ expect(fetchMock).toHaveBeenCalledTimes(1);
+ expect(body).toContain('"finish_reason":"stop"');
+ expect(body).not.toContain("kiro_missing_terminal");
+ });
+
+ it.each(["...", "…"])("repairs exact ellipsis final %s without leaking it", async (ellipsis) => {
+ fetchMock
+ .mockResolvedValueOnce(response([frame("assistantResponseEvent", { content: ellipsis })]))
+ .mockResolvedValueOnce(response([frame("assistantResponseEvent", { content: "Recovered answer." })]));
+
+ const body = await (await execute()).response.text();
+
+ expect(fetchMock).toHaveBeenCalledTimes(2);
+ expect(body).toContain("Recovered answer.");
+ expect(body).not.toContain(`"content":"${ellipsis}"`);
+ });
+
+ it.each([
+ "接下來我只再確認部署結果。",
+ "我會重新抓取最新日誌並確認結果。",
+ "目前證據顯示只在 **03:48:30–03:49:00 TPE** 出現少量 NonKA 504；主池 106/106、副池 50/50，且兩池都沒有重啟。最後補查 504 access log，確認 host／路徑與是否為集中流量。",
+ "Next I'll verify the deployment logs.",
+ "Let me check the remaining failures."
+ ])("repairs conservative future-action final: %s", async (progress) => {
+ fetchMock
+ .mockResolvedValueOnce(response([frame("assistantResponseEvent", { content: progress })]))
+ .mockResolvedValueOnce(response([frame("assistantResponseEvent", { content: "Verification completed." })]));
+
+ const body = await (await execute()).response.text();
+
+ expect(fetchMock).toHaveBeenCalledTimes(2);
+ expect(body).toContain("Verification completed.");
+ expect(body).not.toContain(progress);
+ });
+
+ it.each([
+ "Working...",
+ "I'll check the logs. They show no errors and deployment succeeded.",
+ "Let me check: status is 200 and the checksum matches abc123.",
+ "我會檢查版本。版本是 1.2.3。",
+ "接下來請你先批准部署，我會等待你的確認。",
+ "已完成驗證，所有測試均通過。",
+ "目前證據顯示只有少量 504，且主副池均未重啟。",
+ "目前證據顯示只有少量 504。最後補查結果顯示沒有集中流量。",
+ "目前證據顯示只有少量 504。最後補查，結果顯示沒有集中流量。",
+ "目前證據顯示只有少量 504。最後補查：結果顯示沒有集中流量。",
+ "目前證據顯示只有少量 504。最後補查 504 access log，結果顯示沒有集中流量。",
+ "目前證據顯示只有少量 504。最後補查 504 access log，確認 host／路徑與有無集中流量：無集中流量。",
+ "目前證據顯示只有少量 504。最後補查 504 access log，確認 host／路徑與是否為集中流量（答案是否定的）。",
+ "目前證據顯示只有少量 504。最後補充兩點已確認的結果。",
+ "The verification is complete and all tests passed."
+ ])("does not retry legitimate final: %s", async (finalText) => {
+ fetchMock.mockResolvedValueOnce(response([
+ frame("assistantResponseEvent", { content: finalText })
+ ]));
+
+ const body = await (await execute()).response.text();
+
+ expect(fetchMock).toHaveBeenCalledTimes(1);
+ expect(body).toContain(finalText);
+ });
+
+ it("bounds incomplete-final repair to one retry", async () => {
+ fetchMock
+ .mockResolvedValueOnce(response([frame("assistantResponseEvent", { content: "..." })]))
+ .mockResolvedValueOnce(response([frame("assistantResponseEvent", { content: "…" })]));
+
+ const body = await (await execute()).response.text();
+
+ expect(fetchMock).toHaveBeenCalledTimes(2);
+ expect(body).toContain("kiro_ellipsis_retry_failed");
+ expect(body).not.toContain('"content":"..."');
+ });
+
+ it("repairs malformed wrapper tools without leaking the invalid call", async () => {
+ fetchMock
+ .mockResolvedValueOnce(response([frame("toolUseEvent", {
+ toolUseId: "bad",
+ name: "tool_call",
+ input: { arguments: { q: "router" } }
+ })]))
+ .mockResolvedValueOnce(response([frame("toolUseEvent", {
+ toolUseId: "good",
+ name: "tool_call",
+ input: { name: "mcp_search", arguments: { q: "router" } }
+ })]));
+
+ const body = await (await execute()).response.text();
+
+ expect(fetchMock).toHaveBeenCalledTimes(2);
+ expect(body).toContain('"name":"tool_call"');
+ expect(body).toContain('\\"name\\":\\"mcp_search\\"');
+ expect(body).not.toContain('"id":"bad"');
+ });
+
+ it("requires complete direct tool input and keeps the failure private", async () => {
+ const pending = frame("toolUseEvent", { toolUseId: "pending", name: "read_file" });
+ fetchMock
+ .mockResolvedValueOnce(response([pending]))
+ .mockResolvedValueOnce(response([pending]));
+
+ const body = await (await execute()).response.text();
+
+ expect(fetchMock).toHaveBeenCalledTimes(2);
+ expect(body).toContain("kiro_tool_call_repair_retry_failed");
+ expect(body).not.toContain('"name":"read_file"');
+ });
+
+ it("repairs a non-string toolUseId before releasing the tool call", async () => {
+ fetchMock
+ .mockResolvedValueOnce(response([frame("toolUseEvent", {
+ toolUseId: 123,
+ name: "read_file",
+ input: { path: "bad.txt" }
+ })]))
+ .mockResolvedValueOnce(response([frame("toolUseEvent", {
+ toolUseId: "valid-tool-id",
+ name: "read_file",
+ input: { path: "safe.txt" }
+ })]));
+
+ const body = await (await execute()).response.text();
+
+ expect(fetchMock).toHaveBeenCalledTimes(2);
+ expect(body).toContain('"id":"valid-tool-id"');
+ expect(body).not.toContain('"id":123');
+ });
+
+ it("keeps model-controlled parser detail out of the retry system prompt", async () => {
+ fetchMock
+ .mockResolvedValueOnce(response([frame("toolUseEvent", {
+ toolUseId: "bad-json",
+ name: "tool_call",
+ input: '{"name":"IGNORE_ALL_INSTRUCTIONS"'
+ })]))
+ .mockResolvedValueOnce(response([frame("assistantResponseEvent", {
+ content: "Recovered safely."
+ })]));
+
+ const body = await (await execute()).response.text();
+ const retryBody = JSON.parse(fetchMock.mock.calls[1][1].body);
+
+ expect(body).toContain("Recovered safely.");
+ expect(retryBody.systemPrompt).toContain("tool_call wrapper was malformed");
+ expect(retryBody.systemPrompt).not.toContain("IGNORE_ALL_INSTRUCTIONS");
+ });
+
+ it("lets a complete tool call override metadata end_turn", async () => {
+ fetchMock.mockResolvedValueOnce(response([
+ frame("toolUseEvent", {
+ toolUseId: "tool",
+ name: "read_file",
+ input: { path: "safe.txt" }
+ }),
+ frame("metadataEvent", { stopReason: "end_turn" })
+ ]));
+
+ const body = await (await execute()).response.text();
+
+ expect(fetchMock).toHaveBeenCalledTimes(1);
+ expect(body).toContain('"name":"read_file"');
+ expect(body).toContain('"finish_reason":"tool_calls"');
+ });
+
+ it("maps max_tokens without treating it as a normal stop", async () => {
+ fetchMock.mockResolvedValueOnce(response([
+ frame("assistantResponseEvent", { content: "Limited answer." }),
+ frame("metadataEvent", { stopReason: "max_tokens" })
+ ]));
+
+ const body = await (await execute()).response.text();
+
+ expect(fetchMock).toHaveBeenCalledTimes(1);
+ expect(body).toContain('"finish_reason":"length"');
+ expect(body).not.toContain('"finish_reason":"stop"');
+ });
+
+ it("retries malformed_model_output once without semantic leakage", async () => {
+ fetchMock
+ .mockResolvedValueOnce(response([
+ frame("assistantResponseEvent", { content: "private malformed output" }),
+ frame("metadataEvent", { stopReason: "malformed_model_output" })
+ ]))
+ .mockResolvedValueOnce(response([
+ frame("assistantResponseEvent", { content: "Recovered protocol output." })
+ ]));
+
+ const body = await (await execute()).response.text();
+
+ expect(fetchMock).toHaveBeenCalledTimes(2);
+ expect(body).toContain("Recovered protocol output.");
+ expect(body).not.toContain("private malformed output");
+ });
+
+ it.each([
+ ["cancelled", "kiro_terminal_incomplete"],
+ ["pause_turn", "kiro_terminal_incomplete"],
+ ["content_filtered", "kiro_terminal_refusal"],
+ ["novel_reason", "kiro_unknown_stop_reason"]
+ ])("fails closed for stop reason %s", async (stopReason, code) => {
+ fetchMock.mockResolvedValueOnce(response([
+ frame("assistantResponseEvent", { content: `private-${stopReason}` }),
+ frame("metadataEvent", { stopReason })
+ ]));
+
+ const body = await (await execute()).response.text();
+
+ expect(fetchMock).toHaveBeenCalledTimes(1);
+ expect(body).toContain(code);
+ expect(body).not.toContain(`private-${stopReason}`);
+ expect(body).not.toContain('"finish_reason":"stop"');
+ });
+
+ it.each([
+ [
+ frame("messageStopEvent", { stopReason: "content_filtered" }),
+ frame("metadataEvent", { stopReason: "end_turn" })
+ ],
+ [
+ frame("metadataEvent", { stopReason: "end_turn" }),
+ frame("messageStopEvent", { stopReason: "content_filtered" })
+ ]
+ ])("preserves the most restrictive conflicting stop reason", async (...stopFrames) => {
+ fetchMock.mockResolvedValueOnce(response([
+ frame("assistantResponseEvent", { content: "private filtered output" }),
+ ...stopFrames
+ ]));
+
+ const body = await (await execute()).response.text();
+
+ expect(fetchMock).toHaveBeenCalledTimes(1);
+ expect(body).toContain("kiro_terminal_refusal");
+ expect(body).not.toContain("private filtered output");
+ });
+
+ it("prefers a non-retryable terminal reason over an earlier retryable reason", async () => {
+ fetchMock.mockResolvedValueOnce(response([
+ frame("assistantResponseEvent", { content: "private malformed output" }),
+ frame("metadataEvent", { stopReason: "malformed_model_output" }),
+ frame("messageStopEvent", { stopReason: "cancelled" })
+ ]));
+
+ const body = await (await execute()).response.text();
+
+ expect(fetchMock).toHaveBeenCalledTimes(1);
+ expect(body).toContain("kiro_terminal_incomplete");
+ expect(body).toContain('"stop_reason":"cancelled"');
+ expect(body).not.toContain("private malformed output");
+ });
+
+ it("preserves an authoritative refusal returned by the bounded retry", async () => {
+ fetchMock
+ .mockResolvedValueOnce(response([]))
+ .mockResolvedValueOnce(response([
+ frame("assistantResponseEvent", { content: "private filtered retry" }),
+ frame("metadataEvent", { stopReason: "content_filtered" })
+ ]));
+
+ const body = await (await execute()).response.text();
+
+ expect(fetchMock).toHaveBeenCalledTimes(2);
+ expect(body).toContain("kiro_terminal_refusal");
+ expect(body).not.toContain("kiro_missing_terminal_retry_failed");
+ expect(body).not.toContain("private filtered retry");
+ });
+
+ it.each([
+ ["max_tokens", "kiro_terminal_incomplete"],
+ ["cancelled", "kiro_terminal_incomplete"],
+ ["content_filtered", "kiro_terminal_refusal"],
+ ["novel_reason", "kiro_unknown_stop_reason"]
+ ])("does not let a valid tool override failure stop reason %s", async (stopReason, code) => {
+ fetchMock.mockResolvedValueOnce(response([
+ frame("toolUseEvent", {
+ toolUseId: "blocked-tool",
+ name: "read_file",
+ input: { path: "secret.txt" }
+ }),
+ frame("metadataEvent", { stopReason })
+ ]));
+
+ const body = await (await execute()).response.text();
+
+ expect(fetchMock).toHaveBeenCalledTimes(1);
+ expect(body).toContain(code);
+ expect(body).not.toContain('"name":"read_file"');
+ });
+
+ it.each(["content_filtered", "cancelled", "max_tokens"])(
+ "classifies failure %s before validating a malformed deferred tool",
+ async (stopReason) => {
+ fetchMock.mockResolvedValueOnce(response([
+ frame("toolUseEvent", { toolUseId: "bad-tool", name: "read_file" }),
+ frame("metadataEvent", { stopReason })
+ ]));
+
+ const body = await (await execute()).response.text();
+
+ expect(fetchMock).toHaveBeenCalledTimes(1);
+ expect(body).toContain(stopReason === "content_filtered"
+ ? "kiro_terminal_refusal"
+ : "kiro_terminal_incomplete");
+ expect(body).not.toContain("kiro_tool_call_repair_retry_failed");
+ expect(body).not.toContain('"name":"read_file"');
+ }
+ );
+
+ it.each([
+ ["content_filtered", [frame("toolUseEvent", {
+ toolUseId: 123,
+ name: "read_file",
+ input: { path: "bad.txt" }
+ })], "kiro_terminal_refusal"],
+ ["cancelled", [frame("toolUseEvent", {
+ toolUseId: "missing-name",
+ input: { path: "bad.txt" }
+ })], "kiro_terminal_incomplete"],
+ ["max_tokens", [
+ frame("toolUseEvent", { toolUseId: "changing", name: "read_file" }),
+ frame("toolUseEvent", { toolUseId: "changing", name: "write_file" })
+ ], "kiro_terminal_incomplete"]
+ ])("continues past eager tool-shape errors to authoritative stop %s", async (stopReason, toolFrames, code) => {
+ fetchMock.mockResolvedValueOnce(response([
+ ...toolFrames,
+ frame("metadataEvent", { stopReason })
+ ]));
+
+ const body = await (await execute()).response.text();
+
+ expect(fetchMock).toHaveBeenCalledTimes(1);
+ expect(body).toContain(code);
+ expect(body).not.toContain("kiro_tool_call_repair_retry_failed");
+ expect(body).not.toContain('"tool_calls"');
+ });
+
+ it("retries a TTFT timeout once while preserving cancellation semantics", async () => {
+ process.env.KIRO_TOOL_CALL_REPAIR_TTFT_TIMEOUT_MS = "1";
+ fetchMock
+ .mockResolvedValueOnce(controlledResponse().value)
+ .mockResolvedValueOnce(response([
+ frame("assistantResponseEvent", { content: "Recovered after timeout." })
+ ]));
+
+ const body = await (await execute()).response.text();
+
+ expect(fetchMock).toHaveBeenCalledTimes(2);
+ expect(body).toContain("Recovered after timeout.");
+ });
+
+ it("treats validated non-semantic frames as watchdog activity", async () => {
+ process.env.KIRO_TOOL_CALL_REPAIR_TTFT_TIMEOUT_MS = "30";
+ process.env.KIRO_TOOL_CALL_REPAIR_STALL_TIMEOUT_MS = "30";
+ const upstream = controlledResponse();
+ fetchMock.mockResolvedValueOnce(upstream.value);
+ setTimeout(() => upstream.enqueue(frame("meteringEvent", { usage: 1 })), 20);
+ setTimeout(() => upstream.enqueue(frame("contextUsageEvent", { contextUsagePercentage: 5 })), 40);
+ setTimeout(() => {
+ upstream.enqueue(frame("assistantResponseEvent", { content: "Completed after active frames." }));
+ upstream.close();
+ }, 60);
+
+ const body = await (await execute()).response.text();
+
+ expect(fetchMock).toHaveBeenCalledTimes(1);
+ expect(body).toContain("Completed after active frames.");
+ });
+
+ it("retries a response-body read failure once", async () => {
+ fetchMock
+ .mockResolvedValueOnce(new Response(new ReadableStream({
+ start(controller) {
+ controller.error(new Error("socket reset"));
+ }
+ })))
+ .mockResolvedValueOnce(response([
+ frame("assistantResponseEvent", { content: "Recovered after read failure." })
+ ]));
+
+ const body = await (await execute()).response.text();
+
+ expect(fetchMock).toHaveBeenCalledTimes(2);
+ expect(body).toContain("Recovered after read failure.");
+ expect(body).not.toContain("socket reset");
+ });
+
+ it.each([
+ ["message CRC", () => {
+ const corrupt = frame("assistantResponseEvent", { content: "corrupt CRC" });
+ corrupt[corrupt.byteLength - 1] ^= 0xff;
+ return [corrupt];
+ }],
+ ["prelude CRC", () => {
+ const corrupt = frame("assistantResponseEvent", { content: "corrupt prelude" });
+ corrupt[8] ^= 0xff;
+ return [corrupt];
+ }],
+ ["truncated frame", () => {
+ const truncated = frame("assistantResponseEvent", { content: "truncated" });
+ return [truncated.slice(0, -3)];
+ }],
+ ["out-of-bounds headers", () => {
+ const corrupt = frame("assistantResponseEvent", { content: "bad headers" });
+ new DataView(corrupt.buffer).setUint32(4, corrupt.byteLength - 15, false);
+ return [checksum(corrupt)];
+ }],
+ ["duplicate headers", () => [
+ frameFromEntries([
+ [":event-type", "assistantResponseEvent"],
+ [":event-type", "metadataEvent"]
+ ], { content: "duplicate" })
+ ]]
+ ])("retries %s and releases only the valid attempt", async (_name, invalidFrames) => {
+ fetchMock
+ .mockResolvedValueOnce(response([
+ frame("assistantResponseEvent", { content: "must stay private" }),
+ ...invalidFrames()
+ ]))
+ .mockResolvedValueOnce(response([
+ frame("assistantResponseEvent", { content: "Recovered after validation." })
+ ]));
+
+ const body = await (await execute()).response.text();
+
+ expect(fetchMock).toHaveBeenCalledTimes(2);
+ expect(body).toContain("Recovered after validation.");
+ expect(body).not.toContain("must stay private");
+ });
+
+ it("reports corrupt-frame provenance when the bounded retry also fails", async () => {
+ const corruptFrame = () => {
+ const corrupt = frame("assistantResponseEvent", { content: "corrupt" });
+ corrupt[corrupt.byteLength - 1] ^= 0xff;
+ return corrupt;
+ };
+ fetchMock
+ .mockResolvedValueOnce(response([corruptFrame()]))
+ .mockResolvedValueOnce(response([corruptFrame()]));
+
+ const body = await (await execute()).response.text();
+
+ expect(body).toContain("kiro_missing_terminal_retry_failed");
+ expect(body).toContain('"terminal_provenance":"corrupt_eventstream_frame"');
+ expect(body).toContain('"transport_state":"corrupt_frame"');
+ });
+
+ it("caps diagnostic event-type cardinality", async () => {
+ let terminal;
+ const executor = new KiroExecutor();
+ const frames = Array.from({ length: 100 }, (_, index) =>
+ frame(`unknownEvent${index}`, { index })
+ );
+ frames.push(frame("assistantResponseEvent", { content: "done" }));
+ const transformed = executor.transformEventStreamToSSE(
+ response(frames),
+ "kr/claude-opus-4.8",
+ { onTerminalState: (value) => { terminal = value; } }
+ );
+
+ await transformed.text();
+
+ expect(terminal.event_counts).toEqual({
+ other: 100,
+ assistantResponseEvent: 1
+ });
+ });
+
+ it("rejects a raw chunk before concatenating beyond the protocol bound", async () => {
+ let terminal;
+ const executor = new KiroExecutor();
+ const transformed = executor.transformEventStreamToSSE(
+ response([new Uint8Array(65)]),
+ "kr/claude-opus-4.8",
+ {
+ maxRawBytes: 64,
+ onTerminalState: (value) => { terminal = value; }
+ }
+ );
+
+ const body = await transformed.text();
+
+ expect(body).toContain("buffered bytes exceed the protocol bound");
+ expect(terminal.terminal_provenance).toBe("corrupt_eventstream_frame");
+ });
+
+ it.each(["error", "exception"])("propagates EventStream %s without retry or leakage", async (messageType) => {
+ fetchMock.mockResolvedValueOnce(response([
+ frame("assistantResponseEvent", { content: "must stay private" }),
+ frameFromEntries([
+ [":message-type", messageType],
+ ...(messageType === "exception" ? [[":exception-type", "InternalServerException"]] : [])
+ ], { message: "upstream failed" })
+ ]));
+
+ const body = await (await execute()).response.text();
+
+ expect(fetchMock).toHaveBeenCalledTimes(1);
+ expect(body).toContain("kiro_upstream_eventstream_error");
+ expect(body).toContain("upstream failed");
+ expect(body).not.toContain("must stay private");
+ });
+
+ it("surfaces retry HTTP failures as SSE after heartbeat commits headers", async () => {
+ fetchMock
+ .mockResolvedValueOnce(response([]))
+ .mockResolvedValueOnce(new Response("unauthorized", {
+ status: 401,
+ statusText: "Unauthorized"
+ }));
+
+ const result = await execute();
+ const body = await result.response.text();
+
+ expect(result.response.status).toBe(200);
+ expect(body).toContain("kiro_integrity_retry_upstream_error");
+ expect(body).toContain("unauthorized");
+ });
+
+ it("bounds the retry HTTP error body", async () => {
+ fetchMock
+ .mockResolvedValueOnce(response([]))
+ .mockResolvedValueOnce(new Response(`error-start-${"x".repeat(10_000)}-error-tail`, {
+ status: 401,
+ statusText: "Unauthorized"
+ }));
+
+ const body = await (await execute()).response.text();
+
+ expect(body).toContain("error-start-");
+ expect(body).not.toContain("error-tail");
+ expect(body.length).toBeLessThan(5000);
+ });
+
+ it("propagates cancellation while validation is waiting for EOF", async () => {
+ const upstream = controlledResponse([
+ frame("assistantResponseEvent", { content: "waiting" })
+ ]);
+ fetchMock.mockResolvedValueOnce(upstream.value);
+ const abort = new AbortController();
+
+ const result = await execute(new KiroExecutor(), { signal: abort.signal });
+ const reader = result.response.body.getReader();
+ await reader.read();
+ abort.abort("client cancelled");
+
+ await expect(reader.read()).rejects.toMatchObject({ name: "AbortError" });
+ });
+
+ it("fails safely when the private gate exceeds its configured bound", async () => {
+ process.env.KIRO_TOOL_CALL_REPAIR_BUFFER_MAX_BYTES = "8";
+ fetchMock.mockResolvedValueOnce(response([
+ frame("assistantResponseEvent", { content: "larger than eight bytes" })
+ ]));
+
+ const body = await (await execute()).response.text();
+
+ expect(fetchMock).toHaveBeenCalledTimes(1);
+ expect(body).toContain("integrity buffer exceeded");
+ expect(body).not.toContain("larger than eight bytes");
+ });
+
+ it("counts deferred tool fragments against the private memory bound", async () => {
+ process.env.KIRO_TOOL_CALL_REPAIR_BUFFER_MAX_BYTES = "128";
+ fetchMock.mockResolvedValueOnce(response([
+ frame("toolUseEvent", {
+ toolUseId: "large-tool",
+ name: "read_file",
+ input: { path: "x".repeat(200) }
+ })
+ ]));
+
+ const body = await (await execute()).response.text();
+
+ expect(fetchMock).toHaveBeenCalledTimes(1);
+ expect(body).toContain("kiro_integrity_buffer_exceeded");
+ expect(body).not.toContain('"name":"read_file"');
+ });
+});
diff --git a/tests/unit/kiro-thinking-strip.test.js b/tests/unit/kiro-thinking-strip.test.js
index 1fff6e9ea8..cc20aa4740 100644
--- a/tests/unit/kiro-thinking-strip.test.js
+++ b/tests/unit/kiro-thinking-strip.test.js
@@ -32,10 +32,23 @@ function createMockFrame(eventType, payloadObj) {
offset += headerValueBytes.length;
buffer.set(payloadBytes, offset);
-
+
+ view.setUint32(8, crc32(buffer.subarray(0, 8)), false);
+ view.setUint32(totalLength - 4, crc32(buffer.subarray(0, totalLength - 4)), false);
return buffer;
}
+function crc32(bytes) {
+ let crc = 0xffffffff;
+ for (const byte of bytes) {
+ crc ^= byte;
+ for (let bit = 0; bit < 8; bit++) {
+ crc = (crc >>> 1) ^ ((crc & 1) ? 0xedb88320 : 0);
+ }
+ }
+ return (crc ^ 0xffffffff) >>> 0;
+}
+
async function readAllSSE(stream) {
const reader = stream.getReader();
const decoder = new TextDecoder();
@@ -130,14 +143,16 @@ describe("KiroExecutor thinking tag stripping", () => {
expect(contentChunks.length).toBe(0);
});
- it("emits a terminal chunk at messageStop before the upstream stream closes", async () => {
+ it("waits for clean EOF before emitting stop after messageStop", async () => {
const executor = new KiroExecutor();
const f1 = createMockFrame("assistantResponseEvent", { content: "OK" });
const f2 = createMockFrame("messageStopEvent", {});
+ let upstreamController;
const readableStream = new ReadableStream({
start(controller) {
+ upstreamController = controller;
controller.enqueue(f1);
controller.enqueue(f2);
}
@@ -147,11 +162,16 @@ describe("KiroExecutor thinking tag stripping", () => {
const reader = transformedResponse.body.getReader();
const decoder = new TextDecoder();
let output = "";
- for (let i = 0; i < 4 && !output.includes("\"finish_reason\":\"stop\""); i++) {
- const { value } = await readNextWithTimeout(reader);
- output += decoder.decode(value, { stream: true });
+ const { value } = await readNextWithTimeout(reader);
+ output += decoder.decode(value, { stream: true });
+ expect(output).not.toContain("\"finish_reason\":\"stop\"");
+
+ upstreamController.close();
+ while (!output.includes("\"finish_reason\":\"stop\"")) {
+ const { value: nextValue, done } = await readNextWithTimeout(reader);
+ if (done) break;
+ output += decoder.decode(nextValue, { stream: true });
}
- await reader.cancel();
expect(output).toContain("\"finish_reason\":\"stop\"");
});