From e431ec63d41fe90c6fd68bd2fc2bb969feeca917 Mon Sep 17 00:00:00 2001 From: Dotta Date: Wed, 4 Nov 2026 00:29:37 -0600 Subject: [PATCH] test: wait for native run settlement and retain recovered failures --- scripts/compare-native-ingestion-browser.ts | 95 +++++++-- scripts/describe-native-ingestion-h5.mjs | 2 +- scripts/native-ingestion-h5.test.mjs | 107 +++++++++- ...ative-ingestion-performance-gates.test.mjs | 148 +++++++++++--- tests/e2e/fixtures/native-ingestion-h5.md | 116 +++++++++-- tests/e2e/fixtures/native-ingestion-h5.mjs | 188 +++++++++++++++++- tests/e2e/native-event-ingestion.journey.ts | 70 ++++--- 7 files changed, 621 insertions(+), 105 deletions(-) diff --git a/scripts/compare-native-ingestion-browser.ts b/scripts/compare-native-ingestion-browser.ts index d593deb41db..53fbb6ff03d 100644 --- a/scripts/compare-native-ingestion-browser.ts +++ b/scripts/compare-native-ingestion-browser.ts @@ -3,8 +3,8 @@ import { createHash } from "node:crypto"; import { readFile, writeFile } from "node:fs/promises"; import path from "node:path"; import { noiseExceedingPairedWins } from "./native-ingestion-performance-gates.mjs"; -import { H5_MODE, H5_PROTOCOL, H4D_PROTOCOL, H5_WORKLOAD_SHA256, H5_BASELINE_REVISIONS, ORIGINAL_BUDGETS, - assessH5Arm, requireH5Candidate, assertParticipantSetup, assertParticipantTopology } from "../tests/e2e/fixtures/native-ingestion-h5.mjs"; +import { H5_MODE, H5_PROTOCOL, H4E_PROTOCOL, H5_WORKLOAD_SHA256, H5_BASELINE_REVISIONS, ORIGINAL_BUDGETS, + assessH5Arm, requireH5Candidate, assertParticipantSetup, assertParticipantTopology, assertSettlementObservation, assertRecordedFailures } from "../tests/e2e/fixtures/native-ingestion-h5.mjs"; import { validateArmInventory } from "./native-ingestion-h5-inventory.mjs"; // Each manifest entry points at native-ingestion-evidence JSON attachments @@ -18,6 +18,7 @@ const h5 = arg("observation-mode") === H5_MODE; assert.ok(arg("observation-mode") === undefined || h5, "Unknown observation mode"); const candidatePolicy = arg("candidate-policy"); assert.ok(!arg("original-h4c-proof") && !arg("original-h4c-inventory"), "Old H4c proof arguments cannot confirm the new independent-agent geometry"); +assert.ok(!arg("original-h4d-proof") && !arg("original-h4d-inventory"), "Old H4d proof arguments use the premature settlement observer"); assert.ok(["aa", "compare"].includes(phase)); assert.ok(["improvement", "nonregression"].includes(expectation)); const manifest = JSON.parse(await readFile(manifestPath, "utf8")) as Array<{ baselineFile: string; candidateFile: string; @@ -154,6 +155,8 @@ function assertCaseAuthority(report: any, result: any) { } function assertCaseSettlement(report: any, result: any, validated = assertCaseAuthority(report, result)) { + assertSettlementObservation(result.h5?.terminalObservation ?? result.terminalObservation, result); + assertRecordedFailures(result.recordedFailures, result, report.companyId); const remote = report.remote, local = report.local; const { authority, authorityByRun } = validated; if (result.stop) { @@ -229,7 +232,7 @@ function assertCaseSettlement(report: any, result: any, validated = assertCaseAu for (const pair of pairs) for (const arm of ["baseline", "candidate"] as const) { const report = pair[arm]; - assert.equal(report.protocolVersion, h5 ? H5_PROTOCOL : H4D_PROTOCOL, + assert.equal(report.protocolVersion, h5 ? H5_PROTOCOL : H4E_PROTOCOL, "Fresh browser acceptance requires the independent-agent observation or default-deadline protocol"); assert.equal(report.transport, "sandbox-remote-with-local-stop", "V4 requires remote streaming and the original local Stop transport"); assert.deepEqual(report.transportPolicy, { streaming: "sandbox-remote", operatorStop: "local", @@ -401,25 +404,25 @@ for (const pair of pairs) for (const arm of ["baseline", "candidate"] as const) for (const field of ["runs", "events", "intervalMs", "prelude", "stop"]) assert.equal(result[field], reference[field], `Workload changed: ${name}.${field}`); } } -let originalH4dConfirmation: Record | undefined; +let originalH4eConfirmation: Record | undefined; if (h5 && phase === "compare" && candidatePolicy === "final-b2") { - originalH4dConfirmation = { required: true, passed: false, reason: "Later H4d default-mode full functional proof on the same source and geometry is pending" }; - if (arg("original-h4d-proof")) { + originalH4eConfirmation = { required: true, passed: false, reason: "Later H4e default-mode full functional proof on the same source and geometry is pending" }; + if (arg("original-h4e-proof")) { try { - assert.ok(arg("original-h4d-inventory"), "H4d original-deadline proof requires its full artifact inventory sidecar"); - const { evidence: proof, owner } = await owningResult(path.resolve(arg("original-h4d-proof")!)); + assert.ok(arg("original-h4e-inventory"), "H4e original-deadline proof requires its full artifact inventory sidecar"); + const { evidence: proof, owner } = await owningResult(path.resolve(arg("original-h4e-proof")!)); assert.equal(owner.status, "passed"); assert.equal(owner.retry, 0); assert.deepEqual(owner.errors, []); assert.equal(owner.timeoutMs, ORIGINAL_BUDGETS.fullInvocationMs); assert.ok(owner.durationMs > 0 && owner.durationMs < ORIGINAL_BUDGETS.fullInvocationMs); - assert.equal(proof.protocolVersion, H4D_PROTOCOL, "Old H4c geometry cannot confirm H5b"); + assert.equal(proof.protocolVersion, H4E_PROTOCOL, "Old H4c geometry cannot confirm H5c"); assertParticipantSetup(proof); assert.deepEqual(proof.agentTopology, first.candidate.agentTopology); - assert.equal(proof.observationMode, undefined, "Confirmation must use H4d default deadlines"); - assert.equal(proof.harnessProvenance?.sourceRevision, first.candidate.harnessProvenance.sourceRevision, "H4d confirmation must use the same frozen harness source"); + assert.equal(proof.observationMode, undefined, "Confirmation must use H4e default deadlines"); + assert.equal(proof.harnessProvenance?.sourceRevision, first.candidate.harnessProvenance.sourceRevision, "H4e confirmation must use the same frozen harness source"); assert.deepEqual(proof.harnessProvenance.sourceHashes, first.candidate.harnessProvenance.sourceHashes); assert.equal(proof.harnessProvenance.driverLockSha256, first.candidate.harnessProvenance.driverLockSha256); assert.equal(proof.harnessProvenance.sourceStatus, ""); - assert.equal(proof.measurementSha256, first.candidate.measurementSha256, "H4d confirmation must bind the same measured harness bytes"); + assert.equal(proof.measurementSha256, first.candidate.measurementSha256, "H4e confirmation must bind the same measured harness bytes"); assert.deepEqual(proof.resourceBudgets, ORIGINAL_BUDGETS); assert.equal(proof.complete, true); assert.equal(proof.qualificationOnly, false); assert.equal(proof.diagnosticOnly, false); assert.equal(proof.performanceAcceptanceEligible, true); assert.equal(proof.serverArtifact, "compiled"); @@ -447,12 +450,15 @@ if (h5 && phase === "compare" && candidatePolicy === "final-b2") { const latestCandidateEnd = Math.max(...pairs.map(pair => Date.parse(pair.candidate.owningPlaywright.startedAt) + pair.candidate.owningPlaywright.durationMs)); assert.ok(Date.parse(owner.startedAt) > latestCandidateEnd, "Original confirmation must follow all candidate arms"); assert.ok(pairs.every(pair => proof.companyId !== pair.baseline.companyId && proof.companyId !== pair.candidate.companyId), "Original confirmation needs a fresh disposable company/database"); - const inventory = await validateArmInventory(path.resolve(arg("original-h4d-inventory")!), { evidence: proof, owningPlaywright: owner }); + const inventory = await validateArmInventory(path.resolve(arg("original-h4e-inventory")!), { evidence: proof, owningPlaywright: owner }); for (const field of ["manifestFileSha256", "aggregateSha256", "qualifiedReceiptSha256", "inventorySummarySha256", "revision"] as const) assert.equal(inventory[field], first.candidate.armInventory[field], "Original confirmation used a different full qualified artifact inventory"); - originalH4dConfirmation = { required: true, passed: true, owner, inventory, sourceRevision: proof.sourceRevision }; + originalH4eConfirmation = { required: true, passed: true, owner, inventory, sourceRevision: proof.sourceRevision, + retryObservations: proof.cases.map((item: any) => ({ name: item.name, + ...assertRecordedFailures(item.recordedFailures, item, proof.companyId), + transitions: item.terminalObservation.transitions, recordedFailures: item.recordedFailures })) }; } catch (error) { - originalH4dConfirmation.reason = error instanceof Error ? error.message : String(error); + originalH4eConfirmation.reason = error instanceof Error ? error.message : String(error); } } } @@ -478,6 +484,34 @@ const measurements = names.flatMap(name => { samples }; }); }); +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); + const value = item.h5?.terminalObservation ?? item.terminalObservation; + const recorded = assertRecordedFailures(item.recordedFailures, item, pair[arm].companyId); + return [arm, { runs: item.runs, ...recorded, + retryRate: recorded.uniqueEverFailedRuns / item.runs, + errorClassRates: Object.fromEntries(Object.entries(recorded.recordedErrorClasses).map(([code, count]) => [code, Number(count) / item.runs])), + transitionCount: value.transitions.length, transitions: value.transitions, recordedFailures: item.recordedFailures }]; + }))) as Array>; + const observedErrorClasses = [...new Set(samples.flatMap(sample => ["baseline", "candidate"] + .flatMap(arm => Object.keys(sample[arm].recordedErrorClasses))))].sort(); + const rates = [null, ...observedErrorClasses].map(errorClass => { + const values = samples.map(sample => { + const baselineCount = errorClass === null ? sample.baseline.uniqueEverFailedRuns : sample.baseline.recordedErrorClasses[errorClass] ?? 0; + const candidateCount = errorClass === null ? sample.candidate.uniqueEverFailedRuns : sample.candidate.recordedErrorClasses[errorClass] ?? 0; + assert.equal(sample.baseline.runs, sample.candidate.runs); + const runs = sample.baseline.runs; + return { runs, baselineCount, candidateCount, baseline: baselineCount / runs, candidate: candidateCount / runs, + increaseCount: candidateCount - baselineCount, increase: (candidateCount - baselineCount) / runs }; + }); + return { errorClass, samples: values, medianIncrease: median(values.map(value => value.increase)), + medianIncreaseCount: median(values.map(value => value.increaseCount)), + aaNoiseEnvelopeRuns: phase === "aa" ? Math.max(...values.map(value => Math.abs(value.increaseCount))) : undefined, + aaNoiseEnvelopeAbsolute: phase === "aa" ? Math.max(...values.map(value => Math.abs(value.increase))) : undefined }; + }); + return { name, pairCount: samples.length, observedErrorClasses, samples, rates }; +}); const result: Record = { scope: "Compiled-app streaming with100 independent normal agents and the original local Stop workload; real cloud Stop is separately required", agentTopology: first.baseline.agentTopology, @@ -490,11 +524,11 @@ const result: Record = { gatesSha256: createHash("sha256").update(await readFile(new URL("./native-ingestion-performance-gates.mjs", import.meta.url))).digest("hex"), 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, + primaryLoadedWorkload: "burst512", complete: true, retryObservations, ...(h5 ? { observationMode: H5_MODE, workloadSha256: H5_WORKLOAD_SHA256, candidatePolicy, h5Assessments, - artifactInventories: { baseline: first.baseline.armInventory, candidate: first.candidate.armInventory }, originalH4dConfirmation, - originalProtocolAcceptance: "Baseline SLA misses are retained; H5b complete observation is not an original-deadline PASS", - finalH4dFunctionalPassRequired: candidatePolicy === "final-b2" } : {}), + 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", + finalH4eFunctionalPassRequired: candidatePolicy === "final-b2" } : {}), }; if (phase === "compare") { assert.ok(arg("noise-report"), "Comparison requires its frozen A/A --noise-report"); @@ -529,14 +563,33 @@ if (phase === "compare") { noiseExceedingPairedWins: paired.noiseExceedingPairedWins, requiredPairedWins: paired.requiredPairedWins, passed: improvementRequired ? improvement : noRegression }; }); + assert.equal(noise.retryObservations?.length, names.length, "A/A must retain complete retry-rate calibration"); + result.retryGates = retryObservations.flatMap(item => { + const control = noise.retryObservations.find((entry: any) => entry.name === item.name); + assert.equal(control?.pairCount, pairs.length, "A/A retry calibration must retain every pair"); + return item.rates.map(rate => { + const knownClass = rate.errorClass === null || control.observedErrorClasses.includes(rate.errorClass); + const reference = control.rates.find((entry: any) => entry.errorClass === rate.errorClass); + const envelope = reference?.aaNoiseEnvelopeAbsolute ?? 0; + assert.ok(Number.isFinite(envelope) && envelope >= 0 && envelope <= 1, "Invalid frozen absolute retry-rate noise"); + const envelopeRuns = reference?.aaNoiseEnvelopeRuns ?? 0; + assert.ok(Number.isSafeInteger(envelopeRuns) && envelopeRuns >= 0 + && envelope === envelopeRuns / rate.samples[0].runs, "Retry-rate noise must match its exact run counts"); + const pairedPassed = rate.samples.every(sample => sample.increaseCount <= envelopeRuns); + return { name: item.name, errorClass: rate.errorClass, knownClass, noiseEnvelopeAbsolute: envelope, + medianIncrease: rate.medianIncrease, pairedIncreases: rate.samples.map(sample => sample.increase), + passed: knownClass && pairedPassed && rate.medianIncreaseCount <= envelopeRuns }; + }); + }); result.performancePassed = result.gates.every((entry: any) => entry.passed) + && result.retryGates.every((entry: any) => entry.passed) && (!h5 || h5Assessments.every(entry => entry.candidateSlaPolicyPassed)); - result.acceptancePassed = result.performancePassed && (!originalH4dConfirmation || originalH4dConfirmation.passed); + result.acceptancePassed = result.performancePassed && (!originalH4eConfirmation || originalH4eConfirmation.passed); } if (arg("output")) await writeFile(arg("output")!, JSON.stringify(result, null, 2)); console.log(JSON.stringify(result, null, 2)); if (process.argv.includes("--require-performance")) { - assert.equal(h5, true, "H5b performance-only gating is explicit; default H4d keeps original deadlines"); + assert.equal(h5, true, "H5c performance-only gating is explicit; default H4e keeps original deadlines"); assert.equal(phase, "compare"); assert.equal(result.performancePassed, true, "H5 performance/SLA gates failed; retain results and diagnose"); } diff --git a/scripts/describe-native-ingestion-h5.mjs b/scripts/describe-native-ingestion-h5.mjs index b93b7f2a3e0..205b6aa0faf 100644 --- a/scripts/describe-native-ingestion-h5.mjs +++ b/scripts/describe-native-ingestion-h5.mjs @@ -4,4 +4,4 @@ console.log(JSON.stringify({ dryRun: true, executionAuthorized: false, workloadS registration: H5_REGISTRATION, requiredBeforeRelease: ['Root-reviewed exact driver/comparator hashes', 'New functional observation qualification', 'Fresh B0 and B1 A/A before further product implementation', 'Root-reviewed frozen A/A report hashes before candidate stages', 'Final B2 every-arm original SLA pass', - 'Later H4d default-mode full functional pass on the same source, geometry and final SUT'] }, null, 2)); + 'Later H4e default-mode full functional pass on the same source, geometry and final SUT'] }, null, 2)); diff --git a/scripts/native-ingestion-h5.test.mjs b/scripts/native-ingestion-h5.test.mjs index 9e074d97cd1..0c04e043eca 100644 --- a/scripts/native-ingestion-h5.test.mjs +++ b/scripts/native-ingestion-h5.test.mjs @@ -1,12 +1,27 @@ import assert from 'node:assert/strict'; 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 } from '../tests/e2e/fixtures/native-ingestion-h5.mjs'; + H5_LIMITS, ORIGINAL_BUDGETS, prepareParticipantAgents, observeRunSettlement } 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('real frozen Playwright tiny poll exposes the matcher deadline shape used by H5', async () => { const require = createRequire(import.meta.url), { expect } = require('@playwright/test'); assert.equal(require('@playwright/test/package.json').version, '1.62.1'); @@ -39,10 +54,11 @@ test('real frozen Playwright can report its configured timeout before the nomina assessedAtMs, elapsedMs, observations, matcherName: failure.matcherResult.name, matcherPassed: failure.matcherResult.pass })); }); -test('H5b requires its explicit mode and leaves default H4d selected', () => { +test('H5c requires its explicit mode and leaves default H4e selected', () => { assert.equal(h5Enabled(undefined), false); assert.equal(h5Enabled(H5_MODE), true); assert.throws(() => h5Enabled('almost-h5')); assert.throws(() => h5Enabled('h5-observe-original-sla'), 'Old H5a observations use a different geometry'); + assert.throws(() => h5Enabled('h5b-observe-original-sla-independent-agents'), 'Old H5b observations stop before supported recovery'); }); for (const fail of [false, true]) test(`participant setup uses at most8 workers, preserves index order and settles all attempts${fail ? ' after a failure' : ''}`, async () => { @@ -114,9 +130,96 @@ test('original terminal is not reached after120s miss; observational180s is dist assert.deepEqual(originalTerminal({ original: { status: 'failed' } }, terminal), { status: 'not-reached', reason: 'original_streaming_deadline', startedAtMs: null, deadlineMs: null, assessedAtMs: null }); assert.equal(originalTerminal({ original: { status: 'passed' } }, terminal).status, 'passed'); + assert.equal(originalTerminal({ original: { status: 'passed' } }, { ...terminal, complete: false, + censored: false, error: 'Terminal run failed' }).reason, 'original_terminal_failure'); assert.equal(H5_LIMITS.terminalDrainageMs, ORIGINAL_BUDGETS.terminalDrainageMs); }); +test('terminal observation keeps a supported native retry pending until same-run recovery within the original window', async () => { + let clock = 1000, reads = 0; const calls = [], records = []; + const states = [{ status: 'running', runtimeMode: 'native', nativePhase: 'executing' }, + { status: 'failed', runtimeMode: 'native', nativePhase: 'retryable_failure', errorCode: 'provider_transport_failed' }, + { status: 'succeeded', runtimeMode: 'native', nativePhase: 'committed' }]; + const value = await observeRunSettlement({ expectedRuns: 1, now: () => clock, + onOutcome: value => records.push(structuredClone(value)), + readRuns: async () => [{ id: 'same-run', status: states[Math.min(reads++, 2)].status }], + readRun: async id => { calls.push(id); return { id, ...states[1] }; }, + poll: async (read, milliseconds) => { + assert.equal(milliseconds, 180000); + while (clock < 181000) { if (await read()) return; clock += 30000; } + throw new Error('Unexpected terminal timeout'); + } }); + assert.equal(reads, 3); assert.deepEqual(calls, ['same-run']); + assert.equal(value.runs[0].status, 'succeeded'); assert.equal(value.outcome.complete, true); + assert.equal(value.outcome.startedAtMs, 1000); assert.equal(value.outcome.deadlineMs, 181000); + assert.equal(value.outcome.assessedAtMs, 61000); assert.equal(value.outcome.censored, false); + assert.equal(records.find(value => value.lastStatuses[0]?.nativePhase === 'retryable_failure').terminalRuns, 0); + assert.deepEqual(value.outcome.transitions.map(row => row.status), ['running', 'failed', 'succeeded']); + assert.equal(value.outcome.uniqueRetryableRuns, 1); assert.equal(value.outcome.recoveredRuns, 1); + assert.equal(value.outcome.nonretryableFailures, 0); + assert.deepEqual(value.outcome.retryableErrorClasses, { provider_transport_failed: 1 }); +}); + +test('a permanent native failure rejects immediately without waiting for another provider turn', async () => { + let reads = 0; const records = []; + await assert.rejects(observeRunSettlement({ expectedRuns: 2, + onOutcome: value => records.push(structuredClone(value)), + readRuns: async () => { reads++; return [{ id: 'failed-run', status: 'failed' }, { id: 'still-running', status: 'running' }]; }, + readRun: async id => ({ id, status: 'failed', runtimeMode: 'native', nativePhase: 'terminal_failure' }), + poll: async (read, milliseconds) => { assert.equal(milliseconds, 180000); await read(); } }), /Terminal run failed/); + assert.equal(reads, 1); assert.equal(records.at(-1).complete, false); assert.equal(records.at(-1).censored, false); +}); + +test('an unsettled retry is censored at the one unchanged180s deadline, never a completed latency', async () => { + let clock = 1000, reads = 0; const records = [], calls = []; + const error = Object.assign(new Error('Timeout 180000ms exceeded while waiting on the predicate'), + { matcherResult: { name: 'toBe', pass: false } }); + await assert.rejects(observeRunSettlement({ expectedRuns: 1, now: () => clock, + onOutcome: value => records.push(structuredClone(value)), + readRuns: async () => { reads++; return [{ id: 'same-run', status: 'failed' }]; }, + readRun: async id => ({ id, status: 'failed', runtimeMode: 'native', nativePhase: 'retryable_failure', errorCode: 'provider_transport_failed' }), + poll: async (read, milliseconds) => { + calls.push(milliseconds); + while (clock < 181000) { if (await read()) return; clock += 30000; } + throw error; + } }), value => value === error); + assert.deepEqual(calls, [180000]); assert.equal(reads, 6); + const last = records.at(-1); assert.equal(last.deadlineMs, 181000); assert.equal(last.assessedAtMs, 181000); + assert.equal(last.complete, false); assert.equal(last.censored, true); assert.equal(last.terminalRuns, 0); +}); + +test('failed-list details may already show recovery; detail errors and unknown phases fail closed', async () => { + const value = await observeRunSettlement({ expectedRuns: 1, + readRuns: async () => [{ id: 'run', status: 'failed' }], + readRun: async id => ({ id, status: 'succeeded', runtimeMode: 'native', nativePhase: 'committed' }), + poll: async read => assert.equal(await read(), true) }); + assert.equal(value.runs[0].status, 'succeeded'); assert.equal(value.outcome.uniqueRetryableRuns, 0); + const failure = new Error('normal detail GET failed'); + for (const readRun of [async () => { throw failure; }, + async id => ({ id, status: 'failed', runtimeMode: 'native', nativePhase: 'unknown' })]) { + await assert.rejects(observeRunSettlement({ expectedRuns: 1, + readRuns: async () => [{ id: 'run', status: 'failed' }], readRun, poll: async read => read() }), + error => error === failure || error.message.includes('Terminal run failed')); + } +}); + +test('the real frozen Playwright terminal poll preserves detail errors and exposes its deadline matcher', async () => { + const require = createRequire(import.meta.url), { expect } = require('@playwright/test'); + const failure = new Error('actual callback failure'); + await assert.rejects(expect.poll(async () => { throw failure; }, { timeout: 15 }).toBe(true), error => error === failure); + await assert.rejects(expect.poll(async () => false, { timeout: 15, intervals: [5] }).toBe(true), error => { + assert.equal(error.matcherResult?.name, 'toBe'); assert.equal(error.matcherResult.pass, false); + assert.ok(error.message.includes('Timeout 15ms exceeded while waiting on the predicate')); return true; + }); +}); + +test('bounded state evidence fails on overflow instead of omitting a transition', async () => { + let reads = 0; + await assert.rejects(observeRunSettlement({ expectedRuns: 1, + readRuns: async () => [{ id: 'run', status: 'running', errorCode: String(reads++) }], readRun: async () => assert.fail(), + poll: async read => { for (let index = 0; index < 17; index++) await read(); } }), /evidence overflow/); +}); + test('cursor inherits only paired exact known misses; final candidate inherits none', () => { const base = { misses: ['burst512:streaming120'] }; requireH5Candidate('cursor-nonregression', base, base); diff --git a/scripts/native-ingestion-performance-gates.test.mjs b/scripts/native-ingestion-performance-gates.test.mjs index 4b7ce897e3f..0c0d75ee1ac 100644 --- a/scripts/native-ingestion-performance-gates.test.mjs +++ b/scripts/native-ingestion-performance-gates.test.mjs @@ -1,5 +1,5 @@ import assert from "node:assert/strict"; -import { H5_MODE, H5_PROTOCOL, H4D_PROTOCOL, H5_TOPOLOGY, H5_LIMITS, H5_WORKLOAD_SHA256, H5_BASELINE_REVISIONS, +import { H5_MODE, H5_PROTOCOL, H4E_PROTOCOL, H5_TOPOLOGY, H5_LIMITS, H5_WORKLOAD_SHA256, H5_BASELINE_REVISIONS, originalTerminal, assessH5Arm, requireH5Candidate } from "../tests/e2e/fixtures/native-ingestion-h5.mjs"; import { spawnSync } from "node:child_process"; import { createHash } from "node:crypto"; @@ -77,6 +77,15 @@ const workloads = [ { name: "existing-history", runs: 10, events: 128, intervalMs: 0, prelude: 1000 }, { name: "operator-stop", runs: 1, events: 512, intervalMs: 100, prelude: 0, stop: true }, ]; +function syntheticSettlement(item) { + const states = item.runIds.map(runId => ({ runId, status: item.stop ? "cancelled" : "succeeded", + runtimeMode: null, nativePhase: null, errorCode: null })); + return { startedAtMs: item.settledAtMs, deadlineMs: item.settledAtMs + 180000, + assessedAtMs: item.settledAtMs, lastObservedAtMs: item.settledAtMs, observedRuns: item.runs, terminalRuns: item.runs, + pendingRetryRuns: 0, lastStatuses: states, transitions: states.map(row => ({ observedAtMs: item.settledAtMs, ...row })), + uniqueRetryableRuns: 0, recoveredRuns: 0, nonretryableFailures: 0, retryableErrorClasses: {}, complete: true, censored: false }; +} + function browserReport(revision, latency) { const companyId = "00000000-0000-4000-8000-000000000001", environmentId = "00000000-0000-4000-8000-000000000002"; const localEnvironmentId = "00000000-0000-4000-8000-000000000003", controlAgentId = "00000000-0000-4000-8000-000000000004"; @@ -84,7 +93,7 @@ function browserReport(revision, latency) { const remoteAgentIds = Array.from({ length: 100 }, (_, index) => id(9, 0, index)); const transportPolicy = { streaming: "sandbox-remote", operatorStop: "local", realCloudStop: "Separate required qualification; fake-plugin void receipts grant no remote Stop authority" }; - return { protocolVersion: H4D_PROTOCOL, transport: "sandbox-remote-with-local-stop", transportPolicy, companyId, + return { protocolVersion: H4E_PROTOCOL, transport: "sandbox-remote-with-local-stop", transportPolicy, companyId, bodyStartedAtMs: 1, agentTopology: { ...H5_TOPOLOGY, remoteAgentPoolSize: 100 }, harnessProvenance: { sourceRevision: "b".repeat(40), sourceStatus: "", driverLockSha256: "8".repeat(64), sourceHashes: { "native-event-ingestion.journey.ts": "9".repeat(64), "fixtures/native-ingestion-h5.mjs": "a".repeat(64) } }, @@ -143,7 +152,14 @@ function browserReport(revision, latency) { setupActivationConcurrency: 2, setupReadinessStartedAtMs: 100, readinessDeadlineMs: 300100, allReadyObservedAtMs: 700, participants: workload.runs, activations, activationWaves, epochMs: 1000, preludeProof: workload.prelude ? { individualDeltas: workload.prelude, uniqueSourceIdentities: workload.prelude, verifiedAtMs: 999 } : undefined, + terminalObservation: syntheticSettlement({ ...workload, stop, runIds: rows.map(row => row.run_id), settledAtMs: 1000 + latency }), settledAtMs: 1000 + latency, stopRequestedAtMs: stop ? 1000 : undefined, + recordedFailures: { kind: "recorded-run-failure-codes", transaction: "READ ONLY REPEATABLE READ", + observedReadOnly: "on", observedIsolation: "repeatable read", queryStartedAtMs: 1002 + latency, + checkedAtMs: 1003 + latency, connectionClosedAtMs: 1004 + latency, + rows: rows.map(row => ({ run_id: row.run_id, company_id: companyId, agent_id: row.agent_id, runtime_mode: "native", + status: stop ? "cancelled" : "succeeded", native_phase: "committed", current_error_code: stop ? "cancelled" : null, current_has_error: stop, + recovered_type: null, recovered_schema: null, recovered_error_code: null, recovered_observed_at: null, finalization_attempt: 1 })) }, stopAcknowledgedAtMs: stop ? 1000 + latency : undefined, remoteLeasesCheckedAtMs: stop ? undefined : 1001 + latency, runIds: rows.map(row => row.run_id), controllerAuthority: { ...snapshot, rows, ...(stop ? { localJournal: { schema: "paperclip.runner.durable.control-plane-state.v1", sha256: "8".repeat(64), bytes: 512, checkedAtMs: 999, @@ -165,6 +181,7 @@ function browserReport(revision, latency) { intendedToVisible: { p95Ms: latency }, drainThroughTerminalMs: latency, stopThroughTerminalMs: latency, stopThroughAcknowledgementMs: latency, durable: rows.map((row, index) => ({ runId: row.run_id, token: String(index), ordered: true, + resultAcceptedEvents: stop ? 0 : 1, uniqueSourceIdentities: stop ? 3 : workload.events, individualDeltas: stop ? 3 : workload.events, matchedSourceClocks: stop ? 3 : workload.events })), }; }) }; @@ -185,6 +202,8 @@ function h5Report(revision, loadedLatency) { item.firstVisibleAfterEmissionMs = latency; item.emissionToVisible = { medianMs: latency, p95Ms: latency }; item.intendedToVisible = { p95Ms: latency }; if (!item.stop) item.remoteLeasesCheckedAtMs = item.settledAtMs + 1; + Object.assign(item.recordedFailures, { queryStartedAtMs: item.settledAtMs + 2, + checkedAtMs: item.settledAtMs + 3, connectionClosedAtMs: item.settledAtMs + 4 }); item.visible = Array.from({ length: item.stop ? 3 : item.events }, (_, index) => ({ index, visibleMs: item.settledAtMs })); const missed = latency > 120000, assessedAtMs = missed ? item.epochMs + 120000 : item.settledAtMs; const streaming = { original: { status: missed ? "failed" : "passed", reason: missed ? "original_streaming_deadline" : null, @@ -193,9 +212,7 @@ function h5Report(revision, loadedLatency) { lastObservedAtMs: missed ? assessedAtMs - 1 : assessedAtMs, error: missed ? "Timeout 120000ms exceeded while waiting on the predicate" : null }, observation: { ceilingMs: 600000, deadlineMs: item.epochMs + 600000, complete: true, censored: false, completedAtMs: item.settledAtMs } }; - const terminalObservation = { startedAtMs: item.settledAtMs, deadlineMs: item.settledAtMs + 180000, - assessedAtMs: item.settledAtMs, lastObservedAtMs: item.settledAtMs, observedRuns: item.runs, terminalRuns: item.runs, - lastStatuses: item.runIds.map(runId => ({ runId, status: item.stop ? "cancelled" : "succeeded" })), complete: true, censored: false }; + const terminalObservation = syntheticSettlement(item); item.h5 = { streaming, terminalObservation, terminalOriginal: originalTerminal(streaming, terminalObservation), visibleCountByOriginalPaintDeadline: missed ? 0 : item.visible.length, independentDrain: { verifiedAtMs: item.settledAtMs + 10, elapsedFromEpochMs: latency + 10, @@ -221,7 +238,7 @@ test("H5 CLI retains baseline SLA failure, rejects final SLA misses and requires 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; }; - const run = args => spawnSync(process.execPath, [script.pathname, `--observation-mode=${H5_MODE}`, ...args], { encoding: "utf8" }); + const run = args => spawnSync(process.execPath, [script.pathname, `--observation-mode=${H5_MODE}`, ...args], { encoding: "utf8", maxBuffer: 32 * 1024 * 1024 }); const hash = value => createHash("sha256").update(value).digest("hex"), inventories = new Map(); async function inventory(report, playwrightFile, name) { const variant = ({ [H5_BASELINE_REVISIONS.B0]: "B0", [H5_BASELINE_REVISIONS.B1]: "B1", ["c".repeat(40)]: "B2" })[report.sourceRevision]; @@ -249,7 +266,7 @@ test("H5 CLI retains baseline SLA failure, rejects final SLA misses and requires sourceRevision: report.sourceRevision, compiledBuildRevision: report.compiledBuildRevision, playwrightReportSha256: hash(await readFile(playwrightFile)), ...inventories.get(variant) }); } - async function manifest(prefix, candidateLatency, { sameRevision = false, baselineRevision = "B0", candidateRevision = "B2", mutate, duration = 500000 } = {}) { + async function manifest(prefix, candidateLatency, { sameRevision = false, baselineRevision = "B0", candidateRevision = "B2", mutate, mutateArm, duration = 500000 } = {}) { const entries = []; for (let index = 0; index < 5; index++) { const entry = {}; @@ -258,6 +275,7 @@ test("H5 CLI retains baseline SLA failure, rejects final SLA misses and requires const owner = { timeout: 1200000, expectedStatus: "passed", results: [{ status: "passed", retry: 0, duration, startTime: "2026-11-03T00:00:00.000Z", errors: [], attachments: [] }] }; if (index === 0 && arm === "candidate") mutate?.(report, owner); + mutateArm?.(report, { index, arm }); 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); @@ -270,16 +288,16 @@ test("H5 CLI retains baseline SLA failure, rejects final SLA misses and requires } try { const proof = JSON.parse(JSON.stringify(h5Report("B2", 1000)).replaceAll("00000000-", "10000000-")); - proof.protocolVersion = H4D_PROTOCOL; + proof.protocolVersion = H4E_PROTOCOL; proof.measurementSha256 = "h5-synthetic-fixed-fixture"; - // H4d defaults use the same frozen source/geometry as the observation arms. + // H4e defaults use the same frozen source/geometry as the observation arms. for (const field of ["observationMode", "observationLimits", "observationComplete", "workloadSha256", "actualPlaywrightTimeoutMs"]) delete proof[field]; - for (const item of proof.cases) { delete item.h5; delete item.terminalStatuses; for (const row of item.durable) delete row.resultAcceptedEvents; } - const originalProof = await save("original-h4d-playwright.json", { suites: [{ tests: [{ timeout: 900000, expectedStatus: "passed", results: [{ + for (const item of proof.cases) { item.terminalObservation = item.h5.terminalObservation; delete item.h5; } + const originalProof = await save("original-h4e-playwright.json", { suites: [{ tests: [{ timeout: 900000, expectedStatus: "passed", results: [{ status: "passed", retry: 0, duration: 5000, startTime: "2026-11-04T00:00:00.000Z", errors: [], attachments: [{ name: "native-ingestion-evidence", body: Buffer.from(JSON.stringify(proof)).toString("base64") }] }] }] }] }); const originalInventory = await inventory(proof, originalProof, "original-h4c"); - const finalProofArgs = [`--original-h4d-proof=${originalProof}`, `--original-h4d-inventory=${originalInventory}`]; + 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}`]); assert.equal(aa.status, 0, aa.stderr); @@ -316,37 +334,96 @@ test("H5 CLI retains baseline SLA failure, rejects final SLA misses and requires assert.equal(pendingReport.performancePassed, true); assert.equal(pendingReport.acceptancePassed, false); const performanceOnly = run(["--candidate-policy=final-b2", `--pairs=${goodPairs}`, `--noise-report=${noise}`, "--require-performance"]); assert.equal(performanceOnly.status, 0, performanceOnly.stderr); + const setRetries = (report, count, code = "provider_transport_failed", observed = true) => { + const item = report.cases[3], value = item.h5.terminalObservation; + value.transitions = [...item.runIds.slice(0, observed ? count : 0).map(runId => ({ observedAtMs: value.startedAtMs, + runId, status: "failed", runtimeMode: "native", nativePhase: "retryable_failure", errorCode: code })), + ...value.lastStatuses.map(row => ({ observedAtMs: value.assessedAtMs, ...row }))]; + Object.assign(value, { uniqueRetryableRuns: observed ? count : 0, recoveredRuns: observed ? count : 0, + retryableErrorClasses: observed && count ? { [code]: count } : {} }); + for (let index = 0; index < count; index++) Object.assign(item.recordedFailures.rows[index], { + recovered_type: "object", recovered_schema: "paperclip.recovered_execution_failure.v1", recovered_error_code: code, + recovered_observed_at: new Date(item.settledAtMs - 1).toISOString() }); + }; + const retryNoise = path.join(directory, "retry-aa.json"); + const retryAA = run(["--phase=aa", `--pairs=${await manifest("retry-aa", 181000, { sameRevision: true, + mutateArm: (report, { arm }) => setRetries(report, arm === "candidate" ? 2 : 1) })}`, `--output=${retryNoise}`]); + assert.equal(retryAA.status, 0, retryAA.stderr); + const calibrated = JSON.parse(await readFile(retryNoise, "utf8")).retryObservations[3]; + assert.equal(calibrated.rates.find(rate => rate.errorClass === null).aaNoiseEnvelopeAbsolute, .01); + assert.equal(calibrated.samples[0].baseline.uniqueRetryableRuns, 1); + assert.equal(calibrated.samples[0].candidate.recoveredRuns, 2); + const staleCurrent = run(["--candidate-policy=final-b2", + `--pairs=${await manifest("stale-current-error", 1000, { + mutateArm: report => setRetries(report, 1), + mutate: report => Object.assign(report.cases[3].recordedFailures.rows[0], { + current_has_error: true, current_error_code: "provider_transport_failed" }) })}`, + `--noise-report=${retryNoise}`, "--require-performance"]); + assert.notEqual(staleCurrent.status, 0, "An unresolved current error is not recovered merely because status says succeeded"); + assert.match(staleCurrent.stderr, /must clear their current execution error/); + for (const [name, mutation, expectedPass] of [ + ["retry-at-frozen-noise", (report, { arm }) => setRetries(report, arm === "candidate" ? 2 : 1), true], + ["one-arm-retry-storm", (report, { arm, index }) => setRetries(report, arm === "candidate" && index === 0 ? 3 : 1), false], + ["new-retry-class", (report, { arm }) => setRetries(report, 1, arm === "candidate" ? "new_transport_failure" : "provider_transport_failed"), false], + ]) { + const output = path.join(directory, `${name}.json`); + const attempt = run(["--candidate-policy=final-b2", `--pairs=${await manifest(name, 1000, { mutateArm: mutation })}`, + `--noise-report=${retryNoise}`, `--output=${output}`, "--require-performance"]); + assert.equal(attempt.status === 0, expectedPass, attempt.stderr); + const observed = JSON.parse(await readFile(output, "utf8")); + assert.equal(observed.performancePassed, expectedPass); + if (name === "one-arm-retry-storm") { + const gate = observed.retryGates.find(gate => gate.name === "burst512" && gate.errorClass === null); + assert.equal(gate.medianIncrease, 0); assert.equal(gate.pairedIncreases[0], .02); assert.equal(gate.passed, false); + } + if (name === "new-retry-class") assert.ok(observed.retryGates.some(gate => !gate.knownClass && !gate.passed)); + } + const zeroBaselineRetries = run(["--candidate-policy=final-b2", + `--pairs=${await manifest("zero-baseline-new-retry", 1000, { mutate: report => setRetries(report, 1) })}`, + `--noise-report=${noise}`, "--require-performance"]); + assert.notEqual(zeroBaselineRetries.status, 0, "A zero-retry A/A cannot silently admit a new retry class or rate"); + const earlyFile = path.join(directory, "early-recovery.json"); + const early = run(["--candidate-policy=final-b2", + `--pairs=${await manifest("early-recovery", 1000, { mutate: report => setRetries(report, 1, "provider_transport_failed", false) })}`, + `--noise-report=${noise}`, `--output=${earlyFile}`, "--require-performance"]); + assert.notEqual(early.status, 0, "Recovery before the terminal observation window cannot vanish from failure rates"); + const earlyObservation = JSON.parse(await readFile(earlyFile, "utf8")).retryObservations[3].samples[0].candidate; + assert.equal(earlyObservation.uniqueRetryableRuns, 0); assert.equal(earlyObservation.uniqueEverFailedRuns, 1); + assert.equal(earlyObservation.allRecordedRecoveredRuns, 1); for (const [name, change, message] of [ ["h4c-too-early", (_value, owner) => { owner.results[0].startTime = "2026-11-02T00:00:00.000Z"; }, /follow all candidate/], ["h4c-wrong-sut", value => { value.sourceRevision = H5_BASELINE_REVISIONS.B0; value.compiledBuildRevision = value.sourceRevision; }, /strictly equal/], - ["h4c-observation-mode", value => { value.observationMode = H5_MODE; }, /H4d default deadlines/], + ["h4c-observation-mode", value => { value.observationMode = H5_MODE; }, /H4e default deadlines/], ["old-h4c-proof", value => { value.protocolVersion = "v4-remote-streaming-original-local-stop-clock-stream"; value.harnessProvenance.sourceRevision = "7d7e7a88d562a95a1d407b76474763b666c181f5"; }, /Old H4c geometry/], - ["h4d-wrong-harness", value => { value.harnessProvenance.sourceRevision = "d".repeat(40); }, /same frozen harness source/], - ["h4d-wrong-measurement", value => { value.measurementSha256 = "old-h5a"; }, /same measured harness bytes/], - ["h4d-shared-agent", value => { value.cases[3].controllerAuthority.rows[1].agent_id = value.cases[3].controllerAuthority.rows[0].agent_id; }, /distinct normal agents/], - ["h4d-missing-stop-proof", value => { delete value.cases[5].localStopProof; }, /physical Stop/], - ["h4d-provider-still-alive", value => { value.cases[5].localStopProof.processes.providerAbsent = false; }, /provider remains alive/], - ["h4d-pending-local-lease", value => { value.cases[5].localStopProof.lease.status = "active"; }, /lease cleanup remains unsettled/], - ["h4d-wrong-local-lease", value => { value.cases[5].localStopProof.lease.id = value.cases[3].controllerAuthority.rows[0].lease_id; }, /strictly equal/], - ["h4d-ack-timing-mismatch", value => { value.cases[5].stopAcknowledgedAtMs++; }, /acknowledgement timing boundary/], - ["h4d-missing-remote-leases", value => { delete value.cases[3].remoteLeases; }, /post-timing lease cleanup proof/], - ["h4d-pending-remote-lease", value => { value.cases[3].remoteLeases[0].cleanupStatus = "pending"; }, /lease cleanup remains unsettled/], - ["h4d-wrong-remote-lease", value => { value.cases[3].remoteLeases[0].id = value.cases[5].controllerAuthority.rows[0].lease_id; }, /strictly equal/], - ["h4d-premature-remote-cleanup-proof", value => { value.cases[3].remoteLeasesCheckedAtMs = value.cases[3].settledAtMs - 1; }, /leases after recorded timing/], - ["h4d-wrong-authority-company", value => { value.cases[3].controllerAuthority.rows[0].company_id = "00000000-0000-4000-8000-000000099999"; }, /authority company mismatch/], - ["h4d-wrong-authority-run", value => { value.cases[3].controllerAuthority.rows[0].run_id = value.cases[5].runIds[0]; }, /authority run set mismatch/], + ["old-h4d-proof", value => { value.protocolVersion = "v4d-independent-agents-original-deadlines"; }, /Old H4c geometry/], + ["h4e-wrong-harness", value => { value.harnessProvenance.sourceRevision = "d".repeat(40); }, /same frozen harness source/], + ["h4e-wrong-measurement", value => { value.measurementSha256 = "old-h5a"; }, /same measured harness bytes/], + ["h4e-shared-agent", value => { value.cases[3].controllerAuthority.rows[1].agent_id = value.cases[3].controllerAuthority.rows[0].agent_id; }, /distinct normal agents/], + ["h4e-missing-stop-proof", value => { delete value.cases[5].localStopProof; }, /physical Stop/], + ["h4e-provider-still-alive", value => { value.cases[5].localStopProof.processes.providerAbsent = false; }, /provider remains alive/], + ["h4e-pending-local-lease", value => { value.cases[5].localStopProof.lease.status = "active"; }, /lease cleanup remains unsettled/], + ["h4e-wrong-local-lease", value => { value.cases[5].localStopProof.lease.id = value.cases[3].controllerAuthority.rows[0].lease_id; }, /strictly equal/], + ["h4e-ack-timing-mismatch", value => { value.cases[5].stopAcknowledgedAtMs++; }, /acknowledgement timing boundary/], + ["h4e-missing-remote-leases", value => { delete value.cases[3].remoteLeases; }, /post-timing lease cleanup proof/], + ["h4e-pending-remote-lease", value => { value.cases[3].remoteLeases[0].cleanupStatus = "pending"; }, /lease cleanup remains unsettled/], + ["h4e-wrong-remote-lease", value => { value.cases[3].remoteLeases[0].id = value.cases[5].controllerAuthority.rows[0].lease_id; }, /strictly equal/], + ["h4e-premature-remote-cleanup-proof", value => { value.cases[3].remoteLeasesCheckedAtMs = value.cases[3].settledAtMs - 1; }, /leases after recorded timing/], + ["h4e-wrong-authority-company", value => { value.cases[3].controllerAuthority.rows[0].company_id = "00000000-0000-4000-8000-000000099999"; }, /authority company mismatch/], + ["h4e-wrong-authority-run", value => { value.cases[3].controllerAuthority.rows[0].run_id = value.cases[5].runIds[0]; }, /authority run set mismatch/], + ["h4e-missing-settlement", value => { delete value.cases[3].terminalObservation; }, /canonical run settlement/], + ["h4e-missing-recorded-failures", value => { delete value.cases[3].recordedFailures; }, /post-timing recorded failure proof/], ]) { const report = JSON.parse(await readFile(originalProof, "utf8")), owner = report.suites[0].tests[0], value = JSON.parse(Buffer.from(owner.results[0].attachments[0].body, "base64").toString()); change(value, owner); owner.results[0].attachments[0].body = Buffer.from(JSON.stringify(value)).toString("base64"); const file = await save(`${name}-playwright.json`, report), sidecar = await inventory(value, file, name), output = path.join(directory, `${name}-result.json`); const attempt = run(["--candidate-policy=final-b2", `--pairs=${goodPairs}`, `--noise-report=${noise}`, - `--original-h4d-proof=${file}`, `--original-h4d-inventory=${sidecar}`, `--output=${output}`, "--require-acceptance"]); + `--original-h4e-proof=${file}`, `--original-h4e-inventory=${sidecar}`, `--output=${output}`, "--require-acceptance"]); assert.notEqual(attempt.status, 0, name); const rejected = JSON.parse(await readFile(output, "utf8")); assert.equal(rejected.performancePassed, true); assert.equal(rejected.acceptancePassed, false); - assert.match(rejected.originalH4dConfirmation.reason, message); + assert.match(rejected.originalH4eConfirmation.reason, message); } const badOutput = path.join(directory, "still-misses.json"); const bad = run(["--candidate-policy=final-b2", `--pairs=${await manifest("still-misses", 150000)}`, @@ -357,6 +434,17 @@ test("H5 CLI retains baseline SLA failure, rejects final SLA misses and requires assert.equal(rejected.h5Assessments[0].candidateSlaPolicyPassed, false); for (const [name, mutate, message] of [ ["old-h5a-protocol", report => { report.protocolVersion = "v5-complete-observation-original-sla"; }, /strictly equal/], + ["old-h5b-protocol", report => { report.protocolVersion = "v5b-independent-agents-complete-observation-original-sla"; }, /strictly equal/], + ["retry-ledger-hidden", report => { setRetries(report, 1); report.cases[3].h5.terminalObservation.uniqueRetryableRuns = 0; }, /Retry ledger/], + ["missing-state-history", report => { delete report.cases[3].h5.terminalObservation.transitions; }, /state history/], + ["missing-recorded-failures", report => { delete report.cases[3].recordedFailures; }, /post-timing recorded failure proof/], + ["wrong-recorded-run", report => { report.cases[3].recordedFailures.rows[0].run_id = report.cases[5].runIds[0]; }, /run set mismatch/], + ["wrong-recorded-company", report => { report.cases[3].recordedFailures.rows[0].company_id = "wrong"; }, /company mismatch/], + ["wrong-recovered-schema", report => { setRetries(report, 1); report.cases[3].recordedFailures.rows[0].recovered_schema = "unknown"; }, /strictly equal/], + ["missing-recorded-code", report => { setRetries(report, 1); report.cases[3].recordedFailures.rows[0].recovered_error_code = null; }, /closed recorded error code/], + ["premature-recorded-snapshot", report => { report.cases[3].recordedFailures.queryStartedAtMs = report.cases[3].settledAtMs - 1; }, /post-timing failure snapshot/], + ["unclosed-recorded-snapshot", report => { delete report.cases[3].recordedFailures.connectionClosedAtMs; }, /post-timing failure snapshot/], + ["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/], ["result-missing", report => { report.cases[3].durable[99].resultAcceptedEvents = 0; }, /persisted result/], @@ -402,7 +490,7 @@ test("browser CLI rejects the three-pairs-above-noise counterexample and ineligi } return save(`${prefix}-pairs.json`, entries); } - const run = args => spawnSync(process.execPath, [script.pathname, ...args], { encoding: "utf8" }); + const run = args => spawnSync(process.execPath, [script.pathname, ...args], { encoding: "utf8", maxBuffer: 32 * 1024 * 1024 }); try { const noise = path.join(directory, "noise.json"); const aa = run(["--phase=aa", `--pairs=${await manifest("aa", Array(5).fill(0.10), true)}`, `--output=${noise}`]); diff --git a/tests/e2e/fixtures/native-ingestion-h5.md b/tests/e2e/fixtures/native-ingestion-h5.md index fd0be3f4378..94c16439628 100644 --- a/tests/e2e/fixtures/native-ingestion-h5.md +++ b/tests/e2e/fixtures/native-ingestion-h5.md @@ -1,10 +1,11 @@ -# H5b independent-agent observation registration (prospective; no execution release) +# H5c canonical-settlement observation registration (prospective; no execution release) H4c remains frozen at `7d7e7a88d562a95a1d407b76474763b666c181f5` (`5db4129bbdc1fa38775121aff5d0e68a8a11762519e41cfdf36eacd90ba7256b`). Its observed 120-second streaming failures remain failures. Aborted or profiled -attempts cannot be reused as complete timing arms. H5b uses one normal agent per concurrent remote participant. It repairs the -experiment topology while preserving observation of original deadline outcomes. +attempts cannot be reused as complete timing arms. H5c keeps one normal agent per +concurrent remote participant and waits for the product's canonical settlement, +including supported same-run recovery, within the same terminal deadline. It does not optimize the product or turn earlier failures into passes. The aligned fresh application baselines are B0 `bac53ec3bfeaf6546e7aaeaee516487b30534654` @@ -16,12 +17,14 @@ or matching binary is presumed before those artifacts exist. The historical a365/d8cd/914 application failures remain evidence of their recorded revisions. Select H5 explicitly with -`PAPERCLIP_INGESTION_OBSERVATION_MODE=h5b-observe-original-sla-independent-agents`. -The new default H4d mode uses the same independent-agent geometry with the original -120/180/900-second deadlines. Observation mode uses the explicit H5b protocol. +`PAPERCLIP_INGESTION_OBSERVATION_MODE=h5c-observe-original-sla-independent-agents-settlement`. +The new default H4e mode uses the same independent-agent geometry with the original +120/180/900-second deadlines. Observation mode uses the explicit H5c protocol. Both modes hash the same nine source files, including the shared topology helper. Their source revision and measurement bytes must match for later confirmation. -Old H4c/H5a protocols, geometry, arms and noise are ineligible for H5b. +Old H4c/H5a protocols, geometry, arms and noise are ineligible for H5c. H5b179fe6 +and its H4d default proof use the premature terminal-status observer and are also +ineligible. The new H4e proof must use the exact H5c source and geometry. Run `node scripts/describe-native-ingestion-h5.mjs` to inspect the registration. This is a dry run only: it starts no app, provider, database or browser and @@ -77,7 +80,86 @@ A separate background presentation40P01 involved a run that succeeded. The retained log shows its coordinator→issue→run wait, but lacks PostgreSQL's full opposing lock cycle. Its cause and relationship to other contention are unproved. It is not evidence that the14 run failures came from a database deadlock. All -historical failures remain separate; no failed H4c/H5a arm is reused as H5b noise. +historical failures remain separate; no failed H4c/H5a arm is reused as H5c noise. + +## Retained H5b failure and canonical settlement correction + +H5b179fe6 on aligned B0bac53 admitted100 distinct normal agents before the common +epoch. Its original120s poll failed with61/512 foreground markers; all512 became +visible282.734s after the wait began. The observer stopped54.737s later on83 +succeeded and17 failed statuses. Independent PostgreSQL checks found all51,200 +lossless deltas and one accepted result per run. Qualification failed; history +and Stop were not reached. This arm is not noise or performance evidence. + +All17 errors were provider_transport_failed with the generic message that the +runner did not durably suspend before checkpoint. Existing run logs instead +show runnerSuspended=true and providerDrained=false: the exact runner.drain +receipt was still pending after its five-second drain window. Why the receipt +was late remains unproved. These were not the earlier shared-agent lock errors. +All17 finalizers were retryable_failure at attempt1 with a scheduled+30s workspace +retry; application teardown finished before the earliest retry was due. ACTIVE +leases and prepared workspace-sync receipts remained unfinished. Later recovery +was not observed and is not presumed to pass. + +The exact B0bac53 run-failure-report.ts44–50 contract states: + +> A native run that waits for a same-run retry is stored as `failed` with +> `nativePhase: "retryable_failure"`. That is not terminal: the retry or the +> workspace copy-back retry can still finish the run as succeeded. + +Normal company-run lists omit +runtimeMode/nativePhase. Normal GET /api/heartbeat-runs/:id exposes both through +the existing full run columns. Inside the same single180_000ms terminal poll, +resolve only failed list candidates through that detail GET and classify its +current result. Keep failed/native/retryable_failure pending; permanent/unknown +failures and GET errors reject. The observer sends no retry or provider command. + +Record bounded state changes per run with time, status, runtimeMode, nativePhase +and errorCode. Missing list-only fields are null, not inferred. Cap at16 changes +per run and fail on overflow rather than discard observations. Retain total +unique retryable runs, recovered runs, nonretryable failures and per-error-class +unique-run counts. Every observed retry must recover successfully within +the unchanged180s window; a deadline miss remains censored failure, never a +completed latency. All100 successful native runs, exact deltas/results, reload, +lease cleanup, existing-history and local Stop gates remain required. + +After recorded settlement, acknowledgement, lease timing, reload and normal +event verification, take one company-and-exact-run-set READ ONLY REPEATABLE READ +snapshot through the existing owned PostgreSQL helper. Retain current error-code +fields and the closed schema/errorCode/observedAt fields of +result_json.recoveredExecutionFailure, plus descriptive finalization attempt. +Require every actual run, exact identities, settled status, valid shapes and a +closed connection before the next case. No event HTTP or database polling loop +is added. The existing independent completion upper bound includes this check; +the measured settlement and lease timestamps stay unchanged. + +The durable recovery marker is copied from the run's error fields when successful +finalization clears them (native-run-projection.ts114–121 in B0bac53). It can +therefore retain a failure recovered before terminal-window polling. Union its +codes and current recorded errors with observed transition codes by run identity. +The normal Stop path's closed cancelled code is retained as cancellation +evidence (run-control.ts1345); other current Stop error codes remain failures. +Use these all-recorded unique-ever-failed-run counts/rates for comparison, and +retain the observed-only counters separately. This marker records the retained +failure, not an exhaustive sequence of every error class or attempt. Do not claim +omniscient error history. Finalization attempt is a claim counter that can be +preserved across workspace repair (native-run-finalizer.ts237–240); it is not a +failure count and supplies no rate gate. + +Successful committed ordinary runs must have both current error fields cleared, +as required by native-run-projection.ts180–193. A still-current error fails the +proof; a succeeded status alone cannot turn it into recovered failure evidence. + +Fresh A/A retains all histories and freezes, per case, total and per-error-class +all-recorded unique-ever-failed-run rates plus the maximum absolute paired rate difference. Zero baseline +counts are valid; no relative division is used. Each candidate may use only +error classes observed in that baseline A/A. Every paired total/per-class rate +increase and its median must stay within that frozen absolute noise envelope. +A single-arm retry storm cannot hide behind the median. Recovery and its final +backlog remain inside drainage timing; observing recovery does not mean the +earlier errors disappeared. All recorded failures must have recovered, and every +run still needs its exact independently persisted accepted result. No product +retry policy or drain limit is changed. ## Fixed work and observation boundaries @@ -95,7 +177,7 @@ is added. Fake-plugin streaming remains separate from real-cloud Stop proof. | Boundary | Original requirement | H5 observation | | --- | --- | --- | | Foreground markers | Original Playwright120s wait | Continue to an absolute600s ceiling from that same wait start | -| All terminal runs |180s after foreground completion (and Stop click) | Same180s after eventual foreground completion; hard failure at the ceiling | +| Canonically settled runs |180s after foreground completion (and Stop click) | Same single180s after eventual foreground completion; supported retries remain pending, permanent failures reject, deadline misses fail | | Stop acknowledgement and leases | Existing30s polls and ordering | Unchanged; full click→acknowledgement metric retained | | Invocation | Original900s test budget | Actual1200s observation timeout, separately disclosed | @@ -131,7 +213,7 @@ status, errors and configured timeout. Its observed total duration is an independent conservative assessment of the original900s budget; hooks/teardown can be treated differently by the original timeout. It is not an exact original timer-expiration observation. A final candidate must have total duration<900s -in every arm and later pass one unchanged H4d default-mode full invocation on the +in every arm and later pass one unchanged H4e default-mode full invocation on the same final SUT revision and full qualified artifact inventory. ## Complete evidence and eligibility @@ -167,7 +249,7 @@ not itself attest that the live filesystem stayed unchanged. | --- | --- | --- | | Fresh B0/B1 A/A arms | Known burst512 streaming120 miss; explicit conservative observed-duration900 miss | Complete finite correct observation; root-reviewed noise suitability | | B0→B1 cursor-only candidate | Only an exact registered known miss also present in the paired baseline: burst512 streaming120 or conservative observed-duration900 | No additional deadline failures; latency/control nonregression plus separate incremental SQL evidence; never claim SLA fixed | -| Final B2 candidate in either comparison | None | Every arm meets original streaming/control checks and conservative900 gate; loaded improvements and later H4d proof | +| Final B2 candidate in either comparison | None | Every arm meets original streaming/control checks and conservative900 gate; loaded improvements and later H4e proof | Only burst512's streaming miss is registered as a known workload SLA failure. New failures in other workloads require diagnosis. The existing Stop poll @@ -190,13 +272,13 @@ improve; other fixed latency/control metrics must not regress outside noise. Do not infer p99 from sparse observations. There is no new fixed percentage A/A cutoff: root must reject unsuitable noisy/nonstationary calibration explicitly. -Use `--observation-mode=h5b-observe-original-sla-independent-agents` for the comparator. Candidate +Use `--observation-mode=h5c-observe-original-sla-independent-agents-settlement` for the comparator. Candidate comparisons also require `--candidate-policy=cursor-nonregression --expect=nonregression` or `--candidate-policy=final-b2`. Final comparisons expose `performancePassed` -separately; `acceptancePassed` remains false until `--original-h4d-proof` and -`--original-h4d-inventory` establish the later fresh H4d default-deadline functional pass. -That H4d proof is full, unprofiled, retry0, default mode with original120/180/900s -bounds, on the same frozen H5b harness source, nine-file measurement bytes and +separately; `acceptancePassed` remains false until `--original-h4e-proof` and +`--original-h4e-inventory` establish the later fresh H4e default-deadline functional pass. +That H4e proof is full, unprofiled, retry0, default mode with original120/180/900s +bounds, on the same frozen H5c harness source, nine-file measurement bytes and independent-agent geometry, and the exact final SUT revision/qualified inventory. It uses a new disposable company and starts after all candidate arms. Old7d/H4c proofs and H5a observations cannot confirm this geometry. Prior failed H4c attempts remain in the record. @@ -206,7 +288,7 @@ ordering; no absolute click-to-acknowledgement deadline is added. Use the explicit H5 `--require-performance` gate for the two final comparison stages while that later proof is pending; this requires every performance and candidate original-SLA gate, but grants no final acceptance. After the one later -H4d default-mode qualification, recompute both reports against their unchanged paired +H4e default-mode qualification, recompute both reports against their unchanged paired manifests with the proof and `--require-acceptance`. This re-evaluates retained evidence; it neither reuses an arm in a different comparison nor reruns failures. diff --git a/tests/e2e/fixtures/native-ingestion-h5.mjs b/tests/e2e/fixtures/native-ingestion-h5.mjs index 564511d7a26..3ba362667bc 100644 --- a/tests/e2e/fixtures/native-ingestion-h5.mjs +++ b/tests/e2e/fixtures/native-ingestion-h5.mjs @@ -1,9 +1,9 @@ import assert from 'node:assert/strict'; import { createHash } from 'node:crypto'; -export const H5_MODE = 'h5b-observe-original-sla-independent-agents'; -export const H5_PROTOCOL = 'v5b-independent-agents-complete-observation-original-sla'; -export const H4D_PROTOCOL = 'v4d-independent-agents-original-deadlines'; +export const H5_MODE = 'h5c-observe-original-sla-independent-agents-settlement'; +export const H5_PROTOCOL = 'v5c-independent-agents-complete-settlement-original-sla'; +export const H4E_PROTOCOL = 'v4e-independent-agents-settlement-original-deadlines'; export const H5_TOPOLOGY = Object.freeze({ mode: 'one-normal-agent-per-concurrent-participant', fullRemoteAgentPoolSize: 100, agentSetupConcurrency: 8, maxConcurrentRuns: 50, poolReuse: 'only-across-sequential-cases' }); @@ -12,6 +12,7 @@ export const H5_BASELINE_REVISIONS = Object.freeze({ B0: 'bac53ec3bfeaf6546e7aae export const ORIGINAL_BUDGETS = Object.freeze({ providerReadinessMs: 300_000, fullInvocationMs: 900_000, streamingMs: 120_000, terminalDrainageMs: 180_000, stopAcknowledgementMs: 30_000, remoteLeaseCleanupMs: 30_000 }); export const H5_LIMITS = Object.freeze({ streamingMs: 600_000, terminalDrainageMs: 180_000, fullInvocationMs: 1_200_000 }); +export const SETTLEMENT_MAX_TRANSITIONS_PER_RUN = 16; export const H5_WORKLOADS = Object.freeze([ { name: 'paced', runs: 1, events: 128, intervalMs: 25, prelude: 0, stop: false }, { name: 'burst16', runs: 1, events: 16, intervalMs: 0, prelude: 0, stop: false }, @@ -22,14 +23,19 @@ export const H5_WORKLOADS = Object.freeze([ ]); export const H5_REGISTRATION = Object.freeze({ mode: H5_MODE, protocolVersion: H5_PROTOCOL, baselineRevisions: H5_BASELINE_REVISIONS, - topology: H5_TOPOLOGY, originalDeadlineProtocol: H4D_PROTOCOL, + topology: H5_TOPOLOGY, originalDeadlineProtocol: H4E_PROTOCOL, + settlement: { classification: 'Failed native retryable_failure remains pending within the original terminal window', + detailReads: 'Only failed run-list candidates use the normal detail GET; no event polling or retry dispatch', + maxTransitionsPerRun: SETTLEMENT_MAX_TRANSITIONS_PER_RUN, + retryRatePolicy: 'No new recorded error class after frozen A/A; every paired and median total/per-class unique-ever-failed-run rate increase must stay within absolute paired A/A noise', + recordedFailures: 'Post-timing exact-run-set read-only snapshot unions current and durable recovered error codes with observed transitions; retained recovery metadata is last-failure evidence, not exhaustive per-class history' }, originalBudgets: ORIGINAL_BUDGETS, observationLimits: H5_LIMITS, workloads: H5_WORKLOADS, pairs: 5, order: 'baseline-first on even pair index; candidate-first on odd pair index', stages: ['B0 A/A', 'B1 A/A', 'B0→B1 nonregression', 'B1→B2 improvement', 'B0→B2 improvement'], knownBaselineMisses: ['burst512:streaming120', 'invocation:observed-total-duration900'], - baselineEligibility: 'Finite complete correct observation; SLA misses remain failures of the H4d original deadlines', + baselineEligibility: 'Finite complete correct observation; SLA misses remain failures of the H4e original deadlines', cursorCandidateEligibility: 'Nonregression; only the paired baseline exact registered known misses may be inherited', - finalCandidateEligibility: 'Every original SLA passes in every arm; independently assessed total duration <900s; later H4d default-mode functional PASS on the same harness source and independent-agent geometry required', + finalCandidateEligibility: 'Every original SLA passes in every arm; independently assessed total duration <900s; later H4e default-mode functional PASS on the same harness source and independent-agent geometry required', noise: 'Fresh B0 and B1 A/A frozen before any further product implementation; no old abort/profile samples', performance: 'Loaded p95 emission/intended-to-visible and terminal drainage improve beyond frozen noise in median and at least4/5 pairs; all other metrics nonregress', }); @@ -156,11 +162,178 @@ export async function observeStreaming({ readCount, poll, target, now = Date.now export function originalTerminal(streaming, terminal) { return streaming.original.status === 'passed' - ? { status: terminal.complete ? 'passed' : 'failed', reason: terminal.complete ? null : 'original_terminal_deadline', + ? { status: terminal.complete ? 'passed' : 'failed', reason: terminal.complete ? null + : terminal.censored === false && terminal.error ? 'original_terminal_failure' : 'original_terminal_deadline', startedAtMs: terminal.startedAtMs, deadlineMs: terminal.deadlineMs, assessedAtMs: terminal.assessedAtMs } : { status: 'not-reached', reason: 'original_streaming_deadline', startedAtMs: null, deadlineMs: null, assessedAtMs: null }; } +/** The run list omits nativePhase. Resolve failed rows through the normal run + * detail API before deciding whether the existing terminal window has drained. */ +export async function observeRunSettlement({ readRuns, readRun, poll, expectedRuns, now = Date.now, onOutcome = _value => {} }) { + const startedAtMs = now(); + const outcome = { startedAtMs, deadlineMs: startedAtMs + ORIGINAL_BUDGETS.terminalDrainageMs, + assessedAtMs: 0, lastObservedAtMs: 0, observedRuns: 0, terminalRuns: 0, pendingRetryRuns: 0, + lastStatuses: [], transitions: [], uniqueRetryableRuns: 0, recoveredRuns: 0, nonretryableFailures: 0, + retryableErrorClasses: {}, complete: false, censored: false }; + const previous = new Map(); + let runs = []; + onOutcome(outcome); + try { + await poll(async () => { + runs = []; + for (const listed of await readRuns()) { + const run = listed.status === 'failed' ? await readRun(listed.id) : listed; + assert.equal(run.id, listed.id, 'Run detail belongs to a different run'); + if (listed.status === 'failed') { + assert.ok(typeof run.runtimeMode === 'string', 'Run detail must expose runtimeMode'); + if (run.runtimeMode === 'native') assert.ok(Object.hasOwn(run, 'nativePhase'), 'Run detail must expose nativePhase'); + } + runs.push(run); + } + assert.ok(runs.length <= expectedRuns && new Set(runs.map(run => run.id)).size === runs.length, + 'Terminal observation contains duplicate or additional runs'); + const observedAtMs = now(); + const statuses = runs.map(run => ({ runId: run.id, status: run.status, + runtimeMode: run.runtimeMode ?? null, nativePhase: run.nativePhase ?? null, errorCode: run.errorCode ?? null })); + for (const state of statuses) { + const signature = JSON.stringify(state); + if (previous.get(state.runId) !== signature) { + assert.ok(outcome.transitions.filter(row => row.runId === state.runId).length < SETTLEMENT_MAX_TRANSITIONS_PER_RUN, + 'Terminal state transition evidence overflow; no observation may be dropped'); + outcome.transitions.push({ observedAtMs, ...state }); + } + previous.set(state.runId, signature); + } + const pending = run => run.status === 'failed' && run.runtimeMode === 'native' && run.nativePhase === 'retryable_failure'; + const settled = run => ['succeeded', 'cancelled', 'failed'].includes(run.status) && !pending(run); + Object.assign(outcome, { lastObservedAtMs: observedAtMs, observedRuns: runs.length, + terminalRuns: runs.filter(settled).length, pendingRetryRuns: runs.filter(pending).length, lastStatuses: statuses, + ...summarizeSettlementTransitions(outcome.transitions) }); + onOutcome(outcome); + assert.ok(!runs.some(run => run.status === 'timed_out' || run.status === 'failed' && !pending(run)), + 'Terminal run failed; automatic retry is not pending'); + return runs.length === expectedRuns && runs.every(settled); + }, ORIGINAL_BUDGETS.terminalDrainageMs); + assert.equal(outcome.recoveredRuns, outcome.uniqueRetryableRuns, 'Every observed retryable run must recover successfully'); + outcome.assessedAtMs = now(); outcome.complete = true; onOutcome(outcome); + return { runs, outcome }; + } catch (error) { + outcome.assessedAtMs = now(); + outcome.censored = error?.matcherResult?.name === 'toBe' && error.matcherResult.pass === false + && error.message.includes(`Timeout ${ORIGINAL_BUDGETS.terminalDrainageMs}ms exceeded while waiting on the predicate`); + outcome.error = String(error); onOutcome(outcome); throw error; + } +} + +export function summarizeSettlementTransitions(transitions) { + const retryable = new Set(), recovered = new Set(), permanent = new Set(), classes = new Map(); + for (const row of transitions) { + if (row.status === 'failed' && row.runtimeMode === 'native' && row.nativePhase === 'retryable_failure') { + assert.ok(typeof row.errorCode === 'string' && row.errorCode.length > 0, 'Retryable failure needs its observed error class'); + retryable.add(row.runId); + const runs = classes.get(row.errorCode) ?? new Set(); runs.add(row.runId); classes.set(row.errorCode, runs); + } else if (row.status === 'failed' || row.status === 'timed_out') permanent.add(row.runId); + if (row.status === 'succeeded' && retryable.has(row.runId)) recovered.add(row.runId); + } + return { uniqueRetryableRuns: retryable.size, recoveredRuns: recovered.size, nonretryableFailures: permanent.size, + retryableErrorClasses: Object.fromEntries([...classes].sort(([a], [b]) => a.localeCompare(b)).map(([code, runs]) => [code, runs.size])) }; +} + +export function assertSettlementObservation(value, item) { + assert.ok(value, 'Missing canonical run settlement observation'); + assert.equal(value.complete, true); assert.equal(value.censored, false); + assert.ok(Number.isSafeInteger(value.startedAtMs) && value.startedAtMs > 0); + assert.equal(value.deadlineMs, value.startedAtMs + ORIGINAL_BUDGETS.terminalDrainageMs); + assert.equal(value.assessedAtMs, item.settledAtMs); + assert.ok(value.lastObservedAtMs >= value.startedAtMs && value.lastObservedAtMs <= value.assessedAtMs + && value.assessedAtMs <= value.deadlineMs, 'Settlement must fit the original terminal window'); + assert.equal(value.observedRuns, item.runs); assert.equal(value.terminalRuns, item.runs); assert.equal(value.pendingRetryRuns, 0); + assert.deepEqual(value.lastStatuses.map(row => row.runId).sort(), [...item.runIds].sort()); + assert.ok(value.lastStatuses.every(row => row.status === (item.stop ? 'cancelled' : 'succeeded')), 'Failed runs are not settled success'); + assert.ok(Array.isArray(value.transitions) && value.transitions.length >= item.runs + && value.transitions.length <= item.runs * SETTLEMENT_MAX_TRANSITIONS_PER_RUN, 'Missing or overflowing bounded state history'); + const last = new Map(), counts = new Map(); let priorTime = value.startedAtMs; + for (const row of value.transitions) { + assert.ok(item.runIds.includes(row.runId), 'State history belongs to another run'); + assert.ok(Number.isSafeInteger(row.observedAtMs) && row.observedAtMs >= priorTime + && row.observedAtMs <= value.lastObservedAtMs, 'Invalid state history timestamp'); + priorTime = row.observedAtMs; + for (const field of ['runtimeMode', 'nativePhase', 'errorCode']) assert.ok(row[field] === null || typeof row[field] === 'string'); + const count = (counts.get(row.runId) ?? 0) + 1; counts.set(row.runId, count); + assert.ok(count <= SETTLEMENT_MAX_TRANSITIONS_PER_RUN, 'State history overflow'); + last.set(row.runId, row); + } + for (const row of value.lastStatuses) { + const { observedAtMs: _at, ...state } = last.get(row.runId) ?? {}; + assert.deepEqual(state, row, 'Final state must match retained state history'); + } + const summary = summarizeSettlementTransitions(value.transitions); + for (const [field, expected] of Object.entries(summary)) assert.deepEqual(value[field], expected, 'Retry ledger must match retained transitions'); + assert.equal(summary.nonretryableFailures, 0, 'Nonretryable failures cannot qualify'); + assert.equal(summary.recoveredRuns, summary.uniqueRetryableRuns, 'Every observed retryable run must recover successfully'); + return summary; +} + +export function assertRecordedFailures(proof, item, companyId) { + const observed = assertSettlementObservation(item.h5?.terminalObservation ?? item.terminalObservation, item); + assert.equal(proof?.kind, 'recorded-run-failure-codes', 'Missing post-timing recorded failure proof'); + assert.equal(proof.transaction, 'READ ONLY REPEATABLE READ'); assert.equal(proof.observedReadOnly, 'on'); + assert.equal(proof.observedIsolation, 'repeatable read'); + const after = Math.max(item.settledAtMs, item.remoteLeasesCheckedAtMs ?? 0, item.stopAcknowledgedAtMs ?? 0, + item.localStopProof?.checkedAtMs ?? 0); + assert.ok(Number.isSafeInteger(proof.queryStartedAtMs) && proof.queryStartedAtMs >= after + && Number.isSafeInteger(proof.checkedAtMs) && proof.checkedAtMs >= proof.queryStartedAtMs + && Number.isSafeInteger(proof.connectionClosedAtMs) && proof.connectionClosedAtMs >= proof.checkedAtMs, + 'Close the post-timing failure snapshot after settlement and lease timing'); + assert.equal(proof.rows?.length, item.runs, 'Recorded failure proof needs every actual run'); + assert.deepEqual(proof.rows.map(row => row.run_id).sort(), [...item.runIds].sort(), 'Recorded failure run set mismatch'); + assert.equal(new Set(proof.rows.map(row => row.run_id)).size, item.runs); + const classes = new Map(), ever = new Set(), recovered = new Set(); + const add = (runId, code) => { + assert.ok(typeof code === 'string' && code.length > 0 && code.length <= 128, 'Missing closed recorded error code'); + ever.add(runId); const ids = classes.get(code) ?? new Set(); ids.add(runId); classes.set(code, ids); + }; + const transitions = (item.h5?.terminalObservation ?? item.terminalObservation).transitions; + for (const row of transitions) if (row.status === 'failed') add(row.runId, row.errorCode); + for (const row of proof.rows) { + assert.equal(row.company_id, companyId, 'Recorded failure company mismatch'); + assert.equal(row.agent_id, item.controllerAuthority.rows.find(authority => authority.run_id === row.run_id)?.agent_id, + 'Recorded failure agent mismatch'); + assert.equal(row.runtime_mode, 'native'); + assert.equal(row.status, item.stop ? 'cancelled' : 'succeeded', 'Recorded run is not successfully settled'); + if (!item.stop) assert.equal(row.native_phase, 'committed', 'Recorded native finalization is not committed'); + else assert.ok(row.native_phase === null || typeof row.native_phase === 'string', 'Invalid recorded cancellation phase'); + assert.ok(typeof row.current_has_error === 'boolean'); + assert.ok(row.current_error_code === null || typeof row.current_error_code === 'string'); + if (!item.stop) { + assert.equal(row.current_has_error, false, 'Successful committed runs must clear their current execution error'); + assert.equal(row.current_error_code, null, 'Successful committed runs must clear their current execution error code'); + } + // The normal Stop path writes the closed code "cancelled". Retain it as + // cancellation evidence; every other recorded code remains failure evidence. + if ((row.current_has_error || row.current_error_code !== null) && !(item.stop && row.current_error_code === 'cancelled')) + add(row.run_id, row.current_error_code); + if (row.recovered_type === null || row.recovered_type === 'null') { + for (const key of ['recovered_schema', 'recovered_error_code', 'recovered_observed_at']) assert.equal(row[key], null); + } else { + assert.equal(row.recovered_type, 'object'); assert.equal(row.recovered_schema, 'paperclip.recovered_execution_failure.v1'); + assert.ok(typeof row.recovered_observed_at === 'string' && Number.isFinite(Date.parse(row.recovered_observed_at)) + && Date.parse(row.recovered_observed_at) <= proof.checkedAtMs, 'Invalid recovered failure observation time'); + add(row.run_id, row.recovered_error_code); + } + assert.ok(row.finalization_attempt === null || Number.isSafeInteger(row.finalization_attempt) && row.finalization_attempt >= 0, + 'Invalid descriptive finalization attempt'); + if (ever.has(row.run_id) && row.status === 'succeeded') recovered.add(row.run_id); + } + assert.equal(recovered.size, ever.size, 'Every recorded failed run must have recovered successfully'); + assert.ok(item.durable.every(row => row.resultAcceptedEvents === (item.stop ? 0 : 1)), 'Every run needs its exact independently persisted result'); + if (item.h5?.independentDrain) assert.ok(item.h5.independentDrain.verifiedAtMs >= proof.connectionClosedAtMs, + 'Independent drainage must include the closed recorded-failure snapshot'); + return { ...observed, uniqueEverFailedRuns: ever.size, allRecordedRecoveredRuns: recovered.size, + recordedErrorClasses: Object.fromEntries([...classes].sort(([a], [b]) => a.localeCompare(b)).map(([code, ids]) => [code, ids.size])) }; +} + export function assessH5Arm(report, owner) { assert.equal(report.protocolVersion, H5_PROTOCOL); assert.equal(report.observationMode, H5_MODE); @@ -209,6 +382,7 @@ export function assessH5Arm(report, owner) { assert.equal(streaming.error, null); assert.equal(value.terminalOriginal.status, 'passed'); } + assertSettlementObservation(value.terminalObservation, item); assert.equal(value.terminalObservation.complete, true, 'H5 terminal observation must drain'); assert.equal(value.terminalObservation.censored, false); assert.equal(value.terminalObservation.deadlineMs, value.terminalObservation.startedAtMs + H5_LIMITS.terminalDrainageMs); diff --git a/tests/e2e/native-event-ingestion.journey.ts b/tests/e2e/native-event-ingestion.journey.ts index 14eb1474963..bcb49eedda3 100644 --- a/tests/e2e/native-event-ingestion.journey.ts +++ b/tests/e2e/native-event-ingestion.journey.ts @@ -7,8 +7,8 @@ import { hostname } from "node:os"; import path from "node:path"; import { expect, test } from "@playwright/test"; import { json } from "./agent-chat.shared"; -import { h5Enabled, H5_MODE, H5_PROTOCOL, H4D_PROTOCOL, H5_TOPOLOGY, H5_LIMITS, H5_WORKLOAD_SHA256, ORIGINAL_BUDGETS, - observeStreaming, originalTerminal, prepareParticipantAgents, assertParticipantTopology } from "./fixtures/native-ingestion-h5.mjs"; +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"; type SeenMarker = { markerCode: string; index: number; visibleMs: number }; type Emission = { nonce: string; markerCode: string; index: number; intendedMs: number; emittedMs: number; receivedAtMs: number }; @@ -44,7 +44,7 @@ const stats = (values: number[]) => { async function runtimeAuthority(input: { appRoot: string; companyId: string; environmentId: string; issueIds: string[]; transport: "local" | "sandbox-remote"; - afterStop?: boolean; + afterStop?: boolean; recordedFailureRunIds?: string[]; }) { const configuration = process.env.PAPERCLIP_E2E_SERVER_CONFIG; if (!configuration) throw new Error("Missing owned server configuration"); @@ -68,6 +68,21 @@ async function runtimeAuthority(input: { current_setting('transaction_isolation') as isolation`; expect(settings[0].read_only).toBe("on"); expect(settings[0].isolation).toBe("repeatable read"); + if (input.recordedFailureRunIds) { + const queryStartedAtMs = Date.now(); + const rows = await tx`select r.id as run_id, r.company_id, r.agent_id, r.runtime_mode, r.status, r.native_phase, + r.error_code as current_error_code, r.error is not null as current_has_error, + jsonb_typeof(r.result_json->'recoveredExecutionFailure') as recovered_type, + r.result_json->'recoveredExecutionFailure'->>'schema' as recovered_schema, + r.result_json->'recoveredExecutionFailure'->>'errorCode' as recovered_error_code, + r.result_json->'recoveredExecutionFailure'->>'observedAt' as recovered_observed_at, + f.attempt as finalization_attempt + from heartbeat_runs r left join native_run_finalizations f on f.run_id = r.id and f.company_id = r.company_id + where r.company_id = ${input.companyId} and r.id = any(${input.recordedFailureRunIds}::uuid[])`; + return { kind: "recorded-run-failure-codes", queryStartedAtMs, checkedAtMs: Date.now(), + transaction: "READ ONLY REPEATABLE READ", observedReadOnly: settings[0].read_only, + observedIsolation: settings[0].isolation, rows }; + } const rows = await tx`select r.id as run_id, r.company_id, r.agent_id, r.native_issue_id as issue_id, r.runtime_mode, r.native_session_id, r.runner_instance_id, r.status, r.process_pid, r.process_group_id, r.process_started_at, r.runner_profile_json->'nativeExecutionInput'->'binding'->>'executionWorkspaceId' as admitted_execution_workspace_id, @@ -115,7 +130,7 @@ async function runtimeAuthority(input: { } finally { await sql.end({ timeout: 5 }); } const connectionClosedAtMs = Date.now(); let localJournal; - if (input.transport === "local" && !input.afterStop) { + if (input.transport === "local" && !input.afterStop && !input.recordedFailureRunIds) { expect(input.issueIds).toHaveLength(1); const base = await realpath(path.join(path.dirname(configuration), "runtime/paperclip-runner/durable-sessions")); expect(base.startsWith(`${ownedHome}${path.sep}`)).toBe(true); @@ -275,7 +290,7 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy const browserRecording = { trace: captureMode(info.project.use.trace), video: captureMode(info.project.use.video), screencast: captureMode(info.project.use.trace) !== "off" || captureMode(info.project.use.video) !== "off" }; const report: Record = { - protocolVersion: h5 ? H5_PROTOCOL : H4D_PROTOCOL, transport, bodyStartedAtMs, + protocolVersion: h5 ? H5_PROTOCOL : H4E_PROTOCOL, transport, bodyStartedAtMs, agentTopology: { ...H5_TOPOLOGY, remoteAgentPoolSize: Math.max(...cases.map(item => item.runs)) }, ...(h5 ? { observationMode: H5_MODE, workloadSha256: H5_WORKLOAD_SHA256, observationLimits: H5_LIMITS, observationComplete: false, actualPlaywrightTimeoutMs: info.timeout, @@ -555,25 +570,20 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy } active.phase = "terminal-drainage"; let runs: any[] = []; - const terminalStartedAtMs = h5 ? Date.now() : undefined; - const terminalObservation = h5 ? { startedAtMs: terminalStartedAtMs!, deadlineMs: terminalStartedAtMs! + H5_LIMITS.terminalDrainageMs, - assessedAtMs: 0, lastObservedAtMs: 0, observedRuns: 0, terminalRuns: 0, - lastStatuses: [] as Array<{ runId: string; status: string }>, complete: false, censored: false } : undefined; - if (h5) active.terminalObservation = terminalObservation; - try { await expect.poll(async () => { - runs = (await json(await request.get(`/api/companies/${company.id}/heartbeat-runs?limit=1000`))).filter((run: { issueId?: string; nativeIssueId?: string; contextSnapshot?: { issueId?: string; taskId?: string } }) => issueIds.includes(run.issueId ?? run.nativeIssueId ?? run.contextSnapshot?.issueId ?? run.contextSnapshot?.taskId)); - if (terminalObservation) Object.assign(terminalObservation, { lastObservedAtMs: Date.now(), observedRuns: runs.length, - terminalRuns: runs.filter(run => ["succeeded", "cancelled", "failed"].includes(run.status)).length, - lastStatuses: runs.map(run => ({ runId: run.id, status: run.status })) }); - return runs.length === issueIds.length && runs.every(run => ["succeeded", "cancelled", "failed"].includes(run.status)); - }, { timeout: 180_000 }).toBe(true); } + let settlement; + try { settlement = await observeRunSettlement({ expectedRuns: issueIds.length, + readRuns: async () => (await json(await request.get(`/api/companies/${company.id}/heartbeat-runs?limit=1000`))).filter((run: { issueId?: string; nativeIssueId?: string; contextSnapshot?: { issueId?: string; taskId?: string } }) => issueIds.includes(run.issueId ?? run.nativeIssueId ?? run.contextSnapshot?.issueId ?? run.contextSnapshot?.taskId)), + readRun: async (runId: string) => json(await request.get(`/api/heartbeat-runs/${runId}`)), + poll: (read: () => Promise, milliseconds: number) => expect.poll(read, { timeout: milliseconds }).toBe(true), + onOutcome: (value: Record) => { active.terminalObservation = value; }, + }); } catch (error) { - if (terminalObservation) { terminalObservation.assessedAtMs = Date.now(); terminalObservation.censored = true; - active.terminalOriginal = originalTerminal(streamingObservation!, terminalObservation); } + if (h5) active.terminalOriginal = originalTerminal(streamingObservation!, active.terminalObservation); throw error; } - const settledAt = Date.now(), seenMarkers = await observed(); - if (terminalObservation) { terminalObservation.assessedAtMs = settledAt; terminalObservation.complete = true; } + const terminalObservation = settlement.outcome; + runs = settlement.runs; + const settledAt = terminalObservation.assessedAtMs, seenMarkers = await observed(); active.settledAt = settledAt; active.seenMarkers = seenMarkers; const details = []; for (const run of runs) details.push(await json(await request.get(`/api/heartbeat-runs/${run.id}`))); @@ -651,8 +661,9 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy ...(h5 ? { h5: { streaming: streamingObservation, terminalObservation, terminalOriginal: originalTerminal(streamingObservation!, terminalObservation!), visibleCountByOriginalPaintDeadline: seenMarkers.filter(marker => marker.visibleMs <= streamingObservation!.original.deadlineMs).length, - deadlineCountScope: "Original poll last count/time retained separately; paint-deadline count derives from unchanged2RAF timestamps" } } : {}), - ...(h5 ? { terminalStatuses: runs.map(run => ({ runId: run.id, runtimeMode: run.runtimeMode, status: run.status })) } : {}), + deadlineCountScope: "Original poll last count/time retained separately; paint-deadline count derives from unchanged2RAF timestamps" } } : { terminalObservation }), + terminalStatuses: runs.map(run => ({ runId: run.id, runtimeMode: run.runtimeMode, status: run.status, + nativePhase: run.nativePhase ?? null, errorCode: run.errorCode ?? null })), setupActivationConcurrency, setupReadinessStartedAtMs, readinessDeadlineMs, allReadyObservedAtMs, activations, activationWaves, nonce, markerCode, epochMs, settledAtMs: settledAt, stopRequestedAtMs: stopRequestedAt, stopAcknowledgedAtMs: acknowledgedAt, preludeProof, controllerAuthority, clockEvidence, interChunkGaps, @@ -689,7 +700,7 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy if (!rows.length) break; for (const row of rows) { const event = row.payload?.prpEvent; - if (h5 && (row.eventType === "run.result.accepted" || event?.eventType === "run.result.accepted")) resultAcceptedEvents++; + 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); @@ -712,16 +723,20 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy 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); - if (h5) expect(resultAcceptedEvents).toBe(stop ? 0 : 1); + 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, ...(h5 ? { resultAcceptedEvents } : {}) }); + matchedSourceClocks: deltas.length, ordered: true, resultAcceptedEvents }); } await expect.poll(() => [...state.streams.values()].every(stream => stream.ended || stream.aborted), { timeout: 30_000 }).toBe(true); expect(state.errors).toEqual([]); expect([...state.streams.values()].every(stream => !stream.error)).toBe(true); + const recordedFailures = await runtimeAuthority({ appRoot, companyId: company.id, + environmentId: caseEnvironment.id, issueIds, transport: caseTransport, recordedFailureRunIds: runs.map(run => run.id) }); + active.recordedFailures = recordedFailures; + assertRecordedFailures(recordedFailures, { ...result, durable, runIds: runs.map(run => run.id) }, company.id); if (h5) { const verifiedAtMs = Date.now(); - Object.assign(result.h5!, { independentDrain: { verifiedAtMs, elapsedFromEpochMs: verifiedAtMs - epochMs, + Object.assign((result as any).h5, { independentDrain: { verifiedAtMs, elapsedFromEpochMs: verifiedAtMs - epochMs, deltaCount: durable.reduce((count, row) => count + row.individualDeltas, 0), resultAcceptedCount: durable.reduce((count, row) => count + (row.resultAcceptedEvents ?? 0), 0), clockStreamsClosed: true, @@ -732,6 +747,7 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy hello: stream.hello, ended: stream.ended, aborted: stream.aborted, error: stream.error, partialBytes: stream.partialBytes, providerReadyAtMs: stream.providerReadyAtMs, clockHelloAtMs: stream.clockHelloAtMs, readinessObservedAtMs: stream.readinessObservedAtMs, emissions: [...stream.records.values()].sort((a, b) => a.index - b.index) }))); + Object.assign(result, { recordedFailures }); (report.cases as unknown[]).push({ ...result, durable, persistedAfterReload: true }); await info.attach(`${workload.name}${stop ? "-stop" : ""}-metrics`, { body: JSON.stringify(result, null, 2), contentType: "application/json" }); pending.delete(key); clocks.delete(key); delete report.activeCase;