Merge branch 'codex/sleep-deferred-execution-receipts' into codex/sleep-live-run-resource-recovery

* codex/sleep-deferred-execution-receipts:
  Join receipt acceptance passes before advancing saved deadlines
  Retain moved receipt deadlines and wake released execution gates
  fix: scope comment-triggered recovery to the affected issue
  test: await issue recovery dispatch and cancellation settlement

doc/plans/2026-10-30-deferred-execution-sleep.md

@@ -17,3 +17,7 @@ Three scoped owners reconstruct only their actual durable receipt predicates, wi
 The generic run-dispatch notifier and broad issue review-state notifier deliberately do not fan out deferred eligibility. Queue bookkeeping may write them while processing its own saved input. Separate eligibility commits and settlement-only chat notifications avoid a self-restaging loop.
 
 C2 remains separate and mandatory: active output silence deadlines, remaining orphan/process/controller/remote authority and resource debts, legacy shutdown settlements, stranded agent status, and final aggregate chain removal. Retained workspace and migrated native finalization workers remain their existing owners.
+
+Review controls cover queue edits that move an already loaded grace deadline. Each due check re-reads current stored grace or claim time before admission. Subtree hold release, subtree restore, and member archival publish eligibility inside their own transactions. Member archival preserves the cancelled Retry tombstone created by an earlier user reassignment.
+
+A resume that first records a new execution-wait reason arms its existing 30-second grace once. Writing the same closed-gate reason again does not publish another hint. A clock obligation, an eligibility change, and an infrastructure failure therefore retain separate exits: stored deadline, owning committed producer, and bounded failure backoff. Refreshing a diagnostic timestamp alone does not create permanent polling behind a deliberate pause.

doc/plans/2026-10-30-issue-recovery-sleep.md

@@ -1,6 +1,6 @@
 # Issue recovery without periodic scans
 
-The issue recovery owners reconstruct their current obligations once at startup. Committing source writes invalidate only their company. The coordinator keeps scopes, per-issue deadlines, and per-entry failures in memory. Its pending check does not query the database.
+The issue recovery owners reconstruct their current obligations once at startup. Ordinary comment writes invalidate only the affected issue for assignment and lock recovery. Lifecycle, graph, run, configuration and eligibility changes invalidate their company. The coordinator keeps scopes, per-issue deadlines, and per-entry failures in memory. Its pending check does not query the database.
 
 | Owner | Durable truth and producer boundary | Deadline or event | Exit |
 | --- | --- | --- | --- |
@@ -11,6 +11,10 @@ The issue recovery owners reconstruct their current obligations once at startup.
 | Task watchdog | Active issue watchdogs and their current source subtree; normal watchdog upsert/disable; issue/comment/document/work-product and wait changes | Exact source creation plus the existing 15-second first-run grace. An incomplete scan retains the watchdog with error backoff. | Live path, open/reviewed reusable review, disabled configuration, or ineligible source. |
 | Stale issue lock | Existing checkout/execution run references; issue and run lifecycle changes | A committing ownership change, or the fresh owning controller's saved lease expiry. | No stale reference, a valid owner, or the existing guarded terminalization and lock clear. |
 
+Local comment hints do not broaden into an assignment check for every open task. A real PostgreSQL regression starts with1,000 open assigned issues, writes one comment through `issueService.addComment`, and observes exactly one assignment check; a subsequent company eligibility notification still checks all1,000. Restoring the previous comment hint makes that same regression check all1,000 after the comment. The test substitutes the expensive assignment decision only to count its invocations; it uses the real transaction, notification, coordinator and paged SQL. Separate producer acceptance cases exercise real recovery decisions and process execution.
+
+Parent/descendant disposition repair and shared execution references make status and run changes broader than a comment. Their existing notification boundaries deliberately retain company scope. Orphan blockers, active recovery actions, resolved dependencies and watchdogs also keep their company/subtree duties; an issue comment does not drop those independent obligations. The scope serializer selects known primitive fields and never serializes an entire insert or update object.
+
 Reconstruction uses keyset pages of at most100rows. Repeated company hints coalesce reconstruction for30seconds. Already-known entry deadlines preempt that spacing without an early probe on each comment. Scheduling suppression is read once per pass. A failed read keeps scoped backoff debt. The two subsidiary company repairs do not run inside the per-issue loop. Legacy direct aggregate API calls retain both repairs.
 
 The owner methods accept company and issue/watchdog identity. A narrowed lock check still reads other same-company references before using terminal-issue authority. A stale terminal alias cannot end an active issue's shared run. The controller lease owner supplies its fresh expiry; the existing write predicate revalidates the observed boot ID and database-clock expiry.

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

