From c2add2f2124ff71f2fe955b1e6105a9e71711310 Mon Sep 17 00:00:00 2001 From: Dotta Date: Tue, 3 Nov 2026 12:33:07 -0600 Subject: [PATCH] perf(server): publish committed text before awaited observers Co-Authored-By: Paperclip --- .../paperclip-control-plane-port.test.ts | 90 +++++++++++++++++++ .../paperclip-control-plane-port.ts | 51 ++++++----- 2 files changed, 121 insertions(+), 20 deletions(-) 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 0a767278633..f6528c96345 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 @@ -48,6 +48,7 @@ import { import { startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; import { startRecoveryPostgres } from "../../__tests__/helpers/recovery-postgres.js"; import { PaperclipControlPlanePort } from "./paperclip-control-plane-port.js"; +import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js"; import { nativeSha256 } from "./canonical.js"; import { NativeRunCoordinatorStore } from "./native-run-coordinator-store.js"; import { finalizeNativeRun, repairCommittedNativeChatResponse } from "./native-run-finalizer.js"; @@ -66,6 +67,7 @@ import { childReviewOutcomes } from "./child-review-outcomes.js"; import { createPostgresRunDispatchAdapter } from "../../modules/run-dispatch/adapters/postgres.js"; import { materializeRuntimeQuestionFallback } from "./native-session-executor.js"; import { + publishLiveEvent, subscribeAllCompanyLiveEvents, subscribeCompanyLiveEvents, } from "../live-events.js"; @@ -576,6 +578,94 @@ describe("PaperclipControlPlanePort conformance", () => { } }); + it("publishes all committed text before a callback publishes its later legacy event", async () => { + const identity = { ...CONTROL_PLANE_CONFORMANCE_OPEN.identity, runId: randomUUID(), sessionId: randomUUID() }; + const runnerId = randomUUID(); + 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 callbacks: number[] = [], published: number[] = []; + const port = new PaperclipControlPlanePort(db, { ...identity, completionContractId: contractId, + completionContractSha256: contractSha, sourceInstanceId: runnerId, controlPlaneSourceInstanceId: `nested-${identity.runId}` }, { + onCommittedEvent: async event => { + callbacks.push(event.sourceSeq); + if (event.sourceSeq !== 1) return; + // The production first-text trace sink takes this same legacy append + // and publication path while the PRP commit callback is still pending. + const nested = await appendHeartbeatRunEvent(db, { companyId: identity.companyId, + agentId: identity.agentId, runId: identity.runId, eventType: "run.performance.span", + stream: "system", payload: { span: "task.time_to_first_token" } }); + publishLiveEvent({ companyId: identity.companyId, type: "heartbeat.run.event", payload: { + runId: identity.runId, agentId: identity.agentId, seq: nested.row.seq, + eventType: nested.row.eventType, payload: nested.row.payload, + } }); + }, + }); + await port.openRun({ identity, backendKind: "mock", sourceInstanceId: runnerId }); + const events: PrpEvent[] = [1, 2].map(sourceSeq => ({ + schema: "paperclip.prp.event.v1", schemaVersion: 1, sourceEventId: `${runnerId}:${sourceSeq}`, sourceSeq, + sourceInstanceId: runnerId, sourceKind: "runner", runId: identity.runId, normalizedSessionId: identity.sessionId, + turnId: "nested-turn", itemId: "nested-item", eventType: "item.delta", priority: 1, + emittedAt: "2026-11-03T00:00:00.000Z", payload: { kind: "agentMessage", text: `chunk-${sourceSeq}` }, + })); + const unsubscribe = subscribeCompanyLiveEvents(identity.companyId, event => { + if (event.type === "heartbeat.run.event" && event.payload.runId === identity.runId) published.push(Number(event.payload.seq)); + }); + try { await port.appendEvents(events); } finally { unsubscribe(); } + expect(published).toEqual([1, 2, 3]); + expect(callbacks).toEqual([1, 2]); + const rows = await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, identity.runId)).orderBy(heartbeatRunEvents.seq); + expect(rows.map(row => row.eventType)).toEqual(["item.delta", "item.delta", "run.performance.span"]); + }); + + it.each([[false, "metric"], [true, "metric"], [true, "business"]] as const)( + "publishes committed text before awaiting a blocked %s/%s callback", + async (batch, blocked) => { + const identity = { ...CONTROL_PLANE_CONFORMANCE_OPEN.identity, runId: randomUUID(), sessionId: randomUUID() }; + const runnerId = randomUUID(); + 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 }); + let release!: () => void, entered!: () => void; + const hold = new Promise(resolve => { release = resolve; }); + const reached = new Promise(resolve => { entered = resolve; }); + const published: number[] = [], callbacks: number[] = [], endedAt: number[] = []; + const port = new PaperclipControlPlanePort(db, { ...identity, completionContractId: contractId, + completionContractSha256: contractSha, sourceInstanceId: runnerId, controlPlaneSourceInstanceId: `blocked-${identity.runId}` }, { + onEventPersistence: async (event, _startedAt, persistenceEndedAt) => { + endedAt.push(persistenceEndedAt); + if (blocked === "metric" && event.sourceSeq === 1) { entered(); await hold; } + }, + onCommittedEvent: async event => { + callbacks.push(event.sourceSeq); + if (blocked === "business" && event.sourceSeq === 1) { entered(); await hold; } + }, + }); + await port.openRun({ identity, backendKind: "mock", sourceInstanceId: runnerId }); + const events: PrpEvent[] = Array.from({ length: batch ? 2 : 1 }, (_, index) => ({ + schema: "paperclip.prp.event.v1", schemaVersion: 1, sourceEventId: `${runnerId}:${index + 1}`, sourceSeq: index + 1, + sourceInstanceId: runnerId, sourceKind: "runner", runId: identity.runId, normalizedSessionId: identity.sessionId, + turnId: "blocked-turn", itemId: "blocked-item", eventType: "item.delta", priority: 1, + emittedAt: "2026-11-03T00:00:00.000Z", payload: { kind: "agentMessage", text: `chunk-${index + 1}` }, + })); + const unsubscribe = subscribeCompanyLiveEvents(identity.companyId, event => { + if (event.type === "heartbeat.run.event" && event.payload.runId === identity.runId) published.push(Number(event.payload.seq)); + }); + let settled = false; + const pending = (batch ? port.appendEvents(events) : port.appendEvent(events[0]!)) + .then(value => { settled = true; return value; }); + try { + await reached; + expect(published).toEqual(events.map(event => event.sourceSeq)); + expect(callbacks).toEqual(blocked === "metric" ? [] : [1]); + expect(settled).toBe(false); + expect(await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, identity.runId))).toHaveLength(events.length); + } finally { release(); await pending; unsubscribe(); } + expect(callbacks).toEqual(events.map(event => event.sourceSeq)); + expect(new Set(endedAt).size).toBe(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 68bb07fa38d..ecf0c96cd4b 100644 --- a/server/src/services/native-runtime/paperclip-control-plane-port.ts +++ b/server/src/services/native-runtime/paperclip-control-plane-port.ts @@ -437,9 +437,16 @@ export class PaperclipControlPlanePort implements ControlPlanePort { }); const persistenceEndedAt = persistenceStartedAt + performance.now() - persistenceClock; this.#assertActive(options); + // Publish the whole committed prefix synchronously. A callback can append + // its own later run event (for example a first-text timing span); awaiting + // it between rows would create a false gap and delay already durable text. + for (const receipt of receipts) { + this.#assertActive(options); + if (receipt.disposition === "committed") this.#publishCommittedEvent(receipt); + } for (let index = 0; index < events.length; index += 1) { this.#assertActive(options); - await this.#publishEventReceipt(events[index]!, receipts[index]!, persistenceStartedAt, persistenceEndedAt); + await this.#observeEventReceipt(events[index]!, receipts[index]!, persistenceStartedAt, persistenceEndedAt, options); } return receipts.map(persisted => ({ cursor: persisted.row.seq, @@ -448,27 +455,31 @@ export class PaperclipControlPlanePort implements ControlPlanePort { })); } - async #publishEventReceipt(event: PrpEvent, persisted: AppendHeartbeatRunEventResult, - persistenceStartedAt: number, persistenceEndedAt: number): Promise { + #publishCommittedEvent(persisted: AppendHeartbeatRunEventResult): void { + try { + publishLiveEvent({ + companyId: this.#binding.companyId, + type: "heartbeat.run.event", + payload: { + runId: this.#binding.runId, + agentId: this.#binding.agentId, + issueId: this.#binding.issueId, + seq: persisted.row.seq, + eventType: persisted.row.eventType, + stream: persisted.row.stream, + level: persisted.row.level, + payload: persisted.row.payload, + }, + }); + } catch { + } + } + + async #observeEventReceipt(event: PrpEvent, persisted: AppendHeartbeatRunEventResult, + persistenceStartedAt: number, persistenceEndedAt: number, options?: { signal: AbortSignal }): Promise { try { await this.#onEventPersistence?.(event, persistenceStartedAt, persistenceEndedAt); } catch {} + this.#assertActive(options); if (persisted.disposition === "committed") { - try { - publishLiveEvent({ - companyId: this.#binding.companyId, - type: "heartbeat.run.event", - payload: { - runId: this.#binding.runId, - agentId: this.#binding.agentId, - issueId: this.#binding.issueId, - seq: persisted.row.seq, - eventType: persisted.row.eventType, - stream: persisted.row.stream, - level: persisted.row.level, - payload: persisted.row.payload, - }, - }); - } catch { - } await this.#onCommittedEvent?.(event, persisted.row.seq); } else { // A recovered runner may replay the event whose durable side effects