workspace: delete workers through runtime

This commit is contained in:
2026-07-12 09:06:59 +09:00
parent d8c0853d55
commit 3009bb4ad9
12 changed files with 489 additions and 112 deletions
+12
View File
@@ -192,6 +192,18 @@ impl FsRuntimeStore {
Ok(())
}
pub(crate) fn delete_worker_snapshot(&self, worker_id: &WorkerId) -> Result<(), RuntimeError> {
let worker_dir = self.worker_dir(worker_id);
if !worker_dir.exists() {
return Ok(());
}
fs::remove_dir_all(&worker_dir).map_err(|source| RuntimeError::StoreIo {
operation: "delete worker store",
path: worker_dir,
source,
})
}
pub(crate) fn append_event(&self, event: &RuntimeEvent) -> Result<(), RuntimeError> {
if let Some(worker_ref) = &event.worker_ref {
self.ensure_worker_ref(worker_ref)?;
+23 -2
View File
@@ -15,7 +15,7 @@ use crate::config_bundle::{ConfigBundle, ConfigBundleAvailability, ConfigBundleS
use crate::error::RuntimeError;
use crate::identity::{RuntimeId, WorkerId, WorkerRef};
use crate::interaction::{WorkerInput, WorkerInteractionAck};
use crate::management::{RuntimeLimits, RuntimeSummary};
use crate::management::{RuntimeLimits, RuntimeSummary, WorkerDeleteResult};
#[cfg(feature = "ws-server")]
use crate::observation::WorkerObservationCursor;
use axum::body::{Body, Bytes};
@@ -147,7 +147,10 @@ pub fn runtime_http_router(runtime: Runtime, local_token: Option<String>) -> Rou
get(get_working_directory).delete(cleanup_working_directory),
)
.route("/v1/workers", get(list_workers).post(create_worker))
.route("/v1/workers/{worker_id}", get(get_worker))
.route(
"/v1/workers/{worker_id}",
get(get_worker).delete(delete_worker),
)
.route("/v1/workers/{worker_id}/input", post(send_worker_input))
.route("/v1/workers/{worker_id}/stop", post(stop_worker))
.route("/v1/workers/{worker_id}/cancel", post(cancel_worker));
@@ -219,6 +222,12 @@ pub struct RuntimeHttpWorkerResponse {
pub worker: WorkerDetail,
}
/// Worker delete response.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct RuntimeHttpWorkerDeleteResponse {
pub worker: WorkerDeleteResult,
}
/// Worker input acknowledgement response.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct RuntimeHttpWorkerInputResponse {
@@ -461,6 +470,18 @@ async fn get_worker(
Ok(Json(RuntimeHttpWorkerResponse { worker }))
}
async fn delete_worker(
State(state): State<RuntimeHttpState>,
Path(worker_id): Path<String>,
) -> RestResult<RuntimeHttpWorkerDeleteResponse> {
let worker_ref = worker_ref_for(&state.runtime, worker_id)?;
let worker = state
.runtime
.delete_worker(&worker_ref)
.map_err(RuntimeHttpRestError::runtime)?;
Ok(Json(RuntimeHttpWorkerDeleteResponse { worker }))
}
async fn create_worker(
State(state): State<RuntimeHttpState>,
body: Result<Json<CreateWorkerRequest>, JsonRejection>,
+8 -1
View File
@@ -1,4 +1,4 @@
use crate::identity::RuntimeId;
use crate::identity::{RuntimeId, WorkerId};
use serde::{Deserialize, Serialize};
/// Runtime backend kind.
@@ -54,6 +54,13 @@ fn unknown_platform_component() -> String {
"unknown".to_string()
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct WorkerDeleteResult {
pub runtime_id: RuntimeId,
pub worker_id: WorkerId,
pub deleted: bool,
}
/// Management-plane summary for a Runtime.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct RuntimeSummary {
+1
View File
@@ -42,6 +42,7 @@ pub enum RuntimeEventKind {
WorkerInputAccepted,
WorkerStopped,
WorkerCancelled,
WorkerDeleted,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+74
View File
@@ -25,6 +25,7 @@ use crate::identity::{RuntimeId, WorkerId, WorkerRef};
use crate::interaction::{WorkerInput, WorkerInputKind, WorkerInteractionAck};
use crate::management::{
RuntimeBackendKind, RuntimeLimits, RuntimeOptions, RuntimeStatus, RuntimeSummary,
WorkerDeleteResult,
};
use crate::observation::{
EventCursor, EventSubscription, EventSubscriptionMode, RuntimeEvent, RuntimeEventBatch,
@@ -648,6 +649,46 @@ impl Runtime {
)
}
/// Delete a non-running Worker from Runtime state and persisted Worker storage.
pub fn delete_worker(
&self,
worker_ref: &WorkerRef,
) -> Result<WorkerDeleteResult, RuntimeError> {
let mut state = self.lock()?;
state.ensure_running()?;
state.ensure_worker_ref(worker_ref)?;
let worker = state.worker(worker_ref)?;
if worker.status.is_active() {
return Err(RuntimeError::InvalidRequest(format!(
"worker {} is running and must be stopped before deletion",
worker_ref.worker_id
)));
}
let removed = state.workers.remove(&worker_ref.worker_id).ok_or_else(|| {
RuntimeError::WorkerNotFound {
runtime_id: state.runtime_id.clone(),
worker_id: worker_ref.worker_id,
}
})?;
#[cfg(feature = "ws-server")]
state
.observation_events
.retain(|event| event.worker_ref != *worker_ref);
let event_id = state.push_event(
Some(worker_ref.clone()),
RuntimeEventKind::WorkerDeleted,
"worker deleted",
);
state.persist_runtime_snapshot()?;
state.delete_worker_snapshot(&worker_ref.worker_id)?;
state.persist_event_by_id(event_id)?;
Ok(WorkerDeleteResult {
runtime_id: removed.worker_ref.runtime_id,
worker_id: removed.worker_id,
deleted: true,
})
}
/// Cursor pointing to the beginning of Runtime events.
pub fn event_cursor_from_start(&self) -> Result<EventCursor, RuntimeError> {
let state = self.lock()?;
@@ -1145,6 +1186,14 @@ impl RuntimeState {
Ok(())
}
#[cfg(feature = "fs-store")]
fn delete_worker_snapshot(&self, worker_id: &WorkerId) -> Result<(), RuntimeError> {
if let Some(store) = self.fs_store() {
store.delete_worker_snapshot(worker_id)?;
}
Ok(())
}
#[cfg(feature = "fs-store")]
fn persist_event_by_id(&self, event_id: u64) -> Result<(), RuntimeError> {
if let Some(store) = self.fs_store() {
@@ -1201,6 +1250,11 @@ impl RuntimeState {
Ok(())
}
#[cfg(not(feature = "fs-store"))]
fn delete_worker_snapshot(&self, _worker_id: &WorkerId) -> Result<(), RuntimeError> {
Ok(())
}
#[cfg(not(feature = "fs-store"))]
fn persist_event_by_id(&self, _event_id: u64) -> Result<(), RuntimeError> {
Ok(())
@@ -1997,6 +2051,26 @@ mod tests {
assert_eq!(summary.cancelled_worker_count, 1);
}
#[test]
fn delete_worker_removes_stopped_worker_from_runtime() {
let runtime = runtime_with_backend();
let worker = runtime.create_worker(task_request("delete me")).unwrap();
assert!(runtime.delete_worker(&worker.worker_ref).is_err());
runtime
.stop_worker(&worker.worker_ref, Some("done".to_string()))
.unwrap();
let result = runtime.delete_worker(&worker.worker_ref).unwrap();
assert!(result.deleted);
assert_eq!(result.worker_id, worker.worker_id);
assert!(matches!(
runtime.worker_detail(&worker.worker_ref),
Err(RuntimeError::WorkerNotFound { .. })
));
let summary = runtime.summary().unwrap();
assert_eq!(summary.worker_count, 0);
}
#[test]
fn stop_then_cancel_preserves_stopped_terminal_state() {
let runtime = runtime_with_backend();
+99 -4
View File
@@ -31,10 +31,10 @@ use worker_runtime::execution::{
use worker_runtime::fs_store::FsRuntimeStoreOptions;
use worker_runtime::http_server::{
RuntimeHttpConfigBundleAvailabilityResponse, RuntimeHttpConfigBundleSyncRequest,
RuntimeHttpErrorResponse, RuntimeHttpSummaryResponse, RuntimeHttpWorkerInputResponse,
RuntimeHttpWorkerLifecycleRequest, RuntimeHttpWorkerLifecycleResponse,
RuntimeHttpWorkerResponse, RuntimeHttpWorkersResponse, RuntimeHttpWorkingDirectoriesResponse,
RuntimeHttpWorkingDirectoryResponse,
RuntimeHttpErrorResponse, RuntimeHttpSummaryResponse, RuntimeHttpWorkerDeleteResponse,
RuntimeHttpWorkerInputResponse, RuntimeHttpWorkerLifecycleRequest,
RuntimeHttpWorkerLifecycleResponse, RuntimeHttpWorkerResponse, RuntimeHttpWorkersResponse,
RuntimeHttpWorkingDirectoriesResponse, RuntimeHttpWorkingDirectoryResponse,
};
use worker_runtime::identity::{
RuntimeId as EmbeddedRuntimeId, WorkerId as EmbeddedWorkerId, WorkerRef as EmbeddedWorkerRef,
@@ -442,6 +442,15 @@ pub enum WorkerInputKind {
System,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct WorkerDeleteResult {
pub state: WorkerOperationState,
pub runtime_id: String,
pub worker_id: String,
pub deleted: bool,
pub diagnostics: Vec<RuntimeDiagnostic>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct WorkerInputRequest {
#[serde(default = "default_worker_input_kind")]
@@ -670,6 +679,20 @@ pub trait WorkspaceWorkerRuntime: Send + Sync {
}
}
fn delete_worker(&self, worker_id: &str) -> WorkerDeleteResult {
WorkerDeleteResult {
state: WorkerOperationState::Unsupported,
runtime_id: self.runtime_id().to_string(),
worker_id: worker_id.to_string(),
deleted: false,
diagnostics: vec![diagnostic(
"worker_delete_unsupported",
DiagnosticSeverity::Info,
format!("runtime does not implement worker deletion for '{worker_id}'"),
)],
}
}
fn observation_source(
&self,
_worker_id: &str,
@@ -1041,6 +1064,25 @@ impl RuntimeRegistry {
Ok(runtime.cancel_worker(worker_id, request))
}
pub fn delete_worker(
&self,
runtime_id: &str,
worker_id: &str,
) -> Result<WorkerDeleteResult, RuntimeRegistryError> {
validate_backend_identifier("runtime_id", runtime_id)?;
validate_backend_identifier("worker_id", worker_id)?;
let runtime = self.runtime(runtime_id)?;
let lookup = runtime.worker(worker_id);
if lookup.worker.is_none() {
return Err(operation_failed_or_unknown_worker(
runtime_id,
worker_id,
lookup.diagnostics,
));
}
Ok(runtime.delete_worker(worker_id))
}
pub fn observation_source(
&self,
runtime_id: &str,
@@ -1658,6 +1700,38 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime {
}
}
fn delete_worker(&self, worker_id: &str) -> WorkerDeleteResult {
let Some(worker_ref) = self.worker_ref(worker_id) else {
return WorkerDeleteResult {
state: WorkerOperationState::Rejected,
runtime_id: self.runtime_id.clone(),
worker_id: worker_id.to_string(),
deleted: false,
diagnostics: vec![diagnostic(
"embedded_worker_id_invalid",
DiagnosticSeverity::Warning,
"Worker id was empty and cannot be resolved".to_string(),
)],
};
};
match self.runtime.delete_worker(&worker_ref) {
Ok(result) => WorkerDeleteResult {
state: WorkerOperationState::Accepted,
runtime_id: result.runtime_id.to_string(),
worker_id: result.worker_id.to_string(),
deleted: result.deleted,
diagnostics: Vec::new(),
},
Err(error) => WorkerDeleteResult {
state: WorkerOperationState::Rejected,
runtime_id: self.runtime_id.clone(),
worker_id: worker_id.to_string(),
deleted: false,
diagnostics: vec![embedded_runtime_diagnostic(&error)],
},
}
}
fn observation_source(
&self,
worker_id: &str,
@@ -2315,6 +2389,27 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
}
}
fn delete_worker(&self, worker_id: &str) -> WorkerDeleteResult {
match self
.delete_json::<RuntimeHttpWorkerDeleteResponse>(&format!("/v1/workers/{worker_id}"))
{
Ok(response) => WorkerDeleteResult {
state: WorkerOperationState::Accepted,
runtime_id: response.worker.runtime_id.to_string(),
worker_id: response.worker.worker_id.to_string(),
deleted: response.worker.deleted,
diagnostics: Vec::new(),
},
Err(diagnostic) => WorkerDeleteResult {
state: WorkerOperationState::Rejected,
runtime_id: self.runtime_id.clone(),
worker_id: worker_id.to_string(),
deleted: false,
diagnostics: vec![diagnostic],
},
}
}
fn observation_source(
&self,
worker_id: &str,
+63 -16
View File
@@ -27,7 +27,7 @@ use crate::config::{RemoteRuntimeConfigFile, WorkspaceBackendConfigFile, resolve
use crate::hosts::{
ConfigBundleCheckResult, ConfigBundleSyncResult, DiagnosticSeverity, EmbeddedWorkerRuntime,
HostSummary, RemoteRuntimeConfig, RemoteWorkerRuntime, RuntimeDiagnostic, RuntimeRegistry,
RuntimeRegistryUnregisterResult, RuntimeSummary, WorkerCapabilitySummary,
RuntimeRegistryError, RuntimeRegistryUnregisterResult, RuntimeSummary, WorkerCapabilitySummary,
WorkerImplementationSummary, WorkerInputRequest, WorkerInputResult, WorkerLifecycleRequest,
WorkerLifecycleResult, WorkerOperationState, WorkerSpawnAcceptanceRequirement,
WorkerSpawnIntent, WorkerSpawnRequest, WorkerSpawnResult, WorkerSpawnWorkingDirectoryRequest,
@@ -1662,10 +1662,9 @@ fn build_runtime_cleanup_plan(
runtime_worker_id: record.runtime_worker_id.to_string(),
runtime_id: record.runtime_id.clone(),
reason: if blocking_reason.is_some() {
"Worker registry row cannot be deleted until blocking conditions are cleared"
.to_string()
"Worker cannot be deleted until blocking conditions are cleared".to_string()
} else {
"Stopped or unobserved Worker registry row can be manually deleted".to_string()
"Stopped or missing Worker can be manually deleted".to_string()
},
blocking_reason,
pinned,
@@ -1840,6 +1839,7 @@ fn execute_runtime_cleanup(
));
}
let runtime_worker_id = parse_runtime_worker_id_for_registry(&candidate.runtime_worker_id)?;
cleanup_runtime_worker_for_execution(api, runtime_id, candidate)?;
api.store.delete_worker_registry(
&api.config.workspace_id,
candidate.runtime_id.as_str(),
@@ -1849,8 +1849,7 @@ fn execute_runtime_cleanup(
target_id: candidate.target_id.clone(),
action: candidate.action.clone(),
status: "deleted".to_string(),
message: "Worker registry row deleted; Runtime process state was not touched"
.to_string(),
message: "Worker deleted from Runtime and Backend registry".to_string(),
});
}
@@ -1928,6 +1927,29 @@ fn execute_runtime_cleanup(
})
}
fn cleanup_runtime_worker_for_execution(
api: &WorkspaceApi,
runtime_id: &str,
candidate: &CleanupWorkerCandidate,
) -> ApiResult<()> {
match api
.runtime
.delete_worker(runtime_id, candidate.runtime_worker_id.as_str())
{
Ok(result) if result.deleted && result.state == WorkerOperationState::Accepted => Ok(()),
Ok(result) => Err(ApiError::with_diagnostics(
Error::RuntimeOperationFailed {
runtime_id: runtime_id.to_string(),
code: "workspace_cleanup_worker_runtime_delete_rejected".to_string(),
message: "Runtime did not delete selected Worker".to_string(),
},
result.diagnostics,
)),
Err(RuntimeRegistryError::UnknownWorker { .. }) => Ok(()),
Err(error) => Err(error.into_error().into()),
}
}
fn cleanup_runtime_workdir_for_execution(
api: &WorkspaceApi,
runtime_id: &str,
@@ -3183,6 +3205,7 @@ fn workers_response(api: WorkspaceApi) -> ApiResult<RuntimeListResponse<WorkerSu
);
}
}
let mut diagnostics = runtime_workers.diagnostics;
let worker_records = api
.store
.list_worker_registry(&api.config.workspace_id, limit)?;
@@ -3191,13 +3214,38 @@ fn workers_response(api: WorkspaceApi) -> ApiResult<RuntimeListResponse<WorkerSu
.list_workdir_registry(&api.config.workspace_id, 500)?;
let mut items = Vec::new();
for record in worker_records {
if !observed.contains_key(&(record.runtime_id.clone(), record.runtime_worker_id)) {
match api.runtime.worker(
record.runtime_id.as_str(),
&record.runtime_worker_id.to_string(),
) {
Ok(worker) => {
let _ = sync_worker_observation(&api, &worker);
observed.insert(
(worker.runtime_id.clone(), record.runtime_worker_id),
worker,
);
}
Err(RuntimeRegistryError::UnknownWorker { .. }) => {}
Err(error) => diagnostics.push(RuntimeDiagnostic {
code: "worker_detail_probe_failed".to_string(),
severity: DiagnosticSeverity::Info,
message: format!(
"Could not verify Worker {} on Runtime {}: {}",
record.runtime_worker_id,
record.runtime_id,
sanitize_backend_error(&error.into_error().to_string())
),
}),
}
}
let links = api.store.list_worker_workdir_links(
&api.config.workspace_id,
record.runtime_id.as_str(),
record.runtime_worker_id,
)?;
items.push(merge_worker_registry_projection(
observed.get(&(record.runtime_id.clone(), record.runtime_worker_id.clone())),
observed.get(&(record.runtime_id.clone(), record.runtime_worker_id)),
&record,
links,
&workdir_records,
@@ -3208,7 +3256,7 @@ fn workers_response(api: WorkspaceApi) -> ApiResult<RuntimeListResponse<WorkerSu
limit,
items,
source: "backend_worker_registry".to_string(),
diagnostics: runtime_workers.diagnostics,
diagnostics,
})
}
@@ -4016,7 +4064,7 @@ fn worker_summary_from_registry(record: &WorkerRegistryRecord) -> WorkerSummary
host_id: "backend-registry".to_string(),
role: None,
label: record.display_name.clone(),
state: "archived".to_string(),
state: "missing".to_string(),
last_seen_at: Some(record.updated_at.clone()),
pinned: record.retention_state == "pinned",
retention_state: record.retention_state.clone(),
@@ -4032,14 +4080,14 @@ fn worker_summary_from_registry(record: &WorkerRegistryRecord) -> WorkerSummary
profile: record.profile.clone(),
implementation: WorkerImplementationSummary {
kind: "backend_worker_registry".to_string(),
display_hint: "Archived Worker".to_string(),
display_hint: "Missing Worker".to_string(),
},
working_directory: None,
diagnostics: vec![RuntimeDiagnostic {
code: "backend_worker_registry_only".to_string(),
code: "backend_worker_missing".to_string(),
severity: DiagnosticSeverity::Info,
message:
"Worker is preserved in the Backend registry without a live Runtime observation"
"Worker is preserved in the Backend registry but the Runtime did not find it by id"
.to_string(),
}],
}
@@ -4874,12 +4922,12 @@ mod tests {
const TEST_CREATED_AT: &str = "2026-06-23T06:43:28Z";
#[test]
fn backend_worker_projection_preserves_archive_rows_links_and_redacts_paths() {
fn backend_worker_projection_preserves_missing_rows_links_and_redacts_paths() {
let worker = WorkerRegistryRecord {
workspace_id: "workspace-1".to_string(),
runtime_id: "embedded".to_string(),
runtime_worker_id: 1,
display_name: "Archived Worker".to_string(),
display_name: "Missing Worker".to_string(),
profile: Some("builtin:coder".to_string()),
retention_state: "pinned".to_string(),
transcript_ref: Some("runtime://embedded/workers/worker-1/transcript".to_string()),
@@ -4914,8 +4962,7 @@ mod tests {
let projected = merge_worker_registry_projection(None, &worker, vec![link], &[workdir]);
assert_eq!(projected.state, "archived");
assert_eq!(projected.state, "archived");
assert_eq!(projected.state, "missing");
assert_eq!(
projected.working_directory.as_ref().unwrap().status,
WorkingDirectoryStatusKind::Removed