@@ -1,7 +1,15 @@
+import { setImmediate as immediate } from "node:timers/promises";
+import { DELIVERY_QUEUES, subscribeDeliveryWork } from "../services/delivery-work-notifications.js";
+import { accessService } from "../services/access.js";
+import { 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";
+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, heartbeatRuns, issues } from "@paperclipai/db";
+import { agents, agentWakeupRequests, authUsers, chatActions, chatEndpoints, toolApplications, toolConnections, companies, companyMemberships, 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";
@@ -20,6 +28,7 @@ describe("deferred receipt durable dispatch", () => {
   let coordinator: ReturnType<typeof createDeliveryWorkCoordinator>;
   let drain = false;
   const errors: unknown[] = [];
+  const passes = new Set<Promise<unknown>>();
   beforeAll(async () => {
     const external = resolveExternalTestDatabaseUrl("PAPERCLIP_DEFERRED_RECEIPT_TEST_DATABASE_URL");
     if (!external) temporary = await startEmbeddedPostgresTestDatabase("deferred-receipts-");
@@ -31,6 +40,11 @@ describe("deferred receipt durable dispatch", () => {
     heartbeat = heartbeatService(db);
     coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !drain, canReconcile: () => true,
       onError: error => errors.push(error) });
+    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);
+      void pass.then(() => passes.delete(pass), () => passes.delete(pass)); return pass;
+    } });
   });
   afterEach(async () => {
     await coordinator.stop(); await heartbeat.drainActiveRunExecutions();
@@ -61,6 +75,27 @@ describe("deferred receipt durable dispatch", () => {
     await vi.advanceTimersByTimeAsync(20 * 60_000);
     expect(reads).toHaveBeenCalledTimes(selected); expect(executes).toHaveBeenCalledTimes(executed);
   }
+  async function finish(issueId: string) {
+    await issueService(db).update(issueId, { status: "done" });
+    await vi.waitFor(async () => {
+      await Promise.allSettled([...passes]); await heartbeat.drainActiveRunExecutions();
+      expect(await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.status} in ('queued', 'running')`)).toHaveLength(0);
+    }, { timeout: 15_000 });
+    const saved = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.status, "scheduled_retry"));
+    if (!vi.isFakeTimers()) vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
+    if (saved.length) {
+      const due = Math.max(...saved.map(run => run.scheduledRetryAt!.getTime()));
+      while (Date.now() <= due) {
+        await vi.advanceTimersByTimeAsync(Math.min(500, due - Date.now() + 1));
+        await Promise.allSettled([...passes]); await immediate();
+      }
+    }
+    await vi.waitFor(async () => {
+      await vi.advanceTimersByTimeAsync(500); await Promise.allSettled([...passes]);
+      await heartbeat.drainActiveRunExecutions();
+      expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.status, "scheduled_retry"))).toHaveLength(0);
+    });
+  }
   it("is quiet after empty reconstruction and terminal receipt history", async () => {
     const f = await fixture();
     await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agent.id,
@@ -86,13 +121,7 @@ describe("deferred receipt durable dispatch", () => {
       expect(run.status, JSON.stringify({ error: run.error, code: run.errorCode })).toBe("succeeded");
       expect(run.stdoutExcerpt).toContain("receipt-proof");
     }, { timeout: 15_000 });
-    await issueService(db).update(f.issue.id, { status: "done" });
-    await heartbeat.drainActiveRunExecutions();
-    const saved = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.status, "scheduled_retry"));
-    expect(await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.status} in ('queued', 'running')`)).toHaveLength(0);
-    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
-    if (saved.length) await vi.advanceTimersByTimeAsync(Math.max(0, Math.max(...saved.map(run => run.scheduledRetryAt!.getTime())) - Date.now()) + 1);
-    await vi.waitFor(async () => expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.status, "scheduled_retry"))).toHaveLength(0));
+    await finish(f.issue.id);
     const linked = await db.select().from(agentWakeupRequests).where(and(eq(agentWakeupRequests.reason, "task_retry"), sql`${agentWakeupRequests.runId} is not null`));
     expect(linked).toHaveLength(2);
     expect(new Set(linked.map(row => row.runId))).toEqual(new Set([runId]));
@@ -208,6 +237,178 @@ describe("deferred receipt durable dispatch", () => {
       .toBe("issued"), { timeout: 5_000 });
     await quiet();
   });
