mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-10 01:50:13 +01:00
fix(budgets): price lost receipts and stop restart-derived accounting from hard-stopping companies (#19673)
## 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 <noreply@anthropic.com>
This commit is contained in:
1 parent
785ead5c7a
commit
06bbcea12f
6 files changed
+206
-30
No files matched your search
@@ -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.
|
||||
|
||||
@@ -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<ReturnType<typeof fixture>>, usage: Record<string, unknown>) {
|
||||
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);
|
||||
|
||||
@@ -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<ObservedSpend[]> {
|
||||
async function readObservedSpends(db: Db, policies: SpendPolicy[], now: Date, excludeRunId?: string | null): Promise<ObservedSpend[]> {
|
||||
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<string, SQL> = {}, heldColumns: Record<string, SQL> = {}, runColumns: Record<string, SQL> = {}, browserColumns: Record<string, SQL> = {};
|
||||
@@ -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<ObservedSpend[]> {
|
||||
export function computeObservedSpends(db: Db, policies: SpendPolicy[], now = new Date(), excludeRunId?: string | null): Promise<ObservedSpend[]> {
|
||||
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 });
|
||||
}
|
||||
@@ -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.
|
||||
|
||||
@@ -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<ReturnType<typeof reconcileScope>>) => ({ 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 })),
|
||||
|
||||
@@ -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<string, unknown>) {
|
||||
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" });
|
||||
|
||||
Reference in new issue
Block a user