Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
f3860f3497 | ||
|
|
43e69d345e | ||
|
|
028c283ff9 |
15 changed files with 548 additions and 154 deletions
4
Cargo.lock
generated
4
Cargo.lock
generated
|
|
@ -1835,9 +1835,9 @@ dependencies = [
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "hive-claude"
|
name = "hive-claude"
|
||||||
version = "0.1.2"
|
version = "0.1.4"
|
||||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
checksum = "1e43e06fd09a7b6d6cca3e70f06d00fa795ceddfe1555121ca0cf4fc578eb9c3"
|
checksum = "4d7c5638f8f59e438d6416b3048709c9b3d6a10c23872ecf3825e3fe4c1bd7e8"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"libc",
|
"libc",
|
||||||
"serde",
|
"serde",
|
||||||
|
|
|
||||||
|
|
@ -97,7 +97,7 @@ hive-jobq-metrics = { path = "hive-jobq-metrics" }
|
||||||
hive-jobq-wire = { path = "hive-jobq-wire" }
|
hive-jobq-wire = { path = "hive-jobq-wire" }
|
||||||
hive-log = { path = "hive-log" }
|
hive-log = { path = "hive-log" }
|
||||||
hive-core-agent-sock = { path = "hive-core-agent-sock" }
|
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-host-sock = { path = "hive-host-sock" }
|
||||||
hive-priv-sock = { path = "hive-priv-sock" }
|
hive-priv-sock = { path = "hive-priv-sock" }
|
||||||
hive-runtime = { path = "hive-runtime" }
|
hive-runtime = { path = "hive-runtime" }
|
||||||
|
|
|
||||||
|
|
@ -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
|
||||||
|
|
|
||||||
|
|
@ -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
|
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 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`,
|
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
|
||||||
stop reason to that file as well, so the artifact you were going to read
|
stop reason to that file as well, so the artifact you were going to read
|
||||||
|
|
@ -340,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.
|
||||||
|
|
@ -349,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`).
|
||||||
|
|
|
||||||
|
|
@ -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(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(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(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(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 killed it if it ignored that; `note` says which |
|
| `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] \`<qualified-label>\` claude turn failed:\n<err>` to `operator` via `send_to_operator` |
|
| `Err(Failed(err))` | route `[system] \`<qualified-label>\` claude turn failed:\n<err>` to `operator` via `send_to_operator` |
|
||||||
|
|
||||||
`ApiStall` catches an Anthropic API stall — a multi-retry connection storm where
|
`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 —
|
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
|
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
|
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
|
`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
|
also how a provider HTTP 429 ends an ACP turn when the agent retries it without
|
||||||
|
|
|
||||||
|
|
@ -227,11 +227,11 @@ Slash commands today:
|
||||||
|
|
||||||
- `/help` — list commands locally.
|
- `/help` — list commands locally.
|
||||||
- `/clear` — wipe the local terminal view (server history kept).
|
- `/clear` — wipe the local terminal view (server history kept).
|
||||||
- `/cancel` — `POST /api/cancel` → host shellouts `pkill -INT
|
- `/cancel` — `POST /api/cancel` → SIGINT to the turn's claude and every
|
||||||
claude`, emits a Note. On an ACP agent it sends the agent
|
command it started (its process group), emits a Note. On an ACP agent
|
||||||
`session/cancel` instead, and kills it if the turn hasn't ended 10s
|
it sends the agent `session/cancel` instead, and SIGINT if the turn
|
||||||
later. Also surfaces as a `■ cancel turn` button in the state row
|
hasn't ended 10s later. Also surfaces as a `■ cancel turn` button in
|
||||||
while state=thinking.
|
the state row while state=thinking.
|
||||||
- `/compact` — `POST /api/compact` → sets the deferred
|
- `/compact` — `POST /api/compact` → sets the deferred
|
||||||
`Bus::request_compact()` flag and returns immediately. The harness
|
`Bus::request_compact()` flag and returns immediately. The harness
|
||||||
runs the `/compact` at the next turn boundary (end of the in-flight
|
runs the `/compact` at the next turn boundary (end of the in-flight
|
||||||
|
|
@ -272,8 +272,8 @@ shaped).
|
||||||
- `POST /login/{start,code,cancel}` — claude OAuth login flow.
|
- `POST /login/{start,code,cancel}` — claude OAuth login flow.
|
||||||
Start/cancel emit `LiveEvent::StatusChanged` to flip the
|
Start/cancel emit `LiveEvent::StatusChanged` to flip the
|
||||||
badge to/from `needs_login_in_progress`.
|
badge to/from `needs_login_in_progress`.
|
||||||
- `POST /api/cancel` — SIGINT the in-flight claude turn. Emits a
|
- `POST /api/cancel` — SIGINT the in-flight claude turn and the
|
||||||
`LiveEvent::Note`.
|
commands it started. Emits a `LiveEvent::Note`.
|
||||||
- `POST /api/compact` — run `/compact` on the persistent session
|
- `POST /api/compact` — run `/compact` on the persistent session
|
||||||
(same MCP config + system prompt + allowed tools as a normal
|
(same MCP config + system prompt + allowed tools as a normal
|
||||||
turn — only the stdin payload differs). Flips state to
|
turn — only the stdin payload differs). Flips state to
|
||||||
|
|
|
||||||
|
|
@ -50,8 +50,8 @@ const DEFAULT_SESSION_TITLE: &str = "hive-session";
|
||||||
/// capacity limits.
|
/// capacity limits.
|
||||||
const DEFAULT_RATE_LIMIT_SLEEP_SECS: u64 = 300;
|
const DEFAULT_RATE_LIMIT_SLEEP_SECS: u64 = 300;
|
||||||
|
|
||||||
/// Idle watchdog window: kill claude and surface `ApiStall` if it produces no
|
/// Idle watchdog window: send claude SIGINT and surface `ApiStall` once it
|
||||||
/// stdout for this long. The timer resets on every stdout line, so a large but
|
/// 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.
|
/// still-streaming turn is never cut — only a complete silence trips it.
|
||||||
/// Default is 10 minutes: long enough for legitimate big-context turns, short
|
/// Default is 10 minutes: long enough for legitimate big-context turns, short
|
||||||
/// enough to cut a multi-retry Anthropic connection storm (atlas telemetry:
|
/// enough to cut a multi-retry Anthropic connection storm (atlas telemetry:
|
||||||
|
|
@ -189,14 +189,14 @@ pub enum TurnError {
|
||||||
/// the wake message, the serve loop requeues it so the next turn retries;
|
/// the wake message, the serve loop requeues it so the next turn retries;
|
||||||
/// no status park.
|
/// no status park.
|
||||||
SessionNotFound,
|
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
|
/// output silence — indicative of an Anthropic API stall (e.g. a multi-retry
|
||||||
/// connection storm burning minutes of wall-clock with no stream progress).
|
/// connection storm burning minutes of wall-clock with no stream progress).
|
||||||
/// The serve loop parks for `stall_sleep_secs()` and requeues, like the
|
/// The serve loop parks for `stall_sleep_secs()` and requeues, like the
|
||||||
/// rate-limit path — NOT a crash.
|
/// rate-limit path — NOT a crash.
|
||||||
ApiStall,
|
ApiStall,
|
||||||
/// The ACP backend's idle watchdog stopped a turn that sent nothing for
|
/// 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
|
/// it ignored that. Carries what happened. An ACP agent retrying a failing
|
||||||
/// provider (e.g. HTTP 429) without reporting it ends up here. Handled
|
/// provider (e.g. HTTP 429) without reporting it ends up here. Handled
|
||||||
/// like [`Self::ApiStall`].
|
/// like [`Self::ApiStall`].
|
||||||
|
|
@ -558,11 +558,11 @@ pub fn emit_turn_end(bus: &Bus, outcome: &TurnOutcome) {
|
||||||
bus.emit(LiveEvent::TurnEnd {
|
bus.emit(LiveEvent::TurnEnd {
|
||||||
ok: false,
|
ok: false,
|
||||||
note: Some(format!(
|
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()
|
turn_idle_secs()
|
||||||
)),
|
)),
|
||||||
});
|
});
|
||||||
tracing::warn!("turn killed: API stall (idle watchdog)");
|
tracing::warn!("turn stopped: API stall (idle watchdog)");
|
||||||
}
|
}
|
||||||
Err(TurnError::AgentStall(note)) => {
|
Err(TurnError::AgentStall(note)) => {
|
||||||
bus.emit(LiveEvent::TurnEnd {
|
bus.emit(LiveEvent::TurnEnd {
|
||||||
|
|
|
||||||
|
|
@ -69,7 +69,7 @@ pub(super) async fn post_cancel_turn(State(state): State<AppState>) -> Response
|
||||||
.emit(crate::events::LiveEvent::Note { text: note.into() });
|
.emit(crate::events::LiveEvent::Note { text: note.into() });
|
||||||
return (axum::http::StatusCode::OK, "ok").into_response();
|
return (axum::http::StatusCode::OK, "ok").into_response();
|
||||||
}
|
}
|
||||||
let out = super::sigint_claude().await;
|
let out = super::sigint_claude();
|
||||||
let note = match out {
|
let note = match out {
|
||||||
SigintOutcome::Signalled => {
|
SigintOutcome::Signalled => {
|
||||||
// A process actually got signalled — the *next* turn's wake
|
// A process actually got signalled — the *next* turn's wake
|
||||||
|
|
|
||||||
|
|
@ -84,7 +84,7 @@ pub(super) async fn post_login_cancel(State(state): State<AppState>) -> Response
|
||||||
/// preservation invariants.
|
/// preservation invariants.
|
||||||
pub(super) async fn post_logout(State(state): State<AppState>) -> Response {
|
pub(super) async fn post_logout(State(state): State<AppState>) -> Response {
|
||||||
// Step 1: SIGINT claude (best-effort, matches `post_cancel_turn`).
|
// 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
|
// Step 2: delete OAuth credential files only — login::clear_session owns
|
||||||
// the file set and preserves session-history files alongside them.
|
// the file set and preserves session-history files alongside them.
|
||||||
let dir = crate::paths::claude_dir();
|
let dir = crate::paths::claude_dir();
|
||||||
|
|
|
||||||
|
|
@ -340,21 +340,40 @@ fn find_claude_child() -> Option<u32> {
|
||||||
/// (best-effort). See [`find_claude_child`] for how it's identified —
|
/// (best-effort). See [`find_claude_child`] for how it's identified —
|
||||||
/// **not** a name-based `pkill claude`, which would also hit an unrelated
|
/// **not** a name-based `pkill claude`, which would also hit an unrelated
|
||||||
/// `claude` process someone is running interactively in the container.
|
/// `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 {
|
let Some(pid) = find_claude_child() else {
|
||||||
return SigintOutcome::NoProcess;
|
return SigintOutcome::NoProcess;
|
||||||
};
|
};
|
||||||
match tokio::process::Command::new("kill")
|
sigint_turn(pid)
|
||||||
.args(["-INT", &pid.to_string()])
|
}
|
||||||
.status()
|
|
||||||
.await
|
/// 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
|
||||||
Ok(status) if status.success() => SigintOutcome::Signalled,
|
/// turn's claude as the leader of its own group. A `claude` child that
|
||||||
// Non-zero from `kill` means "no such process" (ESRCH) — it already
|
/// leads none (`plugins.rs`'s `claude plugin` calls) shares the harness's
|
||||||
// exited between the check above and the signal. Same as never
|
/// group, so it is signalled alone.
|
||||||
// having found it.
|
fn sigint_turn(pid: u32) -> SigintOutcome {
|
||||||
Ok(_) => SigintOutcome::NoProcess,
|
// A child is never pid 0 or 1, and `-0` / `-1` would reach the
|
||||||
Err(e) => SigintOutcome::Failed(e),
|
// 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)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -419,3 +438,105 @@ 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::time::Duration;
|
||||||
|
|
||||||
|
use super::{SigintOutcome, sigint_turn};
|
||||||
|
|
||||||
|
/// 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)
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
|
||||||
|
|
@ -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
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -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())
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -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);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -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:#}"),
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -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
|
||||||
|
|
@ -3731,6 +3779,178 @@ mod tests {
|
||||||
std::fs::remove_dir_all(&dir).ok();
|
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]
|
#[test]
|
||||||
fn a_cancelled_run_stops_with_turns_still_on_its_clock() {
|
fn a_cancelled_run_stops_with_turns_still_on_its_clock() {
|
||||||
// The loop half of the same fact, asserted directly: `Cancelled` is
|
// The loop half of the same fact, asserted directly: `Cancelled` is
|
||||||
|
|
@ -3985,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,
|
||||||
|
|
@ -4000,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");
|
||||||
|
|
@ -4016,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")
|
||||||
})
|
})
|
||||||
|
|
@ -4026,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")
|
||||||
})
|
})
|
||||||
|
|
@ -4539,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
|
||||||
|
|
@ -4889,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");
|
||||||
|
|
@ -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
|
// 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);
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue