diff --git a/scripts/compare-native-ingestion-browser.ts b/scripts/compare-native-ingestion-browser.ts index 469105c2d60..44174f879d4 100644 --- a/scripts/compare-native-ingestion-browser.ts +++ b/scripts/compare-native-ingestion-browser.ts @@ -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(); 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 = { - 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"); diff --git a/scripts/native-ingestion-performance-gates.test.mjs b/scripts/native-ingestion-performance-gates.test.mjs index 7ca344767bc..e87a5af5f79 100644 --- a/scripts/native-ingestion-performance-gates.test.mjs +++ b/scripts/native-ingestion-performance-gates.test.mjs @@ -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); diff --git a/tests/e2e/fixtures/in-feed-codex.mjs b/tests/e2e/fixtures/in-feed-codex.mjs index 07806f40184..b36583934d0 100644 --- a/tests/e2e/fixtures/in-feed-codex.mjs +++ b/tests/e2e/fixtures/in-feed-codex.mjs @@ -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 }; } diff --git a/tests/e2e/fixtures/native-ingestion-bin/codex b/tests/e2e/fixtures/native-ingestion-bin/codex index 93acfb1aaef..86473839fb6 100755 --- a/tests/e2e/fixtures/native-ingestion-bin/codex +++ b/tests/e2e/fixtures/native-ingestion-bin/codex @@ -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 "$@" diff --git a/tests/e2e/fixtures/native-ingestion-protocol-qualification.mjs b/tests/e2e/fixtures/native-ingestion-protocol-qualification.mjs index 1622239d877..42204f95eb5 100644 --- a/tests/e2e/fixtures/native-ingestion-protocol-qualification.mjs +++ b/tests/e2e/fixtures/native-ingestion-protocol-qualification.mjs @@ -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 }); diff --git a/tests/e2e/fixtures/native-ingestion-remote-server.mjs b/tests/e2e/fixtures/native-ingestion-remote-server.mjs index 55ff4d79d47..62b9c531278 100644 --- a/tests/e2e/fixtures/native-ingestion-remote-server.mjs +++ b/tests/e2e/fixtures/native-ingestion-remote-server.mjs @@ -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' })); diff --git a/tests/e2e/fixtures/native-ingestion-remote.md b/tests/e2e/fixtures/native-ingestion-remote.md index 94c0d150787..0c13e8826f6 100644 --- a/tests/e2e/fixtures/native-ingestion-remote.md +++ b/tests/e2e/fixtures/native-ingestion-remote.md @@ -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. diff --git a/tests/e2e/native-event-ingestion.journey.ts b/tests/e2e/native-event-ingestion.journey.ts index 15bd0631b01..386e737a432 100644 --- a/tests/e2e/native-event-ingestion.journey.ts +++ b/tests/e2e/native-event-ingestion.journey.ts @@ -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; 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; 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>(); const clocks = new Map(); @@ -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 = { - 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 = { ...workload, stop, nonce, markerCode, phase: "setup", key }; + const active: Record = { ...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> = []; - 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) }))); diff --git a/tests/e2e/playwright-native-ingestion.config.ts b/tests/e2e/playwright-native-ingestion.config.ts index 60893d36650..47c7ce3276f 100644 --- a/tests/e2e/playwright-native-ingestion.config.ts +++ b/tests/e2e/playwright-native-ingestion.config.ts @@ -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")}`, }, }, });