fix(runner): drain Pi snapshot acquisition through shutdown

Cancel and drain pending native snapshot copies through the command owner. Give Pi a finite cleanup grace before process-group termination, preserving lifetime fences and exact-child retirement. Retain the original canonical Stop pass with failed independent cleanup and add free regressions for both boundaries.

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

doc/plans/2026-10-02-pi-production-readiness.md

@@ -40,13 +40,43 @@ old code and passes with the correction; live-child and cleanup-failure regressi
 also pass. No paid cases resume before rebuilding and admitting this repair.
 The three passes on the previous Runner source do not qualify changed host bytes.
 
+Source `b061f31242a29374910b5c02b595e90d21e7fae9` then passed normal Linux
+workspace/public builds, all installation hooks, graph audits, Pi setup and real
+server/UI startup. All 20 archives are hash-verified and retained. Its hosted image
+build, anonymous OCI admission and actual credential-free Daytona guest probe
+passed; probe deletion was confirmed. Current-head CI passed 54 applicable checks,
+with two expected skips and zero new Greptile comments or open review threads.
+
+The first Stop run on that source passed its unchanged canonical Product grader,
+including process cleanup. Independent inspection nevertheless found two native
+snapshot directories: one complete 547 MB distribution and one partial 20 MB
+copy. The original canonical pass, failed independent cleanup observation, trace,
+manifest and every remaining file hash are retained. Paid cases halted; this run
+does not qualify the release. Only those two exact, quiescent owned directories
+were reclaimed after rechecking their retained identities and bytes.
+
+Free regressions reproduce both remaining boundaries: native acquisition ignored
+cancellation, and the Rust process-group supervisor killed Pi cleanup after only
+two seconds. The new repair propagates a command-owner cancellation signal into
+Pi snapshot acquisition, stops admitting copies on cancellation and drains all
+admitted reads before deleting the partial tree. Pi gets a finite 30-second
+sidecar shutdown grace; process-group KILL and inherited lifetime-fence proofs
+remain authoritative. Normal cleanup still waits for exact-child exit.
+The repair passes 271 focused TypeScript tests, one existing skip, all 16 Rust
+transport tests and Runner typecheck. A credential-free Linux probe against the actual installed 15,671-entry, 540 MB
+Pi closure takes 4.22 seconds to copy and 1.38 seconds to delete. Cancellation
+during an admitted read drains and deletes the partial copy in 0.28 seconds; no
+snapshots remain. This free repair-engine proof does not qualify packaged Runner
+bytes. The source must be rebuilt and admitted before another paid attempt. All
+33 release cases remain required.
+
 | Release gate | Current evidence |
 | --- | --- |
 | Offline cancellation proof | Unit and actual subprocess regressions fail on profile-17 bytes and pass on the profile-18 wrapper. All 49 Pi ACP protocol tests pass. Historical profile-17 decoding remains available; exact launch requires profile 18. |
 | Snapshot retirement proof | New actual-subprocess tests prove deletion completes before retirement, live-child bytes remain available, failed deletion is reported and retryable, and a surviving child returns bounded failure with retained bytes. All 150 focused tests pass, six existing skips; Runner TypeScript typecheck passes. The earlier parallel credential-home collision and passing 52-test isolated recheck remain retained. |
 | Existing free checks | The cancellation source passes 2,480 Product unit tests; the final source rerun has one unrelated file-gate timeout, whose seven-test isolated recheck passes. Exact-source Rust checks pass 689 tests, two ignored. Latest Linux CI on `9cd363d` passes all 54 applicable checks. Fresh cleanup-source CI is required. |
 | Normal Linux build and public installation | Source `6a3930934` passes the normal build/install/hook/graph/setup/server/UI chain. Fresh artifacts containing the snapshot-retirement repair must pass the same admission. |
