fix(cloud-workers): reclaim sessions after workspace conflicts (#111714)

* fix(cloud-workers): make reclaim failures retryable * chore: leave changelog to release flow * chore: fix cloud reclaim lint * fix(cloud-workers): preserve committed reclaim recovery
2026-07-21 02:06:43 +00:00 · 2026-07-20 01:16:07 -07:00
parent 83e78d8681
commit 6eee1dccfc
6 changed files with 510 additions and 187 deletions
@@ -30,6 +30,7 @@ export type WorkerDispatchPlacementStore = Pick<
  ReturnType<typeof createWorkerSessionPlacementStore>,
  | "adoptActive"
  | "acceptIdleWorkspaceReconciliation"
+  | "claimReclaimWorkspaceResult"
  | "claimTurn"
  | "fail"
  | "finishReclaim"
@@ -45,6 +46,7 @@ export type WorkerDispatchPlacementStore = Pick<
  | "recordStagedWorkspaceResult"
  | "recordWorkspaceResultConflict"
  | "acceptWorkspaceResult"
+  | "cancelWorkspaceResultAndReleaseTurn"
  | "completeWorkspaceResultAndReleaseTurn"
  | "abandonWorkspaceResult"
  | "listForReconcile"
@@ -1,7 +1,9 @@
 import fs from "node:fs/promises";
 import os from "node:os";
+import path from "node:path";
 import { afterEach, beforeEach, describe, expect, it } from "vitest";
 import { useAutoCleanupTempDirTracker } from "../../../test/helpers/temp-dir.js";
+import { runCommandWithTimeout } from "../../process/exec.js";
 import {
  closeOpenClawStateDatabaseForTest,
  openOpenClawStateDatabase,
@@ -127,4 +129,137 @@ describe("worker placement dispatch reclaim", () => {
    expect(placementStore.listPendingWorkspaceResults()).toEqual([]);
    expect(harness.environments.destroy).toHaveBeenCalledOnce();
  });
+
+  it("reclaims an unchanged worker without clearing a retained keep-local conflict", async () => {
+    const priorConflict = {
+      paths: ["notes.md"],
+      stagedResultRef: "refs/openclaw/worker-results/prior-conflict",
+    };
+    const harness = createHarness(placementStore, {
+      priorWorkspaceResultConflict: priorConflict,
+      reconcileChanged: false,
+      reconcileCommitsManifest: false,
+    });
+    await harness.service.dispatch(REQUEST);
+
+    await expect(
+      harness.service.reclaim({
+        sessionId: REQUEST.sessionId,
+        sessionKey: REQUEST.sessionKey,
+        agentId: REQUEST.agentId,
+      }),
+    ).resolves.toMatchObject({
+      state: "reclaimed",
+      workspaceBaseManifestRef: MANIFEST_REF,
+    });
+
+    expect(harness.placements.current()).toMatchObject({ workspaceResultConflict: priorConflict });
+    expect(harness.reportWorkspaceResultConflict).not.toHaveBeenCalled();
+    expect(placementStore.listPendingWorkspaceResults()).toEqual([]);
+    expect(harness.environments.destroy).toHaveBeenCalledOnce();
+  });
+
+  it("applies a prepared staged result before requiring its manifest commit", async () => {
+    const harness = createHarness(placementStore, {
+      reconcileCommitsManifest: false,
+      reconcileCommitsManifestOnApply: true,
+    });
+    await harness.service.dispatch(REQUEST);
+
+    await expect(
+      harness.service.reclaim({
+        sessionId: REQUEST.sessionId,
+        sessionKey: REQUEST.sessionKey,
+        agentId: REQUEST.agentId,
+      }),
+    ).resolves.toMatchObject({
+      state: "reclaimed",
+      workspaceBaseManifestRef: harness.reconciledManifestRef,
+    });
+
+    expect(harness.log).toContain("workspace:apply-prepared");
+  });
+
+  it("claims and cancels a reclaim workspace result atomically", async () => {
+    const harness = createHarness(placementStore);
+    const active = await harness.service.dispatch(REQUEST);
+    const claim = placementStore.claimReclaimWorkspaceResult({
+      ...REQUEST,
+      owner: {
+        kind: "worker",
+        environmentId: active.environmentId,
+        ownerEpoch: active.activeOwnerEpoch,
+      },
+      claimId: "reclaim-atomic",
+      runId: "reclaim-atomic",
+    });
+
+    expect(placementStore.get(active.sessionId)?.turnClaim).toMatchObject({
+      claimId: claim.claimId,
+    });
+    expect(placementStore.listPendingWorkspaceResults()).toMatchObject([
+      { sessionId: active.sessionId, claimId: claim.claimId },
+    ]);
+
+    expect(placementStore.cancelWorkspaceResultAndReleaseTurn(claim)).toMatchObject({
+      turnClaim: null,
+    });
+    expect(placementStore.listPendingWorkspaceResults()).toEqual([]);
+  });
+
+  it("releases a failed stop claim so reclaim can be retried", async () => {
+    const workspacePath = path.join(root, "retry-workspace");
+    await fs.mkdir(workspacePath);
+    const initialized = await runCommandWithTimeout(
+      ["git", "-C", workspacePath, "init", "--quiet"],
+      { timeoutMs: 10_000 },
+    );
+    expect(initialized.code).toBe(0);
+    const harness = createHarness(placementStore, {
+      reconcileFailureCount: 1,
+      workspacePath,
+    });
+    await harness.service.dispatch(REQUEST);
+    const request = {
+      sessionId: REQUEST.sessionId,
+      sessionKey: REQUEST.sessionKey,
+      agentId: REQUEST.agentId,
+    };
+
+    await expect(harness.service.reclaim(request)).rejects.toThrow("workspace conflict");
+    expect(harness.placements.current()).toMatchObject({ state: "active", turnClaim: null });
+    expect(placementStore.listPendingWorkspaceResults()).toEqual([]);
+
+    await expect(harness.service.reclaim(request)).resolves.toMatchObject({ state: "reclaimed" });
+    expect(harness.environments.destroy).toHaveBeenCalledOnce();
+  });
+
+  it("keeps a committed failed stop result fenced for recovery", async () => {
+    const priorConflict = {
+      paths: ["notes.md"],
+      stagedResultRef: "refs/openclaw/worker-results/prior-conflict",
+    };
+    const harness = createHarness(placementStore, {
+      priorWorkspaceResultConflict: priorConflict,
+      verifyFails: true,
+    });
+    await harness.service.dispatch(REQUEST);
+
+    await expect(
+      harness.service.reclaim({
+        sessionId: REQUEST.sessionId,
+        sessionKey: REQUEST.sessionKey,
+        agentId: REQUEST.agentId,
+      }),
+    ).rejects.toThrow("workspace changed after reconciliation");
+
+    expect(harness.placements.current()).toMatchObject({
+      state: "active",
+      workspaceBaseManifestRef: harness.reconciledManifestRef,
+      turnClaim: { owner: "worker" },
+    });
+    expect(placementStore.listPendingWorkspaceResults()).toMatchObject([
+      { workspaceAcceptedAtMs: null, stagedResultRef: null },
+    ]);
+  });
 });
