feat: recover saved restore operations by their durable decision

2026-10-26 11:50:54 +09:00
parent 9e3f87ac52
commit 6be4e8388b
6 changed files with 945 additions and 56 deletions
@@ -621,9 +621,13 @@ impl fmt::Debug for WorkerProtocolTransport {
 /// authority: Runtime must reread its decision before choosing any action.
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub enum RestoreRecoveryDisposition {
+    /// No live or pending owner for this exact operation. Not cleanup evidence.
+    Absent,
    InFlight,
    Continue,
    Discard,
+    /// A retained cleanup-only owner; never reconstruct or activate it.
+    Reclaim,
 }

 /// The connection selects this backend; WorkerRef identifies an execution,
@@ -704,8 +708,15 @@ pub trait WorkerExecutionBackend: Send + Sync + 'static {
        None
    }

-    /// Opt in to Runtime-owned continuation. Waiting is signaled through the
-    /// original execution context; no callback may wait for Runtime locks.
+    /// Opt in to Runtime-owned autonomous restore continuation. Pending tasks
+    /// signal completion through the original execution context; callbacks must
+    /// not wait for Runtime locks. Independent of resource disposition support.
+    fn supports_restore_continuation(&self) -> bool {
+        false
+    }
+
+    /// Describe exact operation resource ownership for warm/cold recovery
+    /// dispatch. This does not opt in to callbacks or autonomous continuation.
    fn restore_recovery_disposition(
        &self,
        _operation_id: WorkerLifecycleOperationId,
@@ -721,6 +732,24 @@ pub trait WorkerExecutionBackend: Send + Sync + 'static {
        self.restore_worker(request)
    }

+    /// Reclaim an uncommitted operation after restart without preparing,
+    /// repairing, or activating a replacement candidate. Runtime must establish
+    /// the durable uncommitted decision before calling this seam. Unsupported
+    /// cleanup is uncertainty, never a successful rollback.
+    fn reclaim_uncommitted_restore(
+        &self,
+        _request: WorkerExecutionRestoreRequest,
+    ) -> WorkerExecutionSpawnResult {
+        WorkerExecutionSpawnResult::ReconciliationRequired {
+            result: WorkerExecutionResult::unsupported(
+                WorkerExecutionOperation::Restore,
+                "backend cannot reclaim an uncommitted restore after restart",
+            ),
+            worker_state: None,
+            workdir_attachments: Vec::new(),
+        }
+    }
+
    /// Discard only the exact unpublished operation after Runtime has proved
    /// its durable decision uncommitted. This is not Stop and must not affect a
    /// connected execution. RolledBack requires confirmed cleanup and no repair.
@@ -1009,6 +1038,10 @@ impl WorkerExecutionBackendRef {
            .completed_restore_failure(operation_id, worker_ref)
    }

