perf(chat): find the conversation head from open delivery work (#20564)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - External chat messages go into a durable delivery queue. Each
conversation drains in provider order.
> - Each drain step reads the head of one conversation. The head query
filters one endpoint's deliveries by state and thread.
> - On a hosted instance, this query ran about 10,000 times at about 300
ms each. The database was busy, and request reads waited behind
background work.
> - PostgreSQL read every processed delivery of the endpoint and its
`normalized_event` value to find a few open rows.
> - This pull request finds the endpoint's open delivery IDs first and
fetches only those rows.
> - The benefit is less background database load. The selected head and
the delivery order do not change.

## Linked Issues or Issue Description

**What existing behavior does this improve?**
The inbound conversation head query
(`earliestOpenConversationDelivery`).

**Subsystem affected**
Server chat inbound delivery queue.

**Current behavior**
The query has `endpoint_id = $1 and (state in ('received', 'retry',
'processing') or (state = 'processed' and <pending inbound wake>))`, a
thread filter on `normalized_event`, and a sort on `normalized_event`
keys. No index matches the `or`. PostgreSQL scans all deliveries of the
endpoint. For each row it reads `normalized_event` to test the thread.
It also reads all of `chat_actions` to build the pending-wake set. A
busy endpoint keeps all of its processed history, so the cost grows with
that history.

**Proposed behavior**
The query selects `id in (<open delivery IDs of the endpoint>)`. That
set is a `union all` of two small reads:
- received, retry and processing deliveries of the endpoint, through the
existing `chat_deliveries_work_idx`;
- processed deliveries that an issued or processing `inbound_wakeup`
action of the same endpoint (and company) still owns.

PostgreSQL fetches only those rows by primary key, then applies the same
reaction, thread and order rules.

**Reason and benefit**
The cost no longer grows with processed history. On a local fixture with
60,000 processed deliveries, the head query reads 1,309 shared buffers
instead of 262,630.

**Breaking changes**
None. There is no schema change, migration, API change, timer or poll
change.

Related work: #17971 changed the pending-wake predicate of the same
queue. #20400 added the index for the other chat delivery statement in
the same report (`select state ...
normalized_event->'githubAutomatic'->'context'->>'deliveryId'`). That
statement already uses the #20400 index on master, so this pull request
does not change it.

## What Changed

- `delivery-ledger.ts`: add `openConversationDeliveryIds(endpointId)`.
It returns the endpoint's open delivery IDs. It selects the same set as
the old state predicate for that endpoint.
- `conversation-queue.ts`: the conversation head query uses `id in
(openConversationDeliveryIds(endpointId))` in place of the state `or`.
The thread filter, reaction exclusion, order, limit and lifecycle
dependency read do not change.
- `inbound-queue-discovery.test.ts`: add a real PostgreSQL test. It
compares the new set with the old predicate on three endpoints. It
covers each state, issued and processing wakes, duplicate wakes, wakes
of non-processed deliveries, processed, other-kind and null-delivery
actions, an action on a different endpoint of the same company, and
reactions. It also checks that the real captured head query returns the
oracle's first row.
- `benchmark-inbound-queue-discovery.ts`: add
`--endpoint-history=<rows>` (a busy endpoint with 4 KiB payloads) and
`--only=<fixture>`. Add a work budget: the treatment head query must
visit at most 100 delivery rows on the history fixtures.
- `doc/DATABASE.md`: describe the head query and the benchmark command.

## Verification

Local SQL measurements. PostgreSQL 17.10 (Postgres.app, macOS arm64),
disposable database, synthetic data, default settings with 256 MB shared
buffers. The host was under heavy load from other work, so timings are
noisy. Shared buffers and visited rows are stable. Each arm has two
warmups and five alternating samples (median [min..max]). Baseline:
`origin/master` at `8d2bf78323`.

| Fixture | Query | Baseline | Treatment | Shared buffers | Delivery
rows visited | Output |
| --- | --- | ---: | ---: | ---: | ---: | --- |
| 60,000 processed, 4 KiB | head | 290.09 ms [220.21..554.19] | 7.34 ms
[5.85..7.68] | 262,630 → 1,309 | 60,002 → 4 | Equal |
| 20,000 processed, 4 KiB | head | 276.02 ms [211.77..322.12] | 34.61 ms
[4.45..194.15] | 67,618 → 486 | 20,002 → 4 | Equal |
| 1,500 processed, 32 KiB | head | 15.88 ms [13.66..21.79] | 2.29 ms
[1.64..7.75] | 10,602 → 80 | 1,502 → 4 | Equal |
| 20 processed | head | 1.49 ms [0.92..3.66] | 1.85 ms [1.01..1.97] | 2
→ 45 | 22 → 968 | Equal |

- The deadline and ready selectors do not change. Their buffers are
identical in both arms; their timings move only with host noise.
- The small control is neutral. On a near-empty table, PostgreSQL picks
a sequential scan for the open-state branch. This costs about 45
buffers.
- The 60,000-row run passes `--require-work-budget`. The baseline
exceeds the budget (`baselineExceedsWorkBudget: true`). This is the
red/green proof. There is no timing threshold in CI.
- On the treatment, the remaining head cost is the scan of
`chat_actions` for pending wakes of the endpoint. The baseline also
reads all of `chat_actions` for its hashed subplan.
- `pnpm exec vitest run src/__tests__/inbound-queue-discovery.test.ts`
(server): 6 tests pass, including the new equivalence test.
- `chat-channels.integration.test.ts` filtered with `-t
"reorder|orders|FIFO|conversation drain|lease renewal|receipt
order|edit|lifecycle"`: 148 assembled chat tests pass. They cover Slack,
Teams, Telegram, Discord and GitHub provider-order drains, head-of-line
barriers, lease loss, lifecycle edits and wake retries.
- `pnpm exec tsc --noEmit -p tsconfig.json` (server) passes.

Reproduce:

```sh
pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-inbound-queue-discovery.ts --baseline-ref=8d2bf78323 --endpoint-history=60000 --only=endpoint-history --require-work-budget
```

Set `PAPERCLIP_TEST_POSTGRES_ADMIN_URL` to use an existing disposable
PostgreSQL server. The benchmark uses synthetic data only.

## Risks

- Query plans depend on data and settings. These local numbers do not
prove hosted request latency. A follow-up statistics window after deploy
must confirm the gain.
- The pending-wake branch still scans the endpoint's `chat_actions`. A
partial index on pending `inbound_wakeup` actions would remove that
scan. I did not add it, to avoid a migration.
- Low behavior risk. The new set equals the old predicate for each
endpoint, and the order expression is unchanged. Tests compare both on
the same rows.

## Model Used

- Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, with tool use (shell,
file edit) in Claude Code. Extended reasoning was used.

## 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

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

doc/DATABASE.md

@@ -463,6 +463,18 @@ benchmark loads that revision's chat query, uses synthetic data only, compares
 complete returned rows, and prints timings and `EXPLAIN (ANALYZE, BUFFERS)` plans.
 Local results measure query execution, not staging request latency.
 
+The inbound conversation head reads one endpoint's open delivery IDs first. These
+are received, retry and processing deliveries (from `chat_deliveries_work_idx`)
+and processed deliveries that the endpoint's issued or processing `inbound_wakeup`
+actions still own. Only those rows are fetched, filtered by thread and sorted. The
+query does not scan the endpoint's processed history or read its
+`normalized_event` values. To compare it with a pre-change revision on a busy
+endpoint fixture:
+
+```sh
+pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-inbound-queue-discovery.ts --baseline-ref=<commit> --endpoint-history=20000 --only=endpoint-history --require-work-budget
+```
+
 Run history defaults to 200 rows and permits at most 1,000 rows, including
 direct service calls. The UI client requests 200 rows when its caller omits a
 limit. Context summaries expand the stored JSON once per returned row, retaining

scripts/benchmark-inbound-queue-discovery.ts

@@ -9,6 +9,16 @@ import { captureInboundQuery, runQuery, seedInboundDiscovery, type QueueFactory,
 
 const ref = process.argv.find(arg => arg.startsWith("--baseline-ref="))?.slice(15);
 assert.ok(ref, "Pass an explicit pre-change --baseline-ref=<commit>");
+// `--endpoint-history=<rows>` adds one endpoint with a long processed history
+// and 4 KiB payloads, like a busy hosted bot. `--only=<name>` selects one fixture.
+const endpointHistoryArg = process.argv.find(arg => arg.startsWith("--endpoint-history="))?.slice(19);
+const endpointHistory = endpointHistoryArg === undefined ? 0 : Number(endpointHistoryArg);
+assert.ok(endpointHistoryArg === undefined || (Number.isSafeInteger(endpointHistory) && endpointHistory > 0), "--endpoint-history must be a positive integer");
+const only = process.argv.find(arg => arg.startsWith("--only="))?.slice(7) ?? (process.argv.includes("--retained-only") ? "retained-history" : undefined);
+const fixtures = [{ name: "small", history: 20, payloadBytes: 0 }, { name: "retained-history", history: 1500, payloadBytes: 32 * 1024 },
+  ...(endpointHistory > 0 ? [{ name: "endpoint-history", history: endpointHistory, payloadBytes: 4096 }] : [])];
+const selectedFixtures = fixtures.filter(value => !only || value.name === only);
+assert.ok(selectedFixtures.length > 0, `No fixture matches --only=${only}; endpoint-history also needs --endpoint-history=<rows>`);
 const baselineRevision = execFileSync("git", ["rev-parse", "--verify", `${ref}^{commit}`], { encoding: "utf8" }).trim();
 const root = fileURLToPath(new URL("../", import.meta.url));
 const sources = ["server/src/services/chat-channels/inbound/delivery-ledger.ts", "server/src/services/chat-channels/inbound/conversation-queue.ts", "server/src/services/chat-channels.ts"];
@@ -43,7 +53,7 @@ try {
   const malformedReadyControls = [];
   let baselineExceedsWorkBudget = false;
   const budgetFailures: string[] = [];
-  for (const fixture of [{ name: "small", history: 20, payloadBytes: 0 }, { name: "retained-history", history: 1500, payloadBytes: 32 * 1024 }].filter(value => !process.argv.includes("--retained-only") || value.name === "retained-history")) {
+  for (const fixture of selectedFixtures) {
     const f = await seedInboundDiscovery(db, fixture.history, fixture.payloadBytes);
     await db.$client.unsafe("begin");
     if (process.argv.includes("--jit-off")) await db.$client.unsafe("set local jit = off");
@@ -89,9 +99,21 @@ try {
             if (kind !== "ready" && newPlan.JIT) budgetFailures.push(`${kind}: unnecessary JIT on the retained-history fixture`);
             if (kind === "deadline" && newBuffers > 1000) budgetFailures.push(`${kind}: ${newBuffers} shared buffer accesses exceeds fixture budget 1000`);
           }
+          // The conversation head must start from open work. It must not visit
+          // the endpoint's processed delivery history to find the few open rows.
+          const deliveryRowsVisited = {} as Record<"before" | "after", number>;
+          for (const arm of ["before", "after"] as const) {
+            deliveryRowsVisited[arm] = nodes((plans[arm] as { Plan: Record<string, unknown> }[])[0]!.Plan)
+              .filter(node => node["Relation Name"] === "chat_deliveries")
+              .reduce((sum, node) => sum + (Number(node["Actual Rows"] ?? 0) + Number(node["Rows Removed by Filter"] ?? 0)) * Number(node["Actual Loops"] ?? 1), 0);
+          }
+          if (kind === "head" && fixture.name !== "small") {
+            baselineExceedsWorkBudget ||= deliveryRowsVisited.before > 100;
+            if (deliveryRowsVisited.after > 100) budgetFailures.push(`head: visits ${deliveryRowsVisited.after} delivery rows on the ${fixture.name} fixture; budget 100`);
+          }
           const median = (values: number[]) => [...values].sort((a, b) => a - b)[Math.floor(values.length / 2)]!;
-          results.push({ fixture, kind, equalPositionalRows: expected.length, before, after, samples, beforeMedianMs: median(samples.before), afterMedianMs: median(samples.after), actionLoops, plans });
-          process.stderr.write(`${fixture.name}/${kind}: ${median(samples.before).toFixed(2)} -> ${median(samples.after).toFixed(2)} ms; action scans ${actionLoops.before} -> ${actionLoops.after}\n`);
+          results.push({ fixture, kind, equalPositionalRows: expected.length, before, after, samples, beforeMedianMs: median(samples.before), afterMedianMs: median(samples.after), actionLoops, deliveryRowsVisited, plans });
+          process.stderr.write(`${fixture.name}/${kind}: ${median(samples.before).toFixed(2)} -> ${median(samples.after).toFixed(2)} ms; action scans ${actionLoops.before} -> ${actionLoops.after}; delivery rows visited ${deliveryRowsVisited.before} -> ${deliveryRowsVisited.after}\n`);
         }
       } finally { connection.release(); }
       if (fixture.name === "small") {

server/src/__tests__/inbound-queue-discovery.test.ts

@@ -1,9 +1,9 @@
 import { randomUUID } from "node:crypto";
 import { afterAll, beforeAll, describe, expect, it } from "vitest";
-import { and, eq, getTableColumns, sql } from "drizzle-orm";
+import { and, eq, getTableColumns, inArray, notInArray, or, sql } from "drizzle-orm";
 import { chatActions, chatDeliveries, chatEndpoints, createDb, startEmbeddedPostgresTestDatabase, type Db } from "@paperclipai/db";
 import { createChatConversationDeliveryQueue } from "../services/chat-channels/inbound/conversation-queue.js";
-import { pendingInboundWakeupCondition } from "../services/chat-channels/inbound/delivery-ledger.js";
+import { conversationDeliveryOrder, openConversationDeliveryIds, pendingInboundWakeupCondition } from "../services/chat-channels/inbound/delivery-ledger.js";
 import { captureInboundQuery, pendingIds, queueOptions, runQuery, seedInboundDiscovery } from "./helpers/inbound-discovery-fixture.js";
 
 let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
@@ -29,6 +29,56 @@ describe("inbound queue discovery", () => {
     expect((await pendingIds(db, pendingInboundWakeupCondition(), f.companyId)).map(row => row.id).sort()).toEqual([f.pending.id, duplicate.id].sort());
   });
 
+  it("selects the same open conversation work as the endpoint state predicate", async () => {
+    const f = await seedInboundDiscovery(db);
+    const other = await seedInboundDiscovery(db, 0);
+    const otherEndpoint = await f.anotherEndpoint();
+    const open = await Promise.all((["received", "retry", "processing"] as const).map(state => f.delivery({ state })));
+    const closed = await Promise.all((["filtered", "failed", "processed"] as const).map(state => f.delivery({ state })));
+    const processingWake = await f.delivery({ state: "processed" });
+    await f.action(processingWake.id, { status: "processing" });
+    const duplicateWake = await f.delivery({ state: "processed" });
+    await f.action(duplicateWake.id); await f.action(duplicateWake.id);
+    const wakeForOpenDelivery = await f.delivery({ state: "retry" });
+    await f.action(wakeForOpenDelivery.id);
+    const wakeForFailedDelivery = await f.delivery({ state: "failed" });
+    await f.action(wakeForFailedDelivery.id);
+    const notPending = await f.delivery({ state: "processed" });
+    await f.action(notPending.id, { status: "processed" });
+    await f.action(notPending.id, { kind: "other" });
+    await f.action(notPending.id, { endpointId: otherEndpoint });
+    await f.action(null);
+    const crossEndpointWake = await f.delivery({ state: "processed", endpointId: otherEndpoint });
+    await f.action(crossEndpointWake.id);
+    const otherEndpointWake = await f.delivery({ state: "processed", endpointId: otherEndpoint });
+    await f.action(otherEndpointWake.id, { endpointId: otherEndpoint });
+    const reaction = await f.delivery({ state: "received", eventKind: "reaction_added" });
+    const oracle = or(inArray(chatDeliveries.state, ["received", "retry", "processing"]),
+      and(eq(chatDeliveries.state, "processed"), pendingInboundWakeupCondition()));
+    for (const endpointId of [f.endpointId, other.endpointId, otherEndpoint]) {
+      const expected = await db.select({ id: chatDeliveries.id }).from(chatDeliveries)
+        .where(and(eq(chatDeliveries.endpointId, endpointId), oracle)).orderBy(chatDeliveries.id);
+      const actual = await db.select({ id: chatDeliveries.id }).from(chatDeliveries)
+        .where(and(eq(chatDeliveries.endpointId, endpointId), inArray(chatDeliveries.id, openConversationDeliveryIds(endpointId)))).orderBy(chatDeliveries.id);
+      expect(actual).toEqual(expected);
+    }
+    const selected = (await db.select({ id: chatDeliveries.id }).from(chatDeliveries)
+      .where(and(eq(chatDeliveries.endpointId, f.endpointId), inArray(chatDeliveries.id, openConversationDeliveryIds(f.endpointId))))).map(row => row.id).sort();
+    for (const row of [f.pending, ...open, processingWake, duplicateWake, wakeForOpenDelivery, reaction]) expect(selected).toContain(row.id);
+    expect(new Set(selected).size).toBe(selected.length);
+    for (const row of [...closed, wakeForFailedDelivery, notPending, other.pending, crossEndpointWake, otherEndpointWake]) expect(selected).not.toContain(row.id);
+    expect((await db.select({ id: chatDeliveries.id }).from(chatDeliveries)
+      .where(and(eq(chatDeliveries.endpointId, otherEndpoint), inArray(chatDeliveries.id, openConversationDeliveryIds(otherEndpoint))))).map(row => row.id)).toEqual([otherEndpointWake.id]);
+    // The real head query returns the oracle's first row in conversation order.
+    const earliest = await f.delivery({ state: "processed", normalizedEvent: { conversation: { externalThreadId: f.threadId }, message: { providerSentAt: "2025-12-31T00:00:00Z" } } });
+    await f.action(earliest.id, { status: "processing" });
+    const [expectedHead] = await db.select({ id: chatDeliveries.id }).from(chatDeliveries).where(and(eq(chatDeliveries.endpointId, f.endpointId), oracle,
+      notInArray(chatDeliveries.eventKind, ["reaction_added", "reaction_removed"]), sql`${chatDeliveries.normalizedEvent}->'conversation'->>'externalThreadId' = ${f.threadId}`))
+      .orderBy(...conversationDeliveryOrder()).limit(1);
+    expect(expectedHead?.id).toBe(earliest.id);
+    expect((await runQuery(db, await captureInboundQuery(db, f, "head")))[0]?.[0]).toBe(earliest.id);
+  });
+
   it("preserves scoped ready-head probes with unrelated malformed retry data", async () => {
     const f = await seedInboundDiscovery(db);
     const other = await seedInboundDiscovery(db, 0);

server/src/services/chat-channels/inbound/conversation-queue.ts

@@ -60,7 +60,7 @@ import type {
   Message,
   Thread,
 } from "chat";
-import { DELIVERY_PROCESSING_STALE_MS, deliveryContentWasRedacted, setupDestinationActivationSuppressed, normalizedDeliveryThreadId, normalizedLifecycleTargetEventId, normalizedLifecycleEffect, deliveryReady, conversationDeliveryOrder, pendingInboundWakeupCondition, lifecycleRuntimeFence } from "./delivery-ledger.js";
+import { DELIVERY_PROCESSING_STALE_MS, deliveryContentWasRedacted, setupDestinationActivationSuppressed, normalizedDeliveryThreadId, normalizedLifecycleTargetEventId, normalizedLifecycleEffect, deliveryReady, conversationDeliveryOrder, pendingInboundWakeupCondition, openConversationDeliveryIds, lifecycleRuntimeFence } from "./delivery-ledger.js";
 import type { createChatMessageLifecycle } from "./lifecycle.js";
 import type { LiveInboundMessage } from "../runtime-types.js";
 type DeliveryRow = typeof chatDeliveries.$inferSelect;
@@ -102,13 +102,7 @@ export function createChatConversationDeliveryQueue(db: Db, options: Conversatio
       .where(
         and(
           eq(chatDeliveries.endpointId, endpointId),
-          or(
-            inArray(chatDeliveries.state, ["received", "retry", "processing"]),
-            and(
-              eq(chatDeliveries.state, "processed"),
-              pendingInboundWakeupCondition(),
-            ),
-          ),
+          inArray(chatDeliveries.id, openConversationDeliveryIds(endpointId)),
           notInArray(chatDeliveries.eventKind, [
             "reaction_added",
             "reaction_removed",

server/src/services/chat-channels/inbound/delivery-ledger.ts

@@ -210,6 +210,36 @@ const DELIVERY_PROCESSING_STALE_MS = 60_000;
     )`;
   }
 
+  /**
+   * IDs of one endpoint's open conversation work: a received, retry or
+   * processing delivery, or a processed delivery whose inbound wake is still
+   * issued or processing. This is the same set as the endpoint filter with
+   * `state in (...) or (state = 'processed' and pendingInboundWakeupCondition())`.
+   *
+   * Each branch starts from a small set (open states, or the endpoint's
+   * pending wake actions). The caller does not scan the endpoint's processed
+   * history or read its `normalized_event` values to find the few open rows.
+   */
+  function openConversationDeliveryIds(endpointId: string): SQL {
+    return sql`(
+      select open_delivery.id
+      from ${chatDeliveries} open_delivery
+      where open_delivery.endpoint_id = ${endpointId}
+        and open_delivery.state in ('received', 'retry', 'processing')
+      union all
+      select woken_delivery.id
+      from ${chatActions} pending_wake
+      join ${chatDeliveries} woken_delivery
+        on woken_delivery.id = pending_wake.delivery_id
+        and woken_delivery.company_id = pending_wake.company_id
+        and woken_delivery.endpoint_id = pending_wake.endpoint_id
+      where pending_wake.endpoint_id = ${endpointId}
+        and pending_wake.kind = 'inbound_wakeup'
+        and pending_wake.status in ('issued', 'processing')
+        and woken_delivery.state = 'processed'
+    )`;
+  }
+
   function lifecycleRuntimeFence(
     delivery: DeliveryRow,
   ): LifecycleRuntimeFence | null {
@@ -247,4 +277,4 @@ const DELIVERY_PROCESSING_STALE_MS = 60_000;
     return { ...normalized, runtimeContext };
   }
 
-export { DELIVERY_PROCESSING_STALE_MS, deliveryContentWasRedacted, setupDestinationActivationSuppressed, normalizedDeliveryThreadId, normalizedLifecycleTargetEventId, normalizedLifecycleEffect, deliveryReady, conversationDeliveryOrder, pendingInboundWakeupCondition, lifecycleRuntimeFence, normalizedLifecycleWithRuntimeFence };
+export { DELIVERY_PROCESSING_STALE_MS, deliveryContentWasRedacted, setupDestinationActivationSuppressed, normalizedDeliveryThreadId, normalizedLifecycleTargetEventId, normalizedLifecycleEffect, deliveryReady, conversationDeliveryOrder, pendingInboundWakeupCondition, openConversationDeliveryIds, lifecycleRuntimeFence, normalizedLifecycleWithRuntimeFence };