fix(turborepo): use transitive closure of filtered packages in watch …

…mode (#8161)

### Description

We were using the set of filtered packages to limit change events.
That's not exactly right since we need to account for the dependencies
of filtered packages as well. Therefore we take the transitive closure
of the filtered packages and restrict package change events to that set.

### Testing Instructions

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

@@ -23,7 +23,7 @@ use turbopath::AbsoluteSystemPathBuf;
 use turborepo_api_client::{APIAuth, APIClient};
 use turborepo_ci::Vendor;
 use turborepo_env::EnvironmentVariableMap;
-use turborepo_repository::package_graph::{PackageGraph, PackageName};
+use turborepo_repository::package_graph::{PackageGraph, PackageName, PackageNode};
 use turborepo_scm::SCM;
 use turborepo_telemetry::events::generic::GenericEventBuilder;
 use turborepo_ui::{cprint, cprintln, tui, tui::AppSender, BOLD_GREY, GREY, UI};
@@ -118,6 +118,24 @@ impl Run {
         new_run
     }
 
+    // Produces the transitive closure of the filtered packages,
+    // i.e. the packages relevant for this run.
+    pub fn get_relevant_packages(&self) -> HashSet<PackageName> {
+        let packages: Vec<_> = self
+            .filtered_pkgs
+            .iter()
+            .map(|pkg| PackageNode::Workspace(pkg.clone()))
+            .collect();
+        self.pkg_dep_graph
+            .transitive_closure(&packages)
+            .into_iter()
+            .filter_map(|node| match node {
+                PackageNode::Root => None,
+                PackageNode::Workspace(pkg) => Some(pkg.clone()),
+            })
+            .collect()
+    }
+
     pub fn has_experimental_ui(&self) -> bool {
         self.experimental_ui
     }

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

@@ -23,6 +23,7 @@ use crate::{
     DaemonConnector, DaemonPaths,
 };
 
+#[derive(Debug)]
 enum ChangedPackages {
     All,
     Some(HashSet<PackageName>),
@@ -45,6 +46,7 @@ impl ChangedPackages {
 
 pub struct WatchClient {
     run: Run,
+    watched_packages: HashSet<PackageName>,
     persistent_tasks_handle: Option<JoinHandle<Result<i32, run::Error>>>,
     connector: DaemonConnector,
     base: CommandBase,
@@ -114,6 +116,8 @@ impl WatchClient {
             .build(&handler, telemetry.clone())
             .await?;
 
+        let watched_packages = run.get_relevant_packages();
+
         let (sender, handle) = run.start_experimental_ui().unzip();
 
         let connector = DaemonConnector {
@@ -125,6 +129,7 @@ impl WatchClient {
         Ok(Self {
             base,
             run,
+            watched_packages,
             connector,
             handler,
             telemetry,
@@ -228,8 +233,8 @@ impl WatchClient {
                 let packages = packages
                     .into_iter()
                     .filter(|pkg| {
-                        // If not in the filtered pkgs, ignore
-                        self.run.filtered_pkgs.contains(pkg)
+                        // If not in the watched packages set, ignore
+                        self.watched_packages.contains(pkg)
                     })
                     .collect();
 
@@ -297,6 +302,8 @@ impl WatchClient {
                     .build(&self.handler, self.telemetry.clone())
                     .await?;
 
+                self.watched_packages = self.run.get_relevant_packages();
+
                 if let Some(sender) = &self.ui_sender {
                     let task_names = self.run.engine.tasks_with_command(&self.run.pkg_dep_graph);
                     sender