Merge branch 'codex/sleep-stack-07' into codex/sleep-stack-01

* codex/sleep-stack-07:
  fix(ui): invalidate pending startup work on sign-out
  fix(server): park persisted deferred admission refusals

server/src/__tests__/deferred-execution-receipt-work.test.ts

@@ -1,7 +1,11 @@
+import { conflict, HttpError } from "../errors.js";
 import { setImmediate as immediate } from "node:timers/promises";
 import { DELIVERY_QUEUES, subscribeDeliveryWork } from "../services/delivery-work-notifications.js";
+import { instanceSettingsService } from "../services/instance-settings.js";
+import { budgetService } from "../services/budgets.js";
+import { configureAgentLifecycle, createAgentLifecycle } from "../services/agent-lifecycle.js";
 import { accessService } from "../services/access.js";
-import { writeDeferredExecutionWait } from "../services/deferred-execution-wait.js";
+import { PersistedExecutionWaitRefusal, writeDeferredExecutionWait } from "../services/deferred-execution-wait.js";
 import { createQueuedCommentIssueLockWriter } from "../modules/wake-queue/adapters/queued-comment-postgres.js";
 import { createHeartbeatQueue } from "../services/heartbeat/queue.js";
 import { createHeartbeatQueueTestDependencies } from "./helpers/heartbeat-queue.js";
