hive-runtime: TurnStop replaces the subagent daemon's Stopper
TurnStop is the enum over hive-claude's Cancel and the ACP Canceller, with stop() -> bool (false when no turn was in flight). It signals SIGINT only and never waits. The subagent daemon tracks a TurnStop per turn, so interrupt has one stop path; whether the name stays claimed afterwards (claude) or is freed (ACP) still follows the runtime.
This commit is contained in:
parent
baefb32bd5
commit
496f85c8c6
5 changed files with 147 additions and 36 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
///
|
||||
|
|
|
|||
101
hive-runtime/src/stop.rs
Normal file
101
hive-runtime/src/stop.rs
Normal file
|
|
@ -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<Cancel> for TurnStop {
|
||||
fn from(cancel: Cancel) -> Self {
|
||||
Self::Claude(cancel)
|
||||
}
|
||||
}
|
||||
|
||||
impl From<Canceller> 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;
|
||||
}
|
||||
}
|
||||
|
|
@ -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<String> {
|
||||
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<String> {
|
|||
`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!(
|
||||
|
|
|
|||
Loading…
Reference in a new issue