fix: preserve native recovery and settle the latest durable receipt

Keep fresh budget admission blocked without cancelling the native recovery needed to close accounting. Drain every matching disk receipt under the accounting lock and retain files until ledger commit. Add PostgreSQL rollback, backlog, movement, and budget-scope regressions.

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

server/src/services/budgets.ts

@@ -9,6 +9,7 @@ import {
   companies,
   costEvents,
   heartbeatRuns,
+  nativeRunFinalizations,
   projects,
 } from "@paperclipai/db";
 import type {
@@ -152,7 +153,7 @@ export async function computeObservedSpend(
   db: Db,
   policy: Pick<PolicyRow, "companyId" | "scopeType" | "scopeId" | "windowKind" | "metric">,
 ) {
-  if (policy.metric !== "billed_cents") return { total: 0, totalExact: "0.0000000", unpricedEventCount: 0, pendingRunCount: 0 };
+  if (policy.metric !== "billed_cents") return { total: 0, totalExact: "0.0000000", unpricedEventCount: 0, pendingRunCount: 0, recoveringRunCount: 0 };
 
   const conditions = [eq(costEvents.companyId, policy.companyId)];
   if (policy.scopeType === "agent") conditions.push(eq(costEvents.agentId, policy.scopeId));
@@ -178,8 +179,17 @@ export async function computeObservedSpend(
   if (policy.windowKind === "calendar_month_utc") {
     pendingConditions.push(sql`coalesce(${heartbeatRuns.finishedAt}, ${heartbeatRuns.createdAt}) >= ${start.toISOString()}::timestamptz`, sql`coalesce(${heartbeatRuns.finishedAt}, ${heartbeatRuns.createdAt}) < ${end.toISOString()}::timestamptz`);
   }
-  const [pending] = await db.select({ count: sql<number>`count(*)::int` }).from(heartbeatRuns).where(and(...pendingConditions));
-  return { total: Number(row?.total ?? 0), totalExact: normalizeCents(row?.total ?? 0), unpricedEventCount: Number(row?.unpricedEventCount ?? 0), pendingRunCount: pending?.count ?? 0 };
+  const [pending] = await db.select({
+    count: sql<number>`count(*)::int`,
+    recovering: sql<number>`count(*) filter (where ${heartbeatRuns.runtimeMode} = 'native'
+      and exists (select 1 from ${nativeRunFinalizations}
+        where ${nativeRunFinalizations.runId} = ${heartbeatRuns.id}
+          and ${nativeRunFinalizations.companyId} = ${heartbeatRuns.companyId}
+          and ${nativeRunFinalizations.resultId} is null
+          and ${nativeRunFinalizations.phase} <> 'terminal_failure'))::int`,
+  }).from(heartbeatRuns).where(and(...pendingConditions));
+  return { total: Number(row?.total ?? 0), totalExact: normalizeCents(row?.total ?? 0), unpricedEventCount: Number(row?.unpricedEventCount ?? 0),
+    pendingRunCount: pending?.count ?? 0, recoveringRunCount: pending?.recovering ?? 0 };
 }
 
 async function computeObservedAmount(db: Db, policy: PolicyRow) {
@@ -190,7 +200,7 @@ async function reached(db: Db, policy: PolicyRow, threshold: number) {
 }
 function observedBlocks(policy: PolicyRow, observed: Awaited<ReturnType<typeof computeObservedSpend>>) {
   if (!policy.isActive || !policy.hardStopEnabled || policy.amount <= 0) return false;
-  return compareCents(observed.totalExact, policy.amount) >= 0 || observed.pendingRunCount > 0 || (policy.unpricedUsagePolicy !== "allow" && observed.unpricedEventCount > 0);
+  return compareCents(observed.totalExact, policy.amount) >= 0 || (observed.pendingRunCount - observed.recoveringRunCount) > 0 || (policy.unpricedUsagePolicy !== "allow" && observed.unpricedEventCount > 0);
 }
 async function policyBlocks(db: Db, policy: PolicyRow) {
   if (!policy.isActive || !policy.hardStopEnabled || policy.amount <= 0) return false;
@@ -552,10 +562,11 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
     const policies = await db.select().from(budgetPolicies).where(and(
       eq(budgetPolicies.companyId, companyId), eq(budgetPolicies.scopeType, scopeType), eq(budgetPolicies.scopeId, scopeId),
     ));
+    let recoveringPolicy: PolicyRow | undefined;
     for (const policy of policies) {
-      const observed = await computeObservedAmount(db, policy);
-      if (await policyBlocks(db, policy)) {
-        await createIncidentIfNeeded(policy, "hard", observed);
+      const observed = await computeObservedSpend(db, policy);
+      if (observedBlocks(policy, observed)) {
+        await createIncidentIfNeeded(policy, "hard", observed.total);
         await pauseAndCancelScopeForBudget(policy);
         const title = scopeType === "company" ? "Company" : scopeType === "agent" ? "Agent" : "Project";
         if ((await computeObservedSpend(db, policy)).pendingRunCount > 0) {
@@ -568,6 +579,9 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
           ? "Project cannot start work because its budget hard-stop is still exceeded."
           : `${title} is paused because its budget hard-stop was reached.` };
       }
+      // An open native coordinator must finish the original run's accounting.
+      // Deny fresh work without pausing/cancelling that recovery or its hold.
+      if (policy.isActive && policy.hardStopEnabled && policy.amount > 0 && observed.recoveringRunCount > 0) recoveringPolicy ??= policy;
       // A closed UTC window, inactive policy, or budget raise releases only
       // budget-owned pauses. Manual pauses remain operator decisions.
       await resolveOpenIncidentsForPolicy(policy.id, "approved", "budget_service", "hard");
@@ -584,6 +598,8 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
       return { scopeType, scopeId, scopeName: scope.name, reason: "Budget pause requires a policy or an explicit operator resume." };
     }
     if (scope.pauseReason === "budget" && policies[0]) await resumeScopeFromBudget(policies[0]);
+    if (recoveringPolicy) return { scopeType, scopeId, scopeName: scope.name,
+      reason: "New work must wait while a native run recovers its unfinished accounting." };
     return null;
   }
 

server/src/services/run-cost-accounting.ts

@@ -5,6 +5,8 @@ import { withAccountingTransaction } from "./accounting-transaction.js";
 import { createCostEventInTransaction } from "./costs.js";
 import { budgetService, deliverBudgetEnforcement, type BudgetServiceHooks } from "./budgets.js";
 import { logger } from "../middleware/logger.js";
+import { promises as fs } from "node:fs";
+import { recoverPendingRunUsageReceipts } from "./usage-receipts.js";
 
 const terminalStatuses = ["succeeded", "failed", "timed_out", "cancelled", "interrupted"];
 const object = (value: unknown): Record<string, unknown> => value && typeof value === "object" && !Array.isArray(value) ? value as Record<string, unknown> : {};
@@ -16,13 +18,19 @@ const amount = (value: unknown) => typeof value === "number" && Number.isFinite(
 export async function accountRunCost(db: Db, runId: string, hooks: BudgetServiceHooks = {}) {
   const [identity] = await db.select({ companyId: heartbeatRuns.companyId }).from(heartbeatRuns).where(eq(heartbeatRuns.id, runId));
   if (!identity) return false;
+  let recoveredFiles: string[] = [];
   const accounted = await withAccountingTransaction(db, identity.companyId, async (tx, publications) => {
-    const [run] = await tx.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)).for("update");
+    let [run] = await tx.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)).for("update");
     if (!run?.costAccountingPending || !terminalStatuses.includes(run.status)) return false;
     if (run.costAccountedAt) {
       await tx.update(heartbeatRuns).set({ costAccountingPending: false }).where(eq(heartbeatRuns.id, run.id));
       return false;
     }
+    // The bounded global replay can leave a newer receipt behind an older
+    // complete snapshot. Drain this run under the accounting lock before
+    // deciding completeness, and retain disk evidence until the outer commit.
+    recoveredFiles = await recoverPendingRunUsageReceipts(tx, { ...identity, runId }, undefined, true);
+    [run] = await tx.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId));
     const preProviderFailure = object(object(run.resultJson).executionRecovery).providerWorkStarted === false
       || object(run.usageJson).accountingProviderWorkStarted === false;
     // A failed native turn is not necessarily the end of this heartbeat: the
@@ -118,6 +126,10 @@ export async function accountRunCost(db: Db, runId: string, hooks: BudgetService
     await tx.update(heartbeatRuns).set({ costAccountingPending: false, costAccountedAt: new Date(), accountingProjectionVersion: "v2", accountingLastError: null, accountingLastAttemptAt: new Date() }).where(eq(heartbeatRuns.id, run.id));
     return true;
   });
+  for (const file of recoveredFiles) {
+    try { await fs.rm(file, { force: true }); }
+    catch (error) { logger.warn({ err: error, runId }, "Accounting committed; recovered receipt cleanup will retry"); }
+  }
   await deliverBudgetEnforcement(db, hooks, identity.companyId);
   return accounted;
 }