@@ -9,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, createDb, reconcileDatabaseWork, heartbeatRuns, issueComments, issues } from "@paperclipai/db";
+import { agents, agentWakeupRequests, authUsers, chatActions, chatEndpoints, toolApplications, toolConnections, companies, companyMemberships, costEvents, createDb, 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";
@@ -28,18 +32,23 @@ describe("deferred receipt durable dispatch", () => {
   let coordinator: ReturnType<typeof createDeliveryWorkCoordinator>;
   let drain = false;
   const errors: unknown[] = [];
+  const errorTimes: number[] = [];
+  const statements: string[] = [];
   const passes = new Set<Promise<unknown>>();
+  const lifecycleWorkers = new Set<ReturnType<typeof configureAgentLifecycle>>();
   beforeAll(async () => {
     const external = resolveExternalTestDatabaseUrl("PAPERCLIP_DEFERRED_RECEIPT_TEST_DATABASE_URL");
     if (!external) temporary = await startEmbeddedPostgresTestDatabase("deferred-receipts-");
     db = createDb(external ?? temporary!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 });
+    db.$client.options.onnotice = () => {};
+    db.$client.options.debug = (_connection, query) => { statements.push(query); };
     await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`)));
   });
   beforeEach(async () => {
-    await db.execute(sql`truncate companies cascade`); drain = false; errors.length = 0;
+    await db.execute(sql`truncate companies cascade`); drain = false; errors.length = 0; errorTimes.length = 0;
     heartbeat = heartbeatService(db);
     coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !drain, canReconcile: () => true,
-      onError: error => errors.push(error) });
+      onError: error => { errors.push(error); errorTimes.push(Date.now()); } });
     const register = coordinator.register.bind(coordinator);
     coordinator.register = (queue, task) => register(queue, { ...task, run: signal => {
       const pass = Promise.resolve(task.run(signal)); passes.add(pass);
@@ -48,6 +57,7 @@ describe("deferred receipt durable dispatch", () => {
   });
   afterEach(async () => {
     await coordinator.stop(); await heartbeat.drainActiveRunExecutions();
+    await Promise.all([...lifecycleWorkers].map(worker => worker.stop())); lifecycleWorkers.clear();
     vi.useRealTimers(); vi.restoreAllMocks(); expect(errors).toEqual([]);
   });
   afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); });
@@ -72,7 +82,9 @@ describe("deferred receipt durable dispatch", () => {
     const reads = vi.spyOn(db, "select"), executes = vi.spyOn(db, "execute");
     const selected = reads.mock.calls.length, executed = executes.mock.calls.length;
     if (!vi.isFakeTimers()) vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
-    await vi.advanceTimersByTimeAsync(20 * 60_000);
+    const count = statements.length;
+    await vi.advanceTimersByTimeAsync(25 * 60_000);
+    expect(statements).toHaveLength(count);
     expect(reads).toHaveBeenCalledTimes(selected); expect(executes).toHaveBeenCalledTimes(executed);
   }
   async function finish(issueId: string) {
@@ -130,6 +142,156 @@ 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) {
+    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,
+      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,
+        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);
+    await lifecycleWorker.process(f.agent.id);
+    if (clocked) vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
+    await heartbeat.registerDeferredReceiptWork(coordinator, () => !drain).ready;
+    await heartbeat.registerRunDispatchWork(coordinator, () => !drain).ready;
+    await heartbeat.registerScheduledRetryCancellationWork(coordinator, () => !drain).ready;
+    const request = await requestTaskRetry({ db, companyId: f.companyId, issueId: f.issue.id, actorId: f.actorId,
+      observedRunId: f.source.id, idempotencyKey: randomUUID() });
+    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",
+    }), { 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" } });
+    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 => {
+    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.
+    for (let n = 0; n < 3; n++) {
+      await vi.advanceTimersByTimeAsync(31_000); await Promise.allSettled([...passes]); await immediate();
+    }
+    expect(errors).toEqual([]);
+    await quiet();
+    expect(await read()).toMatchObject({ status: "deferred_issue_execution", runId: null });
+    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 => {
+    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);
+    else await lifecycle.resumeAgent(f.agent.id);
+    await lifecycleWorker.process(f.agent.id);
+    let runId: string | null = null;
+    await vi.waitFor(async () => {
+      const wake = await read(); expect(wake.runId).toBeTruthy(); runId = wake.runId;
+    }, { timeout: 40_000 });
+    await vi.waitFor(async () => {
+      await heartbeat.drainActiveRunExecutions();
+      const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId!));
+      expect(run.status, JSON.stringify({ error: run.error, code: run.errorCode })).toBe("succeeded");
+      expect(run.stdoutExcerpt).toContain("receipt-proof");
+    }, { timeout: 15_000 });
+    expect(await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.contextSnapshot}->>'taskRetryRequestId' = ${request.requestId}`)).toHaveLength(1);
+    await finish(f.issue.id); await quiet();
+  });
+
+  it.each(["missing", "cancelled"] as const)("preserves ordinary409 when a %s receipt cannot save the gate", async status => {
+    const f = await fixture(), requestId = randomUUID();
+    if (status === "cancelled") await db.insert(agentWakeupRequests).values({ id: requestId, companyId: f.companyId,
+      agentId: f.agent.id, source: "on_demand", status, payload: { issueId: f.issue.id } });
+    const deps = createHeartbeatQueueTestDependencies(db), queue = createHeartbeatQueue(db, deps);
+    deps.getAgent.mockResolvedValue(f.agent);
+    deps.instanceSettings.getExperimental.mockImplementation(() => instanceSettingsService(db).getExperimental());
+    for (const gate of ["budget", "agent"] as const) {
+      deps.budgets.getInvocationBlock.mockResolvedValue(gate === "budget"
+        ? { scopeType: "agent", scopeId: f.agent.id, scopeName: f.agent.name, reason: "Budget blocked" } : null);
+      deps.getAgentInvokability.mockResolvedValue({ invokable: false, message: "Paused", reason: "paused", invalidOrgChain: false });
+      const error = await queue.enqueueWakeup(f.agent.id, { source: "on_demand", reason: "task_retry",
+        requestedByActorType: "user", requestedByActorId: f.actorId, payload: { issueId: f.issue.id },
+        contextSnapshot: { issueId: f.issue.id, responsibleUserId: f.actorId } }, requestId).catch(cause => cause);
+      expect(error).toBeInstanceOf(HttpError);
+      expect(error).not.toBeInstanceOf(PersistedExecutionWaitRefusal);
+      expect(error).toMatchObject({ status: 409, message: gate === "budget" ? "Budget blocked" : "Paused" });
+    }
+    const rows = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, requestId));
+    expect(rows).toHaveLength(status === "missing" ? 0 : 1);
+    if (status === "cancelled") expect(rows[0]).toMatchObject({ status: "cancelled", runId: null, payload: { issueId: f.issue.id } });
+  });
+
+  it("retains failure backoff for a real wait-persistence database error", async () => {
+    const f = await fixture();
+    await createAgentLifecycle(db).pauseAgent(f.agent.id);
+    const request = await requestTaskRetry({ db, companyId: f.companyId, issueId: f.issue.id, actorId: f.actorId,
+      observedRunId: f.source.id, idempotencyKey: randomUUID() });
+    await db.execute(sql`create function c1_refusal_fault() returns trigger language plpgsql as $$
+      begin if new.reason = 'task_retry' and new.payload ? 'executionWait' then
+        raise exception 'C1 injected wait persistence failure'; end if; return new; end $$`);
+    await db.execute(sql`create trigger c1_refusal_fault before update on agent_wakeup_requests
+      for each row execute function c1_refusal_fault()`);
+    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
+    try {
+      await heartbeat.registerDeferredReceiptWork(coordinator, () => !drain).ready;
+      await vi.waitFor(() => expect(errors).toHaveLength(1));
+      const failure = errors[0] as AggregateError;
+      expect(failure).toBeInstanceOf(AggregateError);
+      expect(String(failure.errors[0]?.errors[0]?.cause?.message)).toContain("C1 injected wait persistence failure");
+      const due = errorTimes[0] + 30_000;
+      await Promise.allSettled([...passes]); await immediate();
+      await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBe(due));
+      expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, request.requestId!)))[0].payload?.executionWait).toBeUndefined();
+      await vi.advanceTimersByTimeAsync(29_999);
+      expect(errors).toHaveLength(1);
+      await db.execute(sql`drop trigger c1_refusal_fault on agent_wakeup_requests`);
+      await db.execute(sql`drop function c1_refusal_fault()`);
+      errors.length = 0;
+      for (let n = 0; n < 3; n++) {
+        await vi.advanceTimersByTimeAsync(31_000); await Promise.allSettled([...passes]); await immediate();
+      }
+      expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, request.requestId!)))[0].payload?.executionWait)
+        .toMatchObject({ reason: "agent.not_invokable" });
+      await quiet();
+    } finally {
+      await db.execute(sql`drop trigger if exists c1_refusal_fault on agent_wakeup_requests`);
+      await db.execute(sql`drop function if exists c1_refusal_fault()`);
+    }
+  });
+
+  it("does not park an unrelated HTTP409 as a persisted refusal", async () => {
+    const f = await fixture();
+    const [wake] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agent.id,
+      source: "on_demand", reason: "task_retry", status: "deferred_issue_execution" }).returning();
+    const failure = conflict("Unexpected admission conflict");
+    let fail = true;
+    const resume = vi.fn(async (scope: { onError?: (error: unknown) => void }) => { if (fail) scope.onError?.(failure); });
+    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
+    await registerDeferredReceiptWork(db, coordinator, { resume, settleTaskChat: async () => {}, dispatchNative: async () => {},
+      schedulingSuppressed: async () => false }, () => !drain).ready;
+    expect(errors).toHaveLength(1);
+    expect((errors[0] as AggregateError).errors[0].errors).toEqual([failure]);
+    expect(coordinator.nextWakeAt()).toBe(Date.now() + 30_000);
+    expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wake.id)))[0].status).toBe("deferred_issue_execution");
+    errors.length = 0; fail = false;
+    await vi.advanceTimersByTimeAsync(30_000);
+    await vi.waitFor(() => expect(resume).toHaveBeenCalledTimes(2));
+    await quiet();
+  });
+
   it.each(["native claim", "execution wait"])("reconstructs an exact fresh %s deadline without probing early", async kind => {
     const f = await fixture(), updatedAt = new Date(), native = kind === "native claim";
     const [wake] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agent.id,

server/src/services/deferred-execution-wait.ts

@@ -1,24 +1,33 @@
+import { HttpError } from "../errors.js";
 import { and, eq, sql } from "drizzle-orm";
 import { agentWakeupRequests, type Db } from "@paperclipai/db";
 import { notifyDeferredExecutionWork } from "./deferred-execution-notifications.js";
 
+/** Admission has saved this refusal on the caller's durable deferred receipt.
+ * HTTP callers still receive the existing 409; receipt recovery waits for its
+ * owning gate producer instead of treating the refusal as infrastructure debt. */
+export class PersistedExecutionWaitRefusal extends HttpError {
+  constructor(message: string, details?: unknown) { super(409, message, details); }
+}
+
 export async function writeDeferredExecutionWait(db: Db, scope: { companyId: string; requestId: string },
   condition: Record<string, unknown>, options: { replace?: boolean } = {}) {
   return db.transaction(async tx => {
     const [current] = await tx.select({ payload: agentWakeupRequests.payload }).from(agentWakeupRequests)
       .where(and(eq(agentWakeupRequests.companyId, scope.companyId), eq(agentWakeupRequests.id, scope.requestId),
         eq(agentWakeupRequests.status, "deferred_issue_execution"))).for("update");
-    if (!current) return;
+    if (!current) return false;
     const previous = current.payload?.executionWait;
     const previousReason = previous && typeof previous === "object" && !Array.isArray(previous)
       ? (previous as Record<string, unknown>).reason : undefined;
     if (!previous || previousReason !== condition.reason) await notifyDeferredExecutionWork(tx, scope);
-    await tx.update(agentWakeupRequests).set({
+    const [saved] = await tx.update(agentWakeupRequests).set({
       payload: sql`jsonb_set(coalesce(${agentWakeupRequests.payload}, '{}'::jsonb), '{executionWait}',
         ${options.replace ? sql`'{}'::jsonb` : sql`coalesce(${agentWakeupRequests.payload}->'executionWait', '{}'::jsonb)`}
           || ${JSON.stringify(condition)}::jsonb)`,
       updatedAt: new Date(),
     }).where(and(eq(agentWakeupRequests.companyId, scope.companyId), eq(agentWakeupRequests.id, scope.requestId),
-      eq(agentWakeupRequests.status, "deferred_issue_execution")));
+      eq(agentWakeupRequests.status, "deferred_issue_execution"))).returning({ id: agentWakeupRequests.id });
+    return Boolean(saved);
   });
 }

server/src/services/heartbeat/deferred-receipt-work.ts

@@ -1,3 +1,4 @@
+import { PersistedExecutionWaitRefusal } from "../deferred-execution-wait.js";
 import { and, asc, eq, gt, inArray, isNull, or, sql } from "drizzle-orm";
 import { agentWakeupRequests, type Db } from "@paperclipai/db";
 import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js";
@@ -105,7 +106,11 @@ export function registerDeferredReceiptWork(
           onError: error => { errors.push(error); },
         });
         else if (kind === "task_chat") await owner.settleTaskChat(scope);
-        else await owner.resume({ ...scope, onError: error => { errors.push(error); } });
+        else {
+          const failed = (error: unknown) => { if (!(error instanceof PersistedExecutionWaitRefusal)) errors.push(error); };
+          try { await owner.resume({ ...scope, onError: failed }); }
+          catch (error) { failed(error); }
+        }
         if (errors.length) throw new AggregateError(errors, `Deferred receipt ${entry.id} could not settle`);
         return next === null ? null : { ...entry, dueAt: next };
       },

