Watch
0
0
Fork
You've already forked hyperhive
0

subagents, ACP: stop with SIGINT only, so claude ends its own tool commands

claude's Bash tool starts every command in a new session, so a signal to
claude's process group never reaches it. claude stops those commands
itself when SIGINT makes it shut down. A SIGKILL leaves them running,
re-parented to init. Per the ruling on #4999, every stop is SIGINT and
nothing escalates.

- Subagent `interrupt` always sends SIGINT. The `force` argument is gone
  from the MCP tool and from `session::interrupt`. A caller that still
  sends `force: true` gets SIGINT; the argument is ignored.
- `interrupt` used to free the session name before it signalled, so a
  `continue` could start a second claude on the session while the first
  was still exiting. The `running` map now holds a three-state `Slot`
  (starting, running, stopping). An interrupted claude turn is
  `Stopping` until the turn task's own `finish_turn` clears it. `status`
  says "stopping", and `start`/`continue` on that name are refused.
  `interrupt` still returns at once. The ACP arm is unchanged: its name
  is freed when `session/cancel` is sent.
- ACP `Connection` drop sends its process group SIGINT, not SIGKILL, and
  the spawn no longer sets tokio's `kill_on_drop`, which SIGKILLed the
  leader after the `Drop` body. The test fixture's background grandchild
  resets SIGINT to default, since `&` starts it ignored.

Tests: a fake claude that records SIGINT, interrupted, fails with
`cancel(true)` (the old `force: true` path). An interrupted fake claude
that keeps running keeps its name claimed and `continue` is refused,
which fails when the slot is freed before the signal. A fake ACP agent
that records SIGINT, dropped, fails with SIGKILL in `Drop`, and also with
SIGINT in `Drop` but `kill_on_drop(true)` left in place.

Not changed: hive-claude's idle watchdog (still SIGKILL, separate
hive-claude change), the units' `KillSignal`, `machinectl kill`, and the
login helper.
This commit is contained in:
atlas 2026-10-10 19:52:25 +02:00 • committed by mara
commit 43e69d345e
7 changed files with 337 additions and 167 deletions

View file

@ -27,7 +27,7 @@ skill's recipe for you:
start(name, prompt_file, model?, effort?, trigger?) start(name, prompt_file, model?, effort?, trigger?)
continue(name, prompt, model?, effort?) continue(name, prompt, model?, effort?)
status(name) status(name)
interrupt(name, force?) interrupt(name)
``` ```
`start` and `continue` return once the turn is under way, not once it `start` and `continue` return once the turn is under way, not once it
@ -42,7 +42,8 @@ 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 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 wedged and worth an `interrupt`. Reach for `continue` only once you
actually have a new instruction for it, since that spends a turn. actually have a new instruction for it, since that spends a turn.
`interrupt` genuinely stops a running turn (`force: true` for SIGKILL). `interrupt` genuinely stops a running turn, with SIGINT so claude stops
the commands it started; `status` says `stopping` until claude has exited.
Model choice, prompt hygiene, splitting big batches, verify-then-report — Model choice, prompt hygiene, splitting big batches, verify-then-report —
everything else in this skill — applies exactly the same whether you're everything else in this skill — applies exactly the same whether you're

View file

@ -166,10 +166,15 @@ from how the child exited. `continue` clears it and hands the session a
fresh turn allowance, which is what makes an interrupt a pause you can fresh turn allowance, which is what makes an interrupt a pause you can
undo rather than a session you have to abandon. undo rather than a session you have to abandon.
On `claude`, the signal goes to claude's whole process group, so the On `claude`, the signal is always SIGINT, and it goes to claude's whole
commands the subagent started (a `nix build`, a test run) get it too. The process group. The signal doesn't reach a command that moved itself out of
signal doesn't reach a command that moved itself out of the group the group (`setsid`), and claude's Bash tool starts every command that way:
(`setsid`). claude stops those commands itself when it shuts down on SIGINT. A SIGKILL
would leave them running, so nothing escalates to one. `interrupt` returns
at once, and the name stays claimed until claude has exited: `status` says
**stopping**, and the daemon refuses `start` and `continue` on that name
until then. A claude that ignores the SIGINT stays stopping until someone kills
it by hand or the unit stops.
When the session knows where its report goes — `start`'s `report_file`, When the session knows where its report goes — `start`'s `report_file`,
or the path the subagent names when it signals — the daemon appends the or the path the subagent names when it signals — the daemon appends the
@ -345,7 +350,7 @@ page applies as written.
On `acp`: On `acp`:
- each run spawns its own copy of the parent's ACP agent. The end of the - each run spawns its own copy of the parent's ACP agent. The end of the
run kills it and every command it started. The unit loads run sends SIGINT to it and every command in its process group. The unit loads
`backendEnvironmentFile` for it. On the `opencode` preset, the daemon `backendEnvironmentFile` for it. On the `opencode` preset, the daemon
reads a provider key that file leaves unset from the swarm secret store reads a provider key that file leaves unset from the swarm secret store
at `swarm/agents/<agent>/acp-provider`, as the harness does. at `swarm/agents/<agent>/acp-provider`, as the harness does.
@ -354,9 +359,10 @@ On `acp`:
agent process, and a `continue` with none recorded fails straight away. agent process, and a `continue` with none recorded fails straight away.
- a role goes in front of the session's first prompt, since ACP has no - a role goes in front of the session's first prompt, since ACP has no
system prompt. system prompt.
- `interrupt` sends `session/cancel`, and `force` changes nothing. The - `interrupt` sends `session/cancel`. The runtime sends SIGINT to the
runtime kills an agent that ignores the cancel for 10s, with every process group of an agent that ignores the cancel for 10s, and when the
command it started. run ends. Nothing escalates past that SIGINT. The name is free as soon as
`interrupt` sends the cancel.
- the runtime sets `model` and `effort` on the session when the agent - the runtime sets `model` and `effort` on the session when the agent
offers that value for its `model` or `thought_level` config option. Any offers that value for its `model` or `thought_level` config option. Any
other value does nothing (`hive_runtime::AcpRuntime`). other value does nothing (`hive_runtime::AcpRuntime`).

