Ensure tower-runtime setup failures surface a terminal status

crates/tower-runtime/src/local.rs

@@ -49,6 +49,31 @@ pub struct LocalApp {
     execute_handle: Option<JoinHandle<Result<(), Error>>>,
 }
 
+// Sent when execute_local_app returns Err before producing a real exit code
+// (uv not found, venv spawn failed, PYTHONPATH construction failed, etc).
+// Distinct from the cancellation sentinel -1.
+pub(crate) const SETUP_FAILURE_EXIT_CODE: i32 = -2;
+
+// Drop guard that ensures the oneshot waiter always receives a code, so `?`
+// paths in execute_local_app cannot leave status() stuck on WaiterClosed.
+struct StatusReporter {
+    tx: Option<oneshot::Sender<i32>>,
+}
+
+impl StatusReporter {
+    fn send(&mut self, code: i32) {
+        if let Some(tx) = self.tx.take() {
+            let _ = tx.send(code);
+        }
+    }
+}
+
+impl Drop for StatusReporter {
+    fn drop(&mut self) {
+        self.send(SETUP_FAILURE_EXIT_CODE);
+    }
+}
+
 // Helper function to check if a file is executable
 async fn is_executable(path: &PathBuf) -> bool {
     let metadata = match fs::metadata(path).await {
@@ -104,6 +129,7 @@ async fn execute_local_app(
     sx: oneshot::Sender<i32>,
     cancel_token: CancellationToken,
 ) -> Result<(), Error> {
+    let mut reporter = StatusReporter { tx: Some(sx) };
     let ctx = opts.ctx.clone();
     let package = opts.package;
     let environment = opts.environment;
@@ -159,7 +185,7 @@ async fn execute_local_app(
     if cancel_token.is_cancelled() {
         // if there's a waiter, we want them to know that the process was cancelled so we have
         // to return something on the relevant channel.
-        let _ = sx.send(-1);
+        reporter.send(-1);
         return Err(Error::Cancelled);
     }
 
@@ -176,7 +202,7 @@ async fn execute_local_app(
         )
         .await?;
 
-        let _ = sx.send(wait_for_process(ctx.clone(), &cancel_token, child).await);
+        reporter.send(wait_for_process(ctx.clone(), &cancel_token, child).await);
     } else {
         // we put Uv in to protected mode when there's no caching configured/enabled.
         let protected_mode = opts.cache_dir.is_none();
@@ -198,7 +224,7 @@ async fn execute_local_app(
         // ensure everything is in place.
         if cancel_token.is_cancelled() {
             // again tell any waiters that we cancelled.
-            let _ = sx.send(-1);
+            reporter.send(-1);
             return Err(Error::Cancelled);
         }
 
@@ -226,15 +252,15 @@ async fn execute_local_app(
 
         if res != 0 {
             // If the venv process failed, we want to return an error.
-            let _ = sx.send(res);
+            reporter.send(res);
             return Err(Error::VirtualEnvCreationFailed);
         }
 
         // Check once more if the process was cancelled before we do a uv sync. The sync itself,
         // once started, will take a while and we have logic for checking for cancellation.
         if cancel_token.is_cancelled() {
             // again tell any waiters that we cancelled.
-            let _ = sx.send(-1);
+            reporter.send(-1);
             return Err(Error::Cancelled);
         }
 
@@ -279,7 +305,7 @@ async fn execute_local_app(
 
                 if res != 0 {
                     // If the sync process failed, we want to return an error.
-                    let _ = sx.send(res);
+                    reporter.send(res);
                     return Err(Error::DependencyInstallationFailed);
                 }
             }
@@ -289,7 +315,7 @@ async fn execute_local_app(
         if cancel_token.is_cancelled() {
             // if there's a waiter, we want them to know that the process was cancelled so we have
             // to return something on the relevant channel.
-            let _ = sx.send(-1);
+            reporter.send(-1);
             return Err(Error::Cancelled);
         }
 
@@ -312,7 +338,7 @@ async fn execute_local_app(
             BufReader::new(stderr),
         ));
 
-        let _ = sx.send(wait_for_process(ctx.clone(), &cancel_token, child).await);
+        reporter.send(wait_for_process(ctx.clone(), &cancel_token, child).await);
     }
 
     // Everything was properly executed I suppose.

crates/tower-runtime/tests/local_test.rs

@@ -4,7 +4,8 @@ use std::path::PathBuf;
 use tower_runtime::{local::LocalApp, App, StartOptions, Status};
 
 use config::Towerfile;
-use tower_package::{Package, PackageSpec};
+use tmpdir::TmpDir;
+use tower_package::{Manifest, Package, PackageSpec};
 use tower_telemetry::{self, debug};
 
 use tokio::sync::mpsc::unbounded_channel;
@@ -329,6 +330,69 @@ async fn test_running_app_with_secret() {
     assert!(status == Status::Exited, "App should be running");
 }
 
+#[cfg(unix)]
+#[tokio::test]
+async fn test_setup_failure_reports_status_instead_of_waiter_closed() {
+    // Regression: execute_local_app used to bubble up `?` errors (e.g. a failing
+    // std::env::join_paths) without ever signalling the oneshot waiter, so
+    // LocalApp::status() returned Err(WaiterClosed) and callers exited 1
+    // silently. A setup failure must now surface as a terminal Status.
+    let tmp = TmpDir::new("test-silent-setup")
+        .await
+        .expect("Failed to create tmp dir");
+    let unpacked_path: PathBuf = tmp.as_ref().to_path_buf();
+
+    // On Unix, std::env::join_paths rejects paths containing ':'. Feeding one
+    // into manifest.import_paths causes the PYTHONPATH construction inside
+    // execute_local_app to fail via `?` before any subprocess is spawned.
+    let package = Package {
+        tmp_dir: None,
+        package_file_path: None,
+        unpacked_path: Some(unpacked_path),
+        manifest: Manifest {
+            version: Some(1),
+            invoke: "main.py".to_string(),
+            parameters: vec![],
+            schedule: None,
+            import_paths: vec!["bad:path".to_string()],
+            app_dir_name: "app".to_string(),
+            modules_dir_name: "modules".to_string(),
+            checksum: "".to_string(),
+        },
+    };
+
+    let (sender, _receiver) = unbounded_channel();
+    let opts = StartOptions {
+        ctx: tower_telemetry::Context::new("runner-id".to_string()),
+        package,
+        output_sender: sender,
+        cwd: None,
+        environment: "local".to_string(),
+        secrets: HashMap::new(),
+        parameters: HashMap::new(),
+        env_vars: HashMap::new(),
+        cache_dir: Some(config::default_cache_dir()),
+    };
+
+    let app = LocalApp::start(opts).await.expect("Failed to start app");
+
+    for _ in 0..20 {
+        let status = app.status().await.expect("status should not be WaiterClosed");
+        if status.is_terminal() {
+            match status {
+                Status::Crashed { code } => {
+                    assert!(code < 0, "expected negative sentinel, got {}", code);
+                    return;
+                }
+                other => panic!("expected Crashed, got {:?}", other),
+            }
+        }
+        tokio::time::sleep(std::time::Duration::from_millis(25)).await;
+    }
+
+    panic!("status never reached a terminal state");
+}
+
 #[tokio::test]
 async fn test_abort_on_dependency_installation_failure() {
     debug!("Running 05-broken-dependencies");