server/src/services/heartbeat/queue.ts

@@ -1,5 +1,5 @@
 import { notifyStatusCardIssueChange } from "../status-card-notifications.js";
-import { writeDeferredExecutionWait } from "../deferred-execution-wait.js";
+import { PersistedExecutionWaitRefusal, writeDeferredExecutionWait } from "../deferred-execution-wait.js";
 import { notifyDeferredExecutionWork } from "../deferred-execution-notifications.js";
 import type { DeferredReceiptRecoveryOptions } from "../deferred-execution-notifications.js";
 import { notifyRunDispatch } from "../run-dispatch-notifications.js";
@@ -2521,17 +2521,17 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
       skipReason: string,
       patch: Partial<typeof agentWakeupRequests.$inferInsert> = {},
       waitCondition?: Record<string, unknown>,
-    ) => {
+    ): Promise<{ created: boolean; waitSaved?: boolean }> => {
       // A temporary refusal must leave the handoff retryable, without burning
       // its one durable receipt ID on a skipped wake.
       if (caseRequest) return { created: false };
       if (executionWaitRequestId) {
-        await writeDeferredExecutionWait(db, { companyId: agent.companyId, requestId: executionWaitRequestId }, {
+        const waitSaved = await writeDeferredExecutionWait(db, { companyId: agent.companyId, requestId: executionWaitRequestId }, {
           reason: skipReason, message: patch.error ?? (skipReason === "issue_tree_hold_active"
             ? "This task is paused. Resume it to send your saved message."
             : "Waiting for task execution to be enabled. Your message is saved."),
         });
-        return { created: false };
+        return { created: false, waitSaved };
       }
       const request = {
         ...durableReceiptFields,
@@ -2803,31 +2803,30 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
       },
     );
     if (budgetBlock) {
-      await writeSkippedRequest("budget.blocked", { error: budgetBlock.reason }, {
+      const wait = await writeSkippedRequest("budget.blocked", { error: budgetBlock.reason }, {
         scopeType: budgetBlock.scopeType, scopeId: budgetBlock.scopeId,
       });
       if (caseRequest) return null;
-      throw conflict(budgetBlock.reason, {
-        scopeType: budgetBlock.scopeType,
-        scopeId: budgetBlock.scopeId,
-      });
+      const details = { scopeType: budgetBlock.scopeType, scopeId: budgetBlock.scopeId };
+      throw wait.waitSaved ? new PersistedExecutionWaitRefusal(budgetBlock.reason, details)
+        : conflict(budgetBlock.reason, details);
     }
 
     // Setup delays execution, but must not discard work accepted by a caller.
     const invokability = await getAgentInvokability(isAgentAwaitingSetup(agent) ? { ...agent, status: "idle" } : agent);
     if (!invokability.invokable) {
       if (caseRequest) return null;
+      let waitSaved = false;
       if (opts.requestedByActorType !== "user" || executionWaitRequestId) {
-        await writeSkippedRequest("agent.not_invokable", {
+        const wait = await writeSkippedRequest("agent.not_invokable", {
           error: invokability.message,
         }, { status: agent.status, reason: invokability.reason });
+        waitSaved = wait.waitSaved === true;
       }
-      throw conflict(invokability.message, {
-        status: agent.status,
-        reason: invokability.reason,
-        invalidOrgChain: invokability.invalidOrgChain,
-        ...invokability.details,
-      });
+      const details = { status: agent.status, reason: invokability.reason,
+        invalidOrgChain: invokability.invalidOrgChain, ...invokability.details };
+      throw waitSaved ? new PersistedExecutionWaitRefusal(invokability.message, details)
+        : conflict(invokability.message, details);
     }
 
     const policy = parseHeartbeatPolicy(agent);