fix: fence original host writers and retain canonical job cleanup children

crates/workspace-server/src/backend_job_host_effects.rs

@@ -1,12 +1,16 @@
-//! Live ownership of Host effects admitted for one exact Job execution.
+//! Exact Backend Job Host admission, independent of Controller and domain completion.
 //!
-//! This supplements durable allocation/submission/domain receipts; it is not a Job state machine.
+//! Each accepted callback retains a shared admission-file lease through actual DB/Host work,
-//! Create it only when initially admitting an execution, before its first Host callback. Recovery
+//! including discarded-response cleanup. After the canonical attempt closes admission, an
-//! must not manufacture a drain receipt by reconstructing an empty owner after a restart. A lost
+//! exclusive lease on that same retained resource fences those callbacks across Server owners;
-//! owner requires reconciliation of the retained domain operations and writer fences instead.
+//! neither an empty counter nor process exit issues a receipt. The original identity file must
-//! Closing admission does not cancel accepted operations. Their existing DbExecutor lifetimes
+//! exist and match. This does not certify detached external work: Backend Job grants permit
-//! retain the request lease through actual work and discarded-result cleanup, even if the HTTP
+//! only result and bounded synchronous Subject operations, not arbitrary external mutations.
-//! waiter disappears. Detached work must explicitly retain this same lease or is not covered.
+//! Canonical notification delivery and Worker removal retain their separate operation owners.
+use sha2::{Digest, Sha256};
+use std::fs::{File, OpenOptions};
+use std::io::{Read, Write};
+use std::path::{Path, PathBuf};
 use std::sync::{Arc, Mutex};
 
 use tokio::sync::Notify;
