fix(ai): the arrival reaction reserves through the shared ledger gate

The arrival-reaction claim carried its own copy of the reservation 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
background ceiling. On the row the wave was written for, 1.2 million tokens
served by the user's own plan, the reaction was refused all day at half the old
ceiling, reproducing the defect the wave exists to remove on the one surface it
forgot to convert. reserveBudget now takes the transaction client, so both
writes still roll back together and there is no second copy left to drift.

The row also remembers who paid for its reservation. The supersede refund in
data-arrival has no chain to ask and could only reverse the total, so the
operator share of every superseded reservation stayed on the row forever,
walking the operator's ceiling upward against spend that never happened. A
supersede is routine, not an edge case: any second measurement of the same kind
on the same local date before the worker runs. The resume path re-derived the
owner from a chain the health ledger may have reordered since, which leaks the
same way or drains another request's balance.

A structural guard makes the invariant executable. Every INSERT column list,
SET clause and RETURNING list that names one counter must name the other, and
the set of files holding raw ledger SQL is pinned. Both defects above were a
grep away and both would have failed this test.

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;

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")

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());
+  });
+});

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,
+    );
+  });
+});

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,
       },
     });

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 {

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");