View file

@ -35,13 +35,15 @@ The crate depends on no hyperhive binary crate, so `hive-agent` and
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, or by the
idle watchdog once no `session/update` has arrived for idle watchdog once no `session/update` has arrived for
`Config::idle_timeout`. An agent that has not answered the prompt 10s later `Config::idle_timeout`. An agent that has not answered the prompt 10s later
is killed, and the next turn respawns it. A watchdog stop fails the turn is sent SIGINT, and the next turn respawns it. A watchdog stop fails the turn
with `AcpError::IdleTimeout` (`IdleKilled` if the agent was killed). This is with `AcpError::IdleTimeout` (`IdleKilled` if the agent was sent SIGINT). This is
the only way a provider error the agent retries without reporting it, such as the only way a provider error the agent retries without reporting it, such as
an HTTP 429, ends the turn. an HTTP 429, ends the turn.
The agent leads its own process group. Killing it, and dropping the runtime, The agent leads its own process group. Stopping it, and dropping the runtime,
`SIGKILL`s that group, so the commands the agent started die with it. sends that group SIGINT, so the commands the agent started get it too.
Nothing escalates to SIGKILL: an agent that ignores SIGINT keeps running,
and the next turn starts a new one beside it.
## ACP backend: compaction ## ACP backend: compaction

View file

