perf: Send engine callback before tracker bookkeeping and add tracing spans (#11970)

## Summary

Moves the engine callback send in task `execute()` to fire immediately
after output is flushed (`output_client.finish()`), before the tracker
bookkeeping (`tracker.cached()`, `tracker.build_succeeded()`, etc.).
This unblocks the engine's DAG walker sooner — dependent tasks can be
dispatched while the completed task's run summary tracking happens in
the background.

| Repo | Tasks | Baseline | After | Delta |
|------|-------|----------|-------|-------|
| Small (~5 packages) | 5 | 164ms | 163ms | ~flat |
| Medium (~125 packages) | 70 | 848ms | 864ms | ~flat |
| Large (~1000 packages) | 1700 | 845ms | 804ms | **-41ms (-4.9%)** |

Profiled with `--profile` across three monorepos of varying size (full
cache hit scenario, median of 5 warm runs).

## Changes

Also adds tracing spans throughout the dispatch loop and builder startup
for profiling visibility:
- `visit_recv_wait`, `task_cache_new`, `exec_context_new` in the
dispatch loop
- `micro_frontends_from_disk`, `task_access_setup`,
`turbo_json_loader_setup`, `root_turbo_json_load`,
`pkg_dep_graph_validate` in builder startup

Output ordering is preserved because the callback fires after all log
replay and output flushing is complete.

- **`crates/turborepo-task-executor/src/exec.rs`** — Restructured
`execute()` to send callback immediately after output flush, before
tracker bookkeeping.
- **`crates/turborepo-lib/src/task_graph/visitor/mod.rs`** — Added
`visit_recv_wait`, `task_cache_new`, `exec_context_new` tracing spans.
- **`crates/turborepo-lib/src/run/builder.rs`** — Added tracing spans
for uninstrumented startup phases.

Author: anthonyshew

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

@@ -260,13 +260,15 @@ impl RunBuilder {
         repo_telemetry.track_package_manager(pkg_dep_graph.package_manager().name().to_string());
         repo_telemetry.track_size(pkg_dep_graph.len());
         run_telemetry.track_run_type(self.opts.run_opts.dry_run.is_some());
-        let micro_frontend_configs =
+        let micro_frontend_configs = {
+            let _span = tracing::info_span!("micro_frontends_from_disk").entered();
             match MicrofrontendsConfigs::from_disk(&self.repo_root, &pkg_dep_graph) {
                 Ok(configs) => configs,
                 Err(err) => {
                     return Err(Error::MicroFrontends(err));
                 }
-            };
+            }
+        };
 
         let (scm, repo_index) = scm_task
             .instrument(tracing::info_span!("scm_task_await"))
@@ -323,53 +325,66 @@ impl RunBuilder {
             )?
         };
 
-        let task_access = TaskAccess::new(self.repo_root.clone(), async_cache.clone(), &scm);
-        task_access.restore_config().await;
+        let task_access = {
+            let _span = tracing::info_span!("task_access_setup").entered();
+            let ta = TaskAccess::new(self.repo_root.clone(), async_cache.clone(), &scm);
+            ta.restore_config().await;
+            ta
+        };
 
         let root_turbo_json_path = self.opts.repo_opts.root_turbo_json_path.clone();
         let future_flags = self.opts.future_flags;
 
         let reader = TurboJsonReader::new(self.repo_root.clone()).with_future_flags(future_flags);
 
