diff --git a/doc/COST-ACCOUNTING.md b/doc/COST-ACCOUNTING.md index 4363f279cdf..54d0472e16a 100644 --- a/doc/COST-ACCOUNTING.md +++ b/doc/COST-ACCOUNTING.md @@ -174,3 +174,10 @@ edits deliver hard-stop cancellation after their transaction commits. Status-card update costs retain the ledger's fractional-cent storage precision. The mutation gate retains partial results and per-run reports/logs under `coverage/accounting/` when its baseline fails or a mutation survives. + +Failed native turns persist observed run-delta usage before semantic-result +finalization. A matching terminal failure, cancellation, or interruption closes +that receipt; missing terminal evidence and session-only totals remain pending. +For multiple provider attempts, known prices remain in the exact spend total +when another attempt is unpriced. The aggregate stays marked unpriced, so the +configured unknown-price policy still controls whether new work may start. diff --git a/server/src/__tests__/cost-accounting-operations-edges.test.ts b/server/src/__tests__/cost-accounting-operations-edges.test.ts index 156d3092c0d..6a97dcc5978 100644 --- a/server/src/__tests__/cost-accounting-operations-edges.test.ts +++ b/server/src/__tests__/cost-accounting-operations-edges.test.ts @@ -175,11 +175,30 @@ const support = await getEmbeddedPostgresTestSupport(); expect(result).toMatchObject({costUsdExact:"0.040000001",complete:true,usage:{inputTokens:14,cachedInputTokens:22,outputTokens:6}}); await recorder.capture({attemptId:randomUUID(),usage,complete:false,billingType:"subscription"}); const uncertain = await recorder.complete({exitCode:1,signal:null,timedOut:false,usage}); - expect(uncertain).toMatchObject({costUsd:null,costUsdExact:null,billingType:"unknown"}); + expect(uncertain).toMatchObject({costUsd:0.040000001,costUsdExact:"0.040000001",costStatus:"unpriced",billingType:"unknown"}); const partial = await recorder.complete({exitCode:1,signal:null,timedOut:false}); expect(partial.complete).toBe(false); expect(partial.usage?.inputTokens).toBe(21); }); + it.each(["block", "allow"] as const)("keeps known attempted-run spend when unknown prices %s new work", async (unpricedUsagePolicy) => { + const f = await fixture(), budgets = budgetService(db); + await budgets.upsertPolicy(f.company.id, { scopeType: "company", scopeId: f.company.id, amount: 100, unpricedUsagePolicy }, "board"); + await reserveRunBudget(db, f.company.id, f.run.id, null); + const recorder = await createRunUsageRecorder(db, { companyId: f.company.id, runId: f.run.id, adapterType: "process" }, path.join(directory, randomUUID())); + await recorder.capture({ attemptId: randomUUID(), complete: true, billingType: "metered_api", usage, costUsdExact: "0.010000001" }); + const aggregate = await recorder.capture({ attemptId: randomUUID(), complete: true, billingType: "metered_api", usage, costUsd: null, costStatus: "unpriced" }); + expect(aggregate).toMatchObject({ costUsdExact: "0.010000001", costStatus: "unpriced", complete: true }); + await db.update(heartbeatRuns).set({ status: "failed", finishedAt: new Date() }).where(eq(heartbeatRuns.id, f.run.id)); + expect(await accountRunCost(db, f.run.id)).toBe(true); + expect(await accountRunCost(db, f.run.id)).toBe(false); + expect(await costService(db).summary(f.company.id)).toMatchObject({ spendCentsExact: "1.0000001", unpricedEventCount: 1, pricingComplete: false, pendingRunCount: 0 }); + const [runtime] = await db.select().from(agentRuntimeState).where(eq(agentRuntimeState.agentId, f.agent.id)); + expect(runtime.totalCostCents).toBe(1.0000001); + const [reservation] = await db.select().from(budgetReservations).where(eq(budgetReservations.runId, f.run.id)); + expect(reservation.state).toBe("settled"); + expect((await budgets.getInvocationBlock(f.company.id, f.agent.id)) !== null).toBe(unpricedUsagePolicy === "block"); + }); + it("refuses invalid checkpoints and does not certify a run after capture failure", async () => { const f = await fixture(); const recorder = await createRunUsageRecorder(db,{companyId:f.company.id,runId:f.run.id,adapterType:"process"},path.join(directory,randomUUID())); diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index 185f3416533..d07fecace03 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -191,6 +191,17 @@ const state = vi.hoisted(() => ({ release: null as null | (() => void), })); +const accountingEvents = vi.hoisted(() => ({ committed: undefined as undefined | ((event: PrpEvent) => Promise) })); +vi.mock("./paperclip-control-plane-port.js", async importOriginal => { + const original = await importOriginal(); + return { ...original, PaperclipControlPlanePort: class extends original.PaperclipControlPlanePort { + constructor(...args: ConstructorParameters) { + super(...args); + accountingEvents.committed = args[2]?.onCommittedEvent; + } + } }; +}); + const grokCopyBack = vi.hoisted(() => vi.fn(async (_input: { readSandboxAuth: () => Promise; hostHomeDir: string }) => undefined)); vi.mock("@paperclipai/adapter-grok-local/server", async importOriginal => ({ ...await importOriginal(), @@ -4873,7 +4884,7 @@ function leaseDb( where: () => query, orderBy: () => query, for: () => query, - limit: () => Promise.resolve(rows), + limit: () => query, }; return query; }, @@ -4881,7 +4892,9 @@ function leaseDb( const insert = (table: unknown) => ({ values: (values: Record) => { updates.push({ table, values }); - return { returning: async () => [values] }; + const query = { returning: async () => [values], onConflictDoUpdate: () => query, onConflictDoNothing: () => query, + then: Promise.resolve([values]).then.bind(Promise.resolve([values])) }; + return query; }, }); const tx = { @@ -5134,6 +5147,27 @@ describe("native startup restart detachment", () => { }); }); +describe("native failed-turn accounting", () => { + beforeEach(() => vi.spyOn(issueServiceModule, "issueService").mockReturnValue({ update: vi.fn(async () => ({ status: "blocked", statusVersion: 1 })) } as unknown as ReturnType)); + afterEach(() => vi.restoreAllMocks()); + it.each(["turn.failed", "turn.cancelled", "turn.interrupted", "missing_terminal", "session_total", "empty_usage", "different_turn"])("preserves run usage for %s without certifying incomplete evidence", async (scenario) => { + const onUsage = vi.fn(async (_receipt: import("@paperclipai/adapter-utils").AdapterUsageCheckpoint) => {}); + const failure = new Error("provider_transport_failed: no semantic result"); + state.execute.mockReset().mockImplementationOnce(async () => { + const usage = { inputTokens: 120, cacheReadTokens: 100, outputTokens: 10, providerCostUsd: 0.25 }; + await accountingEvents.committed!({ eventType: "item.completed", turnId: "turn", emittedAt: new Date().toISOString(), payload: { kind: "usage", usage: scenario === "session_total" ? { total: usage } : { runDelta: scenario === "empty_usage" ? { requests: 1 } : usage } } } as unknown as PrpEvent); + if (scenario !== "missing_terminal") await accountingEvents.committed!({ eventType: scenario.startsWith("turn.") ? scenario : "turn.failed", turnId: scenario === "different_turn" ? "another-turn" : "turn", emittedAt: new Date().toISOString(), payload: { status: "failed" } } as unknown as PrpEvent); + throw failure; + }); + await expect(executePaperclipNativeSession({ db: leaseDb(), execution: { ...execution, provider: { kind: "codex", model: "gpt-6-astra" } } as NativeExecutionInput, runnerInstanceId: "runner", onUsage })).rejects.toBe(failure); + if (["session_total", "empty_usage"].includes(scenario)) expect(onUsage).not.toHaveBeenCalled(); + else { + expect(onUsage).toHaveBeenLastCalledWith(expect.objectContaining({ complete: scenario.startsWith("turn."), costUsd: 0.25, usageBasis: "per_run", usage: { inputTokens: 20, cachedInputTokens: 100, outputTokens: 10 } })); + expect(onUsage.mock.calls[0]?.[0]).toMatchObject({ complete: false }); + } + }); +}); + describe("native resumed preparation timing", () => { it("keeps answered-question ingress at the run root rather than charging it to preparation", async () => { const answeredAtMs = Date.now(); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 4ba162a2e29..4b7a7851f8e 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -7838,6 +7838,22 @@ async function executePaperclipNativeSessionWithinScope( binding: { ...input.execution.binding, normalizedSessionId: nativeSessionKey(input.execution), runnerSourceInstanceId: effectiveRunnerInstanceId }, resolve: resolveNativeRuntimeRequest, }); + let observedAccountingUsage: Record | null = null; + let observedAccountingTurn: string | undefined; + const persistAccountingUsage = async (usage: Record | null, complete: boolean) => { + const accountingUsage = normalizeNativeUsage(usage, { inputIncludesCacheReads: input.execution.provider.kind === "codex" }); + const accountingCost = nativeUsageCostUsd(usage, input.execution.provider); + const billing = resolveNativeBilling(input.execution.provider, input.runnerEnvironment, input.billingIdentity); + await input.onUsage?.({ usage: accountingUsage, ...billing, complete, usageBasis: "per_run", + model: input.execution.provider.model ?? "unknown", costUsd: accountingCost ?? null }); + if (!input.onUsage) await input.db.update(heartbeatRuns).set({ + costAccountingPending: true, + usageJson: sql`coalesce(${heartbeatRuns.usageJson}, '{}'::jsonb) || ${JSON.stringify({ + ...accountingUsage, ...billing, accountingReceiptReady: complete, model: input.execution.provider.model ?? "unknown", + costUsd: accountingCost ?? null, usageSource: "per_run", + })}::jsonb`, + }).where(and(eq(heartbeatRuns.id, input.execution.binding.runId), eq(heartbeatRuns.companyId, input.execution.binding.companyId), isNull(heartbeatRuns.costAccountedAt))); + }; let completedConversationReply: PrpEvent | null = null; const controlPlane = new PaperclipControlPlanePort( input.db, @@ -7854,6 +7870,25 @@ async function executePaperclipNativeSessionWithinScope( }, { onCommittedEvent: async (event) => { + const payload = record(event.payload); + const usage = record(payload.usage); + const delta = record(usage.runDelta); + if (event.eventType === "turn.started" && observedAccountingUsage) { + observedAccountingTurn = undefined; + await persistAccountingUsage(observedAccountingUsage, false); + } + // Only an explicit run delta is safe here: session totals can include + // earlier work. Save provisional evidence before ancillary finalizers. + if (payload.kind === "usage" && numericUsageField(delta, ["inputTokens"]) !== undefined + && numericUsageField(delta, ["outputTokens"]) !== undefined) { + observedAccountingUsage = { runDelta: usage.runDelta }; + observedAccountingTurn = event.turnId; + await persistAccountingUsage(observedAccountingUsage, false); + } + if (["turn.failed", "turn.cancelled", "turn.interrupted"].includes(event.eventType) + && observedAccountingUsage && event.turnId === observedAccountingTurn) { + await persistAccountingUsage(observedAccountingUsage, true); + } await toolTrace.observe(event); if (event.eventType === "item.completed" && record(event.payload).kind === "agentMessage" && @@ -8602,18 +8637,8 @@ async function executePaperclipNativeSessionWithinScope( ); // Persist provider accounting before any workspace/issue finalization. A // detached controller or failed finalizer must not lose a completed turn. - const accountingUsage = normalizeNativeUsage(native.usage, { inputIncludesCacheReads: input.execution.provider.kind === "codex" }); - const accountingCost = nativeUsageCostUsd(native.usage, input.execution.provider); - const billing = resolveNativeBilling(input.execution.provider, input.runnerEnvironment, input.billingIdentity); - await input.onUsage?.({ usage: accountingUsage, ...billing, complete: true, usageBasis: "per_run", - model: input.execution.provider.model ?? "unknown", costUsd: accountingCost ?? null }); - if (!input.onUsage) await input.db.update(heartbeatRuns).set({ - costAccountingPending: true, - usageJson: sql`coalesce(${heartbeatRuns.usageJson}, '{}'::jsonb) || ${JSON.stringify({ - ...accountingUsage, ...billing, accountingReceiptReady: true, model: input.execution.provider.model ?? "unknown", - costUsd: accountingCost ?? null, usageSource: "per_run", - })}::jsonb`, - }).where(and(eq(heartbeatRuns.id, input.execution.binding.runId), eq(heartbeatRuns.companyId, input.execution.binding.companyId), isNull(heartbeatRuns.costAccountedAt))); + native = { ...native, usage: native.usage ?? observedAccountingUsage }; + await persistAccountingUsage(native.usage, true); try { await completeManagedNativeCredentialTurn(managedCredentialSession); } catch { diff --git a/server/src/services/run-cost-accounting.ts b/server/src/services/run-cost-accounting.ts index 895e2443acb..b1126462a8b 100644 --- a/server/src/services/run-cost-accounting.ts +++ b/server/src/services/run-cost-accounting.ts @@ -62,7 +62,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 && billingType !== "subscription_included" && hasBillableEvidence ? "unpriced" : usage.costStatus === "estimated" ? "estimated" : "reported", + costStatus: (costUsd === null || usage.costStatus === "unpriced") && 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, }; diff --git a/server/src/services/usage-receipts.ts b/server/src/services/usage-receipts.ts index c629b41c567..0421712dda4 100644 --- a/server/src/services/usage-receipts.ts +++ b/server/src/services/usage-receipts.ts @@ -152,13 +152,15 @@ export async function createRunUsageRecorder(db: Db, input: { companyId: string; if (parts.length > 1) { let total = 0n, priced = true; for (const part of parts) { - const cost = part.cacheAdjustedCostUsd ?? part.costUsdExact ?? part.costUsd; - if (cost == null) priced = false; else total += usdToUnits(cost); + const subscription = ["subscription", "subscription_included"].includes(part.billingType ?? ""); + const cost = subscription ? 0 : part.cacheAdjustedCostUsd ?? part.costUsdExact ?? part.costUsd; + if (cost == null || (!subscription && part.costStatus === "unpriced")) priced = false; + if (cost != null) total += usdToUnits(cost); } if (new Set(parts.map(part => part.billingType ?? "unknown")).size > 1) receipt.billingType = "unknown"; const decimal = `${total / 1_000_000_000n}.${String(total % 1_000_000_000n).padStart(9, "0")}`; - receipt.costUsdExact = priced ? decimal : null; - receipt.costUsd = priced ? Number(decimal) : null; + receipt.costUsdExact = decimal; + receipt.costUsd = Number(decimal); receipt.cacheAdjustedCostUsd = null; receipt.usageByModel = undefined; receipt.costStatus = priced && parts.some(part => part.costStatus === "estimated") ? "estimated" : priced ? "reported" : "unpriced";