fix(heartbeat): prevent saved input starvation during queue reloads (#18819)

## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Accepted messages and task Retry requests use durable wake receipts. > - The recovery worker loads receipts in pages of 100 and checks batches of 25. > - Company activity can reload those pages before the pending batch drains. > - Reloads reset ready receipts' dispatch times and can repeatedly favor receipts already checked. > - This pull request preserves the priority of unchecked ready receipts so saved input can reach normal admission. ## Linked Issues or Issue Description **What happened?** Saved input can remain deferred after the preceding run finishes. Repeated company eligibility changes reset the dispatch times of pending receipts during paginated reconstruction. A PostgreSQL regression checks only 75 of 178 receipts on the original code despite twelve eligibility changes. **Expected behavior** Each ready receipt must reach its admission check. Existing wait deadlines, retry backoffs, authorization, and execution gates must still apply. **Steps to reproduce** 1. Seed more than one page of deferred receipts. 2. Keep those receipts pending after their checks. 3. Send company eligibility notifications between scheduler passes. 4. Check that every receipt is reconsidered, including receipts later in the backlog. **Paperclip version or commit** Reproduced on `367f29478d`. **Deployment mode** Built from source with PostgreSQL. The defect occurs before provider execution. Related: #10195 reports a similar starvation symptom in a different promotion loop. #18621 added saved-input Retry after a successful run. This change fixes fairness in the deferred-receipt worker. ## What Changed - Add an optional entry merge callback to scoped deadline reconstruction. - Use it in the deferred-receipt worker to retain unread ready receipts' priority while refreshing their state. - Place receipts whose wait has expired into the ready queue during reconstruction. - Keep real wait, claim, and dispatch deadlines authoritative. - Add a paginated starvation regression, real process dispatch coverage, and GitHub lock-collision recovery coverage. - Document the recovery fairness rule. ## Verification - The PostgreSQL starvation regression checks only 75 of 178 receipts before the fix. It checks all 178 afterward, both with ordinary receipts and with expired execution waits. - Seven focused scheduler cases pass, including moved interrupt grace, moved native error deadline, fresh claim/wait deadlines, and unchanged-gate grace. - GitHub endpoint lock-collision recovery, retained board authority, HTTP Retry, and five current admission-gate cases pass. - Real process acceptance passes on the final code: task Retry crosses the repeatedly invalidated backlog, reconstructs its persisted queued run after a worker restart, saves `receipt-proof` in run output, and has one successor for the Retry receipt. The test holds executor capacity while advancing queue time, then starts fresh workers at PostgreSQL time before releasing a real process lease. The defect is in admission scheduling before provider execution; this acceptance uses PostgreSQL and a local process adapter. - The full workspace build and typecheck (`node scripts/typecheck-low-memory.mjs`) pass locally. The affected worker suites pass 52 of 53 cases together; the existing native multi-company batch case times out in that combined run and passes in isolation without code changes. All new regression and acceptance cases pass. - Local `pnpm test:run` stops in its first chat shard on three 15-second timeouts (Teams long native HTTP picture delivery, direct final-response materialization, and contended inbound notices). All three pass in isolation without code changes. Later full-suite groups did not run locally after that shard failure. - [The full GitHub Actions matrix passes](https://github.com/paperclipai/paperclip/actions/runs/41697489374) on `ac91df2cc6db6c77529d923177b375b8c92efd45`, including all general-test, serialized-server, browser, build, and typecheck jobs. Server shard 8 initially failed an upstream process-adoption test and passed its one rerun without code changes. - [Greptile is 5/5](https://github.com/paperclipai/paperclip/pull/18819#issuecomment-6612091535) on the same current head, with no actionable findings or unresolved review threads. The separate Paperclip Review check is neutral; its previous request was bound to an older commit. Focused command: ```sh pnpm exec vitest run server/src/__tests__/deferred-execution-receipt-work.test.ts server/src/__tests__/scoped-deadline-work.test.ts ``` ## Risks - Ready receipt order changes during repeated reconstruction. Admission still reads current durable state. - A ready receipt must not inherit an obsolete priority after it acquires a future deadline. The merge applies only when both pending and loaded entries are immediate. Deadline safety tests cover that boundary. - The merge callback is optional. Other scoped workers retain their current reconstruction behavior. ## Model Used OpenAI `gpt-6.1-sol` in Codex, max reasoning effort, with tool use and local code execution. The runtime does not expose the context-window size. ## 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 (focused regressions and acceptance; full-run timing limitations documented above) - [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>
2026-11-06 14:02:36 +01:00 · 2026-11-03 13:17:00 -06:00
1 parent 8ab89f41e4
commit e0c6992426
@@ -756,6 +756,8 @@ Normal issue-thread feedback can still wake the current assignee according to th

 Restart recovery revisits accepted deferred messages through the latest issue run, including historical mention queues for another agent. Ordinary messages for a former owner retain their assignment handoff. Native sources require their matching committed finalization, settled environment leases, and released finalizer and copyback ownership before recovery clears old issue claims. A successful retained warm release can satisfy the lease check. Released local bookkeeping without a provider allocation and a matching SSH bookkeeping release need no provider cleanup receipt. Plugin releases record their successful acknowledgement. Pending or failed cleanup and durable owners still hold work. A successful fenced cleanup completion clears its in-flight claim in the same database write, so a restart before retry bookkeeping cannot strand saved input. The normal release checks preserve cleanup holds, pauses, and operator Stops. Recovery checks for a competing turn under the issue lock before it promotes saved input.

+Repeated eligibility notifications must not move an unchecked, ready deferred receipt behind receipts already reconsidered. Paginated reconstruction preserves pending ready receipts' dispatch priority while refreshing their current state. Receipts whose wait deadline has passed join the ready queue on reconstruction. Future wait grace periods, claimed-work deadlines, infrastructure retry backoffs, and current admission checks remain effective.
+
 A paused recipient does not fail an accepted deferred request. Release and periodic recovery preserve its pending receipt and may process an eligible request behind it. Resuming the agent allows normal admission to reconsider the saved input. Missing, foreign, terminated, and unapproved recipients retain their rejection rules. Explicit task holds and operator Stops still prevent unauthorized execution. This does not replay historical failed requests.

 Accepted agent feedback must survive a child changing to `done` before its active run exits. Deferred wake promotion may reopen that completed child only for its current assignee, with explicit agent resume intent and live tracked comments from another author. Claim promotion before reopening. Cancelled tasks, deleted comments, self-authored comments, empty continuations, and agent continuations without explicit intent keep their existing suppression rules. Normal pause, ownership, authorization, and budget gates still apply.
@@ -64441,6 +64441,45 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
    } finally { await f.service.shutdown(); }
  });

