fix(runner): fence warm directory successor ownership and recovery

Retain stable projectless scope, transfer collection to the exact successor lease, and preserve stopped remote files without restarting their sandbox. Validate complete remote directories and cover the composed DB/session lifecycle, retirement, and fresh preparation.

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

server/src/services/agent-directory-probe.ts

@@ -15,7 +15,7 @@ export function agentDirectoryBaselineDigest(snapshot: DirectorySnapshot): strin
 // Keep all dependencies explicit so remote observation never reads a local mirror.
 export function probeAgentDirectory(
   root: string,
-  modules = { fs, path, createHash },
+  modules = { fs, path, createHash, platform: process.platform },
 ): { digest: string; identity: string } {
   const { fs, path, createHash } = modules;
   const deadline = Date.now() + 5_000;
@@ -23,7 +23,14 @@ export function probeAgentDirectory(
   const stamp = (s: fs.BigIntStats) => [s.dev, s.ino, s.mode, s.nlink, s.size, s.mtimeNs, s.ctimeNs].join(":");
   const identity = (s: fs.BigIntStats) => [s.dev, s.ino, s.birthtimeNs].join(":");
   if (!path.isAbsolute(root) || path.resolve(root) !== root || root.includes("\0")) throw new Error("Invalid agent directory root");
-  // Fixed macOS system aliases are resolved by the caller, not followed here.
+  // Resolve only the observing host's fixed Darwin aliases. The exact full
+  // counterpart excludes nested/user symlinks; canonical ancestors remain strict.
+  const requestedRoot = root;
+  if (modules.platform === "darwin" && /^\/(tmp|var|etc)\//.test(root)) {
+    const resolved = fs.realpathSync(root);
+    if (resolved !== `/private${root}`) throw new Error("Unsafe agent directory alias");
+    root = resolved;
+  }
   const ancestors = new Map<string, string>();
   let ancestor = path.parse(root).root;
   for (const segment of root.slice(ancestor.length).split(path.sep)) {
@@ -94,11 +101,53 @@ export function probeAgentDirectory(
     const current = fs.lstatSync(ancestor, { bigint: true });
     if (!current.isDirectory() || current.isSymbolicLink() || identity(current) !== expected) throw new Error("Agent directory ancestor changed");
   }
+  if (requestedRoot !== root && fs.realpathSync(requestedRoot) !== root) throw new Error("Agent directory alias changed");
   entries.sort(([a], [b]) => a < b ? -1 : a > b ? 1 : 0);
   return { digest: createHash("sha256").update(JSON.stringify(entries)).digest("hex"), identity: identity(initial) };
 }
 
-export const agentDirectoryProbeProgram = `const probe=${probeAgentDirectory.toString()}; process.stdout.write(JSON.stringify(probe(process.argv[1], {fs:require('node:fs'),path:require('node:path'),createHash:require('node:crypto').createHash})))`;
+export const agentDirectoryProbeProgram = `const probe=${probeAgentDirectory.toString()}; process.stdout.write(JSON.stringify(probe(process.argv[1], {fs:require('node:fs'),path:require('node:path'),createHash:require('node:crypto').createHash,platform:process.platform})))`;
+
+/** Retire only the two empty directories created by initial sandbox transfer.
+ * Run before provider admission; this is never an exclusion from live probing. */
+export function retireAgentDirectoryTransferScratch(root: string, modules = { fs, path, platform: process.platform }): void {
+  const { fs, path } = modules;
+  if (!path.isAbsolute(root) || path.resolve(root) !== root || root.includes("\0")) throw new Error("Invalid agent directory root");
+  const requestedRoot = root;
+  if (modules.platform === "darwin" && /^\/(tmp|var|etc)\//.test(root)) {
+    const resolved = fs.realpathSync(root);
+    if (resolved !== `/private${root}`) throw new Error("Unsafe agent directory alias");
+    root = resolved;
+  }
+  const identities = new Map<string, string>();
+  const identify = (directory: string) => {
+    const stat = fs.lstatSync(directory, { bigint: true });
+    if (!stat.isDirectory() || stat.isSymbolicLink()) throw new Error("Unsafe transfer scratch directory");
+    return [stat.dev, stat.ino, stat.birthtimeNs].join(":");
+  };
+  let ancestor = path.parse(root).root;
+  for (const segment of root.slice(ancestor.length).split(path.sep)) {
+    ancestor = path.join(ancestor, segment); identities.set(ancestor, identify(ancestor));
+  }
+  const parent = path.join(root, ".paperclip-runtime"), scratch = path.join(parent, "agent-files");
+  identities.set(parent, identify(parent)); identities.set(scratch, identify(scratch));
+  const verify = () => {
+    for (const [directory, expected] of identities) if (identify(directory) !== expected) throw new Error("Transfer scratch identity changed");
+    if (requestedRoot !== root && fs.realpathSync(requestedRoot) !== root) throw new Error("Agent directory alias changed");
+  };
+  const entries = fs.readdirSync(parent);
+  if (entries.length !== 1 || entries[0] !== "agent-files" || fs.readdirSync(scratch).length) throw new Error("Unexpected transfer scratch contents");
+  verify();
+  // rmdir atomically refuses nonempty directories. Never recursively delete or
+  // follow a link; any change stops preparation before the provider can start.
+  fs.rmdirSync(scratch); identities.delete(scratch);
+  verify();
+  if (fs.readdirSync(parent).length) throw new Error("Transfer scratch changed during retirement");
+  fs.rmdirSync(parent); identities.delete(parent);
+  verify();
+}
+
+export const agentDirectoryTransferCleanupProgram = `const retire=${retireAgentDirectoryTransferScratch.toString()}; retire(process.argv[1], {fs:require('node:fs'),path:require('node:path'),platform:process.platform})`;
 
 /** Bound child: no shell, inherited credentials, or event-loop hashing. */
 export async function observeLocalAgentDirectory(root: string) {

server/src/services/agent-directory-working-copies.ts

@@ -1,6 +1,6 @@
 import fs from "node:fs/promises";
 import path from "node:path";
-import { and, eq, isNull, sql } from "drizzle-orm";
+import { and, eq, inArray, isNull, sql } from "drizzle-orm";
 import { agents, environmentLeases, environments, agentInstructionWorkingCopies as copies, type Db } from "@paperclipai/db";
 import { syncDirectoryToSsh, restoreWorkspaceFromSshExecution } from "@paperclipai/adapter-utils/ssh";
 import { prepareAdapterExecutionTargetRuntime, runAdapterExecutionTargetShellCommand, type AdapterExecutionTarget, type PreparedAdapterExecutionTargetRuntime } from "@paperclipai/adapter-utils/execution-target";
@@ -13,7 +13,7 @@ import { HttpError, conflict, notFound } from "../errors.js";
 import type { AuthorizationActor } from "./authorization.js";
 import type { EnvironmentRuntimeService } from "./environment-runtime.js";
 import type { Environment, EnvironmentLease } from "@paperclipai/shared";
-import { agentDirectoryBaselineDigest, agentDirectoryProbeProgram, observeLocalAgentDirectory } from "./agent-directory-probe.js";
+import { agentDirectoryBaselineDigest, agentDirectoryProbeProgram, agentDirectoryTransferCleanupProgram, observeLocalAgentDirectory } from "./agent-directory-probe.js";
 import { hasRemoteTerminationReceipt } from "./remote-execution-termination.js";
 
 type Copy = typeof copies.$inferSelect;
@@ -35,11 +35,10 @@ export function agentDirectoryWorkingCopyService(db: Db, get: (companyId: string
   const store = agentFileStore(db);
   async function transport(row: Copy, target: AdapterExecutionTarget, recovering: boolean) {
     if (target.kind !== "remote" || row.location !== `remote:${target.environmentId ?? ""}`) throw new Error("Agent directory environment changed");
-    if (recovering && target.transport === "ssh") throw new Error("The original SSH collector is unavailable");
     if (target.transport === "ssh") {
       // Reuse SSH's plain directory transfer without its task-workspace suffix
       // or Git-history discovery. The registered root is the exact writable root.
-      await syncDirectoryToSsh({ spec: target.spec, localDir: row.localRoot, remoteDir: row.executionRoot, exclude: [".paperclip-runtime"] });
+      if (!recovering) await syncDirectoryToSsh({ spec: target.spec, localDir: row.localRoot, remoteDir: row.executionRoot, exclude: [".paperclip-runtime"] });
       return { target, workspaceRemoteDir: row.executionRoot, runtimeRootDir: null,
         assetDirs: {}, additionalSourceDirs: {}, additionalSourceFailures: [], workspaceSyncSnapshot: null,
         restoreWorkspace: () => restoreWorkspaceFromSshExecution({ spec: target.spec, localDir: row.localRoot,
@@ -53,11 +52,8 @@ export function agentDirectoryWorkingCopyService(db: Db, get: (companyId: string
   }
   async function observe(row: Copy, target?: AdapterExecutionTarget | null) {
     if (row.location === "local") {
-      const real = await fs.realpath(row.localRoot);
-      if (real !== row.localRoot && !(process.platform === "darwin" && real === `/private${row.localRoot}` && /^\/(tmp|var|etc)\//.test(row.localRoot))) throw new Error("Agent directory root changed");
-      // A bounded child keeps hashing off the controller event loop and inherits
-      // no provider credentials. execFile terminates it on timeout.
-      return observeLocalAgentDirectory(real);
+      // The same bounded program validates aliases on its actual host.
+      return observeLocalAgentDirectory(row.localRoot);
     }
     if (!target || target.kind !== "remote" || row.location !== `remote:${target.environmentId ?? ""}`) throw new Error("Agent directory environment changed");
     const origin = typeof row.receipt?.materializationRunId === "string" ? row.receipt.materializationRunId : row.runId;
@@ -142,7 +138,15 @@ export function agentDirectoryWorkingCopyService(db: Db, get: (companyId: string
           `node -e ${quote(instructionGitExcludeProgram)} ${quote(input.target.remoteCwd)}`,
           { cwd: input.target.remoteCwd, env: {}, timeoutSec: 15 });
         if (excluded.exitCode !== 0 || excluded.timedOut) throw new Error("Could not exclude agent files from task Git staging");
-        transports.set(key(row), await transport(row, input.target, false));
+        const runtime = await transport(row, input.target, false);
+        transports.set(key(row), runtime);
+        if (input.target.transport === "sandbox") {
+          if (runtime.runtimeRootDir !== path.posix.join(row.executionRoot, ".paperclip-runtime", "agent-files")) throw new Error("Agent directory transfer scratch path changed");
+          const retired = await runAdapterExecutionTargetShellCommand(input.runId, input.target,
+            `node -e ${quote(agentDirectoryTransferCleanupProgram)} ${quote(row.executionRoot)}`,
+            { cwd: input.target.remoteCwd, env: {}, timeoutSec: 10 });
+          if (retired.exitCode !== 0 || retired.timedOut) throw new Error("Agent directory transfer scratch could not be safely retired");
+        }
       }
       const observed = await observe(row, input.target).catch(() => null);
       return await patch(row, { state: "prepared", receipt: { ...row.receipt,
@@ -170,7 +174,7 @@ export function agentDirectoryWorkingCopyService(db: Db, get: (companyId: string
   }
   async function hasChanges(row: Copy, target?: AdapterExecutionTarget | null) {
     try {
-      if (row.receipt?.retainedByRunId || typeof row.receipt?.materializationIdentity !== "string") return true;
+      if (row.receipt?.retainedByRunId || row.receipt?.retirementRequired || typeof row.receipt?.materializationIdentity !== "string") return true;
       const observed = await observe(row, target);
       if (observed.identity !== row.receipt.materializationIdentity || observed.digest !== agentDirectoryBaselineDigest(baseline(row))) return true;
       const unchanged = await patch(row, { state: "unchanged_turn", nextAttemptAt: null });
@@ -178,16 +182,47 @@ export function agentDirectoryWorkingCopyService(db: Db, get: (companyId: string
     } catch { return true; }
   }
   /** Rebind one physical materialization; earlier runs lose collection authority. */
-  async function adopt(input: { companyId: string; agentId: string; runId: string; previousRunId: string; target?: AdapterExecutionTarget | null; cwd: string }) {
+  async function adopt(input: { companyId: string; agentId: string; runId: string; previousRunId: string; target?: AdapterExecutionTarget | null; cwd: string; allowRetirementHandoff?: boolean }) {
     let prior = await get(input.companyId, input.previousRunId);
     if (!prior || prior.agentId !== input.agentId || prior.state !== "unchanged_turn" || prior.receipt?.retainedByRunId || prior.processStoppedAt
       || prior.receipt?.workspaceCwd !== input.cwd || prior.location !== (input.target?.kind === "remote" ? `remote:${input.target.environmentId ?? ""}` : "local")) return null;
-    if (await hasChanges(prior, input.target)) return null;
-    prior = (await get(input.companyId, input.previousRunId))!;
+    const oldRuntime = transports.get(key(prior));
+    const remote = input.target?.kind === "remote" ? input.target : null;
+    const priorCleanup = prior.receipt?.cleanup as { leaseId?: string; remoteCwd?: string } | undefined;
+    const validLeases = (leases: Array<typeof environmentLeases.$inferSelect>) => {
+      if (!remote) return true;
+      const previous = leases.find(lease => lease.id === priorCleanup?.leaseId);
+      const successor = leases.find(lease => lease.id === remote.leaseId);
+      return Boolean(previous && successor && previous.id !== successor.id
+        && previous.companyId === input.companyId && successor.companyId === input.companyId
+        && previous.heartbeatRunId === input.previousRunId && successor.heartbeatRunId === input.runId
+        && previous.environmentId === remote.environmentId && successor.environmentId === remote.environmentId
+        && previous.provider && previous.provider === successor.provider
+        && previous.providerLeaseId && previous.providerLeaseId === successor.providerLeaseId
+        && successor.status === "active" && !successor.releasedAt
+        && !hasRemoteTerminationReceipt(previous) && !hasRemoteTerminationReceipt(successor)
+        && (remote.transport !== "sandbox" || remote.providerKey === successor.provider)
+        && (successor.metadata?.remoteCwd === undefined || successor.metadata.remoteCwd === remote.remoteCwd));
+    };
+    const leaseIds = remote && priorCleanup?.leaseId && remote.leaseId ? [priorCleanup.leaseId, remote.leaseId] : [];
+    if (remote && (!oldRuntime || oldRuntime.target.kind !== "remote" || oldRuntime.target.transport !== remote.transport
+      || oldRuntime.target.leaseId !== priorCleanup?.leaseId || oldRuntime.target.remoteCwd !== remote.remoteCwd
+      || priorCleanup?.remoteCwd !== remote.remoteCwd || leaseIds.length !== 2
+      || !validLeases(await db.select().from(environmentLeases).where(and(eq(environmentLeases.companyId, input.companyId), inArray(environmentLeases.id, leaseIds)))))) return null;
     const bound = await resolveInstructionActor(db, { type: "agent", companyId: input.companyId, agentId: input.agentId, runId: input.runId });
-    const adopted = await store.locked(input.companyId, input.agentId, bound, false, async (tx, _agent, canonical) => {
+    const observedChange = await hasChanges(prior, input.target);
+    if (observedChange && !input.allowRetirementHandoff) return null;
+    prior = (await get(input.companyId, input.previousRunId))!;
+    // adopt_remote constructs a collector on the successor target without
+    // uploading the stale host mirror over the retained remote directory.
+    const replacement = remote ? await transport({ ...prior, runId: input.runId }, remote, true) : undefined;
+    let adopted: Copy | null;
+    try { adopted = await store.locked(input.companyId, input.agentId, bound, false, async (tx, _agent, canonical) => {
+      if (remote && !validLeases(await tx.select().from(environmentLeases).where(and(
+        eq(environmentLeases.companyId, input.companyId), inArray(environmentLeases.id, leaseIds))).for("update"))) return null;
       const current = await inspectAgentDirectory(canonical);
-      if (agentDirectoryBaselineDigest(current.snapshot) !== agentDirectoryBaselineDigest(baseline(prior))) return null;
+      const canonicalChanged = agentDirectoryBaselineDigest(current.snapshot) !== agentDirectoryBaselineDigest(baseline(prior));
+      if (canonicalChanged && !input.allowRetirementHandoff) return null;
       const [claimed] = await tx.update(copies).set({ receipt: { ...prior.receipt, retainedByRunId: input.runId }, updatedAt: new Date() })
         .where(and(eq(copies.companyId, input.companyId), eq(copies.runId, prior.runId), eq(copies.agentId, input.agentId), eq(copies.state, "unchanged_turn"),
           eq(copies.updatedAt, prior.updatedAt), eq(copies.baseHash, prior.baseHash), eq(copies.localRoot, prior.localRoot),
@@ -195,20 +230,32 @@ export function agentDirectoryWorkingCopyService(db: Db, get: (companyId: string
       if (!claimed) throw conflict("Agent directory ownership changed");
       const { createdAt: _createdAt, updatedAt: _updatedAt, ...values } = prior;
       const [next] = await tx.insert(copies).values({ ...values, runId: input.runId, responsibleUserId: bound.onBehalfOfUserId!, state: "prepared",
-        receipt: { ...prior.receipt, materializationRunId: prior.receipt?.materializationRunId ?? prior.runId }, updatedAt: new Date(),
+        receipt: { ...prior.receipt, materializationRunId: prior.receipt?.materializationRunId ?? prior.runId,
+          ...(remote ? { cleanup: { leaseId: remote.leaseId, remoteCwd: remote.remoteCwd } } : {}),
+          ...(observedChange || canonicalChanged ? { retirementRequired: true } : {}) }, updatedAt: new Date(),
       }).returning();
       return next;
-    });
+    }); } catch (error) { await replacement?.cleanupWorkspaceSnapshot?.(); throw error; }
     // Do not move the live transport until the database ownership commit succeeds.
     if (adopted) {
-      const runtime = transports.get(key(prior));
-      if (runtime) { transports.delete(key(prior)); transports.set(key(adopted), runtime); }
-    }
+      if (replacement && oldRuntime) {
+        const cleanup = replacement.cleanupWorkspaceSnapshot;
+        replacement.cleanupWorkspaceSnapshot = async () => {
+          try { await cleanup?.(); } finally { await oldRuntime.cleanupWorkspaceSnapshot?.(); }
+        };
+        transports.delete(key(prior)); transports.set(key(adopted), replacement);
+      }
+    } else await replacement?.cleanupWorkspaceSnapshot?.();
     return adopted;
   }
   async function collectStopped(row: Copy, target?: AdapterExecutionTarget | null) {
     if (row.receipt?.retainedByRunId) return row;
     if (completed.has(row.state)) { await release(row); return (await get(row.companyId, row.runId))!; }
+    const remoteLease = row.location !== "local" ? await ownedRemoteLease(row) : null;
+    if (row.location !== "local" && (!remoteLease
+      || remoteLease.releasedAt || remoteLease.status !== "active" || hasRemoteTerminationReceipt(remoteLease))) {
+      return recoverStoppedRemote(row);
+    }
     row = await patch(row, { processStoppedAt: row.processStoppedAt ?? new Date() });
     // Adoption may win between the initial read and the stop CAS. The returned
     // current row is authoritative; a transferred alias cannot inspect or save.
@@ -257,16 +304,25 @@ export function agentDirectoryWorkingCopyService(db: Db, get: (companyId: string
     let cleanupPending = false;
     await runtime?.cleanupWorkspaceSnapshot?.().catch(() => { cleanupPending = true; });
     const cleanupTarget = runtime?.target ?? target;
-    if (row.processStoppedAt && completed.has(row.state) && cleanupTarget?.kind === "remote") {
+    const lease = row.location !== "local" ? await ownedRemoteLease(row) : null;
+    const terminated = lease && hasRemoteTerminationReceipt(lease);
+    const destroyed = terminated && (lease.metadata?.remoteExecutionTermination as { state?: string } | undefined)?.state === "destroyed";
+    const cleanupLeaseId = (row.receipt?.cleanup as { leaseId?: string } | undefined)?.leaseId;
+    // Maintenance may run while the collector is still cached. A cached target
+    // cannot override the current lease's stopped/destroyed/uncertain state.
+    const remoteUnavailable = row.location !== "local" && !destroyed && (!lease || terminated
+      || Boolean(lease && (lease.releasedAt || lease.status !== "active"))
+      || Boolean(cleanupLeaseId && (!lease || cleanupTarget?.kind === "remote" && cleanupTarget.leaseId !== lease.id)));
+    if (remoteUnavailable) cleanupPending = true;
+    else if (!destroyed && row.processStoppedAt && completed.has(row.state) && cleanupTarget?.kind === "remote") {
       const expected = path.posix.join(cleanupTarget.remoteCwd, ".paperclip-runtime", "agent-files", row.agentId, typeof row.receipt?.materializationRunId === "string" ? row.receipt.materializationRunId : row.runId);
       if (row.executionRoot !== expected) throw new Error("Agent directory cleanup path changed");
       const quoted = `'${expected.replaceAll("'", `'"'"'`)}'`;
       const remoteCleanupFailed = await runAdapterExecutionTargetShellCommand(row.runId, cleanupTarget, `rm -rf -- ${quoted}`,
         { cwd: cleanupTarget.remoteCwd, env: {}, timeoutSec: 15 }).then(result => result.exitCode !== 0 || result.timedOut, () => true);
       cleanupPending ||= remoteCleanupFailed;
-    } else if (row.processStoppedAt && completed.has(row.state) && row.location !== "local") {
-      // Rebind only the original host-owned lease. Never acquire a replacement
-      // sandbox or persist transport credentials in a working-copy receipt.
+    } else if (!destroyed && row.processStoppedAt && completed.has(row.state) && row.location !== "local") {
+      // Rebind only the current host-owned lease. Never acquire a replacement.
       cleanupPending ||= !await cleanupRemoteAfterRestart(row).catch(() => false);
     }
     if (completed.has(row.state) && row.processStoppedAt) {
@@ -277,23 +333,48 @@ export function agentDirectoryWorkingCopyService(db: Db, get: (companyId: string
         return;
       }
       await patch(row, { receipt: { schema: AGENT_FILES_CONTRACT, state: row.state, appliedCandidateHash: row.candidateHash, storageWarning: row.receipt?.storageWarning ?? null, cleanupPending,
+        materializationRunId: row.receipt?.materializationRunId ?? row.runId,
         ...(cleanupPending ? { cleanup: row.receipt?.cleanup } : {}) },
         nextAttemptAt: cleanupPending ? new Date(Date.now() + 30_000) : null });
     }
   }
+  async function ownedRemoteLease(row: Copy) {
+    const cleanup = row.receipt?.cleanup as { leaseId?: string } | undefined;
+    const leases = await db.select().from(environmentLeases).where(and(
+      eq(environmentLeases.companyId, row.companyId), eq(environmentLeases.heartbeatRunId, row.runId),
+      cleanup?.leaseId ? eq(environmentLeases.id, cleanup.leaseId) : eq(environmentLeases.environmentId, row.location.slice("remote:".length)),
+    ));
+    if (leases.length !== 1) return null;
+    const lease = leases[0]!;
+    const destroyedWithoutEnvironment = lease.environmentId === null && hasRemoteTerminationReceipt(lease)
+      && (lease.metadata?.remoteExecutionTermination as { state?: string } | undefined)?.state === "destroyed";
+    if (row.location !== `remote:${lease.environmentId}` && !destroyedWithoutEnvironment) return null;
+    return lease;
+  }
+  async function recoverStoppedRemote(row: Copy) {
+    const lease = await ownedRemoteLease(row);
+    const destroyed = lease && hasRemoteTerminationReceipt(lease)
+      && (lease.metadata?.remoteExecutionTermination as { state?: string } | undefined)?.state === "destroyed";
+    if (!destroyed) {
+      // execute (including bypassSession) may restart a stopped sandbox. Keep
+      // the only remaining bytes; a stop receipt is not a destruction receipt.
+      return patch(row, { state: "pending_collection", errorCode: lease ? "INSTRUCTION_STOPPED_REMOTE_COLLECTION_PENDING" : "INSTRUCTION_REMOTE_LEASE_UNVERIFIED",
+        errorMessage: lease ? "The remote provider stopped with its agent directory retained. Safe stopped-file retrieval is unavailable; no save or discard is claimed."
+          : "The current remote lease could not be verified. Its agent directory is preserved; no retrieval, save or discard is claimed.", nextAttemptAt: null });
+    }
+    row = await patch(row, { state: "unavailable", processStoppedAt: row.processStoppedAt ?? new Date(),
+      errorCode: "INSTRUCTION_COLLECTION_UNAVAILABLE", errorMessage: "The exact remote sandbox was destroyed before its agent directory was retrieved. No instruction save is claimed.", nextAttemptAt: null });
+    await release(row);
+    return (await get(row.companyId, row.runId))!;
+  }
   async function cleanupRemoteAfterRestart(row: Copy): Promise<boolean> {
     const cleanup = row.receipt?.cleanup as { leaseId?: string; remoteCwd?: string } | undefined;
-    // Older preview receipts did not record the lease ID. Accept only a unique
-    // lease still bound to this run and environment in that compatibility case.
-    const leases = await db.select().from(environmentLeases).where(and(
-      eq(environmentLeases.companyId, row.companyId),
-      cleanup?.leaseId ? eq(environmentLeases.id, cleanup.leaseId) : and(
-        eq(environmentLeases.environmentId, row.location.slice("remote:".length)), eq(environmentLeases.heartbeatRunId, row.runId)),
-    ));
-    if (leases.length !== 1) return false;
-    const lease = leases[0]!;
+    const lease = await ownedRemoteLease(row);
+    if (!lease) return false;
     const termination = lease.metadata?.remoteExecutionTermination as { state?: string } | undefined;
-    if (hasRemoteTerminationReceipt(lease) && termination?.state === "destroyed") return true;
+    if (hasRemoteTerminationReceipt(lease)) return termination?.state === "destroyed";
+    // Never execute a cleanup command against a released/uncertain allocation.
+    if (lease.releasedAt || lease.status !== "active") return false;
     if (!environmentRuntime || !lease.environmentId || row.location !== `remote:${lease.environmentId}`) return false;
     const [environment] = await db.select().from(environments).where(eq(environments.id, lease.environmentId));
     if (!environment) return false;
@@ -313,6 +394,7 @@ export function agentDirectoryWorkingCopyService(db: Db, get: (companyId: string
     });
   }
   return { prepare, hasChanges, adopt,
+    recoverStoppedRemote: (row: Copy) => serial(row, current => current.receipt?.retainedByRunId ? Promise.resolve(current) : recoverStoppedRemote(current)),
     collectStopped: (row: Copy, target?: AdapterExecutionTarget | null) => serial(row, current => collectStopped(current, target)),
     release: (row: Copy) => serial(row, release),
   };

server/src/services/agent-instruction-working-copies.ts

@@ -311,16 +311,8 @@ export function agentInstructionWorkingCopyService(db: Db, options: { environmen
         const result = await collectStopped({ companyId: row.companyId, runId: row.runId });
         if (result && isAgentDirectoryCopy(result) && completed.has(result.state)) await directories.release(result);
       } else if (row.location !== "local" && await remoteExecutionHasStopped(db, row.companyId, row.runId)) {
-        // An unchanged-turn receipt protects a live warm owner. Only this
-        // independently stop-proved branch may turn lost remote bytes into a
-        // terminal no-save receipt and release the retained materialization.
-        const unavailable = isAgentDirectoryCopy(row) && row.state === "unchanged_turn"
-          ? await patch(row, { state: "unavailable", errorCode: "INSTRUCTION_COLLECTION_UNAVAILABLE",
-              errorMessage: "The remote provider stopped before its retained agent directory could be retrieved. No instruction save is claimed.", nextAttemptAt: null })
-          : await reportUnavailable(row.companyId, row.runId);
-        if (unavailable && isAgentDirectoryCopy(unavailable)) {
-          await directories.release(await patch(unavailable, { processStoppedAt: unavailable.processStoppedAt ?? new Date() }));
-        }
+        if (isAgentDirectoryCopy(row)) await directories.recoverStoppedRemote(row);
+        else await reportUnavailable(row.companyId, row.runId);
       } else if (row.state === "prepared") {
         await patch(row, { state: "pending_collection", errorCode: "INSTRUCTION_STOP_UNCONFIRMED",
           errorMessage: "The provider's stop has not been confirmed. Instruction collection is pending; no save is claimed.", nextAttemptAt: null });

server/src/services/heartbeat.ts

@@ -221,6 +221,7 @@ import {
   cancelNativeSession,
   claimNativeRestartRecoveries,
   claimWarmNativeInstructionCopy,
+  nativeSessionWorkspaceScope,
   closeWarmNativeSessionsForEnvironment,
   closeIdleWarmNativeSessionsForRestart,
   currentNativeControllerIdentity,
@@ -22423,7 +22424,10 @@ export function heartbeatService(
       let instructionSave: Record<string, unknown> | null = null;
       const instructionPreparationKey = createHash("sha256").update(JSON.stringify({
         adapterType: agent.adapterType, adapterConfig: agent.adapterConfig, runtimeConfig: agent.runtimeConfig, sessionConfigMetadata,
-        workspaceId: persistedExecutionWorkspace?.id ?? run.id, cwd: executionWorkspace.cwd,
+        workspace: nativeSessionWorkspaceScope({
+          binding: { runId: run.id, executionWorkspaceId: persistedExecutionWorkspace?.id ?? run.id },
+          workspace: executionWorkspace,
+        }), cwd: executionWorkspace.cwd,
       })).digest("hex");
       const collectStoppedInstructions = async () => {
         if (!instructionCopy) return;
@@ -23278,14 +23282,17 @@ export function heartbeatService(
             if (nativeRuntimeResolution.kind === "native" && priorWorkingCopy.kind === "agent_files" && taskSession) {
               releaseWarmInstructionPreparation = await claimWarmNativeInstructionCopy({
                 priorExecution: parseNativeExecutionInput(priorFileInput), companyId: agent.companyId, agentId: agent.id,
-                executionWorkspaceId: persistedExecutionWorkspace?.id ?? run.id, cwd: executionWorkspace.cwd,
-                environmentId: executionTarget?.environmentId ?? null, runId: run.id, preparationKey: taskSessionForRun ? instructionPreparationKey : `reset:${run.id}`,
+                executionWorkspaceId: persistedExecutionWorkspace?.id ?? run.id, workspace: executionWorkspace,
+                environmentId: executionTarget?.environmentId ?? null, runId: run.id, preparationKey: instructionPreparationKey, forceRetirement: !taskSessionForRun,
                 adopt: async previousRunId => {
                   instructionCopy = await instructionCopies.adopt({ companyId: agent.companyId, agentId: agent.id,
-                    runId: run.id, previousRunId, target: executionTarget, cwd: executionWorkspace.cwd });
+                    runId: run.id, previousRunId, target: executionTarget, cwd: executionWorkspace.cwd, allowRetirementHandoff: true });
                   return nativeInstructionWorkingCopy() ?? null;
                 },
               });
+              // A collection-only handoff was retired. Re-enter normal
+              // preparation instead of composing a root already collected.
+              if (!releaseWarmInstructionPreparation) instructionCopy = null;
             }
             instructionCopy ??= await instructionCopies.prepare({
               companyId: agent.companyId, agentId: agent.id, runId: run.id,

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

@@ -2,6 +2,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 import { readFileSync } from "node:fs";
 import {
   access,
+  chmod,
   cp,
   lstat,
   mkdir,
@@ -35,6 +36,7 @@ import {
   validatePrpEvent,
   parseNativeExecutionInput,
   type NativeExecutionInputV1,
+  type NativeExecutionInputV3,
   type NativeExecutionInput,
   type PrpEvent,
 } from "@paperclipai/paperclip-runner";
@@ -282,6 +284,7 @@ import {
   cancelNativeSession,
   closeWarmNativeSessionsForEnvironment,
   claimWarmNativeInstructionCopy,
+  nativeSessionWorkspaceScope,
   closeIdleWarmNativeSessionsForRestart,
   createGovernedWaitEventObservation,
   createRemoteRunnerProcessLauncher,
@@ -5977,7 +5980,251 @@ describe("native session same-turn steering", () => {
 });
 
 describe("native warm session supervision", () => {
-  it.each(["idle", "edit", "add", "remove"])("retains the actual agent_files copy across warm turns, then retires for %s", async ending => {
+  it.each(["collect", "stopped", "destroyed", "missing-lease", "foreign-run", "foreign-environment", "deleted-environment", "claim-edit", "claim-add", "claim-remove", "claim-remove-entry", "claim-canonical", "claim-config", "claim-uncertain"])("composes remote warm ownership with real DB successor leases and current-run collection: %s", async ending => {
+    const tables = await import("@paperclipai/db");
+    const { eq } = await import("drizzle-orm");
+    const { randomUUID } = await import("node:crypto");
+    const { promisify } = await import("node:util");
+    const { execFile } = await import("node:child_process");
+    const executeCommand = promisify(execFile);
+    const { startEmbeddedPostgresTestDatabase } = await import("../../__tests__/helpers/embedded-postgres.js");
+    const { agentInstructionWorkingCopyService } = await import("../agent-instruction-working-copies.js");
+    const { buildNativeRuntimeContext } = await import("./runtime-context.js");
+    const { prepareNativeHeartbeatRun } = await import("./prepare-native-run.js");
+    const { buildNativeCompletionContract } = await import("./completion-contracts.js");
+    const home = await realpath(await mkdtemp(join(tmpdir(), "warm-remote-db-")));
+    const previousHome = process.env.PAPERCLIP_HOME;
+    process.env.PAPERCLIP_HOME = home;
+    const database = await startEmbeddedPostgresTestDatabase("warm-remote-db-");
+    const db = tables.createDb(database.connectionString);
+    const companyId = randomUUID(), agentId = randomUUID(), userId = randomUUID(), environmentId = randomUUID(), issueId = randomUUID();
+    const { resolveManagedInstructionsRoot } = await import("../agent-instructions.js");
+    const canonical = resolveManagedInstructionsRoot({ id: agentId, companyId, name: "Warm remote", adapterConfig: {} }), remoteCwd = join(home, "remote");
+    let currentLease = "", failCleanup = false, failProbe = false, closed = false;
+    const commands: Array<{ leaseId: string; command: string }> = [];
+    const commandRunner = (leaseId: string): import("@paperclipai/adapter-utils/command-managed-runtime").CommandManagedRuntimeRunner => ({ execute: async input => {
+      expect(leaseId).toBe(currentLease); // The replaced collector must never execute.
+      commands.push({ leaseId, command: `${input.command} ${(input.args ?? []).join(" ")}` });
+      if (failProbe) {
+        failProbe = false;
+        expect(commands.at(-1)!.command).toContain("Agent directory probe deadline");
+        return { exitCode: 1, signal: null, timedOut: false, stdout: "", stderr: "injected uncertain observation", pid: null, startedAt: new Date().toISOString() };
+      }
+      if (failCleanup && commands.at(-1)!.command.includes("rm -rf --")) {
+        failCleanup = false;
+        return { exitCode: 1, signal: null, timedOut: false, stdout: "", stderr: "injected cleanup failure", pid: null, startedAt: new Date().toISOString() };
+      }
+      const env = { PATH: process.env.PATH, ...input.env }, args = [...(input.args ?? [])];
+      if (input.stdin != null && (args[0] === "-c" || args[0] === "-lc")) {
+        Object.assign(env, { PAPERCLIP_TEST_STDIN: input.stdin });
+        args[1] = `printf '%s' "$PAPERCLIP_TEST_STDIN" | (${args[1]})`;
+      }
+      const result = await executeCommand(input.command, args, { cwd: input.cwd, env, timeout: input.timeoutMs, maxBuffer: 32 * 1024 * 1024 });
+      return { exitCode: 0, signal: null, timedOut: false, ...result, pid: null, startedAt: new Date().toISOString() };
+    } });
+    let copies = agentInstructionWorkingCopyService(db);
+    const agent = { id: agentId, companyId, name: "Warm remote", adapterConfig: {
+      instructionsBundleMode: "managed", instructionsRootPath: canonical, instructionsEntryFile: "AGENTS.md" } };
+    let active: Awaited<ReturnType<typeof copies.prepare>>;
+    const session = { close: vi.fn(async () => { closed = true; }) };
+    state.execute.mockReset().mockImplementation(async options => {
+      await options.onSession?.(session);
+      return { result: { summary: "completed" }, terminal: { runTerminalState: "succeeded" },
+        turnId: "warm-turn", normalizedSessionId: environmentId, providerSessionId: environmentId,
+        driverKind: "test", driverVersion: "1", nativeEventCount: 1, highestContiguousSourceSeq: 1, usage: null };
+    });
+    let bodyFailure: unknown;
+    try {
+      await db.insert(tables.companies).values({ id: companyId, name: "Warm tests", issuePrefix: "WARM" });
+      await db.insert(tables.authUsers).values({ id: userId, name: "Editor", email: `${userId}@example.test`, createdAt: new Date(), updatedAt: new Date() });
+      await db.insert(tables.agents).values(agent);
+      await db.insert(tables.companyMemberships).values([{ companyId, principalType: "user", principalId: userId, membershipRole: "operator" }, { companyId, principalType: "agent", principalId: agentId, membershipRole: "member" }]);
+      await db.insert(tables.principalPermissionGrants).values({ companyId, principalType: "user", principalId: userId, permissionKey: "agents:configure", scope: { agentIds: [agentId] } });
+      await db.insert(tables.environments).values({ id: environmentId, name: "Owned fake remote", driver: "sandbox" });
+      const [issue] = await db.insert(tables.issues).values({ id: issueId, companyId, title: "Warm remote", status: "in_progress", assigneeAgentId: agentId }).returning();
+      await mkdir(canonical, { recursive: true }); await mkdir(remoteCwd);
+      await writeFile(join(canonical, "AGENTS.md"), "Retain this exact directory.\n");
+      await writeFile(join(canonical, "optional-memory.txt"), "Optional memory\n");
+      let prior: NativeExecutionInputV3 | undefined;
+      let originalRoot = "", materializationRunId = "";
+      for (let turn = 0; turn < 3; turn++) {
+        const runId = randomUUID();
+        if (currentLease) {
+          await db.update(tables.environmentLeases).set({ status: "expired", releasedAt: new Date(), cleanupStatus: "success" }).where(eq(tables.environmentLeases.id, currentLease));
+          await db.update(tables.heartbeatRuns).set({ status: "succeeded" }).where(eq(tables.heartbeatRuns.id, prior!.binding.runId));
+        }
+        currentLease = randomUUID();
+        const [run] = await db.insert(tables.heartbeatRuns).values({ id: runId, companyId, agentId, invocationSource: "on_demand", responsibleUserId: userId,
+          status: "running", runtimeMode: "native", nativeIssueId: issueId, nativeSessionId: environmentId, runnerInstanceId: environmentId,
+          contextSnapshot: { issueId } }).returning();
+        await db.insert(tables.nativeRunFinalizations).values({ runId, companyId, issueId, phase: "observed" });
+        await db.update(tables.issues).set({ executionRunId: runId }).where(eq(tables.issues.id, issueId));
+        await db.insert(tables.environmentLeases).values({ id: currentLease, companyId, environmentId, heartbeatRunId: runId,
+          provider: "daytona", providerLeaseId: "same-owned-allocation", leasePolicy: "reuse_by_environment", metadata: { remoteCwd } });
+        // Match the production preparation seam and existing native DB tests:
+        // pin the issue's completion contract and durable run/lease identities.
+        await prepareNativeHeartbeatRun({ db, run: run!, issue: issue!, environmentLeaseId: currentLease });
+        const [contract] = await db.select().from(tables.completionContracts).where(eq(tables.completionContracts.issueId, issueId));
+        const target = { kind: "remote" as const, transport: "sandbox" as const, providerKey: "daytona", environmentId, leaseId: currentLease, remoteCwd, runner: commandRunner(currentLease) };
+        const current: NativeExecutionInputV3 = { ...execution, schema: "paperclip.native-execution-input.v3",
+          executionMode: "default", planningContext: null, runtimeContext: nativeRuntimeContextFixture(),
+          completionContract: { id: contract!.id, sha256: contract!.canonicalSha256, schemaVersion: "paperclip.completion-contract.v1",
+            contract: buildNativeCompletionContract(issue!, { revision: contract!.revision }) }, binding: { companyId, agentId, issueId, runId, executionWorkspaceId: runId },
+          workspace: { cwd: remoteCwd, repoUrl: null, repoRef: null, branchName: null },
+          session: { ...execution.session, normalizedSessionId: environmentId, lifecyclePolicy: { mode: "warm", idleTimeoutMs: 60_000 } } };
+        const preparationKey = turn === 1 && ending === "claim-config" ? "changed-config" : "same-config";
+        let retiredReceipt: Record<string, unknown> | null = null;
+        const capability = () => ({ runId, preparationKey, hasChanges: () => copies.hasChanges({ companyId, runId, target }),
+          collectStopped: async () => { expect(closed).toBe(true); retiredReceipt = (await copies.get(companyId, runId))?.receipt ?? null; await copies.collectStopped({ companyId, runId, target }); } });
+        let release: (() => Promise<void>) | null = null;
+        if (prior) {
+          // Same environment is insufficient: reject a different physical allocation.
+          await db.update(tables.environmentLeases).set({ providerLeaseId: "foreign-allocation" }).where(eq(tables.environmentLeases.id, currentLease));
+          const before = commands.length;
+          expect(await copies.adopt({ companyId, agentId, runId, previousRunId: prior.binding.runId, target, cwd: remoteCwd })).toBeNull();
+          expect(commands).toHaveLength(before);
+          await db.update(tables.environmentLeases).set({ providerLeaseId: "same-owned-allocation" }).where(eq(tables.environmentLeases.id, currentLease));
+          if (ending === "claim-edit") await writeFile(join(originalRoot, "AGENTS.md"), "Remote edit before next claim\n");
+          if (ending === "claim-add") await writeFile(join(originalRoot, "memory.txt"), "Remote addition before next claim\n");
+          if (ending === "claim-remove") await rm(join(originalRoot, "optional-memory.txt"));
+          if (ending === "claim-remove-entry") await rm(join(originalRoot, "AGENTS.md"));
+          if (ending === "claim-canonical") await writeFile(join(canonical, "AGENTS.md"), "Canonical edit before next claim\n");
+          if (["claim-edit", "claim-add", "claim-remove", "claim-remove-entry", "claim-canonical"].includes(ending)) {
+            // Ordinary adoption remains unchanged-only and cannot alias this owner.
+            expect(await copies.adopt({ companyId, agentId, runId, previousRunId: prior.binding.runId, target, cwd: remoteCwd })).toBeNull();
+            expect((await copies.get(companyId, prior.binding.runId))?.receipt?.retainedByRunId).toBeUndefined();
+          }
+          failProbe = ending === "claim-uncertain";
+          release = await claimWarmNativeInstructionCopy({ priorExecution: prior, companyId, agentId, executionWorkspaceId: runId,
+            workspace: current.workspace, environmentId, runId, preparationKey, adopt: async previousRunId => {
+              active = await copies.adopt({ companyId, agentId, runId, previousRunId, target, cwd: remoteCwd, allowRetirementHandoff: true });
+              return active ? capability() : null;
+            } });
+          if (ending.startsWith("claim-")) {
+            expect(release).toBeNull();
+            expect(session.close).toHaveBeenCalledOnce();
+            expect(state.execute).toHaveBeenCalledOnce(); // No reuse was admitted.
+            if (ending !== "claim-config") expect(retiredReceipt).toMatchObject({ retirementRequired: true });
+            expect(retiredReceipt).toMatchObject({ materializationRunId, cleanup: { leaseId: currentLease, remoteCwd } });
+            expect((await copies.get(companyId, runId))?.processStoppedAt).toBeInstanceOf(Date);
+            const compact = (await copies.get(companyId, runId))!.receipt;
+            expect(compact).toMatchObject({ materializationRunId, cleanupPending: false });
+            expect(compact?.baseline).toBeUndefined();
+            await expect(access(originalRoot)).rejects.toMatchObject({ code: "ENOENT" });
+            if (ending === "claim-edit") expect(await readFile(join(canonical, "AGENTS.md"), "utf8")).toBe("Remote edit before next claim\n");
+            if (ending === "claim-add") expect(await readFile(join(canonical, "memory.txt"), "utf8")).toBe("Remote addition before next claim\n");
+            if (ending === "claim-remove") await expect(access(join(canonical, "optional-memory.txt"))).rejects.toMatchObject({ code: "ENOENT" });
+            if (ending === "claim-remove-entry") {
+              expect((await copies.get(companyId, runId))!).toMatchObject({ state: "unavailable", errorCode: "AGENT_FILES_SAVE_FAILED" });
+              expect(await readFile(join(canonical, "AGENTS.md"), "utf8")).toBe("Retain this exact directory.\n");
+            }
+            if (ending === "claim-canonical") expect(await readFile(join(canonical, "AGENTS.md"), "utf8")).toBe("Canonical edit before next claim\n");
+            // Heartbeat rematerializes only after the successor proved retirement.
+            active = await copies.prepare({ companyId, agentId, runId, target, cwd: remoteCwd });
+            expect(active).toMatchObject({ state: "prepared", executionRoot: originalRoot, processStoppedAt: null });
+            expect(active!.receipt?.retirementRequired).toBeUndefined();
+            expect(await copies.hasChanges({ companyId, runId, target })).toBe(false);
+            await copies.reportUnavailable(companyId, prior.binding.runId);
+            await copies.release(companyId, prior.binding.runId);
+            expect((await lstat(originalRoot)).isDirectory()).toBe(true);
+            await copies.collectStopped({ companyId, runId, target });
+            return;
+          }
+          expect(release).toBeTypeOf("function");
+          await copies.reportUnavailable(companyId, prior.binding.runId);
+          await copies.release(companyId, prior.binding.runId);
+          expect(active!.receipt).toMatchObject({ materializationRunId, cleanup: { leaseId: currentLease, remoteCwd } });
+        } else {
+          active = await copies.prepare({ companyId, agentId, runId, target, cwd: remoteCwd });
+          originalRoot = active!.executionRoot; materializationRunId = runId;
+        }
+        expect(active!.executionRoot).toBe(originalRoot);
+        current.runtimeContext = await buildNativeRuntimeContext({ db, agent, runId, runtimeConfig: {}, runtimeSkillEntries: [],
+          instructionWorkingCopy: { rootPath: active!.executionRoot, entryPath: "AGENTS.md", kind: "agent_files" } });
+        expect(current.runtimeContext.instructions.workingCopy?.rootPath).toBe(originalRoot);
+        await executePaperclipNativeSession({ db, execution: current, runnerInstanceId: environmentId, runnerExecutionTarget: target, instructionWorkingCopy: capability() });
+        if (prior) expect(state.execute.mock.calls.at(-1)?.[0].existingSession).toBe(session);
+        expect(closed).toBe(false);
+        await release?.(); expect(closed).toBe(false);
+        prior = current;
+      }
+      // A write after the live unchanged probe is still collected on retirement.
+      await writeFile(join(originalRoot, "late-memory.txt"), "Latest owner's remote bytes");
+      if (["missing-lease", "foreign-run", "foreign-environment", "deleted-environment"].includes(ending)) {
+        if (ending === "missing-lease") await db.delete(tables.environmentLeases).where(eq(tables.environmentLeases.id, currentLease));
+        if (ending === "foreign-run") await db.update(tables.environmentLeases).set({ heartbeatRunId: materializationRunId }).where(eq(tables.environmentLeases.id, currentLease));
+        if (ending === "foreign-environment") {
+          const other = randomUUID(); await db.insert(tables.environments).values({ id: other, name: "Other", driver: "sandbox" });
+          await db.update(tables.environmentLeases).set({ environmentId: other }).where(eq(tables.environmentLeases.id, currentLease));
+        }
+        if (ending === "deleted-environment") await db.delete(tables.environments).where(eq(tables.environments.id, environmentId));
+        const before = commands.length;
+        await closeWarmNativeSessionsForEnvironment({ environmentId, reason: "lease identity changed" });
+        expect(commands).toHaveLength(before);
+        const pending = (await copies.get(companyId, prior!.binding.runId))!;
+        expect(pending.state).toBe("pending_collection");
+        expect(pending.receipt?.baseline).toBeDefined();
+        expect(await readFile(join(originalRoot, "late-memory.txt"), "utf8")).toBe("Latest owner's remote bytes");
+        return;
+      }
+      if (ending !== "collect") {
+        const { remoteTerminationReceipt } = await import("../remote-execution-termination.js");
+        const [lease] = await db.select().from(tables.environmentLeases).where(eq(tables.environmentLeases.id, currentLease));
+        await db.update(tables.environmentLeases).set({ status: "released", releasedAt: new Date(), cleanupStatus: "success",
+          metadata: { ...lease!.metadata, remoteExecutionTermination: remoteTerminationReceipt(lease!, { providerLeaseId: lease!.providerLeaseId, state: ending }) } }).where(eq(tables.environmentLeases.id, currentLease));
+        await db.update(tables.heartbeatRuns).set({ status: "succeeded" }).where(eq(tables.heartbeatRuns.id, prior!.binding.runId));
+        const before = commands.length;
+        await copies.recoverStopped(); // The original remote collector is still cached.
+        expect(commands).toHaveLength(before);
+        const recovered = (await copies.get(companyId, prior!.binding.runId))!;
+        expect(recovered.state).toBe(ending === "destroyed" ? "unavailable" : "pending_collection");
+        await expect(access(join(canonical, "late-memory.txt"))).rejects.toMatchObject({ code: "ENOENT" });
+        if (ending === "stopped") {
+          expect(await readFile(join(originalRoot, "late-memory.txt"), "utf8")).toBe("Latest owner's remote bytes");
+          expect(recovered.receipt?.baseline).toBeDefined();
+          await copies.collectStopped({ companyId, runId: prior!.binding.runId });
+          await copies.release(companyId, prior!.binding.runId);
+          expect(commands).toHaveLength(before);
+        } else await expect(access(active!.localRoot)).rejects.toMatchObject({ code: "ENOENT" });
+        await closeWarmNativeSessionsForEnvironment({ environmentId, reason: "already terminated owner" });
+        expect(commands).toHaveLength(before);
+        return;
+      }
+      failCleanup = true;
+      await closeWarmNativeSessionsForEnvironment({ environmentId, reason: "test retirement" });
+      expect(session.close).toHaveBeenCalledOnce();
+      expect(await readFile(join(canonical, "late-memory.txt"), "utf8")).toBe("Latest owner's remote bytes");
+      const saved = (await copies.get(companyId, prior!.binding.runId))!;
+      expect(saved).toMatchObject({ state: "saved", receipt: { cleanupPending: true, materializationRunId, cleanup: { leaseId: currentLease } } });
+      const cleanup = vi.fn(async input => {
+        expect(input.lease.id).toBe(currentLease);
+        expect(input.lease.heartbeatRunId).toBe(prior!.binding.runId);
+        return commandRunner(currentLease).execute(input);
+      });
+      copies = agentInstructionWorkingCopyService(db, { environmentRuntime: { execute: cleanup } as never });
+      await db.update(tables.agentInstructionWorkingCopies).set({ nextAttemptAt: null }).where(eq(tables.agentInstructionWorkingCopies.runId, prior!.binding.runId));
+      await copies.recoverCaptured();
+      expect(cleanup).toHaveBeenCalledOnce();
+      await expect(access(originalRoot)).rejects.toMatchObject({ code: "ENOENT" });
+      expect((await copies.get(companyId, prior!.binding.runId))!.receipt?.cleanupPending).toBe(false);
+    } catch (error) { bodyFailure = error; throw error; }
+    finally {
+      const cleanupFailures: unknown[] = [];
+      try { await closeWarmNativeSessionsForEnvironment({ environmentId, reason: "test cleanup" }); } catch (error) { cleanupFailures.push(error); }
+      try { await database.cleanup(); } catch (error) { cleanupFailures.push(error); }
+      if (previousHome === undefined) delete process.env.PAPERCLIP_HOME; else process.env.PAPERCLIP_HOME = previousHome;
+      try {
+        const writable = async (directory: string): Promise<void> => {
+          await chmod(directory, 0o700);
+          for (const entry of await readdir(directory, { withFileTypes: true })) if (entry.isDirectory()) await writable(join(directory, entry.name));
+        };
+        await writable(home); await rm(home, { recursive: true, force: true });
+      } catch (error) { cleanupFailures.push(error); }
+      if (cleanupFailures.length) throw new AggregateError([...(bodyFailure ? [bodyFailure] : []), ...cleanupFailures], "Warm DB test body/cleanup failure");
+    }
+  }, 90_000);
+
+  it.each(["idle", "edit", "add", "remove", "projectless"])("retains the actual agent_files copy across warm turns, then retires for %s", async ending => {
     const root = await realpath(await mkdtemp(join(tmpdir(), "warm-agent-files-")));
     const localRoot = join(root, "run", "live");
     await mkdir(localRoot, { recursive: true });
@@ -5999,10 +6246,10 @@ describe("native warm session supervision", () => {
       expect(close).toHaveBeenCalledOnce();
       // The real DB service suite verifies changed-file persistence. Here the
       // fake backend proves stop precedes granting its collection callback.
-      if (ending === "idle") await copies.collectStopped(row);
+      if (ending === "idle" || ending === "projectless") await copies.collectStopped(row);
     };
     const identity = "warm-real-agent-files";
-    const warmExecution = { ...execution, binding: { ...execution.binding, runId: identity, executionWorkspaceId: `${identity}-workspace` },
+    const warmExecution = { ...execution, binding: { ...execution.binding, runId: identity, executionWorkspaceId: ending === "projectless" ? identity : `${identity}-workspace` },
       session: { ...execution.session, normalizedSessionId: identity, lifecyclePolicy: { mode: "warm" as const, idleTimeoutMs: 60_000 } },
     } as NativeExecutionInputV1;
     const session = { close };
@@ -6019,11 +6266,26 @@ describe("native warm session supervision", () => {
       expect(close).not.toHaveBeenCalled();
       expect(order).toEqual([]);
       expect(await readFile(join(localRoot, "AGENTS.md"), "utf8")).toContain("registered");
-      const second = { ...warmExecution, binding: { ...warmExecution.binding, runId: `${identity}-second` } };
+      const second = { ...warmExecution, binding: { ...warmExecution.binding, runId: `${identity}-second`,
+        ...(ending === "projectless" ? { executionWorkspaceId: `${identity}-second` } : {}) } };
+      expect(nativeSessionWorkspaceScope(second)).toEqual(nativeSessionWorkspaceScope(warmExecution));
       const currentCopy = { runId: second.binding.runId, preparationKey: "same-preparation", hasChanges: () => copies.hasChanges(row), collectStopped };
+      if (ending === "projectless") {
+        const adopt = vi.fn(async () => currentCopy);
+        for (const changed of [
+          { ...second.workspace, cwd: "/different" }, { ...second.workspace, repoUrl: "https://example.test/other" },
+          { ...second.workspace, repoRef: "other" }, { ...second.workspace, branchName: "other" },
+        ]) expect(await claimWarmNativeInstructionCopy({ priorExecution: warmExecution,
+          companyId: second.binding.companyId, agentId: second.binding.agentId, executionWorkspaceId: second.binding.runId,
+          workspace: changed, environmentId: identity, runId: second.binding.runId, preparationKey: "same-preparation", adopt })).toBeNull();
+        expect(await claimWarmNativeInstructionCopy({ priorExecution: warmExecution,
+          companyId: second.binding.companyId, agentId: second.binding.agentId, executionWorkspaceId: "managed-workspace",
+          workspace: second.workspace, environmentId: identity, runId: second.binding.runId, preparationKey: "same-preparation", adopt })).toBeNull();
+        expect(adopt).not.toHaveBeenCalled(); expect(close).not.toHaveBeenCalled();
+      }
       const release = await claimWarmNativeInstructionCopy({ priorExecution: warmExecution,
         companyId: warmExecution.binding.companyId, agentId: warmExecution.binding.agentId,
-        executionWorkspaceId: warmExecution.binding.executionWorkspaceId, cwd: warmExecution.workspace.cwd,
+        executionWorkspaceId: second.binding.executionWorkspaceId, workspace: second.workspace,
         environmentId: identity, runId: second.binding.runId, preparationKey: "same-preparation",
         adopt: async () => currentCopy });
       expect(release).toBeTypeOf("function");
@@ -6035,19 +6297,19 @@ describe("native warm session supervision", () => {
       expect(row.executionRoot).toBe(localRoot);
       await release?.(); // Consumed preparation cannot retire the retained owner.
       expect(close).not.toHaveBeenCalled();
-      if (ending === "idle") {
+      if (ending === "idle" || ending === "projectless") {
         await closeWarmNativeSessionsForEnvironment({ environmentId: identity, reason: "idle retirement" });
         await expect(access(localRoot)).rejects.toMatchObject({ code: "ENOENT" });
       } else {
         if (ending === "edit") await writeFile(join(localRoot, "AGENTS.md"), "Changed instructions");
         if (ending === "add") await writeFile(join(localRoot, "memory.txt"), "New memory");
         if (ending === "remove") await rm(join(localRoot, "AGENTS.md"));
-        const adopt = vi.fn(async () => currentCopy);
+        const adopt = vi.fn(async () => ({ ...currentCopy, runId: `${identity}-third` }));
         expect(await claimWarmNativeInstructionCopy({ priorExecution: second,
           companyId: second.binding.companyId, agentId: second.binding.agentId,
-          executionWorkspaceId: second.binding.executionWorkspaceId, cwd: second.workspace.cwd,
+          executionWorkspaceId: second.binding.executionWorkspaceId, workspace: second.workspace,
           environmentId: identity, runId: `${identity}-third`, preparationKey: "same-preparation", adopt })).toBeNull();
-        expect(adopt).not.toHaveBeenCalled();
+        expect(adopt).toHaveBeenCalledOnce();
       }
       expect(order).toEqual(["closed", "collected"]);
     } finally {
@@ -6079,7 +6341,7 @@ describe("native warm session supervision", () => {
     const current = { ...copy, runId: second.binding.runId };
     const adopt = vi.fn(async () => current);
     const claim = { priorExecution: first, companyId: first.binding.companyId, agentId: first.binding.agentId,
-      executionWorkspaceId: first.binding.executionWorkspaceId, cwd: first.workspace.cwd, environmentId: identity,
+      executionWorkspaceId: first.binding.executionWorkspaceId, workspace: first.workspace, environmentId: identity,
       runId: second.binding.runId, preparationKey: "key", adopt };
     try {
       await executePaperclipNativeSession({ db: leaseDb(first), execution: first, runnerInstanceId: identity, runnerExecutionTarget: target, instructionWorkingCopy: copy });
@@ -6111,6 +6373,56 @@ describe("native warm session supervision", () => {
     } finally { allowClose = true; await closeWarmNativeSessionsForEnvironment({ environmentId: identity, reason: "test cleanup" }); }
   });
 
+  it.each(["config", "reset", "changed", "unavailable", "auth", "wrong-capability", "retirement-failure"])("hands off collection before probing or retiring a warm owner: %s", async scenario => {
+    const identity = `warm-successor-${scenario}`, order: string[] = [];
+    let allowClose = scenario !== "retirement-failure";
+    const session = { close: vi.fn(async () => { order.push("stop"); if (!allowClose) throw new Error("stop unconfirmed"); }) };
+    const priorCopy = { runId: identity, preparationKey: "before", hasChanges: vi.fn(async () => false),
+      collectStopped: vi.fn(async () => { order.push("prior-pending"); }), retirementFailed: vi.fn(async () => {}) };
+    const first = { ...execution, binding: { ...execution.binding, runId: identity, executionWorkspaceId: `${identity}-workspace` },
+      session: { ...execution.session, normalizedSessionId: identity, lifecyclePolicy: { mode: "warm" as const, idleTimeoutMs: 60_000 } } } as NativeExecutionInputV1;
+    const target = { kind: "remote" as const, transport: "sandbox" as const, environmentId: identity, remoteCwd: `/tmp/${identity}` };
+    state.execute.mockReset().mockImplementation(async options => {
+      await options.onSession?.(session);
+      return { result: { summary: "completed" }, terminal: { runTerminalState: "succeeded" }, turnId: identity,
+        normalizedSessionId: identity, providerSessionId: identity, driverKind: "test", driverVersion: "1",
+        nativeEventCount: 1, highestContiguousSourceSeq: 1, usage: null };
+    });
+    const current = { runId: `${identity}-next`, preparationKey: scenario === "config" ? "after" : "before",
+      hasChanges: vi.fn(async () => { order.push("current-probe"); return true; }),
+      collectStopped: vi.fn(async () => { order.push("current-collect"); }), retirementFailed: vi.fn(async () => {}) };
+    const adopt = vi.fn(async () => {
+      order.push("handoff");
+      if (scenario === "auth") throw new Error("authorization denied");
+      if (scenario === "unavailable") return null;
+      return scenario === "wrong-capability" ? { ...current, runId: "foreign" } : current;
+    });
+    try {
+      await executePaperclipNativeSession({ db: leaseDb(first), execution: first, runnerInstanceId: identity, runnerExecutionTarget: target, instructionWorkingCopy: priorCopy });
+      priorCopy.hasChanges.mockClear();
+      const claiming = claimWarmNativeInstructionCopy({ priorExecution: first, companyId: first.binding.companyId, agentId: first.binding.agentId,
+        executionWorkspaceId: first.binding.executionWorkspaceId, workspace: first.workspace, environmentId: identity,
+        runId: current.runId, preparationKey: current.preparationKey, forceRetirement: scenario === "reset", adopt });
+      if (["unavailable", "auth", "wrong-capability", "retirement-failure"].includes(scenario)) {
+        await expect(claiming).rejects.toThrow(({ unavailable: "handoff_unavailable", auth: "authorization denied",
+          "wrong-capability": "handoff_mismatch", "retirement-failure": "stop unconfirmed" } as Record<string, string>)[scenario]);
+      } else expect(await claiming).toBeNull();
+      expect(priorCopy.hasChanges).not.toHaveBeenCalled();
+      expect(state.execute).toHaveBeenCalledOnce();
+      if (["unavailable", "auth", "wrong-capability"].includes(scenario)) {
+        expect(order).toEqual(["handoff", "stop", "prior-pending"]);
+        expect(current.collectStopped).not.toHaveBeenCalled();
+      } else {
+        expect(priorCopy.collectStopped).not.toHaveBeenCalled();
+        expect(order).toEqual(["handoff", ...(["config", "reset"].includes(scenario) ? [] : ["current-probe"]), "stop", ...(scenario === "retirement-failure" ? [] : ["current-collect"])]);
+        if (scenario === "retirement-failure") {
+          expect(current.retirementFailed).toHaveBeenCalledOnce();
+          expect(current.collectStopped).not.toHaveBeenCalled();
+        }
+      }
+    } finally { allowClose = true; await closeWarmNativeSessionsForEnvironment({ environmentId: identity, reason: "test cleanup" }); }
+  });
+
   it.each([
     { changed: false, closeFails: false },
     { changed: true, closeFails: false },

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

@@ -447,15 +447,18 @@ export async function claimWarmNativeInstructionCopy(input: {
   companyId: string;
   agentId: string;
   executionWorkspaceId: string;
-  cwd: string;
+  workspace: NativeExecutionInput["workspace"];
   environmentId: string | null;
   runId: string;
   preparationKey: string;
+  forceRetirement?: boolean;
   adopt: (previousRunId: string) => Promise<NativeInstructionWorkingCopy | null>;
 }): Promise<(() => Promise<void>) | null> {
   const prior = input.priorExecution;
   if (prior.binding.companyId !== input.companyId || prior.binding.agentId !== input.agentId
-    || prior.binding.executionWorkspaceId !== input.executionWorkspaceId || prior.workspace.cwd !== input.cwd) return null;
+    || JSON.stringify(nativeSessionWorkspaceScope(prior)) !== JSON.stringify(nativeSessionWorkspaceScope({
+      binding: { ...prior.binding, runId: input.runId, executionWorkspaceId: input.executionWorkspaceId }, workspace: input.workspace,
+    })) || prior.workspace.cwd !== input.workspace.cwd) return null;
   const scope = nativeSessionScopeKey(prior);
   const entry = warmNativeSessions.get(scope);
   if (!entry || !entry.instructionWorkingCopy?.runId || entry.environmentId !== input.environmentId) return null;
@@ -472,17 +475,21 @@ export async function claimWarmNativeInstructionCopy(input: {
     } finally { entry.busy = false; }
   };
   try {
-    if (entry.closeOnReleaseReason !== undefined || entry.instructionWorkingCopy.preparationKey !== input.preparationKey || await entry.instructionWorkingCopy.hasChanges()) {
+    const preparationChanged = input.forceRetirement === true || entry.instructionWorkingCopy.preparationKey !== input.preparationKey;
+    // The prior capability may close over an expired remote lease. Rebind
+    // collection authority first; adoption itself grants no reuse permission.
+    const adopted = await input.adopt(entry.instructionWorkingCopy.runId);
+    if (!adopted) {
+      await retire("warm instruction materialization cannot be handed off");
+      throw new Error("native_instruction_materialization_handoff_unavailable");
+    }
+    if (adopted.runId !== input.runId || adopted.preparationKey !== input.preparationKey) throw new Error("native_instruction_materialization_handoff_mismatch");
+    entry.instructionWorkingCopy = adopted;
+    entry.instructionPreparationRunId = input.runId;
+    if (entry.closeOnReleaseReason !== undefined || preparationChanged || await adopted.hasChanges()) {
       await retire("warm instruction materialization changed before preparation");
       return null;
     }
-    const adopted = await input.adopt(entry.instructionWorkingCopy.runId);
-    if (!adopted) {
-      await retire("warm instruction materialization cannot be adopted");
-      return null;
-    }
-    entry.instructionWorkingCopy = adopted;
-    entry.instructionPreparationRunId = input.runId;
     return async () => {
       if (entry.instructionPreparationRunId === input.runId) await retire("warm instruction preparation abandoned");
     };
@@ -1556,7 +1563,7 @@ function nativeSessionKey(execution: NativeExecutionInput): string {
   );
 }
 
-function nativeSessionWorkspaceScope(execution: NativeExecutionInput) {
+export function nativeSessionWorkspaceScope(execution: { binding: Pick<NativeExecutionInput["binding"], "runId" | "executionWorkspaceId">; workspace: Pick<NativeExecutionInput["workspace"], "cwd" | "repoUrl" | "repoRef" | "branchName"> }) {
   // Projectless local runs use the heartbeat run id as a durable placeholder
   // rather than fabricating an execution_workspaces row. Do not let that
   // per-run placeholder break continuity for the same provider session; the