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/doc/plans/2026-10-30-status-cards-sleep.md b/doc/plans/2026-10-30-status-cards-sleep.md index 3f3b8933cbc..97566de3bc4 100644 --- a/doc/plans/2026-10-30-status-cards-sleep.md +++ b/doc/plans/2026-10-30-status-cards-sleep.md @@ -155,3 +155,8 @@ closed-range evaluations perform one conversion, not another minute walk. Both caches use bounded LRU retention and hold no company/card data. Tests prove eviction, skipped/repeated hours, half-hour DST, date-line changes, rollback, strict boundaries, and actual twenty-card source delivery followed by quiet. + + +Refresh decisions use the card's PostgreSQL row revision as a private compare-and-swap fence. If a card changes during watched-source reads, the owner rereads it and reconsiders the current policy, for at most three attempts. A replacement future interval deadline remains unchanged; a replacement manual policy does not run automatically. The same fence protects no-change, paused-policy and generation claims. Existing issue idempotency, generation ownership and losing-claim cancellation still apply. The row revision is not exposed through card APIs or shared scheduling contracts. + +Company source hints do not preempt a future deadline owned by a failed evaluation (`retry`). Membership and policy deadlines remain eligible for source-driven rechecks, and actual due retries still run. An explicit policy edit can replace the retry through the existing card mutation path. Real PostgreSQL regressions hold the actual search while committing an ordinary policy edit, then verify the saved interval still fires. Persistent search failures stay quiet through issue hints and restart, retry at their saved deadline, and recover through an ordinary policy edit. 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/__tests__/status-card-evaluation-work.test.ts b/server/src/__tests__/status-card-evaluation-work.test.ts index d012a239efd..7c0f010949d 100644 --- a/server/src/__tests__/status-card-evaluation-work.test.ts +++ b/server/src/__tests__/status-card-evaluation-work.test.ts @@ -588,6 +588,88 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu expect(wakeup).toHaveBeenCalledOnce(); }); + it.each([false, true])("rechecks current policy after a committed edit during held search (changes=%s)", async changed => { + const created = await card(); + if (changed) await issueService(db).create(companyId, { title: "Policy race blocker", status: "blocked", priority: "medium" }); + const entered = Promise.withResolvers(), release = Promise.withResolvers(); + const factory = companySearch.companySearchService; + let hold = true; + const searches = vi.fn(); + vi.spyOn(companySearch, "companySearchService").mockImplementation(owner => { + const service = factory(owner), actual = service.search; + service.search = async (...args) => { + searches(); + const result = await actual(...args); + if (hold) { hold = false; entered.resolve(); await release.promise; } + return result; + }; + return service; + }); + const service = statusCardEvaluationService(db, { wakeup }); + const pass = service.run(new AbortController().signal); + await entered.promise; + const cards = statusCardService(db); + let scheduled!: Date; + try { + scheduled = (await cards.update((await cards.getById(created.id))!, { refreshPolicy: policy("interval") }, actor)).nextEvalAt!; + } finally { release.resolve(); } + await pass; + const current = (await cards.getById(created.id))!; + expect(current.refreshPolicy.mode).toBe("interval"); + expect(current.nextEvalAt!.getTime()).toBe(scheduled.getTime()); + expect(current.nextEvalReason).toBe("refresh"); + expect(searches).toHaveBeenCalledOnce(); + expect(wakeup).not.toHaveBeenCalled(); + const intervalCard = (await cards.getById(created.id))!; + await issueService(db).create(companyId, { title: "Next interval blocker", status: "blocked", priority: "medium" }); + vi.setSystemTime(intervalCard.nextEvalAt!); + await service.run(new AbortController().signal); + expect(wakeup).toHaveBeenCalledOnce(); + }); + + it("keeps future failure retries fenced from ordinary company hints and permits explicit policy recovery", async () => { + const created = await card(); + const factory = companySearch.companySearchService; + let failing = true; + const searches = vi.fn(); + vi.spyOn(companySearch, "companySearchService").mockImplementation(owner => { + const service = factory(owner), actual = service.search; + service.search = async (...args) => { searches(); if (failing) throw new Error("Controlled search unavailable"); return actual(...args); }; + return service; + }); + const service = statusCardEvaluationService(db, { wakeup }); + await service.run(new AbortController().signal); + const failed = (await statusCardService(db).getById(created.id))!; + expect(failed.nextEvalReason).toBe("retry"); + const retryAt = failed.nextEvalAt!.getTime(); + expect(searches).toHaveBeenCalledOnce(); + for (let index = 0; index < 5; index++) { + await issueService(db).create(companyId, { title: `Retry blocker ${index}`, status: "blocked", priority: "medium" }); + service.changed(new Set([`changes:${companyId}`])); + await service.run(new AbortController().signal); + } + expect(searches).toHaveBeenCalledOnce(); + expect((await statusCardService(db).getById(created.id))!.nextEvalAt!.getTime()).toBe(retryAt); + const restarted = statusCardEvaluationService(db, { wakeup }); + restarted.changed(new Set([`changes:${companyId}`])); + await restarted.run(new AbortController().signal); + expect(searches).toHaveBeenCalledOnce(); + expect(await restarted.next()).toBe(retryAt); + vi.setSystemTime(retryAt); + await restarted.run(new AbortController().signal); + expect(searches).toHaveBeenCalledTimes(2); + const cards = statusCardService(db); + failing = false; + await cards.update((await cards.getById(created.id))!, { refreshPolicy: { ...policy(), debounceSeconds: 1 } }, actor); + await restarted.run(new AbortController().signal); + expect(searches).toHaveBeenCalledTimes(3); + const pending = (await cards.getById(created.id))!; + expect(pending.nextEvalReason).toBe("policy"); + vi.setSystemTime(pending.nextEvalAt!); + await restarted.run(new AbortController().signal); + expect(wakeup).toHaveBeenCalledOnce(); + }); + it("retains the durable five-minute retry after evaluation failure across restart and drain", async () => { const created = await card(); await db.update(statusCards).set({ queries: [{ scope: "invalid" } as never], nextEvalAt: new Date() }).where(eq(statusCards.id, created.id)); diff --git a/server/src/services/budgets.ts b/server/src/services/budgets.ts index cd7bc4c5c48..1b31bcf6994 100644 --- a/server/src/services/budgets.ts +++ b/server/src/services/budgets.ts @@ -303,6 +303,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 @@ -312,7 +317,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; } diff --git a/server/src/services/status-cards.ts b/server/src/services/status-cards.ts index d4660227c28..862d9d64ec4 100644 --- a/server/src/services/status-cards.ts +++ b/server/src/services/status-cards.ts @@ -643,14 +643,17 @@ export function statusCardService( ? membershipAt : interval; } - async function requestRefresh(cardId: string, input: { + async function refreshCurrentCard(cardId: string, input: { full?: boolean; trigger?: "manual" | "interval" | "reactive" | "restore"; actor?: StatusCardActor; now?: Date; - } = {}) { - const card = await getById(cardId); - if (!card) throw notFound("Status card not found"); + } = {}, superseded = false) { + const [source] = await db.select({ card: statusCards, revision: sql`${statusCards}.xmin::text` }) + .from(statusCards).where(eq(statusCards.id, cardId)); + if (!source) throw notFound("Status card not found"); + const { card, revision } = source; + const currentRevision = sql`${statusCards}.xmin::text = ${revision}`; if (card.archivedAt) throw unprocessable("Archived status cards cannot be refreshed"); if (card.queries.length === 0) throw conflict("Compile the status-card query before refreshing it"); if (card.generatingIssueId) { @@ -663,6 +666,10 @@ export function statusCardService( } const now = input.now ?? new Date(); + if (superseded && input.trigger !== undefined && input.trigger !== "manual" && input.trigger !== "restore" + && (card.refreshPolicy.mode === "manual" || (card.refreshPolicy.mode === "interval" && card.nextEvalAt && card.nextEvalAt > now))) { + return { card, generatingIssue: null, alreadyGenerating: false, enqueued: false }; + } let membershipAt: Date | null = null; const snapshot = await executeQueries(card, {}, at => { if (at && (!membershipAt || at < membershipAt)) membershipAt = at; }); const fingerprint = buildStatusCardFingerprint(snapshot); @@ -679,8 +686,9 @@ export function statusCardService( state: "active", nextEvalAt, nextEvalReason: nextEvalAt ? card.refreshPolicy.mode === "reactive" ? "membership" : "refresh" : null, - }).where(eq(statusCards.id, card.id)).returning()); - return { card: next!, generatingIssue: null, alreadyGenerating: false, enqueued: false }; + }).where(and(eq(statusCards.id, card.id), currentRevision)).returning()); + if (!next) return undefined; + return { card: next, generatingIssue: null, alreadyGenerating: false, enqueued: false }; } const hourAgo = new Date(now.getTime() - 60 * 60 * 1000); @@ -712,8 +720,9 @@ export function statusCardService( completedUpdateTimes: recent.filter(row => row.kind !== "compile" && row.finishedAt).map(row => row.startedAt), tokensToday: Number(daily[0]?.tokens ?? 0), }), - }).where(eq(statusCards.id, card.id)).returning()); - return { card: next!, generatingIssue: null, alreadyGenerating: false, enqueued: false }; + }).where(and(eq(statusCards.id, card.id), currentRevision)).returning()); + if (!next) return undefined; + return { card: next, generatingIssue: null, alreadyGenerating: false, enqueued: false }; } const history = await listUpdates(card.id); @@ -765,13 +774,14 @@ export function statusCardService( nextEvalAt, nextEvalReason: nextEvalAt ? card.refreshPolicy.mode === "reactive" ? "membership" : "refresh" : null, failureReason: null, - }).where(and(eq(statusCards.id, card.id), isNull(statusCards.archivedAt), or(isNull(statusCards.generatingIssueId), priorGenerationPredicate))).returning()); + }).where(and(eq(statusCards.id, card.id), isNull(statusCards.archivedAt), currentRevision, or(isNull(statusCards.generatingIssueId), priorGenerationPredicate))).returning()); if (!next) { const winner = await getById(card.id); if (!winner?.generatingIssueId) { if (!TERMINAL_ISSUE_STATUSES.has(generationIssue!.status)) { await issuesSvc().update(generationIssue!.id, { status: "cancelled" }); } + if (winner && !winner.archivedAt) return undefined; throw conflict("Status-card refresh claim was lost"); } if (generationIssue!.id !== winner.generatingIssueId && !TERMINAL_ISSUE_STATUSES.has(generationIssue!.status)) { @@ -794,6 +804,14 @@ export function statusCardService( return { card: next, generatingIssue: generationIssue!, alreadyGenerating: deduplicated, enqueued: true, kind, changes }; } + async function requestRefresh(cardId: string, input: Parameters[1] = {}) { + for (let attempt = 0; attempt < 3; attempt += 1) { + const result = await refreshCurrentCard(cardId, input, attempt > 0); + if (result) return result; + } + throw conflict("Status-card refresh was superseded by concurrent edits"); + } + async function tickDueStatusCards(now = new Date(), reactiveCompanies?: ReadonlySet | "changed", reconstruct = false) { const changedScope = reactiveCompanies === "changed" ? sql`true` : reactiveCompanies?.size ? inArray(statusCards.companyId, [...reactiveCompanies]) : sql`false`; @@ -811,7 +829,9 @@ export function statusCardService( .innerJoin(companies, eq(companies.id, statusCards.companyId)) .where(and(eq(companies.status, "active"), isNull(statusCards.archivedAt), isNull(statusCards.generatingIssueId), or(and(isNotNull(statusCards.nextEvalAt), lte(statusCards.nextEvalAt, now)), - and(reactiveScope, ne(statusCards.state, "error"), ne(statusCards.state, "compiling"), sql`${statusCards.refreshPolicy}->>'mode' = 'reactive'`, sql`jsonb_array_length(${statusCards.queries}) > 0`)))) + and(reactiveScope, + or(isNull(statusCards.nextEvalReason), ne(statusCards.nextEvalReason, "retry"), isNull(statusCards.nextEvalAt), lte(statusCards.nextEvalAt, now)), + ne(statusCards.state, "error"), ne(statusCards.state, "compiling"), sql`${statusCards.refreshPolicy}->>'mode' = 'reactive'`, sql`jsonb_array_length(${statusCards.queries}) > 0`)))) .then(rows => rows.map(row => row.card)); const enqueued: Array<{ cardId: string; generatingIssue: typeof issues.$inferSelect }> = []; let evaluated = 0;