hive-runtime: kill an ACP agent's whole process group, not just its pid

The ACP agent was spawned with only `kill_on_drop`, which SIGKILLs the
agent's own pid. Commands the agent had started (a `nix build`, a test
run) were reparented and kept running after the agent was killed: when a
subagent run ends after `interrupt`, when an agent ignores
`session/cancel` past the grace, and when the harness drops the runtime.

The agent now leads its own process group (`process_group(0)`), and
dropping its `Connection` SIGKILLs that group. The soft path is unchanged:
`interrupt` / `Canceller::cancel` still sends `session/cancel`.

Refs #4999. The claude half needs hive-claude 0.1.3, which is not on
crates.io yet.

hive-runtime/Cargo.toml

@@ -9,6 +9,7 @@ workspace = true
 
 [dependencies]
 hive-claude.workspace = true
+libc.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 # The ACP agent's provider key, read under the agent's own certificate

hive-runtime/README.md

@@ -40,6 +40,9 @@ with `AcpError::IdleTimeout` (`IdleKilled` if the agent was killed). This is
 the only way a provider error the agent retries without reporting it, such as
 an HTTP 429, ends the turn.
 
+The agent leads its own process group. Killing it, and dropping the runtime,
+`SIGKILL`s that group, so the commands the agent started die with it.
+
 ## ACP backend: compaction
 
 The same `CompactionPolicy` as the claude backend decides when: after a turn

hive-runtime/src/acp/mod.rs

@@ -1062,6 +1062,10 @@ mod tests {
     const AGENT: &str = r#"
 n=0 prompt= model=m/think effort=low
 printf 'start\n' >> "$1.methods"
+# With GRANDCHILD set: a background `sleep`, its pid in <log>.grandchild.
+if [ -n "$GRANDCHILD" ]; then
+  sh -c 'echo $$ >"$1.grandchild"; exec sleep 300' sh "$1" >/dev/null 2>&1 &
+fi
 update() {
   printf '{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"%s","update":%s}}\n' "$1" "$2"
 }
@@ -1430,6 +1434,101 @@ done
         assert_eq!(count(dir.path(), "start"), 2);
     }
 
+    /// Poll `path` until it holds a complete, newline-terminated pid.
+    async fn read_pid(path: &Path) -> i32 {
+        for _ in 0..250 {
+            if let Ok(text) = std::fs::read_to_string(path)
+                && text.ends_with('\n')
+            {
+                return text.trim().parse().expect("pid file holds a pid");
+            }
+            tokio::time::sleep(Duration::from_millis(20)).await;
+        }
+        unreachable!("no pid in {} within 5s", path.display());
+    }
+
+    /// Whether `pid` is still running. A zombie counts as gone: it has
+    /// exited, and `kill(pid, 0)` only keeps finding it until it is reaped.
+    fn is_running(pid: i32) -> bool {
+        // SAFETY: signal 0 delivers nothing; it only checks that `pid` exists.
+        if unsafe { libc::kill(pid, 0) } != 0 {
+            return false;
+        }
+        // Field 3 of /proc/<pid>/stat, right after the parenthesised comm, is
+        // the state letter. Without /proc the process counts as running.
+        !std::fs::read_to_string(format!("/proc/{pid}/stat")).is_ok_and(|stat| {
+            stat.rsplit_once(") ")
+                .is_some_and(|(_, rest)| rest.starts_with('Z'))
+        })
+    }
+
+    /// Require `pid` to be gone within 5s, killing it if it is not.
+    async fn assert_gone(pid: i32) {
+        let mut alive = is_running(pid);
+        for _ in 0..250 {
+            if !alive {
+                break;
+            }
+            tokio::time::sleep(Duration::from_millis(20)).await;
+            alive = is_running(pid);
+        }
+        if alive {
+            // SAFETY: `kill(2)` on a pid this test started; keeps a failed
+            // run from leaving a 300s `sleep` behind.
+            unsafe {
+                libc::kill(pid, libc::SIGKILL);
+            }
+        }
+        assert!(!alive, "grandchild {pid} outlived the agent");
+    }
+
+    #[tokio::test]
+    async fn an_agent_ignoring_a_cancel_is_killed_with_the_commands_it_started() {
+        let dir = tempfile::tempdir().unwrap();
+        let mut runtime = agent(
+            dir.path(),
+            "deaf",
+            &[("GRANDCHILD", "1")],
+            PercentPolicy::default(),
+        );
+        runtime.cancel_grace = Duration::from_millis(200);
+        let (canceller, config) = (runtime.canceller().unwrap(), config(dir.path()));
+
+        let (done, grandchild) = tokio::join!(runtime.run(&config, "one", &NoopSink), async {
+            prompt_sent(dir.path()).await;
+            let grandchild = read_pid(&dir.path().join("prompts.grandchild")).await;
+            assert!(canceller.cancel());
+            grandchild
+        });
+        assert!(
+            matches!(done, Err(Error::Acp(AcpError::CancelIgnored { .. }))),
+            "{done:?}"
+        );
+        assert_gone(grandchild).await;
+    }
+
+    /// A subagent's run drops its runtime when it ends, an interrupted one
+    /// included.
+    #[tokio::test]
+    async fn dropping_the_runtime_kills_the_commands_its_agent_started() {
+        let dir = tempfile::tempdir().unwrap();
+        let runtime = agent(
+            dir.path(),
+            "ok",
+            &[("GRANDCHILD", "1")],
+            PercentPolicy::default(),
+        );
+        runtime
+            .run(&config(dir.path()), "one", &NoopSink)
+            .await
+            .unwrap();
+        let grandchild = read_pid(&dir.path().join("prompts.grandchild")).await;
+        assert!(is_running(grandchild));
+
+        drop(runtime);
+        assert_gone(grandchild).await;
+    }
+
     fn sent(session: &str, text: &str) -> (String, String) {
         (session.to_owned(), text.to_owned())
     }

