feat: Remove daemon from watch mode, use in-process file watching

Replace the daemon gRPC IPC in turbo watch with in-process file watching
infrastructure. All watcher components (FileSystemWatcher, CookieWriter,
GlobWatcher, PackageWatcher, HashWatcher, PackageChangesWatcher) are
instantiated directly in WatchClient, eliminating the daemon dependency
for watch mode.

Key changes:
- Replace DaemonClient with OutputWatcher trait in RunCache/TaskCache
- Create InProcessOutputWatcher wrapping GlobWatcher
- Build full watcher stack in WatchClient::new()
- Pre-populate hash baselines at startup to prevent spurious rebuilds
- Wait for active runs to complete before processing new change events
- Make TUI start_task resilient to concurrent/out-of-order events
- Handle SIGINT as clean exit (exit code 0, no error message)

The daemon is still used by other consumers (turbo daemon CLI, LSP,
turbo info, diagnostics). This change only decouples watch mode.

Author: anthonyshew

Cargo.lock

@@ -1727,12 +1727,18 @@ pub async fn run(
             let mut client =
                 WatchClient::new(base, *experimental_write_cache, event, query_server.clone())
                     .await?;
-            if let Err(e) = client.start().await {
-                client.shutdown().await;
-                return Err(e.into());
+            match client.start().await {
+                Ok(()) => {}
+                Err(crate::run::watch::Error::SignalInterrupt) => {
+                    // Normal shutdown via Ctrl+C — not an error.
+                }
+                Err(e) => {
+                    client.shutdown().await;
+                    return Err(e.into());
+                }
             }
-            // We only exit if we get a signal, so we return a non-zero exit code
-            return Ok(1);
+            client.shutdown().await;
+            return Ok(0);
         }
         Command::Prune {
             scope,

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

@@ -441,9 +441,27 @@ impl Subscriber {
             else {
                 return;
             };
-            // We store the hash of the package's files. If the hash is already
-            // in here, we don't need to recompute it
+            // Pre-populate hash baselines for all known packages. Without
+            // this, the first file change for each package would always be
+            // treated as "new" (no old hash to compare against), causing
+            // spurious rebuilds from build output writes on the initial run.
             let mut package_file_hashes = HashMap::new();
+            for (name, info) in repo_state.pkg_dep_graph.packages() {
+                let pkg = WorkspacePackage {
+                    name: name.clone(),
+                    path: info.package_path().to_owned(),
+                };
+                if let Ok(hash) = self
+                    .hash_watcher
+                    .get_file_hashes(HashSpec {
+                        package_path: pkg.path.clone(),
+                        inputs: InputGlobs::Default,
+                    })
+                    .await
+                {
+                    package_file_hashes.insert(pkg.path, hash);
+                }
+            }
 
             let mut change_mapper = match repo_state.get_change_mapper() {
                 Some(change_mapper) => change_mapper,

crates/turborepo-lib/src/package_changes_watcher.rs

@@ -11,7 +11,6 @@ use turbopath::{AbsoluteSystemPath, AbsoluteSystemPathBuf};
 use turborepo_analytics::{start_analytics, AnalyticsHandle};
 use turborepo_api_client::{APIAuth, APIClient};
 use turborepo_cache::AsyncCache;
-use turborepo_daemon::{DaemonClient, DaemonConnector};
 use turborepo_env::EnvironmentVariableMap;
 use turborepo_errors::Spanned;
 use turborepo_process::ProcessManager;
@@ -61,11 +60,11 @@ pub struct RunBuilder {
     should_validate_engine: bool,
     // If true, we will add all tasks to the graph, even if they are not specified
     add_all_tasks: bool,
-    // When running under `turbo watch`, a daemon client is needed so that
+    // When running under `turbo watch`, an output watcher is needed so that
     // the run cache can register output globs and skip restoring outputs
     // that are already on disk. Without this, cache restores write files
     // that trigger the file watcher, causing an infinite rebuild loop.
-    daemon_client: Option<DaemonClient<DaemonConnector>>,
+    output_watcher: Option<Arc<dyn turborepo_run_cache::OutputWatcher>>,
     query_server: Option<Arc<dyn turborepo_query_api::QueryServer>>,
 }
 
@@ -108,7 +107,7 @@ impl RunBuilder {
             should_print_prelude_override: None,
             should_validate_engine: true,
             add_all_tasks: false,
-            daemon_client: None,
+            output_watcher: None,
             query_server: None,
         })
     }
@@ -118,8 +117,11 @@ impl RunBuilder {
         self
     }
 
-    pub fn with_daemon_client(mut self, client: DaemonClient<DaemonConnector>) -> Self {
-        self.daemon_client = Some(client);
+    pub fn with_output_watcher(
+        mut self,
+        watcher: Arc<dyn turborepo_run_cache::OutputWatcher>,
+    ) -> Self {
+        self.output_watcher = Some(watcher);
         self
     }
 
@@ -471,7 +473,7 @@ impl RunBuilder {
             &self.repo_root,
             self.opts.runcache_opts,
             &self.opts.cache_opts,
-            self.daemon_client,
+            self.output_watcher,
             self.color_config,
             self.opts.run_opts.dry_run.is_some(),
         ));