Qualify provider admission before timed ingestion workloads

doc/plans/2026-11-03-native-event-ingestion.md

@@ -349,3 +349,44 @@ samples are never reused as v2 evidence.
 Registered v2 browser protocol SHA256: `5498ed8dfc7139285aeda582c540eb572f07a1ad3e73de6e0f12c5af4c45e0ba`.
 Comparator SHA256: `d8cf6f55339ad811ca1743f4bc73b74afa3a321b93e17ef913fc0339b47b4d7d`. Individual source hashes and
 H's exact committed revision are retained with each qualification/execution receipt.
+
+### V2 admission qualification correction, before fresh A/A
+
+H `96b1a5f1a4497d1389c21b0cc52d55b86641690d` passed the compiled
+single-run smoke and completed paced 128, burst 16, and ten runs of 128 events.
+Its first full baseline qualification then failed during setup for the 100-run
+case, before the common emission epoch was released. All 100 issues existed,
+but only one provider/clock hello arrived and no measured deltas were emitted.
+Ten concurrent normal task-activation requests timed out; cleanup also timed
+out and masked the original exception. The complete failed report and an
+independent SELECT-only database inspection remain excluded from acceptance.
+The active issues were 50 todo, 49 backlog, and 1 in_progress; 31 runs existed
+(3 running, 28 queued), with no measured item.delta rows.
+
+Static inspection found a separate potential pool-starvation path: each native
+wakeup transaction can await a responsible-user lookup on the outer database
+client, whose default pool is 10; a group of ten activation wakeups can consume
+that pool. This is an observed pre-epoch startup-contention failure with a
+matching static candidate, not a runtime-proven deadlock diagnosis: no live
+pg_stat_activity capture exists. The product admission hazard remains separate
+from this ingestion change; increasing pool size or request timeouts is not
+part of the experiment.
+
+Before any fresh A/A, H now schedules each background using the same normal
+PATCH, then requires that exact participant's provider GET and original-clock
+hello before scheduling the next. All 100 providers remain live behind the
+closed common emission gate. A **shared 180-second readiness budget starts
+before activation**; this is stricter than starting the same readiness budget
+after activation. The 600-second overall test limit, 20-second request limits,
+event counts, payloads, rates, 100-run measured concurrency and every performance
+gate are unchanged. Source owner-path protocol and gates are byte-identical.
+The report identifies this admission protocol explicitly, and its comparator
+rejects reports without it. Original exceptions, exact activation progress and
+separate cleanup errors are retained without accepting a cleanup failure.
+
+Registered corrected browser protocol SHA256:
+`b955ff1f84d82655e5e1d331b2494cb811f459f44aee0e29205a70636c07dc84`.
+Comparator SHA256:
+`55eab8950528a6ec6a30728fb7614b5a9fb7aa210526cb8b7d3237a82db65672`.
+The corrected H commit and source hashes are recorded before its full baseline
+qualification and new browser A/A. Earlier H96b samples never enter that series.

scripts/compare-native-ingestion-browser.ts

@@ -23,7 +23,10 @@ const first = pairs[0];
 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.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");
   assert.notEqual(report.diagnosticOnly, true, "Diagnostic profiles cannot qualify unprofiled performance");
   assert.notEqual(report.performanceAcceptanceEligible, false, "Profile-only evidence cannot qualify performance");
   assert.equal(report.browserRecording?.trace, "off", "Acceptance requires browser trace capture off");

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", complete: true, 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", 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,
@@ -123,6 +123,9 @@ test("browser CLI rejects the three-pairs-above-noise counterexample and ineligi
       ["profile-ineligible", report => { report.performanceAcceptanceEligible = false; }, /Profile-only evidence/],
       ["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/],
+      ["unfinished-cleanup", report => { report.cleanup.complete = false; }, /completed fixture cleanup/],
+      ["cleanup-error", report => { report.cleanup.errors.push({ step: "restore", message: "timeout" }); }, /cleanup errors/],
       ["unmatched-visible-clock", report => { report.cases[0].matchedVisibleClocks--; }, /original provider clock/],
       ["unmatched-durable-clock", report => { report.cases[0].durable[0].matchedSourceClocks--; }, /original provider clock/],
       ["browser-trace", report => { report.browserRecording.trace = "on"; }, /trace capture off/],

tests/e2e/native-event-ingestion.journey.ts

@@ -146,6 +146,7 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy
     screencast: captureMode(info.project.use.trace) !== "off" || captureMode(info.project.use.video) !== "off" };
   const report: Record<string, unknown> = {
     protocolVersion: "v2-compact-markers-original-clock-stream",
+    admissionProtocol: "sequential-provider-and-clock-readiness-before-common-epoch",
     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),
@@ -245,15 +246,24 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy
         issueIds.push(background.id);
         issuedTokens.set(background.id, backgroundToken);
       }
