From c921d7c7c7f53787c4ed00e2d412bc88308b1688 Mon Sep 17 00:00:00 2001 From: Dotta Date: Tue, 3 Nov 2026 12:28:31 -0600 Subject: [PATCH] test(perf): qualify native ingestion with matched browser evidence Co-Authored-By: Paperclip --- scripts/benchmark-native-event-ingestion.ts | 153 +++++++++-- scripts/compare-native-ingestion-browser.ts | 99 +++++++ .../fixtures/composer-stop-codex-loader.mjs | 2 +- tests/e2e/fixtures/in-feed-codex.mjs | 72 ++++- tests/e2e/fixtures/native-ingestion-bin/codex | 2 + tests/e2e/native-event-ingestion.journey.ts | 250 ++++++++++++++++++ .../e2e/playwright-native-ingestion.config.ts | 37 +++ 7 files changed, 584 insertions(+), 31 deletions(-) create mode 100644 scripts/compare-native-ingestion-browser.ts create mode 100755 tests/e2e/fixtures/native-ingestion-bin/codex create mode 100644 tests/e2e/native-event-ingestion.journey.ts create mode 100644 tests/e2e/playwright-native-ingestion.config.ts diff --git a/scripts/benchmark-native-event-ingestion.ts b/scripts/benchmark-native-event-ingestion.ts index ed9b8924475..b20eaee4bbd 100644 --- a/scripts/benchmark-native-event-ingestion.ts +++ b/scripts/benchmark-native-event-ingestion.ts @@ -14,6 +14,7 @@ import { import { installDatabaseWorkSignals } from "../packages/db/src/work-signals.js"; import type { AppendHeartbeatRunEventInput, AppendHeartbeatRunEventResult } from "../server/src/services/heartbeat-run-events.js"; import { resolveExternalTestDatabaseUrl } from "../server/src/__tests__/helpers/external-test-database.js"; +import { nativeSha256 } from "../server/src/services/native-runtime/canonical.js"; // This measures the real database owner, not browser delivery or a provider. // The assembled application/browser acceptance is a separate required layer. @@ -21,8 +22,19 @@ const root = fileURLToPath(new URL("../", import.meta.url)); const serverRequire = createRequire(new URL("../server/package.json", import.meta.url)); const { drizzle } = await import(serverRequire.resolve("drizzle-orm/postgres-js")); const sourcePath = "server/src/services/heartbeat-run-events.ts"; +const cursorSourcePath = "server/src/services/heartbeat/native-event-cursors.ts"; const arg = (name: string) => process.argv.find(value => value.startsWith(`--${name}=`))?.slice(name.length + 3); const git = (args: string[]) => execFileSync("git", args, { cwd: root, encoding: "utf8" }); +function assertServiceAcceptance(recorded: any) { + assert.equal(recorded.cleanupComplete, true, "Owned benchmark resources have not finished cleanup"); + assert.ok(["compare", "batch"].includes(recorded.phase), "A/A calibration is not performance acceptance"); + assert.equal(recorded.workload?.quick, false, "Qualification smoke cannot establish acceptance"); + assert.ok(recorded.workload.pairs >= 5, "Acceptance requires at least five paired samples"); + assert.ok([1000, 10000, 100000].every(history => recorded.workload.histories.includes(history)), "Acceptance requires the complete retained-history matrix"); + assert.ok(["ordered", "replay", "late-gap", "warm-suffix"].every(scenario => recorded.workload.scenarios.includes(scenario)), "Acceptance requires every fixed service scenario"); + assert.ok(recorded.comparison?.length > 0, "Acceptance requires a same-workload A/A noise comparison"); + assert.ok(recorded.comparison.every((entry: { acceptancePassed: boolean | null }) => entry.acceptancePassed !== false), "Service performance gate failed; inspect comparison and retained failures"); +} if (arg("validate-report")) { const recorded = JSON.parse(await fs.readFile(arg("validate-report")!, "utf8")); assert.equal(recorded.incomplete, false, "Incomplete evidence cannot qualify a revision"); @@ -30,6 +42,9 @@ if (arg("validate-report")) { for (const [file, expected] of Object.entries(recorded.candidateSourceHashes)) { assert.equal(createHash("sha256").update(await fs.readFile(path.join(root, file))).digest("hex"), expected, `Measured source changed: ${file}`); } + if (process.argv.includes("--require-acceptance")) { + assertServiceAcceptance(recorded); + } console.log(JSON.stringify({ valid: true, phase: recorded.phase, baselineRevision: recorded.baselineRevision, candidateRevision: recorded.candidateRevision, scope: recorded.scope })); process.exit(0); @@ -45,11 +60,13 @@ const candidateRevision = git(["rev-parse", "HEAD"]).trim(); const histories = (arg("histories") ?? (quick ? "1000" : "1000,10000,100000")).split(",").map(Number); const pairs = Number(arg("pairs") ?? (quick ? 1 : 5)); const events = Number(arg("events") ?? (quick ? 8 : 32)); +const groups = Number(arg("groups") ?? (quick ? 1 : 16)); assert.ok(histories.every(value => Number.isSafeInteger(value) && value >= 1)); assert.ok(Number.isSafeInteger(pairs) && pairs >= (quick ? 1 : 5), "Acceptance requires at least five paired samples"); assert.ok(Number.isSafeInteger(events) && events >= 2 && events <= 64 && events % 2 === 0); +assert.ok(Number.isSafeInteger(groups) && groups >= 1 && groups <= 128); const scenarios = quick ? ["ordered"] : ["ordered", "replay", "late-gap", "warm-suffix"]; -const workload = { histories, pairs, events, scenarios, payloadTextBytes: 128, warmSourceOffset: 1000, +const workload = { histories, pairs, events, groups, scenarios, payloadTextBytes: 128, warmSourceOffset: 1000, seed: "native-event-ingestion-v1", retainedHistory: "vacuum-analyzed", quick }; const workloadSha256 = createHash("sha256").update(JSON.stringify(workload)).digest("hex"); const output = arg("output"); @@ -65,6 +82,7 @@ type Sample = { elapsedMs: number; eventsPerSecond: number; newEvents: number; duplicateEvents: number; appCpuMicros: number; databaseCpuMillis: number | null; heapDeltaBytes: number; rssBytes: number; eventLoopUtilization: number; eventLoopDelayMaxMs: number | null; + retainedHistoryBefore: number; retainedHistoryAfter: number; }; type Arm = "baseline" | "candidate"; const median = (values: number[]) => [...values].sort((a, b) => a - b)[Math.floor(values.length / 2)]!; @@ -87,7 +105,7 @@ const report: Record = { baselineSourceSha256: createHash("sha256").update(baselineSource).digest("hex"), measurement: { warmups: 2, pairedOrder: "alternating", cache: "warm; seed and projection preparation excluded", databaseCpuResolutionMillis: 10 }, environment: { node: process.version, platform: process.platform, architecture: process.arch, cpus: availableParallelism(), loadBefore }, - results: [], incomplete: true, + results: [], incomplete: true, cleanupComplete: false, }; async function save() { @@ -129,10 +147,10 @@ function diagnosticDb(target: Db, queries: Query[], transactions: Array<{ elapse const decorate = (connection: Db, transaction?: { acquisitionMs?: number; acquiredAt?: number }) => { const session = (connection as unknown as { session: Session }).session; const prepare = session.prepareQuery; - session.prepareQuery = function (...args: any[]) { + session.prepareQuery = function (this: Session, ...args: any[]) { const prepared = prepare.apply(this, args), execute = prepared.execute; const query = args[0] as Query; - prepared.execute = async function (...values: any[]) { + prepared.execute = async function (this: unknown, ...values: any[]) { const started = performance.now(); try { const result = await execute.apply(this, values); @@ -151,9 +169,9 @@ function diagnosticDb(target: Db, queries: Query[], transactions: Array<{ elapse return prepared; }; const original = connection.transaction.bind(connection); - connection.transaction = ((callback: (tx: Db) => Promise, ...options: any[]) => { + connection.transaction = ((callback: (tx: Db) => Promise, options?: Parameters[1]) => { const started = performance.now(), state: { acquisitionMs?: number; acquiredAt?: number } = {}; - return original(async tx => callback(decorate(tx as unknown as Db, state)), ...options).finally(() => { + return original(async tx => callback(decorate(tx as unknown as Db, state)), options).finally(() => { transactions.push({ elapsedMs: performance.now() - started, lockAcquisitionQueryMs: state.acquisitionMs, lockHeldThroughCommitMs: state.acquiredAt === undefined ? undefined : performance.now() - state.acquiredAt }); }); @@ -171,6 +189,10 @@ try { const changed = git(["diff", "--name-only", baselineRevision, "--", "server/src/services"]) .trim().split("\n").filter(file => file.endsWith(".ts") && !file.endsWith(".test.ts")); const snapshots = new Map([[sourcePath, baselineSource]]); + let baselineCursorSource: string | undefined; + try { baselineCursorSource = git(["show", `${baselineRevision}:${cursorSourcePath}`]); } + catch { /* The pre-cursor baseline has no projection to prepare. */ } + if (baselineCursorSource) snapshots.set(cursorSourcePath, baselineCursorSource); for (const file of changed) { if (snapshots.has(file)) continue; try { snapshots.set(file, git(["show", `${baselineRevision}:${file}`])); } @@ -190,6 +212,8 @@ try { const baselineFile = path.join(baselineDir, sourcePath); const baseline = await import(pathToFileURL(baselineFile).href) as EventModule; const candidate = phase === "aa" ? baseline : await import("../server/src/services/heartbeat-run-events.js") as EventModule; + report.projectionPriming = { baseline: Boolean(baselineCursorSource), candidate: phase === "aa" ? Boolean(baselineCursorSource) : true }; + report.effectiveCandidateRevision = phase === "aa" ? baselineRevision : candidateRevision; if (phase === "batch") assert.equal(typeof candidate.appendHeartbeatRunEvents, "function", "Batch phase requires appendHeartbeatRunEvents"); report.candidateSourceSha256 = createHash("sha256").update(await fs.readFile(path.join(root, sourcePath))).digest("hex"); const proofFiles = [...new Set([sourcePath, "scripts/benchmark-native-event-ingestion.ts", @@ -222,10 +246,13 @@ try { } async function prepare(f: Fixture, arm: Arm) { - if (arm === "baseline" || phase === "aa") return; + const usesBaseline = arm === "baseline" || phase === "aa"; + if (usesBaseline && !baselineCursorSource) return; // Seed setup is an old-version append. Prime using the real fallback and // bounded rebuild before steady-state timing; cold behavior is traced below. - const module = await import("../server/src/services/heartbeat/native-event-cursors.js") as Record; + const module = await import(usesBaseline + ? pathToFileURL(path.join(baselineDir, cursorSourcePath)).href + : pathToFileURL(path.join(root, cursorSourcePath)).href) as Record; const rebuild = module.rebuildNativeEventCursorPage as undefined | ((db: Db, binding: { companyId: string; runId: string }) => Promise<{ pending: boolean }>); assert.equal(typeof rebuild, "function", "Cursor owner must expose rebuildNativeEventCursorPage for seeded-history preparation"); await db!.$client.unsafe("update heartbeat_runs set event_cursor_rebuild_pending=true where id=$1::uuid", [f.runId]); @@ -236,18 +263,29 @@ try { throw new Error("Cursor fixture failed to finish bounded rebuild"); } - function emittedInputs(f: Fixture, scenario: string) { + function emittedInputs(f: Fixture, scenario: string, repetitions = 1) { if (scenario === "late-gap") return [input(f, 1)]; const newCount = scenario === "replay" ? events / 2 : events; - const unique = Array.from({ length: newCount }, (_, n) => input(f, f.next + n)); - f.next += newCount; - return scenario === "replay" ? [...unique, ...unique] : unique; + const values: AppendHeartbeatRunEventInput[] = []; + for (let group = 0; group < repetitions; group++) { + const unique = Array.from({ length: newCount }, (_, n) => input(f, f.next + n)); + f.next += newCount; + values.push(...unique); + if (scenario === "replay") values.push(...unique); + } + return values; } async function invoke(target: Db, arm: Arm, values: AppendHeartbeatRunEventInput[]) { const owner = arm === "baseline" ? baseline : candidate; const receipts: AppendHeartbeatRunEventResult[] = []; - if (phase === "batch" && arm === "candidate") receipts.push(...await owner.appendHeartbeatRunEvents!(target, values)); + if (phase === "batch" && arm === "candidate") { + for (let offset = 0; offset < values.length; offset += 64) { + const chunk = values.slice(offset, offset + 64); + assert.ok(chunk.reduce((bytes, value) => bytes + Buffer.byteLength(JSON.stringify(value.nativeSource!.canonicalPayload)), 0) <= 64 * 1024); + receipts.push(...await owner.appendHeartbeatRunEvents!(target, chunk)); + } + } else for (const value of values) receipts.push(await owner.appendHeartbeatRunEvent(target, value)); assert.equal(receipts.length, values.length); const expectedCursor = Math.max(...values.map(value => value.nativeSource!.sourceSeq)); @@ -256,7 +294,8 @@ try { } async function timed(f: Fixture, arm: Arm, scenario: string): Promise { - const values = emittedInputs(f, scenario); + const retainedHistoryBefore = f.next - f.sourceOffset - 1; + const values = emittedInputs(f, scenario, groups); const pgBefore = backendCpuMillis(Number(settings!.backend_pid)); const beforeMemory = process.memoryUsage(), cpuBefore = process.cpuUsage(), eluBefore = performance.eventLoopUtilization(); const loop = monitorEventLoopDelay({ resolution: 10 }); @@ -266,17 +305,38 @@ try { const elapsedMs = performance.now() - started; loop.disable(); const cpu = process.cpuUsage(cpuBefore), memory = process.memoryUsage(); + const eventLoopUtilization = performance.eventLoopUtilization(eluBefore).utilization; const pgAfter = backendCpuMillis(Number(settings!.backend_pid)); const newEvents = receipts.filter(receipt => receipt.disposition === "committed").length; - assert.equal(newEvents, scenario === "replay" ? events / 2 : values.length); + assert.equal(newEvents, scenario === "replay" ? values.length / 2 : values.length); assert.equal(receipts.at(-1)!.highestContiguousSourceSeq, scenario === "late-gap" ? f.history + 1 : scenario === "warm-suffix" ? 0 : f.next - 1); + const expectedRows = scenario === "late-gap" ? f.history + 1 : f.next - f.sourceOffset - 1; + const [durable] = await db!.$client.unsafe("select count(*)::integer as events, max(seq)::bigint as last from heartbeat_run_events where run_id=$1::uuid", [f.runId]); + assert.equal(Number(durable!.events), expectedRows, "Independent history read must account for every committed event"); + assert.equal(Number(durable!.last), expectedRows, "Run sequences must not skip or overlap"); + const expectedBySequence = new Map(values.map(value => [value.nativeSource!.sourceSeq, value])); + const sequences = [...expectedBySequence.keys()]; + const persisted = await db!.$client.unsafe(`select source_instance_id, source_event_id, source_seq, + source_payload_sha256, protocol_schema_version, payload from heartbeat_run_events + where run_id=$1::uuid and source_instance_id=$2 and source_seq between $3::bigint and $4::bigint + order by source_seq`, [f.runId, f.source, Math.min(...sequences), Math.max(...sequences)]); + assert.equal(persisted.length, expectedBySequence.size); + for (const row of persisted) { + const expected = expectedBySequence.get(Number(row.source_seq))!; + assert.equal(row.source_event_id, expected.nativeSource!.sourceEventId); + assert.equal(row.source_instance_id, expected.nativeSource!.sourceInstanceId); + assert.equal(row.protocol_schema_version, expected.nativeSource!.protocolSchemaVersion); + assert.equal(row.source_payload_sha256, nativeSha256(expected.nativeSource!.canonicalPayload)); + assert.deepEqual(row.payload, expected.payload, "Durable event payload must match independently of its append receipt"); + } return { elapsedMs, eventsPerSecond: values.length / elapsedMs * 1000, newEvents, duplicateEvents: values.length - newEvents, appCpuMicros: cpu.user + cpu.system, databaseCpuMillis: pgBefore === null || pgAfter === null ? null : pgAfter - pgBefore, heapDeltaBytes: memory.heapUsed - beforeMemory.heapUsed, rssBytes: memory.rss, - eventLoopUtilization: performance.eventLoopUtilization(eluBefore).utilization, - eventLoopDelayMaxMs: loop.count ? loop.max / 1e6 : null }; + eventLoopUtilization, + eventLoopDelayMaxMs: loop.count ? loop.max / 1e6 : null, + retainedHistoryBefore, retainedHistoryAfter: retainedHistoryBefore + newEvents }; } // Qualify before entering the full benchmark. All comparisons preserve the @@ -285,14 +345,16 @@ try { const samples: Record = { baseline: [], candidate: [] }; let a: Fixture | undefined, b: Fixture | undefined; if (scenario !== "late-gap") { - a = await fixture(history, scenario); b = await fixture(history, scenario); await prepare(b, "candidate"); + a = await fixture(history, scenario); b = await fixture(history, scenario); + await prepare(a, "baseline"); await prepare(b, "candidate"); for (let warmup = 0; warmup < 2; warmup++) { await timed(a, "baseline", scenario); await timed(b, "candidate", scenario); } } for (let pair = 0; pair < pairs; pair++) { if (scenario === "late-gap") { - a = await fixture(history, scenario); b = await fixture(history, scenario); await prepare(b, "candidate"); + a = await fixture(history, scenario); b = await fixture(history, scenario); + await prepare(a, "baseline"); await prepare(b, "candidate"); } for (const arm of pair % 2 ? ["candidate", "baseline"] as const : ["baseline", "candidate"] as const) { samples[arm].push(await timed(arm === "baseline" ? a! : b!, arm, scenario)); @@ -313,14 +375,33 @@ try { const [plan] = await db.$client.unsafe(`EXPLAIN (${read ? "ANALYZE, BUFFERS, " : ""}FORMAT JSON) ${query.sql}`, query.params as never[]); plans.push({ sql: query.sql, analyzed: read, plan: plan!["QUERY PLAN"] }); } + const scanNodes = (node: any): any[] => [ + ...(node["Relation Name"] === "heartbeat_run_events" ? [node] : []), + ...(node.Plans ?? []).flatMap(scanNodes), + ]; + const historyScans = (plans as Array<{ analyzed: boolean; plan: Array<{ Plan: unknown }> }>).filter(plan => plan.analyzed) + .flatMap(plan => scanNodes(plan.plan[0].Plan)); + const usesProjection = arm === "baseline" || phase === "aa" ? Boolean(baselineCursorSource) : true; + const examinedRows = (node: any) => (node["Actual Rows"] + (node["Rows Removed by Filter"] ?? 0) + + (node["Rows Removed by Index Recheck"] ?? 0)) * node["Actual Loops"]; + if (usesProjection) { + assert.ok(historyScans.every(node => node["Node Type"] !== "Seq Scan" && examinedRows(node) <= 64), + "A ready cursor must not scan retained event history, including the optimized baseline arm"); + assert.ok(!queries.some(query => /count\(\*\)/i.test(query.sql) && /heartbeat_run_events/.test(query.sql)), + "A ready cursor unexpectedly used the retained-history aggregate fallback"); + } diagnostics[arm] = { sqlCalls: queries.length, transactions: transactions.length, sqlCallsPerEvent: queries.length / values.length, transactionsPerEvent: transactions.length / values.length, transactionTimings: transactions, lockTimingBoundary: "Acquisition query duration includes network/row-read cost; held interval ends after commit. It is not a measurement of pure lock wait.", - persistedEvents: receipts.filter(receipt => receipt.disposition === "committed").length, queries, plans }; + persistedEvents: receipts.filter(receipt => receipt.disposition === "committed").length, + usesProjection, structuralCacheBoundPassed: usesProjection ? true : null, + historyScanRows: historyScans.map(examinedRows), queries, plans }; } const baselineStats = summary(samples.baseline), candidateStats = summary(samples.candidate); const relativePairedDifference = samples.baseline.map((sample, n) => (samples.candidate[n]!.elapsedMs - sample.elapsedMs) / sample.elapsedMs); - (report.results as unknown[]).push({ history, scenario, baseline: baselineStats, candidate: candidateStats, diagnostics, + (report.results as unknown[]).push({ history, scenario, + latencyGate: scenario === "late-gap" ? "structural-only" : history >= 10_000 ? "improvement" : "nonregression", + baseline: baselineStats, candidate: candidateStats, diagnostics, pairedCandidateWins: relativePairedDifference.filter(value => value < 0).length, medianLatencyChangePercent: 100 * (candidateStats.medianMs / baselineStats.medianMs - 1), aaNoiseEnvelopeRelative: phase === "aa" ? Math.max(...relativePairedDifference.map(Math.abs)) : undefined, @@ -335,25 +416,37 @@ try { const noise = JSON.parse(await fs.readFile(arg("noise-report")!, "utf8")); assert.equal(noise.phase, "aa"); assert.equal(noise.incomplete, false); assert.equal(noise.baselineRevision, baselineRevision); assert.equal(noise.workloadSha256, workloadSha256); + assert.equal(noise.candidateSourceHashes["scripts/benchmark-native-event-ingestion.ts"], + (report.candidateSourceHashes as Record)["scripts/benchmark-native-event-ingestion.ts"], + "Measurement protocol changed after A/A calibration; rerun calibration rather than reuse its noise envelope"); report.noiseReport = arg("noise-report"); report.comparison = (report.results as any[]).map(result => { const control = noise.results.find((row: any) => row.history === result.history && row.scenario === result.scenario); assert.ok(control); const improvement = 1 - result.candidate.medianMs / result.baseline.medianMs; + const improvedBeyondNoise = result.pairedCandidateWins >= Math.ceil(pairs * 0.8) && improvement > control.aaNoiseEnvelopeRelative; + const noRegressionBeyondNoise = -improvement <= control.aaNoiseEnvelopeRelative; return { history: result.history, scenario: result.scenario, pairedCandidateWins: result.pairedCandidateWins, + latencyGate: result.latencyGate, improvementRelative: improvement, noiseEnvelopeRelative: control.aaNoiseEnvelopeRelative, - improvedBeyondNoise: result.pairedCandidateWins >= Math.ceil(pairs * 0.8) && improvement > control.aaNoiseEnvelopeRelative }; + improvedBeyondNoise, noRegressionBeyondNoise, + acceptancePassed: result.latencyGate === "structural-only" ? null + : result.latencyGate === "improvement" ? improvedBeyondNoise : noRegressionBeyondNoise }; }); } await save(); - console.log(JSON.stringify(report, null, 2)); } finally { try { - for (const companyId of companiesToRemove) { - await db?.$client.unsafe("delete from heartbeat_run_events where company_id=$1::uuid", [companyId]); - await db?.$client.unsafe("delete from heartbeat_runs where company_id=$1::uuid", [companyId]); - await db?.$client.unsafe("delete from agents where company_id=$1::uuid", [companyId]); - await db?.$client.unsafe("delete from companies where id=$1::uuid", [companyId]); + // The owned cluster is removed as a whole below. Per-company agent deletes + // would repeatedly scan other fixtures through foreign-key checks. An + // explicitly supplied test database retains unrelated data, so remove all + // our event rows first, then their parents in separate phases. + if (externalUrl) { + for (const table of ["heartbeat_run_events", "heartbeat_runs", "agents", "companies"]) { + for (const companyId of companiesToRemove) { + await db?.$client.unsafe(`delete from ${table} where ${table === "companies" ? "id" : "company_id"}=$1::uuid`, [companyId]); + } + } } } finally { try { await db?.$client.end({ timeout: 1 }); } @@ -362,4 +455,8 @@ try { finally { await fs.rm(baselineDir, { recursive: true, force: true }); } } } + report.cleanupComplete = true; + await save(); } +console.log(JSON.stringify(report, null, 2)); +if (process.argv.includes("--require-acceptance")) assertServiceAcceptance(report); diff --git a/scripts/compare-native-ingestion-browser.ts b/scripts/compare-native-ingestion-browser.ts new file mode 100644 index 00000000000..fb15edfb1a3 --- /dev/null +++ b/scripts/compare-native-ingestion-browser.ts @@ -0,0 +1,99 @@ +import assert from "node:assert/strict"; +import { createHash } from "node:crypto"; +import { readFile, writeFile } from "node:fs/promises"; +import path from "node:path"; + +// Each manifest entry points at native-ingestion-evidence JSON attachments +// from two isolated compiled-app runs. Run the variants in alternating order; +// timestamps here are never reconstructed from screenshots or success text. +const arg = (name: string) => process.argv.find(value => value.startsWith(`--${name}=`))?.slice(name.length + 3); +const manifestPath = arg("pairs"); +assert.ok(manifestPath, "Pass --pairs="); +const phase = arg("phase") ?? "compare", expectation = arg("expect") ?? "improvement"; +assert.ok(["aa", "compare"].includes(phase)); +assert.ok(["improvement", "nonregression"].includes(expectation)); +const manifest = JSON.parse(await readFile(manifestPath, "utf8")) as Array<{ baselineFile: string; candidateFile: string }>; +assert.ok(manifest.length >= 5, "At least five actual paired application runs are required"); +const load = async (file: string) => JSON.parse(await readFile(path.resolve(path.dirname(manifestPath), file), "utf8")); +const pairs = await Promise.all(manifest.map(async entry => ({ baseline: await load(entry.baselineFile), candidate: await load(entry.candidateFile) }))); +const names = ["paced", "burst16", "burst128", "burst512", "existing-history", "operator-stop"]; +const median = (values: number[]) => [...values].sort((a, b) => a - b)[Math.floor(values.length / 2)]!; +const first = pairs[0]; +for (const pair of pairs) for (const arm of ["baseline", "candidate"] as const) { + const report = pair[arm]; + assert.equal(report.complete, true); assert.equal(report.qualificationOnly, false, "Smoke cannot qualify performance"); + assert.equal(report.serverArtifact, "compiled", "Performance acceptance requires the compiled server and shipped UI"); + assert.ok(report.serverEntrySha256 && report.measurementSha256); + assert.equal(report.compiledBuildRevision, report.sourceRevision, "Compiled build stamp does not match the measured revision"); + assert.ok(Object.keys(report.compiledHotPathHashes ?? {}).length >= 3, "Missing compiled ingestion/runtime fingerprints"); + assert.equal(report.sourceRevision, first[arm].sourceRevision, "Every sample must cover the same revision per arm"); + assert.equal(report.serverEntrySha256, first[arm].serverEntrySha256, "Compiled artifact changed between samples"); + assert.deepEqual(report.compiledHotPathHashes, first[arm].compiledHotPathHashes, "Compiled ingestion/runtime changed between samples"); + assert.equal(report.measurementSha256, first.baseline.measurementSha256, "Fixture/observer protocol changed between runs"); + if (phase === "aa") assert.equal(report.sourceRevision, first.baseline.sourceRevision, "A/A must use one source revision"); + for (const name of names) { + const result = report.cases.find((item: any) => item.name === name); + assert.ok(result, `Missing fixed workload ${name}`); + assert.equal(result.persistedAfterReload, true); + assert.equal(result.durable.length, result.runs, "Every background run needs independent delta persistence proof"); + for (const proof of result.durable) { + assert.equal(proof.ordered, true); assert.equal(proof.uniqueSourceIdentities, proof.individualDeltas); + if (!result.stop) assert.equal(proof.individualDeltas, result.events); + } + const reference = first.baseline.cases.find((item: any) => item.name === name); + for (const field of ["runs", "events", "intervalMs", "prelude", "stop"]) assert.equal(result[field], reference[field], `Workload changed: ${name}.${field}`); + } +} +const metric = (item: any, key: string): number => key.split(".").reduce((value, part) => value[part], item); +const measurements = names.flatMap(name => { + const observed = first.baseline.cases.find((item: any) => item.name === name); + const keys = ["firstVisibleAfterEmissionMs", "emissionToVisible.medianMs", "drainThroughTerminalMs", + ...(observed.emissionToVisible.p95Ms === undefined ? [] : ["emissionToVisible.p95Ms", "intendedToVisible.p95Ms"]), + ...(observed.stop ? ["stopThroughTerminalMs", "stopThroughAcknowledgementMs"] : [])]; + return keys.map(key => { + const samples = pairs.map(pair => { + const baseline = metric(pair.baseline.cases.find((item: any) => item.name === name), key); + const candidate = metric(pair.candidate.cases.find((item: any) => item.name === name), key); + assert.ok(Number.isFinite(baseline) && baseline > 0 && Number.isFinite(candidate) && candidate >= 0, `Invalid latency ${name}.${key}`); + return { baseline, candidate, improvementRelative: 1 - candidate / baseline }; + }); + return { name, metric: key, baselineMedianMs: median(samples.map(sample => sample.baseline)), + candidateMedianMs: median(samples.map(sample => sample.candidate)), + medianPairedImprovementRelative: median(samples.map(sample => sample.improvementRelative)), + candidateWins: samples.filter(sample => sample.improvementRelative > 0).length, + aaNoiseEnvelopeRelative: phase === "aa" ? Math.max(...samples.map(sample => Math.abs(sample.improvementRelative))) : undefined, + samples }; + }); +}); +const result: Record = { + scope: "Compiled-app deterministic-provider/browser acceptance; live-provider qualification remains separate", + phase, expectation, pairCount: pairs.length, baselineRevision: first.baseline.sourceRevision, + candidateRevision: first.candidate.sourceRevision, measurementSha256: first.baseline.measurementSha256, + manifestSha256: createHash("sha256").update(await readFile(manifestPath)).digest("hex"), measurements, + primaryLoadedWorkload: "burst512", complete: true, +}; +if (phase === "compare") { + assert.ok(arg("noise-report"), "Comparison requires its frozen A/A --noise-report"); + const noise = JSON.parse(await readFile(arg("noise-report")!, "utf8")); + assert.equal(noise.phase, "aa"); assert.equal(noise.complete, true); + assert.equal(noise.baselineRevision, result.baselineRevision); assert.equal(noise.measurementSha256, result.measurementSha256); + assert.ok(noise.pairCount >= 5); + result.gates = measurements.map(item => { + const control = noise.measurements.find((entry: any) => entry.name === item.name && entry.metric === item.metric); + assert.ok(control); + const improvementRequired = expectation === "improvement" && item.name === "burst512" + && ["emissionToVisible.p95Ms", "intendedToVisible.p95Ms", "drainThroughTerminalMs"].includes(item.metric); + const noRegression = -item.medianPairedImprovementRelative <= control.aaNoiseEnvelopeRelative; + const improvement = item.candidateWins >= Math.ceil(pairs.length * 0.8) + && item.medianPairedImprovementRelative > control.aaNoiseEnvelopeRelative; + return { name: item.name, metric: item.metric, improvementRequired, noiseEnvelopeRelative: control.aaNoiseEnvelopeRelative, + passed: improvementRequired ? improvement : noRegression }; + }); + result.acceptancePassed = result.gates.every((entry: any) => entry.passed); +} +if (arg("output")) await writeFile(arg("output")!, JSON.stringify(result, null, 2)); +console.log(JSON.stringify(result, null, 2)); +if (process.argv.includes("--require-acceptance")) { + assert.equal(phase, "compare", "A/A calibrates noise; it is not performance acceptance"); + assert.equal(result.acceptancePassed, true, "Browser performance acceptance failed; retain results and diagnose"); +} diff --git a/tests/e2e/fixtures/composer-stop-codex-loader.mjs b/tests/e2e/fixtures/composer-stop-codex-loader.mjs index 0815a0c62bf..c7dffc0e27d 100644 --- a/tests/e2e/fixtures/composer-stop-codex-loader.mjs +++ b/tests/e2e/fixtures/composer-stop-codex-loader.mjs @@ -11,7 +11,7 @@ if (process.env.NODE_ENV !== "test" || !command || !isAbsolute(command)) { accessSync(command, constants.X_OK); registerHooks({ load(url, context, nextLoad) { - if (/\/packages\/paperclip-runner\/(src|dist)\/drivers\/codex\/codex-command\.(js|ts)$/.test(url)) { + if (/(?:\/packages\/paperclip-runner\/(?:src|dist)|\/server\/(?:src|dist)\/vendor\/paperclip-runner)\/drivers\/codex\/codex-command\.(?:js|ts)$/.test(url)) { return { format: "module", shortCircuit: true, diff --git a/tests/e2e/fixtures/in-feed-codex.mjs b/tests/e2e/fixtures/in-feed-codex.mjs index 43acbb058e7..0f1a44c47c4 100644 --- a/tests/e2e/fixtures/in-feed-codex.mjs +++ b/tests/e2e/fixtures/in-feed-codex.mjs @@ -2,6 +2,7 @@ // Deterministic provider only. The production Rust runner, tool authority, and // authenticated MCP gateway still execute every tool and enforce access. import { randomUUID } from 'node:crypto'; +import { once } from 'node:events'; import { readFileSync } from 'node:fs'; import { join, resolve } from 'node:path'; import { createInterface } from 'node:readline'; @@ -10,6 +11,9 @@ let threadId = `fixture-${randomUUID()}`; let turnId, toolSequence = 0, declined = false; let helloProbe = false; const recoveryFixture = process.env.PAPERCLIP_RECOVERY_FIXTURE === "1"; +const ingestionFixture = process.argv.includes('--ingestion-stream-fixture'); +let ingestionPlan = null; +let ingestionGeneration = 0; let currentObjective = ""; let emitCeoLineage = false; let completionContract = { revision: "1", criterionIds: ["objective"] }; @@ -20,6 +24,57 @@ const call = (tool, args) => new Promise((resolve, reject) => { pending.set(id, { resolve, reject }); send({ id, method: 'item/tool/call', params: { threadId, turnId, callId: id, tool, arguments: args } }); }); +async function sendIngestion(value) { + if (!process.stdout.write(`${JSON.stringify(value)}\n`)) await once(process.stdout, 'drain'); +} + +async function streamIngestion() { + if (!ingestionPlan) throw new Error('Missing INGESTION_PERF fixture plan'); + const plan = ingestionPlan, generation = ++ingestionGeneration; + const gate = new URL(plan.gate); + if (gate.protocol !== 'http:' || !['127.0.0.1', 'localhost'].includes(gate.hostname) + || !gate.pathname.startsWith('/native-ingestion-gate/') || gate.username || gate.password) { + throw new Error('Ingestion fixture requires its local benchmark gate'); + } + if (plan.prelude) { + const preludeId = `prelude-${turnId}`; + await sendIngestion({ method: 'item/started', params: { threadId, turnId, + item: { id: preludeId, type: 'agentMessage', phase: 'commentary', text: '' } } }); + for (let index = 0; index < plan.prelude; index++) { + await sendIngestion({ method: 'item/agentMessage/delta', params: { threadId, turnId, itemId: preludeId, + delta: index === plan.prelude - 1 ? ` PRELUDE_${plan.nonce}_READY ` : '.' } }); + } + await sendIngestion({ method: 'item/completed', params: { threadId, turnId, + item: { id: preludeId, type: 'agentMessage', phase: 'commentary', + text: `${'.'.repeat(Math.max(0, plan.prelude - 1))} PRELUDE_${plan.nonce}_READY ` } } }); + } + gate.searchParams.set('participant', plan.nonce); + const response = await fetch(gate, { signal: AbortSignal.timeout(300_000) }); + if (!response.ok) throw new Error(`Ingestion gate returned ${response.status}`); + const { epochMs } = await response.json(); + if (!Number.isSafeInteger(epochMs)) throw new Error('Invalid ingestion schedule'); + const itemId = `ingestion-${turnId}`; + await sendIngestion({ method: 'item/started', params: { threadId, turnId, + item: { id: itemId, type: 'agentMessage', phase: 'final_answer', text: '' } } }); + let text = ''; + for (let index = 0; index < plan.events; index++) { + const intended = epochMs + index * plan.intervalMs; + const wait = intended - Date.now(); + if (wait > 0) await new Promise(resolve => setTimeout(resolve, wait)); + if (generation !== ingestionGeneration) return; + const delta = `INGEST_${plan.nonce}_${index}_${intended}_${Date.now()} `; + text += delta; + await sendIngestion({ method: 'item/agentMessage/delta', params: { threadId, turnId, itemId, delta } }); + } + if (generation !== ingestionGeneration) return; + await call('report_progress', { idempotencyKey: `ingestion-${plan.nonce}`, body: `Ingestion fixture emitted ${plan.events} measured chunks.` }); + if (generation !== ingestionGeneration) return; + await finish(`Ingestion fixture ${plan.nonce} completed.`, `fixture:ingestion-${plan.nonce}`); + if (generation !== ingestionGeneration) return; + await sendIngestion({ method: 'item/completed', params: { threadId, turnId, + item: { id: itemId, type: 'agentMessage', phase: 'final_answer', text } } }); + await sendIngestion({ method: 'turn/completed', params: { threadId, turn: { id: turnId, status: 'completed' } } }); +} function unwrap(result) { const texts = result?.contentItems ?? result?.content ?? []; for (const item of texts) { @@ -61,6 +116,7 @@ async function execute() { send({ method: 'turn/completed', params: { threadId, turn: { id: turnId, status: 'completed' } } }); return; } + if (ingestionFixture) return streamIngestion(); if (process.argv.includes('--task-retry-fixture')) { const text = 'The saved result is confirmed. No external publication was repeated.'; await call('report_progress', { idempotencyKey: 'saved-result-confirmation', body: text }); @@ -156,13 +212,21 @@ createInterface({ input: process.stdin }).on('line', (line) => { if (recoveryFixture) send({ method: 'configWarning', params: { message: 'Recovery fixture startup notice before the first turn' } }); if (method === 'thread/resume' && message.params?.threadId) threadId = message.params.threadId; if (message.params?.completionContract) completionContract = message.params.completionContract; - const model = process.argv.includes('--completion-stream-fixture') ? message.params?.model ?? 'in-feed-fixture' : 'in-feed-fixture'; + const model = process.argv.includes('--completion-stream-fixture') || ingestionFixture ? message.params?.model ?? 'in-feed-fixture' : 'in-feed-fixture'; send({ id, result: { model, modelProvider: 'fixture', thread: { id: threadId, sessionId: threadId } } }); } else if (method === 'thread/read') send({ id, result: { thread: { id: threadId, turns: [] } } }); else if (method === 'turn/start') { - helloProbe = process.argv.includes('--completion-stream-fixture') + helloProbe = (process.argv.includes('--completion-stream-fixture') || ingestionFixture) && JSON.stringify(message.params?.input).includes('Respond only with hello. Do not use tools or inspect files.'); + if (ingestionFixture && !helloProbe) { + const match = JSON.stringify(message.params?.input).match(/INGESTION_PERF nonce=([a-z0-9]+) events=(\d+) intervalMs=(\d+) prelude=(\d+) gate=([^\s"\\]+)/); + if (!match) throw new Error('Missing explicit ingestion fixture plan'); + ingestionPlan = { nonce: match[1], events: Number(match[2]), intervalMs: Number(match[3]), prelude: Number(match[4]), gate: match[5] }; + if (ingestionPlan.events < 1 || ingestionPlan.events > 4096 || ingestionPlan.intervalMs > 1000 || ingestionPlan.prelude > 10_000) { + throw new Error('Ingestion fixture exceeds its bounded workload'); + } + } emitCeoLineage = JSON.stringify(message.params).includes('CEO descendant fixture'); declined = /connection_intent/.test(JSON.stringify(message.params)) && /rejected/.test(JSON.stringify(message.params)); for (const part of message.params?.input ?? []) { @@ -184,5 +248,9 @@ createInterface({ input: process.stdin }).on('line', (line) => { process.stderr.write(`In-feed fixture: ${error.message}\n`); send({ method: 'turn/completed', params: { threadId, turn: { id: turnId, status: 'failed', error: { message: error.message } } } }); }), 50); + } else if (ingestionFixture && method === 'turn/interrupt') { + ingestionGeneration += 1; + send({ id, result: {} }); + send({ method: 'turn/completed', params: { threadId, turn: { id: turnId, status: 'interrupted' } } }); } else if (id != null) send({ id, result: {} }); }); diff --git a/tests/e2e/fixtures/native-ingestion-bin/codex b/tests/e2e/fixtures/native-ingestion-bin/codex new file mode 100755 index 00000000000..93acfb1aaef --- /dev/null +++ b/tests/e2e/fixtures/native-ingestion-bin/codex @@ -0,0 +1,2 @@ +#!/bin/sh +exec node "$(dirname "$0")/../in-feed-codex.mjs" --ingestion-stream-fixture "$@" diff --git a/tests/e2e/native-event-ingestion.journey.ts b/tests/e2e/native-event-ingestion.journey.ts new file mode 100644 index 00000000000..cecd73d2338 --- /dev/null +++ b/tests/e2e/native-event-ingestion.journey.ts @@ -0,0 +1,250 @@ +import { createServer, type ServerResponse } from "node:http"; +import { randomUUID, createHash } from "node:crypto"; +import { execFileSync } from "node:child_process"; +import { readFile } from "node:fs/promises"; +import path from "node:path"; +import { expect, test } from "@playwright/test"; +import { json } from "./agent-chat.shared"; + +type Visible = { nonce: string; index: number; intendedMs: number; emittedMs: number; visibleMs: number }; +type Case = { name: string; runs: number; events: number; intervalMs: number; prelude: number }; +const smoke = process.env.PAPERCLIP_INGESTION_SMOKE === "1"; +const cases: Case[] = smoke ? [{ name: "qualification", runs: 1, events: 16, intervalMs: 25, prelude: 0 }] : [ + { name: "paced", runs: 1, events: 128, intervalMs: 25, prelude: 0 }, + { name: "burst16", runs: 1, events: 16, intervalMs: 0, prelude: 0 }, + { name: "burst128", runs: 10, events: 128, intervalMs: 0, prelude: 0 }, + { name: "burst512", runs: 100, events: 512, intervalMs: 0, prelude: 0 }, + { name: "existing-history", runs: 10, events: 128, intervalMs: 0, prelude: 1000 }, +]; +const stats = (values: number[]) => { + const sorted = [...values].sort((a, b) => a - b); + return { count: sorted.length, medianMs: sorted[Math.floor(sorted.length / 2)], minMs: sorted[0], maxMs: sorted.at(-1), + ...(values.length >= 100 ? { p95Ms: sorted[Math.ceil(sorted.length * 0.95) - 1] } : {}) }; +}; + +test("native ingestion reaches the shipped UI promptly and survives reload", async ({ page, request }, info) => { + const pending = new Map>(); + const gate = createServer((req, res) => { + const url = new URL(req.url!, "http://127.0.0.1"); + const key = url.pathname.replace("/native-ingestion-gate/", ""); + const participant = url.searchParams.get("participant"); + if (req.method !== "GET" || !pending.has(key) || !participant || !/^[a-z0-9]+$/.test(participant)) { + res.writeHead(404).end(); return; + } + pending.get(key)!.set(participant, res); + res.on("close", () => pending.get(key)?.delete(participant)); + }); + await new Promise(resolve => gate.listen(0, "127.0.0.1", resolve)); + const address = gate.address(); + if (!address || typeof address === "string") throw new Error("Local ingestion gate failed"); + const base = `http://127.0.0.1:${address.port}/native-ingestion-gate/`; + await page.addInitScript(() => { + const seen = new Set(), visible: Visible[] = []; + Object.assign(window, { nativeIngestionVisible: visible }); + let scheduled = false, dirty = false; + const snapshot = () => { + const text = document.body?.innerText ?? ""; + const pattern = /INGEST_([a-z0-9]+)_(\d+)_(\d+)_(\d+)/g; + const entries: Omit[] = []; + for (const match of text.matchAll(pattern)) { + const key = `${match[1]}:${match[2]}`; + if (seen.has(key)) continue; + entries.push({ nonce: match[1], index: Number(match[2]), intendedMs: Number(match[3]), emittedMs: Number(match[4]) }); + } + return entries; + }; + const schedule = () => { + dirty = true; + if (scheduled) return; + scheduled = true; + requestAnimationFrame(() => { + dirty = false; + const entries = snapshot(); + requestAnimationFrame(() => { + // Only stamp the previous frame's snapshot. A token inserted + // between frames has not had this snapshot's paint opportunity. + const visibleMs = Date.now(); + for (const entry of entries) { + const key = `${entry.nonce}:${entry.index}`; + if (seen.has(key)) continue; + seen.add(key); visible.push({ ...entry, visibleMs }); + } + scheduled = false; + if (dirty) schedule(); + }); + }); + }; + const install = () => { + new MutationObserver(schedule).observe(document.body, { subtree: true, childList: true, characterData: true }); + schedule(); + }; + if (document.body) install(); else addEventListener("DOMContentLoaded", install, { once: true }); + }); + const company = await json(await request.post("/api/companies", { data: { name: `Ingestion ${Date.now()}` } })); + const original = await json(await request.get("/api/instance/settings/experimental")); + const report: Record = { + dependency: "Deterministic Codex app-server fixture; real application/runtime/port/PostgreSQL/live WebSocket/shipped UI", + variant: process.env.PAPERCLIP_INGESTION_VARIANT ?? "unlabelled", qualificationOnly: smoke, + startedAtMs: Date.now(), + emittedClock: "Provider process epoch; original schedule retained even if stdout backpressures", + observedClock: "Browser epoch after a DOM update and one paint opportunity", + cases: [], companyId: company.id, + health: await json(await request.get("/api/health")), + serverArtifact: process.env.PAPERCLIP_INGESTION_SERVER_ARTIFACT, + }; + const appRoot = process.env.PAPERCLIP_INGESTION_APP_ROOT ?? path.resolve(import.meta.dirname, "../.."); + report.sourceRevision = execFileSync("git", ["rev-parse", "HEAD"], { cwd: appRoot, encoding: "utf8" }).trim(); + if (report.serverArtifact === "compiled") { + const stamp = JSON.parse(await readFile(path.join(appRoot, "server/dist/build-info.json"), "utf8")); + report.compiledBuildRevision = stamp.commit; + if (!smoke) expect(stamp.commit, "Build the revision being measured before accepting compiled-app evidence").toBe(report.sourceRevision); + report.serverEntrySha256 = createHash("sha256").update(await readFile(path.join(appRoot, "server/dist/index.js"))).digest("hex"); + report.runtimeArtifact = "Compiled server and vendored runner; TSX resolves workspace DB/shared source exports only; static built UI"; + const hotPaths = ["server/dist/services/heartbeat-run-events.js", "server/dist/services/native-runtime/paperclip-control-plane-port.js", + "server/dist/vendor/paperclip-runner/native-session-runtime.js"]; + report.compiledHotPathHashes = Object.fromEntries(await Promise.all(hotPaths.map(async file => [file, + createHash("sha256").update(await readFile(path.join(appRoot, file))).digest("hex")]))); + } + report.fixtureSha256 = createHash("sha256").update(await readFile(path.resolve(import.meta.dirname, "fixtures/in-feed-codex.mjs"))).digest("hex"); + const protocolFiles = ["native-event-ingestion.journey.ts", "playwright-native-ingestion.config.ts", "fixtures/in-feed-codex.mjs", + "fixtures/native-ingestion-bin/codex", "fixtures/composer-stop-codex-loader.mjs"]; + report.measurementSha256 = createHash("sha256").update(Buffer.concat(await Promise.all(protocolFiles.map(file => readFile(path.resolve(import.meta.dirname, file)))))).digest("hex"); + try { + await json(await request.patch("/api/instance/settings/experimental", { data: { enableNativeRunner: true, enableWorktreeRunExecution: true } })); + await json(await request.post(`/api/companies/${company.id}/user-secret-definitions`, { data: { key: "OPENAI_API_KEY", name: "Fixture credential" } })); + await json(await request.post(`/api/companies/${company.id}/me/user-secrets`, { data: { definitionKey: "OPENAI_API_KEY", value: "not-a-real-model-key" } })); + const agents = []; + const agentCount = Math.ceil(Math.max(...cases.map(item => item.runs)) / 50); + for (let index = 0; index < agentCount; index++) { + const agent = await json(await request.post(`/api/companies/${company.id}/agents`, { data: { + name: `Ingestion ${index + 1}`, role: index === 0 ? "ceo" : "engineer", adapterType: "paperclip_runner", + adapterConfig: { provider: "codex", model: "gpt-5.1-codex-mini", env: { + OPENAI_API_KEY: { type: "user_secret_ref", key: "OPENAI_API_KEY", version: "latest", required: true }, + } }, runtimeConfig: { heartbeat: { enabled: false, wakeOnDemand: true, maxConcurrentRuns: 50 } }, + } })); + await expect.poll(async () => (await json(await request.get(`/api/agents/${agent.id}`))).lifecycleState, + { timeout: 60_000 }).toBe("ready"); + agents.push(agent); + } + + async function perform(workload: Case, stop = false) { + const key = randomUUID().replaceAll("-", ""), nonce = randomUUID().replaceAll("-", ""); + pending.set(key, new Map()); + const prompt = (token: string, prelude: number) => `INGESTION_PERF nonce=${token} events=${workload.events} intervalMs=${workload.intervalMs} prelude=${prelude} gate=${base}${key}`; + const before = new Set((await json(await request.get(`/api/companies/${company.id}/issues`))).map((issue: { id: string }) => issue.id)); + await page.goto(`/${company.issuePrefix}/agents/${agents[0].id}/overview`); + await page.getByRole("link", { name: "Tasks", exact: true }).click(); + await page.getByRole("button", { name: "New Task", exact: true }).last().click(); + const dialog = page.getByRole("dialog"); + await dialog.getByRole("textbox", { name: "editable markdown" }).fill(prompt(nonce, workload.prelude)); + await expect(dialog.getByRole("button", { name: "Select assignee" })).toContainText(agents[0].name); + await dialog.getByRole("button", { name: "Create task", exact: true }).click(); + let issue: any; + await expect.poll(async () => { + issue = (await json(await request.get(`/api/companies/${company.id}/issues`))).find((item: { id: string; description?: string }) => !before.has(item.id) && item.description?.includes(nonce)); + return Boolean(issue); + }).toBe(true); + const issueIds = [issue.id]; + const issuedTokens = new Map([[issue.id, nonce]]); + for (let index = 1; index < workload.runs; index++) { + const backgroundToken = randomUUID().replaceAll("-", ""); + const background = await json(await request.post(`/api/companies/${company.id}/issues`, { data: { + title: `Ingestion background ${key} ${index}`, description: prompt(backgroundToken, 0), + status: "backlog", assigneeAgentId: agents[Math.floor(index / 50)].id, + } })); + issueIds.push(background.id); + issuedTokens.set(background.id, backgroundToken); + } + for (let offset = 1; offset < issueIds.length; offset += 10) { + await Promise.all(issueIds.slice(offset, offset + 10).map(async id => json(await request.patch(`/api/issues/${id}`, { data: { status: "todo" } })))); + } + const route = `/${company.issuePrefix}/issues/${issue.identifier}`; + await page.goto(route); + await expect.poll(() => pending.get(key)!.size, { timeout: 180_000 }).toBe(workload.runs); + if (workload.prelude) await expect(page.getByText(`PRELUDE_${nonce}_READY`, { exact: false }).first()).toBeVisible({ timeout: 120_000 }); + const epochMs = Date.now() + 1000; + const participants = [...pending.get(key)!.keys()]; + for (const response of pending.get(key)!.values()) response.writeHead(200, { "Content-Type": "application/json" }).end(JSON.stringify({ epochMs })); + const observed = () => page.evaluate(token => (window as unknown as { nativeIngestionVisible: Visible[] }).nativeIngestionVisible.filter(item => item.nonce === token), nonce); + await expect.poll(async () => (await observed()).length, { timeout: 120_000 }).toBeGreaterThanOrEqual(stop ? 3 : workload.events); + let stopRequestedAt: number | undefined; + if (stop) { + const button = page.getByRole("button", { name: "Stop", exact: true }); + await expect(button).toBeVisible(); + stopRequestedAt = Date.now(); + await button.click(); + } + let runs: any[] = []; + await expect.poll(async () => { + runs = (await json(await request.get(`/api/companies/${company.id}/heartbeat-runs?limit=1000`))).filter((run: { issueId?: string; nativeIssueId?: string; contextSnapshot?: { issueId?: string; taskId?: string } }) => issueIds.includes(run.issueId ?? run.nativeIssueId ?? run.contextSnapshot?.issueId ?? run.contextSnapshot?.taskId)); + return runs.length === issueIds.length && runs.every(run => ["succeeded", "cancelled", "failed"].includes(run.status)); + }, { timeout: 180_000 }).toBe(true); + const settledAt = Date.now(), visible = await observed(); + expect(runs.every(run => run.runtimeMode === "native")).toBe(true); + expect(runs.every(run => run.status === (stop ? "cancelled" : "succeeded"))).toBe(true); + if (stop) await expect.poll(async () => + (await json(await request.get(`/api/heartbeat-runs/${runs[0].id}`))).resultJson?.nativeCancellation?.dispatchState, + { timeout: 30_000 }).toBe("acknowledged"); + const acknowledgedAt = stop ? Date.now() : undefined; + if (!stop) expect(new Set(visible.map(item => item.index)).size).toBe(workload.events); + const result = { ...workload, stop, nonce, runIds: runs.map(run => run.id), participants: participants.length, visible, + firstVisibleAfterEmissionMs: visible[0].visibleMs - visible[0].emittedMs, + emissionToVisible: stats(visible.map(item => item.visibleMs - item.emittedMs)), + intendedToVisible: stats(visible.map(item => item.visibleMs - item.intendedMs)), + providerLateness: stats(visible.map(item => item.emittedMs - item.intendedMs)), + drainThroughTerminalMs: settledAt - epochMs, + ...(stopRequestedAt === undefined ? {} : { stopThroughTerminalMs: settledAt - stopRequestedAt, + stopThroughAcknowledgementMs: acknowledgedAt! - stopRequestedAt }) }; + await page.screenshot({ path: info.outputPath(`${workload.name}${stop ? "-stop" : ""}.png`), fullPage: true }); + await page.reload(); + if (!stop) await expect(page.getByText(`INGEST_${nonce}_${workload.events - 1}_`, { exact: false }).first()).toBeVisible(); + else await expect(page.getByRole("button", { name: "Stop", exact: true })).toHaveCount(0); + const comments = await json(await request.get(`/api/issues/${issue.id}/comments`)); + if (!stop) expect(JSON.stringify(comments)).toContain(`INGEST_${nonce}_${workload.events - 1}_`); + else { + const lastVisible = visible.at(-1)!; + await expect(page.getByText(`INGEST_${nonce}_${lastVisible.index}_`, { exact: false }).first()).toBeVisible(); + } + const durable = []; + for (const run of runs) { + const taskId = run.issueId ?? run.nativeIssueId ?? run.contextSnapshot?.issueId ?? run.contextSnapshot?.taskId; + const token = issuedTokens.get(taskId)!; + expect(token).toBeTruthy(); + const deltas: Array<{ index: number; identity: string }> = []; + let afterSeq = 0; + while (true) { + const rows = await json(await request.get(`/api/heartbeat-runs/${run.id}/events?afterSeq=${afterSeq}&limit=1000`)); + if (!rows.length) break; + for (const row of rows) { + const event = row.payload?.prpEvent; + if (event?.eventType !== "item.delta") continue; + for (const match of String(event.payload?.text ?? "").matchAll(new RegExp(`INGEST_${token}_(\\d+)_(\\d+)_(\\d+)`, "g"))) { + deltas.push({ index: Number(match[1]), identity: `${event.sourceInstanceId}:${event.sourceEventId}:${event.sourceSeq}` }); + } + } + const next = Math.max(...rows.map((row: { seq: number }) => row.seq)); + expect(next).toBeGreaterThan(afterSeq); afterSeq = next; + } + expect(new Set(deltas.map(item => item.identity)).size).toBe(deltas.length); + expect(deltas.map(item => item.index)).toEqual(Array.from({ length: stop ? deltas.length : workload.events }, (_, index) => index)); + if (stop) for (const item of visible) expect(deltas.some(delta => delta.index === item.index)).toBe(true); + durable.push({ runId: run.id, token, individualDeltas: deltas.length, uniqueSourceIdentities: deltas.length, ordered: true }); + } + (report.cases as unknown[]).push({ ...result, durable, persistedAfterReload: true }); + await info.attach(`${workload.name}${stop ? "-stop" : ""}-metrics`, { body: JSON.stringify(result, null, 2), contentType: "application/json" }); + pending.delete(key); + } + for (const workload of cases) await perform(workload); + await perform({ name: "operator-stop", runs: 1, events: 512, intervalMs: 100, prelude: 0 }, true); + report.complete = true; + } finally { + await info.attach("native-ingestion-evidence", { body: JSON.stringify(report, null, 2), contentType: "application/json" }); + for (const requests of pending.values()) for (const response of requests.values()) response.writeHead(503).end(); + gate.closeAllConnections(); + await new Promise(resolve => gate.close(() => resolve())); + const live = await json(await request.get(`/api/companies/${company.id}/live-runs`)).catch(() => []); + for (const run of live) await request.post(`/api/heartbeat-runs/${run.id}/cancel`); + await request.patch("/api/instance/settings/experimental", { data: { enableNativeRunner: original.enableNativeRunner, + enableWorktreeRunExecution: original.enableWorktreeRunExecution } }); + } +}); diff --git a/tests/e2e/playwright-native-ingestion.config.ts b/tests/e2e/playwright-native-ingestion.config.ts new file mode 100644 index 00000000000..981d719fca5 --- /dev/null +++ b/tests/e2e/playwright-native-ingestion.config.ts @@ -0,0 +1,37 @@ +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { defineConfig, type ReporterDescription } from "@playwright/test"; +import base from "./playwright.config"; + +const server = base.webServer as Exclude; +const evidence = process.env.PAPERCLIP_INGESTION_EVIDENCE_DIR; +const appRoot = path.resolve(process.env.PAPERCLIP_INGESTION_APP_ROOT ?? path.resolve(import.meta.dirname, "../..")); +const command = process.env.PAPERCLIP_INGESTION_SERVER_COMMAND ?? server?.command; +const compiled = Boolean(command?.includes("server/dist/index.js") && !command.includes("server/src/index") && !/onboard[^;&]*--run/.test(command)); +if (process.env.PAPERCLIP_INGESTION_SMOKE !== "1" && !compiled) { + throw new Error("Full ingestion measurements require a compiled server/dist/index.js command; source CLI is qualification smoke only"); +} +process.env.PAPERCLIP_INGESTION_SERVER_ARTIFACT = compiled ? "compiled" : "source-qualification"; +const providerHome = fs.mkdtempSync(path.join(os.tmpdir(), "ingestion-provider-")); +export default defineConfig({ + ...base, + testMatch: "native-event-ingestion.journey.ts", + timeout: 600_000, + ...(evidence ? { outputDir: path.join(evidence, "test-results"), + reporter: [["list"], ["json", { outputFile: path.join(evidence, "playwright.json") }]] as ReporterDescription[] } : {}), + use: { ...base.use, trace: "retain-on-failure", actionTimeout: 20_000 }, + webServer: { + ...server, + cwd: appRoot, + command, + timeout: 240_000, + env: { + ...server?.env, + PAPERCLIP_STOP_CODEX_COMMAND: path.resolve(import.meta.dirname, "fixtures/native-ingestion-bin/codex"), + CODEX_HOME: providerHome, + OPENAI_API_KEY: "ingestion-fixture-key", + NODE_OPTIONS: `${server?.env?.NODE_OPTIONS ?? ""} --import=${path.resolve(import.meta.dirname, "fixtures/composer-stop-codex-loader.mjs")}`, + }, + }, +});