fix(runtime): consume original Stop receipts for worker deletion

crates/worker-runtime/src/runtime.rs

@@ -3266,6 +3266,14 @@ impl Runtime {
         let _guard = operation_lock
             .lock()
             .map_err(|_| RuntimeError::StatePoisoned)?;
+        self.execution_termination_under_lock(worker_ref)
+    }
+
+    // Caller holds the Runtime Worker operation lock, including deletion callers.
+    fn execution_termination_under_lock(
+        &self,
+        worker_ref: &WorkerRef,
+    ) -> Result<Option<crate::execution::WorkerExecutionTermination>, RuntimeError> {
         let (backend, saved) = {
             let mut state = self.lock()?;
             state.ensure_worker_ref(worker_ref)?;
@@ -4630,7 +4638,7 @@ impl Runtime {
         if let Some(scope) = scope {
             self.ensure_worker_in_workspace(scope, worker_ref)?;
         }
-        let backend = {
+        let (backend, has_termination) = {
             let state = self.lock()?;
             state.ensure_running()?;
             state.ensure_worker_ref(worker_ref)?;
@@ -4648,9 +4656,22 @@ impl Runtime {
                     worker_ref.worker_id
                 )));
             }
-            state.execution_backend.clone()
+            (
+                state.execution_backend.clone(),
+                worker.restore_guard.execution_termination.is_some(),
+            )
         };
-        if let Some(backend) = backend {
+        if has_termination {
+            // Consume the canonical Stop's exact receipt. A new Stop could
+            // target a replacement Controller or discard a later candidate;
+            // neither is authorized by the original scope's termination.
+            if self.execution_termination_under_lock(worker_ref)?.is_none() {
+                return Err(RuntimeError::WorkerExecutionUnavailable {
+                    worker_id: worker_ref.worker_id,
+                    message: "canonical Stop receipt no longer matches the execution owner; deletion remains pending".into(),
+                });
+            }
+        } else if let Some(backend) = backend {
             let result = backend.stop_worker(worker_ref);
             if !result.is_accepted() {
                 return Err(RuntimeError::WorkerExecutionRejected {

crates/worker-runtime/src/subject_delete_backend_tests.rs

@@ -196,8 +196,21 @@ fn subject_delete_unregistered_profile_controller_retries_unknown_shutdown_witho
         backend.restore_worker(request.clone()),
         WorkerExecutionSpawnResult::Rejected(_)
     ));
+    let stop_operation_id = backend.admissions.lock().unwrap()[&request.worker_ref]
+        .deletion_stop_operation_id
+        .expect("the original rejected Stop keeps its identity for retry");
     backend.subject_delete_worker(&deletion).unwrap();
+    let original = backend
+        .termination_receipt(&request.worker_ref)
+        .unwrap()
+        .unwrap();
+    assert_eq!(original.stop_operation_id, stop_operation_id);
+    assert_eq!(original.execution_id, execution.tasks.execution_id);
     backend.subject_delete_worker(&deletion).unwrap();
+    assert_eq!(
+        backend.termination_receipt(&request.worker_ref).unwrap(),
+        Some(original)
+    );
     assert!(execution.handle.protocol_is_closed());
     assert!(execution.tasks.tasks.lock().unwrap().is_empty());
     assert!(backend.workers.lock().unwrap().is_empty());
@@ -283,3 +296,111 @@ fn subject_delete_pending_restore_fences_late_controller_and_never_restarts_it()
         WorkerExecutionSpawnResult::Rejected(_)
     ));
 }
+
+#[test]
+fn subject_delete_without_original_owner_retains_child_identity_but_never_issues_receipt() {
+    let root = tempfile::tempdir().unwrap();
+    let backend = WorkerRuntimeExecutionBackend::new(MockFactory {
+        client: MockClient::new(vec![]),
+        runtime_base: root.path().join("must-not-launch"),
+        cwd: root.path().to_path_buf(),
+        store_dir: root.path().join("sessions"),
+        worker_metadata_dir: root.path().join("metadata"),
+        observed_cwds: Arc::default(),
+        observed_workspace_clients: Arc::default(),
+    })
+    .unwrap();
+    let worker_ref = WorkerRef::new(WorkerId::now_v7());
+    let deletion = runtime_api::SubjectDeleteWorkerRequest {
+        workspace_id: "ws-test".into(),
+        worker_id: worker_ref.worker_id,
+        create_fingerprint: "not-admitted".into(),
+        operation_id: "delete-without-owner".into(),
+    };
+    assert!(
+        backend
+            .subject_delete_worker(&deletion)
+            .unwrap_err()
+            .contains("map absence")
+    );
+    let child = backend.admissions.lock().unwrap()[&worker_ref]
+        .deletion_stop_operation_id
+        .unwrap();
+    assert!(
+        backend
+            .subject_delete_worker(&deletion)
+            .unwrap_err()
+            .contains("map absence")
+    );
+    assert_eq!(
+        backend.admissions.lock().unwrap()[&worker_ref].deletion_stop_operation_id,
+        Some(child)
+    );
+    assert!(backend.completed_stops.lock().unwrap().is_empty());
+    assert!(backend.termination_receipt(&worker_ref).unwrap().is_none());
+    assert!(!root.path().join("must-not-launch").exists());
+}
+
+#[test]
+#[cfg(feature = "fs-store")]
+fn subject_delete_consumes_prior_canonical_stop_without_retagging_original_scope() {
+    let temp = tempfile::tempdir_in(".").unwrap();
+    let root = fs::canonicalize(temp.path()).unwrap();
+    let backend = Arc::new(
+        WorkerRuntimeExecutionBackend::new(MockFactory {
+            client: MockClient::new(vec![]),
+            runtime_base: PathBuf::from(temp.path().file_name().unwrap()).join("c"),
+            cwd: root.clone(),
+            store_dir: root.join("sessions"),
+            worker_metadata_dir: root.join("metadata"),
+            observed_cwds: Arc::default(),
+            observed_workspace_clients: Arc::default(),
+        })
+        .unwrap(),
+    );
+    let runtime = EmbeddedRuntime::with_fs_store_and_execution_backend(
+        crate::fs_store::FsRuntimeStoreOptions::new(root.join("catalog"))
+            .with_runtime_id("delete-prior-stop"),
+        backend.clone(),
+    )
+    .unwrap();
+    runtime.store_config_bundle(test_bundle()).unwrap();
+    let scope = crate::runtime::RuntimeWorkspaceScope::new("ws-test", "server-test");
+    let mut create = create_request("prior Stop Subject deletion");
+    create.memory_settings = Some(manifest::WorkspaceMemorySettingsSnapshot {
+        workspace_id: scope.workspace_id.clone(),
+        language: "English".into(),
+    });
+    create.workspace_api = Some(WorkspaceApiRef {
+        workspace_id: scope.workspace_id.clone(),
+        base_url: "https://workspace.invalid".into(),
+    });
+    let worker = runtime
+        .create_worker_scoped(&scope, create.clone())
+        .unwrap();
+    runtime
+        .stop_worker_scoped(&scope, &worker.worker_ref, None)
+        .unwrap();
+    let original = runtime
+        .execution_termination(&worker.worker_ref)
+        .unwrap()
+        .unwrap();
+    let deletion = runtime_api::SubjectDeleteWorkerRequest {
+        workspace_id: scope.workspace_id.clone(),
+        worker_id: create.worker_id,
+        create_fingerprint: create.create_fingerprint,
+        operation_id: "delete-prior-stop".into(),
+    };
+    runtime
+        .subject_delete_worker_scoped(&scope, deletion.clone())
+        .unwrap();
+    backend.subject_delete_worker(&deletion).unwrap();
+    assert_eq!(
+        backend.termination_receipt(&worker.worker_ref).unwrap(),
+        Some(original.clone())
+    );
+    assert_eq!(
+        backend.admissions.lock().unwrap()[&worker.worker_ref].deletion_stop_operation_id,
+        Some(original.stop_operation_id)
+    );
+}

crates/worker-runtime/src/worker_backend.rs

@@ -3903,6 +3903,9 @@ struct WorkerAdmission {
     workspace_id: Option<String>,
     create_fingerprint: String,
     deletion: Option<runtime_api::SubjectDeleteWorkerRequest>,
+    // Stable child Stop for this exact deletion. This is an operation identity,
+    // not termination proof; only the original joined owner can issue a receipt.
+    deletion_stop_operation_id: Option<crate::execution::WorkerLifecycleOperationId>,
 }
 
 /// Worker execution backend backed by real worker crate Workers.
@@ -4007,6 +4010,7 @@ where
                     workspace_id: workspace.map(str::to_string),
                     create_fingerprint: fingerprint.into(),
                     deletion: None,
+                    deletion_stop_operation_id: None,
                 },
             );
         }
