diff --git a/prisma/migrations/0341_arrival_reaction_cost_owner/migration.sql b/prisma/migrations/0341_arrival_reaction_cost_owner/migration.sql new file mode 100644 index 000000000..74fee8a02 --- /dev/null +++ b/prisma/migrations/0341_arrival_reaction_cost_owner/migration.sql @@ -0,0 +1,18 @@ +-- v1.38.19 (Wave E, fix round 1) — the arrival reaction remembers who paid for +-- its reservation. +-- +-- The row persisted the reserved amount and the date key but not the cost +-- owner, and two readers need it. The supersede refund in `data-arrival.ts` +-- reverses a pre-provider reservation with no chain in hand at all, so without +-- this column it could only reverse `total_tokens` — leaving the operator +-- share of every superseded reservation on the row forever, walking the +-- operator's ceiling upward against spend that never happened. The resume path +-- in `reaction-line.ts` re-derived the owner from the chain as it stands NOW, +-- and the chain is re-resolved per run and reordered by the health ledger, so +-- a reservation booked to the operator could be reconciled as the user's (the +-- same leak) or vice versa (draining another request's balance). +-- +-- Nullable, no default: an existing row's owner is genuinely unknown, and the +-- two readers fall back to the previous behaviour for those. +ALTER TABLE "arrival_reactions" + ADD COLUMN IF NOT EXISTS "generation_cost_owner" TEXT; diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 688951392..0a2fe73c0 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -6761,6 +6761,12 @@ model ArrivalReaction { /// `coach_usage` increment. A reclaimed pre-provider attempt reuses it. generationReservedTokens Int? @map("generation_reserved_tokens") generationBudgetDateKey String? @map("generation_budget_date_key") + /// v1.38.19 (Wave E) — who the reservation above was booked to, `"operator"` + /// or `"user"`. Persisted rather than recomputed: the supersede refund in + /// `data-arrival.ts` has no chain to ask, and the resume path's chain may + /// have been reordered by the health ledger since the reservation was made. + /// Null on rows written before the column existed. + generationCostOwner String? @map("generation_cost_owner") /// Durable provider boundary. Once set, this marker is terminal unless the /// encrypted line is committed; no retry may invoke a provider again. generationProviderInvokedAt DateTime? @map("generation_provider_invoked_at") diff --git a/src/__tests__/coach-usage-owner-columns-guard.test.ts b/src/__tests__/coach-usage-owner-columns-guard.test.ts new file mode 100644 index 000000000..bda895bfe --- /dev/null +++ b/src/__tests__/coach-usage-owner-columns-guard.test.ts @@ -0,0 +1,132 @@ +/** + * Structural guard — `coach_usage.total_tokens` and `coach_usage.operator_tokens` + * move together in EVERY raw statement. + * + * v1.38.19 (Wave E) gave the daily ledger a second counter: `operator_tokens` + * is the share of `total_tokens` an operator-funded provider served, and the + * operator ceiling is enforced against it. The counters are only meaningful + * while they stay in step. A statement that writes the total and forgets the + * operator share under-counts the operator's exposure; a statement that + * reverses the total and forgets the operator share leaks budget permanently + * toward the operator cap, because nothing else ever subtracts it; a statement + * that RETURNS the total alone forces its caller to gate on the wrong number, + * which is the exact defect the wave exists to fix. + * + * Two of the three writers shipped with precisely those bugs, and both were a + * grep away. This guard is that grep, executable: every INSERT column list, + * every SET clause and every RETURNING list that names one counter must name + * the other. It also pins the writer set, so a fourth place that learns to + * write the ledger has to come past this file first. + */ +import { readdirSync, readFileSync } from "node:fs"; +import { fileURLToPath } from "node:url"; +import { dirname, relative, resolve } from "node:path"; + +import { describe, it, expect } from "vitest"; + +const SRC_DIR = resolve(dirname(fileURLToPath(import.meta.url)), ".."); +const REPO_ROOT = resolve(SRC_DIR, ".."); + +/** + * The production files allowed to hold raw `coach_usage` SQL. The ledger's + * invariants (two counters in one statement, one row per user and UTC day) are + * reasoned about here and nowhere else; a new entry means a new place those + * invariants can be broken, so it is a deliberate edit, not a side effect. + */ +const EXPECTED_SQL_WRITERS = [ + "src/lib/ai/coach/budget.ts", + "src/lib/jobs/data-arrival.ts", +]; + +function collectSourceFiles(dir: string, out: string[] = []): string[] { + for (const entry of readdirSync(dir, { withFileTypes: true })) { + const full = resolve(dir, entry.name); + if (entry.isDirectory()) { + if (entry.name === "__tests__" || entry.name === "generated") continue; + collectSourceFiles(full, out); + continue; + } + if (entry.name.endsWith(".ts") || entry.name.endsWith(".tsx")) { + if (entry.name.endsWith(".test.ts") || entry.name.endsWith(".test.tsx")) { + continue; + } + out.push(full); + } + } + return out; +} + +/** + * Every backtick-delimited SQL template literal in a source file. A prose + * mention of `coach_usage` in a doc comment sits between backticks too, so a + * literal only counts once it also reads like a statement. + */ +const SQL_STATEMENT = /\b(INSERT\s+INTO|UPDATE|DELETE\s+FROM|SELECT)\b/i; + +function sqlLiterals(source: string): string[] { + return (source.match(/`[^`]*`/g) ?? []).filter( + (literal) => literal.includes("coach_usage") && SQL_STATEMENT.test(literal), + ); +} + +/** + * The clauses of one SQL statement that name columns: the INSERT column list, + * every SET assignment block, and every RETURNING list. Each is checked on its + * own, because a statement can write both counters and still return only one — + * which is how the arrival-reaction reservation ended up gating the operator's + * cap on a mixed total. + */ +function columnClauses(sql: string): { label: string; text: string }[] { + const clauses: { label: string; text: string }[] = []; + for (const m of sql.matchAll(/INSERT\s+INTO\s+[\w"]+\s*\(([^)]*)\)/gi)) { + clauses.push({ label: "INSERT column list", text: m[1] }); + } + for (const m of sql.matchAll( + /\bSET\b([\s\S]*?)(?=\bWHERE\b|\bRETURNING\b|$)/gi, + )) { + clauses.push({ label: "SET clause", text: m[1] }); + } + for (const m of sql.matchAll(/\bRETURNING\b([\s\S]*?)(?=;|$)/gi)) { + clauses.push({ label: "RETURNING list", text: m[1] }); + } + return clauses; +} + +describe("coach_usage — the two counters move together", () => { + const files = collectSourceFiles(SRC_DIR); + + it("names operator_tokens in every clause that names total_tokens", () => { + const offences: string[] = []; + for (const file of files) { + const source = readFileSync(file, "utf8"); + if (!source.includes("coach_usage")) continue; + for (const literal of sqlLiterals(source)) { + for (const clause of columnClauses(literal)) { + if (!clause.text.includes("total_tokens")) continue; + if (clause.text.includes("operator_tokens")) continue; + offences.push( + `${relative(REPO_ROOT, file)} — ${clause.label}: ${clause.text + .replace(/\s+/g, " ") + .trim()}`, + ); + } + } + } + + expect( + offences, + `Every raw coach_usage clause that touches total_tokens must touch operator_tokens too:\n${offences.join("\n")}`, + ).toEqual([]); + }); + + it("keeps the raw-SQL writer set to the files that reason about the ledger", () => { + const writers = files + .filter((file) => { + return sqlLiterals(readFileSync(file, "utf8")).length > 0; + }) + .map((file) => relative(REPO_ROOT, file)) + .sort(); + + expect(writers).toEqual([...EXPECTED_SQL_WRITERS].sort()); + }); +}); diff --git a/src/lib/jobs/__tests__/reaction-line-degradation.test.ts b/src/lib/jobs/__tests__/reaction-line-degradation.test.ts index 8e2ff0ac5..53b3d4ac6 100644 --- a/src/lib/jobs/__tests__/reaction-line-degradation.test.ts +++ b/src/lib/jobs/__tests__/reaction-line-degradation.test.ts @@ -88,7 +88,8 @@ vi.mock("@/lib/ai/coach/budget", () => ({ reserveBudget: (...a: unknown[]) => reserveBudget(...a), reconcileSpend: (...a: unknown[]) => reconcileSpend(...a), resolveDailyCap: () => 200_000, - resolveDailyCapFor: () => 200_000, + // A background surface on an operator-funded chain: half the ceiling. + resolveDailyCapFor: () => 100_000, resolveCostOwner: () => "operator" as const, })); @@ -175,7 +176,7 @@ function expectSurfaceStillWorks() { expect(digest.line.length).toBeGreaterThan(0); } -beforeEach(() => { +beforeEach(async () => { vi.clearAllMocks(); findUnique.mockResolvedValue({ id: "r1", @@ -193,16 +194,22 @@ beforeEach(() => { measurementFindMany.mockResolvedValue([]); workoutFindFirst.mockResolvedValue(null); labResultFindMany.mockResolvedValue([]); - queryRaw.mockResolvedValue([{ total_tokens: 1_400 }]); + queryRaw.mockResolvedValue([{ total_tokens: 1_400, operator_tokens: 1_400 }]); executeRaw.mockResolvedValue(1); isModuleEnabled.mockResolvedValue(true); chainRequiresServerManagedConsent.mockReturnValue(false); hasActiveConsentForSurface.mockResolvedValue(true); - reserveBudget.mockResolvedValue({ - allowed: true, - reserved: 1_400, - totalAfter: 1_400, - }); + // v1.38.19 (Wave E, fix round 1) — the claim reservation is no longer a + // second copy of the upsert; it calls the shared `reserveBudget` with the + // transaction client. The spy therefore runs the REAL gate against the + // mocked `$queryRaw`, so the rows a test hands back are the ledger the gate + // reads — which is the only way this file can pin which counter it compares. + const actualBudget = await vi.importActual< + typeof import("@/lib/ai/coach/budget") + >("@/lib/ai/coach/budget"); + reserveBudget.mockImplementation( + actualBudget.reserveBudget as unknown as (...a: unknown[]) => unknown, + ); reconcileSpend.mockResolvedValue(undefined); loadDailyDigest.mockResolvedValue({ score: { value: 82, band: "good", delta: 3 }, @@ -504,7 +511,11 @@ describe("reaction line — degradation", () => { resolveProviderChain.mockResolvedValue([ { providerType: "openai", instance: {} }, ]); - queryRaw.mockResolvedValue([{ total_tokens: 201_400 }]); + // The OPERATOR-funded share of the day is spent — the counter the job's + // ceiling is about. + queryRaw.mockResolvedValue([ + { total_tokens: 101_400, operator_tokens: 101_400 }, + ]); const outcome = await runReactionLine(JOB); @@ -513,6 +524,44 @@ describe("reaction line — degradation", () => { expectSurfaceStillWorks(); }); + it("a day the user's own plan paid for still admits the reaction", async () => { + // The 2026-09-11 shape, on the surface the wave forgot to convert: the + // operator's shared key answered 500 all morning and `codex` — his own + // ChatGPT plan — served everything, so the day's total is large and the + // operator's counter is empty. The reservation used to compare that total + // against the operator's ceiling and refuse, which is the exact defect the + // wave exists to remove, reproduced at HALF the old ceiling. + resolveProviderChain.mockResolvedValue([ + { + providerType: "openai", + instance: { + generateCompletion: vi.fn().mockResolvedValue({ + content: "Logged and steady.", + tokensUsed: 90, + }), + }, + }, + ]); + queryRaw.mockResolvedValue([ + { total_tokens: 901_400, operator_tokens: 1_400 }, + ]); + + const outcome = await runReactionLine(JOB); + + expect(outcome).not.toMatchObject({ reason: "budget_exceeded" }); + // One implementation, not two: the claim transaction reserves through the + // shared gate, with the job ceiling and the transaction client. + expect(reserveBudget).toHaveBeenCalledWith( + "u1", + 1_400, + "2026-07-16", + 100_000, + "operator", + "job", + expect.anything(), + ); + }); + it("provider invocation is terminal even when the provider throws", async () => { const generateCompletion = vi.fn().mockRejectedValue(new Error("timeout")); resolveProviderChain.mockResolvedValue([ @@ -693,3 +742,104 @@ describe("reaction line — degradation", () => { ); }); }); + +/** + * v1.38.19 (Wave E, fix round 1) — a reservation is never stranded. + * + * Every path that crosses the provider boundary has to settle the ledger, and + * settle it against the owner the reservation was actually booked to. Nothing + * outside `data-arrival.ts` reverses an arrival reservation, and there is no + * retention job on `coach_usage`, so a path that forgets leaves those tokens + * counted until the UTC day rolls — and under an operator-funded chain it + * leaves them counted against the operator's ceiling. + * + * The supersede path (a newer arrival replacing a pre-provider reservation) is + * the fourth exit and lives in the database, so it is pinned in + * `tests/integration/arrival-reaction-claim.test.ts` instead. + */ +describe("reaction line — every exit settles the reservation", () => { + const settled = { + servedBy: expect.any(String), + reservedOwner: "operator", + }; + + it("settles on a provider error", async () => { + resolveProviderChain.mockResolvedValue([ + { + providerType: "openai", + instance: { + generateCompletion: vi.fn().mockRejectedValue(new Error("boom")), + }, + }, + ]); + + const outcome = await runReactionLine(JOB); + + expect(outcome).toEqual({ status: "skipped", reason: "provider_failed" }); + expect(reconcileSpend).toHaveBeenCalledTimes(1); + expect(reconcileSpend).toHaveBeenCalledWith( + "u1", + 1_400, + 1_400, + "2026-07-16", + 0, + settled, + ); + }); + + it("settles on a timeout", async () => { + const timeout = Object.assign(new Error("The operation was aborted"), { + name: "TimeoutError", + }); + resolveProviderChain.mockResolvedValue([ + { + providerType: "openai", + instance: { + generateCompletion: vi.fn().mockRejectedValue(timeout), + }, + }, + ]); + + const outcome = await runReactionLine(JOB); + + expect(outcome).toEqual({ status: "skipped", reason: "provider_failed" }); + // A timed-out call may still have burned the tokens upstream, so the + // conservative reservation is CHARGED rather than refunded — but it is + // settled either way, and against the right owner. + expect(reconcileSpend).toHaveBeenCalledWith( + "u1", + 1_400, + 1_400, + "2026-07-16", + 0, + settled, + ); + }); + + it("settles on an empty reply", async () => { + resolveProviderChain.mockResolvedValue([ + { + providerType: "openai", + instance: { + generateCompletion: vi + .fn() + .mockResolvedValue({ content: " ", tokensUsed: 640 }), + }, + }, + ]); + + const outcome = await runReactionLine(JOB); + + expect(outcome).toEqual({ status: "skipped", reason: "unusable_output" }); + // Burned upstream even though the reply was unusable: recorded, not + // refunded into a free retry loop. + expect(reconcileSpend).toHaveBeenCalledWith( + "u1", + 1_400, + 640, + "2026-07-16", + 0, + settled, + ); + }); +}); diff --git a/src/lib/jobs/data-arrival.ts b/src/lib/jobs/data-arrival.ts index cb4318ac6..6c427ef33 100644 --- a/src/lib/jobs/data-arrival.ts +++ b/src/lib/jobs/data-arrival.ts @@ -61,6 +61,13 @@ interface LockedReaction { occurredAt: Date; generationReservedTokens: number | null; generationBudgetDateKey: string | null; + /** + * v1.38.19 (Wave E) — `"operator"` when the reservation below was booked to + * the operator's counter, `"user"` otherwise, null on a row reserved before + * the column existed. The refund has to reverse the SAME counters the + * reservation moved, and there is no chain here to ask. + */ + generationCostOwner: string | null; generationProviderInvokedAt: Date | null; } @@ -100,6 +107,7 @@ async function claimReaction( occurred_at AS "occurredAt", generation_reserved_tokens AS "generationReservedTokens", generation_budget_date_key AS "generationBudgetDateKey", + generation_cost_owner AS "generationCostOwner", generation_provider_invoked_at AS "generationProviderInvokedAt" FROM arrival_reactions WHERE user_id = ${arrival.userId} @@ -117,12 +125,25 @@ async function claimReaction( current.generationBudgetDateKey != null && current.generationProviderInvokedAt == null ) { + // v1.38.19 (Wave E, fix round 1) — reverse BOTH counters, or neither. + // + // The reservation booked `operator_tokens` too whenever the chain's + // primary was operator-funded. Refunding only the total left that share + // stranded: nothing else in the codebase ever subtracts it, so every + // supersede — a routine event, any second measurement of the same kind on + // the same local date before the worker runs — walked the operator's + // ceiling upward against spend that never happened. + const operatorReserved = + current.generationCostOwner === "operator" + ? current.generationReservedTokens + : 0; await tx.$executeRaw` UPDATE coach_usage SET total_tokens = GREATEST( 0, total_tokens - ${current.generationReservedTokens} ), + operator_tokens = GREATEST(0, operator_tokens - ${operatorReserved}), message_count = GREATEST(0, message_count - 1), updated_at = NOW() WHERE user_id = ${arrival.userId} @@ -142,6 +163,7 @@ async function claimReaction( generationClaimedAt: null, generationReservedTokens: null, generationBudgetDateKey: null, + generationCostOwner: null, generationProviderInvokedAt: null, }, }); diff --git a/src/lib/jobs/reaction-line.ts b/src/lib/jobs/reaction-line.ts index 2d49c25e2..21dbb9bc4 100644 --- a/src/lib/jobs/reaction-line.ts +++ b/src/lib/jobs/reaction-line.ts @@ -41,6 +41,7 @@ import { isModuleEnabled } from "@/lib/modules/gate"; import { buildDateKey, reconcileSpend, + reserveBudget, resolveCostOwner, resolveDailyCapFor, type BudgetCostOwner, @@ -317,10 +318,11 @@ type ReactionReservation = { reserved: number; dateKey: string; /** - * v1.38.19 (Wave E) — the cost owner the reservation was booked to. A - * RESUMED reservation (the claim survived a crash, its tokens are on the - * row) recomputes it from the same chain the reservation was made under, so - * the reconcile reverses exactly what the reservation added. + * v1.38.19 (Wave E) — the cost owner the reservation was booked to. It is + * PERSISTED on the row (`generationCostOwner`), not recomputed: the chain is + * re-resolved per run and the health ledger reorders it, so a resumed + * reservation that asked the chain again could reverse a counter the + * reservation never touched — or leave one it did. */ owner: BudgetCostOwner; }; @@ -361,32 +363,28 @@ async function reserveClaimBudget( }); if (stillOwned.count !== 1) throw new ReactionClaimLostError(); - const reserved = ARRIVAL_REACTION_RESERVE_TOKENS; - // v1.38.19 (Wave E) — this reservation is the claim-linked twin of - // `reserveBudget`, so it books the owner split the same way: an - // operator-funded chain increments `operator_tokens` inside the SAME - // statement as the total, and the refusal path reverses both. - const operatorReserved = owner === "operator" ? reserved : 0; - const rows = await tx.$queryRaw<{ total_tokens: number }[]>` - INSERT INTO coach_usage (id, user_id, date_key, total_tokens, operator_tokens, message_count, created_at, updated_at) - VALUES (gen_random_uuid()::text, ${userId}, ${dateKey}, ${reserved}, ${operatorReserved}, 1, NOW(), NOW()) - ON CONFLICT (user_id, date_key) DO UPDATE SET - total_tokens = coach_usage.total_tokens + ${reserved}, - operator_tokens = coach_usage.operator_tokens + ${operatorReserved}, - message_count = coach_usage.message_count + 1, - updated_at = NOW() - RETURNING total_tokens - `; - const totalAfter = Number(rows[0]?.total_tokens ?? reserved); - if (totalAfter - reserved >= cap) { - await tx.$executeRaw` - UPDATE coach_usage - SET total_tokens = GREATEST(0, total_tokens - ${reserved}), - operator_tokens = GREATEST(0, operator_tokens - ${operatorReserved}), - message_count = GREATEST(0, message_count - 1), - updated_at = NOW() - WHERE user_id = ${userId} AND date_key = ${dateKey} - `; + // v1.38.19 (Wave E, fix round 1) — ONE reservation implementation. + // + // This surface used to carry its own copy of the upsert so the write could + // ride the claim transaction. The copy drifted: it learned to WRITE + // `operator_tokens` and kept gating on the day's mixed total, at the + // halved job ceiling — so on the operator's own 2026-09-11 row (1.2 M + // total, ~0 operator) the arrival reaction was refused all day, which is + // the exact defect this wave exists to remove, reproduced on the one + // surface the wave forgot to convert. `reserveBudget` takes the + // transaction client instead, so both writes still roll back together and + // there is no second copy left to drift. + const reservation = await reserveBudget( + userId, + ARRIVAL_REACTION_RESERVE_TOKENS, + dateKey, + cap, + owner, + "job", + tx, + ); + const reserved = reservation.reserved; + if (!reservation.allowed) { await tx.arrivalReaction.updateMany({ where: { id: rowId, userId, occurredAt: revision, generationClaimId }, data: { generationClaimId: null, generationClaimedAt: null }, @@ -408,6 +406,11 @@ async function reserveClaimBudget( data: { generationReservedTokens: reserved, generationBudgetDateKey: dateKey, + // v1.38.19 (Wave E, fix round 1) — persist the cost owner with the + // amount. Two readers need it and neither can derive it: the resume + // path below (the chain may have been reordered since) and the + // supersede refund in `data-arrival.ts` (which has no chain at all). + generationCostOwner: owner, }, }); if (linked.count !== 1) throw new ReactionClaimLostError(); @@ -447,6 +450,7 @@ export async function runReactionLine( generationClaimedAt: true, generationReservedTokens: true, generationBudgetDateKey: true, + generationCostOwner: true, generationProviderInvokedAt: true, occurredAt: true, refId: true, @@ -543,7 +547,12 @@ export async function runReactionLine( allowed: true, reserved: row.generationReservedTokens!, dateKey: row.generationBudgetDateKey!, - owner: resolveCostOwner(chain), + // The owner the reservation was actually booked to. Rows written before + // the column existed have none; the chain is the only answer left for + // those, and it is the answer the whole resume path used to give. + owner: + (row.generationCostOwner as BudgetCostOwner | null) ?? + resolveCostOwner(chain), }; } else { try { diff --git a/tests/integration/arrival-reaction-claim.test.ts b/tests/integration/arrival-reaction-claim.test.ts index 0db295536..3a4284b0e 100644 --- a/tests/integration/arrival-reaction-claim.test.ts +++ b/tests/integration/arrival-reaction-claim.test.ts @@ -266,6 +266,110 @@ describe("ArrivalReaction — the day claim", () => { expect(usage).toMatchObject({ totalTokens: 280, messageCount: 1 }); }); + // v1.38.19 (Wave E, fix round 1) — the refund reverses BOTH counters. + // + // The reservation booked `operator_tokens` whenever the chain's primary was + // operator-funded. The supersede refund reversed only the total, and nothing + // else in the codebase ever subtracts that share — so every superseded + // reaction (any second measurement of the same kind on the same local date + // before the worker runs) walked the operator's ceiling upward against spend + // that never happened. Twenty of them is 28 000 tokens of a 100 000 job share + // gone for nothing, until the UTC day rolls. + it("refunds the operator share of a superseded reservation too", async () => { + const prisma = getPrismaClient(); + const user = await createUser("arrival-operator-refund"); + const early = new Date("2026-07-14T06:00:00Z"); + const late = new Date("2026-07-14T18:00:00Z"); + await prisma.arrivalReaction.create({ + data: { + ...marker(user.id, "weight", early), + generationClaimId: "pre-provider-claim", + generationClaimedAt: new Date("2026-07-14T06:01:00Z"), + generationReservedTokens: 220, + generationBudgetDateKey: "2026-07-14", + generationCostOwner: "operator", + }, + }); + await prisma.coachUsage.create({ + data: { + userId: user.id, + dateKey: "2026-07-14", + totalTokens: 500, + operatorTokens: 300, + messageCount: 2, + }, + }); + + const { runDataArrival } = await import("@/lib/jobs/data-arrival"); + await runDataArrival(prisma as never, { + userId: user.id, + kind: "weight", + salience: "salient", + localDate: LOCAL_DATE, + occurredAt: late.toISOString(), + count: 1, + source: "withings", + }); + + const usage = await prisma.coachUsage.findUniqueOrThrow({ + where: { userId_dateKey: { userId: user.id, dateKey: "2026-07-14" } }, + }); + expect(usage).toMatchObject({ + totalTokens: 280, + operatorTokens: 80, + messageCount: 1, + }); + + // The advanced marker carries no stale owner into the next attempt. + const advanced = await prisma.arrivalReaction.findFirstOrThrow({ + where: { userId: user.id }, + }); + expect(advanced.generationCostOwner).toBeNull(); + }); + + it("leaves the operator counter alone when the user's own plan reserved", async () => { + const prisma = getPrismaClient(); + const user = await createUser("arrival-user-refund"); + const early = new Date("2026-07-14T06:00:00Z"); + const late = new Date("2026-07-14T18:00:00Z"); + await prisma.arrivalReaction.create({ + data: { + ...marker(user.id, "weight", early), + generationClaimId: "pre-provider-claim", + generationClaimedAt: new Date("2026-07-14T06:01:00Z"), + generationReservedTokens: 220, + generationBudgetDateKey: "2026-07-14", + generationCostOwner: "user", + }, + }); + await prisma.coachUsage.create({ + data: { + userId: user.id, + dateKey: "2026-07-14", + totalTokens: 500, + operatorTokens: 300, + messageCount: 2, + }, + }); + + const { runDataArrival } = await import("@/lib/jobs/data-arrival"); + await runDataArrival(prisma as never, { + userId: user.id, + kind: "weight", + salience: "salient", + localDate: LOCAL_DATE, + occurredAt: late.toISOString(), + count: 1, + source: "withings", + }); + + const usage = await prisma.coachUsage.findUniqueOrThrow({ + where: { userId_dateKey: { userId: user.id, dateKey: "2026-07-14" } }, + }); + // Another request's operator balance is not this reservation's to spend. + expect(usage).toMatchObject({ totalTokens: 280, operatorTokens: 300 }); + }); + it("stores the line as encrypted bytes, and tolerates its absence", async () => { const prisma = getPrismaClient(); const user = await createUser("arrival-crypto");