mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-12 22:56:58 +01:00
fix(admission): admit new work independent of other runs' cleanup and accounting (#19944)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Budget admission decides if new agent work can start in a company > - Five admission gates waited on the cleanup or accounting of other runs > - The restart-chaos campaign found that these gates froze companies for up to 10 minutes, caused false budget pauses, denied in-run writes, and dropped wakes, comments and Retries > - The durable ownership plan (Decision 1) removes the company-wide accounting gate. It does not narrow it again > - This pull request removes the five gates. Unfinished accounting now adds its bounded estimate to exposure > - The benefit is that one run's unsettled accounting never stops the other work in its company ## Linked Issues or Issue Description Refs #19831 (plan, section D), #19838 (inventory, finding 3), #19927 (Phase 2 reconciler data), #19497, #19724, #19826. **What happened?** Admission refused new work in these cases, with no reached hard stop and no pause: - a terminal run awaited accounting (`waitingRunCount`); - a native run recovered its accounting (`recoveringRunCount`); - a budget-owned scope hold awaited reconciliation, after the hard stop no longer applied; - a budget input was older than 15 s (`budget_evaluation_work`); - the predecessor run row of a task was missing (`hasPendingRunCleanup` failed closed). **Expected behavior** Admission never waits on the cleanup or accounting of another run. Only a reached hard stop, an explicit pause or an inactive company refuses new work (invariant 3). **Steps to reproduce** Make a company with an active budget policy. Leave one failed run with `cost_accounting_pending = true`. Request work for another agent. On `master` admission refuses with "Company cannot start work while completed runs await accounting." ## What Changed - `budgets.ts`: admission refuses only for a reached hard stop or an explicit pause. The code removes `observedBlocks`, the `recoveringRunCount` check, the stale `budgetHeld` refusal and the 15 s input-age gate. It also removes the unused `inspectInvocationEligibility`. `reconcileScope` no longer returns accounting holds. - `budget-exposure.ts`: removes the `waiting` and `recovering` aggregates and the own-task subquery. The admission query is smaller. Unfinished accounting stays exposure: the larger of the held reservation and the observed cost. - `budget-evaluation-work.ts`: removes `oldestBudgetInputAt`, which nothing reads now. - `execution-blocker.ts`: `hasPendingRunCleanup` returns false for a missing run row. A run row that is deleted removes its finalization coordinator and nulls its lease references, so the row owns nothing. A predecessor that owns a lease or a coordinator still holds the successor of its own task. - `recovery/ownership-invariant-report.ts`: the I3 check no longer mirrors the removed gate. It reports a wake request or a run that was refused with an accounting-hold reason. Any such report is a regression. - `doc/SPEC-implementation.md`: adds the product rule "Admission is independent of cleanup and accounting". `doc/execution-semantics.md` is aligned. ## Verification - New `server/src/__tests__/admission-independent-of-cleanup.test.ts`, on embedded Postgres. It has one case per gate. All six cases fail on `master` and pass here. A seventh case (an explicit pause still refuses) passes on both. One case also asserts invariant 3 together with exposure: two failed runs that each hold a 400-cent reservation fill a 1000-cent budget, and a hard stop still refuses. - Existing tests that asserted the old gates now assert admission (invariants 3 and 4): `closed-native-receipt-accounting`, `budgets-service`, `budget-evaluation-work`, `heartbeat-held-comment-wake`, `ownership-reconciliation-report`. In `heartbeat-held-comment-wake`, each saved comment, Retry, assignment and continuation is delivered exactly once while another run's accounting is pending. Settling accounting later does not deliver it again. - All six files pass locally, one file per process. The server package typecheck passes. - Speed: admission is on the turn path. This change only removes work from it: two correlated aggregates, one own-task subquery and one `budget_evaluation_work` read per policy. No wait or I/O is added before `turn.start`. - Not run here: the chaos waves (K13 storm, K6 crash loop, K4 long holds). The coordinating session schedules them on the shared harness. ## Risks - Behavior change: work starts while accounting for other runs settles. Spend stays bounded by the reservation capacity check and the hard stop, which read the ledger plus observed cost. A run that has no reservation and no observed cost adds zero exposure until its charge commits. This is the gap that Decision 1 accepts. Budget enforcement was already best effort (SPEC 11.6). - A budget-owned pause with no reached hard stop no longer refuses work in budget admission. The paused company or agent is still refused by the existing pause checks until the budget worker lifts the pause. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, with tool use and code execution in Claude Code. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
DottaandClaude Opus 5.5 authored and GitHub committed
1 parent
7213f64962
commit
f6cb3e43f4
18 files changed
+415
-359
No files matched your search
@@ -3,9 +3,10 @@ import { budgetPolicies, type Db } from "@paperclipai/db";
|
||||
import { policyBlocks } from "./budgets.js";
|
||||
|
||||
/** A reached hard stop (spend at the limit, or blocked unpriced usage) is a
|
||||
* board decision. Any other budget block (completed runs awaiting accounting,
|
||||
* a native run recovering its accounting, a hold awaiting reconciliation)
|
||||
* clears by itself. Read only: the admission check already reconciled. */
|
||||
* board decision. Any other budget block (a scope paused by budget with no
|
||||
* policy left to judge it) is an explicit pause that clears when it is lifted.
|
||||
* Unfinished accounting never blocks admission. Read only: the admission check
|
||||
* already reconciled. */
|
||||
export async function isBoardOwnedBudgetBlock(
|
||||
db: Db,
|
||||
companyId: string,
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import { and, eq, or, sql } from "drizzle-orm";
|
||||
import { sql } from "drizzle-orm";
|
||||
import { budgetEvaluationWork, type Db } from "@paperclipai/db";
|
||||
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
|
||||
|
||||
@@ -18,13 +18,3 @@ export async function hasPendingBudgetEvaluation(db: Db): Promise<boolean> {
|
||||
const [row] = await db.select({ id: budgetEvaluationWork.id }).from(budgetEvaluationWork).limit(1);
|
||||
return Boolean(row);
|
||||
}
|
||||
|
||||
/** A completed idle decision does not age. Only unconsumed relevant inputs do. */
|
||||
export async function oldestBudgetInputAt(db: Db, input: BudgetEvaluationInput): Promise<Date | null> {
|
||||
const [row] = await db.select({ oldest: sql<Date | null>`min(${budgetEvaluationWork.createdAt})` }).from(budgetEvaluationWork)
|
||||
.where(and(eq(budgetEvaluationWork.companyId, input.companyId), !input.agentId && !input.projectId ? undefined : or(
|
||||
input.agentId ? eq(budgetEvaluationWork.agentId, input.agentId) : undefined,
|
||||
input.projectId ? eq(budgetEvaluationWork.projectId, input.projectId) : undefined,
|
||||
)));
|
||||
return row?.oldest ? new Date(row.oldest) : null;
|
||||
}
|
||||
@@ -1,6 +1,6 @@
|
||||
import { addCents, normalizeCents, usdToUnits, unitsToCents } from "@paperclipai/shared";
|
||||
import { and, eq, gte, inArray, lt, or, sql, type SQL } from "drizzle-orm";
|
||||
import { budgetPolicies, budgetReservations, browserUseRuns, browserUseSessions, costEvents, heartbeatRuns, issues, nativeRunFinalizations, type Db } from "@paperclipai/db";
|
||||
import { budgetPolicies, budgetReservations, browserUseRuns, browserUseSessions, costEvents, heartbeatRuns, issues, type Db } from "@paperclipai/db";
|
||||
import { withBudgetSpan } from "./budget-tracing.js";
|
||||
|
||||
export interface BudgetExposure {
|
||||
@@ -16,10 +16,9 @@ export interface BudgetExposure {
|
||||
export type SpendPolicy = Pick<typeof budgetPolicies.$inferSelect, "companyId" | "scopeType" | "scopeId" | "windowKind" | "metric">;
|
||||
export type ObservedSpend = BudgetExposure & {
|
||||
windowStart: Date; windowEnd: Date; total: number; totalExact: string;
|
||||
pendingRunCount: number; recoveringRunCount: number;
|
||||
/** Pending runs that hold admission: they ended without succeeding, so
|
||||
* their provider may not have stopped, and no coordinator is recovering them. */
|
||||
waitingRunCount: number;
|
||||
/** Terminal runs whose accounting is unfinished. Display only: each one
|
||||
* contributes its bounded estimate to exposure and never gates admission. */
|
||||
pendingRunCount: number;
|
||||
};
|
||||
|
||||
export function budgetWindow(kind: string, now = new Date()) {
|
||||
@@ -73,7 +72,7 @@ const liveCost = sql`case when ${heartbeatRuns.costAccountedAt} is null and ${he
|
||||
|
||||
/** Batched ledger, held-operation and unfinished-run aggregates. Never scans
|
||||
* agents or receipt history; each policy retains its own scope and window. */
|
||||
async function readObservedSpends(db: Db, policies: SpendPolicy[], now: Date, excludeRunId?: string | null): Promise<ObservedSpend[]> {
|
||||
async function readObservedSpends(db: Db, policies: SpendPolicy[], now: Date): Promise<ObservedSpend[]> {
|
||||
const windows = policies.map(policy => budgetWindow(policy.windowKind, now));
|
||||
const billable = policies.map((policy, index) => ({ policy, index })).filter(({ policy }) => policy.metric === "billed_cents");
|
||||
const ledgerColumns: Record<string, SQL> = {}, heldColumns: Record<string, SQL> = {}, runColumns: Record<string, SQL> = {}, browserColumns: Record<string, SQL> = {};
|
||||
@@ -120,30 +119,12 @@ async function readObservedSpends(db: Db, policies: SpendPolicy[], now: Date, ex
|
||||
const run = and(eq(heartbeatRuns.companyId, policy.companyId), or(eq(heartbeatRuns.costAccountingPending, true), lateReceiptDebt),
|
||||
policy.scopeType === "agent" ? eq(heartbeatRuns.agentId, policy.scopeId) : undefined,
|
||||
policy.scopeType === "project" ? sql`${usage}->'ledgerScope'->>'projectId' = ${policy.scopeId}` : undefined)!;
|
||||
// A task never waits on its own unfinished accounting: the run being
|
||||
// admitted and earlier runs of the same task (its recovery generation) are
|
||||
// what its continuation and finalization settle. They still count as spend.
|
||||
// coalesce keeps this a definite boolean when the admitted run has no task.
|
||||
const ownTask = excludeRunId ? sql`coalesce(${heartbeatRuns.id} = ${excludeRunId}::uuid or (${heartbeatRuns.nativeIssueId} is not null
|
||||
and ${heartbeatRuns.nativeIssueId} = (select admitted_run.native_issue_id from ${heartbeatRuns} admitted_run
|
||||
where admitted_run.id = ${excludeRunId}::uuid and admitted_run.company_id = ${heartbeatRuns.companyId})), false)` : sql`false`;
|
||||
const terminal = and(run, inArray(heartbeatRuns.status, ["succeeded", "failed", "timed_out", "cancelled", "interrupted"]), or(inWindow, lateReceiptDebt),
|
||||
sql`not ${ownTask}`)!;
|
||||
const terminal = and(run, inArray(heartbeatRuns.status, ["succeeded", "failed", "timed_out", "cancelled", "interrupted"]), or(inWindow, lateReceiptDebt))!;
|
||||
// Keep correlated subqueries in nested SQL chunks. Drizzle strips column
|
||||
// qualifiers from a single-table selection's direct column chunks.
|
||||
const hasReservation = sql`exists (select 1 from ${budgetReservations}
|
||||
where ${budgetReservations.companyId} = ${heartbeatRuns.companyId}
|
||||
and ${budgetReservations.runId} = ${heartbeatRuns.id} and ${budgetReservations.state} = 'held')`;
|
||||
// A terminal failure whose final receipt never closed is still recovering
|
||||
// accounting: once its provider stop is verified, accounting settles the
|
||||
// known part as unpriced. That window holds new work; it is not a hard stop.
|
||||
const closingReceipt = sql`${usage}->>'accountingReceiptReady' = 'false'
|
||||
and ${usage}->>'accountingCaptureFailed' is distinct from 'true'`;
|
||||
const recovering = sql`not (${lateReceiptDebt}) and exists (select 1 from ${nativeRunFinalizations}
|
||||
where ${nativeRunFinalizations.runId} = ${heartbeatRuns.id}
|
||||
and ${nativeRunFinalizations.companyId} = ${heartbeatRuns.companyId}
|
||||
and ${nativeRunFinalizations.resultId} is null
|
||||
and (${nativeRunFinalizations.phase} <> 'terminal_failure' or (${closingReceipt})))`;
|
||||
ledgerPredicates.push(ledger); heldPredicates.push(held); runPredicates.push(run);
|
||||
ledgerColumns[`total${index}`] = sql`coalesce(sum(${costEvents.costCents}) filter (where ${ledger}), 0)::text`;
|
||||
// "Usage with no reliable price" means observed usage that could not be
|
||||
@@ -192,16 +173,10 @@ async function readObservedSpends(db: Db, policies: SpendPolicy[], now: Date, ex
|
||||
// observation describe one operation, so count their maximum, not both.
|
||||
heldColumns[`held${index}`] = sql`coalesce(sum(greatest(${budgetReservations.amountCents}, ${liveCost})) filter (where ${held}), 0)::text`;
|
||||
runColumns[`pending${index}`] = sql`count(*) filter (where ${terminal})::int`;
|
||||
runColumns[`recovering${index}`] = sql`count(*) filter (where ${terminal} and ${heartbeatRuns.runtimeMode} = 'native' and ${recovering})::int`;
|
||||
// A succeeded run whose provider reported observed usage has only its
|
||||
// final usage report outstanding. Like a running run, that observed cost
|
||||
// counts toward spend and reservation capacity, and its settled charge can
|
||||
// still reach the hard stop, but it does not hold the whole scope's new
|
||||
// work. A succeeded run with no observed cost (for example an unproven
|
||||
// Cursor Cloud terminal) still holds new work.
|
||||
const boundedSucceeded = sql`${heartbeatRuns.status} = 'succeeded' and ${usage}->'budgetExposure'->>'costCents' is not null`;
|
||||
runColumns[`waiting${index}`] = sql`count(*) filter (where ${terminal} and not (${boundedSucceeded})
|
||||
and not (${heartbeatRuns.runtimeMode} = 'native' and ${recovering}))::int`;
|
||||
// Unfinished accounting is exposure, not a gate (SPEC 11.6, invariant 3).
|
||||
// Its bounded estimate is the larger of the run's held reservation and its
|
||||
// observed cost, counted above; a hard stop reads the ledger plus observed
|
||||
// cost. No terminal run's cleanup or accounting state refuses new work.
|
||||
}
|
||||
// A single statement preserves hold-to-ledger atomicity even when a caller
|
||||
// owns an otherwise READ COMMITTED transaction. Alias dynamic columns before
|
||||
@@ -233,8 +208,7 @@ async function readObservedSpends(db: Db, policies: SpendPolicy[], now: Date, ex
|
||||
return { windowStart: windows[index].start, windowEnd: windows[index].end,
|
||||
ledgerCents, liveObservedCents, effectiveHeldCents: addCents(String(held[`held${index}`] ?? 0), String(runs[`unreserved${index}`] ?? 0), String(browser[`held${index}`] ?? 0)),
|
||||
total: Number(totalExact), totalExact, pendingRunCount, pendingTerminalRunCount: pendingRunCount,
|
||||
unpricedEventCount: Number(ledger[`unpriced${index}`] ?? 0), unknownBrowserRunCount: Number(browser[`unknown${index}`] ?? 0), recoveringRunCount: Number(runs[`recovering${index}`] ?? 0),
|
||||
waitingRunCount: Number(runs[`waiting${index}`] ?? 0) };
|
||||
unpricedEventCount: Number(ledger[`unpriced${index}`] ?? 0), unknownBrowserRunCount: Number(browser[`unknown${index}`] ?? 0) };
|
||||
});
|
||||
}
|
||||
|
||||
@@ -242,14 +216,14 @@ export async function computeObservedSpend(db: Db, policy: SpendPolicy, now = ne
|
||||
return (await computeObservedSpends(db, [policy], now))[0];
|
||||
}
|
||||
|
||||
export function computeObservedSpends(db: Db, policies: SpendPolicy[], now = new Date(), excludeRunId?: string | null): Promise<ObservedSpend[]> {
|
||||
export function computeObservedSpends(db: Db, policies: SpendPolicy[], now = new Date()): Promise<ObservedSpend[]> {
|
||||
return withBudgetSpan("budget.exposure.evaluate", async () => {
|
||||
// Each policy contributes several aggregate columns. Bound statement width
|
||||
// below PostgreSQL's target-list limit. Multi-scope evaluators/reports own
|
||||
// a repeatable-read transaction; individual admission batches fit one SQL.
|
||||
const result: ObservedSpend[] = [];
|
||||
for (let offset = 0; offset < policies.length; offset += 128)
|
||||
result.push(...await readObservedSpends(db, policies.slice(offset, offset + 128), now, excludeRunId));
|
||||
result.push(...await readObservedSpends(db, policies.slice(offset, offset + 128), now));
|
||||
return result;
|
||||
}, { "budget.policy_count": policies.length });
|
||||
}
|
||||
@@ -42,7 +42,7 @@ import { computeObservedSpend, computeObservedSpends, type ObservedSpend } from
|
||||
import { notifyDeliveryWork, DELIVERY_QUEUES } from "./delivery-work-notifications.js";
|
||||
import { listPendingBudgetCancellationScopes } from "./heartbeat/budget-cancellation.js";
|
||||
import { withBudgetSpan } from "./budget-tracing.js";
|
||||
import { queueBudgetEvaluation, oldestBudgetInputAt } from "./budget-evaluation-work.js";
|
||||
import { queueBudgetEvaluation } from "./budget-evaluation-work.js";
|
||||
export { computeObservedSpend, computeObservedSpends } from "./budget-exposure.js";
|
||||
export type { BudgetExposure } from "./budget-exposure.js";
|
||||
|
||||
@@ -204,11 +204,9 @@ export function budgetPoliciesForRun(companyId: string, agentId: string | null,
|
||||
/** `runId` names the run being admitted, so its own pending accounting never holds it. */
|
||||
type AdmissionContext = { issueId?: string | null; projectId?: string | null; runId?: string | null;
|
||||
/** The caller continues work that was already admitted: a validated
|
||||
* recovery owner, or a running run's own tool mutations. Temporary holds
|
||||
* (accounting that awaits settlement, recovering runs, budget inputs that
|
||||
* await evaluation) gate new work only; admitted work must be able to
|
||||
* finish, or concurrent recoveries after a restart storm wait on each other.
|
||||
* Hard stops and pauses still apply. */
|
||||
* recovery owner, or a running run's own tool mutations. It reuses its
|
||||
* original reservation, so other runs' holds filling the budget never
|
||||
* refuse it. Hard stops and pauses still apply. */
|
||||
admittedRecovery?: boolean };
|
||||
|
||||
async function readAdmissionScopes(db: Db, companyId: string, agentId: string | null, context?: AdmissionContext) {
|
||||
@@ -233,32 +231,6 @@ function scopePauseBlock(scopeType: BudgetScopeType, scopeId: string, scope: Sco
|
||||
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, undefined, context?.runId);
|
||||
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);
|
||||
@@ -276,15 +248,6 @@ function hardStopReached(policy: PolicyRow, observed: Awaited<ReturnType<typeof
|
||||
return compareCents(observed.totalExact, policy.amount) >= 0 || unpricedUsageBlocks(policy, observed);
|
||||
}
|
||||
|
||||
/** Admission also waits while runs that ended without succeeding await
|
||||
* accounting. That unknown is a separate conservative restriction: it holds
|
||||
* new work only. A run that finishes moments before its charge commits must
|
||||
* not pause the company or cancel unrelated running work. A succeeded run with
|
||||
* only usage outstanding is bounded like a running run and does not wait. */
|
||||
function observedBlocks(policy: PolicyRow, observed: Awaited<ReturnType<typeof computeObservedSpend>>) {
|
||||
if (!policy.isActive || !policy.hardStopEnabled || policy.amount <= 0) return false;
|
||||
return hardStopReached(policy, observed) || observed.waitingRunCount > 0;
|
||||
}
|
||||
export async function policyBlocks(db: Db, policy: PolicyRow) {
|
||||
if (!policy.isActive || !policy.hardStopEnabled || policy.amount <= 0) return false;
|
||||
return hardStopReached(policy, await computeObservedSpend(db, policy));
|
||||
@@ -310,11 +273,10 @@ export async function recheckAdmittedRunBudgetReadOnly(db: Db, input: {
|
||||
const policies = await db.select().from(budgetPolicies).where(and(eq(budgetPolicies.companyId, input.companyId),
|
||||
eq(budgetPolicies.isActive, true), eq(budgetPolicies.hardStopEnabled, true), eq(budgetPolicies.metric, "billed_cents"),
|
||||
sql`${budgetPolicies.amount} > 0`, budgetPoliciesForRun(input.companyId, input.agentId, input.projectId)));
|
||||
const observations = await computeObservedSpends(db, policies, undefined, input.runId);
|
||||
const observations = await computeObservedSpends(db, policies);
|
||||
for (const [index, policy] of policies.entries()) {
|
||||
const observed = observations[index];
|
||||
if (input.admittedRecovery ? hardStopReached(policy, observed) : observedBlocks(policy, observed))
|
||||
return "Applicable budget no longer permits provider work";
|
||||
if (hardStopReached(policy, observed)) return "Applicable budget no longer permits provider work";
|
||||
// An owned recovery reuses the hold reserveRunBudget already admitted;
|
||||
// other runs' holds filling the budget must not refuse it.
|
||||
if (input.admittedRecovery) continue;
|
||||
@@ -717,8 +679,6 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
|
||||
const policies = prepared?.policies ?? await db.select().from(budgetPolicies).where(and(
|
||||
eq(budgetPolicies.companyId, companyId), eq(budgetPolicies.scopeType, scopeType), eq(budgetPolicies.scopeId, scopeId),
|
||||
));
|
||||
let recoveringPolicy: PolicyRow | undefined;
|
||||
let accountingHold = false;
|
||||
const observations = prepared?.observations ?? await computeObservedSpends(db, policies, now);
|
||||
const title = scopeType === "company" ? "Company" : scopeType === "agent" ? "Agent" : "Project";
|
||||
for (const [index, policy] of policies.entries()) {
|
||||
@@ -726,9 +686,6 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
|
||||
if (hardStopReached(policy, observed)) {
|
||||
await createIncidentIfNeeded(policy, "hard", observed);
|
||||
await pauseAndCancelScopeForBudget(policy, !scope.budgetHeld);
|
||||
if (observed.pendingRunCount > 0) {
|
||||
return { scopeType, scopeId, scopeName: scope.name, reason: `${title} cannot start work while completed runs await accounting.` };
|
||||
}
|
||||
if (unpricedUsageBlocks(policy, observed)) {
|
||||
return { scopeType, scopeId, scopeName: scope.name, reason: `${title} cannot start work because recorded usage has no reliable price.` };
|
||||
}
|
||||
@@ -736,12 +693,6 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
|
||||
? "Project cannot start work because its budget hard-stop is still exceeded."
|
||||
: `${title} is paused because its budget hard-stop was reached.` };
|
||||
}
|
||||
// Pending accounting holds new work only. It never keeps a budget-owned
|
||||
// pause or a hard incident, including one opened by older releases.
|
||||
if (observedBlocks(policy, observed)) accountingHold = true;
|
||||
// An open native coordinator must finish the original run's accounting.
|
||||
// Deny fresh work without pausing/cancelling that recovery or its hold.
|
||||
if (policy.isActive && policy.hardStopEnabled && policy.amount > 0 && observed.recoveringRunCount > 0) recoveringPolicy ??= policy;
|
||||
// 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");
|
||||
@@ -752,10 +703,6 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
|
||||
const pauseBlock = scopePauseBlock(scopeType, scopeId, scope, policies.length > 0);
|
||||
if (pauseBlock) return pauseBlock;
|
||||
if (scope.budgetHeld && policies[0]) await resumeScopeFromBudget(policies[0]);
|
||||
if (accountingHold) return { scopeType, scopeId, scopeName: scope.name,
|
||||
reason: `${title} cannot start work while completed runs await accounting.` };
|
||||
if (recoveringPolicy) return { scopeType, scopeId, scopeName: scope.name,
|
||||
reason: "New work must wait while a native run recovers its unfinished accounting." };
|
||||
return null;
|
||||
}
|
||||
|
||||
@@ -838,7 +785,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
|
||||
async function evaluateAdmissionInTransaction(companyId: string, agentId: string | null,
|
||||
context?: AdmissionContext) {
|
||||
const { scopes, records, policies } = await readAdmissionScopes(db, companyId, agentId, context);
|
||||
const observations = await computeObservedSpends(db, policies, now, context?.runId);
|
||||
const observations = await computeObservedSpends(db, policies, now);
|
||||
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,
|
||||
policies: reservationIndexes.map(i => policies[i]), observations: reservationIndexes.map(i => observations[i]) });
|
||||
@@ -848,29 +795,17 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
|
||||
.filter(({ policy }) => policy.scopeType === scopeType && policy.scopeId === scopeId);
|
||||
const pause = scopePauseBlock(scopeType, scopeId, scope, scoped.length > 0);
|
||||
if (pause) return admission(pause);
|
||||
// Admission is independent of cleanup (SPEC 11.6, invariant 3): only a
|
||||
// reached hard stop or an explicit pause refuses work. Unfinished
|
||||
// accounting adds its bounded estimate to exposure, never a gate, and a
|
||||
// budget-owned scope hold with no reached hard stop is stale: the budget
|
||||
// worker releases it, admission does not wait for that.
|
||||
for (const { policy, observed } of scoped) {
|
||||
const title = scopeType === "company" ? "Company" : scopeType === "agent" ? "Agent" : "Project";
|
||||
if (context?.admittedRecovery ? hardStopReached(policy, observed) : observedBlocks(policy, observed)) return admission({ scopeType, scopeId, scopeName: scope.name,
|
||||
reason: observed.waitingRunCount > 0 ? `${title} cannot start work while completed runs await accounting.`
|
||||
: unpricedUsageBlocks(policy, observed) ? `${title} cannot start work because recorded usage has no reliable price.`
|
||||
if (hardStopReached(policy, observed)) return admission({ scopeType, scopeId, scopeName: scope.name,
|
||||
reason: unpricedUsageBlocks(policy, observed) ? `${title} cannot start work because recorded usage has no reliable price.`
|
||||
: scopeType === "project" ? "Project cannot start work because its budget hard-stop is still exceeded."
|
||||
: `${title} is paused because its budget hard-stop was reached.` });
|
||||
if (!context?.admittedRecovery && policy.isActive && policy.hardStopEnabled && policy.amount > 0 && observed.recoveringRunCount > 0)
|
||||
return admission({ scopeType, scopeId, scopeName: scope.name,
|
||||
reason: "New work must wait while a native run recovers its unfinished accounting." });
|
||||
}
|
||||
if (scope.budgetHeld) return admission({ scopeType, scopeId, scopeName: scope.name,
|
||||
reason: "Budget hold is awaiting reconciliation." });
|
||||
}
|
||||
if (reservationIndexes.length > 0 && !context?.admittedRecovery) {
|
||||
for (const index of reservationIndexes) {
|
||||
const policy = policies[index];
|
||||
const oldest = await oldestBudgetInputAt(db, { companyId,
|
||||
agentId: policy.scopeType === "agent" ? policy.scopeId : null,
|
||||
projectId: policy.scopeType === "project" ? policy.scopeId : null });
|
||||
if (oldest && now.getTime() - oldest.getTime() > 15_000) return admission({ scopeType: policy.scopeType as BudgetScopeType,
|
||||
scopeId: policy.scopeId, scopeName: records.find((_, i) => scopes[i].scopeType === policy.scopeType)?.name ?? "Budget scope",
|
||||
reason: "Budget accounting changes are awaiting reconciliation." });
|
||||
}
|
||||
}
|
||||
return admission(null);
|
||||
|
||||
@@ -75,7 +75,11 @@ export async function hasPendingRunCleanup(db: Db, companyId: string, runId: str
|
||||
eq(environmentLeases.companyId, companyId), eq(environmentLeases.heartbeatRunId, runId),
|
||||
sql`${environmentLeases.metadata}->'workspaceRecoveryCleanup' is not null`,
|
||||
))).filter(hasRecoveredCleanupStopReceipt).map(lease => lease.id);
|
||||
const [state] = await db.execute<{ pending: boolean }>(sql`SELECT
|
||||
// A missing predecessor row owns nothing: deleting a run nulls its leases'
|
||||
// run reference and removes its finalization coordinator. Failing closed on
|
||||
// absence stranded the successor with no next action (invariants 3 and 7).
|
||||
const [state] = await db.execute<{ pending: boolean }>(sql`SELECT EXISTS (SELECT 1 FROM ${heartbeatRuns}
|
||||
WHERE ${heartbeatRuns.companyId} = ${companyId} AND ${heartbeatRuns.id} = ${runId} AND (
|
||||
${heartbeatRuns.runnerProfileJson}->>'nativeWorkspaceFinalizationOwner' IS NOT NULL
|
||||
OR EXISTS (SELECT 1 FROM ${environmentLeases}
|
||||
WHERE ${environmentLeases.companyId} = ${companyId} AND ${environmentLeases.heartbeatRunId} = ${runId}
|
||||
@@ -93,13 +97,12 @@ export async function hasPendingRunCleanup(db: Db, companyId: string, runId: str
|
||||
OR (${environmentLeases.provider} = 'ssh' AND ${environmentLeases.metadata}->>'driver' = 'ssh'
|
||||
AND ${environmentLeases.providerLeaseId} LIKE 'ssh://%')))))
|
||||
OR (${environmentLeases.status} = 'retained' AND ${environmentLeases.leasePolicy} = 'reuse_by_environment'
|
||||
AND ${environmentLeases.cleanupStatus} = 'success'), false))))
|
||||
AND ${environmentLeases.cleanupStatus} = 'success'), false))))))
|
||||
OR EXISTS (SELECT 1 FROM ${nativeRunFinalizations}
|
||||
WHERE ${nativeRunFinalizations.companyId} = ${companyId} AND ${nativeRunFinalizations.runId} = ${runId}
|
||||
AND (${nativeRunFinalizations.leaseOwner} IS NOT NULL
|
||||
OR ${nativeRunFinalizations.phase} NOT IN ('committed', 'applied', 'terminal_failure'))) AS pending
|
||||
FROM ${heartbeatRuns} WHERE ${heartbeatRuns.companyId} = ${companyId} AND ${heartbeatRuns.id} = ${runId} LIMIT 1`);
|
||||
return state?.pending !== false;
|
||||
OR ${nativeRunFinalizations.phase} NOT IN ('committed', 'applied', 'terminal_failure'))) AS pending`);
|
||||
return state?.pending === true;
|
||||
}
|
||||
|
||||
/** Resolved recovery bookkeeping can still carry an effective no-replay hold. */
|
||||
|
||||
@@ -8363,7 +8363,10 @@ describe("native terminal-turn accounting", () => {
|
||||
const nextRun = randomUUID();
|
||||
await db.insert(tables.heartbeatRuns).values({ id: nextRun, companyId, agentId, status: "queued", runtimeMode: "native" });
|
||||
await expect(reserveRunBudget(db, companyId, nextRun, null)).resolves.toMatchObject({ reused: false });
|
||||
} else expect(block, scenario).not.toBeNull();
|
||||
// Only a settled unpriced charge under a blocking policy refuses work.
|
||||
// Pending accounting is exposure, never a gate (invariant 3).
|
||||
} else if (ready) expect(block, scenario).not.toBeNull();
|
||||
else expect(block, scenario).toBeNull();
|
||||
}
|
||||
} finally { await db.$client.end(); await database.cleanup(); await rm(directory, { recursive: true, force: true }); }
|
||||
}, 90_000);
|
||||
|
||||
@@ -1,9 +1,7 @@
|
||||
import { and, asc, eq, inArray, isNotNull, isNull, lt, notExists, or, sql } from "drizzle-orm";
|
||||
import {
|
||||
agentWakeupRequests,
|
||||
budgetPolicies,
|
||||
heartbeatRuns,
|
||||
nativeRunFinalizations,
|
||||
issueRecoveryActions,
|
||||
issues,
|
||||
type Db,
|
||||
@@ -95,7 +93,8 @@ const COVERAGE_GAPS: Record<string, string> = {
|
||||
I6_generation_fencing: "Runner events carry no generation stamp until section G lands.",
|
||||
I8_fast_turns: "Measured per run by the speed-budget check in the run log, not by this pass.",
|
||||
I7_scope: "I7 measures agent-assigned in_progress issues. todo, in_review and blocked issues are not measured yet.",
|
||||
record_cap: "Each kind, and the I3 and I7 checks, read at most 2,000 records per pass; a larger kind is reported as truncated.",
|
||||
record_cap: "Each kind and the I7 check read at most 2,000 records per pass; a larger kind is reported as truncated. "
|
||||
+ "The I3 and I4 scans read ten pages per pass and continue from a cursor on the next pass.",
|
||||
wake_coverage: "Writers of workspace runtime services, remote workspace operations, snapshots, restore points, "
|
||||
+ "git cache intents, provider artifacts and harness backups publish no work signal. Those kinds are read at "
|
||||
+ "startup and on any other report wake. Phase 2 adds a signal when each kind moves to the shared settle path.",
|
||||
@@ -114,16 +113,20 @@ export type OwnershipReportState = {
|
||||
unwritten: Map<string, { violation: InvariantViolation; attempts: number }>;
|
||||
/** True when the I4 scan stopped with more rows to read. */
|
||||
intentBacklog: boolean;
|
||||
/** Keyset cursors of the I3 refusal scans (wake requests and runs). */
|
||||
admissionWakeCursor: { at: string; id: string } | null;
|
||||
admissionRunCursor: { at: string; id: string } | null;
|
||||
/** True when an I3 scan stopped with more rows to read. */
|
||||
admissionBacklog: boolean;
|
||||
};
|
||||
|
||||
export function createOwnershipReportState(): OwnershipReportState {
|
||||
return { goneLastPass: new Set(), reported: new Set(), intentCursor: null, unwritten: new Map(), intentBacklog: false };
|
||||
return { goneLastPass: new Set(), reported: new Set(), intentCursor: null, unwritten: new Map(), intentBacklog: false,
|
||||
admissionWakeCursor: null, admissionRunCursor: null, admissionBacklog: false };
|
||||
}
|
||||
|
||||
const TERMINAL_RUN_STATUSES = ["succeeded", "failed", "cancelled", "timed_out", "interrupted"];
|
||||
const ACTIVE_RUN_STATUSES = ["queued", "running", "scheduled_retry"];
|
||||
/** Accounting closure deadline used by the admission gate (run-cost-accounting.ts). */
|
||||
const ACCOUNTING_CLOSURE_MS = 10 * 60_000;
|
||||
/** An in-progress issue must show a next action within this time. */
|
||||
const NEXT_ACTION_GRACE_MS = 5 * 60_000;
|
||||
/** Records read per kind (and per I3/I7 check) per pass; beyond this the
|
||||
@@ -256,43 +259,55 @@ async function settleKind(db: Db, kindId: string, options: {
|
||||
|
||||
const violationKey = (violation: InvariantViolation) => `${violation.rule}:${violation.kind}:${violation.recordId}`;
|
||||
|
||||
/** Admission still waits on accounting (inventory finding 3). This mirrors
|
||||
* the `waitingRunCount` gate of budget-exposure.ts, including late-receipt
|
||||
* debt: a terminal run with
|
||||
* pending accounting holds its company's new work unless it is a succeeded
|
||||
* run with observed cost, or a native run whose coordinator is still
|
||||
* recovering. Only companies with an active budget policy have the gate, and
|
||||
* only runs past the accounting closure deadline are reported. */
|
||||
async function checkAdmissionIndependence(db: Db, now: Date, limit: number): Promise<InvariantViolation[]> {
|
||||
const usage = heartbeatRuns.usageJson;
|
||||
const lateReceiptDebt = sql`coalesce(${usage}->>'accountingLateReceiptPending', 'false') = 'true'`;
|
||||
const boundedSucceeded = sql`(${heartbeatRuns.status} = 'succeeded' and ${usage}->'budgetExposure'->>'costCents' is not null)`;
|
||||
const closingReceipt = sql`${usage}->>'accountingReceiptReady' = 'false' and ${usage}->>'accountingCaptureFailed' is distinct from 'true'`;
|
||||
const recovering = sql`(${heartbeatRuns.runtimeMode} = 'native' and not (${lateReceiptDebt}) and exists (select 1 from ${nativeRunFinalizations}
|
||||
where ${nativeRunFinalizations.runId} = ${heartbeatRuns.id} and ${nativeRunFinalizations.companyId} = ${heartbeatRuns.companyId}
|
||||
and ${nativeRunFinalizations.resultId} is null
|
||||
and (${nativeRunFinalizations.phase} <> 'terminal_failure' or (${closingReceipt}))))`;
|
||||
const gated = sql`exists (select 1 from ${budgetPolicies} where ${budgetPolicies.companyId} = ${heartbeatRuns.companyId}
|
||||
and ${budgetPolicies.isActive} = true
|
||||
and (${budgetPolicies.scopeType} = 'company'
|
||||
or (${budgetPolicies.scopeType} = 'agent' and ${budgetPolicies.scopeId} = ${heartbeatRuns.agentId})
|
||||
or (${budgetPolicies.scopeType} = 'project' and ${usage}->'ledgerScope'->>'projectId' = ${budgetPolicies.scopeId}::text))
|
||||
and (${budgetPolicies.windowKind} <> 'calendar_month_utc' or (${lateReceiptDebt})
|
||||
or ${heartbeatRuns.finishedAt} >= (date_trunc('month', ${now.toISOString()}::timestamptz at time zone 'UTC') at time zone 'UTC')))`;
|
||||
/** Refusal texts that name unfinished accounting or cleanup as the reason
|
||||
* work could not start, as a PostgreSQL case-insensitive regular expression.
|
||||
* Phase 3 removed every such gate, so any match is a regression. */
|
||||
const ACCOUNTING_HOLD_PATTERN = "await(s|ing)? accounting|unfinished accounting|awaiting reconciliation";
|
||||
|
||||
/** Invariant 3: admission never waits on another run's cleanup or
|
||||
* accounting. Phase 3 removed the five gates that did (inventory finding 3),
|
||||
* so this reads the outcome instead of mirroring a gate: a wake request or a
|
||||
* run refused within the lookback window with an accounting-hold reason. */
|
||||
async function checkAdmissionIndependence(db: Db, state: OwnershipReportState, now: Date, limit: number): Promise<InvariantViolation[]> {
|
||||
const start = { at: new Date(now.getTime() - INTENT_LOOKBACK_MS).toISOString(), id: "00000000-0000-0000-0000-000000000000" };
|
||||
const wakeText = sql`(coalesce(${agentWakeupRequests.error}, '') || ' '
|
||||
|| coalesce(${agentWakeupRequests.payload}->'executionWait'->>'message', ''))`;
|
||||
const runText = sql`(coalesce(${heartbeatRuns.error}, '') || ' ' || coalesce(${heartbeatRuns.resultJson}->>'admissionBlockReason', ''))`;
|
||||
const violations: InvariantViolation[] = [];
|
||||
for (let offset = 0; offset < MAX_RECORDS_PER_KIND; offset += limit) {
|
||||
const rows = await db.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, issueId: heartbeatRuns.issueId })
|
||||
.from(heartbeatRuns).where(and(or(eq(heartbeatRuns.costAccountingPending, true), lateReceiptDebt),
|
||||
inArray(heartbeatRuns.status, TERMINAL_RUN_STATUSES),
|
||||
lt(heartbeatRuns.finishedAt, new Date(now.getTime() - ACCOUNTING_CLOSURE_MS)),
|
||||
sql`not ${boundedSucceeded}`, sql`not ${recovering}`, gated))
|
||||
.orderBy(asc(heartbeatRuns.id)).limit(limit).offset(offset);
|
||||
for (const row of rows) {
|
||||
violations.push({ rule: "I3_admission_independent", kind: "run_accounting_debt", recordId: row.id,
|
||||
companyId: row.companyId, runId: row.id, issueId: row.issueId, reason: "accounting_debt_holds_admission" });
|
||||
}
|
||||
// Each table pages by (updated_at, id) from its own cursor, as rule I4
|
||||
// does, so a busy window is read in full across passes.
|
||||
let backlog = false;
|
||||
let position = state.admissionWakeCursor ?? start;
|
||||
for (let page = 0; page < 10; page++) {
|
||||
const rows = await db.select({ id: agentWakeupRequests.id, companyId: agentWakeupRequests.companyId, runId: agentWakeupRequests.runId,
|
||||
issueId: sql<string | null>`${agentWakeupRequests.payload}->>'issueId'`, updatedAt: sql<string>`${agentWakeupRequests.updatedAt}::text` })
|
||||
.from(agentWakeupRequests).where(and(
|
||||
sql`(${agentWakeupRequests.updatedAt}, ${agentWakeupRequests.id}) > (${position.at}::timestamptz, ${position.id}::uuid)`,
|
||||
sql`${wakeText} ~* ${ACCOUNTING_HOLD_PATTERN}`))
|
||||
.orderBy(asc(agentWakeupRequests.updatedAt), asc(agentWakeupRequests.id)).limit(limit);
|
||||
for (const row of rows) violations.push({ rule: "I3_admission_independent", kind: "wake_request", recordId: row.id,
|
||||
companyId: row.companyId, runId: row.runId, issueId: row.issueId, reason: "refused_for_accounting_hold" });
|
||||
if (rows.length > 0) position = { at: rows[rows.length - 1]!.updatedAt, id: rows[rows.length - 1]!.id };
|
||||
if (rows.length < limit) break;
|
||||
if (page === 9) backlog = true;
|
||||
}
|
||||
state.admissionWakeCursor = position;
|
||||
position = state.admissionRunCursor ?? start;
|
||||
for (let page = 0; page < 10; page++) {
|
||||
const rows = await db.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, issueId: heartbeatRuns.issueId,
|
||||
updatedAt: sql<string>`${heartbeatRuns.updatedAt}::text` })
|
||||
.from(heartbeatRuns).where(and(
|
||||
sql`(${heartbeatRuns.updatedAt}, ${heartbeatRuns.id}) > (${position.at}::timestamptz, ${position.id}::uuid)`,
|
||||
inArray(heartbeatRuns.status, ["cancelled", "failed"]), sql`${runText} ~* ${ACCOUNTING_HOLD_PATTERN}`))
|
||||
.orderBy(asc(heartbeatRuns.updatedAt), asc(heartbeatRuns.id)).limit(limit);
|
||||
for (const row of rows) violations.push({ rule: "I3_admission_independent", kind: "queued_run", recordId: row.id,
|
||||
companyId: row.companyId, runId: row.id, issueId: row.issueId, reason: "refused_for_accounting_hold" });
|
||||
if (rows.length > 0) position = { at: rows[rows.length - 1]!.updatedAt, id: rows[rows.length - 1]!.id };
|
||||
if (rows.length < limit) break;
|
||||
if (page === 9) backlog = true;
|
||||
}
|
||||
state.admissionRunCursor = position;
|
||||
state.admissionBacklog = backlog;
|
||||
return violations;
|
||||
}
|
||||
|
||||
@@ -444,7 +459,7 @@ export async function runOwnershipReconciliationPass(db: Db, options: {
|
||||
|
||||
let nextDueAt: Date | null = null;
|
||||
for (const check of [
|
||||
() => checkAdmissionIndependence(db, now, limit),
|
||||
() => checkAdmissionIndependence(db, state, now, limit),
|
||||
() => checkExactlyOnceIntents(db, state, now, limit),
|
||||
async () => {
|
||||
const result = await checkVisibleNextAction(db, now, limit);
|
||||
@@ -462,7 +477,7 @@ export async function runOwnershipReconciliationPass(db: Db, options: {
|
||||
|
||||
// Only a first proof needs a second pass; a record already reported for I2
|
||||
// is not a reason to keep scanning.
|
||||
if (state.intentBacklog) retryNeeded = true;
|
||||
if (state.intentBacklog || state.admissionBacklog) retryNeeded = true;
|
||||
const awaitingSecondProof = [...goneThisPass].filter(key => !state.goneLastPass.has(key)).length;
|
||||
state.goneLastPass = goneThisPass;
|
||||
// Violations whose activity row failed last time are written again, even
|
||||
|
||||
Reference in new issue
Block a user