vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

harness: flip into needs_login on 401 mid-turn (closes #419)

Browse source
This commit is contained in:
damocles 2026-05-25 21:32:01 +02:00 • committed by Mara
commit 799804e3d1
6 changed files with 164 additions and 22 deletions
Download patch file Download diff file Expand all files Collapse all files

43
hive-ag3nt/src/bin/hive-ag3nt.rs
View file

@ -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<Mutex<LoginState>>,
login_state: Arc<Mutex<LoginState>>,
claude_dir: std::path::PathBuf,
bus: Bus,
stats: Option<TurnStats>,
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

36
hive-ag3nt/src/bin/hive-m1nd.rs
View file

@ -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<Mutex<LoginState>>,
claude_dir: std::path::PathBuf,
bus: Bus,
stats: Option<TurnStats>,
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

28
hive-ag3nt/src/events.rs
View file

@ -684,19 +684,33 @@ impl Bus {
/// `Arc<Mutex<LoginState>>` 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<String>) {
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 });
}

1
hive-ag3nt/src/serve_common.rs
View file

@ -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 {

57
hive-ag3nt/src/turn.rs
View file

@ -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::<serde_json::Value>(&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 <c> -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::<Vec<_>>().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))
}

21
hive-c0re/src/container_view.rs
View file

@ -117,8 +117,14 @@ pub async fn build_all(coord: &Coordinator) -> Vec<ContainerView> {
};
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`.