diff --git a/src/app/api/ai/test/__tests__/route.test.ts b/src/app/api/ai/test/__tests__/route.test.ts index e983939cb..2cf725575 100644 --- a/src/app/api/ai/test/__tests__/route.test.ts +++ b/src/app/api/ai/test/__tests__/route.test.ts @@ -24,6 +24,17 @@ vi.mock("@/lib/rate-limit", () => ({ checkRateLimit: vi.fn(async () => ({ allowed: true })), })); +vi.mock("@/lib/ai/coach/budget", () => ({ + buildDateKey: () => "2026-01-01", + reserveBudget: vi.fn(async () => ({ + allowed: true, + reserved: 32, + totalAfter: 32, + })), + reconcileSpend: vi.fn(async () => undefined), + resolveDailyCap: () => 200_000, +})); + vi.mock("@/lib/logging/context", () => ({ annotate: vi.fn(), })); @@ -278,3 +289,72 @@ describe("POST /api/ai/test — dropdown-aware override", () => { expect(body.error).toBe("Anthropic API key not configured"); }); }); + +/** + * Cost ceiling. The 5/min bucket bounded burst rate but nothing cumulative — + * sustained, it permits ~7 200 probes a day, and with an empty body this route + * resolves the SAME provider chain generation uses, which can be the + * operator's own credential. So the surface could reach the operator's key + * indefinitely with no daily ceiling and no ledger entry at all. + */ +describe("POST /api/ai/test — daily ceiling + ledger", () => { + function okProvider() { + vi.mocked(resolveProviderForTest).mockResolvedValue({ + type: "anthropic", + generateCompletion: vi.fn(async () => ({ + content: '{"ok":true}', + providerType: "anthropic", + model: "claude-3-5-sonnet-latest", + tokensUsed: 5, + })), + } as never); + } + + it("refuses once the per-user daily probe limit is spent", async () => { + const { checkRateLimit } = await import("@/lib/rate-limit"); + okProvider(); + // Burst bucket fine, cumulative daily bucket exhausted. + vi.mocked(checkRateLimit) + .mockResolvedValueOnce({ allowed: true } as never) + .mockResolvedValueOnce({ allowed: false } as never); + + const response = await POST(emptyRequest() as never); + + expect(response.status).toBe(429); + // Refused before any provider egress. + expect(resolveProviderForTest).not.toHaveBeenCalled(); + }); + + it("refuses when the day's token budget is already exhausted", async () => { + const { reserveBudget } = await import("@/lib/ai/coach/budget"); + const generateCompletion = vi.fn(); + vi.mocked(resolveProviderForTest).mockResolvedValue({ + type: "admin-key", + generateCompletion, + } as never); + vi.mocked(reserveBudget).mockResolvedValueOnce({ + allowed: false, + reserved: 32, + totalAfter: 200_000, + } as never); + + const response = await POST(emptyRequest() as never); + + expect(response.status).toBe(429); + expect(generateCompletion).not.toHaveBeenCalled(); + }); + + it("records a successful probe on the ledger", async () => { + const { reconcileSpend } = await import("@/lib/ai/coach/budget"); + okProvider(); + + await POST(emptyRequest() as never); + + expect(reconcileSpend).toHaveBeenCalledWith( + "u-1", + 32, + expect.any(Number), + "2026-01-01", + ); + }); +}); diff --git a/src/app/api/ai/test/route.ts b/src/app/api/ai/test/route.ts index 14718f4b7..cd7e8a12b 100644 --- a/src/app/api/ai/test/route.ts +++ b/src/app/api/ai/test/route.ts @@ -10,6 +10,32 @@ import { singleUserTurn } from "@/lib/ai/types"; import { apiSuccess, apiError, safeJson } from "@/lib/api-response"; import { checkRateLimit } from "@/lib/rate-limit"; import { annotate } from "@/lib/logging/context"; +import { + buildDateKey, + reconcileSpend, + reserveBudget, + resolveDailyCap, +} from "@/lib/ai/coach/budget"; + +/** + * Output ceiling of the probe below. Mirrors the `maxTokens` on the completion + * so the ledger reserves what the call can actually spend. + */ +const AI_TEST_MAX_TOKENS = 32; + +/** + * Daily ceiling on connection tests per user. + * + * The 5/min bucket alone bounded burst rate but nothing cumulative: sustained, + * it permits ~7 200 probes/day, and with an empty body this route resolves the + * SAME chain generation uses — which can be the operator's own credential. That + * is unmetered operator spend on a surface that exists to answer "are my + * settings right?". A person checks that a handful of times after editing a + * provider; 50 is far above real use and far below anything that matters on the + * bill. + */ +const AI_TEST_DAILY_LIMIT = 50; +const ONE_DAY_MS = 24 * 60 * 60 * 1000; export const dynamic = "force-dynamic"; @@ -34,6 +60,17 @@ export const POST = apiHandler(async (request: NextRequest) => { const rl = await checkRateLimit(`ai-test:${user.id}`, 5, 60_000); if (!rl.allowed) return apiError("Too many test requests", 429); + // Cumulative ceiling on top of the burst bucket — see AI_TEST_DAILY_LIMIT. + const daily = await checkRateLimit( + `ai-test-daily:${user.id}`, + AI_TEST_DAILY_LIMIT, + ONE_DAY_MS, + ); + if (!daily.allowed) { + annotate({ action: { name: "ai.test.daily_limit" } }); + return apiError("Too many test requests today", 429); + } + // Body is optional. Empty body → behaves like before (test the saved // config). Non-empty body → tests the unsaved selection without // mutating the user row. Plaintext keys never persist. @@ -65,19 +102,47 @@ export const POST = apiHandler(async (request: NextRequest) => { return apiError("No AI provider configured", 422); } + // Meter the probe on the same daily ledger every other AI surface writes to, + // so operator-key spend from this route is visible rather than invisible. + // `admin-key` is the operator's own credential and draws the operator + // ceiling; an override names the caller's own key, and every other resolved + // type is the user's own egress, so those draw the user-plan ceiling. + const dateKey = buildDateKey(); + const reservation = await reserveBudget( + user.id, + AI_TEST_MAX_TOKENS, + dateKey, + resolveDailyCap([ + { + providerType: provider.type === "admin-key" ? "admin-openai" : "local", + }, + ]), + ); + if (!reservation.allowed) { + annotate({ action: { name: "ai.test.budget_exceeded" } }); + return apiError("Daily AI budget exhausted", 429); + } + try { const result = await provider.generateCompletion( singleUserTurn({ system: "You are a connection-test responder.", user: 'Reply with the JSON object {"ok": true} and nothing else.', temperature: 0, - maxTokens: 32, + maxTokens: AI_TEST_MAX_TOKENS, // The probe asks for a JSON object — keep the OpenAI / Codex strict // JSON mode it relied on before the response_format gate landed. responseFormat: "json", }), ); + await reconcileSpend( + user.id, + reservation.reserved, + result.tokensUsed ?? reservation.reserved, + dateKey, + ); + return apiSuccess({ ok: true, providerType: result.providerType, @@ -91,6 +156,10 @@ export const POST = apiHandler(async (request: NextRequest) => { // Log full details server-side for the operator and respond with a // categorised, generic reason. const err = e as Error & { httpStatus?: number; bodyExcerpt?: string }; + // A failed probe produced no reported usage — refund the reservation so a + // provider that is simply misconfigured doesn't burn the caller's ledger + // while they fix it. + await reconcileSpend(user.id, reservation.reserved, 0, dateKey); annotate({ meta: { ai_test_error: err.message.slice(0, 500), diff --git a/src/app/api/insights/chat/__tests__/route-snapshot-once.test.ts b/src/app/api/insights/chat/__tests__/route-snapshot-once.test.ts index 39423d43b..d7f2c4299 100644 --- a/src/app/api/insights/chat/__tests__/route-snapshot-once.test.ts +++ b/src/app/api/insights/chat/__tests__/route-snapshot-once.test.ts @@ -220,3 +220,92 @@ describe("coach chat — snapshot sent once per conversation (C4)", () => { expect(prompt).toContain("Your BP looks stable."); }); }); + +/** + * The erosion guard. + * + * The re-ground condition used to be the open-ended `allTurns.length > + * TURN_CAP`, which is true for EVERY turn past the history cap — not just the + * one that crosses it. So a conversation that ran long re-shipped the full + * ~15k-token snapshot on turn 21, 22, 23, … permanently, reinstating exactly + * the per-turn cost the snapshot-once design removes, and paying it on the + * longest (most expensive) conversations. The intent was always to re-ground + * ONCE, at the boundary where the original snapshot scrolls out of the + * verbatim window. + */ +describe("coach chat — snapshot re-grounds at the elision boundary only", () => { + const TURN_CAP = 20; + + beforeEach(() => { + vi.clearAllMocks(); + runStreamingRawCompletionWithFallback.mockResolvedValue({ + result: { content: "Your BP looks stable.", tokensUsed: 42, model: "m" }, + workingProvider: { providerType: "admin-openai" }, + }); + }); + + /** Drive one turn against a conversation that already has `n` turns on disk. */ + async function turnWithPriorCount(n: number): Promise { + fetchConversationWithMessages.mockResolvedValue({ + id: "c1", + summary: "earlier summary", + messages: Array.from({ length: n }, (_, i) => ({ + role: i % 2 === 0 ? "user" : "assistant", + content: `turn ${i}`, + })), + }); + await postAndDrain({ conversationId: "c1", message: "next?" }); + return lastUserPrompt(); + } + + it("re-grounds on the turn that crosses the cap", async () => { + expect(await turnWithPriorCount(TURN_CAP)).toContain(SNAPSHOT_JSON); + }); + + it("does NOT re-ship the snapshot on every turn past the cap", async () => { + // Well past the boundary — these are the turns that used to pay the full + // snapshot cost forever. + for (const prior of [TURN_CAP + 2, TURN_CAP + 6, TURN_CAP + 20, 58]) { + expect(await turnWithPriorCount(prior)).not.toContain(SNAPSHOT_JSON); + } + }); + + it("ships the full snapshot only twice across a 60-turn conversation", async () => { + let shipped = 0; + const shippedAt: number[] = []; + + // Turn 1 starts a fresh conversation (no prior turns on disk); every later + // turn loads the growing history. A successful turn persists both the user + // and the assistant message, so the count advances by 2. + fetchConversationWithMessages.mockResolvedValue(null); + await postAndDrain({ message: "How is my BP?" }); + if (lastUserPrompt().includes(SNAPSHOT_JSON)) { + shipped++; + shippedAt.push(0); + } + + for (let prior = 2; prior <= 60; prior += 2) { + if ((await turnWithPriorCount(prior)).includes(SNAPSHOT_JSON)) { + shipped++; + shippedAt.push(prior); + } + } + + // Turn 1 (grounding) + the single crossing at the cap. Under the old + // predicate this was 1 + every turn from 20 to 60 — 21 full snapshots. + expect(shippedAt).toEqual([0, TURN_CAP]); + expect(shipped).toBe(2); + }); + + /** + * The turn count does not advance in fixed steps: the user message is + * persisted before the provider call, the assistant message only on success, + * so a turn that lost its reply advances the count by 1 instead of 2. An + * exact `=== TURN_CAP` equality would be stepped straight over by such a + * conversation and would then never re-ground at all. The window must catch + * the crossing from either parity. + */ + it("still re-grounds when a lost reply shifts the turn parity", async () => { + expect(await turnWithPriorCount(TURN_CAP + 1)).toContain(SNAPSHOT_JSON); + }); +}); diff --git a/src/app/api/insights/chat/route.ts b/src/app/api/insights/chat/route.ts index caaeeb43d..59f9f46bf 100644 --- a/src/app/api/insights/chat/route.ts +++ b/src/app/api/insights/chat/route.ts @@ -503,15 +503,33 @@ async function handleChatRequest(request: NextRequest): Promise { // figures stay in-context without re-paying for them. We therefore include // the full block only when: // - this is the first turn (no prior turns on disk), OR - // - the history window has begun eliding turns (`allTurns.length > TURN_CAP`) - // — once the oldest turns are folded into the rolling summary the original - // snapshot may have scrolled out of the verbatim window, so we re-ground. + // - the history window has JUST begun eliding turns — once the oldest turns + // are folded into the rolling summary the original snapshot may have + // scrolled out of the verbatim window, so we re-ground exactly once at + // that boundary. // On the cheap path (a follow-up inside the verbatim window) we send a short // pointer instead of the figures, preserving grounding + cross-metric // correlation quality at a fraction of the wire cost. + // + // The elision test is a narrow WINDOW, not the open-ended `allTurns.length > + // TURN_CAP` it replaced. That predicate is true for every turn past the cap, + // not just the one that crosses it, so a long conversation re-shipped the + // whole ~15k-token snapshot on turn 21, 22, 23 … — reinstating the exact + // per-turn cost this design removed, and paying it on the LONGEST + // conversations. A 60-turn conversation paid it ~40 times instead of twice. + // + // Width 2 rather than an exact `=== TURN_CAP` equality, because the turn + // count does not advance in fixed steps: the user turn is persisted before + // the provider call, the assistant turn only on success, so a failed turn + // advances the count by 1 and a successful one by 2. An exact-equality check + // would be stepped straight over by a conversation that ever lost a reply, + // and would then NEVER re-ground. A width-2 window cannot be stepped over by + // steps of 1 or 2, so the crossing is always caught, and it can match at most + // twice — bounded either way. const isFirstTurn = priorTurns.length === 0; - const historyEliding = allTurns.length > TURN_CAP; - const includeFullSnapshot = isFirstTurn || historyEliding; + const historyElisionCrossing = + priorTurns.length >= TURN_CAP && priorTurns.length <= TURN_CAP + 1; + const includeFullSnapshot = isFirstTurn || historyElisionCrossing; const transcript = window .map((t) => `${t.role.toUpperCase()}: ${t.content}`) .join("\n\n"); diff --git a/src/lib/ai/coach/coach-memory-refresh-worker.ts b/src/lib/ai/coach/coach-memory-refresh-worker.ts index 3867844f8..c303e49f0 100644 --- a/src/lib/ai/coach/coach-memory-refresh-worker.ts +++ b/src/lib/ai/coach/coach-memory-refresh-worker.ts @@ -2,11 +2,18 @@ * v1.11.1 — worker pipeline for the combined Coach memory-refresh queue. * * Runs both background generators for one conversation, sequentially so they - * share the wake-up but each pays its own budget check inside - * `runStatusCompletion`: the rolling conversation summary first, then durable - * fact extraction. Each step is fault-isolated — a failure or no-provider in - * one never sinks the other or the job. Kept out of the route bundle (the - * route imports only `enqueueCoachMemoryRefresh` from `coach-memory-shared`). + * share the wake-up: the rolling conversation summary first, then durable fact + * extraction. Each step is fault-isolated — a failure or no-provider in one + * never sinks the other or the job. Kept out of the route bundle (the route + * imports only `enqueueCoachMemoryRefresh` from `coach-memory-shared`). + * + * Both steps reserve and reconcile against the caller's daily token ledger + * inside `runStatusCompletion`, so this off-request work is metered on the same + * `coach_usage` row as the chat turn that triggered it, against the ceiling + * that matches the chain's cost owner. This comment previously asserted the + * same guarantee while no such accounting existed anywhere on the path — the + * chokepoint did no budget work at all, so every generator behind it, not just + * these two, spent unmetered. */ import { annotate } from "@/lib/logging/context"; diff --git a/src/lib/insights/__tests__/status-provider-budget.test.ts b/src/lib/insights/__tests__/status-provider-budget.test.ts new file mode 100644 index 000000000..9e5bd760d --- /dev/null +++ b/src/lib/insights/__tests__/status-provider-budget.test.ts @@ -0,0 +1,196 @@ +/** + * Cost accounting for the status/reference provider chokepoint. + * + * `runStatusCompletion` is the single provider entry for every status and + * reference generator — the specialised status cards, the generic metric + * cards, biomarker cards, the batched assessment, the derived scores, the + * period narrative, and the off-request Coach memory workers. It previously + * ran with NO budget accounting whatsoever: nothing reserved, nothing + * recorded, no ceiling. Everything behind it spent invisibly, and on an + * operator-key account that was unmetered operator spend. + * + * These tests drive the REAL budget module (`reserveBudget` / `reconcileSpend` + * / `resolveDailyCap`) against a stateful in-memory stand-in for the + * `coach_usage` row, so they exercise the actual atomic-upsert path rather + * than merely asserting that a mock was called. + */ +import { describe, it, expect, vi, beforeEach } from "vitest"; + +/** In-memory stand-in for the day's `coach_usage.total_tokens`. */ +let ledgerTotal = 0; + +vi.mock("@/lib/db", () => ({ + prisma: { + // `reserveBudget`'s atomic upsert-increment. Tagged-template call shape: + // (strings, userId, dateKey, reserved, reserved). + $queryRaw: vi.fn( + async (_strings: TemplateStringsArray, ...v: unknown[]) => { + ledgerTotal += Number(v[2] ?? 0); + return [{ total_tokens: ledgerTotal }]; + }, + ), + // `reconcileSpend` (+delta) and `refundReservation` (-reserved). + $executeRaw: vi.fn( + async (strings: TemplateStringsArray, ...v: unknown[]) => { + const sql = strings.join("?"); + const amount = Number(v[0] ?? 0); + if (sql.includes("total_tokens + ")) ledgerTotal += amount; + else if (sql.includes("total_tokens - ")) ledgerTotal -= amount; + ledgerTotal = Math.max(0, ledgerTotal); + return 1; + }, + ), + user: { + findUnique: vi.fn(async () => ({ aiResponseTimeoutSeconds: null })), + }, + }, +})); + +vi.mock("@/lib/logging/context", () => ({ annotate: vi.fn() })); + +const { resolveProviderChain, resolveProvider } = vi.hoisted(() => ({ + resolveProviderChain: vi.fn(), + resolveProvider: vi.fn(), +})); +vi.mock("@/lib/ai/provider", () => ({ resolveProviderChain, resolveProvider })); + +const { runRawCompletionWithFallback } = vi.hoisted(() => ({ + runRawCompletionWithFallback: vi.fn(), +})); +vi.mock("@/lib/ai/provider-runner", () => ({ + AllProvidersFailedError: class extends Error {}, + runRawCompletionWithFallback, +})); + +// Consent gate is exercised by its own suite; keep it open here so the budget +// behaviour is what these tests isolate. +vi.mock("@/lib/ai/consent-guard", () => ({ + chainRequiresServerManagedConsent: vi.fn(() => false), + hasActiveConsentForSurface: vi.fn(async () => true), +})); + +import { runStatusCompletion } from "../status-provider"; +import { annotate } from "@/lib/logging/context"; +import { OPERATOR_COST_CAP, USER_PLAN_CAP } from "@/lib/ai/coach/budget"; + +const OPERATOR_CHAIN = [{ providerType: "admin-openai", instance: {} }]; +const BYOK_CHAIN = [{ providerType: "anthropic", instance: {} }]; + +function completionArgs(overrides: Record = {}) { + return { + userId: "u1", + cacheAction: "status:test", + systemPrompt: "system", + userPrompt: "user", + consentSurface: "insights" as const, + ...overrides, + }; +} + +/** A successful provider reply reporting `tokensUsed`. */ +function mockProviderReply(tokensUsed: number | null) { + runRawCompletionWithFallback.mockResolvedValue({ + result: { content: '{"summary":"ok"}', model: "m", tokensUsed }, + workingProvider: { providerType: "admin-openai" }, + }); +} + +beforeEach(() => { + vi.clearAllMocks(); + ledgerTotal = 0; + resolveProviderChain.mockResolvedValue(OPERATOR_CHAIN); + resolveProvider.mockResolvedValue({ type: "none" }); +}); + +describe("runStatusCompletion — ledger accounting", () => { + it("lands a successful generation on the day's ledger", async () => { + mockProviderReply(1234); + + const result = await runStatusCompletion(completionArgs()); + + expect(result.kind).toBe("ok"); + // Reserved an estimate, then reconciled down to what the provider + // actually reported. Either way the spend is now ON the ledger — before + // this wiring existed the total stayed at zero forever. + expect(ledgerTotal).toBe(1234); + }); + + it("charges the reservation when the provider reports no token count", async () => { + mockProviderReply(null); + + await runStatusCompletion(completionArgs()); + + // An unreported generation must not bill as free. + expect(ledgerTotal).toBeGreaterThan(0); + }); + + it("refuses a generation once the day's cap is already spent", async () => { + ledgerTotal = OPERATOR_COST_CAP; + mockProviderReply(500); + + const result = await runStatusCompletion(completionArgs()); + + // Refused BEFORE any provider egress, and reported as a transient miss + // (`error`) rather than `none` — callers cache `none` as the settled + // "no provider configured" assessment, which a budget refusal is not. + expect(runRawCompletionWithFallback).not.toHaveBeenCalled(); + expect(result.kind).toBe("error"); + expect(annotate).toHaveBeenCalledWith( + expect.objectContaining({ + action: { name: "insights.status.budget_exceeded" }, + }), + ); + // The refused reservation was refunded, not left on the row. + expect(ledgerTotal).toBe(OPERATOR_COST_CAP); + }); + + it("refunds the reservation when the provider errors", async () => { + runRawCompletionWithFallback.mockRejectedValue(new Error("upstream down")); + + const result = await runStatusCompletion(completionArgs()); + + expect(result.kind).toBe("error"); + expect(ledgerTotal).toBe(0); + }); +}); + +describe("runStatusCompletion — cost owner decides the ceiling", () => { + it("measures an operator-key chain against the operator ceiling", async () => { + // Just under the operator ceiling: an operator-key user is refused here. + ledgerTotal = OPERATOR_COST_CAP; + mockProviderReply(100); + + const result = await runStatusCompletion(completionArgs()); + + expect(result.kind).toBe("error"); + expect(runRawCompletionWithFallback).not.toHaveBeenCalled(); + }); + + it("measures a BYOK chain against the user-plan ceiling, not the operator's", async () => { + // Same spend that locked the operator-key user out above. A self-hoster on + // their own key pays their own bill, so the operator's ceiling is a + // category error for them — they must still be served. + resolveProviderChain.mockResolvedValue(BYOK_CHAIN); + ledgerTotal = OPERATOR_COST_CAP; + runRawCompletionWithFallback.mockResolvedValue({ + result: { content: '{"summary":"ok"}', model: "m", tokensUsed: 100 }, + workingProvider: { providerType: "anthropic" }, + }); + + const result = await runStatusCompletion(completionArgs()); + + expect(result.kind).toBe("ok"); + expect(runRawCompletionWithFallback).toHaveBeenCalled(); + }); + + it("still bounds a BYOK chain at the user-plan ceiling", async () => { + resolveProviderChain.mockResolvedValue(BYOK_CHAIN); + ledgerTotal = USER_PLAN_CAP; + mockProviderReply(100); + + const result = await runStatusCompletion(completionArgs()); + + expect(result.kind).toBe("error"); + expect(runRawCompletionWithFallback).not.toHaveBeenCalled(); + }); +}); diff --git a/src/lib/insights/status-provider.ts b/src/lib/insights/status-provider.ts index 107a48466..8f36e8a1d 100644 --- a/src/lib/insights/status-provider.ts +++ b/src/lib/insights/status-provider.ts @@ -10,6 +10,12 @@ import { type ConsentSurface, } from "@/lib/ai/consent-guard"; import { annotate } from "@/lib/logging/context"; +import { + buildDateKey, + reconcileSpend, + reserveBudget, + resolveDailyCap, +} from "@/lib/ai/coach/budget"; import { AI_BUDGETS, REFERENCE_AI_SEED } from "@/lib/ai/ai-budgets"; import { singleUserTurn } from "@/lib/ai/types"; import { STATUS_PROVIDER_TIMEOUT_MS, withTimeout } from "./with-timeout"; @@ -189,6 +195,49 @@ export async function runStatusCompletion( STATUS_PROVIDER_TIMEOUT_MS, ); + const maxTokens = args.maxTokens ?? AI_BUDGETS.status.maxTokens; + + // The day's token ledger. Until now this chokepoint — the provider entry for + // EVERY status/reference family (the specialised cards, the generic metric + // cards, biomarker cards, the batched assessment, the derived scores, the + // period narrative, and the off-request Coach memory workers: rolling + // summary, fact extraction, plan proposals) — ran with no accounting at all, + // so none of that spend appeared in `coach_usage` and no ceiling applied. + // + // The reservation is atomic (single upsert-increment), matching the Coach and + // document paths: a read-then-write check would let concurrent generations + // each observe a sub-cap total and all proceed. We reserve an ESTIMATE up + // front — the output ceiling plus a ~4-chars-per-token approximation of the + // prompt we are about to send — and reconcile against the provider's reported + // count afterwards, refunding in full when nothing was generated. + // + // The cap follows the COST OWNER, not the surface: `resolveDailyCap` charges + // the operator ceiling only when the chain's primary is the operator's own + // credential (`admin-openai` / `admin-codex`). A self-hoster on their own key + // or a local model is measured against the generous user-plan ceiling, so + // their own hardware/plan is never rationed by the operator's bill. + const dateKey = buildDateKey(); + const estimatedTokens = + maxTokens + Math.ceil((systemPrompt.length + userPrompt.length) / 4); + const reservation = await reserveBudget( + userId, + estimatedTokens, + dateKey, + resolveDailyCap(chain), + ); + if (!reservation.allowed) { + // Over the day's ceiling. Reported as `error` — a TRANSIENT miss the caller + // serves the fallback for without persisting it — deliberately NOT `none`, + // which callers cache as the settled "no provider configured" assessment. + // The distinct annotation keeps the refusal observable even though the + // result shape is shared. + annotate({ + action: { name: "insights.status.budget_exceeded" }, + meta: { cacheAction, totalAfter: reservation.totalAfter }, + }); + return { kind: "error" }; + } + const raced = await withTimeout( () => runRawCompletionWithFallback({ @@ -198,7 +247,7 @@ export async function runStatusCompletion( system: systemPrompt, user: userPrompt, temperature: args.temperature ?? AI_BUDGETS.status.temperature, - maxTokens: args.maxTokens ?? AI_BUDGETS.status.maxTokens, + maxTokens, // v1.18.7 — status/reference output is reproducible: pin the // deterministic seed unless a caller overrides it. seed: args.seed ?? REFERENCE_AI_SEED, @@ -213,13 +262,26 @@ export async function runStatusCompletion( ); if (raced.timedOut) { + // A timed-out generation may still have burned upstream tokens, but we have + // no reported count to charge — refund the reservation rather than bill an + // invented figure. + await reconcileSpend(userId, reservation.reserved, 0, dateKey); return { kind: "timeout" }; } if (raced.errored || raced.value === null) { + await reconcileSpend(userId, reservation.reserved, 0, dateKey); return { kind: "error" }; } const { result, workingProvider } = raced.value; + // Reconcile against what the provider actually reported. This runs for the + // empty-content branch too: those tokens were burned upstream even though the + // reply was unusable, so they stay on the ledger rather than being refunded + // into a free retry loop. Falls back to the reservation when the provider + // reports no count, so an unreported generation is never billed as zero. + const actualTokens = result.tokensUsed ?? reservation.reserved; + await reconcileSpend(userId, reservation.reserved, actualTokens, dateKey); + const content = result.content; if (typeof content !== "string" || content.trim().length === 0) { annotate({ diff --git a/src/lib/jobs/__tests__/insight-status-generate.test.ts b/src/lib/jobs/__tests__/insight-status-generate.test.ts index f536ec70c..fe4a9b116 100644 --- a/src/lib/jobs/__tests__/insight-status-generate.test.ts +++ b/src/lib/jobs/__tests__/insight-status-generate.test.ts @@ -108,7 +108,7 @@ describe("enqueueStatusGeneration", () => { ).resolves.toBeUndefined(); }); - it("sends with a per-(user,metric,locale) singletonKey when boss exists", async () => { + it("sends with a per-(user,metric) singletonKey when boss exists", async () => { const send = vi.fn().mockResolvedValue("job-id"); getGlobalBoss.mockReturnValue({ send }); await enqueueStatusGeneration({ @@ -119,10 +119,34 @@ describe("enqueueStatusGeneration", () => { expect(send).toHaveBeenCalledWith( INSIGHT_STATUS_GENERATE_QUEUE, { userId: "u1", metric: "mood", locale: "en" }, - expect.objectContaining({ singletonKey: "u1:mood:en" }), + expect.objectContaining({ singletonKey: "u1:mood" }), ); }); + /** + * Cost guard. The locale on a status read is client-chosen (`?locale=`, or + * the `healthlog-locale` cookie), so when it was part of the singleton key + * each of the six supported locales opened its own de-dupe slot and one + * legitimate generation pass could be multiplied into six. The key must + * collapse them: same user, same metric, one queued generation regardless of + * how many locales ask. + */ + it("collapses every locale for one metric onto a single singletonKey", async () => { + const send = vi.fn().mockResolvedValue("job-id"); + getGlobalBoss.mockReturnValue({ send }); + + for (const locale of ["de", "en", "fr", "es", "it", "pl"] as const) { + await enqueueStatusGeneration({ userId: "u1", metric: "mood", locale }); + } + + const keys = new Set( + send.mock.calls.map( + (call) => (call[2] as { singletonKey: string }).singletonKey, + ), + ); + expect(keys).toEqual(new Set(["u1:mood"])); + }); + it("swallows a boss.send rejection (best-effort enqueue)", async () => { const send = vi.fn().mockRejectedValue(new Error("boom")); getGlobalBoss.mockReturnValue({ send }); diff --git a/src/lib/jobs/insight-status-generate-shared.ts b/src/lib/jobs/insight-status-generate-shared.ts index b1b4ae5e5..0228bcf4f 100644 --- a/src/lib/jobs/insight-status-generate-shared.ts +++ b/src/lib/jobs/insight-status-generate-shared.ts @@ -95,12 +95,29 @@ export interface InsightStatusGeneratePayload { /** * Fire-and-forget enqueue used by the read-only status route on a cache - * miss. A `singletonKey` per `(user, metric, locale)` collapses repeated - * polls into one queued job, so a client polling every few seconds while - * the provider works can't pile up duplicate generations. No-ops cleanly - * when the global boss instance is not available (e.g. the web process - * runs without an embedded worker) — the nightly pre-generate cron and - * the daily status crons remain the catch-net. + * miss. A `singletonKey` per `(user, metric)` collapses repeated polls into + * one queued job, so a client polling every few seconds while the provider + * works can't pile up duplicate generations. No-ops cleanly when the global + * boss instance is not available (e.g. the web process runs without an + * embedded worker) — the nightly pre-generate cron and the daily status + * crons remain the catch-net. + * + * The key deliberately EXCLUDES `locale`. It used to be + * `(user, metric, locale)`, and the locale on a status read is client-chosen + * — the `?locale=` query param, or the `healthlog-locale` cookie, both of + * which a caller sets freely. Each distinct locale therefore opened its own + * singleton slot, so walking the metric registry across the six supported + * locales multiplied one legitimate generation pass into six, none of which + * the de-dupe could collapse. Keying on `(user, metric)` makes the locale a + * property of the queued payload instead of part of its identity: one + * generation per metric per window, whichever locale asked first. + * + * A legitimate language switch still works. The cache stays keyed by + * `(user, metric, locale)`, so the new locale is a genuine cache miss and + * enqueues normally; the only change is that a switch made WHILE a generation + * for the same metric is already in flight waits out the 120 s window instead + * of racing a second one. The next poll re-enqueues and the card converges to + * the requested language. */ export async function enqueueStatusGeneration( payload: InsightStatusGeneratePayload, @@ -115,7 +132,7 @@ export async function enqueueStatusGeneration( } try { await boss.send(INSIGHT_STATUS_GENERATE_QUEUE, payload, { - singletonKey: `${payload.userId}:${payload.metric}:${payload.locale}`, + singletonKey: `${payload.userId}:${payload.metric}`, // De-dupe within a short window so a polling client doesn't enqueue // a fresh job on every tick while a generation is already running. singletonSeconds: 120,