test(perf): qualify bounded browser markers with source clocks

doc/plans/2026-11-03-native-event-ingestion.md

@@ -267,3 +267,85 @@ the combined change still requires the loaded improvement. Comparison rejects
 smoke, source-server evidence, mismatched artifacts/protocols, missing background
 delta proof and incomplete runs. Live-provider walkthroughs remain a separate
 integration gate.
+
+### Browser protocol v2, registered before fresh A/A
+
+The first full v1 baseline qualification failed in the paced 1×128 case. Its
+paint observer accumulated 127 markers and missed index 127. Independent,
+read-only PostgreSQL inspection found all 128 deltas with unique source
+identities and no gaps. The run succeeded. The normalized completed item and
+final comment were each exactly 4,000 characters with a truncation marker and
+only complete marker indices 0–55. Raw Playwright DOM snapshots confirmed the
+same 4,000-character terminal text, after streamed DOM text reached 9,033
+characters. This is not an accessibility-snapshot truncation assumption.
+
+The existing Rust provider normalizer bounds completed text at 4,000 Unicode
+characters. UI projection treats a completed item as authoritative and replaces
+its streamed deltas. Long in-band diagnostic clocks made the fixture's completed
+answer exceed that preexisting bound. No application or normalizer policy is
+changed to make the experiment pass. V1 failed artifacts, trace, decoded paint
+observations and SELECT-only database receipts remain evidence of an unqualified
+experiment; none establishes a speedup.
+
+V2 changes the browser measurement protocol for every arm before any fresh A/A:
+
+- A visible chunk is `[rrII] `: a unique two-base36-character run code followed by
+  its base36 index. A full 512-chunk answer is 3,548 ASCII characters and bytes.
+  The existing-history fixture keeps its 1,000 actual single-character prelude
+  deltas and sentinel; that prelude plus 128 compact chunks totals 1,907 characters.
+  Explicit checks keep these completed payloads below the unchanged 4,000 bound.
+  Before releasing measured emission, an independent normal event-API read verifies
+  all 1,000 prelude deltas and source identities; the report records that pre-gate proof.
+- Full nonce, index, original intended emission time and original provider-process
+  emission time travel on one streamed HTTP clock channel per provider. The
+  channel opens and sends its hello before the common gate releases. There is no
+  per-token HTTP request or wait for clock drain/response before writing text.
+- The browser still captures a DOM snapshot in one animation frame and assigns
+  that snapshot's paint-opportunity timestamp in the next. It records that time
+  immediately, even when the clock channel has not arrived. It joins original
+  clocks afterward, preserving the original browser timestamp.
+- Every visible marker and every independently read durable delta must match its
+  original full nonce/code/index clock. Missing, duplicated or conflicting clocks
+  fail. Stop may terminate the clock stream with a partial unacknowledged raw
+  tail, but no observed or database-acknowledged prefix receives a synthetic clock.
+- The provider completes its assistant item before invoking semantic progress
+  and Finish tools. Tool/result/terminal output does not wait for the clock upload
+  response. Standalone qualification deliberately withholds that response through
+  terminal settlement and Stop acknowledgement.
+- All 16/128/512 event counts, 1/10/100 run counts, emission schedules, history
+  counts, payload-identity checks, drainage checks, timeouts and performance gates
+  remain fixed. Removing diagnostic clocks makes the browser text payload smaller;
+  this is explicit and matched across all v2 arms. The owner-path workload retains
+  its original 128-byte text payload and unchanged source/gate hashes.
+- Timed A/A and comparison runs disable Playwright trace, video and screencast.
+  Their metadata records the actual configured capture modes; the comparator
+  rejects recording, legacy v1, smoke, profile and incomplete evidence. The two
+  animation-frame observer remains declared measurement overhead. Manual
+  screenshots occur after timed terminal drainage. Separate diagnostics may
+  capture browser traces and app-only Node CPU profiles and stay ineligible.
+- A failure attachment retains the active case, requested and ready participants,
+  observed marker timestamps, raw rendered text, DOM marker inventory, received
+  source clocks, stream errors and the latest independent durable read.
+
+The separate browser harness H is based on application ref
+`7922802f069ab3b39e9f79368c46615ba2201ba9` with only the v2 fixture, observer,
+recording metadata, comparator, tests and this record changed. H uses explicitly
+identified driver-only tooling dependencies. Each measured application retains
+its own independent locked dependencies, clean source checkout and matching
+compiled build stamp: B0 `83b5f7d486e70e484acd9a41bc957054f4e10175`, B1
+`7922802f069ab3b39e9f79368c46615ba2201ba9`, B2
+`6afeba4edafd47120e8117f736641d5424b27d4a`. Reports retain H's own revision,
+source status, individual protocol hashes and lockfile hash. No application
+build stamp is rewritten for a test-only change. Exact H commit and combined
+protocol hash are frozen with the qualification receipts before new A/A starts.
+
+Qualification must pass the source-clock protocol smoke and the full B0 browser
+matrix, including actual 100-participant readiness and independent background
+persistence. Fresh A/A still precedes any candidate comparison. The owner-path
+20% maximum paired variance / 10% median drift limits and the requirement that
+four individual candidate pairs exceed observed noise are unchanged. Failed v1
+samples are never reused as v2 evidence.
+
+Registered v2 browser protocol SHA256: `5498ed8dfc7139285aeda582c540eb572f07a1ad3e73de6e0f12c5af4c45e0ba`.
+Comparator SHA256: `d8cf6f55339ad811ca1743f4bc73b74afa3a321b93e17ef913fc0339b47b4d7d`. Individual source hashes and
+H's exact committed revision are retained with each qualification/execution receipt.

scripts/compare-native-ingestion-browser.ts

