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