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