fix(job): reconcile retained owners and diagnose cleanup admission failures

Own all startup recovery pages fairly, consume original scoped removal history, and distinguish ordinary Worker failure from retained cleanup obligations.

Persist exact-owner Host admission preflight diagnostics without inventing execution fences, and retain original Stop identity across uncertain acknowledgements.

Checkpoint only: Runtime normal package passes; standalone selected target closure passes after fixture corrections; Server owner closure is 43/44 with a retained real-Journal convergence failure. Not approved or integrated.

crates/standalone/src/job_store.rs

@@ -366,6 +366,13 @@ impl JobStore {
             .transaction_with_behavior(TransactionBehavior::Immediate)?;
         let snapshot = require_snapshot(&tx, job_id)?;
         check_attempt(&snapshot, attempt_id)?;
+        if snapshot.version_stopped || snapshot.input_disposition != JobInputDisposition::Available
+        {
+            return Err(JobStoreError::InvalidTransition(
+                "stopped or retired input forbids domain reconciliation",
+            ));
+        }
+        snapshot.load_executable_request()?;
         if snapshot.domain_grant.is_none()
             || snapshot.attempt.state != JobAttemptState::Failed
             || !matches!(snapshot.state, JobState::Failed | JobState::Cancelled)
@@ -1263,6 +1270,54 @@ mod tests {
         assert!(reopened.retry("first").is_err());
     }
 
+    #[test]
+    fn granted_reconciliation_rechecks_retired_input_in_its_own_transaction() {
+        let (dir, mut store) = fixture();
+        for (id, erase) in [("live-input", false), ("erased-input", true)] {
+            let reserved = store
+                .reserve_granted(request(id), Some(json!({"domain":"consolidation"})))
+                .unwrap();
+            store.dispatch(id, &reserved.attempt.attempt_id, 8).unwrap();
+            store
+                .finish(
+                    id,
+                    &reserved.attempt.attempt_id,
+                    JobAttemptState::Failed,
+                    Some("provider failure"),
+                )
+                .unwrap();
+            record_runner_cleanup(&mut store, id);
+            let observed = store.get(id).unwrap();
+            assert!(observed.executions_released());
+            if erase {
+                let mut other = JobStore::open(&dir.path().join("jobs.sqlite3")).unwrap();
+                other
+                    .reclaim_input(
+                        id,
+                        &observed.input_digest,
+                        JobInputRetirementReason::ExplicitErasure,
+                    )
+                    .unwrap();
+            }
+            let result = store.reconcile_granted_attempt(
+                id,
+                &observed.attempt.attempt_id,
+                "inspected-original-effects",
+            );
+            assert_eq!(result.is_ok(), !erase);
+            let receipts: i64 = store
+                .conn
+                .query_row(
+                    "SELECT COUNT(*) FROM job_domain_reconciliations WHERE job_id=?1",
+                    [id],
+                    |row| row.get(0),
+                )
+                .unwrap();
+            assert_eq!(receipts, i64::from(!erase));
+            assert_eq!(store.get(id).unwrap().state, JobState::Failed);
+        }
+    }
+
     #[test]
     fn schema_three_cutover_preserves_grant_history_and_execution_fence() {
         let dir = tempfile::tempdir().unwrap();
@@ -2612,6 +2667,7 @@ mod tests {
                 Some("known failure"),
             )
             .unwrap();
+        assert!(!failed.executions_released());
         record_runner_cleanup(&mut store, "atomic");
         let failed = store.get("atomic").unwrap();
         store

crates/standalone/src/jobs.rs

@@ -85,6 +85,10 @@ pub(crate) trait JobDomainProvider: Send + Sync {
     ) -> Result<worker::job::JobFeatureGrant, String>;
 }
 
+type OwnedExecutionReport = Arc<
+    Mutex<Option<Result<worker::job::OwnedInternalJobExecution, worker::job::JobExecutionError>>>,
+>;
+
 struct Execution {
     cancel: watch::Sender<bool>,
     complete: watch::Receiver<Option<Result<(), String>>>,
@@ -95,11 +99,7 @@ struct Execution {
     recovery_issue: Arc<Mutex<Option<JobServiceError>>>,
     task: tokio::task::JoinHandle<Result<(), JobServiceError>>,
     // Retain typed Worker failures and their cleanup resources until the owner can resolve them.
-    _owned: Arc<
-        Mutex<
-            Option<Result<worker::job::OwnedInternalJobExecution, worker::job::JobExecutionError>>,
-        >,
-    >,
+    _owned: OwnedExecutionReport,
 }
 struct Executions {
     closed: bool,
@@ -133,9 +133,30 @@ pub struct StandaloneJobs {
     // Exact restore admissions and typed causes remain Host-owned, even if a
     // dispatcher fails. These are identities, never copies of immutable bodies.
     result_recovery: Arc<Mutex<HashMap<String, Arc<Mutex<Option<JobServiceError>>>>>>,
+    // Process-local typed reports survive successful wait/capacity release.
+    // They are diagnostics, never durable shutdown or replay authority.
+    diagnostics: Arc<Mutex<HashMap<(String, String), OwnedExecutionReport>>>,
 }
 
 impl StandaloneJobs {
+    /// Inspect a retained ordinary Worker failure for one exact attempt in this Host.
+    /// The report survives `wait` and shutdown, but is not restart or cleanup evidence.
+    pub fn with_execution_diagnostic<T>(
+        &self,
+        job_id: &str,
+        attempt_id: &str,
+        inspect: impl FnOnce(&worker::job::OwnedInternalJobExecution) -> T,
+    ) -> Option<T> {
+        let report = self
+            .diagnostics
+            .lock()
+            .expect("Job diagnostics poisoned")
+            .get(&(job_id.to_owned(), attempt_id.to_owned()))
+            .cloned()?;
+        let owned = report.lock().expect("Job owned execution poisoned");
+        owned.as_ref()?.as_ref().ok().map(inspect)
+    }
+
     /// Excludes request/start/result acknowledgement while a Subject Host checks its final
     /// saved obligations and releases ownership. Accepted results alone are insufficient:
     /// every spawned execution must have actually finished successful cleanup.
@@ -178,6 +199,7 @@ impl StandaloneJobs {
             storage_recovered: Arc::default(),
             capacity_changed: Arc::default(),
             result_recovery: Arc::default(),
+            diagnostics: Arc::default(),
         })
     }
 
@@ -635,6 +657,7 @@ impl StandaloneJobs {
         let task_issue = recovery_issue.clone();
         let storage_recovered = self.storage_recovered.clone();
         let capacity_changed = self.capacity_changed.clone();
+        let diagnostics = self.diagnostics.clone();
         let task = tokio::spawn(async move {
             let (sender_tx, mut sender_rx) = tokio::sync::oneshot::channel();
             let run = prepared.run_owned(attempt_id.clone(), sink, move |sender| {
@@ -708,6 +731,19 @@ impl StandaloneJobs {
                 },
             )
             .await;
+            if outcome.is_ok()
+                && retained
+                    .lock()
+                    .expect("Job owned execution poisoned")
+                    .as_ref()
+                    .and_then(|result| result.as_ref().ok())
+                    .is_some_and(|execution| execution.worker_failure().is_some())
+            {
+                diagnostics
+                    .lock()
+                    .expect("Job diagnostics poisoned")
+                    .insert((id.clone(), attempt_id.clone()), retained.clone());
+            }
             complete_tx.send_replace(Some(
                 outcome.as_ref().map(|_| ()).map_err(ToString::to_string),
             ));
@@ -1018,7 +1054,7 @@ fn settle_owned_execution(
         .map_err(|error| JobServiceError::Cleanup(error.to_string()))?;
     if !domain_fenced
         || !execution.feature_shutdown_complete()
-        || execution.worker_failure().is_some()
+        || execution.requires_worker_cleanup_reconciliation()
     {
         return Err(JobServiceError::Cleanup(format!(
             "owned Worker shutdown unconfirmed ({} Feature shutdown failures); typed execution retained",

crates/standalone/src/jobs/recovery_tests.rs

@@ -190,6 +190,13 @@ async fn restored_results_continue_after_wait_error_without_forging_cleanup() {
     let jobs = StandaloneJobs::open(&temp.path().join("jobs.sqlite3"), temp.path().into()).unwrap();
     let failed = jobs.request(request("wait-error")).unwrap();
     let later = jobs.request(request("later-result")).unwrap();
+    let db = rusqlite::Connection::open(temp.path().join("jobs.sqlite3")).unwrap();
+    db.execute_batch(
+        "CREATE TRIGGER reject_cleanup_receipt BEFORE UPDATE OF cleanup_receipt ON job_execution_ownership
+         WHEN NEW.job_id='wait-error' AND NEW.cleanup_receipt IS NOT NULL
+         BEGIN SELECT RAISE(FAIL, 'scripted cleanup receipt rejection'); END;",
+    )
+    .unwrap();
     let ids = jobs.reserved_result_attempts().unwrap();
     let release = Arc::new(Notify::new());
     let _release_on_failure = ReleaseShutdownOnDrop(release.clone());
@@ -206,7 +213,8 @@ async fn restored_results_continue_after_wait_error_without_forging_cleanup() {
     let owner_jobs = jobs.clone();
     let owner = tokio::spawn(async move { owner_jobs.resume_reserved_result_jobs(ids).await });
     // Wait for the later model to enter, then observe its real cleanup. The first
-    // owner still cannot finish; it is held before a known provider failure.
+    // owner is held before provider return and its exact cleanup receipt write
+    // will be rejected independently of that ordinary model failure.
     await_model(&model).await;
     assert_eq!(
         await_job(&jobs, "later-result").await.state,
@@ -233,7 +241,7 @@ async fn restored_results_continue_after_wait_error_without_forging_cleanup() {
     assert!(
         jobs.result_recovery_diagnostic("wait-error")
             .unwrap()
-            .contains("cleanup")
+            .contains("scripted cleanup receipt rejection")
     );
     assert_eq!(model.requests.lock().unwrap().len(), 2);
     assert_eq!(bad_model.requests.lock().unwrap().len(), 2);

crates/standalone/src/jobs/tests.rs

@@ -366,7 +366,7 @@ async fn accepted_result_with_feature_shutdown_failure_retains_typed_owner_and_b
 }
 
 #[tokio::test]
-async fn accepted_result_with_worker_failure_does_not_confirm_execution_cleanup() {
+async fn accepted_result_with_ordinary_worker_failure_confirms_independent_cleanup() {
     let temp = tempfile::tempdir().unwrap();
     let jobs = StandaloneJobs::open(&temp.path().join("jobs.sqlite3"), temp.path().into()).unwrap();
     jobs.request(request("post-result-error")).unwrap();
@@ -380,31 +380,52 @@ async fn accepted_result_with_worker_failure_does_not_confirm_execution_cleanup(
     jobs.start_with_model_client("post-result-error", model)
         .await
         .unwrap();
-    assert!(matches!(
-        tokio::time::timeout(Duration::from_secs(5), jobs.wait("post-result-error"))
-            .await
-            .unwrap(),
-        Err(JobServiceError::Cleanup(_))
-    ));
-    let accepted = jobs.get("post-result-error").unwrap();
-    assert_eq!(accepted.state, JobState::Completed);
-    assert_eq!(accepted.result, Some(json!({"valid":true})));
-    assert!(!accepted.executions_released());
+    let mut completion = jobs.executions.lock().unwrap().jobs["post-result-error"]
+        .complete
+        .clone();
+    completion.wait_for(Option::is_some).await.unwrap();
     {
         let executions = jobs.executions.lock().unwrap();
         let owned = executions.jobs["post-result-error"]._owned.lock().unwrap();
         let owned = owned.as_ref().unwrap().as_ref().unwrap();
         assert!(owned.feature_shutdown_complete());
         assert!(owned.worker_failure().is_some());
+        assert!(!owned.requires_worker_cleanup_reconciliation());
         assert!(owned.result.is_ok());
     }
+    let accepted = tokio::time::timeout(Duration::from_secs(5), jobs.wait("post-result-error"))
+        .await
+        .unwrap()
+        .unwrap();
+    assert_eq!(accepted.state, JobState::Completed);
+    assert_eq!(accepted.result, Some(json!({"valid":true})));
+    assert!(accepted.executions_released());
+    assert_eq!(
+        jobs.with_execution_diagnostic(
+            "post-result-error",
+            &accepted.attempt.attempt_id,
+            |report| {
+                assert_eq!(report.run_id.to_string(), accepted.executions[0].run_id);
+                assert!(report.feature_shutdown_complete());
+                assert!(!report.requires_worker_cleanup_reconciliation());
+                assert!(report.result.is_ok());
+                matches!(
+                    report.worker_failure(),
+                    Some(worker::WorkerError::Engine(_))
+                )
+            }
+        ),
+        Some(true),
+        "typed ordinary failure survives execution-slot release"
+    );
+    assert!(
+        jobs.with_execution_diagnostic("post-result-error", "foreign-attempt", |_| ())
+            .is_none()
+    );
     assert!(jobs.retry("post-result-error").is_err());
-    assert!(jobs.reserve_quiescent_closure().is_err());
-    assert!(matches!(
-        jobs.shutdown().await,
-        Err(JobServiceError::Cleanup(_))
-    ));
-    assert!(jobs.store.lock().unwrap().is_some());
+    assert!(jobs.reserve_quiescent_closure().is_ok());
+    jobs.shutdown().await.unwrap();
+    assert!(jobs.store.lock().unwrap().is_none());
 }
 
 #[tokio::test]

crates/standalone/src/subjektiv.rs

@@ -276,6 +276,7 @@ impl StandaloneSubjects {
             writer,
             self_weak: weak.clone(),
             journal_dispatch: Mutex::new(false),
+            consolidation_recovery: Mutex::new(std::collections::HashMap::new()),
             #[cfg(test)]
             registration_before_attribution: Mutex::new(None),
             language: self.memory_language.clone(),
@@ -372,10 +373,46 @@ impl Drop for SubjectConnection {
     }
 }
 
+#[derive(Debug, thiserror::Error)]
+enum ConsolidationRecoveryError {
+    #[error(transparent)]
+    Host(#[from] SubjektivHostError),
+    #[error(transparent)]
+    Service(#[from] crate::jobs::JobServiceError),
+    #[error(transparent)]
+    Domain(#[from] subjektiv::api::OperationError),
+}
+
+impl ConsolidationRecoveryError {
+    fn retryable_read_or_storage(&self) -> bool {
+        use crate::jobs::{JobServiceError, JobStoreError};
+        matches!(
+            self,
+            Self::Service(JobServiceError::Store(
+                JobStoreError::Sqlite(_) | JobStoreError::Storage(_) | JobStoreError::Io(_)
+            )) | Self::Domain(subjektiv::api::OperationError::Storage(_))
+            // The Host's writer/Subject liveness reads wrap storage errors in
+            // Unavailable. Recheck them within this same finite read-only budget;
+            // their failure NEVER authorizes a receipt, release, or execution.
+            | Self::Host(SubjektivHostError::Unavailable(_))
+        )
+    }
+}
+
+struct ConsolidationRecoveryOwner {
+    running: bool,
+    passes: usize,
+    error: ConsolidationRecoveryError,
+}
+
 pub(crate) struct LocalSubjectHost {
     writer: subjektiv::journal::JournalWriterBinding,
     self_weak: std::sync::Weak<Self>,
     journal_dispatch: Mutex<bool>,
+    // Exact-attempt, process-local bounded recovery owners/diagnostics. The Job
+    // grant and its unchanged input remain the durable obligation, not this map.
+    consolidation_recovery:
+        Mutex<std::collections::HashMap<(String, String, String), ConsolidationRecoveryOwner>>,
     #[cfg(test)]
     registration_before_attribution: Mutex<Option<Box<dyn FnOnce() + Send>>>,
     catalog: StandaloneSubjects,
@@ -767,10 +804,7 @@ impl LocalSubjectHost {
                             schedule_job(jobs.clone(), job.request.job_id, self.self_weak.clone());
                         }
                         Ok(false) => {} // Missing receipts never authorize replay of effects.
-                        Err(error) => eprintln!(
-                            "standalone consolidation Job {} reconciliation blocked: {error}",
-                            job.request.job_id
-                        ),
+                        Err(error) => self.own_consolidation_recovery(jobs.clone(), job, error),
                     }
                 }
             }
@@ -779,24 +813,24 @@ impl LocalSubjectHost {
         self.recover_journals();
     }
 
-    /// Reconstructs a result from domain receipts, never by reentering the finalizer.
-    /// Acceptance does not assert prior owner shutdown or discharge cleanup ownership.
+    /// Prefer a complete saved result. Otherwise close only this failed attempt's
+    /// domain reservation, retaining unresolved work and the original failure.
+    /// Neither path reruns the finalizer or proves prior execution shutdown.
     fn recover_consolidation_job(
         &self,
         jobs: &crate::jobs::StandaloneJobs,
         observed: &crate::jobs::JobSnapshot,
-    ) -> Result<bool, SubjektivHostError> {
-        let service_error = |error: crate::jobs::JobServiceError| {
-            SubjektivHostError::Unavailable(error.to_string())
-        };
+    ) -> Result<bool, ConsolidationRecoveryError> {
         self.ensure_live()?;
-        let current = jobs.get(&observed.request.job_id).map_err(service_error)?;
+        let current = jobs.get(&observed.request.job_id)?;
         if current.attempt.attempt_id != observed.attempt.attempt_id
             || current.input_digest != observed.input_digest
+            || current.domain_grant != observed.domain_grant
         {
             return Err(SubjektivHostError::InvalidArgument(
-                "consolidation recovery attempt/input changed".into(),
-            ));
+                "consolidation recovery attempt/input/grant changed".into(),
+            )
+            .into());
         }
         if current.version_stopped
             || current.input_disposition != crate::jobs::JobInputDisposition::Available
@@ -804,37 +838,156 @@ impl LocalSubjectHost {
         {
             return Err(SubjektivHostError::InvalidArgument(
                 "stopped or retired consolidation cannot be recovered".into(),
-            ));
+            )
+            .into());
         }
         current
             .load_executable_request()
-            .map_err(|error| SubjektivHostError::Unavailable(error.to_string()))?;
+            .map_err(crate::jobs::JobServiceError::from)?;
         let grant = self.consolidation_grant(&current)?;
-        let Some(result) = subjektiv::job::lookup_result(
+        if let Some(result) = subjektiv::job::lookup_result(
             &self.catalog.store,
             &grant.binding.subject_id,
             &grant.candidate_ids,
             &current.request.job_id,
             &current.attempt.attempt_id,
-        )
-        .map_err(operation_error)?
-        else {
-            return Ok(false);
-        };
-        self.ensure_live()?;
-        let settled = jobs
-            .accept_reconciled(&job::JobResultSubmission {
+        )? {
+            self.ensure_live()?;
+            let settled = jobs.accept_reconciled(&job::JobResultSubmission {
                 job_id: current.request.job_id,
                 attempt_id: current.attempt.attempt_id,
                 input_digest: current.input_digest,
                 result,
-            })
-            .map_err(service_error)?;
-        jobs.acknowledge(&settled.request.job_id)
-            .map_err(service_error)?;
+            })?;
+            jobs.acknowledge(&settled.request.job_id)?;
+            return Ok(true);
+        }
+        if current.state != crate::jobs::JobState::Failed
+            || current.attempt.state != crate::jobs::JobAttemptState::Failed
+            || !current.executions_released()
+        {
+            return Ok(false);
+        }
+        // This inspection reads original candidate/disposition and preparation
+        // records transactionally. Missing preparation is valid; stale preparation,
+        // missing candidates, retired Subjects and malformed receipts are not.
+        // Do NOT turn unresolved IDs into dispositions or fail a newer/ready surface.
+        let _retained = subjektiv::job::inspect_abandoned_attempt(
+            &self.catalog.store,
+            &grant.binding.subject_id,
+            &grant.candidate_ids,
+            &current.request.job_id,
+            &current.attempt.attempt_id,
+        )?;
+        // Stable reference to the original immutable grant/input and failed
+        // attempt, not a hash of mutable pending-work projections. Later explicit
+        // work can resolve retained candidates without invalidating exact replay.
+        // The existing Job receipt is the closure; no parallel domain ledger.
+        let reference = job::fingerprint(&(
+            "consolidation-abandoned-v1",
+            &grant.binding,
+            &current.request.job_id,
+            &current.attempt.attempt_id,
+            &current.input_digest,
+        ))
+        .map_err(|error| SubjektivHostError::InvalidArgument(error.to_string()))?;
+        self.ensure_live()?;
+        // Recheck the exact Host/grant before persistence. The service/store also
+        // fence the current attempt and independently require ALL executions released.
+        let checked = jobs.get(&current.request.job_id)?;
+        if checked != current {
+            return Err(SubjektivHostError::InvalidArgument(
+                "consolidation changed during abandoned-attempt inspection".into(),
+            )
+            .into());
+        }
+        self.consolidation_grant(&checked)?;
+        jobs.reconcile_granted_attempt(
+            &current.request.job_id,
+            &current.attempt.attempt_id,
+            &format!("consolidation-abandoned-v1:{reference}"),
+        )?;
         Ok(true)
     }
 
+    /// Keep a finite recovery owner for read/storage failure of this exact
+    /// snapshot. No startup-only repair, disconnect polling, new attempt or model
+    /// call. Exhaustion retains the typed cause and original durable obligation;
+    /// a fresh Host can reconcile it, never infer success from budget exhaustion.
+    fn own_consolidation_recovery(
+        &self,
+        jobs: crate::jobs::StandaloneJobs,
+        observed: crate::jobs::JobSnapshot,
+        error: ConsolidationRecoveryError,
+    ) {
+        eprintln!(
+            "standalone consolidation Job {} reconciliation blocked: {error}",
+            observed.request.job_id
+        );
+        if !error.retryable_read_or_storage() {
+            return;
+        }
+        let key = (
+            observed.request.job_id.clone(),
+            observed.attempt.attempt_id.clone(),
+            observed.input_digest.clone(),
+        );
+        let mut owners = self
+            .consolidation_recovery
+            .lock()
+            .expect("consolidation recovery poisoned");
+        if let Some(owner) = owners.get(&key) {
+            if !owner.running {
+                eprintln!(
+                    "standalone consolidation Job {} retains exhausted reconciliation after {} passes: {}; next action: resolve cause and restore the same Host binding",
+                    key.0, owner.passes, owner.error
+                );
+            }
+            return; // Repeated disconnect cannot refill a running or exhausted budget.
+        }
+        let Some(host) = self.self_weak.upgrade() else {
+            return;
+        };
+        let Ok(runtime) = tokio::runtime::Handle::try_current() else {
+            return;
+        };
+        owners.insert(
+            key.clone(),
+            ConsolidationRecoveryOwner {
+                running: true,
+                passes: 1,
+                error,
+            },
+        );
+        runtime.spawn(async move {
+            const PASSES: usize = 8;
+            for pass in 2..=PASSES {
+                tokio::time::sleep(std::time::Duration::from_millis(100 << (pass - 2).min(5))).await;
+                let result = host.recover_consolidation_job(&jobs, &observed);
+                let mut owners = host.consolidation_recovery.lock().expect("consolidation recovery poisoned");
+                match result {
+                    Ok(_) => {
+                        owners.remove(&key);
+                        drop(owners);
+                        host.wake_journals();
+                        return;
+                    }
+                    Err(error) => {
+                        let retry = error.retryable_read_or_storage() && pass < PASSES;
+                        let owner = owners.get_mut(&key).expect("retained consolidation owner");
+                        owner.passes = pass;
+                        owner.running = retry;
+                        owner.error = error;
+                        if !retry {
+                            eprintln!("standalone consolidation Job {} reconciliation stopped after {} passes: {}; original Failed/input/receipts retained; next action: resolve storage/authority failure and restore the same Host binding", key.0, owner.passes, owner.error);
+                            return;
+                        }
+                    }
+                }
+            }
+        });
+    }
+
     fn consolidation_grant(
         &self,
         snapshot: &crate::jobs::JobSnapshot,
@@ -1596,6 +1749,13 @@ fn schedule_job(
             Ok(result) if result.state == crate::jobs::JobState::Completed => {
                 let _ = jobs.acknowledge(&id);
             }
+            Ok(result) if result.state == crate::jobs::JobState::Failed => {
+                if let Some(host) = host.upgrade() {
+                    if let Err(error) = host.recover_consolidation_job(&jobs, &result) {
+                        host.own_consolidation_recovery(jobs.clone(), result, error);
+                    }
+                }
+            }
             Ok(_) => {}
             Err(error) => eprintln!(
                 "standalone Subject Job {id} cleanup/acknowledgement is unconfirmed: {error}"

crates/standalone/src/subjektiv/tests.rs

@@ -22,39 +22,45 @@ impl Drop for ConsolidationRecoveryFixture {
 
 impl ConsolidationRecoveryFixture {
     fn new() -> Self {
+        Self::with_candidates(&["candidate"])
+    }
+
+    fn with_candidates(candidate_ids: &[&str]) -> Self {
         let temp = tempfile::tempdir().unwrap();
         let catalog = StandaloneSubjects::open(temp.path().join("state")).unwrap();
         let subject = catalog.create("recovery-owner", None).unwrap();
         let body = catalog
             .connect(&subject.id, protocol::WorkerId::now_v7())
             .unwrap();
-        body.host
-            .catalog
-            .store
-            .stage_candidate(subjektiv::SubjectStagingRecord::attach(
-                &subject.id,
-                memory::extract::StagingRecord::from_candidate(
-                    "candidate",
-                    "dedup",
-                    memory::schema::SourceRef {
-                        segment_id: "segment".into(),
-                        range: [1, 1],
-                    },
-                    memory::extract::ExtractedCandidate {
-                        kind: memory::extract::CandidateKind::Lesson,
-                        claim: "claim".into(),
-                        why_useful: "reason".into(),
-                        staleness: None,
-                        evidence_ids: vec![],
-                    },
-                    vec![],
-                    vec![],
-                ),
-            ))
-            .unwrap();
+        for candidate_id in candidate_ids {
+            body.host
+                .catalog
+                .store
+                .stage_candidate(subjektiv::SubjectStagingRecord::attach(
+                    &subject.id,
+                    memory::extract::StagingRecord::from_candidate(
+                        *candidate_id,
+                        *candidate_id,
+                        memory::schema::SourceRef {
+                            segment_id: "segment".into(),
+                            range: [1, 1],
+                        },
+                        memory::extract::ExtractedCandidate {
+                            kind: memory::extract::CandidateKind::Lesson,
+                            claim: "claim".into(),
+                            why_useful: "reason".into(),
+                            staleness: None,
+                            evidence_ids: vec![],
+                        },
+                        vec![],
+                        vec![],
+                    ),
+                ))
+                .unwrap();
+        }
         let grant = LocalConsolidationGrant {
             binding: body.binding.clone(),
-            candidate_ids: vec!["candidate".into()],
+            candidate_ids: candidate_ids.iter().map(|id| (*id).to_owned()).collect(),
         };
         let request = job::JobRequest {
             job_id: "consolidation-recovery".into(),
@@ -93,6 +99,19 @@ impl ConsolidationRecoveryFixture {
         }
     }
 
+    fn domain_closure(&self) -> Option<String> {
+        use rusqlite::OptionalExtension;
+        rusqlite::Connection::open(&self.body.jobs_path)
+            .unwrap()
+            .query_row(
+                "SELECT domain_receipt FROM job_domain_reconciliations WHERE job_id=?1 AND attempt_id=?2",
+                rusqlite::params![self.saved.request.job_id, self.saved.attempt.attempt_id],
+                |row| row.get(0),
+            )
+            .optional()
+            .unwrap()
+    }
+
     fn confirm_execution_joined(&mut self) {
         // The execution boundary is substituted in this domain-consumer fixture;
         // use its exact owner/run to commit the receipt through the real store.
@@ -148,6 +167,244 @@ impl ConsolidationRecoveryFixture {
     }
 }
 
+#[tokio::test]
+async fn failed_consolidation_closes_only_attempt_and_preserves_unresolved_input_and_applied_memory()
+ {
+    for partial_apply in [false, true] {
+        let mut fixture =
+            ConsolidationRecoveryFixture::with_candidates(&["candidate", "unresolved"]);
+        let subject_id = fixture.body.binding.subject_id.clone();
+        let subject = &subject_id;
+        let store = &fixture.body.host.catalog.store;
+        let applied = partial_apply.then(|| {
+            store
+                .decide_candidate(
+                    subject,
+                    subjektiv::CandidateDecisionRequest {
+                        request_id: "applied-before-failure".into(),
+                        candidate_id: "candidate".into(),
+                        reason: "retained applied lesson".into(),
+                        decision: subjektiv::CandidateDecision::Apply {
+                            target: subjektiv::MemoryChangeTarget::Create,
+                            draft: subjektiv::MemoryDraft::active(
+                                memory::extract::CandidateKind::Lesson,
+                                "claim",
+                                "body",
+                                "reason",
+                                "change",
+                            ),
+                        },
+                    },
+                )
+                .unwrap()
+        });
+        // This fixture substitutes only positive execution termination, not the
+        // domain lookup. No model is configured; any accidental execution fails.
+        fixture.confirm_execution_joined();
+        let before = fixture.jobs.get(&fixture.saved.request.job_id).unwrap();
+        let memory = serde_json::to_value(
+            fixture
+                .body
+                .host
+                .catalog
+                .store
+                .list_memories(subject)
+                .unwrap(),
+        )
+        .unwrap();
+        fixture.body.host.recover_jobs();
+        let receipt = fixture
+            .domain_closure()
+            .expect("failed attempt closure persisted");
+        assert!(receipt.starts_with("consolidation-abandoned-v1:"));
+        assert_eq!(fixture.jobs.get(&before.request.job_id).unwrap(), before);
+        assert_eq!(before.state, crate::jobs::JobState::Failed);
+        assert!(before.result.is_none());
+        assert!(fixture.jobs.unreleased_granted().unwrap().is_empty());
+        let retained = subjektiv::job::inspect_abandoned_attempt(
+            &fixture.body.host.catalog.store,
+            subject,
+            &["candidate".into(), "unresolved".into()],
+            &before.request.job_id,
+            &before.attempt.attempt_id,
+        )
+        .unwrap();
+        assert!(
+            retained.prepared_generation_id.is_none(),
+            "no new generation"
+        );
+        assert_eq!(
+            retained.unresolved_candidate_ids,
+            if partial_apply {
+                vec!["unresolved"]
+            } else {
+                vec!["candidate", "unresolved"]
+            }
+        );
+        if let Some(applied) = applied {
+            assert_eq!(retained.disposition_ids, vec![applied.resolution.id]);
+            let saved = fixture
+                .body
+                .host
+                .catalog
+                .store
+                .staging_resolution(subject, "candidate")
+                .unwrap()
+                .unwrap();
+            assert_eq!(saved.affected_memory, applied.resolution.affected_memory);
+        } else {
+            assert!(retained.disposition_ids.is_empty());
+        }
+        assert_eq!(
+            serde_json::to_value(
+                fixture
+                    .body
+                    .host
+                    .catalog
+                    .store
+                    .list_memories(subject)
+                    .unwrap()
+            )
+            .unwrap(),
+            memory
+        );
+        // Stable exact replay; a reference does not mean all candidate work is done.
+        assert!(
+            fixture
+                .body
+                .host
+                .recover_consolidation_job(&fixture.jobs, &before)
+                .unwrap()
+        );
+        assert_eq!(fixture.domain_closure(), Some(receipt));
+        let reopened = crate::job_store::JobStore::open(&fixture.body.jobs_path).unwrap();
+        assert_eq!(reopened.get(&before.request.job_id).unwrap(), before);
+    }
+}
+
+#[tokio::test]
+async fn failed_consolidation_retains_unreleased_stale_inactive_migrated_and_erased_controls() {
+    for case in [
+        "unreleased",
+        "stale",
+        "inactive",
+        "migrated",
+        "erased",
+        "changed_grant",
+    ] {
+        let mut fixture = ConsolidationRecoveryFixture::new();
+        if case != "unreleased" {
+            fixture.confirm_execution_joined();
+        }
+        if case == "stale" {
+            fixture.generation(&fixture.saved.attempt.attempt_id);
+            fixture
+                .body
+                .host
+                .catalog
+                .store
+                .create_memory(
+                    &fixture.body.binding.subject_id,
+                    subjektiv::MemoryDraft::active(
+                        memory::extract::CandidateKind::Lesson,
+                        "new",
+                        "body",
+                        "reason",
+                        "change",
+                    ),
+                )
+                .unwrap();
+        }
+        if case == "inactive" {
+            fixture.body.host.live.store(false, Ordering::Release);
+        }
+        if case == "migrated" {
+            rusqlite::Connection::open(&fixture.body.jobs_path)
+                .unwrap()
+                .execute(
+                    "INSERT INTO job_version_stops(job_id) VALUES (?1)",
+                    [&fixture.saved.request.job_id],
+                )
+                .unwrap();
+        }
+        if case == "erased" {
+            fixture
+                .jobs
+                .reclaim_input(
+                    &fixture.saved.request.job_id,
+                    &fixture.saved.input_digest,
+                    crate::jobs::JobInputRetirementReason::ExplicitErasure,
+                )
+                .unwrap();
+        }
+        let mut observed = fixture.saved.clone();
+        if case == "changed_grant" {
+            observed.domain_grant = None;
+        }
+        let before = fixture.jobs.get(&fixture.saved.request.job_id).unwrap();
+        let outcome = fixture
+            .body
+            .host
+            .recover_consolidation_job(&fixture.jobs, &observed);
+        assert!(!matches!(outcome, Ok(true)), "{case}");
+        assert!(fixture.domain_closure().is_none(), "{case}");
+        assert_eq!(
+            fixture.jobs.get(&fixture.saved.request.job_id).unwrap(),
+            before,
+            "{case}"
+        );
+    }
+}
+
+#[tokio::test]
+async fn failed_consolidation_receipt_storage_error_has_autonomous_exact_attempt_recovery() {
+    let mut fixture = ConsolidationRecoveryFixture::new();
+    fixture.confirm_execution_joined();
+    let before = fixture.saved.clone();
+    let connection = rusqlite::Connection::open(&fixture.body.jobs_path).unwrap();
+    // Fail only the closure write; the read-only domain check and all independent
+    // cleanup proofs are real. A storage error is never a successful closure.
+    connection
+        .execute_batch(
+            "CREATE TRIGGER fail_domain_closure BEFORE INSERT ON job_domain_reconciliations
+        BEGIN SELECT RAISE(FAIL, 'fixture domain receipt storage failure'); END;",
+        )
+        .unwrap();
+    let error = fixture
+        .body
+        .host
+        .recover_consolidation_job(&fixture.jobs, &before)
+        .unwrap_err();
+    assert!(error.retryable_read_or_storage());
+    assert!(fixture.domain_closure().is_none());
+    fixture
+        .body
+        .host
+        .own_consolidation_recovery(fixture.jobs.clone(), before.clone(), error);
+    connection
+        .execute_batch("DROP TRIGGER fail_domain_closure;")
+        .unwrap();
+    // No disconnect, recover_jobs call, restart, new Job, model or finalizer.
+    tokio::time::timeout(std::time::Duration::from_secs(5), async {
+        while fixture.domain_closure().is_none() {
+            tokio::task::yield_now().await;
+        }
+    })
+    .await
+    .unwrap();
+    assert!(fixture.jobs.unreleased_granted().unwrap().is_empty());
+    assert_eq!(fixture.jobs.get(&before.request.job_id).unwrap(), before);
+    assert!(
+        fixture
+            .body
+            .host
+            .consolidation_recovery
+            .lock()
+            .unwrap()
+            .is_empty()
+    );
+}
+
 #[tokio::test]
 async fn consolidation_recovery_accepts_same_attempt_receipts_without_claiming_owner_drain() {
     for availability in ["ready", "failed"] {

crates/standalone/tests/jobs.rs

@@ -580,7 +580,7 @@ async fn process_loss_fails_same_attempt_without_claiming_execution_cleanup() {
 }
 
 #[tokio::test]
-async fn model_failure_retains_unconfirmed_execution_and_forbids_retry() {
+async fn ordinary_model_failure_preserves_cause_and_confirms_independent_cleanup() {
     let temp = tempfile::tempdir().unwrap();
     let (host, _) = host(&temp).await;
     let jobs = host.jobs();
@@ -589,14 +589,14 @@ async fn model_failure_retains_unconfirmed_execution_and_forbids_retry() {
         fail: true,
         ..ScriptedClient::new(Vec::new())
     };
-    jobs.start_with_model_client("model-failure", client)
+    jobs.start_with_model_client("model-failure", client.clone())
         .await
         .unwrap();
-    assert!(matches!(
-        jobs.wait("model-failure").await,
-        Err(JobServiceError::Cleanup(_))
-    ));
-    let failure = jobs.get("model-failure").unwrap();
+    let failure = tokio::time::timeout(Duration::from_secs(10), jobs.wait("model-failure"))
+        .await
+        .unwrap()
+        .unwrap();
+    assert_eq!(jobs.get("model-failure").unwrap(), failure);
     assert_eq!(failure.state, JobState::Failed);
     assert!(failure.result.is_none());
     assert!(
@@ -607,11 +607,40 @@ async fn model_failure_retains_unconfirmed_execution_and_forbids_retry() {
             .unwrap()
             .contains("scripted model failure")
     );
-    assert!(!failure.executions_released());
-    assert!(jobs.retry("model-failure").is_err());
-    assert_eq!(jobs.get("model-failure").unwrap(), failure);
+    assert!(failure.executions_released());
+    assert_eq!(failure.executions.len(), 1);
+    let execution = &failure.executions[0];
+    assert!(execution.writes_revoked);
     assert_eq!(
-        host.shutdown().await,
-        Err(standalone::StandaloneShutdownError::JobCleanup)
+        execution.cleanup_receipt.as_deref(),
+        Some(format!("owned-run:{}", execution.run_id).as_str())
     );
+    assert_eq!(client.requests().len(), 1, "no automatic model retry");
+    assert_eq!(
+        jobs.with_execution_diagnostic("model-failure", &failure.attempt.attempt_id, |report| {
+            assert_eq!(report.run_id.to_string(), execution.run_id);
+            assert!(report.feature_shutdown_complete());
+            assert!(report.worker_failure().is_some());
+            assert!(report.result.is_err());
+            !report.requires_worker_cleanup_reconciliation()
+        }),
+        Some(true)
+    );
+    assert_eq!(jobs.request(request("model-failure")).unwrap(), failure);
+    let retried = jobs.retry("model-failure").unwrap();
+    assert_eq!(retried.attempt.number, 2);
+    assert_ne!(retried.attempt.attempt_id, failure.attempt.attempt_id);
+    assert_eq!(retried.executions, failure.executions);
+    let recovery = success();
+    jobs.start_with_model_client("model-failure", recovery.clone())
+        .await
+        .unwrap();
+    let completed = jobs.wait("model-failure").await.unwrap();
+    assert_eq!(completed.state, JobState::Completed);
+    assert_eq!(completed.executions.len(), 2);
+    assert_eq!(completed.executions[0], failure.executions[0]);
+    assert!(completed.executions_released());
+    assert_eq!(recovery.requests().len(), 2);
+    assert_eq!(client.requests().len(), 1);
+    host.shutdown().await.unwrap();
 }

crates/standalone/tests/subjektiv.rs

@@ -1383,7 +1383,61 @@ async fn transport_loss_after_candidate_apply_preserves_memory_and_later_job_onl
         failed.result.is_none(),
         "partial apply is not an accepted Job result"
     );
+    assert!(
+        failed.executions_released(),
+        "real cleanup is independent of failure"
+    );
+    let requests_for_failed_job = || {
+        losing
+            .requests()
+            .iter()
+            .filter(|request| {
+                request.items.iter().filter_map(Item::as_text).any(|text| {
+                    text.lines()
+                        .any(|line| line == format!("job_id: {first_id}"))
+                })
+            })
+            .count()
+    };
+    let failed_job_requests = requests_for_failed_job();
+    let job_db = rusqlite::Connection::open(
+        state
+            .join("subjektiv/jobs")
+            .join(&subject.id)
+            .join("jobs.sqlite3"),
+    )
+    .unwrap();
+    // Wait for the production post-terminal consumer, NOT disconnect retries or
+    // a manual startup/recovery call. This row is explicitly non-success and the
+    // original Failed snapshot/cause/input remain unchanged.
+    tokio::time::timeout(Duration::from_secs(5), async {
+        loop {
+            use rusqlite::OptionalExtension;
+            let receipt: Option<String> = job_db.query_row(
+                "SELECT domain_receipt FROM job_domain_reconciliations WHERE job_id=?1 AND attempt_id=?2",
+                rusqlite::params![first_id, failed.attempt.attempt_id], |row| row.get(0),
+            ).optional().unwrap();
+            if let Some(receipt) = receipt {
+                assert!(receipt.starts_with("consolidation-abandoned-v1:"));
+                break;
+            }
+            tokio::task::yield_now().await;
+        }
+    }).await.expect("failed domain reconciliation is owned without disconnect/startup");
+    assert_eq!(host.jobs().get(&first_id).unwrap(), failed);
+    assert_eq!(
+        requests_for_failed_job(),
+        failed_job_requests,
+        "no extra model request"
+    );
     disconnect_recorded(&mut host).await;
+    // Disconnect closes this service; compare the durable Failed snapshot after
+    // the next Host reopens the same Subject ledger below.
+    assert_eq!(
+        requests_for_failed_job(),
+        failed_job_requests,
+        "no replay during handoff"
+    );
     host.shutdown().await.unwrap();
     let original = inspect_store(&state, |store| {
         assert!(

crates/subjektiv/src/job.rs

@@ -215,6 +215,93 @@ pub fn lookup_result(
     with_dispositions(store, subject_id, batch, &result).map(Some)
 }
 
+/// A read-only view of work retained by an abandoned attempt, NOT a successful
+/// consolidation result. Receipt IDs point to immutable dispositions; unresolved
+/// IDs remain in the original input/staging store for later explicit work.
+#[derive(Debug, PartialEq, Eq)]
+pub struct AbandonedAttempt {
+    pub disposition_ids: Vec<String>,
+    pub unresolved_candidate_ids: Vec<String>,
+    pub prepared_generation_id: Option<String>,
+}
+
+/// Inspect the original batch and preparation without deciding candidates,
+/// preparing/failing a surface, or calling a model/finalizer. The Host must first
+/// prove the exact failed attempt's execution release and current grant/input.
+/// No preparation is a valid pre-effects cut; an existing but stale preparation
+/// is not. A ready surface (possibly shared by another preparation) is untouched.
+pub fn inspect_abandoned_attempt(
+    store: &SubjektivStore,
+    subject_id: &str,
+    batch: &[String],
+    job_id: &str,
+    attempt_id: &str,
+) -> OperationResult<AbandonedAttempt> {
+    use crate::SubjektivError;
+    use rusqlite::{OptionalExtension, params};
+
+    if batch.len() > CANDIDATE_BATCH_LIMIT
+        || batch
+            .iter()
+            .collect::<std::collections::BTreeSet<_>>()
+            .len()
+            != batch.len()
+    {
+        return Err(OperationError::InvalidInput(
+            "invalid consolidation batch".into(),
+        ));
+    }
+    store.database.try_transaction(|tx| {
+        crate::require_active_subject(tx, subject_id)?;
+        crate::validate_label("Job id", job_id)?;
+        crate::validate_label("attempt id", attempt_id)?;
+        let mut retained = AbandonedAttempt {
+            disposition_ids: Vec::new(),
+            unresolved_candidate_ids: Vec::new(),
+            prepared_generation_id: None,
+        };
+        for id in batch {
+            let raw: Option<String> = tx.query_row(
+                "SELECT record_json FROM staging_records WHERE subject_id=?1 AND candidate_id=?2",
+                params![subject_id, id], |row| row.get(0)
+            ).optional()?;
+            let raw_candidate = raw.ok_or_else(|| SubjektivError::CandidateNotFound(id.clone()))?;
+            let candidate = crate::parse_staging(store.workspace_id(), &raw_candidate)?;
+            if candidate.subject_id != subject_id || candidate.id != *id {
+                return Err(SubjektivError::InvalidRecord("consolidation candidate identity mismatch".into()));
+            }
+            let resolution: Option<String> = tx.query_row(
+                "SELECT resolution_json FROM staging_resolutions WHERE subject_id=?1 AND candidate_id=?2",
+                params![subject_id, id], |row| row.get(0)
+            ).optional()?;
+            if let Some(raw) = resolution {
+                let saved = crate::parse_resolution(&raw)?;
+                if saved.subject_id != subject_id || saved.candidate_id != *id
+                    || !crate::equivalent_staging_payload(&serde_json::to_string(&saved.candidate)?, &raw_candidate)?
+                {
+                    return Err(SubjektivError::InvalidRecord("consolidation disposition identity mismatch".into()));
+                }
+                retained.disposition_ids.push(saved.id);
+            } else {
+                retained.unresolved_candidate_ids.push(id.clone());
+            }
+        }
+        let fingerprint = crate::memory_fingerprint_in_connection(tx, subject_id)?;
+        retained.prepared_generation_id = tx.query_row(
+            "SELECT generation_id FROM surface_generation_runs
+             WHERE subject_id=?1 AND job_id=?2 AND attempt_id=?3 AND memory_fingerprint=?4
+             ORDER BY generation_id LIMIT 1",
+            params![subject_id, job_id, attempt_id, fingerprint], |row| row.get(0)
+        ).optional()?;
+        if retained.prepared_generation_id.is_none() && tx.prepare(
+            "SELECT 1 FROM surface_generation_runs WHERE subject_id=?1 AND job_id=?2 AND attempt_id=?3"
+        )?.exists(params![subject_id, job_id, attempt_id])? {
+            return Err(SubjektivError::InvalidRecord("abandoned consolidation generation input is stale".into()));
+        }
+        Ok(retained)
+    }).map_err(crate::api::subjektiv_store_error)
+}
+
 /// Adds a compact projection of stored dispositions to a validated result.
 /// Reasons, change references, and provenance stay in the receipt store, not duplicated in bounded Job results.
 /// Neither reapplies candidates nor creates a new ledger.

crates/subjektiv/src/job/tests.rs

@@ -6,6 +6,81 @@ use crate::{
 use memory::extract::{CandidateKind, ExtractedCandidate, StagingRecord};
 use memory::schema::SourceRef;
 
+#[test]
+fn abandoned_attempt_inspection_preserves_original_preparation_and_rejects_stale_input() {
+    let temp = tempfile::tempdir().unwrap();
+    let manager = feature_storage::FeatureStorage::new(temp.path());
+    let scope = manager.scope("abandoned-consolidation").unwrap();
+    let registration = SubjektivStore::register(&scope).unwrap();
+    let store = SubjektivStore::open(&scope, &registration).unwrap();
+    let subject = store
+        .create_subject(SubjectRole::new("companion").unwrap())
+        .unwrap();
+    assert_eq!(
+        inspect_abandoned_attempt(&store, &subject.id, &[], "job", "attempt").unwrap(),
+        AbandonedAttempt {
+            disposition_ids: vec![],
+            unresolved_candidate_ids: vec![],
+            prepared_generation_id: None,
+        }
+    );
+    let generation = store
+        .prepare_job_surface_generation(&subject.id, Some(("job", "attempt")))
+        .unwrap();
+    let surface = store.resident_surface(&subject.id).unwrap();
+    for _ in 0..2 {
+        let abandoned =
+            inspect_abandoned_attempt(&store, &subject.id, &[], "job", "attempt").unwrap();
+        assert_eq!(
+            abandoned.prepared_generation_id.as_deref(),
+            Some(generation.id.as_str())
+        );
+        // Inspection never fabricates a terminal surface failure or finalizes it.
+        assert!(
+            lookup_result(&store, &subject.id, &[], "job", "attempt")
+                .unwrap()
+                .is_none()
+        );
+        assert_eq!(store.resident_surface(&subject.id).unwrap(), surface);
+    }
+    let count: i64 = store
+        .database
+        .try_with_connection(|connection| {
+            Ok::<_, crate::SubjektivError>(connection.query_row(
+                "SELECT COUNT(*) FROM surface_generation_runs",
+                [],
+                |row| row.get(0),
+            )?)
+        })
+        .unwrap();
+    assert_eq!(count, 1, "no new generation on abandoned reconciliation");
+    let ready = store
+        .publish_surface_generation(&subject.id, &generation.id, vec![])
+        .unwrap();
+    inspect_abandoned_attempt(&store, &subject.id, &[], "job", "attempt").unwrap();
+    assert_eq!(
+        store.resident_surface(&subject.id).unwrap().snapshot,
+        Some(ready)
+    );
+    assert!(
+        inspect_abandoned_attempt(&store, &subject.id, &["missing".into()], "job", "attempt")
+            .is_err()
+    );
+    store
+        .create_memory(
+            &subject.id,
+            crate::MemoryDraft::active(CandidateKind::Lesson, "new", "body", "reason", "change"),
+        )
+        .unwrap();
+    assert!(inspect_abandoned_attempt(&store, &subject.id, &[], "job", "attempt").is_err());
+    store
+        .begin_subject_deletion(&subject.id, "delete-abandoned-subject")
+        .unwrap();
+    assert!(
+        inspect_abandoned_attempt(&store, &subject.id, &[], "fresh-job", "fresh-attempt").is_err()
+    );
+}
+
 #[test]
 fn prepared_dirty_surface_waits_for_terminal_receipt_and_rejects_stale_input() {
     let temp = tempfile::tempdir().unwrap();

crates/worker-runtime/src/http_server/tests/session_material_tests.rs

@@ -58,6 +58,10 @@ impl Fixture {
         let mut request = task_request("canonical Session fixture");
         request.config_bundle = None;
         request.memory_settings.as_mut().unwrap().workspace_id = scope.workspace_id.clone();
+        request.workspace_api = Some(crate::catalog::WorkspaceApiRef {
+            workspace_id: scope.workspace_id.clone(),
+            base_url: "https://backend.test".into(),
+        });
         let worker = runtime
             .create_worker_scoped(&scope, request)
             .unwrap()
@@ -1265,11 +1269,11 @@ async fn subject_delete_worker_signed_http_requires_delete_scope_without_inventi
         assert_eq!(response.status(), StatusCode::BAD_REQUEST);
         let error: RuntimeHttpErrorResponse = read_json(response).await;
         assert_eq!(error.error.code, "invalid_request");
+        let diagnostic = error.error.message.to_ascii_lowercase();
         assert!(
-            error
-                .error
-                .message
-                .contains("Missing original confirmed Subject shutdown")
+            diagnostic.contains("original") && diagnostic.contains("shutdown"),
+            "expected rejection for absent original shutdown proof, got: {}",
+            error.error.message
         );
         assert!(f.runtime.worker_detail(&f.worker).is_ok());
         assert!(

crates/worker-runtime/src/http_server/tests/subject_job_removal_purge_recovery_tests.rs

@@ -66,28 +66,43 @@ fn subject_job_removal_purge_partial_physical_failure_keeps_fence_and_retry_eras
 fn subject_job_removal_purge_unconfirmed_fence_never_invents_removed_execution_proof() {
     let f = Fixture::new();
     let request = f.removal_purge_request(SessionDisposition::Archive);
-    f.complete_removal(&request);
+    let original_result = f.complete_removal(&request);
     let copies = f.copies();
     let receipt_path = f.receipt_path();
-    let receipt = std::fs::read(&receipt_path).unwrap();
+    let mut legacy_receipt: serde_json::Value =
+        serde_json::from_slice(&std::fs::read(&receipt_path).unwrap()).unwrap();
     let Fixture {
         temp,
         runtime,
         scope,
         store,
+        session,
         ..
     } = f;
     drop(store);
     drop(runtime);
-    // Retained incomplete-proof fixture: deliberately remove the confirmation
-    // after removal. Correct current ordering requires that confirmation before
-    // removal; this is not evidence that such a crash window can occur now.
-    // The old removal result cannot manufacture the missing Subject proof.
+    // Model proofless retained history, not a current crash window. Removing
+    // only the Subject response is insufficient: the original canonical Stop
+    // now survives removal and legitimately permits historical reconciliation.
+    // Remove that proof too, leaving the completed removal request/result intact.
+    assert!(
+        legacy_receipt
+            .as_object_mut()
+            .unwrap()
+            .remove("shutdown")
+            .is_some_and(|shutdown| shutdown.is_object())
+    );
+    let receipt = serde_json::to_vec(&legacy_receipt).unwrap();
+    std::fs::write(&receipt_path, &receipt).unwrap();
     let snapshot_path = temp.path().join("runtime.json");
     let mut snapshot: serde_json::Value =
         serde_json::from_slice(&std::fs::read(&snapshot_path).unwrap()).unwrap();
-    snapshot["subject_delete_worker_fences"][request.deletion.worker_id.to_string()]["response"] =
-        serde_json::Value::Null;
+    let fence =
+        &mut snapshot["subject_delete_worker_fences"][request.deletion.worker_id.to_string()];
+    assert!(fence["response"].is_object());
+    assert!(fence.get("canonical_removal").is_none());
+    fence["response"] = serde_json::Value::Null;
+    let unconfirmed_fence = fence.clone();
     std::fs::write(&snapshot_path, serde_json::to_vec(&snapshot).unwrap()).unwrap();
     let runtime = Runtime::with_fs_store_and_execution_backend(
         crate::FsRuntimeStoreOptions::new(temp.path()).with_runtime_id("runtime-test"),
@@ -102,16 +117,47 @@ fn subject_job_removal_purge_unconfirmed_fence_never_invents_removed_execution_p
     );
     assert_eq!(tree(&temp.path().join("archives")), copies);
     assert_eq!(std::fs::read(&receipt_path).unwrap(), receipt);
+    let removal = serde_json::from_value(serde_json::to_value(&request.removal).unwrap()).unwrap();
+    assert_eq!(
+        runtime.execute_worker_retention(&removal).unwrap(),
+        original_result
+    );
+    assert_eq!(std::fs::read(&receipt_path).unwrap(), receipt);
+    let error = runtime
+        .subject_delete_worker_scoped(&scope, request.deletion.clone())
+        .unwrap_err();
+    let diagnostic = error.to_string().to_ascii_lowercase();
     assert!(
-        runtime
-            .subject_delete_worker_scoped(&scope, request.deletion.clone())
-            .is_err()
+        diagnostic.contains("original") && diagnostic.contains("shutdown"),
+        "expected rejection for absent original shutdown proof, got: {error}"
     );
     assert!(
         runtime
-            .purge_subject_job_removal_scoped(&scope, request)
+            .purge_subject_job_removal_scoped(&scope, request.clone())
             .is_err()
     );
     assert_eq!(tree(&temp.path().join("archives")), copies);
     assert_eq!(std::fs::read(receipt_path).unwrap(), receipt);
+    let snapshot: serde_json::Value =
+        serde_json::from_slice(&std::fs::read(snapshot_path).unwrap()).unwrap();
+    assert_eq!(
+        snapshot["subject_delete_worker_fences"][request.deletion.worker_id.to_string()],
+        unconfirmed_fence,
+        "proofless history must not acquire a Subject response or a fabricated Stop child"
+    );
+    assert!(
+        runtime
+            .worker_detail(&WorkerRef::new(request.deletion.worker_id))
+            .is_err()
+    );
+    assert!(
+        !temp
+            .path()
+            .join("workers")
+            .join(request.deletion.worker_id.to_string())
+            .exists()
+    );
+    runtime_session_retention(temp.path())
+        .ensure_live(session)
+        .unwrap();
 }

crates/worker-runtime/src/http_server/tests/subject_job_removal_purge_tests.rs

@@ -80,6 +80,10 @@ impl Fixture {
         let mut request = task_request("independent Worker");
         request.config_bundle = None;
         request.memory_settings.as_mut().unwrap().workspace_id = self.scope.workspace_id.clone();
+        request.workspace_api = Some(crate::catalog::WorkspaceApiRef {
+            workspace_id: self.scope.workspace_id.clone(),
+            base_url: "https://backend.test".into(),
+        });
         self.runtime
             .create_worker_scoped(&self.scope, request)
             .unwrap()
@@ -145,6 +149,11 @@ fn subject_job_removal_purge_erases_copies_preserves_receipt_and_prior_success_a
         let prior = tree(&f.temp.path().join("archives/workers/prior-archive"));
         assert!(!prior.is_empty());
         let unrelated = f.independent_worker();
+        // Compare settled retained material, not an active Worker whose cold
+        // automatic restore legitimately changes worker.json independently of purge.
+        f.runtime
+            .stop_worker_scoped(&f.scope, &unrelated, None)
+            .unwrap();
         let unrelated_root = f
             .temp
             .path()

crates/worker-runtime/src/retention.rs

@@ -764,6 +764,112 @@ impl FsWorkerRetentionProvider {
         ))
     }
 
+    /// Read, never create or finish, the original canonical removal. The Worker
+    /// operation and Runtime owner locks are held by the caller; the retention
+    /// lock also excludes receipt publication while inspecting history.
+    pub(crate) fn subject_shutdown_removal(
+        &self,
+        deletion: &runtime_api::SubjectDeleteWorkerRequest,
+        runtime_id: &str,
+        expected: Option<(&WorkerRetentionExecutionRequest, &str)>,
+    ) -> Result<
+        (
+            WorkerRetentionExecutionRequest,
+            crate::execution::WorkerExecutionTermination,
+            String,
+        ),
+        RuntimeError,
+    > {
+        let invalid = || {
+            RuntimeError::InvalidRequest(
+                "Missing or mismatched original canonical removal shutdown".into(),
+            )
+        };
+        let lock_path = self.runtime_root.join(RETENTION_LOCK);
+        let lock = OpenOptions::new()
+            .create(true)
+            .read(true)
+            .write(true)
+            .open(&lock_path)
+            .map_err(|source| RuntimeError::StoreIo {
+                operation: "lock Subject removal history",
+                path: lock_path.clone(),
+                source,
+            })?;
+        lock.lock().map_err(|source| RuntimeError::StoreIo {
+            operation: "lock Subject removal history",
+            path: lock_path,
+            source,
+        })?;
+        let operations = self.runtime_root.join("retention").join("operations");
+        let entries = fs::read_dir(&operations).map_err(|_| invalid())?;
+        let mut found = None;
+        for entry in entries {
+            let path = entry.map_err(|_| invalid())?.path();
+            if path.extension().and_then(|value| value.to_str()) != Some("json") {
+                continue;
+            }
+            let receipt = read_operation_receipt(&path)?.ok_or_else(&invalid)?;
+            if receipt.request.worker_id != deletion.worker_id {
+                continue;
+            }
+            // More than one removal of this ID is successor history, not proof
+            // that the original scope covers whichever generation is now gone.
+            if found.is_some() {
+                return Err(invalid());
+            }
+            let request = &receipt.request;
+            // Identity of the immutable child receipt, not another copy of its
+            // shutdown facts in the parent. Detect changed proof on cold retry.
+            let receipt_sha256 =
+                Sha256::digest(serde_json::to_vec(&receipt).map_err(|_| invalid())?)
+                    .iter()
+                    .map(|byte| format!("{byte:02x}"))
+                    .collect::<String>();
+            validate_request(request)?;
+            validate_receipt_request(&receipt, request)?;
+            let shutdown = receipt.shutdown.as_ref().ok_or_else(&invalid)?;
+            let result = &receipt.result;
+            let worker = crate::identity::WorkerRef::new(deletion.worker_id);
+            if path != self.operation_path(&request.operation_id)?
+                || request.workspace_id != deletion.workspace_id
+                || request.source_runtime_id != runtime_id
+                || expected.is_some_and(|(saved, digest)| {
+                    saved != request || digest != receipt_sha256.as_str()
+                })
+                || shutdown.create_fingerprint != deletion.create_fingerprint
+                || !shutdown
+                    .termination
+                    .confirms_scope_shutdown(&worker, shutdown.stop_operation_id)
+                || result.worker_id != request.worker_id
+                || result.session_disposition != request.session_disposition
+                || result.diagnostics_disposition != request.diagnostics_disposition
+                || result.diagnostics_retained
+                    != (request.diagnostics_disposition == DiagnosticsDisposition::Retain)
+                || !result.source_removed
+            {
+                return Err(invalid());
+            }
+            match (&result.archive, request.session_disposition) {
+                (None, SessionDisposition::Purge) => {}
+                (Some(archive), SessionDisposition::Archive)
+                    if archive.workspace_id == request.workspace_id
+                        && archive.source_runtime_id == request.source_runtime_id
+                        && archive.source_worker_id == request.worker_id
+                        && archive.operation_id == request.operation_id
+                        && archive.input_fingerprint == request.input_fingerprint
+                        && Some(&archive.archive_id) == request.archive_id.as_ref() => {}
+                _ => return Err(invalid()),
+            }
+            found = Some((
+                receipt.request.clone(),
+                shutdown.termination.clone(),
+                receipt_sha256,
+            ));
+        }
+        found.ok_or_else(invalid)
+    }
+
     pub(crate) fn recover_after_source_removal(
         &self,
         request: &WorkerRetentionExecutionRequest,
@@ -908,6 +1014,18 @@ impl WorkerRetentionProvider for FsWorkerRetentionProvider {
     fn execute(
         &self,
         request: &WorkerRetentionExecutionRequest,
+    ) -> Result<WorkerRetentionExecutionResult, RuntimeError> {
+        self.execute_with_shutdown(request, None)
+    }
+}
+
+impl FsWorkerRetentionProvider {
+    // Only Runtime's canonical removal caller supplies this evidence, before
+    // deleting its source aggregate. Replays never upgrade a legacy receipt.
+    pub(crate) fn execute_with_shutdown(
+        &self,
+        request: &WorkerRetentionExecutionRequest,
+        shutdown: Option<RetainedWorkerShutdown>,
     ) -> Result<WorkerRetentionExecutionResult, RuntimeError> {
         validate_request(request)?;
         fs::create_dir_all(&self.runtime_root).map_err(|source| RuntimeError::StoreIo {
@@ -938,6 +1056,14 @@ impl WorkerRetentionProvider for FsWorkerRetentionProvider {
             if receipt.result.source_removed {
                 return Ok(receipt.result.clone());
             }
+            if receipt.shutdown.is_some()
+                && self.worker_dir(request.worker_id).is_dir()
+                && receipt.shutdown != shutdown
+            {
+                return Err(RuntimeError::InvalidRequest(
+                    "Canonical removal original shutdown changed".into(),
+                ));
+            }
         }
 
         let worker_dir = self.worker_dir(request.worker_id);
@@ -1026,6 +1152,12 @@ impl WorkerRetentionProvider for FsWorkerRetentionProvider {
             schema_version: OPERATION_SCHEMA_VERSION,
             request: request.clone(),
             result: result.clone(),
+            // The original operation owns this fact. Do not retag a pending
+            // removal with a later Stop, or enrich an older proofless receipt.
+            shutdown: match pending {
+                Some(receipt) => receipt.shutdown,
+                None => shutdown,
+            },
         };
         // Pending receipt makes the delete/final-receipt crash window
         // recoverable without treating an uncommitted archive as completion.
@@ -1079,11 +1211,25 @@ struct CanonicalSessionManifest {
     session_id: String,
 }
 
+/// Evidence destroyed with the canonical aggregate, retained only by its
+/// existing removal operation. This is not a new Stop or an execution ledger.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub(crate) struct RetainedWorkerShutdown {
+    pub(crate) create_fingerprint: String,
+    pub(crate) stop_operation_id: crate::execution::WorkerLifecycleOperationId,
+    pub(crate) termination: crate::execution::WorkerExecutionTermination,
+}
+
 #[derive(Serialize, Deserialize)]
 struct RetentionOperationReceipt {
     schema_version: u32,
     request: WorkerRetentionExecutionRequest,
     result: WorkerRetentionExecutionResult,
+    // Additive read compatibility: an old receipt remains valid retention
+    // history, but cannot establish previously unrecorded Subject shutdown.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    shutdown: Option<RetainedWorkerShutdown>,
 }
 
 fn read_operation_receipt(path: &Path) -> Result<Option<RetentionOperationReceipt>, RuntimeError> {
@@ -2320,6 +2466,7 @@ mod tests {
                 schema_version: OPERATION_SCHEMA_VERSION,
                 request: request.clone(),
                 result: completed.clone(),
+                shutdown: None,
             },
         );
         fs::remove_dir_all(provider.worker_dir(worker)).unwrap();

crates/worker-runtime/src/retention/runtime_v5_tests.rs

@@ -218,6 +218,7 @@ fn completed_v4_archive(
             schema_version: OPERATION_SCHEMA_VERSION,
             request: request.clone(),
             result: completed.clone(),
+            shutdown: None,
         },
     );
     fs::remove_dir_all(provider.worker_dir(worker)).unwrap();

crates/worker-runtime/src/runtime.rs

@@ -6224,7 +6224,59 @@ impl Runtime {
                 "Worker retention cannot remove pending {operation} reconciliation"
             )));
         }
-        let result = provider.execute(request)?;
+        // Preserve only the original canonical Stop, not stopped status, map
+        // absence or a backend receipt for a different execution. The aggregate
+        // is about to disappear; the existing retention operation owns the
+        // historical copy. Missing legacy evidence does not block removal.
+        let shutdown = worker.request.as_ref().and_then(|create| {
+            worker
+                .restore_guard
+                .execution_termination
+                .as_ref()
+                .and_then(|termination| {
+                    termination
+                        .confirms_scope_shutdown(&worker.worker_ref, termination.stop_operation_id)
+                        .then(|| crate::retention::RetainedWorkerShutdown {
+                            create_fingerprint: create.create_fingerprint.clone(),
+                            stop_operation_id: termination.stop_operation_id,
+                            termination: termination.clone(),
+                        })
+                })
+        });
+        let expected = worker.persisted_record();
+        let backend = state.execution_backend.clone();
+        drop(state);
+        if let Some(saved) = shutdown.as_ref() {
+            let current = backend
+                .ok_or_else(|| {
+                    RuntimeError::InvalidRequest(
+                        "Canonical removal shutdown has no execution owner".into(),
+                    )
+                })?
+                .reconcile_termination_receipt(&expected.worker_ref, &saved.termination)
+                .map_err(RuntimeError::InvalidRequest)?;
+            if current.as_ref() != Some(&saved.termination) {
+                return Err(RuntimeError::InvalidRequest(
+                    "Canonical removal shutdown no longer matches execution owner".into(),
+                ));
+            }
+        }
+        let mut state = self.lock()?;
+        state.ensure_running()?;
+        if shutdown.is_some() {
+            let store = state.fs_store().ok_or_else(|| {
+                RuntimeError::InvalidRequest(
+                    "Canonical removal shutdown requires filesystem storage".into(),
+                )
+            })?;
+            let (_, matches) = store.reconcile_worker_record(&expected)?;
+            if !matches {
+                return Err(RuntimeError::InvalidRequest(
+                    "Canonical removal shutdown source changed".into(),
+                ));
+            }
+        }
+        let result = provider.execute_with_shutdown(request, shutdown)?;
         state.workers.remove(&request.worker_id);
         state.persist_runtime_snapshot()?;
         Ok(result)

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

@@ -1,49 +1,98 @@
 use super::*;
 
+struct SubjectOriginalAbsence {
+    // None only for the existing confirmation-before-removal path.
+    removal_shutdown: Option<crate::execution::WorkerExecutionTermination>,
+}
+
 impl RuntimeState {
-    // Absence is usable only after this exact Subject operation confirmed the
-    // original scope and cleanup. A new missing catalog ID is never proof.
+    // Absence needs original scope shutdown evidence, independently of current
+    // resource cleanup. A new missing catalog ID is never proof.
     fn confirmed_subject_delete_originals_absent(
-        &self,
+        &mut self,
         request: &runtime_api::SubjectDeleteWorkerRequest,
-    ) -> Result<bool, RuntimeError> {
+    ) -> Result<Option<SubjectOriginalAbsence>, RuntimeError> {
         if self.workers.contains_key(&request.worker_id) {
-            return Ok(false);
+            return Ok(None);
         }
         if self.unreconciled_worker_writes.contains(&request.worker_id) {
             return Err(RuntimeError::InvalidRequest(
                 "Subject Worker write remains uncertain".into(),
             ));
         }
-        self.subject_delete_worker_fences
-            .get(&request.worker_id)
-            .filter(|fence| fence.request == *request)
-            .and_then(|fence| fence.response.as_ref())
-            .filter(|saved| {
-                saved.workspace_id == request.workspace_id
-                    && saved.worker_id == request.worker_id
-                    && saved.create_fingerprint == request.create_fingerprint
-                    && saved.operation_id == request.operation_id
-            })
-            .ok_or_else(|| {
-                RuntimeError::InvalidRequest("Missing original confirmed Subject shutdown".into())
-            })?;
         #[cfg(feature = "fs-store")]
-        if self
-            .fs_store()
-            .ok_or_else(|| {
-                RuntimeError::InvalidRequest("Subject original absence requires fs storage".into())
-            })?
-            .worker_originals_absent(request.worker_id)?
         {
-            return Ok(true);
+            let fence = self
+                .subject_delete_worker_fences
+                .get(&request.worker_id)
+                .filter(|fence| fence.request == *request)
+                .ok_or_else(|| {
+                    RuntimeError::InvalidRequest("Missing exact Subject fence".into())
+                })?;
+            let store = self.fs_store().ok_or_else(|| {
+                RuntimeError::InvalidRequest("Subject original absence requires fs storage".into())
+            })?;
+            if !store.worker_originals_absent(request.worker_id)? {
+                return Err(RuntimeError::InvalidRequest(
+                    "Subject Worker original material remains".into(),
+                ));
+            }
+            // Preserve confirmation-before-removal replay. A fence confirmed
+            // through retention history must instead reread its original child
+            // every time; its response is not a substitute for that receipt.
+            if fence.canonical_removal.is_none()
+                && fence.response.as_ref().is_some_and(|saved| {
+                    saved.workspace_id == request.workspace_id
+                        && saved.worker_id == request.worker_id
+                        && saved.create_fingerprint == request.create_fingerprint
+                        && saved.operation_id == request.operation_id
+                })
+            {
+                return Ok(Some(SubjectOriginalAbsence {
+                    removal_shutdown: None,
+                }));
+            }
+            let runtime_id = self.runtime_identity.as_deref().ok_or_else(|| {
+                RuntimeError::InvalidRequest("Subject removal Runtime identity missing".into())
+            })?;
+            let (removal, termination, receipt_sha256) =
+                FsWorkerRetentionProvider::new(store.runtime_dir()).subject_shutdown_removal(
+                    request,
+                    runtime_id,
+                    fence
+                        .canonical_removal
+                        .as_ref()
+                        .map(|child| (&child.request, child.receipt_sha256.as_str())),
+                )?;
+            self.subject_delete_worker_fences
+                .get_mut(&request.worker_id)
+                .expect("exact Subject fence held under Worker lock")
+                .canonical_removal = Some(SubjectCanonicalRemoval {
+                request: removal,
+                receipt_sha256,
+            });
+            // Bind the original child before external resource cleanup. Retry
+            // also synchronizes a prior uncertain snapshot acknowledgement.
+            self.persist_runtime_snapshot()?;
+            return Ok(Some(SubjectOriginalAbsence {
+                removal_shutdown: Some(termination),
+            }));
         }
+        #[cfg(not(feature = "fs-store"))]
         Err(RuntimeError::InvalidRequest(
             "Subject Worker original material remains or cannot be inspected".into(),
         ))
     }
 }
 
+#[cfg(feature = "fs-store")]
+#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
+#[serde(deny_unknown_fields)]
+pub(crate) struct SubjectCanonicalRemoval {
+    request: WorkerRetentionExecutionRequest,
+    receipt_sha256: String,
+}
+
 /// Permanent exclusion survives WorkerRemoval. A receipt never substitutes for a
 /// fresh execution reconciliation on exact retry, including after Runtime reopen.
 #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
@@ -51,6 +100,10 @@ impl RuntimeState {
 pub(crate) struct SubjectDeleteWorkerFence {
     pub(crate) request: runtime_api::SubjectDeleteWorkerRequest,
     pub(crate) response: Option<runtime_api::SubjectDeleteWorkerResponse>,
+    // Child identity only; shutdown remains authoritative in the removal receipt.
+    #[cfg(feature = "fs-store")]
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub(crate) canonical_removal: Option<SubjectCanonicalRemoval>,
 }
 
 impl Runtime {
@@ -218,6 +271,8 @@ impl Runtime {
                     SubjectDeleteWorkerFence {
                         request: request.clone(),
                         response: None,
+                        #[cfg(feature = "fs-store")]
+                        canonical_removal: None,
                     },
                 );
             }
@@ -233,17 +288,31 @@ impl Runtime {
         backend
             .fence_subject_delete_worker(&request)
             .map_err(RuntimeError::InvalidRequest)?;
-        if self
-            .lock()?
-            .confirmed_subject_delete_originals_absent(&request)?
-        {
-            // The durable confirmed fence owns the historical shutdown proof;
-            // this separately checks current adapter resources after removal.
+        let absent = {
+            let mut state = self.lock()?;
+            state.confirmed_subject_delete_originals_absent(&request)?
+        };
+        if let Some(absent) = absent {
+            if let Some(saved) = absent.removal_shutdown.as_ref() {
+                let current = backend
+                    .reconcile_termination_receipt(&worker_ref, saved)
+                    .map_err(RuntimeError::InvalidRequest)?;
+                if current.as_ref() != Some(saved) {
+                    return Err(RuntimeError::InvalidRequest(
+                        "Original removal Stop no longer matches execution owner".into(),
+                    ));
+                }
+            }
+            // The durable confirmed fence or original removal owns historical
+            // shutdown proof; separately check current adapter/factory resources.
             backend
                 .confirm_subject_delete_worker_absent(&request)
                 .map_err(RuntimeError::InvalidRequest)?;
             let mut state = self.lock()?;
-            if !state.confirmed_subject_delete_originals_absent(&request)? {
+            if state
+                .confirmed_subject_delete_originals_absent(&request)?
+                .is_none()
+            {
                 return Err(RuntimeError::InvalidRequest(
                     "Subject originals changed during absence reconciliation".into(),
                 ));

crates/worker-runtime/src/subject_delete_backend_tests.rs

@@ -343,7 +343,9 @@ fn subject_delete_without_original_owner_retains_child_identity_but_never_issues
 
 #[test]
 #[cfg(feature = "fs-store")]
-fn subject_delete_cancelled_release_history_consumes_prior_canonical_stop() {
+fn subject_delete_terminal_release_history_consumes_prior_canonical_stop() {
+    use runtime_api::{SubjectReleaseAction as Action, SubjectReleaseState as State};
+    for (action, terminal) in [(Action::Cancel, State::Cancelled), (Action::Disconnect, State::Disconnected)] {
     let temp = tempfile::tempdir_in(".").unwrap();
     let root = fs::canonicalize(temp.path()).unwrap();
     let backend = Arc::new(
@@ -397,20 +399,23 @@ fn subject_delete_cancelled_release_history_consumes_prior_canonical_stop() {
             .stop_worker_scoped(&scope, &worker.worker_ref, None)
             .is_err()
     );
-    release.action = runtime_api::SubjectReleaseAction::Cancel;
+    release.action = action;
     assert_eq!(
         runtime
             .subject_release_scoped(&scope, release.clone())
             .unwrap()
             .state,
-        runtime_api::SubjectReleaseState::Cancelled
+        terminal
     );
     {
         let reservations = backend.subject_reservations.lock().unwrap();
         let saved = &reservations[&(worker.worker_ref.clone(), release.operation_id.clone())];
-        assert_eq!(saved.state, runtime_api::SubjectReleaseState::Cancelled);
+        assert_eq!(saved.state, terminal);
         assert!(saved.guard.is_none());
     }
+    if terminal == State::Disconnected {
+        assert!(runtime.send_input(&worker.worker_ref, WorkerInput::user("must not resume a disconnected body")).is_err());
+    }
     runtime
         .stop_worker_scoped(&scope, &worker.worker_ref, None)
         .unwrap();
@@ -424,7 +429,7 @@ fn subject_delete_cancelled_release_history_consumes_prior_canonical_stop() {
         create_fingerprint: create.create_fingerprint,
         operation_id: "delete-prior-stop".into(),
     };
-    // Retained non-Cancelled state still blocks canonical consumption even if
+    // Retained non-terminal state still blocks canonical consumption even if
     // its local guard is unavailable; only the terminal no-guard row is harmless.
     let key = (worker.worker_ref.clone(), release.operation_id);
     backend
@@ -458,7 +463,7 @@ fn subject_delete_cancelled_release_history_consumes_prior_canonical_stop() {
         .unwrap()
         .get_mut(&key)
         .unwrap()
-        .state = runtime_api::SubjectReleaseState::Cancelled;
+        .state = terminal;
     runtime
         .subject_delete_worker_scoped(&scope, deletion.clone())
         .unwrap();
@@ -478,6 +483,7 @@ fn subject_delete_cancelled_release_history_consumes_prior_canonical_stop() {
     assert!(backend.workers.lock().unwrap().is_empty());
     assert!(backend.pending_spawns.lock().unwrap().is_empty());
     assert!(backend.pending_restores.lock().unwrap().is_empty());
+    }
 }
 
 // Only the extra factory-owned cleanup is faulted. Controller Stop, scope join,
@@ -1217,3 +1223,295 @@ fn subject_delete_confirmed_removal_retries_warm_and_cold_without_new_stop() {
     assert!(cold.pending_restores.lock().unwrap().is_empty());
     assert!(reopened.create_worker_scoped(&scope, create).is_err());
 }
+
+#[test]
+#[cfg(feature = "fs-store")]
+#[serial_test::serial(worker_allocation)]
+fn subject_delete_removal_first_retains_original_stop_and_rechecks_cold_history() {
+    let temp = tempfile::tempdir_in(".").unwrap();
+    let root = fs::canonicalize(temp.path()).unwrap();
+    let spawns = Arc::new(AtomicUsize::new(0));
+    let cleanup_allowed = Arc::new(AtomicBool::new(true));
+    let cleanup_calls = Arc::new(AtomicUsize::new(0));
+    let new_backend = || {
+        Arc::new(
+            WorkerRuntimeExecutionBackend::new(SubjectCleanupFaultFactory {
+                inner: MockFactory {
+                    client: MockClient::new(vec![]),
+                    runtime_base: PathBuf::from(temp.path().file_name().unwrap()).join("c"),
+                    cwd: root.clone(),
+                    store_dir: root.join("sessions"),
+                    worker_metadata_dir: root.join("metadata"),
+                    observed_cwds: Arc::default(),
+                    observed_workspace_clients: Arc::default(),
+                },
+                spawns: spawns.clone(),
+                cleanup_allowed: cleanup_allowed.clone(),
+                cleanup_calls: cleanup_calls.clone(),
+            })
+            .unwrap(),
+        )
+    };
+    let backend = new_backend();
+    let options = crate::fs_store::FsRuntimeStoreOptions::new(root.join("catalog"))
+        .with_runtime_id("removal-first");
+    let runtime =
+        EmbeddedRuntime::with_fs_store_and_execution_backend(options.clone(), backend.clone())
+            .unwrap();
+    runtime.bind_runtime_identity("removal-first").unwrap();
+    runtime.store_config_bundle(test_bundle()).unwrap();
+    let scope = crate::runtime::RuntimeWorkspaceScope::new("ws-test", "server-test");
+    let mut create = create_request("ordinary removal before first Subject confirmation");
+    create.workspace_api = Some(WorkspaceApiRef {
+        workspace_id: scope.workspace_id.clone(),
+        base_url: "https://workspace.invalid".into(),
+    });
+    create.memory_settings = Some(manifest::WorkspaceMemorySettingsSnapshot {
+        workspace_id: scope.workspace_id.clone(),
+        language: "English".into(),
+    });
+    let worker = runtime
+        .create_worker_scoped(&scope, create.clone())
+        .unwrap();
+    let execution = backend.workers.lock().unwrap()[&worker.worker_ref].clone();
+    runtime
+        .stop_worker_scoped(&scope, &worker.worker_ref, None)
+        .unwrap();
+    let original = runtime
+        .execution_termination(&worker.worker_ref)
+        .unwrap()
+        .unwrap();
+    assert_eq!(original.execution_id, execution.tasks.execution_id);
+    assert!(execution.handle.protocol_is_closed());
+    assert!(execution.tasks.tasks.lock().unwrap().is_empty());
+    let removal = crate::retention::WorkerRetentionExecutionRequest {
+        operation_id: "ordinary-removal".into(),
+        input_fingerprint: "ordinary-input".into(),
+        archive_id: None,
+        workspace_id: scope.workspace_id.clone(),
+        source_runtime_id: "removal-first".into(),
+        worker_id: create.worker_id,
+        source_created_at: "2026-11-04T00:00:00Z".into(),
+        removed_at: "2026-11-04T01:00:00Z".into(),
+        effective_profile: None,
+        retention_class: None,
+        policy_id: "ordinary-policy".into(),
+        session_disposition: crate::retention::SessionDisposition::Purge,
+        diagnostics_disposition: crate::retention::DiagnosticsDisposition::Purge,
+    };
+    assert!(
+        runtime
+            .execute_worker_retention(&removal)
+            .unwrap()
+            .source_removed
+    );
+    assert!(runtime.worker_detail(&worker.worker_ref).is_err());
+    let receipt_path = options
+        .root
+        .join("retention/operations/ordinary-removal.json");
+    let receipt_bytes = fs::read(&receipt_path).unwrap();
+    let receipt: serde_json::Value = serde_json::from_slice(&receipt_bytes).unwrap();
+    assert_eq!(
+        receipt["shutdown"]["termination"],
+        serde_json::to_value(&original).unwrap()
+    );
+    let deletion = runtime_api::SubjectDeleteWorkerRequest {
+        workspace_id: scope.workspace_id.clone(),
+        worker_id: create.worker_id,
+        create_fingerprint: create.create_fingerprint.clone(),
+        operation_id: "first-Subject".into(),
+    };
+    // Every failure happens after a real ordinary removal, before any successful
+    // Subject response; an empty adapter/catalog cannot provide the missing fact.
+    let mut bad = receipt.clone();
+    bad.as_object_mut().unwrap().remove("shutdown");
+    fs::write(&receipt_path, serde_json::to_vec(&bad).unwrap()).unwrap();
+    assert!(
+        runtime
+            .execute_worker_retention(&removal)
+            .unwrap()
+            .source_removed
+    );
+    assert!(
+        serde_json::from_slice::<serde_json::Value>(&fs::read(&receipt_path).unwrap())
+            .unwrap()
+            .get("shutdown")
+            .is_none(),
+        "legacy replay must not synthesize shutdown"
+    );
+    assert!(
+        runtime
+            .subject_delete_worker_scoped(&scope, deletion.clone())
+            .is_err()
+    );
+    let mut bad = receipt.clone();
+    bad["shutdown"]["stop_operation_id"] =
+        serde_json::to_value(crate::execution::WorkerLifecycleOperationId::new()).unwrap();
+    fs::write(&receipt_path, serde_json::to_vec(&bad).unwrap()).unwrap();
+    assert!(
+        runtime
+            .subject_delete_worker_scoped(&scope, deletion.clone())
+            .is_err()
+    );
+    let mut bad = receipt.clone();
+    bad["shutdown"]["termination"]["worker_ref"] =
+        serde_json::to_value(WorkerRef::new(WorkerId::now_v7())).unwrap();
+    fs::write(&receipt_path, serde_json::to_vec(&bad).unwrap()).unwrap();
+    assert!(
+        runtime
+            .subject_delete_worker_scoped(&scope, deletion.clone())
+            .is_err()
+    );
+    let mut bad = receipt.clone();
+    bad["request"]["source_runtime_id"] = "other-runtime".into();
+    fs::write(&receipt_path, serde_json::to_vec(&bad).unwrap()).unwrap();
+    assert!(
+        runtime
+            .subject_delete_worker_scoped(&scope, deletion.clone())
+            .is_err()
+    );
+    let mut bad = receipt.clone();
+    bad["request"]["operation_id"] = "other-removal".into();
+    bad["result"]["operation_id"] = "other-removal".into();
+    fs::write(&receipt_path, serde_json::to_vec(&bad).unwrap()).unwrap();
+    assert!(
+        runtime
+            .subject_delete_worker_scoped(&scope, deletion.clone())
+            .is_err()
+    );
+    let mut bad = receipt.clone();
+    bad["request"]["workspace_id"] = "other-workspace".into();
+    fs::write(&receipt_path, serde_json::to_vec(&bad).unwrap()).unwrap();
+    assert!(
+        runtime
+            .subject_delete_worker_scoped(&scope, deletion.clone())
+            .is_err()
+    );
+    let mut bad = receipt.clone();
+    bad["shutdown"]["create_fingerprint"] = "other-create".into();
+    fs::write(&receipt_path, serde_json::to_vec(&bad).unwrap()).unwrap();
+    assert!(
+        runtime
+            .subject_delete_worker_scoped(&scope, deletion.clone())
+            .is_err()
+    );
+    let mut bad = receipt.clone();
+    bad["result"]["source_removed"] = false.into();
+    fs::write(&receipt_path, serde_json::to_vec(&bad).unwrap()).unwrap();
+    assert!(
+        runtime
+            .subject_delete_worker_scoped(&scope, deletion.clone())
+            .is_err()
+    );
+    fs::write(&receipt_path, &receipt_bytes).unwrap();
+    cleanup_allowed.store(false, Ordering::SeqCst);
+    assert!(
+        runtime
+            .subject_delete_worker_scoped(&scope, deletion.clone())
+            .is_err()
+    );
+    cleanup_allowed.store(true, Ordering::SeqCst);
+    let confirmed = runtime
+        .subject_delete_worker_scoped(&scope, deletion.clone())
+        .unwrap();
+    assert!(confirmed.originals_absent);
+    assert_eq!(
+        backend.completed_stops.lock().unwrap()[&worker.worker_ref],
+        original
+    );
+    assert_eq!(fs::read(&receipt_path).unwrap(), receipt_bytes);
+    drop(execution);
+    drop(runtime);
+    drop(backend);
+
+    let cold = new_backend();
+    let reopened =
+        EmbeddedRuntime::with_fs_store_and_execution_backend(options.clone(), cold.clone())
+            .unwrap();
+    reopened.bind_runtime_identity("removal-first").unwrap();
+    // A saved Subject response cannot hide a lost or changed original child.
+    fs::remove_file(&receipt_path).unwrap();
+    assert!(
+        reopened
+            .subject_delete_worker_scoped(&scope, deletion.clone())
+            .is_err()
+    );
+    let mut bad = receipt.clone();
+    bad["request"]["input_fingerprint"] = "changed-input".into();
+    bad["result"]["input_fingerprint"] = "changed-input".into();
+    fs::write(&receipt_path, serde_json::to_vec(&bad).unwrap()).unwrap();
+    assert!(
+        reopened
+            .subject_delete_worker_scoped(&scope, deletion.clone())
+            .is_err()
+    );
+    fs::write(&receipt_path, &receipt_bytes).unwrap();
+    let mut bad = receipt.clone();
+    bad["shutdown"]["termination"]["execution_id"] =
+        serde_json::to_value(uuid::Uuid::now_v7()).unwrap();
+    fs::write(&receipt_path, serde_json::to_vec(&bad).unwrap()).unwrap();
+    assert!(
+        reopened
+            .subject_delete_worker_scoped(&scope, deletion.clone())
+            .is_err(),
+        "cold retry must reread the original whole child receipt, not just request fields"
+    );
+    fs::write(&receipt_path, &receipt_bytes).unwrap();
+    // A later completed Controller is successor evidence even with empty maps.
+    let mut successor = original.clone();
+    successor.execution_id = uuid::Uuid::now_v7();
+    cold.completed_stops
+        .lock()
+        .unwrap()
+        .insert(worker.worker_ref.clone(), successor);
+    assert!(
+        reopened
+            .subject_delete_worker_scoped(&scope, deletion.clone())
+            .is_err()
+    );
+    cold.completed_stops.lock().unwrap().clear();
+    let mut duplicate = receipt.clone();
+    duplicate["request"]["operation_id"] = "later-removal".into();
+    duplicate["result"]["operation_id"] = "later-removal".into();
+    let duplicate_path = options.root.join("retention/operations/later-removal.json");
+    fs::write(&duplicate_path, serde_json::to_vec(&duplicate).unwrap()).unwrap();
+    assert!(
+        reopened
+            .subject_delete_worker_scoped(&scope, deletion.clone())
+            .is_err()
+    );
+    fs::remove_file(duplicate_path).unwrap();
+    cleanup_allowed.store(false, Ordering::SeqCst);
+    assert!(
+        reopened
+            .subject_delete_worker_scoped(&scope, deletion.clone())
+            .is_err()
+    );
+    cleanup_allowed.store(true, Ordering::SeqCst);
+    let material = options
+        .root
+        .join("workers")
+        .join(create.worker_id.to_string());
+    fs::create_dir_all(&material).unwrap();
+    fs::write(material.join("residual"), "still owned original").unwrap();
+    assert!(
+        reopened
+            .subject_delete_worker_scoped(&scope, deletion.clone())
+            .is_err()
+    );
+    fs::remove_dir_all(material).unwrap();
+    assert_eq!(
+        reopened
+            .subject_delete_worker_scoped(&scope, deletion)
+            .unwrap(),
+        confirmed
+    );
+    assert_eq!(spawns.load(Ordering::SeqCst), 1);
+    assert!(cleanup_calls.load(Ordering::SeqCst) > 0);
+    assert!(cold.completed_stops.lock().unwrap().is_empty());
+    assert!(cold.workers.lock().unwrap().is_empty());
+    assert!(cold.pending_spawns.lock().unwrap().is_empty());
+    assert!(cold.pending_restores.lock().unwrap().is_empty());
+    assert_eq!(fs::read(receipt_path).unwrap(), receipt_bytes);
+    assert!(reopened.create_worker_scoped(&scope, create).is_err());
+}

crates/worker-runtime/src/worker_backend.rs

@@ -5442,7 +5442,11 @@ where
                 .any(|((worker, _), reservation)| {
                     worker == worker_ref
                         && (reservation.guard.is_some()
-                            || reservation.state != runtime_api::SubjectReleaseState::Cancelled)
+                            || !matches!(
+                                reservation.state,
+                                runtime_api::SubjectReleaseState::Cancelled
+                                    | runtime_api::SubjectReleaseState::Disconnected
+                            ))
                 })
         {
             return Ok(None);
@@ -6056,7 +6060,11 @@ where
                 .any(|((saved, _), reservation)| {
                     *saved == worker
                         && (reservation.guard.is_some()
-                            || reservation.state != runtime_api::SubjectReleaseState::Cancelled)
+                            || !matches!(
+                                reservation.state,
+                                runtime_api::SubjectReleaseState::Cancelled
+                                    | runtime_api::SubjectReleaseState::Disconnected
+                            ))
                 })
         {
             return Err("Subject Worker still has an execution resource owner".into());
@@ -6213,6 +6221,9 @@ where
                     .as_mut()
                     .ok_or("Controller reservation missing")?
                     .commit_disconnect()?;
+                // The Controller's permanent mutation fence is now committed;
+                // the temporary quiescence lease no longer owns a live handoff.
+                saved.guard.take();
                 saved.state = State::Disconnected;
             }
             (Action::Disconnect, State::Disconnected) | (Action::Cancel, State::Cancelled) => {}

crates/workspace-server/src/backend_job_recovery_store.rs

@@ -189,6 +189,66 @@ pub(super) fn claim_cleanup_owned(
     })
 }
 
+/// A failed Host preflight is diagnostic evidence, never cleanup ownership or
+/// termination proof. In particular, a competing unfenced Executing writer must
+/// remain Executing; only its original admission resource can authorize recovery.
+pub(super) fn record_cleanup_preflight_failure(
+    store: &SqliteWorkspaceStore,
+    workspace: &str,
+    expected_history: &[BackendJobAttemptRecord],
+    expected: &BackendJobAttemptRecord,
+    detail: &str,
+    now: &str,
+) -> Result<BackendJobAttemptRecord> {
+    if detail.trim().is_empty() {
+        return Err(Error::InvalidInput(
+            "cleanup preflight requires a cause detail".into(),
+        ));
+    }
+    store.with_conn_mut(|conn| {
+        let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
+        require_current_contract(&tx, workspace, &expected.job_id)?;
+        require_history(&tx, workspace, expected_history, expected)?;
+        let job = require_retained_owner(&tx, workspace, expected)?;
+        if job.current_attempt != expected.attempt
+            || job.state == BackendJobState::Pending
+            || !expected.state.terminal()
+            || expected.worker.is_none()
+            || expected
+                .runtime_run_id
+                .as_ref()
+                .is_none_or(|run| run.is_empty())
+            || expected.worker_cleanup_completed_at.is_some()
+            || !matches!(
+                expected.worker_cleanup_state,
+                Some(
+                    BackendJobWorkerCleanupState::Pending
+                        | BackendJobWorkerCleanupState::Failed
+                        | BackendJobWorkerCleanupState::Executing
+                )
+            )
+        {
+            return Err(Error::RepositoryConflict(
+                "Cleanup preflight diagnostic owner is no longer eligible".into(),
+            ));
+        }
+        let fenced: bool = tx.query_row(
+            "SELECT EXISTS(SELECT 1 FROM backend_job_execution_fences WHERE workspace_id=?1 AND job_id=?2 AND attempt_id=?3)",
+            params![workspace, expected.job_id, expected.attempt_id], |row| row.get(0),
+        )?;
+        if fenced {
+            return Err(Error::RepositoryConflict(
+                "Cleanup execution fence already recorded; preflight cannot overwrite progress".into(),
+            ));
+        }
+        tx.execute("UPDATE backend_job_attempts SET worker_cleanup_state=CASE WHEN worker_cleanup_state='executing' THEN 'executing' ELSE 'failed' END,worker_cleanup_failure_category='backend_job_cleanup_host_admission_unconfirmed',worker_cleanup_failure_detail=?4,worker_cleanup_updated_at=?5 WHERE workspace_id=?1 AND job_id=?2 AND attempt_id=?3", params![workspace, expected.job_id, expected.attempt_id, bounded_failure_detail(detail), now])?;
+        let result = read_backend_job_attempt(&tx, workspace, &expected.job_id, &expected.attempt_id)?
+            .ok_or_else(|| Error::Store("Cleanup diagnostic owner disappeared".into()))?;
+        tx.commit()?;
+        Ok(result)
+    })
+}
+
 pub(super) fn finish_cleanup_owned(
     store: &SqliteWorkspaceStore,
     workspace: &str,
@@ -1286,6 +1346,349 @@ mod tests {
             .unwrap()
     }
 
+    #[test]
+    fn cleanup_preflight_failure_full_cas_preserves_result_input_and_rolls_back_storage_error() {
+        let (dir, store, reserved) = fixture();
+        let owner = bind_fixture_owner(&store, &reserved);
+        let completed = store
+            .reconcile_backend_job_domain_result(
+                "workspace",
+                &owner,
+                &BackendJobResultSubmission {
+                    job_id: owner.job_id.clone(),
+                    attempt_id: owner.attempt_id.clone(),
+                    input_digest: owner.input_digest.clone(),
+                    result: serde_json::json!({"saved": true}),
+                },
+                NOW,
+            )
+            .unwrap();
+        let history = store
+            .list_backend_job_attempts("workspace", "accepted")
+            .unwrap();
+        let job = store
+            .get_backend_job("workspace", "accepted")
+            .unwrap()
+            .unwrap();
+        let detail =
+            "Original Host admission unavailable; termination unconfirmed; retain result/input";
+        let later = "2026-11-04T00:00:01Z";
+        assert!(
+            store
+                .record_backend_job_cleanup_preflight_failure(
+                    "workspace",
+                    &[],
+                    &completed,
+                    detail,
+                    later,
+                )
+                .is_err()
+        );
+        assert!(
+            store
+                .record_backend_job_cleanup_preflight_failure(
+                    "other-workspace",
+                    &history,
+                    &completed,
+                    detail,
+                    later,
+                )
+                .is_err()
+        );
+        let mut foreign = completed.clone();
+        foreign.runtime_run_id = Some("successor-run".into());
+        assert!(
+            store
+                .record_backend_job_cleanup_preflight_failure(
+                    "workspace",
+                    &[foreign.clone()],
+                    &foreign,
+                    detail,
+                    later,
+                )
+                .is_err()
+        );
+        // Real SQLite failure at the diagnostic write must propagate, leaving
+        // the exact observation available for its existing retry/probe owner.
+        store.with_conn(|conn| {
+            conn.execute_batch("CREATE TEMP TRIGGER preflight_fault BEFORE UPDATE OF worker_cleanup_failure_category ON main.backend_job_attempts BEGIN SELECT RAISE(ABORT,'diagnostic storage fault'); END;")?;
+            Ok(())
+        }).unwrap();
+        assert!(
+            store
+                .record_backend_job_cleanup_preflight_failure(
+                    "workspace",
+                    &history,
+                    &completed,
+                    detail,
+                    later,
+                )
+                .is_err()
+        );
+        assert_eq!(
+            store
+                .list_backend_job_attempts("workspace", "accepted")
+                .unwrap(),
+            history
+        );
+        store
+            .with_conn(|conn| {
+                conn.execute_batch("DROP TRIGGER preflight_fault")?;
+                Ok(())
+            })
+            .unwrap();
+        let failed = store
+            .record_backend_job_cleanup_preflight_failure(
+                "workspace",
+                &history,
+                &completed,
+                detail,
+                later,
+            )
+            .unwrap();
+        let mut expected = completed.clone();
+        expected.worker_cleanup_state = Some(BackendJobWorkerCleanupState::Failed);
+        expected.worker_cleanup_failure_category =
+            Some("backend_job_cleanup_host_admission_unconfirmed".into());
+        expected.worker_cleanup_failure_detail = Some(detail.into());
+        expected.worker_cleanup_updated_at = Some(later.into());
+        assert_eq!(
+            failed, expected,
+            "only cleanup diagnostic columns may change"
+        );
+        assert_eq!(
+            store
+                .get_backend_job("workspace", "accepted")
+                .unwrap()
+                .unwrap(),
+            job
+        );
+        assert!(
+            !store
+                .has_backend_job_execution_fence("workspace", &failed)
+                .unwrap()
+        );
+        assert!(
+            store
+                .record_backend_job_cleanup_preflight_failure(
+                    "workspace",
+                    &history,
+                    &completed,
+                    "late stale failure",
+                    later,
+                )
+                .is_err()
+        );
+        let fresh_history = store
+            .list_backend_job_attempts("workspace", "accepted")
+            .unwrap();
+        assert!(
+            store
+                .finish_backend_job_worker_cleanup_owned(
+                    "workspace",
+                    &fresh_history,
+                    &failed,
+                    BackendJobWorkerCleanupState::Completed,
+                    None,
+                    None,
+                    later,
+                )
+                .is_err(),
+            "diagnosis must not claim Executing ownership"
+        );
+        assert!(
+            store
+                .reserve_backend_job_retry("workspace", "accepted", later)
+                .is_err()
+        );
+        drop(store);
+        let store =
+            SqliteWorkspaceStore::open_for_test(&dir.path().join("recovery.sqlite")).unwrap();
+        assert_eq!(
+            store
+                .list_backend_job_attempts("workspace", "accepted")
+                .unwrap(),
+            vec![failed]
+        );
+        assert_eq!(
+            store
+                .get_backend_job("workspace", "accepted")
+                .unwrap()
+                .unwrap(),
+            job
+        );
+    }
+
+    #[test]
+    fn cleanup_preflight_diagnosis_keeps_executing_and_rejects_fenced_completed_or_successor_owner()
+    {
+        let (_dir, store, reserved) = fixture();
+        let bound = bind_fixture_owner(&store, &reserved);
+        // As in the fence component tests, only the external termination receipt
+        // is fixture evidence; this test does not claim real execution ended.
+        store.with_conn(|conn| {
+            conn.execute("UPDATE backend_job_attempts SET worker_id='019a4ea8-0000-7000-8000-000000000003' WHERE attempt_id=?1", params![bound.attempt_id])?;
+            Ok(())
+        }).unwrap();
+        let failed = store
+            .finish_backend_job_attempt(
+                "workspace",
+                "accepted",
+                &bound.attempt_id,
+                BackendJobAttemptState::Failed,
+                "execution_failed",
+                "saved original failure",
+                NOW,
+            )
+            .unwrap();
+        let history = store
+            .list_backend_job_attempts("workspace", "accepted")
+            .unwrap();
+        let (executing, acquired) = store
+            .claim_backend_job_worker_cleanup_owned("workspace", &history, &failed, NOW)
+            .unwrap();
+        assert!(acquired);
+        let history = store
+            .list_backend_job_attempts("workspace", "accepted")
+            .unwrap();
+        let diagnosed = store
+            .record_backend_job_cleanup_preflight_failure(
+                "workspace",
+                &history,
+                &executing,
+                "Host close/fence unavailable; no termination proof",
+                NOW,
+            )
+            .unwrap();
+        assert_eq!(
+            diagnosed.worker_cleanup_state,
+            Some(BackendJobWorkerCleanupState::Executing)
+        );
+        assert_eq!(
+            diagnosed.failure_detail.as_deref(),
+            Some("saved original failure")
+        );
+        assert_eq!(
+            diagnosed.worker_cleanup_failure_category.as_deref(),
+            Some("backend_job_cleanup_host_admission_unconfirmed")
+        );
+        assert!(diagnosed.worker_cleanup_completed_at.is_none());
+        assert!(
+            !store
+                .has_backend_job_execution_fence("workspace", &diagnosed)
+                .unwrap()
+        );
+        assert!(
+            store
+                .finish_backend_job_worker_cleanup_owned(
+                    "workspace",
+                    &history,
+                    &executing,
+                    BackendJobWorkerCleanupState::Completed,
+                    None,
+                    None,
+                    NOW,
+                )
+                .is_err(),
+            "intervening diagnosis must invalidate stale completion CAS"
+        );
+        let history = store
+            .list_backend_job_attempts("workspace", "accepted")
+            .unwrap();
+        let (controller, host) = evidence(&diagnosed);
+        store
+            .record_backend_job_execution_fence("workspace", &diagnosed, &controller, &host, NOW)
+            .unwrap();
+        assert!(
+            store
+                .record_backend_job_cleanup_preflight_failure(
+                    "workspace",
+                    &history,
+                    &diagnosed,
+                    "late unfenced observation",
+                    NOW,
+                )
+                .is_err()
+        );
+        assert_eq!(
+            store
+                .list_backend_job_attempts("workspace", "accepted")
+                .unwrap(),
+            history
+        );
+        let completed = store
+            .finish_backend_job_worker_cleanup_owned(
+                "workspace",
+                &history,
+                &diagnosed,
+                BackendJobWorkerCleanupState::Completed,
+                None,
+                None,
+                NOW,
+            )
+            .unwrap();
+        let history = store
+            .list_backend_job_attempts("workspace", "accepted")
+            .unwrap();
+        assert!(
+            store
+                .record_backend_job_cleanup_preflight_failure(
+                    "workspace",
+                    &history,
+                    &completed,
+                    "late failure after completion",
+                    NOW,
+                )
+                .is_err()
+        );
+        let next = store
+            .reserve_backend_job_retry("workspace", "accepted", NOW)
+            .unwrap();
+        let successor = bind_fixture_owner(&store, &next);
+        let next_job = store
+            .get_backend_job("workspace", "accepted")
+            .unwrap()
+            .unwrap();
+        let history = store
+            .list_backend_job_attempts("workspace", "accepted")
+            .unwrap();
+        assert!(
+            store
+                .record_backend_job_cleanup_preflight_failure(
+                    "workspace",
+                    &history,
+                    &completed,
+                    "late historical failure",
+                    NOW,
+                )
+                .is_err()
+        );
+        assert!(
+            store
+                .record_backend_job_cleanup_preflight_failure(
+                    "workspace",
+                    &history,
+                    &successor,
+                    "nonterminal owner cannot diagnose cleanup",
+                    NOW,
+                )
+                .is_err()
+        );
+        assert_eq!(
+            store
+                .list_backend_job_attempts("workspace", "accepted")
+                .unwrap(),
+            history
+        );
+        assert_eq!(
+            store
+                .get_backend_job("workspace", "accepted")
+                .unwrap()
+                .unwrap(),
+            next_job
+        );
+    }
+
     #[test]
     fn cleanup_cas_rejects_changed_history_and_wrong_worker_run_without_releasing_owner() {
         let (_dir, store, reserved) = fixture();

crates/workspace-server/src/server.rs

@@ -5582,11 +5582,30 @@ impl WorkspaceApi {
 
     /// Restore owners for current-contract attempts without replaying uncertain effects.
     pub fn recover_backend_jobs(&self) -> Result<()> {
-        for attempt in self.store.list_recoverable_backend_job_attempts(
-            self.workspace_id(),
-            usize::from(ABSOLUTE_MAX_CONCURRENT_JOBS),
-        )? {
-            self.own_backend_job_recovery(&attempt);
+        let page_limit = usize::from(ABSOLUTE_MAX_CONCURRENT_JOBS);
+        let mut after = None;
+        loop {
+            let attempts = self.store.list_recoverable_backend_job_attempts_page(
+                self.workspace_id(),
+                after.as_ref(),
+                page_limit,
+            )?;
+            let Some(last) = attempts.last() else {
+                break;
+            };
+            let next = crate::store::BackendJobRecoveryCursor {
+                job_id: last.job_id.clone(),
+                attempt_id: last.attempt_id.clone(),
+            };
+            for attempt in &attempts {
+                self.own_backend_job_recovery(attempt);
+            }
+            if attempts.len() < page_limit {
+                break;
+            }
+            // Reservations are not bounded by active concurrency. Traverse past
+            // admitted owners even if they remain pending or change timestamps.
+            after = Some(next);
         }
         subjektiv_journal::recover_attempts(self)?;
         self.recover_backend_job_worker_cleanups()?;

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

@@ -22,6 +22,10 @@ pub(super) struct RecoveryOwners {
     submitted: Mutex<HashSet<String>>,
     stopping: std::sync::atomic::AtomicBool,
     changed: tokio::sync::Notify,
+    // Already accepted owners take fair turns before entering the finite Host
+    // lane. Immediate rejection plus independent retry timers otherwise lets
+    // earlier reservations monopolize every slot after a large reopen.
+    step_admission: tokio::sync::Mutex<()>,
     #[cfg(test)]
     pub(super) storage_probe_started: tokio::sync::Notify,
     #[cfg(test)]
@@ -44,6 +48,11 @@ impl RecoveryOwners {
         self.stopping.store(false, Ordering::Release);
     }
 
+    #[cfg(test)]
+    pub(super) fn recovery_owner_ids(&self) -> HashSet<String> {
+        self.running.lock().unwrap().clone()
+    }
+
     #[cfg(test)]
     pub(super) fn cleanup_owner_ids(&self) -> HashSet<String> {
         self.cleanup_running.lock().unwrap().clone()
@@ -498,17 +507,54 @@ impl WorkspaceApi {
                 .host_executors
                 .cleanup()?
                 .run(move || {
+                    // Observe before preflight: a failure may diagnose only this
+                    // exact row/history, never a successor or intervening progress.
+                    let history = api
+                        .store
+                        .list_backend_job_attempts(api.workspace_id(), &owner.job_id)?;
+                    let observed = history
+                        .iter()
+                        .find(|attempt| attempt.attempt_id == owner.attempt_id)
+                        .ok_or_else(|| Error::Store("Cleanup owner disappeared".into()))?;
+                    if !owner.state.terminal()
+                        || !observed.state.terminal()
+                        || observed.attempt != owner.attempt
+                        || JobHostOwner::try_from(observed)? != JobHostOwner::try_from(&owner)?
+                    {
+                        return Err(Error::RepositoryConflict(
+                            "Cleanup execution owner changed before preflight".into(),
+                        ));
+                    }
                     // 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 preflight = (|| {
+                        api.backend_job_recovery
+                            .close_host_admission(&JobHostOwner::try_from(observed)?)?;
+                        #[cfg(test)]
+                        api.backend_job_recovery
+                            .cleanup_before_writer_fence
+                            .notify_one();
+                        JobHostEffects::fence_retained(&api.config.database_path, observed)
+                    })();
+                    let drain = match preflight {
+                        Ok(drain) => drain,
+                        Err(error) => {
+                            api.store.record_backend_job_cleanup_preflight_failure(
+                                api.workspace_id(),
+                                &history,
+                                observed,
+                                &format!(
+                                    "Host admission close/fence failed: {error}; execution and Host termination remain unconfirmed. Retain result/input and reconcile the original admission resource; never create replacement evidence."
+                                ),
+                                &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true),
+                            )?;
+                            // Neither a saved diagnosis nor failure to save it is
+                            // success. Storage errors still reach retry/probe ownership.
+                            return Err(error);
+                        }
+                    };
                     let current = api
                         .store
                         .get_backend_job_attempt(
@@ -667,6 +713,10 @@ impl WorkspaceApi {
             let api = self.clone();
             let owner = expected.clone();
             let failure = definitive_failure.clone();
+            let admission = self.backend_job_recovery.step_admission.lock().await;
+            if self.backend_job_recovery.stopping.load(Ordering::Acquire) {
+                return Ok(());
+            }
             let step = self
                 .host_executors
                 .host()?
@@ -689,6 +739,7 @@ impl WorkspaceApi {
                     }
                 })
                 .await;
+            drop(admission);
             match step {
                 Ok(true) => return Ok(()),
                 Ok(false) => {
@@ -741,12 +792,17 @@ impl WorkspaceApi {
                         self.backend_job_recovery.pause(delay).await;
                         let api = self.clone();
                         let owner = expected.clone();
-                        match self
+                        let admission = self.backend_job_recovery.step_admission.lock().await;
+                        if self.backend_job_recovery.stopping.load(Ordering::Acquire) {
+                            return Ok(());
+                        }
+                        let probe = self
                             .host_executors
                             .host()?
                             .run(move || api.probe_backend_job_storage(&owner))
-                            .await
-                        {
+                            .await;
+                        drop(admission);
+                        match probe {
                             Ok(()) => {
                                 storage_failures = 0;
                                 break;

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

@@ -123,7 +123,14 @@ async fn wait_for_attempt(
         }
     })
     .await
-    .expect("same accepted attempt did not reach its expected durable state")
+    .unwrap_or_else(|error| {
+        panic!(
+            "same accepted attempt did not reach its expected durable state: {error}; current={:?}; recovery_owners={:?}",
+            api.store
+                .get_backend_job_attempt(api.workspace_id(), job, attempt),
+            api.backend_job_recovery.recovery_owner_ids()
+        )
+    })
 }
 
 #[tokio::test]
@@ -1055,7 +1062,7 @@ async fn terminal_backend_job_cleanup_removes_only_bound_worker_and_retains_resu
 }
 
 #[tokio::test]
-async fn backend_job_cleanup_guard_failure_retries_old_attempt_without_touching_new_worker() {
+async fn backend_job_cleanup_commit_failure_retries_old_attempt_without_touching_new_worker() {
     let temp = tempfile::tempdir().unwrap();
     let (api, _execution) = test_api_with_recording_backend(temp.path()).await;
     let request = BackendJobRequest {
@@ -1074,14 +1081,18 @@ async fn backend_job_cleanup_guard_failure_retries_old_attempt_without_touching_
     let first = api.dispatch_backend_job(&request).unwrap();
     let first_worker = first.attempt.worker.clone().unwrap();
     seed_worker_session_for_cleanup(temp.path(), &first_worker);
-    let _ = set_worker_retention(
-        api.clone(),
-        first_worker.runtime_id.clone(),
-        first_worker.worker_id.clone(),
-        true,
-    )
-    .await
-    .unwrap();
+    // Fail the canonical catalog commit after Runtime retention, without changing
+    // Worker/policy authority and thereby invalidating the original removal child.
+    api.config_store
+        .with_conn(|conn| {
+            conn.execute_batch(
+                "CREATE TRIGGER fail_job_cleanup_catalog_commit
+                 BEFORE INSERT ON worker_registry_projection_removals
+                 BEGIN SELECT RAISE(ABORT, 'injected cleanup catalog failure'); END;",
+            )?;
+            Ok(())
+        })
+        .unwrap();
     api.finish_backend_job_attempt(
         &request.job_id,
         &first.attempt.attempt_id,
@@ -1102,33 +1113,87 @@ async fn backend_job_cleanup_guard_failure_retries_old_attempt_without_touching_
         blocked.worker_cleanup_failure_category.as_deref(),
         Some("backend_job_cleanup_removal_rejected")
     );
+    assert!(
+        blocked
+            .worker_cleanup_failure_detail
+            .as_deref()
+            .unwrap()
+            .contains("worker_removal_authority_unavailable"),
+        "canonical catalog commit failure must remain a removal error: {blocked:?}"
+    );
+    let blocked_removal = api
+        .config_store
+        .recover_worker_removal_execution(TEST_WORKSPACE_ID, &first_worker)
+        .unwrap()
+        .unwrap();
+    assert_eq!(
+        blocked_removal.prior_failure_category.as_deref(),
+        Some("metadata_commit_failed")
+    );
+    let first_removal_children = || {
+        api.config_store
+            .with_conn(|conn| {
+                let mut statement = conn.prepare(
+                    "SELECT operation_id, input_fingerprint FROM worker_removal_operations
+                     WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3
+                     ORDER BY operation_id",
+                )?;
+                let children = statement
+                    .query_map(
+                        rusqlite::params![
+                            api.workspace_id(),
+                            first_worker.runtime_id,
+                            first_worker.worker_id
+                        ],
+                        |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)),
+                    )?
+                    .collect::<rusqlite::Result<Vec<_>>>()?;
+                Ok(children)
+            })
+            .unwrap()
+    };
+    let original_children = first_removal_children();
+    assert_eq!(original_children.len(), 1);
+    assert_eq!(blocked.worker, Some(first_worker.clone()));
+    assert_eq!(blocked.runtime_run_id, first.attempt.runtime_run_id);
 
     assert!(
         api.retry_backend_job(&request.job_id).is_err(),
         "retry must not overlap an incompletely cleaned attempt"
     );
-    let _ = set_worker_retention(
-        api.clone(),
-        first_worker.runtime_id.clone(),
-        first_worker.worker_id.clone(),
-        false,
-    )
-    .await
-    .unwrap();
+    api.config_store
+        .with_conn(|conn| {
+            conn.execute_batch("DROP TRIGGER fail_job_cleanup_catalog_commit;")?;
+            Ok(())
+        })
+        .unwrap();
     api.recover_backend_jobs().unwrap();
-    wait_for_backend_job_worker_cleanup(
+    let cleaned = wait_for_backend_job_worker_cleanup(
         &api,
         &request.job_id,
         &first.attempt.attempt_id,
         BackendJobWorkerCleanupState::Completed,
     )
     .await;
+    assert_eq!(first_removal_children(), original_children);
+    assert_eq!(cleaned.worker, blocked.worker);
+    assert_eq!(cleaned.runtime_run_id, blocked.runtime_run_id);
+    api.require_backend_job_attempt_effects_ended(&cleaned)
+        .unwrap();
+    assert!(api.runtime.worker(&first_worker).is_err());
+    assert!(
+        api.store
+            .get_worker_registry(TEST_WORKSPACE_ID, &first_worker)
+            .unwrap()
+            .is_none()
+    );
     let second = api.retry_backend_job(&request.job_id).unwrap();
     let second_worker = second.attempt.worker.clone().unwrap();
     seed_worker_session_for_cleanup(temp.path(), &second_worker);
     assert_ne!(first_worker, second_worker);
     api.recover_backend_jobs().unwrap();
     assert!(api.runtime.worker(&second_worker).is_ok());
+    assert_eq!(first_removal_children(), original_children);
 
     api.accept_backend_job_result(
         &second_worker,
@@ -1312,6 +1377,12 @@ async fn backend_job_restart_without_live_host_owner_keeps_cleanup_unconfirmed()
             .worker_cleanup_state,
         Some(BackendJobWorkerCleanupState::Pending)
     );
+    // This fixture deliberately bypasses producer Host admission. Missing the
+    // original durable resource is not merely losing the live owner map.
+    let mut admission_root = config.database_path.as_os_str().to_owned();
+    admission_root.push(".job-host-admission");
+    let admission_root = std::path::PathBuf::from(admission_root);
+    assert!(!admission_root.exists());
     let attempt_id = reservation.attempt.attempt_id;
     drop(runtime);
     drop(embedded);
@@ -1325,7 +1396,7 @@ async fn backend_job_restart_without_live_host_owner_keeps_cleanup_unconfirmed()
     let restarted = WorkspaceApi::new_with_execution_backend(
         config,
         Arc::new(restarted_store),
-        restarted_execution,
+        restarted_execution.clone(),
     )
     .await
     .unwrap();
@@ -1337,6 +1408,53 @@ async fn backend_job_restart_without_live_host_owner_keeps_cleanup_unconfirmed()
     )
     .await;
     assert_eq!(cleaned.result, Some(serde_json::json!({"persisted": true})));
+    assert_eq!(cleaned.state, BackendJobAttemptState::Completed);
+    assert_eq!(cleaned.worker.as_ref(), Some(&worker));
+    assert_eq!(
+        cleaned.runtime_run_id.as_deref(),
+        Some(runtime_run_id.as_str())
+    );
+    assert_eq!(
+        cleaned.worker_cleanup_failure_category.as_deref(),
+        Some("backend_job_cleanup_host_admission_unconfirmed")
+    );
+    let cause = cleaned.worker_cleanup_failure_detail.as_deref().unwrap();
+    assert!(cause.contains("Original Job Host admission resource unavailable"));
+    assert!(cause.contains("termination remain unconfirmed"));
+    assert!(cleaned.worker_cleanup_completed_at.is_none());
+    assert!(cleaned.allocation_precluded_at.is_none());
+    assert!(
+        !admission_root.exists(),
+        "recovery must not create Host evidence"
+    );
+    let saved = restarted
+        .store
+        .get_backend_job(TEST_WORKSPACE_ID, &request.job_id)
+        .unwrap()
+        .unwrap();
+    assert_eq!(saved.state, BackendJobState::Completed);
+    assert_eq!(saved.result, Some(serde_json::json!({"persisted": true})));
+    assert_eq!(saved.request, request);
+    assert_eq!(
+        saved.input_disposition,
+        crate::backend_job::JobInputDisposition::Available
+    );
+    assert!(
+        !restarted
+            .store
+            .has_backend_job_execution_fence(TEST_WORKSPACE_ID, &cleaned)
+            .unwrap()
+    );
+    restarted.config_store.with_conn(|connection| {
+        let fences: i64 = connection.query_row("SELECT COUNT(*) FROM backend_job_execution_fences WHERE workspace_id=?1 AND job_id=?2 AND attempt_id=?3", rusqlite::params![TEST_WORKSPACE_ID, request.job_id, attempt_id], |row| row.get(0))?;
+        assert_eq!(fences, 0, "preflight failure must not fabricate execution evidence");
+        let removals: i64 = connection.query_row("SELECT COUNT(*) FROM worker_removal_operations WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3", rusqlite::params![TEST_WORKSPACE_ID, worker.runtime_id, worker.worker_id], |row| row.get(0))?;
+        assert_eq!(removals, 0, "preflight failure must not call canonical removal");
+        Ok(())
+    }).unwrap();
+    assert!(restarted_execution.stops.lock().unwrap().is_empty());
+    assert!(restarted_execution.terminations.lock().unwrap().is_empty());
+    assert!(restarted_execution.take_inputs().is_empty());
     assert!(
         restarted
             .require_backend_job_attempt_effects_ended(&cleaned)
@@ -1710,22 +1828,82 @@ async fn backend_job_input_response_loss_is_owned_until_deadline_without_resubmi
         Some("submission_receipt_unconfirmed")
     );
     let worker = failed.worker.clone().unwrap();
-    wait_for_backend_job_worker_cleanup(
+    // The substituted callee returned Errored before saving an acceptance receipt.
+    // Stop can join its fixture execution, but cannot attest this exact run ended.
+    let cleaned = wait_for_backend_job_worker_cleanup(
         &api,
         &request.job_id,
         &failed.attempt_id,
-        BackendJobWorkerCleanupState::Completed,
+        BackendJobWorkerCleanupState::Failed,
     )
     .await;
-    assert!(matches!(
-        api.runtime.worker(&worker),
-        Err(RuntimeRegistryError::UnknownWorker { .. })
-    ));
+    assert_eq!(cleaned.state, BackendJobAttemptState::Failed);
+    assert_eq!(cleaned.failure_category.as_deref(), Some("timeout"));
+    assert!(
+        cleaned
+            .failure_detail
+            .as_deref()
+            .unwrap()
+            .contains("Attempt deadline exceeded")
+    );
+    assert_eq!(cleaned.worker, failed.worker);
+    assert_eq!(cleaned.runtime_run_id, failed.runtime_run_id);
+    assert_eq!(cleaned.deadline_at, failed.deadline_at);
+    assert_eq!(
+        cleaned.worker_cleanup_failure_category.as_deref(),
+        Some("backend_job_cleanup_effects_unconfirmed")
+    );
+    assert!(
+        cleaned
+            .worker_cleanup_failure_detail
+            .as_deref()
+            .unwrap()
+            .contains(
+                "Runtime receipt does not prove exact Controller/Feature/background termination"
+            )
+    );
+    assert!(cleaned.worker_cleanup_completed_at.is_none());
+    assert!(
+        api.require_backend_job_attempt_effects_ended(&cleaned)
+            .is_err()
+    );
+    assert!(api.retry_backend_job(&request.job_id).is_err());
+    let retained = api
+        .store
+        .get_backend_job(TEST_WORKSPACE_ID, &request.job_id)
+        .unwrap()
+        .unwrap();
+    assert_eq!(retained.state, BackendJobState::Failed);
+    assert_eq!(retained.failure_category.as_deref(), Some("timeout"));
+    assert_eq!(retained.request, request);
+    assert_eq!(
+        retained.input_disposition,
+        crate::backend_job::JobInputDisposition::Available
+    );
+    assert!(api.runtime.worker(&worker).is_ok());
+    assert!(
+        api.store
+            .get_worker_registry(TEST_WORKSPACE_ID, &worker)
+            .unwrap()
+            .is_some()
+    );
+    assert!(
+        api.config_store
+            .recover_worker_removal_execution(TEST_WORKSPACE_ID, &worker)
+            .unwrap()
+            .is_none(),
+        "unconfirmed exact run termination must not authorize canonical removal"
+    );
     assert_eq!(
         execution.take_inputs().len(),
         1,
         "ambiguous input was never blindly resubmitted"
     );
+    assert_eq!(
+        execution.take_input_request_ids(),
+        vec![failed.runtime_run_id.clone()],
+        "only the original tracked Submit may reach the execution callee"
+    );
     assert!(
         api.store
             .get_backend_job_attempt(
@@ -1736,6 +1914,7 @@ async fn backend_job_input_response_loss_is_owned_until_deadline_without_resubmi
             .unwrap()
             .is_none()
     );
+    api.stop_backend_job_recovery().await;
 }
 
 #[tokio::test]
@@ -1962,7 +2141,10 @@ async fn backend_job_missing_worker_reaches_failure_without_fabricated_cleanup()
         serialization_key: None,
         source_worker: None,
         notification_target: None,
-        limits: crate::backend_job::BackendJobLimits::default(),
+        limits: crate::backend_job::BackendJobLimits {
+            timeout_seconds: 1,
+            ..Default::default()
+        },
     };
     let reserved = api
         .store
@@ -2021,6 +2203,12 @@ async fn backend_job_missing_worker_reaches_failure_without_fabricated_cleanup()
     )
     .await;
     assert_eq!(attempt.failure_category.as_deref(), Some("timeout"));
+    assert_eq!(attempt.deadline_at, reserved.attempt.deadline_at);
+    assert_eq!(attempt.worker, Some(missing_worker));
+    assert_eq!(
+        attempt.runtime_run_id.as_deref(),
+        Some("missing-runtime-run")
+    );
     assert_ne!(
         attempt.worker_cleanup_state,
         Some(BackendJobWorkerCleanupState::Completed)
@@ -2093,3 +2281,235 @@ async fn backend_job_timeout_fences_late_structured_result() {
         .is_err()
     );
 }
+#[tokio::test]
+async fn backend_job_reopen_owns_later_nonjournal_attempt_behind_queued_reservations() {
+    let temp = tempfile::tempdir().unwrap();
+    let (api, execution) = test_api_with_recording_backend(temp.path()).await;
+    let config = api.config.clone();
+    // Pause only automatic recovery while creating the reachable accepted state.
+    // Dispatch, capacity checks, Runtime acceptance and persistence remain real.
+    api.stop_backend_job_recovery().await;
+    let mut first_request = recovery_request("a-startup-active");
+    first_request.limits.max_concurrent_jobs = 1;
+    first_request.limits.timeout_seconds = crate::backend_job::MAX_JOB_TIMEOUT_SECONDS;
+    let first = api.dispatch_backend_job(&first_request).unwrap();
+    assert_eq!(first.attempt.state, BackendJobAttemptState::Dispatched);
+
+    let mut queued = Vec::new();
+    for index in 0..usize::from(ABSOLUTE_MAX_CONCURRENT_JOBS) {
+        let mut request = recovery_request(&format!("b-startup-queued-{index:03}"));
+        request.limits.max_concurrent_jobs = 1;
+        request.limits.timeout_seconds = crate::backend_job::MAX_JOB_TIMEOUT_SECONDS;
+        let reservation = api.dispatch_backend_job(&request).unwrap();
+        assert_eq!(reservation.attempt.state, BackendJobAttemptState::Reserved);
+        assert!(reservation.attempt.worker.is_none());
+        queued.push(reservation.attempt);
+    }
+    let mut later_request = recovery_request("z-startup-later-active");
+    later_request.limits.max_concurrent_jobs = ABSOLUTE_MAX_CONCURRENT_JOBS;
+    // This attempt's own original deadline supplies the autonomous observation;
+    // older work stays pending. No fixed sleep or production timeout change.
+    later_request.limits.timeout_seconds = 1;
+    let later = api.dispatch_backend_job(&later_request).unwrap();
+    assert_eq!(later.attempt.state, BackendJobAttemptState::Dispatched);
+    assert!(later.job.request.grants.subjektiv_journal.is_none());
+    let original_worker = later.attempt.worker.clone().unwrap();
+    assert_eq!(execution.take_inputs().len(), 2);
+    assert!(
+        execution
+            .take_input_request_ids()
+            .contains(&later.attempt.runtime_run_id)
+    );
+
+    let first_page = api
+        .store
+        .list_recoverable_backend_job_attempts(
+            api.workspace_id(),
+            usize::from(ABSOLUTE_MAX_CONCURRENT_JOBS),
+        )
+        .unwrap();
+    assert_eq!(first_page.len(), usize::from(ABSOLUTE_MAX_CONCURRENT_JOBS));
+    assert!(
+        first_page
+            .iter()
+            .all(|attempt| attempt.job_id != later_request.job_id)
+    );
+    assert!(
+        api.store
+            .list_reserved_backend_job_attempts(api.workspace_id(), queued.len() + 1)
+            .unwrap()
+            .iter()
+            .all(|attempt| attempt.job_id != later_request.job_id)
+    );
+
+    // Reopen the actual workspace DB and construct a fresh API/Runtime. Do not
+    // substitute an empty owner map on a live API or call full recovery manually.
+    shutdown_workspace_feature_storage(&api).await.unwrap();
+    drop(api);
+    drop(execution);
+    let reopened_store =
+        Arc::new(SqliteWorkspaceStore::open_for_test(config.database_path.clone()).unwrap());
+    let reopened_execution = Arc::new(DeterministicExecutionBackend::default());
+    reopened_execution.accept_restores();
+    let reopened = WorkspaceApi::new_with_execution_backend(
+        config,
+        reopened_store,
+        reopened_execution.clone(),
+    )
+    .await
+    .unwrap();
+    assert!(
+        reopened
+            .backend_job_recovery
+            .recovery_owner_ids()
+            .contains(&later.attempt.attempt_id)
+            || reopened
+                .store
+                .get_backend_job_attempt(
+                    reopened.workspace_id(),
+                    &later_request.job_id,
+                    &later.attempt.attempt_id
+                )
+                .unwrap()
+                .unwrap()
+                .state
+                .terminal(),
+        "startup must own later active attempt; owners={:?}; candidate={:?}; job={:?}",
+        reopened.backend_job_recovery.recovery_owner_ids(),
+        reopened.store.list_recoverable_backend_job_attempts_page(
+            reopened.workspace_id(),
+            Some(&crate::store::BackendJobRecoveryCursor {
+                job_id: "b-startup-queued-063".into(),
+                attempt_id: "".into()
+            }),
+            64
+        ),
+        reopened
+            .store
+            .get_backend_job(reopened.workspace_id(), &later_request.job_id)
+    );
+    let recovered = wait_for_attempt(
+        &reopened,
+        &later_request.job_id,
+        &later.attempt.attempt_id,
+        |attempt| attempt.state == BackendJobAttemptState::Failed,
+    )
+    .await;
+    assert_eq!(recovered.failure_category.as_deref(), Some("timeout"));
+    assert_eq!(recovered.worker, Some(original_worker));
+    assert_eq!(recovered.attempt, later.attempt.attempt);
+    assert_eq!(recovered.runtime_run_id, later.attempt.runtime_run_id);
+    assert_eq!(recovered.input_digest, later.attempt.input_digest);
+    assert_eq!(recovered.deadline_at, later.attempt.deadline_at);
+    assert_eq!(
+        reopened
+            .store
+            .list_backend_job_attempts(reopened.workspace_id(), &later_request.job_id)
+            .unwrap()
+            .len(),
+        1,
+        "startup recovery must retain the original accepted attempt"
+    );
+    for attempt in &queued {
+        let saved = reopened
+            .store
+            .get_backend_job_attempt(
+                reopened.workspace_id(),
+                &attempt.job_id,
+                &attempt.attempt_id,
+            )
+            .unwrap()
+            .unwrap();
+        assert_eq!(saved.state, BackendJobAttemptState::Reserved);
+        assert!(saved.worker.is_none());
+    }
+    assert!(
+        reopened_execution.take_inputs().is_empty(),
+        "restored active non-Journal work must not be submitted again"
+    );
+    shutdown_workspace_feature_storage(&reopened).await.unwrap();
+}
+
+#[tokio::test]
+async fn backend_job_recovery_pages_remain_read_only_when_earlier_attempts_terminalize() {
+    let temp = tempfile::tempdir().unwrap();
+    let api = test_api(temp.path()).await;
+    api.stop_backend_job_recovery().await;
+    let page_limit = usize::from(ABSOLUTE_MAX_CONCURRENT_JOBS);
+    let now = Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true);
+    let mut seeded = Vec::new();
+    for index in 0..(page_limit + 2) {
+        let request = recovery_request(&format!("startup-keyset-{index:03}"));
+        seeded.push(
+            api.store
+                .reserve_backend_job(api.workspace_id(), &request, &now)
+                .unwrap()
+                .attempt,
+        );
+    }
+    let first = api
+        .store
+        .list_recoverable_backend_job_attempts_page(api.workspace_id(), None, usize::MAX)
+        .unwrap();
+    assert_eq!(first, seeded[..page_limit]);
+    for attempt in &seeded {
+        assert_eq!(
+            api.store
+                .get_backend_job_attempt(api.workspace_id(), &attempt.job_id, &attempt.attempt_id)
+                .unwrap()
+                .unwrap(),
+            *attempt,
+            "enumeration must not claim, rewrite timestamps, or reset dispatch"
+        );
+    }
+    let last = first.last().unwrap();
+    let cursor = crate::store::BackendJobRecoveryCursor {
+        job_id: last.job_id.clone(),
+        attempt_id: last.attempt_id.clone(),
+    };
+    // An OFFSET or mutable-state cursor would skip the later accepted work after
+    // the first page's owners progress. All creation times are deliberately equal.
+    for attempt in &first {
+        api.store
+            .finish_backend_job_attempt(
+                api.workspace_id(),
+                &attempt.job_id,
+                &attempt.attempt_id,
+                BackendJobAttemptState::Failed,
+                "fixture_preallocation_failure",
+                "First page settled before the next bounded read",
+                &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true),
+            )
+            .unwrap();
+    }
+    let second = api
+        .store
+        .list_recoverable_backend_job_attempts_page(api.workspace_id(), Some(&cursor), page_limit)
+        .unwrap();
+    assert_eq!(second, seeded[page_limit..]);
+    for attempt in &second {
+        assert_eq!(
+            api.store
+                .get_backend_job_attempt(api.workspace_id(), &attempt.job_id, &attempt.attempt_id)
+                .unwrap()
+                .unwrap(),
+            *attempt
+        );
+    }
+    let last = second.last().unwrap();
+    let cursor = crate::store::BackendJobRecoveryCursor {
+        job_id: last.job_id.clone(),
+        attempt_id: last.attempt_id.clone(),
+    };
+    assert!(
+        api.store
+            .list_recoverable_backend_job_attempts_page(
+                api.workspace_id(),
+                Some(&cursor),
+                page_limit
+            )
+            .unwrap()
+            .is_empty()
+    );
+    shutdown_workspace_feature_storage(&api).await.unwrap();
+}

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

@@ -2759,7 +2759,40 @@ async fn connected_explicit_repair(scenario: ConnectedRepair) {
         // independent Journal job, not a second execution of the restored body.
         let before_stop = fixture.execution.worker_snapshot(&worker_ref);
         let original_stop_count = probe.stops.lock().unwrap().len();
-        let stopped = connected_stop_worker(api, worker).await;
+        let mut stopped = connected_stop_worker(api, worker).await;
+        let uncertain = {
+            let stops = probe.stops.lock().unwrap();
+            stops[original_stop_count..]
+                .last()
+                .and_then(|(request, result)| {
+                    result
+                        .message
+                        .as_deref()
+                        .filter(|message| {
+                            message.contains("admission is unknown and waiter is retained")
+                        })
+                        .and_then(|message| message.strip_prefix("Shutdown command "))
+                        .and_then(|message| message.split_whitespace().next()?.parse::<u64>().ok())
+                        .map(|command| (request.clone(), command))
+                })
+        };
+        if let Some((request, _)) = &uncertain {
+            // A bounded admission wait is not definitive rejection. Reconcile
+            // the retained operation/command once, without a readiness method,
+            // replacement Controller, new Stop identity, or a new input.
+            let pending = record();
+            assert_eq!(
+                pending["execution_state"]["state"],
+                "reconciliation_required"
+            );
+            assert_eq!(pending["execution_state"]["execution"]["operation"], "stop");
+            assert_eq!(
+                pending["execution_state"]["execution"]["intent"]["operation_id"],
+                serde_json::to_value(request.operation_id).unwrap()
+            );
+            assert!(pending["restore_guard"]["execution_termination"].is_null());
+            stopped = connected_stop_worker(api, worker).await;
+        }
         let task_finished = probe
             .controller_tasks
             .lock()
@@ -2788,6 +2821,20 @@ async fn connected_explicit_repair(scenario: ConnectedRepair) {
             .expect("teardown must obtain a typed Stop from the real adapter");
         assert_eq!(stop.worker_ref, worker_ref);
         assert!(result.is_accepted(), "real adapter teardown: {result:?}");
+        assert!(
+            task_finished.iter().all(|finished| *finished),
+            "real Controller tasks must finish: {task_finished:?}"
+        );
+        if let Some((original, command)) = uncertain {
+            assert!(stops[original_stop_count..].iter().all(|(request, _)| {
+                request.operation_id == original.operation_id
+                    && request.worker_ref == original.worker_ref
+            }));
+            assert_eq!(
+                result.termination.as_ref().unwrap().shutdown_command_id,
+                Some(command)
+            );
+        }
         assert!(
             result.termination.as_ref().is_some_and(
                 |receipt| receipt.confirms_scope_shutdown(&worker_ref, stop.operation_id)

crates/workspace-server/src/store.rs

@@ -1396,6 +1396,13 @@ pub struct HostingAdminInitialBearer {
     pub expires_at: String,
 }
 
+/// Immutable boundary for the all-state startup recovery scan, not an execution claim.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct BackendJobRecoveryCursor {
+    pub job_id: String,
+    pub attempt_id: String,
+}
+
 #[async_trait]
 pub trait ControlPlaneStore: Send + Sync + WorkspaceDeletionStore {
     fn plugin_upload(
@@ -2054,6 +2061,15 @@ pub trait ControlPlaneStore: Send + Sync + WorkspaceDeletionStore {
         expected: &BackendJobAttemptRecord,
         now: &str,
     ) -> Result<(BackendJobAttemptRecord, bool)>;
+    /// Diagnose failed Host preflight without claiming cleanup or proving termination.
+    fn record_backend_job_cleanup_preflight_failure(
+        &self,
+        workspace_id: &str,
+        expected_history: &[BackendJobAttemptRecord],
+        expected: &BackendJobAttemptRecord,
+        failure_detail: &str,
+        now: &str,
+    ) -> Result<BackendJobAttemptRecord>;
     fn finish_backend_job_worker_cleanup_owned(
         &self,
         workspace_id: &str,
@@ -2064,10 +2080,21 @@ pub trait ControlPlaneStore: Send + Sync + WorkspaceDeletionStore {
         failure_detail: Option<&str>,
         now: &str,
     ) -> Result<BackendJobAttemptRecord>;
+    /// Read the first bounded page; use the paged variant to enumerate all owners.
     fn list_recoverable_backend_job_attempts(
         &self,
         workspace_id: &str,
         limit: usize,
+    ) -> Result<Vec<BackendJobAttemptRecord>> {
+        self.list_recoverable_backend_job_attempts_page(workspace_id, None, limit)
+    }
+    /// Read one bounded page without claiming or changing any attempt.
+    /// Startup must traverse every page, independently of admitted owner progress.
+    fn list_recoverable_backend_job_attempts_page(
+        &self,
+        workspace_id: &str,
+        after: Option<&BackendJobRecoveryCursor>,
+        limit: usize,
     ) -> Result<Vec<BackendJobAttemptRecord>>;
     fn claim_backend_job_attempt_dispatch(
         &self,
@@ -8315,6 +8342,23 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
             now,
         )
     }
+    fn record_backend_job_cleanup_preflight_failure(
+        &self,
+        workspace_id: &str,
+        expected_history: &[BackendJobAttemptRecord],
+        expected: &BackendJobAttemptRecord,
+        failure_detail: &str,
+        now: &str,
+    ) -> Result<BackendJobAttemptRecord> {
+        backend_job_recovery_store::record_cleanup_preflight_failure(
+            self,
+            workspace_id,
+            expected_history,
+            expected,
+            failure_detail,
+            now,
+        )
+    }
     fn finish_backend_job_worker_cleanup_owned(
         &self,
         workspace_id: &str,
@@ -8336,26 +8380,42 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
             now,
         )
     }
-    fn list_recoverable_backend_job_attempts(
+    fn list_recoverable_backend_job_attempts_page(
         &self,
         workspace_id: &str,
+        after: Option<&BackendJobRecoveryCursor>,
         limit: usize,
     ) -> Result<Vec<BackendJobAttemptRecord>> {
-        self.with_conn_mut(|conn| {
-            let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
-            // Preserve the last confirmed dispatch fact. An unbound claim may
-            // already own an external allocation; the Server reconciles its ID.
+        self.with_conn(|conn| {
+            // A deferred read snapshot keeps identities and records consistent
+            // without reserving the writer or changing the last dispatch fact.
+            let tx = conn.unchecked_transaction()?;
             let mut statement = tx.prepare(
-                "SELECT a.job_id, a.attempt_id FROM backend_job_attempts a JOIN backend_jobs j USING(workspace_id,job_id) WHERE a.workspace_id = ?1 AND a.state IN ('reserved', 'dispatching', 'dispatched') AND a.attempt=j.current_attempt AND j.state='pending' AND j.input_disposition='available' AND NOT EXISTS(SELECT 1 FROM backend_job_upgrade_history h WHERE h.workspace_id=a.workspace_id AND h.job_id=a.job_id) ORDER BY a.created_at LIMIT ?2",
+                "SELECT a.job_id, a.attempt_id FROM backend_job_attempts a JOIN backend_jobs j USING(workspace_id,job_id) WHERE a.workspace_id = ?1 AND (?3 IS NULL OR (a.job_id,a.attempt_id) > (?3,?4)) AND a.state IN ('reserved', 'dispatching', 'dispatched') AND a.attempt=j.current_attempt AND j.state='pending' AND j.input_disposition='available' AND NOT EXISTS(SELECT 1 FROM backend_job_upgrade_history h WHERE h.workspace_id=a.workspace_id AND h.job_id=a.job_id) ORDER BY a.job_id, a.attempt_id LIMIT ?2",
             )?;
-            let identities = statement.query_map(params![workspace_id, limit as i64], |row| {
-                Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
-            })?.collect::<rusqlite::Result<Vec<_>>>()?;
+            // Identities remain stable when earlier owners update state or
+            // timestamps. Reading a page never grants allocation/Submit authority.
+            let limit = limit.min(usize::from(crate::backend_job::ABSOLUTE_MAX_CONCURRENT_JOBS));
+            let identities = statement
+                .query_map(
+                    params![
+                        workspace_id,
+                        limit as i64,
+                        after.map(|cursor| cursor.job_id.as_str()),
+                        after.map(|cursor| cursor.attempt_id.as_str()),
+                    ],
+                    |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)),
+                )?
+                .collect::<rusqlite::Result<Vec<_>>>()?;
             drop(statement);
-            let attempts = identities.into_iter().map(|(job_id, attempt_id)| {
-                read_backend_job_attempt(&tx, workspace_id, &job_id, &attempt_id)?
-                    .ok_or_else(|| Error::Store("recoverable Backend Job attempt disappeared".to_string()))
-            }).collect::<Result<Vec<_>>>()?;
+            let attempts = identities
+                .into_iter()
+                .map(|(job_id, attempt_id)| {
+                    read_backend_job_attempt(&tx, workspace_id, &job_id, &attempt_id)?.ok_or_else(|| {
+                        Error::Store("recoverable Backend Job attempt disappeared".to_string())
+                    })
+                })
+                .collect::<Result<Vec<_>>>()?;
             tx.commit()?;
             Ok(attempts)
         })