diff --git a/architecture/compute-runtimes.md b/architecture/compute-runtimes.md index f122fda5d..d02ee72bc 100644 --- a/architecture/compute-runtimes.md +++ b/architecture/compute-runtimes.md @@ -16,12 +16,60 @@ Each runtime receives a sandbox spec from the gateway and is responsible for: - Reporting lifecycle and platform events back to the gateway. - Cleaning up runtime-owned resources. +Drivers report **backend state only**. A driver snapshot with `Ready=True` means +the underlying compute resource (container, pod, VM) is healthy and running — +nothing more. Drivers must not gate on supervisor session state or hold +references to gateway-internal types. The gateway owns the public +`SandboxPhase::Ready` decision. This applies equally to extension drivers +implementing `ComputeDriver` out of tree. + Drivers own runtime-specific platform event interpretation. When an event should drive client provisioning UI, the driver attaches the shared `openshell.progress.*` metadata defined in `openshell-core` instead of requiring clients to parse Kubernetes reasons, VM cache states, or other driver-local reason strings. +## Sandbox Readiness Composition + +The gateway composes driver backend state with supervisor session presence to +produce the public `SandboxPhase`. This composition is gateway-owned and applied +uniformly across all drivers: + +``` +backend_phase = derive_phase(driver_status) + +public_phase = + if backend_phase in {Error, Deleting}: → pass through (terminal precedence) + if backend_phase == Ready && session connected: → Ready + if backend_phase == Ready && no session: → Provisioning + if backend_phase in {Provisioning, Unknown} && session: → Ready + if backend_phase in {Provisioning, Unknown} && no session: → Provisioning +``` + +When `public_phase == Ready` the sandbox is usable through the gateway — both the +backend resource is healthy and a supervisor session is registered. A sandbox whose +backend reports ready but has no supervisor session yet holds `Provisioning`; the +driver's `BackendReady=True` condition is visible in the sandbox status for operators +who need to distinguish that state from a sandbox still provisioning its compute resource. + +**Session precedence over lagging driver snapshots:** A supervisor session can only be +established by a running workload. When `set_supervisor_session_state` promotes the +store record to `Ready` on session connect, a driver watch event may still arrive +shortly after carrying a stale `Provisioning` or `Unknown` backend phase. The +composition rule treats a connected session as the stronger signal and keeps `Ready` +in that case, preventing a lagging snapshot from undoing the session-driven promotion. + +**HA deployments:** Supervisor sessions are process-local. A gateway replica that +does not own the active supervisor session holds the public phase at `Provisioning`. +The owning replica's `supervisor_session_connected` write propagates through the +shared store and reconcile loop. This is correct behavior — a replica should not +claim `Ready` for a session it does not hold. + +**Extension point:** The readiness decision is a safety invariant, not an +operator-configurable hook. The driver contract is the correct extension point for +custom backend readiness semantics. RFC-0010 lifecycle hooks may observe readiness +transitions via `post_commit`; they do not override the composition rule. + The capability RPC reports driver identity, version, and the default sandbox image used by the gateway. GPU availability stays driver-local and is validated when a sandbox create request asks for GPU resources. diff --git a/crates/openshell-driver-docker/src/lib.rs b/crates/openshell-driver-docker/src/lib.rs index d9bea99e7..e8feb5cdb 100644 --- a/crates/openshell-driver-docker/src/lib.rs +++ b/crates/openshell-driver-docker/src/lib.rs @@ -127,17 +127,6 @@ fn resolve_default_docker_supervisor_image_tag( tag.replace('+', "-") } -/// Queried by the Docker driver to decide when a sandbox's supervisor -/// relay is live. Implementations return `true` once a sandbox has an -/// active `ConnectSupervisor` session registered. -/// -/// The driver cannot observe the supervisor's SSH socket directly (it -/// lives inside the container), so it leans on this signal to flip the -/// Ready condition from `DependenciesNotReady` to `True`. -pub trait SupervisorReadiness: Send + Sync + 'static { - fn is_supervisor_connected(&self, sandbox_id: &str) -> bool; -} - /// Gateway-local configuration for the Docker compute driver. #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] #[serde(default, deny_unknown_fields)] @@ -254,7 +243,6 @@ pub struct DockerComputeDriver { config: DockerDriverRuntimeConfig, events: broadcast::Sender, pending: Arc>>, - supervisor_readiness: Arc, gpu_selector: Arc, } @@ -351,11 +339,7 @@ type WatchStream = Pin> + Send + 'static>>; impl DockerComputeDriver { - pub async fn new( - config: &Config, - docker_config: &DockerComputeConfig, - supervisor_readiness: Arc, - ) -> CoreResult { + pub async fn new(config: &Config, docker_config: &DockerComputeConfig) -> CoreResult { let docker = Docker::connect_with_local_defaults() .map_err(|err| Error::execution(format!("failed to create Docker client: {err}")))?; let version = docker.version().await.map_err(|err| { @@ -423,7 +407,6 @@ impl DockerComputeDriver { }, events: broadcast::channel(WATCH_BUFFER).0, pending: Arc::new(Mutex::new(HashMap::new())), - supervisor_readiness, gpu_selector: Arc::new(CdiGpuDefaultSelector::new( cdi_gpu_inventory, allow_all_default_gpu, @@ -634,9 +617,9 @@ impl DockerComputeDriver { let container = self .find_managed_container_summary(sandbox_id, sandbox_name) .await?; - if let Some(sandbox) = container.and_then(|summary| { - sandbox_from_container_summary(&summary, self.supervisor_readiness.as_ref()) - }) { + if let Some(sandbox) = + container.and_then(|summary| sandbox_from_container_summary(&summary)) + { return Ok(Some(sandbox)); } @@ -647,9 +630,7 @@ impl DockerComputeDriver { let containers = self.list_managed_container_summaries().await?; let container_sandboxes = containers .iter() - .filter_map(|summary| { - sandbox_from_container_summary(summary, self.supervisor_readiness.as_ref()) - }) + .filter_map(sandbox_from_container_summary) .collect::>(); let mut by_id = self.pending_snapshot_map().await; for sandbox in container_sandboxes { @@ -1151,8 +1132,7 @@ impl DockerComputeDriver { if let Some(summary) = self .find_managed_container_summary(sandbox_id, sandbox_name) .await? - && let Some(sandbox) = - sandbox_from_container_summary(&summary, self.supervisor_readiness.as_ref()) + && let Some(sandbox) = sandbox_from_container_summary(&summary) { self.publish_sandbox_snapshot(sandbox); } @@ -2761,10 +2741,7 @@ fn parse_memory_limit(value: &str) -> Result, Status> { Ok(Some((amount * multiplier).round() as i64)) } -fn sandbox_from_container_summary( - summary: &ContainerSummary, - readiness: &dyn SupervisorReadiness, -) -> Option { +fn sandbox_from_container_summary(summary: &ContainerSummary) -> Option { let labels = summary.labels.as_ref()?; let id = labels.get(LABEL_SANDBOX_ID)?.clone(); let name = labels.get(LABEL_SANDBOX_NAME)?.clone(); @@ -2773,27 +2750,21 @@ fn sandbox_from_container_summary( .cloned() .unwrap_or_default(); - let supervisor_connected = readiness.is_supervisor_connected(&id); Some(DriverSandbox { id, name: name.clone(), namespace, spec: None, - status: Some(driver_status_from_summary( - summary, - &name, - supervisor_connected, - )), + status: Some(driver_status_from_summary(summary, &name)), }) } fn driver_status_from_summary( summary: &ContainerSummary, sandbox_name: &str, - supervisor_connected: bool, ) -> DriverSandboxStatus { let state = summary.state.unwrap_or(ContainerSummaryStateEnum::EMPTY); - let (ready, reason, message, deleting) = container_ready_condition(state, supervisor_connected); + let (ready, reason, message, deleting) = container_ready_condition(state); DriverSandboxStatus { sandbox_name: summary_container_name(summary).unwrap_or_else(|| sandbox_name.to_string()), @@ -2813,25 +2784,10 @@ fn driver_status_from_summary( fn container_ready_condition( state: ContainerSummaryStateEnum, - supervisor_connected: bool, ) -> (&'static str, &'static str, &'static str, bool) { match state { ContainerSummaryStateEnum::RUNNING => { - if supervisor_connected { - ( - "True", - "SupervisorConnected", - "Supervisor relay is live", - false, - ) - } else { - ( - "False", - "DependenciesNotReady", - "Container is running; waiting for supervisor relay", - false, - ) - } + ("True", "BackendReady", "Container is running", false) } ContainerSummaryStateEnum::CREATED => ("False", "Starting", "Container created", false), ContainerSummaryStateEnum::RESTARTING => ( diff --git a/crates/openshell-driver-docker/src/tests.rs b/crates/openshell-driver-docker/src/tests.rs index d42d41099..3813d3e68 100644 --- a/crates/openshell-driver-docker/src/tests.rs +++ b/crates/openshell-driver-docker/src/tests.rs @@ -169,14 +169,6 @@ fn inspected_volume(driver: &str, options: HashMap) -> bollard:: } } -struct DisconnectedSupervisorReadiness; - -impl SupervisorReadiness for DisconnectedSupervisorReadiness { - fn is_supervisor_connected(&self, _sandbox_id: &str) -> bool { - false - } -} - fn test_driver_with_config(config: DockerDriverRuntimeConfig) -> DockerComputeDriver { let allow_all_default_gpu = config.allow_all_default_gpu; DockerComputeDriver { @@ -187,7 +179,6 @@ fn test_driver_with_config(config: DockerDriverRuntimeConfig) -> DockerComputeDr config, events: broadcast::channel(WATCH_BUFFER).0, pending: Arc::new(tokio::sync::Mutex::new(HashMap::new())), - supervisor_readiness: Arc::new(DisconnectedSupervisorReadiness), gpu_selector: Arc::new(CdiGpuDefaultSelector::new( CdiGpuInventory::default(), allow_all_default_gpu, @@ -1723,34 +1714,19 @@ fn driver_status_keeps_running_sandboxes_provisioning_with_stable_message() { ..running.clone() }; - let running_status = driver_status_from_summary(&running, "demo", false); - let running_later_status = driver_status_from_summary(&running_later, "demo", false); - assert_eq!(running_status.conditions[0].status, "False"); - assert_eq!(running_status.conditions[0].reason, "DependenciesNotReady"); - assert_eq!( - running_status.conditions[0].message, - "Container is running; waiting for supervisor relay" - ); + // A running container always emits Ready=True with BackendReady. The gateway + // composes this with supervisor-session presence to decide public SandboxPhase. + let running_status = driver_status_from_summary(&running, "demo"); + let running_later_status = driver_status_from_summary(&running_later, "demo"); + assert_eq!(running_status.conditions[0].status, "True"); + assert_eq!(running_status.conditions[0].reason, "BackendReady"); + assert_eq!(running_status.conditions[0].message, "Container is running"); assert_eq!(running_status.conditions, running_later_status.conditions); - let exited_status = driver_status_from_summary(&exited, "demo", false); + let exited_status = driver_status_from_summary(&exited, "demo"); assert_eq!(exited_status.conditions[0].status, "False"); assert_eq!(exited_status.conditions[0].reason, "ContainerExited"); assert_eq!(exited_status.conditions[0].message, "Container exited"); - - // With a live supervisor session, a RUNNING container flips Ready=True - // so ExecSandbox and other "sandbox must be ready" gates can proceed. - let running_connected = driver_status_from_summary(&running, "demo", true); - assert_eq!(running_connected.conditions[0].status, "True"); - assert_eq!( - running_connected.conditions[0].reason, - "SupervisorConnected" - ); - - // Supervisor readiness is ignored for non-RUNNING states -- an exited - // container must not report Ready=True. - let exited_connected = driver_status_from_summary(&exited, "demo", true); - assert_eq!(exited_connected.conditions[0].status, "False"); } #[test] @@ -1768,7 +1744,7 @@ fn driver_status_marks_restarting_sandboxes_as_error() { ..Default::default() }; - let status = driver_status_from_summary(&restarting, "demo", false); + let status = driver_status_from_summary(&restarting, "demo"); assert_eq!(status.conditions[0].status, "False"); assert_eq!(status.conditions[0].reason, "ContainerRestarting"); assert_eq!( diff --git a/crates/openshell-server/src/compute/mod.rs b/crates/openshell-server/src/compute/mod.rs index 3a92cd209..12a4b3afd 100644 --- a/crates/openshell-server/src/compute/mod.rs +++ b/crates/openshell-server/src/compute/mod.rs @@ -346,7 +346,7 @@ impl ComputeRuntime { supervisor_sessions: Arc, ) -> Result { let driver = Arc::new( - DockerComputeDriver::new(&config, &docker_config, supervisor_sessions.clone()) + DockerComputeDriver::new(&config, &docker_config) .await .map_err(|err| ComputeError::Message(err.to_string()))?, ); @@ -1129,97 +1129,102 @@ impl ComputeRuntime { .map(decode_sandbox_record) .transpose()?; - // If no existing record, create initial sandbox (first watch event for this sandbox) - if existing.is_none() { - use crate::persistence::WriteCondition; - let now_ms = openshell_core::time::now_ms(); + match (existing.as_ref(), incoming.status.as_ref()) { + (None, _) => self.create_sandbox_record(incoming).await, + (Some(_), None) => Ok(()), // session events are owned by set_supervisor_session_state + (Some(_), Some(_)) => self.update_sandbox_record(incoming).await, + } + } - let session_connected = self.supervisor_sessions.has_session(&incoming.id); - let mut phase = derive_phase(incoming.status.as_ref()); - let sandbox_name = incoming.name.clone(); + // First watch event for a sandbox: build the initial record and persist it. + async fn create_sandbox_record(&self, incoming: DriverSandbox) -> Result<(), String> { + use crate::persistence::WriteCondition; + let now_ms = openshell_core::time::now_ms(); - let supervisor_promoted = session_connected - && matches!(phase, SandboxPhase::Provisioning | SandboxPhase::Unknown); - if supervisor_promoted { - phase = SandboxPhase::Ready; - } + // supervisor_sessions is process-local. In HA deployments this gateway may not + // own the active supervisor session. A replica that does not own the session will + // hold the public phase at Provisioning; the owning replica's + // supervisor_session_connected write to the shared store will propagate through + // the reconcile loop. + let session_connected = self.supervisor_sessions.has_session(&incoming.id); + let sandbox_name = incoming.name.clone(); - let mut status = incoming + let (phase, status) = + incoming .status .as_ref() - .map(|s| public_status_from_driver(s, phase, 0)); - rewrite_user_facing_conditions(&mut status, None); - if supervisor_promoted { - ensure_supervisor_ready_status(&mut status, &sandbox_name); - } - let mut sandbox = Sandbox { - metadata: Some(openshell_core::proto::datamodel::v1::ObjectMeta { - id: incoming.id.clone(), - name: sandbox_name, - created_at_ms: now_ms, - labels: std::collections::HashMap::new(), - resource_version: 0, - }), - spec: None, - status, - }; - sandbox.set_phase(phase as i32); - - self.store - .put_if( - Sandbox::object_type(), - &incoming.id, - sandbox.object_name(), - &sandbox.encode_to_vec(), - None, - WriteCondition::MustCreate, - ) - .await - .map_err(|e| match e { - crate::persistence::PersistenceError::Conflict { - current_resource_version, - } => format!( - "concurrent modification detected during sandbox creation (current resource_version: {})", - current_resource_version - .map_or_else(|| "unknown".to_string(), |v| v.to_string()) - ), - other => other.to_string(), - })?; + .map_or((SandboxPhase::Provisioning, None), |s| { + let composed = ComposedPhase::new(s, session_connected); + let mut status = Some(public_status_from_driver(s, composed.phase, 0)); + composed.apply_readiness_conditions(&mut status, &sandbox_name, None); + (composed.phase, status) + }); - self.sandbox_index.update_from_sandbox(&sandbox); - self.sandbox_watch_bus.notify(sandbox.object_id()); - return Ok(()); - } + let mut sandbox = Sandbox { + metadata: Some(openshell_core::proto::datamodel::v1::ObjectMeta { + id: incoming.id.clone(), + name: sandbox_name, + created_at_ms: now_ms, + labels: std::collections::HashMap::new(), + resource_version: 0, + }), + spec: None, + status, + }; + sandbox.set_phase(phase as i32); + + self.store + .put_if( + Sandbox::object_type(), + &incoming.id, + sandbox.object_name(), + &sandbox.encode_to_vec(), + None, + WriteCondition::MustCreate, + ) + .await + .map_err(|e| match e { + crate::persistence::PersistenceError::Conflict { + current_resource_version, + } => format!( + "concurrent modification detected during sandbox creation (current resource_version: {})", + current_resource_version + .map_or_else(|| "unknown".to_string(), |v| v.to_string()) + ), + other => other.to_string(), + })?; - // Single-attempt CAS: on conflict, the next watch event will naturally retry + self.sandbox_index.update_from_sandbox(&sandbox); + self.sandbox_watch_bus.notify(sandbox.object_id()); + Ok(()) + } + + // Subsequent driver snapshot for an existing sandbox: apply a single-attempt CAS update. + // On conflict the next watch event will naturally retry. + // Caller guarantees incoming.status is Some. + async fn update_sandbox_record(&self, incoming: DriverSandbox) -> Result<(), String> { let session_connected = self.supervisor_sessions.has_session(&incoming.id); let sandbox_name = incoming.name.clone(); + let incoming_status = incoming + .status + .as_ref() + .expect("caller guarantees status is Some"); let sandbox = self .store .update_message_cas::(&incoming.id, 0, |sandbox| { let old_phase = SandboxPhase::try_from(sandbox.phase()).unwrap_or(SandboxPhase::Unknown); - let mut phase = incoming - .status - .as_ref() - .map_or(old_phase, |status| derive_phase(Some(status))); - let supervisor_promoted = session_connected - && matches!(phase, SandboxPhase::Provisioning | SandboxPhase::Unknown); - if supervisor_promoted { - phase = SandboxPhase::Ready; - } + + let composed = ComposedPhase::new(incoming_status, session_connected); let cpv = sandbox.current_policy_version(); - let mut status = incoming - .status - .as_ref() - .map(|s| public_status_from_driver(s, phase, cpv)) - .or_else(|| sandbox.status.clone()); - rewrite_user_facing_conditions(&mut status, sandbox.spec.as_ref()); - if supervisor_promoted { - ensure_supervisor_ready_status(&mut status, &sandbox_name); - } + let mut status = Some(public_status_from_driver(incoming_status, composed.phase, cpv)); + composed.apply_readiness_conditions( + &mut status, + &sandbox_name, + sandbox.spec.as_ref(), + ); if let Some(s) = status.as_mut() && s.sandbox_name.is_empty() @@ -1227,17 +1232,17 @@ impl ComputeRuntime { s.sandbox_name.clone_from(&sandbox_name); } - if old_phase != phase { + if old_phase != composed.phase { info!( sandbox_id = %incoming.id, sandbox_name = %sandbox_name, old_phase = ?old_phase, - new_phase = ?phase, + new_phase = ?composed.phase, "Sandbox phase changed" ); } - if phase == SandboxPhase::Error + if composed.phase == SandboxPhase::Error && let Some(ref status) = status { for condition in &status.conditions { @@ -1256,12 +1261,11 @@ impl ComputeRuntime { } } - // Update metadata fields if let Some(metadata) = sandbox.metadata.as_mut() { metadata.name.clone_from(&sandbox_name); } sandbox.status = status; - sandbox.set_phase(phase as i32); + sandbox.set_phase(composed.phase as i32); sandbox.set_current_policy_version(cpv); }) .await @@ -1866,6 +1870,48 @@ fn ensure_supervisor_ready_status(status: &mut Option, sandbox_na ); } +/// Compose the public `SandboxPhase` from backend driver state and supervisor session presence. +/// +/// The readiness decision is a gateway-owned safety invariant: `SandboxPhase::Ready` means +/// "usable through this gateway." The driver contract is the extension point for custom backend +/// readiness semantics. RFC-0010 lifecycle hooks observe this decision via `post_commit`; they +/// do not modify it. +struct ComposedPhase { + phase: SandboxPhase, + session_connected: bool, +} + +impl ComposedPhase { + fn new(incoming_status: &DriverSandboxStatus, session_connected: bool) -> Self { + let backend_phase = derive_phase(Some(incoming_status)); + // A live supervisor session is a stronger readiness signal than the backend phase. + // set_supervisor_session_state may have already promoted the store record to Ready + // before this driver snapshot arrived. Keep Ready rather than letting a lagging + // backend phase overwrite it. + let phase = match backend_phase { + SandboxPhase::Error | SandboxPhase::Deleting => backend_phase, + _ if session_connected => SandboxPhase::Ready, + _ => SandboxPhase::Provisioning, + }; + Self { + phase, + session_connected, + } + } + + fn apply_readiness_conditions( + &self, + status: &mut Option, + sandbox_name: &str, + spec: Option<&SandboxSpec>, + ) { + rewrite_user_facing_conditions(status, spec); + if self.session_connected && self.phase == SandboxPhase::Ready { + ensure_supervisor_ready_status(status, sandbox_name); + } + } +} + fn ensure_supervisor_not_ready_status(status: &mut Option, sandbox_name: &str) { upsert_ready_condition( status, @@ -2790,6 +2836,10 @@ mod tests { #[tokio::test] async fn apply_sandbox_update_allows_delete_failures_to_recover() { + // A previously-Deleting sandbox that receives a Ready driver snapshot (delete failed, + // container is still running) without a supervisor session stays at Provisioning. + // The gateway composition rule treats backend-ready + no session as Provisioning, not Ready. + // Recovery to Ready requires the supervisor to reconnect. let runtime = test_runtime(Arc::new(TestDriver::default())).await; let sandbox = sandbox_record("sb-1", "sandbox-a", SandboxPhase::Deleting); runtime.store.put_message(&sandbox).await.unwrap(); @@ -2808,8 +2858,8 @@ mod tests { conditions: vec![DriverCondition { r#type: "Ready".to_string(), status: "True".to_string(), - reason: "DependenciesReady".to_string(), - message: "Pod is Ready".to_string(), + reason: "BackendReady".to_string(), + message: "Container is running".to_string(), last_transition_time: String::new(), }], deleting: false, @@ -2826,8 +2876,14 @@ mod tests { .unwrap(); assert_eq!( SandboxPhase::try_from(stored.phase()).unwrap(), - SandboxPhase::Ready + SandboxPhase::Provisioning ); + let ready_condition = stored + .status + .as_ref() + .and_then(|s| s.conditions.iter().find(|c| c.r#type == "Ready")) + .unwrap(); + assert_eq!(ready_condition.reason, "BackendReady"); } #[tokio::test] @@ -3001,6 +3057,276 @@ mod tests { assert_eq!(ready.message, "Supervisor session disconnected"); } + // --- Composition rule tests --- + + fn make_ready_driver_status() -> DriverSandboxStatus { + DriverSandboxStatus { + sandbox_name: "test".to_string(), + instance_id: "test-pod".to_string(), + agent_fd: String::new(), + sandbox_fd: String::new(), + conditions: vec![DriverCondition { + r#type: "Ready".to_string(), + status: "True".to_string(), + reason: "BackendReady".to_string(), + message: "Container is running".to_string(), + last_transition_time: String::new(), + }], + deleting: false, + } + } + + fn make_deleting_driver_status() -> DriverSandboxStatus { + DriverSandboxStatus { + sandbox_name: "test".to_string(), + instance_id: "test-pod".to_string(), + agent_fd: String::new(), + sandbox_fd: String::new(), + conditions: vec![DriverCondition { + r#type: "Ready".to_string(), + status: "False".to_string(), + reason: "Deleting".to_string(), + message: "Container is being removed".to_string(), + last_transition_time: String::new(), + }], + deleting: true, + } + } + + fn ready_condition(sandbox: &Sandbox) -> Option<&SandboxCondition> { + sandbox + .status + .as_ref() + .and_then(|s| s.conditions.iter().find(|c| c.r#type == "Ready")) + } + + #[tokio::test] + async fn backend_ready_without_supervisor_stays_provisioning() { + let runtime = test_runtime(Arc::new(TestDriver::default())).await; + let sandbox = sandbox_record("sb-1", "sandbox-a", SandboxPhase::Provisioning); + runtime.store.put_message(&sandbox).await.unwrap(); + + runtime + .apply_sandbox_update(DriverSandbox { + id: "sb-1".to_string(), + name: "sandbox-a".to_string(), + namespace: "default".to_string(), + spec: None, + status: Some(make_ready_driver_status()), + }) + .await + .unwrap(); + + let stored = runtime + .store + .get_message::("sb-1") + .await + .unwrap() + .unwrap(); + assert_eq!( + SandboxPhase::try_from(stored.phase()).unwrap(), + SandboxPhase::Provisioning + ); + let cond = ready_condition(&stored).unwrap(); + assert_eq!(cond.status, "True"); + assert_eq!(cond.reason, "BackendReady"); + } + + #[tokio::test] + async fn backend_ready_with_supervisor_becomes_ready() { + let runtime = test_runtime(Arc::new(TestDriver::default())).await; + let sandbox = sandbox_record("sb-1", "sandbox-a", SandboxPhase::Provisioning); + runtime.store.put_message(&sandbox).await.unwrap(); + register_test_supervisor_session(&runtime, "sb-1"); + + runtime + .apply_sandbox_update(DriverSandbox { + id: "sb-1".to_string(), + name: "sandbox-a".to_string(), + namespace: "default".to_string(), + spec: None, + status: Some(make_ready_driver_status()), + }) + .await + .unwrap(); + + let stored = runtime + .store + .get_message::("sb-1") + .await + .unwrap() + .unwrap(); + assert_eq!( + SandboxPhase::try_from(stored.phase()).unwrap(), + SandboxPhase::Ready + ); + let cond = ready_condition(&stored).unwrap(); + assert_eq!(cond.status, "True"); + assert_eq!(cond.reason, "DependenciesReady"); + } + + #[tokio::test] + async fn backend_not_ready_with_supervisor_becomes_ready() { + // VM path: supervisor connects before backend reports Ready. + let runtime = test_runtime(Arc::new(TestDriver::default())).await; + let sandbox = sandbox_record("sb-1", "sandbox-a", SandboxPhase::Provisioning); + runtime.store.put_message(&sandbox).await.unwrap(); + register_test_supervisor_session(&runtime, "sb-1"); + + runtime + .apply_sandbox_update(DriverSandbox { + id: "sb-1".to_string(), + name: "sandbox-a".to_string(), + namespace: "default".to_string(), + spec: None, + status: Some(make_driver_status(make_driver_condition( + "Starting", + "VM is starting", + ))), + }) + .await + .unwrap(); + + let stored = runtime + .store + .get_message::("sb-1") + .await + .unwrap() + .unwrap(); + assert_eq!( + SandboxPhase::try_from(stored.phase()).unwrap(), + SandboxPhase::Ready + ); + } + + #[tokio::test] + async fn terminal_failure_ignores_supervisor_session() { + let runtime = test_runtime(Arc::new(TestDriver::default())).await; + let sandbox = sandbox_record("sb-1", "sandbox-a", SandboxPhase::Provisioning); + runtime.store.put_message(&sandbox).await.unwrap(); + register_test_supervisor_session(&runtime, "sb-1"); + + runtime + .apply_sandbox_update(DriverSandbox { + id: "sb-1".to_string(), + name: "sandbox-a".to_string(), + namespace: "default".to_string(), + spec: None, + status: Some(make_driver_status(make_driver_condition( + "ImagePullBackOff", + "Failed to pull image", + ))), + }) + .await + .unwrap(); + + let stored = runtime + .store + .get_message::("sb-1") + .await + .unwrap() + .unwrap(); + assert_eq!( + SandboxPhase::try_from(stored.phase()).unwrap(), + SandboxPhase::Error + ); + } + + #[tokio::test] + async fn later_driver_ready_without_session_does_not_repromote() { + // Re-promotion bug fix: backend-ready snapshot after session disconnect must not + // re-promote the sandbox to Ready. + let runtime = test_runtime(Arc::new(TestDriver::default())).await; + let sandbox = sandbox_record("sb-1", "sandbox-a", SandboxPhase::Provisioning); + runtime.store.put_message(&sandbox).await.unwrap(); + + // Promote to Ready via supervisor session connect. + register_test_supervisor_session(&runtime, "sb-1"); + runtime.supervisor_session_connected("sb-1").await.unwrap(); + let stored = runtime + .store + .get_message::("sb-1") + .await + .unwrap() + .unwrap(); + assert_eq!( + SandboxPhase::try_from(stored.phase()).unwrap(), + SandboxPhase::Ready + ); + + // Session drops. + runtime.supervisor_sessions.cleanup_sandbox("sb-1"); + runtime + .supervisor_session_disconnected("sb-1") + .await + .unwrap(); + let stored = runtime + .store + .get_message::("sb-1") + .await + .unwrap() + .unwrap(); + assert_eq!( + SandboxPhase::try_from(stored.phase()).unwrap(), + SandboxPhase::Provisioning + ); + + // Backend-ready snapshot arrives with no active session — must not re-promote. + runtime + .apply_sandbox_update(DriverSandbox { + id: "sb-1".to_string(), + name: "sandbox-a".to_string(), + namespace: "default".to_string(), + spec: None, + status: Some(make_ready_driver_status()), + }) + .await + .unwrap(); + + let stored = runtime + .store + .get_message::("sb-1") + .await + .unwrap() + .unwrap(); + assert_eq!( + SandboxPhase::try_from(stored.phase()).unwrap(), + SandboxPhase::Provisioning + ); + let cond = ready_condition(&stored).unwrap(); + assert_eq!(cond.reason, "BackendReady"); + } + + #[tokio::test] + async fn deleting_ignores_supervisor_session() { + let runtime = test_runtime(Arc::new(TestDriver::default())).await; + let sandbox = sandbox_record("sb-1", "sandbox-a", SandboxPhase::Provisioning); + runtime.store.put_message(&sandbox).await.unwrap(); + register_test_supervisor_session(&runtime, "sb-1"); + + runtime + .apply_sandbox_update(DriverSandbox { + id: "sb-1".to_string(), + name: "sandbox-a".to_string(), + namespace: "default".to_string(), + spec: None, + status: Some(make_deleting_driver_status()), + }) + .await + .unwrap(); + + let stored = runtime + .store + .get_message::("sb-1") + .await + .unwrap() + .unwrap(); + assert_eq!( + SandboxPhase::try_from(stored.phase()).unwrap(), + SandboxPhase::Deleting + ); + } + #[tokio::test] async fn reconcile_store_with_backend_applies_driver_snapshot() { let runtime = test_runtime(Arc::new(TestDriver { @@ -3058,6 +3384,7 @@ mod tests { }; runtime.store.put_message(&sandbox).await.unwrap(); runtime.sandbox_index.update_from_sandbox(&sandbox); + register_test_supervisor_session(&runtime, "sb-1"); runtime .reconcile_store_with_backend(Duration::ZERO) @@ -3153,6 +3480,7 @@ mod tests { let sandbox = sandbox_record("sb-1", "sandbox-a", SandboxPhase::Provisioning); runtime.store.put_message(&sandbox).await.unwrap(); runtime.sandbox_index.update_from_sandbox(&sandbox); + register_test_supervisor_session(&runtime, "sb-1"); runtime .reconcile_store_with_backend(Duration::ZERO) diff --git a/crates/openshell-server/src/supervisor_session.rs b/crates/openshell-server/src/supervisor_session.rs index 4adf9e8b6..e5225d13c 100644 --- a/crates/openshell-server/src/supervisor_session.rs +++ b/crates/openshell-server/src/supervisor_session.rs @@ -63,12 +63,6 @@ struct LiveSession { /// target-open failure reported by the supervisor. type RelayStreamSender = oneshot::Sender>; -impl openshell_driver_docker::SupervisorReadiness for SupervisorSessionRegistry { - fn is_supervisor_connected(&self, sandbox_id: &str) -> bool { - Self::is_connected(self, sandbox_id) - } -} - /// Registry of active supervisor sessions and pending relay channels. #[derive(Default)] pub struct SupervisorSessionRegistry {