perf: narrow run bindings during event ingestion (#17378)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Agent runs append native events and retry-exhaustion receipts.
> - These writes lock the run row and verify its company and agent
bindings.
> - The binding check fetched the full run, including prompts,
configuration, and output.
> - This pull request reads only the two fields that the check uses.
> - The benefit is less database transfer while the same row lock
protects sequence allocation and replay checks.

## Linked Issues or Issue Description

Related: #17355 and #17348 reduce other run-read payloads. GitHub
searches for event-read and narrow-run PRs found no duplicate of this
event-ingestion change.

**What existing behavior does this improve?**

The run binding read during native event and retry-exhaustion ingestion.

**Current behavior**

The transaction uses a full-row `SELECT ... FOR UPDATE` to check only
`companyId` and `agentId`. Large run context, runner configuration, and
output cross the database connection for each such check.

**Proposed behavior**

Select only the company and agent IDs. Keep the same run ID predicate,
row lock, transaction, sequence allocator, replay cursor, and binding
checks.

**Reason and benefit**

The check does not use the other fields. Narrowing the query removes
that repeated transfer without a cache or an authorization change.

**Breaking changes**

None. No API response, schema, event, or scheduling changes.

## What Changed

- Narrow the existing run-row read inside `appendHeartbeatRunEvent`. The
event service continues to own its transaction and binding validation.
- Exercise native and retry-exhaustion writes and replays with large
stored payloads. Record the executed SQL to prove the lock reads only
the two binding fields.
- Extend missing-run and company/agent mismatch coverage to both
transaction paths. Keep the existing concurrent-writer and rollback
tests.
- Document the read boundary.

## Verification

- Focused PostgreSQL event append and sequencing suites: 9 tests passed.
They include concurrent writers, replay conflicts, rollback, cursor
checks, large payloads, and company/agent/missing-run denials.
- The first new retry-exhaustion fixture failed its duplicate assertion
because it omitted the exhaustion fields from the event payload. The
existing producer contract stores those fields in the payload. The
fixture now does the same; the rerun passed. No production retry or
replay logic changed.
- `pnpm build`: passed.
- `pnpm -r typecheck`: passed.
- Greptile Apex: 5/5 on `76a6d3ee813ae47919b1ec8e8b88e01e58d5655d`,
clean check and no review threads. The [full PR
workflow](https://github.com/paperclipai/paperclip/actions/runs/40028758665)
passed on that exact head: 54 checks passed and two conditional checks
were skipped. The branch is conflict-free.
- The full local all-in-one suite is not repeated for this follow-up.
Full exact-head CI will supply the broader suite coverage.
- No hosted deployment or bill reduction is claimed for this follow-up.

## Risks

- Low scope: only the selected columns change. The `FOR UPDATE` lock,
transaction boundaries, sequence reservation, duplicate checks, and
failure handling remain in place.
- Native replay and legacy event writers must still serialize on the
same row. Existing real PostgreSQL concurrency tests cover that
boundary.
- No migration, provider change, new timer, or operator configuration is
required.

## Model Used

OpenAI Codex (GPT-6 family), with code execution and repository tools.
The runtime does not expose the exact model ID or context window.

## 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
#` 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 (focused suites above)
- [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>

doc/DEVELOPING.md

@@ -2871,6 +2871,10 @@ This check selects only the run ID, company ID, agent ID, and status through
 configuration, or provider results. Keep this read fresh so terminal or deleted
 runs cannot publish late progress from an old executor snapshot.
 
+Native event and retry-exhaustion ingestion lock the run row to verify its
+company and agent bindings. That lock reads only those two fields. Sequence
+allocation, replay checks, and event insertion remain in the same transaction.
+
 Run the deterministic native event replay with the normal server test runner:
 
 ```sh

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

@@ -1,3 +1,4 @@
+import { randomBytes } from "node:crypto";
 import { afterAll, beforeAll, describe, expect, it } from "vitest";
 import { eq, sql } from "drizzle-orm";
 import { drizzle } from "drizzle-orm/postgres-js";
@@ -96,6 +97,45 @@ describeEmbeddedPostgres("atomic ordinary run events", () => {
     expect((await readRun(runId)).nextEventSeq).toBe(41);
   });
 
+  it.each(["native", "retry-exhaustion"])("locks only run bindings for %s events with large run payloads", async (kind) => {
+    const runId = await makeRun();
+    await db.update(heartbeatRuns).set({
+      contextSnapshot: { prompt: randomBytes(128 * 1024).toString("hex") },
+      runnerProfileJson: { configuration: randomBytes(128 * 1024).toString("hex") },
+      resultJson: { providerData: randomBytes(128 * 1024).toString("hex") },
+    }).where(eq(heartbeatRuns.id, runId));
+    const statements: string[] = [];
+    const recorded = drizzle(db.$client, {
+      schema: db._.fullSchema,
+      logger: { logQuery: query => statements.push(query) },
+    }) as typeof db;
+    const exhaustion = { retryReason: "transient_failure", scheduledRetryAttempt: 1, maxAttempts: 1 };
+    const input = {
+      companyId, agentId, runId, eventType: "lifecycle",
+      message: "Bounded retry exhausted after one attempt",
+      ...(kind === "native" ? {
+        nativeSource: { sourceInstanceId: "native-writer", sourceEventId: "event-one",
+          sourceSeq: 1, protocolSchemaVersion: 1, canonicalPayload: { ordinal: 1 } },
+      } : { retryExhaustion: exhaustion, payload: exhaustion }),
+    };
+    const committed = await appendHeartbeatRunEvent(recorded, input);
+    const replay = await appendHeartbeatRunEvent(recorded, input);
+    expect(committed.disposition).toBe("committed");
+    expect(committed.row.seq).toBe(1);
+    expect(replay.disposition).toBe("duplicate");
+    expect(replay.row.id).toBe(committed.row.id);
+    expect(replay.highestContiguousSourceSeq).toBe(kind === "native" ? 1 : 0);
+    expect((await readRun(runId)).nextEventSeq).toBe(2);
+    expect(await readEvents(runId)).toEqual([committed.row]);
+    const lockReads = statements.filter(query => query.startsWith("select ") &&
+      query.includes('from "heartbeat_runs"') && query.endsWith("for update"));
+    expect(lockReads).toHaveLength(2);
+    for (const query of lockReads) {
+      expect(query).toMatch(/^select "company_id", "agent_id" from "heartbeat_runs" /);
+      expect(query).not.toMatch(/context_snapshot|runner_profile_json|result_json|stdout_excerpt/);
+    }
+  });
+
   it("rejects mismatched authority without changing either run", async () => {
     const runId = await makeRun();
     const otherRunId = "30000000-0000-4000-8000-000000000099";
@@ -108,8 +148,15 @@ describeEmbeddedPostgres("atomic ordinary run events", () => {
       { 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");
+      for (const source of [
+        {},
+        { nativeSource: { sourceInstanceId: "native-writer", sourceEventId: "mismatched-event",
+          sourceSeq: 1, protocolSchemaVersion: 1, canonicalPayload: { ordinal: 1 } } },
+        { retryExhaustion: { retryReason: "transient_failure", scheduledRetryAttempt: 1, maxAttempts: 1 } },
+      ]) {
+        await expect(appendHeartbeatRunEvent(db, { ...authority, ...source, eventType: "stdout" }))
+          .rejects.toThrow("heartbeat_run_event_binding_mismatch");
+      }
     }
     expect([await readRun(runId), await readRun(otherRunId)]).toEqual(before);
     expect(await readEvents(runId)).toEqual([]);

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

@@ -94,7 +94,7 @@ export async function appendHeartbeatRunEvent(
   }
   return db.transaction(async (tx) => {
     const run = await tx
-      .select()
+      .select({ companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId })
       .from(heartbeatRuns)
       .where(eq(heartbeatRuns.id, input.runId))
       .for("update")