perf(runner): batch queued native text without buffering delay

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

packages/paperclip-runner/src/contracts/control-plane-port.ts

@@ -30,6 +30,15 @@ export interface AppendControlPlaneEventOptions {
   signal: AbortSignal;
 }
 
+/** Only ordinary display text can share a persistence batch. */
+export function isBatchableNativeTextEvent(event: PrpEvent): boolean {
+  return event.sourceKind === "runner"
+    && event.priority !== 0
+    && event.eventType === "item.delta"
+    && (event.payload.kind === "agentMessage" || event.payload.kind === "reasoning")
+    && typeof event.payload.text === "string";
+}
+
 export interface CheckpointControlPlaneSessionOptions {
   /**
    * Cancellation is a durability boundary: once aborted, checkpoint work
@@ -84,6 +93,15 @@ export interface ControlPlanePort {
     event: NativeRunEvent | PrpEvent,
     options?: AppendControlPlaneEventOptions,
   ): Promise<AppendedEventReceipt>;
+  /**
+   * Commit ordinary text together without changing its source identities.
+   * Receipts have the input order and are returned only after the whole batch
+   * commits. Cancellation has the same durability boundary as appendEvent.
+   */
+  appendEvents?(
+    events: readonly PrpEvent[],
+    options?: AppendControlPlaneEventOptions,
+  ): Promise<AppendedEventReceipt[]>;
   replayEvents(
     input: ReplayControlPlaneEventsInput,
     options?: FinalizeControlPlaneOperationOptions,

packages/paperclip-runner/src/native-event-persistence.test.ts

@@ -0,0 +1,296 @@
+import { describe, expect, it, vi } from "vitest";
+import type { AppendedEventReceipt, AppendControlPlaneEventOptions, ControlPlanePort } from "./contracts/control-plane-port.js";
+import { NativeEventPersistence } from "./native-event-persistence.js";
+import type { PrpEvent } from "./protocol/replay-contract.js";
+
+function deferred<T>() {
+  let resolve!: (value: T) => void;
+  let reject!: (reason: unknown) => void;
+  const promise = new Promise<T>((yes, no) => { resolve = yes; reject = no; });
+  return { promise, resolve, reject };
+}
+
+function text(seq: number, value = `text-${seq}`): PrpEvent {
+  return {
+    schema: "paperclip.prp.event.v1", schemaVersion: 1,
+    sourceEventId: `runner:run:${seq}`, sourceSeq: seq,
+    sourceInstanceId: "runner", sourceKind: "runner", runId: "run",
+    normalizedSessionId: "session", turnId: "turn", itemId: "message",
+    eventType: "item.delta", priority: 1, emittedAt: "2026-11-03T00:00:00.000Z",
+    payload: { kind: "agentMessage", text: value },
+  };
+}
+
+function receipt(event: PrpEvent): AppendedEventReceipt {
+  return { cursor: event.sourceSeq, highestContiguousSourceSeq: event.sourceSeq, disposition: "committed" };
+}
+
+class EventStream implements AsyncIterator<PrpEvent> {
+  #values: PrpEvent[] = [];
+  #pending: ReturnType<typeof deferred<IteratorResult<PrpEvent>>> | undefined;
+  #reads: Array<{ count: number; resolve: () => void }> = [];
+  reads = 0;
+  maxConcurrentReads = 0;
+
+  push(...events: PrpEvent[]) {
+    this.#values.push(...events);
+    if (this.#pending !== undefined && this.#values.length > 0) {
+      this.#pending.resolve({ value: this.#values.shift()!, done: false });
+      this.#pending = undefined;
+    }
+  }
+
+  readCount(count: number): Promise<void> {
+    return this.reads >= count ? Promise.resolve() : new Promise(resolve => this.#reads.push({ count, resolve }));
+  }
+
+  next(): Promise<IteratorResult<PrpEvent>> {
+    if (this.#pending !== undefined) throw new Error("concurrent iterator read");
+    this.reads += 1;
+    this.maxConcurrentReads = Math.max(this.maxConcurrentReads, 1);
+    for (const waiter of this.#reads.splice(0)) {
+      if (this.reads >= waiter.count) waiter.resolve();
+      else this.#reads.push(waiter);
+    }
+    const value = this.#values.shift();
+    if (value !== undefined) return Promise.resolve({ value, done: false });
+    this.#pending = deferred<IteratorResult<PrpEvent>>();
+    return this.#pending.promise;
+  }
+
+  async return(): Promise<IteratorResult<PrpEvent>> {
+    this.#pending?.resolve({ value: undefined, done: true });
+    this.#pending = undefined;
+    return { value: undefined, done: true };
+  }
+}
+
+function port() {
+  return {
+    appendEvent: vi.fn(async (event: PrpEvent, _options?: AppendControlPlaneEventOptions) => receipt(event)),
+    appendEvents: vi.fn(async (events: readonly PrpEvent[], _options?: AppendControlPlaneEventOptions) => events.map(receipt)),
+  };
+}
+
+describe("native text persistence", () => {
+  it("starts the first and isolated text writes immediately, without a batch deadline", async () => {
+    const stream = new EventStream();
+    const persistence = port();
+    const firstWrite = deferred<AppendedEventReceipt>();
+    persistence.appendEvent.mockImplementationOnce(() => firstWrite.promise);
+    const events = new NativeEventPersistence(stream, persistence);
+    const first = text(1);
+    stream.push(first);
+    expect((await events.next()).value).toBe(first);
+    const append = events.appendEvent(first);
+    expect(persistence.appendEvent).toHaveBeenCalledExactlyOnceWith(first, undefined);
+    expect(persistence.appendEvents).not.toHaveBeenCalled();
+    firstWrite.resolve(receipt(first));
+    await append;
+    const second = text(2);
+    stream.push(second);
+    expect((await events.next()).value).toBe(second);
+    const secondAppend = events.appendEvent(second);
+    expect(persistence.appendEvent).toHaveBeenLastCalledWith(second, undefined);
+    await secondAppend;
+    await events.return();
+  });
+
+  it("batches ready text without waiting for an outstanding read or reading concurrently", async () => {
+    const stream = new EventStream();
+    const persistence = port();
+    const firstWrite = deferred<AppendedEventReceipt>();
+    persistence.appendEvent.mockImplementationOnce(() => firstWrite.promise);
+    const batchWrite = deferred<AppendedEventReceipt[]>();
+    persistence.appendEvents.mockImplementationOnce(() => batchWrite.promise);
+    const events = new NativeEventPersistence(stream, persistence);
+    const source = [text(1), text(2), text(3)];
+    stream.push(source[0]!);
+    await events.next();
+    const first = events.appendEvent(source[0]!);
+    stream.push(source[1]!, source[2]!);
+    await stream.readCount(4); // The fourth read remains pending.
+    firstWrite.resolve(receipt(source[0]!));
+    await first;
+    expect((await events.next()).value).toBe(source[1]);
+    const second = events.appendEvent(source[1]!);
+    expect(persistence.appendEvents).toHaveBeenCalledExactlyOnceWith(source.slice(1), undefined);
+    batchWrite.resolve(source.slice(1).map(receipt));
+    await second;
+    expect((await events.next()).value).toBe(source[2]);
+    expect(await events.appendEvent(source[2]!)).toEqual(receipt(source[2]!));
+    expect(persistence.appendEvent).toHaveBeenCalledOnce();
+    const late = text(4);
+    stream.push(late);
+    expect((await events.next()).value).toBe(late);
+    await events.appendEvent(late);
+    expect(stream.maxConcurrentReads).toBe(1);
+    await events.return();
+  });
+
+  it.each([
+    { eventType: "tool.execution.started", priority: 1, payload: {} },
+    { eventType: "run.result.proposed", priority: 0, payload: {} },
+    { eventType: "turn.completed", priority: 0, payload: {} },
+    { eventType: "item.delta", priority: 0, payload: { kind: "agentMessage", text: "urgent" } },
+    { eventType: "item.delta", priority: 1, payload: { kind: "commandExecution", text: "output" } },
+    { eventType: "item.delta", priority: 1, sourceKind: "control_plane", payload: { kind: "agentMessage", text: "evidence", evidenceOutcome: "accepted" } },
+  ] satisfies Array<Partial<PrpEvent>>)("stops read-ahead at $eventType priority $priority", async boundary => {
+    const stream = new EventStream();
+    const persistence = port();
+    const write = deferred<AppendedEventReceipt>();
+    persistence.appendEvent.mockImplementationOnce(() => write.promise);
+    const events = new NativeEventPersistence(stream, persistence);
+    const first = text(1), second = text(2), control = { ...text(3), ...boundary }, tail = text(4);
+    stream.push(first);
+    await events.next();
+    const append = events.appendEvent(first);
+    stream.push(second, control, tail);
+    await stream.readCount(3);
+    // Let the resolved boundary pass through the reader; no fourth read starts.
+    await Promise.resolve();
+    await Promise.resolve();
+    expect(stream.reads).toBe(3);
+    expect(persistence.appendEvent).toHaveBeenCalledOnce();
+    write.resolve(receipt(first));
+    await append;
+    expect((await events.next()).value).toBe(second);
+    await events.appendEvent(second);
+    expect(stream.reads).toBe(3); // The retained boundary still blocks read-ahead.
+    expect((await events.next()).value).toBe(control);
+    await events.appendEvent(control);
+    expect(persistence.appendEvents).not.toHaveBeenCalled();
+    expect(persistence.appendEvent.mock.calls.map(call => call[0])).toEqual([first, second, control]);
+    await events.return();
+  });
+
+  it("caps buffered event count and submitted batches at 64", async () => {
+    const stream = new EventStream();
+    const persistence = port();
+    const write = deferred<AppendedEventReceipt>();
+    persistence.appendEvent.mockImplementationOnce(() => write.promise);
+    const events = new NativeEventPersistence(stream, persistence);
+    const source = Array.from({ length: 130 }, (_, index) => text(index + 1));
+    stream.push(...source);
+    await events.next();
+    const first = events.appendEvent(source[0]!);
+    await stream.readCount(65);
+    await Promise.resolve();
+    await Promise.resolve();
+    expect(stream.reads).toBe(65);
+    write.resolve(receipt(source[0]!));
+    await first;
+    for (let index = 1; index < source.length; index += 1) {
+      expect((await events.next()).value).toBe(source[index]);
+      await events.appendEvent(source[index]!);
+    }
+    expect(persistence.appendEvents.mock.calls[0]![0]).toHaveLength(64);
+    expect(persistence.appendEvents.mock.calls.every(([batch]) => batch.length <= 64)).toBe(true);
+    expect(persistence.appendEvent.mock.calls.length + persistence.appendEvents.mock.calls.flatMap(([batch]) => batch).length)
+      .toBe(source.length);
+    await events.return();
+  });
+
+  it("caps batch bytes and appends an oversized event singly", async () => {
+    const stream = new EventStream();
+    const persistence = port();
+    const write = deferred<AppendedEventReceipt>();
+    persistence.appendEvent.mockImplementationOnce(() => write.promise);
+    const events = new NativeEventPersistence(stream, persistence);
+    const source = [text(1), text(2, "中".repeat(10_000)), text(3, "中".repeat(10_000)), text(4, "x".repeat(70_000))];
+    stream.push(...source);
+    await events.next();
+    const append = events.appendEvent(source[0]!);
+    await stream.readCount(4);
+    write.resolve(receipt(source[0]!));
+    await append;
+    for (let index = 1; index < source.length; index += 1) {
+      expect((await events.next()).value).toBe(source[index]);
+      await events.appendEvent(source[index]!);
+    }
+    expect(persistence.appendEvents).toHaveBeenCalledOnce();
+    expect(persistence.appendEvents.mock.calls.every(([batch]) =>
+      batch.reduce((sum, event) => sum + Buffer.byteLength(JSON.stringify(event)), 0) <= 64 * 1024)).toBe(true);
+    expect(persistence.appendEvent.mock.calls.map(call => call[0])).toContain(source[3]);
+    await events.return();
+  });
+
+  it("preserves singleton behavior for ports without batch support", async () => {
+    const stream = new EventStream();
+    const write = deferred<AppendedEventReceipt>();
+    const persistence = { appendEvent: vi.fn(() => write.promise) };
+    const events = new NativeEventPersistence(stream, persistence);
+    const source = [text(1), text(2)];
+    stream.push(...source);
+    await events.next();
+    const append = events.appendEvent(source[0]!);
+    expect(stream.reads).toBe(1);
+    write.resolve(receipt(source[0]!));
+    await append;
+    expect(stream.reads).toBe(1);
+    expect((await events.next()).value).toBe(source[1]);
+    await events.return();
+  });
+
+  it("does not append buffered text after abort and releases its outstanding reader", async () => {
+    const stream = new EventStream();
+    const abort = new AbortController();
+    let committed = false;
+    const release = deferred<void>();
+    const persistence: Pick<ControlPlanePort, "appendEvent" | "appendEvents"> = {
+      async appendEvent(event, options) {
+        await release.promise;
+        options?.signal.throwIfAborted();
+        committed = true;
+        return receipt(event as PrpEvent);
+      },
+      appendEvents: vi.fn(async batch => batch.map(receipt)),
+    };
+    const events = new NativeEventPersistence(stream, persistence);
+    const source = [text(1), text(2)];
+    stream.push(...source);
+    await events.next();
+    const append = events.appendEvent(source[0]!, { signal: abort.signal });
+    await stream.readCount(3);
+    abort.abort(new Error("stopped"));
+    release.resolve();
+    await expect(append).rejects.toThrow("stopped");
+    expect(committed).toBe(false);
+    await expect(events.appendEvent(source[1]!, { signal: abort.signal })).rejects.toThrow("stopped");
+    expect(persistence.appendEvents).not.toHaveBeenCalled();
+    await events.return();
+    expect(await events.next()).toEqual({ value: undefined, done: true });
+    expect(stream.reads).toBe(3);
+  });
+
+  it("passes the same cancellation boundary to an in-flight batch", async () => {
+    const stream = new EventStream(), abort = new AbortController();
+    const firstWrite = deferred<AppendedEventReceipt>(), batchWrite = deferred<void>();
+    const persistence = port();
+    let committed = false;
+    persistence.appendEvent.mockImplementationOnce(() => firstWrite.promise);
+    persistence.appendEvents.mockImplementationOnce(async (batch, options) => {
+      expect(options?.signal).toBe(abort.signal);
+      await batchWrite.promise;
+      options?.signal.throwIfAborted();
+      committed = true;
+      return batch.map(receipt);
+    });
+    const events = new NativeEventPersistence(stream, persistence);
+    const source = [text(1), text(2), text(3)];
+    stream.push(...source);
+    await events.next();
+    const first = events.appendEvent(source[0]!, { signal: abort.signal });
+    await stream.readCount(4);
+    firstWrite.resolve(receipt(source[0]!));
+    await first;
+    expect((await events.next()).value).toBe(source[1]);
+    const batch = events.appendEvent(source[1]!, { signal: abort.signal });
+    expect(persistence.appendEvents).toHaveBeenCalledOnce();
+    abort.abort(new Error("batch stopped"));
+    batchWrite.resolve();
+    await expect(batch).rejects.toThrow("batch stopped");
+    expect(committed).toBe(false);
+    await events.return();
+  });
+});

packages/paperclip-runner/src/native-event-persistence.ts

@@ -0,0 +1,162 @@
+import {
+  isBatchableNativeTextEvent,
+  type AppendedEventReceipt,
+  type AppendControlPlaneEventOptions,
+  type ControlPlanePort,
+} from "./contracts/control-plane-port.js";
+import type { PrpEvent } from "./protocol/replay-contract.js";
+
+const MAX_BATCH_EVENTS = 64;
+const MAX_BATCH_BYTES = 64 * 1024;
+
+type BufferedEvent = { event: PrpEvent; bytes: number | null };
+type ReadOutcome =
+  | { ok: true; result: IteratorResult<PrpEvent> }
+  | { ok: false; error: unknown };
+
+function bufferedEvent(event: PrpEvent): BufferedEvent {
+  return {
+    event,
+    bytes: isBatchableNativeTextEvent(event)
+      ? Buffer.byteLength(JSON.stringify(event), "utf8")
+      : null,
+  };
+}
+
+function sameStream(left: PrpEvent, right: PrpEvent): boolean {
+  return left.runId === right.runId
+    && left.normalizedSessionId === right.normalizedSessionId
+    && left.sourceInstanceId === right.sourceInstanceId
+    && left.turnId === right.turnId;
+}
+
+/**
+ * One owner of the transformed session iterator and its append operations.
+ * Read ahead only during an ordinary text write; never wait to fill a batch.
+ */
+export class NativeEventPersistence implements AsyncIterator<PrpEvent> {
+  readonly #iterator: AsyncIterator<PrpEvent>;
+  readonly #port: Pick<ControlPlanePort, "appendEvent" | "appendEvents">;
+  readonly #buffered: BufferedEvent[] = [];
+  readonly #committed: Array<{ event: PrpEvent; receipt: AppendedEventReceipt }> = [];
+  readonly #receipts = new WeakMap<PrpEvent, AppendedEventReceipt>();
+  #pendingRead: Promise<void> | undefined;
+  #readOutcome: ReadOutcome | undefined;
+  #stopped = false;
+
+  constructor(
+    iterator: AsyncIterator<PrpEvent>,
+    port: Pick<ControlPlanePort, "appendEvent" | "appendEvents">,
+  ) {
+    this.#iterator = iterator;
+    this.#port = port;
+  }
+
+  #requestRead(): Promise<void> {
+    if (this.#readOutcome !== undefined) return Promise.resolve();
+    if (this.#pendingRead !== undefined) return this.#pendingRead;
+    try {
+      this.#pendingRead = Promise.resolve(this.#iterator.next()).then(
+        result => {
+          this.#readOutcome = { ok: true, result };
+          this.#pendingRead = undefined;
+        },
+        error => {
+          this.#readOutcome = { ok: false, error };
+          this.#pendingRead = undefined;
+        },
+      );
+    } catch (error) {
+      this.#readOutcome = { ok: false, error };
+      return Promise.resolve();
+    }
+    return this.#pendingRead;
+  }
+
+  async next(): Promise<IteratorResult<PrpEvent>> {
+    if (this.#stopped) return { value: undefined, done: true };
+    const committed = this.#committed.shift();
+    if (committed !== undefined) {
+      this.#receipts.set(committed.event, committed.receipt);
+      return { value: committed.event, done: false };
+    }
+    const buffered = this.#buffered.shift();
+    if (buffered !== undefined) return { value: buffered.event, done: false };
+    await this.#requestRead();
+    const outcome = this.#readOutcome!;
+    if (!outcome.ok) throw outcome.error;
+    // Keep end-of-stream stable for the governed accounting drain.
+    if (!outcome.result.done) this.#readOutcome = undefined;
+    return outcome.result;
+  }
+
+  async #collect(scope: PrpEvent, writing: () => boolean): Promise<void> {
+    let bytes = this.#buffered.reduce((sum, entry) => sum + (entry.bytes ?? 0), 0);
+    if (this.#buffered.some(entry => entry.bytes === null || !sameStream(scope, entry.event))) return;
+    while (writing() && !this.#stopped && this.#buffered.length < MAX_BATCH_EVENTS && bytes < MAX_BATCH_BYTES) {
+      // A pending read belongs to this owner even if the write finishes first.
+      // Its eventual outcome remains available to next(); no event is discarded.
+      await this.#requestRead();
+      if (!writing() || this.#stopped) return;
+      const outcome = this.#readOutcome!;
+      if (!outcome.ok || outcome.result.done) return;
+      this.#readOutcome = undefined;
+      const entry = bufferedEvent(outcome.result.value);
+      this.#buffered.push(entry);
+      // Retain at most one nontext, different-stream, or over-budget lookahead.
+      if (entry.bytes === null || !sameStream(scope, entry.event) || bytes + entry.bytes > MAX_BATCH_BYTES) return;
+      bytes += entry.bytes;
+    }
+  }
+
+  async appendEvent(
+    event: PrpEvent,
+    options?: AppendControlPlaneEventOptions,
+  ): Promise<AppendedEventReceipt> {
+    options?.signal.throwIfAborted();
+    if (this.#stopped) throw new Error("native_event_consumer_stopped");
+    const committed = this.#receipts.get(event);
+    if (committed !== undefined) {
+      this.#receipts.delete(event);
+      return committed;
+    }
+    const first = bufferedEvent(event);
+    const batch = [event];
+    let bytes = first.bytes ?? 0;
+    if (this.#port.appendEvents !== undefined && first.bytes !== null && bytes <= MAX_BATCH_BYTES) {
+      while (this.#buffered.length > 0 && batch.length < MAX_BATCH_EVENTS) {
+        const entry = this.#buffered[0]!;
+        if (entry.bytes === null || !sameStream(event, entry.event) || bytes + entry.bytes > MAX_BATCH_BYTES) break;
+        this.#buffered.shift();
+        batch.push(entry.event);
+        bytes += entry.bytes;
+      }
+    }
+
+    // Invoke persistence before starting another iterator read. An isolated
+    // event therefore takes the same immediate singleton path as before.
+    const operation = batch.length > 1
+      ? this.#port.appendEvents!(batch, options)
+      : this.#port.appendEvent(event, options).then(receipt => [receipt]);
+    let writing = true;
+    const write = operation.finally(() => { writing = false; });
+    if (this.#port.appendEvents !== undefined && first.bytes !== null) {
+      void this.#collect(event, () => writing && options?.signal.aborted !== true).catch(error => {
+        this.#readOutcome = { ok: false, error };
+      });
+    }
+    const receipts = await write;
+    if (receipts.length !== batch.length) throw new Error("native_event_batch_receipt_count_mismatch");
+    for (let index = 1; index < batch.length; index += 1) {
+      this.#committed.push({ event: batch[index]!, receipt: receipts[index]! });
+    }
+    return receipts[0]!;
+  }
+
+  async return(): Promise<IteratorResult<PrpEvent>> {
+    this.#stopped = true;
+    this.#buffered.length = 0;
+    this.#committed.length = 0;
+    return await this.#iterator.return?.() ?? { value: undefined, done: true };
+  }
+}

packages/paperclip-runner/src/native-session-runtime.test.ts

@@ -230,6 +230,71 @@ function highestContiguous(events: PrpEvent[]): number {
 }
 
 describe("executeNativeSession recovery", () => {
+  it("commits queued text in one batch and observes each event before the result boundary", async () => {
+    const capabilities = { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true };
+    let releaseFirst!: () => void;
+    const firstReleased = new Promise<void>(resolve => { releaseFirst = resolve; });
+    let reachedBoundary!: () => void;
+    const boundaryReached = new Promise<void>(resolve => { reachedBoundary = resolve; });
+    const textEvents = [1, 2, 3].map(seq => ({
+      ...runnerEvent(seq, "item.delta", { kind: "agentMessage", text: `chunk-${seq}` }), priority: 1 as const,
+    }));
+    const source = [...textEvents, runnerEvent(4, "run.result.proposed", result), runnerEvent(5, "turn.completed")];
+    const session: NativeSession = {
+      identity: () => identity,
+      async capabilities() { return capabilities; },
+      async *events() {
+        for (const event of source) {
+          if (event.eventType === "run.result.proposed") reachedBoundary();
+          yield event;
+        }
+      },
+      async startTurn() { return { turnId: "turn-recovery" }; },
+      async result() { return { result, terminal, turnId: "turn-recovery" }; },
+      async snapshot() {
+        return { backendKind: "mock", sessionId: "driver-recovery", identity, providerSessionId: "provider-recovery",
+          cursor: null, activeTurnId: null, pendingRuntimeRequests: [], lineage: [] };
+      },
+      async close() {},
+    };
+    const backend: NativeSessionBackend = {
+      async descriptor() { return { kind: "mock", name: "batch-fixture", version: "1", capabilities }; },
+      async openSession() { return session; },
+    };
+    const committed: PrpEvent[] = [];
+    const makeReceipt = (event: PrpEvent) => ({ cursor: event.sourceSeq, highestContiguousSourceSeq: event.sourceSeq,
+      disposition: "committed" as const });
+    const appendEvents = vi.fn(async (events: readonly PrpEvent[]) => {
+      committed.push(...events);
+      return events.map(makeReceipt);
+    });
+    const port: ControlPlanePort = {
+      async openRun() {}, async checkpointSession() {},
+      async appendEvent(event) {
+        const prp = event as PrpEvent;
+        if (prp === textEvents[0]) await firstReleased;
+        committed.push(prp);
+        return makeReceipt(prp);
+      },
+      appendEvents,
+      async replayEvents() { return { events: [], highestContiguousSourceSeq: 0 }; },
+      async completeRun() {},
+    };
+    const observed: PrpEvent[] = [];
+    const execution = executeNativeSession({ input, backend, controlPlane: port,
+      runnerInstanceId: "runner-recovery", controlPlaneInstanceId: "control-recovery",
+      resolveGovernedWait({ event }) { observed.push(event); return null; } });
+    await boundaryReached;
+    expect(appendEvents).not.toHaveBeenCalled();
+    expect(observed).toEqual([]);
+    releaseFirst();
+    const completed = await execution;
+    expect(completed).toMatchObject({ result, terminal, nativeEventCount: committed.length });
+    expect(appendEvents).toHaveBeenCalledExactlyOnceWith(textEvents.slice(1), expect.objectContaining({ signal: expect.any(AbortSignal) }));
+    expect(committed.filter(event => event.sourceKind === "runner")).toEqual(source);
+    expect(observed).toEqual(textEvents);
+  });
+
   it("completes Cursor Cloud semantic planning through its default-only harness without losing the pinned revision", async () => {
     await writeFile(join(WORKSPACE, "cloud-plan-instructions.md"), "Make the plan only. Do not implement it.\n");
     const prepared = preparedInput();

packages/paperclip-runner/src/native-session-runtime.ts

@@ -1,5 +1,6 @@
 import { randomUUID } from "node:crypto";
 import { nativeSystemInstructions } from "./backends/runtime-context.js";
+import { NativeEventPersistence } from "./native-event-persistence.js";
 
 import type {
   CheckpointControlPlaneSessionOptions,
@@ -911,7 +912,7 @@ async function consumeTurn(
   let deferredSessionCancellationSettlement: Promise<unknown> | null = null;
   const inputTimers = new Map<string, ReturnType<typeof setTimeout>>();
   const handoffCleanupOperations = new Set<Promise<unknown>>();
-  const eventIterator = session.events()[Symbol.asyncIterator]();
+  const eventIterator = new NativeEventPersistence(session.events()[Symbol.asyncIterator](), controlPlane);
   let stopConsumer = false;
   let rejectHandoff: ((error: unknown) => void) | null = null;
   const handoffFailure = new Promise<never>((_resolve, reject) => {
@@ -1001,7 +1002,7 @@ async function consumeTurn(
             || (trailing.eventType === "item.completed" && trailing.payload.kind === "interrupt_acknowledgement")
             || (trailing.eventType === "run.terminal" && ["failed", "cancelled"].includes(String(trailing.payload.runTerminalState)));
           if (trailing.turnId === event.turnId && (usage || terminal || cancellation)) {
-            const receipt = await controlPlane.appendEvent(trailing, { signal: drainAbort.signal });
+            const receipt = await eventIterator.appendEvent(trailing, { signal: drainAbort.signal });
             if (drainAbort.signal.aborted) return;
             eventCount += receipt.disposition === "committed" ? 1 : 0;
             highestContiguousSourceSeq = Math.max(highestContiguousSourceSeq, receipt.highestContiguousSourceSeq);
@@ -1073,7 +1074,7 @@ async function consumeTurn(
       // the timer stayed live until the receipt returned, both settlement
       // and a durable handoff could commit for the same request.
       if (settlingRequestId !== null) clearInputTimer(settlingRequestId);
-      const receipt = await controlPlane.appendEvent(event, {
+      const receipt = await eventIterator.appendEvent(event, {
         signal: appendAbort.signal,
       });
       if (stopConsumer) throw new Error("native event consumer stopped");

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

@@ -0,0 +1,182 @@
+import { randomUUID } from "node:crypto";
+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, heartbeatRunEventCursors, heartbeatRunEvents, heartbeatRuns } from "@paperclipai/db";
+import { appendHeartbeatRunEvent, appendHeartbeatRunEvents, type AppendHeartbeatRunEventInput } from "../services/heartbeat-run-events.js";
+import { rebuildNativeEventCursorPage } from "../services/heartbeat/native-event-cursors.js";
+import { startRecoveryPostgres } from "./helpers/recovery-postgres.js";
+
+describe("bounded native text batch persistence", () => {
+  let temporary: Awaited<ReturnType<typeof startRecoveryPostgres>>;
+  let db: ReturnType<typeof createDb>, otherDb: ReturnType<typeof createDb>;
+  const companyId = randomUUID(), agentId = randomUUID();
+  beforeAll(async () => {
+    temporary = await startRecoveryPostgres("paperclip-native-event-batch-");
+    db = createDb(temporary.connectionString); otherDb = createDb(temporary.connectionString);
+    await db.insert(companies).values({ id: companyId, name: "Batch events", issuePrefix: "BCH" });
+    await db.insert(agents).values({ id: agentId, companyId, name: "Batch agent" });
+  }, 60_000);
+  afterAll(async () => {
+    await otherDb?.$client.end({ timeout: 1 });
+    await db?.$client.end({ timeout: 1 });
+    await temporary?.cleanup();
+  });
+  async function makeRun() {
+    const runId = randomUUID();
+    await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "running", runtimeMode: "native" });
+    return runId;
+  }
+  function event(runId: string, sourceSeq: number, text = `chunk-${sourceSeq}`, sourceInstanceId = "runner"): AppendHeartbeatRunEventInput {
+    const canonicalPayload = {
+      schema: "paperclip.prp.event.v1", schemaVersion: 1, sourceEventId: `${sourceInstanceId}:${sourceSeq}`,
+      sourceSeq, sourceInstanceId, sourceKind: "runner", runId, normalizedSessionId: "session",
+      turnId: "turn", itemId: "message", eventType: "item.delta", priority: 1,
+      emittedAt: "2026-11-03T00:00:00.000Z", payload: { kind: "agentMessage", text },
+    };
+    return { companyId, agentId, runId, eventType: "item.delta", stream: "system", level: "info",
+      payload: { prpEvent: canonicalPayload }, nativeSource: { sourceInstanceId,
+        sourceEventId: canonicalPayload.sourceEventId, sourceSeq, protocolSchemaVersion: 1, canonicalPayload } };
+  }
+  async function read(runId: string) {
+    return {
+      run: (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)))[0]!,
+      events: await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, runId)).orderBy(heartbeatRunEvents.seq),
+      cursors: await db.select().from(heartbeatRunEventCursors).where(eq(heartbeatRunEventCursors.runId, runId)),
+    };
+  }
+
+  it("uses one run lock, reservation, insert and contiguous interval projection", async () => {
+    const runId = await makeRun(), statements: string[] = [];
+    const recorded = drizzle(db.$client, { schema: db._.fullSchema, logger: { logQuery: query => statements.push(query) } }) as typeof db;
+    const inputs = [event(runId, 1), event(runId, 2, "before\u0000after"), event(runId, 3)];
+    const receipts = await appendHeartbeatRunEvents(recorded, inputs);
+    expect(receipts.map(value => [value.row.seq, value.disposition, value.highestContiguousSourceSeq]))
+      .toEqual([[1, "committed", 3], [2, "committed", 3], [3, "committed", 3]]);
+    expect(receipts.map(value => value.row.payload)).toEqual(inputs.map(value => value.payload));
+    expect(statements.filter(query => query.includes('from "heartbeat_runs"') && query.includes("for update"))).toHaveLength(1);
+    expect(statements.filter(query => query.startsWith('update "heartbeat_runs"'))).toHaveLength(1);
+    expect(statements.filter(query => query.startsWith('insert into "heartbeat_run_events"'))).toHaveLength(1);
+    expect(statements.filter(query => query.includes("insert into heartbeat_run_event_cursors"))).toHaveLength(1);
+    expect(statements.some(query => query.includes("count(*)"))).toBe(false);
+    const state = await read(runId);
+    expect(state.events).toEqual(receipts.map(value => value.row));
+    expect(state.run).toMatchObject({ nextEventSeq: 4, eventCursorProjectionSeq: 3, eventCursorRebuildPending: false });
+  });
+
+  it("returns input-order receipts for stored and in-batch duplicates without allocating for them", async () => {
+    const runId = await makeRun();
+    const [one, two] = await appendHeartbeatRunEvents(db, [event(runId, 1), event(runId, 2)]);
+    const receipts = await appendHeartbeatRunEvents(db, [event(runId, 2), event(runId, 3), event(runId, 3), event(runId, 1), event(runId, 4)]);
+    expect(receipts.map(value => [value.row.seq, value.disposition, value.highestContiguousSourceSeq]))
+      .toEqual([[2, "duplicate", 4], [3, "committed", 4], [3, "duplicate", 4], [1, "duplicate", 4], [4, "committed", 4]]);
+    expect(receipts[0]!.row).toEqual(two!.row); expect(receipts[3]!.row).toEqual(one!.row);
+    expect((await read(runId)).run.nextEventSeq).toBe(5);
+  });
+
+  it.each(["changed payload", "event id", "sequence"])("rejects an in-batch %s conflict without any writes", async conflict => {
+    const runId = await makeRun(), first = event(runId, 1), second = event(runId, 1, conflict === "changed payload" ? "changed" : "chunk-1");
+    if (conflict === "event id") {
+      second.nativeSource!.sourceEventId = "other-id";
+      second.nativeSource!.canonicalPayload.sourceEventId = "other-id";
+    }
+    if (conflict === "sequence") {
+      second.nativeSource!.sourceSeq = 2;
+      second.nativeSource!.canonicalPayload.sourceSeq = 2;
+    }
+    await expect(appendHeartbeatRunEvents(db, [first, second])).rejects.toThrow("native_event_replay_conflict");
+    expect(await read(runId)).toMatchObject({ run: { nextEventSeq: 1, eventCursorProjectionSeq: 0 }, events: [], cursors: [] });
+  });
+
+  it("checks both stored replay identities and rolls back the fresh batch prefix", async () => {
+    const runId = await makeRun();
+    await appendHeartbeatRunEvents(db, [event(runId, 1), event(runId, 2)]);
+    const bridge = event(runId, 2);
+    bridge.nativeSource!.sourceEventId = "runner:1";
+    bridge.nativeSource!.canonicalPayload.sourceEventId = "runner:1";
+    const before = await read(runId);
+    await expect(appendHeartbeatRunEvents(db, [event(runId, 3), bridge])).rejects.toThrow("native_event_replay_conflict");
+    expect(await read(runId)).toEqual(before);
+  });
+
+  it("shares sequence allocation with concurrent singleton and legacy writers on separate connections", async () => {
+    const runId = await makeRun();
+    await Promise.all(Array.from({ length: 10 }, (_, index) => index % 2
+      ? appendHeartbeatRunEvent(otherDb, { companyId, agentId, runId, eventType: "log", message: `legacy-${index}` })
+      : appendHeartbeatRunEvents(db, Array.from({ length: 4 }, (_, offset) => event(runId, index / 2 * 4 + offset + 1)))));
+    await appendHeartbeatRunEvent(otherDb, event(runId, 21));
+    const state = await read(runId);
+    expect(state.events.map(row => row.seq)).toEqual(Array.from({ length: 26 }, (_, index) => index + 1));
+    expect(state.run).toMatchObject({ nextEventSeq: 27, eventCursorProjectionSeq: 26 });
+    expect((await appendHeartbeatRunEvents(db, [event(runId, 1), event(runId, 21)])).map(value => value.highestContiguousSourceSeq))
+      .toEqual([21, 21]);
+  });
+
+  it("keeps independent source gaps and merges a late bridging batch", async () => {
+    const runId = await makeRun();
+    const initial = await appendHeartbeatRunEvents(db, [event(runId, 2), event(runId, 5), event(runId, 6), event(runId, 1, "other", "other-source")]);
+    expect(initial.map(value => value.highestContiguousSourceSeq)).toEqual([0, 0, 0, 1]);
+    const filled = await appendHeartbeatRunEvents(db, [event(runId, 1), event(runId, 4), event(runId, 3)]);
+    expect(filled.map(value => value.highestContiguousSourceSeq)).toEqual([6, 6, 6]);
+    expect((await read(runId)).cursors.map(row => [row.sourceInstanceId, row.contiguousSourceSeq, row.pendingRangeCount]).sort())
+      .toEqual([["other-source", 1, 0], ["runner", 6, 0]]);
+  });
+
+  it("uses the correct fallback for old writes and the ready cursor after rebuild", async () => {
+    const runId = await makeRun();
+    const old = event(runId, 1);
+    await db.insert(heartbeatRunEvents).values({ companyId, agentId, runId, seq: 1, eventType: old.eventType,
+      sourceInstanceId: "runner", sourceSeq: 1, sourceEventId: "old-writer:1" });
+    await db.update(heartbeatRuns).set({ nextEventSeq: 2 }).where(eq(heartbeatRuns.id, runId));
+    expect((await appendHeartbeatRunEvents(db, [event(runId, 2), event(runId, 3)])).map(value => value.highestContiguousSourceSeq)).toEqual([3, 3]);
+    expect((await read(runId)).run).toMatchObject({ eventCursorProjectionSeq: 0, eventCursorRebuildPending: true });
+    expect(await rebuildNativeEventCursorPage(db, { companyId, runId })).toMatchObject({ pending: false });
+    const statements: string[] = [];
+    const recorded = drizzle(db.$client, { schema: db._.fullSchema, logger: { logQuery: query => statements.push(query) } }) as typeof db;
+    expect((await appendHeartbeatRunEvents(recorded, [event(runId, 4), event(runId, 5)])).map(value => value.highestContiguousSourceSeq)).toEqual([5, 5]);
+    expect(statements.some(query => query.includes("count(*)"))).toBe(false);
+  });
+
+  it("rolls back rows, source cursors and reservation with its outer transaction", async () => {
+    const runId = await makeRun();
+    await expect(db.transaction(async tx => {
+      await appendHeartbeatRunEvents(tx as unknown as typeof db, [event(runId, 1), event(runId, 2)]);
+      throw new Error("outer rollback");
+    })).rejects.toThrow("outer rollback");
+    expect(await read(runId)).toMatchObject({ run: { nextEventSeq: 1, eventCursorProjectionSeq: 0 }, events: [], cursors: [] });
+  });
+
+  it("rolls back the whole reservation when an insertion fails", async () => {
+    const runId = await makeRun();
+    await db.execute(sql`create function reject_native_batch_fixture() returns trigger language plpgsql as $$
+      begin if new.source_event_id = 'reject-batch-fixture' then raise exception 'batch insert rejected'; end if; return new; end $$`);
+    await db.execute(sql`create trigger reject_native_batch_fixture before insert on heartbeat_run_events
+      for each row execute function reject_native_batch_fixture()`);
+    try {
+      const bad = event(runId, 2);
+      bad.nativeSource!.sourceEventId = "reject-batch-fixture";
+      bad.nativeSource!.canonicalPayload.sourceEventId = "reject-batch-fixture";
+      await expect(appendHeartbeatRunEvents(db, [event(runId, 1), bad])).rejects.toThrow("batch insert rejected");
+      expect(await read(runId)).toMatchObject({ run: { nextEventSeq: 1, eventCursorProjectionSeq: 0 }, events: [], cursors: [] });
+    } finally {
+      await db.execute(sql`drop trigger reject_native_batch_fixture on heartbeat_run_events`);
+      await db.execute(sql`drop function reject_native_batch_fixture()`);
+    }
+  });
+
+  it("rejects control, mixed binding, oversized and over-count batches", async () => {
+    const runId = await makeRun();
+    const control = event(runId, 1);
+    control.nativeSource!.canonicalPayload.sourceKind = "control_plane";
+    (control.nativeSource!.canonicalPayload.payload as Record<string, unknown>).evidenceOutcome = "accepted";
+    const invalid = [
+      [{ ...event(runId, 1), eventType: "turn.completed" }],
+      [control],
+      [event(runId, 1), event(randomUUID(), 2)],
+      [event(runId, 1, "x".repeat(64 * 1024))],
+      Array.from({ length: 65 }, (_, index) => event(runId, index + 1)),
+    ];
+    for (const batch of invalid) await expect(appendHeartbeatRunEvents(db, batch)).rejects.toThrow("native_event_batch_invalid");
+    expect((await read(runId)).events).toEqual([]);
+  });
+});

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

@@ -6,7 +6,7 @@ import { isNativeResponseEvidence, notifyNativeResponsePresentation } from "./na
 import { isSafeNativeChatProgressEventType } from "./safe-native-chat-progress.js";
 import { notifyChatNativeProgress } from "./chat-native-progress-notifications.js";
 import { boundRunIssueIdSql } from "./run-task-binding.js";
-import { and, eq, getTableColumns, or, sql } from "drizzle-orm";
+import { and, eq, getTableColumns, inArray, or, sql } from "drizzle-orm";
 import type { Db } from "@paperclipai/db";
 import { heartbeatRunEvents, heartbeatRuns } from "@paperclipai/db";
 import { nativeSha256 } from "./native-runtime/canonical.js";
@@ -258,3 +258,162 @@ export async function appendHeartbeatRunEvent(
     };
   });
 }
+
+/**
+ * Ordinary display text keeps its individual replay identities, but shares one
+ * run lock, sequence reservation and insert. Semantic/control events retain
+ * their singleton append and notification boundaries.
+ */
+export async function appendHeartbeatRunEvents(
+  db: Db,
+  values: readonly AppendHeartbeatRunEventInput[],
+): Promise<AppendHeartbeatRunEventResult[]> {
+  if (values.length === 0) return [];
+  if (values.length > 64) throw new Error("native_event_batch_invalid");
+  // Clone and hash before taking the run lock. A caller cannot change the
+  // payload behind the replay check while the transaction is waiting.
+  const inputs = values.map(value => structuredClone(value));
+  const binding = inputs[0]!;
+  let bytes = 0;
+  type Candidate = {
+    input: AppendHeartbeatRunEventInput & { nativeSource: NonNullable<AppendHeartbeatRunEventInput["nativeSource"]> };
+    hash: string;
+    firstIndex: number;
+    row?: typeof heartbeatRunEvents.$inferSelect;
+    fresh?: boolean;
+  };
+  const byEventId = new Map<string, Candidate>();
+  const bySourceSeq = new Map<string, Map<number, Candidate>>();
+  const candidates: Candidate[] = [];
+  const entries: Candidate[] = [];
+  const matches = (candidate: Candidate, source: NonNullable<AppendHeartbeatRunEventInput["nativeSource"]>, hash: string) =>
+    candidate.hash === hash && candidate.input.nativeSource.sourceEventId === source.sourceEventId
+    && candidate.input.nativeSource.sourceInstanceId === source.sourceInstanceId
+    && candidate.input.nativeSource.sourceSeq === source.sourceSeq;
+  for (let index = 0; index < inputs.length; index += 1) {
+    const input = inputs[index]!;
+    const source = input.nativeSource;
+    const event = source?.canonicalPayload;
+    const payload = event?.payload as Record<string, unknown> | undefined;
+    if (input.companyId !== binding.companyId || input.runId !== binding.runId || input.agentId !== binding.agentId
+      || input.eventType !== "item.delta" || input.retryExhaustion !== undefined || source === undefined
+      || !Number.isSafeInteger(source.sourceSeq) || source.sourceSeq < 1
+      || event?.eventType !== "item.delta" || event.sourceKind !== "runner" || (event.priority !== 1 && event.priority !== 2)
+      || event.sourceEventId !== source.sourceEventId || event.sourceInstanceId !== source.sourceInstanceId
+      || event.sourceSeq !== source.sourceSeq || event.runId !== input.runId
+      || (payload?.kind !== "agentMessage" && payload?.kind !== "reasoning") || typeof payload?.text !== "string") {
+      throw new Error("native_event_batch_invalid");
+    }
+    bytes += Buffer.byteLength(JSON.stringify(event), "utf8");
+    if (bytes > 64 * 1024) throw new Error("native_event_batch_invalid");
+    const hash = nativeSha256(event);
+    const eventCandidate = byEventId.get(source.sourceEventId);
+    const sourceCandidates = bySourceSeq.get(source.sourceInstanceId) ?? new Map<number, Candidate>();
+    const sequenceCandidate = sourceCandidates.get(source.sourceSeq);
+    if (eventCandidate && !matches(eventCandidate, source, hash)
+      || sequenceCandidate && !matches(sequenceCandidate, source, hash)) {
+      throw new HeartbeatRunEventConflictError();
+    }
+    const candidate = eventCandidate ?? sequenceCandidate ?? {
+      input: input as Candidate["input"], hash, firstIndex: index,
+    };
+    if (!eventCandidate && !sequenceCandidate) candidates.push(candidate);
+    byEventId.set(source.sourceEventId, candidate);
+    sourceCandidates.set(source.sourceSeq, candidate);
+    bySourceSeq.set(source.sourceInstanceId, sourceCandidates);
+    entries.push(candidate);
+  }
+
+  return db.transaction(async tx => {
+    const [run] = await tx.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId,
+      agentId: heartbeatRuns.agentId, nextEventSeq: heartbeatRuns.nextEventSeq,
+      eventCursorProjectionSeq: heartbeatRuns.eventCursorProjectionSeq,
+      eventCursorRebuildPending: heartbeatRuns.eventCursorRebuildPending,
+    }).from(heartbeatRuns).where(eq(heartbeatRuns.id, binding.runId)).for("update").limit(1);
+    if (!run || run.companyId !== binding.companyId || run.agentId !== binding.agentId) {
+      throw new Error("heartbeat_run_event_binding_mismatch");
+    }
+    const existing = await tx.select().from(heartbeatRunEvents).where(and(
+      eq(heartbeatRunEvents.runId, run.id),
+      or(inArray(heartbeatRunEvents.sourceEventId, [...byEventId.keys()]),
+        ...[...bySourceSeq].map(([sourceInstanceId, sequences]) => and(
+          eq(heartbeatRunEvents.sourceInstanceId, sourceInstanceId),
+          inArray(heartbeatRunEvents.sourceSeq, [...sequences.keys()]),
+        ))),
+    ));
+    for (const row of existing) {
+      const identified = new Set([
+        row.sourceEventId === null ? undefined : byEventId.get(row.sourceEventId),
+        row.sourceInstanceId === null || row.sourceSeq === null ? undefined : bySourceSeq.get(row.sourceInstanceId)?.get(row.sourceSeq),
+      ]);
+      for (const candidate of identified) {
+        if (candidate === undefined) continue;
+        const source = candidate.input.nativeSource;
+        if (row.companyId !== binding.companyId || row.agentId !== binding.agentId
+          || row.sourcePayloadSha256 !== candidate.hash || row.sourceEventId !== source.sourceEventId
+          || row.sourceInstanceId !== source.sourceInstanceId || row.sourceSeq !== source.sourceSeq
+          || candidate.row !== undefined && candidate.row.id !== row.id) {
+          throw new HeartbeatRunEventConflictError();
+        }
+        candidate.row = row;
+      }
+    }
+    const projectionCurrent = nativeEventCursorProjectionCurrent(run);
+    if (!projectionCurrent) await requestNativeEventCursorRebuild(tx as unknown as Db, run);
+    const added = candidates.filter(candidate => candidate.row === undefined);
+    const cursors = new Map<string, number>();
+    if (added.length > 0) {
+      await tx.update(heartbeatRuns).set({
+        nextEventSeq: run.nextEventSeq + added.length,
+        ...(projectionCurrent ? { eventCursorProjectionSeq: run.nextEventSeq + added.length - 1 } : {}),
+        updatedAt: new Date(),
+      }).where(eq(heartbeatRuns.id, run.id));
+      const inserted = await tx.insert(heartbeatRunEvents).values(added.map((candidate, index) => {
+        const { input } = candidate;
+        return { companyId: input.companyId, runId: input.runId, agentId: input.agentId,
+          seq: run.nextEventSeq + index, eventType: input.eventType,
+          stream: input.stream ?? null, level: input.level ?? null, color: input.color ?? null,
+          message: input.message ?? null, payload: input.payload ?? null,
+          sourceInstanceId: input.nativeSource.sourceInstanceId, sourceEventId: input.nativeSource.sourceEventId,
+          sourceSeq: input.nativeSource.sourceSeq, sourcePayloadSha256: candidate.hash,
+          protocolSchemaVersion: input.nativeSource.protocolSchemaVersion,
+        };
+      })).returning();
+      for (const row of inserted) {
+        const candidate = byEventId.get(row.sourceEventId!)!;
+        candidate.row = row;
+        candidate.fresh = true;
+      }
+      if (inserted.length !== added.length) throw new Error("heartbeat_run_event_not_persisted");
+
+      if (projectionCurrent) {
+        const ranges = new Map<string, number[]>();
+        for (const candidate of added) {
+          const source = candidate.input.nativeSource;
+          const values = ranges.get(source.sourceInstanceId) ?? [];
+          values.push(source.sourceSeq);
+          ranges.set(source.sourceInstanceId, values);
+        }
+        for (const [sourceInstanceId, values] of ranges) {
+          values.sort((a, b) => a - b);
+          const scope = { companyId: run.companyId, runId: run.id, sourceInstanceId };
+          let start = values[0]!, end = start;
+          for (const value of values.slice(1)) {
+            if (value === end + 1) { end = value; continue; }
+            await projectNativeEventSourceRange(tx as unknown as Db, scope, start, end);
+            start = end = value;
+          }
+          cursors.set(sourceInstanceId, await projectNativeEventSourceRange(tx as unknown as Db, scope, start, end));
+        }
+      }
+    }
+    for (const sourceInstanceId of bySourceSeq.keys()) if (!cursors.has(sourceInstanceId)) {
+      cursors.set(sourceInstanceId, await (projectionCurrent ? readNativeEventSourceCursor : readRetainedNativeEventCursor)(
+        tx as unknown as Db, { companyId: run.companyId, runId: run.id, sourceInstanceId }));
+    }
+    return entries.map((candidate, index) => ({ row: candidate.row!,
+      disposition: candidate.fresh && candidate.firstIndex === index ? "committed" as const : "duplicate" as const,
+      highestContiguousSourceSeq: cursors.get(candidate.input.nativeSource.sourceInstanceId)!,
+    }));
+  });
+}

server/src/services/native-runtime/paperclip-control-plane-port.test.ts

@@ -404,8 +404,8 @@ describe("PaperclipControlPlanePort conformance", () => {
   });
 
   it.each([
-    ["checkpoint", "detach"], ["result", "detach"], ["event", "detach"],
-    ["checkpoint", "abort"], ["result", "abort"], ["event", "abort"],
+    ["checkpoint", "detach"], ["result", "detach"], ["event", "detach"], ["batch", "detach"],
+    ["checkpoint", "abort"], ["result", "abort"], ["event", "abort"], ["batch", "abort"],
   ] as const)("revokes an in-flight %s on %s", async (operation, reason) => {
     const identity = { ...CONTROL_PLANE_CONFORMANCE_OPEN.identity, runId: randomUUID(), sessionId: randomUUID() };
     const runnerId = randomUUID();
@@ -424,12 +424,16 @@ describe("PaperclipControlPlanePort conformance", () => {
       assertControllerActive: () => { if (detached) throw new Error("controller_detached"); },
     });
     await port.openRun({ identity, backendKind: "mock", sourceInstanceId: runnerId });
-    const mutate = () => operation === "event" ? port.appendEvent({
+    const nativeEvent: PrpEvent = {
       schema: "paperclip.prp.event.v1", sourceEventId: `${runnerId}:1`, sourceSeq: 1,
       sourceInstanceId: runnerId, sourceKind: "runner", runId: identity.runId,
       normalizedSessionId: identity.sessionId, eventType: "turn.interrupted", schemaVersion: 1,
       priority: 0, emittedAt: new Date().toISOString(), payload: { reason: "governed_wait" },
-    }, { signal: controller.signal }) : operation === "checkpoint"
+    };
+    const mutate = () => operation === "batch" ? port.appendEvents([1, 2].map(sourceSeq => ({
+      ...nativeEvent, sourceEventId: `${runnerId}:${sourceSeq}`, sourceSeq,
+      eventType: "item.delta", priority: 1, payload: { kind: "agentMessage", text: `chunk-${sourceSeq}` },
+    })), { signal: controller.signal }) : operation === "event" ? port.appendEvent(nativeEvent, { signal: controller.signal }) : operation === "checkpoint"
       ? port.checkpointSession({ backendKind: "mock", sessionId: identity.sessionId, identity }, { signal: controller.signal })
       : port.completeRun({ result: CONTROL_PLANE_CONFORMANCE_RESULT, terminal: CONTROL_PLANE_CONFORMANCE_TERMINAL }, { signal: controller.signal });
     let pending: Promise<unknown> | undefined;
@@ -464,6 +468,114 @@ describe("PaperclipControlPlanePort conformance", () => {
       terminal: CONTROL_PLANE_CONFORMANCE_TERMINAL })).resolves.toBeUndefined();
   });
 
+  it("commits text batches before ordered callbacks and preserves redaction and duplicate receipts", async () => {
+    const identity = { ...CONTROL_PLANE_CONFORMANCE_OPEN.identity, runId: randomUUID(), sessionId: randomUUID() };
+    const runnerId = randomUUID();
+    const privateKeyPem = generateKeyPairSync("ed25519").privateKey.export({ type: "pkcs8", format: "pem" }).toString();
+    const keyBody = privateKeyPem.split("\n")[1]!;
+    await db.insert(heartbeatRuns).values({ id: identity.runId, companyId: identity.companyId, agentId: identity.agentId,
+      status: "running", runtimeMode: "native", nativeIssueId: identity.issueId, nativeSessionId: identity.sessionId,
+      runnerInstanceId: runnerId, completionContractId: contractId, completionContractSha256: contractSha });
+    const observed: PrpEvent[] = [], duplicates: string[] = [], commitSizes: number[] = [];
+    const port = new PaperclipControlPlanePort(db, { ...identity, completionContractId: contractId,
+      completionContractSha256: contractSha, sourceInstanceId: runnerId, controlPlaneSourceInstanceId: `batch-${identity.runId}` }, {
+      privateKeyPem,
+      onCommittedEvent: async event => {
+        const rows = await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, identity.runId));
+        commitSizes.push(rows.length);
+        observed.push(event);
+      },
+      onDuplicateEvent: async event => { duplicates.push(event.sourceEventId); },
+    });
+    await port.openRun({ identity, backendKind: "mock", sourceInstanceId: runnerId });
+    const events: PrpEvent[] = [keyBody.slice(0, 31), keyBody.slice(31), "after\u0000text"].map((text, index) => ({
+      schema: "paperclip.prp.event.v1", sourceEventId: `${runnerId}:${index + 1}`, sourceSeq: index + 1,
+      sourceInstanceId: runnerId, sourceKind: "runner", runId: identity.runId, normalizedSessionId: identity.sessionId,
+      turnId: "batch-turn", itemId: "batch-item", eventType: "item.delta", schemaVersion: 1, priority: 1,
+      emittedAt: "2026-11-03T00:00:00.000Z", payload: { kind: "agentMessage", text },
+    }));
+    const notifications: string[] = [];
+    const unsubscribe = subscribeCompanyLiveEvents(identity.companyId, event => {
+      if (event.type === "heartbeat.run.event" && event.payload.runId === identity.runId) notifications.push(String(event.payload.seq));
+    });
+    try {
+      expect(await port.appendEvents(events)).toEqual([1, 2, 3].map(cursor => ({ cursor,
+        highestContiguousSourceSeq: 3, disposition: "committed" })));
+      expect(await port.appendEvents(events)).toEqual([1, 2, 3].map(cursor => ({ cursor,
+        highestContiguousSourceSeq: 3, disposition: "duplicate" })));
+    } finally { unsubscribe(); }
+    expect(commitSizes).toEqual([3, 3, 3]);
+    expect(observed.map(event => event.sourceEventId)).toEqual(events.map(event => event.sourceEventId));
+    expect(duplicates).toEqual(events.map(event => event.sourceEventId));
+    expect(notifications).toEqual(["1", "2", "3"]);
+    const saved = await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, identity.runId)).orderBy(heartbeatRunEvents.seq);
+    expect(saved.map(row => row.payload?.prpEvent)).toEqual(observed);
+    expect(observed.map(event => event.payload.text).join("")).toContain("***REDACTED***");
+    expect(JSON.stringify(saved)).not.toContain(keyBody);
+    expect(JSON.stringify(saved)).toContain("after\\u0000text");
+    await expect(port.appendEvents([{ ...events[2]!, payload: { kind: "agentMessage", text: "changed" } }])).rejects.toThrow();
+    expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, identity.runId)))[0]!.nextEventSeq).toBe(4);
+    const control: PrpEvent = { ...events[0]!, sourceKind: "control_plane", sourceInstanceId: `batch-${identity.runId}`,
+      sourceEventId: `batch-${identity.runId}:1`, payload: { kind: "agentMessage", text: "accepted evidence", evidenceOutcome: "accepted" } };
+    await expect(port.appendEvents([control])).rejects.toThrow("native_event_batch_invalid");
+    // A control-plane evidence event retains the singleton notification path.
+    await expect(port.appendEvent(control)).resolves.toEqual({ cursor: 4, highestContiguousSourceSeq: 1, disposition: "committed" });
+  });
+
+  it.each([false, true])("handles redaction expansion of an exact-size text batch atomically (later conflict=%s)", async conflict => {
+    const identity = { ...CONTROL_PLANE_CONFORMANCE_OPEN.identity, runId: randomUUID(), sessionId: randomUUID() };
+    const runnerId = randomUUID();
+    const privateKeyPem = generateKeyPairSync("ed25519").privateKey.export({ type: "pkcs8", format: "pem" }).toString();
+    const prefix = privateKeyPem.split("\n")[1]![0]!;
+    expect(prefix).toBe("M");
+    await db.insert(heartbeatRuns).values({ id: identity.runId, companyId: identity.companyId, agentId: identity.agentId,
+      status: "running", runtimeMode: "native", nativeIssueId: identity.issueId, nativeSessionId: identity.sessionId,
+      runnerInstanceId: runnerId, completionContractId: contractId, completionContractSha256: contractSha });
+    const binding = { ...identity, completionContractId: contractId, completionContractSha256: contractSha,
+      sourceInstanceId: runnerId, controlPlaneSourceInstanceId: `batch-size-${identity.runId}` };
+    const observed: PrpEvent[] = [], callbackRowCounts: number[] = [];
+    const port = new PaperclipControlPlanePort(db, binding, { privateKeyPem,
+      onCommittedEvent: async event => {
+        observed.push(event);
+        callbackRowCounts.push((await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, identity.runId))).length);
+      } });
+    await port.openRun({ identity, backendKind: "mock", sourceInstanceId: runnerId });
+    const textEvent = (sourceSeq: number, text: string): PrpEvent => ({
+      schema: "paperclip.prp.event.v1", schemaVersion: 1, sourceEventId: `${runnerId}:${sourceSeq}`, sourceSeq,
+      sourceInstanceId: runnerId, sourceKind: "runner", runId: identity.runId, normalizedSessionId: identity.sessionId,
+      turnId: "sized-turn", itemId: "sized-item", eventType: "item.delta", priority: 1,
+      emittedAt: "2026-11-03T00:00:00.000Z", payload: { kind: "agentMessage", text },
+    });
+    await port.appendEvent(textEvent(1, prefix)); // The redactor holds this possible secret prefix.
+    const batch = [textEvent(2, ""), textEvent(3, "")];
+    const overhead = batch.reduce((bytes, event) => bytes + Buffer.byteLength(JSON.stringify(event)), 0);
+    const padding = 64 * 1024 - overhead;
+    batch[0]!.payload.text = "x".repeat(Math.floor(padding / 2));
+    batch[1]!.payload.text = "x".repeat(padding - Math.floor(padding / 2));
+    expect(batch.reduce((bytes, event) => bytes + Buffer.byteLength(JSON.stringify(event)), 0)).toBe(64 * 1024);
+    if (conflict) {
+      const older = new PaperclipControlPlanePort(db, binding);
+      await older.appendEvent({ ...batch[1]!, payload: { ...batch[1]!.payload, text: "different stored payload" } });
+    }
+    const beforeRows = await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, identity.runId)).orderBy(heartbeatRunEvents.seq);
+    const [beforeRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, identity.runId));
+    observed.length = 0; callbackRowCounts.length = 0;
+    if (conflict) {
+      await expect(port.appendEvents(batch)).rejects.toThrow("native_event_replay_conflict");
+      expect(await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, identity.runId)).orderBy(heartbeatRunEvents.seq)).toEqual(beforeRows);
+      const [afterRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, identity.runId));
+      expect(afterRun.nextEventSeq).toBe(beforeRun.nextEventSeq);
+      expect(afterRun.eventCursorProjectionSeq).toBe(beforeRun.eventCursorProjectionSeq);
+      expect(observed).toEqual([]);
+    } else {
+      expect((await port.appendEvents(batch)).map(receipt => receipt.disposition)).toEqual(["committed", "committed"]);
+      expect(observed.map(event => event.sourceSeq)).toEqual([2, 3]);
+      expect(callbackRowCounts).toEqual([3, 3]);
+      expect(observed[0]!.payload.text).toBe(prefix + batch[0]!.payload.text);
+      expect(observed.reduce((bytes, event) => bytes + Buffer.byteLength(JSON.stringify(event)), 0)).toBe(64 * 1024 + 1);
+    }
+  });
+
   it("persists panel activation under the authenticated run and recovers it across source replay", async () => {
     await db.insert(plugins).values({ pluginKey: "example.live-view", packageName: "example-live-view", version: "1.0.0", status: "ready",
       manifestJson: { ui: { slots: [{ type: "detailTab", id: "screen", entityTypes: ["agent"], taskSidePanel: { availabilityPath: "/agents/:entityId/ready", activation: { namespace: "example.runtime", activityKey: "desktop.used" } } }] } } as never });

server/src/services/native-runtime/paperclip-control-plane-port.ts

@@ -27,7 +27,7 @@ import {
   validatePrpEvent,
   validatePrpStructuredRunResult,
 } from "../../vendor/paperclip-runner/index.js";
-import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js";
+import { appendHeartbeatRunEvent, appendHeartbeatRunEvents, type AppendHeartbeatRunEventResult } from "../heartbeat-run-events.js";
 import { publishLiveEvent } from "../live-events.js";
 import { nativeSha256 } from "./canonical.js";
 import { redactRunnerApiEvent, redactRunnerApiPayloads } from "./runner-api-log-redaction.js";
@@ -324,9 +324,7 @@ export class PaperclipControlPlanePort implements ControlPlanePort {
     });
   }
 
