diff --git a/Cargo.lock b/Cargo.lock index 824935a0..6b2e88f5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1836,9 +1836,9 @@ dependencies = [ [[package]] name = "hive-claude" -version = "0.1.2" +version = "0.1.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1e43e06fd09a7b6d6cca3e70f06d00fa795ceddfe1555121ca0cf4fc578eb9c3" +checksum = "4d7c5638f8f59e438d6416b3048709c9b3d6a10c23872ecf3825e3fe4c1bd7e8" dependencies = [ "libc", "serde", diff --git a/Cargo.toml b/Cargo.toml index 082db1a5..7f73e0ab 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -97,7 +97,7 @@ hive-jobq-metrics = { path = "hive-jobq-metrics" } hive-jobq-wire = { path = "hive-jobq-wire" } hive-log = { path = "hive-log" } hive-core-agent-sock = { path = "hive-core-agent-sock" } -hive-claude = "0.1.2" +hive-claude = "0.1.4" hive-host-sock = { path = "hive-host-sock" } hive-priv-sock = { path = "hive-priv-sock" } hive-runtime = { path = "hive-runtime" } diff --git a/claude-plugins/plugins/base/skills/claude-subagents/SKILL.md b/claude-plugins/plugins/base/skills/claude-subagents/SKILL.md index dc53f709..d7f37b3d 100644 --- a/claude-plugins/plugins/base/skills/claude-subagents/SKILL.md +++ b/claude-plugins/plugins/base/skills/claude-subagents/SKILL.md @@ -27,7 +27,7 @@ skill's recipe for you: start(name, prompt_file, model?, effort?, trigger?) continue(name, prompt, model?, effort?) status(name) -interrupt(name, force?) +interrupt(name) ``` `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 wedged and worth an `interrupt`. Reach for `continue` only once you 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 — everything else in this skill — applies exactly the same whether you're diff --git a/docs/agent-lifecycle/persistence.md b/docs/agent-lifecycle/persistence.md index bcea11ac..31125900 100644 --- a/docs/agent-lifecycle/persistence.md +++ b/docs/agent-lifecycle/persistence.md @@ -209,6 +209,11 @@ Writes are best-effort: `append_stdout` / `append_stderr` / `finish` log a warning on sqlite error and let the build continue. A failed log row never blocks a rebuild. +A row is open (`finished_at IS NULL`) only while its child runs. At startup +`Coordinator::open` closes every row still open as `fail` and appends a +line saying hive-c0re stopped: the previous process died mid-build, and +the retention sweep never deletes open rows. + ### `/harness/hyperhive-model` (per agent) Single-line text file holding a claude model name. The harness only diff --git a/docs/tools/subagent.md b/docs/tools/subagent.md index 69a05b2c..f75aeace 100644 --- a/docs/tools/subagent.md +++ b/docs/tools/subagent.md @@ -166,6 +166,16 @@ 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 undo rather than a session you have to abandon. +On `claude`, the signal is always SIGINT, and it goes to claude's whole +process group. The signal doesn't reach a command that moved itself out of +the group (`setsid`), and claude's Bash tool starts every command that way: +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`, or the path the subagent names when it signals — the daemon appends the stop reason to that file as well, so the artifact you were going to read @@ -338,7 +348,7 @@ page applies as written. On `acp`: - 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 reads a provider key that file leaves unset from the swarm secret store at `swarm/agents//acp-provider`, as the harness does. @@ -347,9 +357,10 @@ On `acp`: 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 system prompt. -- `interrupt` sends `session/cancel`, and `force` changes nothing. The - runtime kills an agent that ignores the cancel for 10s, with every - command it started. +- `interrupt` sends `session/cancel`. The runtime sends SIGINT to the + process group of an agent that ignores the cancel for 10s, and when the + 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 offers that value for its `model` or `thought_level` config option. Any other value does nothing (`hive_runtime::AcpRuntime`). diff --git a/docs/turn-loop/README.md b/docs/turn-loop/README.md index 961f3a6d..ee2327b1 100644 --- a/docs/turn-loop/README.md +++ b/docs/turn-loop/README.md @@ -142,8 +142,8 @@ else a `TurnError`) drives the post-claude branch: | `Err(RateLimited)` | sleep `HIVE_RATE_LIMIT_SLEEP_SECS` (default 300), requeue inflight, status back to `online` | | `Err(AuthFailed)` | emit `needs_login_idle` sentinel, requeue inflight, park in `wait_for_login` | | `Err(SessionNotFound)` | resume + create self-heal both missed ("shouldn't happen"); requeue inflight so the next turn creates fresh — no status park, message not dropped | -| `Err(ApiStall)` | idle watchdog killed claude after `HIVE_TURN_IDLE_SECS` (default 600) of output silence; sleep `HIVE_STALL_SLEEP_SECS` (default 60), requeue inflight, status back to `online` | -| `Err(AgentStall(note))` | the same, for an ACP agent: after `HIVE_TURN_IDLE_SECS` with no `session/update` the harness sent the agent `session/cancel`, and killed it if it ignored that; `note` says which | +| `Err(ApiStall)` | idle watchdog sent claude SIGINT after `HIVE_TURN_IDLE_SECS` (default 600) of output silence and claude exited; sleep `HIVE_STALL_SLEEP_SECS` (default 60), requeue inflight, status back to `online` | +| `Err(AgentStall(note))` | the same, for an ACP agent: after `HIVE_TURN_IDLE_SECS` with no `session/update` the harness sent the agent `session/cancel`, and SIGINT if it ignored that; `note` says which | | `Err(Failed(err))` | route `[system] \`\` claude turn failed:\n` to `operator` via `send_to_operator` | `ApiStall` catches an Anthropic API stall — a multi-retry connection storm where @@ -152,7 +152,11 @@ the stream goes silent for minutes. On `claude` the idle watchdog lives in resets on every stdout line, so a large but still-streaming turn is never cut — only complete output silence for the window trips it. The harness sets the window from `HIVE_TURN_IDLE_SECS` (`0` disables) and maps the driver's -`Error::IdleTimeout` onto `TurnError::ApiStall`. +`Error::IdleTimeout` onto `TurnError::ApiStall`. The watchdog sends claude's +process group SIGINT, so claude stops the commands it started, and then +waits for claude to exit. Nothing escalates to SIGKILL: a claude that +ignores the SIGINT keeps the turn open until someone kills it or the unit +stops. `hive-runtime` applies the same window to an ACP agent's turns. That's also how a provider HTTP 429 ends an ACP turn when the agent retries it without diff --git a/docs/web-ui/agent.md b/docs/web-ui/agent.md index a6e3d07d..ff04e86a 100644 --- a/docs/web-ui/agent.md +++ b/docs/web-ui/agent.md @@ -229,16 +229,17 @@ Slash commands today: - `/help` — list commands locally. - `/clear` — wipe the local terminal view (server history kept). -- `/cancel` — `POST /api/cancel` → host shellouts `pkill -INT - claude`, emits a Note. On an ACP agent it sends the agent - `session/cancel` instead, and kills it if the turn hasn't ended 10s - later. Also surfaces as a `■ cancel turn` button in the state row - while state=thinking. +- `/cancel` — `POST /api/cancel` → SIGINT to the turn's claude and every + command it started (its process group), emits a Note. On an ACP agent + it sends the agent `session/cancel` instead, and SIGINT if the turn + hasn't ended 10s later. Also surfaces as a `■ cancel turn` button in + the state row while state=thinking. - `/compact` — `POST /api/compact` → sets the deferred `Bus::request_compact()` flag and returns immediately. The harness runs the `/compact` at the next turn boundary (end of the in-flight turn, or `turn::run_pending_compact` when idle); output streams into - the live panel. Works mid-turn, not only when idle. + the live panel. Works mid-turn, not only when idle. Either way the live + panel ends with a `/compact done` or `/compact failed: …` note. - `/new-session` — `POST /api/new-session` (confirms first). Arms a one-shot on the Bus; next turn runs without `--continue`, dropping the resume session entirely. @@ -266,8 +267,8 @@ shaped). - `POST /login/{start,code,cancel}` — claude OAuth login flow. Start/cancel emit `LiveEvent::StatusChanged` to flip the badge to/from `needs_login_in_progress`. -- `POST /api/cancel` — SIGINT the in-flight claude turn. Emits a - `LiveEvent::Note`. +- `POST /api/cancel` — SIGINT the in-flight claude turn and the + commands it started. Emits a `LiveEvent::Note`. - `POST /api/compact` — run `/compact` on the persistent session (same MCP config + system prompt + allowed tools as a normal turn — only the stdin payload differs). Flips state to @@ -363,7 +364,8 @@ shaped). `{"": ""}`). Each entry mounts a transparent reverse-proxy at `/extra//` that forwards every request (method, headers, body) to the configured upstream, strips hop-by-hop headers - on both sides, and buffers the full response body (MVP — SSE + on both sides (repeated response headers such as two `Set-Cookie` lines + are all forwarded), and buffers the full response body (MVP — SSE connections will appear as one large response rather than streaming). The `/extra/` namespace ensures user-declared proxies can never conflict with native agent endpoints. Upstream values are either an diff --git a/hive-agent/src/paths.rs b/hive-agent/src/paths.rs index 7c41cd87..dc4ceddc 100644 --- a/hive-agent/src/paths.rs +++ b/hive-agent/src/paths.rs @@ -132,7 +132,37 @@ pub fn claude_dir() -> PathBuf { PathBuf::from("/root/.claude") } -/// Held by every test that points `HYPERHIVE_HARNESS_DIR` at a temp dir: -/// cargo runs a crate's tests on parallel threads of one process. +/// Run `f` with `HYPERHIVE_HARNESS_DIR` pointing at a fresh temp dir, then +/// restore the previous value. Tests that build a `Bus` or read the harness +/// state files go through this: the variable is process-wide, so the calls +/// are serialised on a lock. #[cfg(test)] -pub(crate) static HARNESS_DIR_ENV: std::sync::Mutex<()> = std::sync::Mutex::new(()); +pub fn with_harness_dir(f: impl FnOnce() -> T) -> T { + use std::ffi::OsString; + use std::sync::Mutex; + + static LOCK: Mutex<()> = Mutex::new(()); + + struct Restore(Option); + impl Drop for Restore { + fn drop(&mut self) { + // SAFETY: still holding `LOCK`, the only writer of this variable. + unsafe { + match self.0.take() { + Some(v) => std::env::set_var("HYPERHIVE_HARNESS_DIR", v), + None => std::env::remove_var("HYPERHIVE_HARNESS_DIR"), + } + } + } + } + + let _lock = LOCK + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let dir = tempfile::tempdir().unwrap(); + // Declared after `_lock` so the variable is restored before the lock drops. + let _restore = Restore(std::env::var_os("HYPERHIVE_HARNESS_DIR")); + // SAFETY: serialised by `LOCK`. + unsafe { std::env::set_var("HYPERHIVE_HARNESS_DIR", dir.path()) }; + f() +} diff --git a/hive-agent/src/swarm_inference.rs b/hive-agent/src/swarm_inference.rs index ad97e910..57d96420 100644 --- a/hive-agent/src/swarm_inference.rs +++ b/hive-agent/src/swarm_inference.rs @@ -320,37 +320,30 @@ mod tests { /// persisted files, so clearing it, or a restart, finds the local value. #[test] fn the_swarm_value_is_not_persisted_over_the_local_one() { - let dir = tempfile::tempdir().unwrap(); - let _env = crate::paths::HARNESS_DIR_ENV - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); - let prev = std::env::var_os("HYPERHIVE_HARNESS_DIR"); - // SAFETY: every test that sets this variable holds `HARNESS_DIR_ENV`. - unsafe { std::env::set_var("HYPERHIVE_HARNESS_DIR", dir.path()) }; - let bus = Bus::new(); - std::fs::write(dir.path().join("hyperhive-model"), "sonnet\n").unwrap(); - std::fs::write(dir.path().join("hyperhive-effort"), "low\n").unwrap(); - let target = Harness { - bus: bus.clone(), - nix_model: None, - nix_effort: None, - }; - let mut o = Overridden::default(); - o.apply(&set(Some("opus"), Some(Effort::Max)), &target); - let swarm = (bus.model(), bus.effort()); - let swarm_sources = (bus.model_source(), bus.effort_source()); - let files = ( - crate::harness_state::load_model(), - crate::harness_state::load_effort(), - ); - o.apply(&AgentInference::default(), &target); - let cleared = (bus.model(), bus.effort()); - let cleared_sources = (bus.model_source(), bus.effort_source()); - match prev { - // SAFETY: as above. - Some(v) => unsafe { std::env::set_var("HYPERHIVE_HARNESS_DIR", v) }, - None => unsafe { std::env::remove_var("HYPERHIVE_HARNESS_DIR") }, - } + let (swarm, swarm_sources, files, cleared, cleared_sources) = + crate::paths::with_harness_dir(|| { + let bus = Bus::new(); + let dir = crate::paths::harness_dir(); + std::fs::write(dir.join("hyperhive-model"), "sonnet\n").unwrap(); + std::fs::write(dir.join("hyperhive-effort"), "low\n").unwrap(); + let target = Harness { + bus: bus.clone(), + nix_model: None, + nix_effort: None, + }; + let mut o = Overridden::default(); + o.apply(&set(Some("opus"), Some(Effort::Max)), &target); + let swarm = (bus.model(), bus.effort()); + let swarm_sources = (bus.model_source(), bus.effort_source()); + let files = ( + crate::harness_state::load_model(), + crate::harness_state::load_effort(), + ); + o.apply(&AgentInference::default(), &target); + let cleared = (bus.model(), bus.effort()); + let cleared_sources = (bus.model_source(), bus.effort_source()); + (swarm, swarm_sources, files, cleared, cleared_sources) + }); assert_eq!(swarm, pair("opus", "max")); assert_eq!(files, (Some("sonnet".to_owned()), Some("low".to_owned()))); assert_eq!(cleared, pair("sonnet", "low")); diff --git a/hive-agent/src/turn.rs b/hive-agent/src/turn.rs index ee970c70..156e83ef 100644 --- a/hive-agent/src/turn.rs +++ b/hive-agent/src/turn.rs @@ -50,8 +50,8 @@ const DEFAULT_SESSION_TITLE: &str = "hive-session"; /// capacity limits. const DEFAULT_RATE_LIMIT_SLEEP_SECS: u64 = 300; -/// Idle watchdog window: kill claude and surface `ApiStall` if it produces no -/// stdout for this long. The timer resets on every stdout line, so a large but +/// Idle watchdog window: send claude SIGINT and surface `ApiStall` once it +/// exits, if it produces no stdout for this long. The timer resets on every stdout line, so a large but /// still-streaming turn is never cut — only a complete silence trips it. /// Default is 10 minutes: long enough for legitimate big-context turns, short /// enough to cut a multi-retry Anthropic connection storm (atlas telemetry: @@ -155,7 +155,7 @@ pub async fn write_system_prompt(socket: &Path, label: &str) -> Result /// One claude turn's outcome: `Ok(compacted)` on success, or a [`TurnError`] /// the serve loop must act on. The `compacted` bool is `true` when a /// compaction ran this turn (reactively on overflow, or proactively per the -/// policy — or an operator `/compact` at turn end); it's recorded as +/// policy — or a successful operator `/compact` at turn end); it's recorded as /// `result_kind = "compacted"` in turn stats so the stats page can distinguish /// those turns. Both `Ok(true)` and `Ok(false)` are ack'd; the error cases /// each map to a distinct serve-loop action (see [`emit_turn_end`] and the @@ -189,14 +189,14 @@ pub enum TurnError { /// the wake message, the serve loop requeues it so the next turn retries; /// no status park. SessionNotFound, - /// The harness-side idle watchdog killed claude after `turn_idle_secs()` of + /// The harness-side idle watchdog stopped claude after `turn_idle_secs()` of /// output silence — indicative of an Anthropic API stall (e.g. a multi-retry /// connection storm burning minutes of wall-clock with no stream progress). /// The serve loop parks for `stall_sleep_secs()` and requeues, like the /// rate-limit path — NOT a crash. ApiStall, /// The ACP backend's idle watchdog stopped a turn that sent nothing for - /// `turn_idle_secs()`: the agent was sent `session/cancel`, and killed if + /// `turn_idle_secs()`: the agent was sent `session/cancel`, and SIGINT if /// it ignored that. Carries what happened. An ACP agent retrying a failing /// provider (e.g. HTTP 429) without reporting it ends up here. Handled /// like [`Self::ApiStall`]. @@ -461,7 +461,7 @@ pub async fn drive_turn( // Reflect `Compacting` in the UI like the idle path (`run_pending_compact`) // does; the serve loop resets to `Idle` once this turn returns. bus.set_state(crate::events::TurnState::Compacting); - let _ = session.compact(&config, &sink).await; + let compact_ok = report_compact(bus, session.compact(&config, &sink).await); // If the compact call asked to be woken (the agent's own `compact` // tool with a `wake_prompt`), stash it — the serve loop reads it // back after this turn returns and drives a synthetic follow-up @@ -470,11 +470,38 @@ pub async fn drive_turn( if let Some(prompt) = request.wake_prompt { bus.set_post_compact_wake(prompt); } - return Ok(true); + return with_turn_end_compact(outcome, compact_ok); } outcome } +/// Emit the `/compact done` / `/compact failed: …` note for a finished +/// operator or agent `/compact` and return whether it succeeded. Shared by the +/// turn-end path in [`drive_turn`] and the idle path in [`run_pending_compact`]. +fn report_compact(bus: &Bus, result: Result<(), E>) -> bool { + match result { + Ok(()) => { + bus.emit(LiveEvent::Note { + text: "/compact done".into(), + }); + true + } + Err(e) => { + bus.emit(LiveEvent::Note { + text: format!("/compact failed: {e}"), + }); + false + } + } +} + +/// Fold a turn-end `/compact` into the turn's own outcome. A compaction that +/// ran during the turn (`Ok(true)`) stays recorded even when the turn-end +/// `/compact` failed; a failed `/compact` never turns it into `Ok(false)`. +fn with_turn_end_compact(outcome: TurnOutcome, compact_ok: bool) -> TurnOutcome { + outcome.map(|turn_compacted| turn_compacted || compact_ok) +} + /// Pre-turn auto-reset check. If context is large AND the prompt cache has /// gone cold (idle time >= cache TTL), archive the current session so the /// next wake-up turn's `--resume ` misses and self-heals into a fresh @@ -558,11 +585,11 @@ pub fn emit_turn_end(bus: &Bus, outcome: &TurnOutcome) { bus.emit(LiveEvent::TurnEnd { ok: false, note: Some(format!( - "claude killed after {}s of output silence — API stall suspected, parking + requeueing", + "claude stopped with SIGINT after {}s of output silence — API stall suspected, parking + requeueing", turn_idle_secs() )), }); - tracing::warn!("turn killed: API stall (idle watchdog)"); + tracing::warn!("turn stopped: API stall (idle watchdog)"); } Err(TurnError::AgentStall(note)) => { bus.emit(LiveEvent::TurnEnd { @@ -600,14 +627,7 @@ pub async fn run_pending_compact(files: &TurnFiles, bus: &Bus, session: &AgentSe bus.set_state(crate::events::TurnState::Compacting); let config = claude_config(bus, files); let sink = BusSink::new(bus); - match session.compact(&config, &sink).await { - Ok(()) => bus.emit(LiveEvent::Note { - text: "/compact done".into(), - }), - Err(e) => bus.emit(LiveEvent::Note { - text: format!("/compact failed: {e}"), - }), - } + report_compact(bus, session.compact(&config, &sink).await); bus.set_state(crate::events::TurnState::Idle); if let Some(prompt) = request.wake_prompt { bus.set_post_compact_wake(prompt); @@ -821,8 +841,10 @@ fn archive_session(bus: &Bus, session: &AgentSession) { #[cfg(test)] mod tests { use super::{ - PermissionAsk, TurnError, acp_permits, nonzero_or, parse_u64, runtime_error_to_turn, + PermissionAsk, TurnError, acp_permits, nonzero_or, parse_u64, report_compact, + runtime_error_to_turn, with_turn_end_compact, }; + use crate::events::{Bus, LiveEvent}; use hive_runtime::{AcpError, Error}; #[test] @@ -959,4 +981,49 @@ mod tests { Err(TurnError::Failed(_)) )); } + + fn compact_report(result: Result<(), &str>) -> (bool, Vec<String>) { + crate::paths::with_harness_dir(|| { + let bus = Bus::new(); + let mut rx = bus.subscribe(); + let ok = report_compact(&bus, result); + let mut notes = Vec::new(); + while let Ok(ev) = rx.try_recv() { + if let LiveEvent::Note { text } = ev.event { + notes.push(text); + } + } + (ok, notes) + }) + } + + #[test] + fn a_failed_compact_is_reported_and_not_counted_as_compacted() { + let (compacted, notes) = compact_report(Err("spawn failed")); + assert!(!compacted); + assert_eq!(notes, ["/compact failed: spawn failed"]); + } + + #[test] + fn a_successful_compact_is_reported_and_counted_as_compacted() { + let (compacted, notes) = compact_report(Ok(())); + assert!(compacted); + assert_eq!(notes, ["/compact done"]); + } + + #[test] + fn a_failed_turn_end_compact_keeps_the_turns_own_compaction() { + assert!(matches!(with_turn_end_compact(Ok(true), false), Ok(true))); + } + + #[test] + fn a_failed_turn_end_compact_leaves_a_plain_turn_uncompacted() { + assert!(matches!(with_turn_end_compact(Ok(false), false), Ok(false))); + } + + #[test] + fn a_successful_turn_end_compact_marks_the_turn_compacted() { + assert!(matches!(with_turn_end_compact(Ok(false), true), Ok(true))); + assert!(matches!(with_turn_end_compact(Ok(true), true), Ok(true))); + } } diff --git a/hive-agent/src/web_ui/actions.rs b/hive-agent/src/web_ui/actions.rs index 42932535..cf399f57 100644 --- a/hive-agent/src/web_ui/actions.rs +++ b/hive-agent/src/web_ui/actions.rs @@ -69,7 +69,7 @@ pub(super) async fn post_cancel_turn(State(state): State<AppState>) -> Response .emit(crate::events::LiveEvent::Note { text: note.into() }); return (axum::http::StatusCode::OK, "ok").into_response(); } - let out = super::sigint_claude().await; + let out = super::sigint_claude(); let note = match out { SigintOutcome::Signalled => { // A process actually got signalled — the *next* turn's wake diff --git a/hive-agent/src/web_ui/auth.rs b/hive-agent/src/web_ui/auth.rs index ea02fb3a..bbc664b3 100644 --- a/hive-agent/src/web_ui/auth.rs +++ b/hive-agent/src/web_ui/auth.rs @@ -84,7 +84,7 @@ pub(super) async fn post_login_cancel(State(state): State<AppState>) -> Response /// preservation invariants. pub(super) async fn post_logout(State(state): State<AppState>) -> Response { // Step 1: SIGINT claude (best-effort, matches `post_cancel_turn`). - let _ = super::sigint_claude().await; + let _ = super::sigint_claude(); // Step 2: delete OAuth credential files only — login::clear_session owns // the file set and preserves session-history files alongside them. let dir = crate::paths::claude_dir(); diff --git a/hive-agent/src/web_ui/mod.rs b/hive-agent/src/web_ui/mod.rs index cdae0793..ba036b23 100644 --- a/hive-agent/src/web_ui/mod.rs +++ b/hive-agent/src/web_ui/mod.rs @@ -343,21 +343,40 @@ fn find_claude_child() -> Option<u32> { /// (best-effort). See [`find_claude_child`] for how it's identified — /// **not** a name-based `pkill claude`, which would also hit an unrelated /// `claude` process someone is running interactively in the container. -async fn sigint_claude() -> SigintOutcome { +fn sigint_claude() -> SigintOutcome { let Some(pid) = find_claude_child() else { return SigintOutcome::NoProcess; }; - match tokio::process::Command::new("kill") - .args(["-INT", &pid.to_string()]) - .status() - .await - { - Ok(status) if status.success() => SigintOutcome::Signalled, - // Non-zero from `kill` means "no such process" (ESRCH) — it already - // exited between the check above and the signal. Same as never - // having found it. - Ok(_) => SigintOutcome::NoProcess, - Err(e) => SigintOutcome::Failed(e), + sigint_turn(pid) +} + +/// SIGINT `pid`, and its whole process group when it leads one, so the +/// commands a turn's claude started stop with it. `hive-claude` spawns the +/// turn's claude as the leader of its own group. A `claude` child that +/// leads none (`plugins.rs`'s `claude plugin` calls) shares the harness's +/// group, so it is signalled alone. +fn sigint_turn(pid: u32) -> SigintOutcome { + // A child is never pid 0 or 1, and `-0` / `-1` would reach the + // harness's own group or every process it may signal. + let Some(pid) = i32::try_from(pid).ok().filter(|&pid| pid > 1) else { + return SigintOutcome::NoProcess; + }; + // SAFETY: `getpgid(2)` only reads; a gone pid returns -1, never `pid`. + let target = if unsafe { libc::getpgid(pid) } == pid { + -pid + } else { + pid + }; + // SAFETY: plain `kill(2)` on the child or the group it leads. + if unsafe { libc::kill(target, libc::SIGINT) } == 0 { + return SigintOutcome::Signalled; + } + let err = std::io::Error::last_os_error(); + if err.raw_os_error() == Some(libc::ESRCH) { + // It exited between the scan and the signal: same as never found. + SigintOutcome::NoProcess + } else { + SigintOutcome::Failed(err) } } @@ -425,14 +444,18 @@ fn broker_error_response(err: &BrokerError, action: &str) -> Response { #[cfg(test)] mod tests { + use std::os::unix::process::CommandExt as _; + use std::path::Path; + use std::process::{Command, Stdio}; use std::sync::atomic::AtomicBool; - use std::sync::{Arc, Mutex, PoisonError}; + use std::sync::{Arc, Mutex}; + use std::time::Duration; use axum::body::Body; use axum::http::{Request, StatusCode, header}; use tower::ServiceExt; - use super::{AppState, routes}; + use super::{AppState, SigintOutcome, routes, sigint_turn}; use crate::events::Bus; use crate::login::LoginState; @@ -447,45 +470,127 @@ mod tests { /// either. #[test] fn neither_model_nor_effort_has_a_route() { - let dir = tempfile::tempdir().unwrap(); - let _env = crate::paths::HARNESS_DIR_ENV - .lock() - .unwrap_or_else(PoisonError::into_inner); - let prev = std::env::var_os("HYPERHIVE_HARNESS_DIR"); - // SAFETY: every test that sets this variable holds `HARNESS_DIR_ENV`. - unsafe { std::env::set_var("HYPERHIVE_HARNESS_DIR", dir.path()) }; - let state = AppState { - label: "test".to_owned(), - login: Arc::new(Mutex::new(LoginState::Online)), - session: Arc::new(Mutex::new(None)), - bus: Bus::new(), - socket: dir.path().join("mcp.sock"), - gui_vnc_port: None, - interrupted: Arc::new(AtomicBool::new(false)), - }; - let app = routes().with_state(state); - let (model, effort, compact) = tokio::runtime::Builder::new_current_thread() - .build() - .unwrap() - .block_on(async { - let model = app.clone().oneshot(form("/api/model", "model=opus")); - let effort = app.clone().oneshot(form("/api/effort", "effort=high")); - // A route the router does serve, so the 404s above are not - // the router refusing everything. - let compact = app.oneshot(form("/api/compact", "")); - ( - model.await.unwrap(), - effort.await.unwrap(), - compact.await.unwrap(), - ) - }); - match prev { - // SAFETY: as above. - Some(v) => unsafe { std::env::set_var("HYPERHIVE_HARNESS_DIR", v) }, - None => unsafe { std::env::remove_var("HYPERHIVE_HARNESS_DIR") }, - } + let (model, effort, compact) = crate::paths::with_harness_dir(|| { + let state = AppState { + label: "test".to_owned(), + login: Arc::new(Mutex::new(LoginState::Online)), + session: Arc::new(Mutex::new(None)), + bus: Bus::new(), + socket: crate::paths::harness_dir().join("mcp.sock"), + gui_vnc_port: None, + interrupted: Arc::new(AtomicBool::new(false)), + }; + let app = routes().with_state(state); + tokio::runtime::Builder::new_current_thread() + .build() + .unwrap() + .block_on(async { + let model = app.clone().oneshot(form("/api/model", "model=opus")); + let effort = app.clone().oneshot(form("/api/effort", "effort=high")); + // A route the router does serve, so the 404s above are not + // the router refusing everything. + let compact = app.oneshot(form("/api/compact", "")); + ( + model.await.unwrap(), + effort.await.unwrap(), + compact.await.unwrap(), + ) + }) + }); assert_eq!(model.status(), StatusCode::NOT_FOUND); assert_eq!(effort.status(), StatusCode::NOT_FOUND); assert_eq!(compact.status(), StatusCode::OK); } + + /// Poll `path` until it holds a complete, newline-terminated pid. + fn read_pid(path: &Path) -> i32 { + for _ in 0..250 { + if let Ok(text) = std::fs::read_to_string(path) + && text.ends_with('\n') + { + return text.trim().parse().expect("pid file holds a pid"); + } + std::thread::sleep(Duration::from_millis(20)); + } + unreachable!("no pid in {} within 5s", path.display()); + } + + /// Whether `pid` is still running. A zombie counts as gone: it has + /// exited, and `kill(pid, 0)` only keeps finding it until it is reaped. + fn is_running(pid: i32) -> bool { + // SAFETY: signal 0 delivers nothing; it only checks that `pid` exists. + if unsafe { libc::kill(pid, 0) } != 0 { + return false; + } + !std::fs::read_to_string(format!("/proc/{pid}/stat")).is_ok_and(|stat| { + stat.rsplit_once(") ") + .is_some_and(|(_, rest)| rest.starts_with('Z')) + }) + } + + #[test] + fn a_cancel_stops_the_commands_a_turn_started() { + let dir = tempfile::tempdir().expect("tempdir"); + let pidfile = dir.path().join("grandchild.pid"); + // A turn's claude as hive-claude spawns it: leading its own group, + // with a command of its own running in the foreground (a + // non-interactive shell starts `&` jobs with SIGINT ignored). The + // trailing `exit` keeps `sh -c` from exec'ing into that command. + let mut turn = Command::new("/bin/sh") + .args([ + "-c", + r#"sh -c 'echo $$ >"$1"; exec sleep 300' sh "$1"; exit"#, + "sh", + ]) + .arg(&pidfile) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .process_group(0) + .spawn() + .expect("spawn the stand-in turn"); + let grandchild = read_pid(&pidfile); + assert_ne!( + i32::try_from(turn.id()).ok(), + Some(grandchild), + "the command must be a child of the turn, not the turn itself" + ); + + assert!(matches!(sigint_turn(turn.id()), SigintOutcome::Signalled)); + + let mut alive = is_running(grandchild); + for _ in 0..250 { + if !alive { + break; + } + std::thread::sleep(Duration::from_millis(20)); + alive = is_running(grandchild); + } + 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); + } + } + let _ = turn.wait(); + assert!(!alive, "grandchild {grandchild} outlived the cancel"); + } + + #[test] + fn a_claude_sharing_the_harness_group_is_signalled_alone() { + // In this test process's own group, as `claude plugin` runs. A group + // signal here would SIGINT the test binary itself. + let mut child = Command::new("sleep") + .arg("300") + .spawn() + .expect("spawn sleep"); + + assert!(matches!(sigint_turn(child.id()), SigintOutcome::Signalled)); + + let status = child.wait().expect("reap sleep"); + assert_eq!( + std::os::unix::process::ExitStatusExt::signal(&status), + Some(libc::SIGINT) + ); + } } diff --git a/hive-agent/src/web_ui/proxy.rs b/hive-agent/src/web_ui/proxy.rs index 1bdb5e45..c44c207f 100644 --- a/hive-agent/src/web_ui/proxy.rs +++ b/hive-agent/src/web_ui/proxy.rs @@ -169,7 +169,7 @@ async fn proxy_handler( HeaderName::from_bytes(name.as_str().as_bytes()), HeaderValue::from_bytes(value.as_bytes()), ) { - resp_headers.insert(n, v); + resp_headers.append(n, v); } } match upstream_resp.bytes().await { @@ -269,8 +269,75 @@ async fn forward_over_unix_socket( if HOP_BY_HOP.contains(&name.as_str()) { continue; } - resp_headers.insert(name.clone(), value.clone()); + resp_headers.append(name.clone(), value.clone()); } let body_bytes = upstream_resp.into_body().collect().await?.to_bytes(); Ok((status, resp_headers, body_bytes).into_response()) } + +#[cfg(test)] +mod tests { + use super::*; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + /// Raw HTTP/1.1 answer with two `Set-Cookie` headers and a body. + const TWO_COOKIES: &[u8] = b"HTTP/1.1 200 OK\r\n\ + Set-Cookie: session=abc\r\n\ + Set-Cookie: csrf=def\r\n\ + Content-Length: 2\r\n\ + Connection: close\r\n\r\nok"; + + /// Read the request head, then answer with [`TWO_COOKIES`]. + async fn answer_once<S: AsyncReadExt + AsyncWriteExt + Unpin>(mut stream: S) { + let mut buf = [0u8; 1024]; + let _ = stream.read(&mut buf).await; + stream.write_all(TWO_COOKIES).await.unwrap(); + stream.shutdown().await.unwrap(); + } + + fn cookies(resp: &Response) -> Vec<&str> { + resp.headers() + .get_all("set-cookie") + .iter() + .map(|v| v.to_str().unwrap()) + .collect() + } + + #[tokio::test] + async fn unix_upstream_keeps_repeated_response_headers() { + let dir = tempfile::tempdir().unwrap(); + let sock = dir.path().join("up.sock"); + let listener = tokio::net::UnixListener::bind(&sock).unwrap(); + tokio::spawn(async move { + let (stream, _) = listener.accept().await.unwrap(); + answer_once(stream).await; + }); + + let resp = + forward_over_unix_socket(&sock, Method::GET, "/", HeaderMap::new(), Bytes::new()) + .await + .unwrap(); + assert_eq!(cookies(&resp), ["session=abc", "csrf=def"]); + } + + #[tokio::test] + async fn http_upstream_keeps_repeated_response_headers() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let base = format!("http://{}", listener.local_addr().unwrap()); + tokio::spawn(async move { + let (stream, _) = listener.accept().await.unwrap(); + answer_once(stream).await; + }); + + let client = reqwest::Client::builder().no_proxy().build().unwrap(); + let resp = proxy_handler( + State((Arc::new(client), Arc::new(base))), + Method::GET, + Uri::from_static("/"), + HeaderMap::new(), + Bytes::new(), + ) + .await; + assert_eq!(cookies(&resp), ["session=abc", "csrf=def"]); + } +} diff --git a/hive-c0re/src/coordinator.rs b/hive-c0re/src/coordinator.rs index d06f5969..9e70ad3d 100644 --- a/hive-c0re/src/coordinator.rs +++ b/hive-c0re/src/coordinator.rs @@ -462,6 +462,15 @@ impl Coordinator { let build_logs = Arc::new( crate::build_logs::BuildLogs::open(build_logs_dir).context("open build_logs")?, ); + // This process is the database's only writer and has started no + // build yet, so a row still in-flight belongs to a dead process. + let reaped = build_logs.reap_orphans()?; + if reaped > 0 { + tracing::warn!( + reaped, + "build_logs: closed rows left in-flight by a previous run" + ); + } // Install the process-wide handle so `lifecycle::run` / // `lifecycle::prebuild_toplevel` can write without us having // to thread an `Arc<BuildLogs>` through every public entry @@ -1376,3 +1385,35 @@ mod agent_paths_tests { assert_eq!(paths.agent, std::path::PathBuf::from("/run/x")); } } + +#[cfg(test)] +mod open_tests { + use super::{Coordinator, HiveEnv}; + + fn open(db: &std::path::Path) -> Coordinator { + Coordinator::open( + db, + HiveEnv::default(), + crate::hive_stats::PriceTable::new(), + 1, + ) + .expect("open coordinator") + } + + /// A build log row the previous process left open is closed as failed + /// when the next one opens the coordinator. + #[tokio::test] + async fn open_closes_build_logs_left_in_flight() { + let dir = tempfile::tempdir().expect("tempdir"); + let db = dir.path().join("broker.sqlite"); + let id = open(&db) + .build_logs + .start("alice", "run", "cmd", None) + .expect("start"); + + let coord = open(&db); + let row = coord.build_logs.get_full(id).expect("get").expect("Some"); + assert_eq!(row.header.status.as_deref(), Some("fail")); + assert!(row.header.finished_at.is_some()); + } +} diff --git a/hive-c0re/src/snapshot_push.rs b/hive-c0re/src/snapshot_push.rs index 1cd65905..3098c6d3 100644 --- a/hive-c0re/src/snapshot_push.rs +++ b/hive-c0re/src/snapshot_push.rs @@ -18,6 +18,8 @@ //! stores would split that chain in two, which is the case the store //! exists to serve. +use std::os::fd::OwnedFd; + use anyhow::{Context as _, Result, bail}; use hive_priv_sock::PrivRequest; use tokio::io::AsyncWriteExt as _; @@ -61,10 +63,7 @@ pub async fn push_agent_snapshot(agent: &str, snapshot: &str, parent: Option<&st // closes our copy as soon as the kernel has it, so the receiver sees // EOF when `btrfs send` finishes rather than hanging on a descriptor // this process is still holding. - let fd = sock - .into_std() - .context("detach snapshot-store socket for hand-off")? - .into(); + let fd = detach_blocking(sock)?; let resp = priv_client::call_with_fd( &PrivRequest::SendAgentSnapshotToFd { @@ -93,6 +92,22 @@ pub async fn push_agent_snapshot(agent: &str, snapshot: &str, parent: Option<&st Ok(()) } +/// Turn a tokio socket into a descriptor safe to hand to `btrfs send`. +/// +/// tokio sockets are `O_NONBLOCK`, and `into_std` does not clear it. The +/// flag lives on the open file description, which `hive-priv` shares with +/// us through `SCM_RIGHTS`, so `btrfs send` would write to a +/// non-blocking socket and get `EAGAIN` whenever the send buffer is full. +fn detach_blocking(sock: TcpStream) -> Result<OwnedFd> { + let std_sock = sock + .into_std() + .context("detach snapshot-store socket for hand-off")?; + std_sock + .set_nonblocking(false) + .context("switch snapshot-store socket to blocking mode for hand-off")?; + Ok(std_sock.into()) +} + /// Require a configured store address. /// /// Split out of [`push_agent_snapshot`] so the "not configured" path is @@ -118,7 +133,30 @@ fn require_store_addr(configured: Option<&str>) -> Result<&str> { #[cfg(test)] mod tests { - use super::require_store_addr; + use std::os::fd::AsRawFd as _; + + use tokio::net::{TcpListener, TcpStream}; + + use super::{detach_blocking, require_store_addr}; + + #[tokio::test] + async fn the_handed_off_descriptor_is_blocking() { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let sock = TcpStream::connect(listener.local_addr().expect("addr")) + .await + .expect("connect"); + + let fd = detach_blocking(sock).expect("detach"); + + // SAFETY: `fd` is a live descriptor owned by this test. + let flags = unsafe { libc::fcntl(fd.as_raw_fd(), libc::F_GETFL) }; + assert!(flags >= 0, "F_GETFL failed"); + assert_eq!( + flags & libc::O_NONBLOCK, + 0, + "btrfs send would get EAGAIN on a full socket buffer" + ); + } #[test] fn a_configured_address_passes_through() { diff --git a/hive-c0re/src/stores/build_logs.rs b/hive-c0re/src/stores/build_logs.rs index b7ca43b9..6af05a71 100644 --- a/hive-c0re/src/stores/build_logs.rs +++ b/hive-c0re/src/stores/build_logs.rs @@ -48,6 +48,9 @@ const KEEP_FAIL_SECS: i64 = 30 * 24 * 3600; /// `finished_at IS NULL`). const KEEP_OK_SECS: i64 = 24 * 3600; +/// Last line of a row `reap_orphans` closed. +const ORPHAN_NOTE: &str = "hive-c0re stopped before this build finished"; + const SCHEMA: &str = " CREATE TABLE IF NOT EXISTS build_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, @@ -175,6 +178,30 @@ impl BuildLogs { }) } + /// Close every row still marked in-flight as failed. Returns the number + /// closed. + /// + /// Only sound when no other writer holds this database: the caller must + /// be the one process that owns it, before it starts any build. A row + /// with `finished_at IS NULL` then belongs to a process that died, and + /// nothing will ever finish it. Kept out of `open` so a second opener + /// can't close another process's live rows. + pub fn reap_orphans(&self) -> Result<u64> { + let now = Utc::now().timestamp(); + let conn = self.conn.lock().unwrap(); + let closed = conn + .execute( + "UPDATE build_logs + SET finished_at = ?1, + status = 'fail', + stderr = stderr || ?2 || x'0a' + WHERE finished_at IS NULL", + params![now, ORPHAN_NOTE], + ) + .context("close orphaned build_logs rows")?; + Ok(u64::try_from(closed).unwrap_or(0)) + } + /// Subscribe to per-build append/finish notifications. Each emitted /// value is the `id` of the row that changed. The SSE stream handler /// calls this once and filters for its target id. @@ -522,6 +549,31 @@ mod tests { assert!(full.stderr.ends_with('\n')); } + #[test] + fn reap_orphans_fails_rows_left_in_flight_by_a_previous_process() { + let dir = tempfile::tempdir().expect("tempdir"); + let (orphan, done) = { + let db = BuildLogs::open(dir.path()).expect("open"); + let orphan = db.start("alice", "run", "cmd one", None).expect("start"); + db.append_stdout(orphan, "building"); + let done = db.start("bob", "run", "cmd two", None).expect("start"); + db.finish(done, BuildStatus::Ok); + (orphan, done) + }; + + let db = BuildLogs::open(dir.path()).expect("reopen"); + assert_eq!(db.reap_orphans().expect("reap"), 1); + + let row = db.get_full(orphan).expect("get").expect("Some"); + assert_eq!(row.header.status.as_deref(), Some("fail")); + assert!(row.header.finished_at.is_some()); + assert!(row.stdout.contains("building")); + assert!(row.stderr.ends_with(&format!("{ORPHAN_NOTE}\n"))); + let ok = db.get_full(done).expect("get").expect("Some"); + assert_eq!(ok.header.status.as_deref(), Some("ok")); + assert_eq!(db.reap_orphans().expect("reap again"), 0); + } + #[test] fn list_recent_orders_newest_first_and_clamps() { let (_d, db) = tmpdb(); diff --git a/hive-runtime/README.md b/hive-runtime/README.md index 11183df4..6b6d4025 100644 --- a/hive-runtime/README.md +++ b/hive-runtime/README.md @@ -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 idle watchdog once no `session/update` has arrived for `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 -with `AcpError::IdleTimeout` (`IdleKilled` if the agent was killed). This is +is sent SIGINT, and the next turn respawns it. A watchdog stop fails the turn +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 an HTTP 429, ends the turn. -The agent leads its own process group. Killing it, and dropping the runtime, -`SIGKILL`s that group, so the commands the agent started die with it. +The agent leads its own process group. Stopping it, and dropping the runtime, +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 diff --git a/hive-runtime/src/acp/mod.rs b/hive-runtime/src/acp/mod.rs index b93be539..64afa350 100644 --- a/hive-runtime/src/acp/mod.rs +++ b/hive-runtime/src/acp/mod.rs @@ -39,7 +39,7 @@ const SETTLE_QUIET: Duration = Duration::from_millis(500); const SETTLE_MAX: Duration = Duration::from_secs(5); /// 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); /// How long a `compact` command may send nothing before it is cancelled and @@ -120,10 +120,12 @@ pub enum AcpError { EmptyEndTurn, #[error( "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 }, - #[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 }, } @@ -141,8 +143,8 @@ struct CancelState { impl Canceller { /// 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 - /// agent that has not ended it within `CANCEL_GRACE` is killed, and the - /// turn fails with [`AcpError::CancelIgnored`]. Returns whether a turn + /// agent that has not ended it within `CANCEL_GRACE` is sent SIGINT, and + /// the turn fails with [`AcpError::CancelIgnored`]. Returns whether a turn /// was in flight; with none, this does nothing. #[must_use] pub fn cancel(&self) -> bool { @@ -1062,9 +1064,14 @@ mod tests { const AGENT: &str = r#" n=0 prompt= model=m/think effort=low 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. +# `&` starts it with SIGINT ignored; `env` puts the default back. 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 update() { printf '{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"%s","update":%s}}\n' "$1" "$2" @@ -1529,6 +1536,33 @@ done 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) { (session.to_owned(), text.to_owned()) } diff --git a/hive-runtime/src/acp/rpc.rs b/hive-runtime/src/acp/rpc.rs index 65681f02..6d07d054 100644 --- a/hive-runtime/src/acp/rpc.rs +++ b/hive-runtime/src/acp/rpc.rs @@ -31,9 +31,10 @@ type Reply = Result<Value, AcpError>; type Pending = Arc<Mutex<HashMap<u64, (&'static str, oneshot::Sender<Reply>)>>>; /// 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 -/// with it. A descendant that left the group (`setsid`, `setpgid`) is not -/// reached. +/// connection sends that group SIGINT, so the agent can stop the commands +/// it started, and they get the signal too. Nothing escalates: an agent +/// that ignores SIGINT keeps running. A descendant that left the group +/// (`setsid`, `setpgid`) is not reached. pub(super) struct Connection { child: Child, /// The agent's pid, which is its group's id. Read at spawn: a reaped @@ -66,7 +67,6 @@ impl Connection { .stdout(Stdio::piped()) .stderr(Stdio::piped()) .process_group(0) - .kill_on_drop(true) .spawn() .map_err(|source| AcpError::Spawn { program: command.command.clone(), @@ -182,7 +182,7 @@ impl Drop for Connection { // SAFETY: plain `kill(2)`; `-pgid` targets the agent's own // process group, which `process_group(0)` made it lead. unsafe { - libc::kill(-pgid, libc::SIGKILL); + libc::kill(-pgid, libc::SIGINT); } } } diff --git a/hive-subagent-mcp/src/mcp.rs b/hive-subagent-mcp/src/mcp.rs index 8035dba1..6d24e0c5 100644 --- a/hive-subagent-mcp/src/mcp.rs +++ b/hive-subagent-mcp/src/mcp.rs @@ -187,10 +187,6 @@ struct StatusArgs { struct InterruptArgs { /// The running session's name to signal. 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)] @@ -276,11 +272,13 @@ impl SubagentMcp { 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 \ 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 \ - SIGKILL, otherwise SIGINT." + retry shortly), same as a name with nothing tracked at all. On claude it sends \ + 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 { - match session::interrupt(&self.state, &args.name, args.force) { + match session::interrupt(&self.state, &args.name) { Ok(msg) => msg, Err(e) => format!("interrupt error: {e:#}"), } diff --git a/hive-subagent-mcp/src/session.rs b/hive-subagent-mcp/src/session.rs index 75031cd1..4a276e82 100644 --- a/hive-subagent-mcp/src/session.rs +++ b/hive-subagent-mcp/src/session.rs @@ -5,8 +5,8 @@ //! continuation loop records. //! //! **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 -//! rest of the maps and for what the `None`/`Some` split is for) — all of +//! in-memory `name -> Slot` map (see `State`'s own doc for the rest of the +//! 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 //! whatever was running gets killed //! with it (`tokio`'s own child-process drop semantics), not adopted, and @@ -230,11 +230,33 @@ enum SubagentRuntime { enum Stopper { /// Signals the turn's `claude` process. Claude(Cancel), - /// Sends the ACP agent `session/cancel`; one that ignores it is killed - /// by the runtime after its own grace period. + /// 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), + /// `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` /// blocks on before it answers. 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 /// task driving a run outlives the tool call that started it. /// -/// The map value is `Option<Stopper>`: `None` means `name` is reserved for -/// an in-flight `start`/`continue` that hasn't reached a confirmed -/// spawn yet; `Some(stopper)` means a real turn is tracked and -/// interruptible. The `None` state exists to close a real TOCTOU window a -/// reviewer caught in the original check-then-insert version: checking "is -/// `name` free" and committing to it are two different lock acquisitions -/// unless the check *is* the reservation — see `reserve`. +/// The map value is a `Slot`: `Starting` means `name` is reserved for an +/// in-flight `start`/`continue` that hasn't reached a confirmed spawn yet; +/// `Running` means a real turn is tracked and interruptible; `Stopping` +/// means it was interrupted and its process is still winding down. Each +/// claims the name. `Starting` closes a TOCTOU window: checking "is `name` +/// free" and committing to it are two different lock acquisitions unless +/// the check *is* the reservation — see `reserve`. /// /// `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 @@ -359,7 +381,7 @@ fn searched_location(config: &Config) -> Option<String> { /// name across directories at your own risk, the tool doesn't disambiguate /// it (flagged in review when `dir` was added). pub struct State { - running: Mutex<HashMap<String, Option<Stopper>>>, + running: Mutex<HashMap<String, Slot>>, dirs: Mutex<HashMap<String, String>>, killed: Mutex<HashMap<String, i32>>, last_event: Mutex<HashMap<String, Instant>>, @@ -547,17 +569,17 @@ impl State { } } - /// `Some(true)` — a live process is tracked, interruptible. `Some(false)` - /// — the name is claimed but no process is confirmed under it: an - /// 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> { + /// Which [`Slot`] `name` is in; `None` with nothing tracked under it. + fn occupancy(&self, name: &str) -> Option<Occupancy> { self.running .lock() .unwrap_or_else(PoisonError::into_inner) .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 @@ -572,21 +594,21 @@ impl State { if running.contains_key(name) { return false; } - running.insert(name.to_owned(), None); + running.insert(name.to_owned(), Slot::Starting); true } /// Release a reservation that never made it to a real spawn (an error /// on the slow path between `reserve` and `Claude::spawn` succeeding). - /// A no-op if the entry was already upgraded to `Some` — this only ever - /// clears a still-`None` placeholder, never a live process. + /// A no-op if the entry was already upgraded to `Running` — this only + /// ever clears a still-`Starting` placeholder, never a live process. /// /// 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 /// one way a route could resolve to a session that isn't there. fn release_reservation(&self, name: &str) { 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 { running.remove(name); } @@ -742,14 +764,14 @@ impl State { } } - /// Upgrade `name`'s `None` 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. + /// 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) { self.running .lock() .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 @@ -763,7 +785,7 @@ impl State { self.running .lock() .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 @@ -1286,7 +1308,7 @@ pub async fn start(state: &Arc<State>, req: StartRequest) -> anyhow::Result<Stri // process spawned. let role = req.role.as_deref().map(crate::role::load).transpose()?; 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 // claimed `name` — see `resolve_dir`'s doc for why the order matters, @@ -1460,9 +1482,11 @@ pub async fn continue_( validate_name(name)?; state.allow_model(model.as_deref()).await?; if !state.reserve(name) { - anyhow::bail!( - "subagent `{name}` is already running — use `interrupt` first if you meant to redirect it" - ); + return Err(name_taken( + 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 // 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. struct StatusFacts<'a> { name: &'a str, - occupancy: Option<bool>, + occupancy: Option<Occupancy>, killed: Option<i32>, session_exists: bool, last_event_age: Option<Duration>, @@ -2224,7 +2248,7 @@ fn describe_status(facts: &StatusFacts<'_>) -> anyhow::Result<String> { let name = facts.name; let turns = describe_turns(facts.turns); match facts.occupancy { - Some(true) => { + Some(Occupancy::Running) => { return Ok(format!( "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 \ @@ -2233,7 +2257,7 @@ fn describe_status(facts: &StatusFacts<'_>) -> anyhow::Result<String> { describe_pending_signal(facts.stop.as_ref()), )); } - Some(false) => { + Some(Occupancy::Starting) => { return Ok(format!( "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 \ @@ -2241,6 +2265,15 @@ fn describe_status(facts: &StatusFacts<'_>) -> anyhow::Result<String> { 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 => {} } 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 -/// `hive_claude::Cancel::cancel`). Refuses a name with nothing running: no -/// entry at all, or one still in the brief not-yet-spawned window (nothing -/// to signal yet — the reservation is put back so a concurrent -/// `start`/`continue` for the same name still gets refused). +/// Stop `name`'s running turn: SIGINT to claude's process group (see +/// `hive_claude::Cancel::cancel`), or `session/cancel` on ACP. Nothing +/// escalates to SIGKILL: each command claude's Bash tool starts leads its +/// own session, and only claude shutting down on SIGINT stops it. /// -/// This stops a goal run, not just the turn in it — and it does so by -/// **recording `StopReason::Cancelled`**, not by relying on how the child -/// dies. +/// Returns without waiting. A claude turn's name stays claimed as +/// `Slot::Stopping` until the turn's own `finish_turn` clears it, so no +/// 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 -/// default is SIGINT, claude handles SIGINT by emitting its terminal -/// `result` event and exiting 0, and a zero exit is `TurnEnd::Complete` — -/// so `spawn_and_track`'s killed-turn early return never fires and the loop -/// goes straight on to spawn turn three of five. The recorded reason is what -/// `plan_after_turn` reads, and it is checked before the goal, so the run -/// stops whichever way the child ends up exiting. `force`'s SIGKILL does -/// 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. +/// It stops the goal run, not just the turn, by **recording +/// `StopReason::Cancelled`**: claude answers SIGINT by emitting its +/// terminal `result` event and exiting 0, which is `TurnEnd::Complete`, so +/// only the recorded reason, which `plan_after_turn` checks before the +/// goal, keeps the loop from re-prompting. It is written while the +/// `running` lock is held, so a turn that ends on its own in between still +/// finds it. /// /// # Errors /// -/// An invalid name, nothing tracked under `name`, or nothing to signal right -/// now (a spawn or ACP turn not yet live, or a goal run between turns). -pub fn interrupt(state: &State, name: &str, force: bool) -> anyhow::Result<String> { +/// 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; the +/// 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)?; let mut running = state.running.lock().unwrap_or_else(PoisonError::into_inner); + let mut claimed = ""; match running.remove(name) { None => anyhow::bail!("no subagent named `{name}` is currently running"), - Some(None) => { - running.insert(name.to_owned(), None); + Some(Slot::Starting) => { + running.insert(name.to_owned(), Slot::Starting); return Err(still_starting(name)); } - Some(Some(Stopper::Claude(cancel))) => { - state.record_stop(name, StopReason::Cancelled); - drop(running); - cancel.cancel(force); + Some(Slot::Stopping) => { + running.insert(name.to_owned(), Slot::Stopping); + anyhow::bail!( + "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 - // agent that ignores `session/cancel`. `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(Some(Stopper::Acp(canceller))) => { + 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 !canceller.cancel() { @@ -2386,7 +2422,7 @@ pub fn interrupt(state: &State, name: &str, force: bool) -> anyhow::Result<Strin None => stops.remove(name), }; 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)); } } @@ -2394,10 +2430,24 @@ pub fn interrupt(state: &State, name: &str, force: bool) -> anyhow::Result<Strin Ok(format!( "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 \ - 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 /// signal yet. fn still_starting(name: &str) -> anyhow::Error { @@ -2539,7 +2589,7 @@ mod tests { state.reserve("n"); assert_eq!( state.occupancy("n"), - Some(false), + Some(Occupancy::Starting), "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, // which is all a caller ever gets back. let running = describe_status(&StatusFacts { - occupancy: Some(true), + occupancy: Some(Occupancy::Running), ..StatusFacts::new("n") }) .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}" ); let starting = describe_status(&StatusFacts { - occupancy: Some(false), + occupancy: Some(Occupancy::Starting), ..StatusFacts::new("n") }) .expect("starting status"); @@ -3238,7 +3288,7 @@ mod tests { #[test] fn a_running_status_reports_the_age_and_an_idle_one_does_not() { let running = describe_status(&StatusFacts { - occupancy: Some(true), + occupancy: Some(Occupancy::Running), last_event_age: Some(Duration::from_secs(4)), ..StatusFacts::new("n") }) @@ -3249,7 +3299,7 @@ mod tests { {running}" ); let wedged = describe_status(&StatusFacts { - occupancy: Some(true), + occupancy: Some(Occupancy::Running), last_event_age: Some(Duration::from_mins(15)), ..StatusFacts::new("n") }) @@ -3281,7 +3331,7 @@ mod tests { // turn can finish between the two. The answer drops the age rather // than inventing one. let raced = describe_status(&StatusFacts { - occupancy: Some(true), + occupancy: Some(Occupancy::Running), ..StatusFacts::new("n") }) .expect("running status"); @@ -3710,9 +3760,7 @@ mod tests { } assert_eq!(turns_spawned(&log), 1, "the first turn never started"); - // An operator's plain `interrupt`, not `force` — SIGINT is the - // default and the case that was broken. - interrupt(&state, "n", false).expect("a running turn is interruptible"); + interrupt(&state, "n").expect("a running turn is interruptible"); // Generous: the fake logs its invocation as its first act, so a // re-prompt shows up here almost immediately once the loop decides @@ -3731,6 +3779,178 @@ mod tests { std::fs::remove_dir_all(&dir).ok(); } + /// A stand-in for claude with a command of its own still running: an + /// inner `sh` writes its pid to `pidfile` and execs `sleep`. It runs in + /// the foreground because a non-interactive shell starts `&` jobs with + /// SIGINT ignored, and its output goes to /dev/null so it cannot hold the + /// driver's pipes open. + fn busy_claude(dir: &Path, pidfile: &Path) -> PathBuf { + let script = dir.join("busy-claude"); + write_executable( + &script, + &format!( + "#!/bin/sh\nsh -c 'echo $$ >\"$1\"; exec sleep 300' sh '{}' >/dev/null 2>&1\n", + pidfile.display() + ), + ); + script + } + + /// Poll `path` until it holds a complete, newline-terminated pid. + async fn read_pid(path: &Path) -> i32 { + for _ in 0..250 { + if let Ok(text) = std::fs::read_to_string(path) + && text.ends_with('\n') + { + return text.trim().parse().expect("pid file holds a pid"); + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + unreachable!("no pid in {} within 5s", path.display()); + } + + /// Whether `pid` is still running. A zombie counts as gone: it has + /// exited, and `kill(pid, 0)` only keeps finding it until it is reaped. + fn is_running(pid: i32) -> bool { + // SAFETY: signal 0 delivers nothing; it only checks that `pid` exists. + if unsafe { libc::kill(pid, 0) } != 0 { + return false; + } + !std::fs::read_to_string(format!("/proc/{pid}/stat")).is_ok_and(|stat| { + stat.rsplit_once(") ") + .is_some_and(|(_, rest)| rest.starts_with('Z')) + }) + } + + #[tokio::test] + async fn an_interrupt_stops_the_commands_the_subagent_started() { + let dir = scratch_dir("group"); + let pidfile = dir.join("grandchild.pid"); + let state = Arc::new(mid_run("n", "carry on indefinitely", 5)); + let config = Config { + program: Some(busy_claude(&dir, &pidfile)), + ..Config::default() + }; + spawn_and_track( + &state, + "n", + &config, + &Attach::Create("n".to_owned()), + "turn one".to_owned(), + None, + ) + .expect("the fake claude spawns"); + let grandchild = read_pid(&pidfile).await; + + interrupt(&state, "n").expect("a running turn is interruptible"); + + let mut alive = is_running(grandchild); + for _ in 0..250 { + if !alive { + break; + } + tokio::time::sleep(Duration::from_millis(20)).await; + alive = is_running(grandchild); + } + 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] fn a_cancelled_run_stops_with_turns_still_on_its_clock() { // The loop half of the same fact, asserted directly: `Cancelled` is @@ -3985,7 +4205,12 @@ mod tests { #[test] 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 { occupancy, session_exists: true, @@ -4000,7 +4225,7 @@ mod tests { ); } let goalless = describe_status(&StatusFacts { - occupancy: Some(true), + occupancy: Some(Occupancy::Running), ..StatusFacts::new("n") }) .expect("status"); @@ -4016,7 +4241,7 @@ mod tests { // it is blocked, its turn hasn't ended yet, and plain "running" would // tell the parent to leave it alone. let answer = describe_status(&StatusFacts { - occupancy: Some(true), + occupancy: Some(Occupancy::Running), stop: Some(StopReason::NeedHelp("no credential".to_owned())), ..StatusFacts::new("n") }) @@ -4026,7 +4251,7 @@ mod tests { "both facts are true at once and both have to be in the answer: {answer}" ); let quiet = describe_status(&StatusFacts { - occupancy: Some(true), + occupancy: Some(Occupancy::Running), stop: Some(StopReason::Done), ..StatusFacts::new("n") }) @@ -4539,10 +4764,8 @@ done turns_spawned(&dir.join("claude-turns")) == 1 }) .await; - interrupt(&state, "n", false).expect("a running claude turn is interruptible"); - // `interrupt` drops the tracking entry itself; the run is over once - // `finish_turn` has cleared its liveness clock too. - until("the run ends", || state.last_event_age("n").is_none()).await; + interrupt(&state, "n").expect("a running claude turn is interruptible"); + until("the run ends", || state.occupancy("n").is_none()).await; // This `start` had no goal, so the fake claude's clean exit after // SIGINT (`TurnEnd::Complete`) is exactly the case that used to @@ -4889,7 +5112,7 @@ echo '{"type":"result","subtype":"success","is_error":false}' }) .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; assert_eq!(lines(&dir.join("prompts")).len(), 1, "no second goal turn"); @@ -4912,7 +5135,7 @@ echo '{"type":"result","subtype":"success","is_error":false}' // The test runtime is single-threaded, so the run's task has not // 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_eq!(state.stop_reason("n"), None);