-        let turbo_json_loader = if task_access.is_enabled() {
-            UnifiedTurboJsonLoader::task_access(
-                reader,
-                root_turbo_json_path.clone(),
-                root_package_json.clone(),
-            )
-        } else if is_single_package {
-            UnifiedTurboJsonLoader::single_package(
-                reader,
-                root_turbo_json_path.clone(),
-                root_package_json.clone(),
-            )
-        } else if !root_turbo_json_path.exists() &&
-        // Infer a turbo.json if allowing no turbo.json is explicitly allowed or if MFE configs are discovered
-        (self.opts.repo_opts.allow_no_turbo_json || micro_frontend_configs.is_some())
-        {
-            UnifiedTurboJsonLoader::workspace_no_turbo_json(
-                reader,
-                pkg_dep_graph.packages(),
-                micro_frontend_configs.clone(),
-            )
-        } else if let Some(micro_frontends) = &micro_frontend_configs {
-            UnifiedTurboJsonLoader::workspace_with_microfrontends(
-                reader,
-                root_turbo_json_path.clone(),
-                pkg_dep_graph.packages(),
-                micro_frontends.clone(),
-            )
-        } else {
-            UnifiedTurboJsonLoader::workspace(
-                reader,
-                root_turbo_json_path.clone(),
-                pkg_dep_graph.packages(),
-            )
+        let turbo_json_loader = {
+            let _span = tracing::info_span!("turbo_json_loader_setup").entered();
+            if task_access.is_enabled() {
+                UnifiedTurboJsonLoader::task_access(
+                    reader,
+                    root_turbo_json_path.clone(),
+                    root_package_json.clone(),
+                )
+            } else if is_single_package {
+                UnifiedTurboJsonLoader::single_package(
+                    reader,
+                    root_turbo_json_path.clone(),
+                    root_package_json.clone(),
+                )
+            } else if !root_turbo_json_path.exists() &&
+            // Infer a turbo.json if allowing no turbo.json is explicitly allowed or if MFE configs are discovered
+            (self.opts.repo_opts.allow_no_turbo_json || micro_frontend_configs.is_some())
+            {
+                UnifiedTurboJsonLoader::workspace_no_turbo_json(
+                    reader,
+                    pkg_dep_graph.packages(),
+                    micro_frontend_configs.clone(),
+                )
+            } else if let Some(micro_frontends) = &micro_frontend_configs {
+                UnifiedTurboJsonLoader::workspace_with_microfrontends(
+                    reader,
+                    root_turbo_json_path.clone(),
+                    pkg_dep_graph.packages(),
+                    micro_frontends.clone(),
+                )
+            } else {
+                UnifiedTurboJsonLoader::workspace(
+                    reader,
+                    root_turbo_json_path.clone(),
+                    pkg_dep_graph.packages(),
+                )
+            }
         };
 
-        let root_turbo_json = turbo_json_loader.load(&PackageName::Root)?.clone();
+        let root_turbo_json = {
+            let _span = tracing::info_span!("root_turbo_json_load").entered();
+            turbo_json_loader.load(&PackageName::Root)?.clone()
+        };
 
-        pkg_dep_graph.validate()?;
+        {
+            let _span = tracing::info_span!("pkg_dep_graph_validate").entered();
+            pkg_dep_graph.validate()?;
+        }
 
         let filtered_pkgs = {
             let _span = tracing::info_span!("calculate_filtered_packages").entered();

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

@@ -15,7 +15,7 @@ use futures::{stream::FuturesUnordered, StreamExt};
 use itertools::Itertools;
 use miette::{Diagnostic, NamedSource, SourceSpan};
 use tokio::sync::mpsc;
-use tracing::{debug, warn, Span};
+use tracing::{debug, warn, Instrument, Span};
 use turbopath::{AbsoluteSystemPath, AnchoredSystemPath};
 use turborepo_ci::{Vendor, VendorBehavior};
 use turborepo_engine::{TaskError, TaskWarning};
@@ -336,7 +336,14 @@ impl<'a> Visitor<'a> {
         let factory = ExecContextFactory::new(self, errors.clone(), self.manager.clone(), &engine)?;
         let cached_vendor_behavior = Vendor::infer().and_then(|vendor| vendor.behavior.as_ref());
 
-        while let Some(message) = node_stream.recv().await {
+        loop {
+            let message = node_stream
+                .recv()
+                .instrument(tracing::info_span!("visit_recv_wait"))
+                .await;
+            let Some(message) = message else {
+                break;
+            };
             let span = tracing::debug_span!(parent: &span, "queue_task", task = %message.info);
             let _enter = span.enter();
             let crate::engine::Message { info, callback } = message;
@@ -382,12 +389,11 @@ impl<'a> Visitor<'a> {
 
             debug!("task {} hash is {}", info, task_hash);
 
-            let task_cache = self.run_cache.task_cache(
-                task_definition,
-                workspace_info,
-                info.clone(),
-                &task_hash,
-            );
+            let task_cache = {
+                let _span = tracing::info_span!("task_cache_new").entered();
+                self.run_cache
+                    .task_cache(task_definition, workspace_info, info.clone(), &task_hash)
+            };
 
             // Drop to avoid holding the span across an await
 
@@ -405,15 +411,18 @@ impl<'a> Visitor<'a> {
                 }
                 false => {
                     let takes_input = task_definition.interactive || task_definition.persistent;
-                    let Some(mut exec_context) = factory.exec_context(
-                        info.clone(),
-                        task_hash,
-                        task_cache,
-                        execution_env,
-                        takes_input,
-                        self.task_access.clone(),
-                    )?
-                    else {
+                    let exec_context = {
+                        let _span = tracing::info_span!("exec_context_new").entered();
+                        factory.exec_context(
+                            info.clone(),
+                            task_hash,
+                            task_cache,
+                            execution_env,
+                            takes_input,
+                            self.task_access.clone(),
+                        )?
+                    };
+                    let Some(mut exec_context) = exec_context else {
                         continue;
                     };
 

crates/turborepo-task-executor/src/exec.rs

@@ -287,41 +287,55 @@ where
             result = Err(InternalError::Io(e));
         }
 
-        match result {
-            Ok(ExecOutcome::Success(outcome)) => {
-                match outcome {
-                    SuccessOutcome::CacheHit => tracker.cached().await,
-                    SuccessOutcome::Run => tracker.build_succeeded(0).await,
-                };
+        // Send the callback as early as possible after output is flushed.
+        // For cache hits, this unblocks dependent tasks before the tracker
+        // bookkeeping runs, shaving latency off the critical path.
+        match &result {
+            Ok(ExecOutcome::Success(_)) => {
                 callback.send(Ok(())).ok();
             }
-            Ok(ExecOutcome::Task { exit_code, message }) => {
-                tracker.build_failed(exit_code, message).await;
+            Ok(ExecOutcome::Task { .. }) => {
                 callback
                     .send(match self.continue_on_error {
                         ContinueMode::Always => Ok(()),
                         ContinueMode::DependenciesSuccessful => Err(StopExecution::DependentTasks),
                         ContinueMode::Never => Err(StopExecution::AllTasks),
                     })
                     .ok();
+            }
+            Ok(ExecOutcome::Shutdown) | Err(_) => {
+                callback.send(Err(StopExecution::AllTasks)).ok();
+            }
+            Ok(ExecOutcome::Restarted) => {
+                callback.send(Err(StopExecution::DependentTasks)).ok();
+            }
+        }
 
+        // Tracker bookkeeping happens after the callback so dependents
+        // can start while we update summaries.
+        match result {
+            Ok(ExecOutcome::Success(outcome)) => {
+                match outcome {
+                    SuccessOutcome::CacheHit => tracker.cached().await,
+                    SuccessOutcome::Run => tracker.build_succeeded(0).await,
+                };
+            }
+            Ok(ExecOutcome::Task { exit_code, message }) => {
+                tracker.build_failed(exit_code, message).await;
                 match self.continue_on_error {
                     ContinueMode::Always | ContinueMode::DependenciesSuccessful => (),
                     ContinueMode::Never => self.manager.stop().await,
                 }
             }
             Ok(ExecOutcome::Shutdown) => {
                 tracker.cancel();
-                callback.send(Err(StopExecution::AllTasks)).ok();
                 self.manager.stop().await;
             }
             Ok(ExecOutcome::Restarted) => {
                 tracker.cancel();
-                callback.send(Err(StopExecution::DependentTasks)).ok();
             }
             Err(e) => {
                 tracker.cancel();
-                callback.send(Err(StopExecution::AllTasks)).ok();
                 self.manager.stop().await;
                 return Err(e);
             }