fix: avoid response recovery self-contention (#18715)

## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - The server must present accepted agent responses as comments under current chat authority. > - The recovery worker can check four tasks at the same time. > - Different tasks can share a chat endpoint or external principal. Their authorization transactions then compete for endpoint or identity locks. > - A failed transaction leaves the accepted response pending and repeats its source queries on a later retry. > - This pull request groups checks that share a task, endpoint or principal before dispatch. It keeps every transactional authority check and lock. > - The benefit is less failed recovery work and prompt, exactly-once presentation of valid accepted responses. ## Linked Issues or Issue Description Refs #17714 for a separate publication-commit lock-order fix. Refs #18686 for separate run-list JSON work. Neither changes this recovery admission path. **What happened?** A response-presentation worker reported `could not obtain lock on row in relation "chat_endpoints"`. Four real PostgreSQL recovery fixtures reproduced three such failures in one pass with no external ingress. The worker grouped by task, while authorization locks shared endpoints and principals. Review also exposed two valid Slack endpoints in one workspace that share one principal and compete for its try-advisory lock. **Expected behavior** Recovery checks that share an endpoint or principal should retain task order without competing with one another. Unrelated endpoints should make progress. The server must still reject stale authority and retry legitimate external contention. **Steps to reproduce** Run the shared-endpoint regression or the portable comparison helper linked below. The fixture creates four accepted historical responses for distinct tasks and principals on one actual endpoint. ## What Changed - Read optional endpoint and principal scheduling hints through a company-scoped join to the canonical wake action. The join does not change eligibility or grant authority. - Extend existing bounded check groups to merge intersecting resource keys. Preserve task order, transitive conflicts, independent-group progress, shutdown, drain and pending scopes. - Add real PostgreSQL controls for shared endpoints, a principal shared across two endpoints, an observed blocked transaction with independent endpoint progress, external locks, exact selected output, and endpoint revocation despite a forged hint. - Add a portable helper that materializes the pinned baseline with `git show`, runs the existing real fixture, retains results, and restores temporary files. ## Verification - Final principal-extension head `57047441b28765ed1cf49b8924efc69f5c03dcb4`: 34 focused scheduler/chat controls passed on Node 24.22.2 / pnpm 9.15.4. Independent exact-head source review is green. Full local `pnpm -r typecheck` and `pnpm build` pass on this head. Fresh Greptile and Paperclip reviews are both 5/5 with no findings. All 59 final-head contexts are complete: 55 success, 3 skipped, and the neutral Paperclip review at 5/5. There are no failures or pending checks. - Earlier endpoint-only head `eea437e7edbe7eb41eb880d4d8fd6d99a9c247da`: full local typecheck/build and all remote CI passed. The redundant local full suite was interrupted for this review fix after 956 passing chat cases; it is not a full local pass. - Final shared-principal counts-only comparison completed the same two exact responses: product calls 165 → 147, rows 114 → 100, outer contention failures 1 → 0. SQL query errors stayed zero; advisory-lock refusal is not a failed SQL query. - The earlier endpoint-only study completed five alternating pairs / ten measured runs: median calls 443 → 293, rows 314 → 200, query errors 6 → 0 for four exact responses. Its instrumented shared-host CPU data predates the principal extension and is not a final-head or hosted CPU claim. Retry clocks were accelerated; wall time is not hosted latency. - [Raw samples, final principal work counts, portable commands and failure accounting](https://github.com/paperclipai/paperclip/blob/57047441b28765ed1cf49b8924efc69f5c03dcb4/doc/performance/2026-11-03-response-presentation-contention.md). - Initial server4 CI reported three provider calls instead of two in the unchanged Anthropic identity fixture. The exact local case passed; the full 88-case file passed with observations showing the two expected identity-route calls. The third CI call's origin remains unknown. One bounded rerun of the failed job and its dependent CI aggregate passed. No assertion or source change was made for that failure. - The causal benchmark baseline worker/query is `82c033ed127d07862e6765aa4b69a3cf339bde81`. The PR base is `36df8933f0`. They are not claims about the currently serving revision. ## Risks - Connected resource groups can reduce parallelism when tasks link several endpoints. This preserves the existing task ordering contract. - Missing or stale endpoint/principal hints retain current transactional checks. Other workers and ingress can still cause legitimate NOWAIT retries. - There is no schema, API, permission or provider-transport change. Endpoint revocation and company isolation remain enforced by their existing owners. ## Model Used OpenAI GPT-6.1 Sol (`gpt-6.1-sol`), with tool use and code execution, produced and reviewed this change. 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>
2026-11-04 11:01:16 +01:00 · 2026-11-03 06:04:25 -06:00
1 parent 76ee5d0886
commit bf7f77d8cd
@@ -0,0 +1,133 @@
+// Run only in a clean, isolated checkout. Reuses the real chat integration fixture.
+import assert from "node:assert/strict";
+import { existsSync, mkdirSync, mkdtempSync, openSync, closeSync, readFileSync, rmSync, writeFileSync } from "node:fs";
+import { tmpdir } from "node:os";
+import path from "node:path";
+import { fileURLToPath } from "node:url";
+import { execFileSync, spawnSync } from "node:child_process";
+
+assert.equal(process.version, "v24.22.2", "Use Node24.22.2 for the recorded comparison");
+const root = fileURLToPath(new URL("../..", import.meta.url));
+const output = path.resolve(process.argv[2] ?? mkdtempSync(path.join(tmpdir(), "presentation-benchmark-")));
+mkdirSync(output, { recursive: true });
+assert.equal(existsSync(path.join(output, "pairs.jsonl")), false, "Use a fresh output directory");
+const baseline = "82c033ed127d07862e6765aa4b69a3cf339bde81";
+const integration = "server/src/__tests__/chat-channels.integration.test.ts";
+const git = (...args) => execFileSync("git", args, { cwd: root, encoding: "utf8" });
+assert.equal(git("status", "--porcelain").trim(), "", "Commit fixture changes first; use an isolated checkout");
+const original = readFileSync(path.join(root, integration), "utf8");
+const anchor = '  it("settles a shared-endpoint accepted-response backlog';
+assert.equal(original.split(anchor).length, 2, "Expected one fixture insertion point");
+const files = ["query", "work"].map(kind => `server/src/services/native-runtime/native-response-presentation-${kind}.baseline.ts`);
+for (const file of files) assert.equal(existsSync(path.join(root, file)), false, `Refusing to replace ${file}`);
+const env = { ...process.env, PATH: `${path.dirname(process.execPath)}${path.delimiter}${process.env.PATH ?? ""}`,
+  PAPERCLIP_PRESENTATION_BENCHMARK_OUTPUT: output };
+assert.equal(execFileSync("pnpm", ["--version"], { cwd: root, env, encoding: "utf8" }).trim(), "9.15.4", "Use pnpm9.15.4");
+writeFileSync(path.join(output, "metadata.json"), JSON.stringify({ baseline, treatment: git("rev-parse", "HEAD").trim(),
+  node: process.version, execPath: process.execPath, pairs: 5, runs: 10,
+  mode: "real PostgreSQL source service under Vitest; accelerated retry timers; instrumented process CPU" }, null, 2));
+const benchmark = String.raw`  it("measures equivalent completed accepted-response work without artificial sleeps", async () => {
+    const { appendFileSync } = await import("node:fs");
+    const output = process.env.PAPERCLIP_PRESENTATION_BENCHMARK_OUTPUT;
+    if (!output) throw new Error("Benchmark output directory is required");
+    const { nativeResponsePresentationWork: baselineWork } = await import("../services/native-runtime/native-response-presentation-work.baseline.js");
+    const { createDatabaseResponseCapture, installDatabaseResponseCapture } = await import("@paperclipai/db/response-capture");
+    const schema = await import("@paperclipai/db");
+    appendFileSync(output + "/child-runtime.jsonl",
+      JSON.stringify({ version: process.version, execPath: process.execPath }) + "\n");
+    expect(process.version).toBe("v24.22.2");
+    const fixedTime = Date.now();
+    const capture = createDatabaseResponseCapture({ start: fixedTime - 1, end: fixedTime + 59_999 }, () => {}, () => fixedTime);
+    let measuring = false, queries = 0, rows = 0, queryErrors = 0;
+    capture.execute = (action => {
+      const counted = measuring;
+      if (counted) queries++;
+      return Promise.resolve().then(action).then(value => {
+        if (counted && Array.isArray(value)) rows += value.length;
+        return value;
+      }, error => { if (counted) queryErrors++; throw error; });
+    }) as typeof capture.execute;
+    installDatabaseResponseCapture(db, schema, capture);
+    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
+    try {
+      for (let round = 0; round < 5; round++) {
+        for (const label of round % 2 ? ["after", "before"] : ["before", "after"]) {
+          const fixture = await seedCompany();
+          const configured = await configuredTelegramEndpoint(fixture);
+          const contexts = [];
+          for (let index = 0; index < 4; index++) contexts.push(await committedChatResponseRecoveryFixture(
+            "telegram", "coordinator", false, undefined, undefined, true,
+            { fixture, configured, suffix: String(311 + index) }));
+          const failures: unknown[] = [];
+          const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => failures.push(error) });
+          queries = rows = queryErrors = 0;
+          const wallStart = performance.now(), cpuStart = process.cpuUsage(), virtualStart = Date.now();
+          measuring = true;
+          const worker = (label === "before" ? baselineWork : nativeResponsePresentationWork)(db).register(coordinator, () => true);
+          let passes = 0, observerIterations = 0;
+          try {
+            await worker.ready;
+            for (;;) {
+              while (idleWorkSnapshot().active !== 0) {
+                observerIterations++;
+                await new Promise(resolve => setTimeout(resolve, 1));
+              }
+              observerIterations++;
+              const next = coordinator.nextWakeAt();
+              if (next === null) break;
+              if (++passes > 10) throw new Error("Response benchmark did not settle");
+              await vi.advanceTimersByTimeAsync(Math.max(0, next - Date.now()) + 1);
+            }
+            measuring = false;
+            const cpu = process.cpuUsage(cpuStart), wallMs = performance.now() - wallStart;
+            for (const context of contexts) {
+              const comments = await db.select().from(issueComments).where(eq(issueComments.createdByRunId, context.runId));
+              expect(comments).toHaveLength(1);
+              expect(comments[0].body).toBe(context.result.summary);
+              const publications = await readChatPublicationsByCommentId(comments[0].id);
+              expect(publications).toHaveLength(1);
+              expect(publications[0]).toMatchObject({ companyId: fixture.companyId, endpointId: configured.endpoint.id,
+                issueId: context.issue.id, conversationId: context.conversation.id, payload: { text: context.result.summary } });
+            }
+            expect(await readChatPublicationsByEndpointId(configured.endpoint.id)).toHaveLength(4);
+            appendFileSync(output + "/pairs.jsonl", JSON.stringify({ round, label, node: process.version, execPath: process.execPath, observerIterations,
+              wallMs, cpuMs: (cpu.user + cpu.system) / 1000, queries, rows, queryErrors,
+              failedPasses: failures.length, replayedRetryDeadlines: passes,
+              virtualElapsedMs: Date.now() - virtualStart, completedResponses: 4,
+              artificialDatabaseSleep: false, endpointCount: 1, issueCount: 4 }) + "\n");
+          } finally { measuring = false; worker.stop(); await coordinator.stop(); await configured.service.shutdown(); }
+        }
+      }
+    } finally { measuring = false; capture.flush(); vi.useRealTimers(); }
+  }, 120_000);
+
+`;
+let status = 1;
+try {
+  for (const [index, kind] of ["query", "work"].entries()) {
+    const source = git("show", `${baseline}:server/src/services/native-runtime/native-response-presentation-${kind}.ts`);
+    const materialized = kind === "work" ? source.replace("./native-response-presentation-query.js", "./native-response-presentation-query.baseline.js") : source;
+    writeFileSync(path.join(root, files[index]), materialized);
+    writeFileSync(path.join(output, path.basename(files[index])), materialized);
+  }
+  writeFileSync(path.join(output, "benchmark-fragment.txt"), benchmark);
+  writeFileSync(path.join(root, integration), original.replace(anchor, benchmark + anchor));
+  const log = openSync(path.join(output, "vitest.log"), "wx");
+  try {
+    const result = spawnSync("pnpm", ["exec", "vitest", "run", "--project", "@paperclipai/server", integration,
+      "-t", "measures equivalent completed accepted-response work without artificial sleeps"],
+      { cwd: root, env, stdio: ["ignore", log, log] });
+    if (result.error) throw result.error;
+    status = result.status ?? 1;
+  } finally { closeSync(log); }
+  if (status === 0) {
+    const pairs = readFileSync(path.join(output, "pairs.jsonl"), "utf8").trim().split("\n").map(line => JSON.parse(line));
+    assert.equal(pairs.length, 10);
+    assert.ok(pairs.every(pair => pair.node === "v24.22.2" && pair.completedResponses === 4));
+  }
+} finally {
+  writeFileSync(path.join(root, integration), original);
+  for (const file of files) rmSync(path.join(root, file), { force: true });
+}
+console.log(`Benchmark ${status === 0 ? "passed" : "failed"}; artifacts: ${output}`);
+process.exitCode = status;
@@ -64125,24 +64125,32 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
    }
  });

