fix(mitm): Kiro binary EventStream crash + add models & TTS tool filtering

- server.js: isBinaryData() skips binary AWS EventStream bodies (fix JSON parse crash)
- kiro.js: isBinaryEventStream detection + migrate to pipeTransformedEventStream pipeline
- base.js: add pipeTransformedSSE / pipeTransformedEventStream helpers
- chatCore.js: filter tool messages + tools for TTS models via getModelType()
- providerModels.js: add getModelType()
- cliTools.js: add gpt-5-mini (Copilot), glm-5 & minimax-m2.5 (Kiro)

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Giao Ho
2026-06-06 11:31:52 +07:00
committed by decolua
co-authored by Cursor
parent c233c7c8fc
commit 0850f0a470
6 changed files with 439 additions and 117 deletions
+7
View File
@@ -858,6 +858,13 @@ export function getModelTargetFormat(aliasOrId, modelId) {
return found?.targetFormat || null;
}
export function getModelType(aliasOrId, modelId) {
const models = PROVIDER_MODELS[aliasOrId];
if (!models) return null;
const found = models.find(m => m.id === modelId);
return found?.type || null;
}
export function getModelUpstreamId(aliasOrId, modelId) {
const models = PROVIDER_MODELS[aliasOrId];
const found = models?.find(m => m.id === modelId);
+7 -1
View File
@@ -5,7 +5,7 @@ import { COLORS } from "../utils/stream.js";
import { createStreamController } from "../utils/streamHandler.js";
import { refreshWithRetry } from "../services/tokenRefresh.js";
import { createRequestLogger } from "../utils/requestLogger.js";
import { getModelTargetFormat, getModelStrip, getModelUpstreamId, PROVIDER_ID_TO_ALIAS } from "../config/providerModels.js";
import { getModelTargetFormat, getModelStrip, getModelUpstreamId, getModelType, PROVIDER_ID_TO_ALIAS } from "../config/providerModels.js";
import { createErrorResult, parseUpstreamError, formatProviderError } from "../utils/error.js";
import { HTTP_STATUS } from "../config/runtimeConfig.js";
import { handleBypassRequest } from "../utils/bypassHandler.js";
@@ -115,6 +115,12 @@ export async function handleChatCore({ body, modelInfo, credentials, log, onCred
// Covers both passthrough (source shape) and translated (target shape) flows
const finalFormat = passthrough ? sourceFormat : targetFormat;
// TTS models don't support tool messages/function calling
if (getModelType(alias, model) === "tts" && translatedBody.messages) {
translatedBody.messages = translatedBody.messages.filter(msg => msg.role !== "tool");
delete translatedBody.tools;
}
// RTK: compress tool_result content
const rtkStats = compressMessages(translatedBody, rtkEnabled);
const rtkLine = formatRtkLog(rtkStats);
+159 -1
View File
@@ -65,4 +65,162 @@ async function pipeSSE(routerRes, res, dumper) {
}
}
module.exports = { fetchRouter, pipeSSE };
/**
* Pipe SSE stream from router, transforming each chunk through a user function.
* Reads SSE data: lines, parses JSON, calls transformFn(parsed, state),
* and writes returned SSE strings to the client response.
*
* @param {Response} routerRes - Fetch Response from 9Router
* @param {http.ServerResponse} res - Client response
* @param {Function} transformFn - (parsedChunk, state) => string|string[]|null
* @param {object} state - Mutable state object shared across chunks and flush
*/
async function pipeTransformedSSE(routerRes, res, transformFn, state) {
const ct = routerRes.headers.get("content-type") || "application/json";
const resHeaders = { "Content-Type": ct, "Cache-Control": "no-cache", "Connection": "keep-alive" };
if (ct.includes("text/event-stream")) resHeaders["X-Accel-Buffering"] = "no";
res.writeHead(200, resHeaders);
if (!routerRes.body) {
res.end(await routerRes.text().catch(() => ""));
return;
}
const reader = routerRes.body.getReader();
const decoder = new TextDecoder("utf-8", { fatal: false });
let buffer = "";
while (true) {
const { done, value } = await reader.read();
if (done) break;
buffer += decoder.decode(value, { stream: true });
const lines = buffer.split("\n");
buffer = lines.pop() || "";
for (const line of lines) {
const trimmed = line.trim();
if (!trimmed || !trimmed.startsWith("data:")) continue;
const data = trimmed.slice(5).trim();
if (data === "[DONE]") continue;
if (process.env.DEBUG_MITM) {
log(`[SSE in] ${data.slice(0, 200)}`);
}
try {
const parsed = JSON.parse(data);
const result = transformFn(parsed, state);
if (result != null) {
const outputs = Array.isArray(result) ? result : [result];
for (const output of outputs) {
if (process.env.DEBUG_MITM) {
const len = output.length || output.byteLength || 0;
log(`[write binary frame] (${len}B) first 20B: ${Array.from(output.slice(0, 20)).join(',')}`);
}
res.write(Buffer.from(output));
}
}
} catch {
// Skip unparseable lines
}
}
}
// Flush: pass null to signal stream end
try {
const flushed = transformFn(null, state);
if (flushed != null) {
const outputs = Array.isArray(flushed) ? flushed : [flushed];
for (const output of outputs) {
res.write(output);
}
}
} catch { /* ignore flush errors */ }
res.end();
}
/**
* Pipe SSE stream from router, transforming each chunk through a user function,
* and writing binary EventStream frames to the client.
*
* Reads SSE data: lines, parses JSON, calls transformFn(parsed, state),
* and writes returned Uint8Array frames to the client response.
*
* @param {Response} routerRes - Fetch Response from 9Router
* @param {http.ServerResponse} res - Client response
* @param {Function} transformFn - (parsedChunk, state) => Uint8Array|Uint8Array[]|null
* @param {object} state - Mutable state object shared across chunks and flush
*/
async function pipeTransformedEventStream(routerRes, res, transformFn, state) {
const resHeaders = {
"Content-Type": "application/vnd.amazon.eventstream",
"Cache-Control": "no-cache",
"Connection": "keep-alive"
};
res.writeHead(200, resHeaders);
if (!routerRes.body) {
res.end(await routerRes.text().catch(() => ""));
return;
}
const reader = routerRes.body.getReader();
const decoder = new TextDecoder("utf-8", { fatal: false });
let buffer = "";
while (true) {
const { done, value } = await reader.read();
if (done) break;
buffer += decoder.decode(value, { stream: true });
const lines = buffer.split("\n");
buffer = lines.pop() || "";
for (const line of lines) {
const trimmed = line.trim();
if (!trimmed || !trimmed.startsWith("data:")) continue;
const data = trimmed.slice(5).trim();
if (data === "[DONE]") continue;
if (process.env.DEBUG_MITM) {
log(`[SSE in] ${data.slice(0, 200)}`);
}
try {
const parsed = JSON.parse(data);
const result = transformFn(parsed, state);
if (result != null) {
const outputs = Array.isArray(result) ? result : [result];
for (const output of outputs) {
if (process.env.DEBUG_MITM) {
const len = output.length || output.byteLength || 0;
log(`[write binary frame] (${len}B) first 20B: ${Array.from(output.slice(0, 20)).join(',')}`);
}
res.write(Buffer.from(output));
}
}
} catch {
// Skip unparseable lines
}
}
}
// Flush: pass null to signal stream end
try {
const flushed = transformFn(null, state);
if (flushed != null) {
const outputs = Array.isArray(flushed) ? flushed : [flushed];
for (const output of outputs) {
res.write(output);
}
}
} catch { /* ignore flush errors */ }
res.end();
}
module.exports = { fetchRouter, pipeSSE, pipeTransformedSSE, pipeTransformedEventStream };
+240 -115
View File
@@ -1,6 +1,6 @@
const { err } = require("../logger");
const { IS_DEV } = require("../config");
const { fetchRouter } = require("./base");
const { fetchRouter, pipeTransformedEventStream } = require("./base");
const fs = require("fs");
const path = require("path");
@@ -32,6 +32,76 @@ function crc32(buf) {
return (crc ^ 0xffffffff) >>> 0;
}
/**
* Initialize state for the Kiro response translator
*/
function initKiroState(modelId) {
return {
modelId: modelId || null, // Model name from first chunk
toolCallInit: {}, // { [index]: { id, name } } — tracks seen tools
hasToolCalls: false, // Whether this response uses tool calls
finishSent: false, // Whether termination has been emitted
usage: null, // Accumulated usage from usage-only chunks
inThink: false, // Whether inside a <thinking> block
thinkBuf: "" // Buffer for partial thinking content
};
}
/**
* Extract thinking blocks from text content.
* Handles both <thinking>...</thinking> and <think>...</think> tags,
* including partial tags split across SSE chunks.
*/
function extractThinking(text, state) {
if (!text) return { thinking: null, text: null };
let working = text;
// Prepend buffered partial thinking from previous chunk
if (state.inThink && state.thinkBuf) {
working = state.thinkBuf + working;
state.thinkBuf = "";
state.inThink = false;
}
// Match <thinking> or <think> opening tags
const startRe = /<thinking>|<think>/i;
const startMatch = working.match(startRe);
if (!startMatch) {
return { thinking: null, text: working };
}
const tag = startMatch[0].toLowerCase();
const closeTag = tag === "<think>" ? "</think>" : "</thinking>";
const startIdx = startMatch.index;
const endIdx = working.indexOf(closeTag, startIdx + tag.length);
if (endIdx === -1) {
// Opening tag without closing — buffer for next chunk
state.inThink = true;
state.thinkBuf = working.slice(startIdx);
const before = working.slice(0, startIdx).trim();
return { thinking: null, text: before || null };
}
// Complete block found
const thinking = working.slice(startIdx + tag.length, endIdx);
const before = working.slice(0, startIdx).trim();
const after = working.slice(endIdx + closeTag.length).trim();
const rest = [before, after].filter(Boolean).join("");
// Recursively process for more blocks
const recurse = rest
? extractThinking(rest, { inThink: false, thinkBuf: "" })
: { thinking: null, text: null };
return {
thinking: thinking || null,
text: recurse.text || null
};
}
// ─── AWS EventStream frame builder ────────────────────────────────────────────
/**
* Encode a single string header into the AWS EventStream binary format.
@@ -233,132 +303,173 @@ function extractTools(body) {
}
// ─── OpenAI SSE → EventStream binary conversion ───────────────────────────────
/**
* Read 9router's OpenAI SSE response and re-encode it as AWS EventStream binary
* frames that Kiro's Smithy SDK expects.
* Convert an OpenAI SSE chunk to AWS EventStream binary frame(s)
* This replaces pipeOpenAIasEventStream and works with pipeTransformedEventStream
*
* OpenAI SSE format: data: { choices:[{ delta:{ content:"..." } }] }\n\n
* EventStream events emitted:
* assistantResponseEvent { content: "..." } — one per SSE chunk with text
* toolUseEvent { toolUseId, name, input } — for tool calls
* messageStopEvent {} — on finish
* @param {object|null} chunk - Parsed OpenAI chat.completion.chunk, or null for flush
* @param {object} state - Mutable state object
* @returns {Uint8Array|Uint8Array[]|null} Binary EventStream frame(s) or null to skip
*/
async function pipeOpenAIasEventStream(routerRes, res) {
if (!routerRes.body) {
res.end(buildEventStreamFrame("messageStopEvent", {}));
return;
function convertOpenAIToKiro(chunk, state) {
// Flush: ensure clean stream termination
if (!chunk) {
if (state.finishSent) return null;
// Flush any remaining buffered thinking
if (state.inThink && state.thinkBuf) {
state.inThink = false;
const thinking = state.thinkBuf;
state.thinkBuf = "";
return buildEventStreamFrame("reasoningContentEvent", {
content: thinking,
modelId: state.modelId || "kiro-unknown"
});
}
return buildEventStreamFrame("messageStopEvent", {});
}
const reader = routerRes.body.getReader();
const decoder = new TextDecoder();
let sseBuffer = "";
let stopSent = false;
const frames = [];
const choice = chunk.choices?.[0];
const delta = choice?.delta || {};
// Accumulated tool-call state keyed by index
const toolCallAccum = {};
// Capture modelId from first chunk (real API includes it in every content frame)
if (!state.modelId && chunk.model) {
state.modelId = chunk.model;
}
const modelId = state.modelId || "unknown";
const sendStop = () => {
if (!stopSent) {
stopSent = true;
res.write(buildEventStreamFrame("messageStopEvent", {}));
}
};
// Handle usage (may arrive standalone or with other chunks)
if (chunk.usage) {
state.usage = chunk.usage;
}
try {
while (true) {
const { done, value } = await reader.read();
if (done) break;
// Handle tool calls — stream incrementally, matching real API format
if (delta.tool_calls) {
state.hasToolCalls = true;
for (const tc of delta.tool_calls) {
const idx = tc.index ?? 0;
sseBuffer += decoder.decode(value, { stream: true });
if (tc.id && tc.function?.name && !state.toolCallInit[idx]) {
// First appearance: emit frame with name + id, no input
state.toolCallInit[idx] = { id: tc.id, name: tc.function.name };
dbg(`toolUseEvent init: ${tc.function.name} (${tc.id})`);
frames.push(buildEventStreamFrame("toolUseEvent", {
name: tc.function.name,
toolUseId: tc.id
}));
}
// Split on newlines; keep the last (possibly incomplete) line in the buffer
const lines = sseBuffer.split("\n");
sseBuffer = lines.pop() ?? "";
for (const line of lines) {
const trimmed = line.trim();
if (!trimmed.startsWith("data:")) continue;
const raw = trimmed.slice(5).trim();
if (raw === "[DONE]") {
sendStop();
continue;
}
let chunk;
try { chunk = JSON.parse(raw); } catch { continue; }
const delta = chunk?.choices?.[0]?.delta;
if (!delta) continue;
// ── Text content ───────────────────────────────────────────────────────
if (delta.content) {
res.write(buildEventStreamFrame("assistantResponseEvent", { content: delta.content }));
}
// ── Tool calls (streamed in pieces by OpenAI SSE) ──────────────────────
if (delta.tool_calls) {
dbg(`TOOL_CALLS delta: ${JSON.stringify(delta.tool_calls).slice(0, 300)}`);
for (const tc of delta.tool_calls) {
const idx = tc.index ?? 0;
if (!toolCallAccum[idx]) {
toolCallAccum[idx] = { id: tc.id ?? "", name: "", args: "" };
}
const acc = toolCallAccum[idx];
if (tc.id) acc.id = tc.id;
if (tc.function?.name) acc.name += tc.function.name;
// safeArgsString prevents `"" + object` → "[object Object]" corruption
// when 9router's Anthropic→OpenAI conversion passes a pre-parsed object
const argType = typeof tc.function?.arguments;
if (tc.function?.arguments != null) {
acc.args += safeArgsString(tc.function.arguments);
}
dbg(` tc[${idx}] argType=${argType} id=${acc.id} name=${acc.name} args_so_far=${acc.args.slice(0, 100)}`);
}
}
// ── Finish ─────────────────────────────────────────────────────────────
const finish = chunk?.choices?.[0]?.finish_reason;
if (finish) {
dbg(`FINISH finish_reason=${finish} toolCallKeys=${JSON.stringify(Object.keys(toolCallAccum))}`);
// Flush accumulated tool calls before stop
if (finish === "tool_calls") {
for (const acc of Object.values(toolCallAccum)) {
// IMPORTANT: Kiro's internal tool dispatcher expects `input` to be a JSON string
// (not a parsed object). The real CodeWhisperer server sends:
// { toolUseId, name, input: "{\"key\":\"value\"}" } ← input is a string
// Kiro then JSON.parses that string to get the tool arguments.
// If we send input as a parsed object, Kiro does String(obj) → "[object Object]".
const inputStr = acc.args || "{}";
dbg(` toolUseEvent: id=${acc.id} name=${acc.name} inputStr=${inputStr.slice(0, 200)}`);
res.write(buildEventStreamFrame("toolUseEvent", {
toolUseId: acc.id,
name: acc.name,
input: inputStr, // Must be a JSON STRING, not a parsed object
}));
}
}
sendStop();
}
// Emit incremental input fragment
if (tc.function?.arguments) {
const init = state.toolCallInit[idx];
dbg(`toolUseEvent fragment: ${tc.function.arguments.slice(0, 100)}`);
frames.push(buildEventStreamFrame("toolUseEvent", {
input: tc.function.arguments,
name: init?.name || tc.function?.name || "",
toolUseId: init?.id || tc.id || ""
}));
}
}
} finally {
sendStop();
res.end();
}
// Handle explicit reasoning_content (type-specific thinking channel)
if (delta.reasoning_content) {
frames.push(buildEventStreamFrame("reasoningContentEvent", {
content: delta.reasoning_content,
modelId
}));
}
// Handle text content — extract thinking blocks, emit rest as assistantResponseEvent
if (delta.content) {
const { thinking, text } = extractThinking(delta.content, state);
if (thinking) {
frames.push(buildEventStreamFrame("reasoningContentEvent", {
content: thinking,
modelId
}));
}
if (text) {
frames.push(buildEventStreamFrame("assistantResponseEvent", {
content: text,
modelId
}));
}
}
// Handle finish_reason
if (choice?.finish_reason) {
const finishFrames = emitFinish(state);
if (finishFrames) {
frames.push(...(Array.isArray(finishFrames) ? finishFrames : [finishFrames]));
}
}
if (frames.length === 0) return null;
return frames.length === 1 ? frames[0] : frames;
}
/**
* Emit termination frames. For tool-call responses, emits stop:true per tool.
* For text-only responses, emits messageStopEvent.
*/
function emitFinish(state) {
const frames = [];
if (state.hasToolCalls) {
// Tool-call response: emit stop:true for each tool
for (const idx of Object.keys(state.toolCallInit).sort()) {
const tc = state.toolCallInit[idx];
frames.push(buildEventStreamFrame("toolUseEvent", {
name: tc.name,
stop: true,
toolUseId: tc.id
}));
}
} else {
// Text-only response: emit messageStopEvent
frames.push(buildEventStreamFrame("messageStopEvent", {}));
}
state.finishSent = true;
// Emit usage if available
if (state.usage) {
frames.push(buildEventStreamFrame("usageEvent", {
inputTokens: state.usage.prompt_tokens || 0,
outputTokens: state.usage.completion_tokens || 0
}));
}
state.toolCallInit = {};
return frames.length > 0 ? frames : null;
}
// ─── MITM intercept entry point ───────────────────────────────────────────────
/**
* Intercept Kiro IDE CodeWhisperer request:
* 1. Parse CodeWhisperer binary/JSON body
* 2. Convert to OpenAI messages[] format
* Intercept Kiro IDE CodeWhisperer request and convert to EventStream response:
* 1. Parse CodeWhisperer JSON body (reject binary EventStream formats)
* 2. Convert CodeWhisperer format to OpenAI messages[] format
* 3. Forward to 9router /v1/chat/completions (OpenAI SSE)
* 4. Convert OpenAI SSE response → AWS EventStream binary frames
* 5. Stream binary frames back to Kiro
* 5. Stream EventStream frames back to Kiro IDE
*
* @param {http.IncomingMessage} req - HTTP request from Kiro IDE
* @param {http.ServerResponse} res - HTTP response to Kiro IDE
* @param {Buffer} bodyBuffer - Request body buffer
* @param {string} mappedModel - Model name after MITM alias mapping
*/
async function intercept(req, res, bodyBuffer, mappedModel) {
try {
// Detect and handle binary data (e.g., continuation requests with EventStream frames)
if (isBinaryEventStream(bodyBuffer)) {
// Binary EventStream requests are typically continuation/streaming frames
// that don't contain model info - pass them through directly to avoid JSON.parse crash
throw new Error(`Binary EventStream format detected (${bodyBuffer.length}B) - request should use passthrough instead of intercept`);
}
const body = JSON.parse(bodyBuffer.toString());
// 1 + 2: CodeWhisperer → OpenAI messages + tools
@@ -380,22 +491,36 @@ async function intercept(req, res, bodyBuffer, mappedModel) {
// 3: Forward to 9router
const routerRes = await fetchRouter(openaiBody, "/v1/chat/completions", req.headers);
// 4 + 5: Re-encode response as AWS EventStream binary
res.writeHead(routerRes.status, {
"Content-Type": "application/vnd.amazon.eventstream",
"x-amzn-requestid": `mitm-${Date.now()}`,
"x-amz-id-2": "mitm",
"Transfer-Encoding": "chunked",
});
// 4 + 5: Re-encode response as AWS EventStream binary using standard pipeline
const state = initKiroState(mappedModel);
await pipeOpenAIasEventStream(routerRes, res);
await pipeTransformedEventStream(routerRes, res, convertOpenAIToKiro, state);
} catch (error) {
err(`[Kiro] ${error.message}`);
err(`[Kiro MITM] Request processing failed: ${error.message}`);
if (!res.headersSent) {
res.writeHead(500, { "Content-Type": "application/json" });
}
res.end(JSON.stringify({ error: { message: error.message, type: "mitm_error" } }));
res.end(JSON.stringify({
error: {
message: error.message,
type: "mitm_error",
handler: "kiro"
}
}));
}
}
// Detect AWS EventStream binary format
function isBinaryEventStream(buffer) {
if (!buffer || buffer.length < 12) return false;
// AWS EventStream signature:
// - First 4 bytes: total frame length (big-endian)
// - Bytes 4-8: headers length (big-endian)
// - Typical frame length: 100-10000 bytes
const totalLen = buffer.readUInt32BE(0);
const headersLen = buffer.readUInt32BE(4);
// Sanity checks: frame length should be reasonable and headers should fit
return totalLen > 12 && totalLen < 1000000 && headersLen < totalLen - 12;
}
module.exports = { intercept };
+23
View File
@@ -95,6 +95,10 @@ function collectBodyRaw(req) {
function extractModel(url, body) {
const urlMatch = url.match(/\/models\/([^/:]+)/);
if (urlMatch) return urlMatch[1];
// Skip parsing if body is binary (AWS EventStream, Protocol Buffers, etc.)
if (isBinaryData(body)) return null;
try {
const parsed = JSON.parse(body.toString());
if (parsed.conversationState) {
@@ -104,6 +108,25 @@ function extractModel(url, body) {
} catch { return null; }
}
// Detect binary data vs JSON text
function isBinaryData(buffer) {
if (!buffer || buffer.length === 0) return false;
// AWS EventStream signature: first 4 bytes = frame length (big-endian uint32)
// Check for non-printable chars in first 100 bytes (common in binary protocols)
const sample = buffer.slice(0, Math.min(100, buffer.length));
let nonPrintable = 0;
for (let i = 0; i < sample.length; i++) {
const byte = sample[i];
// Count non-ASCII printable chars (excluding whitespace)
if (byte < 0x20 && byte !== 0x09 && byte !== 0x0A && byte !== 0x0D) {
nonPrintable++;
}
if (byte > 0x7E) nonPrintable++;
}
// If >30% non-printable, treat as binary
return (nonPrintable / sample.length) > 0.3;
}
function getMappedModel(tool, model) {
if (!model) return null;
try {
+3
View File
@@ -38,6 +38,7 @@ export const MITM_TOOLS = {
defaultModels: [
{ id: "gpt-4o", name: "GPT-4o", alias: "gpt-4o" },
{ id: "gpt-4.1", name: "GPT-4.1", alias: "gpt-4.1" },
{ id: "gpt-5-mini", name: "GPT-5 Mini", alias: "gpt-5-mini" },
{ id: "claude-haiku-4.5", name: "Claude Haiku 4.5", alias: "claude-haiku-4.5" },
],
},
@@ -59,7 +60,9 @@ export const MITM_TOOLS = {
{ id: "claude-sonnet-4", name: "Claude Sonnet 4", alias: "claude-sonnet-4" },
{ id: "claude-haiku-4.5", name: "Claude Haiku 4.5", alias: "claude-haiku-4.5" },
{ id: "deepseek-3.2", name: "DeepSeek 3.2", alias: "deepseek-3.2" },
{ id: "minimax-m2.5", name: "MiniMax M2.5", alias: "minimax-m2.5" },
{ id: "minimax-m2.1", name: "MiniMax M2.1", alias: "minimax-m2.1" },
{ id: "glm-5", name: "GLM 5", alias: "glm-5" },
{ id: "simple-task", name: "Qwen3 Coder Next", alias: "simple-task" },
],
},