test: isolate ingestion streams by normal agent identity

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<string, any>();
+  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<string, any>();
-    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<string, any> | undefined;
+let originalH4dConfirmation: Record<string, any> | 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<string, any> = {
-  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<string, any> = {
   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");
 }

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

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 () => {

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

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.
 

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;

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<string, unknown> = {
-    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 };