diff --git a/hive-ag3nt/src/bin/hive-ag3nt.rs b/hive-ag3nt/src/bin/hive-ag3nt.rs index ae237721..3e4cfb5c 100644 --- a/hive-ag3nt/src/bin/hive-ag3nt.rs +++ b/hive-ag3nt/src/bin/hive-ag3nt.rs @@ -99,6 +99,7 @@ async fn main() -> Result<()> { &cli.socket, Duration::from_millis(poll_ms), login_state, + claude_dir, bus, stats, &files, @@ -116,6 +117,7 @@ async fn main() -> Result<()> { &cli.socket, Duration::from_millis(poll_ms), login_state, + claude_dir, bus, stats, &files, @@ -153,7 +155,8 @@ async fn main() -> Result<()> { async fn serve( socket: &Path, interval: Duration, - _login_state: Arc>, + login_state: Arc>, + claude_dir: std::path::PathBuf, bus: Bus, stats: Option, files: &turn::TurnFiles, @@ -178,7 +181,23 @@ async fn serve( match recv { Ok(AgentResponse::Messages { messages }) if !messages.is_empty() => { let first = messages.into_iter().next().expect("checked non-empty"); - handle_agent_turn(socket, &bus, stats.as_ref(), files, &turn_lock, label, first).await; + let auth_failed = + handle_agent_turn(socket, &bus, stats.as_ref(), files, &turn_lock, label, first) + .await; + if auth_failed { + // Park: flip LoginState + wait for the operator's + // re-auth to repopulate claude_dir. wait_for_login + // emits `online` on resume, which clears the + // needs_login sentinel. + *login_state.lock().unwrap() = LoginState::NeedsLogin; + turn::wait_for_login( + &claude_dir, + login_state.clone(), + &bus, + u64::try_from(interval.as_millis()).unwrap_or(2000), + ) + .await; + } } Ok(AgentResponse::Messages { .. }) => { // Idle: empty list = nothing pending. Brief sleep @@ -208,7 +227,9 @@ async fn serve( } } -/// Drive one turn for a received agent-inbox message. +/// Drive one turn for a received agent-inbox message. Returns `true` +/// when the turn ended with `AuthFailed` so the caller knows to park +/// in `wait_for_login`. async fn handle_agent_turn( socket: &Path, bus: &Bus, @@ -217,7 +238,7 @@ async fn handle_agent_turn( turn_lock: &TurnLock, label: &str, first: hive_sh4re::DeliveredMessage, -) { +) -> bool { let from = first.from; let body = first.body; let redelivered = first.redelivered; @@ -251,6 +272,19 @@ async fn handle_agent_turn( requeue_inflight(socket).await; bus.emit_status("online"); } + // 401: flip into needs_login + requeue the message that triggered + // the turn so it survives the re-auth. The serve loop's outer + // login-state watcher parks until the operator's `/login` flow + // completes; once it does, the requeued message replays the turn + // (closes #419). + if matches!(outcome, turn::TurnOutcome::AuthFailed) { + bus.emit_status("needs_login_idle"); + bus.emit(LiveEvent::Note { + text: "API 401 — waiting for re-login via web UI".into(), + }); + tracing::warn!("auth-failed; parking until re-login"); + requeue_inflight(socket).await; + } // Real crash: PromptTooLong is absorbed by compaction inside drive_turn. if let turn::TurnOutcome::Failed(e) = &outcome { notify_manager_of_failure(socket, label, e).await; @@ -280,6 +314,7 @@ async fn handle_agent_turn( // `request_next_turn` MCP tool: agent wrote a sentinel requesting // an immediate self-continuation. Clear and inject synthetic wake. check_and_inject_continue(socket, label).await; + matches!(outcome, turn::TurnOutcome::AuthFailed) } // Per-turn user prompt: the role/tools/etc. is in the system prompt diff --git a/hive-ag3nt/src/bin/hive-m1nd.rs b/hive-ag3nt/src/bin/hive-m1nd.rs index d85dc231..4c7f362a 100644 --- a/hive-ag3nt/src/bin/hive-m1nd.rs +++ b/hive-ag3nt/src/bin/hive-m1nd.rs @@ -88,6 +88,8 @@ async fn main() -> Result<()> { serve( &cli.socket, Duration::from_millis(poll_ms), + login_state, + claude_dir, bus, stats, &files, @@ -96,10 +98,12 @@ async fn main() -> Result<()> { .await } LoginState::NeedsLogin => { - turn::wait_for_login(&claude_dir, login_state, &bus, poll_ms).await; + turn::wait_for_login(&claude_dir, login_state.clone(), &bus, poll_ms).await; serve( &cli.socket, Duration::from_millis(poll_ms), + login_state, + claude_dir, bus, stats, &files, @@ -113,9 +117,12 @@ async fn main() -> Result<()> { } } +#[allow(clippy::too_many_arguments)] async fn serve( socket: &Path, interval: Duration, + login_state: Arc>, + claude_dir: std::path::PathBuf, bus: Bus, stats: Option, files: &turn::TurnFiles, @@ -144,7 +151,19 @@ async fn serve( match recv { Ok(ManagerResponse::Messages { messages }) if !messages.is_empty() => { let first = messages.into_iter().next().expect("checked non-empty"); - handle_manager_turn(socket, &bus, stats.as_ref(), files, &turn_lock, first).await; + let auth_failed = + handle_manager_turn(socket, &bus, stats.as_ref(), files, &turn_lock, first) + .await; + if auth_failed { + *login_state.lock().unwrap() = LoginState::NeedsLogin; + turn::wait_for_login( + &claude_dir, + login_state.clone(), + &bus, + u64::try_from(interval.as_millis()).unwrap_or(2000), + ) + .await; + } } Ok(ManagerResponse::Messages { .. }) => { // Idle: empty list = nothing pending. Brief sleep @@ -176,6 +195,8 @@ async fn serve( /// Drive one turn for a received manager-inbox message. Called from the /// serve loop for the non-empty-messages arm to keep that loop readable. +/// Returns `true` when the turn ended with `AuthFailed` so the caller +/// can park in `wait_for_login`. async fn handle_manager_turn( socket: &Path, bus: &Bus, @@ -183,7 +204,7 @@ async fn handle_manager_turn( files: &turn::TurnFiles, turn_lock: &TurnLock, first: hive_sh4re::DeliveredMessage, -) { +) -> bool { let from = first.from; let body = first.body; let redelivered = first.redelivered; @@ -229,6 +250,14 @@ async fn handle_manager_turn( requeue_inflight(socket).await; bus.emit_status("online"); } + if matches!(outcome, turn::TurnOutcome::AuthFailed) { + bus.emit_status("needs_login_idle"); + bus.emit(LiveEvent::Note { + text: "API 401 — waiting for re-login via web UI".into(), + }); + tracing::warn!("auth-failed; parking until re-login"); + requeue_inflight(socket).await; + } if let Some(stats) = stats { let ended_at = serve_common::now_unix(); let duration_ms = @@ -251,6 +280,7 @@ async fn handle_manager_turn( if pending > 0 { tracing::info!(%pending, "pending messages after turn; fetching next"); } + matches!(outcome, turn::TurnOutcome::AuthFailed) } /// Best-effort: tell the broker every message popped during the turn diff --git a/hive-ag3nt/src/events.rs b/hive-ag3nt/src/events.rs index f7e9c89f..bae95e30 100644 --- a/hive-ag3nt/src/events.rs +++ b/hive-ag3nt/src/events.rs @@ -684,19 +684,33 @@ impl Bus { /// `Arc>` should also call this so the web UI /// drops its periodic /api/state poll while a turn loop is running. /// - /// `"rate_limited"` sets the rate-limited flag and writes a sentinel - /// file at `{state_dir}/hyperhive-rate-limited` so the host-side - /// dashboard can show the status without a live socket call. - /// Any other status clears the flag and removes the sentinel. + /// Sentinel files survive harness restart so the host-side dashboard + /// can render the status without a live socket call: + /// - `"rate_limited"` writes `{state_dir}/hyperhive-rate-limited` + /// (cleared by any other status). + /// - `"needs_login_idle"` writes `{state_dir}/hyperhive-needs-login` + /// so a 401-triggered re-auth flag persists across harness restart + /// (#419). The web UI's `/login` POST handler clears it via + /// `clear_needs_login_sentinel` once the operator re-auths. + /// - `"online"` clears both sentinels — the agent is healthy again. pub fn emit_status(&self, status: impl Into) { let status = status.into(); - let sentinel = crate::paths::state_dir().join("hyperhive-rate-limited"); + let rate_limited_path = crate::paths::state_dir().join("hyperhive-rate-limited"); + let needs_login_path = crate::paths::state_dir().join("hyperhive-needs-login"); if status == "rate_limited" { self.rate_limited.store(true, Ordering::Relaxed); - let _ = std::fs::write(&sentinel, b""); + let _ = std::fs::write(&rate_limited_path, b""); } else { self.rate_limited.store(false, Ordering::Relaxed); - let _ = std::fs::remove_file(&sentinel); + let _ = std::fs::remove_file(&rate_limited_path); + } + if status == "needs_login_idle" { + let _ = std::fs::write(&needs_login_path, b""); + } else if status == "online" { + // Re-auth completed (or manual flip back to online) — drop + // the sentinel. `needs_login_in_progress` is a transient + // mid-flow status and shouldn't clear yet. + let _ = std::fs::remove_file(&needs_login_path); } self.emit(LiveEvent::StatusChanged { status }); } diff --git a/hive-ag3nt/src/serve_common.rs b/hive-ag3nt/src/serve_common.rs index a84dc9fe..44c80a1f 100644 --- a/hive-ag3nt/src/serve_common.rs +++ b/hive-ag3nt/src/serve_common.rs @@ -66,6 +66,7 @@ pub fn build_row( TurnOutcome::Compacted => ("compacted", None), TurnOutcome::PromptTooLong => ("prompt_too_long", None), TurnOutcome::RateLimited => ("rate_limited", None), + TurnOutcome::AuthFailed => ("auth_failed", None), TurnOutcome::Failed(e) => ("failed", Some(format!("{e:#}"))), }; TurnStatRow { diff --git a/hive-ag3nt/src/turn.rs b/hive-ag3nt/src/turn.rs index 12a7e136..d0297832 100644 --- a/hive-ag3nt/src/turn.rs +++ b/hive-ag3nt/src/turn.rs @@ -47,6 +47,21 @@ const RATE_LIMIT_MARKERS: &[&str] = &[ "Request rate limit exceeded", ]; +/// Substrings that indicate the Anthropic API rejected the request as +/// unauthenticated — the OAuth session in `/root/.claude/` has expired +/// or been revoked. Surfaced as `TurnOutcome::AuthFailed`, which the +/// harness uses to flip the container into `needs_login_idle` so the +/// dashboard's re-auth flow takes over (closes #419). Matched against +/// both stdout JSON `error` events and stderr; the markers come from +/// claude-code's `api_retry` events (`{"error":"authentication_failed", +/// "error_status":401,...}`) and the human-readable +/// "Failed to authenticate. API Error: 401" line claude prints on giveup. +const AUTH_FAIL_MARKERS: &[&str] = &[ + "\"error\":\"authentication_failed\"", + "\"error_status\":401", + "Failed to authenticate. API Error: 401", +]; + /// How long to sleep after detecting a rate-limit before re-entering the /// serve loop. Overridable via `HIVE_RATE_LIMIT_SLEEP_SECS`. Default is /// 5 minutes — enough for most short-lived throttles; the operator can @@ -196,6 +211,11 @@ pub enum TurnOutcome { /// usage cap, or exhausted credit balance. The serve loop should park for /// `rate_limit_sleep_secs()` and retry — NOT bubble up as a crash. RateLimited, + /// The Anthropic API rejected the request with 401 (OAuth session + /// expired or revoked). The serve loop should flip the container + /// into `needs_login_idle` and stop driving turns until the + /// operator re-auths via the per-agent web UI (closes #419). + AuthFailed, Failed(anyhow::Error), } @@ -343,6 +363,9 @@ async fn maybe_checkpoint_and_compact(files: &TurnFiles, bus: &Bus) -> bool { TurnOutcome::RateLimited => bus.emit(LiveEvent::Note { text: "checkpoint turn was rate-limited — compacting anyway".into(), }), + TurnOutcome::AuthFailed => bus.emit(LiveEvent::Note { + text: "checkpoint turn hit 401 — skipping compaction, parking for re-login".into(), + }), TurnOutcome::Failed(e) => bus.emit(LiveEvent::Note { text: format!("checkpoint turn failed ({e:#}) — compacting anyway"), }), @@ -413,6 +436,13 @@ pub fn emit_turn_end(bus: &Bus, outcome: &TurnOutcome) { }); tracing::warn!("turn rate-limited"); } + TurnOutcome::AuthFailed => { + bus.emit(LiveEvent::TurnEnd { + ok: false, + note: Some("authentication failed (401) — waiting for re-login".into()), + }); + tracing::warn!("turn auth-failed (401)"); + } TurnOutcome::Failed(e) => { let note = format!("{e:#}"); bus.emit(LiveEvent::TurnEnd { @@ -461,8 +491,9 @@ pub async fn wait_for_login( /// doesn't stall mid-turn — hyperhive owns compaction. pub async fn run_turn(prompt: &str, files: &TurnFiles, bus: &Bus) -> TurnOutcome { match run_claude(prompt, files, bus).await { - Ok((too_long, _)) if too_long => TurnOutcome::PromptTooLong, - Ok((_, rate_limited)) if rate_limited => TurnOutcome::RateLimited, + Ok((too_long, _, _)) if too_long => TurnOutcome::PromptTooLong, + Ok((_, rate_limited, _)) if rate_limited => TurnOutcome::RateLimited, + Ok((_, _, auth_failed)) if auth_failed => TurnOutcome::AuthFailed, Ok(_) => TurnOutcome::Ok, Err(e) => TurnOutcome::Failed(e), } @@ -483,7 +514,7 @@ pub async fn compact_session(files: &TurnFiles, bus: &Bus) -> Result<()> { bus.emit(LiveEvent::Note { text: "context overflow — running /compact on the persistent session".into(), }); - let (_, _) = run_claude("/compact", files, bus).await?; + let (_, _, _) = run_claude("/compact", files, bus).await?; bus.emit(LiveEvent::Note { text: "/compact done".into(), }); @@ -491,7 +522,7 @@ pub async fn compact_session(files: &TurnFiles, bus: &Bus) -> Result<()> { } #[allow(clippy::too_many_lines)] -async fn run_claude(prompt: &str, files: &TurnFiles, bus: &Bus) -> Result<(bool, bool)> { +async fn run_claude(prompt: &str, files: &TurnFiles, bus: &Bus) -> Result<(bool, bool, bool)> { // Keep the last STDERR_TAIL_LINES of stderr so a non-zero exit can // include real context in the bail message (and downstream in the // failure notification to the manager) instead of just "exit 1". @@ -547,10 +578,13 @@ async fn run_claude(prompt: &str, files: &TurnFiles, bus: &Bus) -> Result<(bool, let prompt_too_long = Arc::new(AtomicBool::new(false)); let rate_limited = Arc::new(AtomicBool::new(false)); + let auth_failed = Arc::new(AtomicBool::new(false)); let flag_out = prompt_too_long.clone(); let flag_err = prompt_too_long.clone(); let rate_out = rate_limited.clone(); let rate_err = rate_limited.clone(); + let auth_out = auth_failed.clone(); + let auth_err = auth_failed.clone(); let bus_out = bus.clone(); let bus_err = bus.clone(); let pump_stdout = tokio::spawn(async move { @@ -566,6 +600,13 @@ async fn run_claude(prompt: &str, files: &TurnFiles, bus: &Bus) -> Result<(bool, if line.contains(PROMPT_TOO_LONG_MARKER) { flag_out.store(true, Ordering::Relaxed); } + // Auth-fail check happens on the raw line first so we + // catch both the `api_retry` JSON events (which can land + // before they're fully parseable) and any stderr-shaped + // text that snuck onto stdout. + if AUTH_FAIL_MARKERS.iter().any(|m| line.contains(m)) { + auth_out.store(true, Ordering::Relaxed); + } match serde_json::from_str::(&line) { Ok(v) => { // Rate-limit detection: only fire on JSON `error` events, @@ -628,6 +669,9 @@ async fn run_claude(prompt: &str, files: &TurnFiles, bus: &Bus) -> Result<(bool, if RATE_LIMIT_MARKERS.iter().any(|m| line.contains(m)) { rate_err.store(true, Ordering::Relaxed); } + if AUTH_FAIL_MARKERS.iter().any(|m| line.contains(m)) { + auth_err.store(true, Ordering::Relaxed); + } // Mirror to journald so post-mortems work without the web UI // or the events sqlite. The bus event is what the dashboard // renders; the tracing line is what `journalctl -M -b` @@ -649,7 +693,8 @@ async fn run_claude(prompt: &str, files: &TurnFiles, bus: &Bus) -> Result<(bool, let _ = pump_stderr.await; let too_long = prompt_too_long.load(Ordering::Relaxed); let is_rate_limited = rate_limited.load(Ordering::Relaxed); - if !status.success() && !too_long && !is_rate_limited { + let is_auth_failed = auth_failed.load(Ordering::Relaxed); + if !status.success() && !too_long && !is_rate_limited && !is_auth_failed { let tail = stderr_tail.lock().unwrap(); if tail.is_empty() { bail!("claude exited {status} (no stderr)"); @@ -657,5 +702,5 @@ async fn run_claude(prompt: &str, files: &TurnFiles, bus: &Bus) -> Result<(bool, let tail_str = tail.iter().cloned().collect::>().join("\n"); bail!("claude exited {status}\nstderr tail:\n{tail_str}"); } - Ok((too_long, is_rate_limited)) + Ok((too_long, is_rate_limited, is_auth_failed)) } diff --git a/hive-c0re/src/container_view.rs b/hive-c0re/src/container_view.rs index 01be7ef7..f0e56e91 100644 --- a/hive-c0re/src/container_view.rs +++ b/hive-c0re/src/container_view.rs @@ -117,8 +117,14 @@ pub async fn build_all(coord: &Coordinator) -> Vec { }; let deployed_full = locked.get(&format!("agent-{logical}")).map(std::string::String::as_str); let needs_update = crate::auto_update::agent_config_pending(&logical, deployed_full); - let needs_login = - !is_manager && !claude_has_session(&Coordinator::agent_claude_dir(&logical)); + // needs_login fires when EITHER the claude session dir is + // missing (boot-time / fresh container) OR the harness wrote + // the auth-failed sentinel because a turn hit 401 (#419). The + // manager has its own session lifecycle and never participates + // in needs_login. + let needs_login = !is_manager + && (!claude_has_session(&Coordinator::agent_claude_dir(&logical)) + || auth_failed_sentinel(&logical)); let deployed_sha = deployed_full.map(|s| s[..s.len().min(12)].to_owned()); // Recipient name the broker uses for this agent — sub-agents // are addressed by logical name, the manager by the @@ -199,6 +205,17 @@ fn is_rate_limited(name: &str) -> bool { .exists() } +/// True when the harness wrote `{state_dir}/hyperhive-needs-login` +/// after a 401 mid-turn. Lets the dashboard surface `needs_login` for +/// agents whose `/root/.claude/` dir still exists (so +/// `claude_has_session` returns true) but whose OAuth credentials +/// inside it have actually expired (#419). +fn auth_failed_sentinel(name: &str) -> bool { + Coordinator::agent_notes_dir(name) + .join("hyperhive-needs-login") + .exists() +} + /// Read the agent's free-text status and the Unix timestamp when it was last set /// (derived from the file's mtime). Returns `(None, None)` when the file is absent /// or empty. `pub` so `agent_server` and `manager_server` can populate `AgentMeta`.