From be2f204585d00e695b81ce73c978d6543a1fc2d3 Mon Sep 17 00:00:00 2001 From: Jacob Stephens Date: Tue, 6 Oct 2026 15:01:48 -0400 Subject: [PATCH 1/2] Unify owned Harness process execution (#448) --- src/harness.rs | 2 +- src/harness/adapter.rs | 2 +- src/harness/muse.rs | 2 +- src/harness/opencode.rs | 2 +- src/harness/opencode/export.rs | 2 +- src/harness/process.rs | 261 +++++++++++++++++++++++++ src/harness/process/tests.rs | 337 +++++++++++++++++++++++++++++++++ src/harness/process_output.rs | 84 -------- src/session.rs | 93 +++------ tests/process_tree.rs | 38 +++- tests/progress_lines.rs | 17 +- 11 files changed, 673 insertions(+), 167 deletions(-) create mode 100644 src/harness/process.rs create mode 100644 src/harness/process/tests.rs delete mode 100644 src/harness/process_output.rs diff --git a/src/harness.rs b/src/harness.rs index b0ecdf1d0..86d978239 100644 --- a/src/harness.rs +++ b/src/harness.rs @@ -17,7 +17,7 @@ pub(crate) mod grok; mod instructions; pub(crate) mod muse; pub(crate) mod opencode; -mod process_output; +pub(crate) mod process; mod process_tree; mod skill_load; diff --git a/src/harness/adapter.rs b/src/harness/adapter.rs index 09aab6690..ffc31faa4 100644 --- a/src/harness/adapter.rs +++ b/src/harness/adapter.rs @@ -61,7 +61,7 @@ pub trait Adapter: Sync { } /// A session's arguments and optional prompt on stdin. `None` keeps stdin -/// null; `Some` is written once and closed before following the stream. +/// null; `Some` is written once and closed while output is drained concurrently. pub struct Invocation { pub args: Vec, pub stdin: Option, diff --git a/src/harness/muse.rs b/src/harness/muse.rs index 44c88b561..60b3e4796 100644 --- a/src/harness/muse.rs +++ b/src/harness/muse.rs @@ -135,7 +135,7 @@ pub fn check_model_and_effort(chosen: &ModelAndEffort) -> Result "checking the Model {model} with a test call to muse" )); let invocation = Muse.session(&choice, None, "Reply with OK."); - let output = super::process_output::output( + let output = super::process::output( &Muse, Command::new(Muse.name()).args(invocation.args), None, diff --git a/src/harness/opencode.rs b/src/harness/opencode.rs index ff4f254b5..413d523e3 100644 --- a/src/harness/opencode.rs +++ b/src/harness/opencode.rs @@ -115,7 +115,7 @@ pub fn check_model_and_effort(chosen: &ModelAndEffort) -> Result<()> { }; crate::progress::step("checking OpenCode's Model and Effort with a standalone test call"); let invocation = OpenCode.session(&choice, None, "Reply with OK."); - let output = super::process_output::output( + let output = super::process::output( &OpenCode, Command::new(OpenCode.name()).args(invocation.args), invocation.stdin.as_deref(), diff --git a/src/harness/opencode/export.rs b/src/harness/opencode/export.rs index e82a4b004..f6df4b9a4 100644 --- a/src/harness/opencode/export.rs +++ b/src/harness/opencode/export.rs @@ -100,7 +100,7 @@ impl SessionExport { pub fn read_after_exit(worktree: &Path, stream: Box) -> Result> { let export = if let Some(id) = stream.session_id() { - let output = super::super::process_output::output( + let output = super::super::process::output( &OpenCode, Command::new(OpenCode.name()) .args(["session", "export", "--standalone", id]) diff --git a/src/harness/process.rs b/src/harness/process.rs new file mode 100644 index 000000000..c1ab047cc --- /dev/null +++ b/src/harness/process.rs @@ -0,0 +1,261 @@ +//! Own a Harness child and all its I/O until completion or whole-tree cleanup. + +use std::io::{Read, Write}; +use std::os::unix::process::CommandExt; +use std::process::{Child, ChildStdout, Command, ExitStatus, Output, Stdio}; +use std::thread::{self, JoinHandle}; +use std::time::Duration; + +use anyhow::{Context, Result, anyhow, bail}; + +use super::Adapter; +use crate::interrupt; + +#[cfg(test)] +mod tests; + +/// Capture both output pipes without interpreting the child's exit status. +pub(super) fn output( + adapter: &dyn Adapter, + command: &mut Command, + input: Option<&str>, +) -> Result { + let cli = adapter.name(); + let (status, stdout, stderr) = execute(adapter, command, input, true, move |pipe| { + read(pipe).with_context(|| format!("could not read {cli}'s stdout")) + })?; + Ok(Output { + status, + stdout, + stderr, + }) +} + +/// Consume stdout concurrently with prompt I/O, inheriting stderr. Return the +/// consumer's opaque result only after the child and every I/O worker finish. +pub(crate) fn streaming( + adapter: &dyn Adapter, + command: &mut Command, + input: Option<&str>, + consume: impl FnOnce(ChildStdout) -> Result + Send + 'static, +) -> Result<(ExitStatus, T)> { + let (status, consumed, _) = execute(adapter, command, input, false, consume)?; + Ok((status, consumed)) +} + +fn read(mut pipe: impl Read) -> std::io::Result> { + let mut bytes = Vec::new(); + pipe.read_to_end(&mut bytes)?; + Ok(bytes) +} + +fn execute( + adapter: &dyn Adapter, + command: &mut Command, + input: Option<&str>, + capture_stderr: bool, + consume: impl FnOnce(ChildStdout) -> Result + Send + 'static, +) -> Result<(ExitStatus, T, Vec)> { + if interrupt::requested() { + bail!("interrupted"); + } + let cli = adapter.name(); + let child = command + .envs(adapter.environment().iter().copied()) + .process_group(0) + .stdin(if input.is_some() { + Stdio::piped() + } else { + Stdio::null() + }) + .stdout(Stdio::piped()) + .stderr(if capture_stderr { + Stdio::piped() + } else { + Stdio::inherit() + }) + .spawn() + .with_context(|| format!("could not run {cli}"))?; + // Install ownership immediately: every fallible step after spawn is + // covered, including pipe extraction and starting workers. + let mut owned = Execution { + adapter, + child, + stdout: None, + stderr: None, + stdin: None, + completed: false, + }; + let result = (|| { + let pipe = owned + .child + .stdout + .take() + .context("no stdout from Harness")?; + let reader = if capture_stderr { + format!("{cli}'s stdout reader") + } else { + "the session stream reader".to_string() + }; + owned.stdout = Some(Worker::start(reader, move || consume(pipe))?); + if capture_stderr { + let pipe = owned.child.stderr.take().context("no captured stderr")?; + owned.stderr = Some(Worker::start( + format!("{cli}'s stderr reader"), + move || read(pipe).with_context(|| format!("could not read {cli}'s stderr")), + )?); + } + if let Some(input) = input { + let mut pipe = owned.child.stdin.take().context("no prompt stdin")?; + let input = input.to_string(); + owned.stdin = Some(Worker::start( + format!("{cli}'s prompt writer"), + move || { + pipe.write_all(input.as_bytes()) + .with_context(|| format!("could not write the prompt to {cli}")) + // Dropping the pipe closes stdin, including for an empty prompt. + }, + )?); + } + let status = owned.wait()?; + let stdout = owned.stdout.as_mut().unwrap().value.take().unwrap(); + let stderr = owned + .stderr + .as_mut() + .and_then(|worker| worker.value.take()) + .unwrap_or_default(); + Ok((status, stdout, stderr)) + })(); + if result.is_err() || interrupt::requested() { + owned.cleanup(); + // Interruption can also arrive during stopping or joining. Preserve + // the original transport failure unless an interruption was observed. + if interrupt::requested() { + bail!("interrupted"); + } + } else { + owned.completed = true; + } + result +} + +struct Execution<'a, T> { + adapter: &'a dyn Adapter, + child: Child, + stdout: Option>, + stderr: Option>>, + stdin: Option>, + completed: bool, +} + +impl Execution<'_, T> { + fn wait(&mut self) -> Result { + let mut status = None; + loop { + if interrupt::requested() { + bail!("interrupted"); + } + if status.is_none() { + status = self + .child + .try_wait() + .with_context(|| format!("could not wait for {}", self.adapter.name()))?; + } + let stdin_done = self + .stdin + .as_mut() + .map(Worker::poll) + .transpose()? + .unwrap_or(true); + let stdout_done = self.stdout.as_mut().unwrap().poll()?; + let stderr_done = self + .stderr + .as_mut() + .map(Worker::poll) + .transpose()? + .unwrap_or(true); + if let Some(status) = status + && stdin_done + && stdout_done + && stderr_done + { + if interrupt::requested() { + bail!("interrupted"); + } + return Ok(status); + } + thread::sleep(Duration::from_millis(100)); + } + } + + fn cleanup(&mut self) { + if self.completed { + return; + } + self.adapter.stop(&mut self.child); + // Pipes may still be attached if worker startup failed. + drop(self.child.stdin.take()); + drop(self.child.stdout.take()); + drop(self.child.stderr.take()); + if let Some(worker) = &mut self.stdin { + worker.join(); + } + if let Some(worker) = &mut self.stdout { + worker.join(); + } + if let Some(worker) = &mut self.stderr { + worker.join(); + } + self.completed = true; + } +} + +impl Drop for Execution<'_, T> { + fn drop(&mut self) { + self.cleanup(); + } +} + +struct Worker { + handle: Option>>, + value: Option, + name: String, +} + +impl Worker { + fn start(name: String, work: impl FnOnce() -> Result + Send + 'static) -> Result + where + T: Send + 'static, + { + let handle = thread::Builder::new() + .spawn(work) + .with_context(|| format!("could not start {name}"))?; + Ok(Self { + handle: Some(handle), + value: None, + name, + }) + } + + /// Join only a completed worker while supervising the child. Its failure + /// or panic must initiate cleanup even if the child has not exited yet. + fn poll(&mut self) -> Result { + if self.handle.as_ref().is_some_and(JoinHandle::is_finished) { + self.value = Some(self.handle.take().unwrap().join().map_err(|panic| { + let cause = panic + .downcast_ref::() + .map(String::as_str) + .or_else(|| panic.downcast_ref::<&str>().copied()) + .unwrap_or("unknown panic"); + anyhow!("{} panicked: {cause}", self.name) + })??); + } + Ok(self.handle.is_none()) + } + + fn join(&mut self) { + if let Some(handle) = self.handle.take() { + let _ = handle.join(); + } + } +} diff --git a/src/harness/process/tests.rs b/src/harness/process/tests.rs new file mode 100644 index 000000000..f3f6a91d7 --- /dev/null +++ b/src/harness/process/tests.rs @@ -0,0 +1,337 @@ +//! Behavior through the same two operations used by sessions and checks. + +use super::*; +use crate::harness::Harness; + +use std::fs; +use std::path::Path; +use std::time::Instant; + +use tempfile::TempDir; + +fn shell(script: &str) -> Command { + let mut command = Command::new("bash"); + command.args(["-c", script]); + command +} + +#[test] +fn captured_completion_preserves_both_byte_streams_and_nonzero_status() { + let result = output( + Harness::OpenCode.adapter(), + &mut shell("printf '\\377out\\000\\n'; printf '\\376err\\000\\n' >&2; exit 7"), + None, + ) + .unwrap(); + + assert_eq!(result.status.code(), Some(7)); + assert_eq!(result.stdout, b"\xffout\0\n"); + assert_eq!(result.stderr, b"\xfeerr\0\n"); +} + +#[test] +fn streamed_completion_returns_raw_bytes_and_the_consumers_result() { + let (status, consumed) = streaming( + Harness::Grok.adapter(), + &mut shell("printf 'unknown\\n\\377malformed\\n'; exit 9"), + None, + |pipe| Ok((read(pipe)?, "finished")), + ) + .unwrap(); + + assert_eq!(status.code(), Some(9)); + assert_eq!(consumed, (b"unknown\n\xffmalformed\n".to_vec(), "finished")); +} + +#[test] +fn captured_prompt_is_null_when_absent_and_written_once_then_closed_when_present() { + for (input, expected) in [ + (None, "nullclosed"), + (Some(""), "closed"), + (Some("once"), "onceclosed"), + ] { + let result = output( + Harness::OpenCode.adapter(), + &mut shell("if [ /dev/fd/0 -ef /dev/null ]; then printf null; fi; cat; printf closed"), + input, + ) + .unwrap(); + assert!(result.status.success()); + assert_eq!(result.stdout, expected.as_bytes()); + } +} + +/// Cleanup uses only this fixture's recorded PID, including on assertion failure. +struct RecordedProcess(TempDir); + +impl RecordedProcess { + fn new() -> Self { + Self(tempfile::tempdir().unwrap()) + } + + fn command(&self, script: &str) -> Command { + let mut command = shell(script); + command.env("PID_FILE", self.0.path().join("pid")); + command + } + + fn pid(&self) -> libc::pid_t { + fs::read_to_string(self.0.path().join("pid")) + .unwrap() + .trim() + .parse() + .unwrap() + } + + fn assert_stopped(&self) { + assert!(!exists(self.pid()), "the owned child survived cleanup"); + } +} + +fn exists(pid: libc::pid_t) -> bool { + // SAFETY: signal 0 probes only the exact PID recorded by this test. + unsafe { libc::kill(pid, 0) == 0 } +} + +impl Drop for RecordedProcess { + fn drop(&mut self) { + if let Ok(pid) = fs::read_to_string(self.0.path().join("pid")) + && let Ok(pid) = pid.trim().parse() + && exists(pid) + { + // SAFETY: this exact PID belongs to the test's finite command. + unsafe { libc::kill(pid, libc::SIGKILL) }; + } + } +} + +#[test] +fn broken_prompt_stops_and_reaps_the_child_with_the_original_stdin_error() { + let child = RecordedProcess::new(); + let started = Instant::now(); + let result = output( + Harness::OpenCode.adapter(), + &mut child.command("echo $$ > \"$PID_FILE\"; exec 0<&-; exec sleep 5"), + Some(&"p".repeat(1024 * 1024)), + ); + + let error = result.unwrap_err(); + assert!( + format!("{error:#}").contains("could not write the prompt to opencode"), + "{error:#}" + ); + assert_eq!( + error.downcast_ref::().unwrap().kind(), + std::io::ErrorKind::BrokenPipe + ); + assert!(started.elapsed() < Duration::from_secs(2)); + child.assert_stopped(); +} + +struct FailingLog; + +impl Write for FailingLog { + fn write(&mut self, _bytes: &[u8]) -> std::io::Result { + Err(std::io::Error::other("log device failed")) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } +} + +#[test] +fn consumer_log_failure_cleans_up_even_with_a_blocked_prompt_writer() { + let child = RecordedProcess::new(); + let escaped = RecordedProcess::new(); + let mut command = child.command( + "set -m; trap 'wait; exit' TERM; echo $$ > \"$PID_FILE\"; \ + sleep 5 /dev/null 2>&1 & echo $! > \"$DESCENDANT_PID_FILE\"; \ + printf ready; wait", + ); + command.env("DESCENDANT_PID_FILE", escaped.0.path().join("pid")); + let started = Instant::now(); + let result = streaming( + Harness::OpenCode.adapter(), + &mut command, + Some(&"p".repeat(1024 * 1024)), + |mut pipe| std::io::copy(&mut pipe, &mut FailingLog).context("could not write Session log"), + ); + + assert_eq!( + format!("{:#}", result.unwrap_err()), + "could not write Session log: log device failed" + ); + assert!(started.elapsed() < Duration::from_secs(2)); + child.assert_stopped(); + escaped.assert_stopped(); +} + +#[test] +fn consumer_panic_stops_and_reaps_the_child_and_reports_its_cause() { + let child = RecordedProcess::new(); + let started = Instant::now(); + let result = streaming::<()>( + Harness::OpenCode.adapter(), + &mut child.command("echo $$ > \"$PID_FILE\"; printf ready; exec sleep 5"), + None, + |mut pipe| { + pipe.read_exact(&mut [0; 5])?; + panic!("consumer exploded"); + }, + ); + + assert_eq!( + format!("{:#}", result.unwrap_err()), + "the session stream reader panicked: consumer exploded" + ); + assert!(started.elapsed() < Duration::from_secs(2)); + child.assert_stopped(); +} + +#[test] +fn unread_captured_prompt_remains_interruptible() { + assert_unread_prompt_interruptible( + "harness::process::tests::unread_captured_prompt_remains_interruptible", + true, + ); +} + +#[test] +fn unread_streamed_prompt_remains_interruptible() { + assert_unread_prompt_interruptible( + "harness::process::tests::unread_streamed_prompt_remains_interruptible", + false, + ); +} + +#[test] +fn eof_does_not_finish_either_operation_before_the_child_exits() { + for captured in [true, false] { + let mut command = shell("exec 1>&- 2>&-; sleep 0.3; exit 7"); + let started = Instant::now(); + let status = if captured { + output(Harness::OpenCode.adapter(), &mut command, None) + .unwrap() + .status + } else { + streaming(Harness::OpenCode.adapter(), &mut command, None, |pipe| { + Ok(read(pipe)?) + }) + .unwrap() + .0 + }; + assert_eq!(status.code(), Some(7)); + assert!(started.elapsed() >= Duration::from_millis(250)); + } +} + +#[test] +fn both_operations_wait_for_output_held_after_child_exit() { + for captured in [true, false] { + let mut command = shell("(sleep 0.3; printf late) & exit 7"); + let (status, bytes) = if captured { + let result = output(Harness::OpenCode.adapter(), &mut command, None).unwrap(); + (result.status, result.stdout) + } else { + streaming(Harness::OpenCode.adapter(), &mut command, None, |pipe| { + Ok(read(pipe)?) + }) + .unwrap() + }; + assert_eq!(status.code(), Some(7)); + assert_eq!(bytes, b"late"); + } +} + +#[test] +fn normal_completion_leaves_background_work_without_owned_io_running() { + let work = RecordedProcess::new(); + let result = output( + Harness::OpenCode.adapter(), + &mut work.command("sleep 5 /dev/null 2>&1 & echo $! > \"$PID_FILE\"; exit 0"), + None, + ) + .unwrap(); + + assert!(result.status.success()); + assert!( + exists(work.pid()), + "normal completion killed background work" + ); +} + +#[test] +fn captured_completion_waits_for_stderr_after_stdout_and_child_exit() { + let result = output( + Harness::OpenCode.adapter(), + &mut shell("(exec 1>&-; sleep 0.3; printf late-diagnostic >&2) & exit 7"), + None, + ) + .unwrap(); + + assert_eq!(result.status.code(), Some(7)); + assert!(result.stdout.is_empty()); + assert_eq!(result.stderr, b"late-diagnostic"); +} + +fn assert_unread_prompt_interruptible(test_name: &str, captured: bool) { + const SIGNAL_DIR: &str = "THIRDSHIFT_PROCESS_TEST_SIGNAL_DIR"; + // Run this interface test alone in another process: the interruption + // predicate is process-wide and must not affect concurrent unit tests. + if let Some(dir) = std::env::var_os(SIGNAL_DIR) { + interrupt::install().unwrap(); + let owned = RecordedProcess::new(); + let mut command = + owned.command("echo $$ > \"$PID_FILE\"; touch \"$READY_FILE\"; exec sleep 5"); + command.env("READY_FILE", Path::new(&dir).join("ready")); + let prompt = "p".repeat(1024 * 1024); + let result = if captured { + output(Harness::OpenCode.adapter(), &mut command, Some(&prompt)).map(|_| ()) + } else { + streaming( + Harness::OpenCode.adapter(), + &mut command, + Some(&prompt), + |pipe| Ok(read(pipe)?), + ) + .map(|_| ()) + }; + assert_eq!(format!("{:#}", result.unwrap_err()), "interrupted"); + owned.assert_stopped(); + return; + } + + let dir = tempfile::tempdir().unwrap(); + let mut child = Command::new(std::env::current_exe().unwrap()) + .args(["--exact", test_name, "--nocapture"]) + .env(SIGNAL_DIR, dir.path()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + let deadline = Instant::now() + Duration::from_secs(10); + while !dir.path().join("ready").exists() && Instant::now() < deadline { + if child.try_wait().unwrap().is_some() { + break; + } + thread::sleep(Duration::from_millis(10)); + } + if !dir.path().join("ready").exists() { + let _ = child.kill(); + let result = child.wait_with_output().unwrap(); + panic!("the isolated test never started: {result:?}"); + } + let interrupted_at = Instant::now(); + // SAFETY: this is the exact PID of our isolated test process. + assert_eq!( + unsafe { libc::kill(child.id() as libc::pid_t, libc::SIGINT) }, + 0 + ); + let result = child.wait_with_output().unwrap(); + assert!(result.status.success(), "{result:?}"); + assert!( + interrupted_at.elapsed() < Duration::from_secs(2), + "the unread prompt prevented interruption" + ); +} diff --git a/src/harness/process_output.rs b/src/harness/process_output.rs deleted file mode 100644 index 2a9c301a1..000000000 --- a/src/harness/process_output.rs +++ /dev/null @@ -1,84 +0,0 @@ -//! Capture a Harness check or export while retaining session interruption rules. - -use std::io::{Read, Write}; -use std::os::unix::process::CommandExt; -use std::process::{Command, Output, Stdio}; -use std::thread; -use std::time::Duration; - -use anyhow::{Context, Result, anyhow, bail}; - -use super::Adapter; - -/// Isolate the CLI's process group and drain both pipes while watching for -/// interruption. Checks and exports use the same fixed environment and -/// whole-process-tree stop as the adapter's live sessions. -pub(super) fn output( - adapter: &dyn Adapter, - command: &mut Command, - input: Option<&str>, -) -> Result { - if crate::interrupt::requested() { - bail!("interrupted"); - } - let cli = adapter.name(); - let mut child = command - .envs(adapter.environment().iter().copied()) - .process_group(0) - .stdin(if input.is_some() { - Stdio::piped() - } else { - Stdio::null() - }) - .stdout(Stdio::piped()) - .stderr(Stdio::piped()) - .spawn() - .with_context(|| format!("could not run {cli}"))?; - - fn read(mut pipe: impl Read + Send + 'static) -> thread::JoinHandle>> { - thread::spawn(move || { - let mut bytes = Vec::new(); - pipe.read_to_end(&mut bytes)?; - Ok(bytes) - }) - } - let stdout = read(child.stdout.take().context("no captured stdout")?); - let stderr = read(child.stderr.take().context("no captured stderr")?); - if let Some(input) = input - && let Some(mut stdin) = child.stdin.take() - && let Err(error) = stdin.write_all(input.as_bytes()) - { - adapter.stop(&mut child); - let _ = stdout.join(); - let _ = stderr.join(); - return Err(error).with_context(|| format!("could not write the prompt to {cli}")); - } - let status = loop { - if crate::interrupt::requested() { - adapter.stop(&mut child); - let _ = stdout.join(); - let _ = stderr.join(); - bail!("interrupted"); - } - if let Some(status) = child - .try_wait() - .with_context(|| format!("could not wait for {cli}"))? - // A child command can retain either pipe after the CLI exits. - // Keep watching interruptions until both readers can be joined. - && stdout.is_finished() - && stderr.is_finished() - { - break status; - } - thread::sleep(Duration::from_millis(100)); - }; - Ok(Output { - status, - stdout: stdout - .join() - .map_err(|_| anyhow!("{cli}'s stdout reader panicked"))??, - stderr: stderr - .join() - .map_err(|_| anyhow!("{cli}'s stderr reader panicked"))??, - }) -} diff --git a/src/session.rs b/src/session.rs index 107d94d07..15694f57f 100644 --- a/src/session.rs +++ b/src/session.rs @@ -3,20 +3,17 @@ use std::cell::RefCell; use std::fs::{self, File}; use std::io::{BufRead, BufReader, Read, Write}; -use std::os::unix::process::CommandExt; use std::path::{Path, PathBuf}; -use std::process::{Command, Stdio}; -use std::thread; +use std::process::Command; use std::time::{Duration, Instant}; -use anyhow::{Context, Result, anyhow, bail}; +use anyhow::{Context, Result, bail}; #[cfg(test)] pub(crate) use crate::harness::claude::claude_args; #[cfg(test)] pub(crate) use crate::harness::codex::{codex_args, codex_prompt}; -use crate::harness::{Adapter, Choice, Invocation}; -use crate::interrupt; +use crate::harness::{Adapter, Choice, Invocation, process}; use crate::issue::{IssueUrl, Repo}; use crate::logs; use crate::progress::{self, Stream}; @@ -289,69 +286,24 @@ fn run( let mut log_file = File::create(log).with_context(|| format!("could not create {}", log.display()))?; let started = Instant::now(); - let mut child = Command::new(cli) - .args(&invocation.args) - .envs(adapter.environment().iter().copied()) - .current_dir(worktree) - .stdin(if invocation.stdin.is_some() { - Stdio::piped() - } else { - Stdio::null() - }) - .stdout(Stdio::piped()) - // Its own process group, so thirdshift decides how it is stopped and - // can stop everything it started. - .process_group(0) - .spawn() - .with_context(|| format!("could not run {cli}"))?; - - // Follow the stream on its own thread, so this one can watch for an - // interrupt while the session runs. - let output = child - .stdout - .take() - .with_context(|| format!("no stdout from {cli}"))?; - let group = -(child.id() as libc::pid_t); let (kind_owned, log_owned) = (kind.to_string(), log.to_path_buf()); - let follower = thread::spawn(move || { - let followed = follow( - &kind_owned, - output, - &mut log_file, - &log_owned, - stream.as_mut(), - ); - if followed.is_err() { - // Nothing reads its output any more, so it could block forever. - // SAFETY: kill has no memory-safety preconditions. - unsafe { libc::kill(group, libc::SIGKILL) }; - } - (followed, stream) - }); - if let Some(input) = invocation.stdin - && let Some(mut stdin) = child.stdin.take() - && let Err(error) = stdin.write_all(input.as_bytes()) - { - adapter.stop(&mut child); - let _ = follower.join(); - return Err(error).with_context(|| format!("could not write the prompt to {cli}")); - } - let status = loop { - if interrupt::requested() { - adapter.stop(&mut child); - bail!("interrupted"); - } - if let Some(status) = child - .try_wait() - .with_context(|| format!("could not wait for {cli}"))? - { - break status; - } - thread::sleep(POLL); - }; - let (followed, stream) = follower - .join() - .map_err(|_| anyhow!("the session stream reader panicked"))?; + let (status, stream) = process::streaming( + adapter, + Command::new(cli) + .args(&invocation.args) + .current_dir(worktree), + invocation.stdin.as_deref(), + move |output| { + follow( + &kind_owned, + output, + &mut log_file, + &log_owned, + stream.as_mut(), + )?; + Ok(stream) + }, + )?; let stream = adapter.read_after_exit(worktree, stream)?; for warning in stream.warnings() { @@ -364,7 +316,6 @@ fn run( progress::step(format_args!( "{kind}: session ended after {elapsed}{summary}" )); - followed?; if status.success() && !stream.failed() { return Ok(stream); } @@ -384,8 +335,6 @@ fn run( } } -const POLL: Duration = Duration::from_millis(100); - /// Copy every line of `stream` to `log_file` and print the progress lines it /// condenses to, until the stream ends. fn follow( @@ -429,6 +378,8 @@ mod tests { use std::collections::VecDeque; use std::rc::Rc; + use anyhow::anyhow; + use super::*; /// What the rules asked of the scripted [`Outside`], in order. diff --git a/tests/process_tree.rs b/tests/process_tree.rs index 6af9d5a38..6654a9025 100644 --- a/tests/process_tree.rs +++ b/tests/process_tree.rs @@ -81,6 +81,35 @@ fn interrupting_a_muse_check_after_cli_exit_stops_the_command_holding_its_stream assert_session_command_stopped("muse", During::CheckAfterExit); } +#[test] +fn interrupting_a_live_session_after_cli_exit_stops_the_command_holding_stdout() { + let scenario = Scenario::new(); + let pid = scenario.path("stdout-holder-pid"); + scenario.agent_does(&format!( + r#"bash -c 'while kill -0 "$1" 2>/dev/null; do sleep 0.01; done +echo $$ > "$2" +touch "$3" +exec sleep 5' holder "$PPID" "{pid}" "{started}" & +"#, + pid = pid.display(), + started = scenario.path("cli-exited").display(), + )); + let url = scenario.issue_url(7); + let mut held = scenario.run_until(&[&url], &[], "cli-exited"); + let command = SessionCommand(fs::read_to_string(pid).unwrap().trim().parse().unwrap()); + let interrupted_at = Instant::now(); + held.signal("INT"); + let result = held.finish(); + + assert_eq!(result.code, Some(1), "{}", result.stderr); + assert!(result.stderr.contains("interrupted"), "{}", result.stderr); + assert!( + interrupted_at.elapsed() < Duration::from_secs(2), + "interruption waited for the stdout holder's natural expiry" + ); + command.assert_stopped("the stdout holder survived interruption"); +} + #[test] fn interrupting_an_opencode_export_stops_its_whole_process_tree() { assert_session_command_stopped("opencode", During::Export); @@ -112,7 +141,7 @@ fn assert_session_command_stopped(harness: &str, during: During) { r#"while kill -0 "$FAKE_MUSE_CHECK_PID" 2>/dev/null; do sleep 0.01; done echo $$ > "{pid}" touch "{started}" -exec sleep 9 +exec sleep 5 "#, pid = pid.display(), started = scenario.path("agent-started").display(), @@ -172,8 +201,13 @@ touch "{started}" assert!(scenario.issue_labels(7).is_empty()); assert_eq!(scenario.entries("work"), ["widgets"]); } + let limit = if matches!(during, During::CheckAfterExit) { + Duration::from_secs(2) + } else { + Duration::from_secs(8) + }; assert!( - interrupted_at.elapsed() < Duration::from_secs(8), + interrupted_at.elapsed() < limit, "interruption waited for the command instead of stopping it" ); command.assert_stopped("the session's command survived interruption"); diff --git a/tests/progress_lines.rs b/tests/progress_lines.rs index 05879c5f1..1beaf0f92 100644 --- a/tests/progress_lines.rs +++ b/tests/progress_lines.rs @@ -178,20 +178,27 @@ fn skips_unknown_and_malformed_stream_events_and_keeps_stdout_to_the_pr_url() { fn still_logs_the_whole_stream() { let scenario = Scenario::new(); scenario.agent_does(&format!( - "echo 'not json at all'\n{AGENT_COMMITS_AND_OPENS_PR}" + "echo 'not json at all'\nprintf '\\377unknown\\000\\n'\necho '{{\"type\":'\n{AGENT_COMMITS_AND_OPENS_PR}" )); - scenario.run(&[&scenario.issue_url(7)]); + let result = scenario.run(&[&scenario.issue_url(7)]); + assert_eq!(result.code, Some(0), "{}", result.stderr); let logs = scenario.entries("home/.thirdshift/logs/acme/widgets/sessions"); - let log = std::fs::read_to_string( + let log = std::fs::read( scenario .path("home/.thirdshift/logs/acme/widgets/sessions") .join(&logs[0]), ) .unwrap(); - assert!(log.contains("not json at all\n"), "log: {log}"); - assert!(log.contains(r#""type": "result""#), "log: {log}"); + let raw = b"not json at all\n\xffunknown\0\n{\"type\":\n"; + assert!(log.windows(raw.len()).any(|bytes| bytes == raw), "{log:?}"); + let result_event = br#""type": "result""#; + assert!( + log.windows(result_event.len()) + .any(|bytes| bytes == result_event), + "{log:?}" + ); } #[test] From e046bdfac521a94f80759242dc227df88c342a1b Mon Sep 17 00:00:00 2001 From: Jacob Stephens Date: Tue, 6 Oct 2026 15:08:28 -0400 Subject: [PATCH 2/2] Preserve session recovery and reporting after stream failures (#448) --- src/session.rs | 54 ++++++++++++++++----- tests/progress_lines.rs | 104 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 146 insertions(+), 12 deletions(-) diff --git a/src/session.rs b/src/session.rs index 15694f57f..9899d8f95 100644 --- a/src/session.rs +++ b/src/session.rs @@ -5,6 +5,7 @@ use std::fs::{self, File}; use std::io::{BufRead, BufReader, Read, Write}; use std::path::{Path, PathBuf}; use std::process::Command; +use std::sync::mpsc; use std::time::{Duration, Instant}; use anyhow::{Context, Result, bail}; @@ -14,6 +15,7 @@ pub(crate) use crate::harness::claude::claude_args; #[cfg(test)] pub(crate) use crate::harness::codex::{codex_args, codex_prompt}; use crate::harness::{Adapter, Choice, Invocation, process}; +use crate::interrupt; use crate::issue::{IssueUrl, Repo}; use crate::logs; use crate::progress::{self, Stream}; @@ -287,25 +289,52 @@ fn run( File::create(log).with_context(|| format!("could not create {}", log.display()))?; let started = Instant::now(); let (kind_owned, log_owned) = (kind.to_string(), log.to_path_buf()); - let (status, stream) = process::streaming( + // A consumer failure still leaves a useful partial stream for recovery + // and reporting. The process owner joins the consumer before returning, + // so this bounded handoff never makes the caller wait on an I/O worker. + let (failed_stream, partial_stream) = mpsc::sync_channel(1); + let executed = process::streaming( adapter, Command::new(cli) .args(&invocation.args) .current_dir(worktree), invocation.stdin.as_deref(), - move |output| { - follow( - &kind_owned, - output, - &mut log_file, - &log_owned, - stream.as_mut(), - )?; - Ok(stream) + move |output| match follow( + &kind_owned, + output, + &mut log_file, + &log_owned, + stream.as_mut(), + ) { + Ok(()) => Ok(stream), + Err(error) => { + let _ = failed_stream.send(stream); + Err(error) + } }, - )?; + ); + let (status, stream) = match executed { + Ok((status, stream)) => (Ok(status), stream), + Err(error) => { + if interrupt::requested() { + return Err(error); + } + match partial_stream.try_recv() { + Ok(stream) => (Err(error), stream), + Err(_) => return Err(error), + } + } + }; - let stream = adapter.read_after_exit(worktree, stream)?; + let stream = match adapter.read_after_exit(worktree, stream) { + Ok(stream) => stream, + Err(error) => { + if interrupt::requested() { + return Err(error); + } + return Err(status.err().unwrap_or(error)); + } + }; for warning in stream.warnings() { progress::step(format_args!("{kind}: {warning}")); } @@ -316,6 +345,7 @@ fn run( progress::step(format_args!( "{kind}: session ended after {elapsed}{summary}" )); + let status = status?; if status.success() && !stream.failed() { return Ok(stream); } diff --git a/tests/progress_lines.rs b/tests/progress_lines.rs index 1beaf0f92..48303a54a 100644 --- a/tests/progress_lines.rs +++ b/tests/progress_lines.rs @@ -5,6 +5,9 @@ mod support; use serde_json::json; +use std::fs::{self, File}; +use std::io::{BufRead, BufReader}; +use std::process::Command; use support::Scenario; /// The agent commits its work and opens a PR, but leaves the pushing to @@ -201,6 +204,107 @@ fn still_logs_the_whole_stream() { ); } +#[test] +fn a_log_failure_still_recovers_and_reports_warnings_and_usage() { + let scenario = Scenario::new(); + let pipe = scenario.path("session-log-pipe"); + assert!( + Command::new("mkfifo") + .arg(&pipe) + .status() + .unwrap() + .success() + ); + let logs = scenario.path("home/.thirdshift/logs/acme/widgets"); + // Install the failing log before the session opens it. The Command log + // supplies the actual start stamp, so this never predicts the clock. + scenario.repo_has_hook( + &scenario.launch_dir(), + "post-checkout", + &format!( + r#"#!/bin/sh +for log in "{logs}/commands/issue/"*.log; do + name=${{log##*/}} + stamp=${{name#7-}} + stamp=${{stamp%.log}} +done +mkdir -p "{logs}/sessions" +ln -s "{pipe}" "{logs}/sessions/7-$stamp-implement.jsonl" +touch "{ready}" +"#, + logs = logs.display(), + pipe = pipe.display(), + ready = scenario.path("log-ready").display(), + ), + ); + // Recovery data is already on disk when the live consumer fails. + fs::write( + scenario.path("opencode-calls.fake-opencode-1.export.json"), + json!({ + "info": {"outcome": "success"}, + "messages": [{ + "type": "assistant", "tokens": {"input": 42, "output": 7}, + "content": [{"type": "text", "text": "recovered reply"}] + }] + }) + .to_string(), + ) + .unwrap(); + let closed = scenario.path("log-closed"); + scenario.agent_does(&format!( + r#"echo ready-to-fail +for attempt in {{1..500}}; do + test -f "{closed}" && break + sleep 0.01 +done +echo another-line +exec sleep 5 +"#, + closed = closed.display(), + )); + let url = scenario.issue_url(7); + let held = scenario.run_until( + &["harness", "opencode", &url], + &[("FAKE_OPENCODE_SKIP_SKILL", "1")], + "log-ready", + ); + let mut reader = BufReader::new(File::open(&pipe).unwrap()); + let mut line = String::new(); + loop { + line.clear(); + if reader.read_line(&mut line).unwrap() == 0 || line == "ready-to-fail\n" { + break; + } + } + // The next log write fails with BrokenPipe after the stream has its id. + drop(reader); + fs::write(closed, "").unwrap(); + let result = held.finish(); + + assert_eq!(result.code, Some(1), "{}", result.stderr); + assert!( + result.stderr.contains("could not write"), + "{}", + result.stderr + ); + assert!(result.stderr.contains("Broken pipe"), "{}", result.stderr); + let warning = "warning: the session never loaded thirdshift-implement with its skill tool"; + assert_eq!( + result.stderr.matches(warning).count(), + 1, + "{}", + result.stderr + ); + assert!( + result.stderr.contains("implement: session ended after") + && result.stderr.contains( + "42 input tokens (0 cache read, 0 cache write), 7 output tokens (0 reasoning)" + ), + "{}", + result.stderr + ); +} + #[test] fn every_line_starts_with_the_local_time_it_was_printed() { let scenario = Scenario::new();