diff --git a/claude-plugins/plugins/base/skills/claude-subagents/SKILL.md b/claude-plugins/plugins/base/skills/claude-subagents/SKILL.md index dc53f709..05f6191a 100644 --- a/claude-plugins/plugins/base/skills/claude-subagents/SKILL.md +++ b/claude-plugins/plugins/base/skills/claude-subagents/SKILL.md @@ -30,17 +30,10 @@ status(name) interrupt(name, force?) ``` -`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 -wedged and worth an `interrupt`. Reach for `continue` only once you +`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 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 1cd7184e..23d67ebf 100644 --- a/docs/tools/subagent.md +++ b/docs/tools/subagent.md @@ -26,70 +26,12 @@ Served under the `subagent` MCP server (`mcp__subagent__`): `start`, ## State -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 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. - -## 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` 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: - -``` -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. +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. ## A killed turn diff --git a/hive-subagent-mcp/Cargo.toml b/hive-subagent-mcp/Cargo.toml index 025735ea..e6d78ef0 100644 --- a/hive-subagent-mcp/Cargo.toml +++ b/hive-subagent-mcp/Cargo.toml @@ -24,12 +24,6 @@ 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 fd5b7bef..511e2bb0 100644 --- a/hive-subagent-mcp/src/mcp.rs +++ b/hive-subagent-mcp/src/mcp.rs @@ -79,12 +79,8 @@ struct ContinueArgs { #[serde(default)] effort: Option, /// Omit to reuse whatever `dir` `start` (or a prior `continue`) used for - /// 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. + /// this name — the daemon remembers it. Only pass this to point the + /// session at a *different* directory than last time. #[serde(default)] dir: Option, } @@ -149,16 +145,12 @@ 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 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. \ + 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." )] - async fn r#continue(&self, Parameters(args): Parameters) -> String { + fn r#continue(&self, Parameters(args): Parameters) -> String { match session::continue_( &self.state, &args.name, @@ -166,9 +158,7 @@ impl SubagentMcp { args.model, args.effort, args.dir.as_deref(), - ) - .await - { + ) { Ok(msg) => msg, Err(e) => format!("continue error: {e:#}"), } @@ -191,9 +181,7 @@ 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. 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`." + and what to do about it." )] 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 e2747ae7..dd1cd51c 100644 --- a/hive-subagent-mcp/src/session.rs +++ b/hive-subagent-mcp/src/session.rs @@ -4,29 +4,17 @@ //! //! **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), a `name -> dir` memory so `start`'s +//! the `None`/`Some` split is for) plus a `name -> dir` memory so `start`'s //! `dir` doesn't have to be repeated on every later `continue`/`status` -//! (`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 — the timestamp says the child is alive, never what it said. -//! -//! **`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`. +//! (`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. //! **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 @@ -52,25 +40,8 @@ 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, 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); +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. @@ -92,40 +63,6 @@ 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. @@ -143,15 +80,7 @@ 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. -/// -/// `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 { +fn classify_end(outcome: hive_claude::Result<()>) -> TurnEnd { let Err(error) = outcome else { return TurnEnd::Complete; }; @@ -160,31 +89,9 @@ fn classify_end(outcome: hive_claude::Result<()>, searched: Option<&str>) -> Tur { 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 it was started elsewhere, pass the `dir` it was started in)", - 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 @@ -204,10 +111,6 @@ fn searched_location(config: &Config) -> Option { /// 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 @@ -219,7 +122,6 @@ pub struct State { running: Mutex>>, dirs: Mutex>, killed: Mutex>, - last_event: Mutex>, socket: PathBuf, } @@ -230,7 +132,6 @@ impl State { running: Mutex::new(HashMap::new()), dirs: Mutex::new(HashMap::new()), killed: Mutex::new(HashMap::new()), - last_event: Mutex::new(HashMap::new()), socket, } } @@ -283,10 +184,6 @@ 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 { @@ -298,10 +195,6 @@ 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. @@ -323,42 +216,6 @@ 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 @@ -557,37 +414,21 @@ 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, ) } /// 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 already running (same +/// daemon restarted, reattaching." Refuses a name with no session on disk +/// at all (nothing to continue) or one already running (same /// concurrent-run hazard as `start`). /// -/// 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 /// so, since a caller that never ran `status` and missed the todo would @@ -595,9 +436,9 @@ fn start_reserved( /// /// # Errors /// -/// 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_( +/// No session under `name`, an invalid name, one already running, or +/// `Claude::spawn` failing. +pub fn continue_( state: &Arc, name: &str, prompt: String, @@ -617,45 +458,11 @@ pub async 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 (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() { + let result = continue_reserved(state, name, prompt, model, effort, dir.as_deref()); + if result.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 @@ -682,93 +489,29 @@ 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. `verdict` is how that answer gets back to the - // caller in time to be its error. See this module's doc. + 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" + ); + } spawn_and_track( state, name, &config, &Attach::Resume(name.to_owned()), prompt, - Some(Arc::clone(verdict)), ) } -/// 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, 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. `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) { - 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) { - self.state.note_event(&self.name); - } - - fn on_stderr_line(&self, _line: &str) { - self.state.note_event(&self.name); - } -} - -/// 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. -/// +/// 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. /// Not `async` itself — `tokio::spawn` needs an active runtime to spawn /// *onto*, not an `async` caller to spawn *from*. fn spawn_and_track( @@ -777,7 +520,6 @@ 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}"))?; @@ -792,26 +534,11 @@ 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 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()); + let end = classify_end(running.wait(&prompt, &NoopSink).await); match &end { TurnEnd::Complete => {} TurnEnd::Killed { signal } => { @@ -826,17 +553,7 @@ fn spawn_and_track( } } state.finish_turn(&task_name, &end); - 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; - } + push_turn_end_todo(&state.socket, &task_name, &end).await; }); Ok(format!("subagent `{name}` started")) @@ -849,11 +566,6 @@ 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 @@ -866,7 +578,6 @@ 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. +/// 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 @@ -919,15 +608,13 @@ 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.{}", - describe_liveness(last_event_age) + `interrupt` it if you want it stopped early." )); } Some(false) => { @@ -1228,19 +915,19 @@ mod tests { #[test] fn classify_end_separates_a_signalled_child_from_a_clean_one() { - assert_eq!(classify_end(Ok(()), None), TurnEnd::Complete); + assert_eq!(classify_end(Ok(())), TurnEnd::Complete); assert_eq!( - classify_end(Err(signalled_exit(libc::SIGKILL)), None), + 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)), None), + classify_end(Err(signalled_exit(libc::SIGTERM))), TurnEnd::Killed { signal: 15 } ); assert!( matches!( - classify_end(Err(hive_claude::Error::PromptTooLong), None), + classify_end(Err(hive_claude::Error::PromptTooLong)), TurnEnd::Failed(_) ), "claude exiting on its own is a failure, not a kill" @@ -1251,9 +938,8 @@ 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, None).expect("killed status"); - let idle = describe_status("n", None, None, true, None).expect("idle status"); + 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}" @@ -1274,30 +960,28 @@ 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, None).expect("running status"); + let running = describe_status("n", Some(true), None, false).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, None).expect("starting status"); + let starting = describe_status("n", Some(false), None, false).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, None).expect("idle status"); + let idle = describe_status("n", None, None, true).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, None).expect("killed status"); + let killed = describe_status("n", None, Some(libc::SIGKILL), true).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, None) + let missing = describe_status("n", None, None, false) .expect_err("nothing tracked and nothing on disk is an error, not a state"); assert!( missing.to_string().contains("`start`"), @@ -1315,7 +999,7 @@ mod tests { ("done", Ok(())), ] { assert!(state.reserve(name)); - state.finish_turn(name, &classify_end(outcome, None)); + state.finish_turn(name, &classify_end(outcome)); assert_eq!( state.occupancy(name), None, @@ -1325,10 +1009,8 @@ 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, None).expect("status"); - let done = - describe_status("done", None, state.killed_by("done"), true, None).expect("status"); + 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}"); } @@ -1414,372 +1096,6 @@ 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(), - verdict: None, - }; - (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 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"); - }; - 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 the `dir` it was started in"), - "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}" - ); - } - - /// 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