@@ -58,19 +62,69 @@ struct Admission {
 
 pub(crate) struct JobHostEffects {
     owner: JobHostOwner,
+    admission_path: PathBuf,
     admission: Mutex<Admission>,
     finished: Notify,
 }
 
 impl JobHostEffects {
-    pub(crate) fn initially_admit(owner: JobHostOwner) -> Arc<Self> {
+    pub(crate) fn initially_admit(owner: JobHostOwner, database: &Path) -> Result<Arc<Self>> {
-        Arc::new(Self {
+        let admission_path = admission_path(database, &owner)?;
+        let parent = admission_path.parent().unwrap();
+        std::fs::create_dir_all(parent)?;
+        // Publish a complete identity before any callback can acquire its lease.
+        // The temporary name is not an execution, release or recovery identity.
+        let temporary = parent.join(format!(".{}.tmp", uuid::Uuid::new_v4()));
+        let publish = (|| -> Result<()> {
+            let mut file = OpenOptions::new()
+                .write(true)
+                .create_new(true)
+                .open(&temporary)?;
+            file.write_all(
+                &serde_json::to_vec(&owner).map_err(|error| Error::Store(error.to_string()))?,
+            )?;
+            file.sync_all()?;
+            match std::fs::hard_link(&temporary, &admission_path) {
+                Ok(()) => {}
+                Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {
+                    open_admission(&admission_path, &owner)?;
+                }
+                Err(error) => return Err(error.into()),
+            }
+            // Resolve a prior publication whose directory sync acknowledgement failed.
+            File::open(parent)?.sync_all()?;
+            Ok(())
+        })();
+        let _ = std::fs::remove_file(temporary);
+        publish?;
+        Ok(Arc::new(Self {
             owner,
+            admission_path,
             admission: Mutex::new(Admission {
                 open: true,
                 requests: 0,
             }),
             finished: Notify::new(),
+        }))
+    }
+
+    /// The caller must recheck the canonical current terminal attempt while holding
+    /// this guard, then persist its fence before releasing it. No resource is created.
+    pub(crate) fn fence_retained(
+        database: &Path,
+        attempt: &BackendJobAttemptRecord,
+    ) -> Result<JobHostDrain> {
+        if !attempt.state.terminal() {
+            return Err(Error::RepositoryConflict(
+                "Job Host admission is not canonically closed".into(),
+            ));
+        }
+        let owner = JobHostOwner::try_from(attempt)?;
+        let file = open_admission(&admission_path(database, &owner)?, &owner)?;
+        file.lock()?;
+        Ok(JobHostDrain {
+            owner,
+            _admission: file,
         })
     }
 
@@ -89,6 +143,12 @@ impl JobHostEffects {
                 "Job Host execution admission is closed".into(),
             ));
         }
+        let file = open_admission(&self.admission_path, expected)?;
+        // Never block a DB lane behind cleanup's exclusive guard: cleanup needs
+        // that lane to commit its evidence. The canonical terminal row rejects replay.
+        file.try_lock_shared().map_err(|error| {
+            Error::RepositoryConflict(format!("Job Host admission is fenced: {error}"))
+        })?;
         admission.requests = admission
             .requests
             .checked_add(1)
@@ -96,16 +156,24 @@ impl JobHostEffects {
         Ok(JobHostRequestLease {
             request: Arc::new(AdmittedRequest {
                 effects: self.clone(),
+                _admission: file,
             }),
         })
     }
 
-    /// The receipt proves only this live Host owner's accepted operations ended. Its constructor
+    pub(crate) fn close_admission(&self, expected: &JobHostOwner) -> Result<()> {
-    /// is private so a missing map entry, Job terminal state, or timeout cannot create evidence.
+        if &self.owner != expected {
-    pub(crate) async fn close_and_drain(
+            return Err(Error::InvalidInput("Job Host close owner mismatch".into()));
-        self: &Arc<Self>,
+        }
-        expected: &JobHostOwner,
+        self.admission
-    ) -> Result<JobHostDrain> {
+            .lock()
+            .map_err(|_| Error::Store("Job Host admission poisoned".into()))?
+            .open = false;
+        Ok(())
+    }
+
+    #[cfg(test)]
+    async fn close_and_drain(self: &Arc<Self>, expected: &JobHostOwner) -> Result<JobHostDrain> {
         if &self.owner != expected {
             return Err(Error::InvalidInput("Job Host close owner mismatch".into()));
         }
@@ -124,8 +192,11 @@ impl JobHostEffects {
                 .requests
                 == 0
             {
+                let file = open_admission(&self.admission_path, expected)?;
+                file.lock()?;
                 return Ok(JobHostDrain {
                     owner: self.owner.clone(),
+                    _admission: file,
                 });
             }
             notified.await;
@@ -148,6 +219,7 @@ impl DbOperationLifetime for JobHostRequestLease {
 
 struct AdmittedRequest {
     effects: Arc<JobHostEffects>,
+    _admission: File,
 }
 
 impl Drop for AdmittedRequest {
@@ -165,6 +237,7 @@ impl Drop for AdmittedRequest {
 
 pub(crate) struct JobHostDrain {
     owner: JobHostOwner,
+    _admission: File,
 }
 
 impl JobHostDrain {
@@ -173,6 +246,41 @@ impl JobHostDrain {
     }
 }
 
+fn admission_path(database: &Path, owner: &JobHostOwner) -> Result<PathBuf> {
+    let identity = serde_json::to_vec(owner).map_err(|error| Error::Store(error.to_string()))?;
+    let key: String = Sha256::digest(identity)
+        .iter()
+        .map(|byte| format!("{byte:02x}"))
+        .collect();
+    let mut root = database.as_os_str().to_owned();
+    root.push(".job-host-admission");
+    Ok(PathBuf::from(root).join(key))
+}
+
+fn open_admission(path: &Path, expected: &JobHostOwner) -> Result<File> {
+    // Recovery must never create a missing predecessor's admission resource.
+    let mut file = OpenOptions::new()
+        .read(true)
+        .write(true)
+        .open(path)
+        .map_err(|error| {
+            Error::RepositoryConflict(format!(
+                "Original Job Host admission resource unavailable: {error}"
+            ))
+        })?;
+    let mut bytes = Vec::new();
+    (&mut file).take(8193).read_to_end(&mut bytes)?;
+    let saved: JobHostOwner = serde_json::from_slice(&bytes).map_err(|_| {
+        Error::RepositoryConflict("Original Job Host admission identity invalid".into())
+    })?;
+    if bytes.len() > 8192 || &saved != expected {
+        return Err(Error::RepositoryConflict(
+            "Original Job Host admission identity changed".into(),
+        ));
+    }
+    Ok(file)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -193,7 +301,12 @@ mod tests {
     #[tokio::test]
     async fn cancelled_waiter_cannot_release_accepted_host_effect_or_admit_after_close() {
         let expected = owner("attempt-1");
-        let effects = JobHostEffects::initially_admit(expected.clone());
+        let directory = tempfile::tempdir().unwrap();
+        let effects = JobHostEffects::initially_admit(
+            expected.clone(),
+            &directory.path().join("store.sqlite"),
+        )
+        .unwrap();
         let lease = effects.admit(&expected).unwrap();
         let executor = DbExecutor::new(DbExecutorLimits {
             workers: NonZeroUsize::new(1).unwrap(),
@@ -234,7 +347,12 @@ mod tests {
     #[tokio::test]
     async fn stale_close_cannot_revoke_a_successor_owner_or_create_its_drain_receipt() {
         let current = owner("attempt-2");
-        let effects = JobHostEffects::initially_admit(current.clone());
+        let directory = tempfile::tempdir().unwrap();
+        let effects = JobHostEffects::initially_admit(
+            current.clone(),
+            &directory.path().join("store.sqlite"),
+        )
+        .unwrap();
         assert!(effects.close_and_drain(&owner("attempt-1")).await.is_err());
         let request = effects.admit(&current).unwrap();
         let child = request.clone();

crates/workspace-server/src/backend_job_recovery_store.rs

@@ -301,6 +301,10 @@ fn validate_fence(
         || ended.execution_id.is_nil()
         || ended.shutdown_command_id.is_none()
         || ended.worker_ref.worker_id.to_string() != owner.worker.worker_id
+        || !ended
+            .submissions
+            .iter()
+            .any(|submission| submission.submission_request_id == owner.runtime_run_id)
     {
         return Err(Error::InvalidInput("Controller/Feature/background and Host drain must confirm the exact owner independently".into()));
     }
@@ -1070,7 +1074,11 @@ mod tests {
             "scope":"controller", "worker_ref":{"worker_id":owner.worker.as_ref().unwrap().worker_id},
             "stop_operation_id":"019a4ea8-0000-7000-8000-000000000001",
             "execution_id":"019a4ea8-0000-7000-8000-000000000002",
-            "shutdown_command_id":1,"submissions":[],"feature_failures":[]
+            "shutdown_command_id":1,"submissions":[{
+                "submission_request_id":owner.runtime_run_id,
+                "submission_id":"accepted-submission",
+                "disposition":"started"
+            }],"feature_failures":[]
         });
         (
             controller.to_string(),
@@ -1103,6 +1111,31 @@ mod tests {
                 .record_backend_job_execution_fence("workspace", &owner, &controller, &wrong, NOW)
                 .is_err()
         );
+        let mut wrong_run: serde_json::Value = serde_json::from_str(&controller).unwrap();
+        wrong_run["submissions"][0]["submission_request_id"] = "successor-run".into();
+        assert!(
+            store
+                .record_backend_job_execution_fence(
+                    "workspace",
+                    &owner,
+                    &wrong_run.to_string(),
+                    &host,
+                    NOW
+                )
+                .is_err()
+        );
+        wrong_run["submissions"] = serde_json::json!([]);
+        assert!(
+            store
+                .record_backend_job_execution_fence(
+                    "workspace",
+                    &owner,
+                    &wrong_run.to_string(),
+                    &host,
+                    NOW
+                )
+                .is_err()
+        );
         let bad = controller.replace("controller", "unpublished_startup_candidate");
         assert!(
             store

crates/workspace-server/src/retention.rs

@@ -310,6 +310,26 @@ impl SqliteWorkspaceStore {
         &self,
         req: &WorkerRemovalPlanRequest,
         inv: &WorkerRetentionInventory,
+    ) -> Result<WorkerRemovalPlan, WorkerRetentionError> {
+        self.plan_worker_removal_inner(req, inv, false)
+    }
+
+    /// An automatic Job cleanup may create its first canonical child, but cannot
+    /// replace one hidden by authority/staleness filters after an uncertain effect.
+    /// The predecessor check and first-plan save share the canonical writer Tx.
+    pub(crate) fn plan_job_cleanup_worker_removal(
+        &self,
+        req: &WorkerRemovalPlanRequest,
+        inv: &WorkerRetentionInventory,
+    ) -> Result<WorkerRemovalPlan, WorkerRetentionError> {
+        self.plan_worker_removal_inner(req, inv, true)
+    }
+
+    fn plan_worker_removal_inner(
+        &self,
+        req: &WorkerRemovalPlanRequest,
+        inv: &WorkerRetentionInventory,
+        first_job_cleanup_child: bool,
     ) -> Result<WorkerRemovalPlan, WorkerRetentionError> {
         validate_plan(req, inv)?;
         if req
@@ -325,6 +345,15 @@ impl SqliteWorkspaceStore {
         let now = Utc::now().to_rfc3339();
         self.with_conn_mut(|conn| {
             let tx=conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
+            if first_job_cleanup_child {
+                let prior: Option<String> = tx.query_row(
+                    "SELECT plan_id FROM worker_removal_operations WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 ORDER BY created_at,operation_id LIMIT 1",
+                    params![req.workspace_id,req.worker.runtime_id,req.worker.worker_id], |row| row.get(0),
+                ).optional()?;
+                if let Some(plan_id) = prior {
+                    return Err(StoreError::InvalidInput(format!("stale:{plan_id}:Job cleanup must reconcile its existing canonical removal; a filtered child is not absence")));
+                }
+            }
             let policy=load_removal_policy(&tx,&req.workspace_id,&req.worker,erased_journal,true)?.ok_or_else(|| StoreError::InvalidInput(format!("policy-missing:{}",req.workspace_id)))?;
             let worker=match load_worker(&tx,&req.workspace_id,&req.worker)? {
                 Some(v)=>v,
@@ -2785,6 +2814,93 @@ mod tests {
         assert_ne!(replacement.plan_id, plan.plan_id);
     }
 
+    #[test]
+    fn automatic_job_cleanup_cannot_replace_a_hidden_or_stale_canonical_child() {
+        for changed in ["policy", "worker"] {
+            let store = setup();
+            let request = req();
+            let original = store
+                .plan_job_cleanup_worker_removal(&request, &inv())
+                .unwrap();
+            store
+                .prepare_worker_removal_execution(
+                    "w",
+                    &original.plan_id,
+                    &original.input_fingerprint,
+                )
+                .unwrap();
+            store
+                .fail_worker_removal(
+                    "w",
+                    &original.operation_id,
+                    &original.input_fingerprint,
+                    "runtime_outcome_unconfirmed",
+                )
+                .unwrap();
+            if changed == "policy" {
+                let policy = store.worker_retention_policy("w").unwrap().unwrap();
+                store
+                    .update_worker_retention_policy(
+                        "w",
+                        &policy.policy_digest,
+                        &WorkerRetentionPolicyUpdate {
+                            policy_id: "changed-after-uncertain-removal".into(),
+                            session_disposition: policy.session_disposition,
+                            metadata_disposition: policy.metadata_disposition,
+                            archive_retention: policy.archive_retention,
+                            diagnostics_disposition: policy.diagnostics_disposition,
+                            diagnostics_retention_seconds: policy.diagnostics_retention_seconds,
+                        },
+                    )
+                    .unwrap();
+                assert!(
+                    store
+                        .prepare_worker_removal_execution(
+                            "w",
+                            &original.plan_id,
+                            &original.input_fingerprint
+                        )
+                        .is_err()
+                );
+            } else {
+                store.with_conn(|connection| {
+                    connection.execute("UPDATE worker_registry SET updated_at='2026-08-24T00:00:01Z' WHERE workspace_id='w' AND runtime_id='r' AND worker_id=?1", [worker_id().to_string()])?;
+                    Ok(())
+                }).unwrap();
+            }
+            assert!(
+                store
+                    .recover_worker_removal_execution("w", &request.worker)
+                    .unwrap()
+                    .is_none()
+            );
+            assert!(
+                store
+                    .plan_job_cleanup_worker_removal(&request, &inv())
+                    .is_err(),
+                "{changed}"
+            );
+            store
+                .with_conn(|connection| {
+                    let count: i64 = connection.query_row(
+                        "SELECT COUNT(*) FROM worker_removal_operations WHERE workspace_id='w'",
+                        [],
+                        |row| row.get(0),
+                    )?;
+                    assert_eq!(
+                        count, 1,
+                        "filtered canonical ownership must not become a new operation"
+                    );
+                    let saved = load_plan(connection, &original.plan_id)?.unwrap();
+                    assert_eq!(saved.operation_id, original.operation_id);
+                    assert_eq!(saved.input_fingerprint, original.input_fingerprint);
+                    assert_ne!(saved.state, WorkerRemovalPlanState::Succeeded);
+                    Ok(())
+                })
+                .unwrap();
+        }
+    }
+
     #[test]
     fn succeeded_worker_removal_recovers_after_registry_purge() {
         let s = setup();

crates/workspace-server/src/server.rs

@@ -2664,7 +2664,7 @@ impl WorkerRemovalService {
         target: &RuntimeWorkerRef,
         reason: &str,
         parent_workspace_deletion_operation_id: Option<&str>,
-        expected_removal: Option<(&str, &str)>,
+        job_cleanup_owner: Option<&Mutex<Option<(String, String)>>>,
     ) -> std::result::Result<worker::WorkspaceResponse, String> {
         let remove_lock = {
             let mut locks = self
@@ -2693,6 +2693,15 @@ impl WorkerRemovalService {
             .store
             .recover_worker_removal_execution(&self.workspace_id, &target)
             .map_err(|_| "Worker removal recovery authority is unavailable".to_string())?;
+        let expected_removal = job_cleanup_owner
+            .map(|owner| {
+                owner
+                    .lock()
+                    .map(|saved| saved.clone())
+                    .map_err(|_| "Job cleanup child owner poisoned".to_string())
+            })
+            .transpose()?
+            .flatten();
         if let Some((operation_id, fingerprint)) = expected_removal {
             if !prepared.as_ref().is_some_and(|saved| {
                 saved.plan.workspace_id == self.workspace_id
@@ -2708,6 +2717,28 @@ impl WorkerRemovalService {
             }
         }
         if let Some(prepared) = prepared {
+            if let Some(owner) = job_cleanup_owner {
+                if prepared.plan.workspace_id != self.workspace_id
+                    || prepared.plan.worker != *target
+                    || prepared.plan.operation_id.is_empty()
+                    || prepared.plan.input_fingerprint.is_empty()
+                    || prepared.runtime_request.workspace_id != self.workspace_id
+                    || prepared.runtime_request.source_runtime_id != target.runtime_id
+                    || prepared.runtime_request.worker_id.to_string() != target.worker_id
+                    || prepared.runtime_request.operation_id != prepared.plan.operation_id
+                    || prepared.runtime_request.input_fingerprint != prepared.plan.input_fingerprint
+                {
+                    return Err("Job cleanup canonical removal binding mismatch".into());
+                }
+                // Retain the first observed canonical child before any external
+                // effect or retry, including children created by this live owner.
+                *owner
+                    .lock()
+                    .map_err(|_| "Job cleanup child owner poisoned".to_string())? = Some((
+                    prepared.plan.operation_id.clone(),
+                    prepared.plan.input_fingerprint.clone(),
+                ));
+            }
             if prepared.plan.state == crate::retention::WorkerRemovalPlanState::Succeeded {
                 self.worker_projection
                     .publish_ordered(|| {
@@ -2835,10 +2866,21 @@ impl WorkerRemovalService {
             reason: reason.to_string(),
             journal_erasure,
         };
-        let plan = match self.store.plan_worker_removal(&request, &inventory) {
+        let plan = match if job_cleanup_owner.is_some() {
+            self.store
+                .plan_job_cleanup_worker_removal(&request, &inventory)
+        } else {
+            self.store.plan_worker_removal(&request, &inventory)
+        } {
             Ok(plan) => plan,
             Err(error) => return Ok(worker_retention_error_response(error)),
         };
+        if let Some(owner) = job_cleanup_owner {
+            *owner
+                .lock()
+                .map_err(|_| "Job cleanup child owner poisoned".to_string())? =
+                Some((plan.operation_id.clone(), plan.input_fingerprint.clone()));
+        }
         let prepared = match self.store.prepare_worker_removal_execution(
             &self.workspace_id,
             &plan.plan_id,
@@ -5129,6 +5171,7 @@ impl WorkspaceApi {
                 worker: worker.worker.clone(),
                 runtime_run_id: runtime_run_id.clone(),
             },
+            &self.config.database_path,
         )?;
         let newly_bound = self.worker_projection.publish_ordered(|| {
             self.store
@@ -5414,7 +5457,8 @@ impl WorkspaceApi {
     async fn cleanup_backend_job_attempt_worker(
         &self,
         attempt: BackendJobAttemptRecord,
-        retained_removal: Option<(String, String)>,
+        removal_owner: &Mutex<Option<(String, String)>>,
+        drain: &crate::backend_job_host_effects::JobHostDrain,
     ) -> Result<()> {
         let history = self
             .store
@@ -5507,7 +5551,7 @@ impl WorkspaceApi {
                 }
                 Err(error) => return fail("backend_job_cleanup_stop_unconfirmed", error.message()),
             }
-            if let Err(error) = self.fence_backend_job_effects(&attempt).await {
+            if let Err(error) = self.fence_backend_job_effects(&attempt, drain).await {
                 return fail("backend_job_cleanup_effects_unconfirmed", error.to_string());
             }
         }
@@ -5518,9 +5562,7 @@ impl WorkspaceApi {
                 &worker,
                 &format!("Terminal Backend Job attempt {attempt_id}"),
                 None,
-                retained_removal
+                Some(removal_owner),
-                    .as_ref()
-                    .map(|(id, fingerprint)| (id.as_str(), fingerprint.as_str())),
             )
             .await
             .map_err(|message| Error::RuntimeOperationFailed {

crates/workspace-server/src/server/backend_job_recovery.rs

@@ -3,7 +3,9 @@
 //! Only initial dispatch opens Host admission; a reconstructed empty map proves nothing.
 use super::*;
 use crate::backend_job::BackendJobRecord;
-use crate::backend_job_host_effects::{JobHostEffects, JobHostOwner, JobHostRequestLease};
+use crate::backend_job_host_effects::{
+    JobHostDrain, JobHostEffects, JobHostOwner, JobHostRequestLease,
+};
 use feature_storage::FeatureStorageError;
 use std::collections::HashSet;
 use std::time::Duration;
@@ -28,6 +30,12 @@ pub(super) struct RecoveryOwners {
     pub(super) nonjournal_lookup_pending: tokio::sync::Notify,
     #[cfg(test)]
     pub(super) cleanup_before_claim: Mutex<Option<Arc<tokio::sync::Semaphore>>>,
+    #[cfg(test)]
+    pub(super) cleanup_before_writer_fence: tokio::sync::Notify,
+    #[cfg(test)]
+    pub(super) cleanup_after_claim: Mutex<Option<Arc<tokio::sync::Semaphore>>>,
+    #[cfg(test)]
+    pub(super) cleanup_claimed_writer: tokio::sync::Notify,
 }
 
 impl RecoveryOwners {
@@ -60,7 +68,11 @@ impl RecoveryOwners {
             .clone()
     }
 
-    pub(super) fn initially_admit(&self, owner: JobHostOwner) -> Result<()> {
+    pub(super) fn initially_admit(
+        &self,
+        owner: JobHostOwner,
+        database: &std::path::Path,
+    ) -> Result<()> {
         let mut hosts = self
             .hosts
             .lock()
@@ -71,7 +83,10 @@ impl RecoveryOwners {
             None => {
                 hosts.insert(
                     owner.attempt_id.clone(),
-                    (owner.clone(), JobHostEffects::initially_admit(owner)),
+                    (
+                        owner.clone(),
+                        JobHostEffects::initially_admit(owner, database)?,
+                    ),
                 );
                 Ok(())
             }
@@ -98,6 +113,19 @@ impl RecoveryOwners {
             .insert(owner.attempt_id.clone()))
     }
 
+    fn close_host_admission(&self, expected: &JobHostOwner) -> Result<()> {
+        let hosts = self
+            .hosts
+            .lock()
+            .map_err(|_| Error::Store("Job Host owners poisoned".into()))?;
+        if let Some((_, effects)) = hosts.get(&expected.attempt_id) {
+            effects.close_admission(expected)?;
+        }
+        // A missing live owner is not evidence. The original retained resource
+        // below must independently fence every accepted Host and cleanup writer.
+        Ok(())
+    }
+
     pub(super) fn effects(&self, expected: &JobHostOwner) -> Result<Arc<JobHostEffects>> {
         self.hosts.lock().map_err(|_| Error::Store("Job Host owners poisoned".into()))?
             .get(&expected.attempt_id).filter(|(owner, _)| owner == expected)
@@ -455,7 +483,6 @@ impl WorkspaceApi {
         &self,
         expected: BackendJobAttemptRecord,
     ) -> Result<()> {
-        let claimed = Arc::new(std::sync::atomic::AtomicBool::new(false));
         // Pin the observed canonical operation across this owner's storage retries.
         let removal_owner = Arc::new(Mutex::new(None::<(String, String)>));
         let mut failures = 0;
@@ -465,13 +492,23 @@ impl WorkspaceApi {
             }
             let api = self.clone();
             let owner = expected.clone();
-            let mine = claimed.clone();
             let removal_owner = removal_owner.clone();
             let runtime = tokio::runtime::Handle::current();
             let result = self
                 .host_executors
                 .cleanup()?
                 .run(move || {
+                    // Every cleanup caller, including its blocking descendants, holds
+                    // the original admission resource exclusively through actual return.
+                    // Recovery therefore fences the predecessor's writers rather than
+                    // treating Executing, an empty map, or process exit as ownership.
+                    api.backend_job_recovery
+                        .close_host_admission(&JobHostOwner::try_from(&owner)?)?;
+                    #[cfg(test)]
+                    api.backend_job_recovery
+                        .cleanup_before_writer_fence
+                        .notify_one();
+                    let drain = JobHostEffects::fence_retained(&api.config.database_path, &owner)?;
                     let current = api
                         .store
                         .get_backend_job_attempt(
@@ -480,7 +517,8 @@ impl WorkspaceApi {
                             &owner.attempt_id,
                         )?
                         .ok_or_else(|| Error::Store("Cleanup owner disappeared".into()))?;
-                    if current.worker != owner.worker
+                    if !current.state.terminal()
+                        || current.worker != owner.worker
                         || current.runtime_run_id != owner.runtime_run_id
                         || current.input_digest != owner.input_digest
                     {
@@ -498,40 +536,9 @@ impl WorkspaceApi {
                     let attempt = if current.worker_cleanup_state
                         == Some(BackendJobWorkerCleanupState::Executing)
                     {
-                        if !mine.load(Ordering::Acquire) {
+                        // Exclusive original-writer fencing authorizes recovery,
-                            // Executing is not a lease we can steal after restart.
+                        // not the Executing marker. Canonical child continuity is
-                            // First require the exact durable execution/Host fence,
+                        // checked under the removal service's own operation lock.
-                            // then recover (never plan) its canonical removal owner.
-                            api.require_backend_job_attempt_effects_ended(&current)?;
-                            let worker = current.worker.as_ref().ok_or_else(||
-                                Error::RepositoryConflict("Executing cleanup Worker missing".into()))?;
-                            let saved = api.config_store.recover_worker_removal_execution(
-                                api.workspace_id(), worker,
-                            ).map_err(|error| Error::Store(error.to_string()))?
-                                .ok_or_else(|| Error::RepositoryConflict(
-                                    "Executing cleanup has no retained canonical removal operation; ownership cannot be reconstructed".into()))?;
-                            if saved.plan.workspace_id != api.workspace_id()
-                                || saved.plan.worker != *worker
-                                || saved.plan.operation_id.is_empty()
-                                || saved.plan.input_fingerprint.is_empty()
-                                || saved.runtime_request.workspace_id != api.workspace_id()
-                                || saved.runtime_request.source_runtime_id != worker.runtime_id
-                                || saved.runtime_request.worker_id.to_string() != worker.worker_id
-                                || saved.runtime_request.operation_id != saved.plan.operation_id
-                                || saved.runtime_request.input_fingerprint != saved.plan.input_fingerprint
-                            {
-                                return Err(Error::RepositoryConflict(
-                                    "Executing cleanup canonical removal owner mismatch".into()));
-                            }
-                            let expected = (saved.plan.operation_id, saved.plan.input_fingerprint);
-                            let mut pinned = removal_owner.lock().map_err(|_| Error::Store(
-                                "Cleanup removal owner poisoned".into()))?;
-                            if pinned.as_ref().is_some_and(|prior| prior != &expected) {
-                                return Err(Error::RepositoryConflict(
-                                    "Executing cleanup canonical removal owner changed".into()));
-                            }
-                            *pinned = Some(expected);
-                        }
                         current
                     } else {
                         let history = api
@@ -549,19 +556,29 @@ impl WorkspaceApi {
                                 "Cleanup belongs to another admitted owner".into(),
                             ));
                         }
-                        mine.store(true, Ordering::Release);
                         attempt
                     };
-                    let retained_removal = removal_owner.lock().map_err(|_| Error::Store(
+                    #[cfg(test)]
-                        "Cleanup removal owner poisoned".into()))?.clone();
+                    let after_claim = api
+                        .backend_job_recovery
+                        .cleanup_after_claim
+                        .lock()
+                        .unwrap()
+                        .clone();
                     runtime.block_on(async {
+                        #[cfg(test)]
+                        if let Some(gate) = after_claim {
+                            api.backend_job_recovery.cleanup_claimed_writer.notify_one();
+                            gate.acquire().await.unwrap().forget();
+                        }
                         if attempt.state == BackendJobAttemptState::Completed
                             || attempt.failure_category.as_deref() == Some("invalid_result")
                         {
                             api.wait_for_backend_job_worker_run_teardown(&attempt)
                                 .await?;
                         }
-                        api.cleanup_backend_job_attempt_worker(attempt, retained_removal).await
+                        api.cleanup_backend_job_attempt_worker(attempt, &removal_owner, &drain)
+                            .await
                     })
                 })
                 .await;
@@ -1231,7 +1248,13 @@ impl WorkspaceApi {
     pub(super) async fn fence_backend_job_effects(
         &self,
         attempt: &BackendJobAttemptRecord,
+        drain: &JobHostDrain,
     ) -> Result<()> {
+        if drain.owner() != &JobHostOwner::try_from(attempt)? {
+            return Err(Error::RepositoryConflict(
+                "Job cleanup admission owner changed".into(),
+            ));
+        }
         if self
             .store
             .has_backend_job_execution_fence(self.workspace_id(), attempt)?
@@ -1252,15 +1275,15 @@ impl WorkspaceApi {
             || termination.scope
                 != worker_runtime::execution::WorkerExecutionTerminationScope::Controller
             || termination.worker_ref.worker_id.to_string() != worker.worker_id
+            || !termination.submissions.iter().any(|submission| {
+                Some(&submission.submission_request_id) == attempt.runtime_run_id.as_ref()
+            })
         {
             return Err(Error::RepositoryConflict(
                 "Runtime receipt does not prove exact Controller/Feature/background termination"
                     .into(),
             ));
         }
-        let owner = JobHostOwner::try_from(attempt)?;
-        let effects = self.backend_job_recovery.effects(&owner)?;
-        let drain = effects.close_and_drain(&owner).await?;
         self.store.record_backend_job_execution_fence(
             self.workspace_id(),
             attempt,
@@ -1321,9 +1344,31 @@ pub(super) async fn admit_host_request(
                 .get_backend_job_attempt(lookup.workspace_id(), &job.request.job_id, &id)?
                 .ok_or_else(|| Error::Store("Job Host attempt missing".into()))?;
             let owner = JobHostOwner::try_from(&attempt)?;
-            Ok(Some(
+            let lease = lookup.backend_job_recovery.effects(&owner)?.admit(&owner)?;
-                lookup.backend_job_recovery.effects(&owner)?.admit(&owner)?,
+            // Close the stale-read window: cleanup's exclusive admission fence and
-            ))
+            // terminal row must preclude every later callback, including another Host.
+            let current_job = lookup
+                .store
+                .get_active_backend_job_for_worker(lookup.workspace_id(), &worker)?
+                .ok_or_else(|| {
+                    Error::WorkspacePermissionDenied(
+                        "Job admission closed before callback acquisition".into(),
+                    )
+                })?;
+            let current = lookup
+                .store
+                .get_backend_job_attempt(lookup.workspace_id(), &job.request.job_id, &id)?
+                .ok_or_else(|| Error::Store("Job Host attempt disappeared".into()))?;
+            if current_job.request.job_id != owner.job_id
+                || current_job.current_attempt != current.attempt
+                || current.state != BackendJobAttemptState::Dispatched
+                || JobHostOwner::try_from(&current)? != owner
+            {
+                return Err(Error::WorkspacePermissionDenied(
+                    "Job callback owner changed during admission".into(),
+                ));
+            }
+            Ok(Some(lease))
         })
         .await;
     match admission {

crates/workspace-server/src/server/tests/backend_job_tests.rs

@@ -194,18 +194,26 @@ async fn backend_job_lost_allocation_response_binds_original_receipt_without_hos
 }
 
 #[tokio::test]
-async fn backend_job_executing_cleanup_restart_requires_fence_and_original_removal() {
+async fn backend_job_executing_cleanup_restart_fences_original_writers_at_each_cut() {
+    for cut in [
+        "claimed",
+        "fenced",
+        "removed",
+        "missing-resource",
+        "wrong-resource",
+    ] {
         let temp = tempfile::tempdir().unwrap();
         let (api, execution) = test_api_with_recording_backend(temp.path()).await;
-    let request = recovery_request("executing-cleanup-restart");
+        let request = recovery_request(&format!("executing-cleanup-{cut}"));
         let dispatched = api.dispatch_backend_job(&request).unwrap();
         let worker = dispatched.attempt.worker.clone().unwrap();
-    execution.take_inputs();
+        assert_eq!(execution.take_inputs().len(), 1);
         seed_worker_session_for_cleanup(temp.path(), &worker);
-    // Stop only the recovery tasks, not execution. Inject a crash boundary after
+        // Quiesce only fixture recovery tasks. The substituted execution boundary
-    // cleanup claim without a live cleanup task. This is not a process-exit test.
+        // and real Host admission resource remain; this is not a process-exit test.
         api.stop_backend_job_recovery().await;
-    api.store
+        let current = api
+            .store
             .finish_backend_job_attempt(
                 api.workspace_id(),
                 &request.job_id,
@@ -216,15 +224,6 @@ async fn backend_job_executing_cleanup_restart_requires_fence_and_original_remov
                 &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true),
             )
             .unwrap();
-    let current = api
-        .store
-        .get_backend_job_attempt(
-            api.workspace_id(),
-            &request.job_id,
-            &dispatched.attempt.attempt_id,
-        )
-        .unwrap()
-        .unwrap();
         let history = api
             .store
             .list_backend_job_attempts(api.workspace_id(), &request.job_id)
@@ -239,16 +238,9 @@ async fn backend_job_executing_cleanup_restart_requires_fence_and_original_remov
             )
             .unwrap();
         assert!(acquired);
-    let mut restarted = api.clone();
-    restarted.backend_job_recovery = Arc::default();
-    let error = restarted
-        .run_backend_job_cleanup(claimed.clone())
-        .await
-        .unwrap_err();
         assert!(
-        error
+            api.require_backend_job_attempt_effects_ended(&claimed)
-            .to_string()
+                .is_err()
-            .contains("Host writer/drain evidence missing")
         );
         assert!(
             api.config_store
@@ -256,7 +248,74 @@ async fn backend_job_executing_cleanup_restart_requires_fence_and_original_remov
                 .unwrap()
                 .is_none()
         );
-
+        // A claimed marker cannot fabricate a missing canonical child.
+        assert!(
+            WorkerRemovalService::new(&api)
+                .execute_target_removal_on_host_checked(
+                    api.runtime.as_ref(),
+                    &worker,
+                    "must not replace missing child",
+                    None,
+                    Some(&Mutex::new(Some((
+                        "missing-operation".into(),
+                        "missing-fingerprint".into()
+                    )))),
+                )
+                .await
+                .is_err()
+        );
+        if matches!(cut, "missing-resource" | "wrong-resource") {
+            let mut path = api.config.database_path.as_os_str().to_owned();
+            path.push(".job-host-admission");
+            let path = std::path::PathBuf::from(path);
+            if cut == "missing-resource" {
+                std::fs::remove_dir_all(&path).unwrap();
+            } else {
+                let file = std::fs::read_dir(&path)
+                    .unwrap()
+                    .next()
+                    .unwrap()
+                    .unwrap()
+                    .path();
+                let mut owner = serde_json::to_value(
+                    crate::backend_job_host_effects::JobHostOwner::try_from(&claimed).unwrap(),
+                )
+                .unwrap();
+                owner["runtime_run_id"] = "successor-run".into();
+                std::fs::write(file, serde_json::to_vec(&owner).unwrap()).unwrap();
+            }
+            let mut restarted = api.clone();
+            restarted.backend_job_recovery = Arc::default();
+            let error = restarted
+                .run_backend_job_cleanup(claimed.clone())
+                .await
+                .unwrap_err();
+            assert!(
+                error.to_string().contains("Original Job Host admission"),
+                "{cut}: {error}"
+            );
+            assert!(
+                api.require_backend_job_attempt_effects_ended(&claimed)
+                    .is_err()
+            );
+            assert!(
+                api.config_store
+                    .recover_worker_removal_execution(api.workspace_id(), &worker)
+                    .unwrap()
+                    .is_none()
+            );
+            assert!(execution.take_inputs().is_empty());
+            if cut == "missing-resource" {
+                assert!(!path.exists());
+            }
+            continue;
+        }
+        let original = if cut != "claimed" {
+            let drain = crate::backend_job_host_effects::JobHostEffects::fence_retained(
+                &api.config.database_path,
+                &claimed,
+            )
+            .unwrap();
             api.runtime
                 .stop_worker(
                     &worker,
@@ -267,54 +326,14 @@ async fn backend_job_executing_cleanup_restart_requires_fence_and_original_remov
                     },
                 )
                 .unwrap();
-    api.fence_backend_job_effects(&claimed).await.unwrap();
+            api.fence_backend_job_effects(&claimed, &drain)
-    let error = restarted
-        .run_backend_job_cleanup(claimed.clone())
                 .await
-        .unwrap_err();
+                .unwrap();
             assert!(
-        error
+                api.require_backend_job_attempt_effects_ended(&claimed)
-            .to_string()
+                    .is_ok()
-            .contains("no retained canonical removal operation")
             );
-    assert!(
+            if cut == "removed" {
-        api.config_store
-            .recover_worker_removal_execution(api.workspace_id(), &worker)
-            .unwrap()
-            .is_none(),
-        "a restarted Executing marker cannot authorize a new removal"
-    );
-    assert_eq!(
-        api.store
-            .get_backend_job_attempt(api.workspace_id(), &request.job_id, &claimed.attempt_id)
-            .unwrap()
-            .unwrap()
-            .worker_cleanup_state,
-        Some(BackendJobWorkerCleanupState::Executing)
-    );
-
-    // Checked restart entry cannot fall through to planning without the pinned
-    // original operation, even though execution/Host fences now exist.
-    assert!(
-        WorkerRemovalService::new(&api)
-            .execute_target_removal_on_host_checked(
-                api.runtime.as_ref(),
-                &worker,
-                "must not create a replacement",
-                None,
-                Some(("missing-operation", "missing-fingerprint"))
-            )
-            .await
-            .is_err()
-    );
-    assert!(
-        api.config_store
-            .recover_worker_removal_execution(api.workspace_id(), &worker)
-            .unwrap()
-            .is_none()
-    );
-
-    // The original owner performs canonical removal, then loses its cleanup save.
                 let response = WorkerRemovalService::new(&api)
                     .execute_target_removal_on_host(
                         api.runtime.as_ref(),
@@ -325,15 +344,17 @@ async fn backend_job_executing_cleanup_restart_requires_fence_and_original_remov
                     .await
                     .unwrap();
                 assert_eq!(response.status, StatusCode::OK.as_u16());
-    let original = api
+            }
-        .config_store
+            api.config_store
                 .recover_worker_removal_execution(api.workspace_id(), &worker)
                 .unwrap()
-        .unwrap();
+        } else {
-    assert_eq!(
+            None
-        original.plan.state,
+        };
-        crate::retention::WorkerRemovalPlanState::Succeeded
+        let mut restarted = api.clone();
-    );
+        restarted.backend_job_recovery = Arc::default();
+        let owner = crate::backend_job_host_effects::JobHostOwner::try_from(&claimed).unwrap();
+        assert!(restarted.backend_job_recovery.effects(&owner).is_err());
         restarted.recover_backend_job_worker_cleanups().unwrap();
         let cleaned = wait_for_backend_job_worker_cleanup(
             &restarted,
@@ -342,20 +363,317 @@ async fn backend_job_executing_cleanup_restart_requires_fence_and_original_remov
             BackendJobWorkerCleanupState::Completed,
         )
         .await;
-    assert_eq!(cleaned.failure_category.as_deref(), Some("fixture_failure"));
+        assert_eq!(
+            cleaned.failure_category.as_deref(),
+            Some("fixture_failure"),
+            "{cut}"
+        );
+        assert_eq!(
+            cleaned.failure_detail.as_deref(),
+            Some("retained original failure")
+        );
+        assert_eq!(cleaned.worker, claimed.worker);
+        assert_eq!(cleaned.runtime_run_id, claimed.runtime_run_id);
+        assert_eq!(cleaned.input_digest, claimed.input_digest);
+        restarted
+            .require_backend_job_attempt_effects_ended(&cleaned)
+            .unwrap();
         let retained = api
             .config_store
             .recover_worker_removal_execution(api.workspace_id(), &worker)
             .unwrap()
             .unwrap();
+        assert_eq!(
+            retained.plan.state,
+            crate::retention::WorkerRemovalPlanState::Succeeded
+        );
+        if let Some(original) = original {
             assert_eq!(retained.plan.operation_id, original.plan.operation_id);
             assert_eq!(
                 retained.plan.input_fingerprint,
                 original.plan.input_fingerprint
             );
+        }
+        assert_eq!(
+            api.store
+                .list_backend_job_attempts(api.workspace_id(), &request.job_id)
+                .unwrap()
+                .len(),
+            1
+        );
         assert!(execution.take_inputs().is_empty());
         assert!(api.runtime.worker(&worker).is_err());
         restarted.stop_backend_job_recovery().await;
+    }
+}
+
+#[tokio::test]
+async fn job_cleanup_policy_staleness_retains_the_first_child_across_retry_and_new_owner() {
+    use crate::retention::{WorkerRemovalPlanRequest, WorkerRetentionPolicyUpdate};
+    let temp = tempfile::tempdir().unwrap();
+    let (api, execution) = test_api_with_recording_backend(temp.path()).await;
+    let request = recovery_request("stale-cleanup-child");
+    let dispatched = api.dispatch_backend_job(&request).unwrap();
+    let worker = dispatched.attempt.worker.clone().unwrap();
+    execution.take_inputs();
+    seed_worker_session_for_cleanup(temp.path(), &worker);
+    api.stop_backend_job_recovery().await;
+    let failed = api
+        .store
+        .finish_backend_job_attempt(
+            api.workspace_id(),
+            &request.job_id,
+            &dispatched.attempt.attempt_id,
+            BackendJobAttemptState::Failed,
+            "fixture_failure",
+            "retained original failure",
+            &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true),
+        )
+        .unwrap();
+    let drain = crate::backend_job_host_effects::JobHostEffects::fence_retained(
+        &api.config.database_path,
+        &failed,
+    )
+    .unwrap();
+    api.runtime
+        .stop_worker(
+            &worker,
+            WorkerLifecycleRequest {
+                stop_intent: None,
+                reason: Some("fixture original scope join".into()),
+                ticket_assignment: None,
+            },
+        )
+        .unwrap();
+    api.fence_backend_job_effects(&failed, &drain)
+        .await
+        .unwrap();
+    let reason = format!("Terminal Backend Job attempt {}", failed.attempt_id);
+    let plan = api
+        .config_store
+        .plan_job_cleanup_worker_removal(
+            &WorkerRemovalPlanRequest {
+                workspace_id: api.workspace_id().into(),
+                worker: worker.clone(),
+                reason: reason.clone(),
+                journal_erasure: None,
+            },
+            &api.runtime.worker_retention_inventory(&worker).unwrap(),
+        )
+        .unwrap();
+    api.config_store
+        .prepare_worker_removal_execution(
+            api.workspace_id(),
+            &plan.plan_id,
+            &plan.input_fingerprint,
+        )
+        .unwrap();
+    // Seed an uncertain canonical child, not a successful provider response.
+    api.config_store
+        .fail_worker_removal(
+            api.workspace_id(),
+            &plan.operation_id,
+            &plan.input_fingerprint,
+            "fixture_runtime_outcome_unconfirmed",
+        )
+        .unwrap();
+    let policy = api
+        .config_store
+        .worker_retention_policy(api.workspace_id())
+        .unwrap()
+        .unwrap();
+    api.config_store
+        .update_worker_retention_policy(
+            api.workspace_id(),
+            &policy.policy_digest,
+            &WorkerRetentionPolicyUpdate {
+                policy_id: "changed-after-uncertain-job-child".into(),
+                session_disposition: policy.session_disposition,
+                metadata_disposition: policy.metadata_disposition,
+                archive_retention: policy.archive_retention,
+                diagnostics_disposition: policy.diagnostics_disposition,
+                diagnostics_retention_seconds: policy.diagnostics_retention_seconds,
+            },
+        )
+        .unwrap();
+    let child = Mutex::new(None);
+    let service = WorkerRemovalService::new(&api);
+    let response = service
+        .execute_target_removal_on_host_checked(
+            api.runtime.as_ref(),
+            &worker,
+            &reason,
+            None,
+            Some(&child),
+        )
+        .await
+        .unwrap();
+    assert_ne!(response.status, StatusCode::OK.as_u16());
+    assert_eq!(
+        *child.lock().unwrap(),
+        Some((plan.operation_id.clone(), plan.input_fingerprint.clone()))
+    );
+    assert!(
+        service
+            .execute_target_removal_on_host_checked(
+                api.runtime.as_ref(),
+                &worker,
+                &reason,
+                None,
+                Some(&child)
+            )
+            .await
+            .is_err()
+    );
+    // A fresh recovery owner's empty memory still cannot treat filtered Stale as
+    // proof that no canonical child was ever created.
+    let cold = Mutex::new(None);
+    let response = service
+        .execute_target_removal_on_host_checked(
+            api.runtime.as_ref(),
+            &worker,
+            &reason,
+            None,
+            Some(&cold),
+        )
+        .await
+        .unwrap();
+    assert_ne!(response.status, StatusCode::OK.as_u16());
+    assert!(cold.lock().unwrap().is_none());
+    api.config_store.with_conn(|connection| {
+        let count: i64 = connection.query_row("SELECT COUNT(*) FROM worker_removal_operations WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3", rusqlite::params![api.workspace_id(), worker.runtime_id, worker.worker_id], |row| row.get(0))?;
+        assert_eq!(count, 1);
+        Ok(())
+    }).unwrap();
+    assert!(api.runtime.worker(&worker).is_ok());
+    assert!(execution.take_inputs().is_empty());
+    assert_eq!(
+        api.store
+            .get_backend_job(api.workspace_id(), &request.job_id)
+            .unwrap()
+            .unwrap()
+            .failure_category
+            .as_deref(),
+        Some("fixture_failure")
+    );
+}
+
+#[tokio::test]
+async fn cancelled_cleanup_waiter_keeps_its_actual_writer_exclusive_across_recovery_owners() {
+    let temp = tempfile::tempdir().unwrap();
+    let (api, execution) = test_api_with_recording_backend(temp.path()).await;
+    let request = recovery_request("cancelled-cleanup-writer");
+    let dispatched = api.dispatch_backend_job(&request).unwrap();
+    let worker = dispatched.attempt.worker.clone().unwrap();
+    assert_eq!(execution.take_inputs().len(), 1);
+    seed_worker_session_for_cleanup(temp.path(), &worker);
+    api.stop_backend_job_recovery().await;
+    let failed = api
+        .store
+        .finish_backend_job_attempt(
+            api.workspace_id(),
+            &request.job_id,
+            &dispatched.attempt.attempt_id,
+            BackendJobAttemptState::Failed,
+            "fixture_failure",
+            "retain original failure",
+            &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true),
+        )
+        .unwrap();
+    let mut original = api.clone();
+    original.backend_job_recovery = Arc::default();
+    let hold = Arc::new(tokio::sync::Semaphore::new(0));
+    *original
+        .backend_job_recovery
+        .cleanup_after_claim
+        .lock()
+        .unwrap() = Some(hold.clone());
+    let original_api = original.clone();
+    let expected = failed.clone();
+    let waiter = tokio::spawn(async move { original_api.run_backend_job_cleanup(expected).await });
+    tokio::time::timeout(
+        std::time::Duration::from_secs(5),
+        original
+            .backend_job_recovery
+            .cleanup_claimed_writer
+            .notified(),
+    )
+    .await
+    .unwrap();
+    waiter.abort();
+    assert!(waiter.await.unwrap_err().is_cancelled());
+    // The real cleanup-lane closure is still gated after claim. Its exclusive
+    // resource must survive the cancelled async response waiter independently.
+    let mut directory = api.config.database_path.as_os_str().to_owned();
+    directory.push(".job-host-admission");
+    let path = std::fs::read_dir(std::path::PathBuf::from(directory))
+        .unwrap()
+        .next()
+        .unwrap()
+        .unwrap()
+        .path();
+    let competing_writer = std::fs::OpenOptions::new()
+        .read(true)
+        .write(true)
+        .open(path)
+        .unwrap();
+    assert!(
+        competing_writer.try_lock().is_err(),
+        "actual predecessor cleanup lost its writer lease"
+    );
+    let mut recovered = api.clone();
+    recovered.backend_job_recovery = Arc::default();
+    recovered.recover_backend_job_worker_cleanups().unwrap();
+    tokio::time::timeout(
+        std::time::Duration::from_secs(5),
+        recovered
+            .backend_job_recovery
+            .cleanup_before_writer_fence
+            .notified(),
+    )
+    .await
+    .unwrap();
+    let pending = api
+        .store
+        .get_backend_job_attempt(api.workspace_id(), &request.job_id, &failed.attempt_id)
+        .unwrap()
+        .unwrap();
+    assert_eq!(
+        pending.worker_cleanup_state,
+        Some(BackendJobWorkerCleanupState::Executing)
+    );
+    assert!(
+        api.require_backend_job_attempt_effects_ended(&pending)
+            .is_err()
+    );
+    assert!(
+        api.config_store
+            .recover_worker_removal_execution(api.workspace_id(), &worker)
+            .unwrap()
+            .is_none()
+    );
+    hold.add_permits(1);
+    let cleaned = wait_for_backend_job_worker_cleanup(
+        &recovered,
+        &request.job_id,
+        &failed.attempt_id,
+        BackendJobWorkerCleanupState::Completed,
+    )
+    .await;
+    assert_eq!(cleaned.worker, failed.worker);
+    assert_eq!(cleaned.runtime_run_id, failed.runtime_run_id);
+    assert_eq!(cleaned.failure_category, failed.failure_category);
+    recovered
+        .require_backend_job_attempt_effects_ended(&cleaned)
+        .unwrap();
+    api.config_store.with_conn(|connection| {
+        let count: i64 = connection.query_row("SELECT COUNT(*) FROM worker_removal_operations WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3", rusqlite::params![api.workspace_id(), worker.runtime_id, worker.worker_id], |row| row.get(0))?;
+        assert_eq!(count, 1, "both admitted cleanup callers must converge on one canonical child");
+        Ok(())
+    }).unwrap();
+    assert!(execution.take_inputs().is_empty());
+    assert!(api.runtime.worker(&worker).is_err());
+    recovered.stop_backend_job_recovery().await;
 }
 
 #[tokio::test]
@@ -417,7 +735,7 @@ async fn backend_job_definitive_input_rejection_is_not_receipt_ambiguity() {
 #[tokio::test]
 async fn backend_job_cancelled_callback_keeps_actual_host_db_effect_owned_until_drain() {
     let temp = tempfile::tempdir().unwrap();
-    let (api, execution) = test_api_with_recording_backend(temp.path()).await;
+    let (api, _execution) = test_api_with_recording_backend(temp.path()).await;
     let request = recovery_request("host-drain");
     let reservation = api.dispatch_backend_job(&request).unwrap();
     let owner =
@@ -489,8 +807,7 @@ async fn backend_job_cancelled_callback_keeps_actual_host_db_effect_owned_until_
     // closure is real and still blocked despite its cancelled async waiter.
     tokio::time::timeout(std::time::Duration::from_secs(5), async {
         loop {
-            if !execution.terminations.lock().unwrap().is_empty() && effects.admit(&owner).is_err()
+            if effects.admit(&owner).is_err() {
-            {
                 break;
             }
             tokio::task::yield_now().await;

crates/workspace-server/src/server/tests/subjektiv_jobs_tests.rs

@@ -476,6 +476,14 @@ async fn subjektiv_job_surface_only_requires_attempt_owned_generation_and_real_o
     );
     let job = resource(&api, &subject.id);
     let attempt = current_attempt(&api, &job);
+    // This test exercises live-grant operations before result reconciliation.
+    // The owner may otherwise complete the ready domain surface during an await.
+    let recovery_gate = api.backend_job_recovery.dispatch_gate(&attempt.attempt_id);
+    let recovery_guard = recovery_gate.lock().unwrap();
+    assert_eq!(
+        current_attempt(&api, &job).state,
+        BackendJobAttemptState::Dispatched
+    );
     let worker = attempt.worker.as_ref().unwrap();
     assert!(
         store
@@ -541,6 +549,7 @@ async fn subjektiv_job_surface_only_requires_attempt_owned_generation_and_real_o
         result: serde_json::json!({"subject_id":subject.id,"candidate_ids":[],"surface":{"availability":"ready","generation_id":generation.id,"memory_fingerprint":generation.memory_fingerprint,"snapshot_id":snapshot.id}}),
     };
     api.accept_backend_job_result(worker, &submission).unwrap();
+    drop(recovery_guard);
     wait_for_backend_job_worker_cleanup(
         &api,
         &job.request.job_id,

crates/workspace-server/src/server/tests/subjektiv_journal_connected_tests.rs

@@ -451,9 +451,10 @@ async fn backend_real_runtime_journal_release_and_next_body_persisted_startup()
         worker_id: old.worker_id.clone(),
         binding_id: old_host.writer_binding_id().unwrap().into(),
     };
-    // Inject a durable result-acceptance gap after real Journal completion. This must
+    // Corrupt only the saved Job projection after real completion, deliberately
-    // reject before writer revocation and confirm cancellation of the actual Controller
+    // leaving its attempt/checkpoint completed. This is a fail-closed consumer
-    // guard; merely returning Err would strand the body and its execution-admission fence.
+    // contrast, not a simulated transaction interruption or a recovery mechanism.
+    // The pre-revoke refusal must cancel the actual Controller guard before retry.
     let recorded = api
         .store
         .list_backend_journal_jobs(TEST_WORKSPACE_ID, &old_attr.session_id, None, 100)
@@ -488,7 +489,7 @@ async fn backend_real_runtime_journal_release_and_next_body_persisted_startup()
         rusqlite::params![TEST_WORKSPACE_ID, uncertain_job], |r| Ok((r.get(0)?, r.get(1)?)),
     ).unwrap();
     conn.execute(
-        "UPDATE backend_jobs SET state='unknown',result_json=NULL,result_digest=NULL WHERE workspace_id=?1 AND job_id=?2",
+        "UPDATE backend_jobs SET state='pending',result_json=NULL,result_digest=NULL WHERE workspace_id=?1 AND job_id=?2",
         rusqlite::params![TEST_WORKSPACE_ID, uncertain_job],
     )
     .unwrap();
@@ -500,7 +501,7 @@ async fn backend_real_runtime_journal_release_and_next_body_persisted_startup()
         serde_json::from_value(incomplete.clone()).unwrap(),
     )
     .unwrap_err();
-    assert!(error.to_string().contains("Unknown"), "{error}");
+    assert!(error.to_string().contains("still pending"), "{error}");
     assert!(store.journal_writer_active(&old_binding).unwrap());
     assert!(
         !api.config_store

crates/workspace-server/src/store.rs

@@ -1616,6 +1616,14 @@ pub trait ControlPlaneStore: Send + Sync + WorkspaceDeletionStore {
         now_seconds: u64,
         consumed_at: &str,
     ) -> Result<bool>;
+    fn plan_job_cleanup_worker_removal(
+        &self,
+        request: &crate::retention::WorkerRemovalPlanRequest,
+        inventory: &worker_runtime::retention::WorkerRetentionInventory,
+    ) -> std::result::Result<
+        crate::retention::WorkerRemovalPlan,
+        crate::retention::WorkerRetentionError,
+    >;
     fn plan_worker_removal(
         &self,
         request: &crate::retention::WorkerRemovalPlanRequest,
@@ -6415,6 +6423,17 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
             .await
     }
 
+    fn plan_job_cleanup_worker_removal(
+        &self,
+        request: &crate::retention::WorkerRemovalPlanRequest,
+        inventory: &worker_runtime::retention::WorkerRetentionInventory,
+    ) -> std::result::Result<
+        crate::retention::WorkerRemovalPlan,
+        crate::retention::WorkerRetentionError,
+    > {
+        SqliteWorkspaceStore::plan_job_cleanup_worker_removal(self, request, inventory)
+    }
+
     fn plan_worker_removal(
         &self,
         request: &crate::retention::WorkerRemovalPlanRequest,