fix(cli): let detached workers exit after post-run shutdown

Move worker control stdin handling off Tokio's blocking shutdown path so
subprocess workers can exit cleanly after success or cooperative
cancellation even when the parent still holds stdin open.

Add regression coverage for retro-enabled success and SIGTERM-driven
cancellation with stdin intentionally left open.

lib/crates/fabro-cli/src/commands/run/runner.rs

@@ -1,3 +1,4 @@
+use std::io::{BufRead as StdBufRead, BufReader as StdBufReader};
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, Ordering};
@@ -16,10 +17,9 @@ use fabro_workflow::event::{Emitter, RunEventSink};
 use fabro_workflow::operations::{self, StartServices};
 use fabro_workflow::run_control::RunControlState;
 use fabro_workflow::runtime_store::{RunStoreBackend, RunStoreHandle};
-use tokio::io::{self, AsyncBufReadExt, AsyncRead, BufReader};
 #[cfg(unix)]
 use tokio::signal::unix::{SignalKind, signal};
-use tokio::sync::Mutex;
+use tokio::sync::{Mutex, mpsc};
 use tokio::time::sleep;
 
 use crate::args::RunWorkerMode;
@@ -72,11 +72,7 @@ pub(crate) async fn execute(
     )));
     let interviewer = Arc::new(ControlInterviewer::new());
     let cancel_token = Arc::new(AtomicBool::new(false));
-    tokio::spawn(read_worker_control_stream(
-        io::stdin(),
-        Arc::clone(&interviewer),
-        Arc::clone(&cancel_token),
-    ));
+    spawn_worker_control_stream(Arc::clone(&interviewer), Arc::clone(&cancel_token))?;
     let run_control = RunControlState::new();
     install_signal_handlers(Arc::clone(&run_control), Arc::clone(&cancel_token))?;
     let github_app = maybe_build_github_app_credentials(&run_record.settings)?;
@@ -112,24 +108,79 @@ pub(crate) async fn execute(
     Ok(())
 }
 
