hyperhive/hive-agent/src/events.rs

//! Live event stream for the per-agent web UI. The harness emits one
//! `LiveEvent` per interesting thing that happens during a turn — wake-up
//! (the popped inbox message), every line claude prints on stdout
//! (parsed from `--output-format stream-json`), and the turn-end summary.
//! The web UI subscribes via SSE and renders rows live.
//!
//! Channel type is `tokio::sync::broadcast`. New subscribers see only
//! future events; the dashboard JS deals with the cold-start case by
//! showing "connecting…" until the first event arrives.

use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU64, Ordering};
use std::sync::{Arc, Mutex};

use chrono::Utc;
use hive_claude::TokenUsage;
use rusqlite::{Connection, params};
use serde::{Deserialize, Serialize};
use tokio::sync::broadcast;

use crate::harness_state::{
    DEFAULT_EFFORT, DEFAULT_MODEL, configured_effort, configured_model, context_window_tokens,
    load_effort, load_model, persist_effort, persist_model, read_harness_state,
    write_harness_state,
};

const CHANNEL_CAPACITY: usize = 256;
/// Max `LiveEvent`s the `Bus` returns from `history()` and keeps in
/// sqlite. Older rows are vacuumed on a periodic sweep.
pub const HISTORY_CAPACITY: usize = 2000;
/// Path to the persisted event db. Overridable via `HYPERHIVE_EVENTS_DB`
/// for dev / tests; otherwise derived from the agent's harness dir.
fn events_db_path() -> PathBuf {
    std::env::var_os("HYPERHIVE_EVENTS_DB").map_or_else(
        || crate::paths::harness_dir().join("hyperhive-events.sqlite"),
        PathBuf::from,
    )
}

const SCHEMA: &str = "
CREATE TABLE IF NOT EXISTS events (
    id           INTEGER PRIMARY KEY AUTOINCREMENT,
    ts           INTEGER NOT NULL,
    kind         TEXT    NOT NULL,
    payload_json TEXT    NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_events_ts ON events (ts);
";

/// Envelope carried over the broadcast channel: the `LiveEvent` itself
/// plus a monotonic per-process seq stamped by `Bus::emit`. SSE consumers
/// serialize this directly (seq becomes a sibling of the `kind` tag);
/// clients use seq to dedupe their buffered live traffic against the
/// snapshot/history responses (drop anything with `seq <= snapshot.seq`).
#[derive(Debug, Clone, Serialize)]
pub struct BusEvent {
    pub seq: u64,
    /// Unix seconds at emit time. Serialized as a sibling of the `kind`
    /// tag so the agent terminal can render turn start/end times (and
    /// turn duration) on the live stream; history rows carry the same
    /// `ts` field sourced from the persisted `events.ts` column, so the
    /// renderer reads `ts` identically for live + scrollback.
    pub ts: i64,
    #[serde(flatten)]
    pub event: LiveEvent,
}

/// A persisted event paired with its stored unix-seconds timestamp.
/// Serializes with `ts` as a sibling of the `kind` tag — same wire shape
/// as a live [`BusEvent`] minus `seq` — so the agent terminal reads `ts`
/// identically whether an event arrives live or is replayed from history.
#[derive(Debug, Clone, Serialize)]
pub struct StoredEvent {
    pub ts: i64,
    #[serde(flatten)]
    pub event: LiveEvent,
}

/// One row of the agent's live stream. Serialised to JSON for SSE delivery.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum LiveEvent {
    /// Harness popped a wake-up message and is about to invoke claude.
    /// `unread` is the count of *other* messages still in the inbox at
    /// that moment — surfaced as a badge in the live panel header.
    TurnStart {
        from: String,
        body: String,
        unread: u64,
    },
    /// One line of claude's `--output-format stream-json` stdout, parsed as
    /// a generic JSON value (so we don't have to track every claude-code
    /// event variant). The frontend pretty-prints by `type` field.
    Stream(serde_json::Value),
    /// Free-form note from the harness (e.g. "claude exited 0",
    /// "stream-json parse error: ..."). Useful when stream-json itself
    /// fails so the UI doesn't just go silent.
    ///
    /// Must be a struct variant (not `Note(String)`): internally-tagged
    /// enums can't flatten a tag onto a primitive newtype, and serde
    /// fails serialization at runtime — silently, because the SSE
    /// handler's `filter_map(... .ok()? ...)` swallows the error. From
    /// 2025-08 through 2026-05 every `Note` emission was a no-op + the
    /// sqlite history persisted them as the literal string `"null"`.
    /// The web UI's `note` renderer already reads `ev.text`, so the
    /// wire shape matches without a JS change.
    Note { text: String },
    /// Turn finished. `ok=false` means claude exited non-zero or the
    /// harness hit a transport error.
    TurnEnd { ok: bool, note: Option<String> },
    /// Harness reachability flipped: `"online"` /
    /// `"needs_login_idle"` / `"needs_login_in_progress"`. The web UI
    /// drives the alive badge from this so the operator sees a login
    /// land (or get revoked) without polling. Session detail
    /// (`url`/`output`/`finished`) is still served by `/api/state`
    /// during the short-lived in-progress window — the client
    /// re-fetches only while that flow is active.
    StatusChanged { status: String },
    /// `/api/model` switched the active claude model. The web UI
    /// updates the chip + the per-turn stats sink will key off this
    /// to mark the boundary in its log.
    ModelChanged { model: String },
    /// `/api/effort` switched the active claude effort level. Applies on
    /// the next session start; the web UI updates the picker chip to
    /// reflect it.
    EffortChanged { effort: String },
    /// Token usage for the turn just ended. Carries two snapshots:
    /// - `ctx` is the LAST inference's usage block (the actual context
    ///   window in use right now — what the operator needs to decide
    ///   whether to compact / reset).
    /// - `cost` is the cumulative usage across every inference in the
    ///   turn (sum of per-call billed tokens — the cost signal). For
    ///   tool-heavy turns the cumulative blows past the model's window
    ///   because each tool call's prompt is rebilled.
    TokenUsageChanged { ctx: TokenUsage, cost: TokenUsage },
    /// Harness's `TurnState` transitioned (idle / thinking /
    /// compacting). `since_unix` matches `Bus::state_snapshot().1`
    /// so the client's elapsed-time ticker keeps progressing across
    /// SSE reconnects without drift.
    TurnStateChanged { state: TurnState, since_unix: i64 },
}