server/src/services/usage-receipts.ts

@@ -113,28 +113,44 @@ 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;
+export async function recoverPendingRunUsageReceipts(db: Db, input: { companyId: string; runId: string }, directory = usageReceiptSpoolPath(), retainFiles = false) {
+  // The bounded sweep may not reach this run. Its durable evidence must
+  // reach the journal before source replacement or ledger acknowledgement.
+  // A failed save aborts the operation so recovery can retry.
+  const recovered = new Set<string>();
+  // A concurrent failed replay can rename a file after releasing its company
+  // lock. Rescan disappearing paths rather than certify an incomplete scan.
+  for (let scan = 0; scan < 3; scan++) {
+    let names: string[];
+    try { names = await fs.readdir(directory); }
+    catch (error) { if ((error as NodeJS.ErrnoException).code === "ENOENT") return [...recovered]; throw error; }
+    let moved = false;
+    for (const name of names.filter(name => name.endsWith(".json"))) {
+      const file = path.join(directory, name);
+      if (recovered.has(file)) continue;
+      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") { moved = true; continue; }
+        if (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));
+      // Settlement uses the caller's transaction: deleting before its commit
+      // would lose the only durable evidence if the later ledger write fails.
+      recovered.add(file);
+      if (!retainFiles) await fs.rm(file, { force: true });
     }
-    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 });
+    if (!moved) return [...recovered];
   }
+  throw new Error("Accounting spool changed during recovery; retry required");
 }
 
 export async function createRunUsageRecorder(db: Db, input: { companyId: string; runId: string; adapterType: string }, directory = usageReceiptSpoolPath()) {