feat: add ability to configure tasks as interactive (#7767)

### Description

This adds a new `"interactive"` field to task definitions which allows
users to interact with tasks when it is set to true.

A few notes on the feature:
 - All persistent tasks automatically are interactive
- Interactive tasks cannot be cached since the interaction can (and
probably will) alter the task's execution. This could be done
automatically, but I chose to require that the task be explicitly marked
as `"cache": false` to reduce perceived magic.
- `turbo` will not attempt to run if any tasks in the graph are marked
as interactive and the experimental UI isn't being used

### Testing Instructions

Snapshot test for various validation errors.

Manual testing: no longer able to select tasks and interact with them
e.g. `turbo_dev build --filter=docs --force`


Closes TURBO-2656

Author: chris-olszewski

crates/turborepo-lib/src/config.rs

@@ -118,6 +118,13 @@ pub enum Error {
         #[source_code]
         text: NamedSource,
     },
+    #[error("Tasks cannot be marked as interactive and cacheable")]
+    InteractiveNoCacheable {
+        #[label("marked interactive here")]
+        span: Option<SourceSpan>,
+        #[source_code]
+        text: NamedSource,
+    },
     #[error("Failed to create APIClient: {0}")]
     ApiClient(#[source] turborepo_api_client::Error),
     #[error("{0} is not UTF8.")]

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

@@ -181,6 +181,7 @@ impl Engine<Built> {
         &self,
         package_graph: &PackageGraph,
         concurrency: u32,
+        experimental_ui: bool,
     ) -> Result<(), Vec<ValidateError>> {
         // TODO(olszewski) once this is hooked up to a real run, we should
         // see if using rayon to parallelize would provide a speedup
@@ -277,11 +278,34 @@ impl Engine<Built> {
             })
         }
 
+        validation_errors.extend(self.validate_interactive(experimental_ui));
+
         match validation_errors.is_empty() {
             true => Ok(()),
             false => Err(validation_errors),
         }
     }
+
+    // Validates that UI is setup if any interactive tasks will be executed
+    fn validate_interactive(&self, experimental_ui: bool) -> Vec<ValidateError> {
+        // If experimental_ui is being used, then we don't need check for interactive
+        // tasks
+        if experimental_ui {
+            return Vec::new();
+        }
+        self.task_definitions
+            .iter()
+            .filter_map(|(task, definition)| {
+                if definition.interactive {
+                    Some(ValidateError::InteractiveNeedsUI {
+                        task: task.to_string(),
+                    })
+                } else {
+                    None
+                }
+            })
+            .collect()
+    }
 }
 
 #[derive(Debug, Error, Diagnostic)]
@@ -310,6 +334,11 @@ pub enum ValidateError {
         persistent_count: u32,
         concurrency: u32,
     },
+    #[error(
+        "Cannot run interactive task \"{task}\" without experimental UI. Set `\"experimentalUI\": \
+         true` in `turbo.json` or `TURBO_EXPERIMENTAL_UI=true` as an environment variable"
+    )]
+    InteractiveNeedsUI { task: String },
 }
 
 impl fmt::Display for TaskNode {
@@ -427,16 +456,16 @@ mod test {
         let graph = graph_builder.build().await.unwrap();
 
         // if our limit is less than, it should fail
-        engine.validate(&graph, 1).expect_err("not enough");
+        engine.validate(&graph, 1, false).expect_err("not enough");
 
         // if our limit is less than, it should fail
-        engine.validate(&graph, 2).expect_err("not enough");
+        engine.validate(&graph, 2, false).expect_err("not enough");
 
         // we have two persistent tasks, and a slot for all other tasks, so this should
         // pass
-        engine.validate(&graph, 3).expect("ok");
+        engine.validate(&graph, 3, false).expect("ok");
 
         // if our limit is greater, then it should pass
-        engine.validate(&graph, 4).expect("ok");
+        engine.validate(&graph, 4, false).expect("ok");
     }
 }

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

