ci(runner): compare retained and streaming snapshots

Add a bounded native Intel ABBA diagnostic using the exact retained baseline and separately compiled unpublished streaming candidate. Keep full output integrity, process ownership, and diagnostic-only attribution.

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-10-02 00:29:34 -05:00
1 parent 3f23a76698
commit 4286d26d05
8 files changed
+1095 -3

No files matched your search

+76 -3
View File
@@ -31,6 +31,10 @@ on:
description: "Verify frozen Pi 1.0 closed startup and SDK contracts on native Intel macOS (no provider prompts)"
type: boolean
default: false
diagnose_pi_snapshot_streaming:
description: "Compare retained B9 baseline and unpublished streaming snapshots in four ABBA children (diagnostic only)"
type: boolean
default: false
diagnose_pi_snapshot_pool:
description: "Measure exact retained B9 snapshots in four process-start pool4/16 ABBA children (diagnostic only)"
type: boolean
@@ -74,7 +78,7 @@ permissions: {}
concurrency:
# Different immutable manual sources retain independent evidence; the same
# source/mode still supersedes itself. Ordinary PR grouping is unchanged.
group: docker-runner-check-${{ github.event.pull_request.number || github.ref }}-${{ inputs.diagnose_pi_snapshot_pool && 'pi-snapshot-pool' || inputs.diagnose_pi_startup && 'pi-startup-diagnostic' || inputs.verify_pi_intel && 'pi-native-intel' || inputs.diagnose_ajv_pack && 'ajv-diagnostic' || inputs.verify_runner && 'runner-verification' || inputs.source_e2e_support_only && 'source-e2e-support' || inputs.source_root_tests_only && 'source-root-tests' || inputs.verify_source && 'verification' || inputs.verify_public_install && 'public-install' || inputs.build_eval_viewer && 'viewer' || 'image' }}${{ github.event_name == 'workflow_dispatch' && inputs.expected_source_sha && format('-{0}', inputs.expected_source_sha) || '' }}
group: docker-runner-check-${{ github.event.pull_request.number || github.ref }}-${{ inputs.diagnose_pi_snapshot_streaming && 'pi-snapshot-streaming' || inputs.diagnose_pi_snapshot_pool && 'pi-snapshot-pool' || inputs.diagnose_pi_startup && 'pi-startup-diagnostic' || inputs.verify_pi_intel && 'pi-native-intel' || inputs.diagnose_ajv_pack && 'ajv-diagnostic' || inputs.verify_runner && 'runner-verification' || inputs.source_e2e_support_only && 'source-e2e-support' || inputs.source_root_tests_only && 'source-root-tests' || inputs.verify_source && 'verification' || inputs.verify_public_install && 'public-install' || inputs.build_eval_viewer && 'viewer' || 'image' }}${{ github.event_name == 'workflow_dispatch' && inputs.expected_source_sha && format('-{0}', inputs.expected_source_sha) || '' }}
cancel-in-progress: true
jobs:
@@ -129,10 +133,11 @@ jobs:
SOURCE_E2E_SUPPORT_ONLY: ${{ inputs.source_e2e_support_only }}
IMAGE_MODE: ${{ inputs.publish_eval_image || inputs.verify_public_install || inputs.build_eval_viewer }}
VERIFY_PI_INTEL: ${{ inputs.verify_pi_intel }}
DIAGNOSE_PI_SNAPSHOT_STREAMING: ${{ inputs.diagnose_pi_snapshot_streaming }}
DIAGNOSE_PI_SNAPSHOT_POOL: ${{ inputs.diagnose_pi_snapshot_pool }}
DIAGNOSE_PI_STARTUP: ${{ inputs.diagnose_pi_startup }}
DIAGNOSE_AJV: ${{ inputs.diagnose_ajv_pack }}
OTHER_MODE: ${{ inputs.publish_eval_image || inputs.verify_source || inputs.verify_public_install || inputs.build_eval_viewer || inputs.verify_pi_intel || inputs.diagnose_pi_startup || inputs.diagnose_pi_snapshot_pool || inputs.source_e2e_support_only }}
OTHER_MODE: ${{ inputs.publish_eval_image || inputs.verify_source || inputs.verify_public_install || inputs.build_eval_viewer || inputs.verify_pi_intel || inputs.diagnose_pi_startup || inputs.diagnose_pi_snapshot_pool || inputs.diagnose_pi_snapshot_streaming || inputs.source_e2e_support_only }}
run: |
set -euo pipefail
test "$REPOSITORY" = paperclipai/paperclip
@@ -176,6 +181,7 @@ jobs:
test "$EXPECTED_LOCK_SHA256" = 5ae57d1ddd475691dac1f1cfbd4836e54f7da1956b47e6e705b218ae3070ae2e
fi
if [ "$VERIFY_PI_INTEL" = true ]; then
test "$DIAGNOSE_PI_SNAPSHOT_STREAMING" != true
test "$DIAGNOSE_PI_SNAPSHOT_POOL" != true
test "$DIAGNOSE_PI_STARTUP" != true
test "$VERIFY_RUNNER" != true
@@ -190,6 +196,7 @@ jobs:
test "$(gh api "repos/$REPOSITORY" --jq '.visibility')" = public
fi
if [ "$DIAGNOSE_PI_STARTUP" = true ]; then
test "$DIAGNOSE_PI_SNAPSHOT_STREAMING" != true
test "$DIAGNOSE_PI_SNAPSHOT_POOL" != true
test "$VERIFY_PI_INTEL" != true
test "$VERIFY_RUNNER" != true
@@ -203,6 +210,21 @@ jobs:
test "$(gh api "repos/$REPOSITORY" --jq '.visibility')" = public
fi
if [ "$DIAGNOSE_PI_SNAPSHOT_POOL" = true ]; then
test "$DIAGNOSE_PI_SNAPSHOT_STREAMING" != true
test "$DIAGNOSE_PI_STARTUP" != true
test "$VERIFY_PI_INTEL" != true
test "$VERIFY_RUNNER" != true
test "$VERIFY_SOURCE" != true
test "$SOURCE_ROOT_TESTS_ONLY" != true
test "$SOURCE_E2E_SUPPORT_ONLY" != true
test "$DIAGNOSE_AJV" != true
test "$IMAGE_MODE" != true
test "$EXPECTED_SOURCE_SHA" = b9e5d6ecdb05ab7244c90976e07c950f8d09b15b
test "$EXPECTED_LOCK_SHA256" = 38338a6867358440c5ab5993eaeb85fb501bc7df65d24acf52fbd63c880ee4ba
test "$(gh api "repos/$REPOSITORY" --jq '.visibility')" = public
fi
if [ "$DIAGNOSE_PI_SNAPSHOT_STREAMING" = true ]; then
test "$DIAGNOSE_PI_SNAPSHOT_POOL" != true
test "$DIAGNOSE_PI_STARTUP" != true
test "$VERIFY_PI_INTEL" != true
test "$VERIFY_RUNNER" != true
@@ -378,6 +400,57 @@ jobs:
retention-days: 7
if-no-files-found: error
manual_pi_snapshot_streaming:
name: Retained B9 native Intel baseline/streaming ABBA (not qualification)
if: github.event_name == 'workflow_dispatch' && inputs.diagnose_pi_snapshot_streaming
needs: authorize_manual
runs-on: macos-15-intel
timeout-minutes: 25
permissions:
contents: read
actions: read
steps:
- name: Checkout trusted diagnostic helpers only
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7
with:
ref: ${{ github.workflow_sha }}
path: trusted-ci
persist-credentials: false
- name: Download exact failed qualification artifacts
timeout-minutes: 5
env:
GH_TOKEN: ${{ github.token }}
run: |
set -euo pipefail
gh api repos/paperclipai/paperclip/actions/artifacts/11208570718/zip \
> "$RUNNER_TEMP/pi-startup-36959948724.zip"
printf '%s %s\n' b82cf6c843006781e2e6d5f0a06e0ff39497ffc457b94a511f8bdda66628aa43 \
"$RUNNER_TEMP/pi-startup-36959948724.zip" | shasum -a 256 --check
- name: Measure four baseline/candidate snapshots with identical process-start pool4
timeout-minutes: 18
run: |
set -euo pipefail
python3 -B trusted-ci/scripts/ci/pi-intel/snapshot_streaming.py \
--archive "$RUNNER_TEMP/pi-startup-36959948724.zip" \
--output "$GITHUB_WORKSPACE/pi-snapshot-streaming-evidence"
- name: Admit complete diagnostic evidence within the storage bound
if: always()
id: diagnostic_evidence
timeout-minutes: 2
run: |
python3 -B trusted-ci/scripts/ci/pi-intel/admit_diagnostic_evidence.py \
--evidence "$GITHUB_WORKSPACE/pi-snapshot-streaming-evidence" \
--github-output "$GITHUB_OUTPUT"
- name: Retain original identities, four snapshot measurements and cleanup
if: always() && steps.diagnostic_evidence.outputs.admitted == 'true'
uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7
with:
name: pi-snapshot-streaming-${{ github.run_id }}
include-hidden-files: true
path: pi-snapshot-streaming-evidence/
retention-days: 7
if-no-files-found: error
manual_runner_verify:
name: Full offline Runner verification on GitHub-hosted Ubuntu
if: github.event_name == 'workflow_dispatch' && inputs.verify_runner
@@ -808,7 +881,7 @@ jobs:
manual_image:
name: Build and verify on EC2
if: github.event_name == 'workflow_dispatch' && !inputs.verify_runner && !inputs.verify_source && !inputs.diagnose_ajv_pack && !inputs.verify_pi_intel && !inputs.diagnose_pi_startup && !inputs.diagnose_pi_snapshot_pool && !inputs.source_e2e_support_only
if: github.event_name == 'workflow_dispatch' && !inputs.verify_runner && !inputs.verify_source && !inputs.diagnose_ajv_pack && !inputs.verify_pi_intel && !inputs.diagnose_pi_startup && !inputs.diagnose_pi_snapshot_pool && !inputs.diagnose_pi_snapshot_streaming && !inputs.source_e2e_support_only
needs: authorize_manual
runs-on: runs-on/fleet=paperclip-public-pr-x64/env=public-ci
timeout-minutes: 90
@@ -0,0 +1,307 @@
import { createHash } from "node:crypto";
import { constants } from "node:fs";
import { chmod, lstat, mkdir, mkdtemp, open, realpath, rm, writeFile } from "node:fs/promises";
import { tmpdir } from "node:os";
import { isAbsolute, join } from "node:path";
const MAX_NATIVE_TREE_BYTES = 1024 * 1024 * 1024;
const MAX_NATIVE_FILE_BYTES = 384 * 1024 * 1024;
const MAX_NATIVE_MANIFEST_BYTES = 4 * 1024 * 1024;
// Bound active copy workers and their retained buffers. Large files stream
// through one chunk instead of excluding other copies or retaining whole files.
const NATIVE_COPY_CONCURRENCY = 32;
const NATIVE_DIRECTORY_CONCURRENCY = 32;
const NATIVE_COPY_BUFFER_BYTES = 32 * 1024 * 1024;
const NATIVE_COPY_CHUNK_BYTES = 1024 * 1024;
const BOOTSTRAP = ".paperclip-native-entry.cjs";
const GUARD = ".paperclip-native-module-guard.cjs";
const COPILOT_ENTRY = ".paperclip-copilot-entry";
export const NATIVE_ACPX_BOOTSTRAP_NAME = BOOTSTRAP;
const sha256 = (value) => createHash("sha256").update(value).digest("hex");
const validPath = (value) => typeof value === "string" && value.length > 0 && value.length <= 4_096 && !isAbsolute(value) && !/[\u0000-\u001f\u007f\\]/.test(value) && value.split("/").every(part => part.length > 0 && part !== "." && part !== "..");
/** Parse a closed, canonical execution inventory before opening any file in it. */
export function parseNativeAcpxDistributionEntries(value, expectedSha256) {
if (!/^[a-f0-9]{64}$/.test(expectedSha256))
throw new Error("Native ACPX closure pin is invalid");
const entries = value !== null && typeof value === "object" ? value.entries : undefined;
if (!Array.isArray(entries) || entries.length === 0 || entries.length > 30_000)
throw new Error("Native ACPX closure inventory is invalid");
let total = 0;
let previous = "";
const parsed = entries.map(raw => {
if (raw === null || typeof raw !== "object" || Array.isArray(raw))
throw new Error("Native ACPX closure entry is invalid");
const entry = raw;
if (Object.keys(entry).some(key => !["path", "sha256", "size", "executable"].includes(key)) || !validPath(entry.path) || entry.path <= previous || entry.path === BOOTSTRAP || entry.path === GUARD || entry.path === "manifest.json" || entry.path.startsWith(".paperclip-"))
throw new Error("Native ACPX closure paths must be safe, unique and sorted");
if (typeof entry.sha256 !== "string" || !/^[a-f0-9]{64}$/.test(entry.sha256) || !Number.isSafeInteger(entry.size) || Number(entry.size) < 0 || Number(entry.size) > MAX_NATIVE_FILE_BYTES || typeof entry.executable !== "boolean")
throw new Error("Native ACPX closure file metadata is invalid");
previous = entry.path;
total += Number(entry.size);
if (total > MAX_NATIVE_TREE_BYTES)
throw new Error("Native ACPX closure exceeds its byte bound");
return { path: entry.path, sha256: entry.sha256, size: Number(entry.size), executable: entry.executable };
});
if (sha256(JSON.stringify(parsed)) !== expectedSha256)
throw new Error("Native ACPX closure manifest digest mismatch");
return parsed;
}
export async function readNativeAcpxDistributionEntries(input) {
if (!validPath(input.executable) || (input.entrypoint !== undefined && !validPath(input.entrypoint)) || input.fixedArguments.length > 128 || input.fixedArguments.some(arg => typeof arg !== "string" || arg.length > 16_384 || arg.includes("\0")) || (input.isolatedCacheEnvironmentName !== undefined && input.isolatedCacheEnvironmentName !== "COPILOT_PKG_CACHE_HOME")
|| (input.copilotDistributionDirectory !== undefined && (!validPath(input.copilotDistributionDirectory) || input.entrypoint !== undefined)))
throw new Error("Native ACPX launch declaration is invalid");
const file = await open(input.manifestPath, constants.O_RDONLY | constants.O_NOFOLLOW);
try {
const before = await file.stat({ bigint: true });
if (!before.isFile() || before.size < 1n || before.size > BigInt(MAX_NATIVE_MANIFEST_BYTES))
throw new Error("Native ACPX manifest must be a bounded regular file");
const bytes = await file.readFile();
if (!same(before, await file.stat({ bigint: true })) || bytes.length !== Number(before.size))
throw new Error("Native ACPX manifest changed while read");
const entries = parseNativeAcpxDistributionEntries(JSON.parse(bytes.toString("utf8")), input.expectedClosureSha256);
if (!entries.some(entry => entry.path === input.executable && entry.executable && entry.size > 0) || (input.entrypoint !== undefined && !entries.some(entry => entry.path === input.entrypoint && entry.size > 0)))
throw new Error("Native ACPX executable or entrypoint is absent from its closure");
if (input.copilotDistributionDirectory !== undefined && !entries.some(entry => entry.path === `${input.copilotDistributionDirectory}/index.js` && entry.size > 0))
throw new Error("Copilot owned entrypoint is absent from its closure");
return entries;
}
finally {
await file.close();
}
}
/**
* Freeze only manifest-admitted bytes. Native trees deliberately have a separate
* bound; the tighter JavaScript/npm snapshot limits remain unchanged.
*/
export async function createNativeAcpxDistributionSnapshot(input, entries) {
// Callers cannot substitute entries between manifest validation and copying.
entries = parseNativeAcpxDistributionEntries({ entries }, input.expectedClosureSha256);
if (process.platform !== "linux" && process.platform !== "darwin")
throw new Error("Native ACPX snapshots require Linux or macOS");
const source = await realpath(input.distributionRoot);
const rootBefore = await lstat(source, { bigint: true });
if (!rootBefore.isDirectory() || rootBefore.isSymbolicLink())
throw new Error("Native ACPX distribution must be a directory");
const heldRoot = await open(source, constants.O_RDONLY | constants.O_NOFOLLOW | constants.O_DIRECTORY);
const privateRoot = await realpath(await mkdtemp(join(tmpdir(), "paperclip-acpx-native-")));
const packageRoot = join(privateRoot, "distribution");
const cacheRoot = join(privateRoot, "state");
const directories = new Set([privateRoot, packageRoot]);
const digests = {};
const close = async () => {
for (const path of directories)
await chmod(path, 0o700).catch(() => undefined);
await rm(privateRoot, { recursive: true, force: true });
};
let commandDirectory;
try {
if (!same(rootBefore, await heldRoot.stat({ bigint: true })))
throw new Error("Native ACPX distribution root changed before snapshot");
await mkdir(packageRoot, { mode: 0o700 });
if (input.isolatedCacheEnvironmentName)
await mkdir(cacheRoot, { mode: 0o700 });
// Build each private parent once. Level ordering prevents child creation
// from racing its parent; batches bound work and drain before any cleanup.
const parentLevels = new Map();
for (const entry of entries) {
const parts = entry.path.split("/");
for (let depth = 1; depth < parts.length; depth++) {
const path = join(packageRoot, ...parts.slice(0, depth));
const level = parentLevels.get(depth) ?? new Set();
level.add(path);
parentLevels.set(depth, level);
directories.add(path);
}
}
const directoryBatch = async (paths, operation) => {
for (let start = 0; start < paths.length; start += NATIVE_DIRECTORY_CONCURRENCY) {
const settled = await Promise.allSettled(paths.slice(start, start + NATIVE_DIRECTORY_CONCURRENCY).map(operation));
const failed = settled.find(result => result.status === "rejected");
if (failed?.status === "rejected")
throw failed.reason;
}
};
for (const depth of [...parentLevels.keys()].sort((a, b) => a - b)) {
await directoryBatch([...parentLevels.get(depth)].sort(), path => mkdir(path, { mode: 0o700 }));
}
let failed = false;
let failure;
const recordFailure = (error) => { if (!failed) {
failed = true;
failure = error;
} };
const copyEntry = async (entry) => {
const path = join(source, ...entry.path.split("/"));
if (await realpath(path) !== path)
throw new Error("Native ACPX closure contains a symbolic link");
// Bind metadata to the descriptor we will read, without a redundant
// pathname stat. NONBLOCK prevents a raced-in FIFO from blocking open;
// only a bounded regular single-link file may reach the read below.
const file = await open(path, constants.O_RDONLY | constants.O_NOFOLLOW | constants.O_NONBLOCK);
let destination;
try {
const before = await file.stat({ bigint: true });
if (!before.isFile() || before.nlink !== 1n || before.size !== BigInt(entry.size) || Boolean(before.mode & 73n) !== entry.executable)
throw new Error("Native ACPX closure file identity is invalid");
const bytes = Buffer.alloc(Math.min(entry.size, NATIVE_COPY_CHUNK_BYTES));
const target = join(packageRoot, ...entry.path.split("/"));
const streaming = entry.size > NATIVE_COPY_CHUNK_BYTES;
const contentHash = streaming ? createHash("sha256") : undefined;
// Partial large files remain inside the unpublished private tree. No
// digests, command lease or handoff exist until every copy is verified.
if (streaming)
destination = await open(target, constants.O_WRONLY | constants.O_CREAT | constants.O_EXCL | constants.O_NOFOLLOW, entry.executable ? 0o500 : 0o400);
let offset = 0;
while (offset < entry.size) {
const bufferOffset = streaming ? 0 : offset;
const read = await file.read(bytes, bufferOffset, Math.min(bytes.length - bufferOffset, entry.size - offset), offset);
if (read.bytesRead === 0)
throw new Error("Native ACPX closure file ended during snapshot");
if (destination) {
contentHash.update(bytes.subarray(0, read.bytesRead));
let written = 0;
while (written < read.bytesRead) {
const result = await destination.write(bytes, written, read.bytesRead - written, offset + written);
if (result.bytesWritten === 0)
throw new Error("Native ACPX snapshot write made no progress");
written += result.bytesWritten;
}
}
offset += read.bytesRead;
}
if (!same(before, await file.stat({ bigint: true })) || !same(before, await lstat(path, { bigint: true })) || await realpath(path) !== path)
throw new Error("Native ACPX closure file changed while read");
if ((contentHash ? contentHash.digest("hex") : sha256(bytes)) !== entry.sha256)
throw new Error(`Native ACPX closure file digest mismatch: ${entry.path}`);
if (destination) {
const copied = await destination.stat({ bigint: true });
if (!copied.isFile() || copied.nlink !== 1n || copied.size !== BigInt(entry.size))
throw new Error("Native ACPX snapshot file identity is invalid");
}
else
await writeFile(target, bytes, { mode: entry.executable ? 0o500 : 0o400, flag: "wx" });
}
catch (error) {
recordFailure(error);
throw error;
}
finally {
// A failed close must not skip the other owned descriptor's retirement.
const closed = await Promise.allSettled([file, ...(destination ? [destination] : [])].map(async (handle) => {
try {
await handle.close();
}
catch (error) {
recordFailure(error);
throw error;
}
}));
const failedClose = closed.find(result => result.status === "rejected");
if (failedClose?.status === "rejected")
throw failedClose.reason;
}
};
// Keep bounded capacity occupied when one file is slower than its peers.
// Every task catches its rejection before releasing capacity; once failed,
// no new copy is admitted and all owned descriptors drain before cleanup.
const active = new Set();
let activeBytes = 0;
for (const entry of entries) {
const bufferBytes = Math.min(entry.size, NATIVE_COPY_CHUNK_BYTES);
while (!failed && (active.size >= NATIVE_COPY_CONCURRENCY
|| activeBytes + bufferBytes > NATIVE_COPY_BUFFER_BYTES)) {
await Promise.race(active);
}
if (failed)
break;
activeBytes += bufferBytes;
const copying = copyEntry(entry).catch(recordFailure)
.finally(() => { activeBytes -= bufferBytes; active.delete(copying); });
active.add(copying);
}
await Promise.all(active);
if (failed)
throw failure;
// Completion order must not change the module guard or manifest.
for (const entry of entries)
digests[join(packageRoot, ...entry.path.split("/"))] = entry.sha256;
if (!same(rootBefore, await heldRoot.stat({ bigint: true })) || !same(rootBefore, await lstat(source, { bigint: true })))
throw new Error("Native ACPX distribution root changed during snapshot");
const executable = join(packageRoot, ...input.executable.split("/"));
const args = [...(input.entrypoint === undefined ? [] : ["--require", join(packageRoot, GUARD), join(packageRoot, ...input.entrypoint.split("/"))]), ...input.fixedArguments];
let copilotEntryDirectory;
if (input.copilotDistributionDirectory !== undefined) {
copilotEntryDirectory = join(packageRoot, COPILOT_ENTRY);
await mkdir(copilotEntryDirectory, { mode: 0o700 });
directories.add(copilotEntryDirectory);
const shim = Buffer.from([
`require(${JSON.stringify(join(packageRoot, GUARD))});`,
`import(require("node:url").pathToFileURL(${JSON.stringify(join(packageRoot, input.copilotDistributionDirectory, "index.js"))}).href).catch(error=>{console.error(error);process.exitCode=1;});`,
].join("\n"));
const shimPath = join(copilotEntryDirectory, "index.js");
await writeFile(shimPath, shim, { mode: 0o400, flag: "wx" });
digests[shimPath] = sha256(shim);
}
if (input.entrypoint !== undefined || copilotEntryDirectory !== undefined) {
const guard = Buffer.from(nativeModuleGuard(digests));
await writeFile(join(packageRoot, GUARD), guard, { mode: 0o400, flag: "wx" });
digests[join(packageRoot, GUARD)] = sha256(guard);
}
const bootstrap = Buffer.from(nativeBootstrap(executable, digests[executable], args, input.isolatedCacheEnvironmentName ? { name: input.isolatedCacheEnvironmentName, value: cacheRoot } : undefined, copilotEntryDirectory));
await writeFile(join(packageRoot, BOOTSTRAP), bootstrap, { mode: 0o400, flag: "wx" });
digests[join(packageRoot, BOOTSTRAP)] = sha256(bootstrap);
// executable is null: commandLease's separately-qualified provider runtime
// slot is unused; this distribution is spawned by the trusted bootstrap.
const manifest = Buffer.from(JSON.stringify({ roots: [packageRoot], executable: null, digests }));
const manifestPath = join(privateRoot, "manifest.json");
await writeFile(manifestPath, manifest, { mode: 0o400, flag: "wx" });
await directoryBatch([...directories], path => chmod(path, 0o500));
commandDirectory = await open(packageRoot, constants.O_RDONLY | constants.O_NOFOLLOW | constants.O_DIRECTORY);
return { commandDirectory, bootstrap, snapshot: { roots: [packageRoot], executable: null, digests, handoff: { path: manifestPath, digest: sha256(manifest) }, close } };
}
catch (error) {
await commandDirectory?.close();
await close();
throw error;
}
finally {
await heldRoot.close();
}
}
function nativeBootstrap(executable, executableDigest, args, cache, copilotEntryDirectory) {
return [
'const {spawn}=require("node:child_process");',
'const fs=require("node:fs");',
// The commandLease's guarded bootstrap owns FDs 5 (guardian), 6 (exit),
// 7/8 (credential fences). Keep them open in the native process too.
'const guarded=process.env.PAPERCLIP_ACPX_NATIVE_GUARDED==="1";',
'const env={...process.env}; delete env.PAPERCLIP_ACPX_NATIVE_GUARDED; delete env.PAPERCLIP_ACPX_PRIVATE_SNAPSHOT; delete env.PAPERCLIP_VERIFIED_RUNTIME_EXECUTABLE;',
'env.NODE_DISABLE_COMPILE_CACHE="1"; delete env.NODE_COMPILE_CACHE;',
...(cache ? [`env[${JSON.stringify(cache.name)}]=${JSON.stringify(cache.value)};`] : []),
...(copilotEntryDirectory ? [
// Never permit the native loader to choose a different cached version,
// app, bootstrap mode or module path supplied by a workspace/caller.
'for(const name of ["COPILOT_CLI_VERSION","COPILOT_CLI_DIST_DIR","COPILOT_CLI_RESOLVED_DIST_DIR","COPILOT_VOICE_SERVER_MODE","COPILOT_SHUTDOWN_FLUSH","NODE_OPTIONS","NODE_PATH"])delete env[name];',
`env.COPILOT_CLI_DIST_DIR=${JSON.stringify(copilotEntryDirectory)};`,
] : []),
'if(guarded)for(const fd of [5,6,7,8])fs.fstatSync(fd);',
`const executable=${JSON.stringify(executable)};const fd=fs.openSync(executable,fs.constants.O_RDONLY|fs.constants.O_NOFOLLOW);`,
`if(require("node:crypto").createHash("sha256").update(fs.readFileSync(fd)).digest("hex")!==${JSON.stringify(executableDigest)})throw new Error("Native ACPX executable changed before spawn");`,
'const stdio=guarded?[0,1,2,5,6,7,8]:[0,1,2]; const childExecutable=process.platform==="linux"?"/proc/self/fd/"+stdio.length:executable;stdio.push(fd);',
`const child=spawn(childExecutable,${JSON.stringify(args)},{env,cwd:process.cwd(),shell:false,detached:false,stdio});fs.closeSync(fd);`,
'child.once("error",()=>process.exit(1)); child.once("exit",(code,signal)=>process.exit(signal?1:code??1));',
'for(const signal of ["SIGTERM","SIGINT","SIGHUP"])process.on(signal,()=>child.kill(signal));',
].join("\n");
}
function nativeModuleGuard(digests) {
return [
'const {registerHooks,isBuiltin}=require("node:module");',
'const {readFileSync,realpathSync}=require("node:fs");',
'const {fileURLToPath}=require("node:url"); const {createHash}=require("node:crypto");',
`const digests=${JSON.stringify(digests)};`,
'const verify=url=>{if(!url.startsWith("file:"))throw new Error("Native ACPX module must be a qualified file or builtin");const path=fileURLToPath(url);if(realpathSync(path)!==path||!Object.hasOwn(digests,path))throw new Error("Native ACPX module escaped its closed distribution");const bytes=readFileSync(path);if(createHash("sha256").update(bytes).digest("hex")!==digests[path])throw new Error("Native ACPX module digest mismatch");return bytes;};',
'registerHooks({resolve(specifier,context,next){const result=next(specifier,context);if(!isBuiltin(specifier)&&!result.url.startsWith("node:"))verify(result.url);return result;},load(url,context,next){if(url.startsWith("node:"))return next(url,context);const bytes=verify(url);if(["module","commonjs","json"].includes(context.format))return {format:context.format,source:bytes,shortCircuit:true};return next(url,context);}});',
].join("\n");
}
function same(a, b) {
return a.dev === b.dev && a.ino === b.ino && a.size === b.size && a.mtimeNs === b.mtimeNs && a.ctimeNs === b.ctimeNs;
}
@@ -0,0 +1,276 @@
import { createHash } from "node:crypto";
import { constants } from "node:fs";
import { chmod, lstat, mkdir, mkdtemp, open, realpath, rm, writeFile, type FileHandle } from "node:fs/promises";
import { tmpdir } from "node:os";
import { isAbsolute, join, relative, resolve } from "node:path";
import type { AcpxPrivateSnapshot } from "./private-snapshot.js";
export interface NativeAcpxDistributionEntry { path: string; sha256: string; size: number; executable: boolean }
export interface NativeAcpxDistributionInput {
distributionRoot: string;
manifestPath: string;
/** Trusted profile pin, never taken from the installed manifest itself. */
expectedClosureSha256: string;
executable: string;
/** When present, execute this JS entry with the distribution's bundled Node. */
entrypoint?: string;
fixedArguments: readonly string[];
isolatedCacheEnvironmentName?: "COPILOT_PKG_CACHE_HOME";
/** Trusted Copilot profile only; resolved inside each immutable spawn lease. */
copilotDistributionDirectory?: string;
}
export interface NativeAcpxDistributionSnapshot {
snapshot: AcpxPrivateSnapshot;
commandDirectory: FileHandle;
bootstrap: Buffer;
}
const MAX_NATIVE_TREE_BYTES = 1024 * 1024 * 1024;
const MAX_NATIVE_FILE_BYTES = 384 * 1024 * 1024;
const MAX_NATIVE_MANIFEST_BYTES = 4 * 1024 * 1024;
// Bound active copy workers and their retained buffers. Large files stream
// through one chunk instead of excluding other copies or retaining whole files.
const NATIVE_COPY_CONCURRENCY = 32;
const NATIVE_DIRECTORY_CONCURRENCY = 32;
const NATIVE_COPY_BUFFER_BYTES = 32 * 1024 * 1024;
const NATIVE_COPY_CHUNK_BYTES = 1024 * 1024;
const BOOTSTRAP = ".paperclip-native-entry.cjs";
const GUARD = ".paperclip-native-module-guard.cjs";
const COPILOT_ENTRY = ".paperclip-copilot-entry";
export const NATIVE_ACPX_BOOTSTRAP_NAME = BOOTSTRAP;
const sha256 = (value: string | Buffer) => createHash("sha256").update(value).digest("hex");
const validPath = (value: unknown): value is string => typeof value === "string" && value.length > 0 && value.length <= 4_096 && !isAbsolute(value) && !/[\u0000-\u001f\u007f\\]/.test(value) && value.split("/").every(part => part.length > 0 && part !== "." && part !== "..");
/** Parse a closed, canonical execution inventory before opening any file in it. */
export function parseNativeAcpxDistributionEntries(value: unknown, expectedSha256: string): NativeAcpxDistributionEntry[] {
if (!/^[a-f0-9]{64}$/.test(expectedSha256)) throw new Error("Native ACPX closure pin is invalid");
const entries = value !== null && typeof value === "object" ? (value as { entries?: unknown }).entries : undefined;
if (!Array.isArray(entries) || entries.length === 0 || entries.length > 30_000) throw new Error("Native ACPX closure inventory is invalid");
let total = 0;
let previous = "";
const parsed = entries.map(raw => {
if (raw === null || typeof raw !== "object" || Array.isArray(raw)) throw new Error("Native ACPX closure entry is invalid");
const entry = raw as Record<string, unknown>;
if (Object.keys(entry).some(key => !["path", "sha256", "size", "executable"].includes(key)) || !validPath(entry.path) || entry.path <= previous || entry.path === BOOTSTRAP || entry.path === GUARD || entry.path === "manifest.json" || entry.path.startsWith(".paperclip-")) throw new Error("Native ACPX closure paths must be safe, unique and sorted");
if (typeof entry.sha256 !== "string" || !/^[a-f0-9]{64}$/.test(entry.sha256) || !Number.isSafeInteger(entry.size) || Number(entry.size) < 0 || Number(entry.size) > MAX_NATIVE_FILE_BYTES || typeof entry.executable !== "boolean") throw new Error("Native ACPX closure file metadata is invalid");
previous = entry.path;
total += Number(entry.size);
if (total > MAX_NATIVE_TREE_BYTES) throw new Error("Native ACPX closure exceeds its byte bound");
return { path: entry.path, sha256: entry.sha256, size: Number(entry.size), executable: entry.executable };
});
if (sha256(JSON.stringify(parsed)) !== expectedSha256) throw new Error("Native ACPX closure manifest digest mismatch");
return parsed;
}
export async function readNativeAcpxDistributionEntries(input: NativeAcpxDistributionInput): Promise<NativeAcpxDistributionEntry[]> {
if (!validPath(input.executable) || (input.entrypoint !== undefined && !validPath(input.entrypoint)) || input.fixedArguments.length > 128 || input.fixedArguments.some(arg => typeof arg !== "string" || arg.length > 16_384 || arg.includes("\0")) || (input.isolatedCacheEnvironmentName !== undefined && input.isolatedCacheEnvironmentName !== "COPILOT_PKG_CACHE_HOME")
|| (input.copilotDistributionDirectory !== undefined && (!validPath(input.copilotDistributionDirectory) || input.entrypoint !== undefined))) throw new Error("Native ACPX launch declaration is invalid");
const file = await open(input.manifestPath, constants.O_RDONLY | constants.O_NOFOLLOW);
try {
const before = await file.stat({ bigint: true });
if (!before.isFile() || before.size < 1n || before.size > BigInt(MAX_NATIVE_MANIFEST_BYTES)) throw new Error("Native ACPX manifest must be a bounded regular file");
const bytes = await file.readFile();
if (!same(before, await file.stat({ bigint: true })) || bytes.length !== Number(before.size)) throw new Error("Native ACPX manifest changed while read");
const entries = parseNativeAcpxDistributionEntries(JSON.parse(bytes.toString("utf8")), input.expectedClosureSha256);
if (!entries.some(entry => entry.path === input.executable && entry.executable && entry.size > 0) || (input.entrypoint !== undefined && !entries.some(entry => entry.path === input.entrypoint && entry.size > 0))) throw new Error("Native ACPX executable or entrypoint is absent from its closure");
if (input.copilotDistributionDirectory !== undefined && !entries.some(entry => entry.path === `${input.copilotDistributionDirectory}/index.js` && entry.size > 0)) throw new Error("Copilot owned entrypoint is absent from its closure");
return entries;
} finally { await file.close(); }
}
/**
* Freeze only manifest-admitted bytes. Native trees deliberately have a separate
* bound; the tighter JavaScript/npm snapshot limits remain unchanged.
*/
export async function createNativeAcpxDistributionSnapshot(input: NativeAcpxDistributionInput, entries: NativeAcpxDistributionEntry[]): Promise<NativeAcpxDistributionSnapshot> {
// Callers cannot substitute entries between manifest validation and copying.
entries = parseNativeAcpxDistributionEntries({ entries }, input.expectedClosureSha256);
if (process.platform !== "linux" && process.platform !== "darwin") throw new Error("Native ACPX snapshots require Linux or macOS");
const source = await realpath(input.distributionRoot);
const rootBefore = await lstat(source, { bigint: true });
if (!rootBefore.isDirectory() || rootBefore.isSymbolicLink()) throw new Error("Native ACPX distribution must be a directory");
const heldRoot = await open(source, constants.O_RDONLY | constants.O_NOFOLLOW | constants.O_DIRECTORY);
const privateRoot = await realpath(await mkdtemp(join(tmpdir(), "paperclip-acpx-native-")));
const packageRoot = join(privateRoot, "distribution");
const cacheRoot = join(privateRoot, "state");
const directories = new Set([privateRoot, packageRoot]);
const digests: Record<string, string> = {};
const close = async () => {
for (const path of directories) await chmod(path, 0o700).catch(() => undefined);
await rm(privateRoot, { recursive: true, force: true });
};
let commandDirectory: FileHandle | undefined;
try {
if (!same(rootBefore, await heldRoot.stat({ bigint: true }))) throw new Error("Native ACPX distribution root changed before snapshot");
await mkdir(packageRoot, { mode: 0o700 });
if (input.isolatedCacheEnvironmentName) await mkdir(cacheRoot, { mode: 0o700 });
// Build each private parent once. Level ordering prevents child creation
// from racing its parent; batches bound work and drain before any cleanup.
const parentLevels = new Map<number, Set<string>>();
for (const entry of entries) {
const parts = entry.path.split("/");
for (let depth = 1; depth < parts.length; depth++) {
const path = join(packageRoot, ...parts.slice(0, depth));
const level = parentLevels.get(depth) ?? new Set<string>();
level.add(path); parentLevels.set(depth, level); directories.add(path);
}
}
const directoryBatch = async (paths: string[], operation: (path: string) => Promise<unknown>): Promise<void> => {
for (let start = 0; start < paths.length; start += NATIVE_DIRECTORY_CONCURRENCY) {
const settled = await Promise.allSettled(paths.slice(start, start + NATIVE_DIRECTORY_CONCURRENCY).map(operation));
const failed = settled.find(result => result.status === "rejected");
if (failed?.status === "rejected") throw failed.reason;
}
};
for (const depth of [...parentLevels.keys()].sort((a, b) => a - b)) {
await directoryBatch([...parentLevels.get(depth)!].sort(), path => mkdir(path, { mode: 0o700 }));
}
let failed = false;
let failure: unknown;
const recordFailure = (error: unknown) => { if (!failed) { failed = true; failure = error; } };
const copyEntry = async (entry: NativeAcpxDistributionEntry): Promise<void> => {
const path = join(source, ...entry.path.split("/"));
if (await realpath(path) !== path) throw new Error("Native ACPX closure contains a symbolic link");
// Bind metadata to the descriptor we will read, without a redundant
// pathname stat. NONBLOCK prevents a raced-in FIFO from blocking open;
// only a bounded regular single-link file may reach the read below.
const file = await open(path, constants.O_RDONLY | constants.O_NOFOLLOW | constants.O_NONBLOCK);
let destination: FileHandle | undefined;
try {
const before = await file.stat({ bigint: true });
if (!before.isFile() || before.nlink !== 1n || before.size !== BigInt(entry.size) || Boolean(before.mode & 0o111n) !== entry.executable) throw new Error("Native ACPX closure file identity is invalid");
const bytes = Buffer.alloc(Math.min(entry.size, NATIVE_COPY_CHUNK_BYTES));
const target = join(packageRoot, ...entry.path.split("/"));
const streaming = entry.size > NATIVE_COPY_CHUNK_BYTES;
const contentHash = streaming ? createHash("sha256") : undefined;
// Partial large files remain inside the unpublished private tree. No
// digests, command lease or handoff exist until every copy is verified.
if (streaming) destination = await open(target, constants.O_WRONLY | constants.O_CREAT | constants.O_EXCL | constants.O_NOFOLLOW, entry.executable ? 0o500 : 0o400);
let offset = 0;
while (offset < entry.size) {
const bufferOffset = streaming ? 0 : offset;
const read = await file.read(bytes, bufferOffset, Math.min(bytes.length - bufferOffset, entry.size - offset), offset);
if (read.bytesRead === 0) throw new Error("Native ACPX closure file ended during snapshot");
if (destination) {
contentHash!.update(bytes.subarray(0, read.bytesRead));
let written = 0;
while (written < read.bytesRead) {
const result = await destination.write(bytes, written, read.bytesRead - written, offset + written);
if (result.bytesWritten === 0) throw new Error("Native ACPX snapshot write made no progress");
written += result.bytesWritten;
}
}
offset += read.bytesRead;
}
if (!same(before, await file.stat({ bigint: true })) || !same(before, await lstat(path, { bigint: true })) || await realpath(path) !== path) throw new Error("Native ACPX closure file changed while read");
if ((contentHash ? contentHash.digest("hex") : sha256(bytes)) !== entry.sha256) throw new Error(`Native ACPX closure file digest mismatch: ${entry.path}`);
if (destination) {
const copied = await destination.stat({ bigint: true });
if (!copied.isFile() || copied.nlink !== 1n || copied.size !== BigInt(entry.size)) throw new Error("Native ACPX snapshot file identity is invalid");
} else await writeFile(target, bytes, { mode: entry.executable ? 0o500 : 0o400, flag: "wx" });
} catch (error) { recordFailure(error); throw error; }
finally {
// A failed close must not skip the other owned descriptor's retirement.
const closed = await Promise.allSettled([file, ...(destination ? [destination] : [])].map(async handle => {
try { await handle.close(); } catch (error) { recordFailure(error); throw error; }
}));
const failedClose = closed.find(result => result.status === "rejected");
if (failedClose?.status === "rejected") throw failedClose.reason;
}
};
// Keep bounded capacity occupied when one file is slower than its peers.
// Every task catches its rejection before releasing capacity; once failed,
// no new copy is admitted and all owned descriptors drain before cleanup.
const active = new Set<Promise<void>>();
let activeBytes = 0;
for (const entry of entries) {
const bufferBytes = Math.min(entry.size, NATIVE_COPY_CHUNK_BYTES);
while (!failed && (active.size >= NATIVE_COPY_CONCURRENCY
|| activeBytes + bufferBytes > NATIVE_COPY_BUFFER_BYTES)) {
await Promise.race(active);
}
if (failed) break;
activeBytes += bufferBytes;
const copying = copyEntry(entry).catch(recordFailure)
.finally(() => { activeBytes -= bufferBytes; active.delete(copying); });
active.add(copying);
}
await Promise.all(active);
if (failed) throw failure;
// Completion order must not change the module guard or manifest.
for (const entry of entries) digests[join(packageRoot, ...entry.path.split("/"))] = entry.sha256;
if (!same(rootBefore, await heldRoot.stat({ bigint: true })) || !same(rootBefore, await lstat(source, { bigint: true }))) throw new Error("Native ACPX distribution root changed during snapshot");
const executable = join(packageRoot, ...input.executable.split("/"));
const args = [...(input.entrypoint === undefined ? [] : ["--require", join(packageRoot, GUARD), join(packageRoot, ...input.entrypoint.split("/"))]), ...input.fixedArguments];
let copilotEntryDirectory: string | undefined;
if (input.copilotDistributionDirectory !== undefined) {
copilotEntryDirectory = join(packageRoot, COPILOT_ENTRY);
await mkdir(copilotEntryDirectory, { mode: 0o700 }); directories.add(copilotEntryDirectory);
const shim = Buffer.from([
`require(${JSON.stringify(join(packageRoot, GUARD))});`,
`import(require("node:url").pathToFileURL(${JSON.stringify(join(packageRoot, input.copilotDistributionDirectory, "index.js"))}).href).catch(error=>{console.error(error);process.exitCode=1;});`,
].join("\n"));
const shimPath = join(copilotEntryDirectory, "index.js");
await writeFile(shimPath, shim, { mode: 0o400, flag: "wx" }); digests[shimPath] = sha256(shim);
}
if (input.entrypoint !== undefined || copilotEntryDirectory !== undefined) {
const guard = Buffer.from(nativeModuleGuard(digests));
await writeFile(join(packageRoot, GUARD), guard, { mode: 0o400, flag: "wx" });
digests[join(packageRoot, GUARD)] = sha256(guard);
}
const bootstrap = Buffer.from(nativeBootstrap(executable, digests[executable]!, args, input.isolatedCacheEnvironmentName ? { name: input.isolatedCacheEnvironmentName, value: cacheRoot } : undefined, copilotEntryDirectory));
await writeFile(join(packageRoot, BOOTSTRAP), bootstrap, { mode: 0o400, flag: "wx" });
digests[join(packageRoot, BOOTSTRAP)] = sha256(bootstrap);
// executable is null: commandLease's separately-qualified provider runtime
// slot is unused; this distribution is spawned by the trusted bootstrap.
const manifest = Buffer.from(JSON.stringify({ roots: [packageRoot], executable: null, digests }));
const manifestPath = join(privateRoot, "manifest.json");
await writeFile(manifestPath, manifest, { mode: 0o400, flag: "wx" });
await directoryBatch([...directories], path => chmod(path, 0o500));
commandDirectory = await open(packageRoot, constants.O_RDONLY | constants.O_NOFOLLOW | constants.O_DIRECTORY);
return { commandDirectory, bootstrap, snapshot: { roots: [packageRoot], executable: null, digests, handoff: { path: manifestPath, digest: sha256(manifest) }, close } };
} catch (error) {
await commandDirectory?.close(); await close(); throw error;
} finally { await heldRoot.close(); }
}
function nativeBootstrap(executable: string, executableDigest: string, args: string[], cache?: { name: string; value: string }, copilotEntryDirectory?: string): string {
return [
'const {spawn}=require("node:child_process");',
'const fs=require("node:fs");',
// The commandLease's guarded bootstrap owns FDs 5 (guardian), 6 (exit),
// 7/8 (credential fences). Keep them open in the native process too.
'const guarded=process.env.PAPERCLIP_ACPX_NATIVE_GUARDED==="1";',
'const env={...process.env}; delete env.PAPERCLIP_ACPX_NATIVE_GUARDED; delete env.PAPERCLIP_ACPX_PRIVATE_SNAPSHOT; delete env.PAPERCLIP_VERIFIED_RUNTIME_EXECUTABLE;',
'env.NODE_DISABLE_COMPILE_CACHE="1"; delete env.NODE_COMPILE_CACHE;',
...(cache ? [`env[${JSON.stringify(cache.name)}]=${JSON.stringify(cache.value)};`] : []),
...(copilotEntryDirectory ? [
// Never permit the native loader to choose a different cached version,
// app, bootstrap mode or module path supplied by a workspace/caller.
'for(const name of ["COPILOT_CLI_VERSION","COPILOT_CLI_DIST_DIR","COPILOT_CLI_RESOLVED_DIST_DIR","COPILOT_VOICE_SERVER_MODE","COPILOT_SHUTDOWN_FLUSH","NODE_OPTIONS","NODE_PATH"])delete env[name];',
`env.COPILOT_CLI_DIST_DIR=${JSON.stringify(copilotEntryDirectory)};`,
] : []),
'if(guarded)for(const fd of [5,6,7,8])fs.fstatSync(fd);',
`const executable=${JSON.stringify(executable)};const fd=fs.openSync(executable,fs.constants.O_RDONLY|fs.constants.O_NOFOLLOW);`,
`if(require("node:crypto").createHash("sha256").update(fs.readFileSync(fd)).digest("hex")!==${JSON.stringify(executableDigest)})throw new Error("Native ACPX executable changed before spawn");`,
'const stdio=guarded?[0,1,2,5,6,7,8]:[0,1,2]; const childExecutable=process.platform==="linux"?"/proc/self/fd/"+stdio.length:executable;stdio.push(fd);',
`const child=spawn(childExecutable,${JSON.stringify(args)},{env,cwd:process.cwd(),shell:false,detached:false,stdio});fs.closeSync(fd);`,
'child.once("error",()=>process.exit(1)); child.once("exit",(code,signal)=>process.exit(signal?1:code??1));',
'for(const signal of ["SIGTERM","SIGINT","SIGHUP"])process.on(signal,()=>child.kill(signal));',
].join("\n");
}
function nativeModuleGuard(digests: Record<string, string>): string {
return [
'const {registerHooks,isBuiltin}=require("node:module");',
'const {readFileSync,realpathSync}=require("node:fs");',
'const {fileURLToPath}=require("node:url"); const {createHash}=require("node:crypto");',
`const digests=${JSON.stringify(digests)};`,
'const verify=url=>{if(!url.startsWith("file:"))throw new Error("Native ACPX module must be a qualified file or builtin");const path=fileURLToPath(url);if(realpathSync(path)!==path||!Object.hasOwn(digests,path))throw new Error("Native ACPX module escaped its closed distribution");const bytes=readFileSync(path);if(createHash("sha256").update(bytes).digest("hex")!==digests[path])throw new Error("Native ACPX module digest mismatch");return bytes;};',
'registerHooks({resolve(specifier,context,next){const result=next(specifier,context);if(!isBuiltin(specifier)&&!result.url.startsWith("node:"))verify(result.url);return result;},load(url,context,next){if(url.startsWith("node:"))return next(url,context);const bytes=verify(url);if(["module","commonjs","json"].includes(context.format))return {format:context.format,source:bytes,shortCircuit:true};return next(url,context);}});',
].join("\n");
}
function same(a: { dev: bigint; ino: bigint; size: bigint; mtimeNs: bigint; ctimeNs: bigint }, b: typeof a): boolean {
return a.dev === b.dev && a.ino === b.ino && a.size === b.size && a.mtimeNs === b.mtimeNs && a.ctimeNs === b.ctimeNs;
}
@@ -0,0 +1,56 @@
// Diagnostic derivative only; original retained pack never changes.
import assert from 'node:assert/strict';
import * as fs from 'node:fs';
import {createHash} from 'node:crypto';
import {join,resolve,dirname} from 'node:path';
import {pathToFileURL,fileURLToPath} from 'node:url';
import {validateEnvironment,verifySealedSnapshot} from './snapshot-pool-fixture.mjs';
const sha=b=>createHash('sha256').update(b).digest('hex');
export async function run(pack, pin, variant, report) {
validateEnvironment(process.env, '4');
assert.ok(['baseline','candidate'].includes(variant));
assert.equal(process.platform, 'darwin'); assert.equal(process.arch, 'x64');
assert.equal(process.version, 'v24.21.0');
const tmp = fs.realpathSync(process.env.TMPDIR);
assert.deepEqual(fs.readdirSync(tmp), []);
const tmpIdentity = fs.lstatSync(tmp);
assert.equal(tmpIdentity.mode & 0o777, 0o700);
const modulePath = variant==='baseline' ? join(pack,pin.modulePath) : join(dirname(fileURLToPath(import.meta.url)),'snapshot-streaming-candidate.mjs');
const moduleSha256=variant==='baseline'?pin.moduleSha256:pin.candidateModuleSha256;
assert.equal(sha(fs.readFileSync(join(pack,pin.modulePath))),pin.moduleSha256);
assert.equal(sha(fs.readFileSync(modulePath)), moduleSha256);
assert.equal(sha(fs.readFileSync(join(pack, pin.closureManifestPath))), pin.closureManifestSha256);
const native = await import(pathToFileURL(modulePath).href);
const input = { distributionRoot: join(pack, pin.distributionPath), manifestPath: join(pack, pin.closureManifestPath), expectedClosureSha256: pin.closureDigest.replace(/^sha256:/, ''), executable: 'node/bin/node', entrypoint: 'pi-entry.cjs', fixedArguments: [] };
const proof = { schema: 'paperclip.snapshot-streaming-copy/v1', variant, isQualification: false, poolRequestedAtProcessStart: 4, actualThreadCountMeasured: false, moduleSha256, admittedEnvironmentKeys: Object.keys(process.env).sort(), osAddedTextEncoding: process.env.__CF_USER_TEXT_ENCODING ?? null, providerCalls: 0, vendorProcessesStarted: 0, snapshotClosed: false, commandDirectoryClosed: false };
let snapshot;
try {
const entries = await native.readNativeAcpxDistributionEntries(input);
assert.equal(entries.length, pin.entries);
assert.equal(entries.reduce((sum, e) => sum + e.size, 0), pin.distributionBytes);
const start = process.hrtime.bigint();
snapshot = await native.createNativeAcpxDistributionSnapshot(input, entries);
proof.snapshotMilliseconds = Number(process.hrtime.bigint() - start) / 1e6;
// Deliberately outside the timed routine; this extra verification is identical in all four children.
proof.sealedOutput = verifySealedSnapshot(snapshot, entries, tmp);
proof.status = 'snapshot_verified_not_executed';
} finally {
const errors = [];
if (snapshot) {
try { await snapshot.commandDirectory.close(); proof.commandDirectoryClosed = true; } catch (error) { errors.push(error); }
try { await snapshot.snapshot.close(); proof.snapshotClosed = true; } catch (error) { errors.push(error); }
}
try {
const current = fs.lstatSync(tmp);
assert.equal(current.dev, tmpIdentity.dev); assert.equal(current.ino, tmpIdentity.ino);
assert.deepEqual(fs.readdirSync(tmp), []); proof.temporaryRootEmpty = true;
} catch (error) { errors.push(error); }
proof.cleanupErrors = errors.map(error => String(error));
fs.writeFileSync(report, JSON.stringify(proof, null, 2) + '\n', { flag: 'wx', mode: 0o600 });
if (errors.length) throw new AggregateError(errors, 'Snapshot cleanup incomplete');
}
}
if (process.argv[1] && import.meta.url === pathToFileURL(resolve(process.argv[1])).href) {
const [pack, pinPath, variant, report] = process.argv.slice(2);
await run(pack, JSON.parse(fs.readFileSync(pinPath)), variant, report);
}
@@ -0,0 +1,64 @@
import assert from 'node:assert/strict';
import { test } from 'node:test';
import * as fs from 'node:fs';
import { createHash } from 'node:crypto';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { pathToFileURL } from 'node:url';
import { validateEnvironment, verifySealedSnapshot } from './snapshot-pool-fixture.mjs';
const modulePath = new URL('./snapshot-streaming-candidate.mjs',import.meta.url).pathname;
assert.ok(modulePath, 'Explicit retained built-module path is required for this calibration');
assert.equal(createHash('sha256').update(fs.readFileSync(modulePath)).digest('hex'), '2ddacbff88558fb055a2037f904b3fc829140b138488f71bae9d23700c744a1d');
const native = await import(pathToFileURL(modulePath).href);
const sha = bytes => createHash('sha256').update(bytes).digest('hex');
test('environment rejects ambient credentials, mismatched or unsupported pools', () => {
const good = { HOME: '/home', LANG: 'C', PATH: '/bin', TMPDIR: '/tmp', UV_THREADPOOL_SIZE: '4' };
validateEnvironment(good, '4');
assert.throws(() => validateEnvironment({ ...good, OPENROUTER_API_KEY: 'dummy' }, '4'));
assert.throws(() => validateEnvironment(good, '16'));
assert.throws(() => validateEnvironment({ ...good, UV_THREADPOOL_SIZE: '8' }, '8'));
validateEnvironment({ ...good, __CF_USER_TEXT_ENCODING: `0x${process.getuid().toString(16)}:0x0:0x0` }, '4');
assert.throws(() => validateEnvironment({ ...good, __CF_USER_TEXT_ENCODING: `0x${(process.getuid() + 1).toString(16)}:0x0:0x0` }, '4'));
assert.throws(() => validateEnvironment({ ...good, __CF_USER_TEXT_ENCODING: 'unknown' }, '4'));
});
for (const mutation of ['none', 'changed bytes', 'wrong mode', 'extra file', 'extra directory', 'symbolic link', 'hardlink', 'handoff drift']) {
test(`actual compiled streaming snapshot calibration: ${mutation}`, async () => {
const source = fs.mkdtempSync(join(tmpdir(), 'pc-pool-calibration-'));
let value;
try {
const body=Buffer.alloc(2*1024*1024+7,42);
const entries = [{ path: 'entry.cjs', sha256: sha(body), size: body.length, executable: false }, { path: 'node', sha256: sha('node'), size: 4, executable: true }];
fs.writeFileSync(join(source, 'entry.cjs'), body, { mode: 0o600 });
fs.writeFileSync(join(source, 'node'), 'node', { mode: 0o700 });
const manifestPath = join(source, 'closure.json');
fs.writeFileSync(manifestPath, JSON.stringify({ entries }));
const input = { distributionRoot: source, manifestPath, expectedClosureSha256: sha(JSON.stringify(entries)), executable: 'node', entrypoint: 'entry.cjs', fixedArguments: [] };
value = await native.createNativeAcpxDistributionSnapshot(input, await native.readNativeAcpxDistributionEntries(input));
const root = value.snapshot.roots[0], target = join(root, 'entry.cjs');
if (mutation !== 'none') fs.chmodSync(root, 0o700);
if (mutation === 'changed bytes') { fs.chmodSync(target, 0o600); fs.writeFileSync(target, 'wrong'); fs.chmodSync(target, 0o400); }
if (mutation === 'wrong mode') fs.chmodSync(target, 0o600);
if (mutation === 'extra file') fs.writeFileSync(join(root, 'extra'), 'x', { mode: 0o400 });
if (mutation === 'extra directory') fs.mkdirSync(join(root, 'extra'), { mode: 0o500 });
if (mutation === 'symbolic link') { fs.unlinkSync(target); fs.symlinkSync(join(source, 'entry.cjs'), target); }
if (mutation === 'hardlink') fs.linkSync(target, join(source, 'linked'));
if (mutation === 'handoff drift') value.snapshot.handoff.digest = '0'.repeat(64);
if (mutation !== 'none') fs.chmodSync(root, 0o500);
if (mutation === 'none') assert.equal(verifySealedSnapshot(value, entries, fs.realpathSync(tmpdir())).files, 4);
else assert.throws(() => verifySealedSnapshot(value, entries, fs.realpathSync(tmpdir())));
} finally {
if (value) {
await value.commandDirectory.close();
// The extra empty directory is an intentionally foreign fixture node;
// make it removable without changing the production close function.
const extra = join(value.snapshot.roots[0], 'extra');
if (fs.existsSync(extra) && fs.lstatSync(extra).isDirectory()) fs.chmodSync(extra, 0o700);
await value.snapshot.close();
}
fs.rmSync(source, { recursive: true, force: true });
}
});
}
@@ -0,0 +1,86 @@
{
"schema": "paperclip.native-intel-snapshot-streaming-input/v1",
"sourceRevision": "b9e5d6ecdb05ab7244c90976e07c950f8d09b15b",
"resolvedLockSha256": "38338a6867358440c5ab5993eaeb85fb501bc7df65d24acf52fbd63c880ee4ba",
"artifactRunId": "36959948724",
"artifactId": "11208570718",
"artifactWorkflowRevision": "9541eb18789bac35242eeea3e569b77fa90d1818",
"artifactZipBytes": 1076365959,
"artifactZipSha256": "b82cf6c843006781e2e6d5f0a06e0ff39497ffc457b94a511f8bdda66628aa43",
"selectedFiles": {
"receipt.json": {
"bytes": 32763,
"sha256": "47ef1f494c59c1249ac42f08308a2fb83edcb0759890960059cc367ebdf12bf0"
},
"provider-pack.tar.gz": {
"bytes": 1029214229,
"sha256": "5a550004ca982c49a2d86b3196676f3ce1877b6655ec730d1971b4cec2ba547d"
},
"provider-pack.json": {
"bytes": 3629,
"sha256": "db5b1d7d825a0533a6506755c6209ccc4d45776c3abbf6c87ff845d1d47e99b2"
},
"pack-inventory.json": {
"bytes": 10633748,
"sha256": "172f0f4f2319f3716cca656cdeafd03b4f0db652def53101dfbf148850a20eaa"
},
"paperclip-runnerd": {
"bytes": 32035568,
"sha256": "29d5b0fc2ae2a552b89a107e3777e2c8ee356ebdbb13a14ceab8865e1cea1d45"
},
"source.tar": {
"bytes": 141742080,
"sha256": "b32d8e32639e6527804fe11bfa1aaefcefa87fca31e9155499e6b9a28f69dbd4"
},
"source-input-inventory.json": {
"bytes": 1033702,
"sha256": "7c160bb1fc4d049378004d014e398d42dabf12e8a73ec9cdda7679113aaa6949"
},
"resolved-pnpm-lock.yaml": {
"bytes": 607788,
"sha256": "38338a6867358440c5ab5993eaeb85fb501bc7df65d24acf52fbd63c880ee4ba"
}
},
"originalSidecarSha256": "7462825f4b82bce8e2a69425eafe4ca8b2c8812df345269c82fb75aa21f3043c",
"originalPackDigest": "sha256:9cf4f88d9fcf63746f7716f773cb8d062544c19e0a78e111e60f659a3596b13f",
"daemonSha256": "29d5b0fc2ae2a552b89a107e3777e2c8ee356ebdbb13a14ceab8865e1cea1d45",
"nodeSha256": "7abcf39bd37ab251015337ff75304d7555f0d8e88c6e0fbf04bce8ce34636f49",
"testPath": "packages/paperclip-runner/test/pi-closed-startup.test.mjs",
"testSha256": "8967cf9c8cd130b68bf9d64abef8cb8d352af00646e2288b341d8c6ae758b47a",
"observerSha256": "1fcf353beb90722f90401a7792a5853bf0807d964a1065e3f55b76a63bc4b4cc",
"profileDigest": "sha256:47306e6d2a9b59e8f9189f725ebb7a0a7f91826044d1739e1a35ab31f228ba1f",
"closureDigest": "sha256:03351f4a250a8db0e79411a9079b43a0ff05f72a2aff41fae17f1fc2de24bd41",
"model": "openrouter/deepseek/deepseek-v4-flash-0731",
"originalFailure": "Original 30s session.open timeout; settled35032ms; runner exit0 and owned cleanup verified;48SDK contracts passed; fresh Rust daemon and complete pack passed.",
"providerCalls": 0,
"timeoutChanges": false,
"archiveBytesMaximum": 268435456,
"retentionDays": 7,
"modulePath": "dist/drivers/acpx/native-distribution-integrity.js",
"moduleSha256": "2c321e2a6b356bdd92e19493ee1d848436bdc895e5aa98c4190a0d8658df15df",
"closureManifestPath": "provider-assets/pi/darwin-x64/native-closure.json",
"closureManifestSha256": "b54c812aa78e072bf9badd0e7e3bf491a948ef5ecef827c40de88be48e2e023b",
"distributionPath": "provider-assets/pi/darwin-x64/runtime",
"maximumCopies": 4,
"experimentDeadlineSeconds": 600,
"perCopyDeadlineSeconds": 90,
"cleanupReserveSeconds": 120,
"finalIntegrityReserveSeconds": 60,
"entries": 15671,
"distributionBytes": 538588650,
"variants": [
"baseline",
"candidate",
"candidate",
"baseline"
],
"candidateModuleSha256": "2ddacbff88558fb055a2037f904b3fc829140b138488f71bae9d23700c744a1d",
"candidateSourceSha256": "4fcd8a9b1f27d0dc11ca9ab582ddaf399e28ccb4fca3fefcd48e147d5ba2dc7f",
"baselineSourceSha256": "943ba0fb2d27d08b9b0d36778f08c7c0dbdb78893c7564babd4e67f67cf065fe",
"compilation": {
"tool": "TypeScript7.0.2",
"options": "targetES2023,moduleESNext,noResolve,noCheck,noSourceMap; emit only; separate narrow strict typecheck passed",
"baselineCalibration": "Exact retained baseline emitted bytes except original nonexecuting sourceMappingURL trailer",
"baselineEmittedSha256": "15a05f6e00e8ca58f07c54ee47d8daf4ab5cdeff99788ca8811ad815c6b44bbc"
}
}
+151
View File
@@ -0,0 +1,151 @@
#!/usr/bin/env python3
"""Four baseline/candidate exact-snapshot ABBA measurements; no provider, daemon or qualification."""
import argparse,datetime,hashlib,json,os,platform,shutil,signal,stat,subprocess,tarfile,tempfile,time
from pathlib import Path
from source_guard import require,sha
from native_daemon_reuse import extract_selected
from closed_inventory import closed_tree
from startup_diagnostic import restore_pack,finalize_diagnostic
from diagnostic_lifecycle import DiagnosticChild,DiagnosticInspection
from owned_processes import OwnedProcesses,atomic_json
HERE=Path(__file__).resolve().parent
PIN=json.loads((HERE/'snapshot-streaming-inputs.json').read_text())
def admit_copy(index,variant,remaining,cancelled,uncertain):
require(not cancelled and not uncertain,'Cancelled or uncertain experiment cannot launch another copy')
require(type(index) is int and 0<=index<4 and variant==["baseline","candidate","candidate","baseline"][index],'Only the exact four-copy ABBA sequence is admitted')
require(remaining>=90+120+60,'Full copy, cleanup and final-integrity windows no longer fit')
def child_environment(home,tmp):
return {'PATH':'/usr/bin:/bin:/usr/sbin:/sbin','HOME':str(home),'LANG':'en_US.UTF-8','TMPDIR':str(tmp),'UV_THREADPOOL_SIZE':'4'}
def verify_report(path,variant,pin=PIN):
st=path.lstat();require(stat.S_ISREG(st.st_mode) and st.st_nlink==1 and st.st_size<=16384,'Unsafe copy receipt')
r=json.loads(path.read_text())
require(variant in ('baseline','candidate') and r['variant']==variant,'Wrong experiment variant')
require(r['status']=='snapshot_verified_not_executed' and r['isQualification'] is False and r['poolRequestedAtProcessStart']==4 and r['actualThreadCountMeasured'] is False,'Copy receipt identity/status mismatch')
expected=pin['moduleSha256'] if variant=='baseline' else pin['candidateModuleSha256']
require(r['moduleSha256']==expected and r['providerCalls']==r['vendorProcessesStarted']==0,'Copy code/execution scope mismatch')
require(all(r[k] is True for k in ['snapshotClosed','commandDirectoryClosed','temporaryRootEmpty']) and r['cleanupErrors']==[],'Copy cleanup incomplete')
require(r['sealedOutput']=={'files':pin['entries']+2,'bytes':pin['distributionBytes'],'handoffVerified':True,'descriptorIdentityVerified':True},'Incomplete sealed snapshot verification')
require(type(r['snapshotMilliseconds']) in (int,float) and 0<r['snapshotMilliseconds']<90000,'Invalid copy duration')
return r
def verify_baseline_source(archive,pin=PIN):
name='packages/paperclip-runner/src/drivers/acpx/native-distribution-integrity.ts'
with tarfile.open(archive,'r:') as source:
rows=[m for m in source.getmembers() if m.name==name]
require(len(rows)==1 and rows[0].isreg() and 0<rows[0].size<=131072,'Missing or unsafe baseline source member')
with source.extractfile(rows[0]) as member:payload=member.read(131073)
require(len(payload)==rows[0].size and hashlib.sha256(payload).hexdigest()==pin['baselineSourceSha256'],'Baseline source identity mismatch')
return payload
def post_integrity(pack,inventory,original,pin):
checks={};errors={}
actions={'pack':lambda:closed_tree(pack)==inventory,'candidateModule':lambda:sha(HERE/'snapshot-streaming-candidate.mjs')==pin['candidateModuleSha256'],'candidateSource':lambda:sha(HERE/'snapshot-streaming-candidate.ts')==pin['candidateSourceSha256']}
for name,row in pin['selectedFiles'].items():actions[name]=lambda name=name,row=row:sha(original/name)==row['sha256']
for name,check in actions.items():
try:require(check(),'Identity mismatch: '+name);checks[name]=True
except BaseException as error:checks[name]=False;errors[name]=repr(error)
return {'complete':True,'passed':not errors,'checks':checks,'errors':errors}
def assert_complete(proof):
integrity=proof.get('postRunIntegrity',{})
require(proof.get('status')=='comparison_complete_not_qualification' and proof.get('cleanupUncertain') is False and proof.get('scratchRemoved') is True and integrity.get('complete') is True and integrity.get('passed') is True and not integrity.get('errors') and integrity.get('checks') and all(v is True for v in integrity['checks'].values()),'Comparison finalization did not establish complete integrity and owned cleanup')
def execute(args):
out=args.output.resolve();out.mkdir(mode=0o700,parents=True,exist_ok=False)
scratch=Path(tempfile.mkdtemp(prefix='pc-snapshot-streaming-',dir='/private/tmp'));scratch.chmod(0o700)
initial=scratch.stat();identity=(initial.st_dev,initial.st_ino,initial.st_uid)
proof={'schema':'paperclip.native-intel-snapshot-streaming/v1','status':'preparing','sourceRevision':PIN['sourceRevision'],'runId':os.environ['GITHUB_RUN_ID'],'trustedWorkflowRevision':os.environ['GITHUB_WORKFLOW_SHA'],'startedAt':datetime.datetime.now(datetime.timezone.utc).isoformat(),'isQualification':False,'providerCalls':0,'runtimePromptsSubmitted':0,'vendorProcessesStarted':0,'nativeRecompiled':False,'dependencyInstallExecuted':False,'sourceOrPackModified':False,'actualThreadCountMeasured':False,'poolBinding':'Identical explicit UV_THREADPOOL_SIZE=4 in all four children','candidateScope':'Separately compiled unpublished streaming derivative; original pack and built baseline remain byte-identical','cacheLimitation':'Full archive admission reads all bytes before ABBA. Four sequential samples measure this host and warm-cache order only, not canonical cold startup or end-to-end qualification.','helpers':{p.name:sha(p) for p in HERE.iterdir() if p.is_file()},'copies':[],'cleanupUncertain':False,'scratchRemoved':False}
active=None;cancelled=None;pack=None;inventory=None;original=None;experiment_started=None
def save():atomic_json(out/'receipt.json',proof)
def cancel(sig,_):
nonlocal cancelled
cancelled='signal '+str(sig)
if active is not None:active.cancel(cancelled)
signal.signal(signal.SIGTERM,cancel);signal.signal(signal.SIGINT,cancel)
save()
try:
require(platform.system()=='Darwin' and platform.machine()=='x86_64','Native Intel required')
env={'PATH':'/usr/bin:/bin:/usr/sbin:/sbin','LANG':'en_US.UTF-8'}
for flag in ['sysctl.proc_translated','hw.optional.arm64']:
r=subprocess.run(['/usr/sbin/sysctl','-in',flag],capture_output=True,text=True,env=env,timeout=10)
require(r.stdout.strip() in ('','0'),'Rosetta/ARM rejected');proof[flag]={'value':r.stdout.strip(),'exitCode':r.returncode}
hardware=subprocess.run(['/usr/sbin/sysctl','hw.machine','machdep.cpu.vendor','machdep.cpu.brand_string'],capture_output=True,text=True,env=env,timeout=10)
require(hardware.returncode==0 and 'GenuineIntel' in hardware.stdout and 'x86_64' in hardware.stdout,'Intel identity missing');(out/'hardware.log').write_text(hardware.stdout)
require(not cancelled,'Cancelled before artifact preparation')
original=extract_selected(args.archive,scratch/'original-artifact',PIN)
baseline_source=verify_baseline_source(original/'source.tar');(out/'baseline-native-distribution-integrity.ts').write_bytes(baseline_source)
original_receipt=json.loads((original/'receipt.json').read_text())
require(all(original_receipt[k]==PIN[v] for k,v in [('sourceRevision','sourceRevision'),('runId','artifactRunId'),('trustedWorkflowRevision','artifactWorkflowRevision')]),'Original source/workflow/run mismatch')
require(original_receipt['status']=='failed_no_retry' and original_receipt['cleanupUncertain'] is False,'Original failure/cleanup provenance mismatch')
refs=out/'original-reference';refs.mkdir(mode=0o700)
for name in ['receipt.json','provider-pack.json','pack-inventory.json','source-input-inventory.json','resolved-pnpm-lock.yaml']:shutil.copy2(original/name,refs/name)
atomic_json(out/'original-input-pins.json',PIN)
inventory=json.loads((original/'pack-inventory.json').read_text());pack=restore_pack(original/'provider-pack.tar.gz',scratch/'unpacked',inventory)
node=pack/'node_modules/node/bin/node'
for path,digest in [(node,PIN['nodeSha256']),(pack/PIN['modulePath'],PIN['moduleSha256']),(pack/PIN['closureManifestPath'],PIN['closureManifestSha256'])]:require(sha(path)==digest,'Exact executable/module/closure mismatch')
require(sha(HERE/'snapshot-streaming-candidate.mjs')==PIN['candidateModuleSha256'] and sha(HERE/'snapshot-streaming-candidate.ts')==PIN['candidateSourceSha256'],'Candidate code changed')
for name in ['snapshot-streaming-fixture.mjs','snapshot-streaming-inputs.json','snapshot-streaming-candidate.mjs','snapshot-streaming-candidate.ts','snapshot-pool-fixture.mjs']:shutil.copy2(HERE/name,out/name)
shutil.copy2(pack/PIN['modulePath'],out/'native-distribution-integrity.js')
require(PIN['variants']==['baseline','candidate','candidate','baseline'] and PIN['maximumCopies']==4 and PIN['experimentDeadlineSeconds']==600 and PIN['perCopyDeadlineSeconds']==90 and PIN['cleanupReserveSeconds']==120 and PIN['finalIntegrityReserveSeconds']==60,'Experiment bounds changed')
experiment_started=time.monotonic();deadline=experiment_started+600
proof.update(status='measuring',experimentStartedAt=datetime.datetime.now(datetime.timezone.utc).isoformat());save()
for index,variant in enumerate(PIN['variants']):
admit_copy(index,variant,deadline-time.monotonic(),cancelled,proof['cleanupUncertain'])
copy_root=scratch/('copy-'+str(index));copy_root.mkdir(mode=0o700)
home=copy_root/'home';home.mkdir(mode=0o700);tmp=copy_root/'tmp';tmp.mkdir(mode=0o700)
report=out/('copy-'+str(index)+'.json');log=out/('copy-'+str(index)+'.log')
owner=OwnedProcesses(out/('copy-'+str(index)+'-processes.json'),scratch,pack,pack/'dist/cli/acpx-runtime-sidecar.cjs')
active=DiagnosticChild(owner);inspection=DiagnosticInspection(owner,active);owner.table=inspection.table;owner.argv=inspection.argv
if cancelled:active.cancel(cancelled)
row={'index':index,'variant':variant,'outerDeadlineSeconds':90,'startedAt':datetime.datetime.now(datetime.timezone.utc).isoformat()};proof['copies'].append(row);save()
def spawn():
# Recheck after publishing the new lifecycle; a cancellation before this
# exact boundary cannot escape through the previous child's gate.
if cancelled:active.cancel(cancelled);raise InterruptedError(cancelled)
admit_copy(index,variant,deadline-time.monotonic(),cancelled,proof['cleanupUncertain'])
return subprocess.Popen([str(node),str(HERE/'snapshot-streaming-fixture.mjs'),str(pack),str(HERE/'snapshot-streaming-inputs.json'),variant,str(report)],env=child_environment(home,tmp),cwd=pack,stdout=output,stderr=output,start_new_session=True)
with log.open('xb') as output:result=active.execute(spawn,90)
row.update(exitCode=result,logSha256=sha(log),finishedAt=datetime.datetime.now(datetime.timezone.utc).isoformat(),shutdown=owner.stop_result)
proof['cleanupUncertain']=active.uncertain;save()
require(result==0,'Snapshot child failed; no later copy will launch')
row['measurement']=verify_report(report,variant);row['reportSha256']=sha(report);save()
require(not cancelled,'Experiment cancelled before completion')
proof.update(status='comparison_complete_not_qualification',meansMilliseconds={v:sum(r['measurement']['snapshotMilliseconds'] for r in proof['copies'] if r['variant']==v)/2 for v in ['baseline','candidate']})
return 0
except BaseException as error:
proof.update(status='comparison_failed_no_retry',errorType=type(error).__name__,error=str(error));raise
finally:
def post():
if pack is None:return {'complete':False,'passed':False,'reason':'pack preparation incomplete'}
return post_integrity(pack,inventory,original,PIN)
def retain():
proof['finishedAt']=datetime.datetime.now(datetime.timezone.utc).isoformat()
if experiment_started is not None:proof['experimentThroughIntegritySeconds']=time.monotonic()-experiment_started
def remove():
st=scratch.lstat();require(stat.S_ISDIR(st.st_mode) and (st.st_dev,st.st_ino,st.st_uid)==identity,'Scratch identity changed')
# A forcibly retired copy may leave sealed 0500 snapshot directories. Only
# this owned scratch is made removable, after retirement and integrity audit.
for root,dirs,_ in os.walk(scratch,followlinks=False):
require(not Path(root).is_symlink(),'Scratch directory replaced by link');os.chmod(root,0o700,follow_symlinks=False)
shutil.rmtree(scratch)
try:
finalize_diagnostic(proof,active,post,retain,remove)
if experiment_started is not None:
proof['experimentIncludingCleanupSeconds']=time.monotonic()-experiment_started
if proof['experimentIncludingCleanupSeconds']>600:
proof['status']='comparison_failed_no_retry'
raise RuntimeError('Experiment exceeded its 600-second bound; cleanup completed')
if cancelled:
proof['status']='comparison_failed_no_retry'
raise InterruptedError(cancelled+'; owned cleanup completed')
try:assert_complete(proof)
except BaseException:
proof['status']='comparison_failed_no_retry';raise
finally:save()
if __name__=='__main__':
p=argparse.ArgumentParser(description=__doc__);p.add_argument('--archive',type=Path,required=True);p.add_argument('--output',type=Path,required=True)
raise SystemExit(execute(p.parse_args()))
@@ -0,0 +1,79 @@
import copy,json,tempfile,unittest
from pathlib import Path
from unittest.mock import patch
from snapshot_streaming import PIN,admit_copy,child_environment,verify_report,post_integrity,assert_complete,verify_baseline_source
from diagnostic_lifecycle import DiagnosticChild
class SnapshotStreamingTests(unittest.TestCase):
def test_terminal_success_cannot_survive_failed_or_incomplete_finalization(self):
good={'status':'comparison_complete_not_qualification','cleanupUncertain':False,'scratchRemoved':True,'postRunIntegrity':{'complete':True,'passed':True,'checks':{'pack':True},'errors':{}}}
assert_complete(good)
for delta in [{'status':'diagnostic_failed'},{'cleanupUncertain':True},{'scratchRemoved':False},{'postRunIntegrity':{}},{'postRunIntegrity':{'complete':True,'passed':False,'checks':{'pack':False},'errors':{'pack':'changed'}}},{'postRunIntegrity':{'complete':True,'passed':True,'checks':{},'errors':{}}}]:
with self.subTest(delta=delta),self.assertRaises(RuntimeError):assert_complete(good|delta)
def test_exact_four_sequence_and_full_window(self):
for i,p in enumerate(['baseline','candidate','candidate','baseline']):admit_copy(i,p,270,None,False)
for args in [(4,'baseline',600,None,False),(0,'candidate',600,None,False),(0,'baseline',269.9,None,False),(0,'baseline',600,'cancel',False),(0,'baseline',600,None,True)]:
with self.assertRaises(RuntimeError):admit_copy(*args)
def test_only_explicit_noncredential_process_environment(self):
with patch.dict('os.environ',{'OPENROUTER_API_KEY':'must-not-propagate','UV_THREADPOOL_SIZE':'128'}):
e=child_environment('/private/home','/private/tmp')
self.assertEqual(e,{'HOME':'/private/home','TMPDIR':'/private/tmp','UV_THREADPOOL_SIZE':'4','LANG':'en_US.UTF-8','PATH':'/usr/bin:/bin:/usr/sbin:/sbin'})
def receipt(self):
return {'variant':'baseline','status':'snapshot_verified_not_executed','isQualification':False,'poolRequestedAtProcessStart':4,'actualThreadCountMeasured':False,'moduleSha256':PIN['moduleSha256'],'providerCalls':0,'vendorProcessesStarted':0,'snapshotClosed':True,'commandDirectoryClosed':True,'temporaryRootEmpty':True,'cleanupErrors':[],'sealedOutput':{'files':PIN['entries']+2,'bytes':PIN['distributionBytes'],'handoffVerified':True,'descriptorIdentityVerified':True},'snapshotMilliseconds':125.0}
def test_exact_receipt(self):
with tempfile.TemporaryDirectory() as d:
path=Path(d)/'receipt.json';r=self.receipt();path.write_text(json.dumps(r));self.assertEqual(verify_report(path,'baseline'),r)
def test_false_or_incomplete_measurements_rejected(self):
deltas=[{'variant':'candidate'},{'poolRequestedAtProcessStart':16},{'moduleSha256':'0'*64},{'providerCalls':1},{'snapshotClosed':False},{'commandDirectoryClosed':False},{'temporaryRootEmpty':False},{'cleanupErrors':['error']},{'snapshotMilliseconds':float('nan')},{'snapshotMilliseconds':float('inf')},{'snapshotMilliseconds':0},{'sealedOutput':{}},{'status':'ok'}]
with tempfile.TemporaryDirectory() as d:
path=Path(d)/'receipt.json'
for delta in deltas:
with self.subTest(delta=delta):
path.write_text(json.dumps(self.receipt()|delta))
with self.assertRaises(RuntimeError):verify_report(path,'baseline')
def test_post_checks_continue_after_pack_failure(self):
calls=[];pin={'candidateModuleSha256':'expected','candidateSourceSha256':'expected','selectedFiles':{'one':{'sha256':'1'},'two':{'sha256':'2'}}}
def digest(path):calls.append(path.name);return path.name=='one' and '1' or 'wrong'
with patch('snapshot_streaming.closed_tree',side_effect=RuntimeError('pack changed')),patch('snapshot_streaming.sha',side_effect=digest):r=post_integrity(Path('/pack'),[],Path('/original'),pin)
self.assertEqual(calls,['snapshot-streaming-candidate.mjs','snapshot-streaming-candidate.ts','one','two']);self.assertEqual(r['checks'],{'pack':False,'candidateModule':False,'candidateSource':False,'one':True,'two':False});self.assertFalse(r['passed'])
def test_before_launch_cancellation_never_calls_factory(self):
owner=type('Owner',(),{})();child=DiagnosticChild(owner);child.cancel('between copies')
with self.assertRaises(RuntimeError):child.execute(lambda:self.fail('cancelled spawn'),90)
self.assertFalse(child.attempted);self.assertFalse(child.uncertain)
def test_during_publication_cancellation_retires_owned_handle(self):
class Process:
pid=123;returncode=0
def wait(self,timeout):return 0
class Owner:
stops=0
def observe(self,pid):child.cancel('during publication')
def stop(self,pid):self.stops+=1;return {'status':'stopped'}
def live(self):return []
owner=Owner();child=DiagnosticChild(owner)
with self.assertRaises(RuntimeError):child.execute(Process,90)
self.assertTrue(child.retired);self.assertEqual(owner.stops,1)
with self.assertRaises(RuntimeError):child.execute(lambda:self.fail('second spawn'),90)
class BaselineSourceTests(unittest.TestCase):
def test_safe_source_member_and_mismatched_hash_link_duplicate_rejected(self):
import tarfile,io,hashlib
name='packages/paperclip-runner/src/drivers/acpx/native-distribution-integrity.ts'
payload=b'baseline source';pin={'baselineSourceSha256':hashlib.sha256(payload).hexdigest()}
with tempfile.TemporaryDirectory() as d:
p=Path(d)/'source.tar'
for kind in ['valid','changed','link','duplicate','missing']:
with tarfile.open(p,'w') as t:
member=tarfile.TarInfo(name);member.size=len(payload)
if kind=='link':member.type=tarfile.SYMTYPE;member.linkname='/foreign';member.size=0
if kind!='missing':t.addfile(member,io.BytesIO(b'changed source!' if kind=='changed' else payload))
if kind=='duplicate':t.addfile(member,io.BytesIO(payload))
if kind=='valid':self.assertEqual(verify_baseline_source(p,pin),payload)
else:
with self.assertRaises(RuntimeError):verify_baseline_source(p,pin)
def test_candidate_report_must_use_candidate_hash(self):
with tempfile.TemporaryDirectory() as d:
p=Path(d)/'receipt.json';r=SnapshotStreamingTests().receipt()|{'variant':'candidate'};p.write_text(json.dumps(r))
with self.assertRaises(RuntimeError):verify_report(p,'candidate')
r['moduleSha256']=PIN['candidateModuleSha256'];p.write_text(json.dumps(r));self.assertEqual(verify_report(p,'candidate'),r)
if __name__=='__main__':unittest.main()