fix: keep runtime responsive during workdir cleanup

crates/worker-runtime/src/http_server.rs

@@ -951,10 +951,12 @@ async fn observe_repository_ref(
 async fn list_working_directories(
     State(state): State<RuntimeHttpState>,
 ) -> RestResult<RuntimeHttpWorkingDirectoriesResponse> {
-    let working_directories = state
-        .runtime
-        .list_working_directories()
-        .map_err(RuntimeHttpRestError::runtime)?;
+    // Inventory providers may inspect the filesystem even for read-only requests.
+    let working_directories =
+        tokio::task::spawn_blocking(move || state.runtime.list_working_directories())
+            .await
+            .map_err(RuntimeHttpRestError::working_directory_join)?
+            .map_err(RuntimeHttpRestError::runtime)?;
     Ok(Json(RuntimeHttpWorkingDirectoriesResponse {
         working_directories,
     }))
@@ -989,10 +991,11 @@ async fn get_working_directory(
     State(state): State<RuntimeHttpState>,
     Path(working_directory_id): Path<String>,
 ) -> RestResult<RuntimeHttpWorkingDirectoryResponse> {
-    let working_directory = state
-        .runtime
-        .working_directory(&working_directory_id)
-        .map_err(RuntimeHttpRestError::runtime)?;
+    let working_directory =
+        tokio::task::spawn_blocking(move || state.runtime.working_directory(&working_directory_id))
+            .await
+            .map_err(RuntimeHttpRestError::working_directory_join)?
+            .map_err(RuntimeHttpRestError::runtime)?;
     Ok(Json(RuntimeHttpWorkingDirectoryResponse {
         working_directory,
     }))
@@ -1130,10 +1133,16 @@ async fn cleanup_working_directory(
     State(state): State<RuntimeHttpState>,
     Path(working_directory_id): Path<String>,
 ) -> RestResult<RuntimeHttpWorkingDirectoryResponse> {
-    let working_directory = state
-        .runtime
-        .cleanup_working_directory(&working_directory_id)
-        .map_err(RuntimeHttpRestError::runtime)?;
+    // Keep Runtime's occupancy check and deletion together, but do not run
+    // synchronous filesystem/Git work on the HTTP executor.
+    let working_directory = tokio::task::spawn_blocking(move || {
+        state
+            .runtime
+            .cleanup_working_directory(&working_directory_id)
+    })
+    .await
+    .map_err(RuntimeHttpRestError::working_directory_join)?
+    .map_err(RuntimeHttpRestError::runtime)?;
     Ok(Json(RuntimeHttpWorkingDirectoryResponse {
         working_directory,
     }))
@@ -2443,6 +2452,16 @@ impl RuntimeHttpRestError {
         Self::new(status, code, error.to_string())
     }
 
+    fn working_directory_join(_error: tokio::task::JoinError) -> Self {
+        // A panic payload can contain paths or credentials. A missing result
+        // also does not prove that cleanup made no changes; do not claim rollback.
+        Self::new(
+            StatusCode::INTERNAL_SERVER_ERROR,
+            "working_directory_task_failed",
+            "Working directory operation failed to return a result",
+        )
+    }
+
     fn json_rejection(error: JsonRejection) -> Self {
         Self::new(
             StatusCode::BAD_REQUEST,
@@ -4341,34 +4360,379 @@ mod tests {
         assert!(!error.to_string().contains('/'));
     }
 
+    enum CleanupHttpOutcome {
+        Removed,
+        Rejected(&'static str),
+        Panicked,
+    }
+
+    struct CleanupHttpGate {
+        entered: tokio::sync::oneshot::Sender<()>,
+        release: std::sync::mpsc::Receiver<()>,
+    }
+
+    struct CleanupHttpBackend {
+        gate: Mutex<Option<CleanupHttpGate>>,
+        outcome: CleanupHttpOutcome,
+    }
+
+    fn removed_working_directory() -> WorkingDirectoryStatus {
+        WorkingDirectoryStatus {
+            summary: crate::catalog::WorkingDirectorySummary {
+                working_directory_id: "cleanup-test".to_string(),
+                display_name: None,
+                repository_id: "repository-test".to_string(),
+                creation_selector: None,
+                creation_ref: None,
+                creation_tree: None,
+                current_selector: None,
+                current_ref: None,
+                current_tree: None,
+                observed_at_epoch_seconds: None,
+                materializer_kind: crate::catalog::MaterializerKind::RuntimeGitClone,
+                cleanup_target: None,
+                status: crate::catalog::WorkingDirectoryStatusKind::NotFound,
+                cleanliness: None,
+                occupied_by: None,
+            },
+        }
+    }
+
+    impl WorkerExecutionBackend for CleanupHttpBackend {
+        fn backend_id(&self) -> &str {
+            "http-cleanup-test"
+        }
+
+        fn spawn_worker(&self, request: WorkerExecutionSpawnRequest) -> WorkerExecutionSpawnResult {
+            AcceptingBackend.spawn_worker(request)
+        }
+
+        fn dispatch_input(
+            &self,
+            worker_ref: &WorkerRef,
+            input: WorkerInput,
+        ) -> WorkerExecutionResult {
+            AcceptingBackend.dispatch_input(worker_ref, input)
+        }
+
+        fn cleanup_working_directory(
+            &self,
+            working_directory_id: &str,
+        ) -> Result<WorkingDirectoryStatus, crate::working_directory::WorkingDirectoryDiagnostic>
+        {
+            assert_eq!(working_directory_id, "cleanup-test");
+            let gate = self.gate.lock().unwrap().take();
+            if let Some(gate) = gate {
+                gate.entered.send(()).expect("cleanup entry receiver");
+                // A watchdog also bounds a regression that blocks the sole
+                // Tokio thread, where an async timeout could not make progress.
+                gate.release
+                    .recv_timeout(std::time::Duration::from_secs(10))
+                    .expect("cleanup must be released after observations complete");
+            }
+            match self.outcome {
+                CleanupHttpOutcome::Removed => Ok(removed_working_directory()),
+                CleanupHttpOutcome::Rejected(code) => Err(
+                    crate::working_directory::WorkingDirectoryDiagnostic::rejected(
+                        code,
+                        "Resolve the current blocker and retry removal.",
+                    ),
+                ),
+                CleanupHttpOutcome::Panicked => panic!("private provider path /secret/cleanup"),
+            }
+        }
+    }
+
+    fn cleanup_http_router(
+        outcome: CleanupHttpOutcome,
+        gate: Option<CleanupHttpGate>,
+    ) -> (Runtime, Router) {
+        let runtime = Runtime::with_execution_backend(
+            RuntimeOptions::default(),
+            Arc::new(CleanupHttpBackend {
+                gate: Mutex::new(gate),
+                outcome,
+            }),
+        )
+        .unwrap();
+        let identity = RuntimeIdentityMaterial::generate("runtime-cleanup-test").unwrap();
+        let workspace_identity =
+            RuntimeIdentityMaterial::generate("workspace-cleanup-key").unwrap();
+        let workspace_public_key =
+            crate::auth::decode_public_key(&workspace_identity.public_key).unwrap();
+        let workspace_fingerprint = format!(
+            "sha256:{}",
+            crate::workspace_issuer::hex_lower(&sha2::Sha256::digest(workspace_public_key))
+        );
+        let app = runtime_http_router_with_workspace_auth(
+            runtime.clone(),
+            Some("cleanup-token".to_string()),
+            WorkspaceRuntimeHttpAuth {
+                verifier: WorkspaceCapabilityVerifier::new(
+                    vec![WorkspaceIssuerTrustRecord {
+                        workspace_id: "local".to_string(),
+                        backend_url: "https://backend.test".to_string(),
+                        key_id: "workspace-cleanup-key".to_string(),
+                        algorithm: "ed25519".to_string(),
+                        public_key: workspace_identity.public_key,
+                        public_key_fingerprint: workspace_fingerprint,
+                        identity_revision: 1,
+                        trust_generation: 1,
+                        state: WorkspaceIssuerTrustState::Active,
+                        registered_at_unix: 1,
+                        updated_at_unix: 1,
+                    }],
+                    Arc::new(InMemoryWorkspaceClaimReplayProtection::default()),
+                )
+                .unwrap(),
+                signer: RuntimeVerificationSigner::from_identity(&identity).unwrap(),
+                verifications: Arc::new(InMemoryWorkspaceRuntimeVerificationAuthority::default()),
+            },
+        );
+        (runtime, app)
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn slow_working_directory_cleanup_does_not_block_http_ping() {
+        let (entered, started) = tokio::sync::oneshot::channel();
+        let (release, wait_for_release) = std::sync::mpsc::channel();
+        let (runtime, app) = cleanup_http_router(
+            CleanupHttpOutcome::Removed,
+            Some(CleanupHttpGate {
+                entered,
+                release: wait_for_release,
+            }),
+        );
+        runtime
+            .store_config_bundle(test_bundle(ProfileSelector::Builtin(
+                "builtin:coder".to_string(),
+            )))
+            .unwrap();
+        let scope = RuntimeWorkspaceScope::new("local", "local-token");
+        let mut request = task_request("unrelated Worker remains observable during cleanup");
+        request.workspace_api = Some(WorkspaceApiRef {
+            workspace_id: "local".to_string(),
+            base_url: "https://backend.test".to_string(),
+        });
+        let worker = runtime.create_worker_scoped(&scope, request).unwrap();
+        assert!(worker.workdir_attachments.is_empty());
+        let worker_path = format!("/v1/workers/{}", worker.worker_id);
+        let path = "/v1/working-directories/cleanup-test";
+        let unauthorized = empty_request(app.clone(), Method::DELETE, path).await;
+        assert_eq!(unauthorized.status(), StatusCode::UNAUTHORIZED);
+        let error: RuntimeHttpErrorResponse = read_json(unauthorized).await;
+        assert_eq!(error.error.code, "unauthorized");
+
+        // A synchronous RuntimeState lock can stall this sole executor thread,
+        // including Tokio timeouts. Release from an independent thread on either
+        // completion, assertion unwind, or watchdog expiry, and report expiry as
+        // failure even if observations resume after the forced release.
+        let (observed, wait_for_observations) = std::sync::mpsc::channel();
+        let watchdog = std::thread::spawn(move || {
+            let timed_out = matches!(
+                wait_for_observations.recv_timeout(std::time::Duration::from_secs(5)),
+                Err(std::sync::mpsc::RecvTimeoutError::Timeout)
+            );
+            let _ = release.send(());
+            timed_out
+        });
+        let cleanup = tokio::spawn(authed_empty_request(
+            app.clone(),
+            Method::DELETE,
+            path,
+            "cleanup-token",
+        ));
+        tokio::time::timeout(std::time::Duration::from_secs(2), started)
+            .await
+            .expect("cleanup provider must be entered")
+            .expect("cleanup entry signal");
+
+        // Unlike ping, these HTTP routes synchronously acquire RuntimeState.
+        // Exercise existing Worker paths so authorization/not-found shortcuts
+        // cannot make a global occupancy-lock regression pass unnoticed.
+        for observation_path in [
+            "/v1/workers".to_string(),
+            worker_path.clone(),
+            format!("{worker_path}/session?workspace_id=local"),
+        ] {
+            let response = tokio::time::timeout(
+                std::time::Duration::from_secs(2),
+                authed_empty_request(app.clone(), Method::GET, &observation_path, "cleanup-token"),
+            )
+            .await
+            .unwrap_or_else(|_| panic!("{observation_path} must finish while cleanup is gated"));
+            let status = response.status();
+            let body: serde_json::Value = read_json(response).await;
+            assert_eq!(status, StatusCode::OK, "{observation_path}: {body}");
+            if observation_path == "/v1/workers" {
+                assert_eq!(body["workers"].as_array().unwrap().len(), 1, "{body}");
+                assert_eq!(
+                    body["workers"][0]["worker_id"],
+                    worker.worker_id.to_string()
+                );
+            } else if observation_path == worker_path {
+                assert_eq!(body["worker"]["worker_id"], worker.worker_id.to_string());
+            } else {
+                // This minimal fake has no session store. Its normal, typed
+                // availability result still requires the real scoped HTTP read.
+                assert_eq!(
+                    body,
+                    serde_json::to_value(runtime_api::WorkerSessionAvailability::Unavailable {
+                        reason: runtime_api::WorkerSessionUnavailableReason::StorageUnavailable,
+                        message: "retained session storage is unavailable".to_string(),
+                    })
+                    .unwrap(),
+                );
+            }
+        }
+
+        // Exercise the same scoped snapshot acquisition used by Runtime WS
+        // subscriptions, without requiring a socket or the ws-server feature.
+        let subscription = runtime
+            .subscribe_event_selector_scoped(
+                &scope,
+                protocol::subscription::EventSubscriptionSelector::RuntimeWorkers,
+            )
+            .expect("Runtime subscription snapshot must complete while cleanup is gated");
+        let protocol::subscription::SubscriptionSnapshot::Workers { workers } =
+            subscription.snapshot()
+        else {
+            panic!("expected Runtime Worker subscription snapshot");
+        };
+        assert_eq!(workers.len(), 1);
+        assert_eq!(workers[0].worker_id.as_str(), worker.worker_id.to_string());
+
+        let ping = tokio::time::timeout(
+            std::time::Duration::from_secs(2),
+            app.oneshot(
+                Request::builder()
+                    .method(Method::GET)
+                    .uri("/v1/ping")
+                    .header(header::AUTHORIZATION, "Bearer cleanup-token")
+                    .header(RUNTIME_WORKSPACE_SCOPE_HEADER, "local")
+                    .body(Body::empty())
+                    .unwrap(),
+            ),
+        )
+        .await
+        .expect("ping must finish while cleanup is gated")
+        .unwrap();
+        let ping_status = ping.status();
+        let ping_body: serde_json::Value = read_json(ping).await;
+        assert_eq!(ping_status, StatusCode::OK, "{ping_body}");
+        assert_eq!(ping_body["runtime_id"], "runtime-cleanup-test");
+        assert_eq!(ping_body["protocol_version"], RUNTIME_HTTP_PROTOCOL_VERSION);
+        assert!(
+            !cleanup.is_finished(),
+            "cleanup must still be gated after ping",
+        );
+        // The watchdog owns the release sender; dropping `observed` on an
+        // assertion failure also releases the provider instead of leaking a task.
+        let observations_completed = observed.send(()).is_ok();
+        let watchdog_expired = watchdog.join().unwrap();
+        assert!(
+            observations_completed && !watchdog_expired,
+            "Worker observations, subscription snapshot, and ping must finish before cleanup is released",
+        );
+        let response = tokio::time::timeout(std::time::Duration::from_secs(2), cleanup)
+            .await
+            .expect("released cleanup must finish")
+            .unwrap();
+        let status = response.status();
+        let body: serde_json::Value = read_json(response).await;
+        assert_eq!(status, StatusCode::OK, "{body}");
+        assert_eq!(
+            body,
+            serde_json::to_value(RuntimeHttpWorkingDirectoryResponse {
+                working_directory: removed_working_directory(),
+            })
+            .unwrap(),
+        );
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn working_directory_cleanup_panic_returns_bounded_http_error() {
+        let (_, app) = cleanup_http_router(CleanupHttpOutcome::Panicked, None);
+        let response = authed_empty_request(
+            app,
+            Method::DELETE,
+            "/v1/working-directories/cleanup-test",
+            "cleanup-token",
+        )
+        .await;
+        let status = response.status();
+        let body: serde_json::Value = read_json(response).await;
+        assert_eq!(status, StatusCode::INTERNAL_SERVER_ERROR, "{body}");
+        assert_eq!(
+            body,
+            serde_json::json!({
+                "error": {
+                    "code": "working_directory_task_failed",
+                    "message": "Working directory operation failed to return a result",
+                },
+            }),
+        );
+    }
+
     #[tokio::test]
     async fn cleanup_cause_codes_survive_http_error_serialization() {
         let cases = [
-            ("mount_present", StatusCode::CONFLICT),
-            ("mount_check_unavailable", StatusCode::SERVICE_UNAVAILABLE),
-            ("permission_denied", StatusCode::FORBIDDEN),
-            ("resource_busy", StatusCode::CONFLICT),
-            ("storage_unavailable", StatusCode::SERVICE_UNAVAILABLE),
-            ("ownership_unknown", StatusCode::CONFLICT),
-            ("changes_present", StatusCode::CONFLICT),
-            ("changes_unknown", StatusCode::CONFLICT),
+            ("working_directory_not_found", StatusCode::NOT_FOUND),
+            (
+                "working_directory_cleanup_mount_present",
+                StatusCode::CONFLICT,
+            ),
+            (
+                "working_directory_cleanup_mount_check_unavailable",
+                StatusCode::SERVICE_UNAVAILABLE,
+            ),
+            (
+                "working_directory_cleanup_permission_denied",
+                StatusCode::FORBIDDEN,
+            ),
+            (
+                "working_directory_cleanup_resource_busy",
+                StatusCode::CONFLICT,
+            ),
+            (
+                "working_directory_cleanup_storage_unavailable",
+                StatusCode::SERVICE_UNAVAILABLE,
+            ),
+            (
+                "working_directory_cleanup_ownership_unknown",
+                StatusCode::CONFLICT,
+            ),
+            (
+                "working_directory_cleanup_changes_present",
+                StatusCode::CONFLICT,
+            ),
+            (
+                "working_directory_cleanup_changes_unknown",
+                StatusCode::CONFLICT,
+            ),
         ];
-        for (cause, status) in cases {
-            let code = format!("working_directory_cleanup_{cause}");
+        for (code, expected_status) in cases {
+            let (_, app) = cleanup_http_router(CleanupHttpOutcome::Rejected(code), None);
+            let response = authed_empty_request(
+                app,
+                Method::DELETE,
+                "/v1/working-directories/cleanup-test",
+                "cleanup-token",
+            )
+            .await;
+            let status = response.status();
+            let json: serde_json::Value = read_json(response).await;
+            assert_eq!(status, expected_status, "{json}");
+            assert_eq!(json["error"]["code"], code, "{json}");
             let diagnostic = crate::working_directory::WorkingDirectoryDiagnostic::rejected(
-                &code,
+                code,
                 "Resolve the current blocker and retry removal.",
             );
-            let response =
-                RuntimeHttpRestError::runtime(RuntimeError::WorkingDirectory(diagnostic))
-                    .into_response();
-            assert_eq!(response.status(), status);
-            let body = axum::body::to_bytes(response.into_body(), 2048)
-                .await
-                .unwrap();
-            let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
-            assert_eq!(json["error"]["code"], code, "{json}");
-            assert!(json["error"]["message"].as_str().unwrap().contains("retry"));
+            assert_eq!(
+                json["error"]["message"],
+                RuntimeError::WorkingDirectory(diagnostic).to_string(),
+                "{json}",
+            );
         }
     }
 

crates/worker-runtime/src/runtime.rs

@@ -59,6 +59,54 @@ use tokio::sync::broadcast;
 use tokio::sync::mpsc;
 use uuid::Uuid;
 
+mod workdir_session;
+
+#[derive(Debug)]
+enum WorkdirOperation {
+    Use(usize),
+    Cleanup,
+}
+
+/// Transient admission authority, not persisted cleanup or recovery state.
+#[derive(Debug)]
+struct WorkdirOperationGuard {
+    state: Weak<Mutex<RuntimeState>>,
+    ids: Vec<String>,
+    cleanup: bool,
+}
+
+impl Drop for WorkdirOperationGuard {
+    fn drop(&mut self) {
+        let Some(state) = self.state.upgrade() else {
+            return;
+        };
+        let mut state = state
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner());
+        for id in &self.ids {
+            match state.workdir_operations.get_mut(id) {
+                Some(WorkdirOperation::Use(count)) if !self.cleanup && *count > 1 => *count -= 1,
+                Some(WorkdirOperation::Use(_)) if !self.cleanup => {
+                    state.workdir_operations.remove(id);
+                }
+                Some(WorkdirOperation::Cleanup) if self.cleanup => {
+                    state.workdir_operations.remove(id);
+                }
+                _ => {}
+            }
+        }
+    }
+}
+
+fn workdir_operation_busy() -> RuntimeError {
+    RuntimeError::WorkingDirectory(
+        crate::working_directory::WorkingDirectoryDiagnostic::rejected(
+            "working_directory_cleanup_resource_busy",
+            "Workdir has an in-progress operation; retry after it finishes",
+        ),
+    )
+}
+
 /// Workspace-scoped Runtime authorization context supplied by a trusted backend.
 #[derive(Clone, Debug, Eq, PartialEq)]
 pub struct RuntimeWorkspaceScope {
@@ -296,6 +344,37 @@ impl Drop for WorkerOperationLease {
 }
 
 impl Runtime {
+    fn reserve_workdir_use<'a>(
+        &self,
+        ids: impl IntoIterator<Item = &'a str>,
+    ) -> Result<WorkdirOperationGuard, RuntimeError> {
+        let mut ids: Vec<String> = ids.into_iter().map(str::to_string).collect();
+        ids.sort();
+        ids.dedup();
+        let mut state = self.lock()?;
+        state.ensure_running()?;
+        for id in &ids {
+            state.ensure_workdir_not_removing(id)?;
+        }
+        for id in &ids {
+            match state
+                .workdir_operations
+                .entry(id.clone())
+                .or_insert(WorkdirOperation::Use(0))
+            {
+                WorkdirOperation::Use(count) => *count += 1,
+                WorkdirOperation::Cleanup => {
+                    unreachable!("cleanup admission checked under state lock")
+                }
+            }
+        }
+        Ok(WorkdirOperationGuard {
+            state: Arc::downgrade(&self.inner),
+            ids,
+            cleanup: false,
+        })
+    }
+
     /// Create a memory-backed Runtime with generated identity.
     pub fn new_memory() -> Self {
         Self::with_options(RuntimeOptions::default())
@@ -515,8 +594,15 @@ impl Runtime {
     /// Create a Runtime-owned working directory through the attached execution backend.
     pub fn create_working_directory(
         &self,
-        request: WorkingDirectoryRequest,
+        mut request: WorkingDirectoryRequest,
     ) -> Result<CatalogWorkingDirectoryStatus, RuntimeError> {
+        let id = request
+            .backend_workdir_id
+            .get_or_insert_with(|| {
+                crate::working_directory::next_working_directory_id(&request.repository.id)
+            })
+            .clone();
+        let _operation = self.reserve_workdir_use([id.as_str()])?;
         let backend = {
             let state = self.lock()?;
             state.ensure_running()?;
@@ -581,6 +667,7 @@ impl Runtime {
         &self,
         request: WorkingDirectoryRepositoryAccessRequest,
     ) -> Result<(), RuntimeError> {
+        let _operation = self.reserve_workdir_use([request.working_directory_id.as_str()])?;
         let backend = {
             let state = self.lock()?;
             state.ensure_running()?;
@@ -775,6 +862,7 @@ impl Runtime {
         working_directory_id: &str,
         owner_worker_ref: Option<&WorkerRef>,
     ) -> Result<workdir::WorkdirSessionHandle, RuntimeError> {
+        let operation = self.reserve_workdir_use([working_directory_id])?;
         let backend = {
             let mut state = self.lock()?;
             state.ensure_running()?;
@@ -809,9 +897,10 @@ impl Runtime {
         backend
             .working_directory(working_directory_id)
             .map_err(RuntimeError::WorkingDirectory)?;
-        backend
+        let session = backend
             .open_workdir_session(working_directory_id)
-            .map_err(RuntimeError::WorkingDirectory)
+            .map_err(RuntimeError::WorkingDirectory)?;
+        Ok(workdir_session::retain_operation(session, operation))
     }
 
     /// Cleanup a Runtime-owned working directory.
@@ -819,30 +908,40 @@ impl Runtime {
         &self,
         working_directory_id: &str,
     ) -> Result<CatalogWorkingDirectoryStatus, RuntimeError> {
-        // Hold occupancy authority through the side effect, preventing an
-        // attachment from entering between the check and physical removal.
-        // This synchronous provider call must not reenter Runtime. The Git
-        // materializer does filesystem/Git work only. The global guard also
-        // delays unrelated state operations during hashing/unlink; releasing it
-        // would require equivalent per-workdir exclusion on ALL attachment paths,
-        // not just moving the occupied check before the provider call.
-        let state = self.lock()?;
-        state.ensure_running()?;
-        if let Some(worker_id) = state.worker_id_for_workdir(working_directory_id) {
-            return Err(RuntimeError::InvalidRequest(format!(
-                "working directory {working_directory_id} is assigned to worker {worker_id}"
-            )));
-        }
-        let backend = state.execution_backend.clone().ok_or_else(|| {
-            RuntimeError::ExecutionBackendUnavailable {
-                message: "working directory cleanup requires an execution backend".to_string(),
+        // Admit cleanup atomically with attachment/use admission, but reserve
+        // only this Workdir while the provider hashes/unlinks. No filesystem
+        // work or wait for a Workdir operation may hold the global state mutex.
+        let backend = {
+            let mut state = self.lock()?;
+            state.ensure_running()?;
+            if state.workdir_operations.contains_key(working_directory_id) {
+                return Err(workdir_operation_busy());
             }
-        })?;
-        let result = backend
+            if let Some(worker_id) = state.worker_id_for_workdir(working_directory_id) {
+                return Err(RuntimeError::InvalidRequest(format!(
+                    "working directory {working_directory_id} is assigned to worker {worker_id}"
+                )));
+            }
+            let backend = state.execution_backend.clone().ok_or_else(|| {
+                RuntimeError::ExecutionBackendUnavailable {
+                    message: "working directory cleanup requires an execution backend".to_string(),
+                }
+            })?;
+            state
+                .workdir_operations
+                .insert(working_directory_id.to_string(), WorkdirOperation::Cleanup);
+            backend
+        };
+        // Construct/drop the RAII owner outside the state guard: failure, panic
+        // and request cancellation must not leave stale exclusion or deadlock.
+        let _operation = WorkdirOperationGuard {
+            state: Arc::downgrade(&self.inner),
+            ids: vec![working_directory_id.to_string()],
+            cleanup: true,
+        };
+        backend
             .cleanup_working_directory(working_directory_id)
-            .map_err(RuntimeError::from);
-        drop(state);
-        result
+            .map_err(RuntimeError::from)
     }
 
     fn annotate_working_directory_statuses(
@@ -1041,7 +1140,7 @@ impl Runtime {
 
     fn create_worker_with_workspace_inner(
         &self,
-        request: CreateWorkerRequest,
+        mut request: CreateWorkerRequest,
         scope: Option<&RuntimeWorkspaceScope>,
     ) -> Result<WorkerDetail, RuntimeError> {
         let operation_lock = self.worker_operation_lock(request.worker_id)?;
@@ -1051,6 +1150,17 @@ impl Runtime {
         if let Some(existing) = self.existing_worker_for_create(&request, scope)? {
             return Ok(existing);
         }
+        // Assign generated materializations before publishing the durable
+        // request, so cleanup cannot discover an unclaimed creation mid-spawn.
+        for attachment in &mut request.workdir_attachment_requests {
+            let directory = &mut attachment.working_directory;
+            if directory.backend_workdir_id.is_none() {
+                directory.backend_workdir_id = Some(
+                    crate::working_directory::next_working_directory_id(&directory.repository.id),
+                );
+            }
+        }
+        let _workdir_operation = self.reserve_workdir_use(requested_workdir_ids(&request))?;
         self.refresh_workspace_config(&request)?;
         let (backend, worker_ref, spawn_request) = {
             let mut state = self.lock()?;
@@ -1565,6 +1675,9 @@ impl Runtime {
         attachments.sort_by(|left, right| left.alias.cmp(&right.alias));
 
         let mut state = self.lock()?;
+        for attachment in &attachments {
+            state.ensure_workdir_not_removing(&attachment.working_directory_id)?;
+        }
         let previous = state.worker(worker_ref)?.clone();
         {
             let worker = state.worker_mut(worker_ref)?;
@@ -1875,6 +1988,13 @@ impl Runtime {
                 if let Some(receipt) = &owner.receipt {
                     return Ok((Some(receipt.clone()), None));
                 }
+                if let Some(preparation) = &owner.request.preparation {
+                    for id in
+                        preparation_workdir_ids(preparation, state.runtime_identity.as_deref())
+                    {
+                        state.ensure_workdir_not_removing(id)?;
+                    }
+                }
                 if candidate.restore_guard.active_request_id.as_deref() != Some(&request.request_id)
                     || candidate
                         .pending_restore
@@ -2074,6 +2194,19 @@ impl Runtime {
                     state.ensure_workspace_owner(scope, true)?;
                     state.persist_runtime_snapshot()?;
                 }
+                // Recovery uses the persisted operation, not new preparation.
+                // Discard it before checking Workdir admission as well as before
+                // publishing the owner; unrelated cleanup must not reject recovery.
+                if recovering_pending {
+                    request.preparation = None;
+                }
+                if let Some(preparation) = &request.preparation {
+                    for id in
+                        preparation_workdir_ids(preparation, state.runtime_identity.as_deref())
+                    {
+                        state.ensure_workdir_not_removing(id)?;
+                    }
+                }
                 let pending = candidate
                     .pending_restore
                     .unwrap_or_else(|| PendingWorkerRestore {
@@ -2083,9 +2216,6 @@ impl Runtime {
                     });
                 let operation_id = pending.operation_id;
                 candidate.pending_restore = Some(pending);
-                if recovering_pending {
-                    request.preparation = None;
-                }
                 candidate.restore_guard.rotate();
                 candidate.restore_guard.active_request_id = Some(request.request_id.clone());
                 candidate.restore_guard.owners.insert(
@@ -2517,6 +2647,28 @@ impl Runtime {
             )
         };
 
+        let _workdir_operation = self.reserve_workdir_use(
+            requested_workdir_ids(&request.request)
+                .into_iter()
+                .chain(
+                    request
+                        .previous_workdir_attachments
+                        .iter()
+                        .map(|attachment| {
+                            attachment
+                                .working_directory
+                                .summary
+                                .working_directory_id
+                                .as_str()
+                        }),
+                )
+                .chain(
+                    request
+                        .logical_workdir_attachments
+                        .iter()
+                        .map(|attachment| attachment.working_directory_id.as_str()),
+                ),
+        )?;
         if !reconcile && let Err(result) = backend.preflight_restore(&request) {
             // A preflight may attest an already-connected Controller snapshot.
             // Repeated restore is read-only; do not replace its operation owner.
@@ -4853,6 +5005,7 @@ struct RuntimeState {
     #[cfg(feature = "fs-store")]
     next_diagnostic_id: u64,
     workers: BTreeMap<WorkerId, WorkerRecord>,
+    workdir_operations: BTreeMap<String, WorkdirOperation>,
     workspace_owners: BTreeMap<String, String>,
     config_bundles: BTreeMap<String, ConfigBundle>,
     workspace_config_latest: BTreeMap<String, ConfigBundleRef>,
@@ -4882,6 +5035,7 @@ impl RuntimeState {
             #[cfg(feature = "fs-store")]
             next_diagnostic_id: 1,
             workers: BTreeMap::new(),
+            workdir_operations: BTreeMap::new(),
             workspace_owners: BTreeMap::new(),
             config_bundles: BTreeMap::new(),
             workspace_config_latest: BTreeMap::new(),
@@ -4912,6 +5066,7 @@ impl RuntimeState {
             #[cfg(feature = "fs-store")]
             next_diagnostic_id: 1,
             workers: BTreeMap::new(),
+            workdir_operations: BTreeMap::new(),
             workspace_owners: BTreeMap::new(),
             config_bundles: BTreeMap::new(),
             workspace_config_latest: BTreeMap::new(),
@@ -5056,6 +5211,7 @@ impl RuntimeState {
             workspace_backend_resource_clients: BTreeMap::new(),
             next_diagnostic_id,
             workers,
+            workdir_operations: BTreeMap::new(),
             config_bundles: BTreeMap::new(),
             workspace_config_latest: BTreeMap::new(),
             workspace_config_fetch_gates: BTreeMap::new(),
@@ -5470,6 +5626,16 @@ 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());
+        }
+        Ok(())
+    }
+
     fn worker_id_for_workdir(&self, working_directory_id: &str) -> Option<WorkerId> {
         self.workers.values().find_map(|worker| {
             let has_status = worker.workdir_attachments.iter().any(|attachment| {
@@ -5478,7 +5644,23 @@ impl RuntimeState {
             let has_request = worker.request.as_ref().is_some_and(|request| {
                 requested_workdir_ids(request).contains(&working_directory_id)
             });
-            (has_status || has_request).then_some(worker.worker_id)
+            let has_logical = worker
+                .logical_workdir_attachments
+                .iter()
+                .any(|attachment| attachment.working_directory_id == working_directory_id);
+            let has_preparation = worker
+                .restore_guard
+                .active_request_id
+                .as_ref()
+                .and_then(|id| worker.restore_guard.owners.get(id))
+                .filter(|owner| owner.receipt.is_none())
+                .and_then(|owner| owner.request.preparation.as_ref())
+                .is_some_and(|preparation| {
+                    preparation_workdir_ids(preparation, self.runtime_identity.as_deref())
+                        .contains(&working_directory_id)
+                });
+            (has_status || has_request || has_logical || has_preparation)
+                .then_some(worker.worker_id)
         })
     }
 
@@ -6179,6 +6361,33 @@ fn durable_create_worker_request(request: &CreateWorkerRequest) -> CreateWorkerR
     durable
 }
 
+fn preparation_workdir_ids<'a>(
+    preparation: &'a runtime_api::WorkerRestorePreparation,
+    runtime_id: Option<&str>,
+) -> Vec<&'a str> {
+    let local = |owner: &str| runtime_id.is_none_or(|runtime_id| runtime_id == owner);
+    preparation
+        .workdir_attachments
+        .iter()
+        .flatten()
+        .map(|attachment| attachment.working_directory_id.as_str())
+        .chain(
+            preparation
+                .repository_access_workdirs
+                .iter()
+                .filter(|reference| local(&reference.runtime_id))
+                .map(|reference| reference.working_directory_id.as_str()),
+        )
+        .chain(
+            preparation
+                .repository_access
+                .iter()
+                .filter(|access| local(&access.materialization.runtime_id))
+                .map(|access| access.working_directory_id.as_str()),
+        )
+        .collect()
+}
+
 fn requested_workdir_ids(request: &CreateWorkerRequest) -> Vec<&str> {
     request
         .workdir_attachments
@@ -7641,6 +7850,35 @@ mod tests {
         }
     }
 
+    struct WorkdirEffectGate {
+        entered: std::sync::mpsc::Sender<String>,
+        release: std::sync::mpsc::Receiver<()>,
+    }
+
+    impl WorkdirEffectGate {
+        fn wait(slot: &Mutex<Option<Self>>, id: &str) {
+            let gate = slot.lock().unwrap().take();
+            if let Some(gate) = gate {
+                gate.entered.send(id.to_string()).unwrap();
+                gate.release
+                    .recv_timeout(std::time::Duration::from_secs(5))
+                    .expect("Workdir effect did not receive release before timeout");
+            }
+        }
+
+        fn install(
+            slot: &Mutex<Option<Self>>,
+        ) -> (
+            std::sync::mpsc::Receiver<String>,
+            std::sync::mpsc::Sender<()>,
+        ) {
+            let (entered, receiver) = std::sync::mpsc::channel();
+            let (sender, release) = std::sync::mpsc::channel();
+            *slot.lock().unwrap() = Some(Self { entered, release });
+            (receiver, sender)
+        }
+    }
+
     #[derive(Default)]
     struct TestExecutionBackend {
         spawn_result: Mutex<Option<WorkerExecutionSpawnResult>>,
@@ -7673,6 +7911,10 @@ mod tests {
         repository_access_available: AtomicBool,
         working_directory_requests: Mutex<Vec<WorkingDirectoryRequest>>,
         cleanup_occupancy_probe: Mutex<Option<Arc<Mutex<RuntimeState>>>>,
+        cleanup_gate: Mutex<Option<WorkdirEffectGate>>,
+        workdir_use_gate: Mutex<Option<WorkdirEffectGate>>,
+        cleanup_calls: std::sync::atomic::AtomicUsize,
+        cleanup_panics: AtomicBool,
         preserve_submission_acknowledgement_id: AtomicBool,
         #[cfg(feature = "ws-server")]
         snapshots: Mutex<BTreeMap<WorkerId, protocol::Event>>,
@@ -7717,10 +7959,21 @@ mod tests {
             &self,
             _id: &str,
         ) -> Result<CatalogWorkingDirectoryStatus, WorkingDirectoryDiagnostic> {
+            self.cleanup_calls.fetch_add(1, Ordering::SeqCst);
+            WorkdirEffectGate::wait(&self.cleanup_gate, _id);
+            if self.cleanup_panics.load(Ordering::SeqCst) {
+                panic!("cleanup provider failed after admission");
+            }
             if let Some(state) = self.cleanup_occupancy_probe.lock().unwrap().as_ref() {
+                let state = state
+                    .try_lock()
+                    .expect("physical cleanup must not hold global Runtime state");
                 assert!(
-                    matches!(state.try_lock(), Err(std::sync::TryLockError::WouldBlock)),
-                    "occupancy must remain exclusive during physical cleanup"
+                    matches!(
+                        state.workdir_operations.get(_id),
+                        Some(WorkdirOperation::Cleanup)
+                    ),
+                    "the target Workdir must remain reserved during physical cleanup"
                 );
             }
             Err(WorkingDirectoryDiagnostic::rejected(
@@ -7749,6 +8002,10 @@ mod tests {
             &self,
             request: &WorkingDirectoryRequest,
         ) -> Result<CatalogWorkingDirectoryStatus, WorkingDirectoryDiagnostic> {
+            WorkdirEffectGate::wait(
+                &self.workdir_use_gate,
+                request.backend_workdir_id.as_deref().unwrap(),
+            );
             self.working_directory_requests
                 .lock()
                 .unwrap()
@@ -7763,6 +8020,7 @@ mod tests {
             &self,
             request: &WorkingDirectoryRepositoryAccessRequest,
         ) -> Result<(), WorkingDirectoryDiagnostic> {
+            WorkdirEffectGate::wait(&self.workdir_use_gate, &request.working_directory_id);
             self.repository_accesses
                 .lock()
                 .unwrap()
@@ -8068,7 +8326,7 @@ mod tests {
     }
 
     #[test]
-    fn workdir_removal_keeps_occupancy_exclusive_during_provider_side_effect() {
+    fn workdir_removal_reserves_only_its_target_and_releases_on_failure() {
         let (runtime, backend) = runtime_and_backend();
         *backend.cleanup_occupancy_probe.lock().unwrap() = Some(runtime.inner.clone());
         let error = runtime
@@ -8078,11 +8336,384 @@ mod tests {
             matches!(error, RuntimeError::WorkingDirectory(ref diagnostic)
             if diagnostic.code == "working_directory_cleanup_resource_busy")
         );
-        // The lock is released on failure, permitting normal attachment/retry.
-        assert!(runtime.inner.try_lock().is_ok());
+        // Failure releases the per-target reservation for normal use/retry.
+        assert!(runtime.inner.lock().unwrap().workdir_operations.is_empty());
         *backend.cleanup_occupancy_probe.lock().unwrap() = None;
     }
 
+    fn workdir_effect_request(id: Option<&str>) -> WorkingDirectoryRequest {
+        WorkingDirectoryRequest {
+            repository: WorkingDirectoryRepository {
+                id: "repository-1".into(),
+                provider: "git".into(),
+                source: server_api::RepositorySource {
+                    kind: server_api::RepositorySourceKind::LocalPath,
+                    uri: "/unused-test-repository".into(),
+                },
+                source_revision: 1,
+                source_fingerprint: "sha256:test-source".into(),
+                selector: None,
+            },
+            display_name: None,
+            materializer: MaterializerKind::RuntimeGitClone,
+            backend_workdir_id: id.map(str::to_string),
+            materialization: None,
+        }
+    }
+
+    fn workdir_access_request(id: &str) -> WorkingDirectoryRepositoryAccessRequest {
+        WorkingDirectoryRepositoryAccessRequest {
+            working_directory_id: id.into(),
+            materialization: RepositoryMaterializationContext {
+                workspace_id: "workspace-a".into(),
+                runtime_id: "runtime-1".into(),
+                operation_id: "operation-1".into(),
+                config_revision: 1,
+                config_projection_digest: "sha256:test-projection".into(),
+                ssh: None,
+            },
+        }
+    }
+
+    fn assert_workdir_busy(error: RuntimeError) {
+        assert!(
+            matches!(error, RuntimeError::WorkingDirectory(ref diagnostic)
+            if diagnostic.code == "working_directory_cleanup_resource_busy"),
+            "{error:?}"
+        );
+    }
+
+    #[test]
+    fn active_workdir_cleanup_rejects_same_id_admissions_without_blocking_other_workers() {
+        let (runtime, backend) = runtime_and_backend();
+        let owner = scope("workspace-a", "server-a");
+        let worker = runtime
+            .create_worker_scoped(&owner, scoped_task_request("unrelated", "workspace-a"))
+            .unwrap();
+        let id = "workdir-cleanup-active";
+        let (entered, release) = WorkdirEffectGate::install(&backend.cleanup_gate);
+        let cleanup_runtime = runtime.clone();
+        let cleanup = std::thread::spawn(move || cleanup_runtime.cleanup_working_directory(id));
+        assert_eq!(
+            entered
+                .recv_timeout(std::time::Duration::from_secs(5))
+                .unwrap(),
+            id
+        );
+
+        assert_eq!(runtime.list_workers_scoped(&owner).unwrap().len(), 1);
+        assert_workdir_busy(runtime.cleanup_working_directory(id).unwrap_err());
+        assert_workdir_busy(
+            runtime
+                .create_working_directory(workdir_effect_request(Some(id)))
+                .unwrap_err(),
+        );
+        assert_workdir_busy(
+            runtime
+                .authorize_working_directory_repository_access(workdir_access_request(id))
+                .unwrap_err(),
+        );
+        assert_workdir_busy(
+            runtime
+                .open_workdir_session_scoped(&owner, id, None)
+                .unwrap_err(),
+        );
+        let mut create = scoped_task_request("conflicting attachment", "workspace-a");
+        create.workdir_attachments = vec![WorkingDirectoryAttachmentClaim {
+            alias: workdir::WorkdirAttachmentAlias::new("checkout").unwrap(),
+            working_directory_id: id.into(),
+            relative_cwd: None,
+            capabilities: workdir::WorkdirSessionCapabilities::ALL,
+        }];
+        assert_workdir_busy(runtime.create_worker_scoped(&owner, create).unwrap_err());
+        let attachment = LogicalWorkdirAttachment {
+            alias: workdir::WorkdirAttachmentAlias::new("checkout").unwrap(),
+            working_directory_id: id.into(),
+            capabilities: workdir::WorkdirSessionCapabilities::ALL,
+        };
+        assert_workdir_busy(
+            runtime
+                .replace_worker_workdir_attachments_scoped(
+                    &owner,
+                    &worker.worker_ref,
+                    vec![attachment.clone()],
+                )
+                .unwrap_err(),
+        );
+        runtime
+            .stop_worker_scoped(&owner, &worker.worker_ref, None)
+            .unwrap();
+        let restore = runtime.test_restore_request(&worker.worker_ref);
+        assert_workdir_busy(
+            runtime
+                .coordinate_worker_restore_operation(
+                    &worker.worker_ref,
+                    runtime_api::WorkerRestoreCoordinationRequest {
+                        expected_observation_token: restore.expected_observation_token,
+                        request_id: restore.request_id,
+                        preparation: Some(runtime_api::WorkerRestorePreparation {
+                            workdir_attachments: Some(vec![
+                                runtime_api::LogicalWorkdirAttachment {
+                                    alias: attachment.alias,
+                                    working_directory_id: attachment.working_directory_id,
+                                    capabilities: attachment.capabilities,
+                                },
+                            ]),
+                            ..Default::default()
+                        }),
+                    },
+                    Some(&owner),
+                )
+                .unwrap_err(),
+        );
+        assert!(
+            runtime
+                .lock()
+                .unwrap()
+                .worker(&worker.worker_ref)
+                .unwrap()
+                .pending_restore
+                .is_none()
+        );
+        assert_eq!(runtime.list_workers_scoped(&owner).unwrap().len(), 1);
+        // Different IDs and Worker creation remain available during cleanup.
+        runtime
+            .create_worker_scoped(
+                &owner,
+                scoped_task_request("another unrelated", "workspace-a"),
+            )
+            .unwrap();
+        let other = runtime
+            .create_working_directory(workdir_effect_request(Some("workdir-other")))
+            .unwrap_err();
+        assert!(
+            matches!(other, RuntimeError::WorkingDirectory(ref diagnostic) if diagnostic.code == "working_directory_unsupported")
+        );
+        assert_eq!(backend.cleanup_calls.load(Ordering::SeqCst), 1);
+        release.send(()).unwrap();
+        assert_workdir_busy(cleanup.join().unwrap().unwrap_err());
+        assert!(runtime.lock().unwrap().workdir_operations.is_empty());
+    }
+
+    #[test]
+    fn in_flight_creation_and_access_exclude_cleanup_including_generated_ids() {
+        for operation in ["explicit-create", "generated-create", "access"] {
+            let (runtime, backend) = runtime_and_backend();
+            let (entered, release) = WorkdirEffectGate::install(&backend.workdir_use_gate);
+            let effect_runtime = runtime.clone();
+            let effect = std::thread::spawn(move || {
+                if operation == "access" {
+                    effect_runtime.authorize_working_directory_repository_access(
+                        workdir_access_request("workdir-in-use"),
+                    )
+                } else {
+                    effect_runtime
+                        .create_working_directory(workdir_effect_request(
+                            (operation == "explicit-create").then_some("workdir-in-use"),
+                        ))
+                        .map(|_| ())
+                }
+            });
+            let id = entered
+                .recv_timeout(std::time::Duration::from_secs(5))
+                .unwrap();
+            assert!(!id.is_empty());
+            assert_workdir_busy(runtime.cleanup_working_directory(&id).unwrap_err());
+            assert_eq!(backend.cleanup_calls.load(Ordering::SeqCst), 0);
+            runtime
+                .create_worker(task_request("unrelated during effect"))
+                .unwrap();
+            release.send(()).unwrap();
+            assert!(effect.join().unwrap().is_err());
+            // Failed provider effects release their reservation for ordinary retry.
+            assert_workdir_busy(runtime.cleanup_working_directory(&id).unwrap_err());
+            assert_eq!(backend.cleanup_calls.load(Ordering::SeqCst), 1);
+        }
+    }
+
+    #[test]
+    fn logical_attachments_and_unsettled_restore_preparation_exclude_cleanup() {
+        for pending in [false, true] {
+            let (runtime, backend) = runtime_and_backend();
+            let owner = scope("workspace-a", "server-a");
+            let worker = runtime
+                .create_worker_scoped(&owner, scoped_task_request("logical owner", "workspace-a"))
+                .unwrap();
+            let id = "workdir-logical-owner";
+            let attachment = LogicalWorkdirAttachment {
+                alias: workdir::WorkdirAttachmentAlias::new("checkout").unwrap(),
+                working_directory_id: id.into(),
+                capabilities: workdir::WorkdirSessionCapabilities::ALL,
+            };
+            if pending {
+                runtime
+                    .stop_worker_scoped(&owner, &worker.worker_ref, None)
+                    .unwrap();
+                let restore = runtime.test_restore_request(&worker.worker_ref);
+                let preparation = runtime_api::WorkerRestorePreparation {
+                    workdir_attachments: Some(vec![runtime_api::LogicalWorkdirAttachment {
+                        alias: attachment.alias,
+                        working_directory_id: attachment.working_directory_id,
+                        capabilities: attachment.capabilities,
+                    }]),
+                    ..Default::default()
+                };
+                let coordinated = runtime
+                    .coordinate_worker_restore_operation(
+                        &worker.worker_ref,
+                        runtime_api::WorkerRestoreCoordinationRequest {
+                            expected_observation_token: restore.expected_observation_token,
+                            request_id: restore.request_id,
+                            preparation: Some(preparation.clone()),
+                        },
+                        Some(&owner),
+                    )
+                    .unwrap();
+                assert_eq!(coordinated, (None, Some(preparation)));
+            } else {
+                runtime
+                    .replace_worker_workdir_attachments_scoped(
+                        &owner,
+                        &worker.worker_ref,
+                        vec![attachment],
+                    )
+                    .unwrap();
+            }
+            assert!(
+                matches!(runtime.cleanup_working_directory(id).unwrap_err(), RuntimeError::InvalidRequest(message) if message.contains("assigned to worker"))
+            );
+            assert_eq!(backend.cleanup_calls.load(Ordering::SeqCst), 0);
+        }
+    }
+
+    #[test]
+    fn unowned_restore_recovery_ignores_new_preparation_for_a_workdir_being_cleaned() {
+        let (runtime, backend) = runtime_and_backend();
+        let owner = scope("workspace-a", "server-a");
+        let worker = runtime
+            .create_worker_scoped(
+                &owner,
+                scoped_task_request("recover persisted operation", "workspace-a"),
+            )
+            .unwrap();
+        runtime
+            .stop_worker_scoped(&owner, &worker.worker_ref, None)
+            .unwrap();
+        // Persisted lifecycle recovery may have no request owner (e.g. after an
+        // interrupted older admission). New preparation is not its authority.
+        runtime
+            .lock()
+            .unwrap()
+            .worker_mut(&worker.worker_ref)
+            .unwrap()
+            .pending_restore = Some(PendingWorkerRestore {
+            operation_id: WorkerLifecycleOperationId::new(),
+            mode: WorkerRestoreMode::Explicit,
+            last_settled_status: WorkerStatus::Stopped,
+        });
+        let id = "workdir-irrelevant-to-recovery";
+        let (entered, release) = WorkdirEffectGate::install(&backend.cleanup_gate);
+        let cleanup_runtime = runtime.clone();
+        let cleanup = std::thread::spawn(move || cleanup_runtime.cleanup_working_directory(id));
+        entered
+            .recv_timeout(std::time::Duration::from_secs(5))
+            .unwrap();
+        let restore = runtime.test_restore_request(&worker.worker_ref);
+        let (result, preparation) = runtime
+            .coordinate_worker_restore_operation(
+                &worker.worker_ref,
+                runtime_api::WorkerRestoreCoordinationRequest {
+                    expected_observation_token: restore.expected_observation_token,
+                    request_id: restore.request_id,
+                    preparation: Some(runtime_api::WorkerRestorePreparation {
+                        workdir_attachments: Some(vec![runtime_api::LogicalWorkdirAttachment {
+                            alias: workdir::WorkdirAttachmentAlias::new("ignored").unwrap(),
+                            working_directory_id: id.into(),
+                            capabilities: workdir::WorkdirSessionCapabilities::ALL,
+                        }]),
+                        ..Default::default()
+                    }),
+                },
+                Some(&owner),
+            )
+            .unwrap();
+        assert_eq!(result.unwrap().state, WorkerRestoreState::Accepted);
+        assert!(preparation.is_none());
+        assert!(
+            runtime
+                .lock()
+                .unwrap()
+                .worker(&worker.worker_ref)
+                .unwrap()
+                .logical_workdir_attachments
+                .is_empty()
+        );
+        release.send(()).unwrap();
+        assert_workdir_busy(cleanup.join().unwrap().unwrap_err());
+    }
+
+    #[tokio::test]
+    async fn live_workdir_session_clones_exclude_cleanup_until_close_or_final_drop() {
+        for close in [false, true] {
+            let (runtime, backend) = runtime_and_backend();
+            let root = tempfile::tempdir().unwrap();
+            let id = "workdir-live-session";
+            let provider: workdir::WorkdirSessionHandle = Arc::new(
+                workdir::LocalWorkdirSession::materialized_bound_with_environment(
+                    workdir::Workdir::new(id),
+                    root.path().to_path_buf(),
+                    root.path().to_path_buf(),
+                    manifest::SharedScope::new(manifest::Scope::writable(root.path()).unwrap()),
+                    workdir::WorkdirSessionCapabilities::ALL,
+                    BTreeMap::new(),
+                    Vec::new(),
+                ),
+            );
+            let session = workdir_session::retain_operation(
+                provider,
+                runtime.reserve_workdir_use([id]).unwrap(),
+            );
+            let clone = session.clone();
+            drop(session);
+            assert_workdir_busy(runtime.cleanup_working_directory(id).unwrap_err());
+            assert_eq!(backend.cleanup_calls.load(Ordering::SeqCst), 0);
+            if close {
+                clone.close().await.unwrap();
+                clone.close().await.unwrap();
+                // A closed wrapper may remain alive without reserving storage.
+                assert_workdir_busy(runtime.cleanup_working_directory(id).unwrap_err());
+                assert_eq!(backend.cleanup_calls.load(Ordering::SeqCst), 1);
+            } else {
+                drop(clone);
+                assert_workdir_busy(runtime.cleanup_working_directory(id).unwrap_err());
+                assert_eq!(backend.cleanup_calls.load(Ordering::SeqCst), 1);
+            }
+        }
+    }
+
+    #[test]
+    fn workdir_cleanup_panic_releases_reservation_without_poisoning_runtime() {
+        let (runtime, backend) = runtime_and_backend();
+        backend.cleanup_panics.store(true, Ordering::SeqCst);
+        assert!(
+            std::panic::catch_unwind(std::panic::AssertUnwindSafe(
+                || runtime.cleanup_working_directory("workdir-panic")
+            ))
+            .is_err()
+        );
+        assert!(runtime.lock().unwrap().workdir_operations.is_empty());
+        backend.cleanup_panics.store(false, Ordering::SeqCst);
+        runtime
+            .create_worker(task_request("after cleanup panic"))
+            .unwrap();
+        assert_workdir_busy(
+            runtime
+                .cleanup_working_directory("workdir-panic")
+                .unwrap_err(),
+        );
+        assert_eq!(backend.cleanup_calls.load(Ordering::SeqCst), 2);
+    }
+
     #[test]
     fn unknown_typed_intent_is_rejected_before_backend_dispatch() {
         let (runtime, backend) = runtime_and_backend();

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

@@ -0,0 +1,165 @@
+//! Retain a Runtime active-use reservation for a live provider session.
+//!
+//! Cloned handles share the reservation. A successful provider close releases
+//! it even while closed handles remain; a failed or cancelled close retains it
+//! until a later successful close or the final wrapper is dropped. Dropping the
+//! wrapper does not close the provider or wait for outstanding commands, so
+//! callers must keep a session handle alive while using provider-owned work.
+
+use super::WorkdirOperationGuard;
+use async_trait::async_trait;
+use std::sync::{Arc, Mutex};
+use tokio::sync::broadcast;
+use workdir::{
+    CheckoutObservation, CheckoutRequest, CheckoutResult, CheckoutSearchRequest,
+    CheckoutSearchResult, CommandEvent, CommandHandle, CommandOutput, CommandOutputRequest,
+    CommandRequest, CommandSnapshot, CommandStatus, EditRequest, EditResult, GlobRequest,
+    GlobResult, GrepRequest, GrepResult, ListRequest, ListResult, ReadBytesRequest,
+    ReadBytesResult, ReadRequest, ReadResult, StatRequest, StatResult, Workdir, WorkdirError,
+    WorkdirPath, WorkdirScopeAuthorizationRequest, WorkdirScopeOverlapRequest, WorkdirSession,
+    WorkdirSessionCapabilities, WorkdirSessionHandle, WriteRequest, WriteResult,
+};
+
+pub(super) fn retain_operation(
+    session: WorkdirSessionHandle,
+    operation: WorkdirOperationGuard,
+) -> WorkdirSessionHandle {
+    Arc::new(RetainedWorkdirSession {
+        inner: session,
+        operation: Mutex::new(Some(operation)),
+    })
+}
+
+struct RetainedWorkdirSession {
+    inner: WorkdirSessionHandle,
+    operation: Mutex<Option<WorkdirOperationGuard>>,
+}
+
+impl std::fmt::Debug for RetainedWorkdirSession {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        // Do not expose provider internals or reservation state containing host paths.
+        f.debug_struct("RetainedWorkdirSession")
+            .field("workdir", self.inner.workdir())
+            .field("capabilities", &self.inner.capabilities())
+            .finish_non_exhaustive()
+    }
+}
+
+#[async_trait]
+impl WorkdirSession for RetainedWorkdirSession {
+    fn workdir(&self) -> &Workdir {
+        self.inner.workdir()
+    }
+
+    fn capabilities(&self) -> WorkdirSessionCapabilities {
+        self.inner.capabilities()
+    }
+
+    async fn authorize_scope_path(
+        &self,
+        request: WorkdirScopeAuthorizationRequest,
+    ) -> Result<(), WorkdirError> {
+        self.inner.authorize_scope_path(request).await
+    }
+
+    async fn scope_rules_overlap(
+        &self,
+        request: WorkdirScopeOverlapRequest,
+    ) -> Result<bool, WorkdirError> {
+        self.inner.scope_rules_overlap(request).await
+    }
+
+    async fn checkout_search(
+        &self,
+        request: CheckoutSearchRequest,
+    ) -> Result<CheckoutSearchResult, WorkdirError> {
+        self.inner.checkout_search(request).await
+    }
+
+    async fn checkout_observe(
+        &self,
+        path: WorkdirPath,
+    ) -> Result<CheckoutObservation, WorkdirError> {
+        self.inner.checkout_observe(path).await
+    }
+
+    async fn checkout_execute(
+        &self,
+        request: CheckoutRequest,
+    ) -> Result<CheckoutResult, WorkdirError> {
+        self.inner.checkout_execute(request).await
+    }
+
+    async fn stat(&self, request: StatRequest) -> Result<StatResult, WorkdirError> {
+        self.inner.stat(request).await
+    }
+
+    async fn read(&self, request: ReadRequest) -> Result<ReadResult, WorkdirError> {
+        self.inner.read(request).await
+    }
+
+    async fn read_bytes(&self, request: ReadBytesRequest) -> Result<ReadBytesResult, WorkdirError> {
+        self.inner.read_bytes(request).await
+    }
+
+    async fn write(&self, request: WriteRequest) -> Result<WriteResult, WorkdirError> {
+        self.inner.write(request).await
+    }
+
+    async fn edit(&self, request: EditRequest) -> Result<EditResult, WorkdirError> {
+        self.inner.edit(request).await
+    }
+
+    async fn list(&self, request: ListRequest) -> Result<ListResult, WorkdirError> {
+        self.inner.list(request).await
+    }
+
+    async fn glob(&self, request: GlobRequest) -> Result<GlobResult, WorkdirError> {
+        self.inner.glob(request).await
+    }
+
+    async fn grep(&self, request: GrepRequest) -> Result<GrepResult, WorkdirError> {
+        self.inner.grep(request).await
+    }
+
+    async fn start_command(&self, request: CommandRequest) -> Result<CommandHandle, WorkdirError> {
+        self.inner.start_command(request).await
+    }
+
+    async fn command_status(&self, handle: CommandHandle) -> Result<CommandStatus, WorkdirError> {
+        self.inner.command_status(handle).await
+    }
+
+    async fn command_output(
+        &self,
+        request: CommandOutputRequest,
+    ) -> Result<CommandOutput, WorkdirError> {
+        self.inner.command_output(request).await
+    }
+
+    async fn cancel_command(&self, handle: CommandHandle) -> Result<(), WorkdirError> {
+        self.inner.cancel_command(handle).await
+    }
+
+    fn subscribe_command_events(&self) -> Option<broadcast::Receiver<CommandEvent>> {
+        self.inner.subscribe_command_events()
+    }
+
+    fn command_snapshot(&self) -> Vec<CommandSnapshot> {
+        self.inner.command_snapshot()
+    }
+
+    async fn close(&self) -> Result<(), WorkdirError> {
+        // Provider close must succeed before releasing the active-use reservation.
+        // No wrapper-local terminal checks may bypass the provider's own checks.
+        self.inner.close().await?;
+        let operation = self
+            .operation
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner())
+            .take();
+        // Release outside the mutex: guard Drop may acquire Runtime state locks.
+        drop(operation);
+        Ok(())
+    }
+}

crates/worker-runtime/src/worker_backend.rs

@@ -1025,7 +1025,9 @@ fn runtime_local_workdir_router(
                     command_output_scope.clone(),
                     capabilities,
                     binding.command_environment(),
-                    binding.session_resources(),
+                    binding.session_resources().map_err(|error| {
+                        format!("open Workdir attachment `{alias}` resources: {error}")
+                    })?,
                 ),
             )
             .map_err(|error| format!("bind Workdir attachment `{alias}`: {error}"))?;
@@ -3453,7 +3455,7 @@ where
             scope,
             WorkdirSessionCapabilities::ALL,
             binding.command_environment(),
-            binding.session_resources(),
+            binding.session_resources()?,
         ))
     }
 
@@ -9501,6 +9503,42 @@ mod tests {
         assert_eq!(status.summary.cleanliness.as_deref(), Some("clean"));
     }
 
+    #[cfg(target_os = "linux")]
+    #[tokio::test]
+    async fn local_sessions_exclude_cleanup_until_close_releases_resources() {
+        let runtime_base = tempfile::tempdir().unwrap();
+        let repo = create_clean_repo();
+        let materializer = RuntimeGitMaterializer::new(runtime_base.path());
+        let binding = materializer
+            .create(&working_directory_request(repo.path()))
+            .unwrap();
+        let id = &binding.working_directory.id;
+        let backend = WorkerRuntimeExecutionBackend::new(FailingFactory)
+            .unwrap()
+            .with_working_directory_materializer(materializer.clone());
+        let session = backend.open_workdir_session(id).unwrap();
+        let session_clone = session.clone();
+        drop(session);
+        assert_eq!(
+            materializer.cleanup_working_directory(id).unwrap_err().code,
+            "working_directory_cleanup_resource_busy"
+        );
+
+        let alias = WorkdirAttachmentAlias::new("checkout").unwrap();
+        let attachments = BTreeMap::from([(alias, binding.clone())]);
+        let scope = manifest::SharedScope::new(manifest::Scope::writable(binding.root()).unwrap());
+        let router = runtime_local_workdir_router(&attachments, &BTreeMap::new(), scope).unwrap();
+        session_clone.close().await.unwrap();
+        assert_eq!(
+            materializer.cleanup_working_directory(id).unwrap_err().code,
+            "working_directory_cleanup_resource_busy"
+        );
+        router.close_all().await.unwrap();
+        // Closed handles, router and passive bindings can remain alive.
+        materializer.cleanup_working_directory(id).unwrap();
+        assert!(!binding.working_directory_root().exists());
+    }
+
     #[test]
     fn spawn_failure_with_existing_working_directory_preserves_workdir() {
         let runtime_base = tempfile::tempdir().unwrap();

crates/worker-runtime/src/working_directory.rs

@@ -23,6 +23,7 @@ use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 use workdir::WorkdirSessionResource;
 
 mod cleanup;
+mod occupancy;
 
 #[cfg(all(test, target_os = "linux"))]
 pub(crate) use cleanup::tests::fail_once as fail_workdir_cleanup_once;
@@ -101,6 +102,9 @@ pub struct WorkingDirectoryBinding {
     working_directory_root: PathBuf,
     command_environment: BTreeMap<String, String>,
     session_resources: Vec<Arc<dyn WorkdirSessionResource>>,
+    occupancy: occupancy::Occupancy,
+    occupancy_id: String,
+    generation: occupancy::Generation,
 }
 
 impl WorkingDirectoryBinding {
@@ -120,8 +124,23 @@ impl WorkingDirectoryBinding {
         self.command_environment.clone()
     }
 
-    pub fn session_resources(&self) -> Vec<Arc<dyn WorkdirSessionResource>> {
-        self.session_resources.clone()
+    /// Activate resources for a local session. Cloning a binding alone is passive;
+    /// the returned lease excludes cleanup until the last resource clone drops.
+    pub fn session_resources(
+        &self,
+    ) -> Result<Vec<Arc<dyn WorkdirSessionResource>>, WorkingDirectoryDiagnostic> {
+        let lease = self
+            .occupancy
+            .acquire_session_use(&self.occupancy_id, &self.generation)?;
+        if !binding_paths_are_available(self) {
+            return Err(WorkingDirectoryDiagnostic::new(
+                "working_directory_not_found",
+                "Workdir binding is no longer available; acquire a fresh binding",
+            ));
+        }
+        let mut resources = self.session_resources.clone();
+        resources.push(Arc::new(lease));
+        Ok(resources)
     }
 
     pub fn status(&self) -> WorkingDirectoryStatus {
@@ -583,6 +602,7 @@ fn repository_access_now_epoch_seconds() -> u64 {
 pub struct RuntimeGitMaterializer {
     runtime_root: PathBuf,
     repository_access: RepositoryAccessExpiryScheduler,
+    occupancy: occupancy::Occupancy,
 }
 
 impl RuntimeGitMaterializer {
@@ -590,6 +610,7 @@ impl RuntimeGitMaterializer {
         Self {
             runtime_root: runtime_root.into(),
             repository_access: RepositoryAccessExpiryScheduler::new(),
+            occupancy: occupancy::Occupancy::default(),
         }
     }
 
@@ -702,6 +723,9 @@ impl RuntimeGitMaterializer {
             working_directory_root,
             command_environment: BTreeMap::new(),
             session_resources: Vec::new(),
+            occupancy: self.occupancy.clone(),
+            occupancy_id: working_directory_id.to_string(),
+            generation: self.occupancy.generation(&working_directory_id),
         })
     }
 
@@ -904,6 +928,7 @@ impl RuntimeGitMaterializer {
         request: &WorkingDirectoryRequest,
     ) -> Result<WorkingDirectoryBinding, WorkingDirectoryDiagnostic> {
         validate_working_directory_id(&working_directory_id)?;
+        let _creation_lease = self.occupancy.acquire_creation(&working_directory_id)?;
         Self::validate_plain_http_source(request)?;
         let request =
             self.request_with_authorized_repository_access(&working_directory_id, request)?;
@@ -911,20 +936,30 @@ impl RuntimeGitMaterializer {
         let selector = request.repository.selector.as_deref().unwrap_or("HEAD");
         let working_directory_root = self.working_directory_root(&working_directory_id);
         let checkout_root = working_directory_root.join(CHECKOUT_DIR);
-        if checkout_root.exists() {
+        if working_directory_root.symlink_metadata().is_ok() {
             return Err(WorkingDirectoryDiagnostic::new(
                 "working_directory_exists",
                 "working directory target already exists; cleanup or choose a new working_directory",
             ));
         }
-        fs::create_dir_all(&working_directory_root).map_err(|_| {
+        fs::create_dir_all(&self.runtime_root).map_err(|_| {
+            WorkingDirectoryDiagnostic::new(
+                "working_directory_create_failed",
+                "failed to create working directory storage",
+            )
+        })?;
+        // Prepare fallible access before owning a root that must not be adopted
+        // by a retry after failure.
+        let access = RepositoryCommandAccess::prepare(&self.runtime_root, &request)?;
+        // Atomically own this new root before any rollback can remove it. Do
+        // not adopt an existing partial-cleanup directory or another creator's root.
+        fs::create_dir(&working_directory_root).map_err(|_| {
             WorkingDirectoryDiagnostic::new(
                 "working_directory_create_failed",
                 "failed to create working directory; backend-private path details were omitted",
             )
         })?;
 
-        let access = RepositoryCommandAccess::prepare(&self.runtime_root, &request)?;
         let mut clone = repository_git_command(&request, access.as_ref());
         clone
             .args([
@@ -1022,6 +1057,9 @@ impl RuntimeGitMaterializer {
             working_directory_root: working_directory_root.clone(),
             command_environment: BTreeMap::new(),
             session_resources: Vec::new(),
+            occupancy: self.occupancy.clone(),
+            occupancy_id: working_directory_id.to_string(),
+            generation: self.occupancy.generation(&working_directory_id),
         };
         if let Err(error) = self.write_record(&binding) {
             let _ = fs::remove_dir_all(&working_directory_root);
@@ -1046,7 +1084,10 @@ impl WorkingDirectoryMaterializer for RuntimeGitMaterializer {
         worker_ref: &WorkerRef,
         request: &WorkingDirectoryRequest,
     ) -> Result<WorkingDirectoryBinding, WorkingDirectoryDiagnostic> {
-        let working_directory_id = Self::working_directory_id(worker_ref, &request.repository.id);
+        let working_directory_id = request
+            .backend_workdir_id
+            .clone()
+            .unwrap_or_else(|| Self::working_directory_id(worker_ref, &request.repository.id));
         self.materialize_with_working_directory_id(working_directory_id, request)
     }
 
@@ -1066,6 +1107,7 @@ impl WorkingDirectoryMaterializer for RuntimeGitMaterializer {
         request: &WorkingDirectoryRepositoryAccessRequest,
     ) -> Result<(), WorkingDirectoryDiagnostic> {
         validate_working_directory_id(&request.working_directory_id)?;
+        let _use_lease = self.occupancy.acquire_use(&request.working_directory_id)?;
         let ssh = request.materialization.ssh.as_ref().ok_or_else(|| {
             WorkingDirectoryDiagnostic::new(
                 "working_directory_remote_repository_access_required",
@@ -1187,6 +1229,8 @@ impl WorkingDirectoryMaterializer for RuntimeGitMaterializer {
         working_directory_id: &str,
         relative_cwd: Option<&str>,
     ) -> Result<WorkingDirectoryBinding, WorkingDirectoryDiagnostic> {
+        validate_working_directory_id(working_directory_id)?;
+        let _use_lease = self.occupancy.acquire_use(working_directory_id)?;
         let binding = self.checked_binding(working_directory_id, relative_cwd)?;
         self.bind_repository_access(working_directory_id, binding)
     }
@@ -1281,7 +1325,10 @@ impl WorkingDirectoryMaterializer for RuntimeGitMaterializer {
         working_directory_id: &str,
     ) -> Result<WorkingDirectoryStatus, WorkingDirectoryDiagnostic> {
         validate_working_directory_id(working_directory_id)?;
-        cleanup::remove(self, working_directory_id)
+        let _cleanup_lease = self.occupancy.acquire_cleanup(working_directory_id)?;
+        let status = cleanup::remove(self, working_directory_id)?;
+        self.occupancy.invalidate_generation(working_directory_id);
+        Ok(status)
     }
 
     fn cleanup(&self, binding: &WorkingDirectoryBinding) -> Result<(), WorkingDirectoryDiagnostic> {
@@ -2878,7 +2925,7 @@ fn sanitize_path_component(value: &str) -> String {
     }
 }
 
-fn next_working_directory_id(_repository_id: &str) -> String {
+pub(crate) fn next_working_directory_id(_repository_id: &str) -> String {
     let now = SystemTime::now()
         .duration_since(UNIX_EPOCH)
         .map(|duration| duration.as_millis() as u64)
@@ -3933,7 +3980,8 @@ mod tests {
         #[cfg(target_os = "linux")]
         {
             // Real SSH reauthorization updated access evidence repeatedly, but
-            // did not replace this clean materialization or require manual recovery.
+            // did not replace this materialization or require manual recovery.
+            // Status does not inspect content; cleanup still verifies it below.
             assert_eq!(
                 restored
                     .working_directory_status(&id)
@@ -3941,7 +3989,7 @@ mod tests {
                     .summary
                     .cleanliness
                     .as_deref(),
-                Some("clean")
+                Some("unknown")
             );
             restored.cleanup_working_directory(&id).unwrap();
             assert!(!restored.working_directory_root(&id).exists());
@@ -4443,6 +4491,307 @@ mod tests {
         assert!(root.exists());
     }
 
+    fn materialization_context(
+        ssh: Option<RepositorySshMaterializationAccess>,
+    ) -> crate::catalog::RepositoryMaterializationContext {
+        crate::catalog::RepositoryMaterializationContext {
+            workspace_id: "workspace-1".to_string(),
+            runtime_id: "runtime-1".to_string(),
+            operation_id: "operation-1".to_string(),
+            config_revision: 7,
+            config_projection_digest: "sha256:projection".to_string(),
+            ssh,
+        }
+    }
+
+    #[cfg(target_os = "linux")]
+    #[test]
+    fn live_session_resources_block_cleanup_until_last_clone_drops() {
+        let repo = create_clean_repo();
+        let runtime_root = tempfile::tempdir().unwrap();
+        let materializer = RuntimeGitMaterializer::new(runtime_root.path());
+        let binding = materializer.create(&request(repo.path())).unwrap();
+        let id = &binding.working_directory.id;
+        // Read bindings and materializer clones must share the same admission.
+        let restored = materializer.bind_working_directory(id, None).unwrap();
+        let resources = restored.session_resources().unwrap();
+        let resource_clone = resources.clone();
+        let second_resources = binding.session_resources().unwrap();
+        let clone = materializer.clone();
+
+        for _ in 0..2 {
+            assert_eq!(
+                clone.cleanup_working_directory(id).unwrap_err().code,
+                "working_directory_cleanup_resource_busy"
+            );
+        }
+        assert!(binding.root().join("README.md").exists());
+        drop(resources);
+        drop(second_resources);
+        assert_eq!(
+            clone.cleanup_working_directory(id).unwrap_err().code,
+            "working_directory_cleanup_resource_busy"
+        );
+        drop(resource_clone);
+        clone.cleanup_working_directory(id).unwrap();
+        // Both bindings are still alive and passive after resource release.
+        assert!(!binding.working_directory_root().exists());
+        assert!(!restored.root().exists());
+    }
+
+    #[test]
+    fn active_cleanup_rejects_creation_and_access_before_effects() {
+        let repo = tempfile::tempdir().unwrap();
+        let runtime_root = tempfile::tempdir().unwrap();
+        let materializer = RuntimeGitMaterializer::new(runtime_root.path());
+        let id = "workdir-cleanup-active";
+        let cleanup = materializer.occupancy.acquire_cleanup(id).unwrap();
+        let mut create_request = request(repo.path());
+        create_request.backend_workdir_id = Some(id.to_string());
+        let access_request = WorkingDirectoryRepositoryAccessRequest {
+            working_directory_id: id.to_string(),
+            materialization: materialization_context(Some(repository_ssh_access())),
+        };
+
+        assert_eq!(
+            materializer.create(&create_request).unwrap_err().code,
+            "working_directory_cleanup_resource_busy"
+        );
+        assert_eq!(
+            materializer
+                .materialize(&worker_ref(1), &create_request)
+                .unwrap_err()
+                .code,
+            "working_directory_cleanup_resource_busy"
+        );
+        assert_eq!(
+            materializer
+                .authorize_repository_access(&access_request)
+                .unwrap_err()
+                .code,
+            "working_directory_cleanup_resource_busy"
+        );
+        assert_eq!(
+            materializer
+                .bind_working_directory(id, None)
+                .unwrap_err()
+                .code,
+            "working_directory_cleanup_resource_busy"
+        );
+        assert!(!materializer.working_directory_root(id).exists());
+        assert_eq!(materializer.repository_access.pending_count(), 0);
+        drop(cleanup);
+        // The cleanup reservation was transient, not a persisted lifecycle.
+        assert_ne!(
+            materializer.create(&create_request).unwrap_err().code,
+            "working_directory_cleanup_resource_busy"
+        );
+    }
+
+    #[cfg(target_os = "linux")]
+    #[test]
+    fn active_cleanup_rejects_existing_binding_resources_but_not_other_ids() {
+        let repo = create_clean_repo();
+        let runtime_root = tempfile::tempdir().unwrap();
+        let materializer = RuntimeGitMaterializer::new(runtime_root.path());
+        let binding = materializer.create(&request(repo.path())).unwrap();
+        let id = &binding.working_directory.id;
+        let cleanup = materializer.occupancy.acquire_cleanup(id).unwrap();
+        assert_eq!(
+            binding.session_resources().unwrap_err().code,
+            "working_directory_cleanup_resource_busy"
+        );
+        assert_eq!(
+            materializer.cleanup_working_directory(id).unwrap_err().code,
+            "working_directory_cleanup_resource_busy"
+        );
+        let other = materializer.create(&request(repo.path())).unwrap();
+        let other_resources = other.session_resources().unwrap();
+        drop(other_resources);
+        materializer
+            .cleanup_working_directory(&other.working_directory.id)
+            .unwrap();
+        drop(cleanup);
+        let resources = binding.session_resources().unwrap();
+        drop(resources);
+        materializer.cleanup_working_directory(id).unwrap();
+    }
+
+    #[cfg(target_os = "linux")]
+    #[test]
+    fn cleanup_error_releases_lease_for_session_and_safe_retry() {
+        let repo = create_clean_repo();
+        let runtime_root = tempfile::tempdir().unwrap();
+        let materializer = RuntimeGitMaterializer::new(runtime_root.path());
+        let binding = materializer.create(&request(repo.path())).unwrap();
+        let changed = binding.root().join("untracked.txt");
+        fs::write(&changed, "preserve me\n").unwrap();
+        assert_eq!(
+            materializer
+                .cleanup_working_directory(&binding.working_directory.id)
+                .unwrap_err()
+                .code,
+            "working_directory_cleanup_changes_present"
+        );
+        let resources = binding.session_resources().unwrap();
+        fs::remove_file(changed).unwrap();
+        drop(resources);
+        materializer
+            .cleanup_working_directory(&binding.working_directory.id)
+            .unwrap();
+    }
+
+    #[cfg(target_os = "linux")]
+    #[test]
+    fn removed_binding_cannot_activate_against_recreated_same_id() {
+        let repo = create_clean_repo();
+        let runtime_root = tempfile::tempdir().unwrap();
+        let materializer = RuntimeGitMaterializer::new(runtime_root.path());
+        let mut create = request(repo.path());
+        create.backend_workdir_id = Some("workdir-reused-id".into());
+        let old = materializer.create(&create).unwrap();
+        materializer
+            .cleanup_working_directory(&old.working_directory.id)
+            .unwrap();
+        assert_eq!(
+            old.session_resources().unwrap_err().code,
+            "working_directory_not_found"
+        );
+        let fresh = materializer.create(&create).unwrap();
+        assert_eq!(
+            old.session_resources().unwrap_err().code,
+            "working_directory_not_found"
+        );
+        let resources = fresh.session_resources().unwrap();
+        assert_eq!(
+            materializer
+                .cleanup_working_directory(&fresh.working_directory.id)
+                .unwrap_err()
+                .code,
+            "working_directory_cleanup_resource_busy"
+        );
+        drop(resources);
+        materializer
+            .cleanup_working_directory(&fresh.working_directory.id)
+            .unwrap();
+    }
+
+    #[test]
+    fn same_id_creation_is_exclusive_and_never_adopts_existing_partial_root() {
+        let repo = tempfile::tempdir().unwrap();
+        let runtime_root = tempfile::tempdir().unwrap();
+        let materializer = RuntimeGitMaterializer::new(runtime_root.path());
+        let id = "workdir-creation-owner";
+        let mut create = request(repo.path());
+        create.backend_workdir_id = Some(id.into());
+        let creation = materializer.occupancy.acquire_creation(id).unwrap();
+        assert_eq!(
+            materializer.clone().create(&create).unwrap_err().code,
+            "working_directory_cleanup_resource_busy"
+        );
+        assert_eq!(
+            materializer
+                .bind_working_directory(id, None)
+                .unwrap_err()
+                .code,
+            "working_directory_cleanup_resource_busy"
+        );
+        assert_eq!(
+            materializer.cleanup_working_directory(id).unwrap_err().code,
+            "working_directory_cleanup_resource_busy"
+        );
+        assert!(!materializer.working_directory_root(id).exists());
+        drop(creation);
+        let partial = materializer.working_directory_root(id);
+        fs::create_dir(&partial).unwrap();
+        fs::write(
+            partial.join("owner-content"),
+            "preserve partial removal state",
+        )
+        .unwrap();
+        assert_eq!(
+            materializer.create(&create).unwrap_err().code,
+            "working_directory_exists"
+        );
+        assert_eq!(
+            fs::read_to_string(partial.join("owner-content")).unwrap(),
+            "preserve partial removal state"
+        );
+    }
+
+    #[test]
+    fn failed_access_preparation_does_not_strand_a_creation_root() {
+        let repo = tempfile::tempdir().unwrap();
+        let runtime_root = tempfile::tempdir().unwrap();
+        let materializer = RuntimeGitMaterializer::new(runtime_root.path());
+        let mut create = request(repo.path());
+        create.backend_workdir_id = Some("workdir-access-setup-failure".into());
+        let mut ssh = repository_ssh_access();
+        ssh.known_hosts_entry =
+            crate::catalog::SensitiveString::new("example.test ssh-ed25519 placeholder");
+        create.repository.source = server_api::RepositorySource {
+            kind: server_api::RepositorySourceKind::Ssh,
+            uri: ssh.repository_uri.clone(),
+        };
+        create.materialization = Some(materialization_context(Some(ssh)));
+        // A local storage failure stops access setup before any SSH transport or
+        // agent can start. The materialization must remain retryable, not adopted.
+        fs::write(
+            runtime_root.path().join(REPOSITORY_ACCESS_DIR),
+            "storage obstruction",
+        )
+        .unwrap();
+        assert_eq!(
+            materializer.create(&create).unwrap_err().code,
+            "working_directory_repository_access_setup_failed"
+        );
+        assert!(
+            !materializer
+                .working_directory_root(create.backend_workdir_id.as_deref().unwrap())
+                .exists()
+        );
+    }
+
+    #[test]
+    fn failed_creation_and_access_release_use_leases() {
+        let repo = tempfile::tempdir().unwrap();
+        let runtime_root = tempfile::tempdir().unwrap();
+        let materializer = RuntimeGitMaterializer::new(runtime_root.path());
+        let id = "workdir-effect-errors";
+        let mut create_request = request(repo.path());
+        create_request.backend_workdir_id = Some(id.to_string());
+        // Validation fails after admission but before effects.
+        create_request.repository.provider = "unsupported".to_string();
+        assert_eq!(
+            materializer.create(&create_request).unwrap_err().code,
+            "working_directory_repository_provider_unsupported"
+        );
+        drop(materializer.occupancy.acquire_cleanup(id).unwrap());
+        // A clone failure has already created the target and must also release
+        // admission after rolling its filesystem effects back.
+        create_request.repository.provider = "git".to_string();
+        assert_eq!(
+            materializer.create(&create_request).unwrap_err().code,
+            "working_directory_repository_clone_failed"
+        );
+        assert!(!materializer.working_directory_root(id).exists());
+        drop(materializer.occupancy.acquire_cleanup(id).unwrap());
+        let access_request = WorkingDirectoryRepositoryAccessRequest {
+            working_directory_id: id.to_string(),
+            materialization: materialization_context(None),
+        };
+        assert_eq!(
+            materializer
+                .authorize_repository_access(&access_request)
+                .unwrap_err()
+                .code,
+            "working_directory_remote_repository_access_required"
+        );
+        drop(materializer.occupancy.acquire_cleanup(id).unwrap());
+        assert!(materializer.bind_working_directory(id, None).is_err());
+        drop(materializer.occupancy.acquire_cleanup(id).unwrap());
+    }
+
     #[test]
     fn cleanup_working_directory_removes_clone_and_record() {
         let repo = create_clean_repo();

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

@@ -76,19 +76,11 @@ pub(super) fn remaining_status(
     }
     let mut summary = saved.working_directory.status_summary();
     summary.status = WorkingDirectoryStatusKind::CleanupPending;
-    summary.cleanliness = Some(
-        match check_mounts(
-            &materializer.working_directory_root(id),
-            &fs::read("/proc/self/mountinfo").ok()?,
-        )
-        .and_then(|()| retry_inventory(materializer, id, &tree, &saved))
-        {
-            Ok(_) => "clean",
-            Err(error) if error.code == "working_directory_cleanup_changes_present" => "dirty",
-            Err(_) => "unknown",
-        }
-        .to_string(),
-    );
+    // Observation is not removal admission: do not inventory or hash the
+    // surviving checkout just to list it. The witness can be stale, so neither
+    // its old cleanliness nor a successful identity check proves current content
+    // is safe to remove. Actual retries still perform the full safety checks.
+    summary.cleanliness = Some("unknown".to_string());
     Some(WorkingDirectoryStatus { summary })
 }
 
@@ -974,7 +966,7 @@ pub(super) mod tests {
     }
 
     #[test]
-    fn cleanup_pending_rechecks_current_cleanliness_and_retries_without_state_edit() {
+    fn cleanup_pending_observation_is_unknown_and_retry_checks_safety() {
         let (_runtime, materializer, binding) = fixture();
         let id = &binding.working_directory.id;
         fail_once(&materializer, id);
@@ -983,12 +975,54 @@ pub(super) mod tests {
             status.summary.status,
             WorkingDirectoryStatusKind::CleanupPending
         );
-        assert_eq!(status.summary.cleanliness.as_deref(), Some("clean"));
+        assert_eq!(status.summary.cleanliness.as_deref(), Some("unknown"));
         let removed = materializer.cleanup_working_directory(id).unwrap();
         assert_eq!(removed.summary.status, WorkingDirectoryStatusKind::NotFound);
         assert!(!binding.working_directory_root.exists());
     }
 
+    #[test]
+    fn cleanup_pending_observation_does_not_open_checkout_content() {
+        use std::os::fd::{AsRawFd, FromRawFd};
+        use std::os::unix::ffi::OsStrExt;
+
+        let (_runtime, materializer, binding) = fixture();
+        let id = &binding.working_directory.id;
+        fail_once(&materializer, id);
+
+        // Observe actual file opens rather than a timing threshold or a large
+        // fixture. Inventory would open this surviving tracked file to hash it.
+        // SAFETY: inotify_init1 has no pointer arguments.
+        let fd = unsafe { libc::inotify_init1(libc::IN_NONBLOCK | libc::IN_CLOEXEC) };
+        assert!(fd >= 0, "{}", io::Error::last_os_error());
+        // SAFETY: fd is a freshly created descriptor, owned only by this File.
+        let mut events = unsafe { fs::File::from_raw_fd(fd) };
+        let path =
+            std::ffi::CString::new(binding.root.join("README.md").as_os_str().as_bytes()).unwrap();
+        // SAFETY: the descriptor is live and path is a valid terminated string.
+        let watch =
+            unsafe { libc::inotify_add_watch(events.as_raw_fd(), path.as_ptr(), libc::IN_OPEN) };
+        assert!(watch >= 0, "{}", io::Error::last_os_error());
+
+        let detail = materializer.working_directory_status(id).unwrap();
+        let listed = materializer.list_working_directories().unwrap();
+        let mut buffer = [0u8; 4096];
+        assert_eq!(
+            events
+                .read(&mut buffer)
+                .expect_err("status observation opened surviving checkout content")
+                .kind(),
+            io::ErrorKind::WouldBlock
+        );
+        assert_eq!(
+            detail.summary.status,
+            WorkingDirectoryStatusKind::CleanupPending
+        );
+        assert_eq!(detail.summary.cleanliness.as_deref(), Some("unknown"));
+        assert_eq!(listed.len(), 1);
+        assert_eq!(listed[0].summary, detail.summary);
+    }
+
     #[test]
     fn partial_deletion_missing_checkout_and_restart_finish_with_durable_content_authority() {
         for removed in [
@@ -1021,7 +1055,7 @@ pub(super) mod tests {
                 retained.summary.cleanup_target.as_ref(),
                 Some(&binding.working_directory.cleanup_target)
             );
-            assert_eq!(retained.summary.cleanliness.as_deref(), Some("clean"));
+            assert_eq!(retained.summary.cleanliness.as_deref(), Some("unknown"));
             assert_eq!(
                 restarted.list_working_directories().unwrap()[0].summary,
                 retained.summary
@@ -1074,7 +1108,7 @@ pub(super) mod tests {
                     .summary
                     .cleanliness
                     .as_deref(),
-                Some("dirty")
+                Some("unknown")
             );
             assert_eq!(
                 materializer.cleanup_working_directory(id).unwrap_err().code,
@@ -1117,7 +1151,7 @@ pub(super) mod tests {
                     .summary
                     .cleanliness
                     .as_deref(),
-                Some("clean"),
+                Some("unknown"),
                 "{resolution}"
             );
             // Renewal itself can fail at unlink. Its new witness must survive a

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

@@ -0,0 +1,196 @@
+//! Transient, clone-shared admission for local filesystem effects. Bindings are
+//! passive; only effects and activated session resources own leases. The mutex
+//! protects admission bookkeeping only, never filesystem work.
+use super::WorkingDirectoryDiagnostic;
+use std::collections::HashMap;
+use std::sync::{Arc, Mutex};
+
+#[derive(Clone, Debug, Default)]
+pub(super) struct Occupancy {
+    entries: Arc<Mutex<HashMap<String, Entry>>>,
+    generations: Arc<Mutex<HashMap<String, Arc<()>>>>,
+}
+
+/// Passive bindings identify a materialization generation, not merely its ID.
+#[derive(Clone, Debug)]
+pub(super) struct Generation(Arc<()>);
+
+#[derive(Debug)]
+enum Entry {
+    Use(usize),
+    Create,
+    Cleanup,
+}
+
+#[derive(Debug)]
+pub(super) struct Lease {
+    occupancy: Occupancy,
+    id: String,
+}
+
+fn busy() -> WorkingDirectoryDiagnostic {
+    WorkingDirectoryDiagnostic::new(
+        "working_directory_cleanup_resource_busy",
+        "A filesystem resource is busy or changed during removal. Release the resource and retry.",
+    )
+}
+
+impl Occupancy {
+    pub(super) fn acquire_use(&self, id: &str) -> Result<Lease, WorkingDirectoryDiagnostic> {
+        let mut entries = self
+            .entries
+            .lock()
+            .unwrap_or_else(|error| error.into_inner());
+        match entries.get_mut(id) {
+            Some(Entry::Cleanup | Entry::Create) => return Err(busy()),
+            Some(Entry::Use(count)) => *count += 1,
+            None => {
+                entries.insert(id.to_string(), Entry::Use(1));
+            }
+        }
+        Ok(Lease {
+            occupancy: self.clone(),
+            id: id.to_string(),
+        })
+    }
+
+    pub(super) fn generation(&self, id: &str) -> Generation {
+        let mut generations = self
+            .generations
+            .lock()
+            .unwrap_or_else(|error| error.into_inner());
+        Generation(
+            generations
+                .entry(id.to_string())
+                .or_insert_with(|| Arc::new(()))
+                .clone(),
+        )
+    }
+
+    // Called only after successful removal, while exclusive cleanup is held.
+    pub(super) fn invalidate_generation(&self, id: &str) {
+        self.generations
+            .lock()
+            .unwrap_or_else(|error| error.into_inner())
+            .remove(id);
+    }
+
+    pub(super) fn acquire_session_use(
+        &self,
+        id: &str,
+        generation: &Generation,
+    ) -> Result<Lease, WorkingDirectoryDiagnostic> {
+        let lease = self.acquire_use(id)?;
+        let valid = self
+            .generations
+            .lock()
+            .unwrap_or_else(|error| error.into_inner())
+            .get(id)
+            .is_some_and(|current| Arc::ptr_eq(current, &generation.0));
+        if !valid {
+            return Err(WorkingDirectoryDiagnostic::new(
+                "working_directory_not_found",
+                "Workdir binding is stale; acquire a fresh binding before opening a session",
+            ));
+        }
+        Ok(lease)
+    }
+
+    pub(super) fn acquire_creation(&self, id: &str) -> Result<Lease, WorkingDirectoryDiagnostic> {
+        let mut entries = self
+            .entries
+            .lock()
+            .unwrap_or_else(|error| error.into_inner());
+        if entries.contains_key(id) {
+            return Err(busy());
+        }
+        entries.insert(id.to_string(), Entry::Create);
+        Ok(Lease {
+            occupancy: self.clone(),
+            id: id.to_string(),
+        })
+    }
+
+    pub(super) fn acquire_cleanup(&self, id: &str) -> Result<Lease, WorkingDirectoryDiagnostic> {
+        let mut entries = self
+            .entries
+            .lock()
+            .unwrap_or_else(|error| error.into_inner());
+        if entries.contains_key(id) {
+            return Err(busy());
+        }
+        entries.insert(id.to_string(), Entry::Cleanup);
+        Ok(Lease {
+            occupancy: self.clone(),
+            id: id.to_string(),
+        })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn use_and_cleanup_leases_release_during_unwind() {
+        let occupancy = Occupancy::default();
+        for operation in ["use", "create", "cleanup"] {
+            let result = std::panic::catch_unwind(|| {
+                let _lease = match operation {
+                    "cleanup" => occupancy.acquire_cleanup("workdir-panic").unwrap(),
+                    "create" => occupancy.acquire_creation("workdir-panic").unwrap(),
+                    _ => occupancy.acquire_use("workdir-panic").unwrap(),
+                };
+                panic!("effect failed after admission");
+            });
+            assert!(result.is_err());
+            drop(occupancy.acquire_use("workdir-panic").unwrap());
+            drop(occupancy.acquire_cleanup("workdir-panic").unwrap());
+        }
+    }
+
+    #[test]
+    fn simultaneous_use_and_cleanup_admit_only_one_direction() {
+        let occupancy = Occupancy::default();
+        let barrier = Arc::new(std::sync::Barrier::new(2));
+        std::thread::scope(|scope| {
+            let use_occupancy = occupancy.clone();
+            let use_barrier = barrier.clone();
+            let use_thread = scope.spawn(move || {
+                use_barrier.wait();
+                let result = use_occupancy.acquire_use("workdir-race");
+                // Keep the winner's lease until both admission attempts finish.
+                use_barrier.wait();
+                result
+            });
+            barrier.wait();
+            let cleanup = occupancy.acquire_cleanup("workdir-race");
+            barrier.wait();
+            let use_result = use_thread.join().unwrap();
+            assert_ne!(use_result.is_ok(), cleanup.is_ok());
+            let error = if let Err(error) = use_result {
+                error
+            } else {
+                cleanup.unwrap_err()
+            };
+            assert_eq!(error.code, "working_directory_cleanup_resource_busy");
+        });
+        drop(occupancy.acquire_cleanup("workdir-race").unwrap());
+    }
+}
+
+impl Drop for Lease {
+    fn drop(&mut self) {
+        let mut entries = self
+            .occupancy
+            .entries
+            .lock()
+            .unwrap_or_else(|error| error.into_inner());
+        match entries.get_mut(&self.id) {
+            Some(Entry::Use(count)) if *count > 1 => *count -= 1,
+            _ => {
+                entries.remove(&self.id);
+            }
+        }
+    }
+}