From 716ea506f44ebc543b8ed8e8766dc87b4f8b3cea Mon Sep 17 00:00:00 2001 From: Evan Lezar Date: Mon, 6 Jul 2026 14:35:43 +0200 Subject: [PATCH 1/2] refactor(compute): extract create_sandbox_record and update_sandbox_record helpers Split apply_sandbox_update_locked into named helpers to make the two distinct paths explicit: create_sandbox_record for first-observation events and update_sandbox_record for subsequent driver snapshots on existing sandboxes. The dispatcher now uses a match on the existing record rather than an early-return guard. No behavior change. Signed-off-by: Evan Lezar --- crates/openshell-server/src/compute/mod.rs | 130 +++++++++++---------- 1 file changed, 69 insertions(+), 61 deletions(-) diff --git a/crates/openshell-server/src/compute/mod.rs b/crates/openshell-server/src/compute/mod.rs index 239feb74d..c0cb837e1 100644 --- a/crates/openshell-server/src/compute/mod.rs +++ b/crates/openshell-server/src/compute/mod.rs @@ -1157,70 +1157,79 @@ 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(); - - 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(); - - let supervisor_promoted = session_connected - && matches!(phase, SandboxPhase::Provisioning | SandboxPhase::Unknown); - if supervisor_promoted { - phase = SandboxPhase::Ready; - } + match existing.as_ref() { + None => self.create_sandbox_record(incoming).await, + Some(_) => self.update_sandbox_record(incoming).await, + } + } - let mut 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, - annotations: std::collections::HashMap::new(), - }), - 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(), - })?; + // 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(); - self.sandbox_index.update_from_sandbox(&sandbox); - self.sandbox_watch_bus.notify(sandbox.object_id()); - return Ok(()); + 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(); + + let supervisor_promoted = session_connected + && matches!(phase, SandboxPhase::Provisioning | SandboxPhase::Unknown); + if supervisor_promoted { + phase = SandboxPhase::Ready; + } + + let mut 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); } - // Single-attempt CAS: on conflict, the next watch event will naturally retry + 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, + annotations: std::collections::HashMap::new(), + }), + 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(), + })?; + + 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. + 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(); @@ -1285,7 +1294,6 @@ impl ComputeRuntime { } } - // Update metadata fields if let Some(metadata) = sandbox.metadata.as_mut() { metadata.name.clone_from(&sandbox_name); } From 761589ecbcc9ed4760e47d093d5320d8017a3354 Mon Sep 17 00:00:00 2001 From: Evan Lezar Date: Mon, 6 Jul 2026 16:21:39 +0200 Subject: [PATCH 2/2] refactor(compute): make sandbox readiness gateway-owned across all drivers Introduce compute_phase_components and apply_readiness_conditions to centralise the gateway's phase composition logic. The public SandboxPhase is now determined by combining the backend phase reported by the driver with supervisor session presence, independent of the driver implementation. Remove SupervisorReadiness from the driver contract. Running containers always report BackendReady; the gateway owns the Ready decision. Rename the dispatcher match to three arms so that status-less events for existing sandboxes are a documented no-op rather than a silent pass-through. Drop backend_ready_no_session and the SupervisorNotConnected condition. The BackendReady driver condition plus the Provisioning phase already communicates that the backend is up but the supervisor has not connected. The redundant condition added noise without new information. Closes #1951 Signed-off-by: Evan Lezar --- architecture/compute-runtimes.md | 48 +++ crates/openshell-driver-docker/src/lib.rs | 107 ++--- crates/openshell-driver-docker/src/tests.rs | 42 +- crates/openshell-server/src/compute/mod.rs | 408 ++++++++++++++++-- .../src/supervisor_session.rs | 6 - 5 files changed, 477 insertions(+), 134 deletions(-) diff --git a/architecture/compute-runtimes.md b/architecture/compute-runtimes.md index ec42277ab..c97f3ad17 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 97a0f6083..be42e3250 100644 --- a/crates/openshell-driver-docker/src/lib.rs +++ b/crates/openshell-driver-docker/src/lib.rs @@ -78,15 +78,53 @@ const HOST_OPENSHELL_INTERNAL: &str = "host.openshell.internal"; const HOST_DOCKER_INTERNAL: &str = "host.docker.internal"; const DOCKER_NETWORK_DRIVER: &str = "bridge"; -/// 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. +/// Default image holding the Linux `openshell-sandbox` binary. The gateway +/// pulls this image and extracts the binary to a host-side cache when no +/// explicit `supervisor_bin`, configured `supervisor_image`, sibling binary, +/// or local build is available. +const DEFAULT_DOCKER_SUPERVISOR_IMAGE_REPO: &str = "ghcr.io/nvidia/openshell/supervisor"; + +/// Return the default `ghcr.io/nvidia/openshell/supervisor:` reference +/// used when no supervisor binary override is provided. +pub fn default_docker_supervisor_image() -> String { + format!( + "{DEFAULT_DOCKER_SUPERVISOR_IMAGE_REPO}:{}", + default_docker_supervisor_image_tag() + ) +} + +/// Image tag baked in at compile time to pair the gateway with a matching +/// supervisor image. /// -/// 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; +/// Build pipelines pass `OPENSHELL_IMAGE_TAG` explicitly. The `IMAGE_TAG` +/// fallback covers image build wrappers that already tag the gateway and +/// supervisor together. Standalone release binaries also patch the Cargo +/// package version, so use it when it has been set to a real release value. +fn default_docker_supervisor_image_tag() -> String { + resolve_default_docker_supervisor_image_tag( + option_env!("OPENSHELL_IMAGE_TAG"), + option_env!("IMAGE_TAG"), + env!("CARGO_PKG_VERSION"), + ) +} + +fn resolve_default_docker_supervisor_image_tag( + openshell_image_tag: Option<&'static str>, + image_tag: Option<&'static str>, + cargo_pkg_version: &'static str, +) -> String { + let tag = openshell_image_tag + .filter(|tag| !tag.is_empty()) + .or_else(|| image_tag.filter(|tag| !tag.is_empty())) + .unwrap_or_else(|| { + if cargo_pkg_version.is_empty() || cargo_pkg_version == "0.0.0" { + "dev" + } else { + cargo_pkg_version + } + }); + + tag.replace('+', "-") } /// Gateway-local configuration for the Docker compute driver. @@ -211,7 +249,6 @@ pub struct DockerComputeDriver { config: DockerDriverRuntimeConfig, events: broadcast::Sender, pending: Arc>>, - supervisor_readiness: Arc, gpu_selector: Arc, } @@ -308,11 +345,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 socket_path = docker_config .socket_path .clone() @@ -394,7 +427,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, @@ -605,9 +637,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)); } @@ -618,9 +650,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 { @@ -1122,8 +1152,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); } @@ -2732,10 +2761,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(); @@ -2744,27 +2770,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()), @@ -2784,25 +2804,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 7ca340525..bcb689613 100644 --- a/crates/openshell-driver-docker/src/tests.rs +++ b/crates/openshell-driver-docker/src/tests.rs @@ -140,14 +140,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 { @@ -158,7 +150,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, @@ -1727,34 +1718,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] @@ -1772,7 +1748,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 c0cb837e1..ddd3a2b94 100644 --- a/crates/openshell-server/src/compute/mod.rs +++ b/crates/openshell-server/src/compute/mod.rs @@ -361,7 +361,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()))?, ); @@ -1157,9 +1157,10 @@ impl ComputeRuntime { .map(decode_sandbox_record) .transpose()?; - match existing.as_ref() { - None => self.create_sandbox_record(incoming).await, - Some(_) => self.update_sandbox_record(incoming).await, + 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, } } @@ -1168,24 +1169,24 @@ impl ComputeRuntime { use crate::persistence::WriteCondition; let now_ms = openshell_core::time::now_ms(); + // 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 mut phase = derive_phase(incoming.status.as_ref()); let sandbox_name = incoming.name.clone(); - let supervisor_promoted = session_connected - && matches!(phase, SandboxPhase::Provisioning | SandboxPhase::Unknown); - if supervisor_promoted { - phase = SandboxPhase::Ready; - } - - let mut 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 (phase, status) = + incoming + .status + .as_ref() + .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) + }); let mut sandbox = Sandbox { metadata: Some(openshell_core::proto::datamodel::v1::ObjectMeta { @@ -1229,35 +1230,30 @@ impl ComputeRuntime { // 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() @@ -1265,17 +1261,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 { @@ -1298,7 +1294,7 @@ impl ComputeRuntime { 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 @@ -1903,6 +1899,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, @@ -2837,6 +2875,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(); @@ -2855,8 +2897,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, @@ -2873,8 +2915,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] @@ -3048,6 +3096,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 { @@ -3105,6 +3423,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) @@ -3200,6 +3519,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 16ddbc0d4..e425a6504 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 {