fix(heartbeat): release task Retries held by a drain as soon as it ends (#20640)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work
> - When a person clicks Retry on a stopped task, Paperclip saves the
request as a held wake and dispatches it right away, unless admission is
held. Admission is held during a task drain, for example while a
deployment restarts the server
> - A Retry held by a drain is released later by the deferred-receipt
worker. That worker shares one queue with every other held wake: comment
waits, execution waits, and interrupted queues
> - On a busy instance that queue holds hundreds of passive waits that
come due every 30 seconds. A Retry is checked in due-time order behind
all of them. Ending a drain does not wake the worker either
> - In a real case, a Retry clicked during a deployment drain was still
held 40 minutes later, after the drain expired and the server restarted.
It ran only when the person clicked Retry again
> - This pull request checks operator Retries ahead of passive waits,
and wakes the worker as soon as a drain stops or expires
> - The benefit is that a Retry held by a drain starts within one worker
pass after the drain ends

## Linked Issues or Issue Description

**What happened?**

A person clicked Retry on a blocked task 14 seconds after a deployment's
task drain started. The Retry was saved as `deferred_issue_execution`,
and the immediate dispatch returned early because of the drain. The
drain had a 20-minute TTL, and the deployment restarted the server about
6 minutes in. The held Retry still did not start, and its `updated_at`
never changed. Forty minutes later the person clicked Retry again, and
the new request started a run and merged with the held one.

That instance had 390 held wakes: 271 execution waits and 116 others,
against 6 task Retries. The execution waits come due every 30 seconds.
Measured on the instance, the deferred-receipt worker took about 7
minutes to get through that backlog once.

**Expected behavior**

A Retry that a drain held starts soon after the drain stops or expires,
or after the server restarts, whatever the size of the passive backlog.

**Steps to reproduce**

1. Start a non-idle task drain (`POST /instance/task-drain`).
2. Click Retry on a stopped task. The receipt is
`deferred_issue_execution`.
3. Stop the drain. On `master`, the deferred-receipt worker only notices
on its next 30-second re-check. It then checks the Retry in due-time
order behind every passive wait that came due before it.

**Paperclip version or commit**

`master` at 8148fc0b1e.

## What Changed

- `server/src/services/scoped-deadline-work.ts`:
- New optional `priority(entry)`. Due entries are sorted by priority
first, then by due time and id as before.
- New `refresh(hint)`. It clears a scope's reconstruction delay and
invalidates it. Failure backoff is kept.
- New optional `onStop` hook, called when the coordinator stops the
worker.
- `server/src/services/heartbeat/deferred-receipt-work.ts`:
- A `task_retry` receipt that a person requested is an operator Retry,
with priority 1. Passive receipts keep priority 0, so their order does
not change.
- Each deferred scan loads operator Retries first, with their own query,
then pages by id as before. A Retry is therefore in memory in the first
pass even beyond the first 100-row page. A receipt loaded twice merges
by id.
- New optional `owner.subscribeRefresh`. The work calls `refresh()` on
all three of its queues when it fires, and releases the subscription
from `onStop`, so the coordinator's own stop removes it.
- `server/src/services/heartbeat.ts`: passes `subscribeRefresh` that
listens to task-drain changes and expiry. Before this change, a deploy
drain left the worker waiting for its 30-second re-check, behind the
scan's reconstruction delay.
- `server/src/__tests__/deferred-execution-receipt-work.test.ts`, four
new tests:
- A real Retry, requested during a deploy drain, starts a run (to
`succeeded`) within 5 seconds after the drain stops.
- An operator Retry whose id sorts after 120 passive execution waits
(more than one load page) is resumed first, in the first worker pass.
- A held Retry inserted with no notification right after a full scan is
resumed as soon as the drain refresh fires.
  - Stopping only the coordinator removes the drain listener.

## Verification

- `npx vitest run src/__tests__/deferred-execution-receipt-work.test.ts
-t "held by a deploy drain|ahead of a backlog|ordinary committed task
Retry"`: 3 pass. The third is the existing ordinary Retry dispatch test.
- Red checks, removing one piece at a time:
- Without the drain subscription, the drain test fails: the Retry is
still `deferred_issue_execution` with no run after 5 seconds.
- Without the priority, or without the prioritized load, the backlog
test fails: the first resumed request is the passive `00000000-…` wake.
- With `invalidate()` instead of `refresh()`, the post-scan test fails
after 5 seconds.
  - Without the `onStop` release, the listener test fails.
- Full `deferred-execution-receipt-work.test.ts`: 50 pass.
- `chat-channels.integration.test.ts -t "still batches several replies
in one finding thread"`: it timed out once on CI and passes 2 of 2 times
locally with this change.
- `scoped-deadline-work`, `heartbeat-task-drain`,
`heartbeat-task-drain-admission-release`, `cloud-control-task-drain`,
`task-admission-deadline`, `onboarding-first-task-admission` and
`delivery-work-coordinator` suites: 164 pass.
- `pnpm run typecheck` in `server` (layered): clean.

## Risks

- Operator Retries are a small set; there were 6 on the instance above.
They now take the first slots of a worker pass, so passive waits can
move by at most that many places. Passive waits keep their own order.
- A Retry that keeps failing its check still gets the existing failure
backoff, so it cannot take every pass.
- A drain start also wakes the worker. That pass ends at once, because
the worker's run gate still reads the drain.

## Model Used

- Claude Opus 5.5 (`claude-opus-5-5`) via Claude Code, with tool use
(shell, file editing, vitest runs against embedded PostgreSQL, read-only
production log and database analysis).

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [x] All Paperclip CI gates are green
- [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

---------

Co-authored-by: Paperclip <noreply@paperclip.ing>

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

@@ -21,7 +21,7 @@ import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.j
 import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
 import { notifyDeferredExecutionEligibility, notifyDeferredExecutionWork, notifyNativeStatusWakeWork, notifyTaskRetrySettlement } from "../services/deferred-execution-notifications.js";
 import { registerDeferredReceiptWork } from "../services/heartbeat/deferred-receipt-work.js";
-import { heartbeatService } from "../services/heartbeat.js";
+import { heartbeatService, startTaskDrain, stopTaskDrain } from "../services/heartbeat.js";
 import { requestTaskRetry } from "../services/task-retry.js";
 import { issueService } from "../services/issues.js";
 import { environmentService } from "../services/environments.js";
@@ -376,6 +376,85 @@ describe("deferred receipt durable dispatch", () => {
     expect(await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.contextSnapshot}->>'taskRetryRequestId' = ${receipt.requestId}`)).toHaveLength(1);
     await quiet();
   });
+  it("dispatches a task Retry held by a deploy drain as soon as the drain stops", async () => {
+    const f = await fixture();
+    await heartbeat.registerDeferredReceiptWork(coordinator, () => !drain).ready;
+    await heartbeat.registerRunDispatchWork(coordinator, () => !drain).ready;
+    await heartbeat.registerScheduledRetryCancellationWork(coordinator, () => !drain).ready;
+    startTaskDrain({ ttlMs: 20 * 60_000 });
+    let requestId: string;
+    try {
+      const receipt = await requestTaskRetry({ db, companyId: f.companyId, issueId: f.issue.id, actorId: f.actorId,
+        observedRunId: f.source.id, idempotencyKey: randomUUID() });
+      expect(receipt.accepted).toBe(true);
+      requestId = receipt.requestId!;
+      await heartbeat.resumeTaskRetry(f.companyId, requestId);
+      await vi.advanceTimersByTimeAsync(1000);
+      const [held] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, requestId));
+      expect(held).toMatchObject({ status: "deferred_issue_execution", runId: null });
+    } finally { stopTaskDrain(); }
+    let runId: string | undefined;
+    await waitForReal(async () => {
+      const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, requestId));
+      expect(wake.runId, JSON.stringify(wake)).toBeTruthy(); runId = wake.runId!;
+    }, { timeout: 5_000 });
+    await waitForReal(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");
+    }, { timeout: 15_000 });
+    await finish(f.issue.id);
+    await quiet();
+  });
+  it("dispatches an operator Retry ahead of a backlog of passive waits", async () => {
+    const f = await fixture();
+    await db.insert(agentWakeupRequests).values(Array.from({ length: 120 }, (_, index) => ({
+      id: `00000000-0000-4000-8000-${String(index).padStart(12, "0")}`, companyId: f.companyId, agentId: f.agent.id,
+      source: "automation", status: "deferred_issue_execution", requestedByActorType: "system",
+      payload: { executionWait: { reason: "recovery" } }, updatedAt: new Date(Date.now() - 60_000),
+    })));
+    const [retry] = await db.insert(agentWakeupRequests).values({ id: "ffffffff-ffff-4fff-bfff-ffffffffffff",
+      companyId: f.companyId, agentId: f.agent.id, source: "on_demand", status: "deferred_issue_execution",
+      reason: "task_retry", requestedByActorType: "user", requestedByActorId: f.actorId,
+      payload: { issueId: f.issue.id } }).returning();
+    const order: string[] = [];
+    const resume = vi.fn(async (scope: { requestId?: string }) => {
+      order.push(scope.requestId!);
+      await db.update(agentWakeupRequests).set({ status: "skipped" }).where(eq(agentWakeupRequests.id, scope.requestId!));
+    });
+    await registerDeferredReceiptWork(db, coordinator, { resume, settleTaskChat: vi.fn(), createNativeBatch: () => vi.fn(),
+      schedulingSuppressed: async () => false }, () => !drain).ready;
+    expect(order[0]).toBe(retry.id);
+    await waitForReal(() => expect(order).toHaveLength(121), { timeout: 30_000 });
+    await quiet();
+  });
+  it("reloads a held Retry without a pending notification as soon as a drain ends, right after a full scan", async () => {
+    const f = await fixture();
+    let refresh: (() => void) | undefined;
+    const resume = vi.fn(async (scope: { requestId?: string }) => {
+      await db.update(agentWakeupRequests).set({ status: "skipped" }).where(eq(agentWakeupRequests.id, scope.requestId!));
+    });
+    await registerDeferredReceiptWork(db, coordinator, { resume, settleTaskChat: vi.fn(), createNativeBatch: () => vi.fn(),
+      schedulingSuppressed: async () => false, subscribeRefresh: listener => { refresh = listener; return () => {}; } },
+    () => !drain).ready;
+    const [retry] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agent.id,
+      source: "on_demand", status: "deferred_issue_execution", reason: "task_retry", requestedByActorType: "user",
+      requestedByActorId: f.actorId, payload: { issueId: f.issue.id } }).returning();
+    refresh!();
+    await waitForReal(() => expect(resume).toHaveBeenCalledWith(expect.objectContaining({ requestId: retry.id })),
+      { timeout: 5_000 });
+    await quiet();
+  });
+  it("removes its task-drain refresh listener when the coordinator stops", async () => {
+    let refresh: (() => void) | undefined;
+    const unsubscribe = vi.fn();
+    await registerDeferredReceiptWork(db, coordinator, { resume: vi.fn(), settleTaskChat: vi.fn(),
+      createNativeBatch: () => vi.fn(), schedulingSuppressed: async () => false,
+      subscribeRefresh: listener => { refresh = listener; return unsubscribe; } }, () => !drain).ready;
+    expect(refresh).toBeTypeOf("function");
+    await coordinator.stop();
+    expect(unsubscribe).toHaveBeenCalledTimes(1);
+  });
   async function blockedRetry(gate: "budget" | "project budget" | "paused agent", conversation = false) {
     const f = await fixture(), lifecycle = createAgentLifecycle(db), budgets = budgetService(db);
     if (conversation) await db.update(issues).set({ status: "in_review", conversationAgentId: f.agent.id,

server/src/services/heartbeat.ts

@@ -561,7 +561,8 @@ import { resolveAndRetainRunTrustPreset } from "./run-trust-preset.js";
 import { effectiveProjectWorkspacePolicy, sourcelessProjectWorkspacePolicy } from "./project-workspace-provider.js";
 import type { PluginWorkerManager } from "./plugin-worker-manager.js";
 
-import { computeTaskDrain, applyTaskDrain, startTaskDrain, stopTaskDrain, readTaskDrain } from "./task-admission.js";
+import { computeTaskDrain, applyTaskDrain, startTaskDrain, stopTaskDrain, readTaskDrain, subscribeTaskDrainChanges,
+  subscribeTaskDrainExpiry } from "./task-admission.js";
 
 const execFile = promisify(execFileCallback);
 const UNSUCCESSFUL_HEARTBEAT_RUN_TERMINAL_STATUSES = [
@@ -4515,6 +4516,10 @@ export function heartbeatService(
         settleTaskChat: scope => reconcileTaskChatRetryReceipts(db, { ...scope, settleOnly: true }),
         createNativeBatch: heartbeatQueue.createNativeStatusWakeBatch,
         schedulingSuppressed: async () => (await getSchedulingSuppression(true)).suppressed,
+        subscribeRefresh: refresh => {
+          const unsubscribe = [subscribeTaskDrainChanges(refresh), subscribeTaskDrainExpiry(refresh)];
+          return () => { for (const stop of unsubscribe) stop(); };
+        },
       }, () => canRun() && readTaskDrain(new Date()) === null),
     readIdleWatchdogSchedule: taskWatchdogs.readIdleSchedule,
     registerIssueStateRecoveryWork: (coordinator: Parameters<typeof registerIssueRecoveryWork>[1], canRun: () => boolean) =>

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

@@ -7,7 +7,7 @@ import type { DeferredExecutionScope, DeferredReceiptRecoveryOptions } from "../
 import { registerScopedDeadlineWork } from "../scoped-deadline-work.js";
 import { dependencyResolvedWakePayloadSql } from "../issue-dependency-wakeups.js";
 
-type Entry = { id: string; companyId: string; dueAt: number; immediate: boolean };
+type Entry = { id: string; companyId: string; dueAt: number; immediate: boolean; operatorRetry?: boolean };
 type NativeIntentOptions = DeferredExecutionScope & {
   onDeferredDeadline(at: number): void;
   onError(error: unknown): void;
@@ -21,9 +21,15 @@ export function registerDeferredReceiptWork(
     settleTaskChat(scope: DeferredExecutionScope): Promise<unknown>;
     createNativeBatch(): (scope: NativeIntentOptions) => Promise<unknown>;
     schedulingSuppressed(): Promise<boolean>;
+    subscribeRefresh?(refresh: () => void): () => void;
   },
   canRun: () => boolean,
 ) {
+  let unsubscribeRefresh: (() => void) | undefined;
+  function releaseRefresh() {
+    unsubscribeRefresh?.();
+    unsubscribeRefresh = undefined;
+  }
   function register(kind: "deferred" | "task_chat" | "native") {
     const cursors = new Map<string, string>();
     let enabled = false;
@@ -45,23 +51,28 @@ export function registerDeferredReceiptWork(
       // receipts already checked; actual wait/claim deadlines still replace.
       mergeEntry: (pending, loaded) => pending.immediate && loaded.immediate
         ? { ...loaded, dueAt: Math.min(pending.dueAt, loaded.dueAt) } : loaded,
+      priority: entry => entry.operatorRetry ? 1 : 0,
+      onStop: releaseRefresh,
       load: async scope => {
         const key = scope ? JSON.stringify({ companyId: scope.companyId, requestId: scope.requestId }) : "*";
         if (!enabled) { cursors.delete(key); return []; }
         const cursor = cursors.get(key);
-        const rows = await db.select({ id: agentWakeupRequests.id, companyId: agentWakeupRequests.companyId,
+        const operatorRetry = sql`coalesce(${agentWakeupRequests.reason}, '') = 'task_retry'
+          and coalesce(${agentWakeupRequests.requestedByActorType}, '') = 'user'`;
+        const columns = { id: agentWakeupRequests.id, companyId: agentWakeupRequests.companyId,
           status: agentWakeupRequests.status, updatedAt: agentWakeupRequests.updatedAt,
           claimedAt: agentWakeupRequests.claimedAt, error: agentWakeupRequests.error,
+          operatorRetry: sql<boolean>`${operatorRetry}`,
           delayed: sql<boolean>`${agentWakeupRequests.payload}->'queuedCommentInterrupt' is not null
             or ${agentWakeupRequests.payload}->'executionWait' is not null`,
           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(
+        };
+        const eligible = and(
           scope ? eq(agentWakeupRequests.companyId, scope.companyId) : undefined,
           scope?.requestId ? eq(agentWakeupRequests.id, scope.requestId) : undefined,
-          cursor ? gt(agentWakeupRequests.id, cursor) : undefined,
           kind === "native" ? and(
             eq(agentWakeupRequests.requestedByActorType, "system"),
             eq(agentWakeupRequests.requestedByActorId, "native-status-committer"),
@@ -80,17 +91,25 @@ export function registerDeferredReceiptWork(
                 and r.runtime_mode = 'legacy' and r.status = 'cancelled'
                 and r.result_json->>'queuedCommentInterruptQueueId' = ${agentWakeupRequests.id}::text)`,
             )),
-        )).orderBy(asc(agentWakeupRequests.id)).limit(100);
+        );
+        const prioritized = kind === "deferred" && !cursor
+          ? await db.select(columns).from(agentWakeupRequests).where(and(eligible, operatorRetry))
+            .orderBy(asc(agentWakeupRequests.id)).limit(100)
+          : [];
+        const rows = await db.select(columns).from(agentWakeupRequests).where(and(eligible,
+          cursor ? gt(agentWakeupRequests.id, cursor) : undefined))
+          .orderBy(asc(agentWakeupRequests.id)).limit(100);
         const complete = rows.length < 100;
         if (complete) cursors.delete(key); else cursors.set(key, rows.at(-1)!.id);
         const now = Date.now();