@@ -22,9 +22,13 @@ const median = (values: number[]) => [...values].sort((a, b) => a - b)[Math.floo
 const first = pairs[0];
 for (const pair of pairs) for (const arm of ["baseline", "candidate"] as const) {
   const report = pair[arm];
+  assert.equal(report.protocolVersion, "v2-compact-markers-original-clock-stream", "Fresh browser acceptance requires the qualified v2 clock protocol");
   assert.equal(report.complete, true); assert.equal(report.qualificationOnly, false, "Smoke cannot qualify performance");
   assert.notEqual(report.diagnosticOnly, true, "Diagnostic profiles cannot qualify unprofiled performance");
   assert.notEqual(report.performanceAcceptanceEligible, false, "Profile-only evidence cannot qualify performance");
+  assert.equal(report.browserRecording?.trace, "off", "Acceptance requires browser trace capture off");
+  assert.equal(report.browserRecording?.video, "off", "Acceptance requires browser video capture off");
+  assert.equal(report.browserRecording?.screencast, false, "Acceptance requires browser screencast capture off");
   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");
@@ -38,9 +42,17 @@ for (const pair of pairs) for (const arm of ["baseline", "candidate"] as const)
     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.matchedVisibleClocks, result.visible?.length, "Every observed marker needs its original provider clock");
+    assert.ok(result.matchedVisibleClocks >= (result.stop ? 3 : result.events), "Missing foreground render/clock observations");
+    if (result.prelude) {
+      assert.equal(result.preludeProof?.individualDeltas, result.prelude, "Existing history requires independently stored prelude deltas");
+      assert.equal(result.preludeProof?.uniqueSourceIdentities, result.prelude);
+      assert.ok(result.preludeProof?.verifiedAtMs < result.epochMs, "Verify retained prelude before releasing measured emission");
+    }
     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);
+      assert.equal(proof.matchedSourceClocks, proof.individualDeltas, "Every durable measured delta needs an original provider clock");
       if (!result.stop) assert.equal(proof.individualDeltas, result.events);
     }
     const reference = first.baseline.cases.find((item: any) => item.name === name);

scripts/native-ingestion-performance-gates.test.mjs

@@ -75,15 +75,17 @@ const workloads = [
   { name: "operator-stop", runs: 1, events: 512, intervalMs: 100, prelude: 0, stop: true },
 ];
 function browserReport(revision, latency) {
-  return { complete: true, qualificationOnly: false, diagnosticOnly: false, performanceAcceptanceEligible: true,
+  return { protocolVersion: "v2-compact-markers-original-clock-stream", complete: true, qualificationOnly: false, diagnosticOnly: false, performanceAcceptanceEligible: true, browserRecording: { trace: "off", video: "off", screencast: false },
     sourceRevision: revision, compiledBuildRevision: revision, serverArtifact: "compiled", serverEntrySha256: `${revision}-compiled`,
     measurementSha256: "synthetic-fixed-fixture", compiledHotPathHashes: { append: "append", port: "port", runtime: "runtime" },
     cases: workloads.map(workload => ({ ...workload, stop: Boolean(workload.stop), persistedAfterReload: true,
+      epochMs: 1000, preludeProof: workload.prelude ? { individualDeltas: workload.prelude, uniqueSourceIdentities: workload.prelude, verifiedAtMs: 999 } : undefined,
+      visible: Array(workload.stop ? 3 : workload.events).fill({}), matchedVisibleClocks: workload.stop ? 3 : workload.events,
       firstVisibleAfterEmissionMs: latency, emissionToVisible: { medianMs: latency, p95Ms: latency },
       intendedToVisible: { p95Ms: latency }, drainThroughTerminalMs: latency,
       stopThroughTerminalMs: latency, stopThroughAcknowledgementMs: latency,
       durable: Array.from({ length: workload.runs }, (_, index) => ({ token: String(index), ordered: true,
-        uniqueSourceIdentities: workload.events, individualDeltas: workload.events })),
+        uniqueSourceIdentities: workload.events, individualDeltas: workload.events, matchedSourceClocks: workload.events })),
     })) };
 }
 
@@ -119,6 +121,13 @@ test("browser CLI rejects the three-pairs-above-noise counterexample and ineligi
       ["source", report => { report.serverArtifact = "source-qualification"; }, /compiled server/],
       ["profile", report => { report.diagnosticOnly = true; }, /Diagnostic profiles/],
       ["profile-ineligible", report => { report.performanceAcceptanceEligible = false; }, /Profile-only evidence/],
+      ["missing-retained-prelude", report => { report.cases.find(row => row.prelude).preludeProof.individualDeltas--; }, /stored prelude deltas/],
+      ["legacy-protocol", report => { delete report.protocolVersion; }, /v2 clock protocol/],
+      ["unmatched-visible-clock", report => { report.cases[0].matchedVisibleClocks--; }, /original provider clock/],
+      ["unmatched-durable-clock", report => { report.cases[0].durable[0].matchedSourceClocks--; }, /original provider clock/],
+      ["browser-trace", report => { report.browserRecording.trace = "on"; }, /trace capture off/],
+      ["browser-video", report => { report.browserRecording.video = "on"; }, /video capture off/],
+      ["browser-screencast", report => { report.browserRecording.screencast = true; }, /screencast capture off/],
     ]) {
       const rejected = run([`--pairs=${await manifest(name, Array(5).fill(0.30), false, change)}`, `--noise-report=${noise}`, "--require-acceptance"]);
       assert.notEqual(rejected.status, 0); assert.match(rejected.stderr, message);

tests/e2e/fixtures/in-feed-codex.mjs

@@ -4,6 +4,7 @@
 import { randomUUID } from 'node:crypto';
 import { once } from 'node:events';
 import { readFileSync } from 'node:fs';
+import { request as httpRequest } from 'node:http';
 import { join, resolve } from 'node:path';
 import { createInterface } from 'node:readline';
 if (process.argv.includes('--version')) { console.log('codex-cli 0.115.0 (in-feed fixture)'); process.exit(0); }
@@ -14,6 +15,7 @@ const recoveryFixture = process.env.PAPERCLIP_RECOVERY_FIXTURE === "1";
 const ingestionFixture = process.argv.includes('--ingestion-stream-fixture');
 let ingestionPlan = null;
 let ingestionGeneration = 0;
+let ingestionClock = null;
 let currentObjective = "";
 let emitCeoLineage = false;
 let completionContract = { revision: "1", criterionIds: ["objective"] };
@@ -42,15 +44,56 @@ function ingestionFixturePlan(value) {
     // The real Markdown composer escapes punctuation, and the native brief
     // also carries a shortened title. Accept a complete fixture plan only.
     for (let pass = 0; pass < 4; pass++) text = text.replace(/\\([\\_=:./-])/g, '$1');
-    const match = text.match(/INGESTION_PERF nonce=([a-z0-9]+) events=(\d+) intervalMs=(\d+) prelude=(\d+) gate=(http:\/\/(?:127\.0\.0\.1|localhost):\d+\/native-ingestion-gate\/[a-f0-9]{32})/);
+    const match = text.match(/INGESTION_PERF nonce=([a-z0-9]+) events=(\d+) intervalMs=(\d+) prelude=(\d+) markerCode=([a-z0-9]{2}) gate=(http:\/\/(?:127\.0\.0\.1|localhost):\d+\/native-ingestion-gate\/[a-f0-9]{32})/);
     if (!match) continue;
-    const plan = { nonce: match[1], events: Number(match[2]), intervalMs: Number(match[3]), prelude: Number(match[4]), gate: match[5] };
+    const plan = { nonce: match[1], events: Number(match[2]), intervalMs: Number(match[3]), prelude: Number(match[4]), markerCode: match[5], gate: match[6] };
     if (plan.events < 1 || plan.events > 4096 || plan.intervalMs > 1000 || plan.prelude > 10_000) throw new Error('Ingestion fixture exceeds its bounded workload');
     return plan;
   }
   throw new Error('Missing complete INGESTION_PERF fixture plan');
 }
 
