hyperhive/hive-ag3nt/src/turn.rs

//! Per-turn claude invocation. The spawn shape, arg-vector, stdin plumbing,
//! and stream-json pumping are shared across all roles (there is only one
//! role: agent).

use std::collections::VecDeque;
use std::path::{Path, PathBuf};
use std::process::Stdio;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;

use anyhow::{Result, bail};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::process::Command;

use crate::events::{Bus, LiveEvent};
use crate::login::LoginState;
use crate::mcp;

// Hive-enforced claude settings ship at `/etc/claude-code/managed-settings.json`
// (wired in `nix/templates/harness-base.nix` from the `prompts/claude-settings.json`
// asset). claude-code auto-discovers that managed path — precedence #1,
// read-only, un-overridable — so the harness no longer passes `--settings`.
// We turn off claude's in-session auto-compaction and its cross-session
// auto-memory because hyperhive owns those concerns (`/compact` on overflow,
// notes persistence under `/state`). Unknown keys are silently ignored by
// claude-code; if a key gets renamed we'll spot it because the
// corresponding behavior will start firing mid-turn again.

/// Regex-ish marker claude-code emits when context overflows. Same string
/// bitburner-agent watches for. Empirically reliable across claude-code
/// versions; if it ever changes, compaction won't fire and we'll see a
/// claude exit with a useful error in the live view.
const PROMPT_TOO_LONG_MARKER: &str = "Prompt is too long";

/// Substrings that indicate the Anthropic API is refusing the request due
/// to a rate limit, per-account usage cap, or exhausted credit balance.
/// Matched against both stdout and stderr; any hit returns
/// `TurnOutcome::RateLimited` so the serve loop can park + retry instead
/// of propagating a hard failure that looks identical to a crash.
const RATE_LIMIT_MARKERS: &[&str] = &[
    "rate_limit_error",
    "overloaded_error",
    "Credit balance is too low",
    "Usage limit reached",
    "Request rate limit exceeded",
];

/// Substrings that indicate the Anthropic API rejected the request as
/// unauthenticated — the OAuth session in `$HOME/.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. 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.
/// See [`docs/turn-loop.md::The loop`](../../docs/turn-loop.md) for the
/// re-auth resumption path.
const AUTH_FAIL_MARKERS: &[&str] = &[
    "\"error\":\"authentication_failed\"",
    "\"error_status\":401",
    "Failed to authenticate. API Error: 401",
];

/// Substring claude-code emits when `--resume <id>` is handed a session id
/// that doesn't exist in this cwd's project (stale persisted id, or a crash
/// before the first turn ever completed a `.jsonl`). On a hit we clear the
/// persisted id so the NEXT turn starts a fresh session and re-captures —
/// the agent self-heals instead of failing `--resume` forever.
const SESSION_NOT_FOUND_MARKER: &str = "No conversation found with session ID";

/// Name of the harness-owned file under `paths::harness_dir()` that holds
/// the claude session id to resume. Written after every turn with the id
/// claude reported on its stream (the id can change across resume turns in
/// some claude-code versions, so we always rewrite with the last-seen value).
const CLAUDE_SESSION_ID_FILE: &str = "claude-session-id";

/// 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
/// tune down for tight retry scenarios or up if they're hitting sustained
/// capacity limits.
const DEFAULT_RATE_LIMIT_SLEEP_SECS: u64 = 300;

/// 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
/// turn re-uploads the full transcript regardless of whether we resume or
/// start fresh. A fresh session with a small context is therefore equally
/// cheap but gives the model a clean slate. Default is 3600s (1h) matching
/// the subscription TTL; API (pay-per-token) users should set
/// `HIVE_CACHE_TTL_SECS=300`. Override via `HIVE_CACHE_TTL_SECS`; set to
/// `0` to disable (always resume).
const DEFAULT_CACHE_TTL_SECS: u64 = 3600;

