perf(server): recover independent native replies concurrently (#18421)

## Thinking Path

> - Paperclip manages work for AI agents.
> - A native run can commit its result before its reply is saved.
> - The response recovery worker saves replies from those committed
results.
> - It checks each candidate in series, so one task can delay unrelated
tasks.
> - This pull request checks up to four independent tasks at once.
> - Each task keeps serial checks, and the existing coordinator owns
retries and shutdown.

## Linked Issues or Issue Description

**What existing behavior does this improve?**

Recovery of committed native task replies.

**Current behavior**

The worker checks one candidate at a time in a batch of up to 25. A
database lock or a slow read for one task delays replies for other
tasks.

**Proposed behavior**

Check up to four company-and-task groups at once. Check candidates
within each group in their existing order. Keep the current batch limit
and company selection.

**Reason and benefit**

A ready task reply can be saved while another task waits for a database
lock. The change reduces waiting under a reply backlog.

**Breaking changes**

Replies for different tasks can be saved in a different order. There are
no API or schema changes. Other deadline workers keep serial dispatch.

Related work: #17696 adds the recovery worker. #17501 reduces its read
payloads. #18367 adds its timing spans. These changes are already
merged. The PR search found no open change that adds this concurrency.

## What Changed

- Add optional keyed concurrency to the shared scoped deadline worker.
- Enable a limit of four for native response recovery, keyed by company
and task.
- Await all active checks before the pass ends. Retain per-candidate
failure backoff, drain behavior and pending work.
- Test concurrency limits, task order, company separation, drain/resume,
shutdown, retries and idle wake behavior.
- Add a PostgreSQL regression that holds one task's finalization row
lock while a second task saves its exact accepted reply.
- Document the worker's concurrency and sleep contract.

## Verification

- The PostgreSQL lock regression fails with serial dispatch. It passes
with this change. Both accepted replies are saved once after the lock is
released.
- `pnpm exec vitest run
server/src/__tests__/scoped-deadline-work.test.ts
server/src/__tests__/delivery-work-coordinator.test.ts
server/src/__tests__/company-work-batch.test.ts
server/src/__tests__/deadline-work.test.ts
server/src/services/execution-preparation-timing.test.ts` — 55 tests
pass.
- `pnpm exec vitest run
server/src/services/presentation-ownership-read.test.ts` with its
documented dedicated PostgreSQL URL — 21 tests pass. The disposable
database was dropped.
- `pnpm -r typecheck` — passes.
- `pnpm build` — passes.
- One matched local replay with 25 older pending replies and one fresh
reply used the real worker, a 10-connection PostgreSQL pool and 8 ms
delay in each network direction. The saved fresh reply was first
observed after 29.193 s with serial dispatch and 9.798 s with this
change. All 26 replies were saved exactly once in both runs, with no
worker errors. Both disposable databases were dropped. This is one
synthetic workload pair, not a staging measurement.
- `pnpm test:run` was attempted. Embedded PostgreSQL could not
initialize its template after five attempts, so the broad run was
stopped. The changed PostgreSQL tests pass on the separate local server.
Remote CI is green. This PR does not claim a measured staging speedup.

- The [first CI browser
shard](https://github.com/paperclipai/paperclip/actions/runs/41162257463/job/124422204506)
failed during company cleanup after its task assertions and empty-run
check passed. Company deletion hit the
`activity_log_company_id_companies_id_fk` constraint. The fixture uses
paused process agents and reported no runs. The one failed-jobs rerun
passed all 45 browser tests. All CI checks are now green.

## Risks

- Four checks can use more database connections than serial recovery.
The limit applies to this queue only; the batch limit stays at 25.
- The shared helper has more dispatch logic. Its default remains serial.
Tests cover task order, retry deadlines, drain and waiting for active
checks during shutdown.
- This improves reply recovery under backlog or lock contention. It does
not establish a target for total agent-turn latency.

## Model Used

OpenAI Codex, based on GPT-6, with code editing, tool use and local test
execution. This session does not expose a more specific model ID or
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
- [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>

doc/idle-sleep-safety.md

@@ -81,6 +81,17 @@ Independent SQL writers and additional processes need an explicit wake protocol;
 process-local notifications do not provide cross-replica delivery. Worker quiet
 alone is not evidence that the host or database actually suspended.
 
+## Native reply recovery
+
+Committed native replies use the shared delivery coordinator and its persisted
+retry deadlines. One pass checks at most four independent tasks at a time.
+Checks for the same company and task run in candidate order within the batch.
+Other scoped deadline workers remain serial unless their owner enables concurrency.
+Drain retains unstarted checks, and shutdown waits for every admitted check before
+releasing tracked work. Completed history creates no recurring discovery reads;
+new committed work reactivates the existing notification path. Pending reply
+obligations keep their existing sleep blockers.
+
 ## Scheduled routine wake protocol
 
 Historical failed ephemeral leases and terminal custom-image setup sessions can

server/src/__tests__/scoped-deadline-work.test.ts

@@ -37,3 +37,166 @@ it("preserves a subsecond replacement deadline when the owner does not require p
   expect(check).toHaveBeenCalledOnce();
   expect(onError).not.toHaveBeenCalled();
 });
+
+type Entry = { id: string; companyId: string; issueId: string; dueAt: number };
+
+function concurrentFixture(entries: Entry[], limit = 4, concurrent = true) {
+  vi.useFakeTimers();
+  const onError = vi.fn();
+  const owner = installDatabaseWorkSignals({ execute: vi.fn(async () => []) });
+  let canRun = true;
+  let releaseAll = false;
+  const releases = new Map<string, () => void>();
+  const signals: AbortSignal[] = [];
+  const started: string[] = [];
+  const active = new Set<string>();
+  const completed = new Set<string>();
+  let maximumActive = 0;
+  const check = vi.fn(async (entry: Entry, signal?: AbortSignal) => {
+    started.push(entry.id);
+    signals.push(signal!);
+    active.add(entry.id);
+    maximumActive = Math.max(maximumActive, active.size);
+    if (!releaseAll) await new Promise<void>(resolve => releases.set(entry.id, resolve));
+    active.delete(entry.id);
+    completed.add(entry.id);
+    return null;
+  });
+  const load = vi.fn(async () => entries.filter(entry => !completed.has(entry.id)));
+  coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => canRun, onError });
+  const options = {
+    queue: DELIVERY_QUEUES.nativeResponsePresentation,
+    concurrentChecks: concurrent ? { limit, key: (entry: Entry) => `${entry.companyId}:${entry.issueId}` } : undefined,
+    decodeScope: () => undefined,
+    matches: () => true,
+    load,
+    check,
+  };
+  const worker = registerScopedDeadlineWork(coordinator, options, () => canRun);
+  return {
+    worker, check, load, started, active, signals, onError, entries,
+    maximumActive: () => maximumActive,
+    pause: () => { canRun = false; },
+    resume: () => { canRun = true; worker.wake(); },
+    release: (id: string) => releases.get(id)?.(),
+    releaseAll: () => { releaseAll = true; for (const resolve of releases.values()) resolve(); },
+  };
+}
+
+const entry = (id: string, issueId = id, companyId = "company"): Entry => ({ id, issueId, companyId, dueAt: 0 });
+
+it("checks independent tasks concurrently with a bounded number of active checks", async () => {
+  const f = concurrentFixture(Array.from({ length: 6 }, (_, index) => entry(String(index))));
+  try {
+    await vi.waitFor(() => expect(f.started).toHaveLength(4), { timeout: 100 });
+    expect(f.maximumActive()).toBe(4);
+    f.release("1");
+    await vi.waitFor(() => expect(f.started).toHaveLength(5), { timeout: 100 });
+    expect(f.active.has("0")).toBe(true);
+    expect(f.maximumActive()).toBe(4);
+  } finally { f.releaseAll(); await f.worker.ready; }
+  expect(f.started).toHaveLength(6);
+  expect(new Set(f.started).size).toBe(6);
+  expect(coordinator!.nextWakeAt()).toBeNull();
+  await vi.advanceTimersByTimeAsync(60_000);
+  expect(f.load).toHaveBeenCalledOnce();
+  expect(f.check).toHaveBeenCalledTimes(6);
+  f.entries.push(entry("fresh"));
+  f.worker.invalidate();
+  await vi.advanceTimersByTimeAsync(1);
+  expect(f.load).toHaveBeenCalledTimes(2);
+  expect(f.check).toHaveBeenCalledTimes(7);
+  expect(f.started.at(-1)).toBe("fresh");
+  expect(coordinator!.nextWakeAt()).toBeNull();
+  expect(f.onError).not.toHaveBeenCalled();
+});
+
+it("serializes each task in candidate order while other tasks and companies progress", async () => {
+  const f = concurrentFixture([
+    entry("1", "shared"), entry("2", "shared"), entry("3", "other"), entry("4", "shared", "other-company"),
+  ]);
+  try {
+    await vi.waitFor(() => expect(new Set(f.started)).toEqual(new Set(["1", "3", "4"])), { timeout: 100 });
+    f.release("3");
+    await vi.waitFor(() => expect(f.active.has("3")).toBe(false), { timeout: 100 });
+    expect(f.started).not.toContain("2");
+    f.release("1");
+    await vi.waitFor(() => expect(f.started).toHaveLength(4), { timeout: 100 });
+    expect(f.started.filter(id => id === "1" || id === "2")).toEqual(["1", "2"]);
+    expect(f.active.has("1")).toBe(false);
+  } finally { f.releaseAll(); await f.worker.ready; }
+  expect(f.maximumActive()).toBe(3);
+  expect(f.onError).not.toHaveBeenCalled();
+});
+
+it("preserves global candidate order and serial dispatch unless concurrency is enabled", async () => {
+  const f = concurrentFixture([entry("1", "shared"), entry("2", "other"), entry("3", "shared")], 4, false);
+  try {
+    await vi.waitFor(() => expect(f.started).toEqual(["1"]), { timeout: 100 });
+    f.release("1");
+    await vi.waitFor(() => expect(f.started).toEqual(["1", "2"]), { timeout: 100 });
+  } finally { f.releaseAll(); await f.worker.ready; }
+  expect(f.started).toEqual(["1", "2", "3"]);
+  expect(f.maximumActive()).toBe(1);
+});
+
+it("retains unstarted checks through drain and resumes without idle discovery reads", async () => {
+  const f = concurrentFixture(Array.from({ length: 6 }, (_, index) => entry(String(index))));
+  try {
+    await vi.waitFor(() => expect(f.started).toHaveLength(4), { timeout: 100 });
+    f.pause();
+  } finally { f.releaseAll(); await f.worker.ready; }
+  expect(f.started).toHaveLength(4);
+  await vi.advanceTimersByTimeAsync(60_000);
+  expect(f.load).toHaveBeenCalledOnce();
+  expect(f.started).toHaveLength(4);
+  f.resume();
+  await vi.advanceTimersByTimeAsync(1);
+  expect(f.started).toHaveLength(6);
+  expect(coordinator!.nextWakeAt()).toBeNull();
+  expect(f.onError).not.toHaveBeenCalled();
+});
+
+it("stops starting checks and awaits every admitted check during shutdown", async () => {
+  const f = concurrentFixture(Array.from({ length: 6 }, (_, index) => entry(String(index))));
+  let stopped = false;
+  let stop: Promise<void> | undefined;
+  try {
+    await vi.waitFor(() => expect(f.started).toHaveLength(4), { timeout: 100 });
+    stop = coordinator!.stop().then(() => { stopped = true; });
+    expect(f.signals.every(signal => signal.aborted)).toBe(true);
+    expect(stopped).toBe(false);
+    f.release("0");
+    await vi.waitFor(() => expect(f.active.has("0")).toBe(false), { timeout: 100 });
+    expect(stopped).toBe(false);
+    expect(f.started).toHaveLength(4);
+  } finally { f.releaseAll(); await f.worker.ready; await stop; }
+  expect(stopped).toBe(true);
+  expect(f.started).toHaveLength(4);
+});
+
+it("keeps failures isolated and retries their deadline without repeating successful checks", async () => {
+  vi.useFakeTimers();
+  const owner = installDatabaseWorkSignals({ execute: vi.fn(async () => []) });
+  const onError = vi.fn();
+  coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => true, onError });
+  const entries = [entry("1", "shared"), entry("2", "shared"), entry("3")];
+  const load = vi.fn(async () => entries);
+  const check = vi.fn(async (_entry: Entry) => null).mockRejectedValueOnce(new Error("temporary failure"));
+  const options = {
+    queue: DELIVERY_QUEUES.nativeResponsePresentation,
+    concurrentChecks: { limit: 4, key: (entry: Entry) => `${entry.companyId}:${entry.issueId}` },
+    decodeScope: () => undefined, matches: () => true, load, check,
+  };
+  const worker = registerScopedDeadlineWork(coordinator, options, () => true);
+  await worker.ready;
+  expect(check.mock.calls.map(([entry]) => entry.id).sort()).toEqual(["1", "2", "3"]);
+  expect(onError).toHaveBeenCalledOnce();
+  expect(onError.mock.calls[0][0]).toBeInstanceOf(AggregateError);
+  await vi.advanceTimersByTimeAsync(29_999);
+  expect(check).toHaveBeenCalledTimes(3);
+  await vi.advanceTimersByTimeAsync(1);
+  expect(check.mock.calls.map(([entry]) => entry.id)).toEqual(["1", "3", "2", "1"]);
+  expect(load).toHaveBeenCalledOnce();
+  expect(coordinator.nextWakeAt()).toBeNull();
+});

