diff --git a/scripts/native-ingestion/README.md b/scripts/native-ingestion/README.md new file mode 100644 index 00000000000..447028dc2e3 --- /dev/null +++ b/scripts/native-ingestion/README.md @@ -0,0 +1,170 @@ +# Native ingestion experiment helpers + +These helpers reproduce the experiment registered on November 3, 2026. They +retain its immutable application revisions, harness revision, workload hashes, +acceptance checks and failure behavior. They are not a benchmark against an +arbitrary current checkout. Measurements and their qualification status belong +in the dated plan and evidence report; copying these helpers proves no gain. + +## Provenance + +The source helpers were snapshotted from the following task-owned files. The +hashes describe the originals before the small packaging changes below. + +| Committed file | Original in `/private/tmp/` | Original SHA256 | +| --- | --- | --- | +| `run-frozen.mjs` | `native-ingestion-run-frozen.mjs` | `09879a9d6a4554b98bd0435cdc222915b0613cb9dfb22b36615ede8794f70435` | +| `diagnostic-run.mjs` | `native-ingestion-diagnostic-run.mjs` | `06cc3de3b67dcaa97dfa90f57562fd72b4fdf39b4f9e33c2cac543d569087681` | +| `diagnostic-preload.mjs` | `native-ingestion-diagnostic-preload.mjs` | `2afb5e0b9fff878cb7abc7e8a4e236e1620ac35eede9ca06accfec845756d10a` | +| `verify-delivery-runtime.py` | `native-ingestion-verify-delivery-runtime.py` | `a2d9d2fe4ed2e2b0bff61b9d264f1678b84266d0407d43e19cf9f4063819a7d4` | + +Packaging changes: the diagnostic runner imports its renamed adjacent +`diagnostic-preload.mjs`; its dry-run message uses the committed path; the +acceptance runner rejects both the original and packaged preload filenames in +`NODE_OPTIONS`. Application, protocol, compiled-anchor and gate hashes are +unchanged. The runtime comparison helper and diagnostic preload are exact copies. + +The frozen experiment revisions are: + +| Role | Revision | +| --- | --- | +| B0 baseline | `83b5f7d486e70e484acd9a41bc957054f4e10175` | +| B1 cursor | `7922802f069ab3b39e9f79368c46615ba2201ba9` | +| B2 cursor and batching | `6afeba4edafd47120e8117f736641d5424b27d4a` | +| H browser driver | `f5c560d80d9d4ee089297d3b3f300c052db17c9e` | + +H is a separate test-driver checkout. Folding its test changes into a later +delivery commit does not change the recorded H revision. Check out these exact +revisions to repeat this experiment; do not update the constants to pass a guard. + +## Preparation and quiet windows + +Use Node 26.4.0 and pnpm 9.15.4, matching the captured experiment environment. +Install each measured application independently with `pnpm install +--frozen-lockfile` and run its normal `pnpm build`. Every application owns its +workspace dependencies, TypeScript output and UI output. The build stamp must +equal that checkout's exact revision. The original experiment shared only a +verified identical Rust target cache; cache sharing is optional. + +The historical H setup linked only its root and server `node_modules` to B1's +locked driver tools. H did not supply application source or compiled output to +any measured application. The runner records this setup in the evidence's +`PAPERCLIP_INGESTION_DRIVER_DEPENDENCIES` description. Preserve it when reproducing +the historical runner. Install the matching Playwright Chromium build first. + +Select a disposable quiet host with PostgreSQL and Chromium available. The +service harness starts a real embedded PostgreSQL database; browser cases start +the compiled application and shipped UI against a disposable database. No cloud +account, production data or provider credential is required. Run one stage at a +time, with no competing build, test, profiler or benchmark. Setup and warmup are +excluded from timed append samples. The production-built browser experiment +includes offered-load queueing and terminal backlog drainage. + +Set the following shell variables to your own absolute checkout and output +paths. `TOOLS_CHECKOUT` contains this committed helper directory. `B1_CHECKOUT`, +`B0_CHECKOUT`, `B2_CHECKOUT` and `H_CHECKOUT` contain the exact revisions above. +Choose a new output directory for every independent experiment; existing evidence +is never overwritten. Always supply the roots explicitly because the fallback +paths preserve the historical local setup. + +```sh +node "$TOOLS_CHECKOUT/scripts/native-ingestion/run-frozen.mjs" \ + --root="$B1_CHECKOUT" --b0-root="$B0_CHECKOUT" --b2-root="$B2_CHECKOUT" \ + --harness-root="$H_CHECKOUT" --output-root="$EVIDENCE_DIR" +``` + +Without `--execute`, this prints the registered stages and launches no work. +After qualification and release of a quiet window, add `--execute` and one +`--stage=` from the following sequence: + +1. `service-aa-b0`, `service-b0-b1`, `service-aa-b1`, `service-b1-b2`, + `service-b0-b2`. +2. `browser-smoke-b0`, `browser-smoke-b1`, `browser-smoke-b2`, then + `browser-full-qualification-b0`. +3. `browser-aa-b0`, `browser-b0-b1`, `browser-aa-b1`, `browser-b1-b2`, + `browser-b0-b2`. + +Stop on failure, retain its evidence, and diagnose before continuing. A timeout +or cleanup error is incomplete evidence. The process watchdogs bound resources; +they do not relax performance gates. Readiness must admit all 100 participants +before the common emission epoch in the full browser workload. + +## Acceptance and interpretation + +Service A/A requires five alternating pairs for all nine latency cases: +ordered, replay and warm-suffix at 1k, 10k and 100k retained histories. Every case +must have at most 20% maximum paired relative difference and 10% median arm +drift. Each timed sample contains 16 groups of 32 events. Late-gap closure has a +separate structural/correctness check; its one-shot timing is not a latency gate. + +Candidate improvement must exceed the observed per-case A/A noise in at least +four of five individual pairs and in the median. Cursor comparisons require +10k/100k improvement and 1k nonregression. Browser comparisons preserve sparse +streaming, inter-chunk, first-render and Stop nonregression gates. The batching +and combined comparisons additionally require improvement for loaded `burst512` +p95 emission-to-render, p95 intended-to-render and terminal drainage. Every +background run requires independent durable delta proof. Browser A/A suitability +still requires review of its actual noise; successful execution alone is not a +repeatability conclusion. + +These are local service/application measurements. They establish neither +deployed production improvement nor qualification of a 10,000-agent fleet. The +live-provider integration journey is separate. Do not interpret sparse raw +transport output as universally durable merely because its wire ACK was sent. + +## Separate diagnostics + +```sh +node "$TOOLS_CHECKOUT/scripts/native-ingestion/diagnostic-run.mjs" \ + --root="$B1_CHECKOUT" --b0-root="$B0_CHECKOUT" --b2-root="$B2_CHECKOUT" \ + --harness-root="$H_CHECKOUT" --tree=B2 --output=/tmp/native-ingestion-profile-new +``` + +This is also a dry run until `--execute` is supplied. Start with `--smoke` to +qualify the assembled diagnostic observer; it has no CPU profile and cannot +establish acceptance. Full diagnostics require successful marker coverage and +all compiled instrumentation anchors to be exercised. Their qualification must +be recorded separately from their presence in this directory. + +The observer records actual normalized queue depth, postcommit source markers, +SQL/transaction attempts without SQL text or parameters, server Node CPU/memory, +event-loop delay, and sampled PostgreSQL process CPU/RSS. Full runs also capture +an app-only Node CPU profile. The outputs are explicitly `diagnosticOnly` and +ineligible for unprofiled acceptance. Queue observation excludes collector, +iterator, in-flight write and raw transport buffers. CPU/SQL window brackets +include disclosed sampling overshoot and may include verification reads. +PostgreSQL RSS sums double-count shared pages. Server Node CPU excludes Rust, +providers, Chromium and the external driver. Preserve these limits with any chart +or table derived from the diagnostic outputs. + +The diagnostic output must be a new owned directory under `/tmp/` or +`/private/tmp/`. The test-only provider wrapper leaves production environment +sanitization unchanged. Unknown compiled-module SHA values fail immediately; +do not add an unreviewed hash to make a different build run. + +## Delivery-source comparison + +```sh +python3 "$TOOLS_CHECKOUT/scripts/native-ingestion/verify-delivery-runtime.py" \ + --repo="$TOOLS_CHECKOUT" \ + --frozen=7922802f069ab3b39e9f79368c46615ba2201ba9 \ + --delivery=3b77c1ca29467fbd742e9b6df4885bccef0acf08 +``` + +For B2, supply its frozen and delivery commits instead. This is a read-only Git +source/input comparison with the two documented test-typecheck script exceptions +and genuine CI helpers pinned to `8ab89f41e499032e3e55d261f457a3463d754a1d`. +It does not compare installed dependencies or compiled artifacts; those require +their separate build/equivalence receipts. Later delivery revisions must remain +explicit full commit IDs. + +## Evidence retention + +Keep paired samples, exact command manifests, source/protocol hashes, environment, +correctness and cleanup results together. Keep diagnostic evidence separate. +Retain failed qualifications and their primary errors, record retry linkage and +confidence in the diagnosis, and never promote a failed report by rewriting its +status. Publish compact JSON and interpretation with the owning PR; link raw +artifacts by a durable location and SHA256. Temporary local paths alone are not a +reviewable final evidence location. Scan live-provider artifacts for credentials +before publishing; these deterministic helpers require none. diff --git a/scripts/native-ingestion/diagnostic-preload.mjs b/scripts/native-ingestion/diagnostic-preload.mjs new file mode 100644 index 00000000000..2ebccbe43b8 --- /dev/null +++ b/scripts/native-ingestion/diagnostic-preload.mjs @@ -0,0 +1,176 @@ +import assert from 'node:assert/strict'; +import { createHash } from 'node:crypto'; +import { appendFileSync, mkdirSync, readFileSync, writeFileSync } from 'node:fs'; +import { registerHooks } from 'node:module'; +import path from 'node:path'; +import { fileURLToPath } from 'node:url'; +import { monitorEventLoopDelay, performance } from 'node:perf_hooks'; + +const directory = process.env.PAPERCLIP_NATIVE_INGESTION_DIAGNOSTIC_DIR; +assert.equal(process.env.NODE_ENV, 'test', 'Diagnostic instrumentation is test-only'); +assert.ok(directory?.startsWith('/private/tmp/') || directory?.startsWith('/tmp/'), 'Use an owned private temporary diagnostic directory'); +mkdirSync(directory, { recursive: true }); +const role = process.argv.includes('--ingestion-stream-fixture') ? 'provider' + : process.argv.some(value => /\/server\/dist\/index\.js$/.test(value)) ? 'app' : 'other'; +const file = path.join(directory, `${role}-${process.pid}.ndjson`); +const metadata = { diagnosticOnly: true, role, pid: process.pid, parentPid: process.ppid, startedAtMs: Date.now(), + cwd: process.cwd(), configPath: role === 'app' ? process.env.PAPERCLIP_CONFIG : undefined, + disclosure: 'Profile-only instrumentation; never use these timings for unprofiled acceptance' }; +writeFileSync(path.join(directory, `${role}-${process.pid}.metadata.json`), JSON.stringify(metadata, null, 2)); +const queue = []; +const record = value => queue.push({ atMs: Date.now(), ...value }); +function flush() { if (queue.length) appendFileSync(file, queue.splice(0).map(value => JSON.stringify(value)).join('\n') + '\n'); } +const markerPattern = /INGEST_([a-z0-9]+)_(\d+)_(\d+)_(\d+)/g; +const markers = text => { + if (typeof text !== 'string') return []; + const compact = [...text.matchAll(/\[([a-z0-9]{2})([a-z0-9]{1,3})\]/g)].map(match => ({ markerCode: match[1], index: Number.parseInt(match[2], 36) })); + if (compact.length) return compact; + return [...text.matchAll(markerPattern)].map(match => ({ nonce: match[1], index: Number(match[2]), intendedMs: Number(match[3]), emittedMs: Number(match[4]) })); +}; +process.once('exit', flush); +setInterval(flush, 250).unref(); + +if (role === 'provider') { + for (const signal of ['SIGINT', 'SIGTERM']) { + const finish = () => { + flush(); + process.removeListener(signal, finish); + if (process.listenerCount(signal) === 0) process.kill(process.pid, signal); + }; + process.once(signal, finish); + } + const original = process.stdout.write; + let pending = ''; + process.stdout.write = function (chunk, ...args) { + const result = original.call(this, chunk, ...args); + pending += Buffer.isBuffer(chunk) ? chunk.toString('utf8') : String(chunk); + for (;;) { + const end = pending.indexOf('\n'); if (end < 0) break; + const line = pending.slice(0, end); pending = pending.slice(end + 1); + let message; try { message = JSON.parse(line); } catch { continue; } + if (message.method === 'item/agentMessage/delta') { + for (const marker of markers(message.params?.delta)) record({ kind: 'provider-emission', ...marker }); + } else if (message.method === 'turn/completed') { + record({ kind: 'provider-terminal' }); + flush(); // Provider teardown can precede the periodic diagnostic flush. + } + } + return result; + }; +} else if (role === 'app') { + // The production provider environment sanitizer intentionally strips loader + // options. A separate test-only command wrapper explicitly starts the + // diagnostic observer; production environment policy stays untouched. + const providerCommand = process.env.PAPERCLIP_NATIVE_INGESTION_DIAGNOSTIC_CODEX_COMMAND; + assert.ok(providerCommand?.startsWith(directory + '/'), 'An owned profile-only provider wrapper is required'); + const key = Symbol.for('paperclip.native-ingestion.diagnostic'); + const queues = new Set(); + const committed = new Set(); + const fatalErrors = []; + const counters = { normalizedEvents: 0, committedEvents: 0, duplicateReceipts: 0, committedMarkers: 0, + sqlCalls: 0, begins: 0, commits: 0, rollbacks: 0, savepoints: 0, releases: 0, queueHighWater: 0 }; + const sqlVerbs = {}; + let queuePeakSinceSample = 0; + function queueDepth() { + let depth = 0, active = 0; + for (const reference of queues) { + const value = reference.deref(); + if (!value) { queues.delete(reference); continue; } + depth += value.__nativeIngestionDiagnosticDepth(); active++; + } + return { depth, active }; + } + globalThis[key] = { + registerQueue(value) { queues.add(new WeakRef(value)); }, + observeQueue() { + const { depth } = queueDepth(); + queuePeakSinceSample = Math.max(queuePeakSinceSample, depth); + counters.queueHighWater = Math.max(counters.queueHighWater, depth); + }, + normalized(event) { counters.normalizedEvents++; }, + postCommit(event, disposition) { try { + if (disposition === 'duplicate') { counters.duplicateReceipts++; return; } + assert.equal(disposition, 'committed'); + const identity = `${event.runId}:${event.sourceInstanceId}:${event.sourceEventId}`; + assert.equal(committed.has(identity), false, 'A committed source identity was observed twice'); + committed.add(identity); counters.committedEvents++; + if (event.eventType !== 'item.delta') return; + for (const marker of markers(event.payload?.text)) { + counters.committedMarkers++; record({ kind: 'postcommit-marker', runId: event.runId, sourceSeq: event.sourceSeq, + identitySha256: createHash('sha256').update(identity).digest('hex'), ...marker }); + } + } catch (error) { + // The existing persistence observer has a deliberately non-consequential + // catch. Persist an explicit fatal flag instead of trusting propagation. + const message = String(error.message ?? error); + fatalErrors.push(message); record({ kind: 'diagnostic-fatal', message }); + } + }, + sql(statement) { + counters.sqlCalls++; + const verb = statement.trim().match(/^[a-z]+/i)?.[0].toLowerCase() ?? 'other'; + sqlVerbs[verb] = (sqlVerbs[verb] ?? 0) + 1; + if (verb === 'begin') counters.begins++; + else if (verb === 'commit') counters.commits++; + else if (verb === 'rollback') counters.rollbacks++; + else if (verb === 'savepoint') counters.savepoints++; + else if (verb === 'release') counters.releases++; + }, + }; + const call = `globalThis[Symbol.for('paperclip.native-ingestion.diagnostic')]`; + const allowed = { + queue: new Set(['f8cb9fd06a567831cff886251c79eb73810a83d1384fb2b61851587a30832869']), + port: new Set(['49a971a961d831dc4dbea39c715abb371609464797bbfcc58ac911792d8b945c', 'd3cf1203feb7f0b74c6799a289799990b932a837475e857732c7340876fd4f66']), + postgres: new Set(['b57261592f135d127b8e7267c78981eb2c5f4751259641b782ba621b89d939d1']), + resolver: new Set(['92031b6800bfb2356812d6fdc28af89143e1ffea37d2060879515d04667c5f3c']), + }; + function replaceExactly(source, anchor, replacement) { + assert.equal(source.split(anchor).length - 1, 1, `Diagnostic anchor must occur exactly once: ${anchor}`); + return source.replace(anchor, replacement); + } + registerHooks({ load(url, context, nextLoad) { + let type; + if (/\/server\/dist\/vendor\/paperclip-runner\/drivers\/codex\/codex-session-state\.js$/.test(url)) type = 'queue'; + else if (/\/server\/dist\/services\/native-runtime\/paperclip-control-plane-port\.js$/.test(url)) type = 'port'; + else if (/\/postgres\/src\/connection\.js$/.test(url)) type = 'postgres'; + else if (/\/server\/dist\/vendor\/paperclip-runner\/drivers\/codex\/codex-command\.js$/.test(url)) type = 'resolver'; + if (!type) return nextLoad(url, context); + const original = readFileSync(fileURLToPath(url), 'utf8'); + const sha256 = createHash('sha256').update(original).digest('hex'); + assert.ok(allowed[type].has(sha256), `Unknown compiled ${type} artifact: ${sha256}`); + let source = original; + if (type === 'queue') { + source = replaceExactly(source, 'class AsyncQueue {', `class AsyncQueue {\nconstructor() { ${call}.registerQueue(this); }\n__nativeIngestionDiagnosticDepth() { return this.#values.length; }`); + source = replaceExactly(source, 'const waiter = this.#waiters.shift();', `${call}.normalized(value);\nconst waiter = this.#waiters.shift();`); + source = replaceExactly(source, 'this.#values.push(value);', `{ this.#values.push(value); ${call}.observeQueue(); }`); + } else if (type === 'port') { + source = replaceExactly(source, 'await this.#onEventPersistence?.(', `${call}.postCommit(event, persisted.disposition);\nawait this.#onEventPersistence?.(`); + } else if (type === 'postgres') { + source = replaceExactly(source, "typeof options.debug === 'function' && options.debug(id, string, parameters, types)", `${call}.sql(string)\n typeof options.debug === 'function' && options.debug(id, string, parameters, types)`); + } else { + source = `export const resolveCodexCommand = () => ${JSON.stringify(providerCommand)}; export const resolvePinnedCodexCommand = resolveCodexCommand;`; + } + record({ kind: 'instrumented-module', type, sha256, file: fileURLToPath(url) }); + return { format: 'module', shortCircuit: true, source }; + } }); + const delay = monitorEventLoopDelay({ resolution: 20 }); delay.enable(); + let previousCpu = process.cpuUsage(), previousTime = performance.now(), previousElu = performance.eventLoopUtilization(); + const sample = () => { + const now = performance.now(), currentCpu = process.cpuUsage(), currentElu = performance.eventLoopUtilization(); + const memory = process.memoryUsage(), actualQueue = queueDepth(); + record({ kind: 'app-sample', intervalMs: now - previousTime, cpuMicros: currentCpu.user + currentCpu.system - previousCpu.user - previousCpu.system, + rssBytes: memory.rss, heapUsedBytes: memory.heapUsed, externalBytes: memory.external, + eventLoopUtilization: performance.eventLoopUtilization(currentElu, previousElu).utilization, + eventLoopDelayCount: delay.count, eventLoopDelayMaxMs: delay.count ? delay.max / 1e6 : null, + eventLoopDelayP95Ms: delay.count ? delay.percentile(95) / 1e6 : null, + actualNormalizedQueueDepth: actualQueue.depth, liveQueueObjects: actualQueue.active, queuePeakSinceSample, + counters: { ...counters }, sqlVerbs: { ...sqlVerbs }, fatalDiagnosticErrors: fatalErrors.length }); + queuePeakSinceSample = 0; previousCpu = currentCpu; previousTime = now; previousElu = currentElu; delay.reset(); + }; + setInterval(sample, 250).unref(); + process.once('exit', () => { sample(); flush(); }); + record({ kind: 'diagnostic-disclosure', queue: 'Actual private normalized AsyncQueue buffered values; excludes runtime collector, iterator pending read, in-flight append and transport queue', + sql: 'All PostgreSQL protocol query-build attempts, transaction and savepoint commands included; parameters and SQL text are never logged', + commit: 'Existing port persistence-observation callsite after outer database transaction resolves; callback timestamp is an upper bound on commit time', + overhead: 'Queue observation scans weak references after every buffered push; diagnostic overhead is intentionally excluded from acceptance measurements' }); +} diff --git a/scripts/native-ingestion/diagnostic-run.mjs b/scripts/native-ingestion/diagnostic-run.mjs new file mode 100644 index 00000000000..5939f160f0a --- /dev/null +++ b/scripts/native-ingestion/diagnostic-run.mjs @@ -0,0 +1,218 @@ +import assert from 'node:assert/strict'; +import { spawn, execFileSync } from 'node:child_process'; +import { createHash } from 'node:crypto'; +import { readdir, readFile, writeFile, mkdir, open } from 'node:fs/promises'; +import net from 'node:net'; +import path from 'node:path'; + +const arg = name => process.argv.find(value => value.startsWith(`--${name}=`))?.slice(name.length + 3); +const root = arg('root') ?? process.cwd(); +const harnessRoot = arg('harness-root') ?? root; +const trees = { B0: arg('b0-root') ?? '/private/tmp/paperclip-native-event-baseline', B1: root, + B2: arg('b2-root') ?? '/private/tmp/paperclip-native-event-batching' }; +const tree = arg('tree'), smoke = process.argv.includes('--smoke'); +const out = arg('output'); +if (!process.argv.includes('--execute')) { + console.log('PROFILE ONLY; no load launched. After quiet-window release: node scripts/native-ingestion/diagnostic-run.mjs --execute --tree=B0|B1|B2 --output=/tmp/ [--smoke]'); + process.exit(0); +} +assert.ok(trees[tree], 'Choose B0, B1 or B2'); assert.ok(out?.startsWith('/private/tmp/') || out?.startsWith('/tmp/')); +await mkdir(out, { recursive: false }); +const profiles = path.join(out, 'profile'); await mkdir(profiles); +const cpuProfileDirectory = path.join(profiles, 'cpu'); +if (!smoke) await mkdir(cpuProfileDirectory); +const getJson = async file => JSON.parse(await readFile(file, 'utf8')); +const saveJson = async (file, value) => writeFile(file, JSON.stringify(value, null, 2) + '\n'); +const readNdjson = async file => (await readFile(file, 'utf8')).split('\n').filter(Boolean).map(value => JSON.parse(value)); +const quote = value => `'${value.replaceAll("'", "'\\''")}'`; +const portServer = net.createServer(); +await new Promise((resolve, reject) => { portServer.once('error', reject); portServer.listen(0, '127.0.0.1', resolve); }); +const port = portServer.address().port; await new Promise(resolve => portServer.close(resolve)); +const bootstrap = path.join(harnessRoot, 'tests/e2e/fixtures/native-ingestion-bootstrap.mjs'); +const preload = path.join(import.meta.dirname, 'diagnostic-preload.mjs'); +const providerCommand = path.join(profiles, 'codex-profile-only'); +await writeFile(providerCommand, `#!/bin/sh\nexec env NODE_ENV=test PAPERCLIP_NATIVE_INGESTION_DIAGNOSTIC_DIR=${quote(profiles)} ${quote(process.execPath)} --import ${quote(preload)} ${quote(path.join(harnessRoot, 'tests/e2e/fixtures/in-feed-codex.mjs'))} --ingestion-stream-fixture "$@"\n`, { mode: 0o755 }); +const cpuProfileFlags = smoke ? '' : ` --cpu-prof --cpu-prof-dir=${quote(cpuProfileDirectory)}`; +const command = `node --import ./server/node_modules/tsx/dist/loader.mjs ${quote(bootstrap)} && node${cpuProfileFlags} --import ${quote(preload)} --import ./server/node_modules/tsx/dist/loader.mjs server/dist/index.js`; +const env = { PAPERCLIP_E2E_PORT: String(port), PAPERCLIP_INGESTION_SMOKE: smoke ? '1' : '0', + PAPERCLIP_INGESTION_VARIANT: `PROFILE-ONLY-${tree}`, PAPERCLIP_INGESTION_APP_ROOT: trees[tree], + PAPERCLIP_INGESTION_EVIDENCE_DIR: out, PAPERCLIP_NATIVE_INGESTION_DIAGNOSTIC_DIR: profiles, + PAPERCLIP_NATIVE_INGESTION_DIAGNOSTIC_CODEX_COMMAND: providerCommand, + PAPERCLIP_INGESTION_DRIVER_DEPENDENCIES: 'Driver-only tooling links to frozen B1 locked dependencies; each app owns independent installs', + PAPERCLIP_INGESTION_SERVER_COMMAND: command }; +const logHandle = await open(path.join(out, 'playwright.log'), 'wx'), log = logHandle.createWriteStream(); +const child = spawn(process.execPath, ['node_modules/@playwright/test/cli.js', 'test', '--config', 'tests/e2e/playwright-native-ingestion.config.ts'], + { cwd: harnessRoot, env: { ...process.env, ...env }, stdio: ['ignore', 'pipe', 'pipe'] }); +child.stdout.pipe(log, { end: false }); child.stderr.pipe(log, { end: false }); +const startedAtMs = Date.now(); +await saveJson(path.join(out, 'command.json'), { diagnosticOnly: true, tree, sourceRevision: execFileSync('git', ['rev-parse', 'HEAD'], { cwd: trees[tree], encoding: 'utf8' }).trim(), + childPid: child.pid, startedAtMs, env, preloadSha256: createHash('sha256').update(await readFile(preload)).digest('hex') }); +let metadata, clusterPid; +const processSamples = []; +const cpuMillis = value => value.split(':').map(Number).reduce((total, part) => total * 60 + part, 0) * 1000; +const clockTicks = process.platform === 'linux' ? Number(execFileSync('getconf', ['CLK_TCK'], { encoding: 'utf8' }).trim()) : null; +const sample = async () => { + if (!metadata) { + const names = (await readdir(profiles)).filter(name => /^app-\d+\.metadata\.json$/.test(name)); + if (!names.length) return; assert.equal(names.length, 1); metadata = await getJson(path.join(profiles, names[0])); + } + if (!clusterPid) { + try { + const config = await getJson(metadata.configPath), directory = config.database.embeddedPostgresDataDir; + assert.ok(directory.includes('/paperclip-e2e-home-'), 'Sampler may only inspect this disposable fixture database'); + clusterPid = Number((await readFile(path.join(directory, 'postmaster.pid'), 'utf8')).split('\n')[0]); + assert.ok(Number.isSafeInteger(clusterPid) && clusterPid > 1); + } catch (error) { if (error.code !== 'ENOENT') throw error; } + } + const output = execFileSync('ps', ['-axo', 'pid=,ppid=,time=,rss=,comm='], { encoding: 'utf8' }); + const processes = output.split('\n').flatMap(line => { + const match = line.match(/^\s*(\d+)\s+(\d+)\s+([\d:.]+)\s+(\d+)\s+(.+)$/); + return match ? [{ pid: Number(match[1]), parentPid: Number(match[2]), cpuMillis: cpuMillis(match[3]), rssBytes: Number(match[4]) * 1024, executable: path.basename(match[5]) }] : []; + }); + const ownedPg = new Set(clusterPid ? [clusterPid] : []); + for (let pass = 0; pass < 4; pass++) for (const proc of processes) if (ownedPg.has(proc.parentPid) && /^postgres/.test(proc.executable)) ownedPg.add(proc.pid); + const pg = processes.filter(proc => ownedPg.has(proc.pid) && /^postgres/.test(proc.executable)); + if (clockTicks) for (const proc of pg) { + try { + const stat = await readFile(`/proc/${proc.pid}/stat`, 'utf8'); + const fields = stat.slice(stat.lastIndexOf(')') + 2).trim().split(/\s+/); + proc.cpuMillis = (Number(fields[11]) + Number(fields[12])) / clockTicks * 1000; + } catch (error) { if (error.code !== 'ENOENT' && error.code !== 'ESRCH') throw error; } + } + const app = processes.find(proc => proc.pid === metadata.pid); + processSamples.push({ atMs: Date.now(), app, postgres: pg }); +}; +let sampling = false, samplingError; +const timer = setInterval(() => { + if (sampling) return; sampling = true; + sample().catch(error => { samplingError = String(error.stack ?? error); child.kill('SIGTERM'); }).finally(() => { sampling = false; }); +}, 1000); +const terminate = signal => child.kill(signal); +const sigint = () => terminate('SIGINT'), sigterm = () => terminate('SIGTERM'); +process.once('SIGINT', sigint); process.once('SIGTERM', sigterm); +const watchdog = setTimeout(() => { samplingError = '20-minute diagnostic resource watchdog expired'; child.kill('SIGTERM'); }, 20 * 60_000); +const exit = await new Promise((resolve, reject) => { child.once('error', reject); child.once('close', (code, signal) => resolve({ code, signal })); }); +clearInterval(timer); clearTimeout(watchdog); process.off('SIGINT', sigint); process.off('SIGTERM', sigterm); +while (sampling) await new Promise(resolve => setTimeout(resolve, 20)); +await new Promise(resolve => log.end(resolve)); +await saveJson(path.join(out, 'process-samples.json'), { diagnosticOnly: true, intervalMs: 1000, childExit: exit, samplingError, + cpuResolutionMillis: clockTicks ? 1000 / clockTicks : 10, + cpuSource: clockTicks ? 'Owned PostgreSQL /proc/PID/stat utime+stime ticks' : 'ps process CPU time, 10ms on this macOS host', + boundary: 'Task-owned app PID and embedded cluster PID plus PostgreSQL descendants; PID/PPID/CPU/RSS/comm only, no argv or credentials', + limitation: '1-second polling can miss short-lived backends and CPU accrued after their final observation; PostgreSQL CPU is sampled process CPU, not exact accounting', samples: processSamples }); +assert.equal(samplingError, undefined, 'Diagnostic process sampler failed; preserve process-samples.json'); +assert.equal(exit.code, 0, `Diagnostic fixture failed: ${JSON.stringify(exit)}; preserve log`); +const browser = await getJson(path.join(out, 'playwright.json')); +const attachments = []; +function visit(value) { + if (Array.isArray(value)) for (const child of value) visit(child); + else if (value && typeof value === 'object') { + if (value.attachments) attachments.push(...value.attachments.filter(item => item.name === 'native-ingestion-evidence')); + for (const [key, child] of Object.entries(value)) if (key !== 'attachments') visit(child); + } +} +visit(browser); assert.equal(attachments.length, 1); +const attachment = attachments[0]; +const evidence = attachment.body ? JSON.parse(Buffer.from(attachment.body, 'base64').toString()) : await getJson(attachment.path); +assert.equal(evidence.complete, true); +assert.equal(evidence.diagnosticOnly, true, 'The raw fixture attachment must identify profiling before extraction'); +assert.equal(evidence.performanceAcceptanceEligible, false); +await saveJson(path.join(out, 'PROFILE-ONLY-browser-evidence.json'), evidence); +const files = await readdir(profiles); +const rows = (await Promise.all(files.filter(file => file.endsWith('.ndjson')).map(file => readNdjson(path.join(profiles, file))))).flat(); +const moduleTypes = new Set(rows.filter(row => row.kind === 'instrumented-module').map(row => row.type)); +assert.deepEqual([...moduleTypes].sort(), ['port', 'postgres', 'queue', 'resolver'], 'All diagnostic anchors must actually be exercised'); +assert.equal(rows.filter(row => row.kind === 'diagnostic-fatal').length, 0, 'A diagnostic assertion failed inside the non-consequential observer'); +const emissions = rows.filter(row => row.kind === 'provider-emission'), commits = rows.filter(row => row.kind === 'postcommit-marker'); +const samples = rows.filter(row => row.kind === 'app-sample').sort((a, b) => a.atMs - b.atMs); +const cases = await Promise.all(evidence.cases.map(async item => { + const codes = new Set(item.durable.map(proof => proof.markerCode)); + const offered = emissions.filter(row => codes.has(row.markerCode)), persisted = commits.filter(row => codes.has(row.markerCode)); + const key = row => `${row.markerCode}:${row.index}`; + const originalClocks = item.clockEvidence.flatMap(stream => stream.emissions); + const originalByMarker = new Map(originalClocks.map(row => [key(row), row])); + assert.equal(originalByMarker.size, originalClocks.length, 'Duplicate original source-clock marker'); + const durableByMarker = new Map(item.durable.flatMap(proof => Array.from({ length: proof.individualDeltas }, (_, index) => + [`${proof.markerCode}:${index}`, { runId: proof.runId, nonce: proof.token, markerCode: proof.markerCode, index }]))); + const rawMarkersWithoutOriginalClock = offered.filter(row => !originalByMarker.has(key(row))).map(key); + if (!item.stop) assert.equal(rawMarkersWithoutOriginalClock.length, 0, 'Every normally emitted marker needs an original clock'); + assert.equal(new Set(offered.map(key)).size, offered.length, 'Duplicate provider emission marker'); + assert.equal(new Set(persisted.map(key)).size, persisted.length, 'Duplicate committed marker'); + const offeredKeys = new Set(offered.map(key)), persistedKeys = new Set(persisted.map(key)); + const markerCoverage = { + expectedDurableMarkers: durableByMarker.size, + missingPostcommitObservation: [...durableByMarker.keys()].filter(value => !persistedKeys.has(value)), + postcommitAbsentFromDurablePrefix: persisted.filter(row => !durableByMarker.has(key(row))).map(key), + postcommitWithoutProviderObservation: persisted.filter(row => !offeredKeys.has(key(row))).map(key), + postcommitWithoutOriginalClock: persisted.filter(row => !originalByMarker.has(key(row))).map(key), + mismatchedRunOrNonce: persisted.filter(row => { + const durable = durableByMarker.get(key(row)), original = originalByMarker.get(key(row)); + return durable && original && (row.runId !== durable.runId || original.nonce !== durable.nonce); + }).map(key), + }; + markerCoverage.complete = Object.entries(markerCoverage).every(([name, value]) => name === 'expectedDurableMarkers' || value.length === 0); + // Incomplete observation is distinct from a correctly observed, cancelled raw + // tail. Preserve the mismatch even if teardown killed a diagnostic observer. + await saveJson(path.join(out, `${item.name}-marker-coverage.json`), markerCoverage); + assert.equal(markerCoverage.complete, true, 'Incomplete diagnostic observer coverage; do not interpret its backlog as all durable commits'); + if (!item.stop) { + assert.equal(offered.length, item.runs * item.events); assert.equal(persisted.length, offered.length); + assert.deepEqual([...new Set(persisted.map(key))].sort(), [...new Set(offered.map(key))].sort()); + } + assert.ok(offered.length > 0 && persisted.length > 0); + const begin = Math.min(...originalClocks.map(row => row.intendedMs)), end = begin + item.drainThroughTerminalMs; + const during = samples.filter(row => row.atMs >= begin && row.atMs <= end); + assert.ok(during.length > 0, 'No app resource samples in measured diagnostic interval'); + const preceding = [...samples].reverse().find(row => row.atMs < begin), following = samples.find(row => row.atMs > end); + assert.ok(preceding && following, 'Resource counter boundaries were not captured'); + const deltas = Object.fromEntries(['sqlCalls', 'begins', 'commits', 'rollbacks', 'savepoints', 'releases', 'committedEvents'].map(name => [name, following.counters[name] - preceding.counters[name]])); + const changes = [...offered.map(row => ({ atMs: row.atMs, delta: 1 })), ...persisted.map(row => ({ atMs: row.atMs, delta: -1 }))].sort((a, b) => a.atMs - b.atMs || b.delta - a.delta); + let backlog = 0, peakBacklog = 0; + for (const change of changes) { backlog += change.delta; peakBacklog = Math.max(peakBacklog, backlog); } + const bracketed = samples.filter(row => row.atMs > preceding.atMs && row.atMs <= following.atMs); + const appCpuMicros = bracketed.reduce((total, row) => total + row.cpuMicros, 0); + const pgBefore = [...processSamples].reverse().find(row => row.atMs <= begin), pgAfter = processSamples.find(row => row.atMs >= end); + assert.ok(pgBefore && pgAfter, 'PG CPU needs an explicit before/after boundary'); + const pg = processSamples.filter(row => row.atMs > pgBefore.atMs && row.atMs <= pgAfter.atMs); + const previousByPid = new Map(pgBefore.postgres.map(proc => [proc.pid, proc.cpuMillis])); let pgCpuMillis = 0; + for (const row of pg) for (const proc of row.postgres) { + const previous = previousByPid.get(proc.pid); + pgCpuMillis += previous === undefined || proc.cpuMillis < previous ? proc.cpuMillis : proc.cpuMillis - previous; + previousByPid.set(proc.pid, proc.cpuMillis); + } + return { name: item.name, interval: { begin, end }, providerMarkers: offered.length, postcommitMarkers: persisted.length, + emittedToCommittedBacklogPeak: peakBacklog, finalUncommittedMarkers: backlog, originalClockCount: originalClocks.length, rawMarkersWithoutOriginalClock, markerCoverage, + finalBacklogInterpretation: item.stop ? 'Cancellation may intentionally leave raw unacknowledged output uncommitted under the preserved transport policy' : 'Must be zero after full terminal drainage', + actualNormalizedQueuePeakInBracketedWindow: Math.max(...bracketed.map(row => row.queuePeakSinceSample)), + actualNormalizedQueueDepthAtLastPreterminalSample: during.at(-1).actualNormalizedQueueDepth, + actualNormalizedQueueDepthAtPostterminalBoundary: following.actualNormalizedQueueDepth, + appCpuWindow: { begin: preceding.atMs, end: following.atMs }, pgCpuWindow: { begin: pgBefore.atMs, end: pgAfter.atMs }, + appCpuMicros, appCpuMicrosPerMarker: appCpuMicros / persisted.length, sampledPgCpuMillis: pgCpuMillis, + appCpuMicrosPerCommittedNormalizedEvent: appCpuMicros / deltas.committedEvents, + sampledPgCpuMillisPerMarker: pgCpuMillis / persisted.length, appRssPeakBytes: Math.max(...during.map(row => row.rssBytes)), + appHeapPeakBytes: Math.max(...during.map(row => row.heapUsedBytes)), + eventLoopDelayPeakMs: Math.max(...during.map(row => row.eventLoopDelayMaxMs ?? 0)), + eventLoopUtilizationMedian: during.map(row => row.eventLoopUtilization).sort((a, b) => a - b)[Math.floor(during.length / 2)], + databaseCounterDeltas: deltas, sqlCallsPerMarker: deltas.sqlCalls / persisted.length, + sqlCallsPerCommittedNormalizedEvent: deltas.sqlCalls / deltas.committedEvents, + transactionsBegunPerMarker: deltas.begins / persisted.length, + transactionsBegunPerCommittedNormalizedEvent: deltas.begins / deltas.committedEvents, + counterBoundaryOvershootMs: { before: begin - preceding.atMs, after: following.atMs - end }, resourceSampleCount: during.length }; +})); +const cpuProfiles = smoke ? [] : await Promise.all((await readdir(cpuProfileDirectory)).filter(file => file.endsWith('.cpuprofile')).map(async file => ({ + file: path.join(cpuProfileDirectory, file), sha256: createHash('sha256').update(await readFile(path.join(cpuProfileDirectory, file))).digest('hex'), +}))); +if (!smoke) assert.ok(cpuProfiles.length > 0, 'Full profile must capture the app-only Node CPU profile'); +const summary = { diagnosticOnly: true, performanceAcceptanceEligible: false, tree, sourceRevision: evidence.sourceRevision, + complete: true, cases, cpuProfiles, processScope: { appCpuAndMemory: 'Server Node process only; excludes Rust Runner/providers and the external load driver', postgresMemory: 'Per-process RSS; summing it double-counts shared pages' }, disclosures: ['CPU/RSS/heap describe the server Node process, not the Rust Runner, provider or whole host fleet', + 'PostgreSQL memory is per-process RSS; a sum would double-count shared pages and is not unique physical memory', + 'Actual normalized queue excludes iterator/collector/in-flight append and raw transport buffers', + 'Raw provider-observer emission-minus-postcommit is a full-path backlog proxy, distinct from the actual normalized queue', + 'Original intended/emitted clocks come from the separate streamed HTTP source channel in browser evidence; no clocks are synthesized for cancelled unmatched raw tail', + 'Resource samples use 250ms app / 1s process polling. CPU/SQL phase boundaries are approximate, with overshoot reported', + 'SQL counts are query-build attempts across the whole app including API reads, lifecycle work, COMMIT and savepoints; no parameters or SQL text are logged', + 'Bracketed whole-app CPU/SQL windows can include postterminal verification reads; per-marker ratios are diagnostic whole-app costs, not append-only costs', + 'Postcommit observation happens after the durable outer transaction, so its timestamp is an upper bound', + 'Full diagnostics also run app-only Node --cpu-prof; CPU sampling and observer overhead are excluded from acceptance timing claims'] }; +await saveJson(path.join(out, 'PROFILE-ONLY-summary.json'), summary); +console.log(JSON.stringify(summary, null, 2)); diff --git a/scripts/native-ingestion/run-frozen.mjs b/scripts/native-ingestion/run-frozen.mjs new file mode 100644 index 00000000000..0b62db97c81 --- /dev/null +++ b/scripts/native-ingestion/run-frozen.mjs @@ -0,0 +1,200 @@ +import assert from 'node:assert/strict'; +import { spawn, execFileSync } from 'node:child_process'; +import { createHash } from 'node:crypto'; +import { readFile, writeFile, mkdir, open } from 'node:fs/promises'; +import net from 'node:net'; +import path from 'node:path'; + +// Automation only. The acceptance harness and browser protocol remain frozen. +// Nothing launches unless --execute AND an explicit --stage are supplied. +const option = name => process.argv.find(value => value.startsWith(`--${name}=`))?.slice(name.length + 3); +const root = option('root') ?? process.cwd(); +const harnessRoot = option('harness-root') ?? root; +const trees = { B0: option('b0-root') ?? '/private/tmp/paperclip-native-event-baseline', B1: root, + B2: option('b2-root') ?? '/private/tmp/paperclip-native-event-batching' }; +const revisions = { B0: '83b5f7d486e70e484acd9a41bc957054f4e10175', B1: '7922802f069ab3b39e9f79368c46615ba2201ba9', B2: '6afeba4edafd47120e8117f736641d5424b27d4a' }; +const out = option('output-root') ?? (process.platform === 'darwin' ? '/private/tmp/native-ingestion-frozen-evidence' : '/tmp/native-ingestion-frozen-evidence'); +const serviceHash = '48e967a5e7a8c803ff3bf4890a37ff70564ba1a618d83bcff5bb110b35ee8133'; +const browserHash = '902c6a208793f0bcde1b326a1fee46d85df3274117754e3bfbbf77a65dc730ee'; +const comparatorHash = 'e50d958532906d037413db6f66bdc02e60c93ddcacfe2e18e5b731d302f6a785'; +const harnessRevision = 'f5c560d80d9d4ee089297d3b3f300c052db17c9e'; +const gatesHash = 'ded8bcbb8f05732348f6520b9c5e41b7c35bc098997eb263a9dcb174eede75fc'; +// Preregistered by the lead before fresh measurements on 2026-11-03. +const aaLimits = { maximumPairedRelativeDifference: 0.20, maximumMedianArmDrift: 0.10 }; +const protocol = ['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']; +const hash = value => createHash('sha256').update(value).digest('hex'); +const getJson = async file => JSON.parse(await readFile(file, 'utf8')); +const saveJson = async (file, value) => writeFile(file, JSON.stringify(value, null, 2) + '\n'); +const stages = ['service-aa-b0', 'service-b0-b1', 'service-aa-b1', 'service-b1-b2', 'service-b0-b2', + 'browser-smoke-b0', 'browser-smoke-b1', 'browser-smoke-b2', 'browser-full-qualification-b0', + 'browser-aa-b0', 'browser-b0-b1', 'browser-aa-b1', 'browser-b1-b2', 'browser-b0-b2']; +const commands = []; +let commandIndex = 0; + +function service(stage, tree, baseline, phase, noise) { + return { stage, cwd: trees[tree], args: ['--import', './server/node_modules/tsx/dist/loader.mjs', 'scripts/benchmark-native-event-ingestion.ts', + `--baseline-ref=${revisions[baseline]}`, `--phase=${phase}`, `--output=${out}/${stage}.json`, + ...(noise ? [`--noise-report=${out}/${noise}.json`, '--require-acceptance'] : ['--require-repeatability'])] }; +} +const serviceCommands = { + 'service-aa-b0': service('service-aa-b0', 'B1', 'B0', 'aa'), + 'service-b0-b1': service('service-b0-b1', 'B1', 'B0', 'compare', 'service-aa-b0'), + 'service-aa-b1': service('service-aa-b1', 'B2', 'B1', 'aa'), + 'service-b1-b2': service('service-b1-b2', 'B2', 'B1', 'batch', 'service-aa-b1'), + 'service-b0-b2': service('service-b0-b2', 'B2', 'B0', 'batch', 'service-aa-b0'), +}; + +async function checkFrozen() { + assert.notEqual(process.env.PAPERCLIP_INGESTION_BROWSER_CAPTURE, '1', 'Browser diagnostic recording cannot enter timed acceptance'); + assert.equal(execFileSync('git', ['rev-parse', 'HEAD'], { cwd: harnessRoot, encoding: 'utf8' }).trim(), harnessRevision, 'Browser harness revision changed'); + assert.equal(process.env.PAPERCLIP_NATIVE_INGESTION_DIAGNOSTIC_DIR, undefined, 'Profile-only environment cannot enter an unprofiled acceptance run'); + assert.ok(!/native-ingestion-diagnostic|diagnostic-preload/.test(process.env.NODE_OPTIONS ?? ''), 'Diagnostic preload cannot enter acceptance'); + for (const [label, tree] of Object.entries(trees)) { + assert.equal(execFileSync('git', ['rev-parse', 'HEAD'], { cwd: tree, encoding: 'utf8' }).trim(), revisions[label], `Revision changed: ${label}`); + const stamp = await getJson(path.join(tree, 'server/dist/build-info.json')); + assert.equal(stamp.commit, revisions[label], `Compiled build does not match: ${label}`); + } + for (const tree of [trees.B1, trees.B2]) { + assert.equal(hash(await readFile(path.join(tree, 'scripts/benchmark-native-event-ingestion.ts'))), serviceHash); + assert.equal(hash(await readFile(path.join(tree, 'scripts/native-ingestion-performance-gates.mjs'))), gatesHash); + } + assert.equal(hash(await readFile(path.join(harnessRoot, 'scripts/compare-native-ingestion-browser.ts'))), comparatorHash); + assert.equal(hash(Buffer.concat(await Promise.all(protocol.map(file => readFile(path.join(harnessRoot, 'tests/e2e', file)))))), browserHash); +} + +async function execute(step) { + const index = ++commandIndex, startedAt = new Date().toISOString(); + const log = path.join(out, `${step.stage}.log`); + const command = { index, stage: step.stage, cwd: step.cwd, executable: process.execPath, args: step.args, + environmentOverrides: step.env ?? {}, startedAt, log }; + commands.push(command); await saveJson(path.join(out, `${option('stage')}-commands.json`), commands); + console.log(`START ${step.stage} ${startedAt}; log=${log}`); + const handle = await open(log, 'wx'); + const stream = handle.createWriteStream(); + const child = spawn(process.execPath, step.args, { cwd: step.cwd, env: { ...process.env, ...step.env }, stdio: ['ignore', 'pipe', 'pipe'] }); + command.childPid = child.pid; + await saveJson(path.join(out, `${option('stage')}-commands.json`), commands); + const interrupt = () => child.kill('SIGINT'), terminate = () => child.kill('SIGTERM'); + process.once('SIGINT', interrupt); process.once('SIGTERM', terminate); + const watchdogMillis = step.stage.startsWith('service-') ? 30 * 60_000 : 20 * 60_000; + const watchdog = setTimeout(() => { command.watchdogExpired = true; child.kill('SIGTERM'); }, watchdogMillis); + child.stdout.pipe(stream, { end: false }); child.stderr.pipe(stream, { end: false }); + const exit = await new Promise((resolve, reject) => { child.once('error', reject); child.once('close', (code, signal) => resolve({ code, signal })); }); + clearTimeout(watchdog); process.off('SIGINT', interrupt); process.off('SIGTERM', terminate); + await new Promise(resolve => stream.end(resolve)); + Object.assign(command, { finishedAt: new Date().toISOString(), ...exit }); + await saveJson(path.join(out, `${option('stage')}-commands.json`), commands); + console.log(`END ${step.stage} code=${exit.code} signal=${exit.signal ?? 'none'}`); + assert.equal(command.watchdogExpired, undefined, 'Resource watchdog expired; run is incomplete, not acceptance'); + assert.equal(exit.code, 0, `Stage ${step.stage} failed; preserve ${log} and diagnose before continuing`); + return command; +} + +async function qualifyAa(file) { + const report = await getJson(file); + assert.equal(report.candidateSourceHashes['scripts/benchmark-native-event-ingestion.ts'], serviceHash); + assert.equal(report.candidateSourceHashes['scripts/native-ingestion-performance-gates.mjs'], gatesHash); + const { evaluateServiceRepeatability } = await import(path.join(root, 'scripts/native-ingestion-performance-gates.mjs')); + const result = { report: file, ...evaluateServiceRepeatability(report) }; + await saveJson(file.replace(/\.json$/, '.repeatability.json'), result); + assert.equal(result.passed, true, 'A/A repeatability failed; no candidate comparison may start'); +} + +async function freePort() { + const server = net.createServer(); + await new Promise((resolve, reject) => { server.once('error', reject); server.listen(0, '127.0.0.1', resolve); }); + const port = server.address().port; + await new Promise(resolve => server.close(resolve)); + return port; +} + +async function extractEvidence(directory) { + const report = await getJson(path.join(directory, 'playwright.json')); + const attachments = []; + const visit = node => { + if (Array.isArray(node)) for (const child of node) visit(child); + else if (node && typeof node === 'object') { + if (Array.isArray(node.attachments)) attachments.push(...node.attachments.filter(value => value.name === 'native-ingestion-evidence')); + for (const [key, child] of Object.entries(node)) if (key !== 'attachments') visit(child); + } + }; + visit(report); assert.equal(attachments.length, 1, 'Exactly one complete native-ingestion evidence attachment is required'); + const attachment = attachments[0]; + const evidence = attachment.body ? JSON.parse(Buffer.from(attachment.body, 'base64').toString()) : await getJson(attachment.path); + assert.equal(evidence.complete, true); assert.equal(evidence.measurementSha256, browserHash); + assert.equal(evidence.protocolVersion, 'v2-compact-markers-original-clock-stream'); + assert.equal(evidence.harnessProvenance?.sourceRevision, harnessRevision); + assert.equal(evidence.harnessProvenance?.sourceStatus, ''); + const file = path.join(directory, 'native-ingestion-evidence.json'); await saveJson(file, evidence); + return { file, evidence }; +} + +async function browserRun(stage, tree, smoke) { + const directory = path.join(out, stage); await mkdir(directory, { recursive: false }); + const port = await freePort(); + await execute({ stage, cwd: harnessRoot, args: ['node_modules/@playwright/test/cli.js', 'test', '--config', 'tests/e2e/playwright-native-ingestion.config.ts'], + env: { PAPERCLIP_E2E_PORT: String(port), PAPERCLIP_INGESTION_SMOKE: smoke ? '1' : '0', PAPERCLIP_INGESTION_VARIANT: `${stage}-${tree}`, + PAPERCLIP_INGESTION_APP_ROOT: trees[tree], PAPERCLIP_INGESTION_EVIDENCE_DIR: directory, + PAPERCLIP_INGESTION_DRIVER_DEPENDENCIES: 'Driver-only tooling links to frozen B1 locked dependencies; measured applications each use independent locked installs' } }); + const { file, evidence } = await extractEvidence(directory); + assert.equal(evidence.sourceRevision, revisions[tree]); assert.equal(evidence.compiledBuildRevision, revisions[tree]); + assert.equal(evidence.serverArtifact, 'compiled'); assert.equal(evidence.qualificationOnly, smoke); + assert.equal(evidence.diagnosticOnly, false); assert.equal(evidence.performanceAcceptanceEligible, true); + if (!smoke) { + const loaded = evidence.cases.find(item => item.name === 'burst512'); + assert.equal(loaded.runs, 100); assert.equal(loaded.participants, 100); assert.equal(loaded.durable.length, 100); + } + // An invocation is complete only when its own HTTP server has released the port. + const listener = net.createServer(); + await new Promise((resolve, reject) => { listener.once('error', reject); listener.listen(port, '127.0.0.1', resolve); }); + await new Promise(resolve => listener.close(resolve)); + return file; +} + +async function browserSeries(stage, baseline, candidate, aa = false, expectation = 'improvement') { + const manifest = [], file = path.join(out, `${stage}-pairs.json`); + if (!aa) { + const noise = await getJson(path.join(out, `browser-aa-${baseline.toLowerCase()}.json`)); + assert.equal(noise.complete, true); assert.equal(noise.phase, 'aa'); + } + for (let pair = 0; pair < 5; pair++) { + const entry = {}; + const arms = pair % 2 ? [['candidate', candidate], ['baseline', baseline]] : [['baseline', baseline], ['candidate', candidate]]; + for (const [arm, tree] of arms) entry[`${arm}File`] = await browserRun(`${stage}-pair${pair + 1}-${arm}`, tree, false); + manifest.push(entry); await saveJson(file, manifest); + } + await execute({ stage: `${stage}-comparison`, cwd: harnessRoot, args: ['scripts/compare-native-ingestion-browser.ts', `--phase=${aa ? 'aa' : 'compare'}`, + `--expect=${expectation}`, `--pairs=${file}`, `--output=${out}/${stage}.json`, + ...(aa ? [] : [`--noise-report=${out}/browser-aa-${baseline.toLowerCase()}.json`, '--require-acceptance'])] }); +} + +const stage = option('stage'); +if (!process.argv.includes('--execute')) { + console.log(JSON.stringify({ mode: 'DRY RUN; no load launched', revisions, out, aaLimits, stages, serviceCommands, + note: 'Execute one released stage at a time with --execute --stage=. Do not run builds/tests concurrently. Browser series perform five fresh alternating pairs, never reuse remote baseline samples.' }, null, 2)); + process.exit(0); +} +assert.ok(stages.includes(stage), 'Choose an explicit --stage from the dry run'); +await mkdir(out, { recursive: true }); +await checkFrozen(); +const stageReceipt = await open(path.join(out, `${stage}.started.json`), 'wx'); +await stageReceipt.writeFile(JSON.stringify({ stage, pid: process.pid, startedAt: new Date().toISOString(), revisions }, null, 2) + '\n'); +await stageReceipt.close(); +// Never overwrite evidence. A retry must use a new output root or a new stage name. +if (serviceCommands[stage]) { + const step = serviceCommands[stage]; + const prior = step.args.find(value => value.startsWith('--noise-report=')); + if (prior) await qualifyAa(prior.slice('--noise-report='.length)); + await execute(step); + if (stage.startsWith('service-aa-')) await qualifyAa(path.join(out, `${stage}.json`)); +} else if (stage.startsWith('browser-smoke-')) { + await browserRun(stage, stage.slice(-2).toUpperCase(), true); +} else if (stage === 'browser-full-qualification-b0') { + await browserRun(stage, 'B0', false); +} else if (stage === 'browser-aa-b0') await browserSeries(stage, 'B0', 'B0', true); +else if (stage === 'browser-aa-b1') await browserSeries(stage, 'B1', 'B1', true); +else if (stage === 'browser-b0-b1') await browserSeries(stage, 'B0', 'B1', false, 'nonregression'); +else if (stage === 'browser-b1-b2') await browserSeries(stage, 'B1', 'B2'); +else if (stage === 'browser-b0-b2') await browserSeries(stage, 'B0', 'B2'); +console.log(`COMPLETE ${stage}`); diff --git a/scripts/native-ingestion/verify-delivery-runtime.py b/scripts/native-ingestion/verify-delivery-runtime.py new file mode 100644 index 00000000000..f9bd57b771e --- /dev/null +++ b/scripts/native-ingestion/verify-delivery-runtime.py @@ -0,0 +1,63 @@ +#!/usr/bin/env python3 +"""Read-only Git comparison for the native-ingestion delivery branches.""" +import argparse +import hashlib +import json +import re +import subprocess + +parser = argparse.ArgumentParser() +parser.add_argument("--repo", required=True) +parser.add_argument("--frozen", required=True) +parser.add_argument("--delivery", required=True) +args = parser.parse_args() +for ref in [args.frozen, args.delivery]: + if not re.fullmatch(r"[a-f0-9]{40}", ref): + parser.error("Use immutable full commit IDs") + +CI_BASE = "8ab89f41e499032e3e55d261f457a3463d754a1d" +SCRIPT_EXCEPTIONS = {"test:e2e:runner:typecheck", "test:lifecycle-baseline:typecheck"} +CI_ONLY_SERVER_FILES = {"server/scripts/typecheck-layers.mjs"} + +def git(*parts): + return subprocess.check_output(["git", *parts], cwd=args.repo) + +def is_runtime(path, expanded=False): + prefixes = ("server/" if expanded else "server/src/", "packages/", "ui/", "cli/", "skills/") + return (path.startswith(prefixes) + and not any(part in ("__tests__", "tests", "test") for part in path.split("/")) + and not path.endswith((".test.ts", ".test.tsx", ".test.js", ".test.mjs", ".spec.ts", ".spec.tsx", ".spec.js")) + and path not in CI_ONLY_SERVER_FILES) + +def manifest(ref, expanded=False): + rows = [] + for entry in git("ls-tree", "-r", "-z", ref).split(b"\0"): + if not entry: + continue + metadata, name = entry.split(b"\t", 1) + path = name.decode() + if is_runtime(path, expanded): + rows.append([path, *metadata.decode().split(" ")]) + rows.sort() + package = json.loads(git("show", ref + ":package.json")) + content = {"files": rows, + "rootPackageNonScripts": {key: value for key, value in package.items() if key != "scripts"}, + "rootScriptsExceptTwoTestTypechecks": {key: value for key, value in package["scripts"].items() if key not in SCRIPT_EXCEPTIONS}} + digest = hashlib.sha256(json.dumps(content, sort_keys=True, separators=(",", ":")).encode()).hexdigest() + return {"count": len(rows), "sha256": digest} + +output = {"frozen": args.frozen, "delivery": args.delivery, "sourceMutation": False} +for name, expanded in [("approvedScope", False), ("expandedServerBuildInputs", True)]: + frozen, delivery = manifest(args.frozen, expanded), manifest(args.delivery, expanded) + assert frozen == delivery, (name, frozen, delivery) + output[name] = {"equal": True, "frozen": frozen, "delivery": delivery} + +approved_package = json.loads(git("show", CI_BASE + ":package.json")) +delivery_package = json.loads(git("show", args.delivery + ":package.json")) +for key in SCRIPT_EXCEPTIONS: + assert delivery_package["scripts"][key] == approved_package["scripts"][key], key +for path in ["scripts/typecheck-low-memory.mjs", "scripts/typecheck-test-projects.mjs", *CI_ONLY_SERVER_FILES]: + assert git("show", args.delivery + ":" + path) == git("show", CI_BASE + ":" + path), path +output["realCiHelpersMatchApproved8ab"] = True +output["allChangedPaths"] = git("diff", "--name-only", args.frozen, args.delivery).decode().splitlines() +print(json.dumps(output, indent=2))