@@ -25,6 +25,10 @@ export function createHarness(
    destroyFails?: boolean;
    claimOnDrain?: boolean;
    reconcileFails?: boolean;
+    reconcileFailureCount?: number;
+    reconcileChanged?: boolean;
+    reconcileCommitsManifest?: boolean;
+    reconcileCommitsManifestOnApply?: boolean;
    verifyFails?: boolean;
    leaseFails?: boolean;
    localVerifyFails?: boolean;
@@ -35,6 +39,7 @@ export function createHarness(
  } = {},
 ) {
  const reconciledManifestRef = MANIFEST_REF.replaceAll("b", "c");
+  let remainingReconcileFailures = options.reconcileFailureCount ?? 0;
  const log: string[] = [];
  const reportWorkspaceResultConflict = vi.fn(async () => {});
  const fail = (stage: DispatchStage) => {
@@ -57,8 +62,11 @@ export function createHarness(
    recordWorkspaceResultConflict: (claim, conflict) =>
      placementStore.recordWorkspaceResultConflict(claim, conflict),
    claimTurn: (params) => placementStore.claimTurn(params),
+    claimReclaimWorkspaceResult: (params) => placementStore.claimReclaimWorkspaceResult(params),
    markWorkspaceResultPending: (claim) => placementStore.markWorkspaceResultPending(claim),
    acceptWorkspaceResult: (claim) => placementStore.acceptWorkspaceResult(claim),
+    cancelWorkspaceResultAndReleaseTurn: (claim) =>
+      placementStore.cancelWorkspaceResultAndReleaseTurn(claim),
    completeWorkspaceResultAndReleaseTurn: (claim, completionOptions) => {
      const completed = placementStore.completeWorkspaceResultAndReleaseTurn(
        claim,
@@ -159,16 +167,19 @@ export function createHarness(
    }),
    reconcileWorkspace: vi.fn(async (request) => {
      log.push("workspace:reconcile");
-      if (options.reconcileFails) {
+      if (options.reconcileFails || remainingReconcileFailures > 0) {
+        remainingReconcileFailures -= 1;
        throw new Error("workspace conflict");
      }
-      request.journal.commit(reconciledManifestRef);
+      if (options.reconcileCommitsManifest !== false) {
+        request.journal.commit(reconciledManifestRef);
+      }
      if (options.reconcileConflictPaths?.length && request.stagedResult) {
        request.stagedResult.record(request.stagedResult.ref);
      }
      return {
        manifestRef: reconciledManifestRef,
-        changed: true,
+        changed: options.reconcileChanged ?? true,
        verifyStable: async () => {
          log.push("workspace:verify");
          if (options.verifyFails) {
@@ -189,6 +200,15 @@ export function createHarness(
              verifyLocalStable: async () => {},
            })
          : undefined,
+        ...(options.reconcileCommitsManifestOnApply
+          ? {
+              applyPreparedStagedResult: async () => {
+                log.push("workspace:apply-prepared");
+                request.journal.commit(reconciledManifestRef);
+              },
+              publishStagedResult: async () => {},
+            }
+          : {}),
      };
    }),
    runWorkspaceCommand: vi.fn(async () => ({
@@ -65,8 +65,11 @@ function createHarness(
    recordWorkspaceResultConflict: (claim, conflict) =>
      placementStore.recordWorkspaceResultConflict(claim, conflict),
    claimTurn: (params) => placementStore.claimTurn(params),
+    claimReclaimWorkspaceResult: (params) => placementStore.claimReclaimWorkspaceResult(params),
    markWorkspaceResultPending: (claim) => placementStore.markWorkspaceResultPending(claim),
    acceptWorkspaceResult: (claim) => placementStore.acceptWorkspaceResult(claim),
+    cancelWorkspaceResultAndReleaseTurn: (claim) =>
+      placementStore.cancelWorkspaceResultAndReleaseTurn(claim),
    completeWorkspaceResultAndReleaseTurn: (claim, completionOptions) => {
      const completed = placementStore.completeWorkspaceResultAndReleaseTurn(
        claim,
@@ -678,7 +681,11 @@ describe("worker placement dispatch", () => {
    expect(harness.placements.current()).toMatchObject({
      state: "active",
      workspaceBaseManifestRef: harness.reconciledManifestRef,
+      turnClaim: { owner: "worker" },
    });
+    expect(placementStore.listPendingWorkspaceResults()).toMatchObject([
+      { workspaceAcceptedAtMs: expect.any(Number) },
+    ]);
    expect(harness.log).not.toContain("placement:draining");
    expect(harness.log).toContain("workspace:resume");
  });
@@ -23,7 +23,9 @@ import type { WorkerWorkspaceOperationCoordinator } from "./workspace-operation-
 import { recoverWorkerWorkspaceReconciliation } from "./workspace-reconcile.js";
 import {
  deleteStagedWorkerWorkspaceResult,
+  hasWorkerWorkspaceResultRef,
  moveStagedWorkerWorkspaceResultToCleanup,
+  preparedWorkerWorkspaceResultRef,
  workerWorkspaceResultRef,
 } from "./workspace-result-staging.js";

@@ -236,7 +238,7 @@ export function createWorkerPlacementDispatchService(options: WorkerPlacementDis
          placementGeneration: current.generation,
        };
        const reclaimClaimId = `reclaim-${randomUUID()}`;
-        const reclaimClaim = placements.claimTurn({
+        const reclaimClaim = placements.claimReclaimWorkspaceResult({
          sessionId: current.sessionId,
          sessionKey: current.sessionKey,
          agentId: current.agentId,
@@ -248,7 +250,7 @@ export function createWorkerPlacementDispatchService(options: WorkerPlacementDis
            ownerEpoch: current.activeOwnerEpoch,
          },
        });
-        placements.markWorkspaceResultPending(reclaimClaim);
+        const reclaimResultRef = workerWorkspaceResultRef(reclaimClaim.claimId);
        let manifestAccepted = false;
        const journal = {
          load: () => placements.loadWorkspaceReconciliation(journalOwner),
@@ -263,133 +265,200 @@ export function createWorkerPlacementDispatchService(options: WorkerPlacementDis
          },
          abort: () => placements.abortWorkspaceReconciliation(journalOwner),
        };
-        const pending = journal.load();
-        if (pending) {
-          await recoverWorkerWorkspaceReconciliation({ root: localPath, journal: pending });
-          journal.abort();
-        }
-        const tunnel = await environments.startTunnel({
-          environmentId: current.environmentId,
-          ownerEpoch: current.activeOwnerEpoch,
-        });
-        const reclaimed = await options.workspaceOperations.run(current.environmentId, async () => {
-          const owned = placements.get(current.sessionId);
-          if (
-            owned?.state !== "active" ||
-            owned.generation !== current.generation ||
-            owned.environmentId !== current.environmentId ||
-            owned.activeOwnerEpoch !== current.activeOwnerEpoch ||
-            owned.turnClaim?.claimId !== reclaimClaim.claimId
-          ) {
-            throw new Error("Cloud worker stop lost its placement owner before reconciliation");
-          }
-          const quiescence = await tunnel.quiesceWorkspace(current.remoteWorkspaceDir);
-          let destroyed = false;
-          try {
-            const reconciliation = await tunnel.reconcileWorkspace({
-              localPath,
-              remoteWorkspaceDir: current.remoteWorkspaceDir,
-              baseManifestRef: current.workspaceBaseManifestRef,
-              journal,
-              stagedResult: {
-                ref: workerWorkspaceResultRef(reclaimClaim.claimId),
-                record: (ref) => placements.recordStagedWorkspaceResult(reclaimClaim, ref),
-              },
-            });
-            if (!manifestAccepted) {
-              throw new Error("Cloud worker stop did not commit its reconciled workspace");
-            }
-            const applied = await verifyReconciledWorkspaceFinal(reconciliation, quiescence);
-            placements.acceptWorkspaceResult(reclaimClaim);
-            const recordedStagedResultRef = placements
-              .listPendingWorkspaceResults()
-              .find(
-                (result) =>
-                  result.sessionId === reclaimClaim.sessionId &&
-                  result.claimId === reclaimClaim.claimId &&
-                  result.runId === reclaimClaim.runId,
-              )?.stagedResultRef;
-            const conflictPaths = applied?.conflictPaths ?? [];
-            if (conflictPaths.length > 0 && !recordedStagedResultRef) {
-              throw new Error("Cloud worker stop conflict has no staged result reference");
-            }
-            const priorWorkspaceResultConflict =
-              current.workspaceResultConflict ??
-              (await options.resolveWorkspaceResultConflict({
-                sessionId: current.sessionId,
-                sessionKey: current.sessionKey,
-                agentId: current.agentId,
-              }));
-            const supersededConflict =
-              priorWorkspaceResultConflict &&
-              (conflictPaths.length === 0 ||
-                priorWorkspaceResultConflict.stagedResultRef !== recordedStagedResultRef)
-                ? priorWorkspaceResultConflict
-                : undefined;
-            if (
-              supersededConflict &&
-              supersededConflict.stagedResultRef !== recordedStagedResultRef
-            ) {
-              await deleteStagedWorkerWorkspaceResult({
-                root: localPath,
-                stagedResultRef: supersededConflict.stagedResultRef,
-              });
-            }
-            if (conflictPaths.length > 0 && recordedStagedResultRef) {
-              const projectedConflict = projectWorkspaceResultConflict(
-                conflictPaths,
-                recordedStagedResultRef,
+        const cancelEmptyFailedReclaim = async (): Promise<void> => {
+          await options.workspaceOperations.run(current.environmentId, async () => {
+            const stillOwnsEmptyResult = (): boolean => {
+              const owned = placements.get(current.sessionId);
+              const currentEnvironment = environments.get(current.environmentId);
+              const pendingResult = placements
+                .listPendingWorkspaceResults()
+                .find(
+                  (pending) =>
+                    pending.sessionId === reclaimClaim.sessionId &&
+                    pending.claimId === reclaimClaim.claimId &&
+                    pending.runId === reclaimClaim.runId,
+                );
+              return (
+                !manifestAccepted &&
+                owned?.state === "active" &&
+                owned.turnClaim?.claimId === reclaimClaim.claimId &&
+                reclaimClaim.owner.kind === "worker" &&
+                currentEnvironment?.state === "attached" &&
+                currentEnvironment.ownerEpoch === reclaimClaim.owner.ownerEpoch &&
+                currentEnvironment.attachedSessionIds.length === 1 &&
+                currentEnvironment.attachedSessionIds[0] === owned.sessionId &&
+                pendingResult?.workspaceAcceptedAtMs === null &&
+                pendingResult.stagedResultRef === null
              );
-              placements.recordWorkspaceResultConflict(reclaimClaim, projectedConflict);
-              await options.reportWorkspaceResultConflict({
-                sessionId: current.sessionId,
-                sessionKey: current.sessionKey,
-                agentId: current.agentId,
-                ...projectedConflict,
-              });
-            } else if (supersededConflict) {
-              placements.recordWorkspaceResultConflict(reclaimClaim, undefined);
-              await options.reportWorkspaceResultConflict({
-                sessionId: current.sessionId,
-                sessionKey: current.sessionKey,
-                agentId: current.agentId,
-                cleared: true,
-              });
+            };
+            if (!stillOwnsEmptyResult()) {
+              return;
            }
-            const cleanupRef =
-              recordedStagedResultRef && conflictPaths.length === 0
-                ? await moveStagedWorkerWorkspaceResultToCleanup({
-                    root: localPath,
-                    stagedResultRef: recordedStagedResultRef,
-                  })
-                : undefined;
-            await environments.destroy(current.environmentId);
-            destroyed = true;
-            const completed = placements.completeWorkspaceResultAndReleaseTurn(reclaimClaim, {
-              reclaim: true,
-            });
-            if (completed.state !== "reclaimed") {
-              throw new Error("Cloud worker stop did not produce a reclaimed placement");
-            }
-            if (cleanupRef) {
-              await deleteStagedWorkerWorkspaceResult({
+            const [canonicalExists, preparedExists] = await Promise.all([
+              hasWorkerWorkspaceResultRef({ root: localPath, stagedResultRef: reclaimResultRef }),
+              hasWorkerWorkspaceResultRef({
                root: localPath,
-                stagedResultRef: cleanupRef,
-              }).catch(() => undefined);
-            }
-            return completed;
-          } finally {
-            if (!destroyed) {
-              await quiescence.resume();
+                stagedResultRef: preparedWorkerWorkspaceResultRef(reclaimResultRef),
+              }),
+            ]);
+            // Recheck after filesystem I/O while the session barrier and workspace
+            // owner lock are still held. A committed manifest or durable ref keeps
+            // recovery authoritative.
+            if (!canonicalExists && !preparedExists && stillOwnsEmptyResult()) {
+              placements.cancelWorkspaceResultAndReleaseTurn(reclaimClaim);
            }
+          });
+        };
+        const finishReclaim = async (): Promise<WorkerReclaimedPlacement> => {
+          const pending = journal.load();
+          if (pending) {
+            await recoverWorkerWorkspaceReconciliation({ root: localPath, journal: pending });
+            journal.abort();
          }
-        });
+          const tunnel = await environments.startTunnel({
+            environmentId: current.environmentId,
+            ownerEpoch: current.activeOwnerEpoch,
+          });
+          const reclaimed = await options.workspaceOperations.run(
+            current.environmentId,
+            async () => {
+              const owned = placements.get(current.sessionId);
+              if (
+                owned?.state !== "active" ||
+                owned.generation !== current.generation ||
+                owned.environmentId !== current.environmentId ||
+                owned.activeOwnerEpoch !== current.activeOwnerEpoch ||
+                owned.turnClaim?.claimId !== reclaimClaim.claimId
+              ) {
+                throw new Error("Cloud worker stop lost its placement owner before reconciliation");
+              }
+              const quiescence = await tunnel.quiesceWorkspace(current.remoteWorkspaceDir);
+              let destroyed = false;
+              try {
+                const reconciliation = await tunnel.reconcileWorkspace({
+                  localPath,
+                  remoteWorkspaceDir: current.remoteWorkspaceDir,
+                  baseManifestRef: current.workspaceBaseManifestRef,
+                  journal,
+                  stagedResult: {
+                    ref: reclaimResultRef,
+                    record: (ref) => placements.recordStagedWorkspaceResult(reclaimClaim, ref),
+                  },
+                });
+                const applied = await verifyReconciledWorkspaceFinal(reconciliation, quiescence);
+                if (reconciliation.changed && !manifestAccepted) {
+                  throw new Error("Cloud worker stop did not commit its reconciled workspace");
+                }
+                placements.acceptWorkspaceResult(reclaimClaim);
+                const recordedStagedResultRef = placements
+                  .listPendingWorkspaceResults()
+                  .find(
+                    (result) =>
+                      result.sessionId === reclaimClaim.sessionId &&
+                      result.claimId === reclaimClaim.claimId &&
+                      result.runId === reclaimClaim.runId,
+                  )?.stagedResultRef;
+                const conflictPaths = applied?.conflictPaths ?? [];
+                if (conflictPaths.length > 0 && !recordedStagedResultRef) {
+                  throw new Error("Cloud worker stop conflict has no staged result reference");
+                }
+                const priorWorkspaceResultConflict =
+                  current.workspaceResultConflict ??
+                  (await options.resolveWorkspaceResultConflict({
+                    sessionId: current.sessionId,
+                    sessionKey: current.sessionKey,
+                    agentId: current.agentId,
+                  }));
+                const retainedWorkspaceResultConflict =
+                  !reconciliation.changed && conflictPaths.length === 0
+                    ? priorWorkspaceResultConflict
+                    : undefined;
+                const supersededConflict =
+                  priorWorkspaceResultConflict &&
+                  !retainedWorkspaceResultConflict &&
+                  (conflictPaths.length === 0 ||
+                    priorWorkspaceResultConflict.stagedResultRef !== recordedStagedResultRef)
+                    ? priorWorkspaceResultConflict
+                    : undefined;
+                if (
+                  supersededConflict &&
+                  supersededConflict.stagedResultRef !== recordedStagedResultRef
+                ) {
+                  await deleteStagedWorkerWorkspaceResult({
+                    root: localPath,
+                    stagedResultRef: supersededConflict.stagedResultRef,
+                  });
+                }
+                if (conflictPaths.length > 0 && recordedStagedResultRef) {
+                  const projectedConflict = projectWorkspaceResultConflict(
+                    conflictPaths,
+                    recordedStagedResultRef,
+                  );
+                  placements.recordWorkspaceResultConflict(reclaimClaim, projectedConflict);
+                  await options.reportWorkspaceResultConflict({
+                    sessionId: current.sessionId,
+                    sessionKey: current.sessionKey,
+                    agentId: current.agentId,
+                    ...projectedConflict,
+                  });
+                } else if (retainedWorkspaceResultConflict) {
+                  // Stopping an unchanged worker is not a later cloud result. Keep the
+                  // already resolved keep-local fence inspectable after placement teardown.
+                  placements.recordWorkspaceResultConflict(
+                    reclaimClaim,
+                    retainedWorkspaceResultConflict,
+                  );
+                } else if (supersededConflict) {
+                  placements.recordWorkspaceResultConflict(reclaimClaim, undefined);
+                  await options.reportWorkspaceResultConflict({
+                    sessionId: current.sessionId,
+                    sessionKey: current.sessionKey,
+                    agentId: current.agentId,
+                    cleared: true,
+                  });
+                }
+                const cleanupRef =
+                  recordedStagedResultRef && conflictPaths.length === 0
+                    ? await moveStagedWorkerWorkspaceResultToCleanup({
+                        root: localPath,
+                        stagedResultRef: recordedStagedResultRef,
+                      })
+                    : undefined;
+                await environments.destroy(current.environmentId);
+                destroyed = true;
+                const completed = placements.completeWorkspaceResultAndReleaseTurn(reclaimClaim, {
+                  reclaim: true,
+                });
+                if (completed.state !== "reclaimed") {
+                  throw new Error("Cloud worker stop did not produce a reclaimed placement");
+                }
+                if (cleanupRef) {
+                  await deleteStagedWorkerWorkspaceResult({
+                    root: localPath,
+                    stagedResultRef: cleanupRef,
+                  }).catch(() => undefined);
+                }
+                return completed;
+              } finally {
+                if (!destroyed) {
+                  await quiescence.resume();
+                }
+              }
+            },
+          );
+          try {
+            await environments.stopTunnel(current.environmentId, current.activeOwnerEpoch);
+          } catch {
+            // Provider teardown is authoritative; local tunnel cleanup is best effort.
+          }
+          return reclaimed;
+        };
        try {
-          await environments.stopTunnel(current.environmentId, current.activeOwnerEpoch);
-        } catch {
-          // Provider teardown is authoritative; local tunnel cleanup is best effort.
+          return await finishReclaim();
+        } catch (error) {
+          await cancelEmptyFailedReclaim().catch(() => undefined);
+          throw error;
        }
-        return reclaimed;
      },
    });

@@ -26,6 +26,11 @@ import {
 } from "./workspace-reconcile.js";

 type TurnClaimReleaseWaiter = () => void;
+type WorkerTurnClaimInput = WorkerSessionPlacementIdentity & {
+  owner: WorkerSessionTurnOwner;
+  claimId: string;
+  runId: string;
+};
 const turnClaimReleaseWaiters = new Map<string, Map<string, Set<TurnClaimReleaseWaiter>>>();
 const workspaceJournalQuery = (db: DatabaseSync) =>
  getNodeSqliteKysely<Pick<StateDatabase, "worker_workspace_reconciliations">>(db);
@@ -61,71 +66,84 @@ export function signalTurnClaimRelease(path: string, sessionId: string): void {

 export function createPlacementTurnClaimOps(runtime: PlacementStoreRuntime) {
  const { instanceId, path, now, read, write } = runtime;
+  const claimTurnInDatabase = (
+    db: DatabaseSync,
+    input: WorkerTurnClaimInput,
+    updatedAtMs: number,
+  ): WorkerSessionTurnClaim => {
+    const identity = normalizeIdentity(input);
+    const claimId = required(input.claimId, "turn claim id");
+    const runId = required(input.runId, "turn claim run id");
+    const owner: WorkerSessionTurnOwner =
+      input.owner.kind === "local"
+        ? { kind: "local" }
+        : {
+            kind: "worker",
+            environmentId: required(input.owner.environmentId, "turn owner environment id"),
+            ownerEpoch: normalizeEpoch(input.owner.ownerEpoch, "turn owner epoch"),
+          };
+    const current = ensureLocal(db, identity, updatedAtMs);
+    if (current.turnClaim) {
+      throw new Error(`Session ${identity.sessionId} already has an active turn claim`);
+    }
+    if (owner.kind === "local") {
+      if (current.state !== "local") {
+        throw new Error(
+          `Local turn rejected for session ${identity.sessionId} in placement ${current.state}`,
+        );
+      }
+    } else if (
+      current.state !== "active" ||
+      current.environmentId !== owner.environmentId ||
+      current.activeOwnerEpoch !== owner.ownerEpoch
+    ) {
+      throw new Error(`Worker turn rejected for session ${identity.sessionId}: stale owner`);
+    }
+    const result = executeSqliteQuerySync(
+      db,
+      query(db)
+        .updateTable("worker_session_placements")
+        .set({
+          turn_claim_owner: owner.kind,
+          turn_claim_id: claimId,
+          turn_claim_run_id: runId,
+          turn_claim_generation: current.generation,
+          turn_claim_owner_epoch: owner.kind === "worker" ? owner.ownerEpoch : null,
+          updated_at_ms: updatedAtMs,
+        })
+        .where("session_id", "=", current.sessionId)
+        .where("state", "=", current.state)
+        .where("transition_generation", "=", current.generation)
+        .where("turn_claim_owner", "is", null),
+    );
+    if (result.numAffectedRows !== 1n) {
+      throw new Error(`Session ${identity.sessionId} placement changed during turn admission`);
+    }
+    return {
+      sessionId: current.sessionId,
+      claimId,
+      runId,
+      placementGeneration: current.generation,
+      owner,
+    };
+  };

  return {
-    claimTurn(
-      input: WorkerSessionPlacementIdentity & {
-        owner: WorkerSessionTurnOwner;
-        claimId: string;
-        runId: string;
-      },
-    ): WorkerSessionTurnClaim {
-      const identity = normalizeIdentity(input);
-      const claimId = required(input.claimId, "turn claim id");
-      const runId = required(input.runId, "turn claim run id");
-      const owner: WorkerSessionTurnOwner =
-        input.owner.kind === "local"
-          ? { kind: "local" }
-          : {
-              kind: "worker",
-              environmentId: required(input.owner.environmentId, "turn owner environment id"),
-              ownerEpoch: normalizeEpoch(input.owner.ownerEpoch, "turn owner epoch"),
-            };
+    claimTurn(input: WorkerTurnClaimInput): WorkerSessionTurnClaim {
+      return write((db) => claimTurnInDatabase(db, input, now()));
+    },
+
+    claimReclaimWorkspaceResult(input: WorkerTurnClaimInput): WorkerSessionTurnClaim {
+      if (input.claimId !== input.runId || !input.claimId.startsWith("reclaim-")) {
+        throw new Error(`Session ${input.sessionId} workspace result is not owned by reclaim`);
+      }
+      // Admission and its recovery fence are inseparable. A crash after this
+      // transaction leaves startup recovery enough state to finish or abandon it.
      return write((db) => {
-        const current = ensureLocal(db, identity, now());
-        if (current.turnClaim) {
-          throw new Error(`Session ${identity.sessionId} already has an active turn claim`);
-        }
-        if (owner.kind === "local") {
-          if (current.state !== "local") {
-            throw new Error(
-              `Local turn rejected for session ${identity.sessionId} in placement ${current.state}`,
-            );
-          }
-        } else if (
-          current.state !== "active" ||
-          current.environmentId !== owner.environmentId ||
-          current.activeOwnerEpoch !== owner.ownerEpoch
-        ) {
-          throw new Error(`Worker turn rejected for session ${identity.sessionId}: stale owner`);
-        }
-        const result = executeSqliteQuerySync(
-          db,
-          query(db)
-            .updateTable("worker_session_placements")
-            .set({
-              turn_claim_owner: owner.kind,
-              turn_claim_id: claimId,
-              turn_claim_run_id: runId,
-              turn_claim_generation: current.generation,
-              turn_claim_owner_epoch: owner.kind === "worker" ? owner.ownerEpoch : null,
-              updated_at_ms: now(),
-            })
-            .where("session_id", "=", current.sessionId)
-            .where("state", "=", current.state)
-            .where("transition_generation", "=", current.generation)
-            .where("turn_claim_owner", "is", null),
-        );
-        if (result.numAffectedRows !== 1n) {
-          throw new Error(`Session ${identity.sessionId} placement changed during turn admission`);
-        }
-        return {
-          sessionId: current.sessionId,
-          claimId,
-          runId,
-          placementGeneration: current.generation,
-          owner,
-        };
+        const updatedAtMs = now();
+        const claim = claimTurnInDatabase(db, input, updatedAtMs);
+        insertWorkerWorkspacePendingResult(db, claim, updatedAtMs, instanceId);
+        return claim;
      });
    },

@@ -241,6 +259,78 @@ export function createPlacementTurnClaimOps(runtime: PlacementStoreRuntime) {
      return released;
    },

+    cancelWorkspaceResultAndReleaseTurn(
+      claim: WorkerSessionTurnClaim,
+    ): WorkerSessionPlacementRecord {
+      const sessionId = required(claim.sessionId, "session id");
+      const claimId = required(claim.claimId, "turn claim id");
+      const runId = required(claim.runId, "turn claim run id");
+      if (claimId !== runId || !claimId.startsWith("reclaim-")) {
+        throw new Error(`Session ${sessionId} workspace result is not owned by reclaim`);
+      }
+      // A failed stop must not expose a claim without its recovery fence (or vice
+      // versa), because either half-state permanently blocks the next reclaim.
+      const released = write((db) => {
+        const current = getRequired(db, sessionId);
+        const persisted = current.turnClaim;
+        const pending = executeSqliteQuerySync(
+          db,
+          getNodeSqliteKysely<Pick<StateDatabase, "worker_workspace_pending_results">>(db)
+            .selectFrom("worker_workspace_pending_results")
+            .selectAll()
+            .where("session_id", "=", sessionId),
+        ).rows[0];
+        if (
+          claim.owner.kind !== "worker" ||
+          (current.state !== "active" && current.state !== "draining") ||
+          current.environmentId !== claim.owner.environmentId ||
+          current.activeOwnerEpoch !== claim.owner.ownerEpoch ||
+          !persisted ||
+          persisted.owner !== "worker" ||
+          persisted.claimId !== claimId ||
+          persisted.runId !== runId ||
+          persisted.generation !== claim.placementGeneration ||
+          persisted.ownerEpoch !== claim.owner.ownerEpoch ||
+          !pending ||
+          pending.environment_id !== claim.owner.environmentId ||
+          pending.owner_epoch !== claim.owner.ownerEpoch ||
+          pending.placement_generation !== claim.placementGeneration ||
+          pending.claim_id !== claimId ||
+          pending.run_id !== runId ||
+          pending.workspace_accepted_at_ms !== null
+        ) {
+          throw new Error(
+            `Session ${sessionId} workspace result owner changed before cancellation`,
+          );
+        }
+        clearWorkerWorkspacePendingResult(db, sessionId);
+        const result = executeSqliteQuerySync(
+          db,
+          query(db)
+            .updateTable("worker_session_placements")
+            .set({
+              turn_claim_owner: null,
+              turn_claim_id: null,
+              turn_claim_run_id: null,
+              turn_claim_generation: null,
+              turn_claim_owner_epoch: null,
+              updated_at_ms: now(),
+            })
+            .where("session_id", "=", sessionId)
+            .where("state", "=", current.state)
+            .where("transition_generation", "=", current.generation)
+            .where("turn_claim_id", "=", claimId)
+            .where("turn_claim_run_id", "=", runId),
+        );
+        if (result.numAffectedRows !== 1n) {
+          throw new Error(`Session ${sessionId} workspace result changed during cancellation`);
+        }
+        return getRequired(db, sessionId);
+      });
+      signalTurnClaimRelease(path, sessionId);
+      return released;
+    },
+
    clearLocalTurnClaimsAfterRestart(): number {
      const clearedSessionIds = write((db) => {
        const sessionIds = executeSqliteQuerySync(