+  it("retains a moved interrupt grace deadline after an ordinary saved queue reorder", async () => {
+    const f = await fixture(), initialAt = new Date();
+    const comments = await db.insert(issueComments).values(["first", "second"].map(body => ({
+      companyId: f.companyId, issueId: f.issue.id, body, authorType: "user", authorUserId: f.actorId,
+    }))).returning();
+    const payload = { issueId: f.issue.id, queuedCommentInterrupt: { actorId: f.actorId },
+      _paperclipWakeContext: { wakeCommentIds: comments.map(row => row.id) } };
+    const [wake] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agent.id,
+      source: "on_demand", status: "deferred_issue_execution", payload, updatedAt: initialAt }).returning();
+    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
+    const deps = createHeartbeatQueueTestDependencies(db), queue = createHeartbeatQueue(db, deps);
+    await registerDeferredReceiptWork(db, coordinator, { resume: scope => queue.resumeDeferredExecutionReceipts(scope),
+      settleTaskChat: async () => {}, dispatchNative: async () => {}, schedulingSuppressed: async () => false }, () => !drain).ready;
+    await vi.advanceTimersByTimeAsync(initialAt.getTime() + 10_000 - Date.now());
+    const movedAt = new Date();
+    const writer = createQueuedCommentIssueLockWriter(db, { syncCommentReferences: async () => {},
+      deleteCommentReferenceSource: async () => {}, syncCommentExternalObjectsSafely: async () => {} });
+    await writer.withLockedQueue({ issue: f.issue, queueId: wake.id,
+      actor: { actorType: "user", actorId: f.actorId, agentId: null, runId: null, agentApiKeyId: null } },
+      async (_locked, tx) => { await tx.updateWakeQueuedCommentIds({ wakeId: wake.id, payload,
+        ids: comments.map(row => row.id).reverse(), updatedAt: movedAt }); });
+    await vi.advanceTimersByTimeAsync(initialAt.getTime() + 30_000 - Date.now());
+    await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0));
+    expect(deps.resumeQueuedCommentInterrupt).not.toHaveBeenCalled();
+    expect(coordinator.nextWakeAt()).toBe(movedAt.getTime() + 30_000);
+    await vi.advanceTimersByTimeAsync(movedAt.getTime() + 30_000 - Date.now());
+    await vi.waitFor(() => expect(deps.resumeQueuedCommentInterrupt).toHaveBeenCalledExactlyOnceWith(f.companyId, wake.id));
+    await quiet();
+  });
+  it("reactivates a held task Retry when the authorized subtree hold is released without wakeAgents", async () => {
+    const f = await fixture(), tree = issueTreeControlService(db);
+    const actor = { actorType: "user" as const, actorId: f.actorId, userId: f.actorId };
+    const hold = await tree.createHold(f.companyId, f.issue.id, { mode: "pause", actor });
+    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.requestId).toBeTruthy();
+    await vi.waitFor(async () => {
+      const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, request.requestId!));
+      expect(wake.payload?.executionWait).toMatchObject({ reason: "issue_tree_hold_active" });
+      expect(wake.runId).toBeNull();
+    }, { timeout: 5_000 });
+    await vi.waitFor(() => { expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); }, { timeout: 40_000 });
+    await tree.releaseHold(f.companyId, f.issue.id, hold.hold.id, { actor, reason: "Resume saved Retry" });
+    let runId: string | null = null;
+    await vi.waitFor(async () => {
+      const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, request.requestId!));
+      expect(wake.runId, JSON.stringify(wake)).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("keeps a moved native dispatch error deadline instead of retrying at its old time", async () => {
+    const f = await fixture(), initialAt = new Date();
+    const [wake] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agent.id,
+      source: "automation", status: "queued", error: "Transient dispatch failure", updatedAt: initialAt,
+      requestedByActorType: "system", requestedByActorId: "native-status-committer" }).returning();
+    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
+    const dispatchNative = vi.fn(async () => {});
+    await registerDeferredReceiptWork(db, coordinator, { resume: async () => {}, settleTaskChat: async () => {},
+      dispatchNative, schedulingSuppressed: async () => false }, () => !drain).ready;
+    await vi.advanceTimersByTimeAsync(initialAt.getTime() + 10_000 - Date.now());
+    const movedAt = new Date();
+    await db.update(agentWakeupRequests).set({ updatedAt: movedAt }).where(eq(agentWakeupRequests.id, wake.id));
+    await vi.advanceTimersByTimeAsync(initialAt.getTime() + 30_000 - Date.now());
+    await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0));
+    expect(dispatchNative).not.toHaveBeenCalled(); expect(coordinator.nextWakeAt()).toBe(movedAt.getTime() + 30_000);
+    await vi.advanceTimersByTimeAsync(movedAt.getTime() + 30_000 - Date.now());
+    await vi.waitFor(() => expect(dispatchNative).toHaveBeenCalledTimes(1)); await quiet();
+  });
+  it("arms one new wait reason grace and stays quiet when that same gate is still closed", async () => {
+    const f = await fixture(); let reason = "operator_pause";
+    const resume = vi.fn(async (scope: { companyId?: string; requestId?: string }) => {
+      await writeDeferredExecutionWait(db, { companyId: scope.companyId!, requestId: scope.requestId! }, { reason });
+    });
+    await registerDeferredReceiptWork(db, coordinator, { resume, settleTaskChat: async () => {}, dispatchNative: async () => {},
+      schedulingSuppressed: async () => false }, () => !drain).ready;
+    const [wake] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agent.id,
+      source: "on_demand", status: "deferred_issue_execution", reason: "task_retry" }).returning();
+    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
+    for (const nextReason of ["operator_pause", "budget_pause"]) {
+      reason = nextReason;
+      await writeDeferredExecutionWait(db, { companyId: f.companyId, requestId: wake.id }, { reason });
+      const [current] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wake.id));
+      const due = current.updatedAt.getTime() + 30_000;
+      await vi.advanceTimersByTimeAsync(1_000);
+      await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBe(due));
+      const checked = resume.mock.calls.length;
+      await vi.advanceTimersByTimeAsync(due - Date.now());
+      await vi.waitFor(() => expect(resume).toHaveBeenCalledTimes(checked + 1));
+      await quiet();
+    }
+    expect(resume).toHaveBeenCalledTimes(2);
+  });
+  it("reactivates a saved Retry through ordinary restore cancelled task", async () => {
+    const f = await fixture(), actor = { actorType: "user" as const, actorId: f.actorId, userId: f.actorId };
+    const request = await requestTaskRetry({ db, companyId: f.companyId, issueId: f.issue.id, actorId: f.actorId,
+      observedRunId: f.source.id, idempotencyKey: randomUUID() });
+    const tree = issueTreeControlService(db);
+    const cancel = await tree.createHold(f.companyId, f.issue.id, { mode: "cancel", actor });
+    await tree.cancelIssueStatusesForHold(f.companyId, f.issue.id, cancel.hold.id);
+    const restore = await tree.createHold(f.companyId, f.issue.id, { mode: "restore", actor });
+    const commitEligibility = () => tree.restoreIssueStatusesForHold(f.companyId, f.issue.id, restore.hold.id, { actor });
+    await heartbeat.registerDeferredReceiptWork(coordinator, () => !drain).ready;
+    await heartbeat.registerRunDispatchWork(coordinator, () => !drain).ready;
+    await heartbeat.registerScheduledRetryCancellationWork(coordinator, () => !drain).ready;
+    await vi.waitFor(() => { expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); }, { timeout: 5_000 });
+    expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, request.requestId!)))[0].runId).toBeNull();
+    await commitEligibility();
+    let runId: string | null = null;
+    await vi.waitFor(async () => {
+      const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, request.requestId!));
+      expect(wake.runId, JSON.stringify(wake)).toBeTruthy(); runId = wake.runId;
+    }, { timeout: 15_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("member archive publishes committed eligibility while preserving a cancelled Retry tombstone", async () => {
+    const f = await fixture(), departing = `departing-${randomUUID()}`;
+    const request = await requestTaskRetry({ db, companyId: f.companyId, issueId: f.issue.id, actorId: f.actorId,
+      observedRunId: f.source.id, idempotencyKey: randomUUID() });
+    await db.insert(authUsers).values({ id: departing, name: "Departing member", email: `${departing}@example.test`,
+      createdAt: new Date(), updatedAt: new Date() });
+    const [member] = await db.insert(companyMemberships).values({ companyId: f.companyId, principalId: departing,
+      principalType: "user", status: "active", membershipRole: "member" }).returning();
+    await issueService(db).update(f.issue.id, { assigneeUserId: departing, assigneeAgentId: null });
+    const readIntent = async () => (await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, request.requestId!)))[0];
+    expect(await readIntent()).toMatchObject({ status: "cancelled", runId: null });
+    const deferred = vi.fn(), native = vi.fn();
+    const unsubscribes = [subscribeDeliveryWork(db, DELIVERY_QUEUES.deferredExecutionReceipts, deferred),
+      subscribeDeliveryWork(db, DELIVERY_QUEUES.nativeStatusWakeIntents, native)];
+    try {
+      await expect(db.transaction(async tx => {
+        await accessService(tx as unknown as typeof db).archiveMember(f.companyId, member.id,
+          { reassignment: { assigneeAgentId: f.agent.id } });
+        expect(deferred).not.toHaveBeenCalled(); expect(native).not.toHaveBeenCalled();
+        throw new Error("roll back member archive");
+      })).rejects.toThrow("roll back member archive");
+      expect(deferred).toHaveBeenCalledTimes(1); expect(native).toHaveBeenCalledTimes(1);
+      expect((await db.select().from(issues).where(eq(issues.id, f.issue.id)))[0].assigneeUserId).toBe(departing);
+      await reconcileDatabaseWork(db, DELIVERY_QUEUES.deferredExecutionReceipts);
+      await reconcileDatabaseWork(db, DELIVERY_QUEUES.nativeStatusWakeIntents);
+      deferred.mockClear(); native.mockClear();
+      await db.transaction(async tx => {
+        await accessService(tx as unknown as typeof db).archiveMember(f.companyId, member.id,
+          { reassignment: { assigneeAgentId: f.agent.id } });
+        expect(deferred).not.toHaveBeenCalled(); expect(native).not.toHaveBeenCalled();
+      });
+      expect(deferred).toHaveBeenCalledTimes(1); expect(native).toHaveBeenCalledTimes(1);
+      expect((await db.select().from(issues).where(eq(issues.id, f.issue.id)))[0]).toMatchObject({
+        assigneeUserId: null, assigneeAgentId: f.agent.id,
+      });
+      expect((await db.select().from(companyMemberships).where(eq(companyMemberships.id, member.id)))[0].status).toBe("archived");
+      expect(await readIntent()).toMatchObject({ status: "cancelled", runId: null });
+      expect(await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.contextSnapshot}->>'taskRetryRequestId' = ${request.requestId}`)).toHaveLength(0);
+    } finally { for (const unsubscribe of unsubscribes) unsubscribe(); }
+    await quiet();
+  });
   it("outer rollback and nested savepoint rollback do not dispatch receipt intent", async () => {
     const f = await fixture(); const dispatchNative = vi.fn(async () => {});
     await registerDeferredReceiptWork(db, coordinator, { resume: vi.fn(), settleTaskChat: vi.fn(), dispatchNative,

server/src/__tests__/issue-recovery-deadlines.test.ts

@@ -1,3 +1,4 @@
+import { setImmediate as immediate } from "node:timers/promises";
 import { idleWorkSnapshot } from "../services/task-admission.js";
 import { randomUUID } from "node:crypto";
 import { and, eq, sql } from "drizzle-orm";
@@ -8,6 +9,7 @@ import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } fro
 import { heartbeatService } from "../services/heartbeat.js";
 import { issueService } from "../services/issues.js";
 import { issueRecoveryActionService } from "../services/issue-recovery-actions.js";
+import { registerIssueRecoveryWork } from "../services/recovery/work.js";
 import { notifyIssueRecovery } from "../services/issue-recovery-notifications.js";
 import { taskWatchdogService } from "../services/task-watchdogs.js";
 import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
@@ -27,6 +29,7 @@ describe.skipIf(!support.supported)("issue recovery production producers and dea
   let draining = false;
   const tasks = new Map<string, { nextRunAt?: () => number | null; hasPending: () => Promise<boolean> }>();
   const errors: unknown[] = [];
+  const passes = new Set<Promise<unknown>>();
   const extraStops: Array<() => Promise<void>> = [];
   const extraDeadlines: Array<() => number | null> = [];
   beforeAll(async () => {
@@ -52,7 +55,14 @@ describe.skipIf(!support.supported)("issue recovery production producers and dea
       onError: error => errors.push(error) });
     tasks.clear();
     const register = coordinator.register.bind(coordinator);
-    coordinator.register = (queue, task) => { tasks.set(queue, task); return register(queue, task); };
+    coordinator.register = (queue, task) => {
+      tasks.set(queue, task);
+      return register(queue, { ...task, run: signal => {
+        const work = task.run(signal); passes.add(work);
+        void work.then(() => passes.delete(work), () => passes.delete(work));
+        return work;
+      } });
+    };
     worker = heartbeat.registerIssueStateRecoveryWork(coordinator, () => !draining);
     dispatch = heartbeat.registerRunDispatchWork(coordinator, () => !draining);
     cancellation = heartbeat.registerScheduledRetryCancellationWork(coordinator, () => !draining);
@@ -89,6 +99,7 @@ describe.skipIf(!support.supported)("issue recovery production producers and dea
   }
   async function quiet() {
     await settled();
+    expect(await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.status} in ('queued', 'running')`)).toHaveLength(0);
     const reads = vi.spyOn(db, "select"); const 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"] });
