diff --git a/scripts/compare-native-ingestion-browser.ts b/scripts/compare-native-ingestion-browser.ts index cc8d6d90e5b..354e7b0ef37 100644 --- a/scripts/compare-native-ingestion-browser.ts +++ b/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 = { 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"); diff --git a/scripts/native-ingestion-performance-gates.test.mjs b/scripts/native-ingestion-performance-gates.test.mjs index 024b375ff85..21753ab312c 100644 --- a/scripts/native-ingestion-performance-gates.test.mjs +++ b/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); diff --git a/tests/e2e/native-event-ingestion.journey.ts b/tests/e2e/native-event-ingestion.journey.ts index 2b2bf6a6e4a..bce7653d9db 100644 --- a/tests/e2e/native-event-ingestion.journey.ts +++ b/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; 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([[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 | 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> = []; 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,