From 482cfbcdadc6138c05f742a37288e3caef436f85 Mon Sep 17 00:00:00 2001 From: diegosouzapw Date: Wed, 27 May 2026 21:44:37 -0300 Subject: [PATCH] feat(inspector): add buffer, sseMerger, conversationNormalizer, llmMetadataExtractor (F4) --- src/mitm/inspector/buffer.ts | 201 ++++++++++ src/mitm/inspector/conversationNormalizer.ts | 393 +++++++++++++++++++ src/mitm/inspector/llmMetadataExtractor.ts | 182 +++++++++ src/mitm/inspector/sseMerger.ts | 316 +++++++++++++++ 4 files changed, 1092 insertions(+) create mode 100644 src/mitm/inspector/buffer.ts create mode 100644 src/mitm/inspector/conversationNormalizer.ts create mode 100644 src/mitm/inspector/llmMetadataExtractor.ts create mode 100644 src/mitm/inspector/sseMerger.ts diff --git a/src/mitm/inspector/buffer.ts b/src/mitm/inspector/buffer.ts new file mode 100644 index 0000000000..955d8a8fe3 --- /dev/null +++ b/src/mitm/inspector/buffer.ts @@ -0,0 +1,201 @@ +/** + * In-memory ring buffer for intercepted traffic. + * + * Stores up to `INSPECTOR_BUFFER_SIZE` (default 1000) entries; rotates + * oldest-first when capacity is reached. Auto-applies kind detection and + * context-key fingerprinting on push, and broadcasts mutations to all + * subscribers (WebSocket consumers). + * + * Body sizes are clamped to `INSPECTOR_MAX_BODY_KB` (default 1024 KiB) and + * marked with a truncation suffix so the UI does not have to guess. + * + * See `_orchestration/master-plan-group-A.md` §3.6 and + * `12-traffic-inspector.plan.md` §4.1. + */ + +import { computeContextKey } from "./contextKey.ts"; +import { detectKind } from "./kindDetector.ts"; +import type { InterceptedRequest, ListFilters, WsEvent } from "./types.ts"; + +const TRUNCATION_MARKER = "\n…(truncated for performance)"; + +function parseEnvNumber(value: string | undefined, fallback: number): number { + if (!value) return fallback; + const parsed = Number(value); + if (!Number.isFinite(parsed) || parsed <= 0) return fallback; + return parsed; +} + +function getMaxBodyBytes(): number { + const kb = parseEnvNumber(process.env.INSPECTOR_MAX_BODY_KB, 1024); + return Math.max(1, Math.floor(kb)) * 1024; +} + +function capBody(body: string | null, maxBytes: number): string | null { + if (body == null) return body; + if (body.length <= maxBytes) return body; + return body.slice(0, maxBytes) + TRUNCATION_MARKER; +} + +function statusBucket(status: InterceptedRequest["status"]): string { + if (status === "error") return "error"; + if (status === "in-flight") return "in-flight"; + if (typeof status !== "number") return "unknown"; + if (status >= 200 && status < 300) return "2xx"; + if (status >= 300 && status < 400) return "3xx"; + if (status >= 400 && status < 500) return "4xx"; + if (status >= 500 && status < 600) return "5xx"; + return "unknown"; +} + +function matchesFilters(req: InterceptedRequest, filters?: ListFilters): boolean { + if (!filters) return true; + + if (filters.profile && filters.profile !== "all") { + if (filters.profile === "llm" && req.detectedKind !== "llm") return false; + if (filters.profile === "custom" && req.source !== "custom-host") return false; + } + + if (filters.host && req.host !== filters.host) return false; + if (filters.agent && req.agent !== filters.agent) return false; + if (filters.source && req.source !== filters.source) return false; + if (filters.sessionId && req.sessionId !== filters.sessionId) return false; + + if (filters.status) { + const bucket = statusBucket(req.status); + if (bucket !== filters.status) return false; + } + + return true; +} + +/** + * Ring buffer with broadcast support. + * + * Designed to be process-singleton (`globalTrafficBuffer`); tests can + * instantiate isolated buffers when needed. + */ +export class TrafficBuffer { + private buffer: InterceptedRequest[] = []; + private subscribers = new Set<(ev: WsEvent) => void>(); + private maxSize: number; + private maxBodyBytes: number; + + constructor( + maxSize: number = parseEnvNumber(process.env.INSPECTOR_BUFFER_SIZE, 1000), + maxBodyBytes: number = getMaxBodyBytes() + ) { + this.maxSize = Math.max(1, Math.floor(maxSize)); + this.maxBodyBytes = Math.max(1, Math.floor(maxBodyBytes)); + } + + /** + * Append a new intercepted request. Applies kind detection and + * context-key fingerprinting if missing, and clamps body sizes. + * Broadcasts a `new` event to all subscribers. + */ + push(req: InterceptedRequest): void { + if (!req.detectedKind) { + req.detectedKind = detectKind(req); + } + if (!req.contextKey && req.detectedKind === "llm") { + const key = computeContextKey(req); + if (key) req.contextKey = key; + } + + req.requestBody = capBody(req.requestBody, this.maxBodyBytes); + req.responseBody = capBody(req.responseBody, this.maxBodyBytes); + + this.buffer.push(req); + while (this.buffer.length > this.maxSize) { + this.buffer.shift(); + } + + this.broadcast({ type: "new", data: req }); + } + + /** + * Update an existing entry in place by id. No-op if the id is unknown + * (e.g. already rotated out). Broadcasts an `update` event on success. + */ + update(id: string, req: InterceptedRequest): void { + const idx = this.buffer.findIndex((r) => r.id === id); + if (idx < 0) return; + + req.requestBody = capBody(req.requestBody, this.maxBodyBytes); + req.responseBody = capBody(req.responseBody, this.maxBodyBytes); + + this.buffer[idx] = req; + this.broadcast({ type: "update", data: req }); + } + + /** + * Lookup by id (linear scan — buffer is bounded to ~1000 entries). + */ + get(id: string): InterceptedRequest | null { + return this.buffer.find((r) => r.id === id) ?? null; + } + + /** + * Return a filtered snapshot of the buffer. Filtering is in-memory and + * cheap (~O(maxSize)). A new array is returned each call. + */ + list(filters?: ListFilters): InterceptedRequest[] { + if (!filters) return [...this.buffer]; + return this.buffer.filter((r) => matchesFilters(r, filters)); + } + + /** + * Empty the buffer and notify subscribers. Subscriber count is preserved. + */ + clear(): void { + this.buffer = []; + this.broadcast({ type: "clear" }); + } + + /** + * Register a listener. Immediately receives a `snapshot` event with the + * current buffer state. Returns an `unsubscribe` function. + */ + subscribe(fn: (ev: WsEvent) => void): () => void { + this.subscribers.add(fn); + try { + fn({ type: "snapshot", data: [...this.buffer] }); + } catch { + // a subscriber's snapshot handler failure must not break subscription + } + return () => { + this.subscribers.delete(fn); + }; + } + + /** + * Current subscriber count — exposed for tests / diagnostics. + */ + subscriberCount(): number { + return this.subscribers.size; + } + + /** + * Current entry count — exposed for tests / diagnostics. + */ + size(): number { + return this.buffer.length; + } + + private broadcast(ev: WsEvent): void { + for (const fn of this.subscribers) { + try { + fn(ev); + } catch { + // one subscriber's failure must not block others + } + } + } +} + +/** + * Process-wide singleton consumed by `agentBridgeHook`, `httpProxyServer`, + * REST/WS routes, and tests. + */ +export const globalTrafficBuffer = new TrafficBuffer(); diff --git a/src/mitm/inspector/conversationNormalizer.ts b/src/mitm/inspector/conversationNormalizer.ts new file mode 100644 index 0000000000..b38e9acd2b --- /dev/null +++ b/src/mitm/inspector/conversationNormalizer.ts @@ -0,0 +1,393 @@ +/** + * Conversation normalizer — converts OpenAI / Anthropic / Gemini request + + * response payloads into a single provider-agnostic shape. + * + * MIT — port from https://github.com/chouzz/llm-interceptor (ui/utils.ts) + * + * Returns `null` for non-LLM requests or payloads we cannot understand — + * never throws — so the renderer can fall back to the raw view. + */ + +import { mergeStream, parseSseStream } from "./sseMerger.ts"; +import type { + InterceptedRequest, + NormalizedBlock, + NormalizedConversation, + NormalizedTurn, +} from "./types.ts"; + +type NormalizedRole = NormalizedTurn["role"]; + +function asRecord(value: unknown): Record | null { + if (value && typeof value === "object" && !Array.isArray(value)) { + return value as Record; + } + return null; +} + +function tryParseJson(value: string | null | undefined): unknown { + if (!value) return null; + try { + return JSON.parse(value); + } catch { + return null; + } +} + +function normalizeRole(raw: unknown): NormalizedRole { + if (raw === "system" || raw === "user" || raw === "assistant" || raw === "tool") { + return raw; + } + if (raw === "model") return "assistant"; + if (raw === "function") return "tool"; + return "user"; +} + +/** + * OpenAI / Anthropic message content can be a string, or an array of blocks. + * Returns a list of normalized blocks. + */ +function blocksFromOpenAiContent(content: unknown): NormalizedBlock[] { + if (content == null) return []; + if (typeof content === "string") { + if (content.length === 0) return []; + return [{ type: "text", text: content }]; + } + if (!Array.isArray(content)) return []; + const out: NormalizedBlock[] = []; + for (const raw of content) { + if (typeof raw === "string") { + out.push({ type: "text", text: raw }); + continue; + } + const block = asRecord(raw); + if (!block) continue; + const type = block.type; + if (type === "text" || type === "output_text") { + const text = typeof block.text === "string" ? block.text : ""; + out.push({ type: "text", text }); + } else if (type === "input_text") { + const text = typeof block.text === "string" ? block.text : ""; + out.push({ type: "text", text }); + } else if (type === "tool_use") { + out.push({ + type: "tool_use", + id: typeof block.id === "string" ? block.id : "", + name: typeof block.name === "string" ? block.name : "", + input: block.input ?? {}, + }); + } else if (type === "tool_result") { + out.push({ + type: "tool_result", + tool_use_id: + typeof block.tool_use_id === "string" ? block.tool_use_id : "", + content: block.content ?? null, + }); + } else if (typeof block.text === "string") { + out.push({ type: "text", text: block.text }); + } + } + return out; +} + +/** + * OpenAI assistant messages may declare `tool_calls`. Each becomes a + * `tool_use` block alongside any text content. + */ +function appendOpenAiToolCalls( + blocks: NormalizedBlock[], + toolCalls: unknown +): NormalizedBlock[] { + if (!Array.isArray(toolCalls)) return blocks; + for (const raw of toolCalls) { + const tc = asRecord(raw); + if (!tc) continue; + const fn = asRecord(tc.function) ?? {}; + let parsedInput: unknown = {}; + if (typeof fn.arguments === "string") { + try { + parsedInput = JSON.parse(fn.arguments); + } catch { + parsedInput = fn.arguments; + } + } else if (fn.arguments != null) { + parsedInput = fn.arguments; + } + blocks.push({ + type: "tool_use", + id: typeof tc.id === "string" ? tc.id : "", + name: typeof fn.name === "string" ? fn.name : "", + input: parsedInput, + }); + } + return blocks; +} + +/** + * Build NormalizedTurn[] from OpenAI / Anthropic chat messages. + */ +function turnsFromOpenAiMessages(messages: unknown[]): NormalizedTurn[] { + const out: NormalizedTurn[] = []; + for (const raw of messages) { + const msg = asRecord(raw); + if (!msg) continue; + const role = normalizeRole(msg.role); + + if (msg.role === "tool" || msg.role === "function") { + const content = msg.content; + out.push({ + role: "tool", + blocks: [ + { + type: "tool_result", + tool_use_id: + typeof msg.tool_call_id === "string" + ? msg.tool_call_id + : typeof msg.name === "string" + ? msg.name + : "", + content, + }, + ], + }); + continue; + } + + const blocks = blocksFromOpenAiContent(msg.content); + if ("tool_calls" in msg) { + appendOpenAiToolCalls(blocks, msg.tool_calls); + } + if (blocks.length === 0 && msg.content == null && !("tool_calls" in msg)) { + continue; + } + out.push({ role, blocks }); + } + return out; +} + +/** + * Gemini contents have a different shape: `[{role, parts: [{text|...}]}]`. + */ +function turnsFromGeminiContents(contents: unknown[]): NormalizedTurn[] { + const out: NormalizedTurn[] = []; + for (const raw of contents) { + const turn = asRecord(raw); + if (!turn) continue; + const role = normalizeRole(turn.role); + const blocks: NormalizedBlock[] = []; + if (Array.isArray(turn.parts)) { + for (const partRaw of turn.parts) { + const part = asRecord(partRaw); + if (!part) continue; + if (typeof part.text === "string") { + blocks.push({ type: "text", text: part.text }); + } else if (part.functionCall) { + const fc = asRecord(part.functionCall) ?? {}; + blocks.push({ + type: "tool_use", + id: typeof fc.name === "string" ? fc.name : "", + name: typeof fc.name === "string" ? fc.name : "", + input: fc.args ?? {}, + }); + } else if (part.functionResponse) { + const fr = asRecord(part.functionResponse) ?? {}; + blocks.push({ + type: "tool_result", + tool_use_id: typeof fr.name === "string" ? fr.name : "", + content: fr.response ?? null, + }); + } + } + } + if (blocks.length > 0) out.push({ role, blocks }); + } + return out; +} + +/** + * Anthropic Messages API requests carry a top-level `system` field (string + * or array of `{type:"text"|text}` blocks). Convert to a `system` turn. + */ +function systemTurnFromAnthropic(system: unknown): NormalizedTurn | null { + if (!system) return null; + if (typeof system === "string") { + return system.length === 0 + ? null + : { role: "system", blocks: [{ type: "text", text: system }] }; + } + if (!Array.isArray(system)) return null; + const blocks: NormalizedBlock[] = []; + for (const raw of system) { + const item = asRecord(raw); + if (item && typeof item.text === "string") { + blocks.push({ type: "text", text: item.text }); + } else if (typeof raw === "string") { + blocks.push({ type: "text", text: raw }); + } + } + if (blocks.length === 0) return null; + return { role: "system", blocks }; +} + +function buildRequestTurns(body: unknown): NormalizedTurn[] | null { + const obj = asRecord(body); + if (!obj) return null; + + if (Array.isArray(obj.messages)) { + const turns: NormalizedTurn[] = []; + const systemTurn = systemTurnFromAnthropic(obj.system); + if (systemTurn) turns.push(systemTurn); + turns.push(...turnsFromOpenAiMessages(obj.messages)); + return turns; + } + + if (Array.isArray(obj.contents)) { + const turns: NormalizedTurn[] = []; + const sysObj = asRecord(obj.systemInstruction); + if (sysObj && Array.isArray(sysObj.parts)) { + const parts: NormalizedBlock[] = []; + for (const partRaw of sysObj.parts) { + const p = asRecord(partRaw); + if (p && typeof p.text === "string") parts.push({ type: "text", text: p.text }); + } + if (parts.length > 0) turns.push({ role: "system", blocks: parts }); + } + turns.push(...turnsFromGeminiContents(obj.contents)); + return turns; + } + + if (typeof obj.prompt === "string") { + return [{ role: "user", blocks: [{ type: "text", text: obj.prompt }] }]; + } + if (typeof obj.input === "string") { + return [{ role: "user", blocks: [{ type: "text", text: obj.input }] }]; + } + if (Array.isArray(obj.input)) { + return turnsFromOpenAiMessages(obj.input); + } + + return null; +} + +function isSseResponse(req: InterceptedRequest): boolean { + const accept = req.requestHeaders["accept"] ?? req.requestHeaders["Accept"] ?? ""; + const ct = req.responseHeaders["content-type"] ?? req.responseHeaders["Content-Type"] ?? ""; + return ( + accept.includes("event-stream") || + ct.includes("event-stream") || + /^\s*event:|^\s*data:/m.test(req.responseBody ?? "") + ); +} + +function extractAnthropicResponseTurn(message: unknown): NormalizedTurn | null { + const obj = asRecord(message); + if (!obj) return null; + const content = obj.content; + if (!Array.isArray(content)) return null; + const blocks: NormalizedBlock[] = []; + for (const raw of content) { + const block = asRecord(raw); + if (!block) continue; + if (block.type === "text" && typeof block.text === "string") { + blocks.push({ type: "text", text: block.text }); + } else if (block.type === "tool_use") { + blocks.push({ + type: "tool_use", + id: typeof block.id === "string" ? block.id : "", + name: typeof block.name === "string" ? block.name : "", + input: block.input ?? {}, + }); + } else if (block.type === "thinking" && typeof block.thinking === "string") { + blocks.push({ type: "text", text: block.thinking }); + } + } + if (blocks.length === 0) return null; + return { role: "assistant", blocks }; +} + +function extractOpenAiResponseTurn(message: unknown): NormalizedTurn | null { + const obj = asRecord(message); + if (!obj || !Array.isArray(obj.choices)) return null; + const first = asRecord(obj.choices[0]); + if (!first) return null; + const msg = asRecord(first.message) ?? asRecord(first.delta); + if (!msg) return null; + const blocks = blocksFromOpenAiContent(msg.content); + if ("tool_calls" in msg) appendOpenAiToolCalls(blocks, msg.tool_calls); + if (blocks.length === 0) return null; + return { role: "assistant", blocks }; +} + +function extractGeminiResponseTurn(message: unknown): NormalizedTurn | null { + const obj = asRecord(message); + if (!obj || !Array.isArray(obj.candidates)) return null; + const first = asRecord(obj.candidates[0]); + if (!first) return null; + const content = asRecord(first.content); + if (!content || !Array.isArray(content.parts)) return null; + const blocks: NormalizedBlock[] = []; + for (const partRaw of content.parts) { + const part = asRecord(partRaw); + if (!part) continue; + if (typeof part.text === "string") { + blocks.push({ type: "text", text: part.text }); + } else if (part.functionCall) { + const fc = asRecord(part.functionCall) ?? {}; + blocks.push({ + type: "tool_use", + id: typeof fc.name === "string" ? fc.name : "", + name: typeof fc.name === "string" ? fc.name : "", + input: fc.args ?? {}, + }); + } + } + if (blocks.length === 0) return null; + return { role: "assistant", blocks }; +} + +function buildResponseTurns(req: InterceptedRequest): NormalizedTurn[] { + const raw = req.responseBody ?? ""; + if (!raw) return []; + + let payload: unknown = null; + + if (isSseResponse(req)) { + const merged = mergeStream(parseSseStream(raw)); + payload = merged.message ?? null; + } else { + payload = tryParseJson(raw); + } + + if (!payload) return []; + + const anth = extractAnthropicResponseTurn(payload); + if (anth) return [anth]; + const oai = extractOpenAiResponseTurn(payload); + if (oai) return [oai]; + const gem = extractGeminiResponseTurn(payload); + if (gem) return [gem]; + + return []; +} + +/** + * Normalize an intercepted LLM request + response into a provider-agnostic + * conversation. Returns `null` for non-LLM requests or unparseable payloads. + */ +export function normalizeConversation( + req: InterceptedRequest +): NormalizedConversation | null { + if (req.detectedKind !== "llm") return null; + + const requestBody = tryParseJson(req.requestBody); + const requestTurns = buildRequestTurns(requestBody); + if (!requestTurns) return null; + + const responseTurns = buildResponseTurns(req); + + return { + request: requestTurns, + response: responseTurns, + contextKey: req.contextKey ?? null, + }; +} diff --git a/src/mitm/inspector/llmMetadataExtractor.ts b/src/mitm/inspector/llmMetadataExtractor.ts new file mode 100644 index 0000000000..97209513db --- /dev/null +++ b/src/mitm/inspector/llmMetadataExtractor.ts @@ -0,0 +1,182 @@ +/** + * Extract LLM-specific metadata from intercepted requests so the UI can + * render summary chips (provider, model, tokens, cost). Provider/api inference + * is host- and path-based; token counts come from the upstream `usage` block. + * + * Replaces the stub `extractLlmMetadata` left in `kindDetector.ts` by F1. + * Cost estimation is deferred to a future audit pass; we return `null`. + */ + +import { detectKind } from "./kindDetector.ts"; +import { mergeStream, parseSseStream } from "./sseMerger.ts"; +import type { InterceptedRequest, LlmMetadata } from "./types.ts"; + +interface ProviderMatch { + pattern: RegExp; + provider: string; +} + +const PROVIDER_MATCHERS: ProviderMatch[] = [ + { pattern: /(^|\.)openai\.com$/i, provider: "openai" }, + { pattern: /(^|\.)openai\.azure\.com$/i, provider: "azure-openai" }, + { pattern: /(^|\.)anthropic\.com$/i, provider: "anthropic" }, + { pattern: /generativelanguage\.googleapis\.com$/i, provider: "gemini" }, + { pattern: /(^|\.)aiplatform\.googleapis\.com$/i, provider: "vertex" }, + { pattern: /(^|\.)mistral\.ai$/i, provider: "mistral" }, + { pattern: /(^|\.)deepseek\.com$/i, provider: "deepseek" }, + { pattern: /(^|\.)groq\.com$/i, provider: "groq" }, + { pattern: /(^|\.)together\.xyz$/i, provider: "together" }, + { pattern: /(^|\.)fireworks\.ai$/i, provider: "fireworks" }, + { pattern: /(^|\.)cohere\.com$/i, provider: "cohere" }, + { pattern: /(^|\.)perplexity\.ai$/i, provider: "perplexity" }, + { pattern: /(^|\.)huggingface\.co$/i, provider: "huggingface" }, + { pattern: /(^|\.)openrouter\.ai$/i, provider: "openrouter" }, + { pattern: /(^|\.)x\.ai$/i, provider: "xai" }, + { pattern: /(^|\.)moonshot\.ai$/i, provider: "moonshot" }, + { pattern: /bigmodel\.cn$/i, provider: "bigmodel" }, + { pattern: /(^|\.)githubcopilot\.com$/i, provider: "github-copilot" }, + { pattern: /(^|\.)cursor\.sh$/i, provider: "cursor" }, + { pattern: /(^|\.)zed\.dev$/i, provider: "zed" }, +]; + +interface ApiKindMatch { + pattern: RegExp; + apiKind: string; +} + +const API_KIND_MATCHERS: ApiKindMatch[] = [ + { pattern: /\/(v1|v1beta)?\/?chat\/completions/i, apiKind: "chat.completions" }, + { pattern: /\/(v1|v1beta)\/messages/i, apiKind: "messages" }, + { pattern: /\/(v1|v1beta)?\/?embeddings/i, apiKind: "embeddings" }, + { pattern: /\/(v1|v1beta)?\/?responses/i, apiKind: "responses" }, + { pattern: /\/streamGenerateContent/i, apiKind: "streamGenerateContent" }, + { pattern: /\/generateContent/i, apiKind: "generateContent" }, + { pattern: /\/(v1|v1beta)\/completions/i, apiKind: "completions" }, +]; + +function asRecord(value: unknown): Record | null { + if (value && typeof value === "object" && !Array.isArray(value)) { + return value as Record; + } + return null; +} + +function safeParseJson(value: string | null | undefined): unknown { + if (!value) return null; + try { + return JSON.parse(value); + } catch { + return null; + } +} + +function inferProvider(host: string): string | null { + for (const m of PROVIDER_MATCHERS) { + if (m.pattern.test(host)) return m.provider; + } + return null; +} + +function inferApiKind(path: string): string | null { + for (const m of API_KIND_MATCHERS) { + if (m.pattern.test(path)) return m.apiKind; + } + return null; +} + +function countMessages(body: Record | null): number { + if (!body) return 0; + if (Array.isArray(body.messages)) return body.messages.length; + if (Array.isArray(body.contents)) return body.contents.length; + if (Array.isArray(body.input)) return body.input.length; + return 0; +} + +function isSseRequest(req: InterceptedRequest): boolean { + const accept = req.requestHeaders["accept"] ?? req.requestHeaders["Accept"] ?? ""; + const ct = req.responseHeaders["content-type"] ?? req.responseHeaders["Content-Type"] ?? ""; + return ( + accept.includes("event-stream") || + ct.includes("event-stream") || + /^\s*event:|^\s*data:/m.test(req.responseBody ?? "") + ); +} + +function maybeNumber(v: unknown): number | null { + return typeof v === "number" && Number.isFinite(v) ? v : null; +} + +function extractUsage(resp: unknown): { tokensIn: number | null; tokensOut: number | null } { + // Direct JSON + const respObj = asRecord(resp); + const usage = asRecord(respObj?.usage); + if (usage) { + const inTok = + maybeNumber(usage.prompt_tokens) ?? + maybeNumber(usage.input_tokens) ?? + maybeNumber((asRecord(usage.promptTokensDetails) ?? {}).total) ?? + null; + const outTok = + maybeNumber(usage.completion_tokens) ?? + maybeNumber(usage.output_tokens) ?? + maybeNumber((asRecord(usage.completionTokensDetails) ?? {}).total) ?? + null; + return { tokensIn: inTok, tokensOut: outTok }; + } + // Gemini-style usageMetadata + const um = asRecord(respObj?.usageMetadata); + if (um) { + const inTok = maybeNumber(um.promptTokenCount); + const outTok = maybeNumber(um.candidatesTokenCount); + return { tokensIn: inTok, tokensOut: outTok }; + } + return { tokensIn: null, tokensOut: null }; +} + +/** + * Extract LLM metadata. Returns `null` for non-LLM requests; otherwise + * returns best-effort fields (any unknown field is `null`). + */ +export function extractLlmMetadata(req: InterceptedRequest): LlmMetadata | null { + const kind = req.detectedKind ?? detectKind(req); + if (kind !== "llm") return null; + + const body = asRecord(safeParseJson(req.requestBody)); + let resp: unknown = safeParseJson(req.responseBody); + + // If response was SSE, try to merge it for usage metadata. + if (!resp && req.responseBody && isSseRequest(req)) { + const merged = mergeStream(parseSseStream(req.responseBody)); + resp = merged.message ?? null; + } + + const respObj = asRecord(resp); + + const provider = inferProvider(req.host); + const apiKind = inferApiKind(req.path); + const model = + (body && typeof body.model === "string" ? body.model : null) ?? + (respObj && typeof respObj.model === "string" ? respObj.model : null) ?? + (respObj && typeof respObj.modelVersion === "string" ? respObj.modelVersion : null) ?? + null; + const messages = countMessages(body); + const { tokensIn, tokensOut } = extractUsage(resp); + const streamed = isSseRequest(req); + const mappedTo = + req.mappedModel ?? + req.requestHeaders["x-omniroute-mapped"] ?? + req.requestHeaders["X-Omniroute-Mapped"] ?? + null; + + return { + provider, + apiKind, + model, + messages, + tokensIn, + tokensOut, + streamed, + mappedTo, + costEstimateUsd: null, // cost table out of scope for F4 + }; +} diff --git a/src/mitm/inspector/sseMerger.ts b/src/mitm/inspector/sseMerger.ts new file mode 100644 index 0000000000..8a1d245603 --- /dev/null +++ b/src/mitm/inspector/sseMerger.ts @@ -0,0 +1,316 @@ +/** + * SSE merger — reconstructs complete LLM response from streaming SSE chunks. + * + * MIT — port from https://github.com/chouzz/llm-interceptor (merger.py) + * + * Detects API format by chunk shape (not URL — robust to URL rewrite) and + * rebuilds Anthropic / OpenAI / Gemini responses. Falls back to a raw event + * list when the format is unrecognised so the caller never crashes. + */ + +export type ApiFormat = "anthropic" | "openai" | "gemini" | "unknown"; + +export interface SseEvent { + event?: string; + data?: string; + // Parsed JSON payload when `data` was valid JSON. + json?: unknown; +} + +export interface MergedResponse { + format: ApiFormat; + message?: unknown; + raw?: SseEvent[]; +} + +function asRecord(value: unknown): Record | null { + if (value && typeof value === "object" && !Array.isArray(value)) { + return value as Record; + } + return null; +} + +/** + * Inspect chunk shapes to determine the upstream API. Matches on the first + * recognisable hint; returns `"unknown"` if none match. + */ +export function detectApiFormat(chunks: SseEvent[]): ApiFormat { + for (const c of chunks) { + const j = asRecord(c.json); + if (!j) continue; + if (j.type === "message_start" || j.type === "content_block_delta") return "anthropic"; + if (Array.isArray(j.choices)) { + const first = j.choices[0]; + if (first && typeof first === "object" && "delta" in first) return "openai"; + } + if (Array.isArray(j.candidates)) return "gemini"; + } + return "unknown"; +} + +/** + * Parse a raw SSE stream (the response body string captured by the proxy) + * into discrete events. Empty blocks and `[DONE]` terminators are skipped + * silently; malformed JSON payloads are kept as raw `data` (no `json`). + */ +export function parseSseStream(raw: string): SseEvent[] { + const events: SseEvent[] = []; + if (!raw) return events; + // SSE blocks separated by blank lines — accept both LF and CRLF. + for (const block of raw.split(/\r?\n\r?\n/)) { + if (!block.trim()) continue; + const ev: SseEvent = {}; + for (const line of block.split(/\r?\n/)) { + if (line.startsWith("event:")) { + ev.event = line.slice(6).trim(); + } else if (line.startsWith("data:")) { + ev.data = (ev.data ?? "") + line.slice(5).trim(); + } + } + if (ev.data === undefined) continue; + if (ev.data === "[DONE]") { + events.push(ev); + continue; + } + try { + ev.json = JSON.parse(ev.data); + } catch { + // keep raw data only + } + events.push(ev); + } + return events; +} + +interface AnthropicBlock { + type: string; + text?: string; + thinking?: string; + id?: string; + name?: string; + input?: unknown; +} + +/** + * Rebuild an Anthropic Messages API response from streaming events. + * Handles `text_delta`, `thinking_delta`, and `input_json_delta` deltas; + * applies `JSON.parse` (best-effort) on accumulated tool-use input. + */ +export function rebuildAnthropic(chunks: SseEvent[]): MergedResponse { + const blocks: AnthropicBlock[] = []; + let message: Record | null = null; + const inputJsonByIndex: Record = {}; + + for (const c of chunks) { + const j = asRecord(c.json); + if (!j) continue; + const t = j.type; + + if (t === "message_start") { + const m = asRecord(j.message); + message = m ? { ...m } : {}; + } else if (t === "content_block_start") { + const idx = typeof j.index === "number" ? j.index : blocks.length; + const cb = asRecord(j.content_block); + const block: AnthropicBlock = { type: "text" }; + if (cb) { + for (const [k, v] of Object.entries(cb)) (block as Record)[k] = v; + } + if (block.type === "text" && block.text === undefined) block.text = ""; + if (block.type === "thinking" && block.thinking === undefined) block.thinking = ""; + if (block.type === "tool_use" && block.input === undefined) block.input = {}; + blocks[idx] = block; + } else if (t === "content_block_delta") { + const idx = typeof j.index === "number" ? j.index : 0; + const d = asRecord(j.delta); + if (!d) continue; + // Ensure a block slot exists (some streams skip content_block_start). + const slot = blocks[idx] ?? (blocks[idx] = { type: "text", text: "" }); + const dType = d.type; + if (dType === "text_delta" && typeof d.text === "string") { + slot.text = (slot.text ?? "") + d.text; + } else if (dType === "thinking_delta" && typeof d.thinking === "string") { + slot.thinking = (slot.thinking ?? "") + d.thinking; + } else if (dType === "input_json_delta" && typeof d.partial_json === "string") { + inputJsonByIndex[idx] = (inputJsonByIndex[idx] ?? "") + d.partial_json; + } + } else if (t === "content_block_stop") { + const idx = typeof j.index === "number" ? j.index : 0; + const slot = blocks[idx]; + if (slot && slot.type === "tool_use" && inputJsonByIndex[idx]) { + try { + slot.input = JSON.parse(inputJsonByIndex[idx]); + } catch { + // keep accumulated string for forensic visibility + slot.input = inputJsonByIndex[idx]; + } + } + } else if (t === "message_delta") { + if (!message) message = {}; + const d = asRecord(j.delta); + if (d && typeof d.stop_reason === "string") { + message.stop_reason = d.stop_reason; + } + const usage = asRecord(j.usage); + if (usage) { + const prev = asRecord(message.usage) ?? {}; + message.usage = { ...prev, ...usage }; + } + } + } + + const filledBlocks = blocks.filter((b) => b !== undefined); + return { + format: "anthropic", + message: { ...(message ?? {}), content: filledBlocks }, + }; +} + +interface OpenAiToolCall { + index: number; + id?: string; + type?: string; + function: { name: string; arguments: string }; +} + +interface OpenAiChoice { + index: number; + message: { + role: string; + content: string; + tool_calls?: OpenAiToolCall[]; + refusal?: string; + }; + finish_reason: string | null; +} + +/** + * Rebuild an OpenAI Chat Completions response from streaming events. + * Accumulates content text and tool-call fragments per choice/index. + */ +export function rebuildOpenAI(chunks: SseEvent[]): MergedResponse { + const choicesByIdx: Record = {}; + let model: string | null = null; + let usage: unknown = null; + let id: string | null = null; + + for (const c of chunks) { + const j = asRecord(c.json); + if (!j) continue; + if (typeof j.model === "string") model = j.model; + if (typeof j.id === "string") id = j.id; + if (j.usage != null) usage = j.usage; + if (!Array.isArray(j.choices)) continue; + + for (const raw of j.choices) { + const ch = asRecord(raw); + if (!ch) continue; + const idx = typeof ch.index === "number" ? ch.index : 0; + const slot = (choicesByIdx[idx] ??= { + index: idx, + message: { role: "assistant", content: "" }, + finish_reason: null, + }); + const delta = asRecord(ch.delta) ?? {}; + if (typeof delta.role === "string") slot.message.role = delta.role; + if (typeof delta.content === "string") slot.message.content += delta.content; + if (typeof delta.refusal === "string") { + slot.message.refusal = (slot.message.refusal ?? "") + delta.refusal; + } + if (Array.isArray(delta.tool_calls)) { + slot.message.tool_calls ??= []; + for (const tcRaw of delta.tool_calls) { + const tc = asRecord(tcRaw); + if (!tc) continue; + const ti = typeof tc.index === "number" ? tc.index : 0; + const tcSlot = + slot.message.tool_calls[ti] ?? + (slot.message.tool_calls[ti] = { + index: ti, + function: { name: "", arguments: "" }, + }); + if (typeof tc.id === "string") tcSlot.id = tc.id; + if (typeof tc.type === "string") tcSlot.type = tc.type; + const fn = asRecord(tc.function); + if (fn) { + if (typeof fn.name === "string") tcSlot.function.name += fn.name; + if (typeof fn.arguments === "string") tcSlot.function.arguments += fn.arguments; + } + } + } + if (typeof ch.finish_reason === "string") slot.finish_reason = ch.finish_reason; + } + } + + return { + format: "openai", + message: { + id, + model, + choices: Object.values(choicesByIdx).sort((a, b) => a.index - b.index), + usage, + }, + }; +} + +/** + * Rebuild a Gemini `generateContent`-style response from streaming events. + * Concatenates all parts across emitted candidates into a single candidate. + */ +export function rebuildGemini(chunks: SseEvent[]): MergedResponse { + const parts: unknown[] = []; + let usageMetadata: unknown = null; + let finishReason: unknown = null; + let modelVersion: string | null = null; + + for (const c of chunks) { + const j = asRecord(c.json); + if (!j) continue; + if (j.usageMetadata != null) usageMetadata = j.usageMetadata; + if (typeof j.modelVersion === "string") modelVersion = j.modelVersion; + if (!Array.isArray(j.candidates)) continue; + for (const candRaw of j.candidates) { + const cand = asRecord(candRaw); + if (!cand) continue; + if (cand.finishReason != null) finishReason = cand.finishReason; + const content = asRecord(cand.content); + if (!content) continue; + const ps = content.parts; + if (Array.isArray(ps)) { + for (const p of ps) parts.push(p); + } + } + } + + return { + format: "gemini", + message: { + candidates: [ + { + content: { parts, role: "model" }, + ...(finishReason != null ? { finishReason } : {}), + }, + ], + ...(modelVersion ? { modelVersion } : {}), + ...(usageMetadata != null ? { usageMetadata } : {}), + }, + }; +} + +/** + * Merge an array of SSE events into a single rebuilt response. Returns + * `{ format: "unknown", raw }` (no throw) for unrecognised shapes. + */ +export function mergeStream(chunks: SseEvent[]): MergedResponse { + const format = detectApiFormat(chunks); + switch (format) { + case "anthropic": + return rebuildAnthropic(chunks); + case "openai": + return rebuildOpenAI(chunks); + case "gemini": + return rebuildGemini(chunks); + default: + return { format: "unknown", raw: chunks }; + } +}