hive-runtime/src/acp/rpc.rs

@@ -30,8 +30,15 @@ pub(super) enum Incoming {
 type Reply = Result<Value, AcpError>;
 type Pending = Arc<Mutex<HashMap<u64, (&'static str, oneshot::Sender<Reply>)>>>;
 
+/// The agent runs as the leader of its own process group; dropping the
+/// connection `SIGKILL`s that group, so the commands the agent started die
+/// with it. A descendant that left the group (`setsid`, `setpgid`) is not
+/// reached.
 pub(super) struct Connection {
     child: Child,
+    /// The agent's pid, which is its group's id. Read at spawn: a reaped
+    /// child has no id, and the group can outlive its leader.
+    pgid: Option<i32>,
     stdin: Arc<tokio::sync::Mutex<ChildStdin>>,
     next_id: AtomicU64,
     pending: Pending,
@@ -58,12 +65,14 @@ impl Connection {
             .stdin(Stdio::piped())
             .stdout(Stdio::piped())
             .stderr(Stdio::piped())
+            .process_group(0)
             .kill_on_drop(true)
             .spawn()
             .map_err(|source| AcpError::Spawn {
                 program: command.command.clone(),
                 source,
             })?;
+        let pgid = group_id(child.id());
         let (Some(stdin), Some(stdout), Some(stderr)) =
             (child.stdin.take(), child.stdout.take(), child.stderr.take())
         else {
@@ -106,6 +115,7 @@ impl Connection {
 
         Ok(Self {
             child,
+            pgid,
             stdin,
             next_id: AtomicU64::new(1),
             pending,
@@ -159,6 +169,25 @@ impl Connection {
     }
 }
 
+/// The process group a leader with `pid` heads, as `kill(2)` takes it.
+/// Never 0: `kill(-0, …)` signals the caller's own group.
+fn group_id(pid: Option<u32>) -> Option<i32> {
+    pid.and_then(|pid| i32::try_from(pid).ok())
+        .filter(|&pgid| pgid > 0)
+}
+
+impl Drop for Connection {
+    fn drop(&mut self) {
+        if let Some(pgid) = self.pgid {
+            // SAFETY: plain `kill(2)`; `-pgid` targets the agent's own
+            // process group, which `process_group(0)` made it lead.
+            unsafe {
+                libc::kill(-pgid, libc::SIGKILL);
+            }
+        }
+    }
+}
+
 /// The stdout side: routes responses to their waiting request, answers the
 /// agent's own requests, and queues its notifications.
 struct Reader {
@@ -309,7 +338,7 @@ pub(super) fn permission_outcome(
 
 #[cfg(test)]
 mod tests {
-    use super::permission_outcome;
+    use super::{group_id, permission_outcome};
     use crate::acp::{PermissionAsk, PermissionPolicy};
     use serde_json::json;
     use std::sync::Arc;
@@ -406,4 +435,12 @@ mod tests {
             json!({ "outcome": { "outcome": "cancelled" } })
         );
     }
+
+    #[test]
+    fn a_group_id_is_a_positive_pid() {
+        assert_eq!(group_id(Some(42)), Some(42));
+        assert_eq!(group_id(Some(0)), None);
+        assert_eq!(group_id(Some(u32::MAX)), None);
+        assert_eq!(group_id(None), None);
+    }
 }