From 179fe6fa6da1b24ca08c1162301538d3e3788af6 Mon Sep 17 00:00:00 2001 From: Dotta Date: Tue, 3 Nov 2026 23:12:50 -0600 Subject: [PATCH] test: isolate ingestion streams by normal agent identity --- scripts/compare-native-ingestion-browser.ts | 281 ++++++++++-------- scripts/describe-native-ingestion-h5.mjs | 2 +- scripts/native-ingestion-h5.test.mjs | 26 +- ...ative-ingestion-performance-gates.test.mjs | 65 +++- tests/e2e/fixtures/native-ingestion-h5.md | 110 +++++-- tests/e2e/fixtures/native-ingestion-h5.mjs | 87 +++++- tests/e2e/native-event-ingestion.journey.ts | 25 +- 7 files changed, 409 insertions(+), 187 deletions(-) diff --git a/scripts/compare-native-ingestion-browser.ts b/scripts/compare-native-ingestion-browser.ts index 4d34d7d127d..d593deb41db 100644 --- a/scripts/compare-native-ingestion-browser.ts +++ b/scripts/compare-native-ingestion-browser.ts @@ -3,7 +3,8 @@ import { createHash } from "node:crypto"; import { readFile, writeFile } from "node:fs/promises"; import path from "node:path"; import { noiseExceedingPairedWins } from "./native-ingestion-performance-gates.mjs"; -import { H5_MODE, H5_PROTOCOL, H5_WORKLOAD_SHA256, H5_BASELINE_REVISIONS, ORIGINAL_BUDGETS, assessH5Arm, requireH5Candidate } from "../tests/e2e/fixtures/native-ingestion-h5.mjs"; +import { H5_MODE, H5_PROTOCOL, H4D_PROTOCOL, H5_WORKLOAD_SHA256, H5_BASELINE_REVISIONS, ORIGINAL_BUDGETS, + assessH5Arm, requireH5Candidate, assertParticipantSetup, assertParticipantTopology } from "../tests/e2e/fixtures/native-ingestion-h5.mjs"; import { validateArmInventory } from "./native-ingestion-h5-inventory.mjs"; // Each manifest entry points at native-ingestion-evidence JSON attachments @@ -16,6 +17,7 @@ const phase = arg("phase") ?? "compare", expectation = arg("expect") ?? "improve const h5 = arg("observation-mode") === H5_MODE; assert.ok(arg("observation-mode") === undefined || h5, "Unknown observation mode"); const candidatePolicy = arg("candidate-policy"); +assert.ok(!arg("original-h4c-proof") && !arg("original-h4c-inventory"), "Old H4c proof arguments cannot confirm the new independent-agent geometry"); assert.ok(["aa", "compare"].includes(phase)); assert.ok(["improvement", "nonregression"].includes(expectation)); const manifest = JSON.parse(await readFile(manifestPath, "utf8")) as Array<{ baselineFile: string; candidateFile: string; @@ -107,10 +109,128 @@ const median = (values: number[]) => [...values].sort((a, b) => a - b)[Math.floo const first = pairs[0]; const sha256 = (value: unknown) => typeof value === "string" && /^[a-f0-9]{64}$/.test(value); const uuid = (value: unknown) => typeof value === "string" && /^[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}$/i.test(value); +function assertCaseAuthority(report: any, result: any) { + const remote = report.remote, local = report.local; + const authority = result.controllerAuthority; + assert.equal(authority?.checkedBeforeMeasuredEpoch, true, "Missing controller authority proof before the measured epoch"); + assert.equal(authority?.kind, result.stop ? "host-file-journal" : "database-controller-checkpoint", "Wrong checkpoint authority for actual transport"); + assert.equal(authority?.observedReadOnly, "on", "Remote authority proof must observe a read-only transaction"); + assert.equal(authority?.observedIsolation, "repeatable read", "Remote authority proof must observe repeatable-read isolation"); + assert.ok(Number.isSafeInteger(authority?.checkedAtMs) && authority.checkedAtMs > 0 + && Number.isSafeInteger(authority.connectionClosedAtMs) && authority.connectionClosedAtMs > 0 + && authority.checkedAtMs <= authority.connectionClosedAtMs && authority.connectionClosedAtMs < result.epochMs, + "Close the actual remote authority snapshot before the measured epoch"); + assert.equal(authority.rows?.length, result.runs, "Every remote run needs its controller checkpoint and lease proof"); + assertParticipantTopology(report, result); + assert.deepEqual(authority.rows.map((row: any) => row.run_id).sort(), [...result.runIds].sort(), "Remote authority run set mismatch"); + assert.equal(new Set(authority.rows.map((row: any) => row.issue_id)).size, result.runs, "Duplicate remote issue authority"); + const authorityByRun = new Map(); + for (const row of authority.rows) { + assert.ok(uuid(row.issue_id) && uuid(row.lease_id), "Missing actual task/lease UUID"); + assert.equal(row.company_id, report.companyId, "Remote authority company mismatch"); + assert.equal(row.environment_id, result.environmentId, "Remote authority environment mismatch"); + assert.equal(row.runtime_mode, "native"); + if (!result.stop) { + assert.ok(remote.agentIds.includes(row.agent_id), "Remote run did not use its normal remote agent"); + assert.equal(row.provider, "fake-plugin"); + assert.match(row.provider_lease_id, /^fake-plugin:\/\//); + assert.equal(row.checkpoint_part, "header", "Missing actual database controller checkpoint header"); + assert.ok(Number.isSafeInteger(row.checkpoint_header_bytes) && row.checkpoint_header_bytes > 0); + } else { + assert.equal(row.agent_id, local.controlAgentId, "Stop did not use its separate local control agent"); + assert.equal(row.provider, "local", "Stop requires its normal local lease"); + assert.equal(row.provider_lease_id, null); assert.equal(row.checkpoint_part, null, "Local Stop must retain host-file authority without a database header"); + assert.ok(BigInt(String(row.process_identity_seq)) > 0n); + assert.equal(row.process_identity_source_event_id, null, "Local identity must be server-authored"); + assert.equal(row.process_identity?.targetKind, "local"); + assert.equal(row.process_identity?.processPid, row.process_pid); + assert.equal(row.process_identity?.processGroupId, row.process_group_id); + assert.equal(row.process_identity?.processStartedAt, row.process_started_at); + assert.ok(uuid(row.admitted_execution_workspace_id), "Missing actual admitted local workspace binding"); + } + authorityByRun.set(row.run_id, row); + } + return { authority, authorityByRun }; +} + +function assertCaseSettlement(report: any, result: any, validated = assertCaseAuthority(report, result)) { + const remote = report.remote, local = report.local; + const { authority, authorityByRun } = validated; + if (result.stop) { + assert.equal(result.stopAcknowledgedAtMs, result.stopRequestedAtMs + result.stopThroughAcknowledgementMs, "Stop acknowledgement timing boundary changed"); + const row = authority.rows[0], journal = authority.localJournal, before = result.localProcesses; + assert.equal(journal?.schema, "paperclip.runner.durable.control-plane-state.v1", "Missing owned local journal authority"); + assert.ok(sha256(journal.sha256) && journal.bytes > 0 && Number.isSafeInteger(journal.checkedAtMs) + && journal.checkedAtMs < result.epochMs, "Verify actual local file journal before the measured epoch"); + assert.notEqual(journal.checkpointEventPolicy, "control-v1", "Preserve the original local raw transport recovery policy"); + for (const [field, expected] of [["runId", row.run_id], ["runnerInstanceId", row.runner_instance_id], + ["normalizedSessionId", row.native_session_id], ["environmentLeaseId", row.admitted_execution_workspace_id]]) assert.equal(journal.identity?.[field], expected, "Local journal identity mismatch"); + assert.ok(before && before.hostname === row.process_identity.hostname && before.runnerPid === row.process_pid + && before.processGroupId === row.process_group_id && before.providerPid === result.providerClocks[0].providerPid, + "Local observed process ownership mismatch"); + assert.ok(before.runnerAlive && before.groupAlive && before.providerAlive && Number.isSafeInteger(before.checkedAtMs) + && before.checkedAtMs < result.epochMs, "Independently observe all owned local processes before emission"); + const stopped = result.localStopProof, database = stopped?.database, after = database?.rows?.[0]; + assert.ok(stopped && Number.isSafeInteger(stopped.checkedAtMs) && stopped.checkedAtMs >= result.stopAcknowledgedAtMs + && stopped.checkedAtMs >= result.settledAtMs, "Verify local physical Stop after timing boundaries"); + assert.equal(database?.kind, "host-file-journal"); assert.equal(database?.observedReadOnly, "on"); + assert.equal(database?.observedIsolation, "repeatable read"); assert.equal(database?.rows?.length, 1); + assert.equal(database?.checkedBeforeMeasuredEpoch, false); + assert.ok(Number.isSafeInteger(database.checkedAtMs) && database.checkedAtMs >= result.stopAcknowledgedAtMs + && Number.isSafeInteger(database.connectionClosedAtMs) && database.connectionClosedAtMs >= database.checkedAtMs + && database.connectionClosedAtMs <= stopped.checkedAtMs, "Local stopped database proof predates acknowledgement"); + for (const field of ["run_id", "company_id", "environment_id", "lease_id", "agent_id", "provider", "provider_lease_id", + "admitted_execution_workspace_id", "process_identity_seq", "process_identity_source_event_id"]) assert.equal(after?.[field], row[field], "Local stopped database identity mismatch"); + assert.equal(after.status, "cancelled"); assert.equal(after.cancellation_dispatch_state, "acknowledged"); + if (!after.process_pid && !after.process_group_id) { + assert.ok(BigInt(String(after.process_stopped_seq)) > BigInt(String(row.process_identity_seq)), "Cleared local process identity needs its server stop receipt"); + assert.equal(after.process_stopped_source_event_id, null); + assert.equal(after.process_stopped?.processPid, before.runnerPid); + assert.equal(after.process_stopped?.processGroupId, before.processGroupId); + assert.equal(after.process_stopped?.processStartedAt, row.process_identity.processStartedAt); + } else { + assert.equal(after.process_pid, before.runnerPid); assert.equal(after.process_group_id, before.processGroupId); + } + for (const field of ["runnerPid", "processGroupId", "providerPid", "hostname"]) assert.equal(stopped.processes?.[field], before[field]); + assert.equal(stopped.processes.runnerAbsent, true); assert.equal(stopped.processes.groupAbsent, true); + assert.equal(stopped.processes.providerAbsent, true, "Local provider remains alive after Stop"); + const lease = stopped.lease; + assert.equal(lease?.id, row.lease_id); assert.equal(lease?.companyId, report.companyId); + assert.equal(lease?.environmentId, local.environmentId); assert.equal(lease?.heartbeatRunId, row.run_id); + assert.equal(lease?.provider, "local"); assert.equal(lease?.providerLeaseId, null); + assert.ok(["released", "expired", "failed"].includes(lease.status) && lease.releasedAt + && lease.cleanupStatus !== "failed", "Local Stop lease cleanup remains unsettled"); + assert.equal(result.remoteLeases, undefined, "Local Stop cannot claim fake remote cleanup"); + } else { + assert.equal(result.localStopProof, undefined, "Streaming cannot claim a local Stop proof"); + assert.equal(result.remoteLeases?.length, result.runs, "Every remote run needs post-timing lease cleanup proof"); + assert.deepEqual(result.remoteLeases.map((lease: any) => lease.runId).sort(), [...result.runIds].sort(), "Remote cleanup run set mismatch"); + assert.ok(Number.isSafeInteger(result.remoteLeasesCheckedAtMs) && result.remoteLeasesCheckedAtMs >= result.settledAtMs, + "Verify remote leases after recorded timing boundaries"); + for (const lease of result.remoteLeases) { + const row = authorityByRun.get(lease.runId); + assert.equal(lease.id, row.lease_id); assert.equal(lease.companyId, report.companyId); + assert.equal(lease.environmentId, remote.environmentId); assert.equal(lease.provider, row.provider); + assert.equal(lease.providerLeaseId, row.provider_lease_id); + assert.ok(["released", "expired", "failed"].includes(lease.status) && lease.releasedAt + && lease.cleanupStatus === "success", "Remote lease cleanup remains unsettled"); + if (lease.terminationReceipt) { + const receipt = lease.terminationReceipt; + assert.equal(receipt.schema, "paperclip.remote-termination.v1"); + assert.equal(receipt.companyId, report.companyId); assert.equal(receipt.runId, lease.runId); + assert.equal(receipt.leaseId, lease.id); assert.equal(receipt.provider, lease.provider); + assert.equal(receipt.providerLeaseId, lease.providerLeaseId); + assert.ok(["stopped", "destroyed"].includes(receipt.state), "Invalid recorded remote termination receipt"); + } + } + } +} + + for (const pair of pairs) for (const arm of ["baseline", "candidate"] as const) { const report = pair[arm]; - assert.equal(report.protocolVersion, h5 ? H5_PROTOCOL : "v4-remote-streaming-original-local-stop-clock-stream", - h5 ? "Fresh browser acceptance requires its qualified H5 observation protocol" : "Fresh browser acceptance requires the qualified v4 mixed clock protocol"); + assert.equal(report.protocolVersion, h5 ? H5_PROTOCOL : H4D_PROTOCOL, + "Fresh browser acceptance requires the independent-agent observation or default-deadline protocol"); assert.equal(report.transport, "sandbox-remote-with-local-stop", "V4 requires remote streaming and the original local Stop transport"); assert.deepEqual(report.transportPolicy, { streaming: "sandbox-remote", operatorStop: "local", realCloudStop: "Separate required qualification; fake-plugin void receipts grant no remote Stop authority" }, "Missing explicit transport and real cloud Stop qualification policy"); @@ -161,6 +281,7 @@ for (const pair of pairs) for (const arm of ["baseline", "candidate"] as const) 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"); + assertParticipantSetup(report); 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"); @@ -178,6 +299,10 @@ for (const pair of pairs) for (const arm of ["baseline", "candidate"] as const) if (h5) for (const field of ["manifestFileSha256", "aggregateSha256", "qualifiedReceiptSha256", "inventorySummarySha256", "revision"]) assert.equal(report.armInventory[field], first[arm].armInventory[field], "Full qualified artifact inventory changed between samples"); assert.equal(report.measurementSha256, first.baseline.measurementSha256, "Fixture/observer protocol changed between runs"); + assert.match(report.harnessProvenance?.sourceRevision ?? "", /^[a-f0-9]{40}$/, "Missing frozen harness source revision"); + assert.equal(report.harnessProvenance.sourceStatus, ""); + for (const field of ["sourceRevision", "sourceHashes", "driverLockSha256"]) + assert.deepEqual(report.harnessProvenance[field], first.baseline.harnessProvenance[field], "Harness source changed between arms"); if (phase === "aa") assert.equal(report.sourceRevision, first.baseline.sourceRevision, "A/A must use one source revision"); for (const name of names) { const result = report.cases.find((item: any) => item.name === name); @@ -204,44 +329,7 @@ for (const pair of pairs) for (const arm of ["baseline", "candidate"] as const) assert.ok(result.runIds.every(uuid), "Missing actual remote run UUID"); assert.equal(new Set(result.runIds).size, result.runs, "Duplicate remote run identity"); assert.deepEqual(result.durable.map((proof: any) => proof.runId).sort(), [...result.runIds].sort(), "Durable proof run set mismatch"); - const authority = result.controllerAuthority; - assert.equal(authority?.checkedBeforeMeasuredEpoch, true, "Missing controller authority proof before the measured epoch"); - assert.equal(authority?.kind, result.stop ? "host-file-journal" : "database-controller-checkpoint", "Wrong checkpoint authority for actual transport"); - assert.equal(authority?.observedReadOnly, "on", "Remote authority proof must observe a read-only transaction"); - assert.equal(authority?.observedIsolation, "repeatable read", "Remote authority proof must observe repeatable-read isolation"); - assert.ok(Number.isSafeInteger(authority?.checkedAtMs) && authority.checkedAtMs > 0 - && Number.isSafeInteger(authority.connectionClosedAtMs) && authority.connectionClosedAtMs > 0 - && authority.checkedAtMs <= authority.connectionClosedAtMs && authority.connectionClosedAtMs < result.epochMs, - "Close the actual remote authority snapshot before the measured epoch"); - assert.equal(authority.rows?.length, result.runs, "Every remote run needs its controller checkpoint and lease proof"); - assert.deepEqual(authority.rows.map((row: any) => row.run_id).sort(), [...result.runIds].sort(), "Remote authority run set mismatch"); - assert.equal(new Set(authority.rows.map((row: any) => row.issue_id)).size, result.runs, "Duplicate remote issue authority"); - const authorityByRun = new Map(); - for (const row of authority.rows) { - assert.ok(uuid(row.issue_id) && uuid(row.lease_id), "Missing actual task/lease UUID"); - assert.equal(row.company_id, report.companyId, "Remote authority company mismatch"); - assert.equal(row.environment_id, result.environmentId, "Remote authority environment mismatch"); - assert.equal(row.runtime_mode, "native"); - if (!result.stop) { - assert.ok(remote.agentIds.includes(row.agent_id), "Remote run did not use its normal remote agent"); - assert.equal(row.provider, "fake-plugin"); - assert.match(row.provider_lease_id, /^fake-plugin:\/\//); - assert.equal(row.checkpoint_part, "header", "Missing actual database controller checkpoint header"); - assert.ok(Number.isSafeInteger(row.checkpoint_header_bytes) && row.checkpoint_header_bytes > 0); - } else { - assert.equal(row.agent_id, local.controlAgentId, "Stop did not use its separate local control agent"); - assert.equal(row.provider, "local", "Stop requires its normal local lease"); - assert.equal(row.provider_lease_id, null); assert.equal(row.checkpoint_part, null, "Local Stop must retain host-file authority without a database header"); - assert.ok(BigInt(String(row.process_identity_seq)) > 0n); - assert.equal(row.process_identity_source_event_id, null, "Local identity must be server-authored"); - assert.equal(row.process_identity?.targetKind, "local"); - assert.equal(row.process_identity?.processPid, row.process_pid); - assert.equal(row.process_identity?.processGroupId, row.process_group_id); - assert.equal(row.process_identity?.processStartedAt, row.process_started_at); - assert.ok(uuid(row.admitted_execution_workspace_id), "Missing actual admitted local workspace binding"); - } - authorityByRun.set(row.run_id, row); - } + const { authority, authorityByRun } = assertCaseAuthority(report, result); assert.equal(result.providerClocks?.length, result.runs, "Every actual provider needs its wrapper launcher and PID"); assert.deepEqual(result.providerClocks.map((clock: any) => clock.runId).sort(), [...result.runIds].sort()); assert.equal(new Set(result.providerClocks.map((clock: any) => clock.providerPid)).size, result.runs); @@ -303,74 +391,7 @@ for (const pair of pairs) for (const arm of ["baseline", "candidate"] as const) } } 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"); - const row = authority.rows[0], journal = authority.localJournal, before = result.localProcesses; - assert.equal(journal?.schema, "paperclip.runner.durable.control-plane-state.v1", "Missing owned local journal authority"); - assert.ok(sha256(journal.sha256) && journal.bytes > 0 && Number.isSafeInteger(journal.checkedAtMs) - && journal.checkedAtMs < result.epochMs, "Verify actual local file journal before the measured epoch"); - assert.notEqual(journal.checkpointEventPolicy, "control-v1", "Preserve the original local raw transport recovery policy"); - for (const [field, expected] of [["runId", row.run_id], ["runnerInstanceId", row.runner_instance_id], - ["normalizedSessionId", row.native_session_id], ["environmentLeaseId", row.admitted_execution_workspace_id]]) assert.equal(journal.identity?.[field], expected, "Local journal identity mismatch"); - assert.ok(before && before.hostname === row.process_identity.hostname && before.runnerPid === row.process_pid - && before.processGroupId === row.process_group_id && before.providerPid === result.providerClocks[0].providerPid, - "Local observed process ownership mismatch"); - assert.ok(before.runnerAlive && before.groupAlive && before.providerAlive && Number.isSafeInteger(before.checkedAtMs) - && before.checkedAtMs < result.epochMs, "Independently observe all owned local processes before emission"); - const stopped = result.localStopProof, database = stopped?.database, after = database?.rows?.[0]; - assert.ok(stopped && Number.isSafeInteger(stopped.checkedAtMs) && stopped.checkedAtMs >= result.stopAcknowledgedAtMs - && stopped.checkedAtMs >= result.settledAtMs, "Verify local physical Stop after timing boundaries"); - assert.equal(database?.kind, "host-file-journal"); assert.equal(database?.observedReadOnly, "on"); - assert.equal(database?.observedIsolation, "repeatable read"); assert.equal(database?.rows?.length, 1); - assert.equal(database?.checkedBeforeMeasuredEpoch, false); - assert.ok(Number.isSafeInteger(database.checkedAtMs) && database.checkedAtMs >= result.stopAcknowledgedAtMs - && Number.isSafeInteger(database.connectionClosedAtMs) && database.connectionClosedAtMs >= database.checkedAtMs - && database.connectionClosedAtMs <= stopped.checkedAtMs, "Local stopped database proof predates acknowledgement"); - for (const field of ["run_id", "company_id", "environment_id", "lease_id", "agent_id", "provider", "provider_lease_id", - "admitted_execution_workspace_id", "process_identity_seq", "process_identity_source_event_id"]) assert.equal(after?.[field], row[field], "Local stopped database identity mismatch"); - assert.equal(after.status, "cancelled"); assert.equal(after.cancellation_dispatch_state, "acknowledged"); - if (!after.process_pid && !after.process_group_id) { - assert.ok(BigInt(String(after.process_stopped_seq)) > BigInt(String(row.process_identity_seq)), "Cleared local process identity needs its server stop receipt"); - assert.equal(after.process_stopped_source_event_id, null); - assert.equal(after.process_stopped?.processPid, before.runnerPid); - assert.equal(after.process_stopped?.processGroupId, before.processGroupId); - assert.equal(after.process_stopped?.processStartedAt, row.process_identity.processStartedAt); - } else { - assert.equal(after.process_pid, before.runnerPid); assert.equal(after.process_group_id, before.processGroupId); - } - for (const field of ["runnerPid", "processGroupId", "providerPid", "hostname"]) assert.equal(stopped.processes?.[field], before[field]); - assert.equal(stopped.processes.runnerAbsent, true); assert.equal(stopped.processes.groupAbsent, true); - assert.equal(stopped.processes.providerAbsent, true, "Local provider remains alive after Stop"); - const lease = stopped.lease; - assert.equal(lease?.id, row.lease_id); assert.equal(lease?.companyId, report.companyId); - assert.equal(lease?.environmentId, local.environmentId); assert.equal(lease?.heartbeatRunId, row.run_id); - assert.equal(lease?.provider, "local"); assert.equal(lease?.providerLeaseId, null); - assert.ok(["released", "expired", "failed"].includes(lease.status) && lease.releasedAt - && lease.cleanupStatus !== "failed", "Local Stop lease cleanup remains unsettled"); - assert.equal(result.remoteLeases, undefined, "Local Stop cannot claim fake remote cleanup"); - } else { - assert.equal(result.localStopProof, undefined, "Streaming cannot claim a local Stop proof"); - assert.equal(result.remoteLeases?.length, result.runs, "Every remote run needs post-timing lease cleanup proof"); - assert.deepEqual(result.remoteLeases.map((lease: any) => lease.runId).sort(), [...result.runIds].sort(), "Remote cleanup run set mismatch"); - assert.ok(Number.isSafeInteger(result.remoteLeasesCheckedAtMs) && result.remoteLeasesCheckedAtMs >= result.settledAtMs, - "Verify remote leases after recorded timing boundaries"); - for (const lease of result.remoteLeases) { - const row = authorityByRun.get(lease.runId); - assert.equal(lease.id, row.lease_id); assert.equal(lease.companyId, report.companyId); - assert.equal(lease.environmentId, remote.environmentId); assert.equal(lease.provider, row.provider); - assert.equal(lease.providerLeaseId, row.provider_lease_id); - assert.ok(["released", "expired", "failed"].includes(lease.status) && lease.releasedAt - && lease.cleanupStatus === "success", "Remote lease cleanup remains unsettled"); - if (lease.terminationReceipt) { - const receipt = lease.terminationReceipt; - assert.equal(receipt.schema, "paperclip.remote-termination.v1"); - assert.equal(receipt.companyId, report.companyId); assert.equal(receipt.runId, lease.runId); - assert.equal(receipt.leaseId, lease.id); assert.equal(receipt.provider, lease.provider); - assert.equal(receipt.providerLeaseId, lease.providerLeaseId); - assert.ok(["stopped", "destroyed"].includes(receipt.state), "Invalid recorded remote termination receipt"); - } - } - } + assertCaseSettlement(report, result, { authority, authorityByRun }); for (const proof of result.durable) { assert.equal(proof.ordered, true); assert.equal(proof.uniqueSourceIdentities, proof.individualDeltas); assert.equal(proof.matchedSourceClocks, proof.individualDeltas, "Every durable measured delta needs an original provider clock"); @@ -380,21 +401,25 @@ for (const pair of pairs) for (const arm of ["baseline", "candidate"] as const) for (const field of ["runs", "events", "intervalMs", "prelude", "stop"]) assert.equal(result[field], reference[field], `Workload changed: ${name}.${field}`); } } -let originalH4cConfirmation: Record | undefined; +let originalH4dConfirmation: Record | undefined; if (h5 && phase === "compare" && candidatePolicy === "final-b2") { - originalH4cConfirmation = { required: true, passed: false, reason: "Later unchanged original H4c full functional proof is pending" }; - if (arg("original-h4c-proof")) { + originalH4dConfirmation = { required: true, passed: false, reason: "Later H4d default-mode full functional proof on the same source and geometry is pending" }; + if (arg("original-h4d-proof")) { try { - assert.ok(arg("original-h4c-inventory"), "Original H4c proof requires its full artifact inventory sidecar"); - const { evidence: proof, owner } = await owningResult(path.resolve(arg("original-h4c-proof")!)); + assert.ok(arg("original-h4d-inventory"), "H4d original-deadline proof requires its full artifact inventory sidecar"); + const { evidence: proof, owner } = await owningResult(path.resolve(arg("original-h4d-proof")!)); assert.equal(owner.status, "passed"); assert.equal(owner.retry, 0); assert.deepEqual(owner.errors, []); assert.equal(owner.timeoutMs, ORIGINAL_BUDGETS.fullInvocationMs); assert.ok(owner.durationMs > 0 && owner.durationMs < ORIGINAL_BUDGETS.fullInvocationMs); - assert.equal(proof.protocolVersion, "v4-remote-streaming-original-local-stop-clock-stream"); - assert.equal(proof.observationMode, undefined, "Confirmation must use unchanged original H4c defaults"); - assert.equal(proof.harnessProvenance?.sourceRevision, "7d7e7a88d562a95a1d407b76474763b666c181f5"); + assert.equal(proof.protocolVersion, H4D_PROTOCOL, "Old H4c geometry cannot confirm H5b"); + assertParticipantSetup(proof); + assert.deepEqual(proof.agentTopology, first.candidate.agentTopology); + assert.equal(proof.observationMode, undefined, "Confirmation must use H4d default deadlines"); + assert.equal(proof.harnessProvenance?.sourceRevision, first.candidate.harnessProvenance.sourceRevision, "H4d confirmation must use the same frozen harness source"); + assert.deepEqual(proof.harnessProvenance.sourceHashes, first.candidate.harnessProvenance.sourceHashes); + assert.equal(proof.harnessProvenance.driverLockSha256, first.candidate.harnessProvenance.driverLockSha256); assert.equal(proof.harnessProvenance.sourceStatus, ""); - assert.equal(proof.measurementSha256, "5db4129bbdc1fa38775121aff5d0e68a8a11762519e41cfdf36eacd90ba7256b"); + assert.equal(proof.measurementSha256, first.candidate.measurementSha256, "H4d confirmation must bind the same measured harness bytes"); assert.deepEqual(proof.resourceBudgets, ORIGINAL_BUDGETS); assert.equal(proof.complete, true); assert.equal(proof.qualificationOnly, false); assert.equal(proof.diagnosticOnly, false); assert.equal(proof.performanceAcceptanceEligible, true); assert.equal(proof.serverArtifact, "compiled"); @@ -407,6 +432,7 @@ if (h5 && phase === "compare" && candidatePolicy === "final-b2") { assert.deepEqual(proof.cases.map((item: any) => item.name), names, "Original confirmation must include all six workloads"); for (const fixed of workloads) { const item = proof.cases.find((item: any) => item.name === fixed.name); + assertCaseSettlement(proof, item); for (const field of ["runs", "events", "intervalMs", "prelude", "stop"] as const) assert.equal(item[field], fixed[field]); assert.equal(item.persistedAfterReload, true); assert.equal(item.durable.length, fixed.runs); assert.equal(item.matchedVisibleClocks, item.visible.length); @@ -421,12 +447,12 @@ if (h5 && phase === "compare" && candidatePolicy === "final-b2") { const latestCandidateEnd = Math.max(...pairs.map(pair => Date.parse(pair.candidate.owningPlaywright.startedAt) + pair.candidate.owningPlaywright.durationMs)); assert.ok(Date.parse(owner.startedAt) > latestCandidateEnd, "Original confirmation must follow all candidate arms"); assert.ok(pairs.every(pair => proof.companyId !== pair.baseline.companyId && proof.companyId !== pair.candidate.companyId), "Original confirmation needs a fresh disposable company/database"); - const inventory = await validateArmInventory(path.resolve(arg("original-h4c-inventory")!), { evidence: proof, owningPlaywright: owner }); + const inventory = await validateArmInventory(path.resolve(arg("original-h4d-inventory")!), { evidence: proof, owningPlaywright: owner }); for (const field of ["manifestFileSha256", "aggregateSha256", "qualifiedReceiptSha256", "inventorySummarySha256", "revision"] as const) assert.equal(inventory[field], first.candidate.armInventory[field], "Original confirmation used a different full qualified artifact inventory"); - originalH4cConfirmation = { required: true, passed: true, owner, inventory, sourceRevision: proof.sourceRevision }; + originalH4dConfirmation = { required: true, passed: true, owner, inventory, sourceRevision: proof.sourceRevision }; } catch (error) { - originalH4cConfirmation.reason = error instanceof Error ? error.message : String(error); + originalH4dConfirmation.reason = error instanceof Error ? error.message : String(error); } } } @@ -453,7 +479,8 @@ 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", + scope: "Compiled-app streaming with100 independent normal agents and the original local Stop workload; real cloud Stop is separately required", + agentTopology: first.baseline.agentTopology, transportPolicy: first.baseline.transportPolicy, additionalRealCloudStopQualificationRequired: true, admissionProtocol: first.baseline.admissionProtocol, setupActivationConcurrency: first.baseline.setupActivationConcurrency, protocolVersion: first.baseline.protocolVersion, transport: first.baseline.transport, @@ -465,9 +492,9 @@ const result: Record = { manifestSha256: createHash("sha256").update(await readFile(manifestPath)).digest("hex"), measurements, primaryLoadedWorkload: "burst512", complete: true, ...(h5 ? { observationMode: H5_MODE, workloadSha256: H5_WORKLOAD_SHA256, candidatePolicy, h5Assessments, - artifactInventories: { baseline: first.baseline.armInventory, candidate: first.candidate.armInventory }, originalH4cConfirmation, - originalProtocolAcceptance: "Baseline SLA misses are retained; H5 complete observation is not H4c PASS", - finalUnchangedH4cFunctionalPassRequired: candidatePolicy === "final-b2" } : {}), + artifactInventories: { baseline: first.baseline.armInventory, candidate: first.candidate.armInventory }, originalH4dConfirmation, + originalProtocolAcceptance: "Baseline SLA misses are retained; H5b complete observation is not an original-deadline PASS", + finalH4dFunctionalPassRequired: candidatePolicy === "final-b2" } : {}), }; if (phase === "compare") { assert.ok(arg("noise-report"), "Comparison requires its frozen A/A --noise-report"); @@ -504,12 +531,12 @@ if (phase === "compare") { }); result.performancePassed = result.gates.every((entry: any) => entry.passed) && (!h5 || h5Assessments.every(entry => entry.candidateSlaPolicyPassed)); - result.acceptancePassed = result.performancePassed && (!originalH4cConfirmation || originalH4cConfirmation.passed); + result.acceptancePassed = result.performancePassed && (!originalH4dConfirmation || originalH4dConfirmation.passed); } if (arg("output")) await writeFile(arg("output")!, JSON.stringify(result, null, 2)); console.log(JSON.stringify(result, null, 2)); if (process.argv.includes("--require-performance")) { - assert.equal(h5, true, "H5 performance-only gating is explicit; original H4c remains unchanged"); + assert.equal(h5, true, "H5b performance-only gating is explicit; default H4d keeps original deadlines"); assert.equal(phase, "compare"); assert.equal(result.performancePassed, true, "H5 performance/SLA gates failed; retain results and diagnose"); } diff --git a/scripts/describe-native-ingestion-h5.mjs b/scripts/describe-native-ingestion-h5.mjs index 8d4bf55bde2..b93b7f2a3e0 100644 --- a/scripts/describe-native-ingestion-h5.mjs +++ b/scripts/describe-native-ingestion-h5.mjs @@ -4,4 +4,4 @@ console.log(JSON.stringify({ dryRun: true, executionAuthorized: false, workloadS registration: H5_REGISTRATION, requiredBeforeRelease: ['Root-reviewed exact driver/comparator hashes', 'New functional observation qualification', 'Fresh B0 and B1 A/A before further product implementation', 'Root-reviewed frozen A/A report hashes before candidate stages', 'Final B2 every-arm original SLA pass', - 'Later unchanged original H4c full functional pass on the final SUT'] }, null, 2)); + 'Later H4d default-mode full functional pass on the same source, geometry and final SUT'] }, null, 2)); diff --git a/scripts/native-ingestion-h5.test.mjs b/scripts/native-ingestion-h5.test.mjs index 9ae2a9f7b9b..9e074d97cd1 100644 --- a/scripts/native-ingestion-h5.test.mjs +++ b/scripts/native-ingestion-h5.test.mjs @@ -2,7 +2,7 @@ import assert from 'node:assert/strict'; import { test } from 'node:test'; import { createRequire } from 'node:module'; import { h5Enabled, H5_MODE, observeStreaming, originalTerminal, requireH5Candidate, - H5_LIMITS, ORIGINAL_BUDGETS } from '../tests/e2e/fixtures/native-ingestion-h5.mjs'; + H5_LIMITS, ORIGINAL_BUDGETS, prepareParticipantAgents } from '../tests/e2e/fixtures/native-ingestion-h5.mjs'; const timeout = milliseconds => Object.assign(new Error(`Timeout ${milliseconds}ms exceeded while waiting on the predicate`), { matcherResult: { name: 'toBeGreaterThanOrEqual', pass: false } }); @@ -39,9 +39,31 @@ test('real frozen Playwright can report its configured timeout before the nomina assessedAtMs, elapsedMs, observations, matcherName: failure.matcherResult.name, matcherPassed: failure.matcherResult.pass })); }); -test('H5 requires explicit mode and leaves default H4c selected', () => { +test('H5b requires its explicit mode and leaves default H4d selected', () => { assert.equal(h5Enabled(undefined), false); assert.equal(h5Enabled(H5_MODE), true); assert.throws(() => h5Enabled('almost-h5')); + assert.throws(() => h5Enabled('h5-observe-original-sla'), 'Old H5a observations use a different geometry'); +}); + +for (const fail of [false, true]) test(`participant setup uses at most8 workers, preserves index order and settles all attempts${fail ? ' after a failure' : ''}`, async () => { + const count = 20, pending = [], records = []; let active = 0, maximum = 0, finished = 0; + const completion = prepareParticipantAgents({ count, onOutcome: value => records.push(structuredClone(value)), + createAndReady: async index => { + active++; maximum = Math.max(maximum, active); + await new Promise(resolve => pending.push(resolve)); active--; finished++; + if (fail && index === 3) throw new Error('actual agent readiness failed'); + return { id: `agent-${index}` }; + } }).then(value => ({ value }), error => ({ error })); + while (finished < count) { + await new Promise(resolve => setImmediate(resolve)); + assert.ok(active <= 8); pending.splice(0).reverse().forEach(resolve => resolve()); + } + const result = await completion, last = records.at(-1); + assert.equal(maximum, 8); assert.equal(active, 0); assert.equal(last.attempts.length, count); + assert.ok(last.attempts.every(attempt => attempt.status !== 'pending')); + assert.equal(last.complete, !fail); + if (fail) { assert.match(result.error.message, /setup failed/); assert.equal(last.attempts.filter(value => value.status === 'rejected').length, 1); } + else assert.deepEqual(result.value.map(agent => agent.id), Array.from({ length: count }, (_, index) => `agent-${index}`)); }); test('first successful observation uses only the original wait with no collection delay', async () => { diff --git a/scripts/native-ingestion-performance-gates.test.mjs b/scripts/native-ingestion-performance-gates.test.mjs index b146615503e..4b7ce897e3f 100644 --- a/scripts/native-ingestion-performance-gates.test.mjs +++ b/scripts/native-ingestion-performance-gates.test.mjs @@ -1,5 +1,6 @@ import assert from "node:assert/strict"; -import { H5_MODE, H5_PROTOCOL, H5_LIMITS, H5_WORKLOAD_SHA256, H5_BASELINE_REVISIONS, originalTerminal, assessH5Arm, requireH5Candidate } from "../tests/e2e/fixtures/native-ingestion-h5.mjs"; +import { H5_MODE, H5_PROTOCOL, H4D_PROTOCOL, H5_TOPOLOGY, H5_LIMITS, H5_WORKLOAD_SHA256, H5_BASELINE_REVISIONS, + originalTerminal, assessH5Arm, requireH5Candidate } from "../tests/e2e/fixtures/native-ingestion-h5.mjs"; import { spawnSync } from "node:child_process"; import { createHash } from "node:crypto"; import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; @@ -79,17 +80,23 @@ const workloads = [ function browserReport(revision, latency) { const companyId = "00000000-0000-4000-8000-000000000001", environmentId = "00000000-0000-4000-8000-000000000002"; const localEnvironmentId = "00000000-0000-4000-8000-000000000003", controlAgentId = "00000000-0000-4000-8000-000000000004"; - const remoteAgentId = "00000000-0000-4000-8000-000000000005"; const id = (kind, caseIndex, index) => `00000000-0000-4000-8000-${(kind * 100000 + caseIndex * 1000 + index).toString(16).padStart(12, "0")}`; + const remoteAgentIds = Array.from({ length: 100 }, (_, index) => id(9, 0, index)); const transportPolicy = { streaming: "sandbox-remote", operatorStop: "local", realCloudStop: "Separate required qualification; fake-plugin void receipts grant no remote Stop authority" }; - return { protocolVersion: "v4-remote-streaming-original-local-stop-clock-stream", transport: "sandbox-remote-with-local-stop", transportPolicy, companyId, + return { protocolVersion: H4D_PROTOCOL, transport: "sandbox-remote-with-local-stop", transportPolicy, companyId, + bodyStartedAtMs: 1, agentTopology: { ...H5_TOPOLOGY, remoteAgentPoolSize: 100 }, + harnessProvenance: { sourceRevision: "b".repeat(40), sourceStatus: "", driverLockSha256: "8".repeat(64), + sourceHashes: { "native-event-ingestion.journey.ts": "9".repeat(64), "fixtures/native-ingestion-h5.mjs": "a".repeat(64) } }, resourceBudgets: { providerReadinessMs: 300000, fullInvocationMs: 900000, streamingMs: 120000, terminalDrainageMs: 180000, stopAcknowledgementMs: 30000, remoteLeaseCleanupMs: 30000 }, local: { transport: "local", environmentId: localEnvironmentId, controlAgentId, selection: "Normal local environment + separate control agent defaultEnvironmentId; existing test-only Codex leaf command fixture", commandSha256: "6".repeat(64), loaderSha256: "7".repeat(64) }, - remote: { transport: "sandbox-remote", revision, buildRevision: revision, environmentId, agentIds: [remoteAgentId], + remote: { transport: "sandbox-remote", revision, buildRevision: revision, environmentId, agentIds: remoteAgentIds, + agentSetup: { count: 100, concurrency: 8, complete: true, startedAtMs: 2, settledAtMs: 40, + attempts: remoteAgentIds.map((agentId, index) => ({ index, agentId, status: "fulfilled", + startedAtMs: 3 + Math.floor(index / 8) * 2, settledAtMs: 4 + Math.floor(index / 8) * 2 })) }, runnerPublicUrl: "wss://127.0.0.1:32100", selection: "Normal plugin install + sandbox environment + agent defaultEnvironmentId; no runtime resolver override", plugin: { pluginKey: "paperclip.fake-sandbox-provider", assetHashes: Object.fromEntries(["worker.js", "plugin.js", "manifest.js", "lease-cleanup.js"].map(file => [file, "5".repeat(64)])) }, @@ -106,7 +113,7 @@ function browserReport(revision, latency) { observedReadOnly: "on", observedIsolation: "repeatable read", kind: stop ? "host-file-journal" : "database-controller-checkpoint" }; const rows = Array.from({ length: workload.runs }, (_, index) => ({ run_id: id(1, caseIndex, index), issue_id: id(2, caseIndex, index), company_id: companyId, - agent_id: stop ? controlAgentId : remoteAgentId, + agent_id: stop ? controlAgentId : remoteAgentIds[index], lease_id: id(3, caseIndex, index), environment_id: actualEnvironmentId, runtime_mode: "native", provider: stop ? "local" : "fake-plugin", provider_lease_id: stop ? null : `fake-plugin://${id(1, caseIndex, index)}/fixture`, checkpoint_part: stop ? null : "header", checkpoint_header_bytes: stop ? null : 256, @@ -263,16 +270,16 @@ test("H5 CLI retains baseline SLA failure, rejects final SLA misses and requires } try { const proof = JSON.parse(JSON.stringify(h5Report("B2", 1000)).replaceAll("00000000-", "10000000-")); - proof.protocolVersion = "v4-remote-streaming-original-local-stop-clock-stream"; - proof.measurementSha256 = "5db4129bbdc1fa38775121aff5d0e68a8a11762519e41cfdf36eacd90ba7256b"; - proof.harnessProvenance = { sourceRevision: "7d7e7a88d562a95a1d407b76474763b666c181f5", sourceStatus: "" }; + proof.protocolVersion = H4D_PROTOCOL; + proof.measurementSha256 = "h5-synthetic-fixed-fixture"; + // H4d defaults use the same frozen source/geometry as the observation arms. for (const field of ["observationMode", "observationLimits", "observationComplete", "workloadSha256", "actualPlaywrightTimeoutMs"]) delete proof[field]; for (const item of proof.cases) { delete item.h5; delete item.terminalStatuses; for (const row of item.durable) delete row.resultAcceptedEvents; } - const originalProof = await save("original-h4c-playwright.json", { suites: [{ tests: [{ timeout: 900000, expectedStatus: "passed", results: [{ + const originalProof = await save("original-h4d-playwright.json", { suites: [{ tests: [{ timeout: 900000, expectedStatus: "passed", results: [{ status: "passed", retry: 0, duration: 5000, startTime: "2026-11-04T00:00:00.000Z", errors: [], attachments: [{ name: "native-ingestion-evidence", body: Buffer.from(JSON.stringify(proof)).toString("base64") }] }] }] }] }); const originalInventory = await inventory(proof, originalProof, "original-h4c"); - const finalProofArgs = [`--original-h4c-proof=${originalProof}`, `--original-h4c-inventory=${originalInventory}`]; + const finalProofArgs = [`--original-h4d-proof=${originalProof}`, `--original-h4d-inventory=${originalInventory}`]; const noise = path.join(directory, "aa.json"); const aa = run(["--phase=aa", `--pairs=${await manifest("aa", 181000, { sameRevision: true })}`, `--output=${noise}`]); assert.equal(aa.status, 0, aa.stderr); @@ -312,18 +319,34 @@ test("H5 CLI retains baseline SLA failure, rejects final SLA misses and requires for (const [name, change, message] of [ ["h4c-too-early", (_value, owner) => { owner.results[0].startTime = "2026-11-02T00:00:00.000Z"; }, /follow all candidate/], ["h4c-wrong-sut", value => { value.sourceRevision = H5_BASELINE_REVISIONS.B0; value.compiledBuildRevision = value.sourceRevision; }, /strictly equal/], - ["h4c-observation-mode", value => { value.observationMode = H5_MODE; }, /unchanged original H4c/], + ["h4c-observation-mode", value => { value.observationMode = H5_MODE; }, /H4d default deadlines/], + ["old-h4c-proof", value => { value.protocolVersion = "v4-remote-streaming-original-local-stop-clock-stream"; + value.harnessProvenance.sourceRevision = "7d7e7a88d562a95a1d407b76474763b666c181f5"; }, /Old H4c geometry/], + ["h4d-wrong-harness", value => { value.harnessProvenance.sourceRevision = "d".repeat(40); }, /same frozen harness source/], + ["h4d-wrong-measurement", value => { value.measurementSha256 = "old-h5a"; }, /same measured harness bytes/], + ["h4d-shared-agent", value => { value.cases[3].controllerAuthority.rows[1].agent_id = value.cases[3].controllerAuthority.rows[0].agent_id; }, /distinct normal agents/], + ["h4d-missing-stop-proof", value => { delete value.cases[5].localStopProof; }, /physical Stop/], + ["h4d-provider-still-alive", value => { value.cases[5].localStopProof.processes.providerAbsent = false; }, /provider remains alive/], + ["h4d-pending-local-lease", value => { value.cases[5].localStopProof.lease.status = "active"; }, /lease cleanup remains unsettled/], + ["h4d-wrong-local-lease", value => { value.cases[5].localStopProof.lease.id = value.cases[3].controllerAuthority.rows[0].lease_id; }, /strictly equal/], + ["h4d-ack-timing-mismatch", value => { value.cases[5].stopAcknowledgedAtMs++; }, /acknowledgement timing boundary/], + ["h4d-missing-remote-leases", value => { delete value.cases[3].remoteLeases; }, /post-timing lease cleanup proof/], + ["h4d-pending-remote-lease", value => { value.cases[3].remoteLeases[0].cleanupStatus = "pending"; }, /lease cleanup remains unsettled/], + ["h4d-wrong-remote-lease", value => { value.cases[3].remoteLeases[0].id = value.cases[5].controllerAuthority.rows[0].lease_id; }, /strictly equal/], + ["h4d-premature-remote-cleanup-proof", value => { value.cases[3].remoteLeasesCheckedAtMs = value.cases[3].settledAtMs - 1; }, /leases after recorded timing/], + ["h4d-wrong-authority-company", value => { value.cases[3].controllerAuthority.rows[0].company_id = "00000000-0000-4000-8000-000000099999"; }, /authority company mismatch/], + ["h4d-wrong-authority-run", value => { value.cases[3].controllerAuthority.rows[0].run_id = value.cases[5].runIds[0]; }, /authority run set mismatch/], ]) { const report = JSON.parse(await readFile(originalProof, "utf8")), owner = report.suites[0].tests[0], value = JSON.parse(Buffer.from(owner.results[0].attachments[0].body, "base64").toString()); change(value, owner); owner.results[0].attachments[0].body = Buffer.from(JSON.stringify(value)).toString("base64"); const file = await save(`${name}-playwright.json`, report), sidecar = await inventory(value, file, name), output = path.join(directory, `${name}-result.json`); const attempt = run(["--candidate-policy=final-b2", `--pairs=${goodPairs}`, `--noise-report=${noise}`, - `--original-h4c-proof=${file}`, `--original-h4c-inventory=${sidecar}`, `--output=${output}`, "--require-acceptance"]); + `--original-h4d-proof=${file}`, `--original-h4d-inventory=${sidecar}`, `--output=${output}`, "--require-acceptance"]); assert.notEqual(attempt.status, 0, name); const rejected = JSON.parse(await readFile(output, "utf8")); assert.equal(rejected.performancePassed, true); assert.equal(rejected.acceptancePassed, false); - assert.match(rejected.originalH4cConfirmation.reason, message); + assert.match(rejected.originalH4dConfirmation.reason, message); } const badOutput = path.join(directory, "still-misses.json"); const bad = run(["--candidate-policy=final-b2", `--pairs=${await manifest("still-misses", 150000)}`, @@ -333,6 +356,7 @@ test("H5 CLI retains baseline SLA failure, rejects final SLA misses and requires assert.equal(rejected.acceptancePassed, false); assert.equal(rejected.h5Assessments[0].candidateSlaPolicyPassed, false); for (const [name, mutate, message] of [ + ["old-h5a-protocol", report => { report.protocolVersion = "v5-complete-observation-original-sla"; }, /strictly equal/], ["censored", report => { report.cases[3].h5.streaming.observation.censored = true; }, /false/], ["terminal-failed", report => { report.cases[3].terminalStatuses[99].status = "failed"; }, /successful drainage/], ["result-missing", report => { report.cases[3].durable[99].resultAcceptedEvents = 0; }, /persisted result/], @@ -404,9 +428,18 @@ test("browser CLI rejects the three-pairs-above-noise counterexample and ineligi ["profile", report => { report.diagnosticOnly = true; }, /Diagnostic profiles/], ["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; }, /v4 mixed clock protocol/], - ["old-local-protocol", report => { report.protocolVersion = "v2-compact-markers-original-clock-stream"; report.transport = "local"; }, /v4 mixed clock protocol/], - ["old-remote-protocol", report => { report.protocolVersion = "v3-supported-remote-original-clock-stream"; }, /v4 mixed clock protocol/], + ["legacy-protocol", report => { delete report.protocolVersion; }, /independent-agent/], + ["old-local-protocol", report => { report.protocolVersion = "v2-compact-markers-original-clock-stream"; report.transport = "local"; }, /independent-agent/], + ["old-remote-protocol", report => { report.protocolVersion = "v3-supported-remote-original-clock-stream"; }, /independent-agent/], + ["old-h4c-geometry", report => { report.protocolVersion = "v4-remote-streaming-original-local-stop-clock-stream"; }, /independent-agent/], + ["shared-participant-agent", report => { report.cases[3].controllerAuthority.rows[1].agent_id = report.cases[3].controllerAuthority.rows[0].agent_id; }, /distinct normal agents/], + ["participant-outside-pool", report => { report.cases[3].controllerAuthority.rows[1].agent_id = "00000000-0000-4000-8000-000000099999"; }, /outside the normal agent pool/], + ["incomplete-agent-pool", report => { report.remote.agentIds.pop(); }, /Incomplete independent/], + ["unfinished-agent-setup", report => { report.remote.agentSetup.complete = false; }, /settle successfully/], + ["rejected-agent-setup", report => { report.remote.agentSetup.attempts[0].status = "rejected"; }, /Rejected agent setup/], + ["agent-setup-concurrency", report => { report.remote.agentSetup.concurrency = 9; }, /strictly equal/], + ["agent-setup-overlap", report => { for (const row of report.remote.agentSetup.attempts.slice(0, 9)) { row.startedAtMs = 3; row.settledAtMs = 39; } }, /exceeded the fixed concurrency/], + ["late-agent-setup", report => { report.remote.agentSetup.settledAtMs = 60; }, /before every case activation/], ["wrong-transport", report => { report.transport = "local"; }, /original local Stop transport/], ["missing-transport-policy", report => { delete report.transportPolicy; }, /explicit transport/], ["fake-remote-stop", report => { report.cases[5].transport = "sandbox-remote"; }, /Wrong per-case transport/], diff --git a/tests/e2e/fixtures/native-ingestion-h5.md b/tests/e2e/fixtures/native-ingestion-h5.md index 7debd036dfd..fd0be3f4378 100644 --- a/tests/e2e/fixtures/native-ingestion-h5.md +++ b/tests/e2e/fixtures/native-ingestion-h5.md @@ -1,30 +1,87 @@ -# H5 complete observation registration (prospective; no execution release) +# H5b independent-agent observation registration (prospective; no execution release) H4c remains frozen at `7d7e7a88d562a95a1d407b76474763b666c181f5` (`5db4129bbdc1fa38775121aff5d0e68a8a11762519e41cfdf36eacd90ba7256b`). Its observed 120-second streaming failures remain failures. Aborted or profiled -attempts cannot be reused as complete timing arms. H5 repairs observation of the -final outcome; it does not optimize the product or turn those failures into passes. +attempts cannot be reused as complete timing arms. H5b uses one normal agent per concurrent remote participant. It repairs the +experiment topology while preserving observation of original deadline outcomes. +It does not optimize the product or turn earlier failures into passes. -The fresh application baselines remain B0 `a36567f9fbf805ad72207d54f539516e3121741e` -and B1 `d8cd3be0aeed4f9fb93ecc3cf725f1d46d5a0310`. Current B2 -`914129afb141d81c0d6d8c5f0ea11c20bd69a3da` has an observed original120s failure; -it is not already qualified. A later candidate uses its exact new immutable -revision, complete build receipt and inventory; no old binary is presumed equal. +The aligned fresh application baselines are B0 `bac53ec3bfeaf6546e7aaeaee516487b30534654` +and B1 `07061a601cdc20603c8c7a46f9ddbad7ac5f5939`. Raw B2 +`ca7cba171cd5198ed569d75012edc858041cbc44` is the unoptimized batching base, +not a qualified future performance candidate. A later candidate needs its actual +new immutable revision, normal build receipt and inventory; no candidate reference +or matching binary is presumed before those artifacts exist. The historical +a365/d8cd/914 application failures remain evidence of their recorded revisions. Select H5 explicitly with -`PAPERCLIP_INGESTION_OBSERVATION_MODE=h5-observe-original-sla`. -The default code path retains the H4c 120/180/900-second behavior. H5 emits a new -protocol version, hashes its added helper with the driver sources, and records a -separate registration/workload SHA. The original comparator mode rejects H5. +`PAPERCLIP_INGESTION_OBSERVATION_MODE=h5b-observe-original-sla-independent-agents`. +The new default H4d mode uses the same independent-agent geometry with the original +120/180/900-second deadlines. Observation mode uses the explicit H5b protocol. +Both modes hash the same nine source files, including the shared topology helper. +Their source revision and measurement bytes must match for later confirmation. +Old H4c/H5a protocols, geometry, arms and noise are ineligible for H5b. Run `node scripts/describe-native-ingestion-h5.mjs` to inspect the registration. This is a dry run only: it starts no app, provider, database or browser and accepts no execution arguments. It grants no runtime release. +## Independent participant geometry and setup + +Create a pool of100 normal remote agents for the full fixture (three for smoke). +Each concurrent remote participant receives a distinct pool agent by index; the +pool is reused only across sequential cases. The100-run case is100 agents × one +run each. Agent POST, normal environment selection, managed files, permission +checks and maxConcurrentRuns50 configuration stay enabled. The separate local +Stop control agent and its normal local environment stay unchanged. + +Use exactly eight setup workers, each awaiting normal agent creation and the +existing60s lifecycle-ready poll. Keep index order despite completion order. +Attempt and settle the whole pool; any setup error fails the invocation before +case activation. Record every attempt, start/settlement time and status. No +partial pool, silent retry or reduced load can qualify. The existing per-case +READ ONLY authority snapshot must prove distinct agent_id values for every +remote participant, all drawn from the ready pool. No event polling is added. + +Agent creation occurs before the existing300s activation/readiness window starts. +That start point remains after the pool and background issue creation. All agent +setup remains inside the owning public900s default/1200s observation invocation; +no setup work is moved outside those timers or into measured epochs. The existing +waves of two provider activations, readiness polls and common source epoch stay +unchanged. All arms use the same geometry and eight-worker setup. Its actual +setup cost and100-agent full qualification are unverified until measured. + +This changes managed-directory contention and agent-list cardinality. A result +will describe independent-agent ingestion, not50 concurrent runs sharing one +agent. Do not generalize it to the retained old2×50 failure below. + +## Retained H5a negative result + +H5a7388 on application B0a365 used two normal agents with50 concurrent runs each. +All100 participants were admitted and clock-ready. The original120s poll failed +with54/512 visible markers; all512 became visible296.555s after that wait began. +All100 runs reached a terminal status55.234s later, with86 succeeded and14 failed. +Independent read-only PostgreSQL checks found all51,200 expected lossless deltas +and one accepted result for each run. This did not pass full qualification or A/A. + +Each of the14 failed runs recorded native_session_interrupted from a30s +agent-directory merge-lock timeout; each finalization retained retryable_failure +with native_workspace_sync_out_failed and an ACTIVE unreleased lease. Seven +failures mapped to each agent's shared file-sync root. Task workspace and lease +identities were distinct; no provider failure was recorded. The new geometry +removes that shared-agent contention from this experiment. It does not fix or +qualify the product's old2×50 concurrency limit or justify disabling file sync. + +A separate background presentation40P01 involved a run that succeeded. The +retained log shows its coordinator→issue→run wait, but lacks PostgreSQL's full +opposing lock cycle. Its cause and relationship to other contention are unproved. +It is not evidence that the14 run failures came from a database deadlock. All +historical failures remain separate; no failed H4c/H5a arm is reused as H5b noise. + ## Fixed work and observation boundaries -Keep the five streaming cases and original local Stop case exactly as H4c: +Keep the five streaming event workloads and original local Stop case exactly as H4c: paced 1×128 at25ms; burst 1×16; burst10×128; burst100×512; existing-history 1000 foreground prelude deltas plus measured10×128; local Stop1×512 at100ms, clicked after at least three observed markers. Keep normal API/UI activation, @@ -74,14 +131,14 @@ status, errors and configured timeout. Its observed total duration is an independent conservative assessment of the original900s budget; hooks/teardown can be treated differently by the original timeout. It is not an exact original timer-expiration observation. A final candidate must have total duration<900s -in every arm and later pass one unchanged original H4c full invocation on the +in every arm and later pass one unchanged H4d default-mode full invocation on the same final SUT revision and full qualified artifact inventory. ## Complete evidence and eligibility H5 `observationComplete` means all six cases reached finite correct outcomes, -with every expected participant and foreground marker. It does not mean H4c -passed. Every remote run must succeed in native mode, retain every normalized +with every expected participant and foreground marker. It does not establish a +pass of the original deadlines. Every remote run must succeed in native mode, retain every normalized delta's complete payload/identity/order, have exactly one independently read `run.result.accepted`, and pass the existing reload/result/clock/lease checks. The100-run case must independently verify all51200 deltas. Stop keeps its @@ -110,7 +167,7 @@ not itself attest that the live filesystem stayed unchanged. | --- | --- | --- | | Fresh B0/B1 A/A arms | Known burst512 streaming120 miss; explicit conservative observed-duration900 miss | Complete finite correct observation; root-reviewed noise suitability | | B0→B1 cursor-only candidate | Only an exact registered known miss also present in the paired baseline: burst512 streaming120 or conservative observed-duration900 | No additional deadline failures; latency/control nonregression plus separate incremental SQL evidence; never claim SLA fixed | -| Final B2 candidate in either comparison | None | Every arm meets original streaming/control checks and conservative900 gate; loaded improvements and later unchanged H4c proof | +| Final B2 candidate in either comparison | None | Every arm meets original streaming/control checks and conservative900 gate; loaded improvements and later H4d proof | Only burst512's streaming miss is registered as a known workload SLA failure. New failures in other workloads require diagnosis. The existing Stop poll @@ -133,18 +190,23 @@ improve; other fixed latency/control metrics must not regress outside noise. Do not infer p99 from sparse observations. There is no new fixed percentage A/A cutoff: root must reject unsuitable noisy/nonstationary calibration explicitly. -Use `--observation-mode=h5-observe-original-sla` for the comparator. Candidate +Use `--observation-mode=h5b-observe-original-sla-independent-agents` for the comparator. Candidate comparisons also require `--candidate-policy=cursor-nonregression --expect=nonregression` or `--candidate-policy=final-b2`. Final comparisons expose `performancePassed` -separately; `acceptancePassed` remains false until `--original-h4c-proof` and -`--original-h4c-inventory` establish the later fresh original functional pass. -That proof is full, unprofiled, retry0, original900s, on the exact final SUT -revision/qualified inventory, with a new disposable company and a start time -after all candidate arms. Prior failed H4c attempts remain in the record. +separately; `acceptancePassed` remains false until `--original-h4d-proof` and +`--original-h4d-inventory` establish the later fresh H4d default-deadline functional pass. +That H4d proof is full, unprofiled, retry0, default mode with original120/180/900s +bounds, on the same frozen H5b harness source, nine-file measurement bytes and +independent-agent geometry, and the exact final SUT revision/qualified inventory. +It uses a new disposable company and starts after all candidate arms. Old7d/H4c +proofs and H5a observations cannot confirm this geometry. Prior failed H4c attempts remain in the record. +Confirmation reuses the measured arms' authority, physical Stop, local lease and +remote lease binding/settlement checks. The original30s polls keep their existing +ordering; no absolute click-to-acknowledgement deadline is added. Use the explicit H5 `--require-performance` gate for the two final comparison stages while that later proof is pending; this requires every performance and candidate original-SLA gate, but grants no final acceptance. After the one later -original H4c qualification, recompute both reports against their unchanged paired +H4d default-mode qualification, recompute both reports against their unchanged paired manifests with the proof and `--require-acceptance`. This re-evaluates retained evidence; it neither reuses an arm in a different comparison nor reruns failures. diff --git a/tests/e2e/fixtures/native-ingestion-h5.mjs b/tests/e2e/fixtures/native-ingestion-h5.mjs index c6242d74d42..564511d7a26 100644 --- a/tests/e2e/fixtures/native-ingestion-h5.mjs +++ b/tests/e2e/fixtures/native-ingestion-h5.mjs @@ -1,10 +1,14 @@ import assert from 'node:assert/strict'; import { createHash } from 'node:crypto'; -export const H5_MODE = 'h5-observe-original-sla'; -export const H5_PROTOCOL = 'v5-complete-observation-original-sla'; -export const H5_BASELINE_REVISIONS = Object.freeze({ B0: 'a36567f9fbf805ad72207d54f539516e3121741e', - B1: 'd8cd3be0aeed4f9fb93ecc3cf725f1d46d5a0310' }); +export const H5_MODE = 'h5b-observe-original-sla-independent-agents'; +export const H5_PROTOCOL = 'v5b-independent-agents-complete-observation-original-sla'; +export const H4D_PROTOCOL = 'v4d-independent-agents-original-deadlines'; +export const H5_TOPOLOGY = Object.freeze({ mode: 'one-normal-agent-per-concurrent-participant', + fullRemoteAgentPoolSize: 100, agentSetupConcurrency: 8, maxConcurrentRuns: 50, + poolReuse: 'only-across-sequential-cases' }); +export const H5_BASELINE_REVISIONS = Object.freeze({ B0: 'bac53ec3bfeaf6546e7aaeaee516487b30534654', + B1: '07061a601cdc20603c8c7a46f9ddbad7ac5f5939' }); export const ORIGINAL_BUDGETS = Object.freeze({ providerReadinessMs: 300_000, fullInvocationMs: 900_000, streamingMs: 120_000, terminalDrainageMs: 180_000, stopAcknowledgementMs: 30_000, remoteLeaseCleanupMs: 30_000 }); export const H5_LIMITS = Object.freeze({ streamingMs: 600_000, terminalDrainageMs: 180_000, fullInvocationMs: 1_200_000 }); @@ -18,18 +22,89 @@ export const H5_WORKLOADS = Object.freeze([ ]); export const H5_REGISTRATION = Object.freeze({ mode: H5_MODE, protocolVersion: H5_PROTOCOL, baselineRevisions: H5_BASELINE_REVISIONS, + topology: H5_TOPOLOGY, originalDeadlineProtocol: H4D_PROTOCOL, originalBudgets: ORIGINAL_BUDGETS, observationLimits: H5_LIMITS, workloads: H5_WORKLOADS, pairs: 5, order: 'baseline-first on even pair index; candidate-first on odd pair index', stages: ['B0 A/A', 'B1 A/A', 'B0→B1 nonregression', 'B1→B2 improvement', 'B0→B2 improvement'], knownBaselineMisses: ['burst512:streaming120', 'invocation:observed-total-duration900'], - baselineEligibility: 'Finite complete correct observation; SLA misses remain failures of H4c', + baselineEligibility: 'Finite complete correct observation; SLA misses remain failures of the H4d original deadlines', cursorCandidateEligibility: 'Nonregression; only the paired baseline exact registered known misses may be inherited', - finalCandidateEligibility: 'Every original SLA passes in every arm; independently assessed total duration <900s; later unchanged H4c full functional PASS required', + finalCandidateEligibility: 'Every original SLA passes in every arm; independently assessed total duration <900s; later H4d default-mode functional PASS on the same harness source and independent-agent geometry required', noise: 'Fresh B0 and B1 A/A frozen before any further product implementation; no old abort/profile samples', performance: 'Loaded p95 emission/intended-to-visible and terminal drainage improve beyond frozen noise in median and at least4/5 pairs; all other metrics nonregress', }); export const H5_WORKLOAD_SHA256 = createHash('sha256').update(JSON.stringify(H5_REGISTRATION)).digest('hex'); +/** Agent setup settles in full before any case activation; its time remains + * inside the owning invocation and outside the existing admission300s window. */ +export async function prepareParticipantAgents({ count, createAndReady, now = Date.now, onOutcome = _value => {} }) { + assert.ok(Number.isSafeInteger(count) && count > 0 && count <= H5_TOPOLOGY.fullRemoteAgentPoolSize); + const agents = new Array(count), failures = []; + const outcome = { count, concurrency: H5_TOPOLOGY.agentSetupConcurrency, startedAtMs: now(), settledAtMs: null, + complete: false, attempts: [] }; + onOutcome(outcome); + let next = 0; + const workers = await Promise.allSettled(Array.from({ length: Math.min(count, outcome.concurrency) }, async () => { + for (;;) { + const index = next++; if (index >= count) return; + const attempt = { index, startedAtMs: now(), settledAtMs: null, status: 'pending', agentId: null }; + outcome.attempts.push(attempt); onOutcome(outcome); + try { + agents[index] = await createAndReady(index); + assert.ok(typeof agents[index]?.id === 'string', 'Missing ready participant agent'); + attempt.agentId = agents[index].id; attempt.status = 'fulfilled'; + } catch (error) { + failures.push(error); attempt.status = 'rejected'; attempt.error = String(error); + } finally { attempt.settledAtMs = now(); onOutcome(outcome); } + } + })); + for (const worker of workers) if (worker.status === 'rejected') failures.push(worker.reason); + if (!failures.length) try { assert.equal(new Set(agents.map(agent => agent.id)).size, count, 'Participant agents must be distinct'); } + catch (error) { failures.push(error); } + outcome.settledAtMs = now(); outcome.complete = failures.length === 0; onOutcome(outcome); + if (failures.length) throw new AggregateError(failures, 'Participant agent setup failed; no case may activate'); + return agents; +} + +export function assertParticipantSetup(report) { + assert.deepEqual(report.agentTopology, { ...H5_TOPOLOGY, remoteAgentPoolSize: report.qualificationOnly ? 3 : 100 }); + const ids = report.remote?.agentIds; + assert.equal(ids?.length, report.agentTopology.remoteAgentPoolSize, 'Incomplete independent participant agent pool'); + assert.equal(new Set(ids).size, ids.length, 'Duplicate participant agent pool identity'); + const setup = report.remote.agentSetup; + assert.equal(setup?.complete, true, 'Agent setup must settle successfully before any case'); + assert.equal(setup.count, ids.length); assert.equal(setup.concurrency, H5_TOPOLOGY.agentSetupConcurrency); + assert.ok(Number.isSafeInteger(report.bodyStartedAtMs) && report.bodyStartedAtMs > 0); + assert.ok(Number.isSafeInteger(setup.startedAtMs) && setup.startedAtMs >= report.bodyStartedAtMs); + assert.ok(Number.isSafeInteger(setup.settledAtMs) && setup.settledAtMs >= setup.startedAtMs); + assert.equal(setup.attempts?.length, ids.length); + assert.deepEqual(setup.attempts.map(attempt => attempt.index).sort((a, b) => a - b), ids.map((_id, index) => index)); + const boundaries = []; + for (const attempt of setup.attempts) { + assert.equal(attempt.status, 'fulfilled', 'Rejected agent setup cannot qualify'); + assert.equal(attempt.agentId, ids[attempt.index]); + assert.ok(Number.isSafeInteger(attempt.startedAtMs) && attempt.startedAtMs >= setup.startedAtMs); + assert.ok(Number.isSafeInteger(attempt.settledAtMs) && attempt.settledAtMs >= attempt.startedAtMs && attempt.settledAtMs <= setup.settledAtMs); + if (attempt.settledAtMs > attempt.startedAtMs) boundaries.push([attempt.startedAtMs, 1], [attempt.settledAtMs, -1]); + } + let active = 0; + for (const [_time, delta] of boundaries.sort((a, b) => a[0] - b[0] || a[1] - b[1])) { + active += delta; assert.ok(active <= setup.concurrency, 'Agent setup exceeded the fixed concurrency'); + } + for (const item of report.cases) assert.ok(item.activations?.[0]?.startedAtMs >= setup.settledAtMs, + 'Agent setup must finish before every case activation'); +} + +export function assertParticipantTopology(report, result) { + const ids = report.remote?.agentIds; + if (!result.stop) { + const rows = result.controllerAuthority?.rows; + assert.equal(rows?.length, result.runs, 'Missing participant authority rows'); + assert.equal(new Set(rows.map(row => row.agent_id)).size, result.runs, 'Concurrent participants must use distinct normal agents'); + assert.ok(rows.every(row => ids.includes(row.agent_id)), 'Participant authority is outside the normal agent pool'); + } +} + export function h5Enabled(value) { assert.ok(value === undefined || value === H5_MODE, 'Explicit supported H5 observation mode required'); return value === H5_MODE; diff --git a/tests/e2e/native-event-ingestion.journey.ts b/tests/e2e/native-event-ingestion.journey.ts index 853549253db..14eb1474963 100644 --- a/tests/e2e/native-event-ingestion.journey.ts +++ b/tests/e2e/native-event-ingestion.journey.ts @@ -7,8 +7,8 @@ import { hostname } from "node:os"; import path from "node:path"; import { expect, test } from "@playwright/test"; import { json } from "./agent-chat.shared"; -import { h5Enabled, H5_MODE, H5_PROTOCOL, H5_LIMITS, H5_WORKLOAD_SHA256, ORIGINAL_BUDGETS, - observeStreaming, originalTerminal } from "./fixtures/native-ingestion-h5.mjs"; +import { h5Enabled, H5_MODE, H5_PROTOCOL, H4D_PROTOCOL, H5_TOPOLOGY, H5_LIMITS, H5_WORKLOAD_SHA256, ORIGINAL_BUDGETS, + observeStreaming, originalTerminal, prepareParticipantAgents, assertParticipantTopology } from "./fixtures/native-ingestion-h5.mjs"; type SeenMarker = { markerCode: string; index: number; visibleMs: number }; type Emission = { nonce: string; markerCode: string; index: number; intendedMs: number; emittedMs: number; receivedAtMs: number }; @@ -275,9 +275,10 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy const browserRecording = { trace: captureMode(info.project.use.trace), video: captureMode(info.project.use.video), screencast: captureMode(info.project.use.trace) !== "off" || captureMode(info.project.use.video) !== "off" }; const report: Record = { - protocolVersion: h5 ? H5_PROTOCOL : "v4-remote-streaming-original-local-stop-clock-stream", transport, + protocolVersion: h5 ? H5_PROTOCOL : H4D_PROTOCOL, transport, bodyStartedAtMs, + agentTopology: { ...H5_TOPOLOGY, remoteAgentPoolSize: Math.max(...cases.map(item => item.runs)) }, ...(h5 ? { observationMode: H5_MODE, workloadSha256: H5_WORKLOAD_SHA256, observationLimits: H5_LIMITS, - observationComplete: false, bodyStartedAtMs, actualPlaywrightTimeoutMs: info.timeout, + observationComplete: false, actualPlaywrightTimeoutMs: info.timeout, originalInvocationBudgetAssessment: "Requires independent owning Playwright observed total duration; conservative original900 assessment, not exact timeout expiration" } : {}), transportPolicy: { streaming: "sandbox-remote", operatorStop: "local", realCloudStop: "Separate required qualification; fake-plugin void receipts grant no remote Stop authority" }, @@ -315,7 +316,7 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy const protocolFiles = ["native-event-ingestion.journey.ts", "playwright-native-ingestion.config.ts", "fixtures/in-feed-codex.mjs", "fixtures/native-ingestion-bin/codex", "fixtures/composer-stop-codex-loader.mjs", "fixtures/native-ingestion-bootstrap.mjs", "fixtures/native-ingestion-remote-server.mjs", "fixtures/native-ingestion-protocol-qualification.mjs", - ...(h5 ? ["fixtures/native-ingestion-h5.mjs"] : [])]; + "fixtures/native-ingestion-h5.mjs"]; const protocolSources = await Promise.all(protocolFiles.map(async file => ({ file, bytes: await readFile(path.resolve(import.meta.dirname, file)) }))); report.measurementSha256 = createHash("sha256").update(Buffer.concat(protocolSources.map(source => source.bytes))).digest("hex"); const harnessRoot = path.resolve(import.meta.dirname, "../.."); @@ -357,9 +358,10 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy expect(localEnvironments).toHaveLength(1); const localEnvironment = localEnvironments[0]; expect(localEnvironment.driver).toBe("local"); - const agents: Array<{ id: string; name: string }> = []; - const agentCount = Math.ceil(Math.max(...cases.map(item => item.runs)) / 50); - for (let index = 0; index < agentCount; index++) { + const agentCount = Math.max(...cases.map(item => item.runs)); + const agents = await prepareParticipantAgents({ count: agentCount, + onOutcome: (value: unknown) => { (report.remote as any).agentSetup = value; }, + createAndReady: async (index: number) => { const agent = await json(await request.post(`/api/companies/${company.id}/agents`, { timeout: 120_000, data: { name: `Ingestion ${index + 1}`, role: "engineer", adapterType: "paperclip_runner", defaultEnvironmentId: remoteEnvironment.id, @@ -369,8 +371,8 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy } })); await expect.poll(async () => (await json(await request.get(`/api/agents/${agent.id}`))).lifecycleState, { timeout: 60_000 }).toBe("ready"); - agents.push(agent); - } + return agent; + } }) as Array<{ id: string; name: string }>; (report.remote as any).agentIds = agents.map(agent => agent.id); const controlAgent = await json(await request.post(`/api/companies/${company.id}/agents`, { timeout: 120_000, data: { name: "Ingestion local Stop control", role: "engineer", adapterType: "paperclip_runner", @@ -437,7 +439,7 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy register(backgroundToken, backgroundCode); const background = await json(await request.post(`/api/companies/${company.id}/issues`, { data: { title: `Ingestion background ${key} ${index}`, description: prompt(backgroundToken, backgroundCode, 0), - status: "backlog", assigneeAgentId: caseAgents[Math.floor(index / 50)].id, + status: "backlog", assigneeAgentId: caseAgents[index].id, } })); issueIds.push(background.id); issuedTokens.set(background.id, backgroundToken); @@ -516,6 +518,7 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy appRoot, companyId: company.id, environmentId: caseEnvironment.id, issueIds, transport: caseTransport, }); active.controllerAuthority = controllerAuthority; + assertParticipantTopology(report, { ...workload, stop, controllerAuthority }); const providerClocks = controllerAuthority.rows.map((row: any) => { const token = issuedTokens.get(row.issue_id)!, stream = state.streams.get(token)!; return { runId: row.run_id, nonce: token, launcher: stream.launcher, providerPid: stream.providerPid };