+  type SharedEndpointProgress = {
+    fixture: Awaited<ReturnType<typeof seedCompany>>;
+    configured: Awaited<ReturnType<typeof configuredTelegramEndpoint>> | Awaited<ReturnType<typeof configuredSlackEndpoint>>;
+    suffix: string;
+  };
+
  async function failedChatRetryFixture(
    provider: "slack" | "telegram" | "github",
    externalActorId?: string,
    linkedOriginalActor = false,
    githubUnavailableFile = false,
    currentGitHubPr = false,
+    shared?: SharedEndpointProgress,
  ) {
    // Keep the policy under which the source was admitted. Inserting an
    // upgraded GitHub policy after receipt correctly invalidates retry proof.
    const context = await safeNativeProgressFixture(
      provider,
-      "91",
+      shared?.suffix ?? "91",
      "channel",
      externalActorId,
      linkedOriginalActor,
      githubUnavailableFile,
      false,
      currentGitHubPr,
+      shared,
    );
    const [issue] = await readIssuesById(context.conversation.issueId);
    const [action] = await db
@@ -68625,9 +68633,10 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
    githubUnavailableFile = false,
    upgradedGitHub = true,
    currentGitHubPr = false,
+    shared?: SharedEndpointProgress,
  ) {
-    const fixture = await seedCompany();
-    const configured =
+    const fixture = shared?.fixture ?? await seedCompany();
+    const configured = shared?.configured ?? (
      provider === "slack"
        ? await configuredSlackEndpoint(fixture)
        : provider === "github"
@@ -68641,7 +68650,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
            ? await configuredDiscordEndpoint(fixture)
            : provider === "microsoft-teams"
              ? await configuredTeamsEndpoint(fixture)
-              : await configuredTelegramEndpoint(fixture);
+              : await configuredTelegramEndpoint(fixture));
    const { callbacks, endpoint, runtime, service } = configured;
    const telegramChatId = `77113${suffix.padStart(3, "0")}`;
    const teamsConversationId = `19:safe-progress-${suffix}@thread.${teamsSurface === "channel" ? "tacv2" : "v2"}`;
@@ -68769,7 +68778,8 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
      message: sourceMessage,
      trigger: thread.thread.isDM ? "direct_message" : "mention",
    });
