feat(#2109): harness-side idle watchdog to bail on anthropic api stall storms

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] \`<qualified-label>\` claude turn failed:\n<err>` to `<parent>` 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

hive-ag3nt/src/bin/hive-agent.rs

@@ -553,6 +553,22 @@ async fn handle_turn<S: Surface>(
         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 {

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-") {

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())),
     }
 }

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<String>,
     /// Program to spawn. `None` defaults to `claude` (resolved on `PATH`).
     pub program: Option<String>,
+    /// 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<Duration>,
 }

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!(
+        let ((), stderr_tail, (status, timed_out)) = tokio::join!(
-            pump_stdout(stdout, sink, &sentinels),
+            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 recognized sentinel takes precedence over everything (most specific
-        // a non-zero exit with no sentinel is a hard failure.
+        // 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<Duration>,
+) -> (std::io::Result<ExitStatus>, 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::<serde_json::Value>(&line) {
             // JSON stdout: classify with the model-content gate so an
             // `assistant`/`user` message quoting a marker can't trip it.

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.