feat: Remove daemon from watch mode (#12175)

## Summary

Replaces the daemon gRPC IPC in `turbo watch` with in-process file
watching. All watcher components are instantiated directly in
`WatchClient`, eliminating the daemon as a dependency for watch mode.

- Introduces `OutputWatcher` trait to abstract output change tracking,
replacing `DaemonClient` in `RunCache`/`TaskCache`
- Builds the full watcher stack in-process: `FileSystemWatcher` →
`CookieWriter` → `GlobWatcher` + `PackageWatcher` → `HashWatcher` →
`PackageChangesWatcher`
- Pre-populates hash baselines at startup so build output writes don't
trigger spurious rebuilds
- Waits for active runs to complete before processing new change events,
preventing concurrent builds of the same package
- Makes TUI `start_task` resilient to out-of-order events from
concurrent runs
- Handles SIGINT as a clean exit (code 0) instead of propagating an
error

The daemon is still used by other consumers (`turbo daemon` CLI, LSP,
`turbo info`).

## Testing

Manually verified on the turborepo monorepo (24 packages):
- Initial build settles without spurious rebuilds
- File edits during builds are not lost
- Rapid successive edits are debounced
- Same-content writes don't trigger rebuilds
- Ctrl+C exits cleanly
- TUI stays alive across rebuilds

Automated: 11 E2E tests, 18 watch unit tests, 8 OutputWatcher trait
tests, 6 GlobWatcher delegation tests, 2 TUI start_task resilience tests
— all passing.

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(),
         ));