/// Synthetic wake prompt for the proactive notes-checkpoint turn. Not an
/// inbox message — the harness injects it directly so the agent gets one
/// turn to persist durable state before `/compact` collapses the
/// turn-by-turn history into a summary.
const CHECKPOINT_PROMPT: &str = "[system] Context checkpoint — no inbox message to handle.\n\n\
Your conversation context has grown large and the harness is about to run `/compact`, \
which collapses the detailed turn-by-turn history into a short summary. Anything you \
do not persist now is effectively lost after the next turn.\n\n\
Use THIS turn to flush anything worth keeping into your durable `/state` files: update \
your notes / CLAUDE.md / TODO.md with in-flight task state, decisions made, important \
file paths, and whatever you would need to resume cleanly with only a summary of this \
conversation to go on. Do not start new work or reply to anyone — just write your notes \
and end the turn.";

/// The set of files claude reads on every invocation: the MCP server
/// config (`--mcp-config`) and the pre-rendered role/tools system
/// prompt (`--system-prompt-file`). Static settings are no longer
/// passed here — they live at `/etc/claude-code/managed-settings.json`
/// and claude auto-discovers them.
/// Materialised once at harness startup; shared between the turn loop
/// and the operator-driven `/compact` path so both invocations look
/// identical to claude (same MCP surface, same allowed tools, same
/// role prompt — only the stdin payload differs).
#[derive(Clone)]
pub struct TurnFiles {
    pub mcp_config: PathBuf,
    pub system_prompt: PathBuf,
}

impl TurnFiles {
    /// Write all three files into the per-agent runtime dir alongside
    /// `socket`. Idempotent — overwrites whatever was there.
    ///
    /// # Errors
    ///
    /// Returns an error if any of the config files cannot be written to disk.
    pub async fn prepare(socket: &Path, label: &str) -> Result<Self> {
        Ok(Self {
            mcp_config: write_mcp_config(socket).await?,
            system_prompt: write_system_prompt(socket, label).await?,
        })
    }
}

/// Drop the MCP config blob claude reads from `--mcp-config <path>`.
/// `socket` is the hyperhive per-container socket (forwarded to the child
/// as `--socket <path>`). The MCP subcommand is always `mcp` on the single
/// `hive` binary resolved from `/proc/self/exe`.
///
/// # Errors
///
/// Returns an error if the config file cannot be written.
pub async fn write_mcp_config(socket: &Path) -> Result<PathBuf> {
    let parent = crate::paths::config_dir();
    tokio::fs::create_dir_all(&parent).await.ok();
    let path = parent.join("claude-mcp-config.json");
    let exe = std::env::current_exe()
        .ok()
        .map_or_else(|| "hive".into(), |p| p.display().to_string());
    let body = mcp::render_claude_config(&exe, socket);
    tokio::fs::write(&path, body).await?;
    tracing::info!(path = %path.display(), "wrote claude MCP config");
    Ok(path)
}

/// Thin re-export of [`crate::prompt::write_system_prompt`] for
/// callers that already import this module. The actual rendering +
/// marker-block logic lives in `prompt.rs`; this is just the public
/// entry point the binaries call.
///
/// # Errors
///
/// Returns an error if the system prompt file cannot be written.
pub async fn write_system_prompt(socket: &Path, label: &str) -> Result<PathBuf> {
    crate::prompt::write_system_prompt(socket, label).await
}

/// One claude turn's outcome. The harness uses this to decide whether to
/// transparently kick off a compaction and retry.
#[derive(Debug)]
pub enum TurnOutcome {
    Ok,
    /// Turn completed and proactive context-size compaction fired afterwards.
    /// Treated like `Ok` for ack and failure-notification purposes; recorded
    /// as `result_kind = "compacted"` in turn stats so the stats page can
    /// distinguish normal turns from turns that triggered a compaction.
    Compacted,
    /// claude saw "Prompt is too long" — the session needs compacting.
    /// Run `compact_session()` then retry the same wake-up prompt.
    PromptTooLong,
    /// The Anthropic API refused the request due to a rate limit, per-account
    /// 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.
    AuthFailed,
    Failed(anyhow::Error),
}

