Merge branch 'worktree-agent-acf4680fc308d77c8' into release/v1.30.25

This commit is contained in:
Marc-André Bombeck committed 2026-07-19 04:24:13 +02:00
commit b1370626fc
9 files changed
+583 -21

No files matched your search

@@ -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",
);
});
});
+70 -1
View File
@@ -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),
@@ -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<string> {
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);
});
});
+23 -5
View File
@@ -503,15 +503,33 @@ async function handleChatRequest(request: NextRequest): Promise<Response> {
// 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");
@@ -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";
@@ -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<string, unknown> = {}) {
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();
});
});
+63 -1
View File
@@ -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({
@@ -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 });
+24 -7
View File
@@ -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,