fix(watch): ensure TUI is shutdown regardless of exit path

crates/turborepo-lib/src/cli/mod.rs

@@ -1391,7 +1391,10 @@ pub async fn run(
             let base = CommandBase::new(cli_args, repo_root, version, color_config);
 
             let mut client = WatchClient::new(base, event).await?;
-            client.start().await?;
+            if let Err(e) = client.start().await {
+                client.shutdown().await;
+                return Err(e.into());
+            }
             // We only exit if we get a signal, so we return a non-zero exit code
             return Ok(1);
         }

crates/turborepo-lib/src/process/child.rs

@@ -464,7 +464,7 @@ impl Child {
             // On Windows it is important that this gets dropped once the child process
             // exits
             let controller = controller;
-            debug!("waiting for task");
+            debug!("waiting for task: {pid:?}");
             let manager = ChildStateManager {
                 shutdown_style,
                 task_state,

crates/turborepo-lib/src/run/watch.rs

@@ -217,13 +217,6 @@ impl WatchClient {
             biased;
             _ = signal_subscriber.listen() => {
                 tracing::info!("shutting down");
-                if let Some(RunHandle { stopper, run_task }) = self.persistent_tasks_handle.take() {
-                    // Shut down the tasks for the run
-                    stopper.stop().await;
-                    // Run should exit shortly after we stop all child tasks, wait for it to finish
-                    // to ensure all messages are flushed.
-                    let _ = run_task.await;
-                }
                 Err(Error::SignalInterrupt)
             }
             result = event_fut => {
@@ -267,6 +260,20 @@ impl WatchClient {
         Ok(())
     }
 
+    /// Shut down any resources that run as part of watch.
+    pub async fn shutdown(&mut self) {
+        if let Some(sender) = &self.ui_sender {
+            sender.stop().await;
+        }
+        if let Some(RunHandle { stopper, run_task }) = self.persistent_tasks_handle.take() {
+            // Shut down the tasks for the run
+            stopper.stop().await;
+            // Run should exit shortly after we stop all child tasks, wait for it to finish
+            // to ensure all messages are flushed.
+            let _ = run_task.await;
+        }
+    }
+
     /// Executes a run with the given changed packages. Splits the run into two
     /// parts:
     /// 1. The persistent tasks that are not allowed to be interrupted