fix(server): consistent sandbox access after the main process exits

crates/openshell-cli/src/run.rs

@@ -2014,15 +2014,6 @@ pub async fn sandbox_exec_grpc(
         .sandbox
         .ok_or_else(|| miette::miette!("sandbox not found"))?;
 
-    // Verify the sandbox is ready before issuing the exec.
-    if SandboxPhase::try_from(sandbox.phase()) != Ok(SandboxPhase::Ready) {
-        return Err(miette::miette!(
-            "sandbox '{}' is not ready (phase: {}); wait for it to reach Ready state",
-            name,
-            phase_name(sandbox.phase())
-        ));
-    }
-
     // Resolve TTY mode: explicit --tty / --no-tty wins, otherwise auto-detect.
     let stdin_is_terminal = std::io::stdin().is_terminal();
     let tty = tty_override.unwrap_or_else(|| stdin_is_terminal && std::io::stdout().is_terminal());
@@ -2115,7 +2106,7 @@ pub async fn sandbox_exec_grpc(
         }
         (response, _) => response,
     }
-    .into_diagnostic()?
+    .map_err(|status| exec_start_error(status, &sandbox))?
     .into_inner();
 
     // Stream output to terminal in real-time.
@@ -2169,7 +2160,7 @@ pub async fn service_forward_tcp(
     let (bind_addr, bind_port) = parse_tcp_forward_spec(local, target_port)?;
     let mut client = grpc_client(server, tls).await?;
 
-    let mut replica = fetch_ready_sandbox_for_forward(&mut client, name, workspace).await?;
+    let mut replica = fetch_sandbox_for_forward(&mut client, name, workspace).await?;
 
     let listener = tokio::net::TcpListener::bind((bind_addr.as_str(), bind_port))
         .await
@@ -2203,7 +2194,7 @@ pub async fn service_forward_tcp(
             }
 
             _ = health_check.tick() => {
-                replica = fetch_ready_sandbox_for_forward(&mut client, name, workspace).await?;
+                replica = fetch_sandbox_for_forward(&mut client, name, workspace).await?;
             }
 
             accepted = listener.accept() => {
@@ -2269,8 +2260,13 @@ fn forward_requires_session_token(status: &Status) -> bool {
         && status.message().contains("authorization_token is required")
 }
 
-/// Confirm the sandbox is still ready, returning the replica that owns it.
-async fn fetch_ready_sandbox_for_forward(
+/// Confirm the sandbox can still serve the forward, returning the replica that
+/// owns it.
+///
+/// The gateway decides whether each connection may reach the sandbox; this
+/// only stops a forward whose sandbox is gone or is stopping, stopped, or
+/// being deleted, since no connection can succeed until a new command acts.
+async fn fetch_sandbox_for_forward(
     client: &mut crate::tls::GrpcClient,
     name: &str,
     workspace: &str,
@@ -2299,17 +2295,25 @@ async fn fetch_ready_sandbox_for_forward(
         .sandbox
         .ok_or_else(|| miette::miette!("sandbox '{name}' not found"))?;
 
-    if SandboxPhase::try_from(sandbox.phase()) != Ok(SandboxPhase::Ready) {
+    if forward_ended_by_phase(sandbox.phase()) {
         return Err(miette::miette!(
-            "sandbox '{}' is no longer ready (phase: {}); stopping service forward",
-            name,
+            "sandbox '{name}' is no longer available (phase: {}); stopping service forward",
             phase_name(sandbox.phase())
         ));
     }
 
     Ok(replica)
 }
 
+/// Phases in which no forwarded connection can succeed until another command
+/// starts or recreates the sandbox.
+fn forward_ended_by_phase(phase: i32) -> bool {
+    matches!(
+        SandboxPhase::try_from(phase),
+        Ok(SandboxPhase::Stopping | SandboxPhase::Stopped | SandboxPhase::Deleting)
+    )
+}
+
 #[derive(Debug)]
 struct ForwardTcpConnectionError {
     message: String,
@@ -2576,6 +2580,28 @@ impl Drop for TaskGuard {
     }
 }
 
+/// The gateway decides whether a sandbox can be used. When it refuses, name the
+/// phase this command saw and what would make the sandbox usable again.
+fn exec_start_error(status: Status, sandbox: &Sandbox) -> miette::Report {
+    if status.code() == Code::FailedPrecondition && status.message() == "sandbox is not ready" {
+        let name = sandbox.object_name();
+        let hint = match SandboxPhase::try_from(sandbox.phase()) {
+            Ok(SandboxPhase::Stopping | SandboxPhase::Stopped) => {
+                format!("start it with `openshell sandbox start {name}`")
+            }
+            Ok(SandboxPhase::Completed | SandboxPhase::Error) => {
+                "its main process exited and its supervisor is no longer connected".to_string()
+            }
+            _ => "wait for it to reach Ready state".to_string(),
+        };
+        return miette::miette!(
+            "sandbox '{name}' is not ready (phase: {}); {hint}",
+            phase_name(sandbox.phase())
+        );
+    }
+    miette::Report::from_err(status)
+}
+
 #[allow(clippy::too_many_arguments)]
 async fn sandbox_exec_streaming_grpc(
     mut client: crate::tls::GrpcClient,
@@ -2638,7 +2664,7 @@ async fn sandbox_exec_streaming_grpc(
         }
         response => response,
     }
-    .into_diagnostic()?
+    .map_err(|status| exec_start_error(status, sandbox))?
     .into_inner();
 
     // Raw mode is only appropriate for an interactive terminal, not a pipe.
@@ -7124,14 +7150,14 @@ fn format_endpoint(endpoint: &openshell_core::proto::NetworkEndpoint) -> String
 mod tests {
     use super::{
         ForwardTcpConnectionError, PolicyGetView, ProvisioningStep, build_sandbox_resource_limits,
-        format_endpoint, format_log_line, forward_requires_session_token, git_sync_files,
-        has_main_process_result, parse_cli_setting_value, parse_credential_expiry_cli_value,
-        parse_driver_config_json, parse_env_from_pairs, parse_secret_material_env_pairs,
-        policy_revision_list_json, policy_revision_to_json, proto_execution_timeout,
-        provisioning_timeout_message, ready_false_condition_message, relay_local_socket,
-        resolve_from, rootfs_tar_sources_supported_for_gateway, sandbox_should_persist,
-        sandbox_upload_plan, service_endpoint_to_json, service_status_error,
-        service_url_for_gateway, workspace_member_to_json,
+        format_endpoint, format_log_line, forward_ended_by_phase, forward_requires_session_token,
+        git_sync_files, has_main_process_result, parse_cli_setting_value,
+        parse_credential_expiry_cli_value, parse_driver_config_json, parse_env_from_pairs,
+        parse_secret_material_env_pairs, policy_revision_list_json, policy_revision_to_json,
+        proto_execution_timeout, provisioning_timeout_message, ready_false_condition_message,
+        relay_local_socket, resolve_from, rootfs_tar_sources_supported_for_gateway,
+        sandbox_should_persist, sandbox_upload_plan, service_endpoint_to_json,
+        service_status_error, service_url_for_gateway, workspace_member_to_json,
     };
     use openshell_core::proto::TcpForwardFrame;
 
@@ -9498,4 +9524,27 @@ mod tests {
             "authorization_token is required for ForwardTcp"
         )));
     }
+
+    #[test]
+    fn forward_leaves_reachability_after_exit_to_the_gateway() {
+        use openshell_core::proto::SandboxPhase;
+
+        for phase in [
+            SandboxPhase::Stopping,
+            SandboxPhase::Stopped,
+            SandboxPhase::Deleting,
+        ] {
+            assert!(forward_ended_by_phase(phase as i32), "{phase:?}");
+        }
+        for phase in [
+            SandboxPhase::Provisioning,
+            SandboxPhase::Starting,
+            SandboxPhase::Ready,
+            SandboxPhase::Completed,
+            SandboxPhase::Error,
+            SandboxPhase::Suspended,
+        ] {
+            assert!(!forward_ended_by_phase(phase as i32), "{phase:?}");
+        }
+    }
 }

crates/openshell-cli/tests/sandbox_exec_streaming_integration.rs

@@ -46,6 +46,7 @@ struct Calls {
 #[derive(Clone)]
 struct MockGateway {
     scenario: Scenario,
+    phase: proto::SandboxPhase,
     calls: Arc<Mutex<Calls>>,
 }
 
@@ -66,6 +67,20 @@ fn exit(code: i32) -> proto::ExecSandboxEvent {
 }
 
 impl MockGateway {
+    /// The gateway's access rule: only it decides whether a sandbox is usable.
+    fn require_reachable(&self) -> Result<(), Status> {
+        if matches!(
+            self.phase,
+            proto::SandboxPhase::Ready
+                | proto::SandboxPhase::Completed
+                | proto::SandboxPhase::Error
+        ) {
+            Ok(())
+        } else {
+            Err(Status::failed_precondition("sandbox is not ready"))
+        }
+    }
+
     async fn exchange(
         self,
         mut input: tonic::Streaming<proto::ExecSandboxInput>,
@@ -189,7 +204,7 @@ macro_rules! mock_gateway {
                             ..Default::default()
                         }),
                         status: Some(proto::SandboxStatus {
-                            phase: proto::SandboxPhase::Ready.into(),
+                            phase: self.phase.into(),
                             ..Default::default()
                         }),
                         ..Default::default()
@@ -200,12 +215,14 @@ macro_rules! mock_gateway {
 
             type ExecSandboxStream = EventStream;
             async fn exec_sandbox(&self, _: Request<proto::ExecSandboxRequest>) -> Result<Response<EventStream>, Status> {
+                self.require_reachable()?;
                 self.calls.lock().unwrap().unary += 1;
                 Err(Status::unimplemented("test requires streaming exec"))
             }
 
             type ExecSandboxInteractiveStream = EventStream;
             async fn exec_sandbox_interactive(&self, request: Request<tonic::Streaming<proto::ExecSandboxInput>>) -> Result<Response<EventStream>, Status> {
+                self.require_reachable()?;
                 let (sender, receiver) = mpsc::channel(1);
                 let service = self.clone();
                 tokio::spawn(async move {
@@ -326,6 +343,10 @@ impl Drop for TestGateway {
 
 impl TestGateway {
     async fn start(scenario: Scenario) -> Self {
+        Self::start_in_phase(scenario, proto::SandboxPhase::Ready).await
+    }
+
+    async fn start_in_phase(scenario: Scenario, phase: proto::SandboxPhase) -> Self {
         let (ca, ca_key) = build_ca();
         let (server_cert, server_key) = build_server_cert(&ca, &ca_key);
         let (client_cert, client_key) = build_client_cert(&ca, &ca_key);
@@ -346,6 +367,7 @@ impl TestGateway {
         let calls = Arc::new(Mutex::new(Calls::default()));
         let service = MockGateway {
             scenario,
+            phase,
             calls: Arc::clone(&calls),
         };
         let task = tokio::spawn(async move {
@@ -478,6 +500,47 @@ async fn default_remote_exit_does_not_wait_for_idle_open_stdin() {
     drop(held_open);
 }
 
+#[tokio::test]
+async fn exec_reaches_gateway_for_completed_and_error_sandboxes() {
+    // The supervisor keeps its access plane after the main process exits, so
+    // the gateway, not the CLI, decides whether the relay is still reachable.
+    for phase in [proto::SandboxPhase::Completed, proto::SandboxPhase::Error] {
+        let gateway = TestGateway::start_in_phase(Scenario::EarlyExit, phase).await;
+        let mut child = gateway.spawn();
+        let held_open = child.stdin.take().unwrap();
+        let output = finish(child).await;
+        drop(held_open);
+        assert!(
+            output.status.success(),
+            "{phase:?}: {}",
+            String::from_utf8_lossy(&output.stderr)
+        );
+        assert_eq!(output.stdout, b"before-exit\n");
+        gateway.assert_one_stream();
+    }
+}
+
+#[tokio::test]
+async fn exec_names_the_phase_when_the_gateway_refuses() {
+    let gateway =
+        TestGateway::start_in_phase(Scenario::EarlyExit, proto::SandboxPhase::Stopped).await;
+    let mut child = gateway.spawn();
+    drop(child.stdin.take());
+    let output = finish(child).await;
+    assert!(!output.status.success());
+    // The error renderer wraps long messages, so compare the words alone.
+    let stderr = String::from_utf8_lossy(&output.stderr)
+        .split_whitespace()
+        .filter(|word| *word != "│")
+        .collect::<Vec<_>>()
+        .join(" ");
+    assert!(
+        stderr.contains("is not ready (phase: Stopped); start it with `openshell sandbox start"),
+        "{stderr}"
+    );
+    assert!(gateway.calls.lock().unwrap().starts.is_empty());
+}
+
 #[tokio::test]
 async fn default_open_pipe_checks_trailer_error_after_exit() {
     let gateway = TestGateway::start(Scenario::ErrorAfterExit).await;

crates/openshell-server/src/compute/mod.rs

@@ -6963,7 +6963,7 @@ fn terminal_stop_exit_code(sandbox: &Sandbox) -> Option<i32> {
     sandbox.status.as_ref()?.exit_code
 }
 
-fn is_failed_main_process_result(sandbox: &Sandbox) -> bool {
+pub fn is_failed_main_process_result(sandbox: &Sandbox) -> bool {
     sandbox.phase() == SandboxPhase::Error as i32
         && sandbox.status.as_ref().is_some_and(|status| {
             status.exit_code.is_some()

crates/openshell-server/src/grpc/mutation_replay/streaming/tests.rs

@@ -167,6 +167,17 @@ async fn receive_relay(state: &ServerState, messages: &mut mpsc::Receiver<Gatewa
         .fail_pending_relay(&open.channel_id, "test relay ended".into());
 }
 
+async fn set_phase(state: &ServerState, req: &ExecSandboxRequest, phase: SandboxPhase) {
+    let mut sandbox = state
+        .store
+        .get_message_by_name::<Sandbox>("default", &req.sandbox)
+        .await
+        .unwrap()
+        .unwrap();
+    sandbox.set_phase(phase as i32);
+    state.store.put_message(&sandbox).await.unwrap();
+}
+
 async fn completion(state: &ServerState) -> Completion {
     let rows = state
         .store
@@ -210,6 +221,36 @@ fn interactive_fingerprint_uses_start_schema_and_ignores_request_id() {
     );
 }
 
+#[tokio::test]
+async fn exec_reaches_terminal_sandbox_only_while_supervisor_is_connected() {
+    // A supervisor retains its access plane after the main process exits.
+    // Exec follows the same reachability rule as SSH and port forwarding.
+    for interactive in [false, true] {
+        for phase in [SandboxPhase::Completed, SandboxPhase::Error] {
+            let directory = tempfile::tempdir().unwrap();
+            let state = setup("sqlite::memory:", &directory).await;
+            let req = request(&state).await;
+            set_phase(&state, &req, phase).await;
+            let (mut client, server) = wire_client(state.clone(), None, None).await;
+
+            let status = exec_rpc(&mut client, req.clone(), interactive)
+                .await
+                .unwrap_err();
+            assert_eq!(status.code(), tonic::Code::FailedPrecondition, "{phase:?}");
+            assert_eq!(status.message(), "sandbox is not ready");
+
+            let mut messages = register(&state, &sandbox_id(&state, &req).await);
+            let mut retry = req;
+            retry.request_id = uuid::Uuid::new_v4().to_string();
+            exec_rpc(&mut client, retry, interactive)
+                .await
+                .unwrap_or_else(|status| panic!("{phase:?} interactive={interactive}: {status}"));
+            receive_relay(&state, &mut messages).await;
+            server.abort();
+        }
+    }
+}
+
 #[tokio::test]
 async fn concurrent_exec_claims_open_one_relay_and_store_no_secrets() {
     let directory = tempfile::tempdir().unwrap();