Bound ingestion fixture activation to two-provider waves

scripts/compare-native-ingestion-browser.ts

@@ -79,7 +79,8 @@ for (const pair of pairs) for (const arm of ["baseline", "candidate"] as const)
   for (const [group, field] of [["runner", "sha256"], ["provider", "nodeSha256"], ["provider", "wrapperSha256"], ["provider", "fixtureSha256"]]) {
     assert.equal(remote[group][field], first.baseline.remote[group][field], "Remote provider/daemon artifact changed across matched samples");
   }
-  assert.equal(report.admissionProtocol, "sequential-provider-and-clock-readiness-before-common-epoch", "Use the qualified pre-measurement admission protocol");
+  assert.equal(report.admissionProtocol, "two-at-a-time-provider-and-clock-readiness-before-common-epoch", "Use the qualified pre-measurement admission protocol");
+  assert.equal(report.setupActivationConcurrency, 2, "Keep the qualified setup activation concurrency of two");
   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");
@@ -105,6 +106,7 @@ for (const pair of pairs) for (const arm of ["baseline", "candidate"] as const)
     for (const field of ["runs", "events", "intervalMs", "prelude", "stop"] as const) assert.equal(result[field], fixed[field], `Fixed workload changed: ${name}.${field}`);
     assert.equal(result.transport, result.stop ? "local" : "sandbox-remote", "Wrong per-case transport; fake-plugin does not qualify remote Stop");
     assert.equal(result.environmentId, result.stop ? local.environmentId : remote.environmentId, "Wrong per-case normal environment");
+    assert.equal(result.setupActivationConcurrency, 2, "Keep each case's qualified setup activation concurrency of two");
     assert.equal(result.persistedAfterReload, true);
     assert.equal(result.matchedVisibleClocks, result.visible?.length, "Every observed marker needs its original provider clock");
     assert.ok(result.matchedVisibleClocks >= (result.stop ? 3 : result.events), "Missing foreground render/clock observations");
@@ -171,6 +173,55 @@ for (const pair of pairs) for (const arm of ["baseline", "candidate"] as const)
       assert.equal(stream?.providerPid, clock.providerPid, "Clock hello provider PID proof missing");
       assert.equal(result.durable.find((proof: any) => proof.runId === clock.runId)?.token, clock.nonce);
     }