+function openIngestionClock(plan, gate) {
+  const endpoint = new URL(gate);
+  endpoint.pathname += '/clock';
+  endpoint.searchParams.set('participant', plan.nonce);
+  let closed = false, failure = null, settle;
+  const done = new Promise(resolve => { settle = resolve; });
+  const fail = (error) => {
+    if (failure) return;
+    failure = new Error(`Ingestion fixture clock ${plan.nonce}: ${error.message}`);
+    process.stderr.write(`${failure.message}\n`);
+    process.exitCode = 1;
+    settle(false);
+  };
+  const request = httpRequest(endpoint, { method: 'POST', agent: false,
+    headers: { 'Content-Type': 'application/x-ndjson' } }, response => {
+    response.resume();
+    response.on('error', fail);
+    response.on('aborted', () => fail(new Error('Response aborted')));
+    response.on('end', () => {
+      if (response.statusCode < 200 || response.statusCode >= 300) fail(new Error(`HTTP ${response.statusCode}`));
+      else settle(true);
+    });
+  });
+  request.on('error', fail);
+  request.setTimeout(300_000, () => request.destroy(new Error('Upload timed out')));
+  const write = record => {
+    if (failure) throw failure;
+    if (closed) throw new Error('Ingestion fixture clock is closed');
+    // One buffered stream, with no per-token request or clock-drain wait.
+    request.write(`${JSON.stringify(record)}\n`);
+  };
+  const close = () => {
+    if (closed) return;
+    closed = true;
+    try { request.end(); } catch (error) { fail(error); }
+  };
+  write({ type: 'hello', nonce: plan.nonce, markerCode: plan.markerCode,
+    events: plan.events, intervalMs: plan.intervalMs });
+  return { write, close, done };
+}
+
 async function streamIngestion() {
   if (!ingestionPlan) throw new Error('Missing INGESTION_PERF fixture plan');
   const plan = ingestionPlan, generation = ++ingestionGeneration;
@@ -71,32 +114,47 @@ async function streamIngestion() {
       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));
+  const clock = openIngestionClock(plan, gate);
+  ingestionClock = clock;
+  try {
+    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');
     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 } });
+    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 intendedMs = epochMs + index * plan.intervalMs;
+      const wait = intendedMs - Date.now();
+      if (wait > 0) await new Promise(resolve => setTimeout(resolve, wait));
+      if (generation !== ingestionGeneration) return;
+      const emittedMs = Date.now();
+      clock.write({ type: 'emission', nonce: plan.nonce, markerCode: plan.markerCode, index, intendedMs, emittedMs });
+      const delta = `[${plan.markerCode}${index.toString(36)}] `;
+      text += delta;
+      await sendIngestion({ method: 'item/agentMessage/delta', params: { threadId, turnId, itemId, delta } });
+    }
+    clock.close();
+    if (generation !== ingestionGeneration) return;
+    await sendIngestion({ method: 'item/completed', params: { threadId, turnId,
+      item: { id: itemId, type: 'agentMessage', phase: 'final_answer', text } } });
+    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: 'turn/completed', params: { threadId, turn: { id: turnId, status: 'completed' } } });
+    // Upload completion cannot delay output, tools, result or protocol terminal.
+    // Errors are already explicit on stderr and set a nonzero process exit;
+    // do not produce a second terminal notification after a completed turn.
+    await clock.done;
+  } finally {
+    clock.close();
+    if (ingestionClock === clock) ingestionClock = null;
   }
