Trying to get experimental UI working with watch mode

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

@@ -1,5 +1,6 @@
 use std::future::Future;
 
+use tracing::error;
 use turborepo_telemetry::events::command::CommandEventBuilder;
 
 use crate::{commands::CommandBase, run, run::builder::RunBuilder, signal::SignalHandler};
@@ -38,16 +39,30 @@ pub async fn run(base: CommandBase, telemetry: CommandEventBuilder) -> Result<i3
 
     let run_fut = async {
         let (analytics_sender, analytics_handle) = run_builder.start_analytics();
+
         let run = run_builder
             .with_analytics_sender(analytics_sender)
             .build(&handler, telemetry)
             .await?;
-        let result = run.run().await;
+
+        let (sender, handle) = run
+            .has_experimental_ui()
+            .then(|| run.start_experimental_ui())
+            .unzip();
+
+        let result = run.run(sender.clone()).await;
 
         if let Some(analytics_handle) = analytics_handle {
             analytics_handle.close_with_timeout().await;
         }
 
+        if let (Some(handle), Some(sender)) = (handle, sender) {
+            sender.stop();
+            if let Err(e) = handle.await.expect("render thread panicked") {
+                error!("error encountered rendering tui: {e}");
+            }
+        }
+
         result
     };
 

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

@@ -402,7 +402,6 @@ impl RunBuilder {
         Ok(Run {
             version: self.version,
             ui: self.ui,
-            experimental_ui: self.experimental_ui,
             start_at,
             processes: self.processes,
             run_telemetry,
@@ -420,6 +419,7 @@ impl RunBuilder {
             run_cache,
             signal_handler: signal_handler.clone(),
             should_print_prelude,
+            experimental_ui: self.experimental_ui,
         })
     }
 

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

@@ -17,6 +17,7 @@ use std::{collections::HashSet, io::Write, sync::Arc};
 pub use cache::{ConfigCache, RunCache, TaskCache};
 use chrono::{DateTime, Local};
 use rayon::iter::ParallelBridge;
+use tokio::task::JoinHandle;
 use tracing::debug;
 use turbopath::AbsoluteSystemPathBuf;
 use turborepo_api_client::{APIAuth, APIClient};
@@ -25,7 +26,7 @@ use turborepo_env::EnvironmentVariableMap;
 use turborepo_repository::package_graph::{PackageGraph, PackageName};
 use turborepo_scm::SCM;
 use turborepo_telemetry::events::generic::GenericEventBuilder;
-use turborepo_ui::{cprint, cprintln, BOLD_GREY, GREY, UI};
+use turborepo_ui::{cprint, cprintln, tui, tui::AppSender, BOLD_GREY, GREY, UI};
 
 pub use crate::run::error::Error;
 use crate::{
@@ -43,7 +44,6 @@ use crate::{
 pub struct Run {
     version: &'static str,
     ui: UI,
-    experimental_ui: bool,
     start_at: DateTime<Local>,
     processes: ProcessManager,
     run_telemetry: GenericEventBuilder,
@@ -61,6 +61,7 @@ pub struct Run {
     engine: Arc<Engine>,
     task_access: TaskAccess,
     should_print_prelude: bool,
+    experimental_ui: bool,
 }
 
 impl Run {
@@ -98,7 +99,19 @@ impl Run {
         }
     }
 
-    pub async fn run(&self) -> Result<i32, Error> {
+    pub fn has_experimental_ui(&self) -> bool {
+        self.experimental_ui
+    }
+
+    pub fn start_experimental_ui(&self) -> (AppSender, JoinHandle<Result<(), tui::Error>>) {
+        let task_names = self.engine.tasks_with_command(&self.pkg_dep_graph);
+        let (sender, receiver) = AppSender::new();
+        let handle = tokio::task::spawn_blocking(move || tui::run_app(task_names, receiver));
+
+        (sender, handle)
+    }
+
+    pub async fn run(&self, experimental_ui_sender: Option<AppSender>) -> Result<i32, Error> {
         if self.should_print_prelude {
             self.print_run_prelude();
         }
@@ -220,7 +233,7 @@ impl Run {
             self.processes.clone(),
             &self.repo_root,
             global_env,
-            self.experimental_ui,
+            experimental_ui_sender,
         );
 
         if self.opts.run_opts.dry_run.is_some() {

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

@@ -6,6 +6,7 @@ use thiserror::Error;
 use tokio::{select, task::JoinHandle};
 use turborepo_repository::package_graph::PackageName;
 use turborepo_telemetry::events::command::CommandEventBuilder;
+use turborepo_ui::tui::AppSender;
 
 use crate::{
     cli::{Command, ExecutionArgs, RunArgs},
@@ -78,6 +79,8 @@ impl WatchClient {
             .build(&handler, telemetry.clone())
             .await?;
 
+        let (sender, handle) = run.start_experimental_ui();
+
         run.print_run_prelude();
 
         let has_persistent_tasks = run.has_persistent_tasks();
@@ -107,6 +110,7 @@ impl WatchClient {
                     &handler,
                     &mut main_run_handle,
                     has_persistent_tasks,
+                    Some(sender.clone()),
                 )
                 .await?;
             }
@@ -137,6 +141,7 @@ impl WatchClient {
         handler: &SignalHandler,
         main_run_handle: &mut Option<JoinHandle<Result<i32, run::error::Error>>>,
         has_persistent_tasks: bool,
+        ui_sender: Option<AppSender>,
     ) -> Result<(), Error> {
         // Should we recover here?
         match event {
@@ -185,7 +190,7 @@ impl WatchClient {
                             .build(&signal_handler, telemetry)
                             .await?;
 
-                        run.run().await
+                        run.run(ui_sender.clone()).await
                     }),
                 );
             }
@@ -217,13 +222,12 @@ impl WatchClient {
                 let handler = handler.clone();
                 let run_fut = async move {
                     let base = CommandBase::new(args, repo_root, get_version(), ui);
-
                     let run = RunBuilder::new(base)?
                         .hide_prelude()
                         .build(&handler, telemetry)
                         .await?;
 
-                    run.run().await
+                    run.run(ui_sender.clone()).await
                 };
 
                 if has_persistent_tasks {

crates/turborepo-lib/src/task_graph/visitor.rs

@@ -23,7 +23,7 @@ use turborepo_telemetry::events::{
     generic::GenericEventBuilder, task::PackageTaskEventBuilder, EventBuilder, TrackedErrors,
 };
 use turborepo_ui::{
-    tui::{self, TuiTask},
+    tui::{self, AppSender, TuiTask},
     ColorSelector, OutputClient, OutputSink, OutputWriter, PrefixedUI, UI,
 };
 use which::which;
@@ -62,7 +62,7 @@ pub struct Visitor<'a> {
     sink: OutputSink<StdWriter>,
     task_hasher: TaskHasher<'a>,
     ui: UI,
-    experimental_ui: bool,
+    experimental_ui_sender: Option<AppSender>,
 }
 
 #[derive(Debug, thiserror::Error)]
@@ -105,7 +105,7 @@ impl<'a> Visitor<'a> {
         manager: ProcessManager,
         repo_root: &'a AbsoluteSystemPath,
         global_env: EnvironmentVariableMap,
-        experimental_ui: bool,
+        experimental_ui_sender: Option<AppSender>,
     ) -> Self {
         let task_hasher = TaskHasher::new(
             package_inputs_hashes,
@@ -132,7 +132,7 @@ impl<'a> Visitor<'a> {
             task_hasher,
             ui,
             global_env,
-            experimental_ui,
+            experimental_ui_sender,
         }
     }
 
@@ -145,16 +145,6 @@ impl<'a> Visitor<'a> {
         let concurrency = self.run_opts.concurrency as usize;
         let (node_sender, mut node_stream) = mpsc::channel(concurrency);
 
-        let (ui, render_thread_handle) = if self.experimental_ui {
-            let task_names = engine.tasks_with_command(&self.package_graph);
-
-            let (handle, receiver) = tui::AppSender::new();
-            let app = tokio::task::spawn_blocking(move || tui::run_app(task_names, receiver));
-            (Some(handle), Some(app))
-        } else {
-            (None, None)
-        };
-
         let engine_handle = {
             let engine = engine.clone();
             tokio::spawn(engine.execute(ExecutionOptions::new(false, concurrency), node_sender))
@@ -282,7 +272,7 @@ impl<'a> Visitor<'a> {
                     let vendor_behavior =
                         Vendor::infer().and_then(|vendor| vendor.behavior.as_ref());
 
-                    let output_client = if let Some(handle) = &ui {
+                    let output_client = if let Some(handle) = &self.experimental_ui_sender {
                         TaskOutput::UI(handle.task(info.to_string()))
                     } else {
                         TaskOutput::Direct(self.output_client(&info, vendor_behavior))
@@ -315,16 +305,6 @@ impl<'a> Visitor<'a> {
             result.unwrap_or_else(|e| panic!("task executor panicked: {e}"));
         }
         drop(factory);
-        if let Some(handle) = ui {
-            handle.stop();
-            if let Err(e) = render_thread_handle
-                .unwrap()
-                .await
-                .expect("render thread panicked")
-            {
-                error!("error encountered rendering tui: {e}");
-            }
-        }
 
         // Write out the traced-config.json file if we have one
         self.task_access.save().await;
@@ -483,7 +463,7 @@ impl<'a> Visitor<'a> {
     pub fn dry_run(&mut self) {
         self.dry = true;
         // No need to start a TUI on dry run
-        self.experimental_ui = false;
+        self.experimental_ui_sender = None;
     }
 }
 
@@ -637,7 +617,7 @@ impl<'a> ExecContextFactory<'a> {
         ExecContext {
             engine: self.engine.clone(),
             ui: self.visitor.ui,
-            experimental_ui: self.visitor.experimental_ui,
+            experimental_ui: self.visitor.experimental_ui_sender.is_some(),
             is_github_actions: self.visitor.run_opts.is_github_actions,
             pretty_prefix: self
                 .visitor