+    assert.equal(result.participants, result.runs, "Every provider GET must remain open before the common epoch");
+    assert.ok(Number.isSafeInteger(result.setupReadinessStartedAtMs) && result.setupReadinessStartedAtMs > 0);
+    assert.equal(result.readinessDeadlineMs, result.setupReadinessStartedAtMs + report.resourceBudgets.providerReadinessMs,
+      "Keep one unchanged shared readiness deadline");
+    assert.ok(Number.isSafeInteger(result.allReadyObservedAtMs) && result.allReadyObservedAtMs >= result.setupReadinessStartedAtMs
+      && result.allReadyObservedAtMs <= result.readinessDeadlineMs && result.allReadyObservedAtMs < result.epochMs,
+    "All actual providers and clocks must be ready before the epoch and shared deadline");
+    assert.ok(authority.checkedAtMs >= result.allReadyObservedAtMs, "Check actual controller authority after every provider is ready");
+    assert.equal(result.activations?.length, result.runs, "Every actual participant needs its setup activation timing");
+    assert.deepEqual(result.activations.map((entry: any) => entry.index).sort((a: number, b: number) => a - b),
+      Array.from({ length: result.runs }, (_, index) => index), "Setup activation indices must cover every participant once");
+    const waves = result.activationWaves;
+    assert.equal(waves?.length, Math.ceil((result.runs - 1) / 2), "Background setup must use bounded waves of two");
+    for (let waveIndex = 0; waveIndex < waves.length; waveIndex++) {
+      const wave = waves[waveIndex], firstIndex = 1 + waveIndex * 2;
+      const indices = Array.from({ length: Math.min(2, result.runs - firstIndex) }, (_, offset) => firstIndex + offset);
+      assert.equal(wave.index, waveIndex); assert.deepEqual(wave.indices, indices, "Wrong setup wave participants");
+      assert.deepEqual(wave.statuses, indices.map(() => "fulfilled"), "A failed or unfinished setup wave cannot qualify");
+      assert.ok(Number.isSafeInteger(wave.startedAtMs) && Number.isSafeInteger(wave.settledAtMs)
+        && wave.startedAtMs >= (waveIndex ? waves[waveIndex - 1].settledAtMs : result.setupReadinessStartedAtMs)
+        && wave.settledAtMs >= wave.startedAtMs && wave.settledAtMs <= result.allReadyObservedAtMs,
+      "Finish the whole setup wave before starting the next one");
+    }
+    for (const activation of result.activations) {
+      const row = authority.rows.find((row: any) => row.issue_id === activation.issueId);
+      assert.ok(row, "Setup activation task must match actual controller authority");
+      const clock = result.providerClocks.find((clock: any) => clock.runId === row.run_id);
+      assert.equal(activation.token, clock.nonce, "Setup activation must match its actual provider clock");
+      assert.equal(activation.status, "fulfilled", "An unfinished setup activation cannot qualify");
+      for (const field of ["startedAtMs", "requestSettledAtMs", "providerReadyAtMs", "clockHelloAtMs", "readinessObservedAtMs", "settledAtMs"]) {
+        assert.ok(Number.isSafeInteger(activation[field]) && activation[field] > 0 && activation[field] < result.epochMs,
+          "Missing actual pre-epoch activation/readiness timing");
+      }
+      assert.ok(activation.requestSettledAtMs >= activation.startedAtMs && activation.providerReadyAtMs >= activation.startedAtMs
+        && activation.clockHelloAtMs >= activation.startedAtMs
+        && activation.readinessObservedAtMs >= Math.max(activation.requestSettledAtMs, activation.providerReadyAtMs, activation.clockHelloAtMs)
+        && activation.settledAtMs >= activation.readinessObservedAtMs && activation.settledAtMs <= result.allReadyObservedAtMs,
+      "Actual provider GET and hello must precede observed readiness");
+      const stream = result.clockEvidence.find((stream: any) => stream.nonce === activation.token);
+      assert.equal(stream.hello, true, "Every actual provider needs its clock hello");
+      for (const field of ["providerReadyAtMs", "clockHelloAtMs", "readinessObservedAtMs"]) assert.equal(stream[field], activation[field], "Readiness metadata must match actual clock observations");
+      if (activation.index === 0) assert.equal(activation.waveIndex, null);
+      else {
+        const waveIndex = Math.floor((activation.index - 1) / 2), wave = waves[waveIndex];
+        assert.equal(activation.waveIndex, waveIndex);
+        assert.ok(activation.startedAtMs >= wave.startedAtMs && activation.settledAtMs <= wave.settledAtMs,
+          "Keep each activation inside its settled setup wave");
+      }
+    }
     assert.equal(result.settledAtMs, result.epochMs + result.drainThroughTerminalMs, "Terminal timing boundary changed");
     if (result.stop) {
       assert.equal(result.stopAcknowledgedAtMs, result.stopRequestedAtMs + result.stopThroughAcknowledgementMs, "Stop acknowledgement timing boundary changed");
@@ -274,6 +325,7 @@ const measurements = names.flatMap(name => {
 const result: Record<string, any> = {
   scope: "Compiled-app remote streaming and original local Stop performance; additional real cloud Stop qualification is required separately",
   transportPolicy: first.baseline.transportPolicy, additionalRealCloudStopQualificationRequired: true,
+  admissionProtocol: first.baseline.admissionProtocol, setupActivationConcurrency: first.baseline.setupActivationConcurrency,
   protocolVersion: first.baseline.protocolVersion, transport: first.baseline.transport,
   resourceBudgets: first.baseline.resourceBudgets,
   phase, expectation, pairCount: pairs.length, baselineRevision: first.baseline.sourceRevision,
@@ -290,6 +342,8 @@ if (phase === "compare") {
   assert.equal(noise.protocolVersion, result.protocolVersion, "Noise report uses a different mixed protocol");
   assert.equal(noise.transport, result.transport, "Noise report uses a different transport");
   assert.deepEqual(noise.transportPolicy, result.transportPolicy, "Transport policy changed after A/A calibration");
+  assert.equal(noise.admissionProtocol, result.admissionProtocol, "Admission protocol changed after A/A calibration");
+  assert.equal(noise.setupActivationConcurrency, result.setupActivationConcurrency, "Setup activation concurrency changed after A/A calibration");
   assert.deepEqual(noise.resourceBudgets, result.resourceBudgets, "Resource budgets changed after A/A calibration");
   assert.equal(noise.baselineRevision, result.baselineRevision); assert.equal(noise.measurementSha256, result.measurementSha256);
   assert.equal(noise.gatesSha256, result.gatesSha256, "Acceptance gates changed after A/A calibration");

scripts/native-ingestion-performance-gates.test.mjs

@@ -94,7 +94,7 @@ function browserReport(revision, latency) {
       tls: { peerVerification: true, caSha256: "e".repeat(64), serverCertificateSha256: "f".repeat(64) },
       runner: { sha256: "d".repeat(64), metadata: { prpTransportModes: ["dial_wss"] } },
       provider: { nodeSha256: "b".repeat(64), wrapperSha256: "c".repeat(64), fixtureSha256: "a".repeat(64) } },
-    fixtureSha256: "a".repeat(64), 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 },
+    fixtureSha256: "a".repeat(64), admissionProtocol: "two-at-a-time-provider-and-clock-readiness-before-common-epoch", setupActivationConcurrency: 2, 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, caseIndex) => {
@@ -118,7 +118,21 @@ function browserReport(revision, latency) {
         provider: "local", providerLeaseId: null, status: "released", releasedAt: "2026-11-03T00:00:01.000Z", cleanupStatus: null };
       const localProcesses = { runnerPid: pid - 1, processGroupId: pid - 1, providerPid: pid, hostname: "fixture-host",
         checkedAtMs: 999, runnerAlive: true, groupAlive: true, providerAlive: true };
+      const activations = rows.map((row, index) => {
+        const waveIndex = index ? Math.floor((index - 1) / 2) : null, startedAtMs = index ? 110 + waveIndex * 10 : 50;
+        return { index, issueId: row.issue_id, token: String(index), waveIndex, startedAtMs,
+          requestSettledAtMs: startedAtMs + (index ? 3 : 5), providerReadyAtMs: startedAtMs + (index ? 1 : 10),
+          clockHelloAtMs: startedAtMs + (index ? 2 : 20), readinessObservedAtMs: index ? startedAtMs + 6 : 700,
+          settledAtMs: index ? startedAtMs + 7 : 700, status: "fulfilled" };
+      });
+      const activationWaves = Array.from({ length: Math.ceil((workload.runs - 1) / 2) }, (_, index) => ({
+        index, indices: Array.from({ length: Math.min(2, workload.runs - 1 - index * 2) }, (_, offset) => 1 + index * 2 + offset),
+        startedAtMs: 110 + index * 10, settledAtMs: 118 + index * 10,
+        statuses: Array(Math.min(2, workload.runs - 1 - index * 2)).fill("fulfilled"),
+      }));
       return { ...workload, stop, transport: stop ? "local" : "sandbox-remote", environmentId: actualEnvironmentId, persistedAfterReload: true,
+        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,
         settledAtMs: 1000 + latency, stopRequestedAtMs: stop ? 1000 : undefined,
         stopAcknowledgedAtMs: stop ? 1000 + latency : undefined, remoteLeasesCheckedAtMs: stop ? undefined : 1001 + latency,
@@ -126,7 +140,9 @@ function browserReport(revision, latency) {
           ...(stop ? { localJournal: { schema: "paperclip.runner.durable.control-plane-state.v1", sha256: "8".repeat(64), bytes: 512, checkedAtMs: 999,
             identity: { runId: rows[0].run_id, runnerInstanceId: rows[0].runner_instance_id, normalizedSessionId: rows[0].native_session_id, environmentLeaseId: rows[0].admitted_execution_workspace_id } } } : {}) },
         providerClocks: rows.map((row, index) => ({ runId: row.run_id, nonce: String(index), launcher, providerPid: pid + index })),
-        clockEvidence: rows.map((row, index) => ({ nonce: String(index), launcher, providerPid: pid + index })),
+        clockEvidence: rows.map((row, index) => ({ nonce: String(index), launcher, providerPid: pid + index,
+          hello: true, providerReadyAtMs: activations[index].providerReadyAtMs, clockHelloAtMs: activations[index].clockHelloAtMs,
+          readinessObservedAtMs: activations[index].readinessObservedAtMs })),
         ...(stop ? { localProcesses, localStopProof: { checkedAtMs: 1001 + latency, lease,
           processes: { ...localProcesses, runnerAbsent: true, groupAbsent: true, providerAbsent: true },
           database: { ...snapshot, checkedBeforeMeasuredEpoch: false, checkedAtMs: 1000 + latency, connectionClosedAtMs: 1000 + latency,
@@ -250,6 +266,24 @@ test("browser CLI rejects the three-pairs-above-noise counterexample and ineligi
       ["lease-check-before-timing", report => { report.cases[3].remoteLeasesCheckedAtMs = report.cases[3].settledAtMs - 1; }, /after recorded timing boundaries/],
       ["stop-timing-restamped", report => { report.cases[5].stopAcknowledgedAtMs++; }, /timing boundary changed/],
       ["unqualified-admission", report => { delete report.admissionProtocol; }, /admission protocol/],
+      ["old-sequential-admission", report => { report.admissionProtocol = "sequential-provider-and-clock-readiness-before-common-epoch"; }, /admission protocol/],
+      ["missing-activation-concurrency", report => { delete report.setupActivationConcurrency; }, /activation concurrency of two/],
+      ["excessive-activation-concurrency", report => { report.setupActivationConcurrency = 10; }, /activation concurrency of two/],
+      ["serial-activation-concurrency", report => { report.setupActivationConcurrency = 1; }, /activation concurrency of two/],
+      ["wrong-case-activation-concurrency", report => { report.cases[3].setupActivationConcurrency = 1; }, /case's qualified setup activation/],
+      ["extended-shared-readiness-deadline", report => { report.cases[3].readinessDeadlineMs++; }, /unchanged shared readiness deadline/],
+      ["authority-before-all-ready", report => { report.cases[3].controllerAuthority.checkedAtMs = 699; }, /authority after every provider/],
+      ["missing-live-provider-get", report => { report.cases[3].participants--; }, /provider GET must remain open/],
+      ["missing-activation-record", report => { report.cases[3].activations.pop(); }, /activation timing/],
+      ["wrong-activation-task", report => { report.cases[3].activations[99].issueId = "other"; }, /task must match actual controller/],
+      ["wrong-activation-token", report => { report.cases[3].activations[99].token = "other"; }, /actual provider clock/],
+      ["unfinished-activation", report => { report.cases[3].activations[99].status = "pending"; }, /unfinished setup activation/],
+      ["unfinished-wave", report => { report.cases[3].activationWaves[49].statuses = ["rejected"]; }, /failed or unfinished setup wave/],
+      ["overlapping-waves", report => { report.cases[3].activationWaves[1].startedAtMs = 110; }, /whole setup wave before/],
+      ["oversized-wave", report => { report.cases[3].activationWaves[0].indices.push(3); }, /Wrong setup wave participants/],
+      ["late-readiness-observation", report => { report.cases[3].activations[99].readinessObservedAtMs = 1001; }, /pre-epoch activation/],
+      ["missing-provider-clock-hello", report => { report.cases[3].clockEvidence[99].hello = false; }, /actual provider needs its clock hello/],
+      ["restamped-provider-readiness", report => { report.cases[3].clockEvidence[99].readinessObservedAtMs++; }, /actual clock observations/],
       ["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/],
@@ -280,6 +314,7 @@ test("browser CLI rejects the three-pairs-above-noise counterexample and ineligi
       ["local-noise-transport", report => { report.transport = "local"; }, /different transport/],
       ["changed-noise-budget", report => { report.resourceBudgets.providerReadinessMs++; }, /budgets changed after A\/A/],
       ["changed-noise-transport-policy", report => { report.transportPolicy.operatorStop = "sandbox-remote"; }, /Transport policy changed after A\/A/],
+      ["changed-noise-activation-concurrency", report => { report.setupActivationConcurrency = 10; }, /concurrency changed after A\/A/],
     ]) {
       const changed = JSON.parse(await readFile(noise, "utf8")); change(changed);
       const file = await save(name + ".json", changed);

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

@@ -17,10 +17,16 @@ type ClockStream = { nonce: string; markerCode: string; opened: boolean; hello:
   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 };
+type SetupActivation = { index: number; issueId: string; token: string; waveIndex: number | null; startedAtMs: number;
+  requestSettledAtMs?: number; providerReadyAtMs?: number; clockHelloAtMs?: number; readinessObservedAtMs?: number;
+  settledAtMs?: number; status: "pending" | "fulfilled" | "rejected"; failure?: string };
+type SetupWave = { index: number; indices: number[]; startedAtMs: number; settledAtMs?: number;
+  statuses?: Array<"fulfilled" | "rejected"> };
 const smoke = process.env.PAPERCLIP_INGESTION_SMOKE === "1";
 const transport = process.env.PAPERCLIP_INGESTION_TRANSPORT ?? "sandbox-remote-with-local-stop";
 const providerReadinessMs = 300_000;
-const cases: Case[] = smoke ? [{ name: "qualification", runs: 1, events: 16, intervalMs: 25, prelude: 0 }] : [
+const setupActivationConcurrency = 2;
+const cases: Case[] = smoke ? [{ name: "qualification", runs: 3, events: 16, intervalMs: 25, prelude: 0 }] : [
   { name: "paced", runs: 1, events: 128, intervalMs: 25, prelude: 0 },
   { name: "burst16", runs: 1, events: 16, intervalMs: 0, prelude: 0 },
   { name: "burst128", runs: 10, events: 128, intervalMs: 0, prelude: 0 },
@@ -270,7 +276,8 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy
       realCloudStop: "Separate required qualification; fake-plugin void receipts grant no remote Stop authority" },
     resourceBudgets: { providerReadinessMs, fullInvocationMs: info.timeout, streamingMs: 120_000,
       terminalDrainageMs: 180_000, stopAcknowledgementMs: 30_000, remoteLeaseCleanupMs: 30_000 },
-    admissionProtocol: "sequential-provider-and-clock-readiness-before-common-epoch",
+    admissionProtocol: "two-at-a-time-provider-and-clock-readiness-before-common-epoch",
+    setupActivationConcurrency,
     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,
@@ -402,6 +409,7 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy
       await page.getByRole("listbox", { name: "Assignees", exact: true }).getByRole("option")
         .filter({ has: page.getByText(caseAgents[0].name, { exact: true }) }).click();
       await expect(dialog.getByRole("button", { name: "Select assignee" })).toContainText(caseAgents[0].name);
+      const foregroundActivationStartedAtMs = Date.now();
       await dialog.getByRole("button", { name: "Create task", exact: true }).click();
       let issue: any;
       await expect.poll(async () => {
@@ -410,6 +418,11 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy
       }).toBe(true);
       const issueIds = [issue.id];
       active.issueIds = issueIds;
+      const activations: SetupActivation[] = [{ index: 0, issueId: issue.id, token: nonce, waveIndex: null,
+        startedAtMs: foregroundActivationStartedAtMs, requestSettledAtMs: Date.now(), status: "pending" }];
+      const activationWaves: SetupWave[] = [];
+      active.activations = activations; active.activationWaves = activationWaves;
+      active.setupActivationConcurrency = setupActivationConcurrency;
       const issuedTokens = new Map<string, string>([[issue.id, nonce]]);
       for (let index = 1; index < workload.runs; index++) {
         const backgroundToken = randomUUID().replaceAll("-", ""), backgroundCode = nextMarkerCode();
@@ -421,17 +434,38 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy
         issueIds.push(background.id);
         issuedTokens.set(background.id, backgroundToken);
       }
-      const readinessDeadlineMs = Date.now() + providerReadinessMs;
+      const setupReadinessStartedAtMs = Date.now();
+      const readinessDeadlineMs = setupReadinessStartedAtMs + providerReadinessMs;
+      active.setupReadinessStartedAtMs = setupReadinessStartedAtMs; active.readinessDeadlineMs = readinessDeadlineMs;
       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()), intervals: [50] }).toBe(true);
-        state.streams.get(token)!.readinessObservedAtMs = Date.now();
+      for (let firstIndex = 1; firstIndex < issueIds.length; firstIndex += setupActivationConcurrency) {
+        expect(Date.now(), "Do not activate a later wave after the shared readiness deadline").toBeLessThanOrEqual(readinessDeadlineMs);
+        const indices = Array.from({ length: Math.min(setupActivationConcurrency, issueIds.length - firstIndex) }, (_, offset) => firstIndex + offset);
+        const wave: SetupWave = { index: activationWaves.length, indices, startedAtMs: Date.now() };
+        activationWaves.push(wave); active.activationWave = wave;
+        const settled = await Promise.allSettled(indices.map(async index => {
+          const id = issueIds[index], token = issuedTokens.get(id)!;
+          const activation: SetupActivation = { index, issueId: id, token, waveIndex: wave.index,
+            startedAtMs: Date.now(), status: "pending" };
+          activations.push(activation);
+          try {
+            await json(await request.patch(`/api/issues/${id}`, { data: { status: "todo" } }));
+            activation.requestSettledAtMs = Date.now();
+            // Keep the gate closed until both actual providers in this wave have
+            // opened their GET and clock hello; never overlap a later wave.
+            await expect.poll(() => pending.get(key)!.has(token) && state.streams.get(token)!.hello,
+              { timeout: Math.max(1, readinessDeadlineMs - Date.now()), intervals: [50] }).toBe(true);
+            const stream = state.streams.get(token)!;
+            stream.readinessObservedAtMs = Date.now();
+            Object.assign(activation, { providerReadyAtMs: stream.providerReadyAtMs, clockHelloAtMs: stream.clockHelloAtMs,
+              readinessObservedAtMs: stream.readinessObservedAtMs, status: "fulfilled" });
+          } catch (error) {
+            activation.status = "rejected"; activation.failure = String(error); throw error;
+          } finally { activation.settledAtMs = Date.now(); }
+        }));
+        wave.settledAtMs = Date.now(); wave.statuses = settled.map(result => result.status);
+        const failure = settled.find(result => result.status === "rejected");
+        if (failure?.status === "rejected") throw failure.reason;
       }
       const route = `/${company.issuePrefix}/issues/${issue.identifier}`;
       await page.goto(route);
@@ -440,6 +474,11 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy
         clocks: [...state.streams.values()].filter(stream => stream.hello).length }),
       { 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);
+      const allReadyObservedAtMs = Date.now(), foregroundStream = state.streams.get(nonce)!;
+      foregroundStream.readinessObservedAtMs = allReadyObservedAtMs;
+      Object.assign(activations[0], { providerReadyAtMs: foregroundStream.providerReadyAtMs, clockHelloAtMs: foregroundStream.clockHelloAtMs,
+        readinessObservedAtMs: allReadyObservedAtMs, settledAtMs: allReadyObservedAtMs, status: "fulfilled" });
+      active.allReadyObservedAtMs = allReadyObservedAtMs;
       let preludeProof: Record<string, unknown> | undefined;
       if (workload.prelude) {
         await expect(page.getByText(`PRELUDE_${nonce}_READY`, { exact: false }).first()).toBeVisible({ timeout: 120_000 });
@@ -483,9 +522,10 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy
           groupAlive: !processAbsent(-row.process_group_id), providerAlive: !processAbsent(providerPid) };
         expect(localProcesses.runnerAlive && localProcesses.groupAlive && localProcesses.providerAlive).toBe(true);
       }
+      const participants = [...pending.get(key)!.keys()];
+      expect(participants, "Every actual provider GET must still be open after the authority snapshot").toHaveLength(workload.runs);
       const epochMs = Date.now() + 1000;
       state.epochMs = epochMs; active.epochMs = epochMs; active.phase = "streaming";
-      const participants = [...pending.get(key)!.keys()];
       active.participants = participants;
       for (const response of pending.get(key)!.values()) response.writeHead(200, { "Content-Type": "application/json" }).end(JSON.stringify({ epochMs }));
       const observed = () => page.evaluate(code => (window as unknown as { nativeIngestionVisible: SeenMarker[] }).nativeIngestionVisible.filter(item => item.markerCode === code), markerCode);
@@ -578,6 +618,7 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy
       const interChunkGaps = { paintCount: paintTimes.length, count, medianMs, maxMs };
       const clockEvidence: Array<Record<string, unknown>> = [];
       const result = { ...workload, stop, transport: caseTransport, environmentId: caseEnvironment.id,
+        setupActivationConcurrency, setupReadinessStartedAtMs, readinessDeadlineMs, allReadyObservedAtMs, activations, activationWaves,
         nonce, markerCode, epochMs, settledAtMs: settledAt, stopRequestedAtMs: stopRequestedAt,
         stopAcknowledgedAtMs: acknowledgedAt, preludeProof, controllerAuthority, clockEvidence, interChunkGaps,
         providerClocks, localProcesses, localStopProof,