diff --git a/crates/protocol/src/lib.rs b/crates/protocol/src/lib.rs index f66be5dc..d54c9232 100644 --- a/crates/protocol/src/lib.rs +++ b/crates/protocol/src/lib.rs @@ -125,6 +125,7 @@ pub enum WorkerCommandDisposition { StaleExecutionGeneration, StaleWorkerStateRevision, StaleCommandId, + Conflict, InvalidState, } diff --git a/crates/protocol/src/subscription.rs b/crates/protocol/src/subscription.rs index 468e241c..bf4a8635 100644 --- a/crates/protocol/src/subscription.rs +++ b/crates/protocol/src/subscription.rs @@ -573,6 +573,11 @@ pub struct SubscriptionWorker { pub resource_key: Option, /// Producer-owned monotonic revision for this Worker subject. pub subject_revision: u64, + /// Latest revisioned foreground state observed from the Worker. This remains + /// absent until an authoritative Worker snapshot/event has been applied. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub worker_state: Option, + /// Runtime catalog lifecycle compatibility projection; not foreground-state authority. pub state: SubscriptionWorkerState, #[serde(default)] pub has_running_internal_workers: bool, @@ -874,6 +879,7 @@ mod tests { runtime_id: None, resource_key: None, subject_revision: 0, + worker_state: None, state: SubscriptionWorkerState::Idle, has_running_internal_workers: false, workspace_id: Some("workspace-1".to_string()), diff --git a/crates/tui/src/app.rs b/crates/tui/src/app.rs index c543f991..9858dd39 100644 --- a/crates/tui/src/app.rs +++ b/crates/tui/src/app.rs @@ -1180,18 +1180,14 @@ impl App { self.assistant_streaming = false; } Event::TurnStart { .. } => { - self.set_worker_status(WorkerStatus::Running); self.run_requests += 1; self.current_tool = None; self.latest_llm_wait_event = None; self.assistant_streaming = false; } - Event::InvokeStart { .. } => { - self.set_worker_status(WorkerStatus::Running); - } + Event::InvokeStart { .. } => {} // UI consumers of per-attempt LlmCall semantics remain out of scope; - // the run-level status starts at InvokeStart and TurnStart counts each - // LLM request within that run. + // authoritative run state comes only from WorkerStateSnapshot. Event::LlmCallStart { .. } | Event::LlmCallEnd { .. } => { self.latest_llm_wait_event = None; } @@ -1398,12 +1394,7 @@ impl App { output_tokens: self.run_output_tokens, }); self.pending_submit_rollback = None; - self.reset_run_state(match result { - RunResult::Paused => WorkerStatus::Paused, - RunResult::Finished | RunResult::LimitReached | RunResult::RolledBack => { - WorkerStatus::Idle - } - }); + self.reset_run_state(); } } Event::CompactStart { .. } => { @@ -1536,7 +1527,7 @@ impl App { }; self.completion = None; self.close_rewind_picker(); - self.reset_run_state(self.worker_status); + self.reset_run_state(); let mut message = if restored_composer { format!( "Rewound session: discarded {} log entries; restored selected input to composer.", @@ -1584,8 +1575,7 @@ impl App { None } - fn reset_run_state(&mut self, status: WorkerStatus) { - self.set_worker_status(status); + fn reset_run_state(&mut self) { self.run_requests = 0; self.run_upload_tokens = 0; self.run_output_tokens = 0; @@ -1615,7 +1605,7 @@ impl App { "Rolled back empty assistant turn; no local submitted input was available to restore." .to_owned() }; - self.reset_run_state(WorkerStatus::Idle); + self.reset_run_state(); self.blocks.push(Block::Alert { level: AlertLevel::Warn, source: AlertSource::Worker, @@ -3583,6 +3573,37 @@ mod completion_flow_tests { assert!(matches!(app.blocks.first(), Some(Block::Greeting(_)))); } + #[test] + fn occurrence_events_do_not_infer_foreground_worker_state() { + let mut app = App::new("test".into()); + app.handle_worker_event(Event::TurnStart { turn: 1 }); + app.handle_worker_event(Event::InvokeStart { + kind: protocol::InvokeKind::UserSend, + }); + app.handle_worker_event(Event::RunEnd { + result: RunResult::Paused, + }); + assert_eq!(app.worker_state.state, protocol::WorkerState::Idle); + assert_eq!(app.worker_status, WorkerStatus::Idle); + + let running = WorkerStateSnapshot { + execution_generation: 1, + revision: 1, + state: protocol::WorkerState::Busy(protocol::WorkerBusyState::Run( + protocol::WorkerRunState::Running, + )), + last_command_id: 0, + }; + app.handle_worker_event(Event::WorkerState { + snapshot: running.clone(), + }); + app.handle_worker_event(Event::RunEnd { + result: RunResult::Finished, + }); + assert_eq!(app.worker_state, running); + assert_eq!(app.worker_status, WorkerStatus::Running); + } + #[test] fn worker_state_events_and_acknowledgements_share_monotonic_application() { let mut app = App::new("test".into()); diff --git a/crates/tui/src/backend_worker_picker.rs b/crates/tui/src/backend_worker_picker.rs index e9a37271..473ca5eb 100644 --- a/crates/tui/src/backend_worker_picker.rs +++ b/crates/tui/src/backend_worker_picker.rs @@ -321,7 +321,7 @@ fn row_line( Span::raw(" "), Span::styled( pad_column(&worker_state(worker), widths.state), - state_style(worker.state.as_str()), + state_style(worker_state_label(worker)), ), Span::raw(" "), Span::styled( @@ -341,8 +341,20 @@ fn worker_name(worker: &BackendWorkerSummary) -> &str { } } +fn worker_state_label(worker: &BackendWorkerSummary) -> &str { + match worker.worker_state.as_ref().map(|state| &state.state) { + Some(protocol::WorkerState::Idle) => "idle", + Some(protocol::WorkerState::Busy(protocol::WorkerBusyState::Run( + protocol::WorkerRunState::Paused, + ))) => "paused", + Some(protocol::WorkerState::Busy(_)) => "running", + None if worker.state == "stopped" => "stopped", + None => "unknown", + } +} + fn worker_state(worker: &BackendWorkerSummary) -> String { - format!("[{}]", worker.state) + format!("[{}]", worker_state_label(worker)) } fn text_width(value: &str) -> usize { @@ -413,7 +425,15 @@ mod tests { identity: "ws".to_string(), workspace_id: Some("ws".to_string()), }, - state: "running".to_string(), + state: "idle".to_string(), + worker_state: Some(protocol::WorkerStateSnapshot { + execution_generation: 1, + revision: 1, + state: protocol::WorkerState::Busy(protocol::WorkerBusyState::Run( + protocol::WorkerRunState::Running, + )), + last_command_id: 0, + }), last_seen_at: None, pinned: false, retention_state: String::new(), @@ -450,6 +470,7 @@ mod tests { worker.display_name = "Coder".to_string(); worker.label = "Coder · T-585".to_string(); worker.state = "stopped".to_string(); + worker.worker_state = None; worker.working_directory = Some( serde_json::from_value(serde_json::json!({ "working_directory_id": "001a06a9f0202000000", @@ -478,12 +499,19 @@ mod tests { short.label = "Coder".to_string(); short.display_name = short.label.clone(); short.state = "idle".to_string(); + short.worker_state = Some(protocol::WorkerStateSnapshot { + execution_generation: 1, + revision: 2, + state: protocol::WorkerState::Idle, + last_command_id: 0, + }); let mut long = worker("runtime-a", "worker-b", None); long.resource_key = "W-100".to_string(); long.label = "Longer worker · T-9".to_string(); long.display_name = long.label.clone(); long.state = "stopped".to_string(); + long.worker_state = None; for worker in [&mut short, &mut long] { worker.working_directory = Some( diff --git a/crates/worker-runtime/src/catalog.rs b/crates/worker-runtime/src/catalog.rs index 3cc27d01..96bb2beb 100644 --- a/crates/worker-runtime/src/catalog.rs +++ b/crates/worker-runtime/src/catalog.rs @@ -307,6 +307,8 @@ pub struct WorkerSummary { pub worker_id: WorkerId, pub status: WorkerStatus, #[serde(default, skip_serializing_if = "Option::is_none")] + pub worker_state: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] pub workspace_id: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub working_directory: Option, @@ -325,6 +327,8 @@ pub struct WorkerDetail { pub worker_id: WorkerId, pub status: WorkerStatus, #[serde(default, skip_serializing_if = "Option::is_none")] + pub worker_state: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] pub workspace_id: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub working_directory: Option, @@ -341,6 +345,8 @@ pub struct WorkerDetail { pub struct WorkerLifecycleAck { pub worker_ref: WorkerRef, pub status: WorkerStatus, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub worker_state: Option, } #[cfg(test)] diff --git a/crates/worker-runtime/src/http_server.rs b/crates/worker-runtime/src/http_server.rs index 28cf26aa..c05bdb91 100644 --- a/crates/worker-runtime/src/http_server.rs +++ b/crates/worker-runtime/src/http_server.rs @@ -3486,7 +3486,16 @@ mod ws_tests { .. }) if delivered_subscription_id == subscription_id && worker.worker_id.as_str() == worker_ref.worker_id.to_string() - && worker.state == protocol::subscription::SubscriptionWorkerState::Running + && worker.state == protocol::subscription::SubscriptionWorkerState::Idle + && matches!( + worker.worker_state, + Some(protocol::WorkerStateSnapshot { + state: protocol::WorkerState::Busy(protocol::WorkerBusyState::Run( + protocol::WorkerRunState::Running + )), + .. + }) + ) )); let unsubscribe_request_id = diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index 808788f4..76066336 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -700,6 +700,7 @@ impl Runtime { worker_ref: worker_ref.clone(), worker_id: worker_id.clone(), status: WorkerStatus::Stopped, + worker_state: None, workspace_id: scope.map(|scope| scope.workspace_id.clone()), request: durable_request, run_generation: 1, @@ -787,7 +788,6 @@ impl Runtime { let detail = self.commit_created_worker( &worker_ref, handle, - WorkerStatus::Running, working_directory, dispatch_result, )?; @@ -797,7 +797,6 @@ impl Runtime { self.commit_created_worker( &worker_ref, handle, - WorkerStatus::Idle, working_directory, WorkerExecutionResult::accepted(WorkerExecutionOperation::Spawn), ) @@ -1220,17 +1219,7 @@ impl Runtime { state.ensure_running()?; let worker = state.worker_mut(worker_ref)?; if let Some(snapshot) = dispatch_result.worker_state.as_ref() { - worker.status = match snapshot.catalog_status() { - protocol::WorkerStatus::Idle => WorkerStatus::Idle, - protocol::WorkerStatus::Running => WorkerStatus::Running, - protocol::WorkerStatus::Paused => WorkerStatus::Paused, - protocol::WorkerStatus::Stopped => WorkerStatus::Stopped, - }; - } else if matches!( - submission.as_ref().map(|ack| ack.disposition), - Some(protocol::SubmissionDisposition::Started) - ) { - worker.status = WorkerStatus::Running; + let _ = worker.apply_worker_state(snapshot); } let status = worker.status; #[cfg(feature = "ws-server")] @@ -1490,16 +1479,19 @@ impl Runtime { &self, worker_ref: &WorkerRef, handle: WorkerExecutionHandle, - status: WorkerStatus, working_directory: Option, - _result: WorkerExecutionResult, + result: WorkerExecutionResult, ) -> Result { let mut state = self.lock()?; let detail = { let worker = state.worker_mut(worker_ref)?; worker.execution_handle = Some(handle); worker.execution_bound = true; - worker.status = status; + worker.status = WorkerStatus::Idle; + worker.worker_state = None; + if let Some(snapshot) = result.worker_state.as_ref() { + let _ = worker.apply_worker_state(snapshot); + } worker.restore_intent = restore_intent_for_status(worker.status); worker.working_directory = working_directory; worker.detail() @@ -1536,16 +1528,14 @@ impl Runtime { let Some(snapshot) = result.worker_state else { return Ok(()); }; - let status = match snapshot.catalog_status() { - protocol::WorkerStatus::Idle => WorkerStatus::Idle, - protocol::WorkerStatus::Running => WorkerStatus::Running, - protocol::WorkerStatus::Paused => WorkerStatus::Paused, - protocol::WorkerStatus::Stopped => WorkerStatus::Stopped, - }; let mut state = self.lock()?; let worker = state.worker_mut(worker_ref)?; - worker.status = status; - worker.restore_intent = restore_intent_for_status(status); + let applied = worker + .apply_worker_state(&snapshot) + .is_ok_and(|result| matches!(result, protocol::WorkerStateSnapshotApply::Applied)); + if !applied { + return Ok(()); + } state.publish_worker_upsert(worker_ref.worker_id)?; state.persist_runtime_snapshot()?; state.persist_worker(&worker_ref.worker_id)?; @@ -1636,20 +1626,26 @@ impl Runtime { worker_ref: &WorkerRef, reason: Option, ) -> Result { - let current = { + { let state = self.lock()?; state.ensure_running()?; - state.worker(worker_ref)?.status - }; - if matches!(current, WorkerStatus::Idle | WorkerStatus::Stopped) { - return Ok(WorkerLifecycleAck { - worker_ref: worker_ref.clone(), - status: current, - }); + if state.worker(worker_ref)?.status == WorkerStatus::Stopped { + return Ok(WorkerLifecycleAck { + worker_ref: worker_ref.clone(), + status: WorkerStatus::Stopped, + worker_state: None, + }); + } } self.dispatch_lifecycle_to_backend(worker_ref, WorkerExecutionOperation::Cancel)?; let _ = reason; - self.transition_worker_preserving_execution(worker_ref, WorkerStatus::Idle) + let state = self.lock()?; + let worker = state.worker(worker_ref)?; + Ok(WorkerLifecycleAck { + worker_ref: worker_ref.clone(), + status: worker.status, + worker_state: worker.worker_state.clone(), + }) } /// Delete a non-running Worker through a workspace-scoped Runtime authorization context. @@ -1795,12 +1791,13 @@ impl Runtime { ) -> Result { let mut state = self.lock()?; state.ensure_worker_ref(worker_ref)?; - let status_changed = state.project_protocol_event_to_status(worker_ref, &payload); + let worker_state_changed = + state.project_protocol_event_to_worker_state(worker_ref, &payload); let activity_changed = state.project_internal_worker_activity(worker_ref, &payload); - if status_changed || activity_changed { + if worker_state_changed || activity_changed { state.publish_worker_upsert(worker_ref.worker_id)?; } - if status_changed { + if worker_state_changed { state.persist_runtime_snapshot()?; state.persist_worker(&worker_ref.worker_id)?; } @@ -1836,26 +1833,6 @@ impl Runtime { Ok(()) } - fn transition_worker_preserving_execution( - &self, - worker_ref: &WorkerRef, - status: WorkerStatus, - ) -> Result { - let mut state = self.lock()?; - state.ensure_running()?; - let worker = state.worker_mut(worker_ref)?; - worker.status = status; - worker.restore_intent = restore_intent_for_status(status); - let status = worker.status; - state.publish_worker_upsert(worker_ref.worker_id)?; - state.persist_runtime_snapshot()?; - state.persist_worker(&worker_ref.worker_id)?; - Ok(WorkerLifecycleAck { - worker_ref: worker_ref.clone(), - status, - }) - } - fn transition_worker( &self, worker_ref: &WorkerRef, @@ -1867,6 +1844,7 @@ impl Runtime { let worker = state.worker_mut(worker_ref)?; worker.status = status; + worker.worker_state = None; worker.restore_intent = restore_intent_for_status(status); worker.execution_handle = None; worker.internal_workers.clear(); @@ -1877,6 +1855,7 @@ impl Runtime { Ok(WorkerLifecycleAck { worker_ref: worker_ref.clone(), status, + worker_state: None, }) } @@ -2301,6 +2280,7 @@ impl RuntimeState { worker_ref: worker.worker_ref, worker_id: worker.worker_id, status: worker.status, + worker_state: None, workspace_id: worker.workspace_id, request: worker.request, run_generation, @@ -2630,6 +2610,7 @@ impl RuntimeState { .get(&worker.worker_id) .copied() .unwrap_or(0), + worker_state: worker.worker_state.clone(), state: subscription_worker_state(worker.status), has_running_internal_workers: worker .internal_workers @@ -2965,7 +2946,7 @@ impl RuntimeState { Self::update_internal_worker_activity(&mut worker.internal_workers, event) } - fn project_protocol_event_to_status( + fn project_protocol_event_to_worker_state( &mut self, worker_ref: &WorkerRef, event: &protocol::Event, @@ -2973,7 +2954,7 @@ impl RuntimeState { let Some(worker) = self.workers.get_mut(&worker_ref.worker_id) else { return false; }; - let next_status = match event { + let incoming = match event { protocol::Event::WorkerState { snapshot } | protocol::Event::Snapshot { state: snapshot, .. @@ -2983,21 +2964,16 @@ impl RuntimeState { protocol::WorkerCommandAcknowledgement { state: snapshot, .. }, - } => Some(match snapshot.catalog_status() { - protocol::WorkerStatus::Idle => WorkerStatus::Idle, - protocol::WorkerStatus::Running => WorkerStatus::Running, - protocol::WorkerStatus::Paused => WorkerStatus::Paused, - protocol::WorkerStatus::Stopped => WorkerStatus::Stopped, - }), - _ => None, + } => snapshot, + _ => return false, }; - if let Some(next_status) = next_status { - let changed = worker.status != next_status; - worker.status = next_status; - worker.restore_intent = restore_intent_for_status(next_status); - changed - } else { - false + match worker.apply_worker_state(incoming) { + Ok(protocol::WorkerStateSnapshotApply::Applied) => true, + Ok( + protocol::WorkerStateSnapshotApply::Duplicate + | protocol::WorkerStateSnapshotApply::Stale, + ) + | Err(_) => false, } } } @@ -3013,6 +2989,7 @@ struct WorkerRecord { worker_ref: WorkerRef, worker_id: WorkerId, status: WorkerStatus, + worker_state: Option, workspace_id: Option, request: CreateWorkerRequest, run_generation: u64, @@ -3024,6 +3001,19 @@ struct WorkerRecord { } impl WorkerRecord { + fn apply_worker_state( + &mut self, + incoming: &protocol::WorkerStateSnapshot, + ) -> Result { + match self.worker_state.as_mut() { + Some(current) => protocol::apply_worker_state_snapshot(current, incoming), + None => { + self.worker_state = Some(incoming.clone()); + Ok(protocol::WorkerStateSnapshotApply::Applied) + } + } + } + fn belongs_to_workspace(&self, workspace_id: &str) -> bool { self.workspace_id.as_deref() == Some(workspace_id) } @@ -3033,6 +3023,7 @@ impl WorkerRecord { worker_ref: self.worker_ref.clone(), worker_id: self.worker_id, status: self.status, + worker_state: self.worker_state.clone(), workspace_id: self.workspace_id.clone(), working_directory: self.working_directory.clone(), profile: self.request.profile.clone(), @@ -3047,6 +3038,7 @@ impl WorkerRecord { worker_ref: self.worker_ref.clone(), worker_id: self.worker_id, status: self.status, + worker_state: self.worker_state.clone(), workspace_id: self.workspace_id.clone(), working_directory: self.working_directory.clone(), profile: self.request.profile.clone(), @@ -4719,7 +4711,7 @@ mod tests { } #[test] - fn create_worker_uses_started_submission_ack_for_initial_running_status() { + fn create_worker_does_not_infer_state_from_started_submission_ack() { let (runtime, backend) = runtime_and_backend(); backend.set_dispatch_result(WorkerExecutionResult::accepted_submission( WorkerExecutionOperation::Input, @@ -4732,7 +4724,69 @@ mod tests { let detail = runtime.create_worker(request).unwrap(); - assert_eq!(detail.status, WorkerStatus::Running); + assert_eq!(detail.status, WorkerStatus::Idle); + assert_eq!(detail.worker_state, None); + } + + #[test] + fn runtime_applies_only_newer_worker_state_snapshots() { + let (runtime, _) = runtime_and_backend(); + let detail = runtime + .create_worker(task_request("state ordering")) + .unwrap(); + let running = protocol::WorkerStateSnapshot { + execution_generation: 7, + revision: 3, + state: protocol::WorkerState::Busy(protocol::WorkerBusyState::Run( + protocol::WorkerRunState::Running, + )), + last_command_id: 2, + }; + assert!({ + let mut state = runtime.lock().unwrap(); + state.project_protocol_event_to_worker_state( + &detail.worker_ref, + &protocol::Event::WorkerState { + snapshot: running.clone(), + }, + ) + }); + assert_eq!( + runtime + .worker_detail(&detail.worker_ref) + .unwrap() + .worker_state, + Some(running.clone()) + ); + + assert!({ + let mut state = runtime.lock().unwrap(); + !state.project_protocol_event_to_worker_state( + &detail.worker_ref, + &protocol::Event::WorkerState { + snapshot: protocol::WorkerStateSnapshot { + revision: 2, + state: protocol::WorkerState::Idle, + ..running.clone() + }, + }, + ) + }); + assert!({ + let mut state = runtime.lock().unwrap(); + !state.project_protocol_event_to_worker_state( + &detail.worker_ref, + &protocol::Event::WorkerState { + snapshot: protocol::WorkerStateSnapshot { + state: protocol::WorkerState::Idle, + ..running.clone() + }, + }, + ) + }); + let after = runtime.worker_detail(&detail.worker_ref).unwrap(); + assert_eq!(after.status, WorkerStatus::Idle); + assert_eq!(after.worker_state, Some(running)); } #[test] @@ -5027,10 +5081,9 @@ mod tests { assert_eq!(*backend.restore_count.lock().unwrap(), 1); assert_eq!(*backend.run_generations.lock().unwrap(), vec![1, 2]); - assert_eq!( - runtime.worker_detail(&detail.worker_ref).unwrap().status, - WorkerStatus::Running - ); + let restored = runtime.worker_detail(&detail.worker_ref).unwrap(); + assert_eq!(restored.status, WorkerStatus::Idle); + assert_eq!(restored.worker_state, None); } #[test] diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index beeb4791..1655a387 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -17,7 +17,7 @@ use crate::ipc::notify_buffer::NotifyBuffer; use crate::ipc::server::SocketServer; use crate::runtime::dir::RuntimeDir; use crate::segment_log_sink::SegmentLogSink; -use crate::shared_state::WorkerSharedState; +use crate::shared_state::{WorkerCommandAdmission, WorkerSharedState}; use crate::shutdown_after_idle::{ ShutdownAfterIdleRequest, TicketIntakeReadyShutdownHook, is_ticket_intake_role, take_shutdown_request_after_status, @@ -181,21 +181,40 @@ impl WorkerHandle { } } +fn command_admission_disposition( + admission: WorkerCommandAdmission, +) -> Result<(), WorkerCommandDisposition> { + match admission { + WorkerCommandAdmission::Accepted => Ok(()), + WorkerCommandAdmission::Retry | WorkerCommandAdmission::StaleCommandId => { + Err(WorkerCommandDisposition::StaleCommandId) + } + WorkerCommandAdmission::Conflict => Err(WorkerCommandDisposition::Conflict), + WorkerCommandAdmission::ExecutionGenerationMismatch => { + Err(WorkerCommandDisposition::StaleExecutionGeneration) + } + WorkerCommandAdmission::StateRevisionMismatch => { + Err(WorkerCommandDisposition::StaleWorkerStateRevision) + } + } +} + fn validate_command( + envelope: WorkerCommandEnvelope, + kind: WorkerCommandKind, + shared_state: &WorkerSharedState, +) -> Result<(), WorkerCommandDisposition> { + command_admission_disposition(shared_state.admit_command(envelope, kind, true)) +} + +fn validate_shutdown_command( envelope: WorkerCommandEnvelope, shared_state: &WorkerSharedState, ) -> Result<(), WorkerCommandDisposition> { - let snapshot = shared_state.snapshot(); - if envelope.expected_execution_generation != snapshot.execution_generation { - return Err(WorkerCommandDisposition::StaleExecutionGeneration); + match shared_state.admit_command(envelope, WorkerCommandKind::Shutdown, false) { + WorkerCommandAdmission::Accepted | WorkerCommandAdmission::Retry => Ok(()), + admission => command_admission_disposition(admission), } - if envelope.expected_worker_state_revision != snapshot.revision { - return Err(WorkerCommandDisposition::StaleWorkerStateRevision); - } - if !shared_state.accept_command_id(envelope.command_id) { - return Err(WorkerCommandDisposition::StaleCommandId); - } - Ok(()) } fn acknowledge_command( @@ -205,6 +224,7 @@ fn acknowledge_command( command: WorkerCommandKind, disposition: WorkerCommandDisposition, ) { + shared_state.complete_command(command_id, command, disposition); let _ = working_event_tx.send(Event::CommandAcknowledged { acknowledgement: WorkerCommandAcknowledgement { command_id, @@ -1913,7 +1933,9 @@ async fn controller_loop( } } Method::Resume { command } => { - if let Err(disposition) = validate_command(command, &shared_state) { + if let Err(disposition) = + validate_command(command, WorkerCommandKind::Resume, &shared_state) + { acknowledge_command( &working_event_tx, &shared_state, @@ -1953,7 +1975,9 @@ async fn controller_loop( } Method::Cancel { command } => { - if let Err(disposition) = validate_command(command, &shared_state) { + if let Err(disposition) = + validate_command(command, WorkerCommandKind::Cancel, &shared_state) + { acknowledge_command( &working_event_tx, &shared_state, @@ -2017,7 +2041,9 @@ async fn controller_loop( } Method::Pause { command } => { - if let Err(disposition) = validate_command(command, &shared_state) { + if let Err(disposition) = + validate_command(command, WorkerCommandKind::Pause, &shared_state) + { acknowledge_command( &working_event_tx, &shared_state, @@ -2036,7 +2062,9 @@ async fn controller_loop( } Method::Compact { command } => { - if let Err(disposition) = validate_command(command, &shared_state) { + if let Err(disposition) = + validate_command(command, WorkerCommandKind::Compact, &shared_state) + { acknowledge_command( &working_event_tx, &shared_state, @@ -2081,7 +2109,11 @@ async fn controller_loop( method = method_rx.recv() => { match method { Some(Method::Cancel { command }) => { - if let Err(disposition) = validate_command(command, &shared_state) { + if let Err(disposition) = validate_command( + command, + WorkerCommandKind::Cancel, + &shared_state, + ) { acknowledge_command( &working_event_tx, &shared_state, @@ -2101,7 +2133,18 @@ async fn controller_loop( let _ = cancel_tx.send(true); } Some(Method::Shutdown { command }) => { - shared_state.accept_command_id(command.command_id); + if let Err(disposition) = + validate_shutdown_command(command, &shared_state) + { + acknowledge_command( + &working_event_tx, + &shared_state, + command.command_id, + WorkerCommandKind::Shutdown, + disposition, + ); + continue; + } shutdown_after_compaction = true; acknowledge_command( &working_event_tx, @@ -2196,9 +2239,18 @@ async fn controller_loop( }, Method::Shutdown { command } => { - // Shutdown remains unconditional/retryable even when the caller's - // live-state fence is stale. - shared_state.accept_command_id(command.command_id); + // Shutdown ignores the state-revision fence but remains bound to the + // current execution generation and command payload identity. + if let Err(disposition) = validate_shutdown_command(command, &shared_state) { + acknowledge_command( + &working_event_tx, + &shared_state, + command.command_id, + WorkerCommandKind::Shutdown, + disposition, + ); + continue; + } acknowledge_command( &working_event_tx, &shared_state, @@ -2544,7 +2596,9 @@ where method = method_rx.recv(), if input_commit.is_none() => { match method { Some(Method::Cancel { command }) => { - if let Err(disposition) = validate_command(command, shared_state) { + if let Err(disposition) = + validate_command(command, WorkerCommandKind::Cancel, shared_state) + { acknowledge_command( working_event_tx, shared_state, @@ -2583,7 +2637,9 @@ where let _ = cancel_tx.try_send(()); } Some(Method::Pause { command }) => { - if let Err(disposition) = validate_command(command, shared_state) { + if let Err(disposition) = + validate_command(command, WorkerCommandKind::Pause, shared_state) + { acknowledge_command( working_event_tx, shared_state, @@ -2623,7 +2679,16 @@ where let _ = pause_tx.try_send(()); } Some(Method::Shutdown { command }) => { - shared_state.accept_command_id(command.command_id); + if let Err(disposition) = validate_shutdown_command(command, shared_state) { + acknowledge_command( + working_event_tx, + shared_state, + command.command_id, + WorkerCommandKind::Shutdown, + disposition, + ); + continue; + } shutdown_requested = true; set_controller_state( shared_state, @@ -2705,7 +2770,9 @@ where } } Some(Method::Resume { command }) => { - if let Err(disposition) = validate_command(command, shared_state) { + if let Err(disposition) = + validate_command(command, WorkerCommandKind::Resume, shared_state) + { acknowledge_command( working_event_tx, shared_state, @@ -2763,7 +2830,9 @@ where } } Some(Method::Compact { command }) => { - if let Err(disposition) = validate_command(command, shared_state) { + if let Err(disposition) = + validate_command(command, WorkerCommandKind::Compact, shared_state) + { acknowledge_command( working_event_tx, shared_state, @@ -3677,6 +3746,7 @@ mod tests { expected_execution_generation: 8, expected_worker_state_revision: 0, }, + WorkerCommandKind::Pause, &shared, ), Err(WorkerCommandDisposition::StaleExecutionGeneration) @@ -3688,6 +3758,7 @@ mod tests { expected_execution_generation: 9, expected_worker_state_revision: 1, }, + WorkerCommandKind::Pause, &shared, ), Err(WorkerCommandDisposition::StaleWorkerStateRevision) @@ -3699,6 +3770,7 @@ mod tests { expected_execution_generation: 9, expected_worker_state_revision: 0, }, + WorkerCommandKind::Pause, &shared, ) .is_ok() @@ -3708,12 +3780,25 @@ mod tests { WorkerCommandEnvelope { command_id: 1, expected_execution_generation: 9, - expected_worker_state_revision: 1, + expected_worker_state_revision: 0, }, + WorkerCommandKind::Pause, &shared, ), Err(WorkerCommandDisposition::StaleCommandId) ); + assert_eq!( + validate_command( + WorkerCommandEnvelope { + command_id: 1, + expected_execution_generation: 9, + expected_worker_state_revision: 0, + }, + WorkerCommandKind::Cancel, + &shared, + ), + Err(WorkerCommandDisposition::Conflict) + ); assert!( validate_command( WorkerCommandEnvelope { @@ -3721,6 +3806,7 @@ mod tests { expected_execution_generation: 9, expected_worker_state_revision: 1, }, + WorkerCommandKind::Pause, &shared, ) .is_ok() diff --git a/crates/worker/src/shared_state.rs b/crates/worker/src/shared_state.rs index 573f2bb5..810b59f4 100644 --- a/crates/worker/src/shared_state.rs +++ b/crates/worker/src/shared_state.rs @@ -1,17 +1,37 @@ +use std::collections::VecDeque; use std::sync::{ OnceLock, RwLock, atomic::{AtomicBool, Ordering}, }; use protocol::{ - WorkerBusyState, WorkerMaintenanceState, WorkerRunState, WorkerState, WorkerStateSnapshot, - WorkerStatus, + WorkerBusyState, WorkerCommandDisposition, WorkerCommandEnvelope, WorkerCommandKind, + WorkerMaintenanceState, WorkerRunState, WorkerState, WorkerStateSnapshot, WorkerStatus, }; use serde_json::json; use session_store::SegmentId; use crate::fs_view::WorkerFsView; +const COMPLETED_COMMAND_RETENTION: usize = 256; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct AcceptedWorkerCommand { + envelope: WorkerCommandEnvelope, + kind: WorkerCommandKind, + disposition: Option, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum WorkerCommandAdmission { + Accepted, + Retry, + Conflict, + StaleCommandId, + ExecutionGenerationMismatch, + StateRevisionMismatch, +} + /// Shared state between WorkerController and runtime directory. /// /// `WorkerStateSnapshot` is the sole live execution-state authority. Runtime @@ -23,6 +43,7 @@ pub struct WorkerSharedState { pub manifest_toml: String, pub greeting: protocol::Greeting, state: RwLock, + accepted_commands: RwLock>, /// Worker-from-the-inside view of the filesystem. Set once in /// `WorkerController::start` after the local WorkdirSession provider is /// materialised, and read from the IPC server layer to answer @@ -55,6 +76,7 @@ impl WorkerSharedState { manifest_toml, greeting, state: RwLock::new(WorkerStateSnapshot::initial(execution_generation)), + accepted_commands: RwLock::new(VecDeque::new()), fs_view: OnceLock::new(), flow_transition_enabled: AtomicBool::new(false), } @@ -92,17 +114,99 @@ impl WorkerSharedState { snapshot.clone() } - pub fn accept_command_id(&self, command_id: u64) -> bool { + pub(crate) fn admit_command( + &self, + envelope: WorkerCommandEnvelope, + kind: WorkerCommandKind, + require_state_revision: bool, + ) -> WorkerCommandAdmission { let mut snapshot = self .state .write() .expect("worker state lock poisoned; refusing command admission"); - if command_id <= snapshot.last_command_id { - return false; + let mut accepted = self + .accepted_commands + .write() + .expect("worker command ledger lock poisoned; refusing command admission"); + if let Some(existing) = accepted + .iter() + .find(|accepted| accepted.envelope.command_id == envelope.command_id) + { + return if existing.envelope == envelope && existing.kind == kind { + WorkerCommandAdmission::Retry + } else { + WorkerCommandAdmission::Conflict + }; } - snapshot.last_command_id = command_id; + if envelope.expected_execution_generation != snapshot.execution_generation { + return WorkerCommandAdmission::ExecutionGenerationMismatch; + } + if require_state_revision && envelope.expected_worker_state_revision != snapshot.revision { + return WorkerCommandAdmission::StateRevisionMismatch; + } + if envelope.command_id <= snapshot.last_command_id { + return WorkerCommandAdmission::StaleCommandId; + } + + snapshot.last_command_id = envelope.command_id; snapshot.revision = snapshot.revision.saturating_add(1); - true + accepted.push_back(AcceptedWorkerCommand { + envelope, + kind, + disposition: None, + }); + WorkerCommandAdmission::Accepted + } + + pub(crate) fn complete_command( + &self, + command_id: u64, + kind: WorkerCommandKind, + disposition: WorkerCommandDisposition, + ) { + if !matches!( + disposition, + WorkerCommandDisposition::Accepted | WorkerCommandDisposition::InvalidState + ) { + return; + } + let mut accepted = self + .accepted_commands + .write() + .expect("worker command ledger lock poisoned; refusing command completion"); + if let Some(command) = accepted + .iter_mut() + .find(|command| command.envelope.command_id == command_id && command.kind == kind) + { + command.disposition.get_or_insert(disposition); + } + while accepted + .iter() + .filter(|command| command.disposition.is_some()) + .count() + > COMPLETED_COMMAND_RETENTION + { + let Some(index) = accepted + .iter() + .position(|command| command.disposition.is_some()) + else { + break; + }; + accepted.remove(index); + } + } + + #[cfg(test)] + pub(crate) fn command_result( + &self, + command_id: u64, + ) -> Option> { + self.accepted_commands + .read() + .expect("worker command ledger lock poisoned") + .iter() + .find(|command| command.envelope.command_id == command_id) + .map(|command| command.disposition) } pub fn snapshot(&self) -> WorkerStateSnapshot { @@ -190,9 +294,17 @@ mod tests { } #[test] - fn accepted_command_id_advances_the_snapshot_revision_atomically() { + fn accepted_command_identity_advances_revision_and_detects_reuse_conflicts() { let state = test_state(); - assert!(state.accept_command_id(9)); + let envelope = WorkerCommandEnvelope { + command_id: 9, + expected_execution_generation: 7, + expected_worker_state_revision: 0, + }; + assert_eq!( + state.admit_command(envelope, WorkerCommandKind::Pause, true), + WorkerCommandAdmission::Accepted + ); assert_eq!( state.snapshot(), WorkerStateSnapshot { @@ -202,7 +314,24 @@ mod tests { state: WorkerState::Idle, } ); - assert!(!state.accept_command_id(9)); + assert_eq!(state.command_result(9), Some(None)); + state.complete_command( + 9, + WorkerCommandKind::Pause, + WorkerCommandDisposition::Accepted, + ); + assert_eq!( + state.command_result(9), + Some(Some(WorkerCommandDisposition::Accepted)) + ); + assert_eq!( + state.admit_command(envelope, WorkerCommandKind::Pause, true), + WorkerCommandAdmission::Retry + ); + assert_eq!( + state.admit_command(envelope, WorkerCommandKind::Cancel, true), + WorkerCommandAdmission::Conflict + ); assert_eq!(state.snapshot().revision, 1); } diff --git a/crates/workspace-api/src/lib.rs b/crates/workspace-api/src/lib.rs index a2cf8f87..9bdd7bce 100644 --- a/crates/workspace-api/src/lib.rs +++ b/crates/workspace-api/src/lib.rs @@ -1568,7 +1568,12 @@ pub struct WorkerSummary { #[serde(default)] pub tags: Vec, pub workspace: WorkerWorkspaceSummary, + /// Runtime catalog lifecycle compatibility state. Live foreground state, when + /// available, is carried separately in `worker_state`. pub state: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + #[cfg_attr(feature = "typescript", ts(optional))] + pub worker_state: Option, pub last_seen_at: Option, #[serde(default)] pub pinned: bool, diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index f615685d..a6b065b2 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -243,7 +243,10 @@ pub struct WorkerSummary { #[serde(default)] pub tags: Vec, pub workspace: WorkerWorkspaceSummary, + /// Runtime catalog lifecycle compatibility state. pub state: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub worker_state: Option, pub last_seen_at: Option, #[serde(default)] pub pinned: bool, @@ -335,6 +338,7 @@ pub(crate) fn workspace_worker_summary( workspace_id: summary.workspace.workspace_id, }, state: summary.state, + worker_state: summary.worker_state, last_seen_at: summary.last_seen_at, pinned: summary.pinned, retention_state: summary.retention_state, @@ -1998,6 +2002,7 @@ impl EmbeddedWorkerRuntime { workspace_id: summary.workspace_id.clone(), }, state: embedded_worker_status_label(summary.status).to_string(), + worker_state: summary.worker_state.clone(), last_seen_at: None, pinned: false, retention_state: "transient".to_string(), @@ -2037,6 +2042,7 @@ impl EmbeddedWorkerRuntime { workspace_id: detail.workspace_id.clone(), }, state: embedded_worker_status_label(detail.status).to_string(), + worker_state: detail.worker_state.clone(), last_seen_at: None, pinned: false, retention_state: "transient".to_string(), @@ -3340,6 +3346,7 @@ impl RemoteWorkerRuntime { workspace_id: summary.workspace_id.clone(), }, state: embedded_worker_status_label(summary.status).to_string(), + worker_state: summary.worker_state.clone(), last_seen_at: None, pinned: false, retention_state: "transient".to_string(), @@ -3383,6 +3390,7 @@ impl RemoteWorkerRuntime { workspace_id: detail.workspace_id.clone(), }, state: embedded_worker_status_label(detail.status).to_string(), + worker_state: detail.worker_state.clone(), last_seen_at: None, pinned: false, retention_state: "transient".to_string(), @@ -4730,6 +4738,7 @@ pub fn placeholder_worker(host_id: impl Into) -> WorkerSummary { workspace_id: None, }, state: "unsupported".to_string(), + worker_state: None, last_seen_at: None, pinned: false, retention_state: "transient".to_string(), @@ -5251,6 +5260,7 @@ mod tests { workspace_id: None, }, state: "available".to_string(), + worker_state: None, last_seen_at: None, pinned: false, retention_state: "transient".to_string(), diff --git a/crates/workspace-server/src/runtime_subscription_tests.rs b/crates/workspace-server/src/runtime_subscription_tests.rs index 5c67f983..ab5d65ac 100644 --- a/crates/workspace-server/src/runtime_subscription_tests.rs +++ b/crates/workspace-server/src/runtime_subscription_tests.rs @@ -202,7 +202,16 @@ async fn equal_downstream_selectors_share_one_upstream_subscription() { BrokerSubscriptionEvent::Event { payload: SubscriptionEventPayload::WorkerUpserted { ref worker }, .. - } if worker.state == SubscriptionWorkerState::Running + } if worker.state == SubscriptionWorkerState::Idle + && matches!( + worker.worker_state, + Some(protocol::WorkerStateSnapshot { + state: protocol::WorkerState::Busy(protocol::WorkerBusyState::Run( + protocol::WorkerRunState::Running + )), + .. + }) + ) )); } let mut late = broker.subscribe("runtime-test", selector.clone()).unwrap(); @@ -212,7 +221,18 @@ async fn equal_downstream_selectors_share_one_upstream_subscription() { assert!(matches!( snapshot, SubscriptionSnapshot::Workers { workers } - if workers.iter().any(|worker| worker.state == SubscriptionWorkerState::Running) + if workers.iter().any(|worker| { + worker.state == SubscriptionWorkerState::Idle + && matches!( + worker.worker_state, + Some(protocol::WorkerStateSnapshot { + state: protocol::WorkerState::Busy( + protocol::WorkerBusyState::Run(protocol::WorkerRunState::Running) + ), + .. + }) + ) + }) )); drop(late); @@ -337,8 +357,18 @@ async fn embedded_runtime_uses_in_process_subscription_source() { ) .unwrap(); assert!(matches!(next_event(&mut subscription).await, - BrokerSubscriptionEvent::Event { payload: SubscriptionEventPayload::WorkerUpserted { worker }, .. } - if worker.runtime_id.as_deref() == Some("embedded-worker-runtime") && worker.state == SubscriptionWorkerState::Running)); + BrokerSubscriptionEvent::Event { payload: SubscriptionEventPayload::WorkerUpserted { worker }, .. } + if worker.runtime_id.as_deref() == Some("embedded-worker-runtime") + && worker.state == SubscriptionWorkerState::Idle + && matches!( + worker.worker_state, + Some(protocol::WorkerStateSnapshot { + state: protocol::WorkerState::Busy(protocol::WorkerBusyState::Run( + protocol::WorkerRunState::Running + )), + .. + }) + ))); let mut late = broker .subscribe( "embedded-worker-runtime", @@ -351,7 +381,18 @@ async fn embedded_runtime_uses_in_process_subscription_source() { assert!(matches!( snapshot, SubscriptionSnapshot::Workers { workers } - if workers.iter().any(|worker| worker.state == SubscriptionWorkerState::Running) + if workers.iter().any(|worker| { + worker.state == SubscriptionWorkerState::Idle + && matches!( + worker.worker_state, + Some(protocol::WorkerStateSnapshot { + state: protocol::WorkerState::Busy( + protocol::WorkerBusyState::Run(protocol::WorkerRunState::Running) + ), + .. + }) + ) + }) )); runtime diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 39206e5d..9cbeb899 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -15252,6 +15252,7 @@ fn worker_summary_from_registry(record: &WorkerRegistryRecord) -> WorkerSummary singleton_key: None, tags: Vec::new(), state: "missing".to_string(), + worker_state: None, last_seen_at: Some(record.updated_at.clone()), pinned: record.retention_state == "pinned", retention_state: record.retention_state.clone(), @@ -24990,6 +24991,7 @@ mod tests { workspace_id: Some(TEST_WORKSPACE_ID.to_string()), }, state: "idle".to_string(), + worker_state: None, last_seen_at: None, pinned: false, retention_state: "normal".to_string(), @@ -25086,6 +25088,7 @@ mod tests { workspace_id: Some(TEST_WORKSPACE_ID.to_string()), }, state: "idle".to_string(), + worker_state: None, last_seen_at: None, pinned: false, retention_state: "normal".to_string(), diff --git a/web/workspace/src/lib/generated/protocol.ts b/web/workspace/src/lib/generated/protocol.ts index c15fbdca..b3482873 100644 --- a/web/workspace/src/lib/generated/protocol.ts +++ b/web/workspace/src/lib/generated/protocol.ts @@ -19,7 +19,7 @@ command_id: number, expected_execution_generation: number, expected_worker_state export type WorkerCommandKind = "resume" | "cancel" | "pause" | "compact" | "shutdown"; -export type WorkerCommandDisposition = "accepted" | "stale_execution_generation" | "stale_worker_state_revision" | "stale_command_id" | "invalid_state"; +export type WorkerCommandDisposition = "accepted" | "stale_execution_generation" | "stale_worker_state_revision" | "stale_command_id" | "conflict" | "invalid_state"; export type WorkerCommandAcknowledgement = { command_id: number, command: WorkerCommandKind, disposition: WorkerCommandDisposition, /** @@ -233,7 +233,16 @@ resource_key?: string | null, /** * Producer-owned monotonic revision for this Worker subject. */ -subject_revision: number, state: SubscriptionWorkerState, has_running_internal_workers: boolean, workspace_id?: string | null, display_name?: string | null, profile?: string | null, +subject_revision: number, +/** + * Latest revisioned foreground state observed from the Worker. This remains + * absent until an authoritative Worker snapshot/event has been applied. + */ +worker_state?: WorkerStateSnapshot | null, +/** + * Runtime catalog lifecycle compatibility projection; not foreground-state authority. + */ +state: SubscriptionWorkerState, has_running_internal_workers: boolean, workspace_id?: string | null, display_name?: string | null, profile?: string | null, /** * Workspace-facing Repository key. Runtime producers leave this unset and * Workspace Server projections replace `repository_id` with this field. diff --git a/web/workspace/src/lib/workspace/sidebar/types.ts b/web/workspace/src/lib/workspace/sidebar/types.ts index 67d7aa10..856076fd 100644 --- a/web/workspace/src/lib/workspace/sidebar/types.ts +++ b/web/workspace/src/lib/workspace/sidebar/types.ts @@ -19,6 +19,7 @@ import type { Event as PodProtocolEvent, Method as PodProtocolMethod, Segment as PodProtocolSegment, + WorkerStateSnapshot, } from "$lib/generated/protocol"; import type { GitCommitSummary as SharedGitCommitSummary, @@ -99,6 +100,7 @@ export type Worker = { tags: string[]; workspace: { visibility: string; identity: string }; state: string; + worker_state?: WorkerStateSnapshot | null; pinned?: boolean; retention_state?: string; last_seen_at?: string | null; diff --git a/web/workspace/src/lib/workspace/sidebar/worker-state.ts b/web/workspace/src/lib/workspace/sidebar/worker-state.ts new file mode 100644 index 00000000..0f45e9f3 --- /dev/null +++ b/web/workspace/src/lib/workspace/sidebar/worker-state.ts @@ -0,0 +1,12 @@ +import type { WorkerStateSnapshot } from "$lib/generated/protocol"; + +export function liveWorkerState(worker: { + state: string; + worker_state?: WorkerStateSnapshot | null; +}): string { + const state = worker.worker_state?.state; + if (!state) return worker.state === "stopped" ? "stopped" : "unknown"; + if (state.kind === "idle") return "idle"; + if (state.state.kind === "maintenance") return "running"; + return state.state.state === "paused" ? "paused" : "running"; +} diff --git a/web/workspace/src/lib/workspace/sidebar/worker-subscription.test.ts b/web/workspace/src/lib/workspace/sidebar/worker-subscription.test.ts index 279afa0d..84e81598 100644 --- a/web/workspace/src/lib/workspace/sidebar/worker-subscription.test.ts +++ b/web/workspace/src/lib/workspace/sidebar/worker-subscription.test.ts @@ -5,6 +5,7 @@ function assertEquals(actual: unknown, expected: unknown): void { throw new Error(`expected ${JSON.stringify(expected)}, got ${JSON.stringify(actual)}`); } } +import { liveWorkerState } from './worker-state'; import { applyWorkspaceWorkersFrame, createWorkspaceWorkersProjection, @@ -33,6 +34,22 @@ function worker( }; } +Deno.test('Worker list state uses the authoritative live snapshot separately from lifecycle', () => { + const active = worker('runtime-a', 'worker-1', 1); + active.worker_state = { + execution_generation: 4, + revision: 2, + last_command_id: 1, + state: { kind: 'busy', state: { kind: 'run', state: 'paused' } }, + }; + assertEquals(liveWorkerState(active), 'paused'); + + const unavailable = worker('runtime-a', 'worker-2', 1); + assertEquals(liveWorkerState(unavailable), 'unknown'); + unavailable.state = 'stopped'; + assertEquals(liveWorkerState(unavailable), 'stopped'); +}); + Deno.test('workspace Worker snapshot keeps equal local ids from different Runtimes', () => { const projection = createWorkspaceWorkersProjection(); const frame: SubscriptionFrame = { diff --git a/web/workspace/src/lib/workspace/sidebar/worker-subscription.ts b/web/workspace/src/lib/workspace/sidebar/worker-subscription.ts index a1eed656..8c9b7557 100644 --- a/web/workspace/src/lib/workspace/sidebar/worker-subscription.ts +++ b/web/workspace/src/lib/workspace/sidebar/worker-subscription.ts @@ -5,6 +5,7 @@ import { applyWorkspaceWorkersFrame, createWorkspaceWorkersProjection, } from './worker-subscription-model'; +import { liveWorkerState } from './worker-state'; import { compareWorkersForSidebar } from './workers'; import type { Worker } from './types'; @@ -86,7 +87,8 @@ function projectWorker(worker: SubscriptionWorker): SidebarWorker { profile: worker.profile ?? null, tags: [], workspace: { visibility: 'workspace', identity: 'runtime_subscription_worker' }, - state: worker.state, + state: liveWorkerState(worker), + worker_state: worker.worker_state, pinned: false, retention_state: 'transient', implementation: { diff --git a/web/workspace/src/routes/w/[workspaceId]/runtimes/[runtimeId]/workers/[workerId]/console/+page.svelte b/web/workspace/src/routes/w/[workspaceId]/runtimes/[runtimeId]/workers/[workerId]/console/+page.svelte index cf28f294..ed8d1df4 100644 --- a/web/workspace/src/routes/w/[workspaceId]/runtimes/[runtimeId]/workers/[workerId]/console/+page.svelte +++ b/web/workspace/src/routes/w/[workspaceId]/runtimes/[runtimeId]/workers/[workerId]/console/+page.svelte @@ -52,11 +52,7 @@ import { pushWorkspaceAlert } from "$lib/workspace/alerts/store"; import { workspaceApiPath } from "$lib/workspace/api/http"; import { workspaceMultiplexer, type WorkspaceMultiplexerSubscription } from "$lib/workspace/multiplexer"; - import type { - Diagnostic, - Worker, - PodProtocolEvent, - } from "$lib/workspace/sidebar/types"; + import type { Diagnostic, Worker } from "$lib/workspace/sidebar/types"; type Props = { data: { @@ -207,7 +203,6 @@ ); let pendingObservationEvents: ConsoleEventInput[] = []; let protocolEventSequence = 0; - let pendingObservedStates: Array = []; let pendingStreamDiagnostics: Diagnostic[] = []; let observationFlushHandle: number | null = null; let nextReloadToken = 0; @@ -249,7 +244,9 @@ const diagnostics = $derived( mergeDiagnostics(worker?.diagnostics ?? [], streamDiagnostics), ); - const workerState = $derived(liveWorkerState ?? worker?.state ?? "loading"); + const workerState = $derived( + liveWorkerState ?? (worker?.state === "stopped" ? "stopped" : "loading"), + ); const workerRunning = $derived(workerState === "running"); const workerPaused = $derived(workerState === "paused"); const composerEditable = $derived(protocolState === "open" && !sending); @@ -343,7 +340,6 @@ observationFlushHandle = null; } pendingObservationEvents = []; - pendingObservedStates = []; pendingStreamDiagnostics = []; } @@ -359,18 +355,15 @@ function flushObservationBatch() { observationFlushHandle = null; const eventBatch = pendingObservationEvents; - const stateBatch = pendingObservedStates; const diagnosticBatch = pendingStreamDiagnostics; pendingObservationEvents = []; - pendingObservedStates = []; pendingStreamDiagnostics = []; if (eventBatch.length > 0) { - const latestState = stateBatch.findLast((state) => state !== null); - if (latestState) { - liveWorkerState = latestState; - } consoleProjection = consoleProjector.append(eventBatch); + liveWorkerState = consoleProjection.status === "shutdown" + ? "shutdown" + : workerStateFromSnapshot(consoleProjection.workerState); advanceEventObservedAtVersion(); } @@ -407,7 +400,6 @@ event: payload, observedAtMs, }); - pendingObservedStates.push(workerStateFromProtocolEvent(payload)); scheduleObservationFlush(); } @@ -925,36 +917,16 @@ handleComposerSubmit(); } - function workerStateFromProtocolEvent( - event: PodProtocolEvent, + function workerStateFromSnapshot( + snapshot: ConsoleProjection["workerState"], ): string | null { - switch (event.event) { - case "snapshot": - return event.data.state.state.kind === "idle" - ? "idle" - : event.data.state.state.state.kind === "run" && - event.data.state.state.state.state === "paused" - ? "paused" - : "running"; - case "worker_state": - return event.data.snapshot.state.kind === "idle" - ? "idle" - : event.data.snapshot.state.state.kind === "run" && - event.data.snapshot.state.state.state === "paused" - ? "paused" - : "running"; - case "command_acknowledged": - return event.data.acknowledgement.state.state.kind === "idle" - ? "idle" - : event.data.acknowledgement.state.state.state.kind === "run" && - event.data.acknowledgement.state.state.state.state === "paused" - ? "paused" - : "running"; - case "shutdown": - return "shutdown"; - default: - return null; - } + if (!snapshot) return null; + return snapshot.state.kind === "idle" + ? "idle" + : snapshot.state.state.kind === "run" && + snapshot.state.state.state === "paused" + ? "paused" + : "running"; } function connectProtocolTransport( diff --git a/web/workspace/src/routes/w/[workspaceId]/workers/+page.svelte b/web/workspace/src/routes/w/[workspaceId]/workers/+page.svelte index e734def2..e3a0d542 100644 --- a/web/workspace/src/routes/w/[workspaceId]/workers/+page.svelte +++ b/web/workspace/src/routes/w/[workspaceId]/workers/+page.svelte @@ -4,6 +4,7 @@ import { workerHref } from '$lib/workspace/resource-links'; import { formatCurrentWorkdirRevision } from '$lib/workspace/settings/workdir-revision'; import { canOpenWorkerConsole } from '$lib/workspace/sidebar/workers'; + import { liveWorkerState } from '$lib/workspace/sidebar/worker-state'; import type { CleanupWorkerCandidate, RuntimeCleanupExecutionResponse, RuntimeCleanupPlanResponse, Worker } from '$lib/workspace/sidebar/types'; import type { PageProps } from './$types'; @@ -136,7 +137,7 @@ } function workerStatus(worker: Worker): string { - return worker.state; + return liveWorkerState(worker); } function workerProfile(worker: Worker): string { diff --git a/web/workspace/src/routes/w/[workspaceId]/workers/[workerRef]/+page.svelte b/web/workspace/src/routes/w/[workspaceId]/workers/[workerRef]/+page.svelte index ef3d04bb..15d3ffe4 100644 --- a/web/workspace/src/routes/w/[workspaceId]/workers/[workerRef]/+page.svelte +++ b/web/workspace/src/routes/w/[workspaceId]/workers/[workerRef]/+page.svelte @@ -1,5 +1,6 @@ @@ -24,7 +25,7 @@ >Open console
-
Status
{data.worker.state}
+
Status
{liveWorkerState(data.worker)}
Profile
{data.worker.profile}
Internal ID
{data.worker.worker_id}