diff --git a/hive-runtime/README.md b/hive-runtime/README.md index a05d096e..6b6d4025 100644 --- a/hive-runtime/README.md +++ b/hive-runtime/README.md @@ -32,8 +32,7 @@ The crate depends on no hyperhive binary crate, so `hive-agent` and ## ACP backend: stopping a turn -A turn is stopped with `session/cancel`: by the `Canceller` handle (`TurnStop` -wraps it, and claude's cancel handle, for callers that run either), or by the +A turn is stopped with `session/cancel`: by the `Canceller` handle, or by the idle watchdog once no `session/update` has arrived for `Config::idle_timeout`. An agent that has not answered the prompt 10s later is sent SIGINT, and the next turn respawns it. A watchdog stop fails the turn diff --git a/hive-runtime/src/acp/mod.rs b/hive-runtime/src/acp/mod.rs index 1f13a7cd..64afa350 100644 --- a/hive-runtime/src/acp/mod.rs +++ b/hive-runtime/src/acp/mod.rs @@ -1029,7 +1029,7 @@ mod tests { use hive_claude::{Config, NoopSink, PercentPolicy, Sink}; use super::{AcpError, AcpRuntime, Choice, ReportedCost, SessionChoices, Spend, TurnKind}; - use crate::{AcpCommand, Error, Runtime, TurnStop}; + use crate::{AcpCommand, Error, Runtime}; /// An ACP agent in plain `sh`. It answers `initialize`, numbers its /// sessions `s1`, `s2`, …, appends every method it is sent to @@ -1422,24 +1422,6 @@ done assert!(!canceller.cancel(), "the turn is over"); } - #[tokio::test] - async fn a_turn_stop_cancels_only_a_turn_in_flight() { - let dir = tempfile::tempdir().unwrap(); - let runtime = runtime(dir.path(), "silent"); - let stop = TurnStop::from(runtime.canceller().unwrap()); - assert!(!stop.stop(), "no turn is in flight yet"); - - let config = config(dir.path()); - let (done, was_in_flight) = tokio::join!(runtime.run(&config, "one", &NoopSink), async { - prompt_sent(dir.path()).await; - stop.stop() - }); - assert!(was_in_flight); - assert!(done.is_ok(), "{done:?}"); - assert_eq!(count(dir.path(), "session/cancel"), 1); - assert!(!stop.stop(), "the turn is over"); - } - #[tokio::test] async fn an_agent_ignoring_a_cancel_is_killed() { let dir = tempfile::tempdir().unwrap(); diff --git a/hive-runtime/src/lib.rs b/hive-runtime/src/lib.rs index 21e1963f..52208e58 100644 --- a/hive-runtime/src/lib.rs +++ b/hive-runtime/src/lib.rs @@ -20,7 +20,6 @@ use std::path::PathBuf; mod acp; mod claude; mod spec; -mod stop; pub use acp::{ AcpError, AcpRuntime, Canceller, Choice, Choices, PermissionAsk, PermissionPolicy, @@ -34,7 +33,6 @@ pub use spec::{ ACP_API_KEY_ENV_ENV, ACP_ARGS_ENV, ACP_COMMAND_ENV, ACP_ENV_ENV, AcpCommand, RUNTIME_ENV, RuntimeSpec, SpecError, }; -pub use stop::TurnStop; /// The operations every runtime backend offers. /// diff --git a/hive-runtime/src/stop.rs b/hive-runtime/src/stop.rs deleted file mode 100644 index 640c7090..00000000 --- a/hive-runtime/src/stop.rs +++ /dev/null @@ -1,101 +0,0 @@ -//! [`TurnStop`]: the handle that stops a turn from outside the call driving it. - -use hive_claude::Cancel; - -use crate::Canceller; - -/// Stops the turn a claude process or an ACP agent has in flight. Cheap to -/// clone. -/// -/// Every stop is SIGINT, directly (claude) or after `session/cancel` went -/// unanswered (ACP): claude shuts down the commands it started when it gets -/// SIGINT, and nothing here escalates past it or waits for the turn to end. -#[derive(Clone)] -pub enum TurnStop { - /// Signals the turn's `claude` process group. - Claude(Cancel), - /// Sends the ACP agent `session/cancel`; the runtime sends SIGINT to one - /// that ignores it for its own grace period. - Acp(Canceller), -} - -impl From for TurnStop { - fn from(cancel: Cancel) -> Self { - Self::Claude(cancel) - } -} - -impl From for TurnStop { - fn from(canceller: Canceller) -> Self { - Self::Acp(canceller) - } -} - -impl TurnStop { - /// Ask the turn to stop, without waiting for it to. Returns `false` when - /// no turn was in flight and nothing was signalled. A claude handle exists - /// only for a spawned process, so it always returns `true`. - #[must_use] - pub fn stop(&self) -> bool { - match self { - Self::Claude(cancel) => { - cancel.cancel(false); - true - } - Self::Acp(canceller) => canceller.cancel(), - } - } -} - -#[cfg(all(test, unix))] -mod tests { - use std::path::Path; - use std::time::Duration; - - use hive_claude::{Attach, Claude, Config}; - - use super::TurnStop; - - /// A `claude` that records its SIGINT in `$MARK` and exits 0, and says it - /// is listening for one in `$READY`. - const FAKE_CLAUDE: &str = "#!/bin/sh\n\ - trap ': > \"$MARK\"; exit 0' INT\n\ - : > \"$READY\"\n\ - sleep 30 >/dev/null 2>&1 &\n\ - wait\n"; - - async fn exists(path: &Path) { - for _ in 0..250 { - if path.exists() { - return; - } - tokio::time::sleep(Duration::from_millis(20)).await; - } - panic!("{} never appeared", path.display()); - } - - #[tokio::test] - async fn a_claude_stop_is_a_sigint_to_the_process() { - use std::os::unix::fs::PermissionsExt as _; - - let dir = tempfile::tempdir().unwrap(); - let program = dir.path().join("fake-claude"); - std::fs::write(&program, FAKE_CLAUDE).unwrap(); - std::fs::set_permissions(&program, std::fs::Permissions::from_mode(0o755)).unwrap(); - let (ready, mark) = (dir.path().join("ready"), dir.path().join("int")); - let config = Config { - program: Some(program), - env: vec![ - ("READY".to_owned(), ready.display().to_string()), - ("MARK".to_owned(), mark.display().to_string()), - ], - ..Config::default() - }; - let running = Claude::spawn(&config, &Attach::Create("n".to_owned())).unwrap(); - let stop = TurnStop::from(running.cancel_handle()); - exists(&ready).await; - - assert!(stop.stop()); - exists(&mark).await; - } -} diff --git a/hive-subagent-mcp/src/session.rs b/hive-subagent-mcp/src/session.rs index 64aa940f..4a276e82 100644 --- a/hive-subagent-mcp/src/session.rs +++ b/hive-subagent-mcp/src/session.rs @@ -92,9 +92,9 @@ use std::path::{Path, PathBuf}; use std::sync::{Arc, Mutex, PoisonError}; use std::time::{Duration, Instant}; -use hive_claude::{Attach, Claude, Config, SessionStore}; +use hive_claude::{Attach, Cancel, Claude, Config, SessionStore}; use hive_runtime::{ - AcpCommand, AcpRuntime, PermissionAsk, PermissionPolicy, Runtime as _, RuntimeSpec, TurnStop, + AcpCommand, AcpRuntime, Canceller, PermissionAsk, PermissionPolicy, Runtime as _, RuntimeSpec, }; use hive_sh4re::permissions::ToolGroup; use tokio::sync::oneshot; @@ -226,13 +226,22 @@ enum SubagentRuntime { }, } +/// How `interrupt` reaches a tracked turn, per runtime. +enum Stopper { + /// Signals the turn's `claude` process. + Claude(Cancel), + /// Sends the ACP agent `session/cancel`; the runtime sends SIGINT to one + /// that ignores it for its own grace period. + Acp(Canceller), +} + /// What `running` holds for a claimed name. enum Slot { /// Claimed, with no process confirmed under it: a `start`/`continue` /// that hasn't spawned yet, or a goal run between turns. Starting, /// A turn is in flight, and this is how `interrupt` reaches it. - Running(TurnStop), + Running(Stopper), /// `interrupt` has signalled a claude turn and its process hasn't exited /// yet. Nothing escalates past that SIGINT, so a claude that ignores it /// stays here until it exits; the turn's own `finish_turn` frees the @@ -758,11 +767,11 @@ impl State { /// Upgrade `name`'s `Starting` reservation to a real, interruptible /// process. Same key as the reservation, so there is no window in which /// `name` reads as unoccupied between the two. - fn track(&self, name: &str, stop: TurnStop) { + fn track(&self, name: &str, stopper: Stopper) { self.running .lock() .unwrap_or_else(PoisonError::into_inner) - .insert(name.to_owned(), Slot::Running(stop)); + .insert(name.to_owned(), Slot::Running(stopper)); } /// Drop `name`'s cancel handle but keep the name claimed — the moment @@ -1738,7 +1747,7 @@ fn spawn_and_track( ) -> anyhow::Result { let running = Claude::spawn(config, attach) .map_err(|e| anyhow::anyhow!("starting the subagent process failed: {e}"))?; - state.track(name, running.cancel_handle().into()); + state.track(name, Stopper::Claude(running.cancel_handle())); // This turn supersedes whatever the previous one did, including having // been killed — the record is about the turn before this one. state.clear_kill(name); @@ -1777,7 +1786,7 @@ fn spawn_and_track( .and_then(|id| Ok(Claude::spawn(&config, &Attach::Resume(id))?)) { Ok(next_running) => { - state.track(&task_name, next_running.cancel_handle().into()); + state.track(&task_name, Stopper::Claude(next_running.cancel_handle())); state.note_event(&task_name); running = next_running; prompt = next; @@ -1865,7 +1874,7 @@ fn spawn_and_track_acp( let canceller = runtime .canceller() .ok_or_else(|| anyhow::anyhow!("the ACP runtime offers no cancel handle"))?; - state.track(name, canceller.clone().into()); + state.track(name, Stopper::Acp(canceller.clone())); state.clear_kill(name); state.note_event(name); @@ -1886,7 +1895,7 @@ fn spawn_and_track_acp( let Some(next) = after_turn(&state, &task_name, end, verdict.as_ref()).await else { return; }; - state.track(&task_name, canceller.clone().into()); + state.track(&task_name, Stopper::Acp(canceller.clone())); state.note_event(&task_name); prompt = next; } @@ -2391,29 +2400,31 @@ pub fn interrupt(state: &State, name: &str) -> anyhow::Result { `status` says when it has" ); } - // `stop()` is `false` when no turn is in flight (an ACP agent not - // prompted yet, or a turn just ended), and then nothing was stopped. - // The stop goes in before the signal and comes out under the same - // `stops` lock, so `plan_after_turn` never reads a cancel that - // didn't happen. - Some(Slot::Running(stop)) => { + Some(Slot::Running(Stopper::Claude(cancel))) => { + state.record_stop(name, StopReason::Cancelled); + running.insert(name.to_owned(), Slot::Stopping); + drop(running); + cancel.cancel(false); + claimed = " The name stays claimed until claude has exited; `status` says `stopping` \ + until then."; + } + // `cancel()` is `false` when no turn is in flight (the agent not + // prompted yet, or a turn just ended), and then nothing was + // cancelled. The stop goes in before the cancel and comes out under + // the same `stops` lock, so `plan_after_turn` never reads a cancel + // that didn't happen. + Some(Slot::Running(Stopper::Acp(canceller))) => { let mut stops = state.stops.lock().unwrap_or_else(PoisonError::into_inner); let prior = stops.insert(name.to_owned(), StopReason::Cancelled); - if !stop.stop() { + if !canceller.cancel() { match prior { Some(prior) => stops.insert(name.to_owned(), prior), None => stops.remove(name), }; drop(stops); - running.insert(name.to_owned(), Slot::Running(stop)); + running.insert(name.to_owned(), Slot::Running(Stopper::Acp(canceller))); return Err(still_starting(name)); } - drop(stops); - if matches!(state.runtime, SubagentRuntime::Claude) { - running.insert(name.to_owned(), Slot::Stopping); - claimed = " The name stays claimed until claude has exited; `status` says \ - `stopping` until then."; - } } } Ok(format!( @@ -5102,11 +5113,6 @@ echo '{"type":"result","subtype":"success","is_error":false}' .await; interrupt(&state, "n").expect("a running ACP turn is interruptible"); - assert_eq!( - state.occupancy("n"), - None, - "an interrupted ACP turn frees its name at once; only claude's stays claimed" - ); until("the run ends", || state.last_event_age("n").is_none()).await; assert_eq!(lines(&dir.join("prompts")).len(), 1, "no second goal turn");