/// 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::<u64>().ok())
        .filter(|&v| v > 0)
        .unwrap_or(DEFAULT_RATE_LIMIT_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`.
/// 2. `HIVE_CONTEXT_WINDOW_TOKENS_*` env vars (Nix-configured per-model defaults).
/// 3. Hard fallback: 200 000.
///
/// The API-reported window is the authoritative per-inference active
/// context limit. It reflects what the model actually enforces — which
/// for models with large prompt caches (e.g. 1 M total cache) may be
/// significantly smaller than the cache capacity (e.g. 200 k active window
/// for `claude-sonnet-4-6`).
fn effective_context_window(bus: &Bus) -> u64 {
    bus.api_context_window()
        .unwrap_or_else(|| crate::events::context_window_tokens(&bus.model()))
}

/// Resolve the auto-reset watermark. Priority order:
/// 1. `HIVE_AUTO_RESET_WATERMARK_TOKENS` env var (explicit override).
/// 2. 50% of `effective_context_window(bus)`.
///
/// `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::<u64>().ok())
    {
        return v;
    }
    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::<u64>().ok())
        .filter(|&v| v > 0)
        .unwrap_or(DEFAULT_CACHE_TTL_SECS)
}

/// Resolve the proactive-compaction watermark. Priority order:
/// 1. `HIVE_COMPACT_WATERMARK_TOKENS` env var (explicit override).
/// 2. 75% of `effective_context_window(bus)`.
///
/// `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::<u64>().ok())
    {
        return v;
    }
    effective_context_window(bus) * 3 / 4
}

/// Post-turn compaction behaviour for [`drive_turn`].
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CompactionMode {
    /// Normal turns: if the context crossed the watermark, run a dedicated
    /// notes-checkpoint turn so the agent can flush durable state, THEN
    /// `/compact`. The checkpoint turn is a separate wake.
    CheckpointThenCompact,
    /// Graceful-stop turn: the turn that just ran WAS the checkpoint (its
    /// prompt already told the agent to flush state), so compact directly
    /// if the watermark was crossed — no second wake. Keeps a cold-start
    /// resume cheap without waking the agent twice.
    CompactOnly,
}

/// Drive one turn end-to-end. Three paths layer on top of the raw `run_turn`:
///
/// - **Auto-reset (pre-turn)** — context is large AND the prompt cache has
///   gone cold (idle gap ≥ cache TTL). Resuming would re-upload the full
///   transcript uncached at the same cost as a fresh start. The harness runs
///   one checkpoint turn (agent flushes state), then arms a one-shot
///   `request_new_session` so the actual turn starts fresh.
/// - **Reactive (on overflow)** — `run_turn` returns `PromptTooLong`: the
///   session is already past the context window and *no* turn can run on it,
///   so we compact immediately and retry the same wake-up prompt once. No
///   notes-checkpoint turn is possible here — the detail is gone.
/// - **Proactive (post-turn)** — the turn finished cleanly but its context
///   size has crept past the watermark: while the session is still healthy we
///   give the agent one dedicated turn to checkpoint its `/state` notes, then
///   compact. This keeps a later turn from hitting the reactive path (where
///   there is no chance to save anything first). The graceful-stop path uses
///   [`CompactionMode::CompactOnly`] to skip the dedicated checkpoint turn.
///
/// Called once per turn by the `hive` serve loop (every agent role).
pub async fn drive_turn(prompt: &str, files: &TurnFiles, bus: &Bus) -> TurnOutcome {
    drive_turn_with(prompt, files, bus, CompactionMode::CheckpointThenCompact).await
}

