diff --git a/doc/plans/2026-11-03-native-event-ingestion.md b/doc/plans/2026-11-03-native-event-ingestion.md index 34727ef872d..7e5961cb2ea 100644 --- a/doc/plans/2026-11-03-native-event-ingestion.md +++ b/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. diff --git a/scripts/compare-native-ingestion-browser.ts b/scripts/compare-native-ingestion-browser.ts index 2512188ee2a..1c1f11fa786 100644 --- a/scripts/compare-native-ingestion-browser.ts +++ b/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); diff --git a/scripts/native-ingestion-performance-gates.test.mjs b/scripts/native-ingestion-performance-gates.test.mjs index 134b649d006..2d88fa4f483 100644 --- a/scripts/native-ingestion-performance-gates.test.mjs +++ b/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); diff --git a/tests/e2e/fixtures/in-feed-codex.mjs b/tests/e2e/fixtures/in-feed-codex.mjs index ced943c1e35..07806f40184 100644 --- a/tests/e2e/fixtures/in-feed-codex.mjs +++ b/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: {} }); diff --git a/tests/e2e/fixtures/native-ingestion-protocol-qualification.mjs b/tests/e2e/fixtures/native-ingestion-protocol-qualification.mjs new file mode 100644 index 00000000000..1622239d877 --- /dev/null +++ b/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; +} diff --git a/tests/e2e/native-event-ingestion.journey.ts b/tests/e2e/native-event-ingestion.journey.ts index f7f88991b0c..44ad16495a0 100644 --- a/tests/e2e/native-event-ingestion.journey.ts +++ b/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; error?: string }; +type ClockCase = { events: number; intervalMs: number; epochMs?: number; streams: Map; 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>(); + const clocks = new Map(); + 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(resolve => gate.listen(0, "127.0.0.1", resolve)); const address = gate.address(); if (!address || typeof address === "string") throw new Error("Local ingestion gate failed"); const base = `http://127.0.0.1:${address.port}/native-ingestion-gate/`; await page.addInitScript(() => { - const seen = new Set(), visible: Visible[] = []; + const seen = new Set(), 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[] = []; + const pattern = /\[([a-z0-9]{2})([a-z0-9]{1,3})\]/g; + const entries: Omit[] = []; 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 = { + 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 = { ...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([[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 | 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(); 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> = []; + 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; + 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(); diff --git a/tests/e2e/playwright-native-ingestion.config.ts b/tests/e2e/playwright-native-ingestion.config.ts index c76b3fdcb8d..84ecdf4bff9 100644 --- a/tests/e2e/playwright-native-ingestion.config.ts +++ b/tests/e2e/playwright-native-ingestion.config.ts @@ -6,6 +6,8 @@ import base from "./playwright.config"; const server = base.webServer as Exclude; 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,