@@ -97,15 +108,32 @@ describe.skipIf(!support.supported)("issue recovery production producers and dea
   }
   async function finish(issueId: string) {
     await issueService(db).update(issueId, { status: "done" });
-    await heartbeat.drainActiveRunExecutions();
+    // Closing one blocker can legitimately start its dependent. Wait for this
+    // task's dispatch and execution; quiet() verifies the whole fixture later.
+    await vi.waitFor(async () => {
+      await Promise.allSettled([...passes]);
+      await heartbeat.drainActiveRunExecutions();
+      expect(await db.select().from(heartbeatRuns).where(and(
+        sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`,
+        sql`${heartbeatRuns.status} in ('queued', 'running')`,
+      ))).toHaveLength(0);
+    }, { timeout: 15_000 });
     const saved = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.status, "scheduled_retry"),
       sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`));
-    expect(await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.status} in ('queued', 'running')`)).toHaveLength(0);
     if (!vi.isFakeTimers()) vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
     if (saved.length > 0) {
       const due = Math.max(...saved.map(run => run.scheduledRetryAt!.getTime()));
-      await vi.advanceTimersByTimeAsync(Math.max(0, due - Date.now()) + 1);
+      // Keep PostgreSQL I/O and coordinator settlement ahead of virtual time.
+      // A single leap can otherwise occur before the retry deadline is armed.
+      while (Date.now() <= due) {
+        await vi.advanceTimersByTimeAsync(Math.min(500, due - Date.now() + 1));
+        await Promise.allSettled([...passes]);
+        await immediate();
+      }
       await vi.waitFor(async () => {
+        await vi.advanceTimersByTimeAsync(500);
+        await Promise.allSettled([...passes]);
+        await heartbeat.drainActiveRunExecutions();
         const remaining = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.status, "scheduled_retry"),
             sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`));
         expect(remaining).toHaveLength(0);
@@ -196,6 +224,52 @@ describe.skipIf(!support.supported)("issue recovery production producers and dea
     expect((await originalSelect().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, foreignAction.id)))[0].status).toBe("active");
     expect(await originalSelect().from(heartbeatRuns)).toHaveLength(0); await quiet();
   });
