diff --git a/crates/client/src/backend_runtime.rs b/crates/client/src/backend_runtime.rs index 8ec2ea84a..6809c6409 100644 --- a/crates/client/src/backend_runtime.rs +++ b/crates/client/src/backend_runtime.rs @@ -18,7 +18,8 @@ pub use workspace_api::{ WorkerLaunchRuntimeOption as BackendWorkerLaunchRuntimeOption, WorkerOperationState as BackendWorkerOperationState, WorkerRestoreResponse as BackendWorkerRestoreResponse, - WorkerRestoreResult as BackendWorkerRestoreResult, WorkerSummary as BackendWorkerSummary, + WorkerRestoreResult as BackendWorkerRestoreResult, + WorkerRestoreState as BackendWorkerRestoreState, WorkerSummary as BackendWorkerSummary, WorkerWorkspaceSummary as BackendWorkerWorkspaceSummary, WorkingDirectoryCreateRequest as BackendWorkingDirectoryCreateRequest, WorkingDirectoryCreateResponse as BackendWorkingDirectoryCreateResponse, diff --git a/crates/client/src/lib.rs b/crates/client/src/lib.rs index 814e08cdf..1bb4b8d74 100644 --- a/crates/client/src/lib.rs +++ b/crates/client/src/lib.rs @@ -27,10 +27,10 @@ pub use backend_runtime::{ BackendWorkerCapabilitySummary, BackendWorkerImplementationSummary, BackendWorkerLaunchOptions, BackendWorkerLaunchProfileCandidate, BackendWorkerLaunchRuntimeOption, BackendWorkerLaunchTarget, BackendWorkerOperationState, BackendWorkerRestoreResponse, - BackendWorkerRestoreResult, BackendWorkerSummary, BackendWorkerWorkspaceSummary, - BackendWorkingDirectorySummary, connect_backend_runtime, create_backend_worker, - get_backend_worker_launch_options, list_backend_stopped_workers, list_backend_workers, - restore_backend_worker, + BackendWorkerRestoreResult, BackendWorkerRestoreState, BackendWorkerSummary, + BackendWorkerWorkspaceSummary, BackendWorkingDirectorySummary, connect_backend_runtime, + create_backend_worker, get_backend_worker_launch_options, list_backend_stopped_workers, + list_backend_workers, restore_backend_worker, }; pub use backend_workspace::{ BackendWorkspace, BackendWorkspaceCatalogTarget, BackendWorkspaceClientError, diff --git a/crates/tui/src/backend_worker_picker.rs b/crates/tui/src/backend_worker_picker.rs index ac7b2a7bb..34ce59560 100644 --- a/crates/tui/src/backend_worker_picker.rs +++ b/crates/tui/src/backend_worker_picker.rs @@ -3,7 +3,7 @@ use std::io; use std::time::Duration; use client::{ - BackendRuntimeListTarget, BackendWorkerOperationState, BackendWorkerRestoreResponse, + BackendRuntimeListTarget, BackendWorkerRestoreResponse, BackendWorkerRestoreState, BackendWorkerSummary, list_backend_stopped_workers, list_backend_workers, restore_backend_worker, }; @@ -110,7 +110,7 @@ pub(crate) async fn run( } fn restored_worker(response: BackendWorkerRestoreResponse) -> Result { - if response.result.state != BackendWorkerOperationState::Accepted { + if response.result.state != BackendWorkerRestoreState::Accepted { let diagnostics = response .result .diagnostics @@ -119,9 +119,10 @@ fn restored_worker(response: BackendWorkerRestoreResponse) -> Result>() .join("; "); let state = match response.result.state { - BackendWorkerOperationState::Accepted => unreachable!(), - BackendWorkerOperationState::Rejected => "rejected", - BackendWorkerOperationState::Unsupported => "unsupported", + BackendWorkerRestoreState::Accepted => unreachable!(), + BackendWorkerRestoreState::Rejected => "rejected", + BackendWorkerRestoreState::RolledBack => "rolled back", + BackendWorkerRestoreState::ReconciliationRequired => "reconciliation required", }; return Err(if diagnostics.is_empty() { format!("restore was {state} without a diagnostic") @@ -494,7 +495,7 @@ mod tests { } fn restore_response( - state: BackendWorkerOperationState, + state: BackendWorkerRestoreState, worker: Option, diagnostics: Vec, ) -> BackendWorkerRestoreResponse { @@ -513,7 +514,7 @@ mod tests { #[test] fn rejected_restore_surfaces_diagnostic_instead_of_attaching_selected_worker() { let error = restored_worker(restore_response( - BackendWorkerOperationState::Rejected, + BackendWorkerRestoreState::Rejected, None, vec![BackendDiagnostic { code: "working_directory_not_found".to_string(), @@ -532,7 +533,7 @@ mod tests { #[test] fn accepted_restore_requires_returned_worker_snapshot() { let error = restored_worker(restore_response( - BackendWorkerOperationState::Accepted, + BackendWorkerRestoreState::Accepted, None, Vec::new(), )) @@ -545,7 +546,7 @@ mod tests { fn accepted_restore_returns_authoritative_worker_snapshot() { let worker = worker("runtime-a", "worker-a", Some("builtin:companion")); let restored = restored_worker(restore_response( - BackendWorkerOperationState::Accepted, + BackendWorkerRestoreState::Accepted, Some(worker.clone()), Vec::new(), )) diff --git a/crates/worker-runtime/src/execution.rs b/crates/worker-runtime/src/execution.rs index 0b0aaae34..dab47f3f3 100644 --- a/crates/worker-runtime/src/execution.rs +++ b/crates/worker-runtime/src/execution.rs @@ -286,7 +286,21 @@ pub enum WorkerExecutionSpawnResult { worker_state: protocol::WorkerStateSnapshot, working_directory: Option, }, + /// A read-only preflight rejected the operation before live work started. Rejected(WorkerExecutionResult), + /// Live work started, failed, and all operation-owned resources were joined/released. + RolledBack(WorkerExecutionResult), + /// The operation crossed the live side-effect boundary and could not prove + /// either commit or complete cleanup. Optional handle/state fields preserve + /// concrete execution evidence for retry reconciliation. + ReconciliationRequired { + result: WorkerExecutionResult, + handle: Option, + worker_state: Option, + working_directory: Option, + }, + /// Legacy spawn failure. Restore implementations must use `RolledBack` or + /// `ReconciliationRequired` after crossing their live side-effect boundary. Errored(WorkerExecutionResult), } @@ -316,6 +330,15 @@ pub trait WorkerExecutionBackend: Send + Sync + 'static { fn spawn_worker(&self, request: WorkerExecutionSpawnRequest) -> WorkerExecutionSpawnResult; + /// Read-only restore validation performed before any live controller, + /// bridge, task, or external registration is created. + fn preflight_restore( + &self, + _request: &WorkerExecutionRestoreRequest, + ) -> Result<(), WorkerExecutionResult> { + Ok(()) + } + fn restore_worker( &self, _request: WorkerExecutionRestoreRequest, @@ -509,6 +532,13 @@ impl WorkerExecutionBackendRef { self.backend.spawn_worker(request) } + pub(crate) fn preflight_restore( + &self, + request: &WorkerExecutionRestoreRequest, + ) -> Result<(), WorkerExecutionResult> { + self.backend.preflight_restore(request) + } + pub(crate) fn restore_worker( &self, request: WorkerExecutionRestoreRequest, diff --git a/crates/worker-runtime/src/http_server.rs b/crates/worker-runtime/src/http_server.rs index 9f631ca3f..fadffb45c 100644 --- a/crates/worker-runtime/src/http_server.rs +++ b/crates/worker-runtime/src/http_server.rs @@ -74,6 +74,7 @@ use workdir::{ WorkdirTransportError, WorkdirTransportErrorCode, }, }; +use workspace_api::WorkerRestoreState; const DEFAULT_RUNTIME_HTTP_PORT: u16 = 38800; pub const RUNTIME_HTTP_PROTOCOL_MIN_VERSION: u32 = 1; @@ -420,6 +421,20 @@ pub struct RuntimeHttpWorkerResponse { pub worker: WorkerDetail, } +/// Explicit restore outcome. This envelope is returned for every reachable +/// Runtime restore attempt, including preflight rejection and uncertain commit. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct RuntimeHttpWorkerRestoreResponse { + pub state: WorkerRestoreState, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub worker: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub reason_code: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub message: Option, +} + /// Replace the Workspace API binding for an existing Worker. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct RuntimeHttpWorkerWorkspaceApiRequest { @@ -1199,14 +1214,21 @@ async fn restore_worker( State(state): State, auth: Option>, Path(worker_id): Path, -) -> RestResult { +) -> RestResult { let worker_ref = worker_ref_for(&state.runtime, worker_id)?; - let worker = match auth_workspace_scope(&state, auth.as_ref())? { - Some(scope) => state.runtime.restore_worker_scoped(&scope, &worker_ref), - None => state.runtime.restore_worker(&worker_ref), + let result = match auth_workspace_scope(&state, auth.as_ref())? { + Some(scope) => state + .runtime + .restore_worker_operation_scoped(&scope, &worker_ref), + None => state.runtime.restore_worker_operation(&worker_ref), } .map_err(RuntimeHttpRestError::runtime)?; - Ok(Json(RuntimeHttpWorkerResponse { worker })) + Ok(Json(RuntimeHttpWorkerRestoreResponse { + state: result.state, + worker: result.worker, + reason_code: result.reason_code, + message: result.message, + })) } #[cfg(feature = "ws-server")] @@ -3322,8 +3344,9 @@ mod tests { ) .await; assert_eq!(response.status(), StatusCode::OK); - let restored: RuntimeHttpWorkerResponse = read_json(response).await; - assert_eq!(restored.worker.status, WorkerStatus::Idle); + let restored: RuntimeHttpWorkerRestoreResponse = read_json(response).await; + assert_eq!(restored.state, WorkerRestoreState::Accepted); + assert_eq!(restored.worker.unwrap().status, WorkerStatus::Idle); let response = authed_empty_request( app.clone(), diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index 18352c690..b00ecde8a 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -54,6 +54,7 @@ use std::sync::{Arc, Mutex, MutexGuard, Weak}; use tokio::sync::broadcast; use tokio::sync::mpsc; use uuid::Uuid; +use workspace_api::WorkerRestoreState; /// Workspace-scoped Runtime authorization context supplied by a trusted backend. #[derive(Clone, Debug, Eq, PartialEq)] @@ -83,6 +84,47 @@ enum WorkerRestoreMode { Automatic, } +/// Runtime-internal restore result consumed by both embedded and HTTP +/// providers. Diagnostics are stable and path-free; detailed backend errors +/// remain in Runtime logs rather than crossing the API boundary. +#[derive(Clone, Debug)] +pub struct RuntimeWorkerRestoreResult { + pub state: WorkerRestoreState, + pub worker: Option, + pub reason_code: Option, + pub message: Option, +} + +impl RuntimeWorkerRestoreResult { + fn accepted(worker: WorkerDetail) -> Self { + Self { + state: WorkerRestoreState::Accepted, + worker: Some(worker), + reason_code: None, + message: None, + } + } + + fn failed(state: WorkerRestoreState, reason_code: &'static str, message: &'static str) -> Self { + Self { + state, + worker: None, + reason_code: Some(reason_code.to_string()), + message: Some(message.to_string()), + } + } + + fn into_legacy_result(self) -> Result { + match (self.state, self.worker) { + (WorkerRestoreState::Accepted, Some(worker)) => Ok(worker), + _ => Err(RuntimeError::InvalidRequest( + self.message + .unwrap_or_else(|| "Worker restore did not commit".to_string()), + )), + } + } +} + #[derive(Clone, Debug)] pub struct RuntimeSubscriptionUpdate { pub subject_revision: u64, @@ -1047,6 +1089,7 @@ impl Runtime { working_directory, } => (handle, worker_state, working_directory), WorkerExecutionSpawnResult::Rejected(result) + | WorkerExecutionSpawnResult::RolledBack(result) | WorkerExecutionSpawnResult::Errored(result) => { self.rollback_failed_create(&worker_ref)?; return Err(RuntimeError::WorkerExecutionRejected { @@ -1057,6 +1100,16 @@ impl Runtime { result, }); } + WorkerExecutionSpawnResult::ReconciliationRequired { result, .. } => { + self.rollback_failed_create(&worker_ref)?; + return Err(RuntimeError::WorkerExecutionRejected { + worker_id: worker_ref.worker_id.clone(), + operation: result.operation, + outcome: result.outcome, + message: result.message_or_default(), + result, + }); + } }; if let Some(mut initial_input) = { @@ -1377,7 +1430,16 @@ impl Runtime { scope: &RuntimeWorkspaceScope, worker_ref: &WorkerRef, ) -> Result { - self.restore_worker_with_scope(worker_ref, Some(scope)) + self.restore_worker_operation_scoped(scope, worker_ref)? + .into_legacy_result() + } + + pub fn restore_worker_operation_scoped( + &self, + scope: &RuntimeWorkspaceScope, + worker_ref: &WorkerRef, + ) -> Result { + self.restore_worker_operation_with_scope(worker_ref, Some(scope)) } /// Attach a live execution to a persisted Worker definition. @@ -1387,14 +1449,22 @@ impl Runtime { /// converges on its already-installed execution rather than spawning another /// controller. pub fn restore_worker(&self, worker_ref: &WorkerRef) -> Result { - self.restore_worker_with_scope(worker_ref, None) + self.restore_worker_operation(worker_ref)? + .into_legacy_result() } - fn restore_worker_with_scope( + pub fn restore_worker_operation( + &self, + worker_ref: &WorkerRef, + ) -> Result { + self.restore_worker_operation_with_scope(worker_ref, None) + } + + fn restore_worker_operation_with_scope( &self, worker_ref: &WorkerRef, scope: Option<&RuntimeWorkspaceScope>, - ) -> Result { + ) -> Result { let operation_lock = self.worker_operation_lock(worker_ref.worker_id)?; let _operation_guard = operation_lock .lock() @@ -1409,58 +1479,54 @@ impl Runtime { &self, worker_ref: &WorkerRef, mode: WorkerRestoreMode, - ) -> Result { + ) -> Result { let (backend, request) = { - let mut state = self.lock()?; + let state = self.lock()?; state.ensure_running()?; - let (worker_request, previous_working_directory) = { - let worker = state.worker(worker_ref)?; - if worker.execution_handle.is_some() { - if worker.status.is_active() { - return Ok(worker.detail()); - } - return Err(RuntimeError::WorkerExecutionUnavailable { - worker_id: worker_ref.worker_id, - message: "previous execution cleanup has not completed".to_string(), - }); + let worker = state.worker(worker_ref)?; + if worker.execution_handle.is_some() { + if worker.status.is_active() { + return Ok(RuntimeWorkerRestoreResult::accepted(worker.detail())); } - match mode { - WorkerRestoreMode::Explicit - if worker.execution_metadata_available - && worker.status != WorkerStatus::Stopped => - { - return Err(RuntimeError::InvalidRequest(format!( - "worker {} is not stopped", - worker_ref.worker_id - ))); - } - WorkerRestoreMode::Automatic - if !worker.status.is_active() - || worker.restore_intent != WorkerRestoreIntent::Automatic => - { - return Ok(worker.detail()); - } - _ => {} - } - let request = worker.request.clone().ok_or_else(|| { - RuntimeError::WorkerExecutionUnavailable { - worker_id: worker.worker_id, - message: "persisted Worker restore request is unavailable".to_string(), - } - })?; - (request, worker.working_directory.clone()) - }; - let backend = state.execution_backend.clone().ok_or_else(|| { - RuntimeError::WorkerExecutionUnavailable { - worker_id: worker_ref.worker_id, - message: "runtime has no execution backend".to_string(), - } - })?; - { - let worker = state.worker_mut(worker_ref)?; - worker.execution_bound = true; + return Ok(RuntimeWorkerRestoreResult::failed( + WorkerRestoreState::ReconciliationRequired, + "worker_restore_cleanup_pending", + "Worker restore cleanup or reconciliation is still pending", + )); } - state.persist_worker(&worker_ref.worker_id)?; + match mode { + WorkerRestoreMode::Explicit + if worker.execution_metadata_available + && worker.status != WorkerStatus::Stopped => + { + return Ok(RuntimeWorkerRestoreResult::failed( + WorkerRestoreState::Rejected, + "worker_restore_not_stopped", + "Worker restore requires a stopped Worker", + )); + } + WorkerRestoreMode::Automatic + if !worker.status.is_active() + || worker.restore_intent != WorkerRestoreIntent::Automatic => + { + return Ok(RuntimeWorkerRestoreResult::accepted(worker.detail())); + } + _ => {} + } + let Some(worker_request) = worker.request.clone() else { + return Ok(RuntimeWorkerRestoreResult::failed( + WorkerRestoreState::Rejected, + "worker_restore_metadata_unavailable", + "Persisted Worker restore metadata is unavailable", + )); + }; + let Some(backend) = state.execution_backend.clone() else { + return Ok(RuntimeWorkerRestoreResult::failed( + WorkerRestoreState::Rejected, + "worker_restore_backend_unavailable", + "Runtime has no Worker execution backend", + )); + }; let workspace_scope = worker_request.workspace_api.as_ref().and_then(|api| { state .workspace_owners @@ -1472,13 +1538,21 @@ impl Runtime { request: worker_request, workspace_scope, context: self.execution_context(worker_ref.clone()), - previous_working_directory, + previous_working_directory: worker.working_directory.clone(), working_directory: None, config_bundle: None, }; (backend, request) }; + if let Err(_result) = backend.preflight_restore(&request) { + return Ok(RuntimeWorkerRestoreResult::failed( + WorkerRestoreState::Rejected, + "worker_restore_preflight_rejected", + "Worker restore preflight rejected the current request", + )); + } + match backend.restore_worker(request) { WorkerExecutionSpawnResult::Connected { handle, @@ -1488,30 +1562,118 @@ impl Runtime { let commit = self.commit_restored_worker_execution( worker_ref, handle.clone(), - worker_state, + worker_state.clone(), WorkerStatus::Idle, - working_directory, + working_directory.clone(), ); - if let Err(error) = commit { - self.cleanup_failed_restore(&backend, worker_ref, &handle)?; - return Err(error); + match commit { + Ok(worker) => Ok(RuntimeWorkerRestoreResult::accepted(worker)), + Err(error) => { + match self.cleanup_failed_restore(&backend, worker_ref, &handle) { + Ok(()) => { + tracing::warn!( + worker_id = %worker_ref.worker_id, + error = %error, + "Worker restore commit failed and was rolled back" + ); + Ok(RuntimeWorkerRestoreResult::failed( + WorkerRestoreState::RolledBack, + "worker_restore_commit_rolled_back", + "Worker restore commit failed and cleanup completed", + )) + } + Err(cleanup_error) => { + tracing::error!( + worker_id = %worker_ref.worker_id, + error = %error, + cleanup_error = %cleanup_error, + "Worker restore requires reconciliation after commit and cleanup failures" + ); + if let Err(retain_error) = self.retain_restore_execution_evidence( + worker_ref, + handle, + worker_state, + working_directory, + ) { + tracing::error!( + worker_id = %worker_ref.worker_id, + error = %retain_error, + "failed to persist restore execution evidence; in-process reconciliation remains required" + ); + } + Ok(RuntimeWorkerRestoreResult::failed( + WorkerRestoreState::ReconciliationRequired, + "worker_restore_reconciliation_required", + "Worker restore result is uncertain; reread or retry the Worker restore", + )) + } + } + } } - self.worker_detail(worker_ref) } - WorkerExecutionSpawnResult::Rejected(result) - | WorkerExecutionSpawnResult::Errored(result) => { + WorkerExecutionSpawnResult::Rejected(_result) => { + Ok(RuntimeWorkerRestoreResult::failed( + WorkerRestoreState::Rejected, + "worker_restore_rejected", + "Worker restore was rejected before live work started", + )) + } + WorkerExecutionSpawnResult::RolledBack(_result) => { #[cfg(feature = "fs-store")] + self.lock()?.record_restore_failure(worker_ref, _result)?; + Ok(RuntimeWorkerRestoreResult::failed( + WorkerRestoreState::RolledBack, + "worker_restore_rolled_back", + "Worker restore failed after live work started; cleanup completed", + )) + } + WorkerExecutionSpawnResult::ReconciliationRequired { + result: _result, + handle, + worker_state, + working_directory, + } => { + if let (Some(handle), Some(worker_state)) = (handle, worker_state) { + if let Err(retain_error) = self.retain_restore_execution_evidence( + worker_ref, + handle, + worker_state, + working_directory, + ) { + tracing::error!( + worker_id = %worker_ref.worker_id, + error = %retain_error, + "failed to persist restore execution evidence; in-process reconciliation remains required" + ); + } + } else if let Err(retain_error) = + self.retain_restore_reconciliation_pending(worker_ref) { - self.lock()? - .record_restore_failure(worker_ref, result.clone())?; + tracing::error!( + worker_id = %worker_ref.worker_id, + error = %retain_error, + "failed to persist pending restore reconciliation marker" + ); } - Err(RuntimeError::WorkerExecutionRejected { - worker_id: worker_ref.worker_id, - operation: result.operation, - outcome: result.outcome, - message: result.message_or_default(), - result, - }) + Ok(RuntimeWorkerRestoreResult::failed( + WorkerRestoreState::ReconciliationRequired, + "worker_restore_reconciliation_required", + "Worker restore result is uncertain; reread or retry the Worker restore", + )) + } + WorkerExecutionSpawnResult::Errored(_result) => { + if let Err(retain_error) = self.retain_restore_reconciliation_pending(worker_ref) { + tracing::error!( + worker_id = %worker_ref.worker_id, + error = %retain_error, + "failed to persist pending restore reconciliation marker" + ); + } + Ok(RuntimeWorkerRestoreResult::failed( + WorkerRestoreState::ReconciliationRequired, + "worker_restore_backend_outcome_unknown", + "Worker restore backend outcome is uncertain; reread or retry the Worker restore", + )) } } } @@ -2469,27 +2631,83 @@ impl Runtime { worker_state: protocol::WorkerStateSnapshot, status: WorkerStatus, working_directory: Option, - ) -> Result<(), RuntimeError> { + ) -> Result { let mut state = self.lock()?; state.ensure_worker_ref(worker_ref)?; - { - let worker = state.worker_mut(worker_ref)?; - if worker.execution_handle.is_some() { - return Err(RuntimeError::InvalidRequest(format!( - "worker {} already has a current execution", - worker_ref.worker_id - ))); - } - worker.execution_handle = Some(handle); - worker.execution_metadata_available = true; - worker.execution_bound = true; - worker.status = status; - let _ = worker.apply_worker_state(&worker_state); - worker.restore_intent = restore_intent_for_status(worker.status); - worker.working_directory = working_directory; + let mut candidate = state.worker(worker_ref)?.clone(); + if candidate.execution_handle.is_some() { + return Err(RuntimeError::InvalidRequest(format!( + "worker {} already has a current execution", + worker_ref.worker_id + ))); } + candidate.execution_handle = Some(handle); + candidate.execution_metadata_available = true; + candidate.execution_bound = true; + candidate.status = status; + let _ = candidate.apply_worker_state(&worker_state); + candidate.restore_intent = restore_intent_for_status(candidate.status); + candidate.working_directory = working_directory; + + // Validate the exact subscription projection before persistence. The + // subsequent publish uses the same pure projection while the Worker + // operation lock keeps this candidate stable. + let _subscription_projection = state.subscription_worker(&candidate)?; + + // Persist the complete candidate before making it observable. A failed + // write therefore leaves both the in-memory catalog and subscriptions + // on the pre-restore record, so cleanup can produce a true rollback. + state.persist_worker_record(&candidate)?; + let detail = candidate.detail(); + state.workers.insert(worker_ref.worker_id, candidate); + state.publish_worker_upsert(worker_ref.worker_id)?; + if let Err(error) = state.persist_runtime_snapshot() { + tracing::warn!( + worker_id = %worker_ref.worker_id, + error = %error, + "Worker restore committed to its canonical record but Runtime snapshot refresh failed" + ); + } + Ok(detail) + } + + fn retain_restore_execution_evidence( + &self, + worker_ref: &WorkerRef, + handle: WorkerExecutionHandle, + worker_state: protocol::WorkerStateSnapshot, + working_directory: Option, + ) -> Result<(), RuntimeError> { + let mut state = self.lock()?; + let mut candidate = state.worker(worker_ref)?.clone(); + candidate.execution_handle = Some(handle); + candidate.execution_metadata_available = true; + candidate.execution_bound = true; + candidate.status = WorkerStatus::Idle; + let _ = candidate.apply_worker_state(&worker_state); + candidate.restore_intent = WorkerRestoreIntent::Automatic; + candidate.working_directory = working_directory; + state.workers.insert(worker_ref.worker_id, candidate); + state.publish_worker_upsert(worker_ref.worker_id)?; + state.persist_worker(&worker_ref.worker_id)?; + Ok(()) + } + + fn retain_restore_reconciliation_pending( + &self, + worker_ref: &WorkerRef, + ) -> Result<(), RuntimeError> { + let mut state = self.lock()?; + let mut candidate = state.worker(worker_ref)?.clone(); + candidate.execution_handle = None; + candidate.execution_metadata_available = false; + candidate.execution_bound = true; + // Active-without-metadata is the durable pending-reconciliation marker: + // it prevents a false Stopped projection and blocks duplicate restore. + candidate.status = WorkerStatus::Idle; + candidate.restore_intent = WorkerRestoreIntent::Explicit; + state.workers.insert(worker_ref.worker_id, candidate); state.publish_worker_upsert(worker_ref.worker_id)?; - state.persist_runtime_snapshot()?; state.persist_worker(&worker_ref.worker_id)?; Ok(()) } @@ -2513,18 +2731,6 @@ impl Runtime { result, }); } - - let mut state = self.lock()?; - let worker = state.worker_mut(worker_ref)?; - worker.execution_handle = None; - worker.execution_bound = false; - worker.status = WorkerStatus::Stopped; - worker.worker_state = None; - worker.restore_intent = WorkerRestoreIntent::Explicit; - worker.internal_workers.clear(); - state.publish_worker_upsert(worker_ref.worker_id)?; - state.persist_runtime_snapshot()?; - state.persist_worker(&worker_ref.worker_id)?; Ok(()) } @@ -2908,6 +3114,14 @@ impl RuntimeState { Ok(()) } + #[cfg(feature = "fs-store")] + fn persist_worker_record(&self, worker: &WorkerRecord) -> Result<(), RuntimeError> { + if let Some(store) = self.fs_store() { + store.write_worker_record(&worker.persisted_record())?; + } + Ok(()) + } + #[cfg(feature = "fs-store")] fn persist_worker(&self, worker_id: &WorkerId) -> Result<(), RuntimeError> { if let Some(store) = self.fs_store() { @@ -2945,6 +3159,11 @@ impl RuntimeState { Ok(()) } + #[cfg(not(feature = "fs-store"))] + fn persist_worker_record(&self, _worker: &WorkerRecord) -> Result<(), RuntimeError> { + Ok(()) + } + #[cfg(not(feature = "fs-store"))] fn persist_worker(&self, _worker_id: &WorkerId) -> Result<(), RuntimeError> { Ok(()) @@ -3689,7 +3908,7 @@ struct RemovedInternalWorkerActivity { parent_session_id: Option, } -#[derive(Debug, Default)] +#[derive(Debug, Default, Clone)] struct InternalWorkerActivityProjection { workers: BTreeMap, removed_workers: BTreeMap, @@ -3708,7 +3927,7 @@ impl InternalWorkerActivityProjection { } } -#[derive(Debug)] +#[derive(Debug, Clone)] struct WorkerRecord { worker_ref: WorkerRef, worker_id: WorkerId, @@ -4998,6 +5217,7 @@ mod tests { stop_gate: Mutex>>, stop_count: Mutex, restore_result: Mutex>, + preflight_restore_result: Mutex>, restore_gate: Mutex>>, restore_count: Mutex, config_bundles: Mutex>>, @@ -5117,6 +5337,16 @@ mod tests { } } + fn preflight_restore( + &self, + _request: &WorkerExecutionRestoreRequest, + ) -> Result<(), WorkerExecutionResult> { + match self.preflight_restore_result.lock().unwrap().clone() { + Some(result) => Err(result), + None => Ok(()), + } + } + fn restore_worker( &self, request: WorkerExecutionRestoreRequest, @@ -7229,6 +7459,72 @@ mod tests { let _ = std::fs::remove_dir_all(root); } + #[test] + fn restore_preflight_rejection_is_typed_and_side_effect_free() { + let (runtime, backend) = runtime_and_backend(); + runtime.store_config_bundle(test_bundle()).unwrap(); + let created = runtime + .create_worker(task_request("restore preflight")) + .unwrap(); + runtime.stop_worker(&created.worker_ref, None).unwrap(); + let before = runtime.worker_detail(&created.worker_ref).unwrap(); + backend + .preflight_restore_result + .lock() + .unwrap() + .replace(WorkerExecutionResult::rejected( + WorkerExecutionOperation::Restore, + "profile is no longer permitted", + )); + + let result = runtime + .restore_worker_operation(&created.worker_ref) + .unwrap(); + + assert_eq!(result.state, WorkerRestoreState::Rejected); + assert!(result.worker.is_none()); + assert_eq!(*backend.restore_count.lock().unwrap(), 0); + assert_eq!(runtime.worker_detail(&created.worker_ref).unwrap(), before); + } + + #[test] + fn uncertain_restore_backend_outcome_requires_reconciliation() { + let (runtime, backend) = runtime_and_backend(); + runtime.store_config_bundle(test_bundle()).unwrap(); + let created = runtime + .create_worker(task_request("restore reconciliation")) + .unwrap(); + runtime.stop_worker(&created.worker_ref, None).unwrap(); + backend.restore_result.lock().unwrap().replace( + WorkerExecutionSpawnResult::ReconciliationRequired { + result: WorkerExecutionResult::errored( + WorkerExecutionOperation::Restore, + "controller handshake outcome was not observable", + ), + handle: None, + worker_state: None, + working_directory: None, + }, + ); + + let result = runtime + .restore_worker_operation(&created.worker_ref) + .unwrap(); + + assert_eq!(result.state, WorkerRestoreState::ReconciliationRequired); + assert!(result.worker.is_none()); + assert_eq!( + runtime.worker_detail(&created.worker_ref).unwrap().status, + WorkerStatus::Idle + ); + backend.restore_result.lock().unwrap().take(); + let retry = runtime + .restore_worker_operation(&created.worker_ref) + .unwrap(); + assert_eq!(retry.state, WorkerRestoreState::Accepted); + assert_eq!(*backend.restore_count.lock().unwrap(), 2); + } + #[cfg(feature = "fs-store")] #[test] fn fs_store_automatically_restores_every_active_lifecycle_state() { @@ -7564,7 +7860,7 @@ mod tests { #[cfg(feature = "fs-store")] #[test] - fn fs_store_stops_worker_and_reports_when_execution_restore_fails() { + fn fs_store_marks_worker_for_reconciliation_when_execution_restore_is_uncertain() { let root = fs_store_root("execution-restore-failed"); let runtime = Runtime::with_fs_store_and_execution_backend( crate::fs_store::FsRuntimeStoreOptions { @@ -7598,27 +7894,20 @@ mod tests { assert_eq!(*restoring_backend.restore_count.lock().unwrap(), 1); let restored_worker = restored.worker_detail(&worker.worker_ref).unwrap(); - assert_eq!(restored_worker.status, WorkerStatus::Stopped); - assert!( - restored - .diagnostics() - .unwrap() - .iter() - .any( - |diagnostic| diagnostic.code == "worker_execution_restore_failed" - && diagnostic.worker_ref.as_ref() == Some(&worker.worker_ref) - ) - ); - let err = restored + assert_eq!(restored_worker.status, WorkerStatus::Idle); + assert!(!restored_worker.execution_metadata_available); + restoring_backend.restore_result.lock().unwrap().take(); + let retry = restored + .restore_worker_operation(&worker.worker_ref) + .unwrap(); + assert_eq!(retry.state, WorkerRestoreState::Accepted); + assert_eq!(*restoring_backend.restore_count.lock().unwrap(), 2); + restored .send_input( &worker.worker_ref, - WorkerInput::user("after failed restore"), + WorkerInput::user("after reconciled restore"), ) - .unwrap_err(); - assert!(matches!( - err, - RuntimeError::WorkerExecutionUnavailable { .. } - )); + .unwrap(); let _ = std::fs::remove_dir_all(root); } diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index e17794e82..ccb564b3b 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -18,7 +18,7 @@ use crate::auth::{BACKEND_RESOURCE_FETCH_PERMISSION, RuntimeIdentityMaterial}; use crate::catalog::{ CreateWorkerRequest, ProfileSourceArchiveSource, RepositoryRefObservation, RepositoryRefObservationRequest, WorkingDirectoryRepositoryAccessRequest, - WorkingDirectoryRequest, WorkingDirectoryStatus, + WorkingDirectoryRequest, WorkingDirectoryStatus, WorkingDirectoryStatusKind, }; use crate::config_bundle::{ConfigBundle, workspace_config_etag}; use crate::execution::{ @@ -59,7 +59,9 @@ fn next_internal_command( .max(floor); Ok(WorkerCommandEnvelope::new(command_id)) } -use session_store::{CombinedStore, WorkerAggregateStore, WorkerSessionStore}; +use session_store::{ + CombinedStore, Store, WorkerAggregateStore, WorkerMetadataStore, WorkerSessionStore, +}; #[cfg(test)] use session_store::{FsStore, FsWorkerStore}; use tokio::runtime::Runtime; @@ -122,6 +124,13 @@ pub trait RuntimeWorkerFactory: Send + Sync + 'static { request: WorkerExecutionSpawnRequest, ) -> Result; + async fn preflight_restore( + &self, + _request: &WorkerExecutionRestoreRequest, + ) -> Result<(), String> { + Ok(()) + } + async fn restore_controller( &self, request: WorkerExecutionRestoreRequest, @@ -1002,6 +1011,56 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory { }) } + async fn preflight_restore( + &self, + request: &WorkerExecutionRestoreRequest, + ) -> Result<(), String> { + let worker_name = Self::runtime_worker_name_for_ref(&request.worker_ref); + let (mut manifest, _) = Self::restore_fallback_manifest(&worker_name)?; + bind_workspace_memory_settings(&mut manifest, &request.request)?; + let worker_aggregate_dir = self.worker_aggregate_dir(&request.worker_ref)?; + if !worker_aggregate_dir.is_dir() { + return Err("Persisted Worker aggregate metadata is unavailable".to_string()); + } + if !worker_aggregate_dir.join("session").is_dir() { + return Err("Persisted Worker Session metadata is unavailable".to_string()); + } + let metadata_store = WorkerAggregateStore::new(&worker_aggregate_dir, &worker_name) + .map_err(|error| format!("failed to read Worker aggregate metadata: {error}"))?; + let metadata = metadata_store + .read_by_name(&worker_name) + .map_err(|error| format!("failed to read Worker metadata: {error}"))? + .ok_or_else(|| "Persisted Worker metadata is unavailable".to_string())?; + if let Some(active) = metadata.active.as_ref() + && let Some(segment_id) = active.segment_id + { + let session_store = WorkerSessionStore::new(worker_aggregate_dir.join("session")) + .map_err(|error| format!("failed to read Worker Session store: {error}"))?; + if !session_store + .exists(active.session_id, segment_id) + .map_err(|error| format!("failed to inspect Worker Session segment: {error}"))? + { + return Err("Persisted Worker Session segment is unavailable".to_string()); + } + session_store + .read_all(active.session_id, segment_id) + .map_err(|error| format!("failed to read Worker Session segment: {error}"))?; + } + if request.request.workspace_api.is_some() + && metadata + .active + .as_ref() + .is_some_and(|active| active.segment_id.is_none()) + && request.config_bundle.is_none() + { + return Err( + "Pending Workspace Worker restore requires current profile launch authority" + .to_string(), + ); + } + Ok(()) + } + async fn restore_controller( &self, request: WorkerExecutionRestoreRequest, @@ -1310,6 +1369,7 @@ pub struct WorkerRuntimeExecutionBackend { working_directory_materializer: Option>, runtime: Mutex>, workers: Mutex>, + restore_lock: Mutex<()>, spawn_restore_timeout: Duration, } @@ -1338,6 +1398,7 @@ where working_directory_materializer: None, runtime: Mutex::new(Some(runtime)), workers: Mutex::new(HashMap::new()), + restore_lock: Mutex::new(()), spawn_restore_timeout: SPAWN_RESTORE_TASK_TIMEOUT, }) } @@ -1569,6 +1630,39 @@ where .unwrap_or_else(|message| WorkerExecutionResult::errored(operation, message)) } + fn retain_uncertain_unconnected_controller( + &self, + worker_ref: &crate::identity::WorkerRef, + handle: WorkerHandle, + shutdown: Arc>>, + tasks: RuntimeExecutionTaskScope, + worker_state: Arc>, + workspace_client: Option>, + ) -> Result { + let mut workers = self + .workers + .lock() + .map_err(|_| "worker adapter registry lock is poisoned".to_string())?; + if workers.contains_key(worker_ref) { + return Err("Worker already has a retained execution handle".to_string()); + } + workers.insert( + worker_ref.clone(), + RuntimeWorkerExecution { + handle, + shutdown, + shutdown_requested: Arc::new(AtomicBool::new(false)), + tasks, + worker_state, + workspace_client, + }, + ); + Ok(WorkerExecutionHandle::new( + worker_ref.clone(), + self.backend_id(), + )) + } + fn cleanup_unconnected_controller( &self, handle: &WorkerHandle, @@ -1618,11 +1712,19 @@ where working_directory: Option, workspace_client: Option>, ) -> WorkerExecutionSpawnResult { - let worker_state = Arc::new(RwLock::new(handle.shared_state.snapshot())); + #[cfg(feature = "ws-server")] + let streams = subscribe_worker_protocol_session(&handle); + #[cfg(feature = "ws-server")] + let worker_state_snapshot = match &streams.snapshot_event { + Event::Snapshot { state, .. } => state.clone(), + _ => unreachable!("Worker protocol subscription snapshot must be a snapshot event"), + }; + #[cfg(not(feature = "ws-server"))] + let worker_state_snapshot = handle.shared_state.snapshot(); + let worker_state = Arc::new(RwLock::new(worker_state_snapshot)); let tasks = RuntimeExecutionTaskScope::new(controller_task); #[cfg(feature = "ws-server")] { - let streams = subscribe_worker_protocol_session(&handle); let mut events = streams.events; let mut entry_events = streams.log_entries; let bridge_worker_state = worker_state.clone(); @@ -1666,15 +1768,50 @@ where }) { Ok(task) => task, Err(message) => { - let cleanup = self - .cleanup_unconnected_controller(&handle, &shutdown, &tasks, &worker_state) - .err() - .map(|cleanup| format!("; controller cleanup failed: {cleanup}")) - .unwrap_or_default(); - return WorkerExecutionSpawnResult::Errored(WorkerExecutionResult::errored( + let cleanup = self.cleanup_unconnected_controller( + &handle, + &shutdown, + &tasks, + &worker_state, + ); + let result = WorkerExecutionResult::errored( operation, - format!("{message}{cleanup}"), - )); + match &cleanup { + Ok(()) => message, + Err(cleanup) => { + format!("{message}; controller cleanup failed: {cleanup}") + } + }, + ); + return if operation == WorkerExecutionOperation::Restore { + match cleanup { + Ok(()) => WorkerExecutionSpawnResult::RolledBack(result), + Err(_) => { + let retained_state = + worker_state.read().ok().map(|state| state.clone()); + let retained_working_directory = + working_directory.as_ref().map(|binding| binding.status()); + let retained_handle = self + .retain_uncertain_unconnected_controller( + &worker_ref, + handle, + shutdown, + tasks, + worker_state, + workspace_client, + ) + .ok(); + WorkerExecutionSpawnResult::ReconciliationRequired { + result, + handle: retained_handle, + worker_state: retained_state, + working_directory: retained_working_directory, + } + } + } + } else { + WorkerExecutionSpawnResult::Errored(result) + }; } }; tasks.push("protocol bridge", bridge_task, true); @@ -1684,31 +1821,40 @@ where let _ = bridge_context; } - let mut workers = match self.workers.lock() { - Ok(workers) => workers, - Err(_) => { - let cleanup = self - .cleanup_unconnected_controller(&handle, &shutdown, &tasks, &worker_state) - .err() - .map(|cleanup| format!("; controller cleanup failed: {cleanup}")) - .unwrap_or_default(); - return WorkerExecutionSpawnResult::Errored(WorkerExecutionResult::errored( - operation, - format!("worker adapter registry lock is poisoned{cleanup}"), - )); - } - }; + let mut workers = self + .workers + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); if workers.contains_key(&worker_ref) { + let existing_worker_state = workers + .get(&worker_ref) + .and_then(|worker| worker.worker_state.read().ok().map(|state| state.clone())); + let existing_handle = WorkerExecutionHandle::new(worker_ref.clone(), self.backend_id()); drop(workers); - let cleanup = self - .cleanup_unconnected_controller(&handle, &shutdown, &tasks, &worker_state) - .err() - .map(|cleanup| format!("; controller cleanup failed: {cleanup}")) - .unwrap_or_default(); - return WorkerExecutionSpawnResult::Rejected(WorkerExecutionResult::busy( + let cleanup = + self.cleanup_unconnected_controller(&handle, &shutdown, &tasks, &worker_state); + let result = WorkerExecutionResult::busy( operation, - format!("Worker is already connected to execution backend{cleanup}"), - )); + match &cleanup { + Ok(()) => "Worker is already connected to execution backend".to_string(), + Err(cleanup) => format!( + "Worker is already connected to execution backend; controller cleanup failed: {cleanup}" + ), + }, + ); + return if operation == WorkerExecutionOperation::Restore { + match cleanup { + Ok(()) => WorkerExecutionSpawnResult::RolledBack(result), + Err(_) => WorkerExecutionSpawnResult::ReconciliationRequired { + result, + handle: Some(existing_handle), + worker_state: existing_worker_state, + working_directory: None, + }, + } + } else { + WorkerExecutionSpawnResult::Rejected(result) + }; } let connected_worker_state = worker_state .read() @@ -1884,11 +2030,15 @@ where } fn spawn_worker(&self, request: WorkerExecutionSpawnRequest) -> WorkerExecutionSpawnResult { + let _start_guard = self + .restore_lock + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); if self .workers .lock() - .map(|workers| workers.contains_key(&request.worker_ref)) - .unwrap_or(false) + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .contains_key(&request.worker_ref) { return WorkerExecutionSpawnResult::Rejected(WorkerExecutionResult::busy( WorkerExecutionOperation::Spawn, @@ -1995,10 +2145,111 @@ where ) } + fn preflight_restore( + &self, + request: &WorkerExecutionRestoreRequest, + ) -> Result<(), WorkerExecutionResult> { + let workers = self + .workers + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + if workers.contains_key(&request.worker_ref) { + return Err(WorkerExecutionResult::busy( + WorkerExecutionOperation::Restore, + "Worker is already connected to execution backend", + )); + } + drop(workers); + + if let Some(previous) = request.previous_working_directory.as_ref() { + let materializer = self + .working_directory_materializer + .as_ref() + .ok_or_else(|| { + WorkerExecutionResult::rejected( + WorkerExecutionOperation::Restore, + "Persisted Worker Workdir binding cannot be restored by this Runtime", + ) + })?; + let current = materializer + .working_directory_status(&previous.summary.working_directory_id) + .map_err(|message| { + WorkerExecutionResult::rejected( + WorkerExecutionOperation::Restore, + format!("Persisted Worker Workdir is unavailable: {message}"), + ) + })?; + if current.summary.status != WorkingDirectoryStatusKind::Active { + return Err(WorkerExecutionResult::rejected( + WorkerExecutionOperation::Restore, + "Persisted Worker Workdir is not active", + )); + } + } + if request.previous_working_directory.is_none() + && request.request.working_directory_request.is_some() + { + return Err(WorkerExecutionResult::rejected( + WorkerExecutionOperation::Restore, + "Persisted Worker Workdir allocation is unavailable", + )); + } + if request.previous_working_directory.is_none() + && request.request.working_directory.is_some() + && self.working_directory_materializer.is_none() + { + return Err(WorkerExecutionResult::rejected( + WorkerExecutionOperation::Restore, + "Persisted Worker Workdir claim cannot be restored by this Runtime", + )); + } + if let Some(workspace_api) = request.request.workspace_api.as_ref() { + let Some(scope) = request.workspace_scope.as_ref() else { + return Err(WorkerExecutionResult::rejected( + WorkerExecutionOperation::Restore, + "Persisted Workspace restore authorization is unavailable", + )); + }; + if scope.workspace_id != workspace_api.workspace_id { + return Err(WorkerExecutionResult::rejected( + WorkerExecutionOperation::Restore, + "Persisted Workspace restore authorization does not match Worker authority", + )); + } + } + + let factory = Arc::clone(&self.factory); + let request = request.clone(); + self.run_on_adapter_runtime(async move { factory.preflight_restore(&request).await }) + .map_err(|message| { + WorkerExecutionResult::rejected(WorkerExecutionOperation::Restore, message) + })?; + Ok(()) + } + fn restore_worker( &self, mut request: WorkerExecutionRestoreRequest, ) -> WorkerExecutionSpawnResult { + // Serialize backend restore starts so duplicate detection is completed + // before any controller or bridge side effect. Runtime also serializes + // per Worker; this guard protects direct/concurrent backend callers. + let _restore_guard = self + .restore_lock + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + let workers = self + .workers + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + if workers.contains_key(&request.worker_ref) { + return WorkerExecutionSpawnResult::Rejected(WorkerExecutionResult::busy( + WorkerExecutionOperation::Restore, + "Worker is already connected to execution backend", + )); + } + drop(workers); + let working_directory = match request.previous_working_directory.clone() { Some(status) => { let Some(materializer) = self.working_directory_materializer.as_ref() else { @@ -2076,10 +2327,15 @@ where let controller = match restore_result { Ok(controller) => controller, Err(message) => { - return WorkerExecutionSpawnResult::Errored(WorkerExecutionResult::errored( - WorkerExecutionOperation::Restore, - message, - )); + return WorkerExecutionSpawnResult::ReconciliationRequired { + result: WorkerExecutionResult::errored( + WorkerExecutionOperation::Restore, + message, + ), + handle: None, + worker_state: None, + working_directory: None, + }; } }; diff --git a/crates/workspace-api/src/lib.rs b/crates/workspace-api/src/lib.rs index c05750ab3..c3452b24f 100644 --- a/crates/workspace-api/src/lib.rs +++ b/crates/workspace-api/src/lib.rs @@ -2327,15 +2327,36 @@ pub enum WorkerOperationState { Rejected, } +/// Cross-layer outcome of an explicit Worker restore operation. +/// +/// `Rejected` is reserved for read-only preflight failures. Once live restore +/// work starts, failure is either a confirmed `RolledBack` operation or a +/// `ReconciliationRequired` result whose commit state must be reread/retried. +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +#[serde(rename_all = "snake_case")] +pub enum WorkerRestoreState { + Accepted, + Rejected, + RolledBack, + ReconciliationRequired, +} + #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +#[serde(deny_unknown_fields)] pub struct WorkerRestoreResult { - pub state: WorkerOperationState, + pub state: WorkerRestoreState, + #[serde(default, skip_serializing_if = "Option::is_none")] + #[cfg_attr(feature = "typescript", ts(optional = nullable))] pub worker: Option, #[serde(default)] pub diagnostics: Vec, } #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +#[serde(deny_unknown_fields)] pub struct WorkerRestoreResponse { pub workspace_id: String, pub runtime_id: String, @@ -3482,6 +3503,42 @@ mod workdir_typescript_tests { mod tests { use super::*; + #[test] + fn worker_restore_state_round_trips_and_rejects_unknown_variants() { + for (state, wire) in [ + (WorkerRestoreState::Accepted, "accepted"), + (WorkerRestoreState::Rejected, "rejected"), + (WorkerRestoreState::RolledBack, "rolled_back"), + ( + WorkerRestoreState::ReconciliationRequired, + "reconciliation_required", + ), + ] { + assert_eq!( + serde_json::to_value(state).unwrap(), + serde_json::json!(wire) + ); + assert_eq!( + serde_json::from_value::(serde_json::json!(wire)).unwrap(), + state + ); + } + assert!( + serde_json::from_value::(serde_json::json!("unknown")).is_err() + ); + } + + #[test] + fn worker_restore_result_rejects_unknown_fields() { + let result = serde_json::from_value::(serde_json::json!({ + "state": "rejected", + "worker": null, + "diagnostics": [], + "unexpected": true + })); + assert!(result.is_err()); + } + #[test] fn historical_http_repository_source_kind_decodes_as_invalid_evidence() { let source: RepositorySource = serde_json::from_value(serde_json::json!({ diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index 6cd702387..1e71e2ab2 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -52,9 +52,10 @@ use worker_runtime::http_server::{ RuntimeHttpWorkerCompletionsResponse, RuntimeHttpWorkerDeleteResponse, RuntimeHttpWorkerInputResponse, RuntimeHttpWorkerLifecycleRequest, RuntimeHttpWorkerLifecycleResponse, RuntimeHttpWorkerResponse, - RuntimeHttpWorkerWorkspaceApiRequest, RuntimeHttpWorkersResponse, - RuntimeHttpWorkingDirectoriesResponse, RuntimeHttpWorkingDirectoryResponse, - RuntimeHttpWorkspacePromptProjectionRequest, RuntimeHttpWorkspacePromptProjectionResponse, + RuntimeHttpWorkerRestoreResponse, RuntimeHttpWorkerWorkspaceApiRequest, + RuntimeHttpWorkersResponse, RuntimeHttpWorkingDirectoriesResponse, + RuntimeHttpWorkingDirectoryResponse, RuntimeHttpWorkspacePromptProjectionRequest, + RuntimeHttpWorkspacePromptProjectionResponse, }; use worker_runtime::identity::{ RuntimeWorkerRef, WorkerId as EmbeddedWorkerId, WorkerRef as EmbeddedWorkerRef, @@ -378,8 +379,9 @@ pub(crate) fn workspace_worker_summary( } #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] pub struct WorkerRestoreResult { - pub state: WorkerOperationState, + pub state: workspace_api::WorkerRestoreState, #[serde(skip_serializing_if = "Option::is_none")] pub worker: Option, pub diagnostics: Vec, @@ -882,7 +884,7 @@ pub trait WorkspaceWorkerRuntime: Send + Sync { fn restore_worker(&self, worker_id: &str) -> WorkerRestoreResult { WorkerRestoreResult { - state: WorkerOperationState::Unsupported, + state: workspace_api::WorkerRestoreState::Rejected, worker: None, diagnostics: vec![diagnostic( "worker_restore_unsupported", @@ -2336,7 +2338,7 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime { fn restore_worker(&self, worker_id: &str) -> WorkerRestoreResult { let Some(worker_ref) = self.worker_ref(worker_id) else { return WorkerRestoreResult { - state: WorkerOperationState::Rejected, + state: workspace_api::WorkerRestoreState::Rejected, worker: None, diagnostics: vec![diagnostic( "embedded_worker_id_invalid", @@ -2345,14 +2347,22 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime { )], }; }; - match self.runtime.restore_worker(&worker_ref) { - Ok(detail) => WorkerRestoreResult { - state: WorkerOperationState::Accepted, - worker: Some(self.map_worker_detail(detail)), - diagnostics: Vec::new(), - }, + match self.runtime.restore_worker_operation(&worker_ref) { + Ok(result) => { + let diagnostics = match (result.reason_code, result.message) { + (Some(code), Some(message)) => { + vec![diagnostic(code, DiagnosticSeverity::Warning, message)] + } + _ => Vec::new(), + }; + WorkerRestoreResult { + state: result.state, + worker: result.worker.map(|detail| self.map_worker_detail(detail)), + diagnostics, + } + } Err(err) => WorkerRestoreResult { - state: WorkerOperationState::Rejected, + state: workspace_api::WorkerRestoreState::ReconciliationRequired, worker: None, diagnostics: vec![embedded_runtime_diagnostic(&err)], }, @@ -4164,17 +4174,27 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { } fn restore_worker(&self, worker_id: &str) -> WorkerRestoreResult { - match self.post_json::<_, RuntimeHttpWorkerResponse>( + match self.post_json::<_, RuntimeHttpWorkerRestoreResponse>( &format!("/v1/workers/{worker_id}/restore"), &serde_json::json!({}), ) { - Ok(response) => WorkerRestoreResult { - state: WorkerOperationState::Accepted, - worker: Some(self.map_worker_detail(response.worker)), - diagnostics: Vec::new(), - }, + Ok(response) => { + let diagnostics = match (response.reason_code, response.message) { + (Some(code), Some(message)) => { + vec![diagnostic(code, DiagnosticSeverity::Warning, message)] + } + _ => Vec::new(), + }; + WorkerRestoreResult { + state: response.state, + worker: response.worker.map(|worker| self.map_worker_detail(worker)), + diagnostics, + } + } Err(diagnostic) => WorkerRestoreResult { - state: WorkerOperationState::Rejected, + // A transport/protocol failure cannot prove that the Runtime + // rejected before side effects. Preserve uncertainty. + state: workspace_api::WorkerRestoreState::ReconciliationRequired, worker: None, diagnostics: vec![diagnostic], }, @@ -5428,6 +5448,38 @@ mod tests { use std::sync::{Arc, Mutex}; use std::thread; + #[test] + fn remote_restore_transport_failure_requires_reconciliation() { + let provider = RemoteWorkerRuntime::new( + RemoteRuntimeConfig { + runtime_id: "runtime-remote".to_string(), + workspace_id: Some("workspace-a".to_string()), + display_name: "Remote Runtime".to_string(), + base_url: "http://127.0.0.1:9".to_string(), + bearer_token: None, + workspace_authorization: None, + strict_public_egress: false, + cached_worker_creation_available: true, + cached_os: "linux".to_string(), + cached_arch: "x86_64".to_string(), + cached_status: "ready".to_string(), + timeout: Duration::from_millis(100), + }, + "workspace-a".to_string(), + "http://127.0.0.1:3000".to_string(), + ) + .unwrap(); + + let result = provider.restore_worker("worker-transport-uncertain"); + + assert_eq!( + result.state, + workspace_api::WorkerRestoreState::ReconciliationRequired + ); + assert!(result.worker.is_none()); + assert_eq!(result.diagnostics.len(), 1); + } + #[test] fn embedded_delete_persistence_failure_diagnostic_is_bounded_and_path_free() { let diagnostic = diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index c3f56388d..721977812 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -2814,7 +2814,7 @@ impl WorkspaceApi { .map_err(|error| error.into_error())?; if binding.state != WorkerOperationState::Accepted { return Ok(WorkerRestoreResult { - state: binding.state, + state: workspace_api::WorkerRestoreState::Rejected, worker: binding.worker, diagnostics: binding.diagnostics, }); @@ -10422,7 +10422,7 @@ async fn scoped_start_workspace_orchestrator( .runtime .restore_worker(&existing.worker) .map_err(|error| error.into_error())?; - if restored.state != WorkerOperationState::Accepted { + if restored.state != workspace_api::WorkerRestoreState::Accepted { return Err(ApiError::with_diagnostics( Error::RuntimeOperationFailed { runtime_id: existing.worker.runtime_id.clone(), @@ -12501,7 +12501,7 @@ async fn scoped_restore_runtime_worker( runtime_id: runtime_id.clone(), worker_id: worker_id.clone(), result: workspace_api::WorkerRestoreResult { - state: workspace_api::WorkerOperationState::Accepted, + state: workspace_api::WorkerRestoreState::Accepted, worker: Some(worker), diagnostics: Vec::new(), }, @@ -15213,7 +15213,7 @@ async fn restore_runtime_worker( runtime_id: runtime_id.clone(), worker_id: worker_id.clone(), result: workspace_api::WorkerRestoreResult { - state: result.state.into(), + state: result.state, worker: projected_worker, diagnostics: result.diagnostics.into_iter().map(Into::into).collect(), }, @@ -20722,7 +20722,7 @@ mod tests { } #[tokio::test] - async fn production_profile_backend_rejects_unrecoverable_pending_orchestrator_restore() { + async fn production_profile_backend_preflight_rejects_pending_orchestrator_restore() { let workspace = tempfile::tempdir().unwrap(); init_clean_git_workspace(workspace.path()); let config = test_server_config(workspace.path()); @@ -20772,11 +20772,11 @@ mod tests { }), ) .await - .expect_err("pending Workspace Orchestrator restore without durable Prompt must fail"); + .expect_err( + "pending Workspace Orchestrator restore without durable Prompt must be rejected", + ); assert!( - format!("{error:?}").contains( - "pending Workspace Worker restore requires operation-owned launch material" - ), + format!("{error:?}").contains("worker_restore_preflight_rejected"), "unexpected restore error: {error:?}" ); } @@ -25277,7 +25277,7 @@ mod tests { assert_eq!(retried_restore.worker_id, first_worker.worker.worker_id); assert_eq!( retried_restore.result.state, - workspace_api::WorkerOperationState::Accepted + workspace_api::WorkerRestoreState::Accepted ); let restored_assignment = api .store diff --git a/docs/report/2026-08-19-coder-toolchain-missing-rustfmt.md b/docs/report/2026-08-19-coder-toolchain-missing-rustfmt.md new file mode 100644 index 000000000..907ccd749 --- /dev/null +++ b/docs/report/2026-08-19-coder-toolchain-missing-rustfmt.md @@ -0,0 +1,17 @@ +# Ticket validation image lacks `rustfmt` + +While implementing T-613, `cargo fmt --all` failed before reading project sources because the active Rust toolchain does not provide the `rustfmt` component. The standalone `rustfmt` binary is also absent. + +Observed commands: + +```text +$ cargo fmt --all +error: 'cargo-fmt' is not installed for the toolchain '1.90.0-x86_64-unknown-linux-gnu'. + +$ rustfmt --version +bash: rustfmt: command not found +``` + +This prevents Workers from executing the repository-required formatting check or automatically normalizing Rust changes. The Coder can still run compile/tests and `git diff --check`, but formatting evidence remains weaker and review may surface avoidable formatting differences. + +Suggested improvement: include the `rustfmt` component in the standard coding Workdir/toolchain image and verify `cargo fmt --all -- --check` during image qualification. diff --git a/web/workspace/src/lib/workspace/sidebar/types.ts b/web/workspace/src/lib/workspace/sidebar/types.ts index 856076fd2..a6642c2c3 100644 --- a/web/workspace/src/lib/workspace/sidebar/types.ts +++ b/web/workspace/src/lib/workspace/sidebar/types.ts @@ -112,6 +112,19 @@ export type Worker = { export type WorkerOperationState = "accepted" | "unsupported" | "rejected"; +/** Typed result of restore across Runtime, Workspace, Web, and TUI clients. */ +export type WorkerRestoreState = + | "accepted" + | "rejected" + | "rolled_back" + | "reconciliation_required"; + +export type WorkerRestoreResult = { + state: WorkerRestoreState; + worker?: WorkerSummary | null; + diagnostics: Diagnostic[]; +}; + export type WorkerLaunchRuntimeOption = SharedWorkerLaunchRuntimeOption; export type WorkerLaunchProfileCandidate = SharedWorkerLaunchProfileCandidate; export type WorkingDirectoryRepositoryOption =