fix: serialize Workdir removal attempts

crates/workspace-server/src/server.rs

@@ -9589,18 +9589,71 @@ fn classify_workdir_provider_error(error: &RuntimeRegistryError) -> (&'static st
     }
 }
 
+trait WorkdirRemovalRuntimeProvider: Send + Sync {
+    fn observe_workdir(
+        &self,
+        runtime_id: &str,
+        working_directory_id: &str,
+    ) -> std::result::Result<crate::hosts::RuntimeWorkingDirectoryResult, RuntimeRegistryError>;
+
+    fn cleanup_workdir(
+        &self,
+        runtime_id: &str,
+        working_directory_id: &str,
+    ) -> std::result::Result<crate::hosts::RuntimeWorkingDirectoryResult, RuntimeRegistryError>;
+}
+
+impl WorkdirRemovalRuntimeProvider for RuntimeRegistry {
+    fn observe_workdir(
+        &self,
+        runtime_id: &str,
+        working_directory_id: &str,
+    ) -> std::result::Result<crate::hosts::RuntimeWorkingDirectoryResult, RuntimeRegistryError>
+    {
+        self.working_directory(runtime_id, working_directory_id)
+    }
+
+    fn cleanup_workdir(
+        &self,
+        runtime_id: &str,
+        working_directory_id: &str,
+    ) -> std::result::Result<crate::hosts::RuntimeWorkingDirectoryResult, RuntimeRegistryError>
+    {
+        self.cleanup_working_directory(runtime_id, working_directory_id)
+    }
+}
+
 fn execute_reserved_workdir_removal(
     api: &WorkspaceApi,
     operation: WorkdirRemovalOperation,
+    recovery: bool,
+) -> Result<WorkdirRemovalOperation> {
+    execute_reserved_workdir_removal_with_provider(api, operation, recovery, api.runtime.as_ref())
+}
+
+fn execute_reserved_workdir_removal_with_provider(
+    api: &WorkspaceApi,
+    operation: WorkdirRemovalOperation,
+    recovery: bool,
+    provider: &dyn WorkdirRemovalRuntimeProvider,
 ) -> Result<WorkdirRemovalOperation> {
     if operation.state == WorkdirRemovalOperationState::Completed {
         return Ok(operation);
     }
-    let operation = api.config_store.begin_workdir_removal_attempt(
+    let operation = if recovery {
+        api.config_store
+            .reclaim_workdir_removal_attempt_for_recovery(
                 &operation.workspace_id,
                 &operation.operation_id,
                 &operation.request_fingerprint,
-    )?;
+            )?
+    } else {
+        api.config_store.begin_workdir_removal_attempt(
+            &operation.workspace_id,
+            &operation.operation_id,
+            &operation.request_fingerprint,
+        )?
+    };
     let guards = match api.config_store.workdir_removal_guards(&operation) {
         Ok(guards) => guards,
         Err(
@@ -9627,10 +9680,8 @@ fn execute_reserved_workdir_removal(
     // Runtime observation is the provider's publication/removal authority. A
     // missing summary without the exact not-found diagnostic is unknown, not a
     // successful delete.
-    let observed = match api
+    let observed = provider.observe_workdir(&operation.runtime_id, &operation.working_directory_id);
-        .runtime
+    let observed = match observed {
-        .working_directory(&operation.runtime_id, &operation.working_directory_id)
-    {
         Ok(observed) => observed,
         Err(error) => {
             let (category, retryable) = classify_workdir_provider_error(&error);
@@ -9659,10 +9710,8 @@ fn execute_reserved_workdir_removal(
         );
     }
 
-    let deleted = match api
+    let deleted = provider.cleanup_workdir(&operation.runtime_id, &operation.working_directory_id);
-        .runtime
+    let deleted = match deleted {
-        .cleanup_working_directory(&operation.runtime_id, &operation.working_directory_id)
-    {
         Ok(deleted) => deleted,
         Err(error) => {
             let (category, retryable) = classify_workdir_provider_error(&error);
@@ -9692,6 +9741,20 @@ fn execute_reserved_workdir_removal(
     api.config_store.commit_workdir_removal_removed(&operation)
 }
 
+fn workdir_removal_execution_lock(
+    api: &WorkspaceApi,
+    working_directory_id: &str,
+) -> Result<Arc<std::sync::Mutex<()>>> {
+    let mut locks = api
+        .workdir_remove_locks
+        .lock()
+        .map_err(|_| Error::Store("Workdir removal lock registry was poisoned".to_string()))?;
+    Ok(locks
+        .entry(working_directory_id.to_string())
+        .or_insert_with(|| Arc::new(std::sync::Mutex::new(())))
+        .clone())
+}
+
 fn execute_workdir_removal(
     api: &WorkspaceApi,
     working_directory_id: &str,
@@ -9704,16 +9767,7 @@ fn execute_workdir_removal(
             "Workdir removal reason must be between 1 and 500 bytes".to_string(),
         ));
     }
-    let lock = {
+    let lock = workdir_removal_execution_lock(api, working_directory_id)?;
-        let mut locks = api
-            .workdir_remove_locks
-            .lock()
-            .map_err(|_| Error::Store("Workdir removal lock registry was poisoned".to_string()))?;
-        locks
-            .entry(working_directory_id.to_string())
-            .or_insert_with(|| Arc::new(std::sync::Mutex::new(())))
-            .clone()
-    };
     let _guard = lock
         .lock()
         .map_err(|_| Error::Store("Workdir removal lock was poisoned".to_string()))?;
@@ -9737,7 +9791,7 @@ fn execute_workdir_removal(
         api.config_store
             .reserve_workdir_removal_operation(&intent)?
     };
-    execute_reserved_workdir_removal(api, operation)
+    execute_reserved_workdir_removal(api, operation, false)
         .map(|operation| workdir_removal_response(&operation))
 }
 
@@ -9746,7 +9800,14 @@ fn recover_workdir_removals(api: &WorkspaceApi) -> Result<()> {
         .config_store
         .recoverable_workdir_removal_operations(api.workspace_id(), 100)?
     {
-        if let Err(error) = execute_reserved_workdir_removal(api, operation.clone()) {
+        let result =
+            workdir_removal_execution_lock(api, &operation.working_directory_id).and_then(|lock| {
+                let _guard = lock
+                    .lock()
+                    .map_err(|_| Error::Store("Workdir removal lock was poisoned".to_string()))?;
+                execute_reserved_workdir_removal(api, operation.clone(), true)
+            });
+        if let Err(error) = result {
             tracing::warn!(
                 workspace_id = %api.workspace_id(),
                 workdir_id = %operation.working_directory_id,
@@ -22287,6 +22348,253 @@ mod tests {
         ));
     }
 
+    struct FakeWorkdirRemovalProvider {
+        observation: Mutex<
+            Option<
+                std::result::Result<
+                    crate::hosts::RuntimeWorkingDirectoryResult,
+                    RuntimeRegistryError,
+                >,
+            >,
+        >,
+        cleanup: Mutex<
+            Option<
+                std::result::Result<
+                    crate::hosts::RuntimeWorkingDirectoryResult,
+                    RuntimeRegistryError,
+                >,
+            >,
+        >,
+        cleanup_calls: std::sync::atomic::AtomicUsize,
+    }
+
+    impl FakeWorkdirRemovalProvider {
+        fn new(
+            observation: crate::hosts::RuntimeWorkingDirectoryResult,
+            cleanup: crate::hosts::RuntimeWorkingDirectoryResult,
+        ) -> Self {
+            Self {
+                observation: Mutex::new(Some(Ok(observation))),
+                cleanup: Mutex::new(Some(Ok(cleanup))),
+                cleanup_calls: std::sync::atomic::AtomicUsize::new(0),
+            }
+        }
+
+        fn cleanup_calls(&self) -> usize {
+            self.cleanup_calls.load(std::sync::atomic::Ordering::SeqCst)
+        }
+    }
+
+    impl WorkdirRemovalRuntimeProvider for FakeWorkdirRemovalProvider {
+        fn observe_workdir(
+            &self,
+            _runtime_id: &str,
+            _working_directory_id: &str,
+        ) -> std::result::Result<crate::hosts::RuntimeWorkingDirectoryResult, RuntimeRegistryError>
+        {
+            self.observation
+                .lock()
+                .unwrap()
+                .take()
+                .expect("one observation fixture")
+        }
+
+        fn cleanup_workdir(
+            &self,
+            _runtime_id: &str,
+            _working_directory_id: &str,
+        ) -> std::result::Result<crate::hosts::RuntimeWorkingDirectoryResult, RuntimeRegistryError>
+        {
+            self.cleanup_calls
+                .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+            std::thread::sleep(std::time::Duration::from_millis(5));
+            self.cleanup
+                .lock()
+                .unwrap()
+                .take()
+                .expect("one cleanup fixture")
+        }
+    }
+
+    fn workdir_removal_result(
+        state: WorkerOperationState,
+        summary: Option<WorkingDirectorySummary>,
+        diagnostics: Vec<RuntimeDiagnostic>,
+    ) -> crate::hosts::RuntimeWorkingDirectoryResult {
+        crate::hosts::RuntimeWorkingDirectoryResult {
+            state,
+            working_directory: summary
+                .map(|summary| worker_runtime::catalog::WorkingDirectoryStatus { summary }),
+            diagnostics,
+        }
+    }
+
+    fn reserve_removal_fixture(
+        api: &WorkspaceApi,
+        working_directory_id: &str,
+    ) -> (WorkdirRemovalOperation, WorkingDirectorySummary) {
+        seed_cleanup_workdir(api, working_directory_id, "present", "clean");
+        let record = api
+            .store
+            .get_workdir_registry(&api.config.workspace_id, working_directory_id)
+            .unwrap()
+            .unwrap();
+        let summary = workdir_summary_from_record(&record);
+        let intent = workdir_removal_intent(
+            &record,
+            "account:owner",
+            &format!("remove {working_directory_id}"),
+        )
+        .unwrap();
+        let operation = api
+            .config_store
+            .reserve_workdir_removal_operation(&intent)
+            .unwrap();
+        (operation, summary)
+    }
+
+    #[tokio::test]
+    async fn injectable_provider_covers_cleanup_not_found_unknown_dirty_and_unsupported() {
+        let workspace = tempfile::tempdir().unwrap();
+        init_clean_git_workspace(workspace.path());
+        let api = test_api(workspace.path()).await;
+
+        let (clean_operation, clean_summary) = reserve_removal_fixture(&api, "clean-provider");
+        let clean_provider = FakeWorkdirRemovalProvider::new(
+            workdir_removal_result(
+                WorkerOperationState::Accepted,
+                Some(clean_summary),
+                Vec::new(),
+            ),
+            workdir_removal_result(WorkerOperationState::Accepted, None, Vec::new()),
+        );
+        let removed = execute_reserved_workdir_removal_with_provider(
+            &api,
+            clean_operation.clone(),
+            false,
+            &clean_provider,
+        )
+        .unwrap();
+        assert_eq!(
+            removed.disposition,
+            Some(WorkdirRemovalDisposition::Removed)
+        );
+        assert_eq!(clean_provider.cleanup_calls(), 1);
+        let replay = execute_reserved_workdir_removal_with_provider(
+            &api,
+            removed.clone(),
+            false,
+            &clean_provider,
+        )
+        .unwrap();
+        assert_eq!(replay, removed);
+        assert_eq!(clean_provider.cleanup_calls(), 1);
+
+        let (missing_operation, _) = reserve_removal_fixture(&api, "provider-not-found");
+        let not_found = RuntimeDiagnostic {
+            code: "working_directory_not_found".to_string(),
+            severity: DiagnosticSeverity::Error,
+            message: "missing".to_string(),
+        };
+        let missing_provider = FakeWorkdirRemovalProvider::new(
+            workdir_removal_result(
+                WorkerOperationState::Rejected,
+                None,
+                vec![not_found.clone()],
+            ),
+            workdir_removal_result(WorkerOperationState::Rejected, None, vec![not_found]),
+        );
+        let removed = execute_reserved_workdir_removal_with_provider(
+            &api,
+            missing_operation,
+            false,
+            &missing_provider,
+        )
+        .unwrap();
+        assert_eq!(
+            removed.disposition,
+            Some(WorkdirRemovalDisposition::Removed)
+        );
+        assert_eq!(missing_provider.cleanup_calls(), 0);
+
+        let (unknown_operation, _) = reserve_removal_fixture(&api, "provider-unknown");
+        let unknown_provider = FakeWorkdirRemovalProvider::new(
+            workdir_removal_result(
+                WorkerOperationState::Accepted,
+                None,
+                vec![RuntimeDiagnostic {
+                    code: "working_directory_provider_timeout".to_string(),
+                    severity: DiagnosticSeverity::Error,
+                    message: "timeout".to_string(),
+                }],
+            ),
+            workdir_removal_result(WorkerOperationState::Accepted, None, Vec::new()),
+        );
+        let unknown = execute_reserved_workdir_removal_with_provider(
+            &api,
+            unknown_operation,
+            false,
+            &unknown_provider,
+        )
+        .unwrap();
+        assert_eq!(unknown.state, WorkdirRemovalOperationState::Failed);
+        assert!(unknown.retryable);
+        assert_eq!(unknown_provider.cleanup_calls(), 0);
+
+        let (dirty_operation, mut dirty_summary) = reserve_removal_fixture(&api, "provider-dirty");
+        dirty_summary.cleanliness = Some("dirty".to_string());
+        let dirty_provider = FakeWorkdirRemovalProvider::new(
+            workdir_removal_result(
+                WorkerOperationState::Accepted,
+                Some(dirty_summary),
+                Vec::new(),
+            ),
+            workdir_removal_result(WorkerOperationState::Accepted, None, Vec::new()),
+        );
+        let dirty = execute_reserved_workdir_removal_with_provider(
+            &api,
+            dirty_operation,
+            false,
+            &dirty_provider,
+        )
+        .unwrap();
+        assert_eq!(dirty.disposition, Some(WorkdirRemovalDisposition::Retained));
+        assert_eq!(dirty_provider.cleanup_calls(), 0);
+
+        let (unsupported_operation, unsupported_summary) =
+            reserve_removal_fixture(&api, "provider-unsupported");
+        let unsupported_provider = FakeWorkdirRemovalProvider::new(
+            workdir_removal_result(
+                WorkerOperationState::Accepted,
+                Some(unsupported_summary),
+                Vec::new(),
+            ),
+            workdir_removal_result(
+                WorkerOperationState::Unsupported,
+                None,
+                vec![RuntimeDiagnostic {
+                    code: "working_directory_unsupported".to_string(),
+                    severity: DiagnosticSeverity::Error,
+                    message: "unsupported".to_string(),
+                }],
+            ),
+        );
+        let unsupported = execute_reserved_workdir_removal_with_provider(
+            &api,
+            unsupported_operation,
+            false,
+            &unsupported_provider,
+        )
+        .unwrap();
+        assert_eq!(unsupported.state, WorkdirRemovalOperationState::Failed);
+        assert!(!unsupported.retryable);
+        assert_eq!(
+            unsupported.failure_category.as_deref(),
+            Some("unsupported_target")
+        );
+        assert_eq!(unsupported_provider.cleanup_calls(), 1);
+    }
+
     #[test]
     fn only_exact_provider_not_found_is_removal_evidence() {
         let not_found = crate::hosts::RuntimeWorkingDirectoryResult {
@@ -22408,9 +22716,19 @@ mod tests {
             .unwrap();
         let intent =
             workdir_removal_intent(&record, "account:owner", "recover provider cleanup").unwrap();
-        api.config_store
+        let reserved = api
+            .config_store
             .reserve_workdir_removal_operation(&intent)
             .unwrap();
+        let interrupted_attempt = api
+            .config_store
+            .begin_workdir_removal_attempt(
+                &reserved.workspace_id,
+                &reserved.operation_id,
+                &reserved.request_fingerprint,
+            )
+            .unwrap();
+        assert_eq!(interrupted_attempt.attempt_count, 1);
 
         recover_workdir_removals(&api).unwrap();
 
@@ -22420,6 +22738,7 @@ mod tests {
             .unwrap()
             .unwrap();
         assert_eq!(operation.state, WorkdirRemovalOperationState::Failed);
+        assert_eq!(operation.attempt_count, 2);
         assert!(operation.retryable);
         assert_eq!(
             operation.failure_category.as_deref(),

crates/workspace-server/src/workdir_removal.rs

@@ -108,6 +108,9 @@ CREATE TABLE workdir_removal_operations (
     PRIMARY KEY (workspace_id, operation_id),
     FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE
 );
+CREATE UNIQUE INDEX idx_workdir_removal_operations_one_pending
+    ON workdir_removal_operations(workspace_id, workdir_id)
+    WHERE state = 'pending';
 CREATE INDEX idx_workdir_removal_operations_recovery
     ON workdir_removal_operations(workspace_id, state, retryable, updated_at);
 CREATE INDEX idx_workdir_removal_operations_workdir
@@ -188,6 +191,18 @@ impl SqliteWorkspaceStore {
                 tx.commit()?;
                 return Ok(existing);
             }
+            let pending_operation: Option<String> = tx
+                .query_row(
+                    "SELECT operation_id FROM workdir_removal_operations WHERE workspace_id=?1 AND workdir_id=?2 AND state='pending' LIMIT 1",
+                    params![intent.workspace_id, intent.working_directory_id],
+                    |row| row.get(0),
+                )
+                .optional()?;
+            if let Some(pending_operation) = pending_operation {
+                return Err(Error::WorkdirAttachmentConflict(format!(
+                    "Workdir removal operation `{pending_operation}` is already pending"
+                )));
+            }
             let current = load_workdir_record(&tx, &intent.workspace_id, &intent.working_directory_id)?
                 .ok_or_else(|| Error::InvalidInput(format!(
                     "Unknown Workdir `{}`",
@@ -237,10 +252,40 @@ impl SqliteWorkspaceStore {
         workspace_id: &str,
         operation_id: &str,
         request_fingerprint: &str,
+    ) -> Result<WorkdirRemovalOperation> {
+        self.begin_workdir_removal_attempt_inner(
+            workspace_id,
+            operation_id,
+            request_fingerprint,
+            false,
+        )
+    }
+
+    pub fn reclaim_workdir_removal_attempt_for_recovery(
+        &self,
+        workspace_id: &str,
+        operation_id: &str,
+        request_fingerprint: &str,
+    ) -> Result<WorkdirRemovalOperation> {
+        self.begin_workdir_removal_attempt_inner(
+            workspace_id,
+            operation_id,
+            request_fingerprint,
+            true,
+        )
+    }
+
+    fn begin_workdir_removal_attempt_inner(
+        &self,
+        workspace_id: &str,
+        operation_id: &str,
+        request_fingerprint: &str,
+        recovery: bool,
     ) -> Result<WorkdirRemovalOperation> {
         self.with_conn_mut(|conn| {
             let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
-            let operation = require_operation(&tx, workspace_id, operation_id, request_fingerprint)?;
+            let operation =
+                require_operation(&tx, workspace_id, operation_id, request_fingerprint)?;
             if operation.state == WorkdirRemovalOperationState::Completed {
                 tx.commit()?;
                 return Ok(operation);
@@ -250,12 +295,21 @@ impl SqliteWorkspaceStore {
                     "Workdir removal operation `{operation_id}` is not retryable"
                 )));
             }
+            if operation.state == WorkdirRemovalOperationState::Pending
+                && operation.attempt_count > 0
+                && !recovery
+            {
+                return Err(Error::WorkdirAttachmentConflict(format!(
+                    "Workdir removal operation `{operation_id}` already has an active attempt"
+                )));
+            }
             let now = Utc::now().to_rfc3339();
             tx.execute(
                 "UPDATE workdir_removal_operations SET state='pending', attempt_count=attempt_count+1, retryable=1, failure_category=NULL, disposition=NULL, updated_at=?1, completed_at=NULL WHERE workspace_id=?2 AND operation_id=?3 AND request_fingerprint=?4",
                 params![now, workspace_id, operation_id, request_fingerprint],
             )?;
-            let operation = require_operation(&tx, workspace_id, operation_id, request_fingerprint)?;
+            let operation =
+                require_operation(&tx, workspace_id, operation_id, request_fingerprint)?;
             tx.commit()?;
             Ok(operation)
         })
@@ -932,6 +986,59 @@ mod tests {
         assert_eq!(replay, completed);
     }
 
+    #[tokio::test]
+    async fn concurrent_claims_invoke_the_simulated_provider_once() {
+        use std::sync::atomic::{AtomicUsize, Ordering};
+        use std::sync::{Arc, Barrier};
+
+        let (store, workdir) = seeded_store().await;
+        let intent =
+            workdir_removal_intent(&workdir, "workspace-api", "remove clean Workdir").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));
+        let mut callers = Vec::new();
+        for _ in 0..2 {
+            let store = store.clone();
+            let operation = operation.clone();
+            let barrier = barrier.clone();
+            let provider_calls = provider_calls.clone();
+            callers.push(std::thread::spawn(move || {
+                barrier.wait();
+                let claim = store.begin_workdir_removal_attempt(
+                    &operation.workspace_id,
+                    &operation.operation_id,
+                    &operation.request_fingerprint,
+                );
+                if claim.is_ok() {
+                    provider_calls.fetch_add(1, Ordering::SeqCst);
+                }
+                claim
+            }));
+        }
+        barrier.wait();
+        let results = callers
+            .into_iter()
+            .map(|caller| caller.join().unwrap())
+            .collect::<Vec<_>>();
+        assert_eq!(results.iter().filter(|result| result.is_ok()).count(), 1);
+        assert_eq!(
+            results
+                .iter()
+                .filter(|result| matches!(result, Err(Error::WorkdirAttachmentConflict(_))))
+                .count(),
+            1
+        );
+        assert_eq!(provider_calls.load(Ordering::SeqCst), 1);
+
+        let claimed = results.into_iter().find_map(|result| result.ok()).unwrap();
+        let completed = store.commit_workdir_removal_removed(&claimed).unwrap();
+        assert_eq!(
+            completed.disposition,
+            Some(WorkdirRemovalDisposition::Removed)
+        );
+    }
+
     #[tokio::test]
     async fn pending_removal_fences_new_attachment_and_retry_rereads_live_reservation() {
         let (store, workdir) = seeded_store().await;

docs/design/durable-operations.md

@@ -13,7 +13,7 @@ The record stores only facts that affect identity, authorization, replay, or the
 - explicit retryability, bounded failure category, and bounded disposition;
 - a factual checkpoint only when a non-idempotent provider effect cannot be safely re-observed or repeated.
 
-A fingerprint excludes Server-generated result identifiers, attempt data, diagnostics, and fresh observations. Reusing one operation identity with a different fingerprint is an error. A completed exact retry replays the committed bounded result.
+A fingerprint excludes Server-generated result identifiers, attempt data, diagnostics, and fresh observations. Reusing one operation identity with a different fingerprint is an error. A completed exact retry replays the committed bounded result. A durable one-pending-operation constraint plus an atomic attempt claim prevents concurrent callers from entering the provider side effect for the same Workdir; the in-process resource lock is an additional serialization layer, not the sole authority.
 
 `pending` means only that the intent remains open. Function names, validation steps, and provider-call positions are not persisted as lifecycle stages. `failed` records the latest terminal attempt outcome; retryability remains separate metadata. `completed` means the required domain result and disposition are durably committed.