-    const [conversation] = await readChatConversationsByEndpointId(endpoint.id);
+    const conversations = await readChatConversationsByEndpointId(endpoint.id);
+    const conversation = shared ? conversations.find(row => row.externalThreadId === thread.thread.id) : conversations[0];
    if (!conversation) throw new Error("Expected safe-progress conversation");
    const providerRuntime = runtime.endpoints.get(endpoint.id);
    if (!providerRuntime) throw new Error("Expected safe-progress runtime");
@@ -74594,12 +74604,15 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
    ) => Promise<void>,
    summary?: (issueId: string) => string,
    finalize = true,
+    shared?: SharedEndpointProgress,
  ) {
    const context = await failedChatRetryFixture(
      provider,
      provider === "github" ? "42" : undefined,
      false,
      githubUnavailableFile,
+      false,
+      shared,
    );
    if (provider === "github") {
      const config = await githubChatManagementService(db).configuration(context.endpoint.id, "owner-user");
@@ -74787,6 +74800,126 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
    };
  }

+  it("settles a shared-endpoint accepted-response backlog without retrying its own locks", async () => {
+    const fixture = await seedCompany();
+    const configured = await configuredTelegramEndpoint(fixture);
+    const contexts = [];
+    for (let index = 0; index < 4; index++) {
+      contexts.push(await committedChatResponseRecoveryFixture("telegram", "coordinator", false,
+        undefined, undefined, true, { fixture, configured, suffix: String(211 + index) }));
+    }
+    expect(new Set(contexts.map(context => context.issue.id)).size).toBe(4);
+    const independent = await committedChatResponseRecoveryFixture("telegram");
+    const errors: unknown[] = [];
+    const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
+    // Hold one real comment write after authority locks are acquired. Another
+    // endpoint must still publish while this endpoint's backlog waits.
+    const functionName = `hold_presentation_${randomUUID().replaceAll("-", "")}`;
+    const gate = Math.floor(Math.random() * 1_000_000_000);
+    let release!: () => void, locked!: () => void;
+    let holderPid = 0;
+    const hold = new Promise<void>(resolve => { release = resolve; });
+    const acquired = new Promise<void>(resolve => { locked = resolve; });
+    const holder = db.transaction(async tx => {
+      const [backend] = await tx.execute(sql`select pg_backend_pid() as pid`);
+      holderPid = Number(backend.pid);
+      await tx.execute(sql`select pg_advisory_xact_lock(${gate})`); locked(); await hold;
+    });
+    await acquired;
+    const sharedRuns = contexts.map(context => `'${context.runId}'::uuid`).join(",");
+    await db.execute(sql.raw(`create function ${functionName}() returns trigger language plpgsql as $$ begin if NEW.created_by_run_id in (${sharedRuns}) then perform pg_advisory_xact_lock(${gate}); end if; return NEW; end $$`));
+    await db.execute(sql.raw(`create trigger ${functionName} before insert on issue_comments for each row execute function ${functionName}()`));
+    const worker = nativeResponsePresentationWork(db).register(coordinator, () => true);
+    try {
+      await expect.poll(async () => {
+        const [state] = await db.execute(sql`select exists (select 1 from pg_stat_activity
+          where datname = current_database() and ${holderPid} = any(pg_blocking_pids(pid))) as blocked`);
+        return state.blocked;
+      }, { timeout: 10_000 }).toBe(true);
+      await expect.poll(async () => (await db.select({ id: issueComments.id }).from(issueComments)
+        .where(eq(issueComments.createdByRunId, independent.runId))).length, { timeout: 10_000 }).toBe(1);
+      expect(await db.select({ id: issueComments.id }).from(issueComments)
+        .where(inArray(issueComments.createdByRunId, contexts.map(context => context.runId)))).toHaveLength(0);
+      release(); await holder;
+      await worker.ready;
+      await expect.poll(() => idleWorkSnapshot().active, { timeout: 10_000 }).toBe(0);
+      expect(errors).toEqual([]);
+      for (const context of contexts) {
+        const comments = await db.select().from(issueComments).where(eq(issueComments.createdByRunId, context.runId));
+        expect(comments).toHaveLength(1);
+        expect(comments[0].body).toBe(context.result.summary);
+        const publications = await readChatPublicationsByCommentId(comments[0].id);
+        expect(publications).toHaveLength(1);
+        expect(publications[0]).toMatchObject({ companyId: fixture.companyId, endpointId: configured.endpoint.id,
+          issueId: context.issue.id, conversationId: context.conversation.id, payload: { text: context.result.summary } });
+      }
+      expect(await readChatPublicationsByEndpointId(configured.endpoint.id)).toHaveLength(4);
+      expect(coordinator.nextWakeAt()).toBeNull();
+    } finally {
+      release(); await holder;
+      worker.stop(); await coordinator.stop();
+      await db.execute(sql.raw(`drop trigger ${functionName} on issue_comments`));
+      await db.execute(sql.raw(`drop function ${functionName}()`));
+      await configured.service.shutdown(); await independent.service.shutdown();
+    }
+  });
+
+  it("settles responses on different endpoints for one principal without retrying its own identity lock", async () => {
+    const fixture = await seedCompany();
+    const first = await configuredSlackEndpoint(fixture);
+    const second = await configuredSlackEndpoint(fixture);
+    const contexts = [
+      await committedChatResponseRecoveryFixture("slack", "coordinator", false, undefined, undefined, true,
+        { fixture, configured: first, suffix: "221" }),
+      await committedChatResponseRecoveryFixture("slack", "coordinator", false, undefined, undefined, true,
+        { fixture, configured: second, suffix: "222" }),
+    ];
+    expect(contexts[0].action.principalId).toBeTruthy();
+    expect(contexts[0].action.principalId).toBe(contexts[1].action.principalId);
+    expect(contexts[0].endpoint.id).not.toBe(contexts[1].endpoint.id);
+    const errors: unknown[] = [];
+    const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
+    const functionName = `hold_identity_${randomUUID().replaceAll("-", "")}`;
+    await db.execute(sql.raw(`create function ${functionName}() returns trigger language plpgsql as $$ begin perform pg_sleep(0.1); return NEW; end $$`));
+    await db.execute(sql.raw(`create trigger ${functionName} before insert on issue_comments for each row execute function ${functionName}()`));
+    const worker = nativeResponsePresentationWork(db).register(coordinator, () => true);
+    try {
+      await worker.ready;
+      await expect.poll(() => idleWorkSnapshot().active, { timeout: 10_000 }).toBe(0);
+      expect(errors).toEqual([]);
+      for (const context of contexts) {
+        const comments = await db.select().from(issueComments).where(eq(issueComments.createdByRunId, context.runId));
+        expect(comments).toHaveLength(1);
+        expect(comments[0].body).toBe(context.result.summary);
+        const publications = await readChatPublicationsByCommentId(comments[0].id);
+        expect(publications).toHaveLength(1);
+        expect(publications[0]).toMatchObject({ companyId: fixture.companyId, endpointId: context.endpoint.id,
+          issueId: context.issue.id, conversationId: context.conversation.id, payload: { text: context.result.summary } });
+      }
+      expect(coordinator.nextWakeAt()).toBeNull();
+    } finally {
+      worker.stop(); await coordinator.stop();
+      await db.execute(sql.raw(`drop trigger ${functionName} on issue_comments`));
+      await db.execute(sql.raw(`drop function ${functionName}()`));
+      await first.service.shutdown(); await second.service.shutdown();
+    }
+  });
+
+  it("does not treat an accepted-response endpoint scheduling hint as current authority", async () => {
+    const context = await committedChatResponseRecoveryFixture("telegram");
+    const work = nativeResponsePresentationWork(db);
+    try {
+      const [entry] = await work.load({ companyId: context.fixture.companyId });
+      expect(entry.endpointId).toBe(context.endpoint.id);
+      await db.update(chatEndpoints).set({ status: "paused" }).where(eq(chatEndpoints.id, context.endpoint.id));
+      // A stale or forged hint cannot authorize presentation against the real source.
+      await expect(work.check({ ...entry, endpointId: randomUUID(), principalId: randomUUID() })).resolves.toBeNull();
+      expect(await db.select({ id: issueComments.id }).from(issueComments)
+        .where(eq(issueComments.createdByRunId, context.runId))).toHaveLength(0);
+      expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0);
+    } finally { await context.service.shutdown(); }
+  });
+
  it("repairs a committed reply and its recovery notice without reversing row locks", async () => {
    const context = await committedChatResponseRecoveryFixture("telegram");
    const scope = { companyId: context.fixture.companyId, issueId: context.issue.id, runId: context.runId };
@@ -76163,7 +76296,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
    } finally { await context.service.shutdown(); }
  });

