fix: resolve pipe-buffer deadlock and add runner observability

Read stdout and stderr concurrently (background thread for stderr)
to prevent the classic deadlock where cargo blocks writing stderr
while the parent blocks reading stdout. Timeout now actually kills
the child process via kill -9. Added eprintln logging for all job
lifecycle events so failures are visible in local-runner.log.

Also switches .ci.yml to use $HOME instead of ~ for shell portability.

Authored by Claude, lovingly guided by Zachery Aaron Shores-Chmielewski

.ci.yml

@@ -5,12 +5,12 @@ pipelines:
         branches: ["*"]
     jobs:
       check:
-        run: . ~/.cargo/env && cargo check --workspace
+        run: echo "HOME=$HOME PATH=$PATH" && . "$HOME/.cargo/env" && which cargo && cargo check --workspace
         timeout: 600
       clippy:
-        run: . ~/.cargo/env && cargo clippy --workspace --all-targets -- -D warnings
+        run: . "$HOME/.cargo/env" && cargo clippy --workspace --all-targets -- -D warnings
         timeout: 600
       test:
         needs: [check]
-        run: . ~/.cargo/env && cargo xtask test essential
+        run: . "$HOME/.cargo/env" && cargo xtask test essential
         timeout: 900

crates/ci/src/local_runner.rs

@@ -4,7 +4,7 @@
 //! is ready to execute.
 
 use std::io::BufRead;
-use std::process::{Command, Stdio};
+use std::process::{Child, Command, Stdio};
 use std::sync::{Arc, Mutex};
 
 use swactor::actor::{ActorAddress, ActorInterface, Ctx};
