diff --git a/open-sse/config/providerModels.js b/open-sse/config/providerModels.js index e3691e74..8f515132 100644 --- a/open-sse/config/providerModels.js +++ b/open-sse/config/providerModels.js @@ -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); diff --git a/open-sse/handlers/chatCore.js b/open-sse/handlers/chatCore.js index 2a2abde1..f6d17432 100644 --- a/open-sse/handlers/chatCore.js +++ b/open-sse/handlers/chatCore.js @@ -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); diff --git a/src/mitm/handlers/base.js b/src/mitm/handlers/base.js index 74ac35bb..5f2d3d12 100644 --- a/src/mitm/handlers/base.js +++ b/src/mitm/handlers/base.js @@ -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 }; \ No newline at end of file diff --git a/src/mitm/handlers/kiro.js b/src/mitm/handlers/kiro.js index fa7b0162..b453d2cb 100644 --- a/src/mitm/handlers/kiro.js +++ b/src/mitm/handlers/kiro.js @@ -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 block + thinkBuf: "" // Buffer for partial thinking content + }; +} + +/** + * Extract thinking blocks from text content. + * Handles both ... and ... 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 or opening tags + const startRe = /|/i; + const startMatch = working.match(startRe); + + if (!startMatch) { + return { thinking: null, text: working }; + } + + const tag = startMatch[0].toLowerCase(); + const closeTag = tag === "" ? "" : ""; + 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 }; diff --git a/src/mitm/server.js b/src/mitm/server.js index caa3932e..2c5c876a 100644 --- a/src/mitm/server.js +++ b/src/mitm/server.js @@ -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 { diff --git a/src/shared/constants/cliTools.js b/src/shared/constants/cliTools.js index 8f0c65ed..e75f8ec1 100644 --- a/src/shared/constants/cliTools.js +++ b/src/shared/constants/cliTools.js @@ -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" }, ], },