test(perf): qualify native ingestion with matched browser evidence

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

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<string, unknown> = {
   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<unknown>, ...options: any[]) => {
+    connection.transaction = ((callback: (tx: Db) => Promise<unknown>, options?: Parameters<Db["transaction"]>[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<string, string>([[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<string, unknown>;
+    const module = await import(usesBaseline
+      ? pathToFileURL(path.join(baselineDir, cursorSourcePath)).href
+      : pathToFileURL(path.join(root, cursorSourcePath)).href) as Record<string, unknown>;
     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<Sample> {
-    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<Arm, Sample[]> = { 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<string, string>)["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);

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=<JSON manifest of baselineFile/candidateFile 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<string, any> = {
+  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");
+}

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,

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: {} });
 });

tests/e2e/fixtures/native-ingestion-bin/codex

@@ -0,0 +1,2 @@
+#!/bin/sh
+exec node "$(dirname "$0")/../in-feed-codex.mjs" --ingestion-stream-fixture "$@"

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<string, Map<string, ServerResponse>>();
+  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<void>(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<string>(), 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<Visible, "visibleMs">[] = [];
+      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<string, unknown> = {
+    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<string, string>([[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<void>(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 } });
+  }
+});

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<typeof base.webServer, unknown[]>;
+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")}`,
+    },
+  },
+});