+  it("limits an ordinary comment to its issue among one thousand open assignments", async () => {
+    const f = await fixture();
+    const rows = await db.insert(issues).values(Array.from({ length: 1000 }, (_, index) => ({
+      companyId: f.companyId, title: `Open work ${index}`, status: "todo", assigneeAgentId: f.agentId,
+    }))).returning({ id: issues.id });
+    const checked: string[] = [];
+    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
+    coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
+    const register = coordinator.register.bind(coordinator);
+    coordinator.register = (queue, task) => register(queue, { ...task, run: signal => {
+      const pass = task.run(signal); passes.add(pass);
+      void pass.then(() => passes.delete(pass), () => passes.delete(pass));
+      return pass;
+    } });
+    // Observe the actual paged reconstruction boundary without executing one
+    // thousand adapters. The producer, transaction, coordinator and SQL are real;
+    // the other acceptance cases exercise the resulting recovery decisions.
+    worker = registerIssueRecoveryWork(db, coordinator, {
+      stranded: async scope => { checked.push(scope.issueId); },
+      orphanBlockers: async () => {}, recoveryActions: async () => {},
+      dependencies: async () => ({ enqueueFailed: 0 }), locks: async () => {},
+      watchdogs: async () => ({ incomplete: 0 }), schedulingSuppressed: async () => false,
+    }, () => true);
+    await worker.ready;
+    async function settleWork() {
+      for (let step = 0; step < 120; step++) {
+        await vi.advanceTimersByTimeAsync(1000);
+        await Promise.allSettled([...passes]); await immediate();
+        if (coordinator.nextWakeAt() === null && idleWorkSnapshot().active === 0) return;
+      }
+      throw new Error("Recovery did not settle its committed work within two virtual minutes");
+    }
+    await settleWork();
+    expect(new Set(checked).size).toBe(1000);
+    checked.length = 0;
+    await issueService(db).addComment(rows[0].id, "New evidence on this task", { userId: "recovery-user" });
+    await settleWork();
+    expect(checked).toEqual([rows[0].id]);
+    // Company eligibility changes still reconstruct all assignments. Narrowing
+    // comment hints must not hide unrelated work after an operator resumes it.
+    checked.length = 0;
+    await db.transaction(tx => notifyIssueRecovery(tx, { companyId: f.companyId }));
+    await settleWork();
+    expect(new Set(checked).size).toBe(1000);
+    await quiet();
+  });
   it("dispatches an ordinary new assignment after commit and returns to quiet after completion", async () => {
     const f = await fixture(); await start(); await quiet();
     const issue = await issueService(db).create(f.companyId, { title: "Recover assignment", status: "todo", assigneeAgentId: f.agentId });

server/src/services/access.ts

@@ -1,3 +1,4 @@
+import { notifyDeferredExecutionEligibility } from "./deferred-execution-notifications.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js";
 import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js";
@@ -743,6 +744,7 @@ export function accessService(db: Db) {
         sql`${issues.status} not in ('done', 'cancelled')`,
       );
       await notifyNativeStatusReassessment(tx, { companyId });
+      await notifyDeferredExecutionEligibility(tx, { companyId });
       await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
       await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
       const resetInProgress = await tx

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

@@ -0,0 +1,24 @@
+import { and, eq, sql } from "drizzle-orm";
+import { agentWakeupRequests, type Db } from "@paperclipai/db";
+import { notifyDeferredExecutionWork } from "./deferred-execution-notifications.js";
+
+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;
+    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({
+      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")));
+  });
+}

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

@@ -79,6 +79,24 @@ export function registerDeferredReceiptWork(
       },
       check: async entry => {
         if (!enabled) return null;
+        if (kind !== "task_chat") {
+          const [current] = await db.select({ status: agentWakeupRequests.status, payload: agentWakeupRequests.payload,
+            updatedAt: agentWakeupRequests.updatedAt, claimedAt: agentWakeupRequests.claimedAt, error: agentWakeupRequests.error,
+            interruptedLegacy: sql<boolean>`exists (select 1 from heartbeat_runs r
+              where r.company_id = ${agentWakeupRequests.companyId} and r.agent_id = ${agentWakeupRequests.agentId}
+                and r.runtime_mode = 'legacy' and r.status = 'cancelled'
+                and r.result_json->>'queuedCommentInterruptQueueId' = ${agentWakeupRequests.id}::text)`,
+          }).from(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, entry.companyId),
+            eq(agentWakeupRequests.id, entry.id)));
+          if (!current) return null;
+          const due = kind === "native" && current.status === "claimed" && current.claimedAt
+            ? current.claimedAt.getTime() + 60_000
+            : kind === "native" && current.status === "queued" && current.error ? current.updatedAt.getTime() + 30_000
+              : kind === "deferred" && current.status === "deferred_issue_execution" && !current.interruptedLegacy
+                && (current.payload?.queuedCommentInterrupt || current.payload?.executionWait)
+                ? current.updatedAt.getTime() + 30_000 : null;
+          if (due !== null && due > Date.now()) return { ...entry, dueAt: due };
+        }
         const errors: unknown[] = [];
         let next: number | null = null;
         const scope = { companyId: entry.companyId, requestId: entry.id };

