fix: guard Workdir removal recovery ownership

crates/workspace-server/src/server.rs

@@ -142,8 +142,8 @@ use crate::store::{
     WorkspaceResourceKind,
 };
 use crate::workdir_removal::{
-    WorkdirRemovalDisposition, WorkdirRemovalOperation, WorkdirRemovalOperationState,
-    workdir_removal_intent,
+    WorkdirRemovalAttemptOwner, WorkdirRemovalDisposition, WorkdirRemovalOperation,
+    WorkdirRemovalOperationState, workdir_removal_intent,
 };
 use crate::workspace_catalog::{WorkspaceCatalogService, WorkspaceCreateRequest};
 use crate::{Error, Result};
@@ -519,6 +519,7 @@ pub struct WorkspaceApi {
     workdir_session_locks: Arc<Mutex<HashMap<RuntimeWorkerRef, Arc<tokio::sync::Mutex<()>>>>>,
     worker_remove_locks: Arc<Mutex<HashMap<RuntimeWorkerRef, Arc<tokio::sync::Mutex<()>>>>>,
     workdir_remove_locks: Arc<Mutex<HashMap<String, Arc<std::sync::Mutex<()>>>>>,
+    workdir_remove_attempt_owner: WorkdirRemovalAttemptOwner,
     worker_control_locks: Arc<Mutex<HashMap<String, Arc<tokio::sync::Mutex<()>>>>>,
 }
 
@@ -1640,6 +1641,7 @@ impl WorkspaceApi {
             workdir_session_locks: Arc::new(Mutex::new(HashMap::new())),
             worker_remove_locks: Arc::new(Mutex::new(HashMap::new())),
             workdir_remove_locks: Arc::new(Mutex::new(HashMap::new())),
+            workdir_remove_attempt_owner: current_workdir_removal_attempt_owner()?,
             worker_control_locks: Arc::new(Mutex::new(HashMap::new())),
         };
         if let Some(dispatcher) = worker_remove_dispatcher {
@@ -9320,6 +9322,16 @@ async fn create_workspace_working_directory(
             )
         });
     }
+    let reserved = if reserved.state == "failed" {
+        api.config_store.begin_failed_workdir_create_retry(
+            workspace_id,
+            &operation_id,
+            &request_fingerprint,
+            &now_registry_timestamp(),
+        )?
+    } else {
+        reserved
+    };
 
     let runtime = match api
         .runtime
