Qualify remote ingestion streaming with original local Stop coverage

This commit is contained in:
Dotta committed 2026-11-03 17:06:16 -06:00
1 parent 2d962322ea
commit 5f5328952f
9 files changed
+477 -177

No files matched your search

+124 -29
View File
@@ -32,8 +32,10 @@ const sha256 = (value: unknown) => typeof value === "string" && /^[a-f0-9]{64}$/
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);
for (const pair of pairs) for (const arm of ["baseline", "candidate"] as const) {
const report = pair[arm];
assert.equal(report.protocolVersion, "v3-supported-remote-original-clock-stream", "Fresh browser acceptance requires the qualified v3 remote clock protocol");
assert.equal(report.transport, "sandbox-remote", "V3 requires the supported sandbox remote transport");
assert.equal(report.protocolVersion, "v4-remote-streaming-original-local-stop-clock-stream", "Fresh browser acceptance requires the qualified v4 mixed clock 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");
assert.deepEqual(report.resourceBudgets, { providerReadinessMs: 300_000, fullInvocationMs: 900_000,
streamingMs: 120_000, terminalDrainageMs: 180_000, stopAcknowledgementMs: 30_000, remoteLeaseCleanupMs: 30_000 },
"Keep the prospectively registered remote setup/resource budgets and existing streaming/control timeouts");
@@ -43,6 +45,22 @@ for (const pair of pairs) for (const arm of ["baseline", "candidate"] as const)
assert.equal(remote?.selection, "Normal plugin install + sandbox environment + agent defaultEnvironmentId; no runtime resolver override",
"Use normal API remote environment selection");
assert.ok(uuid(remote.environmentId) && uuid(report.companyId), "Missing actual remote company/environment identity");
assert.ok(remote.agentIds?.length > 0 && remote.agentIds.every(uuid), "Missing actual remote agent selection");
assert.deepEqual(Object.keys(remote.plugin.assetHashes ?? {}).sort(), ["lease-cleanup.js", "manifest.js", "plugin.js", "worker.js"]);
for (const [file, hash] of Object.entries(remote.plugin.assetHashes)) {
assert.ok(sha256(hash), "Missing actual provider plugin build fingerprint");
assert.equal(hash, first.baseline.remote.plugin.assetHashes[file], "Provider cleanup build changed across matched samples");
}
const local = report.local;
assert.equal(local?.transport, "local", "Missing normal local Stop runtime provenance");
assert.equal(local?.selection, "Normal local environment + separate control agent defaultEnvironmentId; existing test-only Codex leaf command fixture",
"Use normal local Stop environment selection");
assert.ok(uuid(local.environmentId) && uuid(local.controlAgentId) && local.environmentId !== remote.environmentId
&& !remote.agentIds.includes(local.controlAgentId), "Local Stop needs a separate normal control agent and environment");
for (const field of ["commandSha256", "loaderSha256"]) {
assert.ok(sha256(local[field]), "Missing original local provider command provenance");
assert.equal(local[field], first.baseline.local[field], "Local Stop fixture changed across matched samples");
}
assert.equal(remote.revision, report.sourceRevision);
assert.equal(remote.buildRevision, report.compiledBuildRevision);
assert.equal(remote.tls?.peerVerification, true, "Remote TLS peer verification is required");
@@ -85,6 +103,8 @@ for (const pair of pairs) for (const arm of ["baseline", "candidate"] as const)
assert.ok(result, `Missing fixed workload ${name}`);
const fixed = workloads.find(item => item.name === name)!;
for (const field of ["runs", "events", "intervalMs", "prelude", "stop"] as const) assert.equal(result[field], fixed[field], `Fixed workload changed: ${name}.${field}`);
assert.equal(result.transport, result.stop ? "local" : "sandbox-remote", "Wrong per-case transport; fake-plugin does not qualify remote Stop");
assert.equal(result.environmentId, result.stop ? local.environmentId : remote.environmentId, "Wrong per-case normal environment");
assert.equal(result.persistedAfterReload, true);
assert.equal(result.matchedVisibleClocks, result.visible?.length, "Every observed marker needs its original provider clock");
assert.ok(result.matchedVisibleClocks >= (result.stop ? 3 : result.events), "Missing foreground render/clock observations");
@@ -103,7 +123,8 @@ for (const pair of pairs) for (const arm of ["baseline", "candidate"] as const)
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 remote database controller authority proof");
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
@@ -115,36 +136,108 @@ for (const pair of pairs) for (const arm of ["baseline", "candidate"] as const)
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 remote task/lease UUID");
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, remote.environmentId, "Remote authority environment mismatch");
assert.equal(row.environment_id, result.environmentId, "Remote authority environment mismatch");
assert.equal(row.runtime_mode, "native");
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);
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(Number.isSafeInteger(row.process_identity_seq) && row.process_identity_seq > 0);
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);
}
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
&& (!result.stop || result.remoteLeasesCheckedAtMs >= result.stopAcknowledgedAtMs), "Verify remote leases after recorded timing boundaries");
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);
for (const clock of result.providerClocks) {
assert.equal(clock.launcher, result.stop ? "host-local" : "staged-remote", "Actual wrapper launcher does not match transport");
assert.ok(Number.isSafeInteger(clock.providerPid) && clock.providerPid > 1, "Missing actual provider PID");
const stream = result.clockEvidence?.find((stream: any) => stream.nonce === clock.nonce);
assert.equal(stream?.launcher, clock.launcher, "Clock hello launcher proof missing");
assert.equal(stream?.providerPid, clock.providerPid, "Clock hello provider PID proof missing");
assert.equal(result.durable.find((proof: any) => proof.runId === clock.runId)?.token, clock.nonce);
}
assert.equal(result.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");
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);
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(after.process_stopped_seq > 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 === "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");
&& 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 proof of result.durable) {
@@ -179,7 +272,8 @@ const measurements = names.flatMap(name => {
});
});
const result: Record<string, any> = {
scope: "Compiled-app deterministic-provider/browser acceptance; live-provider qualification remains separate",
scope: "Compiled-app remote streaming and original local Stop performance; additional real cloud Stop qualification is required separately",
transportPolicy: first.baseline.transportPolicy, additionalRealCloudStopQualificationRequired: true,
protocolVersion: first.baseline.protocolVersion, transport: first.baseline.transport,
resourceBudgets: first.baseline.resourceBudgets,
phase, expectation, pairCount: pairs.length, baselineRevision: first.baseline.sourceRevision,
@@ -193,8 +287,9 @@ if (phase === "compare") {
assert.ok(arg("noise-report"), "Comparison requires its frozen A/A --noise-report");
const noise = JSON.parse(await readFile(arg("noise-report")!, "utf8"));
assert.equal(noise.phase, "aa"); assert.equal(noise.complete, true);
assert.equal(noise.protocolVersion, result.protocolVersion, "Noise report uses a different remote protocol");
assert.equal(noise.protocolVersion, result.protocolVersion, "Noise report uses a different mixed protocol");
assert.equal(noise.transport, result.transport, "Noise report uses a different transport");
assert.deepEqual(noise.transportPolicy, result.transportPolicy, "Transport policy changed after A/A calibration");
assert.deepEqual(noise.resourceBudgets, result.resourceBudgets, "Resource budgets changed after A/A calibration");
assert.equal(noise.baselineRevision, result.baselineRevision); assert.equal(noise.measurementSha256, result.measurementSha256);
assert.equal(noise.gatesSha256, result.gatesSha256, "Acceptance gates changed after A/A calibration");
@@ -76,41 +76,73 @@ 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")}`;
return { protocolVersion: "v3-supported-remote-original-clock-stream", transport: "sandbox-remote", companyId,
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,
resourceBudgets: { providerReadinessMs: 300000, fullInvocationMs: 900000, streamingMs: 120000,
terminalDrainageMs: 180000, stopAcknowledgementMs: 30000, remoteLeaseCleanupMs: 30000 },
remote: { transport: "sandbox-remote", revision, buildRevision: revision, environmentId,
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],
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" }, tls: { peerVerification: true, caSha256: "e".repeat(64), serverCertificateSha256: "f".repeat(64) },
plugin: { pluginKey: "paperclip.fake-sandbox-provider", assetHashes: Object.fromEntries(["worker.js", "plugin.js", "manifest.js", "lease-cleanup.js"].map(file => [file, "5".repeat(64)])) },
tls: { peerVerification: true, caSha256: "e".repeat(64), serverCertificateSha256: "f".repeat(64) },
runner: { sha256: "d".repeat(64), metadata: { prpTransportModes: ["dial_wss"] } },
provider: { nodeSha256: "b".repeat(64), wrapperSha256: "c".repeat(64), fixtureSha256: "a".repeat(64) } },
fixtureSha256: "a".repeat(64), admissionProtocol: "sequential-provider-and-clock-readiness-before-common-epoch", setupReadinessPollingMs: 50, complete: true, cleanup: { complete: true, errors: [] }, qualificationOnly: false, diagnosticOnly: false, performanceAcceptanceEligible: true, browserRecording: { trace: "off", video: "off", screencast: false },
sourceRevision: revision, compiledBuildRevision: revision, serverArtifact: "compiled", serverEntrySha256: `${revision}-compiled`,
measurementSha256: "synthetic-fixed-fixture", compiledHotPathHashes: { append: "append", port: "port", runtime: "runtime" },
cases: workloads.map((workload, caseIndex) => ({ ...workload, stop: Boolean(workload.stop), persistedAfterReload: true,
epochMs: 1000, preludeProof: workload.prelude ? { individualDeltas: workload.prelude, uniqueSourceIdentities: workload.prelude, verifiedAtMs: 999 } : undefined,
settledAtMs: 1000 + latency, stopRequestedAtMs: workload.stop ? 1000 : undefined,
stopAcknowledgedAtMs: workload.stop ? 1000 + latency : undefined, remoteLeasesCheckedAtMs: 1001 + latency,
runIds: Array.from({ length: workload.runs }, (_, index) => id(1, caseIndex, index)),
controllerAuthority: { checkedBeforeMeasuredEpoch: true, checkedAtMs: 998, connectionClosedAtMs: 999,
observedReadOnly: "on", observedIsolation: "repeatable read", rows: Array.from({ length: workload.runs }, (_, index) => ({
run_id: id(1, caseIndex, index), issue_id: id(2, caseIndex, index), company_id: companyId,
lease_id: id(3, caseIndex, index), environment_id: environmentId, runtime_mode: "native", provider: "fake-plugin",
provider_lease_id: `fake-plugin://${id(1, caseIndex, index)}/fixture`, checkpoint_part: "header", checkpoint_header_bytes: 256,
})) },
remoteLeases: Array.from({ length: workload.runs }, (_, index) => ({ id: id(3, caseIndex, index), runId: id(1, caseIndex, index),
companyId, environmentId, provider: "fake-plugin", providerLeaseId: `fake-plugin://${id(1, caseIndex, index)}/fixture`,
status: "released", cleanupStatus: "success", releasedAt: "2026-11-03T00:00:00.000Z" })),
visible: Array(workload.stop ? 3 : workload.events).fill({}), matchedVisibleClocks: workload.stop ? 3 : workload.events,
firstVisibleAfterEmissionMs: latency, emissionToVisible: { medianMs: latency, p95Ms: latency },
interChunkGaps: { paintCount: workload.events, count: workload.events - 1, medianMs: latency, maxMs: latency },
intendedToVisible: { p95Ms: latency }, drainThroughTerminalMs: latency,
stopThroughTerminalMs: latency, stopThroughAcknowledgementMs: latency,
durable: Array.from({ length: workload.runs }, (_, index) => ({ runId: id(1, caseIndex, index), token: String(index), ordered: true,
uniqueSourceIdentities: workload.stop ? 3 : workload.events, individualDeltas: workload.stop ? 3 : workload.events, matchedSourceClocks: workload.stop ? 3 : workload.events })),
})) };
cases: workloads.map((workload, caseIndex) => {
const stop = Boolean(workload.stop), actualEnvironmentId = stop ? localEnvironmentId : environmentId;
const launcher = stop ? "host-local" : "staged-remote", pid = 20000 + caseIndex * 1000;
const snapshot = { checkedBeforeMeasuredEpoch: true, checkedAtMs: 998, connectionClosedAtMs: 999,
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,
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,
runner_instance_id: id(6, caseIndex, index), native_session_id: id(7, caseIndex, index), admitted_execution_workspace_id: id(8, caseIndex, index),
...(stop ? { process_pid: pid - 1, process_group_id: pid - 1, process_started_at: "2026-11-03T00:00:00.000Z",
process_identity_seq: 1, process_identity_source_event_id: null,
process_identity: { targetKind: "local", hostname: "fixture-host", processPid: pid - 1, processGroupId: pid - 1,
processStartedAt: "2026-11-03T00:00:00.000Z" } } : {}),
}));
const lease = { id: rows[0].lease_id, heartbeatRunId: rows[0].run_id, companyId, environmentId: localEnvironmentId,
provider: "local", providerLeaseId: null, status: "released", releasedAt: "2026-11-03T00:00:01.000Z", cleanupStatus: null };
const localProcesses = { runnerPid: pid - 1, processGroupId: pid - 1, providerPid: pid, hostname: "fixture-host",
checkedAtMs: 999, runnerAlive: true, groupAlive: true, providerAlive: true };
return { ...workload, stop, transport: stop ? "local" : "sandbox-remote", environmentId: actualEnvironmentId, persistedAfterReload: true,
epochMs: 1000, preludeProof: workload.prelude ? { individualDeltas: workload.prelude, uniqueSourceIdentities: workload.prelude, verifiedAtMs: 999 } : undefined,
settledAtMs: 1000 + latency, stopRequestedAtMs: stop ? 1000 : undefined,
stopAcknowledgedAtMs: stop ? 1000 + latency : undefined, remoteLeasesCheckedAtMs: stop ? undefined : 1001 + latency,
runIds: rows.map(row => row.run_id), controllerAuthority: { ...snapshot, rows,
...(stop ? { localJournal: { schema: "paperclip.runner.durable.control-plane-state.v1", sha256: "8".repeat(64), bytes: 512, checkedAtMs: 999,
identity: { runId: rows[0].run_id, runnerInstanceId: rows[0].runner_instance_id, normalizedSessionId: rows[0].native_session_id, environmentLeaseId: rows[0].admitted_execution_workspace_id } } } : {}) },
providerClocks: rows.map((row, index) => ({ runId: row.run_id, nonce: String(index), launcher, providerPid: pid + index })),
clockEvidence: rows.map((row, index) => ({ nonce: String(index), launcher, providerPid: pid + index })),
...(stop ? { localProcesses, localStopProof: { checkedAtMs: 1001 + latency, lease,
processes: { ...localProcesses, runnerAbsent: true, groupAbsent: true, providerAbsent: true },
database: { ...snapshot, checkedBeforeMeasuredEpoch: false, checkedAtMs: 1000 + latency, connectionClosedAtMs: 1000 + latency,
rows: [{ ...rows[0], process_pid: null, process_group_id: null, process_started_at: null, status: "cancelled", cancellation_dispatch_state: "acknowledged",
process_stopped_seq: 2, process_stopped_source_event_id: null, process_stopped: { processPid: pid - 1, processGroupId: pid - 1, processStartedAt: rows[0].process_started_at } }] } } }
: { remoteLeases: rows.map(row => ({ id: row.lease_id, runId: row.run_id, companyId, environmentId, provider: "fake-plugin", providerLeaseId: row.provider_lease_id,
status: "released", cleanupStatus: "success", releasedAt: "2026-11-03T00:00:00.000Z" })) }),
visible: Array(stop ? 3 : workload.events).fill({}), matchedVisibleClocks: stop ? 3 : workload.events,
firstVisibleAfterEmissionMs: latency, emissionToVisible: { medianMs: latency, p95Ms: latency },
interChunkGaps: { paintCount: workload.events, count: workload.events - 1, medianMs: latency, maxMs: latency },
intendedToVisible: { p95Ms: latency }, drainThroughTerminalMs: latency,
stopThroughTerminalMs: latency, stopThroughAcknowledgementMs: latency,
durable: rows.map((row, index) => ({ runId: row.run_id, token: String(index), ordered: true,
uniqueSourceIdentities: stop ? 3 : workload.events, individualDeltas: stop ? 3 : workload.events, matchedSourceClocks: stop ? 3 : workload.events })),
};
}) };
}
test("browser CLI rejects the three-pairs-above-noise counterexample and ineligible evidence", async () => {
@@ -146,9 +178,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; }, /v3 remote clock protocol/],
["old-local-protocol", report => { report.protocolVersion = "v2-compact-markers-original-clock-stream"; report.transport = "local"; }, /v3 remote clock protocol/],
["wrong-transport", report => { report.transport = "local"; }, /sandbox remote transport/],
["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/],
["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/],
["local-loaded-workload", report => { report.cases[3].transport = "local"; }, /Wrong per-case transport/],
["wrong-case-environment", report => { report.cases[5].environmentId = report.remote.environmentId; }, /normal environment/],
["missing-local-selection", report => { delete report.local; }, /local Stop runtime provenance/],
["shared-control-agent", report => { report.local.controlAgentId = report.remote.agentIds[0]; }, /separate normal control agent/],
["changed-local-command", report => { report.local.commandSha256 = "9".repeat(64); }, /Local Stop fixture changed/],
["changed-provider-cleanup", report => { report.remote.plugin.assetHashes["lease-cleanup.js"] = "9".repeat(64); }, /Provider cleanup build changed/],
["unregistered-resource-budget", report => { report.resourceBudgets.providerReadinessMs = 600000; }, /registered remote setup\/resource budgets/],
["missing-normal-remote", report => { delete report.remote; }, /normal remote runtime provenance/],
["missing-plugin", report => { delete report.remote.plugin; }, /plugin provenance/],
@@ -169,6 +210,28 @@ test("browser CLI rejects the three-pairs-above-noise counterexample and ineligi
["authority-wrong-environment", report => { report.cases[3].controllerAuthority.rows[99].environment_id = "other"; }, /environment mismatch/],
["authority-wrong-provider", report => { report.cases[3].controllerAuthority.rows[99].provider = "fake"; }, /fake-plugin/],
["authority-missing-checkpoint", report => { report.cases[3].controllerAuthority.rows[99].checkpoint_part = null; }, /checkpoint header/],
["local-wrong-checkpoint-authority", report => { report.cases[5].controllerAuthority.kind = "database-controller-checkpoint"; }, /Wrong checkpoint authority/],
["local-fake-database-header", report => { report.cases[5].controllerAuthority.rows[0].checkpoint_part = "header"; }, /host-file authority/],
["missing-local-journal", report => { delete report.cases[5].controllerAuthority.localJournal; }, /owned local journal authority/],
["local-journal-wrong-run", report => { report.cases[5].controllerAuthority.localJournal.identity.runId = "other"; }, /Local journal identity mismatch/],
["local-journal-wrong-workspace", report => { report.cases[5].controllerAuthority.rows[0].admitted_execution_workspace_id = report.remote.environmentId; }, /Local journal identity mismatch/],
["local-journal-bookkeeping-alias", report => { report.cases[5].controllerAuthority.localJournal.identity.environmentLeaseId = report.cases[5].controllerAuthority.rows[0].lease_id; }, /Local journal identity mismatch/],
["local-missing-admitted-workspace", report => { delete report.cases[5].controllerAuthority.rows[0].admitted_execution_workspace_id; }, /admitted local workspace binding/],
["local-journal-wrong-policy", report => { report.cases[5].controllerAuthority.localJournal.checkpointEventPolicy = "control-v1"; }, /original local raw transport/],
["missing-wrapper-provenance", report => { delete report.cases[3].providerClocks; }, /wrapper launcher and PID/],
["remote-used-local-wrapper", report => { report.cases[3].providerClocks[99].launcher = "host-local"; }, /wrapper launcher/],
["local-used-remote-wrapper", report => { report.cases[5].providerClocks[0].launcher = "staged-remote"; }, /wrapper launcher/],
["missing-hello-pid", report => { delete report.cases[3].clockEvidence[99].providerPid; }, /hello provider PID/],
["missing-local-identity-event", report => { delete report.cases[5].controllerAuthority.rows[0].process_identity; }, /targetKind|local/],
["provider-authored-local-identity", report => { report.cases[5].controllerAuthority.rows[0].process_identity_source_event_id = "provider-event"; }, /server-authored/],
["missing-local-physical-proof", report => { delete report.cases[5].localStopProof; }, /physical Stop/],
["local-provider-still-alive", report => { report.cases[5].localStopProof.processes.providerAbsent = false; }, /remains alive/],
["local-group-still-alive", report => { report.cases[5].localStopProof.processes.groupAbsent = false; }, /false.*true|false !== true/],
["local-database-ack-missing", report => { report.cases[5].localStopProof.database.rows[0].cancellation_dispatch_state = "pending"; }, /acknowledged/],
["local-cleared-identity-no-receipt", report => { delete report.cases[5].localStopProof.database.rows[0].process_stopped_seq; }, /server stop receipt/],
["local-provider-stop-receipt", report => { report.cases[5].localStopProof.database.rows[0].process_stopped_source_event_id = "provider"; }, /null/],
["local-lease-cleanup-failed", report => { report.cases[5].localStopProof.lease.cleanupStatus = "failed"; }, /Local Stop lease cleanup/],
["local-lease-wrong-binding", report => { report.cases[5].localStopProof.lease.heartbeatRunId = "other"; }, /other/],
["missing-lease-cleanup", report => { delete report.cases[3].remoteLeases; }, /lease cleanup proof/],
["lease-pending", report => { report.cases[3].remoteLeases[99].status = "pending_cleanup"; }, /cleanup remains unsettled/],
["lease-wrong-binding", report => { report.cases[3].remoteLeases[99].providerLeaseId = "other"; }, /fake-plugin/],
@@ -189,9 +252,10 @@ test("browser CLI rejects the three-pairs-above-noise counterexample and ineligi
assert.notEqual(rejected.status, 0); assert.match(rejected.stderr, message);
}
for (const [name, change, message] of [
["old-noise-protocol", report => { report.protocolVersion = "v2-compact-markers-original-clock-stream"; }, /different remote protocol/],
["old-noise-protocol", report => { report.protocolVersion = "v2-compact-markers-original-clock-stream"; }, /different mixed protocol/],
["local-noise-transport", report => { report.transport = "local"; }, /different transport/],
["changed-noise-budget", report => { report.resourceBudgets.providerReadinessMs++; }, /budgets changed after A\/A/],
["changed-noise-transport-policy", report => { report.transportPolicy.operatorStop = "sandbox-remote"; }, /Transport policy changed after A\/A/],
]) {
const changed = JSON.parse(await readFile(noise, "utf8")); change(changed);
const file = await save(name + ".json", changed);
+6 -1
View File
@@ -89,8 +89,13 @@ function openIngestionClock(plan, gate) {
closed = true;
try { request.end(); } catch (error) { fail(error); }
};
const launcher = process.env.PAPERCLIP_INGESTION_FIXTURE_LAUNCHER;
const providerPid = Number(process.env.PAPERCLIP_INGESTION_FIXTURE_PID);
if (!['host-local', 'staged-remote'].includes(launcher) || providerPid !== process.pid) {
throw new Error('Ingestion fixture requires its actual wrapper launcher and process identity');
}
write({ type: 'hello', nonce: plan.nonce, markerCode: plan.markerCode,
events: plan.events, intervalMs: plan.intervalMs });
events: plan.events, intervalMs: plan.intervalMs, launcher, providerPid });
return { write, close, done };
}
@@ -1,2 +1,4 @@
#!/bin/sh
export PAPERCLIP_INGESTION_FIXTURE_LAUNCHER=host-local
export PAPERCLIP_INGESTION_FIXTURE_PID=$$
exec node "$(dirname "$0")/../in-feed-codex.mjs" --ingestion-stream-fixture "$@"
@@ -7,6 +7,7 @@ import { createInterface } from 'node:readline';
import { fileURLToPath } from 'node:url';
const fixture = fileURLToPath(new URL('./in-feed-codex.mjs', import.meta.url));
const wrapper = fileURLToPath(new URL('./native-ingestion-bin/codex', import.meta.url));
const output = process.argv[2];
assert.ok(output, 'Pass an owned output JSON path');
const reports = [];
@@ -63,7 +64,7 @@ async function run({ name, count, intervalMs = 0, stopAfter, clockStatus = 200,
await new Promise(resolve => server.listen(0, '127.0.0.1', resolve));
const gate = `http://127.0.0.1:${server.address().port}/native-ingestion-gate/${key}`;
const send = value => child.stdin.write(JSON.stringify(value) + '\n');
child = spawn(process.execPath, [fixture, '--ingestion-stream-fixture'], { stdio: ['pipe', 'pipe', 'pipe'] });
child = spawn(wrapper, [], { stdio: ['pipe', 'pipe', 'pipe'] });
child.stderr.on('data', value => { stderr += value; });
const watchdog = setTimeout(() => fail(new Error(`${name}: qualification watchdog`)), 10_000);
createInterface({ input: child.stdout }).on('line', line => {
@@ -104,7 +105,8 @@ async function run({ name, count, intervalMs = 0, stopAfter, clockStatus = 200,
if (failure) throw failure;
assert.equal(exit.code, clockStatus === 200 ? 0 : 1, stderr);
assert.equal(clockPosts, 1); assert.equal(gateGets, 1);
assert.deepEqual(hello, { type: 'hello', nonce, markerCode, events: count, intervalMs });
assert.deepEqual(hello, { type: 'hello', nonce, markerCode, events: count, intervalMs,
launcher: 'host-local', providerPid: child.pid });
assert.equal(emissions.length, deltas.length);
assert.ok(deltas.length > 0);
for (let index = 0; index < emissions.length; index++) {
@@ -127,6 +129,7 @@ async function run({ name, count, intervalMs = 0, stopAfter, clockStatus = 200,
}
if (clockStatus !== 200) assert.match(stderr, /Ingestion fixture clock .*HTTP 503/);
reports.push({ name, exit, count, emitted: deltas.length, clockPosts, clockRecords: clockRecords.length,
launcher: hello.launcher, providerPid: hello.providerPid,
completedTextCharacters: completedItem?.text.length, terminal: terminal.status,
uploadHeldUntilAfterTerminal: true, stopAcknowledgedBeforeUploadResponse: stopAfter ? true : undefined,
explicitClockError: clockStatus !== 200 ? stderr.trim() : undefined });
@@ -11,9 +11,10 @@ import { createServer } from 'node:tls';
import { pathToFileURL } from 'node:url';
assert.equal(process.env.NODE_ENV, 'test');
assert.equal(process.env.PAPERCLIP_INGESTION_TRANSPORT, 'sandbox-remote');
assert.equal(process.env.PAPERCLIP_INGESTION_TRANSPORT, 'sandbox-remote-with-local-stop');
assert.ok(process.env.PAPERCLIP_HOME && process.env.PAPERCLIP_CONFIG);
assert.ok(!process.env.DATABASE_URL, 'Remote fixture requires its disposable embedded database');
assert.ok(!process.env.PAPERCLIP_RUNNER_STATE_DIR, 'H4 uses the normal owned instance journal directory');
assert.ok(path.resolve(process.env.PAPERCLIP_CONFIG).startsWith(`${path.resolve(process.env.PAPERCLIP_HOME)}${path.sep}`));
const ownedHome = await realpath(process.env.PAPERCLIP_HOME);
const ownedConfiguration = await realpath(process.env.PAPERCLIP_CONFIG);
@@ -52,7 +53,7 @@ execFileSync('openssl', ['verify', '-CAfile', ca, cert], { stdio: 'ignore' });
const fixture = path.resolve(import.meta.dirname, 'in-feed-codex.mjs');
const wrapper = path.join(directory, 'codex');
const quote = value => `'${value.replaceAll("'", "'\\''")}'`;
await writeFile(wrapper, `#!/bin/sh\nexec ${quote(process.execPath)} ${quote(fixture)} --ingestion-stream-fixture "$@"\n`, { mode: 0o700 });
await writeFile(wrapper, `#!/bin/sh\nexport PAPERCLIP_INGESTION_FIXTURE_LAUNCHER=staged-remote\nexport PAPERCLIP_INGESTION_FIXTURE_PID=$$\nexec ${quote(process.execPath)} ${quote(fixture)} --ingestion-stream-fixture "$@"\n`, { mode: 0o700 });
const runner = process.env.PAPERCLIP_RUNNER_REMOTE_BINARY_PATH
?? path.join(appRoot, 'packages/paperclip-runner/runner/target/release/paperclip-runnerd');
const runnerMetadata = JSON.parse(execFileSync(runner, ['--build-metadata'], { encoding: 'utf8' }));
+72 -75
View File
@@ -1,24 +1,27 @@
# Supported remote ingestion fixture
# Remote streaming and original local Stop fixture (H4)
This test assembly selects the production remote execution path through a
normal plugin installation, sandbox environment and agent `defaultEnvironmentId`.
The first-party `@paperclipai/plugin-fake-sandbox` plugin executes separate OS
processes in private temporary directories on the **same physical host**. It
simulates provisioning; it does not simulate the Runner, PRP connection,
PostgreSQL append path, WebSocket delivery or browser.
The five streaming workloads select the production remote execution path through
normal plugin installation, a sandbox environment and agent `defaultEnvironmentId`.
The first-party `@paperclipai/plugin-fake-sandbox` executes separate OS processes
in private temporary directories on the **same physical host**. Provisioning is
simulated; Runner, authenticated PRP, PostgreSQL ingestion, live WebSocket delivery
and the shipped browser UI are real.
Use a Linux host. Current remote process supervision requires a Linux boot ID
and process start ticks; matching a Darwin daemon's binary architecture alone
does not qualify a Darwin remote environment. The fixture does not weaken this
identity check.
The unchanged 512-event, 100-ms operator Stop workload uses the instance's normal
local environment and a separate control agent. This retains the original local
performance coverage. It does not qualify remote Stop. **A separate real Daytona
Stop journey is an additional required acceptance gate**, with the shared Stop
acknowledgement, a provider-bound termination receipt and independent provider SDK
state. A fake-plugin void return never supplies that authority.
Build and independently install the application normally. The server stamp must
equal the application's exact source revision. The application must include the
built fake-sandbox plugin and matching native Runner daemon. The separate H3
checkout supplies driver tooling and fixture files only.
Use Linux. Normal remote supervision requires its boot ID and process start ticks;
a Darwin artifact alone does not satisfy that contract. Each measured app needs
its own frozen source, locked dependencies, matching compiled server stamp, built
fake-sandbox plugin including cleanup repair `b9e4679e39c4cbe0d0c6b64f51487f011d28fb9d`,
and matching native daemon. H4 owns separate driver tooling and fixtures.
```sh
PAPERCLIP_INGESTION_TRANSPORT=sandbox-remote \
PAPERCLIP_INGESTION_TRANSPORT=sandbox-remote-with-local-stop \
PAPERCLIP_INGESTION_SMOKE=1 \
PAPERCLIP_INGESTION_APP_ROOT="$APP_ROOT" \
PAPERCLIP_INGESTION_EVIDENCE_DIR="$NEW_EVIDENCE_DIR" \
@@ -27,68 +30,62 @@ node node_modules/@playwright/test/cli.js test \
--config tests/e2e/playwright-native-ingestion.config.ts
```
Run this from the test-driver checkout. Smoke emits the original 16-event case
and exercises Stop separately. It is functional qualification, never performance
acceptance. Omit the smoke flag only after the complete protocol, application
revisions, paired-run commands and acceptance rules have been frozen.
Run from the driver checkout with `DATABASE_URL` and `PAPERCLIP_RUNNER_STATE_DIR`
unset. Smoke runs 16 remote events and the original local Stop case. It qualifies
functionality only. Full 100-run readiness and the frozen protocol must pass before
A/A; five alternating matched pairs and all existing noise/latency/throughput gates
still apply. Omit the smoke flag only after the lead freezes exact revisions,
protocol hashes and commands. Do not mix H3 or the failed local protocol into H4.
The test-only server assembly:
The assembly generates a private two-day CA and signed server certificate with
`CA:FALSE`, server authentication and loopback subject alternative names, then
fronts the compiled HTTP app with verified TLS. It supplies normal
`PAPERCLIP_RUNNER_PUBLIC_URL`, `PAPERCLIP_RUNNER_CA_BUNDLE_PATH`,
`PAPERCLIP_RUNNER_REMOTE_BINARY_PATH` and `PAPERCLIP_RUNNER_REMOTE_CODEX_PATH`.
Production code retains artifact staging, version/hash verification,
authentication and process ownership. The remote wrapper uses absolute Node and
fixture paths; the normal sandbox environment includes Node's directory in PATH.
Provenance hashes the daemon, Node, wrappers, fixture, certificates and built
plugin assets, without private keys or checkpoint contents.
- Generates a private two-day CA and a signed server certificate with
`CA:FALSE`, server authentication usage, and `127.0.0.1`/`localhost` subject
alternative names, and terminates verified TLS in front of
the compiled HTTP server using the existing Runner reverse-proxy contract.
- Sets the normal `PAPERCLIP_RUNNER_PUBLIC_URL`,
`PAPERCLIP_RUNNER_CA_BUNDLE_PATH`, `PAPERCLIP_RUNNER_REMOTE_BINARY_PATH` and
`PAPERCLIP_RUNNER_REMOTE_CODEX_PATH` inputs. Artifact staging, version checks,
checksum checks, transport authentication and process identity remain owned
by production code.
- Stages a deterministic Codex wrapper with an absolute Node executable and
absolute fixture path, because staging relocates the wrapper into the remote
runtime. Provenance records the wrapper, Node, provider fixture, daemon and CA
hashes. Private keys are not included in provenance.
- Gives the normal environment an explicit `PATH` containing the recorded Node
directory and fixed system directories. The fake provider's default Unix PATH
does not contain Homebrew Node; this does not override a command resolver.
The existing test-only local command loader changes one Codex executable leaf.
Normal remote staging passes its explicit `codexCommand`, which short-circuits
that resolver. Each actual wrapper exports its own launcher identity and PID
before exec; the provider's pre-epoch clock hello validates the PID against
`process.pid`. The strict observer requires `staged-remote` for every streaming
participant and `host-local` for Stop. Emission timestamps, normalized payloads,
event counts, compact marker bounds and the original two-frame browser observer
remain unchanged. The standalone protocol qualification verifies the new hello
and the unchanged held-clock Stop and terminal behavior.
Remote mode removes the local Codex command-resolver test hook. It never forces
`controllerPersistence`, changes local raw journaling or disables certificate
verification. Before releasing each common emission epoch, a separate raw
PostgreSQL connection verifies the actual run's company/environment lease and
database controller checkpoint header in a read-only repeatable-read snapshot.
The connection reads the actual owned postmaster port, closes before the epoch,
and neither inserts fixture outcomes nor changes state. Lease verification after
timed settlement also uses the normal API.
Before every common epoch, an independent READ ONLY REPEATABLE READ PostgreSQL
snapshot binds every actual run/company/agent/environment/lease. Remote cases
require database controller checkpoint headers. Local Stop instead verifies the
actual owned host-file journal's identity and hash, its original raw-event policy,
and server-authored process identity. Its PRP environment identity binds to the
actual admitted `binding.executionWorkspaceId` read from the run snapshot;
the normal environment bookkeeping lease is validated separately. Only scalar metadata is recorded. The
connection closes before emission. No fixture outcomes are inserted and no
controller persistence policy is overridden.
All existing event counts, rates, marker clocks, payload caps, independent delta
checks, reload, Stop and timing gates remain. Before any Linux remote or paired
timings, the lead registered **300 seconds** for shared pre-epoch remote setup
and **900 seconds** for the whole test, allowing normal plugin/daemon staging.
The local failed protocol retains its original 180/600-second limits. Streaming,
terminal drainage and Stop acknowledgement keep their original 120/180/30-second
timeouts. Post-timing lease cleanup has a separate 30-second verification bound.
Reports record these exact values and the comparator rejects other budgets.
Remote setup and the complete 100-run workload still require qualification before
paired measurements; these resource allowances establish no latency success.
After terminal and Stop timing stamps are retained, normal lease APIs verify
cleanup. Local Stop additionally checks cancelled/acknowledged database state,
matching process ownership and independent runner PID, process-group and provider
PID absence. Cleared mutable identities require the actual server-authored local
process-stop receipt. Remote streaming requires released/expired/failed leases
with successful cleanup; those void provider responses make no remote Stop claim.
Two excluded Mac setup attempts are retained locally:
Prospectively registered budgets stay **300 seconds** for shared pre-epoch
readiness and **900 seconds** for the complete test. Streaming, terminal drainage
and Stop acknowledgement remain **120/180/30 seconds**, with **30 seconds** for
post-timing lease verification. These are resource bounds, not latency acceptance.
The comparator preserves all fixed workloads, four-of-five beyond-noise wins,
sparse-stream, paint-gap, Stop and final-drainage gates, and rejects wrong transport
or missing authority/provenance.
- `native-ingestion-remote-smoke-ahjnfjv3`: normal plugin/environment installation
passed; agent verification failed before a provider turn. The launch shell's
plain Node command was independently shown to exit 127 under the fake
provider's default PATH. The API exposed `runner_remote_process_launch_failed`.
- `native-ingestion-remote-smoke-atl7nu5j`: explicit normal environment PATH fixed
that missing executable; verification still failed before a provider turn.
Static process supervision requires Linux `/proc` fingerprints, so no further
Mac attempts or identity bypass were made. The exact second runtime cause is
not promoted beyond the retained lifecycle failure.
Both original reports remain failures. Their owned app, TLS and actual
PostgreSQL ports were independently checked closed, known lease processes were
absent, and only their recorded temporary homes were removed. No Linux native
turn has yet passed this fixture at the time of this initial recipe.
The exact certificate-generation block passed a separate local TLS check:
the matching CA authorized the connection, an untrusted connection failed, and
an incorrect hostname failed with `ERR_TLS_CERT_ALTNAME_INVALID`. This qualifies
certificate assembly only. It does not qualify the remote Runner or any timing.
H3's first Linux attempt preserved all 16 deltas and committed native finalization,
but correctly failed real provider cleanup on sealed `0555` context directories;
Stop was not reached. That failed evidence remains unchanged. The isolated cleanup
repair has real-filesystem red/green tests and preserves void receipts. **H4 runtime,
100-run readiness, A/A, comparisons and separate real cloud Stop remain pending**
until their fresh evidence passes; source checks alone qualify none of them.
+166 -32
View File
@@ -2,7 +2,8 @@ import { createServer, type ServerResponse } from "node:http";
import { randomUUID, createHash } from "node:crypto";
import { execFileSync } from "node:child_process";
import { createRequire } from "node:module";
import { readFile, realpath } from "node:fs/promises";
import { lstat, readFile, readdir, realpath } from "node:fs/promises";
import { hostname } from "node:os";
import path from "node:path";
import { expect, test } from "@playwright/test";
import { json } from "./agent-chat.shared";
@@ -12,12 +13,13 @@ type Emission = { nonce: string; markerCode: string; index: number; intendedMs:
type Visible = SeenMarker & Emission;
type ClockStream = { nonce: string; markerCode: string; opened: boolean; hello: boolean; ended: boolean; aborted: boolean;
partialBytes: number; records: Map<number, Emission>; error?: string;
expectedLauncher: "host-local" | "staged-remote"; launcher?: string; providerPid?: number;
providerReadyAtMs?: number; clockHelloAtMs?: number; readinessObservedAtMs?: number };
type ClockCase = { events: number; intervalMs: number; epochMs?: number; streams: Map<string, ClockStream>; errors: string[] };
type Case = { name: string; runs: number; events: number; intervalMs: number; prelude: number };
const smoke = process.env.PAPERCLIP_INGESTION_SMOKE === "1";
const transport = process.env.PAPERCLIP_INGESTION_TRANSPORT ?? "local";
const providerReadinessMs = transport === "sandbox-remote" ? 300_000 : 180_000;
const transport = process.env.PAPERCLIP_INGESTION_TRANSPORT ?? "sandbox-remote-with-local-stop";
const providerReadinessMs = 300_000;
const cases: Case[] = smoke ? [{ name: "qualification", runs: 1, events: 16, intervalMs: 25, prelude: 0 }] : [
{ name: "paced", runs: 1, events: 128, intervalMs: 25, prelude: 0 },
{ name: "burst16", runs: 1, events: 16, intervalMs: 0, prelude: 0 },
@@ -31,8 +33,9 @@ const stats = (values: number[]) => {
...(values.length >= 100 ? { p95Ms: sorted[Math.ceil(sorted.length * 0.95) - 1] } : {}) };
};
async function remoteControllerAuthority(input: {
appRoot: string; companyId: string; environmentId: string; issueIds: string[];
async function runtimeAuthority(input: {
appRoot: string; companyId: string; environmentId: string; issueIds: string[]; transport: "local" | "sandbox-remote";
afterStop?: boolean;
}) {
const configuration = process.env.PAPERCLIP_E2E_SERVER_CONFIG;
if (!configuration) throw new Error("Missing owned server configuration");
@@ -56,13 +59,24 @@ async function remoteControllerAuthority(input: {
current_setting('transaction_isolation') as isolation`;
expect(settings[0].read_only).toBe("on");
expect(settings[0].isolation).toBe("repeatable read");
const rows = await tx`select r.id as run_id, r.company_id, r.native_issue_id as issue_id, r.runtime_mode,
const rows = await tx`select r.id as run_id, r.company_id, r.agent_id, r.native_issue_id as issue_id, r.runtime_mode,
r.native_session_id, r.runner_instance_id, r.status, r.process_pid, r.process_group_id, r.process_started_at,
r.runner_profile_json->'nativeExecutionInput'->'binding'->>'executionWorkspaceId' as admitted_execution_workspace_id,
r.result_json->'nativeCancellation'->>'dispatchState' as cancellation_dispatch_state,
i.seq as process_identity_seq, i.payload as process_identity, i.source_event_id as process_identity_source_event_id,
s.seq as process_stopped_seq, s.payload as process_stopped, s.source_event_id as process_stopped_source_event_id,
l.id as lease_id, l.environment_id, l.provider, l.provider_lease_id,
c.part as checkpoint_part, octet_length(c.material::text) as checkpoint_header_bytes
from heartbeat_runs r
join environment_leases l on l.heartbeat_run_id = r.id and l.company_id = r.company_id
left join native_runner_checkpoints c on c.run_id = r.id and c.company_id = r.company_id
and c.issue_id = r.native_issue_id and c.part = 'header'
left join lateral (select seq, payload, source_event_id from heartbeat_run_events e where e.run_id = r.id
and e.company_id = r.company_id and e.event_type = 'native.process_identity_recorded'
and e.source_event_id is null order by seq desc limit 1) i on true
left join lateral (select seq, payload, source_event_id from heartbeat_run_events e where e.run_id = r.id
and e.company_id = r.company_id and e.event_type = 'native.local_process_stopped'
and e.source_event_id is null order by seq desc limit 1) s on true
where r.company_id = ${input.companyId} and r.native_issue_id = any(${input.issueIds}::uuid[])`;
expect(rows).toHaveLength(input.issueIds.length);
expect(new Set(rows.map((row: any) => row.issue_id)).size).toBe(input.issueIds.length);
@@ -70,18 +84,65 @@ async function remoteControllerAuthority(input: {
expect(row.company_id).toBe(input.companyId);
expect(row.runtime_mode).toBe("native");
expect(row.environment_id).toBe(input.environmentId);
expect(row.provider).toBe("fake-plugin");
expect(row.provider_lease_id).toMatch(/^fake-plugin:\/\//);
expect(row.checkpoint_part).toBe("header");
expect(Number(row.checkpoint_header_bytes)).toBeGreaterThan(0);
if (input.transport === "sandbox-remote") {
expect(row.provider).toBe("fake-plugin");
expect(row.provider_lease_id).toMatch(/^fake-plugin:\/\//);
expect(row.checkpoint_part).toBe("header");
expect(Number(row.checkpoint_header_bytes)).toBeGreaterThan(0);
} else {
expect(row.provider).toBe("local"); expect(row.provider_lease_id).toBeNull();
expect(row.checkpoint_part, "Local raw transport keeps its existing host-file journal authority").toBeNull();
expect(row.process_identity?.targetKind).toBe("local"); expect(row.process_identity?.hostname).toBe(hostname());
expect(row.process_identity_source_event_id).toBeNull();
expect(Number(row.process_identity_seq)).toBeGreaterThan(0);
expect(row.admitted_execution_workspace_id).toMatch(/^[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}$/i);
if (row.process_started_at) expect(row.process_identity.processStartedAt).toBe(new Date(row.process_started_at).toISOString());
}
}
return { checkedBeforeMeasuredEpoch: true, checkedAtMs: Date.now(), transaction: "READ ONLY REPEATABLE READ",
return { kind: input.transport === "local" ? "host-file-journal" : "database-controller-checkpoint",
checkedBeforeMeasuredEpoch: !input.afterStop, checkedAtMs: Date.now(), transaction: "READ ONLY REPEATABLE READ",
observedReadOnly: settings[0].read_only, observedIsolation: settings[0].isolation, rows };
});
} finally { await sql.end({ timeout: 5 }); }
return { ...proof, connectionClosedAtMs: Date.now() };
const connectionClosedAtMs = Date.now();
let localJournal;
if (input.transport === "local" && !input.afterStop) {
expect(input.issueIds).toHaveLength(1);
const base = await realpath(path.join(path.dirname(configuration), "runtime/paperclip-runner/durable-sessions"));
expect(base.startsWith(`${ownedHome}${path.sep}`)).toBe(true);
const directories = (await readdir(base, { withFileTypes: true })).filter(entry => entry.isDirectory() && /^[a-f0-9]{64}$/.test(entry.name));
expect(directories.length, "Bound the owned fixture journal metadata search").toBeLessThanOrEqual(256);
for (const directory of directories) {
const file = path.join(base, directory.name, "control-plane/control-plane-state.json");
const metadata = await lstat(file).catch((error: NodeJS.ErrnoException) => {
if (error.code === "ENOENT") return null; throw error;
});
if (!metadata) continue;
expect(metadata.isFile() && !metadata.isSymbolicLink() && metadata.size <= 16 * 1024 * 1024).toBe(true);
expect((await realpath(file)).startsWith(`${base}${path.sep}`)).toBe(true);
const bytes = await readFile(file), state = JSON.parse(bytes.toString("utf8"));
if (state.identity?.runId !== proof!.rows[0].run_id) continue;
expect(localJournal, "Exactly one owned local journal must match the actual run").toBeUndefined();
expect(state.schema).toBe("paperclip.runner.durable.control-plane-state.v1");
expect(state.checkpointEventPolicy).not.toBe("control-v1");
expect(state.identity.runnerInstanceId).toBe(proof!.rows[0].runner_instance_id);
expect(state.identity.normalizedSessionId).toBe(proof!.rows[0].native_session_id);
expect(state.identity.environmentLeaseId).toBe(proof!.rows[0].admitted_execution_workspace_id);
localJournal = { file, sha256: createHash("sha256").update(bytes).digest("hex"), bytes: bytes.length,
schema: state.schema, identity: Object.fromEntries(["runnerInstanceId", "environmentLeaseId", "runId", "normalizedSessionId", "turnId", "itemId"]
.map(key => [key, state.identity[key]])), checkpointEventPolicy: state.checkpointEventPolicy, checkedAtMs: Date.now() };
}
expect(localJournal, "Missing actual owned local raw journal").toBeTruthy();
}
return { ...proof, connectionClosedAtMs, localJournal };
}
const processAbsent = (pid: number) => {
expect(Number.isSafeInteger(pid) && Math.abs(pid) > 1).toBe(true);
try { process.kill(pid, 0); return false; }
catch (error) { if ((error as NodeJS.ErrnoException).code === "ESRCH") return true; throw error; }
};
test("native ingestion reaches the shipped UI promptly and survives reload", async ({ page, request }, info) => {
const pending = new Map<string, Map<string, ServerResponse>>();
const clocks = new Map<string, ClockCase>();
@@ -116,7 +177,10 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy
const value = JSON.parse(text);
if (value.nonce !== participant || value.markerCode !== stream.markerCode) throw new Error("Clock identity mismatch");
if (!stream.hello) {
if (value.type !== "hello" || value.events !== state.events || value.intervalMs !== state.intervalMs) throw new Error("Clock hello mismatch");
if (value.type !== "hello" || value.events !== state.events || value.intervalMs !== state.intervalMs
|| value.launcher !== stream.expectedLauncher || !Number.isSafeInteger(value.providerPid) || value.providerPid <= 1
|| Object.keys(value).sort().join(",") !== "events,intervalMs,launcher,markerCode,nonce,providerPid,type") throw new Error("Clock hello mismatch");
stream.launcher = value.launcher; stream.providerPid = value.providerPid;
stream.hello = true; stream.clockHelloAtMs = Date.now(); return;
}
if (value.type !== "emission" || !Number.isSafeInteger(value.index) || value.index < 0 || value.index >= state.events
@@ -201,8 +265,9 @@ 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: transport === "sandbox-remote" ? "v3-supported-remote-original-clock-stream" : "v2-compact-markers-original-clock-stream",
transport,
protocolVersion: "v4-remote-streaming-original-local-stop-clock-stream", transport,
transportPolicy: { streaming: "sandbox-remote", operatorStop: "local",
realCloudStop: "Separate required qualification; fake-plugin void receipts grant no remote Stop authority" },
resourceBudgets: { providerReadinessMs, fullInvocationMs: info.timeout, streamingMs: 120_000,
terminalDrainageMs: 180_000, stopAcknowledgementMs: 30_000, remoteLeaseCleanupMs: 30_000 },
admissionProtocol: "sequential-provider-and-clock-readiness-before-common-epoch",
@@ -235,7 +300,7 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy
report.fixtureSha256 = createHash("sha256").update(await readFile(path.resolve(import.meta.dirname, "fixtures/in-feed-codex.mjs"))).digest("hex");
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-remote-server.mjs", "fixtures/native-ingestion-protocol-qualification.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, "../..");
@@ -244,7 +309,7 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy
sourceStatus: execFileSync("git", ["status", "--porcelain", "--", "tests/e2e"], { cwd: harnessRoot, encoding: "utf8" }).trim(),
sourceHashes: Object.fromEntries(protocolSources.map(source => [source.file, createHash("sha256").update(source.bytes).digest("hex")])),
driverDependencySource: process.env.PAPERCLIP_INGESTION_DRIVER_DEPENDENCIES
?? "Independent locked H3 driver install; every measured app owns separate dependencies and compiled output",
?? "Independent locked H4 driver install; every measured app owns separate dependencies and compiled output",
driverLockSha256: createHash("sha256").update(await readFile(path.join(harnessRoot, "pnpm-lock.yaml"))).digest("hex"),
scope: "Separate browser harness and driver-only tooling; every measured app uses its own frozen clean source and matching compiled stamp",
};
@@ -252,8 +317,8 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy
await json(await request.patch("/api/instance/settings/experimental", { data: { enableNativeRunner: true, enableWorktreeRunExecution: true } }));
await json(await request.post(`/api/companies/${company.id}/user-secret-definitions`, { data: { key: "OPENAI_API_KEY", name: "Fixture credential" } }));
await json(await request.post(`/api/companies/${company.id}/me/user-secrets`, { data: { definitionKey: "OPENAI_API_KEY", value: "not-a-real-model-key" } }));
let remoteEnvironment: { id: string } | undefined;
if (transport === "sandbox-remote") {
let remoteEnvironment: { id: string };
{
const pluginPath = path.join(appRoot, "packages/plugins/paperclip-plugin-fake-sandbox");
const plugin = await json(await request.post("/api/plugins/install", { timeout: 120_000,
data: { packageName: pluginPath, isLocalPath: true } }));
@@ -269,13 +334,20 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy
report.remote = { ...remoteProvenance, plugin: { id: plugin.id, pluginKey: plugin.pluginKey, path: pluginPath },
environmentId: remoteEnvironment!.id,
selection: "Normal plugin install + sandbox environment + agent defaultEnvironmentId; no runtime resolver override" };
(report.remote as any).plugin.assetHashes = Object.fromEntries(await Promise.all(
["worker.js", "plugin.js", "manifest.js", "lease-cleanup.js"].map(async file => [file,
createHash("sha256").update(await readFile(path.join(pluginPath, "dist", file))).digest("hex")])));
}
const agents = [];
const localEnvironments = await json(await request.get(`/api/companies/${company.id}/environments?driver=local`));
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 agent = await json(await request.post(`/api/companies/${company.id}/agents`, { timeout: 120_000, data: {
name: `Ingestion ${index + 1}`, role: "engineer", adapterType: "paperclip_runner",
...(remoteEnvironment ? { defaultEnvironmentId: remoteEnvironment.id } : {}),
defaultEnvironmentId: remoteEnvironment.id,
adapterConfig: { provider: "codex", model: "gpt-5.1-codex-mini", env: {
OPENAI_API_KEY: { type: "user_secret_ref", key: "OPENAI_API_KEY", version: "latest", required: true },
} }, runtimeConfig: { heartbeat: { enabled: false, wakeOnDemand: true, maxConcurrentRuns: 50 } },
@@ -284,30 +356,49 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy
{ timeout: 60_000 }).toBe("ready");
agents.push(agent);
}
(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",
defaultEnvironmentId: localEnvironment.id,
adapterConfig: { provider: "codex", model: "gpt-5.1-codex-mini", env: {
OPENAI_API_KEY: { type: "user_secret_ref", key: "OPENAI_API_KEY", version: "latest", required: true },
} }, runtimeConfig: { heartbeat: { enabled: false, wakeOnDemand: true, maxConcurrentRuns: 50 } },
} }));
await expect.poll(async () => (await json(await request.get(`/api/agents/${controlAgent.id}`))).lifecycleState,
{ timeout: 60_000 }).toBe("ready");
report.local = { transport: "local", environmentId: localEnvironment.id, controlAgentId: controlAgent.id,
selection: "Normal local environment + separate control agent defaultEnvironmentId; existing test-only Codex leaf command fixture",
commandSha256: (report.harnessProvenance as any).sourceHashes["fixtures/native-ingestion-bin/codex"],
loaderSha256: (report.harnessProvenance as any).sourceHashes["fixtures/composer-stop-codex-loader.mjs"] };
async function perform(workload: Case, stop = false) {
const caseTransport = stop ? "local" : "sandbox-remote";
const caseEnvironment = stop ? localEnvironment : remoteEnvironment;
const caseAgents = stop ? [controlAgent] : agents;
const key = randomUUID().replaceAll("-", ""), nonce = randomUUID().replaceAll("-", "");
const markerCode = nextMarkerCode();
pending.set(key, new Map());
const state: ClockCase = { events: workload.events, intervalMs: workload.intervalMs, streams: new Map(), errors: [] };
clocks.set(key, state);
const register = (token: string, code: string) => state.streams.set(token, { nonce: token, markerCode: code,
opened: false, hello: false, ended: false, aborted: false, partialBytes: 0, records: new Map() });
opened: false, hello: false, ended: false, aborted: false, partialBytes: 0, records: new Map(),
expectedLauncher: stop ? "host-local" : "staged-remote" });
register(nonce, markerCode);
const compactOutput = Array.from({ length: workload.events }, (_, index) => `[${markerCode}${index.toString(36)}] `).join("");
const preludeOutput = workload.prelude ? `${".".repeat(Math.max(0, workload.prelude - 1))} PRELUDE_${nonce}_READY ` : "";
expect([...compactOutput].length, "Full measured completion must fit the unchanged Rust text bound").toBeLessThan(4000);
expect([...preludeOutput + compactOutput].length, "Prelude and measured text must remain below that bound").toBeLessThan(4000);
const active: Record<string, unknown> = { ...workload, stop, nonce, markerCode, phase: "setup", key };
const active: Record<string, unknown> = { ...workload, stop, transport: caseTransport,
environmentId: caseEnvironment.id, nonce, markerCode, phase: "setup", key };
report.activeCase = active;
const prompt = (token: string, code: string, prelude: number) => `Measure native streaming for ${token.slice(0, 8)}.\n\nINGESTION_PERF nonce=${token} events=${workload.events} intervalMs=${workload.intervalMs} prelude=${prelude} markerCode=${code} gate=${base}${key}`;
const before = new Set((await json(await request.get(`/api/companies/${company.id}/issues`))).map((issue: { id: string }) => issue.id));
await page.goto(`/${company.issuePrefix}/agents/${agents[0].id}/overview`);
await page.goto(`/${company.issuePrefix}/agents/${caseAgents[0].id}/overview`);
await page.getByRole("link", { name: "Tasks", exact: true }).click();
await page.getByRole("button", { name: "New Task", exact: true }).last().click();
const dialog = page.getByRole("dialog");
await dialog.getByRole("textbox", { name: "editable markdown" }).fill(prompt(nonce, markerCode, workload.prelude));
await expect(dialog.getByRole("button", { name: "Select assignee" })).toContainText(agents[0].name);
await expect(dialog.getByRole("button", { name: "Select assignee" })).toContainText(caseAgents[0].name);
await dialog.getByRole("button", { name: "Create task", exact: true }).click();
let issue: any;
await expect.poll(async () => {
@@ -322,7 +413,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: agents[Math.floor(index / 50)].id,
status: "backlog", assigneeAgentId: caseAgents[Math.floor(index / 50)].id,
} }));
issueIds.push(background.id);
issuedTokens.set(background.id, backgroundToken);
@@ -371,10 +462,24 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy
active.preludeProof = preludeProof;
expect(count).toBe(workload.prelude); expect(identities.size).toBe(count);
}
const controllerAuthority = remoteEnvironment ? await remoteControllerAuthority({
appRoot, companyId: company.id, environmentId: remoteEnvironment.id, issueIds,
}) : undefined;
if (controllerAuthority) active.controllerAuthority = controllerAuthority;
const controllerAuthority = await runtimeAuthority({
appRoot, companyId: company.id, environmentId: caseEnvironment.id, issueIds, transport: caseTransport,
});
active.controllerAuthority = 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 };
});
let localProcesses;
if (stop) {
const row = controllerAuthority.rows[0], identity = row.process_identity;
expect(identity.processPid).toBe(row.process_pid); expect(identity.processGroupId).toBe(row.process_group_id);
const providerPid = providerClocks[0].providerPid!;
localProcesses = { runnerPid: row.process_pid, processGroupId: row.process_group_id, providerPid,
hostname: hostname(), checkedAtMs: Date.now(), runnerAlive: !processAbsent(row.process_pid),
groupAlive: !processAbsent(-row.process_group_id), providerAlive: !processAbsent(providerPid) };
expect(localProcesses.runnerAlive && localProcesses.groupAlive && localProcesses.providerAlive).toBe(true);
}
const epochMs = Date.now() + 1000;
state.epochMs = epochMs; active.epochMs = epochMs; active.phase = "streaming";
const participants = [...pending.get(key)!.keys()];
@@ -408,7 +513,7 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy
const acknowledgedAt = stop ? Date.now() : undefined;
let remoteLeases: unknown[] | undefined;
let remoteLeasesCheckedAtMs: number | undefined;
if (remoteEnvironment) {
if (!stop) {
let leases: any[] = [];
await expect.poll(async () => {
leases = await json(await request.get(`/api/environments/${remoteEnvironment.id}/leases`));
@@ -426,6 +531,32 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy
terminationReceipt: lease.metadata?.remoteExecutionTermination };
});
}
let localStopProof;
if (stop) {
let leases: any[] = [];
await expect.poll(async () => {
leases = await json(await request.get(`/api/environments/${localEnvironment.id}/leases`));
return leases.some(lease => lease.heartbeatRunId === runs[0].id && lease.provider === "local"
&& lease.releasedAt && lease.cleanupStatus !== "failed");
}, { timeout: 30_000 }).toBe(true);
const processIdentity = localProcesses!;
await expect.poll(() => processAbsent(processIdentity.runnerPid) && processAbsent(-processIdentity.processGroupId)
&& processAbsent(processIdentity.providerPid), { timeout: 30_000 }).toBe(true);
const database = await runtimeAuthority({ appRoot, companyId: company.id, environmentId: localEnvironment.id,
issueIds, transport: "local", afterStop: true });
const row = database.rows[0];
expect(row.status).toBe("cancelled"); expect(row.cancellation_dispatch_state).toBe("acknowledged");
if (!row.process_pid && !row.process_group_id) {
expect(row.process_stopped_seq).toBeGreaterThan(row.process_identity_seq);
expect(row.process_stopped?.processPid).toBe(processIdentity.runnerPid);
expect(row.process_stopped?.processGroupId).toBe(processIdentity.processGroupId);
}
localStopProof = { checkedAtMs: Date.now(), database, processes: { ...processIdentity,
runnerAbsent: processAbsent(processIdentity.runnerPid), groupAbsent: processAbsent(-processIdentity.processGroupId),
providerAbsent: processAbsent(processIdentity.providerPid) },
lease: Object.fromEntries(["id", "companyId", "environmentId", "heartbeatRunId", "provider", "providerLeaseId", "status", "releasedAt", "cleanupStatus"]
.map(key => [key, leases.find(lease => lease.heartbeatRunId === runs[0].id)[key]])) };
}
active.phase = "clock-reconciliation";
await expect.poll(() => [...state.streams.values()].every(stream => !stream.error && (stop
? seenMarkers.every(marker => state.streams.get(nonce)!.records.has(marker.index))
@@ -443,8 +574,10 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy
const { count, medianMs, maxMs } = stats(paintTimes.slice(1).map((at, index) => at - paintTimes[index]));
const interChunkGaps = { paintCount: paintTimes.length, count, medianMs, maxMs };
const clockEvidence: Array<Record<string, unknown>> = [];
const result = { ...workload, stop, nonce, markerCode, epochMs, settledAtMs: settledAt, stopRequestedAtMs: stopRequestedAt,
const result = { ...workload, stop, transport: caseTransport, environmentId: caseEnvironment.id,
nonce, markerCode, epochMs, settledAtMs: settledAt, stopRequestedAtMs: stopRequestedAt,
stopAcknowledgedAtMs: acknowledgedAt, preludeProof, controllerAuthority, clockEvidence, interChunkGaps,
providerClocks, localProcesses, localStopProof,
remoteLeases, remoteLeasesCheckedAtMs, runIds: runs.map(run => run.id), participants: participants.length, visible, matchedVisibleClocks: visible.length,
firstVisibleAfterEmissionMs: visible[0].visibleMs - visible[0].emittedMs,
emissionToVisible: stats(visible.map(item => item.visibleMs - item.emittedMs)),
@@ -503,6 +636,7 @@ test("native ingestion reaches the shipped UI promptly and survives reload", asy
expect(state.errors).toEqual([]);
expect([...state.streams.values()].every(stream => !stream.error)).toBe(true);
clockEvidence.push(...[...state.streams.values()].map(stream => ({ nonce: stream.nonce, markerCode: stream.markerCode,
launcher: stream.launcher, providerPid: stream.providerPid,
hello: stream.hello, ended: stream.ended, aborted: stream.aborted, error: stream.error, partialBytes: stream.partialBytes,
providerReadyAtMs: stream.providerReadyAtMs, clockHelloAtMs: stream.clockHelloAtMs, readinessObservedAtMs: stream.readinessObservedAtMs,
emissions: [...stream.records.values()].sort((a, b) => a.index - b.index) })));
@@ -11,14 +11,13 @@ process.env.PAPERCLIP_INGESTION_BROWSER_RECORDING = browserRecording ? "on" : "o
const appRoot = path.resolve(process.env.PAPERCLIP_INGESTION_APP_ROOT ?? path.resolve(import.meta.dirname, "../.."));
const quote = (value: string) => `'${value.replaceAll("'", "'\\''")}'`;
const bootstrap = path.resolve(import.meta.dirname, "fixtures/native-ingestion-bootstrap.mjs");
const transport = process.env.PAPERCLIP_INGESTION_TRANSPORT ?? "local";
if (!["local", "sandbox-remote"].includes(transport)) throw new Error("Unknown ingestion transport");
if (transport === "sandbox-remote" && process.env.DATABASE_URL) {
const transport = process.env.PAPERCLIP_INGESTION_TRANSPORT ?? "sandbox-remote-with-local-stop";
if (transport !== "sandbox-remote-with-local-stop") throw new Error("H4 requires remote streaming with the original local Stop workload");
if (process.env.DATABASE_URL) {
throw new Error("Remote ingestion fixture requires a new disposable embedded database");
}
const remoteServer = path.resolve(import.meta.dirname, "fixtures/native-ingestion-remote-server.mjs");
const compiledEntry = transport === "sandbox-remote"
? `${quote(remoteServer)} --compiled-entry=server/dist/index.js` : "server/dist/index.js";
const compiledEntry = `${quote(remoteServer)} --compiled-entry=server/dist/index.js`;
const compiledCommand = `node --import ./server/node_modules/tsx/dist/loader.mjs ${quote(bootstrap)} && node --import ./server/node_modules/tsx/dist/loader.mjs ${compiledEntry}`;
const command = process.env.PAPERCLIP_INGESTION_SERVER_COMMAND ?? compiledCommand;
const autoStartingOnboard = /\bonboard\b[^;&]*(?:--yes|-y)\b/.test(command);
@@ -34,7 +33,7 @@ const providerHome = fs.mkdtempSync(path.join(os.tmpdir(), "ingestion-provider-"
export default defineConfig({
...base,
testMatch: "native-event-ingestion.journey.ts",
timeout: transport === "sandbox-remote" ? 900_000 : 600_000,
timeout: 900_000,
...(evidence ? { outputDir: path.join(evidence, "test-results"),
reporter: [["list"], ["json", { outputFile: path.join(evidence, "playwright.json") }]] as ReporterDescription[] } : {}),
use: { ...base.use, trace: browserRecording ? "on" : "off", video: "off", actionTimeout: 20_000 },
@@ -49,7 +48,7 @@ export default defineConfig({
PAPERCLIP_STOP_CODEX_COMMAND: path.resolve(import.meta.dirname, "fixtures/native-ingestion-bin/codex"),
CODEX_HOME: providerHome,
OPENAI_API_KEY: "ingestion-fixture-key",
NODE_OPTIONS: `${server?.env?.NODE_OPTIONS ?? ""}${transport === "local" ? ` --import=${path.resolve(import.meta.dirname, "fixtures/composer-stop-codex-loader.mjs")}` : ""}`,
NODE_OPTIONS: `${server?.env?.NODE_OPTIONS ?? ""} --import=${path.resolve(import.meta.dirname, "fixtures/composer-stop-codex-loader.mjs")}`,
},
},
});