fix: recover pending receipts before replacing run recorder

Co-Authored-By: Paperclip <noreply@paperclip.ing>

doc/COST-ACCOUNTING.md

@@ -71,6 +71,8 @@ Run `pnpm exec playwright test --config tests/e2e/playwright.config.ts tests/e2e
 
 The spool lives at `<instance-root>/accounting-receipts`. Directories use mode 0700 and receipt files 0600. Files contain accounting fields only; transcript text, prompts, tools, and credentials are excluded. Publication awaits file and directory fsync on POSIX. Each record carries company/run identity, a controller source ID, sequence, timestamp, normalized usage, completeness, and price. Replay validates the schema and fingerprint and rotates failures behind fresh entries in bounded batches. Old controller snapshots are retained as evidence but cannot overwrite a replacement controller or an acknowledged run. Multiple provider attempts are accounted together; incomplete attempts and capture failures remain incomplete.
 
+Before replacing a stopped run's recorder, recovery saves every pending spool receipt for that company and run, independently of the bounded startup sweep. A save failure blocks replacement and retains the old source for retry. Native recovery then restores its latest journal snapshot; subsequent cumulative run usage replaces that snapshot rather than adding it twice.
+
 The spool survives process death on persistent local storage, including the tested boundary before the first database write. It does not survive destruction of that storage, and cannot recover provider activity that was never emitted as usage. Back up persistent instance storage together with PostgreSQL. No arbitrary provider-log scraping or automatic provider re-execution occurs during accounting recovery.
 
 Board-only routes under `/api/companies/:companyId/accounting` provide:

server/src/__tests__/cost-accounting-operations-edges.test.ts

@@ -96,6 +96,83 @@ const support = await getEmbeddedPostgresTestSupport();
     expect((await db.select().from(costEvents).where(eq(costEvents.heartbeatRunId, f.run.id)))[0]).toMatchObject({ inputTokens: 150, outputTokens: 15, costCents: 15 });
   });
 
+  it.each([false, true])("recovers pending native receipts beyond the startup batch before replacement (save failure: %s)", async (failSave) => {
+    const f = await fixture(), other = await fixture();
+    await db.update(heartbeatRuns).set({ runtimeMode: "native" }).where(eq(heartbeatRuns.id, f.run.id));
+    await reserveRunBudget(db, f.company.id, f.run.id, null);
+    const input = { companyId: f.company.id, runId: f.run.id, adapterType: "paperclip_runner" };
+    const spool = path.join(directory, randomUUID());
+    const recorder = await createRunUsageRecorder(db, input, spool);
+    await recorder.capture({ complete: true, usageBasis: "per_run", usage: { inputTokens: 100, outputTokens: 10 }, costUsd: 0.1 });
+    const before = (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.run.id)))[0];
+    const unavailable = vi.spyOn(db, "transaction").mockRejectedValue(new Error("Database unavailable"));
+    try {
+      await recorder.capture({ complete: true, usageBasis: "per_run", usage: { inputTokens: 150, outputTokens: 15 }, costUsd: 0.15 });
+      await recorder.capture({ complete: true, usageBasis: "per_run", usage: { inputTokens: 200, outputTokens: 20 }, costUsd: 0.2 });
+    } finally { unavailable.mockRestore(); }
+    const pending = (await fs.readdir(spool)).filter(name => name.endsWith(".json"));
+    expect(pending).toHaveLength(2);
+    for (const name of pending) await fs.rename(path.join(spool, name), path.join(spool, `zzz-${name}`));
+    const unrelated: UsageReceiptEnvelope = { schema: "paperclip/accounting-receipt/v1", id: randomUUID(),
+      companyId: other.company.id, runId: other.run.id, sourceId: randomUUID(), sequence: 1,
+      receivedAt: new Date().toISOString(), adapterType: "process", receipt: { complete: false } };
+    await Promise.all(Array.from({ length: 101 }, async (_, index) => {
+      await fs.writeFile(path.join(spool, `000-${index}.json`), JSON.stringify({ ...unrelated, id: randomUUID(), sequence: index + 1 }));
+    }));
+    expect(await replayUsageReceipts(db, spool)).toEqual({ replayed: 100, failed: 0 });
+    expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.run.id)))[0].usageJson?.inputTokens).toBe(100);
+    if (failSave) {
+      await db.execute(sql`create function reject_recovered_receipt() returns trigger language plpgsql as $$ begin raise exception 'receipt save failed'; end $$`);
+      await db.execute(sql`create trigger reject_recovered_receipt before insert on run_usage_receipts for each row execute function reject_recovered_receipt()`);
+      try {
+        await expect(createRunUsageRecorder(db, input, spool)).rejects.toThrow();
+        expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.run.id)))[0].usageJson?.accountingReceiptSourceId)
+          .toBe(before.usageJson?.accountingReceiptSourceId);
+        expect((await fs.readdir(spool)).filter(name => name.startsWith("zzz-"))).toHaveLength(2);
+      } finally {
+        await db.execute(sql`drop trigger reject_recovered_receipt on run_usage_receipts`);
+        await db.execute(sql`drop function reject_recovered_receipt()`);
+      }
+    }
+    const resumed = await createRunUsageRecorder(db, input, spool);
+    expect((await fs.readdir(spool)).filter(name => name.startsWith("zzz-"))).toHaveLength(0);
+    expect(await db.select().from(runUsageReceipts).where(eq(runUsageReceipts.runId, f.run.id))).toHaveLength(3);
+    expect(await resumed.complete({ exitCode: 0, signal: null, timedOut: false }))
+      .toMatchObject({ complete: false, usage: { inputTokens: 200, outputTokens: 20 } });
+    await resumed.capture({ complete: true, usageBasis: "per_run", usage: { inputTokens: 220, outputTokens: 22 }, costUsd: 0.22 });
+    await db.update(heartbeatRuns).set({ status: "failed", finishedAt: new Date() }).where(eq(heartbeatRuns.id, f.run.id));
+    expect(await accountRunCost(db, f.run.id)).toBe(true);
+    expect(await accountRunCost(db, f.run.id)).toBe(false);
+    expect((await db.select().from(costEvents).where(eq(costEvents.heartbeatRunId, f.run.id)))[0])
+      .toMatchObject({ inputTokens: 220, outputTokens: 22, costCents: 22 });
+    expect((await db.select().from(budgetReservations).where(eq(budgetReservations.runId, f.run.id)))[0].state).toBe("settled");
+    expect(await replayUsageReceipts(db, spool)).toEqual({ replayed: 1, failed: 0 });
+    expect((await costService(db).summary(f.company.id)).spendCents).toBe(22);
+  });
+
+  it("leaves unrelated corrupt spool evidence alone and aborts handoff on unreadable or invalid matching receipts", async () => {
+    const f = await fixture(), spool = path.join(directory, randomUUID());
+    const input = { companyId: f.company.id, runId: f.run.id, adapterType: "paperclip_runner" };
+    await createRunUsageRecorder(db, input, spool);
+    const source = (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.run.id)))[0].usageJson?.accountingReceiptSourceId;
+    const matching = path.join(spool, "matching.json");
+    await fs.writeFile(matching, JSON.stringify({ ...input, receipt: "invalid" }));
+    await expect(createRunUsageRecorder(db, input, spool)).rejects.toThrow();
+    expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.run.id)))[0].usageJson?.accountingReceiptSourceId).toBe(source);
+    const read = vi.spyOn(fs, "readFile").mockRejectedValueOnce(Object.assign(new Error("Unreadable receipt"), { code: "EACCES" }));
+    try { await expect(createRunUsageRecorder(db, input, spool)).rejects.toThrow("Unreadable receipt"); } finally { read.mockRestore(); }
+    expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.run.id)))[0].usageJson?.accountingReceiptSourceId).toBe(source);
+    await fs.rm(matching);
+    const unrelated = ["invalid", "null", "{}", JSON.stringify({ companyId: f.company.id }),
+      JSON.stringify({ companyId: f.company.id, runId: randomUUID() }), "x".repeat(1024 * 1024 + 1)];
+    for (const [index, value] of unrelated.entries()) await fs.writeFile(path.join(spool, `${index}.json`), value);
+    await fs.mkdir(path.join(spool, "directory.json"));
+    const stat = vi.spyOn(fs, "lstat").mockRejectedValueOnce(Object.assign(new Error("Concurrent replay removed file"), { code: "ENOENT" }));
+    try { await createRunUsageRecorder(db, input, spool); } finally { stat.mockRestore(); }
+    await createRunUsageRecorder(db, input, spool);
+    expect(await fs.readdir(spool)).toHaveLength(unrelated.length + 1);
+  });
+
   it("reports over-limit advisory budgets without blocking, and tolerates legacy active zero limits", async () => {
     const f = await fixture(), budgets = budgetService(db);
     const policy = await budgets.upsertPolicy(f.company.id, { scopeType: "company", scopeId: f.company.id, amount: 1, hardStopEnabled: false }, "board");

server/src/services/usage-receipts.ts

@@ -113,6 +113,30 @@ export async function replayUsageReceipts(db: Db, directory = usageReceiptSpoolP
   return { replayed, failed };
 }
 
+async function recoverPendingRunUsageReceipts(db: Db, input: { companyId: string; runId: string }, directory: string) {
+  // Startup's bounded sweep may not reach this run. Its stopped controller's
+  // durable evidence must reach the journal before we fence that source out.
+  // A failed save deliberately aborts replacement so recovery can retry.
+  for (const name of (await fs.readdir(directory)).filter(name => name.endsWith(".json"))) {
+    const file = path.join(directory, name);
+    let raw: unknown;
+    try {
+      const info = await fs.lstat(file);
+      if (!info.isFile() || info.size > 1024 * 1024) continue;
+      raw = JSON.parse(await fs.readFile(file, "utf8"));
+    } catch (error) {
+      // The background replayer can move/remove a file while we scan. Invalid
+      // JSON remains on disk for its normal recovery/error reporting path.
+      if ((error as NodeJS.ErrnoException).code === "ENOENT" || error instanceof SyntaxError) continue;
+      throw error;
+    }
+    if (!raw || typeof raw !== "object" || !("companyId" in raw) || !("runId" in raw)
+      || raw.companyId !== input.companyId || raw.runId !== input.runId) continue;
+    await persistUsageReceipt(db, envelopeSchema.parse(raw));
+    await fs.rm(file, { force: true });
+  }
+}
+
 export async function createRunUsageRecorder(db: Db, input: { companyId: string; runId: string; adapterType: string }, directory = usageReceiptSpoolPath()) {
   const sourceId = randomUUID();
   // Verify durable storage before starting paid work, including a directory
@@ -121,6 +145,7 @@ export async function createRunUsageRecorder(db: Db, input: { companyId: string;
   const probe = await fs.open(path.join(directory, `${sourceId}.probe`), "wx", 0o600);
   try { await probe.sync(); } finally { await probe.close(); await fs.rm(path.join(directory, `${sourceId}.probe`)); }
   await syncDirectory(directory);
+  await recoverPendingRunUsageReceipts(db, input, directory);
   const bound = await db.update(heartbeatRuns).set({ usageJson: sql`coalesce(${heartbeatRuns.usageJson}, '{}'::jsonb) || ${JSON.stringify({ accountingReceiptSourceId: sourceId, accountingReceiptSequence: 0, accountingReceiptReady: false })}::jsonb` })
     .where(and(eq(heartbeatRuns.id, input.runId), eq(heartbeatRuns.companyId, input.companyId), isNull(heartbeatRuns.costAccountedAt))).returning({ id: heartbeatRuns.id, runtimeMode: heartbeatRuns.runtimeMode, usageJson: heartbeatRuns.usageJson });
   if (bound.length !== 1) throw conflict("Run cannot accept a new usage recorder");