/// sqlite-backed event log. Wraps a `Connection` behind a `Mutex` so the
/// `Bus` (which clones cheaply) shares one writer.
struct EventStore {
    conn: Mutex<Connection>,
}

impl EventStore {
    fn open(path: &Path) -> rusqlite::Result<Self> {
        if let Some(parent) = path.parent() {
            let _ = std::fs::create_dir_all(parent);
        }
        let conn = Connection::open(path)?;
        conn.execute_batch(SCHEMA)?;
        Ok(Self {
            conn: Mutex::new(conn),
        })
    }

    fn append(&self, event: &LiveEvent) -> rusqlite::Result<()> {
        let ts = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .ok()
            .and_then(|d| i64::try_from(d.as_secs()).ok())
            .unwrap_or(0);
        let kind = match event {
            LiveEvent::TurnStart { .. } => "turn_start",
            LiveEvent::Stream(_) => "stream",
            LiveEvent::Note { .. } => "note",
            LiveEvent::TurnEnd { .. } => "turn_end",
            LiveEvent::StatusChanged { .. } => "status_changed",
            LiveEvent::ModelChanged { .. } => "model_changed",
            LiveEvent::EffortChanged { .. } => "effort_changed",
            LiveEvent::TokenUsageChanged { .. } => "token_usage_changed",
            LiveEvent::TurnStateChanged { .. } => "turn_state_changed",
        };
        let payload = serde_json::to_string(event).unwrap_or_else(|_| "null".into());
        let conn = self.conn.lock().unwrap();
        conn.execute(
            "INSERT INTO events (ts, kind, payload_json) VALUES (?1, ?2, ?3)",
            params![ts, kind, payload],
        )?;
        Ok(())
    }

    /// Fetch up to `limit` events with id < `before_id` (or the most recent
    /// `limit` events when `before_id` is `None`). Returns
    /// `(events_oldest_first, min_row_id, has_more)`.
    fn page(
        &self,
        before_id: Option<i64>,
        limit: usize,
    ) -> rusqlite::Result<(Vec<StoredEvent>, Option<i64>, bool)> {
        let limit_i = i64::try_from(limit).unwrap_or(i64::MAX);
        let conn = self.conn.lock().unwrap();
        // Fetch one extra row so we can tell whether more exist.
        let fetch = limit_i.saturating_add(1);
        // `ts` is the persisted emit-time unix-seconds stamp; carried out
        // alongside each event so history replay shows the same turn
        // start/end times the live stream did.
        let rows: Vec<(i64, StoredEvent)> = if let Some(bid) = before_id {
            let mut stmt = conn.prepare(
                "SELECT id, ts, payload_json FROM events
                  WHERE id < ?1
                  ORDER BY id DESC
                  LIMIT ?2",
            )?;
            stmt.query_map(params![bid, fetch], |row| {
                let id: i64 = row.get(0)?;
                let ts: i64 = row.get(1)?;
                let s: String = row.get(2)?;
                Ok(serde_json::from_str::<LiveEvent>(&s)
                    .ok()
                    .map(|event| (id, StoredEvent { ts, event })))
            })?
            .flatten()
            .flatten()
            .collect()
        } else {
            let mut stmt = conn.prepare(
                "SELECT id, ts, payload_json FROM events
                  ORDER BY id DESC
                  LIMIT ?1",
            )?;
            stmt.query_map(params![fetch], |row| {
                let id: i64 = row.get(0)?;
                let ts: i64 = row.get(1)?;
                let s: String = row.get(2)?;
                Ok(serde_json::from_str::<LiveEvent>(&s)
                    .ok()
                    .map(|event| (id, StoredEvent { ts, event })))
            })?
            .flatten()
            .flatten()
            .collect()
        };
        let has_more = rows.len() > limit;
        let mut rows: Vec<(i64, StoredEvent)> = rows.into_iter().take(limit).collect();
        rows.reverse(); // oldest first
        let min_id = rows.first().map(|(id, _)| *id);
        let events = rows.into_iter().map(|(_, e)| e).collect();
        Ok((events, min_id, has_more))
    }
}

/// Authoritative turn-loop state. The harness owns it; the web UI
/// reads via `/api/state` and renders. Lives alongside the bus
/// because everyone who has a `Bus` already has the right handle to
/// poke the state on transitions.
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum TurnState {
    /// Inbox is empty / waiting on `Recv`.
    Idle,
    /// `claude --print` is running for a turn.
    Thinking,
    /// Operator-triggered `/compact` is running on the persistent
    /// session.
    Compacting,
}

/// One pending `/compact` request (see `Bus::request_compact`).
/// `wake_prompt` is what to drive as a synthetic follow-up turn once the
/// compaction actually finishes, if anything.
#[derive(Debug, Clone)]
pub struct CompactRequest {
    pub wake_prompt: Option<String>,
}

