perf(server): find the native event cursor through the source seq index (#16140)

Find the contiguous native cursor in PostgreSQL while preserving gaps, late arrivals, source isolation, and event replay. Verify a long-history response budget against current master.

Co-Authored-By: Paperclip <noreply@paperclip.ing>

doc/DATABASE.md

@@ -49,6 +49,12 @@ and leaves all existing cases passive. Source identities are not grants.
 
 ### Task query performance
 
+Native event acknowledgements read a count and maximum through the existing
+source-sequence index. A stream with gaps uses indexed lookups for its first
+missing successor. The acknowledgement keeps its contiguous-prefix rule without
+returning every historical sequence number. All events stay stored. Counting the
+index entries still grows with the stream length.
+
 The attention feed checks for newer runs with company-scoped existence probes.
 The probes match the failed run's agent and issue or task context. They preserve
 the existing millisecond comparison and empty-context rules. Each batch holds

server/src/__tests__/heartbeat-run-event-sequencing.test.ts

@@ -1,5 +1,6 @@
 import { describe, expect, it } from "vitest";
-import { eq } from "drizzle-orm";
+import { eq, sql } from "drizzle-orm";
+import { createDatabaseResponseCapture, installDatabaseResponseCapture } from "@paperclipai/db/response-capture";
 import {
   agents,
   companies,
@@ -15,6 +16,62 @@ import {
 import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
 
 describe("P6-11..13 / P6-17 canonical event allocator", () => {
+  it("keeps native cursor responses bounded with long histories and a late gap fill", async () => {
+    const temporary = await startEmbeddedPostgresTestDatabase("paperclip-native-cursor-budget-");
+    const db = createDb(temporary.connectionString);
+    const companyId = "20000000-0000-4000-8000-000000000001";
+    const agentId = "20000000-0000-4000-8000-000000000002";
+    const runId = "20000000-0000-4000-8000-000000000003";
+    const otherRunId = "20000000-0000-4000-8000-000000000004";
+    try {
+      await db.insert(companies).values({ id: companyId, name: "Cursor budget", issuePrefix: "CBG" });
+      await db.insert(agents).values({ id: agentId, companyId, name: "Cursor agent" });
+      await db.insert(heartbeatRuns).values([
+        { id: runId, companyId, agentId, status: "running", nextEventSeq: 5005 },
+        { id: otherRunId, companyId, agentId, status: "running" },
+      ]);
+      for (let batch = 0; batch < 10; batch += 1) {
+        await db.insert(heartbeatRunEvents).values(Array.from({ length: 500 }, (_, index) => {
+          const seq = batch * 500 + index + 1;
+          return { companyId, agentId, runId, seq, eventType: "item.delta", sourceInstanceId: "runner-a",
+            sourceSeq: seq, sourceEventId: `seed:${seq}` };
+        }));
+      }
+      await db.insert(heartbeatRunEvents).values([
+        ...[-1, 0, null].map((sourceSeq, index) => ({
+          companyId, agentId, runId, seq: 5001 + index, eventType: "item.delta",
+          sourceInstanceId: "runner-a", sourceSeq, sourceEventId: `legacy:${index}`,
+        })),
+        { companyId, agentId, runId, seq: 5004, eventType: "item.delta", sourceInstanceId: "runner-b",
+          sourceSeq: 5001, sourceEventId: "other-source:5001" },
+        { companyId, agentId, runId: otherRunId, seq: 1, eventType: "item.delta", sourceInstanceId: "runner-a",
+          sourceSeq: 5001, sourceEventId: "other-run:5001" },
+      ]);
+      const append = (sourceSeq: number) => appendHeartbeatRunEvent(db, {
+        companyId, agentId, runId, eventType: "item.delta", payload: { sourceSeq },
+        nativeSource: { sourceInstanceId: "runner-a", sourceSeq, sourceEventId: `runner-a:${sourceSeq}`,
+          protocolSchemaVersion: 1, canonicalPayload: { sourceSeq } },
+      });
+      const events: Array<Record<string, unknown>> = [], now = Date.now();
+      const capture = createDatabaseResponseCapture({ start: now - 100, end: now + 59_000 }, event => events.push(event));
+      installDatabaseResponseCapture(db, { heartbeatRunEvents }, capture);
+      try { expect((await append(5002)).highestContiguousSourceSeq).toBe(5000); }
+      finally { capture.flush(); }
+      const reads = events.filter(event => event.event === "paperclip.db_response_capture.entry.v1"
+        && event.table === "heartbeat_run_events" && (event.operation === "select" || event.operation === "execute"));
+      expect(reads.reduce((sum, event) => sum + Number(event.calls), 0)).toBeGreaterThan(0);
+      expect(reads.reduce((sum, event) => sum + Number(event.rows), 0)).toBeLessThan(16);
+      expect(reads.reduce((sum, event) => sum + Number(event.responseValueBytes), 0)).toBeLessThan(2048);
+      expect((await append(5001)).highestContiguousSourceSeq).toBe(5002);
+      const replay = await append(5002);
+      expect(replay.disposition).toBe("duplicate");
+      expect(replay.highestContiguousSourceSeq).toBe(5002);
+      const [remaining] = await db.select({ count: sql<number>`count(*)::int` }).from(heartbeatRunEvents)
+        .where(eq(heartbeatRunEvents.runId, runId));
+      expect(remaining?.count).toBe(5006);
+    } finally { await temporary.cleanup(); }
+  }, 60_000);
+
   it.each([false, true])("deduplicates concurrent retry exhaustion (historical receipt: %s)", async (historical) => {
     const temporary = await startEmbeddedPostgresTestDatabase("paperclip-exhaustion-events-");
     const db = createDb(temporary.connectionString);
@@ -165,4 +222,46 @@ describe("P6-11..13 / P6-17 canonical event allocator", () => {
       await temporary.cleanup();
     }
   }, 60_000);
+
+  it("reports the highest contiguous source seq per source instance across gaps and late arrivals", async () => {
+    const temporary = await startEmbeddedPostgresTestDatabase("paperclip-native-cursor-");
+    const db = createDb(temporary.connectionString);
+    const companyId = "20000000-0000-4000-8000-000000000001";
+    const agentId = "20000000-0000-4000-8000-000000000002";
+    const runId = "20000000-0000-4000-8000-000000000003";
+    try {
+      await db.insert(companies).values({ id: companyId, name: "Cursor fixture", issuePrefix: "CUR" });
+      await db.insert(agents).values({ id: agentId, companyId, name: "Cursor agent" });
+      await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "running" });
+      const append = async (sourceInstanceId: string, sourceSeq: number) => (await appendHeartbeatRunEvent(db, {
+        companyId,
+        runId,
+        agentId,
+        eventType: "item.delta",
+        payload: { sourceInstanceId, sourceSeq },
+        nativeSource: {
+          sourceInstanceId,
+          sourceEventId: `${sourceInstanceId}:${sourceSeq}`,
+          sourceSeq,
+          protocolSchemaVersion: 1,
+          canonicalPayload: { sourceInstanceId, sourceSeq },
+        },
+      })).highestContiguousSourceSeq;
+
+      expect(await append("runner-a", 2)).toBe(0);
+      expect(await append("runner-a", 1)).toBe(2);
+      expect(await append("runner-a", 3)).toBe(3);
+      expect(await append("runner-a", 5)).toBe(3);
+      expect(await append("runner-a", 6)).toBe(3);
+      expect(await append("runner-b", 1)).toBe(1);
+      expect(await append("runner-a", 4)).toBe(6);
+      for (let sourceSeq = 7; sourceSeq <= 200; sourceSeq += 1) {
+        expect(await append("runner-a", sourceSeq)).toBe(sourceSeq);
+      }
+      expect(await append("runner-b", 3)).toBe(1);
+      expect(await append("runner-b", 2)).toBe(3);
+    } finally {
+      await temporary.cleanup();
+    }
+  }, 60_000);
 });

server/src/services/heartbeat-run-events.ts

@@ -202,18 +202,36 @@ export async function appendHeartbeatRunEvent(
 }
 
 async function contiguousCursor(db: Db, runId: string, sourceInstanceId: string): Promise<number> {
-  const rows = await db
+  const scope = and(
+    eq(heartbeatRunEvents.runId, runId),
+    eq(heartbeatRunEvents.sourceInstanceId, sourceInstanceId),
+    sql`${heartbeatRunEvents.sourceSeq} >= 1`,
+  );
+  const [range] = await db
+    .select({
+      count: sql<number>`count(*)`.mapWith(Number),
+      max: sql<number | null>`max(${heartbeatRunEvents.sourceSeq})`.mapWith(Number),
+    })
+    .from(heartbeatRunEvents)
+    .where(scope);
+  if (!range || !range.count) return 0;
+  if (range.max === range.count) return range.count;
+  const [first] = await db
     .select({ sourceSeq: heartbeatRunEvents.sourceSeq })
     .from(heartbeatRunEvents)
-    .where(and(
-      eq(heartbeatRunEvents.runId, runId),
-      eq(heartbeatRunEvents.sourceInstanceId, sourceInstanceId),
-    ))
-    .orderBy(asc(heartbeatRunEvents.sourceSeq));
-  let cursor = 0;
-  for (const row of rows) {
-    if (row.sourceSeq === cursor + 1) cursor += 1;
-    else if (row.sourceSeq !== null && row.sourceSeq > cursor + 1) break;
-  }
-  return cursor;
+    .where(and(scope, eq(heartbeatRunEvents.sourceSeq, 1)))
+    .limit(1);
+  if (!first) return 0;
+  const [last] = await db
+    .select({ sourceSeq: heartbeatRunEvents.sourceSeq })
+    .from(heartbeatRunEvents)
+    .where(and(scope, sql`not exists (
+      select 1 from heartbeat_run_events next_event
+      where next_event.run_id = heartbeat_run_events.run_id
+        and next_event.source_instance_id = heartbeat_run_events.source_instance_id
+        and next_event.source_seq = heartbeat_run_events.source_seq + 1
+    )`))
+    .orderBy(asc(heartbeatRunEvents.sourceSeq))
+    .limit(1);
+  return last?.sourceSeq ?? 0;
 }