@@ -9589,6 +9601,91 @@ fn classify_workdir_provider_error(error: &RuntimeRegistryError) -> (&'static st
     }
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum WorkdirRemovalOwnerObservation {
+    Running { process_start_marker: u64 },
+    Missing,
+    Unobservable,
+}
+
+fn current_workdir_removal_attempt_owner() -> Result<WorkdirRemovalAttemptOwner> {
+    let process_id = std::process::id();
+    match observe_workdir_removal_owner(process_id) {
+        WorkdirRemovalOwnerObservation::Running {
+            process_start_marker,
+        } => Ok(WorkdirRemovalAttemptOwner {
+            process_id,
+            process_start_marker,
+        }),
+        WorkdirRemovalOwnerObservation::Missing | WorkdirRemovalOwnerObservation::Unobservable => {
+            Err(Error::Config(
+                "current Server process identity is unavailable for durable Workdir removal"
+                    .to_string(),
+            ))
+        }
+    }
+}
+
+fn workdir_removal_attempt_is_orphaned(operation: &WorkdirRemovalOperation) -> Result<bool> {
+    let Some(owner) = operation.attempt_owner else {
+        return Ok(operation.attempt_count == 0);
+    };
+    match observe_workdir_removal_owner(owner.process_id) {
+        WorkdirRemovalOwnerObservation::Running {
+            process_start_marker,
+        } if process_start_marker == owner.process_start_marker => Ok(false),
+        WorkdirRemovalOwnerObservation::Running { .. }
+        | WorkdirRemovalOwnerObservation::Missing => Ok(true),
+        WorkdirRemovalOwnerObservation::Unobservable => Err(Error::RegistryInconsistency(
+            "prior Workdir removal attempt owner liveness is unobservable".to_string(),
+        )),
+    }
+}
+
+#[cfg(target_os = "linux")]
+fn observe_workdir_removal_owner(process_id: u32) -> WorkdirRemovalOwnerObservation {
+    let stat = match std::fs::read_to_string(format!("/proc/{process_id}/stat")) {
+        Ok(stat) => stat,
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
+            return if process_id != std::process::id()
+                && std::fs::read_to_string("/proc/self/stat")
+                    .ok()
+                    .and_then(|stat| parse_linux_process_start_marker(&stat))
+                    .is_some()
+            {
+                WorkdirRemovalOwnerObservation::Missing
+            } else {
+                WorkdirRemovalOwnerObservation::Unobservable
+            };
+        }
+        Err(_) => return WorkdirRemovalOwnerObservation::Unobservable,
+    };
+    parse_linux_process_start_marker(&stat)
+        .map(
+            |process_start_marker| WorkdirRemovalOwnerObservation::Running {
+                process_start_marker,
+            },
+        )
+        .unwrap_or(WorkdirRemovalOwnerObservation::Unobservable)
+}
+
+#[cfg(target_os = "linux")]
+fn parse_linux_process_start_marker(stat: &str) -> Option<u64> {
+    let (_, tail) = stat.rsplit_once(") ")?;
+    tail.split_whitespace().nth(19)?.parse().ok()
+}
+
+#[cfg(not(target_os = "linux"))]
+fn observe_workdir_removal_owner(process_id: u32) -> WorkdirRemovalOwnerObservation {
+    if process_id == std::process::id() {
+        WorkdirRemovalOwnerObservation::Running {
+            process_start_marker: 0,
+        }
+    } else {
+        WorkdirRemovalOwnerObservation::Unobservable
+    }
+}
+
 trait WorkdirRemovalRuntimeProvider: Send + Sync {
     fn observe_workdir(
         &self,
@@ -9641,17 +9738,25 @@ fn execute_reserved_workdir_removal_with_provider(
         return Ok(operation);
     }
     let operation = if recovery {
+        let prior_owner_is_orphaned = if operation.state == WorkdirRemovalOperationState::Pending {
+            workdir_removal_attempt_is_orphaned(&operation)?
+        } else {
+            false
+        };
         api.config_store
             .reclaim_workdir_removal_attempt_for_recovery(
                 &operation.workspace_id,
                 &operation.operation_id,
                 &operation.request_fingerprint,
+                api.workdir_remove_attempt_owner,
+                prior_owner_is_orphaned,
             )?
     } else {
         api.config_store.begin_workdir_removal_attempt(
             &operation.workspace_id,
             &operation.operation_id,
             &operation.request_fingerprint,
+            api.workdir_remove_attempt_owner,
         )?
     };
     let guards = match api.config_store.workdir_removal_guards(&operation) {
@@ -22703,6 +22808,51 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn recovery_does_not_reclaim_live_attempt_owner() {
+        let workspace = tempfile::tempdir().unwrap();
+        init_clean_git_workspace(workspace.path());
+        let api = test_api(workspace.path()).await;
+        seed_cleanup_workdir(&api, "live-owner-workdir", "present", "clean");
+        let record = api
+            .store
+            .get_workdir_registry(&api.config.workspace_id, "live-owner-workdir")
+            .unwrap()
+            .unwrap();
+        let intent = workdir_removal_intent(
+            &record,
+            "account:owner",
+            "do not steal a live provider call",
+        )
+        .unwrap();
+        let reserved = api
+            .config_store
+            .reserve_workdir_removal_operation(&intent)
+            .unwrap();
+        api.config_store
+            .begin_workdir_removal_attempt(
+                &reserved.workspace_id,
+                &reserved.operation_id,
+                &reserved.request_fingerprint,
+                api.workdir_remove_attempt_owner,
+            )
+            .unwrap();
+
+        recover_workdir_removals(&api).unwrap();
+
+        let operation = api
+            .config_store
+            .get_workdir_removal_operation(&api.config.workspace_id, &intent.operation_id)
+            .unwrap()
+            .unwrap();
+        assert_eq!(operation.state, WorkdirRemovalOperationState::Pending);
+        assert_eq!(operation.attempt_count, 1);
+        assert_eq!(
+            operation.attempt_owner,
+            Some(api.workdir_remove_attempt_owner)
+        );
+    }
+
     #[tokio::test]
     async fn recovery_retries_same_operation_and_retains_unknown_provider_result() {
         let workspace = tempfile::tempdir().unwrap();
@@ -22726,6 +22876,10 @@ mod tests {
                 &reserved.workspace_id,
                 &reserved.operation_id,
                 &reserved.request_fingerprint,
+                WorkdirRemovalAttemptOwner {
+                    process_id: u32::MAX,
+                    process_start_marker: 1,
+                },
             )
             .unwrap();
         assert_eq!(interrupted_attempt.attempt_count, 1);

crates/workspace-server/src/store.rs

@@ -10104,18 +10104,55 @@ mod tests {
     }
 
     #[test]
-    fn schema_v49_upgrades_v48_with_durable_workdir_removal_authority() {
-        let conn = Connection::open_in_memory().unwrap();
+    fn schema_v49_upgrades_persisted_v48_workdir_fixture() {
+        let temp = tempfile::tempdir().unwrap();
+        let path = temp.path().join("server-v48.db");
+        {
+            let conn = Connection::open(&path).unwrap();
             configure_sqlite(&conn).unwrap();
             apply_migrations_through(&conn, 48).unwrap();
             assert_eq!(current_schema_version(&conn).unwrap(), 48);
             assert!(!table_exists(&conn, "workdir_removal_operations").unwrap());
+            conn.execute_batch(
+                r#"
+                INSERT INTO accounts (
+                    account_id, kind, handle, display_name, created_at, updated_at
+                ) VALUES ('owner-account', 'user', 'owner', 'Owner', '1', '1');
+                INSERT INTO workspaces (
+                    workspace_id, owner_account_id, display_name, state, created_at, updated_at
+                ) VALUES ('workspace-a', 'owner-account', 'Workspace A', 'active', '1', '1');
+                INSERT INTO repositories (
+                    workspace_id, repository_id, name, kind, provider, uri,
+                    source_kind, source_uri, default_ref, source_revision,
+                    source_fingerprint, observed_status, observed_at, created_at, updated_at
+                ) VALUES (
+                    'workspace-a', 'repository-a', 'Repository A', 'git', 'local', '/repo-a',
+                    'local_path', '/repo-a', 'develop', 1,
+                    'sha256:source-a', 'unverified', NULL, '1', '1'
+                );
+                INSERT INTO workdir_registry (
+                    workspace_id, workdir_id, runtime_id, repository_id,
+                    creation_selector, creation_ref, creation_tree,
+                    current_selector, current_ref, current_tree,
+                    observed_at_epoch_seconds, materialization_status, cleanliness,
+                    created_at, updated_at
+                ) VALUES (
+                    'workspace-a', 'workdir-a', 'runtime-a', 'repository-a',
+                    'refs/heads/develop', 'abc', 'tree-a',
+                    'refs/heads/work', 'def', 'tree-b',
+                    1, 'present', 'clean', '1', '1'
+                );
+                "#,
+            )
+            .unwrap();
+        }
 
-        apply_migrations(&conn).unwrap();
-
-        assert_eq!(current_schema_version(&conn).unwrap(), 49);
-        assert!(table_exists(&conn, "workdir_removal_operations").unwrap());
-        let columns = table_columns(&conn, "workdir_removal_operations").unwrap();
+        let store = SqliteWorkspaceStore::open(&path).unwrap();
+        store
+            .with_conn(|conn| {
+                assert_eq!(current_schema_version(conn)?, 49);
+                assert!(table_exists(conn, "workdir_removal_operations")?);
+                let columns = table_columns(conn, "workdir_removal_operations")?;
                 for required in [
                     "workspace_id",
                     "operation_id",
@@ -10131,6 +10168,8 @@ mod tests {
                     "retryable",
                     "disposition",
                     "failure_category",
+                    "attempt_owner_pid",
+                    "attempt_owner_start_marker",
                     "created_at",
                     "updated_at",
                     "completed_at",
@@ -10140,12 +10179,42 @@ mod tests {
                         "missing {required}"
                     );
                 }
-        let foreign_key_failures: i64 = conn
-            .query_row("SELECT count(*) FROM pragma_foreign_key_check", [], |row| {
-                row.get(0)
+                let preserved: (String, String, String) = conn.query_row(
+                    "SELECT workspace_id, repository_id, materialization_status FROM workdir_registry WHERE workdir_id='workdir-a'",
+                    [],
+                    |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
+                )?;
+                assert_eq!(
+                    preserved,
+                    (
+                        "workspace-a".to_string(),
+                        "repository-a".to_string(),
+                        "present".to_string(),
+                    )
+                );
+                let foreign_key_failures: i64 = conn.query_row(
+                    "SELECT count(*) FROM pragma_foreign_key_check",
+                    [],
+                    |row| row.get(0),
+                )?;
+                assert_eq!(foreign_key_failures, 0);
+                Ok(())
             })
             .unwrap();
-        assert_eq!(foreign_key_failures, 0);
+
+        let workdir = store
+            .get_workdir_registry("workspace-a", "workdir-a")
+            .unwrap()
+            .unwrap();
+        let intent = crate::workdir_removal::workdir_removal_intent(
+            &workdir,
+            "migration-test",
+            "remove migrated Workdir",
+        )
+        .unwrap();
+        let operation = store.reserve_workdir_removal_operation(&intent).unwrap();
+        assert_eq!(operation.workspace_id, "workspace-a");
+        assert_eq!(operation.working_directory_id, "workdir-a");
     }
 
     #[test]

crates/workspace-server/src/workdir_create_operations.rs

@@ -1,4 +1,4 @@
-use rusqlite::{OptionalExtension, params};
+use rusqlite::{OptionalExtension, TransactionBehavior, params};
 use sha2::{Digest, Sha256};
 
 use crate::store::WorkdirCreateOperationRecord;
@@ -103,6 +103,65 @@ impl SqliteWorkspaceStore {
         })
     }
 
+    pub fn begin_failed_workdir_create_retry(
+        &self,
+        workspace_id: &str,
+        operation_id: &str,
+        request_fingerprint: &str,
+        updated_at: &str,
+    ) -> Result<WorkdirCreateOperationRecord> {
+        self.with_conn_mut(|conn| {
+            let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
+            let operation = read_workdir_create_operation(&tx, workspace_id, operation_id)?
+                .ok_or_else(|| {
+                    Error::RegistryInconsistency(format!(
+                        "Workdir create operation `{operation_id}` disappeared before retry"
+                    ))
+                })?;
+            if operation.request_fingerprint != request_fingerprint {
+                return Err(Error::InvalidInput(format!(
+                    "Workdir create operation `{operation_id}` was reused with different input"
+                )));
+            }
+            if operation.state != "failed" {
+                return Err(Error::WorkdirAttachmentConflict(format!(
+                    "Workdir create operation `{operation_id}` is not a failed retry"
+                )));
+            }
+            let removal_pending: bool = tx.query_row(
+                "SELECT EXISTS(SELECT 1 FROM workdir_removal_operations WHERE workspace_id=?1 AND workdir_id=?2 AND state='pending')",
+                params![workspace_id, operation.working_directory_id],
+                |row| row.get(0),
+            )?;
+            if removal_pending {
+                return Err(Error::WorkdirAttachmentConflict(format!(
+                    "Workdir {} has a pending durable removal operation",
+                    operation.working_directory_id
+                )));
+            }
+            let changed = tx.execute(
+                r#"UPDATE workdir_create_operations
+                   SET state='pending', failure=NULL, updated_at=?1
+                   WHERE workspace_id=?2 AND operation_id=?3
+                     AND request_fingerprint=?4 AND state='failed'"#,
+                params![updated_at, workspace_id, operation_id, request_fingerprint],
+            )?;
+            if changed != 1 {
+                return Err(Error::WorkdirAttachmentConflict(format!(
+                    "Workdir create operation `{operation_id}` retry was claimed concurrently"
+                )));
+            }
+            let updated = read_workdir_create_operation(&tx, workspace_id, operation_id)?
+                .ok_or_else(|| {
+                    Error::RegistryInconsistency(format!(
+                        "Workdir create operation `{operation_id}` disappeared after retry claim"
+                    ))
+                })?;
+            tx.commit()?;
+            Ok(updated)
+        })
+    }
+
     pub fn bind_workdir_create_repository_access(
         &self,
         workspace_id: &str,
@@ -406,11 +465,32 @@ mod tests {
             .unwrap();
         assert_eq!(replayed, bound);
         assert_eq!(replayed.source_uri.as_deref(), Some("/tmp/repo"));
+        let failed = store
+            .finish_workdir_create_operation(
+                "workspace",
+                "call-1",
+                &record.request_fingerprint,
+                false,
+                Some("provider failed"),
+                "2026-08-24T00:00:03Z",
+            )
+            .unwrap();
+        assert_eq!(failed.state, "failed");
+        let retry = store
+            .begin_failed_workdir_create_retry(
+                "workspace",
+                "call-1",
+                &record.request_fingerprint,
+                "2026-08-24T00:00:04Z",
+            )
+            .unwrap();
+        assert_eq!(retry.state, "pending");
+        assert_eq!(retry.failure, None);
         assert_eq!(
             store
                 .load_workdir_create_operation("workspace", "call-1")
                 .unwrap(),
-            Some(bound.clone())
+            Some(retry.clone())
         );
         let mut changed_input = record.clone();
         changed_input.request_fingerprint =

crates/workspace-server/src/workdir_removal.rs

@@ -44,6 +44,12 @@ impl WorkdirRemovalDisposition {
     }
 }
 
+#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
+pub struct WorkdirRemovalAttemptOwner {
+    pub process_id: u32,
+    pub process_start_marker: u64,
+}
+
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct WorkdirRemovalIntent {
     pub operation_id: String,
@@ -73,6 +79,7 @@ pub struct WorkdirRemovalOperation {
     pub retryable: bool,
     pub disposition: Option<WorkdirRemovalDisposition>,
     pub failure_category: Option<String>,
+    pub attempt_owner: Option<WorkdirRemovalAttemptOwner>,
     pub created_at: String,
     pub updated_at: String,
     pub completed_at: Option<String>,
@@ -102,6 +109,8 @@ CREATE TABLE workdir_removal_operations (
     retryable INTEGER NOT NULL CHECK (retryable IN (0, 1)),
     disposition TEXT CHECK (disposition IN ('removed', 'retained', 'attention_required')),
     failure_category TEXT,
+    attempt_owner_pid INTEGER CHECK (attempt_owner_pid > 0),
+    attempt_owner_start_marker INTEGER CHECK (attempt_owner_start_marker >= 0),
     created_at TEXT NOT NULL,
     updated_at TEXT NOT NULL,
     completed_at TEXT,
@@ -252,11 +261,13 @@ impl SqliteWorkspaceStore {
         workspace_id: &str,
         operation_id: &str,
         request_fingerprint: &str,
+        owner: WorkdirRemovalAttemptOwner,
     ) -> Result<WorkdirRemovalOperation> {
         self.begin_workdir_removal_attempt_inner(
             workspace_id,
             operation_id,
             request_fingerprint,
+            owner,
             false,
         )
     }
@@ -266,12 +277,15 @@ impl SqliteWorkspaceStore {
         workspace_id: &str,
         operation_id: &str,
         request_fingerprint: &str,
+        owner: WorkdirRemovalAttemptOwner,
+        prior_owner_is_orphaned: bool,
     ) -> Result<WorkdirRemovalOperation> {
         self.begin_workdir_removal_attempt_inner(
             workspace_id,
             operation_id,
             request_fingerprint,
-            true,
+            owner,
+            prior_owner_is_orphaned,
         )
     }
 
@@ -280,7 +294,8 @@ impl SqliteWorkspaceStore {
         workspace_id: &str,
         operation_id: &str,
         request_fingerprint: &str,
-        recovery: bool,
+        owner: WorkdirRemovalAttemptOwner,
+        prior_owner_is_orphaned: bool,
     ) -> Result<WorkdirRemovalOperation> {
         self.with_conn_mut(|conn| {
             let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
@@ -297,7 +312,7 @@ impl SqliteWorkspaceStore {
             }
             if operation.state == WorkdirRemovalOperationState::Pending
                 && operation.attempt_count > 0
-                && !recovery
+                && !prior_owner_is_orphaned
             {
                 return Err(Error::WorkdirAttachmentConflict(format!(
                     "Workdir removal operation `{operation_id}` already has an active attempt"
@@ -305,8 +320,19 @@ impl SqliteWorkspaceStore {
             }
             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],
+                "UPDATE workdir_removal_operations SET state='pending', attempt_count=attempt_count+1, retryable=1, failure_category=NULL, disposition=NULL, attempt_owner_pid=?1, attempt_owner_start_marker=?2, updated_at=?3, completed_at=NULL WHERE workspace_id=?4 AND operation_id=?5 AND request_fingerprint=?6",
+                params![
+                    owner.process_id,
+                    i64::try_from(owner.process_start_marker).map_err(|_| {
+                        Error::InvalidInput(
+                            "process start marker is out of SQLite range".to_string(),
+                        )
+                    })?,
+                    now,
+                    workspace_id,
+                    operation_id,
+                    request_fingerprint,
+                ],
             )?;
             let operation =
                 require_operation(&tx, workspace_id, operation_id, request_fingerprint)?;
@@ -456,7 +482,7 @@ impl SqliteWorkspaceStore {
             }
             let now = Utc::now().to_rfc3339();
             tx.execute(
-                "UPDATE workdir_removal_operations SET state='failed', retryable=?1, disposition=?2, failure_category=?3, updated_at=?4 WHERE workspace_id=?5 AND operation_id=?6 AND request_fingerprint=?7",
+                "UPDATE workdir_removal_operations SET state='failed', retryable=?1, disposition=?2, failure_category=?3, attempt_owner_pid=NULL, attempt_owner_start_marker=NULL, updated_at=?4 WHERE workspace_id=?5 AND operation_id=?6 AND request_fingerprint=?7",
                 params![
                     retryable,
                     WorkdirRemovalDisposition::AttentionRequired.as_str(),
@@ -538,7 +564,7 @@ impl SqliteWorkspaceStore {
             }
             let now = Utc::now().to_rfc3339();
             tx.execute(
-                "UPDATE workdir_removal_operations SET state='completed', retryable=?1, disposition=?2, failure_category=?3, updated_at=?4, completed_at=?4 WHERE workspace_id=?5 AND operation_id=?6 AND request_fingerprint=?7",
+                "UPDATE workdir_removal_operations SET state='completed', retryable=?1, disposition=?2, failure_category=?3, attempt_owner_pid=NULL, attempt_owner_start_marker=NULL, updated_at=?4, completed_at=?4 WHERE workspace_id=?5 AND operation_id=?6 AND request_fingerprint=?7",
                 params![
                     retryable,
                     disposition.as_str(),
@@ -620,13 +646,16 @@ fn require_no_removal_blockers(
             UNION ALL
             SELECT 1 FROM worker_workdir_attachment_reservations
              WHERE workspace_id=?1 AND workdir_id=?2
+            UNION ALL
+            SELECT 1 FROM workdir_create_operations
+             WHERE workspace_id=?1 AND working_directory_id=?2 AND state='pending'
         )"#,
         params![workspace_id, workdir_id],
         |row| row.get(0),
     )?;
     if blocked {
         return Err(Error::WorkdirAttachmentConflict(format!(
-            "Workdir {workdir_id} acquired an active or pending attachment during removal"
+            "Workdir {workdir_id} acquired active attachment or materialization authority during removal"
         )));
     }
     Ok(())
@@ -784,6 +813,7 @@ fn operation_select_sql() -> &'static str {
     r#"SELECT operation_id, request_fingerprint, workspace_id, workdir_id, runtime_id,
               repository_id, materialization_fingerprint, source_actor, reason, state,
               attempt_count, retryable, disposition, failure_category,
+              attempt_owner_pid, attempt_owner_start_marker,
               created_at, updated_at, completed_at
        FROM workdir_removal_operations"#
 }
@@ -795,6 +825,26 @@ fn read_operation(row: &rusqlite::Row<'_>) -> rusqlite::Result<WorkdirRemovalOpe
         .map(|value| parse_disposition(&value))
         .transpose()?;
     let attempt_count = row.get::<_, i64>(10)?;
+    let attempt_owner_pid = row.get::<_, Option<i64>>(14)?;
+    let attempt_owner_start_marker = row.get::<_, Option<i64>>(15)?;
+    let attempt_owner = match (attempt_owner_pid, attempt_owner_start_marker) {
+        (None, None) => None,
+        (Some(process_id), Some(process_start_marker)) => Some(WorkdirRemovalAttemptOwner {
+            process_id: process_id
+                .try_into()
+                .map_err(|_| rusqlite::Error::IntegralValueOutOfRange(14, process_id))?,
+            process_start_marker: process_start_marker
+                .try_into()
+                .map_err(|_| rusqlite::Error::IntegralValueOutOfRange(15, process_start_marker))?,
+        }),
+        _ => {
+            return Err(rusqlite::Error::FromSqlConversionFailure(
+                14,
+                rusqlite::types::Type::Integer,
+                "incomplete Workdir removal attempt owner".into(),
+            ));
+        }
+    };
     Ok(WorkdirRemovalOperation {
         operation_id: row.get(0)?,
         request_fingerprint: row.get(1)?,
@@ -812,9 +862,10 @@ fn read_operation(row: &rusqlite::Row<'_>) -> rusqlite::Result<WorkdirRemovalOpe
         retryable: row.get(11)?,
         disposition,
         failure_category: row.get(13)?,
-        created_at: row.get(14)?,
-        updated_at: row.get(15)?,
-        completed_at: row.get(16)?,
+        attempt_owner,
+        created_at: row.get(16)?,
+        updated_at: row.get(17)?,
+        completed_at: row.get(18)?,
     })
 }
 
@@ -857,6 +908,13 @@ mod tests {
     use crate::store::{AccountRecord, ControlPlaneStore, RepositoryRecord, WorkspaceRecord};
     use workspace_api::{RepositoryObservedStatus, RepositorySource, RepositorySourceKind};
 
+    fn attempt_owner() -> WorkdirRemovalAttemptOwner {
+        WorkdirRemovalAttemptOwner {
+            process_id: 100,
+            process_start_marker: 200,
+        }
+    }
+
     async fn seeded_store() -> (SqliteWorkspaceStore, WorkdirRegistryRecord) {
         let store = SqliteWorkspaceStore::in_memory().unwrap();
         store
@@ -950,6 +1008,7 @@ mod tests {
                 &reserved.workspace_id,
                 &reserved.operation_id,
                 &reserved.request_fingerprint,
+                attempt_owner(),
             )
             .unwrap();
         assert_eq!(first.attempt_count, 1);
@@ -962,6 +1021,7 @@ mod tests {
                 &failed.workspace_id,
                 &failed.operation_id,
                 &failed.request_fingerprint,
+                attempt_owner(),
             )
             .unwrap();
         assert_eq!(retry.attempt_count, 2);
@@ -981,6 +1041,7 @@ mod tests {
                 &completed.workspace_id,
                 &completed.operation_id,
                 &completed.request_fingerprint,
+                attempt_owner(),
             )
             .unwrap();
         assert_eq!(replay, completed);
@@ -1009,6 +1070,7 @@ mod tests {
                     &operation.workspace_id,
                     &operation.operation_id,
                     &operation.request_fingerprint,
+                    attempt_owner(),
                 );
                 if claim.is_ok() {
                     provider_calls.fetch_add(1, Ordering::SeqCst);
@@ -1039,6 +1101,63 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn failed_workdir_create_retry_cannot_start_after_removal_claim() {
+        use crate::store::WorkdirCreateOperationRecord;
+
+        let (store, workdir) = seeded_store().await;
+        let create = WorkdirCreateOperationRecord {
+            workspace_id: "workspace-a".to_string(),
+            operation_id: "create-a".to_string(),
+            request_fingerprint: "create-fingerprint".to_string(),
+            repository_id: "repository-a".to_string(),
+            selector: Some("develop".to_string()),
+            requested_runtime_id: Some("runtime-a".to_string()),
+            resolved_runtime_id: "runtime-a".to_string(),
+            config_revision: 1,
+            config_projection_digest: "projection-a".to_string(),
+            source_kind: Some("local_path".to_string()),
+            source_uri: Some("/repository-a".to_string()),
+            source_revision: Some(1),
+            source_fingerprint: Some("source-a".to_string()),
+            credential_id: None,
+            credential_revision: None,
+            host_trust_id: None,
+            host_trust_revision: None,
+            repository_access_mode: None,
+            cache_generation: 0,
+            working_directory_id: "workdir-a".to_string(),
+            state: "pending".to_string(),
+            failure: None,
+            created_at: "1".to_string(),
+            updated_at: "1".to_string(),
+        };
+        store.reserve_workdir_create_operation(&create).unwrap();
+        store
+            .finish_workdir_create_operation(
+                "workspace-a",
+                "create-a",
+                "create-fingerprint",
+                false,
+                Some("provider failed"),
+                "2",
+            )
+            .unwrap();
+        let intent =
+            workdir_removal_intent(&workdir, "workspace-api", "remove clean Workdir").unwrap();
+        store.reserve_workdir_removal_operation(&intent).unwrap();
+
+        let error = store
+            .begin_failed_workdir_create_retry("workspace-a", "create-a", "create-fingerprint", "3")
+            .unwrap_err();
+        assert!(matches!(error, Error::WorkdirAttachmentConflict(_)));
+        let create = store
+            .load_workdir_create_operation("workspace-a", "create-a")
+            .unwrap()
+            .unwrap();
+        assert_eq!(create.state, "failed");
+    }
+
     #[tokio::test]
     async fn pending_removal_fences_new_attachment_and_retry_rereads_live_reservation() {
         let (store, workdir) = seeded_store().await;
@@ -1062,6 +1181,7 @@ mod tests {
                 &failed.workspace_id,
                 &failed.operation_id,
                 &failed.request_fingerprint,
+                attempt_owner(),
             )
             .unwrap();
         let guards = store.workdir_removal_guards(&retry).unwrap();

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 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.
+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. Each active attempt persists the Server process ID and process-start marker. Recovery reclaims only an owner proven missing or replaced; a live or unobservable owner is never stolen.
 
 `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.
 
@@ -30,7 +30,7 @@ Workdir removal is one durable side-effect operation in the Workspace Server DB.
 Each attempt:
 
 1. resolves or revalidates the persisted same-Workspace Workdir, Runtime, Repository, and materialization identity;
-2. checks current attachments, attachment reservations, current assignment occupancy, retention/cleanup holds, and pending materialization authority;
+2. checks current attachments, attachment reservations, current assignment occupancy, retention/cleanup holds, and pending materialization authority; a failed Workdir-create retry must atomically return to `pending` before provider work and is rejected while removal is pending;
 3. retains dirty, occupied, blocked, or otherwise unknown Workdirs without detaching a Worker or forcing deletion;
 4. observes the owning Runtime/provider and calls its existing Workdir cleanup only for an eligible clean Workdir;
 5. treats only successful provider cleanup or exact `working_directory_not_found` as removal evidence;