-      for (let offset = 1; offset < issueIds.length; offset += 10) {
-        await Promise.all(issueIds.slice(offset, offset + 10).map(async id => json(await request.patch(`/api/issues/${id}`, { data: { status: "todo" } }))));
+      const readinessDeadlineMs = Date.now() + 180_000;
+      active.phase = "activation-and-provider-readiness";
+      for (let index = 1; index < issueIds.length; index++) {
+        const id = issueIds[index], token = issuedTokens.get(id)!;
+        active.activation = { index, issueId: id, token, startedAtMs: Date.now(), readinessDeadlineMs };
+        await json(await request.patch(`/api/issues/${id}`, { data: { status: "todo" } }));
+        // 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);
       }
       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: 180_000 }).toEqual({ providers: workload.runs, clocks: workload.runs });
+      { timeout: Math.max(1, readinessDeadlineMs - Date.now()) }).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) {
         await expect(page.getByText(`PRELUDE_${nonce}_READY`, { exact: false }).first()).toBeVisible({ timeout: 120_000 });
@@ -389,6 +399,10 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy
     for (const workload of cases) await perform(workload);
     await perform({ name: "operator-stop", runs: 1, events: 512, intervalMs: 100, prelude: 0 }, true);
     report.complete = true;
+  } catch (error) {
+    report.primaryFailure = { name: error instanceof Error ? error.name : "UnknownError",
+      message: error instanceof Error ? error.message : String(error), stack: error instanceof Error ? error.stack : undefined };
+    throw error;
   } finally {
     if (report.activeCase) {
       const active = report.activeCase as Record<string, unknown>;
@@ -407,13 +421,23 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy
         ...stream, records: [...stream.records.values()],
       })) };
     }
-    await info.attach("native-ingestion-evidence", { body: JSON.stringify(report, null, 2), contentType: "application/json" });
+    // Retain primary failure evidence even if API teardown cannot finish. This
+    // attachment is never eligible for the standalone acceptance comparator.
+    await info.attach("native-ingestion-precleanup-evidence", { body: JSON.stringify(report, null, 2), contentType: "application/json" });
     for (const requests of pending.values()) for (const response of requests.values()) response.writeHead(503).end();
     gate.closeAllConnections();
     await new Promise<void>(resolve => gate.close(() => resolve()));
-    const live = await json(await request.get(`/api/companies/${company.id}/live-runs`)).catch(() => []);
-    for (const run of live) await request.post(`/api/heartbeat-runs/${run.id}/cancel`);
-    await request.patch("/api/instance/settings/experimental", { data: { enableNativeRunner: original.enableNativeRunner,
-      enableWorktreeRunExecution: original.enableWorktreeRunExecution } });
+    const cleanupErrors: Array<{ step: string; message: string }> = [];
+    const attemptCleanup = async (step: string, action: () => Promise<unknown>) => {
+      try { return await action(); } catch (error) { cleanupErrors.push({ step, message: String(error) }); }
+    };
+    const live = await attemptCleanup("list-live-runs", async () => json(await request.get(`/api/companies/${company.id}/live-runs`))) as any[] | undefined;
+    for (const run of live ?? []) await attemptCleanup(`cancel-${run.id}`, async () => json(await request.post(`/api/heartbeat-runs/${run.id}/cancel`)));
+    await attemptCleanup("restore-experimental-settings", async () => json(await request.patch("/api/instance/settings/experimental", {
+      data: { enableNativeRunner: original.enableNativeRunner, enableWorktreeRunExecution: original.enableWorktreeRunExecution } })));
+    report.cleanup = { errors: cleanupErrors, complete: cleanupErrors.length === 0 };
+    if (cleanupErrors.length) { report.complete = false; report.performanceAcceptanceEligible = false; }
+    await info.attach("native-ingestion-evidence", { body: JSON.stringify(report, null, 2), contentType: "application/json" });
+    if (!report.primaryFailure) expect(cleanupErrors, "API cleanup must succeed before accepting an invocation").toEqual([]);
   }
 });