-| Final Product acceptance | Fresh repair-source coverage is 0/26. Retain source `6a3930934`'s three passes and original disk-exhaustion failure as historical evidence. Keep every accepted assertion and zero automatic retries. |
+| Final Product acceptance | Fresh repair-source coverage is 0/26. Retain source `6a3930934`'s three passes and disk-exhaustion failure, and `b061f3124`'s canonical Stop pass with failed independent native-snapshot cleanup, as historical evidence. Keep every accepted assertion and zero automatic retries. |
 | Final Runner acceptance | Fresh repair-source coverage is 0/7. All seven definitions and scoring rules stay frozen. Only their shipping-source metadata changes when the next artifact set is frozen. |
 | Cloud image | [Hosted Linux build](https://github.com/paperclipai/paperclip/actions/runs/37662254178) passes for source `6a3930934`, immutable image `sha256:c05914a36034c3a9a1fee2e385ca67c28ee4f98fd058c004b9a9a68c76bbf976`. Anonymous OCI and actual credential-free Daytona guest admission pass; owned probe deletion is confirmed. A fresh repair-source image and guest admission are required. No local Docker is used. |
 | PR CI and review | `9cd363d` passes 54 checks with no failures; Greptile is 5/5 with no open threads. Fresh cleanup-source CI and review are required. No merge, release or rollout is authorized. |
@@ -59,8 +89,8 @@ remain strict. Task creation binds the explicit public creation-response issue I
 
 The dedicated OpenRouter key retains its $5 lifetime cap inside the approved $100
 total token ceiling, with no reset, BYOK, account-key fallback or unchanged paid
-retry. The last retained key snapshot records $0.478199225 lifetime usage and
-$4.521800775 remaining. Billing snapshots are provisional. Reclaim only inspected,
+retry. The pre-Stop key snapshot records $0.491592211 lifetime usage and
+$4.508407789 remaining. Billing snapshots are provisional. Reclaim only inspected,
 quiescent resources from the owned qualification host after retaining their evidence;
 require disk headroom and zero leaked native snapshots before subsequent paid cases.
 

packages/paperclip-runner/runner/crates/runner-core/src/acpx_sidecar_transport.rs

@@ -276,7 +276,15 @@ impl AcpxSidecarTransport {
         };
         keys.extend(pi_keys.iter().map(String::as_str));
         keys.extend_from_slice(credential_keys);
-        let mut transport = Self::start_with_environment_keys(config, &keys)?;
+        // Pi owns a native distribution copy, including a pending refresh at
+        // suspension. Allow its bounded cleanup to settle before group KILL;
+        // the ordinary two-second grace can cut off deletion mid-tree.
+        config.validate()?;
+        let mut launch_config = config.clone();
+        if agent == "pi" {
+            launch_config.shutdown_grace = Duration::from_secs(30);
+        }
+        let mut transport = Self::start_with_environment_keys(&launch_config, &keys)?;
         transport.session_open_timeout = session_open_timeout(agent, config.request_timeout);
         Ok(transport)
     }

packages/paperclip-runner/runner/crates/runner-core/tests/acpx_sidecar_transport.rs

@@ -134,6 +134,55 @@ fn an_empty_event_poll_does_not_poison_the_transport() {
     transport.shutdown().expect("fake sidecar should stop");
 }
 
