fix(native): admit independent replies during contended publication (#19726)

## Thinking Path

> - Paperclip manages work done by AI agents.
> - A completed native run must deliver its saved final response.
> - The response worker also recovers retained replies that can wait on
publication locks.
> - A pass waited for its old publication groups even when other check
slots were free.
> - This could delay an independent final reply beyond a full page of
retained candidates.
> - This PR enables the existing shared coordinator option that admits
independent due work during an active pass.
> - Existing authority checks, resource keys and the four-check limit
still control publication.

## Linked Issues or Issue Description

**What happened?**
A native run completed after replacement recovery. Its final response
and accounting were saved, but its final issue reply was missing. The
response queue contained more than 100 retained chat replies and
repeated endpoint-lock failures. The existing worker did not admit
independent due entries until its current groups finished.

**Expected behavior**
A contended publication owner must not hold free worker slots. An
independent accepted final reply must publish once, including when its
candidate is on a later SQL page.

**Steps to reproduce**
Create 100 accepted, committed replies that share one publication owner.
Create an independent reply after them in candidate order. Hold the old
coordinator rows with a database transaction and start the real response
worker. Before this change, the independent reply stays unpublished
until the old group is released.

Related: #18977 changes response selection in standing conversations.
This PR changes only worker admission and preserves response selection.

## What Changed

- Enable `admitDueWhileActive` in the native response worker's existing
shared coordinator configuration.
- Add a real PostgreSQL regression with the full 100-row SQL page, the
default dispatch batch, real row-lock contention and a later-page
independent reply.
- Verify one comment per reply after settlement and no further SQL or
scheduled deadline across 60 seconds of idle time.
- Update one existing principal/rescope integration assertion to observe
the reconciled invalidation revision. Keep its actual per-run replies,
authorization, publication and idle checks.

## Verification

- Regression on the unchanged worker: the independent reply remained
unpublished while the first 100 replies were locked. The real PostgreSQL
test failed as expected.
- Final head `748fb9609104c016c33ff7caf282ba9e15e88495`: all six
publication regressions and both principal/rescope controls passed (8/8,
47.96 seconds). Checks cover later-page admission, exact per-run comment
counts, retry and rollback behavior, authorization, and 60 seconds with
no idle SQL or scheduled deadline.
- The tests used the canonical PostgreSQL wrapper and one owned
disposable PostgreSQL 18.1 fixture. All clients and test databases were
closed before fixture cleanup.
- The current-head [full CI
run](https://github.com/paperclipai/paperclip/actions/runs/42338953241)
supplies typecheck, build and broad test coverage. All current-head
checks passed; the merge gate is green.
- Local full typecheck was blocked by unbuilt dependencies; it is not
counted as a pass. The prior timer-state CI failure and the final
successful rescope-generation regression remain recorded.
- Greptile reviewed the final head at 5/5 with no unresolved threads.
`git diff --check` passed.
- Deployed acceptance remains separate. No serving instance was changed
by the PR tests.

## Risks

The worker can now use free slots during a slow pass. The shared
coordinator still limits checks to four and preserves task, endpoint and
principal exclusions. Authorization, result identity, retry limits,
response selection and transaction ownership are unchanged. The
regression checks idle quiet after all concrete obligations settle.

## Model Used

OpenAI GPT-6.1 (`gpt-6.1-sol`), with reasoning, code execution and tool
use. 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
- [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__/chat-channels.integration.test.ts

@@ -76505,6 +76505,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
     const scopedWork = await import("../services/scoped-deadline-work.js");
     const register = scopedWork.registerScopedDeadlineWork;
     let invalidated = false;
+    let invalidationRevision: string | undefined;
     let invalidateSource: (() => void) | undefined;
     const registration = vi.spyOn(scopedWork, "registerScopedDeadlineWork").mockImplementation((owner, options, canRun) =>
       register(owner, { ...options, load: async (scope, signal) => {
@@ -76518,14 +76519,21 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
         return entries;
       } }, canRun));
     const worker = nativeResponsePresentationWork(db).register(coordinator, () => true);
-    invalidateSource = () => worker.invalidate();
+    invalidateSource = () => {
+      worker.invalidate();
+      invalidationRevision = worker.reconciliationSnapshot().revision;
+    };
     registration.mockRestore();
     try {
       await worker.ready;
       await expect.poll(() => idleWorkSnapshot().active, { timeout: 10_000 }).toBe(0);
       if (mode === "rescoped") {
         expect(invalidated).toBe(true);
-        expect(coordinator.nextWakeAt()).not.toBeNull();
+        expect(invalidationRevision).toBeTypeOf("string");
+        // Independent admission can finish the rescope before this check.
+        // Observe its settled generation rather than an intermediate timer.
+        await expect.poll(() => worker.reconciliationSnapshot().reconciled, { timeout: 10_000 }).toBe(true);
+        expect(worker.reconciliationSnapshot().revision).not.toBe(invalidationRevision);
       }
       await expect.poll(async () => Promise.all(contexts.map(async context =>
         (await db.select({ id: issueComments.id }).from(issueComments)

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

@@ -84,7 +84,7 @@ export function nativeResponsePresentationWork(db: Db) {
       queue: DELIVERY_QUEUES.nativeResponsePresentation,
       // Endpoint and principal IDs are only dispatch hints. Recovery still locks and checks
       // the complete current source authority inside its own transaction.
-      concurrentChecks: { limit: 4, key: (entry: Entry) => [
+      concurrentChecks: { limit: 4, admitDueWhileActive: true, key: (entry: Entry) => [
         `issue:${entry.companyId}:${entry.issueId}`,
         ...(entry.endpointId ? [`endpoint:${entry.companyId}:${entry.endpointId}`] : []),
         ...(entry.principalId ? [`principal:${entry.companyId}:${entry.principalId}`] : []),

server/src/services/native-runtime/native-response-publication-retry.test.ts

@@ -1,6 +1,6 @@
 import { randomUUID } from "node:crypto";
-import { and, eq, sql } from "drizzle-orm";
-import { afterAll, beforeAll, describe, expect, it } from "vitest";
+import { and, eq, inArray, sql } from "drizzle-orm";
+import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
 import { activityLog, agents, companies, completionContracts, createDb, heartbeatRunEvents, heartbeatRuns, issueComments, issues,
   nativeRunFinalizations, type Db } from "@paperclipai/db";
 import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js";
@@ -8,6 +8,8 @@ import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external
 import { NativeRunCoordinatorStore } from "./native-run-coordinator-store.js";
 import { finalizeNativeRun } from "./native-run-finalizer.js";
 import { nativeResponsePresentationWork } from "./native-response-presentation-work.js";
+import { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js";
+import { idleWorkSnapshot } from "../task-admission.js";
 
 // An accepted native task result whose final reply was never published must
 // not be abandoned after one refused or failed publication attempt. Retries
@@ -25,20 +27,21 @@ describe.skipIf(!support.supported)("native final reply publication retry", () =
   afterAll(async () => { await db?.$client.end(); await database?.cleanup(); });
 
   const reply = "Keyboard shortcuts are now available on every plan.";
-  async function fixture() {
-    const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID();
+  async function fixture(options: { companyId?: string; agentId?: string; runId?: string; issueId?: string; contract?: typeof completionContracts.$inferSelect } = {}) {
+    const companyId = options.companyId ?? randomUUID(), agentId = options.agentId ?? randomUUID();
+    const issueId = options.issueId ?? randomUUID(), runId = options.runId ?? randomUUID();
     const sessionId = randomUUID(), runnerId = randomUUID();
-    await db.insert(companies).values({ id: companyId, name: "Reply publication fixture", issuePrefix: companyId.slice(0, 8) });
-    await db.insert(agents).values({ id: agentId, companyId, name: "Writer", adapterType: "paperclip_runner" });
-    await db.insert(issues).values({ id: issueId, companyId, title: "Short changelog blurb", status: "in_progress", assigneeAgentId: agentId });
-    const [contract] = await db.insert(completionContracts).values({ companyId, issueId, revision: 1,
+    if (!options.companyId) await db.insert(companies).values({ id: companyId, name: "Reply publication fixture", issuePrefix: companyId.slice(0, 8) });
+    if (!options.agentId) await db.insert(agents).values({ id: agentId, companyId, name: "Writer", adapterType: "paperclip_runner" });
+    if (!options.issueId) await db.insert(issues).values({ id: issueId, companyId, title: "Short changelog blurb", status: "in_progress", assigneeAgentId: agentId });
+    const [contract] = options.contract ? [options.contract] : await db.insert(completionContracts).values({ companyId, issueId, revision: 1,
       schemaVersion: "paperclip.completion-contract.v1", policyVersion: "test", risk: "standard", completionAuthority: "server_arbiter",
       incompleteCriteriaPolicy: "preserve_non_terminal", contractJson: { revision: "test", objective: "Write the blurb", criteria: [] },
       canonicalSha256: "fixture-publication-contract", createdByActorType: "system", createdByActorId: "test" }).returning();
     await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, nativeIssueId: issueId, nativeSessionId: sessionId,
       runnerInstanceId: runnerId, completionContractId: contract.id, completionContractSha256: contract.canonicalSha256,
       status: "running", runtimeMode: "native", contextSnapshot: { issueId, source: "assignment" } });
-    await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, issueId));
+    await db.update(issues).set({ executionRunId: runId, status: "in_progress" }).where(eq(issues.id, issueId));
     const port = new NativeRunCoordinatorStore(db, { companyId, issueId, runId, agentId, normalizedSessionId: sessionId,
       runnerSourceInstanceId: runnerId, completionContractId: contract.id, completionContractSha256: contract.canonicalSha256,
       completionContractRevision: "test", completionContractCriterionIds: [] });
@@ -54,7 +57,7 @@ describe.skipIf(!support.supported)("native final reply publication retry", () =
       payload: { kind: "agentMessage", channel: "final", providerPhase: "final", text: reply } };
     const saveFinalReply = () => db.insert(heartbeatRunEvents).values({ companyId, runId, agentId, seq: 170, eventType: "item.completed",
       payload: { prpEvent } });
-    return { companyId, issueId, runId, agentId, saveFinalReply };
+    return { companyId, issueId, runId, agentId, contract, saveFinalReply };
   }
   const comments = (runId: string) => db.select().from(issueComments).where(eq(issueComments.createdByRunId, runId));
   const run = async (runId: string) => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)))[0];
@@ -64,6 +67,74 @@ describe.skipIf(!support.supported)("native final reply publication retry", () =
   const makeDue = (runId: string) => db.execute(sql`update heartbeat_runs set result_json = jsonb_set(result_json,
     '{responsePublication,nextAttemptAt}', to_jsonb('2000-01-01T00:00:00.000Z'::text)) where id = ${runId}`);
 
+  it("publishes a later-page reply while older publication groups remain contended", async () => {
+    const first = await fixture({ runId: "00000000-0000-4000-8000-000000000001" });
+    const group = { companyId: first.companyId, agentId: first.agentId };
+    const older = [first];
+    // A full real SQL page of retained replies shares one publication owner.
+    // This is the same grouping boundary as retained replies to one endpoint.
+    for (let i = 2; i <= 100; i++) older.push(await fixture({ ...group,
+      issueId: first.issueId, contract: first.contract,
+      runId: `00000000-0000-4000-8000-${i.toString(16).padStart(12, "0")}` }));
+    const independent = await fixture({ ...group, runId: "ffffffff-ffff-4fff-8fff-ffffffffffff" });
+    for (const f of [...older, independent]) await f.saveFinalReply();
+
+    let unlock!: () => void, locked!: () => void;
+    const release = new Promise<void>(resolve => { unlock = resolve; });
+    const acquired = new Promise<void>(resolve => { locked = resolve; });
+    const holder = db.transaction(async tx => {
+      await tx.select({ id: nativeRunFinalizations.runId }).from(nativeRunFinalizations)
+        .where(inArray(nativeRunFinalizations.runId, older.map(f => f.runId))).for("update");
+      locked();
+      await release;
+    });
+    await acquired;
+    const errors: unknown[] = [];
+    const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true,
+      onError: error => errors.push(error) });
+    const worker = nativeResponsePresentationWork(db).register(coordinator, () => true);
+    let firstPassSettled = false;
+    const startup = worker.ready.then(() => { firstPassSettled = true; });
+    const debug = db.$client.options.debug;
+    let queries = 0;
+    db.$client.options.debug = () => { queries++; };
+    try {
+      await vi.waitFor(async () => expect(await comments(independent.runId)).toHaveLength(1), { timeout: 3000 });
+      expect(firstPassSettled).toBe(false);
+      expect(await comments(first.runId)).toEqual([]);
+      expect((await comments(independent.runId))[0].body).toBe(reply);
+      unlock();
+      await holder;
+      await startup;
+      await vi.waitFor(async () => {
+        const posted = await db.select({ runId: issueComments.createdByRunId }).from(issueComments)
+          .where(eq(issueComments.issueId, first.issueId));
+        expect(posted.map(comment => comment.runId).sort()).toEqual(older.map(f => f.runId).sort());
+      }, { timeout: 20_000 });
+      await vi.waitFor(() => {
+        expect(coordinator.nextWakeAt()).toBeNull();
+        expect(idleWorkSnapshot().active).toBe(0);
+      }, { timeout: 5000 });
+      // Neither retained history nor the completed contention owns an idle scan.
+      const settledQueries = queries;
+      vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
+      await vi.advanceTimersByTimeAsync(60_000);
+      expect(queries).toBe(settledQueries);
+      expect(coordinator.nextWakeAt()).toBeNull();
+      vi.useRealTimers();
+      expect(await comments(independent.runId)).toHaveLength(1);
+      expect(errors).toEqual([]);
+    } finally {
+      vi.useRealTimers();
+      unlock();
+      await holder;
+      await startup;
+      worker.stop();
+      await coordinator.stop();
+      db.$client.options.debug = debug;
+    }
+  }, 120_000);
+
   it("keeps an accepted reply scheduled after a single refused publication and posts it exactly once", async () => {
     const f = await fixture();
     const work = nativeResponsePresentationWork(db);