@ -39,7 +39,7 @@ const SETTLE_QUIET: Duration = Duration::from_millis(500);
const SETTLE_MAX: Duration = Duration::from_secs(5); const SETTLE_MAX: Duration = Duration::from_secs(5);
/// How long the agent has to end its turn after `session/cancel` before it is /// How long the agent has to end its turn after `session/cancel` before it is
/// killed. /// sent SIGINT.
const CANCEL_GRACE: Duration = Duration::from_secs(10); const CANCEL_GRACE: Duration = Duration::from_secs(10);
/// How long a `compact` command may send nothing before it is cancelled and /// How long a `compact` command may send nothing before it is cancelled and
@ -120,10 +120,12 @@ pub enum AcpError {
EmptyEndTurn, EmptyEndTurn,
#[error( #[error(
"ACP agent sent nothing for {silent_secs}s and did not stop within {grace_secs}s of \ "ACP agent sent nothing for {silent_secs}s and did not stop within {grace_secs}s of \
`session/cancel`, so it was killed" `session/cancel`, so it was sent SIGINT"
)] )]
IdleKilled { silent_secs: u64, grace_secs: u64 }, IdleKilled { silent_secs: u64, grace_secs: u64 },
#[error("ACP agent did not stop within {grace_secs}s of `session/cancel`, so it was killed")] #[error(
"ACP agent did not stop within {grace_secs}s of `session/cancel`, so it was sent SIGINT"
)]
CancelIgnored { grace_secs: u64 }, CancelIgnored { grace_secs: u64 },
} }
@ -141,8 +143,8 @@ struct CancelState {
impl Canceller { impl Canceller {
/// Ask the turn in flight to stop. The agent is sent `session/cancel`, /// Ask the turn in flight to stop. The agent is sent `session/cancel`,
/// and the turn then ends normally with what it produced so far; an /// and the turn then ends normally with what it produced so far; an
/// agent that has not ended it within `CANCEL_GRACE` is killed, and the /// agent that has not ended it within `CANCEL_GRACE` is sent SIGINT, and
/// turn fails with [`AcpError::CancelIgnored`]. Returns whether a turn /// the turn fails with [`AcpError::CancelIgnored`]. Returns whether a turn
/// was in flight; with none, this does nothing. /// was in flight; with none, this does nothing.
#[must_use] #[must_use]
pub fn cancel(&self) -> bool { pub fn cancel(&self) -> bool {
@ -1062,9 +1064,14 @@ mod tests {
const AGENT: &str = r#" const AGENT: &str = r#"
n=0 prompt= model=m/think effort=low n=0 prompt= model=m/think effort=low
printf 'start\n' >> "$1.methods" printf 'start\n' >> "$1.methods"
# With TRAP_INT set: SIGINT creates <log>.int and exits 0.
if [ -n "$TRAP_INT" ]; then
trap ': > "$1.int"; exit 0' INT
fi
# With GRANDCHILD set: a background `sleep`, its pid in <log>.grandchild. # With GRANDCHILD set: a background `sleep`, its pid in <log>.grandchild.
# `&` starts it with SIGINT ignored; `env` puts the default back.
if [ -n "$GRANDCHILD" ]; then if [ -n "$GRANDCHILD" ]; then
sh -c 'echo $$ >"$1.grandchild"; exec sleep 300' sh "$1" >/dev/null 2>&1 & sh -c 'echo $$ >"$1.grandchild"; exec env --default-signal=INT sleep 300' sh "$1" >/dev/null 2>&1 &
fi fi
update() { update() {
printf '{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"%s","update":%s}}\n' "$1" "$2" printf '{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"%s","update":%s}}\n' "$1" "$2"
@ -1529,6 +1536,33 @@ done
assert_gone(grandchild).await; assert_gone(grandchild).await;
} }
/// SIGINT rather than SIGKILL: an agent only stops the commands it
/// started itself if it gets to shut down.
#[tokio::test]
async fn dropping_the_runtime_lets_its_agent_shut_down_on_sigint() {
let dir = tempfile::tempdir().unwrap();
let runtime = agent(
dir.path(),
"ok",
&[("TRAP_INT", "1")],
PercentPolicy::default(),
);
runtime
.run(&config(dir.path()), "one", &NoopSink)
.await
.unwrap();
drop(runtime);
let marker = dir.path().join("prompts.int");
for _ in 0..250 {
if marker.exists() {
break;
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
assert!(marker.exists(), "the agent never got SIGINT");
}
fn sent(session: &str, text: &str) -> (String, String) { fn sent(session: &str, text: &str) -> (String, String) {
(session.to_owned(), text.to_owned()) (session.to_owned(), text.to_owned())
} }

View file

@ -31,9 +31,10 @@ type Reply = Result<Value, AcpError>;
type Pending = Arc<Mutex<HashMap<u64, (&'static str, oneshot::Sender<Reply>)>>>; type Pending = Arc<Mutex<HashMap<u64, (&'static str, oneshot::Sender<Reply>)>>>;
/// The agent runs as the leader of its own process group; dropping the /// The agent runs as the leader of its own process group; dropping the
/// connection `SIGKILL`s that group, so the commands the agent started die /// connection sends that group SIGINT, so the agent can stop the commands
/// with it. A descendant that left the group (`setsid`, `setpgid`) is not /// it started, and they get the signal too. Nothing escalates: an agent
/// reached. /// that ignores SIGINT keeps running. A descendant that left the group
/// (`setsid`, `setpgid`) is not reached.
pub(super) struct Connection { pub(super) struct Connection {
child: Child, child: Child,
/// The agent's pid, which is its group's id. Read at spawn: a reaped /// The agent's pid, which is its group's id. Read at spawn: a reaped
@ -66,7 +67,6 @@ impl Connection {
.stdout(Stdio::piped()) .stdout(Stdio::piped())
.stderr(Stdio::piped()) .stderr(Stdio::piped())
.process_group(0) .process_group(0)
.kill_on_drop(true)
.spawn() .spawn()
.map_err(|source| AcpError::Spawn { .map_err(|source| AcpError::Spawn {
program: command.command.clone(), program: command.command.clone(),
@ -182,7 +182,7 @@ impl Drop for Connection {
// SAFETY: plain `kill(2)`; `-pgid` targets the agent's own // SAFETY: plain `kill(2)`; `-pgid` targets the agent's own
// process group, which `process_group(0)` made it lead. // process group, which `process_group(0)` made it lead.
unsafe { unsafe {
libc::kill(-pgid, libc::SIGKILL); libc::kill(-pgid, libc::SIGINT);
} }
} }
} }

View file

@ -187,10 +187,6 @@ struct StatusArgs {
struct InterruptArgs { struct InterruptArgs {
/// The running session's name to signal. /// The running session's name to signal.
name: String, name: String,
/// `true` sends SIGKILL immediately; `false` (default) sends SIGINT,
/// letting claude shut down cleanly if it's already mid-response.
#[serde(default)]
force: bool,
} }
#[derive(Clone)] #[derive(Clone)]
@ -276,11 +272,13 @@ impl SubagentMcp {
re-prompted toward it afterwards, and `continue` is what restarts it. Only works \ re-prompted toward it afterwards, and `continue` is what restarts it. Only works \
once it's actually running — a `start`/`continue` still in its brief window before \ once it's actually running — a `start`/`continue` still in its brief window before \
the process is confirmed spawned refuses interrupt too (nothing to signal yet; \ the process is confirmed spawned refuses interrupt too (nothing to signal yet; \
retry shortly), same as a name with nothing tracked at all. `force: true` for \ retry shortly), same as a name with nothing tracked at all. On claude it sends \
SIGKILL, otherwise SIGINT." SIGINT, which claude answers by stopping the commands it started and exiting; \
nothing escalates past it. Returns at once; the name stays claimed until claude \
has exited, and `status` says `stopping` until then."
)] )]
fn interrupt(&self, Parameters(args): Parameters<InterruptArgs>) -> String { fn interrupt(&self, Parameters(args): Parameters<InterruptArgs>) -> String {
match session::interrupt(&self.state, &args.name, args.force) { match session::interrupt(&self.state, &args.name) {
Ok(msg) => msg, Ok(msg) => msg,
Err(e) => format!("interrupt error: {e:#}"), Err(e) => format!("interrupt error: {e:#}"),
} }

View file

@ -5,8 +5,8 @@
//! continuation loop records. //! continuation loop records.
//! //!
//! **No task files, no restart recovery.** The daemon's only state is an //! **No task files, no restart recovery.** The daemon's only state is an
//! in-memory `name -> Option<Stopper>` map (see `State`'s own doc for the //! in-memory `name -> Slot` map (see `State`'s own doc for the rest of the
//! rest of the maps and for what the `None`/`Some` split is for) — all of //! maps, and `Slot`'s for what each state of a claimed name means) — all of
//! it living for exactly as long as the process is. A daemon restart means //! it living for exactly as long as the process is. A daemon restart means
//! whatever was running gets killed //! whatever was running gets killed
//! with it (`tokio`'s own child-process drop semantics), not adopted, and //! with it (`tokio`'s own child-process drop semantics), not adopted, and
@ -230,11 +230,33 @@ enum SubagentRuntime {
enum Stopper { enum Stopper {
/// Signals the turn's `claude` process. /// Signals the turn's `claude` process.
Claude(Cancel), Claude(Cancel),
/// Sends the ACP agent `session/cancel`; one that ignores it is killed /// Sends the ACP agent `session/cancel`; the runtime sends SIGINT to one
/// by the runtime after its own grace period. /// that ignores it for its own grace period.
Acp(Canceller), 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),
/// `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
/// name.
Stopping,
}
/// Which [`Slot`] a claimed name is in, without its cancel handle.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Occupancy {
Starting,
Running,
Stopping,
}
/// What settled a resumed turn's bounded wait — the one thing `continue` /// What settled a resumed turn's bounded wait — the one thing `continue`
/// blocks on before it answers. /// blocks on before it answers.
enum ResumeVerdict { enum ResumeVerdict {
@ -333,13 +355,13 @@ fn searched_location(config: &Config) -> Option<String> {
/// and where to push the completion todo. `Arc`-wrapped so the background /// and where to push the completion todo. `Arc`-wrapped so the background
/// task driving a run outlives the tool call that started it. /// task driving a run outlives the tool call that started it.
/// ///
/// The map value is `Option<Stopper>`: `None` means `name` is reserved for /// The map value is a `Slot`: `Starting` means `name` is reserved for an
/// an in-flight `start`/`continue` that hasn't reached a confirmed /// in-flight `start`/`continue` that hasn't reached a confirmed spawn yet;
/// spawn yet; `Some(stopper)` means a real turn is tracked and /// `Running` means a real turn is tracked and interruptible; `Stopping`
/// interruptible. The `None` state exists to close a real TOCTOU window a /// means it was interrupted and its process is still winding down. Each
/// reviewer caught in the original check-then-insert version: checking "is /// claims the name. `Starting` closes a TOCTOU window: checking "is `name`
/// `name` free" and committing to it are two different lock acquisitions /// free" and committing to it are two different lock acquisitions unless
/// unless the check *is* the reservation — see `reserve`. /// the check *is* the reservation — see `reserve`.
/// ///
/// `killed` is the other half of that map's story: `running` says what is in /// `killed` is the other half of that map's story: `running` says what is in
/// flight *now*, `killed` remembers the names whose last turn ended on a /// flight *now*, `killed` remembers the names whose last turn ended on a
@ -359,7 +381,7 @@ fn searched_location(config: &Config) -> Option<String> {
/// name across directories at your own risk, the tool doesn't disambiguate /// name across directories at your own risk, the tool doesn't disambiguate
/// it (flagged in review when `dir` was added). /// it (flagged in review when `dir` was added).
pub struct State { pub struct State {
running: Mutex<HashMap<String, Option<Stopper>>>, running: Mutex<HashMap<String, Slot>>,
dirs: Mutex<HashMap<String, String>>, dirs: Mutex<HashMap<String, String>>,
killed: Mutex<HashMap<String, i32>>, killed: Mutex<HashMap<String, i32>>,
last_event: Mutex<HashMap<String, Instant>>, last_event: Mutex<HashMap<String, Instant>>,
@ -547,17 +569,17 @@ impl State {
} }
} }
/// `Some(true)` — a live process is tracked, interruptible. `Some(false)` /// Which [`Slot`] `name` is in; `None` with nothing tracked under it.
/// — the name is claimed but no process is confirmed under it: an fn occupancy(&self, name: &str) -> Option<Occupancy> {
/// in-flight start/continue that hasn't spawned yet, or the gap between
/// one continued turn's child exiting and the next one spawning (see
/// `between_turns`). `None` — nothing tracked under `name` at all.
fn occupancy(&self, name: &str) -> Option<bool> {
self.running self.running
.lock() .lock()
.unwrap_or_else(PoisonError::into_inner) .unwrap_or_else(PoisonError::into_inner)
.get(name) .get(name)
.map(Option::is_some) .map(|slot| match slot {
Slot::Starting => Occupancy::Starting,
Slot::Running(_) => Occupancy::Running,
Slot::Stopping => Occupancy::Stopping,
})
} }
/// Atomically claim `name` for an in-flight start/continue: check /// Atomically claim `name` for an in-flight start/continue: check
@ -572,21 +594,21 @@ impl State {
if running.contains_key(name) { if running.contains_key(name) {
return false; return false;
} }
running.insert(name.to_owned(), None); running.insert(name.to_owned(), Slot::Starting);
true true
} }
/// Release a reservation that never made it to a real spawn (an error /// Release a reservation that never made it to a real spawn (an error
/// on the slow path between `reserve` and `Claude::spawn` succeeding). /// on the slow path between `reserve` and `Claude::spawn` succeeding).
/// A no-op if the entry was already upgraded to `Some` — this only ever /// A no-op if the entry was already upgraded to `Running` — this only
/// clears a still-`None` placeholder, never a live process. /// ever clears a still-`Starting` placeholder, never a live process.
/// ///
/// The signal token the failed call minted goes with it: no process ever /// The signal token the failed call minted goes with it: no process ever
/// read that URL, and a token outliving the call that minted it is the /// read that URL, and a token outliving the call that minted it is the
/// one way a route could resolve to a session that isn't there. /// one way a route could resolve to a session that isn't there.
fn release_reservation(&self, name: &str) { fn release_reservation(&self, name: &str) {
let mut running = self.running.lock().unwrap_or_else(PoisonError::into_inner); let mut running = self.running.lock().unwrap_or_else(PoisonError::into_inner);
let released = matches!(running.get(name), Some(None)); let released = matches!(running.get(name), Some(Slot::Starting));
if released { if released {
running.remove(name); running.remove(name);
} }
@ -742,14 +764,14 @@ impl State {
} }
} }
/// Upgrade `name`'s `None` reservation to a real, interruptible process. /// Upgrade `name`'s `Starting` reservation to a real, interruptible
/// Same key as the reservation, so there is no window in which `name` /// process. Same key as the reservation, so there is no window in which
/// reads as unoccupied between the two. /// `name` reads as unoccupied between the two.
fn track(&self, name: &str, stopper: Stopper) { fn track(&self, name: &str, stopper: Stopper) {
self.running self.running
.lock() .lock()
.unwrap_or_else(PoisonError::into_inner) .unwrap_or_else(PoisonError::into_inner)
.insert(name.to_owned(), Some(stopper)); .insert(name.to_owned(), Slot::Running(stopper));
} }
/// Drop `name`'s cancel handle but keep the name claimed — the moment /// Drop `name`'s cancel handle but keep the name claimed — the moment
@ -763,7 +785,7 @@ impl State {
self.running self.running
.lock() .lock()
.unwrap_or_else(PoisonError::into_inner) .unwrap_or_else(PoisonError::into_inner)
.insert(name.to_owned(), None); .insert(name.to_owned(), Slot::Starting);
} }
/// Set (or clear) `name`'s goal and start its turn budget over at turn /// Set (or clear) `name`'s goal and start its turn budget over at turn
@ -1286,7 +1308,7 @@ pub async fn start(state: &Arc<State>, req: StartRequest) -> anyhow::Result<Stri
// process spawned. // process spawned.
let role = req.role.as_deref().map(crate::role::load).transpose()?; let role = req.role.as_deref().map(crate::role::load).transpose()?;
if !state.reserve(name) { if !state.reserve(name) {
anyhow::bail!("subagent `{name}` is already running — use `continue` or `interrupt`"); return Err(name_taken(state, name, "use `continue` or `interrupt`"));
} }
// Only commit the remembered `dir` now that `reserve` has actually // Only commit the remembered `dir` now that `reserve` has actually
// claimed `name` — see `resolve_dir`'s doc for why the order matters, // claimed `name` — see `resolve_dir`'s doc for why the order matters,
@ -1460,9 +1482,11 @@ pub async fn continue_(
validate_name(name)?; validate_name(name)?;
state.allow_model(model.as_deref()).await?; state.allow_model(model.as_deref()).await?;
if !state.reserve(name) { if !state.reserve(name) {
anyhow::bail!( return Err(name_taken(
"subagent `{name}` is already running — use `interrupt` first if you meant to redirect it" state,
); name,
"use `interrupt` first if you meant to redirect it",
));
} }
// Read before the spawn, which clears the record as soon as the new turn // Read before the spawn, which clears the record as soon as the new turn
// is confirmed — this reply is the last chance to mention it. // is confirmed — this reply is the last chance to mention it.
@ -2179,7 +2203,7 @@ fn describe_turns(turns: Option<(u32, u32)>) -> String {
/// them is a fact about the same session. /// them is a fact about the same session.
struct StatusFacts<'a> { struct StatusFacts<'a> {
name: &'a str, name: &'a str,
occupancy: Option<bool>, occupancy: Option<Occupancy>,
killed: Option<i32>, killed: Option<i32>,
session_exists: bool, session_exists: bool,
last_event_age: Option<Duration>, last_event_age: Option<Duration>,
@ -2224,7 +2248,7 @@ fn describe_status(facts: &StatusFacts<'_>) -> anyhow::Result<String> {
let name = facts.name; let name = facts.name;
let turns = describe_turns(facts.turns); let turns = describe_turns(facts.turns);
match facts.occupancy { match facts.occupancy {
Some(true) => { Some(Occupancy::Running) => {
return Ok(format!( return Ok(format!(
"subagent `{name}` is running — its turn is still in flight, so there's nothing \ "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 run ends, or \ to do but let it work: the daemon pushes a todo when the run ends, or \
@ -2233,7 +2257,7 @@ fn describe_status(facts: &StatusFacts<'_>) -> anyhow::Result<String> {
describe_pending_signal(facts.stop.as_ref()), describe_pending_signal(facts.stop.as_ref()),
)); ));
} }
Some(false) => { Some(Occupancy::Starting) => {
return Ok(format!( return Ok(format!(
"subagent `{name}` is starting — the name is claimed but no process is confirmed \ "subagent `{name}` is starting — the name is claimed but no process is confirmed \
under it yet, either because a `start`/`continue` hasn't spawned one or because \ under it yet, either because a `start`/`continue` hasn't spawned one or because \
@ -2241,6 +2265,15 @@ fn describe_status(facts: &StatusFacts<'_>) -> anyhow::Result<String> {
shortly rather than starting anything else under this name.{turns}" shortly rather than starting anything else under this name.{turns}"
)); ));
} }
Some(Occupancy::Stopping) => {
return Ok(format!(
"subagent `{name}` is stopping — `interrupt` sent it SIGINT and its process \
hasn't exited yet. The name stays claimed until it does, so `start` and \
`continue` are refused meanwhile. Nothing escalates past that SIGINT: one that \
never exits needs killing by hand.{turns}{}",
describe_liveness(facts.last_event_age),
));
}
None => {} None => {}
} }
if let Some(signal) = facts.killed { if let Some(signal) = facts.killed {
@ -2327,57 +2360,60 @@ fn describe_stopped(name: &str, stop: &StopReason, turns: &str) -> String {
} }
} }
/// Signal `name`'s running process — `force` picks SIGKILL over SIGINT (see /// Stop `name`'s running turn: SIGINT to claude's process group (see
/// `hive_claude::Cancel::cancel`). Refuses a name with nothing running: no /// `hive_claude::Cancel::cancel`), or `session/cancel` on ACP. Nothing
/// entry at all, or one still in the brief not-yet-spawned window (nothing /// escalates to SIGKILL: each command claude's Bash tool starts leads its
/// to signal yet — the reservation is put back so a concurrent /// own session, and only claude shutting down on SIGINT stops it.
/// `start`/`continue` for the same name still gets refused).
/// ///
/// This stops a goal run, not just the turn in it — and it does so by /// Returns without waiting. A claude turn's name stays claimed as
/// **recording `StopReason::Cancelled`**, not by relying on how the child /// `Slot::Stopping` until the turn's own `finish_turn` clears it, so no
/// dies. /// second claude starts on the session while the first is winding down.
/// ///
/// That distinction is the entire fix. The signal alone is not enough: the /// It stops the goal run, not just the turn, by **recording
/// default is SIGINT, claude handles SIGINT by emitting its terminal /// `StopReason::Cancelled`**: claude answers SIGINT by emitting its
/// `result` event and exiting 0, and a zero exit is `TurnEnd::Complete` — /// terminal `result` event and exiting 0, which is `TurnEnd::Complete`, so
/// so `spawn_and_track`'s killed-turn early return never fires and the loop /// only the recorded reason, which `plan_after_turn` checks before the
/// goes straight on to spawn turn three of five. The recorded reason is what /// goal, keeps the loop from re-prompting. It is written while the
/// `plan_after_turn` reads, and it is checked before the goal, so the run /// `running` lock is held, so a turn that ends on its own in between still
/// stops whichever way the child ends up exiting. `force`'s SIGKILL does /// finds it.
/// produce a `TurnEnd::Killed`; recording the reason anyway costs nothing
/// and keeps one path, and `status` still prefers the kill record it
/// already had.
///
/// The reason is written while the `running` lock is still held, so there is
/// no instant in which the name has lost its cancel handle but not yet
/// gained its stop reason — a turn ending of its own accord in that gap
/// would find nothing to stop it and start the next one.
/// ///
/// # Errors /// # Errors
/// ///
/// An invalid name, nothing tracked under `name`, or nothing to signal right /// An invalid name, nothing tracked under `name`, nothing to signal right
/// now (a spawn or ACP turn not yet live, or a goal run between turns). /// now (a spawn or ACP turn not yet live, or a goal run between turns; the
pub fn interrupt(state: &State, name: &str, force: bool) -> anyhow::Result<String> { /// reservation is put back, so a concurrent `start`/`continue` is still
/// refused), or a claude already signalled and not yet exited.
pub fn interrupt(state: &State, name: &str) -> anyhow::Result<String> {
validate_name(name)?; validate_name(name)?;
let mut running = state.running.lock().unwrap_or_else(PoisonError::into_inner); let mut running = state.running.lock().unwrap_or_else(PoisonError::into_inner);
let mut claimed = "";
match running.remove(name) { match running.remove(name) {
None => anyhow::bail!("no subagent named `{name}` is currently running"), None => anyhow::bail!("no subagent named `{name}` is currently running"),
Some(None) => { Some(Slot::Starting) => {
running.insert(name.to_owned(), None); running.insert(name.to_owned(), Slot::Starting);
return Err(still_starting(name)); return Err(still_starting(name));
} }
Some(Some(Stopper::Claude(cancel))) => { Some(Slot::Stopping) => {
state.record_stop(name, StopReason::Cancelled); running.insert(name.to_owned(), Slot::Stopping);
drop(running); anyhow::bail!(
cancel.cancel(force); "subagent `{name}` was already interrupted and its process hasn't exited yet — \
`status` says when it has"
);
} }
// `force` has no stronger form on ACP: the runtime already kills an Some(Slot::Running(Stopper::Claude(cancel))) => {
// agent that ignores `session/cancel`. `cancel()` is `false` when no state.record_stop(name, StopReason::Cancelled);
// turn is in flight (the agent not prompted yet, or a turn just running.insert(name.to_owned(), Slot::Stopping);
// ended), and then nothing was cancelled. The stop goes in before the drop(running);
// cancel and comes out under the same `stops` lock, so cancel.cancel(false);
// `plan_after_turn` never reads a cancel that didn't happen. claimed = " The name stays claimed until claude has exited; `status` says `stopping` \
Some(Some(Stopper::Acp(canceller))) => { 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 mut stops = state.stops.lock().unwrap_or_else(PoisonError::into_inner);
let prior = stops.insert(name.to_owned(), StopReason::Cancelled); let prior = stops.insert(name.to_owned(), StopReason::Cancelled);
if !canceller.cancel() { if !canceller.cancel() {
@ -2386,7 +2422,7 @@ pub fn interrupt(state: &State, name: &str, force: bool) -> anyhow::Result<Strin
None => stops.remove(name), None => stops.remove(name),
}; };
drop(stops); drop(stops);
running.insert(name.to_owned(), Some(Stopper::Acp(canceller))); running.insert(name.to_owned(), Slot::Running(Stopper::Acp(canceller)));
return Err(still_starting(name)); return Err(still_starting(name));
} }
} }
@ -2394,10 +2430,24 @@ pub fn interrupt(state: &State, name: &str, force: bool) -> anyhow::Result<Strin
Ok(format!( Ok(format!(
"interrupt sent to subagent `{name}` — the run is cancelled, not just the turn that was \ "interrupt sent to subagent `{name}` — the run is cancelled, not just the turn that was \
in flight, so no further goal turn will start. `continue` is what restarts it, with a \ in flight, so no further goal turn will start. `continue` is what restarts it, with a \
fresh turn allowance." fresh turn allowance.{claimed}"
)) ))
} }
/// `start`'s and `continue`'s refusal for a name `reserve` found claimed:
/// `hint` says what to do about a live session, and an interrupted one that
/// hasn't exited yet gets its own answer.
fn name_taken(state: &State, name: &str, hint: &str) -> anyhow::Error {
if state.occupancy(name) == Some(Occupancy::Stopping) {
anyhow::anyhow!(
"subagent `{name}` is stopping — it was interrupted and its process hasn't exited \
yet; the name is free once it has (`status` says when)"
)
} else {
anyhow::anyhow!("subagent `{name}` is already running — {hint}")
}
}
/// `interrupt`'s refusal for a name that is claimed but has nothing to /// `interrupt`'s refusal for a name that is claimed but has nothing to
/// signal yet. /// signal yet.
fn still_starting(name: &str) -> anyhow::Error { fn still_starting(name: &str) -> anyhow::Error {
@ -2539,7 +2589,7 @@ mod tests {
state.reserve("n"); state.reserve("n");
assert_eq!( assert_eq!(
state.occupancy("n"), state.occupancy("n"),
Some(false), Some(Occupancy::Starting),
"reserved-but-not-yet-spawned must read as occupied-but-not-running" "reserved-but-not-yet-spawned must read as occupied-but-not-running"
); );
} }
@ -2984,7 +3034,7 @@ mod tests {
// has to carry its own explanation — checked one state at a time, // has to carry its own explanation — checked one state at a time,
// which is all a caller ever gets back. // which is all a caller ever gets back.
let running = describe_status(&StatusFacts { let running = describe_status(&StatusFacts {
occupancy: Some(true), occupancy: Some(Occupancy::Running),
..StatusFacts::new("n") ..StatusFacts::new("n")
}) })
.expect("running status"); .expect("running status");
@ -2993,7 +3043,7 @@ mod tests {
"a running answer must say the turn is in flight and how to stop it: {running}" "a running answer must say the turn is in flight and how to stop it: {running}"
); );
let starting = describe_status(&StatusFacts { let starting = describe_status(&StatusFacts {
occupancy: Some(false), occupancy: Some(Occupancy::Starting),
..StatusFacts::new("n") ..StatusFacts::new("n")
}) })
.expect("starting status"); .expect("starting status");
@ -3238,7 +3288,7 @@ mod tests {
#[test] #[test]
fn a_running_status_reports_the_age_and_an_idle_one_does_not() { fn a_running_status_reports_the_age_and_an_idle_one_does_not() {
let running = describe_status(&StatusFacts { let running = describe_status(&StatusFacts {
occupancy: Some(true), occupancy: Some(Occupancy::Running),
last_event_age: Some(Duration::from_secs(4)), last_event_age: Some(Duration::from_secs(4)),
..StatusFacts::new("n") ..StatusFacts::new("n")
}) })
@ -3249,7 +3299,7 @@ mod tests {
{running}" {running}"
); );
let wedged = describe_status(&StatusFacts { let wedged = describe_status(&StatusFacts {
occupancy: Some(true), occupancy: Some(Occupancy::Running),
last_event_age: Some(Duration::from_mins(15)), last_event_age: Some(Duration::from_mins(15)),
..StatusFacts::new("n") ..StatusFacts::new("n")
}) })
@ -3281,7 +3331,7 @@ mod tests {
// turn can finish between the two. The answer drops the age rather // turn can finish between the two. The answer drops the age rather
// than inventing one. // than inventing one.
let raced = describe_status(&StatusFacts { let raced = describe_status(&StatusFacts {
occupancy: Some(true), occupancy: Some(Occupancy::Running),
..StatusFacts::new("n") ..StatusFacts::new("n")
}) })
.expect("running status"); .expect("running status");
@ -3710,9 +3760,7 @@ mod tests {
} }
assert_eq!(turns_spawned(&log), 1, "the first turn never started"); assert_eq!(turns_spawned(&log), 1, "the first turn never started");
// An operator's plain `interrupt`, not `force` — SIGINT is the interrupt(&state, "n").expect("a running turn is interruptible");
// default and the case that was broken.
interrupt(&state, "n", false).expect("a running turn is interruptible");
// Generous: the fake logs its invocation as its first act, so a // Generous: the fake logs its invocation as its first act, so a
// re-prompt shows up here almost immediately once the loop decides // re-prompt shows up here almost immediately once the loop decides
@ -3776,53 +3824,131 @@ mod tests {
#[tokio::test] #[tokio::test]
async fn an_interrupt_stops_the_commands_the_subagent_started() { async fn an_interrupt_stops_the_commands_the_subagent_started() {
for force in [false, true] { let dir = scratch_dir("group");
let dir = std::env::temp_dir().join(format!( let pidfile = dir.join("grandchild.pid");
"hive-subagent-group-{}-{:?}-{force}", let state = Arc::new(mid_run("n", "carry on indefinitely", 5));
std::process::id(), let config = Config {
std::thread::current().id() program: Some(busy_claude(&dir, &pidfile)),
)); ..Config::default()
std::fs::create_dir_all(&dir).expect("scratch dir"); };
let pidfile = dir.join("grandchild.pid"); spawn_and_track(
let state = Arc::new(mid_run("n", "carry on indefinitely", 5)); &state,
let config = Config { "n",
program: Some(busy_claude(&dir, &pidfile)), &config,
..Config::default() &Attach::Create("n".to_owned()),
}; "turn one".to_owned(),
spawn_and_track( None,
&state, )
"n", .expect("the fake claude spawns");
&config, let grandchild = read_pid(&pidfile).await;
&Attach::Create("n".to_owned()),
"turn one".to_owned(),
None,
)
.expect("the fake claude spawns");
let grandchild = read_pid(&pidfile).await;
interrupt(&state, "n", force).expect("a running turn is interruptible"); interrupt(&state, "n").expect("a running turn is interruptible");
let mut alive = is_running(grandchild); let mut alive = is_running(grandchild);
for _ in 0..250 { for _ in 0..250 {
if !alive { if !alive {
break; break;
}
tokio::time::sleep(Duration::from_millis(20)).await;
alive = is_running(grandchild);
} }
if alive { tokio::time::sleep(Duration::from_millis(20)).await;
// SAFETY: `kill(2)` on a pid this test started; keeps a failed alive = is_running(grandchild);
// run from leaving a 300s `sleep` behind.
unsafe {
libc::kill(grandchild, libc::SIGKILL);
}
}
std::fs::remove_dir_all(&dir).ok();
assert!(
!alive,
"grandchild {grandchild} outlived interrupt(force = {force})"
);
} }
if alive {
// SAFETY: `kill(2)` on a pid this test started; keeps a failed
// run from leaving a 300s `sleep` behind.
unsafe {
libc::kill(grandchild, libc::SIGKILL);
}
}
std::fs::remove_dir_all(&dir).ok();
assert!(!alive, "grandchild {grandchild} outlived the interrupt");
}
/// A stand-in for claude that shuts down on SIGINT the way claude does:
/// it writes `interrupted` once the signal arrives, then exits 0 as soon
/// as `release` exists, or after 10s. It writes `ready` once its trap is
/// set. A SIGKILL leaves no `interrupted` behind.
fn graceful_claude(dir: &Path) -> PathBuf {
let script = dir.join("graceful-claude");
write_executable(
&script,
&format!(
"#!/bin/sh\ncd '{}'\ntrap ': > interrupted; i=0; until [ -e release ] || \
[ $i -ge 200 ]; do sleep 0.05; i=$((i+1)); done; kill $!; exit 0' INT\n\
sleep 300 >/dev/null 2>&1 &\n: > ready\nwait\n",
dir.display()
),
);
script
}
/// Spawn [`graceful_claude`] as `n`'s turn and wait until its trap is set.
async fn spawn_graceful(state: &Arc<State>, dir: &Path) {
let config = Config {
program: Some(graceful_claude(dir)),
..Config::default()
};
spawn_and_track(
state,
"n",
&config,
&Attach::Create("n".to_owned()),
"turn one".to_owned(),
None,
)
.expect("the fake claude spawns");
until("the fake claude sets its trap", || {
dir.join("ready").exists()
})
.await;
}
#[tokio::test]
async fn an_interrupt_lets_claude_shut_down_on_sigint() {
let dir = scratch_dir("sigint");
std::fs::write(dir.join("release"), "").expect("release file");
let state = Arc::new(mid_run("n", "carry on indefinitely", 5));
spawn_graceful(&state, &dir).await;
interrupt(&state, "n").expect("a running turn is interruptible");
until("the run ends", || state.occupancy("n").is_none()).await;
let interrupted = dir.join("interrupted").exists();
std::fs::remove_dir_all(&dir).ok();
assert!(
interrupted,
"claude must get SIGINT, which it shuts its tool commands down on, never SIGKILL"
);
assert_eq!(state.killed_by("n"), None, "it exited on its own");
}
#[tokio::test]
async fn an_interrupted_name_stays_claimed_until_claude_has_exited() {
let dir = scratch_dir("stopping");
let state = Arc::new(mid_run("n", "carry on indefinitely", 5));
spawn_graceful(&state, &dir).await;
interrupt(&state, "n").expect("a running turn is interruptible");
until("the fake claude gets the SIGINT", || {
dir.join("interrupted").exists()
})
.await;
assert_eq!(state.occupancy("n"), Some(Occupancy::Stopping));
let status = status(&state, "n", None).expect("status");
assert!(status.contains("is stopping"), "{status}");
let refused = continue_(&state, "n", "more".to_owned(), None, None, None)
.await
.expect_err("a second claude must not start on the session");
assert!(refused.to_string().contains("is stopping"), "{refused}");
assert!(interrupt(&state, "n").is_err(), "it was already signalled");
std::fs::write(dir.join("release"), "").expect("release file");
until("the run ends", || state.occupancy("n").is_none()).await;
std::fs::remove_dir_all(&dir).ok();
assert!(
state.reserve("n"),
"the name is free once claude has exited"
);
} }
#[test] #[test]
@ -4079,7 +4205,12 @@ mod tests {
#[test] #[test]
fn status_reports_the_turn_counter_in_every_state_a_goal_run_reaches() { fn status_reports_the_turn_counter_in_every_state_a_goal_run_reaches() {
for occupancy in [Some(true), Some(false), None] { for occupancy in [
Some(Occupancy::Running),
Some(Occupancy::Starting),
Some(Occupancy::Stopping),
None,
] {
let answer = describe_status(&StatusFacts { let answer = describe_status(&StatusFacts {
occupancy, occupancy,
session_exists: true, session_exists: true,
@ -4094,7 +4225,7 @@ mod tests {
); );
} }
let goalless = describe_status(&StatusFacts { let goalless = describe_status(&StatusFacts {
occupancy: Some(true), occupancy: Some(Occupancy::Running),
..StatusFacts::new("n") ..StatusFacts::new("n")
}) })
.expect("status"); .expect("status");
@ -4110,7 +4241,7 @@ mod tests {
// it is blocked, its turn hasn't ended yet, and plain "running" would // it is blocked, its turn hasn't ended yet, and plain "running" would
// tell the parent to leave it alone. // tell the parent to leave it alone.
let answer = describe_status(&StatusFacts { let answer = describe_status(&StatusFacts {
occupancy: Some(true), occupancy: Some(Occupancy::Running),
stop: Some(StopReason::NeedHelp("no credential".to_owned())), stop: Some(StopReason::NeedHelp("no credential".to_owned())),
..StatusFacts::new("n") ..StatusFacts::new("n")
}) })
@ -4120,7 +4251,7 @@ mod tests {
"both facts are true at once and both have to be in the answer: {answer}" "both facts are true at once and both have to be in the answer: {answer}"
); );
let quiet = describe_status(&StatusFacts { let quiet = describe_status(&StatusFacts {
occupancy: Some(true), occupancy: Some(Occupancy::Running),
stop: Some(StopReason::Done), stop: Some(StopReason::Done),
..StatusFacts::new("n") ..StatusFacts::new("n")
}) })
@ -4633,10 +4764,8 @@ done
turns_spawned(&dir.join("claude-turns")) == 1 turns_spawned(&dir.join("claude-turns")) == 1
}) })
.await; .await;
interrupt(&state, "n", false).expect("a running claude turn is interruptible"); interrupt(&state, "n").expect("a running claude turn is interruptible");
// `interrupt` drops the tracking entry itself; the run is over once until("the run ends", || state.occupancy("n").is_none()).await;
// `finish_turn` has cleared its liveness clock too.
until("the run ends", || state.last_event_age("n").is_none()).await;
// This `start` had no goal, so the fake claude's clean exit after // This `start` had no goal, so the fake claude's clean exit after
// SIGINT (`TurnEnd::Complete`) is exactly the case that used to // SIGINT (`TurnEnd::Complete`) is exactly the case that used to
@ -4983,7 +5112,7 @@ echo '{"type":"result","subtype":"success","is_error":false}'
}) })
.await; .await;
interrupt(&state, "n", false).expect("a running ACP turn is interruptible"); interrupt(&state, "n").expect("a running ACP turn is interruptible");
until("the run ends", || state.last_event_age("n").is_none()).await; until("the run ends", || state.last_event_age("n").is_none()).await;
assert_eq!(lines(&dir.join("prompts")).len(), 1, "no second goal turn"); assert_eq!(lines(&dir.join("prompts")).len(), 1, "no second goal turn");
@ -5006,7 +5135,7 @@ echo '{"type":"result","subtype":"success","is_error":false}'
// The test runtime is single-threaded, so the run's task has not // The test runtime is single-threaded, so the run's task has not
// been polled yet: no turn is in flight. // been polled yet: no turn is in flight.
let err = interrupt(&state, "n", false).expect_err("nothing to cancel yet"); let err = interrupt(&state, "n").expect_err("nothing to cancel yet");
assert!(err.to_string().contains("still starting"), "{err}"); assert!(err.to_string().contains("still starting"), "{err}");
assert_eq!(state.stop_reason("n"), None); assert_eq!(state.stop_reason("n"), None);