server/src/services/heartbeat/queue.ts

@@ -1,3 +1,4 @@
+import { 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";
@@ -2523,17 +2524,11 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
       // its one durable receipt ID on a skipped wake.
       if (caseRequest) return { created: false };
       if (executionWaitRequestId) {
-        await db.update(agentWakeupRequests).set({
-          payload: sql`jsonb_set(coalesce(${agentWakeupRequests.payload}, '{}'::jsonb), '{executionWait}',
-            coalesce(${agentWakeupRequests.payload}->'executionWait', '{}'::jsonb) || ${JSON.stringify({
-              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."),
-            })}::jsonb)`,
-          updatedAt: new Date(),
-        }).where(and(eq(agentWakeupRequests.id, executionWaitRequestId),
-          eq(agentWakeupRequests.companyId, agent.companyId), eq(agentWakeupRequests.agentId, agentId),
-          eq(agentWakeupRequests.status, "deferred_issue_execution")));
+        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 };
       }
       const request = {
@@ -3421,10 +3416,8 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
             opts.issueStateGuard?.onExecutionDeferred?.(executionBlocker.cause);
             const condition = { recoveryActionId: executionBlocker.recoveryActionId, ...continuationWait };
             if (executionWaitRequestId) {
-              await tx.update(agentWakeupRequests).set({
-                payload: sql`jsonb_set(coalesce(${agentWakeupRequests.payload}, '{}'::jsonb), '{executionWait}', ${JSON.stringify(condition)}::jsonb)`,
-                updatedAt: new Date(),
-              }).where(eq(agentWakeupRequests.id, executionWaitRequestId));
+              await writeDeferredExecutionWait(tx as unknown as Db, { companyId: agent.companyId,
+                requestId: executionWaitRequestId }, condition, { replace: true });
               return { kind: "deferred" as const, admitted: true };
             }
             if (durableRequest || wakeCommentId ||

server/src/services/heartbeat/run-control.ts

@@ -1,3 +1,4 @@
+import { writeDeferredExecutionWait } from "../deferred-execution-wait.js";
 import type { DeferredReceiptRecoveryOptions } from "../deferred-execution-notifications.js";
 import { observeStoppedWorkspaceRecovery } from "../workspace-recovery.js";
 import { notifyScheduledRetryCancellation } from "../scheduled-retry-cancellation-notifications.js";
@@ -617,12 +618,7 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont
           onBlocked: (reason, message) => { wait = { reason, message }; },
         });
         if (!admitted) {
-          await db.update(agentWakeupRequests).set({
-            payload: sql`jsonb_set(coalesce(${agentWakeupRequests.payload}, '{}'::jsonb), '{executionWait}',
-              coalesce(${agentWakeupRequests.payload}->'executionWait', '{}'::jsonb) || ${JSON.stringify(wait)}::jsonb)`,
-            updatedAt: new Date(),
-          }).where(and(eq(agentWakeupRequests.id, wake.id), eq(agentWakeupRequests.companyId, run.companyId),
-            eq(agentWakeupRequests.status, "deferred_issue_execution")));
+          await writeDeferredExecutionWait(db, { companyId: run.companyId, requestId: wake.id }, wait);
           continue;
         }
       }