#[derive(Clone)]
pub struct Bus {
    tx: Arc<broadcast::Sender<BusEvent>>,
    /// Monotonic per-process counter stamped onto every `BusEvent`.
    /// Persisted nowhere — a harness restart resets seq to 0; clients
    /// always treat reconnect as "fresh state, fresh stream of seqs."
    /// Historical events served from sqlite carry no seq (they predate
    /// the live channel the seq is meant to dedupe against).
    event_seq: Arc<AtomicU64>,
    /// Persistent event log. `None` only if opening the sqlite db failed
    /// at construction — we keep going so the harness doesn't die on a
    /// missing state dir mount in dev / test scenarios.
    store: Option<Arc<EventStore>>,
    /// Current turn-loop state + since-when (unix seconds).
    state: Arc<Mutex<(TurnState, i64)>>,
    /// Model name passed to `claude --model`. Default `haiku`; the
    /// operator can override at runtime via `POST /api/model`.
    model: Arc<Mutex<String>>,
    effort: Arc<Mutex<String>>,
    /// Last-inference token usage from the most recent turn's final
    /// `assistant` event. Represents the actual context window size at
    /// turn-end — the number the operator watches to decide whether to
    /// compact. `None` until the first turn completes.
    last_ctx_usage: Arc<Mutex<Option<TokenUsage>>>,
    /// Cumulative token usage from the most recent turn's `result`
    /// event (sum across every inference in the turn). This is the cost
    /// signal — tool-heavy turns rebill the cached prompt per call and
    /// blow past the model window. `None` until the first turn completes.
    last_cost_usage: Arc<Mutex<Option<TokenUsage>>>,
    /// True while the harness is parked after a rate-limit response.
    /// Set by `emit_status("rate_limited")`, cleared by
    /// `emit_status("online")`. Also mirrored to a sentinel file at
    /// `{state_dir}/hyperhive-rate-limited` so the host-side
    /// `container_view` can surface the status on the dashboard without
    /// a live socket call.
    rate_limited: Arc<AtomicBool>,
    /// One-shot: operator asked to reset the claude session (`POST
    /// /api/new-session`). Consumed at the *next* turn boundary by
    /// `turn::drive_turn`, which archives the current session so the turn
    /// starts fresh. Deferred (not applied in the handler) so the archive
    /// never races a claude process mid-write — one claude per container,
    /// serialized by the serve loop, so the turn boundary is the only point
    /// where no session file is open.
    session_reset_pending: Arc<AtomicBool>,
    /// One-shot: run `/compact` after the next turn ends. Consumed at the end
    /// of the current/next turn by `turn::drive_turn`. Deferring to the turn
    /// boundary keeps compaction from racing a live claude process mid-turn.
    /// `Some(request)` when a compact is pending; `request.wake_prompt` is
    /// what to drive as a synthetic follow-up turn once the compaction
    /// actually completes (`None` = pending but no follow-up wake wanted,
    /// e.g. the operator dashboard's `/compact` button). `None` = no compact
    /// pending. Wrapped in [`CompactRequest`] rather than
    /// `Option<Option<String>>` (clippy pedantic's `option_option` lint,
    /// and the named field reads clearer at call sites than a bare nested
    /// `Option`) so "pending" and "what to wake with" can never desync.
    compact_pending: Arc<Mutex<Option<CompactRequest>>>,
    /// One-shot, written by `turn::drive_turn`/`turn::run_pending_compact`
    /// right after a compaction they served finishes, when that compact's
    /// request carried a `wake_prompt`. Read once by the `hive-agent` serve
    /// loop after either call site to decide whether to drive a synthetic
    /// follow-up turn. Separate from `compact_pending`: by the time this is
    /// set, the compact has already run and that flag has already been
    /// cleared by `take_compact`.
    post_compact_wake: Arc<Mutex<Option<String>>>,
    /// Current fresh-claude-session id (FK to `sessions.id`). Set by the
    /// bin loop after minting a session row on a fresh start; stamped onto
    /// every `turn_stats` row until the next fresh session. `None` before
    /// the first fresh turn or when the stats db is unavailable.
    session_id: Arc<Mutex<Option<i64>>>,
    /// One-shot: `run_claude` flips this true when it creates a fresh claude
    /// session (`--name <title>` on a title miss). The bin loop takes-and-
    /// clears it after the turn to decide whether to mint a new `sessions`
    /// row. Session identity itself is handled entirely in `turn.rs` via the
    /// constant title + archive — this flag is purely a stats signal.
    fresh_session: Arc<AtomicBool>,
    /// Per-turn tool-call counter. Reset by the bin loop between
    /// turns via `take_tool_calls`. Populated by `observe_stream` as
    /// the stdout pump parses each stream-json line. Powers the
    /// `tool_call_count` + `tool_call_breakdown_json` columns on the
    /// per-turn stats sink.
    tool_calls: Arc<Mutex<std::collections::HashMap<String, u64>>>,
    /// Unix timestamp of the most recent completed turn (set by
    /// `record_turn_usage`). Used by the auto-reset heuristic in
    /// `turn.rs` to compute how long the session has been idle and
    /// whether the prompt cache has gone cold. `0` = no turn yet.
    last_turn_ended_unix: Arc<AtomicI64>,
    /// Per-inference context-window size as reported by the Anthropic API
    /// in the stream-json `result` event (`modelUsage.*.contextWindow`).
    /// Set by the stdout pump on every completed turn. Takes precedence
    /// over the Nix-configured `HIVE_CONTEXT_WINDOW_TOKENS_*` env vars
    /// for compaction watermark calculations — it reflects the actual
    /// limit the model enforces, which may differ from what the operator
    /// configured (e.g. 200 k active window on a 1 M cache-enabled model).
    api_context_window: Arc<Mutex<Option<u64>>>,
    /// Resolved model id from the most recent turn's `assistant` events
    /// (the API-echoed `message.model`, e.g. `claude-opus-4-8`), as
    /// opposed to the requested `--model` name which may be a short
    /// alias. Set once per turn by the stdout pump at result-time;
    /// `None` until the first assistant event of a turn is seen. The
    /// per-turn stats sink prefers this over the requested name so the
    /// model-mix + cost rollup reflect the concrete version that ran.
    last_resolved_model: Arc<Mutex<Option<String>>>,
}

