fix(native): recover exact historical provider pack code

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

doc/opencode-cold-recovery.md

@@ -60,12 +60,67 @@ with the prior session state, receipt, and backups. The cache is not a new
 execution authority and must not remain active for the fresh provider session.
 
 Live Runner reattachment does not download or replace the launch files.
-Historical sessions that recorded image-rooted paths keep those paths. This
-change cannot reconstruct files that were never saved. A governed fresh-session
+Historical sessions that recorded image-rooted paths keep those paths. A governed fresh-session
 transition can use a new pack after existing effects and receipts are reconciled.
 It must not replay a previous turn or treat a missing result as proof that no
 effect occurred.
 
+## Historical provider-pack code sources
+
+An older session can retain its exact pack manifest while the current server
+image contains a different pack. When neither the staged nor the preinstalled
+pack verifies, the normal uploader resolves the selected historical code from
+`ghcr.io/paperclipai/paperclip`. It uses the manifest's recorded source revision
+and Linux architecture. The full source-SHA image tag is a lookup hint; a tag
+name alone does not prove the code identity.
+
+The resolver verifies the OCI index, selected platform manifest, image config,
+source revision, and dedicated provider-pack transport layer. It then verifies
+the selected canonical pack manifest, compressed layer hashes, tar paths and
+links, and each file layer's plaintext bytes. The final installed pack still
+passes the normal verification against the original selected manifest. The
+OpenCode version alone cannot admit a different proxy or contract. This source
+selection also serves other providers that use the full provider-pack uploader;
+it does not change their compatibility or qualification rules.
+
+The code source is downloaded and verified before the uploader can clear or
+transfer anything to the target. A historical source requires the existing
+qualified archive transport. An unavailable transport leaves the target intact.
+Missing images, changed bytes, invalid source identity, and unsupported packs
+fail closed. The resolver never resets provider state, substitutes the current
+pack, changes the stored launch paths, or starts an old application image.
+
+The temporary cache contains public code and source provenance only. It uses
+private, process-account-owned directories and exact manifest keys. An empty
+host can reconstruct the cache from the registry. A cached source is checked
+again before upload. Imports stream data with a four-GiB code limit and a
+three-minute deadline, allow two active imports, and bound queued distinct
+sources to 32. Import waits and downloads respond to cancellation; one cancelled
+caller does not cancel a shared import still needed by another caller. Inactive
+instances perform no registry reads or cache scans. Each import requires disk
+headroom before writing; completed cached code remains until temporary storage
+is removed.
+
+The normal per-turn startup owns the source wait, including OpenCode starts
+without a warm execution context. Stop, execution-lease loss, and startup
+completion cancel that wait. An existing execution context can cancel it sooner.
+The source deadline applies before Runner admission. It does not increase the
+separate cold-admission or provider-turn time limits.
+
+The resolver uses only an anonymous token scoped to reading this public code
+repository. It does not use database, sandbox, model, or object-store credentials.
+Its cache and registry dependency are separate from the encrypted S3 backup of
+provider conversation state and workspace files.
+
+The Docker publication workflow publishes full source-SHA tags. There is no
+image-deletion workflow in this repository, but this is not a permanent registry
+retention guarantee. GitHub package administrators can
+[delete package versions](https://docs.github.com/en/packages/learn-github-packages/deleting-and-restoring-a-package).
+Historical continuation therefore depends on retaining the corresponding code
+image in GHCR, or retaining a verified source cache. Deleting both keeps the
+session blocked with its original state intact; an S3 provider-state backup alone
+does not reconstruct legacy image-owned executable code.
+
 Artifact cleanup uses the existing bounded checkpoint and backup expiration
 owners. It requires closed work, settled native ownership, and exact stopped
 remote execution receipts. Backup retirement releases only that backup reference.

server/src/services/native-runtime/native-session-executor.test.ts

@@ -112,6 +112,7 @@ import { forwardRemoteMcpRequest } from "./remote-mcp-relay.js";
 import { createCapabilityRunnerdProviderEnvironment } from "@paperclipai/paperclip-runner/live";
 import * as nativeSessionResume from "./native-session-resume.js";
 import * as nativeLeaseRenewal from "./native-execution-lease.js";
+import * as retainedPackSource from "./retained-provider-pack-source.js";
 import { buildNativeHeartbeatPreparationSpans, createNativeRunTrace } from "./native-run-trace.js";
 import { NativeRunnerOwnershipUnverifiedError } from "./native-runner-ownership.js";
 import type { AdapterRuntimeEvent } from "../../adapters/index.js";
@@ -7488,6 +7489,65 @@ function cancellationDb(options?: {
 }
 
 describe("native startup cancellation fence", () => {
+  it.each(["stop", "lease-loss"])("cancels the actual per-turn OpenCode code-source wait on %s before provider dispatch", async mode => {
+    const root = await mkdtemp(join(tmpdir(), "native-pack-source-caller-"));
+    const daemonPath = join(root, "paperclip-runnerd");
+    const daemon = Buffer.alloc(64); daemon.set([0x7f, 0x45, 0x4c, 0x46, 2, 1]); daemon.writeUInt16LE(62, 18);
+    await writeFile(daemonPath, daemon);
+    const sha = "sha256:" + "a".repeat(64);
+    const artifacts = Object.fromEntries(Object.entries({ grokLauncher: "dist/providers/grok/launcher.cjs", nodeCommand: "node_modules/node/bin/node",
+      productionLock: "pnpm-lock.yaml", opencodeCommand: "node_modules/.bin/opencode", opencodeExecutable: "node_modules/opencode-ai/bin/opencode.exe",
+      opencodeProxy: "dist/cli/opencode-app-server-proxy.cjs", acpxSidecar: "dist/cli/acpx-runtime-sidecar.cjs" }).map(([name, path]) => [name, { path, sha256: sha }]));
+    const payload = { compatibility: { opencode: 1, acpx: 1 }, contractDigest: sha, target: { platform: "linux", architecture: "x64" }, runnerSourceRevision: "a".repeat(40),
+      pins: { nodeMinimum: "24.11.0", codex: QUALIFIED_ACPX_PROFILES.codex.agentRuntimeVersion, opencode: "1.18.34", acpx: "0.13.1",
+        grok: QUALIFIED_ACPX_PROFILES.grok.agentRuntimeVersion, claudeAcp: QUALIFIED_ACPX_PROFILES.claude.agentServerVersion, codexAcp: QUALIFIED_ACPX_PROFILES.codex.agentServerVersion },
+      acpxProfileDigests: Object.fromEntries(["grok", "claude", "codex"].map(agent => [agent, QUALIFIED_ACPX_PROFILES[agent as "grok" | "claude" | "codex"].commandDigest])),
+      artifacts, distDigest: sha, bridgeDigest: "sha256:" + createHash("sha256").update(sha + "\n" + sha + "\n" + sha).digest("hex") };
+    await writeFile(join(root, "provider-pack.json"), JSON.stringify({ schema: "paperclip-runner/remote-provider-pack/v1", payload,
+      digest: "sha256:" + createHash("sha256").update(canonicalProviderPackJson(payload)).digest("hex") }));
+    const current = { ...execution, binding: { ...execution.binding, runId: randomUUID(), executionWorkspaceId: randomUUID() },
+      workspace: { ...execution.workspace, cwd: root }, provider: { kind: "opencode", model: "openai/test" },
+      session: { ...execution.session, normalizedSessionId: randomUUID(), driverKind: "opencode_server", lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } } } as NativeExecutionInputV1;
+    const execute = vi.fn(async (command: { args?: string[] }) => ({ exitCode: command.args?.[0] === "-e" ? 1 : 0, timedOut: false, signal: null, stderr: "", stdout: command.args?.[0] === "--build-metadata" ? JSON.stringify({
+      schema: "paperclip-runner/runnerd-build-metadata/v1", binaryName: "paperclip-runnerd", packageName: "@paperclipai/paperclip-runner", binaryContractVersion: 2,
+      durableSessionCapabilities: ["unlimited_runtime", "connection_lease_renewal"], prpTransportModes: ["listen_ws"], providerPacks: [],
+    }) : "" }));
+    const syncIn = vi.fn(async (_operations: Array<{ files: Array<{ targetPath?: string }> }>) => ({ operations: [] }));
+    let entered!: () => void, release!: () => void, sourceSignal: AbortSignal | undefined, lost: (() => Promise<unknown>) | undefined;
+    const reached = new Promise<void>(done => { entered = done; });
+    const source = vi.spyOn(retainedPackSource, "resolveProviderPackUploadSource").mockImplementation(input => {
+      sourceSignal = input.signal; entered();
+      return new Promise((_resolve, reject) => {
+        release = () => reject(new Error("owned fixture cleanup"));
+        input.signal?.addEventListener("abort", () => reject(input.signal!.reason), { once: true });
+      });
+    });
+    const originalRenewal = nativeLeaseRenewal.startNativeSessionExecutionLeaseRenewal;
+    const renewal = vi.spyOn(nativeLeaseRenewal, "startNativeSessionExecutionLeaseRenewal").mockImplementation(input => {
+      lost = () => Promise.resolve(input.onLost());
+      return originalRenewal(input);
+    });
+    const running = executePaperclipNativeSession({ db: leaseDb(current), execution: current, useRunnerd: true,
+      runnerInstanceId: "source-caller-runner", runnerRemoteBinaryPath: daemonPath, runnerRemoteProviderPackPath: root,
+      runnerEnvironment: { ...configuredEnvironmentProjection({}), PAPERCLIP_NATIVE_MCP_NAME: "source-caller", PAPERCLIP_NATIVE_MCP_URL: "http://127.0.0.1/mcp/gateways/source-caller", PAPERCLIP_NATIVE_MCP_TOKEN: "owned-fixture" },
+      runnerIngressAuthorized: true, runnerExecutionTarget: { kind: "remote", transport: "sandbox", remoteCwd: "/workspace", environmentId: "environment", leaseId: "lease", providerKey: "daytona",
+        effectiveCapabilities: { runnerWebSocketIngress: true }, runner: { execute, syncIn } } as never });
+    const failure = running.then(() => { throw Error("unexpected provider completion"); }, error => error);
+    try {
+      await Promise.race([reached, failure.then(error => { throw error; })]);
+      expect(sourceSignal).toBeInstanceOf(AbortSignal);
+      const stopping = mode === "stop" ? cancelNativeSession(current.binding.runId, "owned source wait Stop") : lost!();
+      expect(await failure).toMatchObject({ message: mode === "stop" ? "native_cancellation_pending_recovery" : "native_controller_detached_for_restart" });
+      await stopping;
+      expect(sourceSignal!.reason).toMatchObject({ message: mode === "stop" ? "native_cancellation_pending_recovery" : "native_session_lease_lost" });
+      expect(sourceSignal!.aborted).toBe(true);
+      expect(state.createBackend).not.toHaveBeenCalled();
+      expect(syncIn.mock.calls.flatMap(call => (call[0] as { files?: Array<{ targetPath?: string }> } | undefined)?.files ?? []).some(file => file.targetPath?.endsWith("/provider-pack"))).toBe(false);
+    } finally {
+      release?.(); await failure; source.mockRestore(); renewal.mockRestore(); await rm(root, { recursive: true, force: true });
+    }
+  });
+
   it.each(["deadline", "stop"])("retires a claimed warm owner when %s interrupts its reuse diagnostic", async mode => {
     const first: NativeExecutionInputV1 = { ...execution,
       binding: { ...execution.binding, runId: `reuse-log-first-${mode}`, executionWorkspaceId: `reuse-log-workspace-${mode}` },

server/src/services/native-runtime/native-session-executor.ts

@@ -378,6 +378,7 @@ type NativeSessionStartup = {
   priorRestartDetach?: symbol;
   leaseOwner?: string;
   executionLeaseRenewal?: NativeExecutionLeaseRenewal;
+  providerPackSourceAbort: AbortController;
   /** A fresh claim whose preparation failed before any provider turn. */
   unstartedClaim?: { leaseOwner: string; attempt: number };
   /** A provider session handle was published; a turn may have followed it. */
@@ -8780,6 +8781,7 @@ async function performNativeCancellation(
     && startup.sessionPublished !== true && startup.providerWorkMayPrecede !== true));
   if (startup) {
     startup.stopRequested = true;
+    startup.providerPackSourceAbort.abort(new NativeCancellationPendingRecoveryError());
     startup.executionContext?.abort(new Error("native_session_cancelled"));
     startup.cancellationSettled = new Promise<void>(resolve => { settleStartupCancellation = resolve; });
   }
@@ -9161,6 +9163,7 @@ export async function executePaperclipNativeSession(input: {
   let resolveStartup!: (session: ActiveNativeSession | null) => void;
   let settle!: () => void;
   const startup: NativeSessionStartup = {
+    providerPackSourceAbort: new AbortController(),
     priorRestartDetach: nativeRunsDetachingForRestart.get(runId),
     providerWorkMayPrecede: Boolean(input.restartRecovery),
     settled: new Promise<void>(resolve => { settle = resolve; }),
@@ -9317,6 +9320,7 @@ export async function executePaperclipNativeSession(input: {
     }
     throw error;
   } finally {
+    startup.providerPackSourceAbort.abort(new Error("native_execution_startup_completed"));
     for (const lease of diagnosticLeases) lease.retire();
     let startupUnsettled = false;
     if (startup.executionContext) {
@@ -9876,6 +9880,7 @@ async function executePaperclipNativeSessionWithinScope(
             remote: input.runnerExecutionTarget?.kind === "remote",
             confirmedDeadline: confirmedLeaseDeadline,
             onLost: () => {
+              startup.providerPackSourceAbort.abort(new Error("native_session_lease_lost"));
               startup.executionContext?.abort(new Error("native_session_lease_lost"));
               return input.runnerExecutionTarget?.kind === "remote"
                 ? detachNativeSessionsForRestart([input.execution.binding.runId], leaseOwner)
@@ -12801,6 +12806,7 @@ export async function prepareRemoteProviderPackOnTarget(input: {
   runner: CommandManagedRuntimeRunner;
   configuredProviderPackRoot: string | null;
   manifest: RemoteProviderPackManifest;
+  signal?: AbortSignal;
   trace?: NativeRunTrace;
   onLog?: (stream: "stdout" | "stderr", chunk: string) => void | Promise<void>;
 }): Promise<{ root: string; manifest: RemoteProviderPackManifest; source: "staged" | "preinstalled" | "uploaded" }> {
@@ -12891,16 +12897,18 @@ export async function prepareRemoteProviderPackOnTarget(input: {
         root: configuredProviderPackRoot,
         sourceDigest: readRemoteProviderPackIdentity(configuredProviderPackRoot, false).digest,
         expectedDigest: expectedProviderPackManifest.digest,
+        manifest: expectedProviderPackManifest,
+        signal: input.signal,
         runner: remoteCommandRunner,
         cwd: remoteTarget.remoteCwd,
         targetPath: stagedRemoteProviderPackRoot,
         reuseRoots: preinstalledProviderPackRoots,
         trace: input.trace,
-        verifyDirectory: () => { readRemoteProviderPackManifest(configuredProviderPackRoot); },
-        stageDirectory: () => stageRemoteRunnerDirectory({
+        verifyDirectory: (sourceRoot) => { readRemoteProviderPackManifest(sourceRoot); },
+        stageDirectory: (sourceRoot) => stageRemoteRunnerDirectory({
           target: remoteTarget,
           runner: remoteCommandRunner,
-          sourcePath: configuredProviderPackRoot,
+          sourcePath: sourceRoot,
           targetPath: stagedRemoteProviderPackRoot,
           mode: 0o700,
         }),
@@ -14895,9 +14903,14 @@ async function createRunnerdBackendWithinSessionClaim(
       expectedProviderPackManifest &&
       stagedRemoteProviderPackRoot
     ) {
+      const sourceRunId = sessionToolAuthorityEpochs.get(sessionScopeId)?.runId ?? input.execution.binding.runId;
+      const startupSignal = nativeSessionStartups.get(sourceRunId)?.providerPackSourceAbort.signal;
+      const contextSignal = input.executionContext?.active ? input.executionContext.signal : undefined;
+      const sourceSignals = [startupSignal, contextSignal].filter((signal): signal is AbortSignal => !!signal);
       const prepared = await prepareRemoteProviderPackOnTarget({
         target: remoteTarget, runner: remoteCommandRunner, configuredProviderPackRoot,
         manifest: expectedProviderPackManifest, trace: input.trace, onLog: input.onLog,
+        signal: sourceSignals.length ? AbortSignal.any(sourceSignals) : undefined,
       });
       activeRemoteProviderPackRoot = prepared.root;
       providerPackVerified = true;

server/src/services/native-runtime/provider-pack-upload.ts

@@ -1,6 +1,9 @@
 import type { CommandManagedRuntimeRunner } from "@paperclipai/adapter-utils/command-managed-runtime";
 import type { NativeRunTrace } from "./native-run-trace.js";
 import { readProviderPackArchive, stageProviderPackArchive } from "./provider-pack-archive.js";
+import type { ProviderPackIdentity } from "./remote-provider-pack-verification.js";
+import { resolveProviderPackUploadSource } from "./retained-provider-pack-source.js";
+import { providerPackExtractionProbe } from "./provider-pack-extraction.js";
 
 const quote = (value: string) => `'${value.replaceAll("'", "'\\''")}'`;
 
@@ -11,13 +14,15 @@ export async function uploadVerifiedProviderPack(input: {
   sourceDigest: string;
   /** Identity selected for this execution; a retained session can pin an older release. */
   expectedDigest: string;
+  manifest?: ProviderPackIdentity;
+  signal?: AbortSignal;
   runner: CommandManagedRuntimeRunner;
   cwd: string;
   targetPath: string;
   reuseRoots: readonly string[];
   trace?: NativeRunTrace;
-  verifyDirectory: () => void;
-  stageDirectory: () => Promise<void>;
+  verifyDirectory: (root: string) => void | Promise<void>;
+  stageDirectory: (root: string) => Promise<void>;
   verifyUploaded: () => Promise<void>;
 }): Promise<void> {
   // The controller can only stage the pack it ships. A retained continuation
@@ -28,13 +33,27 @@ export async function uploadVerifiedProviderPack(input: {
     throw new Error(`runner_remote_provider_artifact_incompatible: selected provider pack ${short(input.expectedDigest)} is not the controller's pack ${short(input.sourceDigest)}; install the matching Paperclip package and Daytona image`);
   }
   const measure = <T>(name: string, fn: () => Promise<T>) => input.trace ? input.trace.measure(name, fn) : fn();
-  const verifyDirectory = () => measure("provider_pack.directory.verify", async () => input.verifyDirectory());
+  input.signal?.throwIfAborted();
+  if (input.manifest && input.manifest.digest !== input.expectedDigest) throw Error("runner_remote_provider_source_invalid");
+  const source = input.manifest ? await measure("provider_pack.source", () => resolveProviderPackUploadSource({
+    root: input.root, manifest: input.manifest!, signal: input.signal,
+  })) : { root: input.root, archiveOnly: false };
+  const verifyDirectory = () => measure("provider_pack.directory.verify", async () => input.verifyDirectory(source.root));
   // Archive installation reads the sealed archives, not the expanded tree.
   // Hash the directory only when it is actually the upload source. The
   // installed pack still passes full verification against the selected identity.
   const archive = await measure("provider_pack.archive.verify", () =>
-    readProviderPackArchive(input.root, input.expectedDigest));
+    readProviderPackArchive(source.root, input.expectedDigest));
+  if (source.archiveOnly) {
+    if (!archive) throw Error("runner_remote_provider_source_invalid");
+    const capability = await measure("provider_pack.archive.probe", () => input.runner.execute({
+      command: "sh", args: ["-c", `${providerPackExtractionProbe}${"kind" in archive ? " && node -e 'process.exit(Number(process.versions.node.split(\".\")[0])>=18?0:1)'" : ""}`],
+      bypassSession: true, timeoutMs: 10_000,
+    }));
+    if (capability.exitCode !== 0 || capability.timedOut) throw Error("runner_remote_provider_archive_transport_unavailable");
+  }
   if (!archive) await verifyDirectory();
+  input.signal?.throwIfAborted();
   const cleared = await measure("provider_pack.clear_stale", () => input.runner.execute({
     command: "sh", args: ["-c", `rm -rf ${quote(input.targetPath)}`],
     cwd: input.cwd, bypassSession: true, timeoutMs: 10_000,
@@ -47,8 +66,9 @@ export async function uploadVerifiedProviderPack(input: {
       archive, runner: input.runner, targetPath: input.targetPath,
       reuseRoots: input.reuseRoots, trace: input.trace,
     })) return;
+    if (source.archiveOnly) throw Error("runner_remote_provider_archive_transport_unavailable");
     if (archive) await verifyDirectory();
-    await input.stageDirectory();
+    await input.stageDirectory(source.root);
   });
   await measure("provider_pack.verify_uploaded", input.verifyUploaded);
 }

server/src/services/native-runtime/retained-provider-pack-oci.ts

@@ -0,0 +1,161 @@
+import { createHash } from "node:crypto";
+import { createReadStream, createWriteStream } from "node:fs";
+import { mkdir } from "node:fs/promises";
+import { join } from "node:path";
+import { Readable, Transform } from "node:stream";
+import { pipeline } from "node:stream/promises";
+import { createGunzip } from "node:zlib";
+import tar from "tar-stream";
+import type { ProviderPackIdentity } from "./remote-provider-pack-verification.js";
+
+const repository = "paperclipai/paperclip";
+const registry = "https://ghcr.io";
+export const MAX_RETAINED_PACK_BYTES = 4 * 1024 ** 3;
+const maxMetadataBytes = 1024 * 1024;
+const digest = (bytes: Buffer) => `sha256:${createHash("sha256").update(bytes).digest("hex")}`;
+const validDigest = (value: unknown): value is string => typeof value === "string" && /^sha256:[a-f0-9]{64}$/.test(value);
+function fail(): never { throw Error("runner_remote_provider_source_invalid"); }
+type Blob = { digest: string; size: number; mediaType?: string };
+type Manifest = { schemaVersion: number; config: Blob; layers: Blob[] };
+export type RetainedPackProvenance = {
+  schema: "paperclip.retained-provider-pack-source.v1";
+  sourceRevision: string;
+  providerPackDigest: string;
+  imageIndexDigest: string;
+  imageManifestDigest: string;
+  imageConfigDigest: string;
+  transportLayerDigest: string;
+  transportLayerBytes: number;
+};
+
+function blob(value: Blob, maximum: number): Blob {
+  if (!value || !validDigest(value.digest) || !Number.isSafeInteger(value.size) || value.size < 1 || value.size > maximum) fail();
+  return value;
+}
+
+async function metadata(response: Response, signal: AbortSignal, expected?: Blob): Promise<Buffer> {
+  if (!response.body) fail();
+  const chunks: Buffer[] = [];
+  let bytes = 0;
+  try {
+    for await (const chunk of Readable.fromWeb(response.body as Parameters<typeof Readable.fromWeb>[0])) {
+      signal.throwIfAborted();
+      bytes += chunk.length;
+      if (bytes > maxMetadataBytes) fail();
+      chunks.push(Buffer.from(chunk));
+    }
+  } catch (error) {
+    await response.body.cancel().catch(() => undefined);
+    throw error;
+  }
+  const result = Buffer.concat(chunks);
+  if (expected && (result.length !== expected.size || digest(result) !== expected.digest)) fail();
+  return result;
+}
+
+function bounded(limit: number) {
+  let bytes = 0;
+  return new Transform({ transform(chunk, _encoding, done) {
+    bytes += chunk.length;
+    done(bytes > limit ? Error("runner_remote_provider_source_size_limit") : null, chunk);
+  } });
+}
+
+async function unpackTransport(layer: string, target: string, signal: AbortSignal) {
+  const archive = tar.extract();
+  const names = new Set<string>();
+  let bytes = 0;
+  archive.on("entry", (header, body, next) => {
+    body.on("error", () => undefined);
+    void (async () => {
+      signal.throwIfAborted();
+      const normalized = header.name.replace(/^\.\//u, "");
+      const name = header.type === "directory" ? normalized.replace(/\/$/u, "") : normalized;
+      if (names.has(name) || names.size >= 10_010) fail();
+      names.add(name);
+      if (header.type === "directory") {
+        if (!["opt", "opt/paperclip-runner", "opt/paperclip-runner/provider-pack.archives"].includes(name)) fail();
+        body.resume();
+        return;
+      }
+      if (header.type !== "file" || !/^opt\/paperclip-runner\/provider-pack(?:\.archive\.json|\.archives\/(?:overlay|[0-9]{1,4})\.tar\.zst)$/u.test(name)
+        || !Number.isSafeInteger(header.size) || header.size! < 1 || (bytes += header.size!) > MAX_RETAINED_PACK_BYTES
+        || name.endsWith(".json") && header.size! > maxMetadataBytes) fail();
+      const relative = name.slice("opt/paperclip-runner/".length);
+      if (relative.includes("/")) await mkdir(join(target, "provider-pack.archives"), { recursive: true, mode: 0o700 });
+      await pipeline(body, createWriteStream(join(target, relative), { flags: "wx", mode: 0o600 }), { signal });
+    })().then(() => next(), error => archive.destroy(error));
+  });
+  await pipeline(createReadStream(layer), createGunzip(), bounded(MAX_RETAINED_PACK_BYTES + 16 * 1024 ** 2), archive, { signal });
+}
+
+export async function receiveRetainedProviderPack(input: {
+  manifest: ProviderPackIdentity;
+  directory: string;
+  signal: AbortSignal;
+}): Promise<RetainedPackProvenance> {
+  const { manifest: selected, directory, signal } = input;
+  const revision = selected.payload.runnerSourceRevision;
+  const architecture = selected.payload.target.architecture === "x64" ? "amd64" : selected.payload.target.architecture === "arm64" ? "arm64" : null;
+  if (!/^[a-f0-9]{40}$/.test(revision) || selected.payload.target.platform !== "linux" || !architecture) fail();
+  const request = async (path: string, token?: string, allowRedirect = false) => {
+    signal.throwIfAborted();
+    let response = await fetch(`${registry}${path}`, { redirect: "manual", signal, headers: {
+      Accept: "application/vnd.oci.image.index.v1+json, application/vnd.oci.image.manifest.v1+json, application/vnd.docker.distribution.manifest.list.v2+json, application/vnd.docker.distribution.manifest.v2+json",
+      ...(token ? { Authorization: `Bearer ${token}` } : {}),
+    } });
+    if ([302, 307].includes(response.status) && allowRedirect) {
+      const location = response.headers.get("location");
+      await response.body?.cancel();
+      if (!location || location.length > 8192) fail();
+      const url = new URL(location, registry);
+      if (url.protocol !== "https:" || url.username || url.password || !["ghcr.io", "pkg-containers.githubusercontent.com"].includes(url.hostname)) fail();
+      response = await fetch(url, { redirect: "error", signal });
+    }
+    if (!response.ok) {
+      await response.body?.cancel();
+      throw Error(`runner_remote_provider_source_unavailable: HTTP ${response.status}`);
+    }
+    return response;
+  };
+  const authentication = JSON.parse((await metadata(await request(`/token?service=ghcr.io&scope=repository:${repository}:pull`), signal)).toString()) as { token?: unknown };
+  if (typeof authentication.token !== "string" || authentication.token.length < 1 || authentication.token.length > 32 * 1024) fail();
+  const token = authentication.token;
+  const indexBytes = await metadata(await request(`/v2/${repository}/manifests/sha-${revision}`, token), signal);
+  const index = JSON.parse(indexBytes.toString()) as { schemaVersion: number; manifests: Array<Blob & { platform?: { os?: string; architecture?: string } }> };
+  if (index.schemaVersion !== 2 || !Array.isArray(index.manifests) || index.manifests.length > 32) fail();
+  const candidates = index.manifests.filter(row => row.platform?.os === "linux" && row.platform.architecture === architecture);
+  if (candidates.length !== 1) fail();
+  const child = blob(candidates[0]!, maxMetadataBytes);
+  const manifestBytes = await metadata(await request(`/v2/${repository}/manifests/${child.digest}`, token), signal, child);
+  const image = JSON.parse(manifestBytes.toString()) as Manifest;
+  if (image.schemaVersion !== 2 || !Array.isArray(image.layers) || image.layers.length > 128) fail();
+  const configBlob = blob(image.config, maxMetadataBytes);
+  const configBytes = await metadata(await request(`/v2/${repository}/blobs/${configBlob.digest}`, token, true), signal, configBlob);
+  const config = JSON.parse(configBytes.toString()) as { architecture: string; os: string; config?: { Labels?: Record<string, string> }; history: Array<{ empty_layer?: boolean; created_by?: string }> };
+  if (config.architecture !== architecture || config.os !== "linux" || config.config?.Labels?.["org.opencontainers.image.revision"] !== revision
+    || config.config.Labels["org.opencontainers.image.source"] !== "https://github.com/paperclipai/paperclip"
+    || !Array.isArray(config.history) || config.history.length > 1024) fail();
+  const history = config.history.filter(row => !row.empty_layer);
+  if (history.length !== image.layers.length) fail();
+  const indices = history.flatMap((row, index) => /^COPY \/provider-pack-transport\/? \/opt\/paperclip-runner\/? # buildkit$/u.test(row.created_by ?? "") ? [index] : []);
+  if (indices.length !== 1) fail();
+  const transport = blob(image.layers[indices[0]!]!, MAX_RETAINED_PACK_BYTES);
+  if (!["application/vnd.oci.image.layer.v1.tar+gzip", "application/vnd.docker.image.rootfs.diff.tar.gzip"].includes(transport.mediaType ?? "")) fail();
+  const response = await request(`/v2/${repository}/blobs/${transport.digest}`, token, true);
+  if (!response.body) fail();
+  let bytes = 0;
+  const hash = createHash("sha256");
+  const meter = new Transform({ transform(chunk, _encoding, done) {
+    bytes += chunk.length;
+    hash.update(chunk);
+    done(bytes > transport.size ? Error("runner_remote_provider_source_size_limit") : null, chunk);
+  } });
+  const layer = join(directory, "transport.tar.gz");
+  await pipeline(Readable.fromWeb(response.body as Parameters<typeof Readable.fromWeb>[0]), meter, createWriteStream(layer, { flags: "wx", mode: 0o600 }), { signal });
+  if (bytes !== transport.size || `sha256:${hash.digest("hex")}` !== transport.digest) fail();
+  await unpackTransport(layer, directory, signal);
+  return { schema: "paperclip.retained-provider-pack-source.v1", sourceRevision: revision, providerPackDigest: selected.digest,
+    imageIndexDigest: digest(indexBytes), imageManifestDigest: child.digest, imageConfigDigest: configBlob.digest,
+    transportLayerDigest: transport.digest, transportLayerBytes: transport.size };
+}

server/src/services/native-runtime/retained-provider-pack-source.test.ts

@@ -0,0 +1,306 @@
+import { execFile } from "node:child_process";
+import { createHash } from "node:crypto";
+import { createServer } from "node:http";
+import { promises as fs } from "node:fs";
+import os from "node:os";
+import path from "node:path";
+import { promisify } from "node:util";
+import { pathToFileURL } from "node:url";
+import { gzipSync, zstdCompressSync } from "node:zlib";
+import tar from "tar-stream";
+import { afterEach, expect, it, vi } from "vitest";
+import { canonicalProviderPackJson } from "./remote-provider-pack-verification.js";
+import { uploadVerifiedProviderPack } from "./provider-pack-upload.js";
+const archiveScript = pathToFileURL(path.resolve(import.meta.dirname, "../../../../scripts/create-provider-pack-archive.mjs")).href;
+const { createProviderPackLayers } = await import(archiveScript);
+
+const exec = promisify(execFile);
+const hash = (value: string | Buffer) => createHash("sha256").update(value).digest("hex");
+const roots: string[] = [];
+const servers: ReturnType<typeof createServer>[] = [];
+const actualFetch = globalThis.fetch;
+
+afterEach(async () => {
+  vi.unstubAllGlobals();
+  vi.unstubAllEnvs();
+  vi.restoreAllMocks();
+  await Promise.all(servers.splice(0).map(server => new Promise<void>((done, reject) => {
+    server.close(error => error ? reject(error) : done());
+    server.closeAllConnections();
+  })));
+  await Promise.all(roots.splice(0).map(root => fs.rm(root, { recursive: true, force: true })));
+});
+
+async function fixture(options: { wrongCode?: boolean; unsafeOverlay?: boolean; changedFileLayer?: boolean; hardlink?: "safe" | "escape" | "unknown" } = {}) {
+  const root = await fs.mkdtemp(path.join(os.tmpdir(), "retained-pack-test-"));
+  roots.push(root);
+  const temporary = path.join(root, "tmp");
+  await fs.mkdir(temporary);
+  vi.stubEnv("TMPDIR", temporary);
+  const makePack = async (name: string, tool: string, source: string) => {
+    const pack = path.join(root, name);
+    await fs.mkdir(path.join(pack, "bin"), { recursive: true });
+    const payload = { pins: { opencode: "1.18.34" }, target: { platform: "linux", architecture: "x64" }, runnerSourceRevision: source,
+      contractDigest: `sha256:${hash(tool)}`, artifacts: { opencodeProxy: { path: "bin/tool", sha256: `sha256:${hash(tool)}` } } };
+    const manifest = { schema: "paperclip-runner/remote-provider-pack/v1", payload, digest: `sha256:${hash(canonicalProviderPackJson(payload))}` };
+    await fs.writeFile(path.join(pack, "provider-pack.json"), JSON.stringify(manifest));
+    await fs.writeFile(path.join(pack, "bin/tool"), tool, { mode: 0o755 });
+    await fs.writeFile(path.join(pack, "bin/large"), Buffer.alloc(1024 * 1024, 42), { mode: 0o755 });
+    await createProviderPackLayers(pack);
+    return { pack, manifest };
+  };
+  const selected = await makePack("old", "retained provider code", "a".repeat(40));
+  const current = await makePack("current", "current incompatible code", "b".repeat(40));
+  if (options.changedFileLayer) {
+    const unsafe = tar.pack(), chunks: Buffer[] = [];
+    const collected = (async () => { for await (const chunk of unsafe) chunks.push(Buffer.from(chunk)); })();
+    const body = Buffer.alloc(1024 * 1024, 43);
+    await new Promise<void>((done, reject) => unsafe.entry({ name: "bin/large", type: "file", size: body.length, mode: 0o755 }, body, error => error ? reject(error) : done()));
+    unsafe.finalize(); await collected;
+    const bytes = zstdCompressSync(Buffer.concat(chunks));
+    await fs.writeFile(selected.pack + ".archives/0.tar.zst", bytes);
+    const receipt = JSON.parse(await fs.readFile(selected.pack + ".archive.json", "utf8"));
+    receipt.files[0].archive = { bytes: bytes.length, sha256: hash(bytes) };
+    await fs.writeFile(selected.pack + ".archive.json", JSON.stringify(receipt));
+  }
+  if (options.unsafeOverlay || options.hardlink) {
+    const unsafe = tar.pack(), chunks: Buffer[] = [];
+    const collected = (async () => { for await (const chunk of unsafe) chunks.push(Buffer.from(chunk)); })();
+    for (const [name, body] of [["provider-pack.json", JSON.stringify(selected.manifest)], ["bin/tool", "retained provider code"], ...(options.unsafeOverlay ? [["../outside", "must not be written"]] : [])]) {
+      await new Promise<void>((done, reject) => unsafe.entry({ name: name!, type: "file", size: Buffer.byteLength(body!) }, body!, error => error ? reject(error) : done()));
+    }
+    if (options.hardlink) await new Promise<void>((done, reject) => unsafe.entry({ name: "bin/tool-copy", type: "link", linkname: options.hardlink === "safe" ? "./bin/tool" : options.hardlink === "escape" ? "../outside" : "bin/missing", mode: 0o644 }, error => error ? reject(error) : done()));
+    unsafe.finalize(); await collected;
+    const bytes = zstdCompressSync(Buffer.concat(chunks));
+    await fs.writeFile(selected.pack + ".archives/overlay.tar.zst", bytes);
+    const receipt = JSON.parse(await fs.readFile(selected.pack + ".archive.json", "utf8"));
+    receipt.overlay = { bytes: bytes.length, sha256: hash(bytes) };
+    await fs.writeFile(selected.pack + ".archive.json", JSON.stringify(receipt));
+  }
+  const transportPack = options.wrongCode ? current.pack : selected.pack;
+  const archive = tar.pack();
+  const chunks: Buffer[] = [];
+  const collected = (async () => { for await (const chunk of archive) chunks.push(Buffer.from(chunk)); })();
+  for (const name of ["provider-pack.archive.json", ...await fs.readdir(transportPack + ".archives").then(names => names.map(name => `provider-pack.archives/${name}`))]) {
+    const bytes = await fs.readFile(path.join(path.dirname(transportPack), name.replace("provider-pack", path.basename(transportPack))));
+    await new Promise<void>((done, reject) => archive.entry({ name: `opt/paperclip-runner/${name}`, type: "file", size: bytes.length }, bytes, error => error ? reject(error) : done()));
+  }
+  archive.finalize();
+  await collected;
+  const layer = gzipSync(Buffer.concat(chunks));
+  const config = Buffer.from(JSON.stringify({ architecture: "amd64", os: "linux", config: { Labels: {
+    "org.opencontainers.image.revision": selected.manifest.payload.runnerSourceRevision,
+    "org.opencontainers.image.source": "https://github.com/paperclipai/paperclip",
+  } }, history: [{ created_by: "COPY /provider-pack-transport/ /opt/paperclip-runner/ # buildkit" }] }));
+  const manifest = Buffer.from(JSON.stringify({ schemaVersion: 2, mediaType: "application/vnd.oci.image.manifest.v1+json", config: { digest: `sha256:${hash(config)}`, size: config.length }, layers: [{ digest: `sha256:${hash(layer)}`, size: layer.length, mediaType: "application/vnd.oci.image.layer.v1.tar+gzip" }] }));
+  const index = Buffer.from(JSON.stringify({ schemaVersion: 2, mediaType: "application/vnd.oci.image.index.v1+json", manifests: [{ digest: `sha256:${hash(manifest)}`, size: manifest.length, platform: { os: "linux", architecture: "amd64" } }] }));
+  const routes = new Map<string, Buffer>([
+    ["/token", Buffer.from(JSON.stringify({ token: "public-code-fixture-token" }))],
+    [`/v2/paperclipai/paperclip/manifests/sha-${selected.manifest.payload.runnerSourceRevision}`, index],
+    [`/v2/paperclipai/paperclip/manifests/sha256:${hash(manifest)}`, manifest],
+    [`/v2/paperclipai/paperclip/blobs/sha256:${hash(config)}`, config],
+    [`/v2/paperclipai/paperclip/blobs/sha256:${hash(layer)}`, layer],
+  ]);
+  const requests: string[] = [];
+  const stalled = new Set<string>();
+  const observers = new Map<string, () => void>();
+  const server = createServer((request, response) => {
+    const route = new URL(request.url!, "http://fixture").pathname;
+    requests.push(route);
+    const bytes = routes.get(route);
+    response.writeHead(bytes ? 200 : 404, { "content-length": bytes?.length ?? 0 });
+    response.flushHeaders();
+    observers.get(route)?.();
+    if (stalled.has(route)) return;
+    response.end(bytes);
+  });
+  servers.push(server);
+  await new Promise<void>(done => server.listen(0, "127.0.0.1", done));
+  const address = server.address();
+  if (!address || typeof address === "string") throw Error("fixture server unavailable");
+  vi.stubGlobal("fetch", (input: string | URL | Request, init?: RequestInit) => {
+    const url = new URL(typeof input === "string" || input instanceof URL ? input : input.url);
+    expect(url.hostname).toBe("ghcr.io");
+    return actualFetch(`http://127.0.0.1:${address.port}${url.pathname}${url.search}`, init);
+  });
+  const bin = path.join(root, "commands");
+  await fs.mkdir(bin);
+  const tarCommand = (await exec("sh", ["-c", "command -v gtar || command -v tar"])).stdout.trim();
+  await fs.symlink(tarCommand, path.join(bin, "tar"));
+  const execute = vi.fn(async (input: { command: string; args?: string[] }) => {
+    try {
+      const result = await exec(input.command, input.args ?? [], { env: { ...process.env, PATH: `${bin}:${process.env.PATH}` } });
+      return { exitCode: 0, timedOut: false, stdout: result.stdout, stderr: result.stderr };
+    } catch (error) { return { exitCode: 1, timedOut: false, stdout: "", stderr: String(error) }; }
+  });
+  const syncIn = vi.fn(async (operations: Array<{ files: Array<{ sourcePath: string; targetPath: string }> }>) => {
+    for (const operation of operations) for (const file of operation.files) {
+      await fs.mkdir(path.dirname(file.targetPath), { recursive: true });
+      await fs.copyFile(file.sourcePath, file.targetPath);
+    }
+    return { operations: [] };
+  });
+  return { root, selected, current, requests, routes, runner: { execute, syncIn }, targetPath: path.join(root, "target"),
+    temporary, layerRoute: `/v2/paperclipai/paperclip/blobs/sha256:${hash(layer)}`,
+    stall: (route: string) => { stalled.add(route); },
+    observed: (route: string) => new Promise<void>(done => { if (requests.includes(route)) done(); else observers.set(route, done); }),
+  };
+}
+
+function upload(input: Awaited<ReturnType<typeof fixture>>, extra: Partial<Parameters<typeof uploadVerifiedProviderPack>[0]> = {}) {
+  return uploadVerifiedProviderPack({ root: input.current.pack, expectedDigest: input.selected.manifest.digest,
+    manifest: input.selected.manifest, runner: input.runner, cwd: input.root, targetPath: input.targetPath,
+    reuseRoots: [], verifyDirectory: () => { throw Error("current_directory_fallback_forbidden"); },
+    stageDirectory: async () => { throw Error("current_directory_fallback_forbidden"); }, verifyUploaded: async () => {}, ...extra } as Parameters<typeof uploadVerifiedProviderPack>[0]);
+}
+
+it("installs the exact retained pack through the real uploader on an empty host instead of the current image archive", async () => {
+  const input = await fixture();
+  const verifyUploaded = vi.fn(async () => {
+    expect(await fs.readFile(path.join(input.targetPath, "bin/tool"), "utf8")).toBe("retained provider code");
+    expect(JSON.parse(await fs.readFile(path.join(input.targetPath, "provider-pack.json"), "utf8"))).toEqual(input.selected.manifest);
+  });
+  await upload(input, { verifyUploaded });
+  expect(verifyUploaded).toHaveBeenCalledOnce();
+  expect(input.requests.some(route => route.includes("/blobs/"))).toBe(true);
+});
+
+it("keeps a matching current pack on its existing path without registry or cache reads", async () => {
+  const input = await fixture();
+  await upload(input, { root: input.selected.pack });
+  expect(input.requests).toEqual([]);
+  await expect(fs.stat(path.join(input.temporary, `paperclip-provider-pack-sources-${process.getuid?.()}`))).rejects.toMatchObject({ code: "ENOENT" });
+  expect(await fs.readFile(path.join(input.targetPath, "bin/tool"), "utf8")).toBe("retained provider code");
+});
+
+it("refuses a different historical code pack even when OCI source labels and all transport hashes are valid", async () => {
+  const input = await fixture({ wrongCode: true });
+  await expect(upload(input)).rejects.toThrow("runner_remote_provider_archive_invalid");
+  expect(input.runner.execute).not.toHaveBeenCalled();
+  expect(input.runner.syncIn).not.toHaveBeenCalled();
+  expect(await fs.readFile(path.join(input.current.pack, "bin/tool"), "utf8")).toBe("current incompatible code");
+});
+
+it("does not mask a corrupt matching current archive by looking up a different code source", async () => {
+  const input = await fixture();
+  await fs.writeFile(input.selected.pack + ".archive.json", "{");
+  await expect(upload(input, { root: input.selected.pack })).rejects.toThrow("runner_remote_provider_archive_invalid");
+  expect(input.requests).toEqual([]);
+  expect(input.runner.execute).not.toHaveBeenCalled();
+  expect(input.runner.syncIn).not.toHaveBeenCalled();
+});
+
+it("refuses an unknown historical source without clearing or falling back to the current image", async () => {
+  const input = await fixture();
+  input.routes.delete(`/v2/paperclipai/paperclip/manifests/sha-${input.selected.manifest.payload.runnerSourceRevision}`);
+  await expect(upload(input)).rejects.toThrow("runner_remote_provider_source_unavailable: HTTP 404");
+  expect(input.runner.execute).not.toHaveBeenCalled();
+  expect(input.runner.syncIn).not.toHaveBeenCalled();
+});
+
+it("rejects escaping inner tar paths despite valid OCI and layer digests before any remote effects", async () => {
+  const input = await fixture({ unsafeOverlay: true });
+  await expect(upload(input)).rejects.toThrow("runner_remote_provider_source_invalid");
+  expect(input.runner.execute).not.toHaveBeenCalled();
+  expect(input.runner.syncIn).not.toHaveBeenCalled();
+  await expect(fs.stat(path.join(input.temporary, "outside"))).rejects.toMatchObject({ code: "ENOENT" });
+});
+
+it("checks the file layer's actual plaintext against its declared code digest before any remote effects", async () => {
+  const input = await fixture({ changedFileLayer: true });
+  await expect(upload(input)).rejects.toThrow("runner_remote_provider_source_invalid");
+  expect(input.runner.execute).not.toHaveBeenCalled();
+  expect(input.runner.syncIn).not.toHaveBeenCalled();
+});
+
+it("preserves historical code hardlinks only to earlier confined regular files", async () => {
+  const input = await fixture({ hardlink: "safe" });
+  await upload(input);
+  const original = await fs.stat(path.join(input.targetPath, "bin/tool")), copy = await fs.stat(path.join(input.targetPath, "bin/tool-copy"));
+  expect(copy.ino).toBe(original.ino);
+  expect(await fs.readFile(path.join(input.targetPath, "bin/tool-copy"), "utf8")).toBe("retained provider code");
+});
+
+it.each(["escape", "unknown"] as const)("refuses a historical code hardlink with an %s target before remote effects", async hardlink => {
+  const input = await fixture({ hardlink });
+  await expect(upload(input)).rejects.toThrow("runner_remote_provider_source_invalid");
+  expect(input.runner.execute).not.toHaveBeenCalled();
+  expect(input.runner.syncIn).not.toHaveBeenCalled();
+});
+
+it("cancels a stalled source body and removes only the unfinished private import", async () => {
+  const input = await fixture();
+  input.stall(input.layerRoute);
+  const controller = new AbortController();
+  const operation = upload(input, { signal: controller.signal });
+  const rejected = expect(operation).rejects.toThrow("owned caller stopped");
+  await input.observed(input.layerRoute);
+  controller.abort(new Error("owned caller stopped"));
+  await rejected;
+  expect(input.runner.execute).not.toHaveBeenCalled();
+  expect(input.runner.syncIn).not.toHaveBeenCalled();
+  const cache = path.join(input.temporary, `paperclip-provider-pack-sources-${process.getuid?.()}`);
+  await vi.waitFor(async () => expect(await fs.readdir(cache)).toEqual([]));
+});
+
+it("enforces the source deadline while the registry is stalled", async () => {
+  const input = await fixture();
+  input.stall(input.layerRoute);
+  const nativeTimeout = AbortSignal.timeout;
+  vi.spyOn(AbortSignal, "timeout").mockImplementation(milliseconds => {
+    expect(milliseconds).toBe(180_000);
+    return nativeTimeout(80);
+  });
+  await expect(upload(input)).rejects.toMatchObject({ name: "TimeoutError" });
+  expect(input.runner.execute).not.toHaveBeenCalled();
+  expect(input.runner.syncIn).not.toHaveBeenCalled();
+});
+
+it("reuses a verified derived cache without registry reads and reconstructs it after an empty-host reset", async () => {
+  const input = await fixture();
+  await upload(input);
+  const first = input.requests.length;
+  await upload(input, { targetPath: input.targetPath + "-cached" });
+  expect(input.requests).toHaveLength(first);
+  await fs.rm(path.join(input.temporary, `paperclip-provider-pack-sources-${process.getuid?.()}`), { recursive: true });
+  await upload(input, { targetPath: input.targetPath + "-empty" });
+  expect(input.requests.length).toBeGreaterThan(first);
+  expect(await fs.readFile(path.join(input.targetPath + "-empty", "bin/tool"), "utf8")).toBe("retained provider code");
+});
+
+it("rejects a changed cached archive before any remote command or transfer", async () => {
+  const input = await fixture();
+  await upload(input);
+  const cache = path.join(input.temporary, `paperclip-provider-pack-sources-${process.getuid?.()}`);
+  const publication = JSON.parse(await fs.readFile(path.join(cache, input.selected.manifest.digest.slice(7) + ".json"), "utf8"));
+  await fs.appendFile(path.join(cache, publication.entry, "provider-pack.archives/overlay.tar.zst"), "changed");
+  input.runner.execute.mockClear();
+  input.runner.syncIn.mockClear();
+  await expect(upload(input, { targetPath: input.targetPath + "-new" })).rejects.toThrow("runner_remote_provider_archive_invalid");
+  expect(input.runner.execute).not.toHaveBeenCalled();
+  expect(input.runner.syncIn).not.toHaveBeenCalled();
+  expect(await fs.readFile(path.join(input.targetPath, "bin/tool"), "utf8")).toBe("retained provider code");
+});
+
+it("refuses an unsafe cache directory instead of following it", async () => {
+  const input = await fixture();
+  const outside = path.join(input.root, "outside");
+  await fs.mkdir(outside);
+  await fs.symlink(outside, path.join(input.temporary, `paperclip-provider-pack-sources-${process.getuid?.()}`));
+  await expect(upload(input)).rejects.toThrow("runner_remote_provider_source_invalid");
+  expect(await fs.readdir(outside)).toEqual([]);
+  expect(input.requests).toEqual([]);
+  expect(input.runner.execute).not.toHaveBeenCalled();
+});
+
+it("leaves a target intact when its archive transport cannot install the verified source", async () => {
+  const input = await fixture();
+  await fs.mkdir(input.targetPath);
+  await fs.writeFile(path.join(input.targetPath, "marker"), "untouched");
+  input.runner.execute.mockResolvedValue({ exitCode: 127, timedOut: false, stdout: "", stderr: "unavailable" });
+  await expect(upload(input)).rejects.toThrow("runner_remote_provider_archive_transport_unavailable");
+  expect(input.runner.execute).toHaveBeenCalledOnce();
+  expect(input.runner.execute.mock.calls[0]?.[0].args?.join(" ")).not.toContain("rm -rf");
+  expect(input.runner.syncIn).not.toHaveBeenCalled();
+  expect(await fs.readFile(path.join(input.targetPath, "marker"), "utf8")).toBe("untouched");
+});

server/src/services/native-runtime/retained-provider-pack-source.ts

@@ -0,0 +1,308 @@
+import { createHash, randomUUID } from "node:crypto";
+import { constants, createReadStream } from "node:fs";
+import { link, lstat, mkdir, mkdtemp, open, readdir, realpath, rm, statfs, unlink, writeFile } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { basename, join } from "node:path";
+import { Transform } from "node:stream";
+import { pipeline } from "node:stream/promises";
+import { createZstdDecompress } from "node:zlib";
+import tar from "tar-stream";
+import { canonicalProviderPackJson, type ProviderPackIdentity } from "./remote-provider-pack-verification.js";
+import { readProviderPackArchive } from "./provider-pack-archive.js";
+import { MAX_RETAINED_PACK_BYTES, receiveRetainedProviderPack, type RetainedPackProvenance } from "./retained-provider-pack-oci.js";
+
+type Source = { root: string; archiveOnly: true };
+type ImportJob = { operation: Promise<Source>; controller: AbortController; waiters: number };
+const pending = new Map<string, ImportJob>();
+const waiting = new Set<() => void>();
+let activeImports = 0;
+const digest = (value: Buffer | string) => createHash("sha256").update(value).digest("hex");
+function fail(): never { throw Error("runner_remote_provider_source_invalid"); }
+const maximumMetadata = 1024 * 1024;
+type Published = RetainedPackProvenance & { entry: string; receiptSha256: string; verifiedTarPaths: true };
+const safeName = (value: string) => value.length > 0 && Buffer.byteLength(value) <= 4096 && !/[\\\u0000-\u001f\u007f]/u.test(value)
+  && !value.startsWith("/") && value.split("/").every(part => part !== "" && part !== "." && part !== "..");
+
+async function waitForSource<T>(operation: Promise<T>, signal: AbortSignal): Promise<T> {
+  signal.throwIfAborted();
+  let stopped!: () => void;
+  const aborted = new Promise<never>((_resolve, reject) => {
+    stopped = () => reject(signal.reason);
+    signal.addEventListener("abort", stopped, { once: true });
+    if (signal.aborted) stopped();
+  });
+  try { return await Promise.race([operation, aborted]); }
+  finally { signal.removeEventListener("abort", stopped); }
+}
+
+async function importSlot(signal: AbortSignal) {
+  while (activeImports >= 2) {
+    let resume!: () => void;
+    const available = new Promise<void>(done => { resume = done; waiting.add(done); });
+    try { await waitForSource(available, signal); }
+    finally { waiting.delete(resume); }
+  }
+  signal.throwIfAborted();
+  activeImports++;
+  return () => { activeImports--; for (const resume of waiting) resume(); };
+}
+
+async function readStable(path: string, maximum = maximumMetadata) {
+  const file = await open(path, constants.O_RDONLY | constants.O_NOFOLLOW | constants.O_NONBLOCK);
+  try {
+    const before = await file.stat();
+    if (!before.isFile() || before.size > maximum) fail();
+    const bytes = await file.readFile();
+    const after = await file.stat();
+    if (before.ino !== after.ino || before.dev !== after.dev || before.size !== after.size || before.mtimeMs !== after.mtimeMs || before.ctimeMs !== after.ctimeMs) fail();
+    return bytes;
+  } finally { await file.close(); }
+}
+
+async function regularDigest(path: string, expected: { sha256: string; bytes: number }, signal: AbortSignal) {
+  const file = await open(path, constants.O_RDONLY | constants.O_NOFOLLOW | constants.O_NONBLOCK);
+  try {
+    const before = await file.stat();
+    if (!before.isFile() || before.size !== expected.bytes) fail();
+    const hash = createHash("sha256");
+    let bytes = 0;
+    for await (const chunk of file.createReadStream({ autoClose: false, signal })) { bytes += chunk.length; hash.update(chunk); }
+    const after = await file.stat();
+    if (bytes !== before.size || before.ino !== after.ino || before.dev !== after.dev || before.size !== after.size || before.mtimeMs !== after.mtimeMs || before.ctimeMs !== after.ctimeMs || hash.digest("hex") !== expected.sha256) fail();
+  } finally { await file.close(); }
+}
+
+async function selectedArchive(root: string, manifest: ProviderPackIdentity, signal: AbortSignal) {
+  signal.throwIfAborted();
+  const value = JSON.parse((await readStable(join(root, "provider-pack.json"))).toString()) as ProviderPackIdentity;
+  if (canonicalProviderPackJson(value) !== canonicalProviderPackJson(manifest)) fail();
+  const archive = await readProviderPackArchive(root, manifest.digest);
+  if (!archive || !("kind" in archive)) fail();
+  const names = ["overlay.tar.zst", ...archive.files.map((_file, index) => `${index}.tar.zst`)];
+  if (canonicalProviderPackJson((await readdir(archive.sourceRoot)).sort()) !== canonicalProviderPackJson(names.sort())) fail();
+  let bytes = archive.overlay.bytes;
+  for (const [index, file] of archive.files.entries()) {
+    signal.throwIfAborted();
+    bytes += file.archive.bytes;
+    if (bytes > MAX_RETAINED_PACK_BYTES) fail();
+    await regularDigest(join(archive.sourceRoot, `${index}.tar.zst`), file.archive, signal);
+  }
+  signal.throwIfAborted();
+  return archive;
+}
+
+async function extractManifest(root: string, manifest: ProviderPackIdentity, signal: AbortSignal) {
+  const source = await readProviderPackArchive(root, manifest.digest);
+  if (!source || !("kind" in source)) fail();
+  const regular = new Set(source.files.map(file => file.path));
+  const archive = tar.extract();
+  const names = new Set<string>(), files = new Set<string>(), links = new Map<string, string>();
+  let bytes = 0, count = 0;
+  let found: Buffer | undefined;
+  archive.on("entry", (header, body, next) => {
+    body.on("error", () => undefined);
+    void (async () => {
+      signal.throwIfAborted();
+      if (++count > 100_000) fail();
+      const normalized = header.name.replace(/^\.\//u, "");
+      const name = header.type === "directory" ? normalized.replace(/\/$/u, "") : normalized;
+      if (!safeName(name) || names.has(name) || !["file", "directory", "symlink", "link"].includes(header.type ?? "")
+        || !Number.isSafeInteger(header.size ?? 0) || (header.size ?? 0) < 0 || ((header.mode ?? 0) & ~0o777) !== 0) fail();
+      names.add(name);
+      if (header.type !== "directory") files.add(name);
+      if (header.type === "link") {
+        const target = header.linkname?.replace(/^\.\//u, "");
+        if (!target || !safeName(target) || !regular.has(target)) fail();
+        regular.add(name);
+      } else if (header.type === "file") regular.add(name);
+      if (header.type === "symlink") {
+        if (!header.linkname || header.linkname.startsWith("/") || /[\\\u0000-\u001f\u007f]/u.test(header.linkname) || Buffer.byteLength(header.linkname) > 4096) fail();
+        links.set(name, header.linkname);
+      }
+      if (name !== "provider-pack.json") { body.resume(); return; }
+      if (found || header.type !== "file" || !Number.isSafeInteger(header.size) || header.size! > maximumMetadata) fail();
+      const chunks: Buffer[] = [];
+      for await (const chunk of body) {
+        bytes += chunk.length;
+        if (bytes > maximumMetadata) fail();
+        chunks.push(Buffer.from(chunk));
+      }
+      found = Buffer.concat(chunks);
+    })().then(() => next(), error => archive.destroy(error));
+  });
+  let expanded = 0;
+  const bound = new Transform({ transform(chunk, _encoding, done) {
+    expanded += chunk.length;
+    done(expanded > MAX_RETAINED_PACK_BYTES ? Error("runner_remote_provider_source_size_limit") : null, chunk);
+  } });
+  await pipeline(createReadStream(root + ".archives/overlay.tar.zst"), createZstdDecompress(), bound, archive, { signal });
+  for (const name of names) {
+    let parent = name;
+    while (parent.includes("/")) {
+      parent = parent.slice(0, parent.lastIndexOf("/"));
+      if (files.has(parent)) fail();
+    }
+  }
+  for (const [name, target] of links) {
+    let parts = [...name.split("/").slice(0, -1), ...target.split("/")];
+    for (let followed = 0; ; followed++) {
+      if (followed > 32) fail();
+      const resolved: string[] = [];
+      let changed = false;
+      for (let index = 0; index < parts.length; index++) {
+        const part = parts[index]!;
+        if (part === "" || part === ".") continue;
+        if (part === "..") { if (!resolved.length) fail(); resolved.pop(); continue; }
+        resolved.push(part);
+        const link = links.get(resolved.join("/"));
+        if (link !== undefined) { parts = [...resolved.slice(0, -1), ...link.split("/"), ...parts.slice(index + 1)]; changed = true; break; }
+      }
+      if (!changed) break;
+    }
+  }
+  if (!found || canonicalProviderPackJson(JSON.parse(found.toString())) !== canonicalProviderPackJson(manifest)) fail();
+  await mkdir(root, { mode: 0o700 });
+  await writeFile(join(root, "provider-pack.json"), found, { flag: "wx", mode: 0o600 });
+  return { expanded, files };
+}
+
+async function verifyFileContents(archivePath: string, file: { path: string; bytes: number; sha256: string; mode: number }, signal: AbortSignal) {
+  const archive = tar.extract();
+  let found = false, count = 0;
+  archive.on("entry", (header, body, next) => {
+    body.on("error", () => undefined);
+    void (async () => {
+      signal.throwIfAborted();
+      const normalized = header.name.replace(/^\.\//u, "");
+      const name = header.type === "directory" ? normalized.replace(/\/$/u, "") : normalized;
+      if (++count > 128 || !safeName(name)) fail();
+      if (header.type === "directory" && file.path.startsWith(name + "/")) { body.resume(); return; }
+      if (found || header.type !== "file" || name !== file.path || header.size !== file.bytes || header.mode !== file.mode) fail();
+      found = true;
+      const hash = createHash("sha256");
+      let bytes = 0;
+      for await (const chunk of body) {
+        signal.throwIfAborted();
+        bytes += chunk.length;
+        if (bytes > file.bytes) fail();
+        hash.update(chunk);
+      }
+      if (bytes !== file.bytes || hash.digest("hex") !== file.sha256) fail();
+    })().then(() => next(), error => archive.destroy(error));
+  });
+  let expanded = 0;
+  const bound = new Transform({ transform(chunk, _encoding, done) {
+    expanded += chunk.length;
+    done(expanded > file.bytes + 2 * 1024 * 1024 ? Error("runner_remote_provider_source_size_limit") : null, chunk);
+  } });
+  await pipeline(createReadStream(archivePath), createZstdDecompress(), bound, archive, { signal });
+  if (!found) fail();
+}
+
+async function privateDirectory(root: string) {
+  const metadata = await lstat(root);
+  if (!metadata.isDirectory() || metadata.isSymbolicLink() || metadata.uid !== process.getuid?.() || (metadata.mode & 0o077) !== 0) fail();
+}
+
+async function cached(cache: string, pointer: string, manifest: ProviderPackIdentity, signal: AbortSignal) {
+  let bytes: Buffer;
+  try { bytes = await readStable(pointer); } catch (error) {
+    if ((error as NodeJS.ErrnoException).code === "ENOENT") return null;
+    throw error;
+  }
+  const publication = JSON.parse(bytes.toString()) as Published;
+  if (publication.schema !== "paperclip.retained-provider-pack-source.v1" || publication.verifiedTarPaths !== true || publication.providerPackDigest !== manifest.digest
+    || publication.sourceRevision !== manifest.payload.runnerSourceRevision || !new RegExp(`^${manifest.digest.slice(7)}-[A-Za-z0-9]{6}$`).test(publication.entry)
+    || !/^[a-f0-9]{64}$/.test(publication.receiptSha256)
+    || [publication.imageIndexDigest, publication.imageManifestDigest, publication.imageConfigDigest, publication.transportLayerDigest].some(value => !/^sha256:[a-f0-9]{64}$/.test(value))) fail();
+  const directory = join(cache, publication.entry);
+  await privateDirectory(directory);
+  const root = join(directory, "provider-pack");
+  await privateDirectory(root);
+  if (digest(await readStable(root + ".archive.json")) !== publication.receiptSha256) fail();
+  await selectedArchive(root, manifest, signal);
+  return { root, archiveOnly: true as const };
+}
+
+async function importSource(cache: string, pointer: string, manifest: ProviderPackIdentity, signal: AbortSignal) {
+  const space = await statfs(cache, { bigint: true });
+  if (space.bavail * space.bsize < BigInt(2 * MAX_RETAINED_PACK_BYTES + 1024 ** 3)) throw Error("runner_remote_provider_source_disk_headroom_required");
+  const directory = await mkdtemp(join(cache, manifest.digest.slice(7) + "-"));
+  const temporaryPointer = join(cache, `.${randomUUID()}.pending`);
+  let published = false;
+  try {
+    const provenance = await receiveRetainedProviderPack({ manifest, directory, signal });
+    const root = join(directory, "provider-pack");
+    const overlay = await extractManifest(root, manifest, signal);
+    const archive = await selectedArchive(root, manifest, signal);
+    let expanded = overlay.expanded;
+    for (const [index, file] of archive.files.entries()) {
+      if (!Number.isSafeInteger(file.bytes) || file.bytes < 1 || (expanded += file.bytes) > MAX_RETAINED_PACK_BYTES || overlay.files.has(file.path)) fail();
+      let parent = file.path;
+      while (parent.includes("/")) { parent = parent.slice(0, parent.lastIndexOf("/")); if (overlay.files.has(parent)) fail(); }
+      await verifyFileContents(join(archive.sourceRoot, `${index}.tar.zst`), file, signal);
+    }
+    await unlink(join(directory, "transport.tar.gz"));
+    signal.throwIfAborted();
+    const publication: Published = { ...provenance, entry: basename(directory), receiptSha256: digest(await readStable(root + ".archive.json")), verifiedTarPaths: true };
+    await writeFile(temporaryPointer, JSON.stringify(publication), { flag: "wx", mode: 0o600 });
+    signal.throwIfAborted();
+    try {
+      await link(temporaryPointer, pointer);
+      published = true;
+    } catch (error) {
+      if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error;
+      const winner = await cached(cache, pointer, manifest, signal);
+      if (!winner) fail();
+      return winner;
+    }
+    return { root, archiveOnly: true as const };
+  } finally {
+    await rm(temporaryPointer, { force: true });
+    if (!published) await rm(directory, { recursive: true, force: true });
+  }
+}
+
+export async function resolveProviderPackUploadSource(input: {
+  root: string;
+  manifest: ProviderPackIdentity;
+  signal?: AbortSignal;
+}): Promise<{ root: string; archiveOnly: boolean }> {
+  const manifest = input.manifest;
+  if (manifest?.schema !== "paperclip-runner/remote-provider-pack/v1" || manifest.digest !== `sha256:${digest(canonicalProviderPackJson(manifest.payload))}`) fail();
+  input.signal?.throwIfAborted();
+  const configured = JSON.parse((await readStable(join(input.root, "provider-pack.json"))).toString()) as ProviderPackIdentity;
+  if (configured?.schema !== manifest.schema || configured.digest !== `sha256:${digest(canonicalProviderPackJson(configured.payload))}`) fail();
+  if (configured.digest === manifest.digest) return { root: input.root, archiveOnly: false };
+  const cache = join(await realpath(tmpdir()), `paperclip-provider-pack-sources-${process.getuid?.()}`);
+  await mkdir(cache, { recursive: true, mode: 0o700 });
+  await privateDirectory(cache);
+  const signal = input.signal ? AbortSignal.any([input.signal, AbortSignal.timeout(180_000)]) : AbortSignal.timeout(180_000);
+  const pointer = join(cache, manifest.digest.slice(7) + ".json");
+  const retained = await cached(cache, pointer, manifest, signal);
+  if (retained) return retained;
+  const key = `${cache}:${manifest.digest}`;
+  let job = pending.get(key);
+  if (!job || job.controller.signal.aborted) {
+    if (pending.size >= 32) throw Error("runner_remote_provider_source_import_busy");
+    const controller = new AbortController();
+    const sharedSignal = AbortSignal.any([controller.signal, AbortSignal.timeout(180_000)]);
+    let created!: ImportJob;
+    const operation = (async () => {
+      const release = await importSlot(sharedSignal);
+      try { return await importSource(cache, pointer, manifest, sharedSignal); }
+      finally { release(); }
+    })().finally(() => { if (pending.get(key) === created) pending.delete(key); });
+    job = created = { operation, controller, waiters: 0 };
+    pending.set(key, job);
+  }
+  job.waiters++;
+  try {
+    const source = await waitForSource(job.operation, signal);
+    signal.throwIfAborted();
+    return source;
+  } finally {
+    job.waiters--;
+    if (job.waiters === 0 && pending.get(key) === job) job.controller.abort(new DOMException("provider source no longer requested", "AbortError"));
+  }
+}