From 0aaa1c368465337944e2b06e3b86cc9ee25ce0cb Mon Sep 17 00:00:00 2001 From: Dotta Date: Tue, 3 Nov 2026 12:00:09 -0600 Subject: [PATCH] perf(runner): batch queued native text without buffering delay Co-Authored-By: Paperclip --- .../src/contracts/control-plane-port.ts | 18 ++ .../src/native-event-persistence.test.ts | 296 ++++++++++++++++++ .../src/native-event-persistence.ts | 162 ++++++++++ .../src/native-session-runtime.test.ts | 65 ++++ .../src/native-session-runtime.ts | 7 +- .../heartbeat-run-event-batch.test.ts | 182 +++++++++++ server/src/services/heartbeat-run-events.ts | 161 +++++++++- .../paperclip-control-plane-port.test.ts | 120 ++++++- .../paperclip-control-plane-port.ts | 83 ++++- 9 files changed, 1069 insertions(+), 25 deletions(-) create mode 100644 packages/paperclip-runner/src/native-event-persistence.test.ts create mode 100644 packages/paperclip-runner/src/native-event-persistence.ts create mode 100644 server/src/__tests__/heartbeat-run-event-batch.test.ts diff --git a/packages/paperclip-runner/src/contracts/control-plane-port.ts b/packages/paperclip-runner/src/contracts/control-plane-port.ts index b219c207b3a..631a2fe13c3 100644 --- a/packages/paperclip-runner/src/contracts/control-plane-port.ts +++ b/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; + /** + * 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; replayEvents( input: ReplayControlPlaneEventsInput, options?: FinalizeControlPlaneOperationOptions, diff --git a/packages/paperclip-runner/src/native-event-persistence.test.ts b/packages/paperclip-runner/src/native-event-persistence.test.ts new file mode 100644 index 00000000000..dd1c34f6ebc --- /dev/null +++ b/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() { + let resolve!: (value: T) => void; + let reject!: (reason: unknown) => void; + const promise = new Promise((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 { + #values: PrpEvent[] = []; + #pending: ReturnType>> | 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 { + return this.reads >= count ? Promise.resolve() : new Promise(resolve => this.#reads.push({ count, resolve })); + } + + next(): Promise> { + 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>(); + return this.#pending.promise; + } + + async return(): Promise> { + 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(); + 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(); + persistence.appendEvent.mockImplementationOnce(() => firstWrite.promise); + const batchWrite = deferred(); + 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>)("stops read-ahead at $eventType priority $priority", async boundary => { + const stream = new EventStream(); + const persistence = port(); + const write = deferred(); + 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(); + 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(); + 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(); + 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(); + const persistence: Pick = { + 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(), batchWrite = deferred(); + 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(); + }); +}); diff --git a/packages/paperclip-runner/src/native-event-persistence.ts b/packages/paperclip-runner/src/native-event-persistence.ts new file mode 100644 index 00000000000..6e74d7dcf52 --- /dev/null +++ b/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 } + | { 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 { + readonly #iterator: AsyncIterator; + readonly #port: Pick; + readonly #buffered: BufferedEvent[] = []; + readonly #committed: Array<{ event: PrpEvent; receipt: AppendedEventReceipt }> = []; + readonly #receipts = new WeakMap(); + #pendingRead: Promise | undefined; + #readOutcome: ReadOutcome | undefined; + #stopped = false; + + constructor( + iterator: AsyncIterator, + port: Pick, + ) { + this.#iterator = iterator; + this.#port = port; + } + + #requestRead(): Promise { + 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> { + 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 { + 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 { + 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> { + this.#stopped = true; + this.#buffered.length = 0; + this.#committed.length = 0; + return await this.#iterator.return?.() ?? { value: undefined, done: true }; + } +} diff --git a/packages/paperclip-runner/src/native-session-runtime.test.ts b/packages/paperclip-runner/src/native-session-runtime.test.ts index 11f665119dd..c9a0b0510c5 100644 --- a/packages/paperclip-runner/src/native-session-runtime.test.ts +++ b/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(resolve => { releaseFirst = resolve; }); + let reachedBoundary!: () => void; + const boundaryReached = new Promise(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(); diff --git a/packages/paperclip-runner/src/native-session-runtime.ts b/packages/paperclip-runner/src/native-session-runtime.ts index fbbb6a98133..57c925ec07e 100644 --- a/packages/paperclip-runner/src/native-session-runtime.ts +++ b/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 | null = null; const inputTimers = new Map>(); const handoffCleanupOperations = new Set>(); - 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((_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"); diff --git a/server/src/__tests__/heartbeat-run-event-batch.test.ts b/server/src/__tests__/heartbeat-run-event-batch.test.ts new file mode 100644 index 00000000000..1c2c524ca9a --- /dev/null +++ b/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>; + let db: ReturnType, otherDb: ReturnType; + 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).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([]); + }); +}); diff --git a/server/src/services/heartbeat-run-events.ts b/server/src/services/heartbeat-run-events.ts index 1f895cc5365..1f3b787ad9d 100644 --- a/server/src/services/heartbeat-run-events.ts +++ b/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 { + 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 }; + hash: string; + firstIndex: number; + row?: typeof heartbeatRunEvents.$inferSelect; + fresh?: boolean; + }; + const byEventId = new Map(); + const bySourceSeq = new Map>(); + const candidates: Candidate[] = []; + const entries: Candidate[] = []; + const matches = (candidate: Candidate, source: NonNullable, 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 | 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(); + 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(); + 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(); + 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)!, + })); + }); +} diff --git a/server/src/services/native-runtime/paperclip-control-plane-port.test.ts b/server/src/services/native-runtime/paperclip-control-plane-port.test.ts index 92aefdd9ad7..0a767278633 100644 --- a/server/src/services/native-runtime/paperclip-control-plane-port.test.ts +++ b/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 | 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 }); diff --git a/server/src/services/native-runtime/paperclip-control-plane-port.ts b/server/src/services/native-runtime/paperclip-control-plane-port.ts index 4e85a8ef843..68bb07fa38d 100644 --- a/server/src/services/native-runtime/paperclip-control-plane-port.ts +++ b/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, }, - }); - // 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 { + 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) {