perf(daemon): run trace normalization on blocking workers

src/daemon.rs

@@ -2725,9 +2725,13 @@ pub struct ActorDaemonCoordinator {
     backend: Arc<crate::daemon::git_backend::SystemGitBackend>,
     coordinator:
         Arc<crate::daemon::coordinator::Coordinator<crate::daemon::git_backend::SystemGitBackend>>,
-    normalizer: AsyncMutex<
-        crate::daemon::trace_normalizer::TraceNormalizer<
-            crate::daemon::git_backend::SystemGitBackend,
+    // std Mutex (not AsyncMutex) so the normalizer can move into
+    // spawn_blocking closures; it is only ever locked from blocking context.
+    normalizer: Arc<
+        Mutex<
+            crate::daemon::trace_normalizer::TraceNormalizer<
+                crate::daemon::git_backend::SystemGitBackend,
+            >,
         >,
     >,
     pending_rebase_original_head_by_worktree: Mutex<HashMap<String, (String, Option<String>)>>,
@@ -2822,8 +2826,8 @@ impl ActorDaemonCoordinator {
             coordinator: Arc::new(crate::daemon::coordinator::Coordinator::new(
                 backend.clone(),
             )),
-            normalizer: AsyncMutex::new(crate::daemon::trace_normalizer::TraceNormalizer::new(
-                backend.clone(),
+            normalizer: Arc::new(Mutex::new(
+                crate::daemon::trace_normalizer::TraceNormalizer::new(backend.clone()),
             )),
             backend,
             pending_rebase_original_head_by_worktree: Mutex::new(HashMap::new()),
@@ -6433,13 +6437,34 @@ impl ActorDaemonCoordinator {
             .and_then(Value::as_str)
             .unwrap_or_default()
             .to_string();
+        let terminal_root_event = is_terminal_root_trace_event(
+            &event,
+            payload
+                .get("sid")
+                .and_then(Value::as_str)
+                .unwrap_or_default(),
+            payload_root_sid.as_deref().unwrap_or_default(),
+        );
         if event == TRACE_CONNECTION_CLOSED_EVENT {
             let Some(root_sid) = payload_root_sid.as_deref() else {
                 return Ok(TracePayloadApplyOutcome::None);
             };
             {
-                let mut normalizer = self.normalizer.lock().await;
-                let _ = normalizer.sweep_orphans_for_roots(&[root_sid.to_string()]);
+                // Orphan sweeps read reflogs and repo state; run them on a
+                // blocking worker so filesystem I/O does not block a Tokio worker.
+                // Recover from poisoning: the ingest worker catches per-payload
+                // panics and continues, so a poisoned lock must not disable
+                // normalization for the daemon's remaining lifetime.
+                let normalizer = Arc::clone(&self.normalizer);
+                let root_sid_owned = root_sid.to_string();
+                crate::tokio_runtime::spawn_blocking_result(move || {
+                    let mut normalizer = normalizer
+                        .lock()
+                        .unwrap_or_else(|poisoned| poisoned.into_inner());
+                    let _ = normalizer.sweep_orphans_for_roots(&[root_sid_owned]);
+                    Ok(())
+                })
+                .await?;
             }
             let replaced_family = self
                 .replace_pending_root_entry(root_sid, FamilySequencerEntry::Canceled)
@@ -6456,19 +6481,27 @@ impl ActorDaemonCoordinator {
         }
 
         self.maybe_append_pending_root_from_trace_payload(&payload)?;
+        // Normalization can touch the filesystem (alias resolution reads git
+        // config files on exit events); run it on a blocking worker so a slow
+        // disk or cold cache does not block a Tokio worker. The ingest task
+        // still awaits each result to preserve trace ordering.
+        // Recover from poisoning: the ingest worker catches per-payload panics
+        // and continues, so a poisoned lock must not disable normalization for
+        // the daemon's remaining lifetime.
         let emitted = {
-            let mut normalizer = self.normalizer.lock().await;
-            normalizer.ingest_payload(&payload)?
+            let normalizer = Arc::clone(&self.normalizer);
+            let payload_for_ingest = payload;
+            crate::tokio_runtime::spawn_blocking_result(move || {
+                normalizer
+                    .lock()
+                    .unwrap_or_else(|poisoned| poisoned.into_inner())
+                    .ingest_payload(&payload_for_ingest)
+            })
+            .await?
         };
         let Some(command) = emitted else {
-            if is_terminal_root_trace_event(
-                &event,
-                payload
-                    .get("sid")
-                    .and_then(Value::as_str)
-                    .unwrap_or_default(),
-                payload_root_sid.as_deref().unwrap_or_default(),
-            ) && let Some(root_sid) = payload_root_sid.as_deref()
+            if terminal_root_event
+                && let Some(root_sid) = payload_root_sid.as_deref()
                 && let Some(family) = self
                     .replace_pending_root_entry(root_sid, FamilySequencerEntry::Canceled)
                     .await?

src/tokio_runtime.rs

@@ -65,10 +65,16 @@ fn build_bounded_runtime(
 }
 
 pub(crate) fn build_daemon_runtime() -> Result<tokio::runtime::Runtime, String> {
-    build_bounded_runtime(
-        DAEMON_RUNTIME_WORKER_THREADS,
-        DAEMON_RUNTIME_MAX_BLOCKING_THREADS,
-    )
+    #[cfg(feature = "test-support")]
+    let worker_threads = std::env::var("GIT_AI_TEST_DAEMON_RUNTIME_WORKER_THREADS")
+        .ok()
+        .and_then(|value| value.parse::<usize>().ok())
+        .filter(|value| *value > 0)
+        .unwrap_or(DAEMON_RUNTIME_WORKER_THREADS);
+    #[cfg(not(feature = "test-support"))]
+    let worker_threads = DAEMON_RUNTIME_WORKER_THREADS;
+
+    build_bounded_runtime(worker_threads, DAEMON_RUNTIME_MAX_BLOCKING_THREADS)
 }
 
 // Post-commit attribution calls this helper from inside the daemon runtime.

tests/daemon_mode.rs

@@ -34,6 +34,8 @@ use std::fs;
 use std::io::{BufRead, BufReader};
 use std::io::{Read, Write};
 use std::net::{TcpListener, TcpStream};
+#[cfg(not(windows))]
+use std::os::unix::fs::OpenOptionsExt;
 use std::path::{Path, PathBuf};
 use std::process::{Child, Command, Stdio};
 use std::sync::atomic::{AtomicBool, Ordering};
@@ -2676,6 +2678,155 @@ fn daemon_trace_connection_close_without_atexit_does_not_block_later_trace() {
     panic!("daemon did not process a later trace after a mutating root closed before atexit");
 }
 
+#[test]
+#[cfg(not(windows))]
+fn daemon_trace_normalization_offload_keeps_single_worker_runtime_responsive() {
+    let repo = TestRepo::new_with_daemon_scope(DaemonTestScope::NoDaemon);
+    let mut daemon = DaemonGuard::start_with_env(
+        &repo,
+        &[
+            ("GIT_AI_TEST_DAEMON_RUNTIME_WORKER_THREADS", "1"),
+            ("GIT_AI_DAEMON_UPDATE_CHECK_INTERVAL", "86400"),
+            ("GIT_AI_DAEMON_MAX_UPTIME_SECS", "86400"),
+        ],
+    );
+    let trace_socket = daemon_trace_socket_path(&repo);
+    let control_socket = daemon_control_socket_path(&repo);
+    let worktree = repo_workdir_string(&repo);
+    let git_dir = repo.path().join(".git").to_string_lossy().to_string();
+
+    // A FIFO-backed repository config gives the real SystemGitBackend a
+    // deterministic blocking read on the first alias lookup, without adding a
+    // delay hook to the latency-sensitive ingestion path.
+    let repo_config_path = repo.path().join(".git/config");
+    let original_config =
+        fs::read_to_string(&repo_config_path).expect("failed to read original repository config");
+    fs::remove_file(&repo_config_path).expect("failed to replace repository config with FIFO");
+    let mkfifo = Command::new("mkfifo")
+        .arg(&repo_config_path)
+        .status()
+        .expect("failed to invoke mkfifo");
+    assert!(mkfifo.success(), "mkfifo failed: {mkfifo}");
+
+    let sid = "blocking-alias-normalization";
+    let mut trace_stream =
+        open_local_socket_stream_with_timeout(&trace_socket, DAEMON_TEST_PROBE_TIMEOUT)
+            .expect("failed to connect to trace socket");
+    write_trace_frames_to_stream(
+        &mut trace_stream,
+        &[
+            json!({
+                "event": "start",
+                "sid": sid,
+                "argv": ["git", "ci", "-m", "synthetic"],
+                "time_ns": 11_000u64,
+            }),
+            json!({
+                "event": "def_repo",
+                "sid": sid,
+                "worktree": worktree,
+                "repo": git_dir,
+                "time_ns": 11_001u64,
+            }),
+        ],
+    );
+    thread::sleep(Duration::from_millis(150));
+
+    let response = send_control_request_with_timeout(
+        &control_socket,
+        &ControlRequest::StatusFamily {
+            repo_working_dir: repo_workdir_string(&repo),
+        },
+        Duration::from_millis(500),
+    )
+    .expect("blocking trace normalization must not occupy the only Tokio worker");
+    assert!(response.ok, "status.family failed: {response:?}");
+
+    let deadline = std::time::Instant::now() + Duration::from_secs(2);
+    let mut alias_writer = loop {
+        match fs::OpenOptions::new()
+            .write(true)
+            .custom_flags(libc::O_NONBLOCK)
+            .open(&repo_config_path)
+        {
+            Ok(writer) => break writer,
+            Err(error)
+                if error.raw_os_error() == Some(libc::ENXIO)
+                    && std::time::Instant::now() < deadline =>
+            {
+                thread::sleep(Duration::from_millis(10));
+            }
+            Err(error) => panic!("failed to open blocking alias config: {error}"),
+        }
+    };
+    alias_writer
+        .write_all(format!("{original_config}\n[alias]\n\tci = commit\n").as_bytes())
+        .expect("failed to release blocking alias config read");
+    drop(alias_writer);
+    fs::remove_file(&repo_config_path).expect("failed to remove blocking repository config FIFO");
+    fs::write(
+        &repo_config_path,
+        format!("{original_config}\n[alias]\n\tci = commit\n"),
+    )
+    .expect("failed to restore repository config");
+
+    let session = repos::test_repo::new_daemon_test_sync_session_id();
+    let session_arg = format!("git-ai.testSyncSession={session}");
+    let followup_sid = "trace-after-blocking-alias-normalization";
+    write_trace_frames_to_stream(
+        &mut trace_stream,
+        &[
+            json!({
+                "event": "exit",
+                "sid": sid,
+                "code": 0,
+                "time_ns": 11_100u64,
+            }),
+            trace_atexit_frame(sid, 0, 11_101u64),
+            json!({
+                "event": "start",
+                "sid": followup_sid,
+                "argv": ["git", "-c", session_arg, "commit", "-m", "followup"],
+                "time_ns": 12_000u64,
+            }),
+            json!({
+                "event": "def_repo",
+                "sid": followup_sid,
+                "worktree": repo_workdir_string(&repo),
+                "repo": repo.path().join(".git").to_string_lossy().to_string(),
+                "time_ns": 12_001u64,
+            }),
+            json!({
+                "event": "exit",
+                "sid": followup_sid,
+                "code": 0,
+                "time_ns": 12_100u64,
+            }),
+            trace_atexit_frame(followup_sid, 0, 12_101u64),
+        ],
+    );
+    let sync = send_control_request_with_timeout(
+        &control_socket,
+        &ControlRequest::SyncFamily {
+            repo_working_dir: repo_workdir_string(&repo),
+        },
+        Duration::from_secs(2),
+    )
+    .expect("sync.family should complete after the blocking config read is released");
+    assert!(sync.ok, "sync.family failed: {sync:?}");
+    let matching_completions = repo
+        .daemon_completion_entries()
+        .into_iter()
+        .filter(|entry| entry.test_sync_session.as_deref() == Some(session.as_str()))
+        .collect::<Vec<_>>();
+    assert_eq!(
+        matching_completions.len(),
+        1,
+        "a later trace must normalize and complete after the blocking read"
+    );
+    daemon.shutdown();
+}
+
 #[test]
 #[cfg(not(windows))]
 fn daemon_control_listener_stalled_connection_does_not_block_later_control_requests() {