From 9db097f27822fedc7cc5f54c6619276c03a1e208 Mon Sep 17 00:00:00 2001 From: Jacob Stephens Date: Tue, 6 Oct 2026 19:02:13 -0400 Subject: [PATCH] Retain Harness interpretation after transport failures (#461) --- src/harness/process.rs | 36 ++++- src/harness/process/tests.rs | 141 +++++++++++++---- src/session.rs | 36 +---- src/session/execution_tests.rs | 275 +++++++++++++++++++++++++++++++++ 4 files changed, 424 insertions(+), 64 deletions(-) create mode 100644 src/session/execution_tests.rs diff --git a/src/harness/process.rs b/src/harness/process.rs index c1ab047cc..888f20bab 100644 --- a/src/harness/process.rs +++ b/src/harness/process.rs @@ -3,6 +3,7 @@ use std::io::{Read, Write}; use std::os::unix::process::CommandExt; use std::process::{Child, ChildStdout, Command, ExitStatus, Output, Stdio}; +use std::sync::mpsc; use std::thread::{self, JoinHandle}; use std::time::Duration; @@ -31,16 +32,39 @@ pub(super) fn output( }) } -/// 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. +/// Recovered consumer state accompanies an independent execution result. +#[derive(Debug)] +pub(crate) struct Streamed { + pub execution: Result, + pub state: T, +} + +/// Consume stdout concurrently with prompt I/O, inheriting stderr. Preserve +/// state after any normal consumer return, even when execution fails. Return +/// only after completion or whole-tree cleanup and every worker is joined. +/// A consumer panic or failure before it starts returns no state. Recorded +/// interruption suppresses recovery even after a normal consumer return. 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)) + mut state: T, + consume: impl FnOnce(ChildStdout, &mut T) -> Result<()> + Send + 'static, +) -> Result> { + // One send fits without waiting for the caller, which must first finish + // supervising and joining all workers, including on transport failure. + let (completed, recovered) = mpsc::sync_channel(1); + let execution = execute(adapter, command, input, false, move |pipe| { + let result = consume(pipe, &mut state); + let _ = completed.send(state); + result + }) + .map(|(status, (), _)| status); + interrupt::check()?; + match recovered.try_recv() { + Ok(state) => Ok(Streamed { execution, state }), + Err(_) => Err(execution.expect_err("a successful consumer always returns its state")), + } } fn read(mut pipe: impl Read) -> std::io::Result> { diff --git a/src/harness/process/tests.rs b/src/harness/process/tests.rs index f3f6a91d7..aafa97cb5 100644 --- a/src/harness/process/tests.rs +++ b/src/harness/process/tests.rs @@ -30,17 +30,25 @@ fn captured_completion_preserves_both_byte_streams_and_nonzero_status() { } #[test] -fn streamed_completion_returns_raw_bytes_and_the_consumers_result() { - let (status, consumed) = streaming( +fn streamed_completion_preserves_raw_bytes_and_nonzero_status_in_consumer_state() { + let consumed = streaming( Harness::Grok.adapter(), &mut shell("printf 'unknown\\n\\377malformed\\n'; exit 9"), None, - |pipe| Ok((read(pipe)?, "finished")), + (Vec::new(), "initial"), + |pipe, state| { + state.0 = read(pipe)?; + state.1 = "finished"; + Ok(()) + }, ) .unwrap(); - assert_eq!(status.code(), Some(9)); - assert_eq!(consumed, (b"unknown\n\xffmalformed\n".to_vec(), "finished")); + assert_eq!(consumed.execution.unwrap().code(), Some(9)); + assert_eq!( + consumed.state, + (b"unknown\n\xffmalformed\n".to_vec(), "finished") + ); } #[test] @@ -155,11 +163,21 @@ fn consumer_log_failure_cleans_up_even_with_a_blocked_prompt_writer() { 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"), - ); + b"initial".to_vec(), + |mut pipe, state| { + let mut bytes = [0; 5]; + pipe.read_exact(&mut bytes)?; + state.extend_from_slice(&bytes); + FailingLog + .write_all(&bytes) + .context("could not write Session log") + }, + ) + .unwrap(); + assert_eq!(result.state, b"initialready"); assert_eq!( - format!("{:#}", result.unwrap_err()), + format!("{:#}", result.execution.unwrap_err()), "could not write Session log: log device failed" ); assert!(started.elapsed() < Duration::from_secs(2)); @@ -167,6 +185,31 @@ fn consumer_log_failure_cleans_up_even_with_a_blocked_prompt_writer() { escaped.assert_stopped(); } +#[test] +fn prompt_failure_retains_state_and_precedes_a_consumer_failure_during_cleanup() { + let child = RecordedProcess::new(); + let result = streaming( + Harness::OpenCode.adapter(), + &mut child.command("echo $$ > \"$PID_FILE\"; printf ready; exec 0<&-; exec sleep 5"), + Some(&"p".repeat(1024 * 1024)), + b"initial".to_vec(), + |mut pipe, state| { + pipe.read_to_end(state)?; + bail!("later consumer failure") + }, + ) + .unwrap(); + + assert_eq!(result.state, b"initialready"); + let error = result.execution.unwrap_err(); + assert_eq!(error.to_string(), "could not write the prompt to opencode"); + assert_eq!( + error.downcast_ref::().unwrap().kind(), + std::io::ErrorKind::BrokenPipe + ); + child.assert_stopped(); +} + #[test] fn consumer_panic_stops_and_reaps_the_child_and_reports_its_cause() { let child = RecordedProcess::new(); @@ -175,7 +218,8 @@ fn consumer_panic_stops_and_reaps_the_child_and_reports_its_cause() { Harness::OpenCode.adapter(), &mut child.command("echo $$ > \"$PID_FILE\"; printf ready; exec sleep 5"), None, - |mut pipe| { + (), + |mut pipe, _state| { pipe.read_exact(&mut [0; 5])?; panic!("consumer exploded"); }, @@ -215,11 +259,19 @@ fn eof_does_not_finish_either_operation_before_the_child_exits() { .unwrap() .status } else { - streaming(Harness::OpenCode.adapter(), &mut command, None, |pipe| { - Ok(read(pipe)?) - }) + streaming( + Harness::OpenCode.adapter(), + &mut command, + None, + Vec::new(), + |pipe, state| { + *state = read(pipe)?; + Ok(()) + }, + ) + .unwrap() + .execution .unwrap() - .0 }; assert_eq!(status.code(), Some(7)); assert!(started.elapsed() >= Duration::from_millis(250)); @@ -234,10 +286,18 @@ fn both_operations_wait_for_output_held_after_child_exit() { 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() + let result = streaming( + Harness::OpenCode.adapter(), + &mut command, + None, + Vec::new(), + |pipe, state| { + *state = read(pipe)?; + Ok(()) + }, + ) + .unwrap(); + (result.execution.unwrap(), result.state) }; assert_eq!(status.code(), Some(7)); assert_eq!(bytes, b"late"); @@ -246,19 +306,36 @@ fn both_operations_wait_for_output_held_after_child_exit() { #[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(); + for captured in [true, false] { + let work = RecordedProcess::new(); + let mut command = + work.command("sleep 5 /dev/null 2>&1 & echo $! > \"$PID_FILE\"; exit 0"); + let status = if captured { + output(Harness::OpenCode.adapter(), &mut command, None) + .unwrap() + .status + } else { + streaming( + Harness::OpenCode.adapter(), + &mut command, + None, + (), + |pipe, _state| { + read(pipe)?; + Ok(()) + }, + ) + .unwrap() + .execution + .unwrap() + }; - assert!(result.status.success()); - assert!( - exists(work.pid()), - "normal completion killed background work" - ); + assert!(status.success()); + assert!( + exists(work.pid()), + "normal completion killed background work" + ); + } } #[test] @@ -293,7 +370,11 @@ fn assert_unread_prompt_interruptible(test_name: &str, captured: bool) { Harness::OpenCode.adapter(), &mut command, Some(&prompt), - |pipe| Ok(read(pipe)?), + Vec::new(), + |pipe, state| { + *state = read(pipe)?; + Ok(()) + }, ) .map(|_| ()) }; diff --git a/src/session.rs b/src/session.rs index 3cf60e500..68a51b2e2 100644 --- a/src/session.rs +++ b/src/session.rs @@ -5,7 +5,6 @@ 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}; #[cfg(test)] @@ -18,13 +17,15 @@ pub(crate) use crate::harness::claude::claude_args; pub(crate) use crate::harness::codex::{codex_args, codex_prompt}; use crate::harness::interpretation::{Ended, Interpretation}; use crate::harness::{Adapter, Choice, Invocation, process}; -use crate::interrupt; use crate::issue::{IssueUrl, Repo}; use crate::logs; use crate::progress; use crate::prompt; use crate::skills; +#[cfg(test)] +mod execution_tests; + /// Where a Run's or an Architect run's Session logs go: in `sessions/` under /// the root of its repository's logs, under the User config's `logs.dir` or /// `~/.thirdshift/logs`, each named for what is run and stamped with the @@ -263,7 +264,7 @@ fn run( worktree: &Path, invocation: Invocation, log: &Path, - mut stream: Interpretation, + stream: Interpretation, ) -> Result { let cli = adapter.name(); if let Some(dir) = log.parent() { @@ -273,38 +274,17 @@ 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()); - // 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| match follow(&kind_owned, output, &mut log_file, &log_owned, &mut stream) { - 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), - } - } - }; + stream, + move |output, stream| follow(&kind_owned, output, &mut log_file, &log_owned, stream), + )?; - let completion = stream.finish(status); + let completion = executed.state.finish(executed.execution); if let Some(report) = completion.report { for warning in report.warnings { progress::step(format_args!("{kind}: {warning}")); diff --git a/src/session/execution_tests.rs b/src/session/execution_tests.rs new file mode 100644 index 000000000..3cf5a14b6 --- /dev/null +++ b/src/session/execution_tests.rs @@ -0,0 +1,275 @@ +//! Exercise real Session execution with local CLIs in isolated processes. + +use super::*; +use crate::harness::Harness; +use crate::interrupt; +use std::os::unix::fs::PermissionsExt; +use std::process::{Output, Stdio}; + +const FIXTURE: &str = "THIRDSHIFT_SESSION_TEST_DIR"; +const WARNING: &str = "warning: the session never loaded thirdshift-implement with its skill tool"; +const USAGE: &str = "42 input tokens (0 cache read, 0 cache write), 7 output tokens (0 reasoning)"; + +struct Fixture(tempfile::TempDir); + +impl Fixture { + fn new(script: &str) -> Self { + let fixture = Self(tempfile::tempdir().unwrap()); + fs::create_dir(fixture.path("bin")).unwrap(); + fs::create_dir(fixture.path("logs")).unwrap(); + let cli = fixture.path("bin/opencode"); + fs::write( + &cli, + format!( + r#"#!/bin/bash +if test "$1" = session; then + echo export >> "$THIRDSHIFT_SESSION_TEST_DIR/calls" + cat "$THIRDSHIFT_SESSION_TEST_DIR/export.json" + exit 0 +fi +echo session >> "$THIRDSHIFT_SESSION_TEST_DIR/calls" +echo $$ > "$THIRDSHIFT_SESSION_TEST_DIR/pid" +echo '{{"type":"step_start","sessionID":"s1","part":{{}}}}' +{script} +"# + ), + ) + .unwrap(); + fs::set_permissions(cli, fs::Permissions::from_mode(0o755)).unwrap(); + // Retained failure and usage must not replace the transport error. + fs::write( + fixture.path("export.json"), + r#"{"info":{"outcome":"failed","error":{"message":"later retained failure"}},"messages":[{"type":"assistant","tokens":{"input":42,"output":7},"content":[{"type":"text","text":"retained reply"}]}]}"#, + ) + .unwrap(); + fixture + } + + fn path(&self, path: &str) -> PathBuf { + self.0.path().join(path) + } + + fn command(&self, test: &str) -> Command { + let mut command = Command::new(std::env::current_exe().unwrap()); + let path = std::env::join_paths( + std::iter::once(self.path("bin")) + .chain(std::env::split_paths(&std::env::var_os("PATH").unwrap())), + ) + .unwrap(); + command + .args(["--exact", test, "--nocapture"]) + .env(FIXTURE, self.0.path()) + .env("PATH", path); + command + } + + fn assert_reporting(&self, output: Output) { + let stderr = String::from_utf8_lossy(&output.stderr); + assert!(output.status.success(), "{output:?}"); + assert_eq!(stderr.matches(WARNING).count(), 1, "{stderr}"); + assert_eq!(stderr.matches(USAGE).count(), 1, "{stderr}"); + assert_eq!(stderr.matches("session ended after").count(), 1, "{stderr}"); + assert!(!stderr.contains("resuming"), "{stderr}"); + assert_eq!( + fs::read_to_string(self.path("calls")).unwrap(), + "session\nexport\n" + ); + } +} + +impl Drop for Fixture { + fn drop(&mut self) { + if let Ok(pid) = fs::read_to_string(self.path("pid")) + && let Ok(pid) = pid.trim().parse::() + { + // SAFETY: clean up only this fixture's exact child and group if + // an assertion failed before production cleanup could finish. + unsafe { + libc::kill(-pid, libc::SIGKILL); + libc::kill(pid, libc::SIGKILL); + } + } + } +} + +fn execute_session(root: &Path) -> anyhow::Error { + interrupt::install().unwrap(); + logs::begin(logs::Begin::ChildRun("fixture")); + let worktree = root.join("worktree"); + fs::create_dir(&worktree).unwrap(); + assert!( + Command::new("git") + .args(["init", "-q"]) + .arg(&worktree) + .status() + .unwrap() + .success() + ); + let logs = Logs { + name: "7".into(), + dir: root.join("logs"), + }; + let choice = Choice { + harness: Harness::OpenCode, + ..Choice::default() + }; + let prompt = format!("/thirdshift-implement {}", "p".repeat(1024 * 1024)); + let (result, log) = Sessions::within(&logs, &worktree, &choice, |sessions| { + sessions.run_to_final_message("implement", &prompt) + }); + assert!(log.unwrap().exists()); + result.unwrap_err() +} + +fn assert_owned_child_stopped(root: &Path) { + let pid: libc::pid_t = fs::read_to_string(root.join("pid")) + .unwrap() + .trim() + .parse() + .unwrap(); + // SAFETY: probe only this fixture's recorded PID. + assert_ne!(unsafe { libc::kill(pid, 0) }, 0, "owned child survived"); +} + +#[test] +fn completed_stdout_is_reported_after_a_broken_prompt() { + if let Some(root) = std::env::var_os(FIXTURE) { + let root = Path::new(&root); + let error = execute_session(root); + assert!( + error + .to_string() + .contains("could not write the prompt to opencode"), + "{error:#}" + ); + assert_eq!( + error.downcast_ref::().unwrap().kind(), + std::io::ErrorKind::BrokenPipe + ); + assert_owned_child_stopped(root); + return; + } + let fixture = Fixture::new( + r#"echo '{"type":"error","sessionID":"s1","error":{"message":"later turn failure"}}' +exec 1>&- +for attempt in {1..500}; do + grep -q 'later turn failure' "$THIRDSHIFT_SESSION_TEST_DIR"/logs/*.jsonl && break + sleep 0.01 +done +exec 0<&- +exec sleep 5"#, + ); + fixture.assert_reporting( + fixture + .command("session::execution_tests::completed_stdout_is_reported_after_a_broken_prompt") + .output() + .unwrap(), + ); +} + +#[test] +fn partial_stdout_is_reported_after_log_failure_with_a_blocked_prompt() { + if let Some(root) = std::env::var_os(FIXTURE) { + let root = Path::new(&root); + let error = execute_session(root); + assert!( + error.to_string().contains("could not write") + && error.to_string().contains("implement.jsonl"), + "{error:#}" + ); + assert_eq!( + error.downcast_ref::().unwrap().kind(), + std::io::ErrorKind::BrokenPipe + ); + assert_owned_child_stopped(root); + return; + } + let fixture = Fixture::new( + r#"echo ready-to-fail +for attempt in {1..500}; do + test -f "$THIRDSHIFT_SESSION_TEST_DIR/log-closed" && break + sleep 0.01 +done +echo another-line +exec sleep 5"#, + ); + let log = fixture.path("logs/7-fixture-implement.jsonl"); + assert!(Command::new("mkfifo").arg(&log).status().unwrap().success()); + let child = fixture + .command("session::execution_tests::partial_stdout_is_reported_after_log_failure_with_a_blocked_prompt") + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + let mut reader = BufReader::new(File::open(log).unwrap()); + let mut line = String::new(); + loop { + line.clear(); + if reader.read_line(&mut line).unwrap() == 0 || line == "ready-to-fail\n" { + break; + } + } + drop(reader); + fs::write(fixture.path("log-closed"), "").unwrap(); + fixture.assert_reporting(child.wait_with_output().unwrap()); +} + +fn assert_no_completion_reporting(output: Output) { + assert!(output.status.success(), "{output:?}"); + let stderr = String::from_utf8_lossy(&output.stderr); + for unexpected in [WARNING, USAGE, "session ended after", "resuming"] { + assert!(!stderr.contains(unexpected), "{stderr}"); + } +} + +#[test] +fn interruption_suppresses_recovered_interpretation_and_retained_reads() { + if let Some(root) = std::env::var_os(FIXTURE) { + let root = Path::new(&root); + let error = execute_session(root); + assert_eq!(format!("{error:#}"), "interrupted"); + assert_owned_child_stopped(root); + return; + } + let fixture = Fixture::new( + r#"exec 1>&- +for attempt in {1..500}; do + grep -q 'sessionID' "$THIRDSHIFT_SESSION_TEST_DIR"/logs/*.jsonl && break + sleep 0.01 +done +kill -INT "$PPID" +exec sleep 5"#, + ); + assert_no_completion_reporting( + fixture.command("session::execution_tests::interruption_suppresses_recovered_interpretation_and_retained_reads").output().unwrap(), + ); + assert_eq!( + fs::read_to_string(fixture.path("calls")).unwrap(), + "session\n" + ); +} + +#[test] +fn startup_failure_preserves_its_cause_without_completion_reporting() { + if let Some(root) = std::env::var_os(FIXTURE) { + let error = execute_session(Path::new(&root)); + assert_eq!(error.to_string(), "could not run opencode"); + assert_eq!( + error.downcast_ref::().unwrap().kind(), + std::io::ErrorKind::PermissionDenied + ); + return; + } + let fixture = Fixture::new("exit 0"); + // Leave the selected CLI on PATH, but make spawning it fail. This never + // falls through to a real installed Harness. + fs::set_permissions( + fixture.path("bin/opencode"), + fs::Permissions::from_mode(0o644), + ) + .unwrap(); + assert_no_completion_reporting( + fixture.command("session::execution_tests::startup_failure_preserves_its_cause_without_completion_reporting").output().unwrap(), + ); + assert!(!fixture.path("calls").exists()); +}