Integrate immutable Boat provider pack cache

Co-Authored-By: Paperclip <noreply@paperclip.ing>

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",

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<void>((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<Awaited<ReturnType<CommandManagedRuntimeRunner["execute"]>>>;
+    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" });
+  });
+});

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<void>;
+  stage: (root: string, runner: CommandManagedRuntimeRunner) => Promise<void>;
+}): 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);
+  }
+}