1197 lines
49 KiB
Rust
1197 lines
49 KiB
Rust
//! 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 hive_claude::TokenUsage;
|
|
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<String> {
|
|
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<String> {
|
|
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::<serde_json::Value>(&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<String>) {
|
|
// 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::<serde_json::Value>(&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<u64, String> {
|
|
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<u64, String> {
|
|
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<S: std::hash::BuildHasher>(
|
|
cursor: &std::collections::HashMap<u64, String, S>,
|
|
) {
|
|
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<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(())
|
|
}
|
|
|
|
fn recent(&self, limit: usize) -> rusqlite::Result<Vec<StoredEvent>> {
|
|
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<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,
|
|
}
|
|
|
|
/// 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()))
|
|
}
|
|
|
|
/// 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; 5] = ["low", "medium", "high", "xhigh", "max"];
|
|
|
|
/// 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_<KEY_UPPER>` 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>` — 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::<u64>()
|
|
&& 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::<u64>()
|
|
&& 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<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.
|
|
compact_pending: Arc<AtomicBool>,
|
|
/// 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, 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)),
|
|
session_reset_pending: Arc::new(AtomicBool::new(false)),
|
|
compact_pending: 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
|
|
}
|
|
|
|
/// 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.
|
|
pub fn request_compact(&self) {
|
|
self.compact_pending.store(true, Ordering::SeqCst);
|
|
}
|
|
|
|
/// Take + clear the compact one-shot. Returns true iff `drive_turn` should
|
|
/// compact at the end of this turn.
|
|
#[must_use]
|
|
pub fn take_compact(&self) -> bool {
|
|
self.compact_pending.swap(false, Ordering::SeqCst)
|
|
}
|
|
|
|
/// 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 [`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(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<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()
|
|
}
|
|
|
|
/// 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("<unnamed>")
|
|
.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<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, 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<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(¤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<BusEvent> {
|
|
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<StoredEvent> {
|
|
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<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()
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::{
|
|
BusEvent, DEFAULT_EFFORT, EFFORT_LEVELS, LiveEvent, StoredEvent, 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 ["lowest", "", "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");
|
|
}
|
|
}
|