+  it("retries saved GitHub input after an endpoint admission lock collision", async () => {
+    const f = await successfulGitHubQueuedInputFixture();
+    const heartbeat = heartbeatService(db);
+    const failures: unknown[] = [];
+    const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true,
+      onError: error => failures.push(error) });
+    let unlock!: () => void, locked!: () => void;
+    const acquired = new Promise<void>(resolve => { locked = resolve; });
+    const released = new Promise<void>(resolve => { unlock = resolve; });
+    const competingAdmission = db.transaction(async tx => {
+      await tx.select({ id: chatEndpoints.id }).from(chatEndpoints)
+        .where(eq(chatEndpoints.id, f.endpoint.id)).for("no key update");
+      locked();
+      await released;
+    });
+    try {
+      await acquired;
+      await heartbeat.registerDeferredReceiptWork(coordinator, () => true).ready;
+      const priorFailures = failures.length;
+      const retry = await requestTaskRetry(f.input);
+      await expect.poll(() => failures.length, { timeout: 5_000 }).toBeGreaterThan(priorFailures);
+      expect((await readAgentWakeupRequestsById(f.pending!.id))[0].runId).toBeNull();
+      unlock(); await competingAdmission;
+      await expect.poll(async () => (await readAgentWakeupRequestsById(f.pending!.id))[0].runId,
+        { timeout: 40_000 }).toBeTruthy();
+      await expect.poll(async () => (await readAgentWakeupRequestsById(retry.requestId!))[0].status,
+        { timeout: 40_000 }).toBe("coalesced");
+      const [saved] = await readAgentWakeupRequestsById(f.pending!.id);
+      expect((await readHeartbeatRunsByCompanyId(f.fixture.companyId))
+        .filter(run => run.wakeupRequestId === f.pending!.id)).toHaveLength(1);
+      expect((await readAgentWakeupRequestsById(retry.requestId!))[0].runId).toBe(saved.runId);
+      expect((await readHeartbeatRunsById(f.runId))[0]).toMatchObject({ status: "succeeded" });
+    } finally {
+      unlock(); await competingAdmission;
+      await coordinator.stop();
+      await f.service.shutdown();
+    }
+  }, 90_000);
+
  it.each(["reconnected", "newer_input", "reconnected_newer_input", "snapshot_newer_input"] as const)(
    "admits current GitHub task Retry without replaying obsolete input: %s", async kind => {
      const f = await currentGitHubTaskRetryFixture(kind.includes("newer_input"));
@@ -72,14 +72,18 @@ describe("deferred receipt durable dispatch", () => {
    // during actual process execution; accelerated steps only settle idle work.
    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
    heartbeat = heartbeatService(db);
-    coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !drain, canReconcile: () => true,
+    coordinator = createCoordinator();
+  });
+  function createCoordinator() {
+    const worker = createDeliveryWorkCoordinator({ owner: db, canRun: () => !drain, canReconcile: () => true,
      onError: error => { errors.push(error); errorTimes.push(Date.now()); } });
-    const register = coordinator.register.bind(coordinator);
-    coordinator.register = (queue, task) => register(queue, { ...task, run: signal => {
+    const register = worker.register.bind(worker);
+    worker.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;
    } });
-  });
+    return worker;
+  }
  afterEach(async () => {
    await coordinator.stop(); await heartbeat.drainActiveRunExecutions();
    for (const stop of extraStops.splice(0)) await stop();
@@ -153,6 +157,78 @@ describe("deferred receipt durable dispatch", () => {
      source: "on_demand", status: "completed", reason: "task_retry", payload: { issueId: f.issue.id } });
    await heartbeat.registerDeferredReceiptWork(coordinator, () => !drain).ready; await quiet();
  });