@@ -5825,18 +5829,42 @@ where
         let lock = self.worker_lock(&worker)?;
         let _operation = lock.lock().map_err(|_| "worker operation lock poisoned")?;
         self.validate_subject_delete_identity(request)?;
-        self.admissions
-            .lock()
-            .map_err(|_| "worker admission lock poisoned")?
-            .insert(
-                worker.clone(),
-                WorkerAdmission {
+        let operation_id = {
+            let mut admissions = self
+                .admissions
+                .lock()
+                .map_err(|_| "worker admission lock poisoned")?;
+            let admission = admissions
+                .entry(worker.clone())
+                .or_insert_with(|| WorkerAdmission {
                     workspace_id: Some(request.workspace_id.clone()),
                     create_fingerprint: request.create_fingerprint.clone(),
-                    deletion: Some(request.clone()),
-                },
-            );
-        let result = self.stop_worker_under_lock(&worker, None);
+                    deletion: None,
+                    deletion_stop_operation_id: None,
+                });
+            admission.deletion = Some(request.clone());
+            match admission.deletion_stop_operation_id {
+                Some(operation_id) => operation_id,
+                None => {
+                    // Reuse a completed original Stop, never retag its receipt
+                    // as a newly generated deletion child. The Stop consumer
+                    // below rejects replay if any replacement owner is present.
+                    let operation_id = self
+                        .completed_stops
+                        .lock()
+                        .map_err(|_| "stop completion receipts lock poisoned")?
+                        .get(&worker)
+                        .map(|receipt| receipt.stop_operation_id)
+                        .unwrap_or_default();
+                    admission.deletion_stop_operation_id = Some(operation_id);
+                    operation_id
+                }
+            }
+        };
+        let result = self.stop_worker_operation_under_lock(WorkerExecutionStopRequest {
+            worker_ref: worker.clone(),
+            operation_id,
+        });
         if !result.is_accepted() {
             return Err(result.message_or_default());
         }