+#[cfg(unix)]
+#[test]
+fn pi_shutdown_allows_owned_snapshot_cleanup_after_the_ordinary_two_second_grace() {
+    let nonce = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .unwrap()
+        .as_nanos();
+    let marker =
+        std::env::temp_dir().join(format!("pi-sidecar-cleanup-{}-{nonce}", std::process::id()));
+    let ready = marker.with_extension("ready");
+    std::fs::write(&marker, b"owned snapshot cleanup pending").unwrap();
+    // TERM is delivered to the same owned process group as in suspension.
+    // Cleanup deliberately takes longer than the ordinary two-second grace.
+    let script =
+        "trap 'sleep 3; rm -- \"$1\"; exit 0' TERM; echo ready > \"$2\"; while :; do sleep 1; done";
+    let config = AcpxSidecarTransportConfig {
+        command: PathBuf::from("/bin/sh"),
+        args: vec![
+            "-c".into(),
+            script.into(),
+            "pi-cleanup-fixture".into(),
+            marker.to_string_lossy().into_owned(),
+            ready.to_string_lossy().into_owned(),
+        ],
+        verified_launch: None,
+        request_timeout: Duration::from_secs(30),
+        shutdown_grace: Duration::from_secs(2),
+    };
+    let mut transport = AcpxSidecarTransport::start_for_agent(&config, "pi").unwrap();
+    let startup = Instant::now();
+    while !ready.exists() && startup.elapsed() < Duration::from_secs(2) {
+        std::thread::sleep(Duration::from_millis(5));
+    }
+    assert!(
+        ready.exists(),
+        "fixture did not install its TERM cleanup handler"
+    );
+    let started = Instant::now();
+    transport.shutdown().unwrap();
+    let removed = !marker.exists();
+    let _ = std::fs::remove_file(&marker);
+    let _ = std::fs::remove_file(&ready);
+    assert!(
+        removed,
+        "Pi sidecar was killed before owned snapshot cleanup completed"
+    );
+    assert!(started.elapsed() < Duration::from_secs(10));
+}
+
 #[test]
 fn rejects_an_unbounded_event_poll_without_poisoning_the_transport() {
     let mut transport = transport("silent", Duration::from_secs(1));

packages/paperclip-runner/src/drivers/acpx/command-lease-owner.test.ts

@@ -76,4 +76,33 @@ describe("ACPX verified command lease owner", () => {
     await owner.command.close();
     expect(initial.close).toHaveBeenCalledOnce();
   });
+
+  it("cancels a pending refresh and joins its partial-copy cleanup before retiring ownership", async () => {
+    const initial = lease();
+    let signal!: AbortSignal;
+    let entered!: () => void; let drained!: () => void;
+    const acquiring = new Promise<void>(resolve => { entered = resolve; });
+    const cleanup = new Promise<void>(resolve => { drained = resolve; });
+    const owner = createAcpxCommandLeaseOwner(initial, async activeSignal => {
+      signal = activeSignal; entered();
+      await new Promise<void>(resolve => activeSignal.addEventListener("abort", () => resolve(), { once: true }));
+      await cleanup;
+      activeSignal.throwIfAborted();
+      throw new Error("cancelled acquisition cannot return a lease");
+    });
+    owner.command.spawn();
+    const refresh = owner.refreshConsumedCommand();
+    const rejectedRefresh = expect(refresh).rejects.toThrow("owner is closing");
+    await acquiring;
+    let retired = false;
+    const close = owner.command.close().then(() => { retired = true; });
+    expect(signal.aborted).toBe(true);
+    await Promise.resolve();
+    expect(retired).toBe(false);
+    expect(initial.close).not.toHaveBeenCalled();
+    drained();
+    await rejectedRefresh; await close;
+    expect(retired).toBe(true);
+    expect(initial.close).toHaveBeenCalledOnce();
+  });
 });

packages/paperclip-runner/src/drivers/acpx/command-lease-owner.ts