-  async appendEvent(value: NativeRunEvent | PrpEvent, options?: { signal: AbortSignal }) {
-    const persistenceStartedAt = Date.now(), persistenceClock = performance.now();
-    this.#assertActive(options);
+  #prepareEvent(value: NativeRunEvent | PrpEvent): PrpEvent {
     if (!isPrpEvent(value)) throw new Error("native_legacy_event_not_supported");
     const validated = validatePrpEvent(value);
     if (!validated.ok) throw new Error(`native_event_schema_invalid:${validated.issues[0]?.message ?? "unknown"}`);
@@ -371,10 +369,33 @@ export class PaperclipControlPlanePort implements ControlPlanePort {
         }
       }
     }
-    event = redactRunnerApiEvent(event);
-    const persisted = await this.#db.transaction(async tx => {
+    return redactRunnerApiEvent(event);
+  }
+
+  async appendEvent(value: NativeRunEvent | PrpEvent, options?: { signal: AbortSignal }) {
+    return (await this.#appendEventBatch([value], options))[0]!;
+  }
+
+  async appendEvents(values: readonly PrpEvent[], options?: { signal: AbortSignal }) {
+    if (values.length === 0) return [];
+    if (values.length > 64 || values.reduce((bytes, event) => bytes + Buffer.byteLength(JSON.stringify(event)), 0) > 64 * 1024
+      || values.some(event => event.sourceKind !== "runner" || event.eventType !== "item.delta" || event.priority === 0
+        || !["agentMessage", "reasoning"].includes(String(event.payload.kind))
+        || typeof event.payload.text !== "string")) {
+      throw new Error("native_event_batch_invalid");
+    }
+    return this.#appendEventBatch(values, options);
+  }
+
+  async #appendEventBatch(values: readonly (NativeRunEvent | PrpEvent)[], options?: { signal: AbortSignal }) {
+    const persistenceStartedAt = Date.now(), persistenceClock = performance.now();
+    this.#assertActive(options);
+    // Redaction is stateful across deltas. Prepare strictly in source order,
+    // retaining the same replay window and lossless envelope as singleton calls.
+    const events = values.map(value => this.#prepareEvent(value));
+    const receipts = await this.#db.transaction(async tx => {
       this.#assertActive(options);
-      const receipt = await appendHeartbeatRunEvent(tx as unknown as Db, {
+      const inputs = events.map(event => ({
         companyId: this.#binding.companyId,
         runId: this.#binding.runId,
         agentId: this.#binding.agentId,
@@ -389,14 +410,47 @@ export class PaperclipControlPlanePort implements ControlPlanePort {
           protocolSchemaVersion: event.schemaVersion,
           canonicalPayload: event as unknown as Record<string, unknown>,
         },
-      });
-      // The nested append has no publication side effects. Revocation rolls
-      // back its event and sequence allocation before the outer commit.
+      }));
+      // Redaction can release a held prefix or add digest metadata. Bound
+      // prepared envelopes too, without rejecting valid raw-sized batches.
+      // Every subgroup remains inside this same outer transaction.
+      const appended: AppendHeartbeatRunEventResult[] = [];
+      let pending: typeof inputs = [], pendingBytes = 0;
+      const flush = async () => {
+        if (pending.length === 0) return;
+        appended.push(...(pending.length === 1
+          ? [await appendHeartbeatRunEvent(tx as unknown as Db, pending[0]!)]
+          : await appendHeartbeatRunEvents(tx as unknown as Db, pending)));
+        pending = []; pendingBytes = 0;
+      };
+      for (let index = 0; index < inputs.length; index += 1) {
+        const bytes = Buffer.byteLength(JSON.stringify(events[index]!));
+        if (pending.length > 0 && pendingBytes + bytes > 64 * 1024) await flush();
+        pending.push(inputs[index]!); pendingBytes += bytes;
+        if (bytes > 64 * 1024) await flush();
+      }
+      await flush();
+      // Revocation before the outer commit rolls back the entire batch,
+      // including sequence reservations and transaction-aware notifications.
       this.#assertActive(options);
-      return receipt;
+      return appended;
     });
