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