-  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 ?? [];
@@ -268,6 +326,7 @@ createInterface({ input: process.stdin }).on('line', (line) => {
     }), 50);
   } else if (ingestionFixture && method === 'turn/interrupt') {
     ingestionGeneration += 1;
+    ingestionClock?.close();
     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-protocol-qualification.mjs

@@ -0,0 +1,144 @@
+import assert from 'node:assert/strict';
+import { spawn, spawnSync } from 'node:child_process';
+import { randomUUID, createHash } from 'node:crypto';
+import { readFile, writeFile } from 'node:fs/promises';
+import { createServer } from 'node:http';
+import { createInterface } from 'node:readline';
+import { fileURLToPath } from 'node:url';
+
+const fixture = fileURLToPath(new URL('./in-feed-codex.mjs', import.meta.url));
+const output = process.argv[2];
+assert.ok(output, 'Pass an owned output JSON path');
+const reports = [];
+assert.equal(spawnSync(process.execPath, [fixture, '--version'], { encoding: 'utf8' }).stdout.trim(), 'codex-cli 0.115.0 (in-feed fixture)');
+
+async function run({ name, count, intervalMs = 0, stopAfter, clockStatus = 200, prelude = 0 }) {
+  const nonce = randomUUID().replaceAll('-', ''), key = randomUUID().replaceAll('-', ''), markerCode = 'q7';
+  const messages = [], emissions = [], clockRecords = [], deltas = [];
+  let clockPosts = 0, gateGets = 0, hello, pendingGate, clockResponse, clockEnded = false;
+  let terminal, interruptRequestedAt, interruptAcknowledgedAt, uploadResponseAt, completedItem;
+  let epochMs, stderr = '', failure, child, finished = false;
+  const fail = error => { failure ??= error; child?.kill('SIGTERM'); };
+  const maybeGate = () => {
+    if (!hello || !pendingGate) return;
+    epochMs = Date.now() + 25;
+    pendingGate.writeHead(200, { 'Content-Type': 'application/json' }).end(JSON.stringify({ epochMs }));
+    pendingGate = null;
+  };
+  const maybeFinishUpload = () => {
+    if (!clockEnded || !terminal || !clockResponse || (stopAfter && !interruptAcknowledgedAt)) return;
+    uploadResponseAt = Date.now();
+    clockResponse.writeHead(clockStatus, { 'Content-Type': 'application/json' }).end(JSON.stringify({ received: emissions.length }));
+    clockResponse = null;
+    child.stdin.end();
+  };
+  const server = createServer((req, res) => {
+    try {
+      const url = new URL(req.url, 'http://127.0.0.1');
+      assert.equal(url.searchParams.get('participant'), nonce);
+      if (req.method === 'POST' && url.pathname === `/native-ingestion-gate/${key}/clock`) {
+        clockPosts++;
+        clockResponse = res;
+        let buffer = '';
+        req.setEncoding('utf8');
+        req.on('data', chunk => {
+          try {
+            buffer += chunk;
+            while (buffer.includes('\n')) {
+              const cut = buffer.indexOf('\n'), record = JSON.parse(buffer.slice(0, cut));
+              buffer = buffer.slice(cut + 1); clockRecords.push(record);
+              if (record.type === 'hello') { assert.equal(hello, undefined); hello = record; maybeGate(); }
+              else { assert.equal(record.type, 'emission'); emissions.push(record); }
+            }
+          } catch (error) { fail(error); }
+        });
+        req.on('end', () => { assert.equal(buffer, ''); clockEnded = true; maybeFinishUpload(); });
+      } else {
+        assert.equal(req.method, 'GET');
+        assert.equal(url.pathname, `/native-ingestion-gate/${key}`);
+        gateGets++; pendingGate = res; maybeGate();
+      }
+    } catch (error) { res.writeHead(500).end(); fail(error); }
+  });
+  await new Promise(resolve => server.listen(0, '127.0.0.1', resolve));
+  const gate = `http://127.0.0.1:${server.address().port}/native-ingestion-gate/${key}`;
+  const send = value => child.stdin.write(JSON.stringify(value) + '\n');
+  child = spawn(process.execPath, [fixture, '--ingestion-stream-fixture'], { stdio: ['pipe', 'pipe', 'pipe'] });
+  child.stderr.on('data', value => { stderr += value; });
+  const watchdog = setTimeout(() => fail(new Error(`${name}: qualification watchdog`)), 10_000);
+  createInterface({ input: child.stdout }).on('line', line => {
+    try {
+      const message = JSON.parse(line); messages.push({ at: Date.now(), message });
+      if (message.id === 'initialize') send({ id: 'thread', method: 'thread/start', params: { model: 'fixture' } });
+      else if (message.id === 'thread') {
+        // Match the normal nested brief plus Markdown punctuation escaping.
+        const prompt = `INGESTION_PERF nonce=${nonce} events=${count} intervalMs=${intervalMs} prelude=${prelude} markerCode=${markerCode} gate=${gate}`;
+        const escaped = prompt.replaceAll('_', '\\_').replaceAll('=', '\\=');
+        send({ id: 'turn', method: 'turn/start', params: { input: [{ type: 'text', text: JSON.stringify({ message: JSON.stringify({ objective: escaped }) }) }] } });
+      } else if (message.method === 'item/agentMessage/delta' && message.params.itemId.startsWith('ingestion-')) {
+        deltas.push({ delta: message.params.delta, at: Date.now() });
+        if (stopAfter && deltas.length === stopAfter) {
+          interruptRequestedAt = Date.now();
+          send({ id: 'stop', method: 'turn/interrupt' });
+        }
+      } else if (message.method === 'item/completed' && message.params.item.id.startsWith('ingestion-')) completedItem = message.params.item;
+      else if (message.method === 'item/tool/call') {
+        assert.ok(completedItem, 'Assistant completion must precede semantic tools');
+        assert.equal(uploadResponseAt, undefined, 'Tools must run before clock upload completes');
+        send({ id: message.id, result: {} });
+      } else if (message.id === 'stop') {
+        interruptAcknowledgedAt = Date.now();
+        assert.equal(uploadResponseAt, undefined, 'Stop acknowledgement cannot wait for clock upload');
+        maybeFinishUpload();
+      } else if (message.method === 'turn/completed') {
+        terminal = message.params.turn;
+        assert.equal(uploadResponseAt, undefined, 'Protocol terminal cannot wait for clock upload');
+        maybeFinishUpload();
+      }
+    } catch (error) { fail(error); }
+  });
+  send({ id: 'initialize', method: 'initialize' });
+  const exit = await new Promise(resolve => child.once('close', (code, signal) => resolve({ code, signal })));
+  clearTimeout(watchdog); finished = true;
+  server.closeAllConnections(); await new Promise(resolve => server.close(resolve));
+  if (failure) throw failure;
+  assert.equal(exit.code, clockStatus === 200 ? 0 : 1, stderr);
+  assert.equal(clockPosts, 1); assert.equal(gateGets, 1);
+  assert.deepEqual(hello, { type: 'hello', nonce, markerCode, events: count, intervalMs });
+  assert.equal(emissions.length, deltas.length);
+  assert.ok(deltas.length > 0);
+  for (let index = 0; index < emissions.length; index++) {
+    const event = emissions[index];
+    assert.deepEqual({ ...event, emittedMs: 0 }, { type: 'emission', nonce, markerCode, index, intendedMs: epochMs + index * intervalMs, emittedMs: 0 });
+    assert.ok(event.emittedMs <= deltas[index].at);
+    assert.equal(deltas[index].delta, `[${markerCode}${index.toString(36)}] `);
+  }
+  if (stopAfter) {
+    assert.equal(terminal.status, 'interrupted'); assert.ok(interruptAcknowledgedAt);
+    assert.ok(deltas.length >= stopAfter && deltas.length < count);
+    assert.ok(uploadResponseAt >= interruptAcknowledgedAt);
+    assert.equal(completedItem, undefined);
+  } else {
+    assert.equal(terminal.status, 'completed'); assert.equal(deltas.length, count);
+    assert.equal(completedItem.text, deltas.map(item => item.delta).join(''));
+    if (count === 512) assert.equal(completedItem.text.length, 3548);
+    const tools = messages.filter(row => row.message.method === 'item/tool/call').map(row => row.message.params.tool);
+    assert.deepEqual(tools, ['report_progress', 'paperclip_finish']);
+  }
+  if (clockStatus !== 200) assert.match(stderr, /Ingestion fixture clock .*HTTP 503/);
+  reports.push({ name, exit, count, emitted: deltas.length, clockPosts, clockRecords: clockRecords.length,
+    completedTextCharacters: completedItem?.text.length, terminal: terminal.status,
+    uploadHeldUntilAfterTerminal: true, stopAcknowledgedBeforeUploadResponse: stopAfter ? true : undefined,
+    explicitClockError: clockStatus !== 200 ? stderr.trim() : undefined });
+}
+try {
+  await run({ name: 'full-512-with-prelude', count: 512, prelude: 8 });
+  await run({ name: 'interrupt-with-held-clock-response', count: 512, intervalMs: 10, stopAfter: 3 });
+  await run({ name: 'explicit-upload-error', count: 16, clockStatus: 503 });
+  const proof = { fixture, fixtureSha256: createHash('sha256').update(await readFile(fixture)).digest('hex'), reports, passed: true };
+  await writeFile(output, JSON.stringify(proof, null, 2) + '\n');
+  console.log(JSON.stringify(proof));
+} catch (error) {
+  await writeFile(output, JSON.stringify({ fixture, reports, passed: false, error: error.stack }, null, 2) + '\n');
+  throw error;
+}

tests/e2e/native-event-ingestion.journey.ts

@@ -6,7 +6,12 @@ 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 SeenMarker = { markerCode: string; index: number; visibleMs: number };
+type Emission = { nonce: string; markerCode: string; index: number; intendedMs: number; emittedMs: number; receivedAtMs: number };
+type Visible = SeenMarker & Emission;
+type ClockStream = { nonce: string; markerCode: string; opened: boolean; hello: boolean; ended: boolean; aborted: boolean;
+  partialBytes: number; records: Map<number, Emission>; error?: string };
+type ClockCase = { events: number; intervalMs: number; epochMs?: number; streams: Map<string, ClockStream>; errors: string[] };
 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 }] : [
@@ -24,32 +29,85 @@ const stats = (values: number[]) => {
 
 test("native ingestion reaches the shipped UI promptly and survives reload", async ({ page, request }, info) => {
   const pending = new Map<string, Map<string, ServerResponse>>();
+  const clocks = new Map<string, ClockCase>();
+  let markerCounter = 0;
+  const nextMarkerCode = () => {
+    expect(markerCounter).toBeLessThan(36 ** 2);
+    return (markerCounter++).toString(36).padStart(2, "0");
+  };
   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;
+    const route = url.pathname.match(/^\/native-ingestion-gate\/([a-f0-9]{32})(\/clock)?$/);
+    const key = route?.[1], participant = url.searchParams.get("participant");
+    const state = key ? clocks.get(key) : undefined;
+    const stream = participant ? state?.streams.get(participant) : undefined;
+    if (!key || !participant || !state || !stream) { res.writeHead(404).end(); return; }
+    if (!route?.[2] && req.method === "GET") {
+      if (pending.get(key)!.has(participant)) { res.writeHead(409).end(); return; }
+      pending.get(key)!.set(participant, res);
+      res.on("close", () => pending.get(key)?.delete(participant));
+      return;
     }
-    pending.get(key)!.set(participant, res);
-    res.on("close", () => pending.get(key)?.delete(participant));
+    if (!route?.[2] || req.method !== "POST" || stream.opened) { res.writeHead(409).end(); return; }
+    stream.opened = true;
+    let buffered = "";
+    req.setEncoding("utf8");
+    const fail = (message: string) => {
+      stream.error = message; state.errors.push(`${participant}: ${message}`);
+      if (!res.writableEnded) res.writeHead(400).end();
+    };
+    const line = (text: string) => {
+      const value = JSON.parse(text);
+      if (value.nonce !== participant || value.markerCode !== stream.markerCode) throw new Error("Clock identity mismatch");
+      if (!stream.hello) {
+        if (value.type !== "hello" || value.events !== state.events || value.intervalMs !== state.intervalMs) throw new Error("Clock hello mismatch");
+        stream.hello = true; return;
+      }
+      if (value.type !== "emission" || !Number.isSafeInteger(value.index) || value.index < 0 || value.index >= state.events
+        || !Number.isSafeInteger(value.intendedMs) || !Number.isSafeInteger(value.emittedMs) || value.emittedMs <= 0
+        || value.intendedMs !== state.epochMs! + value.index * state.intervalMs) throw new Error("Invalid original emission clock");
+      if (stream.records.has(value.index)) throw new Error("Duplicate or conflicting original emission clock");
+      stream.records.set(value.index, { nonce: participant, markerCode: stream.markerCode, index: value.index,
+        intendedMs: value.intendedMs, emittedMs: value.emittedMs, receivedAtMs: Date.now() });
+    };
+    req.on("data", (chunk: string) => {
+      if (stream.error) return;
+      try {
+        buffered += chunk;
+        for (;;) {
+          const end = buffered.indexOf("\n"); if (end < 0) break;
+          const text = buffered.slice(0, end); buffered = buffered.slice(end + 1);
+          if (text.length > 1024) throw new Error("Clock record exceeds bounded metadata size");
+          if (text) line(text);
+        }
+        if (buffered.length > 1024) throw new Error("Clock record exceeds bounded metadata size");
+      } catch (error) { fail(String(error)); }
+    });
+    req.on("end", () => {
+      stream.ended = true; stream.partialBytes = Buffer.byteLength(buffered);
+      if (buffered.length) fail("Clock stream ended with a partial record");
+      if (!stream.error && !res.writableEnded) res.writeHead(200).end();
+    });
+    req.on("aborted", () => { stream.aborted = true; stream.partialBytes = Buffer.byteLength(buffered); });
+    req.on("error", error => { if (!stream.aborted) fail(String(error)); });
   });
   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[] = [];
+    const seen = new Set<string>(), visible: SeenMarker[] = [];
     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">[] = [];
+      const pattern = /\[([a-z0-9]{2})([a-z0-9]{1,3})\]/g;
+      const entries: Omit<SeenMarker, "visibleMs">[] = [];
       for (const match of text.matchAll(pattern)) {
-        const key = `${match[1]}:${match[2]}`;
+        const index = Number.parseInt(match[2], 36);
+        const key = `${match[1]}:${index}`;
         if (seen.has(key)) continue;
-        entries.push({ nonce: match[1], index: Number(match[2]), intendedMs: Number(match[3]), emittedMs: Number(match[4]) });
+        entries.push({ markerCode: match[1], index });
       }
       return entries;
     };
@@ -65,7 +123,7 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy
           // 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}`;
+            const key = `${entry.markerCode}:${entry.index}`;
             if (seen.has(key)) continue;
             seen.add(key); visible.push({ ...entry, visibleMs });
           }
