feat: runtime own restore continuation beyond caller timeout

crates/worker-runtime/src/execution.rs

@@ -242,6 +242,9 @@ enum WorkerExecutionPublicationState {
 #[derive(Clone)]
 pub struct WorkerExecutionContext {
     worker_ref: WorkerRef,
+    restore_progress: Option<Arc<dyn Fn(WorkerLifecycleOperationId) + Send + Sync>>,
+    restore_repair_authorizer:
+        Option<Arc<dyn Fn(WorkerLifecycleOperationId) -> Result<(), RuntimeError> + Send + Sync>>,
     #[cfg(feature = "ws-server")]
     observation_publisher: WorkerObservationPublisher,
     #[cfg(feature = "ws-server")]
@@ -256,6 +259,8 @@ impl WorkerExecutionContext {
     ) -> Self {
         Self {
             worker_ref,
+            restore_progress: None,
+            restore_repair_authorizer: None,
             observation_publisher,
             publication_state: Arc::new(Mutex::new(WorkerExecutionPublicationState::Active)),
         }
@@ -268,6 +273,8 @@ impl WorkerExecutionContext {
     ) -> Self {
         Self {
             worker_ref,
+            restore_progress: None,
+            restore_repair_authorizer: None,
             observation_publisher,
             publication_state: Arc::new(Mutex::new(WorkerExecutionPublicationState::Pending(
                 VecDeque::new(),
@@ -277,12 +284,56 @@ impl WorkerExecutionContext {
 
     #[cfg(not(feature = "ws-server"))]
     pub(crate) fn new(worker_ref: WorkerRef) -> Self {
-        Self { worker_ref }
+        Self {
+            worker_ref,
+            restore_progress: None,
+            restore_repair_authorizer: None,
+        }
     }
 
     #[cfg(not(feature = "ws-server"))]
     pub(crate) fn candidate(worker_ref: WorkerRef) -> Self {
-        Self { worker_ref }
+        Self {
+            worker_ref,
+            restore_progress: None,
+            restore_repair_authorizer: None,
+        }
+    }
+
+    pub(crate) fn with_restore_repair_authorizer(
+        mut self,
+        authorizer: Arc<
+            dyn Fn(WorkerLifecycleOperationId) -> Result<(), RuntimeError> + Send + Sync,
+        >,
+    ) -> Self {
+        self.restore_repair_authorizer = Some(authorizer);
+        self
+    }
+
+    pub(crate) fn authorize_restore_source_repair(
+        &self,
+        operation: WorkerLifecycleOperationId,
+    ) -> Result<(), RuntimeError> {
+        match &self.restore_repair_authorizer {
+            Some(authorizer) => authorizer(operation),
+            // Direct factory/component users own their source repair separately;
+            // only Runtime contexts carry a Runtime aggregate authority callback.
+            None => Ok(()),
+        }
+    }
+
+    pub(crate) fn with_restore_progress(
+        mut self,
+        progress: Arc<dyn Fn(WorkerLifecycleOperationId) + Send + Sync>,
+    ) -> Self {
+        self.restore_progress = Some(progress);
+        self
+    }
+
+    pub(crate) fn restore_progressed(&self, operation_id: WorkerLifecycleOperationId) {
+        if let Some(progress) = &self.restore_progress {
+            progress(operation_id);
+        }
     }
 
     pub fn worker_ref(&self) -> &WorkerRef {
@@ -424,6 +475,7 @@ pub struct WorkerExecutionSpawnRequest {
 /// Request passed to a [`WorkerExecutionBackend`] when restoring a persisted Worker.
 #[derive(Clone, Debug)]
 pub struct WorkerExecutionRestoreRequest {
+    pub source_repair_possible: bool,
     pub operation_id: WorkerLifecycleOperationId,
     pub worker_ref: WorkerRef,
     pub request: crate::catalog::CreateWorkerRequest,
@@ -565,6 +617,15 @@ impl fmt::Debug for WorkerProtocolTransport {
     }
 }
 
+/// Derived resource disposition for an existing restore owner. Not durable
+/// authority: Runtime must reread its decision before choosing any action.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum RestoreRecoveryDisposition {
+    InFlight,
+    Continue,
+    Discard,
+}
+
 /// The connection selects this backend; WorkerRef identifies an execution,
 /// never a capability or authorization token.
 pub trait WorkerExecutionBackend: Send + Sync + 'static {
@@ -643,6 +704,16 @@ 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.
+    fn restore_recovery_disposition(
+        &self,
+        _operation_id: WorkerLifecycleOperationId,
+        _worker_ref: &WorkerRef,
+    ) -> Option<RestoreRecoveryDisposition> {
+        None
+    }
+
     fn reconcile_restore(
         &self,
         request: WorkerExecutionRestoreRequest,
@@ -938,6 +1009,15 @@ impl WorkerExecutionBackendRef {
             .completed_restore_failure(operation_id, worker_ref)
     }
 
+    pub(crate) fn restore_recovery_disposition(
+        &self,
+        operation_id: WorkerLifecycleOperationId,
+        worker_ref: &WorkerRef,
+    ) -> Option<RestoreRecoveryDisposition> {
+        self.backend
+            .restore_recovery_disposition(operation_id, worker_ref)
+    }
+
     pub(crate) fn reconcile_restore(
         &self,
         request: WorkerExecutionRestoreRequest,

crates/worker-runtime/src/fs_store.rs

@@ -43,7 +43,8 @@ use std::sync::atomic::{AtomicU8, AtomicU64, Ordering};
 #[path = "test_support.rs"]
 pub mod test_support;
 
-pub(crate) const SCHEMA_VERSION: u32 = 13;
+pub(crate) const SCHEMA_VERSION: u32 = 14;
+const RESTORE_DECISION_SCHEMA_VERSION: u32 = 13;
 const RESTORE_ADMISSION_SCHEMA_VERSION: u32 = 12;
 // Frozen source schemas: do not move their meaning when adding a migration.
 const SOURCE_SCHEMA_VERSION: u32 = 11;
@@ -653,6 +654,7 @@ pub(crate) struct PersistedWorkerRestoreOperation {
     pub(crate) operation_id: WorkerLifecycleOperationId,
     pub(crate) mode: PersistedWorkerRestoreMode,
     pub(crate) commit: crate::runtime::RestoreCommitFact,
+    pub(crate) source_repair_possible: bool,
     pub(crate) request: CreateWorkerRequest,
     pub(crate) binding: Option<PersistedWorkerExecutionBinding>,
     pub(crate) last_settled_status: WorkerStatus,
@@ -829,11 +831,12 @@ fn plan_runtime_store_migration(
             | EXECUTION_SCHEMA_VERSION
             | SOURCE_SCHEMA_VERSION
             | RESTORE_ADMISSION_SCHEMA_VERSION
+            | RESTORE_DECISION_SCHEMA_VERSION
     ) {
         return Err(runtime_store_corrupt(
             &runtime_path,
             format!(
-                "unsupported Runtime store schema version {schema_version}; expected {OLDEST_SCHEMA_VERSION}, {LEGACY_SCHEMA_VERSION}, {PREVIOUS_SCHEMA_VERSION}, {AGGREGATE_SCHEMA_VERSION}, {EXECUTION_SCHEMA_VERSION}, {SOURCE_SCHEMA_VERSION}, {RESTORE_ADMISSION_SCHEMA_VERSION}, or {SCHEMA_VERSION}"
+                "unsupported Runtime store schema version {schema_version}; expected {OLDEST_SCHEMA_VERSION}, {LEGACY_SCHEMA_VERSION}, {PREVIOUS_SCHEMA_VERSION}, {AGGREGATE_SCHEMA_VERSION}, {EXECUTION_SCHEMA_VERSION}, {SOURCE_SCHEMA_VERSION}, {RESTORE_ADMISSION_SCHEMA_VERSION}, {RESTORE_DECISION_SCHEMA_VERSION}, or {SCHEMA_VERSION}"
             ),
         ));
     }
@@ -1165,7 +1168,10 @@ fn migrate_worker_record(
 ) -> Result<WorkerAggregateRecord, RuntimeError> {
     if matches!(
         source_schema_version,
-        EXECUTION_SCHEMA_VERSION | SOURCE_SCHEMA_VERSION | RESTORE_ADMISSION_SCHEMA_VERSION
+        EXECUTION_SCHEMA_VERSION
+            | SOURCE_SCHEMA_VERSION
+            | RESTORE_ADMISSION_SCHEMA_VERSION
+            | RESTORE_DECISION_SCHEMA_VERSION
     ) {
         // Admission identity survives migration. Older pending records did not
         // distinguish a committed decision from preparation or failed cleanup.
@@ -1179,7 +1185,10 @@ fn migrate_worker_record(
                 "Worker schema does not match Runtime migration source".into(),
             ));
         }
-        migrate_restore_commit_fact(&mut identity_document, identity_path)?;
+        if source_schema_version != RESTORE_DECISION_SCHEMA_VERSION {
+            migrate_restore_commit_fact(&mut identity_document, identity_path)?;
+        }
+        migrate_restore_repair_fact(&mut identity_document, identity_path)?;
         identity_document["schema_version"] = serde_json::Value::from(SCHEMA_VERSION);
         let aggregate: WorkerAggregateRecord = serde_json::from_value(identity_document)
             .map_err(|error| runtime_store_corrupt(identity_path, error.to_string()))?;
@@ -1294,6 +1303,42 @@ fn migrate_restore_commit_fact(
     Ok(())
 }
 
+// Earlier writers could repair another canonical store before commit without
+// retaining that authority in this aggregate. Missing evidence is never false.
+fn migrate_restore_repair_fact(
+    document: &mut serde_json::Value,
+    path: &Path,
+) -> Result<(), RuntimeError> {
+    let execution = &mut document["execution_state"];
+    if execution["state"] != "reconciliation_required" {
+        return Ok(());
+    }
+    let operation = &mut execution["execution"];
+    let restore = match operation["operation"].as_str() {
+        Some("restore") => Some(&mut operation["intent"]),
+        Some("stop") => operation["intent"]
+            .as_object_mut()
+            .and_then(|stop| stop.get_mut("pending_restore"))
+            .filter(|restore| !restore.is_null()),
+        _ => None,
+    };
+    if let Some(restore) = restore {
+        let restore = restore.as_object_mut().ok_or_else(|| {
+            runtime_store_corrupt(path, "legacy pending Restore must be an object".into())
+        })?;
+        if restore
+            .insert("source_repair_possible".into(), serde_json::json!(true))
+            .is_some()
+        {
+            return Err(runtime_store_corrupt(
+                path,
+                "legacy Restore contains unexpected source repair evidence".into(),
+            ));
+        }
+    }
+    Ok(())
+}
+
 // Frozen schema-v9 conversion. Preserve the exact token and all Restore owners
 // and receipts; generating a fresh token here would lose retry/admission authority.
 fn migrate_v9_restore_guard(
@@ -1608,7 +1653,7 @@ fn migrate_runtime_document(
         .or_insert_with(|| serde_json::json!({}));
     if !matches!(
         source_schema_version,
-        SOURCE_SCHEMA_VERSION | RESTORE_ADMISSION_SCHEMA_VERSION
+        SOURCE_SCHEMA_VERSION | RESTORE_ADMISSION_SCHEMA_VERSION | RESTORE_DECISION_SCHEMA_VERSION
     ) {
         object.remove("workers");
         // This legacy cache is not Runtime lifecycle or restore authority. Bundles
@@ -2148,7 +2193,9 @@ fn migrate_runtime_store_in_place(
         if let Some(metadata) = metadata
             && !matches!(
                 plan.current_schema_version,
-                SOURCE_SCHEMA_VERSION | RESTORE_ADMISSION_SCHEMA_VERSION
+                SOURCE_SCHEMA_VERSION
+                    | RESTORE_ADMISSION_SCHEMA_VERSION
+                    | RESTORE_DECISION_SCHEMA_VERSION
             )
         {
             atomic_write_json(
@@ -4001,7 +4048,13 @@ mod tests {
     #[test]
     fn legacy_pending_restore_migrates_as_unknown_and_preserves_stop_ownership() {
         use crate::runtime::RestoreCommitFact;
-        for under_stop in [false, true] {
+        for (source_schema, under_stop) in [
+            RESTORE_ADMISSION_SCHEMA_VERSION,
+            RESTORE_DECISION_SCHEMA_VERSION,
+        ]
+        .into_iter()
+        .flat_map(|schema| [false, true].map(move |stop| (schema, stop)))
+        {
             let (_temp, store, mut worker, path) = canonical_worker_store();
             let root = store.root.clone();
             let PersistedWorkerExecutionState::Available(execution) =
@@ -4013,6 +4066,7 @@ mod tests {
                 operation_id: WorkerLifecycleOperationId::new(),
                 mode: PersistedWorkerRestoreMode::Explicit,
                 commit: RestoreCommitFact::Committed,
+                source_repair_possible: false,
                 request: execution.request.clone(),
                 binding: execution.binding.clone(),
                 last_settled_status: worker.status,
@@ -4033,21 +4087,27 @@ mod tests {
                 PersistedWorkerExecutionState::ReconciliationRequired(operation);
             let mut legacy =
                 serde_json::to_value(WorkerAggregateRecord::from_persisted(&worker)).unwrap();
-            legacy["schema_version"] = serde_json::json!(RESTORE_ADMISSION_SCHEMA_VERSION);
+            legacy["schema_version"] = serde_json::json!(source_schema);
             let intent = &mut legacy["execution_state"]["execution"]["intent"];
             let restore = if under_stop {
                 &mut intent["pending_restore"]
             } else {
                 intent
             };
-            restore.as_object_mut().unwrap().remove("commit");
+            if source_schema == RESTORE_ADMISSION_SCHEMA_VERSION {
+                restore.as_object_mut().unwrap().remove("commit");
+            }
+            restore
+                .as_object_mut()
+                .unwrap()
+                .remove("source_repair_possible");
             let expected_request = restore["request"].clone();
             let before_guard = legacy["restore_guard"].clone();
             fs::write(&path, serde_json::to_vec(&legacy).unwrap()).unwrap();
             write_empty_schema_v8_store(&root);
             let runtime_path = root.join(RUNTIME_FILE);
             let mut runtime: serde_json::Value = read_json(&runtime_path, "read fixture").unwrap();
-            runtime["schema_version"] = serde_json::json!(RESTORE_ADMISSION_SCHEMA_VERSION);
+            runtime["schema_version"] = serde_json::json!(source_schema);
             fs::write(&runtime_path, serde_json::to_vec(&runtime).unwrap()).unwrap();
             drop(store);
 
@@ -4062,7 +4122,18 @@ mod tests {
                 PersistedWorkerLifecycleOperation::Restore(restore) => restore,
                 PersistedWorkerLifecycleOperation::Stop(stop) => stop.pending_restore.unwrap(),
             };
-            assert_eq!(restored.commit, RestoreCommitFact::LegacyUnknown);
+            assert_eq!(
+                restored.commit,
+                if source_schema == RESTORE_DECISION_SCHEMA_VERSION {
+                    RestoreCommitFact::Committed
+                } else {
+                    RestoreCommitFact::LegacyUnknown
+                }
+            );
+            assert!(
+                restored.source_repair_possible,
+                "older source-repair effects must remain possible"
+            );
             assert_eq!(
                 serde_json::to_value(restored.request).unwrap(),
                 expected_request

crates/worker-runtime/src/runtime.rs

@@ -1,3 +1,4 @@
+mod restore_continuation;
 use crate::catalog::{
     ConfigBundleRef, CreateWorkerRequest, LogicalWorkdirAttachment, ProfileSelector,
     ProfileSourceArchiveSource, RepositoryRefObservation, RepositoryRefObservationRequest,
@@ -323,6 +324,7 @@ impl Drop for RuntimeEventSelectorSubscription {
 pub struct Runtime {
     inner: Arc<Mutex<RuntimeState>>,
     worker_operations: Arc<Mutex<BTreeMap<WorkerId, Arc<Mutex<()>>>>>,
+    restore_continuations: Arc<restore_continuation::RestoreContinuations>,
 }
 
 struct WorkerOperationLease {
@@ -586,6 +588,7 @@ impl Runtime {
         Self {
             inner: Arc::new(Mutex::new(state)),
             worker_operations: Arc::new(Mutex::new(BTreeMap::new())),
+            restore_continuations: Arc::default(),
         }
     }
 
@@ -662,6 +665,7 @@ impl Runtime {
         let runtime = Self {
             inner: Arc::new(Mutex::new(state)),
             worker_operations: Arc::new(Mutex::new(BTreeMap::new())),
+            restore_continuations: Arc::default(),
         };
         runtime.restore_persisted_worker_executions()?;
         Ok(runtime)
@@ -2230,6 +2234,7 @@ impl Runtime {
                         operation_id: WorkerLifecycleOperationId::new(),
                         mode: WorkerRestoreMode::Explicit,
                         commit: RestoreCommitFact::NotCommitted,
+                        source_repair_possible: false,
                         last_settled_status: candidate.status,
                     });
                 let operation_id = pending.operation_id;
@@ -2256,6 +2261,10 @@ impl Runtime {
                 match state.persist_restore_worker_record(&candidate) {
                     Ok(()) => {}
                     Err(error @ RuntimeError::StoreCommitOutcomeUnknown { .. }) => {
+                        drop(state);
+                        if !coordinate {
+                            self.start_restore_continuation(worker_ref.clone(), operation_id)?;
+                        }
                         return Ok((
                             Some(RuntimeWorkerRestoreResult::failed(
                                 WorkerRestoreState::ReconciliationRequired,
@@ -2287,8 +2296,10 @@ impl Runtime {
                 return Ok((None, owner.request.preparation.clone()));
             }
         }
-        self.restore_admitted_worker_under_lock(worker_ref, WorkerRestoreMode::Explicit)
+        let result =
-            .map(|result| (Some(result), None))
+            self.restore_admitted_worker_under_lock(worker_ref, WorkerRestoreMode::Explicit);
+        self.ensure_restore_continuation(worker_ref)?;
+        result.map(|result| (Some(result), None))
     }
 
     #[cfg(all(test, feature = "fs-store"))]
@@ -2529,6 +2540,8 @@ impl Runtime {
                     .cloned()
             });
             let request = WorkerExecutionRestoreRequest {
+                source_repair_possible: pending_restore
+                    .is_some_and(|pending| pending.source_repair_possible),
                 operation_id,
                 worker_ref: worker_ref.clone(),
                 request: worker_request,
@@ -4075,12 +4088,26 @@ impl Runtime {
         crate::execution::WorkerExecutionContext::new(worker_ref)
     }
 
+    fn restore_progress_notifier(
+        &self,
+        worker_ref: WorkerRef,
+    ) -> Arc<dyn Fn(WorkerLifecycleOperationId) + Send + Sync> {
+        let continuations = Arc::downgrade(&self.restore_continuations);
+        Arc::new(move |operation| {
+            if let Some(continuations) = continuations.upgrade() {
+                continuations.signal(&worker_ref, operation);
+            }
+        })
+    }
+
     #[cfg(feature = "ws-server")]
     fn restore_candidate_execution_context(
         &self,
         worker_ref: WorkerRef,
     ) -> crate::execution::WorkerExecutionContext {
         let runtime = Arc::downgrade(&self.inner);
+        let progress = self.restore_progress_notifier(worker_ref.clone());
+        let repair = self.restore_repair_authorizer(worker_ref.clone());
         crate::execution::WorkerExecutionContext::candidate(
             worker_ref,
             Arc::new(move |worker_ref, payload| {
@@ -4101,6 +4128,8 @@ impl Runtime {
                 Ok(state.push_worker_observation_event(worker_ref, payload))
             }),
         )
+        .with_restore_progress(progress)
+        .with_restore_repair_authorizer(repair)
     }
 
     #[cfg(not(feature = "ws-server"))]
@@ -4108,7 +4137,11 @@ impl Runtime {
         &self,
         worker_ref: WorkerRef,
     ) -> crate::execution::WorkerExecutionContext {
+        let progress = self.restore_progress_notifier(worker_ref.clone());
+        let repair = self.restore_repair_authorizer(worker_ref.clone());
         crate::execution::WorkerExecutionContext::candidate(worker_ref)
+            .with_restore_progress(progress)
+            .with_restore_repair_authorizer(repair)
     }
 
     #[cfg(feature = "fs-store")]
@@ -4410,6 +4443,9 @@ impl Runtime {
             commit: candidate
                 .pending_restore
                 .map_or(RestoreCommitFact::NotCommitted, |pending| pending.commit),
+            source_repair_possible: candidate
+                .pending_restore
+                .is_some_and(|pending| pending.source_repair_possible),
             last_settled_status: candidate.status,
         });
         candidate.restore_candidate_context = Some(candidate_context);
@@ -6327,6 +6363,9 @@ struct PendingWorkerRestore {
     operation_id: WorkerLifecycleOperationId,
     mode: WorkerRestoreMode,
     commit: RestoreCommitFact,
+    // False proves Runtime never granted this operation authority to repair
+    // canonical source before its decision. True includes legacy uncertainty.
+    source_repair_possible: bool,
     last_settled_status: WorkerStatus,
 }
 
@@ -6460,10 +6499,9 @@ struct WorkerRecord {
 
 impl WorkerRecord {
     fn ensure_restore_rollback_allowed(&self) -> Result<(), RuntimeError> {
-        if self
+        if self.pending_restore.is_some_and(|pending| {
-            .pending_restore
+            pending.commit != RestoreCommitFact::NotCommitted || pending.source_repair_possible
-            .is_some_and(|pending| pending.commit != RestoreCommitFact::NotCommitted)
+        }) {
-        {
             return Err(RuntimeError::InvalidRequest(
                 "A committed or legacy-unknown Restore cannot be settled as a rollback".into(),
             ));
@@ -6682,6 +6720,7 @@ impl WorkerRecord {
                             operation_id: operation.operation_id,
                             mode: operation.mode.into(),
                             commit: operation.commit,
+                            source_repair_possible: operation.source_repair_possible,
                             last_settled_status: operation.last_settled_status,
                         }),
                         None,
@@ -6698,6 +6737,7 @@ impl WorkerRecord {
                             operation_id: restore.operation_id,
                             mode: restore.mode.into(),
                             commit: restore.commit,
+                            source_repair_possible: restore.source_repair_possible,
                             last_settled_status: restore.last_settled_status,
                         }),
                     Some(PendingWorkerStop {
@@ -6762,6 +6802,7 @@ impl WorkerRecord {
                             operation_id: restore.operation_id,
                             mode: restore.mode.into(),
                             commit: restore.commit,
+                            source_repair_possible: restore.source_repair_possible,
                             request: self
                                 .request
                                 .clone()
@@ -6786,6 +6827,7 @@ impl WorkerRecord {
                     operation_id: pending.operation_id,
                     mode: pending.mode.into(),
                     commit: pending.commit,
+                    source_repair_possible: pending.source_repair_possible,
                     request,
                     binding: self
                         .execution_bound
@@ -9156,6 +9198,7 @@ mod tests {
             operation_id: WorkerLifecycleOperationId::new(),
             mode: WorkerRestoreMode::Explicit,
             commit: RestoreCommitFact::NotCommitted,
+            source_repair_possible: false,
             last_settled_status: WorkerStatus::Stopped,
         });
         let id = "workdir-irrelevant-to-recovery";
@@ -13588,7 +13631,7 @@ mod tests {
         })
         .unwrap_err();
         assert!(error.to_string().contains(
-            "unsupported Runtime store schema version 2; expected 6, 7, 8, 9, 10, 11, 12, or 13"
+            "unsupported Runtime store schema version 2; expected 6, 7, 8, 9, 10, 11, 12, 13, or 14"
         ));
 
         let _ = std::fs::remove_dir_all(root);
@@ -15014,6 +15057,87 @@ mod tests {
         let _ = std::fs::remove_dir_all(root);
     }
 
+    #[cfg(feature = "fs-store")]
+    #[test]
+    fn restore_source_repair_authority_is_durable_before_effects_without_reentering_lifecycle_lock()
+    {
+        for fault in [
+            None,
+            Some(crate::fs_store::AtomicWriteFault::BeforeRename),
+            Some(crate::fs_store::AtomicWriteFault::AfterRename),
+        ] {
+            let dir = tempfile::tempdir().unwrap();
+            let options = FsRuntimeStoreOptions::new(dir.path().join("runtime"))
+                .with_runtime_id("repair-authority");
+            let runtime = Runtime::with_fs_store_and_execution_backend(
+                options,
+                Arc::new(TestExecutionBackend::default()),
+            )
+            .unwrap();
+            runtime.store_config_bundle(test_bundle()).unwrap();
+            let worker = runtime
+                .create_worker(task_request("repair authority"))
+                .unwrap();
+            runtime.stop_worker(&worker.worker_ref, None).unwrap();
+            let lease = runtime.worker_operation_lock(worker.worker_id).unwrap();
+            let _guard = lease.lock().unwrap();
+            let operation = WorkerLifecycleOperationId::new();
+            runtime
+                .begin_restore_operation(
+                    &worker.worker_ref,
+                    operation,
+                    WorkerRestoreMode::Explicit,
+                    runtime.restore_candidate_execution_context(worker.worker_ref.clone()),
+                )
+                .unwrap();
+            if let Some(fault) = fault {
+                runtime_store(&runtime).fail_next_worker_write(fault);
+                if matches!(fault, crate::fs_store::AtomicWriteFault::AfterRename) {
+                    runtime_store(&runtime).fail_next_worker_reconcile_sync();
+                }
+            }
+            let authorizer = runtime.restore_repair_authorizer(worker.worker_ref.clone());
+            let callback = authorizer.clone();
+            let (sent, received) = std::sync::mpsc::channel();
+            let task = std::thread::spawn(move || sent.send(callback(operation)).unwrap());
+            // The lifecycle lock above remains held while this exact factory
+            // callback persists its grant. Waiting on that lock would deadlock.
+            let granted = received
+                .recv_timeout(std::time::Duration::from_secs(5))
+                .expect("repair callback reentered lifecycle lock");
+            task.join().unwrap();
+            assert_eq!(granted.is_ok(), fault.is_none());
+            let saved = runtime_store(&runtime)
+                .read_worker_record(&worker.worker_ref)
+                .unwrap();
+            let PersistedWorkerExecutionState::ReconciliationRequired(
+                PersistedWorkerLifecycleOperation::Restore(saved),
+            ) = saved.execution_state
+            else {
+                panic!("repair grant lost operation")
+            };
+            assert_eq!(saved.operation_id, operation);
+            assert_eq!(saved.commit, RestoreCommitFact::NotCommitted);
+            assert_eq!(
+                saved.source_repair_possible,
+                !matches!(fault, Some(crate::fs_store::AtomicWriteFault::BeforeRename))
+            );
+            // A denied/unknown grant must not authorize external repair. Once
+            // storage is healthy the same operation reconciles, never a new ID.
+            authorizer(operation).unwrap();
+            let rollback = RuntimeWorkerRestoreResult::failed(
+                WorkerRestoreState::RolledBack,
+                "unsafe",
+                "repair effects may exist",
+            );
+            assert!(
+                runtime
+                    .settle_restore_rollback(&worker.worker_ref, &rollback)
+                    .is_err()
+            );
+        }
+    }
+
     #[cfg(feature = "fs-store")]
     #[test]
     fn restore_decision_reread_preserves_commit_across_uncertainty_stop_and_restart() {

crates/worker-runtime/src/runtime/restore_continuation.rs

@@ -0,0 +1,498 @@
+//! Runtime-owned continuation of an admitted restore, independent of its caller.
+//!
+//! Completion callbacks only signal a condition variable. They never acquire a
+//! Runtime/lifecycle lock and are never joined by this driver while holding one.
+//! Each step revalidates the canonical owner under the existing Worker lock;
+//! waiting/backoff and driver teardown happen outside it. A repeated obstruction
+//! parks this owner with its reason, rather than spinning or inventing rollback.
+
+use super::*;
+use crate::execution::RestoreRecoveryDisposition;
+use std::collections::HashMap;
+use std::sync::{Condvar, Weak};
+use std::thread::JoinHandle;
+use std::time::Duration;
+
+type Key = (WorkerRef, WorkerLifecycleOperationId);
+
+#[derive(Debug, Default)]
+pub(super) struct RestoreContinuations {
+    owners: Mutex<HashMap<Key, Arc<Continuation>>>,
+    #[cfg(test)]
+    paused: std::sync::atomic::AtomicBool,
+}
+
+#[derive(Debug, Default)]
+struct Schedule {
+    signaled: bool,
+    cancelled: bool,
+    finished: bool,
+    blocked_reason: Option<String>,
+    parked: bool,
+}
+
+#[derive(Debug, Default)]
+struct Continuation {
+    schedule: Mutex<Schedule>,
+    wake: Condvar,
+    task: Mutex<Option<JoinHandle<()>>>,
+}
+
+impl RestoreContinuations {
+    pub(super) fn signal(&self, worker_ref: &WorkerRef, operation: WorkerLifecycleOperationId) {
+        if let Some(owner) = self
+            .owners
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .get(&(worker_ref.clone(), operation))
+        {
+            let mut schedule = owner
+                .schedule
+                .lock()
+                .unwrap_or_else(std::sync::PoisonError::into_inner);
+            schedule.signaled = true;
+            owner.wake.notify_all();
+        }
+    }
+}
+
+impl Drop for RestoreContinuations {
+    fn drop(&mut self) {
+        let owners = self
+            .owners
+            .get_mut()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        for owner in owners.values() {
+            let mut schedule = owner
+                .schedule
+                .lock()
+                .unwrap_or_else(std::sync::PoisonError::into_inner);
+            schedule.cancelled = true;
+            owner.wake.notify_all();
+        }
+        for owner in owners.values() {
+            if let Some(task) = owner
+                .task
+                .lock()
+                .unwrap_or_else(std::sync::PoisonError::into_inner)
+                .take()
+            {
+                if task.thread().id() != std::thread::current().id() {
+                    let _ = task.join();
+                }
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+pub(crate) struct ContinuationPause(Arc<RestoreContinuations>);
+
+#[cfg(test)]
+impl Drop for ContinuationPause {
+    fn drop(&mut self) {
+        self.0
+            .paused
+            .store(false, std::sync::atomic::Ordering::Release);
+        for owner in self.0.owners.lock().unwrap().values() {
+            owner.schedule.lock().unwrap().signaled = true;
+            owner.wake.notify_all();
+        }
+    }
+}
+
+enum Step {
+    Finished,
+    Waiting,
+    Contended,
+    Retry(String),
+}
+
+impl Runtime {
+    pub(super) fn restore_repair_authorizer(
+        &self,
+        worker_ref: WorkerRef,
+    ) -> Arc<dyn Fn(WorkerLifecycleOperationId) -> Result<(), RuntimeError> + Send + Sync> {
+        let inner = Arc::downgrade(&self.inner);
+        Arc::new(move |operation| {
+            let inner = inner.upgrade().ok_or(RuntimeError::RuntimeStopped)?;
+            // The factory caller owns the lifecycle lock. Never reacquire it
+            // here: it may be awaiting this exact factory task.
+            let mut state = inner.lock().map_err(|_| RuntimeError::StatePoisoned)?;
+            state.ensure_running()?;
+            state.refresh_uncertain_worker_write(&worker_ref)?;
+            let worker = state.worker(&worker_ref)?;
+            state.ensure_subject_not_reserved(&worker_ref, false)?;
+            if worker.pending_stop.is_some()
+                || worker.pending_restore.is_none_or(|pending| {
+                    pending.operation_id != operation
+                        || pending.commit == RestoreCommitFact::LegacyUnknown
+                })
+            {
+                return Err(RuntimeError::RestoreObservationConflict {
+                    worker_id: worker_ref.worker_id,
+                });
+            }
+            if let Some(owner) = worker
+                .restore_guard
+                .active_request_id
+                .as_ref()
+                .and_then(|id| worker.restore_guard.owners.get(id))
+            {
+                owner.validate_workdir_preparation_authority(worker)?;
+            }
+            if worker
+                .pending_restore
+                .is_some_and(|pending| pending.source_repair_possible)
+            {
+                return Ok(());
+            }
+            let mut candidate = worker.clone();
+            candidate
+                .pending_restore
+                .as_mut()
+                .expect("validated restore owner")
+                .source_repair_possible = true;
+            // Grant before touching another canonical store. Unknown durability
+            // denies repair until authoritative reconciliation; no old-memory write.
+            state.persist_restore_worker_record(&candidate)?;
+            state.workers.insert(worker_ref.worker_id, candidate);
+            Ok(())
+        })
+    }
+
+    #[cfg(test)]
+    pub(crate) fn test_wait_restore_continuation_parked(
+        &self,
+        worker_ref: &WorkerRef,
+        operation: WorkerLifecycleOperationId,
+        timeout: Duration,
+    ) -> String {
+        let owner = self
+            .restore_continuations
+            .owners
+            .lock()
+            .unwrap()
+            .get(&(worker_ref.clone(), operation))
+            .cloned()
+            .unwrap();
+        let (schedule, waited) = owner
+            .wake
+            .wait_timeout_while(owner.schedule.lock().unwrap(), timeout, |s| {
+                !s.parked && !s.finished
+            })
+            .unwrap();
+        assert!(
+            !waited.timed_out() && schedule.parked,
+            "continuation did not park: {:?}",
+            schedule.blocked_reason
+        );
+        schedule.blocked_reason.clone().expect("parked reason")
+    }
+
+    /// Gate autonomous steps when a component test observes a single fault
+    /// boundary. Dedicated continuation tests run without this gate.
+    #[cfg(test)]
+    pub(crate) fn test_pause_restore_continuation(&self) -> ContinuationPause {
+        assert!(
+            !self
+                .restore_continuations
+                .paused
+                .swap(true, std::sync::atomic::Ordering::AcqRel)
+        );
+        ContinuationPause(self.restore_continuations.clone())
+    }
+
+    #[cfg(test)]
+    pub(crate) fn test_wait_restore_continuation(
+        &self,
+        worker_ref: &WorkerRef,
+        operation: WorkerLifecycleOperationId,
+        timeout: Duration,
+    ) {
+        let owner = self
+            .restore_continuations
+            .owners
+            .lock()
+            .unwrap()
+            .get(&(worker_ref.clone(), operation))
+            .cloned()
+            .expect("continuation owner");
+        let schedule = owner.schedule.lock().unwrap();
+        let (schedule, waited) = owner
+            .wake
+            .wait_timeout_while(schedule, timeout, |s| !s.finished && !s.cancelled)
+            .unwrap();
+        assert!(
+            !waited.timed_out() && schedule.finished,
+            "restore continuation did not settle: {:?}",
+            schedule.blocked_reason
+        );
+    }
+
+    pub(super) fn ensure_restore_continuation(
+        &self,
+        worker_ref: &WorkerRef,
+    ) -> Result<(), RuntimeError> {
+        let operation = {
+            let state = self.lock()?;
+            let worker = state.worker(worker_ref)?;
+            worker
+                .pending_restore
+                .map(|pending| pending.operation_id)
+                .or_else(|| {
+                    worker
+                        .restore_guard
+                        .active_request_id
+                        .as_ref()
+                        .and_then(|id| worker.restore_guard.owners.get(id))
+                        .filter(|owner| owner.receipt.is_none())
+                        .map(|owner| owner.operation_id)
+                })
+        };
+        if let Some(operation) = operation {
+            self.start_restore_continuation(worker_ref.clone(), operation)?;
+        }
+        Ok(())
+    }
+
+    pub(super) fn start_restore_continuation(
+        &self,
+        worker_ref: WorkerRef,
+        operation: WorkerLifecycleOperationId,
+    ) -> Result<(), RuntimeError> {
+        {
+            let state = self.lock()?;
+            let Some(backend) = &state.execution_backend else {
+                return Ok(());
+            };
+            if backend
+                .restore_recovery_disposition(operation, &worker_ref)
+                .is_none()
+            {
+                return Ok(());
+            }
+        }
+        let key = (worker_ref.clone(), operation);
+        let mut owners = self
+            .restore_continuations
+            .owners
+            .lock()
+            .map_err(|_| RuntimeError::StatePoisoned)?;
+        if owners.contains_key(&key) {
+            return Ok(());
+        }
+        let owner = Arc::new(Continuation::default());
+        let retained = owner.clone();
+        let state = Arc::downgrade(&self.inner);
+        let operations = Arc::downgrade(&self.worker_operations);
+        let continuations = Arc::downgrade(&self.restore_continuations);
+        let task = std::thread::Builder::new()
+            .name("restore-continuation".into())
+            .spawn(move || {
+                run(
+                    state,
+                    operations,
+                    continuations,
+                    retained,
+                    worker_ref,
+                    operation,
+                );
+            })
+            .map_err(|error| {
+                RuntimeError::InvalidRequest(format!(
+                    "Restore continuation could not start: {error}"
+                ))
+            })?;
+        *owner.task.lock().map_err(|_| RuntimeError::StatePoisoned)? = Some(task);
+        owners.insert(key, owner);
+        Ok(())
+    }
+
+    fn continue_restore_operation(
+        &self,
+        worker_ref: &WorkerRef,
+        operation: WorkerLifecycleOperationId,
+    ) -> Result<Step, RuntimeError> {
+        #[cfg(test)]
+        if self
+            .restore_continuations
+            .paused
+            .load(std::sync::atomic::Ordering::Acquire)
+        {
+            return Ok(Step::Contended);
+        }
+        let lease = self.worker_operation_lock(worker_ref.worker_id)?;
+        // Another caller may be waiting for a backend task. Do not queue a join
+        // of that caller/driver while retaining the lock it needs to complete.
+        let _guard = match lease.try_lock() {
+            Ok(guard) => guard,
+            Err(std::sync::TryLockError::WouldBlock) => return Ok(Step::Contended),
+            Err(_) => return Err(RuntimeError::StatePoisoned),
+        };
+        let (backend, pending) = {
+            let mut state = self.lock()?;
+            state.ensure_running()?;
+            state.refresh_uncertain_worker_write(worker_ref)?;
+            let worker = state.worker(worker_ref)?;
+            if worker.pending_stop.is_some() {
+                return Ok(Step::Finished);
+            }
+            let owner = worker
+                .restore_guard
+                .active_request_id
+                .as_ref()
+                .and_then(|id| worker.restore_guard.owners.get(id));
+            if owner.is_some_and(|owner| owner.operation_id != operation || owner.receipt.is_some())
+            {
+                return Ok(Step::Finished);
+            }
+            let Some(pending) = worker.pending_restore else {
+                return Ok(Step::Finished);
+            };
+            if pending.operation_id != operation {
+                return Ok(Step::Finished);
+            }
+            if pending.commit == RestoreCommitFact::LegacyUnknown {
+                return Ok(Step::Retry(
+                    "Saved Restore has no authoritative commit evidence".into(),
+                ));
+            }
+            state.ensure_subject_not_reserved(worker_ref, false)?;
+            if let Some(owner) = owner {
+                owner.validate_workdir_preparation_authority(worker)?;
+            }
+            let Some(backend) = state.execution_backend.clone() else {
+                return Ok(Step::Finished);
+            };
+            (backend, pending)
+        };
+        match backend.restore_recovery_disposition(operation, worker_ref) {
+            None => return Ok(Step::Finished),
+            Some(RestoreRecoveryDisposition::InFlight) => return Ok(Step::Waiting),
+            Some(RestoreRecoveryDisposition::Discard)
+                if pending.commit == RestoreCommitFact::NotCommitted =>
+            {
+                match backend.discard_uncommitted_restore(operation, worker_ref) {
+                    WorkerExecutionSpawnResult::RolledBack(result) => {
+                        let receipt = RuntimeWorkerRestoreResult::failed(
+                            WorkerRestoreState::RolledBack,
+                            "worker_restore_rolled_back",
+                            result.message_or_default(),
+                        );
+                        self.settle_restore_rollback(worker_ref, &receipt)?;
+                        return Ok(Step::Finished);
+                    }
+                    result => {
+                        return Ok(Step::Retry(format!(
+                            "Restore cleanup remains unresolved: {result:?}"
+                        )));
+                    }
+                }
+            }
+            Some(_) => {}
+        }
+        let result = self.restore_admitted_worker_under_lock(worker_ref, pending.mode)?;
+        if backend.restore_recovery_disposition(operation, worker_ref)
+            == Some(RestoreRecoveryDisposition::InFlight)
+        {
+            return Ok(Step::Waiting);
+        }
+        if result.state == WorkerRestoreState::ReconciliationRequired {
+            Ok(Step::Retry(result.message.unwrap_or_else(|| {
+                "Restore reconciliation remains pending".into()
+            })))
+        } else {
+            Ok(Step::Finished)
+        }
+    }
+}
+
+fn run(
+    inner: Weak<Mutex<RuntimeState>>,
+    operations: Weak<Mutex<BTreeMap<WorkerId, Arc<Mutex<()>>>>>,
+    continuations: Weak<RestoreContinuations>,
+    owner: Arc<Continuation>,
+    worker_ref: WorkerRef,
+    operation: WorkerLifecycleOperationId,
+) {
+    let mut repeated = 0u32;
+    let mut last_reason = None;
+    loop {
+        let (Some(inner), Some(worker_operations), Some(restore_continuations)) = (
+            inner.upgrade(),
+            operations.upgrade(),
+            continuations.upgrade(),
+        ) else {
+            break;
+        };
+        owner
+            .schedule
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .parked = false;
+        let runtime = Runtime {
+            inner,
+            worker_operations,
+            restore_continuations,
+        };
+        let step = runtime.continue_restore_operation(&worker_ref, operation);
+        // No lifecycle lock or strong Runtime/registry reference survives waiting.
+        drop(runtime);
+        let mut schedule = owner
+            .schedule
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        if schedule.cancelled {
+            break;
+        }
+        let contended = matches!(&step, Ok(Step::Contended));
+        let reason = match step {
+            Ok(Step::Finished) => {
+                schedule.finished = true;
+                owner.wake.notify_all();
+                break;
+            }
+            Ok(Step::Waiting) | Ok(Step::Contended) => None,
+            Ok(Step::Retry(reason)) => Some(reason),
+            Err(error) => Some(sanitize_worker_create_failure_message(&error.to_string())),
+        };
+        if let Some(reason) = &reason {
+            repeated = repeated.saturating_add(1);
+            last_reason = Some(reason.clone());
+            schedule.blocked_reason = Some(reason.clone());
+        }
+        // Completed failed attempts can have signaled while this step joined
+        // them. Do not let our own unchanged failure reset the retry budget.
+        if reason.is_some() {
+            schedule.signaled = false;
+        }
+        if (!contended && reason.is_none()) || repeated >= 4 {
+            if repeated >= 4 {
+                schedule.parked = true;
+                owner.wake.notify_all();
+                tracing::warn!(worker_id = %worker_ref.worker_id, operation_id = %operation, reason = ?last_reason,
+                    "Restore continuation parked after bounded attempts; operation remains unresolved");
+            }
+            schedule = owner
+                .wake
+                .wait_while(schedule, |s| !s.signaled && !s.cancelled)
+                .unwrap_or_else(std::sync::PoisonError::into_inner);
+        } else {
+            let delay = Duration::from_millis(100 * (1u64 << repeated.min(4)));
+            schedule = owner
+                .wake
+                .wait_timeout_while(schedule, delay, |s| !s.signaled && !s.cancelled)
+                .unwrap_or_else(std::sync::PoisonError::into_inner)
+                .0;
+        }
+        if schedule.cancelled {
+            break;
+        }
+        if schedule.signaled {
+            schedule.signaled = false;
+            // A completion wakes this same owner, not a new request. Its next
+            // observed result, not the signal itself, determines progress.
+        }
+    }
+}

crates/worker-runtime/src/test_support_tests.rs

@@ -564,6 +564,7 @@ fn extraction_rejects_nonstopped_pending_schema_and_metadata_mismatch() {
                         operation_id: WorkerLifecycleOperationId::new(),
                         mode: PersistedWorkerRestoreMode::Explicit,
                         commit: crate::runtime::RestoreCommitFact::NotCommitted,
+                        source_repair_possible: false,
                         request: request(),
                         binding: None,
                         last_settled_status: WorkerStatus::Stopped,

crates/worker-runtime/src/worker_backend.rs

@@ -370,10 +370,23 @@ async fn prepare_restore_operation<F: RuntimeWorkerFactory + ?Sized>(
     factory: &F,
     request: WorkerExecutionRestoreRequest,
 ) -> Result<Box<dyn RuntimeRestoreCandidate>, RuntimeRestoreFailure> {
+    let prior_effects = request.source_repair_possible;
     factory.preflight_restore(&request).await.map_err(|_| {
-        RuntimeRestoreFailure::refused(RuntimeRestoreStage::Preflight, "restore preflight refused")
+        let mut failure = RuntimeRestoreFailure::refused(
+            RuntimeRestoreStage::Preflight,
+            "restore preflight refused",
+        );
+        failure.durable_effects_possible = prior_effects;
+        failure
     })?;
-    let repaired = factory.recover_restore_source(&request).await?;
+    let repaired = factory
+        .recover_restore_source(&request)
+        .await
+        .map_err(|mut failure| {
+            failure.durable_effects_possible |= prior_effects;
+            failure
+        })?
+        || prior_effects;
     let candidate = factory
         .prepare_restore(request)
         .await
@@ -804,6 +817,7 @@ struct RestoreFaults {
     cleanup_fail: AtomicBool,
     socket_failure: AtomicBool,
     publication_failure: AtomicBool,
+    publication_attempts: std::sync::atomic::AtomicUsize,
     factory_gate: Mutex<Option<Arc<tokio::sync::Notify>>>,
     factory_entered: Mutex<Option<mpsc::Sender<()>>>,
     prepared_gate: Mutex<Option<Arc<tokio::sync::Notify>>>,
@@ -2626,6 +2640,11 @@ impl RuntimeRestoreCandidate for ProfileRestoreCandidate {
         }
         let started = async {
             #[cfg(test)]
+            self.factory
+                .restore_faults
+                .publication_attempts
+                .fetch_add(1, Ordering::SeqCst);
+            #[cfg(test)]
             if self
                 .factory
                 .restore_faults
@@ -3120,6 +3139,7 @@ struct RuntimeWorkerExecution {
 struct PendingRuntimeRestore {
     request: WorkerExecutionRestoreRequest,
     task: tokio::sync::Mutex<Option<tokio::task::JoinHandle<()>>>,
+    task_in_flight: AtomicBool,
     result: Mutex<Option<Result<Box<dyn RuntimeRestoreCandidate>, RuntimeRestoreFailure>>>,
     activated: Mutex<Option<Result<RuntimeWorkerController, RuntimeRestoreFailure>>>,
     // Monotonic for this operation, not a claim that its current candidate has
@@ -3131,7 +3151,24 @@ struct PendingRuntimeRestore {
     failure: Mutex<Option<String>>,
 }
 
+struct RestoreTaskCompletion(Arc<PendingRuntimeRestore>);
+
+impl Drop for RestoreTaskCompletion {
+    fn drop(&mut self) {
+        self.0.task_in_flight.store(false, Ordering::Release);
+        self.0
+            .request
+            .context
+            .restore_progressed(self.0.request.operation_id);
+    }
+}
+
 impl PendingRuntimeRestore {
+    fn task_completion(self: &Arc<Self>) -> RestoreTaskCompletion {
+        self.task_in_flight.store(true, Ordering::Release);
+        RestoreTaskCompletion(self.clone())
+    }
+
     fn owns_prepared_candidate(&self) -> bool {
         matches!(
             self.result
@@ -3455,7 +3492,9 @@ where
             if task.is_some() {
                 return Err("restore task has not been joined".into());
             }
+            let completion = pending.task_completion();
             *task = Some(self.spawn_on_adapter_runtime(async move {
+                let _completion = completion;
                 if resume_activation
                     && !failure.needs_cleanup()
                     && (failure.durable_effects_possible
@@ -5340,6 +5379,7 @@ where
                 let pending = Arc::new(PendingRuntimeRestore {
                     request: request.clone(),
                     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),
@@ -5361,7 +5401,9 @@ where
                 resources.insert(worker_ref.clone(), Arc::clone(&pending));
                 let factory = Arc::clone(&self.factory);
                 let result_owner = Arc::clone(&pending);
+                let completion = pending.task_completion();
                 let task = self.spawn_on_adapter_runtime(async move {
+                    let _completion = completion;
                     let result = match prepare_restore_operation(factory.as_ref(), request.clone())
                         .await
                     {
@@ -5551,6 +5593,42 @@ where
         }
     }
 
+    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();
+        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
+                    .discard_result
+                    .lock()
+                    .ok()
+                    .is_some_and(|r| matches!(*r, Some(Ok(()))))
+                {
+                    // Cleanup is proven; prior repair effects may now require
+                    // reconstruction, not repeated discard of a closed candidate.
+                    D::Continue
+                } else if pending.discard_started.load(Ordering::Acquire)
+                    || pending
+                        .discard_result
+                        .lock()
+                        .ok()
+                        .is_some_and(|r| r.is_some())
+                {
+                    D::Discard
+                } else {
+                    D::Continue
+                }
+            }
+            _ => D::Continue,
+        })
+    }
+
     fn discard_uncommitted_restore(
         &self,
         operation_id: crate::execution::WorkerLifecycleOperationId,
@@ -5582,8 +5660,10 @@ where
             self.wait_pending_restore(&pending, self.spawn_restore_timeout)?;
             if !pending.discard_started.swap(true, Ordering::AcqRel) {
                 let owner = pending.clone();
+                let completion = pending.task_completion();
                 let task =
                     self.spawn_on_adapter_runtime(async move {
+                        let _completion = completion;
                         let retained = owner
                             .result
                             .lock()
@@ -5709,8 +5789,10 @@ where
                 pending.activation_started.store(true, Ordering::Release);
                 let owner = pending.clone();
                 let factory = self.factory.clone();
+                let completion = pending.task_completion();
                 let task =
                     self.spawn_on_adapter_runtime(async move {
+                        let _completion = completion;
                         let candidate = owner
                             .result
                             .lock()
@@ -6683,6 +6765,7 @@ mod tests {
             });
         store.write(&metadata).unwrap();
         WorkerExecutionRestoreRequest {
+            source_repair_possible: false,
             operation_id: crate::execution::WorkerLifecycleOperationId::new(),
             worker_ref: worker_ref.clone(),
             request,
@@ -7614,6 +7697,7 @@ mod tests {
             Arc::new(PendingRuntimeRestore {
                 request: direct_restore_request(worker_ref.clone()),
                 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),
@@ -7657,6 +7741,7 @@ mod tests {
         let pending = Arc::new(PendingRuntimeRestore {
             request: request.clone(),
             task: tokio::sync::Mutex::new(Some(failed)),
+            task_in_flight: AtomicBool::new(false),
             result: Mutex::new(None),
             activated: Mutex::new(None),
             activation_started: AtomicBool::new(false),
@@ -10156,6 +10241,7 @@ mod tests {
 
     fn direct_restore_request(worker_ref: WorkerRef) -> WorkerExecutionRestoreRequest {
         WorkerExecutionRestoreRequest {
+            source_repair_possible: false,
             operation_id: crate::execution::WorkerLifecycleOperationId::new(),
             request: create_request("pending operation"),
             workspace_scope: None,
@@ -10729,6 +10815,7 @@ mod tests {
             .with_runtime_store_dir(&runtime_store_dir)
             .with_remote_worker_mutation_identity(identity)
             .restore_controller(WorkerExecutionRestoreRequest {
+                source_repair_possible: false,
                 operation_id: crate::execution::WorkerLifecycleOperationId::new(),
                 worker_ref: worker_ref.clone(),
                 request,
@@ -10826,6 +10913,7 @@ mod tests {
             .with_runtime_store_dir(&runtime_store_dir)
             .with_controller_transport(WorkerControllerTransport::InProcess)
             .restore_controller(WorkerExecutionRestoreRequest {
+                source_repair_possible: false,
                 operation_id: crate::execution::WorkerLifecycleOperationId::new(),
                 worker_ref: worker_ref.clone(),
                 request: create_request("embedded restore"),
@@ -10990,6 +11078,7 @@ mod tests {
         runtime.stop_worker(&worker.worker_ref, None).unwrap();
         let operation_id = crate::execution::WorkerLifecycleOperationId::new();
         let restore_request = WorkerExecutionRestoreRequest {
+            source_repair_possible: false,
             operation_id,
             worker_ref: worker.worker_ref.clone(),
             request,
@@ -11383,6 +11472,7 @@ mod tests {
         let detail = runtime.create_worker(request.clone()).unwrap();
         runtime.stop_worker(&detail.worker_ref, None).unwrap();
         let restore = WorkerExecutionRestoreRequest {
+            source_repair_possible: false,
             operation_id: crate::execution::WorkerLifecycleOperationId::new(),
             worker_ref: detail.worker_ref.clone(),
             request,

crates/worker-runtime/src/worker_backend/restore_recovery.rs

@@ -33,6 +33,10 @@ impl ProfileRuntimeWorkerFactory {
         if !pending && !incomplete {
             return Ok(false);
         }
+        request
+            .context
+            .authorize_restore_source_repair(request.operation_id)
+            .map_err(|_| cause("canonical source repair authority is not durably confirmed"))?;
         // The accepted Runtime operation already owns this Session. Its stable
         // Session identity and sole committed seed, not a new restore request,
         // are used to reconcile an interrupted first-head publication.

crates/worker-runtime/src/worker_backend_restore_failure_tests.rs

@@ -688,6 +688,7 @@ fn profile_restore_prior_effects_survive_read_only_reprepare_failure() {
     for read_stage in [RuntimeRestoreStage::Metadata, RuntimeRestoreStage::Reconstruction] {
         let activation_stage = RuntimeRestoreStage::ActivationEffects;
         let fixture = ProfileFailureFixture::new(Duration::from_secs(5));
+    let _pause = fixture.runtime.test_pause_restore_continuation();
         let (artifacts, pinned) = fixture.pin_unclaimed_restore_upload();
         fixture.fail(activation_stage);
         let request = fixture
@@ -882,6 +883,7 @@ fn profile_restore_prior_effects_from_pending_head_repair_survive_read_only_repr
 #[serial_test::serial(worker_allocation)]
 fn profile_restore_prior_effects_survive_rebuilt_candidate_pre_effect_activation_failure() {
     let fixture = ProfileFailureFixture::new(Duration::from_secs(5));
+    let _pause = fixture.runtime.test_pause_restore_continuation();
     let (artifacts, pinned) = fixture.pin_unclaimed_restore_upload();
     fixture.fail(RuntimeRestoreStage::ActivationEffects);
     let request = fixture

crates/worker-runtime/src/worker_backend_restore_publication_tests.rs

@@ -94,6 +94,141 @@ mod publication {
         (release, received)
     }
 
+    #[test]
+    #[serial_test::serial(worker_allocation)]
+    fn autonomous_late_prepare_and_publication_settle_without_another_request() {
+        for publication_failure in [false, true] {
+            let mut fixture = Fixture::no_model();
+            Arc::get_mut(&mut fixture.backend).unwrap().spawn_restore_timeout = Duration::from_millis(250);
+            let runtime = stopped_runtime(&fixture);
+            let worker_ref = fixture.request.worker_ref.clone();
+            let request = runtime.test_restore_request(&worker_ref);
+            let (release, prepared) = prepared_gate(&fixture);
+            fixture.backend.factory.restore_faults.publication_failure.store(publication_failure, Ordering::SeqCst);
+            let first = runtime.restore_worker_operation_scoped(fixture.request.workspace_scope.as_ref().unwrap(), &worker_ref, request.clone()).unwrap();
+            prepared.recv_timeout(WAIT).unwrap();
+            assert_eq!(first.state, server_api::WorkerRestoreState::ReconciliationRequired);
+            let operation = durable_operation(&runtime, &worker_ref).unwrap();
+            assert_eq!(operation.commit, crate::runtime::RestoreCommitFact::NotCommitted);
+            fixture.assert_no_allocation();
+            release.notify_one();
+            runtime.test_wait_restore_continuation(&worker_ref, operation.operation_id, WAIT);
+            let loaded = runtime.test_runtime_store().load_runtime_state().unwrap();
+            let receipt = serde_json::to_value(&loaded.workers[&worker_ref.worker_id].restore_guard).unwrap();
+            assert_eq!(receipt["owners"][&request.request_id]["receipt"]["state"], "accepted");
+            assert!(durable_operation(&runtime, &worker_ref).is_none());
+            let transport = fixture.backend.clone().attach_worker_protocol(&worker_ref).unwrap();
+            transport.validate().unwrap();
+            drop(transport);
+            assert_eq!(fixture.backend.factory.restore_faults.executions.load(Ordering::SeqCst), 1);
+            fixture.assert_quiet();
+            runtime.stop_worker_scoped(fixture.request.workspace_scope.as_ref().unwrap(), &worker_ref, None).unwrap();
+            fixture.assert_released();
+        }
+    }
+
+    #[test]
+    #[serial_test::serial(worker_allocation)]
+    fn autonomous_cleanup_dispatch_settles_uncommitted_candidate_without_repost_or_stop() {
+        let fixture = Fixture::no_model();
+        let runtime = Arc::new(stopped_runtime(&fixture));
+        let pause = runtime.test_pause_restore_continuation();
+        let worker_ref = fixture.request.worker_ref.clone();
+        let scope = fixture.request.workspace_scope.clone().unwrap();
+        let request = runtime.test_restore_request(&worker_ref);
+        let before = fixture.canonical_files();
+        let (release, prepared) = prepared_gate(&fixture);
+        let restoring = {
+            let runtime = runtime.clone();
+            let worker_ref = worker_ref.clone();
+            let scope = scope.clone();
+            let request = request.clone();
+            std::thread::spawn(move || runtime.restore_worker_operation_scoped(&scope, &worker_ref, request))
+        };
+        prepared.recv_timeout(WAIT).unwrap();
+        runtime.test_runtime_store().fail_next_worker_write(AtomicWriteFault::BeforeRename);
+        fixture.backend.factory.restore_faults.cleanup_fail.store(true, Ordering::SeqCst);
+        release.notify_one();
+        let first = restoring.join().unwrap().unwrap();
+        assert_eq!(first.state, server_api::WorkerRestoreState::ReconciliationRequired);
+        let operation = durable_operation(&runtime, &worker_ref).unwrap();
+        assert_eq!(operation.commit, crate::runtime::RestoreCommitFact::NotCommitted);
+        let pending = fixture.backend.pending_restore(&worker_ref).unwrap().unwrap();
+        assert!(matches!(pending.discard_result.lock().unwrap().as_ref(), Some(Err(_))));
+        fixture.backend.factory.restore_faults.cleanup_fail.store(false, Ordering::SeqCst);
+        drop(pending);
+        drop(pause);
+        runtime.test_wait_restore_continuation(&worker_ref, operation.operation_id, WAIT);
+        let loaded = runtime.test_runtime_store().load_runtime_state().unwrap();
+        let receipt = serde_json::to_value(&loaded.workers[&worker_ref.worker_id].restore_guard).unwrap();
+        assert_eq!(receipt["owners"][&request.request_id]["receipt"]["state"], "rolled_back");
+        assert_eq!(fixture.canonical_files(), before);
+        assert_eq!(fixture.backend.factory.restore_faults.executions.load(Ordering::SeqCst), 1);
+        fixture.assert_released();
+        fixture.assert_quiet();
+    }
+
+    #[test]
+    #[serial_test::serial(worker_allocation)]
+    fn autonomous_unchanged_publication_obstruction_parks_without_false_terminal_outcome() {
+        let fixture = Fixture::no_model();
+        let runtime = stopped_runtime(&fixture);
+        let worker_ref = fixture.request.worker_ref.clone();
+        fixture.backend.factory.restore_faults.socket_failure.store(true, Ordering::SeqCst);
+        let request = runtime.test_restore_request(&worker_ref);
+        let first = runtime.restore_worker_operation_scoped(fixture.request.workspace_scope.as_ref().unwrap(), &worker_ref, request.clone()).unwrap();
+        assert_eq!(first.state, server_api::WorkerRestoreState::ReconciliationRequired);
+        let pending = fixture.backend.pending_restore(&worker_ref).unwrap().unwrap();
+        let reason = runtime.test_wait_restore_continuation_parked(&worker_ref, pending.request.operation_id, WAIT);
+        assert!(reason.contains("publication failed"), "{reason}");
+        let durable = durable_operation(&runtime, &worker_ref).unwrap();
+        assert_eq!(durable.commit, crate::runtime::RestoreCommitFact::Committed);
+        let loaded = runtime.test_runtime_store().load_runtime_state().unwrap();
+        let guard = serde_json::to_value(&loaded.workers[&worker_ref.worker_id].restore_guard).unwrap();
+        assert!(guard["owners"][&request.request_id]["receipt"].is_null());
+        assert_eq!(fixture.backend.factory.restore_faults.executions.load(Ordering::SeqCst), 1);
+        // Initial call plus at most four backed-off continuation attempts. The
+        // parked condition is observed directly, not guessed using a sleep.
+        assert!(fixture.backend.factory.restore_faults.publication_attempts.load(Ordering::SeqCst) <= 5);
+        assert!(fixture.backend.workers.lock().unwrap().is_empty());
+        assert!(fixture.backend.factory.observation_hub.get(&worker_ref).is_none());
+        drop(pending);
+        // Permanent path-length obstruction: explicit fixture teardown only,
+        // not the successful completion path asserted by autonomous tests.
+        runtime.stop_worker_scoped(fixture.request.workspace_scope.as_ref().unwrap(), &worker_ref, None).unwrap();
+        fixture.assert_released();
+        fixture.assert_quiet();
+    }
+
+    #[test]
+    #[serial_test::serial(worker_allocation)]
+    fn autonomous_late_failure_settles_without_another_request() {
+        let mut fixture = Fixture::no_model();
+        Arc::get_mut(&mut fixture.backend).unwrap().spawn_restore_timeout = Duration::from_millis(250);
+        let runtime = stopped_runtime(&fixture);
+        let worker_ref = fixture.request.worker_ref.clone();
+        let request = runtime.test_restore_request(&worker_ref);
+        let release = Arc::new(tokio::sync::Notify::new());
+        let (entered, prepared) = mpsc::channel();
+        *fixture.backend.factory.restore_faults.factory_gate.lock().unwrap() = Some(release.clone());
+        *fixture.backend.factory.restore_faults.factory_entered.lock().unwrap() = Some(entered);
+        *fixture.backend.factory.restore_faults.fail_at.lock().unwrap() = Some(RuntimeRestoreStage::Feature);
+        let before = fixture.canonical_files();
+        let first = runtime.restore_worker_operation_scoped(fixture.request.workspace_scope.as_ref().unwrap(), &worker_ref, request.clone()).unwrap();
+        prepared.recv_timeout(WAIT).unwrap();
+        assert_eq!(first.state, server_api::WorkerRestoreState::ReconciliationRequired);
+        let operation = durable_operation(&runtime, &worker_ref).unwrap();
+        release.notify_one();
+        runtime.test_wait_restore_continuation(&worker_ref, operation.operation_id, WAIT);
+        let loaded = runtime.test_runtime_store().load_runtime_state().unwrap();
+        let receipt = serde_json::to_value(&loaded.workers[&worker_ref.worker_id].restore_guard).unwrap();
+        assert_eq!(receipt["owners"][&request.request_id]["receipt"]["state"], "rolled_back");
+        assert_eq!(fixture.canonical_files(), before);
+        assert_eq!(fixture.backend.factory.restore_faults.executions.load(Ordering::SeqCst), 1);
+        fixture.assert_released();
+        fixture.assert_quiet();
+    }
+
     #[test]
     #[serial_test::serial(worker_allocation)]
     fn runtime_accepted_follows_real_prepared_snapshot_commit_attach_and_input() {
@@ -288,6 +423,7 @@ mod publication {
         ] {
             let fixture = Fixture::no_model();
             let runtime = Arc::new(stopped_runtime(&fixture));
+            let _pause = runtime.test_pause_restore_continuation();
             let worker_ref = fixture.request.worker_ref.clone();
             let scope = fixture.request.workspace_scope.clone().unwrap();
             let request = runtime.test_restore_request(&worker_ref);
@@ -396,6 +532,7 @@ mod publication {
     fn uncommitted_cleanup_failure_fences_discarding_candidate_from_activation() {
         let fixture = Fixture::no_model();
         let runtime = Arc::new(stopped_runtime(&fixture));
+        let _pause = runtime.test_pause_restore_continuation();
         let worker_ref = fixture.request.worker_ref.clone();
         let scope = fixture.request.workspace_scope.clone().unwrap();
         let request = runtime.test_restore_request(&worker_ref);
@@ -443,6 +580,7 @@ mod publication {
     fn publication_failure_retains_same_committed_candidate_without_reconstruction() {
         let fixture = Fixture::no_model();
         let runtime = stopped_runtime(&fixture);
+        let _pause = runtime.test_pause_restore_continuation();
         let worker_ref = fixture.request.worker_ref.clone();
         let scope = fixture.request.workspace_scope.clone().unwrap();
         let request = runtime.test_restore_request(&worker_ref);