From bf93a81e1d136b10f805d7deca5d9eb9cd0240bf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?m=C3=BCde?= Date: Sun, 5 Jul 2026 19:08:48 +0200 Subject: [PATCH] refactor(agent): move login-wait into login.rs, dedupe env resolvers --- hive-ag3nt/src/bin/hive.rs | 4 +- hive-ag3nt/src/login.rs | 217 +++++++++++++++++++++++++++++++ hive-ag3nt/src/turn.rs | 259 +++---------------------------------- 3 files changed, 238 insertions(+), 242 deletions(-) diff --git a/hive-ag3nt/src/bin/hive.rs b/hive-ag3nt/src/bin/hive.rs index 967ff414..7248fa4d 100644 --- a/hive-ag3nt/src/bin/hive.rs +++ b/hive-ag3nt/src/bin/hive.rs @@ -465,7 +465,7 @@ async fn serve_main(socket: &Path, poll_ms: u64) -> Result<()> { } }); if matches!(initial, LoginState::NeedsLogin) { - turn::wait_for_login(&claude_dir, login_state.clone(), &bus, poll_ms).await; + login::wait_for_login(&claude_dir, login_state.clone(), &bus, poll_ms).await; } else { // Clear any stale `hyperhive-needs-login` sentinel left over // from a prior boot — `online` status writes the sentinel @@ -550,7 +550,7 @@ async fn serve_loop( let ctrl = handle_turn::(socket, &bus, stats.as_ref(), files, &turn_lock, next).await; if ctrl.auth_failed { *login_state.lock().unwrap() = LoginState::NeedsLogin; - turn::wait_for_login( + login::wait_for_login( &claude_dir, login_state.clone(), &bus, diff --git a/hive-ag3nt/src/login.rs b/hive-ag3nt/src/login.rs index dd5d429a..33a50a43 100644 --- a/hive-ag3nt/src/login.rs +++ b/hive-ag3nt/src/login.rs @@ -9,6 +9,10 @@ //! exact layout is locked in. use std::path::{Path, PathBuf}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use crate::events::Bus; /// Returns the Claude credentials directory for this agent. Delegates /// to `paths::claude_dir`, which reads `$HOME/.claude`. The service @@ -56,3 +60,216 @@ impl LoginState { } } } + +/// Block until the bound `~/.claude/` dir contains a session that +/// post-dates this call, polling on a `poll_ms` interval (min 2s). +/// Flips `state` to `Online` when login lands; caller resumes its +/// serve loop. Snapshots the dir at entry and only resumes when the +/// snapshot advances (mtime OR file-count change), avoiding the +/// infinite-401 loop a bare-existence check would produce when stale +/// credentials are already on disk. Mtime-snapshot resumption rationale +/// and `DirSnapshot` two-axis design: see +/// [`docs/turn-loop.md::The loop`](../../docs/turn-loop.md). +/// +/// # Panics +/// +/// Panics if the internal login-state lock is poisoned. +pub async fn wait_for_login( + claude_dir: &Path, + state: Arc>, + bus: &Bus, + poll_ms: u64, +) { + tracing::warn!( + claude_dir = %claude_dir.display(), + "no claude session — staying in partial-run mode (web UI only)" + ); + // Announce `needs_login_idle` to the bus so the sentinel file + // (`{state_dir}/hyperhive-needs-login`) gets written on every entry + // path — cold-boot, 401-mid-turn, and `/api/logout`. The host's + // `auth_failed_sentinel` reads that file to surface `needs_login` + // on the dashboard. Idempotent — `emit_status` is a `write` on a + // small empty file, so re-entering this function after a transient + // operator action is a no-op for the on-disk state. + bus.emit_status("needs_login_idle"); + let snapshot = snapshot_dir(claude_dir); + let probe = Duration::from_millis(poll_ms.max(2000)); + loop { + tokio::time::sleep(probe).await; + if session_refreshed(snapshot, snapshot_dir(claude_dir)) { + tracing::info!("claude session refreshed — entering turn loop"); + *state.lock().unwrap() = LoginState::Online; + bus.emit_status("online"); + return; + } + } +} + +/// Snapshot of the credentials dir at a point in time: number of +/// regular files + newest `mtime` across them. The two axes are both +/// load-bearing for `wait_for_login`'s refresh check (`session_refreshed`): +/// mtime catches the common case (re-login overwrites an existing +/// credentials file in-place), `file_count` catches the pathological case +/// where `meta.modified()` errors on every file (exotic fs, NFS quirks) +/// so the mtime axis stays `None` forever but new files still trigger a +/// resume. Defaults to `{0, None}` on `read_dir` failure (missing or +/// unreadable dir) — `wait_for_login` then resumes when files first +/// appear. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +struct DirSnapshot { + file_count: usize, + newest_mtime: Option, +} + +fn snapshot_dir(dir: &Path) -> DirSnapshot { + let Ok(entries) = std::fs::read_dir(dir) else { + return DirSnapshot::default(); + }; + let mut snap = DirSnapshot::default(); + for entry in entries.flatten() { + if !entry.file_type().is_ok_and(|t| t.is_file()) { + continue; + } + snap.file_count += 1; + let Ok(meta) = entry.metadata() else { continue }; + let Ok(mtime) = meta.modified() else { continue }; + if snap.newest_mtime.is_none_or(|cur| mtime > cur) { + snap.newest_mtime = Some(mtime); + } + } + snap +} + +/// Has the credentials dir been written since `prev`? Used as the +/// exit condition for `wait_for_login`: +/// +/// - `file_count` changed → something was added or removed, treat as +/// refresh (covers the "all files have unreadable mtime" edge case). +/// - `newest_mtime` advanced → existing file was rewritten in place +/// (the common claude re-login path). +/// - prev had no mtime (empty or all-unreadable) and now has one → +/// first useful signal we've seen, treat as refresh. +fn session_refreshed(prev: DirSnapshot, now: DirSnapshot) -> bool { + if now.file_count != prev.file_count { + return true; + } + match (prev.newest_mtime, now.newest_mtime) { + (None, Some(_)) => true, + (Some(p), Some(n)) => n > p, + _ => false, + } +} + +#[cfg(test)] +mod tests { + use std::fs; + use std::time::{Duration, SystemTime}; + + use super::{DirSnapshot, session_refreshed, snapshot_dir}; + + #[test] + fn snapshot_dir_empty_dir_is_default() { + let dir = tempfile::tempdir().unwrap(); + let snap = snapshot_dir(dir.path()); + assert_eq!(snap.file_count, 0); + assert!(snap.newest_mtime.is_none()); + } + + #[test] + fn snapshot_dir_missing_dir_is_default() { + // Defensive: a nonexistent dir must NOT panic. Bind mounts that + // disappear mid-poll (host purge during operator intervention) + // would otherwise crash the harness. + let missing = tempfile::tempdir() + .unwrap() + .path() + .join("never-created-subdir"); + let snap = snapshot_dir(&missing); + assert_eq!(snap, DirSnapshot::default()); + } + + #[test] + fn snapshot_dir_picks_latest_mtime_and_counts_files() { + let dir = tempfile::tempdir().unwrap(); + fs::write(dir.path().join("old.json"), b"{}").unwrap(); + // Sleep so the second file's mtime is strictly greater than + // the first on filesystems with low timestamp resolution. + std::thread::sleep(Duration::from_millis(20)); + let newer_path = dir.path().join("newer.json"); + fs::write(&newer_path, b"{}").unwrap(); + let snap = snapshot_dir(dir.path()); + assert_eq!(snap.file_count, 2); + let newer_meta = fs::metadata(&newer_path).unwrap().modified().unwrap(); + assert_eq!(snap.newest_mtime, Some(newer_meta)); + } + + #[test] + fn session_refreshed_first_login_flips_on_any_file() { + // Empty-dir snapshot → any file appearing means a fresh + // login landed. First-time login semantics. + let dir = tempfile::tempdir().unwrap(); + let snapshot = snapshot_dir(dir.path()); + assert!(!session_refreshed(snapshot, snapshot_dir(dir.path()))); + fs::write(dir.path().join("credentials.json"), b"{}").unwrap(); + assert!(session_refreshed(snapshot, snapshot_dir(dir.path()))); + } + + #[test] + fn session_refreshed_stale_creds_dont_flip_immediately() { + // Stale credentials.json already exists at entry; wait_for_login + // must NOT immediately return — it would loop straight into + // another 401-failing turn. + let dir = tempfile::tempdir().unwrap(); + fs::write(dir.path().join("credentials.json"), b"{}").unwrap(); + let snapshot = snapshot_dir(dir.path()); + assert_eq!(snapshot.file_count, 1); + // No change to the file → loop must NOT exit. + assert!(!session_refreshed(snapshot, snapshot_dir(dir.path()))); + } + + #[test] + fn session_refreshed_after_creds_rewrite_flips() { + // After the stale-creds snapshot, the operator's `/login/code` + // flow lands a refreshed credentials file — its mtime bumps + // strictly past the snapshot and wait_for_login resumes. + let dir = tempfile::tempdir().unwrap(); + fs::write(dir.path().join("credentials.json"), b"{}").unwrap(); + let snapshot = snapshot_dir(dir.path()); + std::thread::sleep(Duration::from_millis(20)); + fs::write(dir.path().join("credentials.json"), b"{\"v\":2}").unwrap(); + assert!(session_refreshed(snapshot, snapshot_dir(dir.path()))); + } + + #[test] + fn session_refreshed_snapshot_with_future_mtime_doesnt_flip() { + // Defensive: a snapshot set to a future timestamp (e.g. clock + // skew between snapshot and probe) must keep waiting until a + // file's mtime actually exceeds it, not return on first poll. + let dir = tempfile::tempdir().unwrap(); + fs::write(dir.path().join("credentials.json"), b"{}").unwrap(); + let snapshot = DirSnapshot { + file_count: 1, + newest_mtime: Some(SystemTime::now() + Duration::from_hours(1)), + }; + assert!(!session_refreshed(snapshot, snapshot_dir(dir.path()))); + } + + #[test] + fn session_refreshed_count_change_flips_when_mtime_unreadable() { + // Defensive: if all files have unreadable `meta.modified()` + // (exotic fs / NFS), newest_mtime stays `None` forever — but + // file_count axis still catches new files appearing. Simulated + // here by forging a snapshot with file_count=1 + no mtime, then + // writing a second file. + let dir = tempfile::tempdir().unwrap(); + fs::write(dir.path().join("a"), b"{}").unwrap(); + let forged = DirSnapshot { + file_count: 1, + newest_mtime: None, + }; + fs::write(dir.path().join("b"), b"{}").unwrap(); + // Real snapshot has file_count=2, so refresh fires even + // though the mtime axis would be inconclusive. + assert!(session_refreshed(forged, snapshot_dir(dir.path()))); + } +} diff --git a/hive-ag3nt/src/turn.rs b/hive-ag3nt/src/turn.rs index 3e428d38..c48e4875 100644 --- a/hive-ag3nt/src/turn.rs +++ b/hive-ag3nt/src/turn.rs @@ -6,15 +6,13 @@ //! compaction / auto-reset / retry state machine (`drive_turn`). use std::path::{Path, PathBuf}; -use std::sync::{Arc, Mutex}; -use std::time::Duration; +use std::sync::Mutex; use anyhow::Result; use hive_claude::{Claude, Config, Session, Sink}; use serde_json::Value; use crate::events::{Bus, LiveEvent, TokenUsage}; -use crate::login::LoginState; use crate::mcp; // Hive-enforced claude settings ship at `/etc/claude-code/managed-settings.json` @@ -163,15 +161,25 @@ pub enum TurnOutcome { Failed(anyhow::Error), } +/// Parse an env var as `u64`, ignoring absent / blank / unparseable values. +/// Returns the raw value including `0` (several knobs use `0` as "disable"). +fn env_u64(name: &str) -> Option { + std::env::var(name) + .ok() + .and_then(|s| s.trim().parse::().ok()) +} + +/// Like [`env_u64`] but also rejects `0`, falling back to `default` — for +/// knobs where `0` is meaningless rather than a "disable" sentinel. +fn env_u64_positive(name: &str, default: u64) -> u64 { + env_u64(name).filter(|&v| v > 0).unwrap_or(default) +} + /// How long to sleep after a rate-limit before re-entering the serve loop. /// Reads `HIVE_RATE_LIMIT_SLEEP_SECS` if set to a valid positive integer. #[must_use] pub fn rate_limit_sleep_secs() -> u64 { - std::env::var("HIVE_RATE_LIMIT_SLEEP_SECS") - .ok() - .and_then(|s| s.trim().parse::().ok()) - .filter(|&v| v > 0) - .unwrap_or(DEFAULT_RATE_LIMIT_SLEEP_SECS) + env_u64_positive("HIVE_RATE_LIMIT_SLEEP_SECS", DEFAULT_RATE_LIMIT_SLEEP_SECS) } /// Resolve the effective context-window size for watermark calculations. @@ -196,23 +204,13 @@ fn effective_context_window(bus: &Bus) -> u64 { /// /// `0` disables auto-reset entirely. fn auto_reset_watermark_tokens(bus: &Bus) -> u64 { - if let Some(v) = std::env::var("HIVE_AUTO_RESET_WATERMARK_TOKENS") - .ok() - .and_then(|s| s.trim().parse::().ok()) - { - return v; - } - effective_context_window(bus) / 2 + env_u64("HIVE_AUTO_RESET_WATERMARK_TOKENS").unwrap_or_else(|| effective_context_window(bus) / 2) } /// Resolve the assumed cache TTL: `HIVE_CACHE_TTL_SECS` if set, else /// `DEFAULT_CACHE_TTL_SECS`. fn cache_ttl_secs() -> u64 { - std::env::var("HIVE_CACHE_TTL_SECS") - .ok() - .and_then(|s| s.trim().parse::().ok()) - .filter(|&v| v > 0) - .unwrap_or(DEFAULT_CACHE_TTL_SECS) + env_u64_positive("HIVE_CACHE_TTL_SECS", DEFAULT_CACHE_TTL_SECS) } /// Resolve the proactive-compaction watermark. Priority order: @@ -221,13 +219,7 @@ fn cache_ttl_secs() -> u64 { /// /// `0` disables proactive compaction (reactive path still applies). fn compact_watermark_tokens(bus: &Bus) -> u64 { - if let Some(v) = std::env::var("HIVE_COMPACT_WATERMARK_TOKENS") - .ok() - .and_then(|s| s.trim().parse::().ok()) - { - return v; - } - effective_context_window(bus) * 3 / 4 + env_u64("HIVE_COMPACT_WATERMARK_TOKENS").unwrap_or_else(|| effective_context_window(bus) * 3 / 4) } /// Drive one turn end-to-end. Three paths layer on top of the raw `run_turn`: @@ -459,105 +451,6 @@ pub fn emit_turn_end(bus: &Bus, outcome: &TurnOutcome) { } } -/// Block until the bound `~/.claude/` dir contains a session that -/// post-dates this call, polling on a `poll_ms` interval (min 2s). -/// Flips `state` to `Online` when login lands; caller resumes its -/// serve loop. Snapshots the dir at entry and only resumes when the -/// snapshot advances (mtime OR file-count change), avoiding the -/// infinite-401 loop a bare-existence check would produce when stale -/// credentials are already on disk. Mtime-snapshot resumption rationale -/// and `DirSnapshot` two-axis design: see -/// [`docs/turn-loop.md::The loop`](../../docs/turn-loop.md). -/// -/// # Panics -/// -/// Panics if the internal login-state lock is poisoned. -pub async fn wait_for_login( - claude_dir: &Path, - state: Arc>, - bus: &Bus, - poll_ms: u64, -) { - tracing::warn!( - claude_dir = %claude_dir.display(), - "no claude session — staying in partial-run mode (web UI only)" - ); - // Announce `needs_login_idle` to the bus so the sentinel file - // (`{state_dir}/hyperhive-needs-login`) gets written on every entry - // path — cold-boot, 401-mid-turn, and `/api/logout`. The host's - // `auth_failed_sentinel` reads that file to surface `needs_login` - // on the dashboard. Idempotent — `emit_status` is a `write` on a - // small empty file, so re-entering this function after a transient - // operator action is a no-op for the on-disk state. - bus.emit_status("needs_login_idle"); - let snapshot = snapshot_dir(claude_dir); - let probe = Duration::from_millis(poll_ms.max(2000)); - loop { - tokio::time::sleep(probe).await; - if session_refreshed(snapshot, snapshot_dir(claude_dir)) { - tracing::info!("claude session refreshed — entering turn loop"); - *state.lock().unwrap() = LoginState::Online; - bus.emit_status("online"); - return; - } - } -} - -/// Snapshot of the credentials dir at a point in time: number of -/// regular files + newest `mtime` across them. The two axes are both -/// load-bearing for `wait_for_login`'s refresh check (`session_refreshed`): -/// mtime catches the common case (re-login overwrites an existing -/// credentials file in-place), `file_count` catches the pathological case -/// where `meta.modified()` errors on every file (exotic fs, NFS quirks) -/// so the mtime axis stays `None` forever but new files still trigger a -/// resume. Defaults to `{0, None}` on `read_dir` failure (missing or -/// unreadable dir) — `wait_for_login` then resumes when files first -/// appear. -#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] -struct DirSnapshot { - file_count: usize, - newest_mtime: Option, -} - -fn snapshot_dir(dir: &Path) -> DirSnapshot { - let Ok(entries) = std::fs::read_dir(dir) else { - return DirSnapshot::default(); - }; - let mut snap = DirSnapshot::default(); - for entry in entries.flatten() { - if !entry.file_type().is_ok_and(|t| t.is_file()) { - continue; - } - snap.file_count += 1; - let Ok(meta) = entry.metadata() else { continue }; - let Ok(mtime) = meta.modified() else { continue }; - if snap.newest_mtime.is_none_or(|cur| mtime > cur) { - snap.newest_mtime = Some(mtime); - } - } - snap -} - -/// Has the credentials dir been written since `prev`? Used as the -/// exit condition for `wait_for_login`: -/// -/// - `file_count` changed → something was added or removed, treat as -/// refresh (covers the "all files have unreadable mtime" edge case). -/// - `newest_mtime` advanced → existing file was rewritten in place -/// (the common claude re-login path). -/// - prev had no mtime (empty or all-unreadable) and now has one → -/// first useful signal we've seen, treat as refresh. -fn session_refreshed(prev: DirSnapshot, now: DirSnapshot) -> bool { - if now.file_count != prev.file_count { - return true; - } - match (prev.newest_mtime, now.newest_mtime) { - (None, Some(_)) => true, - (Some(p), Some(n)) => n > p, - _ => false, - } -} - /// Run one turn against the constant-title session, resuming it or creating /// it on first use (`hive_claude::run_resume_or_create`). The session is /// pinned by a fixed `--resume`/`--name ` (NOT bare `--continue`, which @@ -826,117 +719,3 @@ fn archive_session(bus: &Bus) { } } - -#[cfg(test)] -mod tests { - use std::fs; - use std::time::{Duration, SystemTime}; - - use super::{DirSnapshot, session_refreshed, snapshot_dir}; - - #[test] - fn snapshot_dir_empty_dir_is_default() { - let dir = tempfile::tempdir().unwrap(); - let snap = snapshot_dir(dir.path()); - assert_eq!(snap.file_count, 0); - assert!(snap.newest_mtime.is_none()); - } - - #[test] - fn snapshot_dir_missing_dir_is_default() { - // Defensive: a nonexistent dir must NOT panic. Bind mounts that - // disappear mid-poll (host purge during operator intervention) - // would otherwise crash the harness. - let missing = tempfile::tempdir() - .unwrap() - .path() - .join("never-created-subdir"); - let snap = snapshot_dir(&missing); - assert_eq!(snap, DirSnapshot::default()); - } - - #[test] - fn snapshot_dir_picks_latest_mtime_and_counts_files() { - let dir = tempfile::tempdir().unwrap(); - fs::write(dir.path().join("old.json"), b"{}").unwrap(); - // Sleep so the second file's mtime is strictly greater than - // the first on filesystems with low timestamp resolution. - std::thread::sleep(Duration::from_millis(20)); - let newer_path = dir.path().join("newer.json"); - fs::write(&newer_path, b"{}").unwrap(); - let snap = snapshot_dir(dir.path()); - assert_eq!(snap.file_count, 2); - let newer_meta = fs::metadata(&newer_path).unwrap().modified().unwrap(); - assert_eq!(snap.newest_mtime, Some(newer_meta)); - } - - #[test] - fn session_refreshed_first_login_flips_on_any_file() { - // Empty-dir snapshot → any file appearing means a fresh - // login landed. First-time login semantics. - let dir = tempfile::tempdir().unwrap(); - let snapshot = snapshot_dir(dir.path()); - assert!(!session_refreshed(snapshot, snapshot_dir(dir.path()))); - fs::write(dir.path().join("credentials.json"), b"{}").unwrap(); - assert!(session_refreshed(snapshot, snapshot_dir(dir.path()))); - } - - #[test] - fn session_refreshed_stale_creds_dont_flip_immediately() { - // Stale credentials.json already exists at entry; wait_for_login - // must NOT immediately return — it would loop straight into - // another 401-failing turn. - let dir = tempfile::tempdir().unwrap(); - fs::write(dir.path().join("credentials.json"), b"{}").unwrap(); - let snapshot = snapshot_dir(dir.path()); - assert_eq!(snapshot.file_count, 1); - // No change to the file → loop must NOT exit. - assert!(!session_refreshed(snapshot, snapshot_dir(dir.path()))); - } - - #[test] - fn session_refreshed_after_creds_rewrite_flips() { - // After the stale-creds snapshot, the operator's `/login/code` - // flow lands a refreshed credentials file — its mtime bumps - // strictly past the snapshot and wait_for_login resumes. - let dir = tempfile::tempdir().unwrap(); - fs::write(dir.path().join("credentials.json"), b"{}").unwrap(); - let snapshot = snapshot_dir(dir.path()); - std::thread::sleep(Duration::from_millis(20)); - fs::write(dir.path().join("credentials.json"), b"{\"v\":2}").unwrap(); - assert!(session_refreshed(snapshot, snapshot_dir(dir.path()))); - } - - #[test] - fn session_refreshed_snapshot_with_future_mtime_doesnt_flip() { - // Defensive: a snapshot set to a future timestamp (e.g. clock - // skew between snapshot and probe) must keep waiting until a - // file's mtime actually exceeds it, not return on first poll. - let dir = tempfile::tempdir().unwrap(); - fs::write(dir.path().join("credentials.json"), b"{}").unwrap(); - let snapshot = DirSnapshot { - file_count: 1, - newest_mtime: Some(SystemTime::now() + Duration::from_hours(1)), - }; - assert!(!session_refreshed(snapshot, snapshot_dir(dir.path()))); - } - - #[test] - fn session_refreshed_count_change_flips_when_mtime_unreadable() { - // Defensive: if all files have unreadable `meta.modified()` - // (exotic fs / NFS), newest_mtime stays `None` forever — but - // file_count axis still catches new files appearing. Simulated - // here by forging a snapshot with file_count=1 + no mtime, then - // writing a second file. - let dir = tempfile::tempdir().unwrap(); - fs::write(dir.path().join("a"), b"{}").unwrap(); - let forged = DirSnapshot { - file_count: 1, - newest_mtime: None, - }; - fs::write(dir.path().join("b"), b"{}").unwrap(); - // Real snapshot has file_count=2, so refresh fires even - // though the mtime axis would be inconclusive. - assert!(session_refreshed(forged, snapshot_dir(dir.path()))); - } -}