From e96fde0632aac4d34f36883c1ad5773c5ae59767 Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 2 Sep 2026 00:01:16 +0900 Subject: [PATCH] feat: add durable Workdir removal authority --- .../src/feature/builtin/manage_workdir.rs | 64 +- crates/workspace-api/src/lib.rs | 28 + crates/workspace-server/src/lib.rs | 1 + crates/workspace-server/src/server.rs | 593 ++++++++--- crates/workspace-server/src/store.rs | 90 +- .../workspace-server/src/workdir_removal.rs | 952 ++++++++++++++++++ 6 files changed, 1522 insertions(+), 206 deletions(-) create mode 100644 crates/workspace-server/src/workdir_removal.rs diff --git a/crates/worker/src/feature/builtin/manage_workdir.rs b/crates/worker/src/feature/builtin/manage_workdir.rs index bd138eb7..74419af6 100644 --- a/crates/worker/src/feature/builtin/manage_workdir.rs +++ b/crates/worker/src/feature/builtin/manage_workdir.rs @@ -23,8 +23,9 @@ use workdir::{ use workspace_api::{ WorkingDirectoryCreateRequest as WorkdirCreateRequest, WorkingDirectoryCreateResponse as WorkdirCreateResponse, - WorkingDirectoryDetailResponse as WorkdirDetailResponse, WorkingDirectoryListResponse as WorkdirListResponse, + WorkingDirectoryRemovalRequest as WorkdirRemovalRequest, + WorkingDirectoryRemovalResponse as WorkdirRemovalResponse, }; use crate::feature::{ @@ -51,7 +52,7 @@ const LIST_DESCRIPTION: &str = "List persistent Workdirs in the current Workspac const CREATE_DESCRIPTION: &str = "Materialize a persistent Workdir on a selected Runtime from a Workspace repository and optional selector. This does not change this Worker's attachment; use WorkdirAttach explicitly after creation."; const ATTACH_DESCRIPTION: &str = "Attach this Worker to one existing Workdir. The Backend enforces one active Workdir per Worker and one active Worker per Workdir, then opens an ephemeral operation session."; const DETACH_DESCRIPTION: &str = "Detach this Worker from its active Workdir and release Workdir occupancy. Any ephemeral operation session is closed."; -const DELETE_DESCRIPTION: &str = "Delete one persistent Workdir by id through Backend Workspace API authority. Occupied, blocked, or dirty Workdirs requiring confirmation are rejected."; +const DELETE_DESCRIPTION: &str = "Request removal of one persistent Workdir by id through durable Backend Workspace authority. The input includes only the Workdir id and a bounded reason. The result reports removed, retained, or attention_required without exposing operation-table or provider internals."; #[derive(Clone, Debug)] pub struct ManageWorkdirFeature { @@ -484,12 +485,21 @@ impl WorkspaceHttpWorkdirBackend { )?; let workspace_id = encode_path_segment(self.workspace_id()?); let workdir_path = encode_path_segment(workdir_id); - let response = self.execute_json::(WorkspaceRequest { - method: WorkspaceRequestMethod::Delete, - path: format!("/api/w/{workspace_id}/working-directories/{workdir_path}"), - body: None, - })?; - workdir_output(format!("Deleted Workdir {workdir_id}"), &response) + let response = self.execute_json::(WorkspaceRequest::json( + WorkspaceRequestMethod::Delete, + format!("/api/w/{workspace_id}/working-directories/{workdir_path}"), + serde_json::to_string(&WorkdirRemovalRequest { + reason: validate_delete_reason(&input.reason)?.to_string(), + }) + .map_err(decode_error)?, + ))?; + workdir_output( + format!( + "Workdir {workdir_id} removal disposition: {:?}", + response.disposition + ), + &response, + ) } fn execute_json Deserialize<'de>>( @@ -611,6 +621,17 @@ fn validate_identity<'a>( Ok(value) } +fn validate_delete_reason(reason: &str) -> Result<&str, ToolError> { + let reason = reason.trim(); + if reason.is_empty() || reason.len() > 500 || reason.chars().any(char::is_control) { + return Err(ToolError::InvalidArgument( + "WorkdirDelete reason must be non-empty, contain no control characters, and be at most 500 bytes" + .to_string(), + )); + } + Ok(reason) +} + fn validate_optional_selector(selector: Option) -> Result, ToolError> { let Some(selector) = selector else { return Ok(None); @@ -689,9 +710,10 @@ fn delete_schema() -> serde_json::Value { json!({ "type": "object", "additionalProperties": false, - "required": ["working_directory_id"], + "required": ["working_directory_id", "reason"], "properties": { - "working_directory_id": {"type": "string", "minLength": 1} + "working_directory_id": {"type": "string", "minLength": 1}, + "reason": {"type": "string", "minLength": 1, "maxLength": 500} } }) } @@ -736,6 +758,7 @@ struct WorkdirAttachmentResponse { #[serde(deny_unknown_fields)] struct WorkdirDeleteInput { working_directory_id: String, + reason: String, } #[cfg(test)] @@ -959,7 +982,10 @@ mod tests { assert!(create["properties"].get("session_id").is_none()); assert_eq!(attach_schema()["required"], json!(["workdir_id"])); assert!(attach_schema()["properties"].get("session_id").is_none()); - assert_eq!(delete_schema()["required"], json!(["working_directory_id"])); + assert_eq!( + delete_schema()["required"], + json!(["working_directory_id", "reason"]) + ); } #[test] @@ -999,15 +1025,9 @@ mod tests { "attached": false })), response(json!({ - "workspace_id": "workspace/test", - "runtime_id": "runtime/one", - "item": { - "working_directory_id": "wd-created", - "repository_id": "main", - "materializer_kind": "local_git_worktree", - "status": "not_found" - }, - "diagnostics": [] + "working_directory_id": "wd-created", + "disposition": "removed", + "retryable": false })), ])); let backend = WorkspaceHttpWorkdirBackend::new(client.clone()); @@ -1052,6 +1072,7 @@ mod tests { backend .delete(WorkdirDeleteInput { working_directory_id: "wd-created".to_string(), + reason: "remove stale Workdir".to_string(), }) .unwrap(); @@ -1087,6 +1108,9 @@ mod tests { "/api/w/workspace%2Ftest/working-directories/wd-created" ); assert_eq!(requests[4].method, WorkspaceRequestMethod::Delete); + let body: serde_json::Value = + serde_json::from_str(requests[4].body.as_deref().unwrap()).unwrap(); + assert_eq!(body, json!({"reason": "remove stale Workdir"})); } #[tokio::test] diff --git a/crates/workspace-api/src/lib.rs b/crates/workspace-api/src/lib.rs index 6860c5dc..8570dd26 100644 --- a/crates/workspace-api/src/lib.rs +++ b/crates/workspace-api/src/lib.rs @@ -397,6 +397,34 @@ pub struct WorkingDirectoryCleanupTarget { pub repository_id: String, } +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +#[serde(deny_unknown_fields)] +pub struct WorkingDirectoryRemovalRequest { + pub reason: String, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +#[serde(rename_all = "snake_case")] +pub enum WorkingDirectoryRemovalDisposition { + Removed, + Retained, + AttentionRequired, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +#[cfg_attr(feature = "typescript", ts(optional_fields = nullable))] +#[serde(deny_unknown_fields)] +pub struct WorkingDirectoryRemovalResponse { + pub working_directory_id: String, + pub disposition: WorkingDirectoryRemovalDisposition, + pub retryable: bool, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub failure_category: Option, +} + /// Durable Workspace occupancy projection for one Workdir. #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] #[cfg_attr(feature = "typescript", derive(ts_rs::TS))] diff --git a/crates/workspace-server/src/lib.rs b/crates/workspace-server/src/lib.rs index 30b301d6..87d47955 100644 --- a/crates/workspace-server/src/lib.rs +++ b/crates/workspace-server/src/lib.rs @@ -30,6 +30,7 @@ pub mod server; pub mod skills; pub mod store; pub mod workdir_create_operations; +mod workdir_removal; pub mod worker_source; pub mod workspace_catalog; mod workspace_subscription; diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index e043a97d..2498265c 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -70,10 +70,11 @@ use workspace_api::{ WorkingDirectoryCreateResponse as BrowserWorkingDirectoryCreateResponse, WorkingDirectoryDetailResponse as BrowserWorkingDirectoryDetailResponse, WorkingDirectoryListResponse as BrowserWorkingDirectoryListResponse, - WorkspaceCatalogListResponse, WorkspaceCreateResponse, WorkspaceExtensionPointState, - WorkspaceExtensionPoints, WorkspacePermissionSummary, WorkspaceRepositoryRecord, - WorkspaceResponse, WorkspaceRuntimeResource, WorkspaceSummary, WorkspaceWorkerDiscoveryItem, - WorkspaceWorkerDiscoveryPage, WorkspaceWorkerSubject, + WorkingDirectoryRemovalDisposition, WorkingDirectoryRemovalRequest, + WorkingDirectoryRemovalResponse, WorkspaceCatalogListResponse, WorkspaceCreateResponse, + WorkspaceExtensionPointState, WorkspaceExtensionPoints, WorkspacePermissionSummary, + WorkspaceRepositoryRecord, WorkspaceResponse, WorkspaceRuntimeResource, WorkspaceSummary, + WorkspaceWorkerDiscoveryItem, WorkspaceWorkerDiscoveryPage, WorkspaceWorkerSubject, }; use crate::auth::{ @@ -140,6 +141,10 @@ use crate::store::{ WorkerControlGrantRecord, WorkerRegistryRecord, WorkerWorkdirLinkRecord, WorkspaceRecord, WorkspaceResourceKind, }; +use crate::workdir_removal::{ + WorkdirRemovalDisposition, WorkdirRemovalOperation, WorkdirRemovalOperationState, + workdir_removal_intent, +}; use crate::workspace_catalog::{WorkspaceCatalogService, WorkspaceCreateRequest}; use crate::{Error, Result}; use worker_runtime::catalog::{ @@ -513,6 +518,7 @@ pub struct WorkspaceApi { workdir_sessions: Arc>, workdir_session_locks: Arc>>>>, worker_remove_locks: Arc>>>>, + workdir_remove_locks: Arc>>>>, worker_control_locks: Arc>>>>, } @@ -1633,6 +1639,7 @@ impl WorkspaceApi { workdir_sessions: Arc::new(Mutex::new(WorkdirSessionRegistry::default())), workdir_session_locks: Arc::new(Mutex::new(HashMap::new())), worker_remove_locks: Arc::new(Mutex::new(HashMap::new())), + workdir_remove_locks: Arc::new(Mutex::new(HashMap::new())), worker_control_locks: Arc::new(Mutex::new(HashMap::new())), }; if let Some(dispatcher) = worker_remove_dispatcher { @@ -1640,6 +1647,7 @@ impl WorkspaceApi { .install_executor(Arc::new(WorkspaceWorkerRemoveExecutor::new(&api))) .map_err(|message| Error::Config(message.to_string()))?; } + recover_workdir_removals(&api)?; Ok(api) } @@ -8979,9 +8987,34 @@ async fn scoped_runtime_working_directory_detail( async fn scoped_cleanup_runtime_working_directory( State(api): State, AxumPath(path): AxumPath, -) -> ApiResult> { + worker_source: Option>, + request_actor: Option>, + Json(request): Json, +) -> ApiResult> { validate_workspace_scope(&api, &path.workspace_id)?; - cleanup_working_directory_for_runtime(api, &path.runtime_id, &path.working_directory_id) + let registered_runtime = registered_workdir_runtime_id(&api, &path.working_directory_id)?; + if registered_runtime != path.runtime_id { + return Err(ApiError::from(Error::WorkspacePermissionDenied( + "Workdir does not belong to the requested Runtime".to_string(), + ))); + } + let source_actor = if let Some(Extension(source)) = worker_source { + format!("worker:{}:{}", source.runtime_id, source.worker_id) + } else if let Some(Extension(actor)) = request_actor { + format!("account:{}", actor.account_id) + } else { + return Err(ApiError::from(Error::WorkspacePermissionDenied( + "Workdir removal requires authenticated source authority".to_string(), + ))); + }; + execute_workdir_removal( + &api, + &path.working_directory_id, + &source_actor, + &request.reason, + ) + .map(Json) + .map_err(ApiError::from) } async fn scoped_list_working_directories( @@ -9017,10 +9050,28 @@ async fn scoped_working_directory_detail( async fn scoped_cleanup_working_directory( State(api): State, AxumPath(path): AxumPath, -) -> ApiResult> { + worker_source: Option>, + request_actor: Option>, + Json(request): Json, +) -> ApiResult> { validate_workspace_scope(&api, &path.workspace_id)?; - let runtime_id = registered_workdir_runtime_id(&api, &path.working_directory_id)?; - cleanup_working_directory_for_runtime(api, &runtime_id, &path.working_directory_id) + let source_actor = if let Some(Extension(source)) = worker_source { + format!("worker:{}:{}", source.runtime_id, source.worker_id) + } else if let Some(Extension(actor)) = request_actor { + format!("account:{}", actor.account_id) + } else { + return Err(ApiError::from(Error::WorkspacePermissionDenied( + "Workdir removal requires authenticated source authority".to_string(), + ))); + }; + execute_workdir_removal( + &api, + &path.working_directory_id, + &source_actor, + &request.reason, + ) + .map(Json) + .map_err(ApiError::from) } fn registered_workdir_runtime_id( @@ -9497,54 +9548,218 @@ fn working_directory_detail_for_runtime( )) } -fn cleanup_working_directory_for_runtime( - api: WorkspaceApi, - runtime_id: &str, - working_directory_id: &str, -) -> ApiResult> { - if let Some(candidate) = build_runtime_cleanup_plan(&api, runtime_id)? - .workdirs - .into_iter() - .find(|candidate| candidate.workdir_id == working_directory_id) - { - if let Some(reason) = candidate.blocking_reason { - return Err(cleanup_api_error( - runtime_id, - "workspace_cleanup_workdir_blocked", - &reason, - )); +fn workdir_removal_response( + operation: &WorkdirRemovalOperation, +) -> WorkingDirectoryRemovalResponse { + let disposition = match operation.disposition { + Some(WorkdirRemovalDisposition::Removed) => WorkingDirectoryRemovalDisposition::Removed, + Some(WorkdirRemovalDisposition::Retained) => WorkingDirectoryRemovalDisposition::Retained, + Some(WorkdirRemovalDisposition::AttentionRequired) | None => { + WorkingDirectoryRemovalDisposition::AttentionRequired } - if candidate.action == CleanupTargetKind::WorkdirDirtyDiscard { - return Err(cleanup_api_error( - runtime_id, - "workspace_cleanup_dirty_confirmation_required", - "dirty Workdir discard requires the cleanup execution API with explicit confirmation", - )); + }; + WorkingDirectoryRemovalResponse { + working_directory_id: operation.working_directory_id.clone(), + disposition, + retryable: operation.retryable, + failure_category: operation.failure_category.clone(), + } +} + +fn runtime_reports_workdir_not_found(result: &crate::hosts::RuntimeWorkingDirectoryResult) -> bool { + result.working_directory.is_none() + && result + .diagnostics + .iter() + .any(|diagnostic| diagnostic.code == "working_directory_not_found") +} + +fn classify_workdir_provider_error(error: &RuntimeRegistryError) -> (&'static str, bool) { + match error { + RuntimeRegistryError::UnknownRuntime(_) => ("runtime_unavailable", true), + RuntimeRegistryError::RuntimeOperationFailed { code, .. } + if code == "working_directory_unsupported" => + { + ("unsupported_target", false) + } + RuntimeRegistryError::RuntimeOperationFailed { .. } => ("provider_unavailable", true), + RuntimeRegistryError::InvalidIdentifier { .. } + | RuntimeRegistryError::UnknownHost(_) + | RuntimeRegistryError::UnknownWorker { .. } => ("authority_invalid", false), + } +} + +fn execute_reserved_workdir_removal( + api: &WorkspaceApi, + operation: WorkdirRemovalOperation, +) -> Result { + if operation.state == WorkdirRemovalOperationState::Completed { + return Ok(operation); + } + let operation = api.config_store.begin_workdir_removal_attempt( + &operation.workspace_id, + &operation.operation_id, + &operation.request_fingerprint, + )?; + let guards = match api.config_store.workdir_removal_guards(&operation) { + Ok(guards) => guards, + Err( + Error::InvalidInput(_) + | Error::WorkdirAttachmentConflict(_) + | Error::RegistryInconsistency(_), + ) => { + return api.config_store.fail_workdir_removal_operation( + &operation, + "authority_changed", + false, + ); + } + Err(error) => return Err(error), + }; + if !guards.is_empty() { + return api.config_store.complete_workdir_removal_retained( + &operation, + WorkdirRemovalDisposition::Retained, + "blocked_by_live_authority", + ); + } + + // Runtime observation is the provider's publication/removal authority. A + // missing summary without the exact not-found diagnostic is unknown, not a + // successful delete. + let observed = match api + .runtime + .working_directory(&operation.runtime_id, &operation.working_directory_id) + { + Ok(observed) => observed, + Err(error) => { + let (category, retryable) = classify_workdir_provider_error(&error); + return api + .config_store + .fail_workdir_removal_operation(&operation, category, retryable); + } + }; + if runtime_reports_workdir_not_found(&observed) { + return api.config_store.commit_workdir_removal_removed(&operation); + } + let Some(status) = observed.working_directory.as_ref() else { + return api.config_store.fail_workdir_removal_operation( + &operation, + "provider_observation_unknown", + true, + ); + }; + if status.summary.cleanliness.as_deref() != Some("clean") + || !matches!( + status.summary.status, + WorkingDirectoryStatusKind::Active | WorkingDirectoryStatusKind::CleanupPending + ) + { + return api.config_store.complete_workdir_removal_retained( + &operation, + WorkdirRemovalDisposition::Retained, + "dirty_or_unknown", + ); + } + + let deleted = match api + .runtime + .cleanup_working_directory(&operation.runtime_id, &operation.working_directory_id) + { + Ok(deleted) => deleted, + Err(error) => { + let (category, retryable) = classify_workdir_provider_error(&error); + return api + .config_store + .fail_workdir_removal_operation(&operation, category, retryable); + } + }; + if deleted.state != WorkerOperationState::Accepted + && !runtime_reports_workdir_not_found(&deleted) + { + let category = if deleted + .diagnostics + .iter() + .any(|diagnostic| diagnostic.code == "working_directory_unsupported") + { + "unsupported_target" + } else { + "provider_cleanup_failed" + }; + return api.config_store.fail_workdir_removal_operation( + &operation, + category, + category != "unsupported_target", + ); + } + api.config_store.commit_workdir_removal_removed(&operation) +} + +fn execute_workdir_removal( + api: &WorkspaceApi, + working_directory_id: &str, + source_actor: &str, + reason: &str, +) -> Result { + let reason = reason.trim(); + if reason.is_empty() || reason.len() > 500 { + return Err(Error::InvalidInput( + "Workdir removal reason must be between 1 and 500 bytes".to_string(), + )); + } + let lock = { + let mut locks = api + .workdir_remove_locks + .lock() + .map_err(|_| Error::Store("Workdir removal lock registry was poisoned".to_string()))?; + locks + .entry(working_directory_id.to_string()) + .or_insert_with(|| Arc::new(std::sync::Mutex::new(()))) + .clone() + }; + let _guard = lock + .lock() + .map_err(|_| Error::Store("Workdir removal lock was poisoned".to_string()))?; + + let operation = if let Some(existing) = + api.config_store.find_workdir_removal_operation_by_intent( + api.workspace_id(), + working_directory_id, + source_actor, + reason, + )? { + existing + } else { + let workdir = api + .config_store + .get_workdir_registry(api.workspace_id(), working_directory_id)? + .ok_or_else(|| { + Error::InvalidInput(format!("Unknown Workdir `{working_directory_id}`")) + })?; + let intent = workdir_removal_intent(&workdir, source_actor, reason)?; + api.config_store + .reserve_workdir_removal_operation(&intent)? + }; + execute_reserved_workdir_removal(api, operation) + .map(|operation| workdir_removal_response(&operation)) +} + +fn recover_workdir_removals(api: &WorkspaceApi) -> Result<()> { + for operation in api + .config_store + .recoverable_workdir_removal_operations(api.workspace_id(), 100)? + { + if let Err(error) = execute_reserved_workdir_removal(api, operation.clone()) { + tracing::warn!( + workspace_id = %api.workspace_id(), + workdir_id = %operation.working_directory_id, + operation_id = %operation.operation_id, + category = "workdir_removal_recovery_failed", + "durable Workdir removal recovery failed: {error}" + ); } } - let result = api - .runtime - .cleanup_working_directory(runtime_id, working_directory_id) - .map_err(|err| err.into_error())?; - let Some(working_directory) = result.working_directory else { - return Err(ApiError::with_diagnostics( - Error::RuntimeOperationFailed { - runtime_id: runtime_id.to_string(), - code: "workspace_working_directory_cleanup_failed".to_string(), - message: "Runtime did not cleanup working directory".to_string(), - }, - result.diagnostics, - )); - }; - let mut summary = working_directory.summary; - persist_workdir_cleanup_observation(&api, runtime_id, &summary)?; - apply_workdir_occupancy_projection(&api, &mut summary)?; - Ok(Json(BrowserWorkingDirectoryDetailResponse { - workspace_id: api.config.workspace_id.clone(), - runtime_id: runtime_id.to_string(), - item: summary, - diagnostics: working_directory_diagnostics(result.diagnostics), - })) + Ok(()) } async fn set_worker_retention( @@ -9813,11 +10028,6 @@ async fn execute_runtime_cleanup( } let worker_targets: HashSet<_> = request.worker_target_ids.iter().cloned().collect(); let workdir_targets: HashSet<_> = request.workdir_target_ids.iter().cloned().collect(); - let dirty_confirmations: HashSet<_> = request - .confirm_dirty_discard_target_ids - .iter() - .cloned() - .collect(); let mut results = Vec::new(); for candidate in plan @@ -9891,72 +10101,41 @@ async fn execute_runtime_cleanup( } match candidate.action { CleanupTargetKind::WorkdirDirtyDiscard => { - if !dirty_confirmations.contains(candidate.target_id.as_str()) { - return Err(cleanup_api_error( - runtime_id, - "workspace_cleanup_dirty_confirmation_required", - "dirty Workdir discard requires explicit confirmation", - )); - } - cleanup_runtime_workdir_for_execution(api, runtime_id, candidate)?; - let deleted = api.store.delete_workdir_registry( - &api.config.workspace_id, - candidate.workdir_id.as_str(), - )?; - if !deleted { - return Err(cleanup_api_error( - runtime_id, - "workspace_cleanup_workdir_registry_not_found", - "Backend Workdir registry row was not found after Runtime cleanup", - )); - } results.push(RuntimeCleanupExecutionResult { target_id: candidate.target_id.clone(), action: candidate.action.clone(), - status: "deleted".to_string(), + status: "retained".to_string(), message: - "Dirty/unknown Workdir was deleted from Runtime storage and Backend registry after explicit confirmation" + "Dirty or unknown Workdir was retained; forced deletion is not supported" .to_string(), }); } - CleanupTargetKind::WorkdirCleanCleanup => { - cleanup_runtime_workdir_for_execution(api, runtime_id, candidate)?; - let deleted = api.store.delete_workdir_registry( - &api.config.workspace_id, + CleanupTargetKind::WorkdirCleanCleanup | CleanupTargetKind::WorkdirRecordDelete => { + let removal = execute_workdir_removal( + api, candidate.workdir_id.as_str(), + &format!("runtime-cleanup:{runtime_id}"), + &format!("cleanup target {}", candidate.target_id), )?; - if !deleted { - return Err(cleanup_api_error( - runtime_id, - "workspace_cleanup_workdir_registry_not_found", - "Backend Workdir registry row was not found after Runtime cleanup", - )); - } + let (status, message) = match removal.disposition { + WorkingDirectoryRemovalDisposition::Removed => ( + "deleted", + "Workdir removed through the durable Backend operation", + ), + WorkingDirectoryRemovalDisposition::Retained => ( + "retained", + "Workdir retained after live authority revalidation", + ), + WorkingDirectoryRemovalDisposition::AttentionRequired => ( + "attention_required", + "Workdir removal requires attention and may be retried", + ), + }; results.push(RuntimeCleanupExecutionResult { target_id: candidate.target_id.clone(), action: candidate.action.clone(), - status: "deleted".to_string(), - message: "Workdir deleted from Runtime storage and Backend registry" - .to_string(), - }); - } - CleanupTargetKind::WorkdirRecordDelete => { - let deleted = api.store.delete_workdir_registry( - &api.config.workspace_id, - candidate.workdir_id.as_str(), - )?; - if !deleted { - return Err(cleanup_api_error( - runtime_id, - "workspace_cleanup_workdir_registry_not_found", - "Backend Workdir registry row was not found", - )); - } - results.push(RuntimeCleanupExecutionResult { - target_id: candidate.target_id.clone(), - action: candidate.action.clone(), - status: "deleted".to_string(), - message: "Not-found Workdir registry row deleted".to_string(), + status: status.to_string(), + message: message.to_string(), }); } CleanupTargetKind::WorkerDelete => { @@ -10033,28 +10212,6 @@ fn cleanup_runtime_worker_for_execution( } } -fn cleanup_runtime_workdir_for_execution( - api: &WorkspaceApi, - runtime_id: &str, - candidate: &CleanupWorkdirCandidate, -) -> ApiResult<()> { - let result = api - .runtime - .cleanup_working_directory(runtime_id, candidate.workdir_id.as_str()) - .map_err(|err| err.into_error())?; - if result.working_directory.is_none() { - return Err(ApiError::with_diagnostics( - Error::RuntimeOperationFailed { - runtime_id: runtime_id.to_string(), - code: "workspace_cleanup_workdir_runtime_failed".to_string(), - message: "Runtime did not cleanup selected Workdir".to_string(), - }, - result.diagnostics, - )); - }; - Ok(()) -} - fn cleanup_api_error(runtime_id: &str, code: &str, message: &str) -> ApiError { Error::RuntimeOperationFailed { runtime_id: runtime_id.to_string(), @@ -14342,10 +14499,12 @@ fn sync_runtime_workdir_observations( for status in &response.items { observed.insert(status.summary.working_directory_id.clone()); if status.summary.status == WorkingDirectoryStatusKind::NotFound { - api.store.delete_workdir_registry( + if let Some(record) = api.store.get_workdir_registry( &api.config.workspace_id, &status.summary.working_directory_id, - )?; + )? { + persist_workdir_not_found(api, record)?; + } continue; } let existing = api.store.get_workdir_registry( @@ -14369,10 +14528,7 @@ fn sync_runtime_workdir_observations( Ok(result) => { if let Some(status) = result.working_directory { if status.summary.status == WorkingDirectoryStatusKind::NotFound { - api.store.delete_workdir_registry( - &api.config.workspace_id, - record.workdir_id.as_str(), - )?; + persist_workdir_not_found(api, record)?; } else { let mut updated = workdir_record_from_summary(api, runtime_id, &status.summary); @@ -14403,10 +14559,12 @@ fn persist_workdir_cleanup_observation( summary: &WorkingDirectorySummary, ) -> ApiResult<()> { if summary.status == WorkingDirectoryStatusKind::NotFound { - api.store.delete_workdir_registry( + if let Some(record) = api.store.get_workdir_registry( &api.config.workspace_id, summary.working_directory_id.as_str(), - )?; + )? { + persist_workdir_not_found(api, record)?; + } } else { let record = workdir_record_from_summary(api, runtime_id, summary); api.store.upsert_workdir_registry(&record)?; @@ -14428,14 +14586,28 @@ fn workdir_status_from_runtime_miss(diagnostics: &[RuntimeDiagnostic]) -> &'stat } } +fn persist_workdir_not_found( + api: &WorkspaceApi, + mut record: WorkdirRegistryRecord, +) -> ApiResult<()> { + record.materialization_status = "not_found".to_string(); + record.cleanliness = "unknown".to_string(); + record.observed_at_epoch_seconds = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .ok() + .map(|duration| duration.as_secs()); + record.updated_at = now_registry_timestamp(); + api.store.upsert_workdir_registry(&record)?; + Ok(()) +} + fn persist_workdir_runtime_miss( api: &WorkspaceApi, mut record: WorkdirRegistryRecord, diagnostics: &[RuntimeDiagnostic], ) -> ApiResult<()> { if workdir_runtime_miss_is_not_found(diagnostics) { - api.store - .delete_workdir_registry(&api.config.workspace_id, record.workdir_id.as_str())?; + persist_workdir_not_found(api, record)?; } else { record.materialization_status = "unknown".to_string(); record.cleanliness = "unknown".to_string(); @@ -22154,31 +22326,120 @@ mod tests { } #[tokio::test] - async fn simple_workdir_cleanup_rejects_dirty_and_blocked_candidates() { + async fn durable_workdir_removal_replays_completed_provider_not_found_result() { let workspace = tempfile::tempdir().unwrap(); init_clean_git_workspace(workspace.path()); let api = test_api(workspace.path()).await; - seed_cleanup_workdir(&api, "dirty-workdir", "present", "dirty"); - let dirty = - cleanup_working_directory_for_runtime(api.clone(), "runtime-test", "dirty-workdir") - .unwrap_err(); - assert!(matches!( - dirty.error, - Error::RuntimeOperationFailed { ref code, .. } - if code == "workspace_cleanup_dirty_confirmation_required" - )); + seed_cleanup_workdir(&api, "missing-clean-workdir", "present", "clean"); - let pinned = seed_cleanup_worker(&api, 17, "pinned"); - seed_cleanup_workdir(&api, "blocked-workdir", "present", "clean"); - seed_cleanup_link(&api, pinned.as_str(), "blocked-workdir"); - let blocked = - cleanup_working_directory_for_runtime(api.clone(), "runtime-test", "blocked-workdir") - .unwrap_err(); - assert!(matches!( - blocked.error, - Error::RuntimeOperationFailed { ref code, .. } - if code == "workspace_cleanup_workdir_blocked" - )); + let first = execute_workdir_removal( + &api, + "missing-clean-workdir", + "account:owner", + "remove stale clean Workdir", + ) + .unwrap(); + assert_eq!( + first.disposition, + WorkingDirectoryRemovalDisposition::Removed, + "response: {first:?}" + ); + assert!(!first.retryable); + assert!( + api.store + .get_workdir_registry(&api.config.workspace_id, "missing-clean-workdir") + .unwrap() + .is_none() + ); + + let replay = execute_workdir_removal( + &api, + "missing-clean-workdir", + "account:owner", + "remove stale clean Workdir", + ) + .unwrap(); + assert_eq!(replay, first); + let operation = api + .config_store + .find_workdir_removal_operation_by_intent( + &api.config.workspace_id, + "missing-clean-workdir", + "account:owner", + "remove stale clean Workdir", + ) + .unwrap() + .unwrap(); + assert_eq!(operation.attempt_count, 1); + } + + #[tokio::test] + async fn durable_workdir_removal_retains_occupied_and_pinned_workdir() { + let workspace = tempfile::tempdir().unwrap(); + init_clean_git_workspace(workspace.path()); + let api = test_api(workspace.path()).await; + let worker = seed_cleanup_worker(&api, 27, "pinned"); + seed_cleanup_workdir(&api, "occupied-workdir", "present", "clean"); + seed_cleanup_link(&api, worker.as_str(), "occupied-workdir"); + + let result = execute_workdir_removal( + &api, + "occupied-workdir", + "account:owner", + "remove occupied Workdir", + ) + .unwrap(); + assert_eq!( + result.disposition, + WorkingDirectoryRemovalDisposition::Retained + ); + assert_eq!( + result.failure_category.as_deref(), + Some("blocked_by_live_authority") + ); + assert!( + api.store + .get_workdir_registry(&api.config.workspace_id, "occupied-workdir") + .unwrap() + .is_some() + ); + } + + #[tokio::test] + async fn recoverable_workdir_removal_converges_after_provider_side_effect() { + let workspace = tempfile::tempdir().unwrap(); + init_clean_git_workspace(workspace.path()); + let api = test_api(workspace.path()).await; + seed_cleanup_workdir(&api, "recovery-workdir", "present", "clean"); + let record = api + .store + .get_workdir_registry(&api.config.workspace_id, "recovery-workdir") + .unwrap() + .unwrap(); + let intent = + workdir_removal_intent(&record, "account:owner", "recover provider cleanup").unwrap(); + api.config_store + .reserve_workdir_removal_operation(&intent) + .unwrap(); + + recover_workdir_removals(&api).unwrap(); + + let operation = api + .config_store + .get_workdir_removal_operation(&api.config.workspace_id, &intent.operation_id) + .unwrap() + .unwrap(); + assert_eq!(operation.state, WorkdirRemovalOperationState::Completed); + assert_eq!( + operation.disposition, + Some(WorkdirRemovalDisposition::Removed) + ); + assert!( + api.store + .get_workdir_registry(&api.config.workspace_id, "recovery-workdir") + .unwrap() + .is_none() + ); } #[tokio::test] diff --git a/crates/workspace-server/src/store.rs b/crates/workspace-server/src/store.rs index 8114d56c..271bd05c 100644 --- a/crates/workspace-server/src/store.rs +++ b/crates/workspace-server/src/store.rs @@ -267,6 +267,11 @@ const MIGRATIONS: &[Migration] = &[ name: "require one account owner for every Workspace", apply: require_workspace_account_owner, }, + Migration { + version: 49, + name: "create durable Workdir removal operations", + apply: crate::workdir_removal::create_workdir_removal_operations, + }, ]; struct Migration { @@ -10085,6 +10090,51 @@ mod tests { .unwrap(); } + #[test] + fn schema_v49_upgrades_v48_with_durable_workdir_removal_authority() { + let conn = Connection::open_in_memory().unwrap(); + configure_sqlite(&conn).unwrap(); + apply_migrations_through(&conn, 48).unwrap(); + assert_eq!(current_schema_version(&conn).unwrap(), 48); + assert!(!table_exists(&conn, "workdir_removal_operations").unwrap()); + + apply_migrations(&conn).unwrap(); + + assert_eq!(current_schema_version(&conn).unwrap(), 49); + assert!(table_exists(&conn, "workdir_removal_operations").unwrap()); + let columns = table_columns(&conn, "workdir_removal_operations").unwrap(); + for required in [ + "workspace_id", + "operation_id", + "request_fingerprint", + "workdir_id", + "runtime_id", + "repository_id", + "materialization_fingerprint", + "source_actor", + "reason", + "state", + "attempt_count", + "retryable", + "disposition", + "failure_category", + "created_at", + "updated_at", + "completed_at", + ] { + assert!( + columns.iter().any(|column| column == required), + "missing {required}" + ); + } + let foreign_key_failures: i64 = conn + .query_row("SELECT count(*) FROM pragma_foreign_key_check", [], |row| { + row.get(0) + }) + .unwrap(); + assert_eq!(foreign_key_failures, 0); + } + #[test] fn schema_v44_migrates_repository_sources_without_promoting_legacy_auth_refs() { let conn = Connection::open_in_memory().unwrap(); @@ -10118,7 +10168,7 @@ mod tests { assign_explicit_test_workspace_owner(&conn); apply_migrations(&conn).unwrap(); - assert_eq!(current_schema_version(&conn).unwrap(), 48); + assert_eq!(current_schema_version(&conn).unwrap(), 49); let remote = conn .query_row( "SELECT source_kind, source_uri, source_revision, source_fingerprint, observed_status \ @@ -10197,7 +10247,7 @@ mod tests { let before = std::fs::read(&path).unwrap(); let plan = SqliteWorkspaceStore::migration_plan(&path).unwrap(); assert_eq!(plan.current_schema_version, 36); - assert_eq!(plan.target_schema_version, 48); + assert_eq!(plan.target_schema_version, 49); assert!(plan.migration_required); assert_eq!(plan.worker_count, 1); assert_eq!(plan.mappings[0].legacy_worker_id, 7); @@ -10211,7 +10261,7 @@ mod tests { store .with_conn(|conn| { assert!(table_exists(conn, "worker_diagnostics_archives")?); - assert_eq!(current_schema_version(conn)?, 48); + assert_eq!(current_schema_version(conn)?, 49); Ok(()) }) .unwrap(); @@ -10351,7 +10401,7 @@ mod tests { ), ] ); - assert_eq!(current_schema_version(&conn).unwrap(), 48); + assert_eq!(current_schema_version(&conn).unwrap(), 49); let foreign_key_error: Option = conn .query_row("PRAGMA foreign_key_check", [], |row| row.get(0)) .optional() @@ -10481,7 +10531,7 @@ INSERT INTO worker_orphan_diagnostics ( assign_explicit_test_workspace_owner(&conn); apply_migrations(&conn).unwrap(); - assert_eq!(current_schema_version(&conn).unwrap(), 48); + assert_eq!(current_schema_version(&conn).unwrap(), 49); assert!(!table_exists(&conn, "worker_control_delegation_operations").unwrap()); let controller_worker_id: String = conn .query_row( @@ -10599,7 +10649,7 @@ INSERT INTO worker_orphan_diagnostics ( apply_migrations(&conn).unwrap(); - assert_eq!(current_schema_version(&conn).unwrap(), 48); + assert_eq!(current_schema_version(&conn).unwrap(), 49); assert!(table_exists(&conn, "worker_workdir_attachment_reservations").unwrap()); } @@ -10618,7 +10668,7 @@ INSERT INTO worker_orphan_diagnostics ( assign_explicit_test_workspace_owner(&conn); apply_migrations(&conn).unwrap(); - assert_eq!(current_schema_version(&conn).unwrap(), 48); + assert_eq!(current_schema_version(&conn).unwrap(), 49); let settings = conn .query_row( "SELECT settings_revision, language FROM workspace_memory_settings \ @@ -10659,7 +10709,7 @@ CREATE TABLE flow_events (event_id TEXT PRIMARY KEY); apply_migrations(&conn).unwrap(); - assert_eq!(current_schema_version(&conn).unwrap(), 48); + assert_eq!(current_schema_version(&conn).unwrap(), 49); assert!(table_exists(&conn, "flow_sources").unwrap()); assert!(table_exists(&conn, "flow_source_revisions").unwrap()); assert!(!table_exists(&conn, "flow_instances").unwrap()); @@ -10727,7 +10777,7 @@ INSERT INTO worker_workdir_attachment_reservations ( assign_explicit_test_workspace_owner(&conn); apply_migrations(&conn).unwrap(); - assert_eq!(current_schema_version(&conn).unwrap(), 48); + assert_eq!(current_schema_version(&conn).unwrap(), 49); let repositories_sql: String = conn .query_row( "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'repositories'", @@ -10910,7 +10960,7 @@ INSERT INTO workdir_registry ( let db = dir.path().join("control-plane.sqlite"); let store = SqliteWorkspaceStore::open(&db).unwrap(); - assert_eq!(store.schema_version().await.unwrap(), 48); + assert_eq!(store.schema_version().await.unwrap(), 49); assert!( !store .with_conn(|conn| table_exists(conn, "worker_workspace_credentials")) @@ -10927,7 +10977,7 @@ INSERT INTO workdir_registry ( store.upsert_workspace(&record).await.unwrap(); let reopened = SqliteWorkspaceStore::open(&db).unwrap(); - assert_eq!(reopened.schema_version().await.unwrap(), 48); + assert_eq!(reopened.schema_version().await.unwrap(), 49); assert_eq!( reopened.get_workspace("local-dev").await.unwrap(), Some(record) @@ -11693,7 +11743,7 @@ INSERT INTO worker_registry ( let migrated = SqliteWorkspaceStore::open(&db_path).unwrap(); migrated .with_conn(|conn| { - assert_eq!(current_schema_version(conn)?, 48); + assert_eq!(current_schema_version(conn)?, 49); assert_eq!( conn.query_row("PRAGMA foreign_keys", [], |row| row.get::<_, i64>(0))?, 1, @@ -12044,7 +12094,7 @@ INSERT INTO worker_registry ( assert_eq!(current_schema_version(&conn).unwrap(), 44); apply_migrations(&conn).unwrap(); - assert_eq!(current_schema_version(&conn).unwrap(), 48); + assert_eq!(current_schema_version(&conn).unwrap(), 49); assert!(table_exists(&conn, "workdir_create_operations").unwrap()); let columns = table_columns(&conn, "workdir_create_operations").unwrap(); for required in [ @@ -12071,7 +12121,7 @@ INSERT INTO worker_registry ( assert_eq!(current_schema_version(&conn).unwrap(), 45); apply_migrations(&conn).unwrap(); - assert_eq!(current_schema_version(&conn).unwrap(), 48); + assert_eq!(current_schema_version(&conn).unwrap(), 49); for table in [ "repository_ssh_credentials", "repository_ssh_credential_revisions", @@ -12098,7 +12148,7 @@ INSERT INTO worker_registry ( assert_eq!(current_schema_version(&conn).unwrap(), 46); apply_migrations(&conn).unwrap(); - assert_eq!(current_schema_version(&conn).unwrap(), 48); + assert_eq!(current_schema_version(&conn).unwrap(), 49); let columns = table_columns(&conn, "workdir_create_operations").unwrap(); for required in [ "source_kind", @@ -12362,7 +12412,7 @@ VALUES ('workspace-b', 'ticket-b', 'related', 'ticket-a', NULL, 'tester', '2026- assign_explicit_test_workspace_owner(&conn); apply_migrations(&mut conn).unwrap(); - assert_eq!(current_schema_version(&conn).unwrap(), 48); + assert_eq!(current_schema_version(&conn).unwrap(), 49); let workspace_id: Option = conn .query_row( "SELECT workspace_id FROM trusted_runtime_records WHERE runtime_id = 'runtime-a'", @@ -12991,7 +13041,7 @@ WHERE workspace_id = 'workspace-a' assign_explicit_test_workspace_owner(&conn); let store = SqliteWorkspaceStore::from_connection(conn).unwrap(); - assert_eq!(store.schema_version().await.unwrap(), 48); + assert_eq!(store.schema_version().await.unwrap(), 49); store .with_conn(|conn| { @@ -13180,7 +13230,7 @@ CREATE TABLE ticket_assignment_operations ( #[tokio::test] async fn repository_records_round_trip() { let store = SqliteWorkspaceStore::in_memory().unwrap(); - assert_eq!(store.schema_version().await.unwrap(), 48); + assert_eq!(store.schema_version().await.unwrap(), 49); let workspace = WorkspaceRecord { workspace_id: "local-dev".to_string(), owner_account_id: "owner-account".to_string(), @@ -13258,7 +13308,7 @@ CREATE TABLE ticket_assignment_operations ( #[tokio::test] async fn memory_authority_records_round_trip_and_close_staging() { let store = SqliteWorkspaceStore::in_memory().unwrap(); - assert_eq!(store.schema_version().await.unwrap(), 48); + assert_eq!(store.schema_version().await.unwrap(), 49); let workspace = WorkspaceRecord { workspace_id: "local-dev".to_string(), owner_account_id: "owner-account".to_string(), @@ -13671,7 +13721,7 @@ CREATE TABLE ticket_assignment_operations ( #[tokio::test] async fn account_and_login_records_round_trip() { let store = SqliteWorkspaceStore::in_memory().unwrap(); - assert_eq!(store.schema_version().await.unwrap(), 48); + assert_eq!(store.schema_version().await.unwrap(), 49); let now = "2026-07-22T00:00:00Z".to_string(); let account = AccountRecord { account_id: "acct-user-alice".to_string(), diff --git a/crates/workspace-server/src/workdir_removal.rs b/crates/workspace-server/src/workdir_removal.rs new file mode 100644 index 00000000..28abc84c --- /dev/null +++ b/crates/workspace-server/src/workdir_removal.rs @@ -0,0 +1,952 @@ +//! Durable, retryable Backend authority for persistent Workdir removal. +//! +//! Runtime cleanup is an external side effect, so callers reserve one immutable +//! operation before invoking the provider. The final registry deletion and +//! operation completion are committed atomically. If a process stops after the +//! provider side effect but before that transaction, recovery re-observes the +//! provider and converges through the same operation. + +use chrono::Utc; +use rusqlite::{Connection, OptionalExtension, TransactionBehavior, params}; +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; + +use crate::store::WorkdirRegistryRecord; +use crate::{Error, Result, SqliteWorkspaceStore}; + +const MAX_REASON_BYTES: usize = 500; +const MAX_ACTOR_BYTES: usize = 200; +const MAX_FAILURE_CATEGORY_BYTES: usize = 128; + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum WorkdirRemovalOperationState { + Pending, + Failed, + Completed, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum WorkdirRemovalDisposition { + Removed, + Retained, + AttentionRequired, +} + +impl WorkdirRemovalDisposition { + fn as_str(self) -> &'static str { + match self { + Self::Removed => "removed", + Self::Retained => "retained", + Self::AttentionRequired => "attention_required", + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct WorkdirRemovalIntent { + pub operation_id: String, + pub request_fingerprint: String, + pub workspace_id: String, + pub working_directory_id: String, + pub runtime_id: String, + pub repository_id: String, + pub materialization_fingerprint: String, + pub source_actor: String, + pub reason: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct WorkdirRemovalOperation { + pub operation_id: String, + pub request_fingerprint: String, + pub workspace_id: String, + pub working_directory_id: String, + pub runtime_id: String, + pub repository_id: String, + pub materialization_fingerprint: String, + pub source_actor: String, + pub reason: String, + pub state: WorkdirRemovalOperationState, + pub attempt_count: u64, + pub retryable: bool, + pub disposition: Option, + pub failure_category: Option, + pub created_at: String, + pub updated_at: String, + pub completed_at: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct WorkdirRemovalGuard { + pub category: &'static str, + pub detail: &'static str, +} + +pub(crate) fn create_workdir_removal_operations(conn: &Connection) -> Result<()> { + conn.execute_batch( + r#" +CREATE TABLE workdir_removal_operations ( + workspace_id TEXT NOT NULL, + operation_id TEXT NOT NULL, + request_fingerprint TEXT NOT NULL, + workdir_id TEXT NOT NULL, + runtime_id TEXT NOT NULL, + repository_id TEXT NOT NULL, + materialization_fingerprint TEXT NOT NULL, + source_actor TEXT NOT NULL, + reason TEXT NOT NULL, + state TEXT NOT NULL CHECK (state IN ('pending', 'failed', 'completed')), + attempt_count INTEGER NOT NULL DEFAULT 0 CHECK (attempt_count >= 0), + retryable INTEGER NOT NULL CHECK (retryable IN (0, 1)), + disposition TEXT CHECK (disposition IN ('removed', 'retained', 'attention_required')), + failure_category TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + completed_at TEXT, + PRIMARY KEY (workspace_id, operation_id), + FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE, + FOREIGN KEY (workspace_id, repository_id) + REFERENCES repositories(workspace_id, repository_id) ON DELETE RESTRICT +); +CREATE INDEX idx_workdir_removal_operations_recovery + ON workdir_removal_operations(workspace_id, state, retryable, updated_at); +CREATE INDEX idx_workdir_removal_operations_workdir + ON workdir_removal_operations(workspace_id, workdir_id, created_at DESC); +"#, + )?; + Ok(()) +} + +pub fn workdir_materialization_fingerprint(record: &WorkdirRegistryRecord) -> String { + let bytes = serde_json::to_vec(&serde_json::json!([ + record.workspace_id, + record.workdir_id, + record.runtime_id, + record.repository_id, + record.creation_selector, + record.creation_ref, + record.creation_tree, + ])) + .expect("Workdir materialization identity is serializable"); + hex_sha256(&bytes) +} + +pub fn workdir_removal_intent( + record: &WorkdirRegistryRecord, + source_actor: &str, + reason: &str, +) -> Result { + validate_bounded("source actor", source_actor, MAX_ACTOR_BYTES)?; + validate_bounded("reason", reason, MAX_REASON_BYTES)?; + let materialization_fingerprint = workdir_materialization_fingerprint(record); + let fingerprint_bytes = serde_json::to_vec(&serde_json::json!([ + record.workspace_id, + record.workdir_id, + record.runtime_id, + record.repository_id, + materialization_fingerprint, + source_actor, + reason, + ])) + .map_err(|error| Error::Store(format!("Workdir removal fingerprint failed: {error}")))?; + let request_fingerprint = hex_sha256(&fingerprint_bytes); + Ok(WorkdirRemovalIntent { + operation_id: format!("wdr_{}", &request_fingerprint[..32]), + request_fingerprint, + workspace_id: record.workspace_id.clone(), + working_directory_id: record.workdir_id.clone(), + runtime_id: record.runtime_id.clone(), + repository_id: record.repository_id.clone(), + materialization_fingerprint, + source_actor: source_actor.to_string(), + reason: reason.to_string(), + }) +} + +impl SqliteWorkspaceStore { + pub fn reserve_workdir_removal_operation( + &self, + intent: &WorkdirRemovalIntent, + ) -> Result { + validate_intent(intent)?; + self.with_conn_mut(|conn| { + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + if let Some(existing) = load_operation(&tx, &intent.workspace_id, &intent.operation_id)? { + if existing.request_fingerprint != intent.request_fingerprint + || existing.working_directory_id != intent.working_directory_id + || existing.runtime_id != intent.runtime_id + || existing.repository_id != intent.repository_id + || existing.materialization_fingerprint != intent.materialization_fingerprint + || existing.source_actor != intent.source_actor + || existing.reason != intent.reason + { + return Err(Error::WorkdirAttachmentConflict(format!( + "Workdir removal operation `{}` was reused with different intent", + intent.operation_id + ))); + } + tx.commit()?; + return Ok(existing); + } + let current = load_workdir_record(&tx, &intent.workspace_id, &intent.working_directory_id)? + .ok_or_else(|| Error::InvalidInput(format!( + "Unknown Workdir `{}`", + intent.working_directory_id + )))?; + require_matching_materialization(intent, ¤t)?; + let repository_exists: bool = tx.query_row( + "SELECT EXISTS(SELECT 1 FROM repositories WHERE workspace_id=?1 AND repository_id=?2)", + params![intent.workspace_id, intent.repository_id], + |row| row.get(0), + )?; + if !repository_exists { + return Err(Error::RegistryInconsistency( + "Workdir Repository authority is missing".to_string(), + )); + } + let now = Utc::now().to_rfc3339(); + tx.execute( + r#"INSERT INTO workdir_removal_operations ( + workspace_id, operation_id, request_fingerprint, workdir_id, runtime_id, + repository_id, materialization_fingerprint, source_actor, reason, state, + attempt_count, retryable, disposition, failure_category, + created_at, updated_at, completed_at + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, 'pending', 0, 1, NULL, NULL, ?10, ?10, NULL)"#, + params![ + intent.workspace_id, + intent.operation_id, + intent.request_fingerprint, + intent.working_directory_id, + intent.runtime_id, + intent.repository_id, + intent.materialization_fingerprint, + intent.source_actor, + intent.reason, + now, + ], + )?; + let operation = load_operation(&tx, &intent.workspace_id, &intent.operation_id)? + .ok_or_else(|| Error::Store("reserved Workdir removal operation is missing".to_string()))?; + tx.commit()?; + Ok(operation) + }) + } + + pub fn begin_workdir_removal_attempt( + &self, + workspace_id: &str, + operation_id: &str, + request_fingerprint: &str, + ) -> Result { + self.with_conn_mut(|conn| { + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + let operation = require_operation(&tx, workspace_id, operation_id, request_fingerprint)?; + if operation.state == WorkdirRemovalOperationState::Completed { + tx.commit()?; + return Ok(operation); + } + if operation.state == WorkdirRemovalOperationState::Failed && !operation.retryable { + return Err(Error::InvalidInput(format!( + "Workdir removal operation `{operation_id}` is not retryable" + ))); + } + let now = Utc::now().to_rfc3339(); + tx.execute( + "UPDATE workdir_removal_operations SET state='pending', attempt_count=attempt_count+1, retryable=1, failure_category=NULL, disposition=NULL, updated_at=?1, completed_at=NULL WHERE workspace_id=?2 AND operation_id=?3 AND request_fingerprint=?4", + params![now, workspace_id, operation_id, request_fingerprint], + )?; + let operation = require_operation(&tx, workspace_id, operation_id, request_fingerprint)?; + tx.commit()?; + Ok(operation) + }) + } + + pub fn workdir_removal_guards( + &self, + operation: &WorkdirRemovalOperation, + ) -> Result> { + self.with_conn(|conn| { + let current = load_workdir_record( + conn, + &operation.workspace_id, + &operation.working_directory_id, + )? + .ok_or_else(|| Error::InvalidInput(format!( + "Unknown Workdir `{}`", + operation.working_directory_id + )))?; + require_operation_materialization(operation, ¤t)?; + let active_attachment: bool = conn.query_row( + "SELECT EXISTS(SELECT 1 FROM worker_workdir_links WHERE workspace_id=?1 AND workdir_id=?2 AND unlinked_at IS NULL)", + params![operation.workspace_id, operation.working_directory_id], + |row| row.get(0), + )?; + let pending_attachment: bool = conn.query_row( + "SELECT EXISTS(SELECT 1 FROM worker_workdir_attachment_reservations WHERE workspace_id=?1 AND workdir_id=?2)", + params![operation.workspace_id, operation.working_directory_id], + |row| row.get(0), + )?; + let current_assignment: bool = conn.query_row( + r#"SELECT EXISTS( + SELECT 1 FROM worker_workdir_links AS link + JOIN ticket_current_worker_assignments AS current + ON current.workspace_id=link.workspace_id + JOIN ticket_worker_assignments AS assignment + ON assignment.workspace_id=current.workspace_id + AND assignment.ticket_id=current.ticket_id + AND assignment.role=current.role + AND assignment.assignment_id=current.assignment_id + AND assignment.runtime_id=link.runtime_id + AND assignment.worker_id=link.worker_id + WHERE link.workspace_id=?1 AND link.workdir_id=?2 AND link.unlinked_at IS NULL + )"#, + params![operation.workspace_id, operation.working_directory_id], + |row| row.get(0), + )?; + let retention_hold: bool = conn.query_row( + r#"SELECT EXISTS( + SELECT 1 FROM worker_workdir_links AS link + JOIN worker_registry AS worker + ON worker.workspace_id=link.workspace_id + AND worker.runtime_id=link.runtime_id + AND worker.worker_id=link.worker_id + WHERE link.workspace_id=?1 AND link.workdir_id=?2 + AND link.unlinked_at IS NULL AND worker.retention_state='pinned' + )"#, + params![operation.workspace_id, operation.working_directory_id], + |row| row.get(0), + )?; + let pending_materialization: bool = conn.query_row( + "SELECT EXISTS(SELECT 1 FROM workdir_create_operations WHERE workspace_id=?1 AND working_directory_id=?2 AND state='pending')", + params![operation.workspace_id, operation.working_directory_id], + |row| row.get(0), + )?; + let mut guards = Vec::new(); + if active_attachment { + guards.push(WorkdirRemovalGuard { + category: "active_attachment", + detail: "Workdir has an active Worker attachment", + }); + } + if pending_attachment { + guards.push(WorkdirRemovalGuard { + category: "pending_attachment", + detail: "Workdir has a pending Worker attachment reservation", + }); + } + if current_assignment { + guards.push(WorkdirRemovalGuard { + category: "current_assignment", + detail: "Workdir is bound to a Worker with a current Ticket assignment", + }); + } + if retention_hold { + guards.push(WorkdirRemovalGuard { + category: "retention_hold", + detail: "Workdir is bound to a retained Worker", + }); + } + if pending_materialization { + guards.push(WorkdirRemovalGuard { + category: "materialization_pending", + detail: "Workdir materialization is still pending", + }); + } + Ok(guards) + }) + } + + pub fn complete_workdir_removal_retained( + &self, + operation: &WorkdirRemovalOperation, + disposition: WorkdirRemovalDisposition, + category: &str, + ) -> Result { + validate_failure_category(category)?; + if disposition == WorkdirRemovalDisposition::Removed { + return Err(Error::InvalidInput( + "retained completion cannot use removed disposition".to_string(), + )); + } + self.finish_workdir_removal_operation(operation, disposition, false, Some(category), false) + } + + pub fn fail_workdir_removal_operation( + &self, + operation: &WorkdirRemovalOperation, + category: &str, + retryable: bool, + ) -> Result { + validate_failure_category(category)?; + self.with_conn_mut(|conn| { + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + let current = require_operation( + &tx, + &operation.workspace_id, + &operation.operation_id, + &operation.request_fingerprint, + )?; + if current.state == WorkdirRemovalOperationState::Completed { + tx.commit()?; + return Ok(current); + } + let now = Utc::now().to_rfc3339(); + tx.execute( + "UPDATE workdir_removal_operations SET state='failed', retryable=?1, disposition=?2, failure_category=?3, updated_at=?4 WHERE workspace_id=?5 AND operation_id=?6 AND request_fingerprint=?7", + params![ + retryable, + WorkdirRemovalDisposition::AttentionRequired.as_str(), + category, + now, + operation.workspace_id, + operation.operation_id, + operation.request_fingerprint, + ], + )?; + let updated = require_operation( + &tx, + &operation.workspace_id, + &operation.operation_id, + &operation.request_fingerprint, + )?; + tx.commit()?; + Ok(updated) + }) + } + + pub fn commit_workdir_removal_removed( + &self, + operation: &WorkdirRemovalOperation, + ) -> Result { + self.finish_workdir_removal_operation( + operation, + WorkdirRemovalDisposition::Removed, + false, + None, + true, + ) + } + + fn finish_workdir_removal_operation( + &self, + operation: &WorkdirRemovalOperation, + disposition: WorkdirRemovalDisposition, + retryable: bool, + category: Option<&str>, + delete_registry: bool, + ) -> Result { + self.with_conn_mut(|conn| { + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + let current = require_operation( + &tx, + &operation.workspace_id, + &operation.operation_id, + &operation.request_fingerprint, + )?; + if current.state == WorkdirRemovalOperationState::Completed { + tx.commit()?; + return Ok(current); + } + if delete_registry { + let registry = load_workdir_record( + &tx, + &operation.workspace_id, + &operation.working_directory_id, + )? + .ok_or_else(|| Error::RegistryInconsistency( + "Workdir registry row disappeared before durable removal commit".to_string(), + ))?; + require_operation_materialization(operation, ®istry)?; + require_no_removal_blockers( + &tx, + &operation.workspace_id, + &operation.working_directory_id, + )?; + let deleted = tx.execute( + "DELETE FROM workdir_registry WHERE workspace_id=?1 AND workdir_id=?2", + params![operation.workspace_id, operation.working_directory_id], + )?; + if deleted != 1 { + return Err(Error::RegistryInconsistency( + "Workdir registry deletion did not remove exactly one row".to_string(), + )); + } + } + let now = Utc::now().to_rfc3339(); + tx.execute( + "UPDATE workdir_removal_operations SET state='completed', retryable=?1, disposition=?2, failure_category=?3, updated_at=?4, completed_at=?4 WHERE workspace_id=?5 AND operation_id=?6 AND request_fingerprint=?7", + params![ + retryable, + disposition.as_str(), + category, + now, + operation.workspace_id, + operation.operation_id, + operation.request_fingerprint, + ], + )?; + let updated = require_operation( + &tx, + &operation.workspace_id, + &operation.operation_id, + &operation.request_fingerprint, + )?; + tx.commit()?; + Ok(updated) + }) + } + + pub fn find_workdir_removal_operation_by_intent( + &self, + workspace_id: &str, + working_directory_id: &str, + source_actor: &str, + reason: &str, + ) -> Result> { + self.with_conn(|conn| { + conn.query_row( + &format!( + "{} WHERE workspace_id=?1 AND workdir_id=?2 AND source_actor=?3 AND reason=?4 ORDER BY created_at DESC, operation_id DESC LIMIT 1", + operation_select_sql() + ), + params![workspace_id, working_directory_id, source_actor, reason], + read_operation, + ) + .optional() + .map_err(Error::from) + }) + } + + pub fn recoverable_workdir_removal_operations( + &self, + workspace_id: &str, + limit: usize, + ) -> Result> { + self.with_conn(|conn| { + let mut statement = conn.prepare( + &format!( + "{} WHERE workspace_id=?1 AND (state='pending' OR (state='failed' AND retryable=1)) ORDER BY updated_at ASC, operation_id ASC LIMIT ?2", + operation_select_sql() + ), + )?; + let rows = statement.query_map(params![workspace_id, limit as i64], read_operation)?; + rows.collect::, _>>() + .map_err(Error::from) + }) + } + + pub fn get_workdir_removal_operation( + &self, + workspace_id: &str, + operation_id: &str, + ) -> Result> { + self.with_conn(|conn| load_operation(conn, workspace_id, operation_id)) + } +} + +fn require_no_removal_blockers( + conn: &Connection, + workspace_id: &str, + workdir_id: &str, +) -> Result<()> { + let blocked: bool = conn.query_row( + r#"SELECT EXISTS( + SELECT 1 FROM worker_workdir_links + WHERE workspace_id=?1 AND workdir_id=?2 AND unlinked_at IS NULL + UNION ALL + SELECT 1 FROM worker_workdir_attachment_reservations + WHERE workspace_id=?1 AND workdir_id=?2 + )"#, + params![workspace_id, workdir_id], + |row| row.get(0), + )?; + if blocked { + return Err(Error::WorkdirAttachmentConflict(format!( + "Workdir {workdir_id} acquired an active or pending attachment during removal" + ))); + } + Ok(()) +} + +fn validate_intent(intent: &WorkdirRemovalIntent) -> Result<()> { + for (label, value) in [ + ("operation id", intent.operation_id.as_str()), + ("request fingerprint", intent.request_fingerprint.as_str()), + ("Workspace id", intent.workspace_id.as_str()), + ("Workdir id", intent.working_directory_id.as_str()), + ("Runtime id", intent.runtime_id.as_str()), + ("Repository id", intent.repository_id.as_str()), + ( + "materialization fingerprint", + intent.materialization_fingerprint.as_str(), + ), + ] { + validate_bounded(label, value, 256)?; + } + validate_bounded("source actor", &intent.source_actor, MAX_ACTOR_BYTES)?; + validate_bounded("reason", &intent.reason, MAX_REASON_BYTES) +} + +fn validate_bounded(label: &str, value: &str, max: usize) -> Result<()> { + if value.trim().is_empty() || value.len() > max { + return Err(Error::InvalidInput(format!( + "{label} must be non-empty and at most {max} bytes" + ))); + } + Ok(()) +} + +fn validate_failure_category(category: &str) -> Result<()> { + if category.is_empty() + || category.len() > MAX_FAILURE_CATEGORY_BYTES + || !category + .bytes() + .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || byte == b'_') + { + return Err(Error::InvalidInput( + "Workdir removal failure category is invalid".to_string(), + )); + } + Ok(()) +} + +fn require_matching_materialization( + intent: &WorkdirRemovalIntent, + record: &WorkdirRegistryRecord, +) -> Result<()> { + if intent.workspace_id != record.workspace_id + || intent.working_directory_id != record.workdir_id + || intent.runtime_id != record.runtime_id + || intent.repository_id != record.repository_id + || intent.materialization_fingerprint != workdir_materialization_fingerprint(record) + { + return Err(Error::WorkdirAttachmentConflict( + "Workdir authority changed before removal reservation".to_string(), + )); + } + Ok(()) +} + +fn require_operation_materialization( + operation: &WorkdirRemovalOperation, + record: &WorkdirRegistryRecord, +) -> Result<()> { + if operation.workspace_id != record.workspace_id + || operation.working_directory_id != record.workdir_id + || operation.runtime_id != record.runtime_id + || operation.repository_id != record.repository_id + || operation.materialization_fingerprint != workdir_materialization_fingerprint(record) + { + return Err(Error::WorkdirAttachmentConflict( + "Workdir authority changed after removal reservation".to_string(), + )); + } + Ok(()) +} + +fn load_workdir_record( + conn: &Connection, + workspace_id: &str, + workdir_id: &str, +) -> Result> { + conn.query_row( + r#"SELECT workspace_id, workdir_id, runtime_id, repository_id, + creation_selector, creation_ref, creation_tree, + current_selector, current_ref, current_tree, observed_at_epoch_seconds, + materialization_status, cleanliness, created_at, updated_at + FROM workdir_registry WHERE workspace_id=?1 AND workdir_id=?2"#, + params![workspace_id, workdir_id], + |row| { + Ok(WorkdirRegistryRecord { + workspace_id: row.get(0)?, + workdir_id: row.get(1)?, + runtime_id: row.get(2)?, + repository_id: row.get(3)?, + creation_selector: row.get(4)?, + creation_ref: row.get(5)?, + creation_tree: row.get(6)?, + current_selector: row.get(7)?, + current_ref: row.get(8)?, + current_tree: row.get(9)?, + observed_at_epoch_seconds: row.get::<_, Option>(10)?.map(|value| value as u64), + materialization_status: row.get(11)?, + cleanliness: row.get(12)?, + created_at: row.get(13)?, + updated_at: row.get(14)?, + }) + }, + ) + .optional() + .map_err(Error::from) +} + +fn require_operation( + conn: &Connection, + workspace_id: &str, + operation_id: &str, + request_fingerprint: &str, +) -> Result { + let operation = load_operation(conn, workspace_id, operation_id)?.ok_or_else(|| { + Error::InvalidInput(format!( + "Unknown Workdir removal operation `{operation_id}`" + )) + })?; + if operation.request_fingerprint != request_fingerprint { + return Err(Error::WorkdirAttachmentConflict(format!( + "Workdir removal operation `{operation_id}` fingerprint mismatch" + ))); + } + Ok(operation) +} + +fn load_operation( + conn: &Connection, + workspace_id: &str, + operation_id: &str, +) -> Result> { + conn.query_row( + &format!( + "{} WHERE workspace_id=?1 AND operation_id=?2", + operation_select_sql() + ), + params![workspace_id, operation_id], + read_operation, + ) + .optional() + .map_err(Error::from) +} + +fn operation_select_sql() -> &'static str { + r#"SELECT operation_id, request_fingerprint, workspace_id, workdir_id, runtime_id, + repository_id, materialization_fingerprint, source_actor, reason, state, + attempt_count, retryable, disposition, failure_category, + created_at, updated_at, completed_at + FROM workdir_removal_operations"# +} + +fn read_operation(row: &rusqlite::Row<'_>) -> rusqlite::Result { + let state = parse_state(&row.get::<_, String>(9)?)?; + let disposition = row + .get::<_, Option>(12)? + .map(|value| parse_disposition(&value)) + .transpose()?; + let attempt_count = row.get::<_, i64>(10)?; + Ok(WorkdirRemovalOperation { + operation_id: row.get(0)?, + request_fingerprint: row.get(1)?, + workspace_id: row.get(2)?, + working_directory_id: row.get(3)?, + runtime_id: row.get(4)?, + repository_id: row.get(5)?, + materialization_fingerprint: row.get(6)?, + source_actor: row.get(7)?, + reason: row.get(8)?, + state, + attempt_count: attempt_count + .try_into() + .map_err(|_| rusqlite::Error::IntegralValueOutOfRange(10, attempt_count))?, + retryable: row.get(11)?, + disposition, + failure_category: row.get(13)?, + created_at: row.get(14)?, + updated_at: row.get(15)?, + completed_at: row.get(16)?, + }) +} + +fn parse_state(value: &str) -> rusqlite::Result { + match value { + "pending" => Ok(WorkdirRemovalOperationState::Pending), + "failed" => Ok(WorkdirRemovalOperationState::Failed), + "completed" => Ok(WorkdirRemovalOperationState::Completed), + _ => Err(invalid_enum(9, value)), + } +} + +fn parse_disposition(value: &str) -> rusqlite::Result { + match value { + "removed" => Ok(WorkdirRemovalDisposition::Removed), + "retained" => Ok(WorkdirRemovalDisposition::Retained), + "attention_required" => Ok(WorkdirRemovalDisposition::AttentionRequired), + _ => Err(invalid_enum(12, value)), + } +} + +fn invalid_enum(column: usize, value: &str) -> rusqlite::Error { + rusqlite::Error::FromSqlConversionFailure( + column, + rusqlite::types::Type::Text, + format!("invalid Workdir removal value `{value}`").into(), + ) +} + +fn hex_sha256(bytes: &[u8]) -> String { + Sha256::digest(bytes) + .iter() + .map(|byte| format!("{byte:02x}")) + .collect() +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::store::{AccountRecord, ControlPlaneStore, RepositoryRecord, WorkspaceRecord}; + use workspace_api::{RepositoryObservedStatus, RepositorySource, RepositorySourceKind}; + + async fn seeded_store() -> (SqliteWorkspaceStore, WorkdirRegistryRecord) { + let store = SqliteWorkspaceStore::in_memory().unwrap(); + store + .upsert_account(&AccountRecord { + account_id: "account-a".to_string(), + kind: "user".to_string(), + handle: "owner".to_string(), + display_name: "Owner".to_string(), + created_at: "1".to_string(), + updated_at: "1".to_string(), + }) + .unwrap(); + store + .upsert_workspace(&WorkspaceRecord { + workspace_id: "workspace-a".to_string(), + owner_account_id: "account-a".to_string(), + display_name: "Workspace A".to_string(), + state: "active".to_string(), + created_at: "1".to_string(), + updated_at: "1".to_string(), + }) + .await + .unwrap(); + store + .upsert_repository(&RepositoryRecord { + workspace_id: "workspace-a".to_string(), + repository_id: "repository-a".to_string(), + name: "Repository A".to_string(), + kind: "git".to_string(), + provider: Some("local".to_string()), + source: RepositorySource { + kind: RepositorySourceKind::LocalPath, + uri: "/repository-a".to_string(), + }, + default_ref: Some("develop".to_string()), + source_revision: 1, + source_fingerprint: "source-a".to_string(), + observed_status: RepositoryObservedStatus::Unverified, + observed_at: None, + created_at: "1".to_string(), + updated_at: "1".to_string(), + }) + .unwrap(); + let workdir = WorkdirRegistryRecord { + workspace_id: "workspace-a".to_string(), + workdir_id: "workdir-a".to_string(), + runtime_id: "runtime-a".to_string(), + repository_id: "repository-a".to_string(), + creation_selector: Some("refs/heads/develop".to_string()), + creation_ref: Some("abc".to_string()), + creation_tree: Some("tree-a".to_string()), + current_selector: Some("refs/heads/work".to_string()), + current_ref: Some("def".to_string()), + current_tree: Some("tree-b".to_string()), + observed_at_epoch_seconds: Some(1), + materialization_status: "present".to_string(), + cleanliness: "clean".to_string(), + created_at: "1".to_string(), + updated_at: "1".to_string(), + }; + store.upsert_workdir_registry(&workdir).unwrap(); + (store, workdir) + } + + #[tokio::test] + async fn exact_replay_reuses_operation_and_conflicting_intent_fails() { + let (store, workdir) = seeded_store().await; + let intent = + workdir_removal_intent(&workdir, "worker:W-1", "remove stale Workdir").unwrap(); + let first = store.reserve_workdir_removal_operation(&intent).unwrap(); + let replay = store.reserve_workdir_removal_operation(&intent).unwrap(); + assert_eq!(replay, first); + + let mut conflict = intent.clone(); + conflict.reason = "different intent".to_string(); + conflict.request_fingerprint = "f".repeat(64); + let error = store + .reserve_workdir_removal_operation(&conflict) + .unwrap_err(); + assert!(matches!(error, Error::WorkdirAttachmentConflict(_))); + } + + #[tokio::test] + async fn failed_attempt_is_retryable_and_completed_retry_replays() { + let (store, workdir) = seeded_store().await; + let intent = + workdir_removal_intent(&workdir, "workspace-api", "remove clean Workdir").unwrap(); + let reserved = store.reserve_workdir_removal_operation(&intent).unwrap(); + let first = store + .begin_workdir_removal_attempt( + &reserved.workspace_id, + &reserved.operation_id, + &reserved.request_fingerprint, + ) + .unwrap(); + assert_eq!(first.attempt_count, 1); + let failed = store + .fail_workdir_removal_operation(&first, "runtime_unavailable", true) + .unwrap(); + assert_eq!(failed.state, WorkdirRemovalOperationState::Failed); + let retry = store + .begin_workdir_removal_attempt( + &failed.workspace_id, + &failed.operation_id, + &failed.request_fingerprint, + ) + .unwrap(); + assert_eq!(retry.attempt_count, 2); + let completed = store.commit_workdir_removal_removed(&retry).unwrap(); + assert_eq!( + completed.disposition, + Some(WorkdirRemovalDisposition::Removed) + ); + assert!( + store + .get_workdir_registry("workspace-a", "workdir-a") + .unwrap() + .is_none() + ); + let replay = store + .begin_workdir_removal_attempt( + &completed.workspace_id, + &completed.operation_id, + &completed.request_fingerprint, + ) + .unwrap(); + assert_eq!(replay, completed); + } + + #[tokio::test] + async fn retained_completion_keeps_registry_and_is_auditable() { + let (store, workdir) = seeded_store().await; + let intent = + workdir_removal_intent(&workdir, "workspace-api", "inspect dirty Workdir").unwrap(); + let operation = store.reserve_workdir_removal_operation(&intent).unwrap(); + let retained = store + .complete_workdir_removal_retained( + &operation, + WorkdirRemovalDisposition::Retained, + "dirty_or_unknown", + ) + .unwrap(); + assert_eq!(retained.state, WorkdirRemovalOperationState::Completed); + assert_eq!( + retained.failure_category.as_deref(), + Some("dirty_or_unknown") + ); + assert!( + store + .get_workdir_registry("workspace-a", "workdir-a") + .unwrap() + .is_some() + ); + } +}