server/src/services/issue-comment-writes.ts

@@ -7,7 +7,8 @@ type WriteDb = Db | Parameters<Parameters<Db["transaction"]>[0]>[0];
 
 export function insertIssueComment(db: WriteDb, input: PgInsertValue<typeof issueComments> & { companyId: string }) {
   return db.transaction(async tx => {
-    await notifyIssueRecovery(tx, { companyId: input.companyId });
+    await notifyIssueRecovery(tx, { companyId: input.companyId,
+      issueId: typeof input.issueId === "string" ? input.issueId : undefined });
     await notifyDeliveryWork(tx, DELIVERY_QUEUES.dotEvents, `company:${input.companyId}`);
     return tx.insert(issueComments).values(input).returning();
   });

server/src/services/issue-recovery-notifications.ts

@@ -1,11 +1,16 @@
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 
-export type IssueRecoveryScope = { companyId: string };
+/** An issue hint is reserved for local facts such as comments. Lifecycle,
+ * graph, run and eligibility changes retain company scope because their
+ * consequences can include ancestors and shared execution references. */
+export type IssueRecoveryScope = { companyId: string; issueId?: string };
 
 export async function notifyIssueRecovery(transaction: object, scope?: IssueRecoveryScope) {
-  const hint = scope ? JSON.stringify({ companyId: scope.companyId }) : "*";
+  const companyHint = scope ? JSON.stringify({ companyId: scope.companyId }) : "*";
+  const issueHint = scope ? JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId }) : "*";
   for (const queue of [DELIVERY_QUEUES.orphanBlockers, DELIVERY_QUEUES.activeRecoveryActions, DELIVERY_QUEUES.strandedAssignments, DELIVERY_QUEUES.resolvedDependencies,
     DELIVERY_QUEUES.taskWatchdogs, DELIVERY_QUEUES.staleIssueLocks]) {
-    await notifyDeliveryWork(transaction, queue, hint);
+    await notifyDeliveryWork(transaction, queue,
+      queue === DELIVERY_QUEUES.strandedAssignments || queue === DELIVERY_QUEUES.staleIssueLocks ? issueHint : companyHint);
   }
 }

