fix: retain original controller termination and exact submission evidence

crates/worker-runtime/src/execution.rs

@@ -1157,6 +1157,18 @@ pub trait WorkerExecutionBackend: Send + Sync + 'static {
         Err("execution backend has no scope-termination receipt reader".into())
     }
 
+    /// Reconcile canonical execution-owner evidence against current backend
+    /// ownership. The default cannot certify a receipt lost by this backend.
+    fn reconcile_termination_receipt(
+        &self,
+        worker_ref: &WorkerRef,
+        saved: &WorkerExecutionTermination,
+    ) -> Result<Option<WorkerExecutionTermination>, String> {
+        Ok(self
+            .termination_receipt(worker_ref)?
+            .filter(|current| current == saved))
+    }
+
     fn dispatch_input(&self, worker_ref: &WorkerRef, input: WorkerInput) -> WorkerExecutionResult;
 
     fn upload_file(
@@ -1458,11 +1470,13 @@ impl WorkerExecutionBackendRef {
         self.backend.reconcile_submission(worker_ref, input)
     }
 
-    pub(crate) fn termination_receipt(
+    pub(crate) fn reconcile_termination_receipt(
         &self,
         worker_ref: &WorkerRef,
+        saved: &WorkerExecutionTermination,
     ) -> Result<Option<WorkerExecutionTermination>, String> {
-        self.backend.termination_receipt(worker_ref)
+        self.backend
+            .reconcile_termination_receipt(worker_ref, saved)
     }
 
     pub(crate) fn dispatch_input(

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 = 15;
+pub(crate) const SCHEMA_VERSION: u32 = 16;
+const SESSION_ARCHIVE_SCHEMA_VERSION: u32 = 15;
 const RESTORE_REPAIR_SCHEMA_VERSION: u32 = 14;
 const RESTORE_DECISION_SCHEMA_VERSION: u32 = 13;
 const RESTORE_ADMISSION_SCHEMA_VERSION: u32 = 12;
@@ -912,11 +913,12 @@ fn plan_runtime_store_migration(
             | RESTORE_ADMISSION_SCHEMA_VERSION
             | RESTORE_DECISION_SCHEMA_VERSION
             | RESTORE_REPAIR_SCHEMA_VERSION
+            | SESSION_ARCHIVE_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}, {RESTORE_DECISION_SCHEMA_VERSION}, {RESTORE_REPAIR_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}, {RESTORE_REPAIR_SCHEMA_VERSION}, {SESSION_ARCHIVE_SCHEMA_VERSION}, or {SCHEMA_VERSION}"
             ),
         ));
     }
@@ -1253,6 +1255,7 @@ fn migrate_worker_record(
             | RESTORE_ADMISSION_SCHEMA_VERSION
             | RESTORE_DECISION_SCHEMA_VERSION
             | RESTORE_REPAIR_SCHEMA_VERSION
+            | SESSION_ARCHIVE_SCHEMA_VERSION
     ) {
         // Admission identity survives migration. Older pending records did not
         // distinguish a committed decision from preparation or failed cleanup.
@@ -1744,6 +1747,7 @@ fn migrate_runtime_document(
             | RESTORE_ADMISSION_SCHEMA_VERSION
             | RESTORE_DECISION_SCHEMA_VERSION
             | RESTORE_REPAIR_SCHEMA_VERSION
+            | SESSION_ARCHIVE_SCHEMA_VERSION
     ) {
         object.remove("workers");
         // This legacy cache is not Runtime lifecycle or restore authority. Bundles
@@ -2295,6 +2299,7 @@ fn migrate_runtime_store_in_place(
                     | RESTORE_ADMISSION_SCHEMA_VERSION
                     | RESTORE_DECISION_SCHEMA_VERSION
                     | RESTORE_REPAIR_SCHEMA_VERSION
+                    | SESSION_ARCHIVE_SCHEMA_VERSION
             )
         {
             atomic_write_json(

crates/worker-runtime/src/runtime.rs

@@ -2400,6 +2400,7 @@ impl Runtime {
                         last_settled_status: candidate.status,
                     });
                 let operation_id = pending.operation_id;
+                candidate.restore_guard.execution_termination = None;
                 candidate.pending_restore = Some(pending);
                 candidate.restore_guard.rotate();
                 candidate.restore_guard.active_request_id = Some(request.request_id.clone());
@@ -3265,30 +3266,52 @@ impl Runtime {
         let _guard = operation_lock
             .lock()
             .map_err(|_| RuntimeError::StatePoisoned)?;
-        let backend = {
+        let (backend, saved) = {
-            let state = self.lock()?;
+            let mut state = self.lock()?;
             state.ensure_worker_ref(worker_ref)?;
-            state.worker(worker_ref)?;
+            // Unknown save acknowledgements must resolve against the same
-            state.execution_backend.clone().ok_or_else(|| {
+            // canonical operation before exposing its termination evidence.
+            state.refresh_uncertain_worker_write(worker_ref)?;
+            let worker = state.worker(worker_ref)?;
+            let saved = worker
+                .restore_guard
+                .execution_termination
+                .clone()
+                .filter(|receipt| {
+                    worker.pending_stop.map_or_else(
+                        || {
+                            worker.status == WorkerStatus::Stopped
+                                && worker.pending_restore.is_none()
+                        },
+                        |pending| pending.operation_id == receipt.stop_operation_id,
+                    )
+                });
+            let backend = state.execution_backend.clone().ok_or_else(|| {
                 RuntimeError::WorkerExecutionUnavailable {
                     worker_id: worker_ref.worker_id,
                     message: "runtime has no execution backend".into(),
                 }
-            })?
+            })?;
+            (backend, saved)
         };
-        let receipt = backend.termination_receipt(worker_ref).map_err(|message| {
+        let Some(saved) = saved else {
-            RuntimeError::WorkerExecutionUnavailable {
+            // A backend map or stopped observation alone is not durable evidence.
+            return Ok(None);
+        };
+        let receipt = backend
+            .reconcile_termination_receipt(worker_ref, &saved)
+            .map_err(|message| RuntimeError::WorkerExecutionUnavailable {
                 worker_id: worker_ref.worker_id,
                 message,
-            }
+            })?;
-        })?;
+        if receipt.as_ref().is_some_and(|receipt| {
-        if receipt
+            !receipt.confirms_scope_shutdown(worker_ref, receipt.stop_operation_id)
-            .as_ref()
+                || &saved != receipt
-            .is_some_and(|receipt| &receipt.worker_ref != worker_ref)
+        }) {
-        {
             return Err(RuntimeError::WorkerExecutionUnavailable {
                 worker_id: worker_ref.worker_id,
-                message: "scope-termination receipt Worker mismatch".into(),
+                message: "scope-termination receipt does not match canonical operation evidence"
+                    .into(),
             });
         }
         Ok(receipt)
@@ -4354,15 +4377,6 @@ impl Runtime {
             operation_id,
             worker_ref: worker_ref.clone(),
         });
-        if !result.is_accepted() {
-            return Err(RuntimeError::WorkerExecutionRejected {
-                worker_id: worker_ref.worker_id,
-                operation: result.operation,
-                outcome: result.outcome,
-                message: result.message_or_default(),
-                result,
-            });
-        }
         self.commit_stopped_worker(worker_ref, operation_id, result)
     }
 
@@ -5102,15 +5116,6 @@ impl Runtime {
             operation_id,
             worker_ref: worker_ref.clone(),
         });
-        if !result.is_accepted() {
-            return Err(RuntimeError::WorkerExecutionRejected {
-                worker_id: worker_ref.worker_id,
-                operation: result.operation,
-                outcome: result.outcome,
-                message: result.message_or_default(),
-                result,
-            });
-        }
         self.commit_stopped_worker(worker_ref, operation_id, result)?;
         Ok(())
     }
@@ -5144,6 +5149,7 @@ impl Runtime {
             )));
         }
         candidate.restore_guard.rotate();