@@ -3,13 +3,14 @@ import type { VerifiedAcpxCommandLease } from "./installation-integrity.js";
 /** Keep each launch single-use while owning replacements for transient ACP controls. */
 export function createAcpxCommandLeaseOwner(
   initial: VerifiedAcpxCommandLease,
-  openCommand: () => Promise<VerifiedAcpxCommandLease>,
+  openCommand: (signal: AbortSignal) => Promise<VerifiedAcpxCommandLease>,
 ) {
   const leases = new Set([initial]);
   let current = initial;
   let consumed = false;
   let closing = false;
   let refresh: Promise<void> | null = null;
+  const admission = new AbortController();
   const command: VerifiedAcpxCommandLease = {
     spawn(...args) {
       if (closing) throw new Error("Verified ACPX command owner is closing");
@@ -18,6 +19,7 @@ export function createAcpxCommandLeaseOwner(
     },
     async close() {
       closing = true;
+      admission.abort(new Error("Verified ACPX command owner is closing"));
       // Late acquisitions remain owned. Retry every lease whose close fails.
       await refresh?.catch(() => undefined);
       const failures: unknown[] = [];
@@ -39,7 +41,7 @@ export function createAcpxCommandLeaseOwner(
       if (!consumed) return;
       if (!refresh) {
         refresh = Promise.resolve()
-          .then(openCommand)
+          .then(() => openCommand(admission.signal))
           .then((replacement) => {
             leases.add(replacement);
             if (closing) throw new Error("Verified ACPX command owner closed during refresh");

packages/paperclip-runner/src/drivers/acpx/installation-integrity.ts

@@ -409,7 +409,7 @@ export interface VerifiedAcpxInstallation {
   readonly commandDigest: string;
   readonly agentServerPackageJsonPath: string | null;
   readonly agentRuntimePackageJsonPath: string | null;
-  openCommand(): Promise<VerifiedAcpxCommandLease>;
+  openCommand(options?: { signal?: AbortSignal }): Promise<VerifiedAcpxCommandLease>;
 }
 
 /**
@@ -428,8 +428,8 @@ export async function verifyNativeAcpxInstallation(
     commandDigest: `sha256:${declaration.expectedClosureSha256}`,
     agentServerPackageJsonPath: declaration.manifestPath,
     agentRuntimePackageJsonPath: null,
-    async openCommand(): Promise<VerifiedAcpxCommandLease> {
-      const native = await createNativeAcpxDistributionSnapshot(declaration, entries);
+    async openCommand(options?: { signal?: AbortSignal }): Promise<VerifiedAcpxCommandLease> {
+      const native = await createNativeAcpxDistributionSnapshot(declaration, entries, options?.signal);
       const lease = commandLease(
         native.snapshot.roots[0]!, NATIVE_ACPX_BOOTSTRAP_NAME, "commonjs",
         native.bootstrap, native.commandDirectory, [], 0, "commonjs", [],

packages/paperclip-runner/src/drivers/acpx/native-distribution-integrity.test.ts

@@ -552,6 +552,36 @@ describe("native ACPX execution closure", () => {
     await new Promise<void>(resolve => setImmediate(resolve));
     await expect(stat(removals[0]!)).rejects.toMatchObject({ code: "ENOENT" });
   });
+  it("aborts a pending native copy, drains admitted reads and removes its partial snapshot", async () => {
+    const { declaration, entries } = await manyFileFixture(Array.from({ length: 40 }, (_, index) => 91 + index));
+    const prototype = await filePrototype(join(declaration.distributionRoot, "runtime"));
+    const originalRead = prototype.read;
+    const entered = gate(); const hold = gate(); const controller = new AbortController();
+    const readSizes: number[] = []; const removalStart = vi.mocked(rm).mock.calls.length;
+    vi.spyOn(prototype, "read").mockImplementation(async function (this: FileHandle, ...args: any[]): Promise<any> {
+      readSizes.push(args[0].length); entered.release(); await hold.promise;
+      return originalRead.apply(this, args as never);
+    });
+    const creating = createNativeAcpxDistributionSnapshot(declaration, entries, controller.signal);
+    let settled = false; let unexpected: Awaited<typeof creating> | undefined;
+    void creating.then(value => { settled = true; unexpected = value; }, () => { settled = true; });
+    try {
+      await entered.promise;
+      controller.abort(new Error("owned command refresh cancelled"));
+      await new Promise<void>(resolve => setImmediate(resolve));
+      expect(settled).toBe(false);
+      expect(vi.mocked(rm).mock.calls).toHaveLength(removalStart);
+      hold.release();
+      await expect(creating).rejects.toThrow("owned command refresh cancelled");
+      expect(readSizes).not.toContain(123);
+      const removals = vi.mocked(rm).mock.calls.slice(removalStart).map(([path]) => String(path)).filter(path => /paperclip-acpx-native-/.test(path));
+      expect(removals).toHaveLength(1);
+      await expect(stat(removals[0]!)).rejects.toMatchObject({ code: "ENOENT" });
+    } finally {
+      hold.release(); await creating.catch(() => undefined);
+      if (unexpected) { await unexpected.commandDirectory.close(); await unexpected.snapshot.close(); }
+    }
+  });
   it("rejects a source mutation while another file is being copied", async () => {
     const { declaration, entries } = await manyFileFixture([101, 102]);
     const prototype = await filePrototype(join(declaration.distributionRoot, "runtime"));

packages/paperclip-runner/src/drivers/acpx/native-distribution-integrity.ts

@@ -81,7 +81,8 @@ export async function readNativeAcpxDistributionEntries(input: NativeAcpxDistrib
  * Freeze only manifest-admitted bytes. Native trees deliberately have a separate
  * bound; the tighter JavaScript/npm snapshot limits remain unchanged.
  */
-export async function createNativeAcpxDistributionSnapshot(input: NativeAcpxDistributionInput, entries: NativeAcpxDistributionEntry[]): Promise<NativeAcpxDistributionSnapshot> {
+export async function createNativeAcpxDistributionSnapshot(input: NativeAcpxDistributionInput, entries: NativeAcpxDistributionEntry[], signal?: AbortSignal): Promise<NativeAcpxDistributionSnapshot> {
+  signal?.throwIfAborted();
   // Callers cannot substitute entries between manifest validation and copying.
   entries = parseNativeAcpxDistributionEntries({ entries }, input.expectedClosureSha256);
   if (process.platform !== "linux" && process.platform !== "darwin") throw new Error("Native ACPX snapshots require Linux or macOS");
@@ -100,6 +101,7 @@ export async function createNativeAcpxDistributionSnapshot(input: NativeAcpxDist
   };
   let commandDirectory: FileHandle | undefined;
   try {
+    signal?.throwIfAborted();
     if (!same(rootBefore, await heldRoot.stat({ bigint: true }))) throw new Error("Native ACPX distribution root changed before snapshot");
     await mkdir(packageRoot, { mode: 0o700 });
     if (input.isolatedCacheEnvironmentName) await mkdir(cacheRoot, { mode: 0o700 });
@@ -125,15 +127,18 @@ export async function createNativeAcpxDistributionSnapshot(input: NativeAcpxDist
     }
     const directoryBatch = async (paths: string[], operation: (path: string) => Promise<unknown>): Promise<void> => {
       for (let start = 0; start < paths.length; start += NATIVE_DIRECTORY_CONCURRENCY) {
+        signal?.throwIfAborted();
         const settled = await Promise.allSettled(paths.slice(start, start + NATIVE_DIRECTORY_CONCURRENCY).map(operation));
         const failed = settled.find(result => result.status === "rejected");
         if (failed?.status === "rejected") throw failed.reason;
+        signal?.throwIfAborted();
       }
     };
     for (const depth of [...parentLevels.keys()].sort((a, b) => a - b)) {
       await directoryBatch([...parentLevels.get(depth)!].sort(), path => mkdir(path, { mode: 0o700 }));
     }
     const copyEntry = async (entry: NativeAcpxDistributionEntry): Promise<void> => {
+      signal?.throwIfAborted();
       const path = join(source, ...entry.path.split("/"));
       if (await realpath(path) !== path) throw new Error("Native ACPX closure contains a symbolic link");
       // Bind metadata to the descriptor we will read, without a redundant
@@ -146,6 +151,7 @@ export async function createNativeAcpxDistributionSnapshot(input: NativeAcpxDist
         const bytes = Buffer.alloc(entry.size);
         let offset = 0;
         while (offset < bytes.length) {
+          signal?.throwIfAborted();
           const read = await file.read(bytes, offset, bytes.length - offset, offset);
           if (read.bytesRead === 0) throw new Error("Native ACPX closure file ended during snapshot");
           offset += read.bytesRead;
@@ -153,6 +159,7 @@ export async function createNativeAcpxDistributionSnapshot(input: NativeAcpxDist
         if (!same(before, await file.stat({ bigint: true })) || !same(before, await lstat(path, { bigint: true })) || await realpath(path) !== path) throw new Error("Native ACPX closure file changed while read");
         if (sha256(bytes) !== entry.sha256) throw new Error(`Native ACPX closure file digest mismatch: ${entry.path}`);
         const target = join(packageRoot, ...entry.path.split("/"));
+        signal?.throwIfAborted();
         await writeFile(target, bytes, { mode: entry.executable ? 0o500 : 0o400, flag: "wx" });
       } finally { await file.close(); }
     };
@@ -168,9 +175,14 @@ export async function createNativeAcpxDistributionSnapshot(input: NativeAcpxDist
     // copy on each admission (including long-lived large-file reads).
     let capacityAvailable: (() => void) | undefined;
     for (const entry of entries) {
+      // An aborted acquisition must still drain its admitted reads before
+      // deleting the partial tree. Stop scheduling copies at the same bound
+      // used for any observed copy failure.
+      if (signal?.aborted && !failed) { failed = true; failure = signal.reason; }
       while (!failed && (active.size >= NATIVE_COPY_CONCURRENCY
         || (active.size > 0 && activeBytes + entry.size > NATIVE_COPY_BUFFER_BYTES))) {
         await new Promise<void>(resolve => { capacityAvailable = resolve; });
+        if (signal?.aborted && !failed) { failed = true; failure = signal.reason; }
       }
       if (failed) break;
       activeBytes += entry.size;
@@ -187,6 +199,7 @@ export async function createNativeAcpxDistributionSnapshot(input: NativeAcpxDist
     }
     await Promise.all(active);
     if (failed) throw failure;
+    signal?.throwIfAborted();
     // Completion order must not change the module guard or manifest.
     for (const entry of entries) digests[join(packageRoot, ...entry.path.split("/"))] = entry.sha256;
     if (!same(rootBefore, await heldRoot.stat({ bigint: true })) || !same(rootBefore, await lstat(source, { bigint: true }))) throw new Error("Native ACPX distribution root changed during snapshot");

packages/paperclip-runner/src/drivers/acpx/pi-installation.ts

@@ -91,12 +91,13 @@ export async function verifyPiInstallation(profile: AcpxReleaseProfile): Promise
     commandDigest: profile.commandDigest,
     agentServerPackageJsonPath: join(runtimeRoot, "node_modules/pi-acp/package.json"),
     agentRuntimePackageJsonPath: join(runtimeRoot, "node_modules/@earendil-works/pi-coding-agent/package.json"),
-    async openCommand() {
+    async openCommand(options?: { signal?: AbortSignal }) {
+      options?.signal?.throwIfAborted();
       await assertDirectories();
       // The native primitive reads every admitted file through held descriptors
       // into a new immutable snapshot. The bootstrap derives its own launch
       // environment from that snapshot, never these mutable installation paths.
-      return native.openCommand();
+      return native.openCommand(options);
     },
   });
 }

packages/paperclip-runner/src/drivers/acpx/runtime-host.ts

@@ -536,7 +536,7 @@ export class AcpxRuntimeHost {
       }
       command = await acquireAbortableAdmissionResource({
         signal: options.signal,
-        acquire: () => installation.openCommand(),
+        acquire: () => installation.openCommand({ signal: options.signal }),
         resource: "command",
         releaseLate: (lateCommand) => lateCommand.close(),
         reportFailure: (failure) =>
@@ -544,7 +544,7 @@ export class AcpxRuntimeHost {
       });
       const commandOwner = createAcpxCommandLeaseOwner(
         command,
-        () => installation.openCommand(),
+        signal => installation.openCommand({ signal }),
       );
       command = commandOwner.command;
       toolBridge = options.semanticTools