mirror of
https://github.com/diegosouzapw/OmniRoute.git
synced 2026-07-26 09:52:11 +03:00
feat(inspector): add buffer, sseMerger, conversationNormalizer, llmMetadataExtractor (F4)
This commit is contained in:
diegosouzapw
201
src/mitm/inspector/buffer.ts
Normal file
201
src/mitm/inspector/buffer.ts
Normal file
@@ -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();
|
||||
393
src/mitm/inspector/conversationNormalizer.ts
Normal file
393
src/mitm/inspector/conversationNormalizer.ts
Normal file
@@ -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<string, unknown> | null {
|
||||
if (value && typeof value === "object" && !Array.isArray(value)) {
|
||||
return value as Record<string, unknown>;
|
||||
}
|
||||
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,
|
||||
};
|
||||
}
|
||||
182
src/mitm/inspector/llmMetadataExtractor.ts
Normal file
182
src/mitm/inspector/llmMetadataExtractor.ts
Normal file
@@ -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<string, unknown> | null {
|
||||
if (value && typeof value === "object" && !Array.isArray(value)) {
|
||||
return value as Record<string, unknown>;
|
||||
}
|
||||
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<string, unknown> | 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
|
||||
};
|
||||
}
|
||||
316
src/mitm/inspector/sseMerger.ts
Normal file
316
src/mitm/inspector/sseMerger.ts
Normal file
@@ -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<string, unknown> | null {
|
||||
if (value && typeof value === "object" && !Array.isArray(value)) {
|
||||
return value as Record<string, unknown>;
|
||||
}
|
||||
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<string, unknown> | null = null;
|
||||
const inputJsonByIndex: Record<number, string> = {};
|
||||
|
||||
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<string, unknown>)[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<number, OpenAiChoice> = {};
|
||||
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 };
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user