+  it.each([false, true])("reconsiders every saved receipt during repeated company eligibility changes (expired wait: %s)", async expiredWait => {
+    const f = await fixture();
+    const ids = Array.from({ length: 178 }, (_, index) => `${(index + 1).toString(16).padStart(8, "0")}-0000-4000-8000-000000000000`);
+    await db.insert(agentWakeupRequests).values(ids.map(id => ({ id, companyId: f.companyId, agentId: f.agent.id,
+      source: "on_demand", status: "deferred_issue_execution",
+      updatedAt: new Date(Date.now() - 60_000),
+      payload: { issueId: f.issue.id, _paperclipWakeContext: { wakeCommentIds: [randomUUID()] },
+        ...(expiredWait ? { executionWait: { reason: "recovery" } } : {}) },
+    })));
+    const checked = new Set<string>();
+    await registerDeferredReceiptWork(db, coordinator, {
+      resume: async scope => { checked.add(scope.requestId!); },
+      settleTaskChat: async () => {}, createNativeBatch: () => async () => {}, schedulingSuppressed: async () => false,
+    }, () => !drain).ready;
+    for (let pass = 0; pass < 12; pass++) {
+      await db.transaction(tx => notifyDeferredExecutionEligibility(tx, { companyId: f.companyId }));
+      await vi.advanceTimersByTimeAsync(31_000);
+      await Promise.allSettled([...passes]);
+      await immediate();
+    }
+    expect(checked.size).toBe(ids.length);
+    await quiet();
+  });
+  it("dispatches task Retry past a repeatedly invalidated backlog into real process output", async () => {
+    const f = await fixture();
+    // Hold executor capacity while queue time advances; reconstruct the saved
+    // run with fresh workers at PostgreSQL time before real process execution.
+    const [occupant] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agent.id,
+      status: "running", startedAt: new Date() }).returning();
+    await db.update(agents).set({ runtimeConfig: { heartbeat: { enabled: true, intervalSec: 0,
+      wakeOnDemand: true, maxConcurrentRuns: 1 } } }).where(eq(agents.id, f.agent.id));
+    const [closed] = await db.insert(issues).values({ companyId: f.companyId, title: "Closed task history",
+      status: "done", assigneeAgentId: f.agent.id }).returning();
+    await db.insert(agentWakeupRequests).values(Array.from({ length: 178 }, (_, index) => ({
+      id: `${(index + 1).toString(16).padStart(8, "0")}-0000-4000-8000-000000000000`,
+      companyId: f.companyId, agentId: f.agent.id, source: "on_demand", status: "deferred_issue_execution",
+      payload: { issueId: closed.id, _paperclipWakeContext: { wakeCommentIds: [randomUUID()] } },
+    })));
+    await heartbeat.registerDeferredReceiptWork(coordinator, () => !drain).ready;
+    await heartbeat.registerScheduledRetryCancellationWork(coordinator, () => !drain).ready;
+    const retry = await requestTaskRetry({ db, companyId: f.companyId, issueId: f.issue.id, actorId: f.actorId,
+      observedRunId: f.source.id, idempotencyKey: randomUUID() });
+    const read = async () => (await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, retry.requestId!)))[0];
+    for (let pass = 0; pass < 12 && !(await read()).runId; pass++) {
+      await db.transaction(tx => notifyDeferredExecutionEligibility(tx, { companyId: f.companyId }));
+      await vi.advanceTimersByTimeAsync(31_000);
+      await Promise.allSettled([...passes]);
+      await immediate();
+    }
+    const receipt = await read();
+    expect(receipt.runId).toBeTruthy();
+    expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, receipt.runId!)))[0].status).toBe("queued");
+    await coordinator.stop();
+    await heartbeat.drainActiveRunExecutions();
+    vi.setSystemTime(vi.getRealSystemTime());
+    heartbeat = heartbeatService(db);
+    coordinator = createCoordinator();
+    await heartbeat.registerDeferredReceiptWork(coordinator, () => !drain).ready;
+    await heartbeat.registerScheduledRetryCancellationWork(coordinator, () => !drain).ready;
+    await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, occupant.id));
+    await heartbeat.registerRunDispatchWork(coordinator, () => !drain).ready;
+    await waitForReal(async () => {
+      await heartbeat.drainActiveRunExecutions();
+      const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, receipt.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({ id: heartbeatRuns.id }).from(heartbeatRuns)
+      .where(sql`${heartbeatRuns.contextSnapshot}->>'taskRetryRequestId' = ${retry.requestId}`)).toHaveLength(1);
+    await finish(f.issue.id);
+    await quiet();
+  });
  it("dispatches an ordinary committed task Retry into saved process output", async () => {
    const f = await fixture();
    await heartbeat.registerDeferredReceiptWork(coordinator, () => !drain).ready;
@@ -6,7 +6,7 @@ import { DELIVERY_QUEUES } from "../delivery-work-notifications.js";
 import type { DeferredExecutionScope, DeferredReceiptRecoveryOptions } from "../deferred-execution-notifications.js";
 import { registerScopedDeadlineWork } from "../scoped-deadline-work.js";

-type Entry = { id: string; companyId: string; dueAt: number };
+type Entry = { id: string; companyId: string; dueAt: number; immediate: boolean };
 type NativeIntentOptions = DeferredExecutionScope & {
  onDeferredDeadline(at: number): void;
  onError(error: unknown): void;
@@ -39,6 +39,11 @@ export function registerDeferredReceiptWork(
      },
      decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint),
      matches: (entry, scope) => !scope || entry.companyId === scope.companyId && (!scope.requestId || entry.id === scope.requestId),
+      // A busy company can invalidate a paginated scan before the pending
+      // batch drains. Keep unread ready receipts ahead of freshly reloaded
+      // 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,
      load: async scope => {
        const key = scope ? JSON.stringify({ companyId: scope.companyId, requestId: scope.requestId }) : "*";
        if (!enabled) { cursors.delete(key); return []; }
@@ -76,12 +81,15 @@ export function registerDeferredReceiptWork(
        )).orderBy(asc(agentWakeupRequests.id)).limit(100);
        const complete = rows.length < 100;
        if (complete) cursors.delete(key); else cursors.set(key, rows.at(-1)!.id);
-        return { rows: rows.map(row => ({ id: row.id, companyId: row.companyId,
-          dueAt: kind === "native" && row.status === "claimed" && row.claimedAt ? row.claimedAt.getTime() + 60_000
+        const now = Date.now();
+        return { rows: 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
-            : Date.now(),
-        })), complete };
+            : null;
+          const immediate = deadline === null || deadline <= now;
+          return { id: row.id, companyId: row.companyId, dueAt: immediate ? now : deadline, immediate };
+        }), complete };
      },
      check: async entry => {
        if (!enabled) return null;
@@ -101,7 +109,7 @@ export function registerDeferredReceiptWork(
              : 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 };
+          if (due !== null && due > Date.now()) return { ...entry, dueAt: due, immediate: false };
        }
        const errors: unknown[] = [];
        let next: number | null = null;