impl Bus {
    /// Open the events db (path from `events_db_path()`). On failure, fall back
    /// to a no-store bus — the harness still works, just without persistent history.
    #[must_use]
    pub fn new() -> Self {
        let path = events_db_path();
        let store = match EventStore::open(&path) {
            Ok(s) => Some(Arc::new(s)),
            Err(e) => {
                tracing::warn!(error = ?e, path = %path.display(), "events db open failed; running without history");
                None
            }
        };
        let (tx, _) = broadcast::channel(CHANNEL_CAPACITY);
        // Priority: HIVE_DEFAULT_MODEL (from hyperhive.model in agent.nix) >
        // persisted runtime override > compiled-in DEFAULT_MODEL.
        // The nix config always wins on rebuild; the persisted file is kept
        // for within-session tracking only (see persist_model / set_model).
        let initial_model = configured_model().map_or_else(
            || load_model().unwrap_or_else(|| DEFAULT_MODEL.to_owned()),
            str::to_owned,
        );
        // Effort precedence is the inverse of model: a persisted operator
        // pick wins over the nix `HIVE_DEFAULT_EFFORT` baseline so the
        // runtime choice survives harness restart (it only resets on
        // `--purge`). nix is the default for a never-touched agent.
        let initial_effort = load_effort()
            .or_else(|| configured_effort().map(str::to_owned))
            .unwrap_or_else(|| DEFAULT_EFFORT.to_owned());
        // Restore rate_limited (and needs_login) from the consolidated
        // harness state file so the dashboard shows the correct status
        // on cold load if the harness crashed while parked.
        let (was_rate_limited, was_needs_login, _) = read_harness_state();
        // Write the resolved active model to hyperhive-harness.json on
        // startup so hive-c0re can surface the model badge without reading
        // hyperhive-model directly. Written once here; updated on every
        // set_model() call (runtime override) and emit_status() call.
        write_harness_state(was_rate_limited, was_needs_login, Some(&initial_model));
        Self {
            tx: Arc::new(tx),
            event_seq: Arc::new(AtomicU64::new(0)),
            store,
            state: Arc::new(Mutex::new((TurnState::Idle, Utc::now().timestamp()))),
            model: Arc::new(Mutex::new(initial_model)),
            effort: Arc::new(Mutex::new(initial_effort)),
            last_ctx_usage: Arc::new(Mutex::new(None)),
            last_cost_usage: Arc::new(Mutex::new(None)),
            rate_limited: Arc::new(AtomicBool::new(was_rate_limited)),
            session_reset_pending: Arc::new(AtomicBool::new(false)),
            compact_pending: Arc::new(Mutex::new(None)),
            post_compact_wake: Arc::new(Mutex::new(None)),
            session_id: Arc::new(Mutex::new(None)),
            fresh_session: Arc::new(AtomicBool::new(false)),
            tool_calls: Arc::new(Mutex::new(std::collections::HashMap::new())),
            last_turn_ended_unix: Arc::new(AtomicI64::new(0)),
            api_context_window: Arc::new(Mutex::new(None)),
            last_resolved_model: Arc::new(Mutex::new(None)),
        }
    }

    /// Current high-water seq. Snapshot endpoints read this before
    /// gathering state so the resulting (snapshot.seq, snapshot) pair
    /// satisfies: any live event with seq > snapshot.seq is post-snapshot
    /// (not yet reflected). Clients dedupe buffered SSE traffic against
    /// this value.
    #[must_use]
    pub fn current_seq(&self) -> u64 {
        self.event_seq.load(Ordering::SeqCst)
    }

    fn next_seq(&self) -> u64 {
        self.event_seq.fetch_add(1, Ordering::SeqCst) + 1
    }

    /// Request a session reset (operator `POST /api/new-session`). Deferred:
    /// the flag is consumed at the next turn boundary by `drive_turn`, which
    /// archives the current session so no claude process is mid-write when the
    /// file is renamed. Idempotent — two clicks before the next turn still
    /// archive once.
    pub fn request_session_reset(&self) {
        self.session_reset_pending.store(true, Ordering::SeqCst);
    }

    /// Take + clear the session-reset one-shot. Returns true iff `drive_turn`
    /// should archive the current session before this turn.
    #[must_use]
    pub fn take_session_reset(&self) -> bool {
        self.session_reset_pending.swap(false, Ordering::SeqCst)
    }

    /// Request a compaction after the next turn ends (deferred to the turn
    /// boundary). Idempotent — a second request before the first is
    /// serviced just overwrites `wake_prompt` with the latest ask. `Some
    /// (wake_prompt)` schedules a synthetic follow-up turn (driven with
    /// `wake_prompt` as its body) once the compaction actually completes;
    /// `None` requests a plain compact with no follow-up wake (the operator
    /// dashboard's `/compact` button).
    pub fn request_compact(&self, wake_prompt: Option<String>) {
        *self.compact_pending.lock().unwrap() = Some(CompactRequest { wake_prompt });
    }

    /// Take + clear the compact one-shot. `Some(request)` means
    /// `drive_turn`/`run_pending_compact` should compact now —
    /// `request.wake_prompt` is what to pass to `set_post_compact_wake` once
    /// that compaction finishes. `None` means no compact is pending.
    #[must_use]
    pub fn take_compact(&self) -> Option<CompactRequest> {
        self.compact_pending.lock().unwrap().take()
    }

