From b35c983eb047e0a4fd33673b84914fc5eba24c32 Mon Sep 17 00:00:00 2001 From: Parth Sharma Date: Wed, 28 Oct 2026 16:35:48 +0530 Subject: [PATCH 1/3] Fix IPC usage, IPC session clobbering and dropped fail-closed hooks - IPC: turn_end usage and session_end costs come from core's run report, which includes sub-agent model calls (their events are disabled, so the streamed model_text totals left them out). Requests add the report's subagent_turns; a turn without a report (aborted, errored) falls back to the streamed calls. Teardown usage is counted the same way. - IPC: a resumable session (resume: true) takes an exclusive advisory lock on .gitagent/sessions//lock for the process's life. A second process opening the same sessionId gets system/error + session_end and exits 2 instead of loading and later overwriting the other's conversation. - Loader: a hook marked compliance: true or fail_open: false that cannot be loaded (missing or escaping script, bad timeout, unknown event) now fails loading, so the run refuses to start. A compliance hook may not set fail_open: true. Ordinary hooks keep warn-and-skip. --- README.md | 6 +- crates/cli/src/ipc.rs | 171 +++++++++++++++++++++++-------- crates/cli/tests/ipc.rs | 109 ++++++++++++++++++++ crates/loader/src/agent_hooks.rs | 111 +++++++++++++++++--- crates/loader/src/lib.rs | 2 +- docs/HOW-IT-WORKS.md | 2 +- 6 files changed, 339 insertions(+), 62 deletions(-) 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..d5de64dd 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); @@ -772,6 +846,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 +887,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 +1040,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 +1080,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 +1126,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 +1228,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 +1264,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..3be67cfc 100644 --- a/crates/cli/tests/ipc.rs +++ b/crates/cli/tests/ipc.rs @@ -371,6 +371,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/docs/HOW-IT-WORKS.md b/docs/HOW-IT-WORKS.md index 12a0b314..8627e59f 100644 --- a/docs/HOW-IT-WORKS.md +++ b/docs/HOW-IT-WORKS.md @@ -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. From fe7426cc1d3cca4cebbc0a006852d60afedfdf7c Mon Sep 17 00:00:00 2001 From: Parth Sharma Date: Wed, 28 Oct 2026 16:41:12 +0530 Subject: [PATCH 2/3] Stop the shells' commands on cancel and time-up; strip CR from the marker cwd - Terminal: on macOS the end marker line can end in \r\n, so the parsed cwd kept a trailing \r and relative read_file/apply_patch paths failed. The marker line is now trimmed of \r before parsing. - Terminal: Session::kill (end of run, --timeout-s, IPC abort/kill/closed stdin, Terminal drop) only hung up the shell. Job control puts the running command and each background job in its own process group, so they survived with ppid 1. kill now sends SIGTERM (and SIGCONT) to every process group in the shell's session, waits up to 1 s, then SIGKILLs what is left, the shell included. Groups are listed from /proc on Linux (by session id) and with `ps -t ` on macOS, plus the foreground group and the shell's own group. - Terminal: a finished background session and the "did not stop on Ctrl-C" fallback only hang up (SIGTERM/SIGKILL to the foreground command, SIGHUP to the shell), so `nohup` servers started there keep running as before. --linger is unchanged: it never calls kill. - IPC: SIGTERM/SIGINT now drop the session (and its shells) before exiting with 128+signal, instead of dying with the commands still running. --- Cargo.lock | 1 + crates/cli/src/ipc.rs | 28 ++++++- crates/cli/tests/ipc.rs | 57 +++++++++++++ crates/terminal/Cargo.toml | 1 + crates/terminal/src/buffer.rs | 16 +++- crates/terminal/src/lib.rs | 39 ++++++++- crates/terminal/src/manager.rs | 38 ++++++++- crates/terminal/src/stop.rs | 149 +++++++++++++++++++++++++++++++++ crates/terminal/src/tests.rs | 73 ++++++++++++++++ docs/HOW-IT-WORKS.md | 2 +- 10 files changed, 397 insertions(+), 7 deletions(-) create mode 100644 crates/terminal/src/stop.rs 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/crates/cli/src/ipc.rs b/crates/cli/src/ipc.rs index d5de64dd..32450689 100644 --- a/crates/cli/src/ipc.rs +++ b/crates/cli/src/ipc.rs @@ -704,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 @@ -713,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| { diff --git a/crates/cli/tests/ipc.rs b/crates/cli/tests/ipc.rs index 3be67cfc..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(); 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..94ef50cf 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,34 @@ 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. + 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); + match ending { + Ending::Session => { + let tty = shell.master.tty_name(); + stop::stop_session(pid, tty.as_deref(), foreground); + } + Ending::Foreground => stop::stop_foreground(pid, foreground), + } + } + // 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 +449,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..2660645f --- /dev/null +++ b/crates/terminal/src/stop.rs @@ -0,0 +1,149 @@ +//! 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); + +/// 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); +} + +/// 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 8627e59f..06b59a2d 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. ## 4. The output shaper From 89d7cbd51964ba315c99a79e81c2f22de803f2f0 Mon Sep 17 00:00:00 2001 From: Parth Sharma Date: Wed, 28 Oct 2026 16:46:54 +0530 Subject: [PATCH 3/3] Let a finished background session's jobs start before hanging it up The shell passes its SIGHUP on to every job, and nohup ignores SIGHUP only once it runs. A background session whose last command was 'nohup server &' was hung up the moment its end marker arrived, so the SIGHUP reached the job while it was still being forked and killed it. hang_up now waits 200 ms first when the shell has jobs (stop::let_jobs_start), so nohup and disowned jobs survive; plain jobs still die with the shell. kill/drop are unchanged. --- crates/terminal/src/lib.rs | 12 +++++++----- crates/terminal/src/stop.rs | 13 +++++++++++++ docs/HOW-IT-WORKS.md | 2 +- 3 files changed, 21 insertions(+), 6 deletions(-) diff --git a/crates/terminal/src/lib.rs b/crates/terminal/src/lib.rs index 94ef50cf..c9c21aa3 100644 --- a/crates/terminal/src/lib.rs +++ b/crates/terminal/src/lib.rs @@ -305,7 +305,8 @@ impl 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. + /// 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); } @@ -315,12 +316,13 @@ impl Session { 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 => { - let tty = shell.master.tty_name(); - stop::stop_session(pid, tty.as_deref(), foreground); + 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()); } - Ending::Foreground => stop::stop_foreground(pid, foreground), } } // SIGHUP, then SIGKILL, to the shell. Errors only mean it already diff --git a/crates/terminal/src/stop.rs b/crates/terminal/src/stop.rs index 2660645f..7f57d50b 100644 --- a/crates/terminal/src/stop.rs +++ b/crates/terminal/src/stop.rs @@ -16,6 +16,12 @@ use nix::unistd::Pid; 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 @@ -47,6 +53,13 @@ pub(crate) fn stop_foreground(shell: Pid, foreground: Option) { 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) { diff --git a/docs/HOW-IT-WORKS.md b/docs/HOW-IT-WORKS.md index 06b59a2d..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. 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. +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