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

### Description

I noticed in #9389 that the
TUI wasn't exiting if `watch` exited from anything other than a `Ctrl-C`
(including a direct SIGINT to the process). This PR ensures that we shut
down the TUI and the persistent task handle regardless of how
`watch.start()` exits.

### Testing Instructions

- Start up a watch task `turbo_dev watch dev`
- Find the pid file via `turbo_dev daemon status`
- Kill the daemon `kill $(cat /path/to/pid/file)`

Before
<img width="1365" alt="Screenshot 2024-11-07 at 5 42 22 PM"
src="https://github.com/user-attachments/assets/141831b4-8a82-485e-ac14-c0d72802cf81">


After
<img width="1286" alt="Screenshot 2024-11-07 at 5 39 08 PM"
src="https://github.com/user-attachments/assets/21acddb9-ae6a-4b78-b38d-1cbe0e0d3309">

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