diff --git a/docs/turn-loop.md b/docs/turn-loop.md index 2a16037e..9a4ea427 100644 --- a/docs/turn-loop.md +++ b/docs/turn-loop.md @@ -114,8 +114,17 @@ 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(Failed(err))` | route `[system] \`\` claude turn failed:\n` to `` via `send_to_parent` | +`ApiStall` catches an Anthropic API stall — a multi-retry connection storm where +the stream goes silent for minutes. The idle watchdog lives in `hive-claude`'s +driver (`Config::idle_timeout`, enforced around `child.wait()`): the timer +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`. + After the outcome handler, the stats sink records a row and the `hyperhive-continue` sentinel (dropped by the `request_next_turn` MCP tool) is consumed if present. `handle_turn` reports the result diff --git a/hive-ag3nt/src/bin/hive-agent.rs b/hive-ag3nt/src/bin/hive-agent.rs index 8c1fafed..d1aa5a1e 100644 --- a/hive-ag3nt/src/bin/hive-agent.rs +++ b/hive-ag3nt/src/bin/hive-agent.rs @@ -553,6 +553,22 @@ async fn handle_turn( S::requeue_inflight(socket).await; bus.emit_status("online"); } + if matches!(outcome, Err(turn::TurnError::ApiStall)) { + // Idle watchdog killed claude on a suspected API stall. Park briefly to + // let the API recover, then requeue — same shape as the rate-limit path. + let secs = turn::stall_sleep_secs(); + bus.emit_status("api_stall"); + bus.emit(LiveEvent::Note { + text: format!( + "API stall timeout — sleeping {secs}s before retry \ + (tune HIVE_STALL_SLEEP_SECS; disable the watchdog with HIVE_TURN_IDLE_SECS=0)" + ), + }); + tracing::warn!(sleep_secs = secs, "API stall; parking before retry"); + tokio::time::sleep(Duration::from_secs(secs)).await; + S::requeue_inflight(socket).await; + bus.emit_status("online"); + } if matches!(outcome, Err(turn::TurnError::AuthFailed)) { bus.emit_status("needs_login_idle"); bus.emit(LiveEvent::Note { diff --git a/hive-ag3nt/src/serve_common.rs b/hive-ag3nt/src/serve_common.rs index e8149d41..c43af732 100644 --- a/hive-ag3nt/src/serve_common.rs +++ b/hive-ag3nt/src/serve_common.rs @@ -98,6 +98,7 @@ pub fn build_row(args: TurnRowArgs<'_>) -> TurnStatRow { Err(TurnError::RateLimited) => ("rate_limited", None), Err(TurnError::AuthFailed) => ("auth_failed", None), Err(TurnError::SessionNotFound) => ("session_not_found", None), + Err(TurnError::ApiStall) => ("api_stall", None), Err(TurnError::Failed(e)) => ("failed", Some(format!("{e:#}"))), }; let wake_from = if wake_from.starts_with("bash-task-") { diff --git a/hive-ag3nt/src/turn.rs b/hive-ag3nt/src/turn.rs index 2f633bd1..4da56c2b 100644 --- a/hive-ag3nt/src/turn.rs +++ b/hive-ag3nt/src/turn.rs @@ -47,6 +47,19 @@ 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 +/// 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: +/// attempt:11 took ~6.7min on a 371k-token context). Override via +/// `HIVE_TURN_IDLE_SECS`; `0` disables the watchdog (wait indefinitely). +const DEFAULT_TURN_IDLE_SECS: u64 = 600; + +/// How long to park after an `ApiStall` before requeueing the message, giving +/// the API a chance to recover. Overridable via `HIVE_STALL_SLEEP_SECS`. +const DEFAULT_STALL_SLEEP_SECS: u64 = 60; + /// Assumed prompt-cache TTL. Claude caches prompt prefixes — ~5 minutes on /// the API (pay-per-token), ~1 hour on Claude Max (subscription). When the /// idle gap exceeds this, the cache prefix has likely expired and the next @@ -178,6 +191,12 @@ 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 + /// 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, /// A hard failure with no recovery — the serve loop escalates it to the /// parent (`send_to_parent`). Failed(anyhow::Error), @@ -204,6 +223,20 @@ pub fn rate_limit_sleep_secs() -> u64 { env_u64_positive("HIVE_RATE_LIMIT_SLEEP_SECS", DEFAULT_RATE_LIMIT_SLEEP_SECS) } +/// Idle-watchdog window in seconds. Reads `HIVE_TURN_IDLE_SECS`; `0` disables +/// the watchdog. Absent / unparseable falls back to [`DEFAULT_TURN_IDLE_SECS`]. +#[must_use] +pub fn turn_idle_secs() -> u64 { + env_u64("HIVE_TURN_IDLE_SECS").unwrap_or(DEFAULT_TURN_IDLE_SECS) +} + +/// How long to park after an `ApiStall` before requeueing. Reads +/// `HIVE_STALL_SLEEP_SECS` if set to a valid positive integer. +#[must_use] +pub fn stall_sleep_secs() -> u64 { + env_u64_positive("HIVE_STALL_SLEEP_SECS", DEFAULT_STALL_SLEEP_SECS) +} + /// Resolve the effective context-window size for watermark calculations. /// Priority order (first wins): /// 1. API-reported window from the last `result` event's `modelUsage.*.contextWindow`. @@ -445,6 +478,16 @@ pub fn emit_turn_end(bus: &Bus, outcome: &TurnOutcome) { }); tracing::warn!("turn session-not-found; requeueing message"); } + Err(TurnError::ApiStall) => { + bus.emit(LiveEvent::TurnEnd { + ok: false, + note: Some(format!( + "claude killed after {}s of output silence — API stall suspected, parking + requeueing", + turn_idle_secs() + )), + }); + tracing::warn!("turn killed: API stall (idle watchdog)"); + } Err(TurnError::Failed(e)) => { let note = format!("{e:#}"); bus.emit(LiveEvent::TurnEnd { @@ -542,6 +585,12 @@ fn claude_config(bus: &Bus, files: &TurnFiles) -> Config { tools: Some(mcp_config::builtin_tools_arg()), allowed_tools: Some(mcp_config::allowed_tools_arg()), add_dirs, + // Idle watchdog: `0` disables (wait indefinitely), any positive value + // caps output silence. Policy lives here; the driver just enforces it. + idle_timeout: match turn_idle_secs() { + 0 => None, + secs => Some(std::time::Duration::from_secs(secs)), + }, ..Config::default() } } @@ -557,6 +606,7 @@ fn error_to_turn(err: hive_claude::Error) -> TurnOutcome { Error::RateLimited => Err(TurnError::RateLimited), Error::AuthFailed => Err(TurnError::AuthFailed), Error::SessionNotFound => Err(TurnError::SessionNotFound), + Error::IdleTimeout => Err(TurnError::ApiStall), other => Err(TurnError::Failed(other.into())), } } diff --git a/hive-claude/src/config.rs b/hive-claude/src/config.rs index ae0cff93..b150cef3 100644 --- a/hive-claude/src/config.rs +++ b/hive-claude/src/config.rs @@ -1,6 +1,7 @@ //! Invocation config: how to build one `claude --print` command line. use std::path::PathBuf; +use std::time::Duration; /// How a run attaches to a claude session — the low-level session flag. Kept /// separate from [`Config`] so one config can drive resume + create + @@ -56,4 +57,11 @@ pub struct Config { pub extra_args: Vec, /// Program to spawn. `None` defaults to `claude` (resolved on `PATH`). pub program: Option, + /// Idle watchdog: kill the child and return [`crate::Error::IdleTimeout`] + /// if no stdout line arrives for this long. The timer resets on every + /// stdout line, so a large/slow but still-streaming turn is never cut; + /// only complete output silence trips it. `None` waits indefinitely. + /// The driver stays policy-free — the caller decides the window (and + /// whether to read it from the environment). + pub idle_timeout: Option, } diff --git a/hive-claude/src/driver.rs b/hive-claude/src/driver.rs index 7b808ad1..deba5f24 100644 --- a/hive-claude/src/driver.rs +++ b/hive-claude/src/driver.rs @@ -2,10 +2,12 @@ //! assemble the result. use std::collections::VecDeque; -use std::process::Stdio; +use std::process::{ExitStatus, Stdio}; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::{Duration, Instant}; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; -use tokio::process::{ChildStderr, ChildStdout, Command}; +use tokio::process::{Child, ChildStderr, ChildStdout, Command}; use crate::classify::Sentinels; use crate::{Attach, Config, Error, Result, Sink}; @@ -16,6 +18,11 @@ const DEFAULT_PROGRAM: &str = "claude"; /// How many trailing stderr lines to keep for [`Error::Exit`]. const STDERR_TAIL_LINES: usize = 20; +/// Idle-watchdog probe cadence: re-check output silence this often while the +/// child runs. Fine-grained enough to fire within ~one probe of the deadline, +/// cheap enough to ignore. +const IDLE_PROBE: Duration = Duration::from_secs(5); + /// The low-level driver entry point. A namespace for the run function — there /// is nothing to construct; call `Claude::run(…)` directly. For a durable, /// self-compacting session, use [`crate::InfiniteSession`] instead. @@ -37,6 +44,8 @@ impl Claude { /// [`Error::AuthFailed`], [`Error::SessionNotFound`]. /// - [`Error::Spawn`] if the binary can't be launched. /// - [`Error::Stdin`] / [`Error::Wait`] on stdin-write / child-wait failure. + /// - [`Error::IdleTimeout`] if `config.idle_timeout` is set and no stdout + /// line arrives within that window (the child is killed). /// - [`Error::Exit`] on a non-zero exit that raised no sentinel. pub async fn run( config: &Config, @@ -69,20 +78,29 @@ impl Claude { let stderr = child.stderr.take().expect("stderr piped"); let sentinels = Sentinels::default(); + // Idle watchdog clock: `last_activity` holds seconds-since-`base` of the + // last stdout line, bumped by the pump; the waiter reads it to detect a + // fully silent stall. Monotonic (`Instant`) so a wall-clock jump can't + // spuriously fire it. + let base = Instant::now(); + let last_activity = AtomicU64::new(0); // Pump both streams and wait for exit concurrently on this task — no // `spawn`, so the sink needn't be `'static` and borrows stay simple. - let ((), stderr_tail, status) = tokio::join!( - pump_stdout(stdout, sink, &sentinels), + let ((), stderr_tail, (status, timed_out)) = tokio::join!( + pump_stdout(stdout, sink, &sentinels, base, &last_activity), pump_stderr(stderr, sink, &sentinels), - child.wait(), + wait_with_idle(&mut child, base, &last_activity, config.idle_timeout), ); let status = status.map_err(Error::Wait)?; - // A recognized sentinel takes precedence over the exit code; otherwise - // a non-zero exit with no sentinel is a hard failure. + // A recognized sentinel takes precedence over everything (most specific + // reason). Then an idle-watchdog kill; then a plain non-zero exit. if let Some(sentinel) = sentinels.soft_error() { return Err(sentinel); } + if timed_out { + return Err(Error::IdleTimeout); + } if !status.success() { return Err(Error::Exit { status, @@ -93,6 +111,36 @@ impl Claude { } } +/// Wait for the child to exit, enforcing the optional idle watchdog. With no +/// `idle_timeout` this is a plain `child.wait()`. Otherwise it re-checks on a +/// fixed probe cadence: if no stdout line arrived within `idle_timeout` (per +/// `last_activity`, bumped by [`pump_stdout`]), it kills the child and reaps +/// it. Returns the exit status and whether the watchdog fired. +async fn wait_with_idle( + child: &mut Child, + base: Instant, + last_activity: &AtomicU64, + idle_timeout: Option, +) -> (std::io::Result, bool) { + let Some(window) = idle_timeout else { + return (child.wait().await, false); + }; + // `child.wait()` is cancel-safe, so dropping it on a probe timeout doesn't + // lose the exit. + loop { + match tokio::time::timeout(IDLE_PROBE, child.wait()).await { + Ok(status) => return (status, false), + Err(_probe_expired) => { + let last = Duration::from_secs(last_activity.load(Ordering::Relaxed)); + if base.elapsed().saturating_sub(last) >= window { + let _ = child.kill().await; + return (child.wait().await, true); + } + } + } + } +} + /// Assemble the argv. `--print --verbose --output-format stream-json` are /// mandatory (the driver parses that shape); everything else is gated on the /// [`Config`] / [`Attach`]. @@ -149,9 +197,18 @@ fn build_command(program: &str, config: &Config, attach: &Attach) -> Command { /// Read stdout line by line: classify each line, parse JSON, hand events (or /// raw non-JSON lines) to the sink. -async fn pump_stdout(stdout: ChildStdout, sink: &impl Sink, sentinels: &Sentinels) { +async fn pump_stdout( + stdout: ChildStdout, + sink: &impl Sink, + sentinels: &Sentinels, + base: Instant, + last_activity: &AtomicU64, +) { let mut lines = BufReader::new(stdout).lines(); while let Ok(Some(line)) = lines.next_line().await { + // Poke the idle watchdog: any stdout line resets the silence timer. + // Seconds granularity is plenty — the probe cadence is coarser still. + last_activity.store(base.elapsed().as_secs(), Ordering::Relaxed); if let Ok(event) = serde_json::from_str::(&line) { // JSON stdout: classify with the model-content gate so an // `assistant`/`user` message quoting a marker can't trip it. diff --git a/hive-claude/src/error.rs b/hive-claude/src/error.rs index e29deecd..227f290d 100644 --- a/hive-claude/src/error.rs +++ b/hive-claude/src/error.rs @@ -57,6 +57,14 @@ pub enum Error { #[error("waiting on claude failed: {0}")] Wait(#[source] std::io::Error), + /// The child produced no stdout for longer than the configured idle + /// window (`Config::idle_timeout`) and was killed. Indicative of an + /// Anthropic API stall (e.g. a multi-retry connection storm that goes + /// silent for minutes). Callers typically park briefly and retry, like + /// the rate-limit path. + #[error("claude idle timeout: no output for the configured window")] + IdleTimeout, + /// claude exited non-zero and raised none of the recognized sentinels. /// `stderr_tail` is the last handful of stderr lines (empty if there were /// none), included so the caller can surface a real diagnostic.