fix(tui): avoid tui shutdown on unrecognized input (#9289)

### Description

This PR does 3 things:
- Fix run summary from being printed out during watch mode which
corrupts the TUI
- Adds additional debug logs recording the reason for the TUI shutting
down
- Fixes bug where `poll` could now return `None` even if input streams
were still available

The bug came down to:
 - A crossterm event was received
- Translating the crossterm event via
`input_options.handle_crossterm_event` returned `None` (This makes total
sense, not every terminal event requires a response from the TUI)
- This `None` was returned instead of waiting for more input or a tick
event
- The `while let Some(event) = poll(...)` loop would exit which triggers
a TUI shut down

### Testing Instructions

`npx create-turbo@latest` and then start up the TUI `turbo dev
--ui=tui`, without hitting `Enter` to focus the task hit `q` multiple
times and you should see the TUI shut down.

Use `turbo` built from this PR and try the same thing. The TUI should
not shut down on the press of `q`.

crates/turborepo-lib/src/run/summary/mod.rs

@@ -382,7 +382,7 @@ impl<'a> RunSummary<'a> {
         }
 
         if let Some(spaces_client_handle) = self.spaces_client_handle.take() {
-            self.send_to_space(spaces_client_handle, end_time, exit_code)
+            self.send_to_space(spaces_client_handle, end_time, exit_code, is_watch)
                 .await;
         }
 
@@ -395,26 +395,35 @@ impl<'a> RunSummary<'a> {
         spaces_client_handle: SpacesClientHandle,
         ended_at: DateTime<Local>,
         exit_code: i32,
+        is_watch: bool,
     ) {
-        let spinner = tokio::spawn(async {
-            tokio::time::sleep(std::time::Duration::from_millis(1000)).await;
-            turborepo_ui::start_spinner("...sending run summary...");
+        let spinner = (!is_watch).then(|| {
+            tokio::spawn(async {
+                tokio::time::sleep(std::time::Duration::from_millis(1000)).await;
+                turborepo_ui::start_spinner("...sending run summary...");
+            })
         });
 
         // We log the error here but don't fail because
         // failing to send the space shouldn't fail the run.
         if let Err(err) = spaces_client_handle.finish_run(exit_code, ended_at).await {
-            warn!("Error sending to space: {}", err);
+            if !is_watch {
+                warn!("Error sending to space: {}", err);
+            }
         };
 
         let result = spaces_client_handle.close().await;
 
-        spinner.abort();
+        if let Some(spinner) = spinner {
+            spinner.abort();
+        }
 
-        Self::print_errors(&result.errors);
+        if !is_watch {
+            Self::print_errors(&result.errors);
 
-        if let Some(run) = result.run {
-            println!("Run: {}\n", run.url);
+            if let Some(run) = result.run {
+                println!("Run: {}\n", run.url);
+            }
         }
     }
 

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

@@ -101,7 +101,7 @@ pub enum Error {
     PackageChange(#[from] tonic::Status),
     #[error(transparent)]
     UI(#[from] turborepo_ui::Error),
-    #[error("could not connect to UI thread")]
+    #[error("could not connect to UI thread: {0}")]
     UISend(String),
     #[error("cannot use root turbo.json at {0} with watch mode")]
     NonStandardTurboJsonPath(String),
@@ -317,7 +317,7 @@ impl WatchClient {
                     let task_names = run.engine.tasks_with_command(&run.pkg_dep_graph);
                     sender
                         .restart_tasks(task_names)
-                        .map_err(|err| Error::UISend(err.to_string()))?;
+                        .map_err(|err| Error::UISend(format!("some packages changed: {err}")))?;
                 }
 
                 let ui_sender = self.ui_sender.clone();
@@ -371,7 +371,7 @@ impl WatchClient {
                     let task_names = self.run.engine.tasks_with_command(&self.run.pkg_dep_graph);
                     sender
                         .update_tasks(task_names)
-                        .map_err(|err| Error::UISend(err.to_string()))?;
+                        .map_err(|err| Error::UISend(format!("all packages changed {err}")))?;
                 }
 
                 if self.run.has_non_interruptible_tasks() {

crates/turborepo-ui/src/tui/app.rs

@@ -568,7 +568,10 @@ pub async fn run_app(tasks: Vec<String>, receiver: AppReceiver) -> Result<(), Er
     let (result, callback) =
         match run_app_inner(&mut terminal, &mut app, receiver, crossterm_rx).await {
             Ok(callback) => (Ok(()), callback),
-            Err(err) => (Err(err), None),
+            Err(err) => {
+                debug!("tui shutting down: {err}");
+                (Err(err), None)
+            }
         };
 
     cleanup(terminal, app, callback)?;
@@ -634,25 +637,28 @@ async fn poll<'a>(
     crossterm_rx: &mut mpsc::Receiver<crossterm::event::Event>,
 ) -> Option<Event> {
     let input_closed = crossterm_rx.is_closed();
-    let input_fut = async {
-        crossterm_rx
-            .recv()
-            .await
-            .and_then(|event| input_options.handle_crossterm_event(event))
-    };
-    let receiver_fut = async { receiver.recv().await };
-    let event_fut = async move {
-        if input_closed {
-            receiver_fut.await
-        } else {
+
+    if input_closed {
+        receiver.recv().await
+    } else {
+        // tokio::select is messing with variable read detection
+        #[allow(unused_assignments)]
+        let mut event = None;
+        loop {
             tokio::select! {
-                e = input_fut => e,
-                e = receiver_fut => e,
+                e = crossterm_rx.recv() => {
+                    event = e.and_then(|e| input_options.handle_crossterm_event(e));
+                }
+                e = receiver.recv() => {
+                    event = e;
+                }
+            }
+            if event.is_some() {
+                break;
             }
         }
-    };
-
-    event_fut.await
+        event
+    }
 }
 
 const MIN_HEIGHT: u16 = 10;
@@ -729,9 +735,11 @@ fn update(
             app.set_status(task, status, result)?;
         }
         Event::InternalStop => {
+            debug!("shutting down due to internal failure");
             app.done = true;
         }
         Event::Stop(callback) => {
+            debug!("shutting down due to message");
             app.done = true;
             return Ok(Some(callback));
         }

crates/turborepo-ui/src/tui/input.rs

@@ -1,6 +1,7 @@
 use crossterm::event::{EventStream, KeyCode, KeyEvent, KeyEventKind, KeyModifiers};
 use futures::StreamExt;
 use tokio::{sync::mpsc, task::JoinHandle};
+use tracing::debug;
 
 use super::{
     app::LayoutSections,
@@ -122,7 +123,10 @@ fn ctrl_c() -> Option<Event> {
     match signal::raise(signal::SIGINT) {
         Ok(_) => None,
         // We're unable to send the signal, stop rendering to force shutdown
-        Err(_) => Some(Event::InternalStop),
+        Err(_) => {
+            debug!("unable to send sigint, shutting down");
+            Some(Event::InternalStop)
+        }
     }
 }
 
@@ -146,6 +150,7 @@ fn ctrl_c() -> Option<Event> {
         None
     } else {
         // We're unable to send the Ctrl-C event, stop rendering to force shutdown
+        debug!("unable to send sigint, shutting down");
         Some(Event::InternalStop)
     }
 }