From b85a69b334f42f3d7683b83ed49237cb57689237 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Sun, 18 Oct 2026 21:58:32 -0500 Subject: [PATCH] test(server): record observer load in native event replay (#16800) Record observers and polling in synthetic native replay measurements, with zero-viewer controls and correctness checks. Co-Authored-By: Paperclip --- doc/DEVELOPING.md | 37 +++ .../helpers/observer-replay-report.ts | 65 ++++ .../__tests__/native-observer-replay.test.ts | 310 ++++++++++++++++++ 3 files changed, 412 insertions(+) create mode 100644 server/src/__tests__/helpers/observer-replay-report.ts create mode 100644 server/src/__tests__/native-observer-replay.test.ts diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index cfc68beb523..c56e8dfb675 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -2541,6 +2541,43 @@ package avatar PNGs using the API renderer; static hosting needs no API proxy. ### Investigating polling load +Run the deterministic native event replay with the normal server test runner: + +```sh +pnpm --filter @paperclipai/server exec vitest run src/__tests__/native-observer-replay.test.ts +``` + +The test uses disposable PostgreSQL and the production event persistence and +authenticated WebSocket paths. Its `OBSERVER_REPLAY` JSON line follows the +existing benchmark stdout convention. Each case records the measured interval, +active viewers and subscribers, polling cadence and actual read offsets, timing, +query counts, and event/log/delivery/duplicate-replay assertions. Push-only cases +mark polling as not applicable. Subscriber counts include one diagnostic listener +that checks duplicate publication synchronously and does no database work. +The polling fixture reads the production replay +port; it does not model the full browser query graph. Query counts include +Drizzle statement attempts on the measured connections, including authorization, +duplicate replay, and separately counted delivery-time visibility proof reads. +Seed, handshake, post-interval proof reads, and driver SQL are outside that count. + +`interval.workMs` stops after event ingestion, delivery, duplicate replay, and +visibility proofs. Performance comparisons use this elapsed time. The separate +`interval.wallMs` includes the minimum observation window and polling drain, so +that deliberate wait cannot hide work-time differences. Polling cases wait for +two completed read batches with a bounded deadline before stopping observation. +The `slow-first-poll` regression delays its first batch until 100 ms after work +completion and still requires two completed real reads per viewer. That injected +delay is explicit in the polling evidence and comparison matching rules. + +Every viewer-sensitive case names its zero-viewer control. The report shows a +matched push pair, unmatched viewer-count and polling pairs, and explicit +observation-load experiments. Unmatched pairs have no timing ratio. A different +fixture or observer churn stays unmatched even in an observation experiment. +These local diagnostics do not establish hosted latency or provider performance. +The output is test evidence; there is no product API or dashboard consumer to +migrate. Keep future fields additive and check the schema and comparison status +before interpreting timings. + The company heartbeat-run and live-run lists load secret registries in one company-scoped query per response. Registry reads project only `paperclipSecretRedactions` from the run context. They do not load the full diff --git a/server/src/__tests__/helpers/observer-replay-report.ts b/server/src/__tests__/helpers/observer-replay-report.ts new file mode 100644 index 00000000000..4bdf0e7946c --- /dev/null +++ b/server/src/__tests__/helpers/observer-replay-report.ts @@ -0,0 +1,65 @@ +// Test evidence only. Keep observation load beside timing and query counts. +export interface ObserverLoad { + activeViewers: { start: number; end: number; min: number; max: number }; + activeSubscribers: { start: number; end: number; min: number; max: number }; + diagnosticSubscribers: number; + transport: "authenticated_websocket"; + polling: { + applicability: "applicable" | "not_applicable"; + reason: "scheduled_log_reads" | "push_only" | "no_viewers"; + cadenceMs: number | null; + firstBatchDelayAfterWorkMs: number; + completedBatches: number; + offsetsMs: number[][]; + }; +} + +export interface ObserverReplayMeasurement { + id: string; + fixture: string; + zeroViewerControlId: string; + observers: ObserverLoad; + interval: { startedAt: string; finishedAt: string; wallMs: number; workMs: number }; + queries: { persistence: number; viewerDelivery: number; polling: number; visibilityProof: number; total: number }; + correctness: { + committedEvents: number; + durableEvents: number; + duplicateReceipts: number; + deliveriesPerViewer: number[]; + durableOutput: boolean; + orderedEvents: boolean; + duplicateReplay: boolean; + safeViewerPayloads: boolean; + }; +} + +export function compareObserverReplays( + baseline: ObserverReplayMeasurement, + candidate: ObserverReplayMeasurement, + axis: "implementation" | "observation_load" = "implementation", +) { + const differences: string[] = []; + if (baseline.fixture !== candidate.fixture) differences.push("fixture"); + for (const field of ["activeViewers", "activeSubscribers", "diagnosticSubscribers", "transport"] as const) { + if (JSON.stringify(baseline.observers[field]) !== JSON.stringify(candidate.observers[field])) differences.push(field); + } + // Actual offsets describe scheduling jitter. Compare the configured cadence, + // not the number of reads accumulated in differently sized timing windows. + if (baseline.observers.polling.cadenceMs !== candidate.observers.polling.cadenceMs || + baseline.observers.polling.firstBatchDelayAfterWorkMs !== candidate.observers.polling.firstBatchDelayAfterWorkMs || + baseline.observers.polling.applicability !== candidate.observers.polling.applicability) differences.push("polling"); + const stable = [baseline, candidate].every(({ observers }) => + [observers.activeViewers, observers.activeSubscribers].every(count => + count.start === count.end && count.start === count.min && count.start === count.max)); + if (!stable) differences.push("observer_churn"); + const status = differences.length === 0 ? "matched" + : axis === "observation_load" && stable && !differences.includes("fixture") ? "observation_experiment" + : "unmatched"; + return { + baselineId: baseline.id, candidateId: candidate.id, axis, status, differences, + // Never attribute a performance ratio to a confounded implementation pair. + timingBasis: "interval.workMs", + wallTimeRatio: status === "matched" && baseline.interval.workMs > 0 + ? candidate.interval.workMs / baseline.interval.workMs : null, + }; +} diff --git a/server/src/__tests__/native-observer-replay.test.ts b/server/src/__tests__/native-observer-replay.test.ts new file mode 100644 index 00000000000..468722bc2d9 --- /dev/null +++ b/server/src/__tests__/native-observer-replay.test.ts @@ -0,0 +1,310 @@ +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { createHash, randomUUID } from "node:crypto"; +import { execFileSync } from "node:child_process"; +import { once } from "node:events"; +import { createServer } from "node:http"; +import type { AddressInfo } from "node:net"; +import { setTimeout as delay } from "node:timers/promises"; +import { asc, eq } from "drizzle-orm"; +import { drizzle } from "drizzle-orm/postgres-js"; +import { WebSocket } from "ws"; +import { + agents, authUsers, companies, companyMemberships, completionContracts, + createDb, heartbeatRunEvents, heartbeatRuns, issues, type Db, +} from "@paperclipai/db"; +import * as schema from "@paperclipai/db/schema/index"; +import type { LiveEvent } from "@paperclipai/shared"; +import type { PrpEvent } from "../vendor/paperclip-runner/index.js"; +import { setupLiveEventsWebSocketServer } from "../realtime/live-events-ws.js"; +import { PaperclipControlPlanePort } from "../services/native-runtime/paperclip-control-plane-port.js"; +import { subscribeCompanyLiveEvents } from "../services/live-events.js"; +import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { compareObserverReplays, type ObserverReplayMeasurement } from "./helpers/observer-replay-report.js"; + +const output = "Synthetic replay output.\n"; +const fixturePayloads = [ + ...Array.from({ length: 6 }, (_, index) => ({ + eventType: "tool.execution.started", + payload: { + schema: "paperclip.tool.execution.v1", executionId: `fixture-tool-${index + 1}`, + transport: "builtin", operation: "execute", name: "synthetic-tool", target: null, + namespace: null, readOnly: true, status: "running", durationMs: null, exitCode: null, + progress: null, output: null, outputBytes: 0, outputTruncated: false, outputDigest: null, + }, + })), + { eventType: "item.delta", payload: { kind: "agentMessage", text: output } }, + { eventType: "item.completed", payload: { kind: "agentMessage", item: { id: "answer", text: output } } }, +]; +const fixture = createHash("sha256").update(JSON.stringify(fixturePayloads)).digest("hex"); +const expectedDeliveries = 7; // item.delta is durable but does not emit a live signal. + +describe("native event replay observation evidence", () => { + let temporary: Awaited>; + let db: Db; + const companyId = randomUUID(); + const agentId = randomUUID(); + const userId = `observer-${randomUUID()}`; + + beforeAll(async () => { + temporary = await startEmbeddedPostgresTestDatabase("observer-replay-"); + db = createDb(temporary.connectionString); + await db.insert(companies).values({ id: companyId, name: "Synthetic replay", issuePrefix: "REPLAY" }); + await db.insert(agents).values({ id: agentId, companyId, name: "Fixture", adapterType: "codex_local" }); + await db.insert(authUsers).values({ id: userId, name: "Fixture viewer", email: `${userId}@example.test`, + createdAt: new Date(), updatedAt: new Date() }); + await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: userId, + membershipRole: "owner", status: "active" }); + }); + + afterAll(async () => { await temporary?.cleanup(); }); + + async function replay(id: string, viewerCount: number, cadenceMs: number | null, + firstBatchDelayAfterWorkMs = 0): Promise { + const runId = randomUUID(), issueId = randomUUID(), sessionId = randomUUID(); + const sourceInstanceId = randomUUID(), completionContractId = randomUUID(); + const identity = { companyId, agentId, issueId, runId, sessionId }; + const binding = { ...identity, sourceInstanceId, controlPlaneSourceInstanceId: `control-${runId}`, + completionContractId, completionContractSha256: "observer-replay-contract" }; + await db.insert(issues).values({ id: issueId, companyId, title: "Synthetic replay", status: "in_progress", assigneeAgentId: agentId }); + await db.insert(completionContracts).values({ id: completionContractId, companyId, issueId, revision: 1, + schemaVersion: "paperclip.completion-contract.v1", policyVersion: "fixture-v1", risk: "standard", + completionAuthority: "server_arbiter", incompleteCriteriaPolicy: "preserve_non_terminal", + contractJson: { revision: "fixture-v1", objective: "Synthetic replay", criteria: [{ id: "objective", requirement: "Replay" }] }, + canonicalSha256: binding.completionContractSha256, createdByActorType: "system", createdByActorId: "test" }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, scopeKind: "issue", issueId, + status: "running", runtimeMode: "native", nativeIssueId: issueId, nativeSessionId: sessionId, + runnerInstanceId: sourceInstanceId, completionContractId, completionContractSha256: binding.completionContractSha256 }); + + const queries = { persistence: 0, viewerDelivery: 0, polling: 0, visibilityProof: 0, total: 0 }; + let measuring = false; + const counted = (category: "persistence" | "viewerDelivery" | "polling") => drizzle(db.$client, { + schema, logger: { logQuery() { if (measuring) queries[category] += 1; } }, + }) as Db; + const port = new PaperclipControlPlanePort(counted("persistence"), binding); + const pollPort = new PaperclipControlPlanePort(counted("polling"), binding); + await port.openRun({ identity, backendKind: "mock", sourceInstanceId }); + const server = createServer(); + const wss = setupLiveEventsWebSocketServer(server, counted("viewerDelivery"), { + deploymentMode: "authenticated", + resolveCloudActor: async () => ({ userId, companyIds: [companyId] }), + }); + const sockets: WebSocket[] = []; + const received: LiveEvent[][] = []; + const completedDeliveries: Promise[] = []; + const deliveryReads: Promise[] = []; + const visibilityDb = createDb(temporary.connectionString); + const proofDb = drizzle(visibilityDb.$client, { schema, logger: { + logQuery() { if (measuring) queries.visibilityProof += 1; }, + } }); + const pollOffsets: number[][] = Array.from({ length: viewerCount }, () => []); + const signals: LiveEvent[] = []; + const unsubscribe = subscribeCompanyLiveEvents(companyId, event => { + if (event.type === "heartbeat.run.event" && event.payload.runId === runId) signals.push(event); + }); + let polling = false; + let pollWork = Promise.resolve(); + let completedPollBatches = 0; + let pollingFailure: unknown; + const workFinished = Promise.withResolvers(); + let activeViewers = 0; + let minViewers = viewerCount; + try { + server.listen(0, "127.0.0.1"); + await once(server, "listening"); + const url = `ws://127.0.0.1:${(server.address() as AddressInfo).port}/api/companies/${companyId}/events/ws`; + // The production upgrade path must reject a company outside the actor's scope. + const denied = new WebSocket(url.replace(companyId, randomUUID())); + const rejection = await new Promise((resolve, reject) => { + denied.once("unexpected-response", (_request, response) => { response.resume(); denied.terminate(); resolve(response.statusCode!); }); + denied.once("error", () => {}); + denied.once("open", () => { denied.terminate(); reject(new Error("foreign-company viewer admitted")); }); + }); + expect(rejection).toBe(403); + for (let index = 0; index < viewerCount; index += 1) { + const socket = new WebSocket(url); + sockets.push(socket); + const messages: LiveEvent[] = []; + const delivered = Promise.withResolvers(); + completedDeliveries.push(delivered.promise); + received.push(messages); + socket.on("message", bytes => { + const event = JSON.parse(bytes.toString()) as LiveEvent; + messages.push(event); + // A separate connection must see the committed row at delivery time. + deliveryReads.push((async () => { + const rows = await proofDb.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, runId)); + expect(rows.some(row => row.seq === event.payload.seq)).toBe(true); + })()); + if (messages.length === expectedDeliveries) delivered.resolve(); + }); + await once(socket, "open"); + activeViewers += 1; + socket.once("close", () => { + activeViewers -= 1; + if (measuring) minViewers = Math.min(minViewers, activeViewers); + }); + } + const events: PrpEvent[] = fixturePayloads.map((entry, index) => ({ + schema: "paperclip.prp.event.v1", sourceEventId: `fixture:${index + 1}`, sourceSeq: index + 1, + sourceInstanceId, sourceKind: "runner", runId, normalizedSessionId: sessionId, + turnId: "fixture-turn", itemId: "answer", eventType: entry.eventType, payload: entry.payload, + schemaVersion: 1, priority: 0, emittedAt: "2026-10-01T00:00:00.000Z", + })); + const startedAt = new Date().toISOString(); + const started = performance.now(); + measuring = true; + polling = cadenceMs !== null; + pollWork = (async () => { + while (polling) { + // Controlled slow-read regression: keep the first batch pending past + // work completion and the minimum window, while retaining real reads. + if (completedPollBatches === 0 && firstBatchDelayAfterWorkMs > 0) { + await workFinished.promise; + await delay(firstBatchDelayAfterWorkMs); + } + await Promise.all(pollOffsets.map(async offsets => { + offsets.push(performance.now() - started); + await pollPort.replayEvents({ runId, sourceInstanceId, afterSourceSeq: 0, limit: 100 }); + })); + completedPollBatches += 1; + if (polling) await delay(cadenceMs!); + } + })().catch(error => { pollingFailure = error; polling = false; }); + for (const event of events) expect((await port.appendEvent(event)).disposition).toBe("committed"); + // Observe delivery directly so a polling assertion's retry interval does + // not quantize work timing. Keep a bounded failure path for lost frames. + const deliveryDeadline = new AbortController(); + try { + await Promise.race([ + Promise.all(completedDeliveries), + delay(5_000, undefined, { signal: deliveryDeadline.signal }).then(() => { + throw new Error("replay viewer delivery deadline exceeded"); + }), + ]); + } finally { deliveryDeadline.abort(); } + for (const event of events) expect((await port.appendEvent(event)).disposition).toBe("duplicate"); + // This synchronous guard catches duplicate publication even if an async + // authorization queue has not reached the client by the final pong. + expect(signals).toHaveLength(expectedDeliveries); + // A pong follows earlier server frames; duplicate replay must add none. + await Promise.all(sockets.map(async socket => { + const pong = once(socket, "pong"); socket.ping(); await pong; + })); + await Promise.all(deliveryReads); + // Work timing stops before the deliberate observation extension. This + // preserves timing differences even when both runs finish within 60 ms. + const workMs = performance.now() - started; + workFinished.resolve(); + // Give every case the same minimum observation window, including the control. + await delay(Math.max(0, 60 - (performance.now() - started))); + if (cadenceMs !== null) { + // Read offsets count starts, not completions. Slow database reads must + // finish two batches before the observation window can close. + await expect.poll(() => { + if (pollingFailure) throw pollingFailure; + return completedPollBatches; + }, { timeout: 5_000 }).toBeGreaterThanOrEqual(2); + } + polling = false; + await pollWork; + if (pollingFailure) throw pollingFailure; + const interval = { startedAt, finishedAt: new Date().toISOString(), wallMs: performance.now() - started, workMs }; + measuring = false; + queries.total = queries.persistence + queries.viewerDelivery + queries.polling + queries.visibilityProof; + + // Read through a fresh production port after ingestion, not a writer cache. + const reopened = new PaperclipControlPlanePort(visibilityDb, binding); + const durable = await reopened.replayEvents({ runId, sourceInstanceId, afterSourceSeq: 0, limit: 100 }); + expect(durable.events).toEqual(events); + expect(durable.highestContiguousSourceSeq).toBe(events.length); + const rows = await visibilityDb.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, runId)).orderBy(asc(heartbeatRunEvents.seq)); + expect(rows.map(row => row.seq)).toEqual(events.map(event => event.sourceSeq)); + expect(durable.events.at(-1)?.payload).toEqual({ kind: "agentMessage", item: { id: "answer", text: output } }); + for (const messages of received) { + expect(messages.map(event => event.payload)).toEqual(events.filter(event => event.eventType !== "item.delta").map(event => ({ + runId, agentId, issueId, seq: event.sourceSeq, eventType: event.eventType, + }))); + expect(messages.every(event => event.companyId === companyId)).toBe(true); + expect(JSON.stringify(messages)).not.toContain(output.trim()); + } + expect(activeViewers).toBe(viewerCount); + if (cadenceMs !== null) expect(pollOffsets.every(offsets => offsets.length >= 2)).toBe(true); + return { + id, fixture, zeroViewerControlId: "zero-viewers", + interval, queries, + observers: { + activeViewers: { start: viewerCount, end: activeViewers, min: minViewers, max: viewerCount }, + activeSubscribers: { start: viewerCount + 1, end: wss.clients.size + 1, min: minViewers + 1, max: viewerCount + 1 }, + diagnosticSubscribers: 1, + transport: "authenticated_websocket", + polling: { applicability: cadenceMs === null ? "not_applicable" : "applicable", + reason: cadenceMs !== null ? "scheduled_log_reads" : viewerCount === 0 ? "no_viewers" : "push_only", + cadenceMs, firstBatchDelayAfterWorkMs, completedBatches: completedPollBatches, offsetsMs: pollOffsets }, + }, + correctness: { committedEvents: events.length, durableEvents: rows.length, duplicateReceipts: events.length, + deliveriesPerViewer: received.map(messages => messages.length), durableOutput: true, + orderedEvents: true, duplicateReplay: true, safeViewerPayloads: true }, + }; + } finally { + measuring = false; + polling = false; + workFinished.resolve(); + unsubscribe(); + await pollWork; + for (const socket of sockets) socket.terminate(); + // Wait for server-side close handlers to remove the subscriptions. + await Promise.all([...wss.clients].map(socket => { + const closed = new Promise(resolve => socket.on("close", resolve)); socket.terminate(); return closed; + })); + // Closing the real WebSocket server emits its close event and clears the + // production heartbeat timer. Closing only HTTP would leave it running. + await new Promise((resolve, reject) => { + (wss as typeof wss & { close(callback: (error?: Error) => void): void }) + .close(error => error ? reject(error) : resolve()); + }); + if (server.listening) await new Promise((resolve, reject) => server.close(error => error ? reject(error) : resolve())); + await visibilityDb.$client.end(); + } + } + + it("reports matched, confounded, and intentional observer-load comparisons with a zero-viewer control", async () => { + const zero = await replay("zero-viewers", 0, null); + const first = await replay("push-a", 2, null); + const second = await replay("push-b", 2, null); + const polled = await replay("polled", 2, 20); + const slowPolled = await replay("slow-first-poll", 2, 20, 100); + const comparisons = [ + compareObserverReplays(first, second), + compareObserverReplays(zero, first), + compareObserverReplays(first, polled), + compareObserverReplays(zero, first, "observation_load"), + compareObserverReplays(zero, polled, "observation_load"), + compareObserverReplays(polled, slowPolled), + ]; + expect(comparisons.map(pair => pair.status)).toEqual(["matched", "unmatched", "unmatched", "observation_experiment", "observation_experiment", "unmatched"]); + expect(comparisons[5].differences).toEqual(["polling"]); + expect(comparisons[1].wallTimeRatio).toBeNull(); + expect(comparisons[2].differences).toEqual(["polling"]); + // Equal observation windows must not hide different work-completion times. + expect(compareObserverReplays( + { ...first, interval: { ...first.interval, wallMs: 60, workMs: 15 } }, + { ...second, interval: { ...second.interval, wallMs: 60, workMs: 45 } }, + )).toMatchObject({ timingBasis: "interval.workMs", wallTimeRatio: 3 }); + expect(first.queries.viewerDelivery).toBeGreaterThan(zero.queries.viewerDelivery); + expect(zero.queries.polling).toBe(0); + expect(polled.queries.polling).toBeGreaterThan(0); + // A declared experiment cannot excuse a different fixture or observer churn. + expect(compareObserverReplays(first, { ...second, fixture: "different" }, "observation_load").status).toBe("unmatched"); + const churn = structuredClone(second); + churn.observers.activeSubscribers.min = 0; + expect(compareObserverReplays(first, churn, "observation_load").status).toBe("unmatched"); + process.stdout.write(`OBSERVER_REPLAY ${JSON.stringify({ + schema: "paperclip.observer_replay.v1", + sourceRevision: execFileSync("git", ["rev-parse", "HEAD"], { encoding: "utf8" }).trim(), + node: process.version, platform: process.platform, + queryScope: "Drizzle statement attempts during ingestion, duplicate replay, and viewer/poll drain, including separately counted delivery-time visibility proofs; excludes seed, handshake, post-interval proof reads, and driver-internal SQL", + scope: "Synthetic local production persistence and authorized WebSocket delivery; no browser, provider, hosted latency, or exact historical query-count claim", + cases: [zero, first, second, polled, slowPolled], comparisons, + })}\n`); + }, 30_000); +});