diff --git a/crates/client/src/backend_runtime.rs b/crates/client/src/backend_runtime.rs index 40eb2259..bdd960d3 100644 --- a/crates/client/src/backend_runtime.rs +++ b/crates/client/src/backend_runtime.rs @@ -168,6 +168,23 @@ pub struct BackendWorkerSummary { pub diagnostics: Vec, } +#[derive(Debug, Clone, Deserialize, PartialEq, Eq)] +pub struct BackendWorkerRestoreResult { + pub state: String, + #[serde(default)] + pub worker: Option, + #[serde(default)] + pub diagnostics: Vec, +} + +#[derive(Debug, Clone, Deserialize, PartialEq, Eq)] +pub struct BackendWorkerRestoreResponse { + pub workspace_id: String, + pub runtime_id: String, + pub worker_id: String, + pub result: BackendWorkerRestoreResult, +} + #[derive(Debug)] pub struct BackendRuntimeClient { target: BackendRuntimeTarget, @@ -266,6 +283,44 @@ pub async fn list_backend_workers( }) } +pub async fn list_backend_stopped_workers( + target: &BackendRuntimeListTarget, +) -> Result, BackendRuntimeClientError> { + validate_list_target(target)?; + let Some(runtime_id) = target.runtime_id.as_deref() else { + return Err(BackendRuntimeClientError::InvalidTarget( + "stopped worker listing requires a runtime id".to_string(), + )); + }; + let http = reqwest::Client::new(); + let path = backend_runtime_workers_path(target.workspace_id.as_deref(), runtime_id); + let url = join_base_and_path(&target.base_url, &format!("{path}?status=stopped")); + Ok(http + .get(url) + .send() + .await? + .error_for_status()? + .json::>() + .await?) +} + +pub async fn restore_backend_worker( + target: &BackendRuntimeTarget, +) -> Result { + validate_target(target)?; + let http = reqwest::Client::new(); + let path = backend_runtime_worker_restore_path(None, &target.runtime_id, &target.worker_id); + let url = join_base_and_path(&target.base_url, &path); + Ok(http + .post(url) + .json(&serde_json::json!({})) + .send() + .await? + .error_for_status()? + .json::() + .await?) +} + impl BackendRuntimeClient { pub async fn connect(target: BackendRuntimeTarget) -> Result { validate_target(&target)?; @@ -479,6 +534,26 @@ fn backend_runtime_workers_path(workspace_id: Option<&str>, runtime_id: &str) -> } } +fn backend_runtime_worker_restore_path( + workspace_id: Option<&str>, + runtime_id: &str, + worker_id: &str, +) -> String { + match workspace_id { + Some(workspace_id) => format!( + "/api/w/{}/runtimes/{}/workers/{}/restore", + path_segment_encode(workspace_id), + path_segment_encode(runtime_id), + path_segment_encode(worker_id) + ), + None => format!( + "/api/runtimes/{}/workers/{}/restore", + path_segment_encode(runtime_id), + path_segment_encode(worker_id) + ), + } +} + fn protocol_ws_url(target: &BackendRuntimeTarget) -> String { let path = format!( "/api/runtimes/{}/workers/{}/protocol/ws", @@ -542,4 +617,21 @@ mod tests { "ws://127.0.0.1:8787/api/runtimes/runtime%2Fone/workers/worker%20one/protocol/ws" ); } + + #[test] + fn workers_path_can_be_workspace_scoped_for_status_queries() { + let path = backend_runtime_workers_path(Some("team main"), "runtime/one"); + assert_eq!( + format!("{path}?status=stopped"), + "/api/w/team%20main/runtimes/runtime%2Fone/workers?status=stopped" + ); + } + + #[test] + fn restore_worker_path_uses_backend_runtime_worker_identity() { + assert_eq!( + backend_runtime_worker_restore_path(None, "runtime/one", "worker one"), + "/api/runtimes/runtime%2Fone/workers/worker%20one/restore" + ); + } } diff --git a/crates/client/src/lib.rs b/crates/client/src/lib.rs index 84ffa477..67fdeb92 100644 --- a/crates/client/src/lib.rs +++ b/crates/client/src/lib.rs @@ -23,8 +23,10 @@ pub use backend_auth::{ pub use backend_runtime::{ BackendRuntimeClient, BackendRuntimeClientError, BackendRuntimeListResponse, BackendRuntimeListTarget, BackendRuntimeSummary, BackendRuntimeTarget, - BackendWorkerCapabilitySummary, BackendWorkerImplementationSummary, BackendWorkerSummary, - BackendWorkerWorkspaceSummary, BackendWorkingDirectorySummary, list_backend_workers, + BackendWorkerCapabilitySummary, BackendWorkerImplementationSummary, + BackendWorkerRestoreResponse, BackendWorkerRestoreResult, BackendWorkerSummary, + BackendWorkerWorkspaceSummary, BackendWorkingDirectorySummary, list_backend_stopped_workers, + list_backend_workers, restore_backend_worker, }; pub use runtime_command::WorkerRuntimeCommand; pub use target::{ diff --git a/crates/worker-runtime/src/http_server.rs b/crates/worker-runtime/src/http_server.rs index 2c8a0797..a8d8a3d7 100644 --- a/crates/worker-runtime/src/http_server.rs +++ b/crates/worker-runtime/src/http_server.rs @@ -234,6 +234,17 @@ struct RuntimeHttpConfigBundleAvailabilityQuery { digest: String, } +#[derive(Clone, Debug, Default, Deserialize)] +struct RuntimeHttpWorkersQuery { + status: Option, +} + +#[derive(Clone, Copy, Debug, Deserialize)] +#[serde(rename_all = "snake_case")] +enum RuntimeHttpWorkerStatusFilter { + Stopped, +} + /// `GET /v1/workers` response. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct RuntimeHttpWorkersResponse { @@ -372,11 +383,14 @@ async fn check_config_bundle( async fn list_workers( State(state): State, + query: Result, QueryRejection>, ) -> RestResult { - let workers = state - .runtime - .list_workers() - .map_err(RuntimeHttpRestError::runtime)?; + let Query(query) = query.map_err(RuntimeHttpRestError::query_rejection)?; + let workers = match query.status { + Some(RuntimeHttpWorkerStatusFilter::Stopped) => state.runtime.list_stopped_workers(), + None => state.runtime.list_workers(), + } + .map_err(RuntimeHttpRestError::runtime)?; Ok(Json(RuntimeHttpWorkersResponse { workers })) } diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index ca3f4642..4d19f342 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -7,9 +7,7 @@ use crate::config_bundle::{ ConfigBundle, ConfigBundleAvailability, ConfigBundleSummary, validate_config_bundle, validate_config_bundle_ref, }; -#[cfg(feature = "fs-store")] -use crate::diagnostics::DiagnosticSeverity; -use crate::diagnostics::RuntimeDiagnostic; +use crate::diagnostics::{DiagnosticSeverity, RuntimeDiagnostic}; use crate::error::RuntimeError; use crate::execution::WorkerExecutionRestoreRequest; use crate::execution::{ @@ -437,6 +435,17 @@ impl Runtime { Ok(state.workers.values().map(WorkerRecord::summary).collect()) } + /// List stopped Workers known to this Runtime. + pub fn list_stopped_workers(&self) -> Result, RuntimeError> { + let state = self.lock()?; + Ok(state + .workers + .values() + .filter(|worker| worker.status == WorkerStatus::Stopped) + .map(WorkerRecord::summary) + .collect()) + } + /// Fetch Worker detail. The supplied [`WorkerRef`] must match this Runtime. pub fn worker_detail(&self, worker_ref: &WorkerRef) -> Result { let state = self.lock()?; @@ -2049,6 +2058,24 @@ mod tests { assert_eq!(fetched.profile, detail.profile); } + #[test] + fn stopped_worker_list_excludes_alive_and_cancelled_workers() { + let runtime = runtime_with_backend(); + let alive = runtime.create_worker(task_request("alive")).unwrap(); + let stopped = runtime.create_worker(task_request("stopped")).unwrap(); + let cancelled = runtime.create_worker(task_request("cancelled")).unwrap(); + + runtime.stop_worker(&stopped.worker_ref, None).unwrap(); + runtime.cancel_worker(&cancelled.worker_ref, None).unwrap(); + + let candidates = runtime.list_stopped_workers().unwrap(); + assert_eq!(candidates.len(), 1); + assert_eq!(candidates[0].worker_ref, stopped.worker_ref); + assert_eq!(candidates[0].status, WorkerStatus::Stopped); + assert_ne!(candidates[0].worker_ref, alive.worker_ref); + assert_ne!(candidates[0].worker_ref, cancelled.worker_ref); + } + #[test] fn synced_config_bundle_is_stored_checked_and_used_for_worker_creation() { let runtime = runtime_with_backend(); diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index 0a7a7eee..9bc479cc 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -253,6 +253,14 @@ pub struct WorkerSummary { pub diagnostics: Vec, } +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct WorkerRestoreResult { + pub state: WorkerOperationState, + #[serde(skip_serializing_if = "Option::is_none")] + pub worker: Option, + pub diagnostics: Vec, +} + #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub struct RuntimeList { pub items: Vec, @@ -576,8 +584,24 @@ pub trait WorkspaceWorkerRuntime: Send + Sync { fn list_workers(&self, limit: usize) -> RuntimeList; + fn list_stopped_workers(&self, _limit: usize) -> RuntimeList { + RuntimeList::new(Vec::new(), Vec::new()) + } + fn worker(&self, worker_id: &str) -> WorkerLookupResult; + fn restore_worker(&self, worker_id: &str) -> WorkerRestoreResult { + WorkerRestoreResult { + state: WorkerOperationState::Unsupported, + worker: None, + diagnostics: vec![diagnostic( + "worker_restore_unsupported", + DiagnosticSeverity::Info, + format!("runtime does not implement worker restore for `{worker_id}`"), + )], + } + } + fn create_working_directory( &self, _request: WorkingDirectoryRequest, @@ -955,6 +979,26 @@ impl RuntimeRegistry { Ok(RuntimeList::new(items, diagnostics)) } + pub fn list_stopped_workers_for_runtime( + &self, + runtime_id: &str, + limit: usize, + ) -> Result, RuntimeRegistryError> { + validate_backend_identifier("runtime_id", runtime_id)?; + let runtime = self.runtime(runtime_id)?; + let worker_list = runtime.list_stopped_workers(limit); + let mut items: Vec<_> = worker_list + .items + .into_iter() + .filter(|worker| !is_retired_companion_worker(worker)) + .take(limit) + .collect(); + items.truncate(limit); + let mut diagnostics = worker_list.diagnostics; + diagnostics.truncate(MAX_DIAGNOSTICS); + Ok(RuntimeList::new(items, diagnostics)) + } + pub fn list_workers_for_host( &self, host_id: &str, @@ -1015,6 +1059,17 @@ impl RuntimeRegistry { Ok(worker) } + pub fn restore_worker( + &self, + runtime_id: &str, + worker_id: &str, + ) -> Result { + validate_backend_identifier("runtime_id", runtime_id)?; + validate_backend_identifier("worker_id", worker_id)?; + let runtime = self.runtime(runtime_id)?; + Ok(runtime.restore_worker(worker_id)) + } + pub fn spawn_worker( &self, runtime_id: &str, @@ -1501,6 +1556,23 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime { } } + fn list_stopped_workers(&self, limit: usize) -> RuntimeList { + if limit == 0 { + return RuntimeList::new(Vec::new(), Vec::new()); + } + match self.runtime.list_stopped_workers() { + Ok(workers) => RuntimeList::new( + workers + .into_iter() + .take(limit) + .map(|worker| self.map_worker_summary(worker)) + .collect(), + Vec::new(), + ), + Err(err) => RuntimeList::new(Vec::new(), vec![embedded_runtime_diagnostic(&err)]), + } + } + fn worker(&self, worker_id: &str) -> WorkerLookupResult { let Some(worker_ref) = self.worker_ref(worker_id) else { return WorkerLookupResult { @@ -1528,6 +1600,32 @@ 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, + worker: None, + diagnostics: vec![diagnostic( + "embedded_worker_id_invalid", + DiagnosticSeverity::Warning, + "Worker id was empty and cannot be restored".to_string(), + )], + }; + }; + match self.runtime.restore_worker(&worker_ref) { + Ok(detail) => WorkerRestoreResult { + state: WorkerOperationState::Accepted, + worker: Some(self.map_worker_detail(detail)), + diagnostics: Vec::new(), + }, + Err(err) => WorkerRestoreResult { + state: WorkerOperationState::Rejected, + worker: None, + diagnostics: vec![embedded_runtime_diagnostic(&err)], + }, + } + } + fn create_working_directory( &self, _request: WorkingDirectoryRequest, @@ -2446,6 +2544,24 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { } } + fn list_stopped_workers(&self, limit: usize) -> RuntimeList { + if limit == 0 { + return RuntimeList::new(Vec::new(), Vec::new()); + } + match self.get_json::("/v1/workers?status=stopped") { + Ok(response) => RuntimeList::new( + response + .workers + .into_iter() + .take(limit) + .map(|worker| self.map_worker_summary(worker)) + .collect(), + Vec::new(), + ), + Err(diagnostic) => RuntimeList::new(Vec::new(), vec![diagnostic]), + } + } + fn worker(&self, worker_id: &str) -> WorkerLookupResult { match self.get_json::(&format!("/v1/workers/{worker_id}")) { Ok(response) => WorkerLookupResult { @@ -2463,6 +2579,24 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { } } + fn restore_worker(&self, worker_id: &str) -> WorkerRestoreResult { + match self.post_json::<_, RuntimeHttpWorkerResponse>( + &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(), + }, + Err(diagnostic) => WorkerRestoreResult { + state: WorkerOperationState::Rejected, + worker: None, + diagnostics: vec![diagnostic], + }, + } + } + fn create_working_directory( &self, request: WorkingDirectoryRequest, diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index fb85e4ec..3853452c 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -56,9 +56,9 @@ use crate::hosts::{ RuntimeRegistryError, RuntimeRegistryUnregisterResult, RuntimeSummary, WorkerCapabilitySummary, WorkerCompletionsRequest, WorkerCompletionsResult, WorkerImplementationSummary, WorkerInputKind, WorkerInputRequest, WorkerInputResult, WorkerLifecycleRequest, - WorkerLifecycleResult, WorkerOperationState, WorkerSpawnAcceptanceRequirement, - WorkerSpawnIntent, WorkerSpawnRequest, WorkerSpawnResult, WorkerSpawnWorkingDirectoryRequest, - WorkerSummary, WorkerWorkspaceSummary, + WorkerLifecycleResult, WorkerOperationState, WorkerRestoreResult, + WorkerSpawnAcceptanceRequirement, WorkerSpawnIntent, WorkerSpawnRequest, WorkerSpawnResult, + WorkerSpawnWorkingDirectoryRequest, WorkerSummary, WorkerWorkspaceSummary, }; use crate::identity::WorkspaceIdentity; use crate::memory_backend::execute_memory_backend_operation_with_authority; @@ -668,10 +668,18 @@ pub fn build_router(api: WorkspaceApi) -> Router { "/api/runtimes/{runtime_id}/workers/{worker_id}", get(get_runtime_worker), ) + .route( + "/api/runtimes/{runtime_id}/workers/{worker_id}/restore", + post(restore_runtime_worker), + ) .route( "/api/w/{workspace_id}/runtimes/{runtime_id}/workers/{worker_id}", get(scoped_get_runtime_worker), ) + .route( + "/api/w/{workspace_id}/runtimes/{runtime_id}/workers/{worker_id}/restore", + post(scoped_restore_runtime_worker), + ) .route( "/api/w/{workspace_id}/runtimes/{runtime_id}/workers/{worker_id}/pin", put(scoped_pin_runtime_worker).delete(scoped_unpin_runtime_worker), @@ -786,6 +794,25 @@ pub struct RuntimeListResponse { pub diagnostics: Vec, } +#[derive(Debug, Default, Deserialize)] +struct RuntimeWorkersQuery { + status: Option, +} + +#[derive(Debug, Clone, Copy, Deserialize)] +#[serde(rename_all = "snake_case")] +enum RuntimeWorkersStatusFilter { + Stopped, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct WorkerRestoreResponse { + pub workspace_id: String, + pub runtime_id: String, + pub worker_id: String, + pub result: WorkerRestoreResult, +} + #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] #[serde(rename_all = "snake_case")] pub enum CleanupTargetKind { @@ -2733,9 +2760,10 @@ async fn scoped_post_companion_cancel( async fn scoped_list_runtime_workers( State(api): State, AxumPath(path): AxumPath, + Query(query): Query, ) -> ApiResult>> { validate_workspace_scope(&api, &path.workspace_id)?; - list_runtime_workers(State(api), AxumPath(path.runtime_id)).await + list_runtime_workers(State(api), AxumPath(path.runtime_id), Query(query)).await } async fn scoped_create_runtime_worker( @@ -2778,6 +2806,14 @@ async fn scoped_get_runtime_worker( get_runtime_worker(State(api), AxumPath((path.runtime_id, path.worker_id))).await } +async fn scoped_restore_runtime_worker( + State(api): State, + AxumPath(path): AxumPath, +) -> ApiResult> { + validate_workspace_scope(&api, &path.workspace_id)?; + restore_runtime_worker(State(api), AxumPath((path.runtime_id, path.worker_id))).await +} + async fn scoped_pin_runtime_worker( State(api): State, AxumPath(path): AxumPath, @@ -4322,6 +4358,39 @@ async fn get_runtime_worker( ))) } +async fn restore_runtime_worker( + State(api): State, + AxumPath((runtime_id, worker_id)): AxumPath<(String, String)>, +) -> ApiResult> { + let mut result = api + .runtime + .restore_worker(&runtime_id, &worker_id) + .map_err(|err| err.into_error())?; + if let Some(worker) = result.worker.as_ref() { + let record = sync_worker_observation(&api, worker)?; + let links = api.store.list_worker_workdir_links( + &api.config.workspace_id, + record.runtime_id.as_str(), + record.runtime_worker_id, + )?; + let workdirs = api + .store + .list_workdir_registry(&api.config.workspace_id, 500)?; + result.worker = Some(merge_worker_registry_projection( + Some(worker), + &record, + links, + &workdirs, + )); + } + Ok(Json(WorkerRestoreResponse { + workspace_id: api.workspace_id().to_string(), + runtime_id, + worker_id, + result, + })) +} + #[derive(Debug, Serialize, Deserialize)] pub struct RuntimeConfigBundleSyncRequest { pub bundle: ConfigBundle, @@ -4378,17 +4447,28 @@ fn reject_no_workdir_for_non_embedded_runtime( async fn list_runtime_workers( State(api): State, AxumPath(runtime_id): AxumPath, + Query(query): Query, ) -> ApiResult>> { let limit = api.config.max_records.min(200); - let worker_list = api - .runtime - .list_workers_for_runtime(&runtime_id, limit) - .map_err(|err| err.into_error())?; + let (worker_list, source) = match query.status { + Some(RuntimeWorkersStatusFilter::Stopped) => ( + api.runtime + .list_stopped_workers_for_runtime(&runtime_id, limit) + .map_err(|err| err.into_error())?, + "runtime_registry_stopped", + ), + None => ( + api.runtime + .list_workers_for_runtime(&runtime_id, limit) + .map_err(|err| err.into_error())?, + "runtime_registry", + ), + }; Ok(Json(RuntimeListResponse { workspace_id: api.workspace_id().to_string(), limit, items: worker_list.items, - source: "runtime_registry".to_string(), + source: source.to_string(), diagnostics: worker_list.diagnostics, })) } @@ -9322,6 +9402,22 @@ mod tests { assert_eq!(worker["worker_id"], worker_id); assert_eq!(worker["runtime_id"], "embedded-worker-runtime"); + let stopped_workers = get_json( + app.clone(), + "/api/runtimes/embedded-worker-runtime/workers?status=stopped", + ) + .await; + assert!(stopped_workers["items"].as_array().unwrap().is_empty()); + + let restored = post_json( + app.clone(), + &format!("/api/runtimes/embedded-worker-runtime/workers/{worker_id}/restore"), + json!({}), + ) + .await; + assert_eq!(restored["result"]["state"], "accepted"); + assert_eq!(restored["result"]["worker"]["worker_id"], worker_id); + let accepted = post_json( app.clone(), &format!("/api/runtimes/embedded-worker-runtime/workers/{worker_id}/input"), @@ -9373,7 +9469,10 @@ mod tests { .unwrap(); assert_eq!(wrong_runtime.status(), StatusCode::NOT_FOUND); - let projected = format!("{}{}{}{}", embedded_summary, spawned, worker, accepted); + let projected = format!( + "{}{}{}{}{}{}", + embedded_summary, spawned, worker, stopped_workers, restored, accepted + ); for forbidden in [ dir.path().to_string_lossy().as_ref(), "metadata.json",