turn: wait_for_login resumes only on credentials mtime bump (closes #542)

hive-ag3nt/Cargo.toml

@@ -24,6 +24,9 @@ tower-http.workspace = true
 tracing.workspace = true
 tracing-subscriber.workspace = true
 
+[dev-dependencies]
+tempfile = "3"
+
 [[bin]]
 name = "hive-ag3nt"
 path = "src/bin/hive-ag3nt.rs"

hive-ag3nt/src/turn.rs

@@ -15,7 +15,7 @@ use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
 use tokio::process::Command;
 
 use crate::events::{Bus, LiveEvent};
-use crate::login::{self, LoginState};
+use crate::login::LoginState;
 use crate::mcp;
 
 /// `--settings` JSON applied to every claude invocation. Lives as a
@@ -464,9 +464,20 @@ pub fn emit_turn_end(bus: &Bus, outcome: &TurnOutcome) {
     }
 }
 
-/// Block until the bound `~/.claude/` dir contains a session, polling
-/// `claude_dir` on a `poll_ms` interval (min 2s). Flips `state` to
-/// `Online` when login lands; caller resumes its serve loop.
+/// 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.
+///
+/// **Mtime-progress, not bare existence (closes #542)**: an
+/// existence-only check (the pre-#542 behaviour) immediately returns
+/// after a 401 because the stale `credentials.json` is still on disk
+/// — the next turn then 401s on the same tokens and the harness
+/// loops forever. We snapshot the newest file mtime in `claude_dir`
+/// at entry and only resume when something has been written since
+/// that snapshot (the operator's `/login/code` flow lands a refreshed
+/// credentials file, bumping its mtime). First-time login (empty
+/// dir → `None` snapshot) still flips on the first file appearing.
 ///
 /// # Panics
 ///
@@ -481,11 +492,12 @@ pub async fn wait_for_login(
         claude_dir = %claude_dir.display(),
         "no claude session — staying in partial-run mode (web UI only)"
     );
+    let snapshot = newest_file_mtime(claude_dir);
     let probe = Duration::from_millis(poll_ms.max(2000));
     loop {
         tokio::time::sleep(probe).await;
-        if login::has_session(claude_dir) {
-            tracing::info!("claude session detected — entering turn loop");
+        if session_refreshed_since(claude_dir, snapshot) {
+            tracing::info!("claude session refreshed — entering turn loop");
             *state.lock().unwrap() = LoginState::Online;
             bus.emit_status("online");
             return;
@@ -493,6 +505,46 @@ pub async fn wait_for_login(
     }
 }
 
+/// Newest `mtime` across all regular files in `dir`. `None` when the
+/// dir is empty / unreadable. Used by `wait_for_login` to snapshot
+/// the credentials state at entry so a re-auth that overwrites the
+/// existing credentials file is detected (closes #542).
+fn newest_file_mtime(dir: &Path) -> Option<std::time::SystemTime> {
+    let entries = std::fs::read_dir(dir).ok()?;
+    let mut newest: Option<std::time::SystemTime> = None;
+    for entry in entries.flatten() {
+        if !entry.file_type().is_ok_and(|t| t.is_file()) {
+            continue;
+        }
+        let Ok(meta) = entry.metadata() else {
+            continue;
+        };
+        let Ok(mtime) = meta.modified() else {
+            continue;
+        };
+        if newest.is_none_or(|cur| mtime > cur) {
+            newest = Some(mtime);
+        }
+    }
+    newest
+}
+
+/// Has the credentials dir been written since `snapshot`? Used as the
+/// exit condition for `wait_for_login`:
+///
+/// - `snapshot == None` (entry dir was empty) → any file appearing
+///   means a fresh login landed.
+/// - `snapshot == Some(t)` (entry dir had stale creds) → resume only
+///   once a file with `mtime > t` is present, so a stale 401 doesn't
+///   immediately flip us back to `Online`.
+fn session_refreshed_since(dir: &Path, snapshot: Option<std::time::SystemTime>) -> bool {
+    match (snapshot, newest_file_mtime(dir)) {
+        (None, Some(_)) => true,
+        (Some(prev), Some(now)) => now > prev,
+        _ => false,
+    }
+}
+
 /// Spawn `claude` for one turn and pump `stream-json` stdout into the
 /// live event bus. Prompt goes over stdin (variadic
 /// `--allowedTools`/`--tools` would otherwise eat a trailing positional
@@ -737,3 +789,90 @@ async fn run_claude(prompt: &str, files: &TurnFiles, bus: &Bus) -> Result<(bool,
     }
     Ok((too_long, is_rate_limited, is_auth_failed))
 }
+
+#[cfg(test)]
+mod tests {
+    use std::fs;
+    use std::time::{Duration, SystemTime};
+
+    use super::{newest_file_mtime, session_refreshed_since};
+
+    #[test]
+    fn newest_file_mtime_empty_dir_is_none() {
+        let dir = tempfile::tempdir().unwrap();
+        assert!(newest_file_mtime(dir.path()).is_none());
+    }
+
+    #[test]
+    fn newest_file_mtime_missing_dir_is_none() {
+        // 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");
+        assert!(newest_file_mtime(&missing).is_none());
+    }
+
+    #[test]
+    fn newest_file_mtime_picks_latest_across_multiple_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 newest = newest_file_mtime(dir.path()).expect("must find a file");
+        let newer_meta = fs::metadata(&newer_path).unwrap().modified().unwrap();
+        assert_eq!(newest, newer_meta);
+    }
+
+    #[test]
+    fn session_refreshed_first_login_flips_on_any_file() {
+        // Empty-dir snapshot (None) → any file appearing means a fresh
+        // login landed. Pre-#542 semantics for first-time login.
+        let dir = tempfile::tempdir().unwrap();
+        assert!(!session_refreshed_since(dir.path(), None));
+        fs::write(dir.path().join("credentials.json"), b"{}").unwrap();
+        assert!(session_refreshed_since(dir.path(), None));
+    }
+
+    #[test]
+    fn session_refreshed_stale_creds_dont_flip_immediately() {
+        // The #542 repro: 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 = newest_file_mtime(dir.path());
+        assert!(snapshot.is_some());
+        // No change to the file → loop must NOT exit.
+        assert!(!session_refreshed_since(dir.path(), snapshot));
+    }
+
+    #[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 = newest_file_mtime(dir.path());
+        std::thread::sleep(Duration::from_millis(20));
+        fs::write(dir.path().join("credentials.json"), b"{\"v\":2}").unwrap();
+        assert!(session_refreshed_since(dir.path(), snapshot));
+    }
+
+    #[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 way_future = SystemTime::now() + Duration::from_secs(3600);
+        assert!(!session_refreshed_since(dir.path(), Some(way_future)));
+    }
+}