diff --git a/src/agent_launch.rs b/src/agent_launch.rs index 9061ef24..96bacd00 100644 --- a/src/agent_launch.rs +++ b/src/agent_launch.rs @@ -6,6 +6,8 @@ use agent_registry::{Agent, AgentSpec, Tier, headless_args}; use std::io::Read; use std::path::Path; use std::process::{Command, Stdio}; +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; use std::time::{Duration, Instant}; #[derive(Debug)] @@ -122,8 +124,12 @@ pub struct Captured { pub stdout: String, /// Everything the child wrote to stderr. pub stderr: String, - /// True iff the child was killed for outliving the timeout. + /// True iff the child was killed for outliving `hard_cap` or going idle + /// for `idle_timeout` (see `idle_killed` to tell the two apart). pub timed_out: bool, + /// True iff `timed_out` was caused by the idle window elapsing with no + /// new stdout/stderr bytes, rather than `hard_cap` being reached. + pub idle_killed: bool, } /// Kill `child` and everything it spawned, not just the direct process. A @@ -163,11 +169,21 @@ pub(crate) fn kill_tree(child: &mut std::process::Child) { } /// Run `cmd` to completion, capturing stdout, and kill the child (and its -/// whole process tree) if it outlives `timeout` (reporting `timed_out`). -/// Stdout is drained on a separate thread so a child that fills the OS pipe -/// buffer can't deadlock the wait loop. +/// whole process tree) if either `hard_cap` elapses regardless of activity +/// (a backstop against runaway output, not the primary signal — see +/// `idle_timeout`), or `idle_timeout` elapses with no new stdout/stderr +/// bytes (the primary liveness signal: a task producing steady output can +/// run all the way to `hard_cap` even if that takes hours, while a task +/// that's genuinely stuck is caught quickly). Both streams are drained on +/// separate threads so a child that fills the OS pipe buffer can't deadlock +/// the wait loop; each thread bumps a shared byte counter per chunk read so +/// the wait loop can observe activity without waiting for EOF. #[allow(dead_code)] -pub fn run_captured(mut cmd: Command, timeout: Duration) -> std::io::Result { +pub fn run_captured( + mut cmd: Command, + hard_cap: Duration, + idle_timeout: Duration, +) -> std::io::Result { cmd.stdout(Stdio::piped()); cmd.stderr(Stdio::piped()); cmd.stdin(Stdio::null()); @@ -181,29 +197,70 @@ pub fn run_captured(mut cmd: Command, timeout: Duration) -> std::io::Result = Vec::new(); + let mut chunk = [0u8; 8192]; + // Read in chunks (rather than `read_to_string` straight to EOF) so + // `activity` reflects bytes as they arrive, not only once the child + // exits — the wait loop below needs that to detect a stalled child + // before it closes its pipes. + loop { + match pipe.read(&mut chunk) { + Ok(0) | Err(_) => break, + Ok(n) => { + buf.extend_from_slice(&chunk[..n]); + stdout_activity.fetch_add(n as u64, Ordering::Relaxed); + } + } + } + String::from_utf8_lossy(&buf).into_owned() }); let mut err_pipe = child.stderr.take().expect("stderr piped above"); + let stderr_activity = activity.clone(); let err_reader = std::thread::spawn(move || { - let mut buf = String::new(); - let _ = err_pipe.read_to_string(&mut buf); - buf + let mut buf: Vec = Vec::new(); + let mut chunk = [0u8; 8192]; + loop { + match err_pipe.read(&mut chunk) { + Ok(0) | Err(_) => break, + Ok(n) => { + buf.extend_from_slice(&chunk[..n]); + stderr_activity.fetch_add(n as u64, Ordering::Relaxed); + } + } + } + String::from_utf8_lossy(&buf).into_owned() }); let start = Instant::now(); + let mut last_activity_bytes = 0u64; + let mut last_activity_at = Instant::now(); let mut timed_out = false; + let mut idle_killed = false; let status = loop { if let Some(status) = child.try_wait()? { break status; } - if start.elapsed() >= timeout { + let bytes_so_far = activity.load(Ordering::Relaxed); + if bytes_so_far != last_activity_bytes { + last_activity_bytes = bytes_so_far; + last_activity_at = Instant::now(); + } + if start.elapsed() >= hard_cap { + kill_tree(&mut child); + let status = child.wait()?; + timed_out = true; + break status; + } + if last_activity_at.elapsed() >= idle_timeout { kill_tree(&mut child); let status = child.wait()?; timed_out = true; + idle_killed = true; break status; } std::thread::sleep(Duration::from_millis(20)); @@ -216,6 +273,7 @@ pub fn run_captured(mut cmd: Command, timeout: Duration) -> std::io::Result HeadlessOutcome { let Some(spec) = registry.iter().find(|s| s.id.as_str() == agent) else { @@ -291,13 +352,20 @@ pub fn run_headless( // See the matching strip in `run_launch_env` above (item #139) — same // rationale applies to headless child processes. cmd.env_remove("CARGO_TARGET_DIR"); - match run_captured(cmd, timeout) { + match run_captured(cmd, hard_cap, idle_timeout) { Ok(c) if c.success => HeadlessOutcome::Ok(c.stdout), - Ok(c) if c.timed_out => HeadlessOutcome::Failed(format!( - "{} timed out after {timeout:?}{}", - spec.display_name, - diagnostic_suffix(&c) - )), + Ok(c) if c.timed_out => { + let reason = if c.idle_killed { + format!("went idle for {idle_timeout:?} (no new output)") + } else { + format!("exceeded hard cap of {hard_cap:?}") + }; + HeadlessOutcome::Failed(format!( + "{} timed out — {reason}{}", + spec.display_name, + diagnostic_suffix(&c) + )) + } Ok(_) => HeadlessOutcome::Failed(format!("{} exited non-zero", spec.display_name)), Err(e) => HeadlessOutcome::Failed(format!("failed to run {}: {e}", spec.display_name)), } @@ -399,7 +467,12 @@ mod tests { fn run_captured_captures_stdout() { let mut cmd = Command::new("sh"); cmd.arg("-c").arg("printf 'hello world'"); - let out = run_captured(cmd, std::time::Duration::from_secs(5)).unwrap(); + let out = run_captured( + cmd, + std::time::Duration::from_secs(5), + std::time::Duration::from_secs(5), + ) + .unwrap(); assert!(out.success); assert!(!out.timed_out); assert_eq!(out.stdout, "hello world"); @@ -410,7 +483,12 @@ mod tests { fn run_captured_keeps_output_written_before_a_timeout_kill() { let mut cmd = Command::new("sh"); cmd.arg("-c").arg("printf 'made it here'; sleep 5"); - let out = run_captured(cmd, std::time::Duration::from_millis(150)).unwrap(); + let out = run_captured( + cmd, + std::time::Duration::from_millis(150), + std::time::Duration::from_millis(150), + ) + .unwrap(); assert!(out.timed_out); assert_eq!( out.stdout, "made it here", @@ -424,7 +502,12 @@ mod tests { let start = std::time::Instant::now(); let mut cmd = Command::new("sh"); cmd.arg("-c").arg("sleep 5"); - let out = run_captured(cmd, std::time::Duration::from_millis(150)).unwrap(); + let out = run_captured( + cmd, + std::time::Duration::from_millis(150), + std::time::Duration::from_millis(150), + ) + .unwrap(); assert!(out.timed_out, "should report timeout"); assert!(!out.success, "a killed child is not a success"); // Must return promptly, not wait out the full 5s sleep. @@ -434,6 +517,57 @@ mod tests { ); } + // A process that keeps producing output must NOT be killed just because + // it has run past what used to be the single fixed timeout — only a + // genuinely stalled child (no new bytes for `idle_timeout`) should be. + // This is the behavior item #20 exists to add. + #[cfg(unix)] + #[test] + fn run_captured_does_not_idle_timeout_while_output_keeps_arriving() { + let mut cmd = Command::new("sh"); + cmd.arg("-c") + .arg("for i in 1 2 3 4 5; do printf 'x'; sleep 0.05; done"); + let out = run_captured( + cmd, + std::time::Duration::from_secs(30), + std::time::Duration::from_millis(300), + ) + .unwrap(); + assert!( + !out.timed_out, + "steady output growth must reset the idle clock, not get killed" + ); + assert_eq!(out.stdout, "xxxxx"); + } + + // Mirrors `run_captured_keeps_output_written_before_a_timeout_kill` but + // with a generous hard cap and a tight idle window, proving the idle + // timeout — not the hard cap — is what catches a child that produced + // real output and then went quiet. + #[cfg(unix)] + #[test] + fn run_captured_idle_timeout_kills_a_stalled_child_well_before_the_hard_cap() { + let start = std::time::Instant::now(); + let mut cmd = Command::new("sh"); + cmd.arg("-c").arg("printf 'did some work'; sleep 5"); + let out = run_captured( + cmd, + std::time::Duration::from_secs(30), + std::time::Duration::from_millis(150), + ) + .unwrap(); + assert!(out.timed_out, "should time out"); + assert!( + out.idle_killed, + "should be killed for going idle, not for exceeding the hard cap" + ); + assert_eq!(out.stdout, "did some work"); + assert!( + start.elapsed() < std::time::Duration::from_secs(2), + "idle timeout should fire well before the 30s hard cap" + ); + } + // Proves the fix for the "descendant outlives the direct child" hang: the // direct child backgrounds a grandchild that inherits the piped stdout fd, // then waits on it. If timeout only killed the direct child (the old @@ -446,7 +580,12 @@ mod tests { let start = std::time::Instant::now(); let mut cmd = Command::new("sh"); cmd.arg("-c").arg("sleep 5 & wait"); - let out = run_captured(cmd, std::time::Duration::from_millis(150)).unwrap(); + let out = run_captured( + cmd, + std::time::Duration::from_millis(150), + std::time::Duration::from_millis(150), + ) + .unwrap(); assert!(out.timed_out, "should report timeout"); assert!(!out.success, "a killed child is not a success"); assert!( @@ -609,7 +748,14 @@ mod tests { #[test] fn run_headless_unknown_agent() { let reg = headless_registry(); - match run_headless(®, "nope", "hi", Duration::from_secs(1), &[]) { + match run_headless( + ®, + "nope", + "hi", + Duration::from_secs(1), + Duration::from_secs(1), + &[], + ) { HeadlessOutcome::UnknownAgent(m) => assert!(m.contains("nope")), other => panic!("expected UnknownAgent, got {other:?}"), } @@ -618,7 +764,14 @@ mod tests { #[test] fn run_headless_agent_without_print_mode() { let reg = headless_registry(); - match run_headless(®, "aider", "hi", Duration::from_secs(1), &[]) { + match run_headless( + ®, + "aider", + "hi", + Duration::from_secs(1), + Duration::from_secs(1), + &[], + ) { HeadlessOutcome::NotHeadless(m) => assert!(m.contains("aider")), other => panic!("expected NotHeadless, got {other:?}"), } @@ -627,7 +780,14 @@ mod tests { #[test] fn run_headless_binary_not_found() { let reg = headless_registry(); - match run_headless(®, "codex", "hi", Duration::from_secs(1), &[]) { + match run_headless( + ®, + "codex", + "hi", + Duration::from_secs(1), + Duration::from_secs(1), + &[], + ) { HeadlessOutcome::NotFound(m) => assert!(m.contains("not found")), other => panic!("expected NotFound, got {other:?}"), } @@ -661,6 +821,7 @@ mod tests { stdout: String::new(), stderr: String::new(), timed_out: true, + idle_killed: false, }; assert_eq!(diagnostic_suffix(&c), " (no output captured)"); } @@ -672,6 +833,7 @@ mod tests { stdout: "working on task 3...".to_string(), stderr: "some warning".to_string(), timed_out: true, + idle_killed: false, }; let suffix = diagnostic_suffix(&c); assert!(suffix.contains("last stdout before kill")); @@ -685,6 +847,7 @@ mod tests { stdout: String::new(), stderr: "panic: something broke".to_string(), timed_out: true, + idle_killed: false, }; let suffix = diagnostic_suffix(&c); assert!(suffix.contains("last stderr before kill")); diff --git a/src/agents.rs b/src/agents.rs index e49721d0..a1f1b7cc 100644 --- a/src/agents.rs +++ b/src/agents.rs @@ -227,7 +227,17 @@ pub fn cli_run( /// print its captured reply to stdout, and return a process exit code (0 on /// success, 1 on any failure). The caller decides whether to `exit`. pub fn cli_run_headless(agent: &str, prompt: &str, timeout: std::time::Duration) -> i32 { - match agent_launch::run_headless(agent_registry::REGISTRY, agent, prompt, timeout, &[]) { + // `agentflare run --print` is a single one-off invocation, not dispatched + // background work — pass `timeout` as both the hard cap and the idle + // window so behavior matches the old flat wall-clock deadline exactly. + match agent_launch::run_headless( + agent_registry::REGISTRY, + agent, + prompt, + timeout, + timeout, + &[], + ) { agent_launch::HeadlessOutcome::Ok(reply) => { print!("{reply}"); 0 diff --git a/src/cli/work.rs b/src/cli/work.rs index 08f68128..8d093b9b 100644 --- a/src/cli/work.rs +++ b/src/cli/work.rs @@ -16,9 +16,19 @@ pub struct WorkArgs { /// and any `~/.agentflare/config.toml` `[router]` rules. #[arg(long)] pub agent: Option, - /// Headless run timeout in seconds (default 1800 = 30 min). - #[arg(long, default_value_t = 1800)] + /// Absolute hard-cap timeout in seconds, regardless of activity + /// (default 21600 = 6h). A backstop against a runaway process, not the + /// primary signal for whether to keep a job alive — see --idle-timeout. + #[arg(long, default_value_t = 21600)] pub timeout: u64, + /// Kill the agent if it produces no new stdout/stderr output for this + /// many seconds (default 300 = 5 min). This is the primary liveness + /// signal: a task that keeps producing output can run all the way to + /// --timeout even if that takes hours; a genuinely stuck task is caught + /// quickly instead of running out the full --timeout with nothing + /// happening. + #[arg(long, default_value_t = 300)] + pub idle_timeout: u64, /// Max agent turns before forced stop (Claude Code only). #[arg(long)] pub max_turns: Option, @@ -271,6 +281,7 @@ impl WorkArgs { fn run_work(args: WorkArgs) -> i32 { let mcp = AgentflareMcp::default(); let timeout = Duration::from_secs(args.timeout); + let idle_timeout = Duration::from_secs(args.idle_timeout); // An explicit --agent fails fast, before claiming anything, through the // same resolver `resolve_agent` uses further down post-claim — so this @@ -424,6 +435,7 @@ fn run_work(args: WorkArgs) -> i32 { agent_enum.as_str(), &prompt, timeout, + idle_timeout, &extra_args, ); diff --git a/src/supervisor.rs b/src/supervisor.rs index bb646155..553145d1 100644 --- a/src/supervisor.rs +++ b/src/supervisor.rs @@ -10,11 +10,19 @@ const DISPATCHED_LABEL: &str = "dispatched"; const NEEDS_MANUAL_LABEL: &str = "needs-manual-dispatch"; const NEEDS_HUMAN_GATE_LABEL: &str = "needs-human-gate"; -/// `agentflare work`'s own `--timeout` defaults to 1800s (its real budget -/// for a headless agent run) -- this is that budget plus margin for the -/// claim/worktree setup and done/push/PR steps around it, so the outer job -/// timeout never cuts off a run before work's own inner timeout would. -const WORK_JOB_TIMEOUT_SECS: u64 = 2100; +/// `agentflare work`'s own `--timeout` is its hard-cap safety net, not the +/// primary judge of whether it's still making progress -- that's +/// `--idle-timeout`, which lets a job run for hours as long as it keeps +/// producing output (see item #20). It defaults to 21600s (6h); this is +/// that budget plus margin for the claim/worktree setup and done/push/PR +/// steps around it, so the outer job timeout never cuts off a run before +/// work's own inner timeout would. Before item #20 this outer timeout was +/// 2100s (aligned to work's old 1800s fixed timeout) -- left unaligned +/// after work's default grew, it would have silently reintroduced the same +/// "killed a legitimately-progressing job" bug for every job actually +/// dispatched by the daemon, since this is the timeout that governs them, +/// not work's own. +const WORK_JOB_TIMEOUT_SECS: u64 = 21_900; /// Returns the matching `Agent` only if `agent_registry::autonomous_args` /// confirms it has a headless permission-bypass flag — the same gate