From 06bbcea12f02f24863366f78676d49dd9bb76975 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Fri, 6 Nov 2026 18:55:26 -0600 Subject: [PATCH] fix(budgets): price lost receipts and stop restart-derived accounting from hard-stopping companies (#19673) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Budget admission and run accounting decide whether any task in a company may start > - #19497 made pending accounting an admission hold and settles a stopped run's lost receipt as an unpriced charge > - Restart-chaos waves still stopped whole companies. A lost Codex receipt with complete token counts settled as unpriced because the rate card only prices complete receipts. Under the default blocking policy that hard-stopped the company at 0.38 of 3000 cents. In another storm, a resumed run's continuation was refused by its own task's recovering-accounting hold. Separately, unpriced charges with zero tokens or a known partial cost hard-stopped a company at 120 of 2000 cents > - This pull request prices a lost receipt's observed usage from the rate card, lets a task's continuation pass its own recovering accounting, and stops restart-derived or evidence-free unpriced charges from tripping the blocking policy > - The benefit is that a restart no longer pauses a company by itself, while genuinely unpriced usage still blocks ## Linked Issues or Issue Description Follow-up to #19497. Refs #19606. **What happened?** - Wave A (K6 crash loop): native Codex run `5037b8d1` was killed mid-turn. Its receipt was lost, but `usage_json` held gpt-6.1-sol with 28690 input and 220 output tokens. The charge was written `cost_status=unpriced`, `pricing_provenance.source=unknown`, because `priceCodexReceipt` returns early when `!receipt.complete`. 0.2 s later the default `unpricedUsagePolicy=block` opened a hard incident at 0.38 of 3000 cents. The company was paused, and 6/6 tasks were blocked with `canRetry=false`. - Wave B (K1 storm): Claude run `2a7a8abf` resumed after the restart. `reserveRunBudget` refused its interaction continuation with "New work must wait while a native run recovers its unfinished accounting", so the task could not settle its own recovery. A 0-token unpriced charge and an unpriced charge with a known 0.81-cent cost then hard-stopped the company at 120 of 2000 cents. **What did you expect?** A restart-lost receipt with known numbers is priced as an estimate, and no restart-derived accounting state pauses a company by itself. A task's own continuation is never held by its own accounting. Real usage above the limit, and genuinely unpriced usage, still stop work. **Steps to reproduce** 1. Give a company a hard-stop budget with the default unpriced-usage policy. 2. Kill a native Codex turn after it has reported tokens, so the final receipt is lost and native recovery ends in terminal failure with a verified stop. 3. Observe an unpriced charge for known tokens, a hard incident, and a paused company. ## What Changed - `run-cost-accounting.ts`: when a lost receipt settles and its usage has known tokens and no known cost, price it with `priceCodexReceipt`/`priceAnthropicReceipt` as a complete per-run checkpoint. The charge is `estimated`, its provenance keeps the rate-card source, and its evidence says "final receipt lost; observed usage is a lower bound". The estimate is also written to the run's usage so a later receipt reconciles against it. Zero observed tokens, and per-model usage, stay unpriced (unknown is never shown as zero, and aggregate totals never take one model's rate). - `budget-exposure.ts`: - The pending/recovering accounting hold excludes the run being admitted and earlier runs of the same task (same `native_issue_id`). A task's continuation and finalization can therefore settle its own accounting. Other tasks stay held, and those runs still count as spend. - The blocking unpriced-usage policy no longer counts these unpriced charges: the settled final charge of a lost receipt (restart-derived, any usage; later growth still counts), a charge with a known partial cost, and metered API or subscription billing with no observed tokens. Unknown billing with no tokens (for example Cursor Cloud) and complete-receipt usage that has no price still block. - `budgets.ts` / `budget-reservations.ts`: admission accepts the admitted `runId` and passes it to the exposure read. `reserveRunBudget`, the final dispatch recheck and the read-only admitted recheck all supply it, so a continuation that reserves can also dispatch. - Daytona wave s12 (run f1d98917, a blocked remote recovery with a 0-cent held reservation and every lease destroyed) is closed by #19606. A Retry for that same task passes the hold through the same-task exclusion here. ## Verification - `closed-native-receipt-accounting.test.ts` (embedded Postgres): - Lost Codex (gpt-6.1-sol) and Claude (claude-sonnet-5) receipts with observed tokens settle as `estimated` rate-card charges with the lower-bound evidence. The company stays active and admission is clear under the blocking policy. - A lost receipt for a model with no rate card stays unpriced but opens no incident and no pause. - A task's continuation reserves budget past its own recovering accounting, while another task's run is still held. - Unpriced charges with no tokens on metered billing, or with a known partial cost, do not hard-stop. - Complete-receipt usage with no price still blocks. - The six new cases fail on master and pass with this change. - Existing budget and accounting suites pass locally: budget-enforcement-scoping, budgets-service, heartbeat-cost-accounting, run-cost-accounting, native-stop-accounting, cost-data-quality, budget-live-attempts, budget-live-usage. Server typecheck passes. ## Risks - A lost receipt's estimate is a lower bound. The provenance says so, and a later complete receipt still reconciles. - Restart-derived unpriced charges no longer trip the blocking policy. Usage lost with a killed provider is not counted against that policy, which is consistent with budget enforcement being best effort. Unknown-billing and complete-receipt unpriced usage still block. - The same-task exemption relies on `native_issue_id`. Legacy runs without it keep the old behavior. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, via Claude Code with tool use (shell, file editing, embedded Postgres tests). ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 --- doc/SPEC-implementation.md | 7 ++ .../closed-native-receipt-accounting.test.ts | 119 +++++++++++++++++- server/src/services/budget-exposure.ts | 42 +++++-- server/src/services/budget-reservations.ts | 6 +- server/src/services/budgets.ts | 15 +-- server/src/services/run-cost-accounting.ts | 47 ++++++- 6 files changed, 206 insertions(+), 30 deletions(-) diff --git a/doc/SPEC-implementation.md b/doc/SPEC-implementation.md index 922c5cfbe17..fd7c9cca5fb 100644 --- a/doc/SPEC-implementation.md +++ b/doc/SPEC-implementation.md @@ -1890,6 +1890,13 @@ termination receipt (stopped or destroyed); a sandbox that may still run keeps the debt pending. Accounting then records its known part as an unpriced charge, so the unpriced-usage policy governs admission. A later receipt still posts through late-receipt reconciliation. +When a lost receipt's observed usage has a known model and token counts, the +rate card prices it as an estimate whose provenance marks the usage as a lower +bound. A charge settled from a lost receipt never trips the blocking +unpriced-usage policy by itself, and neither does an unpriced charge with a +known partial cost or with no observed tokens on metered billing. A task's own +continuation and finalization are never held by that task's pending +accounting. A controller that detaches for server restart loses checkpoint and completion write authority. A closing event stream is not proof that the governed run finished; the replacement controller must adopt and settle the original run. diff --git a/server/src/__tests__/closed-native-receipt-accounting.test.ts b/server/src/__tests__/closed-native-receipt-accounting.test.ts index 15bb4b01243..e7aaaf80598 100644 --- a/server/src/__tests__/closed-native-receipt-accounting.test.ts +++ b/server/src/__tests__/closed-native-receipt-accounting.test.ts @@ -7,9 +7,10 @@ import { nativeRunFinalizations, type Db, } from "@paperclipai/db"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; -import { budgetService, flushBudgetEvaluationWork } from "../services/budgets.js"; +import { budgetService, flushBudgetEvaluationWork, recheckAdmittedRunBudgetReadOnly } from "../services/budgets.js"; import { accountRunCost, reconcileRunCosts } from "../services/run-cost-accounting.js"; import { persistUsageReceipt } from "../services/usage-receipts.js"; +import { recheckRunBudgetReservationInTransaction, reserveRunBudget } from "../services/budget-reservations.js"; // A control-plane SIGKILL during a provider turn that reports usage only at // turn end leaves the run terminal with an incomplete receipt. Once native @@ -115,9 +116,121 @@ const support = await getEmbeddedPostgresTestSupport(); expect(await f.block()).toBeNull(); }); - it("keeps a blocking policy for lost-receipt usage that has tokens but no price", async () => { + // RC-06d: a crash loop killed a Codex turn whose observed usage was complete + // token counts. The lost receipt settled as unpriced and, under the default + // blocking policy, hard-stopped the company at 0.38 of 3000 cents. + async function lostUsage(f: Awaited>, usage: Record) { + await db.update(heartbeatRuns).set({ usageJson: sql`${heartbeatRuns.usageJson} || ${JSON.stringify(usage)}::jsonb` }) + .where(eq(heartbeatRuns.id, f.run.id)); + } + + it.each([ + ["Codex", { provider: "openai", biller: "openai", model: "gpt-6.1-sol", billingType: "metered_api", + inputTokens: 28690, cachedInputTokens: 0, outputTokens: 220 }], + ["Claude", { provider: "anthropic", biller: "anthropic", model: "claude-sonnet-5", billingType: "metered_api", + inputTokens: 1200, cachedInputTokens: 300, cacheWriteTokens: 100, outputTokens: 80 }], + ] as const)("prices a lost %s receipt's observed usage from the rate card", async (_label, usage) => { const f = await fixture({ unpricedUsagePolicy: "block" }); - await db.update(heartbeatRuns).set({ usageJson: sql`${heartbeatRuns.usageJson} || '{"inputTokens":1200,"outputTokens":80}'::jsonb` }) + await lostUsage(f, usage); + expect(await accountRunCost(db, f.run.id)).toBe(true); + await budgetService(db).evaluateAffectedScopes(f.companyId, f.agentId, null); + await flushBudgetEvaluationWork(db); + const [charge] = await f.charges(); + expect(charge).toMatchObject({ costStatus: "estimated", model: usage.model }); + expect(Number(charge.costCents)).toBeGreaterThan(0); + // Observed usage before the loss is a lower bound, and says so. + expect(charge.pricingProvenance).toMatchObject({ source: "rate_card", + evidence: expect.stringContaining("observed usage is a lower bound") }); + const [company] = await db.select().from(companies).where(eq(companies.id, f.companyId)); + expect(company).toMatchObject({ status: "active", pauseReason: null }); + expect(await f.block()).toBeNull(); + }); + + it("leaves a lost multi-model receipt unpriced instead of applying one model's rate", async () => { + const f = await fixture({ unpricedUsagePolicy: "block" }); + await lostUsage(f, { provider: "openai", biller: "openai", model: "gpt-6.1-sol", billingType: "metered_api", + inputTokens: 28690, cachedInputTokens: 0, outputTokens: 220, + usageByModel: [{ model: "gpt-6.1-sol", usage: { inputTokens: 20000 } }, { model: "gpt-6-luna", usage: { inputTokens: 8690 } }] }); + expect(await accountRunCost(db, f.run.id)).toBe(true); + expect(await f.charges()).toMatchObject([{ costStatus: "unpriced" }]); + expect(await f.block()).toBeNull(); + }); + + it("never hard-stops a company for a lost receipt whose usage has no rate card", async () => { + const f = await fixture({ unpricedUsagePolicy: "block" }); + await lostUsage(f, { provider: "openai", biller: "openai", model: "unlisted-model", inputTokens: 1200, outputTokens: 80 }); + expect(await accountRunCost(db, f.run.id)).toBe(true); + await budgetService(db).evaluateAffectedScopes(f.companyId, f.agentId, null); + await flushBudgetEvaluationWork(db); + expect(await f.charges()).toMatchObject([{ costStatus: "unpriced" }]); + const [company] = await db.select().from(companies).where(eq(companies.id, f.companyId)); + expect(company).toMatchObject({ status: "active", pauseReason: null }); + expect(await db.select().from(budgetIncidents).where(eq(budgetIncidents.companyId, f.companyId))).toHaveLength(0); + expect(await f.block()).toBeNull(); + }); + + // RC-06e: a resumed run's own continuation was refused by the accounting hold + // of its task's interrupted run, so the task could never settle itself. + it("admits a task's continuation past its own recovering accounting, but not other tasks", async () => { + const f = await fixture({ phase: "observed" }); + expect(await f.block()).toMatchObject({ reason: "New work must wait while a native run recovers its unfinished accounting." }); + const continuation = randomUUID(); + await db.insert(heartbeatRuns).values({ id: continuation, companyId: f.companyId, agentId: f.agentId, status: "queued", + runtimeMode: "native", scopeKind: "issue", issueId: f.issue.id, nativeIssueId: f.issue.id }); + const reservation = await reserveRunBudget(db, f.companyId, continuation, null); + expect(reservation).toMatchObject({ reused: false }); + // The later dispatch gates must agree, or the continuation fails there instead. + const [continuationRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, continuation)); + expect(await db.transaction(tx => recheckRunBudgetReservationInTransaction(tx as unknown as typeof db, [], + continuationRun, reservation.id))).toBeNull(); + expect(await recheckAdmittedRunBudgetReadOnly(db, { companyId: f.companyId, agentId: f.agentId, runId: continuation, + projectId: null, reservationId: reservation.id })).toBeNull(); + // A legacy run without a task identity is not exempt from another task's hold. + const legacy = randomUUID(); + await db.insert(heartbeatRuns).values({ id: legacy, companyId: f.companyId, agentId: f.otherAgentId, status: "queued" }); + await expect(reserveRunBudget(db, f.companyId, legacy, null)).rejects.toThrow(/recovers its unfinished accounting/); + const [otherIssue] = await db.insert(issues).values({ companyId: f.companyId, title: "Other task", status: "todo", + assigneeAgentId: f.otherAgentId }).returning(); + const other = randomUUID(); + await db.insert(heartbeatRuns).values({ id: other, companyId: f.companyId, agentId: f.otherAgentId, status: "queued", + runtimeMode: "native", scopeKind: "issue", issueId: otherIssue.id, nativeIssueId: otherIssue.id }); + await expect(reserveRunBudget(db, f.companyId, other, null)).rejects.toThrow(/recovers its unfinished accounting/); + }); + + // RC-06e: unpriced charges with no observed usage, or with a known partial + // cost, hard-stopped a company at 120 of 2000 cents. + it.each([ + ["no observed tokens on metered billing", { billingType: "metered_api", inputTokens: 0, outputTokens: 0, costUsdExact: null }], + ["a known partial cost", { billingType: "metered_api", inputTokens: 900, outputTokens: 40, costUsdExact: "0.0081" }], + ] as const)("does not hard-stop a blocking policy for an unpriced charge with %s", async (_label, usage) => { + const f = await fixture({ unpricedUsagePolicy: "block", stopEvidence: false }); + await db.update(heartbeatRuns).set({ usageJson: sql`${heartbeatRuns.usageJson} || ${JSON.stringify({ + ...usage, model: "unlisted-model", costStatus: "unpriced", accountingReceiptReady: true })}::jsonb` }) + .where(eq(heartbeatRuns.id, f.run.id)); + await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.run.id)); + expect(await accountRunCost(db, f.run.id)).toBe(true); + expect(await f.charges()).toMatchObject([{ costStatus: "unpriced" }]); + await budgetService(db).evaluateAffectedScopes(f.companyId, f.agentId, null); + await flushBudgetEvaluationWork(db); + const [company] = await db.select().from(companies).where(eq(companies.id, f.companyId)); + expect(company).toMatchObject({ status: "active", pauseReason: null }); + expect(await f.block()).toBeNull(); + }); + + it("keeps the blocking policy for fixed or credit billing with no observed tokens", async () => { + const f = await fixture({ unpricedUsagePolicy: "block", stopEvidence: false }); + await db.update(heartbeatRuns).set({ usageJson: sql`${heartbeatRuns.usageJson} || ${JSON.stringify({ + billingType: "credits", inputTokens: 0, outputTokens: 0, model: "unlisted-model", costStatus: "unpriced", + accountingReceiptReady: true })}::jsonb` }).where(eq(heartbeatRuns.id, f.run.id)); + await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.run.id)); + expect(await accountRunCost(db, f.run.id)).toBe(true); + await flushBudgetEvaluationWork(db); + expect((await f.block())?.reason).toMatch(/no reliable price|budget hard-stop/); + }); + + it("keeps the blocking policy for complete-receipt usage with no price", async () => { + const f = await fixture({ unpricedUsagePolicy: "block" }); + await db.update(heartbeatRuns).set({ usageJson: sql`${heartbeatRuns.usageJson} || '{"accountingReceiptReady":true,"inputTokens":1200,"outputTokens":80,"model":"unlisted-model"}'::jsonb` }) .where(eq(heartbeatRuns.id, f.run.id)); expect(await accountRunCost(db, f.run.id)).toBe(true); await flushBudgetEvaluationWork(db); diff --git a/server/src/services/budget-exposure.ts b/server/src/services/budget-exposure.ts index 960533c3b1b..3a542eb38fd 100644 --- a/server/src/services/budget-exposure.ts +++ b/server/src/services/budget-exposure.ts @@ -70,7 +70,7 @@ const liveCost = sql`case when ${heartbeatRuns.costAccountedAt} is null and ${he /** Batched ledger, held-operation and unfinished-run aggregates. Never scans * agents or receipt history; each policy retains its own scope and window. */ -async function readObservedSpends(db: Db, policies: SpendPolicy[], now: Date): Promise { +async function readObservedSpends(db: Db, policies: SpendPolicy[], now: Date, excludeRunId?: string | null): Promise { const windows = policies.map(policy => budgetWindow(policy.windowKind, now)); const billable = policies.map((policy, index) => ({ policy, index })).filter(({ policy }) => policy.metric === "billed_cents"); const ledgerColumns: Record = {}, heldColumns: Record = {}, runColumns: Record = {}, browserColumns: Record = {}; @@ -117,7 +117,15 @@ async function readObservedSpends(db: Db, policies: SpendPolicy[], now: Date): P const run = and(eq(heartbeatRuns.companyId, policy.companyId), or(eq(heartbeatRuns.costAccountingPending, true), lateReceiptDebt), policy.scopeType === "agent" ? eq(heartbeatRuns.agentId, policy.scopeId) : undefined, policy.scopeType === "project" ? sql`${usage}->'ledgerScope'->>'projectId' = ${policy.scopeId}` : undefined)!; - const terminal = and(run, inArray(heartbeatRuns.status, ["succeeded", "failed", "timed_out", "cancelled", "interrupted"]), or(inWindow, lateReceiptDebt))!; + // A task never waits on its own unfinished accounting: the run being + // admitted and earlier runs of the same task (its recovery generation) are + // what its continuation and finalization settle. They still count as spend. + // coalesce keeps this a definite boolean when the admitted run has no task. + const ownTask = excludeRunId ? sql`coalesce(${heartbeatRuns.id} = ${excludeRunId}::uuid or (${heartbeatRuns.nativeIssueId} is not null + and ${heartbeatRuns.nativeIssueId} = (select admitted_run.native_issue_id from ${heartbeatRuns} admitted_run + where admitted_run.id = ${excludeRunId}::uuid and admitted_run.company_id = ${heartbeatRuns.companyId})), false)` : sql`false`; + const terminal = and(run, inArray(heartbeatRuns.status, ["succeeded", "failed", "timed_out", "cancelled", "interrupted"]), or(inWindow, lateReceiptDebt), + sql`not ${ownTask}`)!; // Keep correlated subqueries in nested SQL chunks. Drizzle strips column // qualifiers from a single-table selection's direct column chunks. const hasReservation = sql`exists (select 1 from ${budgetReservations} @@ -135,16 +143,24 @@ async function readObservedSpends(db: Db, policies: SpendPolicy[], now: Date): P and (${nativeRunFinalizations.phase} <> 'terminal_failure' or (${closingReceipt})))`; ledgerPredicates.push(ledger); heldPredicates.push(held); runPredicates.push(run); ledgerColumns[`total${index}`] = sql`coalesce(sum(${costEvents.costCents}) filter (where ${ledger}), 0)::text`; - // A receipt lost with its verified-stopped provider settles as an unpriced - // charge. With no observed usage it is unknown, not usage without a price: - // it stays visible in Costs but does not trip the policy. Complete - // receipts that report no tokens (for example Cursor Cloud) still count. - const lostReceiptWithoutUsage = sql`(${costEvents.inputTokens} = 0 and ${costEvents.cachedInputTokens} = 0 and ${costEvents.outputTokens} = 0 - and exists (select 1 from ${heartbeatRuns} lost_receipt_run where lost_receipt_run.id = ${costEvents.heartbeatRunId} - and lost_receipt_run.company_id = ${costEvents.companyId} - and lost_receipt_run.usage_json->>'accountingReceiptUnavailable' = 'true'))`; + // "Usage with no reliable price" means observed usage that could not be + // priced. These unpriced charges are not that, so they never trip the + // blocking policy (they stay visible in Costs): + // - a receipt lost with its stopped provider (restart-derived, any usage); + // - a charge that carries a known partial cost; + // - metered API or subscription billing with no observed tokens. + // Unknown billing with no tokens (for example Cursor Cloud) still counts. + // The lost-receipt exemption covers only that run's settled final charge; + // later growth from a complete receipt is ordinary usage. + const exemptFromUnpricedBlock = sql`((${costEvents.idempotencyKey} = 'heartbeat:' || ${costEvents.heartbeatRunId}::text || ':final' + and exists (select 1 from ${heartbeatRuns} lost_receipt_run where lost_receipt_run.id = ${costEvents.heartbeatRunId} + and lost_receipt_run.company_id = ${costEvents.companyId} + and lost_receipt_run.usage_json->>'accountingReceiptUnavailable' = 'true')) + or ${costEvents.costCents} > 0 + or (${costEvents.billingType} in ('metered_api', 'api', 'subscription') and ${costEvents.inputTokens} = 0 and ${costEvents.cachedInputTokens} = 0 + and ${costEvents.outputTokens} = 0))`; ledgerColumns[`unpriced${index}`] = sql`count(*) filter (where ${ledger} and ${costEvents.costStatus} = 'unpriced' and ${costEvents.billingType} <> 'subscription_included' - and not ${lostReceiptWithoutUsage})::int`; + and not ${exemptFromUnpricedBlock})::int`; runColumns[`live${index}`] = sql`coalesce(sum(${liveCost}) filter (where ${run} and ${inWindow}), 0)::text`; runColumns[`unreserved${index}`] = sql`coalesce(sum(${liveCost}) filter (where ${run} and not ${hasReservation}), 0)::text`; // Reservations carry over UTC windows. A reservation and its running @@ -191,14 +207,14 @@ export async function computeObservedSpend(db: Db, policy: SpendPolicy, now = ne return (await computeObservedSpends(db, [policy], now))[0]; } -export function computeObservedSpends(db: Db, policies: SpendPolicy[], now = new Date()): Promise { +export function computeObservedSpends(db: Db, policies: SpendPolicy[], now = new Date(), excludeRunId?: string | null): Promise { return withBudgetSpan("budget.exposure.evaluate", async () => { // Each policy contributes several aggregate columns. Bound statement width // below PostgreSQL's target-list limit. Multi-scope evaluators/reports own // a repeatable-read transaction; individual admission batches fit one SQL. const result: ObservedSpend[] = []; for (let offset = 0; offset < policies.length; offset += 128) - result.push(...await readObservedSpends(db, policies.slice(offset, offset + 128), now)); + result.push(...await readObservedSpends(db, policies.slice(offset, offset + 128), now, excludeRunId)); return result; }, { "budget.policy_count": policies.length }); } diff --git a/server/src/services/budget-reservations.ts b/server/src/services/budget-reservations.ts index 95228a8d021..45ef666e3cc 100644 --- a/server/src/services/budget-reservations.ts +++ b/server/src/services/budget-reservations.ts @@ -54,7 +54,7 @@ export async function recheckRunBudgetReservationInTransaction( } if (stopRecovery) return null; const { block, policies, observations } = await budgetServiceInTransaction(tx, publications) - .evaluateAdmission(run.companyId, run.agentId, { projectId: reservation.projectId }); + .evaluateAdmission(run.companyId, run.agentId, { projectId: reservation.projectId, runId: run.id }); if (block) return block.reason; const [agent] = await tx.select({ status: agents.status, lifecycleState: agents.lifecycleState }).from(agents) .where(and(eq(agents.companyId, run.companyId), eq(agents.id, run.agentId))); @@ -102,7 +102,7 @@ export async function reserveRunBudget(db: Db, companyId: string, runId: string, // Recovery has already made this run queued/running, so its own pending // accounting is excluded. Current overruns, other unfinished runs and // scope pauses still apply before any more paid work starts. - const block = await budgetServiceInTransaction(tx, publications).getInvocationBlock(companyId, run.agentId, { projectId }); + const block = await budgetServiceInTransaction(tx, publications).getInvocationBlock(companyId, run.agentId, { projectId, runId }); if (block) return { admissionBlockReason: block.reason }; const [agent] = await tx.select({ status: agents.status, lifecycleState: agents.lifecycleState }).from(agents) .where(and(eq(agents.companyId, companyId), eq(agents.id, run.agentId))); @@ -113,7 +113,7 @@ export async function reserveRunBudget(db: Db, companyId: string, runId: string, throw conflict("Provider dispatch has already reserved this run"); } if (stopRecovery) throw conflict("Native Stop recovery requires the original held reservation"); - const { block, policies, observations } = await budgetServiceInTransaction(tx, publications).evaluateAdmission(companyId, run.agentId, { projectId }); + const { block, policies, observations } = await budgetServiceInTransaction(tx, publications).evaluateAdmission(companyId, run.agentId, { projectId, runId }); if (block) return { admissionBlockReason: block.reason }; // A budget release can start asynchronous resume hooks. Capacity alone // does not authorize dispatch before that lifecycle operation is ready. diff --git a/server/src/services/budgets.ts b/server/src/services/budgets.ts index b7868af8d24..5e1654ceba7 100644 --- a/server/src/services/budgets.ts +++ b/server/src/services/budgets.ts @@ -200,7 +200,8 @@ export function budgetPoliciesForRun(companyId: string, agentId: string | null, ); } -type AdmissionContext = { issueId?: string | null; projectId?: string | null }; +/** `runId` names the run being admitted, so its own pending accounting never holds it. */ +type AdmissionContext = { issueId?: string | null; projectId?: string | null; runId?: string | null }; async function readAdmissionScopes(db: Db, companyId: string, agentId: string | null, context?: AdmissionContext) { const scopes: Array<{ scopeType: BudgetScopeType; scopeId: string }> = [ @@ -228,7 +229,7 @@ async function inspectInvocationEligibility(db: Db, companyId: string, agentId: const { scopes, records, policies } = await readAdmissionScopes(db, companyId, agentId, context); if (policies.length > 0) { if (records.some(scope => scope.budgetHeld)) return { needsReconciliation: true } as const; - const observations = await computeObservedSpends(db, policies); + const observations = await computeObservedSpends(db, policies, undefined, context?.runId); if (policies.some((policy, index) => observedBlocks(policy, observations[index]) || (policy.isActive && policy.hardStopEnabled && policy.amount > 0 && observations[index].recoveringRunCount > 0))) { return { needsReconciliation: true } as const; @@ -298,7 +299,7 @@ export async function recheckAdmittedRunBudgetReadOnly(db: Db, input: { const policies = await db.select().from(budgetPolicies).where(and(eq(budgetPolicies.companyId, input.companyId), eq(budgetPolicies.isActive, true), eq(budgetPolicies.hardStopEnabled, true), eq(budgetPolicies.metric, "billed_cents"), sql`${budgetPolicies.amount} > 0`, budgetPoliciesForRun(input.companyId, input.agentId, input.projectId))); - const observations = await computeObservedSpends(db, policies); + const observations = await computeObservedSpends(db, policies, undefined, input.runId); for (const [index, policy] of policies.entries()) { const observed = observations[index]; if (observedBlocks(policy, observed)) return "Applicable budget no longer permits provider work"; @@ -819,9 +820,9 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica } async function evaluateAdmissionInTransaction(companyId: string, agentId: string | null, - context?: { issueId?: string | null; projectId?: string | null }) { + context?: AdmissionContext) { const { scopes, records, policies } = await readAdmissionScopes(db, companyId, agentId, context); - const observations = await computeObservedSpends(db, policies, now); + const observations = await computeObservedSpends(db, policies, now, context?.runId); const reservationIndexes = policies.map((policy, i) => policy.isActive && policy.hardStopEnabled && policy.metric === "billed_cents" && policy.amount > 0 ? i : -1).filter(i => i >= 0); const admission = (block: Awaited>) => ({ block, policies: reservationIndexes.map(i => policies[i]), observations: reservationIndexes.map(i => observations[i]) }); @@ -1044,7 +1045,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica : queueBudgetEvaluation(db, { companyId: event.companyId, agentId: event.agentId, projectId: event.projectId }), evaluateAdmission, - getInvocationBlock: async (companyId: string, agentId: string | null, context?: { issueId?: string | null; projectId?: string | null }) => + getInvocationBlock: async (companyId: string, agentId: string | null, context?: AdmissionContext) => (await evaluateAdmission(companyId, agentId, context)).block, resolveIncident: async ( @@ -1387,7 +1388,7 @@ export function budgetService(db: Db, _hooks: BudgetServiceHooks = {}) { tx => budgetServiceInTransaction(tx).listPolicies(companyId)), overview: (companyId: string) => withAccountingReadSnapshot(db, companyId, tx => budgetServiceInTransaction(tx).overview(companyId)), - evaluateAdmission: (companyId: string, agentId: string | null, context?: { issueId?: string | null; projectId?: string | null }) => + evaluateAdmission: (companyId: string, agentId: string | null, context?: AdmissionContext) => withAccountingReadSnapshot(db, companyId, tx => budgetServiceInTransaction(tx).evaluateAdmission(companyId, agentId, context)), evaluateAffectedScopes: (companyId: string, agentId: string | null, projectId: string | null) => withOperationAccountingTransaction(db, companyId, tx => queueBudgetEvaluation(tx, { companyId, agentId, projectId })), diff --git a/server/src/services/run-cost-accounting.ts b/server/src/services/run-cost-accounting.ts index cbd92ff10e7..69c56bdeeca 100644 --- a/server/src/services/run-cost-accounting.ts +++ b/server/src/services/run-cost-accounting.ts @@ -11,6 +11,9 @@ import { queueBudgetEvaluation } from "./budget-evaluation-work.js"; import { ensureRuntimeAccountingBaseline } from "./runtime-accounting-totals.js"; import { logger } from "../middleware/logger.js"; import { promises as fs } from "node:fs"; +import { priceAnthropicReceipt } from "./anthropic-pricing.js"; +import { priceCodexReceipt } from "./codex-pricing.js"; +import type { AdapterUsageCheckpoint } from "@paperclipai/adapter-utils"; import { replayUsageReceipts, indexPendingUsageReceipts, readPendingRunUsageReceipts, persistUsageReceiptInTransaction, reconcileLateRunUsageReceiptInTransaction, type UsageReceiptIndex } from "./usage-receipts.js"; const terminalStatuses = ["succeeded", "failed", "timed_out", "cancelled", "interrupted"]; @@ -89,6 +92,35 @@ async function nativeReceiptSourceClosed(db: Db, runId: string) { return row?.closed === true; } +/** A receipt lost with its stopped provider still carries the usage observed + * before the loss. When the model and token counts are known, price them from + * the rate card as an estimate marked as a lower bound, instead of leaving + * known numbers unpriced. Zero observed tokens stay unpriced: unknown is never + * presented as zero cost. */ +function estimateLostReceipt(usage: Record) { + const count = (key: string) => amount(usage[key]); + const inputTokens = count("inputTokens") ?? 0, outputTokens = count("outputTokens") ?? 0; + const cachedInputTokens = count("cachedInputTokens"), cacheWriteTokens = count("cacheWriteTokens"); + if (inputTokens + outputTokens + (cachedInputTokens ?? 0) <= 0) return null; + if ((usage.usageSource ?? "per_run") !== "per_run") return null; + // Aggregate totals across several models cannot take one model's rate. + if (Array.isArray(usage.usageByModel) && usage.usageByModel.length > 0) return null; + const billingType = text(usage.billingType); + const priced = priceAnthropicReceipt(priceCodexReceipt({ + complete: true, usageBasis: "per_run", provider: text(usage.provider) ?? undefined, + biller: text(usage.biller) ?? undefined, model: text(usage.model) ?? undefined, + billingType: billingType === "metered_api" ? "api" : billingType ?? undefined, + usage: { inputTokens, outputTokens, + ...(cachedInputTokens !== null ? { cachedInputTokens } : {}), + ...(cacheWriteTokens !== null ? { cacheWriteTokens } : {}) }, + } as AdapterUsageCheckpoint)); + const exact = text(priced.costUsdExact); + if (exact === null || !priced.pricingProvenance) return null; + const evidence = text(priced.pricingProvenance.evidence); + return { costUsdExact: exact, pricingProvenance: { ...priced.pricingProvenance, + evidence: [evidence, "final receipt lost; observed usage is a lower bound"].filter(Boolean).join("; ") } }; +} + /** The run's persisted usage is a durable receipt. Ledger, reservation handoff * and acknowledgement commit together, including after a restart or failed run. */ export async function accountRunCost(db: Db, runId: string, hooks: BudgetServiceHooks = {}, receiptIndex?: UsageReceiptIndex) { @@ -180,7 +212,9 @@ export async function accountRunCost(db: Db, runId: string, hooks: BudgetService if (!linked) scope[key] = null; } } - const costUsd = amount(usage.cacheAdjustedCostUsd) ?? text(usage.costUsdExact) ?? amount(usage.costUsd); + const knownCostUsd = amount(usage.cacheAdjustedCostUsd) ?? text(usage.costUsdExact) ?? amount(usage.costUsd); + const lostReceiptEstimate = receiptUnavailable && knownCostUsd === null ? estimateLostReceipt(usage) : null; + const costUsd = knownCostUsd ?? lostReceiptEstimate?.costUsdExact ?? null; const billingType = text(usage.billingType) ?? "unknown"; const costUnits = billingType === "subscription_included" ? 0n : usdToUnits(costUsd ?? 0); const costCents = unitsToCents(costUnits); @@ -194,9 +228,9 @@ export async function accountRunCost(db: Db, runId: string, hooks: BudgetService issueId: text(scope.issueId), projectId: text(scope.projectId), billingCode: text(scope.billingCode), provider: text(usage.provider) ?? "unknown", biller: text(usage.biller) ?? text(usage.provider) ?? "unknown", providerRequestId: text(usage.providerRequestId), - pricingProvenance: object(usage.pricingProvenance), + pricingProvenance: lostReceiptEstimate?.pricingProvenance ?? object(usage.pricingProvenance), model: text(usage.model) ?? "unknown", billingType, - costStatus: (costUsd === null || usage.costStatus === "unpriced" || receiptUnavailable) && billingType !== "subscription_included" && hasBillableEvidence ? "unpriced" : usage.costStatus === "estimated" ? "estimated" : "reported", + costStatus: lostReceiptEstimate ? "estimated" : (costUsd === null || usage.costStatus === "unpriced" || receiptUnavailable) && billingType !== "subscription_included" && hasBillableEvidence ? "unpriced" : usage.costStatus === "estimated" ? "estimated" : "reported", inputTokens: amount(usage.inputTokens) ?? 0, cachedInputTokens: amount(usage.cachedInputTokens) ?? 0, outputTokens: amount(usage.outputTokens) ?? 0, costCents, occurredAt: run.finishedAt ?? run.createdAt, }; @@ -231,7 +265,12 @@ export async function accountRunCost(db: Db, runId: string, hooks: BudgetService await tx.update(heartbeatRuns).set({ costAccountingPending: lateDebt, costAccountedAt: new Date(), accountingProjectionVersion: "v2", accountingLastError: lateDebt ? "late_usage_requires_reconciliation" : null, accountingLastAttemptAt: new Date(), // A complete receipt arriving later may price this charge (see usage-receipts). - ...(receiptUnavailable ? { usageJson: sql`coalesce(${heartbeatRuns.usageJson}, '{}'::jsonb) || '{"accountingReceiptUnavailable":true}'::jsonb` } : {}), + ...(receiptUnavailable ? { usageJson: sql`coalesce(${heartbeatRuns.usageJson}, '{}'::jsonb) || ${JSON.stringify({ + accountingReceiptUnavailable: true, + // Late reconciliation measures a later receipt from the accepted money. + ...(lostReceiptEstimate ? { costUsdExact: lostReceiptEstimate.costUsdExact, costStatus: "estimated", + pricingProvenance: lostReceiptEstimate.pricingProvenance } : {}), + })}::jsonb` } : {}), }).where(eq(heartbeatRuns.id, run.id)); return true; }), { "budget.accounting.owner": "run" });