diff --git a/doc/SPEC-implementation.md b/doc/SPEC-implementation.md index e607a41a137..b69d63a9bc5 100644 --- a/doc/SPEC-implementation.md +++ b/doc/SPEC-implementation.md @@ -1763,7 +1763,10 @@ publication is fenced separately from settlement. Inputs arriving or committing later stay pending; no global sequence watermark consumes them. Limited-budget admission refuses relevant unevaluated inputs older than fifteen seconds. Unchanged idle decisions do not expire; unknown accounting retains its separate -conservative admission restriction. Provider reporting and cancellation delays +conservative admission restriction. Completed runs that await accounting hold +new work only: they never open a hard-stop incident, pause a scope, or stop +running work. Only spend at the limit, or unpriced usage under a blocking +policy, does that. Provider reporting and cancellation delays remain additional to evaluation lag. Cancellation delivery is asynchronous and scoped; committed per-run Stop survives policy changes and restart until existing stop proof succeeds. Budget resume cannot clear a manual hold. @@ -1878,6 +1881,12 @@ Native runs retain final usage receipts during a bounded accounting-only drain after a governed wait cancels provider work. This does not accept late messages, tool calls, or new completion proposals. Missing or incomplete receipts continue to block budget admission. +After native recovery reaches terminal failure, an incomplete receipt holds new +work as recovering accounting; it does not open a hard-stop incident. When the +run no longer records a provider process (recovery clears it only after a +verified stop), no producer can deliver the receipt. 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. 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__/budget-enforcement-scoping.test.ts b/server/src/__tests__/budget-enforcement-scoping.test.ts index c471669a2eb..124e5015f91 100644 --- a/server/src/__tests__/budget-enforcement-scoping.test.ts +++ b/server/src/__tests__/budget-enforcement-scoping.test.ts @@ -246,20 +246,22 @@ const support = await getEmbeddedPostgresTestSupport(); expect(await db.select().from(costEvents).where(and(eq(costEvents.companyId, f.companyId), eq(costEvents.heartbeatRunId, f.runId)))).toHaveLength(0); }); - it.each(["pre-provider release", "stale pending marker"])("releases accounting-induced holds after %s", async kind => { + // Pending accounting holds admission only; it never places a budget hold. + it.each(["pre-provider release", "stale pending marker"])("releases the accounting admission hold after %s", async kind => { const f = await fixture(); await db.update(heartbeatRuns).set({ status: "cancelled", finishedAt: new Date() }).where(eq(heartbeatRuns.id, f.runId)); expect(await budgetService(db).getInvocationBlock(f.companyId, f.agentId, { projectId: f.projectId })).not.toBeNull(); await budgetService(db).evaluateAffectedScopes(f.companyId, f.agentId, f.projectId); await flushBudgetEvaluationWork(db); const [held] = await db.select().from(agents).where(eq(agents.id, f.agentId)); - expect(held.lifecycleHolds).toContain("budget"); + expect(held.lifecycleHolds).not.toContain("budget"); await db.update(heartbeatRuns).set(kind === "pre-provider release" ? { resultJson: { executionRecovery: { providerWorkStarted: false } } } : { costAccountedAt: new Date() }).where(eq(heartbeatRuns.id, f.runId)); expect(await accountRunCost(db, f.runId)).toBe(kind === "pre-provider release"); const [released] = await db.select().from(agents).where(eq(agents.id, f.agentId)); expect(released.lifecycleHolds).not.toContain("budget"); + expect(await budgetService(db).getInvocationBlock(f.companyId, f.agentId, { projectId: f.projectId })).toBeNull(); expect((await f.readRun()).costAccountingPending).toBe(false); expect(await db.select().from(costEvents).where(eq(costEvents.companyId, f.companyId))).toHaveLength(0); }); diff --git a/server/src/__tests__/closed-native-receipt-accounting.test.ts b/server/src/__tests__/closed-native-receipt-accounting.test.ts new file mode 100644 index 00000000000..fde4b43b357 --- /dev/null +++ b/server/src/__tests__/closed-native-receipt-accounting.test.ts @@ -0,0 +1,284 @@ +import { randomUUID } from "node:crypto"; +import { and, eq, sql } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; +import { + agents, budgetIncidents, budgetPolicies, budgetReservations, companies, costEvents, createDb, heartbeatRunEvents, heartbeatRuns, issues, + nativeRunFinalizations, type Db, +} from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { budgetService, flushBudgetEvaluationWork } from "../services/budgets.js"; +import { accountRunCost, reconcileRunCosts } from "../services/run-cost-accounting.js"; +import { persistUsageReceipt } from "../services/usage-receipts.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 +// recovery has given up and verified the provider process stopped, no producer +// can deliver that receipt. The debt must leave "awaiting receipt" or every +// later admission in the company refuses forever, whatever the budget allows. +const support = await getEmbeddedPostgresTestSupport(); +(support.supported ? describe : describe.skip)("closed native receipt accounting", () => { + let database: Awaited>; + let db: Db; + beforeAll(async () => { + database = await startEmbeddedPostgresTestDatabase("closed-native-receipt-"); + db = createDb(database.connectionString); + }, 60_000); + afterEach(async () => { await db.transaction(async tx => { + await tx.execute(sql`SET LOCAL client_min_messages = warning`); + await tx.execute(sql`TRUNCATE companies CASCADE`); + }); }); + afterAll(async () => { await database?.cleanup(); }); + + async function fixture(options: { + unpricedUsagePolicy?: "block" | "allow"; + phase?: string; + processPid?: number | null; + stopEvidence?: boolean; + failureCode?: string; + } = {}) { + const companyId = randomUUID(), agentId = randomUUID(), otherAgentId = randomUUID(), sourceId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Chaos Lab", issuePrefix: "CHA" }); + await db.insert(agents).values([ + { id: agentId, companyId, name: "Claude", role: "engineer", status: "idle", adapterType: "paperclip_runner" }, + { id: otherAgentId, companyId, name: "Codex", role: "engineer", status: "idle", adapterType: "paperclip_runner" }, + ]); + const [policy] = await db.insert(budgetPolicies).values({ companyId, scopeType: "company", scopeId: companyId, + metric: "billed_cents", windowKind: "calendar_month_utc", amount: 500, notifyEnabled: false, + unpricedUsagePolicy: options.unpricedUsagePolicy ?? "allow" }).returning(); + const [issue] = await db.insert(issues).values({ companyId, title: "Interrupted task", status: "blocked", + assigneeAgentId: agentId }).returning(); + // The shape persisted for the killed ACPX Claude run: the final provider + // turn failed without a per-turn usage report, so the receipt never closed. + const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, invocationSource: "automation", + scopeKind: "issue", issueId: issue.id, nativeIssueId: issue.id, runtimeMode: "native", status: "failed", + errorCode: "native_provider_terminal_failed", finishedAt: new Date(), costAccountingPending: true, + processPid: options.processPid ?? null, processGroupId: options.processPid ?? null, + processStartedAt: options.processPid ? new Date() : null, + usageJson: { accountingReceiptOwner: "native_producer", accountingReceiptSourceId: sourceId, + accountingReceiptSequence: 1, accountingReceiptReady: false, provider: "anthropic", biller: "anthropic", + model: "claude-sonnet-5", billingType: "metered_api", costStatus: "unpriced", costUsd: null, + inputTokens: 0, cachedInputTokens: 0, outputTokens: 0, cacheWriteTokens: 0, usageSource: "per_run", + budgetExposure: { sourceId, sequence: 1, costCents: "0.0000000" }, + ledgerScope: { issueId: issue.id, projectId: null, billingCode: null } } }).returning(); + await db.insert(nativeRunFinalizations).values({ companyId, issueId: issue.id, runId: run.id, + phase: options.phase ?? "terminal_failure", attempt: 3, failureCode: options.failureCode ?? "native_provider_terminal_failed" }); + // Server-authored process trail: launch, recorded identity, then the + // verified local stop recovery writes before clearing the identity. + await db.insert(heartbeatRunEvents).values([ + { companyId, runId: run.id, agentId, seq: 1, eventType: "native.process_start_requested", stream: "system" }, + { companyId, runId: run.id, agentId, seq: 2, eventType: "native.process_identity_recorded", stream: "system" }, + ...(options.stopEvidence === false ? [] : [ + { companyId, runId: run.id, agentId, seq: 3, eventType: "native.local_process_stopped", stream: "system" }]), + ]); + await db.insert(budgetReservations).values({ companyId, runId: run.id, agentId, amountCents: "0", + state: "held", providerStartedAt: new Date() }); + const readRun = async () => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0]; + const charges = () => db.select().from(costEvents).where(and(eq(costEvents.companyId, companyId), + eq(costEvents.heartbeatRunId, run.id))); + const reservation = async () => (await db.select().from(budgetReservations) + .where(eq(budgetReservations.runId, run.id)))[0]; + const block = () => budgetService(db).getInvocationBlock(companyId, otherAgentId, { issueId: null, projectId: null }); + return { companyId, agentId, otherAgentId, sourceId, policy, issue, run, readRun, charges, reservation, block }; + } + + it("settles a stopped native terminal failure without a final receipt as unpriced usage", async () => { + const f = await fixture(); + expect(await f.block()).toMatchObject({ reason: "New work must wait while a native run recovers its unfinished accounting." }); + + // The background reconciler must select this debt; it is not merely retryable by hand. + expect(await reconcileRunCosts(db)).toMatchObject({ accounted: 1 }); + await flushBudgetEvaluationWork(db); + + const settled = await f.readRun(); + expect(settled.costAccountingPending).toBe(false); + expect(settled.costAccountedAt).not.toBeNull(); + const charges = await f.charges(); + expect(charges).toHaveLength(1); + // Unknown spend is never presented as a zero-cost receipt. + expect(charges[0]).toMatchObject({ costStatus: "unpriced", model: "claude-sonnet-5", provider: "anthropic" }); + expect((await f.reservation()).state).toBe("settled"); + // The operator's unpriced-usage policy now decides admission. + expect(await f.block()).toBeNull(); + }); + + // Chaos wave: under the default blocking policy, a lost zero-token receipt + // opened a hard stop at 0 of 500 cents and blocked every task. + it("does not hard-stop a blocking policy for a lost receipt with no usage", async () => { + const f = await fixture({ unpricedUsagePolicy: "block" }); + 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 f.block()).toBeNull(); + }); + + it("keeps a blocking policy for lost-receipt usage that has tokens but no price", async () => { + const f = await fixture({ unpricedUsagePolicy: "block" }); + await db.update(heartbeatRuns).set({ usageJson: sql`${heartbeatRuns.usageJson} || '{"inputTokens":1200,"outputTokens":80}'::jsonb` }) + .where(eq(heartbeatRuns.id, 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 debt pending while the provider process identity is still recorded", async () => { + const f = await fixture({ processPid: 424242 }); + expect(await reconcileRunCosts(db)).toMatchObject({ accounted: 0 }); + expect(await accountRunCost(db, f.run.id)).toBe(false); + expect((await f.readRun()).costAccountingPending).toBe(true); + expect(await f.charges()).toHaveLength(0); + expect((await f.reservation()).state).toBe("held"); + }); + + // Between a native terminal failure and the verified provider stop, the debt + // is closing, not over budget. A hard stop here paused the company and + // cancelled an unrelated running task in the live restart wave. + it("holds new work without a hard stop while a terminal native receipt is closing", async () => { + const f = await fixture({ processPid: 424242 }); + 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.status).toBe("active"); + expect(company.pauseReason).toBeNull(); + expect(await db.select().from(budgetIncidents).where(eq(budgetIncidents.companyId, f.companyId))).toHaveLength(0); + expect(await f.block()).toMatchObject({ reason: "New work must wait while a native run recovers its unfinished accounting." }); + }); + + // Chaos waves: a run that had just succeeded was terminal for ~100 ms before + // its charge committed. The evaluator read that gap as a hard stop, paused the + // company and cancelled unrelated running work, which then never settled. + it("holds admission without a hard stop while a finished run's charge commits", async () => { + const f = await fixture(); + await db.update(heartbeatRuns).set({ status: "succeeded", errorCode: null, + usageJson: sql`${heartbeatRuns.usageJson} || '{"accountingReceiptReady":true,"costUsd":0.1}'::jsonb` }) + .where(eq(heartbeatRuns.id, f.run.id)); + // Its coordinator already applied the result, so nothing counts it as recovering. + await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.run.id)); + const runningId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: runningId, companyId: f.companyId, agentId: f.otherAgentId, + invocationSource: "assignment", status: "running" }); + 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 db.select().from(budgetIncidents).where(eq(budgetIncidents.companyId, f.companyId))).toHaveLength(0); + const [policy] = await db.select().from(budgetPolicies).where(eq(budgetPolicies.id, f.policy.id)); + expect(policy.enforcementVersion).toBe(f.policy.enforcementVersion); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runningId)))[0].status).toBe("running"); + expect(await f.block()).toMatchObject({ reason: "Company cannot start work while completed runs await accounting." }); + }); + + it("settles a finalization-deadline failure once its local stop is verified", async () => { + const f = await fixture({ failureCode: "execution_finalization_deadline_exceeded" }); + expect(await f.block()).not.toBeNull(); + expect(await reconcileRunCosts(db)).toMatchObject({ accounted: 1 }); + await flushBudgetEvaluationWork(db); + expect(await f.block()).toBeNull(); + }); + + it.each([ + ["no verified stop event", { stopEvidence: false }], + ["a finalization deadline without a verified stop", { failureCode: "execution_finalization_deadline_exceeded", stopEvidence: false }], + ] as const)("keeps the debt pending with %s", async (_label, options) => { + const f = await fixture(options); + expect(await reconcileRunCosts(db)).toMatchObject({ accounted: 0 }); + expect(await accountRunCost(db, f.run.id)).toBe(false); + expect((await f.readRun()).costAccountingPending).toBe(true); + expect(await f.charges()).toHaveLength(0); + }); + + // Upgrade path: an older release paused the company for pending accounting. + it("releases a budget-owned pause that only pending accounting kept", async () => { + const f = await fixture({ stopEvidence: false }); + await db.update(companies).set({ status: "paused", pauseReason: "budget", pausedAt: new Date() }) + .where(eq(companies.id, f.companyId)); + await db.insert(budgetIncidents).values({ companyId: f.companyId, policyId: f.policy.id, scopeType: "company", + scopeId: f.companyId, metric: "billed_cents", windowKind: "calendar_month_utc", + windowStart: new Date(Date.UTC(2026, 10, 1)), windowEnd: new Date(Date.UTC(2026, 11, 1)), + thresholdType: "hard", amountLimit: 500, amountObserved: 0 }); + await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.run.id)); + 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 db.select().from(budgetIncidents).where(eq(budgetIncidents.companyId, f.companyId)))[0].status).not.toBe("open"); + expect(await f.block()).toMatchObject({ reason: "Company cannot start work while completed runs await accounting." }); + const [policy] = await db.select().from(budgetPolicies).where(eq(budgetPolicies.id, f.policy.id)); + expect((await budgetService(db).overview(f.companyId)).policies[0]?.status).not.toBe("hard_stop"); + expect(policy.id).toBe(f.policy.id); + }); + + it("keeps the debt pending while native recovery can still produce a result", async () => { + const f = await fixture({ phase: "observed" }); + expect(await reconcileRunCosts(db)).toMatchObject({ accounted: 0 }); + expect((await f.readRun()).costAccountingPending).toBe(true); + expect(await f.charges()).toHaveLength(0); + }); + + it("reconciles a final receipt that arrives after the unavailable receipt settled", async () => { + const f = await fixture(); + expect(await accountRunCost(db, f.run.id)).toBe(true); + await persistUsageReceipt(db, { schema: "paperclip/accounting-receipt/v1", id: randomUUID(), + companyId: f.companyId, runId: f.run.id, adapterType: "paperclip_runner", sourceId: f.sourceId, sequence: 2, + receivedAt: new Date().toISOString(), receipt: { complete: true, usageBasis: "per_run", provider: "anthropic", + biller: "anthropic", model: "claude-sonnet-5", billingType: "metered_api", + usage: { inputTokens: 1200, cachedInputTokens: 0, outputTokens: 80 } } }); + const late = await f.readRun(); + expect(late.costAccountingPending).toBe(true); + expect(late.usageJson).toMatchObject({ accountingLateReceiptPending: true }); + }); + + it("does not reopen the hold for a later receipt the rate card could price but that carries no price", async () => { + const f = await fixture(); + expect(await accountRunCost(db, f.run.id)).toBe(true); + await persistUsageReceipt(db, { schema: "paperclip/accounting-receipt/v1", id: randomUUID(), + companyId: f.companyId, runId: f.run.id, adapterType: "paperclip_runner", sourceId: f.sourceId, sequence: 2, + receivedAt: new Date().toISOString(), + producerEvent: { sourceEventId: randomUUID(), eventType: "turn.completed" }, + receipt: { complete: true, usageBasis: "per_run", provider: "anthropic", biller: "anthropic", + model: "claude-sonnet-5", billingType: "metered_api", + usage: { inputTokens: 0, cachedInputTokens: 0, outputTokens: 0, cacheWriteTokens: 0 } } }); + expect((await f.readRun()).costAccountingPending).toBe(false); + expect(await f.block()).toBeNull(); + }); + + it("does not reopen the hold for a later complete receipt that adds no price or usage", async () => { + const f = await fixture(); + expect(await accountRunCost(db, f.run.id)).toBe(true); + await persistUsageReceipt(db, { schema: "paperclip/accounting-receipt/v1", id: randomUUID(), + companyId: f.companyId, runId: f.run.id, adapterType: "paperclip_runner", sourceId: f.sourceId, sequence: 2, + receivedAt: new Date().toISOString(), + producerEvent: { sourceEventId: randomUUID(), eventType: "turn.completed" }, + receipt: { complete: true, usageBasis: "per_run", provider: "anthropic", biller: "anthropic", + model: "claude-sonnet-5", billingType: "metered_api", costStatus: "unpriced", + usage: { inputTokens: 0, cachedInputTokens: 0, outputTokens: 0 } } }); + expect((await f.readRun()).costAccountingPending).toBe(false); + expect(await f.block()).toBeNull(); + }); + + // A producer receipt priced only from the rate card is normally not a money + // correction. For a charge settled without its receipt, it is the first price. + it("prices the unavailable charge when a complete rate-card receipt arrives later", async () => { + const f = await fixture({ unpricedUsagePolicy: "block" }); + expect(await accountRunCost(db, f.run.id)).toBe(true); + await persistUsageReceipt(db, { schema: "paperclip/accounting-receipt/v1", id: randomUUID(), + companyId: f.companyId, runId: f.run.id, adapterType: "paperclip_runner", sourceId: f.sourceId, sequence: 2, + receivedAt: new Date().toISOString(), + producerEvent: { sourceEventId: randomUUID(), eventType: "turn.completed" }, + receipt: { complete: true, usageBasis: "per_run", provider: "anthropic", biller: "anthropic", + model: "claude-sonnet-5", billingType: "metered_api", costUsd: 0.25, costStatus: "estimated", + pricingProvenance: { source: "rate_card", version: "anthropic-test" }, + usage: { inputTokens: 0, cachedInputTokens: 0, outputTokens: 0 } } }); + expect((await f.readRun()).usageJson).toMatchObject({ accountingLateReceiptPending: true }); + await accountRunCost(db, f.run.id); + await flushBudgetEvaluationWork(db); + expect((await f.readRun()).costAccountingPending).toBe(false); + const charges = await f.charges(); + expect(charges).toHaveLength(1); + expect(charges[0]).toMatchObject({ costStatus: "estimated" }); + expect(Number(charges[0].costCents)).toBeCloseTo(25, 5); + expect(await f.block()).toBeNull(); + }); +}); diff --git a/server/src/services/budget-exposure.ts b/server/src/services/budget-exposure.ts index 22345462b3c..960533c3b1b 100644 --- a/server/src/services/budget-exposure.ts +++ b/server/src/services/budget-exposure.ts @@ -123,13 +123,28 @@ async function readObservedSpends(db: Db, policies: SpendPolicy[], now: Date): P const hasReservation = sql`exists (select 1 from ${budgetReservations} where ${budgetReservations.companyId} = ${heartbeatRuns.companyId} and ${budgetReservations.runId} = ${heartbeatRuns.id} and ${budgetReservations.state} = 'held')`; + // A terminal failure whose final receipt never closed is still recovering + // accounting: once its provider stop is verified, accounting settles the + // known part as unpriced. That window holds new work; it is not a hard stop. + const closingReceipt = sql`${usage}->>'accountingReceiptReady' = 'false' + and ${usage}->>'accountingCaptureFailed' is distinct from 'true'`; const recovering = sql`not (${lateReceiptDebt}) and exists (select 1 from ${nativeRunFinalizations} where ${nativeRunFinalizations.runId} = ${heartbeatRuns.id} and ${nativeRunFinalizations.companyId} = ${heartbeatRuns.companyId} - and ${nativeRunFinalizations.resultId} is null and ${nativeRunFinalizations.phase} <> 'terminal_failure')`; + and ${nativeRunFinalizations.resultId} is null + 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`; - ledgerColumns[`unpriced${index}`] = sql`count(*) filter (where ${ledger} and ${costEvents.costStatus} = 'unpriced' and ${costEvents.billingType} <> 'subscription_included')::int`; + // 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'))`; + ledgerColumns[`unpriced${index}`] = sql`count(*) filter (where ${ledger} and ${costEvents.costStatus} = 'unpriced' and ${costEvents.billingType} <> 'subscription_included' + and not ${lostReceiptWithoutUsage})::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 diff --git a/server/src/services/budgets.ts b/server/src/services/budgets.ts index 98404b0ae63..b7868af8d24 100644 --- a/server/src/services/budgets.ts +++ b/server/src/services/budgets.ts @@ -260,13 +260,24 @@ function shouldResolveSoftIncident(policy: PolicyRow, observed: ObservedSpend) { || compareCents(observed.totalExact, Math.ceil(policy.amount * policy.warnPercent / 100)) < 0; } +/** Spend reached the limit, or the operator blocks unpriced usage. Only this + * opens a hard-stop incident, pauses a scope, or stops running work. */ +function hardStopReached(policy: PolicyRow, observed: Awaited>) { + if (!policy.isActive || !policy.hardStopEnabled || policy.amount <= 0) return false; + return compareCents(observed.totalExact, policy.amount) >= 0 || unpricedUsageBlocks(policy, observed); +} + +/** Admission also waits while completed runs await accounting. That unknown + * is a separate conservative restriction: it holds new work only. A run that + * finishes moments before its charge commits must not pause the company or + * cancel unrelated running work. */ function observedBlocks(policy: PolicyRow, observed: Awaited>) { if (!policy.isActive || !policy.hardStopEnabled || policy.amount <= 0) return false; - return compareCents(observed.totalExact, policy.amount) >= 0 || (observed.pendingRunCount - observed.recoveringRunCount) > 0 || unpricedUsageBlocks(policy, observed); + return hardStopReached(policy, observed) || (observed.pendingRunCount - observed.recoveringRunCount) > 0; } export async function policyBlocks(db: Db, policy: PolicyRow) { if (!policy.isActive || !policy.hardStopEnabled || policy.amount <= 0) return false; - return observedBlocks(policy, await computeObservedSpend(db, policy)); + return hardStopReached(policy, await computeObservedSpend(db, policy)); } /** Supplemental handoff check after authoritative accounting admission commits. @@ -425,7 +436,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica eq(budgetPolicies.scopeId, policy.scopeId), eq(budgetPolicies.isActive, true), eq(budgetPolicies.hardStopEnabled, true), )); const observations = await computeObservedSpends(db, policies, now); - if (policies.some((candidate, index) => observedBlocks(candidate, observations[index]))) return; + if (policies.some((candidate, index) => hardStopReached(candidate, observations[index]))) return; if (policy.scopeType === "agent") { const { applyAgentBudgetHoldInTransaction } = await import("./agent-lifecycle.js"); await applyAgentBudgetHoldInTransaction(db, policy.companyId, policy.scopeId, false); @@ -516,7 +527,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica hardStopEnabled: policy.hardStopEnabled, notifyEnabled: policy.notifyEnabled, isActive: policy.isActive, - status: observedBlocks(policy, observed) ? "hard_stop" : policy.isActive + status: hardStopReached(policy, observed) ? "hard_stop" : policy.isActive ? budgetStatusFromObserved(observedAmountExact, amount, policy.warnPercent) : "ok", paused: scope.paused, @@ -690,13 +701,14 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica eq(budgetPolicies.companyId, companyId), eq(budgetPolicies.scopeType, scopeType), eq(budgetPolicies.scopeId, scopeId), )); let recoveringPolicy: PolicyRow | undefined; + let accountingHold = false; const observations = prepared?.observations ?? await computeObservedSpends(db, policies, now); + const title = scopeType === "company" ? "Company" : scopeType === "agent" ? "Agent" : "Project"; for (const [index, policy] of policies.entries()) { const observed = observations[index]; - if (observedBlocks(policy, observed)) { + if (hardStopReached(policy, observed)) { await createIncidentIfNeeded(policy, "hard", observed); await pauseAndCancelScopeForBudget(policy, !scope.budgetHeld); - const title = scopeType === "company" ? "Company" : scopeType === "agent" ? "Agent" : "Project"; if (observed.pendingRunCount > 0) { return { scopeType, scopeId, scopeName: scope.name, reason: `${title} cannot start work while completed runs await accounting.` }; } @@ -707,6 +719,9 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica ? "Project cannot start work because its budget hard-stop is still exceeded." : `${title} is paused because its budget hard-stop was reached.` }; } + // Pending accounting holds new work only. It never keeps a budget-owned + // pause or a hard incident, including one opened by older releases. + if (observedBlocks(policy, observed)) accountingHold = true; // An open native coordinator must finish the original run's accounting. // Deny fresh work without pausing/cancelling that recovery or its hold. if (policy.isActive && policy.hardStopEnabled && policy.amount > 0 && observed.recoveringRunCount > 0) recoveringPolicy ??= policy; @@ -720,6 +735,8 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica const pauseBlock = scopePauseBlock(scopeType, scopeId, scope, policies.length > 0); if (pauseBlock) return pauseBlock; if (scope.budgetHeld && policies[0]) await resumeScopeFromBudget(policies[0]); + if (accountingHold) return { scopeType, scopeId, scopeName: scope.name, + reason: `${title} cannot start work while completed runs await accounting.` }; if (recoveringPolicy) return { scopeType, scopeId, scopeName: scope.name, reason: "New work must wait while a native run recovers its unfinished accounting." }; return null; @@ -766,7 +783,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica } } - if (observedBlocks(policy, observed)) { + if (hardStopReached(policy, observed)) { await resolveOpenSoftIncidents(policy.id, policy.companyId); const hardIncident = await createIncidentIfNeeded(policy, "hard", observed); await pauseAndCancelScopeForBudget(policy); @@ -793,7 +810,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica // Release only this scope's budget-owned hold after all its policies agree. const healthyScopes = new Map(); for (const [index, policy] of policies.entries()) { - if (!observedBlocks(policy, observations[index])) healthyScopes.set(`${policy.scopeType}:${policy.scopeId}`, policy); + if (!hardStopReached(policy, observations[index])) healthyScopes.set(`${policy.scopeType}:${policy.scopeId}`, policy); } for (const policy of healthyScopes.values()) { const scope = await resolveScopeRecord(db, policy.scopeType as BudgetScopeType, policy.scopeId); @@ -959,7 +976,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica if (row.isActive && amount > 0) { const observed = await computeObservedSpend(db, row); - if (!observedBlocks(row, observed)) { + if (!hardStopReached(row, observed)) { await resumeScopeFromBudget(row); await resolveOpenIncidentsForPolicy(row.id, row.companyId, "approved", actorUserId ?? "budget_service", "hard"); if (row.notifyEnabled && compareCents(observed.totalExact, Math.ceil(row.amount * row.warnPercent / 100)) >= 0) { @@ -972,7 +989,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica if (row.notifyEnabled && compareCents(observed.totalExact, softThreshold) >= 0 && (!row.hardStopEnabled || compareCents(observed.totalExact, row.amount) < 0)) { await createIncidentIfNeeded(row, "soft", observed); } - if (observedBlocks(row, observed)) { + if (hardStopReached(row, observed)) { await resolveOpenSoftIncidents(row.id, row.companyId); await createIncidentIfNeeded(row, "hard", observed); await pauseAndCancelScopeForBudget(row); @@ -1062,7 +1079,6 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica if (unpricedUsageBlocks(policy, observed)) { throw unprocessable("Resolve unpriced usage or explicitly allow it in the budget policy before resuming"); } - if (observed.pendingRunCount > 0) throw unprocessable("Completed runs must finish accounting before resuming"); if (compareCents(observed.totalExact, nextAmount) >= 0) { throw unprocessable("New budget must exceed current observed spend"); } diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index df6bd364dd3..6dff2d1c475 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -278,7 +278,7 @@ import { } from "./agent-conversations.js"; import { withAdapterExecutionPhase } from "@paperclipai/adapter-utils/execution-phase"; import { - recordNativeLocalProcessStop, + recordNativeLocalProcessStopInTransaction, settleNativeRunProcess, } from "./native-local-process-stop.js"; @@ -1968,7 +1968,7 @@ export function heartbeatService( ) .returning() .then((rows) => rows[0] ?? null); - if (cleared) await recordNativeLocalProcessStop(tx as unknown as Db, run); + if (cleared) await recordNativeLocalProcessStopInTransaction(tx as unknown as Db, run); return cleared; }); if (!cleared) { diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index 3d5695339bc..aedc155f5fd 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -24,7 +24,7 @@ import { hasStopOnlyCleanup, settleStopOnlyCleanup, } from "../sandbox-stop-and-retain.js"; -import { discoverSettledNativeRunProcesses, PROCESS_IDENTITY_RECORDED, PROCESS_START_REQUESTED, recordNativeLocalProcessStop, releaseSettledNativeRunProcess, retryPendingSettledNativeRunProcesses } from "../native-local-process-stop.js"; +import { discoverSettledNativeRunProcesses, PROCESS_IDENTITY_RECORDED, PROCESS_START_REQUESTED, recordNativeLocalProcessStopInTransaction, releaseSettledNativeRunProcess, retryPendingSettledNativeRunProcesses } from "../native-local-process-stop.js"; import { pendingCaseHandoffForRun } from "./case-handoff-lifecycle.js"; import { legacyControllerBootId, @@ -2162,7 +2162,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD ) .returning({ id: heartbeatRuns.id }) .then((rows) => rows[0] ?? null); - if (cleared) await recordNativeLocalProcessStop(tx as unknown as Db, run); + if (cleared) await recordNativeLocalProcessStopInTransaction(tx as unknown as Db, run); return cleared; }); if (cleared) claimableNativeRunIds.add(cleared.id); diff --git a/server/src/services/native-local-process-stop.ts b/server/src/services/native-local-process-stop.ts index 85d10f95359..3cbe1bb1aae 100644 --- a/server/src/services/native-local-process-stop.ts +++ b/server/src/services/native-local-process-stop.ts @@ -3,10 +3,10 @@ import { and, asc, desc, eq, exists, gt, inArray, isNotNull, isNull, or, sql } f import { environmentLeases, heartbeatRunEvents, heartbeatRuns, nativeRunFinalizations, type Db } from "@paperclipai/db"; import { notifyRetainedNativeSessionCleanup } from "./retained-native-session-cleanup-notifications.js"; import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { nativeSessionProviderEvidence } from "./native-runtime/native-session-resume.js"; import { hasRemoteTerminationReceipt } from "./remote-execution-termination.js"; import os from "node:os"; -import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; export const PROCESS_START_REQUESTED = "native.process_start_requested"; export const PROCESS_IDENTITY_RECORDED = "native.process_identity_recorded"; @@ -73,6 +73,15 @@ export async function recordNativeLocalProcessStop(db: Db, run: Pick[1]) { + const recorded = await recordNativeLocalProcessStop(tx, run); + if (recorded) await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, run.companyId); + return recorded; +} + /** Only server-authored evidence counts. A later launch invalidates the receipt. */ export async function hasNativeLocalProcessStop(db: Db, companyId: string, runId: string) { const [event] = await db.select({ eventType: heartbeatRunEvents.eventType }) @@ -203,6 +212,8 @@ export async function releaseSettledNativeRunProcess(db: Db, runId: string, opti } await notifyLiveRunRecovery(tx, { companyId: run.companyId, runId: run.id, agentId: run.agentId }); await notifyRetainedNativeSessionCleanup(tx, { companyId: run.companyId, runId: run.id }); + // Local and remote stops both close the run's usage source. + await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, run.companyId); await tx.update(heartbeatRuns).set({ processPid: null, processGroupId: null, processStartedAt: null, updatedAt: new Date(), }).where(eq(heartbeatRuns.id, run.id)); diff --git a/server/src/services/native-runtime/native-restart-recovery.ts b/server/src/services/native-runtime/native-restart-recovery.ts index efb32887539..1baceb4e514 100644 --- a/server/src/services/native-runtime/native-restart-recovery.ts +++ b/server/src/services/native-runtime/native-restart-recovery.ts @@ -18,7 +18,7 @@ import { activeNativeExecutionRunIds, NATIVE_PREPARATION_FENCING_VERSION } from import { PROCESS_IDENTITY_RECORDED, PROCESS_START_REQUESTED, - recordNativeLocalProcessStop, + recordNativeLocalProcessStopInTransaction, } from "../native-local-process-stop.js"; import { randomUUID } from "node:crypto"; import { and, desc, eq, gt, inArray, isNotNull, isNull, lte, ne, or, sql } from "drizzle-orm"; @@ -1264,7 +1264,7 @@ export async function claimNativeRestartRecoveries(input: { claimKind !== "reattach_existing_runner" && claimKind !== "reattach_remote_runner" ) { - await recordNativeLocalProcessStop(tx as unknown as Db, row.run); + await recordNativeLocalProcessStopInTransaction(tx as unknown as Db, row.run); } await tx diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 2f0faafc871..01c60d35d0e 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -2135,11 +2135,11 @@ export function recoveryService( return result?.awaiting === true; } - async function isInvocationBudgetBlocked( + async function readInvocationBudgetBlock( issue: RecoveryIssue, agentId: string, ) { - const budgetBlock = await budgets.getInvocationBlock( + return budgets.getInvocationBlock( issue.companyId, agentId, { @@ -2147,7 +2147,13 @@ export function recoveryService( projectId: issue.projectId, }, ); - return Boolean(budgetBlock); + } + + async function isInvocationBudgetBlocked( + issue: RecoveryIssue, + agentId: string, + ) { + return Boolean(await readInvocationBudgetBlock(issue, agentId)); } async function reconcileUnassignedBlockingIssues(opts?: { companyId?: string }) { @@ -4823,7 +4829,8 @@ export function recoveryService( continue; } } - if (await isInvocationBudgetBlocked(issue, agentId)) { + const budgetBlock = await readInvocationBudgetBlock(issue, agentId); + if (budgetBlock) { const classification = classifyContinuationFailure(latestRun); if ( classification.kind === "deliberate_wait_without_target" || @@ -4845,8 +4852,9 @@ export function recoveryService( issue.status === "in_review" ? EXECUTION_REVIEW_PARTICIPANT_RECOVERY_REASON : undefined, + // Name the actual gate: unpriced or unrecorded usage is not spend over the limit. comment: - "Paperclip cannot safely continue automatic recovery because the original recovery target is over budget. " + + `Paperclip cannot safely continue automatic recovery because the budget does not allow new work: ${budgetBlock.reason} ` + "The source assignment is unchanged and the board must choose the next action.", }); if (updated) { diff --git a/server/src/services/run-cost-accounting.ts b/server/src/services/run-cost-accounting.ts index ea9f8bd2b17..aaa192c08f0 100644 --- a/server/src/services/run-cost-accounting.ts +++ b/server/src/services/run-cost-accounting.ts @@ -1,7 +1,7 @@ import { notifyStrandedAgentStatus } from "./live-run-recovery-notifications.js"; import { usdToUnits, unitsToCents } from "@paperclipai/shared"; import { and, asc, eq, inArray, or, sql } from "drizzle-orm"; -import { budgetReservations, agents, heartbeatRuns, nativeRunFinalizations, issues, projects, decisionInvocations, promptInjectionInvocations, type Db } from "@paperclipai/db"; +import { budgetReservations, agents, heartbeatRunEvents, heartbeatRuns, nativeRunFinalizations, issues, projects, decisionInvocations, promptInjectionInvocations, type Db } from "@paperclipai/db"; import { withOperationAccountingTransaction } from "./accounting-transaction.js"; import { withBudgetSpan } from "./budget-tracing.js"; import { conflict } from "../errors.js"; @@ -32,12 +32,39 @@ async function readAccountingClosure(db: Db, runId: string) { preProviderFailure: sql`(${heartbeatRuns.resultJson}->'executionRecovery'->>'providerWorkStarted' = 'false' or ${heartbeatRuns.usageJson}->>'accountingProviderWorkStarted' = 'false') is true`, coordinatorId: nativeRunFinalizations.runId, resultId: nativeRunFinalizations.resultId, phase: nativeRunFinalizations.phase, + processPid: heartbeatRuns.processPid, processGroupId: heartbeatRuns.processGroupId, }).from(heartbeatRuns).leftJoin(nativeRunFinalizations, and( eq(nativeRunFinalizations.runId, heartbeatRuns.id), eq(nativeRunFinalizations.companyId, heartbeatRuns.companyId), )).where(eq(heartbeatRuns.id, runId)); return closure; } +/** A failed native run's receipt source is closed only with positive stop + * evidence: recovery ended in terminal failure, the run no longer records a + * process identity, and its latest server-authored process event is the + * verified local stop. A deadline failure whose process still runs keeps its + * identity and has no stop event, so it stays pending. Missing identity alone is not proof. A cancelled run + * keeps its accounting drain: Stop recovery can still deliver the final + * receipt through the original or a reattached recorder. Remote runs keep + * their debt until a receipt arrives. One predicate serves the reconciler + * scan and settlement so they cannot disagree. */ +const closedNativeReceiptSource = sql`${heartbeatRuns.runtimeMode} = 'native' and ${heartbeatRuns.status} = 'failed' + and ${heartbeatRuns.processPid} is null and ${heartbeatRuns.processGroupId} is null + and exists (select 1 from ${nativeRunFinalizations} where ${nativeRunFinalizations.runId} = ${heartbeatRuns.id} + and ${nativeRunFinalizations.companyId} = ${heartbeatRuns.companyId} + and ${nativeRunFinalizations.resultId} is null and ${nativeRunFinalizations.phase} = 'terminal_failure') + and (select process_event.event_type from ${heartbeatRunEvents} process_event + where process_event.company_id = ${heartbeatRuns.companyId} and process_event.run_id = ${heartbeatRuns.id} + and process_event.source_event_id is null + and process_event.event_type in ('native.process_start_requested', 'native.process_identity_recorded', 'native.local_process_stopped') + order by process_event.seq desc limit 1) = 'native.local_process_stopped'`; + +async function nativeReceiptSourceClosed(db: Db, runId: string) { + const [row] = await db.select({ closed: sql`${closedNativeReceiptSource}` }).from(heartbeatRuns) + .where(eq(heartbeatRuns.id, runId)); + return row?.closed === true; +} + /** 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) { @@ -100,8 +127,12 @@ export async function accountRunCost(db: Db, runId: string, hooks: BudgetService } // A stop can mark the run terminal before the provider's shutdown receipt // arrives. Keep the debt pending instead of acknowledging an invented zero - // and permanently discarding that late receipt. - if (!preProviderFailure && object(run.usageJson).accountingReceiptReady === false) return false; + // and permanently discarding that late receipt. Once native recovery has + // 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; + if (receiptUnavailable && !(await nativeReceiptSourceClosed(tx, run.id))) return false; if (!preProviderFailure) await ensureRuntimeAccountingBaseline(tx, run.companyId, run.agentId); // Acknowledgement is part of the same transaction. Budget checks inside // this transaction must not mistake this receipt for missing accounting. @@ -141,7 +172,7 @@ export async function accountRunCost(db: Db, runId: string, hooks: BudgetService providerRequestId: text(usage.providerRequestId), pricingProvenance: object(usage.pricingProvenance), model: text(usage.model) ?? "unknown", billingType, - costStatus: (costUsd === null || usage.costStatus === "unpriced") && billingType !== "subscription_included" && hasBillableEvidence ? "unpriced" : usage.costStatus === "estimated" ? "estimated" : "reported", + costStatus: (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, }; @@ -174,7 +205,10 @@ export async function accountRunCost(db: Db, runId: string, hooks: BudgetService // Late usage reported by a later run can arrive before this charge posts. const lateDebt = usage.accountingLateReceiptPending === true; await tx.update(heartbeatRuns).set({ costAccountingPending: lateDebt, costAccountedAt: new Date(), accountingProjectionVersion: "v2", - accountingLastError: lateDebt ? "late_usage_requires_reconciliation" : null, accountingLastAttemptAt: new Date() }).where(eq(heartbeatRuns.id, run.id)); + 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` } : {}), + }).where(eq(heartbeatRuns.id, run.id)); return true; }), { "budget.accounting.owner": "run" }); for (const file of recoveredFiles) { @@ -197,7 +231,8 @@ function actionableRunAccounting() { 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')`, + sql`(${heartbeatRuns.usageJson}->>'accountingReceiptReady' is distinct from 'false' or ${heartbeatRuns.resultJson}->'executionRecovery'->>'providerWorkStarted' = 'false' or ${heartbeatRuns.usageJson}->>'accountingProviderWorkStarted' = 'false' + or (${closedNativeReceiptSource}))`, )), ); } diff --git a/server/src/services/task-interruption-presentation.test.ts b/server/src/services/task-interruption-presentation.test.ts index fab317dc824..4f29d3a02d9 100644 --- a/server/src/services/task-interruption-presentation.test.ts +++ b/server/src/services/task-interruption-presentation.test.ts @@ -113,6 +113,14 @@ it("explains a persisted Retry wait without exposing the saved payload", () => { .toMatchObject({ state: "waiting_for_user", userAction: "resume" }); }); +it("names a budget gate instead of generic execution checks for a refused Retry", () => { + const failed = { ...run, errorCode: "native_provider_terminal_failed", resultJson: null }; + expect(projectTaskInterruption(issue, { run: failed, agent, paused: false, approval: false, + wait: { id: "retry", requestedAt: new Date(), sourceRunId: failed.id, executionWaitReason: "budget.blocked" } })) + .toMatchObject({ state: "failed", canRetry: true, retryWait: { reason: "budget.blocked", + message: "Retry is waiting for the budget to allow new work. Check Costs for unrecorded or unpriced usage." } }); +}); + it.each(["cleanup_quarantined", "cancelled_by_operator", "retry_source_changed"])("does not offer Retry for terminal Retry waits: %s", reason => { const failed = { ...run, errorCode: "native_provider_terminal_failed", resultJson: null }; expect(projectTaskInterruption(issue, { run: failed, agent, paused: false, approval: false, diff --git a/server/src/services/task-interruption-presentation.ts b/server/src/services/task-interruption-presentation.ts index dec56770f76..c24db5150b4 100644 --- a/server/src/services/task-interruption-presentation.ts +++ b/server/src/services/task-interruption-presentation.ts @@ -71,6 +71,7 @@ const retryWaitMessages: Record = { process_running: "Retry is waiting for the previous execution to stop. Your message is saved.", execution_active: "Retry is waiting for the current run to finish. Your message is saved.", decision_pending: "Retry is waiting for a pending approval or question to be resolved. Your message is saved.", + "budget.blocked": "Retry is waiting for the budget to allow new work. Check Costs for unrecorded or unpriced usage.", finalizing_saved_result: "Paperclip is finishing the saved result before Retry can continue.", cleanup_quarantined: "Retry cannot continue this attempt. Send a new message after the previous execution has stopped.", cancelled_by_operator: "This attempt was stopped. Send a new message to continue from saved work.", diff --git a/server/src/services/usage-receipts.ts b/server/src/services/usage-receipts.ts index d858d841c8f..595daaa8358 100644 --- a/server/src/services/usage-receipts.ts +++ b/server/src/services/usage-receipts.ts @@ -127,14 +127,24 @@ export async function persistUsageReceiptInTransaction(tx: Db, raw: UsageReceipt const authoritative = envelope.receipt.pricingProvenance && !["rate_card", "unknown"].includes(envelope.receipt.pricingProvenance.source); const rateCardOnly = envelope.producerEvent && envelope.receipt.pricingProvenance?.source === "rate_card"; - const changed = envelope.lateDelta !== undefined + // A charge settled without its final receipt accepts a complete receipt's + // own price, including a rate-card price, as a correction. Reconciliation + // can only post a receipt that carries its price; token changes are + // detected below as usual. + const ownPrice = envelope.receipt.cacheAdjustedCostUsd ?? envelope.receipt.costUsdExact ?? envelope.receipt.costUsd; + const unavailableClosed = previous.accountingReceiptUnavailable === true && envelope.receipt.complete === true + && ownPrice != null && envelope.receipt.costStatus !== "unpriced" && envelope.receipt.usageBasis !== "session_cumulative"; + const changed = unavailableClosed || envelope.lateDelta !== undefined || (envelope.producerEvent?.eventType === "usage.reported" && !envelope.receipt.usage) || envelope.receipt.usageBasis === "session_cumulative" || ["provider", "biller", "model", "billingType"].some(key => { const value = envelope.receipt[key as keyof typeof envelope.receipt]; return value !== undefined && value !== null && value !== previous[key]; }) || Object.entries(envelope.receipt.usage ?? {}).some(([key, value]) => previous[key] !== value) - || (!rateCardOnly && incomingMoney !== null && (acceptedMoney === null || compareCents(incomingMoney, acceptedMoney) !== 0)) + // A projection alone cannot be posted against a charge settled without + // its receipt; only the receipt's own price (above) can correct it. + || (!rateCardOnly && incomingMoney !== null && (previous.accountingReceiptUnavailable !== true || ownPrice != null) + && (acceptedMoney === null || compareCents(incomingMoney, acceptedMoney) !== 0)) || (!rateCardOnly && envelope.receipt.costStatus !== undefined && envelope.receipt.costStatus !== comparison.costStatus) || (authoritative && receiptFingerprint(envelope.receipt.pricingProvenance) !== receiptFingerprint(comparison.pricingProvenance ?? null)); @@ -311,7 +321,8 @@ export async function reconcileLateRunUsageReceiptInTransaction(tx: Db, run: typ : receipt?.billingType === "subscription" ? "subscription_included" : receipt?.billingType ?? "unknown"; const biller = receipt?.biller ?? receipt?.provider ?? "unknown"; const qualified = late && receipt?.complete && receipt.usageBasis !== "session_cumulative" - && late.sourceId === previous.accountingReceiptSourceId && sameTokens && authoritative + && late.sourceId === previous.accountingReceiptSourceId && sameTokens + && (authoritative || (previous.accountingReceiptUnavailable === true && receipt.pricingProvenance?.source === "rate_card")) && rawAmount != null && receipt.costStatus !== "unpriced" && events.length === 1 && receipt.model === events[0].event.model && receipt.provider === events[0].event.provider && biller === events[0].event.biller && billingType === events[0].event.billingType