@@ -117,7 +125,7 @@ export function registerDeferredReceiptWork(
          catch (error) { failed(error); }
        }
        if (errors.length) throw new AggregateError(errors, `Deferred receipt ${entry.id} could not settle`);
-        return next === null ? null : { ...entry, dueAt: next };
+        return next === null ? null : { ...entry, dueAt: next, immediate: false };
      },
    }, canRun);
  }
@@ -18,6 +18,8 @@ export function registerScopedDeadlineWork<Entry extends ScheduledEntry, Scope>(
    decodeScope(hint: string): Scope | undefined;
    matches(entry: Entry, scope?: Scope): boolean;
    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;
    check(entry: Entry, signal?: AbortSignal): Promise<Entry | null>;
  },
  canRun: () => boolean,
@@ -90,7 +92,8 @@ export function registerScopedDeadlineWork<Entry extends ScheduledEntry, Scope>(
          const loaded = await options.load(scope, signal);
          const result = Array.isArray(loaded) ? { rows: loaded, complete: true } : loaded;
          for (const entry of result.rows ?? []) {
-            entries.set(entry.id, entry);
+            const pending = entries.get(entry.id);
+            entries.set(entry.id, pending && options.mergeEntry ? options.mergeEntry(pending, entry) : entry);
            for (const [other, progress] of scans) if (options.matches(entry, options.decodeScope(other))) progress.seen.add(entry.id);
          }
          if (!result.complete) { dirty.add(hint); continue; }