    /// Record that a just-finished compaction should drive a synthetic
    /// follow-up turn with `prompt` as its body. Called by
    /// `turn::drive_turn`/`turn::run_pending_compact` right after the
    /// compaction they served (whose `take_compact()` returned a request
    /// with `wake_prompt: Some(prompt)`) completes.
    pub fn set_post_compact_wake(&self, prompt: String) {
        *self.post_compact_wake.lock().unwrap() = Some(prompt);
    }

    /// Take + clear the post-compact wake one-shot. The serve loop calls
    /// this after either compact call site to decide whether to
    /// synthesize a follow-up turn.
    #[must_use]
    pub fn take_post_compact_wake(&self) -> Option<String> {
        self.post_compact_wake.lock().unwrap().take()
    }

    /// Mark that the current turn started a fresh claude session.
    /// `run_claude` calls this when it creates a new titled session.
    pub fn mark_fresh_session(&self) {
        self.fresh_session.store(true, Ordering::SeqCst);
    }

    /// Take + clear the fresh-session one-shot. The bin loop calls this
    /// after the turn to decide whether to mint a new `sessions` row.
    #[must_use]
    pub fn take_fresh_session(&self) -> bool {
        self.fresh_session.swap(false, Ordering::SeqCst)
    }

    /// Currently-active session id (FK to `sessions.id`), or `None`.
    ///
    /// # Panics
    ///
    /// Panics if the internal lock is poisoned.
    #[must_use]
    pub fn current_session_id(&self) -> Option<i64> {
        *self.session_id.lock().unwrap()
    }

    /// Set the active session id after minting a `sessions` row on a
    /// fresh start.
    ///
    /// # Panics
    ///
    /// Panics if the internal lock is poisoned.
    pub fn set_session_id(&self, id: Option<i64>) {
        *self.session_id.lock().unwrap() = id;
    }

    /// Currently-selected claude model name. Read on every turn so a
    /// `/model <name>` flip takes effect on the next turn.
    ///
    /// # Panics
    ///
    /// Panics if the internal lock is poisoned.
    #[must_use]
    pub fn model(&self) -> String {
        self.model.lock().unwrap().clone()
    }

    /// Switch the model for future turns. The current turn (if any)
    /// keeps the model it was already running. Persisted to the agent's
    /// state dir (`hyperhive-model`) so the override survives harness
    /// restart and container rebuild (gone on `--purge`, matching
    /// every other piece of agent state).
    ///
    /// # Panics
    ///
    /// Panics if the internal lock is poisoned.
    pub fn set_model(&self, name: impl Into<String>) {
        let value: String = name.into();
        self.model.lock().unwrap().clone_from(&value);
        if let Err(e) = persist_model(&value) {
            tracing::warn!(error = ?e, "model: persist failed");
        }
        // Mirror the resolved model into hyperhive-harness.json so
        // hive-c0re can surface it on the dashboard badge without reading
        // the hyperhive-model override file directly.
        let (rate_limited, needs_login, _) = read_harness_state();
        write_harness_state(rate_limited, needs_login, Some(&value));
        self.emit(LiveEvent::ModelChanged { model: value });
    }

    /// Currently-selected claude effort level. Read at session start to
    /// build the `--effort` arg.
    ///
    /// # Panics
    ///
    /// Panics if the internal lock is poisoned.
    #[must_use]
    pub fn effort(&self) -> String {
        self.effort.lock().unwrap().clone()
    }

    /// Switch the effort level for future sessions. Applies on the next
    /// claude launch (no mid-session swap). Persisted to the agent's
    /// state dir (`hyperhive-effort`) so the override survives harness
    /// restart and container rebuild (gone on `--purge`). Callers must
    /// pre-validate with [`crate::harness_state::is_valid_effort`].
    ///
    /// # Panics
    ///
    /// Panics if the internal lock is poisoned.
    pub fn set_effort(&self, level: impl Into<String>) {
        let value: String = level.into();
        self.effort.lock().unwrap().clone_from(&value);
        if let Err(e) = persist_effort(&value) {
            tracing::warn!(error = ?e, "effort: persist failed");
        }
        self.emit(LiveEvent::EffortChanged { effort: value });
    }

    /// Seed `last_ctx_usage` + `last_cost_usage` at startup without
    /// emitting a SSE event. Used by the bin entrypoints to backfill
    /// from the most recent `turn_stats` row so the per-agent web UI's
    /// ctx + cost badges paint real numbers on cold load.
    ///
    /// # Panics
    ///
    /// Panics if an internal lock is poisoned.
    pub fn seed_usage(&self, ctx: Option<TokenUsage>, cost: Option<TokenUsage>) {
        if ctx.is_some() {
            *self.last_ctx_usage.lock().unwrap() = ctx;
        }
        if cost.is_some() {
            *self.last_cost_usage.lock().unwrap() = cost;
        }
    }

    /// Record the just-ended turn's usage. `ctx` is the last inference's
    /// usage (current context size); `cost` is the cumulative across
    /// every inference in the turn (cost signal). One SSE event fires
    /// per turn carrying both.
    ///
    /// # Panics
    ///
    /// Panics if an internal lock is poisoned.
    pub fn record_turn_usage(&self, ctx: TokenUsage, cost: TokenUsage) {
        *self.last_ctx_usage.lock().unwrap() = Some(ctx);
        *self.last_cost_usage.lock().unwrap() = Some(cost);
        self.last_turn_ended_unix
            .store(Utc::now().timestamp(), Ordering::Relaxed);
        self.emit(LiveEvent::TokenUsageChanged { ctx, cost });
    }

    /// Unix timestamp of the most recent completed turn (`record_turn_usage`
    /// call), or `0` if no turn has finished yet.
    #[must_use]
    pub fn last_turn_ended_unix(&self) -> i64 {
        self.last_turn_ended_unix.load(Ordering::Relaxed)
    }

