mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-04 11:01:16 +01:00
Merge reviewed master updates into sleep ingress repair
This commit is contained in:
commit
d1e3afa4af
14 files changed
+1090
-48
No files matched your search
@@ -1729,7 +1729,10 @@ Scheduler must skip invocation when:
|
||||
- an existing run is active
|
||||
- hard budget limit has been hit
|
||||
|
||||
Budget admission and accounting serialize under the company accounting lock.
|
||||
Authoritative paid-work admission and accounting mutations serialize under the
|
||||
company accounting lock. Preliminary invocation checks use a read-only snapshot
|
||||
and take that lock only when a hold, incident, or enforcement transition needs
|
||||
reconciliation; reconciliation rereads current state under the lock.
|
||||
The dispatch boundary rechecks current company, agent, and project policies and
|
||||
reserves capacity before paid work starts. Accounting commits budget blocks and
|
||||
durable stop intent without waiting for provider shutdown or reconciling unrelated
|
||||
|
||||
@@ -1602,8 +1602,9 @@ agent, workspace, retained session, and termination of prior execution. The
|
||||
checkpoint must match the exact Runner, lease, run, session, turn, and item.
|
||||
A missing restart claim does not remove these checks or authorize a fresh session.
|
||||
|
||||
After a plugin session passes provider recovery and durable run admission, its
|
||||
local suspension receipt is removed. The running session then owns authority;
|
||||
After a remote sandbox session passes provider recovery and durable run admission,
|
||||
its local suspension receipt is removed. This applies to built-in sandbox runners
|
||||
and plugin tasks. The running session then owns authority;
|
||||
the old receipt must not conflict with a later warm run's journal identity.
|
||||
Failed admission retains the receipt. Normal suspension writes a new receipt.
|
||||
|
||||
|
||||
@@ -45,10 +45,15 @@ Keep credential values out of logs and diagnostic reports.
|
||||
## Budget admission, accounting, and enforcement
|
||||
|
||||
The operator-gated `paperclip.budgets` tracer records
|
||||
`budget.admission.inspect`, `budget.admission.reconcile`,
|
||||
`budget.admission.evaluate`, `budget.admission.reserve`,
|
||||
`budget.accounting.transaction`, `budget.accounting.settlement`, `budget.accounting.lock_wait`,
|
||||
`budget.exposure.evaluate`, `budget.enforcement.deliver`,
|
||||
`budget.enforcement.scope`, `budget.cancellation.deliver`, and `budget.accounting.recover`.
|
||||
Inspection spans cover the preliminary read-only snapshot. Reconciliation spans
|
||||
exist only when that inspection finds a transition requiring the company lock;
|
||||
compare their counts to measure how often preliminary checks bypass the lock.
|
||||
Authoritative reservation and transactional evaluation retain their own spans.
|
||||
Transaction spans include commit time. Lock-wait spans cover acquisition of the
|
||||
company accounting lock. Settlement spans cover the ledger and reservation
|
||||
transaction. Enforcement spans measure asynchronous delivery; cancellation
|
||||
|
||||
@@ -0,0 +1,400 @@
|
||||
# Accounting contention and best-effort budget enforcement
|
||||
|
||||
Date: 2026-11-03
|
||||
Status: Investigation complete; first admission optimization implemented locally, remaining stages proposed
|
||||
Source revision: `367f29478d14003cc5c0f161b7ce8f3164993634`
|
||||
|
||||
Follow-up: [Lock-free accounting design](2026-11-03-lock-free-accounting-design.md)
|
||||
defines the requested target architecture. Its append-only receipt path replaces
|
||||
the run-lock proposal below; this document retains the investigation evidence
|
||||
and the possible behavior-preserving first migration step.
|
||||
|
||||
## Recommendation
|
||||
|
||||
Keep receipt persistence, deduplication, final settlement, and audit transactional.
|
||||
Make budget detection and automatic stopping a measured, short-lag control. Move
|
||||
frequent receipt writes to transactions scoped to their run or paid operation.
|
||||
Evaluate budget policies in coalesced batches outside those transactions.
|
||||
|
||||
First remove avoidable admission transactions and measure the remaining critical
|
||||
sections. Do not begin with distributed spending allocations or a new accounting
|
||||
service. A short lock at an infrequent boundary can be acceptable; serializing
|
||||
every company's receipt stream and admission preflight is the scaling concern.
|
||||
|
||||
The first implementation now inspects preliminary eligibility in a repeatable-read,
|
||||
read-only snapshot. Healthy checks and checks with no policies bypass the company
|
||||
writer lock. Holds, hard blocks, recovering accounting and incidents needing
|
||||
resolution fall back to the existing mutating owner, which revalidates current
|
||||
state under the lock. Transactional admission, reservations and receipt accounting
|
||||
retain their authority. Separate inspection/reconciliation spans expose the bypass
|
||||
ratio. This does not implement the later append-only accounting architecture.
|
||||
|
||||
Configured-policy tradeoffs: a healthy check adds an incident lookup, and an
|
||||
unheld blocking/recovering scope can compute exposure once during inspection
|
||||
and again during locked revalidation. Already-held scopes request reconciliation
|
||||
before computing exposure. Qualify configured-policy workloads as well as the
|
||||
predominant no-policy case; the change is not a claim of lower CPU everywhere.
|
||||
|
||||
The first milestone's budget-service suite passes 18 tests against real
|
||||
PostgreSQL. A controlled writer holds the company accounting lock while routine
|
||||
checks finish; authoritative admission on the same reader instead reaches its
|
||||
500 ms lock timeout. Coverage includes healthy/warning policies, manual and
|
||||
orphaned pauses, hold release, incident cleanup and cross-company rejection.
|
||||
This establishes the lock-bypass contract, not deployed latency or CPU savings.
|
||||
Seven additional accounting/enforcement/replay/tracing suites pass 55 tests,
|
||||
for 73 passing targeted tests across eight files with no skips.
|
||||
The full workspace typecheck passes via `node scripts/typecheck-low-memory.mjs`,
|
||||
and `node scripts/check-module-boundaries.mjs` passes.
|
||||
The full workspace build passes via `PAPERCLIP_TYPECHECK_LOW_MEMORY=1 pnpm -r
|
||||
--workspace-concurrency=1 build`, including the current-source Runner release
|
||||
binary, server, UI and CLI. The initial incomplete local Copilot build artifact
|
||||
was repaired with the pinned installer before the successful canonical retry.
|
||||
|
||||
This proposes a product-contract change for the later stages. It does not change
|
||||
the current spec or silently redefine existing configured budgets. The first
|
||||
implementation stage below preserves existing enforcement semantics.
|
||||
|
||||
## What the investigation established on the source revision above
|
||||
|
||||
### The lock protects several different responsibilities
|
||||
|
||||
[`withAccountingTransaction`](../../server/src/services/accounting-transaction.ts)
|
||||
takes `FOR NO KEY UPDATE` on the company row before invoking its callback. It
|
||||
holds that lock until the transaction ends. Different companies can proceed
|
||||
independently; writers for one company serialize. Ordinary report reads already
|
||||
use a read-only repeatable-read snapshot and do not take this writer lock.
|
||||
|
||||
| Path | Work sharing the company lock before this change |
|
||||
| --- | --- |
|
||||
| Usage receipt | Lock run, validate source/sequence and replay hash, insert immutable receipt, update running usage, evaluate affected budgets if monetary exposure changed |
|
||||
| Admission | Read company/agent/project state and policies, calculate exposure, reconcile budget holds/incidents, reserve estimated capacity before dispatch |
|
||||
| Final settlement | Recover durable receipt files, validate completion, settle reservation, insert ledger entries, update projections and runtime totals, acknowledge the run, evaluate budgets |
|
||||
| Other mutations | Direct cost reports, billing imports/corrections, accounting repairs, paid browser/decision/fast-response/classifier operations, and agent lifecycle operations |
|
||||
|
||||
The wrapper was unconditional. Having no configured budget did not bypass the
|
||||
lock. The public `getInvocationBlock` method also used this mutating wrapper.
|
||||
The implementation above separates its preliminary reads while preserving
|
||||
the reconciliation and resume effects that some callers need.
|
||||
|
||||
Primary owners are [usage receipts](../../server/src/services/usage-receipts.ts),
|
||||
[budgets](../../server/src/services/budgets.ts),
|
||||
[reservations](../../server/src/services/budget-reservations.ts), and
|
||||
[run settlement](../../server/src/services/run-cost-accounting.ts).
|
||||
Queue, dispatch, and lifecycle code should consume their contracts rather than
|
||||
implement another budgeting system.
|
||||
|
||||
### The current guarantee is already narrower than a dollar cap
|
||||
|
||||
Reservations serialize admission against known exposure and estimated holds.
|
||||
They do not restrict how much an already-dispatched provider can charge.
|
||||
`reservationCents` defaults to zero. A zero reservation still prevents duplicate
|
||||
dispatch, but does not reserve a positive allowance for future model spend.
|
||||
|
||||
The [implementation spec](../SPEC-implementation.md) explicitly acknowledges that
|
||||
mid-run enforcement depends on available provider reports or supported prices.
|
||||
Providers reporting only at completion cannot provide exact mid-call detection.
|
||||
Browser work can pass a provider cost cap; preserve such supported limits.
|
||||
|
||||
The current rules are approximately:
|
||||
|
||||
- Admission counts ledger spend plus the maximum of each unfinished operation's
|
||||
reservation and observed spend, plus separate browser exposure.
|
||||
- Threshold stopping uses ledger plus observed running spend, rather than
|
||||
treating reservation estimates as money already spent.
|
||||
- Pending terminal accounting and configured unknown-price restrictions can
|
||||
independently block work.
|
||||
- Final settlement replaces live exposure with the ledger charge atomically.
|
||||
|
||||
Accurate recording of received evidence and instantaneous control of external
|
||||
spend are separate contracts. Some prices remain estimates or unknown even when
|
||||
their receipt is recorded correctly.
|
||||
|
||||
### The critical section can do substantial work
|
||||
|
||||
When applicable policies exist, [exposure evaluation](../../server/src/services/budget-exposure.ts)
|
||||
executes four sequential aggregate statements over ledger entries, held
|
||||
reservations joined to runs, pending/live runs, and browser work. Policies are
|
||||
batched into those statements; it is not four statements per policy. With no
|
||||
billable policies, these four queries are skipped.
|
||||
|
||||
A changed monetary receipt can invoke this evaluation synchronously. Duplicate,
|
||||
stale, unchanged, and subscription receipts take different branches; not every
|
||||
receipt performs the aggregate scans. Native delivery already suppresses
|
||||
accepted source-event replays in its receiver, so duplicate suppression is a
|
||||
measurement question rather than an assumed large win.
|
||||
|
||||
The hot inputs are usage and turn-lifecycle events, not every streamed text
|
||||
token. Native capture handles usage reports and turn start/terminal events.
|
||||
Legacy capture also durably spools before its database attempt. Recovery can
|
||||
still read receipt files while the settlement transaction holds the company
|
||||
lock, although it builds an initial file-identity index outside the lock.
|
||||
|
||||
[Monthly spend projections](../../server/src/services/costs.ts) update
|
||||
`companies.spentMonthlyCents` for each ledger insert, and the applicable agent
|
||||
counter. Merely deleting the explicit company lock would leave an implicit
|
||||
shared-row write bottleneck and break assumptions around settlement and repair.
|
||||
PostgreSQL updates acquire conflicting row locks too. See the
|
||||
[PostgreSQL locking contract](https://www.postgresql.org/docs/current/explicit-locking.html#LOCKING-ROWS).
|
||||
|
||||
### Live traces justify a focused optimization, not a CPU diagnosis
|
||||
|
||||
Read-only observability queries examined one live host. The matched one-hour
|
||||
sample is **2026-11-03 16:45–17:45 UTC**, service revision
|
||||
`3910352c923afda320b2a5134e523932ee1d0633`.
|
||||
The aggregate measurements are reproduced below. Private trace links and host
|
||||
identifiers remain outside the repository.
|
||||
|
||||
| Observed span | Count | p50 | p95 | p99 | Maximum |
|
||||
| --- | ---: | ---: | ---: | ---: | ---: |
|
||||
| Company lock acquisition | 9,421 | 13.00 ms | 206.85 ms | 725.14 ms | 4,301.53 ms |
|
||||
| Accounting transaction | 9,421 | 87.16 ms | 486.95 ms | 1,147.10 ms | 5,594.19 ms |
|
||||
| Admission evaluation | 8,082 | 43.45 ms | 166.84 ms | 342.13 ms | 1,212.92 ms |
|
||||
| Final settlement | 216 | 355.90 ms | 1,894.05 ms | 3,149.67 ms | 5,594.33 ms |
|
||||
|
||||
These are span counts; nested spans must not be added as independent operations.
|
||||
There were 23 `budget.admission.reserve` spans in the same window. The much larger
|
||||
admission-evaluation count warrants caller attribution, not an assumption that
|
||||
all evaluations represent new paid dispatches.
|
||||
|
||||
Of 8,292 exposure evaluations on that host in that hour, **8,280 had zero policies**
|
||||
and 12 had one. That is 99.86% with zero policies at this call boundary; it does
|
||||
not prove that every company has no budgets. These empty evaluations are cheap,
|
||||
but their surrounding admission/accounting work still takes the company lock.
|
||||
|
||||
One inspected admission transaction took 4.183 seconds, with 4.090 seconds in
|
||||
lock acquisition. Its exposure child had zero policies and took about 0.026 ms.
|
||||
A separate settlement transaction took 5.594 seconds, including 4.302 seconds
|
||||
in acquisition and a zero-policy exposure evaluation.
|
||||
Both inspected traces report mean sample rate 1.
|
||||
|
||||
Limits of this evidence:
|
||||
|
||||
- The acquisition span measures the complete awaited lock query. It includes
|
||||
database/network round trips and Node scheduling; it is not a PostgreSQL
|
||||
lock-wait measurement. Transaction duration also includes transaction setup
|
||||
and commit. Do not subtract unrelated percentiles to estimate hold time.
|
||||
- The spans do not identify the locking company, blocker, receipt frequency,
|
||||
or operation caller. They do not establish one company's throughput ceiling
|
||||
or the cause of the previously observed CPU utilization.
|
||||
- The sample identifies one host. It does not establish fleet-wide behavior or
|
||||
independently qualify the public instance serving each request.
|
||||
- The local accounting wrapper, receipt, exposure, budget service, and run-cost
|
||||
files match the traced revision in Git. Other deployed components and database
|
||||
workload have not been qualified against this checkout.
|
||||
- The existing [budget benchmark](../../scripts/benchmark-budget-enforcement.ts)
|
||||
varies unrelated idle-agent count. It does not test concurrent receipt writers,
|
||||
10,000 active runs, or growing ledger history. It was inspected, not run here.
|
||||
|
||||
## Proposed behavior
|
||||
|
||||
| Responsibility | Proposed contract |
|
||||
| --- | --- |
|
||||
| Receipt acceptance | Durable commit before acknowledgement; source ownership, sequence, hash conflicts and company scope remain enforced |
|
||||
| Ledger and settlement | Idempotent transaction scoped to the run/invocation; no lost charges, duplicate application, or transient live-plus-final double counting |
|
||||
| Budget detection | Coalesced evaluation from committed evidence; a measurable delay is allowed |
|
||||
| New work | Check current authoritative pause/Stop and known budget-block state at dispatch; concurrent admission against a not-yet-reflected spend increase can overshoot |
|
||||
| Automatic cancellation | Commit durable stop intent, then use the existing scoped cancellation worker; provider shutdown remains a separate obligation |
|
||||
| Manual Stop and authorization | Preserve current execution-authority checks; budget approximation does not weaken them |
|
||||
| Unknown accounting | Retain unresolved obligations and existing configured unknown-usage restrictions; missing evidence is never a zero-cost receipt |
|
||||
|
||||
Suggested initial service targets, to validate before adopting as promises:
|
||||
|
||||
- Coalesce dirty budget work over approximately one second while work exists.
|
||||
- p99 from committed priced receipt to durable threshold-stop intent at most
|
||||
five seconds under the qualified workload. Measure provider reporting delay
|
||||
and cancellation latency separately.
|
||||
- For applicable spend limits, stop admitting additional paid work when the
|
||||
oldest unevaluated accounting change exceeds fifteen seconds. This is a
|
||||
proposed protection against a failed evaluator, not a hard spending cap.
|
||||
An idle company's old evaluation timestamp alone must never trigger it.
|
||||
- With no dirty changes, unresolved retries, or actual budget deadlines, the
|
||||
evaluator performs no recurring database scans.
|
||||
|
||||
Best effort needs a stated overspend model. As an illustration, 10,000 runs each
|
||||
spending $0.001/second consume $10/second; five seconds of extra detection delay
|
||||
adds about $50, before unreported and uncancellable in-flight work. Those are
|
||||
hypothetical rates, not a measured cost estimate or maximum.
|
||||
|
||||
In general, additional spend depends on fleet burn rate multiplied by reporting,
|
||||
evaluation, and stop delay, plus already-committed work. Without bounds on those
|
||||
quantities, do not advertise a bounded dollar overshoot. A strict end-to-end cap
|
||||
would require provider-enforced limits or preauthorized bounded work; distributed
|
||||
allocations alone cannot cap a provider that can exceed its allocation.
|
||||
|
||||
## Implementation stages
|
||||
|
||||
### 1. Attribute the cost and eliminate unnecessary admission transactions
|
||||
|
||||
Owners: accounting transaction/tracing services, budget admission API, and its
|
||||
queue/dispatch/lifecycle callers.
|
||||
|
||||
Add a closed operation-kind label to accounting spans so receipt, finalization,
|
||||
preflight, reservation, policy edit, correction, and lifecycle work can be
|
||||
distinguished. Capture acquisition-query time, time from acquisition return to
|
||||
transaction completion, query count, and replay/outcome categories. Keep the
|
||||
operator-gated Observability contract and content-free attributes; this is not
|
||||
first-party Telemetry. Label the client-observed held interval honestly because
|
||||
it includes commit/return latency. Use sampled PostgreSQL wait/blocker evidence
|
||||
to establish actual contention, with SQL values and credentials excluded.
|
||||
|
||||
Separate read-only preliminary eligibility from mutating reconciliation.
|
||||
Preflights should not take a company writer lock just to discover no applicable
|
||||
policy. Keep the existing authoritative reservation and policy check at final
|
||||
dispatch in this stage. Budget changes, rollover, settlement and explicit
|
||||
reconciliation continue to own their necessary effects. Do not introduce a
|
||||
stale cached "no budgets" result that can bypass a newly configured limit.
|
||||
|
||||
Measure identical receipt replay and unchanged snapshots before prioritizing
|
||||
their optimization. Any early replay return must prove the immutable receipt
|
||||
and relevant projection state agree. Native event identity and lifecycle events
|
||||
cannot be discarded just because their monetary amount is unchanged.
|
||||
|
||||
Deliver a real-PostgreSQL benchmark using the service boundaries below, and the
|
||||
small behavior-preserving optimization justified by the caller measurements.
|
||||
If this removes today's significant delay, ship it independently; later stages
|
||||
remain a deliberate scale/product decision.
|
||||
|
||||
### 2. Persist receipts per run and evaluate budgets asynchronously
|
||||
|
||||
Owners: usage receipts, native usage delivery, budget exposure, budgets, and the
|
||||
existing accounting/enforcement delivery lifecycle.
|
||||
|
||||
In one run-scoped transaction:
|
||||
|
||||
1. Validate current source/ownership and lock the run.
|
||||
2. Insert or verify the receipt with its unique identity and fingerprint.
|
||||
3. Replace the run's accepted cumulative usage/exposure projection.
|
||||
4. Persist a versioned pending-evaluation marker and register the normal
|
||||
transaction-aware wake.
|
||||
5. Commit before acknowledging native delivery or removing legacy spool evidence.
|
||||
|
||||
Do not take the company accounting writer lock or evaluate all affected policies
|
||||
inside this receipt transaction. Keep final and partial receipts durable; batch
|
||||
derived budget evaluation rather than dropping intermediate accounting evidence.
|
||||
|
||||
Use the shared delivery coordinator for coalescing, retry and deadlines. Pending
|
||||
state must be scoped to the changed run/operation, or an append-only change;
|
||||
updating one company dirty counter for every receipt would recreate the lock.
|
||||
Consumption must compare revisions or claim durable work so a concurrent change
|
||||
cannot be cleared by an older evaluator. A timestamp/sequence high-water mark
|
||||
alone cannot safely assume database transactions commit in allocation order.
|
||||
|
||||
Read ledger and live contributions in one consistent database snapshot, or use
|
||||
transactionally consistent replacement contributions. The present four separate
|
||||
read-committed aggregates are not safe to reuse unchanged once other operations
|
||||
in the same company can settle concurrently: live-to-ledger
|
||||
replacement must not be missed or counted twice across queries.
|
||||
|
||||
Evaluate each affected policy once per batch, revalidate its revision before
|
||||
committing block/resume state, and retain durable cancellation generations.
|
||||
Updates arriving during evaluation remain pending. Recovery reconstructs pending
|
||||
work after restart; rollback must not publish a false wake or acknowledgement.
|
||||
Use short policy-state transactions where necessary. Preserve manual holds and
|
||||
committed stop obligations during policy changes and price corrections.
|
||||
|
||||
This is the first stage that permits an additional detection/admission race
|
||||
window. Update the product contract together with implementation; removing only
|
||||
the wrapper is not a valid intermediate change.
|
||||
|
||||
### 3. Remove remaining company writes from frequent settlement
|
||||
|
||||
Owners: run-cost accounting, costs, billing reconciliation/import, accounting
|
||||
integrity, paid-service invocation owners, and DB/shared contracts.
|
||||
|
||||
Settle ledger entries, reservation state, final receipt acknowledgement, and
|
||||
run-specific accounting together under the operation's lock. Use database
|
||||
uniqueness plus conflict/hash verification for concurrent idempotency; a
|
||||
read-before-insert check previously protected by the company lock is insufficient.
|
||||
Retain narrow agent coordination for shared runtime totals where needed.
|
||||
|
||||
Move company monthly display-counter updates to versioned, coalesced projection
|
||||
work. Update integrity inspection/repair and all counter consumers to understand
|
||||
projection freshness and checkpoint identity. Reporting from ledger snapshots
|
||||
remains available; intentional projection lag must not look like lost money.
|
||||
Handle backdated events, credits/corrections, UTC rollover and lifetime budgets.
|
||||
|
||||
For receipt-file recovery, stage immutable candidate envelopes outside the broad
|
||||
transaction, then validate source/version and closure under the operation lock.
|
||||
Retain disk evidence until the final owning transaction commits. A concurrent
|
||||
late receipt or replaced source must still prevent premature acknowledgement.
|
||||
|
||||
Apply the same boundary to browser runs, decision-model invocations, fast
|
||||
responses, prompt-injection classification, and externally reported charges.
|
||||
Inventory administrative users of `withAccountingTransaction` separately;
|
||||
agent deletion, repair, and lifecycle safety must not depend on a lock removed
|
||||
from receipt writers. Specify and test the new lock order, including issue/run
|
||||
ownership and foreign-key interactions. Keep supported provider spending caps.
|
||||
|
||||
Start with bounded batches and indexed aggregates. If history growth or batch
|
||||
time misses the target, maintain incremental scope/window contributions. If
|
||||
those counters become the next measured hot row, partition their writes and
|
||||
sum a bounded set of partitions. Do not synchronously increment a single new
|
||||
company total on every receipt and call the bottleneck solved.
|
||||
|
||||
### 4. Qualify workload and roll out
|
||||
|
||||
Extend the existing benchmark rather than using idle-agent count as scale proof.
|
||||
Use real PostgreSQL with at least two independent controller processes and
|
||||
explicitly recorded connection-pool limits. Simulated provider events should
|
||||
exercise real receipt ingestion and settlement, followed by a smaller real
|
||||
provider qualification. Keep load tests off the active staging instance.
|
||||
|
||||
| Dimension | Required cases |
|
||||
| --- | --- |
|
||||
| Distribution | 100, 1,000, then 10,000 active runs; one company and many companies |
|
||||
| Receipt cadence | One update per run per 60 seconds, 10 seconds and 1 second; record actual target-provider cadence before choosing the release load |
|
||||
| Policies | None; company; agent/project only; overlapping scopes and windows |
|
||||
| History | Increasing ledger history independently of active-run count |
|
||||
| Event shape | Cost-changing, unchanged, subscription, duplicate, stale, conflicting and out-of-order reports; reconnect bursts |
|
||||
| Races/failures | Simultaneous settlement and receipt, worker crash, failed/ambiguous commit, source replacement, policy change, manual Stop, late final charge, unknown price and UTC rollover |
|
||||
|
||||
At 10,000 runs the illustrative receipt cadences imply about 167, 1,000, and
|
||||
10,000 receipts/second before admission or settlement. This defines a workload,
|
||||
not a claim that current or proposed code supports it. A shared critical section
|
||||
held for an average of H seconds has an ideal ceiling near 1/H operations/second
|
||||
for that company; do not estimate H from the unrelated percentile summaries.
|
||||
|
||||
Acceptance requires durable receipt/ledger equality after quiescence, no duplicate
|
||||
effect or acknowledgement loss, stable pending-work depth, correct policy/manual
|
||||
hold behavior, and measured detection/cancellation/freshness targets. Record
|
||||
p50/p95/p99 ingestion, admission and settlement latency, actual PostgreSQL wait
|
||||
events, query count/rows/bytes, CPU/event-loop delay, and reconnect recovery time.
|
||||
Prove unrelated companies remain responsive and writers for unrelated runs no
|
||||
longer contend on a shared company update in the frequent path.
|
||||
|
||||
Initially compute the proposed evaluation in shadow mode without stop/resume
|
||||
effects and compare it against the authoritative ledger and existing decisions.
|
||||
Then canary the new enforcement contract on an explicitly selected environment,
|
||||
with its versioned pending work rebuilt and validated before switching. Do not
|
||||
run old and new enforcement authorities simultaneously for the same company.
|
||||
Rollback must drain/reconcile accepted receipts and rebuild authoritative state
|
||||
before restoring the older path; flipping a flag alone is insufficient.
|
||||
|
||||
Relevant existing coverage includes scoped budget enforcement, live attempts,
|
||||
native usage delivery, run-cost accounting, budget cancellation, worker recovery,
|
||||
and idle-accounting work. Add focused concurrency/failure tests for the changed
|
||||
boundaries; run the repository's full checks before a PR-ready implementation
|
||||
handoff. Browser qualification is needed if dispatch/Stop or budget UI semantics
|
||||
change, not to validate this plan file.
|
||||
|
||||
## Contract and delivery boundaries
|
||||
|
||||
Before enabling the new behavior, update `doc/SPEC.md`,
|
||||
`doc/SPEC-implementation.md` (especially scheduler and budget enforcement), the
|
||||
applicable contributor invariants, shared budget contracts, and user-facing
|
||||
budget explanations. The UI should say what auto-stop does and disclose possible
|
||||
in-flight overspend; it should not expose internal partition mechanics. Decide
|
||||
how configured existing reservations retain their estimate/visibility role
|
||||
without promising atomic aggregate admission in the new mode.
|
||||
|
||||
Distributed bounded spending allocations are deferred. Revisit them only if
|
||||
customers need stricter aggregate admission or qualified batch evaluation cannot
|
||||
meet latency/overspend needs. They require provider cost bounds as well as
|
||||
replenishment, revocation and policy-version coordination to become a monetary
|
||||
guarantee. They are not required for the recommended first implementation.
|
||||
|
||||
Investigation validation: inspected the source and existing test/benchmark
|
||||
contracts, queried live spans and two individual traces, and checked the relevant
|
||||
deployed-revision source diff. No service, configuration, database, or deployed
|
||||
code was changed. Runtime tests and load tests were not run for this planning
|
||||
change; dependency installation and implementation belong to the next stage.
|
||||
@@ -0,0 +1,432 @@
|
||||
# Lock-free accounting and best-effort budgets
|
||||
|
||||
Date: 2026-11-03
|
||||
Status: Target architecture proposed; preliminary admission optimization implemented locally
|
||||
Scope: Accounting ingestion, settlement, budget projections and budget admission
|
||||
Evidence: [Accounting contention investigation](2026-11-03-accounting-contention-and-budget-enforcement.md)
|
||||
|
||||
## Decision
|
||||
|
||||
Make the frequent accounting write path append-only. A receipt writes its own
|
||||
immutable fact and commits. It does not lock or update the company, agent, run,
|
||||
budget, aggregate counter, or worker checkpoint. Independent runs use independent
|
||||
keys, including when they belong to the same company or agent.
|
||||
|
||||
Asynchronous workers derive operation state, ledger entries and partitioned spend
|
||||
totals. Budget admission reads a recent view without making a synchronized debit
|
||||
against a company balance. Threshold detection and cancellation may lag; durable
|
||||
evidence, deterministic accounting and execution authorization remain required.
|
||||
|
||||
"Lock-free" here means no application mutex, advisory lock, or `SELECT FOR
|
||||
UPDATE` in the receipt or budget-check path, and no shared accounting row written
|
||||
for each receipt. It is not a formal nonblocking progress guarantee: PostgreSQL
|
||||
still uses internal locks, unique-index arbitration, foreign-key checks and
|
||||
transaction commit. In particular, two inserts with the same unique key can
|
||||
wait for each other. See [PostgreSQL INSERT](https://www.postgresql.org/docs/current/sql-insert.html).
|
||||
|
||||
Background batches and infrequent policy changes use optimistic conditional
|
||||
writes. Those also take database row locks internally. Moving a compare-and-swap
|
||||
onto the same company counter for every receipt would fail this design.
|
||||
|
||||
Existing issue ownership, provider-dispatch fencing, manual Stop and resource
|
||||
ownership are outside this lock-removal scope. Accounting cannot replace those
|
||||
contracts with an eventually consistent spend view.
|
||||
|
||||
## Problem fit and first milestone
|
||||
|
||||
This design removes the identified company-wide accounting serialization point
|
||||
if receipt ingestion, admission, settlement and shared counters all migrate.
|
||||
It does not establish that accounting is the current main-loop CPU culprit or
|
||||
that the resulting system supports 10,000 active agents. Those remain separate
|
||||
claims requiring measurements.
|
||||
|
||||
The matched investigation sample contained 8,280 zero-policy exposure evaluations
|
||||
out of 8,292. Those exposure evaluations returned almost immediately; the
|
||||
surrounding admission path still read state and policies inside the accounting
|
||||
transaction. This makes separating eligibility reads from reconciliation the
|
||||
most directly supported first change. It does not show that large exposure
|
||||
scans or a high receipt rate caused the sampled delays.
|
||||
|
||||
The 9,421 accounting transactions in that hour average about 2.6 per second
|
||||
across the observed host. Burst rate, company distribution and actual lock hold
|
||||
time are missing. The acquisition spans include database/network time and Node
|
||||
scheduling, so their long tails alone do not prove PostgreSQL row-lock saturation.
|
||||
|
||||
| Problem | What this design changes | Remaining condition |
|
||||
| --- | --- | --- |
|
||||
| Eligibility checks enter a mutating company transaction | Read-only admission, separate reconciliation | Classify callers and preserve manual holds, resume behavior and execution fencing |
|
||||
| Independent receipts serialize by company | Independent immutable inserts | Qualify actual receipt rate, index/FK waits and durable ingestion capacity |
|
||||
| Repeated exposure work and shared counters limit growth | Coalesced evaluation and partitioned projections | Show less work per receipt and sufficient projection throughput |
|
||||
| Main-loop CPU/event-loop contention | Some synchronous work leaves the request path | Same-process reducers still consume CPU; batching must reduce work or processing must move to another runtime |
|
||||
|
||||
The immutable identity model, closure changes and partition publication protocol
|
||||
are target-architecture work, not prerequisites for the first latency improvement.
|
||||
An asynchronous pipeline also adds backlog and recovery obligations: its stale
|
||||
decision gate can become an admission outage if reducers cannot keep up. Merely
|
||||
moving identical work behind an ACK would conceal latency without solving load.
|
||||
|
||||
Start with an independently shippable milestone: attribute admission calls and
|
||||
transaction work, provide a pure eligibility read (including a no-policy fast
|
||||
path), and move required reconciliation to its explicit owner. Retain the current
|
||||
final admission/reservation authority during this initial behavior-preserving
|
||||
step. Having no configured policy must not bypass manual company/project pauses,
|
||||
existing holds, unknown-accounting restrictions or duplicate-dispatch protection.
|
||||
|
||||
Compare end-to-end dispatch latency, accounting transactions and queries per
|
||||
dispatch, database waits, CPU and event-loop delay under the same workload.
|
||||
Then qualify receipt coalescing and the remaining single-company bottleneck.
|
||||
Use the full migration below when that evidence or the agreed capacity target
|
||||
justifies it. The accounting rewrite does not cover the other independently
|
||||
identified work in run-event persistence, recovery, snapshots or fanout.
|
||||
|
||||
## Architecture
|
||||
|
||||
```mermaid
|
||||
flowchart LR
|
||||
P["Runner / paid service"] --> I["Validate and append receipt"]
|
||||
I --> J[("Immutable receipt journal")]
|
||||
J --> A["ACK after durable commit"]
|
||||
J --> R["Independent partition reducers"]
|
||||
R --> L[("Ledger + operation views")]
|
||||
R --> T[("Spend totals by scope / window / partition")]
|
||||
T --> B["Coalesced budget evaluator"]
|
||||
B --> V[("Versioned budget decisions")]
|
||||
V --> D["Dispatch: read decision, record own admission"]
|
||||
B --> S[("Durable stop intent")]
|
||||
S --> C["Existing scoped cancellation path"]
|
||||
```
|
||||
|
||||
The journal, reducers and evaluator can start in the existing server and
|
||||
PostgreSQL deployment. Independent scaling is a module/runtime boundary, not a
|
||||
requirement for Kafka, Redis, another service, or a separate database today.
|
||||
|
||||
## 1. Give every paid operation a stable identity
|
||||
|
||||
Use one accounting operation for each heartbeat run, browser invocation,
|
||||
decision-model invocation, fast response, classifier invocation, or separately
|
||||
reported charge. Record its company and attribution before paid dispatch.
|
||||
Retries of the same external operation reuse its identity; a genuinely new paid
|
||||
attempt has a distinct attempt identity beneath that operation.
|
||||
|
||||
An accounting operation's immutable identity is separate from mutable run state.
|
||||
Receipts reference this identity, rather than acquiring foreign-key protection
|
||||
on a busy mutable `heartbeat_runs` row. The existing journal directly references
|
||||
that row, so removing the explicit lock alone does not eliminate this possible
|
||||
wait. Use a company-scoped composite reference to the immutable identity; keep
|
||||
company-boundary validation and the existing deletion/retention obligations.
|
||||
Separate any nullable live-run pointer from retained original attribution.
|
||||
|
||||
Admitted accounting sources have durable identities and provenance. Controller
|
||||
reconnection does not manufacture a new paid attempt. Source replacement must
|
||||
retain the existing verified handoff/baseline semantics. A higher timestamp,
|
||||
random UUID, or newly generated controller source ID is not proof of authority
|
||||
over earlier billing evidence.
|
||||
|
||||
Source handoff remains an infrequent execution-authority operation. Ordinary
|
||||
checkpoints do not mutate its ownership record. Existing authentication and
|
||||
company access checks still apply. Authority to report a closing receipt after
|
||||
Stop is accounting-only; it never grants authority to continue execution.
|
||||
|
||||
## 2. Receipt ingestion only appends evidence
|
||||
|
||||
Extend/generalize the existing `run_usage_receipts` contract. The conceptual
|
||||
receipt contains:
|
||||
|
||||
| Field | Purpose |
|
||||
| --- | --- |
|
||||
| Company and operation IDs | Tenant boundary and accounting unit |
|
||||
| Attempt and admitted source identity | Separate paid attempts from reconnect/replay |
|
||||
| Stable source event ID and source sequence | Replay identity and ordering within that source |
|
||||
| Receipt kind and normalization version | Partial usage, final usage, source handoff, closure, correction |
|
||||
| Cumulative usage, price and provenance | Evidence; unknown price stays unknown |
|
||||
| Payload fingerprint | Detect conflicting reuse of an identity |
|
||||
| Server-assigned ingestion time and deterministic partition | Recovery, pending-work age and routing |
|
||||
|
||||
Provider event time remains separate. Freshness protection never trusts a caller's
|
||||
timestamp. External charge callers need a stable request identity to obtain
|
||||
idempotent retries; matching dollar amounts or payload similarity cannot prove
|
||||
that two otherwise unidentified charges are the same purchase.
|
||||
|
||||
Within a small transaction, validate the admitted identity and input, register
|
||||
the existing transaction-aware work signal, and insert the immutable receipt.
|
||||
Enforce the logical identity with a unique constraint. On conflict, fetch the
|
||||
committed existing receipt and compare its fingerprint; equal means replay,
|
||||
different means a conflict requiring reconciliation. Do not use `DO UPDATE`
|
||||
to replace receipt evidence. At read committed isolation, a conflict check may
|
||||
need a subsequent statement to see the winning committed row.
|
||||
|
||||
Commit before ACK. An ACK means the evidence is durable, not that the spend view
|
||||
has already incorporated it. Preserve legacy durable spooling and native
|
||||
redelivery when the database write or its acknowledgement is uncertain.
|
||||
|
||||
There is no read-modify-write of `usageJson`, `spentMonthlyCents`, reservation
|
||||
capacity, or a company-wide dirty flag here. The new receipt is itself durable
|
||||
pending work. Mutable delivery metadata may be attached to it by the consumer;
|
||||
its evidence fields are immutable.
|
||||
|
||||
Batch inserts with count/byte bounds when useful. Keep lifecycle, final, unknown
|
||||
and conflicting evidence intact. Do not drop a receipt solely because its dollar
|
||||
amount matches the previous one: it may change completeness or provenance.
|
||||
|
||||
## 3. Reduce independent partitions, then publish batches atomically
|
||||
|
||||
Route by a stable hash of **operation ID**, not company ID. For example, a
|
||||
starting benchmark can use 64 virtual partitions per company. That number is a
|
||||
test parameter, not a capacity claim. One company with 10,000 active runs spreads
|
||||
over all partitions; one agent with many runs spreads too.
|
||||
|
||||
Persist the layout version and partition with the operation's admission. All its
|
||||
receipts, closure and corrections follow that assignment. Changing partition
|
||||
count requires an explicit migration; do not silently rehash active operations.
|
||||
|
||||
The minimal conceptual storage is:
|
||||
|
||||
| Record | Write frequency and ownership |
|
||||
| --- | --- |
|
||||
| Operation identity | Once per admitted operation; immutable attribution |
|
||||
| Receipt journal | Append per receipt; unique identity; durable pending metadata |
|
||||
| Operation view | Replaced by its partition reducer, at most once per operation per batch |
|
||||
| Scope/window/partition totals | Updated in a bounded batch, never by receipt ingestion |
|
||||
| Partition publication head | One conditional generation advance per batch |
|
||||
| Budget policy and decision | Policy edits and coalesced evaluations; independent of receipt acceptance |
|
||||
| Ledger, adjustments, incidents, stop intent | Existing domain records, adapted to operation identities and idempotent writes |
|
||||
|
||||
Some records can extend existing tables. Separate immutable identities from
|
||||
mutable projections; do not add a table for each timer or delivery stage.
|
||||
|
||||
For a reducer batch:
|
||||
|
||||
1. In one repeatable-read snapshot, capture partition generation `g`, a bounded
|
||||
set of unprocessed receipt IDs, and their prior operation views.
|
||||
2. Reduce those operations and compute replacement contributions outside the
|
||||
publishing transaction. No provider calls or filesystem work occurs during
|
||||
publication.
|
||||
3. In one short transaction, conditionally advance the partition head from `g`
|
||||
to `g + 1`, bulk-write operation views and aggregate deltas, insert any
|
||||
qualified ledger settlement, and mark exactly the consumed receipt IDs.
|
||||
4. If the conditional advance loses, roll back and retry from a fresh snapshot.
|
||||
Use bounded retries with scheduler backoff under contention.
|
||||
|
||||
All writers of those views, including repairs, follow this publication protocol.
|
||||
A normally assigned reducer avoids duplicate work; overlapping workers or an
|
||||
old worker after failover remain safe because publication requires the captured
|
||||
generation. No durable exclusive lease is needed for correctness of this batch.
|
||||
|
||||
An input arriving during computation remains pending. Never advance a global
|
||||
`last_seen_id` or timestamp past all smaller values: sequence allocation and
|
||||
transaction commit order differ. An earlier allocated receipt that commits
|
||||
later must still be discoverable. Retain an indexed pending predicate.
|
||||
|
||||
This has bounded coordination per partition **per batch**. It cannot stall
|
||||
receipt ACKs behind the calculation, and different partitions do not write the
|
||||
same aggregate rows. Shared company/agent/project totals are sums of these
|
||||
partition contributions, not counters incremented by every producer.
|
||||
|
||||
## 4. Define canonical money without depending on arrival order
|
||||
|
||||
The reducer preserves current normalization and source-ownership rules as a
|
||||
versioned pure function over admitted evidence. Within a source, order by the
|
||||
producer sequence, not database arrival time. Across sources, follow verified
|
||||
handoff/continuation provenance rather than arbitrarily choosing a winner.
|
||||
|
||||
Cumulative reports replace earlier reports for the same accounting component.
|
||||
Reports of $2, $3 and $5 represent $5, not $10. Distinct paid attempts contribute
|
||||
separately; session-cumulative counters require their established baseline.
|
||||
An incomplete/unknown report cannot erase known partial exposure. Final prices
|
||||
and audited corrections may legitimately reduce earlier estimates.
|
||||
|
||||
Each operation view owns mutually exclusive contribution categories:
|
||||
|
||||
```text
|
||||
unsettled: ledger = 0; live = known observed amount; hold = max(estimate, live)
|
||||
settled: ledger = final; live = 0; hold = 0
|
||||
```
|
||||
|
||||
This is a contribution model, not a declaration that unknown exposure is zero.
|
||||
Carry unknown-price, pending-terminal and unresolved-dispatch counts separately
|
||||
and apply their configured restrictions. Hosted browser work remains a separate
|
||||
paid operation from its parent run.
|
||||
|
||||
Settlement and that operation's replacement contribution publish atomically.
|
||||
Company/agent/project views sum one contribution per operation. A reader sees
|
||||
the old or new contribution, never both the live estimate and its final charge.
|
||||
UTC-month and lifetime attribution, unsettled carryover, credits, backdated
|
||||
charges and corrections need explicit reducer rules matching the selected
|
||||
product contract; do not assign costs to the month in which a worker ran.
|
||||
|
||||
### Finalization requires closure, not merely a "complete" checkpoint
|
||||
|
||||
The governing execution coordinator supplies immutable closure evidence only
|
||||
after the operation cannot start another ordinary paid attempt. It references
|
||||
the admitted attempts/sources and the final usage evidence or unresolved debt.
|
||||
An adapter's `complete` usage flag alone does not establish that boundary.
|
||||
|
||||
A closure record must prove the final cumulative coverage it names. The current
|
||||
native `sourceSeq` includes non-accounting events, so gaps between usage receipts
|
||||
are not automatically lost usage. Use explicit final references/coverage, or a
|
||||
dedicated accounting sequence if the producer contract is extended. Do not
|
||||
invent a contiguous usage sequence from today's sparse source sequence.
|
||||
|
||||
Publish a settlement certificate and its ledger components in one transaction,
|
||||
with one base settlement per `(company_id, operation_id)` and a verified content
|
||||
fingerprint. A different closure for an already-settled operation is a conflict
|
||||
or an explicit correction, not a second base settlement.
|
||||
Only the transaction that inserts that certificate inserts its components; a
|
||||
replay verifies the existing certificate. Ledger IDs and component keys are
|
||||
stable across retries. Unknown external outcomes remain unresolved.
|
||||
|
||||
Late evidence within the declared closure coverage is replay/recovery. Evidence
|
||||
beyond it remains durable and creates a reconciliation obligation or an explicit
|
||||
correction; it cannot silently reopen the same final charge or disappear because
|
||||
the run was marked terminal. Corrections use immutable adjustment identities.
|
||||
Existing mutable correction/reporting APIs need an adapter to this ledger model.
|
||||
|
||||
## 5. Budget checks read; they do not reserve a shared balance
|
||||
|
||||
For dispatch, read current company/agent/project controls and applicable budget
|
||||
decisions in one bounded read from the authoritative database, rather than an
|
||||
unbounded-lag replica or cache. A policy without a current decision must
|
||||
be evaluated or treated as pending; it is not implicitly unlimited. The decision
|
||||
records policy revision, budget window, projection generations, status and
|
||||
freshness evidence.
|
||||
|
||||
If applicable decisions allow work, record this operation's own admission with
|
||||
the decision versions and estimated cost. Preserve duplicate-dispatch fencing
|
||||
in the execution subsystem. An admission receipt is not proof that an external
|
||||
provider call occurred exactly once; existing unknown-outcome recovery remains.
|
||||
|
||||
There is no atomic decrement of company remaining budget. Estimate-fit checks
|
||||
can use the same snapshot, and admitted estimates can enter the asynchronous
|
||||
exposure view. Two dispatches can see the same remaining money and both start.
|
||||
That race is accepted by this best-effort contract. Reservations become recorded
|
||||
estimates/obligations, not exclusive claims on a shared spending allowance.
|
||||
|
||||
New snapshots after a committed budget block deny additional work. Admissions
|
||||
already in flight can still race that block. Manual Stop, pause, permissions,
|
||||
issue ownership and dispatch authority retain their existing final gates.
|
||||
Keep provider-enforced cost caps where supported.
|
||||
|
||||
## 6. Coalesce budget decisions and preserve stop authority
|
||||
|
||||
The evaluator reads the applicable scope/window partition totals in one
|
||||
consistent snapshot, rather than scanning ledger and receipt history on each
|
||||
checkpoint. Capture the policy's configuration revision and prior decision
|
||||
revision in that same snapshot.
|
||||
|
||||
Publish with a conditional update against both captured revisions. A policy
|
||||
edit advances its configuration revision and invalidates the previous decision.
|
||||
An older evaluator cannot overwrite a newer decision or a changed policy. A
|
||||
losing evaluator recomputes; it never retries publication using a fresh version
|
||||
number attached to stale calculated totals.
|
||||
|
||||
Publish the decision, any incident/approval transition, and durable stop intent
|
||||
in the same transaction. Generate idempotency identities from the winning
|
||||
transition; losing or repeated evaluations produce no duplicate side effects.
|
||||
Only actual state transitions need notification/audit effects. Cancellation
|
||||
delivery remains at least once, scoped and recoverable. Already-committed
|
||||
per-run Stop obligations survive later policy changes. Resuming a budget hold
|
||||
cannot clear a manual hold.
|
||||
|
||||
These occasional policy-row updates can contend with one another. Producers
|
||||
and admission readers do not wait on that row to record receipts or debit money.
|
||||
Do not replace them with per-receipt policy updates.
|
||||
|
||||
## 7. Bound lag and recover work without an idle polling loop
|
||||
|
||||
Retain the proposed healthy-load targets from the investigation: roughly
|
||||
one-second coalescing and a five-second p99 target from durable priced receipt
|
||||
to stop intent. Provider reporting and shutdown delay are additional. Targets
|
||||
require a workload qualification; none is a guarantee on final billed dollars.
|
||||
|
||||
For scopes with spending limits, reject additional paid admissions if an
|
||||
unrepresented accounting change has been waiting over fifteen seconds. This
|
||||
check considers pending journal inputs **and** reduced contributions not yet
|
||||
covered by the decision's projection generations. Carry their oldest pending
|
||||
time through publication. Consuming the journal must not hide evaluator lag.
|
||||
An old decision with no subsequent relevant changes remains fresh while idle.
|
||||
Policy edits and UTC-window changes are explicit new evaluation obligations.
|
||||
|
||||
Durable pending rows are truth; notifications are hints. Reuse the existing
|
||||
delivery coordinator for local coalescing, finite retries, actual deadlines,
|
||||
drain and restart. No unconditional per-company timer scans empty queues.
|
||||
|
||||
The current DB work signals are process-local by default, with optional
|
||||
commit-bound PostgreSQL notifications. Separate reducers therefore require
|
||||
qualified cross-process wake delivery, subscribed before the startup scan and
|
||||
reconciled after reconnect. `NOTIFY` alone is not durable storage or a way to
|
||||
wake stopped compute. See the [existing signal implementation](../../packages/db/src/work-signals.ts)
|
||||
and [PostgreSQL notification behavior](https://www.postgresql.org/docs/current/sql-notify.html).
|
||||
Do not claim distributed recovery or host sleep until the actual notification,
|
||||
startup, disconnect and external-wake paths are qualified.
|
||||
|
||||
Receipt acceptance continues while budgets are blocked. Otherwise stopping spend
|
||||
would also stop the evidence needed to settle it.
|
||||
|
||||
## Failure cases the design must pass
|
||||
|
||||
| Event/race | Required outcome |
|
||||
| --- | --- |
|
||||
| Same receipt delivered twice | One logical receipt; equal fingerprint accepted as replay |
|
||||
| Same identity, different content | Conflict retained/reported; no silent overwrite or second charge |
|
||||
| Sequence 8 arrives before 7 | Correct cumulative state; 7 cannot rewind it |
|
||||
| Earlier allocated receipt commits after a batch | It remains discoverable as pending work |
|
||||
| Producer loses ACK after commit | Safe redelivery with the same identity |
|
||||
| Reducer dies before or during publication | All batch effects commit together or remain replayable |
|
||||
| Two reducers compute from generation 12 | One publishes generation 13; the other recomputes |
|
||||
| Final receipt races partial receipt | Closure and canonical reduction determine one contribution, independent of arrival order |
|
||||
| Final bill is lower than live estimate | Correct signed adjustment/replacement; no double counting |
|
||||
| Source changes during recovery | Verified handoff retains prior debt; no guessed new baseline |
|
||||
| Policy is lowered during evaluation | Old policy revision cannot publish a decision |
|
||||
| Manual Stop races budget resume | Manual execution hold remains authoritative |
|
||||
| Evaluator stops while receipts continue | Evidence accumulates durably; stale-view admission protection activates |
|
||||
| No new accounting work | Reducers and evaluator become quiet |
|
||||
|
||||
## Implementation and migration
|
||||
|
||||
These steps implement the full target after the first milestone above; they are
|
||||
not required as one change to improve today's eligibility path.
|
||||
|
||||
1. Add immutable operation identities and the journal acceptance boundary,
|
||||
initially alongside the existing authoritative path. Preserve existing native
|
||||
and legacy receipts; adapt source/closure evidence before removing their
|
||||
current completion fences.
|
||||
2. Implement deterministic operation reduction, idempotent settlement and
|
||||
partition publication. Run it in shadow mode with isolated projections and
|
||||
no production ledger, stop, or incident side effects. Compare at a common
|
||||
receipt/closure boundary after draining, not at arbitrary wall times.
|
||||
3. Implement versioned budget decisions, freshness, read-only admission and
|
||||
transition-based stop delivery. Replace shared monthly/runtime counters with
|
||||
partitioned projections and update all report, repair and integrity consumers.
|
||||
4. At a controlled cutover, fence the old settlement/evaluation authority, drain
|
||||
committed inputs, seed verified ledger/adjustment baselines, and select one
|
||||
authoritative writer path. Resume admission only with current decisions.
|
||||
Continue accepting/replaying durable receipts through this transition.
|
||||
5. Rollback similarly requires reconciliation and an authority handoff. Do not
|
||||
let old and new settlement writers charge the same operation independently.
|
||||
|
||||
Owning modules are usage receipts/native usage delivery, run-cost accounting,
|
||||
costs/reconciliation/integrity, budgets and the existing enforcement worker.
|
||||
Dispatch consumes a narrow decision contract; provider normalization remains
|
||||
with adapters. Update DB/shared/API/UI contracts and the budget guarantees in
|
||||
`doc/SPEC.md`, `doc/SPEC-implementation.md` and contributor guidance before enabling
|
||||
the change. The earlier per-run-lock design is not this target's hot path.
|
||||
|
||||
Qualification uses real PostgreSQL and multiple independent processes, both one
|
||||
large company and many companies, overlapping scope budgets, ledger-history
|
||||
growth, duplicate/reconnect bursts and the races above. Verify producer SQL has
|
||||
no shared counter/head/policy writes and no company/run `FOR UPDATE`. Audit
|
||||
foreign-key, trigger and uniqueness behavior too; string searches cannot prove
|
||||
absence of contention.
|
||||
|
||||
Measure sustained receipts and bytes per second, ACK/admission/settlement latency,
|
||||
projection backlog age, stop latency, database waits and event-loop load. Ten
|
||||
thousand agents at one receipt per second would create 864 million receipts per
|
||||
day before batching/retention effects. Eliminating serialization does not remove
|
||||
database, WAL, index, storage or provider-capacity limits. Start with bounded
|
||||
batches and measured retention; introduce a different log substrate only when
|
||||
its need is demonstrated.
|
||||
|
||||
Design validation in this chat: inspected the existing journal, budget incident,
|
||||
enforcement and work-signal contracts and checked PostgreSQL concurrency and
|
||||
notification semantics. This is an architecture document, not a prototype or
|
||||
load-test result. Implementation details and capacity remain to be qualified.
|
||||
@@ -19,6 +19,10 @@ same Runner and provider history through another healthy follow-up. This is a
|
||||
controlled cache fault, not a reconstruction of a historical incident.
|
||||
Each recovery case includes another healthy turn after the recovered turn and
|
||||
asserts that it does not start another runner or provider process.
|
||||
The `warm-suspended` case suspends the first turn, resumes the second turn,
|
||||
then keeps that controller warm for the third turn. It must retire the first
|
||||
turn's local suspension receipt so the third turn can reuse the resumed Runner
|
||||
and provider history without a second process or repeated turn.
|
||||
The same-server case also starts preparation lease renewal during recovery and
|
||||
holds instruction collection open past a renewal tick after coordinator release.
|
||||
The executor must acquire renewal before preparation releases its reference, then
|
||||
|
||||
@@ -101,13 +101,15 @@ function registerModuleMocks() {
|
||||
}));
|
||||
}
|
||||
|
||||
registerModuleMocks();
|
||||
const [{ agentRoutes }, { errorHandler }] = await Promise.all([
|
||||
vi.importActual<typeof import("../routes/agents.js")>("../routes/agents.js"),
|
||||
vi.importActual<typeof import("../middleware/index.js")>("../middleware/index.js"),
|
||||
]);
|
||||
|
||||
const refreshableAdapterType = "refreshable_adapter_route_test";
|
||||
|
||||
async function createApp() {
|
||||
const [{ agentRoutes }, { errorHandler }] = await Promise.all([
|
||||
vi.importActual<typeof import("../routes/agents.js")>("../routes/agents.js"),
|
||||
vi.importActual<typeof import("../middleware/index.js")>("../middleware/index.js"),
|
||||
]);
|
||||
const app = express();
|
||||
app.use(express.json());
|
||||
app.use((req, _res, next) => {
|
||||
@@ -159,11 +161,6 @@ async function unregisterTestAdapter(type: string) {
|
||||
|
||||
describe("adapter model refresh route", () => {
|
||||
beforeEach(async () => {
|
||||
vi.resetModules();
|
||||
vi.doUnmock("../routes/agents.js");
|
||||
vi.doUnmock("../routes/authz.js");
|
||||
vi.doUnmock("../middleware/index.js");
|
||||
registerModuleMocks();
|
||||
vi.clearAllMocks();
|
||||
mockCompanySkillService.listRuntimeSkillEntries.mockResolvedValue([]);
|
||||
mockCompanySkillService.resolveRequestedSkillKeys.mockResolvedValue([]);
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
|
||||
import { DELIVERY_QUEUES, subscribeDeliveryWork } from "../services/delivery-work-notifications.js";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { eq, sql } from "drizzle-orm";
|
||||
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import {
|
||||
agents,
|
||||
@@ -11,10 +11,14 @@ import {
|
||||
companies,
|
||||
costEvents,
|
||||
createDb,
|
||||
heartbeatRuns,
|
||||
issues,
|
||||
nativeRunFinalizations,
|
||||
projects,
|
||||
} from "@paperclipai/db";
|
||||
import { budgetService, deliverBudgetEnforcement, hasPendingBudgetEnforcement } from "../services/budgets.ts";
|
||||
import {
|
||||
EMBEDDED_POSTGRES_TEST_TIMEOUT_MS,
|
||||
getEmbeddedPostgresTestSupport,
|
||||
startEmbeddedPostgresTestDatabase,
|
||||
} from "./helpers/embedded-postgres.js";
|
||||
@@ -40,13 +44,16 @@ describeEmbeddedPostgres("budgetService release gate enforcement", () => {
|
||||
await import("../services/agent-lifecycle.js");
|
||||
if (!externalUrl) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-budgets-service-");
|
||||
db = createDb(externalUrl ?? tempDb!.connectionString);
|
||||
}, 20_000);
|
||||
}, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS);
|
||||
|
||||
afterEach(async () => {
|
||||
await db.delete(budgetIncidents);
|
||||
await db.delete(approvals);
|
||||
await db.delete(budgetPolicies);
|
||||
await db.delete(costEvents);
|
||||
await db.delete(nativeRunFinalizations);
|
||||
await db.delete(heartbeatRuns);
|
||||
await db.delete(issues);
|
||||
await db.delete(projects);
|
||||
await db.delete(agents);
|
||||
await db.delete(companies);
|
||||
@@ -118,6 +125,131 @@ describeEmbeddedPostgres("budgetService release gate enforcement", () => {
|
||||
return event!;
|
||||
}
|
||||
|
||||
it.each(["no policy", "healthy policy", "warning policy", "manual company pause", "manual project pause", "orphan budget pause", "recovering run"])(
|
||||
"checks %s while another accounting writer holds the company lock",
|
||||
async state => {
|
||||
const { companyId, agentId, projectId } = await createBudgetFixture();
|
||||
if (state === "healthy policy" || state === "warning policy" || state === "recovering run") {
|
||||
const [policy] = await db.insert(budgetPolicies).values({ companyId, scopeType: "company", scopeId: companyId,
|
||||
windowKind: "calendar_month_utc", amount: 100 }).returning();
|
||||
if (state !== "recovering run") await insertCostEvent({ companyId, agentId, costCents: state === "warning policy" ? 80 : 20 });
|
||||
if (state === "warning policy") {
|
||||
const now = new Date();
|
||||
await db.insert(budgetIncidents).values({ companyId, policyId: policy.id, scopeType: "company", scopeId: companyId,
|
||||
metric: "billed_cents", windowKind: "calendar_month_utc",
|
||||
windowStart: new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), 1)),
|
||||
windowEnd: new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth() + 1, 1)),
|
||||
thresholdType: "soft", amountLimit: 100, amountObserved: 80 });
|
||||
}
|
||||
}
|
||||
if (state === "recovering run") {
|
||||
const [issue] = await db.insert(issues).values({ companyId, projectId, title: "Recover accounting" }).returning();
|
||||
const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, scopeKind: "issue", issueId: issue.id,
|
||||
nativeIssueId: issue.id, runtimeMode: "native", status: "failed", finishedAt: new Date(), costAccountingPending: true }).returning();
|
||||
await db.insert(nativeRunFinalizations).values({ companyId, issueId: issue.id, runId: run.id, phase: "observed" });
|
||||
}
|
||||
if (state === "manual company pause" || state === "orphan budget pause") {
|
||||
await db.update(companies).set({ status: "paused", pauseReason: state === "orphan budget pause" ? "budget" : "manual" })
|
||||
.where(eq(companies.id, companyId));
|
||||
}
|
||||
if (state === "manual project pause") {
|
||||
await db.update(projects).set({ pausedAt: new Date(), pauseReason: "manual" }).where(eq(projects.id, projectId));
|
||||
}
|
||||
|
||||
const reader = createDb(externalUrl ?? tempDb!.connectionString, { maxConnections: 1 });
|
||||
let acquired!: () => void, release!: () => void;
|
||||
const lockAcquired = new Promise<void>(resolve => { acquired = resolve; });
|
||||
const holdLock = new Promise<void>(resolve => { release = resolve; });
|
||||
const writer = db.transaction(async tx => {
|
||||
await tx.select({ id: companies.id }).from(companies).where(eq(companies.id, companyId)).for("no key update");
|
||||
acquired();
|
||||
await holdLock;
|
||||
});
|
||||
try {
|
||||
await Promise.race([lockAcquired, writer]);
|
||||
await reader.execute(sql`set lock_timeout = '500ms'`);
|
||||
if (state === "recovering run") {
|
||||
await expect(budgetService(reader).getInvocationBlock(companyId, agentId, { projectId }))
|
||||
.rejects.toMatchObject({ cause: { code: "55P03" } });
|
||||
} else {
|
||||
const block = await budgetService(reader).getInvocationBlock(companyId, agentId, { projectId });
|
||||
if (state === "manual company pause") {
|
||||
expect(block).toMatchObject({ scopeType: "company", scopeId: companyId,
|
||||
reason: "Company is paused and cannot start new work." });
|
||||
} else if (state === "manual project pause") {
|
||||
expect(block).toMatchObject({ scopeType: "project", scopeId: projectId,
|
||||
reason: "Project is paused and cannot start new work." });
|
||||
} else if (state === "orphan budget pause") {
|
||||
expect(block).toMatchObject({ scopeType: "company", scopeId: companyId,
|
||||
reason: "Budget pause requires a policy or an explicit operator resume." });
|
||||
} else expect(block).toBeNull();
|
||||
if (state === "warning policy") {
|
||||
expect(await reader.select().from(budgetIncidents).where(eq(budgetIncidents.companyId, companyId)))
|
||||
.toMatchObject([{ thresholdType: "soft", status: "open", resolvedAt: null }]);
|
||||
}
|
||||
if (state === "no policy") {
|
||||
await expect(budgetService(reader).evaluateAdmission(companyId, agentId, { projectId }))
|
||||
.rejects.toMatchObject({ cause: { code: "55P03" } });
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
release();
|
||||
try { await writer; }
|
||||
finally { await reader.$client.end({ timeout: 1 }); }
|
||||
}
|
||||
if (state === "recovering run") {
|
||||
const cancelWorkForScope = vi.fn();
|
||||
expect(await budgetService(db, { cancelWorkForScope }).getInvocationBlock(companyId, agentId, { projectId }))
|
||||
.toMatchObject({ scopeType: "company", scopeId: companyId,
|
||||
reason: "New work must wait while a native run recovers its unfinished accounting." });
|
||||
expect(cancelWorkForScope).not.toHaveBeenCalled();
|
||||
expect((await db.select().from(companies).where(eq(companies.id, companyId)))[0])
|
||||
.toMatchObject({ status: "active", pauseReason: null });
|
||||
expect((await db.select().from(agents).where(eq(agents.id, agentId)))[0])
|
||||
.toMatchObject({ status: "active", pauseReason: null });
|
||||
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, companyId)))
|
||||
.toMatchObject([{ status: "failed", costAccountingPending: true }]);
|
||||
expect(await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.companyId, companyId)))
|
||||
.toMatchObject([{ phase: "observed", resultId: null }]);
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
it.each([true, false])("releases an agent budget hold when its policy is healthy (active: %s)", async isActive => {
|
||||
const { companyId, agentId } = await createBudgetFixture();
|
||||
await db.insert(budgetPolicies).values({ companyId, scopeType: "agent", scopeId: agentId,
|
||||
windowKind: "calendar_month_utc", amount: 100, isActive });
|
||||
await db.update(agents).set({ status: "paused", lifecycleState: "paused", pauseReason: "budget", pausedAt: new Date(), lifecycleHolds: ["budget"] })
|
||||
.where(eq(agents.id, agentId));
|
||||
|
||||
expect(await budgetService(db).getInvocationBlock(companyId, agentId)).toBeNull();
|
||||
expect((await db.select().from(agents).where(eq(agents.id, agentId)))[0]).toMatchObject({
|
||||
lifecycleState: "resuming", pauseReason: null, lifecycleHolds: [],
|
||||
});
|
||||
});
|
||||
|
||||
it.each(["hard", "soft"])("resolves an obsolete %s incident even when the scope no longer has a budget hold", async thresholdType => {
|
||||
const { companyId, agentId } = await createBudgetFixture();
|
||||
const [policy] = await db.insert(budgetPolicies).values({ companyId, scopeType: "agent", scopeId: agentId,
|
||||
windowKind: "calendar_month_utc", amount: 100 }).returning();
|
||||
const [incident] = await db.insert(budgetIncidents).values({ companyId, policyId: policy.id, scopeType: "agent", scopeId: agentId,
|
||||
metric: "billed_cents", windowKind: "calendar_month_utc", windowStart: new Date(0), windowEnd: new Date(1),
|
||||
thresholdType, amountLimit: 100, amountObserved: 100 }).returning();
|
||||
|
||||
expect(await budgetService(db).getInvocationBlock(companyId, agentId)).toBeNull();
|
||||
expect((await db.select().from(budgetIncidents).where(eq(budgetIncidents.id, incident.id)))[0])
|
||||
.toMatchObject({ status: "resolved", resolvedAt: expect.any(Date) });
|
||||
});
|
||||
|
||||
it("rejects another company's agent or project before allowing a no-policy invocation", async () => {
|
||||
const own = await createBudgetFixture();
|
||||
const other = await createBudgetFixture();
|
||||
const service = budgetService(db);
|
||||
await expect(service.getInvocationBlock(own.companyId, other.agentId)).rejects.toThrow("Budget scope not found");
|
||||
await expect(service.getInvocationBlock(own.companyId, own.agentId, { projectId: other.projectId }))
|
||||
.rejects.toThrow("Budget scope not found");
|
||||
});
|
||||
|
||||
it("raises one soft incident per window before hard-stopping and preserving safe local audit details", async () => {
|
||||
const { companyId, agentId } = await createBudgetFixture();
|
||||
const cancelWorkForScope = vi.fn().mockResolvedValue(undefined);
|
||||
|
||||
@@ -16,6 +16,7 @@ import { prepareNativeHeartbeatRun } from "../../services/native-runtime/prepare
|
||||
import { createIsolatedTestHome } from "./isolated-test-home.js";
|
||||
import { type NativeControllerIdentity } from "../../services/native-runtime/native-restart-recovery.js";
|
||||
import { waitForPendingRunFailureReports } from "../../services/run-failure-report.js";
|
||||
import { registerLocalRuntimeApiServer } from "../../runtime-api.js";
|
||||
|
||||
// An explicitly supplied disposable database also permits qualification on hosts
|
||||
// where native Postgres cannot allocate semaphores. Never fall back silently.
|
||||
@@ -106,6 +107,7 @@ export function nativeRestartFixture() {
|
||||
: new Date(),
|
||||
};
|
||||
server = createServer();
|
||||
registerLocalRuntimeApiServer(server);
|
||||
await new Promise<void>((resolveListen) =>
|
||||
server!.listen(0, "127.0.0.1", resolveListen),
|
||||
);
|
||||
|
||||
@@ -3,7 +3,7 @@ import { withDatabaseOperation } from "@paperclipai/db";
|
||||
|
||||
type BudgetOperation = "budget.accounting.lock_wait" | "budget.accounting.recover"
|
||||
| "budget.accounting.settlement" | "budget.accounting.transaction"
|
||||
| "budget.admission.evaluate" | "budget.admission.reserve"
|
||||
| "budget.admission.evaluate" | "budget.admission.inspect" | "budget.admission.reconcile" | "budget.admission.reserve"
|
||||
| "budget.cancellation.deliver" | "budget.enforcement.deliver"
|
||||
| "budget.enforcement.scope" | "budget.exposure.evaluate";
|
||||
|
||||
|
||||
@@ -34,7 +34,7 @@ import type {
|
||||
} from "@paperclipai/shared";
|
||||
import { HttpError, notFound, unprocessable } from "../errors.js";
|
||||
import { logger } from "../middleware/logger.js";
|
||||
import { withAccountingTransaction } from "./accounting-transaction.js";
|
||||
import { withAccountingReadSnapshot, withAccountingTransaction } from "./accounting-transaction.js";
|
||||
import { logActivity, type LogActivityInput, type ActivityPublication } from "./activity-log.js";
|
||||
import { computeObservedSpend, computeObservedSpends, type ObservedSpend } from "./budget-exposure.js";
|
||||
import { notifyDeliveryWork, DELIVERY_QUEUES } from "./delivery-work-notifications.js";
|
||||
@@ -198,11 +198,66 @@ export function budgetPoliciesForRun(companyId: string, agentId: string | null,
|
||||
);
|
||||
}
|
||||
|
||||
type AdmissionContext = { issueId?: string | null; projectId?: string | null };
|
||||
|
||||
async function readAdmissionScopes(db: Db, companyId: string, agentId: string | null, context?: AdmissionContext) {
|
||||
const scopes: Array<{ scopeType: BudgetScopeType; scopeId: string }> = [
|
||||
{ scopeType: "company", scopeId: companyId }, ...(agentId ? [{ scopeType: "agent" as const, scopeId: agentId }] : []),
|
||||
...(context?.projectId ? [{ scopeType: "project" as const, scopeId: context.projectId }] : []),
|
||||
];
|
||||
const records = await Promise.all(scopes.map(scope => resolveScopeRecord(db, scope.scopeType, scope.scopeId)));
|
||||
if (records.some(scope => scope.companyId !== companyId)) throw notFound("Budget scope not found");
|
||||
const policies = await db.select().from(budgetPolicies).where(and(eq(budgetPolicies.companyId, companyId),
|
||||
budgetPoliciesForRun(companyId, agentId, context?.projectId ?? null)));
|
||||
return { scopes, records, policies };
|
||||
}
|
||||
|
||||
function scopePauseBlock(scopeType: BudgetScopeType, scopeId: string, scope: ScopeRecord, hasPolicies: boolean) {
|
||||
if (scopeType !== "agent" && scope.paused && scope.pauseReason !== "budget") {
|
||||
return { scopeType, scopeId, scopeName: scope.name, reason: `${scopeType === "company" ? "Company" : "Project"} is paused and cannot start new work.` };
|
||||
}
|
||||
if (scope.paused && scope.pauseReason === "budget" && !hasPolicies) {
|
||||
return { scopeType, scopeId, scopeName: scope.name, reason: "Budget pause requires a policy or an explicit operator resume." };
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
async function inspectInvocationEligibility(db: Db, companyId: string, agentId: string | null, context?: AdmissionContext) {
|
||||
const { scopes, records, policies } = await readAdmissionScopes(db, companyId, agentId, context);
|
||||
if (policies.length > 0) {
|
||||
if (records.some(scope => scope.budgetHeld)) return { needsReconciliation: true } as const;
|
||||
const observations = await computeObservedSpends(db, policies);
|
||||
if (policies.some((policy, index) => observedBlocks(policy, observations[index])
|
||||
|| (policy.isActive && policy.hardStopEnabled && policy.amount > 0 && observations[index].recoveringRunCount > 0))) {
|
||||
return { needsReconciliation: true } as const;
|
||||
}
|
||||
const incidents = await db.select({ policyId: budgetIncidents.policyId, thresholdType: budgetIncidents.thresholdType })
|
||||
.from(budgetIncidents).where(and(eq(budgetIncidents.companyId, companyId),
|
||||
inArray(budgetIncidents.policyId, policies.map(policy => policy.id)), eq(budgetIncidents.status, "open")));
|
||||
const byId = new Map(policies.map((policy, index) => [policy.id, { policy, observed: observations[index] }]));
|
||||
if (incidents.some(incident => {
|
||||
const { policy, observed } = byId.get(incident.policyId)!;
|
||||
return incident.thresholdType === "hard" || shouldResolveSoftIncident(policy, observed);
|
||||
})) return { needsReconciliation: true } as const;
|
||||
}
|
||||
for (const [index, { scopeType, scopeId }] of scopes.entries()) {
|
||||
const block = scopePauseBlock(scopeType, scopeId, records[index],
|
||||
policies.some(policy => policy.scopeType === scopeType && policy.scopeId === scopeId));
|
||||
if (block) return { needsReconciliation: false, block } as const;
|
||||
}
|
||||
return { needsReconciliation: false, block: null } as const;
|
||||
}
|
||||
|
||||
function unpricedUsageBlocks(policy: PolicyRow, observed: ObservedSpend) {
|
||||
return policy.unpricedUsagePolicy !== "allow"
|
||||
&& (observed.unpricedEventCount > 0 || observed.unknownBrowserRunCount > 0);
|
||||
}
|
||||
|
||||
function shouldResolveSoftIncident(policy: PolicyRow, observed: ObservedSpend) {
|
||||
return !policy.isActive || !policy.notifyEnabled
|
||||
|| compareCents(observed.totalExact, Math.ceil(policy.amount * policy.warnPercent / 100)) < 0;
|
||||
}
|
||||
|
||||
function observedBlocks(policy: PolicyRow, observed: Awaited<ReturnType<typeof computeObservedSpend>>) {
|
||||
if (!policy.isActive || !policy.hardStopEnabled || policy.amount <= 0) return false;
|
||||
return compareCents(observed.totalExact, policy.amount) >= 0 || (observed.pendingRunCount - observed.recoveringRunCount) > 0 || unpricedUsageBlocks(policy, observed);
|
||||
@@ -654,18 +709,12 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
|
||||
// A closed UTC window, inactive policy, or budget raise releases only
|
||||
// budget-owned pauses. Manual pauses remain operator decisions.
|
||||
await resolveOpenIncidentsForPolicy(policy.id, policy.companyId, "approved", "budget_service", "hard");
|
||||
if (!policy.isActive || !policy.notifyEnabled || compareCents(observed.totalExact, Math.ceil(policy.amount * policy.warnPercent / 100)) < 0) {
|
||||
if (shouldResolveSoftIncident(policy, observed)) {
|
||||
await resolveOpenIncidentsForPolicy(policy.id, policy.companyId, "approved", "budget_service", "soft");
|
||||
}
|
||||
}
|
||||
// Agent status has its own admission gate and error code. Preserve a
|
||||
// manual pause without misclassifying it as a budget failure.
|
||||
if (scopeType !== "agent" && scope.paused && scope.pauseReason !== "budget") {
|
||||
return { scopeType, scopeId, scopeName: scope.name, reason: `${scopeType === "company" ? "Company" : "Project"} is paused and cannot start new work.` };
|
||||
}
|
||||
if (scope.paused && scope.pauseReason === "budget" && !policies.length) {
|
||||
return { scopeType, scopeId, scopeName: scope.name, reason: "Budget pause requires a policy or an explicit operator resume." };
|
||||
}
|
||||
const pauseBlock = scopePauseBlock(scopeType, scopeId, scope, policies.length > 0);
|
||||
if (pauseBlock) return pauseBlock;
|
||||
if (scope.budgetHeld && policies[0]) await resumeScopeFromBudget(policies[0]);
|
||||
if (recoveringPolicy) return { scopeType, scopeId, scopeName: scope.name,
|
||||
reason: "New work must wait while a native run recovers its unfinished accounting." };
|
||||
@@ -748,14 +797,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
|
||||
|
||||
async function evaluateAdmissionInTransaction(companyId: string, agentId: string | null,
|
||||
context?: { issueId?: string | null; projectId?: string | null }) {
|
||||
const scopes: Array<{ scopeType: BudgetScopeType; scopeId: string }> = [
|
||||
{ scopeType: "company", scopeId: companyId }, ...(agentId ? [{ scopeType: "agent" as const, scopeId: agentId }] : []),
|
||||
...(context?.projectId ? [{ scopeType: "project" as const, scopeId: context.projectId }] : []),
|
||||
];
|
||||
const records = await Promise.all(scopes.map(scope => resolveScopeRecord(db, scope.scopeType, scope.scopeId)));
|
||||
if (records.some(scope => scope.companyId !== companyId)) throw notFound("Budget scope not found");
|
||||
const policies = await db.select().from(budgetPolicies).where(and(eq(budgetPolicies.companyId, companyId),
|
||||
budgetPoliciesForRun(companyId, agentId, context?.projectId ?? null)));
|
||||
const { scopes, records, policies } = await readAdmissionScopes(db, companyId, agentId, context);
|
||||
const observations = await computeObservedSpends(db, policies);
|
||||
const reservationIndexes = policies.map((policy, i) => policy.isActive && policy.hardStopEnabled && policy.metric === "billed_cents" && policy.amount > 0 ? i : -1).filter(i => i >= 0);
|
||||
const admission = (block: Awaited<ReturnType<typeof reconcileScope>>) => ({ block,
|
||||
@@ -1188,8 +1230,13 @@ export function budgetService(db: Db, _hooks: BudgetServiceHooks = {}) {
|
||||
upsertPolicy: (companyId: string, input: BudgetPolicyUpsertInput, actorUserId: string | null) =>
|
||||
mutate(companyId, service => service.upsertPolicy(companyId, input, actorUserId)),
|
||||
evaluateCostEvent: (event: typeof costEvents.$inferSelect) => mutate(event.companyId, service => service.evaluateCostEvent(event)),
|
||||
getInvocationBlock: (companyId: string, agentId: string | null, context?: { issueId?: string | null; projectId?: string | null }) =>
|
||||
mutate(companyId, service => service.getInvocationBlock(companyId, agentId, context)),
|
||||
getInvocationBlock: async (companyId: string, agentId: string | null, context?: AdmissionContext) => {
|
||||
const eligibility = await withBudgetSpan("budget.admission.inspect", () =>
|
||||
withAccountingReadSnapshot(db, companyId, tx => inspectInvocationEligibility(tx, companyId, agentId, context)));
|
||||
if (!eligibility.needsReconciliation) return eligibility.block;
|
||||
return withBudgetSpan("budget.admission.reconcile", () =>
|
||||
mutate(companyId, service => service.getInvocationBlock(companyId, agentId, context)));
|
||||
},
|
||||
resolveIncident: (companyId: string, incidentId: string, input: BudgetIncidentResolutionInput, actorUserId: string) =>
|
||||
mutate(companyId, service => service.resolveIncident(companyId, incidentId, input, actorUserId)),
|
||||
};
|
||||
|
||||
+8
-5
@@ -150,12 +150,13 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", (
|
||||
expect(next.kind).toBe("bootstrap_incomplete");
|
||||
});
|
||||
|
||||
it.skipIf(!process.env.PAPERCLIP_TEST_RUNNER_CONTAINER).each(["warm", "hard", "stale-projection", "legacy", "changed", "same-server", "diskless", "diskless-per-turn", "diskless-per-turn-read-retry", "diskless-warm-suspended", "diskless-warm-suspended-read-retry", "diskless-live-inspection"])("preserves remote warm continuity across completed turns: %s", async (scenario) => {
|
||||
const hardRestart = !["warm", "same-server"].includes(scenario);
|
||||
it.skipIf(!process.env.PAPERCLIP_TEST_RUNNER_CONTAINER).each(["warm", "hard", "warm-suspended", "stale-projection", "legacy", "changed", "same-server", "diskless", "diskless-per-turn", "diskless-per-turn-read-retry", "diskless-warm-suspended", "diskless-warm-suspended-read-retry", "diskless-live-inspection"])("preserves remote warm continuity across completed turns: %s", async (scenario) => {
|
||||
const hardRestart = !["warm", "same-server", "warm-suspended"].includes(scenario);
|
||||
const diskless = scenario.startsWith("diskless");
|
||||
const perTurn = scenario.startsWith("diskless-per-turn");
|
||||
const warmSuspended = scenario.startsWith("diskless-warm-suspended");
|
||||
const warmSuspended = scenario === "warm-suspended" || scenario.startsWith("diskless-warm-suspended");
|
||||
const resumedAfterSuspension = perTurn || warmSuspended;
|
||||
const suspendedBeforeThird = perTurn || (diskless && warmSuspended);
|
||||
const container = process.env.PAPERCLIP_TEST_RUNNER_CONTAINER!;
|
||||
const tunnel = await createRunnerTestTlsTunnel({ port: Number(process.env.PAPERCLIP_TEST_RUNNER_PORT),
|
||||
key: process.env.PAPERCLIP_TEST_RUNNER_TLS_KEY!, cert: process.env.NODE_EXTRA_CA_CERTS! });
|
||||
@@ -310,6 +311,8 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", (
|
||||
};
|
||||
const first = hardRestart ? await executeInController(firstOptions) : await executeFirst();
|
||||
expect(first.exitCode).toBe(0);
|
||||
if (scenario === "warm-suspended") expect(await closeIdleWarmNativeSessionsForRestart())
|
||||
.toEqual({ closed: 1, busy: 0, failed: 0 });
|
||||
await finalizeNativeRun({ db: fixture.db, runId: fixture.runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true });
|
||||
const snapshot = await runner.execute({ command: "sh", args: ["-c", `cat '${remoteCwd}'/.paperclip-runtime/paperclip-runner/sessions/*/runner/runner-state.json`] });
|
||||
expect(snapshot.exitCode).toBe(0);
|
||||
@@ -476,9 +479,9 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", (
|
||||
const finalCalls = await runner.execute({ command: "cat", args: [`${remoteCwd}/provider-calls.log`] });
|
||||
expect(finalCalls.stdout.split("\n").filter(line => line === "turn/start")).toHaveLength(3);
|
||||
expect(finalCalls.stdout.split("\n").filter(line => line === "thread/start")).toHaveLength(1);
|
||||
expect(finalCalls.stdout.split("\n").filter(line => line === "process-start")).toHaveLength(providerStarts + (resumedAfterSuspension ? 1 : 0));
|
||||
expect(finalCalls.stdout.split("\n").filter(line => line === "process-start")).toHaveLength(providerStarts + (suspendedBeforeThird ? 1 : 0));
|
||||
const finalProcess = await runner.execute({ command: "sh", args: ["-c", `cat '${remoteCwd}'/.paperclip-runtime/paperclip-runner/sessions/*/runner/runner-process.identity`] });
|
||||
if (resumedAfterSuspension) expect(finalProcess.stdout).not.toBe(processAfter.stdout);
|
||||
if (suspendedBeforeThird) expect(finalProcess.stdout).not.toBe(processAfter.stdout);
|
||||
else expect(finalProcess.stdout).toBe(processAfter.stdout);
|
||||
expect(await closeIdleWarmNativeSessionsForRestart()).toEqual({ closed: perTurn ? 0 : 1, busy: 0, failed: 0 });
|
||||
const suspended = await runner.execute({ command: "sh", args: ["-c", `cat '${remoteCwd}'/.paperclip-runtime/paperclip-runner/sessions/*/runner/runner-state.json`] });
|
||||
|
||||
@@ -12967,9 +12967,10 @@ describe("runnerd provider runtime wiring", () => {
|
||||
},
|
||||
);
|
||||
|
||||
it.each((["codex", "claude"] as const).flatMap(provider => (["unchanged", "external-checkpoint", "changed", "runtime-context", "retirement-failed", "stale-owner", "active-owner", "missing-owner", "foreign-environment", "plugin-replaced", "plugin-managed-github"] as const).filter(scenario => provider === "codex" || scenario.startsWith("plugin-")).map(scenario => [provider, scenario] as const)))("rotates an owned warm %s remote runner before checking changed configuration: %s", async (provider, scenario) => {
|
||||
it.each((["codex", "claude"] as const).flatMap(provider => (["unchanged", "external-checkpoint", "suspension-receipt", "suspension-active-owner", "changed", "runtime-context", "retirement-failed", "stale-owner", "active-owner", "missing-owner", "foreign-environment", "plugin-replaced", "plugin-managed-github"] as const).filter(scenario => provider === "codex" || scenario.startsWith("plugin-")).map(scenario => [provider, scenario] as const)))("rotates an owned warm %s remote runner before checking changed configuration: %s", async (provider, scenario) => {
|
||||
const plugin = scenario === "plugin-replaced" || scenario === "plugin-managed-github";
|
||||
const reusable = ["unchanged", "external-checkpoint", "plugin-managed-github"].includes(scenario);
|
||||
const suspensionReceipt = scenario.startsWith("suspension-");
|
||||
const reusable = suspensionReceipt || ["unchanged", "external-checkpoint", "plugin-managed-github"].includes(scenario);
|
||||
const { permissionPolicy: _permissionPolicy, ...claudeProfile } = QUALIFIED_ACPX_PROFILES.claude;
|
||||
const first = { ...scopedNativeExecution(`warm-admission-${provider}-${scenario}`),
|
||||
schema: "paperclip.native-execution-input.v5" as const,
|
||||
@@ -12978,7 +12979,7 @@ describe("runnerd provider runtime wiring", () => {
|
||||
if (provider === "claude") first.session.driverKind = "acpx_runtime";
|
||||
first.session.lifecyclePolicy = { mode: "warm", idleTimeoutMs: 60_000 };
|
||||
const next = { ...first, binding: { ...first.binding, runId: `${first.binding.runId}-next` },
|
||||
session: { ...first.session, lifecyclePolicy: { mode: "warm" as const, idleTimeoutMs: ["unchanged", "external-checkpoint", "runtime-context", "stale-owner", "missing-owner", "plugin-replaced", "plugin-managed-github"].includes(scenario) ? 60_000 : 90_000 } } };
|
||||
session: { ...first.session, lifecyclePolicy: { mode: "warm" as const, idleTimeoutMs: suspensionReceipt || ["unchanged", "external-checkpoint", "runtime-context", "stale-owner", "missing-owner", "plugin-replaced", "plugin-managed-github"].includes(scenario) ? 60_000 : 90_000 } } };
|
||||
if (scenario === "runtime-context") {
|
||||
next.runtimeContext = { ...first.runtimeContext,
|
||||
mcp: { ...first.runtimeContext.mcp, digest: "1".repeat(64) } };
|
||||
@@ -13084,11 +13085,17 @@ describe("runnerd provider runtime wiring", () => {
|
||||
const ordinarySelect = nextDb.select.bind(nextDb);
|
||||
nextDb.select = ((fields?: Record<string, unknown>) => fields?.status === heartbeatRuns.status
|
||||
&& fields?.runnerProfileJson === heartbeatRuns.runnerProfileJson
|
||||
? priorNativeRunDb(scenario === "stale-owner" ? older : first, scenario === "active-owner" ? "running" : "succeeded").select(fields as never)
|
||||
? priorNativeRunDb(scenario === "stale-owner" ? older : first, ["active-owner", "suspension-active-owner"].includes(scenario) ? "running" : "succeeded").select(fields as never)
|
||||
: ordinarySelect(fields as never)) as Db["select"];
|
||||
try {
|
||||
await executePaperclipNativeSession({ db: firstDb, execution: first,
|
||||
runnerInstanceId: identity.runnerInstanceId, useRunnerd: true, runnerExecutionTarget: pluginTarget as never,...pluginOptions });
|
||||
if (suspensionReceipt) {
|
||||
// Remote suspension leaves a local identity receipt, while the Runner
|
||||
// owns its full state in the sandbox. A resumed turn must retire it.
|
||||
await writeFile(join(root, "plugin-runner-state.json"), JSON.stringify(durableRunnerState(identity, "suspended")));
|
||||
await rm(join(root, "runner/runner-state.json"));
|
||||
}
|
||||
if (scenario === "stale-owner") await writeRunnerAuthority(root, { ...identity, runId: older.binding.runId }, "ready");
|
||||
if (scenario === "missing-owner") {
|
||||
await closeWarmNativeSessionsForEnvironment({ environmentId, reason: "fixture owner loss" });
|
||||
@@ -13101,12 +13108,14 @@ describe("runnerd provider runtime wiring", () => {
|
||||
runnerInstanceId: `runner-next-${scenario}`, useRunnerd: true,
|
||||
runnerExecutionTarget: (scenario === "foreign-environment" ? { ...target, environmentId: `${environmentId}-other` }
|
||||
: plugin ? {...pluginTarget,leaseId:"next-core-lease"} : target) as never,...pluginOptions });
|
||||
if (["stale-owner", "active-owner", "missing-owner", "foreign-environment"].includes(scenario)) {
|
||||
await expect(continuation).rejects.toThrow(scenario === "active-owner"
|
||||
if (["stale-owner", "active-owner", "suspension-active-owner", "missing-owner", "foreign-environment"].includes(scenario)) {
|
||||
await expect(continuation).rejects.toThrow(["active-owner", "suspension-active-owner"].includes(scenario)
|
||||
? "prior_owner_active" : "prior_owner_terminal_state_indeterminate");
|
||||
expect(state.execute).toHaveBeenCalledOnce();
|
||||
expect(firstSession.close).not.toHaveBeenCalled();
|
||||
expect(order).toEqual([]);
|
||||
if (suspensionReceipt) expect(JSON.parse(await readFile(join(root, "plugin-runner-state.json"), "utf8")))
|
||||
.toMatchObject({ ...identity, lifecycle: "suspended" });
|
||||
} else if (scenario === "retirement-failed") {
|
||||
await expect(continuation).rejects.toThrow("retirement unconfirmed");
|
||||
expect(state.execute).toHaveBeenCalledOnce();
|
||||
@@ -13120,8 +13129,9 @@ describe("runnerd provider runtime wiring", () => {
|
||||
}
|
||||
if (scenario === "plugin-managed-github") expect(rebind).toHaveBeenCalledOnce();
|
||||
expect(order).toEqual(reusable ? ["execute-next"] : ["retire", "execute-next"]);
|
||||
if (["unchanged", "external-checkpoint"].includes(scenario)) {
|
||||
if (["unchanged", "external-checkpoint", "suspension-receipt"].includes(scenario)) {
|
||||
if (!persistedController) await writeRunnerAuthority(root, { ...identity, runId: next.binding.runId }, "ready");
|
||||
if (suspensionReceipt) await rm(join(root, "runner/runner-state.json"));
|
||||
const third = { ...next, binding: { ...next.binding, runId: `${next.binding.runId}-third` } };
|
||||
const thirdDb = leaseDb(third);
|
||||
const thirdSelect = thirdDb.select.bind(thirdDb);
|
||||
@@ -13139,6 +13149,11 @@ describe("runnerd provider runtime wiring", () => {
|
||||
runnerExecutionTarget: pluginTarget as never,...pluginOptions })).resolves.toMatchObject({ exitCode: 0 });
|
||||
expect(firstSession.close).not.toHaveBeenCalled();
|
||||
expect(state.execute).toHaveBeenCalledTimes(3);
|
||||
if (suspensionReceipt) {
|
||||
expect(existsSync(join(root, "plugin-runner-state.json"))).toBe(false);
|
||||
expect(JSON.parse(await readFile(join(root, "control-plane/control-plane-state.json"), "utf8")).identity)
|
||||
.toMatchObject({ ...identity, runId: next.binding.runId });
|
||||
}
|
||||
if (restore) {
|
||||
expect(restore.mock.calls.map(([, binding]) => binding.runId)).toEqual([first.binding.runId, next.binding.runId]);
|
||||
expect(existsSync(root)).toBe(true);
|
||||
|
||||
@@ -9844,7 +9844,8 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
);
|
||||
},
|
||||
onSession: async (session) => {
|
||||
if (session && usesPluginTask) {
|
||||
if (session && input.runnerExecutionTarget?.kind === "remote"
|
||||
&& input.runnerExecutionTarget.transport === "sandbox") {
|
||||
rmSync(resolve(scopedRunnerdStateRoot(input.execution), "plugin-runner-state.json"), { force: true });
|
||||
}
|
||||
if (session && session === existingWarmSession) remoteWarmRecoveryPending = false;
|
||||
|
||||
Reference in new issue
Block a user