hyperhive/hive-agent/src/swarm_permissions.rs

//! The permissions the swarm grants this agent.
//!
//! swarm-controller keeps the agent's set under its key in the
//! `agent-permissions` KV bucket. This task reads that key every
//! [`POLL_INTERVAL`] with the one subject the agent's own queue credential is
//! granted there, and [`current`] answers with the last value read. The turn
//! loop asks [`current`] for every turn's `--tools`, `--allowedTools` and MCP
//! config, so a change reaches the next turn.
//!
//! Until the first successful read after a start, the agent holds what
//! hive-c0re stored for it ([`HiveCopy::before_read`]), so a queue that is down
//! or slow leaves it on its last known grant, not the default. A read is
//! judged by [`HiveCopy::after_read`]: an unset key is the default, except
//! while the hive's import is pending. A failed read keeps the last value.

use std::path::Path;
use std::sync::{PoisonError, RwLock};
use std::time::Duration;

use hive_sh4re::permissions::Permission;
use swarm_queue_client::agent_permissions::{HIVE_COPY_PATH, HiveCopy, Read, read};

/// How often the key is read. Also the longest a change can take to reach
/// the agent's next turn.
const POLL_INTERVAL: Duration = Duration::from_secs(10);

/// The last value read; `None` until the first read succeeds.
static CURRENT: RwLock<Option<Vec<Permission>>> = RwLock::new(None);

/// The permissions this agent holds now.
#[must_use]
pub fn current() -> Vec<Permission> {
    let last = CURRENT
        .read()
        .unwrap_or_else(PoisonError::into_inner)
        .clone();
    current_with(last, Path::new(HIVE_COPY_PATH), &crate::identity::label())
}

/// [`current`] with the last read and the hive's copy given.
fn current_with(last: Option<Vec<Permission>>, copy: &Path, agent: &str) -> Vec<Permission> {
    last.unwrap_or_else(|| HiveCopy::read_at(copy).before_read(agent))
}

/// What one read of the key leaves the agent holding, judged against the
/// hive's copy at `copy`; `None` (keep the last value) when the read failed
/// or the bucket did not answer.
fn after(agent: &str, copy: &Path, read: Option<Read>) -> Option<Vec<Permission>> {
    match read? {
        Read::Current(record) => Some(HiveCopy::read_at(copy).after_read(agent, &record)),
        Read::NoBucket => None,
    }
}
/// Run the poller for the harness's lifetime, if this agent has a label to
/// name its key with.
pub async fn run() {
    let agent = crate::identity::label();
    if agent.is_empty() {
        tracing::warn!("this agent has no label; not reading its swarm permissions");
        return;
    }
    let client = crate::swarm_queue::client().await;
    tracing::info!(%agent, "reading swarm permissions from the swarm queue");
    loop {
        let read = read_once(&client, &agent).await;
        if let Some(perms) = after(&agent, Path::new(HIVE_COPY_PATH), read) {
            let mut current = CURRENT.write().unwrap_or_else(PoisonError::into_inner);
            if current.as_ref() != Some(&perms) {
                tracing::info!(?perms, "permissions: the swarm's set changed");
                *current = Some(perms);
            }
        }
        tokio::time::sleep(POLL_INTERVAL).await;
    }
}

/// The key's current value, or `None` when it could not be read.
async fn read_once(client: &async_nats::Client, agent: &str) -> Option<Read> {
    // An unconnected client does not fail a JetStream request, it hangs on it.
    swarm_queue_client::ensure_connected(client).ok()?;
    match read(client, agent).await {
        Ok(read) => Some(read),
        Err(e) => {
            tracing::warn!(error = %e, "permissions: reading this agent's key failed");
            None
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn record(json: &str) -> Read {
        Read::Current(serde_json::from_str(json).expect("decodes"))
    }

    /// A hive's copy holding `json`, in a tempdir kept alive by the guard.
    fn copy(json: &str) -> (tempfile::TempDir, std::path::PathBuf) {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("swarm-permissions.json");
        std::fs::write(&path, json).expect("write the copy");
        (dir, path)
    }

    #[test]
    fn a_read_sets_the_list_or_the_default_and_a_failure_keeps_the_last() {
        let none = Path::new("/nonexistent/swarm-permissions.json");
        assert_eq!(
            after(
                "atlas",
                none,
                Some(record(r#"{"permissions":["web_tools"]}"#))
            ),
            Some(vec![Permission::WebTools])
        );
        assert_eq!(
            after("atlas", none, Some(record(r#"{"permissions":[]}"#))),
            Some(Vec::new())
        );
        assert_eq!(
            after("atlas", none, Some(record("{}"))),
            Some(Permission::DEFAULT.to_vec())
        );
        assert_eq!(after("atlas", none, Some(Read::NoBucket)), None);
        assert_eq!(after("atlas", none, None), None);
    }

    #[test]
    fn before_the_first_read_the_stored_grant_applies() {
        let (_dir, path) = copy(r#"{"permissions":["web_tools"]}"#);
        assert_eq!(
            current_with(None, &path, "atlas"),
            vec![Permission::WebTools]
        );
        let (_dir, path) = copy(r#"{"permissions":[]}"#);
        assert_eq!(current_with(None, &path, "atlas"), Vec::new());
        assert_eq!(
            current_with(Some(vec![Permission::Meta]), &path, "atlas"),
            vec![Permission::Meta],
            "a read outranks the copy"
        );
    }

    #[test]
    fn while_the_import_is_pending_an_unset_key_keeps_the_stored_grant() {
        let (_dir, path) = copy(r#"{"permissions":["web_tools"],"import_pending":true}"#);
        assert_eq!(
            after("atlas", &path, Some(record("{}"))),
            Some(vec![Permission::WebTools])
        );
        assert_eq!(
            after("atlas", &path, Some(record(r#"{"permissions":["meta"]}"#))),
            Some(vec![Permission::Meta]),
            "a set key still wins"
        );
        let (_dir, path) = copy(r#"{"permissions":["web_tools"]}"#);
        assert_eq!(
            after("atlas", &path, Some(record("{}"))),
            Some(Permission::DEFAULT.to_vec()),
            "once imported, unset is the default"
        );
    }

    /// No read ever succeeds: the last value stays unset and the agent holds
    /// the hive's stored grant.
    #[test]
    fn with_the_queue_down_the_agent_holds_the_stored_grant() {
        let (_dir, path) = copy(r#"{"permissions":["messaging"],"import_pending":true}"#);
        let mut last = None;
        for read in [None, Some(Read::NoBucket), None] {
            if let Some(perms) = after("atlas", &path, read) {
                last = Some(perms);
            }
        }
        assert_eq!(
            current_with(last, &path, "atlas"),
            vec![Permission::Messaging]
        );
    }
}