diff --git a/scripts/compare-native-ingestion-browser.ts b/scripts/compare-native-ingestion-browser.ts index 53fbb6ff03d..f9e3f206586 100644 --- a/scripts/compare-native-ingestion-browser.ts +++ b/scripts/compare-native-ingestion-browser.ts @@ -107,6 +107,10 @@ const workloads = [ ]; const names = workloads.map(item => item.name); const median = (values: number[]) => [...values].sort((a, b) => a - b)[Math.floor(values.length / 2)]!; +const pooledMedian = (values: number[]) => { + const sorted = [...values].sort((a, b) => a - b), middle = Math.floor(sorted.length / 2); + return sorted.length % 2 ? sorted[middle]! : (sorted[middle - 1]! + sorted[middle]!) / 2; +}; const first = pairs[0]; const sha256 = (value: unknown) => typeof value === "string" && /^[a-f0-9]{64}$/.test(value); const uuid = (value: unknown) => typeof value === "string" && /^[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}$/i.test(value); @@ -463,27 +467,112 @@ if (h5 && phase === "compare" && candidatePolicy === "final-b2") { } } const metric = (item: any, key: string): number => key.split(".").reduce((value, part) => value[part], item); +// Prospective main-campaign rule. The 2ms floor is timestamp contrast +// representation only: it never pads observed noise or a regression gate. +const browserAaRule = Object.freeze({ pairCount: 5, pairedFraction: 0.20, medianDriftFraction: 0.10, + contrastRepresentationFloorMs: 2, representationLimitedMedianMs: 20 }); +function assertArmChronology(rows: any[]) { + assert.equal(rows?.length, 10, "Browser A/A/comparison requires ten actual owning invocations"); + const sorted = [...rows].sort((a,b) => a.startedAtMs - b.startedAtMs); + const companies = new Set(), runIds = new Set(), owners = new Set(); + for (const [index, row] of sorted.entries()) { + assert.ok(Number.isSafeInteger(row.startedAtMs) && row.startedAtMs > 0 && Number.isFinite(row.durationMs) + && row.durationMs > 0, "Missing finite actual owning timestamp/duration"); + assert.equal(row.finishedAtMs, row.startedAtMs + row.durationMs); + if (index) assert.ok(sorted[index - 1].finishedAtMs <= row.startedAtMs, "Browser owning invocations overlap"); + assert.ok(sha256(row.ownerReportSha256) && !owners.has(row.ownerReportSha256), "Reused owning report"); owners.add(row.ownerReportSha256); + assert.ok(uuid(row.companyId) && !companies.has(row.companyId), "Each browser arm needs its fresh owned company"); companies.add(row.companyId); + assert.ok(Array.isArray(row.runIds) && row.runIds.length > 0); + for (const runId of row.runIds) { + assert.ok(uuid(runId) && !runIds.has(runId), "Browser arms reuse an owned run"); runIds.add(runId); + } + const pairIndex = Math.floor(index / 2), firstArm = pairIndex % 2 ? "candidate" : "baseline"; + assert.equal(row.pairIndex, pairIndex, "Owner chronology must match registered pair order"); + assert.equal(row.arm, index % 2 ? firstArm === "baseline" ? "candidate" : "baseline" : firstArm, + "Browser pairs must alternate baseline-first then candidate-first"); + } + return sorted; +} +const armChronology = h5 ? assertArmChronology(pairs.flatMap((pair, pairIndex) => (["baseline", "candidate"] as const).map(arm => { + const report = pair[arm], owner = report.owningPlaywright, startedAtMs = Date.parse(owner.startedAt); + return { pairIndex, arm, startedAtMs, durationMs: owner.durationMs, finishedAtMs: startedAtMs + owner.durationMs, + ownerReportSha256: owner.reportSha256, companyId: report.companyId, runIds: report.cases.flatMap((item: any) => item.runIds) }; +}))) : undefined; +function assessAaMetric(samples: any[], chronology: any[]) { + assert.equal(samples.length, 5); + for (const sample of samples) { + assert.ok(Number.isFinite(sample.baseline) && sample.baseline > 0 && Number.isFinite(sample.candidate) && sample.candidate >= 0); + assert.equal(sample.improvementMs, sample.baseline - sample.candidate, "A/A paired absolute contrast was forged"); + assert.equal(sample.improvementRelative, sample.improvementMs / sample.baseline, "A/A paired relative contrast was forged"); + } + const pairedDifferencesMs = samples.map(sample => Math.abs(sample.improvementMs)); + const pairedLimitsMs = samples.map(sample => Math.max(browserAaRule.pairedFraction * sample.baseline, browserAaRule.contrastRepresentationFloorMs)); + const baselineMedianMs = median(samples.map(sample => sample.baseline)), candidateMedianMs = median(samples.map(sample => sample.candidate)); + const chronologicalValues = chronology.map(row => samples[row.pairIndex][row.arm]); + const firstFiveMedianMs = median(chronologicalValues.slice(0,5)), lastFiveMedianMs = median(chronologicalValues.slice(5)); + const chronologicalMedianDriftMs = Math.abs(lastFiveMedianMs - firstFiveMedianMs); + const chronologicalDriftLimitMs = Math.max(browserAaRule.medianDriftFraction * firstFiveMedianMs, browserAaRule.contrastRepresentationFloorMs); + const roleMedianDriftMs = Math.abs(candidateMedianMs - baselineMedianMs); + const roleDriftLimitMs = Math.max(browserAaRule.medianDriftFraction * baselineMedianMs, browserAaRule.contrastRepresentationFloorMs); + const pooledMedianMs = pooledMedian(samples.flatMap(sample => [sample.baseline, sample.candidate])); + return { sampleCount: 10, pairCount: 5, pairedDifferencesMs, pairedLimitsMs, baselineMedianMs, candidateMedianMs, + pooledMedianMs, firstFiveMedianMs, lastFiveMedianMs, chronologicalMedianDriftMs, chronologicalDriftLimitMs, + roleMedianDriftMs, roleDriftLimitMs, representationLimited: pooledMedianMs <= browserAaRule.representationLimitedMedianMs, + observedNoiseAbsoluteMs: Math.max(...pairedDifferencesMs), + observedNoiseRelative: Math.max(...samples.map(sample => Math.abs(sample.improvementRelative))), + passed: pairedDifferencesMs.every((value,index) => value <= pairedLimitsMs[index]) + && chronologicalMedianDriftMs <= chronologicalDriftLimitMs && roleMedianDriftMs <= roleDriftLimitMs }; +} const measurements = names.flatMap(name => { const observed = first.baseline.cases.find((item: any) => item.name === name); const keys = ["firstVisibleAfterEmissionMs", "emissionToVisible.medianMs", "drainThroughTerminalMs", ...(name === "paced" ? ["interChunkGaps.maxMs"] : []), ...(observed.emissionToVisible.p95Ms === undefined ? [] : ["emissionToVisible.p95Ms", "intendedToVisible.p95Ms"]), - ...(observed.stop ? ["stopThroughTerminalMs", "stopThroughAcknowledgementMs"] : [])]; + ...(observed.stop ? ["stopThroughTerminalMs", "stopThroughAcknowledgementMs"] : []), + ...(h5 && !observed.stop ? ["h5.independentDrain.elapsedFromEpochMs"] : [])]; return keys.map(key => { const samples = pairs.map(pair => { const baseline = metric(pair.baseline.cases.find((item: any) => item.name === name), key); const candidate = metric(pair.candidate.cases.find((item: any) => item.name === name), key); assert.ok(Number.isFinite(baseline) && baseline > 0 && Number.isFinite(candidate) && candidate >= 0, `Invalid latency ${name}.${key}`); - return { baseline, candidate, improvementRelative: (baseline - candidate) / baseline }; + return { baseline, candidate, improvementMs: baseline - candidate, improvementRelative: (baseline - candidate) / baseline }; }); + const pooledMedianMs = pooledMedian(samples.flatMap(sample => [sample.baseline, sample.candidate])); + const representationLimited = h5 && pooledMedianMs <= browserAaRule.representationLimitedMedianMs; + const aaSuitability = h5 && phase === "aa" ? assessAaMetric(samples, armChronology!) : undefined; return { name, metric: key, baselineMedianMs: median(samples.map(sample => sample.baseline)), candidateMedianMs: median(samples.map(sample => sample.candidate)), - medianPairedImprovementRelative: median(samples.map(sample => sample.improvementRelative)), + medianPairedImprovementRelative: representationLimited ? null : median(samples.map(sample => sample.improvementRelative)), + ...(h5 ? { pooledMedianMs, representationLimited, percentageGainClaimAllowed: !representationLimited, + sub2msOrP99GainClaimAllowed: false, medianPairedImprovementMs: median(samples.map(sample => sample.improvementMs)), + aaSuitability } : {}), candidateWins: samples.filter(sample => sample.improvementRelative > 0).length, aaNoiseEnvelopeRelative: phase === "aa" ? Math.max(...samples.map(sample => Math.abs(sample.improvementRelative))) : undefined, + ...(h5 ? { aaNoiseEnvelopeAbsoluteMs: phase === "aa" ? Math.max(...samples.map(sample => Math.abs(sample.improvementMs))) : undefined } : {}), samples }; }); }); +// A descriptive rate from the independently verified completion upper bound. +// Gate duration against its own A/A noise; invent no separate rate threshold. +const verifiedCompletionRates = h5 ? names.filter(name => !first.baseline.cases.find((item: any) => item.name === name).stop).map(name => { + const samples = pairs.map(pair => { + const baseline = pair.baseline.cases.find((item: any) => item.name === name); + const candidate = pair.candidate.cases.find((item: any) => item.name === name); + const counts = [baseline, candidate].map(item => item.h5.independentDrain.deltaCount); + assert.equal(counts[0], counts[1], "Verified completion rates require equal matched measured-delta counts"); + assert.ok([baseline, candidate].every(item => item.h5.independentDrain.elapsedFromEpochMs > 0), + "Verified completion rate needs positive observed duration"); + return { measuredDeltas: counts[0], baselineElapsedMs: baseline.h5.independentDrain.elapsedFromEpochMs, + candidateElapsedMs: candidate.h5.independentDrain.elapsedFromEpochMs, + baselineDeltasPerSecond: 1000 * counts[0] / baseline.h5.independentDrain.elapsedFromEpochMs, + candidateDeltasPerSecond: 1000 * counts[1] / candidate.h5.independentDrain.elapsedFromEpochMs }; + }); + return { name, sampleCount: samples.length, samples, + baselineMedianDeltasPerSecond: median(samples.map(sample => sample.baselineDeltasPerSecond)), + candidateMedianDeltasPerSecond: median(samples.map(sample => sample.candidateDeltasPerSecond)), + gatedDurationMetric: "h5.independentDrain.elapsedFromEpochMs", + scope: "Matched measured normalized deltas / independently verified completion upper bound; includes oracle, reload, screenshot, API and lease verification; not exact last-commit throughput" }; +}) : undefined; const retryObservations = names.map(name => { const samples = pairs.map(pair => Object.fromEntries((["baseline", "candidate"] as const).map(arm => { const item = pair[arm].cases.find((value: any) => value.name === name); @@ -525,6 +614,9 @@ const result: Record = { comparatorSha256: createHash("sha256").update(await readFile(new URL(import.meta.url))).digest("hex"), manifestSha256: createHash("sha256").update(await readFile(manifestPath)).digest("hex"), measurements, primaryLoadedWorkload: "burst512", complete: true, retryObservations, + ...(h5 ? { verifiedCompletionRates, armChronology, + ...(phase === "aa" ? { aaSuitability: { rule: browserAaRule, metricCount: measurements.length, + passed: measurements.every(item => item.aaSuitability?.passed) } } : {}) } : {}), ...(h5 ? { observationMode: H5_MODE, workloadSha256: H5_WORKLOAD_SHA256, candidatePolicy, h5Assessments, artifactInventories: { baseline: first.baseline.armInventory, candidate: first.candidate.armInventory }, originalH4eConfirmation, originalProtocolAcceptance: "Baseline SLA misses are retained; H5c complete observation is not an original-deadline PASS", @@ -550,16 +642,55 @@ if (phase === "compare") { assert.equal(noise.gatesSha256, result.gatesSha256, "Acceptance gates changed after A/A calibration"); assert.equal(noise.comparatorSha256, result.comparatorSha256, "Comparator changed after A/A calibration"); assert.ok(noise.pairCount >= 5); + if (h5) { + assert.equal(noise.pairCount, 5); assert.deepEqual(noise.aaSuitability?.rule, browserAaRule, "Browser A/A suitability rule changed"); + const chronology = assertArmChronology(noise.armChronology); + assert.deepEqual(chronology, noise.armChronology, "Frozen A/A chronology was reordered"); + assert.ok(armChronology![0].startedAtMs >= chronology.at(-1).finishedAtMs, + "Comparison must follow all owning calibration invocations"); + const priorOwners = new Set(chronology.map(row => row.ownerReportSha256)); + const priorCompanies = new Set(chronology.map(row => row.companyId)); + const priorRuns = new Set(chronology.flatMap(row => row.runIds)); + for (const arm of armChronology!) { + assert.ok(!priorOwners.has(arm.ownerReportSha256) && !priorCompanies.has(arm.companyId) + && arm.runIds.every((runId: string) => !priorRuns.has(runId)), "Comparison reuses a calibration owner/company/run"); + } + assert.equal(noise.measurements?.length, measurements.length, "Frozen A/A is missing a predeclared metric"); + assert.equal(noise.aaSuitability.metricCount, measurements.length); + for (const item of measurements) { + const controls = noise.measurements.filter((entry: any) => entry.name === item.name && entry.metric === item.metric); + assert.equal(controls.length, 1, "Frozen A/A needs exactly one calibration per metric"); + const control = controls[0], actual = assessAaMetric(control.samples, chronology); + assert.equal(actual.passed, true, "Unsuitable browser A/A cannot freeze noise"); + assert.deepEqual(control.aaSuitability, actual, "Browser A/A suitability was forged"); + assert.equal(control.aaNoiseEnvelopeAbsoluteMs, actual.observedNoiseAbsoluteMs, "Browser absolute noise was forged"); + assert.equal(control.aaNoiseEnvelopeRelative, actual.observedNoiseRelative, "Browser relative noise was forged"); + assert.equal(control.representationLimited, actual.representationLimited, "Browser representation flag was forged"); + } + assert.equal(noise.aaSuitability.passed, true, "Unsuitable browser A/A cannot freeze noise"); + } result.gates = measurements.map(item => { const control = noise.measurements.find((entry: any) => entry.name === item.name && entry.metric === item.metric); assert.ok(control); const improvementRequired = expectation === "improvement" && item.name === "burst512" - && ["emissionToVisible.p95Ms", "intendedToVisible.p95Ms", "drainThroughTerminalMs"].includes(item.metric); - const noRegression = -item.medianPairedImprovementRelative <= control.aaNoiseEnvelopeRelative; - const paired = noiseExceedingPairedWins(item.samples.map(sample => sample.improvementRelative), control.aaNoiseEnvelopeRelative); + && ["emissionToVisible.p95Ms", "intendedToVisible.p95Ms", "drainThroughTerminalMs", + "h5.independentDrain.elapsedFromEpochMs"].includes(item.metric); + const representationLimited = h5 && (control.representationLimited || item.representationLimited); + if (representationLimited) { + item.representationLimited = true; item.percentageGainClaimAllowed = false; item.medianPairedImprovementRelative = null; + } + const medianImprovementRelative = median(item.samples.map(sample => sample.improvementRelative)); + const medianImprovementMs = median(item.samples.map(sample => sample.improvementMs)); + const noRegression = representationLimited ? -medianImprovementMs <= control.aaNoiseEnvelopeAbsoluteMs + : -medianImprovementRelative <= control.aaNoiseEnvelopeRelative; + const paired = noiseExceedingPairedWins(item.samples.map(sample => h5 && sample.improvementMs <= 2 ? 0 : sample.improvementRelative), control.aaNoiseEnvelopeRelative); const improvement = paired.pairedWinRequirementPassed - && item.medianPairedImprovementRelative > control.aaNoiseEnvelopeRelative; + && medianImprovementRelative > control.aaNoiseEnvelopeRelative && (!h5 || medianImprovementMs > 2); return { name: item.name, metric: item.metric, improvementRequired, noiseEnvelopeRelative: control.aaNoiseEnvelopeRelative, + ...(h5 ? { noiseEnvelopeAbsoluteMs: control.aaNoiseEnvelopeAbsoluteMs, representationLimited, + percentageGainClaimAllowed: !representationLimited, sub2msOrP99GainClaimAllowed: false, + nonregressionScale: representationLimited ? "observed-absolute-ms" : "observed-relative", + requiredAbsoluteImprovementAboveMs: improvementRequired ? 2 : undefined } : {}), noiseExceedingPairedWins: paired.noiseExceedingPairedWins, requiredPairedWins: paired.requiredPairedWins, passed: improvementRequired ? improvement : noRegression }; }); @@ -588,6 +719,8 @@ if (phase === "compare") { } if (arg("output")) await writeFile(arg("output")!, JSON.stringify(result, null, 2)); console.log(JSON.stringify(result, null, 2)); +if (h5 && phase === "aa") assert.equal(result.aaSuitability.passed, true, + "Browser A/A suitability failed; retain all ten arms, diagnose and review a fresh calibration"); if (process.argv.includes("--require-performance")) { assert.equal(h5, true, "H5c performance-only gating is explicit; default H4e keeps original deadlines"); assert.equal(phase, "compare"); diff --git a/scripts/native-ingestion-h5.test.mjs b/scripts/native-ingestion-h5.test.mjs index 0c04e043eca..a54364447d3 100644 --- a/scripts/native-ingestion-h5.test.mjs +++ b/scripts/native-ingestion-h5.test.mjs @@ -3,23 +3,20 @@ import { test } from 'node:test'; import { createRequire } from 'node:module'; import { readFile } from 'node:fs/promises'; import { h5Enabled, H5_MODE, observeStreaming, originalTerminal, requireH5Candidate, - H5_LIMITS, ORIGINAL_BUDGETS, prepareParticipantAgents, observeRunSettlement } from '../tests/e2e/fixtures/native-ingestion-h5.mjs'; + H5_LIMITS, ORIGINAL_BUDGETS, prepareParticipantAgents, observeRunSettlement, collectDurableRunEvents, assertDurableRunEvents, + assertPreludeEvents, assertVisibleMarkerClocks } from '../tests/e2e/fixtures/native-ingestion-h5.mjs'; const timeout = milliseconds => Object.assign(new Error(`Timeout ${milliseconds}ms exceeded while waiting on the predicate`), { matcherResult: { name: 'toBeGreaterThanOrEqual', pass: false } }); -test('the actual paginated result counter counts accepted results in default H4e and observation H5c', async () => { - const source = await readFile(new URL('../tests/e2e/native-event-ingestion.journey.ts', import.meta.url), 'utf8'); - const statements = source.split('\n').filter(line => line.includes('resultAcceptedEvents++')); - assert.equal(statements.length, 1, 'Exercise the actual event-loop counter statement'); - const count = new Function('h5', 'rows', `let resultAcceptedEvents = 0; - for (const row of rows) { const event = row.payload?.prpEvent; ${statements[0]} } return resultAcceptedEvents;`); - for (const h5 of [false, true]) { - assert.equal(count(h5, [{ eventType: 'run.result.accepted', payload: { prpEvent: { eventType: 'run.result.accepted' } } }]), 1); - assert.equal(count(h5, [{ payload: { prpEvent: { eventType: 'run.result.accepted' } } }]), 1); - assert.equal(count(h5, [{ eventType: 'run.result.proposed' }, { payload: { prpEvent: { eventType: 'item.delta' } } }]), 0); - assert.equal(count(h5, []), 0); - } +test('the actual durable collector counts each accepted row once in default H4e and observation H5c', () => { + for (const row of [ + { eventType: 'run.result.accepted', payload: { prpEvent: { eventType: 'run.result.accepted' } } }, + { payload: { prpEvent: { eventType: 'run.result.accepted' } } }, + { eventType: 'run.result.accepted' }]) assert.equal(collectDurableRunEvents([row], 'q7').resultAcceptedEvents, 1); + assert.equal(collectDurableRunEvents([{ eventType: 'run.result.proposed' }, + { payload: { prpEvent: { eventType: 'item.delta' } } }], 'q7').resultAcceptedEvents, 0); + assert.equal(collectDurableRunEvents([], 'q7').resultAcceptedEvents, 0); }); test('real frozen Playwright tiny poll exposes the matcher deadline shape used by H5', async () => { @@ -231,3 +228,87 @@ test('cursor inherits only paired exact known misses; final candidate inherits n assert.throws(() => requireH5Candidate('final-b2', base, base), /every original SLA/); assert.throws(() => requireH5Candidate('final-b2', slowInvocation, slowInvocation), /every original SLA/); }); + + +const durableFixture = ({ stop = false, count = 4 } = {}) => { + const runId = 'run-1', token = 'provider-1', markerCode = 'q7'; + const rows = Array.from({ length: count }, (_, index) => ({ seq: index+1, payload: { prpEvent: { + eventType: 'item.delta', runId, sourceInstanceId: 'source-1', sourceEventId: `event-${index}`, sourceSeq: index+1, + payload: { text: `[${markerCode}${index.toString(36)}] ` } } } })); + if (!stop) rows.push({ seq: count+1, eventType: 'run.result.accepted' }); + return { rows, input: { runId, token, markerCode, events: 4, stop, requireNativeIdentity: true, + sourceClocks: new Map(Array.from({ length: 4 }, (_, index) => [index, { nonce: token, markerCode, index, + intendedMs: 1000+index*25, emittedMs: 1000+index*25, receivedAtMs: 1001+index*25 }])), + visible: Array.from({ length: stop ? count : 4 }, (_, index) => ({ index, markerCode, visibleMs: 1003+index*25 })) } }; +}; +const checkDurable = fixture => assertDurableRunEvents({ ...fixture.input, + collected: collectDurableRunEvents(fixture.rows, fixture.input.markerCode) }); + +test('durable oracle retains complete per-run evidence in H4e and H5c and permits only the canonical Stop prefix', () => { + for (const requireNativeIdentity of [false,true]) { + const fixture = durableFixture(); fixture.input.requireNativeIdentity = requireNativeIdentity; + assert.deepEqual(checkDurable(fixture), { runId: 'run-1', token: 'provider-1', markerCode: 'q7', individualDeltas: 4, + uniqueSourceIdentities: 4, matchedSourceClocks: 4, ordered: true, resultAcceptedEvents: 1 }); + } + const stopped = durableFixture({ stop: true, count: 2 }); + assert.equal(checkDurable(stopped).individualDeltas, 2); + assert.equal(checkDurable(stopped).resultAcceptedEvents, 0); + assert.equal(stopped.input.sourceClocks.size, 4, 'Raw emissions outside the durable Stop prefix remain permitted'); +}); +test('durable oracle rejects corrupted payloads, coalescing, replay identity conflicts, gaps and foreign runs', () => { + const changes = [ + fixture => { fixture.rows[0].payload.prpEvent.payload.text = '[q70] corruption'; }, + fixture => { fixture.rows[0].payload.prpEvent.payload.text = '[q70] [q71] '; }, + fixture => { fixture.rows[1].payload.prpEvent.sourceEventId = 'event-0'; fixture.rows[1].payload.prpEvent.sourceSeq = 1; }, + fixture => { fixture.rows.splice(1,1); }, fixture => { fixture.rows.splice(1,0,fixture.rows[0]); }, + fixture => { fixture.rows.reverse(); }, fixture => { fixture.rows[0].payload.prpEvent.runId = 'foreign-run'; }, + fixture => { fixture.rows[0].payload.prpEvent.sourceInstanceId = ''; }, + fixture => { fixture.rows[0].payload.prpEvent.sourceEventId = ''; }, + fixture => { fixture.rows[0].payload.prpEvent.sourceSeq = 0; }, + fixture => { fixture.rows[0].payload.prpEvent.sourceSeq = '1'; }, + ]; + for (const change of changes) { + const fixture = durableFixture(); change(fixture); + assert.throws(() => checkDurable(fixture), /Durable run run-1/, 'Failure must identify its exact run'); + } +}); +test('durable oracle rejects wrong or missing original clock joins, wrong source counts and result counts', () => { + for (const change of [ + fixture => { fixture.input.sourceClocks.delete(0); }, + fixture => { fixture.input.sourceClocks.get(0).nonce = 'foreign-source'; }, + fixture => { fixture.input.sourceClocks.get(0).markerCode = 'z9'; }, + fixture => { fixture.input.sourceClocks.get(0).index = 1; }, + fixture => { fixture.input.sourceClocks.set(9, { nonce: 'provider-1', markerCode: 'q7', index: 9 }); }, + fixture => { fixture.rows.pop(); }, fixture => { fixture.rows.push({ eventType: 'run.result.accepted' }); }, + ]) { const fixture = durableFixture(); change(fixture); assert.throws(() => checkDurable(fixture), /Durable run run-1/); } +}); +test('durable Stop oracle rejects visible markers outside the ordered durable prefix or an accepted result', () => { + const missing = durableFixture({ stop: true, count: 2 }); missing.input.visible.push({ index: 2 }); + assert.throws(() => checkDurable(missing), /visible Stop marker 2/); + const gap = durableFixture({ stop: true, count: 2 }); gap.rows.shift(); assert.throws(() => checkDurable(gap), /canonical Stop prefix/); + const result = durableFixture({ stop: true, count: 2 }); result.rows.push({ eventType: 'run.result.accepted' }); + assert.throws(() => checkDurable(result), /exact accepted result count/); +}); +test('prelude oracle preserves exact setup payloads, source tuple uniqueness, count and prelude-only selection', () => { + const rows = [0,1,2].map(index => ({ payload: { prpEvent: { eventType: 'item.delta', itemId: 'prelude-turn', + sourceInstanceId: 'source-1', sourceEventId: `event-${index}`, sourceSeq: index+1, + payload: { text: index === 2 ? ' PRELUDE_provider-1_READY ' : '.' } } } })); + const check = selected => assertPreludeEvents({ rows: selected, runId: 'run-1', token: 'provider-1', expectedEvents: 3 }); + assert.deepEqual(check([...rows,{ payload: { prpEvent: { eventType: 'item.delta', itemId: 'measured-turn' } } }]), + { individualDeltas: 3, uniqueSourceIdentities: 3 }); + assert.throws(() => check(rows.slice(1)), /Prelude run run-1/); + const bad = structuredClone(rows); bad[0].payload.prpEvent.payload.text = '..'; assert.throws(() => check(bad), /complete normalized payload/); + const replay = structuredClone(rows); Object.assign(replay[1].payload.prpEvent, { sourceEventId: 'event-0', sourceSeq: 1 }); + assert.throws(() => check(replay), /unique source identity tuples/); +}); +test('visible-clock oracle preserves original two-frame timestamps and rejects missing clocks or incomplete unique output', () => { + const fixture = durableFixture(), { visible: seenMarkers, sourceClocks } = fixture.input; + const check = (seen = seenMarkers, clocks = sourceClocks, stop = false) => assertVisibleMarkerClocks({ + seenMarkers: seen, sourceClocks: clocks, markerCode: 'q7', expectedEvents: 4, stop }); + const visible = check(); assert.deepEqual(visible.map(row => row.visibleMs), seenMarkers.map(row => row.visibleMs)); + assert.deepEqual(visible.map(row => row.emittedMs), [...sourceClocks.values()].map(row => row.emittedMs)); + assert.throws(() => check(seenMarkers, new Map()), /Missing original provider clock/); + assert.throws(() => check(seenMarkers.slice(1)), /complete unique foreground count/); + assert.throws(() => check([seenMarkers[0],seenMarkers[0],seenMarkers[2],seenMarkers[3]]), /complete unique foreground count/); + assert.equal(check(seenMarkers.slice(0,2), sourceClocks, true).length, 2); +}); diff --git a/scripts/native-ingestion-performance-gates.test.mjs b/scripts/native-ingestion-performance-gates.test.mjs index 0c0d75ee1ac..6b818188106 100644 --- a/scripts/native-ingestion-performance-gates.test.mjs +++ b/scripts/native-ingestion-performance-gates.test.mjs @@ -234,7 +234,7 @@ test("H5 assessor retains an original SLA failure assessed before its nominal st assert.throws(() => requireH5Candidate("final-b2", outcome, outcome), /every original SLA/); }); -test("H5 CLI retains baseline SLA failure, rejects final SLA misses and requires complete owning evidence", async () => { +test("H5 CLI retains baseline SLA failure, rejects final SLA misses and requires complete owning evidence", async context => { const directory = await mkdtemp(path.join(os.tmpdir(), "native-ingestion-h5-gates-")); const script = new URL("./compare-native-ingestion-browser.ts", import.meta.url); const save = async (name, value) => { const file = path.join(directory, name); await writeFile(file, JSON.stringify(value)); return file; }; @@ -271,11 +271,17 @@ test("H5 CLI retains baseline SLA failure, rejects final SLA misses and requires for (let index = 0; index < 5; index++) { const entry = {}; for (const arm of ["baseline", "candidate"]) { - const report = h5Report(arm === "baseline" || sameRevision ? baselineRevision : candidateRevision, arm === "baseline" ? 180000 : candidateLatency); + const ordinal = 2 * index + (arm === "baseline" ? index % 2 : index % 2 ? 0 : 1); + const startedAtMs = Date.parse(sameRevision ? "2026-11-02T00:00:00.000Z" : "2026-11-03T00:00:00.000Z") + ordinal * 2000000; + const ownedIdPrefix = createHash("sha256").update(prefix).digest("hex").slice(0,6) + ordinal.toString(16).padStart(2,"0"); + const report = JSON.parse(JSON.stringify(h5Report(arm === "baseline" || sameRevision ? baselineRevision : candidateRevision, + arm === "baseline" ? 180000 : candidateLatency)).replaceAll("00000000-", `${ownedIdPrefix}-`), (key, value) => + typeof value === "number" && (key.endsWith("AtMs") || ["epochMs", "deadlineMs", "readinessDeadlineMs", "visibleMs", "intendedMs", "emittedMs"].includes(key)) ? value + startedAtMs : value); + report.bodyFinishedAtMs = startedAtMs + duration - 1; const owner = { timeout: 1200000, expectedStatus: "passed", results: [{ status: "passed", retry: 0, - duration, startTime: "2026-11-03T00:00:00.000Z", errors: [], attachments: [] }] }; + duration, startTime: new Date(startedAtMs).toISOString(), errors: [], attachments: [] }] }; if (index === 0 && arm === "candidate") mutate?.(report, owner); - mutateArm?.(report, { index, arm }); + mutateArm?.(report, { index, arm, owner }); owner.results[0].attachments.push({ name: "native-ingestion-evidence", contentType: "application/json", body: Buffer.from(JSON.stringify(report)).toString("base64") }); entry[`${arm}File`] = await save(`${prefix}-${index}-${arm}.json`, report); @@ -299,11 +305,90 @@ test("H5 CLI retains baseline SLA failure, rejects final SLA misses and requires const originalInventory = await inventory(proof, originalProof, "original-h4c"); const finalProofArgs = [`--original-h4e-proof=${originalProof}`, `--original-h4e-inventory=${originalInventory}`]; const noise = path.join(directory, "aa.json"); - const aa = run(["--phase=aa", `--pairs=${await manifest("aa", 181000, { sameRevision: true })}`, `--output=${noise}`]); + const aaManifest = await manifest("aa", 181000, { sameRevision: true }); + const aa = run(["--phase=aa", `--pairs=${aaManifest}`, `--output=${noise}`]); assert.equal(aa.status, 0, aa.stderr); const aaReport = JSON.parse(await readFile(noise, "utf8")); assert.equal(aaReport.h5Assessments.length, 5); + assert.equal(aaReport.measurements.filter(item => item.metric === "h5.independentDrain.elapsedFromEpochMs").length, 5, + "Every non-Stop case needs its own verified-completion A/A noise"); + assert(aaReport.measurements.find(item => item.name === "burst512" && item.metric === "h5.independentDrain.elapsedFromEpochMs").aaNoiseEnvelopeRelative > 0); assert.deepEqual(aaReport.h5Assessments[0].baseline.misses, ["burst512:streaming120"]); + assert.equal(aaReport.aaSuitability.passed, true); assert.equal(aaReport.armChronology.length, 10); + assert.equal(new Set(aaReport.armChronology.map(arm => arm.companyId)).size, 10); + for (const [name, mutateArm, expectation] of [ + ["aa-paired-jitter", (report, { arm, index }) => { if (arm === "candidate" && index === 4) report.cases[0].firstVisibleAfterEmissionMs = 1300; }, "paired"], + ["aa-chronological-drift", (report, { index }) => { report.cases[0].firstVisibleAfterEmissionMs = 1000 + index * 60; }, "chronological"], + ["aa-role-bias", (report, { arm }) => { report.cases[0].firstVisibleAfterEmissionMs = arm === "candidate" ? 1120 : 1000; }, "role"], + ]) { + const output = path.join(directory, `${name}.json`); + const attempt = run(["--phase=aa", `--pairs=${await manifest(name, 180000, { sameRevision: true, mutateArm })}`, `--output=${output}`]); + assert.notEqual(attempt.status, 0, name); assert.match(attempt.stderr, /A\/A suitability failed/); + const unsuitable = JSON.parse(await readFile(output, "utf8")); + assert.equal(unsuitable.aaSuitability.passed, false); assert.equal(unsuitable.armChronology.length, 10); + const metric = unsuitable.measurements.find(item => item.name === "paced" && item.metric === "firstVisibleAfterEmissionMs").aaSuitability; + if (expectation === "paired") assert(metric.pairedDifferencesMs.some((value,index) => value > metric.pairedLimitsMs[index])); + if (expectation === "chronological") { assert.equal(metric.observedNoiseAbsoluteMs, 0); assert(metric.chronologicalMedianDriftMs > metric.chronologicalDriftLimitMs); } + if (expectation === "role") assert(metric.roleMedianDriftMs > metric.roleDriftLimitMs); + } + for (const [name, mutateArm] of [ + ["aa-missing-owner-time", (_report, { arm, index, owner }) => { if (arm === "candidate" && index === 0) delete owner.results[0].startTime; }], + ["aa-overlap", (_report, { arm, index, owner }) => { if (arm === "candidate" && index === 0) owner.results[0].startTime = "2026-11-02T00:00:00.001Z"; }], + ["aa-wrong-order", (_report, { arm, index, owner }) => { if (index === 1) owner.results[0].startTime = new Date(Date.parse("2026-11-02T00:00:00.000Z") + (arm === "baseline" ? 4 : 5) * 1000000).toISOString(); }], + ["aa-reused-company-run", (report) => { const prefix = report.companyId.slice(0,8); Object.assign(report, JSON.parse(JSON.stringify(report).replaceAll(`${prefix}-`, "00000000-"))); }], + ]) { + const attempt = run(["--phase=aa", `--pairs=${await manifest(name, 180000, { sameRevision: true, mutateArm })}`]); + assert.notEqual(attempt.status, 0, name); + } + for (const [name, forge] of [ + ["forged-absolute-noise", noise => { noise.measurements[0].aaNoiseEnvelopeAbsoluteMs = 9999; }], + ["forged-relative-noise", noise => { noise.measurements[0].aaNoiseEnvelopeRelative = 9; }], + ["forged-representation", noise => { noise.measurements[0].representationLimited = !noise.measurements[0].representationLimited; }], + ["forged-suitability", noise => { noise.measurements[0].aaSuitability.pairedLimitsMs[0] = 9999; }], + ["forged-sample", noise => { noise.measurements[0].samples[0].improvementMs++; }], + ["forged-rule", noise => { noise.aaSuitability.rule.contrastRepresentationFloorMs = 1000; }], + ]) { + const forged = structuredClone(aaReport); forge(forged); + const attempt = run(["--expect=nonregression", "--candidate-policy=cursor-nonregression", + `--pairs=${await manifest(name, 180000, { candidateRevision: "B1" })}`, `--noise-report=${await save(`${name}-noise.json`, forged)}`, "--require-performance"]); + assert.notEqual(attempt.status, 0, name); assert.match(attempt.stderr, /forged|rule changed/); + } + for (const repeat of [1,2]) { + const shortNoise = path.join(directory, `short-${repeat}-noise.json`); + const aa = run(["--phase=aa", `--pairs=${await manifest(`short-${repeat}-aa`, 180000, { sameRevision: true, + mutateArm: (report, { arm }) => { report.cases[0].firstVisibleAfterEmissionMs = arm === "candidate" ? repeat : 1; } })}`, `--output=${shortNoise}`]); + assert.equal(aa.status, 0, aa.stderr); + const calibrated = JSON.parse(await readFile(shortNoise, "utf8")); + const control = calibrated.measurements.find(item => item.name === "paced" && item.metric === "firstVisibleAfterEmissionMs"); + assert.equal(control.representationLimited, true); assert.equal(control.percentageGainClaimAllowed, false); + assert.equal(control.medianPairedImprovementRelative, null); assert.equal(control.aaNoiseEnvelopeAbsoluteMs, repeat - 1); + const output = path.join(directory, `short-${repeat}-regression.json`); + const attempt = run(["--expect=nonregression", "--candidate-policy=cursor-nonregression", + `--pairs=${await manifest(`short-${repeat}-regression`, 180000, { candidateRevision: "B1", mutateArm: (report, { arm }) => { + report.cases[0].firstVisibleAfterEmissionMs = repeat === 1 ? arm === "candidate" ? 2 : 1 : arm === "candidate" ? 12 : 10; + } })}`, `--noise-report=${shortNoise}`, `--output=${output}`, "--require-performance"]); + assert.notEqual(attempt.status, 0, "The suitability floor must not pad actual observed absolute noise"); + const compared = JSON.parse(await readFile(output, "utf8")); + const gate = compared.gates.find(item => item.name === "paced" && item.metric === "firstVisibleAfterEmissionMs"); + assert.equal(gate.noiseEnvelopeAbsoluteMs, repeat - 1); assert.equal(gate.representationLimited, true); assert.equal(gate.passed, false); + if (repeat === 2) { + const improving = path.join(directory, "short-calibrated-improvement.json"); + const changed = run(["--expect=nonregression", "--candidate-policy=cursor-nonregression", + `--pairs=${await manifest("short-calibrated-improvement", 180000, { candidateRevision: "B1", mutateArm: (report, { arm }) => { + report.cases[0].firstVisibleAfterEmissionMs = arm === "candidate" ? 1 : 100; + } })}`, `--noise-report=${shortNoise}`, `--output=${improving}`, "--require-performance"]); + assert.equal(changed.status, 0, changed.stderr); + const measured = JSON.parse(await readFile(improving, "utf8")).measurements.find(item => item.name === "paced" && item.metric === "firstVisibleAfterEmissionMs"); + assert.equal(measured.representationLimited, true); assert.equal(measured.percentageGainClaimAllowed, false); + assert.equal(measured.medianPairedImprovementRelative, null, "Short calibration cannot turn into a percentage gain claim"); + } + } + const replayedOwners = JSON.parse(await readFile(await manifest("reused-aa-owner", 180000, { candidateRevision: "B1" }), "utf8")); + const oldOwners = JSON.parse(await readFile(aaManifest, "utf8")); + for (const field of ["baselineFile", "baselinePlaywrightFile", "baselineInventoryFile"]) replayedOwners[0][field] = oldOwners[0][field]; + const replay = run(["--expect=nonregression", "--candidate-policy=cursor-nonregression", + `--pairs=${await save("reused-aa-owner-pairs.json", replayedOwners)}`, `--noise-report=${noise}`, "--require-performance"]); + assert.notEqual(replay.status, 0); assert.match(replay.stderr, /calibration|reus/); const cursor = run(["--expect=nonregression", "--candidate-policy=cursor-nonregression", `--pairs=${await manifest("cursor", 180000, { candidateRevision: "B1" })}`, `--noise-report=${noise}`, "--require-acceptance"]); assert.equal(cursor.status, 0, cursor.stderr); @@ -318,6 +403,69 @@ test("H5 CLI retains baseline SLA failure, rejects final SLA misses and requires const goodPairs = await manifest("good", 1000); const good = run(["--candidate-policy=final-b2", `--pairs=${goodPairs}`, `--noise-report=${noise}`, ...finalProofArgs, "--require-acceptance"]); assert.equal(good.status, 0, good.stderr); + const completionRegressionPairs = await manifest("verified-drain-regression", 1000, { + mutateArm: (report, { arm }) => { + if (arm !== "candidate") return; + const item = report.cases[3]; + Object.assign(item.h5.independentDrain, { elapsedFromEpochMs: 200000, verifiedAtMs: item.epochMs + 200000 }); + } }); + const completionRegressionOutput = path.join(directory, "verified-drain-regression.json"); + const completionRegression = run(["--candidate-policy=final-b2", `--pairs=${completionRegressionPairs}`, + `--noise-report=${noise}`, `--output=${completionRegressionOutput}`, "--require-performance"]); + const completionRegressionReport = JSON.parse(await readFile(completionRegressionOutput, "utf8")); + context.diagnostic(JSON.stringify({ regression: "browser and terminal faster but independently verified completion slower", + comparatorSha256: completionRegressionReport.comparatorSha256, exit: completionRegression.status, + performancePassed: completionRegressionReport.performancePassed, baselineVerifiedMs: 180010, candidateVerifiedMs: 200000, + verifiedCompletionGate: completionRegressionReport.gates.find(gate => gate.name === "burst512" + && gate.metric === "h5.independentDrain.elapsedFromEpochMs") ?? null })); + assert.notEqual(completionRegression.status, 0, "Faster browser/terminal must not conceal slower fully verified completion"); + const regressionGate = completionRegressionReport.gates.find(gate => gate.name === "burst512" + && gate.metric === "h5.independentDrain.elapsedFromEpochMs"); + assert.equal(regressionGate.improvementRequired, true); assert.equal(regressionGate.passed, false); + assert(completionRegressionReport.gates.filter(gate => gate.name === "burst512" && + ["emissionToVisible.p95Ms", "intendedToVisible.p95Ms", "drainThroughTerminalMs"].includes(gate.metric)).every(gate => gate.passed)); + const goodReport = JSON.parse(good.stdout); + const rate = goodReport.verifiedCompletionRates.find(item => item.name === "burst512"); + assert.equal(rate.samples[0].measuredDeltas, 51200); + assert.equal(rate.samples[0].baselineDeltasPerSecond, 51200000 / 180010); + assert.equal(rate.samples[0].candidateDeltasPerSecond, 51200000 / 1010); + assert(rate.candidateMedianDeltasPerSecond > rate.baselineMedianDeltasPerSecond); + for (const beyondNoise of [3,4]) { + const output = path.join(directory, `verified-drain-${beyondNoise}-wins.json`); + const attempt = run(["--candidate-policy=final-b2", `--pairs=${await manifest(`verified-drain-${beyondNoise}-wins`, 1000, { + mutateArm: (report, { arm, index }) => { + if (arm !== "candidate") return; + const item = report.cases[3], elapsed = index < beyondNoise ? 90000 : 179500; + Object.assign(item.h5.independentDrain, { elapsedFromEpochMs: elapsed, verifiedAtMs: item.epochMs + elapsed }); + } })}`, `--noise-report=${noise}`, `--output=${output}`, "--require-performance"]); + assert.equal(attempt.status === 0, beyondNoise === 4, attempt.stderr); + const compared = JSON.parse(await readFile(output, "utf8")); + const gate = compared.gates.find(item => item.name === "burst512" && item.metric === "h5.independentDrain.elapsedFromEpochMs"); + assert.equal(gate.noiseExceedingPairedWins, beyondNoise); assert.equal(gate.requiredPairedWins, 4); + assert.equal(gate.passed, beyondNoise === 4); + } + const zeroNoise = path.join(directory, "zero-noise.json"); + const zeroAA = run(["--phase=aa", `--pairs=${await manifest("zero-noise-aa", 180000, { sameRevision: true })}`, `--output=${zeroNoise}`]); + assert.equal(zeroAA.status, 0, zeroAA.stderr); + for (const resolvedPairs of [0,3,4]) { + const output = path.join(directory, `timestamp-gain-${resolvedPairs}.json`); + const attempt = run(["--candidate-policy=final-b2", `--pairs=${await manifest(`timestamp-gain-${resolvedPairs}`, 1000, { + mutateArm: (report, { arm, index }) => { + if (arm !== "candidate") return; + const item = report.cases[3], gain = resolvedPairs === 0 ? 1 : index < resolvedPairs ? 3 : 2; + Object.assign(item.h5.independentDrain, { elapsedFromEpochMs: 180010 - gain, verifiedAtMs: item.epochMs + 180010 - gain }); + } })}`, `--noise-report=${zeroNoise}`, `--output=${output}`, "--require-performance"]); + assert.equal(attempt.status === 0, resolvedPairs === 4, attempt.stderr); + const gate = JSON.parse(await readFile(output, "utf8")).gates.find(item => item.name === "burst512" + && item.metric === "h5.independentDrain.elapsedFromEpochMs"); + assert.equal(gate.noiseExceedingPairedWins, resolvedPairs); assert.equal(gate.requiredAbsoluteImprovementAboveMs, 2); + } + const cursorCompletionRegression = run(["--expect=nonregression", "--candidate-policy=cursor-nonregression", + `--pairs=${await manifest("cursor-verified-drain-regression", 180000, { candidateRevision: "B1", mutateArm: (report, { arm }) => { + if (arm !== "candidate") return; + const item = report.cases[3]; Object.assign(item.h5.independentDrain, { elapsedFromEpochMs: 200000, verifiedAtMs: item.epochMs + 200000 }); + } })}`, `--noise-report=${noise}`, "--require-performance"]); + assert.notEqual(cursorCompletionRegression.status, 0, "Cursor nonregression must include fully verified completion"); const wrongPolicy = run(["--expect=nonregression", "--candidate-policy=cursor-nonregression", `--pairs=${goodPairs}`, `--noise-report=${noise}`, "--require-acceptance"]); assert.notEqual(wrongPolicy.status, 0); assert.match(wrongPolicy.stderr, /Cursor policy requires B0→B1/); @@ -447,6 +595,8 @@ test("H5 CLI retains baseline SLA failure, rejects final SLA misses and requires ["stop-recorded-provider-failure", report => { report.cases[5].recordedFailures.rows[0].current_error_code = "provider_transport_failed"; }, /recorded failed run must have recovered/], ["censored", report => { report.cases[3].h5.streaming.observation.censored = true; }, /false/], ["terminal-failed", report => { report.cases[3].terminalStatuses[99].status = "failed"; }, /successful drainage/], + ["missing-independent-drain", report => { delete report.cases[3].h5.independentDrain; }, /independentDrain|verifiedAtMs|Cannot read/], + ["tampered-independent-drain", report => { report.cases[3].h5.independentDrain.elapsedFromEpochMs++; }, /strictly equal/], ["result-missing", report => { report.cases[3].durable[99].resultAcceptedEvents = 0; }, /persisted result/], ["missing-delta", report => { report.cases[3].durable[99].individualDeltas--; }, /individualDeltas|result.events|512/], ["wrong-workload", report => { report.workloadSha256 = "old"; }, /registration changed/], diff --git a/tests/e2e/fixtures/native-ingestion-h5.md b/tests/e2e/fixtures/native-ingestion-h5.md index 94c16439628..a503f1cdca8 100644 --- a/tests/e2e/fixtures/native-ingestion-h5.md +++ b/tests/e2e/fixtures/native-ingestion-h5.md @@ -305,3 +305,165 @@ queue/backlog, delay and memory with identical capture hooks. Aborted profiles remain diagnostic-only and cannot establish complete CPU/event costs. No CPU requirement or profiling is added to this timing comparator. Local results do not establish deployed improvement. + +## Prospective H5d oracle instrumentation and verified-completion gate + +H5d adds oracle instrumentation and a prospective completion gate from frozen H5c +`e431ec63d41fe90c6fd68bd2fc2bb969feeca917`. It keeps the exact H5/H4e mode, +protocol, registration, workload and deadline constants. The new driver commit +and changed nine-file measurement digest distinguish the experiment. No product +optimization, new buffering, setup/control flow, payload bound, provider fixture, +transport, public API, P0 ordering or raw-ACK durability policy changes here. +All fifty main arms, including fresh B0/B1 A/A, must be rerun on the qualified new +source; old noise or the failed H5c attempt cannot qualify H5d. + +The retained H5c FULL B0 attempt reached the unchanged 1,200,000-ms observation +ceiling while fetching/validating independent durable events. Root's retained +raw evidence records all 100 runs succeeded within 75.926 seconds of the same +180-second settlement window, all 100 clock uploads ended with 51,200 emissions +and no clock errors, and only roughly 30 of 100 per-run durable reads completed +before timeout. Foreground completion remained 283.391 seconds, with only +66/512 markers at the original streaming deadline. This owning attempt failed; +complete in-journey transcript/results verification and later cases were not +finished. A separate stopped read-only supplement cannot turn it into an arm +pass. Product improvement is not established by that failure or this correction. + +Frozen H5c performed five Playwright assertions for every measured normalized +delta and one Playwright deep-equality assertion for its original clock join. +It also asserted each prelude payload and each visible clock through Playwright. +H5d uses the same actual paginated API rows and provider clock maps, then validates +those retained observations through Node assertions. Only verified per-run +summaries enter the existing durable evidence array. The active failure record +also identifies the current run, pagination cursor and returned row count. + +| Frozen check | H5d equivalent | +| --- | --- | +| Exact complete compact payload for every measured delta | `assertDurableRunEvents` checks the same full text, rejecting coalescing or added bytes | +| Exact run, truthy source-instance/event IDs, positive safe integer sequence in H5 | Same assertions under the unchanged `requireNativeIdentity: h5` condition | +| Original nonce, marker code and index for every durable event | Same tuple deep equality against the original provider clock map | +| Unique normalized identity tuple and complete ordered indices | Same set count and exact ordered index array; Stop permits only its canonical prefix | +| Every visible Stop marker is durable; normal result one / Stop result zero | Same prefix membership and exact accepted-result count | +| Complete normal source-clock count and closed uploads | Existing clock-readiness/closure polls remain; Node oracle also checks the normal count | +| Prelude exact dots/final readiness text, count and unique identity tuples | `assertPreludeEvents` checks the same filtered prelude rows before epoch release | +| Every visible marker has an original clock; complete normal unique visible count | `assertVisibleMarkerClocks` checks the same observations and preserves the original two-frame timestamp | + +The same `/events?afterSeq=…&limit=1000` pages, advancing cursor checks and final +empty-page read remain. The 30-second per-run provider-clock availability poll +is unchanged and still occurs before the Node assertion. The 120/180/900-second +original gates, 600/180/1,200-second observation ceilings, original late Stop ACK +and lease polls, and outer resource watchdog remain unchanged. No per-event +check is dropped to make the invocation finish. + +The frozen comparator9e01 validated `h5.independentDrain` correctness but omitted +its elapsed duration from quantitative gates. A synthetic RED counterexample +actually passed that comparator with faster browser/terminal delivery while full +verified completion regressed from180,010 to200,000 ms. The prospective comparator +adds `h5.independentDrain.elapsedFromEpochMs` to every non-Stop H5 case's existing +measurement, own A/A noise and nonregression loops. For loaded burst512, final B2 +must also improve that duration beyond its own noise in the median and at least +four of five pairs, alongside the existing browser p95 and terminal-duration +gates. Three improving pairs do not pass. Cursor nonregression includes the same +verified-completion metric. All previous gates and timing limits remain. + +Reports also derive measured normalized deltas per second from equal matched +counts divided by that independently verified completion upper bound. For the +loaded burst, N is51,200 complete measured deltas, with all100 accepted results +independently checked. The duration includes oracle, reload, screenshot, normal +API and lease verification overhead; it is not the exact last database-commit +instant. This conservative observed completion rate is descriptive: the existing +noise gate applies to duration, and no separate rate-noise threshold is invented. +Equal counts make a shorter fully verified duration monotonically higher rate. +Missing/tampered completion proof still fails the existing functional oracle. +No accepted A/A exists. The following finite browser repeatability rule is +chosen prospectively; source freezing still requires the combined root/peer review. + +### Prospective main-browser A/A suitability (H5 only) + +Before any B0 or B1 calibration data, choose exactly five alternating pairs and +apply these rules independently to every declared case/metric after all existing +functional, owning-report and cleanup checks. With pair durations a,b: + +- Every paired difference must satisfy `abs(a-b) <= max(0.20*a, 2 ms)`. +- Sort the ten actual owning invocations by validated start. First-five versus + last-five median drift must satisfy `abs(lastMedian-firstMedian) <= + max(0.10*firstMedian, 2 ms)`. +- The two five-sample role medians must also satisfy `abs(candidateMedian- + baselineMedian) <= max(0.10*baselineMedian, 2 ms)`. + +Require parseable owner start/duration, nonoverlap, registered baseline-first then +candidate-first alternation, distinct owning reports, companies and run sets. +A comparison must follow all calibration owners and reuse none of their evidence +or owned data. Synthetic oracle fixtures represent this chronology and freshness +explicitly; they do not establish that an application or provider ran. + +Retain all raw samples and owner chronology. Freeze the actual maximum paired +absolute difference AND actual maximum paired relative difference. The 2-ms +floor applies only to suitability and never replaces either observed envelope. +Comparisons recompute these envelopes and suitability diagnostics from retained +samples and reject altered envelopes, rules, flags or timestamps. Unsuitable A/A +exits unsuccessfully after writing its complete diagnostic report; retain all ten +arms and diagnose before a newly reviewed prospective observation. Do not inflate +caps or silently replace an arm. + +If the pooled ten-value median is at most20 ms, label that metric +representation-limited, suppress percentage-improvement claims and use its actual +frozen absolute noise for nonregression: `median(candidate-baseline) <= observed +absolute noise`. In particular, zero observed noise permits no positive regression +even when the suitability floor was2 ms. Longer metrics retain the existing +relative-noise nonregression rule. Loaded improvement still requires its existing +relative-noise median and at least4/5 individual wins; each counted win and the +median absolute improvement must additionally exceed2 ms before a numerical gain +is claimed. No sub-2ms, p99 or percentage gain is claimed for representation-limited +metrics. All original hard product/control deadlines remain unchanged. + +The floor describes integer-millisecond timestamp contrast representation only. +It is not a frame period, paint guarantee, clock-accuracy bound or a bound on the +settlement/Stop observer. The unchanged Playwright polls can wait1000 ms after a +callback and incur API/scheduler time as well; no blanket1–2second tolerance is +introduced. The blocked-Stop companion keeps its separate frozen100-ms policy. +The default H4e comparator and journey semantics remain unchanged. + +### Deferred diagnostic design: Playwright assertion cost only + +This diagnostic has **not run**. After root review and an explicit quiet-window +release, create an owned temporary standalone Playwright test/config outside the +measured source. Do not inherit a config with `webServer`. Request no `page`, +`browser` or `request` fixture; use one Node worker, no retries, trace/video off, +and a bounded 300-second diagnostic timeout. It must start no app, database, +provider or browser. Record the temporary source/config, Node and Playwright +versions, source hashes, owning test status, raw durations and all failures. + +Generate 1,000 deterministic synthetic normalized rows and a matching clock map +inside that test process. They are CPU-oracle inputs only: no provider output, +application event or persisted outcome is seeded. Use the same six checks for +both branches on the same objects: + +```ts +// Planned scratch test functions, not an executed measurement. +const checkPw = (event, original, index) => { + expect(event.payload.text).toBe(`[q7${index.toString(36)}] `); + expect(event.runId).toBe("diagnostic-run"); + expect(event.sourceInstanceId).toBeTruthy(); + expect(event.sourceEventId).toBeTruthy(); + expect(Number.isSafeInteger(event.sourceSeq) && event.sourceSeq > 0).toBe(true); + expect({ nonce: original.nonce, markerCode: original.markerCode, index: original.index }) + .toEqual({ nonce: "diagnostic-source", markerCode: "q7", index }); +}; +const checkNode = (event, original, index) => { + assert.equal(event.payload.text, `[q7${index.toString(36)}] `); + assert.equal(event.runId, "diagnostic-run"); + assert.ok(event.sourceInstanceId); + assert.ok(event.sourceEventId); + assert.ok(Number.isSafeInteger(event.sourceSeq) && event.sourceSeq > 0); + assert.deepEqual({ nonce: original.nonce, markerCode: original.markerCode, index: original.index }, + { nonce: "diagnostic-source", markerCode: "q7", index }); +}; +``` + +Warm each branch once on 100 inputs outside timing. Then record five alternating +pairs of all 1,000 inputs using monotonic `performance.now()` boundaries, retaining +each elapsed time and six-check count. Do not disable normal Playwright assertion +steps in its branch. Compare total completion duration and spread; keep diagnostic +profiles separate. This tests assertion instrumentation overhead, not application +ingestion throughput, browser responsiveness or acceptance. Full H5d qualification +and the fresh frozen fifty-arm protocol remain required regardless of its result. diff --git a/tests/e2e/fixtures/native-ingestion-h5.mjs b/tests/e2e/fixtures/native-ingestion-h5.mjs index 3ba362667bc..b2efeda1713 100644 --- a/tests/e2e/fixtures/native-ingestion-h5.mjs +++ b/tests/e2e/fixtures/native-ingestion-h5.mjs @@ -334,6 +334,79 @@ export function assertRecordedFailures(proof, item, companyId) { recordedErrorClasses: Object.fromEntries([...classes].sort(([a], [b]) => a.localeCompare(b)).map(([code, ids]) => [code, ids.size])) }; } +// H5d only changes post-observation oracle instrumentation. These functions +// check the same retained API rows and original provider clocks without making +// a Playwright step for every normalized delta. They never drive the SUT. +export function collectDurableRunEvents(rows, markerCode) { + const deltas = []; let resultAcceptedEvents = 0; + for (const row of rows) { + const event = row.payload?.prpEvent; + if (row.eventType === 'run.result.accepted' || event?.eventType === 'run.result.accepted') resultAcceptedEvents++; + if (event?.eventType !== 'item.delta') continue; + for (const match of String(event.payload?.text ?? '').matchAll(new RegExp(`\\[${markerCode}([a-z0-9]{1,3})\\]`, 'g'))) { + const index = Number.parseInt(match[1], 36); + deltas.push({ index, identity: `${event.sourceInstanceId}:${event.sourceEventId}:${event.sourceSeq}`, event }); + } + } + return { deltas, resultAcceptedEvents }; +} + +export function assertDurableRunEvents({ collected, runId, token, markerCode, events, stop, sourceClocks, visible, + requireNativeIdentity }) { + const { deltas, resultAcceptedEvents } = collected; + for (const [ordinal,delta] of deltas.entries()) { + const { index, event } = delta, context = `Durable run ${runId} source ${token} delta ${ordinal} marker ${markerCode}:${index}`; + assert.equal(event.payload.text, `[${markerCode}${index.toString(36)}] `, `${context}: complete normalized payload`); + if (requireNativeIdentity) { + assert.equal(event.runId, runId, `${context}: exact run identity`); + assert.ok(event.sourceInstanceId, `${context}: nonempty source instance identity`); + assert.ok(event.sourceEventId, `${context}: nonempty source event identity`); + assert.ok(Number.isSafeInteger(event.sourceSeq) && event.sourceSeq > 0, `${context}: positive integer source sequence`); + } + const original = sourceClocks.get(index); + assert.ok(original, `${context}: missing original provider clock`); + assert.deepEqual({ nonce: original.nonce, markerCode: original.markerCode, index: original.index }, + { nonce: token, markerCode, index }, `${context}: exact original emission clock join`); + } + const identities = new Set(deltas.map(delta => delta.identity)), indices = deltas.map(delta => delta.index); + assert.equal(identities.size, deltas.length, `Durable run ${runId}: duplicate normalized source identity tuple`); + assert.deepEqual(indices, Array.from({ length: stop ? deltas.length : events }, (_, index) => index), + `Durable run ${runId}: complete ordered ${stop ? 'canonical Stop prefix' : 'measured events'}`); + if (stop) { + const present = new Set(indices); + for (const item of visible) assert.ok(present.has(item.index), + `Durable run ${runId}: visible Stop marker ${item.index} must remain durable`); + } else assert.equal(sourceClocks.size, events, `Durable run ${runId}: complete source clock count`); + assert.equal(resultAcceptedEvents, stop ? 0 : 1, `Durable run ${runId}: exact accepted result count`); + return { runId, token, markerCode, individualDeltas: deltas.length, uniqueSourceIdentities: identities.size, + matchedSourceClocks: deltas.length, ordered: true, resultAcceptedEvents }; +} + +export function assertPreludeEvents({ rows, runId, token, expectedEvents }) { + const identities = new Set(); let count = 0; + for (const row of rows) { + const event = row.payload?.prpEvent; + if (event?.eventType !== 'item.delta' || !String(event.itemId ?? '').startsWith('prelude-')) continue; + assert.equal(event.payload?.text, count === expectedEvents - 1 ? ` PRELUDE_${token}_READY ` : '.', + `Prelude run ${runId} delta ${count}: complete normalized payload`); + count++; identities.add(`${event.sourceInstanceId}:${event.sourceEventId}:${event.sourceSeq}`); + } + assert.equal(count, expectedEvents, `Prelude run ${runId}: exact individually persisted delta count`); + assert.equal(identities.size, count, `Prelude run ${runId}: unique source identity tuples`); + return { individualDeltas: count, uniqueSourceIdentities: identities.size }; +} + +export function assertVisibleMarkerClocks({ seenMarkers, sourceClocks, markerCode, expectedEvents, stop }) { + const visible = seenMarkers.map(marker => { + const original = sourceClocks.get(marker.index); + assert.ok(original, `Missing original provider clock for visible ${marker.markerCode}:${marker.index}`); + return { ...original, ...marker }; // Retain the original two-frame paint timestamp. + }); + if (!stop) assert.equal(new Set(visible.map(item => item.index)).size, expectedEvents, + `Visible marker ${markerCode}: complete unique foreground count`); + return visible; +} + export function assessH5Arm(report, owner) { assert.equal(report.protocolVersion, H5_PROTOCOL); assert.equal(report.observationMode, H5_MODE); diff --git a/tests/e2e/native-event-ingestion.journey.ts b/tests/e2e/native-event-ingestion.journey.ts index bcb49eedda3..d30e5efad1d 100644 --- a/tests/e2e/native-event-ingestion.journey.ts +++ b/tests/e2e/native-event-ingestion.journey.ts @@ -8,7 +8,8 @@ import path from "node:path"; import { expect, test } from "@playwright/test"; import { json } from "./agent-chat.shared"; import { h5Enabled, H5_MODE, H5_PROTOCOL, H4E_PROTOCOL, H5_TOPOLOGY, H5_LIMITS, H5_WORKLOAD_SHA256, ORIGINAL_BUDGETS, - observeStreaming, observeRunSettlement, originalTerminal, prepareParticipantAgents, assertParticipantTopology, assertRecordedFailures } from "./fixtures/native-ingestion-h5.mjs"; + observeStreaming, observeRunSettlement, originalTerminal, prepareParticipantAgents, assertParticipantTopology, assertRecordedFailures, collectDurableRunEvents, assertDurableRunEvents, + assertPreludeEvents, assertVisibleMarkerClocks } from "./fixtures/native-ingestion-h5.mjs"; type SeenMarker = { markerCode: string; index: number; visibleMs: number }; type Emission = { nonce: string; markerCode: string; index: number; intendedMs: number; emittedMs: number; receivedAtMs: number }; @@ -511,23 +512,18 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy 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; + const preludeRows: any[] = []; let 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}`); - } + preludeRows.push(...rows); 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, + const checked = assertPreludeEvents({ rows: preludeRows, runId: foreground.id, token: nonce, expectedEvents: workload.prelude }); + preludeProof = { runId: foreground.id, ...checked, 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 controllerAuthority = await runtimeAuthority({ appRoot, companyId: company.id, environmentId: caseEnvironment.id, issueIds, transport: caseTransport, @@ -645,12 +641,8 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy ? 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 visible: Visible[] = assertVisibleMarkerClocks({ seenMarkers, sourceClocks: state.streams.get(nonce)!.records, + markerCode, expectedEvents: workload.events, stop }); // Terminal drainage has finished. Derive paint gaps from the retained // observer timestamps without adding work to the streaming observer. const paintTimes = [...new Set(visible.map(item => item.visibleMs))].sort((a, b) => a - b); @@ -693,39 +685,23 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy 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, resultAcceptedEvents = 0; + const persistedRows: any[] = []; + let afterSeq = 0; + active.verifyingRun = { runId: run.id, nonce: token, markerCode: code, afterSeq, rows: 0 }; while (true) { const rows = await json(await request.get(`/api/heartbeat-runs/${run.id}/events?afterSeq=${afterSeq}&limit=1000`)); if (!rows.length) break; - for (const row of rows) { - const event = row.payload?.prpEvent; - if (row.eventType === "run.result.accepted" || event?.eventType === "run.result.accepted") resultAcceptedEvents++; - if (event?.eventType !== "item.delta") continue; - 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)}] `); - if (h5) { expect(event.runId).toBe(run.id); expect(event.sourceInstanceId).toBeTruthy(); - expect(event.sourceEventId).toBeTruthy(); expect(Number.isSafeInteger(event.sourceSeq) && event.sourceSeq > 0).toBe(true); } - deltas.push({ index, identity: `${event.sourceInstanceId}:${event.sourceEventId}:${event.sourceSeq}` }); - } - } + persistedRows.push(...rows); const next = Math.max(...rows.map((row: { seq: number }) => row.seq)); expect(next).toBeGreaterThan(afterSeq); afterSeq = next; + active.verifyingRun = { runId: run.id, nonce: token, markerCode: code, afterSeq, rows: persistedRows.length }; } - 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); - expect(resultAcceptedEvents).toBe(stop ? 0 : 1); - durable.push({ runId: run.id, token, markerCode: code, individualDeltas: deltas.length, uniqueSourceIdentities: deltas.length, - matchedSourceClocks: deltas.length, ordered: true, resultAcceptedEvents }); + const collected = collectDurableRunEvents(persistedRows, code); + active.latestDurableRead = { runId: run.id, nonce: token, markerCode: code, + deltas: collected.deltas.map(({ index, identity }) => ({ index, identity })) }; + await expect.poll(() => collected.deltas.every(delta => state.streams.get(token)!.records.has(delta.index)), { timeout: 30_000 }).toBe(true); + durable.push(assertDurableRunEvents({ collected, runId: run.id, token, markerCode: code, events: workload.events, stop, + sourceClocks: state.streams.get(token)!.records, visible, requireNativeIdentity: h5 })); } await expect.poll(() => [...state.streams.values()].every(stream => stream.ended || stream.aborted), { timeout: 30_000 }).toBe(true); expect(state.errors).toEqual([]);