Finish stop-proved recovery of retained agent directories

Mark lost remote unchanged-turn copies unavailable only after exact termination proof, then perform owned cleanup. Preserve the live unchanged-turn guard and document warm ownership and crash preservation boundaries.

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

doc/agent-files.md

@@ -78,8 +78,10 @@ from the current browser edit. Later successful saves do not erase run history.
 Larger folders take longer to hash, copy, and transfer on each run. There is one
 canonical folder plus temporary working copies for currently active runs (and
 remote staging when the transport needs it). No additional captured tree is
-created. Terminal runs remove their private trees and baseline metadata, keeping
-only a small receipt. Restart recovery retries interrupted cleanup without
+created. After verified provider retirement and collection, runs remove their
+private trees and baseline metadata, keeping only a small receipt. A successful
+warm turn can retain its unchanged copy with the still-running provider. Restart
+recovery retries interrupted cleanup without
 removing a running provider's files. These are not aggregate disk quotas; the
 operator still provisions storage for agents and the configured run concurrency.
 
@@ -101,10 +103,23 @@ operator still provisions storage for agents and the configured run concurrency.
    runs. No per-run file versions, conflict copies, or review queue accumulate.
    The next run starts with the current directory.
 
-The whole-directory contract closes the provider process to establish a safe
-collection boundary, including child processes. It preserves the provider's
-resumable conversation. Only the loaded instruction entry participates in the
-new runtime instruction digest; adding or editing another file does not change
+A successful warm turn may retain its complete, unchanged materialized directory
+and the exact `AGENT_HOME` root with the same provider process. A bounded read-only
+probe must verify the full baseline and root identity without unsafe paths or
+concurrent changes. Local hashing runs in an isolated child; remote observation
+runs at the registered remote root. A failed or uncertain probe requires stopped
+collection. The next authorized run claims that same materialization within the
+company, agent, workspace, environment and configuration scope. The prior run
+loses collection authority; stale cleanup cannot remove the current owner's root.
+
+Edits, additions, removals or changed configuration retire the owner before
+collection and fresh preparation. A configuration or copy change found at the
+final dispatch fence fails that run after retirement; it does not replay the
+request automatically. Idle expiry, restart and abandoned preparation also retire
+before collection. The whole-directory contract closes the provider process,
+including child processes, to establish this safe collection boundary. It
+preserves the provider's resumable conversation. Only the loaded instruction entry
+participates in the new runtime instruction digest; adding or editing another file does not change
 that digest. Relative supporting files are read from `AGENT_HOME`, not from the
 read-only prompt snapshot.
 
@@ -155,12 +170,24 @@ save receipt. An interrupted apply can replay its changed files with the same
 last-sync-wins rule. Cleanup resumes for terminal runs; no copy is retained as
 an archive after cleanup succeeds.
 
+An unchanged-turn observation is neither a save nor proof that a provider stopped.
+If retirement fails, the controller records unresolved ownership and preserves
+the materialized root. A crash after ownership transfer can leave the new run
+without independently recorded stop proof; terminal run status or the old run's
+alias is not enough to collect that root. Recovery leaves it preserved until the
+required proof is available. For a remote run with an exact termination receipt
+but unretrievable bytes, recovery records an explicit unavailable/no-save outcome
+and completes owned cleanup without restarting a provider.
+
 ## Verification
 
 `agent-directory-working-copies.test.ts` exercises nested/binary files, directory
 isolation, last-sync-wins edits and deletions, terminal cleanup, link rejection, old-head
-adoption, and stable prompt digests. The legacy working-copy and native-tool
-suites exercise compatibility. Workspace merge tests exercise preflight and
+adoption, stable prompt digests, warm ownership transfer, concurrent stale claims,
+and stop-proved recovery. `agent-directory-probe.test.ts` covers stable snapshots,
+unsafe paths, mutation races and bounded child failures; composed native executor
+tests cover retained roots and retirement before collection. The legacy
+working-copy and native-tool suites exercise compatibility. Workspace merge tests exercise preflight and
 interrupted replay.
 
 The explicit Product E2E `instruction-persistence` suite creates a file through

server/src/__tests__/agent-directory-working-copies.test.ts

@@ -540,6 +540,35 @@ describe("persistent agent directories", () => {
     expect(execute).toHaveBeenCalledTimes(2);
   });
 
+  it.each([false, true])("recovers a retained remote unchanged turn only with exact stop proof: %s", async stopped => {
+    const copy = await run();
+    const environmentId = randomUUID(), leaseId = randomUUID(), remoteCwd = "/fixture/task";
+    const lease = { id: leaseId, companyId, environmentId, heartbeatRunId: copy.runId, provider: "daytona", providerLeaseId: "retained-sandbox" };
+    await db.insert(environments).values({ id: environmentId, name: environmentId, driver: "sandbox" });
+    await db.insert(environmentLeases).values({ ...lease, status: "released", releasedAt: new Date(), cleanupStatus: "success",
+      metadata: stopped ? { remoteExecutionTermination: remoteTerminationReceipt(lease, { providerLeaseId: lease.providerLeaseId, state: "destroyed" }) } : {} });
+    await db.update(heartbeatRuns).set({ status: "succeeded", runtimeMode: "native" }).where(eq(heartbeatRuns.id, copy.runId));
+    await db.update(agentInstructionWorkingCopies).set({ state: "unchanged_turn", location: `remote:${environmentId}`,
+      executionRoot: path.posix.join(remoteCwd, ".paperclip-runtime", "agent-files", agentId, copy.runId),
+      receipt: { ...copy.receipt, cleanup: { leaseId, remoteCwd } } }).where(eq(agentInstructionWorkingCopies.runId, copy.runId));
+    const execute = vi.fn();
+    copies = agentInstructionWorkingCopyService(db, { environmentRuntime: { execute } as unknown as EnvironmentRuntimeService });
+    expect((await copies.reportUnavailable(companyId, copy.runId))?.state).toBe("unchanged_turn");
+    await copies.recoverStopped();
+    const recovered = (await copies.get(companyId, copy.runId))!;
+    if (stopped) {
+      expect(recovered).toMatchObject({ state: "unavailable", errorCode: "INSTRUCTION_COLLECTION_UNAVAILABLE", receipt: { cleanupPending: false } });
+      expect(recovered.processStoppedAt).toBeInstanceOf(Date);
+      await expect(fs.stat(copy.localRoot)).rejects.toMatchObject({ code: "ENOENT" });
+      await copies.recoverStopped();
+      expect((await copies.get(companyId, copy.runId))?.updatedAt).toEqual(recovered.updatedAt);
+    } else {
+      expect(recovered).toMatchObject({ state: "unchanged_turn", processStoppedAt: null });
+      expect(await fs.readFile(path.join(copy.localRoot, entryFile), "utf8")).toBe(initial);
+    }
+    expect(execute).not.toHaveBeenCalled(); // Never restart a remote provider to recover bytes.
+  });
+
   it("finishes remote cleanup from a destruction receipt after the environment is deleted", async () => {
     const copy = await run();
     const environmentId = randomUUID(), leaseId = randomUUID(), remoteCwd = "/fixture/task";

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

@@ -311,7 +311,13 @@ 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)) {
-        const unavailable = await reportUnavailable(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() }));
         }