server/src/services/native-runtime/native-response-presentation-work.ts

@@ -56,6 +56,7 @@ export function nativeResponsePresentationWork(db: Db) {
   function register(coordinator: ReturnType<typeof createDeliveryWorkCoordinator>, canRun: () => boolean) {
     const worker = registerScopedDeadlineWork(coordinator, {
       queue: DELIVERY_QUEUES.nativeResponsePresentation,
+      concurrentChecks: { limit: 4, key: (entry: Entry) => `${entry.companyId}:${entry.issueId}` },
       decodeScope: (hint: string): NativeResponsePresentationScope | undefined => hint === "*" ? undefined : JSON.parse(hint),
       matches: (entry: Entry, scope?: NativeResponsePresentationScope) => !scope || entry.companyId === scope.companyId && (!scope.issueId || entry.issueId === scope.issueId) && (!scope.runId || entry.id === scope.runId) && (!scope.agentId || entry.agentId === scope.agentId),
       load: (scope, signal) => load(scope, signal, canRun),

server/src/services/presentation-ownership-read.test.ts

@@ -10,6 +10,7 @@ import { getConversationOwnershipBlocker } from "./conversation-continuation.js"
 import { isExternalChatPresentationContext } from "./heartbeat-run-summary.js";
 import { finalizeNativeRun, recoverCommittedNativeChatResponse, repairCommittedNativeChatResponse } from "./native-runtime/native-run-finalizer.js";
 import { nativeResponsePresentationWork } from "./native-runtime/native-response-presentation-work.js";
+import { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
 
 import { resolveExternalTestDatabaseUrl } from "../__tests__/helpers/external-test-database.js";
 
@@ -24,10 +25,12 @@ describe.skipIf(!support.supported)("presentation and ownership PostgreSQL read
   }, 60_000);
   afterAll(async () => { await db?.$client.end(); await database?.cleanup(); });
 
-  async function fixture(patch: Partial<typeof heartbeatRuns.$inferInsert> = {}) {
-    const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID();
-    await db.insert(companies).values({ id: companyId, name: "Read fixture", issuePrefix: companyId.slice(0, 8) });
-    await db.insert(agents).values({ id: agentId, companyId, name: "Read agent", adapterType: "codex_local" });
+  async function fixture(patch: Partial<typeof heartbeatRuns.$inferInsert> = {}, identity?: { companyId: string; agentId: string }) {
+    const companyId = identity?.companyId ?? randomUUID(), agentId = identity?.agentId ?? randomUUID(), issueId = randomUUID();
+    if (!identity) {
+      await db.insert(companies).values({ id: companyId, name: "Read fixture", issuePrefix: companyId.slice(0, 8) });
+      await db.insert(agents).values({ id: agentId, companyId, name: "Read agent", adapterType: "codex_local" });
+    }
     await db.insert(issues).values({ id: issueId, companyId, title: "Read task", status: "blocked", assigneeAgentId: agentId });
     const payload = randomBytes(128 * 1024).toString("base64");
     const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, nativeIssueId: issueId,
@@ -138,10 +141,10 @@ describe.skipIf(!support.supported)("presentation and ownership PostgreSQL read
     expect(await repairCommittedNativeChatResponse(db, { companyId: f.companyId, issueId: other.id, runId: f.run.id })).toBe(false);
   });
 
-  async function unpublishedAcceptedReply() {
+  async function unpublishedAcceptedReply(id?: string, identity?: { companyId: string; agentId: string }) {
     const payload = randomBytes(128 * 1024).toString("base64");
-    const f = await fixture({ runtimeMode: "native", status: "running", finishedAt: null,
-      contextSnapshot: { source: "internal", prompt: payload, paperclipWake: { prompt: payload } } });
+    const f = await fixture({ id, runtimeMode: "native", status: "running", finishedAt: null,
+      contextSnapshot: { source: "internal", prompt: payload, paperclipWake: { prompt: payload } } }, identity);
     const [contract] = await db.insert(completionContracts).values({ companyId: f.companyId, issueId: f.issueId,
       revision: 1, schemaVersion: "paperclip.completion-contract.v1", policyVersion: "test", risk: "standard",
       completionAuthority: "server_arbiter", incompleteCriteriaPolicy: "preserve_non_terminal",
@@ -173,6 +176,40 @@ describe.skipIf(!support.supported)("presentation and ownership PostgreSQL read
     return { ...f, before, reply };
   }
 
+  it("saves an independent task reply while an earlier task waits for its database lock", async () => {
+    const suffix = randomUUID().replaceAll("-", "").slice(0, 12);
+    const first = await unpublishedAcceptedReply(`00000000-0000-4000-8000-${suffix}`);
+    const second = await unpublishedAcceptedReply(`00000000-0000-4001-8000-${suffix}`, first);
+    let release!: () => void;
+    let locked!: () => void;
+    const hold = new Promise<void>(resolve => { release = resolve; });
+    const acquired = new Promise<void>(resolve => { locked = resolve; });
+    const blocker = db.transaction(async tx => {
+      await tx.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, first.run.id)).for("update");
+      locked(); await hold;
+    });
+    await Promise.race([acquired, blocker]);
+    const onError = vi.fn();
+    const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError });
+    const worker = nativeResponsePresentationWork(db).register(coordinator, () => true);
+    try {
+      await vi.waitFor(async () => {
+        const rows = await db.select().from(issueComments).where(eq(issueComments.createdByRunId, second.run.id));
+        expect(rows).toHaveLength(1);
+        expect(rows[0].body).toBe(second.reply);
+      }, { timeout: 5000 });
+      expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, first.run.id))).toHaveLength(0);
+    } finally {
+      release(); await blocker; await worker.ready; worker.stop(); await coordinator.stop();
+    }
+    for (const f of [first, second]) {
+      const rows = await db.select().from(issueComments).where(eq(issueComments.createdByRunId, f.run.id));
+      expect(rows).toHaveLength(1);
+      expect(rows[0].body).toBe(f.reply);
+    }
+    expect(onError).not.toHaveBeenCalled();
+  });
+
   async function repairFixture(patch: Partial<typeof heartbeatRuns.$inferInsert>) {
 
 

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

@@ -12,6 +12,7 @@ export function registerScopedDeadlineWork<Entry extends ScheduledEntry, Scope>(
     queue: DeliveryQueue;
     batchSize?: number;
     minimumPassIntervalMs?: number;
+    concurrentChecks?: { limit: number; key(entry: Entry): string };
     decodeScope(hint: string): Scope | undefined;
     matches(entry: Entry, scope?: Scope): boolean;
     load(scope?: Scope, signal?: AbortSignal): Promise<Entry[] | { rows: Entry[] | null; complete: boolean }>;
@@ -19,6 +20,9 @@ export function registerScopedDeadlineWork<Entry extends ScheduledEntry, Scope>(
   },
   canRun: () => boolean,
 ) {
+  if (options.concurrentChecks && (!Number.isInteger(options.concurrentChecks.limit) || options.concurrentChecks.limit < 1)) {
+    throw new RangeError("Concurrent check limit must be a positive integer");
+  }
   const dirty = new Set<string>(["*"]);
   const versions = new Map<string, number>();
   const scans = new Map<string, { version: number; seen: Set<string> }>();
@@ -94,14 +98,30 @@ export function registerScopedDeadlineWork<Entry extends ScheduledEntry, 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));
       const batch = takeCompanyWorkBatch(due, options.batchSize ?? 25, lastCompany); lastCompany = batch.lastCompany;
+      const groups = new Map<string, Entry[]>();
       for (const entry of batch.entries) {
-        if (signal.aborted || !canRun()) break;
-        try {
-          const current = await options.check(entry, signal);
-          if (current) entries.set(entry.id, current); else entries.delete(entry.id);
-          failures.delete(entry.id);
-        } catch (error) { failed(failures, entry.id); errors.push(error); }
+        const key = options.concurrentChecks?.key(entry) ?? "";
+        const group = groups.get(key) ?? [];
+        group.push(entry); groups.set(key, group);
       }
+      const pending = [...groups.values()];
+      let nextGroup = 0;
+      const checkGroups = async () => {
+        while (!signal.aborted && canRun()) {
+          const group = pending[nextGroup++];
+          if (!group) return;
+          for (const entry of group) {
+            if (signal.aborted || !canRun()) return;
+            try {
+              const current = await options.check(entry, signal);
+              if (current) entries.set(entry.id, current); else entries.delete(entry.id);
+              failures.delete(entry.id);
+            } catch (error) { failed(failures, entry.id); errors.push(error); }
+          }
+        }
+      };
+      const checks = await Promise.allSettled(Array.from({ length: Math.min(options.concurrentChecks?.limit ?? 1, pending.length) }, checkGroups));
+      for (const result of checks) if (result.status === "rejected") errors.push(result.reason);
       if (errors.length) throw new AggregateError(errors, `${options.queue} could not settle`);
     },
     // Retain scopes/deadlines for resume without retaining a settled write's