From aa8372fd5dd75fadae96665614885e68d2af9265 Mon Sep 17 00:00:00 2001 From: Nicky Leach Date: Wed, 4 Nov 2026 15:30:55 -0800 Subject: [PATCH] fix(accounting): post late usage to its owning run (#18991) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Native Runner sessions keep one provider conversation across task turns, so Paperclip accounts usage per run. > - Codex can report more usage for a turn after its run's charge was posted. > - The Runner then marked the session `reconciliation_required`. If another run was attached, it also failed that run. Core refused the delivery with `native_usage_after_finalization_requires_reconciliation`. > - That usage has already happened. Blocking the session cannot undo it, and the debt waited for an operator while the run totals stayed wrong. > - Codex totals are cumulative, and the Runner already knows the owning run in most cases. So the correct amount can be posted automatically. > - This pull request reports the exact increment with its owner, and posts it to the owning run as a separate cost event. > - The benefit is that run accounting becomes correct without stopping work, and an old run's usage can no longer fail a newer run. ## Linked Issues or Issue Description No issue exists. I searched open and closed pull requests and issues for "late usage", `native_usage_after_finalization_requires_reconciliation` and `codex_usage_accounting_integrity`. I found no duplicate. **What happened?** Code review found this while diagnosing session recovery failures. When Codex reported usage above the known total for a turn that was not active, the Runner blocked the session. If that turn belonged to a previous run, the Runner also failed the attached run with `codex_usage_accounting_integrity`. Core kept the receipt as debt, and could correct it automatically only if the token counts were unchanged. **Expected behavior** Late usage does not stop any run or session. The charge goes to the run that used the tokens, and it is auditable. **Steps to reproduce** 1. Finish a Codex turn on a native Runner session and attach the next run. 2. Have Codex send `thread/tokenUsage/updated` with a higher total for the finished turn. 3. On the unfixed code, the attached run fails and later attaches are refused. `late_codex_usage_requires_durable_reconciliation_before_or_after_handoff` encoded this behavior. **Paperclip version or commit** `master` at `04a24f4e06`. **Deployment mode** Any deployment that uses a native Runner with Codex. ## What Changed - **Runner:** late usage no longer sets `reconciliation_required` or fails a run. For a turn that belongs to a previous run, the Runner emits `usage.reported` with the increment in `runDelta` and the owner in `reconciliation.priorOrigin`. `runDeltaAvailable` stays `false`, so an older Core does not bill it to the attached run. - **`CodexRunUsage::absorb_late_usage`:** treats only usage above the known thread total as late, and keeps the attached run's delta unchanged. A report at or below that total adds nothing, so the sum across runs always equals the thread total and a replay cannot charge twice. - **Delivery:** `captureLateUsageForOwner` stores the increment as a late receipt on the owning run. The owner must be a finished run in the same company and native session. If the Runner cannot name the owner, Core uses the session's last finished run and records the attribution as `inferred`. - **Local Runner:** late usage stays with the run that is active when it arrives, or the next run. Usage reported between two runs stays in the next run's delta, so the total stays exact. - **Accounting:** `reconcileLateRunUsageReceiptInTransaction` posts a token increase as a new cost event with the key `heartbeat::late:`. It updates the run totals so the accounting integrity check still agrees, records `cost.late_usage_posted` in the activity log, and queues budget evaluation. - **No more refusal:** `persistUsageReceipt` no longer throws after it saves late evidence, so late usage cannot wedge a Runner outbox. - **Covered receipts:** an older same-source receipt at or below the accepted totals, with matching billing, is cleared because a newer cumulative receipt already covers it. Conflicting billing stays as debt. - **Older Runners:** a zero-filled invalidation stays on the delivering run. - **Unchanged:** a newer lower total, a changed model or billing type, and per-model receipts stay as visible accounting debt for an operator. - **Unchanged:** an unexpected provider message during the warm attachment drain still fails closed. A separate change addresses that gate. - **Docs:** added "Late provider usage" to `doc/execution-semantics.md`. ## Verification - `cd packages/paperclip-runner/runner && cargo test --locked -p paperclip-runner-core --test codex_provider`: 99 passed, 2 ignored. - `cargo test --locked -p paperclip-runner-core --lib codex_usage`: 7 passed. This includes new tests for the out-of-order and repeated report cases. - `cd server && pnpm exec vitest run src/services/run-cost-accounting.test.ts src/services/native-runtime/native-usage-delivery.integration.test.ts src/services/native-runtime/native-recovery-usage-delivery.test.ts src/__tests__/cost-data-quality.test.ts`: passed against embedded PostgreSQL before the last review fixes, and the accounting and delivery files were re-run after them. - `npx tsc --noEmit -p .` in `server` reports no errors. - New tests cover exact and inferred owners, repeated delivery, late usage that arrives before the owner's charge posts, another session's run, an older Runner's report with no increment, a same-run increase, and a lower total. They also run the accounting integrity check after posting. - **Review:** Greptile Apex scored 1/5, 4/5, 3/5, 2/5 and 4/5 while findings were fixed, then 5/5 at the current head. - **Local Runner:** driver tests cover usage reported between two runs, including across a restore. No end-to-end local run has exercised it. - **Not run locally:** the full test suite. CI provides that coverage. - **Not verified live:** a read-only check of a test deployment found no late receipts in its history, so this path has only run against fixtures. ## Risks - **Run totals can rise after a run finishes.** This is deliberate. The correction is a separate cost event and an activity log entry. - **Budgets are corrected after the fact.** A late charge can put an agent over budget after later work has started. Budget evaluation runs after each correction, but it cannot prevent that work. - **Local sessions accept imperfect attribution by design.** Their late usage can land on the active or next run, because local sessions have no durable path for routing it. This also applies after a Stop or a pause. The total stays exact. - **Out-of-order reports can leave tokens on the wrong run in the session.** If a prior turn's report arrives after the attached run has already counted past that total, those tokens stay on the attached run. Moving them would need a lower correction to the attached run. The total stays exact. - **Inferred attribution can name the wrong run in the session.** The usage still belongs to that session and the total stays correct. The attribution is recorded as `inferred`. - **A session with no finished run charges the delivering run,** as an inferred late charge after its own charge posts. - **Rollout order:** deploy Core before the Runner. An older Core would keep this event on the attached run as incomplete usage, and would still refuse a same-run late receipt. - **Migrations:** this change has no migration. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be reworked. ## Model Used - Anthropic Claude Opus 5.5 (`claude-opus-5-5`), through Claude Code, with tool use and code execution. A human directed the design and approved the scope. ## 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 - [x] All Paperclip CI gates are green - [x] 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 --- doc/execution-semantics.md | 18 ++ doc/testing-native-remote-continuity.md | 2 +- .../crates/runner-core/src/codex_usage.rs | 71 +++++++ .../runner-core/src/provider_backend.rs | 74 ++++--- .../runner-core/tests/codex_provider.rs | 119 +++++------ .../src/contracts/harness-driver.ts | 2 +- .../codex-app-server-driver.events.test.ts | 45 ++++ .../drivers/codex/codex-harness-session.ts | 8 +- .../drivers/codex/codex-resume-usage.test.ts | 31 +++ .../codex/codex-session-notifications.ts | 1 + .../src/drivers/codex/codex-usage-baseline.ts | 5 +- .../native-usage-delivery.integration.test.ts | 197 ++++++++++++++++-- .../native-runtime/native-usage-delivery.ts | 36 +++- .../src/services/run-cost-accounting.test.ts | 94 ++++++++- server/src/services/run-cost-accounting.ts | 5 +- server/src/services/usage-receipts.ts | 136 ++++++++++-- 16 files changed, 706 insertions(+), 138 deletions(-) diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 3dad5628de7..7ac3ff61700 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1683,6 +1683,24 @@ interrupt. A verified native stop must not become an unexplained failure simply because it came from a subtree action. The explicit pause hold still prevents future execution until Resume, and missing stop proof still blocks continuation. +### Late provider usage + +A provider can report more usage after its run's charge was posted. That usage +already happened, so it never fails a run or blocks a session. Codex usage totals +are cumulative, so the Runner reports the exact increment above the known total and +names its owning run when the provider turn proves the owner. A report at or below +the known total was already counted by the run that saw it, and it adds nothing, so +the sum across runs always equals the thread total. For a remote session, Core posts the increment to that run as a +separate cost event and records `cost.late_usage_posted` in the activity log. If +the owner is not proven, Core charges the session's last finished run, or the +delivering run if the session has none, and records the attribution as inferred. A late charge can put an agent over budget after later +work started; budget evaluation runs after each correction. Conflicting evidence, +such as a lower total or a changed model, stays visible as accounting debt for +an operator to review. It does not change the charge or stop the session. A local +session has no durable path for this, so its late usage stays with the run that is +active when it arrives, or the next run. The total stays exact, but the attribution +can be off by those tokens. + ### Session settlement after a run Attaching a later run to a session asks whether anything is live or waiting: an diff --git a/doc/testing-native-remote-continuity.md b/doc/testing-native-remote-continuity.md index 810c16e9060..4f3a1a7c67e 100644 --- a/doc/testing-native-remote-continuity.md +++ b/doc/testing-native-remote-continuity.md @@ -68,7 +68,7 @@ without target tool dispatch or a fabricated response. The staged restart helper and classifier tests cover target discovery before its first session snapshot, exact encrypted source/header/artifact bindings and the existing compatibility rebinder. Source accounting tests use real PostgreSQL to distinguish immutable -receipt replay, equal finalized usage and late usage requiring reconciliation. +receipt replay, equal finalized usage and late usage posted to its owning run. Handoff crash tests must restore the exact queued command or authenticated receipt, reuse the verified physical Runner and preserve endpoint binding. Test both direct callbacks and the same guest listener when that ingress is enabled; diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/codex_usage.rs b/packages/paperclip-runner/runner/crates/runner-core/src/codex_usage.rs index 41e282ffe97..89480675680 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/codex_usage.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/codex_usage.rs @@ -65,6 +65,31 @@ impl CodexRunUsage { .then(|| self.latest.clone()); } + /// Records usage above the known thread total and returns that increment. + /// Raising the baseline with the total leaves this run's delta unchanged, so + /// the sum across runs always equals the total. Anything at or below the known + /// total was already counted, so a replay or out-of-order report adds nothing. + pub(crate) fn absorb_late_usage(&mut self, total: &Value) -> Value { + let mut late = serde_json::Map::new(); + for (key, value) in total.as_object().into_iter().flatten() { + let Some(reported) = value.as_u64() else { + continue; + }; + let known = self.latest.get(key).and_then(Value::as_u64).unwrap_or(0); + let increment = reported.saturating_sub(known); + late.insert(key.clone(), json!(increment)); + if increment == 0 { + continue; + } + self.latest.insert(key.clone(), json!(reported)); + if let Some(baseline) = self.baseline.as_mut() { + let base = baseline.get(key).and_then(Value::as_u64).unwrap_or(0); + baseline.insert(key.clone(), json!(base + increment)); + } + } + Value::Object(late) + } + pub(crate) fn observe(&mut self, total: &Value, historical: bool) -> (Value, bool) { let measurement: BTreeMap = total .as_object() @@ -183,6 +208,52 @@ mod tests { ); } + #[test] + fn late_usage_moves_to_its_owner_without_changing_the_attached_runs_delta() { + let mut usage = CodexRunUsage::new(true); + usage.observe(&json!({"inputTokens":100,"outputTokens":10}), true); + usage.begin_run(); + let (paid, _) = usage.observe(&json!({"inputTokens":130,"outputTokens":14}), false); + assert_eq!(paid, json!({"inputTokens":30,"outputTokens":4})); + let late = usage.absorb_late_usage(&json!({"inputTokens":150,"outputTokens":14})); + assert_eq!(late, json!({"inputTokens":20,"outputTokens":0})); + let (next, complete) = usage.observe(&json!({"inputTokens":160,"outputTokens":20}), false); + assert!(complete); + assert_eq!(next, json!({"inputTokens":40,"outputTokens":10})); + } + + #[test] + fn already_counted_late_usage_stays_with_the_run_that_counted_it() { + let mut usage = CodexRunUsage::new(true); + usage.observe(&json!({"inputTokens":100,"outputTokens":10}), true); + usage.begin_run(); + usage.observe(&json!({"inputTokens":150,"outputTokens":15}), false); + let late = usage.absorb_late_usage(&json!({"inputTokens":120,"outputTokens":12})); + assert_eq!(late, json!({"inputTokens":0,"outputTokens":0})); + let (run, _) = usage.observe(&json!({"inputTokens":150,"outputTokens":15}), false); + assert_eq!(run, json!({"inputTokens":50,"outputTokens":5})); + } + + #[test] + fn late_totals_sum_exactly_with_live_usage_and_replays_between_them() { + let mut usage = CodexRunUsage::new(true); + usage.observe(&json!({"inputTokens":100,"outputTokens":10}), true); + usage.begin_run(); + usage.observe(&json!({"inputTokens":150,"outputTokens":15}), false); + let mut moved = 0; + for reported in [180, 180, 190, 180] { + let late = usage.absorb_late_usage(&json!({"inputTokens":reported,"outputTokens":15})); + moved += late["inputTokens"].as_u64().unwrap(); + } + assert_eq!(moved, 40); + let (run, _) = usage.observe(&json!({"inputTokens":200,"outputTokens":15}), false); + assert_eq!(run["inputTokens"], 60); + let late = usage.absorb_late_usage(&json!({"inputTokens":195,"outputTokens":15})); + assert_eq!(late["inputTokens"], 0); + // The prior run's 100, the attached run's 60 and the 40 moved late. + assert_eq!(100 + 60 + moved, 200); + } + #[test] fn attachment_before_paid_work_preserves_a_proven_empty_baseline() { let mut usage = CodexRunUsage::new(false); diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs b/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs index 76f5c5594fa..82a22d33c90 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs @@ -648,48 +648,46 @@ fn normalize_provider_notification( .is_some_and(|usage| usage.increases_run_baseline(total)); if historical && (increases_known_usage || increases_prior_baseline) - && params.get("turnId").and_then(Value::as_str) - != state.active_provider_turn_id.as_deref() + && received_turn != state.active_provider_turn_id.as_deref() { - state.lifecycle = "reconciliation_required".to_owned(); let owns_completed_turn = state.active_provider_turn_id.is_none() - && state - .completed_provider_turn_id - .as_deref() - .is_some_and(|id| params.get("turnId").and_then(Value::as_str) == Some(id)); + && state.completed_provider_turn_id.is_some() + && received_turn == state.completed_provider_turn_id.as_deref(); if !owns_completed_turn { - if state.active_provider_turn_id.is_some() { - state.settle_active_provider_turn_identity()?; + // The usage happened and cannot be undone or billed to this run. + // Report the increment with its owner and keep working: Core + // posts it to the owning run. Moving both counters together + // keeps this run's own delta unchanged. + let late = state + .codex_usage + .get_or_insert_with(|| CodexRunUsage::new(true)) + .absorb_late_usage(total); + if late + .as_object() + .is_some_and(|late| late.values().all(|value| value.as_u64() == Some(0))) + { + // Usage at or below the known total was already counted. + return Ok(normalize_codex_notification( + "paperclip/resumeUsageSnapshot", + params, + )); } - state.active_provider_turn_id = None; - let mut invalidation = normalize_codex_notification("thread/tokenUsage/updated", &json!({ - "threadId":params.get("threadId"), "model":state.config.model, "tokenUsage":{"total":total} - })).remove(0); - invalidation.payload["runDeltaBasis"] = json!("run"); - invalidation.payload["reconciliation"] = json!({"reason":"accounting_reconciliation_required", - "receivedTurnId":params.get("turnId"),"cumulative":invalidation.payload["cumulative"], - "priorOrigin":state.previous_accounting_origin.as_ref().filter(|origin| received_turn == Some(origin.provider_turn_id.as_str())).map(|origin| &origin.identity)}); - return Ok(vec![ - NormalizedProviderEvent { - event_type: "harness.diagnostic".to_owned(), - priority: EventPriority::P0, - payload: json!({"code":"codex_usage_accounting_integrity", "classification":"accounting_reconciliation_required", - "receivedTurnId":params.get("turnId"),"cumulative":total,"priorOrigin":state.previous_accounting_origin.as_ref().filter(|origin| params.get("turnId").and_then(Value::as_str) == Some(origin.provider_turn_id.as_str())).map(|origin| &origin.identity), - "message":"Late provider usage cannot be attributed to the attached run; explicit reconciliation is required"}), - }, - invalidation, - NormalizedProviderEvent { - event_type: "turn.failed".to_owned(), - priority: EventPriority::P0, - payload: json!({"provider":"codex","status":"failed","code":"codex_usage_accounting_integrity","recoverable":false, - "message":"Late provider usage requires explicit accounting reconciliation"}), - }, - NormalizedProviderEvent { - event_type: "run.terminal".to_owned(), - priority: EventPriority::P0, - payload: json!({"schema":"paperclip.prp.terminal.v1","provider":"codex","turnTerminalState":"failed","runTerminalState":"failed","reportedWorkDisposition":"needs_review"}), - }, - ]); + let mut event = normalize_codex_notification( + "thread/tokenUsage/updated", + &json!({ + "threadId":params.get("threadId"), "model":state.config.model, + "tokenUsage":{"total":total,"last":late} + }), + ) + .remove(0); + event.payload["runDeltaAvailable"] = json!(false); + event.payload["runDeltaBasis"] = json!("run"); + event.payload["reconciliation"] = json!({"reason":"accounting_reconciliation_required", + "receivedTurnId":params.get("turnId"),"cumulative":event.payload["cumulative"], + "priorOrigin":state.previous_accounting_origin.as_ref() + .filter(|origin| received_turn == Some(origin.provider_turn_id.as_str())) + .map(|origin| &origin.identity)}); + return Ok(vec![event]); } } let had_baseline = state diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs index 5c59b3338f8..840883c58f1 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs @@ -7301,7 +7301,7 @@ fn durable_codex_usage_keeps_run_baselines_through_requests_recovery_and_attachm } #[test] -fn late_codex_usage_requires_durable_reconciliation_before_or_after_handoff() { +fn late_codex_usage_reports_its_owner_without_blocking_the_session() { for mode in ["before", "after", "after-paid", "unsafe"] { let handoff_first = mode.starts_with("after"); let directory = temporary_directory("late-accounting-reconciliation"); @@ -7425,87 +7425,82 @@ fn late_codex_usage_requires_durable_reconciliation_before_or_after_handoff() { } else { executor.poll_events().unwrap() }; + // In "after-paid" the successor already counted past this total. if events.iter().any(|event| { event.event_type == "usage.reported" - || event.payload["code"] == "codex_usage_accounting_integrity" + || (mode == "after-paid" + && event.payload["code"] == "codex_resume_usage_snapshot") }) { break events; } executor.acknowledge_events(events.len()).unwrap(); }; - if handoff_first { - assert!(retained - .iter() - .filter(|event| event.event_type == "usage.reported") - .all(|event| event.payload["runDeltaAvailable"] == false)); - assert!(retained - .iter() - .any(|event| event.event_type == "usage.reported" - && event.payload["runDeltaAvailable"] == false)); - assert!(retained - .iter() - .any(|event| event.event_type == "run.terminal" - && event.payload["runTerminalState"] == "failed")); - let required = retained - .iter() - .find(|event| event.event_type == "usage.reported") - .unwrap(); + assert!(!retained + .iter() + .any(|event| event.event_type == "run.terminal" + || event.payload["code"] == "codex_usage_accounting_integrity")); + let reported = retained + .iter() + .find(|event| event.event_type == "usage.reported"); + if mode == "after-paid" { + assert!(reported.is_none()); + } else if handoff_first { + let reported = reported.unwrap(); + assert_eq!(reported.payload["runDeltaAvailable"], false); + assert_eq!(reported.payload["runDelta"]["inputTokens"], 20); assert_eq!( - required.payload["reconciliation"]["cumulative"]["inputTokens"], + reported.payload["reconciliation"]["cumulative"]["inputTokens"], 120 ); assert_eq!( - required.payload["reconciliation"]["priorOrigin"]["runId"], + reported.payload["reconciliation"]["priorOrigin"]["runId"], "run-1" ); - let evidence = retained - .iter() - .find(|event| event.payload["code"] == "codex_usage_accounting_integrity") - .unwrap(); - assert_eq!(evidence.payload["priorOrigin"]["runId"], "run-1"); - assert_eq!(evidence.payload["receivedTurnId"], "provider-turn-1"); - assert_eq!(evidence.payload["cumulative"]["inputTokens"], 120); } else { - assert_eq!( - retained - .iter() - .find(|event| event.event_type == "usage.reported") - .unwrap() - .payload["runDelta"]["inputTokens"], - 120 - ); + assert_eq!(reported.unwrap().payload["runDelta"]["inputTokens"], 120); } let saved: Value = serde_json::from_slice(&fs::read(directory.join("codex-provider-state.json")).unwrap()) .unwrap(); - assert_eq!(saved["lifecycle"], "reconciliation_required"); - assert!(!saved["pendingEvents"].as_array().unwrap().is_empty()); - let error = executor - .execute(&command( - "blocked-attach", - 5, - "run.attach", - json!({"authorizedTools":task_context_tool_set()}), - )) - .unwrap_err(); - assert!(error - .to_string() - .contains("requires explicit reconciliation")); + if mode == "unsafe" { + // An unexpected tail still fails closed during the drain itself. + assert_eq!(saved["lifecycle"], "reconciliation_required"); + let error = executor + .execute(&command( + "blocked-attach", + 5, + "run.attach", + json!({"authorizedTools":task_context_tool_set()}), + )) + .unwrap_err(); + assert!(error + .to_string() + .contains("requires explicit reconciliation")); + } else { + if mode == "after-paid" { + assert_eq!(saved["lifecycle"], "turn_active"); + assert!(saved["activeProviderTurnId"].is_string()); + } else { + assert_eq!(saved["lifecycle"], "session_open"); + executor.acknowledge_events(retained.len()).unwrap(); + for _ in 0..50 { + let events = poll_and_ack(&mut executor).unwrap(); + if events.is_empty() { + break; + } + } + executor + .execute(&command( + "next-attach", + 5, + "run.attach", + json!({"authorizedTools":task_context_tool_set()}), + )) + .expect("late usage must not block the next run"); + } + } executor.shutdown().unwrap(); drop(executor); - let mut restored = CodexCommandExecutor::with_runner_config(&directory, &runner_config); - let error = restored - .execute(&command( - "blocked-after-restart", - 6, - "run.attach", - json!({"authorizedTools":task_context_tool_set()}), - )) - .unwrap_err(); - assert!(error - .to_string() - .contains("requires explicit reconciliation")); - restored.shutdown().unwrap(); fs::remove_dir_all(directory).unwrap(); } } diff --git a/packages/paperclip-runner/src/contracts/harness-driver.ts b/packages/paperclip-runner/src/contracts/harness-driver.ts index adccad8d49b..5914b4a6550 100644 --- a/packages/paperclip-runner/src/contracts/harness-driver.ts +++ b/packages/paperclip-runner/src/contracts/harness-driver.ts @@ -478,7 +478,7 @@ export interface PersistedHarnessSession { activeTurnId?: string | null; semanticResult?: PersistedHarnessSemanticResult | null; terminalTurns?: PersistedHarnessTurnTerminal[]; - codexUsageBaseline?: { baseline: Record; latest: Record }; + codexUsageBaseline?: { baseline: Record; latest: Record; accounted?: Record }; /** A result-less terminal task may spend this fail-closed one-shot recovery allowance. */ dispositionOnlyRecoveryConsumed?: boolean; /** Exact accepted provider turn that spent the disposition-only allowance. */ diff --git a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.events.test.ts b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.events.test.ts index c9e857ea401..f1dcb3073c5 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.events.test.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.events.test.ts @@ -404,6 +404,51 @@ describe("Codex app-server Codex driver", () => { expect(await session.usage()).toMatchObject({ runDeltaComplete: false }); }); + it("keeps usage reported between runs in the next run's delta so totals stay exact", async () => { + const transport = new FakeCodexTransport(); + const session = await makeDriver([transport]).openSession({ + runId: "run-first", normalizedSessionId: "normalized-between-runs", workingDirectory: WORKSPACE, + }); + const first = await session.startTurn({ message: { role: "user", text: "First." } }); + transport.push("turn/started", { threadId: "thread-1", turn: { id: first.turnId, status: "inProgress" } }); + transport.push("thread/tokenUsage/updated", { threadId: "thread-1", turnId: first.turnId, tokenUsage: { total: { inputTokens: 100, outputTokens: 10 } } }); + transport.push("turn/completed", { threadId: "thread-1", turn: { id: first.turnId, status: "completed", items: [] } }); + await collectUntilTerminal(session.events()); + // The first run's charge is final at 100; this report arrives after it. + transport.push("thread/tokenUsage/updated", { threadId: "thread-1", turnId: first.turnId, tokenUsage: { total: { inputTokens: 130, outputTokens: 10 } } }); + await new Promise(resolve => setImmediate(resolve)); + await session.attachRun!({ runId: "run-second" }); + transport.turnStartResponse = Promise.resolve({ turn: { id: "turn-2", status: "inProgress", items: [] } }); + const second = await session.startTurn({ message: { role: "user", text: "Second." } }); + transport.push("turn/started", { threadId: "thread-1", turn: { id: second.turnId, status: "inProgress" } }); + transport.push("thread/tokenUsage/updated", { threadId: "thread-1", turnId: second.turnId, tokenUsage: { total: { inputTokens: 150, outputTokens: 16 } } }); + for await (const event of session.events()) if (event.payload.kind === "usage") break; + expect(await session.usage!()).toMatchObject({ runDelta: { inputTokens: 50, outputTokens: 6 } }); + }); + + it("carries uncharged usage across a run that reports no usage", async () => { + const transport = new FakeCodexTransport(); + const session = await makeDriver([transport]).openSession({ + runId: "run-first", normalizedSessionId: "normalized-three-runs", workingDirectory: WORKSPACE, + }); + const first = await session.startTurn({ message: { role: "user", text: "First." } }); + transport.push("turn/started", { threadId: "thread-1", turn: { id: first.turnId, status: "inProgress" } }); + transport.push("thread/tokenUsage/updated", { threadId: "thread-1", turnId: first.turnId, tokenUsage: { total: { inputTokens: 100, outputTokens: 10 } } }); + transport.push("turn/completed", { threadId: "thread-1", turn: { id: first.turnId, status: "completed", items: [] } }); + await collectUntilTerminal(session.events()); + transport.push("thread/tokenUsage/updated", { threadId: "thread-1", turnId: first.turnId, tokenUsage: { total: { inputTokens: 130, outputTokens: 10 } } }); + await new Promise(resolve => setImmediate(resolve)); + // The second run attaches and ends without any usage report of its own. + await session.attachRun!({ runId: "run-second" }); + await session.attachRun!({ runId: "run-third" }); + transport.turnStartResponse = Promise.resolve({ turn: { id: "turn-3", status: "inProgress", items: [] } }); + const third = await session.startTurn({ message: { role: "user", text: "Third." } }); + transport.push("turn/started", { threadId: "thread-1", turn: { id: third.turnId, status: "inProgress" } }); + transport.push("thread/tokenUsage/updated", { threadId: "thread-1", turnId: third.turnId, tokenUsage: { total: { inputTokens: 150, outputTokens: 16 } } }); + for await (const event of session.events()) if (event.payload.kind === "usage") break; + expect(await session.usage!()).toMatchObject({ runDelta: { inputTokens: 50, outputTokens: 6 } }); + }); + it("accepts a thread usage snapshot replayed before a resumed turn starts", async () => { const transport = new FakeCodexTransport(); const session = await makeDriver([transport]).openSession({ diff --git a/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts b/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts index d95eb72f2e5..e2a3a9e8557 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts @@ -134,7 +134,10 @@ export class CodexHarnessSession this.eventQueue.clear(); } if (this.codexUsageBaseline && input.runId !== this.runId) { - this.codexUsageBaseline = { baseline: { ...this.codexUsageBaseline.latest }, latest: { ...this.codexUsageBaseline.latest } }; + // Keep the accounted total until a run reports usage, so a run with no + // report cannot absorb usage that was carried forward for the next one. + const { accounted, latest } = this.codexUsageBaseline; + this.codexUsageBaseline = { baseline: { ...(accounted ?? latest) }, latest: { ...latest }, ...(accounted ? { accounted: { ...accounted } } : {}) }; this.usageSnapshot = codexRunUsage(this.codexUsageBaseline); } this.runId = input.runId; @@ -729,6 +732,9 @@ export class CodexHarnessSession reconciledUsage.total, false, ); + // Only an active turn's report belongs to this run. An idle thread read + // can include late usage that no run has been charged for yet. + if (this.activeTurnId !== null) this.codexUsageBaseline.accounted = { ...this.codexUsageBaseline.latest }; this.usageSnapshot = { ...reconciledUsage, ...codexRunUsage(this.codexUsageBaseline), diff --git a/packages/paperclip-runner/src/drivers/codex/codex-resume-usage.test.ts b/packages/paperclip-runner/src/drivers/codex/codex-resume-usage.test.ts index ef3f0a4cc64..f49d899b6ef 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-resume-usage.test.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-resume-usage.test.ts @@ -125,6 +125,7 @@ describe("Codex resume accounting through the production driver", () => { expect(advanced.codexUsageBaseline).toEqual({ baseline: { inputTokens: 100, outputTokens: 10 }, latest: { inputTokens: 150, outputTokens: 19 }, + accounted: { inputTokens: 150, outputTokens: 19 }, }); expect(persisted.codexUsageBaseline.latest).toEqual({ inputTokens: 140, @@ -277,6 +278,7 @@ describe("Codex resume accounting through the production driver", () => { expect((await session.snapshot!()).codexUsageBaseline).toEqual({ baseline: { inputTokens: 100, outputTokens: 10 }, latest: { inputTokens: 100, outputTokens: 10 }, + accounted: { inputTokens: 100, outputTokens: 10 }, }); second.turnStartResponse = Promise.resolve({ turn: { id: "turn-2", status: "inProgress", items: [] }, @@ -329,6 +331,35 @@ describe("Codex resume accounting through the production driver", () => { await session.close(); }); + it("keeps usage reported between runs in the next run's delta after a restore", async () => { + const first = new FakeCodexTransport(); + const second = new FakeCodexTransport(); + const driver = makeDriver([first, second], { conversationMode: "direct" }); + const session = await driver.openSession({ runId: "first", normalizedSessionId: "between-runs-restore", workingDirectory: WORKSPACE }); + await session.startTurn({ message: { role: "user", text: "First" } }); + first.push("thread/tokenUsage/updated", { threadId: "thread-1", turnId: "turn-1", tokenUsage: { total: { inputTokens: 100, outputTokens: 10 } } }); + first.push("turn/completed", { threadId: "thread-1", turn: { id: "turn-1", status: "completed", items: [] } }); + await collectUntilTerminal(session.events()); + // The first run's charge is final at 100; this report arrives after it. + first.push("thread/tokenUsage/updated", { threadId: "thread-1", turnId: "turn-1", tokenUsage: { total: { inputTokens: 130, outputTokens: 10 } } }); + await new Promise(resolve => setImmediate(resolve)); + const persisted = JSON.parse(JSON.stringify(await session.snapshot!())); + expect(persisted.codexUsageBaseline).toMatchObject({ latest: { inputTokens: 130 }, accounted: { inputTokens: 100 } }); + await session.close(); + second.readResponse = { thread: { id: "thread-1", sessionId: "provider-session-1", cwd: WORKSPACE, + turns: [{ id: "turn-1", status: "completed", items: [] }], tokenUsage: { total: { inputTokens: 130, outputTokens: 10 } } } }; + const recovery = await driver.recoverSession(persisted); + expect(recovery.recovered).toBe(true); + const recovered = recovery.session!; + await recovered.attachRun!({ runId: "second" }); + second.turnStartResponse = Promise.resolve({ turn: { id: "turn-2", status: "inProgress", items: [] } }); + await recovered.startTurn({ message: { role: "user", text: "Second" } }); + second.push("thread/tokenUsage/updated", { threadId: "thread-1", turnId: "turn-2", tokenUsage: { total: { inputTokens: 150, outputTokens: 16 } } }); + for await (const event of recovered.events()) if (event.payload.kind === "usage") break; + expect(await recovered.usage!()).toMatchObject({ runDelta: { inputTokens: 50, outputTokens: 6 } }); + await recovered.close(); + }); + it("keeps the startup external sandbox profile on subsequent turns", async () => { const transport = new FakeCodexTransport(); const session = await makeDriver([transport], { diff --git a/packages/paperclip-runner/src/drivers/codex/codex-session-notifications.ts b/packages/paperclip-runner/src/drivers/codex/codex-session-notifications.ts index 31ccd65a1fb..e6c1a415866 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-session-notifications.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-session-notifications.ts @@ -589,6 +589,7 @@ async function mapNotificationBody(state: CodexSessionState, notification: Codex // is being attached, before the next turn has started. Keep the snapshot, // but do not turn that benign replay into a fatal turn-binding violation. if (state.activeTurnId === null || turnId !== state.activeTurnId) return; + if (state.codexUsageBaseline) state.codexUsageBaseline.accounted = { ...state.codexUsageBaseline.latest }; state.emit( "item.completed", { kind: "usage", usage: state.usageSnapshot }, diff --git a/packages/paperclip-runner/src/drivers/codex/codex-usage-baseline.ts b/packages/paperclip-runner/src/drivers/codex/codex-usage-baseline.ts index fc777bc1a57..b6d6266e58c 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-usage-baseline.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-usage-baseline.ts @@ -2,6 +2,9 @@ export interface CodexUsageBaseline { baseline: Record; latest: Record; + /** Totals the current run has accounted for. Usage above this at attachment + * was never charged, so it stays in the next run's delta and is not lost. */ + accounted?: Record; } export function codexUsageMeasurement(value: unknown): Record { @@ -27,7 +30,7 @@ export function observeCodexUsage( for (const [key, value] of Object.entries(measurement)) { latest[key] = Math.max(latest[key] ?? 0, value); } - return { baseline, latest }; + return { baseline, latest, ...(state?.accounted ? { accounted: state.accounted } : {}) }; } export function codexRunUsage( diff --git a/server/src/services/native-runtime/native-usage-delivery.integration.test.ts b/server/src/services/native-runtime/native-usage-delivery.integration.test.ts index 9a998272e3e..f49ff9b41ec 100644 --- a/server/src/services/native-runtime/native-usage-delivery.integration.test.ts +++ b/server/src/services/native-runtime/native-usage-delivery.integration.test.ts @@ -8,10 +8,11 @@ import { resolve } from "node:path"; import { fileURLToPath } from "node:url"; import { and, eq, sql } from "drizzle-orm"; import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; -import { agents, budgetPolicies, companies, costEvents, createDb, heartbeatRuns, runUsageReceipts, type Db } from "@paperclipai/db"; +import { activityLog, agents, budgetPolicies, companies, costEvents, createDb, heartbeatRuns, runUsageReceipts, type Db } from "@paperclipai/db"; import { type PrpEvent } from "../../vendor/paperclip-runner/index.js"; import { startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; import { flushBudgetEvaluationWork } from "../budgets.js"; +import { accountingIntegrityService } from "../accounting-integrity.js"; import { accountRunCost } from "../run-cost-accounting.js"; import { usageReceiptSpoolPath, type UsageReceiptEnvelope } from "../usage-receipts.js"; import { createNativeUsageContinuationDelivery, createNativeUsageDelivery } from "./native-usage-delivery.js"; @@ -90,7 +91,7 @@ describe("producer-owned native usage", () => { }); it.each(["tokens", "cost", "incomplete", "unpriceable", "basis"])( - "retains a differing finalized source tail as unaccepted durable reconciliation debt: %s", async change => { + "retains a differing finalized source tail as durable debt without failing delivery: %s", async change => { const { input, source } = await finalizedSource(); const continuation = await createNativeUsageContinuationDelivery(db, source); const late = event(input, 3, "usage.reported"); @@ -100,15 +101,14 @@ describe("producer-owned native usage", () => { inputTokens: change === "tokens" ? 150 : change === "unpriceable" ? "unknown" : 120, outputTokens: 10, providerCostUsd: change === "cost" ? 0.5 : 0.25, }; - await expect(continuation.capture(late)).rejects.toThrow("native_usage_after_finalization_requires_reconciliation"); - await expect(continuation.capture(late)).rejects.toThrow("native_usage_after_finalization_requires_reconciliation"); + await continuation.capture(late); + await continuation.capture(late); expect(await db.select().from(runUsageReceipts).where(eq(runUsageReceipts.runId, input.runId))).toHaveLength(3); const [pending] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, input.runId)); expect(pending.usageJson).toMatchObject({ accountingLateReceiptPending: true }); - // Loading the durable journal on restart cannot turn rejected evidence - // into an acknowledged producer delivery. + // Loading the durable journal on restart keeps the evidence owed. const restarted = await createNativeUsageContinuationDelivery(db, source); - await expect(restarted.capture(late)).rejects.toThrow("native_usage_after_finalization_requires_reconciliation"); + await restarted.capture(late); expect((await db.select().from(costEvents).where(eq(costEvents.heartbeatRunId, input.runId)))[0].costCents).toBe(25); }); @@ -216,11 +216,11 @@ describe("producer-owned native usage", () => { expect(ledger[0].costCents).toBe(25); const failedBeforeLateReport = failed; await request(receiver, { kind: "command", type: "turn.steer", commandId: "release-late-usage" }); - await vi.waitFor(() => expect(failed).toBeGreaterThan(failedBeforeLateReport), { timeout: 10_000 }); - expect(lastCaptureError).toContain("native_usage_after_finalization_requires_reconciliation"); - const retainedLate = producerState().outbox.find(report => report.eventType === "usage.reported"); - expect(retainedLate).toBeDefined(); - expect(retainedLate!.envelope.payload).toMatchObject({ runId: input.runId, sourceInstanceId: input.runnerInstanceId }); + await vi.waitFor(async () => expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, input.runId)))[0].usageJson) + .toMatchObject({ accountingLateReceiptPending: true }), { timeout: 10_000 }); + // Late evidence is acknowledged: it must not wedge the producer's outbox. + await vi.waitFor(() => expect(producerState().outbox).toEqual([]), { timeout: 10_000 }); + expect(failed).toBe(failedBeforeLateReport); expect((await receipts()).filter(row => (row.receiptJson.producerEvent as Record)?.eventType === "usage.reported")).toHaveLength(2); expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, input.runId)))[0].usageJson) .toMatchObject({ accountingLateReceiptPending: true }); @@ -236,6 +236,171 @@ describe("producer-owned native usage", () => { } }, 240_000); + describe("late usage reported on a later run", () => { + const sessionId = randomUUID(); + const lateReport = (input: Awaited>, priorOrigin: Record | null, sequence = 7) => { + const report = event(input, sequence, "usage.reported"); + report.normalizedSessionId = sessionId; + report.payload = { runDeltaBasis: "run", runDeltaAvailable: false, runDelta: { inputTokens: 30, outputTokens: 3, providerCostUsd: 0.05 }, + reconciliation: { reason: "accounting_reconciliation_required", receivedTurnId: "provider-turn-1", cumulative: { inputTokens: 150, outputTokens: 13 }, priorOrigin } }; + return report; + }; + const successor = async (owner: Awaited>) => { + const runId = randomUUID(); + await db.update(heartbeatRuns).set({ nativeSessionId: sessionId }).where(eq(heartbeatRuns.id, owner.input.runId)); + await db.insert(heartbeatRuns).values({ id: runId, companyId: owner.input.companyId, agentId: owner.input.agentId, invocationSource: "on_demand", + status: "running", runtimeMode: "native", nativeSessionId: sessionId, costAccountingPending: true }); + return { ...owner.input, runId }; + }; + const ledger = (runId: string) => db.select().from(costEvents).where(eq(costEvents.heartbeatRunId, runId)); + + it.each([["exact", true], ["inferred", false]] as const)("posts the increment to the %s owner and leaves the delivering run alone", async (attribution, named) => { + const owner = await finalizedSource(); + const current = await successor(owner); + const delivery = await createNativeUsageDelivery(db, current); + const report = lateReport(current, named ? { runId: owner.input.runId, normalizedSessionId: sessionId } : null); + await delivery.capture(report); + await delivery.capture(report); + const [before] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, current.runId)); + expect(await db.select().from(runUsageReceipts).where(eq(runUsageReceipts.runId, current.runId))).toHaveLength(0); + expect(before.usageJson).not.toHaveProperty("inputTokens"); + expect(await accountRunCost(db, owner.input.runId)).toBe(true); + expect(await accountRunCost(db, owner.input.runId)).toBe(false); + const charges = await ledger(owner.input.runId); + expect(charges).toHaveLength(2); + expect(charges.map(row => row.inputTokens).sort((a, b) => a - b)).toEqual([30, 120]); + expect(charges.reduce((sum, row) => sum + row.costCents, 0)).toBeCloseTo(30); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, owner.input.runId)); + expect(run.usageJson).toMatchObject({ inputTokens: 150, outputTokens: 13, accountingLateReceiptPending: false }); + expect(run.accountingLastError).toBeNull(); + expect(run.costAccountingPending).toBe(false); + const [activity] = await db.select().from(activityLog).where(and(eq(activityLog.entityId, owner.input.runId), eq(activityLog.action, "cost.late_usage_posted"))); + expect(activity.details).toMatchObject({ attribution, reportedByRunId: current.runId, inputTokens: 30 }); + const inspection = await accountingIntegrityService(db).inspect(owner.input.companyId); + expect(inspection.findings.filter(finding => finding.entityId === owner.input.runId)).toEqual([]); + }); + + it("waits for the owner's charge before posting its late usage", async () => { + const ownerInput = await seed(); + await db.update(heartbeatRuns).set({ runnerInstanceId: ownerInput.runnerInstanceId, nativeSessionId: sessionId }).where(eq(heartbeatRuns.id, ownerInput.runId)); + const ownerDelivery = await createNativeUsageDelivery(db, ownerInput); + await ownerDelivery.capture(event(ownerInput, 1, "usage.reported")); + await ownerDelivery.capture(event(ownerInput, 2, "turn.completed")); + await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, ownerInput.runId)); + const runId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: runId, companyId: ownerInput.companyId, agentId: ownerInput.agentId, invocationSource: "on_demand", + status: "running", runtimeMode: "native", nativeSessionId: sessionId, costAccountingPending: true }); + const current = { ...ownerInput, runId }; + await (await createNativeUsageDelivery(db, current)).capture(lateReport(current, { runId: ownerInput.runId, normalizedSessionId: sessionId })); + expect(await accountRunCost(db, ownerInput.runId)).toBe(true); + expect(await ledger(ownerInput.runId)).toHaveLength(1); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, ownerInput.runId)))[0].costAccountingPending).toBe(true); + expect(await accountRunCost(db, ownerInput.runId)).toBe(true); + expect(await ledger(ownerInput.runId)).toHaveLength(2); + }); + + it("keeps an older Runner's invalidation without an increment on the delivering run", async () => { + const owner = await finalizedSource(); + const current = await successor(owner); + const report = lateReport(current, { runId: owner.input.runId, normalizedSessionId: sessionId }); + (report.payload as Record).runDelta = { inputTokens: 0, cacheReadTokens: 0, cacheWriteTokens: 0, outputTokens: 0, providerCostUsd: 0, requests: 0, activeSeconds: 0 }; + await (await createNativeUsageDelivery(db, current)).capture(report); + expect(await db.select().from(runUsageReceipts).where(eq(runUsageReceipts.runId, current.runId))).toHaveLength(1); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, owner.input.runId)))[0].usageJson).not.toHaveProperty("accountingLateReceiptPending", true); + }); + + it("charges the delivering run when its session has no finished run, never another session's", async () => { + const other = await finalizedSource(); + await db.update(heartbeatRuns).set({ nativeSessionId: randomUUID() }).where(eq(heartbeatRuns.id, other.input.runId)); + const runId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: runId, companyId: other.input.companyId, agentId: other.input.agentId, invocationSource: "on_demand", + status: "running", runtimeMode: "native", nativeSessionId: sessionId, costAccountingPending: true }); + const current = { ...other.input, runId }; + const delivery = await createNativeUsageDelivery(db, current); + await delivery.capture(lateReport(current, { runId: other.input.runId, normalizedSessionId: sessionId })); + await delivery.capture(event(current, 8, "usage.reported")); + await delivery.capture(event(current, 9, "turn.completed")); + await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, runId)); + expect(await accountRunCost(db, runId)).toBe(true); + expect(await accountRunCost(db, runId)).toBe(true); + expect((await ledger(runId)).map(row => row.inputTokens).sort((a, b) => a - b)).toEqual([30, 120]); + expect(await ledger(other.input.runId)).toHaveLength(1); + const [activity] = await db.select().from(activityLog).where(and(eq(activityLog.entityId, runId), eq(activityLog.action, "cost.late_usage_posted"))); + expect(activity.details).toMatchObject({ attribution: "inferred", reportedByRunId: runId }); + expect((await accountingIntegrityService(db).inspect(current.companyId)).findings.filter(finding => finding.entityId === runId)).toEqual([]); + }); + }); + + it("posts a same-run token increase as a separate auditable charge", async () => { + const { input, source } = await finalizedSource(); + const continuation = await createNativeUsageContinuationDelivery(db, source); + const late = event(input, 3, "usage.reported"); + (late.payload as Record).runDelta = { inputTokens: 150, outputTokens: 15, providerCostUsd: 0.5 }; + await continuation.capture(late); + expect(await accountRunCost(db, input.runId)).toBe(true); + const charges = await db.select().from(costEvents).where(eq(costEvents.heartbeatRunId, input.runId)); + expect(charges.map(row => row.inputTokens).sort((a, b) => a - b)).toEqual([30, 120]); + expect(charges.reduce((sum, row) => sum + row.costCents, 0)).toBeCloseTo(50); + expect((await accountingIntegrityService(db).inspect(input.companyId)).findings.filter(finding => finding.entityId === input.runId)).toEqual([]); + }); + + it("prices a same-run late report from the rate card instead of posting free tokens", async () => { + const input = { ...await seed(), provider: { kind: "codex" as const, model: "gpt-6.1-sol" } }; + const delivery = await createNativeUsageDelivery(db, input); + const report = (sequence: number, outputTokens: number) => { + const reported = event(input, sequence, "usage.reported"); + reported.payload = { runDeltaBasis: "run", runDeltaAvailable: true, runDelta: { inputTokens: 1000, outputTokens } }; + return reported; + }; + await delivery.capture(report(1, 100000)); + await delivery.capture(event(input, 2, "turn.completed")); + await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, input.runId)); + expect(await accountRunCost(db, input.runId)).toBe(true); + await delivery.capture(report(3, 120000)); + expect(await accountRunCost(db, input.runId)).toBe(true); + const charges = await db.select().from(costEvents).where(eq(costEvents.heartbeatRunId, input.runId)); + expect(charges).toHaveLength(2); + expect(charges.every(charge => charge.costStatus !== "unpriced")).toBe(true); + const late = charges.find(charge => charge.outputTokens === 20000)!; + expect(late.costCents).toBeGreaterThan(0); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, input.runId)); + expect(run.usageJson).toMatchObject({ outputTokens: 120000, accountingLateReceiptPending: false }); + expect((await accountingIntegrityService(db).inspect(input.companyId)).findings.filter(finding => finding.entityId === input.runId)).toEqual([]); + }); + + it("labels a rate-card late charge as an estimate when the original charge was reported", async () => { + const input = { ...await seed(), provider: { kind: "codex" as const, model: "gpt-6.1-sol" } }; + const delivery = await createNativeUsageDelivery(db, input); + const report = (sequence: number, runDelta: Record) => { + const reported = event(input, sequence, "usage.reported"); + reported.payload = { runDeltaBasis: "run", runDeltaAvailable: true, runDelta }; + return reported; + }; + await delivery.capture(report(1, { inputTokens: 1000, outputTokens: 100000, providerCostUsd: 0.5 })); + await delivery.capture(event(input, 2, "turn.completed")); + await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, input.runId)); + expect(await accountRunCost(db, input.runId)).toBe(true); + await delivery.capture(report(3, { inputTokens: 1000, outputTokens: 120000 })); + expect(await accountRunCost(db, input.runId)).toBe(true); + const charges = await db.select().from(costEvents).where(eq(costEvents.heartbeatRunId, input.runId)); + const original = charges.find(charge => charge.outputTokens === 100000)!; + const late = charges.find(charge => charge.outputTokens === 20000)!; + expect(original.costStatus).not.toBe("estimated"); + expect(late.costStatus).toBe("estimated"); + expect(late.pricingProvenance).toMatchObject({ source: "rate_card" }); + }); + + it("keeps lower late usage as visible debt instead of guessing", async () => { + const { input, source } = await finalizedSource(); + const continuation = await createNativeUsageContinuationDelivery(db, source); + const late = event(input, 3, "usage.reported"); + (late.payload as Record).runDelta = { inputTokens: 90, outputTokens: 15, providerCostUsd: 0.5 }; + await continuation.capture(late); + expect(await accountRunCost(db, input.runId)).toBe(false); + expect(await db.select().from(costEvents).where(eq(costEvents.heartbeatRunId, input.runId))).toHaveLength(1); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, input.runId)))[0].accountingLastError).toBe("late_usage_requires_reconciliation"); + }); + it("preserves source takeover fencing, budget holds, and missing-usage debt", async () => { const input = await seed(); const first = await createNativeUsageDelivery(db, input); @@ -259,7 +424,7 @@ describe("producer-owned native usage", () => { expect((await db.select().from(costEvents).where(eq(costEvents.heartbeatRunId, input.runId)))[0].costCents).toBe(50); }); - it.each(["tokens", "cost", "incomplete"])("accepts equal finalized snapshots and rejects changed %s without repricing", async change => { + it.each(["tokens", "cost", "incomplete"])("accepts equal finalized snapshots and keeps changed %s as debt without repricing", async change => { const input = await seed(); const delivery = await createNativeUsageDelivery(db, input); await delivery.capture(event(input, 1, "usage.reported")); @@ -270,8 +435,8 @@ describe("producer-owned native usage", () => { const late = event(input, 4, "usage.reported"); (late.payload as Record).runDelta = { inputTokens: change === "tokens" ? 150 : 120, outputTokens: change === "tokens" ? 15 : 10, providerCostUsd: change === "incomplete" ? 0.25 : 0.5 }; if (change === "incomplete") (late.payload as Record).runDeltaAvailable = false; - await expect(delivery.capture(late)).rejects.toThrow("native_usage_after_finalization_requires_reconciliation"); - await expect(delivery.capture(late)).rejects.toThrow("native_usage_after_finalization_requires_reconciliation"); + await delivery.capture(late); + await delivery.capture(late); expect((await db.select().from(costEvents).where(eq(costEvents.heartbeatRunId, input.runId)))[0].costCents).toBe(25); }); @@ -296,7 +461,7 @@ describe("producer-owned native usage", () => { expect(ledger[0].costCents).toBe(100.2); const correction = report(4); (correction.payload as Record).runDelta = { inputTokens: 1000, outputTokens: 100000, providerCostUsd: 2 }; - await expect(delivery.capture(correction)).rejects.toThrow("native_usage_after_finalization_requires_reconciliation"); + await delivery.capture(correction); }); it("keeps successor exposure pending after prior-turn accounting invalidates its live snapshot", async () => { diff --git a/server/src/services/native-runtime/native-usage-delivery.ts b/server/src/services/native-runtime/native-usage-delivery.ts index 964e3aadd36..412cfa51c29 100644 --- a/server/src/services/native-runtime/native-usage-delivery.ts +++ b/server/src/services/native-runtime/native-usage-delivery.ts @@ -1,5 +1,5 @@ import { createHash } from "node:crypto"; -import { and, asc, eq, isNull, sql } from "drizzle-orm"; +import { and, asc, desc, eq, inArray, isNull, ne, sql } from "drizzle-orm"; import { heartbeatRuns, runUsageReceipts, type Db } from "@paperclipai/db"; import type { AdapterUsageCheckpoint } from "@paperclipai/adapter-utils"; import type { NativeExecutionInput, PrpEvent } from "../../vendor/paperclip-runner/index.js"; @@ -75,6 +75,39 @@ function loadNativeUsageReceipts(db: Db, input: Pick'accountingLateReceiptDebts', '{}'::jsonb) ? ${runUsageReceipts.id}::text))`)).orderBy(asc(runUsageReceipts.sequence)); } +const isUuid = (value: string) => /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i.test(value); +const finishedRunStatuses = ["succeeded", "failed", "timed_out", "cancelled", "interrupted"]; + +/** The Runner names a late report's owner when the turn proves it. A session's + * usage belongs to that session, so an unproven owner is the session's last + * finished run, recorded as inferred. */ +async function captureLateUsageForOwner(db: Db, input: NativeUsageDeliveryInput, event: PrpEvent, payload: Record) { + const reconciliation = payload.reconciliation; + if (!reconciliation || typeof reconciliation !== "object" || !isUuid(event.normalizedSessionId)) return false; + const measured = { runDelta: payload.runDelta }; + const usage = normalizeNativeUsage(measured, { inputIncludesCacheReads: input.provider.kind === "codex" }); + // An older Runner reports only an invalidation, with zero-filled counters. + // It has no increment to move and still belongs to the delivering run. + if (!usage || (usage.inputTokens + usage.outputTokens + (usage.cachedInputTokens ?? 0)) <= 0) return false; + const prior = (reconciliation as Record).priorOrigin as Record | null | undefined; + const runs = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.nativeSessionId, event.normalizedSessionId), + inArray(heartbeatRuns.status, finishedRunStatuses), ne(heartbeatRuns.id, input.runId), + )).orderBy(desc(heartbeatRuns.finishedAt)); + const exact = runs.find(run => run.id === prior?.runId && prior?.normalizedSessionId === event.normalizedSessionId); + // With no finished run in this session, the delivering run is the only owner + // left. Its own delta already excludes this usage, so it is never counted twice. + const owner = exact ?? runs[0] ?? { id: input.runId }; + const receipt = priceAnthropicReceipt(priceCodexReceipt({ ...input.billing, model: input.provider.model ?? "unknown", usage, usageBasis: "per_run", + complete: true, costUsd: nativeUsageCostUsd(measured, input.provider) ?? null })); + await persistUsageReceipt(db, { schema: "paperclip/accounting-receipt/v1", id: uuid(`${owner.id}:${input.runnerInstanceId}:${event.sourceEventId}`), + companyId: input.companyId, runId: owner.id, adapterType: "paperclip_runner", sourceId: uuid(`late:${owner.id}:${input.runId}:${input.runnerInstanceId}`), + sequence: event.sourceSeq, receivedAt: event.emittedAt, receipt, + producerEvent: { sourceEventId: event.sourceEventId, turnId: event.turnId, eventType: event.eventType, usage: { runDelta: payload.runDelta, reconciliation } }, + lateDelta: { attribution: exact ? "exact" : "inferred", reportedByRunId: input.runId } }); + return true; +} + function createNativeUsageReceiver(db: Db, input: NativeUsageDeliveryInput, sourceId: string, saved: Awaited>, verifySaved = false) { const accepted = new Map(); @@ -120,6 +153,7 @@ function createNativeUsageReceiver(db: Db, input: NativeUsageDeliveryInput, sour if (event.eventType !== "usage.reported" && event.eventType !== "turn.started" && !terminal.has(event.eventType)) return; if (event.runId !== input.runId || event.sourceInstanceId !== input.runnerInstanceId) throw new Error("native_usage_source_mismatch"); const payload = event.payload as Record; + if (event.eventType === "usage.reported" && await captureLateUsageForOwner(db, input, event, payload)) return; const fact: NonNullable = { sourceEventId: event.sourceEventId, turnId: event.turnId, eventType: event.eventType, ...(typeof payload.providerTurnId === "string" ? { providerTurnId: payload.providerTurnId } : {}), diff --git a/server/src/services/run-cost-accounting.test.ts b/server/src/services/run-cost-accounting.test.ts index 08f0ad287d9..29800c24559 100644 --- a/server/src/services/run-cost-accounting.test.ts +++ b/server/src/services/run-cost-accounting.test.ts @@ -180,13 +180,13 @@ const support = await getEmbeddedPostgresTestSupport(); expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, owner.runId)))[0].costAccountingPending).toBe(false); }); - it("commits native late evidence before reporting reconciliation and keeps ambiguous debt quiet", async () => { + it("commits conflicting native late evidence and keeps its debt quiet", async () => { const sourceId = randomUUID(); const owner = await seed({ usageJson: { accountingReceiptSourceId: sourceId, accountingReceiptSequence: 0 } }); await persistUsageReceipt(db, receipt(owner, sourceId, 1, 0.01)); await accountRunCost(db, owner.runId); - const late = { ...receipt(owner, sourceId, 2, 0.02, 20), producerEvent: { sourceEventId: "late", eventType: "usage.reported" } }; - await expect(persistUsageReceipt(db, late)).rejects.toThrow("requires_reconciliation"); + const late = { ...receipt(owner, sourceId, 2, 0.02, 5), producerEvent: { sourceEventId: "late", eventType: "usage.reported" } }; + await persistUsageReceipt(db, late); expect(await db.select().from(runUsageReceipts).where(eq(runUsageReceipts.id, late.id))).toHaveLength(1); expect((await readCostAccountingWork(db, Date.now(), directory)).pending).toBe(true); expect(await accountRunCost(db, owner.runId)).toBe(false); @@ -197,6 +197,94 @@ const support = await getEmbeddedPostgresTestSupport(); expect((await db.select().from(costEvents).where(eq(costEvents.heartbeatRunId, owner.runId)))[0].costCents).toBe(1); }); + it("never counts money twice when late growth and a price correction combine", async () => { + const sourceId = randomUUID(); + const owner = await seed({ usageJson: { accountingReceiptSourceId: sourceId, accountingReceiptSequence: 0 } }); + await persistUsageReceipt(db, receipt(owner, sourceId, 1, 1, 100)); + await accountRunCost(db, owner.runId); + const total = async () => { + const charges = await db.select({ cents: sql`${costEvents.costCents}::text` }).from(costEvents).where(eq(costEvents.heartbeatRunId, owner.runId)); + return charges.reduce((sum, charge) => sum + Number(charge.cents), 0); + }; + await persistUsageReceipt(db, receipt(owner, sourceId, 2, 1.2, 120)); + expect(await accountRunCost(db, owner.runId)).toBe(true); + expect(await total()).toBeCloseTo(120); + // The same tokens at a corrected whole-run price. + await persistUsageReceipt(db, receipt(owner, sourceId, 3, 1.3, 120)); + expect(await accountRunCost(db, owner.runId)).toBe(true); + expect(await total()).toBeCloseTo(130); + // More tokens after that correction charge only the new difference. + await persistUsageReceipt(db, receipt(owner, sourceId, 4, 1.5, 140)); + expect(await accountRunCost(db, owner.runId)).toBe(true); + expect(await total()).toBeCloseTo(150); + // Run totals follow provider-reported amounts, and corrections stay separate. + expect((await accountingIntegrityService(db).inspect(owner.companyId)).findings.filter(finding => finding.entityId === owner.runId)).toEqual([]); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, owner.runId)))[0].costAccountingPending).toBe(false); + }); + + it.each(["cumulative first", "delta first"])("adds a later run's increment to a queued source total: %s", async order => { + const sourceId = randomUUID(); + const owner = await seed({ usageJson: { accountingReceiptSourceId: sourceId, accountingReceiptSequence: 0 } }); + await persistUsageReceipt(db, receipt(owner, sourceId, 1, 1, 100)); + await accountRunCost(db, owner.runId); + const cumulative = receipt(owner, sourceId, 2, 1.2, 120); + const delta: UsageReceiptEnvelope = { ...receipt(owner, randomUUID(), 1, 0.1, 10), lateDelta: { attribution: "exact", reportedByRunId: randomUUID() } }; + for (const item of order === "cumulative first" ? [cumulative, delta] : [delta, cumulative]) await persistUsageReceipt(db, item); + for (let attempt = 0; attempt < 3; attempt++) await accountRunCost(db, owner.runId); + const charges = await db.select().from(costEvents).where(eq(costEvents.heartbeatRunId, owner.runId)); + expect(charges.reduce((sum, charge) => sum + charge.inputTokens, 0)).toBe(130); + expect(charges.reduce((sum, charge) => sum + charge.costCents, 0)).toBeCloseTo(130); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, owner.runId)); + expect(run.usageJson).toMatchObject({ inputTokens: 130, accountingLateReceiptPending: false }); + expect((await accountingIntegrityService(db).inspect(owner.companyId)).findings.filter(finding => finding.entityId === owner.runId)).toEqual([]); + }); + + it("keeps late increment totals through a heartbeat metadata write", async () => { + const owner = await seed({ usageJson: { accountingReceiptOwner: "native_producer", accountingLateDeltas: { inputTokens: 10, units: "100000000" } } }); + await db.update(heartbeatRuns).set({ usageJson: usageReceipts.preserveAcceptedRunUsageProjectionSql({ accountingLateDeltas: null, note: "completion" }) }) + .where(eq(heartbeatRuns.id, owner.runId)); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, owner.runId)); + expect(run.usageJson).toMatchObject({ note: "completion", accountingLateDeltas: { inputTokens: 10, units: "100000000" } }); + }); + + it("clears an older growth receipt that a newer cumulative receipt already covers", async () => { + const sourceId = randomUUID(); + const owner = await seed({ usageJson: { accountingReceiptSourceId: sourceId, accountingReceiptSequence: 0 } }); + await persistUsageReceipt(db, receipt(owner, sourceId, 1, 1, 100)); + await accountRunCost(db, owner.runId); + await persistUsageReceipt(db, receipt(owner, sourceId, 2, 1.2, 120)); + await persistUsageReceipt(db, receipt(owner, sourceId, 3, 1.4, 140)); + await accountRunCost(db, owner.runId); + // Clearing the remaining debt must wake budgets that the debt was holding. + await db.delete(budgetEvaluationWork).where(eq(budgetEvaluationWork.companyId, owner.companyId)); + await accountRunCost(db, owner.runId); + expect((await db.select().from(budgetEvaluationWork).where(eq(budgetEvaluationWork.companyId, owner.companyId))).length).toBeGreaterThan(0); + const charges = await db.select().from(costEvents).where(eq(costEvents.heartbeatRunId, owner.runId)); + expect(charges.reduce((sum, charge) => sum + charge.inputTokens, 0)).toBe(140); + expect(charges.reduce((sum, charge) => sum + charge.costCents, 0)).toBeCloseTo(140); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, owner.runId)); + expect(run.usageJson).toMatchObject({ accountingLateReceiptPending: false }); + expect(run.costAccountingPending).toBe(false); + expect(run.accountingLastError).toBeNull(); + }); + + it("keeps an older receipt with conflicting billing as debt even when its tokens are covered", async () => { + const sourceId = randomUUID(); + const owner = await seed({ usageJson: { accountingReceiptSourceId: sourceId, accountingReceiptSequence: 0 } }); + await persistUsageReceipt(db, receipt(owner, sourceId, 1, 1, 100)); + await accountRunCost(db, owner.runId); + const conflicting = receipt(owner, sourceId, 2, 1.2, 120); + conflicting.receipt.biller = "another-biller"; + await persistUsageReceipt(db, conflicting); + await persistUsageReceipt(db, receipt(owner, sourceId, 3, 1.4, 140)); + for (let attempt = 0; attempt < 3; attempt++) await accountRunCost(db, owner.runId); + const charges = await db.select().from(costEvents).where(eq(costEvents.heartbeatRunId, owner.runId)); + expect(charges.reduce((sum, charge) => sum + charge.inputTokens, 0)).toBe(140); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, owner.runId)); + expect(run.usageJson).toMatchObject({ accountingLateReceiptPending: true, accountingLateReceiptId: conflicting.id }); + expect(run.accountingLastError).toBe("late_usage_requires_reconciliation"); + }); + it("corrects authoritative same-token late money on the original charge", async () => { const sourceId = randomUUID(); const owner = await seed({ usageJson: { accountingReceiptSourceId: sourceId, accountingReceiptSequence: 0 } }); diff --git a/server/src/services/run-cost-accounting.ts b/server/src/services/run-cost-accounting.ts index 350c90e6e97..ea9f8bd2b17 100644 --- a/server/src/services/run-cost-accounting.ts +++ b/server/src/services/run-cost-accounting.ts @@ -171,7 +171,10 @@ export async function accountRunCost(db: Db, runId: string, hooks: BudgetService await createCostEventInTransaction(tx, run.companyId, receipt, publications, undefined, { deferBudgetEvaluation: true }); } await queueBudgetEvaluation(tx, { companyId: run.companyId, agentId: run.agentId, projectId: text(scope.projectId) }); - await tx.update(heartbeatRuns).set({ costAccountingPending: false, costAccountedAt: new Date(), accountingProjectionVersion: "v2", accountingLastError: null, accountingLastAttemptAt: new Date() }).where(eq(heartbeatRuns.id, run.id)); + // 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)); return true; }), { "budget.accounting.owner": "run" }); for (const file of recoveredFiles) { diff --git a/server/src/services/usage-receipts.ts b/server/src/services/usage-receipts.ts index 02bf8b549df..d858d841c8f 100644 --- a/server/src/services/usage-receipts.ts +++ b/server/src/services/usage-receipts.ts @@ -8,7 +8,7 @@ import { z } from "zod"; import { and, eq, isNull, sql } from "drizzle-orm"; import { heartbeatRuns, runUsageReceipts, costEvents, costAdjustments, type Db } from "@paperclipai/db"; import type { AdapterExecutionResult, AdapterUsageCheckpoint } from "@paperclipai/adapter-utils"; -import { compareCents, normalizeCents, pricingProvenanceSchema, usdToUnits, unitsToCents } from "@paperclipai/shared"; +import { centsToUnits, compareCents, normalizeCents, pricingProvenanceSchema, usdToUnits, unitsToCents } from "@paperclipai/shared"; import { idleAccountingSpoolPath, registerIdleSpoolDirectory } from "./idle-local-work.js"; import { conflict, notFound } from "../errors.js"; import { logger } from "../middleware/logger.js"; @@ -19,7 +19,8 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notificatio import { knownRunCostCents } from "./budget-exposure.js"; import { createLiveRunUsageProjection, liveRunUsageCostCents } from "./budget-live-usage.js"; import { applyCostCorrectionInTransaction } from "./accounting-corrections.js"; -import type { ActivityPublication } from "./activity-log.js"; +import { logActivity, type ActivityPublication } from "./activity-log.js"; +import { createCostEventInTransaction } from "./costs.js"; import { queueBudgetEvaluation } from "./budget-evaluation-work.js"; const token = z.number().int().nonnegative().max(2_147_483_647); @@ -49,6 +50,9 @@ const envelopeSchema = z.object({ liveCostCents: z.string().max(160).refine(value => { try { return compareCents(value, 0) >= 0; } catch { return false; } }, "Invalid live cost").transform(normalizeCents).nullable().optional(), + // Set when another run's Runner stream reported this run's late usage. The + // receipt's usage and cost are then an increment, not a run total. + lateDelta: z.object({ attribution: z.enum(["exact", "inferred"]), reportedByRunId: z.string().uuid() }).optional(), }); export type UsageReceiptEnvelope = z.infer; export const usageReceiptSpoolPath = idleAccountingSpoolPath; @@ -106,7 +110,7 @@ export async function persistUsageReceiptInTransaction(tx: Db, raw: UsageReceipt sequence: envelope.sequence, receiptHash: hash, receiptJson: envelope, complete: envelope.receipt.complete, receivedAt: new Date(envelope.receivedAt), }); const previous = run.usageJson ?? {}; - if (run.costAccountedAt) { + if (run.costAccountedAt || envelope.lateDelta) { const activeSource = previous.accountingReceiptSourceId === envelope.sourceId; const closedSources = previous.accountingReceiptSourceFences as Record | undefined; if (!activeSource && typeof closedSources?.[envelope.sourceId] === "number" && envelope.sequence <= closedSources[envelope.sourceId]) @@ -123,7 +127,8 @@ 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.producerEvent?.eventType === "usage.reported" && !envelope.receipt.usage) + const changed = 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]; @@ -212,6 +217,66 @@ export async function persistUsageReceiptInTransaction(tx: Db, raw: UsageReceipt return { reconciliationRequired: false }; } +const receiptTokenKeys = ["inputTokens", "cachedInputTokens", "outputTokens"] as const; + +function lateDeltaTotals(usage: Record) { + const saved = (usage.accountingLateDeltas ?? {}) as Record; + const count = (key: string) => typeof saved[key] === "number" ? saved[key] as number : 0; + return { tokens: { inputTokens: count("inputTokens"), cachedInputTokens: count("cachedInputTokens"), outputTokens: count("outputTokens") }, + units: typeof saved.units === "string" ? BigInt(saved.units) : 0n }; +} + +function formatUsdUnits(units: bigint) { + return `${units / 1_000_000_000n}.${String(units % 1_000_000_000n).padStart(9, "0")}`; +} + +/** A provider can report more usage after its run's charge was posted. Cumulative + * reports make the increment exact, so post it instead of leaving the run wrong. + * Lower or conflicting evidence stays visible debt for an operator to review. */ +function lateUsageGrowth(previous: Record, late: UsageReceiptEnvelope, + event: typeof costEvents.$inferSelect, newer: boolean, reconciled?: UsageReceiptEnvelope["receipt"]) { + // A non-terminal report is never rate-card priced when it arrives, but its run + // totals are all this growth posts. Price them now, or the tokens stay free. + const receipt = priceAnthropicReceipt(priceCodexReceipt({ ...late.receipt, complete: true } as AdapterUsageCheckpoint)) as UsageReceiptEnvelope["receipt"]; + const accepted = (key: string) => typeof previous[key] === "number" ? previous[key] as number : 0; + const billingType = receipt.billingType === "api" ? "metered_api" + : receipt.billingType === "subscription" ? "subscription_included" : receipt.billingType ?? "unknown"; + if (!receipt.usage || receipt.usageBasis === "session_cumulative" || receipt.usageByModel?.length + || (Array.isArray(previous.usageByModel) && previous.usageByModel.length) || receipt.model !== event.model + || receipt.provider !== event.provider || (receipt.biller ?? receipt.provider) !== event.biller || billingType !== event.billingType) return null; + if (!late.lateDelta && (!newer || late.sourceId !== previous.accountingReceiptSourceId)) return null; + const reported = (key: typeof receiptTokenKeys[number]) => receipt.usage?.[key] ?? 0; + const deltas = lateDeltaTotals(previous); + // A source's cumulative receipts never include increments from later runs. + const tokens = Object.fromEntries(receiptTokenKeys.map(key => + [key, late.lateDelta ? reported(key) : reported(key) - (accepted(key) - deltas.tokens[key])])) as Record; + if (receiptTokenKeys.some(key => tokens[key] < 0) || receiptTokenKeys.every(key => tokens[key] === 0)) return null; + const moneyValue = (value: unknown) => typeof value === "number" || typeof value === "string" ? usdToUnits(value) : null; + const acceptedUnits = moneyValue(previous.cacheAdjustedCostUsd) ?? moneyValue(previous.costUsdExact) ?? moneyValue(previous.costUsd); + const reportedUnits = moneyValue(receipt.cacheAdjustedCostUsd) ?? moneyValue(receipt.costUsdExact) ?? moneyValue(receipt.costUsd); + const subscription = billingType === "subscription_included"; + // A price correction already moved the accepted total, so measure from it. + const correctedTotal = reconciled ? moneyValue(reconciled.cacheAdjustedCostUsd) ?? moneyValue(reconciled.costUsdExact) ?? moneyValue(reconciled.costUsd) : null; + const units = subscription ? 0n : reportedUnits === null ? null + : late.lateDelta ? reportedUnits : reportedUnits - (correctedTotal ?? (acceptedUnits ?? 0n) - deltas.units); + if (units !== null && units < 0n) return null; + const totalUnits = units === null ? acceptedUnits : (acceptedUnits ?? 0n) + units; + return { tokens, units, pricingProvenance: receipt.pricingProvenance, costStatus: receipt.costStatus, usage: { + inputTokens: accepted("inputTokens") + tokens.inputTokens, + cachedInputTokens: accepted("cachedInputTokens") + tokens.cachedInputTokens, + outputTokens: accepted("outputTokens") + tokens.outputTokens, + ...(late.lateDelta ? { accountingLateDeltas: { + inputTokens: deltas.tokens.inputTokens + tokens.inputTokens, + cachedInputTokens: deltas.tokens.cachedInputTokens + tokens.cachedInputTokens, + outputTokens: deltas.tokens.outputTokens + tokens.outputTokens, + units: String(deltas.units + (units ?? 0n)), + } } : {}), + // Keep the integrity check's receipt total equal to the run's charges. + ...(subscription || units === null || totalUnits === null ? {} : { cacheAdjustedCostUsd: null, costUsd: null, costUsdExact: formatUsdUnits(totalUnits) }), + ...(units === null && !subscription ? { costStatus: "unpriced" } : {}), + } }; +} + /** Caller owns the run row. A money-only authoritative revision can correct one * existing final charge. Ambiguous usage/source/model evidence remains debt. */ export async function reconcileLateRunUsageReceiptInTransaction(tx: Db, run: typeof heartbeatRuns.$inferSelect, publications: ActivityPublication[]) { @@ -238,8 +303,9 @@ export async function reconcileLateRunUsageReceiptInTransaction(tx: Db, run: typ .where(and(eq(costAdjustments.companyId, run.companyId), eq(costAdjustments.costEventId, events[0].event.id), sql`not (${costAdjustments.actorId} = 'cost_accounting' and left(${costAdjustments.idempotencyKey}, length(${automaticCorrectionPrefix})) = ${automaticCorrectionPrefix})`)).limit(1) : []; - const sameTokens = receipt?.usage && ["inputTokens", "cachedInputTokens", "outputTokens"].every(key => - Number(previous[key] ?? 0) === Number(receipt.usage?.[key as keyof typeof receipt.usage] ?? 0)); + const lateDeltas = lateDeltaTotals(previous); + const sameTokens = receipt?.usage && receiptTokenKeys.every(key => + Number(previous[key] ?? 0) - lateDeltas.tokens[key] === Number(receipt.usage?.[key] ?? 0)); const authoritative = receipt?.pricingProvenance && !["rate_card", "unknown"].includes(receipt.pricingProvenance.source); const billingType = receipt?.billingType === "api" ? "metered_api" : receipt?.billingType === "subscription" ? "subscription_included" : receipt?.billingType ?? "unknown"; @@ -267,9 +333,55 @@ export async function reconcileLateRunUsageReceiptInTransaction(tx: Db, run: typ }, }).where(eq(heartbeatRuns.id, run.id)); } + const growth = late && events.length === 1 && !reviewedValuation + ? lateUsageGrowth(previous, late, events[0].event, late.sequence > acceptedSequence, + reconciled ? envelopeSchema.parse(reconciled.receiptJson).receipt : undefined) : null; + if (growth && late) { + const { event } = events[0]; + const costStatus = growth.units === null && event.billingType !== "subscription_included" ? "unpriced" : growth.costStatus ?? event.costStatus; + await createCostEventInTransaction(tx, run.companyId, { + idempotencyKey: `${automaticCorrectionPrefix}${late.id}`, agentId: run.agentId, heartbeatRunId: run.id, + issueId: event.issueId, projectId: event.projectId, goalId: event.goalId, billingCode: event.billingCode, + provider: event.provider, biller: event.biller, providerRequestId: receipt?.providerRequestId ?? null, + pricingProvenance: growth.pricingProvenance ?? event.pricingProvenance, model: event.model, billingType: event.billingType, + costStatus, inputTokens: growth.tokens.inputTokens, cachedInputTokens: growth.tokens.cachedInputTokens, + outputTokens: growth.tokens.outputTokens, costCents: growth.units === null ? "0" : unitsToCents(growth.units), + occurredAt: new Date(late.receivedAt), + }, publications, undefined, { deferBudgetEvaluation: true }); + Object.assign(previous, growth.usage); + await logActivity(tx, { companyId: run.companyId, actorType: "system", actorId: "cost_accounting", action: "cost.late_usage_posted", + entityType: "heartbeat_run", entityId: run.id, runId: run.id, agentId: run.agentId, + details: { receiptId: late.id, attribution: late.lateDelta?.attribution ?? "exact", reportedByRunId: late.lateDelta?.reportedByRunId ?? run.id, + inputTokens: growth.tokens.inputTokens, cachedInputTokens: growth.tokens.cachedInputTokens, outputTokens: growth.tokens.outputTokens, + costCents: growth.units === null ? null : unitsToCents(growth.units) } }, publications); + await queueBudgetEvaluation(tx, { companyId: run.companyId, agentId: run.agentId, projectId: event.projectId }); + await updateDebtState(true, late.lateDelta === undefined); + if (Object.keys(debts).length === 0) await notifyDeliveryWork(tx, DELIVERY_QUEUES.strandedAgentStatus, + JSON.stringify({ companyId: run.companyId, agentId: run.agentId })); + return true; + } + // Totals are cumulative, so an older receipt at or below the accepted totals + // from this source is already covered. Conflicting billing stays visible debt. + if (late && !late.lateDelta && late.sourceId === previous.accountingReceiptSourceId && late.sequence <= acceptedSequence && receipt?.usage + && receipt.usageBasis !== "session_cumulative" && 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 + && receiptTokenKeys.every(key => (receipt.usage?.[key] ?? 0) <= Number(previous[key] ?? 0) - lateDeltas.tokens[key])) { + await queueBudgetEvaluation(tx, { companyId: run.companyId, agentId: run.agentId, projectId: events[0].event.projectId }); + await updateDebtState(true, false); + if (Object.keys(debts).length === 0) await notifyDeliveryWork(tx, DELIVERY_QUEUES.strandedAgentStatus, + JSON.stringify({ companyId: run.companyId, agentId: run.agentId })); + return true; + } if (!qualified) { await updateDebtState(false); return false; } + // A receipt's money covers the whole run. Earlier late charges already hold + // their share, so the original charge may only be corrected to the remainder. + const lateCharges = await tx.select({ exact: sql`${costEvents.costCents}::text` }).from(costEvents).where(and( + eq(costEvents.companyId, run.companyId), eq(costEvents.heartbeatRunId, run.id), + sql`left(${costEvents.idempotencyKey}, length(${automaticCorrectionPrefix})) = ${automaticCorrectionPrefix}`)); + const correctedUnits = usdToUnits(rawAmount) + lateDeltas.units - lateCharges.reduce((sum, charge) => sum + centsToUnits(charge.exact), 0n); + if (correctedUnits < 0n) { await updateDebtState(false); return false; } const corrected = receipt.billingType === "subscription_included" || receipt.billingType === "subscription" - ? "0.0000000" : unitsToCents(usdToUnits(rawAmount)); + ? "0.0000000" : unitsToCents(correctedUnits); if (reviewedValuation) { if (compareCents(events[0].exact, corrected) !== 0) { await updateDebtState(false); return false; } // The operator has already reconciled this exact billing evidence. Resolve @@ -312,12 +424,10 @@ export async function reconcileLateRunUsageReceiptInTransaction(tx: Db, run: typ export async function persistUsageReceipt(db: Db, raw: UsageReceiptEnvelope) { const envelope = envelopeSchema.parse(raw); - const result = await withOperationAccountingTransaction(db, envelope.companyId, + // Late evidence is committed as run-level debt that accounting resolves or + // surfaces. It never fails the delivering session or an unrelated later run. + await withOperationAccountingTransaction(db, envelope.companyId, tx => persistUsageReceiptInTransaction(tx, envelope)); - // Native delivery keeps its existing explicit reconciliation response, after - // the journal/debt commits. Retrying cannot discard the paid evidence. - if (result.reconciliationRequired && envelope.producerEvent) - throw conflict("native_usage_after_finalization_requires_reconciliation"); } export async function replayUsageReceipts(db: Db, directory = usageReceiptSpoolPath(), limit = 100) { @@ -615,7 +725,7 @@ const producerAccountingProjectionFields = [ "accountingReceiptOwner", "accountingReceiptSourceId", "accountingReceiptSequence", "accountingReceiptReady", "accountingReceiptId", "accountingReceiptReceivedAt", "accountingCaptureFailed", "accountingReceiptSourceFences", "accountingSourceReceiptPending", "accountingLateReceiptPending", "accountingLateReceiptId", - "accountingLateReceiptAttemptedId", "accountingReconciledReceiptId", "accountingLateReceiptDebts", + "accountingLateReceiptAttemptedId", "accountingReconciledReceiptId", "accountingLateReceiptDebts", "accountingLateDeltas", "accountingReconciledReceiptSourceId", "accountingReconciledReceiptSequence", "inputTokens", "outputTokens", "cachedInputTokens", "cacheWriteTokens", "provider", "biller", "model", "billingType", "usageSource", "providerRequestId", "usageByModel",