diff --git a/server/src/__tests__/closed-native-receipt-accounting.test.ts b/server/src/__tests__/closed-native-receipt-accounting.test.ts index b2c2ecce72e..607e9f29806 100644 --- a/server/src/__tests__/closed-native-receipt-accounting.test.ts +++ b/server/src/__tests__/closed-native-receipt-accounting.test.ts @@ -620,6 +620,95 @@ const support = await getEmbeddedPostgresTestSupport(); expect(await f.block()).toBeNull(); }); + async function captureFailed(f: { run: { id: string } }, finishedMinutesAgo: number, extra: Record = {}) { + await db.update(heartbeatRuns).set({ finishedAt: new Date(Date.now() - finishedMinutesAgo * 60_000), + usageJson: sql`${heartbeatRuns.usageJson} || ${JSON.stringify({ accountingCaptureFailed: true, accountingReceiptReady: false, ...extra })}::jsonb` }) + .where(eq(heartbeatRuns.id, f.run.id)); + } + + // A failed usage capture never completes its receipt. It used to be skipped + // by accounting and held company admission until the UTC month rolled over. + it("settles a capture failure as unpriced at the closure bound and releases held work", async () => { + const f = await fixture(); + await captureFailed(f, 11, { inputTokens: 1200, outputTokens: 80 }); + expect(await f.block()).not.toBeNull(); + expect((await readCostAccountingWork(db, Date.now())).pending).toBe(true); + expect(await reconcileRunCosts(db)).toMatchObject({ accounted: 1 }); + await flushBudgetEvaluationWork(db); + const run = await f.readRun(); + expect(run).toMatchObject({ costAccountingPending: false }); + expect(run.usageJson).toMatchObject({ accountingReceiptUnavailable: true, accountingClosureTimedOut: true }); + const [charge] = await f.charges(); + expect(charge).toMatchObject({ inputTokens: 1200, outputTokens: 80 }); + expect(charge?.costStatus).not.toBe("reported"); + expect(JSON.stringify(charge?.pricingProvenance)).toContain("usage capture failed"); + expect(await f.block()).toBeNull(); + }); + + it("waits for the closure bound before settling a capture failure", async () => { + const f = await fixture(); + await captureFailed(f, 1); + expect(await reconcileRunCosts(db)).toMatchObject({ accounted: 0 }); + expect((await f.readRun()).costAccountingPending).toBe(true); + const next = (await readCostAccountingWork(db, Date.now())).nextRunAt; + expect(next).not.toBeNull(); + expect(next!).toBeGreaterThan(Date.now()); + }); + + it("keeps a settled capture failure under the blocking unpriced-usage policy", async () => { + const f = await fixture({ unpricedUsagePolicy: "block" }); + await captureFailed(f, 11, { inputTokens: 1200, outputTokens: 80, model: "unknown-model" }); + expect(await reconcileRunCosts(db)).toMatchObject({ accounted: 1 }); + await flushBudgetEvaluationWork(db); + expect((await f.block())?.reason).toMatch(/no reliable price|budget hard-stop/); + }); + + it("keeps a capture failure after a receipt-source handoff under the blocking policy", async () => { + const f = await fixture({ unpricedUsagePolicy: "block" }); + await captureFailed(f, 11, { inputTokens: 1200, outputTokens: 80, model: "unknown-model", + accountingReceiptSourceFences: { [f.sourceId]: 2 } }); + expect(await reconcileRunCosts(db)).toMatchObject({ accounted: 1 }); + await flushBudgetEvaluationWork(db); + expect((await f.block())?.reason).toMatch(/no reliable price|budget hard-stop/); + }); + + it("settles a capture failure at no less than its observed live price", async () => { + const f = await fixture(); + await captureFailed(f, 11, { inputTokens: 1200, outputTokens: 80, model: "unknown-model", costUsd: null, + budgetExposure: { sourceId: f.sourceId, sequence: 2, costCents: "600.0000000" } }); + expect(await reconcileRunCosts(db)).toMatchObject({ accounted: 1 }); + await flushBudgetEvaluationWork(db); + const [charge] = await f.charges(); + expect(Number(charge?.costCents)).toBeGreaterThanOrEqual(600); + expect(charge?.costStatus).toBe("estimated"); + // 600 cents exceeds the 500-cent policy: the hard stop still holds. + expect((await f.block())?.reason).toMatch(/hard-stop|budget/i); + }); + + it.each([["no raw cost", {}], ["a lower cache-adjusted cost", { cacheAdjustedCostUsd: 5 }]] as const)( + "settles a lost receipt at no less than its observed live price (%s)", async (_name, raw) => { + const f = await fixture(); + await db.update(heartbeatRuns).set({ usageJson: sql`${heartbeatRuns.usageJson} || ${JSON.stringify({ + ...raw, inputTokens: 1200, outputTokens: 80, model: "unknown-model", costUsd: null, + budgetExposure: { sourceId: f.sourceId, sequence: 2, costCents: "600.0000000" } })}::jsonb` }).where(eq(heartbeatRuns.id, f.run.id)); + expect(await reconcileRunCosts(db)).toMatchObject({ accounted: 1 }); + const [charge] = await f.charges(); + expect(Number(charge?.costCents)).toBeGreaterThanOrEqual(600); + expect(charge?.costStatus).toBe("estimated"); + // A complete receipt arriving later posts only its growth over the accepted price. + await persistUsageReceipt(db, { schema: "paperclip/accounting-receipt/v1", id: randomUUID(), + companyId: f.companyId, runId: f.run.id, adapterType: "paperclip_runner", sourceId: f.sourceId, sequence: 6, + receivedAt: new Date().toISOString(), + producerEvent: { sourceEventId: randomUUID(), eventType: "turn.completed" }, + receipt: { complete: true, usageBasis: "per_run", provider: "anthropic", biller: "anthropic", + model: "unknown-model", billingType: "metered_api", costUsd: 7, costStatus: "reported", + pricingProvenance: { source: "provider_reported", version: "test" }, + usage: { inputTokens: 2400, cachedInputTokens: 0, outputTokens: 160 } } }); + await accountRunCost(db, f.run.id); + const total = (await f.charges()).reduce((sum, row) => sum + Number(row.costCents), 0); + expect(total).toBeCloseTo(700, 5); + }); + it("never schedules a wake for a bound that already passed or for a capture failure", async () => { const past = await acceptedResultFixture({ stopEvidence: false, finishedMinutesAgo: 11 }); await db.update(heartbeatRuns).set({ usageJson: sql`${heartbeatRuns.usageJson} || '{"accountingCaptureFailed":true}'::jsonb` }) diff --git a/server/src/__tests__/cost-data-quality.test.ts b/server/src/__tests__/cost-data-quality.test.ts index d36b9a13961..02cabc5728b 100644 --- a/server/src/__tests__/cost-data-quality.test.ts +++ b/server/src/__tests__/cost-data-quality.test.ts @@ -166,9 +166,16 @@ describe("actionable cost data quality", () => { expect(report.runs.find(item => item.id === captureFailed.id)?.reason).toBe("waiting_for_receipt"); expect(report.runs.find(item => item.id === invalidProof.id)?.reason).toBe("waiting_for_receipt"); const integrity = accountingIntegrityService(db); + // A failed capture is never certified free. Before the accounting closure + // bound it stays pending; past it, its known usage settles as unpriced. + await db.update(heartbeatRuns).set({ finishedAt: new Date() }).where(eq(heartbeatRuns.id, captureFailed.id)); expect(await integrity.retry(owner.companyId, captureFailed.id, "board")).toEqual({ accounted: false }); expect(await integrity.retry(owner.companyId, noWork.id, "board")).toEqual({ accounted: true }); expect(await db.select().from(costEvents).where(eq(costEvents.companyId, owner.companyId))).toEqual([]); + await db.update(heartbeatRuns).set({ finishedAt: now }).where(eq(heartbeatRuns.id, captureFailed.id)); + expect(await integrity.retry(owner.companyId, captureFailed.id, "board")).toEqual({ accounted: true }); + const charges = await db.select().from(costEvents).where(eq(costEvents.companyId, owner.companyId)); + expect(charges.map(charge => [charge.heartbeatRunId, charge.costStatus])).toEqual([[captureFailed.id, "unpriced"]]); }); it("does not offer retry while a native coordinator can still resume provider work", async () => { diff --git a/server/src/services/budget-exposure.ts b/server/src/services/budget-exposure.ts index 1ac270e5f78..901dfab6775 100644 --- a/server/src/services/budget-exposure.ts +++ b/server/src/services/budget-exposure.ts @@ -173,7 +173,9 @@ async function readObservedSpends(db: Db, policies: SpendPolicy[], now: Date, ex const closureTimedOutCharge = sql`(${finalCharge} and exists (select 1 from ${heartbeatRuns} timed_out_run where timed_out_run.id = ${costEvents.heartbeatRunId} and timed_out_run.company_id = ${costEvents.companyId} - and timed_out_run.usage_json->>'accountingClosureTimedOut' = 'true' and not ${restartDerived("timed_out_run")}))`; + and timed_out_run.usage_json->>'accountingClosureTimedOut' = 'true' + -- A failed capture is unknown spend whatever its source history. + and (timed_out_run.usage_json->>'accountingCaptureFailed' = 'true' or not ${restartDerived("timed_out_run")})))`; const exemptFromUnpricedBlock = sql`(not ${closureTimedOutCharge} and ((${finalCharge} and exists (select 1 from ${heartbeatRuns} lost_receipt_run where lost_receipt_run.id = ${costEvents.heartbeatRunId} and lost_receipt_run.company_id = ${costEvents.companyId} diff --git a/server/src/services/run-cost-accounting.ts b/server/src/services/run-cost-accounting.ts index fc244d914b3..2ed480afa39 100644 --- a/server/src/services/run-cost-accounting.ts +++ b/server/src/services/run-cost-accounting.ts @@ -2,7 +2,7 @@ import { notifyStrandedAgentStatus } from "./live-run-recovery-notifications.js" import { notifyDeferredExecutionEligibility } from "./deferred-execution-notifications.js"; import { notifyRunDispatch } from "./run-dispatch-notifications.js"; import { notifyIssueRecovery } from "./issue-recovery-notifications.js"; -import { usdToUnits, unitsToCents } from "@paperclipai/shared"; +import { centsToUnits, centsToUsd, usdToUnits, unitsToCents } from "@paperclipai/shared"; import { and, asc, eq, inArray, isNull, or, sql } from "drizzle-orm"; import { budgetReservations, agents, environmentLeases, heartbeatRunEvents, heartbeatRuns, nativeRunFinalizations, runUsageReceipts, issues, projects, decisionInvocations, promptInjectionInvocations, type Db } from "@paperclipai/db"; import { withOperationAccountingTransaction } from "./accounting-transaction.js"; @@ -143,6 +143,26 @@ function closureEvidenceTimedOut(at: Date = new Date()) { <= ${at.toISOString()}::timestamptz - make_interval(secs => ${ACCOUNTING_CLOSURE_TIMEOUT_MS / 1000})`; } +/** A failed usage capture can never complete its receipt, and replay cannot + * certify what it missed. It is bounded like missing closure evidence: past + * the same bound, once native recovery will not run the provider again, its + * known usage settles as unpriced (or as a lower-bound estimate) and is + * flagged so the blocking policy still counts it. Without the bound such a + * run held company admission until its UTC budget window rolled over. */ +function captureFailureBoundPassed(at: Date = new Date()) { + return sql`${heartbeatRuns.usageJson}->>'accountingCaptureFailed' = 'true' + and ${heartbeatRuns.status} in ('succeeded', 'failed', 'cancelled', 'timed_out', 'interrupted') + and (${heartbeatRuns.runtimeMode} is distinct from 'native' or ${settledNativeCoordinator}) + and coalesce(${heartbeatRuns.finishedAt}, ${heartbeatRuns.updatedAt}) + <= ${at.toISOString()}::timestamptz - make_interval(secs => ${ACCOUNTING_CLOSURE_TIMEOUT_MS / 1000})`; +} + +async function captureFailureSettles(db: Db, runId: string): Promise { + const [row] = await db.select({ due: sql`(${captureFailureBoundPassed()}) is true` }).from(heartbeatRuns) + .where(eq(heartbeatRuns.id, runId)); + return row?.due === true; +} + async function readNativeReceiptClosure(db: Db, runId: string) { const [row] = await db.select({ closed: sql`${closedNativeReceiptSource}`, timedOut: sql`${closureEvidenceTimedOut()}`, neverStarted: sql`${nativeProviderWorkNeverStarted}` }).from(heartbeatRuns) @@ -152,11 +172,14 @@ async function readNativeReceiptClosure(db: Db, runId: string) { return row?.timedOut === true ? "timed_out" as const : null; } -function withClosureTimeoutEvidence(provenance: Record, timedOut: boolean) { +function withClosureTimeoutEvidence(provenance: Record, timedOut: boolean, captureFailed = false) { if (!timedOut) return provenance; const evidence = text(provenance.evidence); + const reason = captureFailed + ? "usage capture failed; the usage observed before the failure settled after the accounting closure timeout, and the run may have spent more" + : "closure evidence never arrived; settled after the accounting closure timeout, and the run may have spent more"; return { source: "unknown", version: "accounting-receipt/v1", ...provenance, - evidence: [evidence, "closure evidence never arrived; settled after the accounting closure timeout, and the run may have spent more"].filter(Boolean).join("; ") }; + evidence: [evidence, reason].filter(Boolean).join("; ") }; } /** A receipt lost with its stopped provider still carries the usage observed @@ -246,12 +269,15 @@ export async function accountRunCost(db: Db, runId: string, hooks: BudgetService throw conflict("Run cost attribution changed. Retry accounting."); } // A failed capture may follow an older complete attempt. Neither replay - // nor terminal status can certify that the missing work was free. - if (object(run.usageJson).accountingCaptureFailed === true) return false; - const providerWorkNeverStarted = run.runtimeMode === "native" && object(run.usageJson).accountingReceiptReady === false + // nor terminal status can certify that the missing work was free, so it + // never settles as pre-provider work. It waits for the closure bound and + // then settles its known usage as possibly incomplete spend. + const captureFailed = object(run.usageJson).accountingCaptureFailed === true; + if (captureFailed && !await captureFailureSettles(tx, run.id)) return false; + const providerWorkNeverStarted = !captureFailed && run.runtimeMode === "native" && object(run.usageJson).accountingReceiptReady === false && await readNativeReceiptClosure(tx, run.id) === "no_provider_work"; - const preProviderFailure = object(object(run.resultJson).executionRecovery).providerWorkStarted === false - || object(run.usageJson).accountingProviderWorkStarted === false || providerWorkNeverStarted; + const preProviderFailure = !captureFailed && (object(object(run.resultJson).executionRecovery).providerWorkStarted === false + || object(run.usageJson).accountingProviderWorkStarted === false || providerWorkNeverStarted); // A failed native turn is not necessarily the end of this heartbeat: the // coordinator can resume the same run. Only its closed provider boundary // authorizes acknowledgement; finalization may still run after a result. @@ -265,8 +291,9 @@ export async function accountRunCost(db: Db, runId: string, hooks: BudgetService // given up and the provider process is verified stopped, no producer can // deliver it: settle the known part as unpriced usage. A later receipt // still posts through late-receipt reconciliation. - const receiptUnavailable = !preProviderFailure && object(run.usageJson).accountingReceiptReady === false; - const receiptClosure = receiptUnavailable ? await readNativeReceiptClosure(tx, run.id) : null; + const receiptUnavailable = !preProviderFailure && (captureFailed || object(run.usageJson).accountingReceiptReady === false); + const receiptClosure = captureFailed ? "timed_out" as const + : receiptUnavailable ? await readNativeReceiptClosure(tx, run.id) : null; if (receiptUnavailable && !receiptClosure) return false; const closureTimedOut = receiptClosure === "timed_out"; if (!preProviderFailure) await ensureRuntimeAccountingBaseline(tx, run.companyId, run.agentId); @@ -293,7 +320,22 @@ export async function accountRunCost(db: Db, runId: string, hooks: BudgetService if (!linked) scope[key] = null; } } - const knownCostUsd = amount(usage.cacheAdjustedCostUsd) ?? text(usage.costUsdExact) ?? amount(usage.costUsd); + const reportedCostUsd = amount(usage.cacheAdjustedCostUsd) ?? text(usage.costUsdExact) ?? amount(usage.costUsd); + // A later checkpoint can clear the raw cost fields while the live budget + // projection keeps the price already observed. A receipt that will never + // complete (lost, timed out, or a failed capture) settles at no less than + // that observed money, so settlement never lowers spend. + const observedCents = receiptUnavailable ? text(object(usage.budgetExposure).costCents) : null; + let observedCostUsd: string | null = null; + if (observedCents) { + try { + const observedUnits = centsToUnits(observedCents); + if (observedUnits > 0n && (reportedCostUsd === null || observedUnits > usdToUnits(reportedCostUsd))) { + observedCostUsd = centsToUsd(observedCents); + } + } catch { /* An unreadable projection proves nothing; keep the reported cost. */ } + } + const knownCostUsd = observedCostUsd ?? reportedCostUsd; const lostReceiptEstimate = receiptUnavailable && knownCostUsd === null ? estimateLostReceipt(usage) : null; const costUsd = knownCostUsd ?? lostReceiptEstimate?.costUsdExact ?? null; const billingType = text(usage.billingType) ?? "unknown"; @@ -317,7 +359,7 @@ 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: withClosureTimeoutEvidence(lostReceiptEstimate?.pricingProvenance ?? lowerBoundProvenance ?? object(usage.pricingProvenance), closureTimedOut), + pricingProvenance: withClosureTimeoutEvidence(lostReceiptEstimate?.pricingProvenance ?? lowerBoundProvenance ?? object(usage.pricingProvenance), closureTimedOut, captureFailed), model: text(usage.model) ?? "unknown", billingType, costStatus: lostReceiptEstimate || reportedLowerBound ? "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, @@ -364,11 +406,19 @@ export async function accountRunCost(db: Db, runId: string, hooks: BudgetService ...(lostReceiptEstimate ? { costUsdExact: lostReceiptEstimate.costUsdExact, costStatus: "estimated", pricingProvenance: lostReceiptEstimate.pricingProvenance } : {}), ...(lowerBoundProvenance ? { costStatus: "estimated", pricingProvenance: lowerBoundProvenance } : {}), + // The observed price became the accepted money; late reconciliation + // measures growth from it rather than posting the full total again. + // cacheAdjustedCostUsd outranks costUsdExact in every reader, so a + // lower one would otherwise survive and win. + ...(observedCostUsd ? { costUsdExact: observedCostUsd, cacheAdjustedCostUsd: null } : {}), })}::jsonb` } : {}), }).where(eq(heartbeatRuns.id, run.id)); if (closureTimedOut) await logActivity(tx, { companyId: run.companyId, actorType: "system", actorId: "cost_accounting", action: "cost.accounting_closure_timed_out", entityType: "heartbeat_run", entityId: run.id, runId: run.id, agentId: run.agentId, - details: { reason: "Final usage never arrived and no stop of its provider was verified; the run's known usage was recorded as a lower bound, or as unpriced when it had no price.", + details: { reason: captureFailed + ? "Usage capture failed, so the final usage can never be confirmed; the run's known usage was recorded as a lower bound, or as unpriced when it had no price." + : "Final usage never arrived and no stop of its provider was verified; the run's known usage was recorded as a lower bound, or as unpriced when it had no price.", + ...(captureFailed ? { captureFailed: true } : {}), timeoutMinutes: ACCOUNTING_CLOSURE_TIMEOUT_MS / 60_000 } }, publications); return true; }), { "budget.accounting.owner": "run" }); @@ -386,14 +436,16 @@ function actionableRunAccounting(at: Date = new Date()) { or(sql`${heartbeatRuns.usageJson}->>'accountingLateReceiptPending' is distinct from 'true'`, sql`${heartbeatRuns.usageJson}->>'accountingLateReceiptId' is distinct from ${heartbeatRuns.usageJson}->>'accountingLateReceiptAttemptedId'`)), and( sql`${heartbeatRuns.costAccountedAt} is null`, - sql`${heartbeatRuns.usageJson}->>'accountingCaptureFailed' is distinct from 'true'`, - sql`(${heartbeatRuns.resultJson}->'executionRecovery'->>'providerWorkStarted' = 'false' - or ${heartbeatRuns.usageJson}->>'accountingProviderWorkStarted' = 'false' - or not exists (select 1 from ${nativeRunFinalizations} where ${nativeRunFinalizations.runId} = ${heartbeatRuns.id} - and ${heartbeatRuns.runtimeMode} = 'native' and ${nativeRunFinalizations.resultId} is null - and ${nativeRunFinalizations.phase} <> 'terminal_failure'))`, - sql`(${heartbeatRuns.usageJson}->>'accountingReceiptReady' is distinct from 'false' or ${heartbeatRuns.resultJson}->'executionRecovery'->>'providerWorkStarted' = 'false' or ${heartbeatRuns.usageJson}->>'accountingProviderWorkStarted' = 'false' - or (${closedNativeReceiptSource}) or (${closureEvidenceTimedOut(at)}) or (${nativeProviderWorkNeverStarted}))`, + or(sql`(${captureFailureBoundPassed(at)})`, and( + sql`${heartbeatRuns.usageJson}->>'accountingCaptureFailed' is distinct from 'true'`, + sql`(${heartbeatRuns.resultJson}->'executionRecovery'->>'providerWorkStarted' = 'false' + or ${heartbeatRuns.usageJson}->>'accountingProviderWorkStarted' = 'false' + or not exists (select 1 from ${nativeRunFinalizations} where ${nativeRunFinalizations.runId} = ${heartbeatRuns.id} + and ${heartbeatRuns.runtimeMode} = 'native' and ${nativeRunFinalizations.resultId} is null + and ${nativeRunFinalizations.phase} <> 'terminal_failure'))`, + sql`(${heartbeatRuns.usageJson}->>'accountingReceiptReady' is distinct from 'false' or ${heartbeatRuns.resultJson}->'executionRecovery'->>'providerWorkStarted' = 'false' or ${heartbeatRuns.usageJson}->>'accountingProviderWorkStarted' = 'false' + or (${closedNativeReceiptSource}) or (${closureEvidenceTimedOut(at)}) or (${nativeProviderWorkNeverStarted}))`, + )), )), ); } @@ -414,10 +466,17 @@ export async function readCostAccountingWork(db: Db, now = Date.now(), receiptDi // Wake when a run waiting for closure evidence reaches its bound. const [closure] = await db.select({ at: sql`min(coalesce(${heartbeatRuns.finishedAt}, ${heartbeatRuns.updatedAt}))` }) .from(heartbeatRuns).where(and(eq(heartbeatRuns.costAccountingPending, true), isNull(heartbeatRuns.costAccountedAt), - sql`${heartbeatRuns.usageJson}->>'accountingReceiptReady' = 'false'`, - sql`${heartbeatRuns.usageJson}->>'accountingCaptureFailed' is distinct from 'true'`, - sql`${heartbeatRuns.runtimeMode} = 'native' and ${heartbeatRuns.status} in ('succeeded', 'failed', 'cancelled', 'timed_out', 'interrupted')`, - settledNativeCoordinator, + sql`${heartbeatRuns.status} in ('succeeded', 'failed', 'cancelled', 'timed_out', 'interrupted')`, + or(and( + sql`${heartbeatRuns.usageJson}->>'accountingReceiptReady' = 'false'`, + sql`${heartbeatRuns.usageJson}->>'accountingCaptureFailed' is distinct from 'true'`, + sql`${heartbeatRuns.runtimeMode} = 'native'`, + settledNativeCoordinator, + ), and( + // A failed capture waits for the same bound; see captureFailureBoundPassed. + sql`${heartbeatRuns.usageJson}->>'accountingCaptureFailed' = 'true'`, + sql`(${heartbeatRuns.runtimeMode} is distinct from 'native' or ${settledNativeCoordinator})`, + )), // Only future bounds: a run already past it is in the actionable scan, // and a past deadline must never schedule an immediate pass again. sql`coalesce(${heartbeatRuns.finishedAt}, ${heartbeatRuns.updatedAt})