diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index fb970884ff3..8f61f5674a9 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -42,6 +42,7 @@ import { nativeCompletionFeedback } from "./native-completion-feedback.js"; import { hasAcknowledgedNativeReassignmentStopIntent, hasAcknowledgedNativeStopIntent } from "../acknowledged-native-stop.js"; import { stoppedCodexTurnIsTextOnly } from "./stopped-codex-turn.js"; import { prepareVerifiedRemoteProviderPack } from "./remote-provider-pack.js"; +import { computerProviderPackCachePath, prepareComputerProviderPackCache } from "./remote-provider-pack-cache.js"; import { selectRemotePiCompanion } from "./remote-pi-companion.js"; import { readNativeLocalProcessStop, PROCESS_START_REQUESTED } from "../native-local-process-stop.js"; import { remoteLeaseCleanupScope } from "../remote-execution-termination.js"; @@ -11963,9 +11964,9 @@ async function createRunnerdBackendWithinSessionClaim( configuredProviderPackRoot, ); } - const stagedRemoteProviderPackRoot = remoteRuntimeRoot - ? posix.join(remoteRuntimeRoot, "provider-pack") - : null; + const stagedRemoteProviderPackRoot = remoteTarget?.transport === "computer" && expectedProviderPackManifest + ? computerProviderPackCachePath(remoteTarget.fileAuthority.agentHome, expectedProviderPackManifest.digest) + : remoteRuntimeRoot ? posix.join(remoteRuntimeRoot, "provider-pack") : null; let activeRemoteProviderPackRoot: string | null = null; const remoteBinary = remoteRuntimeRoot ? posix.join(remoteRuntimeRoot, "bin", "paperclip-runnerd") @@ -12510,7 +12511,31 @@ async function createRunnerdBackendWithinSessionClaim( expectedProviderPackManifest && stagedRemoteProviderPackRoot ) { - const packSource = await prepareVerifiedRemoteProviderPack({ + const packSource = remoteTarget.transport === "computer" + ? await prepareComputerProviderPackCache({ + cacheRoot: stagedRemoteProviderPackRoot, + sessionKey: nativeSessionKey(input.execution), + owner: remoteTarget.resourceAuthority, + runner: remoteCommandRunner, + verify: (root) => measureNativeRunnerSpan(input.trace, "provider_pack.verify", () => verifyRemoteProviderPack(root)), + stage: async (root, runner) => { + if (!configuredProviderPackRoot) { + throw new Error("runner_remote_provider_artifact_incompatible: configure the build-owned provider pack for first use on this computer"); + } + await input.onLog?.("stderr", "[paperclip-runner] Preparing the provider pack for first use.\n"); + await stageRemoteRunnerDirectory({ + target: remoteTarget, + runner, + sourcePath: configuredProviderPackRoot, + targetPath: root, + mode: 0o700, + onProgress: async (completed, total) => { + await input.onLog?.("stderr", `[paperclip-runner] Uploading provider pack: ${Math.round(completed / 1024 / 1024)} / ${Math.round(total / 1024 / 1024)} MiB.\n`); + }, + }); + }, + }) + : await prepareVerifiedRemoteProviderPack({ verifyStaged: () => measureNativeRunnerSpan( input.trace, "provider_pack.verify", @@ -12600,7 +12625,7 @@ async function createRunnerdBackendWithinSessionClaim( }, }); activeRemoteProviderPackRoot = stagedRemoteProviderPackRoot; - if (packSource === "staged") { + if (packSource === "staged" || packSource === "reused") { await input.onLog?.( "stderr", "[paperclip-runner] reusing manifest-matched provider pack from the workspace\n", diff --git a/server/src/services/native-runtime/remote-provider-pack-cache.test.ts b/server/src/services/native-runtime/remote-provider-pack-cache.test.ts new file mode 100644 index 00000000000..310ca24d148 --- /dev/null +++ b/server/src/services/native-runtime/remote-provider-pack-cache.test.ts @@ -0,0 +1,145 @@ +import { spawn } from "node:child_process"; +import { promises as fs } from "node:fs"; +import { tmpdir } from "node:os"; +import { dirname, join } from "node:path"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import type { CommandManagedRuntimeRunner } from "@paperclipai/adapter-utils/command-managed-runtime"; +import { computerProviderPackCachePath, prepareComputerProviderPackCache } from "./remote-provider-pack-cache.js"; + +const roots: string[] = []; +const digest = `sha256:${"a".repeat(64)}`; +const runner: CommandManagedRuntimeRunner = { + execute: (input) => new Promise((resolve, reject) => { + const child = spawn(input.command, input.args, { cwd: input.cwd, env: { ...process.env, ...input.env } }); + let stdout = ""; let stderr = ""; + child.stdout.on("data", (chunk) => { stdout += chunk; void input.onLog?.("stdout", String(chunk)); }); + child.stderr.on("data", (chunk) => { stderr += chunk; }); + child.on("error", reject); + child.on("close", (exitCode, signal) => resolve({ exitCode, signal, stdout, stderr, timedOut: false, pid: child.pid ?? null, startedAt: null })); + child.stdin.end(input.stdin); + }), +}; +async function fixture() { + const home = await fs.realpath(await fs.mkdtemp(join(tmpdir(), "provider-cache-"))); + roots.push(home); + return { + cacheRoot: computerProviderPackCachePath(home, digest), sessionKey: "task-a", + owner: { ownerId: "owner-a", generation: 1 }, runner, + verify: async (root: string) => { expect(await fs.readFile(join(root, "payload"), "utf8")).toBe("verified bytes"); }, + stage: async (root: string, guarded: CommandManagedRuntimeRunner) => { + const result = await guarded.execute({ command: "python3", args: ["-c", "import os,sys;os.mkdir(sys.argv[1],0o700);open(os.path.join(sys.argv[1],'payload'),'w').write(sys.stdin.read())", root], stdin: "verified bytes" }); + if (result.exitCode !== 0) throw new Error(result.stderr); + }, + }; +} +function deferred() { let resolve!: () => void; const promise = new Promise((done) => { resolve = done; }); return { promise, resolve }; } +afterEach(async () => { await Promise.all(roots.splice(0).map((root) => fs.rm(root, { recursive: true, force: true }))); }); + +describe("persistent computer provider pack cache", () => { + it("shares immutable identity across tasks, with separate agents and digests", () => { + expect(computerProviderPackCachePath("/agents/a", digest)).not.toBe(computerProviderPackCachePath("/agents/b", digest)); + expect(computerProviderPackCachePath("/agents/a", digest)).not.toBe(computerProviderPackCachePath("/agents/a", `sha256:${"b".repeat(64)}`)); + expect(() => computerProviderPackCachePath("/agents/../a", digest)).toThrow("identity_invalid"); + }); + + it("verifies a shared hit before the compression/upload callback and excludes its ownership marker", async () => { + const input = await fixture(); + const stage = vi.fn(input.stage); const verify = vi.fn(input.verify); + await expect(prepareComputerProviderPackCache({ ...input, stage, verify })).resolves.toBe("published"); + await expect(prepareComputerProviderPackCache({ ...input, stage, verify, sessionKey: "task-b" })).resolves.toBe("reused"); + expect(stage).toHaveBeenCalledTimes(1); + expect(verify).toHaveBeenCalledTimes(3); + expect(await fs.readdir(input.cacheRoot)).toEqual(["payload"]); + }); + + it("fails closed on a corrupt published pack, preserving its exact contents", async () => { + const input = await fixture(); + await fs.mkdir(input.cacheRoot, { recursive: true }); + await fs.writeFile(join(input.cacheRoot, "payload"), "corrupt"); + const stage = vi.fn(input.stage); + await expect(prepareComputerProviderPackCache({ ...input, stage })).rejects.toThrow("operator remove or quarantine this exact cache"); + expect(stage).not.toHaveBeenCalled(); + expect(await fs.readFile(join(input.cacheRoot, "payload"), "utf8")).toBe("corrupt"); + }); + + it("publishes without replacement when two tasks race and verifies the winner", async () => { + const input = await fixture(); const both = deferred(); let staged = 0; + const stage = async (root: string, guarded: CommandManagedRuntimeRunner) => { + await input.stage(root, guarded); if (++staged === 2) both.resolve(); await both.promise; + }; + await Promise.all(["task-a", "task-b"].map((sessionKey) => prepareComputerProviderPackCache({ ...input, sessionKey, stage }))); + expect(staged).toBe(2); + await input.verify(input.cacheRoot); + const directories = await fs.readdir(join(dirname(input.cacheRoot), ".uploads")); + for (const session of directories) expect(await fs.readdir(join(dirname(input.cacheRoot), ".uploads", session, "a".repeat(64)))).toEqual(["lock"]); + }); + + it("cleans only its marked private upload after failure", async () => { + const input = await fixture(); const unrelated = join(dirname(input.cacheRoot), ".uploads", "user-data"); + await fs.mkdir(unrelated, { recursive: true }); await fs.writeFile(join(unrelated, "keep"), "mine"); + let attempted = ""; + await expect(prepareComputerProviderPackCache({ ...input, stage: async (root, guarded) => { + attempted = dirname(root); await input.stage(root, guarded); throw new Error("upload disconnected"); + } })).rejects.toThrow("upload disconnected"); + await expect(fs.stat(attempted)).rejects.toMatchObject({ code: "ENOENT" }); + expect(await fs.readFile(join(unrelated, "keep"), "utf8")).toBe("mine"); + }); + + it("rejects delayed old writes and keeps the replacement when a session retries", async () => { + const input = await fixture(); const oldReady = deferred(); const releaseOld = deferred(); + let oldRoot = ""; let oldRunner!: CommandManagedRuntimeRunner; + const old = prepareComputerProviderPackCache({ ...input, stage: async (root, guarded) => { + oldRoot = root; oldRunner = guarded; await input.stage(root, guarded); oldReady.resolve(); await releaseOld.promise; + throw new Error("old attempt retired"); + } }); + const rejectedOld = expect(old).rejects.toThrow("old attempt retired"); + await oldReady.promise; + await prepareComputerProviderPackCache({ ...input, owner: { ...input.owner, generation: 2 } }); + const staleWrite = await oldRunner.execute({ command: "python3", args: ["-c", "import os,sys;os.makedirs(sys.argv[1],exist_ok=True)", oldRoot] }); + expect(staleWrite.exitCode).not.toBe(0); + expect(staleWrite.stderr).toContain("stale upload generation"); + releaseOld.resolve(); await rejectedOld; + await input.verify(input.cacheRoot); + await expect(fs.stat(dirname(oldRoot))).rejects.toMatchObject({ code: "ENOENT" }); + }); + + it("waits for an in-flight guarded writer before replacing its receipt", async () => { + const input = await fixture(); const entered = deferred(); const releaseStage = deferred(); + const releaseFile = join(roots.at(-1)!, "release-writer"); + let write!: Promise>>; + const old = prepareComputerProviderPackCache({ ...input, stage: async (root, guarded) => { + await input.stage(root, guarded); + write = guarded.execute({ command: "python3", args: ["-c", "import os,sys,time;print('entered',flush=True)\nwhile not os.path.exists(sys.argv[1]):time.sleep(0.01)", releaseFile], onLog: async (_stream, text) => { if (text.includes("entered")) entered.resolve(); } }); + await releaseStage.promise; await write; throw new Error("old retired"); + } }); + const rejectedOld = expect(old).rejects.toThrow("old retired"); + await entered.promise; + let replacementStaged = false; + const replacement = prepareComputerProviderPackCache({ ...input, owner: { ...input.owner, generation: 2 }, stage: async (root, guarded) => { + replacementStaged = true; await input.stage(root, guarded); + } }); + await new Promise((resolve) => setTimeout(resolve, 100)); + expect(replacementStaged).toBe(false); + await fs.writeFile(releaseFile, "release"); + expect((await write).exitCode).toBe(0); + await replacement; releaseStage.resolve(); await rejectedOld; + await input.verify(input.cacheRoot); + }); + + it("reclaims an interrupted marked upload on the next attempt", async () => { + const input = await fixture(); let attempted = ""; + const disconnected: CommandManagedRuntimeRunner = { execute: (command) => { + if (command.args?.[2] === "cleanup") return Promise.reject(new Error("controller disconnected")); + return runner.execute(command); + } }; + await expect(prepareComputerProviderPackCache({ ...input, runner: disconnected, stage: async (root, guarded) => { + attempted = dirname(root); await input.stage(root, guarded); throw new Error("interrupted"); + } })).rejects.toThrow("interrupted"); + expect((await fs.stat(attempted)).isDirectory()).toBe(true); + // Another task can publish while this session is disconnected. Retrying + // this session must reclaim its exact old upload even on the shared hit. + await prepareComputerProviderPackCache({ ...input, sessionKey: "another-task" }); + await expect(prepareComputerProviderPackCache({ ...input, owner: { ...input.owner, generation: 2 } })).resolves.toBe("reused"); + await expect(fs.stat(attempted)).rejects.toMatchObject({ code: "ENOENT" }); + }); +}); diff --git a/server/src/services/native-runtime/remote-provider-pack-cache.ts b/server/src/services/native-runtime/remote-provider-pack-cache.ts new file mode 100644 index 00000000000..97a8454d56f --- /dev/null +++ b/server/src/services/native-runtime/remote-provider-pack-cache.ts @@ -0,0 +1,149 @@ +import { createHash, randomUUID } from "node:crypto"; +import { posix } from "node:path"; +import type { CommandManagedRuntimeRunner } from "@paperclipai/adapter-utils/command-managed-runtime"; + +export function computerProviderPackCachePath(agentHome: string, digest: string): string { + if (!posix.isAbsolute(agentHome) || posix.normalize(agentHome) !== agentHome || !/^sha256:[a-f0-9]{64}$/.test(digest)) { + throw new Error("runner_remote_provider_cache_identity_invalid"); + } + return posix.join(agentHome, ".paperclip-runtime", "paperclip-runner", "provider-packs", digest.slice(7)); +} + +// Publication never replaces shared bytes. The per-session receipt fences every +// staging command; a retry waits for in-flight writes before replacing its receipt. +// A session never retried can retain its private partial upload for operator cleanup. +const cacheProgram = String.raw` +import os,sys,json,stat,fcntl,ctypes,errno,shutil,subprocess +op,root,session,identity_json=sys.argv[1:5];identity=json.loads(identity_json) +def fail(message):raise RuntimeError(message) +def directory(path): + fd=os.open('/',os.O_RDONLY|os.O_DIRECTORY) + try: + for part in path.split('/'): + if not part:continue + if part in ('.','..'):fail('invalid cache path') + try:os.mkdir(part,0o700,dir_fd=fd) + except FileExistsError:pass + nxt=os.open(part,os.O_RDONLY|os.O_DIRECTORY|os.O_NOFOLLOW,dir_fd=fd);os.close(fd);fd=nxt + return fd + except BaseException:os.close(fd);raise +parent=directory(os.path.dirname(root));name=os.path.basename(root) +try: + if op=='exists': + try: + info=os.stat(name,dir_fd=parent,follow_symlinks=False) + if not stat.S_ISDIR(info.st_mode):fail('cache is not a real directory') + print('true') + except FileNotFoundError:print('false') + sys.exit(0) + slot=directory(os.path.join(os.path.dirname(root),'.uploads',session,name)) + lock=os.open('lock',os.O_CREAT|os.O_RDWR|os.O_NOFOLLOW,0o600,dir_fd=slot) + if not stat.S_ISREG(os.fstat(lock).st_mode) or os.fstat(lock).st_nlink!=1:fail('invalid cache lock') + fcntl.flock(lock,fcntl.LOCK_SH if op=='guard' else fcntl.LOCK_EX) + def read(name,fd=slot): + source=os.open(name,os.O_RDONLY|os.O_NOFOLLOW,dir_fd=fd) + with os.fdopen(source,'rb') as f: + info=os.fstat(f.fileno()) + if not stat.S_ISREG(info.st_mode) or info.st_nlink!=1 or info.st_size>4096:fail('invalid cache receipt') + return json.load(f) + def owned(value): + return isinstance(value,dict) and value.get('purpose')=='paperclip-provider-pack-upload-v1' and value.get('cacheRoot')==root and value.get('session')==session and isinstance(value.get('attemptId'),str) and len(value['attemptId'])==36 and all(c in '0123456789abcdef-' for c in value['attemptId']) + def cleanup(value): + if not owned(value):fail('invalid upload ownership') + try:fd=os.open(value['attemptId'],os.O_RDONLY|os.O_DIRECTORY|os.O_NOFOLLOW,dir_fd=slot) + except FileNotFoundError:return + try: + try:marker=read('owner.json',fd) + except FileNotFoundError: + os.rmdir(value['attemptId'],dir_fd=slot);return + if marker!=value:fail('upload ownership mismatch') + finally:os.close(fd) + shutil.rmtree(value['attemptId'],dir_fd=slot) + try:current=read('current.json') + except FileNotFoundError:current=None + if op in ('begin','reap'): + if current is not None: + if not owned(current):fail('invalid upload receipt') + if current.get('ownerId')==identity['ownerId'] and current.get('generation',0)>identity['generation']:fail('stale upload generation') + cleanup(current) + if op=='reap': + if current is not None:os.unlink('current.json',dir_fd=slot) + sys.exit(0) + # Reserve the exact attempt before creating its directory. An interrupted + # begin is recoverable without recursively deleting unmarked user content. + try:os.stat(identity['attemptId'],dir_fd=slot,follow_symlinks=False);fail('upload already exists') + except FileNotFoundError:pass + fd=os.open('current.tmp',os.O_CREAT|os.O_WRONLY|os.O_NOFOLLOW,0o600,dir_fd=slot) + with os.fdopen(fd,'wb') as f: + info=os.fstat(f.fileno()) + if not stat.S_ISREG(info.st_mode) or info.st_nlink!=1:fail('invalid receipt temporary file') + f.truncate(0);f.write(identity_json.encode());f.flush();os.fsync(f.fileno()) + os.rename('current.tmp','current.json',src_dir_fd=slot,dst_dir_fd=slot);os.fsync(slot) + os.mkdir(identity['attemptId'],0o700,dir_fd=slot) + upload=os.open(identity['attemptId'],os.O_RDONLY|os.O_DIRECTORY|os.O_NOFOLLOW,dir_fd=slot) + try: + marker=os.open('owner.json',os.O_CREAT|os.O_EXCL|os.O_WRONLY|os.O_NOFOLLOW,0o600,dir_fd=upload) + with os.fdopen(marker,'wb') as f:f.write(identity_json.encode());f.flush();os.fsync(f.fileno()) + finally:os.close(upload) + elif op=='cleanup': + cleanup(identity) + if current==identity:os.unlink('current.json',dir_fd=slot) + elif op in ('guard','publish'): + if current!=identity:fail('stale upload generation') + if op=='guard':sys.exit(subprocess.run(sys.argv[5:],stdin=sys.stdin,pass_fds=(lock,)).returncode) + upload=os.open(identity['attemptId'],os.O_RDONLY|os.O_DIRECTORY|os.O_NOFOLLOW,dir_fd=slot) + try: + if read('owner.json',upload)!=identity:fail('upload ownership mismatch') + libc=ctypes.CDLL(None,use_errno=True) + if hasattr(libc,'renameat2'):status=libc.renameat2(upload,b'pack',parent,os.fsencode(name),1) + elif hasattr(libc,'renameatx_np'):status=libc.renameatx_np(upload,b'pack',parent,os.fsencode(name),4) + else:fail('atomic publication unavailable') + if status!=0 and ctypes.get_errno() not in (errno.EEXIST,errno.ENOTEMPTY):raise OSError(ctypes.get_errno(),'cache publication failed') + finally:os.close(upload) + else:fail('unknown cache operation') +finally:os.close(parent) +`; + +export async function prepareComputerProviderPackCache(input: { + cacheRoot: string; + sessionKey: string; + owner: { ownerId: string; generation: number }; + runner: CommandManagedRuntimeRunner; + verify: (root: string) => Promise; + stage: (root: string, runner: CommandManagedRuntimeRunner) => Promise; +}): Promise<"reused" | "published"> { + const session = createHash("sha256").update(input.sessionKey).digest("hex"); + const identity = { purpose: "paperclip-provider-pack-upload-v1", cacheRoot: input.cacheRoot, session, + attemptId: randomUUID(), ownerId: input.owner.ownerId, generation: input.owner.generation }; + const args = (op: string) => ["-c", cacheProgram, op, input.cacheRoot, session, JSON.stringify(identity)]; + const operation = async (op: string) => { + const result = await input.runner.execute({ command: "python3", args: args(op), bypassSession: true, timeoutMs: 180_000 }); + if (result.exitCode !== 0 || result.timedOut) throw new Error(`runner_remote_provider_cache_${op}_failed`); + return result.stdout.trim(); + }; + const verifyShared = async () => { + try { await input.verify(input.cacheRoot); } + catch (cause) { throw new Error(`runner_remote_provider_cache_corrupt: ${input.cacheRoot}; stop tasks using this pack and have an operator remove or quarantine this exact cache directory before retrying`, { cause }); } + }; + if (await operation("exists") === "true") { + await verifyShared(); + await operation("reap"); + return "reused"; + } + const stagingRoot = posix.join(posix.dirname(input.cacheRoot), ".uploads", session, posix.basename(input.cacheRoot), identity.attemptId, "pack"); + const guardedRunner: CommandManagedRuntimeRunner = { + execute: (command) => input.runner.execute({ ...command, command: "python3", args: [...args("guard"), command.command, ...(command.args ?? [])] }), + }; + try { + await operation("begin"); + await input.stage(stagingRoot, guardedRunner); + await input.verify(stagingRoot); + await operation("publish"); + await verifyShared(); + return "published"; + } finally { + // The marker and attempt identity prevent a delayed cleanup from touching + // a successor, even when the same session retries under a new generation. + await operation("cleanup").catch(() => undefined); + } +}