diff --git a/Cargo.lock b/Cargo.lock index 447ae62b..1e68b91b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1996,6 +1996,7 @@ dependencies = [ "hive-claude", "hive-sock-client", "hive-types", + "libc", "rmcp", "schemars", "serde", diff --git a/docs/tools/subagent.md b/docs/tools/subagent.md index ec3a18ed..82ae25ec 100644 --- a/docs/tools/subagent.md +++ b/docs/tools/subagent.md @@ -33,6 +33,26 @@ claude's own on-disk session (`hive_claude::SessionStore`), which `continue` reattaches to independent of the daemon's own lifetime — a restart loses the _in-flight turn_, not the subagent's history. +## A killed turn + +A subagent whose `claude` process dies on a signal — the kernel's OOM +killer under memory pressure, a stopped unit, an `interrupt` from the +owning agent — did not finish its turn, and the daemon says so rather +than letting it settle back into `idle`: + +- `status` reports the session **killed**, naming the signal, instead of + the `idle` it reports for a turn that ended on its own; +- the end-of-turn todo the daemon pushes without being asked says the + subagent was killed mid-turn, not that it finished; +- `continue` still resumes such a session — often what you want — but + its reply says the previous turn was killed, so nobody carries on from + cut-off work believing it was complete. + +The record is per-name, in memory with the rest of this daemon's state, +and the next confirmed spawn under that name clears it. A daemon restart +loses it along with everything else — the daemon has to outlive the kill +to report it, which is what `OOMPolicy=continue` on the unit is for. + ## Compaction trade-off Built on `hive_claude::Claude::spawn` + `RunningClaude::wait` directly diff --git a/hive-subagent-mcp/Cargo.toml b/hive-subagent-mcp/Cargo.toml index b3315719..e6d78ef0 100644 --- a/hive-subagent-mcp/Cargo.toml +++ b/hive-subagent-mcp/Cargo.toml @@ -15,6 +15,7 @@ hive-agent-sock.workspace = true hive-claude.workspace = true hive-sock-client.workspace = true hive-types.workspace = true +libc.workspace = true rmcp.workspace = true schemars.workspace = true serde.workspace = true diff --git a/hive-subagent-mcp/src/mcp.rs b/hive-subagent-mcp/src/mcp.rs index f0f3be86..0d59958b 100644 --- a/hive-subagent-mcp/src/mcp.rs +++ b/hive-subagent-mcp/src/mcp.rs @@ -131,7 +131,9 @@ impl SubagentMcp { because its previous turn finished and you have a follow-up instruction, or you're \ reattaching after this daemon restarted (the session itself survives independently \ of the daemon that spawned it). Returns as soon as confirmed running, same as \ - `start`. Refuses a name with no session on disk at all, or one already running." + `start`. Refuses a name with no session on disk at all, or one already running. \ + Resuming a session whose last turn was killed is allowed — the reply says so, since \ + that turn's work stopped wherever it had got to." )] fn r#continue(&self, Parameters(args): Parameters) -> String { match session::continue_( @@ -164,8 +166,10 @@ impl SubagentMcp { description = "Report whether a subagent is currently running — a zero-cost check that \ never launches a process, unlike `continue`. Distinguishes running, starting (a \ `start`/`continue` is in flight but not yet a confirmed spawn — this is normally \ - over in well under a second), idle (a session exists but nothing is in flight — \ - `continue` to give it another turn), and no such session at all." + over in well under a second), killed (its last turn died on a signal — the kernel's \ + OOM killer, a stopped unit, an `interrupt` — so its work was cut off mid-turn rather \ + than finished), idle (a session exists, its last turn finished, nothing is in flight \ + — `continue` to give it another turn), and no such session at all." )] fn status(&self, Parameters(args): Parameters) -> String { match session::status(&self.state, &args.name, args.dir.as_deref()) { diff --git a/hive-subagent-mcp/src/session.rs b/hive-subagent-mcp/src/session.rs index 98678a93..fdf6cf4b 100644 --- a/hive-subagent-mcp/src/session.rs +++ b/hive-subagent-mcp/src/session.rs @@ -1,5 +1,6 @@ //! The claude-facing half of this daemon: spawn a subagent turn, track it -//! only while it's alive, and push exactly one todo when it finishes. +//! only while it's alive, and push exactly one todo when it ends — saying +//! whether it finished or was killed. //! //! **No task files, no restart recovery.** The daemon's only state is an //! in-memory `name -> Option` map (see `State`'s own doc for what @@ -15,6 +16,15 @@ //! restarted and I want to pick this back up" — after a restart, re-supply //! `dir` once if the session isn't in the daemon's own default directory. //! + +//! **A killed turn is not a finished turn.** A child that died on a signal +//! arrives as a `hive_claude::Error::Exit` carrying its `ExitStatus`, so the +//! "how" is there to be read: `classify_end` takes the signal out of it and +//! `State::finish_turn` remembers it against the name, which is what lets +//! `status`, the end-of-turn todo and `continue`'s own reply all say the +//! session was killed rather than let it read as finished. See +//! `docs/tools/subagent.md`. +//! //! **No mid-turn compaction.** Building on `hive_claude::Claude::spawn` + //! `RunningClaude::wait` directly (not `InfiniteSession::run`) is what makes //! `interrupt` possible at all — `InfiniteSession` has no cancel handle to @@ -28,11 +38,61 @@ //! assumption stops holding, not shipped here. use std::collections::HashMap; +use std::os::unix::process::ExitStatusExt as _; use std::path::PathBuf; use std::sync::{Arc, Mutex, PoisonError}; use hive_claude::{Attach, Cancel, Claude, Config, NoopSink, SessionStore}; +/// How a subagent's turn ended, as far as the daemon can tell from what +/// [`hive_claude::RunningClaude::wait`] returned. +/// +/// The distinction that matters is `Killed` vs everything else: a child that +/// died on a signal was cut off mid-turn by something outside this daemon +/// (the kernel's OOM killer, a `systemctl stop`, an operator's `kill`) or by +/// an `interrupt` call here — either way its work stopped wherever it got +/// to, which is not what `Complete` means. +#[derive(Debug, Clone, PartialEq, Eq)] +enum TurnEnd { + /// The turn ran to completion. + Complete, + /// The child was terminated by `signal` instead of exiting on its own. + Killed { signal: i32 }, + /// claude exited on its own but did not complete the turn — a + /// recognized sentinel (rate limit, prompt too long, …) or a plain + /// non-zero exit. Carries the message shown to the caller. + Failed(String), +} + +/// Name a signal number for a human: `9` reads as `SIGKILL (signal 9)`. Only +/// the three that end a subagent in practice are named — anything else keeps +/// the number, which is still enough to look up. +fn describe_signal(signal: i32) -> String { + match signal { + libc::SIGKILL => "SIGKILL (signal 9)".to_owned(), + libc::SIGTERM => "SIGTERM (signal 15)".to_owned(), + libc::SIGINT => "SIGINT (signal 2)".to_owned(), + other => format!("signal {other}"), + } +} + +/// Read how the turn ended out of what the driver returned. A signalled +/// child surfaces as [`hive_claude::Error::Exit`] whose `ExitStatus` has a +/// `signal()` — the driver's own docs point at exactly this check — so the +/// "how" is preserved by the time it reaches here rather than having to be +/// recovered. +fn classify_end(outcome: hive_claude::Result<()>) -> TurnEnd { + let Err(error) = outcome else { + return TurnEnd::Complete; + }; + if let hive_claude::Error::Exit { status, .. } = &error + && let Some(signal) = status.signal() + { + return TurnEnd::Killed { signal }; + } + TurnEnd::Failed(format!("claude error: {error}")) +} + /// This daemon's whole state: which names currently have a live process or /// a reservation in flight, and where to push the completion todo. /// `Arc`-wrapped so the background task that drives a turn to completion @@ -46,6 +106,12 @@ use hive_claude::{Attach, Cancel, Claude, Config, NoopSink, SessionStore}; /// `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 +/// signal rather than on its own, keyed to the signal number. Without it a +/// killed session is indistinguishable from a finished one the moment its +/// entry leaves `running` — the whole point of this record. +/// /// ⚠️ This concurrency guard is keyed by `name` alone — a `start`/`continue` /// for `name` with a *different* `dir` than one already in flight under /// that name is refused as "already running," even though the two would @@ -56,6 +122,7 @@ use hive_claude::{Attach, Cancel, Claude, Config, NoopSink, SessionStore}; pub struct State { running: Mutex>>, dirs: Mutex>, + killed: Mutex>, socket: PathBuf, } @@ -65,6 +132,7 @@ impl State { Self { running: Mutex::new(HashMap::new()), dirs: Mutex::new(HashMap::new()), + killed: Mutex::new(HashMap::new()), socket, } } @@ -107,6 +175,48 @@ impl State { } } + /// Retire the finished turn's tracking for `name` and remember how it + /// ended: a `Killed` end is recorded so `status`, the end-of-turn todo + /// and a later `continue` can all say so; any other end clears a stale + /// record from an earlier turn. + /// + /// The kill is recorded *before* the `running` entry goes away, so there + /// is no instant in which `name` is neither running nor known-killed — + /// a `status` landing between the two would otherwise read the session + /// as plainly idle, which is the exact confusion this record exists to + /// remove. + fn finish_turn(&self, name: &str, end: &TurnEnd) { + let mut killed = self.killed.lock().unwrap_or_else(PoisonError::into_inner); + match end { + TurnEnd::Killed { signal } => killed.insert(name.to_owned(), *signal), + TurnEnd::Complete | TurnEnd::Failed(_) => killed.remove(name), + }; + drop(killed); + self.running + .lock() + .unwrap_or_else(PoisonError::into_inner) + .remove(name); + } + + /// The signal `name`'s last turn died on, if it died on one. + fn killed_by(&self, name: &str) -> Option { + self.killed + .lock() + .unwrap_or_else(PoisonError::into_inner) + .get(name) + .copied() + } + + /// Forget that `name`'s last turn was killed — called once a new turn is + /// confirmed spawned, since the record describes the turn before it and + /// would otherwise keep flagging a session that has since run again. + fn clear_kill(&self, name: &str) { + self.killed + .lock() + .unwrap_or_else(PoisonError::into_inner) + .remove(name); + } + /// An explicit `dir` is remembered against `name` and returned as-is; an /// omitted one (`None`) falls back to whatever was last remembered for /// `name`, so a `start` that gave a `dir` doesn't force every later @@ -301,6 +411,11 @@ fn start_reserved( /// at all (nothing to continue) or one already running (same /// concurrent-run hazard as `start`). /// +/// Resuming a session whose last turn was *killed* is allowed — that is +/// often exactly what the caller wants — but never silent: the reply says +/// so, since a caller that never ran `status` and missed the todo would +/// otherwise carry on from cut-off work believing it was finished work. +/// /// # Errors /// /// No session under `name`, an invalid name, one already running, or @@ -318,6 +433,9 @@ pub fn continue_( "subagent `{name}` is already running — use `interrupt` first if you meant to redirect it" ); } + // Read before the spawn, which clears the record as soon as the new turn + // is confirmed — this reply is the last chance to mention it. + let killed = state.killed_by(name); // Only commit the remembered `dir` now that `reserve` has actually // claimed `name` — see `resolve_dir`'s doc for why the order matters. let dir = state.resolve_dir(name, dir); @@ -325,7 +443,21 @@ pub fn continue_( if result.is_err() { state.release_reservation(name); } - result + result.map(|msg| note_resumed_after_kill(&msg, killed)) +} + +/// Append the "you are resuming a killed session" note to `continue`'s reply +/// when its previous turn was signalled; pass the reply through unchanged +/// otherwise. +fn note_resumed_after_kill(msg: &str, killed: Option) -> String { + match killed { + None => msg.to_owned(), + Some(signal) => format!( + "{msg} — note: its previous turn was killed ({}) rather than finishing, so this turn \ + resumes from work that was cut off mid-way", + describe_signal(signal) + ), + } } /// The slow, fallible part of `continue_`, run only after `reserve` has @@ -379,45 +511,79 @@ fn spawn_and_track( .lock() .unwrap_or_else(PoisonError::into_inner) .insert(name.to_owned(), Some(running.cancel_handle())); + // This turn supersedes whatever the previous one did, including having + // been killed — the record is about the turn before this one. + state.clear_kill(name); let state = Arc::clone(state); let task_name = name.to_owned(); tokio::spawn(async move { - let outcome = running.wait(&prompt, &NoopSink).await; - state - .running - .lock() - .unwrap_or_else(PoisonError::into_inner) - .remove(&task_name); - let summary = match outcome { - Ok(()) => "turn complete".to_owned(), - Err(e) => { - tracing::warn!(name = %task_name, error = %e, "subagent: turn failed"); - format!("claude error: {e}") + let end = classify_end(running.wait(&prompt, &NoopSink).await); + match &end { + TurnEnd::Complete => {} + TurnEnd::Killed { signal } => { + tracing::warn!( + name = %task_name, + signal, + "subagent: turn killed — the child died on a signal, it did not finish" + ); } - }; - push_completion_todo(&state.socket, &task_name, &summary).await; + TurnEnd::Failed(e) => { + tracing::warn!(name = %task_name, error = %e, "subagent: turn failed"); + } + } + state.finish_turn(&task_name, &end); + push_turn_end_todo(&state.socket, &task_name, &end).await; }); Ok(format!("subagent `{name}` started")) } /// Report whether `name` is currently running — a zero-cost check that -/// never launches a process, unlike `continue`. Distinguishes four states: +/// never launches a process, unlike `continue`. Distinguishes five states: /// running, starting (reserved, not yet a confirmed spawn — see `State`'s -/// doc), idle (a session exists but nothing is in flight), and no such -/// session at all. +/// doc), killed (its last turn died on a signal), idle (a session exists, +/// its last turn finished, nothing is in flight), and no such session at +/// all. /// /// # Errors /// -/// An invalid name, or no session — running or on disk — under `name`. +/// An invalid name, or no session — running, killed or on disk — under +/// `name`. pub fn status(state: &State, name: &str, dir: Option<&str>) -> anyhow::Result { validate_name(name)?; // Read-only: an explicit `dir` here is a one-off "check this other // directory's session" per this fn's own doc, not a new remembered // default — `peek_dir` resolves the same way but never writes. let dir = state.peek_dir(name, dir); - match state.occupancy(name) { + let occupancy = state.occupancy(name); + let killed = state.killed_by(name); + // Nothing on disk to look for while something is tracked in memory — + // those states answer on their own, and the store read is the only + // expensive part of this call. + let session_exists = if occupancy.is_none() && killed.is_none() { + let config = build_config(name, None, None, dir.as_deref()); + build_store(&config)?.find_by_title(name).is_some() + } else { + false + }; + describe_status(name, occupancy, killed, session_exists) +} + +/// Render `status`'s answer from the three facts it gathers. Split out from +/// the gathering so the killed-versus-idle distinction is exercisable +/// without a real spawn, a real signal and a real on-disk claude session. +/// +/// A recorded kill outranks the on-disk session: the session file exists +/// either way, so "a session is there" is precisely the fact that cannot +/// tell the two apart. +fn describe_status( + name: &str, + occupancy: Option, + killed: Option, + session_exists: bool, +) -> anyhow::Result { + match occupancy { Some(true) => return Ok(format!("subagent `{name}` is running")), Some(false) => { return Ok(format!( @@ -426,9 +592,16 @@ pub fn status(state: &State, name: &str, dir: Option<&str>) -> anyhow::Result {} } - let config = build_config(name, None, None, dir.as_deref()); - let store = build_store(&config)?; - if store.find_by_title(name).is_some() { + if let Some(signal) = killed { + return Ok(format!( + "subagent `{name}` was killed — its last turn died on {}, so its work stopped \ + wherever it had got to rather than finishing. `continue` still resumes it, but \ + whatever it was told to do is unfinished: check what it actually left behind before \ + trusting it.", + describe_signal(signal) + )); + } + if session_exists { Ok(format!( "subagent `{name}` is idle — its last turn finished; `continue` to give it another" )) @@ -467,20 +640,37 @@ pub fn interrupt(state: &State, name: &str, force: bool) -> anyhow::Result String { + match end { + TurnEnd::Complete => format!("subagent `{name}` finished: turn complete"), + TurnEnd::Failed(e) => format!("subagent `{name}` finished: {e}"), + TurnEnd::Killed { signal } => format!( + "subagent `{name}` was KILLED mid-turn ({}) — it did not finish, and its work stopped \ + wherever it had got to. Check what it left behind before you act on it; `continue` \ + resumes the session if you want it carried on.", + describe_signal(*signal) + ), } } @@ -671,6 +861,144 @@ mod tests { ); } + /// The error the driver hands back for a child that died on `signal` — + /// `ExitStatus::from_raw` takes a raw `wait(2)` status, whose low seven + /// bits are the terminating signal, so this is the same value + /// `RunningClaude::wait` would have produced from a real kill. + fn signalled_exit(signal: i32) -> hive_claude::Error { + hive_claude::Error::Exit { + status: std::process::ExitStatus::from_raw(signal), + stderr_tail: String::new(), + } + } + + #[test] + fn classify_end_separates_a_signalled_child_from_a_clean_one() { + assert_eq!(classify_end(Ok(())), TurnEnd::Complete); + assert_eq!( + classify_end(Err(signalled_exit(libc::SIGKILL))), + TurnEnd::Killed { signal: 9 }, + "a SIGKILLed child must not read as a turn that ended on its own" + ); + assert_eq!( + classify_end(Err(signalled_exit(libc::SIGTERM))), + TurnEnd::Killed { signal: 15 } + ); + assert!( + matches!( + classify_end(Err(hive_claude::Error::PromptTooLong)), + TurnEnd::Failed(_) + ), + "claude exiting on its own is a failure, not a kill" + ); + } + + #[test] + fn status_reports_killed_not_idle_for_a_signalled_session() { + // Both sessions exist on disk and neither is running: the only thing + // that can tell them apart is the recorded signal. + let killed = describe_status("n", None, Some(libc::SIGKILL), true).expect("killed status"); + let idle = describe_status("n", None, None, true).expect("idle status"); + assert!( + killed.contains("killed") && killed.contains("SIGKILL (signal 9)"), + "a killed session must say so, and name the signal: {killed}" + ); + assert!( + !killed.contains("idle"), + "a killed session must not also read as idle: {killed}" + ); + assert!(idle.contains("idle"), "a finished turn still reads idle"); + assert_ne!( + killed, idle, + "collapsing the two back together is the bug this reports" + ); + } + + #[test] + fn a_killed_turn_and_a_clean_one_do_not_land_in_the_same_state() { + // The whole path a real turn takes, minus the process: what the + // driver returned -> what the daemon records -> what `status` says. + let state = State::new(PathBuf::from("/dev/null")); + for (name, outcome) in [ + ("gone", Err(signalled_exit(libc::SIGKILL))), + ("done", Ok(())), + ] { + assert!(state.reserve(name)); + state.finish_turn(name, &classify_end(outcome)); + assert_eq!( + state.occupancy(name), + None, + "the turn is over either way — nothing stays tracked as running" + ); + } + assert_eq!(state.killed_by("gone"), Some(9)); + assert_eq!(state.killed_by("done"), None); + + let gone = describe_status("gone", None, state.killed_by("gone"), true).expect("status"); + let done = describe_status("done", None, state.killed_by("done"), true).expect("status"); + assert!(gone.contains("killed"), "{gone}"); + assert!(done.contains("idle"), "{done}"); + } + + #[test] + fn a_new_turn_clears_the_previous_turn_s_kill() { + let state = State::new(PathBuf::from("/dev/null")); + state.finish_turn("n", &TurnEnd::Killed { signal: 9 }); + assert_eq!(state.killed_by("n"), Some(9)); + // What `spawn_and_track` does once the next turn is confirmed + // spawned, and then what its own clean end does. + state.clear_kill("n"); + assert_eq!(state.killed_by("n"), None); + state.finish_turn("n", &TurnEnd::Complete); + assert_eq!( + state.killed_by("n"), + None, + "a turn that finished must not leave the old kill standing" + ); + } + + #[test] + fn the_killed_todo_does_not_read_like_a_completion() { + let complete = turn_end_summary("n", &TurnEnd::Complete); + let killed = turn_end_summary("n", &TurnEnd::Killed { signal: 9 }); + assert_eq!( + complete, "subagent `n` finished: turn complete", + "the ordinary completion todo is unchanged" + ); + assert!( + killed.contains("KILLED mid-turn") && killed.contains("SIGKILL (signal 9)"), + "the killed todo must state the kill and the signal: {killed}" + ); + assert!( + !killed.contains("finished"), + "the one notification pushed without being asked must not read as a finished turn: \ + {killed}" + ); + assert_ne!(complete, killed); + } + + #[test] + fn continue_tells_the_caller_it_is_resuming_a_killed_session() { + let plain = note_resumed_after_kill("subagent `n` started", None); + assert_eq!( + plain, "subagent `n` started", + "an ordinary resume is unchanged" + ); + let after_kill = note_resumed_after_kill("subagent `n` started", Some(libc::SIGKILL)); + assert!( + after_kill.contains("killed") && after_kill.contains("SIGKILL (signal 9)"), + "resuming a killed session is allowed, but the caller has to be told: {after_kill}" + ); + } + + #[test] + fn describe_signal_names_the_ones_that_end_a_subagent() { + assert_eq!(describe_signal(9), "SIGKILL (signal 9)"); + assert_eq!(describe_signal(15), "SIGTERM (signal 15)"); + assert_eq!(describe_signal(2), "SIGINT (signal 2)"); + assert_eq!(describe_signal(7), "signal 7"); + } + #[test] fn build_config_sets_cwd_only_when_a_dir_is_given() { let with = build_config("n", None, None, Some("/tmp/some-worktree"));