diff --git a/Cargo.lock b/Cargo.lock index ae9d2c7..705fe39 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4,13 +4,15 @@ version = 4 [[package]] name = "agentctl" -version = "0.1.0" +version = "0.1.1" dependencies = [ "anyhow", "assert_cmd", "clap", + "libc", "serde", "serde_json", + "signal-hook", "tempfile", "toml", ] @@ -433,6 +435,26 @@ dependencies = [ "serde", ] +[[package]] +name = "signal-hook" +version = "0.3.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d881a16cf4426aa584979d30bd82cb33429027e42122b169753d6ef1085ed6e2" +dependencies = [ + "libc", + "signal-hook-registry", +] + +[[package]] +name = "signal-hook-registry" +version = "1.4.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c4db69cba1110affc0e9f7bcd48bbf87b3f4fc7c61fc9155afd4c469eb3d6c1b" +dependencies = [ + "errno", + "libc", +] + [[package]] name = "strsim" version = "0.11.1" diff --git a/Cargo.toml b/Cargo.toml index e4701d4..bd424f0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "agentctl" -version = "0.1.0" +version = "0.1.1" edition = "2024" license = "MIT" description = "Compact, bounded command and agent reports for LLM tool loops" @@ -17,6 +17,10 @@ serde = { version = "1.0.219", features = ["derive"] } serde_json = "1.0.140" toml = "0.8.23" +[target.'cfg(unix)'.dependencies] +libc = "0.2.186" +signal-hook = "0.3.18" + [dev-dependencies] assert_cmd = "2.0.17" tempfile = "3.20.0" diff --git a/README.md b/README.md index 29ae720..36b7cee 100644 --- a/README.md +++ b/README.md @@ -52,6 +52,13 @@ handling still works. A failing `cargo test` returns a compact JSON summary: The full stdout and stderr remain in `raw_log_path` for later inspection. Use `--raw` when you deliberately want the untouched command output streamed back. +Output is streamed to the raw log while only the configured head/tail window is +retained in memory. Agent commands run in an isolated process group. Interrupting +agentctl forwards the signal to the complete command tree, waits for it, and +terminates descendants left behind by a command that exits early. On Linux, +agentctl also adopts and reaps orphaned descendants so repeated tool loops do not +accumulate zombie or live background agent processes. + ## Run a detached agent task Configure agents in TOML: @@ -132,6 +139,9 @@ agentctl list --json `logs` is intentionally the raw-output command. `status` and `list` stay bounded and machine-readable for agent loops. +Detached supervisors also run in an isolated process group and terminate the +remaining group after recording their exit status. + ## JSON contract Every public command that has `--json` uses stable field names. diff --git a/src/exec.rs b/src/exec.rs index bb275b0..a5b5426 100644 --- a/src/exec.rs +++ b/src/exec.rs @@ -1,6 +1,8 @@ +use std::collections::VecDeque; +use std::ffi::OsString; use std::fs; use std::io::{Read, Write}; -use std::path::PathBuf; +use std::path::{Path, PathBuf}; use std::process::{Command as ProcessCommand, ExitStatus, Stdio}; use std::thread; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; @@ -8,6 +10,7 @@ use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use anyhow::{Context, Result, bail}; use crate::config::CONFIG_DIR; +use crate::process::ManagedChild; #[derive(Debug, Clone)] pub struct CaptureOptions { pub max_output_bytes: usize, @@ -26,46 +29,95 @@ pub struct CapturedRun { pub raw_log_path: PathBuf, } +#[derive(Debug, Clone, Default)] +pub(crate) struct ProcessOptions { + pub cwd: Option, + pub env: Vec<(OsString, OsString)>, + pub stdin: Option>, +} + pub fn run_captured(argv: &[String], options: &CaptureOptions) -> Result { + run_captured_with(argv, options, &ProcessOptions::default()) +} + +pub(crate) fn run_captured_with( + argv: &[String], + options: &CaptureOptions, + process_options: &ProcessOptions, +) -> Result { if argv.is_empty() { bail!("exec requires a command after `--`"); } let raw_log_path = create_raw_log_path(options.label.as_deref())?; + let stdout_path = raw_log_path.with_extension("stdout.tmp"); + let stderr_path = raw_log_path.with_extension("stderr.tmp"); let start = Instant::now(); let mut command = ProcessCommand::new(&argv[0]); if argv.len() > 1 { command.args(&argv[1..]); } command.stdout(Stdio::piped()).stderr(Stdio::piped()); + if let Some(cwd) = &process_options.cwd { + command.current_dir(cwd); + } + command.envs(process_options.env.iter().cloned()); + if process_options.stdin.is_some() { + command.stdin(Stdio::piped()); + } - let mut child = command - .spawn() + let mut child = ManagedChild::spawn(&mut command) .with_context(|| format!("executing `{}`", format_command(argv)))?; - let stdout = child.stdout.take().context("capturing child stdout")?; - let stderr = child.stderr.take().context("capturing child stderr")?; + let stdout = child.take_stdout().context("capturing child stdout")?; + let stderr = child.take_stderr().context("capturing child stderr")?; let stream_stdout = options.stream_raw; - let stdout_handle = - thread::spawn(move || read_stream(stdout, stream_stdout, StreamKind::Stdout)); + let max_output_bytes = options.max_output_bytes; + let tail_bytes = options.tail_bytes.min(max_output_bytes); + let stdout_capture_path = stdout_path.clone(); + let stdout_handle = thread::spawn(move || { + read_stream( + stdout, + stream_stdout, + StreamKind::Stdout, + max_output_bytes, + tail_bytes, + &stdout_capture_path, + ) + }); let stream_stderr = options.stream_raw; - let stderr_handle = - thread::spawn(move || read_stream(stderr, stream_stderr, StreamKind::Stderr)); + let stderr_capture_path = stderr_path.clone(); + let stderr_handle = thread::spawn(move || { + read_stream( + stderr, + stream_stderr, + StreamKind::Stderr, + max_output_bytes, + tail_bytes, + &stderr_capture_path, + ) + }); + + if let Some(input) = &process_options.stdin + && let Some(mut stdin) = child.take_stdin() + { + stdin + .write_all(input) + .context("writing managed child stdin")?; + } let status = child .wait() .with_context(|| format!("waiting on `{}`", format_command(argv)))?; let elapsed = start.elapsed(); - let stdout_bytes = stdout_handle + let stdout_raw = stdout_handle .join() .map_err(|_| anyhow::anyhow!("stdout capture thread panicked"))??; - let stderr_bytes = stderr_handle + let stderr_raw = stderr_handle .join() .map_err(|_| anyhow::anyhow!("stderr capture thread panicked"))??; - let stdout_raw = String::from_utf8_lossy(&stdout_bytes).to_string(); - let stderr_raw = String::from_utf8_lossy(&stderr_bytes).to_string(); - persist_raw_log(&raw_log_path, &stdout_raw, &stderr_raw) + persist_raw_log(&raw_log_path, &stdout_path, &stderr_path) .with_context(|| format!("writing raw log {}", raw_log_path.display()))?; Ok(CapturedRun { @@ -83,35 +135,104 @@ enum StreamKind { Stderr, } -fn read_stream(mut reader: R, stream_raw: bool, kind: StreamKind) -> Result> { - let mut captured = Vec::new(); +fn read_stream( + mut reader: R, + stream_raw: bool, + kind: StreamKind, + max_output_bytes: usize, + tail_bytes: usize, + capture_path: &Path, +) -> Result { + let mut captured = BoundedCapture::new(max_output_bytes, tail_bytes); + let mut capture_file = fs::File::create(capture_path) + .with_context(|| format!("creating stream capture {}", capture_path.display()))?; let mut buffer = [0_u8; 8192]; loop { let bytes_read = reader.read(&mut buffer).context("reading child output")?; if bytes_read == 0 { break; } - captured.extend_from_slice(&buffer[..bytes_read]); + let chunk = &buffer[..bytes_read]; + capture_file + .write_all(chunk) + .with_context(|| format!("writing stream capture {}", capture_path.display()))?; + captured.push(chunk); if stream_raw { match kind { StreamKind::Stdout => { let mut stdout = std::io::stdout().lock(); - stdout - .write_all(&buffer[..bytes_read]) - .context("streaming child stdout")?; + stdout.write_all(chunk).context("streaming child stdout")?; stdout.flush().context("flushing child stdout")?; } StreamKind::Stderr => { let mut stderr = std::io::stderr().lock(); - stderr - .write_all(&buffer[..bytes_read]) - .context("streaming child stderr")?; + stderr.write_all(chunk).context("streaming child stderr")?; stderr.flush().context("flushing child stderr")?; } } } } - Ok(captured) + capture_file + .flush() + .with_context(|| format!("flushing stream capture {}", capture_path.display()))?; + Ok(captured.finish()) +} + +struct BoundedCapture { + max_bytes: usize, + head_limit: usize, + tail_limit: usize, + total_bytes: usize, + head: Vec, + tail: VecDeque, +} + +impl BoundedCapture { + fn new(max_bytes: usize, tail_bytes: usize) -> Self { + let tail_limit = tail_bytes.min(max_bytes); + Self { + max_bytes, + head_limit: max_bytes.saturating_sub(tail_limit), + tail_limit, + total_bytes: 0, + head: Vec::with_capacity(max_bytes.saturating_sub(tail_limit)), + tail: VecDeque::with_capacity(tail_limit), + } + } + + fn push(&mut self, bytes: &[u8]) { + self.total_bytes = self.total_bytes.saturating_add(bytes.len()); + let head_remaining = self.head_limit.saturating_sub(self.head.len()); + let head_bytes = head_remaining.min(bytes.len()); + self.head.extend_from_slice(&bytes[..head_bytes]); + + if self.tail_limit == 0 { + return; + } + self.tail.extend(&bytes[head_bytes..]); + while self.tail.len() > self.tail_limit { + self.tail.pop_front(); + } + } + + fn finish(self) -> String { + if self.max_bytes == 0 { + return String::new(); + } + let mut bytes = self.head; + if self.total_bytes > self.max_bytes { + let retained = bytes.len() + self.tail.len(); + bytes.extend_from_slice( + format!( + "\n\n[... omitted {} bytes ...]\n\n", + self.total_bytes.saturating_sub(retained) + ) + .as_bytes(), + ); + } + bytes.extend(self.tail); + String::from_utf8_lossy(&bytes).into_owned() + } } pub fn format_command(argv: &[String]) -> String { @@ -169,21 +290,50 @@ fn sanitize_label(label: &str) -> String { .collect() } -fn persist_raw_log(path: &PathBuf, stdout: &str, stderr: &str) -> Result<()> { - let mut raw = String::new(); - if !stdout.is_empty() { - raw.push_str("stdout:\n"); - raw.push_str(stdout); - if !stdout.ends_with('\n') { - raw.push('\n'); - } +fn persist_raw_log(path: &Path, stdout_path: &Path, stderr_path: &Path) -> Result<()> { + let mut raw = fs::File::create(path).context("creating raw command output")?; + append_capture(&mut raw, "stdout:\n", stdout_path)?; + append_capture(&mut raw, "stderr:\n", stderr_path)?; + raw.flush().context("flushing raw command output")?; + let _ = fs::remove_file(stdout_path); + let _ = fs::remove_file(stderr_path); + Ok(()) +} + +fn append_capture(raw: &mut fs::File, heading: &str, capture_path: &Path) -> Result<()> { + let mut capture = fs::File::open(capture_path) + .with_context(|| format!("opening stream capture {}", capture_path.display()))?; + if capture.metadata()?.len() == 0 { + return Ok(()); } - if !stderr.is_empty() { - raw.push_str("stderr:\n"); - raw.push_str(stderr); - if !stderr.ends_with('\n') { - raw.push('\n'); - } + raw.write_all(heading.as_bytes())?; + std::io::copy(&mut capture, raw)?; + raw.write_all(b"\n")?; + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn bounded_capture_keeps_head_and_tail_without_retaining_the_middle() { + let mut capture = BoundedCapture::new(10, 4); + capture.push(b"012345"); + capture.push(b"6789abcdef"); + + let output = capture.finish(); + + assert!(output.starts_with("012345")); + assert!(output.contains("omitted 6 bytes")); + assert!(output.ends_with("cdef")); + } + + #[test] + fn bounded_capture_preserves_small_output_exactly() { + let mut capture = BoundedCapture::new(10, 4); + capture.push(b"hello"); + + assert_eq!(capture.finish(), "hello"); } - fs::write(path, raw).context("writing raw command output") } diff --git a/src/jobs.rs b/src/jobs.rs index 3a34977..938d034 100644 --- a/src/jobs.rs +++ b/src/jobs.rs @@ -1,15 +1,16 @@ use std::fs; use std::io::{Read, Write}; use std::path::{Path, PathBuf}; -use std::process::{Command as ProcessCommand, Stdio}; +use std::process::{Command as ProcessCommand, ExitStatus, Stdio}; use std::thread; -use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; use anyhow::{Context, Result, anyhow, bail}; use serde::{Deserialize, Serialize}; use crate::config::{AgentConfig, CONFIG_DIR, SummaryConfig}; -use crate::exec::{CapturedRun, format_command}; +use crate::exec::{CaptureOptions, CapturedRun, ProcessOptions, format_command, run_captured_with}; +use crate::process::spawn_detached; use crate::summary::{Summary, summarize, summarize_with_command}; #[derive(Debug, Clone)] @@ -71,6 +72,17 @@ struct StoredJobRecord { exit_code_path: PathBuf, } +#[derive(Debug, Clone, Deserialize, Serialize)] +struct SupervisorSpec { + argv: Vec, + prompt_stdin: bool, + prompt: String, + cwd: PathBuf, + iterations: u32, + exit_code_path: PathBuf, + label: String, +} + #[derive(Debug, Clone, Default)] struct ExitRecord { exit_code: Option, @@ -84,7 +96,7 @@ pub fn run_foreground( validate_options(options)?; let mut reports = Vec::new(); for iteration in 1..=options.iterations { - let run = run_agent_once(options, iteration)?; + let run = run_agent_once(options, summary_config, iteration)?; let summary = match &summary_config.summarizer_command { Some(command) => summarize_with_command(&run, summary_config, command) .unwrap_or_else(|_| summarize(&run, summary_config)), @@ -115,40 +127,43 @@ pub fn launch_detached(options: &AgentRunOptions) -> Result { let log_path = job_dir.join("output.log"); let exit_code_path = job_dir.join("exit-code"); let job_path = job_dir.join("job.toml"); + let supervisor_path = job_dir.join("supervisor.toml"); let stdout = fs::File::create(&log_path) .with_context(|| format!("creating agent log {}", log_path.display()))?; let stderr = stdout .try_clone() .with_context(|| format!("cloning agent log {}", log_path.display()))?; - let script = detached_script( - &command_argv(&options.agent, &options.prompt), - options.agent.prompt_stdin, - options.iterations, - &exit_code_path, - ); - let mut command = ProcessCommand::new("bash"); - command.arg("-lc").arg(script); + let argv = command_argv(&options.agent, &options.prompt); + let supervisor = SupervisorSpec { + argv: argv.clone(), + prompt_stdin: options.agent.prompt_stdin, + prompt: options.prompt.clone(), + cwd: options.cwd.clone(), + iterations: options.iterations, + exit_code_path: exit_code_path.clone(), + label: options.agent_name.clone(), + }; + fs::write(&supervisor_path, toml::to_string_pretty(&supervisor)?) + .with_context(|| format!("writing {}", supervisor_path.display()))?; + + let executable = std::env::current_exe().context("locating agentctl executable")?; + let mut command = ProcessCommand::new(executable); + command.arg("__supervise").arg(&supervisor_path); command.current_dir(&options.cwd); - command.env("AGENTCTL_AGENT_PROMPT", &options.prompt); - command.env("AGENTCTL_AGENT_CWD", &options.cwd); command.stdin(Stdio::null()); command.stdout(Stdio::from(stdout)); command.stderr(Stdio::from(stderr)); - let child = command.spawn().with_context(|| { - format!( - "launching detached agent `{}`", - format_command(&command_argv(&options.agent, &options.prompt)) - ) - })?; + let child = spawn_detached(&mut command) + .with_context(|| format!("launching detached agent `{}`", format_command(&argv)))?; let stored = StoredJobRecord { id: job_id, agent: options.agent_name.clone(), pid: child.id(), started_at, - command: format_command(&command_argv(&options.agent, &options.prompt)), + command: format_command(&argv), cwd: options.cwd.clone(), iterations: options.iterations, log_path, @@ -159,6 +174,54 @@ pub fn launch_detached(options: &AgentRunOptions) -> Result { record_from_stored(stored) } +pub fn supervise(path: &Path) -> Result { + let raw = fs::read_to_string(path).with_context(|| format!("reading {}", path.display()))?; + let spec: SupervisorSpec = + toml::from_str(&raw).with_context(|| format!("parsing {}", path.display()))?; + // The prompt only needs to exist long enough for the detached supervisor + // to start. Do not leave the transient spec on disk for completed jobs. + let _ = fs::remove_file(path); + + let mut exit_code = 0; + for iteration in 1..=spec.iterations { + if spec.iterations > 1 { + println!( + "[iteration {iteration}/{}] {}", + spec.iterations, + format_command(&spec.argv) + ); + } + let capture_options = CaptureOptions { + max_output_bytes: 16_384, + tail_bytes: 4_096, + label: Some(format!("{}-detached-iteration-{iteration}", spec.label)), + stream_raw: true, + }; + let process_options = agent_process_options( + &spec.cwd, + &spec.prompt, + spec.prompt_stdin, + iteration, + spec.iterations, + ); + match run_captured_with(&spec.argv, &capture_options, &process_options) { + Ok(run) => { + exit_code = exit_status_code(run.exit_status); + if exit_code != 0 { + break; + } + } + Err(error) => { + eprintln!("agent supervisor failed: {error:#}"); + exit_code = 1; + break; + } + } + } + write_exit_record(&spec.exit_code_path, exit_code)?; + Ok(exit_code) +} + pub fn status(job_id: &str, tail: usize) -> Result { let stored = read_job(job_id)?; let tail_lines = read_tail_lines(&stored.log_path, tail)?; @@ -210,48 +273,51 @@ pub fn logs(job_id: &str, follow: bool) -> Result<()> { Ok(()) } -fn run_agent_once(options: &AgentRunOptions, iteration: u32) -> Result { +fn run_agent_once( + options: &AgentRunOptions, + summary_config: &SummaryConfig, + iteration: u32, +) -> Result { let argv = command_argv(&options.agent, &options.prompt); - let raw_log_path = - create_job_output_log(&format!("{}-iteration-{iteration}", options.agent_name))?; - let start = Instant::now(); - let mut command = ProcessCommand::new(&argv[0]); - command.args(&argv[1..]); - command.current_dir(&options.cwd); - command.env("AGENTCTL_AGENT_PROMPT", &options.prompt); - command.env("AGENTCTL_AGENT_CWD", &options.cwd); - command.env("AGENTCTL_AGENT_ITERATION", iteration.to_string()); - command.env("AGENTCTL_AGENT_ITERATIONS", options.iterations.to_string()); - command.stdout(Stdio::piped()).stderr(Stdio::piped()); - if options.agent.prompt_stdin { - command.stdin(Stdio::piped()); - } + let capture_options = CaptureOptions { + max_output_bytes: summary_config.max_output_bytes, + tail_bytes: summary_config.tail_bytes, + label: Some(format!("{}-iteration-{iteration}", options.agent_name)), + stream_raw: false, + }; + let process_options = agent_process_options( + &options.cwd, + &options.prompt, + options.agent.prompt_stdin, + iteration, + options.iterations, + ); + run_captured_with(&argv, &capture_options, &process_options) +} - let mut child = command - .spawn() - .with_context(|| format!("launching agent `{}`", format_command(&argv)))?; - if options.agent.prompt_stdin - && let Some(mut stdin) = child.stdin.take() - { - stdin - .write_all(options.prompt.as_bytes()) - .context("writing prompt to agent stdin")?; +fn agent_process_options( + cwd: &Path, + prompt: &str, + prompt_stdin: bool, + iteration: u32, + iterations: u32, +) -> ProcessOptions { + ProcessOptions { + cwd: Some(cwd.to_path_buf()), + env: vec![ + ("AGENTCTL_AGENT_PROMPT".into(), prompt.into()), + ("AGENTCTL_AGENT_CWD".into(), cwd.as_os_str().to_owned()), + ( + "AGENTCTL_AGENT_ITERATION".into(), + iteration.to_string().into(), + ), + ( + "AGENTCTL_AGENT_ITERATIONS".into(), + iterations.to_string().into(), + ), + ], + stdin: prompt_stdin.then(|| prompt.as_bytes().to_vec()), } - - let output = child.wait_with_output().context("waiting for agent")?; - let elapsed = start.elapsed(); - let stdout = String::from_utf8_lossy(&output.stdout).to_string(); - let stderr = String::from_utf8_lossy(&output.stderr).to_string(); - persist_raw_log(&raw_log_path, &stdout, &stderr)?; - - Ok(CapturedRun { - command: argv, - exit_status: output.status, - elapsed, - stdout, - stderr, - raw_log_path, - }) } fn validate_options(options: &AgentRunOptions) -> Result<()> { @@ -275,38 +341,6 @@ fn command_argv(agent: &AgentConfig, prompt: &str) -> Vec { argv } -fn detached_script( - argv: &[String], - prompt_stdin: bool, - iterations: u32, - exit_code_path: &Path, -) -> String { - let command = shell_command(argv); - let display = shell_single_quote(&format_command(argv)); - let exit_code_path = shell_single_quote(&exit_code_path.display().to_string()); - let invoke = if prompt_stdin { - format!( - "printf '%s' \"$AGENTCTL_AGENT_PROMPT\" | AGENTCTL_AGENT_ITERATION=\"$iteration\" {command}" - ) - } else { - format!("AGENTCTL_AGENT_ITERATION=\"$iteration\" {command}") - }; - format!( - "status=0; for iteration in $(seq 1 {iterations}); do if [ {iterations} -gt 1 ]; then printf '[iteration %s/{iterations}] %s\\n' \"$iteration\" {display}; fi; {invoke}; status=$?; if [ $status -ne 0 ]; then break; fi; done; finished_at=$(date +%s); {{ printf '%s\\n' \"$status\"; printf 'finished_at=%s\\n' \"$finished_at\"; }} > {exit_code_path}; exit $status" - ) -} - -fn shell_command(argv: &[String]) -> String { - argv.iter() - .map(|argument| shell_single_quote(argument)) - .collect::>() - .join(" ") -} - -fn shell_single_quote(value: &str) -> String { - format!("'{}'", value.replace('\'', "'\\''")) -} - fn record_from_stored(stored: StoredJobRecord) -> Result { let exit_record = read_exit_record(&stored.exit_code_path)?; let state = if process_is_running(stored.pid) && exit_record.exit_code.is_none() { @@ -352,6 +386,29 @@ fn read_exit_record(path: &Path) -> Result { }) } +fn write_exit_record(path: &Path, exit_code: i32) -> Result<()> { + let finished_at = unix_timestamp()?; + fs::write(path, format!("{exit_code}\nfinished_at={finished_at}\n")) + .with_context(|| format!("writing {}", path.display())) +} + +fn exit_status_code(status: ExitStatus) -> i32 { + if let Some(code) = status.code() { + return code; + } + + #[cfg(unix)] + { + use std::os::unix::process::ExitStatusExt; + status.signal().map(|signal| 128 + signal).unwrap_or(1) + } + + #[cfg(not(unix))] + { + 1 + } +} + fn read_job(job_id: &str) -> Result { read_job_path(&jobs_dir()?.join(job_id).join("job.toml")) } @@ -378,35 +435,6 @@ fn jobs_dir() -> Result { Ok(PathBuf::from(home).join(CONFIG_DIR).join("jobs")) } -fn create_job_output_log(label: &str) -> Result { - let timestamp = SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap_or_default() - .as_nanos(); - let dir = jobs_dir()?.join(format!("{}-{}", timestamp, safe_name(label))); - fs::create_dir_all(&dir).with_context(|| format!("creating {}", dir.display()))?; - Ok(dir.join("output.log")) -} - -fn persist_raw_log(path: &Path, stdout: &str, stderr: &str) -> Result<()> { - let mut raw = String::new(); - if !stdout.is_empty() { - raw.push_str("stdout:\n"); - raw.push_str(stdout); - if !stdout.ends_with('\n') { - raw.push('\n'); - } - } - if !stderr.is_empty() { - raw.push_str("stderr:\n"); - raw.push_str(stderr); - if !stderr.ends_with('\n') { - raw.push('\n'); - } - } - fs::write(path, raw).with_context(|| format!("writing {}", path.display())) -} - fn job_id(started_at: u64, agent_name: &str) -> String { format!( "{}-{}-{}", diff --git a/src/lib.rs b/src/lib.rs index e7c8019..0d0c698 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -8,4 +8,5 @@ pub mod config; pub mod exec; pub mod jobs; pub mod output; +mod process; pub mod summary; diff --git a/src/main.rs b/src/main.rs index 65dbe22..f6749c8 100644 --- a/src/main.rs +++ b/src/main.rs @@ -56,6 +56,8 @@ enum Command { #[arg(long)] follow: bool, }, + #[command(name = "__supervise", hide = true)] + Supervise { spec: PathBuf }, Config { #[command(subcommand)] command: ConfigCommand, @@ -188,6 +190,10 @@ fn main() -> Result<()> { Command::Logs { job_id, follow } => { agentctl::jobs::logs(&job_id, follow)?; } + Command::Supervise { spec } => { + let exit_code = agentctl::jobs::supervise(&spec)?; + std::process::exit(exit_code); + } Command::Config { command } => match command { ConfigCommand::Generate => { let path = agentctl::config::generate()?; diff --git a/src/process.rs b/src/process.rs new file mode 100644 index 0000000..7619dc2 --- /dev/null +++ b/src/process.rs @@ -0,0 +1,306 @@ +use std::io; +use std::process::{Child, ChildStderr, ChildStdin, ChildStdout, Command, ExitStatus}; +use std::time::{Duration, Instant}; + +#[cfg(unix)] +use std::sync::OnceLock; +#[cfg(unix)] +use std::sync::atomic::{AtomicI32, Ordering}; + +#[cfg(unix)] +use signal_hook::consts::signal::{SIGHUP, SIGINT, SIGQUIT, SIGTERM}; +#[cfg(unix)] +use signal_hook::iterator::Signals; +#[cfg(unix)] +use std::os::unix::process::CommandExt; + +const TERMINATION_GRACE: Duration = Duration::from_millis(750); + +#[cfg(unix)] +static ACTIVE_PROCESS_GROUP: AtomicI32 = AtomicI32::new(0); +#[cfg(unix)] +static SIGNAL_FORWARDER: OnceLock<()> = OnceLock::new(); + +/// A child that owns an isolated process group. Waiting for the command also +/// terminates descendants which outlive their group leader. Dropping it on an +/// error follows the same TERM-then-KILL cleanup path and always reaps the +/// direct child. +pub(crate) struct ManagedChild { + child: Child, + #[cfg(unix)] + process_group: i32, + reaped: bool, +} + +/// Spawn a detached supervisor as its own process-group leader. The supervisor +/// is responsible for recording completion and terminating its group. +pub(crate) fn spawn_detached(command: &mut Command) -> io::Result { + configure_process_group(command); + command.spawn() +} + +impl ManagedChild { + pub(crate) fn spawn(command: &mut Command) -> io::Result { + enable_subreaper()?; + configure_process_group(command); + install_signal_forwarder()?; + let child = command.spawn()?; + + #[cfg(unix)] + { + let process_group = child.id() as i32; + ACTIVE_PROCESS_GROUP.store(process_group, Ordering::SeqCst); + Ok(Self { + child, + process_group, + reaped: false, + }) + } + + #[cfg(not(unix))] + Ok(Self { + child, + reaped: false, + }) + } + + pub(crate) fn take_stdin(&mut self) -> Option { + self.child.stdin.take() + } + + pub(crate) fn take_stdout(&mut self) -> Option { + self.child.stdout.take() + } + + pub(crate) fn take_stderr(&mut self) -> Option { + self.child.stderr.take() + } + + pub(crate) fn wait(&mut self) -> io::Result { + let status = self.child.wait()?; + self.reaped = true; + self.clean_remaining_descendants(); + self.clear_active_group(); + Ok(status) + } + + fn clean_remaining_descendants(&self) { + #[cfg(unix)] + terminate_process_tree(self.process_group); + } + + fn clear_active_group(&self) { + #[cfg(unix)] + { + let _ = ACTIVE_PROCESS_GROUP.compare_exchange( + self.process_group, + 0, + Ordering::SeqCst, + Ordering::SeqCst, + ); + } + } +} + +impl Drop for ManagedChild { + fn drop(&mut self) { + if self.reaped { + return; + } + + #[cfg(unix)] + terminate_process_tree(self.process_group); + #[cfg(not(unix))] + let _ = self.child.kill(); + + let _ = self.child.wait(); + self.reaped = true; + self.clear_active_group(); + } +} + +#[cfg(unix)] +fn configure_process_group(command: &mut Command) { + command.process_group(0); +} + +#[cfg(not(unix))] +fn configure_process_group(_command: &mut Command) {} + +#[cfg(unix)] +fn install_signal_forwarder() -> io::Result<()> { + if SIGNAL_FORWARDER.get().is_some() { + return Ok(()); + } + + let mut signals = Signals::new([SIGINT, SIGTERM, SIGHUP, SIGQUIT])?; + SIGNAL_FORWARDER.get_or_init(|| { + std::thread::spawn(move || { + for signal in signals.forever() { + let process_group = ACTIVE_PROCESS_GROUP.load(Ordering::SeqCst); + if process_group > 0 { + // Negative PIDs address the complete process group. + unsafe { + libc::kill(-process_group, signal); + } + } else { + // Do not turn agentctl into a process that ignores Ctrl-C + // while no managed child is active. + let _ = signal_hook::low_level::emulate_default_handler(signal); + } + } + }); + }); + Ok(()) +} + +#[cfg(target_os = "linux")] +fn enable_subreaper() -> io::Result<()> { + let result = unsafe { libc::prctl(libc::PR_SET_CHILD_SUBREAPER, 1, 0, 0, 0) }; + if result == 0 { + Ok(()) + } else { + Err(io::Error::last_os_error()) + } +} + +#[cfg(not(target_os = "linux"))] +fn enable_subreaper() -> io::Result<()> { + Ok(()) +} + +#[cfg(not(unix))] +fn install_signal_forwarder() -> io::Result<()> { + Ok(()) +} + +#[cfg(unix)] +fn terminate_process_tree(process_group: i32) { + if process_group <= 0 { + return; + } + + unsafe { + libc::kill(-process_group, SIGTERM); + } + signal_adopted_descendants(SIGTERM); + let deadline = Instant::now() + TERMINATION_GRACE; + while Instant::now() < deadline { + reap_adopted_children(); + signal_adopted_descendants(SIGTERM); + if !process_tree_exists(process_group) { + return; + } + std::thread::sleep(Duration::from_millis(20)); + } + + unsafe { + libc::kill(-process_group, libc::SIGKILL); + } + signal_adopted_descendants(libc::SIGKILL); + let kill_deadline = Instant::now() + TERMINATION_GRACE; + while Instant::now() < kill_deadline { + reap_adopted_children(); + signal_adopted_descendants(libc::SIGKILL); + if !process_tree_exists(process_group) { + return; + } + std::thread::sleep(Duration::from_millis(20)); + } + reap_adopted_children(); +} + +#[cfg(unix)] +fn process_tree_exists(process_group: i32) -> bool { + process_group_exists(process_group) || adopted_children_exist() +} + +#[cfg(unix)] +fn process_group_exists(process_group: i32) -> bool { + let result = unsafe { libc::kill(-process_group, 0) }; + if result == 0 { + return true; + } + io::Error::last_os_error().raw_os_error() == Some(libc::EPERM) +} + +#[cfg(target_os = "linux")] +fn reap_adopted_children() { + loop { + let mut status = 0; + let result = unsafe { libc::waitpid(-1, &mut status, libc::WNOHANG) }; + if result <= 0 { + break; + } + } +} + +#[cfg(all(unix, not(target_os = "linux")))] +fn reap_adopted_children() {} + +#[cfg(target_os = "linux")] +fn adopted_children_exist() -> bool { + !linux_child_pids(std::process::id() as i32).is_empty() +} + +#[cfg(all(unix, not(target_os = "linux")))] +fn adopted_children_exist() -> bool { + false +} + +#[cfg(target_os = "linux")] +fn signal_adopted_descendants(signal: i32) { + let mut pending = linux_child_pids(std::process::id() as i32); + while let Some(pid) = pending.pop() { + pending.extend(linux_child_pids(pid)); + unsafe { + libc::kill(pid, signal); + } + } +} + +#[cfg(all(unix, not(target_os = "linux")))] +fn signal_adopted_descendants(_signal: i32) {} + +#[cfg(target_os = "linux")] +fn linux_child_pids(pid: i32) -> Vec { + let path = format!("/proc/{pid}/task/{pid}/children"); + std::fs::read_to_string(path) + .unwrap_or_default() + .split_whitespace() + .filter_map(|value| value.parse().ok()) + .collect() +} + +#[cfg(test)] +mod tests { + use super::*; + use std::process::Stdio; + + #[test] + #[cfg(unix)] + fn wait_terminates_descendants_left_by_the_direct_child() { + let temp = tempfile::tempdir().expect("tempdir"); + let pid_path = temp.path().join("descendant.pid"); + let script = format!( + "setsid sleep 30 & printf '%s' $! > '{}'; exit 0", + pid_path.display() + ); + let mut command = Command::new("bash"); + command + .args(["-c", &script]) + .stdout(Stdio::null()) + .stderr(Stdio::null()); + + let mut child = ManagedChild::spawn(&mut command).expect("spawn managed child"); + let status = child.wait().expect("wait managed child"); + assert!(status.success()); + + let pid = std::fs::read_to_string(pid_path) + .expect("descendant pid") + .parse::() + .expect("numeric pid"); + let running = unsafe { libc::kill(pid, 0) } == 0; + assert!(!running, "descendant {pid} survived its agent command"); + } +} diff --git a/tests/agent_jobs_cli.rs b/tests/agent_jobs_cli.rs index 89efe80..8202082 100644 --- a/tests/agent_jobs_cli.rs +++ b/tests/agent_jobs_cli.rs @@ -173,6 +173,87 @@ fn detached_trivial_agent_round_trips_run_status_logs() { assert!(logs.contains("done")); } +#[test] +#[cfg(unix)] +fn detached_supervisor_reaps_descendants_that_create_a_new_session() { + let home_dir = tempfile::tempdir().expect("home tempdir"); + let project_dir = tempfile::tempdir().expect("project tempdir"); + let descendant_pid_path = project_dir.path().join("detached-descendant.pid"); + std::fs::write( + project_dir.path().join("agentctl.toml"), + format!( + r#" +[summary] +max_output_bytes = 1024 +tail_bytes = 256 +max_preview_lines = 12 + +[agents.detached_tree] +command = ["bash", "-c", "setsid sleep 30 & echo $! > '{}'; exit 0"] +prompt_stdin = false +"#, + descendant_pid_path.display() + ), + ) + .expect("write config"); + + let run_output = Command::cargo_bin("agentctl") + .expect("agentctl binary") + .current_dir(project_dir.path()) + .env("HOME", home_dir.path()) + .args([ + "run", + "detached_tree", + "--prompt", + "test", + "--detach", + "--json", + ]) + .output() + .expect("run detached process tree"); + assert!(run_output.status.success()); + let run_json: Value = serde_json::from_slice(&run_output.stdout).expect("run json"); + let job_id = run_json["id"].as_str().expect("job id"); + + let mut succeeded = false; + for _ in 0..100 { + let status_output = Command::cargo_bin("agentctl") + .expect("agentctl binary") + .current_dir(project_dir.path()) + .env("HOME", home_dir.path()) + .args(["status", job_id, "--json"]) + .output() + .expect("status detached process tree"); + let status: Value = serde_json::from_slice(&status_output.stdout).expect("status json"); + if status["record"]["state"] == "succeeded" { + succeeded = true; + break; + } + std::thread::sleep(std::time::Duration::from_millis(20)); + } + assert!(succeeded, "detached supervisor did not record success"); + + let descendant_pid = std::fs::read_to_string(&descendant_pid_path) + .expect("detached descendant pid") + .trim() + .parse::() + .expect("numeric detached descendant pid"); + let descendant_exists = unsafe { libc::kill(descendant_pid, 0) } == 0; + assert!( + !descendant_exists, + "detached descendant {descendant_pid} survived its supervisor" + ); + assert!( + !home_dir + .path() + .join(".agentctl/jobs") + .join(job_id) + .join("supervisor.toml") + .exists(), + "transient supervisor spec should be removed" + ); +} + #[test] fn dead_pid_without_exit_code_is_stale() { let home_dir = tempfile::tempdir().expect("home tempdir"); @@ -258,6 +339,66 @@ fn foreground_agent_accepts_prompt_file_and_cwd_override() { ); } +#[test] +#[cfg(unix)] +fn terminating_foreground_agentctl_terminates_and_reaps_the_agent_tree() { + use std::os::unix::process::ExitStatusExt; + + let home_dir = tempfile::tempdir().expect("home tempdir"); + let project_dir = tempfile::tempdir().expect("project tempdir"); + let descendant_pid_path = project_dir.path().join("descendant.pid"); + std::fs::write( + project_dir.path().join("agentctl.toml"), + format!( + r#" +[summary] +max_output_bytes = 1024 +tail_bytes = 256 +max_preview_lines = 12 + +[agents.tree] +command = ["bash", "-c", "setsid sleep 30 & echo $! > '{}'; wait"] +prompt_stdin = false +"#, + descendant_pid_path.display() + ), + ) + .expect("write config"); + + let mut agentctl = std::process::Command::new(assert_cmd::cargo::cargo_bin!("agentctl")) + .current_dir(project_dir.path()) + .env("HOME", home_dir.path()) + .args(["run", "tree", "--prompt", "test"]) + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::null()) + .spawn() + .expect("spawn foreground agentctl"); + + for _ in 0..100 { + if descendant_pid_path.exists() { + break; + } + std::thread::sleep(std::time::Duration::from_millis(20)); + } + let descendant_pid = std::fs::read_to_string(&descendant_pid_path) + .expect("descendant pid") + .trim() + .parse::() + .expect("numeric descendant pid"); + + let signal_result = unsafe { libc::kill(agentctl.id() as i32, libc::SIGTERM) }; + assert_eq!(signal_result, 0, "signal foreground agentctl"); + let status = agentctl.wait().expect("wait for foreground agentctl"); + assert!(!status.success()); + assert!(status.code().is_some() || status.signal().is_some()); + + let descendant_exists = unsafe { libc::kill(descendant_pid, 0) } == 0; + assert!( + !descendant_exists, + "agent descendant {descendant_pid} survived agentctl termination" + ); +} + fn assert_foreground_iteration_schema(json: &Value) { let object = json.as_object().expect("foreground iteration object"); assert_key_set(