From ac7a6027bde11a71471df4eb9b1be599090372dd Mon Sep 17 00:00:00 2001 From: Hare Date: Mon, 28 Sep 2026 04:28:35 +0900 Subject: [PATCH] fix: restore backend workers from resume --- crates/tui/src/backend_worker_picker.rs | 313 ++++++++++++++++++++++-- crates/tui/src/lib.rs | 43 ++-- crates/yoi/src/main.rs | 64 ++++- 3 files changed, 385 insertions(+), 35 deletions(-) diff --git a/crates/tui/src/backend_worker_picker.rs b/crates/tui/src/backend_worker_picker.rs index 547089b19..df62a759b 100644 --- a/crates/tui/src/backend_worker_picker.rs +++ b/crates/tui/src/backend_worker_picker.rs @@ -3,9 +3,10 @@ use std::io; use std::time::Duration; use client::{ - BackendRuntimeListTarget, BackendWorkerSummary, BackendWorkingDirectorySource, - WorkerSessionAvailability, list_backend_stopped_workers, list_backend_workers, - observe_backend_worker_session, + BackendRuntimeListTarget, BackendWorkerRestoreResponse, BackendWorkerRestoreState, + BackendWorkerSummary, BackendWorkingDirectorySource, WorkerSessionAvailability, + WorkspaceWorkerSessionResponse, list_backend_stopped_workers, list_backend_workers, + observe_backend_worker_session, restore_backend_worker, }; use crossterm::event::{self, Event as TermEvent, KeyCode, KeyEventKind, KeyModifiers}; use ratatui::Frame; @@ -15,6 +16,7 @@ use ratatui::text::{Line, Span}; use ratatui::widgets::Paragraph; use unicode_width::UnicodeWidthStr; +use crate::BackendWorkerPickerIntent; use crate::backend_workspace_picker::select_backend_workspace; use crate::console; use crate::inline_terminal::with_inline_terminal; @@ -22,9 +24,53 @@ use crate::inline_terminal::with_inline_terminal; const MAX_ROWS: usize = 10; const VIEWPORT_LINES: u16 = MAX_ROWS as u16 + 4; +#[async_trait::async_trait] +trait BackendWorkerLifecycle { + async fn restore( + &self, + target: &client::BackendRuntimeTarget, + ) -> Result; + + async fn observe( + &self, + target: &client::BackendRuntimeTarget, + ) -> Result; +} + +struct LiveBackendWorkerLifecycle; + +#[async_trait::async_trait] +impl BackendWorkerLifecycle for LiveBackendWorkerLifecycle { + async fn restore( + &self, + target: &client::BackendRuntimeTarget, + ) -> Result { + restore_backend_worker(target).await.map_err(|error| { + io::Error::other(format!( + "failed to restore Backend Worker {}: {error}", + target.display_label() + )) + }) + } + + async fn observe( + &self, + target: &client::BackendRuntimeTarget, + ) -> Result { + observe_backend_worker_session(target) + .await + .map_err(|error| { + io::Error::other(format!( + "failed to observe Worker Session {}: {error}", + target.display_label() + )) + }) + } +} + pub(crate) async fn run( mut target: BackendRuntimeListTarget, - include_stopped: bool, + intent: BackendWorkerPickerIntent, ) -> Result<(), Box> { loop { if target.workspace_id().is_none() { @@ -40,7 +86,10 @@ pub(crate) async fn run( target.base_url )) })?; - if include_stopped { + if intent == BackendWorkerPickerIntent::Resume { + response.items.retain(is_resume_candidate); + } + if intent.include_stopped() { match list_backend_stopped_workers(&target).await { Ok(stopped) => { response.items.extend(stopped.items); @@ -81,22 +130,80 @@ pub(crate) async fn run( } WorkerPickerResult::Selected(selected) => selected, }; - let mut attach_target = target + let attach_target = target .runtime_target(selected.runtime_id.clone(), selected.worker_id.clone()) .map_err(|error| io::Error::other(error.to_string()))?; - let response = observe_backend_worker_session(&attach_target) - .await - .map_err(|error| { - io::Error::other(format!( - "failed to observe Worker Session {}/{}: {error}", - selected.runtime_id, selected.worker_id - )) - })?; - apply_worker_session_observation(&mut attach_target, response.observation); + let attach_target = + prepare_selected_worker(&LiveBackendWorkerLifecycle, attach_target, intent).await?; return console::run_backend_runtime(attach_target).await; } } +async fn prepare_selected_worker( + lifecycle: &impl BackendWorkerLifecycle, + mut target: client::BackendRuntimeTarget, + intent: BackendWorkerPickerIntent, +) -> Result { + if intent == BackendWorkerPickerIntent::Resume { + let response = lifecycle.restore(&target).await?; + ensure_restore_accepted(&response)?; + let observation = lifecycle.observe(&target).await?; + require_live_resume_observation(observation.observation)?; + return Ok(target); + } + + let observation = lifecycle.observe(&target).await?; + apply_worker_session_observation(&mut target, observation.observation); + Ok(target) +} + +fn is_resume_candidate(worker: &BackendWorkerSummary) -> bool { + worker.worker_state.is_none() +} + +fn ensure_restore_accepted(response: &BackendWorkerRestoreResponse) -> Result<(), io::Error> { + if response.result.state == BackendWorkerRestoreState::Accepted { + return Ok(()); + } + let detail = response + .result + .diagnostics + .iter() + .take(3) + .map(|diagnostic| { + let message = diagnostic.message.chars().take(512).collect::(); + format!("{}: {message}", diagnostic.code) + }) + .collect::>() + .join("; "); + let detail = if detail.is_empty() { + "no Backend diagnostics".to_string() + } else { + detail + }; + Err(io::Error::other(format!( + "Backend Worker restore for {}/{} returned {:?} ({detail})", + response.runtime_id, response.worker_id, response.result.state + ))) +} + +fn require_live_resume_observation( + observation: WorkerSessionAvailability, +) -> Result<(), io::Error> { + match observation { + WorkerSessionAvailability::LiveProtocol => Ok(()), + WorkerSessionAvailability::RetainedSnapshot { .. } => Err(io::Error::other( + "Backend Worker restore completed without a live protocol Session; retained history was not opened because resume requires a live Worker", + )), + WorkerSessionAvailability::Unavailable { reason, message } => { + let message = message.chars().take(512).collect::(); + Err(io::Error::other(format!( + "Backend Worker restore completed without a live protocol Session: {reason:?}: {message}" + ))) + } + } +} + fn dedup_workers(workers: &mut Vec) { let mut seen = std::collections::HashSet::new(); workers.retain(|worker| seen.insert((worker.runtime_id.clone(), worker.worker_id.clone()))); @@ -432,7 +539,8 @@ fn working_directory_text(worker: &BackendWorkerSummary) -> String { mod tests { use super::*; use client::{ - BackendWorkerCapabilitySummary, BackendWorkerImplementationSummary, + BackendDiagnostic, BackendDiagnosticSeverity, BackendWorkerCapabilitySummary, + BackendWorkerImplementationSummary, BackendWorkerRestoreResult, BackendWorkerWorkspaceSummary, }; @@ -453,6 +561,7 @@ mod tests { workspace_id: Some("ws".to_string()), }, state: "idle".to_string(), + availability: protocol::subscription::SubscriptionWorkerAvailability::Observed, worker_state: Some(protocol::WorkerStateSnapshot { state: protocol::WorkerState::Busy(protocol::WorkerBusyState::Run( protocol::WorkerRunState::Running, @@ -488,6 +597,178 @@ mod tests { text_width(&text[..byte_offset]) } + fn restore_response(state: BackendWorkerRestoreState) -> BackendWorkerRestoreResponse { + BackendWorkerRestoreResponse { + workspace_id: "workspace-a".to_string(), + runtime_id: "runtime-a".to_string(), + worker_id: "worker-a".to_string(), + result: BackendWorkerRestoreResult { + state, + worker: None, + diagnostics: Vec::new(), + }, + } + } + + struct FakeBackendWorkerLifecycle { + restore_response: BackendWorkerRestoreResponse, + observation: WorkspaceWorkerSessionResponse, + calls: std::sync::Mutex>, + } + + #[async_trait::async_trait] + impl BackendWorkerLifecycle for FakeBackendWorkerLifecycle { + async fn restore( + &self, + _target: &client::BackendRuntimeTarget, + ) -> Result { + self.calls.lock().unwrap().push("restore"); + Ok(self.restore_response.clone()) + } + + async fn observe( + &self, + _target: &client::BackendRuntimeTarget, + ) -> Result { + self.calls.lock().unwrap().push("observe"); + Ok(self.observation.clone()) + } + } + + fn session_observation(observation: serde_json::Value) -> WorkspaceWorkerSessionResponse { + let mut value = serde_json::json!({ + "subject": { + "kind": "runtime_worker", + "runtime_id": "runtime-a", + "worker_id": "worker-a" + } + }); + value.as_object_mut().unwrap().extend( + observation + .as_object() + .expect("observation must be an object") + .clone(), + ); + serde_json::from_value(value).unwrap() + } + + #[tokio::test] + async fn resume_restores_before_requiring_live_observation() { + let lifecycle = FakeBackendWorkerLifecycle { + restore_response: restore_response(BackendWorkerRestoreState::Accepted), + observation: session_observation(serde_json::json!({ + "availability": "live_protocol" + })), + calls: std::sync::Mutex::new(Vec::new()), + }; + let target = client::BackendRuntimeTarget::new( + "http://127.0.0.1:3000", + "workspace-a", + "runtime-a", + "worker-a", + ); + + let prepared = + prepare_selected_worker(&lifecycle, target, BackendWorkerPickerIntent::Resume) + .await + .unwrap(); + + assert_eq!(*lifecycle.calls.lock().unwrap(), ["restore", "observe"]); + assert!(prepared.initial_snapshot.is_none()); + } + + #[tokio::test] + async fn rejected_resume_does_not_observe_or_open_retained_state() { + let lifecycle = FakeBackendWorkerLifecycle { + restore_response: restore_response(BackendWorkerRestoreState::Rejected), + observation: session_observation(serde_json::json!({ + "availability": "unavailable", + "reason": "retention_missing", + "message": "must not be opened" + })), + calls: std::sync::Mutex::new(Vec::new()), + }; + let target = client::BackendRuntimeTarget::new( + "http://127.0.0.1:3000", + "workspace-a", + "runtime-a", + "worker-a", + ); + + let error = prepare_selected_worker(&lifecycle, target, BackendWorkerPickerIntent::Resume) + .await + .unwrap_err() + .to_string(); + + assert!(error.contains("Rejected")); + assert_eq!(*lifecycle.calls.lock().unwrap(), ["restore"]); + } + + #[test] + fn resume_candidates_have_no_current_worker_state() { + let live = worker("runtime-a", "worker-live", None); + let mut stopped = worker("runtime-a", "worker-stopped", None); + stopped.state = "stopped".to_string(); + stopped.worker_state = None; + let mut unknown = worker("runtime-a", "worker-unknown", None); + unknown.worker_state = None; + + assert!(!is_resume_candidate(&live)); + assert!(is_resume_candidate(&stopped)); + assert!(is_resume_candidate(&unknown)); + } + + #[test] + fn backend_resume_requires_an_accepted_restore_result() { + assert!( + ensure_restore_accepted(&restore_response(BackendWorkerRestoreState::Accepted)).is_ok() + ); + for state in [ + BackendWorkerRestoreState::Rejected, + BackendWorkerRestoreState::RolledBack, + BackendWorkerRestoreState::ReconciliationRequired, + ] { + let mut response = restore_response(state); + response.result.diagnostics.push(BackendDiagnostic { + code: "restore_failed".to_string(), + severity: BackendDiagnosticSeverity::Error, + message: "bounded reason".to_string(), + }); + let error = ensure_restore_accepted(&response).unwrap_err().to_string(); + assert!(error.contains("restore_failed: bounded reason")); + } + } + + #[test] + fn backend_resume_requires_live_protocol_after_restore() { + assert!(require_live_resume_observation(WorkerSessionAvailability::LiveProtocol).is_ok()); + + let retained = + require_live_resume_observation(WorkerSessionAvailability::RetainedSnapshot { + identity: client::RetainedSessionIdentity { + session_id: "session-a".to_string(), + segment_id: "segment-a".to_string(), + entry_count: 1, + }, + snapshot: protocol::SessionSnapshot { + pending_submissions: protocol::PendingSubmissionsSnapshot::default(), + entries: Vec::new(), + }, + }) + .unwrap_err() + .to_string(); + assert!(retained.contains("resume requires a live Worker")); + + let unavailable = require_live_resume_observation(WorkerSessionAvailability::Unavailable { + reason: client::WorkerSessionUnavailableReason::RetentionMissing, + message: "retained state missing".to_string(), + }) + .unwrap_err() + .to_string(); + assert!(unavailable.contains("RetentionMissing")); + assert!(unavailable.contains("retained state missing")); + } + #[test] fn authoritative_observation_selects_live_retained_and_unavailable_detail_modes() { let mut live = client::BackendRuntimeTarget::new( diff --git a/crates/tui/src/lib.rs b/crates/tui/src/lib.rs index c9c48bd8e..5bc9ca090 100644 --- a/crates/tui/src/lib.rs +++ b/crates/tui/src/lib.rs @@ -42,6 +42,21 @@ pub struct LaunchOptions { pub workspace_root: PathBuf, } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum BackendWorkerPickerIntent { + Attach { include_stopped: bool }, + Resume, +} + +impl BackendWorkerPickerIntent { + pub(crate) fn include_stopped(self) -> bool { + match self { + Self::Attach { include_stopped } => include_stopped, + Self::Resume => true, + } + } +} + #[derive(Debug, Clone)] pub enum LaunchMode { /// Start one client-owned in-process Standalone Worker. @@ -54,10 +69,10 @@ pub enum LaunchMode { StandaloneResume { include_all: bool }, /// Create one Backend Worker and attach to it. BackendSpawn, - /// List Backend Workers and attach to the selected Worker. + /// List Backend Workers and open the selected Worker according to the caller's intent. Workers { runtime_id: Option, - include_stopped: bool, + intent: BackendWorkerPickerIntent, }, /// Open one Backend Worker through the selected connection target. OpenWorker { @@ -212,20 +227,18 @@ async fn launch_mode( Ok(launch) => backend_spawn::run(launch.target).await, Err(e) => Err(Box::new(e) as Box), }, - LaunchMode::Workers { - runtime_id, - include_stopped, - } => match target.list_workers(if include_stopped { - WorkerListRequest::with_stopped(runtime_id) - } else { - WorkerListRequest::new(runtime_id) - }) { - Ok(worker_list) => { - backend_worker_picker::run(worker_list.backend_target, worker_list.include_stopped) - .await + LaunchMode::Workers { runtime_id, intent } => { + match target.list_workers(if intent.include_stopped() { + WorkerListRequest::with_stopped(runtime_id) + } else { + WorkerListRequest::new(runtime_id) + }) { + Ok(worker_list) => { + backend_worker_picker::run(worker_list.backend_target, intent).await + } + Err(e) => Err(Box::new(e) as Box), } - Err(e) => Err(Box::new(e) as Box), - }, + } LaunchMode::OpenWorker { runtime_id, worker_id, diff --git a/crates/yoi/src/main.rs b/crates/yoi/src/main.rs index b7e4e888e..b46b54a6d 100644 --- a/crates/yoi/src/main.rs +++ b/crates/yoi/src/main.rs @@ -23,7 +23,7 @@ use cli_connection::{ }; use client::{BackendAuthTarget, Target, TargetKind, start_device_login, wait_for_device_login}; use serde::Deserialize; -use tui::{LaunchMode, LaunchOptions}; +use tui::{BackendWorkerPickerIntent, LaunchMode, LaunchOptions}; use workspace_bootstrap::{ InitOptions, discover_repository_root, run_init, select_backend_workspace_for_repository, }; @@ -1028,7 +1028,9 @@ fn parse_console_options( target, mode: LaunchMode::Workers { runtime_id, - include_stopped: false, + intent: BackendWorkerPickerIntent::Attach { + include_stopped: false, + }, }, workspace_root, }); @@ -1154,7 +1156,7 @@ fn parse_workers_args( target, mode: LaunchMode::Workers { runtime_id, - include_stopped, + intent: BackendWorkerPickerIntent::Attach { include_stopped }, }, workspace_root, }) @@ -1272,7 +1274,7 @@ fn parse_resume_args( } LaunchMode::Workers { runtime_id, - include_stopped: true, + intent: BackendWorkerPickerIntent::Resume, } }; @@ -2614,6 +2616,60 @@ backend = "shared" } } + #[test] + fn parse_backend_resume_uses_restore_picker_intent() { + match parse_args_from([ + "--backend", + "http://127.0.0.1:8787", + "--workspace-id", + "workspace-a", + "resume", + "--runtime-id", + "arcadia", + ]) + .unwrap() + { + Mode::Tui { + mode: + LaunchMode::Workers { + runtime_id, + intent: BackendWorkerPickerIntent::Resume, + }, + .. + } => assert_eq!(runtime_id.as_deref(), Some("arcadia")), + other => panic!("expected Backend resume picker, got {other:?}"), + } + } + + #[test] + fn parse_workers_stopped_keeps_retained_attach_intent() { + match parse_args_from([ + "--backend", + "http://127.0.0.1:8787", + "--workspace-id", + "workspace-a", + "workers", + "--stopped", + "--runtime-id", + "arcadia", + ]) + .unwrap() + { + Mode::Tui { + mode: + LaunchMode::Workers { + runtime_id, + intent: BackendWorkerPickerIntent::Attach { include_stopped }, + }, + .. + } => { + assert_eq!(runtime_id.as_deref(), Some("arcadia")); + assert!(include_stopped); + } + other => panic!("expected retained Backend worker picker, got {other:?}"), + } + } + #[test] fn parse_resume_subcommand_uses_standalone_store_by_default() { match parse_args_from(["resume"]).unwrap() {