fix: preserve restore effects and fence prepared attachment replay

crates/worker-runtime/src/runtime.rs

@@ -2150,6 +2150,16 @@ impl Runtime {
                 {
                     return Err(conflict());
                 }
+                if let Err(error) = owner.validate_workdir_preparation_authority(&candidate) {
+                    return Ok((
+                        Some(RuntimeWorkerRestoreResult::failed(
+                            WorkerRestoreState::ReconciliationRequired,
+                            "worker_restore_authority_changed",
+                            sanitize_worker_create_failure_message(&error.to_string()),
+                        )),
+                        None,
+                    ));
+                }
             } else {
                 // Startup/older persisted pending operations have a lifecycle
                 // owner but no public request receipt. Bind recovery to that
@@ -2215,12 +2225,18 @@ impl Runtime {
                 candidate.pending_restore = Some(pending);
                 candidate.restore_guard.rotate();
                 candidate.restore_guard.active_request_id = Some(request.request_id.clone());
+                let workdir_attachment_preparation_base = request
+                    .preparation
+                    .as_ref()
+                    .and_then(|preparation| preparation.workdir_attachments.as_ref())
+                    .map(|_| candidate.logical_workdir_attachments.clone());
                 candidate.restore_guard.owners.insert(
                     request.request_id.clone(),
                     RestoreRequestOwner {
                         fingerprint,
                         operation_id,
                         request,
+                        workdir_attachment_preparation_base,
                         preparation_applied: recovering_pending,
                         execution_started: recovering_pending,
                         receipt: None,
@@ -2433,14 +2449,15 @@ impl Runtime {
             {
                 return Ok(RuntimeWorkerRestoreResult::accepted(worker.detail()));
             }
-            let preparation = worker
+            let owner = worker
                 .restore_guard
                 .active_request_id
                 .as_ref()
-                .and_then(|id| worker.restore_guard.owners.get(id))
+                .and_then(|id| worker.restore_guard.owners.get(id));
-                .filter(|owner| !owner.preparation_applied)
+            let prepared_worker = match owner.map_or_else(
-                .and_then(|owner| owner.request.preparation.as_ref());
+                || Ok(worker.clone()),
-            let prepared_worker = match worker.prepared_restore_record(preparation) {
+                |owner| owner.prepared_worker_record(worker),
+            ) {
                 Ok(prepared) => prepared,
                 Err(error) => {
                     return Ok(RuntimeWorkerRestoreResult::failed(
@@ -4398,8 +4415,7 @@ impl Runtime {
                     worker_id: worker_ref.worker_id,
                 });
             }
-            let mut candidate =
+            let mut candidate = owner.prepared_worker_record(previous)?;
-                previous.prepared_restore_record(owner.request.preparation.as_ref())?;
             candidate
                 .restore_guard
                 .owners
@@ -6277,12 +6293,48 @@ pub(crate) struct RestoreRequestOwner {
     fingerprint: String,
     operation_id: WorkerLifecycleOperationId,
     request: runtime_api::WorkerRestoreRequest,
+    // This is the admission comparison input, not a second attachment authority.
+    // Without it an old preparation could hide a later canonical revocation.
+    // Older owners lacking this fact cannot safely reapply attachment overrides.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    workdir_attachment_preparation_base: Option<Vec<LogicalWorkdirAttachment>>,
     preparation_applied: bool,
     #[serde(default)]
     execution_started: bool,
     receipt: Option<RuntimeWorkerRestoreResult>,
 }
 
+impl RestoreRequestOwner {
+    fn validate_workdir_preparation_authority(
+        &self,
+        worker: &WorkerRecord,
+    ) -> Result<(), RuntimeError> {
+        if !self.preparation_applied
+            && self
+                .request
+                .preparation
+                .as_ref()
+                .is_some_and(|preparation| preparation.workdir_attachments.is_some())
+            && self.workdir_attachment_preparation_base.as_ref()
+                != Some(&worker.logical_workdir_attachments)
+        {
+            return Err(RuntimeError::InvalidRequest(
+                "current Worker authority differs from admitted Restore attachment preparation"
+                    .into(),
+            ));
+        }
+        Ok(())
+    }
+
+    fn prepared_worker_record(&self, worker: &WorkerRecord) -> Result<WorkerRecord, RuntimeError> {
+        if self.preparation_applied {
+            return Ok(worker.clone());
+        }
+        self.validate_workdir_preparation_authority(worker)?;
+        worker.prepared_restore_record(self.request.preparation.as_ref())
+    }
+}
+
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
 #[serde(deny_unknown_fields)]
 pub(crate) struct RestoreGuard {
@@ -11995,7 +12047,7 @@ mod tests {
         let mut request = runtime.test_restore_request(&worker.worker_ref);
         request.preparation = Some(serde_json::from_value(serde_json::json!({
             "workspace_api": { "workspace_id": "workspace-a", "base_url": "https://prepared.example" },
-            "workdir_attachments": [], "repository_access": []
+            "workdir_attachments": [{ "alias": "checkout", "working_directory_id": "prepared-logical-workdir" }], "repository_access": []
         })).unwrap());
         let gate = Arc::new(RestoreGate::default());
         *backend.restore_gate.lock().unwrap() = Some(gate.clone());
@@ -12014,6 +12066,26 @@ mod tests {
             let state = runtime.lock().unwrap();
             let record = state.worker(&worker.worker_ref).unwrap();
             assert!(record.request.as_ref().unwrap().workspace_api.is_none());
+            assert!(record.logical_workdir_attachments.is_empty());
+            let owner = record.restore_guard.owners.values().next().unwrap();
+            let mut saved_owner = serde_json::to_value(owner).unwrap();
+            let decoded: RestoreRequestOwner = serde_json::from_value(saved_owner.clone()).unwrap();
+            assert!(
+                decoded
+                    .validate_workdir_preparation_authority(record)
+                    .is_ok()
+            );
+            saved_owner
+                .as_object_mut()
+                .unwrap()
+                .remove("workdir_attachment_preparation_base");
+            let legacy: RestoreRequestOwner = serde_json::from_value(saved_owner).unwrap();
+            assert!(
+                legacy
+                    .validate_workdir_preparation_authority(record)
+                    .is_err(),
+                "a missing historical comparison input must not be invented from current authority"
+            );
             assert!(
                 !record
                     .restore_guard
@@ -12033,6 +12105,11 @@ mod tests {
                 .base_url,
             "https://prepared.example"
         );
+        assert_eq!(
+            backend.restore_requests.lock().unwrap()[0].logical_workdir_attachments[0]
+                .working_directory_id,
+            "prepared-logical-workdir"
+        );
         assert!(backend.activation_operation_ids.lock().unwrap().is_empty());
         gate.release();
         assert_eq!(
@@ -12053,6 +12130,25 @@ mod tests {
                 .base_url,
             "https://prepared.example"
         );
+        let record = state.worker(&worker.worker_ref).unwrap();
+        assert_eq!(
+            record.logical_workdir_attachments[0].working_directory_id,
+            "prepared-logical-workdir"
+        );
+        let owner = record.restore_guard.owners.values().next().unwrap();
+        assert!(owner.preparation_applied);
+        let mut revoked = record.clone();
+        revoked
+            .apply_logical_workdir_attachments(Vec::new())
+            .unwrap();
+        assert!(
+            owner
+                .prepared_worker_record(&revoked)
+                .unwrap()
+                .logical_workdir_attachments
+                .is_empty(),
+            "a committed preparation is not a perpetual grant"
+        );
     }
 
     #[cfg(feature = "ws-server")]

crates/worker-runtime/src/worker_backend.rs

@@ -360,20 +360,21 @@ async fn prepare_restore_operation<F: RuntimeWorkerFactory + ?Sized>(
             failure
         })?;
     if repaired {
-        Ok(Box::new(RepairedRestoreCandidate { candidate }))
+        Ok(Box::new(EffectRetainingRestoreCandidate { candidate }))
     } else {
         Ok(candidate)
     }
 }
 
-// A completed repair is a canonical effect even when later activation fails
+// Completed repair or an earlier activation may have changed canonical state.
-// before its own mutation boundary. Carry that evidence with the resource owner.
+// Keep that operation-owned evidence through reconstruction and even an
-struct RepairedRestoreCandidate {
+// activation failure before the replacement candidate's own mutation boundary.
+struct EffectRetainingRestoreCandidate {
     candidate: Box<dyn RuntimeRestoreCandidate>,
 }
 
 #[async_trait]
-impl RuntimeRestoreCandidate for RepairedRestoreCandidate {
+impl RuntimeRestoreCandidate for EffectRetainingRestoreCandidate {
     fn initial_state(&self) -> protocol::WorkerStateSnapshot {
         self.candidate.initial_state()
     }
@@ -2953,6 +2954,9 @@ struct PendingRuntimeRestore {
     task: tokio::sync::Mutex<Option<tokio::task::JoinHandle<()>>>,
     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
+    // started. A reconstructed candidate has a new result and no completion in
+    // `activated`; prior activation/effect evidence must survive that handoff.
     activation_started: AtomicBool,
     discard_started: AtomicBool,
     discard_result: Mutex<Option<Result<(), RuntimeRestoreCleanupFailure>>>,
@@ -2960,6 +2964,16 @@ struct PendingRuntimeRestore {
 }
 
 impl PendingRuntimeRestore {
+    fn owns_prepared_candidate(&self) -> bool {
+        matches!(
+            self.result
+                .lock()
+                .unwrap_or_else(std::sync::PoisonError::into_inner)
+                .as_ref(),
+            Some(Ok(_))
+        )
+    }
+
     async fn wait(&self, timeout: Duration) -> Result<(), String> {
         let mut task = self.task.lock().await;
         if let Some(message) = self
@@ -3279,23 +3293,40 @@ where
                     && (failure.durable_effects_possible
                         || owner.activation_started.load(Ordering::Acquire))
                 {
+                    let prior_effects = failure.durable_effects_possible
+                        || owner.activation_started.load(Ordering::Acquire);
                     let result =
                         match prepare_restore_operation(factory.as_ref(), owner.request.clone())
                             .await
                         {
-                            Ok(candidate) => Ok(candidate),
+                            Ok(candidate) => {
-                            Err(failure) => Err(finish_restore_failure(
+                                Ok(Box::new(EffectRetainingRestoreCandidate { candidate })
+                                    as Box<dyn RuntimeRestoreCandidate>)
+                            }
+                            Err(mut failure) => {
+                                failure.durable_effects_possible |= prior_effects;
+                                Err(finish_restore_failure(
                                     factory.as_ref(),
                                     &owner.request,
                                     failure,
                                 )
-                            .await),
+                                .await)
+                            }
+                        };
+                    // A completed reprepare replaces attempt resources, never
+                    // the operation's effect history. Only a ready candidate
+                    // may clear the old activation completion for a new attempt.
+                    let activated = match &result {
+                        Ok(_) => None,
+                        Err(failure) if owner.activation_started.load(Ordering::Acquire) => {
+                            Some(Err(failure.clone()))
+                        }
+                        Err(_) => None,
                     };
-                    owner.activation_started.store(false, Ordering::Release);
                     *owner
                         .activated
                         .lock()
-                        .unwrap_or_else(std::sync::PoisonError::into_inner) = None;
+                        .unwrap_or_else(std::sync::PoisonError::into_inner) = activated;
                     *owner
                         .result
                         .lock()
@@ -4019,7 +4050,9 @@ where
                     failure.to_string(),
                 );
             }
-            if pending.activation_started.load(Ordering::Acquire) {
+            if pending.activation_started.load(Ordering::Acquire)
+                && !pending.owns_prepared_candidate()
+            {
                 let mut activated = match pending.activated.lock() {
                     Ok(result) => result,
                     Err(_) => {
@@ -5205,7 +5238,8 @@ where
                 };
             }
         }
-        if pending.activation_started.load(Ordering::Acquire) {
+        if pending.activation_started.load(Ordering::Acquire) && !pending.owns_prepared_candidate()
+        {
             let activated = pending
                 .activated
                 .lock()
@@ -5330,7 +5364,8 @@ where
                 return Err("restored Worker operation identity does not match".into());
             }
             self.wait_pending_restore(&pending, self.spawn_restore_timeout)?;
-            if !pending.activation_started.swap(true, Ordering::AcqRel) {
+            if pending.owns_prepared_candidate() {
+                pending.activation_started.store(true, Ordering::Release);
                 let owner = pending.clone();
                 let factory = self.factory.clone();
                 let task = self.spawn_on_adapter_runtime(async move {

crates/worker-runtime/src/worker_backend_restore_candidate_tests.rs

@@ -895,4 +895,210 @@ mod restore_candidate {
             assert_eq!(model.call_count.load(Ordering::SeqCst), 0);
         }
     }
+
+    #[test]
+    #[serial_test::serial(worker_allocation)]
+    fn attachment_preparation_replay_cannot_hide_logical_authority_revocation() {
+        use crate::fs_store::{
+            FsRuntimeStoreOptions, PersistedWorkerExecution, PersistedWorkerExecutionState,
+            PersistedWorkerRecord,
+        };
+        let mut fixture = Fixture::new();
+        Arc::get_mut(&mut fixture.backend)
+            .unwrap()
+            .spawn_restore_timeout = Duration::from_secs(2);
+        let worker_ref = fixture.request.worker_ref.clone();
+        let scope = fixture.request.workspace_scope.clone().unwrap();
+        let options = FsRuntimeStoreOptions {
+            root: fixture._root.path().join("catalog"),
+            runtime_id: "candidate-runtime".into(),
+            display_name: None,
+        };
+        let attachment = crate::catalog::LogicalWorkdirAttachment {
+            alias: WorkdirAttachmentAlias::new("checkout").unwrap(),
+            working_directory_id: "workspace-only-workdir".into(),
+            capabilities: workdir::WorkdirSessionCapabilities::READ_ONLY,
+        };
+        // Only the initial Runtime catalog is seeded. The saved Profile, queue,
+        // restore admission, revocation, replay and Stop use their real owners.
+        let runtime = EmbeddedRuntime::with_fs_store_and_execution_backend(
+            options.clone(),
+            fixture.backend.clone(),
+        )
+        .unwrap();
+        let create = fixture.request.request.clone();
+        runtime
+            .test_runtime_store()
+            .write_worker_record(&PersistedWorkerRecord {
+                restore_guard: crate::runtime::RestoreGuard::default(),
+                worker_ref: worker_ref.clone(),
+                worker_id: worker_ref.worker_id,
+                profile: create.profile.clone(),
+                display_name: None,
+                profile_source: create.profile_source.reference(),
+                config_bundle: create.config_bundle.clone(),
+                created_at_ms: None,
+                status: crate::catalog::WorkerStatus::Stopped,
+                last_finished_submission_request_id: None,
+                execution_state: PersistedWorkerExecutionState::Available(
+                    PersistedWorkerExecution {
+                        request: create,
+                        binding: None,
+                        restore_intent: crate::catalog::WorkerRestoreIntent::Explicit,
+                    },
+                ),
+                workspace_id: Some(scope.workspace_id.clone()),
+                workdir_attachments: Vec::new(),
+                logical_workdir_attachments: vec![attachment.clone()],
+            })
+            .unwrap();
+        drop(runtime);
+        let runtime = EmbeddedRuntime::with_fs_store_and_execution_backend(
+            options.clone(),
+            fixture.backend.clone(),
+        )
+        .unwrap();
+        assert!(
+            runtime
+                .worker_detail(&worker_ref)
+                .unwrap()
+                .workdir_attachments
+                .is_empty()
+        );
+        let before = fixture.canonical_files();
+        let authority = || -> serde_json::Value {
+            serde_json::from_slice(
+                &fs::read(
+                    options
+                        .root
+                        .join("workers")
+                        .join(worker_ref.worker_id.to_string())
+                        .join("worker.json"),
+                )
+                .unwrap(),
+            )
+            .unwrap()
+        };
+        let release = Arc::new(tokio::sync::Notify::new());
+        let (entered_tx, entered) = 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_tx);
+        let mut request = runtime.test_restore_request(&worker_ref);
+        request.preparation.as_mut().unwrap().workdir_attachments =
+            Some(serde_json::from_value(serde_json::to_value(vec![attachment]).unwrap()).unwrap());
+        let first = runtime
+            .restore_worker_operation_scoped(&scope, &worker_ref, request.clone())
+            .unwrap();
+        entered.recv_timeout(WAIT).unwrap();
+        assert_eq!(
+            first.state,
+            server_api::WorkerRestoreState::ReconciliationRequired,
+            "{first:?}"
+        );
+        let pending = fixture
+            .backend
+            .pending_restore(&worker_ref)
+            .unwrap()
+            .unwrap();
+        assert!(!pending.request.logical_workdir_attachments.is_empty());
+        assert!(pending.request.previous_workdir_attachments.is_empty());
+        assert!(pending.request.deferred_workdir_sessions.is_empty());
+        runtime
+            .replace_worker_workdir_attachments_scoped(&scope, &worker_ref, Vec::new())
+            .unwrap();
+        assert!(authority().get("logical_workdir_attachments").is_none());
+        release.notify_one();
+        let replay = runtime
+            .restore_worker_operation_scoped(&scope, &worker_ref, request.clone())
+            .unwrap();
+        assert_eq!(
+            replay.state,
+            server_api::WorkerRestoreState::ReconciliationRequired,
+            "revocation hidden by saved preparation: {replay:?}"
+        );
+        assert!(
+            replay
+                .message
+                .as_deref()
+                .unwrap_or_default()
+                .contains("current Worker authority differs"),
+            "{replay:?}"
+        );
+        let record = authority();
+        assert!(record.get("logical_workdir_attachments").is_none());
+        let owner = &record["restore_guard"]["owners"][&request.request_id];
+        assert_eq!(
+            owner["operation_id"],
+            serde_json::to_value(pending.request.operation_id).unwrap()
+        );
+        assert!(owner["receipt"].is_null());
+        assert_eq!(
+            owner["workdir_attachment_preparation_base"]
+                .as_array()
+                .unwrap()
+                .len(),
+            1
+        );
+        let loaded = runtime.test_runtime_store().load_runtime_state().unwrap();
+        let durable_guard =
+            serde_json::to_value(&loaded.workers[&worker_ref.worker_id].restore_guard).unwrap();
+        assert_eq!(durable_guard["owners"][&request.request_id], *owner);
+        let (coordinated, preparation) = 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: None,
+                },
+                Some(&scope),
+            )
+            .unwrap();
+        assert_eq!(
+            coordinated.unwrap().state,
+            server_api::WorkerRestoreState::ReconciliationRequired
+        );
+        assert!(
+            preparation.is_none(),
+            "coordinator must not export the revoked preparation again"
+        );
+        assert!(fixture.backend.workers.lock().unwrap().is_empty());
+        assert!(Arc::ptr_eq(
+            &pending,
+            &fixture
+                .backend
+                .pending_restore(&worker_ref)
+                .unwrap()
+                .unwrap()
+        ));
+        assert!(!pending.activation_started.load(Ordering::SeqCst));
+        fixture.assert_quiet();
+        assert_eq!(fixture.canonical_files(), before);
+        runtime
+            .stop_worker_scoped(&scope, &worker_ref, None)
+            .unwrap();
+        fixture.assert_released();
+        fixture.assert_quiet();
+        assert_eq!(fixture.canonical_files(), before);
+        assert!(authority().get("logical_workdir_attachments").is_none());
+        assert_ne!(
+            runtime
+                .restore_worker_operation_scoped(&scope, &worker_ref, request)
+                .unwrap()
+                .state,
+            server_api::WorkerRestoreState::Accepted
+        );
+    }
 }

crates/worker-runtime/src/worker_backend_restore_failure_tests.rs

@@ -100,6 +100,148 @@ impl WorkingDirectoryMaterializer for RestoreOpenCountingMaterializer {
     }
 }
 
+// Substitute only initial creation: a new aggregate/Session is left pending
+// before its first head, without ever materializing or rewinding a saved head.
+// All Restore preflight, repair, retry, activation and Stop use the real backend.
+struct PendingProfileSeedBackend {
+    inner: Arc<WorkerRuntimeExecutionBackend>,
+}
+
+impl WorkerExecutionBackend for PendingProfileSeedBackend {
+    fn backend_id(&self) -> &str {
+        self.inner.backend_id()
+    }
+
+    fn spawn_worker(&self, request: WorkerExecutionSpawnRequest) -> WorkerExecutionSpawnResult {
+        let aggregate = self
+            .inner
+            .factory
+            .worker_aggregate_dir(&request.worker_ref)
+            .unwrap();
+        let name = ProfileRuntimeWorkerFactory::runtime_worker_name_for_ref(&request.worker_ref);
+        self.inner
+            .run_on_adapter_runtime(async move {
+                assert!(
+                    !aggregate.join("session").exists(),
+                    "pending fixture requires a new Session aggregate"
+                );
+                let manifest = manifest::WorkerManifest::from_toml(&format!(
+                    r#"
+[worker]
+name = "{name}"
+pwd = "./"
+[scope]
+allow = []
+[model]
+scheme = "anthropic"
+model_id = "test-model"
+auth = {{ kind = "none" }}
+[engine]
+max_tokens = 100
+"#
+                ))
+                .unwrap();
+                let sessions = WorkerSessionStore::new(aggregate.join("session")).unwrap();
+                let metadata = WorkerAggregateStore::new(&aggregate, &name).unwrap();
+                assert!(metadata.read_by_name(&name).unwrap().is_none());
+                let engine =
+                    Engine::<_, agen::state::Mutable, worker::SessionHistoryMetadata>::new_annotated(
+                        MockClient::new(Vec::new()),
+                    );
+                let mut head = Worker::new(
+                    manifest.clone(),
+                    engine,
+                    CombinedStore::new(sessions.clone(), metadata.clone()),
+                    WorkerWorkspaceContext::unavailable(
+                        None,
+                        "initial creation interrupted before first head",
+                    ),
+                    WorkerFilesystemAuthority::None,
+                    Scope::empty(),
+                )
+                .await
+                .unwrap();
+                head.enable_worker_metadata_write_through().unwrap();
+                let mut saved = metadata.read_by_name(&name).unwrap().unwrap();
+                let active = saved.active.as_ref().unwrap();
+                assert!(active.segment_id.is_none());
+                assert!(sessions.list_segments(active.session_id).unwrap().is_empty());
+                saved.resolved_manifest_snapshot =
+                    Some(manifest::write_persisted_worker_manifest_snapshot(&manifest).unwrap());
+                metadata.write(&saved).unwrap();
+                drop(head);
+                Ok(())
+            })
+            .unwrap();
+        WorkerExecutionSpawnResult::Connected {
+            worker_state: protocol::WorkerStateSnapshot::initial(),
+            workdir_attachments: Vec::new(),
+        }
+    }
+
+    fn fetch_workspace_config(
+        &self,
+        request: WorkspaceConfigFetchRequest,
+    ) -> Result<WorkspaceConfigFetchResult, String> {
+        self.inner.fetch_workspace_config(request)
+    }
+
+    fn observe_workspace_prompt_projection(
+        &self,
+        projection: worker::WorkspacePromptProjection,
+    ) -> Result<(), String> {
+        self.inner.observe_workspace_prompt_projection(projection)
+    }
+
+    fn preflight_restore(
+        &self,
+        request: &WorkerExecutionRestoreRequest,
+    ) -> Result<(), WorkerExecutionResult> {
+        self.inner.preflight_restore(request)
+    }
+
+    fn completed_restore_failure(
+        &self,
+        operation_id: crate::execution::WorkerLifecycleOperationId,
+        worker_ref: &WorkerRef,
+    ) -> Option<WorkerExecutionSpawnResult> {
+        self.inner
+            .completed_restore_failure(operation_id, worker_ref)
+    }
+
+    fn restore_worker(&self, request: WorkerExecutionRestoreRequest) -> WorkerExecutionSpawnResult {
+        self.inner.restore_worker(request)
+    }
+
+    fn reconcile_restore(
+        &self,
+        request: WorkerExecutionRestoreRequest,
+    ) -> WorkerExecutionSpawnResult {
+        self.inner.reconcile_restore(request)
+    }
+
+    fn activate_restored_worker(
+        &self,
+        operation_id: crate::execution::WorkerLifecycleOperationId,
+        worker_ref: &WorkerRef,
+    ) -> Result<(), String> {
+        self.inner
+            .activate_restored_worker(operation_id, worker_ref)
+    }
+
+    fn dispatch_input(&self, worker_ref: &WorkerRef, input: WorkerInput) -> WorkerExecutionResult {
+        self.inner.dispatch_input(worker_ref, input)
+    }
+
+    fn dispatch_method(&self, worker_ref: &WorkerRef, method: Method) -> WorkerExecutionResult {
+        self.inner.dispatch_method(worker_ref, method)
+    }
+
+    fn stop_worker_operation(&self, request: WorkerExecutionStopRequest) -> WorkerExecutionResult {
+        self.inner.stop_worker_operation(request)
+    }
+}
+
 struct ProfileFailureFixture {
     runtime: EmbeddedRuntime,
     backend: Arc<WorkerRuntimeExecutionBackend>,
@@ -115,6 +257,14 @@ impl ProfileFailureFixture {
     }
 
     fn with_saved_inputs(timeout: Duration, model: Option<MockClient>) -> Self {
+        Self::with_seed(timeout, model, false)
+    }
+
+    fn with_pending_head(timeout: Duration) -> Self {
+        Self::with_seed(timeout, None, true)
+    }
+
+    fn with_seed(timeout: Duration, model: Option<MockClient>, pending_head: bool) -> Self {
         let root = restore_failure_tempdir();
         let repo = root.path().join("g");
         fs::create_dir(&repo).unwrap();
@@ -141,22 +291,31 @@ impl ProfileFailureFixture {
                     opens: provider_opens.clone(),
                 }),
         );
+        let execution: Arc<dyn WorkerExecutionBackend> = if pending_head {
+            Arc::new(PendingProfileSeedBackend {
+                inner: backend.clone(),
+            })
+        } else {
+            backend.clone()
+        };
         let runtime = EmbeddedRuntime::with_fs_store_and_execution_backend(
             crate::fs_store::FsRuntimeStoreOptions {
                 root: store,
                 runtime_id: "r".into(),
                 display_name: None,
             },
-            backend.clone(),
+            execution,
         )
         .unwrap();
         runtime.store_config_bundle(test_bundle()).unwrap();
         let mut request = create_request("restore failure recovery");
         request.profile = ProfileSelector::Builtin("default".into());
+        if !pending_head {
             request.workdir_attachment_requests = vec![WorkingDirectoryAttachmentRequest {
                 alias: WorkdirAttachmentAlias::new("checkout").unwrap(),
                 working_directory: working_directory_request(&repo),
             }];
+        }
         let worker = runtime.create_worker(request).unwrap();
         let seed_handle = queue_inputs.then(|| {
             backend.workers.lock().unwrap()[&worker.worker_ref]
@@ -203,6 +362,127 @@ impl ProfileFailureFixture {
         *self.backend.factory.restore_faults.fail_at.lock().unwrap() = Some(stage);
     }
 
+    fn pin_unclaimed_restore_upload(&self) -> (PathBuf, BTreeMap<PathBuf, Vec<u8>>) {
+        let aggregate = self
+            .backend
+            .factory
+            .worker_aggregate_dir(&self.worker.worker_ref)
+            .unwrap();
+        let name =
+            ProfileRuntimeWorkerFactory::runtime_worker_name_for_ref(&self.worker.worker_ref);
+        let metadata = WorkerAggregateStore::open_read_only(&aggregate, name.clone()).unwrap();
+        let session_id = metadata
+            .read_by_name(&name)
+            .unwrap()
+            .unwrap()
+            .active
+            .unwrap()
+            .session_id;
+        let session = WorkerSessionStore::new(aggregate.join("session")).unwrap();
+        let orphan = session
+            .write_uploaded_file(
+                session_id,
+                "orphan.txt",
+                "text/plain",
+                b"restore effect evidence",
+                session_store::UploadedFileLimits::default(),
+            )
+            .unwrap();
+        session
+            .pin_uploaded_file(session_id, &orphan, "uncommitted-restore-input")
+            .unwrap();
+        let artifacts = aggregate.join("session/artifacts/paste");
+        let pinned = persisted_files(&artifacts);
+        assert!(!pinned.is_empty());
+        (artifacts, pinned)
+    }
+
+    fn assert_retained_restore_effects(
+        &self,
+        request: &runtime_api::WorkerRestoreRequest,
+        pending: &Arc<PendingRuntimeRestore>,
+        stage: RuntimeRestoreStage,
+    ) {
+        let retained = self
+            .backend
+            .pending_restore(&self.worker.worker_ref)
+            .unwrap()
+            .expect("resource cleanup must not remove the operation's effect owner");
+        assert!(Arc::ptr_eq(pending, &retained));
+        {
+            let result = retained.result.lock().unwrap();
+            let failure = result.as_ref().unwrap().as_ref().err().unwrap();
+            assert_eq!(failure.stage, stage);
+            assert_eq!(failure.cleanup, RuntimeRestoreCleanup::Confirmed);
+            assert!(
+                failure.durable_effects_possible,
+                "resource cleanup cannot erase prior canonical effects: {failure:?}"
+            );
+        }
+        let record: serde_json::Value = serde_json::from_slice(
+            &fs::read(
+                self.root
+                    .path()
+                    .join("s/workers")
+                    .join(self.worker.worker_ref.worker_id.to_string())
+                    .join("worker.json"),
+            )
+            .unwrap(),
+        )
+        .unwrap();
+        assert_eq!(
+            record["restore_guard"]["active_request_id"],
+            request.request_id
+        );
+        let operation = serde_json::to_value(pending.request.operation_id).unwrap();
+        let owner = &record["restore_guard"]["owners"][&request.request_id];
+        assert_eq!(owner["operation_id"], operation);
+        assert!(
+            owner["receipt"].is_null(),
+            "false terminal receipt: {owner:?}"
+        );
+        assert_eq!(
+            record["execution_state"]["state"],
+            "reconciliation_required"
+        );
+        assert_eq!(
+            record["execution_state"]["execution"]["operation"],
+            "restore"
+        );
+        assert_eq!(
+            record["execution_state"]["execution"]["intent"]["operation_id"],
+            operation
+        );
+        assert!(
+            self.backend
+                .completed_restore_failures
+                .lock()
+                .unwrap()
+                .is_empty()
+        );
+        assert!(self.backend.workers.lock().unwrap().is_empty());
+        assert!(
+            self.backend
+                .factory
+                .failed_restore_resources
+                .lock()
+                .unwrap()
+                .is_empty(),
+            "cleanup confirmed resources, not rollback of canonical effects"
+        );
+        assert!(
+            self.backend
+                .factory
+                .observation_hub
+                .get(&self.worker.worker_ref)
+                .is_none()
+        );
+        if !pending.request.deferred_workdir_sessions.is_empty() {
+            self.assert_closed_deferred_sessions(pending);
+        }
+        assert_eq!(self.provider_opens.load(Ordering::SeqCst), 0);
+    }
+
     fn assert_released(&self) {
         assert_eq!(self.provider_opens.load(Ordering::SeqCst), 0);
         assert!(self.backend.pending_restores.lock().unwrap().is_empty());
@@ -422,6 +702,311 @@ impl ProfileFailureFixture {
     }
 }
 
+#[test]
+#[serial_test::serial(worker_allocation)]
+fn profile_restore_prior_effects_survive_read_only_reprepare_failure() {
+    for (activation_stage, read_stage) in [
+        (RuntimeRestoreStage::Feature, RuntimeRestoreStage::Metadata),
+        (
+            RuntimeRestoreStage::Controller,
+            RuntimeRestoreStage::Reconstruction,
+        ),
+    ] {
+        let fixture = ProfileFailureFixture::new(Duration::from_secs(5));
+        let (artifacts, pinned) = fixture.pin_unclaimed_restore_upload();
+        fixture.fail(activation_stage);
+        let request = fixture
+            .runtime
+            .test_restore_request(&fixture.worker.worker_ref);
+        let first = fixture
+            .runtime
+            .restore_worker_operation(&fixture.worker.worker_ref, request.clone())
+            .unwrap();
+        assert_eq!(
+            first.state,
+            server_api::WorkerRestoreState::ReconciliationRequired,
+            "{activation_stage:?}: {first:?}"
+        );
+        let pending = fixture
+            .backend
+            .pending_restore(&fixture.worker.worker_ref)
+            .unwrap()
+            .unwrap();
+        assert!(pending.activation_started.load(Ordering::SeqCst));
+        fixture.assert_retained_restore_effects(&request, &pending, activation_stage);
+        let effected = persisted_files(&artifacts);
+        assert_ne!(
+            effected, pinned,
+            "activation must actually reconcile the orphan pin"
+        );
+
+        // The next attempt needs no repair, and fails before any new mutation.
+        // Resource cleanup is confirmed, but the first attempt's pin change
+        // was not rolled back. Repeated sends must retain the same owner.
+        for _ in 0..2 {
+            fixture.fail(read_stage);
+            let replay = fixture
+                .runtime
+                .restore_worker_operation(&fixture.worker.worker_ref, request.clone())
+                .unwrap();
+            assert_eq!(
+                replay.state,
+                server_api::WorkerRestoreState::ReconciliationRequired,
+                "{activation_stage:?} then {read_stage:?}: {replay:?}"
+            );
+            fixture.assert_retained_restore_effects(&request, &pending, read_stage);
+            assert!(pending.activation_started.load(Ordering::SeqCst));
+            assert_eq!(persisted_files(&artifacts), effected);
+        }
+        fixture.resume_same_operation(request, pending.request.operation_id);
+    }
+}
+
+#[test]
+#[serial_test::serial(worker_allocation)]
+fn profile_restore_prior_effects_from_pending_head_repair_survive_read_only_reprepare_failure() {
+    let fixture = ProfileFailureFixture::with_pending_head(Duration::from_secs(5));
+    let aggregate = fixture
+        .backend
+        .factory
+        .worker_aggregate_dir(&fixture.worker.worker_ref)
+        .unwrap();
+    let name = ProfileRuntimeWorkerFactory::runtime_worker_name_for_ref(&fixture.worker.worker_ref);
+    let metadata = WorkerAggregateStore::new(&aggregate, &name).unwrap();
+    let initial = metadata.read_by_name(&name).unwrap().unwrap();
+    let pending_session = initial.active.as_ref().unwrap().session_id;
+    assert!(initial.active.as_ref().unwrap().segment_id.is_none());
+    {
+        let before = WorkerSessionStore::new(aggregate.join("session")).unwrap();
+        assert!(before.list_sessions().unwrap().is_empty());
+        assert!(before.list_segments(pending_session).unwrap().is_empty());
+    }
+    // The independent new Worker has never published a head. Its first segment
+    // and active pointer must be created by production repair, not fixture
+    // rewinding/deletion of a materialized Session.
+    fixture.fail(RuntimeRestoreStage::Metadata);
+    let request = fixture
+        .runtime
+        .test_restore_request(&fixture.worker.worker_ref);
+    let first = fixture
+        .runtime
+        .restore_worker_operation(&fixture.worker.worker_ref, request.clone())
+        .unwrap();
+    assert_eq!(
+        first.state,
+        server_api::WorkerRestoreState::ReconciliationRequired,
+        "{first:?}"
+    );
+    let pending = fixture
+        .backend
+        .pending_restore(&fixture.worker.worker_ref)
+        .unwrap()
+        .unwrap();
+    assert!(!pending.activation_started.load(Ordering::SeqCst));
+    assert!(pending.request.deferred_workdir_sessions.is_empty());
+    fixture.assert_retained_restore_effects(&request, &pending, RuntimeRestoreStage::Metadata);
+    let failure = pending
+        .result
+        .lock()
+        .unwrap()
+        .as_ref()
+        .unwrap()
+        .as_ref()
+        .err()
+        .unwrap()
+        .clone();
+    let repaired = WorkerAggregateStore::open_read_only(&aggregate, &name)
+        .unwrap()
+        .read_by_name(&name)
+        .unwrap()
+        .unwrap();
+    eprintln!(
+        "pending-head repair first={first:?} failure={failure:?} active={:?}",
+        repaired.active
+    );
+    let active = repaired
+        .active
+        .as_ref()
+        .expect("real repair must retain the pending Session's active pointer");
+    assert_eq!(active.session_id, pending_session);
+    // Store handles opened before the first Manifest retain session_id=None;
+    // they cannot observe that another owner has since published this Session.
+    // Reopen read-only to require the canonical Manifest as well as its logs.
+    let session = WorkerSessionStore::open_read_only(aggregate.join("session"))
+        .unwrap_or_else(|error| {
+            panic!(
+                "repair Manifest reread failed: {error}; first={first:?}; failure={failure:?}; active={active:?}"
+            )
+        });
+    assert_eq!(session.list_sessions().unwrap(), vec![pending_session]);
+    let segments = session.list_segments(pending_session).unwrap();
+    assert_eq!(
+        segments.len(),
+        1,
+        "real repair must commit exactly one seed; failure={failure:?}; active={active:?}"
+    );
+    assert_eq!(active.segment_id, Some(segments[0]));
+    let entries = session
+        .read_all_read_only(pending_session, segments[0])
+        .unwrap();
+    assert!(matches!(
+        entries.first(),
+        Some(LogEntry::AnnotatedSegmentStart { .. })
+    ));
+    assert!(
+        session_store::collect_state(&entries)
+            .system_prompt
+            .is_some()
+    );
+    assert_ne!(
+        repaired, initial,
+        "repair must publish the first active head"
+    );
+    let repaired_session = persisted_files(&aggregate.join("session"));
+
+    for _ in 0..2 {
+        fixture.fail(RuntimeRestoreStage::Reconstruction);
+        let replay = fixture
+            .runtime
+            .restore_worker_operation(&fixture.worker.worker_ref, request.clone())
+            .unwrap();
+        assert_eq!(
+            replay.state,
+            server_api::WorkerRestoreState::ReconciliationRequired,
+            "repair effect cannot become a rollback on the next read failure: {replay:?}"
+        );
+        fixture.assert_retained_restore_effects(
+            &request,
+            &pending,
+            RuntimeRestoreStage::Reconstruction,
+        );
+        let current_metadata = WorkerAggregateStore::open_read_only(&aggregate, &name).unwrap();
+        assert_eq!(
+            current_metadata.read_by_name(&name).unwrap().unwrap(),
+            repaired
+        );
+        let current_session =
+            WorkerSessionStore::open_read_only(aggregate.join("session")).unwrap();
+        assert_eq!(
+            current_session.list_sessions().unwrap(),
+            vec![pending_session]
+        );
+        assert_eq!(
+            current_session.list_segments(pending_session).unwrap(),
+            segments
+        );
+        assert_eq!(
+            persisted_files(&aggregate.join("session")),
+            repaired_session
+        );
+    }
+    fixture.resume_same_operation(request, pending.request.operation_id);
+}
+
+#[test]
+#[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 (artifacts, pinned) = fixture.pin_unclaimed_restore_upload();
+    fixture.fail(RuntimeRestoreStage::Feature);
+    let request = fixture
+        .runtime
+        .test_restore_request(&fixture.worker.worker_ref);
+    let first = fixture
+        .runtime
+        .restore_worker_operation(&fixture.worker.worker_ref, request.clone())
+        .unwrap();
+    assert_eq!(
+        first.state,
+        server_api::WorkerRestoreState::ReconciliationRequired,
+        "{first:?}"
+    );
+    let pending = fixture
+        .backend
+        .pending_restore(&fixture.worker.worker_ref)
+        .unwrap()
+        .unwrap();
+    fixture.assert_retained_restore_effects(&request, &pending, RuntimeRestoreStage::Feature);
+    let effected = persisted_files(&artifacts);
+    assert_ne!(effected, pinned);
+
+    let release = Arc::new(tokio::sync::Notify::new());
+    let (entered_tx, entered) = 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_tx);
+    let runtime = fixture.runtime.clone();
+    let worker_ref = fixture.worker.worker_ref.clone();
+    let resend = request.clone();
+    let retry = std::thread::spawn(move || runtime.restore_worker_operation(&worker_ref, resend));
+    let reached = entered.recv_timeout(Duration::from_secs(5));
+    // This gate is after real read-only reconstruction. Change only saved
+    // metadata, so the rebuilt candidate's first activation revalidation fails
+    // before allocation/pin/metadata/delegation mutation. No factory or repair
+    // implementation is replaced by a fake.
+    let aggregate = fixture
+        .backend
+        .factory
+        .worker_aggregate_dir(&fixture.worker.worker_ref)
+        .unwrap();
+    let name = ProfileRuntimeWorkerFactory::runtime_worker_name_for_ref(&fixture.worker.worker_ref);
+    let metadata = WorkerAggregateStore::new(&aggregate, &name).unwrap();
+    let mut changed = metadata.read_by_name(&name).unwrap().unwrap();
+    changed.workspace_root = Some(fixture.root.path().join("changed-after-reconstruction"));
+    metadata.write(&changed).unwrap();
+    release.notify_one();
+    let rebuilt = retry.join().unwrap().unwrap();
+    reached.expect("same-operation retry must reach the real reconstructed candidate");
+    assert_eq!(
+        rebuilt.state,
+        server_api::WorkerRestoreState::ReconciliationRequired,
+        "{rebuilt:?}"
+    );
+    assert_eq!(
+        fixture
+            .backend
+            .factory
+            .restore_faults
+            .executions
+            .load(Ordering::SeqCst),
+        2
+    );
+    assert!(pending.activation_started.load(Ordering::SeqCst));
+    fixture.assert_retained_restore_effects(
+        &request,
+        &pending,
+        RuntimeRestoreStage::Reconstruction,
+    );
+    assert_eq!(metadata.read_by_name(&name).unwrap().unwrap(), changed);
+    assert_eq!(persisted_files(&artifacts), effected);
+
+    fixture.fail(RuntimeRestoreStage::Metadata);
+    let replay = fixture
+        .runtime
+        .restore_worker_operation(&fixture.worker.worker_ref, request.clone())
+        .unwrap();
+    assert_eq!(
+        replay.state,
+        server_api::WorkerRestoreState::ReconciliationRequired,
+        "successful reconstruction must not erase the older effect: {replay:?}"
+    );
+    fixture.assert_retained_restore_effects(&request, &pending, RuntimeRestoreStage::Metadata);
+    assert_eq!(metadata.read_by_name(&name).unwrap().unwrap(), changed);
+    assert_eq!(persisted_files(&artifacts), effected);
+    fixture.resume_same_operation(request, pending.request.operation_id);
+}
+
 #[test]
 #[serial_test::serial(worker_allocation)]
 fn profile_restore_timeout_attachment_revocation_cannot_reactivate_old_candidate() {