Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

runtime: filter stopped workers

Browse Source
This commit is contained in:
Keisuke Hirata 2026-07-29 02:54:23 +09:00
parent 98c1599d1a
commit acc7281414
No known key found for this signature in database
6 changed files with 387 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

92
crates/client/src/backend_runtime.rs
View File

@ -168,6 +168,23 @@ pub struct BackendWorkerSummary {
pub diagnostics: Vec<BackendDiagnostic>,
}
#[derive(Debug, Clone, Deserialize, PartialEq, Eq)]
pub struct BackendWorkerRestoreResult {
pub state: String,
#[serde(default)]
pub worker: Option<BackendWorkerSummary>,
#[serde(default)]
pub diagnostics: Vec<BackendDiagnostic>,
}
#[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<BackendRuntimeListResponse<BackendWorkerSummary>, 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::<BackendRuntimeListResponse<BackendWorkerSummary>>()
.await?)
}
pub async fn restore_backend_worker(
target: &BackendRuntimeTarget,
) -> Result<BackendWorkerRestoreResponse, BackendRuntimeClientError> {
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::<BackendWorkerRestoreResponse>()
.await?)
}
impl BackendRuntimeClient {
pub async fn connect(target: BackendRuntimeTarget) -> Result<Self, BackendRuntimeClientError> {
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"
);
}
}

6
crates/client/src/lib.rs
View File

@ -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::{

22
crates/worker-runtime/src/http_server.rs
View File

@ -234,6 +234,17 @@ struct RuntimeHttpConfigBundleAvailabilityQuery {
digest: String,
}
#[derive(Clone, Debug, Default, Deserialize)]
struct RuntimeHttpWorkersQuery {
status: Option<RuntimeHttpWorkerStatusFilter>,
}
#[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<RuntimeHttpState>,
query: Result<Query<RuntimeHttpWorkersQuery>, QueryRejection>,
) -> RestResult<RuntimeHttpWorkersResponse> {
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 }))
}

33
crates/worker-runtime/src/runtime.rs
View File

@ -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<Vec<WorkerSummary>, 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<WorkerDetail, RuntimeError> {
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();

134
crates/workspace-server/src/hosts.rs
View File

@ -253,6 +253,14 @@ pub struct WorkerSummary {
pub diagnostics: Vec<RuntimeDiagnostic>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct WorkerRestoreResult {
pub state: WorkerOperationState,
#[serde(skip_serializing_if = "Option::is_none")]
pub worker: Option<WorkerSummary>,
pub diagnostics: Vec<RuntimeDiagnostic>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct RuntimeList<T> {
pub items: Vec<T>,
@ -576,8 +584,24 @@ pub trait WorkspaceWorkerRuntime: Send + Sync {
fn list_workers(&self, limit: usize) -> RuntimeList<WorkerSummary>;
fn list_stopped_workers(&self, _limit: usize) -> RuntimeList<WorkerSummary> {
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<RuntimeList<WorkerSummary>, 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<WorkerRestoreResult, RuntimeRegistryError> {
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<WorkerSummary> {
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<WorkerSummary> {
if limit == 0 {
return RuntimeList::new(Vec::new(), Vec::new());
}
match self.get_json::<RuntimeHttpWorkersResponse>("/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::<RuntimeHttpWorkerResponse>(&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,

119
crates/workspace-server/src/server.rs
View File

@ -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<T> {
pub diagnostics: Vec<RuntimeDiagnostic>,
}
#[derive(Debug, Default, Deserialize)]
struct RuntimeWorkersQuery {
status: Option<RuntimeWorkersStatusFilter>,
}
#[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<WorkspaceApi>,
AxumPath(path): AxumPath<ScopedRuntimePath>,
Query(query): Query<RuntimeWorkersQuery>,
) -> ApiResult<Json<RuntimeListResponse<WorkerSummary>>> {
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<WorkspaceApi>,
AxumPath(path): AxumPath<ScopedRuntimeWorkerPath>,
) -> ApiResult<Json<WorkerRestoreResponse>> {
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<WorkspaceApi>,
AxumPath(path): AxumPath<ScopedRuntimeWorkerPath>,
@ -4322,6 +4358,39 @@ async fn get_runtime_worker(
)))
}
async fn restore_runtime_worker(
State(api): State<WorkspaceApi>,
AxumPath((runtime_id, worker_id)): AxumPath<(String, String)>,
) -> ApiResult<Json<WorkerRestoreResponse>> {
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<WorkspaceApi>,
AxumPath(runtime_id): AxumPath<String>,
Query(query): Query<RuntimeWorkersQuery>,
) -> ApiResult<Json<RuntimeListResponse<WorkerSummary>>> {
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",