Observe setup readiness without adaptive polling backoff

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`.

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");

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/],

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<number, Emission>; error?: string };
+  partialBytes: number; records: Map<number, Emission>; error?: string;
+  providerReadyAtMs?: number; clockHelloAtMs?: number; readinessObservedAtMs?: number };
 type ClockCase = { events: number; intervalMs: number; epochMs?: number; streams: Map<string, ClockStream>; 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<string, unknown> = {
     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<string, unknown> | 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" });