    /// Record the resolved model id observed for the just-ended turn
    /// (from `assistant` events' `message.model`). `None` clears it so a
    /// degenerate turn that produced no assistant event doesn't inherit a
    /// stale id — the stats sink then falls back to the requested name.
    ///
    /// # Panics
    ///
    /// Panics if the internal lock is poisoned.
    pub fn set_resolved_model(&self, model: Option<String>) {
        *self.last_resolved_model.lock().unwrap() = model;
    }

    /// Resolved model id from the most recent turn, if an `assistant`
    /// event reported one. The per-turn stats sink prefers this over the
    /// requested `--model` name.
    ///
    /// # Panics
    ///
    /// Panics if the internal lock is poisoned.
    #[must_use]
    pub fn last_resolved_model(&self) -> Option<String> {
        self.last_resolved_model.lock().unwrap().clone()
    }

    /// Update the API-reported context-window size from the stream-json
    /// `result` event's `modelUsage.*.contextWindow` field. Called by the
    /// stdout pump once per completed turn. `0` is ignored (sentinel for
    /// "not reported").
    pub fn set_api_context_window(&self, window: u64) {
        if window > 0 {
            *self.api_context_window.lock().unwrap() = Some(window);
        }
    }

    /// Return the API-reported per-inference context-window size, if the
    /// harness has seen at least one completed turn for this session.
    /// `None` until the first result event is processed.
    #[must_use]
    pub fn api_context_window(&self) -> Option<u64> {
        *self.api_context_window.lock().unwrap()
    }

    /// The effective context window for `model`: the API-reported window if a
    /// turn has completed (`api_context_window`), else the per-model default
    /// ([`context_window_tokens`]). Single accessor so the state + dashboard
    /// endpoints agree by construction.
    #[must_use]
    pub fn effective_context_window(&self, model: &str) -> u64 {
        self.api_context_window()
            .unwrap_or_else(|| context_window_tokens(model))
    }

    /// Walk a stream-json value for `tool_use` blocks and bump the
    /// per-turn counter for each one we find. Called by the stdout
    /// pump on every parsed line. Cheap when the line isn't an
    /// assistant message — the field-check short-circuits.
    ///
    /// A `Skill` invocation is one meta-tool (`name == "Skill"`) dispatching
    /// to whichever skill matched, so the bare tool name collapses every
    /// skill into one undifferentiated count. Special-cased (via
    /// [`breakdown_key`]) to key by `Skill:<skill>` instead.
    ///
    /// # Panics
    ///
    /// Panics if the internal lock is poisoned.
    pub fn observe_stream(&self, v: &serde_json::Value) {
        if v.get("type").and_then(|t| t.as_str()) != Some("assistant") {
            return;
        }
        let Some(content) = v
            .get("message")
            .and_then(|m| m.get("content"))
            .and_then(|c| c.as_array())
        else {
            return;
        };
        let mut counts = self.tool_calls.lock().unwrap();
        for block in content {
            if block.get("type").and_then(|t| t.as_str()) != Some("tool_use") {
                continue;
            }
            let name = block
                .get("name")
                .and_then(|n| n.as_str())
                .unwrap_or("<unnamed>");
            let key = breakdown_key(name, block.get("input"));
            *counts.entry(key).or_insert(0) += 1;
        }
    }

    /// Inspect the per-turn `system`/`init` stream event's `mcp_servers`
    /// array and surface a Note + `warn` when a configured MCP server
    /// failed to connect or is missing entirely. Pure observability: it
    /// never touches the session. It exists because the harness is
    /// otherwise blind to the init event — the only place claude reports
    /// MCP-server status — so a dropped/failed stdio bridge (e.g. matrix
    /// after a core bounce) went silent.
    ///
    /// `pending` is deliberately NOT treated as degraded: it's the claude
    /// CLI's normal init-event race for a freshly spawned stdio server
    /// (status is reported before the handshake completes, then flips to
    /// `connected` moments later in the same turn — see the upstream
    /// claude-agent-sdk-typescript repo, issue number 368, for confirmation).
    /// Flagging it fired on nearly every turn for all three configured
    /// servers with zero actual impact (confirmed: tools kept working),
    /// which is why the MCP-dropped-after-restart tracker issue's
    /// persistent-vs-one-turn repro question looked answered "persistent"
    /// when it wasn't — that was this false positive, not the real bug.
    pub(crate) fn observe_mcp_health(&self, v: &serde_json::Value) {
        if v.get("type").and_then(|t| t.as_str()) != Some("system")
            || v.get("subtype").and_then(|s| s.as_str()) != Some("init")
        {
            return;
        }
        // claude reports one entry per configured server: `{name, status}`.
        // Absent-from-array = claude dropped the server entirely; a present
        // entry with `status != "connected"` = it failed to register.
        let reported: std::collections::HashMap<&str, &str> = v
            .get("mcp_servers")
            .and_then(|m| m.as_array())
            .map(|arr| {
                arr.iter()
                    .filter_map(|s| {
                        let name = s.get("name").and_then(|n| n.as_str())?;
                        let status = s
                            .get("status")
                            .and_then(|st| st.as_str())
                            .unwrap_or("unknown");
                        Some((name, status))
                    })
                    .collect()
            })
            .unwrap_or_default();
        let degraded =
            degraded_mcp_servers(&crate::mcp_config::configured_server_names(), &reported);
        if degraded.is_empty() {
            return;
        }
        let list = degraded.join(", ");
        tracing::warn!(degraded = %list, "mcp servers not connected at turn start");
        self.emit(LiveEvent::Note {
            text: format!("⚠ MCP degraded at turn start: {list}"),
        });
    }

