//! 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 rusqlite::{Connection, params}; use serde::{Deserialize, Serialize}; use tokio::sync::broadcast; 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, ) } /// Path to the persisted model file. Overridable via `HYPERHIVE_MODEL_FILE` /// for dev / tests; otherwise derived from the agent's harness dir. fn model_file_path() -> PathBuf { std::env::var_os("HYPERHIVE_MODEL_FILE").map_or_else( || crate::paths::harness_dir().join("hyperhive-model"), PathBuf::from, ) } fn load_model() -> Option { let s = std::fs::read_to_string(model_file_path()).ok()?; let name = s.trim(); if name.is_empty() { None } else { Some(name.to_owned()) } } fn persist_model(name: &str) -> std::io::Result<()> { let path = model_file_path(); if let Some(parent) = path.parent() { let _ = std::fs::create_dir_all(parent); } std::fs::write(path, format!("{name}\n")) } /// Path to the persisted effort-level file. Sibling of `hyperhive-model`, /// overridable via `HYPERHIVE_EFFORT_FILE` for dev / tests; otherwise /// derived from the agent's harness dir. fn effort_file_path() -> PathBuf { std::env::var_os("HYPERHIVE_EFFORT_FILE").map_or_else( || crate::paths::harness_dir().join("hyperhive-effort"), PathBuf::from, ) } fn load_effort() -> Option { let s = std::fs::read_to_string(effort_file_path()).ok()?; let level = s.trim(); if level.is_empty() { None } else { Some(level.to_owned()) } } fn persist_effort(level: &str) -> std::io::Result<()> { let path = effort_file_path(); if let Some(parent) = path.parent() { let _ = std::fs::create_dir_all(parent); } std::fs::write(path, format!("{level}\n")) } // --------------------------------------------------------------------------- // Consolidated harness state file // --------------------------------------------------------------------------- // // `hyperhive-harness.json` replaces the two legacy boolean sentinel files // (`hyperhive-rate-limited`, `hyperhive-needs-login`) that grew organically // and had no shared schema. A single JSON file is self-documenting, atomic // to write, and cheaper for hive-c0re to read on each sweep (one fopen vs // two stat calls). See `docs/persistence.md::Harness state files`. // // Legacy sentinel files written by older harness builds are still honoured // by `read_harness_state` so in-place upgrades don't lose state (the new // harness re-normalises on first write). Old files are not deleted — they // expire naturally when the state dir is purged. `hive-c0re::container_view` // also checks the legacy paths as a fallback during the transition window. const HARNESS_JSON: &str = "hyperhive-harness.json"; fn harness_json_path() -> PathBuf { crate::paths::state_dir().join(HARNESS_JSON) } // Serialises the read-modify-write of `hyperhive-harness.json`. Two // harness tasks touch it in the same process — the turn loop (rate-limit // / needs-login / active-model) and the forge_notify poller (the // delivery-dedupe cursor) — writing disjoint fields, so each writer must // preserve the other's. The lock closes the lost-update window between a // writer's read and its rename. static HARNESS_JSON_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(()); /// Read the consolidated state file as a JSON object, or an empty object /// when it is missing / unparseable / not an object. fn read_harness_json() -> serde_json::Value { std::fs::read_to_string(harness_json_path()) .ok() .and_then(|raw| serde_json::from_str::(&raw).ok()) .filter(serde_json::Value::is_object) .unwrap_or_else(|| serde_json::json!({})) } /// Atomically overwrite the state file (`.tmp` + rename) so hive-c0re /// never reads a partial file. fn write_harness_json(v: &serde_json::Value) { let path = harness_json_path(); let tmp = path.with_extension("json.tmp"); if std::fs::write(&tmp, v.to_string()).is_ok() { let _ = std::fs::rename(&tmp, &path); } } fn read_harness_state() -> (bool, bool, Option) { // Try the new consolidated file first. if let Ok(raw) = std::fs::read_to_string(harness_json_path()) && let Ok(v) = serde_json::from_str::(&raw) { let rate_limited = v .get("rate_limited") .and_then(serde_json::Value::as_bool) .unwrap_or(false); let needs_login = v .get("needs_login") .and_then(serde_json::Value::as_bool) .unwrap_or(false); let active_model = v .get("active_model") .and_then(serde_json::Value::as_str) .filter(|s| !s.is_empty()) .map(str::to_owned); return (rate_limited, needs_login, active_model); } // Fall back to legacy sentinel files written by older harness builds. let state_dir = crate::paths::state_dir(); let rate_limited = state_dir.join("hyperhive-rate-limited").exists(); let needs_login = state_dir.join("hyperhive-needs-login").exists(); (rate_limited, needs_login, None) } /// Write the turn-loop's harness state fields via a read-modify-write so /// any other writer's fields (e.g. `forge_notify`'s `forge_cursor`) survive. /// Pass `active_model: Some(s)` to update the resolved model (surfaced in /// the dashboard badge); `None` leaves the stored value untouched. fn write_harness_state(rate_limited: bool, needs_login: bool, active_model: Option<&str>) { let _guard = HARNESS_JSON_LOCK .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); let mut v = read_harness_json(); v["rate_limited"] = rate_limited.into(); v["needs_login"] = needs_login.into(); if let Some(model) = active_model { v["active_model"] = model.into(); } write_harness_json(&v); } /// Parse the `forge_notify` delivery-dedupe cursor (notification thread id /// -> last-delivered `updated_at`) out of a harness-state JSON value. /// Empty when the field is absent (first boot) or malformed. fn forge_cursor_from_json(v: &serde_json::Value) -> std::collections::HashMap { v.get("forge_cursor") .cloned() .and_then(|c| serde_json::from_value(c).ok()) .unwrap_or_default() } /// Restore the `forge_notify` delivery-dedupe cursor from the consolidated /// state file so a container rebuild/restart doesn't re-deliver the whole /// currently-unread backlog. pub fn read_forge_cursor() -> std::collections::HashMap { forge_cursor_from_json(&read_harness_json()) } /// Persist the `forge_notify` delivery-dedupe cursor into the consolidated /// state file, read-modify-write under the shared lock so the turn-loop's /// own fields survive. Best-effort: a serialize failure is a no-op. pub fn write_forge_cursor( cursor: &std::collections::HashMap, ) { let Ok(value) = serde_json::to_value(cursor) else { return; }; let _guard = HARNESS_JSON_LOCK .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); let mut v = read_harness_json(); v["forge_cursor"] = value; write_harness_json(&v); } fn now_unix() -> i64 { std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .ok() .and_then(|d| i64::try_from(d.as_secs()).ok()) .unwrap_or(0) } 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 }, /// 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, } impl EventStore { fn open(path: &Path) -> rusqlite::Result { 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(()) } fn recent(&self, limit: usize) -> rusqlite::Result> { let (events, _, _) = self.page(None, limit)?; Ok(events) } /// 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, limit: usize, ) -> rusqlite::Result<(Vec, Option, 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::(&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::(&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)) } } /// Token usage emitted by claude in the final `result` stream-json event. /// All counts are in tokens. `None` fields mean the server didn't report them. #[derive(Debug, Clone, Copy, Default, Serialize, Deserialize, PartialEq, Eq)] pub struct TokenUsage { pub input_tokens: u64, pub output_tokens: u64, pub cache_read_input_tokens: u64, pub cache_creation_input_tokens: u64, } impl TokenUsage { /// Total context consumed this turn (input + cache reads + cache writes). /// This is the per-inference context footprint that counts against the /// model's `contextWindow` limit. Tracked from the last `assistant` event /// in the stream-json (per-inference usage, not the cumulative `result` /// event which sums across all inferences in a tool-heavy turn and can /// far exceed the per-inference window). #[must_use] pub fn context_tokens(&self) -> u64 { self.input_tokens + self.cache_read_input_tokens + self.cache_creation_input_tokens } /// Parse usage from the terminal `result` stream-json event. This is the /// **cumulative** sum across every inference in the turn — useful as a /// cost signal, but NOT the current context size (a tool-heavy turn /// sums per-call cached prompts and easily exceeds the model window). #[must_use] pub fn from_stream_event(v: &serde_json::Value) -> Option { if v.get("type").and_then(|t| t.as_str()) != Some("result") { return None; } Some(Self::from_usage_obj(v.get("usage")?)) } /// Parse usage from a per-inference `assistant` event's /// `.message.usage` block. Each turn fires one of these for every /// model call; tracking the LAST one over the turn gives the actual /// conversation context size — the number to watch for compaction. #[must_use] pub fn from_assistant_event(v: &serde_json::Value) -> Option { if v.get("type").and_then(|t| t.as_str()) != Some("assistant") { return None; } Some(Self::from_usage_obj(v.get("message")?.get("usage")?)) } fn from_usage_obj(u: &serde_json::Value) -> Self { let field = |k: &str| u.get(k).and_then(serde_json::Value::as_u64).unwrap_or(0); Self { input_tokens: field("input_tokens"), output_tokens: field("output_tokens"), cache_read_input_tokens: field("cache_read_input_tokens"), cache_creation_input_tokens: field("cache_creation_input_tokens"), } } /// Extract the per-inference context-window limit from a `result` /// stream-json event's `modelUsage` map. The API reports this as /// `modelUsage..contextWindow`; we take the first non-zero /// value across all model keys. /// /// Returns `None` if the event is not a `result` type or has no /// `contextWindow` field. The returned value is the authoritative /// per-inference active window (e.g. 200 000 for `claude-sonnet-4-6`). /// It may be smaller than the full prompt-cache capacity (which can /// be several million tokens via cache reads). #[must_use] pub fn context_window_from_result_event(v: &serde_json::Value) -> Option { if v.get("type").and_then(|t| t.as_str()) != Some("result") { return None; } let model_usage = v.get("modelUsage")?; let map = model_usage.as_object()?; for (_model, stats) in map { if let Some(w) = stats .get("contextWindow") .and_then(serde_json::Value::as_u64) && w > 0 { return Some(w); } } None } /// Extract the *resolved* model id from an `assistant` stream-json /// event (`message.model`). Unlike the requested `--model` name /// (which may be a short alias like `opus` or a default), the API /// echoes the concrete version it actually ran on (e.g. /// `claude-opus-4-8`). Recording the resolved id (not the requested /// name) is what lets the ST4TS model-mix + cost rollup label the /// exact version that ran. Returns `None` for non-assistant events /// or ones missing `message.model`. #[must_use] pub fn model_from_assistant_event(v: &serde_json::Value) -> Option { if v.get("type").and_then(|t| t.as_str()) != Some("assistant") { return None; } v.get("message") .and_then(|m| m.get("model")) .and_then(serde_json::Value::as_str) .filter(|s| !s.is_empty()) .map(ToOwned::to_owned) } } /// 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, } /// Compiled-in fallback model used when neither `HIVE_DEFAULT_MODEL` nor a /// persisted runtime override is present. pub const DEFAULT_MODEL: &str = "haiku"; /// Return the model declared in `HIVE_DEFAULT_MODEL` (set from /// `hyperhive.model` in `agent.nix`), or `None` if the env var is absent / /// empty. When `Some`, this takes precedence over any persisted runtime /// override so that nix config changes always take effect on rebuild. #[must_use] pub fn configured_model() -> Option<&'static str> { // Leak once at startup — acceptable for a single config value. std::env::var("HIVE_DEFAULT_MODEL") .ok() .filter(|s| !s.trim().is_empty()) .map(|s| &*Box::leak(s.into_boxed_str())) } /// Return the model to use when no config and no persisted override exist. #[must_use] pub fn default_model() -> &'static str { configured_model().unwrap_or(DEFAULT_MODEL) } /// Compiled-in fallback effort level — matches the `effortLevel` baked /// into `prompts/claude-settings.json`. pub const DEFAULT_EFFORT: &str = "medium"; /// Valid claude `--effort` levels, ascending. The operator picker is /// constrained to these; [`is_valid_effort`] guards the persist path. pub const EFFORT_LEVELS: [&str; 3] = ["medium", "high", "xhigh"]; /// True iff `level` is one of [`EFFORT_LEVELS`]. #[must_use] pub fn is_valid_effort(level: &str) -> bool { EFFORT_LEVELS.contains(&level) } /// Return the effort level declared in `HIVE_DEFAULT_EFFORT` (set from /// `hyperhive.effortLevel` in `agent.nix`), or `None` if absent / empty. /// Mirrors [`configured_model`]'s env shape. Unlike model, the persisted /// runtime override takes precedence over this baseline (see `Bus::new`): /// the operator's effort pick sticks across harness restart, with the nix /// value only the default when no override was ever set. #[must_use] pub fn configured_effort() -> Option<&'static str> { std::env::var("HIVE_DEFAULT_EFFORT") .ok() .filter(|s| !s.trim().is_empty()) .map(|s| &*Box::leak(s.into_boxed_str())) } /// Context-window size in tokens for a given model name. /// /// Canonical per-model sizes are declared in `harness-base.nix` as /// `hyperhive.contextWindowTokens` and injected as /// `HIVE_CONTEXT_WINDOW_TOKENS_` env vars — so this function /// normally just reads them. The Rust code carries no model knowledge; /// updating model families only requires a Nix change. /// /// Resolution order (first match wins): /// 1. `HIVE_CONTEXT_WINDOW_TOKENS_` — key (lowercased) is a /// substring of the active model name. Populated by the Nix default /// map for all known families; add/override in `agent.nix`. /// 2. `HIVE_CONTEXT_WINDOW_TOKENS` — single global override (any model). /// 3. Hard fallback: `200_000` (conservative; only hit outside NixOS). #[must_use] pub fn context_window_tokens(model: &str) -> u64 { let m = model.to_ascii_lowercase(); // Per-model env vars set by `hyperhive.contextWindowTokens` in Nix. for (key, val) in std::env::vars() { if let Some(suffix) = key.strip_prefix("HIVE_CONTEXT_WINDOW_TOKENS_") && !suffix.is_empty() && m.contains(&suffix.to_ascii_lowercase()) && let Ok(v) = val.trim().parse::() && v > 0 { return v; } } // Global override (single value, any model). if let Ok(s) = std::env::var("HIVE_CONTEXT_WINDOW_TOKENS") && let Ok(v) = s.trim().parse::() && v > 0 { return v; } // Hard fallback for dev/test outside NixOS where env vars aren't set. 200_000 } #[derive(Clone)] pub struct Bus { tx: Arc>, /// 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, /// 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>, /// Current turn-loop state + since-when (unix seconds). state: Arc>, /// Model name passed to `claude --model`. Default `haiku`; the /// operator can override at runtime via `POST /api/model`. model: Arc>, effort: Arc>, /// 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>>, /// 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>>, /// 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, /// One-shot: next `run_claude` call drops `--continue`, starting /// a fresh claude session. Set by `POST /api/new-session` from /// the per-agent web UI; consumed (cleared back to false) by the /// next turn. Subsequent turns resume normal `--continue` /// behavior. Atomic so the consumer can take-and-clear without a /// lock. skip_continue_once: Arc, /// 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>>, /// One-shot: `run_claude` flips this true when it suppresses /// `--continue` (a fresh session). The bin loop takes-and-clears it /// after the turn to decide whether to mint a new `sessions` row. fresh_session: Arc, /// 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>>, /// 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, /// 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>>, /// 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>>, } 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, now_unix()))), 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)), skip_continue_once: Arc::new(AtomicBool::new(false)), 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 } /// Arm the one-shot: the next claude invocation will run without /// `--continue`, dropping any prior session context. Idempotent /// — calling twice in a row before the next turn still consumes /// to a single fresh-start. pub fn request_new_session(&self) { self.skip_continue_once.store(true, Ordering::SeqCst); } /// Take + clear the one-shot. Returns true iff the caller should /// run claude without `--continue` for this turn. #[must_use] pub fn take_skip_continue(&self) -> bool { self.skip_continue_once.swap(false, Ordering::SeqCst) } /// Mark that the current turn started a fresh claude session. /// `run_claude` calls this when it suppresses `--continue`. 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 { *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) { *self.session_id.lock().unwrap() = id; } /// Currently-selected claude model name. Read on every turn so a /// `/model ` 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) { 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 [`is_valid_effort`]. /// /// # Panics /// /// Panics if the internal lock is poisoned. pub fn set_effort(&self, level: impl Into) { 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, cost: Option) { 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(now_unix(), 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) { *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 { 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 { *self.api_context_window.lock().unwrap() } /// 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. /// /// # 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("") .to_owned(); *counts.entry(name).or_insert(0) += 1; } } /// 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 { 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 { *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 { *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, now_unix()); 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>` 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) { 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(¤t_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: now_unix(), 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 { self.tx.subscribe() } /// Most recent events, oldest first, capped at `HISTORY_CAPACITY`. /// Drives the terminal pre-fill when the operator opens the agent /// page; without a store (db open failed) this is empty. #[must_use] pub fn history(&self) -> Vec { let Some(store) = &self.store else { return Vec::new(); }; store.recent(HISTORY_CAPACITY).unwrap_or_default() } /// 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, limit: usize, ) -> (Vec, Option, 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() } } #[cfg(test)] mod tests { use super::{ BusEvent, DEFAULT_EFFORT, EFFORT_LEVELS, LiveEvent, StoredEvent, TokenUsage, forge_cursor_from_json, is_valid_effort, }; use serde_json::json; #[test] fn forge_cursor_absent_field_is_empty() { // First boot / a state file that only carries the turn-loop fields: // no cursor yet, so we re-deliver the currently-unread set once. assert!(forge_cursor_from_json(&json!({ "rate_limited": false })).is_empty()); } #[test] fn forge_cursor_malformed_field_is_empty() { // A wrong-typed / corrupt cursor degrades to empty rather than // aborting the poller. assert!(forge_cursor_from_json(&json!({ "forge_cursor": "nonsense" })).is_empty()); } #[test] fn forge_cursor_roundtrips_u64_keys() { // serde_json stringifies integer map keys; confirm the u64 thread // ids the cursor is keyed on survive the JSON round-trip. let v = json!({ "forge_cursor": { "42": "2026-06-22T16:00:00Z", "99": "2026-06-22T17:30:00Z" } }); let cursor = forge_cursor_from_json(&v); assert_eq!(cursor.len(), 2); assert_eq!(cursor.get(&42), Some(&"2026-06-22T16:00:00Z".to_owned())); assert_eq!(cursor.get(&99), Some(&"2026-06-22T17:30:00Z".to_owned())); } #[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 effort_validation_accepts_only_known_levels() { for level in EFFORT_LEVELS { assert!(is_valid_effort(level), "{level} should be valid"); } // Out-of-set, empty, and wrong-case inputs are rejected so a bad // picker POST never reaches `claude --effort`. for bad in ["low", "", "MEDIUM", "ultra", "high "] { assert!(!is_valid_effort(bad), "{bad:?} should be rejected"); } } #[test] fn compiled_default_effort_is_selectable() { // The fallback must itself be a valid level, and match the value // baked into prompts/claude-settings.json. assert!(EFFORT_LEVELS.contains(&DEFAULT_EFFORT)); assert_eq!(DEFAULT_EFFORT, "medium"); } #[test] fn resolved_model_from_assistant_event() { let v = json!({ "type": "assistant", "message": { "model": "claude-opus-4-8", "role": "assistant" } }); assert_eq!( TokenUsage::model_from_assistant_event(&v), Some("claude-opus-4-8".to_owned()) ); } #[test] fn resolved_model_ignores_non_assistant_and_missing() { // Wrong event type. let result = json!({ "type": "result", "message": { "model": "claude-opus-4-8" } }); assert_eq!(TokenUsage::model_from_assistant_event(&result), None); // Assistant event missing message.model. let no_model = json!({ "type": "assistant", "message": { "role": "assistant" } }); assert_eq!(TokenUsage::model_from_assistant_event(&no_model), None); // Empty model string is treated as absent. let empty = json!({ "type": "assistant", "message": { "model": "" } }); assert_eq!(TokenUsage::model_from_assistant_event(&empty), None); } }