@@ -41,7 +41,18 @@ impl LocalRunner {
         let work_dir = &self.start_job.work_dir;
         let timeout_secs = self.start_job.job_def.timeout_secs;
 
+        eprintln!(
+            "[runner] job {}/{} starting ({} commands, timeout {}s, workdir {})",
+            job_id.pipeline_id.0,
+            job_id.job_name,
+            self.start_job.job_def.run.len(),
+            timeout_secs,
+            work_dir,
+        );
+
         for cmd_str in &self.start_job.job_def.run {
+            eprintln!("[runner] exec: {cmd_str}");
+
             // Send progress: command being run.
             let _ = ctx.send(
                 self.coordinator_addr,
@@ -64,6 +75,7 @@ impl LocalRunner {
             let mut child = match child_result {
                 Ok(c) => c,
                 Err(e) => {
+                    eprintln!("[runner] spawn failed: {e}");
                     let _ = ctx.send(
                         self.coordinator_addr,
                         LocalCoordinatorMsg::JobComplete(JobComplete {
@@ -77,70 +89,36 @@ impl LocalRunner {
                 }
             };
 
-            // Timeout mechanism: share child handle, spawn thread that kills after timeout.
+            // Timeout: spawn a thread that kills the child after timeout_secs.
+            let child_id = child.id();
             let kill_flag = Arc::new(Mutex::new(false));
             let kill_flag_clone = Arc::clone(&kill_flag);
-            // Use a pipe to signal the timeout thread when the command finishes.
             let (done_tx, done_rx) = std::sync::mpsc::channel::<()>();
             let timeout_handle = std::thread::spawn(move || {
-                // Wait for either timeout or command completion.
+                if done_rx
-                if done_rx.recv_timeout(std::time::Duration::from_secs(timeout_secs)).is_err() {
+                    .recv_timeout(std::time::Duration::from_secs(timeout_secs))
+                    .is_err()
+                {
                     *kill_flag_clone.lock().unwrap() = true;
+                    // Actually kill the child process so the pipe readers unblock.
+                    let _ = std::process::Command::new("kill")
+                        .args(["-9", &child_id.to_string()])
+                        .status();
                 }
             });
 
-            // Read stdout line-by-line.
+            // Read stdout and stderr concurrently to avoid pipe-buffer deadlock.
-            let stdout = child.stdout.take();
+            let (last_lines, timed_out) =
-            let stderr = child.stderr.take();
+                drain_child_output(&mut child, job_id, ctx, self.coordinator_addr, &kill_flag);
-
-            let mut last_lines: Vec<String> = Vec::new();
-
-            if let Some(stdout) = stdout {
-                let reader = std::io::BufReader::new(stdout);
-                for line in reader.lines() {
-                    if let Ok(line) = line {
-                        let _ = ctx.send(
-                            self.coordinator_addr,
-                            LocalCoordinatorMsg::JobProgress(JobProgress {
-                                job_id: job_id.clone(),
-                                output_line: line.clone(),
-                            }),
-                        );
-                        last_lines.push(line);
-                        if last_lines.len() > 50 {
-                            last_lines.remove(0);
-                        }
-                    }
-                }
-            }
-
-            if let Some(stderr) = stderr {
-                let reader = std::io::BufReader::new(stderr);
-                for line in reader.lines() {
-                    if let Ok(line) = line {
-                        let _ = ctx.send(
-                            self.coordinator_addr,
-                            LocalCoordinatorMsg::JobProgress(JobProgress {
-                                job_id: job_id.clone(),
-                                output_line: format!("[stderr] {line}"),
-                            }),
-                        );
-                        last_lines.push(line);
-                        if last_lines.len() > 50 {
-                            last_lines.remove(0);
-                        }
-                    }
-                }
-            }
 
             let status = child.wait();
 
-            // Signal timeout thread that command finished.
+            // Signal timeout thread that we're done.
             let _ = done_tx.send(());
             let _ = timeout_handle.join();
 
-            // Check if killed by timeout.
+            if timed_out || *kill_flag.lock().unwrap() {
-            if *kill_flag.lock().unwrap() {
+                eprintln!("[runner] command timed out after {timeout_secs}s");
                 let _ = ctx.send(
                     self.coordinator_addr,
                     LocalCoordinatorMsg::JobComplete(JobComplete {
@@ -155,10 +133,14 @@ impl LocalRunner {
 
             match status {
                 Ok(exit) if exit.success() => {
-                    // Command passed, continue to next.
+                    eprintln!("[runner] command succeeded");
                 }
                 Ok(exit) => {
                     let exit_code = exit.code().unwrap_or(-1);
+                    eprintln!("[runner] command failed (exit {exit_code})");
+                    for line in &last_lines {
+                        eprintln!("[runner]   {line}");
+                    }
                     let _ = ctx.send(
                         self.coordinator_addr,
                         LocalCoordinatorMsg::JobComplete(JobComplete {
@@ -174,6 +156,7 @@ impl LocalRunner {
                     return;
                 }
                 Err(e) => {
+                    eprintln!("[runner] wait failed: {e}");
                     let _ = ctx.send(
                         self.coordinator_addr,
                         LocalCoordinatorMsg::JobComplete(JobComplete {
@@ -188,7 +171,10 @@ impl LocalRunner {
             }
         }
 
-        // All commands passed.
+        eprintln!(
+            "[runner] job {}/{} passed",
+            job_id.pipeline_id.0, job_id.job_name
+        );
         let _ = ctx.send(
             self.coordinator_addr,
             LocalCoordinatorMsg::JobComplete(JobComplete {
@@ -201,6 +187,87 @@ impl LocalRunner {
     }
 }
 
+/// Drain stdout and stderr from a child process concurrently.
+///
+/// Spawns a background thread for stderr so that both pipes are consumed
+/// in parallel, preventing the classic pipe-buffer deadlock where the child
+/// blocks writing to a full stderr while the parent blocks reading stdout.
+///
+/// Returns (last_lines, timed_out).
+fn drain_child_output(
+    child: &mut Child,
+    job_id: &crate::JobId,
+    ctx: &Ctx,
+    coordinator_addr: ActorAddress,
+    kill_flag: &Arc<Mutex<bool>>,
+) -> (Vec<String>, bool) {
+    let stdout = child.stdout.take();
+    let stderr = child.stderr.take();
+
+    // Collect stderr on a background thread.
+    let stderr_job_id = job_id.clone();
+    let stderr_kill = Arc::clone(kill_flag);
+    let stderr_handle = std::thread::spawn(move || {
+        let mut lines = Vec::new();
+        if let Some(stderr) = stderr {
+            let reader = std::io::BufReader::new(stderr);
+            for line in reader.lines() {
+                if *stderr_kill.lock().unwrap() {
+                    break;
+                }
+                if let Ok(line) = line {
+                    lines.push(line);
+                }
+            }
+        }
+        lines
+    });
+
+    // Read stdout on the current thread, streaming progress.
+    let mut last_lines: Vec<String> = Vec::new();
+    if let Some(stdout) = stdout {
+        let reader = std::io::BufReader::new(stdout);
+        for line in reader.lines() {
+            if let Ok(line) = line {
+                let _ = ctx.send(
+                    coordinator_addr,
+                    LocalCoordinatorMsg::JobProgress(JobProgress {
+                        job_id: job_id.clone(),
+                        output_line: line.clone(),
+                    }),
+                );
+                last_lines.push(line);
+                if last_lines.len() > 50 {
+                    last_lines.remove(0);
+                }
+            }
+        }
+    }
+
+    // Join stderr thread and stream its lines as progress.
+    let timed_out = *kill_flag.lock().unwrap();
+    let stderr_lines = stderr_handle.join().unwrap_or_default();
+    for line in &stderr_lines {
+        let _ = ctx.send(
+            coordinator_addr,
+            LocalCoordinatorMsg::JobProgress(JobProgress {
+                job_id: stderr_job_id.clone(),
+                output_line: format!("[stderr] {line}"),
+            }),
+        );
+    }
+
+    // Merge stderr into last_lines tail.
+    for line in stderr_lines {
+        last_lines.push(line);
+        if last_lines.len() > 50 {
+            last_lines.remove(0);
+        }
+    }
+
+    (last_lines, timed_out)
+}
+
 impl ActorInterface for LocalRunner {
     type Incoming = LocalRunnerMsg;
     type Response = ();

crates/ci/src/yaml.rs

@@ -429,4 +429,5 @@ pipelines:
         let multi = to_job_definition("multi", &check.jobs["multi"]);
         assert_eq!(multi.run, vec!["cargo fmt -- --check", "cargo test"]);
     }
+
 }