    /// Snapshot + clear the per-turn tool-call counter. The harness
    /// calls this between turns to fold the breakdown into a
    /// `turn_stats` row, then start the next turn with an empty map.
    ///
    /// # Panics
    ///
    /// Panics if the internal lock is poisoned.
    #[must_use]
    pub fn take_tool_calls(&self) -> std::collections::HashMap<String, u64> {
        std::mem::take(&mut *self.tool_calls.lock().unwrap())
    }

    /// Last context-size snapshot (last inference of the most recent
    /// turn), or `None` if no turn has completed yet.
    ///
    /// # Panics
    ///
    /// Panics if the internal lock is poisoned.
    #[must_use]
    pub fn last_ctx_usage(&self) -> Option<TokenUsage> {
        *self.last_ctx_usage.lock().unwrap()
    }

    /// Last cumulative cost snapshot (sum across the most recent turn's
    /// inferences), or `None` if no turn has completed yet.
    ///
    /// # Panics
    ///
    /// Panics if the internal lock is poisoned.
    #[must_use]
    pub fn last_cost_usage(&self) -> Option<TokenUsage> {
        *self.last_cost_usage.lock().unwrap()
    }

    /// Update the harness's authoritative turn-loop state. Records
    /// the transition time so `state_snapshot` can return a since-age.
    ///
    /// # Panics
    ///
    /// Panics if the internal lock is poisoned.
    pub fn set_state(&self, next: TurnState) {
        let since;
        {
            let mut guard = self.state.lock().unwrap();
            if guard.0 == next {
                return;
            }
            *guard = (next, Utc::now().timestamp());
            since = guard.1;
        }
        self.emit(LiveEvent::TurnStateChanged {
            state: next,
            since_unix: since,
        });
    }

    /// Broadcast a status flip (online / `needs_login_*` / `rate_limited`).
    /// Called by the bin entry points + `turn::wait_for_login` + the
    /// `post_login_*` handlers — every site that mutates the
    /// `Arc<Mutex<LoginState>>` should also call this so the web UI
    /// drops its periodic /api/state poll while a turn loop is running.
    ///
    /// `hyperhive-harness.json` persists across harness restarts so the
    /// host-side dashboard can render the status without a live socket call:
    /// - `"rate_limited"` sets `rate_limited: true` in the JSON.
    /// - `"needs_login_idle"` sets `needs_login: true` in the JSON.
    /// - `"online"` clears both fields — the agent is healthy again.
    /// - Other statuses clear `rate_limited` only; `needs_login` is sticky
    ///   until `"online"` (re-auth completed successfully).
    ///
    /// Writes are atomic (`.tmp` + `rename`) so hive-c0re never reads a
    /// partial file during its ~10s sweep.
    pub fn emit_status(&self, status: impl Into<String>) {
        let status = status.into();
        let new_rate_limited = status == "rate_limited";
        if new_rate_limited {
            self.rate_limited.store(true, Ordering::Relaxed);
        } else {
            self.rate_limited.store(false, Ordering::Relaxed);
        }
        // Read the current persisted needs_login so we don't flip it on
        // statuses that shouldn't touch it (e.g. `needs_login_in_progress`
        // is a transient mid-flow status; only `needs_login_idle` and
        // `online` should change the persistent flag).
        //
        // Known non-atomicity: this is a read-modify-write. Two concurrent
        // `emit_status` calls could clobber each other's `needs_login`
        // change if they raced between the read and the write. In practice
        // this is safe: the turn loop is sequential and the login flow
        // (`needs_login_idle` / `online`) only fires outside of active
        // turns, so the two callers never overlap. Documented rather than
        // locked because adding a Mutex here would be overkill for the
        // actual call pattern.
        let (_, current_needs_login, _) = read_harness_state();
        let new_needs_login = if status == "needs_login_idle" {
            true
        } else if status == "online" {
            false
        } else {
            current_needs_login
        };
        let current_model = self.model.lock().unwrap().clone();
        write_harness_state(new_rate_limited, new_needs_login, Some(&current_model));
        self.emit(LiveEvent::StatusChanged { status });
    }

    /// Returns true while the harness is parked after a rate-limit response.
    #[must_use]
    pub fn is_rate_limited(&self) -> bool {
        self.rate_limited.load(Ordering::Relaxed)
    }

    /// Current state + since-when (unix seconds). Snapshot copy, no lock held.
    ///
    /// # Panics
    ///
    /// Panics if the internal lock is poisoned.
    #[must_use]
    pub fn state_snapshot(&self) -> (TurnState, i64) {
        *self.state.lock().unwrap()
    }

    pub fn emit(&self, event: LiveEvent) {
        if let Some(store) = &self.store
            && let Err(e) = store.append(&event)
        {
            tracing::warn!(error = ?e, "events: append failed");
        }
        let envelope = BusEvent {
            seq: self.next_seq(),
            ts: Utc::now().timestamp(),
            event,
        };
        // Lagged subscribers drop events — fine; the UI is a tail, not a log.
        let _ = self.tx.send(envelope);
    }

    #[must_use]
    pub fn subscribe(&self) -> broadcast::Receiver<BusEvent> {
        self.tx.subscribe()
    }

    /// Paginated history: up to `limit` events before `before_id`
    /// (or the most recent `limit` when `before_id` is `None`).
    /// Returns `(events_oldest_first, min_row_id, has_more)`.
    /// `min_row_id` is the cursor for the next page; pass it as
    /// `before_id` on the next call.
    #[must_use]
    pub fn history_page(
        &self,
        before_id: Option<i64>,
        limit: usize,
    ) -> (Vec<StoredEvent>, Option<i64>, bool) {
        let Some(store) = &self.store else {
            return (Vec::new(), None, false);
        };
        store.page(before_id, limit).unwrap_or_default()
    }
}