@@ -82,13 +140,19 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy
   });
   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 captureMode = (value: unknown) => typeof value === "string" ? value
+    : value && typeof value === "object" && "mode" in value ? String(value.mode) : "off";
+  const browserRecording = { trace: captureMode(info.project.use.trace), video: captureMode(info.project.use.video),
+    screencast: captureMode(info.project.use.trace) !== "off" || captureMode(info.project.use.video) !== "off" };
   const report: Record<string, unknown> = {
+    protocolVersion: "v2-compact-markers-original-clock-stream",
     dependency: "Deterministic Codex app-server fixture; real application/runtime/port/PostgreSQL/live WebSocket/shipped UI",
     variant: process.env.PAPERCLIP_INGESTION_VARIANT ?? "unlabelled", qualificationOnly: smoke,
     diagnosticOnly: Boolean(process.env.PAPERCLIP_NATIVE_INGESTION_DIAGNOSTIC_DIR),
-    performanceAcceptanceEligible: !process.env.PAPERCLIP_NATIVE_INGESTION_DIAGNOSTIC_DIR,
+    performanceAcceptanceEligible: !process.env.PAPERCLIP_NATIVE_INGESTION_DIAGNOSTIC_DIR && !browserRecording.screencast,
+    browserRecording,
     startedAtMs: Date.now(),
-    emittedClock: "Provider process epoch; original schedule retained even if stdout backpressures",
+    emittedClock: "Original provider-process clocks on one pre-opened streamed HTTP channel per provider; no per-token request or flush wait",
     observedClock: "Browser epoch after a DOM update and one paint opportunity",
     cases: [], companyId: company.id,
     health: await json(await request.get("/api/health")),
@@ -110,7 +174,17 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy
   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", "fixtures/native-ingestion-bootstrap.mjs"];
-  report.measurementSha256 = createHash("sha256").update(Buffer.concat(await Promise.all(protocolFiles.map(file => readFile(path.resolve(import.meta.dirname, file)))))).digest("hex");
+  const protocolSources = await Promise.all(protocolFiles.map(async file => ({ file, bytes: await readFile(path.resolve(import.meta.dirname, file)) })));
+  report.measurementSha256 = createHash("sha256").update(Buffer.concat(protocolSources.map(source => source.bytes))).digest("hex");
+  const harnessRoot = path.resolve(import.meta.dirname, "../..");
+  report.harnessProvenance = {
+    sourceRevision: execFileSync("git", ["rev-parse", "HEAD"], { cwd: harnessRoot, encoding: "utf8" }).trim(),
+    sourceStatus: execFileSync("git", ["status", "--porcelain", "--", "tests/e2e"], { cwd: harnessRoot, encoding: "utf8" }).trim(),
+    sourceHashes: Object.fromEntries(protocolSources.map(source => [source.file, createHash("sha256").update(source.bytes).digest("hex")])),
+    driverDependencySource: process.env.PAPERCLIP_INGESTION_DRIVER_DEPENDENCIES ?? "Checkout tooling dependencies",
+    driverLockSha256: createHash("sha256").update(await readFile(path.join(harnessRoot, "pnpm-lock.yaml"))).digest("hex"),
+    scope: "Separate browser harness and driver-only tooling; every measured app uses its own frozen clean source and matching compiled stamp",
+  };
   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" } }));
@@ -131,14 +205,26 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy
 
     async function perform(workload: Case, stop = false) {
       const key = randomUUID().replaceAll("-", ""), nonce = randomUUID().replaceAll("-", "");
+      const markerCode = nextMarkerCode();
       pending.set(key, new Map());
-      const prompt = (token: string, prelude: number) => `Measure native streaming for ${token.slice(0, 8)}.\n\nINGESTION_PERF nonce=${token} events=${workload.events} intervalMs=${workload.intervalMs} prelude=${prelude} gate=${base}${key}`;
+      const state: ClockCase = { events: workload.events, intervalMs: workload.intervalMs, streams: new Map(), errors: [] };
+      clocks.set(key, state);
+      const register = (token: string, code: string) => state.streams.set(token, { nonce: token, markerCode: code,
+        opened: false, hello: false, ended: false, aborted: false, partialBytes: 0, records: new Map() });
+      register(nonce, markerCode);
+      const compactOutput = Array.from({ length: workload.events }, (_, index) => `[${markerCode}${index.toString(36)}] `).join("");
+      const preludeOutput = workload.prelude ? `${".".repeat(Math.max(0, workload.prelude - 1))} PRELUDE_${nonce}_READY ` : "";
+      expect([...compactOutput].length, "Full measured completion must fit the unchanged Rust text bound").toBeLessThan(4000);
+      expect([...preludeOutput + compactOutput].length, "Prelude and measured text must remain below that bound").toBeLessThan(4000);
+      const active: Record<string, unknown> = { ...workload, stop, nonce, markerCode, phase: "setup", key };
+      report.activeCase = active;
+      const prompt = (token: string, code: string, prelude: number) => `Measure native streaming for ${token.slice(0, 8)}.\n\nINGESTION_PERF nonce=${token} events=${workload.events} intervalMs=${workload.intervalMs} prelude=${prelude} markerCode=${code} 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 dialog.getByRole("textbox", { name: "editable markdown" }).fill(prompt(nonce, markerCode, 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;
@@ -147,11 +233,13 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy
         return Boolean(issue);
       }).toBe(true);
       const issueIds = [issue.id];
+      active.issueIds = issueIds;
       const issuedTokens = new Map<string, string>([[issue.id, nonce]]);
       for (let index = 1; index < workload.runs; index++) {
-        const backgroundToken = randomUUID().replaceAll("-", "");
+        const backgroundToken = randomUUID().replaceAll("-", ""), backgroundCode = nextMarkerCode();
+        register(backgroundToken, backgroundCode);
         const background = await json(await request.post(`/api/companies/${company.id}/issues`, { data: {
-          title: `Ingestion background ${key} ${index}`, description: prompt(backgroundToken, 0),
+          title: `Ingestion background ${key} ${index}`, description: prompt(backgroundToken, backgroundCode, 0),
           status: "backlog", assigneeAgentId: agents[Math.floor(index / 50)].id,
         } }));
         issueIds.push(background.id);
@@ -162,12 +250,41 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy
       }
       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 });
+      active.phase = "provider-and-clock-readiness";
+      await expect.poll(() => ({ providers: pending.get(key)!.size,
+        clocks: [...state.streams.values()].filter(stream => stream.hello).length }),
+      { timeout: 180_000 }).toEqual({ providers: workload.runs, clocks: workload.runs });
+      let preludeProof: Record<string, unknown> | undefined;
+      if (workload.prelude) {
+        await expect(page.getByText(`PRELUDE_${nonce}_READY`, { exact: false }).first()).toBeVisible({ timeout: 120_000 });
+        active.phase = "independent-prelude-verification";
+        const listed = await json(await request.get(`/api/companies/${company.id}/heartbeat-runs?limit=1000`));
+        const foreground = listed.find((run: any) => (run.issueId ?? run.nativeIssueId ?? run.contextSnapshot?.issueId ?? run.contextSnapshot?.taskId) === issue.id);
+        expect(foreground).toBeTruthy();
+        const identities = new Set<string>(); let count = 0, afterSeq = 0;
+        while (true) {
+          const rows = await json(await request.get(`/api/heartbeat-runs/${foreground.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" || !String(event.itemId ?? "").startsWith("prelude-")) continue;
+            expect(event.payload?.text).toBe(count === workload.prelude - 1 ? ` PRELUDE_${nonce}_READY ` : ".");
+            count++; identities.add(`${event.sourceInstanceId}:${event.sourceEventId}:${event.sourceSeq}`);
+          }
+          const next = Math.max(...rows.map((row: { seq: number }) => row.seq));
+          expect(next).toBeGreaterThan(afterSeq); afterSeq = next;
+        }
+        preludeProof = { runId: foreground.id, individualDeltas: count, uniqueSourceIdentities: identities.size,
+          retainedGlobalHead: afterSeq, verifiedAtMs: Date.now(), method: "Independent normal event API read before common emission gate release" };
+        active.preludeProof = preludeProof;
+        expect(count).toBe(workload.prelude); expect(identities.size).toBe(count);
+      }
       const epochMs = Date.now() + 1000;
