From a668bf8bbe8641ed6f7677ba0bcaf804e8ff1ba3 Mon Sep 17 00:00:00 2001 From: Ravi Tharuma Date: Thu, 3 Sep 2026 22:08:28 +0200 Subject: [PATCH] feat: map CLIProxyAPI usage and tok/s onto OpenCode payloads Copy gateway tokens_per_second onto usage. Never invent tok/s from latency. Closes #5 --- src/index.ts | 34 ++++++++ src/telemetry.test.ts | 57 +++++++++++++ src/telemetry.ts | 180 ++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 271 insertions(+) create mode 100644 src/telemetry.test.ts create mode 100644 src/telemetry.ts diff --git a/src/index.ts b/src/index.ts index a35dd02..8e25c60 100644 --- a/src/index.ts +++ b/src/index.ts @@ -6,6 +6,7 @@ import { type CatalogModel, type ModelProtocolCatalog, } from "./catalog.js" +import { applyGatewayInferenceTelemetry } from "./telemetry.js" const DEFAULT_BASE_URL = "http://localhost:8317/v1" const DEFAULT_PROVIDER_ID = "cliproxyapi" @@ -29,7 +30,21 @@ type ModelConfig = NonNullable[string] export const CLIProxyAPIPlugin: Plugin = async ({ client }, rawOptions) => { const options = readOptions(rawOptions) + const fetchInterceptor = createCliproxyFetchInterceptor(options) + return { + auth: { + provider: options.providerID ?? DEFAULT_PROVIDER_ID, + methods: [ + { + type: "api", + label: "CLIProxyAPI API key", + }, + ], + loader: async () => ({ + fetch: fetchInterceptor, + }), + }, config: async (config) => { const providerID = options.providerID ?? DEFAULT_PROVIDER_ID const existing = config.provider?.[providerID] @@ -262,3 +277,22 @@ function isReasoningModel(modelID: string) { function supportsAttachments(modelID: string) { return /^(?:claude|gemini|gpt)/i.test(modelID) } + +function createCliproxyFetchInterceptor(options: ConnectorOptions) { + const baseURL = normalizeBaseURL( + options.baseURL ?? process.env.CLIPROXY_BASE_URL ?? DEFAULT_BASE_URL, + ) + const normalized = baseURL.endsWith("/") ? baseURL : `${baseURL}/` + return async (input: RequestInfo | URL, init?: RequestInit) => { + const url = input instanceof Request ? input.url : input.toString() + if (url !== baseURL && !url.startsWith(normalized)) { + return fetch(input, init) + } + const headers = new Headers(input instanceof Request ? input.headers : undefined) + if (init?.headers) new Headers(init.headers).forEach((value, key) => headers.set(key, value)) + const apiKey = options.apiKey ?? process.env.CLIPROXY_API_KEY + if (apiKey) headers.set("Authorization", `Bearer ${apiKey}`) + return applyGatewayInferenceTelemetry(await fetch(input, { ...init, headers })) + } +} + diff --git a/src/telemetry.test.ts b/src/telemetry.test.ts new file mode 100644 index 0000000..3d5ed77 --- /dev/null +++ b/src/telemetry.test.ts @@ -0,0 +1,57 @@ +import { describe, expect, test } from "bun:test" +import { + applyGatewayInferenceTelemetry, + attachGatewayTelemetryToPayload, + parseGatewayInferenceTelemetry, + tokensPerSecondFromUsage, +} from "./telemetry.js" + +describe("gateway inference telemetry", () => { + test("copies tok/s and winning model from gateway headers", () => { + const headers = new Headers({ + "X-CLIProxyAPI-Tokens-Per-Second": "80.5", + "X-OmniRoute-Model": "winner-model", + "X-OmniRoute-Response-Cost": "0.0123", + }) + const got = parseGatewayInferenceTelemetry(headers) + expect(got.tokensPerSecond).toBe(80.5) + expect(got.model).toBe("winner-model") + expect(got.costUsd).toBe(0.0123) + }) + + test("does not invent tok/s from tokens / latency", () => { + const headers = new Headers({ + "X-OmniRoute-Tokens-Out": "200", + "X-OmniRoute-Latency-Ms": "2000", + }) + const got = parseGatewayInferenceTelemetry(headers) + expect(got.tokensPerSecond).toBeUndefined() + const payload = attachGatewayTelemetryToPayload( + { object: "chat.completion", usage: { completion_tokens: 200 } }, + got, + ) as { usage: { tokens_per_second?: number } } + expect(payload.usage.tokens_per_second).toBeUndefined() + expect(tokensPerSecondFromUsage({ completion_tokens: 200, latency_ms: 2000 })).toBeUndefined() + }) + + test("JSON response attach writes usage.tokens_per_second", async () => { + const response = new Response( + JSON.stringify({ + object: "chat.completion", + model: "combo/auto", + usage: { completion_tokens: 20 }, + }), + { + headers: { + "Content-Type": "application/json", + "X-CLIProxyAPI-Tokens-Per-Second": "40", + "X-OmniRoute-Model": "winner", + }, + }, + ) + const next = await applyGatewayInferenceTelemetry(response) + const body = JSON.parse(await next.text()) as { model: string; usage: { tokens_per_second: number } } + expect(body.model).toBe("winner") + expect(body.usage.tokens_per_second).toBe(40) + }) +}) diff --git a/src/telemetry.ts b/src/telemetry.ts new file mode 100644 index 0000000..74c78fd --- /dev/null +++ b/src/telemetry.ts @@ -0,0 +1,180 @@ +/** + * Prefer gateway-reported inference telemetry. Never invent tok/s from + * tokens / latency (that includes TTFT and tool pauses). + */ +export type GatewayInferenceTelemetry = { + costUsd?: number; + tokensPerSecond?: number; + ttftMs?: number; + model?: string; +}; + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +function readFiniteNumber(value: unknown): number | undefined { + if (typeof value === "number" && Number.isFinite(value)) return value; + if (typeof value === "string" && value.trim() !== "") { + const parsed = Number(value); + if (Number.isFinite(parsed)) return parsed; + } + return undefined; +} + +function readPositiveNumber(value: unknown): number | undefined { + const parsed = readFiniteNumber(value); + if (parsed === undefined || parsed <= 0) return undefined; + return parsed; +} + +function header(headers: Headers, name: string): string | null { + return headers.get(name); +} + +export function parseGatewayInferenceTelemetry(headers: Headers): GatewayInferenceTelemetry { + const out: GatewayInferenceTelemetry = {}; + const cost = readFiniteNumber( + header(headers, "x-omniroute-response-cost") ?? header(headers, "x-cliproxyapi-response-cost"), + ); + if (cost !== undefined && cost >= 0) out.costUsd = cost; + const tps = readPositiveNumber( + header(headers, "x-omniroute-tokens-per-second") ?? + header(headers, "x-cliproxyapi-tokens-per-second") ?? + header(headers, "x-cliproxy-tokens-per-second"), + ); + if (tps !== undefined) out.tokensPerSecond = tps; + const ttft = readPositiveNumber( + header(headers, "x-omniroute-ttft-ms") ?? header(headers, "x-cliproxyapi-ttft-ms"), + ); + if (ttft !== undefined) out.ttftMs = ttft; + const model = + header(headers, "x-omniroute-model") ?? header(headers, "x-cliproxyapi-model"); + if (model && model.trim()) out.model = model.trim(); + return out; +} + +export function tokensPerSecondFromUsage(usage: unknown): number | undefined { + if (!isRecord(usage)) return undefined; + return ( + readPositiveNumber(usage.tokens_per_second) ?? + readPositiveNumber(usage.tokensPerSecond) + ); +} + +export function formatToksDisplay(tps: number | undefined): string { + return tps === undefined ? "—" : String(tps); +} + +function isInferencePayload(payload: Record): boolean { + return ( + isRecord(payload.usage) || + Array.isArray(payload.choices) || + payload.object === "chat.completion" || + payload.object === "chat.completion.chunk" || + payload.object === "response" || + payload.type === "message" || + Array.isArray(payload.output) + ); +} + +function attachToUsage( + usage: Record, + telemetry: GatewayInferenceTelemetry, +): Record { + const next = { ...usage }; + const existing = tokensPerSecondFromUsage(next); + if (existing === undefined && telemetry.tokensPerSecond !== undefined) { + next.tokens_per_second = telemetry.tokensPerSecond; + } + if (telemetry.ttftMs !== undefined && readPositiveNumber(next.ttft_ms) === undefined) { + next.ttft_ms = telemetry.ttftMs; + } + if (telemetry.costUsd !== undefined && typeof next.cost !== "number") { + next.cost = telemetry.costUsd; + } + return next; +} + +export function attachGatewayTelemetryToPayload( + payload: unknown, + telemetry: GatewayInferenceTelemetry, +): unknown { + if (!isRecord(payload) || !isInferencePayload(payload)) return payload; + const next: Record = { ...payload }; + if (telemetry.model) next.model = telemetry.model; + if (isRecord(next.usage)) { + next.usage = attachToUsage(next.usage, { + ...telemetry, + tokensPerSecond: telemetry.tokensPerSecond ?? tokensPerSecondFromUsage(next.usage), + }); + } + return next; +} + +export function attachGatewayTelemetryToSseLine( + line: string, + telemetry: GatewayInferenceTelemetry, +): string { + const trimmed = line.trim(); + if (!trimmed.startsWith("data:")) return line; + const jsonText = trimmed.slice("data:".length).trim(); + if (!jsonText.startsWith("{")) return line; + try { + const parsed = JSON.parse(jsonText) as unknown; + const updated = attachGatewayTelemetryToPayload(parsed, telemetry); + if (updated === parsed) return line; + const prefix = line.slice(0, line.indexOf(jsonText)); + return `${prefix}${JSON.stringify(updated)}`; + } catch { + return line; + } +} + +export async function applyGatewayInferenceTelemetry(response: Response): Promise { + const telemetry = parseGatewayInferenceTelemetry(response.headers); + const contentType = response.headers.get("content-type") ?? ""; + if (contentType.includes("text/event-stream") && response.body) { + const decoder = new TextDecoder(); + const encoder = new TextEncoder(); + let pending = ""; + const mapped = response.body.pipeThrough( + new TransformStream({ + transform(chunk, controller) { + pending += decoder.decode(chunk, { stream: true }); + const lines = pending.split("\n"); + pending = lines.pop() ?? ""; + for (const line of lines) { + controller.enqueue(encoder.encode(`${attachGatewayTelemetryToSseLine(line, telemetry)}\n`)); + } + }, + flush(controller) { + if (pending.length > 0) { + controller.enqueue(encoder.encode(attachGatewayTelemetryToSseLine(pending, telemetry))); + } + }, + }), + ); + return new Response(mapped, { + status: response.status, + statusText: response.statusText, + headers: response.headers, + }); + } + if (!contentType.includes("json")) return response; + const text = await response.text(); + try { + const parsed = JSON.parse(text) as unknown; + const next = attachGatewayTelemetryToPayload(parsed, telemetry); + if (next === parsed) { + return new Response(text, { status: response.status, statusText: response.statusText, headers: response.headers }); + } + return new Response(JSON.stringify(next), { + status: response.status, + statusText: response.statusText, + headers: response.headers, + }); + } catch { + return new Response(text, { status: response.status, statusText: response.statusText, headers: response.headers }); + } +}