/// As [`drive_turn`] but with explicit control over the post-turn
/// compaction behaviour — see [`CompactionMode`]. The graceful-stop path
/// passes [`CompactionMode::CompactOnly`] so the agent isn't woken a
/// second time for a checkpoint it already did.
pub async fn drive_turn_with(
    prompt: &str,
    files: &TurnFiles,
    bus: &Bus,
    compaction: CompactionMode,
) -> TurnOutcome {
    maybe_auto_reset(bus);
    let outcome = match run_turn(prompt, files, bus).await {
        TurnOutcome::PromptTooLong => {
            // Compact has its own three-flag surface (it's the same claude
            // binary). Treat any non-Ok outcome the same as if `run_turn`
            // had returned it — the serve loop already knows what to do
            // with each variant, no point re-wrapping. PromptTooLong from
            // /compact itself would be absurd recursion; bubble it up as
            // a normal failure path.
            match compact_session(files, bus).await {
                TurnOutcome::Ok | TurnOutcome::Compacted => run_turn(prompt, files, bus).await,
                other => return other,
            }
        }
        // Rate-limited: no point retrying immediately — bubble up so the
        // serve loop can park + emit status before the next attempt.
        TurnOutcome::RateLimited => return TurnOutcome::RateLimited,
        // Auth failed: may be a transient token-refresh race or brief API
        // hiccup. Retry once before bubbling up so the serve loop parks for
        // re-login. The retry outcome is passed through unchanged — if it
        // fails again the serve loop handles it as usual.
        TurnOutcome::AuthFailed => {
            bus.emit(LiveEvent::Note {
                text: "got 401 — retrying once before parking for re-login".into(),
            });
            run_turn(prompt, files, bus).await
        }
        other => other,
    };
    // Proactive: a turn just completed on a still-healthy session. If its
    // context crossed the watermark, compact before a later turn overflows
    // into the reactive path. In the normal mode this first runs a separate
    // notes-checkpoint turn so the agent can flush durable state; in
    // `CompactOnly` mode (graceful stop) the turn that just ran was itself
    // the checkpoint, so we compact directly without a second wake.
    // Best-effort — never changes the outcome of the turn that already
    // succeeded, but records it as `Compacted` so turn stats can distinguish
    // it from a plain `Ok`.
    if matches!(outcome, TurnOutcome::Ok) {
        let compacted = match compaction {
            CompactionMode::CheckpointThenCompact => maybe_checkpoint_and_compact(files, bus).await,
            CompactionMode::CompactOnly => maybe_compact(files, bus).await,
        };
        if compacted {
            return TurnOutcome::Compacted;
        }
    }
    outcome
}

/// Proactive post-turn compaction. If the last inference's context size
/// has crossed the watermark, run one notes-checkpoint turn so the agent
/// can persist durable state, then `/compact`. Best-effort: a failed
/// checkpoint or compaction is logged + surfaced as a Note but never
/// fails the turn that already succeeded. Returns `true` if compaction
/// was attempted (watermark crossed), `false` if skipped.
async fn maybe_checkpoint_and_compact(files: &TurnFiles, bus: &Bus) -> bool {
    let Some(used) = watermark_crossed(bus) else {
        return false;
    };
    let watermark = compact_watermark_tokens(bus);
    bus.emit(LiveEvent::Note {
        text: format!(
            "context at {used} tokens (watermark {watermark}) — running a \
             notes-checkpoint turn before /compact"
        ),
    });
    // Give the agent one turn to flush durable state into /state. If the
    // session is somehow already too far gone to run even this, fall
    // through to compaction anyway — the checkpoint is best-effort.
    match run_turn(CHECKPOINT_PROMPT, files, bus).await {
        TurnOutcome::Ok | TurnOutcome::Compacted => {}
        TurnOutcome::PromptTooLong => bus.emit(LiveEvent::Note {
            text: "checkpoint turn overflowed the window — compacting without it".into(),
        }),
        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"),
        }),
    }
    do_compact(files, bus).await;
    true
}

/// Compact-only proactive path: if the context crossed the watermark,
/// `/compact` directly — NO preceding checkpoint turn. Used by the
/// graceful-stop path, where the turn that just ran already flushed
/// durable state (its prompt said so), so a second checkpoint wake would
/// be redundant. Compacting before the container stops keeps a later
/// cold-start resume cheap rather than re-uploading a huge transcript.
/// Returns `true` if compaction was attempted.
async fn maybe_compact(files: &TurnFiles, bus: &Bus) -> bool {
    let Some(used) = watermark_crossed(bus) else {
        return false;
    };
    let watermark = compact_watermark_tokens(bus);
    bus.emit(LiveEvent::Note {
        text: format!(
            "context at {used} tokens (watermark {watermark}) — compacting before \
             graceful stop so a later cold start resumes cheap"
        ),
    });
    do_compact(files, bus).await;
    true
}

/// Returns `Some(used_tokens)` when proactive compaction is enabled AND the
/// last inference's context size has reached the watermark; `None` when
/// compaction is disabled (watermark 0), there's no usage reading yet, or
/// the context is still below the watermark.
fn watermark_crossed(bus: &Bus) -> Option<u64> {
    let watermark = compact_watermark_tokens(bus);
    if watermark == 0 {
        return None; // proactive compaction disabled
    }
    let used = bus.last_ctx_usage().map(|u| u.context_tokens())?;
    (used >= watermark).then_some(used)
}

