perf(server): append ordinary run events in one statement (#16569)

Ordinary run events now validate their binding, allocate a sequence and insert the decoded receipt in one atomic PostgreSQL statement. Native replay and retry-exhaustion transactions keep their existing behavior.

Verification: exact-head full CI/E2E, Superagent, legacy security and Apex 5/5 passed. PostgreSQL concurrency, rollback and payload controls pass. Controlled network-delay comparison reduced median event persistence from about192ms to48ms; no staging response-time claim is made before deployment.

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

doc/run-log-events.md

@@ -5,6 +5,16 @@ Run-log events write to the `heartbeat_run_events` table
 Paperclip Telemetry events, and they are not OpenTelemetry exports. A run-log
 event needs no operator endpoint.
 
+## Ordinary event persistence
+
+Events without a native source identity or retry-exhaustion receipt use one
+atomic PostgreSQL statement. The statement checks the stored run's company and
+agent, allocates its next sequence, and inserts the event. An insertion failure
+also rolls back the sequence allocation. The returned row uses the existing
+column decoders, including the lossless payload codec. Callers still wait for
+the durable write before they publish or acknowledge the event. Native-source
+replay and retry-exhaustion deduplication retain their existing transaction path.
+
 ## Provider-pack preparation timing
 
 The existing `run.performance.span` event also records `provider_pack.select`,

server/src/__tests__/heartbeat-run-event-atomic-append.test.ts

@@ -0,0 +1,145 @@
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+import { eq, sql } from "drizzle-orm";
+import { drizzle } from "drizzle-orm/postgres-js";
+import { agents, companies, createDb, heartbeatRunEvents, heartbeatRuns } from "@paperclipai/db";
+import { appendHeartbeatRunEvent } from "../services/heartbeat-run-events.js";
+import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
+
+const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
+const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
+
+describeEmbeddedPostgres("atomic ordinary run events", () => {
+  let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
+  let db: ReturnType<typeof createDb>;
+  let otherDb: ReturnType<typeof createDb>;
+  const companyId = "30000000-0000-4000-8000-000000000001";
+  const agentId = "30000000-0000-4000-8000-000000000002";
+  const otherCompanyId = "30000000-0000-4000-8000-000000000003";
+  const otherAgentId = "30000000-0000-4000-8000-000000000004";
+  let ordinal = 10;
+
+  const makeRun = async () => {
+    const id = `30000000-0000-4000-8000-${String(ordinal++).padStart(12, "0")}`;
+    await db.insert(heartbeatRuns).values({ id, companyId, agentId, status: "running" });
+    return id;
+  };
+  const readRun = async (id: string) => (await db.select().from(heartbeatRuns)
+    .where(eq(heartbeatRuns.id, id)))[0]!;
+  const readEvents = (id: string) => db.select().from(heartbeatRunEvents)
+    .where(eq(heartbeatRunEvents.runId, id)).orderBy(heartbeatRunEvents.seq);
+
+  beforeAll(async () => {
+    temporary = await startEmbeddedPostgresTestDatabase("paperclip-atomic-events-");
+    db = createDb(temporary.connectionString);
+    otherDb = createDb(temporary.connectionString);
+    await db.insert(companies).values([
+      { id: companyId, name: "Atomic events", issuePrefix: "ATM" },
+      { id: otherCompanyId, name: "Other events", issuePrefix: "OTH" },
+    ]);
+    await db.insert(agents).values([
+      { id: agentId, companyId, name: "Event agent" },
+      { id: otherAgentId, companyId: otherCompanyId, name: "Other agent" },
+    ]);
+  }, 60_000);
+  afterAll(async () => { await temporary?.cleanup(); });
+
+  it("persists a decoded receipt in one executed statement", async () => {
+    const runId = await makeRun();
+    const statements: string[] = [];
+    const recorded = drizzle(db.$client, {
+      schema: db._.fullSchema,
+      logger: { logQuery: query => statements.push(query) },
+    }) as typeof db;
+    const payload = { nested: ["before\u0000after"], $paperclipRunEventJsonV1: "literal" };
+    const receipt = await appendHeartbeatRunEvent(recorded, {
+      companyId, agentId, runId, eventType: "run.performance.span", stream: "system",
+      level: "info", color: "green", message: "runner span", payload,
+    });
+    expect(receipt.disposition).toBe("committed");
+    expect(receipt.highestContiguousSourceSeq).toBe(0);
+    expect(receipt.row).toEqual(expect.objectContaining({
+      companyId, agentId, runId, seq: 1, eventType: "run.performance.span", stream: "system",
+      level: "info", color: "green", message: "runner span", payload,
+      sourceInstanceId: null, sourceEventId: null, sourceSeq: null,
+      sourcePayloadSha256: null, protocolSchemaVersion: null,
+    }));
+    expect(receipt.row.id).toEqual(expect.any(Number));
+    expect(receipt.row.createdAt).toBeInstanceOf(Date);
+    expect(await readEvents(runId)).toEqual([receipt.row]);
+    expect(statements).toHaveLength(1);
+  });
+
+  it("shares the run lock and sequence with native writers on separate connections", async () => {
+    const runId = await makeRun();
+    const receipts = await Promise.all(Array.from({ length: 40 }, (_, index) =>
+      appendHeartbeatRunEvent(index % 2 ? db : otherDb, {
+        companyId, agentId, runId, eventType: index % 2 ? "run.performance.span" : "item.completed",
+        message: `event-${index}`,
+        ...(index % 2 ? {} : { nativeSource: {
+          sourceInstanceId: "native-writer", sourceEventId: `event-${index}`,
+          sourceSeq: index / 2 + 1, protocolSchemaVersion: 1,
+          canonicalPayload: { ordinal: index },
+        } }),
+      })));
+    expect(new Set(receipts.map(receipt => receipt.row.id)).size).toBe(40);
+    expect((await readEvents(runId)).map(row => row.seq)).toEqual(
+      Array.from({ length: 40 }, (_, index) => index + 1),
+    );
+    expect((await readRun(runId)).nextEventSeq).toBe(41);
+    const replay = await appendHeartbeatRunEvent(db, {
+      companyId, agentId, runId, eventType: "item.completed", message: "event-0",
+      nativeSource: { sourceInstanceId: "native-writer", sourceEventId: "event-0",
+        sourceSeq: 1, protocolSchemaVersion: 1, canonicalPayload: { ordinal: 0 } },
+    });
+    expect(replay.disposition).toBe("duplicate");
+    expect(replay.highestContiguousSourceSeq).toBe(20);
+    expect((await readRun(runId)).nextEventSeq).toBe(41);
+  });
+
+  it("rejects mismatched authority without changing either run", async () => {
+    const runId = await makeRun();
+    const otherRunId = "30000000-0000-4000-8000-000000000099";
+    await db.insert(heartbeatRuns).values({ id: otherRunId, companyId: otherCompanyId,
+      agentId: otherAgentId, status: "running" });
+    const before = [await readRun(runId), await readRun(otherRunId)];
+    for (const authority of [
+      { companyId: otherCompanyId, agentId, runId },
+      { companyId, agentId: otherAgentId, runId },
+      { companyId, agentId, runId: otherRunId },
+      { companyId, agentId, runId: "30000000-0000-4000-8000-000000000098" },
+    ]) {
+      await expect(appendHeartbeatRunEvent(db, { ...authority, eventType: "stdout" }))
+        .rejects.toThrow("heartbeat_run_event_binding_mismatch");
+    }
+    expect([await readRun(runId), await readRun(otherRunId)]).toEqual(before);
+    expect(await readEvents(runId)).toEqual([]);
+    expect(await readEvents(otherRunId)).toEqual([]);
+  });
+
+  it("rolls back allocation on insertion failure and in an outer transaction", async () => {
+    const runId = await makeRun();
+    const before = await readRun(runId);
+    await db.execute(sql`alter table heartbeat_run_events add constraint atomic_event_failure
+      check (message is distinct from 'force-event-failure') not valid`);
+    try {
+      await expect(appendHeartbeatRunEvent(db, {
+        companyId, agentId, runId, eventType: "stdout", message: "force-event-failure",
+      })).rejects.toThrow();
+      expect(await readRun(runId)).toEqual(before);
+      expect(await readEvents(runId)).toEqual([]);
+    } finally {
+      await db.execute(sql`alter table heartbeat_run_events drop constraint atomic_event_failure`);
+    }
+    await expect(db.transaction(async tx => {
+      await appendHeartbeatRunEvent(tx as unknown as typeof db, {
+        companyId, agentId, runId, eventType: "stdout",
+      });
+      throw new Error("outer rollback");
+    })).rejects.toThrow("outer rollback");
+    expect(await readRun(runId)).toEqual(before);
+    expect(await readEvents(runId)).toEqual([]);
+    const receipt = await appendHeartbeatRunEvent(db, { companyId, agentId, runId, eventType: "stdout" });
+    expect(receipt.row.seq).toBe(1);
+    expect(receipt.row.payload).toBeNull();
+  });
+});

server/src/__tests__/recovery-stale-issue-lock-sweep.test.ts

@@ -1,5 +1,5 @@
 import { randomUUID } from "node:crypto";
-import { eq } from "drizzle-orm";
+import { eq, sql } from "drizzle-orm";
 import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
 import {
   costEvents,
@@ -1049,27 +1049,34 @@ describeEmbeddedPostgres("recovery sweepStaleIssueLocks", () => {
     // Make only the audit-event insert fail. The run update commits the
     // terminal status first, so the audit write is best-effort. The sweep must
     // catch the failure and still clear the lock.
-    const transactionSpy = vi
-      .spyOn(db, "transaction")
-      .mockRejectedValueOnce(new Error("simulated audit write failure"));
+    await db.execute(sql`
+      create function reject_stale_lock_audit() returns trigger language plpgsql as $$
+      begin raise exception 'simulated audit write failure'; end;
+      $$
+    `);
 
     try {
+      await db.execute(sql`
+        create trigger reject_stale_lock_audit before insert on heartbeat_run_events
+        for each row execute function reject_stale_lock_audit()
+      `);
       const heartbeat = heartbeatService(db);
       const result = await heartbeat.sweepStaleIssueLocks();
 
       expect(result.terminalizedRunIds).toEqual([runningRunId]);
       expect(result.cleared).toBe(1);
     } finally {
-      transactionSpy.mockRestore();
+      await db.execute(sql`drop trigger if exists reject_stale_lock_audit on heartbeat_run_events`);
+      await db.execute(sql`drop function reject_stale_lock_audit()`);
     }
 
     // The run reached its terminal status even though the audit write failed.
     const runStatus = await db
-      .select({ status: heartbeatRuns.status })
+      .select({ status: heartbeatRuns.status, nextEventSeq: heartbeatRuns.nextEventSeq })
       .from(heartbeatRuns)
       .where(eq(heartbeatRuns.id, runningRunId))
-      .then((rows) => rows[0]?.status);
-    expect(runStatus).toBe("interrupted");
+      .then((rows) => rows[0]);
+    expect(runStatus).toEqual({ status: "interrupted", nextEventSeq: 1 });
 
     // The sweep cleared the lock in the same pass.
     const lock = await db

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

@@ -1,4 +1,4 @@
-import { and, asc, eq, or, sql } from "drizzle-orm";
+import { and, asc, eq, getTableColumns, or, sql } from "drizzle-orm";
 import type { Db } from "@paperclipai/db";
 import { heartbeatRunEvents, heartbeatRuns } from "@paperclipai/db";
 import { nativeSha256 } from "./native-runtime/canonical.js";
@@ -68,6 +68,30 @@ export async function appendHeartbeatRunEvent(
   db: Db,
   input: AppendHeartbeatRunEventInput,
 ): Promise<AppendHeartbeatRunEventResult> {
+  if (!input.nativeSource && !input.retryExhaustion) {
+    const allocated = db.$with("allocated_run_event", {}).as(sql`
+      update ${heartbeatRuns}
+      set next_event_seq = ${heartbeatRuns.nextEventSeq} + 1,
+          updated_at = ${sql.param(new Date(), heartbeatRuns.updatedAt)}
+      where ${heartbeatRuns.id} = ${input.runId}
+        and ${heartbeatRuns.companyId} = ${input.companyId}
+        and ${heartbeatRuns.agentId} = ${input.agentId}
+      returning company_id, id as run_id, agent_id, next_event_seq - 1 as seq
+    `);
+    const appended = db.$with("appended_run_event", getTableColumns(heartbeatRunEvents)).as(sql`
+      insert into ${heartbeatRunEvents}
+        (company_id, run_id, agent_id, seq, event_type, stream, level, color, message, payload)
+      select company_id, run_id, agent_id, seq,
+        ${input.eventType}, ${input.stream ?? null}, ${input.level ?? null},
+        ${input.color ?? null}, ${input.message ?? null},
+        ${sql.param(input.payload ?? null, heartbeatRunEvents.payload)}
+      from ${allocated}
+      returning *
+    `);
+    const [row] = await db.with(allocated, appended).select().from(appended);
+    if (!row) throw new Error("heartbeat_run_event_binding_mismatch");
+    return { row, disposition: "committed", highestContiguousSourceSeq: 0 };
+  }
   return db.transaction(async (tx) => {
     const run = await tx
       .select()

server/src/services/native-runtime/native-session-executor.test.ts

@@ -66,6 +66,7 @@ import {
   NativeSessionProtocolIntegrityError,
 } from "../../vendor/paperclip-runner/index.js";
 import * as issueServiceModule from "../issues.js";
+import * as runEventWriter from "../heartbeat-run-events.js";
 import { NativePermissionDeclinedError } from "./native-permission-decline.js";
 import { resolvePaperclipRunnerNativeProviderInput } from "./provider-profile.js";
 import {
@@ -9678,12 +9679,24 @@ describe("native process ownership", () => {
       startedAt: "2026-08-18T18:00:00.000Z",
     };
     const onSpawn = vi.fn(async () => undefined);
+    const db = leaseDb(execution);
+    const admissionWriter = vi.spyOn(runEventWriter, "appendHeartbeatRunEvent").mockResolvedValueOnce({
+      row: {
+        id: 1, companyId: execution.binding.companyId, runId: execution.binding.runId,
+        agentId: execution.binding.agentId, seq: 1, eventType: "native.process_start_requested",
+        stream: "system", level: "info", color: null, message: null, payload: null,
+        sourceInstanceId: null, sourceEventId: null, sourceSeq: null,
+        sourcePayloadSha256: null, protocolSchemaVersion: null, createdAt: new Date(),
+      },
+      disposition: "committed", highestContiguousSourceSeq: 0,
+    });
     state.createBackend.mockClear();
     state.execute.mockReset().mockImplementation(async (options) => {
       await options.onSessionAdmission();
-      expect(updates).toContainEqual({ table: heartbeatRunEvents, values: expect.objectContaining({
+      expect(admissionWriter).toHaveBeenCalledWith(db, expect.objectContaining({
         eventType: "native.process_start_requested", runId: execution.binding.runId,
-      }) });
+        companyId: execution.binding.companyId, agentId: execution.binding.agentId,
+      }));
       await options.backend.onSpawn(processMetadata);
       return {
         result: { summary: "completed" },
@@ -9697,18 +9710,17 @@ describe("native process ownership", () => {
         highestContiguousSourceSeq: 1,
       };
     });
-    const updates: Array<{ table: unknown; values: Record<string, unknown> }> = [];
     state.createBackend.mockImplementationOnce((_input, options) => {
-      expect(updates.some(update => update.values.eventType === "native.process_start_requested")).toBe(false);
+      expect(admissionWriter).not.toHaveBeenCalled();
       return { kind: "test", onSpawn: options.onSpawn };
     });
 
-    await executePaperclipNativeSession({
-      db: leaseDb(execution, {}, {}, updates),
-      execution,
-      runnerInstanceId: "runner",
-      onSpawn,
-    });
+    try {
+      await executePaperclipNativeSession({ db, execution, runnerInstanceId: "runner", onSpawn });
+      expect(admissionWriter).toHaveBeenCalledOnce();
+    } finally {
+      admissionWriter.mockRestore();
+    }
 
     expect(state.createBackend).toHaveBeenCalledWith(
       execution,