@@ -423,7 +423,11 @@ impl RunBuilder {
 
         if !self.opts.run_opts.parallel {
             engine
-                .validate(pkg_dep_graph, self.opts.run_opts.concurrency)
+                .validate(
+                    pkg_dep_graph,
+                    self.opts.run_opts.concurrency,
+                    self.experimental_ui,
+                )
                 .map_err(Error::EngineValidation)?;
         }
 

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

@@ -13,7 +13,7 @@ use crate::{
 
 #[derive(Debug, Error, Diagnostic)]
 pub enum Error {
-    #[error("invalid persistent task configuration")]
+    #[error("invalid task configuration")]
     EngineValidation(#[related] Vec<ValidateError>),
     #[error(transparent)]
     Graph(#[from] graph_visualizer::Error),

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

@@ -105,6 +105,7 @@ pub struct TaskSummaryTaskDefinition {
     env: Vec<String>,
     pass_through_env: Option<Vec<String>>,
     dot_env: Option<Vec<RelativeUnixPathBuf>>,
+    interactive: bool,
 }
 
 #[derive(Debug, Serialize, Clone)]
@@ -282,6 +283,7 @@ impl From<TaskDefinition> for TaskSummaryTaskDefinition {
             mut inputs,
             output_mode,
             persistent,
+            interactive,
         } = value;
 
         let mut outputs = inclusions;
@@ -313,6 +315,7 @@ impl From<TaskDefinition> for TaskSummaryTaskDefinition {
             inputs,
             output_mode,
             persistent,
+            interactive,
             env,
             pass_through_env,
             // This should _not_ be sorted.
@@ -372,6 +375,7 @@ mod test {
             "inputs": [],
             "outputMode": "full",
             "persistent": false,
+            "interactive": false,
             "env": [],
             "passThroughEnv": null,
             "dotEnv": null,

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

@@ -73,6 +73,11 @@ pub struct TaskDefinition {
     // Persistent indicates whether the Task is expected to exit or not
     // Tasks marked Persistent do not exit (e.g. --watch mode or dev servers)
     pub persistent: bool,
+
+    // Interactive marks that a task can have it's stdin written to.
+    // Tasks that take stdin input cannot be cached as their outputs may depend on the
+    // input.
+    pub interactive: bool,
 }
 
 impl Default for TaskDefinition {
@@ -88,6 +93,7 @@ impl Default for TaskDefinition {
             output_mode: Default::default(),
             persistent: Default::default(),
             dot_env: Default::default(),
+            interactive: Default::default(),
         }
     }
 }

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

@@ -268,14 +268,14 @@ impl<'a> Visitor<'a> {
 
                     let workspace_directory = self.repo_root.resolve(workspace_info.package_path());
 
-                    let persistent = task_definition.persistent;
+                    let takes_input = task_definition.interactive || task_definition.persistent;
                     let mut exec_context = factory.exec_context(
                         info.clone(),
                         task_hash,
                         task_cache,
                         workspace_directory,
                         execution_env,
-                        persistent,
+                        takes_input,
                         self.task_access.clone(),
                     );
 
@@ -629,7 +629,7 @@ impl<'a> ExecContextFactory<'a> {
         task_cache: TaskCache,
         workspace_directory: AbsoluteSystemPathBuf,
         execution_env: EnvironmentVariableMap,
-        persistent: bool,
+        takes_input: bool,
         task_access: TaskAccess,
     ) -> ExecContext {
         let task_id_for_display = self.visitor.display_task_id(&task_id);
@@ -655,7 +655,7 @@ impl<'a> ExecContextFactory<'a> {
             continue_on_error: self.visitor.run_opts.continue_on_error,
             pass_through_args,
             errors: self.errors.clone(),
-            persistent,
+            takes_input,
             task_access,
         }
     }
@@ -691,7 +691,7 @@ struct ExecContext {
     continue_on_error: bool,
     pass_through_args: Option<Vec<String>>,
     errors: Arc<Mutex<Vec<TaskError>>>,
-    persistent: bool,
+    takes_input: bool,
     task_access: TaskAccess,
 }
 
@@ -876,25 +876,7 @@ impl ExecContext {
             cmd.env(task_access_trace_key, trace_file.to_string());
         }
 
-        // Many persistent tasks if started hooked up to a pseudoterminal
-        // will shut down if stdin is closed, so we open it even if we don't pass
-        // anything to it.
-        if self.persistent {
-            cmd.open_stdin();
-        }
-
-        let mut stdout_writer = match self.task_cache.output_writer(if self.experimental_ui {
-            Either::Left(output_client.stdout())
-        } else {
-            Either::Right(prefixed_ui.output_prefixed_writer())
-        }) {
-            Ok(w) => w,
-            Err(e) => {
-                telemetry.track_error(TrackedErrors::FailedToCaptureOutputs);
-                error!("failed to capture outputs for \"{}\": {e}", self.task_id);
-                return ExecOutcome::Internal;
-            }
-        };
+        cmd.open_stdin();
 
         let mut process = match self.manager.spawn(cmd, Duration::from_millis(500)) {
             Some(Ok(child)) => child,
@@ -918,14 +900,27 @@ impl ExecContext {
             }
         };
 
-        if self.experimental_ui {
+        if self.experimental_ui && self.takes_input {
             if let TaskOutput::UI(task) = output_client {
                 if let Some(stdin) = process.stdin() {
                     task.set_stdin(stdin);
                 }
             }
         }
 
+        let mut stdout_writer = match self.task_cache.output_writer(if self.experimental_ui {
+            Either::Left(output_client.stdout())
+        } else {
+            Either::Right(prefixed_ui.output_prefixed_writer())
+        }) {
+            Ok(w) => w,
+            Err(e) => {
+                telemetry.track_error(TrackedErrors::FailedToCaptureOutputs);
+                error!("failed to capture outputs for \"{}\": {e}", self.task_id);
+                return ExecOutcome::Internal;
+            }
+        };
+
         let exit_status = match process.wait_with_piped_outputs(&mut stdout_writer).await {
             Ok(Some(exit_status)) => exit_status,
             Err(e) => {

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

@@ -181,6 +181,8 @@ pub struct RawTaskDefinition {
     outputs: Option<Vec<Spanned<UnescapedString>>>,
     #[serde(skip_serializing_if = "Option::is_none")]
     output_mode: Option<Spanned<OutputLogsMode>>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    interactive: Option<Spanned<bool>>,
 }
 
 macro_rules! set_field {
@@ -197,7 +199,10 @@ impl RawTaskDefinition {
     pub fn merge(&mut self, other: RawTaskDefinition) {
         set_field!(self, other, outputs);
 
-        if other.cache.value.is_some() {
+        if other.cache.value.is_some()
+            // If other has range info and we're missing it, carry it over
+            || (other.cache.range.is_some() && self.cache.range.is_none())
+        {
             self.cache = other.cache;
         }
         set_field!(self, other, depends_on);
@@ -207,6 +212,7 @@ impl RawTaskDefinition {
         set_field!(self, other, env);
         set_field!(self, other, pass_through_env);
         set_field!(self, other, dot_env);
+        set_field!(self, other, interactive);
     }
 }
 
@@ -262,7 +268,19 @@ impl TryFrom<RawTaskDefinition> for TaskDefinition {
     fn try_from(raw_task: RawTaskDefinition) -> Result<Self, Error> {
         let outputs = raw_task.outputs.unwrap_or_default().try_into()?;
 
-        let cache = raw_task.cache;
+        let cache = raw_task.cache.into_inner().unwrap_or(true);
+        let interactive = raw_task
+            .interactive
+            .as_ref()
+            .map(|value| value.value)
+            .unwrap_or_default();
+
+        if let Some(interactive) = raw_task.interactive {
+            let (span, text) = interactive.span_and_text("turbo.json");
+            if cache && interactive.value {
+                return Err(Error::InteractiveNoCacheable { span, text });
+            }
+        }
 
         let mut env_var_dependencies = HashSet::new();
         let mut topological_dependencies: Vec<Spanned<TaskName>> = Vec::new();
@@ -350,7 +368,7 @@ impl TryFrom<RawTaskDefinition> for TaskDefinition {
 
         Ok(TaskDefinition {
             outputs,
-            cache: cache.into_inner().unwrap_or(true),
+            cache,
             topological_dependencies,
             task_dependencies,
             env,
@@ -359,6 +377,7 @@ impl TryFrom<RawTaskDefinition> for TaskDefinition {
             dot_env,
             output_mode: *raw_task.output_mode.unwrap_or_default(),
             persistent: *raw_task.persistent.unwrap_or_default(),
+            interactive,
         })
     }
 }
@@ -910,7 +929,8 @@ mod tests {
           "cache": false,
           "inputs": ["package/a/src/**"],
           "outputMode": "full",
-          "persistent": true
+          "persistent": true,
+          "interactive": true
         }"#,
         RawTaskDefinition {
             depends_on: Some(Spanned::new(vec![Spanned::<UnescapedString>::new("cli#build".into()).with_range(26..37)]).with_range(25..38)),
@@ -922,6 +942,7 @@ mod tests {
             inputs: Some(vec![Spanned::<UnescapedString>::new("package/a/src/**".into()).with_range(241..259)]),
             output_mode: Some(Spanned::new(OutputLogsMode::Full).with_range(286..292)),
             persistent: Some(Spanned::new(true).with_range(318..322)),
+            interactive: Some(Spanned::new(true).with_range(349..353)),
         },
         TaskDefinition {
           dot_env: Some(vec![RelativeUnixPathBuf::new("package/a/.env").unwrap()]),
@@ -937,6 +958,7 @@ mod tests {
           task_dependencies: vec![Spanned::<TaskName<'_>>::new("cli#build".into()).with_range(26..37)],
           topological_dependencies: vec![],
           persistent: true,
+          interactive: true,
         }
       ; "full"
     )]
@@ -962,6 +984,7 @@ mod tests {
             inputs: Some(vec![Spanned::<UnescapedString>::new("package\\a\\src\\**".into()).with_range(273..294)]),
             output_mode: Some(Spanned::new(OutputLogsMode::Full).with_range(325..331)),
             persistent: Some(Spanned::new(true).with_range(361..365)),
+            interactive: None,
         },
         TaskDefinition {
             dot_env: Some(vec![RelativeUnixPathBuf::new("package\\a\\.env").unwrap()]),
@@ -977,6 +1000,7 @@ mod tests {
             task_dependencies: vec![Spanned::<TaskName<'_>>::new("cli#build".into()).with_range(30..41)],
             topological_dependencies: vec![],
             persistent: true,
+            interactive: false,
         }
       ; "full (windows)"
     )]