fix: restore backend workers from resume

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<BackendWorkerRestoreResponse, io::Error>;
+
+    async fn observe(
+        &self,
+        target: &client::BackendRuntimeTarget,
+    ) -> Result<WorkspaceWorkerSessionResponse, io::Error>;
+}
+
+struct LiveBackendWorkerLifecycle;
+
+#[async_trait::async_trait]
+impl BackendWorkerLifecycle for LiveBackendWorkerLifecycle {
+    async fn restore(
+        &self,
+        target: &client::BackendRuntimeTarget,
+    ) -> Result<BackendWorkerRestoreResponse, io::Error> {
+        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<WorkspaceWorkerSessionResponse, io::Error> {
+        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<dyn Error>> {
     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<client::BackendRuntimeTarget, io::Error> {
+    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::<String>();
+            format!("{}: {message}", diagnostic.code)
+        })
+        .collect::<Vec<_>>()
+        .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::<String>();
+            Err(io::Error::other(format!(
+                "Backend Worker restore completed without a live protocol Session: {reason:?}: {message}"
+            )))
+        }
+    }
+}
+
 fn dedup_workers(workers: &mut Vec<BackendWorkerSummary>) {
     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<Vec<&'static str>>,
+    }
+
+    #[async_trait::async_trait]
+    impl BackendWorkerLifecycle for FakeBackendWorkerLifecycle {
+        async fn restore(
+            &self,
+            _target: &client::BackendRuntimeTarget,
+        ) -> Result<BackendWorkerRestoreResponse, io::Error> {
+            self.calls.lock().unwrap().push("restore");
+            Ok(self.restore_response.clone())
+        }
+
+        async fn observe(
+            &self,
+            _target: &client::BackendRuntimeTarget,
+        ) -> Result<WorkspaceWorkerSessionResponse, io::Error> {
+            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(

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<String>,
-        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<dyn std::error::Error>),
         },
-        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<dyn std::error::Error>),
             }
-            Err(e) => Err(Box::new(e) as Box<dyn std::error::Error>),
-        },
+        }
         LaunchMode::OpenWorker {
             runtime_id,
             worker_id,

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<R: CliConnectionResolver + ?Sized>(
             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<R: CliConnectionResolver + ?Sized>(
         target,
         mode: LaunchMode::Workers {
             runtime_id,
-            include_stopped,
+            intent: BackendWorkerPickerIntent::Attach { include_stopped },
         },
         workspace_root,
     })
@@ -1272,7 +1274,7 @@ fn parse_resume_args<R: CliConnectionResolver + ?Sized>(
         }
         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() {