fix: merge hare/develop and reconcile lifecycle contracts

crates/merge-request/src/lib.rs

@@ -210,12 +210,7 @@ impl MergeRequestThreadEvent {
     }
     fn bound_bodies(&mut self) {
         match self {
-            Self::Review(value) => {
-                truncate_body(&mut value.body);
-                for finding in &mut value.findings {
-                    truncate_body(&mut finding.body);
-                }
-            }
+            Self::Review(_) => {}
             Self::ReviewRevoked(value) => truncate_body(&mut value.reason),
             Self::ReviewCancelled(value) => truncate_body(&mut value.reason),
             Self::Comment(value) => truncate_body(&mut value.body),
@@ -939,10 +934,6 @@ impl MergeRequestStore {
         &self,
         i: SubmitMergeRequestReview,
     ) -> Result<ReviewEvent, MergeRequestError> {
-        bounded_body("review body", &i.body)?;
-        for finding in &i.findings {
-            bounded_body("review finding", &finding.body)?;
-        }
         let mut c = self.lock()?;
         let t = c.transaction()?;
         let g: Option<(String, String, String, String, String, String)> = t

crates/merge-request/tests/store.rs

@@ -376,6 +376,49 @@ fn approve_for(
 fn approve(s: &MergeRequestStore, subject: &str, token: &str) -> ReviewEvent {
     approve_for(s, "MR", "R", subject, token)
 }
+#[test]
+fn review_submission_preserves_long_body_and_findings_on_reload() {
+    for (body, finding_body) in [
+        ("b".repeat(16 * 1024 + 1), "finding".into()),
+        ("review".into(), "指摘".repeat(4096)),
+    ] {
+        let (_dir, store) = fixture();
+        open(&store);
+        request(&store, "source", "token");
+        let findings = vec![ReviewFinding {
+            severity: FindingSeverity::Major,
+            code: None,
+            path: None,
+            line: None,
+            body: finding_body,
+        }];
+        let submitted = store
+            .submit_review(SubmitMergeRequestReview {
+                merge_request_id: "MR".into(),
+                ticket_id: "T".into(),
+                current_subject_ref: "source".into(),
+                capability_token: "token".into(),
+                decision: ReviewDecision::RequestChanges,
+                body: body.clone(),
+                findings: findings.clone(),
+                now: at(4),
+            })
+            .unwrap();
+        assert_eq!(submitted.body, body);
+        assert_eq!(submitted.findings, findings);
+        let reloaded = store.get_by_id("W", "MR").unwrap();
+        let saved = reloaded
+            .thread
+            .iter()
+            .find_map(|event| match event {
+                MergeRequestThreadEvent::Review(review) => Some(review),
+                _ => None,
+            })
+            .unwrap();
+        assert_eq!(saved, &submitted);
+    }
+}
+
 #[test]
 fn review_submission_authorization_rejects_invalid_grants_before_side_effects() {
     let (_d, store) = fixture();

crates/server-api/src/lib.rs

@@ -7348,6 +7348,8 @@ pub struct WorkingDirectoryCleanupTarget {
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
 #[serde(deny_unknown_fields)]
 pub struct WorkingDirectoryRemovalRequest {
+    /// Stable for an exact resend; a new explicit request uses a new identity.
+    pub operation_id: String,
     pub reason: String,
 }
 
@@ -7886,6 +7888,9 @@ pub struct CurrentWorkerWorkdirAttachmentResponse {
     #[serde(skip_serializing_if = "Option::is_none")]
     pub environment: Option<workdir::WorkdirSessionEnvironment>,
     pub attached: bool,
+    /// The Workdir was already attached to this Worker; alias is its existing name.
+    #[serde(default)]
+    pub already_attached: bool,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)]
@@ -13691,9 +13696,11 @@ mod tests {
             capabilities: workdir::WorkdirSessionCapabilities::EMPTY,
             environment: None,
             attached: false,
+            already_attached: false,
         };
         let value = serde_json::to_value(&response).unwrap();
-        assert_eq!(value.as_object().unwrap().len(), 5);
+        assert_eq!(value.as_object().unwrap().len(), 6);
+        assert_eq!(value["already_attached"], false);
         assert!(value.get("connection_id").is_none());
         assert_eq!(
             serde_json::from_value::<CurrentWorkerWorkdirAttachmentResponse>(value).unwrap(),

crates/subjektiv/src/journal/binding.rs

@@ -202,6 +202,29 @@ impl SubjektivStore {
         })
     }
 
+    /// End a body after the Host has confirmed its Runtime removal. This is not a
+    /// completed Journal handoff: source coverage, positions and saved receipts stay
+    /// unchanged. Revocation and closing generations are atomic, so no retained Job
+    /// can write after deletion even while Backend Job cleanup is being retried.
+    pub fn release_removed_journal_body(&self, host: &JournalHostContext) -> JournalResult<()> {
+        self.database.try_transaction(|tx| {
+            super::store::check_session_host(self, tx, host)?;
+            let attr = host.attribution.as_ref().expect("checked Session authority");
+            let binding: Option<String> = tx.query_row(
+                "SELECT binding_id FROM journal_writer_bindings WHERE subject_id=?1 AND runtime_id=?2 AND worker_id=?3 ORDER BY rowid DESC LIMIT 1",
+                params![host.subject_id, attr.runtime_id, attr.worker_id], |r| r.get(0),
+            ).optional()?;
+            let binding = binding.unwrap_or_else(|| crate::issued_id("journal-removed-body"));
+            tx.execute("INSERT OR IGNORE INTO journal_writer_bindings(binding_id,subject_id,runtime_id,worker_id,active) VALUES(?1,?2,?3,?4,0)",params![binding,host.subject_id,attr.runtime_id,attr.worker_id])?;
+            tx.execute("UPDATE journal_writer_bindings SET active=0 WHERE subject_id=?1 AND runtime_id=?2 AND worker_id=?3",params![host.subject_id,attr.runtime_id,attr.worker_id])?;
+            // Bind legacy attributed Sessions to an inactive lifetime, never to the
+            // successor. Retained unfinished history is not an active writer.
+            tx.execute("INSERT OR IGNORE INTO journal_session_bindings(session_id,binding_id) SELECT session_id,?4 FROM subject_session_attributions WHERE subject_id=?1 AND runtime_id=?2 AND worker_id=?3",params![host.subject_id,attr.runtime_id,attr.worker_id,binding])?;
+            tx.execute("UPDATE journal_generations SET active=0 WHERE journal_id IN (SELECT j.id FROM journals j JOIN subject_session_attributions a ON a.session_id=j.source_session_id WHERE a.subject_id=?1 AND a.runtime_id=?2 AND a.worker_id=?3)",params![host.subject_id,attr.runtime_id,attr.worker_id])?;
+            Ok(())
+        })
+    }
+
     /// The Host must hold its lifecycle fence, confirm Idle, reconcile all committed original
     /// Sessions (including the save/notification gap), and exclude queued/running/Unknown Jobs.
     /// This transaction rejects retained writers, pending source bytes and active generations.

crates/subjektiv/src/journal/store.rs

@@ -653,6 +653,32 @@ impl SubjektivStore {
             Ok(())
         })
     }
+    /// Read the exact retained Job/attempt snapshot without granting writer
+    /// authority or creating a generation. Used for outcome recovery after removal.
+    pub fn retained_journal_generation(
+        &self,
+        host: &JournalHostContext,
+        id: &str,
+        binding: &JobAttemptBinding,
+    ) -> JournalResult<Option<JournalGeneration>> {
+        self.database.try_with_connection(|tx| {
+            check_host(self, tx, host, false)?;
+            check_session_host(self, tx, host)?;
+            journal(tx, host, id, true)?;
+            let raw: Option<String> = tx.query_row(
+                "SELECT snapshot_json FROM journal_generations WHERE journal_id=?1 AND job_id=?2 AND attempt_id=?3",
+                params![id, binding.job_id, binding.attempt_id], |r| r.get(0),
+            ).optional()?;
+            raw.map(|raw| {
+                let generation: JournalGeneration = serde_json::from_str(&raw)?;
+                if generation.journal_id != id || &generation.job != binding {
+                    return Err(conflict("retained generation identity mismatch"));
+                }
+                Ok(generation)
+            }).transpose()
+        })
+    }
+
     pub fn read_journal_generation(
         &self,
         host: &JournalHostContext,

crates/worker-runtime/src/runtime.rs

@@ -103,7 +103,18 @@ impl Drop for WorkdirOperationGuard {
     }
 }
 
-fn workdir_operation_busy() -> RuntimeError {
+fn workdir_operation_busy(id: &str, action: &str, blocking: &WorkdirOperation) -> RuntimeError {
+    let (busy_reason, active_use_count) = match blocking {
+        WorkdirOperation::Use(count) => ("active_use", *count),
+        WorkdirOperation::Cleanup => ("cleanup_in_progress", 0),
+    };
+    tracing::warn!(target: "yoi::workdir_cleanup",
+        working_directory_id = id,
+        action,
+        busy_source = "runtime_operation",
+        busy_reason,
+        active_use_count,
+        "Workdir operation blocked by busy resource");
     RuntimeError::WorkingDirectory(
         crate::working_directory::WorkingDirectoryDiagnostic::rejected(
             "working_directory_cleanup_resource_busy",
@@ -1150,8 +1161,12 @@ impl Runtime {
         let backend = {
             let mut state = self.lock()?;
             state.ensure_running()?;
-            if state.workdir_operations.contains_key(working_directory_id) {
-                return Err(workdir_operation_busy());
+            if let Some(blocking) = state.workdir_operations.get(working_directory_id) {
+                return Err(workdir_operation_busy(
+                    working_directory_id,
+                    "cleanup_working_directory",
+                    blocking,
+                ));
             }
             if let Some(worker_id) = state.worker_id_for_workdir(working_directory_id) {
                 return Err(RuntimeError::InvalidRequest(format!(
@@ -6344,11 +6359,8 @@ impl RuntimeState {
     }
 
     fn ensure_workdir_not_removing(&self, id: &str) -> Result<(), RuntimeError> {
-        if matches!(
-            self.workdir_operations.get(id),
-            Some(WorkdirOperation::Cleanup)
-        ) {
-            return Err(workdir_operation_busy());
+        if let Some(blocking @ WorkdirOperation::Cleanup) = self.workdir_operations.get(id) {
+            return Err(workdir_operation_busy(id, "admit_use", blocking));
         }
         Ok(())
     }
@@ -9409,6 +9421,84 @@ mod tests {
             .block_on(subscription.recv())
     }
 
+    #[test]
+    fn busy_admission_logs_runtime_use_and_cleanup_blockers() {
+        use std::io::{Read, Seek, SeekFrom};
+
+        let mut output = tempfile::tempfile().unwrap();
+        let writer = output.try_clone().unwrap();
+        let subscriber = tracing_subscriber::fmt()
+            .with_ansi(false)
+            .without_time()
+            .with_writer(move || writer.try_clone().unwrap())
+            .finish();
+        tracing::subscriber::with_default(subscriber, || {
+            tracing::callsite::rebuild_interest_cache();
+            let (runtime, backend) = runtime_and_backend();
+            let first = runtime.reserve_workdir_use(["workdir-used"]).unwrap();
+            let second = runtime.reserve_workdir_use(["workdir-used"]).unwrap();
+            assert_workdir_busy(
+                runtime
+                    .cleanup_working_directory("workdir-used")
+                    .unwrap_err(),
+            );
+            drop((first, second));
+            assert!(runtime.inner.lock().unwrap().workdir_operations.is_empty());
+
+            let (entered, release) = WorkdirEffectGate::install(&backend.cleanup_gate);
+            let cleanup_runtime = runtime.clone();
+            let cleanup = std::thread::spawn(move || {
+                cleanup_runtime.cleanup_working_directory("workdir-removing")
+            });
+            entered
+                .recv_timeout(std::time::Duration::from_secs(5))
+                .unwrap();
+            assert_workdir_busy(
+                runtime
+                    .cleanup_working_directory("workdir-removing")
+                    .unwrap_err(),
+            );
+            assert_workdir_busy(
+                runtime
+                    .reserve_workdir_use(["workdir-removing"])
+                    .unwrap_err(),
+            );
+            release.send(()).unwrap();
+            assert_workdir_busy(cleanup.join().unwrap().unwrap_err());
+            assert!(runtime.inner.lock().unwrap().workdir_operations.is_empty());
+        });
+        output.seek(SeekFrom::Start(0)).unwrap();
+        let mut log = String::new();
+        output.read_to_string(&mut log).unwrap();
+        let events: Vec<_> = log
+            .lines()
+            .filter(|line| line.contains("busy_source="))
+            .collect();
+        assert_eq!(events.len(), 3, "{log}");
+        for (event, (id, action, reason, count)) in events.iter().zip([
+            ("workdir-used", "cleanup_working_directory", "active_use", 2),
+            (
+                "workdir-removing",
+                "cleanup_working_directory",
+                "cleanup_in_progress",
+                0,
+            ),
+            ("workdir-removing", "admit_use", "cleanup_in_progress", 0),
+        ]) {
+            for field in [
+                "WARN".to_string(),
+                "yoi::workdir_cleanup".to_string(),
+                format!("working_directory_id=\"{id}\""),
+                format!("action=\"{action}\""),
+                "busy_source=\"runtime_operation\"".to_string(),
+                format!("busy_reason=\"{reason}\""),
+                format!("active_use_count={count}"),
+            ] {
+                assert!(event.contains(&field), "missing {field}: {event}");
+            }
+        }
+    }
+
     #[test]
     fn workdir_removal_reserves_only_its_target_and_releases_on_failure() {
         let (runtime, backend) = runtime_and_backend();

crates/worker-runtime/src/working_directory/occupancy.rs

@@ -28,7 +28,19 @@ pub(super) struct Lease {
     id: String,
 }
 
-fn busy() -> WorkingDirectoryDiagnostic {
+fn busy(id: &str, action: &str, blocking: &Entry) -> WorkingDirectoryDiagnostic {
+    let (busy_reason, active_use_count) = match blocking {
+        Entry::Use(count) => ("active_use", *count),
+        Entry::Create => ("creation_in_progress", 0),
+        Entry::Cleanup => ("cleanup_in_progress", 0),
+    };
+    tracing::warn!(target: "yoi::workdir_cleanup",
+        working_directory_id = id,
+        action,
+        busy_source = "materializer_occupancy",
+        busy_reason,
+        active_use_count,
+        "Workdir operation blocked by busy resource");
     WorkingDirectoryDiagnostic::new(
         "working_directory_cleanup_resource_busy",
         "A filesystem resource is busy or changed during removal. Release the resource and retry.",
@@ -42,7 +54,9 @@ impl Occupancy {
             .lock()
             .unwrap_or_else(|error| error.into_inner());
         match entries.get_mut(id) {
-            Some(Entry::Cleanup | Entry::Create) => return Err(busy()),
+            Some(blocking @ (Entry::Cleanup | Entry::Create)) => {
+                return Err(busy(id, "acquire_use", blocking));
+            }
             Some(Entry::Use(count)) => *count += 1,
             None => {
                 entries.insert(id.to_string(), Entry::Use(1));
@@ -101,8 +115,8 @@ impl Occupancy {
             .entries
             .lock()
             .unwrap_or_else(|error| error.into_inner());
-        if entries.contains_key(id) {
-            return Err(busy());
+        if let Some(blocking) = entries.get(id) {
+            return Err(busy(id, "acquire_creation", blocking));
         }
         entries.insert(id.to_string(), Entry::Create);
         Ok(Lease {
@@ -116,8 +130,8 @@ impl Occupancy {
             .entries
             .lock()
             .unwrap_or_else(|error| error.into_inner());
-        if entries.contains_key(id) {
-            return Err(busy());
+        if let Some(blocking) = entries.get(id) {
+            return Err(busy(id, "acquire_cleanup", blocking));
         }
         entries.insert(id.to_string(), Entry::Cleanup);
         Ok(Lease {
@@ -131,6 +145,71 @@ impl Occupancy {
 mod tests {
     use super::*;
 
+    #[test]
+    fn busy_admission_logs_blocker_and_use_count_without_changing_error() {
+        use std::io::{Read, Seek, SeekFrom};
+
+        let mut output = tempfile::tempfile().unwrap();
+        let writer = output.try_clone().unwrap();
+        let subscriber = tracing_subscriber::fmt()
+            .with_ansi(false)
+            .without_time()
+            .with_writer(move || writer.try_clone().unwrap())
+            .finish();
+        tracing::subscriber::with_default(subscriber, || {
+            tracing::callsite::rebuild_interest_cache();
+            let occupancy = Occupancy::default();
+            for (id, leases) in [
+                (
+                    "workdir-used",
+                    vec![
+                        occupancy.acquire_use("workdir-used").unwrap(),
+                        occupancy.acquire_use("workdir-used").unwrap(),
+                    ],
+                ),
+                (
+                    "workdir-creating",
+                    vec![occupancy.acquire_creation("workdir-creating").unwrap()],
+                ),
+                (
+                    "workdir-removing",
+                    vec![occupancy.acquire_cleanup("workdir-removing").unwrap()],
+                ),
+            ] {
+                let error = occupancy.acquire_cleanup(id).unwrap_err();
+                assert_eq!(error.code, "working_directory_cleanup_resource_busy");
+                assert_eq!(
+                    error.message,
+                    "A filesystem resource is busy or changed during removal. Release the resource and retry."
+                );
+                drop(leases);
+                drop(occupancy.acquire_cleanup(id).unwrap());
+            }
+        });
+        output.seek(SeekFrom::Start(0)).unwrap();
+        let mut log = String::new();
+        output.read_to_string(&mut log).unwrap();
+        let events: Vec<_> = log.lines().collect();
+        assert_eq!(events.len(), 3, "{log}");
+        for (event, (id, reason, count)) in events.iter().zip([
+            ("workdir-used", "active_use", 2),
+            ("workdir-creating", "creation_in_progress", 0),
+            ("workdir-removing", "cleanup_in_progress", 0),
+        ]) {
+            for field in [
+                "WARN".to_string(),
+                "yoi::workdir_cleanup".to_string(),
+                format!("working_directory_id=\"{id}\""),
+                "action=\"acquire_cleanup\"".to_string(),
+                "busy_source=\"materializer_occupancy\"".to_string(),
+                format!("busy_reason=\"{reason}\""),
+                format!("active_use_count={count}"),
+            ] {
+                assert!(event.contains(&field), "missing {field}: {event}");
+            }
+        }
+    }
+
     #[test]
     fn use_and_cleanup_leases_release_during_unwind() {
         let occupancy = Occupancy::default();

crates/worker/src/feature/builtin/manage_workdir.rs

@@ -776,25 +776,39 @@ impl WorkspaceHttpWorkdirBackend {
             ATTACH_TOOL,
             "working_directory_id",
         )?;
-        let alias_key = workdir::WorkdirAttachmentAlias::new(alias)
-            .map_err(|error| ToolError::InvalidArgument(error.to_string()))?;
-        if self.session_router.session(&alias_key).is_some() {
-            return Err(ToolError::InvalidArgument(format!(
-                "attachment alias `{alias}` already exists"
-            )));
-        }
         let response = self.attach_response(alias, working_directory_id)?;
-        if let Err(error) = self.session_router.attach(
-            alias_key.clone(),
+        if !response.attachment.attached
+            || response.attachment.working_directory_id != working_directory_id
+        {
+            return Err(ToolError::ExecutionFailed(
+                "Workdir attachment response does not match the target".into(),
+            ));
+        }
+        // Backend ownership decides the name. A requested alias cannot rename
+        // an existing attachment or create a second route to the same Workdir.
+        let attached_alias = &response.attachment.alias;
+        let alias_key = workdir::WorkdirAttachmentAlias::new(attached_alias.clone())
+            .map_err(|error| ToolError::ExecutionFailed(error.to_string()))?;
+        if let Some(session) = self.session_router.session(&alias_key) {
+            if session.workdir().id().as_str() != working_directory_id {
+                return Err(ToolError::ExecutionFailed("Local Workdir attachment differs from Backend; refresh attachments before retrying".into()));
+            }
+        } else if let Err(error) = self.session_router.attach(
+            alias_key,
             WorkspaceAttachedWorkdirSession::handle_for_workdir_with_environment(
                 self.client.clone(),
-                alias,
+                attached_alias,
                 working_directory_id,
                 response.attachment.capabilities,
                 response.environment.clone(),
             ),
         ) {
-            let rollback = self.detach(alias);
+            if response.attachment.already_attached {
+                return Err(ToolError::ExecutionFailed(format!(
+                    "attach Workdir session route: {error}; existing Backend attachment preserved"
+                )));
+            }
+            let rollback = self.detach(attached_alias);
             return Err(ToolError::ExecutionFailed(match rollback {
                 Ok(_) => {
                     format!("attach Workdir session route: {error}; Backend attachment rolled back")
@@ -805,7 +819,11 @@ impl WorkspaceHttpWorkdirBackend {
             }));
         }
         workdir_output(
-            format!("Attached Workdir {working_directory_id} as {alias}"),
+            if response.attachment.already_attached {
+                format!("Workdir {working_directory_id} is already attached as {attached_alias}")
+            } else {
+                format!("Attached Workdir {working_directory_id} as {attached_alias}")
+            },
             &response,
         )
     }
@@ -817,15 +835,31 @@ impl WorkspaceHttpWorkdirBackend {
     ) -> Result<WorkdirSessionAttachmentResponse, ToolError> {
         let workspace_id = encode_path_segment(self.workspace_id()?);
         self.authorize_workdir("workdir.attach", working_directory_id)?;
-        self.execute_json::<WorkdirSessionAttachmentResponse>(WorkspaceRequest::json(
-            WorkspaceRequestMethod::Post,
-            format!("/api/w/{workspace_id}/workers/self/workdir-attachments"),
-            serde_json::to_string(&WorkdirAttachRequest {
-                alias: alias.to_string(),
-                working_directory_id: working_directory_id.to_string(),
-            })
-            .map_err(decode_error)?,
-        ))
+        let response = self
+            .client
+            .execute(WorkspaceRequest::json(
+                WorkspaceRequestMethod::Post,
+                format!("/api/w/{workspace_id}/workers/self/workdir-attachments"),
+                serde_json::to_string(&WorkdirAttachRequest {
+                    alias: alias.to_string(),
+                    working_directory_id: working_directory_id.to_string(),
+                })
+                .map_err(decode_error)?,
+            ))
+            .map_err(|error| {
+                ToolError::ExecutionFailed(format!("Workspace Workdir request failed: {error}"))
+            })?;
+        if response.status == 409
+            && let Ok(error) =
+                serde_json::from_str::<server_api::RepositoryApiError>(&response.body)
+            && error.error == "workdir_occupied"
+        {
+            return Err(ToolError::StructuredConflict {
+                code: "workdir_occupied".into(),
+                message: "Workdir is occupied by another Worker".into(),
+            });
+        }
+        decode_response(response)
     }
 
     fn detach(&self, alias: &str) -> Result<ToolOutput, ToolError> {
@@ -860,7 +894,11 @@ impl WorkspaceHttpWorkdirBackend {
         )
     }
 
-    fn delete(&self, input: WorkdirDeleteInput) -> Result<ToolOutput, ToolError> {
+    fn delete(
+        &self,
+        input: WorkdirDeleteInput,
+        operation_id: String,
+    ) -> Result<ToolOutput, ToolError> {
         let workdir_id = validate_identity(
             &input.working_directory_id,
             DELETE_TOOL,
@@ -874,6 +912,7 @@ impl WorkspaceHttpWorkdirBackend {
             WorkspaceRequestMethod::Delete,
             format!("/api/w/{workspace_id}/working-directories/{workdir_path}"),
             serde_json::to_string(&WorkdirRemovalRequest {
+                operation_id,
                 reason: reason.to_string(),
             })
             .map_err(decode_error)?,
@@ -884,7 +923,7 @@ impl WorkspaceHttpWorkdirBackend {
                 response.disposition,
                 removal_guidance(response.failure_category.as_deref()),
                 if response.retryable {
-                    " After resolving the cause, repeat this WorkdirDelete with the same id and reason; current safety conditions will be rechecked."
+                    " After resolving the cause, a new explicit WorkdirDelete request will recheck current safety conditions. This failed request will not be automatically re-executed."
                 } else {
                     ""
                 }
@@ -1047,9 +1086,10 @@ impl Tool for WorkspaceHttpWorkdirTool {
                 self.backend.detach_managed(&input.alias, None).await
             }
 
-            WorkdirOperation::Delete => self
-                .backend
-                .delete(parse_input::<WorkdirDeleteInput>(input_json)?),
+            WorkdirOperation::Delete => self.backend.delete(
+                parse_input::<WorkdirDeleteInput>(input_json)?,
+                ctx.call_id.to_string(),
+            ),
         }
     }
 }
@@ -1111,8 +1151,14 @@ fn removal_guidance(category: Option<&str>) -> &'static str {
         Some("runtime_unavailable" | "provider_unavailable") => {
             "Restore the Runtime connection; the registry remains until deletion is confirmed."
         }
-        Some("provider_observation_unknown" | "provider_cleanup_outcome_unknown") => {
-            "Deletion is not confirmed. Re-observe with normal removal after restoring provider access; the registry is retained."
+        Some("provider_cleanup_outcome_unknown") => {
+            "Deletion outcome is unknown. Restore provider access and reconcile the existing request; do not submit a fresh deletion intent. Recovery only observes the target and never repeats deletion."
+        }
+        Some("provider_observation_unknown") => {
+            "The target could not be observed before deletion. Restore provider access before a new request."
+        }
+        Some("request_interrupted") => {
+            "The request was interrupted before execution. A new explicit request is required."
         }
         Some("unsupported_target") => {
             "Ask the owner for the provider's supported cleanup operation."
@@ -1312,6 +1358,8 @@ struct WorkdirAttachmentResponse {
     working_directory_id: String,
     capabilities: WorkdirSessionCapabilities,
     attached: bool,
+    #[serde(default)]
+    already_attached: bool,
 }
 
 #[derive(Debug, Serialize, Deserialize, PartialEq, Eq)]
@@ -1467,8 +1515,44 @@ mod tests {
         })
     }
 
+    #[tokio::test]
+    async fn delete_call_identity_distinguishes_new_requests_from_exact_resends() {
+        let client = Arc::new(RecordingWorkspaceClient::new(
+            (0..3)
+                .map(|_| {
+                    response(json!({
+                        "working_directory_id": "wd-request",
+                        "disposition": "attention_required",
+                        "retryable": true,
+                        "failure_category": "resource_busy"
+                    }))
+                })
+                .collect(),
+        ));
+        let tool = WorkspaceHttpWorkdirTool {
+            backend: WorkspaceHttpWorkdirBackend::new(client.clone()),
+            operation: WorkdirOperation::Delete,
+        };
+        let input = r#"{"working_directory_id":"wd-request","reason":"remove unused Workdir"}"#;
+        for id in ["delete-1", "delete-1", "delete-2"] {
+            tool.execute(input, ToolExecutionContext::new(id, "batch", 0))
+                .await
+                .unwrap();
+        }
+        let bodies: Vec<serde_json::Value> = client
+            .requests()
+            .iter()
+            .map(|r| serde_json::from_str(r.body.as_deref().unwrap()).unwrap())
+            .collect();
+        assert_eq!(bodies[0], bodies[1]);
+        assert_eq!(bodies[0]["operation_id"], "delete-1");
+        assert_eq!(bodies[2]["operation_id"], "delete-2");
+        assert_eq!(bodies[0]["reason"], bodies[2]["reason"]);
+        assert!(delete_schema()["properties"].get("operation_id").is_none());
+    }
+
     #[test]
-    fn ordinary_workdir_delete_reports_safe_cause_and_exact_retry_guidance() {
+    fn ordinary_workdir_delete_reports_safe_cause_and_new_request_guidance() {
         let client = Arc::new(RecordingWorkspaceClient::new(vec![response(json!({
             "working_directory_id": "wd-retry",
             "disposition": "attention_required",
@@ -1477,10 +1561,13 @@ mod tests {
         }))]));
         let backend = WorkspaceHttpWorkdirBackend::new(client.clone());
         let output = backend
-            .delete(WorkdirDeleteInput {
-                working_directory_id: "wd-retry".into(),
-                reason: "remove unused Workdir".into(),
-            })
+            .delete(
+                WorkdirDeleteInput {
+                    working_directory_id: "wd-retry".into(),
+                    reason: "remove unused Workdir".into(),
+                },
+                "delete-call-1".to_string(),
+            )
             .unwrap();
         assert!(output.summary.contains("Confirm its owner"));
         assert!(
@@ -1488,7 +1575,11 @@ mod tests {
                 .summary
                 .contains("Do not recursively delete mounted data")
         );
-        assert!(output.summary.contains("same id and reason"));
+        assert!(
+            output
+                .summary
+                .contains("new explicit WorkdirDelete request")
+        );
         assert!(output.summary.len() <= 600);
         let content: serde_json::Value =
             serde_json::from_str(output.content.as_deref().unwrap()).unwrap();
@@ -1726,10 +1817,13 @@ mod tests {
             .unwrap();
         backend.detach("checkout").unwrap();
         backend
-            .delete(WorkdirDeleteInput {
-                working_directory_id: "wd-created".to_string(),
-                reason: "remove stale Workdir".to_string(),
-            })
+            .delete(
+                WorkdirDeleteInput {
+                    working_directory_id: "wd-created".to_string(),
+                    reason: "remove stale Workdir".to_string(),
+                },
+                "delete-call-1".to_string(),
+            )
             .unwrap();
 
         let requests = client.requests();
@@ -1767,7 +1861,102 @@ mod tests {
         assert_eq!(requests[4].method, WorkspaceRequestMethod::Delete);
         let body: serde_json::Value =
             serde_json::from_str(requests[4].body.as_deref().unwrap()).unwrap();
-        assert_eq!(body, json!({"reason": "remove stale Workdir"}));
+        assert_eq!(
+            body,
+            json!({"reason": "remove stale Workdir", "operation_id": "delete-call-1"})
+        );
+    }
+
+    #[test]
+    fn attach_reports_other_worker_occupancy_without_exposing_backend_details() {
+        let client = Arc::new(RecordingWorkspaceClient::new(vec![WorkspaceResponse {
+            status: 409,
+            body: serde_json::to_string(&server_api::RepositoryApiError::new(
+                409,
+                "workdir_occupied",
+                "private backend detail",
+                Vec::new(),
+            ))
+            .unwrap(),
+        }]));
+        let backend = WorkspaceHttpWorkdirBackend::new(client.clone());
+        let error = backend
+            .attach(WorkdirAttachInput {
+                alias: "work".into(),
+                working_directory_id: "occupied".into(),
+            })
+            .unwrap_err();
+        assert!(
+            matches!(error, ToolError::StructuredConflict { code, message } if code == "workdir_occupied" && message == "Workdir is occupied by another Worker")
+        );
+        assert_eq!(client.requests().len(), 1);
+    }
+
+    #[test]
+    fn attach_reuses_backend_alias_without_duplicate_routes_or_detach() {
+        let client = Arc::new(RecordingWorkspaceClient::new((0..3).map(|_| response(json!({
+            "workspace_id": "workspace/test", "alias": "existing", "working_directory_id": "wd-existing",
+            "capabilities": {"bits": 63}, "attached": true, "already_attached": true,
+            "environment": workdir::WorkdirSessionEnvironment::root(WorkdirSessionCapabilities::ALL)
+        }))).collect()));
+        let backend = WorkspaceHttpWorkdirBackend::new(client.clone());
+        let occupied_alias = workdir::WorkdirAttachmentAlias::new("other").unwrap();
+        backend
+            .session_router
+            .attach(
+                occupied_alias.clone(),
+                WorkspaceAttachedWorkdirSession::handle_for_workdir(
+                    client.clone(),
+                    "other",
+                    "wd-other",
+                ),
+            )
+            .unwrap();
+        for alias in ["new-name", "existing", "other"] {
+            let output = backend
+                .attach(WorkdirAttachInput {
+                    alias: alias.into(),
+                    working_directory_id: "wd-existing".into(),
+                })
+                .unwrap();
+            assert_eq!(
+                output.summary,
+                "Workdir wd-existing is already attached as existing"
+            );
+            let body: serde_json::Value =
+                serde_json::from_str(output.content.as_deref().unwrap()).unwrap();
+            assert_eq!(body["alias"], "existing");
+            assert_eq!(body["already_attached"], true);
+        }
+        assert!(
+            backend
+                .session_router
+                .session(&workdir::WorkdirAttachmentAlias::new("new-name").unwrap())
+                .is_none()
+        );
+        assert_eq!(
+            backend
+                .session_router
+                .session(&occupied_alias)
+                .unwrap()
+                .workdir()
+                .id()
+                .as_str(),
+            "wd-other"
+        );
+        assert!(
+            backend
+                .session_router
+                .session(&workdir::WorkdirAttachmentAlias::new("existing").unwrap())
+                .is_some()
+        );
+        assert_eq!(client.requests().len(), 3);
+        assert!(
+            client
+                .requests()
+                .iter()
+                .all(|request| request.method == WorkspaceRequestMethod::Post)
+        );
     }
 
     #[tokio::test]
@@ -2377,10 +2566,13 @@ mod tests {
         );
         assert!(
             backend
-                .delete(WorkdirDeleteInput {
-                    working_directory_id: "wd".into(),
-                    reason: "remove unused".into()
-                })
+                .delete(
+                    WorkdirDeleteInput {
+                        working_directory_id: "wd".into(),
+                        reason: "remove unused".into(),
+                    },
+                    "delete-intent".into(),
+                )
                 .is_err()
         );
         assert!(backend.detach_managed("checkout", None).await.is_err());

crates/worker/src/feature/builtin/manage_workdir/wip.rs

@@ -914,7 +914,7 @@ impl WipOperationHandler for ItemHandler {
                     let mut result = raw;
                     result["workdir_path"] = json!(item_path(Domain::Workdir, &self.id));
                     result["attachments_path"] = json!(ATTACHMENTS);
-                    result["attachment"] = input
+                    result["attachment"] = result
                         .get("alias")
                         .and_then(Json::as_str)
                         .and_then(|alias| self.provider.attached_reference(alias, &self.id))
@@ -1421,10 +1421,19 @@ pub(crate) mod tests {
                     json!({"workspace_id":"test-workspace", "runtime_id":"runtime", "item":raw, "diagnostics":[]})
                 }
                 (WorkspaceRequestMethod::Post, "workers/self/workdir-attachments") => {
-                    state.generation += 1;
-                    let raw = json!({"alias":body["alias"], "working_directory_id":body["working_directory_id"], "connection_id":format!("connection-{}", state.generation), "capabilities":{"bits":63}});
-                    state.attachments.push(raw);
-                    json!({"workspace_id":"test-workspace", "alias":body["alias"], "working_directory_id":body["working_directory_id"], "capabilities":{"bits":63}, "environment":workdir::WorkdirSessionEnvironment::root(workdir::WorkdirSessionCapabilities::ALL), "attached":true})
+                    let existing = state
+                        .attachments
+                        .iter()
+                        .find(|raw| raw["working_directory_id"] == body["working_directory_id"])
+                        .cloned();
+                    let already_attached = existing.is_some();
+                    let raw = existing.unwrap_or_else(|| {
+                        state.generation += 1;
+                        let raw = json!({"alias":body["alias"], "working_directory_id":body["working_directory_id"], "connection_id":format!("connection-{}", state.generation), "capabilities":{"bits":63}});
+                        state.attachments.push(raw.clone());
+                        raw
+                    });
+                    json!({"workspace_id":"test-workspace", "alias":raw["alias"], "working_directory_id":raw["working_directory_id"], "capabilities":{"bits":63}, "environment":workdir::WorkdirSessionEnvironment::root(workdir::WorkdirSessionCapabilities::ALL), "attached":true, "already_attached":already_attached})
                 }
                 (WorkspaceRequestMethod::Delete, p)
                     if p.starts_with("workers/self/workdir-attachments/") =>

crates/worker/src/feature/builtin/manage_workdir/wip/model_tests.rs

@@ -413,7 +413,16 @@ async fn model_repository_create_attach_selected_entries_read_and_detach_preserv
         .invoke(&workdir, "attach", json!({"alias":"checkout"}))
         .await;
     assert_eq!(attached["attached"], true);
+    assert_eq!(attached["already_attached"], false);
     assert_eq!(attached["attachment"]["$case"], "available");
+    let (reused, reused_entries) = f
+        .invoke(&workdir, "attach", json!({"alias":"another-name"}))
+        .await;
+    assert_eq!(reused["already_attached"], true);
+    assert_eq!(reused["alias"], "checkout");
+    assert_eq!(reused["attachment"], attached["attachment"]);
+    assert_eq!(reused_entries, entries);
+    assert_eq!(f.client.catalog.state.lock().unwrap().attachments.len(), 1);
     assert_eq!(
         f.writes("workdir-session/operations"),
         0,
@@ -432,7 +441,7 @@ async fn model_repository_create_attach_selected_entries_read_and_detach_preserv
             .unwrap()
             .contains("isolated content")
     );
-    assert_eq!(f.writes("workdir-attachments"), 1);
+    assert_eq!(f.writes("workdir-attachments"), 2);
     f.invoke(&attachment, "detach", json!({})).await;
     assert!(
         f.execute("Inspect", json!({"path":attachment,"refresh":true}))

crates/worker/src/feature/builtin/manage_workdir/wip/result.rs

@@ -132,6 +132,7 @@ pub(super) fn attached() -> TypeExpr {
         field("capabilities", true, TypeExpr::Json),
         field("environment", true, TypeExpr::Json),
         field("attached", true, TypeExpr::Boolean),
+        field("already_attached", true, TypeExpr::Boolean),
         field("workdir_path", true, TypeExpr::Entry),
         field("attachments_path", true, TypeExpr::Entry),
         field("attachment", true, TypeExpr::Union { cases: vec![

crates/workspace-server/src/lib.rs

@@ -144,6 +144,12 @@ pub enum Error {
     TicketAssignmentConflict(String),
     #[error("Workdir attachment conflict: {0}")]
     WorkdirAttachmentConflict(String),
+    #[error("Workdir {workdir_id} is already attached to Worker {runtime_id}:{worker_id}")]
+    WorkdirOccupied {
+        workdir_id: String,
+        runtime_id: String,
+        worker_id: String,
+    },
     #[error("Workspace permission denied: {0}")]
     WorkspacePermissionDenied(String),
     #[error("Workspace config update conflict: {0}")]

crates/workspace-server/src/retention.rs

@@ -325,7 +325,7 @@ impl SqliteWorkspaceStore {
         let now = Utc::now().to_rfc3339();
         self.with_conn_mut(|conn| {
             let tx=conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
-            let policy=load_removal_policy(&tx,&req.workspace_id,&req.worker,erased_journal)?.ok_or_else(|| StoreError::InvalidInput(format!("policy-missing:{}",req.workspace_id)))?;
+            let policy=load_removal_policy(&tx,&req.workspace_id,&req.worker,erased_journal,true)?.ok_or_else(|| StoreError::InvalidInput(format!("policy-missing:{}",req.workspace_id)))?;
             let worker=match load_worker(&tx,&req.workspace_id,&req.worker)? {
                 Some(v)=>v,
                 None=>{
@@ -385,7 +385,7 @@ impl SqliteWorkspaceStore {
             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}")))?;
+                let policy=load_removal_policy(&tx,workspace_id,&plan.worker,plan.policy_id==ERASED_JOURNAL_POLICY_ID,plan.policy_id=="backend-subject-removal-archive-v1")?.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"));
@@ -1068,6 +1068,7 @@ fn load_removal_policy(
     w: &str,
     worker: &RuntimeWorkerRef,
     erased_journal: bool,
+    preserve_subject_history: bool,
 ) -> crate::Result<Option<WorkerRetentionPolicy>> {
     let Some(mut policy) = load_policy(c, w)? else {
         return Ok(None);
@@ -1135,6 +1136,25 @@ fn load_removal_policy(
         policy.archive_retention = update.archive_retention;
         policy.diagnostics_disposition = update.diagnostics_disposition;
         policy.diagnostics_retention_seconds = None;
+    } else if preserve_subject_history && policy.session_disposition == SessionDisposition::Purge && c.query_row(
+        "SELECT EXISTS(SELECT 1 FROM worker_create_reservations WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND substr(singleton_key,1,10)='subjektiv:')",
+        params![w, worker.runtime_id, worker.worker_id],
+        |r| r.get::<_, bool>(0),
+    )? {
+        // Removing a Subject body releases occupancy, not its original history.
+        // A configured Purge must not silently erase an unfinished Journal. Explicit
+        // Session erasure uses the separate, proof-bound branch above.
+        let update = WorkerRetentionPolicyUpdate {
+            policy_id: "backend-subject-removal-archive-v1".into(),
+            session_disposition: SessionDisposition::Archive,
+            metadata_disposition: policy.metadata_disposition,
+            archive_retention: policy.archive_retention,
+            diagnostics_disposition: policy.diagnostics_disposition,
+            diagnostics_retention_seconds: policy.diagnostics_retention_seconds,
+        };
+        policy.policy_id = update.policy_id.clone();
+        policy.policy_digest = worker_retention_policy_digest(&update);
+        policy.session_disposition = update.session_disposition;
     }
     Ok(Some(policy))
 }
@@ -1167,7 +1187,7 @@ fn require_current_subject_deletion_plan(
         return Ok(());
     }
     if crate::store::subject_deletion_worker_kind(c, &plan.workspace_id, &plan.worker)?.is_some() {
-        let current = load_removal_policy(c, &plan.workspace_id, &plan.worker, false)?
+        let current = load_removal_policy(c, &plan.workspace_id, &plan.worker, false, true)?
             .ok_or_else(|| StoreError::InvalidInput("Subject deletion policy missing".into()))?;
         if plan.session_disposition != current.session_disposition
             || plan.metadata_disposition != current.metadata_disposition
@@ -2934,7 +2954,7 @@ mod tests {
                 .is_err()
         );
         let fresh = store
-            .with_conn(|c| load_removal_policy(c, "w", &req().worker, false))
+            .with_conn(|c| load_removal_policy(c, "w", &req().worker, false, true))
             .unwrap()
             .unwrap();
         assert_eq!(fresh.session_disposition, SessionDisposition::Purge);

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

@@ -101,6 +101,20 @@ pub(super) fn prepare(api: &WorkspaceApi, reservation: &BackendJobReservation) -
     authorize_source(api, &attribution, "journal.job.input.read")?;
     authorize_source(api, &attribution, "journal.job.input.supply")?;
     let host = host(api, &store, grant)?;
+    let attr = store
+        .session_attribution(&grant.session_id)
+        .map_err(|e| Error::Store(e.to_string()))?
+        .ok_or_else(|| {
+            Error::WorkspacePermissionDenied("Journal Session attribution missing".into())
+        })?;
+    if store
+        .journal_worker_was_released(&attr.subject_id, &attr.runtime_id, &attr.worker_id)
+        .map_err(domain)?
+    {
+        return Err(Error::WorkspacePermissionDenied(
+            "released Journal body cannot dispatch another attempt".into(),
+        ));
+    }
     let input = saved_input(&reservation.job.request, grant)?;
     let g = store
         .prepare_journal_input(
@@ -640,6 +654,18 @@ fn dispatch(
         .map_err(|_| Error::Store("Journal lifecycle lock poisoned".into()))?;
     authorize_source(api, &attribution, "journal.recording.request")?;
     let store = store(api)?;
+    if store
+        .journal_worker_was_released(
+            &attribution.subject_id,
+            &attribution.runtime_id,
+            &attribution.worker_id,
+        )
+        .map_err(domain)?
+    {
+        // Deletion can retain an unfinished prefix. It must not mint attempts or
+        // transfer the removed body's write capability to its successor.
+        return Ok(());
+    }
     let host =
         JournalHostContext::validated_session(&store, attribution.clone()).map_err(domain)?;
     if let Some(g) = store
@@ -1823,8 +1849,8 @@ fn finish_erased_release(
         .release_subject_singleton(api.workspace_id(), &lease, binding_id, &operation)
 }
 
-/// Normal deletion cannot substitute for Subject release, even for a stopped or disabled body.
-/// Explicit erasure remains a separate monotonic source/diagnostics destruction path.
+/// Ordinary removal owns writer revocation and occupancy release. Before execution
+/// only preserve/erase source requirements are checked; an active writer is not a blocker.
 pub(super) fn prepare_worker_removal(
     service: &WorkerRemovalService,
     runtime: &RuntimeRegistry,
@@ -1852,7 +1878,6 @@ pub(super) fn prepare_worker_removal(
         &service.journal_registration,
     )
     .map_err(|e| Error::Store(e.to_string()))?;
-    let host = JournalHostContext::validated_session(&store, attr.clone()).map_err(domain)?;
     let source = runtime_api::SessionPublicSource::Retained {
         worker_id: worker_runtime::identity::WorkerId::parse(&target.worker_id)
             .ok_or_else(|| Error::Store("Worker identity invalid".into()))?,
@@ -1869,25 +1894,9 @@ pub(super) fn prepare_worker_removal(
         finish_erased_release(&api, &attr)?;
         return Ok(Some(proof));
     }
-    let Some(binding_id) = host.writer_binding_id() else {
-        return Err(Error::RepositoryConflict(
-            "Subject writer evidence is missing; ordinary deletion cannot bypass release".into(),
-        ));
-    };
-    let writer = JournalWriterBinding {
-        subject_id: attr.subject_id.clone(),
-        runtime_id: attr.runtime_id.clone(),
-        worker_id: attr.worker_id.clone(),
-        binding_id: binding_id.into(),
-    };
-    if store.journal_writer_active(&writer).map_err(domain)? {
-        return Err(Error::RepositoryConflict("Subject remains occupied: restore to actual Idle, finish Journal recording, then explicitly release before ordinary deletion".into()));
-    }
-    if store.journal_retention_required(&host).map_err(domain)? {
-        return Err(Error::RepositoryConflict(
-            "Released Subject has unresolved Journal material; retain occupancy and source".into(),
-        ));
-    }
+    // Do not require a live Controller, complete Journal, or prior release here.
+    // Runtime removal is the physical fence; finish_worker_removal revokes the
+    // exact old body only after its successful, matching retention receipt.
     Ok(None)
 }
 
@@ -1990,20 +1999,6 @@ pub(super) fn validate_removal_retention(
         // its retained permanent fence, including recovery after the aggregate is gone.
         return Ok(());
     }
-    let binding_id = host.writer_binding_id().ok_or_else(|| {
-        Error::RepositoryConflict("Subject writer evidence missing during removal recovery".into())
-    })?;
-    let binding = JournalWriterBinding {
-        subject_id: attr.subject_id.clone(),
-        runtime_id: attr.runtime_id.clone(),
-        worker_id: attr.worker_id.clone(),
-        binding_id: binding_id.into(),
-    };
-    if store.journal_writer_active(&binding).map_err(domain)? {
-        return Err(Error::RepositoryConflict(
-            "Prepared removal cannot bypass explicit Subject release".into(),
-        ));
-    }
     let source = runtime_api::SessionPublicSource::Retained {
         worker_id: worker_runtime::identity::WorkerId::parse(&target.worker_id)
             .ok_or_else(|| Error::Store("Worker identity invalid".into()))?,
@@ -2011,37 +2006,160 @@ pub(super) fn validate_removal_retention(
     let runtime = service.runtime.upgrade().ok_or_else(|| {
         Error::Store("Runtime unavailable for source retention confirmation".into())
     })?;
-    if store.journal_retention_required(&host).map_err(domain)? {
-        if prepared.runtime_request.session_disposition
-            != worker_runtime::retention::SessionDisposition::Archive
-            || prepared.runtime_request.archive_id.is_none()
-        {
-            return Err(Error::RepositoryConflict(
-                "unreflected Journal requires a retained original Session archive".into(),
-            ));
+    let retention_required = store.journal_retention_required(&host).map_err(domain)?;
+    if prepared.runtime_request.session_disposition
+        == worker_runtime::retention::SessionDisposition::Purge
+        && prepared.runtime_request.archive_id.is_none()
+        && !retention_required
+        && store
+            .journal_worker_was_released(&attr.subject_id, &attr.runtime_id, &attr.worker_id)
+            .map_err(domain)?
+    {
+        // Compatibility for an already issued, history-complete ordinary Purge
+        // plan. New Subject plans always Archive. Keep the old fingerprint and
+        // disposition; never turn this into an erasure or allow an active writer.
+        // A missing Runtime Worker is NOT success: execute_worker_retention still
+        // has to return the exact persisted receipt before anything can commit.
+        // A pending receipt may outlive the physical aggregate while Runtime's
+        // in-memory catalog still contains the Worker. Pin release is best effort
+        // in this historical branch; exact Runtime retention replay still enforces
+        // pins if the source exists, and requires a receipt if it is already gone.
+        if let Err(error) = retain_source(&runtime, &service.workspace_id, &attr, source, true) {
+            tracing::debug!(%error, "Historical Purge source unavailable; reconcile exact Runtime receipt");
         }
-        if let Err(original) = retain_source(&runtime, &service.workspace_id, &attr, source, false)
-        {
-            // Recovery may already have transferred the physical source before its Backend
-            // catalog commit; authenticate the exact reserved archive, never another Session.
-            let archive = runtime_api::SessionPublicSource::Archived {
-                archive_id: prepared
-                    .runtime_request
-                    .archive_id
-                    .clone()
-                    .expect("checked"),
-            };
-            retain_source(&runtime, &service.workspace_id, &attr, archive, false)
-                .map_err(|_| original)?;
-        }
-    } else {
-        // This stopped/removal-fenced Session has no admitted remainder; purge need not
-        // force a generation. A failed release conservatively blocks ordinary deletion.
-        retain_source(&runtime, &service.workspace_id, &attr, source, true)?;
+        return Ok(());
+    }
+    if prepared.runtime_request.session_disposition
+        != worker_runtime::retention::SessionDisposition::Archive
+        || prepared.runtime_request.archive_id.is_none()
+    {
+        return Err(Error::RepositoryConflict(
+            "Subject removal retains its original Session; explicit erasure is required for Purge"
+                .into(),
+        ));
+    }
+    let release = !retention_required;
+    if let Err(original) = retain_source(&runtime, &service.workspace_id, &attr, source, release) {
+        // A lost reply may follow the physical transfer. Reconcile only the exact
+        // reserved archive and Session, whether or not an admitted remainder exists.
+        let archive = runtime_api::SessionPublicSource::Archived {
+            archive_id: prepared
+                .runtime_request
+                .archive_id
+                .clone()
+                .expect("checked"),
+        };
+        retain_source(&runtime, &service.workspace_id, &attr, archive, release)
+            .map_err(|_| original)?;
     }
     Ok(())
 }
 
+/// A successful Runtime receipt, not catalog liveness, authorizes ending the old
+/// writer. The ordinary removal commit releases occupancy atomically with catalog
+/// deletion. Until then a crash leaves the singleton occupied and the same receipt
+/// can safely repeat this step; no Controller restore or Journal generation is needed.
+pub(super) fn finish_worker_removal<T>(
+    service: &WorkerRemovalService,
+    target: &RuntimeWorkerRef,
+    prepared: &crate::retention::PreparedWorkerRemoval,
+    result: &worker_runtime::retention::WorkerRetentionExecutionResult,
+    commit: impl FnOnce() -> T,
+) -> Result<T> {
+    if let Some(api) = service.journal_dispatch_api.upgrade()
+        && subjektiv_deletion::deleting_worker(&api, target)?
+    {
+        // Full Subject deletion already fenced the domain. Its canonical retention
+        // commit, not ordinary Journal writer validation, owns completion.
+        return Ok(commit());
+    }
+    let Some(attr) = removal_attribution(service, target)? else {
+        return Ok(commit());
+    };
+    let request = &prepared.runtime_request;
+    if prepared.plan.worker != *target
+        || request.workspace_id != service.workspace_id
+        || request.source_runtime_id != target.runtime_id
+        || request.worker_id.to_string() != target.worker_id
+        || !result.source_removed
+        || result.worker_id != request.worker_id
+        || result.operation_id != request.operation_id
+        || result.input_fingerprint != request.input_fingerprint
+        || result.session_disposition != request.session_disposition
+        || result.diagnostics_disposition != request.diagnostics_disposition
+        || result.diagnostics_retained
+            != (request.diagnostics_disposition
+                == worker_runtime::retention::DiagnosticsDisposition::Retain)
+    {
+        return Err(Error::RepositoryConflict(
+            "Subject removal receipt does not match its plan".into(),
+        ));
+    }
+    if request.session_disposition == worker_runtime::retention::SessionDisposition::Archive {
+        let archive = result.archive.as_ref().ok_or_else(|| {
+            Error::RepositoryConflict("Subject removal archive receipt is missing".into())
+        })?;
+        if Some(&archive.archive_id) != request.archive_id.as_ref()
+            || archive.workspace_id != request.workspace_id
+            || archive.source_runtime_id != request.source_runtime_id
+            || archive.source_worker_id != request.worker_id
+            || archive.source_session_id != attr.session_id
+            || archive.operation_id != request.operation_id
+            || archive.input_fingerprint != request.input_fingerprint
+            || archive.policy_id != request.policy_id
+        {
+            return Err(Error::RepositoryConflict(
+                "Subject removal archive identity mismatch".into(),
+            ));
+        }
+    }
+    let api = service.journal_dispatch_api.upgrade().ok_or_else(|| {
+        Error::Store("Workspace API unavailable for Subject removal completion".into())
+    })?;
+    let gate = lifecycle_lock(&api, &attr.subject_id)?;
+    let _guard = gate
+        .lock()
+        .map_err(|_| Error::Store("Journal lifecycle lock poisoned".into()))?;
+    let store = store(&api)?;
+    let host = JournalHostContext::validated_session(&store, attr.clone()).map_err(domain)?;
+    if store.journal_session_erased(&host).map_err(domain)? {
+        if request.session_disposition == worker_runtime::retention::SessionDisposition::Archive {
+            // Erasure may win after Runtime archive transfer. Do not publish that
+            // archive or skip writer revocation: the same removal retry must take
+            // erased_archive_removal_proof and its physical-erasure commit path.
+            return Err(Error::RepositoryConflict(
+                "Session erased during removal; reconcile the same removal operation".into(),
+            ));
+        }
+        return Ok(commit());
+    }
+    store.release_removed_journal_body(&host).map_err(domain)?;
+    // Result delivery takes the same Subject gate. Revocation is durable; release
+    // the gate while reconciling jobs, then recheck erasure before publishing.
+    drop(_guard);
+    for session in store
+        .subject_session_attributions(&attr.subject_id)
+        .map_err(|e| Error::Store(e.to_string()))?
+        .into_iter()
+        .filter(|a| a.runtime_id == target.runtime_id && a.worker_id == target.worker_id)
+    {
+        reconcile_session_jobs(&api, &session, false)?;
+    }
+    let _guard = gate
+        .lock()
+        .map_err(|_| Error::Store("Journal lifecycle lock poisoned".into()))?;
+    if store.journal_session_erased(&host).map_err(domain)?
+        && request.session_disposition == worker_runtime::retention::SessionDisposition::Archive
+    {
+        return Err(Error::RepositoryConflict(
+            "Session erased during removal; reconcile the same removal operation".into(),
+        ));
+    }
+    // Keep erasure excluded through the catalog commit: otherwise a tombstone
+    // could win between revocation and publication of the retained archive.
+    Ok(commit())
+}
+
 #[derive(Deserialize)]
 #[serde(deny_unknown_fields)]
 pub(super) struct Empty {}
@@ -2196,7 +2314,28 @@ pub(super) fn reconcile_unknown(
         .as_ref()
         .ok_or_else(|| Error::InvalidInput("Journal recovery requires a grant".into()))?;
     let store = store(api)?;
-    let host = host(api, &store, grant)?;
+    let attr = store
+        .session_attribution(&grant.session_id)
+        .map_err(|e| Error::Store(e.to_string()))?
+        .filter(|a| a.subject_id == grant.subject_id)
+        .ok_or_else(|| Error::WorkspacePermissionDenied("Journal attribution mismatch".into()))?;
+    let released = store
+        .journal_worker_was_released(&attr.subject_id, &attr.runtime_id, &attr.worker_id)
+        .map_err(domain)?;
+    let host = if released {
+        // Recovery reads immutable evidence, never renews a legacy writer grant.
+        let host = JournalHostContext::validated_session(&store, attr.clone()).map_err(domain)?;
+        if !grant.binding_id.is_empty()
+            && host.writer_binding_id() != Some(grant.binding_id.as_str())
+        {
+            return Err(Error::WorkspacePermissionDenied(
+                "Journal recovery binding mismatch".into(),
+            ));
+        }
+        host
+    } else {
+        host(api, &store, grant)?
+    };
     let erased = store.journal_session_erased(&host).map_err(domain)?;
     if !erased {
         job.load_executable_request()?;
@@ -2204,70 +2343,93 @@ pub(super) fn reconcile_unknown(
             job_id: job.request.job_id.clone(),
             attempt_id: attempt.attempt_id.clone(),
         };
-        let g = if legacy_descriptor(&job.request) {
-            // Only reconcile already-saved legacy receipts; never redispatch descriptors.
+        let generation = if released {
             store
-                .prepare_journal_generation(
-                    &host,
-                    &grant.journal_id,
-                    binding,
-                    grant.selected_ranges.clone(),
-                )
+                .retained_journal_generation(&host, &grant.journal_id, &binding)
                 .map_err(domain)?
+        } else if legacy_descriptor(&job.request) {
+            Some(
+                store
+                    .prepare_journal_generation(
+                        &host,
+                        &grant.journal_id,
+                        binding,
+                        grant.selected_ranges.clone(),
+                    )
+                    .map_err(domain)?,
+            )
         } else {
-            let input = saved_input(&job.request, grant)?;
-            let generation = store
-                .prepare_journal_input(&host, &input, binding)
-                .map_err(domain)?;
-            input.validate_generation(&generation).map_err(domain)?;
-            generation
+            // Preserve active-body recovery: its saved intent can still establish
+            // the original generation fence. A released body must only read it.
+            Some(
+                store
+                    .prepare_journal_input(&host, &saved_input(&job.request, grant)?, binding)
+                    .map_err(domain)?,
+            )
         };
-        let ctx = JournalJobContext::validated_recovery(&store, &host, &g.id, g.job.clone())
-            .map_err(domain)?;
-        match journal::service::finish(&store, &ctx) {
-            Ok(completion) => {
-                let submission = BackendJobResultSubmission {
-                    job_id: job.request.job_id.clone(),
-                    attempt_id: attempt.attempt_id.clone(),
-                    input_digest: attempt.input_digest.clone(),
-                    result: serde_json::to_value(completion)
-                        .map_err(|e| Error::Store(e.to_string()))?,
-                };
-                let current = api.store.reconcile_backend_journal_attempt(
-                    api.workspace_id(),
-                    attempt,
-                    Some(&submission),
-                    &Utc::now().to_rfc3339(),
-                )?;
-                let saved = api
-                    .store
-                    .get_backend_job(api.workspace_id(), &job.request.job_id)?
-                    .ok_or_else(|| Error::Store("reconciled Job missing".into()))?;
-                notification_delivery::deliver_job_result(
-                    api,
-                    &BackendJobResultAcceptance {
-                        job: saved,
-                        attempt: current.clone(),
-                        replayed: false,
-                    },
-                );
-                if current.worker_cleanup_state == Some(BackendJobWorkerCleanupState::Completed) {
-                    drain_after_completion(api, &job.request);
-                    if job.input_disposition == crate::backend_job::JobInputDisposition::Available {
-                        api.store.reclaim_backend_job_input(
-                            api.workspace_id(),
-                            &job.request.job_id,
-                            &job.input_digest,
-                            crate::backend_job::JobInputRetirementReason::Completed,
-                        )?;
+        if let Some(g) = generation {
+            if g.selected_ranges != grant.selected_ranges {
+                return Err(Error::WorkspacePermissionDenied(
+                    "Journal recovery selection mismatch".into(),
+                ));
+            }
+            if !legacy_descriptor(&job.request) {
+                saved_input(&job.request, grant)?
+                    .validate_generation(&g)
+                    .map_err(domain)?;
+            }
+            let ctx = JournalJobContext::validated_recovery(&store, &host, &g.id, g.job.clone())
+                .map_err(domain)?;
+            match journal::service::finish(&store, &ctx) {
+                Ok(completion) => {
+                    let submission = BackendJobResultSubmission {
+                        job_id: job.request.job_id.clone(),
+                        attempt_id: attempt.attempt_id.clone(),
+                        input_digest: attempt.input_digest.clone(),
+                        result: serde_json::to_value(completion)
+                            .map_err(|e| Error::Store(e.to_string()))?,
+                    };
+                    let current = api.store.reconcile_backend_journal_attempt(
+                        api.workspace_id(),
+                        attempt,
+                        Some(&submission),
+                        &Utc::now().to_rfc3339(),
+                    )?;
+                    let saved = api
+                        .store
+                        .get_backend_job(api.workspace_id(), &job.request.job_id)?
+                        .ok_or_else(|| Error::Store("reconciled Job missing".into()))?;
+                    notification_delivery::deliver_job_result(
+                        api,
+                        &BackendJobResultAcceptance {
+                            job: saved,
+                            attempt: current.clone(),
+                            replayed: false,
+                        },
+                    );
+                    if current.worker_cleanup_state == Some(BackendJobWorkerCleanupState::Completed)
+                    {
+                        if !released {
+                            drain_after_completion(api, &job.request);
+                        }
+                        if job.input_disposition
+                            == crate::backend_job::JobInputDisposition::Available
+                        {
+                            api.store.reclaim_backend_job_input(
+                                api.workspace_id(),
+                                &job.request.job_id,
+                                &job.input_digest,
+                                crate::backend_job::JobInputRetirementReason::Completed,
+                            )?;
+                        }
+                    } else {
+                        api.schedule_backend_job_worker_cleanup(&current);
                     }
-                } else {
-                    api.schedule_backend_job_worker_cleanup(&current);
+                    return Ok(());
                 }
-                return Ok(());
+                Err(JournalError::Conflict(_) | JournalError::PermissionDenied(_)) => {}
+                Err(error) => return Err(domain(error)),
             }
-            Err(JournalError::Conflict(_) | JournalError::PermissionDenied(_)) => {}
-            Err(error) => return Err(domain(error)),
         }
     }
     if attempt.worker_cleanup_state != Some(BackendJobWorkerCleanupState::Completed) {
@@ -2286,11 +2448,17 @@ pub(super) fn reconcile_unknown(
         false
     };
     if !confirmed_purge {
+        if released {
+            // Source deletion already fenced this generation. An unbound Unknown
+            // still lacks execution-cleanup proof: preserve Unknown, but it no
+            // longer owns the Subject or blocks removal of the original body.
+            return Ok(());
+        }
         return Err(Error::RepositoryConflict(
             "Unknown Journal attempt has no committed stop/Purge fence; not retried".into(),
         ));
     }
-    if !erased {
+    if !erased && !released {
         close(api, &job.request, attempt)?;
     }
     api.store.reconcile_backend_journal_attempt(
@@ -2299,7 +2467,7 @@ pub(super) fn reconcile_unknown(
         None,
         &Utc::now().to_rfc3339(),
     )?;
-    if !erased {
+    if !erased && !released {
         // Resume the same saved body input only after canonical execution/domain fences.
         // dispatch selects its retained intent before it can capture any newer source.
         let attr = store
@@ -2316,6 +2484,10 @@ fn reconcile_session_jobs(
     attr: &crate::subjektiv::SubjectSessionAttribution,
     erased: bool,
 ) -> Result<()> {
+    let released = !erased
+        && store(api)?
+            .journal_worker_was_released(&attr.subject_id, &attr.runtime_id, &attr.worker_id)
+            .map_err(domain)?;
     let mut after: Option<String> = None;
     loop {
         let jobs = api.store.list_backend_journal_jobs(
@@ -2330,6 +2502,29 @@ fn reconcile_session_jobs(
                 .store
                 .get_backend_job_attempt(api.workspace_id(), &job.request.job_id, &id)?
                 .ok_or_else(|| Error::Store("Journal current attempt missing".into()))?;
+            if released {
+                // Writer revocation has already fenced all Journal mutations. End
+                // pending/Unknown attempts without claiming their input was reflected;
+                // saved input, operations and original source remain retained.
+                if !attempt.state.terminal() {
+                    attempt = api.store.finish_backend_job_attempt(
+                        api.workspace_id(),
+                        &job.request.job_id,
+                        &id,
+                        BackendJobAttemptState::Failed,
+                        "journal_body_released",
+                        "Source Worker was removed; unfinished Journal history retained",
+                        &Utc::now().to_rfc3339(),
+                    )?;
+                }
+                api.schedule_backend_job_worker_cleanup(&attempt);
+                if attempt.state == BackendJobAttemptState::Unknown {
+                    // A saved checkpoint may settle success; otherwise retain Unknown
+                    // until canonical Job-worker cleanup has supplied its Purge fence.
+                    reconcile_unknown(api, job, &attempt)?;
+                }
+                continue;
+            }
             if !erased
                 && legacy_descriptor(&job.request)
                 && job.input_disposition == crate::backend_job::JobInputDisposition::Available

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

@@ -314,6 +314,7 @@ impl Fixture {
                 &workdir_id,
                 "account:close-test",
                 "remove normally after session close",
+                &uuid::Uuid::now_v7().to_string(),
             )
         })
         .await

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

@@ -652,6 +652,7 @@ async fn source_workdir_removal_keeps_published_drive_bytes_and_receipt() {
             &removal_id,
             "account:drive-test",
             "source no longer needed",
+            "drive-source-removal",
         )
     })
     .await

crates/workspace-server/src/server_workspace_config_tests.rs

@@ -819,6 +819,7 @@ mod workspace_config_integration {
                 },
                 "owner".into(),
                 WorkingDirectoryRemovalRequest {
+                    operation_id: "unsupported-logical-cleanup".into(),
                     reason: "unsupported logical cleanup".into()
                 }
             )

crates/workspace-server/src/store.rs

@@ -3253,8 +3253,8 @@ impl SqliteWorkspaceStore {
                         },
                     )
                     .optional()?;
-                // Subject occupancy survives removal until an explicit release
-                // proof. It must not resurrect the removed creation reservation:
+                // An unconfirmed legacy removal cannot release Subject occupancy.
+                // Completed removal or explicit release does; never resurrect a reservation:
                 // Runtime callbacks require a live reservation or catalog member.
                 if stored.as_ref().is_some_and(|row| row.6 == "removed") {
                     return Err(Error::RepositoryConflict(format!(
@@ -11443,49 +11443,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
                     record.workdir_id
                 )));
             }
-            let active_for_alias = tx
-                .query_row(
-                    r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at, connection_id
-                       FROM worker_workdir_links
-                       WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3
-                         AND alias = ?4 AND unlinked_at IS NULL"#,
-                    params![
-                        record.workspace_id,
-                        record.worker.runtime_id,
-                        record.worker.worker_id,
-                        record.alias,
-                    ],
-                    read_worker_workdir_link_record,
-                )
-                .optional()?;
-            if let Some(mut active) = active_for_alias {
-                if active.workdir_id == record.workdir_id {
-                    let capabilities = active.capabilities.intersection(record.capabilities);
-                    if capabilities != active.capabilities {
-                        tx.execute(
-                            r#"UPDATE worker_workdir_links
-                               SET capabilities = ?1
-                               WHERE workspace_id = ?2 AND runtime_id = ?3 AND worker_id = ?4
-                                 AND workdir_id = ?5 AND alias = ?6 AND unlinked_at IS NULL"#,
-                            params![
-                                encode_workdir_link_capabilities(capabilities)?,
-                                record.workspace_id,
-                                record.worker.runtime_id,
-                                record.worker.worker_id,
-                                record.workdir_id,
-                                record.alias,
-                            ],
-                        )?;
-                        active.capabilities = capabilities;
-                    }
-                    tx.commit()?;
-                    return Ok(active);
-                }
-                return Err(Error::WorkdirAttachmentConflict(format!(
-                    "Worker {}:{} attachment alias `{}` is already bound to Workdir {}",
-                    record.worker.runtime_id, record.worker.worker_id, record.alias, active.workdir_id
-                )));
-            }
+            // Ownership is keyed by Workdir, not by the requested local name.
             let active_for_workdir = tx
                 .query_row(
                     r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at, connection_id
@@ -11495,10 +11453,45 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
                     read_worker_workdir_link_record,
                 )
                 .optional()?;
-            if let Some(active) = active_for_workdir {
+            if let Some(mut active) = active_for_workdir {
+                if active.worker != record.worker {
+                    return Err(Error::WorkdirOccupied {
+                        workdir_id: record.workdir_id.clone(),
+                        runtime_id: active.worker.runtime_id,
+                        worker_id: active.worker.worker_id,
+                    });
+                }
+                let capabilities = active.capabilities.intersection(record.capabilities);
+                if capabilities != active.capabilities {
+                    tx.execute(
+                        r#"UPDATE worker_workdir_links SET capabilities = ?1
+                           WHERE workspace_id = ?2 AND runtime_id = ?3 AND worker_id = ?4
+                             AND workdir_id = ?5 AND alias = ?6 AND unlinked_at IS NULL"#,
+                        params![
+                            encode_workdir_link_capabilities(capabilities)?,
+                            active.workspace_id, active.worker.runtime_id, active.worker.worker_id,
+                            active.workdir_id, active.alias,
+                        ],
+                    )?;
+                    active.capabilities = capabilities;
+                }
+                tx.commit()?;
+                return Ok(active);
+            }
+            let active_for_alias = tx
+                .query_row(
+                    r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at, connection_id
+                       FROM worker_workdir_links
+                       WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3
+                         AND alias = ?4 AND unlinked_at IS NULL"#,
+                    params![record.workspace_id, record.worker.runtime_id, record.worker.worker_id, record.alias],
+                    read_worker_workdir_link_record,
+                )
+                .optional()?;
+            if let Some(active) = active_for_alias {
                 return Err(Error::WorkdirAttachmentConflict(format!(
-                    "Workdir {} is already attached to Worker {}:{}",
-                    record.workdir_id, active.worker.runtime_id, active.worker.worker_id
+                    "Worker {}:{} attachment alias `{}` is already bound to Workdir {}",
+                    record.worker.runtime_id, record.worker.worker_id, record.alias, active.workdir_id
                 )));
             }
             // Direct attachments (including ordinary recovery/reconnection)
@@ -16746,7 +16739,7 @@ fn current_worker_singleton_owner(
                     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'))) \
+                    AND NOT EXISTS(SELECT 1 FROM worker_removal_operations r WHERE r.workspace_id=owner.workspace_id AND r.worker_id=owner.worker_id AND r.runtime_id=owner.runtime_id AND r.state='succeeded') \
                       AND NOT EXISTS(SELECT 1 FROM subject_deletion_workers 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='subjektiv:'||d.subject_id AND d.originals_absent=1) \
                     THEN 'created' ELSE reservation.state END \
              FROM worker_singleton_owners owner \
@@ -26065,6 +26058,24 @@ INSERT INTO worker_registry (
             2
         );
 
+        for alias in ["another-name", "docs"] {
+            let requested = WorkerWorkdirLinkRecord {
+                alias: alias.into(),
+                ..link.clone()
+            };
+            assert_eq!(
+                store.attach_worker_workdir(&requested).unwrap(),
+                downgraded_link
+            );
+            assert_eq!(
+                store
+                    .list_worker_workdir_links("local-dev", &worker.worker)
+                    .unwrap()
+                    .len(),
+                2
+            );
+        }
+
         let second_worker = WorkerRegistryRecord {
             worker: RuntimeWorkerRef::new("embedded", "2"),
             display_name: "Browser 2".to_string(),
@@ -26080,7 +26091,7 @@ INSERT INTO worker_registry (
         };
         assert!(matches!(
             store.attach_worker_workdir(&workdir_conflict),
-            Err(Error::WorkdirAttachmentConflict(_))
+            Err(Error::WorkdirOccupied { .. })
         ));
         assert!(matches!(
             store.detach_worker_workdir("local-dev", &worker.worker, Some("wrong-workdir"), "6",),

crates/workspace-server/src/subject_deletion_store.rs

@@ -1289,8 +1289,9 @@ mod tests {
         reservation(&s, &body, "body", Some("subjektiv:subject"));
         receipt(&s, &body, "succeeded", "purge", "purge");
         let body_receipt = format!("receipt-{}-succeeded", body.worker_id);
-        // A different Subject's already-removed body remains occupied until its
-        // own deletion/release. Deleting a delegating Subject must not release it.
+        // Another Subject's completed removal already releases its occupancy.
+        // Deleting the delegating Subject must preserve that independent outcome
+        // and must not erase the other Subject's Job.
         let other_body = worker(33);
         reservation(&s, &other_body, "other-body", Some("subjektiv:another"));
         receipt(&s, &other_body, "succeeded", "purge", "purge");
@@ -1307,6 +1308,11 @@ mod tests {
         }).unwrap();
         let archive_before = s.worker_session_archive("w", &jw, "session").unwrap();
         assert!(archive_before.is_some());
+        assert!(
+            s.current_worker_singleton_owner("w", "subjektiv:another")
+                .unwrap()
+                .is_none()
+        );
         s.prepare_subject_deletion("w", "subject", "delete", std::slice::from_ref(&other_body))
             .unwrap();
         assert_eq!(saved_prior(&s, "subject", &jw), Some(job_receipt.clone()));
@@ -1352,7 +1358,7 @@ mod tests {
         assert!(
             s.current_worker_singleton_owner("w", "subjektiv:another")
                 .unwrap()
-                .is_some()
+                .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",[&job_receipt],|r|Ok((r.get(0)?,r.get(1)?,r.get(2)?)))?;

crates/workspace-server/src/workdir_removal.rs

@@ -1,24 +1,15 @@
-//! Persists durable deletion intent and attempts for Repository-backed Workdirs.
+//! Records one Repository-backed Workdir removal request and its result.
 //!
-//! Deletion intent fixes Workspace, Workdir, owning Runtime, Repository/materialization, actor,
-//! and reason. Endpoints, host paths, and provider handles do not grant caller deletion authority.
+//! Caller-generated request identities distinguish an exact resend from a new explicit request;
+//! reason text is audit input, never a replay key. Failed and completed records are terminal,
+//! including legacy failed records marked retryable. Retryability only advises a new explicit
+//! request after resolving its cause; it is not continuing authority to delete.
+//!
+//! Only one unresolved request per Workdir is allowed. Attempts are claimed transactionally and
+//! fenced by process/start marker and attempt identity. Recovery observes uncertain external
+//! effects but never reissues deletion. Confirmed absence and the matching materialization permit
+//! registry removal and the saved result to commit together. Unconfirmed effects remain pending.
 //! External Workdirs follow grant revocation, not this operation.
-//!
-//! Only one pending operation per Workdir is allowed; attempts are claimed transactionally.
-//! Process IDs and startup markers are retained. Recovery reclaims the exact owner/attempt by
-//! compare-and-set only after confirming the previous owner is gone; live or unobservable owners
-//! are not displaced. In-process locks alone are insufficient.
-//!
-//! Server execution revalidates stored materialization and current attachments, reservations,
-//! assignments, retention, and creation operations, then observes occupancy and cleanliness via
-//! Runtime. Dirty resources remain; unknown observations or provider failures retain the registry
-//! and record attention/retryability. Only confirmed deletion or exact working_directory_not_found
-//! is deletion evidence; registry removal and completed/removed commit in the same transaction.
-//!
-//! Normal deletion, Workspace deletion, cleanup, and startup recovery use the same reserved Server
-//! operation execution. Failed DB commits after provider deletion are reobserved through that
-//! operation. Recovery relies on provider identity and results, not an exactly-once guarantee
-//! across all interruption points.
 
 use chrono::Utc;
 use rusqlite::{Connection, OptionalExtension, TransactionBehavior, params};
@@ -118,10 +109,22 @@ pub fn workdir_materialization_fingerprint(record: &WorkdirRegistryRecord) -> St
     hex_sha256(&bytes)
 }
 
+pub fn workdir_removal_request_id(
+    workspace_id: &str,
+    source_actor: &str,
+    request_id: &str,
+) -> Result<String> {
+    validate_bounded("operation id", request_id, 256)?;
+    let bytes = serde_json::to_vec(&(workspace_id, source_actor, request_id))
+        .map_err(|error| Error::Store(error.to_string()))?;
+    Ok(format!("wdr_{}", &hex_sha256(&bytes)[..32]))
+}
+
 pub fn workdir_removal_intent(
     record: &WorkdirRegistryRecord,
     source_actor: &str,
     reason: &str,
+    request_id: &str,
 ) -> Result<WorkdirRemovalIntent> {
     validate_bounded("source actor", source_actor, MAX_ACTOR_BYTES)?;
     validate_bounded("reason", reason, MAX_REASON_BYTES)?;
@@ -146,7 +149,7 @@ pub fn workdir_removal_intent(
         ));
     };
     Ok(WorkdirRemovalIntent {
-        operation_id: format!("wdr_{}", &request_fingerprint[..32]),
+        operation_id: workdir_removal_request_id(&record.workspace_id, source_actor, request_id)?,
         request_fingerprint,
         workspace_id: record.workspace_id.clone(),
         working_directory_id: record.workdir_id.clone(),
@@ -285,17 +288,10 @@ impl SqliteWorkspaceStore {
             let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
             let operation =
                 require_operation(&tx, workspace_id, operation_id, request_fingerprint)?;
-            if operation.state == WorkdirRemovalOperationState::Completed
-                && operation.disposition == Some(WorkdirRemovalDisposition::Removed)
-            {
+            if operation.state != WorkdirRemovalOperationState::Pending {
                 tx.commit()?;
                 return Ok(operation);
             }
-            if operation.state == WorkdirRemovalOperationState::Failed && !operation.retryable {
-                return Err(Error::InvalidInput(format!(
-                    "Workdir removal operation `{operation_id}` is not retryable"
-                )));
-            }
             if let Some((expected_owner, expected_attempt_id)) = recovery_expected {
                 if operation.state != WorkdirRemovalOperationState::Pending
                     || operation.attempt_owner != expected_owner
@@ -473,7 +469,7 @@ impl SqliteWorkspaceStore {
                 &operation.operation_id,
                 &operation.request_fingerprint,
             )?;
-            if current.state == WorkdirRemovalOperationState::Completed {
+            if current.state != WorkdirRemovalOperationState::Pending {
                 tx.commit()?;
                 return Ok(current);
             }
@@ -502,6 +498,39 @@ impl SqliteWorkspaceStore {
         })
     }
 
+    /// Keep an uncertain external outcome for observation, not automatic re-execution.
+    pub fn record_workdir_removal_unknown(
+        &self,
+        operation: &WorkdirRemovalOperation,
+    ) -> Result<WorkdirRemovalOperation> {
+        self.with_conn_mut(|conn| {
+            let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
+            let current = require_operation(
+                &tx,
+                &operation.workspace_id,
+                &operation.operation_id,
+                &operation.request_fingerprint,
+            )?;
+            if current.state != WorkdirRemovalOperationState::Pending {
+                tx.commit()?;
+                return Ok(current);
+            }
+            require_matching_attempt(operation, &current)?;
+            tx.execute(
+                "UPDATE workdir_removal_operations SET retryable=0, disposition='attention_required', failure_category='provider_cleanup_outcome_unknown', attempt_owner_pid=NULL, attempt_owner_start_marker=NULL, updated_at=?1 WHERE workspace_id=?2 AND operation_id=?3",
+                params![Utc::now().to_rfc3339(), operation.workspace_id, operation.operation_id],
+            )?;
+            let updated = require_operation(
+                &tx,
+                &operation.workspace_id,
+                &operation.operation_id,
+                &operation.request_fingerprint,
+            )?;
+            tx.commit()?;
+            Ok(updated)
+        })
+    }
+
     pub fn commit_workdir_removal_removed(
         &self,
         operation: &WorkdirRemovalOperation,
@@ -531,7 +560,7 @@ impl SqliteWorkspaceStore {
                 &operation.operation_id,
                 &operation.request_fingerprint,
             )?;
-            if current.state == WorkdirRemovalOperationState::Completed {
+            if current.state != WorkdirRemovalOperationState::Pending {
                 tx.commit()?;
                 return Ok(current);
             }
@@ -541,24 +570,29 @@ impl SqliteWorkspaceStore {
                     &tx,
                     &operation.workspace_id,
                     &operation.working_directory_id,
-                )?
-                .ok_or_else(|| Error::RegistryInconsistency(
-                    "Workdir registry row disappeared before durable removal commit".to_string(),
-                ))?;
-                require_operation_materialization(operation, &registry)?;
-                require_no_removal_blockers(
-                    &tx,
-                    &operation.workspace_id,
-                    &operation.working_directory_id,
                 )?;
-                let deleted = tx.execute(
-                    "DELETE FROM workdir_registry WHERE workspace_id=?1 AND workdir_id=?2",
-                    params![operation.workspace_id, operation.working_directory_id],
-                )?;
-                if deleted != 1 {
-                    return Err(Error::RegistryInconsistency(
-                        "Workdir registry deletion did not remove exactly one row".to_string(),
-                    ));
+                if let Some(registry) = registry {
+                    require_operation_materialization(operation, &registry)?;
+                    require_no_removal_blockers(
+                        &tx,
+                        &operation.workspace_id,
+                        &operation.working_directory_id,
+                    )?;
+                    tx.execute(
+                        "DELETE FROM workdir_registry WHERE workspace_id=?1 AND workdir_id=?2",
+                        params![operation.workspace_id, operation.working_directory_id],
+                    )?;
+                } else {
+                    let removed: bool = tx.query_row(
+                        "SELECT EXISTS(SELECT 1 FROM workdir_removal_operations WHERE workspace_id=?1 AND workdir_id=?2 AND materialization_fingerprint=?3 AND state='completed' AND disposition='removed')",
+                        params![operation.workspace_id, operation.working_directory_id, operation.materialization_fingerprint],
+                        |row| row.get(0),
+                    )?;
+                    if !removed {
+                        return Err(Error::RegistryInconsistency(
+                            "Workdir registry disappeared without confirmed removal".to_string(),
+                        ));
+                    }
                 }
             }
             let now = Utc::now().to_rfc3339();
@@ -585,20 +619,18 @@ impl SqliteWorkspaceStore {
         })
     }
 
-    pub fn find_workdir_removal_operation_by_intent(
+    pub fn completed_workdir_removal(
         &self,
         workspace_id: &str,
         working_directory_id: &str,
-        source_actor: &str,
-        reason: &str,
     ) -> Result<Option<WorkdirRemovalOperation>> {
         self.with_conn(|conn| {
             conn.query_row(
                 &format!(
-                    "{} WHERE workspace_id=?1 AND workdir_id=?2 AND source_actor=?3 AND reason=?4 ORDER BY created_at DESC, operation_id DESC LIMIT 1",
+                    "{} WHERE workspace_id=?1 AND workdir_id=?2 AND state='completed' AND disposition='removed' ORDER BY completed_at DESC, operation_id DESC LIMIT 1",
                     operation_select_sql()
                 ),
-                params![workspace_id, working_directory_id, source_actor, reason],
+                params![workspace_id, working_directory_id],
                 read_operation,
             )
             .optional()
@@ -614,7 +646,7 @@ impl SqliteWorkspaceStore {
         self.with_conn(|conn| {
             let mut statement = conn.prepare(
                 &format!(
-                    "{} WHERE workspace_id=?1 AND (state='pending' OR (state='failed' AND retryable=1)) ORDER BY updated_at ASC, operation_id ASC LIMIT ?2",
+                    "{} WHERE workspace_id=?1 AND state='pending' ORDER BY updated_at ASC, operation_id ASC LIMIT ?2",
                     operation_select_sql()
                 ),
             )?;
@@ -1007,8 +1039,13 @@ mod tests {
     #[tokio::test]
     async fn exact_replay_reuses_operation_and_conflicting_intent_fails() {
         let (store, workdir) = seeded_store().await;
-        let intent =
-            workdir_removal_intent(&workdir, "worker:W-1", "remove stale Workdir").unwrap();
+        let intent = workdir_removal_intent(
+            &workdir,
+            "worker:W-1",
+            "remove stale Workdir",
+            "test-removal-request",
+        )
+        .unwrap();
         let first = store.reserve_workdir_removal_operation(&intent).unwrap();
         let replay = store.reserve_workdir_removal_operation(&intent).unwrap();
         assert_eq!(replay, first);
@@ -1023,10 +1060,15 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn failed_attempt_is_retryable_and_completed_retry_replays() {
+    async fn failed_request_is_terminal_and_new_request_has_independent_result() {
         let (store, workdir) = seeded_store().await;
-        let intent =
-            workdir_removal_intent(&workdir, "workspace-api", "remove clean Workdir").unwrap();
+        let intent = workdir_removal_intent(
+            &workdir,
+            "workspace-api",
+            "remove clean Workdir",
+            "test-removal-request",
+        )
+        .unwrap();
         let reserved = store.reserve_workdir_removal_operation(&intent).unwrap();
         let first = store
             .begin_workdir_removal_attempt(
@@ -1049,7 +1091,32 @@ mod tests {
                 attempt_owner(),
             )
             .unwrap();
-        assert_ne!(retry.attempt_id, first.attempt_id);
+        assert_eq!(retry, failed);
+        assert!(
+            store
+                .recoverable_workdir_removal_operations("workspace-a", 100)
+                .unwrap()
+                .is_empty()
+        );
+        let next_intent = workdir_removal_intent(
+            &workdir,
+            "workspace-api",
+            "remove clean Workdir",
+            "new-request",
+        )
+        .unwrap();
+        let next = store
+            .reserve_workdir_removal_operation(&next_intent)
+            .unwrap();
+        let retry = store
+            .begin_workdir_removal_attempt(
+                &next.workspace_id,
+                &next.operation_id,
+                &next.request_fingerprint,
+                attempt_owner(),
+            )
+            .unwrap();
+        assert_ne!(retry.operation_id, failed.operation_id);
         let completed = store.commit_workdir_removal_removed(&retry).unwrap();
         assert_eq!(
             completed.disposition,
@@ -1075,7 +1142,13 @@ mod tests {
     #[tokio::test]
     async fn recovery_and_completion_are_fenced_by_attempt_identity() {
         let (store, workdir) = seeded_store().await;
-        let intent = workdir_removal_intent(&workdir, "workspace-api", "remove Workdir").unwrap();
+        let intent = workdir_removal_intent(
+            &workdir,
+            "workspace-api",
+            "remove Workdir",
+            "test-removal-request",
+        )
+        .unwrap();
         let reserved = store.reserve_workdir_removal_operation(&intent).unwrap();
         let first = store
             .begin_workdir_removal_attempt(
@@ -1130,8 +1203,13 @@ mod tests {
         use std::sync::{Arc, Barrier};
 
         let (store, workdir) = seeded_store().await;
-        let intent =
-            workdir_removal_intent(&workdir, "workspace-api", "remove clean Workdir").unwrap();
+        let intent = workdir_removal_intent(
+            &workdir,
+            "workspace-api",
+            "remove clean Workdir",
+            "test-removal-request",
+        )
+        .unwrap();
         let operation = store.reserve_workdir_removal_operation(&intent).unwrap();
         let barrier = Arc::new(Barrier::new(3));
         let provider_calls = Arc::new(AtomicUsize::new(0));
@@ -1218,8 +1296,13 @@ mod tests {
                 "2",
             )
             .unwrap();
-        let intent =
-            workdir_removal_intent(&workdir, "workspace-api", "remove clean Workdir").unwrap();
+        let intent = workdir_removal_intent(
+            &workdir,
+            "workspace-api",
+            "remove clean Workdir",
+            "test-removal-request",
+        )
+        .unwrap();
         store.reserve_workdir_removal_operation(&intent).unwrap();
 
         let error = store
@@ -1236,8 +1319,13 @@ mod tests {
     #[tokio::test]
     async fn pending_removal_fences_new_attachment_and_retry_rereads_live_reservation() {
         let (store, workdir) = seeded_store().await;
-        let intent =
-            workdir_removal_intent(&workdir, "workspace-api", "remove clean Workdir").unwrap();
+        let intent = workdir_removal_intent(
+            &workdir,
+            "workspace-api",
+            "remove clean Workdir",
+            "test-removal-request",
+        )
+        .unwrap();
         let pending = store.reserve_workdir_removal_operation(&intent).unwrap();
 
         let error = store
@@ -1251,11 +1339,22 @@ mod tests {
         store
             .reserve_worker_workdir_attachment("workspace-a", "workdir-a", "reservation-b", "3")
             .unwrap();
+        let next_intent = workdir_removal_intent(
+            &workdir,
+            "workspace-api",
+            "remove clean Workdir",
+            "new-request",
+        )
+        .unwrap();
+        let next = store
+            .reserve_workdir_removal_operation(&next_intent)
+            .unwrap();
+        assert_ne!(next.operation_id, failed.operation_id);
         let retry = store
             .begin_workdir_removal_attempt(
-                &failed.workspace_id,
-                &failed.operation_id,
-                &failed.request_fingerprint,
+                &next.workspace_id,
+                &next.operation_id,
+                &next.request_fingerprint,
                 attempt_owner(),
             )
             .unwrap();
@@ -1324,7 +1423,13 @@ mod tests {
                     Ok(())
                 })
                 .unwrap();
-        let intent = workdir_removal_intent(&workdir, "workspace-api", "remove Workdir").unwrap();
+        let intent = workdir_removal_intent(
+            &workdir,
+            "workspace-api",
+            "remove Workdir",
+            "test-removal-request",
+        )
+        .unwrap();
         let operation = store.reserve_workdir_removal_operation(&intent).unwrap();
         let categories = |guards: Vec<WorkdirRemovalGuard>| {
             guards
@@ -1447,8 +1552,13 @@ mod tests {
     #[tokio::test]
     async fn retained_completion_keeps_registry_and_is_auditable() {
         let (store, workdir) = seeded_store().await;
-        let intent =
-            workdir_removal_intent(&workdir, "workspace-api", "inspect dirty Workdir").unwrap();
+        let intent = workdir_removal_intent(
+            &workdir,
+            "workspace-api",
+            "inspect dirty Workdir",
+            "test-removal-request",
+        )
+        .unwrap();
         let operation = store.reserve_workdir_removal_operation(&intent).unwrap();
         let retained = store
             .complete_workdir_removal_retained(
@@ -1477,10 +1587,12 @@ mod tests {
                 attempt_owner(),
             )
             .unwrap();
-        assert_eq!(retry.state, WorkdirRemovalOperationState::Pending);
-        assert!(retry.attempt_id.is_some());
-        assert_eq!(retry.disposition, None);
-        assert_eq!(retry.failure_category, None);
-        assert!(retry.retryable);
+        assert_eq!(retry, retained);
+        assert!(
+            store
+                .recoverable_workdir_removal_operations("workspace-a", 100)
+                .unwrap()
+                .is_empty()
+        );
     }
 }

openapi/server-api.json

@@ -20698,11 +20698,16 @@
       "WorkingDirectoryRemovalRequest": {
         "additionalProperties": false,
         "properties": {
+          "operation_id": {
+            "description": "Stable for an exact resend; a new explicit request uses a new identity.",
+            "type": "string"
+          },
           "reason": {
             "type": "string"
           }
         },
         "required": [
+          "operation_id",
           "reason"
         ],
         "type": "object"
@@ -22715,7 +22720,7 @@
   "info": {
     "title": "Yoi Server API",
     "version": "0.1.0",
-    "x-yoi-source-digest": "sha256:8cc3c45d082e52b1ea9a014721008aeb4c0699b987729d362c0c14a6f485df10"
+    "x-yoi-source-digest": "sha256:2ac605f7ea70d0c725f9993a5c59c4a8c4266cf69184902db7b3ea7fd5932224"
   },
   "openapi": "3.1.0",
   "paths": {

web/workspace/src/lib/generated/companion-api.ts

@@ -1,6 +1,6 @@
 // @generated by yoi-companion-openapi-typescript@1. Do not edit.
 // Input: openapi/server-api.json
-// Input contract: 0.1.0 (sha256:8cc3c45d082e52b1ea9a014721008aeb4c0699b987729d362c0c14a6f485df10)
+// Input contract: 0.1.0 (sha256:2ac605f7ea70d0c725f9993a5c59c4a8c4266cf69184902db7b3ea7fd5932224)
 // Options: companion-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output
 // Output: web/workspace/src/lib/generated/companion-api.ts
 // Regenerate/check: cargo run -q -p server-api --example generate_companion_api_types -- --check

web/workspace/src/lib/generated/repository-api.ts

@@ -1,6 +1,6 @@
 // @generated by yoi-repository-openapi-typescript@1. Do not edit.
 // Input: openapi/server-api.json
-// Input contract: 0.1.0 (sha256:8cc3c45d082e52b1ea9a014721008aeb4c0699b987729d362c0c14a6f485df10)
+// Input contract: 0.1.0 (sha256:2ac605f7ea70d0c725f9993a5c59c4a8c4266cf69184902db7b3ea7fd5932224)
 // Options: repository-operation-schema-closure,openapi-3.1,format:uint64=safe-number,strict-safe-integers,sorted-output
 // Output: web/workspace/src/lib/generated/repository-api.ts
 // Regenerate/check: cargo run -q -p server-api --example generate_repository_openapi_types -- --check

web/workspace/src/lib/generated/runtime-api.ts

@@ -1,6 +1,6 @@
 // @generated by yoi-runtime-openapi-typescript@1. Do not edit.
 // Input: openapi/server-api.json
-// Input contract: 0.1.0 (sha256:8cc3c45d082e52b1ea9a014721008aeb4c0699b987729d362c0c14a6f485df10)
+// Input contract: 0.1.0 (sha256:2ac605f7ea70d0c725f9993a5c59c4a8c4266cf69184902db7b3ea7fd5932224)
 // Options: runtime-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output
 // Output: web/workspace/src/lib/generated/runtime-api.ts
 // Regenerate/check: cargo run -q -p server-api --example generate_runtime_api_types -- --check

web/workspace/src/lib/generated/ticket-api.ts

@@ -1,6 +1,6 @@
 // @generated by yoi-ticket-browser-openapi-typescript@1. Do not edit.
 // Input: openapi/server-api.json
-// Input contract: 0.1.0 (sha256:8cc3c45d082e52b1ea9a014721008aeb4c0699b987729d362c0c14a6f485df10)
+// Input contract: 0.1.0 (sha256:2ac605f7ea70d0c725f9993a5c59c4a8c4266cf69184902db7b3ea7fd5932224)
 // Options: ticket-objective-merge-request-browser-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output
 // Output: web/workspace/src/lib/generated/ticket-api.ts
 // Regenerate/check: cargo run -q -p server-api --example generate_ticket_api_types -- --check

web/workspace/src/lib/generated/workdir-api.ts

@@ -1,6 +1,6 @@
 // @generated by yoi-workdir-openapi-typescript@1. Do not edit.
 // Input: openapi/server-api.json
-// Input contract: 0.1.0 (sha256:1053b09bb77ff0fc09fec2d3f96af04b76974c5ea52870f805381302e1b29fe4)
+// Input contract: 0.1.0 (sha256:2ac605f7ea70d0c725f9993a5c59c4a8c4266cf69184902db7b3ea7fd5932224)
 // Options: workdir-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output
 // Output: web/workspace/src/lib/generated/workdir-api.ts
 // Regenerate/check: cargo run -q -p server-api --example generate_workdir_api_types -- --check
@@ -33,7 +33,7 @@ export type WorkingDirectoryOccupancy = { display_name: string; linked_at: strin
 
 export type WorkingDirectoryRemovalDisposition = "removed" | "retained" | "attention_required";
 
-export type WorkingDirectoryRemovalRequest = { reason: string; };
+export type WorkingDirectoryRemovalRequest = { operation_id: string; reason: string; };
 
 export type WorkingDirectoryRemovalResponse = { disposition: WorkingDirectoryRemovalDisposition; failure_category?: string | null; retryable: boolean; working_directory_id: string; };
 

web/workspace/src/lib/generated/worker-launch-api.ts

@@ -1,6 +1,6 @@
 // @generated by yoi-worker-openapi-typescript@1. Do not edit.
 // Input: openapi/server-api.json
-// Input contract: 0.1.0 (sha256:8cc3c45d082e52b1ea9a014721008aeb4c0699b987729d362c0c14a6f485df10)
+// Input contract: 0.1.0 (sha256:2ac605f7ea70d0c725f9993a5c59c4a8c4266cf69184902db7b3ea7fd5932224)
 // Options: worker-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output
 // Output: web/workspace/src/lib/generated/worker-launch-api.ts
 // Regenerate/check: cargo run -q -p server-api --example generate_worker_launch_api_types -- --check

web/workspace/src/lib/workspace/settings/workdir-removal.ts

@@ -30,7 +30,9 @@ const causes: Readonly<Record<string, string>> = {
   provider_observation_unknown:
     "The provider could not confirm the remaining Workdir. Restore provider access so the next removal can recheck its current contents.",
   provider_cleanup_outcome_unknown:
-    "The provider has not confirmed deletion. Restore provider access and retry the same removal request to recheck the remaining target. The Workdir stays listed until deletion is confirmed.",
+    "The provider has not confirmed deletion. Restore provider access and reconcile the same request. Recovery only observes the target; it never repeats deletion.",
+  request_interrupted:
+    "The request was interrupted before execution. Submit a new request if removal is still wanted.",
   authority_invalid:
     "The removal authority could not be verified. Ask the Workspace administrator to verify the Workdir target and owning Runtime before requesting removal again.",
   provider_identity_changed:
@@ -125,11 +127,39 @@ async function removalBody(response: Response): Promise<string> {
   }
 }
 
+// 要求IDは結果不明時だけ保持する。確定した失敗の次の明示要求は別IDとなる。
+export class WorkdirRemovalRequests {
+  private readonly unresolved = new Map<string, string>();
+
+  async remove(
+    fetchFn: typeof fetch,
+    workspaceId: string,
+    runtimeId: string,
+    workdirId: string,
+  ): Promise<WorkingDirectoryRemovalResponse> {
+    const key = JSON.stringify([workspaceId, runtimeId, workdirId]);
+    const operationId = this.unresolved.get(key) ?? crypto.randomUUID();
+    this.unresolved.set(key, operationId);
+    const result = await removeWorkdir(
+      fetchFn,
+      workspaceId,
+      runtimeId,
+      workdirId,
+      operationId,
+    );
+    if (result.failure_category !== "provider_cleanup_outcome_unknown") {
+      this.unresolved.delete(key);
+    }
+    return result;
+  }
+}
+
 export async function removeWorkdir(
   fetchFn: typeof fetch,
   workspaceId: string,
   runtimeId: string,
   workdirId: string,
+  operationId: string,
 ): Promise<WorkingDirectoryRemovalResponse> {
   const response = await fetchFn(
     workspaceApiPath(
@@ -141,8 +171,10 @@ export async function removeWorkdir(
     {
       method: "DELETE",
       headers: { "content-type": "application/json" },
-      // An ordinary repeated request uses the existing durable Backend removal authority.
-      body: JSON.stringify({ reason: "User requested Workdir deletion" }),
+      body: JSON.stringify({
+        operation_id: operationId,
+        reason: "User requested Workdir deletion",
+      }),
     },
   );
   if (!response.ok) {

web/workspace/src/routes/w/[workspaceId]/settings/runtimes/[runtimeId]/workdirs/PageContent.svelte

@@ -2,7 +2,7 @@
   import { loadJson, workspaceApiPath } from '#lib/workspace/api/http.ts';
   import { parseRuntimeCleanupPlan } from '#lib/workspace/api/runtime-workers.ts';
   import { parseWorkingDirectoryListResponse } from '#lib/workspace/api/workdirs.ts';
-  import { canRemoveWorkdir, removalCause, removalGuard, removeWorkdir, WorkdirRemovalError } from '#lib/workspace/settings/workdir-removal.ts';
+  import { canRemoveWorkdir, removalCause, removalGuard, WorkdirRemovalRequests, WorkdirRemovalError } from '#lib/workspace/settings/workdir-removal.ts';
   import { formatCurrentWorkdirCheckout } from '#lib/workspace/settings/workdir-checkout.ts';
   import type { CleanupWorkdirCandidate, RuntimeCleanupPlanResponse, WorkingDirectorySummary } from '#lib/workspace/sidebar/types.ts';
   import type { PageData as RouteData } from './$types';
@@ -10,6 +10,7 @@
   type PageProps = { data: ResolvedPageData<RouteData> };
 
   let { data }: PageProps = $props();
+  const removalRequests = new WorkdirRemovalRequests();
   let cleanupBusyTarget = $state<string | null>(null);
   let refreshing = $state(false);
   let refreshError = $state<string | null>(null);
@@ -72,7 +73,7 @@
     cleanupBusyTarget = workdir.working_directory_id;
     feedback = null;
     try {
-      const result = await removeWorkdir(fetch, workspaceId, runtimeId, workdir.working_directory_id);
+      const result = await removalRequests.remove(fetch, workspaceId, runtimeId, workdir.working_directory_id);
       if (workspaceId !== data.workspaceId || runtimeId !== data.runtimeId) return;
       feedback = {
         id: workdir.working_directory_id,

web/workspace/tests/workdir-removal.test.ts

@@ -6,6 +6,7 @@ import {
   canRemoveWorkdir,
   removalCause,
   removeWorkdir,
+  WorkdirRemovalRequests,
 } from "../src/lib/workspace/settings/workdir-removal.ts";
 import type {
   CleanupWorkdirCandidate,
@@ -57,6 +58,7 @@ Deno.test("removal wire boundary rejects malformed results and mismatched identi
       "workspace",
       "runtime",
       "dir",
+      "request-1",
     );
   } catch {
     rejected = true;
@@ -72,6 +74,7 @@ Deno.test("removal response size is bounded before decoding provider content", a
       "workspace",
       "runtime",
       "dir",
+      "request-1",
     );
   } catch {
     rejected = true;
@@ -179,7 +182,7 @@ Deno.test("pending unknown cleanup permits ordinary removal while current dirty
   );
 });
 
-Deno.test("ordinary retries send the same scoped DELETE body and expose only allowlisted HTTP causes", async () => {
+Deno.test("new requests after confirmed failure get distinct IDs and expose only allowlisted HTTP causes", async () => {
   const requests: { path: string; body: string; method: string }[] = [];
   const fetchFn = (async (path: RequestInfo | URL, init?: RequestInit) => {
     requests.push({
@@ -194,8 +197,9 @@ Deno.test("ordinary retries send the same scoped DELETE body and expose only all
       failure_category: "mount_present",
     });
   }) as typeof fetch;
+  const removalRequests = new WorkdirRemovalRequests();
   for (let i = 0; i < 2; i++) {
-    const result = await removeWorkdir(
+    const result = await removalRequests.remove(
       fetchFn,
       "workspace/x",
       "runtime/x",
@@ -204,8 +208,9 @@ Deno.test("ordinary retries send the same scoped DELETE body and expose only all
     assert(result.retryable && result.failure_category === "mount_present");
   }
   assert(
-    JSON.stringify(requests[0]) === JSON.stringify(requests[1]),
-    "retry changed ordinary deletion request",
+    JSON.parse(requests[0].body).operation_id !==
+      JSON.parse(requests[1].body).operation_id,
+    "new explicit request reused a failed request identity",
   );
   assert(
     requests[0].path ===
@@ -213,8 +218,9 @@ Deno.test("ordinary retries send the same scoped DELETE body and expose only all
   );
   assert(
     requests[0].method === "DELETE" &&
-      requests[0].body ===
-        JSON.stringify({ reason: "User requested Workdir deletion" }),
+      JSON.parse(requests[0].body).reason ===
+        "User requested Workdir deletion" &&
+      typeof JSON.parse(requests[0].body).operation_id === "string",
   );
   for (
     const [status, category] of [
@@ -236,6 +242,7 @@ Deno.test("ordinary retries send the same scoped DELETE body and expose only all
         "workspace",
         "runtime",
         "dir",
+        "request-1",
       );
     } catch (error) {
       message = error instanceof Error ? error.message : String(error);
@@ -246,3 +253,34 @@ Deno.test("ordinary retries send the same scoped DELETE body and expose only all
     );
   }
 });
+
+Deno.test("unknown removal outcomes keep the same identity and body until a terminal result", async () => {
+  for (const transportLoss of [false, true]) {
+    const calls: string[] = [];
+    const requester = new WorkdirRemovalRequests();
+    const fetchFn = (async (_path: RequestInfo | URL, init?: RequestInit) => {
+      calls.push(String(init?.body));
+      if (calls.length === 1 && transportLoss) {
+        throw new Error("connection lost");
+      }
+      return Response.json({
+        working_directory_id: "dir",
+        disposition: "attention_required",
+        retryable: calls.length !== 1,
+        failure_category: calls.length === 1
+          ? "provider_cleanup_outcome_unknown"
+          : "resource_busy",
+      });
+    }) as typeof fetch;
+    await requester.remove(fetchFn, "workspace", "runtime", "dir").catch(() =>
+      undefined
+    );
+    await requester.remove(fetchFn, "workspace", "runtime", "dir");
+    await requester.remove(fetchFn, "workspace", "runtime", "dir");
+    assert(calls[0] === calls[1], "unknown outcome changed request identity");
+    assert(
+      calls[1] !== calls[2],
+      "confirmed failure kept old request identity",
+    );
+  }
+});