Retain moved receipt deadlines and wake released execution gates

Co-Authored-By: Paperclip <noreply@paperclip.ing>

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

@@ -1,7 +1,14 @@
+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";
@@ -61,6 +68,15 @@ 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 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));
+  }
   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 +102,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 +218,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/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-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({