impl Default for Bus {
    fn default() -> Self {
        Self::new()
    }
}

/// Compare the configured MCP server names against the `{name -> status}`
/// map parsed from the init event, returning display strings for the
/// degraded ones: a configured server absent from the report was dropped
/// entirely; one present with a non-`connected`, non-`pending` status
/// failed to register. `pending` is excluded — it's the CLI's normal
/// init-event race for a server still completing its handshake, not a
/// failure (see the doc comment on `observe_mcp_health`). Pure so it's
/// unit-testable without a live `Bus`.
fn degraded_mcp_servers(
    configured: &[String],
    reported: &std::collections::HashMap<&str, &str>,
) -> Vec<String> {
    configured
        .iter()
        .filter_map(|name| match reported.get(name.as_str()).copied() {
            Some("connected" | "pending") => None,
            Some(status) => Some(format!("{name} ({status})")),
            None => Some(format!("{name} (absent)")),
        })
        .collect()
}

/// The `tool_call_breakdown_json` key for one `tool_use` block: the bare
/// tool `name`, except a `Skill` invocation (one meta-tool dispatching to
/// whichever skill matched) is keyed `Skill:<skill>` using the invocation's
/// own `input.skill` field — the fully-qualified `plugin:skill-name` — so
/// distinct skills don't collapse into one undifferentiated `"Skill"`
/// count. Falls back to the bare `"Skill"` key if `input.skill` is ever
/// missing/non-string, so a schema change degrades safely instead of losing
/// the count. Pure so it's unit-testable without a live `Bus`.
fn breakdown_key(name: &str, input: Option<&serde_json::Value>) -> String {
    if name != "Skill" {
        return name.to_owned();
    }
    input
        .and_then(|i| i.get("skill"))
        .and_then(|s| s.as_str())
        .map_or_else(|| name.to_owned(), |skill| format!("Skill:{skill}"))
}

#[cfg(test)]
mod tests {
    use super::{BusEvent, LiveEvent, StoredEvent};

    #[test]
    fn stored_event_serializes_ts_beside_kind() {
        // History-row wire shape: `ts` is a flattened sibling of `kind`,
        // which is what the agent terminal reads to time turn boundaries.
        let v = serde_json::to_value(StoredEvent {
            ts: 1_700_000_000,
            event: LiveEvent::Note { text: "hi".into() },
        })
        .unwrap();
        assert_eq!(v["ts"], 1_700_000_000_i64);
        assert_eq!(v["kind"], "note");
        assert_eq!(v["text"], "hi");
    }

    #[test]
    fn bus_event_serializes_ts_and_seq_beside_kind() {
        // Live SSE frame: same `ts` sibling as history (plus `seq`), so the
        // renderer is path-agnostic between live + scrollback.
        let v = serde_json::to_value(BusEvent {
            seq: 7,
            ts: 1_700_000_000,
            event: LiveEvent::Note { text: "yo".into() },
        })
        .unwrap();
        assert_eq!(v["seq"], 7);
        assert_eq!(v["ts"], 1_700_000_000_i64);
        assert_eq!(v["kind"], "note");
    }

    #[test]
    fn degraded_mcp_servers_flags_absent_and_failed_only() {
        use std::collections::HashMap;
        let configured = [
            "hyperhive".to_owned(),
            "matrix".to_owned(),
            "bash".to_owned(),
        ];
        // hyperhive connected; matrix failed; bash absent (claude dropped it).
        let reported: HashMap<&str, &str> = [("hyperhive", "connected"), ("matrix", "failed")]
            .into_iter()
            .collect();
        let degraded = super::degraded_mcp_servers(&configured, &reported);
        assert_eq!(
            degraded,
            vec!["matrix (failed)".to_owned(), "bash (absent)".to_owned()]
        );
    }

    #[test]
    fn degraded_mcp_servers_empty_when_all_connected() {
        use std::collections::HashMap;
        let configured = ["hyperhive".to_owned(), "matrix".to_owned()];
        let reported: HashMap<&str, &str> = [("hyperhive", "connected"), ("matrix", "connected")]
            .into_iter()
            .collect();
        assert!(super::degraded_mcp_servers(&configured, &reported).is_empty());
    }

    #[test]
    fn breakdown_key_keys_skill_by_input_field() {
        let input = serde_json::json!({"skill": "base:async-task-hygiene"});
        assert_eq!(
            super::breakdown_key("Skill", Some(&input)),
            "Skill:base:async-task-hygiene"
        );
    }

    #[test]
    fn breakdown_key_non_skill_tool_is_bare_name() {
        assert_eq!(super::breakdown_key("Read", None), "Read");
    }

    #[test]
    fn breakdown_key_skill_missing_input_field_degrades_to_bare_name() {
        // Schema-drift fallback: an unexpected/missing `input.skill` still
        // counts the invocation, just undifferentiated, rather than losing
        // it entirely.
        assert_eq!(super::breakdown_key("Skill", None), "Skill");
        let input = serde_json::json!({"unexpected": "field"});
        assert_eq!(super::breakdown_key("Skill", Some(&input)), "Skill");
    }

    #[test]
    fn degraded_mcp_servers_ignores_pending() {
        // `pending` is the CLI's normal init-event race for a stdio server
        // still completing its handshake (flips to `connected` moments
        // later in the same turn) — not a real failure, per the
        // MCP-dropped-after-restart tracker issue.
        use std::collections::HashMap;
        let configured = ["hyperhive".to_owned(), "matrix".to_owned()];
        let reported: HashMap<&str, &str> = [("hyperhive", "pending"), ("matrix", "pending")]
            .into_iter()
            .collect();
        assert!(super::degraded_mcp_servers(&configured, &reported).is_empty());
    }
}