/// Run `/compact`, surfacing each failure mode as a best-effort Note.
/// Never changes the outcome of the turn that already succeeded — the
/// next real turn surfaces any underlying issue (rate-limit / 401 / etc.)
/// through the normal path anyway.
async fn do_compact(files: &TurnFiles, bus: &Bus) {
    match compact_session(files, bus).await {
        TurnOutcome::Ok | TurnOutcome::Compacted => {}
        TurnOutcome::PromptTooLong => bus.emit(LiveEvent::Note {
            text: "/compact unexpectedly returned PromptTooLong — next turn will retry".into(),
        }),
        TurnOutcome::RateLimited => bus.emit(LiveEvent::Note {
            text: "/compact was rate-limited — next turn will park + retry".into(),
        }),
        TurnOutcome::AuthFailed => bus.emit(LiveEvent::Note {
            text: "/compact hit 401 — next turn will trigger the re-login flow".into(),
        }),
        TurnOutcome::Failed(e) => {
            tracing::warn!(error = %format!("{e:#}"), "post-checkpoint compact failed");
            bus.emit(LiveEvent::Note {
                text: format!("/compact after checkpoint failed: {e:#}"),
            });
        }
    }
}

/// Pre-turn auto-reset check. If context is large AND the prompt cache has
/// gone cold (idle time >= cache TTL), arm `request_new_session` so the
/// next wake-up turn starts fresh. No preceding checkpoint turn — running
/// any turn before the reset would re-upload and re-warm the cache, which
/// defeats the cost-optimisation purpose entirely.
fn maybe_auto_reset(bus: &Bus) {
    let watermark = auto_reset_watermark_tokens(bus);
    if watermark == 0 {
        return; // auto-reset disabled
    }
    let Some(ctx_tokens) = bus.last_ctx_usage().map(|u| u.context_tokens()) else {
        return; // no usage reading yet — first turn, nothing to reset
    };
    if ctx_tokens < watermark {
        return;
    }
    let last_ended = bus.last_turn_ended_unix();
    if last_ended == 0 {
        return; // no completed turn yet
    }
    // Compute idle seconds using the same clock as now_unix (unix epoch, i64).
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map_or(0, |d| d.as_secs());
    let idle_secs = now.saturating_sub(u64::try_from(last_ended).unwrap_or(0));
    let ttl = cache_ttl_secs();
    if idle_secs < ttl {
        return;
    }
    bus.emit(LiveEvent::Note {
        text: format!(
            "context {ctx_tokens} tokens, idle {idle_secs}s >= cache TTL {ttl}s \
             — dropping session (cache cold, fresh start is equally cheap)"
        ),
    });
    bus.request_new_session();
}

/// Emit the per-turn `TurnEnd` event + log line. Single owner so outcome
/// semantics stay consistent across every agent role.
pub fn emit_turn_end(bus: &Bus, outcome: &TurnOutcome) {
    match outcome {
        TurnOutcome::Ok | TurnOutcome::Compacted | TurnOutcome::PromptTooLong => {
            bus.emit(LiveEvent::TurnEnd {
                ok: true,
                note: None,
            });
            tracing::info!("turn finished");
        }
        TurnOutcome::RateLimited => {
            bus.emit(LiveEvent::TurnEnd {
                ok: false,
                note: Some("rate limited — parking until quota resets".into()),
            });
            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 {
                ok: false,
                note: Some(note.clone()),
            });
            tracing::warn!(error = %note, "turn failed");
        }
    }
}

/// 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<Mutex<LoginState>>,
    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<std::time::SystemTime>,
}

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

/// 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
/// prompt). The session is persistent across turns via `--resume <id>`
/// against the harness's own captured session id (NOT bare `--continue`,
/// which resumes the *latest* session in this cwd and so lets a `choom`
/// session hijack the live harness context). claude's in-session
/// auto-compact is disabled via the managed
/// settings at `/etc/claude-code/managed-settings.json` so it 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((true, _, _)) => TurnOutcome::PromptTooLong,
        Ok((_, true, _)) => TurnOutcome::RateLimited,
        Ok((_, _, true)) => TurnOutcome::AuthFailed,
        Ok(_) => TurnOutcome::Ok,
        Err(e) => TurnOutcome::Failed(e),
    }
}