-  it("leaves a contended accepted response retryable without reclassifying or partially publishing it", async () => {
+  it.each(["action", "endpoint"] as const)("leaves a contended accepted response retryable without reclassifying or partially publishing it (%s)", async target => {
    const context = await committedChatResponseRecoveryFixture("telegram");
    let release!: () => void;
    let observed!: () => void;
@@ -76174,11 +76307,11 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
      observed = resolve;
    });
    const holder = db.transaction(async (tx) => {
-      await tx
-        .select()
-        .from(chatActions)
-        .where(eq(chatActions.id, context.action.id))
-        .for("update");
+      if (target === "endpoint") {
+        await tx.select().from(chatEndpoints).where(eq(chatEndpoints.id, context.endpoint.id)).for("no key update");
+      } else {
+        await tx.select().from(chatActions).where(eq(chatActions.id, context.action.id)).for("update");
+      }
      observed();
      await hold;
    });
@@ -40,7 +40,7 @@ it("preserves a subsecond replacement deadline when the owner does not require p

 type Entry = { id: string; companyId: string; issueId: string; dueAt: number };

-function concurrentFixture(entries: Entry[], limit = 4, concurrent = true) {
+function concurrentFixture(entries: Entry[], limit = 4, concurrent = true, key?: (entry: Entry) => string | readonly string[]) {
  vi.useFakeTimers();
  const onError = vi.fn();
  const owner = installDatabaseWorkSignals({ execute: vi.fn(async () => []), transaction: vi.fn(async () => undefined) });
@@ -66,7 +66,7 @@ function concurrentFixture(entries: Entry[], limit = 4, concurrent = true) {
  coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => canRun, onError });
  const options = {
    queue: DELIVERY_QUEUES.nativeResponsePresentation,
-    concurrentChecks: concurrent ? { limit, key: (entry: Entry) => `${entry.companyId}:${entry.issueId}` } : undefined,
+    concurrentChecks: concurrent ? { limit, key: key ?? ((entry: Entry) => `${entry.companyId}:${entry.issueId}`) } : undefined,
    decodeScope: () => undefined,
    matches: () => true,
    load,
@@ -85,6 +85,88 @@ function concurrentFixture(entries: Entry[], limit = 4, concurrent = true) {

 const entry = (id: string, issueId = id, companyId = "company"): Entry => ({ id, issueId, companyId, dueAt: 0 });

+it("merges transitive resource conflicts in candidate order without blocking independent work", async () => {
+  const keys: Record<string, string[]> = {
+    "1": ["issue:a", "endpoint:first", "endpoint:first"],
+    "2": ["issue:b", "endpoint:second"],
+    "3": ["issue:a", "endpoint:second"],
+    "4": ["issue:c", "endpoint:independent"],
+    "5": ["issue:without-endpoint"],
+    "6": ["issue:without-endpoint"],
+  };
+  const f = concurrentFixture([1, 2, 3, 4, 5, 6].map(id => entry(String(id))), 4, true, entry => keys[entry.id]);
+  try {
+    await vi.waitFor(() => expect(f.started).toEqual(["1", "4", "5"]), { timeout: 100 });
+    f.release("4"); f.release("5");
+    await vi.waitFor(() => expect(f.started).toContain("6"), { timeout: 100 });
+    expect(f.started).not.toContain("2");
+    f.release("1");
+    await vi.waitFor(() => expect(f.started).toContain("2"), { timeout: 100 });
+    expect(f.started).not.toContain("3");
+    f.release("2");
+    await vi.waitFor(() => expect(f.started).toContain("3"), { timeout: 100 });
+    expect(f.started.filter(id => ["1", "2", "3"].includes(id))).toEqual(["1", "2", "3"]);
+  } finally { f.releaseAll(); await f.worker.ready; }
+  expect(f.onError).not.toHaveBeenCalled();
+  expect(coordinator!.nextWakeAt()).toBeNull();
+});
+
+it("retains unstarted resource siblings through drain and resumes without reconstruction", async () => {
+  const f = concurrentFixture([entry("1"), entry("2"), entry("3")], 4, true,
+    entry => [entry.issueId, ...(entry.id === "3" ? ["independent"] : ["shared-endpoint"])]);
+  try {
+    await vi.waitFor(() => expect(f.started).toEqual(["1", "3"]), { timeout: 100 });
+    f.pause(); f.releaseAll(); await f.worker.ready;
+    expect(f.started).not.toContain("2");
+    await vi.advanceTimersByTimeAsync(60_000);
+    expect(f.started).not.toContain("2");
+    f.resume(); await vi.advanceTimersByTimeAsync(1);
+    expect(f.started).toEqual(["1", "3", "2"]);
+    expect(f.load).toHaveBeenCalledOnce();
+    expect(f.onError).not.toHaveBeenCalled();
+  } finally { f.releaseAll(); await f.worker.ready; }
+});
+
+it("retains an in-flight rescope and admits a bridging newcomer once after active groups settle", async () => {
+  const keys: Record<string, string[]> = {
+    "1": ["issue:a", "endpoint:first"], "2": ["issue:b", "endpoint:second"],
+    "3": ["issue:a", "endpoint:second"],
+  };
+  const f = concurrentFixture([entry("1"), entry("2")], 4, true, entry => keys[entry.id]);
+  try {
+    await vi.waitFor(() => expect(f.started).toEqual(["1", "2"]), { timeout: 100 });
+    f.entries.push(entry("3")); f.worker.invalidate("new-scope");
+    f.release("1");
+    await vi.waitFor(() => expect(f.active.has("1")).toBe(false), { timeout: 100 });
+    expect(f.load).toHaveBeenCalledOnce();
+    expect(f.started).not.toContain("3");
+    f.release("2"); await f.worker.ready;
+    await vi.advanceTimersByTimeAsync(1000);
+    expect(f.load).toHaveBeenCalledTimes(2);
+    expect(f.started).toEqual(["1", "2", "3"]);
+  } finally { f.releaseAll(); await vi.advanceTimersByTimeAsync(1); await f.worker.ready; }
+  expect(f.started.filter(id => id === "3")).toHaveLength(1);
+  expect(f.onError).not.toHaveBeenCalled();
+});
+
+it("serializes a shared principal across endpoints while unrelated resources progress", async () => {
+  const keys: Record<string, string[]> = {
+    "1": ["issue:a", "endpoint:a", "principal:shared"],
+    "2": ["issue:b", "endpoint:b", "principal:shared"],
+    "3": ["issue:c", "endpoint:c", "principal:independent"],
+  };
+  const f = concurrentFixture([entry("1"), entry("2"), entry("3")], 4, true, entry => keys[entry.id]);
+  try {
+    await vi.waitFor(() => expect(f.started).toEqual(["1", "3"]), { 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).toEqual(["1", "3", "2"]), { timeout: 100 });
+  } finally { f.releaseAll(); await f.worker.ready; }
+  expect(f.onError).not.toHaveBeenCalled();
+});
+
 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 {
@@ -157,22 +239,23 @@ it("retains unstarted checks through drain and resumes without idle discovery re
  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))));
+it.each(["task", "resources"] as const)("stops starting checks and awaits every admitted %s check during shutdown", async mode => {
+  const f = concurrentFixture(Array.from({ length: 6 }, (_, index) => entry(String(index))), 4, true,
+    mode === "resources" ? entry => [entry.issueId, `endpoint:${Number(entry.id) % 2}`] : undefined);
  let stopped = false;
  let stop: Promise<void> | undefined;
  try {
-    await vi.waitFor(() => expect(f.started).toHaveLength(4), { timeout: 100 });
+    await vi.waitFor(() => expect(f.started).toHaveLength(mode === "resources" ? 2 : 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);
+    expect(f.started).toHaveLength(mode === "resources" ? 2 : 4);
  } finally { f.releaseAll(); await f.worker.ready; await stop; }
  expect(stopped).toBe(true);
-  expect(f.started).toHaveLength(4);
+  expect(f.started).toHaveLength(mode === "resources" ? 2 : 4);
 });

 it("keeps failures isolated and retries their deadline without repeating successful checks", async () => {
@@ -1,15 +1,18 @@
 import { QueryBuilder } from "drizzle-orm/pg-core";
 import { and, asc, eq, exists, gt, inArray, isNotNull, isNull, notExists, or, sql } from "drizzle-orm";
-import { heartbeatRunEvents, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, statusDecisions, statusDecisionEffects } from "@paperclipai/db";
+import { chatActions, heartbeatRunEvents, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, statusDecisions, statusDecisionEffects } from "@paperclipai/db";
 import type { NativeResponsePresentationScope } from "../native-response-presentation-notifications.js";

 export function nativeResponsePresentationCandidates(scope?: NativeResponsePresentationScope, after?: string, runId?: string) {
  const db = new QueryBuilder();
  return db.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId,
    issueId: nativeRunFinalizations.issueId, agentId: heartbeatRuns.agentId, nextAttemptAt: nativeRunFinalizations.nextAttemptAt,
+    endpointId: chatActions.endpointId, principalId: chatActions.principalId,
  }).from(nativeRunFinalizations)
    .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.id, nativeRunFinalizations.runId),
      eq(heartbeatRuns.companyId, nativeRunFinalizations.companyId), eq(heartbeatRuns.nativeIssueId, nativeRunFinalizations.issueId)))
+    .leftJoin(chatActions, and(eq(chatActions.id, heartbeatRuns.wakeupRequestId),
+      eq(chatActions.companyId, heartbeatRuns.companyId)))
    .innerJoin(issues, and(eq(issues.id, nativeRunFinalizations.issueId), eq(issues.companyId, nativeRunFinalizations.companyId)))
    .innerJoin(nativeRunResults, and(eq(nativeRunResults.id, nativeRunFinalizations.resultId),
      eq(nativeRunResults.companyId, nativeRunFinalizations.companyId), eq(nativeRunResults.issueId, nativeRunFinalizations.issueId),
@@ -14,7 +14,7 @@ export function registerNativeResponsePresentationWork(db: Db,
  return nativeResponsePresentationWork(db).register(coordinator, canRun);
 }

-type Entry = { id: string; companyId: string; issueId: string; agentId: string; dueAt: number };
+type Entry = { id: string; companyId: string; issueId: string; agentId: string; endpointId?: string | null; principalId?: string | null; dueAt: number };

 export function nativeResponsePresentationWork(db: Db) {
  const candidates = (scope?: NativeResponsePresentationScope, after?: string, runId?: string) => {
@@ -31,7 +31,7 @@ export function nativeResponsePresentationWork(db: Db) {
    let after: string | undefined;
    while (canRun() && !signal?.aborted) {
      const rows = await measureRunPhase("response.resolve", () => candidates(scope, after));
-      for (const row of rows) entries.push({ id: row.id, companyId: row.companyId, issueId: row.issueId, agentId: row.agentId, dueAt: row.nextAttemptAt?.getTime() ?? 0 });
+      for (const row of rows) entries.push({ id: row.id, companyId: row.companyId, issueId: row.issueId, agentId: row.agentId, endpointId: row.endpointId, principalId: row.principalId, dueAt: row.nextAttemptAt?.getTime() ?? 0 });
      if (rows.length < 100) return entries;
      after = rows[rows.length - 1].id;
    }
@@ -56,7 +56,13 @@ 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}` },
+      // 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) => [
+        `issue:${entry.companyId}:${entry.issueId}`,
+        ...(entry.endpointId ? [`endpoint:${entry.companyId}:${entry.endpointId}`] : []),
+        ...(entry.principalId ? [`principal:${entry.companyId}:${entry.principalId}`] : []),
+      ] },
      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),
@@ -12,7 +12,7 @@ export function registerScopedDeadlineWork<Entry extends ScheduledEntry, Scope>(
    queue: DeliveryQueue;
    batchSize?: number;
    minimumPassIntervalMs?: number;
-    concurrentChecks?: { limit: number; key(entry: Entry): string };
+    concurrentChecks?: { limit: number; key(entry: Entry): string | readonly string[] };
    reconstructionIntervalMs?: number | ((scope: Scope | undefined) => number);
    beforePass?(): Promise<void>;
    decodeScope(hint: string): Scope | undefined;
@@ -116,9 +116,27 @@ 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) {
-        const key = options.concurrentChecks?.key(entry) ?? "";
+      // An owner may share more than one resource (for example a task and an
+      // endpoint). Merge intersecting keys before dispatch so waiting siblings
+      // do not occupy the bounded check slots ahead of independent work.
+      const parents = batch.entries.map((_, index) => index);
+      const root = (index: number): number => {
+        while (parents[index] !== index) index = parents[index];
+        return index;
+      };
+      const owners = new Map<string, number>();
+      for (const [index, entry] of batch.entries.entries()) {
+        const configured = options.concurrentChecks?.key(entry) ?? "";
+        const keys = typeof configured === "string" ? [configured] : configured;
+        for (const key of keys) {
+          const previous = owners.get(key);
+          if (previous === undefined) owners.set(key, index);
+          else parents[root(index)] = root(previous);
+        }
+      }
+      const groups = new Map<number, Entry[]>();
+      for (const [index, entry] of batch.entries.entries()) {
+        const key = root(index);
        const group = groups.get(key) ?? [];
        group.push(entry); groups.set(key, group);
      }