server/src/services/issue-tree-control.ts

@@ -1,3 +1,4 @@
+import { notifyDeferredExecutionEligibility } from "./deferred-execution-notifications.js";
 import { notifyIssueRecovery } from "./issue-recovery-notifications.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js";
@@ -760,6 +761,7 @@ export function issueTreeControlService(db: Db) {
 
       const { hold: resumeHold } = await db.transaction(async (tx) => {
         await notifyIssueRecovery(tx, { companyId });
+        await notifyDeferredExecutionEligibility(tx, { companyId });
         const [createdHold] = await tx
           .insert(issueTreeHolds)
           .values({
@@ -839,6 +841,7 @@ export function issueTreeControlService(db: Db) {
 
     const { hold, members } = await db.transaction(async (tx) => {
       await notifyIssueRecovery(tx, { companyId });
+      await notifyDeferredExecutionEligibility(tx, { companyId });
       const [createdHold] = await tx
         .insert(issueTreeHolds)
         .values({
@@ -1003,6 +1006,7 @@ export function issueTreeControlService(db: Db) {
     const now = new Date();
     const releasedCancelHoldIds = activeCancelHolds.map((hold) => hold.id);
     const updatedIssues = await db.transaction(async (tx) => {
+      await notifyDeferredExecutionEligibility(tx, { companyId });
       await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
       await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
       const restored: TreeStatusUpdateResult["updatedIssues"] = [];
@@ -1182,6 +1186,7 @@ export function issueTreeControlService(db: Db) {
 
     const [updated] = await db.transaction(async tx => {
       await notifyIssueRecovery(tx, { companyId });
+      await notifyDeferredExecutionEligibility(tx, { companyId });
       return tx
       .update(issueTreeHolds)
       .set({

server/src/services/native-status-reassessment-notifications.ts

@@ -4,7 +4,7 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notificatio
 export type NativeStatusReassessmentScope = { companyId: string; issueId?: string; runId?: string; agentId?: string };
 
 export async function notifyNativeStatusReassessment(tx: object, scope: NativeStatusReassessmentScope) {
-  await notifyIssueRecovery(tx, scope);
+  await notifyIssueRecovery(tx, scope ? { companyId: scope.companyId } : undefined);
   return notifyDeliveryWork(tx, DELIVERY_QUEUES.nativeStatusReassessment,
     JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId, runId: scope.runId, agentId: scope.agentId }));
 }

server/src/services/recovery/work.ts

@@ -32,13 +32,14 @@ export function registerIssueRecoveryWork(
       reconstructionIntervalMs: 30_000,
       beforePass: async () => { enabled = !(await owner.schedulingSuppressed()); },
       decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint) as IssueRecoveryScope,
-      matches: (entry, scope) => !scope || entry.companyId === scope.companyId,
+      matches: (entry, scope) => !scope || (entry.companyId === scope.companyId && (!scope.issueId || entry.id === scope.issueId)),
       load: async scope => {
-        const key = scope?.companyId ?? "*";
+        const key = scope ? JSON.stringify([scope.companyId, scope.issueId]) : "*";
         if (!enabled) { cursors.delete(key); return []; }
         const cursor = cursors.get(key);
         const rows = await db.select({ id: issues.id, companyId: issues.companyId }).from(issues).where(and(
           scope ? eq(issues.companyId, scope.companyId) : undefined,
+          scope?.issueId ? eq(issues.id, scope.issueId) : undefined,
           cursor ? gt(issues.id, cursor) : undefined,
           statuses ? inArray(issues.status, statuses) : or(isNotNull(issues.checkoutRunId), isNotNull(issues.executionRunId)),
           statuses ? isNull(issues.hiddenAt) : undefined,

server/src/services/run-dispatch-notifications.ts

@@ -4,7 +4,7 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notificatio
 export type RunDispatchScope = { companyId: string };
 
 export async function notifyRunDispatch(transaction: object, scope?: RunDispatchScope) {
-  await notifyIssueRecovery(transaction, scope);
+  await notifyIssueRecovery(transaction, scope ? { companyId: scope.companyId } : undefined);
   const hint = scope ? JSON.stringify({ companyId: scope.companyId }) : "*";
   await notifyDeliveryWork(transaction, DELIVERY_QUEUES.scheduledRetryDispatch, hint);
   await notifyDeliveryWork(transaction, DELIVERY_QUEUES.queuedRunDispatch, hint);