/// Run claude's built-in `/compact` slash command on the persistent
/// session. Takes the *same* params as `run_turn` because compact
/// re-initialises claude with the full session shape — same MCP
/// surface, same system prompt, same allowed-tools — so the post-
/// compact state matches a normal turn's. Only the prompt over stdin
/// differs (`/compact` vs the wake-up payload).
///
/// Returns the same `TurnOutcome` shape as `run_turn` so callers can
/// react identically to all three failure flags (`prompt_too_long`,
/// `rate_limited`, `auth_failed`). The reactive caller bubbles any
/// non-Ok outcome up so the serve loop's normal handling (park +
/// retry on rate-limit, flip to `needs_login` on 401, etc.) kicks in;
/// the proactive post-checkpoint caller stays best-effort and only
/// emits a Note for each failure mode.
pub async fn compact_session(files: &TurnFiles, bus: &Bus) -> TurnOutcome {
    bus.emit(LiveEvent::Note {
        text: "context overflow — running /compact on the persistent session".into(),
    });
    let outcome = match run_claude("/compact", files, bus).await {
        Ok((true, _, _)) => TurnOutcome::PromptTooLong,
        Ok((_, true, _)) => TurnOutcome::RateLimited,
        Ok((_, _, true)) => TurnOutcome::AuthFailed,
        Ok(_) => TurnOutcome::Ok,
        Err(e) => TurnOutcome::Failed(e),
    };
    match &outcome {
        TurnOutcome::Ok | TurnOutcome::Compacted => bus.emit(LiveEvent::Note {
            text: "/compact done".into(),
        }),
        TurnOutcome::PromptTooLong => bus.emit(LiveEvent::Note {
            text: "/compact reported PromptTooLong — bubbling up".into(),
        }),
        TurnOutcome::RateLimited => bus.emit(LiveEvent::Note {
            text: "/compact was rate-limited — bubbling up".into(),
        }),
        TurnOutcome::AuthFailed => bus.emit(LiveEvent::Note {
            text: "/compact hit 401 — bubbling up for re-login flow".into(),
        }),
        TurnOutcome::Failed(e) => bus.emit(LiveEvent::Note {
            text: format!("/compact failed: {e:#}"),
        }),
    }
    outcome
}

