diff --git a/open-sse/handlers/chatCore.js b/open-sse/handlers/chatCore.js index eb1e88b5..190cbb44 100644 --- a/open-sse/handlers/chatCore.js +++ b/open-sse/handlers/chatCore.js @@ -22,7 +22,7 @@ import { dedupeTools } from "../utils/toolDeduper.js"; import { injectCaveman } from "../rtk/caveman.js"; import { injectPonytail } from "../rtk/ponytail.js"; import { compressMessages, formatRtkLog } from "../rtk/index.js"; -import { compressWithHeadroom, formatHeadroomLog } from "../rtk/headroom.js"; +import { compressWithHeadroom, formatHeadroomLog, formatHeadroomSizeLog, isHeadroomPhantomSavings } from "../rtk/headroom.js"; import { getCapabilitiesForModel } from "../providers/capabilities.js"; import { stripUnsupportedModalities } from "../translator/concerns/modality.js"; import { prefetchRemoteImages } from "../translator/concerns/prefetch.js"; @@ -162,9 +162,16 @@ export async function handleChatCore({ body, modelInfo, credentials, log, onCred if (rtkLine) console.log(rtkLine); // Headroom: optional external proxy compression; fail open if proxy is absent. - const headroomStats = await compressWithHeadroom(translatedBody, { enabled: headroomEnabled, url: headroomUrl, model: upstreamModel, format: finalFormat, compressUserMessages: headroomCompressUserMessages }); + const headroomDiagnostics = {}; + const headroomStats = await compressWithHeadroom(translatedBody, { enabled: headroomEnabled, url: headroomUrl, model: upstreamModel, format: finalFormat, compressUserMessages: headroomCompressUserMessages, diagnostics: headroomDiagnostics }); const headroomLine = formatHeadroomLog(headroomStats); - if (headroomLine) log?.info?.("HEADROOM", headroomLine); + const headroomSizeLine = formatHeadroomSizeLog(headroomDiagnostics); + if (headroomLine) { + log?.info?.("HEADROOM", `${headroomLine}${headroomSizeLine ? ` | ${headroomSizeLine}` : ""}`); + if (isHeadroomPhantomSavings(headroomStats, headroomDiagnostics)) { + log?.warn?.("HEADROOM", `reported token delta, but outbound JSON shrank <5%; provider may bill near-original payload | ${headroomSizeLine}`); + } + } else if (headroomEnabled) log?.warn?.("HEADROOM", `skipped: ${headroomDiagnostics.reason || "compression unavailable"}${headroomDiagnostics.endpoint ? ` (${headroomDiagnostics.endpoint})` : ""}`); // Caveman: inject terse-style system prompt if (cavemanEnabled && cavemanLevel) { diff --git a/open-sse/rtk/headroom.js b/open-sse/rtk/headroom.js index ac2f4db5..df81ee42 100644 --- a/open-sse/rtk/headroom.js +++ b/open-sse/rtk/headroom.js @@ -3,39 +3,135 @@ import { openaiToClaudeRequest } from "../translator/request/openai-to-claude.js const DEFAULT_TIMEOUT_MS = 3000; +function jsonBytes(value) { + try { + return new TextEncoder().encode(JSON.stringify(value) || "").length; + } catch { + return 0; + } +} + +function messagePayload(body) { + if (Array.isArray(body?.messages)) return body.messages; + if (Array.isArray(body?.input)) return body.input; + return null; +} + +function captureSizeSnapshot(body) { + const messages = messagePayload(body); + return { + bodyBytes: jsonBytes(body), + messageBytes: messages ? jsonBytes(messages) : 0, + }; +} + +function setDiagnostic(diagnostics, reason) { + if (diagnostics && !diagnostics.reason) diagnostics.reason = reason; +} + +function scrubSensitiveUrlText(text) { + return String(text) + .replace(/\/\/[^/@\s]+@/g, "//") + .replace(/(https?:\/\/[^\s?#]+)[?#][^\s)]*/g, "$1"); +} + +function describeFetchError(error) { + const cause = error?.cause; + const code = cause?.code || error?.code; + const message = scrubSensitiveUrlText(cause?.message || error?.message || String(error)); + return code ? `${code}: ${message}` : message; +} + +function buildCompressEndpoint(url) { + try { + const parsed = new URL(url); + parsed.pathname = `${parsed.pathname.replace(/\/$/, "")}/v1/compress`; + parsed.hash = ""; + return parsed.toString(); + } catch { + const raw = String(url).replace(/#.*$/, ""); + const [base, query = ""] = raw.split("?", 2); + const endpoint = `${base.replace(/\/$/, "")}/v1/compress`; + return query ? `${endpoint}?${query}` : endpoint; + } +} + +function maskEndpoint(endpoint) { + try { + const parsed = new URL(endpoint); + parsed.username = ""; + parsed.password = ""; + parsed.search = ""; + parsed.hash = ""; + return parsed.toString(); + } catch { + return String(endpoint).replace(/\/\/[^/@\s]+@/, "//").replace(/[?#].*$/, ""); + } +} + // POST messages to Headroom /v1/compress; returns compressed messages + stats or null. -async function callCompress(url, messages, model, timeoutMs, compressUserMessages) { - const endpoint = `${String(url).replace(/\/$/, "")}/v1/compress`; +async function callCompress(url, messages, model, timeoutMs, compressUserMessages, diagnostics) { + const endpoint = buildCompressEndpoint(url); + diagnostics.endpoint = maskEndpoint(endpoint); const payload = { messages, model }; if (compressUserMessages) payload.config = { compress_user_messages: true }; - const res = await fetch(endpoint, { - method: "POST", - headers: { "Content-Type": "application/json" }, - body: JSON.stringify(payload), - signal: AbortSignal.timeout(timeoutMs), - }); - if (!res.ok) return null; + let res; + try { + res = await fetch(endpoint, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify(payload), + signal: AbortSignal.timeout(timeoutMs), + }); + } catch (error) { + setDiagnostic(diagnostics, `request failed: ${describeFetchError(error)}`); + return null; + } + if (!res.ok) { + setDiagnostic(diagnostics, `proxy returned HTTP ${res.status}`); + return null; + } const data = await res.json(); - if (!Array.isArray(data?.messages)) return null; + if (!Array.isArray(data?.messages)) { + setDiagnostic(diagnostics, "proxy response missing messages[]"); + return null; + } return data; } // Compress request body via Headroom proxy. Fail-open: returns null on any error. // /v1/compress only understands OpenAI shape, so Claude bodies are translated // to OpenAI, compressed, then translated back using 9Router's own translators. -export async function compressWithHeadroom(body, { enabled, url, model, format, compressUserMessages, timeoutMs = DEFAULT_TIMEOUT_MS } = {}) { - if (!enabled || !url || !body) return null; +export async function compressWithHeadroom(body, { enabled, url, model, format, compressUserMessages, timeoutMs = DEFAULT_TIMEOUT_MS, diagnostics = null } = {}) { + if (!enabled) { + setDiagnostic(diagnostics, "disabled"); + return null; + } + if (!url) { + setDiagnostic(diagnostics, "missing proxy URL"); + return null; + } + if (!body) { + setDiagnostic(diagnostics, "missing request body"); + return null; + } try { + if (diagnostics) diagnostics.before = captureSizeSnapshot(body); + // Claude shape: translate → OpenAI → compress → translate back. if (format === "claude") { const oai = claudeToOpenAIRequest(model, body, false); - if (!Array.isArray(oai?.messages)) return null; - const data = await callCompress(url, oai.messages, model, timeoutMs, compressUserMessages); + if (!Array.isArray(oai?.messages)) { + setDiagnostic(diagnostics, "Claude request did not translate to messages[]"); + return null; + } + const data = await callCompress(url, oai.messages, model, timeoutMs, compressUserMessages, diagnostics || {}); if (!data) return null; const claudeBody = openaiToClaudeRequest(model, { ...oai, messages: data.messages }, false); if (Array.isArray(claudeBody?.messages)) body.messages = claudeBody.messages; if (claudeBody?.system !== undefined) body.system = claudeBody.system; + if (diagnostics) diagnostics.after = captureSizeSnapshot(body); return data; } @@ -43,12 +139,17 @@ export async function compressWithHeadroom(body, { enabled, url, model, format, const key = Array.isArray(body.messages) ? "messages" : Array.isArray(body.input) ? "input" : null; - if (!key) return null; - const data = await callCompress(url, body[key], model, timeoutMs, compressUserMessages); + if (!key) { + setDiagnostic(diagnostics, `unsupported ${format || "unknown"} request shape`); + return null; + } + const data = await callCompress(url, body[key], model, timeoutMs, compressUserMessages, diagnostics || {}); if (!data) return null; body[key] = data.messages; + if (diagnostics) diagnostics.after = captureSizeSnapshot(body); return data; - } catch { + } catch (error) { + setDiagnostic(diagnostics, `unexpected error: ${error?.message || String(error)}`); return null; } } @@ -57,7 +158,22 @@ export function formatHeadroomLog(stats) { if (!stats) return null; const before = stats.tokens_before || 0; const after = stats.tokens_after || 0; - const saved = stats.tokens_saved || 0; - const pct = before > 0 ? ((saved / before) * 100).toFixed(1) : "0"; - return `saved ${saved} tokens / ${before} (${pct}%) ${after ? `after=${after}` : ""}`.trim(); + const delta = stats.tokens_saved || 0; + const pct = before > 0 ? ((delta / before) * 100).toFixed(1) : "0"; + return `reported token delta=${delta} before=${before}${after ? ` after=${after}` : ""} (${pct}%)`.trim(); +} + +export function formatHeadroomSizeLog(diagnostics) { + const before = diagnostics?.before; + const after = diagnostics?.after; + if (!before || !after) return ""; + return `body=${before.bodyBytes}B→${after.bodyBytes}B messages=${before.messageBytes}B→${after.messageBytes}B`; +} + +export function isHeadroomPhantomSavings(stats, diagnostics, minShrinkRatio = 0.05) { + if (!stats?.tokens_saved || stats.tokens_saved <= 0) return false; + const before = diagnostics?.before?.bodyBytes || 0; + const after = diagnostics?.after?.bodyBytes || 0; + if (before <= 0 || after <= 0) return false; + return after >= before * (1 - minShrinkRatio); } diff --git a/tests/unit/headroom-chat-core.test.js b/tests/unit/headroom-chat-core.test.js new file mode 100644 index 00000000..5c552ab9 --- /dev/null +++ b/tests/unit/headroom-chat-core.test.js @@ -0,0 +1,253 @@ +import { describe, it, expect, vi, beforeEach } from "vitest"; + +const { executeMock } = vi.hoisted(() => ({ + executeMock: vi.fn(), +})); + +vi.mock("../../open-sse/executors/index.js", () => ({ + getExecutor: () => ({ + noAuth: true, + execute: executeMock, + }), +})); + +vi.mock("../../open-sse/utils/requestLogger.js", () => ({ + createRequestLogger: async () => ({ + logClientRawRequest: vi.fn(), + logRawRequest: vi.fn(), + logTargetRequest: vi.fn(), + logProviderResponse: vi.fn(), + logConvertedResponse: vi.fn(), + logError: vi.fn(), + }), +})); + +vi.mock("../../open-sse/utils/stream.js", () => ({ + COLORS: { red: "", reset: "" }, + createPassthroughStreamWithLogger: vi.fn(() => new TransformStream()), +})); + +vi.mock("@/lib/usageDb.js", () => ({ + trackPendingRequest: vi.fn(), + appendRequestLog: vi.fn(async () => {}), + saveRequestDetail: vi.fn(async () => {}), +})); + +const { handleChatCore } = await import("../../open-sse/handlers/chatCore.js"); + +describe("handleChatCore Headroom diagnostics", () => { + beforeEach(() => { + vi.clearAllMocks(); + global.fetch = vi.fn(async (url) => { + if (String(url).includes("/v1/compress")) { + throw Object.assign(new Error("connect ECONNREFUSED 127.0.0.1:8787"), { code: "ECONNREFUSED" }); + } + throw new Error(`unexpected fetch: ${url}`); + }); + executeMock.mockResolvedValue({ + response: new Response(JSON.stringify({ + id: "chatcmpl-test", + object: "chat.completion", + choices: [{ message: { role: "assistant", content: "ok" }, finish_reason: "stop", index: 0 }], + }), { status: 200, headers: { "content-type": "application/json" } }), + url: "https://api.openai.com/v1/chat/completions", + headers: {}, + transformedBody: null, + }); + }); + + it("logs why Headroom was skipped on chat completions", async () => { + const log = { debug: vi.fn(), info: vi.fn(), warn: vi.fn() }; + + await handleChatCore({ + body: { model: "gpt-4o", stream: false, messages: [{ role: "user", content: "hello" }] }, + modelInfo: { provider: "openai", model: "gpt-4o" }, + credentials: { apiKey: "test-key", providerSpecificData: {} }, + log, + connectionId: "test-conn", + headroomEnabled: true, + headroomUrl: "http://localhost:8787", + headroomCompressUserMessages: false, + rtkEnabled: false, + cavemanEnabled: false, + ponytailEnabled: false, + clientRawRequest: { + endpoint: "/v1/chat/completions", + body: {}, + headers: { accept: "application/json" }, + }, + }); + + expect(log.warn).toHaveBeenCalledWith( + "HEADROOM", + expect.stringContaining("skipped: request failed") + ); + expect(log.warn).toHaveBeenCalledWith( + "HEADROOM", + expect.stringContaining("ECONNREFUSED") + ); + expect(log.warn).toHaveBeenCalledWith( + "HEADROOM", + expect.stringContaining("http://localhost:8787/v1/compress") + ); + }); + + it("scrubs credentials and query strings from Headroom fetch errors", async () => { + const log = { debug: vi.fn(), info: vi.fn(), warn: vi.fn() }; + + global.fetch = vi.fn(async () => { + throw new Error("failed to fetch https://user:secret@example.com:8787/proxy/v1/compress?token=abc123"); + }); + + await handleChatCore({ + body: { model: "gpt-4o", stream: false, messages: [{ role: "user", content: "hello" }] }, + modelInfo: { provider: "openai", model: "gpt-4o" }, + credentials: { apiKey: "test-key", providerSpecificData: {} }, + log, + connectionId: "test-conn", + headroomEnabled: true, + headroomUrl: "https://user:secret@example.com:8787/proxy?token=abc123", + headroomCompressUserMessages: false, + rtkEnabled: false, + cavemanEnabled: false, + ponytailEnabled: false, + clientRawRequest: { + endpoint: "/v1/chat/completions", + body: {}, + headers: { accept: "application/json" }, + }, + }); + + const logs = JSON.stringify(log.warn.mock.calls); + expect(logs).toContain("https://example.com:8787/proxy/v1/compress"); + expect(logs).not.toContain("user"); + expect(logs).not.toContain("secret"); + expect(logs).not.toContain("abc123"); + }); + + it("masks credentials and query strings in Headroom endpoint diagnostics", async () => { + const log = { debug: vi.fn(), info: vi.fn(), warn: vi.fn() }; + + await handleChatCore({ + body: { model: "gpt-4o", stream: false, messages: [{ role: "user", content: "hello" }] }, + modelInfo: { provider: "openai", model: "gpt-4o" }, + credentials: { apiKey: "test-key", providerSpecificData: {} }, + log, + connectionId: "test-conn", + headroomEnabled: true, + headroomUrl: "https://user:secret@example.com:8787/proxy?token=abc123", + headroomCompressUserMessages: false, + rtkEnabled: false, + cavemanEnabled: false, + ponytailEnabled: false, + clientRawRequest: { + endpoint: "/v1/chat/completions", + body: {}, + headers: { accept: "application/json" }, + }, + }); + + const logs = JSON.stringify(log.warn.mock.calls); + expect(global.fetch).toHaveBeenCalledWith( + "https://user:secret@example.com:8787/proxy/v1/compress?token=abc123", + expect.any(Object) + ); + expect(logs).toContain("https://example.com:8787/proxy/v1/compress"); + expect(logs).not.toContain("user"); + expect(logs).not.toContain("secret"); + expect(logs).not.toContain("abc123"); + }); + + it("sends Headroom-compressed messages to the provider executor", async () => { + const log = { debug: vi.fn(), info: vi.fn(), warn: vi.fn() }; + const original = "very large context that should be replaced"; + const compressed = "compressed context"; + + global.fetch = vi.fn(async (url) => { + if (String(url).includes("/v1/compress")) { + return new Response(JSON.stringify({ + messages: [{ role: "user", content: compressed }], + tokens_before: 100, + tokens_after: 10, + tokens_saved: 90, + }), { status: 200, headers: { "content-type": "application/json" } }); + } + throw new Error(`unexpected fetch: ${url}`); + }); + + await handleChatCore({ + body: { model: "gpt-4o", stream: false, messages: [{ role: "user", content: original }] }, + modelInfo: { provider: "openai", model: "gpt-4o" }, + credentials: { apiKey: "test-key", providerSpecificData: {} }, + log, + connectionId: "test-conn", + headroomEnabled: true, + headroomUrl: "http://localhost:8787", + headroomCompressUserMessages: false, + rtkEnabled: false, + cavemanEnabled: false, + ponytailEnabled: false, + clientRawRequest: { + endpoint: "/v1/chat/completions", + body: {}, + headers: { accept: "application/json" }, + }, + }); + + expect(executeMock).toHaveBeenCalledWith(expect.objectContaining({ + body: expect.objectContaining({ + messages: [{ role: "user", content: compressed }], + }), + })); + expect(JSON.stringify(executeMock.mock.calls[0][0].body)).not.toContain(original); + expect(log.info).toHaveBeenCalledWith("HEADROOM", expect.stringContaining("reported token delta=90 before=100 after=10")); + expect(log.info).toHaveBeenCalledWith("HEADROOM", expect.stringContaining("body=")); + expect(log.info).toHaveBeenCalledWith("HEADROOM", expect.stringContaining("messages=")); + + const logs = JSON.stringify([...log.info.mock.calls, ...log.warn.mock.calls]); + expect(logs).not.toContain("saved"); + expect(logs).not.toContain(original); + }); + + it("warns when Headroom reports savings but outbound body barely shrinks", async () => { + const log = { debug: vi.fn(), info: vi.fn(), warn: vi.fn() }; + const original = "x".repeat(1000); + const nearlySame = "x".repeat(990); + + global.fetch = vi.fn(async (url) => { + if (String(url).includes("/v1/compress")) { + return new Response(JSON.stringify({ + messages: [{ role: "user", content: nearlySame }], + tokens_before: 1000, + tokens_after: 100, + tokens_saved: 900, + }), { status: 200, headers: { "content-type": "application/json" } }); + } + throw new Error(`unexpected fetch: ${url}`); + }); + + await handleChatCore({ + body: { model: "gpt-4o", stream: false, messages: [{ role: "user", content: original }] }, + modelInfo: { provider: "openai", model: "gpt-4o" }, + credentials: { apiKey: "test-key", providerSpecificData: {} }, + log, + connectionId: "test-conn", + headroomEnabled: true, + headroomUrl: "http://localhost:8787", + headroomCompressUserMessages: false, + rtkEnabled: false, + cavemanEnabled: false, + ponytailEnabled: false, + clientRawRequest: { + endpoint: "/v1/chat/completions", + body: {}, + headers: { accept: "application/json" }, + }, + }); + + expect(log.warn).toHaveBeenCalledWith( + "HEADROOM", + expect.stringContaining("reported token delta, but outbound JSON shrank <5%; provider may bill near-original payload") + ); + }); +}); diff --git a/tests/unit/headroom.test.js b/tests/unit/headroom.test.js index f4e4baca..46468595 100644 --- a/tests/unit/headroom.test.js +++ b/tests/unit/headroom.test.js @@ -66,8 +66,8 @@ describe("compressWithHeadroom", () => { }); describe("formatHeadroomLog", () => { - it("formats savings", () => { + it("formats reported token deltas without implying provider billing savings", () => { expect(formatHeadroomLog({ tokens_before: 100, tokens_after: 25, tokens_saved: 75 })) - .toBe("saved 75 tokens / 100 (75.0%) after=25"); + .toBe("reported token delta=75 before=100 after=25 (75.0%)"); }); });