hyperhive/hive-ag3nt/src/turn.rs

//! Per-turn claude policy layer. The generic subprocess mechanics — spawning
//! `claude --print`, streaming + classifying stream-json, session
//! lookup/archive — live in the `hive-claude` crate. This module owns the
//! hyperhive-specific policy on top: building the per-turn config from the
//! bus, bridging the output stream onto the event bus (`BusSink`), and the
//! compaction / auto-reset / retry state machine (`drive_turn`).

use std::path::{Path, PathBuf};
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::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.
//
// The subprocess mechanics — spawning `claude --print`, streaming +
// classifying stream-json, session lookup/archive — live in the generic
// `hive-claude` crate. This module is the hyperhive *policy* layer on top:
// it builds the per-turn [`Config`] from the bus, forwards the stream to the
// event bus via [`BusSink`], and owns compaction / auto-reset / retry.

/// Fixed, harness-owned claude session title. Every turn / compact /
/// checkpoint resumes THIS title (`--resume <title>`); the create path
/// names it (`--name <title>`). One constant identity per agent means
/// compaction and the post-compact retry provably target the same session —
/// there is no scraped UUID to go stale, empty, or diverge. Each agent runs
/// in its own container (own `~/.claude` + own `/state` cwd), so even the
/// shared default never collides across agents. Override via
/// `HIVE_SESSION_TITLE`.
const DEFAULT_SESSION_TITLE: &str = "hive-session";

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

