From 323d9cb5d94c4173881690e15699a4e224622803 Mon Sep 17 00:00:00 2001 From: BigSimmo <87357024+BigSimmo@users.noreply.github.com> Date: Mon, 13 Jul 2026 21:54:41 +0800 Subject: [PATCH] feat(observability): OpenAI answer spend snapshot on the deep health probe MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wave 2a of the repo-productivity program. Turns the already-persisted per-answer token counts (rag_retrieval_logs.metadata.answer.tokens) into a cost signal so a budget regression — like the reasoning-token starvation incident, or a pricing/ traffic change — is visible before the bill arrives. - src/lib/observability/spend-metrics.ts: spendSnapshot() aggregates answer-path token usage over a trailing window, derives USD from a configurable per-token price, and splits by route (fast/strong) and model, with a 24h projection and an alert flag. Zero-migration (reads existing JSONB); reasoning tokens surfaced but billed within output (never double-counted); cached input billed at the cheaper rate. Throws on query error so the caller nulls the block. - health-response.ts: optional `spend` block gated exactly like `slo`/`cache` (token-authorized deep probe + healthy Supabase, errors → null, never flips liveness). Suppressible via includeSpend: false. - env.ts: OPENAI_PRICE_{INPUT,CACHED_INPUT,OUTPUT}_PER_MTOK (operator-tunable placeholders) + SPEND_ALERT_DAILY_USD. - tests/spend-metrics.test.ts: 9 offline tests (cost math, route/model split, daily projection + alert, empty window, truncated sample, error path). Verified: 20/20 vitest (spend + health-route + answer-slo), tsc --noEmit, eslint, prettier — all green. No live provider calls. Co-Authored-By: Claude Fable 5 --- src/lib/env.ts | 12 ++ src/lib/health-response.ts | 26 +++ src/lib/observability/spend-metrics.ts | 219 +++++++++++++++++++++++++ tests/spend-metrics.test.ts | 125 ++++++++++++++ 4 files changed, 382 insertions(+) create mode 100644 src/lib/observability/spend-metrics.ts create mode 100644 tests/spend-metrics.test.ts diff --git a/src/lib/env.ts b/src/lib/env.ts index e2d849e4d..1ba265d8b 100644 --- a/src/lib/env.ts +++ b/src/lib/env.ts @@ -46,6 +46,18 @@ const envSchema = z.object({ // in openai.ts), and the answer path self-heals a truncation by retrying with a larger // cap before falling back. OPENAI_MAX_OUTPUT_TOKENS: z.coerce.number().int().positive().default(16000), + // Answer-generation pricing for the /api/health `spend` block (USD per 1,000,000 + // tokens). Operator-tunable PLACEHOLDERS — set these to the live OpenAI price for + // the answer model. They only derive a cost SIGNAL from already-recorded token + // counts (src/lib/observability/spend-metrics.ts); they have no billing effect. + // Cached input is a subset of input billed at the cached rate; reasoning tokens + // are billed within output, so output covers them. + OPENAI_PRICE_INPUT_PER_MTOK: z.coerce.number().nonnegative().default(1.25), + OPENAI_PRICE_CACHED_INPUT_PER_MTOK: z.coerce.number().nonnegative().default(0.125), + OPENAI_PRICE_OUTPUT_PER_MTOK: z.coerce.number().nonnegative().default(10), + // Projected-daily-spend alert threshold (USD) for the `spend` block. 0 disables + // the alert flag (the spend figures are still reported). + SPEND_ALERT_DAILY_USD: z.coerce.number().nonnegative().default(0), OPENAI_QUERY_CACHE_SIZE: z.coerce.number().int().nonnegative().default(200), // Max inputs per embeddings request. The OpenAI embeddings endpoint caps a single // request at 2048 inputs / ~300k tokens; a full-corpus re-embed of ~400k texts in one diff --git a/src/lib/health-response.ts b/src/lib/health-response.ts index 628f1773a..85c8a5a8c 100644 --- a/src/lib/health-response.ts +++ b/src/lib/health-response.ts @@ -3,12 +3,14 @@ import { allowDeepHealthProbe } from "@/lib/deep-probe-auth"; import { env, isDemoMode } from "@/lib/env"; import type { AnswerSloSnapshot } from "@/lib/observability/answer-slo"; import { cacheMetricsSnapshot, type CacheMetricsSnapshot } from "@/lib/observability/cache-metrics"; +import type { SpendProbeClient, SpendSnapshot } from "@/lib/observability/spend-metrics"; type HealthResponseOptions = { forceDeep?: boolean; allowUnauthenticatedDeep?: boolean; includeSlo?: boolean; includeCache?: boolean; + includeSpend?: boolean; }; export async function healthResponse(request: Request, options: HealthResponseOptions = {}) { @@ -20,6 +22,7 @@ export async function healthResponse(request: Request, options: HealthResponseOp }; let slo: AnswerSloSnapshot | null = null; let cache: CacheMetricsSnapshot | null = null; + let spend: SpendSnapshot | null = null; if (deep) { const tokenAuthorized = allowDeepHealthProbe(request); @@ -53,6 +56,28 @@ export async function healthResponse(request: Request, options: HealthResponseOp slo = null; } } + // Answer-generation spend, gated like `slo` (token-authorized deep probe + // + healthy Supabase). Derives USD from already-recorded token counts and + // a configurable price; errors are swallowed to null and never flip + // liveness. Suppressed only when explicitly disabled. + if (health.ok && tokenAuthorized && options.includeSpend !== false) { + try { + const { spendSnapshot } = await import("@/lib/observability/spend-metrics"); + // admin is structurally a SpendProbeClient; cast avoids a deep + // instantiation check against the full PostgREST client type (TS2589), + // matching how answer-slo/health treat the admin client. + spend = await spendSnapshot(admin as unknown as SpendProbeClient, { + pricing: { + inputPerMTok: env.OPENAI_PRICE_INPUT_PER_MTOK, + cachedInputPerMTok: env.OPENAI_PRICE_CACHED_INPUT_PER_MTOK, + outputPerMTok: env.OPENAI_PRICE_OUTPUT_PER_MTOK, + }, + alertDailyUsd: env.SPEND_ALERT_DAILY_USD, + }); + } catch { + spend = null; + } + } } catch { checks.supabase = "error"; } @@ -75,6 +100,7 @@ export async function healthResponse(request: Request, options: HealthResponseOp checks, ...(slo ? { slo } : {}), ...(cache ? { cache } : {}), + ...(spend ? { spend } : {}), }, { status: ready ? 200 : 503, headers: { "Cache-Control": "no-store" } }, ); diff --git a/src/lib/observability/spend-metrics.ts b/src/lib/observability/spend-metrics.ts new file mode 100644 index 000000000..8eafd6e38 --- /dev/null +++ b/src/lib/observability/spend-metrics.ts @@ -0,0 +1,219 @@ +// OpenAI answer-generation spend snapshot for the deep /api/health probe. +// +// Token counts are already the durable primitive: src/lib/answer-telemetry.ts +// writes one rag_retrieval_logs row per answered request with +// metadata.answer.tokens { input, output, total, cached_input, reasoning_output } +// plus metadata.answer.route and .model. Nothing turns those into a cost signal, +// so a budget regression (e.g. the reasoning-token starvation incident, or a +// pricing/traffic change) is invisible until the bill arrives. This aggregates +// the persisted counts over a trailing window and derives USD from a configurable +// per-token price, split by route and model, plus a 24h projection for alerting. +// +// USD is derived, never stored (per the answer-telemetry design note). Reasoning +// output tokens are a SUBSET of output_tokens in the OpenAI Responses API, so they +// are surfaced for visibility but billed as part of output — never double-counted. +// Cached input tokens are a subset of input, billed at the cheaper cached rate. +// +// Like `slo`/`cache`, this runs behind the secret-gated deep probe, reads only +// aggregate token metadata (never raw query text), and never flips liveness. + +export type SpendPricing = { + // USD per 1,000,000 tokens. + inputPerMTok: number; + cachedInputPerMTok: number; + outputPerMTok: number; +}; + +export type ModelSpend = { + model: string; + answers: number; + inputTokens: number; + cachedInputTokens: number; + outputTokens: number; + reasoningOutputTokens: number; + usd: number; +}; + +export type SpendSnapshot = { + windowMinutes: number; + answers: number; + tokens: { + input: number; + cachedInput: number; + output: number; + reasoningOutput: number; + total: number; + }; + usd: number; + usdByRoute: Record; + usdByModel: ModelSpend[]; + // usd normalized to a 24h rate, so a short probe window still yields a + // comparable daily figure for the alert threshold. + projectedDailyUsd: number; + alertDailyUsdThreshold: number | null; + alerting: boolean; + // True when the row sample hit the cap, so the totals are a lower bound. + sampleTruncated: boolean; + pricing: SpendPricing; +}; + +type AnswerTokens = { + input?: number | null; + output?: number | null; + total?: number | null; + cached_input?: number | null; + reasoning_output?: number | null; +}; + +type SpendRow = { + query_class?: string | null; + metadata?: { + answer?: { + route?: string | null; + model?: string | null; + tokens?: AnswerTokens | null; + } | null; + } | null; +}; + +type SelectResult = { data: SpendRow[] | null; error: unknown }; + +// Minimal structural view of the PostgREST select we use — the Supabase admin +// client is assignable to this, and tests pass a small fake (same approach as +// answer-slo.ts / supabase/health.ts). +type SpendQueryBuilder = PromiseLike & { + gt(column: string, value: string): SpendQueryBuilder; + eq(column: string, value: string): SpendQueryBuilder; + limit(count: number): SpendQueryBuilder; +}; + +export type SpendProbeClient = { + from(table: string): { + select(columns: string): SpendQueryBuilder; + }; +}; + +const DEFAULT_ROW_CAP = 5000; + +function num(value: number | null | undefined): number { + return typeof value === "number" && Number.isFinite(value) ? value : 0; +} + +function round(value: number, dp = 6): number { + const factor = 10 ** dp; + return Math.round(value * factor) / factor; +} + +/** + * Cost of one answer's token usage. Uncached input and cached input are billed at + * their respective rates (cached is a subset of input); output already includes + * reasoning tokens, so output is billed once. Pure — unit-tested directly. + */ +export function computeAnswerCostUsd(tokens: AnswerTokens, pricing: SpendPricing): number { + const input = num(tokens.input); + const cached = Math.min(num(tokens.cached_input), input); + const uncachedInput = input - cached; + const output = num(tokens.output); + const usd = + (uncachedInput * pricing.inputPerMTok + cached * pricing.cachedInputPerMTok + output * pricing.outputPerMTok) / + 1_000_000; + return usd; +} + +export type SpendSnapshotOptions = { + windowMinutes?: number; + pricing: SpendPricing; + alertDailyUsd?: number | null; + rowCap?: number; +}; + +/** + * Aggregate answer-path token spend over the trailing window. Throws on a query + * error so the caller can null the block rather than report a falsely-zero spend. + */ +export async function spendSnapshot(client: SpendProbeClient, options: SpendSnapshotOptions): Promise { + const windowMinutes = options.windowMinutes ?? 60; + const rowCap = options.rowCap ?? DEFAULT_ROW_CAP; + const pricing = options.pricing; + const alertDailyUsdThreshold = + typeof options.alertDailyUsd === "number" && options.alertDailyUsd > 0 ? options.alertDailyUsd : null; + const sinceIso = new Date(Date.now() - windowMinutes * 60_000).toISOString(); + + const result = await client + .from("rag_retrieval_logs") + .select("query_class, metadata") + .gt("created_at", sinceIso) + .eq("metadata->answer->>log_source", "answer") + .limit(rowCap); + + if (result.error) { + if (result.error instanceof Error) throw result.error; + const message = + result.error && typeof result.error === "object" && "message" in result.error + ? String((result.error as { message?: unknown }).message ?? "") + : String(result.error); + throw new Error(message || "spend snapshot query failed"); + } + + const rows = result.data ?? []; + const totals = { input: 0, cachedInput: 0, output: 0, reasoningOutput: 0, total: 0 }; + const usdByRoute: Record = {}; + const byModel = new Map(); + let usd = 0; + + for (const row of rows) { + const answer = row.metadata?.answer; + const tokens = answer?.tokens ?? {}; + const input = num(tokens.input); + const cached = Math.min(num(tokens.cached_input), input); + const output = num(tokens.output); + const reasoning = num(tokens.reasoning_output); + const rowUsd = computeAnswerCostUsd(tokens, pricing); + + totals.input += input; + totals.cachedInput += cached; + totals.output += output; + totals.reasoningOutput += reasoning; + totals.total += num(tokens.total) || input + output; + usd += rowUsd; + + const route = answer?.route ?? "unknown"; + usdByRoute[route] = (usdByRoute[route] ?? 0) + rowUsd; + + const model = answer?.model ?? "unknown"; + const entry = byModel.get(model) ?? { + model, + answers: 0, + inputTokens: 0, + cachedInputTokens: 0, + outputTokens: 0, + reasoningOutputTokens: 0, + usd: 0, + }; + entry.answers += 1; + entry.inputTokens += input; + entry.cachedInputTokens += cached; + entry.outputTokens += output; + entry.reasoningOutputTokens += reasoning; + entry.usd += rowUsd; + byModel.set(model, entry); + } + + const projectedDailyUsd = windowMinutes > 0 ? usd * (1440 / windowMinutes) : 0; + + return { + windowMinutes, + answers: rows.length, + tokens: totals, + usd: round(usd), + usdByRoute: Object.fromEntries(Object.entries(usdByRoute).map(([route, value]) => [route, round(value)])), + usdByModel: [...byModel.values()] + .map((entry) => ({ ...entry, usd: round(entry.usd) })) + .sort((a, b) => b.usd - a.usd), + projectedDailyUsd: round(projectedDailyUsd, 2), + alertDailyUsdThreshold, + alerting: alertDailyUsdThreshold !== null && projectedDailyUsd > alertDailyUsdThreshold, + sampleTruncated: rows.length >= rowCap, + pricing, + }; +} diff --git a/tests/spend-metrics.test.ts b/tests/spend-metrics.test.ts new file mode 100644 index 000000000..088b979cd --- /dev/null +++ b/tests/spend-metrics.test.ts @@ -0,0 +1,125 @@ +import { describe, expect, it } from "vitest"; + +import { + computeAnswerCostUsd, + type SpendPricing, + type SpendProbeClient, + spendSnapshot, +} from "@/lib/observability/spend-metrics"; + +const PRICING: SpendPricing = { inputPerMTok: 1, cachedInputPerMTok: 0.1, outputPerMTok: 10 }; + +type FakeRow = { + query_class?: string | null; + metadata?: { answer?: { route?: string; model?: string; tokens?: Record } }; +}; + +// Fake PostgREST select builder: from().select().gt().eq().limit() resolves to +// { data, error }. Mirrors the answer-slo test's structural fake. +function fakeClient(rows: FakeRow[], error?: unknown): SpendProbeClient { + const builder = { + gt: () => builder, + eq: () => builder, + limit: () => builder, + then: (resolve: (value: { data: FakeRow[] | null; error: unknown }) => unknown) => + resolve({ data: error ? null : rows, error: error ?? null }), + }; + return { from: () => ({ select: () => builder }) } as unknown as SpendProbeClient; +} + +const answerRow = (route: string, model: string, tokens: Record): FakeRow => ({ + metadata: { answer: { route, model, tokens } }, +}); + +describe("computeAnswerCostUsd", () => { + it("bills uncached input, cached input, and output at their rates", () => { + // 1M uncached input @1 + 1M cached @0.1 is impossible (cached ⊆ input); use a subset. + const usd = computeAnswerCostUsd({ input: 1_000_000, cached_input: 200_000, output: 1_000_000 }, PRICING); + // uncached 800k @1 + cached 200k @0.1 + output 1M @10 = 0.8 + 0.02 + 10 = 10.82 + expect(usd).toBeCloseTo(0.8 + 0.02 + 10, 6); + }); + + it("never counts cached input beyond total input", () => { + const usd = computeAnswerCostUsd({ input: 100, cached_input: 500, output: 0 }, PRICING); + // cached clamped to 100 → all cached → 100 @0.1 / 1e6 + expect(usd).toBeCloseTo((100 * 0.1) / 1_000_000, 9); + }); + + it("treats missing/NaN token fields as zero", () => { + expect(computeAnswerCostUsd({}, PRICING)).toBe(0); + }); +}); + +describe("spendSnapshot", () => { + it("aggregates tokens and USD, split by route and model", async () => { + const rows = [ + answerRow("fast", "gpt-5.5", { input: 1000, cached_input: 0, output: 500, reasoning_output: 100, total: 1500 }), + answerRow("strong", "gpt-5.5", { + input: 2000, + cached_input: 500, + output: 4000, + reasoning_output: 3000, + total: 6000, + }), + ]; + const snap = await spendSnapshot(fakeClient(rows), { windowMinutes: 60, pricing: PRICING }); + + expect(snap.answers).toBe(2); + expect(snap.tokens.input).toBe(3000); + expect(snap.tokens.output).toBe(4500); + expect(snap.tokens.reasoningOutput).toBe(3100); + // fast: 1000@1 + 500@10 = 0.001 + 0.005 = 0.006 + // strong: uncached 1500@1 + cached 500@0.1 + 4000@10 = 0.0015 + 0.00005 + 0.04 = 0.04155 + expect(snap.usd).toBeCloseTo(0.006 + 0.04155, 6); + expect(snap.usdByRoute.fast).toBeCloseTo(0.006, 6); + expect(snap.usdByRoute.strong).toBeCloseTo(0.04155, 6); + expect(snap.usdByModel).toHaveLength(1); + expect(snap.usdByModel[0].model).toBe("gpt-5.5"); + expect(snap.usdByModel[0].answers).toBe(2); + }); + + it("projects a daily rate from a short window and flags the alert", async () => { + const rows = [answerRow("strong", "gpt-5.5", { input: 0, output: 1_000_000 })]; // $10 in the window + const snap = await spendSnapshot(fakeClient(rows), { + windowMinutes: 60, + pricing: PRICING, + alertDailyUsd: 100, + }); + // $10/hour → $240/day > $100 threshold + expect(snap.projectedDailyUsd).toBeCloseTo(240, 2); + expect(snap.alerting).toBe(true); + }); + + it("does not alert when under threshold or when threshold is 0/disabled", async () => { + const rows = [answerRow("fast", "gpt-5.5", { input: 100, output: 100 })]; + const under = await spendSnapshot(fakeClient(rows), { windowMinutes: 60, pricing: PRICING, alertDailyUsd: 1000 }); + expect(under.alerting).toBe(false); + const disabled = await spendSnapshot(fakeClient(rows), { windowMinutes: 60, pricing: PRICING, alertDailyUsd: 0 }); + expect(disabled.alertDailyUsdThreshold).toBeNull(); + expect(disabled.alerting).toBe(false); + }); + + it("returns a zeroed snapshot for an empty window without dividing by zero", async () => { + const snap = await spendSnapshot(fakeClient([]), { windowMinutes: 60, pricing: PRICING, alertDailyUsd: 10 }); + expect(snap.answers).toBe(0); + expect(snap.usd).toBe(0); + expect(snap.projectedDailyUsd).toBe(0); + expect(snap.alerting).toBe(false); + expect(snap.sampleTruncated).toBe(false); + }); + + it("flags a truncated sample when the row cap is hit", async () => { + const rows = [ + answerRow("fast", "gpt-5.5", { input: 1, output: 1 }), + answerRow("fast", "gpt-5.5", { input: 1, output: 1 }), + ]; + const snap = await spendSnapshot(fakeClient(rows), { windowMinutes: 60, pricing: PRICING, rowCap: 2 }); + expect(snap.sampleTruncated).toBe(true); + }); + + it("throws on a query error so the caller can null the block", async () => { + await expect( + spendSnapshot(fakeClient([], { message: "boom" }), { windowMinutes: 60, pricing: PRICING }), + ).rejects.toThrow("boom"); + }); +});