+    const persistenceEndedAt = persistenceStartedAt + performance.now() - persistenceClock;
     this.#assertActive(options);
-    try { await this.#onEventPersistence?.(event, persistenceStartedAt, persistenceStartedAt + performance.now() - persistenceClock); } catch {}
+    for (let index = 0; index < events.length; index += 1) {
+      this.#assertActive(options);
+      await this.#publishEventReceipt(events[index]!, receipts[index]!, persistenceStartedAt, persistenceEndedAt);
+    }
+    return receipts.map(persisted => ({
+      cursor: persisted.row.seq,
+      highestContiguousSourceSeq: persisted.highestContiguousSourceSeq,
+      disposition: persisted.disposition,
+    }));
+  }
+
+  async #publishEventReceipt(event: PrpEvent, persisted: AppendHeartbeatRunEventResult,
+    persistenceStartedAt: number, persistenceEndedAt: number): Promise<void> {
+    try { await this.#onEventPersistence?.(event, persistenceStartedAt, persistenceEndedAt); } catch {}
     if (persisted.disposition === "committed") {
       try {
         publishLiveEvent({
@@ -423,11 +477,6 @@ export class PaperclipControlPlanePort implements ControlPlanePort {
       // returns to its synchronous governed-wait boundary.
       await this.#onDuplicateEvent?.(event, persisted.row.seq);
     }
-    return {
-      cursor: persisted.row.seq,
-      highestContiguousSourceSeq: persisted.highestContiguousSourceSeq,
-      disposition: persisted.disposition,
-    };
   }
 
   async replayEvents(input: ReplayControlPlaneEventsInput) {