-async fn read_worker_control_stream<R>(
-    reader: R,
+#[derive(Debug, PartialEq, Eq)]
+enum WorkerControlStreamEvent {
+    Line(String),
+    Eof,
+}
+
+fn spawn_worker_control_stream(
     interviewer: Arc<ControlInterviewer>,
     cancel_token: Arc<AtomicBool>,
+) -> Result<()> {
+    let (event_tx, event_rx) = mpsc::unbounded_channel();
+    tokio::spawn(handle_worker_control_stream_events(
+        interviewer,
+        cancel_token,
+        event_rx,
+    ));
+    std::thread::Builder::new()
+        .name("fabro-worker-control".to_string())
+        .spawn(move || {
+            read_worker_control_stream_blocking(StdBufReader::new(std::io::stdin()), event_tx);
+        })
+        .context("failed to spawn worker control reader thread")?;
+    Ok(())
+}
+
+fn read_worker_control_stream_blocking<R>(
+    mut reader: R,
+    event_tx: mpsc::UnboundedSender<WorkerControlStreamEvent>,
 ) where
-    R: AsyncRead + Unpin,
+    R: StdBufRead,
 {
-    let mut lines = BufReader::new(reader).lines();
+    let mut line = String::new();
     loop {
-        if let Ok(Some(line)) = lines.next_line().await {
-            apply_worker_control_line(&interviewer, &cancel_token, &line).await;
-        } else {
-            interviewer.interrupt_all().await;
-            break;
+        line.clear();
+        match reader.read_line(&mut line) {
+            Ok(0) => {
+                let _ = event_tx.send(WorkerControlStreamEvent::Eof);
+                break;
+            }
+            Ok(_) => {
+                let line = line.trim_end_matches(['\r', '\n']).to_string();
+                if event_tx.send(WorkerControlStreamEvent::Line(line)).is_err() {
+                    break;
+                }
+            }
+            Err(_) => {
+                let _ = event_tx.send(WorkerControlStreamEvent::Eof);
+                break;
+            }
         }
     }
 }
 
+async fn handle_worker_control_stream_events(
+    interviewer: Arc<ControlInterviewer>,
+    cancel_token: Arc<AtomicBool>,
+    mut event_rx: mpsc::UnboundedReceiver<WorkerControlStreamEvent>,
+) {
+    while let Some(event) = event_rx.recv().await {
+        match event {
+            WorkerControlStreamEvent::Line(line) => {
+                apply_worker_control_line(&interviewer, &cancel_token, &line).await;
+            }
+            WorkerControlStreamEvent::Eof => {
+                interviewer.interrupt_all().await;
+                return;
+            }
+        }
+    }
+
+    interviewer.interrupt_all().await;
+}
+
 async fn apply_worker_control_line(
     interviewer: &ControlInterviewer,
     cancel_token: &AtomicBool,
@@ -494,9 +545,9 @@ mod tests {
     use std::sync::atomic::{AtomicBool, Ordering};
 
     use super::{
-        MissingArtifactUploadTokenUploader, WorkerTitlePhase, apply_worker_control_line,
-        initial_worker_title_phase, read_worker_control_stream, worker_title,
-        worker_title_phase_for_event,
+        MissingArtifactUploadTokenUploader, WorkerControlStreamEvent, WorkerTitlePhase,
+        apply_worker_control_line, handle_worker_control_stream_events, initial_worker_title_phase,
+        read_worker_control_stream_blocking, worker_title, worker_title_phase_for_event,
     };
     use crate::args::RunWorkerMode;
     use fabro_interview::{AnswerValue, ControlInterviewer, Interviewer, Question, QuestionType};
@@ -664,18 +715,49 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn worker_control_stream_eof_interrupts_pending_interviews() {
+    async fn blocking_worker_control_stream_emits_lines_and_eof() {
+        let (event_tx, mut event_rx) = tokio::sync::mpsc::unbounded_channel();
+
+        read_worker_control_stream_blocking(
+            std::io::Cursor::new(
+                b"{\"v\":1,\"type\":\"run.cancel\"}\n{\"v\":1,\"type\":\"interview.answer\",\"qid\":\"q-1\",\"answer\":{\"kind\":\"yes\"}}\n",
+            ),
+            event_tx,
+        );
+
+        assert_eq!(
+            event_rx.try_recv(),
+            Ok(WorkerControlStreamEvent::Line(
+                r#"{"v":1,"type":"run.cancel"}"#.to_string()
+            ))
+        );
+        assert_eq!(
+            event_rx.try_recv(),
+            Ok(WorkerControlStreamEvent::Line(
+                r#"{"v":1,"type":"interview.answer","qid":"q-1","answer":{"kind":"yes"}}"#
+                    .to_string()
+            ))
+        );
+        assert_eq!(event_rx.try_recv(), Ok(WorkerControlStreamEvent::Eof));
+    }
+
+    #[tokio::test]
+    async fn worker_control_event_loop_eof_interrupts_pending_interviews() {
         let interviewer = Arc::new(ControlInterviewer::new());
         let cancel_token = Arc::new(AtomicBool::new(false));
         let mut question = Question::new("Approve?", QuestionType::YesNo);
         question.id = "q-1".to_string();
         let ask_interviewer = Arc::clone(&interviewer);
         let answer_task = tokio::spawn(async move { ask_interviewer.ask(question).await });
+        let (event_tx, event_rx) = tokio::sync::mpsc::unbounded_channel();
 
-        read_worker_control_stream(
-            tokio::io::empty(),
+        event_tx.send(WorkerControlStreamEvent::Eof).unwrap();
+        drop(event_tx);
+
+        handle_worker_control_stream_events(
             Arc::clone(&interviewer),
             Arc::clone(&cancel_token),
+            event_rx,
         )
         .await;
 

lib/crates/fabro-cli/tests/it/cmd/runner.rs

@@ -1,9 +1,16 @@
+use std::io::Read;
+use std::process::{Child, ExitStatus, Output, Stdio};
+use std::time::{Duration, Instant};
+
 use fabro_store::EventEnvelope;
 use fabro_test::{fabro_snapshot, test_context};
-use fabro_types::{EventBody, RunEvent};
+use fabro_types::{EventBody, RunEvent, StatusReason};
 use httpmock::MockServer;
 
-use super::support::{output_stderr, run_events, run_state, server_target};
+use super::support::{
+    output_stderr, run_events, run_state, server_target, wait_for_event_names, wait_for_status,
+    write_gated_workflow,
+};
 use crate::support::{fabro_json_snapshot, unique_run_id};
 
 const SHARED_DAEMON_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
@@ -28,6 +35,69 @@ fn assert_worker_succeeded(run_dir: &std::path::Path, stdout: &[u8]) {
     )));
 }
 
+fn spawn_worker_process(
+    context: &fabro_test::TestContext,
+    server: &str,
+    run_dir: &std::path::Path,
+    run_id: &str,
+    mode: &str,
+) -> Child {
+    let mut cmd = std::process::Command::new(env!("CARGO_BIN_EXE_fabro"));
+    cmd.current_dir(&context.temp_dir);
+    cmd.env("NO_COLOR", "1");
+    cmd.env("HOME", &context.home_dir);
+    cmd.env("FABRO_NO_UPGRADE_CHECK", "true");
+    cmd.env("FABRO_SERVER_MAX_CONCURRENT_RUNS", "64");
+    cmd.env("FABRO_TEST_IN_MEMORY_STORE", "1");
+    cmd.args([
+        "__run-worker",
+        "--server",
+        server,
+        "--run-dir",
+        run_dir.to_str().unwrap(),
+        "--run-id",
+        run_id,
+        "--mode",
+        mode,
+    ]);
+    cmd.stdin(Stdio::piped());
+    cmd.stdout(Stdio::piped());
+    cmd.stderr(Stdio::piped());
+    cmd.spawn().expect("worker should spawn")
+}
+
+fn wait_for_child_exit(child: &mut Child, timeout: Duration) -> ExitStatus {
+    let deadline = Instant::now() + timeout;
+    loop {
+        if let Some(status) = child.try_wait().expect("worker wait should succeed") {
+            return status;
+        }
+        assert!(
+            Instant::now() < deadline,
+            "timed out waiting for worker to exit"
+        );
+        std::thread::sleep(Duration::from_millis(50));
+    }
+}
+
+fn child_output(mut child: Child, status: ExitStatus) -> Output {
+    let mut stdout = Vec::new();
+    let mut stderr = Vec::new();
+    if let Some(mut pipe) = child.stdout.take() {
+        pipe.read_to_end(&mut stdout)
+            .expect("worker stdout should be readable");
+    }
+    if let Some(mut pipe) = child.stderr.take() {
+        pipe.read_to_end(&mut stderr)
+            .expect("worker stderr should be readable");
+    }
+    Output {
+        status,
+        stdout,
+        stderr,
+    }
+}
+
 fn server_endpoint(storage_dir: &std::path::Path) -> (reqwest::Client, String) {
     let target = server_target(storage_dir);
     if target.starts_with('/') {
@@ -557,3 +627,126 @@ fn detached_run_answers_pending_question_without_interview_scratch_files() {
             if props.question_id == question_id && props.answer == "A"
     )));
 }
+
+#[test]
+fn worker_exits_with_retro_enabled_even_when_stdin_stays_open() {
+    let context = test_context!();
+    let run_id = unique_run_id();
+    let workflow_path = context.temp_dir.join("retro-success.fabro");
+
+    context.write_temp(
+        "fabro.toml",
+        r#"_version = 1
+
+[run.execution]
+retros = true
+"#,
+    );
+    context.write_temp(
+        "retro-success.fabro",
+        r#"digraph RetroSuccess {
+  graph [goal="Finish successfully with retro enabled"]
+  start [shape=Mdiamond, label="Start"]
+  exit  [shape=Msquare, label="Exit"]
+  work  [shape=parallelogram, label="Work", script="true"]
+  start -> work -> exit
+}
+"#,
+    );
+
+    context
+        .command()
+        .args([
+            "create",
+            "--dry-run",
+            "--auto-approve",
+            "--run-id",
+            run_id.as_str(),
+            workflow_path.to_str().unwrap(),
+        ])
+        .assert()
+        .success();
+
+    let run_dir = context.find_run_dir(&run_id);
+    let server = server_target(&context.storage_dir);
+    let mut child = spawn_worker_process(&context, &server, &run_dir, &run_id, "start");
+    let stdin = child.stdin.take().expect("worker stdin should be piped");
+
+    wait_for_event_names(&run_dir, &["run.completed", "retro.completed"]);
+    let status = wait_for_child_exit(&mut child, SHARED_DAEMON_TIMEOUT);
+    drop(stdin);
+    let output = child_output(child, status);
+
+    assert!(
+        output.status.success(),
+        "worker should exit successfully after retro even with stdin open:\nstdout:\n{}\nstderr:\n{}",
+        String::from_utf8_lossy(&output.stdout),
+        String::from_utf8_lossy(&output.stderr)
+    );
+
+    let events = stored_worker_events(&run_dir);
+    let run_completed_index = events
+        .iter()
+        .position(|event| matches!(&event.body, EventBody::RunCompleted(_)))
+        .expect("run.completed should be present");
+    let retro_completed_index = events
+        .iter()
+        .position(|event| matches!(&event.body, EventBody::RetroCompleted(_)))
+        .expect("retro.completed should be present");
+    assert!(
+        run_completed_index < retro_completed_index,
+        "retro should still run after run.completed"
+    );
+}
+
+#[cfg(unix)]
+#[test]
+fn worker_exits_after_sigterm_cancel_even_when_stdin_stays_open() {
+    let context = test_context!();
+    let run_id = unique_run_id();
+    let workflow_path = context.temp_dir.join("cancel-gated.fabro");
+    let _gate = write_gated_workflow(&workflow_path, "cancel_gated", "Wait for cancellation");
+
+    context
+        .command()
+        .args([
+            "create",
+            "--auto-approve",
+            "--no-retro",
+            "--sandbox",
+            "local",
+            "--run-id",
+            run_id.as_str(),
+            workflow_path.to_str().unwrap(),
+        ])
+        .assert()
+        .success();
+
+    let run_dir = context.find_run_dir(&run_id);
+    let server = server_target(&context.storage_dir);
+    let mut child = spawn_worker_process(&context, &server, &run_dir, &run_id, "start");
+    let stdin = child.stdin.take().expect("worker stdin should be piped");
+
+    wait_for_event_names(&run_dir, &["run.running"]);
+    let worker_pid = child.id();
+    assert!(worker_pid > 0, "worker pid should be present");
+    fabro_proc::sigterm(worker_pid);
+
+    wait_for_status(&run_dir, &["failed"]);
+    let status = wait_for_child_exit(&mut child, SHARED_DAEMON_TIMEOUT);
+    drop(stdin);
+    let output = child_output(child, status);
+
+    assert!(
+        output.status.success(),
+        "worker should exit cleanly after SIGTERM cancellation:\nstdout:\n{}\nstderr:\n{}",
+        String::from_utf8_lossy(&output.stdout),
+        String::from_utf8_lossy(&output.stderr)
+    );
+
+    let status_record = run_state(&run_dir)
+        .status
+        .expect("cancelled run should have a status record");
+    assert_eq!(status_record.status.to_string(), "failed");
+    assert_eq!(status_record.reason, Some(StatusReason::Cancelled));
+}