fix(tui): avoid panic on restarting tasks during watch (#9032)

### Description

Fixes #9016

Crash was caused by two issues:
- Mistakenly calling `update_tasks` instead of `restart_tasks` when
handling a restart event 🙃 . This meant if a package changed that did
not matter (a package in scope, but that didn't have a task definited)
we would "restart" no tasks, but this would end up clearing out the
entire task list.
- The watch client can declare it is restarting a task that is not in
the current list. This happens if a `build` script is added to a
workspace `package.json` after `turbo watch build` is run.

### Testing Instructions

Added unit tests that test we no longer crash when:
 - updating task list to nothing (TUI cannot display this)
 - restarting tasks that were not in original list

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

@@ -23,7 +23,7 @@ pub use cache::{CacheOutput, ConfigCache, Error as CacheError, RunCache, TaskCac
 use chrono::{DateTime, Local};
 use rayon::iter::ParallelBridge;
 use tokio::{select, task::JoinHandle};
-use tracing::debug;
+use tracing::{debug, instrument};
 use turbopath::AbsoluteSystemPathBuf;
 use turborepo_api_client::{APIAuth, APIClient};
 use turborepo_ci::Vendor;
@@ -130,6 +130,7 @@ impl Run {
 
     // Produces the transitive closure of the filtered packages,
     // i.e. the packages relevant for this run.
+    #[instrument(skip(self), ret)]
     pub fn get_relevant_packages(&self) -> HashSet<PackageName> {
         let packages: Vec<_> = self
             .filtered_pkgs

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

@@ -8,6 +8,7 @@ use tokio::{
     sync::{Mutex, Notify},
     task::JoinHandle,
 };
+use tracing::{instrument, trace};
 use turborepo_repository::package_graph::PackageName;
 use turborepo_telemetry::events::command::CommandEventBuilder;
 use turborepo_ui::{tui, tui::AppSender};
@@ -200,6 +201,7 @@ impl WatchClient {
         }
     }
 
+    #[instrument(skip(changed_packages))]
     async fn handle_change_event(
         changed_packages: &Mutex<RefCell<ChangedPackages>>,
         event: proto::package_change_event::Event,
@@ -233,6 +235,7 @@ impl WatchClient {
 
     async fn execute_run(&mut self, changed_packages: ChangedPackages) -> Result<i32, Error> {
         // Should we recover here?
+        trace!("handling run with changed packages: {changed_packages:?}");
         match changed_packages {
             ChangedPackages::Some(packages) => {
                 let packages = packages

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

@@ -242,6 +242,10 @@ impl<W> App<W> {
 
     #[tracing::instrument(skip(self))]
     pub fn update_tasks(&mut self, tasks: Vec<String>) {
+        if tasks.is_empty() {
+            debug!("got request to update task list to empty list, ignoring request");
+            return;
+        }
         debug!("updating task list: {tasks:?}");
         let highlighted_task = self.active_task().to_owned();
         // Make sure all tasks have a terminal output
@@ -615,7 +619,7 @@ fn update(
             app.copy_selection();
         }
         Event::RestartTasks { tasks } => {
-            app.update_tasks(tasks);
+            app.restart_tasks(tasks);
         }
         Event::Resize { rows, cols } => {
             app.resize(rows, cols);
@@ -954,4 +958,32 @@ mod test {
             );
         }
     }
+
+    #[test]
+    fn test_update_empty_task_list() {
+        let mut app: App<()> = App::new(
+            100,
+            100,
+            vec!["a".to_string(), "b".to_string(), "c".to_string()],
+        );
+        app.next();
+        app.update_tasks(Vec::new());
+
+        assert_eq!(app.active_task(), "b", "selected b");
+    }
+
+    #[test]
+    fn test_restart_missing_task() {
+        let mut app: App<()> = App::new(
+            100,
+            100,
+            vec!["a".to_string(), "b".to_string(), "c".to_string()],
+        );
+        app.next();
+        app.restart_tasks(vec!["d".to_string()]);
+
+        assert_eq!(app.active_task(), "b", "selected b");
+
+        app.start_task("d", OutputLogs::Full).unwrap();
+    }
 }

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

@@ -154,7 +154,7 @@ impl TasksByStatus {
     }
 
     pub fn restart_tasks<'a>(&mut self, tasks: impl Iterator<Item = &'a str>) {
-        let tasks_to_restart = tasks.collect::<HashSet<_>>();
+        let mut tasks_to_restart = tasks.collect::<HashSet<_>>();
 
         let (restarted_running, keep_running): (Vec<_>, Vec<_>) = mem::take(&mut self.running)
             .into_iter()
@@ -171,6 +171,17 @@ impl TasksByStatus {
                 .map(|task| task.restart())
                 .chain(restarted_finished.into_iter().map(|task| task.restart())),
         );
+        // There is a chance that watch might attempt to restart a task that did not
+        // exist before.
+        for task in &self.planned {
+            tasks_to_restart.remove(task.name());
+        }
+        self.planned.extend(
+            tasks_to_restart
+                .into_iter()
+                .map(ToOwned::to_owned)
+                .map(Task::new),
+        );
         self.planned.sort_unstable();
     }
 }