From 466e32b69fb066eebe5d0e38526c808236ace734 Mon Sep 17 00:00:00 2001 From: atlas Date: Sat, 10 Oct 2026 20:00:11 +0200 Subject: [PATCH 1/9] hive-c0re: close build-log rows a dead process left in flight A build_logs row is opened with finished_at NULL and closed only by the process that runs the build. If hive-c0re stopped in between, the row stayed open for good: the dashboard listed it with no status, its log stream never sent `done`, and the retention sweep, which skips open rows, never deleted it. Coordinator::open now closes every open row as `fail` and appends a line saying hive-c0re stopped. It runs before this process starts any build, and only there: BuildLogs::open itself leaves rows alone so a second opener can't close another process's live rows. --- docs/agent-lifecycle/persistence.md | 5 +++ hive-c0re/src/coordinator.rs | 41 +++++++++++++++++++++++ hive-c0re/src/stores/build_logs.rs | 52 +++++++++++++++++++++++++++++ 3 files changed, 98 insertions(+) diff --git a/docs/agent-lifecycle/persistence.md b/docs/agent-lifecycle/persistence.md index d579fdd1..ddddcf7c 100644 --- a/docs/agent-lifecycle/persistence.md +++ b/docs/agent-lifecycle/persistence.md @@ -209,6 +209,11 @@ Writes are best-effort: `append_stdout` / `append_stderr` / `finish` log a warning on sqlite error and let the build continue. A failed log row never blocks a rebuild. +A row is open (`finished_at IS NULL`) only while its child runs. At startup +`Coordinator::open` closes every row still open as `fail` and appends a +line saying hive-c0re stopped: the previous process died mid-build, and +the retention sweep never deletes open rows. + ### `/harness/hyperhive-model` (per agent) Single-line text file holding the claude model name currently diff --git a/hive-c0re/src/coordinator.rs b/hive-c0re/src/coordinator.rs index d06f5969..9e70ad3d 100644 --- a/hive-c0re/src/coordinator.rs +++ b/hive-c0re/src/coordinator.rs @@ -462,6 +462,15 @@ impl Coordinator { let build_logs = Arc::new( crate::build_logs::BuildLogs::open(build_logs_dir).context("open build_logs")?, ); + // This process is the database's only writer and has started no + // build yet, so a row still in-flight belongs to a dead process. + let reaped = build_logs.reap_orphans()?; + if reaped > 0 { + tracing::warn!( + reaped, + "build_logs: closed rows left in-flight by a previous run" + ); + } // Install the process-wide handle so `lifecycle::run` / // `lifecycle::prebuild_toplevel` can write without us having // to thread an `Arc` through every public entry @@ -1376,3 +1385,35 @@ mod agent_paths_tests { assert_eq!(paths.agent, std::path::PathBuf::from("/run/x")); } } + +#[cfg(test)] +mod open_tests { + use super::{Coordinator, HiveEnv}; + + fn open(db: &std::path::Path) -> Coordinator { + Coordinator::open( + db, + HiveEnv::default(), + crate::hive_stats::PriceTable::new(), + 1, + ) + .expect("open coordinator") + } + + /// A build log row the previous process left open is closed as failed + /// when the next one opens the coordinator. + #[tokio::test] + async fn open_closes_build_logs_left_in_flight() { + let dir = tempfile::tempdir().expect("tempdir"); + let db = dir.path().join("broker.sqlite"); + let id = open(&db) + .build_logs + .start("alice", "run", "cmd", None) + .expect("start"); + + let coord = open(&db); + let row = coord.build_logs.get_full(id).expect("get").expect("Some"); + assert_eq!(row.header.status.as_deref(), Some("fail")); + assert!(row.header.finished_at.is_some()); + } +} diff --git a/hive-c0re/src/stores/build_logs.rs b/hive-c0re/src/stores/build_logs.rs index b7ca43b9..6af05a71 100644 --- a/hive-c0re/src/stores/build_logs.rs +++ b/hive-c0re/src/stores/build_logs.rs @@ -48,6 +48,9 @@ const KEEP_FAIL_SECS: i64 = 30 * 24 * 3600; /// `finished_at IS NULL`). const KEEP_OK_SECS: i64 = 24 * 3600; +/// Last line of a row `reap_orphans` closed. +const ORPHAN_NOTE: &str = "hive-c0re stopped before this build finished"; + const SCHEMA: &str = " CREATE TABLE IF NOT EXISTS build_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, @@ -175,6 +178,30 @@ impl BuildLogs { }) } + /// Close every row still marked in-flight as failed. Returns the number + /// closed. + /// + /// Only sound when no other writer holds this database: the caller must + /// be the one process that owns it, before it starts any build. A row + /// with `finished_at IS NULL` then belongs to a process that died, and + /// nothing will ever finish it. Kept out of `open` so a second opener + /// can't close another process's live rows. + pub fn reap_orphans(&self) -> Result { + let now = Utc::now().timestamp(); + let conn = self.conn.lock().unwrap(); + let closed = conn + .execute( + "UPDATE build_logs + SET finished_at = ?1, + status = 'fail', + stderr = stderr || ?2 || x'0a' + WHERE finished_at IS NULL", + params![now, ORPHAN_NOTE], + ) + .context("close orphaned build_logs rows")?; + Ok(u64::try_from(closed).unwrap_or(0)) + } + /// Subscribe to per-build append/finish notifications. Each emitted /// value is the `id` of the row that changed. The SSE stream handler /// calls this once and filters for its target id. @@ -522,6 +549,31 @@ mod tests { assert!(full.stderr.ends_with('\n')); } + #[test] + fn reap_orphans_fails_rows_left_in_flight_by_a_previous_process() { + let dir = tempfile::tempdir().expect("tempdir"); + let (orphan, done) = { + let db = BuildLogs::open(dir.path()).expect("open"); + let orphan = db.start("alice", "run", "cmd one", None).expect("start"); + db.append_stdout(orphan, "building"); + let done = db.start("bob", "run", "cmd two", None).expect("start"); + db.finish(done, BuildStatus::Ok); + (orphan, done) + }; + + let db = BuildLogs::open(dir.path()).expect("reopen"); + assert_eq!(db.reap_orphans().expect("reap"), 1); + + let row = db.get_full(orphan).expect("get").expect("Some"); + assert_eq!(row.header.status.as_deref(), Some("fail")); + assert!(row.header.finished_at.is_some()); + assert!(row.stdout.contains("building")); + assert!(row.stderr.ends_with(&format!("{ORPHAN_NOTE}\n"))); + let ok = db.get_full(done).expect("get").expect("Some"); + assert_eq!(ok.header.status.as_deref(), Some("ok")); + assert_eq!(db.reap_orphans().expect("reap again"), 0); + } + #[test] fn list_recent_orders_newest_first_and_clamps() { let (_d, db) = tmpdb(); From 63ab0bbd210cfc61e8416b5e5c0e16eaac51add2 Mon Sep 17 00:00:00 2001 From: atlas Date: Sat, 10 Oct 2026 20:00:14 +0200 Subject: [PATCH 2/9] hive-c0re: clear O_NONBLOCK on the snapshot-store socket before btrfs send gets it The tokio socket stays non-blocking after into_std, and the flag is shared with hive-priv through SCM_RIGHTS, so btrfs send wrote into a non-blocking socket. --- hive-c0re/src/snapshot_push.rs | 48 ++++++++++++++++++++++++++++++---- 1 file changed, 43 insertions(+), 5 deletions(-) diff --git a/hive-c0re/src/snapshot_push.rs b/hive-c0re/src/snapshot_push.rs index 1cd65905..3098c6d3 100644 --- a/hive-c0re/src/snapshot_push.rs +++ b/hive-c0re/src/snapshot_push.rs @@ -18,6 +18,8 @@ //! stores would split that chain in two, which is the case the store //! exists to serve. +use std::os::fd::OwnedFd; + use anyhow::{Context as _, Result, bail}; use hive_priv_sock::PrivRequest; use tokio::io::AsyncWriteExt as _; @@ -61,10 +63,7 @@ pub async fn push_agent_snapshot(agent: &str, snapshot: &str, parent: Option<&st // closes our copy as soon as the kernel has it, so the receiver sees // EOF when `btrfs send` finishes rather than hanging on a descriptor // this process is still holding. - let fd = sock - .into_std() - .context("detach snapshot-store socket for hand-off")? - .into(); + let fd = detach_blocking(sock)?; let resp = priv_client::call_with_fd( &PrivRequest::SendAgentSnapshotToFd { @@ -93,6 +92,22 @@ pub async fn push_agent_snapshot(agent: &str, snapshot: &str, parent: Option<&st Ok(()) } +/// Turn a tokio socket into a descriptor safe to hand to `btrfs send`. +/// +/// tokio sockets are `O_NONBLOCK`, and `into_std` does not clear it. The +/// flag lives on the open file description, which `hive-priv` shares with +/// us through `SCM_RIGHTS`, so `btrfs send` would write to a +/// non-blocking socket and get `EAGAIN` whenever the send buffer is full. +fn detach_blocking(sock: TcpStream) -> Result { + let std_sock = sock + .into_std() + .context("detach snapshot-store socket for hand-off")?; + std_sock + .set_nonblocking(false) + .context("switch snapshot-store socket to blocking mode for hand-off")?; + Ok(std_sock.into()) +} + /// Require a configured store address. /// /// Split out of [`push_agent_snapshot`] so the "not configured" path is @@ -118,7 +133,30 @@ fn require_store_addr(configured: Option<&str>) -> Result<&str> { #[cfg(test)] mod tests { - use super::require_store_addr; + use std::os::fd::AsRawFd as _; + + use tokio::net::{TcpListener, TcpStream}; + + use super::{detach_blocking, require_store_addr}; + + #[tokio::test] + async fn the_handed_off_descriptor_is_blocking() { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let sock = TcpStream::connect(listener.local_addr().expect("addr")) + .await + .expect("connect"); + + let fd = detach_blocking(sock).expect("detach"); + + // SAFETY: `fd` is a live descriptor owned by this test. + let flags = unsafe { libc::fcntl(fd.as_raw_fd(), libc::F_GETFL) }; + assert!(flags >= 0, "F_GETFL failed"); + assert_eq!( + flags & libc::O_NONBLOCK, + 0, + "btrfs send would get EAGAIN on a full socket buffer" + ); + } #[test] fn a_configured_address_passes_through() { From 028c283ff972d43cb57c3912ce1c8b38ef5ef16b Mon Sep 17 00:00:00 2001 From: atlas Date: Sat, 10 Oct 2026 13:44:10 +0200 Subject: [PATCH 3/9] claude runtime: interrupting a session kills every process it started hive-claude 0.1.3 spawns claude as the leader of its own process group, and its `Cancel::cancel` and idle watchdog signal that group instead of claude's pid alone. Commands claude started (a `nix build`, a test run) used to survive a subagent `interrupt` and keep running. Bump hive-claude to 0.1.3. The subagent `interrupt` call site needs no code change: `Cancel::cancel(force)` kept its signature. The harness's `/cancel` and `/logout` did not go through hive-claude: they ran `kill -INT ` on the claude child they found in /proc. They now SIGINT the group that child leads, through `libc::kill`. A `claude` child that leads no group (the `claude plugin` calls) shares the harness's group and is still signalled alone. Tests: an interrupted subagent's grandchild dies, for both SIGINT and `force`, and fails on hive-claude 0.1.2; a harness cancel stops a turn's grandchild, and fails when only the pid is signalled. --- Cargo.lock | 4 +- Cargo.toml | 2 +- docs/tools/subagent.md | 5 ++ docs/web-ui/agent.md | 14 +-- hive-agent/src/web_ui/actions.rs | 2 +- hive-agent/src/web_ui/auth.rs | 2 +- hive-agent/src/web_ui/mod.rs | 145 ++++++++++++++++++++++++++++--- hive-subagent-mcp/src/session.rs | 94 ++++++++++++++++++++ 8 files changed, 244 insertions(+), 24 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index fef61d4b..e8c42a52 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1835,9 +1835,9 @@ dependencies = [ [[package]] name = "hive-claude" -version = "0.1.2" +version = "0.1.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1e43e06fd09a7b6d6cca3e70f06d00fa795ceddfe1555121ca0cf4fc578eb9c3" +checksum = "8c87ac627823f7bfc38478d7494e614e3f680dda6007dbcd306af51d3eed3bb3" dependencies = [ "libc", "serde", diff --git a/Cargo.toml b/Cargo.toml index 082db1a5..ab3840c7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -97,7 +97,7 @@ hive-jobq-metrics = { path = "hive-jobq-metrics" } hive-jobq-wire = { path = "hive-jobq-wire" } hive-log = { path = "hive-log" } hive-core-agent-sock = { path = "hive-core-agent-sock" } -hive-claude = "0.1.2" +hive-claude = "0.1.3" hive-host-sock = { path = "hive-host-sock" } hive-priv-sock = { path = "hive-priv-sock" } hive-runtime = { path = "hive-runtime" } diff --git a/docs/tools/subagent.md b/docs/tools/subagent.md index 12ef7381..0d59ec8c 100644 --- a/docs/tools/subagent.md +++ b/docs/tools/subagent.md @@ -166,6 +166,11 @@ from how the child exited. `continue` clears it and hands the session a fresh turn allowance, which is what makes an interrupt a pause you can undo rather than a session you have to abandon. +On `claude`, the signal goes to claude's whole process group, so the +commands the subagent started (a `nix build`, a test run) get it too. The +signal doesn't reach a command that moved itself out of the group +(`setsid`). + When the session knows where its report goes — `start`'s `report_file`, or the path the subagent names when it signals — the daemon appends the stop reason to that file as well, so the artifact you were going to read diff --git a/docs/web-ui/agent.md b/docs/web-ui/agent.md index 40cfcec7..ea836349 100644 --- a/docs/web-ui/agent.md +++ b/docs/web-ui/agent.md @@ -227,11 +227,11 @@ Slash commands today: - `/help` — list commands locally. - `/clear` — wipe the local terminal view (server history kept). -- `/cancel` — `POST /api/cancel` → host shellouts `pkill -INT - claude`, emits a Note. On an ACP agent it sends the agent - `session/cancel` instead, and kills it if the turn hasn't ended 10s - later. Also surfaces as a `■ cancel turn` button in the state row - while state=thinking. +- `/cancel` — `POST /api/cancel` → SIGINT to the turn's claude and every + command it started (its process group), emits a Note. On an ACP agent + it sends the agent `session/cancel` instead, and kills it if the turn + hasn't ended 10s later. Also surfaces as a `■ cancel turn` button in + the state row while state=thinking. - `/compact` — `POST /api/compact` → sets the deferred `Bus::request_compact()` flag and returns immediately. The harness runs the `/compact` at the next turn boundary (end of the in-flight @@ -272,8 +272,8 @@ shaped). - `POST /login/{start,code,cancel}` — claude OAuth login flow. Start/cancel emit `LiveEvent::StatusChanged` to flip the badge to/from `needs_login_in_progress`. -- `POST /api/cancel` — SIGINT the in-flight claude turn. Emits a - `LiveEvent::Note`. +- `POST /api/cancel` — SIGINT the in-flight claude turn and the + commands it started. Emits a `LiveEvent::Note`. - `POST /api/compact` — run `/compact` on the persistent session (same MCP config + system prompt + allowed tools as a normal turn — only the stdin payload differs). Flips state to diff --git a/hive-agent/src/web_ui/actions.rs b/hive-agent/src/web_ui/actions.rs index 21d2243f..ebd3d8bc 100644 --- a/hive-agent/src/web_ui/actions.rs +++ b/hive-agent/src/web_ui/actions.rs @@ -69,7 +69,7 @@ pub(super) async fn post_cancel_turn(State(state): State) -> Response .emit(crate::events::LiveEvent::Note { text: note.into() }); return (axum::http::StatusCode::OK, "ok").into_response(); } - let out = super::sigint_claude().await; + let out = super::sigint_claude(); let note = match out { SigintOutcome::Signalled => { // A process actually got signalled — the *next* turn's wake diff --git a/hive-agent/src/web_ui/auth.rs b/hive-agent/src/web_ui/auth.rs index ea02fb3a..bbc664b3 100644 --- a/hive-agent/src/web_ui/auth.rs +++ b/hive-agent/src/web_ui/auth.rs @@ -84,7 +84,7 @@ pub(super) async fn post_login_cancel(State(state): State) -> Response /// preservation invariants. pub(super) async fn post_logout(State(state): State) -> Response { // Step 1: SIGINT claude (best-effort, matches `post_cancel_turn`). - let _ = super::sigint_claude().await; + let _ = super::sigint_claude(); // Step 2: delete OAuth credential files only — login::clear_session owns // the file set and preserves session-history files alongside them. let dir = crate::paths::claude_dir(); diff --git a/hive-agent/src/web_ui/mod.rs b/hive-agent/src/web_ui/mod.rs index 88d848e6..2b8a1c0b 100644 --- a/hive-agent/src/web_ui/mod.rs +++ b/hive-agent/src/web_ui/mod.rs @@ -340,21 +340,40 @@ fn find_claude_child() -> Option { /// (best-effort). See [`find_claude_child`] for how it's identified — /// **not** a name-based `pkill claude`, which would also hit an unrelated /// `claude` process someone is running interactively in the container. -async fn sigint_claude() -> SigintOutcome { +fn sigint_claude() -> SigintOutcome { let Some(pid) = find_claude_child() else { return SigintOutcome::NoProcess; }; - match tokio::process::Command::new("kill") - .args(["-INT", &pid.to_string()]) - .status() - .await - { - Ok(status) if status.success() => SigintOutcome::Signalled, - // Non-zero from `kill` means "no such process" (ESRCH) — it already - // exited between the check above and the signal. Same as never - // having found it. - Ok(_) => SigintOutcome::NoProcess, - Err(e) => SigintOutcome::Failed(e), + sigint_turn(pid) +} + +/// SIGINT `pid`, and its whole process group when it leads one, so the +/// commands a turn's claude started stop with it. `hive-claude` spawns the +/// turn's claude as the leader of its own group. A `claude` child that +/// leads none (`plugins.rs`'s `claude plugin` calls) shares the harness's +/// group, so it is signalled alone. +fn sigint_turn(pid: u32) -> SigintOutcome { + // A child is never pid 0 or 1, and `-0` / `-1` would reach the + // harness's own group or every process it may signal. + let Some(pid) = i32::try_from(pid).ok().filter(|&pid| pid > 1) else { + return SigintOutcome::NoProcess; + }; + // SAFETY: `getpgid(2)` only reads; a gone pid returns -1, never `pid`. + let target = if unsafe { libc::getpgid(pid) } == pid { + -pid + } else { + pid + }; + // SAFETY: plain `kill(2)` on the child or the group it leads. + if unsafe { libc::kill(target, libc::SIGINT) } == 0 { + return SigintOutcome::Signalled; + } + let err = std::io::Error::last_os_error(); + if err.raw_os_error() == Some(libc::ESRCH) { + // It exited between the scan and the signal: same as never found. + SigintOutcome::NoProcess + } else { + SigintOutcome::Failed(err) } } @@ -419,3 +438,105 @@ fn broker_error_response(err: &BrokerError, action: &str) -> Response { ), } } + +#[cfg(test)] +mod tests { + use std::os::unix::process::CommandExt as _; + use std::path::Path; + use std::process::{Command, Stdio}; + use std::time::Duration; + + use super::{SigintOutcome, sigint_turn}; + + /// Poll `path` until it holds a complete, newline-terminated pid. + 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"); + } + std::thread::sleep(Duration::from_millis(20)); + } + 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; + } + !std::fs::read_to_string(format!("/proc/{pid}/stat")).is_ok_and(|stat| { + stat.rsplit_once(") ") + .is_some_and(|(_, rest)| rest.starts_with('Z')) + }) + } + + #[test] + fn a_cancel_stops_the_commands_a_turn_started() { + let dir = tempfile::tempdir().expect("tempdir"); + let pidfile = dir.path().join("grandchild.pid"); + // A turn's claude as hive-claude spawns it: leading its own group, + // with a command of its own running in the foreground (a + // non-interactive shell starts `&` jobs with SIGINT ignored). The + // trailing `exit` keeps `sh -c` from exec'ing into that command. + let mut turn = Command::new("/bin/sh") + .args([ + "-c", + r#"sh -c 'echo $$ >"$1"; exec sleep 300' sh "$1"; exit"#, + "sh", + ]) + .arg(&pidfile) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .process_group(0) + .spawn() + .expect("spawn the stand-in turn"); + let grandchild = read_pid(&pidfile); + assert_ne!( + i32::try_from(turn.id()).ok(), + Some(grandchild), + "the command must be a child of the turn, not the turn itself" + ); + + assert!(matches!(sigint_turn(turn.id()), SigintOutcome::Signalled)); + + let mut alive = is_running(grandchild); + for _ in 0..250 { + if !alive { + break; + } + std::thread::sleep(Duration::from_millis(20)); + alive = is_running(grandchild); + } + if alive { + // SAFETY: `kill(2)` on a pid this test started; keeps a failed + // run from leaving a 300s `sleep` behind. + unsafe { + libc::kill(grandchild, libc::SIGKILL); + } + } + let _ = turn.wait(); + assert!(!alive, "grandchild {grandchild} outlived the cancel"); + } + + #[test] + fn a_claude_sharing_the_harness_group_is_signalled_alone() { + // In this test process's own group, as `claude plugin` runs. A group + // signal here would SIGINT the test binary itself. + let mut child = Command::new("sleep") + .arg("300") + .spawn() + .expect("spawn sleep"); + + assert!(matches!(sigint_turn(child.id()), SigintOutcome::Signalled)); + + let status = child.wait().expect("reap sleep"); + assert_eq!( + std::os::unix::process::ExitStatusExt::signal(&status), + Some(libc::SIGINT) + ); + } +} diff --git a/hive-subagent-mcp/src/session.rs b/hive-subagent-mcp/src/session.rs index 75031cd1..8411e45a 100644 --- a/hive-subagent-mcp/src/session.rs +++ b/hive-subagent-mcp/src/session.rs @@ -3731,6 +3731,100 @@ mod tests { std::fs::remove_dir_all(&dir).ok(); } + /// A stand-in for claude with a command of its own still running: an + /// inner `sh` writes its pid to `pidfile` and execs `sleep`. It runs in + /// the foreground because a non-interactive shell starts `&` jobs with + /// SIGINT ignored, and its output goes to /dev/null so it cannot hold the + /// driver's pipes open. + fn busy_claude(dir: &Path, pidfile: &Path) -> PathBuf { + let script = dir.join("busy-claude"); + write_executable( + &script, + &format!( + "#!/bin/sh\nsh -c 'echo $$ >\"$1\"; exec sleep 300' sh '{}' >/dev/null 2>&1\n", + pidfile.display() + ), + ); + script + } + + /// 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; + } + !std::fs::read_to_string(format!("/proc/{pid}/stat")).is_ok_and(|stat| { + stat.rsplit_once(") ") + .is_some_and(|(_, rest)| rest.starts_with('Z')) + }) + } + + #[tokio::test] + async fn an_interrupt_stops_the_commands_the_subagent_started() { + for force in [false, true] { + let dir = std::env::temp_dir().join(format!( + "hive-subagent-group-{}-{:?}-{force}", + std::process::id(), + std::thread::current().id() + )); + std::fs::create_dir_all(&dir).expect("scratch dir"); + let pidfile = dir.join("grandchild.pid"); + let state = Arc::new(mid_run("n", "carry on indefinitely", 5)); + let config = Config { + program: Some(busy_claude(&dir, &pidfile)), + ..Config::default() + }; + spawn_and_track( + &state, + "n", + &config, + &Attach::Create("n".to_owned()), + "turn one".to_owned(), + None, + ) + .expect("the fake claude spawns"); + let grandchild = read_pid(&pidfile).await; + + interrupt(&state, "n", force).expect("a running turn is interruptible"); + + let mut alive = is_running(grandchild); + for _ in 0..250 { + if !alive { + break; + } + tokio::time::sleep(Duration::from_millis(20)).await; + alive = is_running(grandchild); + } + if alive { + // SAFETY: `kill(2)` on a pid this test started; keeps a failed + // run from leaving a 300s `sleep` behind. + unsafe { + libc::kill(grandchild, libc::SIGKILL); + } + } + std::fs::remove_dir_all(&dir).ok(); + assert!( + !alive, + "grandchild {grandchild} outlived interrupt(force = {force})" + ); + } + } + #[test] fn a_cancelled_run_stops_with_turns_still_on_its_clock() { // The loop half of the same fact, asserted directly: `Cancelled` is From 43e69d345eeef851d4a5e68512443c4a4d6190aa Mon Sep 17 00:00:00 2001 From: atlas Date: Sat, 10 Oct 2026 19:52:25 +0200 Subject: [PATCH 4/9] subagents, ACP: stop with SIGINT only, so claude ends its own tool commands claude's Bash tool starts every command in a new session, so a signal to claude's process group never reaches it. claude stops those commands itself when SIGINT makes it shut down. A SIGKILL leaves them running, re-parented to init. Per the ruling on #4999, every stop is SIGINT and nothing escalates. - Subagent `interrupt` always sends SIGINT. The `force` argument is gone from the MCP tool and from `session::interrupt`. A caller that still sends `force: true` gets SIGINT; the argument is ignored. - `interrupt` used to free the session name before it signalled, so a `continue` could start a second claude on the session while the first was still exiting. The `running` map now holds a three-state `Slot` (starting, running, stopping). An interrupted claude turn is `Stopping` until the turn task's own `finish_turn` clears it. `status` says "stopping", and `start`/`continue` on that name are refused. `interrupt` still returns at once. The ACP arm is unchanged: its name is freed when `session/cancel` is sent. - ACP `Connection` drop sends its process group SIGINT, not SIGKILL, and the spawn no longer sets tokio's `kill_on_drop`, which SIGKILLed the leader after the `Drop` body. The test fixture's background grandchild resets SIGINT to default, since `&` starts it ignored. Tests: a fake claude that records SIGINT, interrupted, fails with `cancel(true)` (the old `force: true` path). An interrupted fake claude that keeps running keeps its name claimed and `continue` is refused, which fails when the slot is freed before the signal. A fake ACP agent that records SIGINT, dropped, fails with SIGKILL in `Drop`, and also with SIGINT in `Drop` but `kill_on_drop(true)` left in place. Not changed: hive-claude's idle watchdog (still SIGKILL, separate hive-claude change), the units' `KillSignal`, `machinectl kill`, and the login helper. --- .../base/skills/claude-subagents/SKILL.md | 5 +- docs/tools/subagent.md | 22 +- hive-runtime/README.md | 10 +- hive-runtime/src/acp/mod.rs | 46 +- hive-runtime/src/acp/rpc.rs | 10 +- hive-subagent-mcp/src/mcp.rs | 12 +- hive-subagent-mcp/src/session.rs | 405 ++++++++++++------ 7 files changed, 340 insertions(+), 170 deletions(-) diff --git a/claude-plugins/plugins/base/skills/claude-subagents/SKILL.md b/claude-plugins/plugins/base/skills/claude-subagents/SKILL.md index dc53f709..d7f37b3d 100644 --- a/claude-plugins/plugins/base/skills/claude-subagents/SKILL.md +++ b/claude-plugins/plugins/base/skills/claude-subagents/SKILL.md @@ -27,7 +27,7 @@ skill's recipe for you: start(name, prompt_file, model?, effort?, trigger?) continue(name, prompt, model?, effort?) status(name) -interrupt(name, force?) +interrupt(name) ``` `start` and `continue` return once the turn is under way, not once it @@ -42,7 +42,8 @@ reports how long since that turn last produced output, so a few seconds means it's working and an age climbing into the minutes means it's wedged and worth an `interrupt`. Reach for `continue` only once you actually have a new instruction for it, since that spends a turn. -`interrupt` genuinely stops a running turn (`force: true` for SIGKILL). +`interrupt` genuinely stops a running turn, with SIGINT so claude stops +the commands it started; `status` says `stopping` until claude has exited. Model choice, prompt hygiene, splitting big batches, verify-then-report — everything else in this skill — applies exactly the same whether you're diff --git a/docs/tools/subagent.md b/docs/tools/subagent.md index 0d59ec8c..b4c76da9 100644 --- a/docs/tools/subagent.md +++ b/docs/tools/subagent.md @@ -166,10 +166,15 @@ from how the child exited. `continue` clears it and hands the session a fresh turn allowance, which is what makes an interrupt a pause you can undo rather than a session you have to abandon. -On `claude`, the signal goes to claude's whole process group, so the -commands the subagent started (a `nix build`, a test run) get it too. The -signal doesn't reach a command that moved itself out of the group -(`setsid`). +On `claude`, the signal is always SIGINT, and it goes to claude's whole +process group. The signal doesn't reach a command that moved itself out of +the group (`setsid`), and claude's Bash tool starts every command that way: +claude stops those commands itself when it shuts down on SIGINT. A SIGKILL +would leave them running, so nothing escalates to one. `interrupt` returns +at once, and the name stays claimed until claude has exited: `status` says +**stopping**, and the daemon refuses `start` and `continue` on that name +until then. A claude that ignores the SIGINT stays stopping until someone kills +it by hand or the unit stops. When the session knows where its report goes — `start`'s `report_file`, or the path the subagent names when it signals — the daemon appends the @@ -345,7 +350,7 @@ page applies as written. On `acp`: - each run spawns its own copy of the parent's ACP agent. The end of the - run kills it and every command it started. The unit loads + run sends SIGINT to it and every command in its process group. The unit loads `backendEnvironmentFile` for it. On the `opencode` preset, the daemon reads a provider key that file leaves unset from the swarm secret store at `swarm/agents//acp-provider`, as the harness does. @@ -354,9 +359,10 @@ On `acp`: agent process, and a `continue` with none recorded fails straight away. - a role goes in front of the session's first prompt, since ACP has no system prompt. -- `interrupt` sends `session/cancel`, and `force` changes nothing. The - runtime kills an agent that ignores the cancel for 10s, with every - command it started. +- `interrupt` sends `session/cancel`. The runtime sends SIGINT to the + process group of an agent that ignores the cancel for 10s, and when the + run ends. Nothing escalates past that SIGINT. The name is free as soon as + `interrupt` sends the cancel. - the runtime sets `model` and `effort` on the session when the agent offers that value for its `model` or `thought_level` config option. Any other value does nothing (`hive_runtime::AcpRuntime`). diff --git a/hive-runtime/README.md b/hive-runtime/README.md index 11183df4..6b6d4025 100644 --- a/hive-runtime/README.md +++ b/hive-runtime/README.md @@ -35,13 +35,15 @@ The crate depends on no hyperhive binary crate, so `hive-agent` and A turn is stopped with `session/cancel`: by the `Canceller` handle, or by the idle watchdog once no `session/update` has arrived for `Config::idle_timeout`. An agent that has not answered the prompt 10s later -is killed, and the next turn respawns it. A watchdog stop fails the turn -with `AcpError::IdleTimeout` (`IdleKilled` if the agent was killed). This is +is sent SIGINT, and the next turn respawns it. A watchdog stop fails the turn +with `AcpError::IdleTimeout` (`IdleKilled` if the agent was sent SIGINT). 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. +The agent leads its own process group. Stopping it, and dropping the runtime, +sends that group SIGINT, so the commands the agent started get it too. +Nothing escalates to SIGKILL: an agent that ignores SIGINT keeps running, +and the next turn starts a new one beside it. ## ACP backend: compaction diff --git a/hive-runtime/src/acp/mod.rs b/hive-runtime/src/acp/mod.rs index b93be539..64afa350 100644 --- a/hive-runtime/src/acp/mod.rs +++ b/hive-runtime/src/acp/mod.rs @@ -39,7 +39,7 @@ const SETTLE_QUIET: Duration = Duration::from_millis(500); const SETTLE_MAX: Duration = Duration::from_secs(5); /// How long the agent has to end its turn after `session/cancel` before it is -/// killed. +/// sent SIGINT. const CANCEL_GRACE: Duration = Duration::from_secs(10); /// How long a `compact` command may send nothing before it is cancelled and @@ -120,10 +120,12 @@ pub enum AcpError { EmptyEndTurn, #[error( "ACP agent sent nothing for {silent_secs}s and did not stop within {grace_secs}s of \ - `session/cancel`, so it was killed" + `session/cancel`, so it was sent SIGINT" )] IdleKilled { silent_secs: u64, grace_secs: u64 }, - #[error("ACP agent did not stop within {grace_secs}s of `session/cancel`, so it was killed")] + #[error( + "ACP agent did not stop within {grace_secs}s of `session/cancel`, so it was sent SIGINT" + )] CancelIgnored { grace_secs: u64 }, } @@ -141,8 +143,8 @@ struct CancelState { impl Canceller { /// Ask the turn in flight to stop. The agent is sent `session/cancel`, /// and the turn then ends normally with what it produced so far; an - /// agent that has not ended it within `CANCEL_GRACE` is killed, and the - /// turn fails with [`AcpError::CancelIgnored`]. Returns whether a turn + /// agent that has not ended it within `CANCEL_GRACE` is sent SIGINT, and + /// the turn fails with [`AcpError::CancelIgnored`]. Returns whether a turn /// was in flight; with none, this does nothing. #[must_use] pub fn cancel(&self) -> bool { @@ -1062,9 +1064,14 @@ mod tests { const AGENT: &str = r#" n=0 prompt= model=m/think effort=low printf 'start\n' >> "$1.methods" +# With TRAP_INT set: SIGINT creates .int and exits 0. +if [ -n "$TRAP_INT" ]; then + trap ': > "$1.int"; exit 0' INT +fi # With GRANDCHILD set: a background `sleep`, its pid in .grandchild. +# `&` starts it with SIGINT ignored; `env` puts the default back. if [ -n "$GRANDCHILD" ]; then - sh -c 'echo $$ >"$1.grandchild"; exec sleep 300' sh "$1" >/dev/null 2>&1 & + sh -c 'echo $$ >"$1.grandchild"; exec env --default-signal=INT 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" @@ -1529,6 +1536,33 @@ done assert_gone(grandchild).await; } + /// SIGINT rather than SIGKILL: an agent only stops the commands it + /// started itself if it gets to shut down. + #[tokio::test] + async fn dropping_the_runtime_lets_its_agent_shut_down_on_sigint() { + let dir = tempfile::tempdir().unwrap(); + let runtime = agent( + dir.path(), + "ok", + &[("TRAP_INT", "1")], + PercentPolicy::default(), + ); + runtime + .run(&config(dir.path()), "one", &NoopSink) + .await + .unwrap(); + + drop(runtime); + let marker = dir.path().join("prompts.int"); + for _ in 0..250 { + if marker.exists() { + break; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + assert!(marker.exists(), "the agent never got SIGINT"); + } + fn sent(session: &str, text: &str) -> (String, String) { (session.to_owned(), text.to_owned()) } diff --git a/hive-runtime/src/acp/rpc.rs b/hive-runtime/src/acp/rpc.rs index 65681f02..6d07d054 100644 --- a/hive-runtime/src/acp/rpc.rs +++ b/hive-runtime/src/acp/rpc.rs @@ -31,9 +31,10 @@ type Reply = Result; type Pending = Arc)>>>; /// 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. +/// connection sends that group SIGINT, so the agent can stop the commands +/// it started, and they get the signal too. Nothing escalates: an agent +/// that ignores SIGINT keeps running. 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 @@ -66,7 +67,6 @@ impl Connection { .stdout(Stdio::piped()) .stderr(Stdio::piped()) .process_group(0) - .kill_on_drop(true) .spawn() .map_err(|source| AcpError::Spawn { program: command.command.clone(), @@ -182,7 +182,7 @@ impl Drop for Connection { // 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); + libc::kill(-pgid, libc::SIGINT); } } } diff --git a/hive-subagent-mcp/src/mcp.rs b/hive-subagent-mcp/src/mcp.rs index 8035dba1..6d24e0c5 100644 --- a/hive-subagent-mcp/src/mcp.rs +++ b/hive-subagent-mcp/src/mcp.rs @@ -187,10 +187,6 @@ struct StatusArgs { struct InterruptArgs { /// The running session's name to signal. name: String, - /// `true` sends SIGKILL immediately; `false` (default) sends SIGINT, - /// letting claude shut down cleanly if it's already mid-response. - #[serde(default)] - force: bool, } #[derive(Clone)] @@ -276,11 +272,13 @@ impl SubagentMcp { re-prompted toward it afterwards, and `continue` is what restarts it. Only works \ once it's actually running — a `start`/`continue` still in its brief window before \ the process is confirmed spawned refuses interrupt too (nothing to signal yet; \ - retry shortly), same as a name with nothing tracked at all. `force: true` for \ - SIGKILL, otherwise SIGINT." + retry shortly), same as a name with nothing tracked at all. On claude it sends \ + SIGINT, which claude answers by stopping the commands it started and exiting; \ + nothing escalates past it. Returns at once; the name stays claimed until claude \ + has exited, and `status` says `stopping` until then." )] fn interrupt(&self, Parameters(args): Parameters) -> String { - match session::interrupt(&self.state, &args.name, args.force) { + match session::interrupt(&self.state, &args.name) { Ok(msg) => msg, Err(e) => format!("interrupt error: {e:#}"), } diff --git a/hive-subagent-mcp/src/session.rs b/hive-subagent-mcp/src/session.rs index 8411e45a..4a276e82 100644 --- a/hive-subagent-mcp/src/session.rs +++ b/hive-subagent-mcp/src/session.rs @@ -5,8 +5,8 @@ //! continuation loop records. //! //! **No task files, no restart recovery.** The daemon's only state is an -//! in-memory `name -> Option` map (see `State`'s own doc for the -//! rest of the maps and for what the `None`/`Some` split is for) — all of +//! in-memory `name -> Slot` map (see `State`'s own doc for the rest of the +//! maps, and `Slot`'s for what each state of a claimed name means) — all of //! it living for exactly as long as the process is. A daemon restart means //! whatever was running gets killed //! with it (`tokio`'s own child-process drop semantics), not adopted, and @@ -230,11 +230,33 @@ enum SubagentRuntime { enum Stopper { /// Signals the turn's `claude` process. Claude(Cancel), - /// Sends the ACP agent `session/cancel`; one that ignores it is killed - /// by the runtime after its own grace period. + /// Sends the ACP agent `session/cancel`; the runtime sends SIGINT to one + /// that ignores it for its own grace period. Acp(Canceller), } +/// What `running` holds for a claimed name. +enum Slot { + /// Claimed, with no process confirmed under it: a `start`/`continue` + /// that hasn't spawned yet, or a goal run between turns. + Starting, + /// A turn is in flight, and this is how `interrupt` reaches it. + Running(Stopper), + /// `interrupt` has signalled a claude turn and its process hasn't exited + /// yet. Nothing escalates past that SIGINT, so a claude that ignores it + /// stays here until it exits; the turn's own `finish_turn` frees the + /// name. + Stopping, +} + +/// Which [`Slot`] a claimed name is in, without its cancel handle. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum Occupancy { + Starting, + Running, + Stopping, +} + /// What settled a resumed turn's bounded wait — the one thing `continue` /// blocks on before it answers. enum ResumeVerdict { @@ -333,13 +355,13 @@ fn searched_location(config: &Config) -> Option { /// and where to push the completion todo. `Arc`-wrapped so the background /// task driving a run outlives the tool call that started it. /// -/// The map value is `Option`: `None` means `name` is reserved for -/// an in-flight `start`/`continue` that hasn't reached a confirmed -/// spawn yet; `Some(stopper)` means a real turn is tracked and -/// interruptible. The `None` state exists to close a real TOCTOU window a -/// reviewer caught in the original check-then-insert version: checking "is -/// `name` free" and committing to it are two different lock acquisitions -/// unless the check *is* the reservation — see `reserve`. +/// The map value is a `Slot`: `Starting` means `name` is reserved for an +/// in-flight `start`/`continue` that hasn't reached a confirmed spawn yet; +/// `Running` means a real turn is tracked and interruptible; `Stopping` +/// means it was interrupted and its process is still winding down. Each +/// claims the name. `Starting` closes a TOCTOU window: checking "is `name` +/// free" and committing to it are two different lock acquisitions unless +/// the check *is* the reservation — see `reserve`. /// /// `killed` is the other half of that map's story: `running` says what is in /// flight *now*, `killed` remembers the names whose last turn ended on a @@ -359,7 +381,7 @@ fn searched_location(config: &Config) -> Option { /// name across directories at your own risk, the tool doesn't disambiguate /// it (flagged in review when `dir` was added). pub struct State { - running: Mutex>>, + running: Mutex>, dirs: Mutex>, killed: Mutex>, last_event: Mutex>, @@ -547,17 +569,17 @@ impl State { } } - /// `Some(true)` — a live process is tracked, interruptible. `Some(false)` - /// — the name is claimed but no process is confirmed under it: an - /// in-flight start/continue that hasn't spawned yet, or the gap between - /// one continued turn's child exiting and the next one spawning (see - /// `between_turns`). `None` — nothing tracked under `name` at all. - fn occupancy(&self, name: &str) -> Option { + /// Which [`Slot`] `name` is in; `None` with nothing tracked under it. + fn occupancy(&self, name: &str) -> Option { self.running .lock() .unwrap_or_else(PoisonError::into_inner) .get(name) - .map(Option::is_some) + .map(|slot| match slot { + Slot::Starting => Occupancy::Starting, + Slot::Running(_) => Occupancy::Running, + Slot::Stopping => Occupancy::Stopping, + }) } /// Atomically claim `name` for an in-flight start/continue: check @@ -572,21 +594,21 @@ impl State { if running.contains_key(name) { return false; } - running.insert(name.to_owned(), None); + running.insert(name.to_owned(), Slot::Starting); true } /// Release a reservation that never made it to a real spawn (an error /// on the slow path between `reserve` and `Claude::spawn` succeeding). - /// A no-op if the entry was already upgraded to `Some` — this only ever - /// clears a still-`None` placeholder, never a live process. + /// A no-op if the entry was already upgraded to `Running` — this only + /// ever clears a still-`Starting` placeholder, never a live process. /// /// The signal token the failed call minted goes with it: no process ever /// read that URL, and a token outliving the call that minted it is the /// one way a route could resolve to a session that isn't there. fn release_reservation(&self, name: &str) { let mut running = self.running.lock().unwrap_or_else(PoisonError::into_inner); - let released = matches!(running.get(name), Some(None)); + let released = matches!(running.get(name), Some(Slot::Starting)); if released { running.remove(name); } @@ -742,14 +764,14 @@ impl State { } } - /// Upgrade `name`'s `None` reservation to a real, interruptible process. - /// Same key as the reservation, so there is no window in which `name` - /// reads as unoccupied between the two. + /// Upgrade `name`'s `Starting` reservation to a real, interruptible + /// process. Same key as the reservation, so there is no window in which + /// `name` reads as unoccupied between the two. fn track(&self, name: &str, stopper: Stopper) { self.running .lock() .unwrap_or_else(PoisonError::into_inner) - .insert(name.to_owned(), Some(stopper)); + .insert(name.to_owned(), Slot::Running(stopper)); } /// Drop `name`'s cancel handle but keep the name claimed — the moment @@ -763,7 +785,7 @@ impl State { self.running .lock() .unwrap_or_else(PoisonError::into_inner) - .insert(name.to_owned(), None); + .insert(name.to_owned(), Slot::Starting); } /// Set (or clear) `name`'s goal and start its turn budget over at turn @@ -1286,7 +1308,7 @@ pub async fn start(state: &Arc, req: StartRequest) -> anyhow::Result) -> String { /// them is a fact about the same session. struct StatusFacts<'a> { name: &'a str, - occupancy: Option, + occupancy: Option, killed: Option, session_exists: bool, last_event_age: Option, @@ -2224,7 +2248,7 @@ fn describe_status(facts: &StatusFacts<'_>) -> anyhow::Result { let name = facts.name; let turns = describe_turns(facts.turns); match facts.occupancy { - Some(true) => { + Some(Occupancy::Running) => { return Ok(format!( "subagent `{name}` is running — its turn is still in flight, so there's nothing \ to do but let it work: the daemon pushes a todo when the run ends, or \ @@ -2233,7 +2257,7 @@ fn describe_status(facts: &StatusFacts<'_>) -> anyhow::Result { describe_pending_signal(facts.stop.as_ref()), )); } - Some(false) => { + Some(Occupancy::Starting) => { return Ok(format!( "subagent `{name}` is starting — the name is claimed but no process is confirmed \ under it yet, either because a `start`/`continue` hasn't spawned one or because \ @@ -2241,6 +2265,15 @@ fn describe_status(facts: &StatusFacts<'_>) -> anyhow::Result { shortly rather than starting anything else under this name.{turns}" )); } + Some(Occupancy::Stopping) => { + return Ok(format!( + "subagent `{name}` is stopping — `interrupt` sent it SIGINT and its process \ + hasn't exited yet. The name stays claimed until it does, so `start` and \ + `continue` are refused meanwhile. Nothing escalates past that SIGINT: one that \ + never exits needs killing by hand.{turns}{}", + describe_liveness(facts.last_event_age), + )); + } None => {} } if let Some(signal) = facts.killed { @@ -2327,57 +2360,60 @@ fn describe_stopped(name: &str, stop: &StopReason, turns: &str) -> String { } } -/// Signal `name`'s running process — `force` picks SIGKILL over SIGINT (see -/// `hive_claude::Cancel::cancel`). Refuses a name with nothing running: no -/// entry at all, or one still in the brief not-yet-spawned window (nothing -/// to signal yet — the reservation is put back so a concurrent -/// `start`/`continue` for the same name still gets refused). +/// Stop `name`'s running turn: SIGINT to claude's process group (see +/// `hive_claude::Cancel::cancel`), or `session/cancel` on ACP. Nothing +/// escalates to SIGKILL: each command claude's Bash tool starts leads its +/// own session, and only claude shutting down on SIGINT stops it. /// -/// This stops a goal run, not just the turn in it — and it does so by -/// **recording `StopReason::Cancelled`**, not by relying on how the child -/// dies. +/// Returns without waiting. A claude turn's name stays claimed as +/// `Slot::Stopping` until the turn's own `finish_turn` clears it, so no +/// second claude starts on the session while the first is winding down. /// -/// That distinction is the entire fix. The signal alone is not enough: the -/// default is SIGINT, claude handles SIGINT by emitting its terminal -/// `result` event and exiting 0, and a zero exit is `TurnEnd::Complete` — -/// so `spawn_and_track`'s killed-turn early return never fires and the loop -/// goes straight on to spawn turn three of five. The recorded reason is what -/// `plan_after_turn` reads, and it is checked before the goal, so the run -/// stops whichever way the child ends up exiting. `force`'s SIGKILL does -/// produce a `TurnEnd::Killed`; recording the reason anyway costs nothing -/// and keeps one path, and `status` still prefers the kill record it -/// already had. -/// -/// The reason is written while the `running` lock is still held, so there is -/// no instant in which the name has lost its cancel handle but not yet -/// gained its stop reason — a turn ending of its own accord in that gap -/// would find nothing to stop it and start the next one. +/// It stops the goal run, not just the turn, by **recording +/// `StopReason::Cancelled`**: claude answers SIGINT by emitting its +/// terminal `result` event and exiting 0, which is `TurnEnd::Complete`, so +/// only the recorded reason, which `plan_after_turn` checks before the +/// goal, keeps the loop from re-prompting. It is written while the +/// `running` lock is held, so a turn that ends on its own in between still +/// finds it. /// /// # Errors /// -/// An invalid name, nothing tracked under `name`, or nothing to signal right -/// now (a spawn or ACP turn not yet live, or a goal run between turns). -pub fn interrupt(state: &State, name: &str, force: bool) -> anyhow::Result { +/// An invalid name, nothing tracked under `name`, nothing to signal right +/// now (a spawn or ACP turn not yet live, or a goal run between turns; the +/// reservation is put back, so a concurrent `start`/`continue` is still +/// refused), or a claude already signalled and not yet exited. +pub fn interrupt(state: &State, name: &str) -> anyhow::Result { validate_name(name)?; let mut running = state.running.lock().unwrap_or_else(PoisonError::into_inner); + let mut claimed = ""; match running.remove(name) { None => anyhow::bail!("no subagent named `{name}` is currently running"), - Some(None) => { - running.insert(name.to_owned(), None); + Some(Slot::Starting) => { + running.insert(name.to_owned(), Slot::Starting); return Err(still_starting(name)); } - Some(Some(Stopper::Claude(cancel))) => { - state.record_stop(name, StopReason::Cancelled); - drop(running); - cancel.cancel(force); + Some(Slot::Stopping) => { + running.insert(name.to_owned(), Slot::Stopping); + anyhow::bail!( + "subagent `{name}` was already interrupted and its process hasn't exited yet — \ + `status` says when it has" + ); } - // `force` has no stronger form on ACP: the runtime already kills an - // agent that ignores `session/cancel`. `cancel()` is `false` when no - // turn is in flight (the agent not prompted yet, or a turn just - // ended), and then nothing was cancelled. The stop goes in before the - // cancel and comes out under the same `stops` lock, so - // `plan_after_turn` never reads a cancel that didn't happen. - Some(Some(Stopper::Acp(canceller))) => { + Some(Slot::Running(Stopper::Claude(cancel))) => { + state.record_stop(name, StopReason::Cancelled); + running.insert(name.to_owned(), Slot::Stopping); + drop(running); + cancel.cancel(false); + claimed = " The name stays claimed until claude has exited; `status` says `stopping` \ + until then."; + } + // `cancel()` is `false` when no turn is in flight (the agent not + // prompted yet, or a turn just ended), and then nothing was + // cancelled. The stop goes in before the cancel and comes out under + // the same `stops` lock, so `plan_after_turn` never reads a cancel + // that didn't happen. + Some(Slot::Running(Stopper::Acp(canceller))) => { let mut stops = state.stops.lock().unwrap_or_else(PoisonError::into_inner); let prior = stops.insert(name.to_owned(), StopReason::Cancelled); if !canceller.cancel() { @@ -2386,7 +2422,7 @@ pub fn interrupt(state: &State, name: &str, force: bool) -> anyhow::Result stops.remove(name), }; drop(stops); - running.insert(name.to_owned(), Some(Stopper::Acp(canceller))); + running.insert(name.to_owned(), Slot::Running(Stopper::Acp(canceller))); return Err(still_starting(name)); } } @@ -2394,10 +2430,24 @@ pub fn interrupt(state: &State, name: &str, force: bool) -> anyhow::Result anyhow::Error { + if state.occupancy(name) == Some(Occupancy::Stopping) { + anyhow::anyhow!( + "subagent `{name}` is stopping — it was interrupted and its process hasn't exited \ + yet; the name is free once it has (`status` says when)" + ) + } else { + anyhow::anyhow!("subagent `{name}` is already running — {hint}") + } +} + /// `interrupt`'s refusal for a name that is claimed but has nothing to /// signal yet. fn still_starting(name: &str) -> anyhow::Error { @@ -2539,7 +2589,7 @@ mod tests { state.reserve("n"); assert_eq!( state.occupancy("n"), - Some(false), + Some(Occupancy::Starting), "reserved-but-not-yet-spawned must read as occupied-but-not-running" ); } @@ -2984,7 +3034,7 @@ mod tests { // has to carry its own explanation — checked one state at a time, // which is all a caller ever gets back. let running = describe_status(&StatusFacts { - occupancy: Some(true), + occupancy: Some(Occupancy::Running), ..StatusFacts::new("n") }) .expect("running status"); @@ -2993,7 +3043,7 @@ mod tests { "a running answer must say the turn is in flight and how to stop it: {running}" ); let starting = describe_status(&StatusFacts { - occupancy: Some(false), + occupancy: Some(Occupancy::Starting), ..StatusFacts::new("n") }) .expect("starting status"); @@ -3238,7 +3288,7 @@ mod tests { #[test] fn a_running_status_reports_the_age_and_an_idle_one_does_not() { let running = describe_status(&StatusFacts { - occupancy: Some(true), + occupancy: Some(Occupancy::Running), last_event_age: Some(Duration::from_secs(4)), ..StatusFacts::new("n") }) @@ -3249,7 +3299,7 @@ mod tests { {running}" ); let wedged = describe_status(&StatusFacts { - occupancy: Some(true), + occupancy: Some(Occupancy::Running), last_event_age: Some(Duration::from_mins(15)), ..StatusFacts::new("n") }) @@ -3281,7 +3331,7 @@ mod tests { // turn can finish between the two. The answer drops the age rather // than inventing one. let raced = describe_status(&StatusFacts { - occupancy: Some(true), + occupancy: Some(Occupancy::Running), ..StatusFacts::new("n") }) .expect("running status"); @@ -3710,9 +3760,7 @@ mod tests { } assert_eq!(turns_spawned(&log), 1, "the first turn never started"); - // An operator's plain `interrupt`, not `force` — SIGINT is the - // default and the case that was broken. - interrupt(&state, "n", false).expect("a running turn is interruptible"); + interrupt(&state, "n").expect("a running turn is interruptible"); // Generous: the fake logs its invocation as its first act, so a // re-prompt shows up here almost immediately once the loop decides @@ -3776,53 +3824,131 @@ mod tests { #[tokio::test] async fn an_interrupt_stops_the_commands_the_subagent_started() { - for force in [false, true] { - let dir = std::env::temp_dir().join(format!( - "hive-subagent-group-{}-{:?}-{force}", - std::process::id(), - std::thread::current().id() - )); - std::fs::create_dir_all(&dir).expect("scratch dir"); - let pidfile = dir.join("grandchild.pid"); - let state = Arc::new(mid_run("n", "carry on indefinitely", 5)); - let config = Config { - program: Some(busy_claude(&dir, &pidfile)), - ..Config::default() - }; - spawn_and_track( - &state, - "n", - &config, - &Attach::Create("n".to_owned()), - "turn one".to_owned(), - None, - ) - .expect("the fake claude spawns"); - let grandchild = read_pid(&pidfile).await; + let dir = scratch_dir("group"); + let pidfile = dir.join("grandchild.pid"); + let state = Arc::new(mid_run("n", "carry on indefinitely", 5)); + let config = Config { + program: Some(busy_claude(&dir, &pidfile)), + ..Config::default() + }; + spawn_and_track( + &state, + "n", + &config, + &Attach::Create("n".to_owned()), + "turn one".to_owned(), + None, + ) + .expect("the fake claude spawns"); + let grandchild = read_pid(&pidfile).await; - interrupt(&state, "n", force).expect("a running turn is interruptible"); + interrupt(&state, "n").expect("a running turn is interruptible"); - let mut alive = is_running(grandchild); - for _ in 0..250 { - if !alive { - break; - } - tokio::time::sleep(Duration::from_millis(20)).await; - alive = is_running(grandchild); + let mut alive = is_running(grandchild); + for _ in 0..250 { + if !alive { + break; } - if alive { - // SAFETY: `kill(2)` on a pid this test started; keeps a failed - // run from leaving a 300s `sleep` behind. - unsafe { - libc::kill(grandchild, libc::SIGKILL); - } - } - std::fs::remove_dir_all(&dir).ok(); - assert!( - !alive, - "grandchild {grandchild} outlived interrupt(force = {force})" - ); + tokio::time::sleep(Duration::from_millis(20)).await; + alive = is_running(grandchild); } + if alive { + // SAFETY: `kill(2)` on a pid this test started; keeps a failed + // run from leaving a 300s `sleep` behind. + unsafe { + libc::kill(grandchild, libc::SIGKILL); + } + } + std::fs::remove_dir_all(&dir).ok(); + assert!(!alive, "grandchild {grandchild} outlived the interrupt"); + } + + /// A stand-in for claude that shuts down on SIGINT the way claude does: + /// it writes `interrupted` once the signal arrives, then exits 0 as soon + /// as `release` exists, or after 10s. It writes `ready` once its trap is + /// set. A SIGKILL leaves no `interrupted` behind. + fn graceful_claude(dir: &Path) -> PathBuf { + let script = dir.join("graceful-claude"); + write_executable( + &script, + &format!( + "#!/bin/sh\ncd '{}'\ntrap ': > interrupted; i=0; until [ -e release ] || \ + [ $i -ge 200 ]; do sleep 0.05; i=$((i+1)); done; kill $!; exit 0' INT\n\ + sleep 300 >/dev/null 2>&1 &\n: > ready\nwait\n", + dir.display() + ), + ); + script + } + + /// Spawn [`graceful_claude`] as `n`'s turn and wait until its trap is set. + async fn spawn_graceful(state: &Arc, dir: &Path) { + let config = Config { + program: Some(graceful_claude(dir)), + ..Config::default() + }; + spawn_and_track( + state, + "n", + &config, + &Attach::Create("n".to_owned()), + "turn one".to_owned(), + None, + ) + .expect("the fake claude spawns"); + until("the fake claude sets its trap", || { + dir.join("ready").exists() + }) + .await; + } + + #[tokio::test] + async fn an_interrupt_lets_claude_shut_down_on_sigint() { + let dir = scratch_dir("sigint"); + std::fs::write(dir.join("release"), "").expect("release file"); + let state = Arc::new(mid_run("n", "carry on indefinitely", 5)); + spawn_graceful(&state, &dir).await; + + interrupt(&state, "n").expect("a running turn is interruptible"); + + until("the run ends", || state.occupancy("n").is_none()).await; + let interrupted = dir.join("interrupted").exists(); + std::fs::remove_dir_all(&dir).ok(); + assert!( + interrupted, + "claude must get SIGINT, which it shuts its tool commands down on, never SIGKILL" + ); + assert_eq!(state.killed_by("n"), None, "it exited on its own"); + } + + #[tokio::test] + async fn an_interrupted_name_stays_claimed_until_claude_has_exited() { + let dir = scratch_dir("stopping"); + let state = Arc::new(mid_run("n", "carry on indefinitely", 5)); + spawn_graceful(&state, &dir).await; + + interrupt(&state, "n").expect("a running turn is interruptible"); + until("the fake claude gets the SIGINT", || { + dir.join("interrupted").exists() + }) + .await; + + assert_eq!(state.occupancy("n"), Some(Occupancy::Stopping)); + let status = status(&state, "n", None).expect("status"); + assert!(status.contains("is stopping"), "{status}"); + let refused = continue_(&state, "n", "more".to_owned(), None, None, None) + .await + .expect_err("a second claude must not start on the session"); + assert!(refused.to_string().contains("is stopping"), "{refused}"); + assert!(interrupt(&state, "n").is_err(), "it was already signalled"); + + std::fs::write(dir.join("release"), "").expect("release file"); + until("the run ends", || state.occupancy("n").is_none()).await; + std::fs::remove_dir_all(&dir).ok(); + assert!( + state.reserve("n"), + "the name is free once claude has exited" + ); } #[test] @@ -4079,7 +4205,12 @@ mod tests { #[test] fn status_reports_the_turn_counter_in_every_state_a_goal_run_reaches() { - for occupancy in [Some(true), Some(false), None] { + for occupancy in [ + Some(Occupancy::Running), + Some(Occupancy::Starting), + Some(Occupancy::Stopping), + None, + ] { let answer = describe_status(&StatusFacts { occupancy, session_exists: true, @@ -4094,7 +4225,7 @@ mod tests { ); } let goalless = describe_status(&StatusFacts { - occupancy: Some(true), + occupancy: Some(Occupancy::Running), ..StatusFacts::new("n") }) .expect("status"); @@ -4110,7 +4241,7 @@ mod tests { // it is blocked, its turn hasn't ended yet, and plain "running" would // tell the parent to leave it alone. let answer = describe_status(&StatusFacts { - occupancy: Some(true), + occupancy: Some(Occupancy::Running), stop: Some(StopReason::NeedHelp("no credential".to_owned())), ..StatusFacts::new("n") }) @@ -4120,7 +4251,7 @@ mod tests { "both facts are true at once and both have to be in the answer: {answer}" ); let quiet = describe_status(&StatusFacts { - occupancy: Some(true), + occupancy: Some(Occupancy::Running), stop: Some(StopReason::Done), ..StatusFacts::new("n") }) @@ -4633,10 +4764,8 @@ done turns_spawned(&dir.join("claude-turns")) == 1 }) .await; - interrupt(&state, "n", false).expect("a running claude turn is interruptible"); - // `interrupt` drops the tracking entry itself; the run is over once - // `finish_turn` has cleared its liveness clock too. - until("the run ends", || state.last_event_age("n").is_none()).await; + interrupt(&state, "n").expect("a running claude turn is interruptible"); + until("the run ends", || state.occupancy("n").is_none()).await; // This `start` had no goal, so the fake claude's clean exit after // SIGINT (`TurnEnd::Complete`) is exactly the case that used to @@ -4983,7 +5112,7 @@ echo '{"type":"result","subtype":"success","is_error":false}' }) .await; - interrupt(&state, "n", false).expect("a running ACP turn is interruptible"); + interrupt(&state, "n").expect("a running ACP turn is interruptible"); until("the run ends", || state.last_event_age("n").is_none()).await; assert_eq!(lines(&dir.join("prompts")).len(), 1, "no second goal turn"); @@ -5006,7 +5135,7 @@ echo '{"type":"result","subtype":"success","is_error":false}' // The test runtime is single-threaded, so the run's task has not // been polled yet: no turn is in flight. - let err = interrupt(&state, "n", false).expect_err("nothing to cancel yet"); + let err = interrupt(&state, "n").expect_err("nothing to cancel yet"); assert!(err.to_string().contains("still starting"), "{err}"); assert_eq!(state.stop_reason("n"), None); From f3860f34973192b883ffe60c7b3c0d34a92a8cc2 Mon Sep 17 00:00:00 2001 From: atlas Date: Sat, 10 Oct 2026 21:54:07 +0200 Subject: [PATCH 5/9] hive-claude 0.1.4: the idle watchdog stops claude with SIGINT hive-claude 0.1.4 (hive-claude #28) sends the idle watchdog's signal as SIGINT to claude's process group instead of SIGKILL. It then waits for claude to exit, with no SIGKILL fallback. claude stops the commands its Bash tool started when it shuts down on SIGINT, and a SIGKILL left them running. This is the last SIGKILL of a claude process from #4999. hive-agent's `idle_timeout` -> `Error::IdleTimeout` -> `TurnError::ApiStall` path is unchanged. The difference is that the error now arrives only once claude has exited. A claude that ignores the SIGINT keeps the turn open until it is killed by hand or the unit stops. The Cargo.lock change is only hive-claude's version and checksum. A plain `cargo update -p hive-claude --precise 0.1.4` also moved seven crates' Windows-only `windows-sys` edges from 0.61.2 to 0.52.0, unrelated to this bump. The minimal lock passes `--locked`. The turn-loop and web-UI docs, and hive-agent's watchdog docs and stall note, now say SIGINT instead of "killed". That includes the ACP spots that the previous commit's `Drop` change made stale. --- Cargo.lock | 4 ++-- Cargo.toml | 2 +- docs/turn-loop/README.md | 10 +++++++--- docs/web-ui/agent.md | 2 +- hive-agent/src/turn.rs | 12 ++++++------ 5 files changed, 17 insertions(+), 13 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index e8c42a52..3ff35e3b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1835,9 +1835,9 @@ dependencies = [ [[package]] name = "hive-claude" -version = "0.1.3" +version = "0.1.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8c87ac627823f7bfc38478d7494e614e3f680dda6007dbcd306af51d3eed3bb3" +checksum = "4d7c5638f8f59e438d6416b3048709c9b3d6a10c23872ecf3825e3fe4c1bd7e8" dependencies = [ "libc", "serde", diff --git a/Cargo.toml b/Cargo.toml index ab3840c7..7f73e0ab 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -97,7 +97,7 @@ hive-jobq-metrics = { path = "hive-jobq-metrics" } hive-jobq-wire = { path = "hive-jobq-wire" } hive-log = { path = "hive-log" } hive-core-agent-sock = { path = "hive-core-agent-sock" } -hive-claude = "0.1.3" +hive-claude = "0.1.4" hive-host-sock = { path = "hive-host-sock" } hive-priv-sock = { path = "hive-priv-sock" } hive-runtime = { path = "hive-runtime" } diff --git a/docs/turn-loop/README.md b/docs/turn-loop/README.md index 961f3a6d..ee2327b1 100644 --- a/docs/turn-loop/README.md +++ b/docs/turn-loop/README.md @@ -142,8 +142,8 @@ else a `TurnError`) drives the post-claude branch: | `Err(RateLimited)` | sleep `HIVE_RATE_LIMIT_SLEEP_SECS` (default 300), requeue inflight, status back to `online` | | `Err(AuthFailed)` | emit `needs_login_idle` sentinel, requeue inflight, park in `wait_for_login` | | `Err(SessionNotFound)` | resume + create self-heal both missed ("shouldn't happen"); requeue inflight so the next turn creates fresh — no status park, message not dropped | -| `Err(ApiStall)` | idle watchdog killed claude after `HIVE_TURN_IDLE_SECS` (default 600) of output silence; sleep `HIVE_STALL_SLEEP_SECS` (default 60), requeue inflight, status back to `online` | -| `Err(AgentStall(note))` | the same, for an ACP agent: after `HIVE_TURN_IDLE_SECS` with no `session/update` the harness sent the agent `session/cancel`, and killed it if it ignored that; `note` says which | +| `Err(ApiStall)` | idle watchdog sent claude SIGINT after `HIVE_TURN_IDLE_SECS` (default 600) of output silence and claude exited; sleep `HIVE_STALL_SLEEP_SECS` (default 60), requeue inflight, status back to `online` | +| `Err(AgentStall(note))` | the same, for an ACP agent: after `HIVE_TURN_IDLE_SECS` with no `session/update` the harness sent the agent `session/cancel`, and SIGINT if it ignored that; `note` says which | | `Err(Failed(err))` | route `[system] \`\` claude turn failed:\n` to `operator` via `send_to_operator` | `ApiStall` catches an Anthropic API stall — a multi-retry connection storm where @@ -152,7 +152,11 @@ the stream goes silent for minutes. On `claude` the idle watchdog lives in resets on every stdout line, so a large but still-streaming turn is never cut — only complete output silence for the window trips it. The harness sets the window from `HIVE_TURN_IDLE_SECS` (`0` disables) and maps the driver's -`Error::IdleTimeout` onto `TurnError::ApiStall`. +`Error::IdleTimeout` onto `TurnError::ApiStall`. The watchdog sends claude's +process group SIGINT, so claude stops the commands it started, and then +waits for claude to exit. Nothing escalates to SIGKILL: a claude that +ignores the SIGINT keeps the turn open until someone kills it or the unit +stops. `hive-runtime` applies the same window to an ACP agent's turns. That's also how a provider HTTP 429 ends an ACP turn when the agent retries it without diff --git a/docs/web-ui/agent.md b/docs/web-ui/agent.md index ea836349..50ef0adf 100644 --- a/docs/web-ui/agent.md +++ b/docs/web-ui/agent.md @@ -229,7 +229,7 @@ Slash commands today: - `/clear` — wipe the local terminal view (server history kept). - `/cancel` — `POST /api/cancel` → SIGINT to the turn's claude and every command it started (its process group), emits a Note. On an ACP agent - it sends the agent `session/cancel` instead, and kills it if the turn + it sends the agent `session/cancel` instead, and SIGINT if the turn hasn't ended 10s later. Also surfaces as a `■ cancel turn` button in the state row while state=thinking. - `/compact` — `POST /api/compact` → sets the deferred diff --git a/hive-agent/src/turn.rs b/hive-agent/src/turn.rs index ee970c70..3bcc338f 100644 --- a/hive-agent/src/turn.rs +++ b/hive-agent/src/turn.rs @@ -50,8 +50,8 @@ const DEFAULT_SESSION_TITLE: &str = "hive-session"; /// capacity limits. const DEFAULT_RATE_LIMIT_SLEEP_SECS: u64 = 300; -/// Idle watchdog window: kill claude and surface `ApiStall` if it produces no -/// stdout for this long. The timer resets on every stdout line, so a large but +/// Idle watchdog window: send claude SIGINT and surface `ApiStall` once it +/// exits, if it produces no stdout for this long. The timer resets on every stdout line, so a large but /// still-streaming turn is never cut — only a complete silence trips it. /// Default is 10 minutes: long enough for legitimate big-context turns, short /// enough to cut a multi-retry Anthropic connection storm (atlas telemetry: @@ -189,14 +189,14 @@ pub enum TurnError { /// the wake message, the serve loop requeues it so the next turn retries; /// no status park. SessionNotFound, - /// The harness-side idle watchdog killed claude after `turn_idle_secs()` of + /// The harness-side idle watchdog stopped claude after `turn_idle_secs()` of /// output silence — indicative of an Anthropic API stall (e.g. a multi-retry /// connection storm burning minutes of wall-clock with no stream progress). /// The serve loop parks for `stall_sleep_secs()` and requeues, like the /// rate-limit path — NOT a crash. ApiStall, /// The ACP backend's idle watchdog stopped a turn that sent nothing for - /// `turn_idle_secs()`: the agent was sent `session/cancel`, and killed if + /// `turn_idle_secs()`: the agent was sent `session/cancel`, and SIGINT if /// it ignored that. Carries what happened. An ACP agent retrying a failing /// provider (e.g. HTTP 429) without reporting it ends up here. Handled /// like [`Self::ApiStall`]. @@ -558,11 +558,11 @@ pub fn emit_turn_end(bus: &Bus, outcome: &TurnOutcome) { bus.emit(LiveEvent::TurnEnd { ok: false, note: Some(format!( - "claude killed after {}s of output silence — API stall suspected, parking + requeueing", + "claude stopped with SIGINT after {}s of output silence — API stall suspected, parking + requeueing", turn_idle_secs() )), }); - tracing::warn!("turn killed: API stall (idle watchdog)"); + tracing::warn!("turn stopped: API stall (idle watchdog)"); } Err(TurnError::AgentStall(note)) => { bus.emit(LiveEvent::TurnEnd { From a20093521646194be894397d76a5e038972fcf51 Mon Sep 17 00:00:00 2001 From: atlas Date: Sat, 10 Oct 2026 20:37:07 +0200 Subject: [PATCH 6/9] hive-agent: extra web proxies forward every repeated response header Both proxy paths copied upstream response headers with HeaderMap::insert, which replaces earlier values of the same name, so an upstream sending two Set-Cookie (or Link, Vary, ...) headers reached the browser with only the last one. Use append in both loops. Tests cover the http and unix: paths. --- docs/web-ui/agent.md | 3 +- hive-agent/src/web_ui/proxy.rs | 71 +++++++++++++++++++++++++++++++++- 2 files changed, 71 insertions(+), 3 deletions(-) diff --git a/docs/web-ui/agent.md b/docs/web-ui/agent.md index 50ef0adf..63f5b708 100644 --- a/docs/web-ui/agent.md +++ b/docs/web-ui/agent.md @@ -380,7 +380,8 @@ shaped). `{"": ""}`). Each entry mounts a transparent reverse-proxy at `/extra//` that forwards every request (method, headers, body) to the configured upstream, strips hop-by-hop headers - on both sides, and buffers the full response body (MVP — SSE + on both sides (repeated response headers such as two `Set-Cookie` lines + are all forwarded), and buffers the full response body (MVP — SSE connections will appear as one large response rather than streaming). The `/extra/` namespace ensures user-declared proxies can never conflict with native agent endpoints. Upstream values are either an diff --git a/hive-agent/src/web_ui/proxy.rs b/hive-agent/src/web_ui/proxy.rs index 1bdb5e45..c44c207f 100644 --- a/hive-agent/src/web_ui/proxy.rs +++ b/hive-agent/src/web_ui/proxy.rs @@ -169,7 +169,7 @@ async fn proxy_handler( HeaderName::from_bytes(name.as_str().as_bytes()), HeaderValue::from_bytes(value.as_bytes()), ) { - resp_headers.insert(n, v); + resp_headers.append(n, v); } } match upstream_resp.bytes().await { @@ -269,8 +269,75 @@ async fn forward_over_unix_socket( if HOP_BY_HOP.contains(&name.as_str()) { continue; } - resp_headers.insert(name.clone(), value.clone()); + resp_headers.append(name.clone(), value.clone()); } let body_bytes = upstream_resp.into_body().collect().await?.to_bytes(); Ok((status, resp_headers, body_bytes).into_response()) } + +#[cfg(test)] +mod tests { + use super::*; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + /// Raw HTTP/1.1 answer with two `Set-Cookie` headers and a body. + const TWO_COOKIES: &[u8] = b"HTTP/1.1 200 OK\r\n\ + Set-Cookie: session=abc\r\n\ + Set-Cookie: csrf=def\r\n\ + Content-Length: 2\r\n\ + Connection: close\r\n\r\nok"; + + /// Read the request head, then answer with [`TWO_COOKIES`]. + async fn answer_once(mut stream: S) { + let mut buf = [0u8; 1024]; + let _ = stream.read(&mut buf).await; + stream.write_all(TWO_COOKIES).await.unwrap(); + stream.shutdown().await.unwrap(); + } + + fn cookies(resp: &Response) -> Vec<&str> { + resp.headers() + .get_all("set-cookie") + .iter() + .map(|v| v.to_str().unwrap()) + .collect() + } + + #[tokio::test] + async fn unix_upstream_keeps_repeated_response_headers() { + let dir = tempfile::tempdir().unwrap(); + let sock = dir.path().join("up.sock"); + let listener = tokio::net::UnixListener::bind(&sock).unwrap(); + tokio::spawn(async move { + let (stream, _) = listener.accept().await.unwrap(); + answer_once(stream).await; + }); + + let resp = + forward_over_unix_socket(&sock, Method::GET, "/", HeaderMap::new(), Bytes::new()) + .await + .unwrap(); + assert_eq!(cookies(&resp), ["session=abc", "csrf=def"]); + } + + #[tokio::test] + async fn http_upstream_keeps_repeated_response_headers() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let base = format!("http://{}", listener.local_addr().unwrap()); + tokio::spawn(async move { + let (stream, _) = listener.accept().await.unwrap(); + answer_once(stream).await; + }); + + let client = reqwest::Client::builder().no_proxy().build().unwrap(); + let resp = proxy_handler( + State((Arc::new(client), Arc::new(base))), + Method::GET, + Uri::from_static("/"), + HeaderMap::new(), + Bytes::new(), + ) + .await; + assert_eq!(cookies(&resp), ["session=abc", "csrf=def"]); + } +} From 7bb1959a94d82403bc400bfc8be59a7287434ab7 Mon Sep 17 00:00:00 2001 From: atlas Date: Sat, 10 Oct 2026 21:50:35 +0200 Subject: [PATCH 7/9] hive-agent: report a failed mid-turn /compact instead of recording it as compacted MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The turn-end /compact (operator button or the agent's `compact` tool) dropped the compaction error with `let _ =`, emitted no note, and returned `Ok(true)`, so the turn was recorded `result_kind = "compacted"` although nothing compacted. The idle path already printed `/compact failed: …`. Both paths now go through one `report_compact` helper that emits the done / failed note; the turn-end path returns whether the compact succeeded. --- docs/web-ui/agent.md | 3 +- hive-agent/src/turn.rs | 66 ++++++++++++++++++++++++++++++++++-------- 2 files changed, 56 insertions(+), 13 deletions(-) diff --git a/docs/web-ui/agent.md b/docs/web-ui/agent.md index 63f5b708..ba47e4cc 100644 --- a/docs/web-ui/agent.md +++ b/docs/web-ui/agent.md @@ -236,7 +236,8 @@ Slash commands today: `Bus::request_compact()` flag and returns immediately. The harness runs the `/compact` at the next turn boundary (end of the in-flight turn, or `turn::run_pending_compact` when idle); output streams into - the live panel. Works mid-turn, not only when idle. + the live panel. Works mid-turn, not only when idle. Either way the live + panel ends with a `/compact done` or `/compact failed: …` note. - `/model ` — `POST /api/model` flipping `Bus::set_model`. Takes effect on the next turn; persisted to `/state/hyperhive-model` so the override survives harness diff --git a/hive-agent/src/turn.rs b/hive-agent/src/turn.rs index 3bcc338f..f90d07dc 100644 --- a/hive-agent/src/turn.rs +++ b/hive-agent/src/turn.rs @@ -155,7 +155,7 @@ pub async fn write_system_prompt(socket: &Path, label: &str) -> Result /// One claude turn's outcome: `Ok(compacted)` on success, or a [`TurnError`] /// the serve loop must act on. The `compacted` bool is `true` when a /// compaction ran this turn (reactively on overflow, or proactively per the -/// policy — or an operator `/compact` at turn end); it's recorded as +/// policy — or a successful operator `/compact` at turn end); it's recorded as /// `result_kind = "compacted"` in turn stats so the stats page can distinguish /// those turns. Both `Ok(true)` and `Ok(false)` are ack'd; the error cases /// each map to a distinct serve-loop action (see [`emit_turn_end`] and the @@ -461,7 +461,7 @@ pub async fn drive_turn( // Reflect `Compacting` in the UI like the idle path (`run_pending_compact`) // does; the serve loop resets to `Idle` once this turn returns. bus.set_state(crate::events::TurnState::Compacting); - let _ = session.compact(&config, &sink).await; + let compacted = report_compact(bus, session.compact(&config, &sink).await); // If the compact call asked to be woken (the agent's own `compact` // tool with a `wake_prompt`), stash it — the serve loop reads it // back after this turn returns and drives a synthetic follow-up @@ -470,11 +470,31 @@ pub async fn drive_turn( if let Some(prompt) = request.wake_prompt { bus.set_post_compact_wake(prompt); } - return Ok(true); + return Ok(compacted); } outcome } +/// Emit the `/compact done` / `/compact failed: …` note for a finished +/// operator or agent `/compact` and return whether it succeeded. Shared by the +/// turn-end path in [`drive_turn`] and the idle path in [`run_pending_compact`]. +fn report_compact(bus: &Bus, result: Result<(), E>) -> bool { + match result { + Ok(()) => { + bus.emit(LiveEvent::Note { + text: "/compact done".into(), + }); + true + } + Err(e) => { + bus.emit(LiveEvent::Note { + text: format!("/compact failed: {e}"), + }); + false + } + } +} + /// Pre-turn auto-reset check. If context is large AND the prompt cache has /// gone cold (idle time >= cache TTL), archive the current session so the /// next wake-up turn's `--resume ` misses and self-heals into a fresh @@ -600,14 +620,7 @@ pub async fn run_pending_compact(files: &TurnFiles, bus: &Bus, session: &AgentSe bus.set_state(crate::events::TurnState::Compacting); let config = claude_config(bus, files); let sink = BusSink::new(bus); - match session.compact(&config, &sink).await { - Ok(()) => bus.emit(LiveEvent::Note { - text: "/compact done".into(), - }), - Err(e) => bus.emit(LiveEvent::Note { - text: format!("/compact failed: {e}"), - }), - } + report_compact(bus, session.compact(&config, &sink).await); bus.set_state(crate::events::TurnState::Idle); if let Some(prompt) = request.wake_prompt { bus.set_post_compact_wake(prompt); @@ -821,8 +834,10 @@ fn archive_session(bus: &Bus, session: &AgentSession) { #[cfg(test)] mod tests { use super::{ - PermissionAsk, TurnError, acp_permits, nonzero_or, parse_u64, runtime_error_to_turn, + PermissionAsk, TurnError, acp_permits, nonzero_or, parse_u64, report_compact, + runtime_error_to_turn, }; + use crate::events::{Bus, LiveEvent}; use hive_runtime::{AcpError, Error}; #[test] @@ -959,4 +974,31 @@ mod tests { Err(TurnError::Failed(_)) )); } + + fn compact_report(result: Result<(), &str>) -> (bool, Vec<String>) { + let bus = Bus::new(); + let mut rx = bus.subscribe(); + let ok = report_compact(&bus, result); + let mut notes = Vec::new(); + while let Ok(ev) = rx.try_recv() { + if let LiveEvent::Note { text } = ev.event { + notes.push(text); + } + } + (ok, notes) + } + + #[test] + fn a_failed_compact_is_reported_and_not_counted_as_compacted() { + let (compacted, notes) = compact_report(Err("spawn failed")); + assert!(!compacted); + assert_eq!(notes, ["/compact failed: spawn failed"]); + } + + #[test] + fn a_successful_compact_is_reported_and_counted_as_compacted() { + let (compacted, notes) = compact_report(Ok(())); + assert!(compacted); + assert_eq!(notes, ["/compact done"]); + } } From 2c4a3c2fa731fed3de7b8915e93697d944dcd0c2 Mon Sep 17 00:00:00 2001 From: atlas <atlas@hyperhive.local> Date: Sat, 10 Oct 2026 22:28:59 +0200 Subject: [PATCH 8/9] hive-agent: a failed turn-end /compact keeps the turn's own compacted flag The turn-end path returned the /compact's success as the turn's outcome, so a turn that had already compacted (reactively or per policy) was recorded as not compacted when its queued /compact then failed. The outcome now folds the two with `||`, via `with_turn_end_compact`, which the tests pin. --- hive-agent/src/turn.rs | 29 ++++++++++++++++++++++++++--- 1 file changed, 26 insertions(+), 3 deletions(-) diff --git a/hive-agent/src/turn.rs b/hive-agent/src/turn.rs index f90d07dc..4828d753 100644 --- a/hive-agent/src/turn.rs +++ b/hive-agent/src/turn.rs @@ -461,7 +461,7 @@ pub async fn drive_turn( // Reflect `Compacting` in the UI like the idle path (`run_pending_compact`) // does; the serve loop resets to `Idle` once this turn returns. bus.set_state(crate::events::TurnState::Compacting); - let compacted = report_compact(bus, session.compact(&config, &sink).await); + let compact_ok = report_compact(bus, session.compact(&config, &sink).await); // If the compact call asked to be woken (the agent's own `compact` // tool with a `wake_prompt`), stash it — the serve loop reads it // back after this turn returns and drives a synthetic follow-up @@ -470,7 +470,7 @@ pub async fn drive_turn( if let Some(prompt) = request.wake_prompt { bus.set_post_compact_wake(prompt); } - return Ok(compacted); + return with_turn_end_compact(outcome, compact_ok); } outcome } @@ -495,6 +495,13 @@ fn report_compact<E: std::fmt::Display>(bus: &Bus, result: Result<(), E>) -> boo } } +/// Fold a turn-end `/compact` into the turn's own outcome. A compaction that +/// ran during the turn (`Ok(true)`) stays recorded even when the turn-end +/// `/compact` failed; a failed `/compact` never turns it into `Ok(false)`. +fn with_turn_end_compact(outcome: TurnOutcome, compact_ok: bool) -> TurnOutcome { + outcome.map(|turn_compacted| turn_compacted || compact_ok) +} + /// Pre-turn auto-reset check. If context is large AND the prompt cache has /// gone cold (idle time >= cache TTL), archive the current session so the /// next wake-up turn's `--resume <title>` misses and self-heals into a fresh @@ -835,7 +842,7 @@ fn archive_session(bus: &Bus, session: &AgentSession) { mod tests { use super::{ PermissionAsk, TurnError, acp_permits, nonzero_or, parse_u64, report_compact, - runtime_error_to_turn, + runtime_error_to_turn, with_turn_end_compact, }; use crate::events::{Bus, LiveEvent}; use hive_runtime::{AcpError, Error}; @@ -1001,4 +1008,20 @@ mod tests { assert!(compacted); assert_eq!(notes, ["/compact done"]); } + + #[test] + fn a_failed_turn_end_compact_keeps_the_turns_own_compaction() { + assert!(matches!(with_turn_end_compact(Ok(true), false), Ok(true))); + } + + #[test] + fn a_failed_turn_end_compact_leaves_a_plain_turn_uncompacted() { + assert!(matches!(with_turn_end_compact(Ok(false), false), Ok(false))); + } + + #[test] + fn a_successful_turn_end_compact_marks_the_turn_compacted() { + assert!(matches!(with_turn_end_compact(Ok(false), true), Ok(true))); + assert!(matches!(with_turn_end_compact(Ok(true), true), Ok(true))); + } } From 3c03d828586e8fab9b944c77b95d438059d56429 Mon Sep 17 00:00:00 2001 From: atlas <atlas@hyperhive.local> Date: Sun, 11 Oct 2026 02:17:16 +0200 Subject: [PATCH 9/9] hive-agent: compact tests set the harness dir like the other Bus tests The two report_compact tests build a Bus, which resolves the events db under HYPERHIVE_HARNESS_DIR and panics when it is unset, as in the nix test sandbox. Add a test-only paths::with_harness_dir that points the variable at a temp dir under a lock, and use it in both the new tests and the existing swarm_inference Bus test, which set the variable unguarded. --- hive-agent/src/paths.rs | 35 +++++++++++++++++++++ hive-agent/src/swarm_inference.rs | 51 ++++++++++++++----------------- hive-agent/src/turn.rs | 20 ++++++------ 3 files changed, 69 insertions(+), 37 deletions(-) diff --git a/hive-agent/src/paths.rs b/hive-agent/src/paths.rs index fdc4e968..dc4ceddc 100644 --- a/hive-agent/src/paths.rs +++ b/hive-agent/src/paths.rs @@ -131,3 +131,38 @@ pub fn claude_dir() -> PathBuf { } PathBuf::from("/root/.claude") } + +/// Run `f` with `HYPERHIVE_HARNESS_DIR` pointing at a fresh temp dir, then +/// restore the previous value. Tests that build a `Bus` or read the harness +/// state files go through this: the variable is process-wide, so the calls +/// are serialised on a lock. +#[cfg(test)] +pub fn with_harness_dir<T>(f: impl FnOnce() -> T) -> T { + use std::ffi::OsString; + use std::sync::Mutex; + + static LOCK: Mutex<()> = Mutex::new(()); + + struct Restore(Option<OsString>); + impl Drop for Restore { + fn drop(&mut self) { + // SAFETY: still holding `LOCK`, the only writer of this variable. + unsafe { + match self.0.take() { + Some(v) => std::env::set_var("HYPERHIVE_HARNESS_DIR", v), + None => std::env::remove_var("HYPERHIVE_HARNESS_DIR"), + } + } + } + } + + let _lock = LOCK + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let dir = tempfile::tempdir().unwrap(); + // Declared after `_lock` so the variable is restored before the lock drops. + let _restore = Restore(std::env::var_os("HYPERHIVE_HARNESS_DIR")); + // SAFETY: serialised by `LOCK`. + unsafe { std::env::set_var("HYPERHIVE_HARNESS_DIR", dir.path()) }; + f() +} diff --git a/hive-agent/src/swarm_inference.rs b/hive-agent/src/swarm_inference.rs index 3be47c2d..eebb3d50 100644 --- a/hive-agent/src/swarm_inference.rs +++ b/hive-agent/src/swarm_inference.rs @@ -320,34 +320,29 @@ mod tests { /// persisted files, so clearing it, or a restart, finds the local value. #[test] fn the_swarm_value_is_not_persisted_over_the_local_one() { - let dir = tempfile::tempdir().unwrap(); - let prev = std::env::var_os("HYPERHIVE_HARNESS_DIR"); - // SAFETY: no other test in this crate reads or sets this variable. - unsafe { std::env::set_var("HYPERHIVE_HARNESS_DIR", dir.path()) }; - let bus = Bus::new(); - bus.set_model("sonnet"); - bus.set_effort("low"); - let target = Harness { - bus: bus.clone(), - nix_model: None, - nix_effort: None, - }; - let mut o = Overridden::default(); - o.apply(&set(Some("opus"), Some(Effort::Max)), &target); - let swarm = (bus.model(), bus.effort()); - let swarm_sources = (bus.model_source(), bus.effort_source()); - let files = ( - crate::harness_state::load_model(), - crate::harness_state::load_effort(), - ); - o.apply(&AgentInference::default(), &target); - let cleared = (bus.model(), bus.effort()); - let cleared_sources = (bus.model_source(), bus.effort_source()); - match prev { - // SAFETY: as above. - Some(v) => unsafe { std::env::set_var("HYPERHIVE_HARNESS_DIR", v) }, - None => unsafe { std::env::remove_var("HYPERHIVE_HARNESS_DIR") }, - } + let (swarm, swarm_sources, files, cleared, cleared_sources) = + crate::paths::with_harness_dir(|| { + let bus = Bus::new(); + bus.set_model("sonnet"); + bus.set_effort("low"); + let target = Harness { + bus: bus.clone(), + nix_model: None, + nix_effort: None, + }; + let mut o = Overridden::default(); + o.apply(&set(Some("opus"), Some(Effort::Max)), &target); + let swarm = (bus.model(), bus.effort()); + let swarm_sources = (bus.model_source(), bus.effort_source()); + let files = ( + crate::harness_state::load_model(), + crate::harness_state::load_effort(), + ); + o.apply(&AgentInference::default(), &target); + let cleared = (bus.model(), bus.effort()); + let cleared_sources = (bus.model_source(), bus.effort_source()); + (swarm, swarm_sources, files, cleared, cleared_sources) + }); assert_eq!(swarm, pair("opus", "max")); assert_eq!(files, (Some("sonnet".to_owned()), Some("low".to_owned()))); assert_eq!(cleared, pair("sonnet", "low")); diff --git a/hive-agent/src/turn.rs b/hive-agent/src/turn.rs index 4828d753..156e83ef 100644 --- a/hive-agent/src/turn.rs +++ b/hive-agent/src/turn.rs @@ -983,16 +983,18 @@ mod tests { } fn compact_report(result: Result<(), &str>) -> (bool, Vec<String>) { - let bus = Bus::new(); - let mut rx = bus.subscribe(); - let ok = report_compact(&bus, result); - let mut notes = Vec::new(); - while let Ok(ev) = rx.try_recv() { - if let LiveEvent::Note { text } = ev.event { - notes.push(text); + crate::paths::with_harness_dir(|| { + let bus = Bus::new(); + let mut rx = bus.subscribe(); + let ok = report_compact(&bus, result); + let mut notes = Vec::new(); + while let Ok(ev) = rx.try_recv() { + if let LiveEvent::Note { text } = ev.event { + notes.push(text); + } } - } - (ok, notes) + (ok, notes) + }) } #[test]