From f25fee35323bfafb9660b179e6adf01c48306737 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Wed, 11 Nov 2026 00:27:12 -0600 Subject: [PATCH] fix(native): hash runner artifacts and run archive tar off the event loop (#20722) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - One server process prepares native runs, renews every controller lease on timers, and serves every request > - On a run path, the server hashed the 31 MB runner binary with `readFileSync` (more than once per run), hashed provider packs and harness backups synchronously, and ran `tar` with `execFileSync` (timeouts up to 60 s, one with no timeout) > - On a slow disk, each of these calls freezes the whole process. A 103 s stall on a hosted instance lapsed every lease. The last event before that stall was immediately before a synchronous read and hash of the runner binary (see #20694 and #20709) > - This pull request moves those reads, hashes and archive commands off the event loop. It computes the runner binary digest once per boot > - The benefit is that a slow disk delays only the run that waits for it. Turns also start faster, because the runner binary is no longer re-read for every run ## Linked Issues or Issue Description Refs #20694, #20709 **What happened?** On a hosted instance, the server process stopped for 103 s without a restart. Every controller lease lapsed, and two tasks needed a manual Retry. The last run event before the gap was immediately before `matchesControllerRunnerArtifact`, which does `createHash("sha256").update(readFileSync(controllerRunnerBinary))`. Four workspace restores were running at the same time, and file I/O was very slow (a 43 s archive hash, 20 s snapshot decrypts). The trigger is not proven. But each synchronous call below blocks the event loop for as long as the disk is slow. **Expected behavior** File hashing and archive commands on a run path must not block the event loop. A slow disk must delay only the run that waits for it. **Steps to reproduce** 1. Make a remote runner binary slow to read (the test uses a FIFO whose writer waits), or make `tar` slow (the test puts a `tar` wrapper that sleeps first on `PATH`). 2. Call `resolveRemotePaperclipRunnerBinary`, `stageRemoteRunnerDirectory`, or `syncRemoteRunnerDirectoryOut` while a 10 ms timer runs. 3. Before this change, the timer stops for the whole slow step (814 ms, 1,764 ms and 3,379 ms). After this change, it keeps firing. ## What Changed - `server/src/services/native-runtime/off-loop-file-work.ts` (new): - `sha256FileStreamed` - `cachedFileSha256`: a streamed digest, reused while the file's device, inode, size, mtime and ctime do not change - `binaryTargetFromHeader`: reads 64 KiB; reads the rest of the file only when the header cannot answer, so the result equals `runnerBinaryTarget(readFileSync(path))` - `sha256DirectoryTreeStreamed` - `execFileBuffered`: `execFile` with stdin input, `maxBuffer`, `timeout` and `env`. It rejects on a non-zero exit, a timeout or a `maxBuffer` overflow, like `execFileSync`. - `native-session-executor.ts`: - `matchesControllerRunnerArtifact` uses `cachedFileSha256`. Each boot reads the binary once, not once or twice for every run. - The remote Claude daemon target is read from the binary header. - The bundled remote runner binary is checked by `verifiedBundledRemoteRunnerBinary`: the same checks and errors, with a cached digest. - Provider-pack upload verification uses the new `verifyRemoteProviderPackManifest`. It runs the same checks, in the same order, with the same errors. One generator drives both the sync and the streamed readers. - Harness backup build and verify on run paths use streamed digests (`buildNativeHarnessBackupManifestStreamed`, `verifyNativeHarnessBackupStreamed`). The digest format does not change. - `stageRemoteRunnerDirectory` (3 calls), `assertSafeRemoteCheckpointArchive` (3 calls) and the checkpoint extract use `execFileBuffered`, with the same arguments, timeouts and buffer limits. `stageRemoteRunnerFile` uses `readFile`. - `packages/paperclip-runner/src/live/runner-binary.ts`: the default daemon lookup (`resolvePackagedRunnerBinary`, used by the local and remote resolvers) classifies the daemon with `runnerBinaryFileTarget`. It reads the 64 KiB header, not the whole file. - `packages/paperclip-runner/src/live/bundled-remote-provider-pack.ts`: the release-manifest checks now live in one helper, `bundledRemoteRunnerRelease`. The sync `bundledRemoteRunnerBinary` and the server's streamed check both use it. - `native-codex-runner.ts`: - `resolveRemotePaperclipRunnerBinary` reads only the header. - The local `runnerDigest` uses `cachedFileSha256`. - New `verifiedBundledRemoteRunnerBinary`. **Ordering and atomicity.** I checked each converted call for code that relied on the sync call being atomic with respect to other runs: - **Checkpoint copy-back.** It extracts into a private `mkdtemp` directory. The replace sequence (`existsSync`, `rename`, `rename`, `rm`) stays synchronous, so no other run can see a half-installed checkpoint. This function already awaited a remote command before the extract, so this change adds no new interleaving with the target. - **Harness backup verify then restore.** The verify was sync, but the restore after it was already async. The window between them existed before. One owner serializes the turns of a session, so no other turn publishes a backup for that session while a turn verifies it. - **Staging and provider-pack hashing.** These read build-owned, immutable trees, or the private staging copies of a run. ## Verification - Red/green: `pnpm --filter @paperclipai/server exec vitest run src/services/native-runtime/off-loop-file-work.test.ts` - With `native-session-executor.ts` and `native-codex-runner.ts` from `origin/master`: 3 failed. The longest timer gaps were 814 ms (runner binary target read), 1,764 ms (directory staging tar) and 3,379 ms (checkpoint list, size check and extract). - With this change: 6 passed. The gaps stay under 400 ms. The same file also checks the digest cache, the header fallback, and the `execFileSync`-like errors (exit code, timeout kill, `ERR_CHILD_PROCESS_STDIO_MAXBUFFER`). - `native-session-executor.test.ts`: - The streamed provider-pack verifier returns the same manifest as the sync reader, and rejects a changed asset tree with the same error. - The streamed harness backup manifest equals the sync one. - The streamed verify agrees with the sync verify, for a valid backup and for a corrupt one. - Run together: `native-session-executor`, `off-loop-file-work`, `native-codex-runner`, `setup-readiness`, `antigravity-connection-probe` (1,653 passed). `run-completion.test.ts` passed all 153 tests. Its `afterAll` embedded-Postgres cleanup timed out locally, which is not related to this change; CI runs it. - `pnpm --filter @paperclipai/server exec tsc --noEmit` passes. ## Risks - **Turn start.** This change adds no wait before `turn.start`; it removes work. The first run of a boot streams the runner binary once. Later runs reuse the digest after one `stat`. - **Cache staleness.** The cache keys on device, inode, size, mtime and ctime. An in-place rewrite changes mtime and ctime, and a replaced file changes its inode, so a changed binary is hashed again. - **stderr.** `execFileSync` wrote a child's stderr to the server's stderr. `execFileBuffered` captures it. Tar warnings no longer appear in server logs; errors and exit codes are unchanged. - **Persisted state.** No persisted shape changes. Old saved state and new code (and the reverse) read the same manifests and digests. - **Not converted in this PR:** small metadata syscalls (`existsSync`, `lstatSync`, `readdirSync`, small JSON reads and writes, renames) in the native state machines. Some of these rely on running synchronously. Also not converted: the sync harness backup verify inside the sync restart-authority checks, and the sync stamp verify in `native-harness-backup-storage.ts`. Both are listed for a follow-up. ## Model Used - Anthropic Claude Opus 5.5 (`claude-opus-5-5`), used through Claude Code with tool use, shell, and code execution. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes (code comments carry the contract) - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude --- packages/paperclip-runner/src/index.ts | 3 +- .../src/live/bundled-remote-provider-pack.ts | 20 +- packages/paperclip-runner/src/live/index.ts | 2 +- .../src/live/runner-binary.test.ts | 25 +- .../src/live/runner-binary.ts | 98 ++++++- .../src/live/runnerd-codex-transport.ts | 15 +- .../antigravity-connection-probe.test.ts | 5 +- .../services/heartbeat/run-completion.test.ts | 6 +- .../native-codex-runner.test.ts | 16 +- .../native-runtime/native-codex-runner.ts | 74 +++++- .../native-session-executor.test.ts | 27 ++ .../native-runtime/native-session-executor.ts | 245 +++++++++++++++--- .../native-runtime/off-loop-file-work.test.ts | 185 +++++++++++++ .../native-runtime/off-loop-file-work.ts | 134 ++++++++++ server/src/vendor/paperclip-runner/index.ts | 4 + 15 files changed, 784 insertions(+), 75 deletions(-) create mode 100644 server/src/services/native-runtime/off-loop-file-work.test.ts create mode 100644 server/src/services/native-runtime/off-loop-file-work.ts diff --git a/packages/paperclip-runner/src/index.ts b/packages/paperclip-runner/src/index.ts index b5dc415aef8..25f2a3c0217 100644 --- a/packages/paperclip-runner/src/index.ts +++ b/packages/paperclip-runner/src/index.ts @@ -95,6 +95,7 @@ export { NativeEnvironmentProbeCleanupError, runnerCodexDynamicToolsFit, defaultCapabilityRunnerdBinary, + defaultCapabilityRunnerdBinaryAsync, readRunnerdArtifactBinding, drainRetainedRunnerdMaintenanceOperations, coldAdmissionTimeoutMs, @@ -124,7 +125,7 @@ export * from "./compatibility.js"; export { probeAcpxClaudeInstallation, probeAcpxGrokInstallation } from "./drivers/acpx/installation-integrity.js"; export { probeAcpxPiInstallation, probeAcpxKimiInstallation } from "./drivers/acpx/profile-installation.js"; -export { PROVIDER_PACK_COMPATIBILITY, bundledRemoteProviderPackManifestPath, bundledRemoteRunnerBinary } from "./live/bundled-remote-provider-pack.js"; +export { BUNDLED_REMOTE_RUNNER_IDENTITY_MISMATCH, PROVIDER_PACK_COMPATIBILITY, bundledRemoteProviderPackManifestPath, bundledRemoteRunnerBinary, bundledRemoteRunnerRelease } from "./live/bundled-remote-provider-pack.js"; export { RunnerdMuseDriver, describeRunnerdMuseDriver, type RunnerdMuseDriverOptions } from "./drivers/muse/runnerd-muse-driver.js"; export { RunnerdDotDriver, type RunnerdDotDriverOptions } from "./drivers/dot/runnerd-dot-driver.js"; diff --git a/packages/paperclip-runner/src/live/bundled-remote-provider-pack.ts b/packages/paperclip-runner/src/live/bundled-remote-provider-pack.ts index 7b0f6265692..1803f564ff8 100644 --- a/packages/paperclip-runner/src/live/bundled-remote-provider-pack.ts +++ b/packages/paperclip-runner/src/live/bundled-remote-provider-pack.ts @@ -24,8 +24,11 @@ export function bundledRemoteProviderPackManifestPath(moduleUrl = import.meta.ur throw new Error("Bundled remote provider pack requires a package file layout"); } -/** A macOS controller must authorize Linux bytes for its Daytona daemon. */ -export function bundledRemoteRunnerBinary(moduleUrl = import.meta.url, target: BundledRemoteProviderPackTarget = "linux-x64"): string { +/** + * The packaged Linux daemon and its release digest, after the release-manifest + * checks. Callers verify the bytes; the server streams that digest. + */ +export function bundledRemoteRunnerRelease(moduleUrl = import.meta.url, target: BundledRemoteProviderPackTarget = "linux-x64"): { binary: string; sha256: string } { const manifestPath = bundledRemoteProviderPackManifestPath(moduleUrl, target); const outputRoot = resolve(dirname(manifestPath), "../.."); const release = JSON.parse(readFileSync(join(outputRoot, "bin/release-manifest.json"), "utf8")); @@ -34,10 +37,17 @@ export function bundledRemoteRunnerBinary(moduleUrl = import.meta.url, target: B artifact?.path !== `${target}/paperclip-runnerd` || !/^sha256:[a-f0-9]{64}$/.test(artifact.sha256 ?? "")) { throw new Error("runner_remote_artifact_unavailable: install the complete Paperclip release with its Linux daemon"); } - const binary = join(outputRoot, "bin", artifact.path); + return { binary: join(outputRoot, "bin", artifact.path), sha256: artifact.sha256 }; +} + +export const BUNDLED_REMOTE_RUNNER_IDENTITY_MISMATCH = "runner_remote_artifact_incompatible: packaged Linux daemon identity mismatch"; + +/** A macOS controller must authorize Linux bytes for its Daytona daemon. */ +export function bundledRemoteRunnerBinary(moduleUrl = import.meta.url, target: BundledRemoteProviderPackTarget = "linux-x64"): string { + const { binary, sha256 } = bundledRemoteRunnerRelease(moduleUrl, target); const bytes = readFileSync(binary); - if (runnerBinaryTarget(bytes) !== target || "sha256:" + createHash("sha256").update(bytes).digest("hex") !== artifact.sha256) { - throw new Error("runner_remote_artifact_incompatible: packaged Linux daemon identity mismatch"); + if (runnerBinaryTarget(bytes) !== target || "sha256:" + createHash("sha256").update(bytes).digest("hex") !== sha256) { + throw new Error(BUNDLED_REMOTE_RUNNER_IDENTITY_MISMATCH); } return binary; } diff --git a/packages/paperclip-runner/src/live/index.ts b/packages/paperclip-runner/src/live/index.ts index 4a65609addd..8ea4740c43b 100644 --- a/packages/paperclip-runner/src/live/index.ts +++ b/packages/paperclip-runner/src/live/index.ts @@ -13,4 +13,4 @@ export { probeAcpxCursorInstallation, probeAcpxPiInstallation, probeAcpxKimiInst export { probeKimiModels, KimiMetadataProbeError, type KimiVerifiedModelMetadata } from "../drivers/acpx/kimi-models.js"; export { probeKimiMetadata, validateKimiMetadata, kimiMetadataProbeDigest, kimiMetadataProbeNodeDigest, type KimiMetadataResult } from "../drivers/acpx/kimi-metadata-probe.js"; -export { bundledRemoteProviderPackManifestPath, bundledRemoteRunnerBinary } from "./bundled-remote-provider-pack.js"; +export { BUNDLED_REMOTE_RUNNER_IDENTITY_MISMATCH, bundledRemoteProviderPackManifestPath, bundledRemoteRunnerBinary, bundledRemoteRunnerRelease } from "./bundled-remote-provider-pack.js"; diff --git a/packages/paperclip-runner/src/live/runner-binary.test.ts b/packages/paperclip-runner/src/live/runner-binary.test.ts index 6f27822fc6b..5fda227c4ef 100644 --- a/packages/paperclip-runner/src/live/runner-binary.test.ts +++ b/packages/paperclip-runner/src/live/runner-binary.test.ts @@ -2,7 +2,7 @@ import { mkdtempSync, mkdirSync, rmSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { afterEach, describe, expect, it } from "vitest"; -import { resolvePackagedRunnerBinary, runnerBinaryTarget } from "./runner-binary.js"; +import { resolvePackagedRunnerBinary, resolvePackagedRunnerBinaryAsync, runnerBinaryFileTarget, runnerBinaryFileTargetAsync, runnerBinaryTarget } from "./runner-binary.js"; const roots: string[] = []; afterEach(() => { for (const root of roots.splice(0)) rmSync(root, { recursive: true, force: true }); }); function header(target: string) { @@ -46,6 +46,29 @@ describe("packaged native daemon target", () => { const root = fixture(); expect(resolvePackagedRunnerBinary(join(root, "dist"), "linux", "x64")).toBeNull(); const legacy = binary(root, null, header("linux-x64")); expect(resolvePackagedRunnerBinary(join(root, "dist"), "linux", "x64")).toBe(legacy); }); + it("classifies a daemon file from its header and matches a whole-file read", () => { + const root = fixture(); + // A 40 MB daemon whose target is in its first bytes. + const large = Buffer.alloc(40 * 1024 * 1024); header("linux-x64").copy(large); + expect(runnerBinaryFileTarget(binary(root, "linux-x64", large))).toBe("linux-x64"); + // A PE header placed past the first 64 KiB needs the rest of the file. + const pe = Buffer.alloc(70_000); pe.set([0x4d, 0x5a]); pe.writeUInt32LE(68_000, 60); + pe.writeUInt32LE(0x00004550, 68_000); pe.writeUInt16LE(0x8664, 68_004); + expect(runnerBinaryTarget(pe)).toBe("win32-x64"); + expect(runnerBinaryFileTarget(binary(root, "win32-x64", pe))).toBe("win32-x64"); + expect(runnerBinaryFileTarget(binary(root, "empty", Buffer.alloc(0)))).toBeNull(); + }); + it("resolves the packaged daemon asynchronously with the same selection and errors", async () => { + const root = fixture(); + const selected = binary(root, "linux-x64", header("linux-x64")); + expect(await resolvePackagedRunnerBinaryAsync(join(root, "dist"), "linux", "x64")).toBe(resolvePackagedRunnerBinary(join(root, "dist"), "linux", "x64")); + expect(await resolvePackagedRunnerBinaryAsync(join(root, "dist"), "linux", "x64")).toBe(selected); + expect(await runnerBinaryFileTargetAsync(selected)).toBe("linux-x64"); + expect(await resolvePackagedRunnerBinaryAsync(join(root, "dist"), "darwin", "arm64")).toBeNull(); + binary(root, "linux-arm64", header("linux-x64")); + await expect(resolvePackagedRunnerBinaryAsync(join(root, "dist"), "linux", "arm64")).rejects.toThrow("packaged daemon does not match linux-arm64"); + expect(() => resolvePackagedRunnerBinary(join(root, "dist"), "linux", "arm64")).toThrow("packaged daemon does not match linux-arm64"); + }); it("rejects malformed, truncated and unknown binary formats", () => { for (const bytes of [Buffer.alloc(0), Buffer.alloc(64), header("linux-x64").subarray(0, 19)]) expect(runnerBinaryTarget(bytes)).toBeNull(); }); diff --git a/packages/paperclip-runner/src/live/runner-binary.ts b/packages/paperclip-runner/src/live/runner-binary.ts index 0537c10b0e9..d4be8d5d2a4 100644 --- a/packages/paperclip-runner/src/live/runner-binary.ts +++ b/packages/paperclip-runner/src/live/runner-binary.ts @@ -1,4 +1,5 @@ -import { existsSync, readFileSync } from "node:fs"; +import { closeSync, existsSync, openSync, readFileSync, readSync, statSync } from "node:fs"; +import { open, stat } from "node:fs/promises"; import { resolve } from "node:path"; /** Native binary headers bind the packaged executable to its actual target. */ @@ -22,13 +23,96 @@ export function runnerBinaryTarget(bytes: Buffer): string | null { return null; } -export function resolvePackagedRunnerBinary(outputRoot: string, platform = process.platform, architecture = process.arch): string | null { - const target = `${platform}-${architecture}`, executable = platform === "win32" ? "paperclip-runnerd.exe" : "paperclip-runnerd"; - for (const binary of [resolve(outputRoot, "bin", target, executable), resolve(outputRoot, "bin", executable)]) { - if (!existsSync(binary)) continue; - if (runnerBinaryTarget(readFileSync(binary)) !== target) { - throw new Error(`runner_runtime_incompatible: packaged daemon does not match ${target}; install the matching Paperclip release`); +const RUNNER_BINARY_HEADER_BYTES = 64 * 1024; + +/** + * `runnerBinaryTarget(readFileSync(path))`, reading only the 64 KiB header + * unless the header cannot answer. A daemon is tens of megabytes; reading all + * of it to classify the target blocked the server's event loop on a slow disk. + */ +export function runnerBinaryFileTarget(path: string): string | null { + // The packaged daemon is release-owned and does not change while the process + // runs. After the first read, each run costs one stat: the target is reused + // while the device, inode, size, modification and change times match. + const metadata = statSync(path, { bigint: true }); + const key = `${metadata.dev}:${metadata.ino}:${metadata.size}:${metadata.mtimeNs}:${metadata.ctimeNs}`; + const cached = targetCache.get(path); + if (cached?.key === key) return cached.target; + const target = readRunnerBinaryFileTarget(path); + targetCache.set(path, { key, target }); + return target; +} + +const targetCache = new Map(); + +/** `runnerBinaryFileTarget` without blocking the event loop; shares its cache. */ +export async function runnerBinaryFileTargetAsync(path: string): Promise { + const metadata = await stat(path, { bigint: true }); + const key = `${metadata.dev}:${metadata.ino}:${metadata.size}:${metadata.mtimeNs}:${metadata.ctimeNs}`; + const cached = targetCache.get(path); + if (cached?.key === key) return cached.target; + const handle = await open(path, "r"); + let target: string | null; + try { + const header = Buffer.alloc(RUNNER_BINARY_HEADER_BYTES); + let length = 0; + let ended = false; + while (length < header.length) { + const { bytesRead } = await handle.read(header, length, header.length - length, null); + if (bytesRead === 0) { ended = true; break; } + length += bytesRead; } + target = runnerBinaryTarget(header.subarray(0, length)); + if (target === null && !ended) target = runnerBinaryTarget(Buffer.concat([header.subarray(0, length), await handle.readFile()])); + } finally { + await handle.close(); + } + targetCache.set(path, { key, target }); + return target; +} + +function readRunnerBinaryFileTarget(path: string): string | null { + const descriptor = openSync(path, "r"); + try { + const header = Buffer.alloc(RUNNER_BINARY_HEADER_BYTES); + let length = 0; + let ended = false; + while (length < header.length) { + const bytesRead = readSync(descriptor, header, length, header.length - length, null); + if (bytesRead === 0) { ended = true; break; } + length += bytesRead; + } + const target = runnerBinaryTarget(header.subarray(0, length)); + return target !== null || ended ? target : runnerBinaryTarget(readFileSync(path)); + } finally { + closeSync(descriptor); + } +} + +function packagedRunnerBinaryCandidates(outputRoot: string, platform: string, architecture: string) { + const target = `${platform}-${architecture}`, executable = platform === "win32" ? "paperclip-runnerd.exe" : "paperclip-runnerd"; + return { target, binaries: [resolve(outputRoot, "bin", target, executable), resolve(outputRoot, "bin", executable)] }; +} + +const packagedRunnerMismatch = (target: string) => + new Error(`runner_runtime_incompatible: packaged daemon does not match ${target}; install the matching Paperclip release`); + +export function resolvePackagedRunnerBinary(outputRoot: string, platform = process.platform, architecture = process.arch): string | null { + const { target, binaries } = packagedRunnerBinaryCandidates(outputRoot, platform, architecture); + for (const binary of binaries) { + if (!existsSync(binary)) continue; + if (runnerBinaryFileTarget(binary) !== target) throw packagedRunnerMismatch(target); + return binary; + } + return null; +} + +/** `resolvePackagedRunnerBinary` without blocking the event loop. */ +export async function resolvePackagedRunnerBinaryAsync(outputRoot: string, platform = process.platform, architecture = process.arch): Promise { + const { target, binaries } = packagedRunnerBinaryCandidates(outputRoot, platform, architecture); + for (const binary of binaries) { + if (!await stat(binary).then(() => true, () => false)) continue; + if (await runnerBinaryFileTargetAsync(binary) !== target) throw packagedRunnerMismatch(target); return binary; } return null; diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index e3e972a6a05..03dcf56ff2d 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -103,7 +103,7 @@ import { prepareIsolatedCodexRecoveryHome, releaseMaterializedNativeRuntimeSkills, } from "../drivers/runtime-context-materializer.js"; -import { resolvePackagedRunnerBinary } from "./runner-binary.js"; +import { resolvePackagedRunnerBinary, resolvePackagedRunnerBinaryAsync } from "./runner-binary.js"; import { RUNNERD_CANONICAL_ITEM, RUNNERD_RUNTIME_ACTIVITY } from "../drivers/codex/codex-driver-values.js"; import { crashPoint } from "../crash-point.js"; import { isTerminalOperation, settleEarlierToolCalls, staleTerminalCallResult } from "./terminal-call-freshness.js"; @@ -8834,7 +8834,18 @@ export function defaultCapabilityRunnerdBinary(platform = process.platform, arch const outputRoot = resolve(fileURLToPath(new URL("../", import.meta.url))); const staged = resolvePackagedRunnerBinary(outputRoot, platform, architecture) ?? resolvePackagedRunnerBinary(resolve(packageRoot, "dist"), platform, architecture); - if (staged) return staged; + return staged ?? unstagedCapabilityRunnerdBinary(platform, architecture); +} + +/** `defaultCapabilityRunnerdBinary` without blocking the event loop. */ +export async function defaultCapabilityRunnerdBinaryAsync(platform = process.platform, architecture = process.arch): Promise { + const outputRoot = resolve(fileURLToPath(new URL("../", import.meta.url))); + const staged = await resolvePackagedRunnerBinaryAsync(outputRoot, platform, architecture) + ?? await resolvePackagedRunnerBinaryAsync(resolve(packageRoot, "dist"), platform, architecture); + return staged ?? unstagedCapabilityRunnerdBinary(platform, architecture); +} + +function unstagedCapabilityRunnerdBinary(platform: string, architecture: string): string { if (platform !== process.platform || architecture !== process.arch) { throw new Error(`runner_remote_artifact_unavailable: install the Paperclip release containing the ${platform}-${architecture} daemon or configure PAPERCLIP_RUNNER_REMOTE_BINARY_PATH`); } diff --git a/server/src/__tests__/antigravity-connection-probe.test.ts b/server/src/__tests__/antigravity-connection-probe.test.ts index d49fc933eed..c35889b58f2 100644 --- a/server/src/__tests__/antigravity-connection-probe.test.ts +++ b/server/src/__tests__/antigravity-connection-probe.test.ts @@ -12,7 +12,10 @@ vi.mock("../services/native-runtime/acpx-qualification.js", () => ({ resolveAnti vi.mock("../services/native-runtime/antigravity-runner-credentials.js", () => ({ prepareAntigravityRunnerCredentials: mocks.credentials, antigravityCredentialTurn: mocks.turn, })); -vi.mock("../services/native-runtime/native-codex-runner.js", () => ({ resolvePaperclipRunnerBinary: () => "/owned/runnerd" })); +vi.mock("../services/native-runtime/native-codex-runner.js", () => ({ + resolvePaperclipRunnerBinary: () => "/owned/runnerd", + resolvePaperclipRunnerBinaryAsync: async () => "/owned/runnerd", +})); import { antigravityConnectionProbeFailure, testAntigravityConnection } from "../services/antigravity-connection-probe.js"; import { NativeEnvironmentProbeCleanupError, resolveAcpxRuntimeRoot } from "../vendor/paperclip-runner/index.js"; diff --git a/server/src/services/heartbeat/run-completion.test.ts b/server/src/services/heartbeat/run-completion.test.ts index 89ae9892542..c0962b38469 100644 --- a/server/src/services/heartbeat/run-completion.test.ts +++ b/server/src/services/heartbeat/run-completion.test.ts @@ -64,7 +64,11 @@ const diagnosticNativeFixture = vi.hoisted(() => ({ })); vi.mock("../native-runtime/native-codex-runner.js", async importOriginal => { const actual = await importOriginal(); - return { ...actual, resolvePaperclipRunnerBinary: () => diagnosticNativeFixture.enabled ? process.execPath : actual.resolvePaperclipRunnerBinary() }; + return { + ...actual, + resolvePaperclipRunnerBinary: () => diagnosticNativeFixture.enabled ? process.execPath : actual.resolvePaperclipRunnerBinary(), + resolvePaperclipRunnerBinaryAsync: async () => diagnosticNativeFixture.enabled ? process.execPath : actual.resolvePaperclipRunnerBinaryAsync(), + }; }); vi.mock("../../vendor/paperclip-runner/index.js", async importOriginal => { const actual = await importOriginal(); diff --git a/server/src/services/native-runtime/native-codex-runner.test.ts b/server/src/services/native-runtime/native-codex-runner.test.ts index c7a5a9ad02e..bd664953290 100644 --- a/server/src/services/native-runtime/native-codex-runner.test.ts +++ b/server/src/services/native-runtime/native-codex-runner.test.ts @@ -66,7 +66,7 @@ describe("selected remote daemon target", () => { const binary = join(root, "paperclip-runnerd"); try { await writeFile(binary, header(`${platform}-${architecture}`)); - const select = vi.spyOn(runner, "defaultCapabilityRunnerdBinary").mockReturnValue(binary); + const select = vi.spyOn(runner, "defaultCapabilityRunnerdBinaryAsync").mockResolvedValue(binary); const target = remote(`${os}\n${arch}\n`); expect(await resolveRemotePaperclipRunnerBinary({ runner: target, remoteCwd: "/selected-workspace" })).toBe(binary); expect(select).toHaveBeenCalledExactlyOnceWith(platform, architecture); @@ -83,7 +83,7 @@ describe("selected remote daemon target", () => { const controllerBinary = join(root, "controller-runnerd"); await writeFile(controllerBinary, header(`darwin-${controllerArch}`), { mode: 0o700 }); vi.stubEnv("PAPERCLIP_RUNNER_BINARY", controllerBinary); - const select = vi.spyOn(runner, "defaultCapabilityRunnerdBinary").mockReturnValue(binary); + const select = vi.spyOn(runner, "defaultCapabilityRunnerdBinaryAsync").mockResolvedValue(binary); const target = remote(`Linux\n${targetArch === "arm64" ? "aarch64" : "x86_64"}\n`); expect(await resolveRemotePaperclipRunnerBinary({ runner: target, remoteCwd: "/sandbox-workspace" })).toBe(binary); expect(select).toHaveBeenCalledExactlyOnceWith("linux", targetArch); @@ -100,7 +100,7 @@ describe("selected remote daemon target", () => { await writeFile(fresh, Buffer.concat([header(`${platform}-${architecture}`), Buffer.from("fresh")]), { mode: 0o700 }); await writeFile(stale, Buffer.concat([header(`${platform}-${architecture}`), Buffer.from("stale")]), { mode: 0o700 }); vi.stubEnv("PAPERCLIP_RUNNER_BINARY", fresh); - const select = vi.spyOn(runner, "defaultCapabilityRunnerdBinary").mockReturnValue(stale); + const select = vi.spyOn(runner, "defaultCapabilityRunnerdBinaryAsync").mockResolvedValue(stale); const target = remote(`${platform === "linux" ? "Linux" : "Darwin"}\n${architecture === "arm64" ? "arm64" : "x86_64"}\n`); expect(await resolveRemotePaperclipRunnerBinary({ runner: target, remoteCwd: "/workspace" })).toBe(fresh); expect(select).not.toHaveBeenCalled(); @@ -115,7 +115,7 @@ describe("selected remote daemon target", () => { if (scenario !== "missing") await writeFile(override, scenario === "malformed" ? Buffer.alloc(64) : header("linux-arm64"), { mode: 0o700 }); await writeFile(fallback, header("linux-x64"), { mode: 0o700 }); vi.stubEnv("PAPERCLIP_RUNNER_BINARY", scenario === "relative" ? "relative-runnerd" : override); - vi.spyOn(runner, "defaultCapabilityRunnerdBinary").mockReturnValue(fallback); + vi.spyOn(runner, "defaultCapabilityRunnerdBinaryAsync").mockResolvedValue(fallback); const request = resolveRemotePaperclipRunnerBinary({ runner: remote("Linux\nx86_64\n"), remoteCwd: "/workspace" }); if (scenario === "missing") expect(await request).toBe(fallback); else await expect(request).rejects.toThrow(scenario === "relative" ? "PAPERCLIP_RUNNER_BINARY must be an absolute path" : "runner_remote_artifact_platform_mismatch"); @@ -128,7 +128,7 @@ describe("selected remote daemon target", () => { vi.stubEnv("PAPERCLIP_RUNNER_BINARY", "relative-controller-runnerd"); const binary = join(root, "linux-runnerd"); await writeFile(binary, header("linux-arm64")); - const select = vi.spyOn(runner, "defaultCapabilityRunnerdBinary").mockReturnValue(binary); + const select = vi.spyOn(runner, "defaultCapabilityRunnerdBinaryAsync").mockResolvedValue(binary); expect(await resolveRemotePaperclipRunnerBinary({ runner: remote("Linux\naarch64\n"), remoteCwd: "/workspace" })).toBe(binary); expect(select).toHaveBeenCalledExactlyOnceWith("linux", "arm64"); } finally { await rm(root, { recursive: true, force: true }); } @@ -140,7 +140,7 @@ describe("selected remote daemon target", () => { vi.stubGlobal("process", { ...process, platform: "linux", arch: "arm64" }); vi.stubEnv("PAPERCLIP_RUNNER_BINARY", "invalid-relative-controller-override"); if (scenario !== "missing") await writeFile(binary, scenario === "malformed" ? Buffer.alloc(64) : header(scenario === "matching" ? "linux-arm64" : "linux-x64")); - const select = vi.spyOn(runner, "defaultCapabilityRunnerdBinary"); + const select = vi.spyOn(runner, "defaultCapabilityRunnerdBinaryAsync"); const request = resolveRemotePaperclipRunnerBinary({ runner: remote("Linux\naarch64\n"), remoteCwd: "/workspace", configuredPath: binary }); if (scenario === "matching") expect(await request).toBe(binary); else await expect(request).rejects.toThrow(scenario === "missing" ? "runner_remote_artifact_unavailable" : "runner_remote_artifact_platform_mismatch"); @@ -148,7 +148,7 @@ describe("selected remote daemon target", () => { } finally { await rm(root, { recursive: true, force: true }); } }); it("keeps missing package targets actionable without a controller or legacy fallback", async () => { - const select = vi.spyOn(runner, "defaultCapabilityRunnerdBinary").mockImplementation(() => { throw new Error("runner_remote_artifact_unavailable: install the matching release"); }); + const select = vi.spyOn(runner, "defaultCapabilityRunnerdBinaryAsync").mockRejectedValue(new Error("runner_remote_artifact_unavailable: install the matching release")); await expect(resolveRemotePaperclipRunnerBinary({ runner: remote("Linux\naarch64\n"), remoteCwd: "/workspace" })).rejects.toThrow("runner_remote_artifact_unavailable"); expect(select).toHaveBeenCalledExactlyOnceWith("linux", "arm64"); }); @@ -156,7 +156,7 @@ describe("selected remote daemon target", () => { ["Linux\naarch64\n", { exitCode: 1 }], ["Linux\naarch64\n", { timedOut: true }], ["Linux\nriscv64\n", {}], ["FreeBSD\nx86_64\n", {}], ["", {}], ])("fails a missing, failed or unsupported platform probe (%s)", async (stdout, overrides) => { - const select = vi.spyOn(runner, "defaultCapabilityRunnerdBinary"); + const select = vi.spyOn(runner, "defaultCapabilityRunnerdBinaryAsync"); await expect(resolveRemotePaperclipRunnerBinary({ runner: remote(stdout as string, overrides), remoteCwd: "/workspace" })).rejects.toThrow("runner_remote_artifact_platform_mismatch"); expect(select).not.toHaveBeenCalled(); }); diff --git a/server/src/services/native-runtime/native-codex-runner.ts b/server/src/services/native-runtime/native-codex-runner.ts index 9e184e2a202..af65604e550 100644 --- a/server/src/services/native-runtime/native-codex-runner.ts +++ b/server/src/services/native-runtime/native-codex-runner.ts @@ -1,6 +1,7 @@ import { spawn, type ChildProcess } from "node:child_process"; import { createHash } from "node:crypto"; -import { accessSync, chmodSync, constants, mkdirSync, readFileSync } from "node:fs"; +import { accessSync, chmodSync, constants, mkdirSync } from "node:fs"; +import { access } from "node:fs/promises"; import { dirname, isAbsolute, resolve } from "node:path"; import { fileURLToPath } from "node:url"; import { and, eq } from "drizzle-orm"; @@ -13,12 +14,16 @@ import { agentSessionGoalActions, agentTaskSessions } from "@paperclipai/db"; import { resolvePaperclipInstanceRoot } from "../../home-paths.js"; import { failRunnerGoalAction } from "../runner-goals.js"; import { + BUNDLED_REMOTE_RUNNER_IDENTITY_MISMATCH, + bundledRemoteRunnerRelease, createPaperclipRunnerAuthorizedToolSet, defaultCapabilityRunnerdBinary, + defaultCapabilityRunnerdBinaryAsync, runnerBinaryTarget, type PaperclipSemanticToolDefinition, } from "../../vendor/paperclip-runner/index.js"; import { runnerPrpCoordinator } from "./runner-prp-coordinator.js"; +import { binaryTargetFromHeader, cachedFileSha256 } from "./off-loop-file-work.js"; const moduleDirectory = dirname(fileURLToPath(import.meta.url)); const RUNNER_VERSION = "paperclip-runner-v1"; @@ -218,6 +223,42 @@ export function resolvePaperclipRunnerBinary( ); } +/** `resolvePaperclipRunnerBinary` without blocking the event loop (run paths). */ +export async function resolvePaperclipRunnerBinaryAsync( + configuredPath = controllerPaperclipRunnerBinaryConfiguration(), +): Promise { + assertPaperclipRunnerBinaryConfiguration(configuredPath); + if (configuredPath) { + try { + await access(configuredPath, constants.R_OK | (process.platform === "win32" ? 0 : constants.X_OK)); + return configuredPath; + } catch { + // Preserve the existing fallback for an unavailable configured path. + } + } + const candidates = [ + // The runner owns its compiled/vendored root and verifies the target header. + // Never prefer a flat build-host executable over a staged release target. + await defaultCapabilityRunnerdBinaryAsync(), + resolve( + moduleDirectory, + "../../../../packages/paperclip-runner/runner/target/release", + executableName(), + ), + ].filter((candidate): candidate is string => Boolean(candidate)); + for (const candidate of candidates) { + try { + await access(candidate, constants.R_OK | (process.platform === "win32" ? 0 : constants.X_OK)); + return candidate; + } catch { + // Continue through the fixed production and workspace locations. + } + } + throw new Error( + "paperclip_runner_binary_missing: build @paperclipai/paperclip-runner or set PAPERCLIP_RUNNER_BINARY", + ); +} + export async function resolveRemotePaperclipRunnerBinary(input: { runner: CommandManagedRuntimeRunner; remoteCwd: string; @@ -236,17 +277,34 @@ export async function resolveRemotePaperclipRunnerBinary(input: { const controllerOverride = platform === process.platform && architecture === process.arch ? process.env.PAPERCLIP_RUNNER_BINARY : undefined; const binary = configured || (controllerOverride - ? resolvePaperclipRunnerBinary(controllerOverride) - : defaultCapabilityRunnerdBinary(platform, architecture)); - let bytes: Buffer; - try { bytes = readFileSync(binary); } + ? await resolvePaperclipRunnerBinaryAsync(controllerOverride) + : await defaultCapabilityRunnerdBinaryAsync(platform, architecture)); + // The header names the target; reading the whole daemon here blocked the + // event loop on every remote run. + let target: string | null; + try { target = await binaryTargetFromHeader(binary, runnerBinaryTarget); } catch (cause) { throw new Error(`runner_remote_artifact_unavailable: install the Paperclip release containing the ${platform}-${architecture} daemon or configure PAPERCLIP_RUNNER_REMOTE_BINARY_PATH`, { cause }); } - if (runnerBinaryTarget(bytes) !== `${platform}-${architecture}`) { + if (target !== `${platform}-${architecture}`) { throw new Error(`runner_remote_artifact_platform_mismatch: the selected daemon does not match ${platform}-${architecture}; install the matching Paperclip release or configure PAPERCLIP_RUNNER_REMOTE_BINARY_PATH`); } return binary; } +/** + * `bundledRemoteRunnerBinary()` without blocking the event loop: the same + * release-manifest, target and digest checks and errors, with the digest + * streamed once per boot (it is cached while the file is unchanged). + */ +export async function verifiedBundledRemoteRunnerBinary( + target: "linux-x64" | "linux-arm64" = "linux-x64", +): Promise { + const { binary, sha256 } = bundledRemoteRunnerRelease(undefined, target); + if (await binaryTargetFromHeader(binary, runnerBinaryTarget) !== target || await cachedFileSha256(binary) !== sha256) { + throw new Error(BUNDLED_REMOTE_RUNNER_IDENTITY_MISMATCH); + } + return binary; +} + export function buildNativeRunnerArguments(input: { connectUrl: string; stateDirectory: string; @@ -363,8 +421,8 @@ export async function executeNativeCodexRunner(input: { startedAt: string; }) => Promise; }): Promise { - const binary = input.runnerBinary ?? resolvePaperclipRunnerBinary(); - const runnerDigest = `sha256:${createHash("sha256").update(readFileSync(binary)).digest("hex")}`; + const binary = input.runnerBinary ?? await resolvePaperclipRunnerBinaryAsync(); + const runnerDigest = await cachedFileSha256(binary); const runtimeRoot = input.runtimeRoot ? resolve(input.runtimeRoot) : resolve(resolvePaperclipInstanceRoot(), "runtime", "paperclip-runner"); diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index e37262c5b72..65aa6b99f5a 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -390,7 +390,11 @@ vi.mock("../issue-recovery-actions.js", () => ({ vi.mock("./native-codex-runner.js", () => ({ resolvePaperclipRunnerBinary: state.resolveRunnerBinary, + resolvePaperclipRunnerBinaryAsync: async (...args: unknown[]) => state.resolveRunnerBinary(...(args as [])), resolveRemotePaperclipRunnerBinary: state.resolveRemoteRunnerBinary, + // Resolves through the (mocked) vendor bundle, like the production checks. + verifiedBundledRemoteRunnerBinary: async () => + (await import("../../vendor/paperclip-runner/index.js")).bundledRemoteRunnerBinary(), })); import { @@ -398,6 +402,7 @@ import { buildNativeProviderEnvironment, resolveNativeProviderEnvironment, buildNativeHarnessBackupManifest, + buildNativeHarnessBackupManifestStreamed, cancelNativeSession, closeWarmNativeSessionsForEnvironment, claimWarmNativeInstructionCopy, @@ -448,6 +453,7 @@ import { stopStaleRemoteRunnerBeforeBootstrap, verifyRemoteRunnerOrSettledExit, readRemoteProviderPackManifest, + verifyRemoteProviderPackManifest, prepareRemoteProviderPackOnTarget, resolveRemoteProviderPackSource, readBundledRemoteProviderPackManifest, @@ -474,6 +480,7 @@ import { remoteProviderStateUnreadable, retryTransientRemoteCheckpointRead, verifyNativeHarnessBackup, + verifyNativeHarnessBackupStreamed, shouldRestoreNativeHarnessBackupIntoSandbox, remoteProviderStateNeverOpened, } from "./native-session-executor.js"; @@ -2920,6 +2927,8 @@ describe("remote provider pack manifest", () => { Object.assign(payload, { providers: { cursor } }); await writeManifest(); expect(readRemoteProviderPackManifest(root).payload.providers?.cursor?.version).toBe(QUALIFIED_ACPX_PROFILES.cursor.agentServerVersion); + // The run path streams the same checks off the event loop. + await expect(verifyRemoteProviderPackManifest(root)).resolves.toEqual(readRemoteProviderPackManifest(root)); const releaseMetadata = await mkdtemp(join(tmpdir(), "paperclip-image-identity-")); const manifestPath = join(releaseMetadata, "provider-pack.json"); await cp(join(root, "provider-pack.json"), manifestPath); @@ -2944,6 +2953,7 @@ describe("remote provider pack manifest", () => { await rm(releaseMetadata, { recursive: true, force: true }); await writeFile(join(root, cursorPath, "runtime"), "substitute Cursor runtime"); expect(() => readRemoteProviderPackManifest(root)).toThrow("asset tree digest mismatch"); + await expect(verifyRemoteProviderPackManifest(root)).rejects.toThrow("asset tree digest mismatch"); await writeFile(join(root, cursorPath, "runtime"), "pinned Cursor runtime"); const kimiPath = "provider-assets/kimi/linux-x64"; await mkdir(join(root, kimiPath), { recursive: true }); @@ -3507,7 +3517,22 @@ describe("verified native harness backups", () => { sourceProviderLeaseId: "sandbox-1", completedAt: "2026-08-26T00:00:00.000Z", }); + // The run path builds and verifies the same manifest by streaming. + await expect(buildNativeHarnessBackupManifestStreamed({ + backupRoot: current, + execution: backupExecution, + runnerInstanceId: "runner-1", + providerSessionIdentity: { + providerSessionId: "thread-1", + providerBackendSessionId: "session-1", + providerSessionIdentity: null, + }, + sourceProviderLeaseId: "sandbox-1", + completedAt: "2026-08-26T00:00:00.000Z", + })).resolves.toEqual(manifest); await writeFile(join(current, "manifest.json"), JSON.stringify(manifest)); + await expect(verifyNativeHarnessBackupStreamed({ root, execution: backupExecution, runnerInstanceId: "runner-1" })) + .resolves.toEqual(verifyNativeHarnessBackup({ root, execution: backupExecution, runnerInstanceId: "runner-1" })); expect( verifyNativeHarnessBackup({ @@ -3553,6 +3578,8 @@ describe("verified native harness backups", () => { runnerInstanceId: "runner-1", }), ).toBeNull(); + await expect(verifyNativeHarnessBackupStreamed({ root, execution: backupExecution, runnerInstanceId: "runner-1" })) + .resolves.toBeNull(); } finally { await rm(root, { recursive: true, force: true }); } diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index e138de3c335..3f50677c75a 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -81,7 +81,7 @@ import { dotRunnerBroker } from "../dot-runner-broker.js"; import { CURSOR_DISTRIBUTION_PINS, KIMI_DISTRIBUTION_CLOSURE_SHA256, QUALIFIED_ACPX_PROFILES, QUALIFIED_ACPX_VERSION } from "../../vendor/paperclip-runner/index.js"; import { isProviderMode } from "../../vendor/paperclip-runner/index.js"; import { codexUsageBaselineForNextRun } from "../../vendor/paperclip-runner/index.js"; -import { PROVIDER_PACK_COMPATIBILITY, bundledRemoteProviderPackManifestPath, bundledRemoteRunnerBinary, runnerBinaryTarget } from "../../vendor/paperclip-runner/index.js"; +import { PROVIDER_PACK_COMPATIBILITY, bundledRemoteProviderPackManifestPath, runnerBinaryTarget } from "../../vendor/paperclip-runner/index.js"; import { nativeRetryCancellationEligible, rethrowNativeCancellationLockConflict, assertCancellationRequest, cancellationIntentId as callerCancellationIntentId, cancellationRequestId } from "./native-cancellation-request.js"; import { readNativePlanWait } from "./native-plan-wait.js"; import { nativeProviderLifecycle } from "./provider-lifecycle.js"; @@ -130,10 +130,11 @@ import { EXECUTION_CONTROL_DEADLINE_MS, } from "../execution-control-deadline.js"; import { createHash, createHmac, randomUUID } from "node:crypto"; -import { execFile, execFileSync } from "node:child_process"; +import { execFile } from "node:child_process"; import { chmodSync, copyFileSync, + createReadStream, closeSync, constants, existsSync, @@ -152,6 +153,7 @@ import { rmSync, writeFileSync, } from "node:fs"; +import { lstat, readFile, readdir, readlink } from "node:fs/promises"; import { hostname, tmpdir, userInfo } from "node:os"; import { DatabaseSync } from "node:sqlite"; import { NATIVE_CHECKPOINT_HEADER_INSPECTION_MAX_BYTES, loadNativeRunnerCheckpoint, loadNativeRunnerCheckpointHeader, loadNativeUnclaimedBootstrapCheckpoint, nativeRunnerCheckpointPersistence, nativeRunnerCheckpointRegistration, restoreNativeRunnerCheckpoint } from "./native-runner-checkpoint.js"; @@ -296,7 +298,7 @@ import { import { conflict, HttpError } from "../../errors.js"; import { redactSensitiveText } from "../../redaction.js"; import { classifyRemoteConnectionError } from "../remote-connection-failure.js"; -import { resolvePaperclipRunnerBinary, resolveRemotePaperclipRunnerBinary } from "./native-codex-runner.js"; +import { resolvePaperclipRunnerBinary, resolvePaperclipRunnerBinaryAsync, resolveRemotePaperclipRunnerBinary, verifiedBundledRemoteRunnerBinary } from "./native-codex-runner.js"; import { createNativeRunTrace, isNativeRunRootHistoricalSpan, @@ -316,6 +318,13 @@ import { registerLiveRunnerGoalController } from "../runner-goal-control-broker. import { applyRunnerGoalPrpEvent } from "../runner-goals.js"; import { currentControllerProcessStamp, currentProcessSpaceHash } from "../controller-process-identity.js"; import { proveOwnerGone } from "../owner-proof.js"; +import { + binaryTargetFromHeader, + cachedFileSha256, + execFileBuffered, + sha256DirectoryTreeStreamed, + sha256FileStreamed, +} from "./off-loop-file-work.js"; import { NativeRunnerOwnershipUnverifiedError, isNativeRunnerOwnershipHeld, @@ -7775,6 +7784,60 @@ function digestBackupDirectory(directory: string): { return { sha256: `sha256:${hash.digest("hex")}`, bytes }; } +// `digestBackupDirectory` for run paths: the same digest, streamed. A harness +// backup can hold large provider databases, and hashing it synchronously +// blocked every lease renewal for the whole read. +async function digestBackupDirectoryStreamed(directory: string): Promise<{ + sha256: string; + bytes: number; +}> { + let hash = createHash("sha256"); + let bytes = 0; + const visit = async (current: string, relative: string): Promise => { + const entries = (await readdir(current, { withFileTypes: true })).sort( + (left, right) => left.name.localeCompare(right.name), + ); + if (entries.length === 0) hash.update(`directory:${relative}\0`); + for (const entry of entries) { + const entryPath = resolve(current, entry.name); + const entryRelative = relative ? `${relative}/${entry.name}` : entry.name; + const stats = await lstat(entryPath); + if (entry.isDirectory()) { + hash.update(`directory:${entryRelative}:${stats.mode & 0o777}\0`); + await visit(entryPath, entryRelative); + } else if (entry.isSymbolicLink()) { + hash.update(`symlink:${entryRelative}:${await readlink(entryPath)}\0`); + } else if (entry.isFile()) { + // The record names the byte count before the contents. Stream with the + // size from lstat; if the file changed size meanwhile, hash the bytes + // actually read, exactly as the synchronous digest would. + const before = hash.copy(); + const header = (size: number) => `file:${entryRelative}:${stats.mode & 0o777}:${size}\0`; + hash.update(header(stats.size)); + let fileBytes = 0; + for await (const chunk of createReadStream(entryPath)) { + fileBytes += (chunk as Buffer).byteLength; + hash.update(chunk as Buffer); + } + if (fileBytes !== stats.size) { + const contents = await readFile(entryPath); + hash = before; + hash.update(header(contents.byteLength)); + hash.update(contents); + fileBytes = contents.byteLength; + } + bytes += fileBytes; + } else { + throw new Error( + `runner_harness_backup_unsupported_entry:${entryRelative}`, + ); + } + } + }; + await visit(directory, ""); + return { sha256: `sha256:${hash.digest("hex")}`, bytes }; +} + function harnessBackupRoot(root: string): string { return resolve(root, "failover-backups"); } @@ -7866,14 +7929,16 @@ function compatibleNativeHarnessBackupManifests(input: { return compatible; } -export function buildNativeHarnessBackupManifest(input: { +type NativeHarnessBackupManifestInput = { backupRoot: string; execution: NativeExecutionInput; runnerInstanceId: string; providerSessionIdentity: unknown; sourceProviderLeaseId: string; completedAt?: string; -}): NativeHarnessBackupManifest { +}; + +export function buildNativeHarnessBackupManifest(input: NativeHarnessBackupManifestInput): NativeHarnessBackupManifest { if (!providerSessionIdentityIsPresent(input.providerSessionIdentity)) { throw new Error("runner_harness_state_mismatch: backup_provider_identity_missing"); } @@ -7883,6 +7948,31 @@ export function buildNativeHarnessBackupManifest(input: { if (!existsSync(path)) throw new Error("runner_harness_state_mismatch: backup_directory_missing"); return { name: directory.name, ...digestBackupDirectory(path) }; }); + return nativeHarnessBackupManifest(input, profile, directories); +} + +/** `buildNativeHarnessBackupManifest` for run paths: the backup is hashed by streaming. */ +export async function buildNativeHarnessBackupManifestStreamed( + input: NativeHarnessBackupManifestInput, +): Promise { + if (!providerSessionIdentityIsPresent(input.providerSessionIdentity)) { + throw new Error("runner_harness_state_mismatch: backup_provider_identity_missing"); + } + const profile = resolveNativeHarnessPersistenceProfile(input.execution); + const directories: NativeHarnessBackupManifest["directories"] = []; + for (const directory of profile.directories) { + const path = resolve(input.backupRoot, directory.name); + if (!existsSync(path)) throw new Error("runner_harness_state_mismatch: backup_directory_missing"); + directories.push({ name: directory.name, ...await digestBackupDirectoryStreamed(path) }); + } + return nativeHarnessBackupManifest(input, profile, directories); +} + +function nativeHarnessBackupManifest( + input: NativeHarnessBackupManifestInput, + profile: ReturnType, + directories: NativeHarnessBackupManifest["directories"], +): NativeHarnessBackupManifest { return { schema: "paperclip.native-harness-backup.v1", normalizedSessionId: nativeSessionKey(input.execution), @@ -7900,6 +7990,44 @@ export function buildNativeHarnessBackupManifest(input: { }; } +/** `verifyNativeHarnessBackup` for run paths: the same checks, streamed. */ +export async function verifyNativeHarnessBackupStreamed(input: { + root: string; + execution: NativeExecutionInput; + runnerInstanceId: string; +}): Promise { + for (const { + root: candidateRoot, + manifest, + } of compatibleNativeHarnessBackupManifests(input)) { + let bytes = 0; + let valid = true; + for (const declared of manifest.directories) { + const directoryPath = resolve(candidateRoot, declared.name); + if (!existsSync(directoryPath)) { + valid = false; + break; + } + try { + const digest = await digestBackupDirectoryStreamed(directoryPath); + if ( + digest.sha256 !== declared.sha256 || + digest.bytes !== declared.bytes + ) { + valid = false; + break; + } + bytes += digest.bytes; + } catch { + valid = false; + break; + } + } + if (valid) return { root: candidateRoot, manifest, bytes }; + } + return null; +} + export function verifyNativeHarnessBackup(input: { root: string; execution: NativeExecutionInput; @@ -13120,6 +13248,10 @@ function sha256File(path: string): string { return `sha256:${createHash("sha256").update(readFileSync(path)).digest("hex")}`; } +const unsupportedProviderDistEntry = (relativePath: string) => new Error( + `runner_remote_provider_artifact_incompatible: unsupported dist entry ${relativePath}`, +); + export function sha256DirectoryTree(root: string): string { const hash = createHash("sha256"); const visit = (directory: string, prefix = "") => { @@ -13139,9 +13271,7 @@ export function sha256DirectoryTree(root: string): string { `symlink\0${relativePath}\0${readlinkSync(absolutePath)}\n`, ); } else { - throw new Error( - `runner_remote_provider_artifact_incompatible: unsupported dist entry ${relativePath}`, - ); + throw unsupportedProviderDistEntry(relativePath); } } }; @@ -13169,6 +13299,23 @@ export function readRemoteProviderPackManifest(packRoot: string): RemoteProvider return readRemoteProviderPackIdentity(packRoot, true); } +/** + * `readRemoteProviderPackManifest` for a run path: the same checks in the same + * order, but the controller files are hashed by streaming, off the event loop. + */ +export async function verifyRemoteProviderPackManifest(packRoot: string): Promise { + const checks = providerPackIdentityChecks(packRoot, true); + let step = checks.next(); + while (!step.done) { + const request = step.value; + const digest = request.kind === "file" + ? await sha256FileStreamed(request.path) + : await sha256DirectoryTreeStreamed(request.path, unsupportedProviderDistEntry); + step = checks.next(digest); + } + return step.value; +} + export function readBundledRemoteProviderPackManifest(manifestPath = bundledRemoteProviderPackManifestPath(), target: "linux-x64" | "linux-arm64" = "linux-x64"): RemoteProviderPackManifest { const manifest = readRemoteProviderPackIdentity(dirname(manifestPath), false); if (`${manifest.payload.target.platform}-${manifest.payload.target.architecture}` !== target) { @@ -13177,7 +13324,25 @@ export function readBundledRemoteProviderPackManifest(manifestPath = bundledRemo return manifest; } +type ProviderPackHashRequest = { kind: "file" | "tree"; path: string }; + function readRemoteProviderPackIdentity(packRoot: string, verifyControllerFiles: boolean): RemoteProviderPackManifest { + const checks = providerPackIdentityChecks(packRoot, verifyControllerFiles); + let step = checks.next(); + while (!step.done) { + const request = step.value; + step = checks.next(request.kind === "file" ? sha256File(request.path) : sha256DirectoryTree(request.path)); + } + return step.value; +} + +// Validates a provider pack manifest. Each controller-file digest it needs is +// yielded to the caller, so the sync and streamed readers share one order of +// checks and one set of errors. +function* providerPackIdentityChecks( + packRoot: string, + verifyControllerFiles: boolean, +): Generator { let manifest: RemoteProviderPackManifest; try { manifest = JSON.parse( @@ -13267,14 +13432,14 @@ function readRemoteProviderPackIdentity(packRoot: string, verifyControllerFiles: } if ( !/^sha256:[a-f0-9]{64}$/.test(artifact?.sha256 ?? "") || - (verifyControllerFiles && sha256File(resolve(packRoot, artifactPath)) !== artifact.sha256) + (verifyControllerFiles && (yield { kind: "file", path: resolve(packRoot, artifactPath) }) !== artifact.sha256) ) { throw new Error( `runner_remote_provider_artifact_incompatible: ${label} digest mismatch`, ); } } - if (verifyControllerFiles && sha256DirectoryTree(resolve(packRoot, "dist")) !== payload.distDigest) { + if (verifyControllerFiles && (yield { kind: "tree", path: resolve(packRoot, "dist") }) !== payload.distDigest) { throw new Error( "runner_remote_provider_artifact_incompatible: provider dist tree digest mismatch", ); @@ -13332,7 +13497,7 @@ function readRemoteProviderPackIdentity(packRoot: string, verifyControllerFiles: throw new Error("runner_remote_provider_artifact_incompatible: Kimi profile or closure does not match this release"); } } - if (verifyControllerFiles && sha256DirectoryTree(resolve(packRoot, candidatePath)) !== candidate.sha256) { + if (verifyControllerFiles && (yield { kind: "tree", path: resolve(packRoot, candidatePath) }) !== candidate.sha256) { throw new Error("runner_remote_provider_artifact_incompatible: candidate asset tree digest mismatch"); } } @@ -13418,7 +13583,7 @@ export async function stageRemoteRunnerFile(input: { ]); return; } - const bytes = readFileSync(input.sourcePath); + const bytes = await readFile(input.sourcePath); const directory = posix.dirname(input.targetPath); const escapedDirectory = directory.replaceAll("'", "'\\''"); const escapedTarget = input.targetPath.replaceAll("'", "'\\''"); @@ -13494,16 +13659,15 @@ export async function stageRemoteRunnerDirectory(input: { try { if (excludeArgs.length > 0) { stagingRoot = mkdtempSync(join(tmpdir(), "paperclip-runner-restore-")); - const archive = execFileSync( + const archive = await execFileBuffered( "tar", [...excludeArgs, "--no-xattrs", "-czf", "-", "-C", input.sourcePath, "."], { - encoding: "buffer", maxBuffer: 64 * 1024 * 1024, env: { ...process.env, COPYFILE_DISABLE: "1" }, }, ); - execFileSync("tar", ["-xzf", "-", "-C", stagingRoot], { + await execFileBuffered("tar", ["-xzf", "-", "-C", stagingRoot], { input: archive, maxBuffer: 64 * 1024 * 1024, }); @@ -13527,11 +13691,10 @@ export async function stageRemoteRunnerDirectory(input: { } return; } - const archive = execFileSync( + const archive = await execFileBuffered( "tar", [...excludeArgs, "--no-xattrs", "-czf", "-", "-C", input.sourcePath, "."], { - encoding: "buffer", maxBuffer: 64 * 1024 * 1024, env: { ...process.env, COPYFILE_DISABLE: "1" }, }, @@ -13668,7 +13831,7 @@ export async function prepareRemoteProviderPackOnTarget(input: { targetPath: stagedRemoteProviderPackRoot, reuseRoots: preinstalledProviderPackRoots, trace: input.trace, - verifyDirectory: (sourceRoot) => { readRemoteProviderPackManifest(sourceRoot); }, + verifyDirectory: async (sourceRoot) => { await verifyRemoteProviderPackManifest(sourceRoot); }, stageDirectory: (sourceRoot) => stageRemoteRunnerDirectory({ target: remoteTarget, runner: remoteCommandRunner, @@ -13717,27 +13880,27 @@ async function remoteRunnerPathExists(input: { return result.exitCode === 0 && !result.timedOut; } -function assertSafeRemoteCheckpointArchive(archive: Buffer): void { +async function assertSafeRemoteCheckpointArchive(archive: Buffer): Promise { if (archive.length > MAX_REMOTE_CHECKPOINT_ARCHIVE_BYTES) { throw new Error("runner_remote_checkpoint_archive_too_large"); } let names: string[]; let verboseEntries: string[]; try { - names = execFileSync("tar", ["-tzf", "-"], { + names = (await execFileBuffered("tar", ["-tzf", "-"], { input: archive, - encoding: "utf8", maxBuffer: 8 * 1024 * 1024, timeout: 30_000, - }) + })) + .toString("utf8") .split("\n") .filter((line) => line.length > 0); - verboseEntries = execFileSync("tar", ["-tvzf", "-"], { + verboseEntries = (await execFileBuffered("tar", ["-tvzf", "-"], { input: archive, - encoding: "utf8", maxBuffer: 8 * 1024 * 1024, timeout: 30_000, - }) + })) + .toString("utf8") .split("\n") .filter((line) => line.length > 0); } catch { @@ -13767,9 +13930,8 @@ function assertSafeRemoteCheckpointArchive(archive: Buffer): void { } } try { - execFileSync("tar", ["-xOzf", "-"], { + await execFileBuffered("tar", ["-xOzf", "-"], { input: archive, - stdio: ["pipe", "pipe", "pipe"], maxBuffer: MAX_REMOTE_CHECKPOINT_EXPANDED_BYTES, timeout: 60_000, }); @@ -13972,7 +14134,7 @@ export async function syncRemoteRunnerDirectoryOut(input: { throw new Error("runner_remote_checkpoint_failed"); } const archive = Buffer.from(result.stdout.replace(/\s+/g, ""), "base64"); - assertSafeRemoteCheckpointArchive(archive); + await assertSafeRemoteCheckpointArchive(archive); const parent = resolve(input.targetPath, ".."); const stagingRoot = mkdtempSync(join(parent, ".paperclip-checkpoint-")); const stagedTarget = join(stagingRoot, "payload"); @@ -13984,7 +14146,10 @@ export async function syncRemoteRunnerDirectoryOut(input: { let replacementInstalled = false; try { mkdirSync(stagedTarget, { recursive: true, mode: input.mode }); - execFileSync( + // Extraction runs off the event loop into this call's private staging + // directory. The swap below stays synchronous, so no other run can observe + // or interleave with a half-installed checkpoint. + await execFileBuffered( "tar", [ "--no-same-owner", @@ -15458,7 +15623,7 @@ async function createRunnerdBackendWithinSessionClaim( const providerArtifactsSignal = () => input.executionContext?.signal ? AbortSignal.any([input.executionContext.signal, AbortSignal.timeout(300_000)]) : AbortSignal.timeout(300_000); const liveProviderRecovery = input.restartRecovery?.kind === "reattach_remote_runner" || input.remoteWarmRecovery !== undefined; - const artifactBackup = artifactStorage ? verifyNativeHarnessBackup({ root, execution: input.execution, + const artifactBackup = artifactStorage ? await verifyNativeHarnessBackupStreamed({ root, execution: input.execution, runnerInstanceId: input.runnerInstanceId }) : null; let providerArtifacts = artifactStorage ? artifactBackup?.manifest.providerArtifacts ?? input.providerPackAdmissionSource?.artifacts as NativeProviderArtifactReceipt | undefined : undefined; @@ -15476,7 +15641,7 @@ async function createRunnerdBackendWithinSessionClaim( const remoteClaudeRunnerBinary = remoteClaude && !adoptingPersistedRemoteRunner ? await resolveRemotePaperclipRunnerBinary({ runner: remoteCommandRunner!, remoteCwd: remoteTarget!.remoteCwd, configuredPath: input.runnerRemoteBinaryPath }) : null; - const remoteClaudeTarget = remoteClaudeRunnerBinary ? runnerBinaryTarget(readFileSync(remoteClaudeRunnerBinary)) : null; + const remoteClaudeTarget = remoteClaudeRunnerBinary ? await binaryTargetFromHeader(remoteClaudeRunnerBinary, runnerBinaryTarget) : null; const useBundledClaudePack = remoteClaude && !adoptingPersistedRemoteRunner && !input.runnerRemoteProviderPackPath?.trim(); if (useBundledClaudePack && remoteClaudeTarget !== "linux-x64" && remoteClaudeTarget !== "linux-arm64") { throw new Error("runner_remote_provider_artifact_incompatible: Claude remote provider packs require a qualified Linux target"); @@ -15537,12 +15702,12 @@ async function createRunnerdBackendWithinSessionClaim( // The persisted admission owns a surviving runner's identity. Local release // files are launch inputs and may no longer exist on a replacement controller. const controllerRunnerBinary = adoptingPersistedRemoteRunner - ? input.runnerRemoteBinaryPath?.trim() || resolvePaperclipRunnerBinary() + ? input.runnerRemoteBinaryPath?.trim() || await resolvePaperclipRunnerBinaryAsync() : targetMatchedCodexArtifact ? await timeController("runner_binary", () => resolveRemotePaperclipRunnerBinary({ runner: remoteCommandRunner!, remoteCwd: remoteTarget.remoteCwd, configuredPath: input.runnerRemoteBinaryPath })) : remoteTarget - ? remoteClaudeRunnerBinary || input.runnerRemoteBinaryPath?.trim() || remotePiCompanion?.runnerBinary || (useBundledRemoteImageAssets ? bundledRemoteRunnerBinary() : resolvePaperclipRunnerBinary()) - : resolvePaperclipRunnerBinary(); + ? remoteClaudeRunnerBinary || input.runnerRemoteBinaryPath?.trim() || remotePiCompanion?.runnerBinary || (useBundledRemoteImageAssets ? await verifiedBundledRemoteRunnerBinary() : await resolvePaperclipRunnerBinaryAsync()) + : await resolvePaperclipRunnerBinaryAsync(); const selectProviderPackAuthority = (reattach: boolean) => requiresRemoteProviderPack ? resolveRemoteProviderPackAuthority({ stateDirectory: root, binding: { companyId: input.execution.binding.companyId, agentId: input.execution.binding.agentId, @@ -16053,7 +16218,7 @@ async function createRunnerdBackendWithinSessionClaim( // hashed once. let retiredSourceBackup: VerifiedHarnessBackup | null = null; const retiredSourceBackupAuthorized = async (): Promise => { - const backup = verifyNativeHarnessBackup({ root, execution: input.execution, runnerInstanceId: input.runnerInstanceId }); + const backup = await verifyNativeHarnessBackupStreamed({ root, execution: input.execution, runnerInstanceId: input.runnerInstanceId }); const sessionId = nativeSessionKey(input.execution); if (!backup || !sessionId) return false; retiredSourceBackup = backup; @@ -16069,7 +16234,7 @@ async function createRunnerdBackendWithinSessionClaim( if (!remoteTarget || !remoteCommandRunner) { throw new Error("runner_harness_backup_unavailable"); } - const backup = retiredSourceBackup ?? verifyNativeHarnessBackup({ + const backup = retiredSourceBackup ?? await verifyNativeHarnessBackupStreamed({ root, execution: input.execution, runnerInstanceId: input.runnerInstanceId, @@ -16528,7 +16693,7 @@ async function createRunnerdBackendWithinSessionClaim( `[paperclip-runner] remote checkpoint read interrupted (attempt ${attempt} of ${attempts}); reading the settled state again\n`, ), ); - const manifest = buildNativeHarnessBackupManifest({ + const manifest = await buildNativeHarnessBackupManifestStreamed({ backupRoot: pendingRoot, execution: input.execution, runnerInstanceId: input.runnerInstanceId, @@ -17548,9 +17713,9 @@ export function createRemoteNativeArtifactPreparation(input: { const prepare = async (requiredMode: "dial_wss" | "listen_ws") => { const matchesControllerRunnerArtifact = async (executable: string): Promise => { - const expected = createHash("sha256") - .update(readFileSync(controllerRunnerBinary)) - .digest("hex"); + // Streamed once per boot and reused while the binary is unchanged; a + // synchronous 31 MB read here blocked the event loop on every run. + const expected = (await cachedFileSha256(controllerRunnerBinary)).slice("sha256:".length); try { const probe = await remoteCommandRunner.execute({ command: "sh", diff --git a/server/src/services/native-runtime/off-loop-file-work.test.ts b/server/src/services/native-runtime/off-loop-file-work.test.ts new file mode 100644 index 00000000000..246604dc8bc --- /dev/null +++ b/server/src/services/native-runtime/off-loop-file-work.test.ts @@ -0,0 +1,185 @@ +import { execFileSync, spawn } from "node:child_process"; +import { createHash } from "node:crypto"; +import { chmodSync, mkdirSync, mkdtempSync, readFileSync, rmSync, utimesSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { resolveRemotePaperclipRunnerBinary } from "./native-codex-runner.js"; +import { stageRemoteRunnerDirectory, syncRemoteRunnerDirectoryOut } from "./native-session-executor.js"; +import { + binaryTargetFromHeader, + cachedFileSha256, + clearCachedFileSha256, + execFileBuffered, +} from "./off-loop-file-work.js"; + +// Measures the longest gap between 10 ms timer ticks while `work` runs. A +// synchronous read or child process stops every tick for its whole duration. +async function longestTimerGapDuring(work: () => Promise): Promise { + let last = performance.now(); + let longest = 0; + const timer = setInterval(() => { + const now = performance.now(); + longest = Math.max(longest, now - last); + last = now; + }, 10); + try { + await work(); + const now = performance.now(); + longest = Math.max(longest, now - last); + } finally { + clearInterval(timer); + } + return longest; +} + +// An ELF64 little-endian x86-64 header, padded like a small executable. +const linuxX64Header = () => { + const bytes = Buffer.alloc(4096); + Buffer.from([0x7f, 0x45, 0x4c, 0x46, 2, 1]).copy(bytes); + bytes.writeUInt16LE(62, 18); + return bytes; +}; + +const SLOW_MS = 2_500; +// Generous for a loaded CI runner, and well below one blocked slow step. +const RESPONSIVE_GAP_MS = 1_500; +const SLOW_TEST_TIMEOUT_MS = 30_000; + +// The fixtures use POSIX tools (mkfifo, sh, sleep, find); the controller +// paths under test are the Linux and macOS ones. +describe.skipIf(process.platform === "win32")("run-path file work stays off the event loop", () => { + let root: string; + let originalPath: string | undefined; + + beforeEach(() => { + root = mkdtempSync(join(tmpdir(), "paperclip-off-loop-")); + originalPath = process.env.PATH; + clearCachedFileSha256(); + }); + afterEach(() => { + process.env.PATH = originalPath; + rmSync(root, { recursive: true, force: true }); + }); + + // Every `tar` the server starts now takes SLOW_MS longer, like a slow disk. + const slowTar = () => { + const realTar = execFileSync("sh", ["-c", "command -v tar"], { encoding: "utf8" }).trim(); + const bin = join(root, "slow-bin"); + mkdirSync(bin); + writeFileSync(join(bin, "tar"), `#!/bin/sh\nsleep ${SLOW_MS / 1000}\nexec '${realTar}' "$@"\n`); + chmodSync(join(bin, "tar"), 0o755); + process.env.PATH = `${bin}:${originalPath}`; + }; + + it("checks the remote daemon's target without blocking while its file is slow to read", async () => { + // A FIFO whose writer waits stands in for a binary on a stalled disk. + const binary = join(root, "paperclip-runnerd"); + execFileSync("mkfifo", [binary]); + const header = join(root, "header"); + writeFileSync(header, linuxX64Header()); + const writer = spawn("sh", ["-c", `sleep ${SLOW_MS / 1000}; cat '${header}' > '${binary}'`], { stdio: "ignore" }); + const runner = { execute: async () => ({ stdout: "Linux\nx86_64\n", stderr: "", exitCode: 0, timedOut: false }) }; + + let resolved = ""; + const gap = await longestTimerGapDuring(async () => { + resolved = await resolveRemotePaperclipRunnerBinary({ runner: runner as never, remoteCwd: "/workspace", configuredPath: binary }); + }); + writer.kill(); + + expect(resolved).toBe(binary); + expect(gap).toBeLessThan(RESPONSIVE_GAP_MS); + }, SLOW_TEST_TIMEOUT_MS); + + it("archives a staged directory without blocking while tar is slow", async () => { + slowTar(); + const source = join(root, "source"); + mkdirSync(join(source, "keep"), { recursive: true }); + mkdirSync(join(source, "skip"), { recursive: true }); + writeFileSync(join(source, "keep", "file.txt"), "kept"); + writeFileSync(join(source, "skip", "file.txt"), "skipped"); + let staged: string[] = []; + const runner = { + execute: async () => ({ stdout: "", stderr: "", exitCode: 0, timedOut: false }), + syncIn: async (operations: Array<{ files: Array<{ sourcePath: string }> }>) => { + const sourcePath = operations[0]!.files[0]!.sourcePath; + staged = execFileSync("find", [".", "-type", "f"], { cwd: sourcePath, encoding: "utf8" }).trim().split("\n"); + }, + }; + + const gap = await longestTimerGapDuring(() => stageRemoteRunnerDirectory({ + target: { kind: "remote" } as never, runner: runner as never, + sourcePath: source, targetPath: "/remote/target", mode: 0o700, excludeEntries: ["skip"], + })); + + expect(staged).toEqual(["./keep/file.txt"]); + expect(gap).toBeLessThan(RESPONSIVE_GAP_MS); + }, SLOW_TEST_TIMEOUT_MS); + + it("checks and installs a copied-back checkpoint without blocking while tar is slow", async () => { + const payload = join(root, "payload"); + mkdirSync(join(payload, "state"), { recursive: true }); + writeFileSync(join(payload, "state", "session.json"), "{\"ok\":true}"); + const archive = execFileSync("tar", ["-czf", "-", "-C", payload, "."]); + slowTar(); + const runner = { + execute: async ({ args }: { args: string[] }) => args[1]?.includes("base64") + ? { stdout: archive.toString("base64"), stderr: "", exitCode: 0, timedOut: false } + : { stdout: "", stderr: "", exitCode: 0, timedOut: false }, + }; + const target = join(root, "checkpoint"); + + const gap = await longestTimerGapDuring(() => syncRemoteRunnerDirectoryOut({ + runner: runner as never, sourcePath: "/remote/state", targetPath: target, mode: 0o700, + })); + + expect(readFileSync(join(target, "state", "session.json"), "utf8")).toBe("{\"ok\":true}"); + expect(gap).toBeLessThan(RESPONSIVE_GAP_MS); + }, SLOW_TEST_TIMEOUT_MS); +}); + +describe.skipIf(process.platform === "win32")("off-loop file helpers", () => { + let root: string; + beforeEach(() => { + root = mkdtempSync(join(tmpdir(), "paperclip-off-loop-helpers-")); + clearCachedFileSha256(); + }); + afterEach(() => rmSync(root, { recursive: true, force: true })); + + it("reuses a binary digest until the file changes", async () => { + const binary = join(root, "runnerd"); + writeFileSync(binary, "first"); + const first = await cachedFileSha256(binary); + expect(first).toBe(`sha256:${createHash("sha256").update("first").digest("hex")}`); + expect(await cachedFileSha256(binary)).toBe(first); + + writeFileSync(binary, "second"); + utimesSync(binary, new Date(Date.now() + 5_000), new Date(Date.now() + 5_000)); + expect(await cachedFileSha256(binary)).toBe(`sha256:${createHash("sha256").update("second").digest("hex")}`); + }); + + it("classifies a binary from its header, or from the whole file when the header cannot answer", async () => { + const elf = join(root, "elf"); + writeFileSync(elf, linuxX64Header()); + const lengths: number[] = []; + const classify = (bytes: Buffer) => { + lengths.push(bytes.length); + return bytes.length >= 70_000 ? "whole-file" : bytes[0] === 0x7f ? "header" : null; + }; + expect(await binaryTargetFromHeader(elf, classify)).toBe("header"); + + const large = join(root, "large"); + writeFileSync(large, Buffer.alloc(70_000)); + lengths.length = 0; + expect(await binaryTargetFromHeader(large, classify)).toBe("whole-file"); + expect(lengths).toEqual([64 * 1024, 70_000]); + }); + + it("rejects like execFileSync on a non-zero exit, a timeout, and output past maxBuffer", async () => { + expect((await execFileBuffered("sh", ["-c", "cat"], { input: Buffer.from("echo"), maxBuffer: 1024 })).toString()).toBe("echo"); + await expect(execFileBuffered("sh", ["-c", "exit 3"], { maxBuffer: 1024 })).rejects.toMatchObject({ code: 3 }); + await expect(execFileBuffered("sh", ["-c", "sleep 5"], { maxBuffer: 1024, timeout: 100 })).rejects.toMatchObject({ killed: true }); + await expect(execFileBuffered("sh", ["-c", "head -c 4096 /dev/zero"], { maxBuffer: 1024 })) + .rejects.toMatchObject({ code: "ERR_CHILD_PROCESS_STDIO_MAXBUFFER" }); + }); +}); diff --git a/server/src/services/native-runtime/off-loop-file-work.ts b/server/src/services/native-runtime/off-loop-file-work.ts new file mode 100644 index 00000000000..1e4305b5125 --- /dev/null +++ b/server/src/services/native-runtime/off-loop-file-work.ts @@ -0,0 +1,134 @@ +import { execFile } from "node:child_process"; +import { createHash } from "node:crypto"; +import { createReadStream } from "node:fs"; +import { open, readdir, readlink, stat } from "node:fs/promises"; +import { join } from "node:path"; + +// File hashing and archive commands on a run path must not block the event +// loop: on a slow disk one synchronous read or `execFileSync` freezes every +// lease renewal and every request in the process. These helpers keep the +// result and the error shape of the synchronous calls they replace. + +/** `sha256:` of a file, streamed. */ +export async function sha256FileStreamed(path: string, signal?: AbortSignal): Promise { + const hash = createHash("sha256"); + for await (const chunk of createReadStream(path, { signal })) hash.update(chunk as Buffer); + return `sha256:${hash.digest("hex")}`; +} + +type CachedDigest = { key: string; digest: Promise }; +const digestCache = new Map(); +const DIGEST_CACHE_LIMIT = 64; + +/** + * `sha256:` of an immutable artifact such as the runner binary. The digest + * is reused while the file's device, inode, size, modification and change times + * are unchanged, so every run of a boot reads it at most once. + */ +export async function cachedFileSha256(path: string): Promise { + const metadata = await stat(path, { bigint: true }); + const key = [metadata.dev, metadata.ino, metadata.size, metadata.mtimeNs, metadata.ctimeNs].join(":"); + const cached = digestCache.get(path); + if (cached?.key === key) return cached.digest; + const digest = sha256FileStreamed(path); + digestCache.set(path, { key, digest }); + if (digestCache.size > DIGEST_CACHE_LIMIT) digestCache.delete(digestCache.keys().next().value!); + digest.catch(() => { + if (digestCache.get(path)?.digest === digest) digestCache.delete(path); + }); + return digest; +} + +/** Test seam: forget every cached digest. */ +export function clearCachedFileSha256(): void { + digestCache.clear(); +} + +const HEADER_BYTES = 64 * 1024; + +/** + * The executable target of a native binary from its header. Reads the rest of + * the file only when the header alone cannot answer (a PE header past 64 KiB), + * so the result always equals `classify(readFileSync(path))`. + */ +export async function binaryTargetFromHeader( + path: string, + classify: (bytes: Buffer) => string | null, +): Promise { + const handle = await open(path, "r"); + try { + const header = Buffer.alloc(HEADER_BYTES); + let length = 0; + let ended = false; + while (length < HEADER_BYTES) { + const { bytesRead } = await handle.read(header, length, HEADER_BYTES - length, null); + if (bytesRead === 0) { ended = true; break; } + length += bytesRead; + } + const target = classify(header.subarray(0, length)); + if (target !== null || ended) return target; + return classify(Buffer.concat([header.subarray(0, length), await handle.readFile()])); + } finally { + await handle.close(); + } +} + +/** `sha256:` of a directory tree, in the format of `sha256DirectoryTree`. */ +export async function sha256DirectoryTreeStreamed( + root: string, + unsupported: (relativePath: string) => Error, +): Promise { + const hash = createHash("sha256"); + const visit = async (directory: string, prefix = ""): Promise => { + const entries = (await readdir(directory, { withFileTypes: true })).sort( + (left, right) => left.name.localeCompare(right.name), + ); + for (const entry of entries) { + const relativePath = prefix ? `${prefix}/${entry.name}` : entry.name; + const absolutePath = join(directory, entry.name); + if (entry.isDirectory()) { + hash.update(`directory\0${relativePath}\n`); + await visit(absolutePath, relativePath); + } else if (entry.isFile()) { + hash.update(`file\0${relativePath}\0${await sha256FileStreamed(absolutePath)}\n`); + } else if (entry.isSymbolicLink()) { + hash.update(`symlink\0${relativePath}\0${await readlink(absolutePath)}\n`); + } else { + throw unsupported(relativePath); + } + } + }; + await visit(root); + return `sha256:${hash.digest("hex")}`; +} + +export type FileCommandOptions = { + input?: Buffer; + maxBuffer: number; + timeout?: number; + env?: NodeJS.ProcessEnv; +}; + +/** + * Runs a command like `execFileSync(file, args, { encoding: "buffer", ... })`, + * without blocking the event loop: it resolves with stdout and rejects when the + * command exits non-zero, is killed by the timeout, or exceeds `maxBuffer`. + */ +export function execFileBuffered(file: string, args: readonly string[], options: FileCommandOptions): Promise { + return new Promise((resolveOutput, rejectOutput) => { + const child = execFile(file, [...args], { + encoding: "buffer", + maxBuffer: options.maxBuffer, + ...(options.timeout === undefined ? {} : { timeout: options.timeout }), + ...(options.env ? { env: options.env } : {}), + }, (error, stdout) => { + if (error) rejectOutput(error); + else resolveOutput(stdout); + }); + // A command can exit before it reads all of its input; its exit status, + // not the broken pipe, decides the result (as with execFileSync). + child.stdin?.on("error", () => undefined); + if (options.input) child.stdin?.end(options.input); + else child.stdin?.end(); + }); +} diff --git a/server/src/vendor/paperclip-runner/index.ts b/server/src/vendor/paperclip-runner/index.ts index d5d5d79456c..c528316cae7 100644 --- a/server/src/vendor/paperclip-runner/index.ts +++ b/server/src/vendor/paperclip-runner/index.ts @@ -149,6 +149,8 @@ export const createRunnerdCodexTransport: ( runner.createRunnerdCodexTransport; export const defaultCapabilityRunnerdBinary = runner.defaultCapabilityRunnerdBinary; +export const defaultCapabilityRunnerdBinaryAsync = + runner.defaultCapabilityRunnerdBinaryAsync; export const runnerBinaryTarget = runner.runnerBinaryTarget; export const coldAdmissionTimeoutMs = runner.coldAdmissionTimeoutMs; export const unreadSteeredMessageIds = runner.unreadSteeredMessageIds; @@ -217,6 +219,8 @@ export const AntigravityModelDiscoveryCleanupError = runner.AntigravityModelDisc export const PROVIDER_PACK_COMPATIBILITY = runner.PROVIDER_PACK_COMPATIBILITY; export const bundledRemoteProviderPackManifestPath = runner.bundledRemoteProviderPackManifestPath; export const bundledRemoteRunnerBinary = runner.bundledRemoteRunnerBinary; +export const bundledRemoteRunnerRelease = runner.bundledRemoteRunnerRelease; +export const BUNDLED_REMOTE_RUNNER_IDENTITY_MISMATCH = runner.BUNDLED_REMOTE_RUNNER_IDENTITY_MISMATCH; export const externalOperationDigest = runner.externalOperationDigest; export const digestPaperclipSemanticContent = runner.digestPaperclipSemanticContent;