+    pub(crate) fn supports_restore_continuation(&self) -> bool {
+        self.backend.supports_restore_continuation()
+    }
+
    pub(crate) fn restore_recovery_disposition(
        &self,
        operation_id: WorkerLifecycleOperationId,
@@ -1025,6 +1058,13 @@ impl WorkerExecutionBackendRef {
        self.backend.reconcile_restore(request)
    }

+    pub(crate) fn reclaim_uncommitted_restore(
+        &self,
+        request: WorkerExecutionRestoreRequest,
+    ) -> WorkerExecutionSpawnResult {
+        self.backend.reclaim_uncommitted_restore(request)
+    }
+
    pub(crate) fn discard_uncommitted_restore(
        &self,
        operation_id: WorkerLifecycleOperationId,
@@ -171,6 +171,12 @@ enum WorkerRestoreMode {
    Automatic,
 }

+#[derive(Clone, Copy, Eq, PartialEq)]
+enum RestoreRecoveryAction {
+    Continue,
+    ReclaimUncommitted,
+}
+
 #[cfg(feature = "fs-store")]
 impl From<WorkerRestoreMode> for PersistedWorkerRestoreMode {
    fn from(mode: WorkerRestoreMode) -> Self {
@@ -2332,6 +2338,56 @@ impl Runtime {
        }
    }

+    // Seed durable crash checkpoints without launching a second Controller.
+    // Tests then reopen the real store with empty backend resource maps.
+    #[cfg(all(test, feature = "fs-store"))]
+    pub(crate) fn test_stage_restore_checkpoint(
+        &self,
+        worker_ref: &WorkerRef,
+        commit: RestoreCommitFact,
+        source_repair_possible: bool,
+    ) -> WorkerLifecycleOperationId {
+        let lease = self.worker_operation_lock(worker_ref.worker_id).unwrap();
+        let _guard = lease.lock().unwrap();
+        let operation = self
+            .lock()
+            .unwrap()
+            .worker(worker_ref)
+            .unwrap()
+            .pending_restore
+            .unwrap();
+        self.begin_restore_operation(
+            worker_ref,
+            operation.operation_id,
+            operation.mode,
+            self.restore_candidate_execution_context(worker_ref.clone()),
+        )
+        .unwrap();
+        if source_repair_possible {
+            self.restore_repair_authorizer(worker_ref.clone())(operation.operation_id).unwrap();
+        }
+        match commit {
+            RestoreCommitFact::Committed => {
+                self.commit_restored_worker_execution(
+                    worker_ref,
+                    protocol::WorkerStatus::Idle.into(),
+                    WorkerStatus::Idle,
+                    Vec::new(),
+                )
+                .unwrap();
+            }
+            RestoreCommitFact::LegacyUnknown => {
+                let mut state = self.lock().unwrap();
+                let mut worker = state.worker(worker_ref).unwrap().clone();
+                worker.pending_restore.as_mut().unwrap().commit = commit;
+                state.persist_restore_worker_record(&worker).unwrap();
+                state.workers.insert(worker_ref.worker_id, worker);
+            }
+            RestoreCommitFact::NotCommitted => {}
+        }
+        operation.operation_id
+    }
+
    #[cfg(test)]
    pub(crate) fn test_restore_request(
        &self,
@@ -2357,6 +2413,19 @@ impl Runtime {
        &self,
        worker_ref: &WorkerRef,
        mode: WorkerRestoreMode,
+    ) -> Result<RuntimeWorkerRestoreResult, RuntimeError> {
+        self.restore_admitted_with_recovery_under_lock(
+            worker_ref,
+            mode,
+            RestoreRecoveryAction::Continue,
+        )
+    }
+
+    fn restore_admitted_with_recovery_under_lock(
+        &self,
+        worker_ref: &WorkerRef,
+        mode: WorkerRestoreMode,
+        recovery: RestoreRecoveryAction,
    ) -> Result<RuntimeWorkerRestoreResult, RuntimeError> {
        self.lock()?
            .ensure_subject_not_reserved(worker_ref, false)?;
@@ -2407,14 +2476,28 @@ impl Runtime {
                    }
                };
                if let Some(detail) = committed_detail {
-                    // Recover a receipt gap without launching execution again.
+                    // A historical binding is not current readiness. The exact
+                    // execution must still attest its operation and endpoint.
+                    let backend = self.lock()?.execution_backend.clone();
+                    let readiness = backend
+                        .ok_or_else(|| "Runtime has no Worker execution backend".to_owned())
+                        .and_then(|backend| {
+                            backend.activate_restored_worker(owner.operation_id, worker_ref)
+                        });
+                    if let Err(message) = readiness {
+                        return Ok(RuntimeWorkerRestoreResult::failed(
+                            WorkerRestoreState::ReconciliationRequired,
+                            "worker_restore_saved_binding_not_ready",
+                            sanitize_worker_create_failure_message(&message),
+                        ));
+                    }
                    let result = RuntimeWorkerRestoreResult::accepted(detail);
                    self.finish_restore_receipt(worker_ref, &result)?;
                    return Ok(result);
                }
            }
        }
-        let result = self.restore_worker_under_lock(worker_ref, mode)?;
+        let result = self.restore_worker_with_recovery_under_lock(worker_ref, mode, recovery)?;
        self.finish_restore_receipt(worker_ref, &result)?;
        Ok(result)
    }
@@ -2451,6 +2534,19 @@ impl Runtime {
        &self,
        worker_ref: &WorkerRef,
        mode: WorkerRestoreMode,
+    ) -> Result<RuntimeWorkerRestoreResult, RuntimeError> {
+        self.restore_worker_with_recovery_under_lock(
+            worker_ref,
+            mode,
+            RestoreRecoveryAction::Continue,
+        )
+    }
+
+    fn restore_worker_with_recovery_under_lock(
+        &self,
+        worker_ref: &WorkerRef,
+        mode: WorkerRestoreMode,
+        recovery: RestoreRecoveryAction,
    ) -> Result<RuntimeWorkerRestoreResult, RuntimeError> {
        self.lock()?
            .ensure_subject_not_reserved(worker_ref, false)?;
@@ -2562,8 +2658,12 @@ impl Runtime {
                    .cloned()
            });
            let request = WorkerExecutionRestoreRequest {
-                source_repair_possible: pending_restore
-                    .is_some_and(|pending| pending.source_repair_possible),
+                // A committed decision also permits canonical activation effects.
+                // Cold reconstruction must retain that possibility through even a
+                // read-only failure, not cache it as an uncommitted rollback.
+                source_repair_possible: pending_restore.is_some_and(|pending| {
+                    pending.source_repair_possible || pending.commit == RestoreCommitFact::Committed
+                }),
                operation_id,
                worker_ref: worker_ref.clone(),
                request: worker_request,
@@ -2585,6 +2685,27 @@ impl Runtime {
            )
        };

+        let uncommitted = self
+            .lock()?
+            .worker(worker_ref)?
+            .pending_restore
+            .is_some_and(|pending| {
+                pending.operation_id == operation_id
+                    && pending.commit == RestoreCommitFact::NotCommitted
+            });
+        let reclaim = uncommitted
+            && (recovery == RestoreRecoveryAction::ReclaimUncommitted
+                || (reconcile
+                    && match backend.restore_recovery_disposition(operation_id, worker_ref) {
+                        Some(crate::execution::RestoreRecoveryDisposition::Reclaim) | None => true,
+                        Some(crate::execution::RestoreRecoveryDisposition::Absent) => self
+                            .lock()?
+                            .worker(worker_ref)?
+                            .restore_candidate_context
+                            .is_none(),
+                        Some(_) => false,
+                    }));
+        let reconcile = reconcile || reclaim;
        if !reconcile && let Err(result) = backend.preflight_restore(&request) {
            // A preflight may attest an already-connected Controller snapshot.
            // Repeated restore is read-only; do not replace its operation owner.
@@ -2643,6 +2764,8 @@ impl Runtime {
        let has_completed_failure = completed_failure.is_some();
        let restore_result = if let Some(result) = completed_failure {
            result
+        } else if reclaim {
+            backend.reclaim_uncommitted_restore(request)
        } else if reconcile {
            backend.reconcile_restore(request)
        } else {
@@ -4205,6 +4328,7 @@ impl Runtime {
                .lock()
                .map_err(|_| RuntimeError::StatePoisoned)?;
            if let Err(error) = self.reconcile_pending_stop_under_lock(&worker_ref) {
+                self.ensure_restore_continuation(&worker_ref)?;
                tracing::warn!(
                    worker_id = %worker_ref.worker_id,
                    error = %error,
@@ -4242,8 +4366,34 @@ impl Runtime {
            let _operation_guard = operation_lock
                .lock()
                .map_err(|_| RuntimeError::StatePoisoned)?;
-            let result = self.restore_admitted_worker_under_lock(&worker_ref, mode)?;
-            self.record_startup_restore_result(&worker_ref, &result)?;
+            let recovery = {
+                let mut state = self.lock()?;
+                state.refresh_uncertain_worker_write(&worker_ref)?;
+                let pending = state.worker(&worker_ref)?.pending_restore;
+                let retained = pending.is_some_and(|pending| {
+                    state.execution_backend.as_ref().is_some_and(|backend| {
+                        matches!(
+                            backend.restore_recovery_disposition(pending.operation_id, &worker_ref),
+                            Some(
+                                crate::execution::RestoreRecoveryDisposition::Continue
+                                    | crate::execution::RestoreRecoveryDisposition::InFlight
+                                    | crate::execution::RestoreRecoveryDisposition::Discard
+                            )
+                        )
+                    })
+                });
+                if pending.is_some_and(|pending| pending.commit == RestoreCommitFact::NotCommitted)
+                    && !retained
+                {
+                    RestoreRecoveryAction::ReclaimUncommitted
+                } else {
+                    RestoreRecoveryAction::Continue
+                }
+            };
+            let result =
+                self.restore_admitted_with_recovery_under_lock(&worker_ref, mode, recovery);
+            self.ensure_restore_continuation(&worker_ref)?;
+            self.record_startup_restore_result(&worker_ref, &result?)?;
        }

        for worker_ref in candidates {
@@ -4252,8 +4402,9 @@ impl Runtime {
                .lock()
                .map_err(|_| RuntimeError::StatePoisoned)?;
            let result =
-                self.restore_admitted_worker_under_lock(&worker_ref, WorkerRestoreMode::Automatic)?;
-            self.record_startup_restore_result(&worker_ref, &result)?;
+                self.restore_admitted_worker_under_lock(&worker_ref, WorkerRestoreMode::Automatic);
+            self.ensure_restore_continuation(&worker_ref)?;
+            self.record_startup_restore_result(&worker_ref, &result?)?;
        }
        Ok(())
    }
@@ -8411,6 +8562,7 @@ mod tests {
        restore_gate: Mutex<Option<Arc<RestoreGate>>>,
        restore_count: Mutex<u64>,
        restore_operation_ids: Mutex<Vec<WorkerLifecycleOperationId>>,
+        reclaim_operation_ids: Mutex<Vec<WorkerLifecycleOperationId>>,
        activation_result: Mutex<Option<Result<(), String>>>,
        activation_gate: Mutex<Option<Arc<RestoreGate>>>,
        activation_operation_ids: Mutex<Vec<WorkerLifecycleOperationId>>,
@@ -8708,6 +8860,57 @@ mod tests {
            }
        }

+        fn restore_recovery_disposition(
+            &self,
+            operation_id: WorkerLifecycleOperationId,
+            _worker_ref: &WorkerRef,
+        ) -> Option<crate::execution::RestoreRecoveryDisposition> {
+            Some(
+                if self
+                    .restore_operation_ids
+                    .lock()
+                    .unwrap()
+                    .contains(&operation_id)
+                {
+                    crate::execution::RestoreRecoveryDisposition::Continue
+                } else {
+                    crate::execution::RestoreRecoveryDisposition::Absent
+                },
+            )
+        }
+
+        fn reclaim_uncommitted_restore(
+            &self,
+            request: WorkerExecutionRestoreRequest,
+        ) -> WorkerExecutionSpawnResult {
+            self.reclaim_operation_ids
+                .lock()
+                .unwrap()
+                .push(request.operation_id);
+            assert!(
+                !self
+                    .contexts
+                    .lock()
+                    .unwrap()
+                    .contains_key(&request.worker_ref.worker_id)
+            );
+            if request.source_repair_possible {
+                WorkerExecutionSpawnResult::ReconciliationRequired {
+                    result: WorkerExecutionResult::errored(
+                        WorkerExecutionOperation::Restore,
+                        "prior repair may remain",
+                    ),
+                    worker_state: None,
+                    workdir_attachments: Vec::new(),
+                }
+            } else {
+                WorkerExecutionSpawnResult::RolledBack(WorkerExecutionResult::errored(
+                    WorkerExecutionOperation::Restore,
+                    "cold request resources confirmed absent",
+                ))
+            }
+        }
+
        fn reconcile_restore(
            &self,
            request: WorkerExecutionRestoreRequest,
@@ -9242,7 +9445,9 @@ mod tests {
                Some(&owner),
            )
            .unwrap();
-        assert_eq!(result.unwrap().state, WorkerRestoreState::Accepted);
+        assert_eq!(result.unwrap().state, WorkerRestoreState::RolledBack);
+        assert_eq!(*backend.restore_count.lock().unwrap(), 0);
+        assert_eq!(backend.reclaim_operation_ids.lock().unwrap().len(), 1);
        assert!(preparation.is_none());
        assert!(
            runtime
@@ -11212,7 +11417,7 @@ mod tests {

    #[cfg(feature = "fs-store")]
    #[test]
-    fn restore_guard_interrupted_preparation_replays_original_pending_operation() {
+    fn restore_guard_interrupted_preparation_reclaims_original_pending_operation() {
        let dir = tempfile::tempdir().unwrap();
        let options = FsRuntimeStoreOptions::new(dir.path().join("runtime"))
            .with_runtime_id("guard-partial-preparation");
@@ -11280,10 +11485,10 @@ mod tests {
        let receipt = resumed
            .restore_worker_operation(&worker.worker_ref, request)
            .unwrap();
-        assert_eq!(receipt.state, WorkerRestoreState::Accepted);
-        assert_eq!(*backend.restore_count.lock().unwrap(), 1);
+        assert_eq!(receipt.state, WorkerRestoreState::RolledBack);
+        assert_eq!(*backend.restore_count.lock().unwrap(), 0);
        assert_eq!(
-            backend.restore_operation_ids.lock().unwrap().as_slice(),
+            backend.reclaim_operation_ids.lock().unwrap().as_slice(),
            &[operation_id]
        );
        assert!(
@@ -11422,7 +11627,7 @@ mod tests {

    #[cfg(feature = "fs-store")]
    #[test]
-    fn restore_guard_unknown_preparation_write_retries_same_pending_operation() {
+    fn restore_guard_unknown_preparation_write_reclaims_same_pending_operation() {
        let dir = tempfile::tempdir().unwrap();
        let options = FsRuntimeStoreOptions::new(dir.path().join("runtime"))
            .with_runtime_id("guard-unknown-preparation");
@@ -11484,12 +11689,16 @@ mod tests {
                .restore_worker_operation(&worker.worker_ref, request)
                .unwrap()
                .state,
-            WorkerRestoreState::Accepted
+            WorkerRestoreState::RolledBack
+        );
+        assert_eq!(*resumed_backend.restore_count.lock().unwrap(), 0);
+        assert_eq!(
+            resumed_backend.reclaim_operation_ids.lock().unwrap().len(),
+            1
        );
-        assert_eq!(*resumed_backend.restore_count.lock().unwrap(), 1);
        assert_eq!(
            resumed.worker_detail(&worker.worker_ref).unwrap().status,
-            WorkerStatus::Idle
+            WorkerStatus::Stopped
        );
    }

@@ -11667,7 +11876,7 @@ mod tests {
            Runtime::with_fs_store_and_execution_backend(options, resumed_backend.clone()).unwrap();
        assert_eq!(
            resumed_backend
-                .restore_operation_ids
+                .reclaim_operation_ids
                .lock()
                .unwrap()
                .as_slice(),
@@ -11676,8 +11885,8 @@ mod tests {
        let receipt = resumed
            .restore_worker_operation(&worker.worker_ref, request)
            .unwrap();
-        assert_eq!(receipt.state, WorkerRestoreState::Accepted);
-        assert_eq!(*resumed_backend.restore_count.lock().unwrap(), 1);
+        assert_eq!(receipt.state, WorkerRestoreState::RolledBack);
+        assert_eq!(*resumed_backend.restore_count.lock().unwrap(), 0);
    }

    #[test]
@@ -15521,6 +15730,21 @@ mod tests {
            restarted_backend.clone(),
        )
        .unwrap();
+        assert_eq!(
+            restarted_backend
+                .reclaim_operation_ids
+                .lock()
+                .unwrap()
+                .as_slice(),
+            &[operation_id]
+        );
+        assert_eq!(*restarted_backend.restore_count.lock().unwrap(), 0);
+        assert_eq!(
+            restarted.worker_detail(&worker.worker_ref).unwrap().status,
+            WorkerStatus::Stopped
+        );
+        // A deliberate fresh intent may restore after the old uncommitted
+        // request was reclaimed. It must allocate a different operation.
        let retry = restarted
            .restore_worker_operation(
                &worker.worker_ref,
@@ -15529,13 +15753,9 @@ mod tests {
            .unwrap();
        assert_eq!(retry.state, WorkerRestoreState::Accepted);
        assert_eq!(*restarted_backend.restore_count.lock().unwrap(), 1);
-        assert_eq!(
-            restarted_backend
-                .restore_operation_ids
-                .lock()
-                .unwrap()
-                .as_slice(),
-            &[operation_id]
+        assert_ne!(
+            restarted_backend.restore_operation_ids.lock().unwrap()[0],
+            operation_id
        );
        assert!(
            restarted
@@ -266,7 +266,8 @@ impl Runtime {
            let Some(backend) = &state.execution_backend else {
                return Ok(());
            };
-            if backend
+            if !backend.supports_restore_continuation()
+                || backend
                    .restore_recovery_disposition(operation, &worker_ref)
                    .is_none()
            {
@@ -396,6 +397,10 @@ impl Runtime {
                            "worker_restore_rolled_back",
                            result.message_or_default(),
                        );
+                        #[cfg(feature = "fs-store")]
+                        self.lock()?
+                            .record_restore_failure(worker_ref, &result, &receipt)?;
+                        #[cfg(not(feature = "fs-store"))]
                        self.settle_restore_rollback(worker_ref, &receipt)?;
                        return Ok(Step::Finished);
                    }
@@ -2455,14 +2455,17 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
                .await
                .map_err(|_| RuntimeRestoreCleanupFailure::Workdir)?;
        }
-        if let Some(resources) = retained {
-            if let Some(allocation) = resources.allocation {
+        if let Some(allocation) = retained
+            .as_ref()
+            .and_then(|resources| resources.allocation.as_ref())
+        {
            allocation
                .release()
                .map_err(|_| RuntimeRestoreCleanupFailure::Allocation)?;
        } else {
-                // Reconstruction may fail before returning its allocation guard.
-                // Absence is checked at the allocation authority, not inferred from Err.
+            // A cold restart loses the retained map as well as allocation guards.
+            // Prove absence at the allocation authority even with an empty map;
+            // never release a possibly different owner by Worker name.
            let path = worker::runtime::worker_allocation::default_allocation_path()
                .map_err(|_| RuntimeRestoreCleanupFailure::Allocation)?;
            let table = worker::runtime::worker_allocation::LockFileGuard::open(&path)
@@ -2475,7 +2478,6 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
                return Err(RuntimeRestoreCleanupFailure::Allocation);
            }
        }
-        }
        // restore_controller starts the Controller only as its final fallible
        // step. A completed error leaves no live Controller; remove only this
        // operation's disposable run artifacts, never Session/metadata authority.
@@ -2486,10 +2488,18 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
            .map_err(|_| RuntimeRestoreCleanupFailure::Artifacts)?;
        Self::remove_restore_run_artifacts(&run_dir)
            .map_err(|_| RuntimeRestoreCleanupFailure::Artifacts)?;
-        self.failed_restore_resources
+        let mut resources = self
+            .failed_restore_resources
            .lock()
-            .map_err(|_| RuntimeRestoreCleanupFailure::Ownership)?
-            .remove(&request.worker_ref);
+            .map_err(|_| RuntimeRestoreCleanupFailure::Ownership)?;
+        if resources
+            .get(&request.worker_ref)
+            .is_some_and(|owner| owner.operation_id != request.operation_id)
+        {
+            return Err(RuntimeRestoreCleanupFailure::Ownership);
+        }
+        resources.remove(&request.worker_ref);
+        drop(resources);
        self.observation_hub.get(&request.worker_ref);
        Ok(())
    }
@@ -2763,6 +2773,29 @@ impl ProfileRuntimeWorkerFactory {
            let run_dir = self
                .worker_restore_run_dir(&request.worker_ref, request.operation_id)
                .map_err(|_| "restore run directory unavailable")?;
+            // A process restart loses the candidate, not its exact operation's
+            // disposable directory. Never infer writer absence from empty maps.
+            match std::fs::symlink_metadata(&run_dir) {
+                Ok(_) => {
+                    self.stopped_restore_run_dirs(&request.worker_ref)
+                        .map_err(|_| "restore run artifacts cannot be validated")?;
+                    let path = worker::runtime::worker_allocation::default_allocation_path()
+                        .map_err(|_| "restore allocation authority unavailable")?;
+                    let table = worker::runtime::worker_allocation::LockFileGuard::open(&path)
+                        .map_err(|_| "restore allocation authority unavailable")?;
+                    if table
+                        .data()
+                        .find(&Self::runtime_worker_name_for_ref(&request.worker_ref))
+                        .is_some()
+                    {
+                        return Err("restore run directory still has an allocation owner");
+                    }
+                    Self::remove_restore_run_artifacts(&run_dir)
+                        .map_err(|_| "restore run artifacts cannot be reclaimed")?;
+                }
+                Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
+                Err(_) => return Err("restore run artifacts cannot be inspected"),
+            }
            let transport = self.controller_transport;
            #[cfg(test)]
            let transport = if self.restore_faults.socket_failure.load(Ordering::SeqCst) {
@@ -3138,6 +3171,9 @@ struct RuntimeWorkerExecution {
 // marker. Timeout never drops the factory future or a late Controller result.
 struct PendingRuntimeRestore {
    request: WorkerExecutionRestoreRequest,
+    // Transient ownership mode, not a durable phase. Cold restart reclamation
+    // must never enter preparation, canonical repair, or activation.
+    cleanup_only: bool,
    task: tokio::sync::Mutex<Option<tokio::task::JoinHandle<()>>>,
    task_in_flight: AtomicBool,
    result: Mutex<Option<Result<Box<dyn RuntimeRestoreCandidate>, RuntimeRestoreFailure>>>,
@@ -3469,6 +3505,7 @@ where
        pending: &Arc<PendingRuntimeRestore>,
        resume_activation: bool,
    ) -> Result<(), String> {
+        let resume_activation = resume_activation && !pending.cleanup_only;
        let failure = pending
            .result
            .lock()
@@ -5317,6 +5354,16 @@ where
            }
        };
        if let Some(pending) = admitted_pending.as_ref() {
+            if pending.cleanup_only {
+                return WorkerExecutionSpawnResult::ReconciliationRequired {
+                    result: WorkerExecutionResult::rejected(
+                        WorkerExecutionOperation::Restore,
+                        "cleanup-only restore cannot prepare or reconstruct execution",
+                    ),
+                    worker_state: None,
+                    workdir_attachments: pending.request.previous_workdir_attachments.clone(),
+                };
+            }
            if request.request != pending.request.request
                || request.workspace_scope != pending.request.workspace_scope
                || request.logical_workdir_attachments
@@ -5378,6 +5425,7 @@ where
                    .remove(&worker_ref);
                let pending = Arc::new(PendingRuntimeRestore {
                    request: request.clone(),
+                    cleanup_only: false,
                    task: tokio::sync::Mutex::new(None),
                    task_in_flight: AtomicBool::new(false),
                    result: Mutex::new(None),
@@ -5593,17 +5641,23 @@ where
        }
    }

+    fn supports_restore_continuation(&self) -> bool {
+        true
+    }
+
    fn restore_recovery_disposition(
        &self,
        operation_id: crate::execution::WorkerLifecycleOperationId,
        worker_ref: &WorkerRef,
    ) -> Option<crate::execution::RestoreRecoveryDisposition> {
        use crate::execution::RestoreRecoveryDisposition as D;
-        let pending = self.pending_restore(worker_ref).ok().flatten();
+        let pending = self.pending_restore(worker_ref).ok()?;
        Some(match pending {
            Some(pending) if pending.request.operation_id == operation_id => {
                if pending.task_in_flight.load(Ordering::Acquire) {
                    D::InFlight
+                } else if pending.cleanup_only {
+                    D::Reclaim
                } else if pending
                    .discard_result
                    .lock()
@@ -5625,10 +5679,132 @@ where
                    D::Continue
                }
            }
-            _ => D::Continue,
+            _ => {
+                let workers = self.workers.lock().ok()?;
+                if workers
+                    .get(worker_ref)
+                    .is_some_and(|execution| execution.restore_operation_id == Some(operation_id))
+                {
+                    D::Continue
+                } else {
+                    D::Absent
+                }
+            }
        })
    }

+    fn reclaim_uncommitted_restore(
+        &self,
+        request: WorkerExecutionRestoreRequest,
+    ) -> WorkerExecutionSpawnResult {
+        let reclaimed = (|| -> Result<WorkerExecutionSpawnResult, String> {
+            let worker_ref = &request.worker_ref;
+            let operation_id = request.operation_id;
+            let operation_lock = self.worker_lock(worker_ref)?;
+            let _guard = operation_lock
+                .lock()
+                .map_err(|_| "worker operation lock is poisoned")?;
+            if request.context.worker_ref() != worker_ref {
+                return Err("restore cleanup context does not match Worker authority".into());
+            }
+            if self
+                .workers
+                .lock()
+                .map_err(|_| "worker registry lock is poisoned")?
+                .contains_key(worker_ref)
+            {
+                return Err("uncommitted reclaim cannot target a connected execution".into());
+            }
+            let pending = match self.pending_restore(worker_ref)? {
+                Some(pending) => {
+                    if pending.request.operation_id != operation_id || !pending.cleanup_only {
+                        return Err(
+                            "uncommitted reclaim cannot replace an in-memory restore owner".into(),
+                        );
+                    }
+                    if request.request != pending.request.request
+                        || request.workspace_scope != pending.request.workspace_scope
+                        || request.logical_workdir_attachments
+                            != pending.request.logical_workdir_attachments
+                        || request.previous_workdir_attachments
+                            != pending.request.previous_workdir_attachments
+                        || request.source_repair_possible != pending.request.source_repair_possible
+                    {
+                        return Err(
+                            "current Worker authority differs from the owned restore cleanup"
+                                .into(),
+                        );
+                    }
+                    pending
+                }
+                None => {
+                    if let Some(result) = self.completed_restore_failure(operation_id, worker_ref) {
+                        return Ok(result);
+                    }
+                    let mut failure = RuntimeRestoreFailure::unproven(
+                        RuntimeRestoreStage::Controller,
+                        "restore decision was not committed before restart",
+                    );
+                    failure.durable_effects_possible = request.source_repair_possible;
+                    let pending = Arc::new(PendingRuntimeRestore {
+                        request: request.clone(),
+                        cleanup_only: true,
+                        task: tokio::sync::Mutex::new(None),
+                        task_in_flight: AtomicBool::new(false),
+                        result: Mutex::new(Some(Err(failure))),
+                        activated: Mutex::new(None),
+                        activation_started: AtomicBool::new(false),
+                        discard_started: AtomicBool::new(false),
+                        discard_result: Mutex::new(None),
+                        failure: Mutex::new(None),
+                    });
+                    self.pending_restores
+                        .lock()
+                        .map_err(|_| "pending restore resources lock is poisoned")?
+                        .insert(worker_ref.clone(), pending.clone());
+                    pending
+                }
+            };
+            // Join the prior task before retrying. A timeout retains this owner,
+            // its original context notifier, and its exact task/JoinHandle.
+            self.wait_pending_restore(&pending, self.spawn_restore_timeout)?;
+            self.retry_pending_restore_cleanup(&pending, false)?;
+            self.wait_pending_restore(&pending, self.spawn_restore_timeout)?;
+            let failure = pending
+                .result
+                .lock()
+                .map_err(|_| "restore result lock is poisoned")?
+                .as_ref()
+                .and_then(|result| result.as_ref().err())
+                .cloned()
+                .ok_or("restore cleanup has no retained failure result")?;
+            if failure.needs_cleanup() {
+                return Err(failure.to_string());
+            }
+            if failure.durable_effects_possible {
+                return Err("Restore resource cleanup is confirmed; possible earlier canonical effects still require reconciliation".into());
+            }
+            self.completed_restore_failures
+                .lock()
+                .map_err(|_| "restore failure receipts lock is poisoned")?
+                .insert(worker_ref.clone(), (operation_id, failure.clone()));
+            self.remove_pending_restore(worker_ref)?;
+            Ok(WorkerExecutionSpawnResult::RolledBack(
+                WorkerExecutionResult::errored(
+                    WorkerExecutionOperation::Restore,
+                    failure.to_string(),
+                ),
+            ))
+        })();
+        reclaimed.unwrap_or_else(
+            |message| WorkerExecutionSpawnResult::ReconciliationRequired {
+                result: WorkerExecutionResult::errored(WorkerExecutionOperation::Restore, message),
+                worker_state: None,
+                workdir_attachments: request.previous_workdir_attachments.clone(),
+            },
+        )
+    }
+
    fn discard_uncommitted_restore(
        &self,
        operation_id: crate::execution::WorkerLifecycleOperationId,
@@ -5643,6 +5819,7 @@ where
                .pending_restore(worker_ref)?
                .ok_or("unpublished restore owner is unavailable")?;
            if pending.request.operation_id != operation_id
+                || pending.cleanup_only
                || pending.activation_started.load(Ordering::Acquire)
            {
                return Err(
@@ -5784,6 +5961,9 @@ where
            if pending.request.operation_id != operation_id {
                return Err("restored Worker operation identity does not match".into());
            }
+            if pending.cleanup_only {
+                return Err("cleanup-only restore cannot activate execution".into());
+            }
            self.wait_pending_restore(&pending, self.spawn_restore_timeout)?;
            if pending.owns_prepared_candidate() {
                pending.activation_started.store(true, Ordering::Release);
@@ -7696,6 +7876,7 @@ mod tests {
            worker_ref.clone(),
            Arc::new(PendingRuntimeRestore {
                request: direct_restore_request(worker_ref.clone()),
+                cleanup_only: false,
                task: tokio::sync::Mutex::new(None),
                task_in_flight: AtomicBool::new(false),
                result: Mutex::new(None),
@@ -7740,6 +7921,7 @@ mod tests {
            .unwrap();
        let pending = Arc::new(PendingRuntimeRestore {
            request: request.clone(),
+            cleanup_only: false,
            task: tokio::sync::Mutex::new(Some(failed)),
            task_in_flight: AtomicBool::new(false),
            result: Mutex::new(None),
@@ -10255,6 +10437,309 @@ mod tests {
        }
    }

+    // Backend component seam only: no saved Worker, model, or Controller is
+    // constructed. Any accidental preparation/recovery is an immediate failure.
+    struct CleanupOnlyFactory {
+        cleanup_calls: Arc<AtomicUsize>,
+        cleanup_allowed: Arc<AtomicBool>,
+        release: Option<Arc<tokio::sync::Notify>>,
+    }
+
+    #[async_trait]
+    impl RuntimeWorkerFactory for CleanupOnlyFactory {
+        async fn spawn_controller(
+            &self,
+            _: WorkerExecutionSpawnRequest,
+        ) -> Result<RuntimeWorkerController, String> {
+            panic!("reclaim must not spawn")
+        }
+
+        async fn preflight_restore(&self, _: &WorkerExecutionRestoreRequest) -> Result<(), String> {
+            panic!("reclaim must not preflight preparation")
+        }
+
+        async fn recover_restore_source(
+            &self,
+            _: &WorkerExecutionRestoreRequest,
+        ) -> Result<bool, RuntimeRestoreFailure> {
+            panic!("reclaim must not repair canonical source")
+        }
+
+        async fn prepare_restore(
+            &self,
+            _: WorkerExecutionRestoreRequest,
+        ) -> Result<Box<dyn RuntimeRestoreCandidate>, RuntimeRestoreFailure> {
+            panic!("reclaim must not prepare")
+        }
+
+        async fn cleanup_failed_restore(
+            &self,
+            _: &WorkerExecutionRestoreRequest,
+        ) -> Result<(), RuntimeRestoreCleanupFailure> {
+            if self.cleanup_calls.fetch_add(1, Ordering::SeqCst) == 0 {
+                if let Some(release) = &self.release {
+                    release.notified().await;
+                }
+            }
+            if self.cleanup_allowed.load(Ordering::SeqCst) {
+                Ok(())
+            } else {
+                Err(RuntimeRestoreCleanupFailure::Workdir)
+            }
+        }
+
+        async fn reconcile_stopped_worker(&self, _: &WorkerRef) -> Result<(), String> {
+            panic!("reclaim must not clean all Worker operations through Stop")
+        }
+    }
+
+    #[test]
+    fn cold_reclaim_retries_cleanup_only_and_keeps_cleaned_receipt_idempotent() {
+        use crate::execution::RestoreRecoveryDisposition as D;
+        let cleanup_calls = Arc::new(AtomicUsize::new(0));
+        let cleanup_allowed = Arc::new(AtomicBool::new(false));
+        let backend = WorkerRuntimeExecutionBackend::new(CleanupOnlyFactory {
+            cleanup_calls: cleanup_calls.clone(),
+            cleanup_allowed: cleanup_allowed.clone(),
+            release: None,
+        })
+        .unwrap();
+        let request = direct_restore_request(WorkerRef::new(WorkerId::now_v7()));
+        let worker_ref = &request.worker_ref;
+        assert_eq!(
+            backend.restore_recovery_disposition(request.operation_id, worker_ref),
+            Some(D::Absent)
+        );
+        assert!(matches!(
+            backend.reclaim_uncommitted_restore(request.clone()),
+            WorkerExecutionSpawnResult::ReconciliationRequired { .. }
+        ));
+        let owner = backend.pending_restore(worker_ref).unwrap().unwrap();
+        assert!(owner.cleanup_only);
+        assert_eq!(cleanup_calls.load(Ordering::SeqCst), 1);
+        assert_eq!(
+            backend.restore_recovery_disposition(request.operation_id, worker_ref),
+            Some(D::Reclaim)
+        );
+        // Wrong operation and changed authority must not replace or retry it.
+        let mut different = request.clone();
+        different.operation_id = crate::execution::WorkerLifecycleOperationId::new();
+        assert!(matches!(
+            backend.reclaim_uncommitted_restore(different),
+            WorkerExecutionSpawnResult::ReconciliationRequired { .. }
+        ));
+        let mut mismatched = request.clone();
+        mismatched.request = create_request("changed cleanup authority");
+        assert!(matches!(
+            backend.reclaim_uncommitted_restore(mismatched),
+            WorkerExecutionSpawnResult::ReconciliationRequired { .. }
+        ));
+        assert!(Arc::ptr_eq(
+            &owner,
+            &backend.pending_restore(worker_ref).unwrap().unwrap()
+        ));
+        // Erroneous generic callers cannot reconstruct, activate, or substitute
+        // in-process candidate discard for cold cleanup.
+        assert!(matches!(
+            backend.restore_worker(request.clone()),
+            WorkerExecutionSpawnResult::ReconciliationRequired { .. }
+        ));
+        assert!(
+            backend
+                .activate_restored_worker(request.operation_id, worker_ref)
+                .is_err()
+        );
+        assert!(matches!(
+            backend.discard_uncommitted_restore(request.operation_id, worker_ref),
+            WorkerExecutionSpawnResult::ReconciliationRequired { .. }
+        ));
+        assert_eq!(cleanup_calls.load(Ordering::SeqCst), 1);
+        cleanup_allowed.store(true, Ordering::SeqCst);
+        assert!(matches!(
+            backend.reclaim_uncommitted_restore(request.clone()),
+            WorkerExecutionSpawnResult::RolledBack(_)
+        ));
+        assert!(backend.pending_restore(worker_ref).unwrap().is_none());
+        assert!(matches!(
+            backend.reclaim_uncommitted_restore(request.clone()),
+            WorkerExecutionSpawnResult::RolledBack(_)
+        ));
+        assert!(matches!(
+            backend.restore_worker(request),
+            WorkerExecutionSpawnResult::RolledBack(_)
+        ));
+        assert_eq!(cleanup_calls.load(Ordering::SeqCst), 2);
+        assert!(backend.workers.lock().unwrap().is_empty());
+    }
+
+    #[test]
+    fn cold_reclaim_preserves_possible_effects_without_repreparing() {
+        use crate::execution::RestoreRecoveryDisposition as D;
+        let cleanup_calls = Arc::new(AtomicUsize::new(0));
+        let backend = WorkerRuntimeExecutionBackend::new(CleanupOnlyFactory {
+            cleanup_calls: cleanup_calls.clone(),
+            cleanup_allowed: Arc::new(AtomicBool::new(true)),
+            release: None,
+        })
+        .unwrap();
+        let mut request = direct_restore_request(WorkerRef::new(WorkerId::now_v7()));
+        request.source_repair_possible = true;
+        for _ in 0..2 {
+            assert!(matches!(
+                backend.reclaim_uncommitted_restore(request.clone()),
+                WorkerExecutionSpawnResult::ReconciliationRequired { .. }
+            ));
+        }
+        let owner = backend
+            .pending_restore(&request.worker_ref)
+            .unwrap()
+            .unwrap();
+        assert!(owner.cleanup_only);
+        let retained = owner.result.lock().unwrap();
+        let failure = retained.as_ref().unwrap().as_ref().err().unwrap();
+        assert!(failure.durable_effects_possible);
+        assert_eq!(failure.cleanup, RuntimeRestoreCleanup::Confirmed);
+        drop(retained);
+        // Even an internal retry with the wrong resume flag cannot prepare.
+        backend.retry_pending_restore_cleanup(&owner, true).unwrap();
+        assert!(owner.task.try_lock().unwrap().is_none());
+        assert!(
+            backend
+                .completed_restore_failure(request.operation_id, &request.worker_ref)
+                .is_none()
+        );
+        assert_eq!(
+            backend.restore_recovery_disposition(request.operation_id, &request.worker_ref),
+            Some(D::Reclaim)
+        );
+        assert!(matches!(
+            backend.reconcile_restore(request.clone()),
+            WorkerExecutionSpawnResult::ReconciliationRequired { .. }
+        ));
+        assert!(
+            backend
+                .activate_restored_worker(request.operation_id, &request.worker_ref)
+                .is_err()
+        );
+        assert_eq!(cleanup_calls.load(Ordering::SeqCst), 1);
+    }
+
+    #[test]
+    fn cold_reclaim_timeout_retains_same_task_and_notifies_before_cleanup_retry() {
+        use crate::execution::RestoreRecoveryDisposition as D;
+        let cleanup_calls = Arc::new(AtomicUsize::new(0));
+        let cleanup_allowed = Arc::new(AtomicBool::new(false));
+        let release = Arc::new(tokio::sync::Notify::new());
+        let mut backend = WorkerRuntimeExecutionBackend::new(CleanupOnlyFactory {
+            cleanup_calls: cleanup_calls.clone(),
+            cleanup_allowed: cleanup_allowed.clone(),
+            release: Some(release.clone()),
+        })
+        .unwrap()
+        .with_spawn_restore_timeout(Duration::from_millis(1));
+        let mut request = direct_restore_request(WorkerRef::new(WorkerId::now_v7()));
+        let (progress, progressed) = mpsc::channel();
+        request.context = request
+            .context
+            .with_restore_progress(Arc::new(move |operation| {
+                let _ = progress.send(operation);
+            }));
+        assert!(matches!(
+            backend.reclaim_uncommitted_restore(request.clone()),
+            WorkerExecutionSpawnResult::ReconciliationRequired { .. }
+        ));
+        let owner = backend
+            .pending_restore(&request.worker_ref)
+            .unwrap()
+            .unwrap();
+        let task_id = owner.task.try_lock().unwrap().as_ref().unwrap().id();
+        assert_eq!(
+            backend.restore_recovery_disposition(request.operation_id, &request.worker_ref),
+            Some(D::InFlight)
+        );
+        assert!(matches!(
+            backend.reclaim_uncommitted_restore(request.clone()),
+            WorkerExecutionSpawnResult::ReconciliationRequired { .. }
+        ));
+        assert!(Arc::ptr_eq(
+            &owner,
+            &backend
+                .pending_restore(&request.worker_ref)
+                .unwrap()
+                .unwrap()
+        ));
+        assert_eq!(
+            owner.task.try_lock().unwrap().as_ref().unwrap().id(),
+            task_id
+        );
+        release.notify_one();
+        assert_eq!(
+            progressed.recv_timeout(Duration::from_secs(5)).unwrap(),
+            request.operation_id
+        );
+        backend
+            .wait_pending_restore(&owner, Duration::from_secs(5))
+            .unwrap();
+        assert!(owner.task.try_lock().unwrap().is_none());
+        assert_eq!(cleanup_calls.load(Ordering::SeqCst), 1);
+        assert_eq!(
+            backend.restore_recovery_disposition(request.operation_id, &request.worker_ref),
+            Some(D::Reclaim)
+        );
+        backend.spawn_restore_timeout = Duration::from_secs(5);
+        cleanup_allowed.store(true, Ordering::SeqCst);
+        assert!(matches!(
+            backend.reclaim_uncommitted_restore(request.clone()),
+            WorkerExecutionSpawnResult::RolledBack(_)
+        ));
+        assert_eq!(
+            progressed.recv_timeout(Duration::from_secs(5)).unwrap(),
+            request.operation_id
+        );
+        assert_eq!(cleanup_calls.load(Ordering::SeqCst), 2);
+        assert!(backend.workers.lock().unwrap().is_empty());
+    }
+
+    #[test]
+    fn cold_reclaim_refuses_existing_candidate_owner_even_for_same_operation() {
+        let backend = WorkerRuntimeExecutionBackend::new(CleanupOnlyFactory {
+            cleanup_calls: Arc::new(AtomicUsize::new(0)),
+            cleanup_allowed: Arc::new(AtomicBool::new(true)),
+            release: None,
+        })
+        .unwrap();
+        let request = direct_restore_request(WorkerRef::new(WorkerId::now_v7()));
+        let owner = Arc::new(PendingRuntimeRestore {
+            request: request.clone(),
+            cleanup_only: false,
+            task: tokio::sync::Mutex::new(None),
+            task_in_flight: AtomicBool::new(false),
+            result: Mutex::new(None),
+            activated: Mutex::new(None),
+            activation_started: AtomicBool::new(false),
+            discard_started: AtomicBool::new(false),
+            discard_result: Mutex::new(None),
+            failure: Mutex::new(None),
+        });
+        backend
+            .pending_restores
+            .lock()
+            .unwrap()
+            .insert(request.worker_ref.clone(), owner.clone());
+        assert!(matches!(
+            backend.reclaim_uncommitted_restore(request.clone()),
+            WorkerExecutionSpawnResult::ReconciliationRequired { .. }
+        ));
+        assert!(Arc::ptr_eq(
+            &owner,
+            &backend
+                .pending_restore(&request.worker_ref)
+                .unwrap()
+                .unwrap()
+        ));
+        assert_eq!(backend.factory.cleanup_calls.load(Ordering::SeqCst), 0);
+    }
+
    #[test]
    fn absent_execution_stop_reconciles_pending_restore_cleanup_without_launching() {
        let restore_calls = Arc::new(AtomicUsize::new(0));
@@ -193,6 +193,18 @@ max_tokens = 100
        self.inner.restore_worker(request)
    }

+    fn restore_recovery_disposition(
+        &self,
+        operation_id: crate::execution::WorkerLifecycleOperationId,
+        worker_ref: &WorkerRef,
+    ) -> Option<crate::execution::RestoreRecoveryDisposition> {
+        self.inner.restore_recovery_disposition(operation_id, worker_ref)
+    }
+
+    fn reclaim_uncommitted_restore(&self, request: WorkerExecutionRestoreRequest) -> WorkerExecutionSpawnResult {
+        self.inner.reclaim_uncommitted_restore(request)
+    }
+
    fn reconcile_restore(
        &self,
        request: WorkerExecutionRestoreRequest,
@@ -94,6 +94,133 @@ mod publication {
        (release, received)
    }

+    #[test]
+    #[serial_test::serial(worker_allocation)]
+    fn startup_fault_recovery_retains_cleanup_or_committed_reconstruction_without_repost() {
+        use crate::runtime::RestoreCommitFact as F;
+        for committed in [false, true] {
+            let fixture = Fixture::no_model();
+            let runtime = stopped_runtime(&fixture);
+            let worker_ref = fixture.request.worker_ref.clone();
+            let request = runtime.test_restore_request(&worker_ref);
+            runtime.coordinate_worker_restore_operation(&worker_ref, runtime_api::WorkerRestoreCoordinationRequest {
+                expected_observation_token: request.expected_observation_token.clone(),
+                request_id: request.request_id.clone(), preparation: request.preparation.clone(),
+            }, fixture.request.workspace_scope.as_ref()).unwrap();
+            let operation = runtime.test_stage_restore_checkpoint(&worker_ref, if committed { F::Committed } else { F::NotCommitted }, false);
+            let before = fixture.canonical_files();
+            drop(runtime);
+            if committed {
+                *fixture.backend.factory.restore_faults.fail_at.lock().unwrap() = Some(RuntimeRestoreStage::Metadata);
+            } else {
+                fixture.backend.factory.restore_faults.cleanup_fail.store(true, Ordering::SeqCst);
+            }
+            let options = FsRuntimeStoreOptions::new(fixture._root.path().join("catalog")).with_runtime_id("candidate-runtime");
+            let restarted = EmbeddedRuntime::with_fs_store_and_execution_backend(options, fixture.backend.clone()).unwrap();
+            let pause = restarted.test_pause_restore_continuation();
+            assert_eq!(durable_operation(&restarted, &worker_ref).unwrap().operation_id, operation);
+            assert!(fixture.backend.completed_restore_failure(operation, &worker_ref).is_none());
+            assert!(fixture.backend.workers.lock().unwrap().is_empty());
+            *fixture.backend.factory.restore_faults.fail_at.lock().unwrap() = None;
+            fixture.backend.factory.restore_faults.cleanup_fail.store(false, Ordering::SeqCst);
+            drop(pause);
+            restarted.test_wait_restore_continuation(&worker_ref, operation, WAIT);
+            let receipt = restarted.restore_worker_operation(&worker_ref, request).unwrap();
+            assert_eq!(receipt.state, if committed { server_api::WorkerRestoreState::Accepted } else { server_api::WorkerRestoreState::RolledBack });
+            if committed {
+                let transport = fixture.backend.clone().attach_worker_protocol(&worker_ref).unwrap();
+                transport.validate().unwrap();
+                drop(transport);
+                restarted.stop_worker(&worker_ref, None).unwrap();
+            } else {
+                assert_eq!(fixture.backend.factory.restore_faults.executions.load(Ordering::SeqCst), 0);
+                assert_eq!(fixture.canonical_files(), before);
+            }
+            fixture.assert_released();
+            fixture.assert_quiet();
+        }
+    }
+
+    #[test]
+    #[serial_test::serial(worker_allocation)]
+    fn startup_dispatches_saved_decisions_without_replaying_uncommitted_execution() {
+        use crate::runtime::RestoreCommitFact as F;
+        for (commit, repaired, expected) in [
+            (F::NotCommitted, false, server_api::WorkerRestoreState::RolledBack),
+            (F::NotCommitted, true, server_api::WorkerRestoreState::ReconciliationRequired),
+            (F::Committed, false, server_api::WorkerRestoreState::Accepted),
+            (F::LegacyUnknown, true, server_api::WorkerRestoreState::ReconciliationRequired),
+        ] {
+            let fixture = Fixture::no_model();
+            let runtime = stopped_runtime(&fixture);
+            let worker_ref = fixture.request.worker_ref.clone();
+            let scope = fixture.request.workspace_scope.as_ref().unwrap();
+            let request = runtime.test_restore_request(&worker_ref);
+            assert!(runtime.coordinate_worker_restore_operation(&worker_ref, runtime_api::WorkerRestoreCoordinationRequest {
+                expected_observation_token: request.expected_observation_token.clone(),
+                request_id: request.request_id.clone(),
+                preparation: request.preparation.clone(),
+            }, Some(scope)).unwrap().0.is_none());
+            let operation = runtime.test_stage_restore_checkpoint(&worker_ref, commit, repaired);
+            let before = fixture.canonical_files();
+            let run = fixture.backend.factory.worker_restore_run_dir(&worker_ref, operation).unwrap();
+            std::fs::create_dir_all(&run).unwrap();
+            std::fs::write(run.join("before-restart.marker"), "disposable").unwrap();
+            if commit == F::NotCommitted && !repaired {
+                // Fixture for an interrupted automatic restore. A rollback must
+                // not leave an active/Automatic projection that restores again.
+                let store = runtime.test_runtime_store();
+                let mut record = store.read_worker_record(&worker_ref).unwrap();
+                record.status = crate::catalog::WorkerStatus::Idle;
+                let PersistedWorkerExecutionState::ReconciliationRequired(
+                    crate::fs_store::PersistedWorkerLifecycleOperation::Restore(saved),
+                ) = &mut record.execution_state else { panic!("pending fixture") };
+                saved.mode = crate::fs_store::PersistedWorkerRestoreMode::Automatic;
+                saved.last_settled_status = crate::catalog::WorkerStatus::Idle;
+                saved.binding = Some(crate::fs_store::PersistedWorkerExecutionBinding {});
+                store.write_worker_record(&record).unwrap();
+            }
+            drop(runtime);
+            assert!(fixture.backend.pending_restore(&worker_ref).unwrap().is_none());
+            assert!(fixture.backend.workers.lock().unwrap().is_empty());
+            let options = FsRuntimeStoreOptions::new(fixture._root.path().join("catalog")).with_runtime_id("candidate-runtime");
+            let restarted = EmbeddedRuntime::with_fs_store_and_execution_backend(options.clone(), fixture.backend.clone()).unwrap();
+            if expected == server_api::WorkerRestoreState::ReconciliationRequired {
+                let reason = restarted.test_wait_restore_continuation_parked(&worker_ref, operation, WAIT);
+                assert!(!reason.is_empty());
+                let pending = durable_operation(&restarted, &worker_ref).unwrap();
+                assert_eq!(pending.operation_id, operation);
+                assert_eq!(pending.commit, commit);
+                assert_eq!(pending.source_repair_possible, repaired);
+            }
+            assert!(restarted.worker_detail(&worker_ref).is_ok(), "{commit:?}: {:?}", restarted.diagnostics());
+            let receipt = restarted.restore_worker_operation(&worker_ref, request.clone()).unwrap();
+            assert_eq!(receipt.state, expected, "{commit:?}: {receipt:?}; backend failure: {:?}", fixture.backend.completed_restore_failure(operation, &worker_ref));
+            assert_eq!(fixture.backend.factory.restore_faults.executions.load(Ordering::SeqCst), usize::from(commit == F::Committed));
+            if commit == F::Committed {
+                let transport = fixture.backend.clone().attach_worker_protocol(&worker_ref).unwrap();
+                transport.validate().unwrap();
+                drop(transport);
+            } else {
+                assert_eq!(fixture.canonical_files(), before);
+                assert!(fixture.backend.workers.lock().unwrap().is_empty());
+                assert!(fixture.backend.factory.observation_hub.get(&worker_ref).is_none());
+                assert_eq!(run.exists(), commit == F::LegacyUnknown);
+            }
+            if expected == server_api::WorkerRestoreState::RolledBack {
+                assert_eq!(restarted.worker_detail(&worker_ref).unwrap().status, crate::catalog::WorkerStatus::Stopped);
+                drop(restarted);
+                let second = EmbeddedRuntime::with_fs_store_and_execution_backend(options, fixture.backend.clone()).unwrap();
+                assert_eq!(second.restore_worker_operation_scoped(scope, &worker_ref, request).unwrap(), receipt);
+                assert_eq!(fixture.backend.factory.restore_faults.executions.load(Ordering::SeqCst), 0);
+            } else {
+                restarted.stop_worker_scoped(scope, &worker_ref, None).unwrap();
+            }
+            fixture.assert_released();
+            fixture.assert_quiet();
+        }
+    }
+
    #[test]
    #[serial_test::serial(worker_allocation)]
    fn stop_timeout_retains_autonomous_late_cleanup_and_blocks_remove_until_confirmed() {