@@ -8737,12 +8765,17 @@ mod tests {
             })
             .await
             .unwrap();
-        if let Some(receiver) = controller.shutdown.lock().await.take() {
-            tokio::time::timeout(Duration::from_secs(5), receiver)
-                .await
-                .unwrap()
-                .unwrap();
-        }
+        let receiver = controller
+            .shutdown
+            .lock()
+            .await
+            .take()
+            .expect("fixture must retain the original Controller shutdown report");
+        let report = tokio::time::timeout(Duration::from_secs(5), receiver)
+            .await
+            .unwrap()
+            .unwrap();
+        assert!(report.feature_failures.is_empty(), "{report:?}");
         tokio::time::timeout(Duration::from_secs(5), controller.controller_task)
             .await
             .unwrap()
@@ -9795,11 +9828,22 @@ mod tests {
                 .contains("still present"),
             "closed Controller is still residual execution"
         );
-        assert!(backend.stop_worker(&request.worker_ref).is_accepted());
+        let stop = WorkerExecutionStopRequest {
+            worker_ref: request.worker_ref.clone(),
+            operation_id: crate::execution::WorkerLifecycleOperationId::new(),
+        };
+        let original = backend.stop_worker_operation(stop.clone());
+        assert!(original.is_accepted(), "{original:?}");
+        let receipt = original.termination.as_ref().unwrap();
+        assert_eq!(receipt.execution_id, execution.tasks.execution_id);
         assert!(backend.workers.lock().unwrap().is_empty());
         backend.recover_subject_create(&recovery).unwrap();
         assert!(execution.tasks.tasks.lock().unwrap().is_empty());
