From 307df77948e99686314f20bb7ef1e64abde9ece5 Mon Sep 17 00:00:00 2001 From: atlas Date: Mon, 14 Sep 2026 19:53:05 +0200 Subject: [PATCH 1/3] subagent: report turn liveness, and stop pre-checking `continue` MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `status` could only answer running / starting / idle / killed / none, because every turn ran against `&NoopSink` and the whole stream-json stream was discarded. "Running" describes a wedged subagent exactly as well as a busy one, leaving a caller to tell them apart from `ps` output and CPU-time deltas. So the daemon now keeps a `name -> last_event_at` clock, bumped by `LivenessSink` on every line of every stream — stream-json events, plain stdout chatter and stderr alike — and `status` reports its age on a running answer: a few seconds means working, an age climbing into the minutes with no end-of-turn todo means wedged. Nothing is read out of the content; classifying *what* a subagent is doing is a separate question and waits on its own driver work. In memory with the rest of this daemon's state, dropped when the turn ends, no persistence. The clock is seeded at the spawn rather than at the first line, so a subagent that wedged before emitting anything still reports a climbing age rather than no age at all — the case an age is worth most in. Separately, `continue`'s existence pre-check is gone. It could only repeat the lookup `Claude::spawn` was about to do, and its message — "no session named `x` exists" — was false in the common failure: the session existed, just not under the claude home + cwd `build_store` resolved from. claude's own `--resume` is the authority and exits non-zero (`does not match any session title`) rather than quietly starting a fresh session, so the turn fails on its own. `classify_end` appends the one fact the CLI's message lacks — the directory searched: claude error: no session matched the requested id or title (searched for cwd ; if the session was started elsewhere, pass `dir`) The `dirs` map's durability is untouched; whether to persist it stays an open operator decision. Module doc, `docs/tools/subagent.md`, the `continue`/`status` tool descriptions and the `base:claude-subagents` skill all updated — including `continue`'s `dir` doc, which said "the daemon remembers it" without saying that a restart is both when it forgets and when you most want it. Refs #4330 Refs #4405 --- .../base/skills/claude-subagents/SKILL.md | 5 +- docs/tools/subagent.md | 57 ++- hive-subagent-mcp/src/mcp.rs | 20 +- hive-subagent-mcp/src/session.rs | 464 ++++++++++++++++-- 4 files changed, 490 insertions(+), 56 deletions(-) diff --git a/claude-plugins/plugins/base/skills/claude-subagents/SKILL.md b/claude-plugins/plugins/base/skills/claude-subagents/SKILL.md index 05f6191a..3d4081d8 100644 --- a/claude-plugins/plugins/base/skills/claude-subagents/SKILL.md +++ b/claude-plugins/plugins/base/skills/claude-subagents/SKILL.md @@ -33,7 +33,10 @@ interrupt(name, force?) `start` and `continue` return as soon as the process is confirmed running, not once it finishes — a completion lands as a todo (`get_loose_ends`), same as any other producer. Use `status` for a -zero-cost "is it still going" check; reach for `continue` only once you +zero-cost "is it still going" check — for a running subagent it also +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). diff --git a/docs/tools/subagent.md b/docs/tools/subagent.md index 23d67ebf..6dc02208 100644 --- a/docs/tools/subagent.md +++ b/docs/tools/subagent.md @@ -26,12 +26,57 @@ Served under the `subagent` MCP server (`mcp__subagent__`): `start`, ## State -In-memory only: a map of currently running processes, live only as long -as the daemon process is. A daemon restart stops whatever was running -rather than adopting it. The durable record of a subagent's existence is -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. +In-memory only: what's running now, where each name's session lives, how +each name's last turn ended, and when each running turn last produced +output. All of it lives only as long as the daemon process does. A daemon +restart stops whatever was running rather than adopting it. The durable +record of a subagent's existence is 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. + +Because the remembered directory goes with the rest of it, a `continue` +after a restart has to re-supply `dir` when the session lives anywhere +other than the daemon's own working directory — and a restart is the +situation you reach for `continue` in most often. + +## Is it working, or is it wedged? + +`status` reporting **running** says a process is tracked, which a wedged +subagent satisfies as fully as a busy one. A running answer therefore +carries the age of that turn's last event too: seconds means it's working, +an age climbing into the minutes with no end-of-turn todo means it's +stuck. That one number replaces inferring the same thing from `ps` output +and CPU-time deltas. + +Every line the subagent's `claude` process writes bumps the timestamp — +stream-json events, plain stdout chatter and stderr alike — and nothing +about the content is inspected. The record says the child is alive, not +what it's doing. The clock starts at the spawn, so a subagent that wedged +before it ever emitted anything still reports a climbing age rather than +no age at all. It's dropped when the turn ends, since a finished turn has +no progress left to describe. + +## A `continue` that finds no session + +`continue` doesn't check for the session before spawning. claude's own +`--resume` is the authority, and it exits non-zero rather than quietly +starting a fresh session, so the check could only duplicate the lookup +the driver was about to do — while answering as though the session were +gone. The usual truth is that the session exists somewhere else. + +`continue` returns as soon as the process is confirmed running, same as +always, so the failure lands where every other failed turn lands: that +turn's end-of-turn todo, carrying claude's own message plus the location +this daemon searched. + +``` +claude error: no session matched the requested id or title (searched /home/agent/.claude for cwd /home/agent/work; if the session was started elsewhere, pass `dir`) +``` + +The directory is the part claude's own message never names, and the part +that resolves the confusion: pass `dir` to point `continue` at the +directory the session was started in. ## A killed turn diff --git a/hive-subagent-mcp/src/mcp.rs b/hive-subagent-mcp/src/mcp.rs index 511e2bb0..4fca71a0 100644 --- a/hive-subagent-mcp/src/mcp.rs +++ b/hive-subagent-mcp/src/mcp.rs @@ -79,8 +79,12 @@ struct ContinueArgs { #[serde(default)] effort: Option, /// Omit to reuse whatever `dir` `start` (or a prior `continue`) used for - /// this name — the daemon remembers it. Only pass this to point the - /// session at a *different* directory than last time. + /// this name — the daemon remembers it until it restarts, and a restart + /// is exactly when you're most likely to be reaching for `continue`. So + /// pass it when pointing the session at a *different* directory than + /// last time, and pass it again after a restart if the session lives + /// anywhere other than the daemon's own working directory. A resume that + /// finds nothing says which directory it searched. #[serde(default)] dir: Option, } @@ -146,9 +150,11 @@ 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. \ - 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." + `start`; a name already running is refused. A name with no session to resume is not \ + refused up front — claude's own `--resume` decides that, and a miss arrives as a \ + failed turn naming the directory that was searched, so check `dir` before concluding \ + the session is gone. 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_( @@ -181,7 +187,9 @@ impl SubagentMcp { #[tool( description = "Report whether a subagent is currently running — a zero-cost check that \ never launches a process, unlike `continue`. The answer says what state it found \ - and what to do about it." + and what to do about it. For a running one it also reports how long since that turn \ + last produced any output, which is how you tell a subagent that's working from one \ + that has wedged without resorting to `ps`." )] 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 dd1cd51c..d81e13bc 100644 --- a/hive-subagent-mcp/src/session.rs +++ b/hive-subagent-mcp/src/session.rs @@ -4,17 +4,30 @@ //! //! **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 -//! the `None`/`Some` split is for) plus a `name -> dir` memory so `start`'s +//! the `None`/`Some` split is for), a `name -> dir` memory so `start`'s //! `dir` doesn't have to be repeated on every later `continue`/`status` -//! (`State::resolve_dir`) — both live 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 the -//! remembered `dir` is gone too. The durable -//! record of a subagent's existence is `hive_claude::SessionStore` — claude's -//! own on-disk session, found again by name. `continue` is how a caller -//! reattaches to it, whether that's "give it a new turn" or "the daemon -//! 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. +//! (`State::resolve_dir`), and a `name -> last_event_at` liveness clock +//! (`State::note_event`) — 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 +//! the remembered `dir` is gone too. The durable record of a subagent's +//! existence is `hive_claude::SessionStore` — claude's own on-disk +//! session, found again by name. `continue` is how a caller reattaches to +//! it, whether that's "give it a new turn" or "the daemon 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 running turn is not necessarily a working turn.** A wedged child +//! satisfies "running" as fully as a busy one, so every line of the +//! child's streams bumps `last_event_at` (`LivenessSink`) and `status` +//! reports its age. Deliberately *unclassified* — the timestamp says the +//! child is alive, not what it's doing. +//! +//! **`continue` doesn't pre-check the session's existence.** claude's own +//! `--resume` is the authority and errors rather than silently starting a +//! fresh session; `classify_end` appends the one fact its message lacks — +//! the directory searched (`searched_location`), which is where this +//! actually goes wrong. See `docs/tools/subagent.md` for both. //! **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 @@ -40,8 +53,9 @@ use std::collections::HashMap; use std::os::unix::process::ExitStatusExt as _; use std::path::PathBuf; use std::sync::{Arc, Mutex, PoisonError}; +use std::time::{Duration, Instant}; -use hive_claude::{Attach, Cancel, Claude, Config, NoopSink, SessionStore}; +use hive_claude::{Attach, Cancel, Claude, Config, SessionStore}; /// How a subagent's turn ended, as far as the daemon can tell from what /// [`hive_claude::RunningClaude::wait`] returned. @@ -80,7 +94,15 @@ fn describe_signal(signal: i32) -> String { /// `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 { +/// +/// `searched` is the resume lookup's location, present only for a turn that +/// attached with [`Attach::Resume`] (see `searched_location`). A missed +/// resume arrives as [`hive_claude::Error::SessionNotFound`], whose message +/// names the *value* that matched nothing but not the *directory* it was +/// looked for in — and "the session lives in another directory" is the way +/// this actually fails in practice, so the caller gets told where the daemon +/// looked rather than left to conclude the session is gone. +fn classify_end(outcome: hive_claude::Result<()>, searched: Option<&str>) -> TurnEnd { let Err(error) = outcome else { return TurnEnd::Complete; }; @@ -89,9 +111,31 @@ fn classify_end(outcome: hive_claude::Result<()>) -> TurnEnd { { return TurnEnd::Killed { signal }; } + if matches!(error, hive_claude::Error::SessionNotFound) + && let Some(searched) = searched + { + return TurnEnd::Failed(format!("claude error: {error} {searched}")); + } TurnEnd::Failed(format!("claude error: {error}")) } +/// Where a `--resume` against `config` would have looked for the session, +/// phrased to append to claude's own not-found message. Both halves come +/// from the same resolution `build_store` and the driver itself use, so this +/// names the directory that was genuinely searched rather than a guess at +/// it. `None` when either half fails to resolve (no `HOME`, a `cwd` that +/// doesn't exist) — an unresolvable location is worse than none, and the +/// underlying error still reaches the caller unadorned. +fn searched_location(config: &Config) -> Option { + let home = config.resolved_claude_home().ok()?; + let cwd = config.resolved_cwd().ok()?; + Some(format!( + "(searched {} for cwd {}; if the session was started elsewhere, pass `dir`)", + home.display(), + cwd.display() + )) +} + /// 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 @@ -111,6 +155,10 @@ fn classify_end(outcome: hive_claude::Result<()>) -> TurnEnd { /// killed session is indistinguishable from a finished one the moment its /// entry leaves `running` — the whole point of this record. /// +/// `last_event` splits `running` a second way — *in flight* against +/// *making progress*, since a wedged child is tracked exactly like a busy +/// one (`note_event` / `last_event_age`). +/// /// ⚠️ 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 @@ -122,6 +170,7 @@ pub struct State { running: Mutex>>, dirs: Mutex>, killed: Mutex>, + last_event: Mutex>, socket: PathBuf, } @@ -132,6 +181,7 @@ impl State { running: Mutex::new(HashMap::new()), dirs: Mutex::new(HashMap::new()), killed: Mutex::new(HashMap::new()), + last_event: Mutex::new(HashMap::new()), socket, } } @@ -184,6 +234,10 @@ impl State { /// a `status` landing between the two would otherwise read the session /// as plainly idle, which is the exact confusion this record exists to /// remove. + /// + /// The liveness clock goes with the `running` entry, for the same reason + /// it's kept at all: it answers "is this turn still making progress", + /// and a turn that has ended has no progress left to make. fn finish_turn(&self, name: &str, end: &TurnEnd) { let mut killed = self.killed.lock().unwrap_or_else(PoisonError::into_inner); match end { @@ -195,6 +249,10 @@ impl State { .lock() .unwrap_or_else(PoisonError::into_inner) .remove(name); + self.last_event + .lock() + .unwrap_or_else(PoisonError::into_inner) + .remove(name); } /// The signal `name`'s last turn died on, if it died on one. @@ -216,6 +274,42 @@ impl State { .remove(name); } + /// `name`'s turn just produced output — record *when*, and nothing else. + /// Called from `LivenessSink` for every line of every stream, so it runs + /// far more often than anything else in this type and stays a single + /// map write for that reason. + /// + /// A monotonic [`Instant`], not a wall clock: what's ever read back out + /// is an elapsed age, which a clock adjustment must not be able to + /// distort into a stall that never happened. + /// + /// Also called once by `spawn_and_track` the moment the child exists, so + /// the clock starts at the spawn rather than at the first line. Without + /// that seed a subagent that wedged before emitting anything at all + /// would report no age forever — the one case where an age is most worth + /// having. The value therefore reads as "how long since the daemon last + /// heard anything from this child", counting the spawn itself as the + /// first thing it heard. + fn note_event(&self, name: &str) { + self.last_event + .lock() + .unwrap_or_else(PoisonError::into_inner) + .insert(name.to_owned(), Instant::now()); + } + + /// How long since `name`'s turn last produced output. `None` once the + /// turn is over (`finish_turn` drops the entry) or for a name that never + /// reached a spawn — in both cases there's no *running* turn whose + /// progress the age would describe, and a leftover age from a turn that + /// has already ended would read as a stall that isn't one. + fn last_event_age(&self, name: &str) -> Option { + self.last_event + .lock() + .unwrap_or_else(PoisonError::into_inner) + .get(name) + .map(Instant::elapsed) + } + /// 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 @@ -425,10 +519,18 @@ fn start_reserved( /// Give an existing named session a new turn — resuming it whether that /// means "the previous turn finished, here's the next instruction" or "the -/// daemon restarted, reattaching." Refuses a name with no session on disk -/// at all (nothing to continue) or one already running (same +/// daemon restarted, reattaching." Refuses a name already running (same /// concurrent-run hazard as `start`). /// +/// A name with **no** session to resume is not refused here: claude's own +/// `--resume` answers that, and its answer arrives with the turn rather +/// than before it — as a `SessionNotFound` failure carrying the directory +/// that was searched (`classify_end`), reaching the caller through the +/// end-of-turn todo and `status` like any other failed turn. The pre-check +/// that used to live here could only repeat the lookup the driver was about +/// to do anyway, and told the caller a session didn't exist when the true +/// answer was almost always that it exists somewhere else. +/// /// 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 @@ -436,8 +538,7 @@ fn start_reserved( /// /// # Errors /// -/// No session under `name`, an invalid name, one already running, or -/// `Claude::spawn` failing. +/// An invalid name, one already running, or `Claude::spawn` failing. pub fn continue_( state: &Arc, name: &str, @@ -491,13 +592,9 @@ fn continue_reserved( dir: Option<&str>, ) -> anyhow::Result { let config = build_config(name, model, effort, None, dir); - let store = build_store(&config)?; - if store.find_by_title(name).is_none() { - anyhow::bail!( - "no session named `{name}` exists — `continue` resumes an existing subagent, `start` \ - creates one" - ); - } + // No existence pre-check: claude's own `--resume` is the authority on + // whether the session is there, and it errors rather than quietly + // starting a fresh one. See this module's doc. spawn_and_track( state, name, @@ -507,6 +604,41 @@ fn continue_reserved( ) } +/// Bumps `name`'s liveness clock on every line of the turn's output, and +/// does nothing else with it. Replaces the `NoopSink` this daemon used to +/// run turns against, which discarded the stream wholesale and left `status` +/// unable to tell a working child from a wedged one. +/// +/// **All three callbacks, deliberately.** A stderr line or a stdout line +/// that didn't parse as JSON is proof the child is alive every bit as much +/// as a stream-json event is, and the failure that matters here is reporting +/// a live subagent as wedged — so anything the child says counts. Nothing is +/// read out of the content: classifying *what* the subagent is doing is a +/// separate question from whether it's doing anything. +/// +/// Sink methods are called synchronously from the driver's stream readers as +/// lines arrive, so the body has to stay cheap — one uncontended map write +/// is, and forwarding to a channel to do the same write elsewhere would cost +/// more than it saved. +struct LivenessSink { + state: Arc, + name: String, +} + +impl hive_claude::Sink for LivenessSink { + fn on_event(&self, _event: &serde_json::Value) { + self.state.note_event(&self.name); + } + + fn on_stdout_line(&self, _line: &str) { + self.state.note_event(&self.name); + } + + fn on_stderr_line(&self, _line: &str) { + self.state.note_event(&self.name); + } +} + /// Spawn the child (synchronous — returns with a real pid the instant the /// process exists, which *is* "confirmed running": there is no stronger /// signal to wait for without slowing every call down for no reason), track @@ -534,11 +666,25 @@ fn spawn_and_track( // 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); + // Start the liveness clock at the spawn, so the age is already an answer + // before the child's first line — see `note_event`. + state.note_event(name); + + // Resolved here, on the calling thread, while the config is still to + // hand: only a resume can miss, and only `classify_end` finding a + // `SessionNotFound` ever uses it. + let searched = matches!(attach, Attach::Resume(_)) + .then(|| searched_location(config)) + .flatten(); let state = Arc::clone(state); let task_name = name.to_owned(); tokio::spawn(async move { - let end = classify_end(running.wait(&prompt, &NoopSink).await); + let sink = LivenessSink { + state: Arc::clone(&state), + name: task_name.clone(), + }; + let end = classify_end(running.wait(&prompt, &sink).await, searched.as_deref()); match &end { TurnEnd::Complete => {} TurnEnd::Killed { signal } => { @@ -566,6 +712,11 @@ fn spawn_and_track( /// its last turn finished, nothing is in flight), and no such session at /// all. /// +/// A *running* answer also carries how long since that turn last produced +/// output, which is the part of this answer a caller can act on: "running" +/// describes a wedged child and a busy one identically, and the age +/// separates them (see `State`'s `last_event` doc). +/// /// # Errors /// /// An invalid name, or no session — running, killed or on disk — under @@ -578,6 +729,7 @@ pub fn status(state: &State, name: &str, dir: Option<&str>) -> anyhow::Result) -> anyhow::Result) -> String { + match age { + None => String::new(), + Some(age) => format!( + " Last event {}s ago — how long since this turn's claude process produced any output \ + at all, whatever it was: a few seconds means it's working, an age that keeps \ + climbing into the minutes with no end-of-turn todo means it's wedged.", + age.as_secs() + ), + } +} + +/// Render `status`'s answer from the facts it gathers. Split out from the +/// gathering so the killed-versus-idle distinction — and the liveness age on +/// a running turn — are exercisable without a real spawn, a real signal, a +/// real stream of events 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 @@ -608,13 +782,15 @@ fn describe_status( occupancy: Option, killed: Option, session_exists: bool, + last_event_age: Option, ) -> anyhow::Result { match occupancy { Some(true) => { 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 turn ends, or \ - `interrupt` it if you want it stopped early." + `interrupt` it if you want it stopped early.{}", + describe_liveness(last_event_age) )); } Some(false) => { @@ -915,19 +1091,19 @@ mod tests { #[test] fn classify_end_separates_a_signalled_child_from_a_clean_one() { - assert_eq!(classify_end(Ok(())), TurnEnd::Complete); + assert_eq!(classify_end(Ok(()), None), TurnEnd::Complete); assert_eq!( - classify_end(Err(signalled_exit(libc::SIGKILL))), + classify_end(Err(signalled_exit(libc::SIGKILL)), None), 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))), + classify_end(Err(signalled_exit(libc::SIGTERM)), None), TurnEnd::Killed { signal: 15 } ); assert!( matches!( - classify_end(Err(hive_claude::Error::PromptTooLong)), + classify_end(Err(hive_claude::Error::PromptTooLong), None), TurnEnd::Failed(_) ), "claude exiting on its own is a failure, not a kill" @@ -938,8 +1114,9 @@ mod tests { 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"); + let killed = + describe_status("n", None, Some(libc::SIGKILL), true, None).expect("killed status"); + let idle = describe_status("n", None, None, true, None).expect("idle status"); assert!( killed.contains("killed") && killed.contains("SIGKILL (signal 9)"), "a killed session must say so, and name the signal: {killed}" @@ -960,28 +1137,30 @@ mod tests { // The tool description no longer lists the states, so each answer // has to carry its own explanation — checked one state at a time, // which is all a caller ever gets back. - let running = describe_status("n", Some(true), None, false).expect("running status"); + let running = describe_status("n", Some(true), None, false, None).expect("running status"); assert!( running.contains("running") && running.contains("`interrupt`"), "a running answer must say the turn is in flight and how to stop it: {running}" ); - let starting = describe_status("n", Some(false), None, false).expect("starting status"); + let starting = + describe_status("n", Some(false), None, false, None).expect("starting status"); assert!( starting.contains("starting") && starting.contains("check again"), "a starting answer must say the spawn isn't confirmed yet and to retry: {starting}" ); - let idle = describe_status("n", None, None, true).expect("idle status"); + let idle = describe_status("n", None, None, true, None).expect("idle status"); assert!( idle.contains("idle") && idle.contains("`continue`"), "an idle answer must say the last turn ended on its own and how to give it another: \ {idle}" ); - let killed = describe_status("n", None, Some(libc::SIGKILL), true).expect("killed status"); + let killed = + describe_status("n", None, Some(libc::SIGKILL), true, None).expect("killed status"); assert!( killed.contains("killed") && killed.contains("`continue`"), "a killed answer must name the kill and say resuming is still possible: {killed}" ); - let missing = describe_status("n", None, None, false) + let missing = describe_status("n", None, None, false, None) .expect_err("nothing tracked and nothing on disk is an error, not a state"); assert!( missing.to_string().contains("`start`"), @@ -999,7 +1178,7 @@ mod tests { ("done", Ok(())), ] { assert!(state.reserve(name)); - state.finish_turn(name, &classify_end(outcome)); + state.finish_turn(name, &classify_end(outcome, None)); assert_eq!( state.occupancy(name), None, @@ -1009,8 +1188,10 @@ mod tests { 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"); + let gone = + describe_status("gone", None, state.killed_by("gone"), true, None).expect("status"); + let done = + describe_status("done", None, state.killed_by("done"), true, None).expect("status"); assert!(gone.contains("killed"), "{gone}"); assert!(done.contains("idle"), "{done}"); } @@ -1096,6 +1277,203 @@ mod tests { assert_eq!(config.effort, Some("high".to_owned())); } + #[test] + fn an_event_starts_the_liveness_clock_and_the_turn_ending_stops_it() { + let state = State::new(PathBuf::from("/dev/null")); + assert_eq!( + state.last_event_age("n"), + None, + "a name that never spawned has no clock to read" + ); + state.note_event("n"); + assert!( + state.last_event_age("n").is_some(), + "the first event must give `status` an age to report" + ); + state.finish_turn("n", &TurnEnd::Complete); + assert_eq!( + state.last_event_age("n"), + None, + "the age describes a turn in flight — a finished turn's leftover age would read as a \ + stall that never happened" + ); + } + + #[test] + fn a_later_event_resets_the_age_rather_than_letting_it_climb() { + // The distinction the whole record exists for: a child still + // producing output must not accumulate the age of a wedged one. + let state = State::new(PathBuf::from("/dev/null")); + state.note_event("n"); + std::thread::sleep(Duration::from_millis(20)); + let before = state.last_event_age("n").expect("clock started"); + state.note_event("n"); + let after = state.last_event_age("n").expect("clock still running"); + assert!( + after < before, + "a fresh event must reset the age, not extend it: {after:?} vs {before:?}" + ); + } + + #[test] + fn the_liveness_clock_does_not_cross_names() { + let state = State::new(PathBuf::from("/dev/null")); + state.note_event("a"); + assert_eq!( + state.last_event_age("b"), + None, + "one subagent's output says nothing about another's" + ); + } + + #[test] + fn every_sink_callback_counts_as_liveness() { + // All three, deliberately: a stderr line or a non-JSON stdout line is + // proof the child is alive exactly as much as a stream-json event, + // and reporting a live subagent as wedged is the failure that costs. + use hive_claude::Sink as _; + + fn fresh() -> (Arc, LivenessSink) { + let state = Arc::new(State::new(PathBuf::from("/dev/null"))); + let sink = LivenessSink { + state: Arc::clone(&state), + name: "n".to_owned(), + }; + (state, sink) + } + + let (state, sink) = fresh(); + sink.on_event(&serde_json::json!({"type": "system"})); + assert!( + state.last_event_age("n").is_some(), + "a stream-json event is liveness" + ); + + let (state, sink) = fresh(); + sink.on_stdout_line("not json"); + assert!( + state.last_event_age("n").is_some(), + "so is a stdout line that didn't parse as JSON" + ); + + let (state, sink) = fresh(); + sink.on_stderr_line("some chatter"); + assert!( + state.last_event_age("n").is_some(), + "so is a stderr line — anything the child says at all counts" + ); + } + + #[test] + fn a_running_status_reports_the_age_and_an_idle_one_does_not() { + let running = describe_status("n", Some(true), None, false, Some(Duration::from_secs(4))) + .expect("running status"); + assert!( + running.contains("Last event 4s ago"), + "a running answer must carry the age — the one part of it a caller can act on: \ + {running}" + ); + let wedged = describe_status("n", Some(true), None, false, Some(Duration::from_mins(15))) + .expect("running status"); + assert!( + wedged.contains("Last event 900s ago"), + "the wedged case is the one this exists for: {wedged}" + ); + assert_ne!( + running, wedged, + "4s-ago and 900s-ago must not render identically — collapsing them back together is \ + the bug this reports" + ); + + let idle = describe_status("n", None, None, true, None).expect("idle status"); + assert!( + !idle.contains("Last event"), + "an idle session has no in-flight turn whose progress an age would describe: {idle}" + ); + } + + #[test] + fn a_running_status_still_answers_when_the_turn_ended_mid_call() { + // `status` reads `running` and the clock under separate locks, so a + // turn can finish between the two. The answer drops the age rather + // than inventing one. + let raced = describe_status("n", Some(true), None, false, None).expect("running status"); + assert!( + raced.contains("is running") && !raced.contains("Last event"), + "a missing age must cost the sentence, not the answer: {raced}" + ); + } + + #[test] + fn a_missed_resume_names_the_directory_that_was_searched() { + // The failure that cost an afternoon: the session existed, just not + // where this daemon looked. claude's own message names the value it + // failed to match but never the directory, which is the fact that + // was missing. + let searched = "(searched /home/agent/.claude for cwd /home/agent/work; if the session \ + was started elsewhere, pass `dir`)"; + let end = classify_end(Err(hive_claude::Error::SessionNotFound), Some(searched)); + let TurnEnd::Failed(msg) = end else { + panic!("a missed resume is a failed turn"); + }; + assert!( + msg.contains("no session matched"), + "claude's own diagnosis must survive: {msg}" + ); + assert!( + msg.contains("/home/agent/.claude") && msg.contains("/home/agent/work"), + "and the search location must be appended to it: {msg}" + ); + assert!( + msg.contains("pass `dir`"), + "naming the directory is only half of it — say what to do about it: {msg}" + ); + } + + #[test] + fn only_a_missed_resume_gets_the_search_location() { + // A `start` never passes one (nothing to resume), and an unrelated + // failure must not be decorated with a location that had no part in + // it. + let no_hint = classify_end(Err(hive_claude::Error::SessionNotFound), None); + assert_eq!( + no_hint, + TurnEnd::Failed( + "claude error: no session matched the requested id or title".to_owned() + ), + "with no resolvable location the underlying error still reaches the caller, plain" + ); + + let unrelated = classify_end(Err(hive_claude::Error::PromptTooLong), Some("(searched …)")); + let TurnEnd::Failed(msg) = unrelated else { + panic!("an overflowed context is a failed turn"); + }; + assert!( + !msg.contains("searched"), + "a context overflow has nothing to do with where the session lives: {msg}" + ); + } + + #[test] + fn searched_location_names_the_same_place_the_store_would_look() { + // The point of reading both halves off the `Config` rather than + // re-deriving them: the location reported must be the one actually + // searched, so it can't drift from `build_store`/the driver. + let cwd = std::env::current_dir().expect("a cwd"); + let config = build_config("n", None, None, None, Some(&cwd.to_string_lossy())); + let Some(located) = searched_location(&config) else { + // No `HOME` in this environment — nothing to compare against. + return; + }; + let home = config.resolved_claude_home().expect("home resolved"); + let resolved_cwd = config.resolved_cwd().expect("cwd resolved"); + assert!( + located.contains(&home.display().to_string()) + && located.contains(&resolved_cwd.display().to_string()), + "both halves must come from the config's own resolution: {located}" + ); + } + #[test] fn build_config_defaults_effort_to_medium_when_omitted() { // Unlike `model`, an omitted `effort` does not fall through to From 31c76ddf327bbcdb4640c5cea573acaa580b6923 Mon Sep 17 00:00:00 2001 From: atlas Date: Mon, 14 Sep 2026 20:20:58 +0200 Subject: [PATCH 2/3] subagent: say the dir a session was started in, not "pass dir" A session cannot be moved between directories, so a hint reading "pass dir" could be misread as pointing an existing session at any directory. Say instead that dir names the directory the session was started in. Refs #4405 --- docs/tools/subagent.md | 2 +- hive-subagent-mcp/src/session.rs | 8 ++++---- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/docs/tools/subagent.md b/docs/tools/subagent.md index 6dc02208..ca2c0e7c 100644 --- a/docs/tools/subagent.md +++ b/docs/tools/subagent.md @@ -71,7 +71,7 @@ turn's end-of-turn todo, carrying claude's own message plus the location this daemon searched. ``` -claude error: no session matched the requested id or title (searched /home/agent/.claude for cwd /home/agent/work; if the session was started elsewhere, pass `dir`) +claude error: no session matched the requested id or title (searched /home/agent/.claude for cwd /home/agent/work; if it was started elsewhere, pass the `dir` it was started in) ``` The directory is the part claude's own message never names, and the part diff --git a/hive-subagent-mcp/src/session.rs b/hive-subagent-mcp/src/session.rs index d81e13bc..9ae11203 100644 --- a/hive-subagent-mcp/src/session.rs +++ b/hive-subagent-mcp/src/session.rs @@ -130,7 +130,7 @@ fn searched_location(config: &Config) -> Option { let home = config.resolved_claude_home().ok()?; let cwd = config.resolved_cwd().ok()?; Some(format!( - "(searched {} for cwd {}; if the session was started elsewhere, pass `dir`)", + "(searched {} for cwd {}; if it was started elsewhere, pass the `dir` it was started in)", home.display(), cwd.display() )) @@ -1410,8 +1410,8 @@ mod tests { // where this daemon looked. claude's own message names the value it // failed to match but never the directory, which is the fact that // was missing. - let searched = "(searched /home/agent/.claude for cwd /home/agent/work; if the session \ - was started elsewhere, pass `dir`)"; + let searched = "(searched /home/agent/.claude for cwd /home/agent/work; if it was started \ + elsewhere, pass the `dir` it was started in)"; let end = classify_end(Err(hive_claude::Error::SessionNotFound), Some(searched)); let TurnEnd::Failed(msg) = end else { panic!("a missed resume is a failed turn"); @@ -1425,7 +1425,7 @@ mod tests { "and the search location must be appended to it: {msg}" ); assert!( - msg.contains("pass `dir`"), + msg.contains("pass the `dir` it was started in"), "naming the directory is only half of it — say what to do about it: {msg}" ); } From 6e2de33f2626623d284d1e25c559231aef949b47 Mon Sep 17 00:00:00 2001 From: atlas Date: Mon, 14 Sep 2026 20:34:54 +0200 Subject: [PATCH 3/3] subagent: make a missed `continue` the tool call's own error MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `continue` returned "started" the instant `Claude::spawn` handed back a pid, and a resume that matched nothing only surfaced later, as an end-of-turn todo. By then the caller had moved on believing it had a running subagent. A pid is proof enough for `start`, which creates its session: the spawn succeeding is the whole story. It is not proof for a resume — claude exits non-zero a fraction of a second *after* the process exists. So `continue` now waits for the first real answer and reports a miss as its own `Err`, carrying claude's message and the directory searched. The wait ends on whichever comes first, so a successful `continue` pays no fixed delay: the turn's first non-terminal stream event settles it at about the same moment a miss's exit would have. Measured on this box: 14 runs of the driver's own invocation against a missing session took 550-1087 ms spawn to exit, and a healthy turn's first event lands at roughly 500 ms. The five-second cap is ~4.6x the slowest miss and is only ever reached by a child that neither speaks nor exits. The underway signal reads the event's kind, not its content: a missed resume is not silent — it emits a terminal `result` event and stderr before exiting — so "any sink callback" would have reported every miss as a successful start. Liveness still counts all three callbacks. The end-of-turn todo is unchanged for every failure later in the turn; the only one it no longer repeats is the miss the caller was just handed. Refs #4405 --- .../base/skills/claude-subagents/SKILL.md | 10 +- docs/tools/subagent.md | 29 +- hive-subagent-mcp/Cargo.toml | 6 + hive-subagent-mcp/src/mcp.rs | 20 +- hive-subagent-mcp/src/session.rs | 372 ++++++++++++++++-- 5 files changed, 385 insertions(+), 52 deletions(-) diff --git a/claude-plugins/plugins/base/skills/claude-subagents/SKILL.md b/claude-plugins/plugins/base/skills/claude-subagents/SKILL.md index 3d4081d8..dc53f709 100644 --- a/claude-plugins/plugins/base/skills/claude-subagents/SKILL.md +++ b/claude-plugins/plugins/base/skills/claude-subagents/SKILL.md @@ -30,9 +30,13 @@ status(name) interrupt(name, force?) ``` -`start` and `continue` return as soon as the process is confirmed -running, not once it finishes — a completion lands as a todo -(`get_loose_ends`), same as any other producer. Use `status` for a +`start` and `continue` return once the turn is under way, not once it +finishes — a completion lands as a todo (`get_loose_ends`), same as any +other producer. `continue` takes the extra moment to confirm the resume +actually attached, so a `continue` naming a session that isn't there +fails the tool call outright instead of looking like it worked; the error +says which directory it searched, which is usually the fix (pass `dir`). +Use `status` for a zero-cost "is it still going" check — for a running subagent it also 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 diff --git a/docs/tools/subagent.md b/docs/tools/subagent.md index ca2c0e7c..1cd7184e 100644 --- a/docs/tools/subagent.md +++ b/docs/tools/subagent.md @@ -50,9 +50,9 @@ stuck. That one number replaces inferring the same thing from `ps` output and CPU-time deltas. Every line the subagent's `claude` process writes bumps the timestamp — -stream-json events, plain stdout chatter and stderr alike — and nothing -about the content is inspected. The record says the child is alive, not -what it's doing. The clock starts at the spawn, so a subagent that wedged +stream-json events, plain stdout chatter and stderr alike — and what the +subagent actually said is never read. The record says the child is alive, +not what it's doing. The clock starts at the spawn, so a subagent that wedged before it ever emitted anything still reports a climbing age rather than no age at all. It's dropped when the turn ends, since a finished turn has no progress left to describe. @@ -65,19 +65,32 @@ starting a fresh session, so the check could only duplicate the lookup the driver was about to do — while answering as though the session were gone. The usual truth is that the session exists somewhere else. -`continue` returns as soon as the process is confirmed running, same as -always, so the failure lands where every other failed turn lands: that -turn's end-of-turn todo, carrying claude's own message plus the location -this daemon searched. +`continue` waits for that answer instead. Where `start` returns the +instant the process exists — it creates its session, so the spawn +succeeding is the whole story — a resumed turn can fail a moment _after_ +a pid exists, and a pid is no proof that turn began. `continue` holds the +tool call open until the turn is underway or the resume has come back +missed, and reports a miss as the call's own error, carrying claude's +message plus the location this daemon searched: ``` -claude error: no session matched the requested id or title (searched /home/agent/.claude for cwd /home/agent/work; if it was started elsewhere, pass the `dir` it was started in) +continue error: claude error: no session matched the requested id or title (searched /home/agent/.claude for cwd /home/agent/work; if it was started elsewhere, pass the `dir` it was started in) ``` The directory is the part claude's own message never names, and the part that resolves the confusion: pass `dir` to point `continue` at the directory the session was started in. +Nothing here is a fixed delay on the way to a successful turn. The wait +ends on whichever comes first — the turn's first stream event or its +early exit — and on this box both land inside a second, so a `continue` +that works answers about as fast as it did before. A five-second cap +bounds the one case neither covers: a child that neither speaks nor +exits, reported as started, with the end-of-turn todo left to say how it +goes. That todo still carries every failure that happens later in the +turn, exactly as before; the only one it no longer repeats is the miss +the caller has just been handed to its face. + ## A killed turn A subagent whose `claude` process dies on a signal — the kernel's OOM diff --git a/hive-subagent-mcp/Cargo.toml b/hive-subagent-mcp/Cargo.toml index e6d78ef0..025735ea 100644 --- a/hive-subagent-mcp/Cargo.toml +++ b/hive-subagent-mcp/Cargo.toml @@ -24,6 +24,12 @@ tokio.workspace = true tracing.workspace = true tracing-subscriber.workspace = true +# `test-util` for `#[tokio::test(start_paused = true)]`: the `continue` +# resume-grace tests assert what happens when the bound is actually reached, +# and paused time gets that answer without a five-second unit test. +[dev-dependencies] +tokio = { workspace = true, features = ["test-util"] } + # `hive-subagent-daemon` — long-running per-agent claude-subagent runner. # Independent of `hive-bash-mcp` (own crate, own binary, own MCP server) — # see lib.rs's module doc for why. Serves its MCP tools (`start`/ diff --git a/hive-subagent-mcp/src/mcp.rs b/hive-subagent-mcp/src/mcp.rs index 4fca71a0..fd5b7bef 100644 --- a/hive-subagent-mcp/src/mcp.rs +++ b/hive-subagent-mcp/src/mcp.rs @@ -149,14 +149,16 @@ impl SubagentMcp { description = "Give an existing named subagent session a new turn — whether that's \ 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`; a name already running is refused. A name with no session to resume is not \ - refused up front — claude's own `--resume` decides that, and a miss arrives as a \ - failed turn naming the directory that was searched, so check `dir` before concluding \ - the session is gone. 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." + of the daemon that spawned it). Returns once the turn is underway rather than the \ + instant the process exists — a second or so, not the length of the turn — so a \ + reply saying the turn started means it started; a name already running is refused. \ + A name with no session to resume is not refused up front, because claude's own \ + `--resume` decides that: a miss comes back as this call's own error, naming the \ + directory that was searched, so check `dir` before concluding the session is gone. \ + 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 { + async fn r#continue(&self, Parameters(args): Parameters) -> String { match session::continue_( &self.state, &args.name, @@ -164,7 +166,9 @@ impl SubagentMcp { args.model, args.effort, args.dir.as_deref(), - ) { + ) + .await + { Ok(msg) => msg, Err(e) => format!("continue error: {e:#}"), } diff --git a/hive-subagent-mcp/src/session.rs b/hive-subagent-mcp/src/session.rs index 9ae11203..e2747ae7 100644 --- a/hive-subagent-mcp/src/session.rs +++ b/hive-subagent-mcp/src/session.rs @@ -18,16 +18,15 @@ //! the session isn't in the daemon's own default directory. //! //! **A running turn is not necessarily a working turn.** A wedged child -//! satisfies "running" as fully as a busy one, so every line of the -//! child's streams bumps `last_event_at` (`LivenessSink`) and `status` -//! reports its age. Deliberately *unclassified* — the timestamp says the -//! child is alive, not what it's doing. +//! satisfies "running" as fully as a busy one, so every line of the child's +//! streams bumps `last_event_at` (`LivenessSink`) and `status` reports its +//! age — the timestamp says the child is alive, never what it said. //! -//! **`continue` doesn't pre-check the session's existence.** claude's own -//! `--resume` is the authority and errors rather than silently starting a -//! fresh session; `classify_end` appends the one fact its message lacks — -//! the directory searched (`searched_location`), which is where this -//! actually goes wrong. See `docs/tools/subagent.md` for both. +//! **`continue` doesn't pre-check the session's existence — it waits for +//! the answer instead.** claude's own `--resume` is the authority, so +//! `continue` holds its tool call open for up to `RESUME_GRACE` and reports +//! a missed resume as its own error, naming the directory searched +//! (`classify_end`). See `docs/tools/subagent.md`. //! **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 @@ -56,6 +55,22 @@ use std::sync::{Arc, Mutex, PoisonError}; use std::time::{Duration, Instant}; use hive_claude::{Attach, Cancel, Claude, Config, SessionStore}; +use tokio::sync::oneshot; + +/// How long a `continue` holds its tool call open waiting to find out +/// whether the resume landed. A cap, not a delay: both real outcomes settle +/// it well inside this, and it is only ever reached by a child that neither +/// speaks nor exits. +/// +/// Measured rather than guessed, on this box, running the driver's own +/// invocation (`--print --verbose --output-format stream-json --resume +/// `) against a session name that matches nothing: 14 runs, +/// 550–1087 ms wall from spawn to exit. A healthy turn's first stream event +/// lands at roughly 500 ms, so the two paths finish within ~100 ms of each +/// other and neither waits on this bound. Five seconds is ~4.6× the slowest +/// miss observed — headroom for a loaded box starting node far slower than +/// any of those runs, while still bounding the one case that reaches it. +const RESUME_GRACE: Duration = Duration::from_secs(5); /// How a subagent's turn ended, as far as the daemon can tell from what /// [`hive_claude::RunningClaude::wait`] returned. @@ -77,6 +92,40 @@ enum TurnEnd { Failed(String), } +/// What settled a resumed turn's bounded wait — the one thing `continue` +/// blocks on before it answers. +enum ResumeVerdict { + /// The child emitted a stream event that isn't the turn's own terminal + /// `result` — which a resume that matched nothing never gets as far as + /// emitting. The turn is genuinely running; there is nothing left to + /// wait for. + Underway, + /// The turn ended before that happened. A `Failed` end here is the + /// resume miss `continue` exists to report. + Ended(TurnEnd), +} + +/// Write half of [`ResumeVerdict`]'s channel, shared between the turn's sink +/// (which reports `Underway`) and its background task (which reports +/// `Ended`). `Option` because a `oneshot::Sender` is consumed by its single +/// send: whichever side gets there first takes it, and the loser finds +/// nothing left to send on — exactly the "first answer wins" the wait wants. +/// The whole thing is `None` on a `start`, which has no resume to miss. +type VerdictTx = Arc>>>; + +/// Report `verdict`, if nothing has been reported yet. +/// +/// Returns `true` only when this call both won that race *and* found a +/// receiver still listening — i.e. the verdict genuinely reached a +/// `continue` that is about to act on it. The background task reads that +/// answer to decide whether its end-of-turn todo would be a second +/// notification for something the caller has already been told to its face. +fn settle(tx: Option<&VerdictTx>, verdict: ResumeVerdict) -> bool { + let Some(tx) = tx else { return false }; + let taken = tx.lock().unwrap_or_else(PoisonError::into_inner).take(); + taken.is_some_and(|tx| tx.send(verdict).is_ok()) +} + /// 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. @@ -508,12 +557,16 @@ fn start_reserved( .archive_by_title(name) .map_err(|e| anyhow::anyhow!("archiving the prior `{name}` session failed: {e}"))?; } + // No verdict channel: a `start` creates its session, so there is no + // resume to miss and nothing for the caller to wait on past the spawn — + // see `spawn_and_track`'s doc. spawn_and_track( state, name, &config, &Attach::Create(name.to_owned()), trigger, + None, ) } @@ -522,14 +575,18 @@ fn start_reserved( /// daemon restarted, reattaching." Refuses a name already running (same /// concurrent-run hazard as `start`). /// -/// A name with **no** session to resume is not refused here: claude's own -/// `--resume` answers that, and its answer arrives with the turn rather -/// than before it — as a `SessionNotFound` failure carrying the directory -/// that was searched (`classify_end`), reaching the caller through the -/// end-of-turn todo and `status` like any other failed turn. The pre-check -/// that used to live here could only repeat the lookup the driver was about -/// to do anyway, and told the caller a session didn't exist when the true -/// answer was almost always that it exists somewhere else. +/// A name with **no** session to resume is still not *pre-checked* here: +/// claude's own `--resume` answers that, and the pre-check that used to live +/// here could only repeat the lookup the driver was about to do anyway, +/// while telling the caller a session didn't exist when the true answer was +/// almost always that it exists somewhere else. What changed is that +/// `continue` now waits for the driver's answer instead of returning ahead +/// of it: a missed resume comes back as this call's own `Err`, carrying +/// claude's message and the directory that was searched (`classify_end`), +/// rather than only as an end-of-turn todo the caller had already stopped +/// looking for. A `continue` that reports "started" therefore means the turn +/// started. See `await_resume` for the bound, and `spawn_and_track`'s doc for +/// why `start` keeps the older, unconditional contract. /// /// Resuming a session whose last turn was *killed* is allowed — that is /// often exactly what the caller wants — but never silent: the reply says @@ -538,8 +595,9 @@ fn start_reserved( /// /// # Errors /// -/// An invalid name, one already running, or `Claude::spawn` failing. -pub fn continue_( +/// An invalid name, one already running, `Claude::spawn` failing, or the +/// resumed turn failing within `RESUME_GRACE` — in practice a missed resume. +pub async fn continue_( state: &Arc, name: &str, prompt: String, @@ -559,11 +617,45 @@ pub fn continue_( // 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); - let result = continue_reserved(state, name, prompt, model, effort, dir.as_deref()); - if result.is_err() { + let (tx, rx) = oneshot::channel(); + let verdict: VerdictTx = Arc::new(Mutex::new(Some(tx))); + let started = continue_reserved(state, name, prompt, model, effort, dir.as_deref(), &verdict); + if started.is_err() { state.release_reservation(name); + return started; + } + // Nothing to release on this path: a turn that ended inside the grace + // has already been through `finish_turn`, which clears the tracking the + // spawn put there. + await_resume(rx).await?; + started.map(|msg| note_resumed_after_kill(&msg, killed)) +} + +/// Hold a `continue` open until its resume is known to have landed — or +/// until [`RESUME_GRACE`] says that waiting any longer costs more than the +/// answer is worth. +/// +/// Returns as soon as *either* side of the race reports, so a successful +/// `continue` pays no fixed delay: its first stream event settles the wait +/// at roughly the same moment a miss's exit would have. The timeout is the +/// floor under a child that does neither, and a turn that reaches it is +/// reported as started — which it is, with the end-of-turn todo left to say +/// how it goes. +/// +/// # Errors +/// +/// The turn's own failure message, verbatim: the caller asked claude to +/// resume a session and claude said why it couldn't, which is a better +/// answer than anything this daemon could paraphrase it into. +async fn await_resume(rx: oneshot::Receiver) -> anyhow::Result<()> { + match tokio::time::timeout(RESUME_GRACE, rx).await { + Ok(Ok(ResumeVerdict::Ended(TurnEnd::Failed(e)))) => anyhow::bail!("{e}"), + // Everything else is a turn that started: it spoke (`Underway`), or + // it ended on its own terms within the grace — `Complete`, or a + // `Killed` that some concurrent `interrupt` asked for and whose todo + // says so — or it is still going when the grace runs out. + _ => Ok(()), } - result.map(|msg| note_resumed_after_kill(&msg, killed)) } /// Append the "you are resuming a killed session" note to `continue`'s reply @@ -590,17 +682,20 @@ fn continue_reserved( model: Option, effort: Option, dir: Option<&str>, + verdict: &VerdictTx, ) -> anyhow::Result { let config = build_config(name, model, effort, None, dir); // No existence pre-check: claude's own `--resume` is the authority on // whether the session is there, and it errors rather than quietly - // starting a fresh one. See this module's doc. + // starting a fresh one. `verdict` is how that answer gets back to the + // caller in time to be its error. See this module's doc. spawn_and_track( state, name, &config, &Attach::Resume(name.to_owned()), prompt, + Some(Arc::clone(verdict)), ) } @@ -612,22 +707,41 @@ fn continue_reserved( /// **All three callbacks, deliberately.** A stderr line or a stdout line /// that didn't parse as JSON is proof the child is alive every bit as much /// as a stream-json event is, and the failure that matters here is reporting -/// a live subagent as wedged — so anything the child says counts. Nothing is -/// read out of the content: classifying *what* the subagent is doing is a +/// a live subagent as wedged — so anything the child says counts, and what +/// it said is never read: classifying *what* the subagent is doing is a /// separate question from whether it's doing anything. /// /// Sink methods are called synchronously from the driver's stream readers as /// lines arrive, so the body has to stay cheap — one uncontended map write /// is, and forwarding to a channel to do the same write elsewhere would cost -/// more than it saved. +/// more than it saved. `settle` adds a second uncontended lock on a +/// `resume`d turn only, and finds an already-emptied slot after the first +/// event — strictly less work than the map write next to it. +/// +/// **Liveness counts every callback; "the turn is underway" does not.** A +/// resume that matched nothing is not silent: claude writes the reason to +/// stderr *and* emits a terminal stream-json `result` event before exiting, +/// so treating any callback at all as proof the turn began would report +/// every missed resume as a successful start. The narrowest fact that +/// separates the two is the event's own kind — a `result` is stream-json's +/// end-of-turn marker, so an event that isn't one is a turn still in +/// progress. That's the envelope, not the content: nothing here reads what +/// the subagent said. struct LivenessSink { state: Arc, name: String, + /// Where to report the first non-terminal event, on a `resume` whose + /// caller is still waiting to hear whether it landed. `None` on a + /// `start`, which has nobody waiting. + verdict: Option, } impl hive_claude::Sink for LivenessSink { - fn on_event(&self, _event: &serde_json::Value) { + fn on_event(&self, event: &serde_json::Value) { self.state.note_event(&self.name); + if event.get("type").and_then(serde_json::Value::as_str) != Some("result") { + settle(self.verdict.as_ref(), ResumeVerdict::Underway); + } } fn on_stdout_line(&self, _line: &str) { @@ -639,11 +753,22 @@ impl hive_claude::Sink for LivenessSink { } } -/// Spawn the child (synchronous — returns with a real pid the instant the -/// process exists, which *is* "confirmed running": there is no stronger -/// signal to wait for without slowing every call down for no reason), track -/// it in `running`, and hand the actual turn off to a background task so -/// the caller returns immediately instead of blocking on the whole turn. +/// Spawn the child (synchronous — `Claude::spawn` returns with a real pid +/// the instant the process exists), track it in `running`, and hand the +/// actual turn off to a background task so nobody blocks on the whole turn. +/// +/// **A pid is "confirmed running" for a `start`, and only for a `start`.** +/// A `start` creates its session, so the only thing that can go wrong at +/// attach time is the spawn itself, which has already either succeeded or +/// returned here as an error — there is no stronger signal to wait for, and +/// waiting would slow every call down for no reason. A `resume` breaks that +/// reasoning: the session it names may not be there, and claude only says so +/// a fraction of a second *after* the process exists, so for a `continue` a +/// pid is not proof the turn began. That's what `verdict` is for — `Some` +/// only on the resume path, carrying the first real answer back to a caller +/// that is waiting for it (`await_resume`); `None` on a `start`, which keeps +/// the immediate-return contract unchanged. +/// /// Not `async` itself — `tokio::spawn` needs an active runtime to spawn /// *onto*, not an `async` caller to spawn *from*. fn spawn_and_track( @@ -652,6 +777,7 @@ fn spawn_and_track( config: &Config, attach: &Attach, prompt: String, + verdict: Option, ) -> anyhow::Result { let running = Claude::spawn(config, attach) .map_err(|e| anyhow::anyhow!("starting the subagent process failed: {e}"))?; @@ -683,6 +809,7 @@ fn spawn_and_track( let sink = LivenessSink { state: Arc::clone(&state), name: task_name.clone(), + verdict: verdict.clone(), }; let end = classify_end(running.wait(&prompt, &sink).await, searched.as_deref()); match &end { @@ -699,7 +826,17 @@ fn spawn_and_track( } } state.finish_turn(&task_name, &end); - push_turn_end_todo(&state.socket, &task_name, &end).await; + let failed = matches!(end, TurnEnd::Failed(_)); + // The todo is how a turn's end reaches an agent that is no longer + // looking — so it is pushed for every end *except* the one the + // caller is being handed as a tool-call error right now. `settle` + // saying the verdict was delivered is what makes that certain: a + // `continue` whose grace had already run out gets `false` here and + // its todo, same as before. + let reported = settle(verdict.as_ref(), ResumeVerdict::Ended(end.clone())); + if !(failed && reported) { + push_turn_end_todo(&state.socket, &task_name, &end).await; + } }); Ok(format!("subagent `{name}` started")) @@ -1338,6 +1475,7 @@ mod tests { let sink = LivenessSink { state: Arc::clone(&state), name: "n".to_owned(), + verdict: None, }; (state, sink) } @@ -1474,6 +1612,174 @@ mod tests { ); } + /// The `Failed` end a real missed resume produces, message and all — + /// `classify_end`'s own output for the error the driver raises when + /// `--resume` matches nothing. + fn missed_resume() -> TurnEnd { + classify_end( + Err(hive_claude::Error::SessionNotFound), + Some( + "(searched /home/agent/.claude for cwd /home/agent/work; if it was started \ + elsewhere, pass the `dir` it was started in)", + ), + ) + } + + #[tokio::test] + async fn a_continue_whose_resume_missed_is_an_error_not_a_started_message() { + // The whole point of the bounded wait: a `continue` naming a session + // that isn't there must fail the tool call, not answer "started" and + // leave the real answer to a todo nobody is waiting on any more. + let (tx, rx) = oneshot::channel(); + tx.send(ResumeVerdict::Ended(missed_resume())) + .map_err(|_| ()) + .expect("the waiter is still listening"); + let err = await_resume(rx) + .await + .expect_err("a missed resume must reach the caller as an error"); + let msg = err.to_string(); + assert!( + msg.contains("no session matched"), + "claude's own diagnosis is the error: {msg}" + ); + assert!( + msg.contains("pass the `dir` it was started in"), + "and the searched location travels with it: {msg}" + ); + } + + #[tokio::test(start_paused = true)] + async fn a_turn_that_got_underway_answers_without_waiting_out_the_grace() { + // A successful `continue` must not pay a fixed delay — it returns on + // the first sign of the turn, not on the bound. Paused time makes + // that measurable: the clock only advances if something awaits it. + let (tx, rx) = oneshot::channel(); + tx.send(ResumeVerdict::Underway) + .map_err(|_| ()) + .expect("the waiter is still listening"); + let started = tokio::time::Instant::now(); + await_resume(rx) + .await + .expect("an underway turn is not an error"); + assert!( + started.elapsed() < RESUME_GRACE, + "a turn that spoke must settle the wait immediately, not on the bound: {:?}", + started.elapsed() + ); + } + + #[tokio::test(start_paused = true)] + async fn a_turn_that_neither_speaks_nor_exits_is_reported_as_started() { + // The bound is a floor under a child that does nothing at all. The + // sender is held open for the whole wait, so only the timeout can + // end it. + let (tx, rx) = oneshot::channel::(); + let started = tokio::time::Instant::now(); + await_resume(rx) + .await + .expect("a silent child is not a failed resume"); + assert!( + started.elapsed() >= RESUME_GRACE, + "the wait must actually run to the bound: {:?}", + started.elapsed() + ); + drop(tx); + } + + #[tokio::test] + async fn an_early_end_that_is_not_a_failure_still_reads_as_started() { + // A turn that completed, or that a concurrent `interrupt` killed, + // inside the grace: both ended a turn that genuinely began, and both + // have a todo of their own to explain themselves. + for end in [TurnEnd::Complete, TurnEnd::Killed { signal: 9 }] { + let (tx, rx) = oneshot::channel(); + tx.send(ResumeVerdict::Ended(end.clone())) + .map_err(|_| ()) + .expect("the waiter is still listening"); + assert!( + await_resume(rx).await.is_ok(), + "{end:?} ended a turn that started — only a failure is the caller's error" + ); + } + } + + #[test] + fn only_the_first_verdict_is_reported_and_only_to_a_live_caller() { + // `settle`'s two jobs: the sink and the background task race for one + // sender, and the winner's `true` is what tells the task the caller + // has already been handed the failure. + let (tx, rx) = oneshot::channel(); + let verdict: VerdictTx = Arc::new(Mutex::new(Some(tx))); + assert!( + settle(Some(&verdict), ResumeVerdict::Underway), + "the first report reaches a listening caller" + ); + assert!( + !settle(Some(&verdict), ResumeVerdict::Ended(missed_resume())), + "the loser of the race has nothing left to send on — and so must still push its todo" + ); + drop(rx); + + let (tx, rx) = oneshot::channel(); + let verdict: VerdictTx = Arc::new(Mutex::new(Some(tx))); + drop(rx); + assert!( + !settle(Some(&verdict), ResumeVerdict::Ended(missed_resume())), + "a caller whose grace already expired is not listening, so its todo must still be \ + pushed" + ); + + assert!( + !settle(None, ResumeVerdict::Underway), + "a `start` has no verdict channel and nobody waiting on one" + ); + } + + #[test] + fn a_terminal_result_event_is_liveness_but_not_proof_the_turn_began() { + // Measured, and the reason the underway signal reads the event's + // kind at all: a resume that matched nothing still emits a + // stream-json `result` event (and stderr chatter) before exiting, so + // treating any callback as "underway" would report every missed + // resume as a successful start. + use hive_claude::Sink as _; + + fn fresh() -> (oneshot::Receiver, LivenessSink) { + let (tx, rx) = oneshot::channel(); + let sink = LivenessSink { + state: Arc::new(State::new(PathBuf::from("/dev/null"))), + name: "n".to_owned(), + verdict: Some(Arc::new(Mutex::new(Some(tx)))), + }; + (rx, sink) + } + + let (mut rx, sink) = fresh(); + sink.on_event(&serde_json::json!({"type": "result", "is_error": true})); + assert!( + rx.try_recv().is_err(), + "the turn's own end-of-turn marker must not settle the wait as underway" + ); + assert!( + sink.state.last_event_age("n").is_some(), + "it is still liveness — the child did say something" + ); + + let (mut rx, sink) = fresh(); + sink.on_stderr_line("Error: --resume requires a valid session ID or session title"); + assert!( + rx.try_recv().is_err(), + "nor does stderr, which a missed resume writes to before it exits" + ); + + let (mut rx, sink) = fresh(); + sink.on_event(&serde_json::json!({"type": "system", "subtype": "init"})); + assert!( + matches!(rx.try_recv(), Ok(ResumeVerdict::Underway)), + "a non-terminal event is a turn in progress, which is the signal continue waits for" + ); + } + #[test] fn build_config_defaults_effort_to_medium_when_omitted() { // Unlike `model`, an omitted `effort` does not fall through to