+      state.epochMs = epochMs; active.epochMs = epochMs; active.phase = "streaming";
       const participants = [...pending.get(key)!.keys()];
+      active.participants = participants;
       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);
+      const observed = () => page.evaluate(code => (window as unknown as { nativeIngestionVisible: SeenMarker[] }).nativeIngestionVisible.filter(item => item.markerCode === code), markerCode);
       await expect.poll(async () => (await observed()).length, { timeout: 120_000 }).toBeGreaterThanOrEqual(stop ? 3 : workload.events);
       let stopRequestedAt: number | undefined;
       if (stop) {
@@ -176,12 +293,14 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy
         stopRequestedAt = Date.now();
         await button.click();
       }
+      active.phase = "terminal-drainage";
       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();
+      const settledAt = Date.now(), seenMarkers = await observed();
+      active.settledAt = settledAt; active.seenMarkers = seenMarkers;
       const details = [];
       for (const run of runs) details.push(await json(await request.get(`/api/heartbeat-runs/${run.id}`)));
       runs = details;
@@ -191,8 +310,19 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy
         (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;
+      active.phase = "clock-reconciliation";
+      await expect.poll(() => [...state.streams.values()].every(stream => !stream.error && (stop
+        ? seenMarkers.every(marker => state.streams.get(nonce)!.records.has(marker.index))
+        : stream.records.size === workload.events && stream.ended)), { timeout: 30_000 }).toBe(true);
+      expect(state.errors).toEqual([]);
+      const visible: Visible[] = seenMarkers.map(marker => {
+        const original = state.streams.get(nonce)!.records.get(marker.index);
+        expect(original, `Missing original provider clock for visible ${marker.markerCode}:${marker.index}`).toBeTruthy();
+        return { ...original!, ...marker }; // Keep the original paint timestamp; never restamp after clock arrival.
+      });
       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,
+      const clockEvidence: Array<Record<string, unknown>> = [];
+      const result = { ...workload, stop, nonce, markerCode, epochMs, preludeProof, clockEvidence, runIds: runs.map(run => run.id), participants: participants.length, visible, matchedVisibleClocks: visible.length,
         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)),