-        assert!(backend.stop_worker(&request.worker_ref).is_accepted());
+        assert_eq!(backend.stop_worker_operation(stop), original);
+        assert!(
+            !backend.stop_worker(&request.worker_ref).is_accepted(),
+            "identity-less absence is not a replay"
+        );
     }
 
     #[test]
@@ -11263,6 +11307,8 @@ mod tests {
     struct PendingCreateFactory {
         started: mpsc::Sender<()>,
         dropped: mpsc::Sender<()>,
+        release: Arc<tokio::sync::Notify>,
+        inner: MockFactory,
     }
 
     struct FactoryFutureDrop(mpsc::Sender<()>);
@@ -11286,11 +11332,12 @@ mod tests {
 
         async fn spawn_controller(
             &self,
-            _request: WorkerExecutionSpawnRequest,
+            request: WorkerExecutionSpawnRequest,
         ) -> Result<RuntimeWorkerController, RuntimeRestoreFailure> {
             let _drop = FactoryFutureDrop(self.dropped.clone());
             self.started.send(()).unwrap();
-            std::future::pending().await
+            self.release.notified().await;
+            self.inner.spawn_controller(request).await
         }
 
         async fn restore_controller(
@@ -11303,15 +11350,26 @@ mod tests {
 
     #[cfg(feature = "fs-store")]
     #[test]
-    fn create_timeout_cancels_factory_and_removes_persisted_worker() {
+    fn create_timeout_retains_factory_and_persisted_worker_until_original_controller_is_joined() {
         let root = tempfile::tempdir().unwrap();
         let runtime_store_dir = root.path().join("runtime");
         let (started_tx, started) = mpsc::channel();
         let (dropped_tx, dropped) = mpsc::channel();
+        let release = Arc::new(tokio::sync::Notify::new());
         let backend = Arc::new(
             WorkerRuntimeExecutionBackend::new(PendingCreateFactory {
                 started: started_tx,
                 dropped: dropped_tx,
+                release: release.clone(),
+                inner: MockFactory {
+                    client: MockClient::new(vec![]),
+                    runtime_base: root.path().join("control"),
+                    cwd: root.path().to_path_buf(),
+                    store_dir: root.path().join("sessions"),
+                    worker_metadata_dir: root.path().join("metadata"),
+                    observed_cwds: Arc::default(),
+                    observed_workspace_clients: Arc::default(),
+                },
             })
             .unwrap()
             .with_spawn_restore_timeout(Duration::from_millis(20)),
@@ -11339,23 +11397,46 @@ mod tests {
             .unwrap_err();
         let error = create.join().unwrap().unwrap_err();
 
+        let worker_ref = WorkerRef::new(worker_id);
         assert!(matches!(
             delete_error,
-            crate::error::RuntimeError::WorkerNotFound { worker_id: missing } if missing == worker_id
+            crate::error::RuntimeError::InvalidRequest(_)
         ));
-        assert!(error.to_string().contains("was cancelled"));
+        assert!(error.to_string().contains("pending"), "{error}");
+        assert!(matches!(dropped.try_recv(), Err(mpsc::TryRecvError::Empty)));
+        assert!(backend.pending_spawn_exists(&worker_ref).unwrap());
+        assert!(
+            runtime
+                .execution_termination(&worker_ref)
+                .unwrap()
+                .is_none()
+        );
+        let worker_path = runtime_store_dir
+            .join("workers")
+            .join(worker_id.to_string());
+        assert!(
+            worker_path.exists(),
+            "timeout must retain the unresolved Worker"
+        );
+        assert_eq!(runtime.list_workers().unwrap().len(), 1);
+
+        // Release the original factory, not a replacement. Stop must transfer
+        // its actual late Controller into the owned scope and join/report it.
+        release.notify_one();
+        runtime.stop_worker(&worker_ref, None).unwrap();
         dropped
             .try_recv()
-            .expect("cancelled factory future must be dropped before create returns");
-        assert!(runtime.list_workers().unwrap().is_empty());
-        assert!(backend.workers.lock().unwrap().is_empty());
-        assert!(
-            !runtime_store_dir
-                .join("workers")
-                .join(worker_id.to_string())
-                .exists(),
-            "failed create must remove its persisted Worker aggregate"
+            .expect("factory completed normally before Stop settled");
+        assert!(!backend.pending_spawn_exists(&worker_ref).unwrap());
+        let original = runtime.execution_termination(&worker_ref).unwrap().unwrap();
+        assert_eq!(
+            original.scope,
+            crate::execution::WorkerExecutionTerminationScope::Controller
         );
+        assert!(original.shutdown_command_id.is_some());
+        assert!(backend.workers.lock().unwrap().is_empty());
+        runtime.delete_worker(&worker_ref).unwrap();
+        assert!(!worker_path.exists());
     }
 
     // Generic controller tests explicitly substitute the Host's non-Subject
@@ -11922,7 +12003,13 @@ mod tests {
             &self,
             _request: WorkerExecutionSpawnRequest,
         ) -> Result<RuntimeWorkerController, RuntimeRestoreFailure> {
-            Err("spawn failed".into())
+            // This fixture returns before constructing a Worker, Controller or
+            // background resource. Unlike a string failure, these are known
+            // owner facts and allow only the adapter-created workdir rollback.
+            Err(RuntimeRestoreFailure::refused(
+                RuntimeRestoreStage::Preflight,
+                "spawn failed",
+            ))
         }
 
         async fn restore_controller(
@@ -14104,7 +14191,19 @@ mod tests {
             fs::read_to_string(checkout.join("README.md")).unwrap(),
             "owner recovery in progress"
         );
-        assert!(backend.stop_worker(&request.worker_ref).is_accepted());
+        let stop = WorkerExecutionStopRequest {
+            worker_ref: request.worker_ref.clone(),
+            operation_id: crate::execution::WorkerLifecycleOperationId::new(),
+        };
+        let result = backend.stop_worker_operation(stop.clone());
+        assert!(result.is_accepted(), "{result:?}");
+        assert!(
+            result
+                .termination
+                .as_ref()
+                .unwrap()
+                .confirms_scope_shutdown(&stop.worker_ref, stop.operation_id)
+        );
     }
 
     #[cfg(target_os = "linux")]
@@ -14176,7 +14275,19 @@ mod tests {
                 before,
                 "damaged checkout must be retained"
             );
-            assert!(backend.stop_worker(&request.worker_ref).is_accepted());
+            let stop = WorkerExecutionStopRequest {
+                worker_ref: request.worker_ref.clone(),
+                operation_id: crate::execution::WorkerLifecycleOperationId::new(),
+            };
+            let result = backend.stop_worker_operation(stop.clone());
+            assert!(result.is_accepted(), "{result:?}");
+            assert!(
+                result
+                    .termination
+                    .as_ref()
+                    .unwrap()
+                    .confirms_scope_shutdown(&stop.worker_ref, stop.operation_id)
+            );
             println!(
                 "damage={damage}; reason={:?}; classification={:?}; cwd_bind_count={}; retained_files={}; retained_data_unchanged=true; stopped=true",
                 error.failure_reason(),

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

@@ -253,9 +253,15 @@ mod tests {
     #[test]
     fn typed_factory_failure_survives_adapter_and_protocol_without_changing_disposition() {
         let backend = WorkerRuntimeExecutionBackend::new(TypedFailingFactory).unwrap();
-        let WorkerExecutionSpawnResult::Errored(result) = backend.spawn_worker(request()) else {
-            panic!("cause attachment must not reclassify the existing spawn result")
+        let request = request();
+        let worker_ref = request.worker_ref.clone();
+        let WorkerExecutionSpawnResult::ReconciliationRequired { result, .. } =
+            backend.spawn_worker(request)
+        else {
+            panic!("unproven resources and failed cleanup must retain the original factory owner")
         };
+        assert!(backend.pending_spawn_exists(&worker_ref).unwrap());
+        assert!(backend.termination_receipt(&worker_ref).unwrap().is_none());
         assert_primary(&result);
         let transferred: WorkerExecutionResult =
             serde_json::from_slice(&serde_json::to_vec(&result).unwrap()).unwrap();

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

@@ -97,6 +97,15 @@ fn stop_termination_real_join_survives_fs_reopen_rejects_replacement_and_legacy_
             .unwrap()
             .is_none()
     );
+    assert!(reopened.delete_worker(&worker.worker_ref).is_err());
+    assert!(worker_path.exists());
+    assert!(
+        backend
+            .pending_restore(&worker.worker_ref)
+            .unwrap()
+            .is_some(),
+        "deletion must not stop/discard a replacement candidate with the old receipt"
+    );
     let replacement_stop = backend.stop_worker_operation(WorkerExecutionStopRequest {
         worker_ref: worker.worker_ref.clone(),
         operation_id: crate::execution::WorkerLifecycleOperationId::new(),
@@ -110,6 +119,8 @@ fn stop_termination_real_join_survives_fs_reopen_rejects_replacement_and_legacy_
         reopened.execution_termination(&worker.worker_ref).is_err(),
         "a different run/stop-operation receipt must not replace canonical evidence"
     );
+    assert!(reopened.delete_worker(&worker.worker_ref).is_err());
+    assert!(worker_path.exists());
     drop(reopened);
     drop(backend);
 
@@ -155,3 +166,97 @@ fn stop_termination_real_join_survives_fs_reopen_rejects_replacement_and_legacy_
             .is_none()
     );
 }
+
+#[cfg(feature = "fs-store")]
+#[test]
+fn delete_after_reopen_consumes_original_stop_without_starting_or_restopping_controller() {
+    let temp = tempfile::tempdir_in(".").unwrap();
+    let root = fs::canonicalize(temp.path()).unwrap();
+    let options = crate::fs_store::FsRuntimeStoreOptions::new(root.join("runtime"))
+        .with_runtime_id("delete-original-stop");
+    let backend = Arc::new(
+        WorkerRuntimeExecutionBackend::new(MockFactory {
+            client: MockClient::new(vec![]),
+            runtime_base: PathBuf::from(temp.path().file_name().unwrap()).join("c"),
+            cwd: root.clone(),
+            store_dir: root.join("sessions"),
+            worker_metadata_dir: root.join("metadata"),
+            observed_cwds: Arc::default(),
+            observed_workspace_clients: Arc::default(),
+        })
+        .unwrap(),
+    );
+    let runtime =
+        EmbeddedRuntime::with_fs_store_and_execution_backend(options.clone(), backend.clone())
+            .unwrap();
+    runtime.store_config_bundle(test_bundle()).unwrap();
+    let worker = runtime
+        .create_worker(create_request("delete original stop"))
+        .unwrap();
+    runtime.stop_worker(&worker.worker_ref, None).unwrap();
+    let original = runtime
+        .execution_termination(&worker.worker_ref)
+        .unwrap()
+        .unwrap();
+    // Inject mismatched observations, never positive termination evidence.
+    // Deletion must keep the canonical original and refuse every substitution.
+    for mismatch in ["worker", "execution", "operation"] {
+        let mut observed = original.clone();
+        match mismatch {
+            "worker" => observed.worker_ref = WorkerRef::new(WorkerId::now_v7()),
+            "execution" => observed.execution_id = uuid::Uuid::now_v7(),
+            "operation" => {
+                observed.stop_operation_id = crate::execution::WorkerLifecycleOperationId::new()
+            }
+            _ => unreachable!(),
+        }
+        backend
+            .completed_stops
+            .lock()
+            .unwrap()
+            .insert(worker.worker_ref.clone(), observed);
+        assert!(
+            runtime.delete_worker(&worker.worker_ref).is_err(),
+            "{mismatch}"
+        );
+        assert!(
+            options
+                .root
+                .join("workers")
+                .join(worker.worker_id.to_string())
+                .exists()
+        );
+    }
+    backend
+        .completed_stops
+        .lock()
+        .unwrap()
+        .insert(worker.worker_ref.clone(), original.clone());
+    drop(runtime);
+    drop(backend);
+    let no_launch = root.join("must-not-launch");
+    let backend = Arc::new(
+        WorkerRuntimeExecutionBackend::new(ProfileRuntimeWorkerFactory::new(&no_launch)).unwrap(),
+    );
+    let reopened =
+        EmbeddedRuntime::with_fs_store_and_execution_backend(options.clone(), backend.clone())
+            .unwrap();
+    assert!(backend.completed_stops.lock().unwrap().is_empty());
+    assert_eq!(
+        reopened.execution_termination(&worker.worker_ref).unwrap(),
+        Some(original)
+    );
+    reopened.delete_worker(&worker.worker_ref).unwrap();
+    assert!(
+        !options
+            .root
+            .join("workers")
+            .join(worker.worker_id.to_string())
+            .exists()
+    );
+    assert!(
+        backend.completed_stops.lock().unwrap().is_empty(),
+        "no new Stop was issued"
+    );
+    assert!(!no_launch.exists());
+}

crates/worker-runtime/src/worker_backend_restore_candidate_tests.rs

@@ -1246,11 +1246,18 @@ mod restore_candidate {
                 .is_some(),
             "retain the real prepared Worker's close-only resource owner"
         );
+        let stop = WorkerExecutionStopRequest {
+            worker_ref: fixture.request.worker_ref.clone(),
+            operation_id: crate::execution::WorkerLifecycleOperationId::new(),
+        };
+        let result = fixture.backend.stop_worker_operation(stop.clone());
+        assert!(result.is_accepted(), "{result:?}");
         assert!(
-            fixture
-                .backend
-                .stop_worker(&fixture.request.worker_ref)
-                .is_accepted()
+            result
+                .termination
+                .as_ref()
+                .unwrap()
+                .confirms_scope_shutdown(&stop.worker_ref, stop.operation_id)
         );
         fixture.assert_released();
 
@@ -1285,12 +1292,10 @@ mod restore_candidate {
         assert_eq!(opens.load(Ordering::SeqCst), 0);
         fixture.assert_quiet();
         assert_eq!(fixture.canonical_files(), before);
-        assert!(
-            fixture
-                .backend
-                .stop_worker(&fixture.request.worker_ref)
-                .is_accepted()
-        );
+        let replay = fixture.backend.stop_worker_operation(stop);
+        assert!(replay.is_accepted(), "{replay:?}");
+        assert_eq!(replay.termination, result.termination);
+        assert_eq!(replay.execution_id, result.execution_id);
         assert_eq!(fixture.canonical_files(), before);
     }