//! Durable per-agent power *intent* (`wanted: Up | Offline`) — the //! spec half of spec-vs-status desired-state reconciliation. //! `container_view` remains the observed *status*; the job queue's //! `Reconcile` nodes are the mechanism that converges the two. //! //! Stored as the `agent_power` table in the coordinator DB //! (`/var/lib/hyperhive/db/broker.sqlite`, one tiny row per agent) — //! same one-file-many-modules pattern as `approvals` / //! `scheduled_prompts`, each with its own connection. Intent persists //! across hive-c0re restarts; in-flight //! queue work deliberately does not. Setting `wanted` is never a //! queued node: operator/intent actions update the row synchronously //! at request time, then submit the DAG whose terminal `Reconcile` //! reads the fresh value — rapid toggles are last-writer-wins and the //! reconciles converge. Power toggles never commit to the meta repo. use std::path::Path; use std::sync::Mutex; use anyhow::{Context, Result}; use chrono::Utc; use rusqlite::{Connection, OptionalExtension, params}; const SCHEMA: &str = " CREATE TABLE IF NOT EXISTS agent_power ( agent TEXT PRIMARY KEY, wanted TEXT NOT NULL, updated_at INTEGER NOT NULL ); "; /// Per-agent power intent. #[derive(Debug, Clone, Copy, PartialEq, Eq, strum::IntoStaticStr, strum::EnumString)] #[strum(serialize_all = "snake_case")] pub enum Wanted { Up, Offline, } impl Wanted { /// Seed value from an observed running state (first boot after /// this store lands, or an agent spawned outside the normal path). pub fn from_running(running: bool) -> Self { if running { Wanted::Up } else { Wanted::Offline } } } /// What a `Reconcile` should do given intent + observation. Pure so /// the `{Up,Offline} × {up,down}` matrix is unit-testable. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum ReconcileAction { Start, Stop, Noop, } #[must_use] pub fn reconcile_action(wanted: Wanted, running: bool) -> ReconcileAction { match (wanted, running) { (Wanted::Up, false) => ReconcileAction::Start, (Wanted::Offline, true) => ReconcileAction::Stop, (Wanted::Up, true) | (Wanted::Offline, false) => ReconcileAction::Noop, } } /// Sqlite-backed store. `Arc`-friendly: all methods take `&self`, the /// internal `Mutex` serializes access. pub struct PowerStore { conn: Mutex, } impl PowerStore { /// Open (a connection to) the shared coordinator DB and ensure the /// `agent_power` table exists. `db_path` is the same sqlite file /// the broker / approvals stores open. pub fn open(db_path: &Path) -> Result { let conn = crate::db::open(db_path, "agent_power")?; conn.execute_batch(SCHEMA) .context("apply agent_power schema")?; Ok(Self { conn: Mutex::new(conn), }) } /// In-memory store for tests. #[cfg(test)] pub fn open_in_memory() -> Result { let conn = Connection::open_in_memory().context("open in-memory agent_power db")?; conn.execute_batch(SCHEMA) .context("apply agent_power schema")?; Ok(Self { conn: Mutex::new(conn), }) } /// Read an agent's intent. `None` when the agent has no row yet /// (callers seed from observed state via [`Self::get_or_seed`]). /// /// A row that exists but holds an unparseable value is an **error**, not /// `None`. The two used to collapse, and the damage was not that a caller /// read the wrong value: [`Self::get_or_seed`] takes `None` to mean "never /// seeded" and writes what the container is *currently doing*, so a /// corrupted row silently overwrote itself with the observed state this /// store exists to overrule. pub fn get(&self, agent: &str) -> Result> { let conn = self.conn.lock().expect("agent_power mutex poisoned"); let row: Option = conn .query_row( "SELECT wanted FROM agent_power WHERE agent = ?1", params![agent], |r| r.get(0), ) .optional() .context("select agent_power")?; match row { None => Ok(None), Some(stored) => stored.parse::().ok().map(Some).with_context(|| { format!("agent_power row for {agent} holds an unknown intent {stored:?}") }), } } /// Write an agent's intent (last-writer-wins, synchronous at /// request time). pub fn set(&self, agent: &str, wanted: Wanted) -> Result<()> { let conn = self.conn.lock().expect("agent_power mutex poisoned"); conn.execute( "INSERT INTO agent_power (agent, wanted, updated_at) VALUES (?1, ?2, ?3) ON CONFLICT(agent) DO UPDATE SET wanted = ?2, updated_at = ?3", params![agent, <&str>::from(wanted), Utc::now().timestamp()], ) .context("upsert agent_power")?; Ok(()) } /// Read an agent's intent, seeding the row from the observed /// running state when absent — the migration rule for agents that /// predate this store (running ⇒ `Up`, stopped ⇒ `Offline`), after /// which the DB is authoritative. pub fn get_or_seed(&self, agent: &str, running: bool) -> Result { if let Some(w) = self.get(agent)? { return Ok(w); } let seeded = Wanted::from_running(running); self.set(agent, seeded)?; tracing::info!(%agent, wanted = <&str>::from(seeded), "agent_power: seeded from observed state"); Ok(seeded) } /// Drop an agent's row (container destroyed). pub fn remove(&self, agent: &str) -> Result<()> { let conn = self.conn.lock().expect("agent_power mutex poisoned"); conn.execute("DELETE FROM agent_power WHERE agent = ?1", params![agent]) .context("delete agent_power")?; Ok(()) } } #[cfg(test)] mod tests { use super::*; /// The full `{Up,Offline} × {up,down}` reconcile matrix: /// start / stop / noop / noop. #[test] fn reconcile_matrix() { assert_eq!(reconcile_action(Wanted::Up, false), ReconcileAction::Start); assert_eq!( reconcile_action(Wanted::Offline, true), ReconcileAction::Stop ); assert_eq!(reconcile_action(Wanted::Up, true), ReconcileAction::Noop); assert_eq!( reconcile_action(Wanted::Offline, false), ReconcileAction::Noop ); } /// Write a value `set` cannot produce, the way corruption would. fn write_raw(store: &PowerStore, agent: &str, wanted: &str) { let conn = store.conn.lock().expect("agent_power mutex poisoned"); conn.execute( "INSERT INTO agent_power (agent, wanted, updated_at) VALUES (?1, ?2, 0)", params![agent, wanted], ) .expect("insert raw row"); } /// A row that exists but does not parse is an error, not an absent one. /// The two rows beside it are the control: this must not pass by breaking /// `get` for everything. #[test] fn an_unparseable_row_is_an_error_not_an_absent_one() { let store = PowerStore::open_in_memory().expect("open"); store.set("valid", Wanted::Up).expect("set"); write_raw(&store, "corrupt", "sideways"); assert_eq!(store.get("valid").expect("valid reads"), Some(Wanted::Up)); assert_eq!(store.get("missing").expect("absent reads"), None); let err = store .get("corrupt") .expect_err("corrupt must not read as absent"); assert!( format!("{err:#}").contains("sideways"), "the error should name the offending value: {err:#}" ); } /// The damage the error arm exists to prevent: `get_or_seed` treats /// `None` as "never seeded" and writes observed state, so a corrupt row /// used to overwrite itself with whatever the container was doing. #[test] fn a_corrupt_row_is_not_reseeded_from_observed_state() { let store = PowerStore::open_in_memory().expect("open"); write_raw(&store, "corrupt", "sideways"); assert!(store.get_or_seed("corrupt", true).is_err()); let stored: String = store .conn .lock() .expect("agent_power mutex poisoned") .query_row( "SELECT wanted FROM agent_power WHERE agent = ?1", params!["corrupt"], |r| r.get(0), ) .expect("row still there"); assert_eq!(stored, "sideways", "the corrupt row must be left alone"); } #[test] fn get_set_roundtrip_and_seed() { let store = PowerStore::open_in_memory().expect("open"); assert_eq!(store.get("alice").expect("get"), None); // Seed from observed running state, once. assert_eq!(store.get_or_seed("alice", true).expect("seed"), Wanted::Up); // Thereafter the DB is authoritative — observed state no longer // overrides. assert_eq!( store.get_or_seed("alice", false).expect("seeded"), Wanted::Up ); store.set("alice", Wanted::Offline).expect("set"); assert_eq!(store.get("alice").expect("get"), Some(Wanted::Offline)); store.remove("alice").expect("remove"); assert_eq!(store.get("alice").expect("get"), None); } }