-        return { rows: rows.map(row => {
+        return { rows: [...prioritized, ...rows].map(row => {
           const deadline = kind === "native" && row.status === "claimed" && row.claimedAt ? row.claimedAt.getTime() + 60_000
             : kind === "native" && row.error ? row.updatedAt.getTime() + 30_000
             : kind === "deferred" && row.delayed && !row.interruptedLegacy ? row.updatedAt.getTime() + 30_000
             : null;
           const immediate = deadline === null || deadline <= now;
-          return { id: row.id, companyId: row.companyId, dueAt: immediate ? now : deadline, immediate };
+          return { id: row.id, companyId: row.companyId, dueAt: immediate ? now : deadline, immediate,
+            operatorRetry: kind === "deferred" && row.operatorRetry };
         }), complete };
       },
       check: async entry => {
@@ -132,6 +151,7 @@ export function registerDeferredReceiptWork(
     }, canRun);
   }
   const workers = [register("deferred"), register("task_chat"), register("native")];
+  unsubscribeRefresh = owner.subscribeRefresh?.(() => { for (const worker of workers) worker.refresh(); });
   return { ready: Promise.all(workers.map(worker => worker.ready)),
     wake() { for (const worker of workers) worker.invalidate(); },
     async stop() { await Promise.all(workers.map(worker => worker.stop())); } };

server/src/services/scoped-deadline-work.ts

@@ -32,7 +32,9 @@ export function registerScopedDeadlineWork<Entry extends ScheduledEntry, Scope>(
     load(scope?: Scope, signal?: AbortSignal): Promise<Entry[] | { rows: Entry[] | null; complete: boolean }>;
     // Owners may retain pending queue priority while refreshing eligibility.
     mergeEntry?(pending: Entry, loaded: Entry): Entry;
+    priority?(entry: Entry): number;
     check(entry: Entry, signal?: AbortSignal): Promise<Entry | null>;
+    onStop?(): void;
   },
   canRun: () => boolean,
 ) {
@@ -73,7 +75,7 @@ export function registerScopedDeadlineWork<Entry extends ScheduledEntry, Scope>(
   const worker = coordinator.register(options.queue, {
     retryMs: 30_000,
     pausePassiveMaintenance: true,
-    onStop: () => { stopped = true; revision++; },
+    onStop: () => { stopped = true; revision++; options.onStop?.(); },
     onSettled: hints => { revision++; for (const hint of hints ?? ["*"]) {
       dirty.add(hint); versions.set(hint, (versions.get(hint) ?? 0) + 1);
     } },
@@ -134,7 +136,8 @@ export function registerScopedDeadlineWork<Entry extends ScheduledEntry, Scope>(
         const unavailable = dirtyScopes();
         const due = [...entries.values()].filter(entry => !unavailable.some(scope => options.matches(entry, scope))
           && Math.max(entry.dueAt, failures.get(entry.id)?.retryAt ?? 0) <= Date.now())
-          .sort((a, b) => a.dueAt - b.dueAt || a.id.localeCompare(b.id));
+          .sort((a, b) => (options.priority?.(b) ?? 0) - (options.priority?.(a) ?? 0)
+            || a.dueAt - b.dueAt || a.id.localeCompare(b.id));
         const batch = takeCompanyWorkBatch(due, options.batchSize ?? 25, lastCompany); lastCompany = batch.lastCompany;
         // An owner may share more than one resource (for example a task and an
         // endpoint). Merge intersecting keys before dispatch so waiting siblings
@@ -275,6 +278,11 @@ export function registerScopedDeadlineWork<Entry extends ScheduledEntry, Scope>(
     nextRunAt: next,
     nextRunAtOnError: next,
   });
+  function invalidate(hint = "*") {
+    revision++;
+    dirty.add(hint); versions.set(hint, (versions.get(hint) ?? 0) + 1);
+    worker.wake();
+  }
   return { ...worker,
     reconciliationSnapshot: (): DeadlineReconciliationSnapshot => ({
       revision: `${ownerRevision}:${revision}`,
@@ -282,10 +290,10 @@ export function registerScopedDeadlineWork<Entry extends ScheduledEntry, Scope>(
         && failures.size === 0 && scopeFailures.size === 0,
     }),
     wake() { revision++; worker.wake(); },
-    invalidate(hint = "*") {
-      revision++;
-      dirty.add(hint); versions.set(hint, (versions.get(hint) ?? 0) + 1);
-      worker.wake();
+    invalidate,
+    refresh(hint = "*") {
+      reconstructAfter.delete(hint);
+      invalidate(hint);
     },
   };
 }