fix(job): retain source-aware cleanup storage recovery

Preserve typed removal storage causes across initial/resumed canonical cleanup and retain exact Feature operation ownership after settlement. Keep public removal response adaptation at the outer boundary. Exercise actual Feature lock and catalog capacity outages beyond budget, original-child recovery, stale-child and permanent-constraint negatives.

Focused cleanup 7 passed; affected removal/fence boundaries 10 passed, 1 real Journal recording wait failed at first-body before checkpoint. G1 latency cause remains unresolved; no whole-suite/all-AC success or approval claimed. Root check, format, state names, diff and source hash checks passed. No operating migration, integration or close.

crates/workspace-server/src/server.rs

@@ -2212,6 +2212,97 @@ impl ServerApiContractService {
     }
 }
 
+#[derive(Debug, thiserror::Error)]
+enum WorkerRemovalFailure {
+    #[error("{0}")]
+    Internal(String),
+    #[error("{error}")]
+    Storage {
+        error: Error,
+        response: Option<worker::WorkspaceResponse>,
+    },
+}
+
+impl From<String> for WorkerRemovalFailure {
+    fn from(message: String) -> Self {
+        Self::Internal(message)
+    }
+}
+
+impl From<&str> for WorkerRemovalFailure {
+    fn from(message: &str) -> Self {
+        Self::Internal(message.to_owned())
+    }
+}
+
+impl From<Error> for WorkerRemovalFailure {
+    fn from(error: Error) -> Self {
+        Self::Storage {
+            error,
+            response: None,
+        }
+    }
+}
+
+impl WorkerRemovalFailure {
+    fn retention_response(
+        error: crate::retention::WorkerRetentionError,
+    ) -> std::result::Result<worker::WorkspaceResponse, Self> {
+        match error {
+            crate::retention::WorkerRetentionError::Store(error) => Err(Self::Storage {
+                error,
+                response: Some(worker_remove_error_response(
+                    StatusCode::SERVICE_UNAVAILABLE,
+                    "worker_removal_authority_unavailable",
+                    "Worker removal authority is unavailable; removal can be retried",
+                )),
+            }),
+            error => Ok(worker_retention_error_response(error)),
+        }
+    }
+
+    fn storage_response(
+        response: worker::WorkspaceResponse,
+        error: Error,
+    ) -> std::result::Result<worker::WorkspaceResponse, Self> {
+        match error {
+            Error::Io(_)
+            | Error::Sqlite(_)
+            | Error::SubjektivStorage(_)
+            | Error::DbCapacityExceeded
+            | Error::DbExecutorUnavailable
+            | Error::DbOperationOutcomeUnknown
+            | Error::Store(_) => Err(Self::Storage {
+                error,
+                response: Some(response),
+            }),
+            _ => Ok(response),
+        }
+    }
+
+    fn into_error(self, runtime_id: &str) -> Error {
+        match self {
+            Self::Internal(message) => Error::RuntimeOperationFailed {
+                runtime_id: runtime_id.to_owned(),
+                code: "backend_job_cleanup_removal_failed".to_string(),
+                message,
+            },
+            Self::Storage { error, .. } => error,
+        }
+    }
+
+    fn into_response(self) -> std::result::Result<worker::WorkspaceResponse, String> {
+        match self {
+            Self::Internal(message) => Err(message),
+            Self::Storage {
+                response: Some(response),
+                ..
+            } => Ok(response),
+            Self::Storage { error, .. } => Err(error.to_string()),
+        }
+    }
+}
+
 #[derive(Clone)]
 struct WorkerRemovalService {
     workspace_id: String,
@@ -2376,7 +2467,7 @@ impl WorkerRemovalService {
         target: &RuntimeWorkerRef,
         mut prepared: crate::retention::PreparedWorkerRemoval,
         parent_workspace_deletion_operation_id: Option<&str>,
-    ) -> std::result::Result<worker::WorkspaceResponse, String> {
+    ) -> std::result::Result<worker::WorkspaceResponse, WorkerRemovalFailure> {
         if prepared.plan.session_disposition
             == worker_runtime::retention::SessionDisposition::Archive
         {
@@ -2385,11 +2476,14 @@ impl WorkerRemovalService {
             ) {
                 Ok(ack) => ack,
                 Err(error) => {
-                    return Ok(worker_remove_error_response(
+                    return WorkerRemovalFailure::storage_response(
-                        StatusCode::CONFLICT,
+                        worker_remove_error_response(
-                        "erased_removal_reconciliation_unconfirmed",
+                            StatusCode::CONFLICT,
-                        &error.to_string(),
+                            "erased_removal_reconciliation_unconfirmed",
-                    ));
+                            &error.to_string(),
+                        ),
+                        error,
+                    );
                 }
             };
             if let Some(ack) = erasure_ack {
@@ -2401,7 +2495,7 @@ impl WorkerRemovalService {
                             .store
                             .audit_erased_worker_removal_receipt(&prepared, &ack, &result)
                         {
-                            return Ok(worker_retention_error_response(error));
+                            return WorkerRemovalFailure::retention_response(error);
                         }
                         return match self.worker_projection.publish_ordered(|| {
                             self.store
@@ -2414,7 +2508,7 @@ impl WorkerRemovalService {
                                 .map(|commit| ((), commit.catalog))
                         }) {
                             Ok(()) => Ok(worker_remove_success_response(target)),
-                            Err(error) => Ok(worker_retention_error_response(error)),
+                            Err(error) => WorkerRemovalFailure::retention_response(error),
                         };
                     }
                     Err(_) => {
@@ -2434,7 +2528,7 @@ impl WorkerRemovalService {
                             .store
                             .supersede_erased_worker_removal(&prepared, &ack, &inventory)
                         {
-                            return Ok(worker_retention_error_response(error));
+                            return WorkerRemovalFailure::retention_response(error);
                         }
                         let request = crate::retention::WorkerRemovalPlanRequest {
                             workspace_id: self.workspace_id.clone(),
@@ -2444,7 +2538,7 @@ impl WorkerRemovalService {
                         };
                         let plan = match self.store.plan_worker_removal(&request, &inventory) {
                             Ok(plan) => plan,
-                            Err(error) => return Ok(worker_retention_error_response(error)),
+                            Err(error) => return WorkerRemovalFailure::retention_response(error),
                         };
                         prepared = match self.store.prepare_worker_removal_execution(
                             &self.workspace_id,
@@ -2452,7 +2546,7 @@ impl WorkerRemovalService {
                             &plan.input_fingerprint,
                         ) {
                             Ok(prepared) => prepared,
-                            Err(error) => return Ok(worker_retention_error_response(error)),
+                            Err(error) => return WorkerRemovalFailure::retention_response(error),
                         };
                         if let Some(parent) = parent_workspace_deletion_operation_id {
                             self.store
@@ -2460,18 +2554,21 @@ impl WorkerRemovalService {
                                     parent,
                                     &prepared.plan.operation_id,
                                 )
-                                .map_err(|error| error.to_string())?;
+                                .map_err(WorkerRemovalFailure::from)?;
                         }
                     }
                 }
             }
         }
         if let Err(error) = subjektiv_journal::validate_removal_retention(self, target, &prepared) {
-            return Ok(worker_remove_error_response(
+            return WorkerRemovalFailure::storage_response(
-                StatusCode::CONFLICT,
+                worker_remove_error_response(
-                "journal_retention_unconfirmed",
+                    StatusCode::CONFLICT,
-                &error.to_string(),
+                    "journal_retention_unconfirmed",
-            ));
+                    &error.to_string(),
+                ),
+                error,
+            );
         }
         let result =
             match runtime.execute_worker_retention(target, prepared.runtime_request.clone()) {
@@ -2500,11 +2597,14 @@ impl WorkerRemovalService {
             }) {
                 Ok(result) => result,
                 Err(error) => {
-                    return Ok(worker_remove_error_response(
+                    return WorkerRemovalFailure::storage_response(
-                        StatusCode::CONFLICT,
+                        worker_remove_error_response(
-                        "subject_removal_completion_unconfirmed",
+                            StatusCode::CONFLICT,
-                        &error.to_string(),
+                            "subject_removal_completion_unconfirmed",
-                    ));
+                            &error.to_string(),
+                        ),
+                        error,
+                    );
                 }
             };
         match commit_result {
@@ -2512,7 +2612,7 @@ impl WorkerRemovalService {
                 subjektiv_journal::after_worker_removal(self, &result);
                 Ok(worker_remove_success_response(target))
             }
-            Err(error) => Ok(worker_retention_error_response(error)),
+            Err(error) => WorkerRemovalFailure::retention_response(error),
         }
     }
 
@@ -2653,6 +2753,7 @@ impl WorkerRemovalService {
             None,
         )
         .await
+        .or_else(WorkerRemovalFailure::into_response)
     }
 
     /// Restarted Job cleanup may resume only its already-observed canonical
@@ -2665,7 +2766,7 @@ impl WorkerRemovalService {
         reason: &str,
         parent_workspace_deletion_operation_id: Option<&str>,
         job_cleanup_owner: Option<&Mutex<Option<(String, String)>>>,
-    ) -> std::result::Result<worker::WorkspaceResponse, String> {
+    ) -> std::result::Result<worker::WorkspaceResponse, WorkerRemovalFailure> {
         let remove_lock = {
             let mut locks = self
                 .worker_remove_locks
@@ -2692,7 +2793,14 @@ impl WorkerRemovalService {
         let prepared = self
             .store
             .recover_worker_removal_execution(&self.workspace_id, &target)
-            .map_err(|_| "Worker removal recovery authority is unavailable".to_string())?;
+            .map_err(|error| match error {
+                crate::retention::WorkerRetentionError::Store(error) => {
+                    WorkerRemovalFailure::from(error)
+                }
+                _ => WorkerRemovalFailure::Internal(
+                    "Worker removal recovery authority is unavailable".to_string(),
+                ),
+            })?;
         let expected_removal = job_cleanup_owner
             .map(|owner| {
                 owner
@@ -2746,7 +2854,7 @@ impl WorkerRemovalService {
                             .recover_succeeded_worker_removal_catalog(&self.workspace_id, target)
                             .map(|commit| ((), commit))
                     })
-                    .map_err(|error| error.to_string())?;
+                    .map_err(WorkerRemovalFailure::from)?;
                 return Ok(worker_remove_success_response(target));
             }
             let prepared = if matches!(
@@ -2760,7 +2868,7 @@ impl WorkerRemovalService {
                     &prepared.plan.input_fingerprint,
                 ) {
                     Ok(prepared) => prepared,
-                    Err(error) => return Ok(worker_retention_error_response(error)),
+                    Err(error) => return WorkerRemovalFailure::retention_response(error),
                 }
             } else {
                 prepared
@@ -2771,7 +2879,7 @@ impl WorkerRemovalService {
                         parent_operation_id,
                         &prepared.plan.operation_id,
                     )
-                    .map_err(|error| error.to_string())?;
+                    .map_err(WorkerRemovalFailure::from)?;
             }
             if close_worker_workdir_sessions(&self.workdir_sessions, &target)
                 .await
@@ -2789,27 +2897,26 @@ impl WorkerRemovalService {
                     "Worker Workdir sessions could not be closed; removal can be retried",
                 ));
             }
-            if self
+            if let Err(error) = self.store.detach_worker_workdir(
-                .store
+                &self.workspace_id,
-                .detach_worker_workdir(
+                &target,
-                    &self.workspace_id,
+                None,
-                    &target,
+                &Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true),
-                    None,
+            ) {
-                    &Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true),
-                )
-                .is_err()
-            {
                 let _ = self.store.fail_worker_removal(
                     &self.workspace_id,
                     &prepared.plan.operation_id,
                     &prepared.plan.input_fingerprint,
                     "workdir_attachment_release_failed",
                 );
-                return Ok(worker_remove_error_response(
+                return WorkerRemovalFailure::storage_response(
-                    StatusCode::SERVICE_UNAVAILABLE,
+                    worker_remove_error_response(
-                    "attachment_release_failed",
+                        StatusCode::SERVICE_UNAVAILABLE,
-                    "Worker Workdir attachment could not be released; removal can be retried",
+                        "attachment_release_failed",
-                ));
+                        "Worker Workdir attachment could not be released; removal can be retried",
+                    ),
+                    error,
+                );
             }
             return self
                 .resume_worker_retention(
@@ -2843,11 +2950,14 @@ impl WorkerRemovalService {
         {
             Ok(proof) => proof,
             Err(error) => {
-                return Ok(worker_remove_error_response(
+                return WorkerRemovalFailure::storage_response(
-                    StatusCode::CONFLICT,
+                    worker_remove_error_response(
-                    "journal_handoff_incomplete",
+                        StatusCode::CONFLICT,
-                    &error.to_string(),
+                        "journal_handoff_incomplete",
-                ));
+                        &error.to_string(),
+                    ),
+                    error,
+                );
             }
         };
         let inventory = match runtime.worker_retention_inventory(&target) {
@@ -2873,7 +2983,7 @@ impl WorkerRemovalService {
             self.store.plan_worker_removal(&request, &inventory)
         } {
             Ok(plan) => plan,
-            Err(error) => return Ok(worker_retention_error_response(error)),
+            Err(error) => return WorkerRemovalFailure::retention_response(error),
         };
         if let Some(owner) = job_cleanup_owner {
             *owner
@@ -2887,7 +2997,7 @@ impl WorkerRemovalService {
             &plan.input_fingerprint,
         ) {
             Ok(prepared) => prepared,
-            Err(error) => return Ok(worker_retention_error_response(error)),
+            Err(error) => return WorkerRemovalFailure::retention_response(error),
         };
 
         if let Some(parent_operation_id) = parent_workspace_deletion_operation_id {
@@ -2896,7 +3006,7 @@ impl WorkerRemovalService {
                     parent_operation_id,
                     &prepared.plan.operation_id,
                 )
-                .map_err(|error| error.to_string())?;
+                .map_err(WorkerRemovalFailure::from)?;
         }
         if close_worker_workdir_sessions(&self.workdir_sessions, &target)
             .await
@@ -2915,7 +3025,7 @@ impl WorkerRemovalService {
             ));
         }
 
-        if let Err(_) = self.store.detach_worker_workdir(
+        if let Err(error) = self.store.detach_worker_workdir(
             &self.workspace_id,
             &target,
             None,
@@ -2927,19 +3037,25 @@ impl WorkerRemovalService {
                 &plan.input_fingerprint,
                 "workdir_attachment_release_failed",
             );
-            return Ok(worker_remove_error_response(
+            return WorkerRemovalFailure::storage_response(
-                StatusCode::SERVICE_UNAVAILABLE,
+                worker_remove_error_response(
-                "attachment_release_failed",
+                    StatusCode::SERVICE_UNAVAILABLE,
-                "Worker Workdir attachment could not be released; removal can be retried",
+                    "attachment_release_failed",
-            ));
+                    "Worker Workdir attachment could not be released; removal can be retried",
+                ),
+                error,
+            );
         }
 
         if let Err(error) = subjektiv_journal::validate_removal_retention(self, target, &prepared) {
-            return Ok(worker_remove_error_response(
+            return WorkerRemovalFailure::storage_response(
-                StatusCode::CONFLICT,
+                worker_remove_error_response(
-                "journal_retention_unconfirmed",
+                    StatusCode::CONFLICT,
-                &error.to_string(),
+                    "journal_retention_unconfirmed",
-            ));
+                    &error.to_string(),
+                ),
+                error,
+            );
         }
         let retention_result =
             match runtime.execute_worker_retention(&target, prepared.runtime_request.clone()) {
@@ -2979,11 +3095,14 @@ impl WorkerRemovalService {
         ) {
             Ok(result) => result,
             Err(error) => {
-                return Ok(worker_remove_error_response(
+                return WorkerRemovalFailure::storage_response(
-                    StatusCode::CONFLICT,
+                    worker_remove_error_response(
-                    "subject_removal_completion_unconfirmed",
+                        StatusCode::CONFLICT,
-                    &error.to_string(),
+                        "subject_removal_completion_unconfirmed",
-                ));
+                        &error.to_string(),
+                    ),
+                    error,
+                );
             }
         };
         match commit_result {
@@ -2995,7 +3114,7 @@ impl WorkerRemovalService {
                     &plan.input_fingerprint,
                     "metadata_commit_failed",
                 );
-                return Ok(worker_retention_error_response(error));
+                return WorkerRemovalFailure::retention_response(error);
             }
         };
         subjektiv_journal::after_worker_removal(self, &retention_result);
@@ -5469,22 +5588,39 @@ impl WorkspaceApi {
             Error::Store("claimed Backend Job Worker cleanup lost its binding".to_string())
         })?;
 
-        let fail = |category: &str, detail: String| -> Result<()> {
+        let record_failure = |category: &str, detail: &str| -> Result<()> {
             self.store.finish_backend_job_worker_cleanup_owned(
                 &self.config.workspace_id,
                 &history,
                 &attempt,
                 BackendJobWorkerCleanupState::Failed,
                 Some(category),
-                Some(&detail),
+                Some(detail),
                 &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true),
             )?;
+            Ok(())
+        };
+        let fail = |category: &str, detail: String| -> Result<()> {
+            record_failure(category, &detail)?;
             Err(Error::RuntimeOperationFailed {
                 runtime_id: worker.runtime_id.clone(),
                 code: category.to_string(),
                 message: detail,
             })
         };
+        let fail_with_error = |category: &str, error: Error| -> Result<()> {
+            if let Err(diagnostic) = record_failure(category, &error.to_string()) {
+                tracing::warn!(%error, %diagnostic, "Cleanup failure diagnosis was not confirmed");
+                // Keep transient storage ownership even when a definitive operation
+                // failure could not be diagnosed durably. Neither error proves cleanup.
+                if !backend_job_recovery::transient_storage(&error)
+                    && backend_job_recovery::transient_storage(&diagnostic)
+                {
+                    return Err(diagnostic);
+                }
+            }
+            Err(error)
+        };
         let complete = || -> Result<()> {
             self.require_backend_job_attempt_effects_ended(&attempt)?;
             let journal_job = self
@@ -5495,7 +5631,10 @@ impl WorkspaceApi {
                 let removal = self
                     .config_store
                     .recover_worker_removal_execution(self.workspace_id(), &worker)
-                    .map_err(|e| Error::Store(e.to_string()))?;
+                    .map_err(|error| match error {
+                        crate::retention::WorkerRetentionError::Store(error) => error,
+                        error => Error::Store(error.to_string()),
+                    })?;
                 if !removal.is_some_and(|p| {
                     p.plan.state == crate::retention::WorkerRemovalPlanState::Succeeded
                         && p.runtime_request.session_disposition
@@ -5552,7 +5691,7 @@ impl WorkspaceApi {
                 Err(error) => return fail("backend_job_cleanup_stop_unconfirmed", error.message()),
             }
             if let Err(error) = self.fence_backend_job_effects(&attempt, drain).await {
-                return fail("backend_job_cleanup_effects_unconfirmed", error.to_string());
+                return fail_with_error("backend_job_cleanup_effects_unconfirmed", error);
             }
         }
 
@@ -5564,15 +5703,15 @@ impl WorkspaceApi {
                 None,
                 Some(removal_owner),
             )
-            .await
+            .await;
-            .map_err(|message| Error::RuntimeOperationFailed {
-                runtime_id: worker.runtime_id.clone(),
-                code: "backend_job_cleanup_removal_failed".to_string(),
-                message,
-            });
         let response = match response {
             Ok(response) => response,
-            Err(error) => return fail("backend_job_cleanup_removal_failed", error.to_string()),
+            Err(error) => {
+                return fail_with_error(
+                    "backend_job_cleanup_removal_failed",
+                    error.into_error(&worker.runtime_id),
+                );
+            }
         };
         if response.status == StatusCode::OK.as_u16() {
             return complete();

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

@@ -33,6 +33,10 @@ pub(super) struct RecoveryOwners {
     #[cfg(test)]
     pub(super) nonjournal_lookup_pending: tokio::sync::Notify,
     #[cfg(test)]
+    pub(super) cleanup_storage_settlement: tokio::sync::Notify,
+    #[cfg(test)]
+    pub(super) cleanup_stopped: tokio::sync::Notify,
+    #[cfg(test)]
     pub(super) cleanup_before_claim: Mutex<Option<Arc<tokio::sync::Semaphore>>>,
     #[cfg(test)]
     pub(super) cleanup_before_writer_fence: tokio::sync::Notify,
@@ -205,7 +209,7 @@ fn transient_subjektiv_storage(error: &Error) -> bool {
     }
 }
 
-fn transient_storage(error: &Error) -> bool {
+pub(super) fn transient_storage(error: &Error) -> bool {
     transient_subjektiv_storage(error) || workspace_storage_failure(error)
 }
 
@@ -483,6 +487,8 @@ impl WorkspaceApi {
             if let Err(error) = api.run_backend_job_cleanup(expected.clone()).await {
                 tracing::warn!(job_id=%expected.job_id, attempt_id=%expected.attempt_id, %error,
                     "Job cleanup stopped; execution/input/resource fences remain retained");
+                #[cfg(test)]
+                api.backend_job_recovery.cleanup_stopped.notify_one();
             }
         });
         true
@@ -494,7 +500,7 @@ impl WorkspaceApi {
     ) -> Result<()> {
         // Pin the observed canonical operation across this owner's storage retries.
         let removal_owner = Arc::new(Mutex::new(None::<(String, String)>));
-        let mut failures = 0;
+        let mut failures = 0usize;
         loop {
             if self.backend_job_recovery.stopping.load(Ordering::Acquire) {
                 return Ok(());
@@ -611,7 +617,7 @@ impl WorkspaceApi {
                         .lock()
                         .unwrap()
                         .clone();
-                    runtime.block_on(async {
+                    let result = runtime.block_on(async {
                         #[cfg(test)]
                         if let Some(gate) = after_claim {
                             api.backend_job_recovery.cleanup_claimed_writer.notify_one();
@@ -625,9 +631,40 @@ impl WorkspaceApi {
                         }
                         api.cleanup_backend_job_attempt_worker(attempt, &removal_owner, &drain)
                             .await
-                    })
+                    });
+                    if let Err(error) = &result
+                        && transient_subjektiv_storage(error)
+                    {
+                        let history = api.store.list_backend_job_attempts(api.workspace_id(), &owner.job_id)?;
+                        let current = history.iter().find(|attempt| attempt.attempt_id == owner.attempt_id)
+                            .ok_or_else(|| Error::Store("Feature cleanup owner disappeared".into()))?;
+                        if JobHostOwner::try_from(current)? != JobHostOwner::try_from(&owner)?
+                            || !current.state.terminal()
+                        {
+                            return Err(Error::RepositoryConflict("Feature cleanup diagnostic owner changed".into()));
+                        }
+                        api.store.finish_backend_job_worker_cleanup_owned(
+                            api.workspace_id(), &history, current,
+                            BackendJobWorkerCleanupState::Failed,
+                            Some("backend_job_cleanup_feature_storage_unavailable"),
+                            Some(&format!("{error}; original cleanup owner retains input and canonical child; reconcile the same Feature operation after backoff, never infer termination or resubmit input")),
+                            &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true),
+                        )?;
+                    }
+                    result
                 })
                 .await;
+            #[cfg(test)]
+            if failures >= 5
+                && result.as_ref().is_err_and(|error| {
+                    transient_storage(error)
+                        || matches!(error, Error::RuntimeOperationFailed { .. })
+                })
+            {
+                self.backend_job_recovery
+                    .cleanup_storage_settlement
+                    .notify_one();
+            }
             match result {
                 Ok(()) => return Ok(()),
                 Err(error)
@@ -640,6 +677,16 @@ impl WorkspaceApi {
                         .await;
                     failures += 1;
                 }
+                Err(error) if transient_subjektiv_storage(&error) => {
+                    // This is the original Feature operation, not a healthy Workspace
+                    // probe. Re-enter its owner/history, retained-writer and canonical
+                    // child checks; no new execution or replacement removal is allowed.
+                    // Terminal Job deadlines do not authorize discarding failed cleanup.
+                    let delay =
+                        Duration::from_secs((1u64 << failures.saturating_sub(5).min(5)).min(30));
+                    failures = failures.saturating_add(1);
+                    self.backend_job_recovery.pause(delay).await;
+                }
                 Err(error) if workspace_storage_failure(&error) => {
                     tracing::warn!(attempt_id=%expected.attempt_id, %error, "Cleanup settlement budget exhausted; exact-owner storage probe only");
                     let mut delay = Duration::from_secs(1);

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

@@ -1111,15 +1111,15 @@ async fn backend_job_cleanup_commit_failure_retries_old_attempt_without_touching
     assert_eq!(blocked.failure_category.as_deref(), Some("model_failure"));
     assert_eq!(
         blocked.worker_cleanup_failure_category.as_deref(),
-        Some("backend_job_cleanup_removal_rejected")
+        Some("backend_job_cleanup_removal_failed")
     );
     assert!(
         blocked
             .worker_cleanup_failure_detail
             .as_deref()
             .unwrap()
-            .contains("worker_removal_authority_unavailable"),
+            .contains("injected cleanup catalog failure"),
-        "canonical catalog commit failure must remain a removal error: {blocked:?}"
+        "cleanup must retain the original catalog cause, not a flattened HTTP rejection: {blocked:?}"
     );
     let blocked_removal = api
         .config_store

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

@@ -1339,6 +1339,342 @@ fn lock_journal_domain(f: &RuntimeFixture) -> JournalCompletionSaveFault {
     JournalCompletionSaveFault(connection)
 }
 
+#[tokio::test]
+async fn cleanup_feature_storage_outage_beyond_budget_recovers_without_sweep() {
+    let f = RuntimeFixture::new().await;
+    let gate = Arc::new(tokio::sync::Semaphore::new(0));
+    *f.api
+        .backend_job_recovery
+        .cleanup_after_claim
+        .lock()
+        .unwrap() = Some(gate.clone());
+    let dispatched = f.dispatch_input();
+    let worker = dispatched.attempt.worker.as_ref().unwrap();
+    f.execution.take_inputs();
+    f.api
+        .finish_backend_job_attempt(
+            &f.request.job_id,
+            &dispatched.attempt.attempt_id,
+            BackendJobAttemptState::Failed,
+            "cancelled",
+            "retain the uncheckpointed original input",
+        )
+        .unwrap();
+    tokio::time::timeout(
+        std::time::Duration::from_secs(10),
+        f.api.backend_job_recovery.cleanup_claimed_writer.notified(),
+    )
+    .await
+    .unwrap();
+    let fault = lock_journal_domain(&f);
+    *f.api
+        .backend_job_recovery
+        .cleanup_after_claim
+        .lock()
+        .unwrap() = None;
+    gate.add_permits(1);
+    tokio::time::timeout(
+        std::time::Duration::from_secs(20),
+        f.api
+            .backend_job_recovery
+            .cleanup_storage_settlement
+            .notified(),
+    )
+    .await
+    .expect("real Feature storage fault did not exhaust initial cleanup settlement");
+    let blocked = current_journal_attempt(&f, &dispatched.attempt.attempt_id);
+    assert_eq!(blocked.state, BackendJobAttemptState::Failed);
+    assert_eq!(blocked.failure_category.as_deref(), Some("cancelled"));
+    assert_ne!(
+        blocked.worker_cleanup_state,
+        Some(BackendJobWorkerCleanupState::Completed)
+    );
+    assert!(
+        f.api
+            .backend_job_recovery
+            .cleanup_owner_ids()
+            .contains(&blocked.attempt_id),
+        "Feature outage exhausted cleanup retries and lost its autonomous owner"
+    );
+    assert_eq!(
+        blocked.worker_cleanup_failure_category.as_deref(),
+        Some("backend_job_cleanup_feature_storage_unavailable")
+    );
+    // The Workspace is healthy, but cannot certify the locked Feature operation.
+    assert!(
+        f.api
+            .store
+            .probe_backend_job_recovery(TEST_WORKSPACE_ID, &blocked)
+            .unwrap()
+    );
+    tokio::time::timeout(
+        std::time::Duration::from_secs(10),
+        f.api
+            .backend_job_recovery
+            .cleanup_storage_settlement
+            .notified(),
+    )
+    .await
+    .expect("source-aware cleanup reconciliation lost its owner after settlement");
+    assert_ne!(
+        current_journal_attempt(&f, &blocked.attempt_id).worker_cleanup_state,
+        Some(BackendJobWorkerCleanupState::Completed)
+    );
+    assert!(
+        f.api
+            .backend_job_recovery
+            .cleanup_owner_ids()
+            .contains(&blocked.attempt_id)
+    );
+    let original = f
+        .api
+        .config_store
+        .recover_worker_removal_execution(TEST_WORKSPACE_ID, worker)
+        .unwrap()
+        .unwrap();
+    drop(fault);
+    // Only release the original Feature writer lock; do not sweep, reopen or replay.
+    let cleaned = wait_for_backend_job_worker_cleanup(
+        &f.api,
+        &f.request.job_id,
+        &blocked.attempt_id,
+        BackendJobWorkerCleanupState::Completed,
+    )
+    .await;
+    assert_eq!(cleaned.worker, blocked.worker);
+    assert_eq!(cleaned.runtime_run_id, blocked.runtime_run_id);
+    assert_eq!(cleaned.failure_category, blocked.failure_category);
+    let after = f
+        .api
+        .config_store
+        .recover_worker_removal_execution(TEST_WORKSPACE_ID, worker)
+        .unwrap()
+        .unwrap();
+    assert_eq!(after.plan.operation_id, original.plan.operation_id);
+    assert_eq!(
+        after.plan.input_fingerprint,
+        original.plan.input_fingerprint
+    );
+    assert!(f.execution.take_inputs().is_empty());
+    let saved = f
+        .api
+        .store
+        .get_backend_job(TEST_WORKSPACE_ID, &f.request.job_id)
+        .unwrap()
+        .unwrap();
+    assert_eq!(
+        saved.input_disposition,
+        crate::backend_job::JobInputDisposition::Available
+    );
+    assert!(f.store.journal_retention_required(&f.host).unwrap());
+    f.api.stop_backend_job_recovery().await;
+}
+
+struct CleanupCatalogCapacityFault {
+    store: Arc<SqliteWorkspaceStore>,
+    previous_limit: i64,
+}
+
+impl CleanupCatalogCapacityFault {
+    fn install(f: &RuntimeFixture) -> Self {
+        let previous_limit = f
+            .job_store
+            .with_conn(|connection| {
+                connection.execute_batch(
+                    "CREATE TABLE cleanup_capacity_fault(payload BLOB);
+                 CREATE TRIGGER fill_cleanup_catalog_capacity
+                 BEFORE INSERT ON worker_registry_projection_removals
+                 BEGIN INSERT INTO cleanup_capacity_fault VALUES(zeroblob(4194304)); END;",
+                )?;
+                let previous: i64 =
+                    connection.query_row("PRAGMA max_page_count", [], |row| row.get(0))?;
+                let pages: i64 = connection.query_row("PRAGMA page_count", [], |row| row.get(0))?;
+                // Small owner/diagnostic writes fit; the catalog commit's trigger does not.
+                connection.pragma_update(None, "max_page_count", pages + 32)?;
+                let error = connection
+                    .execute(
+                        "INSERT INTO cleanup_capacity_fault VALUES(zeroblob(4194304))",
+                        [],
+                    )
+                    .unwrap_err();
+                assert!(matches!(error, rusqlite::Error::SqliteFailure(error, _)
+                if error.code == rusqlite::ErrorCode::DiskFull));
+                Ok(previous)
+            })
+            .unwrap();
+        Self {
+            store: f.job_store.clone(),
+            previous_limit,
+        }
+    }
+}
+
+impl Drop for CleanupCatalogCapacityFault {
+    fn drop(&mut self) {
+        self.store.with_conn(|connection| {
+            connection.pragma_update(None, "max_page_count", self.previous_limit)?;
+            connection.execute_batch("DROP TRIGGER fill_cleanup_catalog_capacity; DROP TABLE cleanup_capacity_fault;")?;
+            Ok(())
+        }).unwrap();
+    }
+}
+
+#[tokio::test]
+async fn cleanup_catalog_storage_outage_beyond_budget_recovers_original_child() {
+    cleanup_catalog_storage_reconciliation(false).await;
+}
+
+#[tokio::test]
+async fn cleanup_catalog_storage_recovery_rejects_changed_canonical_authority() {
+    cleanup_catalog_storage_reconciliation(true).await;
+}
+
+async fn cleanup_catalog_storage_reconciliation(change_authority: bool) {
+    let f = RuntimeFixture::new().await;
+    let dispatched = f.dispatch_input();
+    let worker = dispatched.attempt.worker.as_ref().unwrap();
+    f.execution.take_inputs();
+    let fault = CleanupCatalogCapacityFault::install(&f);
+    f.api
+        .finish_backend_job_attempt(
+            &f.request.job_id,
+            &dispatched.attempt.attempt_id,
+            BackendJobAttemptState::Failed,
+            "cancelled",
+            "retain the uncheckpointed original input",
+        )
+        .unwrap();
+    tokio::time::timeout(
+        std::time::Duration::from_secs(20),
+        f.api
+            .backend_job_recovery
+            .cleanup_storage_settlement
+            .notified(),
+    )
+    .await
+    .expect("real catalog capacity fault did not exhaust initial cleanup settlement");
+    let blocked = current_journal_attempt(&f, &dispatched.attempt.attempt_id);
+    assert_eq!(blocked.failure_category.as_deref(), Some("cancelled"));
+    assert_ne!(
+        blocked.worker_cleanup_state,
+        Some(BackendJobWorkerCleanupState::Completed)
+    );
+    assert!(
+        f.api
+            .backend_job_recovery
+            .cleanup_owner_ids()
+            .contains(&blocked.attempt_id),
+        "catalog capacity outage lost its autonomous cleanup owner"
+    );
+    let original = f
+        .api
+        .config_store
+        .recover_worker_removal_execution(TEST_WORKSPACE_ID, worker)
+        .unwrap()
+        .unwrap();
+    assert_eq!(
+        original.prior_failure_category.as_deref(),
+        Some("metadata_commit_failed")
+    );
+    assert!(
+        f.api
+            .require_backend_job_attempt_effects_ended(&blocked)
+            .is_ok()
+    );
+    if change_authority {
+        // Simulate a canonical child's invalidation while storage is unavailable.
+        // Normal pin updates are fenced during an admitted removal and cannot do this.
+        f.api.config_store.with_conn(|connection| {
+            assert_eq!(connection.execute("UPDATE worker_removal_operations SET state='stale', failure_category='fixture_child_invalidated' WHERE workspace_id=?1 AND operation_id=?2 AND input_fingerprint=?3",
+                rusqlite::params![TEST_WORKSPACE_ID, original.plan.operation_id, original.plan.input_fingerprint])?, 1);
+            Ok(())
+        }).unwrap();
+    }
+    drop(fault);
+    // Restore only the failed catalog write's capacity; no sweep/reopen/result replay.
+    if change_authority {
+        tokio::time::timeout(
+            std::time::Duration::from_secs(20),
+            f.api.backend_job_recovery.cleanup_stopped.notified(),
+        )
+        .await
+        .unwrap_or_else(|error| panic!("changed removal authority did not reach bounded rejection: {error}; current={:?}; owners={:?}; removal={:?}",
+            current_journal_attempt(&f, &blocked.attempt_id), f.api.backend_job_recovery.cleanup_owner_ids(),
+            f.api.config_store.recover_worker_removal_execution(TEST_WORKSPACE_ID, worker)));
+        assert!(
+            !f.api
+                .backend_job_recovery
+                .cleanup_owner_ids()
+                .contains(&blocked.attempt_id)
+        );
+        let rejected = current_journal_attempt(&f, &blocked.attempt_id);
+        assert_eq!(
+            rejected.worker_cleanup_state,
+            Some(BackendJobWorkerCleanupState::Failed)
+        );
+        assert_eq!(rejected.failure_category, blocked.failure_category);
+        f.api.config_store.with_conn(|connection| {
+            let count: i64 = connection.query_row("SELECT COUNT(*) FROM worker_removal_operations WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3", rusqlite::params![TEST_WORKSPACE_ID, worker.runtime_id, worker.worker_id], |row| row.get(0))?;
+            assert_eq!(count, 1);
+            Ok(())
+        }).unwrap();
+        assert!(f.execution.take_inputs().is_empty());
+        assert_eq!(
+            f.api
+                .store
+                .get_backend_job(TEST_WORKSPACE_ID, &f.request.job_id)
+                .unwrap()
+                .unwrap()
+                .input_disposition,
+            crate::backend_job::JobInputDisposition::Available
+        );
+        assert!(f.store.journal_retention_required(&f.host).unwrap());
+        f.api.stop_backend_job_recovery().await;
+        return;
+    }
+    let cleaned = wait_for_backend_job_worker_cleanup(
+        &f.api,
+        &f.request.job_id,
+        &blocked.attempt_id,
+        BackendJobWorkerCleanupState::Completed,
+    )
+    .await;
+    let after = f
+        .api
+        .config_store
+        .recover_worker_removal_execution(TEST_WORKSPACE_ID, worker)
+        .unwrap()
+        .unwrap();
+    assert_eq!(after.plan.operation_id, original.plan.operation_id);
+    assert_eq!(
+        after.plan.input_fingerprint,
+        original.plan.input_fingerprint
+    );
+    assert_eq!(cleaned.worker, blocked.worker);
+    assert_eq!(cleaned.runtime_run_id, blocked.runtime_run_id);
+    assert_eq!(cleaned.failure_category, blocked.failure_category);
+    assert!(f.execution.take_inputs().is_empty());
+    assert_eq!(
+        f.api
+            .store
+            .list_backend_job_attempts(TEST_WORKSPACE_ID, &f.request.job_id)
+            .unwrap()
+            .len(),
+        1
+    );
+    assert_eq!(
+        f.api
+            .store
+            .get_backend_job(TEST_WORKSPACE_ID, &f.request.job_id)
+            .unwrap()
+            .unwrap()
+            .input_disposition,
+        crate::backend_job::JobInputDisposition::Available
+    );
+    assert!(f.store.journal_retention_required(&f.host).unwrap());
+    f.api.stop_backend_job_recovery().await;
+}
+
 #[tokio::test]
 async fn domain_storage_lock_beyond_initial_settlement_recovers_original_journal_attempt() {
     for already_prepared in [false, true] {
@@ -2377,7 +2713,8 @@ async fn saved_checkpoint_recovery_completes_without_proving_failed_cleanup_ende
         let dispatched = f.dispatch_input();
         let worker = dispatched.attempt.worker.as_ref().unwrap();
 
-        // Inject a recoverable cleanup commit failure before checkpoint publication.
+        // A deterministic SQL constraint failure needs explicit fault correction,
+        // not autonomous transient-storage probing. Preserve the original child.
         // Pin/unpin changes Worker authority and deliberately makes the original child stale;
         // that negative contract is covered by the canonical-cleanup staleness tests.
         f.api
@@ -2388,6 +2725,14 @@ async fn saved_checkpoint_recovery_completes_without_proving_failed_cleanup_ende
                  BEFORE INSERT ON worker_registry_projection_removals
                  BEGIN SELECT RAISE(ABORT, 'injected checkpoint cleanup failure'); END;",
                 )?;
+                let error = conn
+                    .execute(
+                        "INSERT INTO worker_registry_projection_removals DEFAULT VALUES",
+                        [],
+                    )
+                    .unwrap_err();
+                assert!(matches!(error, rusqlite::Error::SqliteFailure(error, _)
+                    if error.extended_code == rusqlite::ffi::SQLITE_CONSTRAINT_TRIGGER));
                 Ok(())
             })
             .unwrap();
@@ -2443,6 +2788,19 @@ async fn saved_checkpoint_recovery_completes_without_proving_failed_cleanup_ende
             f.execution.take_inputs().is_empty(),
             "saved checkpoint must not cause model rerun"
         );
+        tokio::time::timeout(
+            std::time::Duration::from_secs(15),
+            f.api.backend_job_recovery.cleanup_stopped.notified(),
+        )
+        .await
+        .expect("deterministic commit failure did not stop with its retained cause");
+        assert!(
+            !f.api
+                .backend_job_recovery
+                .cleanup_owner_ids()
+                .contains(&blocked.attempt_id),
+            "permanent constraint failure must not become an endless transient probe"
+        );
         f.api
             .config_store
             .with_conn(|conn| {