@@ -202,18 +332,20 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy
           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();
+      if (!stop) await expect(page.getByText(`[${markerCode}${(workload.events - 1).toString(36)}]`, { 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}_`);
+      if (!stop) expect(JSON.stringify(comments)).toContain(`[${markerCode}${(workload.events - 1).toString(36)}]`);
       else {
         const lastVisible = visible.at(-1)!;
-        await expect(page.getByText(`INGEST_${nonce}_${lastVisible.index}_`, { exact: false }).first()).toBeVisible();
+        await expect(page.getByText(`[${markerCode}${lastVisible.index.toString(36)}]`, { exact: false }).first()).toBeVisible();
       }
+      active.phase = "independent-durable-verification";
       const durable = [];
       for (const run of runs) {
         const taskId = run.issueId ?? run.nativeIssueId ?? run.contextSnapshot?.issueId ?? run.contextSnapshot?.taskId;
         const token = issuedTokens.get(taskId)!;
+        const code = state.streams.get(token)!.markerCode;
         expect(token).toBeTruthy();
         const deltas: Array<{ index: number; identity: string }> = [];
         let afterSeq = 0;
@@ -223,26 +355,58 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy
           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}` });
+            for (const match of String(event.payload?.text ?? "").matchAll(new RegExp(`\\[${code}([a-z0-9]{1,3})\\]`, "g"))) {
+              const index = Number.parseInt(match[1], 36);
+              expect(event.payload.text, "Every durable normalized delta retains its complete compact payload").toBe(`[${code}${index.toString(36)}] `);
+              deltas.push({ index, identity: `${event.sourceInstanceId}:${event.sourceEventId}:${event.sourceSeq}` });
             }
           }
           const next = Math.max(...rows.map((row: { seq: number }) => row.seq));
           expect(next).toBeGreaterThan(afterSeq); afterSeq = next;
         }
