From d3d2b28adc5d6eb1f87b720c5f4e853e1c353f92 Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 24 Jul 2026 11:07:04 +0900 Subject: [PATCH] runtime: simplify worker lifecycle --- crates/worker-runtime/src/execution.rs | 221 ++++++++++++++++--------- crates/worker-runtime/src/fs_store.rs | 2 +- crates/worker-runtime/src/runtime.rs | 109 +++++++----- crates/workspace-server/src/hosts.rs | 212 ++++++++---------------- crates/workspace-server/src/server.rs | 9 +- 5 files changed, 279 insertions(+), 274 deletions(-) diff --git a/crates/worker-runtime/src/execution.rs b/crates/worker-runtime/src/execution.rs index 74dac542..a2b544fb 100644 --- a/crates/worker-runtime/src/execution.rs +++ b/crates/worker-runtime/src/execution.rs @@ -1,4 +1,4 @@ -use crate::catalog::{CreateWorkerRequest, WorkingDirectoryRequest, WorkingDirectoryStatus}; +use crate::catalog::{WorkingDirectoryRequest, WorkingDirectoryStatus}; use crate::config_bundle::ConfigBundle; use crate::error::RuntimeError; use crate::identity::WorkerRef; @@ -11,24 +11,29 @@ use serde::{Deserialize, Serialize}; use std::fmt; use std::sync::Arc; -/// Coarse execution attachment visible through Worker catalog/detail responses. +/// Persisted execution lifecycle visible through Worker catalog/detail responses. /// -/// This deliberately does not expose backend handles, process paths, sockets, -/// credentials, session files, or manifest paths. It only says whether Runtime -/// has an execution backend attached for the Worker. +/// This is intentionally a worker lifecycle projection, not a transport/backend +/// handle state. Runtime restart boundaries invalidate live handles, so a +/// persisted `alive` worker is restored on startup; restore deferral keeps the +/// worker `stopped`, while structural restore failure marks it `corrupted`. #[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum WorkerExecutionBackendKind { + /// Restoreable persisted state exists, but no live execution handle is attached. #[default] - Unconnected, - /// A durable execution binding was restored, but no live handle was recovered. - Stale, - Connected, + #[serde(alias = "unconnected", alias = "stale")] + Stopped, + /// A live execution handle is currently attached. Legacy `connected` maps here. + #[serde(alias = "connected")] + Alive, + /// Persisted execution state is structurally invalid and cannot be restored. + Corrupted, } /// Durable, non-authority execution binding projection. /// -/// This records only enough identity to diagnose stale mappings after restore. +/// This records only enough identity to restore through the same backend kind. /// It is not a live handle and must not contain sockets, paths, credentials, or /// provider-private authority. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] @@ -49,12 +54,11 @@ impl WorkerExecutionBindingIdentity { #[serde(rename_all = "snake_case")] pub enum WorkerExecutionRunState { #[default] - Unconnected, + Stopped, Idle, Busy, Rejected, Errored, - Stopped, } /// Execution operation that produced a result. @@ -69,7 +73,18 @@ pub enum WorkerExecutionOperation { Cancel, } -/// Typed execution result class. +/// Typed execution result class. Results are transient operation outcomes and +/// are not persisted as Worker lifecycle authority. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct WorkerExecutionResult { + pub operation: WorkerExecutionOperation, + pub outcome: WorkerExecutionOutcome, + pub run_state: WorkerExecutionRunState, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub message: Option, +} + +/// Backend result class for a Worker execution operation. #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum WorkerExecutionOutcome { @@ -80,16 +95,6 @@ pub enum WorkerExecutionOutcome { Unsupported, } -/// Backend result for a Worker execution operation. -#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] -pub struct WorkerExecutionResult { - pub operation: WorkerExecutionOperation, - pub outcome: WorkerExecutionOutcome, - pub run_state: WorkerExecutionRunState, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub message: Option, -} - impl WorkerExecutionResult { pub fn accepted( operation: WorkerExecutionOperation, @@ -116,7 +121,7 @@ impl WorkerExecutionResult { Self { operation, outcome: WorkerExecutionOutcome::Rejected, - run_state: WorkerExecutionRunState::Rejected, + run_state: WorkerExecutionRunState::Stopped, message: Some(message.into()), } } @@ -134,7 +139,7 @@ impl WorkerExecutionResult { Self { operation, outcome: WorkerExecutionOutcome::Unsupported, - run_state: WorkerExecutionRunState::Rejected, + run_state: WorkerExecutionRunState::Stopped, message: Some(message.into()), } } @@ -159,28 +164,31 @@ pub struct WorkerExecutionStatus { pub binding: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub working_directory: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub last_result: Option, } impl WorkerExecutionStatus { - pub fn unconnected() -> Self { + pub fn stopped() -> Self { Self::default() } - pub fn connected(run_state: WorkerExecutionRunState) -> Self { + pub fn alive(run_state: WorkerExecutionRunState) -> Self { Self { - backend: WorkerExecutionBackendKind::Connected, + backend: WorkerExecutionBackendKind::Alive, run_state, binding: None, working_directory: None, - last_result: None, } } - pub fn stale(mut previous: Self) -> Self { - previous.backend = WorkerExecutionBackendKind::Stale; - previous.run_state = WorkerExecutionRunState::Unconnected; + pub fn stopped_from(mut previous: Self) -> Self { + previous.backend = WorkerExecutionBackendKind::Stopped; + previous.run_state = WorkerExecutionRunState::Stopped; + previous + } + + pub fn corrupted(mut previous: Self) -> Self { + previous.backend = WorkerExecutionBackendKind::Corrupted; + previous.run_state = WorkerExecutionRunState::Errored; previous } @@ -196,7 +204,6 @@ impl WorkerExecutionStatus { pub fn with_result(mut self, result: WorkerExecutionResult) -> Self { self.run_state = result.run_state; - self.last_result = Some(result); self } } @@ -266,13 +273,20 @@ impl WorkerExecutionContext { &self.worker_ref } - /// Publish a protocol event into the Runtime observation bus. + #[cfg(feature = "ws-server")] + pub fn publish_observation( + &self, + payload: protocol::Event, + ) -> Result { + (self.observation_publisher)(self.worker_ref.clone(), payload) + } + #[cfg(feature = "ws-server")] pub fn publish_protocol_event( &self, payload: protocol::Event, ) -> Result { - (self.observation_publisher)(self.worker_ref.clone(), payload) + self.publish_observation(payload) } } @@ -284,32 +298,29 @@ impl fmt::Debug for WorkerExecutionContext { } } -/// Spawn/initialization request passed to an execution backend. +/// Request passed to a [`WorkerExecutionBackend`] when spawning a Worker. #[derive(Clone, Debug)] pub struct WorkerExecutionSpawnRequest { pub worker_ref: WorkerRef, - pub request: CreateWorkerRequest, + pub request: crate::catalog::CreateWorkerRequest, pub context: WorkerExecutionContext, pub working_directory: Option, pub config_bundle: Option, } -/// Restore request passed to an execution backend for a persisted Runtime Worker. -/// -/// The persisted execution status is a restore hint, not a live handle. Backends -/// must create a fresh controller/handle before returning `Connected`. +/// Request passed to a [`WorkerExecutionBackend`] when restoring a persisted Worker. #[derive(Clone, Debug)] pub struct WorkerExecutionRestoreRequest { pub worker_ref: WorkerRef, - pub request: CreateWorkerRequest, + pub request: crate::catalog::CreateWorkerRequest, pub context: WorkerExecutionContext, pub previous_execution: WorkerExecutionStatus, pub working_directory: Option, pub config_bundle: Option, } -/// Result of backend Worker spawn/initialization. -#[derive(Clone, Debug, PartialEq, Eq)] +/// Backend outcome for Worker spawn/restore operations. +#[derive(Clone, Debug)] pub enum WorkerExecutionSpawnResult { Connected { handle: WorkerExecutionHandle, @@ -320,11 +331,20 @@ pub enum WorkerExecutionSpawnResult { Errored(WorkerExecutionResult), } -/// Backend boundary for Worker execution. -/// -/// Runtime owns Worker catalog, protocol observation, and lifecycle state. A -/// backend owns concrete execution. The default Runtime has no backend, so input -/// to those Workers is rejected instead of producing providerless responses. +impl WorkerExecutionSpawnResult { + pub fn connected( + handle: WorkerExecutionHandle, + run_state: WorkerExecutionRunState, + working_directory: Option, + ) -> Self { + Self::Connected { + handle, + run_state, + working_directory, + } + } +} + pub trait WorkerExecutionBackend: Send + Sync + 'static { fn backend_id(&self) -> &str; @@ -393,6 +413,15 @@ pub trait WorkerExecutionBackend: Send + Sync + 'static { ) } + fn worker_completions( + &self, + _handle: &WorkerExecutionHandle, + _kind: protocol::CompletionKind, + _prefix: &str, + ) -> Vec { + Vec::new() + } + fn stop_worker(&self, _handle: &WorkerExecutionHandle) -> WorkerExecutionResult { WorkerExecutionResult::unsupported( WorkerExecutionOperation::Stop, @@ -411,32 +440,30 @@ pub trait WorkerExecutionBackend: Send + Sync + 'static { fn worker_snapshot(&self, _handle: &WorkerExecutionHandle) -> Option { None } - - fn worker_completions( - &self, - _handle: &WorkerExecutionHandle, - _kind: protocol::CompletionKind, - _prefix: &str, - ) -> Vec { - Vec::new() - } } #[derive(Clone)] pub(crate) struct WorkerExecutionBackendRef { - id: String, + backend_id: String, backend: Arc, } impl WorkerExecutionBackendRef { pub(crate) fn new(backend: Arc) -> Result { - let id = backend.backend_id().trim().to_string(); - if id.is_empty() { + let backend_id = backend.backend_id().to_string(); + if backend_id.trim().is_empty() { return Err(RuntimeError::InvalidRequest( "execution backend id must not be empty".to_string(), )); } - Ok(Self { id, backend }) + Ok(Self { + backend_id, + backend, + }) + } + + pub(crate) fn backend_id(&self) -> &str { + &self.backend_id } pub(crate) fn spawn_worker( @@ -446,12 +473,6 @@ impl WorkerExecutionBackendRef { self.backend.spawn_worker(request) } - #[cfg(feature = "fs-store")] - pub(crate) fn backend_id(&self) -> &str { - &self.id - } - - #[cfg(feature = "fs-store")] pub(crate) fn restore_worker( &self, request: WorkerExecutionRestoreRequest, @@ -500,14 +521,6 @@ impl WorkerExecutionBackendRef { self.backend.dispatch_method(handle, method) } - pub(crate) fn stop_worker(&self, handle: &WorkerExecutionHandle) -> WorkerExecutionResult { - self.backend.stop_worker(handle) - } - - pub(crate) fn cancel_worker(&self, handle: &WorkerExecutionHandle) -> WorkerExecutionResult { - self.backend.cancel_worker(handle) - } - #[cfg(feature = "ws-server")] pub(crate) fn worker_snapshot( &self, @@ -524,12 +537,62 @@ impl WorkerExecutionBackendRef { ) -> Vec { self.backend.worker_completions(handle, kind, prefix) } + + pub(crate) fn stop_worker(&self, handle: &WorkerExecutionHandle) -> WorkerExecutionResult { + self.backend.stop_worker(handle) + } + + pub(crate) fn cancel_worker(&self, handle: &WorkerExecutionHandle) -> WorkerExecutionResult { + self.backend.cancel_worker(handle) + } } impl fmt::Debug for WorkerExecutionBackendRef { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("WorkerExecutionBackendRef") - .field("id", &self.id) + .field("backend_id", &self.backend_id) .finish_non_exhaustive() } } + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + #[test] + fn execution_backend_kind_accepts_legacy_values() { + let connected: WorkerExecutionStatus = serde_json::from_value(json!({ + "backend": "connected", + "run_state": "idle", + "binding": { "backend_id": "worker-crate" } + })) + .unwrap(); + assert_eq!(connected.backend, WorkerExecutionBackendKind::Alive); + + let stale: WorkerExecutionStatus = serde_json::from_value(json!({ + "backend": "stale", + "run_state": "stopped", + "binding": { "backend_id": "worker-crate" } + })) + .unwrap(); + assert_eq!(stale.backend, WorkerExecutionBackendKind::Stopped); + + let unconnected: WorkerExecutionStatus = serde_json::from_value(json!({ + "backend": "unconnected", + "run_state": "stopped" + })) + .unwrap(); + assert_eq!(unconnected.backend, WorkerExecutionBackendKind::Stopped); + } + + #[test] + fn execution_status_serializes_without_last_result() { + let status = WorkerExecutionStatus::alive(WorkerExecutionRunState::Idle).with_result( + WorkerExecutionResult::rejected(WorkerExecutionOperation::Input, "transient"), + ); + let serialized = serde_json::to_value(status).unwrap(); + assert_eq!(serialized["backend"], "alive"); + assert!(serialized.get("last_result").is_none()); + } +} diff --git a/crates/worker-runtime/src/fs_store.rs b/crates/worker-runtime/src/fs_store.rs index 77d7d201..745e196e 100644 --- a/crates/worker-runtime/src/fs_store.rs +++ b/crates/worker-runtime/src/fs_store.rs @@ -473,7 +473,7 @@ struct WorkerSnapshot { worker_id: WorkerId, status: WorkerStatus, request: CreateWorkerRequest, - #[serde(default = "WorkerExecutionStatus::unconnected")] + #[serde(default = "WorkerExecutionStatus::stopped")] execution: WorkerExecutionStatus, last_event_id: u64, } diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index be362126..8f987acb 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -357,7 +357,7 @@ impl Runtime { worker_id: worker_id.clone(), status: WorkerStatus::Running, request: request.clone(), - execution: WorkerExecutionStatus::unconnected(), + execution: WorkerExecutionStatus::stopped(), execution_handle: None, last_event_id: event_id, }; @@ -471,14 +471,13 @@ impl Runtime { match (backend, handle) { (Some(backend), Some(handle)) => (backend, handle), _ => { - let result = WorkerExecutionResult::rejected( - WorkerExecutionOperation::Input, - "worker has no execution backend", - ); let worker = state.worker_mut(worker_ref)?; - let mut execution = WorkerExecutionStatus::unconnected().with_result(result); - execution.binding = worker.execution.binding.clone(); - worker.execution = execution; + worker.execution = + if worker.execution.backend == WorkerExecutionBackendKind::Corrupted { + worker.execution.clone() + } else { + WorkerExecutionStatus::stopped_from(worker.execution.clone()) + }; state.persist_worker(&worker_ref.worker_id)?; return Err(RuntimeError::WorkerExecutionUnavailable { worker_id: worker_ref.worker_id.clone(), @@ -510,11 +509,10 @@ impl Runtime { let worker = state.worker_mut(worker_ref)?; worker.last_event_id = event_id; worker.execution = WorkerExecutionStatus { - backend: WorkerExecutionBackendKind::Connected, + backend: WorkerExecutionBackendKind::Alive, run_state: dispatch_result.run_state, binding: worker.execution.binding.clone(), working_directory: worker.execution.working_directory.clone(), - last_result: Some(dispatch_result), }; let status = worker.status; @@ -592,14 +590,13 @@ impl Runtime { match (backend, handle) { (Some(backend), Some(handle)) => (backend, handle), _ => { - let result = WorkerExecutionResult::rejected( - WorkerExecutionOperation::ProtocolMethod, - "worker has no execution backend", - ); let worker = state.worker_mut(worker_ref)?; - let mut execution = WorkerExecutionStatus::unconnected().with_result(result); - execution.binding = worker.execution.binding.clone(); - worker.execution = execution; + worker.execution = + if worker.execution.backend == WorkerExecutionBackendKind::Corrupted { + worker.execution.clone() + } else { + WorkerExecutionStatus::stopped_from(worker.execution.clone()) + }; state.persist_worker(&worker_ref.worker_id)?; return Err(RuntimeError::WorkerExecutionUnavailable { worker_id: worker_ref.worker_id.clone(), @@ -638,7 +635,7 @@ impl Runtime { let binding = WorkerExecutionBindingIdentity::from_handle(&handle); let worker = state.worker_mut(worker_ref)?; worker.execution_handle = Some(handle); - let mut execution = WorkerExecutionStatus::connected(run_state).with_binding(binding); + let mut execution = WorkerExecutionStatus::alive(run_state).with_binding(binding); if let Some(status) = working_directory { execution = execution.with_working_directory(status); } @@ -669,11 +666,10 @@ impl Runtime { let mut state = self.lock()?; let worker = state.worker_mut(worker_ref)?; worker.execution = WorkerExecutionStatus { - backend: WorkerExecutionBackendKind::Connected, + backend: WorkerExecutionBackendKind::Alive, run_state: result.run_state, binding: worker.execution.binding.clone(), working_directory: worker.execution.working_directory.clone(), - last_result: Some(result), }; state.persist_worker(&worker_ref.worker_id)?; Ok(()) @@ -1048,6 +1044,24 @@ impl Runtime { let candidates = { let mut state = self.lock()?; let Some(backend) = state.execution_backend.clone() else { + let worker_ids: Vec<_> = state.workers.keys().cloned().collect(); + for worker_id in worker_ids { + let Some(worker) = state.workers.get(&worker_id) else { + continue; + }; + if worker.execution.backend != WorkerExecutionBackendKind::Alive { + continue; + } + let worker_ref = worker.worker_ref.clone(); + let worker = state.worker_mut(&worker_ref)?; + worker.execution = + if worker.execution.backend == WorkerExecutionBackendKind::Corrupted { + worker.execution.clone() + } else { + WorkerExecutionStatus::stopped_from(worker.execution.clone()) + }; + state.persist_worker(&worker_ref.worker_id)?; + } return Ok(()); }; let backend_id = backend.backend_id().to_string(); @@ -1059,7 +1073,10 @@ impl Runtime { }; if !worker.status.is_active() || worker.execution_handle.is_some() - || worker.execution.backend != WorkerExecutionBackendKind::Stale + || !matches!( + worker.execution.backend, + WorkerExecutionBackendKind::Alive | WorkerExecutionBackendKind::Stopped + ) || worker .execution .binding @@ -1152,7 +1169,7 @@ impl Runtime { { let worker = state.worker_mut(worker_ref)?; worker.execution_handle = Some(handle.clone()); - let mut execution = WorkerExecutionStatus::connected(run_state) + let mut execution = WorkerExecutionStatus::alive(run_state) .with_binding(WorkerExecutionBindingIdentity::from_handle(&handle)); if let Some(status) = working_directory { execution = execution.with_working_directory(status); @@ -1269,22 +1286,21 @@ impl RuntimeState { let mut diagnostics = persisted.diagnostics; let mut next_diagnostic_id = persisted.next_diagnostic_id; for (worker_id, worker) in persisted.workers { - let execution = if worker.execution.binding.is_some() - && worker.execution.backend == WorkerExecutionBackendKind::Connected + let execution = if worker.execution.backend == WorkerExecutionBackendKind::Alive + && worker.execution.binding.is_none() { - let stale = WorkerExecutionStatus::stale(worker.execution); diagnostics.push(RuntimeDiagnostic { id: next_diagnostic_id, - severity: DiagnosticSeverity::Warning, - code: "worker_execution_mapping_stale".to_string(), + severity: DiagnosticSeverity::Error, + code: "worker_execution_binding_missing".to_string(), message: format!( - "worker {} has persisted execution binding identity but no live execution handle was restored", + "worker {} was persisted as alive but has no execution binding identity", worker.worker_id ), worker_ref: Some(worker.worker_ref.clone()), }); next_diagnostic_id += 1; - stale + WorkerExecutionStatus::corrupted(worker.execution) } else { worker.execution }; @@ -1596,9 +1612,7 @@ impl RuntimeState { }); let worker = self.worker_mut(worker_ref)?; worker.execution_handle = None; - let mut execution = WorkerExecutionStatus::stale(worker.execution.clone()); - execution.last_result = Some(result); - worker.execution = execution; + worker.execution = WorkerExecutionStatus::corrupted(worker.execution.clone()); self.persist_runtime_snapshot()?; self.persist_worker(&worker_ref.worker_id)?; Ok(()) @@ -2623,7 +2637,7 @@ mod tests { assert_eq!(restored_worker.status, WorkerStatus::Stopped); assert_eq!( restored_worker.execution.backend, - WorkerExecutionBackendKind::Stale + WorkerExecutionBackendKind::Stopped ); assert_eq!( restored_worker @@ -2633,16 +2647,6 @@ mod tests { .map(|binding| binding.backend_id.as_str()), Some("test-execution-backend") ); - assert!( - restored - .diagnostics() - .unwrap() - .iter() - .any( - |diagnostic| diagnostic.code == "worker_execution_mapping_stale" - && diagnostic.worker_ref.as_ref() == Some(&worker.worker_ref) - ) - ); #[cfg(feature = "ws-server")] { let observations = restored @@ -2708,6 +2712,19 @@ mod tests { .unwrap(); drop(runtime); + let backendless = Runtime::with_fs_store(crate::fs_store::FsRuntimeStoreOptions { + root: root.clone(), + display_name: None, + limits: RuntimeLimits::default(), + }) + .unwrap(); + let stopped_worker = backendless.worker_detail(&worker.worker_ref).unwrap(); + assert_eq!( + stopped_worker.execution.backend, + WorkerExecutionBackendKind::Stopped + ); + drop(backendless); + let restoring_backend = Arc::new(TestExecutionBackend::default()); let restored = Runtime::with_fs_store_and_execution_backend( crate::fs_store::FsRuntimeStoreOptions { @@ -2724,7 +2741,7 @@ mod tests { assert_eq!(restored_worker.status, WorkerStatus::Running); assert_eq!( restored_worker.execution.backend, - WorkerExecutionBackendKind::Connected + WorkerExecutionBackendKind::Alive ); assert!(restored_worker.execution.binding.is_some()); restored @@ -2748,7 +2765,7 @@ mod tests { #[cfg(feature = "fs-store")] #[test] - fn fs_store_keeps_worker_stale_when_execution_restore_fails() { + fn fs_store_marks_worker_corrupted_when_execution_restore_fails() { let root = fs_store_root("execution-restore-failed"); let runtime = Runtime::with_fs_store_and_execution_backend( crate::fs_store::FsRuntimeStoreOptions { @@ -2785,7 +2802,7 @@ mod tests { assert_eq!(restored_worker.status, WorkerStatus::Running); assert_eq!( restored_worker.execution.backend, - WorkerExecutionBackendKind::Stale + WorkerExecutionBackendKind::Corrupted ); assert!( restored diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index 1717b58d..9987c24d 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -1629,38 +1629,24 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime { }), }; match self.runtime.create_worker(create_request) { - Ok(detail) => { - let execution_failure = - embedded_spawn_execution_failure_diagnostic(&detail.execution); - if let Some(diagnostic) = execution_failure { - diagnostics.push(diagnostic); - WorkerSpawnResult { - state: WorkerOperationState::Rejected, - worker: Some(self.map_worker_detail(detail)), - acceptance_evidence: Vec::new(), - diagnostics, - } - } else { - WorkerSpawnResult { - state: WorkerOperationState::Accepted, - worker: Some(self.map_worker_detail(detail)), - acceptance_evidence: vec![ - WorkerSpawnAcceptanceEvidence { - kind: "embedded_runtime_worker_created".to_string(), - detail: "worker-runtime catalog accepted a backend-internal tools-less Worker" - .to_string(), - }, - WorkerSpawnAcceptanceEvidence { - kind: "embedded_runtime_backend_internal_projection".to_string(), - detail: - "only runtime_id plus worker_id backend projections were exposed" - .to_string(), - }, - ], - diagnostics, - } - } - } + Ok(detail) => WorkerSpawnResult { + state: WorkerOperationState::Accepted, + worker: Some(self.map_worker_detail(detail)), + acceptance_evidence: vec![ + WorkerSpawnAcceptanceEvidence { + kind: "embedded_runtime_worker_created".to_string(), + detail: + "worker-runtime catalog accepted a backend-internal tools-less Worker" + .to_string(), + }, + WorkerSpawnAcceptanceEvidence { + kind: "embedded_runtime_backend_internal_projection".to_string(), + detail: "only runtime_id plus worker_id backend projections were exposed" + .to_string(), + }, + ], + diagnostics, + }, Err(err) => { diagnostics.push(embedded_runtime_diagnostic(&err)); WorkerSpawnResult { @@ -2710,38 +2696,6 @@ fn embedded_runtime_status_label(status: RuntimeStatus) -> &'static str { } } -fn embedded_spawn_execution_failure_diagnostic( - execution: &worker_runtime::execution::WorkerExecutionStatus, -) -> Option { - let result = execution.last_result.as_ref()?; - let severity = match result.outcome { - worker_runtime::execution::WorkerExecutionOutcome::Accepted => return None, - worker_runtime::execution::WorkerExecutionOutcome::Rejected - | worker_runtime::execution::WorkerExecutionOutcome::Busy - | worker_runtime::execution::WorkerExecutionOutcome::Unsupported => { - DiagnosticSeverity::Warning - } - worker_runtime::execution::WorkerExecutionOutcome::Errored => DiagnosticSeverity::Error, - }; - let status = match result.outcome { - worker_runtime::execution::WorkerExecutionOutcome::Accepted => "accepted", - worker_runtime::execution::WorkerExecutionOutcome::Rejected => "rejected", - worker_runtime::execution::WorkerExecutionOutcome::Busy => "busy", - worker_runtime::execution::WorkerExecutionOutcome::Unsupported => "unsupported", - worker_runtime::execution::WorkerExecutionOutcome::Errored => "errored", - }; - let detail = result - .message - .as_deref() - .map(|message| format!(": {message}")) - .unwrap_or_default(); - Some(diagnostic( - format!("embedded_worker_execution_spawn_{status}"), - severity, - format!("Embedded Worker execution spawn was {status} during setup{detail}"), - )) -} - fn runtime_worker_can_stop( execution_enabled: bool, status: EmbeddedWorkerStatus, @@ -2749,26 +2703,11 @@ fn runtime_worker_can_stop( ) -> bool { execution_enabled && status == EmbeddedWorkerStatus::Running - && execution.backend == worker_runtime::execution::WorkerExecutionBackendKind::Connected - && !matches!( + && execution.backend == worker_runtime::execution::WorkerExecutionBackendKind::Alive + && matches!( execution.run_state, - WorkerExecutionRunState::Stopped - | WorkerExecutionRunState::Rejected - | WorkerExecutionRunState::Errored - | WorkerExecutionRunState::Unconnected + WorkerExecutionRunState::Idle | WorkerExecutionRunState::Busy ) - && !execution_last_result_blocks_control(execution) -} - -fn execution_last_result_blocks_control(execution: &WorkerExecutionStatus) -> bool { - execution.last_result.as_ref().is_some_and(|result| { - matches!( - result.outcome, - worker_runtime::execution::WorkerExecutionOutcome::Rejected - | worker_runtime::execution::WorkerExecutionOutcome::Errored - | worker_runtime::execution::WorkerExecutionOutcome::Unsupported - ) - }) } fn embedded_worker_status_label(status: EmbeddedWorkerStatus) -> &'static str { @@ -2788,27 +2727,26 @@ fn embedded_worker_projection_diagnostics( "Worker identity is projected only as runtime_id plus worker_id; embedded runtime internals remain backend-private".to_string(), )]; - if execution.backend == WorkerExecutionBackendKind::Stale { - diagnostics.push(diagnostic( - "embedded_worker_execution_stale", - DiagnosticSeverity::Warning, - "Worker execution handle is not connected in this server process; persisted execution binding was marked stale".to_string(), - )); - } else if execution.backend == WorkerExecutionBackendKind::Unconnected - || execution.run_state == WorkerExecutionRunState::Unconnected - { - diagnostics.push(diagnostic( - "embedded_worker_execution_unconnected", - DiagnosticSeverity::Warning, - "Worker execution handle is not connected in this server process".to_string(), - )); - } else if execution.run_state == WorkerExecutionRunState::Rejected { - diagnostics.push(diagnostic( - "embedded_worker_execution_spawn_rejected", + match execution.backend { + WorkerExecutionBackendKind::Stopped => diagnostics.push(diagnostic( + "embedded_worker_execution_stopped", + DiagnosticSeverity::Info, + "Worker execution is stopped; the runtime will restore it before dispatching input or protocol methods".to_string(), + )), + WorkerExecutionBackendKind::Corrupted => diagnostics.push(diagnostic( + "embedded_worker_execution_corrupted", DiagnosticSeverity::Error, - "Worker execution spawn was rejected; backend-private details are not exposed" - .to_string(), - )); + "Worker execution state is corrupted and cannot be restored without repair".to_string(), + )), + WorkerExecutionBackendKind::Alive => { + if execution.run_state == WorkerExecutionRunState::Rejected { + diagnostics.push(diagnostic( + "embedded_worker_execution_rejected", + DiagnosticSeverity::Warning, + "Worker execution rejected the last transient operation; retry or inspect runtime logs".to_string(), + )); + } + } } diagnostics @@ -2821,19 +2759,19 @@ fn embedded_worker_execution_status_label( match status { EmbeddedWorkerStatus::Stopped => "stopped", EmbeddedWorkerStatus::Cancelled => "cancelled", - EmbeddedWorkerStatus::Running => { - if execution.backend == worker_runtime::execution::WorkerExecutionBackendKind::Stale { - return "stale"; + EmbeddedWorkerStatus::Running => match execution.backend { + worker_runtime::execution::WorkerExecutionBackendKind::Stopped => "stopped", + worker_runtime::execution::WorkerExecutionBackendKind::Corrupted => "corrupted", + worker_runtime::execution::WorkerExecutionBackendKind::Alive => { + match execution.run_state { + WorkerExecutionRunState::Idle => "idle", + WorkerExecutionRunState::Busy => "running", + WorkerExecutionRunState::Stopped => "stopped", + WorkerExecutionRunState::Rejected => "rejected", + WorkerExecutionRunState::Errored => "errored", + } } - match execution.run_state { - WorkerExecutionRunState::Idle => "idle", - WorkerExecutionRunState::Busy => "running", - WorkerExecutionRunState::Stopped => "stopped", - WorkerExecutionRunState::Rejected => "rejected", - WorkerExecutionRunState::Errored => "errored", - WorkerExecutionRunState::Unconnected => "unconnected", - } - } + }, } } @@ -4331,7 +4269,7 @@ mod tests { } #[test] - fn remote_runtime_projection_blocks_stale_and_unconnected_execution_input() { + fn remote_runtime_projection_blocks_stopped_and_corrupted_execution_stop() { let (base_url, server) = serve_mock_http(vec![ mock_response( "GET", @@ -4343,30 +4281,26 @@ mod tests { worker_json_with_execution( "remote:primary", "1", - "stale", - "unconnected", - None, + "stopped", + "stopped", ), worker_json_with_execution( "remote:primary", "2", - "unconnected", - "unconnected", - None, + "corrupted", + "errored", ), worker_json_with_execution( "remote:primary", "3", - "connected", + "alive", "rejected", - Some("rejected"), ), worker_json_with_execution( "remote:primary", "4", - "connected", + "alive", "errored", - Some("errored"), ) ] }) @@ -4381,9 +4315,8 @@ mod tests { "worker": worker_json_with_execution( "remote:primary", "1", - "stale", - "unconnected", - None, + "stopped", + "stopped", )}) .to_string(), ), @@ -4411,14 +4344,14 @@ mod tests { worker.worker_id ); } - assert_eq!(workers.items[0].state, "stale"); - assert_eq!(workers.items[1].state, "unconnected"); + assert_eq!(workers.items[0].state, "stopped"); + assert_eq!(workers.items[1].state, "corrupted"); assert_eq!(workers.items[2].state, "rejected"); assert_eq!(workers.items[3].state, "errored"); - let stale_detail = registry.worker("remote:primary", "1").unwrap(); - assert!(!stale_detail.capabilities.can_stop); - assert_eq!(stale_detail.state, "stale"); + let stopped_detail = registry.worker("remote:primary", "1").unwrap(); + assert!(!stopped_detail.capabilities.can_stop); + assert_eq!(stopped_detail.state, "stopped"); server.join().expect("mock remote server finished"); } @@ -4599,7 +4532,7 @@ mod tests { } fn worker_json(runtime_id: &str, worker_id: &str) -> serde_json::Value { - worker_json_with_execution(runtime_id, worker_id, "connected", "idle", None) + worker_json_with_execution(runtime_id, worker_id, "alive", "idle") } fn worker_json_with_execution( @@ -4607,23 +4540,14 @@ mod tests { worker_id: &str, backend: &str, run_state: &str, - last_outcome: Option<&str>, ) -> serde_json::Value { - let last_result = last_outcome.map(|outcome| { - json!({ - "operation": "input", - "outcome": outcome, - "run_state": run_state, - "message": format!("{outcome} result") - }) - }); let worker_id = worker_id.parse::().unwrap(); json!({ "worker_ref": { "runtime_id": runtime_id, "worker_id": worker_id }, "runtime_id": runtime_id, "worker_id": worker_id, "status": "running", - "execution": { "backend": backend, "run_state": run_state, "last_result": last_result }, + "execution": { "backend": backend, "run_state": run_state }, "intent": { "kind": "role", "role": "coder", "purpose": "remote test" }, "profile": { "kind": "builtin", "value": "coder" }, "profile_source": { diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 9a1e5732..5ff3a4ff 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -8452,7 +8452,8 @@ mod tests { } #[tokio::test] - async fn embedded_runtime_fs_store_restores_catalog_config_bundle_and_stale_execution() { + async fn embedded_runtime_fs_store_restores_catalog_config_bundle_and_corrupts_failed_execution() + { let dir = tempfile::tempdir().unwrap(); let config = test_server_config(dir.path().join("workspace")); let store_root = config.embedded_runtime_store_root.clone(); @@ -8539,13 +8540,13 @@ mod tests { .runtime .worker("embedded-worker-runtime", &worker_id) .expect("restored worker"); - assert_eq!(restored_worker.state, "stale"); + assert_eq!(restored_worker.state, "corrupted"); assert!(!restored_worker.capabilities.can_stop); assert!( restored_worker .diagnostics .iter() - .any(|diagnostic| diagnostic.code == "embedded_worker_execution_stale") + .any(|diagnostic| diagnostic.code == "embedded_worker_execution_corrupted") ); let bundles = restored @@ -8566,7 +8567,7 @@ mod tests { &worker_id, WorkerInputRequest { kind: WorkerInputKind::User, - content: "should not be routed to stale handle".to_string(), + content: "should not be routed to corrupted handle".to_string(), segments: None, }, )