From af7b7d3316c86cdefc0845dc90f26b7cb436b9d1 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 21:13:42 -0500 Subject: [PATCH] fix: park unchanged project budget execution waits Lock and re-read project pause eligibility before publishing budget intent. Cover actual project Retry parking/release and concurrent pause admission without no-op hints. Co-Authored-By: Paperclip --- .../2026-10-30-deferred-execution-sleep.md | 2 ++ server/src/__tests__/budgets-service.test.ts | 26 +++++++++++++++++++ .../deferred-execution-receipt-work.test.ts | 25 +++++++++++------- server/src/services/budgets.ts | 7 ++++- 4 files changed, 49 insertions(+), 11 deletions(-) diff --git a/doc/plans/2026-10-30-deferred-execution-sleep.md b/doc/plans/2026-10-30-deferred-execution-sleep.md index e84d9abf165..ded551cdf22 100644 --- a/doc/plans/2026-10-30-deferred-execution-sleep.md +++ b/doc/plans/2026-10-30-deferred-execution-sleep.md @@ -25,3 +25,5 @@ A resume that first records a new execution-wait reason arms its existing 30-sec Persisted budget and agent refusals are normalized by the existing deferred-wait owner. The wait writer reports whether its locked conditional update saved the receipt, and only that successful path can produce the internal `PersistedExecutionWaitRefusal` subtype. It remains an HTTP409 with the original message and details. Receipt recovery parks that outcome instead of creating failure debt; the committed first/changed reason still grants its one stored30-second grace. Unrelated409s, failed writes and unexpected provider/infrastructure errors retain failure backoff. This adds no admission policy and changes no native/task-chat ownership. Production-path controls use actual task Retry, heartbeat, PostgreSQL budget/lifecycle services and a child process writing saved stdout. The lifecycle host completion boundary is controlled explicitly. One set of real budget/agent refusal fixtures consumes its finite grace and becomes quiet for25 minutes of timer advancement with zero driver SQL and no execution. Independent warm-worker fixtures keep the real PostgreSQL clock throughout admission, actual gate-producer release, controller authority and child-process exit, proving exactly one saved execution. Long-idle and provider-release evidence are separate; no clock reset or restart is used to claim warm recovery. A PostgreSQL trigger fault proves failed wait persistence retains recovery backoff, and an unrelated unbranded409 remains a failure. + +Project budget admission locks and re-reads the company-scoped project before publishing pause intent. An already-paused project emits no new eligibility hint and preserves its pause reason and timestamp, including a manual pause. The actual transition still publishes dispatch and deferred eligibility intent before updating the project in the accounting transaction. Concurrent admission checks serialize through that authority. Project-scoped saved Retry controls cover its one grace followed by25 minutes with zero driver SQL, plus a separate real-clock budget increase through one succeeded process with saved stdout. diff --git a/server/src/__tests__/budgets-service.test.ts b/server/src/__tests__/budgets-service.test.ts index 0ea5a94dfc6..1b2495cf7b6 100644 --- a/server/src/__tests__/budgets-service.test.ts +++ b/server/src/__tests__/budgets-service.test.ts @@ -1,4 +1,5 @@ import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { DELIVERY_QUEUES, subscribeDeliveryWork } from "../services/delivery-work-notifications.js"; import { randomUUID } from "node:crypto"; import { eq } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; @@ -264,6 +265,31 @@ describeEmbeddedPostgres("budgetService release gate enforcement", () => { }); }); + it.each([false, true])("only signals a real project budget pause with concurrent admission checks (already paused: %s)", async alreadyPaused => { + const { companyId, agentId, projectId } = await createBudgetFixture(); + await db.insert(budgetPolicies).values({ companyId, scopeType: "project", scopeId: projectId, + metric: "billed_cents", windowKind: "lifetime", amount: 100, hardStopEnabled: true, notifyEnabled: false }); + await insertCostEvent({ companyId, agentId, projectId, costCents: 125 }); + const pausedAt = new Date("2020-01-01T00:00:00.000Z"); + if (alreadyPaused) await db.update(projects).set({ pausedAt, pauseReason: "manual" }).where(eq(projects.id, projectId)); + const dispatch = vi.fn(), deferred = vi.fn(); + const unsubscribe = [subscribeDeliveryWork(db, DELIVERY_QUEUES.queuedRunDispatch, dispatch), + subscribeDeliveryWork(db, DELIVERY_QUEUES.deferredExecutionReceipts, deferred)]; + try { + const service = budgetService(db); + const blocks = await Promise.all(Array.from({ length: 4 }, () => service.getInvocationBlock(companyId, agentId, { projectId }))); + expect(blocks.every(block => block?.scopeType === "project" && block.scopeId === projectId)).toBe(true); + const [paused] = await db.select().from(projects).where(eq(projects.id, projectId)); + expect(paused).toMatchObject({ pauseReason: alreadyPaused ? "manual" : "budget", pausedAt: alreadyPaused ? pausedAt : expect.any(Date) }); + expect(dispatch).toHaveBeenCalledTimes(alreadyPaused ? 0 : 1); + expect(deferred).toHaveBeenCalledTimes(alreadyPaused ? 0 : 1); + await Promise.all(Array.from({ length: 4 }, () => service.getInvocationBlock(companyId, agentId, { projectId }))); + expect((await db.select().from(projects).where(eq(projects.id, projectId)))[0]).toEqual(paused); + expect(dispatch).toHaveBeenCalledTimes(alreadyPaused ? 0 : 1); + expect(deferred).toHaveBeenCalledTimes(alreadyPaused ? 0 : 1); + } finally { unsubscribe.forEach(stop => stop()); } + }); + it("hard-stops project work until a valid budget raise resumes it and overview reconciles ledger spend", async () => { const { companyId, agentId, projectId } = await createBudgetFixture(); const cancelWorkForScope = vi.fn().mockResolvedValue(undefined); diff --git a/server/src/__tests__/deferred-execution-receipt-work.test.ts b/server/src/__tests__/deferred-execution-receipt-work.test.ts index 18af0f94f97..78c56ff9a83 100644 --- a/server/src/__tests__/deferred-execution-receipt-work.test.ts +++ b/server/src/__tests__/deferred-execution-receipt-work.test.ts @@ -13,7 +13,7 @@ import { issueTreeControlService } from "../services/issue-tree-control.js"; import { randomUUID } from "node:crypto"; import { and, eq, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; -import { agents, agentWakeupRequests, authUsers, chatActions, chatEndpoints, toolApplications, toolConnections, companies, companyMemberships, costEvents, createDb, reconcileDatabaseWork, heartbeatRuns, issueComments, issues } from "@paperclipai/db"; +import { agents, agentWakeupRequests, authUsers, chatActions, chatEndpoints, toolApplications, toolConnections, companies, companyMemberships, costEvents, createDb, projects, reconcileDatabaseWork, heartbeatRuns, issueComments, issues } from "@paperclipai/db"; import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; @@ -142,17 +142,22 @@ describe("deferred receipt durable dispatch", () => { expect(await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.contextSnapshot}->>'taskRetryRequestId' = ${receipt.requestId}`)).toHaveLength(1); await quiet(); }); - async function blockedRetry(gate: "budget" | "paused agent", clocked = false) { + async function blockedRetry(gate: "budget" | "project budget" | "paused agent", clocked = false) { const f = await fixture(), lifecycle = createAgentLifecycle(db), budgets = budgetService(db); // Host lifecycle completion is controlled; all state, admission and the // eventual process adapter execution use their production implementations. const lifecycleWorker = configureAgentLifecycle(db, { requiredPluginIds: async () => [], runHost: async () => "complete", runPlugin: async () => "complete" }); lifecycleWorkers.add(lifecycleWorker); - const policy = { scopeType: "agent" as const, scopeId: f.agent.id, amount: 100, + const projectId = gate === "project budget" ? randomUUID() : null; + if (projectId) { + await db.insert(projects).values({ id: projectId, companyId: f.companyId, name: "Retry budget project" }); + await db.update(issues).set({ projectId }).where(eq(issues.id, f.issue.id)); + } + const policy = { scopeType: projectId ? "project" as const : "agent" as const, scopeId: projectId ?? f.agent.id, amount: 100, metric: "billed_cents" as const, windowKind: "lifetime" as const, hardStopEnabled: true, notifyEnabled: false }; - if (gate === "budget") { - await db.insert(costEvents).values({ companyId: f.companyId, agentId: f.agent.id, + if (gate !== "paused agent") { + await db.insert(costEvents).values({ companyId: f.companyId, agentId: f.agent.id, projectId, provider: "test", model: "receipt-proof", costCents: 200, occurredAt: new Date() }); await budgets.upsertPolicy(f.companyId, policy, f.actorId); } else await lifecycle.pauseAgent(f.agent.id); @@ -166,18 +171,18 @@ describe("deferred receipt durable dispatch", () => { expect(request.accepted).toBe(true); const read = async () => (await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, request.requestId!)))[0]; await vi.waitFor(async () => expect((await read()).payload?.executionWait).toMatchObject({ - reason: gate === "budget" ? "budget.blocked" : "agent.not_invokable", + reason: gate !== "paused agent" ? "budget.blocked" : "agent.not_invokable", }), { timeout: 10_000 }); const refusals: unknown[] = []; await heartbeat.resumeExecutionWaitComments({ companyId: f.companyId, requestId: request.requestId!, onError: error => refusals.push(error) }); expect(refusals).toHaveLength(1); expect(refusals[0]).toBeInstanceOf(HttpError); expect(refusals[0]).toMatchObject({ status: 409, - details: gate === "budget" ? { scopeType: "agent", scopeId: f.agent.id } : { status: "paused", reason: "paused" } }); + details: gate !== "paused agent" ? { scopeType: policy.scopeType, scopeId: policy.scopeId } : { status: "paused", reason: "paused" } }); return { f, lifecycle, budgets, policy, lifecycleWorker, request, read }; } - it.each(["budget", "paused agent"] as const)("parks real task Retry behind unchanged %s without recurring SQL", async gate => { + it.each(["budget", "project budget", "paused agent"] as const)("parks real task Retry behind unchanged %s without recurring SQL", async gate => { const { request, read } = await blockedRetry(gate, true); // The clock was installed before worker registration and every timer. // No execution exists while we observe actual driver SQL across25min. @@ -190,11 +195,11 @@ describe("deferred receipt durable dispatch", () => { expect(await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.contextSnapshot}->>'taskRetryRequestId' = ${request.requestId}`)).toHaveLength(0); }); - it.each(["budget", "paused agent"] as const)("releases real %s Retry into exactly one saved process execution", async gate => { + it.each(["budget", "project budget", "paused agent"] as const)("releases real %s Retry into exactly one saved process execution", async gate => { const { f, lifecycle, budgets, policy, lifecycleWorker, request, read } = await blockedRetry(gate); // This independent warm-worker fixture stays on the real PostgreSQL clock // through gate release, controller authority and actual child-process exit. - if (gate === "budget") await budgets.upsertPolicy(f.companyId, { ...policy, amount: 1000 }, f.actorId); + if (gate !== "paused agent") await budgets.upsertPolicy(f.companyId, { ...policy, amount: 1000 }, f.actorId); else await lifecycle.resumeAgent(f.agent.id); await lifecycleWorker.process(f.agent.id); let runId: string | null = null; diff --git a/server/src/services/budgets.ts b/server/src/services/budgets.ts index f7eacd7af55..9a18c8069a6 100644 --- a/server/src/services/budgets.ts +++ b/server/src/services/budgets.ts @@ -302,6 +302,11 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica } if (policy.scopeType === "project") { + // Admission rechecks the same hard stop. Lock current eligibility before + // publishing intent so a no-op pause cannot reactivate its own receipts. + const [project] = await db.select({ pausedAt: projects.pausedAt }).from(projects) + .where(and(eq(projects.id, policy.scopeId), eq(projects.companyId, policy.companyId))).for("update"); + if (!project || project.pausedAt !== null) return; await notifyRunDispatch(db, { companyId: policy.companyId }); await notifyDeferredExecutionEligibility(db, { companyId: policy.companyId }); await db @@ -311,7 +316,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica pausedAt: now, updatedAt: now, }) - .where(and(eq(projects.id, policy.scopeId), isNull(projects.pausedAt))); + .where(and(eq(projects.id, policy.scopeId), eq(projects.companyId, policy.companyId), isNull(projects.pausedAt))); return; }