From f6cb3e43f4b16eab72093a06bf4070be10116ee8 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Sat, 7 Nov 2026 21:46:39 -0600 Subject: [PATCH] fix(admission): admit new work independent of other runs' cleanup and accounting (#19944) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - 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 --- doc/SPEC-implementation.md | 51 ++++--- doc/execution-semantics.md | 4 +- .../admission-independent-of-cleanup.test.ts | 128 ++++++++++++++++++ .../budget-enforcement-scoping.test.ts | 12 +- .../__tests__/budget-evaluation-work.test.ts | 21 ++- server/src/__tests__/budgets-service.test.ts | 13 +- .../chat-channels.integration.test.ts | 7 +- .../closed-native-receipt-accounting.test.ts | 52 +++---- .../heartbeat-held-comment-wake.test.ts | 107 +++++---------- .../heartbeat-refused-assignment-wake.test.ts | 38 ++++-- .../ownership-reconciliation-report.test.ts | 60 +++++--- server/src/services/budget-block-ownership.ts | 7 +- server/src/services/budget-evaluation-work.ts | 12 +- server/src/services/budget-exposure.ts | 52 ++----- server/src/services/budgets.ts | 93 ++----------- server/src/services/execution-blocker.ts | 13 +- .../native-session-executor.test.ts | 5 +- .../recovery/ownership-invariant-report.ts | 99 ++++++++------ 18 files changed, 415 insertions(+), 359 deletions(-) create mode 100644 server/src/__tests__/admission-independent-of-cleanup.test.ts diff --git a/doc/SPEC-implementation.md b/doc/SPEC-implementation.md index 637f509072b..12ebce5d86d 100644 --- a/doc/SPEC-implementation.md +++ b/doc/SPEC-implementation.md @@ -1761,26 +1761,37 @@ Financial changes append independent durable evaluation inputs. The existing worker coalesces notifications over one second and publishes budget holds, incidents and durable Stop intent from a consistent snapshot. Policy/control publication is fenced separately from settlement. Inputs arriving or committing -later stay pending; no global sequence watermark consumes them. Limited-budget -admission refuses relevant unevaluated inputs older than fifteen seconds. -Unchanged idle decisions do not expire; unknown accounting retains its separate -conservative admission restriction. Runs that ended without succeeding and -await accounting hold new work only: they never open a hard-stop incident, pause -a scope, or stop running work. Only spend at the limit, or unpriced usage under -a blocking policy, does that. A succeeded run with only its usage report -outstanding, whose provider already reported observed usage, does not hold new -work: its provider work is over, so it is bounded like a running run. A -succeeded run with no observed cost still holds new work. Its observed cost counts toward spend and reservation -capacity, and its settled charge can still reach the hard stop. A run that -failed, was cancelled, timed out or was interrupted still holds new work, -because its provider may not have stopped. Temporary holds (accounting that awaits -settlement, recovering runs, budget inputs that await evaluation) gate new work -only. They never refuse an admitted run's own tool mutations or an owned -recovery; a hard stop or pause still does, through the normal pause and -cancellation path. Provider reporting and cancellation delays -remain additional to evaluation lag. Cancellation delivery is asynchronous and -scoped; committed per-run Stop survives policy changes and restart until existing -stop proof succeeds. Budget resume cannot clear a manual hold. +later stay pending; no global sequence watermark consumes them. Admission does +not wait for them: it reads the ledger and live exposure itself, so a reached +hard stop refuses new work even while evaluation lags. + +**Admission is independent of cleanup and accounting.** Admitting new work never +waits on another run's cleanup or accounting completeness. Only these refuse it: + +- spend at the limit (ledger plus observed running spend), or unpriced usage + under a blocking policy (the hard stop); +- an explicit pause of the company, project or agent; +- an inactive company. + +A run whose accounting is unfinished (its final receipt has not closed, its +native coordinator is still recovering, its usage capture failed, or a late +receipt is outstanding) contributes its bounded estimate to exposure: the +larger of its held reservation and its observed cost. It never creates a +company-wide or scope-wide gate, and it never pauses a scope or opens a +hard-stop incident. A budget-owned scope hold whose policies no longer reach a +hard stop is stale; the budget worker releases it, and budget admission does +not refuse work while it waits. Unevaluated budget inputs are not a gate. +Provider reporting and cancellation delays remain additional to evaluation lag. +Cancellation delivery is asynchronous and scoped; committed per-run Stop +survives policy changes and restart until existing stop proof succeeds. Budget +resume cannot clear a manual hold. + +A successor run on the same task still waits while its predecessor owns +resources that the successor would reuse (a live environment lease, an open +native finalization coordinator, or workspace finalization). That wait is +per task, ends when the owner releases them, and wakes the successor through +the work-intent promotion path. A predecessor run row that no longer exists +owns nothing and never holds its successor. Accepted cumulative per-run usage checkpoints contribute observed running spend before the final ledger charge. Admission counts each unfinished operation's diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 770b3be5b1b..474bdd9aec6 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -510,9 +510,9 @@ and the user need not certify infrastructure state in a reconciliation form. Repeated automatic signals for an unchanged gate share one durable skipped-wake diagnostic, scoped to company, agent, issue, gate code, and condition identity. The diagnostic retains the first request and counts later observations. This applies to execution reconciliation, dependencies, pause holds, company and agent availability, budget blocks, and disabled heartbeats. These diagnostics do not consume provider attempts and are never proof that a future wake was delivered. All current gates are checked again on the next wake, including the periodic dependency reconciliation sweep. Clearing one gate does not bypass another. -A new task whose first assignment wake is refused by a temporary budget hold (completed runs awaiting accounting, a native run recovering its accounting, or a hold awaiting reconciliation) stays `todo`. Stranded-work recovery does not move a task with no run to `blocked` for such a hold. The task shows the hold it waits for, and the assigned-task liveness sweep admits it once the hold clears, through normal admission and with one run. A reached hard stop or blocked unpriced usage remains a board decision. Removed assignees and cancelled tasks still do not wake. +Another run's unfinished accounting or cleanup never refuses new work (see `doc/SPEC-implementation.md`, "Admission is independent of cleanup and accounting"). A new task whose first assignment wake is refused by a temporary condition (an explicit pause) stays `todo`. Stranded-work recovery does not move a task with no run to `blocked` for such a condition. The task shows what it waits for, and the assigned-task liveness sweep admits it once the condition clears, through normal admission and with one run. A reached hard stop or blocked unpriced usage remains a board decision. Removed assignees and cancelled tasks still do not wake. -The same holds for accepted work beyond a first assignment. A temporary budget hold never terminally cancels or escalates accepted work: an unstarted queued run (a promoted comment, a Retry, or a recovery) stays queued, and stranded-work recovery leaves the task as it is instead of moving it to `blocked`. These waits are passive, so settling a run's accounting wakes queued-run dispatch, the saved-receipt worker, and stranded-work recovery for the company. The held work then starts through normal admission, once. When a native run reaches a terminal status through recovery, or releases its environment, each message saved behind it is hinted by its own receipt, so delivery does not wait out a company-wide rescan interval. +The same holds for accepted work beyond a first assignment. A temporary refusal never terminally cancels or escalates accepted work: an unstarted queued run (a promoted comment, a Retry, or a recovery) stays queued, and stranded-work recovery leaves the task as it is instead of moving it to `blocked`. These waits are passive, so settling a run's accounting wakes queued-run dispatch, the saved-receipt worker, and stranded-work recovery for the company. The held work then starts through normal admission, once. When a native run reaches a terminal status through recovery, or releases its environment, each message saved behind it is hinted by its own receipt, so delivery does not wait out a company-wide rescan interval. New comments received during an execution hold retain their individual deferred receipts and ordered comment ids. Release cannot drain those receipts while replay remains blocked; the next eligible wake can adopt them. Authorized external-chat requests also remain deferred with their exact durable receipt. They must use normal promotion and current authorization; a generic wake cannot adopt only their comment ids and discard their actor or session contract. A wait does not authorize replay, reset an incident retry budget, or bypass an interaction's delivery rules. diff --git a/server/src/__tests__/admission-independent-of-cleanup.test.ts b/server/src/__tests__/admission-independent-of-cleanup.test.ts new file mode 100644 index 00000000000..4395b36d848 --- /dev/null +++ b/server/src/__tests__/admission-independent-of-cleanup.test.ts @@ -0,0 +1,128 @@ +import { randomUUID } from "node:crypto"; +import { eq, sql } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; +import { + agents, budgetEvaluationWork, budgetPolicies, budgetReservations, companies, costEvents, createDb, heartbeatRuns, issues, + nativeRunFinalizations, type Db, +} from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { budgetService } from "../services/budgets.js"; +import { reserveRunBudget } from "../services/budget-reservations.js"; +import { hasPendingRunCleanup } from "../services/execution-blocker.js"; + +/** + * Phase 3, section D (invariant 3): admitting new work never waits on another + * run's cleanup or accounting completeness. Only a reached hard stop, an + * explicit pause or an inactive company refuses it. Each case below was one of + * the five gates of inventory finding 3; every one refused on master. + */ +const support = await getEmbeddedPostgresTestSupport(); +(support.supported ? describe : describe.skip)("admission independent of cleanup and accounting", () => { + let database: Awaited>; + let db: Db; + beforeAll(async () => { + database = await startEmbeddedPostgresTestDatabase("admission-independent-"); + db = createDb(database.connectionString); + }, 60_000); + afterEach(async () => { await db.transaction(async tx => { + await tx.execute(sql`SET LOCAL client_min_messages = warning`); + await tx.execute(sql`TRUNCATE companies CASCADE`); + }); }); + afterAll(async () => { await database?.cleanup(); }); + + async function fixture(limit = 1_000, reservationCents = "0") { + const companyId = randomUUID(), agentId = randomUUID(), otherAgentId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Admission lab", issuePrefix: `A${companyId.slice(0, 5)}` }); + await db.insert(agents).values([ + { id: agentId, companyId, name: "Worker", role: "engineer", status: "idle", adapterType: "paperclip_runner" }, + { id: otherAgentId, companyId, name: "Other", role: "engineer", status: "idle", adapterType: "paperclip_runner" }, + ]); + await db.insert(budgetPolicies).values({ companyId, scopeType: "company", scopeId: companyId, metric: "billed_cents", + windowKind: "calendar_month_utc", amount: limit, reservationCents, notifyEnabled: false, unpricedUsagePolicy: "allow" }); + const [issue] = await db.insert(issues).values({ companyId, title: "Earlier task", status: "in_progress", + assigneeAgentId: agentId }).returning(); + const block = () => budgetService(db).getInvocationBlock(companyId, otherAgentId, { projectId: null }); + /** A run of the earlier task that ended without settling its accounting. */ + const pendingRun = async (status: "failed" | "succeeded", options: { recovering?: boolean; reservedCents?: string } = {}) => { + const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, invocationSource: "automation", scopeKind: "issue", + issueId: issue.id, nativeIssueId: issue.id, runtimeMode: "native", status, finishedAt: new Date(), + costAccountingPending: true, usageJson: { accountingReceiptReady: false, billingType: "metered_api", + ledgerScope: { issueId: issue.id, projectId: null } } }).returning(); + if (options.recovering) await db.insert(nativeRunFinalizations).values({ companyId, issueId: issue.id, runId: run.id, + phase: "observed", attempt: 1 }); + await db.insert(budgetReservations).values({ companyId, runId: run.id, agentId, amountCents: options.reservedCents ?? "0", + state: "held", providerStartedAt: new Date() }); + return run; + }; + const queuedRun = async () => { + const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId: otherAgentId, invocationSource: "assignment", + status: "queued" }).returning(); + return run; + }; + return { companyId, agentId, otherAgentId, issue, block, pendingRun, queuedRun }; + } + + it("admits new work while a failed run awaits accounting (waitingRunCount gate)", async () => { + const f = await fixture(); + await f.pendingRun("failed"); + expect(await f.block()).toBeNull(); + await expect(reserveRunBudget(db, f.companyId, (await f.queuedRun()).id, null)).resolves.toMatchObject({ reused: false }); + }); + + it("admits new work while a native run recovers its accounting (recoveringRunCount gate)", async () => { + const f = await fixture(); + await f.pendingRun("failed", { recovering: true }); + expect(await f.block()).toBeNull(); + await expect(reserveRunBudget(db, f.companyId, (await f.queuedRun()).id, null)).resolves.toMatchObject({ reused: false }); + }); + + it("admits new work while a budget-owned scope hold awaits reconciliation (scope budget hold gate)", async () => { + const f = await fixture(); + // The hard stop that set the hold no longer applies; nothing has released it yet. + await db.update(agents).set({ pauseReason: "budget", lifecycleHolds: ["budget"] }).where(eq(agents.id, f.otherAgentId)); + await db.insert(budgetPolicies).values({ companyId: f.companyId, scopeType: "agent", scopeId: f.otherAgentId, + metric: "billed_cents", windowKind: "calendar_month_utc", amount: 1_000, notifyEnabled: false }); + expect(await f.block()).toBeNull(); + }); + + it("admits new work while budget inputs await evaluation (15 s budget_evaluation_work gate)", async () => { + const f = await fixture(); + await db.insert(budgetEvaluationWork).values({ companyId: f.companyId, createdAt: new Date(Date.now() - 60_000) }); + expect(await f.block()).toBeNull(); + await expect(reserveRunBudget(db, f.companyId, (await f.queuedRun()).id, null)).resolves.toMatchObject({ reused: false }); + }); + + it("never holds a successor for a predecessor run row that no longer exists (hasPendingRunCleanup gate)", async () => { + const f = await fixture(); + const missing = randomUUID(); + expect(await hasPendingRunCleanup(db, f.companyId, missing)).toBe(false); + // A predecessor that still owns a finalization coordinator holds it. + const owner = await f.pendingRun("failed"); + await db.update(heartbeatRuns).set({ costAccountingPending: false }).where(eq(heartbeatRuns.id, owner.id)); + await db.insert(nativeRunFinalizations).values({ companyId: f.companyId, issueId: f.issue.id, runId: owner.id, + phase: "observed", attempt: 1, leaseOwner: "live-owner", leaseExpiresAt: new Date(Date.now() + 60_000) }); + expect(await hasPendingRunCleanup(db, f.companyId, owner.id)).toBe(true); + }); + + // Unfinished accounting is exposure: its bounded estimate (the held + // reservation) still consumes capacity, and a hard stop still refuses. + it("counts unfinished accounting's bounded estimate as exposure and still refuses a reached hard stop", async () => { + const f = await fixture(1_000, "400"); + await f.pendingRun("failed", { reservedCents: "400" }); + await f.pendingRun("failed", { reservedCents: "400", recovering: true }); + expect(await f.block()).toBeNull(); + await expect(reserveRunBudget(db, f.companyId, (await f.queuedRun()).id, null)) + .rejects.toThrow(/reserved by unfinished runs/); + await db.insert(costEvents).values({ companyId: f.companyId, agentId: f.agentId, provider: "openai", model: "gpt", + costCents: 1_000, occurredAt: new Date() }); + expect((await f.block())?.reason).toMatch(/hard-stop/); + }); + + it("still refuses an explicit pause", async () => { + const f = await fixture(); + await f.pendingRun("failed"); + await db.update(companies).set({ status: "paused", pauseReason: "manual", pausedAt: new Date() }) + .where(eq(companies.id, f.companyId)); + expect((await f.block())?.reason).toMatch(/paused/); + }); +}); diff --git a/server/src/__tests__/budget-enforcement-scoping.test.ts b/server/src/__tests__/budget-enforcement-scoping.test.ts index 124e5015f91..bfe44479e7f 100644 --- a/server/src/__tests__/budget-enforcement-scoping.test.ts +++ b/server/src/__tests__/budget-enforcement-scoping.test.ts @@ -236,21 +236,23 @@ const support = await getEmbeddedPostgresTestSupport(); await expect(reserveRunBudget(db, f.companyId, await f.queuedRun(), otherProject.id)).resolves.toMatchObject({ reused: false }); }); - it("keeps missing-price terminal work held until accounting evidence arrives", async () => { + // Invariant 3: the missing price stays pending accounting debt, but it is + // exposure, never an admission gate. + it("keeps missing-price terminal accounting pending without holding new work", async () => { const f = await fixture(); await db.update(heartbeatRuns).set({ status: "cancelled", finishedAt: new Date() }).where(eq(heartbeatRuns.id, f.runId)); - expect(await budgetService(db).getInvocationBlock(f.companyId, f.agentId, { projectId: f.projectId })).not.toBeNull(); + expect(await budgetService(db).getInvocationBlock(f.companyId, f.agentId, { projectId: f.projectId })).toBeNull(); expect(await f.exposure()).toMatchObject({ pendingTerminalRunCount: 1, ledgerCents: "0.0000000", liveObservedCents: "0.0000000" }); expect(await accountRunCost(db, f.runId)).toBe(false); expect((await f.readRun()).costAccountingPending).toBe(true); expect(await db.select().from(costEvents).where(and(eq(costEvents.companyId, f.companyId), eq(costEvents.heartbeatRunId, f.runId)))).toHaveLength(0); }); - // Pending accounting holds admission only; it never places a budget hold. - it.each(["pre-provider release", "stale pending marker"])("releases the accounting admission hold after %s", async kind => { + // Pending accounting never places a budget hold or an admission gate. + it.each(["pre-provider release", "stale pending marker"])("settles pending accounting after %s without a hold", async kind => { const f = await fixture(); await db.update(heartbeatRuns).set({ status: "cancelled", finishedAt: new Date() }).where(eq(heartbeatRuns.id, f.runId)); - expect(await budgetService(db).getInvocationBlock(f.companyId, f.agentId, { projectId: f.projectId })).not.toBeNull(); + expect(await budgetService(db).getInvocationBlock(f.companyId, f.agentId, { projectId: f.projectId })).toBeNull(); await budgetService(db).evaluateAffectedScopes(f.companyId, f.agentId, f.projectId); await flushBudgetEvaluationWork(db); const [held] = await db.select().from(agents).where(eq(agents.id, f.agentId)); diff --git a/server/src/__tests__/budget-evaluation-work.test.ts b/server/src/__tests__/budget-evaluation-work.test.ts index 85e1c4b10f8..9bd733a9bee 100644 --- a/server/src/__tests__/budget-evaluation-work.test.ts +++ b/server/src/__tests__/budget-evaluation-work.test.ts @@ -86,14 +86,19 @@ postgres("independent accounting and deferred budget decisions", () => { expect(await pending(f.companyId)).toHaveLength(1); }); - it("blocks stale limited admissions until committed work is evaluated, then stays fresh while idle", async () => { + // Invariant 3: unevaluated inputs are not a gate. Admission reads the ledger + // and live exposure itself, so a reached hard stop refuses at once while the + // evaluator is still behind, and an old input under the limit never refuses. + it("admits limited work while inputs await evaluation, and refuses a reached hard stop inline", async () => { const f = await fixture(1000); await costService(db).createEvent(f.companyId, charge(f.agentId, randomUUID(), 1)); await db.update(budgetEvaluationWork).set({ createdAt: new Date(Date.now() - 16_000) }).where(eq(budgetEvaluationWork.companyId, f.companyId)); - expect((await budgetService(db).evaluateAdmission(f.companyId, f.agentId)).block?.reason).toContain("awaiting reconciliation"); - await flushBudgetEvaluationWork(db, f); expect((await budgetService(db).evaluateAdmission(f.companyId, f.agentId)).block).toBeNull(); - expect(await flushBudgetEvaluationWork(db, f)).toBe(0); + await costService(db).createEvent(f.companyId, charge(f.agentId, randomUUID(), 1000)); + expect(await pending(f.companyId)).not.toHaveLength(0); + expect((await budgetService(db).evaluateAdmission(f.companyId, f.agentId)).block?.reason).toMatch(/hard-stop/); + await flushBudgetEvaluationWork(db, f); + expect((await budgetService(db).evaluateAdmission(f.companyId, f.agentId)).block?.reason).toMatch(/hard-stop/); }); it("retains policy raise and manual pause when evaluation races them", async () => { @@ -108,7 +113,7 @@ postgres("independent accounting and deferred budget decisions", () => { expect((await db.select().from(companies).where(eq(companies.id, f.companyId)))[0].pauseReason).toBe("manual"); }); - it("does not apply another agent's stale backlog to a separately limited agent", async () => { + it("does not apply another agent's unevaluated hard stop to a separately limited agent", async () => { const f = await fixture(0); const otherAgentId = randomUUID(); await db.insert(agents).values({ id: otherAgentId, companyId: f.companyId, name: "Other worker", role: "engineer", status: "active", adapterType: "process" }); @@ -116,9 +121,11 @@ postgres("independent accounting and deferred budget decisions", () => { for (const agentId of [f.agentId, otherAgentId]) await service.upsertPolicy(f.companyId, { scopeType: "agent", scopeId: agentId, amount: 1000 }, null); await flushBudgetEvaluationWork(db, f); - await costService(db).createEvent(f.companyId, charge(f.agentId, randomUUID(), 1)); + await costService(db).createEvent(f.companyId, charge(f.agentId, randomUUID(), 1000)); await db.update(budgetEvaluationWork).set({ createdAt: new Date(Date.now() - 16_000) }).where(eq(budgetEvaluationWork.companyId, f.companyId)); - expect((await service.evaluateAdmission(f.companyId, f.agentId)).block).not.toBeNull(); + // The unevaluated charge reaches this agent's hard stop and refuses it + // inline; the other agent's separate limit is untouched. + expect((await service.evaluateAdmission(f.companyId, f.agentId)).block?.reason).toMatch(/hard-stop/); expect((await service.evaluateAdmission(f.companyId, otherAgentId)).block).toBeNull(); }); diff --git a/server/src/__tests__/budgets-service.test.ts b/server/src/__tests__/budgets-service.test.ts index 03b1c0e75fe..1276e8f8b4a 100644 --- a/server/src/__tests__/budgets-service.test.ts +++ b/server/src/__tests__/budgets-service.test.ts @@ -169,8 +169,8 @@ describeEmbeddedPostgres("budgetService release gate enforcement", () => { 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 })) - .resolves.toMatchObject({ reason: "New work must wait while a native run recovers its unfinished accounting." }); + // Invariant 3: a recovering run's accounting never gates admission. + await expect(budgetService(reader).getInvocationBlock(companyId, agentId, { projectId })).resolves.toBeNull(); } else { const block = await budgetService(reader).getInvocationBlock(companyId, agentId, { projectId }); if (state === "manual company pause") { @@ -199,9 +199,7 @@ describeEmbeddedPostgres("budgetService release gate enforcement", () => { } 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(await budgetService(db, { cancelWorkForScope }).getInvocationBlock(companyId, agentId, { projectId })).toBeNull(); expect(cancelWorkForScope).not.toHaveBeenCalled(); expect((await db.select().from(companies).where(eq(companies.id, companyId)))[0]) .toMatchObject({ status: "active", pauseReason: null }); @@ -222,7 +220,10 @@ describeEmbeddedPostgres("budgetService release gate enforcement", () => { 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)).not.toBeNull(); + // A budget hold with no reached hard stop is stale. Budget admission does + // not wait for its reconciliation (invariant 3); the paused agent itself is + // an explicit pause that agent invokability refuses until the worker lifts it. + expect(await budgetService(db).getInvocationBlock(companyId, agentId)).toBeNull(); await budgetService(db).evaluateAffectedScopes(companyId, agentId, null); await flushBudgetEvaluationWork(db, { companyId }); expect(await budgetService(db).getInvocationBlock(companyId, agentId)).toBeNull(); diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index b35cdbda554..aa8aa1a61fb 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -35,7 +35,7 @@ import * as durableChatWakeup from "../services/durable-chat-wakeup.js"; import * as adapterModule from "../adapters/index.js"; import { conflict } from "../errors.js"; import * as githubTaskPreparation from "../services/chat-channels/github/task-preparation.js"; -import { budgetPolicies } from "@paperclipai/db"; +import { budgetPolicies, costEvents } from "@paperclipai/db"; import { budgetService } from "../services/budgets.js"; import { idleWorkSnapshot } from "../services/task-admission.js"; import { readIdleLocalWork } from "../services/idle-local-work.js"; @@ -66573,9 +66573,12 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { if (mode === "legacy_no_prior") await db.delete(chatActions).where(and( eq(chatActions.deliveryId, context.action.deliveryId!), eq(chatActions.kind, "github_response_comment"))); if (realBudgetIngress) { - await db.update(heartbeatRuns).set({ costAccountingPending: true }).where(eq(heartbeatRuns.id, context.runId)); + // A reached hard stop the board later lifts. Pending accounting is no + // longer a budget hold (invariant 3). await db.insert(budgetPolicies).values({ companyId, scopeType: "agent", scopeId: context.fixture.assignedAgentId, metric: "billed_cents", windowKind: "lifetime", amount: 100, notifyEnabled: false }); + await db.insert(costEvents).values({ companyId, agentId: context.fixture.assignedAgentId, provider: "openai", model: "gpt", + costCents: 100, occurredAt: new Date() }); context.wakeup.mockImplementation(heartbeatService(db).wakeup); } else context.wakeup.mockImplementation(async (agentId, opts) => { const request = opts.durableChatRequest!; diff --git a/server/src/__tests__/closed-native-receipt-accounting.test.ts b/server/src/__tests__/closed-native-receipt-accounting.test.ts index 607e9f29806..fb1e56abd27 100644 --- a/server/src/__tests__/closed-native-receipt-accounting.test.ts +++ b/server/src/__tests__/closed-native-receipt-accounting.test.ts @@ -85,7 +85,7 @@ const support = await getEmbeddedPostgresTestSupport(); it("settles a stopped native terminal failure without a final receipt as unpriced usage", async () => { const f = await fixture(); - expect(await f.block()).toMatchObject({ reason: "New work must wait while a native run recovers its unfinished accounting." }); + expect(await f.block()).toBeNull(); // Invariant 3: unfinished accounting is exposure, never a gate. // The background reconciler must select this debt; it is not merely retryable by hand. expect(await reconcileRunCosts(db)).toMatchObject({ accounted: 1 }); @@ -171,9 +171,9 @@ const support = await getEmbeddedPostgresTestSupport(); // RC-06e: a resumed run's own continuation was refused by the accounting hold // of its task's interrupted run, so the task could never settle itself. - it("admits a task's continuation past its own recovering accounting, but not other tasks", async () => { + it("admits a task's continuation and other tasks past recovering accounting", async () => { const f = await fixture({ phase: "observed" }); - expect(await f.block()).toMatchObject({ reason: "New work must wait while a native run recovers its unfinished accounting." }); + expect(await f.block()).toBeNull(); // Invariant 3: unfinished accounting is exposure, never a gate. const continuation = randomUUID(); await db.insert(heartbeatRuns).values({ id: continuation, companyId: f.companyId, agentId: f.agentId, status: "queued", runtimeMode: "native", scopeKind: "issue", issueId: f.issue.id, nativeIssueId: f.issue.id }); @@ -185,16 +185,16 @@ const support = await getEmbeddedPostgresTestSupport(); continuationRun, reservation.id))).toBeNull(); expect(await recheckAdmittedRunBudgetReadOnly(db, { companyId: f.companyId, agentId: f.agentId, runId: continuation, projectId: null, reservationId: reservation.id })).toBeNull(); - // A legacy run without a task identity is not exempt from another task's hold. + // Other tasks and runs without a task identity are admitted too (invariant 3). const legacy = randomUUID(); await db.insert(heartbeatRuns).values({ id: legacy, companyId: f.companyId, agentId: f.otherAgentId, status: "queued" }); - await expect(reserveRunBudget(db, f.companyId, legacy, null)).rejects.toThrow(/recovers its unfinished accounting/); + await expect(reserveRunBudget(db, f.companyId, legacy, null)).resolves.toMatchObject({ reused: false }); const [otherIssue] = await db.insert(issues).values({ companyId: f.companyId, title: "Other task", status: "todo", assigneeAgentId: f.otherAgentId }).returning(); const other = randomUUID(); await db.insert(heartbeatRuns).values({ id: other, companyId: f.companyId, agentId: f.otherAgentId, status: "queued", runtimeMode: "native", scopeKind: "issue", issueId: otherIssue.id, nativeIssueId: otherIssue.id }); - await expect(reserveRunBudget(db, f.companyId, other, null)).rejects.toThrow(/recovers its unfinished accounting/); + await expect(reserveRunBudget(db, f.companyId, other, null)).resolves.toMatchObject({ reused: false }); }); // Storm 4 (run 6d3c76bf): after a restart storm several tasks recover at once. @@ -211,13 +211,13 @@ const support = await getEmbeddedPostgresTestSupport(); phase: "observed", attempt: 2, leaseOwner: owner, leaseExpiresAt: new Date(Date.now() + 60_000) }); await db.insert(budgetReservations).values({ companyId: f.companyId, runId: recovering, agentId: f.otherAgentId, amountCents: "0", state: "held", providerStartedAt: new Date() }); - expect(await f.block()).toMatchObject({ reason: "New work must wait while a native run recovers its unfinished accounting." }); + expect(await f.block()).toBeNull(); // Invariant 3: unfinished accounting is exposure, never a gate. await expect(reserveRunBudget(db, f.companyId, recovering, null, {}, owner)).resolves.toMatchObject({ reused: true }); - // New work for another task still waits. + // New work for another task is admitted too (invariant 3). const fresh = randomUUID(); await db.insert(heartbeatRuns).values({ id: fresh, companyId: f.companyId, agentId: f.otherAgentId, status: "queued", runtimeMode: "native", scopeKind: "issue", issueId: otherIssue.id, nativeIssueId: null }); - await expect(reserveRunBudget(db, f.companyId, fresh, null)).rejects.toThrow(/recovers its unfinished accounting/); + await expect(reserveRunBudget(db, f.companyId, fresh, null)).resolves.toMatchObject({ reused: false }); }); it("lets an owned recovery dispatch when other runs' held reservations fill the budget", async () => { @@ -300,7 +300,7 @@ const support = await getEmbeddedPostgresTestSupport(); // Between a native terminal failure and the verified provider stop, the debt // is closing, not over budget. A hard stop here paused the company and // cancelled an unrelated running task in the live restart wave. - it("holds new work without a hard stop while a terminal native receipt is closing", async () => { + it("neither pauses nor holds new work while a terminal native receipt is closing", async () => { const f = await fixture({ processPid: 424242 }); await budgetService(db).evaluateAffectedScopes(f.companyId, f.agentId, null); await flushBudgetEvaluationWork(db); @@ -308,13 +308,13 @@ const support = await getEmbeddedPostgresTestSupport(); expect(company.status).toBe("active"); expect(company.pauseReason).toBeNull(); expect(await db.select().from(budgetIncidents).where(eq(budgetIncidents.companyId, f.companyId))).toHaveLength(0); - expect(await f.block()).toMatchObject({ reason: "New work must wait while a native run recovers its unfinished accounting." }); + expect(await f.block()).toBeNull(); // Invariant 3: unfinished accounting is exposure, never a gate. }); // Chaos waves: a run that had just succeeded was terminal for ~100 ms before // its charge committed. The evaluator read that gap as a hard stop, paused the // company and cancelled unrelated running work, which then never settled. - it("neither pauses nor holds admission while a succeeded run's charge commits", async () => { + it("neither pauses nor holds admission while a run's charge commits", async () => { const f = await fixture(); await db.update(heartbeatRuns).set({ status: "succeeded", errorCode: null, usageJson: sql`${heartbeatRuns.usageJson} || '{"accountingReceiptReady":true,"costUsd":0.1}'::jsonb` }) @@ -334,14 +334,14 @@ const support = await getEmbeddedPostgresTestSupport(); expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runningId)))[0].status).toBe("running"); // Its provider work is over and its observed cost counts, like a running run. expect(await f.block()).toBeNull(); - // A run that ended without succeeding still holds new work. + // A run that ended without succeeding no longer holds new work (invariant 3). await db.update(heartbeatRuns).set({ status: "failed" }).where(eq(heartbeatRuns.id, f.run.id)); - expect(await f.block()).toMatchObject({ reason: "Company cannot start work while completed runs await accounting." }); + expect(await f.block()).toBeNull(); // Invariant 3: unfinished accounting is exposure, never a gate. }); it("settles a finalization-deadline failure once its local stop is verified", async () => { const f = await fixture({ failureCode: "execution_finalization_deadline_exceeded" }); - expect(await f.block()).not.toBeNull(); + expect(await f.block()).toBeNull(); // Invariant 3: the pending debt never gates admission. expect(await reconcileRunCosts(db)).toMatchObject({ accounted: 1 }); await flushBudgetEvaluationWork(db); expect(await f.block()).toBeNull(); @@ -359,7 +359,7 @@ const support = await getEmbeddedPostgresTestSupport(); }); // Upgrade path: an older release paused the company for pending accounting. - it("releases a budget-owned pause that only pending accounting kept", async () => { + it("releases a budget-owned pause that only pending accounting kept, and admits work", async () => { const f = await fixture({ stopEvidence: false }); await db.update(companies).set({ status: "paused", pauseReason: "budget", pausedAt: new Date() }) .where(eq(companies.id, f.companyId)); @@ -373,7 +373,7 @@ const support = await getEmbeddedPostgresTestSupport(); const [company] = await db.select().from(companies).where(eq(companies.id, f.companyId)); expect(company).toMatchObject({ status: "active", pauseReason: null }); expect((await db.select().from(budgetIncidents).where(eq(budgetIncidents.companyId, f.companyId)))[0].status).not.toBe("open"); - expect(await f.block()).toMatchObject({ reason: "Company cannot start work while completed runs await accounting." }); + expect(await f.block()).toBeNull(); // Invariant 3: unfinished accounting is exposure, never a gate. const [policy] = await db.select().from(budgetPolicies).where(eq(budgetPolicies.id, f.policy.id)); expect((await budgetService(db).overview(f.companyId)).policies[0]?.status).not.toBe("hard_stop"); expect(policy.id).toBe(f.policy.id); @@ -406,7 +406,7 @@ const support = await getEmbeddedPostgresTestSupport(); it.each(["failed", "cancelled"] as const)("settles a %s remote run once every sandbox has a termination receipt", async status => { const f = await remoteFixture({ status, leases: [{ receipt: "destroyed" }, { receipt: "stopped" }] }); - expect(await f.block()).not.toBeNull(); + expect(await f.block()).toBeNull(); // Invariant 3: the pending debt never gates admission. expect(await reconcileRunCosts(db)).toMatchObject({ accounted: 1 }); await flushBudgetEvaluationWork(db); expect((await f.readRun()).costAccountingPending).toBe(false); @@ -468,7 +468,7 @@ const support = await getEmbeddedPostgresTestSupport(); await db.update(heartbeatRuns).set({ usageJson: sql`${heartbeatRuns.usageJson} || ${JSON.stringify({ budgetExposure: { sourceId: f.sourceId, sequence: 6, costCents: "600" } })}::jsonb` }) .where(eq(heartbeatRuns.id, f.run.id)); - expect((await f.block())?.reason).toMatch(/hard-stop|await accounting/); + expect((await f.block())?.reason).toMatch(/hard-stop/); }); // s7504 run bda81857 (ACPX Claude, K4): succeeded with a committed result, @@ -503,7 +503,7 @@ const support = await getEmbeddedPostgresTestSupport(); it("closes a result-accepted run's lost receipt once its local stop is verified", async () => { const f = await acceptedResultFixture(); - expect(await f.block()).toMatchObject({ reason: "Company cannot start work while completed runs await accounting." }); + expect(await f.block()).toBeNull(); // Invariant 3: unfinished accounting is exposure, never a gate. expect(await reconcileRunCosts(db)).toMatchObject({ accounted: 1 }); await flushBudgetEvaluationWork(db); expect(await f.charges()).toMatchObject([{ costStatus: "unpriced" }]); @@ -512,10 +512,10 @@ const support = await getEmbeddedPostgresTestSupport(); // RC-06g: with no closure evidence at all, one run must not hold a company's // new work forever. - it("holds a run without closure evidence only until the bound, then settles it visibly", async () => { + it("settles a run without closure evidence at the bound, visibly, without holding work", async () => { const recent = await acceptedResultFixture({ stopEvidence: false, finishedMinutesAgo: 5 }); expect(await reconcileRunCosts(db)).toMatchObject({ accounted: 0 }); - expect(await recent.block()).not.toBeNull(); + expect(await recent.block()).toBeNull(); // Invariant 3: the open debt never gates admission. // The worker schedules a wake for the bound instead of polling. const work = await readCostAccountingWork(db, Date.now()); expect(work.nextRunAt).not.toBeNull(); @@ -540,12 +540,12 @@ const support = await getEmbeddedPostgresTestSupport(); // Storm 4 (run f879895f): session.open timed out, the local stop was never // verified (the identity is still recorded), and recovery ended in terminal // failure. That one run held all new company work for hours. - it("bounds the hold of a terminal failure whose stop was never verified", async () => { + it("never holds work for a terminal failure whose stop was never verified", async () => { const f = await fixture({ processPid: 15099, stopEvidence: false }); await db.update(heartbeatRuns).set({ errorCode: "native_session_interrupted", finishedAt: new Date(Date.now() - 2 * 60_000) }) .where(eq(heartbeatRuns.id, f.run.id)); expect(await reconcileRunCosts(db)).toMatchObject({ accounted: 0 }); - expect(await f.block()).toMatchObject({ reason: "New work must wait while a native run recovers its unfinished accounting." }); + expect(await f.block()).toBeNull(); // Invariant 3: unfinished accounting is exposure, never a gate. await db.update(heartbeatRuns).set({ finishedAt: new Date(Date.now() - 11 * 60_000) }).where(eq(heartbeatRuns.id, f.run.id)); expect(await reconcileRunCosts(db)).toMatchObject({ accounted: 1 }); await flushBudgetEvaluationWork(db); @@ -628,10 +628,10 @@ const support = await getEmbeddedPostgresTestSupport(); // A failed usage capture never completes its receipt. It used to be skipped // by accounting and held company admission until the UTC month rolled over. - it("settles a capture failure as unpriced at the closure bound and releases held work", async () => { + it("settles a capture failure as unpriced at the closure bound without holding work", async () => { const f = await fixture(); await captureFailed(f, 11, { inputTokens: 1200, outputTokens: 80 }); - expect(await f.block()).not.toBeNull(); + expect(await f.block()).toBeNull(); // Invariant 3: the open debt never gates admission. expect((await readCostAccountingWork(db, Date.now())).pending).toBe(true); expect(await reconcileRunCosts(db)).toMatchObject({ accounted: 1 }); await flushBudgetEvaluationWork(db); diff --git a/server/src/__tests__/heartbeat-held-comment-wake.test.ts b/server/src/__tests__/heartbeat-held-comment-wake.test.ts index 8f57d676e30..2e0efced5a5 100644 --- a/server/src/__tests__/heartbeat-held-comment-wake.test.ts +++ b/server/src/__tests__/heartbeat-held-comment-wake.test.ts @@ -19,11 +19,11 @@ if (!recoveryPostgresSupport.supported) console.warn(`Skipping held comment wake /** * RC-37: a follow-up comment saved behind a run is promoted when that run ends. - * A temporary company admission hold (accounting that is still settling after a - * restart) can refuse the promoted run before it starts. The comment must stay - * saved and be delivered exactly once after the hold clears. + * Another run's unfinished accounting used to hold company admission and + * refuse it. Since Phase 3 accounting never gates admission (invariant 3), and + * each saved intent is delivered exactly once (invariant 4). */ -describePostgres("a promoted comment refused by a temporary admission hold", () => { +describePostgres("saved intents while another run's accounting is pending", () => { let temporary: Awaited>; let db: ReturnType; let heartbeat: ReturnType; @@ -121,119 +121,76 @@ describePostgres("a promoted comment refused by a temporary admission hold", () await heartbeat.drainActiveRunExecutions(); } - it("keeps the comment saved through the hold and delivers it exactly once after the hold clears", async () => { + // Phase 3 (invariant 3): another run's unfinished accounting never holds + // admission. Each intent that used to wait out the accounting hold is now + // delivered at once, exactly once, and settling accounting later never + // delivers it again. Before Phase 3 every one of these waited. + it("delivers a saved comment once while another run's accounting is still pending", async () => { const f = await seedHeldComment(); - // The source run has ended: the saved comment is promoted while the hold lasts. - await settle(f); await settle(f); - expect(deliveries(f.commentId)).toHaveLength(0); - // The comment stays pending: either saved or on an unstarted queued run. - const held = await readWake(f.wakeId); - expect(["deferred_issue_execution", "queued"]).toContain(held.status); - const pending = held.runId ? (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, held.runId)))[0] : null; - if (pending) expect(pending).toMatchObject({ status: "queued", startedAt: null }); - expect((await issueRuns(f)).filter(run => run.status === "cancelled")).toEqual([]); - const [waiting] = await db.select().from(issues).where(eq(issues.id, f.issueId)); - expect(waiting.status).not.toBe("blocked"); - - // Accounting settles and the hold clears. Settlement wakes the held work. - expect(await settleAccounting(f)).toEqual(new Set([DELIVERY_QUEUES.queuedRunDispatch, - DELIVERY_QUEUES.deferredExecutionReceipts, DELIVERY_QUEUES.strandedAssignments])); - await db.update(agentWakeupRequests).set({ updatedAt: new Date(Date.now() - 60_000) }) - .where(eq(agentWakeupRequests.id, f.wakeId)); await settle(f); await settle(f); expect(deliveries(f.commentId)).toHaveLength(1); - - // Later sweeps never deliver it again. - await settle(f); + expect((await issueRuns(f)).filter(run => run.status === "cancelled")).toEqual([]); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.pendingRunId)))[0].costAccountingPending).toBe(true); + await settleAccounting(f); + await settle(f); await settle(f); expect(deliveries(f.commentId)).toHaveLength(1); }, 180_000); - it("saves a comment whose wake the hold refuses at enqueue and delivers it once after the hold clears", async () => { - // s7532/t06: the comment arrived during the hold; its wake was refused - // before admission and written as a terminal skip, so it was lost. + it("admits a comment posted during pending accounting at enqueue and delivers it once", async () => { + // s7532/t06: the enqueue gate refused the comment during the hold. const f = await seedHeldComment(); await db.update(agentWakeupRequests).set({ status: "cancelled", finishedAt: new Date() }).where(eq(agentWakeupRequests.id, f.wakeId)); const [comment] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, - authorUserId: f.userId, body: "Posted during the hold." }).returning(); - await expect(heartbeat.wakeup(f.agentId, { source: "automation", triggerDetail: "system", reason: "issue_commented", + authorUserId: f.userId, body: "Posted while accounting is pending." }).returning(); + await heartbeat.wakeup(f.agentId, { source: "automation", triggerDetail: "system", reason: "issue_commented", payload: { issueId: f.issueId, commentId: comment.id, mutation: "comment" }, contextSnapshot: { issueId: f.issueId, taskId: f.issueId, commentId: comment.id, wakeCommentId: comment.id, source: "issue.comment", wakeReason: "issue_commented" }, - requestedByActorType: "user", requestedByActorId: f.userId })).rejects.toThrow(/await accounting/); - const saved = (await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))) + requestedByActorType: "user", requestedByActorId: f.userId }); + const wakes = (await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))) .filter(wake => (wake.payload as { commentId?: string }).commentId === comment.id); - expect(saved).toEqual([expect.objectContaining({ status: "deferred_issue_execution", runId: null, - payload: expect.objectContaining({ executionWait: expect.objectContaining({ reason: "budget.blocked" }) }) })]); - await settle(f); - expect(deliveries(comment.id)).toHaveLength(0); - - await settleAccounting(f); - await db.update(agentWakeupRequests).set({ updatedAt: new Date(Date.now() - 60_000) }).where(eq(agentWakeupRequests.id, saved[0].id)); + expect(wakes.map(wake => wake.status)).not.toContain("skipped"); await settle(f); await settle(f); expect(deliveries(comment.id)).toHaveLength(1); + await settleAccounting(f); await settle(f); expect(deliveries(comment.id)).toHaveLength(1); }, 180_000); - it("keeps a Retry clicked during the hold pending and starts it exactly once after the hold clears", async () => { + it("starts a Retry clicked while another run's accounting is pending exactly once", async () => { const f = await seedHeldComment(); - // The run the user retries failed; its saved comment was already discarded. await db.update(heartbeatRuns).set({ status: "failed", errorCode: "adapter_failed" }).where(eq(heartbeatRuns.id, f.sourceRunId)); await db.update(agentWakeupRequests).set({ status: "cancelled", finishedAt: new Date() }).where(eq(agentWakeupRequests.id, f.wakeId)); const retry = await requestTaskRetry({ db, companyId: f.companyId, issueId: f.issueId, actorId: f.userId, idempotencyKey: randomUUID(), observedRunId: f.sourceRunId }); expect(retry).toMatchObject({ accepted: true }); await settle(f); await settle(f); - const retried = () => dispatched.filter(turn => turn.context.wakeReason === "task_retry" - || (turn.context.taskRetry as { generationId?: string } | undefined)?.generationId === retry.requestId); - expect(retried()).toHaveLength(0); - const held = await readWake(retry.requestId!); - expect(["deferred_issue_execution", "queued"]).toContain(held.status); - expect((await issueRuns(f)).filter(run => run.id !== f.sourceRunId && run.status === "cancelled")).toEqual([]); - - // Nothing polls a held Retry: settling accounting must wake its owner. - expect(await settleAccounting(f)).toContain(DELIVERY_QUEUES.deferredExecutionReceipts); - await db.update(agentWakeupRequests).set({ updatedAt: new Date(Date.now() - 60_000) }) - .where(eq(agentWakeupRequests.id, retry.requestId!)); - await settle(f); await settle(f); - const started = (await issueRuns(f)).filter(run => run.id !== f.sourceRunId && run.startedAt); - expect(started).toHaveLength(1); + const started = () => issueRuns(f).then(runs => runs.filter(run => run.id !== f.sourceRunId && run.startedAt)); + expect(await started()).toHaveLength(1); + await settleAccounting(f); await settle(f); - expect((await issueRuns(f)).filter(run => run.id !== f.sourceRunId && run.startedAt)).toHaveLength(1); + expect(await started()).toHaveLength(1); }, 180_000); - it("starts a new task whose assignment was refused during the hold once accounting settles", async () => { + it("starts a new task assigned while another run's accounting is pending", async () => { const f = await seedHeldComment(); - // A second task is created and assigned while the hold lasts. - const [task] = await db.insert(issues).values({ companyId: f.companyId, title: "Created during the hold", status: "todo", + const [task] = await db.insert(issues).values({ companyId: f.companyId, title: "Created during pending accounting", status: "todo", assigneeAgentId: f.agentId, responsibleUserId: f.userId, issueNumber: 2 }).returning(); - await expect(queueIssueAssignmentWakeup({ heartbeat, issue: { id: task.id, assigneeAgentId: f.agentId, status: "todo" }, + await queueIssueAssignmentWakeup({ heartbeat, issue: { id: task.id, assigneeAgentId: f.agentId, status: "todo" }, reason: "issue_assigned", mutation: "create", contextSource: "issue.create", - requestedByActorType: "user", requestedByActorId: f.userId, rethrowOnError: true })).rejects.toMatchObject({ status: 409 }); - await heartbeat.reconcileStrandedAssignedIssues({ companyId: f.companyId }); - expect((await issueRuns({ companyId: f.companyId, issueId: task.id }))).toHaveLength(0); - // Nothing polls a refused assignment: settling accounting must wake its recovery. - expect(await settleAccounting(f)).toContain(DELIVERY_QUEUES.strandedAssignments); - await heartbeat.reconcileStrandedAssignedIssues({ companyId: f.companyId }); + requestedByActorType: "user", requestedByActorId: f.userId, rethrowOnError: true }); await heartbeat.drainActiveRunExecutions(); const started = (await issueRuns({ companyId: f.companyId, issueId: task.id })).filter(run => run.startedAt); expect(started).toHaveLength(1); }, 180_000); - it("waits out the hold instead of blocking a task whose continuation it refused", async () => { + it("continues a task whose run ended while another run's accounting is pending", async () => { const f = await seedHeldComment(); - // The task's run ended without settling the task; its comment was already consumed. await db.update(agentWakeupRequests).set({ status: "cancelled", finishedAt: new Date() }).where(eq(agentWakeupRequests.id, f.wakeId)); await heartbeat.reconcileStrandedAssignedIssues({ companyId: f.companyId }); - await heartbeat.reconcileStrandedAssignedIssues({ companyId: f.companyId }); - const [waiting] = await db.select().from(issues).where(eq(issues.id, f.issueId)); - expect(waiting.status).toBe("in_progress"); - expect((await issueRuns(f)).filter(run => run.id !== f.sourceRunId)).toHaveLength(0); - - await settleAccounting(f); - await heartbeat.reconcileStrandedAssignedIssues({ companyId: f.companyId }); await heartbeat.drainActiveRunExecutions(); expect((await issueRuns(f)).filter(run => run.id !== f.sourceRunId && run.startedAt)).toHaveLength(1); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0].status).not.toBe("blocked"); }, 180_000); it.each([ diff --git a/server/src/__tests__/heartbeat-refused-assignment-wake.test.ts b/server/src/__tests__/heartbeat-refused-assignment-wake.test.ts index c3cef9ee955..53fa461398f 100644 --- a/server/src/__tests__/heartbeat-refused-assignment-wake.test.ts +++ b/server/src/__tests__/heartbeat-refused-assignment-wake.test.ts @@ -5,7 +5,7 @@ import path from "node:path"; import { and, eq } from "drizzle-orm"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; import { agents, agentWakeupRequests, authUsers, budgetPolicies, companies, companyMemberships, createDb, heartbeatRuns, - issues } from "@paperclipai/db"; + issues, projects } from "@paperclipai/db"; import { registerServerAdapter, unregisterServerAdapter } from "../adapters/index.js"; import { heartbeatService } from "../services/heartbeat.js"; import { queueIssueAssignmentWakeup } from "../services/issue-assignment-wakeup.js"; @@ -16,8 +16,9 @@ const describePostgres = recoveryPostgresSupport.supported ? describe : describe if (!recoveryPostgresSupport.supported) console.warn(`Skipping refused assignment wake tests: ${recoveryPostgresSupport.reason}`); /** - * A short company admission hold (for example accounting that is still settling - * after a restart) can refuse a new task's assignment wake. The task must wait + * A temporary budget hold can refuse a new task's assignment wake. Since Phase + * 3 another run's accounting never holds admission (invariant 3); the hold + * left is a budget-owned pause that is not a reached hard stop. The task must wait * honestly and start once the hold clears, instead of resting `blocked` with no run. */ describePostgres("refused assignment wake during a temporary admission hold", () => { @@ -55,8 +56,10 @@ describePostgres("refused assignment wake during a temporary admission hold", () if (home) await rm(home, { recursive: true, force: true }); }); - /** A company with a hard-stop budget and one completed run whose accounting - * has not settled yet: admission refuses new work until it settles. */ + /** A company with a hard-stop budget whose task's project carries a + * budget-owned pause that no project policy justifies: admission refuses the + * task's work until the pause lifts. A completed run with pending + * accounting is present too; it must not hold anything (invariant 3). */ async function seedHeldCompany() { const companyId = randomUUID(), agentId = randomUUID(), otherAgentId = randomUUID(), issueId = randomUUID(); const userId = `hold-user-${companyId}`; @@ -72,9 +75,15 @@ describePostgres("refused assignment wake during a temporary admission hold", () const [pending] = await db.insert(heartbeatRuns).values({ companyId, agentId: otherAgentId, invocationSource: "automation", triggerDetail: "system", status: "succeeded", startedAt: new Date(Date.now() - 60_000), finishedAt: new Date(), costAccountingPending: true, contextSnapshot: {} }).returning(); - await db.insert(issues).values({ id: issueId, companyId, title: "New task created during the hold", status: "todo", + const [project] = await db.insert(projects).values({ companyId, name: "Budget-paused project", status: "in_progress", + pauseReason: "budget", pausedAt: new Date() }).returning(); + await db.insert(issues).values({ id: issueId, companyId, projectId: project.id, title: "New task created during the hold", status: "todo", assigneeAgentId: agentId, responsibleUserId: userId, issueNumber: 1 }); - return { companyId, agentId, issueId, userId, pendingRunId: pending.id }; + return { companyId, agentId, issueId, userId, projectId: project.id, pendingRunId: pending.id }; + } + + async function liftHold(f: { projectId: string }) { + await db.update(projects).set({ pauseReason: null, pausedAt: null }).where(eq(projects.id, f.projectId)); } async function assign(f: Awaited>) { @@ -99,12 +108,13 @@ describePostgres("refused assignment wake during a temporary admission hold", () const [waiting] = await db.select().from(issues).where(eq(issues.id, f.issueId)); expect(waiting.status).toBe("todo"); expect(await issueRuns(f)).toHaveLength(0); - // The task names what it waits for; it does not offer a Retry that cannot start either. - expect(await taskInterruptionPresentation(db, waiting)).toMatchObject({ responsibility: "system", state: "working", - canRetry: false, retryWait: { reason: "budget.blocked", message: expect.stringContaining("starts automatically") } }); + // This pause needs an operator resume, so the task asks the user; it does + // not offer a Retry that cannot start either. + expect(await taskInterruptionPresentation(db, waiting)).toMatchObject({ responsibility: "user", state: "waiting_for_user", + canRetry: false }); - // Accounting settles: the hold clears. - await db.update(heartbeatRuns).set({ costAccountingPending: false }).where(eq(heartbeatRuns.id, f.pendingRunId)); + // The budget pause lifts: the hold clears. + await liftHold(f); await heartbeat.reconcileStrandedAssignedIssues({ companyId: f.companyId }); await heartbeat.reconcileStrandedAssignedIssues({ companyId: f.companyId }); await heartbeat.drainActiveRunExecutions(); @@ -131,7 +141,7 @@ describePostgres("refused assignment wake during a temporary admission hold", () const f = await seedHeldCompany(); await expect(assign(f)).rejects.toMatchObject({ status: 409 }); await db.update(agents).set({ status: "terminated" }).where(eq(agents.id, f.agentId)); - await db.update(heartbeatRuns).set({ costAccountingPending: false }).where(eq(heartbeatRuns.id, f.pendingRunId)); + await liftHold(f); await heartbeat.reconcileStrandedAssignedIssues({ companyId: f.companyId }); await heartbeat.drainActiveRunExecutions(); expect(await issueRuns(f)).toHaveLength(0); @@ -141,7 +151,7 @@ describePostgres("refused assignment wake during a temporary admission hold", () const f = await seedHeldCompany(); await expect(assign(f)).rejects.toMatchObject({ status: 409 }); await db.update(issues).set({ status: "cancelled" }).where(eq(issues.id, f.issueId)); - await db.update(heartbeatRuns).set({ costAccountingPending: false }).where(eq(heartbeatRuns.id, f.pendingRunId)); + await liftHold(f); await heartbeat.reconcileStrandedAssignedIssues({ companyId: f.companyId }); await heartbeat.drainActiveRunExecutions(); expect(await issueRuns(f)).toHaveLength(0); diff --git a/server/src/__tests__/ownership-reconciliation-report.test.ts b/server/src/__tests__/ownership-reconciliation-report.test.ts index 3cb6f8f483a..ecc246bb638 100644 --- a/server/src/__tests__/ownership-reconciliation-report.test.ts +++ b/server/src/__tests__/ownership-reconciliation-report.test.ts @@ -78,6 +78,7 @@ describe("report-only ownership reconciliation", () => { liveRun: randomUUID(), deadRun: randomUUID(), debtRun: randomUUID(), boundedRun: randomUUID(), liveIssue: randomUUID(), stalledIssue: randomUUID(), orphanLease: randomUUID(), unownedCleanup: randomUUID(), foreignIntent: randomUUID(), wake: randomUUID(), + heldWake: randomUUID(), }; await db.insert(issues).values([ { id: ids.liveIssue, companyId, title: "Live", status: "in_progress", assigneeAgentId: agentId }, @@ -116,6 +117,11 @@ describe("report-only ownership reconciliation", () => { // A wake request ended terminally on a temporary refusal. await db.insert(agentWakeupRequests).values({ id: ids.wake, companyId, agentId, source: "assignment", status: "skipped", error: "Company budget hold: admission paused" }); + // A work intent refused for another run's unfinished accounting (the gate + // Phase 3 removed). Saved, not dropped, so it is not an I4 violation. + await db.insert(agentWakeupRequests).values({ id: ids.heldWake, companyId, agentId, source: "automation", + status: "deferred_issue_execution", payload: { executionWait: { reason: "budget.blocked", + message: "Company cannot start work while completed runs await accounting." } } }); } async function snapshot() { @@ -160,9 +166,11 @@ describe("report-only ownership reconciliation", () => { expect(rules(first, "environment_pending_cleanup")).toEqual(["I1_owned"]); // I2 needs the owner proven gone on two consecutive passes. expect(rules(first, "environment_run_lease_active")).toEqual([]); + // Phase 3: pending accounting is exposure, not a gate. Only a refusal that + // names an accounting hold is an I3 violation. expect(first.violations.filter(violation => violation.rule === "I3_admission_independent") - .map(violation => violation.recordId)).toEqual([ids.debtRun]); - expect(rules(first, "wake_request")).toEqual(["I4_exactly_once_intent"]); + .map(violation => violation.recordId)).toEqual([ids.heldWake]); + expect(rules(first, "wake_request")).toEqual(["I3_admission_independent", "I4_exactly_once_intent"]); expect(first.violations.filter(violation => violation.rule === "I7_visible_next_action") .map(violation => violation.recordId)).toEqual([ids.stalledIssue]); expect(Object.keys(first.coverageGaps)).toEqual(["I5_work_based_budgets", "I6_generation_fencing", "I8_fast_turns", @@ -285,10 +293,11 @@ describe("report-only ownership reconciliation", () => { expect(await db.select().from(activityLog)).toEqual([]); }); - it("does not report admission holds without an active budget policy", async () => { + it("never reports pending accounting itself as an admission violation", async () => { await seed(); - await db.update(budgetPolicies).set({ isActive: false }); + await db.delete(agentWakeupRequests).where(eq(agentWakeupRequests.id, ids.heldWake)); const report = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false }); + expect(kindReport(report, "run_accounting_debt")).toMatchObject({ inProgress: 2 }); expect(report.violations.filter(violation => violation.rule === "I3_admission_independent")).toEqual([]); }); @@ -319,15 +328,31 @@ describe("report-only ownership reconciliation", () => { .where(eq(activityLog.action, OWNERSHIP_INVARIANT_VIOLATION_ACTION))).toEqual([{ entityId: "orphan-1" }]); }); - it("counts admission holds only for policies whose scope includes the run", async () => { - await seed(); - await db.update(budgetPolicies).set({ scopeType: "agent", scopeId: randomUUID() }); - const other = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false }); - expect(other.violations.filter(violation => violation.rule === "I3_admission_independent")).toEqual([]); - await db.update(budgetPolicies).set({ scopeId: agentId }); - const own = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false }); - expect(own.violations.filter(violation => violation.rule === "I3_admission_independent").map(violation => violation.recordId)) - .toEqual([ids.debtRun]); + it("reports a run cancelled for an accounting hold", async () => { + const runId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "cancelled", invocationSource: "manual", + error: "New work must wait while a native run recovers its unfinished accounting." }); + const report = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false }); + expect(report.violations.filter(violation => violation.rule === "I3_admission_independent")).toEqual([ + expect.objectContaining({ kind: "queued_run", recordId: runId, reason: "refused_for_accounting_hold" })]); + }); + + it("reads every accounting-hold refusal in a busy window across passes", async () => { + const state = createOwnershipReportState(); + await runOwnershipReconciliationPass(db, { state, record: false, limitPerKind: 2 }); + const refused = Array.from({ length: 25 }, () => randomUUID()); + await db.insert(agentWakeupRequests).values(refused.map(id => ({ id, companyId, agentId, source: "assignment", + status: "skipped", error: "Company cannot start work while completed runs await accounting." }))); + const seen: string[] = []; + const first = await runOwnershipReconciliationPass(db, { state, record: false, limitPerKind: 2 }); + // Ten pages of two: the rest waits for the next pass, which is kept due. + expect(first.retryNeeded).toBe(true); + seen.push(...first.violations.filter(violation => violation.rule === "I3_admission_independent").map(violation => violation.recordId)); + const second = await runOwnershipReconciliationPass(db, { state, record: false, limitPerKind: 2 }); + seen.push(...second.violations.filter(violation => violation.rule === "I3_admission_independent").map(violation => violation.recordId)); + expect(seen.sort()).toEqual([...refused].sort()); + const third = await runOwnershipReconciliationPass(db, { state, record: false, limitPerKind: 2 }); + expect(third.violations.filter(violation => violation.rule === "I3_admission_independent")).toEqual([]); }); it("treats a saved monitor deadline as a next action", async () => { @@ -372,15 +397,6 @@ describe("report-only ownership reconciliation", () => { expect(passes).toEqual([true, true, true, true, false, false]); }); - it("counts late-receipt debt as an admission hold, as the gate does", async () => { - await seed(); - await db.update(heartbeatRuns).set({ costAccountingPending: false, usageJson: { accountingLateReceiptPending: true } }) - .where(eq(heartbeatRuns.id, ids.debtRun)); - const report = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false }); - expect(report.violations.filter(violation => violation.rule === "I3_admission_independent").map(violation => violation.recordId)) - .toEqual([ids.debtRun]); - }); - it("reads a kind larger than one page in a single pass, so second proofs compare the same records", async () => { const runId = randomUUID(); await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "failed", invocationSource: "manual" }); diff --git a/server/src/services/budget-block-ownership.ts b/server/src/services/budget-block-ownership.ts index bb89f33e94c..ea2920dec7f 100644 --- a/server/src/services/budget-block-ownership.ts +++ b/server/src/services/budget-block-ownership.ts @@ -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, diff --git a/server/src/services/budget-evaluation-work.ts b/server/src/services/budget-evaluation-work.ts index 3e1ac4bd699..5123b323e0a 100644 --- a/server/src/services/budget-evaluation-work.ts +++ b/server/src/services/budget-evaluation-work.ts @@ -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 { 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 { - const [row] = await db.select({ oldest: sql`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; -} diff --git a/server/src/services/budget-exposure.ts b/server/src/services/budget-exposure.ts index 901dfab6775..dcb5ace3101 100644 --- a/server/src/services/budget-exposure.ts +++ b/server/src/services/budget-exposure.ts @@ -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; 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 { +async function readObservedSpends(db: Db, policies: SpendPolicy[], now: Date): Promise { 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 = {}, heldColumns: Record = {}, runColumns: Record = {}, browserColumns: Record = {}; @@ -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 { +export function computeObservedSpends(db: Db, policies: SpendPolicy[], now = new Date()): Promise { 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 }); } diff --git a/server/src/services/budgets.ts b/server/src/services/budgets.ts index 1c1762b1736..dde87350c09 100644 --- a/server/src/services/budgets.ts +++ b/server/src/services/budgets.ts @@ -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= 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>) { - 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>) => ({ 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); diff --git a/server/src/services/execution-blocker.ts b/server/src/services/execution-blocker.ts index 55209710ea0..8be6f38c394 100644 --- a/server/src/services/execution-blocker.ts +++ b/server/src/services/execution-blocker.ts @@ -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. */ diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index 20078cd1f0f..c2fcfea2347 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -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); diff --git a/server/src/services/recovery/ownership-invariant-report.ts b/server/src/services/recovery/ownership-invariant-report.ts index cd6da5c1e84..490e7bcb918 100644 --- a/server/src/services/recovery/ownership-invariant-report.ts +++ b/server/src/services/recovery/ownership-invariant-report.ts @@ -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 = { 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; /** 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 { - 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 { + 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`${agentWakeupRequests.payload}->>'issueId'`, updatedAt: sql`${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`${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