/// 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<u64> {
    std::env::var(name)
        .ok()
        .and_then(|s| s.trim().parse::<u64>().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 {
    env_u64_positive("HIVE_RATE_LIMIT_SLEEP_SECS", 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 {
    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 {
    env_u64_positive("HIVE_CACHE_TTL_SECS", 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 {
    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`:
///
/// - **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
///   archives the current session (so the next `--resume <title>` misses and
///   self-heals into a fresh `--name <title>`); no checkpoint turn runs here
///   because any turn before the archive would just re-warm the cache.
/// - **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 takes
///   the same proactive route — a checkpoint turn before `/compact` is cheap
///   insurance and keeps a later cold-start resume small.
///
/// Called once per turn by the `hive` serve loop (every agent role).
pub async fn drive_turn(prompt: &str, files: &TurnFiles, bus: &Bus) -> TurnOutcome {
    // Start this turn on a fresh session when either trigger fires. Both
    // archive the current session at this turn boundary (no claude is mid-
    // write — one claude per container, serialized by the serve loop) and
    // produce the same end state, so they're mutually exclusive: an explicit
    // operator reset makes the auto-reset heuristic moot for this turn.
    if bus.take_session_reset() {
        // Operator-requested (deferred from `POST /api/new-session`).
        bus.emit(LiveEvent::Note {
            text: "operator: resetting session — archiving before this turn".into(),
        });
        archive_session(bus);
    } else {
        // Heuristic: context large AND prompt cache gone cold.
        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, run a separate notes-checkpoint turn so
    // the agent can flush durable state, then compact before a later turn
    // overflows into the reactive path. 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) && maybe_checkpoint_and_compact(files, bus).await {
        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
}

/// 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), archive the current session so the
/// next wake-up turn's `--resume <title>` misses and self-heals into a fresh
/// `--name <title>` session. 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)"
        ),
    });
    archive_session(bus);
}

/// 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");
        }
    }
}

/// 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 <title>` (NOT bare `--continue`, which
/// resumes the *latest* session in this cwd and lets a `choom` session hijack
/// the live harness context — a constant title is immune since choom won't
/// carry it). 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 {
    let config = claude_config(bus, files);
    let sink = BusSink::new(bus);
    match Claude::run_resume_or_create(&config, &session_title(), prompt, &sink).await {
        Ok(created) => {
            if created {
                // Fresh session minted this turn → flag it so the bin loop
                // mints a `sessions` row + stamps its id onto this turn's stats.
                bus.mark_fresh_session();
                bus.emit(LiveEvent::Note {
                    text: format!("created fresh session titled \"{}\"", session_title()),
                });
            }
            TurnOutcome::Ok
        }
        Err(e) => error_to_turn(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(),
    });
    // Resume-only: never create a session just to compact it. If the titled
    // session doesn't exist there's genuinely nothing to compact — compacting
    // a freshly-minted empty session would just print "not enough messages"
    // (the historical version-B failure), so treat a session miss as a no-op Ok.
    let config = claude_config(bus, files);
    let sink = BusSink::new(bus);
    let session = Session::Resume(session_title());
    let outcome = match Claude::run(&config, &session, "/compact", &sink).await {
        Ok(()) => TurnOutcome::Ok,
        Err(hive_claude::Error::SessionNotFound) => {
            bus.emit(LiveEvent::Note {
                text: "no titled session to compact — skipping".into(),
            });
            TurnOutcome::Ok
        }
        Err(e) => error_to_turn(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
}

/// The constant session title for this agent. `HIVE_SESSION_TITLE` overrides
/// the compiled-in [`DEFAULT_SESSION_TITLE`]; each agent runs in its own
/// container (own `~/.claude` + own `/state` cwd), so even the shared default
/// never collides across agents.
#[must_use]
pub fn session_title() -> String {
    std::env::var("HIVE_SESSION_TITLE")
        .ok()
        .map(|s| s.trim().to_string())
        .filter(|s| !s.is_empty())
        .unwrap_or_else(|| DEFAULT_SESSION_TITLE.to_string())
}

/// The cwd claude is spawned in: the agent's durable `/state` dir when it
/// exists, else the harness process cwd. Claude derives its per-project
/// session dir from this path, so the same value feeds both the [`Config`] and
/// the [`hive_claude::SessionStore`].
fn session_cwd() -> PathBuf {
    let state_dir = crate::paths::state_dir();
    if state_dir.is_dir() {
        state_dir
    } else {
        std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."))
    }
}

/// The on-disk session store for this agent (claude home + spawn cwd), used to
/// locate + archive the harness session by title.
fn session_store() -> hive_claude::SessionStore {
    hive_claude::SessionStore::new(crate::paths::claude_dir(), session_cwd())
}

/// Build the per-turn `hive_claude::Config` from the bus (model / effort) and
/// the materialised `TurnFiles` (system prompt + MCP config), plus the fixed
/// tool allow-lists and the optional docs `--add-dir`.
fn claude_config(bus: &Bus, files: &TurnFiles) -> Config {
    let mut add_dirs = Vec::new();
    // hyperhive.docs.enable wires HIVE_DOCS_DIR to the in-container reference
    // docs; expose it as an additional readable directory when set.
    if let Some(docs_dir) = std::env::var_os("HIVE_DOCS_DIR")
        && !docs_dir.is_empty()
    {
        add_dirs.push(PathBuf::from(docs_dir));
    }
    let cwd = {
        let state_dir = crate::paths::state_dir();
        state_dir.is_dir().then_some(state_dir)
    };
    Config {
        model: bus.model(),
        effort: Some(bus.effort()),
        cwd,
        system_prompt_file: Some(files.system_prompt.clone()),
        mcp_config: Some(files.mcp_config.clone()),
        strict_mcp_config: true,
        tools: Some(mcp::builtin_tools_arg()),
        allowed_tools: Some(mcp::allowed_tools_arg()),
        add_dirs,
        ..Config::default()
    }
}

/// Map a `hive_claude::Error` onto the harness's `TurnOutcome`. The recognized
/// sentinels become their matching outcomes; a residual `SessionNotFound`
/// (create path itself missed — shouldn't happen) settles as `Ok`; genuine
/// failures become `Failed` (converting the typed lib error into `anyhow`).
fn error_to_turn(err: hive_claude::Error) -> TurnOutcome {
    use hive_claude::Error;
    match err {
        Error::PromptTooLong => TurnOutcome::PromptTooLong,
        Error::RateLimited => TurnOutcome::RateLimited,
        Error::AuthFailed => TurnOutcome::AuthFailed,
        Error::SessionNotFound => TurnOutcome::Ok,
        other => TurnOutcome::Failed(other.into()),
    }
}

/// Bridges a claude run's output stream onto the hyperhive event bus: parses
/// per-turn token usage / resolved model / context-window from stream-json,
/// mirrors every event to the SSE bus, and surfaces non-JSON stdout + stderr
/// as Notes. Interior mutability (a `Mutex`) tracks the last inference across
/// events; the driver calls the `Sink` methods synchronously from one reader
/// task, so contention is nil — the lock only satisfies the `&self` trait
/// signature (and keeps `BusSink: Sync` for the driver's `Send` future).
struct BusSink<'a> {
    bus: &'a Bus,
    state: Mutex<BusSinkState>,
}

#[derive(Default)]
struct BusSinkState {
    last_inference: Option<TokenUsage>,
    last_model: Option<String>,
}

impl<'a> BusSink<'a> {
    fn new(bus: &'a Bus) -> Self {
        Self {
            bus,
            state: Mutex::new(BusSinkState::default()),
        }
    }
}

impl Sink for BusSink<'_> {
    fn on_event(&self, event: &Value) {
        {
            let mut st = self.state.lock().unwrap();
            // `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
            // cost usage; both update the badges together.
            if let Some(u) = TokenUsage::from_assistant_event(event) {
                st.last_inference = Some(u);
            }
            if let Some(m) = TokenUsage::model_from_assistant_event(event) {
                st.last_model = Some(m);
            }
            if let Some(cost) = TokenUsage::from_stream_event(event) {
                let ctx = st.last_inference.unwrap_or(cost);
                self.bus.record_turn_usage(ctx, cost);
                self.bus.set_resolved_model(st.last_model.clone());
            }
        }
        // Seed the API-reported context-window from the result event's
        // `modelUsage.*.contextWindow` — the authoritative active window for
        // compaction watermarks.
        if let Some(w) = TokenUsage::context_window_from_result_event(event) {
            self.bus.set_api_context_window(w);
        }
        self.bus.observe_stream(event);
        self.bus.emit(LiveEvent::Stream(event.clone()));
    }

    fn on_stdout_line(&self, line: &str) {
        self.bus.emit(LiveEvent::Note {
            text: format!("(non-json) {line}"),
        });
    }

    fn on_stderr_line(&self, line: &str) {
        // Mirror to journald so post-mortems work without the web UI / events
        // sqlite; the bus Note is what the dashboard renders.
        tracing::warn!(line = %line, "claude stderr");
        self.bus.emit(LiveEvent::Note {
            text: format!("stderr: {line}"),
        });
    }
}

/// Archive (do NOT delete) the harness's own session so the next turn's
/// `--resume <title>` misses and self-heals into a fresh `--name <title>`
/// session. Delegates the rename to [`hive_claude::SessionStore::archive_by_title`]
/// (which touches only the file carrying OUR `customTitle`, leaving any `choom`
/// session sharing the cwd alone) and surfaces the result as a Note. Best-
/// effort: never fails a turn. Only ever called at a turn boundary (top of
/// `drive_turn` for an operator reset, or `maybe_auto_reset` pre-turn) so no
/// claude process holds the session file open when it's renamed.
fn archive_session(bus: &Bus) {
    let title = session_title();
    match session_store().archive_by_title(&title) {
        Ok(Some(path)) => {
            let name = path
                .file_name()
                .and_then(|n| n.to_str())
                .unwrap_or("?")
                .to_string();
            tracing::info!(path = %path.display(), "archived claude session");
            bus.emit(LiveEvent::Note {
                text: format!("archived session \"{title}\" ({name}) — next turn starts fresh"),
            });
        }
        Ok(None) => bus.emit(LiveEvent::Note {
            text: format!(
                "no existing session titled \"{title}\" to archive — next turn starts fresh"
            ),
        }),
        Err(e) => {
            tracing::warn!(error = %e, "failed to archive claude session");
            bus.emit(LiveEvent::Note {
                text: format!("failed to archive session \"{title}\": {e}"),
            });
        }
    }
}