From e1c61054ccfc85fec3ee775da657f8f39b2f7d02 Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Fri, 2 Oct 2026 17:33:13 -0700 Subject: [PATCH] fix: settle interrupted usage and retain estimate provenance Co-Authored-By: Paperclip --- .../cost-accounting-operators.test.ts | 29 +++++++++++++++++-- server/src/services/accounting-integrity.ts | 8 ++--- server/src/services/billing-reconciliation.ts | 2 +- server/src/services/budgets.ts | 2 +- server/src/services/costs.ts | 2 +- server/src/services/run-cost-accounting.ts | 2 +- 6 files changed, 34 insertions(+), 11 deletions(-) diff --git a/server/src/__tests__/cost-accounting-operators.test.ts b/server/src/__tests__/cost-accounting-operators.test.ts index e47565f1894..52c2e96a57f 100644 --- a/server/src/__tests__/cost-accounting-operators.test.ts +++ b/server/src/__tests__/cost-accounting-operators.test.ts @@ -10,7 +10,7 @@ import { billingReconciliationService } from "../services/billing-reconciliation import { budgetService } from "../services/budgets.js"; import { reserveRunBudget } from "../services/budget-reservations.js"; import { costService } from "../services/costs.js"; -import { accountRunCost } from "../services/run-cost-accounting.js"; +import { accountRunCost, reconcileRunCosts } from "../services/run-cost-accounting.js"; import { createRunUsageRecorder, persistUsageReceipt, replayUsageReceipts, spoolUsageReceipt, type UsageReceiptEnvelope } from "../services/usage-receipts.js"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; @@ -65,6 +65,28 @@ const support = await getEmbeddedPostgresTestSupport(); expect((await accountingIntegrityService(db).inspect(f.company.id)).findings).toEqual([]); }); + it("keeps an interrupted partial receipt pending and settles its reservation when evidence becomes complete", async () => { + const f = await fixture(); const run = await runFor(f); + await budgetService(db).upsertPolicy(f.company.id, { scopeType: "agent", scopeId: f.agent.id, amount: 100, reservationCents: "10" }, "board"); + await reserveRunBudget(db, f.company.id, run.id, null); + const usage = { accountingReceiptReady: false, costUsdExact: "0.01", inputTokens: 5, provider: "test", billingType: "metered_api" }; + await db.update(heartbeatRuns).set({ status: "interrupted", usageJson: usage }).where(eq(heartbeatRuns.id, run.id)); + const integrity = accountingIntegrityService(db); + expect((await integrity.health(f.company.id)).pendingRunCount).toBe(1); + expect((await integrity.health(f.company.id)).items[0]).toMatchObject({ runId: run.id, state: "waiting_for_receipt" }); + expect((await costService(db).summary(f.company.id)).pendingRunCount).toBe(1); + expect((await integrity.inspect(f.company.id)).findings.some(f => f.kind === "missing_acknowledgement")).toBe(true); + expect(await budgetService(db).getInvocationBlock(f.company.id, f.agent.id)).not.toBeNull(); + expect(await accountRunCost(db, run.id)).toBe(false); + expect((await integrity.health(f.company.id)).heldReservationCents).toBe("10.0000000"); + await db.update(heartbeatRuns).set({ usageJson: { ...usage, accountingReceiptReady: true } }).where(eq(heartbeatRuns.id, run.id)); + await reconcileRunCosts(db); + expect((await costService(db).summary(f.company.id))).toMatchObject({ pendingRunCount: 0, spendCents: 1 }); + expect((await integrity.health(f.company.id)).heldReservationCents).toBe("0.0000000"); + expect((await integrity.inspect(f.company.id)).findings).toEqual([]); + expect(await accountRunCost(db, run.id)).toBe(false); + }); + it("independently detects and repairs corrupted projections with a stale-review guard and audit", async () => { const f = await fixture(); const run = await runFor(f, "succeeded"); await db.update(heartbeatRuns).set({ costAccountingPending: true, usageJson: { accountingReceiptReady: true, costUsdExact: "0.012345678", inputTokens: 7, cachedInputTokens: 3, outputTokens: 11 } }).where(eq(heartbeatRuns.id, run.id)); @@ -163,7 +185,7 @@ const support = await getEmbeddedPostgresTestSupport(); expect(await db.select().from(costAdjustments).where(eq(costAdjustments.companyId, f.company.id))).toEqual([]); }); - it("prices unknown usage from an invoice and updates runtime, admission and independent verification", async () => { + it.each(["provider_invoice", "rate_card"] as const)("prices unknown usage from %s and preserves its estimate status", async (source) => { const f = await fixture(); const run = await runFor(f, "failed"); await budgetService(db).upsertPolicy(f.company.id, { scopeType: "agent", scopeId: f.agent.id, amount: 10 }, "board"); await db.update(heartbeatRuns).set({ costAccountingPending: true, usageJson: { accountingReceiptReady: true, inputTokens: 9, provider: "test" } }).where(eq(heartbeatRuns.id, run.id)); @@ -171,9 +193,10 @@ const support = await getEmbeddedPostgresTestSupport(); expect((await accountingIntegrityService(db).health(f.company.id)).unpricedEventCount).toBe(1); expect(await budgetService(db).getInvocationBlock(f.company.id, f.agent.id)).not.toBeNull(); const [event] = await db.select().from(costEvents).where(eq(costEvents.heartbeatRunId, run.id)); - await billingReconciliationService(db).adjust(f.company.id, event.id, { idempotencyKey: "pricing", expectedCents: 0, correctedCents: "1.0000001", reason: "Verified provider invoice", pricing: { source: "provider_invoice", evidence: "billing-export" } }, "board"); + await billingReconciliationService(db).adjust(f.company.id, event.id, { idempotencyKey: "pricing", expectedCents: 0, correctedCents: "1.0000001", reason: "Verified provider invoice", pricing: { source, evidence: "billing-export" } }, "board"); expect(await budgetService(db).getInvocationBlock(f.company.id, f.agent.id)).toBeNull(); expect((await accountingIntegrityService(db).health(f.company.id)).unpricedEventCount).toBe(0); + expect((await costService(db).summary(f.company.id)).estimatedEventCount).toBe(source === "rate_card" ? 1 : 0); expect((await accountingIntegrityService(db).inspect(f.company.id)).findings).toEqual([]); }); diff --git a/server/src/services/accounting-integrity.ts b/server/src/services/accounting-integrity.ts index 2e03fdd8e53..56c0412169b 100644 --- a/server/src/services/accounting-integrity.ts +++ b/server/src/services/accounting-integrity.ts @@ -39,7 +39,7 @@ async function inspect(tx: Db, companyId: string): Promise const incomplete = await tx.execute<{ id: string; acknowledged: boolean; events: number }>(sql` select r.id, r.cost_accounted_at is not null as acknowledged, count(e.id)::int as events from heartbeat_runs r left join cost_events e on e.heartbeat_run_id = r.id and e.company_id = r.company_id and e.idempotency_key like ('heartbeat:' || r.id::text || ':%') - where r.company_id = ${companyId} and r.status in ('succeeded','failed','timed_out','cancelled') + where r.company_id = ${companyId} and r.status in ('succeeded','failed','timed_out','cancelled','interrupted') and (r.cost_accounting_pending or r.accounting_projection_version = 'v2') and r.result_json->'executionRecovery'->>'providerWorkStarted' is distinct from 'false' and r.usage_json->>'accountingProviderWorkStarted' is distinct from 'false' @@ -100,9 +100,9 @@ export function accountingIntegrityService(db: Db, hooks: BudgetServiceHooks = { }, health: (companyId: string): Promise => withAccountingTransaction(db, companyId, async tx => { const [counts] = await tx.execute<{ pending: number; unpriced: number; oldest: string | null; cancellations: number; reserved: string }>(sql` - select (select count(*)::int from heartbeat_runs where company_id = ${companyId} and cost_accounting_pending and status in ('succeeded','failed','timed_out','cancelled')) as pending, + select (select count(*)::int from heartbeat_runs where company_id = ${companyId} and cost_accounting_pending and status in ('succeeded','failed','timed_out','cancelled','interrupted')) as pending, (select count(*)::int from cost_events where company_id = ${companyId} and cost_status = 'unpriced' and billing_type <> 'subscription_included') as unpriced, - (select min(coalesce(finished_at,created_at))::text from heartbeat_runs where company_id = ${companyId} and cost_accounting_pending and status in ('succeeded','failed','timed_out','cancelled')) as oldest, + (select min(coalesce(finished_at,created_at))::text from heartbeat_runs where company_id = ${companyId} and cost_accounting_pending and status in ('succeeded','failed','timed_out','cancelled','interrupted')) as oldest, (select count(*)::int from budget_policies where company_id = ${companyId} and enforcement_version > enforcement_delivered_version) as cancellations, (select coalesce(sum(amount_cents),0)::text from budget_reservations where company_id = ${companyId} and state = 'held') as reserved`); const pending = await tx.execute<{ runId: string; agentId: string; state: "waiting_for_receipt" | "retryable"; lastError: string | null; attempts: number; since: string; lastAttemptAt: string | null }>(sql` @@ -110,7 +110,7 @@ export function accountingIntegrityService(db: Db, hooks: BudgetServiceHooks = { and usage_json->>'accountingProviderWorkStarted' is distinct from 'false' and result_json->'executionRecovery'->>'providerWorkStarted' is distinct from 'false' then 'waiting_for_receipt' else 'retryable' end as state, accounting_last_error as "lastError", accounting_attempt_count as attempts, coalesce(finished_at,created_at)::text as since, accounting_last_attempt_at::text as "lastAttemptAt" - from heartbeat_runs where company_id = ${companyId} and cost_accounting_pending and status in ('succeeded','failed','timed_out','cancelled') order by coalesce(finished_at,created_at),id limit 100`); + from heartbeat_runs where company_id = ${companyId} and cost_accounting_pending and status in ('succeeded','failed','timed_out','cancelled','interrupted') order by coalesce(finished_at,created_at),id limit 100`); const unpriced = await tx.execute<{ runId: string | null; costEventId: string; agentId: string; since: string }>(sql` select heartbeat_run_id as "runId", id as "costEventId", agent_id as "agentId", occurred_at::text as since from cost_events where company_id = ${companyId} and cost_status = 'unpriced' and billing_type <> 'subscription_included' order by occurred_at,id limit 100`); diff --git a/server/src/services/billing-reconciliation.ts b/server/src/services/billing-reconciliation.ts index 21cec4110ed..99e2af8689c 100644 --- a/server/src/services/billing-reconciliation.ts +++ b/server/src/services/billing-reconciliation.ts @@ -117,7 +117,7 @@ export function billingReconciliationService(db: Db, hooks: BudgetServiceHooks = // Original provider value and fingerprint remain immutable. cost_cents // is the effective valuation, backed by this append-only correction. const [updated] = await tx.update(costEvents).set({ costCents: sql`${input.correctedCents}::numeric`, - reportedCostCents: event.event.reportedCostCents ?? normalizeCents(event.exact), costStatus: "reported", pricingProvenance: input.pricing, + reportedCostCents: event.event.reportedCostCents ?? normalizeCents(event.exact), costStatus: input.pricing.source === "rate_card" ? "estimated" : "reported", pricingProvenance: input.pricing, }).where(eq(costEvents.id, eventId)).returning(); await updateMonthlySpendProjections(tx, companyId, updated.agentId, subtractCents(input.correctedCents, input.expectedCents), updated.occurredAt); if (updated.heartbeatRunId) { diff --git a/server/src/services/budgets.ts b/server/src/services/budgets.ts index b26ffe092e2..45236d29c4e 100644 --- a/server/src/services/budgets.ts +++ b/server/src/services/budgets.ts @@ -170,7 +170,7 @@ export async function computeObservedSpend( .where(and(...conditions)); const pendingConditions = [eq(heartbeatRuns.companyId, policy.companyId), eq(heartbeatRuns.costAccountingPending, true), - inArray(heartbeatRuns.status, ["succeeded", "failed", "timed_out", "cancelled"])]; + inArray(heartbeatRuns.status, ["succeeded", "failed", "timed_out", "cancelled", "interrupted"])]; if (policy.scopeType === "agent") pendingConditions.push(eq(heartbeatRuns.agentId, policy.scopeId)); if (policy.scopeType === "project") pendingConditions.push(sql`${heartbeatRuns.usageJson}->'ledgerScope'->>'projectId' = ${policy.scopeId}`); if (policy.windowKind === "calendar_month_utc") { diff --git a/server/src/services/costs.ts b/server/src/services/costs.ts index 431dd5a136a..b25a3e3bb9d 100644 --- a/server/src/services/costs.ts +++ b/server/src/services/costs.ts @@ -178,7 +178,7 @@ export function costService(db: Db, budgetHooks: BudgetServiceHooks = {}) { .where(and(...conditions)); const pendingConditions = [eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.costAccountingPending, true), - inArray(heartbeatRuns.status, ["succeeded", "failed", "timed_out", "cancelled"])]; + inArray(heartbeatRuns.status, ["succeeded", "failed", "timed_out", "cancelled", "interrupted"])]; if (range?.from) pendingConditions.push(sql`coalesce(${heartbeatRuns.finishedAt}, ${heartbeatRuns.createdAt}) >= ${range.from.toISOString()}::timestamptz`); if (range?.to) pendingConditions.push(sql`coalesce(${heartbeatRuns.finishedAt}, ${heartbeatRuns.createdAt}) <= ${range.to.toISOString()}::timestamptz`); const [pending] = await db.select({ count: sql`count(*)::int` }).from(heartbeatRuns).where(and(...pendingConditions)); diff --git a/server/src/services/run-cost-accounting.ts b/server/src/services/run-cost-accounting.ts index 84b7b2d2f05..a04accf0774 100644 --- a/server/src/services/run-cost-accounting.ts +++ b/server/src/services/run-cost-accounting.ts @@ -6,7 +6,7 @@ import { createCostEventInTransaction } from "./costs.js"; import { budgetService, deliverBudgetEnforcement, type BudgetServiceHooks } from "./budgets.js"; import { logger } from "../middleware/logger.js"; -const terminalStatuses = ["succeeded", "failed", "timed_out", "cancelled"]; +const terminalStatuses = ["succeeded", "failed", "timed_out", "cancelled", "interrupted"]; const object = (value: unknown): Record => value && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; const text = (value: unknown) => typeof value === "string" && value.length > 0 ? value : null; const amount = (value: unknown) => typeof value === "number" && Number.isFinite(value) && value >= 0 ? value : null;