mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-01 23:43:16 +01:00
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 <noreply@paperclip.ing>
This commit is contained in:
1 parent
51d63d74f7
commit
af7b7d3316
4 files changed
+49
-11
No files matched your search
@@ -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.
|
||||
@@ -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);
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user