#[allow(
    clippy::too_many_lines,
    reason = "one linear subprocess driver: spawn claude, stream + classify \
              stdout/stderr, then assemble the outcome; splitting it would \
              fragment the streaming state across helpers"
)]
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".
    const STDERR_TAIL_LINES: usize = 20;
    let model = bus.model();
    let effort = bus.effort();
    // Resolve which claude session to resume. We NEVER pass bare
    // `--continue`: that resumes the latest session in this cwd, which a
    // `choom` invocation (same cwd) can hijack, wiping the harness context.
    // Instead we `--resume <id>` against the id claude reported on a prior
    // turn, persisted under `harness_dir()/claude-session-id`.
    let persist_path = crate::paths::harness_dir().join(CLAUDE_SESSION_ID_FILE);
    let resume_id: Option<String> = if bus.take_skip_continue() {
        // Fresh session requested: mint a new one (no --resume). Flag it so
        // the bin loop mints a new `sessions` row + stamps its id onto this
        // turn's stats. Drop any stale persisted id — the new id claude
        // reports this turn is captured + written below.
        bus.mark_fresh_session();
        let _ = std::fs::remove_file(&persist_path);
        bus.emit(LiveEvent::Note {
            text: "fresh session (continue suppressed for this turn)".into(),
        });
        None
    } else {
        // Continue: resume OUR captured id. Absent (first turn / just
        // self-healed from a stale id) → fall through to a fresh session
        // and capture the new id below.
        match std::fs::read_to_string(&persist_path) {
            Ok(s) if !s.trim().is_empty() => Some(s.trim().to_string()),
            _ => None,
        }
    };
    let mut cmd = Command::new("claude");
    // Spawn inside the agent's state dir so relative paths in tool calls
    // (Read foo.md, Bash ls, Write notes.md) land in the durable dir
    // instead of wherever the harness systemd unit started. Falls back
    // silently if the dir is missing (dev / test without the bind mount).
    let state_dir = crate::paths::state_dir();
    if state_dir.is_dir() {
        cmd.current_dir(&state_dir);
    }
    cmd.arg("--print")
        .arg("--verbose")
        .arg("--output-format")
        .arg("stream-json")
        .arg("--model")
        .arg(&model)
        .arg("--effort")
        .arg(&effort);
    if let Some(id) = &resume_id {
        cmd.arg("--resume").arg(id);
    }
    cmd.arg("--system-prompt-file").arg(&files.system_prompt);
    cmd.arg("--mcp-config")
        .arg(&files.mcp_config)
        .arg("--strict-mcp-config")
        .arg("--tools")
        .arg(mcp::builtin_tools_arg())
        .arg("--allowedTools")
        .arg(mcp::allowed_tools_arg());
    // hyperhive.docs.enable wires HIVE_DOCS_DIR to the in-container
    // reference-docs tree. Expose it to claude as an additional
    // directory so the docs are readable; the agent is NOT pointed at
    // them (no CLAUDE.md autoload) — surfacing a pointer is the open
    // follow-up. Unset (docs disabled) → the flag is not passed.
    if let Some(docs_dir) = std::env::var_os("HIVE_DOCS_DIR")
        && !docs_dir.is_empty()
    {
        cmd.arg("--add-dir").arg(&docs_dir);
    }
    let mut child = cmd
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .spawn()?;

    if let Some(mut stdin) = child.stdin.take() {
        stdin.write_all(prompt.as_bytes()).await?;
        stdin.shutdown().await.ok();
        drop(stdin);
    }
    let stdout = child.stdout.take().expect("piped stdout");
    let stderr = child.stderr.take().expect("piped stderr");

    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));
    // `--resume` against a stale/missing id: clear the persist file so the
    // next turn self-heals into a fresh session.
    let session_not_found = Arc::new(AtomicBool::new(false));
    // Last `session_id` claude reported on its stream this turn; persisted
    // after the child exits so the next turn `--resume`s it.
    let session_id_seen = Arc::new(Mutex::new(None::<String>));
    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 notfound_out = session_not_found.clone();
    let notfound_err = session_not_found.clone();
    let session_id_out = session_id_seen.clone();
    let bus_out = bus.clone();
    let bus_err = bus.clone();
    let pump_stdout = tokio::spawn(async move {
        let mut reader = BufReader::new(stdout).lines();
        // Track usage as the turn unfolds. `last_inference` overwrites on
        // every assistant event so at result-time it holds the most recent
        // model call's usage — the actual context size. The `result` event
        // carries the cumulative-across-the-turn usage (cost signal). Both
        // get handed to `record_turn_usage` together so a single SSE
        // event updates both badges.
        let mut last_inference: Option<crate::events::TokenUsage> = None;
        // Resolved model id (API-echoed `message.model`) from this turn's
        // assistant events; recorded onto the bus at result-time so the
        // per-turn stats label the concrete version that ran, not the
        // requested `--model` alias.
        let mut last_model: Option<String> = None;
        while let Ok(Some(line)) = reader.next_line().await {
            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);
            }
            if line.contains(SESSION_NOT_FOUND_MARKER) {
                notfound_out.store(true, Ordering::Relaxed);
            }
            if let Ok(v) = serde_json::from_str::<serde_json::Value>(&line) {
                // Track the session id claude reports (init + result events
                // both carry it). Persisted after exit so the next turn
                // `--resume`s it; re-captured each turn since the id can
                // change across resumes in some claude-code versions.
                if let Some(sid) = v.get("session_id").and_then(|s| s.as_str())
                    && !sid.is_empty()
                {
                    *session_id_out.lock().unwrap() = Some(sid.to_string());
                }
                // Rate-limit detection: only fire on JSON `error` events,
                // not on arbitrary text content. An agent discussing a past
                // rate limit in its response would otherwise trigger a false
                // positive (the full conversation flows through stdout as
                // stream-json, so any text the model outputs is visible here).
                if v.get("type").and_then(|t| t.as_str()) == Some("error") {
                    let raw = v.to_string();
                    if RATE_LIMIT_MARKERS.iter().any(|m| raw.contains(m)) {
                        rate_out.store(true, Ordering::Relaxed);
                    }
                }
                if let Some(u) = crate::events::TokenUsage::from_assistant_event(&v) {
                    last_inference = Some(u);
                }
                if let Some(m) = crate::events::TokenUsage::model_from_assistant_event(&v) {
                    last_model = Some(m);
                }
                if let Some(cost) = crate::events::TokenUsage::from_stream_event(&v) {
                    // Fallback to `cost` if the turn somehow produced
                    // a result without any assistant event — keeps the
                    // ctx badge from going stale on a degenerate turn.
                    let ctx = last_inference.unwrap_or(cost);
                    bus_out.record_turn_usage(ctx, cost);
                    // Pin the resolved model for this turn's stats row
                    // (cleared to None if no assistant event reported one
                    // → stats sink falls back to the requested name).
                    bus_out.set_resolved_model(last_model.clone());
                }
                // Seed the API-reported context-window from the result
                // event's `modelUsage.*.contextWindow` field. This is
                // the authoritative per-inference active window used for
                // compaction watermarks — it reflects what the model
                // actually enforces, which may differ from the Nix
                // config (e.g. 200k active window on a 1M cache model).
                if let Some(w) = crate::events::TokenUsage::context_window_from_result_event(&v) {
                    bus_out.set_api_context_window(w);
                }
                bus_out.observe_stream(&v);
                bus_out.emit(LiveEvent::Stream(v));
            } else {
                // Non-JSON stdout: raw text check is fine here since these
                // are claude CLI messages, not conversation content.
                if RATE_LIMIT_MARKERS.iter().any(|m| line.contains(m)) {
                    rate_out.store(true, Ordering::Relaxed);
                }
                bus_out.emit(LiveEvent::Note {
                    text: format!("(non-json) {line}"),
                });
            }
        }
    });
    let stderr_tail: Arc<Mutex<VecDeque<String>>> =
        Arc::new(Mutex::new(VecDeque::with_capacity(STDERR_TAIL_LINES)));
    let tail_clone = stderr_tail.clone();
    let pump_stderr = tokio::spawn(async move {
        let mut reader = BufReader::new(stderr).lines();
        while let Ok(Some(line)) = reader.next_line().await {
            if line.contains(PROMPT_TOO_LONG_MARKER) {
                flag_err.store(true, Ordering::Relaxed);
            }
            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);
            }
            if line.contains(SESSION_NOT_FOUND_MARKER) {
                notfound_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`
            // surfaces when claude exits non-zero.
            tracing::warn!(line = %line, "claude stderr");
            bus_err.emit(LiveEvent::Note {
                text: format!("stderr: {line}"),
            });
            let mut t = tail_clone.lock().unwrap();
            if t.len() >= STDERR_TAIL_LINES {
                t.pop_front();
            }
            t.push_back(line);
        }
    });

    let status = child.wait().await?;
    let _ = pump_stdout.await;
    let _ = pump_stderr.await;
    let too_long = prompt_too_long.load(Ordering::Relaxed);
    let is_rate_limited = rate_limited.load(Ordering::Relaxed);
    let is_auth_failed = auth_failed.load(Ordering::Relaxed);
    // Session-id bookkeeping. On a stale/missing `--resume` id, drop the
    // persist file so the next turn starts fresh and self-heals. Otherwise
    // rewrite it with the id claude reported this turn (handles the id
    // changing across resumes in some claude-code versions).
    if session_not_found.load(Ordering::Relaxed) {
        let _ = std::fs::remove_file(&persist_path);
    } else if let Some(sid) = session_id_seen.lock().unwrap().clone() {
        if let Some(parent) = persist_path.parent() {
            let _ = std::fs::create_dir_all(parent);
        }
        let _ = std::fs::write(&persist_path, sid);
    }
    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)");
        }
        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, is_auth_failed))
}

#[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())));
    }
}