fix: reconcile unfinished retention before Subject deletion

crates/runtime-api/src/lib.rs

@@ -1219,8 +1219,34 @@ pub struct SubjectDeleteWorkerResponse {
     pub originals_absent: bool,
 }
 
+/// Additional erasure of a Subject Job's newly reconciled removal copies.
+/// Both identities are exact saved requests; this never changes the old receipt.
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(deny_unknown_fields)]
+pub struct SubjectJobRemovalPurgeRequest {
+    pub deletion: SubjectDeleteWorkerRequest,
+    pub removal: WorkerRetentionExecutionRequest,
+}
+
+/// Confirms current physical erasure under the permanent deletion fence.
+/// Operation identity belongs to `removal`, not the Subject deletion operation.
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(deny_unknown_fields)]
+pub struct SubjectJobRemovalPurgeResponse {
+    pub workspace_id: String,
+    pub worker_id: WorkerId,
+    pub operation_id: String,
+    pub input_fingerprint: String,
+}
+
 #[api(reqwest, axum)]
 pub trait RuntimeApi {
+    #[post("/v1/subject-job-removal-purge", status = 200, error_status = 400)]
+    async fn purge_subject_job_removal(
+        &self,
+        #[body] request: SubjectJobRemovalPurgeRequest,
+    ) -> Result<SubjectJobRemovalPurgeResponse, RuntimeApiError>;
+
     #[post("/v1/subject-delete-worker", status = 200, error_status = 400)]
     async fn subject_delete_worker(
         &self,
@@ -1787,6 +1813,18 @@ mod tests {
         ) -> Result<SessionMaterialPinResponse, RuntimeApiError> {
             Err(test_error(501))
         }
+        async fn purge_subject_job_removal(
+            &self,
+            request: SubjectJobRemovalPurgeRequest,
+        ) -> Result<SubjectJobRemovalPurgeResponse, RuntimeApiError> {
+            Ok(SubjectJobRemovalPurgeResponse {
+                workspace_id: request.removal.workspace_id,
+                worker_id: request.removal.worker_id,
+                operation_id: request.removal.operation_id,
+                input_fingerprint: request.removal.input_fingerprint,
+            })
+        }
+
         async fn subject_delete_worker(
             &self,
             request: SubjectDeleteWorkerRequest,
@@ -1989,9 +2027,35 @@ mod tests {
             .subject_delete_worker(deletion.clone())
             .await
             .unwrap();
-        let mut expected = serde_json::to_value(deletion).unwrap();
+        let mut expected = serde_json::to_value(&deletion).unwrap();
         expected["originals_absent"] = true.into();
         assert_eq!(serde_json::to_value(receipt).unwrap(), expected);
+        let purge = SubjectJobRemovalPurgeRequest {
+            removal: WorkerRetentionExecutionRequest {
+                operation_id: "old-removal".into(),
+                input_fingerprint: "old-removal-fingerprint".into(),
+                archive_id: Some("old-archive".into()),
+                workspace_id: deletion.workspace_id.clone(),
+                source_runtime_id: "runtime-a".into(),
+                worker_id: deletion.worker_id,
+                source_created_at: "2026-01-01T00:00:00Z".into(),
+                removed_at: "2026-01-02T00:00:00Z".into(),
+                effective_profile: None,
+                retention_class: None,
+                policy_id: "old-policy".into(),
+                session_disposition: SessionDisposition::Archive,
+                diagnostics_disposition: DiagnosticsDisposition::Retain,
+            },
+            deletion,
+        };
+        let receipt = client
+            .purge_subject_job_removal(purge.clone())
+            .await
+            .unwrap();
+        assert_eq!(receipt.workspace_id, purge.removal.workspace_id);
+        assert_eq!(receipt.worker_id, purge.removal.worker_id);
+        assert_eq!(receipt.operation_id, purge.removal.operation_id);
+        assert_eq!(receipt.input_fingerprint, purge.removal.input_fingerprint);
         let summary = client.runtime_summary().await.unwrap();
         assert_eq!(summary.runtime.display_name.as_deref(), Some("round-trip"));
         let workers = client

crates/subjektiv/README.md

@@ -35,6 +35,14 @@ originals and pre-existing archives have independent retention lifecycles: the W
 Host archives source originals rather than erasing them. Those archives may contain
 previously injected Memory; Workdirs, Tickets, exports and external provider copies are
 also outside Subject content erasure. These boundaries must be explained to the user.
+An unfinished, previously authorized Worker retention operation is snapshotted separately
+from prior successful removals. Its exact original request is reconciled under the permanent
+fence; its disposition, fingerprint and canonical result are never rewritten. Body originals
+follow that earlier independent request. For a Subject Job, any archive/diagnostic copies
+produced by the unfinished operation require a distinct Runtime physical-erasure acknowledgement
+before deletion can complete. Lost responses or failed receipt commits retain the same request
+and incomplete state across restart. Successes that already existed at deletion start remain
+under their independent archive lifecycle; a later success is not inserted into that snapshot.
 The domain rechecks after out-of-DB capture callbacks and before returning Job read
 responses; already returned DTOs cannot be revoked by SQLite. Host lifecycle locking
 must cover delivery/reinjection of previously fetched content as well as finish.

crates/worker-runtime/src/http_server.rs

@@ -2104,7 +2104,9 @@ fn required_runtime_permission(method: &Method, path: &str) -> Option<&'static s
     }
     if matches!(
         path,
-        "/v1/subject-create-recovery" | "/v1/subject-delete-worker"
+        "/v1/subject-create-recovery"
+            | "/v1/subject-delete-worker"
+            | "/v1/subject-job-removal-purge"
     ) && *method == Method::POST
     {
         return Some("workers:delete");

crates/worker-runtime/src/http_server/runtime_management_api.rs

@@ -406,6 +406,18 @@ impl runtime_api::RuntimeApi for RuntimeManagementApi {
         }
     }
 
+    async fn purge_subject_job_removal(
+        &self,
+        request: runtime_api::SubjectJobRemovalPurgeRequest,
+    ) -> Result<runtime_api::SubjectJobRemovalPurgeResponse, runtime_api::RuntimeApiError> {
+        let scope = material_scope(&self.state, &request.deletion.workspace_id)?;
+        self.state
+            .runtime
+            .purge_subject_job_removal_scoped(&scope, request)
+            .map_err(RuntimeHttpRestError::runtime)
+            .map_err(api_error)
+    }
+
     async fn subject_delete_worker(
         &self,
         request: runtime_api::SubjectDeleteWorkerRequest,

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

@@ -3,6 +3,8 @@
 //! verifier are connected. Only controller execution is replaced by AcceptingBackend;
 //! no model, external server, user credentials or running Runtime is used.
 use super::*;
+#[path = "subject_job_removal_purge_tests.rs"]
+mod subject_job_removal_purge_tests;
 use crate::retention::{DiagnosticsDisposition, SessionDisposition, runtime_session_retention};
 use runtime_api::{
     SessionMaterialCatalogAvailability as Catalog, SessionMaterialReadAvailability as Read,

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

@@ -0,0 +1,117 @@
+//! Real filesystem failures and reopened durable fences for the additional purge.
+//! Uses the same production stores/router fixture as the copy-erasure suite.
+use super::*;
+
+#[test]
+fn subject_job_removal_purge_partial_physical_failure_keeps_fence_and_retry_erases_after_reopen() {
+    let f = Fixture::new();
+    let mut request = f.removal_purge_request(SessionDisposition::Archive);
+    request.removal.diagnostics_disposition = runtime_api::DiagnosticsDisposition::Retain;
+    f.complete_removal(&request);
+    let receipt_path = f.receipt_path();
+    let receipt = std::fs::read(&receipt_path).unwrap();
+    let session_root = f
+        .temp
+        .path()
+        .join("archives/workers/canonical-archive/session");
+    let parked = f.temp.path().join("interrupted-session-fixture");
+    std::fs::rename(&session_root, &parked).unwrap();
+    std::fs::write(&session_root, b"deterministic directory removal failure").unwrap();
+    assert!(
+        f.runtime
+            .purge_subject_job_removal_scoped(&f.scope, request.clone())
+            .is_err()
+    );
+    assert!(matches!(
+        runtime_session_retention(f.temp.path()).ensure_live(f.session),
+        Err(session_store::StoreError::SessionErased(_))
+    ));
+    assert!(
+        !f.temp
+            .path()
+            .join("archives/diagnostics/remove-source")
+            .exists()
+    );
+    assert_eq!(std::fs::read(&receipt_path).unwrap(), receipt);
+    std::fs::remove_file(&session_root).unwrap();
+    std::fs::rename(parked, &session_root).unwrap();
+    let Fixture {
+        temp,
+        runtime,
+        scope,
+        store,
+        session,
+        ..
+    } = f;
+    drop(store);
+    drop(runtime);
+    let runtime = Runtime::with_fs_store_and_execution_backend(
+        crate::FsRuntimeStoreOptions::new(temp.path()).with_runtime_id("runtime-test"),
+        Arc::new(AcceptingBackend),
+    )
+    .unwrap();
+    runtime.bind_runtime_identity("runtime-test").unwrap();
+    runtime
+        .purge_subject_job_removal_scoped(&scope, request)
+        .unwrap();
+    assert!(!session_root.exists());
+    assert!(matches!(
+        runtime_session_retention(temp.path()).lock_live(session),
+        Err(session_store::StoreError::SessionErased(_))
+    ));
+    assert_eq!(std::fs::read(receipt_path).unwrap(), receipt);
+}
+
+#[test]
+fn subject_job_removal_purge_unconfirmed_durable_deletion_fence_preserves_copies_until_reconciled()
+{
+    let f = Fixture::new();
+    let request = f.removal_purge_request(SessionDisposition::Archive);
+    f.complete_removal(&request);
+    let copies = f.copies();
+    let receipt_path = f.receipt_path();
+    let receipt = std::fs::read(&receipt_path).unwrap();
+    let Fixture {
+        temp,
+        runtime,
+        scope,
+        store,
+        ..
+    } = f;
+    drop(store);
+    drop(runtime);
+    // Frozen durable crash-window fixture: permanent admission exclusion was
+    // saved, but no confirmed execution-shutdown response was committed.
+    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;
+    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"),
+        Arc::new(AcceptingBackend),
+    )
+    .unwrap();
+    runtime.bind_runtime_identity("runtime-test").unwrap();
+    assert!(
+        runtime
+            .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);
+    runtime
+        .subject_delete_worker_scoped(&scope, request.deletion.clone())
+        .unwrap();
+    runtime
+        .purge_subject_job_removal_scoped(&scope, request)
+        .unwrap();
+    assert!(
+        !temp
+            .path()
+            .join("archives/workers/canonical-archive/session")
+            .exists()
+    );
+    assert_eq!(std::fs::read(receipt_path).unwrap(), receipt);
+}

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

@@ -0,0 +1,486 @@
+//! Additional removal-copy erasure uses the real Runtime, retention provider,
+//! Session tombstone, filesystem and signed HTTP router. Only execution is fake
+//! (the parent fixture's AcceptingBackend); no Backend recovery is simulated here.
+use super::*;
+#[path = "subject_job_removal_purge_recovery_tests.rs"]
+mod recovery;
+use runtime_api::{SubjectDeleteWorkerRequest, SubjectJobRemovalPurgeRequest};
+use std::collections::BTreeMap;
+use std::path::Path;
+
+fn tree(root: &Path) -> BTreeMap<PathBuf, Vec<u8>> {
+    fn walk(root: &Path, at: &Path, files: &mut BTreeMap<PathBuf, Vec<u8>>) {
+        if !at.exists() {
+            return;
+        }
+        for entry in std::fs::read_dir(at).unwrap() {
+            let entry = entry.unwrap();
+            if entry.file_type().unwrap().is_dir() {
+                walk(root, &entry.path(), files);
+            } else {
+                files.insert(
+                    entry.path().strip_prefix(root).unwrap().into(),
+                    std::fs::read(entry.path()).unwrap(),
+                );
+            }
+        }
+    }
+    let mut files = BTreeMap::new();
+    walk(root, root, &mut files);
+    files
+}
+
+impl Fixture {
+    fn deletion_request(&self) -> SubjectDeleteWorkerRequest {
+        SubjectDeleteWorkerRequest {
+            workspace_id: self.scope.workspace_id.clone(),
+            worker_id: self.worker.worker_id,
+            create_fingerprint: task_request("identity only").create_fingerprint,
+            operation_id: "subject-deletion".into(),
+        }
+    }
+
+    fn removal_purge_request(
+        &self,
+        disposition: SessionDisposition,
+    ) -> SubjectJobRemovalPurgeRequest {
+        let removal = self.retention_request(disposition);
+        SubjectJobRemovalPurgeRequest {
+            deletion: self.deletion_request(),
+            removal: serde_json::from_value(serde_json::to_value(removal).unwrap()).unwrap(),
+        }
+    }
+
+    fn complete_removal(
+        &self,
+        request: &SubjectJobRemovalPurgeRequest,
+    ) -> crate::retention::WorkerRetentionExecutionResult {
+        let diagnostics = self.root().parent().unwrap().join("runs/attempt");
+        std::fs::create_dir_all(&diagnostics).unwrap();
+        std::fs::write(diagnostics.join("worker.out.log"), b"Subject Job output").unwrap();
+        self.runtime
+            .subject_delete_worker_scoped(&self.scope, request.deletion.clone())
+            .unwrap();
+        let removal =
+            serde_json::from_value(serde_json::to_value(&request.removal).unwrap()).unwrap();
+        self.runtime.execute_worker_retention(&removal).unwrap()
+    }
+
+    fn receipt_path(&self) -> PathBuf {
+        self.temp
+            .path()
+            .join("retention/operations/remove-source.json")
+    }
+
+    fn copies(&self) -> BTreeMap<PathBuf, Vec<u8>> {
+        tree(&self.temp.path().join("archives"))
+    }
+
+    fn independent_worker(&self) -> WorkerRef {
+        let mut request = task_request("independent Worker");
+        request.config_bundle = None;
+        request.memory_settings.as_mut().unwrap().workspace_id = self.scope.workspace_id.clone();
+        self.runtime
+            .create_worker_scoped(&self.scope, request)
+            .unwrap()
+            .worker_ref
+    }
+
+    // Independent successful snapshot made before the Subject deletion starts.
+    fn prior_archive(&self) -> WorkerRef {
+        let worker = self.independent_worker();
+        let aggregate = self
+            .temp
+            .path()
+            .join("workers")
+            .join(worker.worker_id.to_string());
+        let session = session_store::new_session_id();
+        let segment = session_store::new_segment_id();
+        WorkerSessionStore::new(aggregate.join("session"))
+            .unwrap()
+            .create_segment(
+                session,
+                segment,
+                &[LogEntry::AnnotatedSegmentStart {
+                    ts: 1,
+                    session_id: session,
+                    system_prompt: None,
+                    config: Default::default(),
+                    history: vec![],
+                    forked_from: None,
+                    compacted_from: None,
+                }],
+            )
+            .unwrap();
+        WorkerAggregateStore::new(&aggregate, format!("worker-runtime-{}", worker.worker_id))
+            .unwrap()
+            .write(
+                &session_store::WorkerMetadata::new(
+                    format!("worker-runtime-{}", worker.worker_id),
+                    Some(session_store::WorkerActiveSegmentRef::active_segment(
+                        session, segment,
+                    )),
+                )
+                .with_workspace_id(&self.scope.workspace_id),
+            )
+            .unwrap();
+        self.runtime
+            .stop_worker_scoped(&self.scope, &worker, None)
+            .unwrap();
+        let mut removal = self.retention_request(SessionDisposition::Archive);
+        removal.worker_id = worker.worker_id;
+        removal.operation_id = "prior-removal".into();
+        removal.archive_id = Some("prior-archive".into());
+        removal.input_fingerprint = "prior-fingerprint".into();
+        self.runtime.execute_worker_retention(&removal).unwrap();
+        worker
+    }
+}
+
+#[test]
+fn subject_job_removal_purge_erases_copies_preserves_receipt_and_prior_success_after_reopen() {
+    for disposition in [SessionDisposition::Archive, SessionDisposition::Purge] {
+        let f = Fixture::new();
+        f.prior_archive();
+        let prior = tree(&f.temp.path().join("archives/workers/prior-archive"));
+        assert!(!prior.is_empty());
+        let unrelated = f.independent_worker();
+        let unrelated_root = f
+            .temp
+            .path()
+            .join("workers")
+            .join(unrelated.worker_id.to_string());
+        let unrelated_before = tree(&unrelated_root);
+        let mut request = f.removal_purge_request(disposition);
+        request.removal.diagnostics_disposition = runtime_api::DiagnosticsDisposition::Retain;
+        let original_result = f.complete_removal(&request);
+        let receipt = std::fs::read(f.receipt_path()).unwrap();
+        let archive = f.temp.path().join("archives/workers/canonical-archive");
+        let diagnostics = f.temp.path().join("archives/diagnostics/remove-source");
+        assert!(diagnostics.is_dir());
+        let manifest = (disposition == SessionDisposition::Archive)
+            .then(|| std::fs::read(archive.join("manifest.json")).unwrap());
+        let archive_staging = f
+            .temp
+            .path()
+            .join("archives/workers/.canonical-archive.tmp-123-1");
+        if let Some(manifest) = &manifest {
+            std::fs::create_dir_all(archive_staging.join("session/segments")).unwrap();
+            std::fs::write(archive_staging.join("manifest.json"), manifest).unwrap();
+            std::fs::write(
+                archive_staging.join("session/segments/interrupted.jsonl"),
+                b"interrupted Subject Job copy",
+            )
+            .unwrap();
+        }
+        // Diagnostics copying precedes publication of its manifest. Interrupted
+        // staging is still bound to the exact removal operation name.
+        let diagnostics_staging = f
+            .temp
+            .path()
+            .join("archives/diagnostics/.remove-source.tmp-123-2");
+        std::fs::create_dir_all(&diagnostics_staging).unwrap();
+        std::fs::write(
+            diagnostics_staging.join("worker.out.log"),
+            b"interrupted diagnostic copy",
+        )
+        .unwrap();
+        for _ in 0..2 {
+            let response = f
+                .runtime
+                .purge_subject_job_removal_scoped(&f.scope, request.clone())
+                .unwrap();
+            assert_eq!(response.operation_id, request.removal.operation_id);
+            assert_eq!(
+                response.input_fingerprint,
+                request.removal.input_fingerprint
+            );
+            assert_eq!(response.worker_id, request.removal.worker_id);
+            assert_eq!(response.workspace_id, request.removal.workspace_id);
+            assert!(!diagnostics.exists());
+            assert!(!diagnostics_staging.exists());
+            assert!(!archive_staging.join("session").exists());
+            assert!(!archive.join("session").exists());
+            assert_eq!(std::fs::read(f.receipt_path()).unwrap(), receipt);
+            if let Some(manifest) = &manifest {
+                assert_eq!(
+                    std::fs::read(archive.join("manifest.json")).unwrap(),
+                    *manifest
+                );
+                assert!(matches!(
+                    runtime_session_retention(f.temp.path()).ensure_live(f.session),
+                    Err(session_store::StoreError::SessionErased(_))
+                ));
+            }
+        }
+        assert_eq!(
+            tree(&f.temp.path().join("archives/workers/prior-archive")),
+            prior
+        );
+        assert_eq!(tree(&unrelated_root), unrelated_before);
+        let Fixture {
+            temp,
+            runtime,
+            scope,
+            store,
+            session,
+            ..
+        } = f;
+        drop(store);
+        drop(runtime);
+        let runtime = Runtime::with_fs_store_and_execution_backend(
+            crate::FsRuntimeStoreOptions::new(temp.path()).with_runtime_id("runtime-test"),
+            Arc::new(AcceptingBackend),
+        )
+        .unwrap();
+        runtime.bind_runtime_identity("runtime-test").unwrap();
+        // Lost response/restart is an exact retry, not a new erasure operation.
+        runtime
+            .purge_subject_job_removal_scoped(&scope, request.clone())
+            .unwrap();
+        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!(!archive.join("session").exists());
+        assert!(!diagnostics.exists());
+        assert_eq!(
+            std::fs::read(temp.path().join("retention/operations/remove-source.json")).unwrap(),
+            receipt
+        );
+        assert_eq!(
+            tree(&temp.path().join("archives/workers/prior-archive")),
+            prior
+        );
+        assert_eq!(tree(&unrelated_root), unrelated_before);
+        if disposition == SessionDisposition::Archive {
+            assert!(matches!(
+                runtime_session_retention(temp.path()).lock_live(session),
+                Err(session_store::StoreError::SessionErased(_))
+            ));
+            assert!(
+                runtime
+                    .session_public_material_reader_scoped(
+                        &scope,
+                        &Source::Archived {
+                            archive_id: "canonical-archive".into()
+                        },
+                        &session.to_string()
+                    )
+                    .is_err()
+            );
+        }
+    }
+}
+
+#[test]
+fn subject_job_removal_purge_requires_complete_old_request_and_confirmed_exact_fence() {
+    let f = Fixture::new();
+    let request = f.removal_purge_request(SessionDisposition::Archive);
+    let copies = f.copies();
+    assert!(
+        f.runtime
+            .purge_subject_job_removal_scoped(&f.scope, request.clone())
+            .is_err()
+    );
+    assert_eq!(f.copies(), copies);
+    f.complete_removal(&request);
+    let copies = f.copies();
+    let receipt = std::fs::read(f.receipt_path()).unwrap();
+    let original = serde_json::to_value(&request).unwrap();
+    // Every saved request field is bound, not just operation/fingerprint. These
+    // are external serialized requests, including the pre-policy-change values.
+    for (section, field, value) in [
+        (
+            "deletion",
+            "workspace_id",
+            serde_json::json!("foreign-workspace"),
+        ),
+        (
+            "deletion",
+            "worker_id",
+            serde_json::json!(WorkerId::now_v7()),
+        ),
+        (
+            "deletion",
+            "create_fingerprint",
+            serde_json::json!("foreign-create"),
+        ),
+        (
+            "deletion",
+            "operation_id",
+            serde_json::json!("foreign-deletion"),
+        ),
+        (
+            "removal",
+            "operation_id",
+            serde_json::json!("unknown-removal"),
+        ),
+        (
+            "removal",
+            "input_fingerprint",
+            serde_json::json!("foreign-fingerprint"),
+        ),
+        (
+            "removal",
+            "archive_id",
+            serde_json::json!("foreign-archive"),
+        ),
+        (
+            "removal",
+            "workspace_id",
+            serde_json::json!("foreign-workspace"),
+        ),
+        (
+            "removal",
+            "source_runtime_id",
+            serde_json::json!("foreign-runtime"),
+        ),
+        (
+            "removal",
+            "worker_id",
+            serde_json::json!(WorkerId::now_v7()),
+        ),
+        ("removal", "source_created_at", serde_json::json!("changed")),
+        ("removal", "removed_at", serde_json::json!("changed")),
+        ("removal", "effective_profile", serde_json::json!("changed")),
+        ("removal", "retention_class", serde_json::json!("changed")),
+        ("removal", "policy_id", serde_json::json!("new-policy")),
+        ("removal", "session_disposition", serde_json::json!("purge")),
+        (
+            "removal",
+            "diagnostics_disposition",
+            serde_json::json!("retain"),
+        ),
+    ] {
+        let mut changed = original.clone();
+        changed[section][field] = value;
+        let changed = serde_json::from_value(changed).unwrap();
+        assert!(
+            f.runtime
+                .purge_subject_job_removal_scoped(&f.scope, changed)
+                .is_err(),
+            "{section}.{field}"
+        );
+        assert_eq!(f.copies(), copies, "{section}.{field}");
+        assert_eq!(std::fs::read(f.receipt_path()).unwrap(), receipt);
+    }
+    for scope in [
+        RuntimeWorkspaceScope::new("workspace-b", "https://backend.test"),
+        RuntimeWorkspaceScope::new("workspace-a", "https://other-backend.test"),
+    ] {
+        assert!(
+            f.runtime
+                .purge_subject_job_removal_scoped(&scope, request.clone())
+                .is_err()
+        );
+        assert_eq!(f.copies(), copies);
+    }
+}
+
+#[test]
+fn subject_job_removal_purge_missing_pending_or_foreign_receipt_preserves_copies() {
+    for kind in [
+        "missing",
+        "pending",
+        "foreign-worker",
+        "foreign-disposition",
+    ] {
+        let f = Fixture::new();
+        let mut request = f.removal_purge_request(SessionDisposition::Archive);
+        request.removal.diagnostics_disposition = runtime_api::DiagnosticsDisposition::Retain;
+        f.complete_removal(&request);
+        let path = f.receipt_path();
+        let mut receipt: serde_json::Value =
+            serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap();
+        match kind {
+            "missing" => std::fs::remove_file(&path).unwrap(),
+            _ => {
+                match kind {
+                    "pending" => receipt["result"]["source_removed"] = false.into(),
+                    "foreign-worker" => {
+                        receipt["result"]["worker_id"] = serde_json::json!(WorkerId::now_v7())
+                    }
+                    _ => receipt["result"]["session_disposition"] = "purge".into(),
+                }
+                std::fs::write(&path, serde_json::to_vec(&receipt).unwrap()).unwrap();
+            }
+        }
+        let copies = f.copies();
+        let retained_receipt = std::fs::read(&path).ok();
+        assert!(
+            f.runtime
+                .purge_subject_job_removal_scoped(&f.scope, request)
+                .is_err(),
+            "{kind}"
+        );
+        assert_eq!(f.copies(), copies, "{kind}");
+        assert_eq!(std::fs::read(path).ok(), retained_receipt);
+        runtime_session_retention(f.temp.path())
+            .ensure_live(f.session)
+            .unwrap();
+    }
+}
+
+#[tokio::test]
+async fn subject_job_removal_purge_signed_http_requires_delete_and_matching_scope() {
+    let f = Fixture::new();
+    let mut request = f.removal_purge_request(SessionDisposition::Archive);
+    request.removal.diagnostics_disposition = runtime_api::DiagnosticsDisposition::Retain;
+    f.complete_removal(&request);
+    let identity = RuntimeIdentityMaterial::generate("purge-key").unwrap();
+    let trust = workspace_trust(&identity);
+    let app = runtime_http_router_with_workspace_auth(
+        f.runtime.clone(),
+        None,
+        workspace_test_auth(vec![trust.clone()]),
+    );
+    let path = "/v1/subject-job-removal-purge";
+    let copies = f.copies();
+    for operation in ["workers:read", "workers:stop", "workers:create"] {
+        let body = serde_json::to_vec(&request).unwrap();
+        let mut claims = signed_claims(&trust, "POST", path, &body);
+        claims.operation = operation.into();
+        let response = signed_request(app.clone(), &identity, &claims, body, "workspace-a").await;
+        assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
+        assert_eq!(f.copies(), copies);
+    }
+    let mut foreign = request.clone();
+    foreign.deletion.workspace_id = "foreign-workspace".into();
+    let body = serde_json::to_vec(&foreign).unwrap();
+    let claims = signed_claims(&trust, "POST", path, &body);
+    let response = signed_request(app.clone(), &identity, &claims, body, "workspace-a").await;
+    assert_eq!(response.status(), StatusCode::FORBIDDEN);
+    assert_eq!(f.copies(), copies);
+    for _ in 0..2 {
+        let body = serde_json::to_vec(&request).unwrap();
+        let claims = signed_claims(&trust, "POST", path, &body);
+        assert_eq!(claims.operation, "workers:delete");
+        let response = signed_request(app.clone(), &identity, &claims, body, "workspace-a").await;
+        assert_eq!(response.status(), StatusCode::OK);
+        let bytes = axum::body::to_bytes(response.into_body(), 16384)
+            .await
+            .unwrap();
+        let response: runtime_api::SubjectJobRemovalPurgeResponse =
+            serde_json::from_slice(&bytes).unwrap();
+        assert_eq!(response.operation_id, request.removal.operation_id);
+        assert_eq!(
+            response.input_fingerprint,
+            request.removal.input_fingerprint
+        );
+        assert!(
+            !f.temp
+                .path()
+                .join("archives/workers/canonical-archive/session")
+                .exists()
+        );
+        assert!(
+            !f.temp
+                .path()
+                .join("archives/diagnostics/remove-source")
+                .exists()
+        );
+    }
+}

crates/worker-runtime/src/retention.rs

@@ -359,28 +359,91 @@ fn erase_archived_diagnostics(
             return Err(invalid());
         }
     }
-    let parent = runtime_root.join("archives/diagnostics");
-    let entries = match fs::read_dir(&parent) {
-        Ok(entries) => entries,
-        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(()),
-        Err(error) => return Err(error.into()),
-    };
-    let staging_prefix = format!(".{}.tmp-", archive.operation_id);
-    for entry in entries {
+    erase_operation_diagnostics(
+        runtime_root,
+        &archive.workspace_id,
+        &archive.source_runtime_id,
+        archive.source_worker_id,
+        &archive.operation_id,
+        &archive.input_fingerprint,
+    )
+}
+
+fn erase_removal_session_copies(
+    runtime_root: &Path,
+    archive: &WorkerSessionArchiveManifest,
+) -> Result<(), session_store::StoreError> {
+    let parent = runtime_root.join("archives/workers");
+    let prefix = format!(".{}.tmp-", archive.archive_id);
+    for entry in fs::read_dir(&parent)? {
         let entry = entry?;
         let name = entry.file_name();
-        let Some(name) = name.to_str() else {
+        let Some(name) = name.to_str() else { continue };
+        if name != archive.archive_id && !is_operation_staging(name, &prefix) {
             continue;
-        };
-        let staged = name.strip_prefix(&staging_prefix).is_some_and(|suffix| {
+        }
+        if !entry.file_type()?.is_dir() {
+            return Err(std::io::Error::other("Removal archive source is not a directory").into());
+        }
+        let mut saved: WorkerSessionArchiveManifest =
+            serde_json::from_slice(&fs::read(entry.path().join("manifest.json"))?)?;
+        // Each interrupted attempt attests the same input, with its own time.
+        saved.archived_at_unix_seconds = archive.archived_at_unix_seconds;
+        if saved != *archive {
+            return Err(std::io::Error::other("Removal archive copy identity mismatch").into());
+        }
+        match fs::remove_dir_all(entry.path().join("session")) {
+            Ok(()) => (),
+            Err(error) if error.kind() == std::io::ErrorKind::NotFound => (),
+            Err(error) => return Err(error.into()),
+        }
+        sync_directory(&entry.path(), "erase removal Session")
+            .map_err(|error| std::io::Error::other(error.to_string()))?;
+    }
+    Ok(())
+}
+
+fn is_operation_staging(name: &str, prefix: &str) -> bool {
+    name.strip_prefix(prefix).is_some_and(|suffix| {
         suffix.split_once('-').is_some_and(|(pid, sequence)| {
             !pid.is_empty()
                 && !sequence.is_empty()
                 && pid.bytes().all(|b| b.is_ascii_digit())
                 && sequence.bytes().all(|b| b.is_ascii_digit())
         })
-        });
-        if name != archive.operation_id && !staged {
+    })
+}
+
+fn erase_operation_diagnostics(
+    runtime_root: &Path,
+    workspace_id: &str,
+    source_runtime_id: &str,
+    worker_id: WorkerId,
+    operation_id: &str,
+    input_fingerprint: &str,
+) -> Result<(), session_store::StoreError> {
+    let invalid = || {
+        session_store::StoreError::Io(std::io::Error::new(
+            std::io::ErrorKind::InvalidData,
+            "Archived diagnostics erasure identity mismatch",
+        ))
+    };
+    validate_id("operation_id", operation_id).map_err(|_| invalid())?;
+    let parent = runtime_root.join("archives/diagnostics");
+    let entries = match fs::read_dir(&parent) {
+        Ok(entries) => entries,
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(()),
+        Err(error) => return Err(error.into()),
+    };
+    let staging_prefix = format!(".{operation_id}.tmp-");
+    for entry in entries {
+        let entry = entry?;
+        let name = entry.file_name();
+        let Some(name) = name.to_str() else {
+            continue;
+        };
+        let staged = is_operation_staging(name, &staging_prefix);
+        if name != operation_id && !staged {
             continue;
         }
         if !entry.file_type()?.is_dir() {
@@ -395,11 +458,11 @@ fn erase_archived_diagnostics(
         };
         if let Some(manifest) = manifest {
             if manifest.schema_version != ARCHIVE_SCHEMA_VERSION
-                || manifest.workspace_id != archive.workspace_id
-                || manifest.source_runtime_id != archive.source_runtime_id
-                || manifest.source_worker_id != archive.source_worker_id
-                || manifest.operation_id != archive.operation_id
-                || manifest.input_fingerprint != archive.input_fingerprint
+                || manifest.workspace_id != workspace_id
+                || manifest.source_runtime_id != source_runtime_id
+                || manifest.source_worker_id != worker_id
+                || manifest.operation_id != operation_id
+                || manifest.input_fingerprint != input_fingerprint
             {
                 return Err(invalid());
             }
@@ -480,6 +543,109 @@ impl FsWorkerRetentionProvider {
         Ok(receipt.result.source_removed.then_some(receipt.result))
     }
 
+    /// Caller holds the Worker operation lock and has verified the durable
+    /// Subject deletion fence. Never execute/recover a missing or pending receipt
+    /// here: the original canonical operation must have completed first.
+    pub(crate) fn purge_subject_job_removal(
+        &self,
+        request: &WorkerRetentionExecutionRequest,
+    ) -> Result<(), RuntimeError> {
+        validate_request(request)?;
+        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 Job removal purge",
+                path: lock_path.clone(),
+                source,
+            })?;
+        lock.lock().map_err(|source| RuntimeError::StoreIo {
+            operation: "lock Subject Job removal purge",
+            path: lock_path,
+            source,
+        })?;
+        let receipt = read_operation_receipt(&self.operation_path(&request.operation_id)?)?
+            .ok_or_else(|| {
+                RuntimeError::InvalidRequest("Canonical removal receipt missing".into())
+            })?;
+        validate_receipt_request(&receipt, request)?;
+        let result = &receipt.result;
+        if !result.source_removed
+            || 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.archive.is_some()
+                != (request.session_disposition == SessionDisposition::Archive)
+            || self.worker_dir(request.worker_id).exists()
+        {
+            return Err(RuntimeError::InvalidRequest(
+                "Canonical removal is not an exact successful receipt".into(),
+            ));
+        }
+        let archive = if let Some(manifest) = &result.archive {
+            if request.archive_id.as_ref() != Some(&manifest.archive_id)
+                || manifest.workspace_id != request.workspace_id
+                || manifest.source_runtime_id != request.source_runtime_id
+                || manifest.source_worker_id != request.worker_id
+                || manifest.operation_id != request.operation_id
+                || manifest.input_fingerprint != request.input_fingerprint
+            {
+                return Err(RuntimeError::InvalidRequest(
+                    "Removal archive identity mismatch".into(),
+                ));
+            }
+            let (saved, _) = self.session_archive_identity(&manifest.archive_id)?;
+            if saved != *manifest {
+                return Err(RuntimeError::InvalidRequest(
+                    "Removal archive receipt mismatch".into(),
+                ));
+            }
+            let session = manifest
+                .source_session_id
+                .parse::<session_store::SessionId>()
+                .map_err(|_| {
+                    RuntimeError::InvalidRequest("Invalid removal Session identity".into())
+                })?;
+            Some(session)
+        } else {
+            None
+        };
+        let erase = || -> Result<(), session_store::StoreError> {
+            erase_operation_diagnostics(
+                &self.runtime_root,
+                &request.workspace_id,
+                &request.source_runtime_id,
+                request.worker_id,
+                &request.operation_id,
+                &request.input_fingerprint,
+            )?;
+            if let Some(manifest) = &result.archive {
+                erase_removal_session_copies(&self.runtime_root, manifest)?;
+            }
+            Ok(())
+        };
+        if let Some(session) = &archive {
+            runtime_session_retention(&self.runtime_root)
+                .delete(
+                    *session,
+                    session_store::retention::SessionDeletionIntent::ErasePermanently,
+                    erase,
+                )
+                .map_err(retention_error)?;
+        } else {
+            // Diagnostics-only old Purge operations have no archive Session ID.
+            // The existing permanent Worker fence plus immutable completed receipt
+            // prevents execution and re-publication; hold the retention lock here.
+            erase().map_err(retention_error)?;
+        }
+        Ok(())
+    }
+
     pub(crate) fn snapshot(
         &self,
         workspace_id: &str,

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

@@ -10,6 +10,79 @@ pub(crate) struct SubjectDeleteWorkerFence {
 }
 
 impl Runtime {
+    /// Erase only copies bound to an exact, completed canonical removal. The
+    /// Subject fence already excludes execution; this does not mint erase authority
+    /// for an arbitrary Worker or reinterpret the historical retention result.
+    pub fn purge_subject_job_removal_scoped(
+        &self,
+        scope: &RuntimeWorkspaceScope,
+        request: runtime_api::SubjectJobRemovalPurgeRequest,
+    ) -> Result<runtime_api::SubjectJobRemovalPurgeResponse, RuntimeError> {
+        let deletion = &request.deletion;
+        let removal = &request.removal;
+        if deletion.workspace_id != scope.workspace_id
+            || removal.workspace_id != deletion.workspace_id
+            || removal.worker_id != deletion.worker_id
+        {
+            return Err(RuntimeError::InvalidRequest(
+                "Subject Job removal scope mismatch".into(),
+            ));
+        }
+        let lease = self.worker_operation_lock(deletion.worker_id)?;
+        let _operation = lease.lock().map_err(|_| RuntimeError::StatePoisoned)?;
+        let mut state = self.lock()?;
+        state.ensure_running()?;
+        state.ensure_workspace_owner(scope, false)?;
+        if state.runtime_identity.as_deref() != Some(removal.source_runtime_id.as_str()) {
+            return Err(RuntimeError::InvalidRequest(
+                "Subject Job removal Runtime identity mismatch".into(),
+            ));
+        }
+        state
+            .subject_delete_worker_fences
+            .get(&deletion.worker_id)
+            .filter(|fence| fence.request == *deletion && fence.response.is_some())
+            .ok_or_else(|| {
+                RuntimeError::InvalidRequest(
+                    "Exact confirmed Subject deletion fence required".into(),
+                )
+            })?;
+        if state.workers.contains_key(&deletion.worker_id)
+            || state
+                .unreconciled_worker_writes
+                .contains(&deletion.worker_id)
+        {
+            return Err(RuntimeError::InvalidRequest(
+                "Subject Job canonical removal must be reconciled first".into(),
+            ));
+        }
+        #[cfg(feature = "fs-store")]
+        {
+            let store = state.fs_store().ok_or_else(|| {
+                RuntimeError::InvalidRequest("Subject Job purge requires filesystem storage".into())
+            })?;
+            let removal: WorkerRetentionExecutionRequest = serde_json::from_value(
+                serde_json::to_value(removal)
+                    .map_err(|error| RuntimeError::InvalidRequest(error.to_string()))?,
+            )
+            .map_err(|error| RuntimeError::InvalidRequest(error.to_string()))?;
+            // Worker -> Runtime -> retention -> Session is the canonical removal
+            // lock order. The provider validates the receipt under its disk lock.
+            FsWorkerRetentionProvider::new(store.runtime_dir())
+                .purge_subject_job_removal(&removal)?;
+            Ok(runtime_api::SubjectJobRemovalPurgeResponse {
+                workspace_id: removal.workspace_id,
+                worker_id: removal.worker_id,
+                operation_id: removal.operation_id,
+                input_fingerprint: removal.input_fingerprint,
+            })
+        }
+        #[cfg(not(feature = "fs-store"))]
+        Err(RuntimeError::InvalidRequest(
+            "Subject Job purge requires filesystem storage".into(),
+        ))
+    }
+
     pub fn subject_delete_worker_scoped(
         &self,
         scope: &RuntimeWorkspaceScope,

crates/workspace-server/src/hosts.rs

@@ -1292,6 +1292,13 @@ pub trait WorkspaceWorkerRuntime: Send + Sync {
         })
     }
 
+    fn purge_subject_job_removal(
+        &self,
+        _request: runtime_api::SubjectJobRemovalPurgeRequest,
+    ) -> Result<runtime_api::SubjectJobRemovalPurgeResponse, String> {
+        Err("Subject Job removal source erasure is unavailable on this Runtime".into())
+    }
+
     fn subject_delete_worker(
         &self,
         _request: runtime_api::SubjectDeleteWorkerRequest,
@@ -2276,6 +2283,32 @@ impl RuntimeRegistry {
         runtime.worker_session_attachment(workspace_id, &worker_ref, session_id, attachment_id)
     }
 
+    pub fn purge_subject_job_removal(
+        &self,
+        runtime_id: &str,
+        deletion: runtime_api::SubjectDeleteWorkerRequest,
+        removal: WorkerRetentionExecutionRequest,
+    ) -> Result<runtime_api::SubjectJobRemovalPurgeResponse, RuntimeRegistryError> {
+        validate_backend_identifier("runtime_id", runtime_id)?;
+        let removal = runtime_contract_convert(removal).map_err(|diagnostic| {
+            RuntimeRegistryError::RuntimeOperationFailed {
+                runtime_id: runtime_id.into(),
+                code: "subject_job_removal_request_invalid".into(),
+                message: diagnostic.message,
+            }
+        })?;
+        self.runtime(runtime_id)?
+            .purge_subject_job_removal(runtime_api::SubjectJobRemovalPurgeRequest {
+                deletion,
+                removal,
+            })
+            .map_err(|message| RuntimeRegistryError::RuntimeOperationFailed {
+                runtime_id: runtime_id.into(),
+                code: "subject_job_removal_purge_unconfirmed".into(),
+                message,
+            })
+    }
+
     pub fn subject_delete_worker(
         &self,
         runtime_id: &str,
@@ -3402,6 +3435,19 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime {
             .map_err(|e| e.to_string())
     }
 
+    fn purge_subject_job_removal(
+        &self,
+        request: runtime_api::SubjectJobRemovalPurgeRequest,
+    ) -> Result<runtime_api::SubjectJobRemovalPurgeResponse, String> {
+        if request.deletion.workspace_id != self.workspace_id {
+            return Err("Subject Job removal erasure Workspace mismatch".into());
+        }
+        let scope = RuntimeWorkspaceScope::new(&request.deletion.workspace_id, "embedded-backend");
+        self.runtime
+            .purge_subject_job_removal_scoped(&scope, request)
+            .map_err(|e| e.to_string())
+    }
+
     fn recover_subject_create(
         &self,
         request: runtime_api::SubjectCreateRecoveryRequest,
@@ -5694,6 +5740,18 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
         })
     }
 
+    fn purge_subject_job_removal(
+        &self,
+        request: runtime_api::SubjectJobRemovalPurgeRequest,
+    ) -> Result<runtime_api::SubjectJobRemovalPurgeResponse, String> {
+        self.run_runtime_api(
+            self.request_timeout,
+            MAX_REMOTE_RUNTIME_RESPONSE_BYTES,
+            move |client| async move { client.purge_subject_job_removal(request).await },
+        )
+        .map_err(|diagnostic| diagnostic.message)
+    }
+
     fn subject_delete_worker(
         &self,
         request: runtime_api::SubjectDeleteWorkerRequest,

crates/workspace-server/src/implementation_result_tests.rs

@@ -407,7 +407,8 @@ fn reconstruct_implementation_schema96(conn: &Connection) -> Result<()> {
     }
     // Children before parents; dropping the tables also drops snapshot guards.
     conn.execute_batch(
-        "DROP TABLE subject_deletion_allocations;
+        "DROP TABLE subject_deletion_removals;
+        DROP TABLE subject_deletion_allocations;
         DROP TABLE subject_deletion_jobs;
         DROP TABLE subject_deletion_workers;
         DROP TABLE subject_deletions;

crates/workspace-server/src/retention.rs

@@ -333,6 +333,8 @@ impl SqliteWorkspaceStore {
                     return Err(StoreError::InvalidInput(if other{"cross-workspace".into()}else{"worker-missing".into()}));
                 }
             };
+            let pending:bool=tx.query_row("SELECT EXISTS(SELECT 1 FROM subject_deletion_removals WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND completed=0)",params![req.workspace_id,req.worker.runtime_id,req.worker.worker_id],|row|row.get(0))?;
+            if pending {return Err(StoreError::InvalidInput("Pinned pre-deletion removal must reconcile before another retention plan".into()));}
             if crate::store::subject_deletion_worker_kind(&tx, &req.workspace_id, &req.worker)?.is_some() {
                 let prior = tx.prepare("SELECT plan_id FROM worker_removal_operations WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND state IN ('executing','failed')")?
                     .query_map(params![req.workspace_id,req.worker.runtime_id,req.worker.worker_id],|r|r.get::<_,String>(0))?
@@ -381,23 +383,26 @@ impl SqliteWorkspaceStore {
                 return Err(StoreError::InvalidInput(format!("stale:{plan_id}:plan state {} is not executable", state_s(plan.state))));
             }
             if !plan.blockers.is_empty(){return Err(StoreError::InvalidInput(format!("blocked:{}",serde_json::to_string(&plan.blockers).unwrap())));}
+            let reconciliation=crate::store::subject_deletion_removal_matches(&tx,workspace_id,&plan.worker,&plan.operation_id,&plan.input_fingerprint)?;
+            if !reconciliation {
                 let policy=load_removal_policy(&tx,workspace_id,&plan.worker,plan.policy_id==ERASED_JOURNAL_POLICY_ID)?.ok_or_else(||StoreError::InvalidInput(format!("policy-missing:{workspace_id}")))?;
                 if policy.policy_id!=plan.policy_id||policy.policy_digest!=plan.policy_digest{
                     mark_stale(&tx,&plan,"policy content changed")?;tx.commit()?;
                     return Err(stale_error(&plan,"policy content changed"));
                 }
+            }
             let worker=load_worker(&tx,workspace_id,&plan.worker)?.ok_or_else(||StoreError::InvalidInput(format!("stale:{plan_id}:Worker missing")))?;
-            if worker.updated_at!=plan.worker_updated_at{
+            if !reconciliation && worker.updated_at!=plan.worker_updated_at{
                 mark_stale(&tx,&plan,"Worker updated_at changed")?;tx.commit()?;
                 return Err(stale_error(&plan,"Worker updated_at changed"));
             }
             if worker.retention_state=="pinned"{
-                mark_stale(&tx,&plan,"hold added")?;tx.commit()?;
+                if !reconciliation {mark_stale(&tx,&plan,"hold added")?;}tx.commit()?;
                 return Err(stale_error(&plan,"hold added"));
             }
             let assigned:bool=tx.query_row("SELECT EXISTS(SELECT 1 FROM ticket_active_worker_assignments c JOIN ticket_worker_assignments a ON a.workspace_id=c.workspace_id AND a.ticket_id=c.ticket_id AND a.assignment_id=c.assignment_id WHERE a.workspace_id=?1 AND a.runtime_id=?2 AND a.worker_id=?3 UNION ALL SELECT 1 FROM ticket_assignment_operations WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND claim_state='pending' AND action IN ('assign','reassign'))",params![workspace_id,plan.worker.runtime_id,plan.worker.worker_id],|r|r.get(0))?;
             if assigned{
-                mark_stale(&tx,&plan,"unfinished work added")?;tx.commit()?;
+                if !reconciliation {mark_stale(&tx,&plan,"unfinished work added")?;}tx.commit()?;
                 return Err(stale_error(&plan,"unfinished work added"));
             }
             let now=Utc::now().to_rfc3339();
@@ -416,6 +421,15 @@ impl SqliteWorkspaceStore {
         input_fingerprint: &str,
     ) -> Result<PreparedWorkerRemoval, WorkerRetentionError> {
         let plan = self.begin_worker_removal(workspace_id, plan_id, input_fingerprint)?;
+        if let Some(reconciliation) = self.subject_deletion_removal(workspace_id, &plan.worker)?
+            && reconciliation.request.operation_id == plan.operation_id
+        {
+            return Ok(PreparedWorkerRemoval {
+                runtime_request: reconciliation.request,
+                prior_failure_category: plan.failure_category.clone(),
+                plan,
+            });
+        }
         let worker = self
             .with_conn(|conn| load_worker(conn, workspace_id, &plan.worker))?
             .ok_or_else(|| WorkerRetentionError::StalePlan {
@@ -466,12 +480,18 @@ impl SqliteWorkspaceStore {
                  WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3
                    AND state IN ('planned','executing','failed','succeeded')
                    AND (
-                     state='succeeded' OR worker_updated_at=(
+                     state='succeeded' OR EXISTS(
+                       SELECT 1 FROM subject_deletion_removals d
+                       WHERE d.workspace_id=worker_removal_operations.workspace_id
+                         AND d.runtime_id=worker_removal_operations.runtime_id
+                         AND d.worker_id=worker_removal_operations.worker_id
+                         AND d.operation_id=worker_removal_operations.operation_id
+                     ) OR worker_updated_at=(
                        SELECT updated_at FROM worker_registry
                        WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3
                      )
                    )
-                 ORDER BY CASE WHEN EXISTS(SELECT 1 FROM subject_deletion_workers d WHERE d.workspace_id=worker_removal_operations.workspace_id AND d.runtime_id=worker_removal_operations.runtime_id AND d.worker_id=worker_removal_operations.worker_id AND d.prior_removal_operation_id=worker_removal_operations.operation_id) THEN 0 WHEN state='succeeded' THEN 1 ELSE 2 END,
+                 ORDER BY CASE WHEN EXISTS(SELECT 1 FROM subject_deletion_workers d WHERE d.workspace_id=worker_removal_operations.workspace_id AND d.runtime_id=worker_removal_operations.runtime_id AND d.worker_id=worker_removal_operations.worker_id AND d.prior_removal_operation_id=worker_removal_operations.operation_id) THEN 0 WHEN EXISTS(SELECT 1 FROM subject_deletion_removals d WHERE d.workspace_id=worker_removal_operations.workspace_id AND d.runtime_id=worker_removal_operations.runtime_id AND d.worker_id=worker_removal_operations.worker_id AND d.operation_id=worker_removal_operations.operation_id) THEN 1 WHEN state='succeeded' THEN 2 ELSE 3 END,
                           created_at DESC,operation_id DESC LIMIT 1",
                 params![workspace_id, worker.runtime_id, worker.worker_id],
                 |row| row.get::<_, String>(0),
@@ -489,6 +509,15 @@ impl SqliteWorkspaceStore {
         self.with_conn(|conn| require_current_subject_deletion_plan(conn, &plan))
             .map_err(map_error)?;
         let prior_failure_category = plan.failure_category.clone();
+        if let Some(reconciliation) = self.subject_deletion_removal(workspace_id, &plan.worker)?
+            && reconciliation.request.operation_id == plan.operation_id
+        {
+            return Ok(Some(PreparedWorkerRemoval {
+                runtime_request: reconciliation.request,
+                prior_failure_category,
+                plan,
+            }));
+        }
         let worker_id = plan
             .worker
             .worker_id
@@ -698,8 +727,12 @@ impl SqliteWorkspaceStore {
                     return Err(StoreError::InvalidInput("Erased removal receipt requires matching source and physical Runtime erasure acknowledgement".into()));
                 }
             }
+            let reconciliation=crate::store::subject_deletion_removal_matches(&tx,workspace_id,&plan.worker,&plan.operation_id,&plan.input_fingerprint)?;
             let worker=load_worker(&tx,workspace_id,&plan.worker)?.ok_or_else(||StoreError::InvalidInput("Worker missing before commit".into()))?;
-            if worker.updated_at!=plan.worker_updated_at{return Err(StoreError::InvalidInput(format!("stale:{}:Worker updated_at changed",plan.plan_id)));}
+            // A pinned pre-deletion request is reconciled unchanged under the permanent
+            // Subject/Runtime exclusion. Its Worker identity cannot be reused. Metadata
+            // version movement must not discard an already-applied canonical result.
+            if !reconciliation && worker.updated_at!=plan.worker_updated_at{return Err(StoreError::InvalidInput(format!("stale:{}:Worker updated_at changed",plan.plan_id)));}
             if worker.retention_state=="pinned" { return Err(StoreError::InvalidInput(format!("stale:{}:hold added",plan.plan_id))); }
             let assigned:bool=tx.query_row("SELECT EXISTS(SELECT 1 FROM ticket_active_worker_assignments c JOIN ticket_worker_assignments a ON a.workspace_id=c.workspace_id AND a.ticket_id=c.ticket_id AND a.assignment_id=c.assignment_id WHERE a.workspace_id=?1 AND a.runtime_id=?2 AND a.worker_id=?3 UNION ALL SELECT 1 FROM ticket_assignment_operations WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND claim_state='pending' AND action IN ('assign','reassign'))",params![workspace_id,plan.worker.runtime_id,plan.worker.worker_id],|row|row.get(0))?;
             if assigned { return Err(StoreError::InvalidInput(format!("stale:{}:unfinished work added",plan.plan_id))); }
@@ -771,7 +804,7 @@ impl SqliteWorkspaceStore {
                 tx.execute("INSERT OR IGNORE INTO worker_tombstones(workspace_id,runtime_id,worker_id,display_name,profile,worker_created_at,removed_at,archive_id,policy_id,policy_digest,operation_id) VALUES(?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11)",params![workspace_id,plan.worker.runtime_id,plan.worker.worker_id,worker.display_name,worker.profile,worker.created_at,now,if erasure.is_some() { None } else { plan.archive_id.clone() },plan.policy_id,plan.policy_digest,operation_id])?;
             }
             tx.execute("UPDATE worker_workdir_links SET unlinked_at=?4 WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND unlinked_at IS NULL",params![workspace_id,plan.worker.runtime_id,plan.worker.worker_id,now])?;
-            let deleted=tx.execute("DELETE FROM worker_registry WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND updated_at=?4",params![workspace_id,plan.worker.runtime_id,plan.worker.worker_id,plan.worker_updated_at])?;
+            let deleted=tx.execute("DELETE FROM worker_registry WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND (updated_at=?4 OR ?5)",params![workspace_id,plan.worker.runtime_id,plan.worker.worker_id,plan.worker_updated_at,reconciliation])?;
             tx.execute("UPDATE worker_create_reservations SET state='removed',updated_at=?4 WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND state IN ('reserved','created')",params![workspace_id,plan.worker.runtime_id,plan.worker.worker_id,now])?;
             if deleted!=1{return Err(StoreError::InvalidInput(format!("stale:{}:removal fence changed",plan.plan_id)));}
             let catalog=commit_worker_catalog_removal(&tx,workspace_id,&plan.worker)?;
@@ -1109,6 +1142,18 @@ fn require_current_subject_deletion_plan(
     c: &Connection,
     plan: &WorkerRemovalPlan,
 ) -> crate::Result<()> {
+    if crate::store::subject_deletion_removal_matches(
+        c,
+        &plan.workspace_id,
+        &plan.worker,
+        &plan.operation_id,
+        &plan.input_fingerprint,
+    )? {
+        // An unfinished canonical request is pinned separately from successes. Only
+        // its exact original input may reconcile; any required Job copy erasure is
+        // a subsequent physical operation and never changes this historical result.
+        return Ok(());
+    }
     if plan.state == WorkerRemovalPlanState::Succeeded
         && crate::store::subject_deletion_prior_removal_matches(
             c,
@@ -2831,7 +2876,7 @@ mod tests {
     #[test]
     fn subject_deletion_snapshot_overrides_policy_without_changing_normal_removal() {
         let store = setup();
-        let ordinary = store.plan_worker_removal(&req(), &inv()).unwrap();
+        let ordinary = setup().plan_worker_removal(&req(), &inv()).unwrap();
         store
             .prepare_subject_deletion("w", "subject", "delete", &[req().worker.clone()])
             .unwrap();
@@ -2842,12 +2887,8 @@ mod tests {
         assert_eq!(body.diagnostics_disposition, DiagnosticsDisposition::Retain);
         assert!(body.diagnostics_retention_seconds.is_some());
         assert_ne!(body.policy_id, ordinary.policy_id);
-        // A saved plan that would discard diagnostics cannot silently stand in
-        // for the Subject conversation archive plan.
-        assert!(matches!(
-            store.begin_worker_removal("w", &ordinary.plan_id, &ordinary.input_fingerprint),
-            Err(WorkerRetentionError::StalePlan { .. })
-        ));
+        // A new deletion-owned plan uses the archival policy; independently saved
+        // old requests are reconciled separately, without borrowing this policy.
         let ordinary_store = setup();
         assert_eq!(
             ordinary_store
@@ -2859,10 +2900,10 @@ mod tests {
     }
 
     #[test]
-    fn subject_deletion_job_purge_rejects_recovery_of_old_executing_archive() {
+    fn subject_deletion_pins_old_archive_input_without_counting_it_as_job_purge() {
         let store = setup();
         let old = store.plan_worker_removal(&req(), &inv()).unwrap();
-        store
+        let original = store
             .prepare_worker_removal_execution("w", &old.plan_id, &old.input_fingerprint)
             .unwrap();
         store
@@ -2877,14 +2918,21 @@ mod tests {
                 Ok(())
             })
             .unwrap();
-        assert!(matches!(
-            store.recover_worker_removal_execution("w", &req().worker),
-            Err(WorkerRetentionError::StalePlan { .. })
-        ));
-        assert!(matches!(
-            store.plan_worker_removal(&req(), &inv()),
-            Err(WorkerRetentionError::StalePlan { .. })
-        ));
+        let recovered = store
+            .recover_worker_removal_execution("w", &req().worker)
+            .unwrap()
+            .unwrap();
+        assert_eq!(recovered.runtime_request, original.runtime_request);
+        let prepared = store
+            .prepare_worker_removal_execution("w", &old.plan_id, &old.input_fingerprint)
+            .unwrap();
+        assert_eq!(prepared.runtime_request, original.runtime_request);
+        assert!(store.plan_worker_removal(&req(), &inv()).is_err());
+        assert!(
+            store
+                .erase_subject_deleted_jobs("w", "subject", "delete")
+                .is_err()
+        );
         let fresh = store
             .with_conn(|c| load_removal_policy(c, "w", &req().worker, false))
             .unwrap()

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

@@ -182,6 +182,45 @@ async fn remove_worker(
         )
         .into());
     }
+    if let Some(reconciliation) = api
+        .config_store
+        .subject_deletion_removal(api.workspace_id(), worker)?
+    {
+        if reconciliation.completed {
+            return Ok(());
+        }
+        // First reconcile the original canonical result, without changing any old
+        // disposition, fingerprint or success snapshot. A lost Runtime response or
+        // Backend commit remains unknown until that same child succeeds.
+        execute_removal(api, operation, worker).await?;
+        let result = api
+            .runtime
+            .execute_worker_retention(worker, reconciliation.request.clone())
+            .map_err(RuntimeRegistryError::into_error)?;
+        let purge = if reconciliation.is_job {
+            Some(
+                api.runtime
+                    .purge_subject_job_removal(
+                        &worker.runtime_id,
+                        request,
+                        reconciliation.request.clone(),
+                    )
+                    .map_err(RuntimeRegistryError::into_error)?,
+            )
+        } else {
+            None
+        };
+        // The extra physical Job erasure is a distinct receipt. An old Archive
+        // success is never counted as a Purge, nor pinned as an older success later.
+        api.config_store.confirm_subject_deletion_removal(
+            api.workspace_id(),
+            worker,
+            &reconciliation.request,
+            &result,
+            purge.as_ref(),
+        )?;
+        return Ok(());
+    }
     // Replay the canonical child even when physical removal already happened:
     // its retained receipt can finish the Backend catalog midpoint. The Runtime
     // fence above also prevents an older create from arriving after that removal.

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

@@ -727,3 +727,473 @@ async fn unmaterialized_body_and_job_reservations_finish_only_with_runtime_stora
         }
     }
 }
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn old_job_removal_reconciles_exact_input_then_erases_copies_across_http_retry_and_restart() {
+    use worker_runtime::retention::{DiagnosticsDisposition, SessionDisposition};
+    for phase in [
+        "planned",
+        "executing",
+        "failed",
+        "response_lost",
+        "catalog_failure",
+        "purge_ack_commit_failure",
+    ] {
+        let dir = tempfile::tempdir().unwrap();
+        let (api, execution) = test_api_with_recording_backend(dir.path()).await;
+        let store = open_subjektiv_store(&api).unwrap();
+        let subject = store
+            .create_subject(
+                crate::subjektiv::SubjectRole::new(format!("old plan {phase}")).unwrap(),
+            )
+            .unwrap();
+        let other = store
+            .create_subject_with_behavior(
+                crate::subjektiv::SubjectRole::new("keep other").unwrap(),
+                "untouched".into(),
+            )
+            .unwrap();
+        let candidate = serde_json::from_value(json!({
+            "schema_version": 2, "id": "old-job-candidate", "extract_run_id": "extract",
+            "source": {"segment_id":"segment", "range":[1,2]}, "kind":"constraint",
+            "claim":"private candidate", "why_useful":"erase after explicit deletion",
+            "evidence": [], "source_refs": []
+        }))
+        .unwrap();
+        store
+            .stage_candidate(crate::subjektiv::SubjectStagingRecord::attach(
+                &subject.id,
+                candidate,
+            ))
+            .unwrap();
+        start_subjektiv_staging_consolidation(
+            api.clone(),
+            &subject.id,
+            MemoryConsolidateStagingOperation { force: true },
+        )
+        .await
+        .unwrap();
+        let job = api
+            .store
+            .find_active_backend_job_for_resource(
+                TEST_WORKSPACE_ID,
+                &crate::backend_job::subjektiv_serialization_key(&subject.id),
+            )
+            .unwrap()
+            .unwrap();
+        let attempt = api
+            .store
+            .get_backend_job_attempt(
+                TEST_WORKSPACE_ID,
+                &job.request.job_id,
+                &crate::backend_job::attempt_id(&job.request.job_id, job.current_attempt),
+            )
+            .unwrap()
+            .unwrap();
+        let worker = attempt.worker.clone().unwrap();
+        seed_worker_session_for_cleanup(dir.path(), &worker);
+        api.runtime
+            .stop_worker(
+                &worker,
+                WorkerLifecycleRequest {
+                    reason: Some("Prior independent removal".into()),
+                    ticket_assignment: None,
+                },
+            )
+            .unwrap();
+        let diagnostic = dir
+            .path()
+            .join(".test-embedded-runtime-store/workers")
+            .join(&worker.worker_id)
+            .join("runs/old/worker.out.log");
+        std::fs::create_dir_all(diagnostic.parent().unwrap()).unwrap();
+        std::fs::write(&diagnostic, "private old Job diagnostic").unwrap();
+        let policy = api
+            .config_store
+            .worker_retention_policy(TEST_WORKSPACE_ID)
+            .unwrap()
+            .unwrap();
+        api.config_store
+            .update_worker_retention_policy(
+                TEST_WORKSPACE_ID,
+                &policy.policy_digest,
+                &crate::retention::WorkerRetentionPolicyUpdate {
+                    policy_id: "prior-job-archive-policy".into(),
+                    session_disposition: SessionDisposition::Archive,
+                    metadata_disposition: crate::retention::MetadataDisposition::Tombstone,
+                    archive_retention: crate::retention::ArchiveRetention::ForSeconds {
+                        seconds: 600,
+                    },
+                    diagnostics_disposition: DiagnosticsDisposition::Retain,
+                    diagnostics_retention_seconds: Some(600),
+                },
+            )
+            .unwrap();
+        let inventory = api.runtime.worker_retention_inventory(&worker).unwrap();
+        assert!(inventory.diagnostics_bytes > 0);
+        let old = api
+            .config_store
+            .plan_worker_removal(
+                &crate::retention::WorkerRemovalPlanRequest {
+                    workspace_id: TEST_WORKSPACE_ID.into(),
+                    worker: worker.clone(),
+                    reason: "Previously requested retention".into(),
+                    journal_erasure: None,
+                },
+                &inventory,
+            )
+            .unwrap();
+        assert_eq!(old.session_disposition, SessionDisposition::Archive);
+        let saved = if phase == "planned" {
+            api.config_store
+                .recover_worker_removal_execution(TEST_WORKSPACE_ID, &worker)
+                .unwrap()
+                .unwrap()
+        } else {
+            api.config_store
+                .prepare_worker_removal_execution(
+                    TEST_WORKSPACE_ID,
+                    &old.plan_id,
+                    &old.input_fingerprint,
+                )
+                .unwrap()
+        };
+        if phase == "failed" {
+            api.config_store
+                .fail_worker_removal(
+                    TEST_WORKSPACE_ID,
+                    &old.operation_id,
+                    &old.input_fingerprint,
+                    "old_unknown_outcome",
+                )
+                .unwrap();
+        }
+        if phase == "catalog_failure" {
+            api.config_store.with_conn(|c| { c.execute_batch("CREATE TRIGGER injected_old_catalog_failure BEFORE DELETE ON worker_registry BEGIN SELECT RAISE(ABORT,'catalog failure'); END;")?; Ok(()) }).unwrap();
+        }
+        let old_result = if matches!(phase, "response_lost" | "catalog_failure") {
+            let result = api
+                .runtime
+                .execute_worker_retention(&worker, saved.runtime_request.clone())
+                .unwrap();
+            if phase == "catalog_failure" {
+                assert!(
+                    api.config_store
+                        .commit_worker_removal(
+                            TEST_WORKSPACE_ID,
+                            &old.operation_id,
+                            &old.input_fingerprint,
+                            &result
+                        )
+                        .is_err()
+                );
+            }
+            Some(result)
+        } else {
+            None
+        };
+        if phase == "purge_ack_commit_failure" {
+            api.config_store.with_conn(|c| { c.execute_batch("CREATE TRIGGER injected_purge_ack_commit_failure BEFORE UPDATE OF completed ON subject_deletion_removals WHEN NEW.completed=1 BEGIN SELECT RAISE(ABORT,'purge receipt commit failure'); END;")?; Ok(()) }).unwrap();
+        }
+        // Force a midpoint AFTER domain/snapshot commit but BEFORE any deletion-owned
+        // Runtime effect. It must retain exactly the original canonical input.
+        *execution.subject_delete_failure.lock().unwrap() = Some("shutdown outcome unknown".into());
+        let owner =
+            seed_test_workspace_owner_token(api.store.as_ref(), &format!("old-removal-{phase}"))
+                .await;
+        let (status, pending) = delete_request(
+            &api,
+            &subject.id,
+            "same-delete",
+            Some(&owner),
+            TEST_WORKSPACE_ID,
+        )
+        .await;
+        assert_eq!(status, StatusCode::OK, "{phase}: {pending}");
+        assert_eq!(pending["completed"], false, "{phase}");
+        let pinned = api
+            .config_store
+            .subject_deletion_removal(TEST_WORKSPACE_ID, &worker)
+            .unwrap()
+            .unwrap();
+        assert_eq!(pinned.request, saved.runtime_request, "{phase}");
+        assert!(!pinned.completed);
+        assert!(pinned.is_job);
+        assert!(
+            api.config_store
+                .erase_subject_deleted_jobs(TEST_WORKSPACE_ID, &subject.id, "same-delete")
+                .is_err()
+        );
+        *execution.subject_delete_failure.lock().unwrap() = None;
+        let mut reopened = api.clone();
+        let reopened_store =
+            Arc::new(SqliteWorkspaceStore::open_for_test(&api.config.database_path).unwrap());
+        reopened.config_store = reopened_store.clone();
+        reopened.store = reopened_store;
+        if matches!(phase, "catalog_failure" | "purge_ack_commit_failure") {
+            let (status, pending) = delete_request(
+                &reopened,
+                &subject.id,
+                "same-delete",
+                Some(&owner),
+                TEST_WORKSPACE_ID,
+            )
+            .await;
+            assert_eq!(status, StatusCode::OK, "{phase}: {pending}");
+            assert_eq!(pending["completed"], false, "{phase}");
+            assert!(
+                !store
+                    .subject_deletion(&subject.id)
+                    .unwrap()
+                    .unwrap()
+                    .completed
+            );
+            assert!(
+                reopened
+                    .store
+                    .get_backend_job(TEST_WORKSPACE_ID, &job.request.job_id)
+                    .unwrap()
+                    .is_some()
+            );
+            reopened
+                .config_store
+                .with_conn(|c| {
+                    c.execute_batch(if phase == "catalog_failure" {
+                        "DROP TRIGGER injected_old_catalog_failure;"
+                    } else {
+                        "DROP TRIGGER injected_purge_ack_commit_failure;"
+                    })?;
+                    Ok(())
+                })
+                .unwrap();
+        }
+        // Exercise the same production restart recovery that starts from the saved
+        // domain operation, then HTTP replay of the same full request.
+        let recovery = reopened.clone();
+        let handle = tokio::runtime::Handle::current();
+        reopened
+            .host_executors
+            .host()
+            .unwrap()
+            .run(move || handle.block_on(subjektiv_deletion::recover(&recovery)))
+            .await
+            .unwrap();
+        let (status, done) = delete_request(
+            &reopened,
+            &subject.id,
+            "same-delete",
+            Some(&owner),
+            TEST_WORKSPACE_ID,
+        )
+        .await;
+        assert_eq!(status, StatusCode::OK, "{phase}: {done}");
+        assert_eq!(done["completed"], true, "{phase}: {done}");
+        let reconciled = reopened
+            .config_store
+            .subject_deletion_removal(TEST_WORKSPACE_ID, &worker)
+            .unwrap()
+            .unwrap();
+        assert!(reconciled.completed);
+        assert_eq!(reconciled.request, saved.runtime_request);
+        let canonical = reopened
+            .config_store
+            .recover_worker_removal_execution(TEST_WORKSPACE_ID, &worker)
+            .unwrap()
+            .unwrap();
+        assert_eq!(
+            canonical.plan.state,
+            crate::retention::WorkerRemovalPlanState::Succeeded
+        );
+        assert_eq!(canonical.runtime_request, saved.runtime_request);
+        assert_eq!(
+            canonical.plan.session_disposition,
+            SessionDisposition::Archive,
+            "old result was not relabeled Purge"
+        );
+        let result = reopened
+            .runtime
+            .execute_worker_retention(&worker, saved.runtime_request.clone())
+            .unwrap();
+        if let Some(old_result) = old_result {
+            assert_eq!(result, old_result);
+        }
+        assert_eq!(result.session_disposition, SessionDisposition::Archive);
+        assert!(result.source_removed);
+        let archive = result.archive.as_ref().unwrap();
+        let root = dir.path().join(".test-embedded-runtime-store");
+        assert!(
+            !root
+                .join("archives/workers")
+                .join(&archive.archive_id)
+                .join("session")
+                .exists(),
+            "new Job archive copy survives {phase}"
+        );
+        if result.diagnostics_disposition == DiagnosticsDisposition::Retain {
+            assert!(
+                !root
+                    .join("archives/diagnostics")
+                    .join(&old.operation_id)
+                    .exists(),
+                "Job diagnostics survive {phase}"
+            );
+        }
+        assert!(
+            reopened
+                .store
+                .get_backend_job(TEST_WORKSPACE_ID, &job.request.job_id)
+                .unwrap()
+                .is_none()
+        );
+        assert!(
+            reopened
+                .store
+                .current_worker_singleton_owner(
+                    TEST_WORKSPACE_ID,
+                    &crate::backend_job::subjektiv_serialization_key(&subject.id)
+                )
+                .unwrap()
+                .is_none()
+        );
+        assert!(store.subject(&subject.id).unwrap().is_none());
+        assert_eq!(
+            store.subject(&other.id).unwrap().unwrap().behavior_md,
+            "untouched"
+        );
+        let (status, replay) = delete_request(
+            &reopened,
+            &subject.id,
+            "same-delete",
+            Some(&owner),
+            TEST_WORKSPACE_ID,
+        )
+        .await;
+        assert_eq!(status, StatusCode::OK);
+        assert_eq!(done, replay);
+    }
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn old_body_archive_reconciles_unchanged_and_remains_retained_after_subject_deletion() {
+    let dir = tempfile::tempdir().unwrap();
+    init_clean_git_workspace(dir.path());
+    let (api, execution) = test_api_with_recording_backend(dir.path()).await;
+    let store = open_subjektiv_store(&api).unwrap();
+    let subject = store
+        .create_subject(crate::subjektiv::SubjectRole::new("old body archive").unwrap())
+        .unwrap();
+    let Json(created) = create_workspace_worker(
+        State(api.clone()),
+        HeaderMap::new(),
+        Json(CreateWorkspaceWorkerRequest {
+            runtime_id: EMBEDDED_WORKER_RUNTIME_ID.into(),
+            display_name: "Existing body removal".into(),
+            singleton_key: None,
+            profile: Some("builtin:companion".into()),
+            ticket_assignment: None,
+            initial_submit: vec![],
+            workdir_attachments: vec![],
+            feature_connections: server_api::WorkspaceWorkerFeatureConnectionsRequest {
+                subjektiv: Some(server_api::SubjektivWorkerConnectionRequest {
+                    subject_id: subject.id.clone(),
+                }),
+            },
+            control_operation_id: None,
+            execution_grant: None,
+        }),
+    )
+    .await
+    .unwrap();
+    let worker = RuntimeWorkerRef::new(created.runtime_id, created.worker_id);
+    seed_worker_session_for_cleanup(dir.path(), &worker);
+    api.runtime
+        .stop_worker(
+            &worker,
+            WorkerLifecycleRequest {
+                reason: Some("Earlier removal".into()),
+                ticket_assignment: None,
+            },
+        )
+        .unwrap();
+    let inventory = api.runtime.worker_retention_inventory(&worker).unwrap();
+    let old = api
+        .config_store
+        .plan_worker_removal(
+            &crate::retention::WorkerRemovalPlanRequest {
+                workspace_id: TEST_WORKSPACE_ID.into(),
+                worker: worker.clone(),
+                reason: "Earlier authorized archive".into(),
+                journal_erasure: None,
+            },
+            &inventory,
+        )
+        .unwrap();
+    let saved = api
+        .config_store
+        .prepare_worker_removal_execution(TEST_WORKSPACE_ID, &old.plan_id, &old.input_fingerprint)
+        .unwrap();
+    let result = api
+        .runtime
+        .execute_worker_retention(&worker, saved.runtime_request.clone())
+        .unwrap();
+    // Simulate a lost result before canonical Backend commit. Deletion must pin an
+    // unfinished request, not retrospectively classify its archive as a prior success.
+    *execution.subject_delete_failure.lock().unwrap() = Some("unknown shutdown".into());
+    let owner = seed_test_workspace_owner_token(api.store.as_ref(), "old-body-removal").await;
+    let (status, pending) = delete_request(
+        &api,
+        &subject.id,
+        "same-body-delete",
+        Some(&owner),
+        TEST_WORKSPACE_ID,
+    )
+    .await;
+    assert_eq!(status, StatusCode::OK, "{pending}");
+    assert_eq!(pending["completed"], false);
+    let snapshot = api
+        .config_store
+        .subject_deletion_removal(TEST_WORKSPACE_ID, &worker)
+        .unwrap()
+        .unwrap();
+    assert!(!snapshot.is_job && !snapshot.completed);
+    assert_eq!(snapshot.request, saved.runtime_request);
+    *execution.subject_delete_failure.lock().unwrap() = None;
+    let (status, done) = delete_request(
+        &api,
+        &subject.id,
+        "same-body-delete",
+        Some(&owner),
+        TEST_WORKSPACE_ID,
+    )
+    .await;
+    assert_eq!(status, StatusCode::OK, "{done}");
+    assert_eq!(done["completed"], true, "{done}");
+    let completed = api
+        .config_store
+        .subject_deletion_removal(TEST_WORKSPACE_ID, &worker)
+        .unwrap()
+        .unwrap();
+    assert!(completed.completed);
+    assert_eq!(completed.request, saved.runtime_request);
+    let replay = api
+        .runtime
+        .execute_worker_retention(&worker, saved.runtime_request)
+        .unwrap();
+    assert_eq!(replay, result);
+    let archive = result.archive.unwrap();
+    assert!(
+        dir.path()
+            .join(".test-embedded-runtime-store/archives/workers")
+            .join(&archive.archive_id)
+            .join("session")
+            .is_dir()
+    );
+    assert!(api.config_store.with_conn(|c| Ok(c.query_row("SELECT prior_removal_operation_id IS NULL FROM subject_deletion_workers WHERE workspace_id=?1 AND subject_id=?2 AND runtime_id=?3 AND worker_id=?4", rusqlite::params![TEST_WORKSPACE_ID,subject.id,worker.runtime_id,worker.worker_id], |r|r.get::<_,bool>(0))?)).unwrap());
+    assert!(
+        store
+            .subject_deletion(&subject.id)
+            .unwrap()
+            .unwrap()
+            .completed
+    );
+}

crates/workspace-server/src/store.rs

@@ -95,7 +95,8 @@ use crate::{Error, Result};
 #[path = "subject_deletion_store.rs"]
 mod subject_deletion_store;
 pub(crate) use subject_deletion_store::{
-    prior_removal_matches as subject_deletion_prior_removal_matches,
+    SubjectDeletionRemoval, prior_removal_matches as subject_deletion_prior_removal_matches,
+    removal_reconciliation_matches as subject_deletion_removal_matches,
     worker_kind as subject_deletion_worker_kind,
 };
 
@@ -16255,7 +16256,8 @@ fn current_worker_singleton_owner(
     let owner = conn
         .query_row(
             "SELECT owner.runtime_id, owner.worker_id, owner.generation, \
-                    CASE WHEN substr(owner.singleton_key,1,10)='subjektiv:' AND reservation.state='removed' \
+                    CASE WHEN reservation.state='removed' AND EXISTS(SELECT 1 FROM subject_deletion_removals d WHERE d.workspace_id=owner.workspace_id AND d.runtime_id=owner.runtime_id AND d.worker_id=owner.worker_id AND owner.singleton_key IN ('subjektiv:'||d.subject_id,'subjektiv-consolidation:'||d.subject_id)) THEN 'created' \
+                    WHEN substr(owner.singleton_key,1,10)='subjektiv:' AND reservation.state='removed' \
                       AND NOT EXISTS(SELECT 1 FROM subject_worker_releases r WHERE r.workspace_id=owner.workspace_id AND r.worker_id=owner.worker_id AND r.runtime_id=owner.runtime_id AND r.singleton_key=owner.singleton_key AND r.singleton_generation=owner.generation AND r.completed=1) \
                       AND NOT EXISTS(SELECT 1 FROM subject_create_recoveries r WHERE r.workspace_id=owner.workspace_id AND r.worker_id=owner.worker_id AND r.runtime_id=owner.runtime_id AND r.singleton_key=owner.singleton_key AND r.singleton_generation=owner.generation AND r.create_fingerprint=reservation.create_fingerprint AND r.completed=1) \
                       AND NOT EXISTS(SELECT 1 FROM worker_removal_operations r JOIN subject_deletion_workers d ON d.workspace_id=r.workspace_id AND d.runtime_id=r.runtime_id AND d.worker_id=r.worker_id WHERE r.workspace_id=owner.workspace_id AND r.runtime_id=owner.runtime_id AND r.worker_id=owner.worker_id AND r.state='succeeded' AND owner.singleton_key='subjektiv:'||d.subject_id AND (d.prior_removal_operation_id=r.operation_id OR (d.prior_removal_operation_id IS NULL AND d.is_job=0 AND r.session_disposition='archive' AND r.archive_retention_kind='forever' AND r.metadata_disposition='tombstone'))) \
@@ -16278,7 +16280,15 @@ fn current_worker_singleton_owner(
         )
         .optional()?;
     owner
-        .map(|(runtime_id, worker_id, generation, state)| {
+        .map(|(runtime_id, worker_id, generation, mut state)| {
+            let worker = RuntimeWorkerRef::new(runtime_id, worker_id);
+            if let Some(subject) = singleton_key.strip_prefix("subjektiv:").or_else(|| singleton_key.strip_prefix("subjektiv-consolidation:"))
+                && state.as_deref() == Some("created")
+                && subject_deletion_store::completed_removal_matches(conn, workspace_id, subject, &worker)?
+            {
+                let removed: bool = conn.query_row("SELECT EXISTS(SELECT 1 FROM worker_create_reservations WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND state='removed')",params![workspace_id,worker.runtime_id,worker.worker_id],|r|r.get(0))?;
+                if removed { state = Some("removed".into()); }
+            }
             let generation = generation.try_into().map_err(|_| {
                 Error::Store(format!(
                     "Worker singleton key {singleton_key:?} has an invalid generation"
@@ -16288,7 +16298,7 @@ fn current_worker_singleton_owner(
                 WorkerSingletonLease {
                     key: singleton_key.to_string(),
                     generation,
-                    worker: RuntimeWorkerRef::new(runtime_id, worker_id),
+                    worker,
                 },
                 state,
             ))
@@ -26903,7 +26913,7 @@ INSERT INTO worker_registry (
                 .query_map([],|r|Ok((r.get::<_,String>(0)?,r.get::<_,String>(1)?)))?.collect::<rusqlite::Result<Vec<_>>>()?;
             for (name,table) in triggers {conn.execute_batch(&format!("DROP TRIGGER {name}; CREATE TRIGGER {name} BEFORE DELETE ON {table} BEGIN SELECT RAISE(ABORT,'Notification evidence is immutable'); END;"))?;}
             for name in ["subject_deleted_job_insert","subject_deleted_job_result","subject_deleted_attempt_insert","subject_deleted_attempt_result","subject_deleted_worker_reservation","subject_deleted_notification_source","subject_deleted_notification_payload","subject_deleted_notification_send","subject_deletion_worker_prior_insert","subject_deletion_worker_identity_immutable"] {conn.execute_batch(&format!("DROP TRIGGER {name}"))?;}
-            conn.execute_batch("DROP TABLE subject_deletion_allocations; DROP TABLE subject_deletion_jobs; DROP TABLE subject_deletion_workers; DROP TABLE subject_deletions;")?;
+            conn.execute_batch("DROP TABLE subject_deletion_removals; DROP TABLE subject_deletion_allocations; DROP TABLE subject_deletion_jobs; DROP TABLE subject_deletion_workers; DROP TABLE subject_deletions;")?;
             conn.execute_batch("DROP TRIGGER backend_job_input_retirement; DROP TRIGGER backend_job_identity_immutable;
                 ALTER TABLE backend_jobs DROP COLUMN input_disposition;
                 DROP TRIGGER passkey_metadata_insert;

crates/workspace-server/src/subject_deletion_schema.sql

@@ -35,6 +35,41 @@ BEGIN SELECT RAISE(ABORT,'Subject deletion Worker snapshot identity is immutable
 CREATE TRIGGER subject_deletion_original_absence_monotonic BEFORE UPDATE OF originals_absent ON subject_deletion_workers
 WHEN OLD.originals_absent=1 AND NEW.originals_absent=0
 BEGIN SELECT RAISE(ABORT,'Subject deletion original absence proof is permanent'); END;
+-- In-flight removals are distinct from successful prior receipts. Pin the exact
+-- old Runtime request once; Subject deletion never rewrites its retention policy.
+CREATE TABLE subject_deletion_removals (
+    workspace_id TEXT NOT NULL,
+    subject_id TEXT NOT NULL,
+    runtime_id TEXT NOT NULL,
+    worker_id TEXT NOT NULL,
+    operation_id TEXT NOT NULL REFERENCES worker_removal_operations(operation_id),
+    input_fingerprint TEXT NOT NULL,
+    request_json TEXT NOT NULL CHECK(json_valid(request_json)),
+    completed INTEGER NOT NULL DEFAULT 0 CHECK(completed IN (0,1)),
+    job_purge_receipt_json TEXT CHECK(job_purge_receipt_json IS NULL OR json_valid(job_purge_receipt_json)),
+    PRIMARY KEY(workspace_id,subject_id,runtime_id,worker_id),
+    FOREIGN KEY(workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE,
+    FOREIGN KEY(workspace_id,subject_id,runtime_id,worker_id)
+      REFERENCES subject_deletion_workers(workspace_id,subject_id,runtime_id,worker_id) ON DELETE CASCADE
+);
+CREATE TRIGGER subject_deletion_removal_insert BEFORE INSERT ON subject_deletion_removals
+WHEN NEW.completed!=0 OR NEW.job_purge_receipt_json IS NOT NULL OR NOT EXISTS(
+ SELECT 1 FROM subject_deletion_workers w JOIN worker_removal_operations r
+ ON r.workspace_id=w.workspace_id AND r.runtime_id=w.runtime_id AND r.worker_id=w.worker_id
+ WHERE w.workspace_id=NEW.workspace_id AND w.subject_id=NEW.subject_id
+ AND w.runtime_id=NEW.runtime_id AND w.worker_id=NEW.worker_id AND w.prior_removal_operation_id IS NULL
+ AND r.operation_id=NEW.operation_id AND r.input_fingerprint=NEW.input_fingerprint
+ AND r.state IN ('planned','executing','failed')
+ AND json_extract(NEW.request_json,'$.operation_id')=r.operation_id
+ AND json_extract(NEW.request_json,'$.input_fingerprint')=r.input_fingerprint)
+BEGIN SELECT RAISE(ABORT,'Subject deletion reconciliation must pin an unfinished canonical removal'); END;
+CREATE TRIGGER subject_deletion_removal_identity_immutable BEFORE UPDATE OF
+ workspace_id,subject_id,runtime_id,worker_id,operation_id,input_fingerprint,request_json ON subject_deletion_removals
+BEGIN SELECT RAISE(ABORT,'Subject deletion removal request is immutable'); END;
+CREATE TRIGGER subject_deletion_removal_completion_monotonic BEFORE UPDATE ON subject_deletion_removals
+WHEN (OLD.completed=1 AND NEW.completed!=1)
+ OR (OLD.job_purge_receipt_json IS NOT NULL AND NEW.job_purge_receipt_json IS NOT OLD.job_purge_receipt_json)
+BEGIN SELECT RAISE(ABORT,'Subject deletion removal receipt is permanent'); END;
 -- Job/attempt identities remain after payload erasure, preventing stale allocation
 -- or caller-selected Job-id reuse from recreating the deleted Subject's inputs.
 CREATE TABLE subject_deletion_jobs (

crates/workspace-server/src/subject_deletion_store.rs

@@ -5,14 +5,184 @@
 //! All historical Job attempts and dispatch allocations are snapshotted, not just
 //! the current attempt. Pending/unknown work becomes failed/subject_deleted (not a
 //! successful Job). Originals in body/delegated Sessions are archived forever;
-//! Existing canonical Worker removals and archives are outside this deletion:
+//! Successful canonical Worker removals and archives are outside this deletion:
 //! prepare pins their exact successful receipt, regardless of its disposition.
+//! Unfinished removals instead pin their immutable old execution request; exact
+//! reconciliation and (for Jobs) additional physical erasure must complete first.
 //! Newly removed Job Sessions/diagnostics need canonical Purge before input,
 //! result, notification payloads and live Job/cache references can be erased.
 //! This DB has no separate Job content cache. Runtime caches are the canonical
 //! Worker removal's responsibility. Notification text already delivered into an
 //! ordinary conversation follows that conversation's archive lifecycle.
 use super::*;
+use worker_runtime::retention::{
+    DiagnosticsDisposition, WorkerRetentionExecutionRequest, WorkerRetentionExecutionResult,
+};
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub(crate) struct SubjectDeletionRemoval {
+    pub request: WorkerRetentionExecutionRequest,
+    pub completed: bool,
+    pub is_job: bool,
+}
+
+fn decode_request(json: &str) -> Result<WorkerRetentionExecutionRequest> {
+    serde_json::from_str(json).map_err(|e| Error::Store(e.to_string()))
+}
+
+// Only source metadata comes from the saved Worker snapshot. Everything owned by
+// the canonical plan is rechecked, even after its success/catalog removal.
+fn canonical_request(
+    conn: &Connection,
+    operation: &str,
+    source_created_at: &str,
+    profile: Option<&str>,
+) -> Result<Option<(WorkerRetentionExecutionRequest, String)>> {
+    let row: Option<(String, String)> = conn.query_row(
+        "SELECT json_object('operation_id',operation_id,'input_fingerprint',input_fingerprint,
+          'archive_id',archive_id,'workspace_id',workspace_id,'source_runtime_id',runtime_id,
+          'worker_id',worker_id,'source_created_at',?2,'removed_at',created_at,
+          'effective_profile',?3,'retention_class',NULL,'policy_id',policy_id,
+          'session_disposition',session_disposition,'diagnostics_disposition',diagnostics_disposition),state
+         FROM worker_removal_operations WHERE operation_id=?1",
+        params![operation,source_created_at,profile], |r| Ok((r.get(0)?,r.get(1)?)),
+    ).optional()?;
+    row.map(|(json, state)| Ok((decode_request(&json)?, state)))
+        .transpose()
+}
+
+fn saved_removal(
+    conn: &Connection,
+    workspace: &str,
+    subject: Option<&str>,
+    worker: &RuntimeWorkerRef,
+) -> Result<Option<SubjectDeletionRemoval>> {
+    let rows = conn.prepare("SELECT d.request_json,d.completed,w.is_job FROM subject_deletion_removals d
+        JOIN subject_deletion_workers w USING(workspace_id,subject_id,runtime_id,worker_id)
+        WHERE d.workspace_id=?1 AND d.runtime_id=?2 AND d.worker_id=?3 AND (?4 IS NULL OR d.subject_id=?4)
+        ORDER BY d.subject_id")?
+        .query_map(params![workspace,worker.runtime_id,worker.worker_id,subject], |r| Ok((r.get::<_,String>(0)?,r.get::<_,bool>(1)?,r.get::<_,bool>(2)?)))?
+        .collect::<rusqlite::Result<Vec<_>>>()?;
+    let mut saved: Option<SubjectDeletionRemoval> = None;
+    for (json, completed, is_job) in rows {
+        let request = decode_request(&json)?;
+        if let Some(saved) = saved.as_mut() {
+            if saved.request != request {
+                return Err(Error::RepositoryConflict(
+                    "Conflicting Subject deletion removal snapshots".into(),
+                ));
+            }
+            saved.completed &= completed;
+            saved.is_job |= is_job;
+        } else {
+            saved = Some(SubjectDeletionRemoval {
+                request,
+                completed,
+                is_job,
+            });
+        }
+    }
+    Ok(saved)
+}
+
+pub(crate) fn removal_reconciliation_matches(
+    conn: &Connection,
+    workspace: &str,
+    worker: &RuntimeWorkerRef,
+    operation: &str,
+    fp: &str,
+) -> Result<bool> {
+    let Some(saved) = saved_removal(conn, workspace, None, worker)? else {
+        return Ok(false);
+    };
+    let request = &saved.request;
+    if request.workspace_id != workspace
+        || request.source_runtime_id != worker.runtime_id
+        || request.worker_id.to_string() != worker.worker_id
+        || request.operation_id != operation
+        || request.input_fingerprint != fp
+    {
+        return Ok(false);
+    }
+    Ok(canonical_request(
+        conn,
+        operation,
+        &request.source_created_at,
+        request.effective_profile.as_deref(),
+    )?
+    .is_some_and(|(canonical, state)| {
+        canonical == *request
+            && matches!(
+                state.as_str(),
+                "planned" | "executing" | "failed" | "succeeded"
+            )
+    }))
+}
+
+pub(super) fn completed_removal_matches(
+    conn: &Connection,
+    workspace: &str,
+    subject: &str,
+    worker: &RuntimeWorkerRef,
+) -> Result<bool> {
+    let Some(saved) = saved_removal(conn, workspace, Some(subject), worker)? else {
+        return Ok(false);
+    };
+    if !saved.completed
+        || !removal_reconciliation_matches(
+            conn,
+            workspace,
+            worker,
+            &saved.request.operation_id,
+            &saved.request.input_fingerprint,
+        )?
+    {
+        return Ok(false);
+    }
+    Ok(conn.query_row(
+        "SELECT EXISTS(SELECT 1 FROM subject_deletion_removals d
+         JOIN worker_removal_operations r ON r.operation_id=d.operation_id AND r.input_fingerprint=d.input_fingerprint
+         WHERE d.workspace_id=?1 AND d.subject_id=?2 AND d.runtime_id=?3 AND d.worker_id=?4
+         AND r.state='succeeded' AND d.completed=1 AND (?5=0 OR (
+           json_extract(d.job_purge_receipt_json,'$.workspace_id')=d.workspace_id
+           AND json_extract(d.job_purge_receipt_json,'$.worker_id')=d.worker_id
+           AND json_extract(d.job_purge_receipt_json,'$.operation_id')=d.operation_id
+           AND json_extract(d.job_purge_receipt_json,'$.input_fingerprint')=d.input_fingerprint)))",
+        params![workspace,subject,worker.runtime_id,worker.worker_id,saved.is_job],
+        |r| r.get(0),
+    )?)
+}
+
+fn pin_removal(
+    conn: &Connection,
+    workspace: &str,
+    subject: &str,
+    worker: &RuntimeWorkerRef,
+) -> Result<()> {
+    let operation: Option<String> = conn.query_row("SELECT operation_id FROM worker_removal_operations
+        WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND state IN ('planned','executing','failed')
+        AND (state IN ('executing','failed') OR worker_updated_at=(SELECT updated_at FROM worker_registry WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3))
+        ORDER BY created_at DESC,operation_id DESC LIMIT 1",
+        params![workspace,worker.runtime_id,worker.worker_id],|r|r.get(0)).optional()?;
+    let Some(operation) = operation else {
+        return Ok(());
+    };
+    let source: Option<(String,Option<String>)> = conn.query_row("SELECT created_at,profile FROM worker_registry WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3",
+        params![workspace,worker.runtime_id,worker.worker_id], |r| Ok((r.get(0)?,r.get(1)?))).optional()?;
+    let (created_at, profile) = source.ok_or_else(|| {
+        Error::RepositoryConflict(
+            "Unfinished Worker removal has no source identity to snapshot".into(),
+        )
+    })?;
+    let (request, _) = canonical_request(conn, &operation, &created_at, profile.as_deref())?
+        .ok_or_else(|| {
+            Error::RepositoryConflict("Canonical removal disappeared while snapshotting".into())
+        })?;
+    let json = serde_json::to_string(&request).map_err(|e| Error::Store(e.to_string()))?;
+    conn.execute("INSERT INTO subject_deletion_removals(workspace_id,subject_id,runtime_id,worker_id,operation_id,input_fingerprint,request_json)
+        VALUES(?1,?2,?3,?4,?5,?6,?7)",params![workspace,subject,worker.runtime_id,worker.worker_id,operation,request.input_fingerprint,json])?;
+    Ok(())
+}
 
 const MIGRATION_NAME: &str = "Durable Subject deletion snapshots and Job admission fences";
 
@@ -103,6 +273,10 @@ pub(super) fn verify(conn: &Connection) -> Result<()> {
     for name in [
         "subject_deletions",
         "subject_deletion_workers",
+        "subject_deletion_removals",
+        "subject_deletion_removal_insert",
+        "subject_deletion_removal_identity_immutable",
+        "subject_deletion_removal_completion_monotonic",
         "subject_deletion_jobs",
         "subject_deletion_allocations",
         "subject_deleted_job_insert",
@@ -251,6 +425,72 @@ fn worker_fence_request(
 }
 
 impl SqliteWorkspaceStore {
+    pub(crate) fn subject_deletion_removal(
+        &self,
+        workspace: &str,
+        worker: &RuntimeWorkerRef,
+    ) -> Result<Option<super::SubjectDeletionRemoval>> {
+        self.with_conn(|conn| saved_removal(conn, workspace, None, worker))
+    }
+
+    /// A canonical success alone is not proof of additional Job erasure. Keep the
+    /// old receipt/dispositions intact and only complete this separate snapshot
+    /// after the exact Runtime result and (for Jobs) physical purge response.
+    pub(crate) fn confirm_subject_deletion_removal(
+        &self,
+        workspace: &str,
+        worker: &RuntimeWorkerRef,
+        request: &WorkerRetentionExecutionRequest,
+        result: &WorkerRetentionExecutionResult,
+        purge: Option<&runtime_api::SubjectJobRemovalPurgeResponse>,
+    ) -> Result<()> {
+        self.with_conn_mut(|conn| {
+            let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
+            let saved = saved_removal(&tx,workspace,None,worker)?
+                .ok_or_else(|| Error::RepositoryConflict("No saved Subject deletion removal".into()))?;
+            if saved.request != *request
+                || !removal_reconciliation_matches(&tx,workspace,worker,&request.operation_id,&request.input_fingerprint)?
+                || !tx.query_row("SELECT state='succeeded' FROM worker_removal_operations WHERE operation_id=?1",[&request.operation_id],|r|r.get::<_,bool>(0))?
+                || result.operation_id != request.operation_id || result.input_fingerprint != request.input_fingerprint
+                || result.worker_id != request.worker_id || result.session_disposition != request.session_disposition
+                || result.diagnostics_disposition != request.diagnostics_disposition || !result.source_removed
+                || result.diagnostics_retained != (request.diagnostics_disposition == DiagnosticsDisposition::Retain) {
+                return Err(Error::RepositoryConflict("Subject deletion removal requires exact saved request and canonical Runtime success".into()));
+            }
+            let archive_matches = match (&result.archive, request.session_disposition) {
+                (Some(archive), worker_runtime::retention::SessionDisposition::Archive) => {
+                    request.archive_id.as_deref() == Some(archive.archive_id.as_str())
+                        && archive.workspace_id == workspace
+                        && archive.source_runtime_id == worker.runtime_id
+                        && archive.source_worker_id == request.worker_id
+                        && archive.operation_id == request.operation_id
+                        && archive.input_fingerprint == request.input_fingerprint
+                        && archive.policy_id == request.policy_id
+                        && archive.source_created_at == request.source_created_at
+                }
+                (None, worker_runtime::retention::SessionDisposition::Purge) => true,
+                _ => false,
+            };
+            if !archive_matches {
+                return Err(Error::RepositoryConflict("Subject deletion removal archive identity mismatch".into()));
+            }
+            if saved.is_job && purge.is_none() {
+                return Err(Error::RepositoryConflict("Subject Job removal requires an additional physical purge receipt".into()));
+            }
+            if let Some(purge) = purge {
+                if purge.workspace_id != workspace || purge.worker_id != request.worker_id
+                    || purge.operation_id != request.operation_id || purge.input_fingerprint != request.input_fingerprint {
+                    return Err(Error::RepositoryConflict("Subject Job physical purge receipt identity mismatch".into()));
+                }
+            }
+            let purge_json = purge.map(serde_json::to_string).transpose().map_err(|e| Error::Store(e.to_string()))?;
+            tx.execute("UPDATE subject_deletion_removals SET completed=1,job_purge_receipt_json=coalesce(job_purge_receipt_json,?4)
+                WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3",params![workspace,worker.runtime_id,worker.worker_id,purge_json])?;
+            tx.commit()?;
+            Ok(())
+        })
+    }
+
     pub(crate) fn subject_deletion_worker_request(
         &self,
         workspace: &str,
@@ -316,6 +556,12 @@ impl SqliteWorkspaceStore {
                 tx.execute("INSERT INTO subject_deletion_workers(workspace_id,subject_id,runtime_id,worker_id,is_job,prior_removal_operation_id) VALUES(?1,?2,?3,?4,?5,?6)
                     ON CONFLICT(workspace_id,subject_id,runtime_id,worker_id) DO UPDATE SET is_job=max(is_job,excluded.is_job)",params![workspace,subject,worker.runtime_id,worker.worker_id,is_job,prior])?;
             }
+            // Deduplicate workers before pinning: an identity can be both a
+            // delegated Worker and a Job attempt; the final is_job wins.
+            let pending = tx.prepare("SELECT runtime_id,worker_id FROM subject_deletion_workers WHERE workspace_id=?1 AND subject_id=?2 AND prior_removal_operation_id IS NULL")?
+                .query_map(params![workspace,subject],|r|Ok(RuntimeWorkerRef::new(r.get::<_,String>(0)?,r.get::<_,String>(1)?)))?
+                .collect::<rusqlite::Result<Vec<_>>>()?;
+            for worker in pending { pin_removal(&tx,workspace,subject,&worker)?; }
             let now = chrono::Utc::now().to_rfc3339();
             tx.execute("UPDATE backend_job_attempts SET state='failed',failure_category='subject_deleted',failure_detail=NULL,completed_at=?3,updated_at=?3,
                 worker_cleanup_state=CASE WHEN worker_id IS NULL THEN 'completed' ELSE 'pending' END,
@@ -362,6 +608,12 @@ impl SqliteWorkspaceStore {
                 || fence.create_fingerprint != request.create_fingerprint || fence.operation_id != request.operation_id {
                 return Err(Error::RepositoryConflict("Subject deletion absence fence identity mismatch".into()));
             }
+            // An in-flight pin cannot be bypassed by UnknownWorker/absence or
+            // a different later canonical success, even if that success is Purge.
+            if saved_removal(conn,workspace,Some(subject),worker)?.is_some() {
+                if completed_removal_matches(conn,workspace,subject,worker)? { return Ok(()); }
+                return Err(Error::RepositoryConflict("Subject deletion removal reconciliation is incomplete".into()));
+            }
             if fence.originals_absent {
                 let registered: bool = conn.query_row("SELECT EXISTS(SELECT 1 FROM worker_registry WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3)", params![workspace,worker.runtime_id,worker.worker_id], |r|r.get(0))?;
                 if registered { return Err(Error::RepositoryConflict("Registered Worker requires canonical original retention".into())); }
@@ -474,7 +726,19 @@ fn require_worker_receipts(
     subject: &str,
     jobs_only: bool,
 ) -> Result<()> {
+    let pinned = conn.prepare("SELECT d.runtime_id,d.worker_id FROM subject_deletion_removals d JOIN subject_deletion_workers w USING(workspace_id,subject_id,runtime_id,worker_id)
+        WHERE d.workspace_id=?1 AND d.subject_id=?2 AND (?3=0 OR w.is_job=1)")?
+        .query_map(params![workspace,subject,jobs_only],|r|Ok(RuntimeWorkerRef::new(r.get::<_,String>(0)?,r.get::<_,String>(1)?)))?
+        .collect::<rusqlite::Result<Vec<_>>>()?;
+    for worker in pinned {
+        if !completed_removal_matches(conn, workspace, subject, &worker)? {
+            return Err(Error::RepositoryConflict(
+                "Subject deletion removal reconciliation receipt is incomplete".into(),
+            ));
+        }
+    }
     let missing: bool = conn.query_row("SELECT EXISTS(SELECT 1 FROM subject_deletion_workers w WHERE w.workspace_id=?1 AND w.subject_id=?2 AND (?3=0 OR w.is_job=1) AND w.originals_absent=0
+        AND NOT EXISTS(SELECT 1 FROM subject_deletion_removals d WHERE d.workspace_id=w.workspace_id AND d.subject_id=w.subject_id AND d.runtime_id=w.runtime_id AND d.worker_id=w.worker_id)
         AND NOT (
           (w.prior_removal_operation_id IS NOT NULL AND EXISTS(SELECT 1 FROM worker_removal_operations r WHERE r.operation_id=w.prior_removal_operation_id AND r.workspace_id=w.workspace_id AND r.runtime_id=w.runtime_id AND r.worker_id=w.worker_id AND r.state='succeeded'))
           OR (w.prior_removal_operation_id IS NULL AND (
@@ -1108,6 +1372,8 @@ mod tests {
             "job-owner",
             Some("subjektiv-consolidation:subject"),
         );
+        registry(&s, &job_worker);
+        registry(&s, &body);
         receipt(&s, &job_worker, "executing", "archive", "retain");
         receipt(&s, &body, "failed", "purge", "purge");
         s.prepare_subject_deletion("w", "subject", "delete", std::slice::from_ref(&body))
@@ -1134,8 +1400,11 @@ mod tests {
             c.execute("UPDATE worker_create_reservations SET state='removed' WHERE workspace_id='w'",[])?;
             Ok(())
         }).unwrap();
+        assert!(
             s.erase_subject_deleted_jobs("w", "subject", "delete")
-            .unwrap();
+                .is_err(),
+            "replacement canonical Purge cannot bypass the pinned unfinished removal"
+        );
         assert!(
             s.finish_subject_deletion_record("w", "subject", "delete")
                 .is_err(),
@@ -1143,6 +1412,389 @@ mod tests {
         );
     }
 
+    fn in_flight(s: &SqliteWorkspaceStore, w: &RuntimeWorkerRef, state: &str, is_job: bool) {
+        registry(s, w);
+        reservation(
+            s,
+            w,
+            "pending",
+            Some(if is_job {
+                "subjektiv-consolidation:subject"
+            } else {
+                "subjektiv:subject"
+            }),
+        );
+        receipt(s, w, state, "archive", "retain");
+        s.with_conn(|c| {
+            c.execute("UPDATE worker_registry SET profile='builtin:old-profile' WHERE worker_id=?1",[&w.worker_id])?;
+            c.execute("UPDATE worker_removal_operations SET archive_id='old-archive',metadata_disposition='purge',archive_retention_kind='for_seconds',archive_retention_seconds=60 WHERE worker_id=?1",[&w.worker_id])?;
+            Ok(())
+        }).unwrap();
+    }
+
+    fn execution_result(
+        request: &WorkerRetentionExecutionRequest,
+    ) -> WorkerRetentionExecutionResult {
+        WorkerRetentionExecutionResult {
+            operation_id: request.operation_id.clone(),
+            input_fingerprint: request.input_fingerprint.clone(),
+            worker_id: request.worker_id,
+            session_disposition: request.session_disposition,
+            diagnostics_disposition: request.diagnostics_disposition,
+            archive: request.archive_id.as_ref().map(|archive_id| {
+                worker_runtime::retention::WorkerSessionArchiveManifest {
+                    schema_version: 1,
+                    archive_id: archive_id.clone(),
+                    workspace_id: request.workspace_id.clone(),
+                    source_runtime_id: request.source_runtime_id.clone(),
+                    source_worker_id: request.worker_id,
+                    source_session_id: "old-session".into(),
+                    source_active_segment_id: None,
+                    segment_ids: vec![],
+                    source_created_at: request.source_created_at.clone(),
+                    removed_at: request.removed_at.clone(),
+                    archived_at_unix_seconds: 0,
+                    effective_profile: request.effective_profile.clone(),
+                    retention_class: request.retention_class.clone(),
+                    content_checksum_sha256: "fixture-checksum".into(),
+                    content_bytes: 0,
+                    content_file_count: 0,
+                    policy_id: request.policy_id.clone(),
+                    operation_id: request.operation_id.clone(),
+                    input_fingerprint: request.input_fingerprint.clone(),
+                }
+            }),
+            source_removed: true,
+            diagnostics_retained: request.diagnostics_disposition == DiagnosticsDisposition::Retain,
+        }
+    }
+
+    fn succeed(s: &SqliteWorkspaceStore, request: &WorkerRetentionExecutionRequest) {
+        // Isolated store proof fixture. HTTP/Runtime tests exercise canonical
+        // commit; here success alone must still leave reconciliation incomplete.
+        s.with_conn(|c| {
+            c.execute("UPDATE worker_removal_operations SET state='succeeded' WHERE operation_id=?1",[&request.operation_id])?;
+            c.execute("DELETE FROM worker_registry WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3",params![request.workspace_id,request.source_runtime_id,request.worker_id.to_string()])?;
+            c.execute("UPDATE worker_create_reservations SET state='removed' WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3",params![request.workspace_id,request.source_runtime_id,request.worker_id.to_string()])?;
+            Ok(())
+        }).unwrap();
+    }
+
+    #[test]
+    fn unfinished_removal_pins_full_identity_and_requires_exact_success_before_release() {
+        let s = setup();
+        let w = worker(71);
+        in_flight(&s, &w, "executing", false);
+        // An executing plan may have an older registry epoch. Its exact Runtime
+        // request is still authoritative; do not turn this into permanent stale.
+        s.with_conn(|c| {
+            c.execute(
+                "UPDATE worker_registry SET updated_at='later' WHERE worker_id=?1",
+                [&w.worker_id],
+            )?;
+            Ok(())
+        })
+        .unwrap();
+        s.prepare_subject_deletion("w", "subject", "delete", &[])
+            .unwrap();
+        let saved = s.subject_deletion_removal("w", &w).unwrap().unwrap();
+        let req = &saved.request;
+        assert!(!saved.completed);
+        assert!(!saved.is_job);
+        assert_eq!(req.source_created_at, NOW);
+        assert_eq!(req.removed_at, NOW);
+        assert_eq!(
+            req.effective_profile.as_deref(),
+            Some("builtin:old-profile")
+        );
+        assert_eq!(req.retention_class, None);
+        assert_eq!(req.archive_id.as_deref(), Some("old-archive"));
+        assert_eq!(req.policy_id, "test");
+        assert_eq!(saved_prior(&s, "subject", &w), None);
+        s.with_conn(|c| {
+            assert!(removal_reconciliation_matches(c,"w",&w,&req.operation_id,&req.input_fingerprint)?);
+            assert!(!removal_reconciliation_matches(c,"w",&w,&req.operation_id,"changed")?);
+            assert!(!removal_reconciliation_matches(c,"other",&w,&req.operation_id,&req.input_fingerprint)?);
+            assert!(c.execute("UPDATE subject_deletion_removals SET request_json=request_json WHERE workspace_id='w'",[]).is_err());
+            Ok(())
+        }).unwrap();
+        let result = execution_result(req);
+        assert!(
+            s.confirm_subject_deletion_removal("w", &w, req, &result, None)
+                .is_err()
+        );
+        succeed(&s, req);
+        assert!(
+            s.current_worker_singleton_owner("w", "subjektiv:subject")
+                .unwrap()
+                .is_some()
+        );
+        assert!(
+            s.finish_subject_deletion_record("w", "subject", "delete")
+                .is_err()
+        );
+        let fence_request = s
+            .subject_deletion_worker_request("w", "subject", "delete", &w)
+            .unwrap();
+        let fence = runtime_api::SubjectDeleteWorkerResponse {
+            workspace_id: fence_request.workspace_id,
+            worker_id: fence_request.worker_id,
+            create_fingerprint: fence_request.create_fingerprint,
+            operation_id: fence_request.operation_id,
+            originals_absent: true,
+        };
+        assert!(
+            s.confirm_subject_deletion_absence("w", "subject", "delete", &w, &fence)
+                .is_err()
+        );
+        for (field, value) in [
+            ("operation_id", serde_json::json!("changed")),
+            ("input_fingerprint", serde_json::json!("changed")),
+            ("archive_id", serde_json::json!("changed")),
+            ("workspace_id", serde_json::json!("other")),
+            ("source_runtime_id", serde_json::json!("other-runtime")),
+            ("worker_id", serde_json::json!(worker(72).worker_id)),
+            ("source_created_at", serde_json::json!("changed")),
+            ("removed_at", serde_json::json!("changed")),
+            ("effective_profile", serde_json::json!("changed")),
+            ("retention_class", serde_json::json!("changed")),
+            ("policy_id", serde_json::json!("changed")),
+            ("session_disposition", serde_json::json!("purge")),
+            ("diagnostics_disposition", serde_json::json!("purge")),
+        ] {
+            let mut changed = serde_json::to_value(req).unwrap();
+            changed[field] = value;
+            let changed: WorkerRetentionExecutionRequest = serde_json::from_value(changed).unwrap();
+            assert!(
+                s.confirm_subject_deletion_removal("w", &w, &changed, &result, None)
+                    .is_err(),
+                "{field}"
+            );
+            assert!(
+                !s.subject_deletion_removal("w", &w)
+                    .unwrap()
+                    .unwrap()
+                    .completed
+            );
+        }
+        for (field, value) in [
+            ("operation_id", serde_json::json!("changed")),
+            ("input_fingerprint", serde_json::json!("changed")),
+            ("worker_id", serde_json::json!(worker(72).worker_id)),
+            ("session_disposition", serde_json::json!("purge")),
+            ("diagnostics_disposition", serde_json::json!("purge")),
+            ("source_removed", serde_json::json!(false)),
+            ("diagnostics_retained", serde_json::json!(false)),
+            ("archive", serde_json::Value::Null),
+        ] {
+            let mut changed = serde_json::to_value(&result).unwrap();
+            changed[field] = value;
+            let changed = serde_json::from_value(changed).unwrap();
+            assert!(
+                s.confirm_subject_deletion_removal("w", &w, req, &changed, None)
+                    .is_err(),
+                "{field}"
+            );
+        }
+        s.confirm_subject_deletion_removal("w", &w, req, &result, None)
+            .unwrap();
+        assert_eq!(
+            saved_prior(&s, "subject", &w),
+            None,
+            "not a late prior success"
+        );
+        assert!(
+            s.current_worker_singleton_owner("w", "subjektiv:subject")
+                .unwrap()
+                .is_none()
+        );
+        s.confirm_subject_deletion_absence("w", "subject", "delete", &w, &fence)
+            .unwrap();
+        s.erase_subject_deleted_jobs("w", "subject", "delete")
+            .unwrap();
+        s.finish_subject_deletion_record("w", "subject", "delete")
+            .unwrap();
+        s.with_conn(|c| {
+            assert!(removal_reconciliation_matches(
+                c,
+                "w",
+                &w,
+                &req.operation_id,
+                &req.input_fingerprint
+            )?);
+            c.execute(
+                "UPDATE worker_removal_operations SET policy_id='changed' WHERE operation_id=?1",
+                [&req.operation_id],
+            )?;
+            assert!(!completed_removal_matches(c, "w", "subject", &w)?);
+            Ok(())
+        })
+        .unwrap();
+        assert!(
+            s.finish_subject_deletion_record("w", "subject", "delete")
+                .is_err()
+        );
+    }
+
+    #[test]
+    fn job_reconciliation_requires_additional_purge_and_survives_reopen_and_retry() {
+        let dir = tempfile::tempdir().unwrap();
+        let path = dir.path().join("snapshot.db");
+        let s = SqliteWorkspaceStore::open_for_test(&path).unwrap();
+        seed(&s);
+        let w = worker(81);
+        in_flight(&s, &w, "failed", true);
+        let other = worker(82);
+        registry(&s, &other);
+        reservation(&s, &other, "other-allocation", Some("subjektiv:another"));
+        receipt(&s, &other, "executing", "archive", "retain");
+        s.prepare_subject_deletion("w", "another", "other-delete", &[])
+            .unwrap();
+        let other_before = s.subject_deletion_removal("w", &other).unwrap();
+        s.prepare_subject_deletion("w", "subject", "delete", &[])
+            .unwrap();
+        let req = s
+            .subject_deletion_removal("w", &w)
+            .unwrap()
+            .unwrap()
+            .request;
+        drop(s);
+        let s = SqliteWorkspaceStore::open_for_test(&path).unwrap();
+        s.prepare_subject_deletion("w", "subject", "delete", &[])
+            .unwrap();
+        assert_eq!(
+            s.subject_deletion_removal("w", &w)
+                .unwrap()
+                .unwrap()
+                .request,
+            req
+        );
+        assert!(s.subject_deletion_removal("other", &w).unwrap().is_none());
+        let result = execution_result(&req);
+        let purge = runtime_api::SubjectJobRemovalPurgeResponse {
+            workspace_id: "w".into(),
+            worker_id: req.worker_id,
+            operation_id: req.operation_id.clone(),
+            input_fingerprint: req.input_fingerprint.clone(),
+        };
+        assert!(
+            s.confirm_subject_deletion_removal("w", &w, &req, &result, Some(&purge))
+                .is_err(),
+            "failed is not success"
+        );
+        succeed(&s, &req);
+        assert!(
+            s.current_worker_singleton_owner("w", "subjektiv-consolidation:subject")
+                .unwrap()
+                .is_some(),
+            "canonical success cannot release Job occupancy before physical purge"
+        );
+        assert!(
+            s.confirm_subject_deletion_removal("w", &w, &req, &result, None)
+                .is_err()
+        );
+        assert!(
+            s.erase_subject_deleted_jobs("w", "subject", "delete")
+                .is_err()
+        );
+        for (field, value) in [
+            ("workspace_id", serde_json::json!("other")),
+            ("worker_id", serde_json::json!(worker(82).worker_id)),
+            ("operation_id", serde_json::json!("changed")),
+            ("input_fingerprint", serde_json::json!("changed")),
+        ] {
+            let mut changed = serde_json::to_value(&purge).unwrap();
+            changed[field] = value;
+            let changed = serde_json::from_value(changed).unwrap();
+            assert!(
+                s.confirm_subject_deletion_removal("w", &w, &req, &result, Some(&changed))
+                    .is_err(),
+                "{field}"
+            );
+        }
+        assert!(
+            !s.subject_deletion_removal("w", &w)
+                .unwrap()
+                .unwrap()
+                .completed
+        );
+        s.confirm_subject_deletion_removal("w", &w, &req, &result, Some(&purge))
+            .unwrap();
+        drop(s);
+        let s = SqliteWorkspaceStore::open_for_test(&path).unwrap();
+        s.confirm_subject_deletion_removal("w", &w, &req, &result, Some(&purge))
+            .unwrap();
+        assert!(
+            s.subject_deletion_removal("w", &w)
+                .unwrap()
+                .unwrap()
+                .completed
+        );
+        assert_eq!(saved_prior(&s, "subject", &w), None);
+        assert!(
+            s.current_worker_singleton_owner("w", "subjektiv-consolidation:subject")
+                .unwrap()
+                .is_none()
+        );
+        s.with_conn(|c| {
+            let old:(String,String,String) = c.query_row("SELECT state,session_disposition,metadata_disposition FROM worker_removal_operations WHERE operation_id=?1",[&req.operation_id],|r|Ok((r.get(0)?,r.get(1)?,r.get(2)?)))?;
+            assert_eq!(old,("succeeded".into(),"archive".into(),"purge".into()));
+            assert!(c.execute("UPDATE subject_deletion_removals SET completed=0",[]).is_err());
+            assert_eq!(c.query_row("SELECT count(*) FROM pragma_foreign_key_check",[],|r|r.get::<_,i64>(0))?,0);
+            Ok(())
+        }).unwrap();
+        assert_eq!(
+            s.subject_deletion_removal("w", &other).unwrap(),
+            other_before
+        );
+        assert!(
+            s.current_worker_singleton_owner("w", "subjektiv:another")
+                .unwrap()
+                .is_some()
+        );
+        s.erase_subject_deleted_jobs("w", "subject", "delete")
+            .unwrap();
+        s.finish_subject_deletion_record("w", "subject", "delete")
+            .unwrap();
+    }
+
+    #[test]
+    fn planned_pin_uses_current_epoch_and_never_replaces_prior_success() {
+        for (state, current, expected) in [
+            ("planned", true, true),
+            ("planned", false, false),
+            ("failed", false, true),
+        ] {
+            let s = setup();
+            let w = worker(91);
+            in_flight(&s, &w, state, false);
+            if !current {
+                s.with_conn(|c| {
+                    c.execute("UPDATE worker_registry SET updated_at='later'", [])?;
+                    Ok(())
+                })
+                .unwrap();
+            }
+            s.prepare_subject_deletion("w", "subject", "delete", &[])
+                .unwrap();
+            assert_eq!(
+                s.subject_deletion_removal("w", &w).unwrap().is_some(),
+                expected,
+                "{state}/{current}"
+            );
+        }
+        let s = setup();
+        let w = worker(92);
+        in_flight(&s, &w, "executing", true);
+        receipt(&s, &w, "succeeded", "archive", "retain");
+        s.prepare_subject_deletion("w", "subject", "delete", &[])
+            .unwrap();
+        assert!(saved_prior(&s, "subject", &w).is_some());
+        assert!(s.subject_deletion_removal("w", &w).unwrap().is_none());
+        s.erase_subject_deleted_jobs("w", "subject", "delete")
+            .unwrap();
+    }
+
     #[test]
     fn saved_prior_receipt_must_remain_the_exact_success_not_a_replacement() {
         let s = setup();

crates/workspace-server/src/workspace_deletion.rs

@@ -96,6 +96,7 @@ const WORKSPACE_DELETION_PURGE_TABLES: &[&str] = &[
     "worker_control_grants",
     // Snapshot children precede their fence parent and canonical Worker removal
     // receipts (subject_deletion_workers.prior_removal_operation_id).
+    "subject_deletion_removals",
     "subject_deletion_allocations",
     "subject_deletion_jobs",
     "subject_deletion_workers",
@@ -1286,6 +1287,7 @@ mod tests {
     fn workspace_purge_removes_subject_deletion_snapshots_and_preserves_other_workspace() {
         let (store, owner, workspace_id) = setup();
         let tables = [
+            "subject_deletion_removals",
             "subject_deletion_allocations",
             "subject_deletion_jobs",
             "subject_deletion_workers",
@@ -1299,9 +1301,11 @@ mod tests {
                 .position(|name| *name == table)
                 .unwrap()
         };
-        for child in &tables[..3] {
+        for child in &tables[..4] {
             assert!(position(*child) < position("subject_deletions"));
         }
+        assert!(position("subject_deletion_removals") < position("subject_deletion_workers"));
+        assert!(position("subject_deletion_removals") < position("worker_removal_operations"));
         assert!(position("subject_deletion_workers") < position("worker_removal_operations"));
         for workspace in ["workspace-a", "workspace-b"] {
             store.with_conn(|conn| {
@@ -1310,11 +1314,16 @@ mod tests {
                     worker_updated_at,policy_id,policy_digest,session_disposition,metadata_disposition,
                     archive_retention_kind,diagnostics_disposition,blockers_json,state,reason,created_at,updated_at)
                     VALUES(?1,?1,'fingerprint',?1,'runtime','worker','1','policy','digest',
-                        'purge','purge','forever','purge','[]','succeeded','removed','1','1')", [workspace])?;
+                        'purge','purge','forever','purge','[]','executing','removed','1','1')", [workspace])?;
                 conn.execute("INSERT INTO subject_deletions(workspace_id,subject_id,operation_id,completed,jobs_erased)
                     VALUES(?1,'subject','subject-delete',1,1)", [workspace])?;
                 conn.execute("INSERT INTO subject_deletion_workers(workspace_id,subject_id,runtime_id,worker_id,is_job,prior_removal_operation_id,originals_absent)
-                    VALUES(?1,'subject','runtime','worker',1,?1,1)", [workspace])?;
+                    VALUES(?1,'subject','runtime','worker',1,NULL,0)", [workspace])?;
+                conn.execute("INSERT INTO subject_deletion_removals(workspace_id,subject_id,runtime_id,worker_id,operation_id,input_fingerprint,request_json)
+                    VALUES(?1,'subject','runtime','worker',?1,'fingerprint',json_object('operation_id',?1,'input_fingerprint','fingerprint'))", [workspace])?;
+                // Settle canonical removal so Workspace deletion's independent
+                // preflight is unblocked; the reconciliation row still needs purge.
+                conn.execute("UPDATE worker_removal_operations SET state='succeeded' WHERE operation_id=?1", [workspace])?;
                 conn.execute("INSERT INTO subject_deletion_jobs VALUES(?1,'subject','job')", [workspace])?;
                 conn.execute("INSERT INTO subject_deletion_allocations VALUES(?1,'subject','allocation')", [workspace])?;
                 Ok(())

web/workspace/src/lib/workspace/memory/SubjectDelete.svelte

@@ -91,7 +91,7 @@
     {#if phase === 'confirming'}
       <form onsubmit={(event) => { event.preventDefault(); void submit(); }}>
         <p><strong>This is irreversible.</strong> This Subject, its behavior, Memory, change history, staging, Surface, Journal, permissions, delegations, Writer/Session associations, and Subject Job inputs/results will be deleted. Connected and delegated Workers and Jobs must stop before completion can be confirmed.</p>
-        <p>Original Worker/Session text, existing archives and Workdirs, Tickets, and other separately retained resources are <strong>not deleted</strong>. Minimal deletion retry/audit metadata is retained. Other Subjects and Workspaces are not affected.</p>
+        <p>Original Worker/Session text, existing archives and Workdirs, Tickets, and other separately retained resources are <strong>not deleted</strong>. Minimal deletion retry/audit metadata is retained. Any earlier Worker retention request still follows its original disposition. Other Subjects and Workspaces are not affected.</p>
         <label class="delete-confirm"><input type="checkbox" bind:checked={confirmed} /> I have checked the target, deletion scope, retained data, and irreversible effect.</label>
         <div class="delete-actions">
           <button type="submit" class="workspace-danger-button" disabled={!confirmed}>Permanently delete Subject</button>