@@ -801,11 +801,8 @@ pub async fn sandbox_create(
     // Non-interactive mode: track start time for timestamps.
     let provision_start = Instant::now();
 
-    // Don't use stop_on_terminal on the server — the Kubernetes CRD may
-    // briefly report a stale Ready status before the controller reconciles
-    // a newly created sandbox.  Instead we handle termination client-side:
-    // we wait until we have observed at least one non-Ready phase followed
-    // by Ready (a genuine Provisioning → Ready transition).
+    // Handle terminal states here so a provisional container exit can wait
+    // for the supervisor's canonical-process result before cleanup.
     let sandbox_name = sandbox.object_name().to_string();
     let sandbox_workspace = sandbox.object_workspace().to_string();
     let mut stream = client
@@ -833,8 +830,6 @@ pub async fn sandbox_create(
     let mut last_sandbox = sandbox.clone();
     let mut last_error_reason = String::new();
     let mut last_condition_message = ready_false_condition_message(sandbox.status.as_ref());
-    // Track whether we have seen a non-Ready phase during the watch.
-    let mut saw_non_ready = SandboxPhase::try_from(sandbox.phase()) != Ok(SandboxPhase::Ready);
     let provision_timeout = Duration::from_secs(
         std::env::var("OPENSHELL_PROVISION_TIMEOUT")
             .ok()
@@ -912,10 +907,6 @@ pub async fn sandbox_create(
                     last_condition_message = Some(message);
                 }
 
-                if phase != SandboxPhase::Ready {
-                    saw_non_ready = true;
-                }
-
                 let main_process_result = has_main_process_result(&s);
                 if matches!(
                     phase,
@@ -950,9 +941,10 @@ pub async fn sandbox_create(
                     break;
                 }
 
-                // Only accept Ready as terminal after we've observed a
-                // non-Ready phase, proving the controller has reconciled.
-                if saw_non_ready && phase == SandboxPhase::Ready {
+                // The gateway owns readiness. Its initial watch snapshot may
+                // already be Ready if provisioning finished before CREATE
+                // returned; requiring an earlier phase would miss that state.
+                if phase == SandboxPhase::Ready {
                     if let Some(d) = display.as_interactive_mut() {
                         d.clear();
                     }
 
@@ -70,6 +70,7 @@ struct SandboxState {
     vm_error_with_observed_exit: Arc<AtomicBool>,
     vm_slow_progress_before_ready: Arc<AtomicBool>,
     vm_log_churn_before_ready: Arc<AtomicBool>,
+    ready_before_create_returns: Arc<AtomicBool>,
     terminal_before_relay: Arc<AtomicBool>,
     terminal_after_provisional_container_exit: Arc<AtomicBool>,
     provisional_container_exit_without_result: Arc<AtomicBool>,
@@ -194,7 +195,17 @@ impl OpenShell for TestOpenShell {
             }),
             ..Sandbox::default()
         };
-        sandbox.set_phase(SandboxPhase::Provisioning as i32);
+        sandbox.set_phase(
+            if self
+                .state
+                .ready_before_create_returns
+                .load(Ordering::SeqCst)
+            {
+                SandboxPhase::Ready as i32
+            } else {
+                SandboxPhase::Provisioning as i32
+            },
+        );
         Ok(Response::new(SandboxResponse {
             sandbox: Some(sandbox),
             service_urls,
@@ -677,6 +688,10 @@ impl OpenShell for TestOpenShell {
             .vm_slow_progress_before_ready
             .load(Ordering::SeqCst);
         let vm_log_churn_before_ready = self.state.vm_log_churn_before_ready.load(Ordering::SeqCst);
+        let ready_before_create_returns = self
+            .state
+            .ready_before_create_returns
+            .load(Ordering::SeqCst);
         let terminal_before_relay = self.state.terminal_before_relay.load(Ordering::SeqCst);
         let terminal_after_provisional_container_exit = self
             .state
@@ -723,6 +738,18 @@ impl OpenShell for TestOpenShell {
             }
             let mut ready = provisioning.clone();
             ready.set_phase(SandboxPhase::Ready as i32);
+            if ready_before_create_returns {
+                // A watch starts with the current snapshot. Keep it open so
+                // stream closure cannot hide a client that ignores Ready.
+                let _ = tx
+                    .send(Ok(SandboxStreamEvent {
+                        payload: Some(sandbox_stream_event::Payload::Sandbox(ready)),
+                        cursor: String::new(),
+                    }))
+                    .await;
+                tx.closed().await;
+                return;
+            }
             let mut completed = provisioning.clone();
             completed.status = Some(SandboxStatus {
                 exit_code: Some(0),
@@ -2372,6 +2399,55 @@ async fn sandbox_create_preserves_vm_error_when_exit_code_is_observed() {
     assert!(rendered.contains("ProcessExited: VM process exited with status 0"));
 }
 
+#[tokio::test]
+async fn sandbox_create_accepts_ready_before_create_returns() {
+    let server = run_server().await;
+    server
+        .openshell
+        .state
+        .ready_before_create_returns
+        .store(true, Ordering::SeqCst);
+    let fake_ssh_dir = tempfile::tempdir().unwrap();
+    let xdg_dir = tempfile::tempdir().unwrap();
+    let _env = test_env_with(
+        &fake_ssh_dir,
+        &xdg_dir,
+        &[("OPENSHELL_PROVISION_TIMEOUT", "1".to_string())],
+    );
+    let tls = test_tls(&server);
+    install_fake_ssh(&fake_ssh_dir);
+
+    let exit_code = tokio::time::timeout(
+        Duration::from_secs(10),
+        run::sandbox_create(
+            &server.endpoint,
+            "openshell",
+            run::SandboxCreateConfig {
+                name: Some("already-ready"),
+                command: &["echo".into(), "OK".into()],
+                ..test_config()
+            },
+            "default",
+            &tls,
+        ),
+    )
+    .await
+    .expect("creation must finish while the watch remains open")
+    .expect("an already-Ready sandbox must not wait for a new provisioning transition");
+
+    assert_eq!(exit_code, 0);
+    assert_eq!(create_requests(&server).await.len(), 1);
+    assert_eq!(
+        server
+            .openshell
+            .state
+            .ssh_session_requests
+            .load(Ordering::SeqCst),
+        1,
+        "the initial Ready snapshot must allow the command to attach"
+    );
+}
+
 #[tokio::test]
 async fn sandbox_create_keeps_waiting_while_vm_progress_arrives() {
     let server = run_server().await;
 
@@ -8419,7 +8419,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn create_compensation_deletes_backend_when_delete_transition_cannot_be_stored() {
+    async fn create_compensation_retains_backend_when_delete_transition_cannot_be_stored() {
         let directory = tempfile::tempdir().expect("temporary database directory");
         let database_url = format!("sqlite://{}", directory.path().join("gateway.db").display());
         let store = Arc::new(Store::connect(&database_url).await.expect("connect store"));
@@ -8462,14 +8462,10 @@ mod tests {
                 .message()
                 .contains("could not claim the sandbox record")
         );
-        assert_eq!(driver.delete_calls(), 1);
-        assert_eq!(
-            driver.delete_requests(),
-            vec![(
-                sandbox.object_id().to_string(),
-                sandbox.object_name().to_string()
-            )]
-        );
+        // A rejected durable claim leaves cleanup ownership unknown. An
+        // unclaimed DELETE could destroy newer compute with the same ID.
+        assert_eq!(driver.delete_calls(), 0);
+        assert!(driver.delete_requests().is_empty());
         let retained = runtime
             .store
             .get_message::<Sandbox>(sandbox.object_id())
 
@@ -12,6 +12,7 @@ use std::time::Duration;
 
 use openshell_core::{ObjectId, proto::Sandbox};
 use tonic::Status;
+use tracing::Instrument as _;
 
 use super::{
     ComputeRuntime, provisioning_deadline, sandbox_provisioning_attempt_id,
@@ -127,7 +128,7 @@ impl ComputeRuntime {
         let owned_attempt = starting.clone();
         // Dropping a JoinHandle detaches its task. Do not abort it on a monitor
         // error or deadline: the remote side may still commit the request.
-        let mut worker = tokio::spawn(async move {
+        let worker = async move {
             let result = operation.await;
             let settled = if tracked {
                 let settled = runtime
@@ -155,7 +156,10 @@ impl ComputeRuntime {
                     settled: Box::new(settled),
                 }),
             }
-        });
+        };
+        // Retain the request span so detaching ownership preserves the driver
+        // call's parent trace, including when the caller stops waiting.
+        let mut worker = tokio::spawn(worker.in_current_span());
         loop {
             tokio::select! {
                 result = &mut worker => return result.map_err(|error| {