Retain Harness interpretation after transport failures

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<T> {
+    pub execution: Result<ExitStatus>,
+    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<T: Send + 'static>(
     adapter: &dyn Adapter,
     command: &mut Command,
     input: Option<&str>,
-    consume: impl FnOnce(ChildStdout) -> Result<T> + 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<Streamed<T>> {
+    // 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<Vec<u8>> {

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,18 +163,53 @@ 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));
     child.assert_stopped();
     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::<std::io::Error>().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 >/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 >/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(|_| ())
         };

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<Ended> {
     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}"));