diff --git a/hive-runtime/README.md b/hive-runtime/README.md index 6b6d4025..a05d096e 100644 --- a/hive-runtime/README.md +++ b/hive-runtime/README.md @@ -32,7 +32,8 @@ 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, or by the +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 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 64afa350..1f13a7cd 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}; + use crate::{AcpCommand, Error, Runtime, TurnStop}; /// An ACP agent in plain `sh`. It answers `initialize`, numbers its /// sessions `s1`, `s2`, …, appends every method it is sent to @@ -1422,6 +1422,24 @@ 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 52208e58..21e1963f 100644 --- a/hive-runtime/src/lib.rs +++ b/hive-runtime/src/lib.rs @@ -20,6 +20,7 @@ use std::path::PathBuf; mod acp; mod claude; mod spec; +mod stop; pub use acp::{ AcpError, AcpRuntime, Canceller, Choice, Choices, PermissionAsk, PermissionPolicy, @@ -33,6 +34,7 @@ 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 new file mode 100644 index 00000000..640c7090 --- /dev/null +++ b/hive-runtime/src/stop.rs @@ -0,0 +1,101 @@ +//! [`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 4a276e82..1739b01a 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, Cancel, Claude, Config, SessionStore}; +use hive_claude::{Attach, Claude, Config, SessionStore}; use hive_runtime::{ - AcpCommand, AcpRuntime, Canceller, PermissionAsk, PermissionPolicy, Runtime as _, RuntimeSpec, + AcpCommand, AcpRuntime, PermissionAsk, PermissionPolicy, Runtime as _, RuntimeSpec, TurnStop, }; use hive_sh4re::permissions::ToolGroup; use tokio::sync::oneshot; @@ -226,22 +226,13 @@ 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(Stopper), + Running(TurnStop), /// `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 @@ -767,11 +758,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, stopper: Stopper) { + fn track(&self, name: &str, stop: TurnStop) { self.running .lock() .unwrap_or_else(PoisonError::into_inner) - .insert(name.to_owned(), Slot::Running(stopper)); + .insert(name.to_owned(), Slot::Running(stop)); } /// Drop `name`'s cancel handle but keep the name claimed — the moment @@ -1747,7 +1738,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, Stopper::Claude(running.cancel_handle())); + state.track(name, running.cancel_handle().into()); // 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); @@ -1786,7 +1777,7 @@ fn spawn_and_track( .and_then(|id| Ok(Claude::spawn(&config, &Attach::Resume(id))?)) { Ok(next_running) => { - state.track(&task_name, Stopper::Claude(next_running.cancel_handle())); + state.track(&task_name, next_running.cancel_handle().into()); state.note_event(&task_name); running = next_running; prompt = next; @@ -1874,7 +1865,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, Stopper::Acp(canceller.clone())); + state.track(name, canceller.clone().into()); state.clear_kill(name); state.note_event(name); @@ -1895,7 +1886,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, Stopper::Acp(canceller.clone())); + state.track(&task_name, canceller.clone().into()); state.note_event(&task_name); prompt = next; } @@ -2400,31 +2391,29 @@ pub fn interrupt(state: &State, name: &str) -> anyhow::Result { `status` says when it has" ); } - 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))) => { + // `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)) => { let mut stops = state.stops.lock().unwrap_or_else(PoisonError::into_inner); let prior = stops.insert(name.to_owned(), StopReason::Cancelled); - if !canceller.cancel() { + if !stop.stop() { match prior { Some(prior) => stops.insert(name.to_owned(), prior), None => stops.remove(name), }; drop(stops); - running.insert(name.to_owned(), Slot::Running(Stopper::Acp(canceller))); + running.insert(name.to_owned(), Slot::Running(stop)); 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!(