diff --git a/Cargo.lock b/Cargo.lock index e859cac3..50c1db7b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -610,6 +610,7 @@ name = "gitagent-terminal" version = "3.0.0" dependencies = [ "anyhow", + "nix", "portable-pty", "tempfile", "tokio", diff --git a/README.md b/README.md index 2730d747..2b0840ef 100644 --- a/README.md +++ b/README.md @@ -490,11 +490,11 @@ done | `system/ready`, `system/session_start` | After `initialize`: the session id, the agent, the model, the tools | | `user`, `delta`, `assistant`, `tool_use`, `tool_result` | As the turn runs; `assistant` carries `usage` and `costUsd` | | `callback_request` (`kind`: `hook`, `tool` or `approval`) | The binary needs your answer; reply with `callback_response` and the same `id` | -| `system/turn_end` | A turn finished, with its `outcome` and usage | +| `system/turn_end` | A turn finished, with its `outcome` and usage (sub-agents' model calls included) | | `system/warning`, `system/error`, `system/hook_blocked` | Something to show; the session goes on unless the error was fatal | -| `system/session_end` | Last message, with the session's costs | +| `system/session_end` | Last message, with the session's costs (sub-agents' included) | -With `resume: true` the conversation is stored after every turn under `.gitagent/sessions//`, and a new process started with the same `sessionId` continues it. +With `resume: true` the conversation is stored after every turn under `.gitagent/sessions//`, and a new process started with the same `sessionId` continues it. One process at a time holds a stored session: a second one opened while the first runs gets `system/error` and stops. ### Rust diff --git a/crates/cli/src/ipc.rs b/crates/cli/src/ipc.rs index eb0855e5..32450689 100644 --- a/crates/cli/src/ipc.rs +++ b/crates/cli/src/ipc.rs @@ -30,11 +30,12 @@ use std::sync::{Arc, Mutex}; use std::time::Duration; use gitagent_core::{ - Approval, ApprovalRequest, Approver, ExternalTools, Outcome, Resume, RunOptions, Steering, + Approval, ApprovalRequest, Approver, ExternalTools, Outcome, Resume, RunOptions, RunReport, + Steering, }; use gitagent_hooks::{Decision, ToolHooks}; use gitagent_loader::v1; -use gitagent_provider::{Message, ToolSpec}; +use gitagent_provider::{Message, ToolSpec, Usage}; use gitagent_record::Events; use serde_json::{json, Map, Value}; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; @@ -401,6 +402,47 @@ struct Totals { } impl Totals { + /// `usage` from `requests` model calls, priced at `prices`. + fn priced(prices: Option, usage: &Usage, requests: u64) -> Totals { + Totals { + input: usage.input_tokens, + output: usage.output_tokens, + cache_read: usage.cache_read_tokens, + cache_write: usage.cache_write_tokens, + cost: gitagent_sdk::cost_of( + prices, + &gitagent_sdk::Usage { + input_tokens: usage.input_tokens, + output_tokens: usage.output_tokens, + cache_read_input_tokens: usage.cache_read_tokens, + cache_creation_input_tokens: usage.cache_write_tokens, + }, + ), + requests, + } + } + + fn add(&mut self, other: &Totals) { + self.input += other.input; + self.output += other.output; + self.cache_read += other.cache_read; + self.cache_write += other.cache_write; + self.cost += other.cost; + self.requests += other.requests; + } + + /// What was added since `before`. + fn since(&self, before: &Totals) -> Totals { + Totals { + input: self.input - before.input, + output: self.output - before.output, + cache_read: self.cache_read - before.cache_read, + cache_write: self.cache_write - before.cache_write, + cost: self.cost - before.cost, + requests: self.requests - before.requests, + } + } + fn json(&self) -> Value { json!({ "inputTokens": self.input, @@ -420,10 +462,31 @@ struct Stream { cb: Arc, provider: String, prices: Option, - totals: Mutex, + /// The model calls streamed so far: the parent's only, since sub-agents + /// run without events. + streamed: Mutex, } impl Stream { + fn streamed(&self) -> Totals { + *self.streamed.lock().unwrap_or_else(|e| e.into_inner()) + } + + /// What a run used. A finished run's report has every model call, + /// its sub-agents' included; without one, only the streamed calls since + /// `before` are known. + fn run_usage(&self, report: Option<&RunReport>, before: &Totals) -> Totals { + let streamed = self.streamed().since(before); + match report { + Some(r) => Totals::priced( + self.prices, + &r.usage, + streamed.requests + r.counters.get("subagent_turns"), + ), + None => streamed, + } + } + fn feed(&self, e: &Value) { let kind = e["event"].as_str().unwrap_or_default(); match kind { @@ -472,31 +535,17 @@ impl Stream { } let u = &e["usage"]; let n = |k: &str| u[k].as_u64().unwrap_or(0); - let turn = Totals { - input: n("input_tokens"), - output: n("output_tokens"), - cache_read: n("cache_read_tokens"), - cache_write: n("cache_write_tokens"), - cost: gitagent_sdk::cost_of( - self.prices, - &gitagent_sdk::Usage { - input_tokens: n("input_tokens"), - output_tokens: n("output_tokens"), - cache_read_input_tokens: n("cache_read_tokens"), - cache_creation_input_tokens: n("cache_write_tokens"), - }, - ), - requests: 1, + let usage = Usage { + input_tokens: n("input_tokens"), + output_tokens: n("output_tokens"), + cache_read_tokens: n("cache_read_tokens"), + cache_write_tokens: n("cache_write_tokens"), }; - { - let mut t = self.totals.lock().unwrap_or_else(|e| e.into_inner()); - t.input += turn.input; - t.output += turn.output; - t.cache_read += turn.cache_read; - t.cache_write += turn.cache_write; - t.cost += turn.cost; - t.requests += 1; - } + let turn = Totals::priced(self.prices, &usage, 1); + self.streamed + .lock() + .unwrap_or_else(|e| e.into_inner()) + .add(&turn); let (stop, error) = match ( e["tool_calls"].as_u64().unwrap_or(0), e["stop_reason"].as_str(), @@ -613,6 +662,31 @@ fn session_file(dir: &Path, id: &str) -> Result { .join("messages.json")) } +/// Locks the stored session `id` for this process (an advisory lock on a +/// `lock` file next to the conversation). The lock lasts as long as the +/// returned file is open; the system drops it when the process ends, so a +/// crashed process leaves nothing to clean up. +fn lock_session(store: &Path, id: &str) -> Result { + let path = store.with_file_name("lock"); + let cannot = |e: std::io::Error| format!("the session lock {} failed: {e}", path.display()); + if let Some(dir) = path.parent() { + std::fs::create_dir_all(dir).map_err(cannot)?; + } + let file = std::fs::OpenOptions::new() + .create(true) + .truncate(false) + .write(true) + .open(&path) + .map_err(cannot)?; + match file.try_lock() { + Ok(()) => Ok(file), + Err(std::fs::TryLockError::WouldBlock) => Err(format!( + "session {id:?} is open in another process; end that one first or use another sessionId" + )), + Err(std::fs::TryLockError::Error(e)) => Err(cannot(e)), + } +} + pub(crate) async fn run(cli: Cli) -> ! { let (tx, mut rx) = mpsc::unbounded_channel::(); let out = Out(tx); @@ -630,8 +704,12 @@ pub(crate) async fn run(cli: Cli) -> ! { let _ = stdout.flush().await; } }); - // `session` drops its shells and MCP servers before returning. - let code = session(cli, out.clone()).await; + // `session` drops its shells and MCP servers before returning. A stop + // signal drops it too, so the commands the shells run do not outlive us. + let code = tokio::select! { + code = session(cli, out.clone()) => code, + signal = stop_signal() => 128 + signal, + }; out.send(Value::Null); let _ = writer.await; // The stdin reader may still be waiting in a blocking read the runtime @@ -639,6 +717,26 @@ pub(crate) async fn run(cli: Cli) -> ! { std::process::exit(code) } +/// Waits for SIGTERM or SIGINT and returns its number. +async fn stop_signal() -> i32 { + use tokio::signal::unix::{signal, SignalKind}; + let wait_for = |kind: SignalKind| async move { + match signal(kind) { + Ok(mut stream) => { + stream.recv().await; + kind.as_raw_value() + } + // No handler was installed, so the signal keeps its default + // action (ending the process) and this branch never fires. + Err(_) => std::future::pending().await, + } + }; + tokio::select! { + number = wait_for(SignalKind::terminate()) => number, + number = wait_for(SignalKind::interrupt()) => number, + } +} + async fn session(cli: Cli, out: Out) -> i32 { let mut lines = BufReader::new(tokio::io::stdin()).lines(); let fatal = |out: &Out, why: String| { @@ -772,6 +870,16 @@ async fn session(cli: Cli, out: Out) -> i32 { } else { None }; + // Held until the process ends, so a second process cannot load and + // overwrite the same stored conversation. + let _session_lock = match store + .as_deref() + .map(|p| lock_session(p, &session_id)) + .transpose() + { + Ok(lock) => lock, + Err(e) => return fatal(&out, e), + }; let agent_name = s .spec .agent @@ -803,7 +911,7 @@ async fn session(cli: Cli, out: Out) -> i32 { cb: cb.clone(), provider: s.spec.model.provider.clone(), prices, - totals: Mutex::new(Totals::default()), + streamed: Mutex::new(Totals::default()), }); let (mcp, warnings) = gitagent_core::connect_mcp(&s.spec).await; @@ -956,6 +1064,8 @@ async fn session(cli: Cli, out: Out) -> i32 { .has("approval") .then(|| Arc::new(ClientApprover(cb.clone())) as Arc); + // Every model call of the session, sub-agents' included. + let mut session_usage = Totals::default(); let mut queue: VecDeque = VecDeque::new(); let mut ending = false; let mut n = 0; @@ -994,7 +1104,7 @@ async fn session(cli: Cli, out: Out) -> i32 { let stream = stream.clone(); Events::to_callback(move |e| stream.feed(&e)) }; - let before = *stream.totals.lock().unwrap_or_else(|e| e.into_inner()); + let before = stream.streamed(); let opts = RunOptions { instruction: content, work_dir: s.work_dir.clone(), @@ -1040,15 +1150,12 @@ async fn session(cli: Cli, out: Out) -> i32 { for late in ctl.steering.take().into_iter().rev() { queue.push_front(late); } - let after = *stream.totals.lock().unwrap_or_else(|e| e.into_inner()); - let turn_usage = Totals { - input: after.input - before.input, - output: after.output - before.output, - cache_read: after.cache_read - before.cache_read, - cache_write: after.cache_write - before.cache_write, - cost: after.cost - before.cost, - requests: after.requests - before.requests, - }; + let report = result + .as_ref() + .and_then(|r| r.as_ref().ok()) + .map(|f| &f.report); + let turn_usage = stream.run_usage(report, &before); + session_usage.add(&turn_usage); match result { None => { out.system( @@ -1145,7 +1252,14 @@ async fn session(cli: Cli, out: Out) -> i32 { external_tools: external.clone(), throttle: None, }; - match gitagent_core::teardown(&s.spec, s.provider.as_ref(), opts).await { + let before = stream.streamed(); + let done = gitagent_core::teardown(&s.spec, s.provider.as_ref(), opts).await; + let report = done + .as_ref() + .and_then(|r| r.as_ref().ok()) + .map(|f| &f.report); + session_usage.add(&stream.run_usage(report, &before)); + match done { Some(Ok(done)) => out.system( "teardown", done.report.summary.clone().unwrap_or_default(), @@ -1174,8 +1288,7 @@ async fn session(cli: Cli, out: Out) -> i32 { } } } - let totals = *stream.totals.lock().unwrap_or_else(|e| e.into_inner()); - end["costs"] = totals.json(); + end["costs"] = session_usage.json(); out.system( "session_end", format!("Agent {agent_name} finished"), diff --git a/crates/cli/tests/ipc.rs b/crates/cli/tests/ipc.rs index 59679c15..bddfa49e 100644 --- a/crates/cli/tests/ipc.rs +++ b/crates/cli/tests/ipc.rs @@ -325,6 +325,63 @@ fn steering_and_abort() { assert!(is("system", "session_end")(rest.last().unwrap())); } +fn process_exists(pid: &str) -> bool { + Command::new("kill") + .args(["-0", pid]) + .stderr(Stdio::null()) + .status() + .unwrap() + .success() +} + +#[test] +fn sigterm_stops_the_running_command() { + let tmp = tempfile::tempdir().unwrap(); + let work = tmp.path().join("work"); + std::fs::create_dir_all(&work).unwrap(); + let model = script( + tmp.path(), + "s.yaml", + "- tool_calls: [{name: exec, arguments: {cmd: \"sh -c 'echo $$ > pid.txt; exec sleep 300'\", yield_ms: 60000}}]\n\ + - text: \"unreachable\"\n", + ); + let mut s = Session::start(tmp.path(), &tmp.path().join("out")); + s.send(json!({"type": "initialize", "protocol": 1, "options": {"model": model, "dir": work, "sessionId": "sig"}})); + s.until(is("system", "session_start")); + s.send(json!({"type": "user", "content": "sleep"})); + s.until(|m| m["type"] == "tool_use"); + let pid_file = work.join("pid.txt"); + let mut pid = String::new(); + for _ in 0..100 { + pid = std::fs::read_to_string(&pid_file) + .unwrap_or_default() + .trim() + .to_string(); + if !pid.is_empty() { + break; + } + std::thread::sleep(Duration::from_millis(50)); + } + assert!(process_exists(&pid), "the command did not start: {pid:?}"); + + let gitagent = s.child.id().to_string(); + assert!(Command::new("kill") + .args(["-TERM", &gitagent]) + .status() + .unwrap() + .success()); + let code = s.child.wait().unwrap().code(); + assert_eq!(code, Some(143)); + // The shell is gone at once; its orphaned command is reaped shortly after. + for _ in 0..100 { + if !process_exists(&pid) { + return; + } + std::thread::sleep(Duration::from_millis(50)); + } + panic!("pid {pid} survived SIGTERM"); +} + #[test] fn startup_problems_and_blocking_hooks() { let tmp = tempfile::tempdir().unwrap(); @@ -371,6 +428,115 @@ fn startup_problems_and_blocking_hooks() { assert_eq!(code, 0); } +/// Sub-agents run without events, but what they use counts in the turn's +/// and the session's usage. +#[test] +fn usage_includes_sub_agents() { + let tmp = tempfile::tempdir().unwrap(); + let work = tmp.path().join("work"); + agent(&work, ""); + std::fs::create_dir_all(work.join("agents")).unwrap(); + std::fs::write( + work.join("agents/helper.md"), + "---\nname: helper\ndescription: Looks things up.\n---\nAnswer briefly.\n", + ) + .unwrap(); + // The parent starts the helper, the helper answers, the parent replies. + let model = script( + tmp.path(), + "s.yaml", + "- tool_calls: [{name: subagent, arguments: {description: look up, prompt: \"Find the answer.\", agent: helper}}]\n\ + - text: \"The answer is 42.\"\n\ + - text: \"It is 42.\"\n", + ); + let mut s = Session::start(tmp.path(), &tmp.path().join("out")); + s.send(json!({ + "type": "initialize", "protocol": 1, + "options": {"model": model, "dir": work, "prices": {"input": 100.0, "output": 100.0}} + })); + s.until(is("system", "session_start")); + s.send(json!({"type": "user", "content": "what is the answer?"})); + let turn = s.until(is("system", "turn_end")); + let streamed: Vec<&Value> = turn.iter().filter(|m| m["type"] == "assistant").collect(); + assert_eq!( + streamed.len(), + 2, + "only the parent's model calls are streamed" + ); + let streamed_input: u64 = streamed + .iter() + .map(|m| m["usage"]["inputTokens"].as_u64().unwrap()) + .sum(); + let streamed_cost: f64 = streamed + .iter() + .map(|m| m["usage"]["costUsd"].as_f64().unwrap()) + .sum(); + let end = turn.last().unwrap(); + assert_eq!(end["metadata"]["result"], "It is 42."); + let usage = &end["metadata"]["usage"]; + assert_eq!(usage["requests"], 3, "{usage}"); + assert!( + usage["inputTokens"].as_u64().unwrap() > streamed_input, + "{usage}, streamed {streamed_input}" + ); + assert!( + usage["costUsd"].as_f64().unwrap() > streamed_cost, + "{usage}, streamed {streamed_cost}" + ); + s.send(json!({"type": "end"})); + let (code, rest) = s.finish(); + assert_eq!(code, 0); + let session_end = rest.last().unwrap(); + assert!(is("system", "session_end")(session_end)); + assert_eq!(&session_end["metadata"]["costs"], usage); +} + +/// Two processes on one stored session would each save over the other's +/// conversation, so the second is refused while the first runs. +#[test] +fn a_stored_session_is_open_in_one_process_at_a_time() { + let tmp = tempfile::tempdir().unwrap(); + let work = tmp.path().join("work"); + std::fs::create_dir_all(&work).unwrap(); + let model = script(tmp.path(), "s.yaml", "- text: \"hi\"\n"); + let open = |out: &str| { + let mut s = Session::start(tmp.path(), &tmp.path().join(out)); + s.send(json!({ + "type": "initialize", "protocol": 1, + "options": {"model": model, "dir": work, "sessionId": "shared", "resume": true} + })); + s + }; + let mut first = open("out1"); + first.until(is("system", "session_start")); + + let (code, rest) = open("out2").finish(); + assert_eq!(code, 2); + let error = rest + .iter() + .find(|m| is("system", "error")(m)) + .expect("an error"); + assert!( + error["content"] + .as_str() + .unwrap() + .contains("session \"shared\" is open in another process"), + "{error}" + ); + assert!(is("system", "session_end")(rest.last().unwrap())); + assert!(!rest.iter().any(|m| is("system", "session_start")(m))); + + // The first goes on, and once it has ended the session opens again. + first.send(json!({"type": "user", "content": "hello"})); + first.until(is("system", "turn_end")); + first.send(json!({"type": "end"})); + assert_eq!(first.finish().0, 0); + let mut again = open("out3"); + again.until(is("system", "session_start")); + again.send(json!({"type": "end"})); + assert_eq!(again.finish().0, 0); +} + /// 1.x's `repo` option: the session works in a clone on a session branch /// and the changes are committed and pushed when it ends; a second session /// with `session` set to that branch reuses the clone and resumes it. diff --git a/crates/loader/src/agent_hooks.rs b/crates/loader/src/agent_hooks.rs index 403023f5..a0fb7420 100644 --- a/crates/loader/src/agent_hooks.rs +++ b/crates/loader/src/agent_hooks.rs @@ -16,7 +16,9 @@ //! ``` //! //! Script paths are relative to `hooks/` and must stay inside the agent -//! directory. Problems are warnings and the hook is left out. +//! directory. Problems are warnings and the hook is left out, except for a +//! compliance or fail-closed hook: one that cannot be loaded stops the agent +//! from starting, since leaving it out would let through what it blocks. use std::path::Path; use std::time::Duration; @@ -113,8 +115,9 @@ pub struct HookEntry { } /// The hooks in an agent directory's `hooks/hooks.yaml`, in file order, and -/// what was wrong with the rest. -pub fn load(agent: &AgentDir) -> (Vec, Vec) { +/// what was wrong with the rest. A compliance or fail-closed hook that +/// cannot be loaded is an error. +pub fn load(agent: &AgentDir) -> Result<(Vec, Vec), String> { // The child's hooks, else the nearest parent's. let base = agent .file_root("hooks/hooks.yaml") @@ -122,33 +125,51 @@ pub fn load(agent: &AgentDir) -> (Vec, Vec) { let path = base.join("hooks/hooks.yaml"); let mut warnings = Vec::new(); let Ok(text) = std::fs::read_to_string(&path) else { - return (Vec::new(), warnings); + return Ok((Vec::new(), warnings)); }; let root: Value = match serde_saphyr::from_str(&text) { Ok(v) => v, Err(e) => { warnings.push(format!("hooks/hooks.yaml: not valid YAML: {e}")); - return (Vec::new(), warnings); + return Ok((Vec::new(), warnings)); } }; let Some(events) = root.get("hooks").and_then(Value::as_object) else { warnings.push("hooks/hooks.yaml: no `hooks:` mapping".into()); - return (Vec::new(), warnings); + return Ok((Vec::new(), warnings)); + }; + let required = |name: &str, i: usize, why: &str| { + format!( + "hooks/hooks.yaml: {name}[{i}]: {why}; it is a compliance or fail-closed hook, \ + so the agent cannot run without it" + ) }; let mut entries = Vec::new(); for (name, list) in events { + let hooks = list.as_array().map(Vec::as_slice).unwrap_or_default(); let Some(event) = HookEvent::parse(name) else { + if let Some(i) = hooks.iter().position(is_fail_closed) { + return Err(required(name, i, &format!("unknown event {name:?}"))); + } warnings.push(format!("hooks/hooks.yaml: unknown event {name:?}")); continue; }; - for (i, v) in list.as_array().into_iter().flatten().enumerate() { + for (i, v) in hooks.iter().enumerate() { match entry(event, v, &base) { Ok(e) => entries.push(e), + Err(why) if is_fail_closed(v) => return Err(required(name, i, &why)), Err(why) => warnings.push(format!("hooks/hooks.yaml: {name}[{i}]: {why}")), } } } - (entries, warnings) + Ok((entries, warnings)) +} + +/// Whether a hook's settings say a failure must block: a compliance hook, +/// or one with `fail_open: false`. +fn is_fail_closed(v: &Value) -> bool { + v.get("compliance").and_then(Value::as_bool) == Some(true) + || v.get("fail_open").and_then(Value::as_bool) == Some(false) } /// The event named `name`, if there is one. @@ -231,6 +252,16 @@ fn entry(event: HookEvent, v: &Value, root: &Path) -> Result crate::sdk::normalize_tool_list(&names) }) .unwrap_or_default(); + let compliance = flag("compliance"); + // As in 1.x, a hook that fails is a warning, not a block, unless it + // is a compliance hook or says otherwise. A compliance hook always blocks. + let fail_open = match v.get("fail_open").and_then(Value::as_bool) { + Some(true) if compliance => { + return Err("a compliance hook cannot set `fail_open: true`".into()) + } + Some(open) => open, + None => !compliance, + }; let mut command = crate::declared::interpreter(None, &path); command.push(path.display().to_string()); Ok(HookEntry { @@ -243,13 +274,8 @@ fn entry(event: HookEvent, v: &Value, root: &Path) -> Result .unwrap_or_default() .to_string(), timeout, - compliance: flag("compliance"), - // As in 1.x, a hook that fails is a warning, not a block, unless it - // is a compliance hook or says otherwise. - fail_open: v - .get("fail_open") - .and_then(Value::as_bool) - .unwrap_or(!flag("compliance")), + compliance, + fail_open, applies_to, }) } @@ -280,7 +306,7 @@ mod tests { ) .unwrap(); let agent = AgentDir::load(root).unwrap().unwrap(); - let (hooks, warnings) = load(&agent); + let (hooks, warnings) = load(&agent).unwrap(); assert_eq!(hooks.len(), 3); let a = &hooks[0]; assert_eq!( @@ -314,4 +340,57 @@ mod tests { ); } } + + /// An agent directory whose hooks/ holds `present.sh` and `hooks_yaml`. + fn agent_with_hooks(dir: &Path, hooks_yaml: &str) -> AgentDir { + std::fs::write( + dir.join("agent.yaml"), + "name: t\nversion: 1.0.0\ndescription: d\n", + ) + .unwrap(); + std::fs::write(dir.join("SOUL.md"), "I test.\n").unwrap(); + std::fs::create_dir_all(dir.join("hooks")).unwrap(); + std::fs::write(dir.join("hooks/present.sh"), "true\n").unwrap(); + std::fs::write(dir.join("hooks/hooks.yaml"), hooks_yaml).unwrap(); + AgentDir::load(dir).unwrap().unwrap() + } + + #[test] + fn a_compliance_or_fail_closed_hook_that_cannot_load_is_an_error() { + for (hooks_yaml, expected) in [ + ( + "hooks:\n pre_tool_use:\n - {script: present.sh}\n - {script: missing.sh, compliance: true}\n", + "pre_tool_use[1]: the script hooks/missing.sh does not exist", + ), + ( + "hooks:\n pre_tool_use:\n - {script: missing.sh, fail_open: false}\n", + "the script hooks/missing.sh does not exist", + ), + ( + "hooks:\n pre_tool_usee:\n - {script: present.sh, compliance: true}\n", + "unknown event \"pre_tool_usee\"", + ), + ( + "hooks:\n pre_tool_use:\n - {script: present.sh, compliance: true, fail_open: true}\n", + "a compliance hook cannot set `fail_open: true`", + ), + ] { + let dir = tempfile::tempdir().unwrap(); + let agent = agent_with_hooks(dir.path(), hooks_yaml); + let err = load(&agent).expect_err(hooks_yaml); + assert!(err.contains(expected), "{err}"); + assert!(err.contains("cannot run without it"), "{err}"); + } + + // An ordinary hook that cannot load is still only a warning. + let dir = tempfile::tempdir().unwrap(); + let agent = agent_with_hooks( + dir.path(), + "hooks:\n pre_tool_use:\n - {script: missing.sh}\n - {script: present.sh, compliance: true}\n", + ); + let (hooks, warnings) = load(&agent).unwrap(); + assert_eq!(hooks.len(), 1); + assert!(hooks[0].compliance && !hooks[0].fail_open); + assert_eq!(warnings.len(), 1, "{warnings:?}"); + } } diff --git a/crates/loader/src/lib.rs b/crates/loader/src/lib.rs index d700accb..5205baef 100644 --- a/crates/loader/src/lib.rs +++ b/crates/loader/src/lib.rs @@ -312,7 +312,7 @@ pub fn load(opts: &LoadOptions) -> Result { Some(a) => { let (mut tools, warnings) = declared::discover(a); a.warnings.extend(warnings); - let (mut hooks, warnings) = agent_hooks::load(a); + let (mut hooks, warnings) = agent_hooks::load(a).map_err(anyhow::Error::msg)?; a.warnings.extend(warnings); // Plugins add their tools and hooks after the agent's own. let taken: Vec = tools.iter().map(|t| t.name.clone()).collect(); diff --git a/crates/terminal/Cargo.toml b/crates/terminal/Cargo.toml index f8e26876..e631ac93 100644 --- a/crates/terminal/Cargo.toml +++ b/crates/terminal/Cargo.toml @@ -10,6 +10,7 @@ workspace = true [dependencies] anyhow.workspace = true +nix = { version = "0.28", default-features = false, features = ["signal", "process"] } portable-pty = "0.9.0" tokio.workspace = true diff --git a/crates/terminal/src/buffer.rs b/crates/terminal/src/buffer.rs index a9aecf8c..86a90ac2 100644 --- a/crates/terminal/src/buffer.rs +++ b/crates/terminal/src/buffer.rs @@ -80,7 +80,10 @@ impl OutputBuffer { return None; }; let line = String::from_utf8_lossy(&self.data[line_start..line_start + nl]); - let (code, cwd) = line.split_once(' ').unwrap_or((&line, "")); + // A terminal that still maps `\n` to `\r\n` (seen on macOS) ends + // the line with `\r`, which would otherwise end up in the cwd. + let line = line.trim_end_matches('\r'); + let (code, cwd) = line.split_once(' ').unwrap_or((line, "")); let Ok(exit) = code.parse::() else { // Not our marker (for example the text of the printf format). from = line_start; @@ -170,6 +173,17 @@ mod tests { assert_eq!(t.output, b"x\n__GA_END_2__ 0 /\n\n__GA_END_3__ %s %s\ny"); } + #[test] + fn strips_carriage_return_from_the_marker_cwd() { + let mut b = OutputBuffer::new(1 << 20); + b.push(b"out\r\n__GA_END_4__ 0 /tmp/dir\r\nrest"); + let (m, t) = b.take_marker(4).unwrap(); + assert_eq!(m.exit, 0); + assert_eq!(m.cwd, "/tmp/dir"); + assert_eq!(t.output, b"out\r"); + assert_eq!(b.take_all().output, b"rest"); + } + #[test] fn partial_take_holds_back_a_split_marker() { let mut b = OutputBuffer::new(1 << 20); diff --git a/crates/terminal/src/lib.rs b/crates/terminal/src/lib.rs index 0e149aaf..c9c21aa3 100644 --- a/crates/terminal/src/lib.rs +++ b/crates/terminal/src/lib.rs @@ -9,6 +9,7 @@ mod buffer; mod keys; mod manager; +mod stop; pub use keys::decode_keys; pub use manager::{Terminal, MAIN_SESSION, MAX_SESSIONS}; @@ -18,6 +19,7 @@ use std::path::{Path, PathBuf}; use std::time::{Duration, Instant}; use anyhow::{bail, Context, Result}; +use nix::unistd::Pid; use portable_pty::{native_pty_system, Child, CommandBuilder, MasterPty, PtySize}; use tokio::sync::mpsc; use tokio::time::timeout_at; @@ -109,6 +111,14 @@ enum ReadEvent { Eof, } +/// What ending a shell stops besides the shell. +enum Ending { + /// Every process in the shell's session, background jobs included. + Session, + /// Only the command in the terminal's foreground. + Foreground, +} + struct Shell { master: Box, writer: Box, @@ -287,9 +297,36 @@ impl Session { } } - /// Stops the shell and everything started from it. + /// Stops the shell and everything started from it: the running command + /// and background jobs, which job control puts in groups of their own. pub fn kill(&mut self) { + self.end_shell(Ending::Session); + } + + /// Stops the shell and the command it is running. Background jobs get + /// the shell's SIGHUP, as when a terminal window closes, so jobs started + /// with `nohup` (e.g. a server) keep running, including one started just + /// before the hang-up. + pub(crate) fn hang_up(&mut self) { + self.end_shell(Ending::Foreground); + } + + fn end_shell(&mut self, ending: Ending) { if let Some(mut shell) = self.shell.take() { + if let Some(pid) = shell.child.process_id().and_then(|p| i32::try_from(p).ok()) { + let pid = Pid::from_raw(pid); + let foreground = shell.master.process_group_leader().map(Pid::from_raw); + let tty = shell.master.tty_name(); + match ending { + Ending::Session => stop::stop_session(pid, tty.as_deref(), foreground), + Ending::Foreground => { + stop::stop_foreground(pid, foreground); + stop::let_jobs_start(pid, tty.as_deref()); + } + } + } + // SIGHUP, then SIGKILL, to the shell. Errors only mean it already + // exited; `wait` then just reaps it. let _ = shell.child.kill(); let _ = shell.child.wait(); } @@ -414,7 +451,7 @@ impl Session { ]); } } - self.kill(); + self.hang_up(); Ok(vec![ "The previous command did not stop on Ctrl-C, so its shell was killed.".into(), ]) diff --git a/crates/terminal/src/manager.rs b/crates/terminal/src/manager.rs index 873f8452..c9032e98 100644 --- a/crates/terminal/src/manager.rs +++ b/crates/terminal/src/manager.rs @@ -84,8 +84,12 @@ impl Terminal { let mut session = Session::start(cfg).await?; let mut r = session.exec(cmd, yield_ms).await?; r.session = id; + // A finished session only hangs up, so a server it started with + // `nohup` keeps running for the rest of the run (and for `--linger`). if session.is_running() { self.sessions.insert(id, session); + } else { + session.hang_up(); } Ok(r) } @@ -103,7 +107,9 @@ impl Terminal { r.session = id; // A background session is done once its command finishes. if id != MAIN_SESSION && !session.is_running() { - self.sessions.remove(&id); + if let Some(mut done) = self.sessions.remove(&id) { + done.hang_up(); + } } Ok(r) } @@ -251,4 +257,34 @@ mod tests { t.kill_all(); assert!(t.background_sessions().is_empty()); } + + #[tokio::test] + async fn a_finished_background_session_leaves_nohup_jobs_until_the_terminal_ends() { + use crate::tests::{parse_pid, process_exists, wait_until_gone}; + let (_d, mut t) = start().await; + let r = t + .exec("nohup sleep 300 >/dev/null 2>&1 & echo $!", 5000, true) + .await + .unwrap(); + assert!(r.finished); + assert!(t.background_sessions().is_empty()); + let pid = parse_pid(&r); + tokio::time::sleep(Duration::from_millis(500)).await; + assert!( + process_exists(pid), + "the nohup job was stopped with its session" + ); + + let r = t.exec("sleep 300 & echo $!", 5000, false).await.unwrap(); + let main_job = parse_pid(&r); + drop(t); + assert!(wait_until_gone(main_job).await, "pid {main_job} survived"); + // The finished session's job is no longer the terminal's to stop. + assert!(process_exists(pid)); + nix::sys::signal::kill( + nix::unistd::Pid::from_raw(pid), + nix::sys::signal::Signal::SIGKILL, + ) + .expect("stopping the nohup job"); + } } diff --git a/crates/terminal/src/stop.rs b/crates/terminal/src/stop.rs new file mode 100644 index 00000000..7f57d50b --- /dev/null +++ b/crates/terminal/src/stop.rs @@ -0,0 +1,162 @@ +//! Stopping everything a shell started. +//! +//! The shell leads its own session on its own pty (portable-pty calls +//! `setsid` before it starts the shell, and the sandbox launcher replaces +//! itself with the shell), so the session id is the shell's pid. With job +//! control on, every foreground and background job gets its own process group +//! in that session, so killing the shell alone leaves its jobs running. + +use std::path::Path; +use std::time::{Duration, Instant}; + +use nix::sys::signal::{killpg, Signal}; +use nix::unistd::Pid; + +/// How long programs get to exit after SIGTERM before they are killed. +const TERM_GRACE: Duration = Duration::from_secs(1); +const TERM_POLL: Duration = Duration::from_millis(20); + +/// How long the shell's jobs get to start before it is hung up. The shell +/// passes its SIGHUP on to every job, and `nohup` ignores SIGHUP only once it +/// runs, so a job forked just before the hang-up (`nohup server &` as the +/// last command) would die with the shell. +const JOB_START_GRACE: Duration = Duration::from_millis(200); + +/// Sends SIGTERM to every process group in the shell's session, waits up to +/// [`TERM_GRACE`] for them to exit, then sends SIGKILL to whatever is left, +/// the shell included. `tty` is the shell's terminal; `foreground` is that +/// terminal's foreground group, signalled even when the session's processes +/// cannot be listed. +pub(crate) fn stop_session(shell: Pid, tty: Option<&Path>, foreground: Option) { + let mut groups = session_groups(shell, tty); + groups.push(shell); + groups.extend(foreground); + groups.sort(); + groups.dedup(); + + terminate(&groups, shell); + + // Jobs started during the grace period (e.g. by a trap) are caught too. + groups.extend(session_groups(shell, tty)); + groups.sort(); + groups.dedup(); + signal_groups(&groups, Signal::SIGKILL); +} + +/// Stops the command running in the foreground of the shell's terminal, if +/// it is not the shell itself: SIGTERM, then SIGKILL after [`TERM_GRACE`]. +pub(crate) fn stop_foreground(shell: Pid, foreground: Option) { + let Some(group) = foreground.filter(|&g| g != shell) else { + return; + }; + terminate(&[group], shell); + signal_groups(&[group], Signal::SIGKILL); +} + +/// Waits [`JOB_START_GRACE`] if the shell has jobs, before it is hung up. +pub(crate) fn let_jobs_start(shell: Pid, tty: Option<&Path>) { + if session_groups(shell, tty).iter().any(|&group| group != shell) { + std::thread::sleep(JOB_START_GRACE); + } +} + +/// Sends SIGTERM to `groups` and waits up to [`TERM_GRACE`] for all of them +/// but the shell's own to exit (an interactive shell ignores SIGTERM). +fn terminate(groups: &[Pid], shell: Pid) { + signal_groups(groups, Signal::SIGTERM); + // Stopped jobs act on SIGTERM only once they are continued. + signal_groups(groups, Signal::SIGCONT); + let deadline = Instant::now() + TERM_GRACE; + while Instant::now() < deadline + && groups + .iter() + .any(|&g| g != shell && killpg(g, None).is_ok()) + { + std::thread::sleep(TERM_POLL); + } +} + +fn signal_groups(groups: &[Pid], signal: Signal) { + for &group in groups { + // ESRCH means the group already exited, which is the goal; there is + // nothing more to do for a group that cannot be signalled. + let _ = killpg(group, signal); + } +} + +/// Process groups with a member in the shell's session. Empty when the +/// processes cannot be listed; the caller still signals the groups it knows. +#[cfg(target_os = "linux")] +fn session_groups(shell: Pid, _tty: Option<&Path>) -> Vec { + let Ok(entries) = std::fs::read_dir("/proc") else { + return Vec::new(); + }; + entries + .flatten() + .filter(|e| { + e.file_name() + .to_str() + .is_some_and(|n| n.parse::().is_ok()) + }) + // A process can exit between listing and reading; it is skipped. + .filter_map(|e| std::fs::read_to_string(e.path().join("stat")).ok()) + .filter_map(|stat| proc_stat_group(&stat)) + .filter(|&(_, session)| session == shell) + .map(|(group, _)| group) + .collect() +} + +/// Process group and session from a `/proc//stat` line: +/// `pid (comm) state ppid pgrp session ...`, where comm may contain spaces +/// and parentheses. +#[cfg(target_os = "linux")] +fn proc_stat_group(stat: &str) -> Option<(Pid, Pid)> { + let (_, rest) = stat.rsplit_once(')')?; + let mut fields = rest.split_whitespace().skip(2); + let group = fields.next()?.parse().ok()?; + let session = fields.next()?.parse().ok()?; + Some((Pid::from_raw(group), Pid::from_raw(session))) +} + +/// Process groups with a member on the shell's terminal (macOS `ps` does not +/// report session ids). Empty when the processes cannot be listed; the caller +/// still signals the groups it knows. +#[cfg(target_os = "macos")] +fn session_groups(_shell: Pid, tty: Option<&Path>) -> Vec { + let Some(name) = tty.and_then(Path::file_name) else { + return Vec::new(); + }; + let Ok(out) = std::process::Command::new("/bin/ps") + .arg("-t") + .arg(name) + .args(["-o", "pgid="]) + .output() + else { + return Vec::new(); + }; + String::from_utf8_lossy(&out.stdout) + .split_whitespace() + .filter_map(|group| group.parse().ok()) + .map(Pid::from_raw) + .collect() +} + +/// Other systems: only the shell's and the foreground group are known. +#[cfg(not(any(target_os = "linux", target_os = "macos")))] +fn session_groups(_shell: Pid, _tty: Option<&Path>) -> Vec { + Vec::new() +} + +#[cfg(all(test, target_os = "linux"))] +mod tests { + use super::*; + + #[test] + fn reads_group_and_session_from_proc_stat() { + let stat = "4242 (a (b) c) S 1 4240 4200 34816 4240 4194560 0 0"; + assert_eq!( + proc_stat_group(stat), + Some((Pid::from_raw(4240), Pid::from_raw(4200))) + ); + } +} diff --git a/crates/terminal/src/tests.rs b/crates/terminal/src/tests.rs index 916a8b71..c6a82ee7 100644 --- a/crates/terminal/src/tests.rs +++ b/crates/terminal/src/tests.rs @@ -144,3 +144,76 @@ async fn large_output_is_capped() { assert!(t.starts_with("1\n2\n")); assert!(t.ends_with("200000\n")); } + +/// How long a killed process may take to disappear (reaping is asynchronous +/// once its parent, the shell, is gone). +const PROCESS_GONE_WAIT: Duration = Duration::from_secs(5); + +pub(crate) fn process_exists(pid: i32) -> bool { + nix::sys::signal::kill(nix::unistd::Pid::from_raw(pid), None).is_ok() +} + +pub(crate) async fn wait_until_gone(pid: i32) -> bool { + let deadline = Instant::now() + PROCESS_GONE_WAIT; + while process_exists(pid) { + if Instant::now() > deadline { + return false; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + true +} + +pub(crate) fn parse_pid(r: &ExecResult) -> i32 { + let out = text(r); + out.trim() + .parse() + .unwrap_or_else(|_| panic!("no pid in {out:?}")) +} + +#[tokio::test] +async fn a_command_ignoring_ctrl_c_is_stopped_with_its_shell() { + let mut f = start().await; + let r = f + .session + .exec("sh -c 'trap \"\" INT; echo $$; exec sleep 300'", 1000) + .await + .unwrap(); + let pid = parse_pid(&r); + let r = f.session.exec("echo next", 15_000).await.unwrap(); + assert_eq!(text(&r), "next\n"); + assert!(r.notes[0].contains("did not stop"), "{:?}", r.notes); + assert!(wait_until_gone(pid).await, "pid {pid} survived"); +} + +#[tokio::test] +async fn kill_stops_the_running_foreground_command() { + let mut f = start().await; + let r = f + .session + .exec("sh -c 'echo $$; exec sleep 300'", 1000) + .await + .unwrap(); + assert!(!r.finished); + let pid = parse_pid(&r); + assert!(process_exists(pid)); + f.session.kill(); + assert!(wait_until_gone(pid).await, "pid {pid} survived kill"); +} + +#[tokio::test] +async fn dropping_the_session_stops_background_jobs() { + let mut f = start().await; + // The job ignores the SIGHUP the shell passes on to its jobs, so only + // signalling its own process group stops it. + let r = f + .session + .exec("(trap '' HUP; exec sleep 300) & echo $!", 5000) + .await + .unwrap(); + assert!(r.finished); + let pid = parse_pid(&r); + assert!(process_exists(pid)); + drop(f); + assert!(wait_until_gone(pid).await, "pid {pid} survived drop"); +} diff --git a/docs/HOW-IT-WORKS.md b/docs/HOW-IT-WORKS.md index 12a0b314..fac8314a 100644 --- a/docs/HOW-IT-WORKS.md +++ b/docs/HOW-IT-WORKS.md @@ -41,7 +41,7 @@ A command is not typed into the shell. It is written to a script file, and the s The end marker carries the exit status and the working directory, and because the command itself is sourced from a file, quotes, heredocs and commands that read stdin cannot swallow the marker. Working directory, environment and background jobs carry from one turn to the next, like a person's terminal. -A command that has not finished after `exec_default_yield_ms` (10 seconds by default, `runtime.timeout` in agent.yaml) is left running in the background: the model gets the output so far and a session id, and can wait for more, send input or interrupt it with `write_stdin`. Up to `MAX_SESSIONS` shells exist at once. With `--linger` the shells stay open after the run, so a server the agent started keeps serving whoever needs it next. +A command that has not finished after `exec_default_yield_ms` (10 seconds by default, `runtime.timeout` in agent.yaml) is left running in the background: the model gets the output so far and a session id, and can wait for more, send input or interrupt it with `write_stdin`. Up to `MAX_SESSIONS` shells exist at once. With `--linger` the shells stay open after the run, so a server the agent started keeps serving whoever needs it next. Otherwise, when the run ends, times out or is cancelled (an IPC `abort`, `kill`, closed stdin, or SIGTERM/SIGINT to `gitagent --ipc`), every process group in each shell's session gets SIGTERM and then, after at most a second, SIGKILL, so background jobs and the running command stop with the shell. A background session whose command finished only hangs up its shell, so a server it started with `nohup` keeps running. If that shell has jobs, the hang-up waits 0.2 s first: the shell passes its SIGHUP on to every job, and `nohup` only ignores it once it runs, so `nohup server &` as the last command would otherwise die with the shell. ## 4. The output shaper @@ -142,6 +142,6 @@ A vote from a task raises a lesson's importance only the first time that task vo **Memory** is files in `memory/`, with layers and line limits from `memory/memory.yaml`. The `memory` tool's `save` appends an addition; a save that sends the whole memory back with edits, carrying at least 60% of the existing lines, replaces the file, as gitagent 1.x did; overflow beyond a layer's limit is archived to `memory/archive/`. Each save is a git commit when the agent directory is a repository. -**Hooks** are scripts in `hooks/hooks.yaml` for seven events (session start and end, before and after a tool call, a changed file, before the reply, on error) plus 1.x's `user_prompt_submit`. Each gets JSON on stdin and answers `{"action": "allow" | "block" | "modify", ...}`; a script that prints something else allows, and one that crashes or times out is a warning. A hook marked `compliance: true` or `fail_open: false` fails closed instead: a crash, a timeout or an answer it cannot read blocks the action. The same protocol serves the SDKs' in-process hooks over the IPC stream. +**Hooks** are scripts in `hooks/hooks.yaml` for seven events (session start and end, before and after a tool call, a changed file, before the reply, on error) plus 1.x's `user_prompt_submit`. Each gets JSON on stdin and answers `{"action": "allow" | "block" | "modify", ...}`; a script that prints something else allows, and one that crashes or times out is a warning. A hook marked `compliance: true` or `fail_open: false` fails closed instead: a crash, a timeout or an answer it cannot read blocks the action, and if the hook cannot be loaded at all (a missing script, an unknown event) the agent does not start. A compliance hook cannot set `fail_open: true`. The same protocol serves the SDKs' in-process hooks over the IPC stream. **Approvals** come from `agent.yaml`'s compliance section: human-in-the-loop rules, escalation triggers by tool, kill switch, segregation of duties, spending limits with a monthly ledger in `.gitagent/state.json`. A gated call pauses the run until a person answers, at the terminal, through the SDK's approver, or by a timeout policy.