refactor(#2442): split harness-state persistence out of events.rs
This commit is contained in:
parent
bf657177d3
commit
91607f3896
7 changed files with 349 additions and 328 deletions
|
|
@ -18,6 +18,12 @@ use rusqlite::{Connection, params};
|
||||||
use serde::{Deserialize, Serialize};
|
use serde::{Deserialize, Serialize};
|
||||||
use tokio::sync::broadcast;
|
use tokio::sync::broadcast;
|
||||||
|
|
||||||
|
use crate::harness_state::{
|
||||||
|
DEFAULT_EFFORT, DEFAULT_MODEL, configured_effort, configured_model, context_window_tokens,
|
||||||
|
load_effort, load_model, persist_effort, persist_model, read_harness_state,
|
||||||
|
write_harness_state,
|
||||||
|
};
|
||||||
|
|
||||||
const CHANNEL_CAPACITY: usize = 256;
|
const CHANNEL_CAPACITY: usize = 256;
|
||||||
/// Max `LiveEvent`s the `Bus` returns from `history()` and keeps in
|
/// Max `LiveEvent`s the `Bus` returns from `history()` and keeps in
|
||||||
/// sqlite. Older rows are vacuumed on a periodic sweep.
|
/// sqlite. Older rows are vacuumed on a periodic sweep.
|
||||||
|
|
@ -31,189 +37,6 @@ fn events_db_path() -> PathBuf {
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
/// 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);
|
|
||||||
}
|
|
||||||
|
|
||||||
const SCHEMA: &str = "
|
const SCHEMA: &str = "
|
||||||
CREATE TABLE IF NOT EXISTS events (
|
CREATE TABLE IF NOT EXISTS events (
|
||||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||||
|
|
@ -442,90 +265,6 @@ pub enum TurnState {
|
||||||
Compacting,
|
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 the harness nix modules 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)]
|
#[derive(Clone)]
|
||||||
pub struct Bus {
|
pub struct Bus {
|
||||||
tx: Arc<broadcast::Sender<BusEvent>>,
|
tx: Arc<broadcast::Sender<BusEvent>>,
|
||||||
|
|
@ -798,7 +537,7 @@ impl Bus {
|
||||||
/// claude launch (no mid-session swap). Persisted to the agent's
|
/// claude launch (no mid-session swap). Persisted to the agent's
|
||||||
/// state dir (`hyperhive-effort`) so the override survives harness
|
/// state dir (`hyperhive-effort`) so the override survives harness
|
||||||
/// restart and container rebuild (gone on `--purge`). Callers must
|
/// restart and container rebuild (gone on `--purge`). Callers must
|
||||||
/// pre-validate with [`is_valid_effort`].
|
/// pre-validate with [`crate::harness_state::is_valid_effort`].
|
||||||
///
|
///
|
||||||
/// # Panics
|
/// # Panics
|
||||||
///
|
///
|
||||||
|
|
@ -1116,38 +855,7 @@ impl Default for Bus {
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use super::{
|
use super::{BusEvent, LiveEvent, StoredEvent};
|
||||||
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]
|
#[test]
|
||||||
fn stored_event_serializes_ts_beside_kind() {
|
fn stored_event_serializes_ts_beside_kind() {
|
||||||
|
|
@ -1177,24 +885,4 @@ mod tests {
|
||||||
assert_eq!(v["ts"], 1_700_000_000_i64);
|
assert_eq!(v["ts"], 1_700_000_000_i64);
|
||||||
assert_eq!(v["kind"], "note");
|
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");
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -182,7 +182,7 @@ pub async fn run(socket: PathBuf) {
|
||||||
// forge's own read-state, so it doesn't reintroduce the
|
// forge's own read-state, so it doesn't reintroduce the
|
||||||
// read-before-comment coupling that the mark-read-on-delivery approach
|
// read-before-comment coupling that the mark-read-on-delivery approach
|
||||||
// suffered.
|
// suffered.
|
||||||
let mut delivered: HashMap<u64, String> = crate::events::read_forge_cursor();
|
let mut delivered: HashMap<u64, String> = crate::harness_state::read_forge_cursor();
|
||||||
if !delivered.is_empty() {
|
if !delivered.is_empty() {
|
||||||
info!(
|
info!(
|
||||||
entries = delivered.len(),
|
entries = delivered.len(),
|
||||||
|
|
@ -917,7 +917,7 @@ async fn poll_once(
|
||||||
// changed, so a rebuild/restart reloads it instead of re-delivering
|
// changed, so a rebuild/restart reloads it instead of re-delivering
|
||||||
// the whole unread backlog.
|
// the whole unread backlog.
|
||||||
if cursor_dirty {
|
if cursor_dirty {
|
||||||
crate::events::write_forge_cursor(delivered);
|
crate::harness_state::write_forge_cursor(delivered);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
332
hive-ag3nt/src/harness_state.rs
Normal file
332
hive-ag3nt/src/harness_state.rs
Normal file
|
|
@ -0,0 +1,332 @@
|
||||||
|
//! File-backed harness state, split out of [`crate::events`] (which owns
|
||||||
|
//! the live event bus + sqlite store). This module holds the runtime
|
||||||
|
//! model/effort selection, the consolidated `hyperhive-harness.json`
|
||||||
|
//! state file, and the `forge_notify` delivery-dedupe cursor — the
|
||||||
|
//! persisted knobs the harness reads/writes across turns, none of which
|
||||||
|
//! are about the live event stream.
|
||||||
|
|
||||||
|
use std::path::PathBuf;
|
||||||
|
|
||||||
|
/// 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,
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
pub(crate) 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())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
pub(crate) 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,
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
pub(crate) 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())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
pub(crate) 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);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
pub(crate) 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.
|
||||||
|
pub(crate) 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);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// 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 the harness nix modules 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
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::{DEFAULT_EFFORT, EFFORT_LEVELS, 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 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");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
@ -6,6 +6,7 @@
|
||||||
pub mod client;
|
pub mod client;
|
||||||
pub mod events;
|
pub mod events;
|
||||||
pub mod forge_notify;
|
pub mod forge_notify;
|
||||||
|
pub mod harness_state;
|
||||||
pub mod identity;
|
pub mod identity;
|
||||||
pub mod login;
|
pub mod login;
|
||||||
pub mod login_session;
|
pub mod login_session;
|
||||||
|
|
|
||||||
|
|
@ -245,7 +245,7 @@ pub fn stall_sleep_secs() -> u64 {
|
||||||
/// for `claude-sonnet-4-6`).
|
/// for `claude-sonnet-4-6`).
|
||||||
fn effective_context_window(bus: &Bus) -> u64 {
|
fn effective_context_window(bus: &Bus) -> u64 {
|
||||||
bus.api_context_window()
|
bus.api_context_window()
|
||||||
.unwrap_or_else(|| crate::events::context_window_tokens(&bus.model()))
|
.unwrap_or_else(|| crate::harness_state::context_window_tokens(&bus.model()))
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Resolve the auto-reset watermark. Priority order:
|
/// Resolve the auto-reset watermark. Priority order:
|
||||||
|
|
|
||||||
|
|
@ -130,7 +130,7 @@ pub(super) struct EffortForm {
|
||||||
|
|
||||||
/// Switch the claude effort level for future sessions. Operator-only
|
/// Switch the claude effort level for future sessions. Operator-only
|
||||||
/// (the dashboard picker POSTs here through the gateway). Validated
|
/// (the dashboard picker POSTs here through the gateway). Validated
|
||||||
/// server-side against [`crate::events::EFFORT_LEVELS`] — an out-of-set
|
/// server-side against [`crate::harness_state::EFFORT_LEVELS`] — an out-of-set
|
||||||
/// value is rejected rather than handed to `claude --effort`, since an
|
/// value is rejected rather than handed to `claude --effort`, since an
|
||||||
/// unknown level would fail every subsequent launch. Applies on the next
|
/// unknown level would fail every subsequent launch. Applies on the next
|
||||||
/// session start (no mid-session swap).
|
/// session start (no mid-session swap).
|
||||||
|
|
@ -139,12 +139,12 @@ pub(super) async fn post_set_effort(
|
||||||
Form(form): Form<EffortForm>,
|
Form(form): Form<EffortForm>,
|
||||||
) -> Response {
|
) -> Response {
|
||||||
let level = form.effort.trim();
|
let level = form.effort.trim();
|
||||||
if !crate::events::is_valid_effort(level) {
|
if !crate::harness_state::is_valid_effort(level) {
|
||||||
return error_response(
|
return error_response(
|
||||||
StatusCode::BAD_REQUEST,
|
StatusCode::BAD_REQUEST,
|
||||||
&format!(
|
&format!(
|
||||||
"effort: level must be one of {}",
|
"effort: level must be one of {}",
|
||||||
crate::events::EFFORT_LEVELS.join(", ")
|
crate::harness_state::EFFORT_LEVELS.join(", ")
|
||||||
),
|
),
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -62,7 +62,7 @@ pub(super) async fn api_state(State(state): State<AppState>) -> axum::Json<State
|
||||||
swarm_name: crate::identity::swarm_name(),
|
swarm_name: crate::identity::swarm_name(),
|
||||||
available_models: available_models(),
|
available_models: available_models(),
|
||||||
effort,
|
effort,
|
||||||
available_efforts: crate::events::EFFORT_LEVELS
|
available_efforts: crate::harness_state::EFFORT_LEVELS
|
||||||
.iter()
|
.iter()
|
||||||
.map(ToString::to_string)
|
.map(ToString::to_string)
|
||||||
.collect(),
|
.collect(),
|
||||||
|
|
@ -181,7 +181,7 @@ pub(super) struct StateSnapshot {
|
||||||
effort: String,
|
effort: String,
|
||||||
/// Selectable effort levels for the picker, ascending. Fixed set
|
/// Selectable effort levels for the picker, ascending. Fixed set
|
||||||
/// (`low`, `medium`, `high`, `xhigh`, `max`) — sourced from
|
/// (`low`, `medium`, `high`, `xhigh`, `max`) — sourced from
|
||||||
/// [`crate::events::EFFORT_LEVELS`], not operator-configurable like
|
/// [`crate::harness_state::EFFORT_LEVELS`], not operator-configurable like
|
||||||
/// `available_models`. The frontend renders one button per entry.
|
/// `available_models`. The frontend renders one button per entry.
|
||||||
available_efforts: Vec<String>,
|
available_efforts: Vec<String>,
|
||||||
}
|
}
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue