diff --git a/doc/plans/2026-11-03-native-event-ingestion.md b/doc/plans/2026-11-03-native-event-ingestion.md index c829f49b1cc..962656bce07 100644 --- a/doc/plans/2026-11-03-native-event-ingestion.md +++ b/doc/plans/2026-11-03-native-event-ingestion.md @@ -401,3 +401,32 @@ five-pair A/A noise envelope. This adds the previously missing acceptance check before browser A/A; it changes neither the streaming observer nor emitted work. The independent source-owner A/A may run during local H work because its workload, measurement code and gates remain unchanged. + +### Setup readiness observation correction, before browser A/A + +The full H `e27b338315034c062b18606c0f3ff7d883e3dfed` qualification +reached activation index 95, with 95 provider GETs and clock hellos, then hit the +unchanged shared 180-second deadline. No common epoch was released and no +measured emission clocks existed. The retained report records the primary +984-millisecond remaining-budget timeout and a separate cleanup 409 response; +it is ineligible for acceptance. After teardown, an independent process check +verified no owned Node, PostgreSQL or Rust Runner processes remained before the +separate source-owner cursor comparison began. + +The default readiness poll backoff is a suspected contributor to the approximately +1.85-second setup cadence. That failed run did not record clock-hello receipt or +readiness-observation times, so it does not prove poll quantization. Before fresh +browser A/A, H changes only the two setup predicates to a fixed 50-millisecond +observation interval. Both predicates read local provider/clock maps; they issue +no database or application request. The shared 180-second hard deadline, 600-second +overall limit, exact-participant checks, 100 live providers, common epoch, measured +event schedules, counts, payloads and performance gates remain unchanged. + +Future qualification records pre-epoch provider GET receipt, clock-hello receipt +and background readiness-observation timestamps to expose any observation lag. +These are setup diagnostics, distinct from the unchanged original provider +emission clocks and browser paint observer. The comparator requires the frozen +50-millisecond setup interval. Earlier failed samples remain excluded. + +Registered browser protocol after setup observation correction: `902c6a208793f0bcde1b326a1fee46d85df3274117754e3bfbbf77a65dc730ee`. +Comparator SHA256: `e50d958532906d037413db6f66bdc02e60c93ddcacfe2e18e5b731d302f6a785`. diff --git a/scripts/compare-native-ingestion-browser.ts b/scripts/compare-native-ingestion-browser.ts index c22c88b3c65..9e87365fe77 100644 --- a/scripts/compare-native-ingestion-browser.ts +++ b/scripts/compare-native-ingestion-browser.ts @@ -24,6 +24,7 @@ for (const pair of pairs) for (const arm of ["baseline", "candidate"] as const) const report = pair[arm]; assert.equal(report.protocolVersion, "v2-compact-markers-original-clock-stream", "Fresh browser acceptance requires the qualified v2 clock protocol"); assert.equal(report.admissionProtocol, "sequential-provider-and-clock-readiness-before-common-epoch", "Use the qualified pre-measurement admission protocol"); + assert.equal(report.setupReadinessPollingMs, 50, "Keep the qualified setup readiness observation interval"); assert.equal(report.complete, true); assert.equal(report.qualificationOnly, false, "Smoke cannot qualify performance"); assert.equal(report.cleanup?.complete, true, "Acceptance requires completed fixture cleanup"); assert.deepEqual(report.cleanup?.errors, [], "Acceptance rejects fixture cleanup errors"); diff --git a/scripts/native-ingestion-performance-gates.test.mjs b/scripts/native-ingestion-performance-gates.test.mjs index 5ecb86f0d3c..e5aedffb42d 100644 --- a/scripts/native-ingestion-performance-gates.test.mjs +++ b/scripts/native-ingestion-performance-gates.test.mjs @@ -75,7 +75,7 @@ const workloads = [ { name: "operator-stop", runs: 1, events: 512, intervalMs: 100, prelude: 0, stop: true }, ]; function browserReport(revision, latency) { - return { protocolVersion: "v2-compact-markers-original-clock-stream", admissionProtocol: "sequential-provider-and-clock-readiness-before-common-epoch", complete: true, cleanup: { complete: true, errors: [] }, qualificationOnly: false, diagnosticOnly: false, performanceAcceptanceEligible: true, browserRecording: { trace: "off", video: "off", screencast: false }, + return { protocolVersion: "v2-compact-markers-original-clock-stream", admissionProtocol: "sequential-provider-and-clock-readiness-before-common-epoch", setupReadinessPollingMs: 50, complete: true, cleanup: { complete: true, errors: [] }, qualificationOnly: false, diagnosticOnly: false, performanceAcceptanceEligible: true, browserRecording: { trace: "off", video: "off", screencast: false }, sourceRevision: revision, compiledBuildRevision: revision, serverArtifact: "compiled", serverEntrySha256: `${revision}-compiled`, measurementSha256: "synthetic-fixed-fixture", compiledHotPathHashes: { append: "append", port: "port", runtime: "runtime" }, cases: workloads.map(workload => ({ ...workload, stop: Boolean(workload.stop), persistedAfterReload: true, @@ -125,6 +125,7 @@ test("browser CLI rejects the three-pairs-above-noise counterexample and ineligi ["missing-retained-prelude", report => { report.cases.find(row => row.prelude).preludeProof.individualDeltas--; }, /stored prelude deltas/], ["legacy-protocol", report => { delete report.protocolVersion; }, /v2 clock protocol/], ["unqualified-admission", report => { delete report.admissionProtocol; }, /admission protocol/], + ["unqualified-readiness-observation", report => { delete report.setupReadinessPollingMs; }, /readiness observation interval/], ["unfinished-cleanup", report => { report.cleanup.complete = false; }, /completed fixture cleanup/], ["cleanup-error", report => { report.cleanup.errors.push({ step: "restore", message: "timeout" }); }, /cleanup errors/], ["missing-paced-gaps", report => { delete report.cases[0].interChunkGaps; }, /inter-chunk paint gaps/], diff --git a/tests/e2e/native-event-ingestion.journey.ts b/tests/e2e/native-event-ingestion.journey.ts index 69d04dd28ae..cb1ca8960ef 100644 --- a/tests/e2e/native-event-ingestion.journey.ts +++ b/tests/e2e/native-event-ingestion.journey.ts @@ -10,7 +10,8 @@ type SeenMarker = { markerCode: string; index: number; visibleMs: number }; type Emission = { nonce: string; markerCode: string; index: number; intendedMs: number; emittedMs: number; receivedAtMs: number }; type Visible = SeenMarker & Emission; type ClockStream = { nonce: string; markerCode: string; opened: boolean; hello: boolean; ended: boolean; aborted: boolean; - partialBytes: number; records: Map; error?: string }; + partialBytes: number; records: Map; error?: string; + providerReadyAtMs?: number; clockHelloAtMs?: number; readinessObservedAtMs?: number }; type ClockCase = { events: number; intervalMs: number; epochMs?: number; streams: Map; errors: string[] }; type Case = { name: string; runs: number; events: number; intervalMs: number; prelude: number }; const smoke = process.env.PAPERCLIP_INGESTION_SMOKE === "1"; @@ -44,6 +45,7 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy if (!key || !participant || !state || !stream) { res.writeHead(404).end(); return; } if (!route?.[2] && req.method === "GET") { if (pending.get(key)!.has(participant)) { res.writeHead(409).end(); return; } + stream.providerReadyAtMs = Date.now(); pending.get(key)!.set(participant, res); res.on("close", () => pending.get(key)?.delete(participant)); return; @@ -61,7 +63,7 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy if (value.nonce !== participant || value.markerCode !== stream.markerCode) throw new Error("Clock identity mismatch"); if (!stream.hello) { if (value.type !== "hello" || value.events !== state.events || value.intervalMs !== state.intervalMs) throw new Error("Clock hello mismatch"); - stream.hello = true; return; + stream.hello = true; stream.clockHelloAtMs = Date.now(); return; } if (value.type !== "emission" || !Number.isSafeInteger(value.index) || value.index < 0 || value.index >= state.events || !Number.isSafeInteger(value.intendedMs) || !Number.isSafeInteger(value.emittedMs) || value.emittedMs <= 0 @@ -147,6 +149,7 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy const report: Record = { protocolVersion: "v2-compact-markers-original-clock-stream", admissionProtocol: "sequential-provider-and-clock-readiness-before-common-epoch", + setupReadinessPollingMs: 50, dependency: "Deterministic Codex app-server fixture; real application/runtime/port/PostgreSQL/live WebSocket/shipped UI", variant: process.env.PAPERCLIP_INGESTION_VARIANT ?? "unlabelled", qualificationOnly: smoke, diagnosticOnly: Boolean(process.env.PAPERCLIP_NATIVE_INGESTION_DIAGNOSTIC_DIR), @@ -255,14 +258,15 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy // PATCH schedules wakeup asynchronously. Wait for this actual provider, // while the common emission gate remains closed, before scheduling more. await expect.poll(() => pending.get(key)!.has(token) && state.streams.get(token)!.hello, - { timeout: Math.max(1, readinessDeadlineMs - Date.now()) }).toBe(true); + { timeout: Math.max(1, readinessDeadlineMs - Date.now()), intervals: [50] }).toBe(true); + state.streams.get(token)!.readinessObservedAtMs = Date.now(); } const route = `/${company.issuePrefix}/issues/${issue.identifier}`; await page.goto(route); active.phase = "provider-and-clock-readiness"; await expect.poll(() => ({ providers: pending.get(key)!.size, clocks: [...state.streams.values()].filter(stream => stream.hello).length }), - { timeout: Math.max(1, readinessDeadlineMs - Date.now()) }).toEqual({ providers: workload.runs, clocks: workload.runs }); + { timeout: Math.max(1, readinessDeadlineMs - Date.now()), intervals: [50] }).toEqual({ providers: workload.runs, clocks: workload.runs }); expect(Date.now(), "All participants must be ready within the shared activation/readiness budget").toBeLessThanOrEqual(readinessDeadlineMs); let preludeProof: Record | undefined; if (workload.prelude) { @@ -396,6 +400,7 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy expect([...state.streams.values()].every(stream => !stream.error)).toBe(true); clockEvidence.push(...[...state.streams.values()].map(stream => ({ nonce: stream.nonce, markerCode: stream.markerCode, hello: stream.hello, ended: stream.ended, aborted: stream.aborted, error: stream.error, partialBytes: stream.partialBytes, + providerReadyAtMs: stream.providerReadyAtMs, clockHelloAtMs: stream.clockHelloAtMs, readinessObservedAtMs: stream.readinessObservedAtMs, emissions: [...stream.records.values()].sort((a, b) => a.index - b.index) }))); (report.cases as unknown[]).push({ ...result, durable, persistedAfterReload: true }); await info.attach(`${workload.name}${stop ? "-stop" : ""}-metrics`, { body: JSON.stringify(result, null, 2), contentType: "application/json" });