test: integrate reviewed predecessor fixes for PR 18382

* commit '06c3c6e9d1fd3495643c82696b3de07fcc83c600':
  fix(server): fence status card refresh decisions and retry debt
  fix: park unchanged project budget execution waits

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);

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;

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<void>(), release = Promise.withResolvers<void>();
+    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));

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;
     }
 

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<string>`${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<typeof refreshCurrentCard>[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<string> | "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;