+        active.latestDurableRead = { runId: run.id, nonce: token, markerCode: code, deltas };
+        await expect.poll(() => deltas.every(delta => state.streams.get(token)!.records.has(delta.index)), { timeout: 30_000 }).toBe(true);
+        for (const delta of deltas) {
+          const original = state.streams.get(token)!.records.get(delta.index)!;
+          expect({ nonce: original.nonce, markerCode: original.markerCode, index: original.index })
+            .toEqual({ nonce: token, markerCode: code, index: delta.index });
+        }
         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 });
+        durable.push({ runId: run.id, token, markerCode: code, individualDeltas: deltas.length, uniqueSourceIdentities: deltas.length, matchedSourceClocks: deltas.length, ordered: true });
       }
+      await expect.poll(() => [...state.streams.values()].every(stream => stream.ended || stream.aborted), { timeout: 30_000 }).toBe(true);
+      expect(state.errors).toEqual([]);
+      expect([...state.streams.values()].every(stream => !stream.error)).toBe(true);
+      clockEvidence.push(...[...state.streams.values()].map(stream => ({ nonce: stream.nonce, markerCode: stream.markerCode,
+        hello: stream.hello, ended: stream.ended, aborted: stream.aborted, error: stream.error, partialBytes: stream.partialBytes,
+        emissions: [...stream.records.values()].sort((a, b) => a.index - b.index) })));
       (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);
+      pending.delete(key); clocks.delete(key); delete report.activeCase;
     }
     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 {
+    if (report.activeCase) {
+      const active = report.activeCase as Record<string, unknown>;
+      try {
+        active.failureBrowser = await page.evaluate(code => ({ capturedAtMs: Date.now(),
+          observed: (window as unknown as { nativeIngestionVisible?: SeenMarker[] }).nativeIngestionVisible?.filter(item => item.markerCode === code),
+          domTextLength: (document.body?.innerText ?? "").length,
+          renderedTexts: [...document.querySelectorAll("[data-conversation-text]")].map(node => ({ kind: node.getAttribute("data-conversation-text"), text: node.textContent })),
+          domMarkers: [...(document.body?.innerText ?? "").matchAll(/\[([a-z0-9]{2})([a-z0-9]{1,3})\]/g)]
+            .filter(match => match[1] === code).map(match => ({ markerCode: match[1], index: Number.parseInt(match[2], 36) })),
+        }), active.markerCode as string);
+      } catch (error) { active.captureError = String(error); }
+      active.failurePendingProviders = [...(pending.get(active.key as string)?.keys() ?? [])];
+      const state = clocks.get(active.key as string);
+      active.failureClockEvidence = state && { errors: state.errors, streams: [...state.streams.values()].map(stream => ({
+        ...stream, records: [...stream.records.values()],
+      })) };
+    }
     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();

tests/e2e/playwright-native-ingestion.config.ts

@@ -6,6 +6,8 @@ import base from "./playwright.config";
 
 const server = base.webServer as Exclude<typeof base.webServer, unknown[]>;
 const evidence = process.env.PAPERCLIP_INGESTION_EVIDENCE_DIR;
+const browserRecording = Boolean(process.env.PAPERCLIP_NATIVE_INGESTION_DIAGNOSTIC_DIR) || process.env.PAPERCLIP_INGESTION_BROWSER_CAPTURE === "1";
+process.env.PAPERCLIP_INGESTION_BROWSER_RECORDING = browserRecording ? "on" : "off";
 const appRoot = path.resolve(process.env.PAPERCLIP_INGESTION_APP_ROOT ?? path.resolve(import.meta.dirname, "../.."));
 const quote = (value: string) => `'${value.replaceAll("'", "'\\''")}'`;
 const bootstrap = path.resolve(import.meta.dirname, "fixtures/native-ingestion-bootstrap.mjs");
@@ -27,7 +29,7 @@ export default defineConfig({
   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 },
+  use: { ...base.use, trace: browserRecording ? "on" : "off", video: "off", actionTimeout: 20_000 },
   webServer: {
     ...server,
     cwd: appRoot,