+        candidate.restore_guard.execution_termination = None;
         candidate.pending_stop = Some(PendingWorkerStop {
             operation_id,
             last_settled_status: candidate.status,
@@ -5179,12 +5185,25 @@ impl Runtime {
         operation_id: WorkerLifecycleOperationId,
         mut result: WorkerExecutionResult,
     ) -> Result<WorkerLifecycleAck, RuntimeError> {
-        if !result.is_accepted()
+        let termination = result
-            || result.operation != WorkerExecutionOperation::Stop
+            .termination
-            || result.termination.as_ref().is_none_or(|termination| {
+            .as_ref()
-                !termination.confirms_scope_shutdown(worker_ref, operation_id)
+            .filter(|termination| {
+                result.operation == WorkerExecutionOperation::Stop
+                    && result.execution_id == Some(termination.execution_id)
+                    && termination.confirms_scope_shutdown(worker_ref, operation_id)
             })
-        {
+            .cloned();
+        if !result.is_accepted() || termination.is_none() {
+            if let Some(termination) = termination {
+                // Scope join is independent of an artifact/resource cleanup error.
+                // Retain the original evidence without settling the pending Stop.
+                let mut state = self.lock()?;
+                let mut candidate = state.worker(worker_ref)?.clone();
+                candidate.retain_stop_termination(operation_id, termination)?;
+                state.persist_stop_worker_record(&candidate)?;
+                state.workers.insert(worker_ref.worker_id, candidate);
+            }
             if result.is_accepted() {
                 result.outcome = crate::execution::WorkerExecutionOutcome::Errored;
                 result.message = Some("stop has no matching actual scope termination evidence; reconciliation remains pending".into());
@@ -5212,6 +5231,8 @@ impl Runtime {
         if let Some(context) = candidate.restore_candidate_context.take() {
             context.discard_candidate();
         }
+        candidate
+            .retain_stop_termination(operation_id, termination.expect("validated termination"))?;
         let (candidate, ack) = candidate.stopped_candidate(operation_id);
         state.persist_stop_worker_record(&candidate)?;
         state.workers.insert(worker_ref.worker_id, candidate);
@@ -5264,6 +5285,7 @@ impl Runtime {
             owner.execution_started = true;
         }
         candidate.restore_guard.rotate();
+        candidate.restore_guard.execution_termination = None;
         candidate.pending_restore = Some(PendingWorkerRestore {
             operation_id,
             mode,
@@ -7509,6 +7531,10 @@ pub(crate) struct RestoreGuard {
     // eviction would forget request-ID conflicts and historical receipts.
     #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
     stop_owners: BTreeMap<String, StopRequestOwner>,
+    // Original execution-owner evidence, independent of lifecycle/cleanup success.
+    // Older aggregates lack it; neither migration nor a stopped status creates it.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    execution_termination: Option<crate::execution::WorkerExecutionTermination>,
 }
 
 impl Default for RestoreGuard {
@@ -7518,6 +7544,7 @@ impl Default for RestoreGuard {
             active_request_id: None,
             owners: BTreeMap::new(),
             stop_owners: BTreeMap::new(),
+            execution_termination: None,
         }
     }
 }
@@ -7579,6 +7606,29 @@ struct WorkerRecord {
 }
 
 impl WorkerRecord {
+    fn retain_stop_termination(
+        &mut self,
+        operation_id: WorkerLifecycleOperationId,
+        termination: crate::execution::WorkerExecutionTermination,
+    ) -> Result<(), RuntimeError> {
+        if self
+            .pending_stop
+            .is_none_or(|pending| pending.operation_id != operation_id)
+            || !termination.confirms_scope_shutdown(&self.worker_ref, operation_id)
+            || self
+                .restore_guard
+                .execution_termination
+                .as_ref()
+                .is_some_and(|saved| saved != &termination)
+        {
+            return Err(RuntimeError::InvalidRequest(
+                "scope termination differs from the canonical Stop operation".into(),
+            ));
+        }
+        self.restore_guard.execution_termination = Some(termination);
+        Ok(())
+    }
+
     // Pure projection shared by terminal commit and stop-owner capacity preflight.
     // Resource cleanup is deliberately outside this method.
     fn stopped_candidate(
@@ -10329,6 +10379,19 @@ mod tests {
             result
         }
 
+        fn termination_receipt(
+            &self,
+            worker_ref: &WorkerRef,
+        ) -> Result<Option<crate::execution::WorkerExecutionTermination>, String> {
+            Ok(self
+                .stop_terminations
+                .lock()
+                .unwrap()
+                .values()
+                .find(|receipt| &receipt.worker_ref == worker_ref)
+                .cloned())
+        }
+
         fn stop_worker_operation(
             &self,
             request: WorkerExecutionStopRequest,
@@ -12417,6 +12480,9 @@ mod tests {
         assert_eq!(ids[0], ids[1]);
     }
 
+    #[cfg(feature = "fs-store")]
+    include!("runtime/execution_termination_tests.rs");
+
     #[cfg(feature = "fs-store")]
     #[test]
     fn legacy_restore_guard_without_stop_owners_reopens_without_losing_observation() {
@@ -16318,9 +16384,16 @@ mod tests {
             display_name: None,
         })
         .unwrap_err();
-        assert!(error.to_string().contains(
+        assert!(
-            "unsupported Runtime store schema version 2; expected 6, 7, 8, 9, 10, 11, 12, 13, 14, or 15"
+            error
-        ));
+                .to_string()
+                .contains("unsupported Runtime store schema version 2")
+        );
+        assert!(
+            error
+                .to_string()
+                .contains(&format!("or {}", crate::fs_store::SCHEMA_VERSION))
+        );
 
         let _ = std::fs::remove_dir_all(root);
     }

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

@@ -0,0 +1,201 @@
+// These catalog tests substitute the execution response. The real Controller
+// join and fs-store reopen connection is covered in worker_backend tests.
+#[test]
+fn stop_termination_unknown_save_reconciles_original_operation_before_and_after_reopen() {
+    for reconcile_in_process in [true, false] {
+        let dir = tempfile::tempdir().unwrap();
+        let options = crate::fs_store::FsRuntimeStoreOptions::new(dir.path())
+            .with_runtime_id("termination-unknown-save");
+        let backend = Arc::new(TestExecutionBackend::default());
+        let runtime =
+            Runtime::with_fs_store_and_execution_backend(options.clone(), backend.clone()).unwrap();
+        runtime.store_config_bundle(test_bundle()).unwrap();
+        let worker = runtime
+            .create_worker(task_request("original receipt"))
+            .unwrap();
+        let operation_id = WorkerLifecycleOperationId::new();
+        runtime
+            .begin_stop_operation(&worker.worker_ref, operation_id)
+            .unwrap();
+        let result = backend.stop_worker_operation(WorkerExecutionStopRequest {
+            worker_ref: worker.worker_ref.clone(),
+            operation_id,
+        });
+        let original = result.termination.clone().unwrap();
+        runtime_store(&runtime)
+            .fail_next_worker_write(crate::fs_store::AtomicWriteFault::AfterRename);
+        assert!(matches!(
+            runtime.commit_stopped_worker(&worker.worker_ref, operation_id, result),
+            Err(RuntimeError::StoreCommitOutcomeUnknown { .. })
+        ));
+        if reconcile_in_process {
+            runtime_store(&runtime).fail_next_worker_reconcile_sync();
+            assert!(
+                runtime.execution_termination(&worker.worker_ref).is_err(),
+                "unknown durability must not fall back to the backend's in-memory receipt"
+            );
+            assert_eq!(
+                runtime.execution_termination(&worker.worker_ref).unwrap(),
+                Some(original.clone())
+            );
+            assert_eq!(
+                runtime.worker_detail(&worker.worker_ref).unwrap().status,
+                WorkerStatus::Stopped
+            );
+        }
+        assert_eq!(
+            backend.stop_operation_ids.lock().unwrap().as_slice(),
+            &[operation_id]
+        );
+        drop(runtime);
+        drop(backend);
+        let backend = Arc::new(
+            crate::worker_backend::WorkerRuntimeExecutionBackend::new(
+                crate::worker_backend::ProfileRuntimeWorkerFactory::new(
+                    dir.path().join("no-launch"),
+                ),
+            )
+            .unwrap(),
+        );
+        let reopened = Runtime::with_fs_store_and_execution_backend(options, backend).unwrap();
+        assert_eq!(
+            reopened.execution_termination(&worker.worker_ref).unwrap(),
+            Some(original)
+        );
+        assert!(!dir.path().join("no-launch").exists());
+    }
+}
+
+#[test]
+fn stop_termination_rejects_missing_wrong_operation_run_worker_and_feature_failure() {
+    for invalid in ["missing", "operation", "execution", "worker", "feature"] {
+        let (runtime, backend) = runtime_and_backend();
+        runtime.store_config_bundle(test_bundle()).unwrap();
+        let worker = runtime.create_worker(task_request(invalid)).unwrap();
+        let operation_id = WorkerLifecycleOperationId::new();
+        runtime
+            .begin_stop_operation(&worker.worker_ref, operation_id)
+            .unwrap();
+        let mut result = backend.stop_worker_operation(WorkerExecutionStopRequest {
+            worker_ref: worker.worker_ref.clone(),
+            operation_id,
+        });
+        match invalid {
+            "missing" => result.termination = None,
+            "operation" => {
+                result.termination.as_mut().unwrap().stop_operation_id =
+                    WorkerLifecycleOperationId::new()
+            }
+            "execution" => result.execution_id = Some(uuid::Uuid::now_v7()),
+            "worker" => {
+                result.termination.as_mut().unwrap().worker_ref = WorkerRef::new(WorkerId::now_v7())
+            }
+            "feature" => result.termination.as_mut().unwrap().feature_failures.push(
+                crate::execution::WorkerFeatureShutdownFailure {
+                    category: worker::hook::HookErrorCategory::Timeout,
+                    diagnostic: "Feature did not terminate".into(),
+                },
+            ),
+            _ => unreachable!(),
+        }
+        assert!(
+            runtime
+                .commit_stopped_worker(&worker.worker_ref, operation_id, result)
+                .is_err(),
+            "{invalid}"
+        );
+        assert!(
+            runtime
+                .execution_termination(&worker.worker_ref)
+                .unwrap()
+                .is_none(),
+            "{invalid}"
+        );
+        assert_eq!(
+            runtime
+                .lock()
+                .unwrap()
+                .worker(&worker.worker_ref)
+                .unwrap()
+                .pending_stop
+                .unwrap()
+                .operation_id,
+            operation_id
+        );
+    }
+}
+
+#[test]
+fn stop_termination_cleanup_error_retains_join_without_settling_stop_or_overwriting_receipt() {
+    let dir = tempfile::tempdir().unwrap();
+    let options = crate::fs_store::FsRuntimeStoreOptions::new(dir.path())
+        .with_runtime_id("termination-cleanup");
+    let backend = Arc::new(TestExecutionBackend::default());
+    let runtime =
+        Runtime::with_fs_store_and_execution_backend(options.clone(), backend.clone()).unwrap();
+    runtime.store_config_bundle(test_bundle()).unwrap();
+    let worker = runtime
+        .create_worker(task_request("cleanup failed"))
+        .unwrap();
+    let operation_id = WorkerLifecycleOperationId::new();
+    runtime
+        .begin_stop_operation(&worker.worker_ref, operation_id)
+        .unwrap();
+    let mut result = backend.stop_worker_operation(WorkerExecutionStopRequest {
+        worker_ref: worker.worker_ref.clone(),
+        operation_id,
+    });
+    result.outcome = crate::execution::WorkerExecutionOutcome::Errored;
+    result.message = Some("artifact cleanup failed after scope join".into());
+    let original = result.termination.clone().unwrap();
+    assert!(
+        matches!(runtime.commit_stopped_worker(&worker.worker_ref, operation_id, result.clone()),
+        Err(RuntimeError::WorkerExecutionRejected { result: rejected, .. }) if rejected == result)
+    );
+    assert_eq!(
+        runtime.execution_termination(&worker.worker_ref).unwrap(),
+        Some(original.clone())
+    );
+    let mut changed = result.clone();
+    changed
+        .termination
+        .as_mut()
+        .unwrap()
+        .submissions
+        .push(crate::execution::WorkerSubmissionAck {
+            submission_request_id: "not-original".into(),
+            submission_id: "other".into(),
+            disposition: protocol::SubmissionDisposition::Started,
+        });
+    assert!(matches!(
+        runtime.commit_stopped_worker(&worker.worker_ref, operation_id, changed),
+        Err(RuntimeError::InvalidRequest(_))
+    ));
+    assert_eq!(
+        runtime.execution_termination(&worker.worker_ref).unwrap(),
+        Some(original.clone())
+    );
+    drop(runtime);
+    let backend = Arc::new(
+        crate::worker_backend::WorkerRuntimeExecutionBackend::new(
+            crate::worker_backend::ProfileRuntimeWorkerFactory::new(dir.path().join("no-launch")),
+        )
+        .unwrap(),
+    );
+    let reopened = Runtime::with_fs_store_and_execution_backend(options, backend).unwrap();
+    assert_eq!(
+        reopened.execution_termination(&worker.worker_ref).unwrap(),
+        Some(original)
+    );
+    assert_eq!(
+        reopened
+            .lock()
+            .unwrap()
+            .worker(&worker.worker_ref)
+            .unwrap()
+            .pending_stop
+            .unwrap()
+            .operation_id,
+        operation_id
+    );
+}

crates/worker-runtime/src/worker_backend.rs

@@ -158,8 +158,6 @@ use workdir::{
     WorkdirSessionHandle, WorkdirSessionRouter,
 };
 
-#[cfg(test)]
-use worker::WorkerController;
 use worker::feature::builtin::manage_workdir::WorkspaceAttachedWorkdirSession;
 use worker::feature::contracts::{
     WorkerObservationError, WorkerObservationProvider, WorkerObservationSubject,
@@ -3290,6 +3288,9 @@ struct RuntimeExecutionTaskScope {
     execution_id: uuid::Uuid,
     shutdown_report: Arc<Mutex<Option<Vec<crate::execution::WorkerFeatureShutdownFailure>>>>,
     submissions: Arc<Mutex<Vec<crate::execution::WorkerSubmissionAck>>>,
+    // Dispatch provenance belongs to this original Controller scope, not its
+    // WorkerRef or the session receipts inherited by a replacement Controller.
+    submission_attempts: Arc<Mutex<HashMap<String, [u8; 32]>>>,
     #[cfg(feature = "ws-server")]
     protocol_bridge: Arc<Mutex<Option<RuntimeProtocolBridge>>>,
     tasks: Arc<Mutex<Vec<Arc<RuntimeExecutionTask>>>>,
@@ -3392,6 +3393,7 @@ impl RuntimeExecutionTaskScope {
             execution_id: uuid::Uuid::now_v7(),
             shutdown_report: Arc::default(),
             submissions: Arc::default(),
+            submission_attempts: Arc::default(),
             #[cfg(feature = "ws-server")]
             protocol_bridge: Arc::default(),
             tasks: Arc::new(Mutex::new(Vec::new())),
@@ -3402,6 +3404,83 @@ impl RuntimeExecutionTaskScope {
         scope
     }
 
+    fn submission_fingerprint(
+        input: &[Segment],
+        source: &protocol::AuthenticatedInputSource,
+    ) -> Result<[u8; 32], String> {
+        use sha2::Digest as _;
+        let payload = serde_json::to_vec(&(input, source))
+            .map_err(|_| "cannot bind original Controller submission payload".to_string())?;
+        Ok(sha2::Sha256::digest(payload).into())
+    }
+
+    fn prepare_submission(
+        &self,
+        handle: &WorkerHandle,
+        request_id: &str,
+        input: &[Segment],
+        source: &protocol::AuthenticatedInputSource,
+    ) -> Result<[u8; 32], String> {
+        let fingerprint = Self::submission_fingerprint(input, source)?;
+        let mut attempts = self
+            .submission_attempts
+            .lock()
+            .map_err(|_| "original Controller submission ownership is poisoned".to_string())?;
+        // An inherited receipt can be read or replayed, but neither attributes
+        // its original acceptance to this new Controller. None alone grants no
+        // acceptance/replay: bind only the actual dispatch that follows, then
+        // require the same handle's exact durable receipt before retaining an ack.
+        if !attempts.contains_key(request_id)
+            && matches!(
+                handle.reconcile_submission(request_id, input, source),
+                Ok(None)
+            )
+        {
+            attempts.insert(request_id.to_string(), fingerprint);
+        }
+        Ok(fingerprint)
+    }
+
+    fn retain_submission_acceptance(
+        &self,
+        ack: &crate::execution::WorkerSubmissionAck,
+        fingerprint: &[u8; 32],
+    ) -> Result<(), String> {
+        let attempts = self
+            .submission_attempts
+            .lock()
+            .map_err(|_| "original Controller submission ownership is poisoned".to_string())?;
+        if attempts.get(&ack.submission_request_id) != Some(fingerprint) {
+            return Ok(());
+        }
+        let mut submissions = self
+            .submissions
+            .lock()
+            .map_err(|_| "original Controller submission receipts are poisoned".to_string())?;
+        if let Some(saved) = submissions
+            .iter()
+            .find(|saved| saved.submission_request_id == ack.submission_request_id)
+        {
+            if saved != ack {
+                return Err("original Controller submission acceptance changed".into());
+            }
+        } else {
+            // A joined scope may already have an original canonical receipt,
+            // including a pending Stop after resource-cleanup failure. A later
+            // read must never amend that immutable evidence.
+            if self
+                .shutdown_report
+                .lock()
+                .map_err(|_| "Controller shutdown report lock is poisoned".to_string())?
+                .is_some()
+            {
+                return Ok(());
+            }
+            submissions.push(ack.clone());
+        }
+        Ok(())
+    }
+
     fn has_cleanup_candidates(&self) -> Result<bool, String> {
         self.cleanup_candidates
             .lock()
@@ -3793,6 +3872,8 @@ struct WorkerAdmission {
 pub struct WorkerRuntimeExecutionBackend<F = ProfileRuntimeWorkerFactory> {
     #[cfg(test)]
     bridge_fault: Mutex<Option<crate::execution::RestoreTestGate<WorkerHandle>>>,
+    #[cfg(test)]
+    lost_submission_ack: Mutex<Option<String>>,
     backend_id: String,
     factory: Arc<F>,
     working_directory_materializer: Option<Arc<dyn WorkingDirectoryMaterializer>>,
@@ -3833,6 +3914,8 @@ where
         Ok(Self {
             #[cfg(test)]
             bridge_fault: Mutex::default(),
+            #[cfg(test)]
+            lost_submission_ack: Mutex::default(),
             backend_id: DEFAULT_BACKEND_ID.to_string(),
             factory: Arc::new(factory),
             working_directory_materializer: None,
@@ -4550,6 +4633,8 @@ where
         submission_request_id: String,
     ) -> WorkerExecutionResult {
         let request_id = submission_request_id.clone();
+        #[cfg(test)]
+        let lost_ack = self.lost_submission_ack.lock().unwrap().take();
         self.run_cancellable_on_adapter_runtime(USER_INPUT_TASK_TIMEOUT, async move {
             // Subscribe before enqueueing so a fast durable acceptance cannot
             // race the Runtime acknowledgement.
@@ -4567,6 +4652,12 @@ where
                             submission_id,
                             disposition,
                         }) if submission_request_id == request_id => {
+                            #[cfg(test)]
+                            if lost_ack.as_deref() == Some(request_id.as_str()) {
+                                // Lose only the caller's delivery after the real
+                                // Controller emitted its durable acceptance.
+                                continue;
+                            }
                             return Ok((submission_id, disposition));
                         }
                         Ok(Event::SubmissionRejected {
@@ -5134,6 +5225,57 @@ where
         Ok(())
     }
 
+    fn read_termination_receipt(
+        &self,
+        worker_ref: &WorkerRef,
+        saved: Option<&crate::execution::WorkerExecutionTermination>,
+    ) -> Result<Option<crate::execution::WorkerExecutionTermination>, String> {
+        let operation = self.worker_lock(worker_ref)?;
+        let _guard = operation
+            .lock()
+            .map_err(|_| "Worker operation lock poisoned".to_string())?;
+        if self
+            .pending_spawns
+            .lock()
+            .map_err(|_| "Worker spawn registry poisoned".to_string())?
+            .contains_key(worker_ref)
+            || self
+                .pending_restores
+                .lock()
+                .map_err(|_| "Worker restore registry poisoned".to_string())?
+                .contains_key(worker_ref)
+            || self
+                .subject_reservations
+                .lock()
+                .map_err(|_| "Subject reservation registry poisoned".to_string())?
+                .keys()
+                .any(|(worker, _)| worker == worker_ref)
+        {
+            return Ok(None);
+        }
+        let current_execution = self
+            .workers
+            .lock()
+            .map_err(|_| "Worker execution registry poisoned".to_string())?
+            .get(worker_ref)
+            .map(|execution| execution.tasks.execution_id);
+        let receipt = self
+            .completed_stops
+            .lock()
+            .map_err(|_| "Worker termination receipts poisoned".to_string())?
+            .get(worker_ref)
+            .cloned();
+        if let (Some(current), Some(saved)) = (&receipt, saved)
+            && current != saved
+        {
+            return Err("retained scope termination differs from canonical Stop receipt".into());
+        }
+        let receipt = receipt.or_else(|| saved.cloned());
+        // The positive receipt comes only from the joined scope, never from
+        // absence. A Restore's new Controller cannot reuse its predecessor's end.
+        Ok(receipt.filter(|receipt| current_execution.is_none_or(|id| id == receipt.execution_id)))
+    }
+
     fn stop_worker_operation_under_lock(
         &self,
         request: WorkerExecutionStopRequest,
@@ -7320,44 +7462,20 @@ where
         &self,
         worker_ref: &WorkerRef,
     ) -> Result<Option<crate::execution::WorkerExecutionTermination>, String> {
-        let operation = self.worker_lock(worker_ref)?;
+        self.read_termination_receipt(worker_ref, None)
-        let _guard = operation
+    }
-            .lock()
+
-            .map_err(|_| "Worker operation lock poisoned".to_string())?;
+    fn reconcile_termination_receipt(
-        if self
+        &self,
-            .pending_spawns
+        worker_ref: &WorkerRef,
-            .lock()
+        saved: &crate::execution::WorkerExecutionTermination,
-            .map_err(|_| "Worker spawn registry poisoned".to_string())?
+    ) -> Result<Option<crate::execution::WorkerExecutionTermination>, String> {
-            .contains_key(worker_ref)
+        if !saved.confirms_scope_shutdown(worker_ref, saved.stop_operation_id) {
-            || self
+            return Err(
-                .pending_restores
+                "canonical scope termination has a Worker mismatch or Feature failure".into(),
-                .lock()
+            );
-                .map_err(|_| "Worker restore registry poisoned".to_string())?
-                .contains_key(worker_ref)
-            || self
-                .subject_reservations
-                .lock()
-                .map_err(|_| "Subject reservation registry poisoned".to_string())?
-                .keys()
-                .any(|(worker, _)| worker == worker_ref)
-        {
-            return Ok(None);
         }
-        let current_execution = self
+        self.read_termination_receipt(worker_ref, Some(saved))
-            .workers
-            .lock()
-            .map_err(|_| "Worker execution registry poisoned".to_string())?
-            .get(worker_ref)
-            .map(|execution| execution.tasks.execution_id);
-        let receipt = self
-            .completed_stops
-            .lock()
-            .map_err(|_| "Worker termination receipts poisoned".to_string())?
-            .get(worker_ref)
-            .cloned();
-        // The positive receipt comes only from the joined scope, never from
-        // absence. A Restore's new Controller cannot reuse its predecessor's end.
-        Ok(receipt.filter(|receipt| current_execution.is_none_or(|id| id == receipt.execution_id)))
     }
 
     fn reconcile_submission(
@@ -7443,8 +7561,22 @@ where
                 "exact submission receipt lookup failed; submission acceptance remains unknown",
             ),
         };
+        if result.is_accepted()
+            && let Some(ack) = result.submission.as_ref()
+            && let Err(message) = RuntimeExecutionTaskScope::submission_fingerprint(
+                &segments, &source,
+            )
+            .and_then(|fingerprint| {
+                execution
+                    .tasks
+                    .retain_submission_acceptance(ack, &fingerprint)
+            })
+        {
+            result = unknown(&message);
+        }
         // Identify the Controller observed, not the originating run of a receipt
-        // retained across Controller replacement. This is not termination evidence.
+        // retained across Controller replacement. Only this scope's original
+        // dispatch binding above permits adding it to termination evidence.
         result.execution_id = Some(execution.tasks.execution_id);
         result
     }
@@ -7469,13 +7601,27 @@ where
                 error.to_string(),
             );
         }
-        let (worker, worker_state, _workspace_client) = match self.get_execution(worker_ref) {
+        // Pin the actual endpoint and scope while holding the per-Worker lock;
-            Ok(execution) => execution,
+        // do not attach a late acceptance through a fresh WorkerRef lookup.
-            Err(mut result) => {
+        let execution = match self.workers.lock() {
-                result.operation = WorkerExecutionOperation::Input;
+            Ok(workers) => match workers.get(worker_ref) {
-                return result;
+                Some(execution) => execution.clone(),
+                None => {
+                    return WorkerExecutionResult::rejected(
+                        WorkerExecutionOperation::Input,
+                        "WorkerRef does not reference a live Worker execution",
+                    );
+                }
+            },
+            Err(_) => {
+                return WorkerExecutionResult::errored(
+                    WorkerExecutionOperation::Input,
+                    "worker adapter registry lock is poisoned",
+                );
             }
         };
+        let worker = execution.handle.clone();
+        let worker_state = execution.worker_state.clone();
 
         if input.kind == WorkerInputKind::Notify {
             let notification_request_id = input
@@ -7512,6 +7658,7 @@ where
             );
         }
 
+        let mut submission_binding = None;
         let (method, submission_request_id) = match input.kind {
             WorkerInputKind::User | WorkerInputKind::UserIfIdle => {
                 let Some(submission_id) = input
@@ -7526,6 +7673,20 @@ where
                     );
                 };
                 let (segments, source) = runtime_submission_payload(&input, &submission_id);
+                submission_binding = match execution.tasks.prepare_submission(
+                    &worker,
+                    &submission_id,
+                    &segments,
+                    &source,
+                ) {
+                    Ok(fingerprint) => Some(fingerprint),
+                    Err(message) => {
+                        return WorkerExecutionResult::errored(
+                            WorkerExecutionOperation::Input,
+                            message,
+                        );
+                    }
+                };
                 let method = if input.kind == WorkerInputKind::UserIfIdle {
                     Method::SubmitIfIdle {
                         submission_request_id: submission_id.clone(),
@@ -7566,27 +7727,17 @@ where
         } else {
             self.send_method(WorkerExecutionOperation::Input, worker, method)
         };
-        if let Some(execution) = self
+        if result.is_accepted()
-            .workers
+            && let Some(ack) = result.submission.as_ref()
-            .lock()
+            && Some(&ack.submission_request_id) == submission_request_id.as_ref()
-            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            && let Some(fingerprint) = submission_binding
-            .get(worker_ref)
+            && let Err(message) = execution
+                .tasks
+                .retain_submission_acceptance(ack, &fingerprint)
         {
-            result.execution_id = Some(execution.tasks.execution_id);
+            result = WorkerExecutionResult::errored(WorkerExecutionOperation::Input, message);
-            if result.is_accepted()
-                && let Some(ack) = result.submission.as_ref()
-                && Some(&ack.submission_request_id) == submission_request_id.as_ref()
-            {
-                let mut submissions = execution
-                    .tasks
-                    .submissions
-                    .lock()
-                    .unwrap_or_else(std::sync::PoisonError::into_inner);
-                if !submissions.contains(ack) {
-                    submissions.push(ack.clone());
-                }
-            }
         }
+        result.execution_id = Some(execution.tasks.execution_id);
         result
     }
 
@@ -8004,6 +8155,8 @@ mod tests {
 
     include!("subject_delete_backend_tests.rs");
     include!("worker_backend/startup_binding_tests.rs");
+    include!("worker_backend/termination_persistence_tests.rs");
+    include!("worker_backend/submission_termination_tests.rs");
     #[cfg(feature = "fs-store")]
     include!("subject_delete_create_tests.rs");
 
@@ -11263,20 +11416,11 @@ mod tests {
         observed_workspace_clients: Arc<Mutex<Vec<(String, Option<String>, bool)>>>,
     }
 
-    #[async_trait]
+    impl MockFactory {
-    impl RuntimeWorkerFactory for MockFactory {
+        async fn spawn_controller_with_saved_source(
-        async fn prepare_restore(
-            &self,
-            request: WorkerExecutionRestoreRequest,
-        ) -> Result<Box<dyn RuntimeRestoreCandidate>, RuntimeRestoreFailure> {
-            self.restore_controller(request).await.map(|controller| {
-                Box::new(TestRestoreCandidate(Some(controller))) as Box<dyn RuntimeRestoreCandidate>
-            })
-        }
-
-        async fn spawn_controller(
             &self,
             request: WorkerExecutionSpawnRequest,
+            save_restore_source: bool,
         ) -> Result<RuntimeWorkerController, RuntimeRestoreFailure> {
             let manifest = WorkerManifest::from_toml(
                 r#"
@@ -11356,7 +11500,7 @@ mod tests {
                 workspace_client.is_available(),
             ));
             let scope = Scope::writable(&scope_root).map_err(|err| err.to_string())?;
-            let worker = Worker::new(
+            let mut worker = Worker::new(
                 manifest,
                 Engine::<_, agen::state::Mutable, worker::SessionHistoryMetadata>::new_annotated(
                     self.client.clone(),
@@ -11368,18 +11512,50 @@ mod tests {
             )
             .await
             .map_err(|err| err.to_string())?;
-            let (handle, shutdown_rx) =
+            if save_restore_source {
-                WorkerController::spawn_runtime_managed(worker, &self.runtime_base)
+                worker
-                    .await
+                    .engine_mut()
-                    .map_err(|err| err.to_string())?;
+                    .set_system_prompt("saved original Controller fixture prompt");
+                worker
+                    .enable_worker_metadata_write_through()
+                    .map_err(|error| error.to_string())?;
+            }
+            let started = worker::bootstrap::start_worker_controller(
+                worker,
+                worker::bootstrap::WorkerBootstrapLayout::Direct {
+                    runtime_base: self.runtime_base.clone(),
+                },
+            )
+            .await
+            .map_err(|err| err.to_string())?;
             Ok(RuntimeWorkerController {
-                handle,
+                handle: started.handle,
-                shutdown: Arc::new(tokio::sync::Mutex::new(Some(shutdown_rx))),
+                shutdown: Arc::new(tokio::sync::Mutex::new(Some(started.shutdown))),
-                controller_task: tokio::spawn(async {}),
+                controller_task: started.controller_task,
                 workspace_client,
                 restore_streams: None,
             })
         }
+    }
+
+    #[async_trait]
+    impl RuntimeWorkerFactory for MockFactory {
+        async fn prepare_restore(
+            &self,
+            request: WorkerExecutionRestoreRequest,
+        ) -> Result<Box<dyn RuntimeRestoreCandidate>, RuntimeRestoreFailure> {
+            self.restore_controller(request).await.map(|controller| {
+                Box::new(TestRestoreCandidate(Some(controller))) as Box<dyn RuntimeRestoreCandidate>
+            })
+        }
+
+        async fn spawn_controller(
+            &self,
+            request: WorkerExecutionSpawnRequest,
+        ) -> Result<RuntimeWorkerController, RuntimeRestoreFailure> {
+            self.spawn_controller_with_saved_source(request, false)
+                .await
+        }
         async fn restore_controller(
             &self,
             request: WorkerExecutionRestoreRequest,

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

@@ -0,0 +1,368 @@
+// Connect real Submit/receipt persistence, original Controller ownership and
+// Runtime's canonical stop receipt. Only model I/O and one caller ACK delivery
+// are substituted; no receipt, Shutdown report or joined Controller is faked.
+#[cfg(feature = "fs-store")]
+struct SavedSourceMockFactory(MockFactory);
+
+#[cfg(feature = "fs-store")]
+#[async_trait]
+impl RuntimeWorkerFactory for SavedSourceMockFactory {
+    async fn spawn_controller(
+        &self,
+        request: WorkerExecutionSpawnRequest,
+    ) -> Result<RuntimeWorkerController, RuntimeRestoreFailure> {
+        self.0
+            .spawn_controller_with_saved_source(request, true)
+            .await
+    }
+
+    async fn prepare_restore(
+        &self,
+        _: WorkerExecutionRestoreRequest,
+    ) -> Result<Box<dyn RuntimeRestoreCandidate>, RuntimeRestoreFailure> {
+        panic!("this fixture explicitly restores the original saved Session below")
+    }
+}
+
+#[cfg(feature = "fs-store")]
+#[test]
+#[serial_test::serial(worker_allocation)]
+fn stop_termination_lost_submit_ack_reconciles_only_original_controller_after_timeout() {
+    let temp = tempfile::tempdir_in(".").unwrap();
+    let root = fs::canonicalize(temp.path()).unwrap();
+    let socket_base = PathBuf::from(temp.path().file_name().unwrap()).join("c");
+    let options = crate::fs_store::FsRuntimeStoreOptions::new(root.join("catalog"))
+        .with_runtime_id("lost-submit-ack");
+    let client = MockClient::new(simple_text_events());
+    let calls = client.call_count.clone();
+    let new_backend = || {
+        Arc::new(
+            WorkerRuntimeExecutionBackend::new(SavedSourceMockFactory(MockFactory {
+                client: client.clone(),
+                runtime_base: socket_base.clone(),
+                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 backend = new_backend();
+    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("lost ACK")).unwrap();
+    let original_execution = backend.workers.lock().unwrap()[&worker.worker_ref].clone();
+    let mut input = WorkerInput::user("  original immutable input  ");
+    input.kind = WorkerInputKind::UserIfIdle;
+    input.submission_request_id = Some("original-lost-ack".into());
+    *backend.lost_submission_ack.lock().unwrap() = input.submission_request_id.clone();
+
+    // The real Controller accepts and persists; only this caller's matching ACK
+    // is lost. The production acceptance timer, not a sleep, ends the wait.
+    let lost = backend.dispatch_input(&worker.worker_ref, input.clone());
+    assert_eq!(
+        lost.outcome,
+        crate::execution::WorkerExecutionOutcome::Errored
+    );
+    assert!(
+        lost.message
+            .as_deref()
+            .unwrap()
+            .contains("timed out waiting")
+    );
+    assert!(lost.submission.is_none());
+    assert_eq!(
+        lost.execution_id,
+        Some(original_execution.tasks.execution_id)
+    );
+    assert!(
+        original_execution
+            .tasks
+            .submissions
+            .lock()
+            .unwrap()
+            .is_empty()
+    );
+    assert_eq!(calls.load(Ordering::SeqCst), 1);
+    let (segments, source) = runtime_submission_payload(&input, "original-lost-ack");
+    let durable = original_execution
+        .handle
+        .reconcile_submission("original-lost-ack", &segments, &source)
+        .unwrap()
+        .unwrap();
+
+    let mut wrong = input.clone();
+    wrong.segments = Some(vec![Segment::text("different immutable input")]);
+    assert!(
+        !backend
+            .reconcile_submission(&worker.worker_ref, wrong)
+            .is_accepted()
+    );
+    assert!(
+        original_execution
+            .tasks
+            .submissions
+            .lock()
+            .unwrap()
+            .is_empty()
+    );
+    let found = backend.reconcile_submission(&worker.worker_ref, input.clone());
+    assert!(found.is_accepted(), "{found:?}");
+    assert_eq!(
+        found.execution_id,
+        Some(original_execution.tasks.execution_id)
+    );
+    let ack = found.submission.unwrap();
+    assert_eq!(ack.submission_request_id, "original-lost-ack");
+    assert_eq!(ack.submission_id, durable.submission_id);
+    assert_eq!(ack.disposition, durable.disposition);
+    assert_eq!(
+        backend
+            .reconcile_submission(&worker.worker_ref, input.clone())
+            .submission,
+        Some(ack.clone())
+    );
+    assert_eq!(
+        original_execution
+            .tasks
+            .submissions
+            .lock()
+            .unwrap()
+            .as_slice(),
+        &[ack.clone()]
+    );
+    assert_eq!(
+        calls.load(Ordering::SeqCst),
+        1,
+        "reconcile never redispatches"
+    );
+
+    // A second real acceptance is deliberately not reconciled until after
+    // shutdown. It must not amend an already-issued original termination.
+    let mut late_input = WorkerInput::user("late receipt lookup");
+    late_input.submission_request_id = Some("late-lost-ack".into());
+    *backend.lost_submission_ack.lock().unwrap() = late_input.submission_request_id.clone();
+    let late_lost = backend.dispatch_input(&worker.worker_ref, late_input.clone());
+    assert!(
+        late_lost
+            .message
+            .as_deref()
+            .unwrap()
+            .contains("timed out waiting")
+    );
+    let (late_segments, late_source) = runtime_submission_payload(&late_input, "late-lost-ack");
+    let late_fingerprint =
+        RuntimeExecutionTaskScope::submission_fingerprint(&late_segments, &late_source).unwrap();
+
+    runtime.stop_worker(&worker.worker_ref, None).unwrap();
+    assert!(original_execution.handle.protocol_is_closed());
+    assert!(original_execution.tasks.tasks.lock().unwrap().is_empty());
+    let original = runtime
+        .execution_termination(&worker.worker_ref)
+        .unwrap()
+        .unwrap();
+    assert_eq!(original.execution_id, original_execution.tasks.execution_id);
+    assert_eq!(original.submissions, vec![ack.clone()]);
+    assert!(original.shutdown_command_id.is_some());
+    assert!(original.feature_failures.is_empty());
+    let late_receipt = original_execution
+        .handle
+        .reconcile_submission("late-lost-ack", &late_segments, &late_source)
+        .unwrap()
+        .unwrap();
+    let late_ack = crate::execution::WorkerSubmissionAck {
+        submission_request_id: late_receipt.submission_request_id,
+        submission_id: late_receipt.submission_id,
+        disposition: late_receipt.disposition,
+    };
+    original_execution
+        .tasks
+        .retain_submission_acceptance(&late_ack, &late_fingerprint)
+        .unwrap();
+    assert_eq!(
+        original_execution
+            .tasks
+            .termination(&worker.worker_ref, original.stop_operation_id),
+        Some(original.clone()),
+        "a late lookup cannot rewrite the original scope receipt"
+    );
+    assert_eq!(
+        runtime.execution_termination(&worker.worker_ref).unwrap(),
+        Some(original.clone())
+    );
+    drop(runtime);
+    drop(backend);
+
+    let backend = new_backend();
+    let reopened =
+        EmbeddedRuntime::with_fs_store_and_execution_backend(options, backend.clone()).unwrap();
+    assert_eq!(
+        reopened.execution_termination(&worker.worker_ref).unwrap(),
+        Some(original.clone())
+    );
+
+    // Restore the same real saved Session into another actual Controller. Its
+    // receipt reader must see the old ACK, but it cannot claim that acceptance
+    // for this different scope, even if the caller replays the same Submit.
+    let restore_root = root.clone();
+    let replacement_socket = socket_base.with_file_name("c2");
+    let replacement_client = MockClient::new(simple_text_events());
+    let replacement_calls = replacement_client.call_count.clone();
+    let replacement = backend
+        .run_on_adapter_runtime(async move {
+            let store = CombinedStore::new(
+                FsStore::new(restore_root.join("sessions")).map_err(|error| error.to_string())?,
+                FsWorkerStore::new(restore_root.join("metadata"))
+                    .map_err(|error| error.to_string())?,
+            );
+            let fallback = WorkerManifest::from_toml(
+                r#"
+            [worker]
+            name = "r"
+            [model]
+            scheme = "anthropic"
+            model_id = "test-model"
+            auth = { kind = "none" }
+            [engine]
+            max_tokens = 100
+            [scope]
+            allow = []
+        "#,
+            )
+            .map_err(|error| error.to_string())?;
+            let prepared =
+                Worker::prepare_restored_from_worker_metadata_with_context_and_model_client(
+                    "r",
+                    fallback,
+                    store,
+                    PromptCatalogSource::builtins_only(),
+                    WorkerWorkspaceContext::no_workspace(),
+                    WorkerFilesystemAuthority::None,
+                    Some(Box::new(replacement_client)),
+                )
+                .await
+                .map_err(|error| error.to_string())?;
+            let restored = prepared
+                .activate()
+                .await
+                .map_err(|error| error.to_string())?;
+            worker::bootstrap::start_worker_controller(
+                restored,
+                WorkerBootstrapLayout::Direct {
+                    runtime_base: replacement_socket,
+                },
+            )
+            .await
+            .map_err(|error| error.to_string())
+        })
+        .unwrap();
+    assert!(
+        !original_execution
+            .handle
+            .same_controller(&replacement.handle)
+    );
+    let read_old = replacement
+        .handle
+        .reconcile_submission("original-lost-ack", &segments, &source)
+        .unwrap()
+        .unwrap();
+    assert_eq!(read_old, durable);
+    let operation = backend.worker_lock(&worker.worker_ref).unwrap();
+    let connected = {
+        let _guard = operation.lock().unwrap();
+        backend.connect_handle(
+            WorkerExecutionOperation::Restore,
+            worker.worker_ref.clone(),
+            test_execution_context(worker.worker_ref.clone()),
+            replacement.handle,
+            Arc::new(tokio::sync::Mutex::new(Some(replacement.shutdown))),
+            replacement.controller_task,
+            None,
+            Vec::new(),
+            None,
+            None,
+        )
+    };
+    assert!(
+        matches!(connected, WorkerExecutionSpawnResult::Connected { .. }),
+        "{connected:?}"
+    );
+    let replacement_execution = backend.workers.lock().unwrap()[&worker.worker_ref].clone();
+    assert_ne!(
+        replacement_execution.tasks.execution_id,
+        original.execution_id
+    );
+    assert!(
+        reopened
+            .execution_termination(&worker.worker_ref)
+            .unwrap()
+            .is_none()
+    );
+    let read = backend.reconcile_submission(&worker.worker_ref, input.clone());
+    assert!(read.is_accepted(), "{read:?}");
+    assert_eq!(read.submission, Some(ack.clone()));
+    assert_eq!(
+        read.execution_id,
+        Some(replacement_execution.tasks.execution_id)
+    );
+    assert!(
+        replacement_execution
+            .tasks
+            .submissions
+            .lock()
+            .unwrap()
+            .is_empty()
+    );
+    let replay = backend.dispatch_input(&worker.worker_ref, input);
+    assert!(replay.is_accepted(), "{replay:?}");
+    assert_eq!(replay.submission, Some(ack.clone()));
+    assert!(
+        replacement_execution
+            .tasks
+            .submission_attempts
+            .lock()
+            .unwrap()
+            .is_empty()
+    );
+    assert!(
+        replacement_execution
+            .tasks
+            .submissions
+            .lock()
+            .unwrap()
+            .is_empty()
+    );
+    assert_eq!(replacement_calls.load(Ordering::SeqCst), 0);
+    let mut own_input = WorkerInput::user("new Controller's own input");
+    own_input.submission_request_id = Some("replacement-own-input".into());
+    let own = backend.dispatch_input(&worker.worker_ref, own_input);
+    assert!(own.is_accepted(), "{own:?}");
+    let own_ack = own.submission.unwrap();
+    assert_eq!(
+        own.execution_id,
+        Some(replacement_execution.tasks.execution_id)
+    );
+    let stopped = backend.stop_worker_operation(WorkerExecutionStopRequest {
+        worker_ref: worker.worker_ref.clone(),
+        operation_id: crate::execution::WorkerLifecycleOperationId::new(),
+    });
+    assert!(stopped.is_accepted(), "{stopped:?}");
+    let replacement_receipt = stopped.termination.unwrap();
+    assert_eq!(
+        replacement_receipt.execution_id,
+        replacement_execution.tasks.execution_id
+    );
+    assert_ne!(replacement_receipt.execution_id, original.execution_id);
+    assert!(!replacement_receipt.submissions.contains(&ack));
+    assert_eq!(replacement_receipt.submissions, vec![own_ack]);
+    assert!(replacement_receipt.shutdown_command_id.is_some());
+    assert!(replacement_execution.tasks.tasks.lock().unwrap().is_empty());
+    assert!(
+        reopened.execution_termination(&worker.worker_ref).is_err(),
+        "replacement cannot overwrite original canonical receipt"
+    );
+}

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

@@ -0,0 +1,157 @@
+#[cfg(feature = "fs-store")]
+#[test]
+fn stop_termination_real_join_survives_fs_reopen_rejects_replacement_and_legacy_missing_receipt() {
+    // Keep socket names relative and short. The focused invocation runs from an
+    // isolated scratch cwd; data paths are still explicit absolute fixture paths.
+    let temp = tempfile::tempdir_in(".").unwrap();
+    let root = fs::canonicalize(temp.path()).unwrap();
+    let socket_base = PathBuf::from(temp.path().file_name().unwrap()).join("c");
+    let options = crate::fs_store::FsRuntimeStoreOptions::new(root.join("runtime"))
+        .with_runtime_id("original-scope-receipt");
+    let new_backend = || {
+        Arc::new(
+            WorkerRuntimeExecutionBackend::new(MockFactory {
+                client: MockClient::sequential(vec![]),
+                runtime_base: socket_base.clone(),
+                cwd: root.clone(),
+                store_dir: root.join("sessions"),
+                worker_metadata_dir: root.join("metadata"),
+                observed_cwds: Arc::new(Mutex::new(Vec::new())),
+                observed_workspace_clients: Arc::new(Mutex::new(Vec::new())),
+            })
+            .unwrap(),
+        )
+    };
+    let backend = new_backend();
+    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("durable original receipt"))
+        .unwrap();
+    assert!(
+        runtime
+            .execution_termination(&worker.worker_ref)
+            .unwrap()
+            .is_none()
+    );
+    let mut input = WorkerInput::user("original submission");
+    input.submission_request_id = Some("original-request".into());
+    let acceptance = runtime
+        .send_input(&worker.worker_ref, input)
+        .unwrap()
+        .submission
+        .unwrap();
+    runtime.stop_worker(&worker.worker_ref, None).unwrap();
+    let original = runtime
+        .execution_termination(&worker.worker_ref)
+        .unwrap()
+        .unwrap();
+    assert_eq!(
+        original.scope,
+        crate::execution::WorkerExecutionTerminationScope::Controller
+    );
+    assert!(original.shutdown_command_id.is_some());
+    assert!(original.feature_failures.is_empty());
+    assert!(original.submissions.contains(&acceptance));
+    assert_eq!(
+        backend.termination_receipt(&worker.worker_ref).unwrap(),
+        Some(original.clone())
+    );
+    assert!(backend.workers.lock().unwrap().is_empty());
+    let worker_path = options
+        .root
+        .join("workers")
+        .join(worker.worker_id.to_string())
+        .join("worker.json");
+    let document: serde_json::Value =
+        serde_json::from_slice(&fs::read(&worker_path).unwrap()).unwrap();
+    assert_eq!(
+        document["restore_guard"]["execution_termination"],
+        serde_json::to_value(&original).unwrap()
+    );
+    drop(runtime);
+    drop(backend);
+
+    let backend = new_backend();
+    let reopened =
+        EmbeddedRuntime::with_fs_store_and_execution_backend(options.clone(), backend.clone())
+            .unwrap();
+    assert!(backend.completed_stops.lock().unwrap().is_empty());
+    assert!(backend.workers.lock().unwrap().is_empty());
+    assert_eq!(
+        reopened.execution_termination(&worker.worker_ref).unwrap(),
+        Some(original.clone())
+    );
+
+    // Exercise the backend ownership check even if a replacement was installed
+    // outside Runtime's lifecycle coordinator. Empty old maps cannot bless it.
+    assert!(matches!(
+        backend.restore_worker(direct_restore_request(worker.worker_ref.clone())),
+        WorkerExecutionSpawnResult::Connected { .. }
+    ));
+    assert!(
+        reopened
+            .execution_termination(&worker.worker_ref)
+            .unwrap()
+            .is_none()
+    );
+    let replacement_stop = backend.stop_worker_operation(WorkerExecutionStopRequest {
+        worker_ref: worker.worker_ref.clone(),
+        operation_id: crate::execution::WorkerLifecycleOperationId::new(),
+    });
+    assert!(replacement_stop.is_accepted(), "{replacement_stop:?}");
+    assert_ne!(
+        replacement_stop.termination.as_ref().unwrap().execution_id,
+        original.execution_id
+    );
+    assert!(
+        reopened.execution_termination(&worker.worker_ref).is_err(),
+        "a different run/stop-operation receipt must not replace canonical evidence"
+    );
+    drop(reopened);
+    drop(backend);
+
+    // Released schema 15 never held an original termination. Its migration must
+    // preserve status/observation but must not reconstruct shutdown from them.
+    let mut legacy = document;
+    legacy["schema_version"] = serde_json::json!(15);
+    legacy["restore_guard"]
+        .as_object_mut()
+        .unwrap()
+        .remove("execution_termination");
+    let observation = legacy["restore_guard"]["observation_token"].clone();
+    fs::write(&worker_path, serde_json::to_vec_pretty(&legacy).unwrap()).unwrap();
+    let runtime_path = options.root.join("runtime.json");
+    let mut runtime_document: serde_json::Value =
+        serde_json::from_slice(&fs::read(&runtime_path).unwrap()).unwrap();
+    runtime_document["schema_version"] = serde_json::json!(15);
+    fs::write(
+        &runtime_path,
+        serde_json::to_vec_pretty(&runtime_document).unwrap(),
+    )
+    .unwrap();
+    crate::fs_store::FsRuntimeStore::migrate(&options).unwrap();
+    let backend = new_backend();
+    let reopened =
+        EmbeddedRuntime::with_fs_store_and_execution_backend(options.clone(), backend).unwrap();
+    assert_eq!(
+        reopened.worker_detail(&worker.worker_ref).unwrap().status,
+        crate::catalog::WorkerStatus::Stopped
+    );
+    assert!(
+        reopened
+            .execution_termination(&worker.worker_ref)
+            .unwrap()
+            .is_none()
+    );
+    let migrated: serde_json::Value =
+        serde_json::from_slice(&fs::read(&worker_path).unwrap()).unwrap();
+    assert_eq!(migrated["restore_guard"]["observation_token"], observation);
+    assert!(
+        migrated["restore_guard"]
+            .get("execution_termination")
+            .is_none()
+    );
+}

crates/worker-runtime/src/working_directory.rs

@@ -4518,6 +4518,11 @@ mod tests {
         let initial_ref = bound.status().summary.creation_ref.expect("creation ref");
 
         git(&bound.root, &["switch", "-c", "observed-branch"]);
+        git(&bound.root, &["config", "user.name", "Yoi Test"]);
+        git(
+            &bound.root,
+            &["config", "user.email", "test@example.invalid"],
+        );
         fs::write(bound.root.join("observed.txt"), "observed\n").unwrap();
         git(&bound.root, &["add", "observed.txt"]);
         git(&bound.root, &["commit", "-m", "advance workdir"]);