hyperhive/hive-c0re/src/stores/power.rs

//! 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)]
pub enum Wanted {
    Up,
    Offline,
}

impl Wanted {
    pub fn as_str(self) -> &'static str {
        match self {
            Wanted::Up => "up",
            Wanted::Offline => "offline",
        }
    }

    fn parse(s: &str) -> Option<Self> {
        match s {
            "up" => Some(Wanted::Up),
            "offline" => Some(Wanted::Offline),
            _ => None,
        }
    }

    /// 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<Connection>` serializes access.
pub struct PowerStore {
    conn: Mutex<Connection>,
}

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 / questions stores open.
    pub fn open(db_path: &Path) -> Result<Self> {
        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<Self> {
        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`]).
    pub fn get(&self, agent: &str) -> Result<Option<Wanted>> {
        let conn = self.conn.lock().expect("agent_power mutex poisoned");
        let row: Option<String> = conn
            .query_row(
                "SELECT wanted FROM agent_power WHERE agent = ?1",
                params![agent],
                |r| r.get(0),
            )
            .optional()
            .context("select agent_power")?;
        Ok(row.and_then(|s| Wanted::parse(&s)))
    }

    /// 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, wanted.as_str(), 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<Wanted> {
        if let Some(w) = self.get(agent)? {
            return Ok(w);
        }
        let seeded = Wanted::from_running(running);
        self.set(agent, seeded)?;
        tracing::info!(%agent, wanted = seeded.as_str(), "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
        );
    }

    #[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);
    }
}