vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

agent capabilities live in a swarm KV record, set from swarmctl

Browse source 
An agent's capabilities move from the hive's meta repo
(`meta/capabilities.json`, injected as `HIVE_CAPABILITIES`) into the
`capabilities` field of its key in a new `agent-permissions` KV bucket,
which swarm-controller alone writes.

- swarm-queue-client: `agent_permissions` (bucket, key, `AgentPermissions`,
  direct-get read, the import subject and its wire types).
- swarm-controller: creates the bucket at start; `GET|PUT|DELETE
  /api/agents/{name}/permissions` (deny_unknown_fields and unknown
  capability ⇒ 422, duplicate ⇒ 400, agent must be declared by a hive);
  answers each hive's one-shot import, writing a key only where there is
  none and only for agents that hive's wanted state declares.
- swarmctl: `agent permissions get|set|clear`.
- swarm-nats-auth: a hive reads every key of the bucket and sends its
  import on its own subject; the controller subscribes to the imports and
  may answer them (`responses`).
- hive-c0re: reads each of its agents' keys every 30s into
  /var/lib/hyperhive/swarm-agent-permissions.json and restarts an agent,
  gracefully, when its capabilities change. Unset is the compiled-in
  default (ruth: manage_root_agent; everyone else: none). A failed read
  keeps the stored value. On first boot it seeds that file from
  meta/capabilities.json, sends the grants to the controller, and deletes
  the file once the controller answers.

Deleted: the capabilities.json reader/writer, `HIVE_CAPABILITIES` in the
rendered meta flake, ruth's capability seed, the capabilities half of the
permissions rebuild job (`PermPayload::Capabilities`/`Combined`), the
dashboard's capabilities table and `capabilities_changed` event, and the
now-unused `Capability::description`.

Part of #5170.
This commit is contained in:
atlas 2026-10-15 00:24:41 +02:00 • committed by mara
commit f862354e43
47 changed files with 1943 additions and 1146 deletions
Download patch file Download diff file Expand all files Collapse all files

110
hive-c0re/src/workers/auto_update.rs
View file

@ -124,10 +124,6 @@ pub async fn ensure_root_agent(coord: &Arc<Coordinator>) -> Result<()> {
);
return Ok(());
}
// Before the create/start branch, not inside it: on a hive whose root
// container already exists this is the only run that can still seed the
// grant, and it has to land before her next rebuild bakes the binds.
seed_manager_capabilities();
let list_result = lifecycle::list().await;
if let Err(e) = &list_result {
tracing::warn!(
@ -257,68 +253,6 @@ fn seed_manager_tool_groups() {
}
}
/// The capability the root agent is seeded with, as the `snake_case` string
/// `capabilities.json` stores and `capabilities::has_cap` compares against.
/// Taken from the enum rather than written out, so the seed cannot drift into
/// a name `capabilities::prune_unknown` would silently drop.
fn manager_seed_caps() -> Vec<String> {
vec![<&str>::from(hive_sh4re::permissions::Capability::ManageRootAgent).to_owned()]
}
/// Whether the root agent's default capability grant should be written.
///
/// `store_written` is "`capabilities.json` exists" — see
/// [`seed_manager_capabilities`] for why that, and not the absence of an
/// entry for the manager, is the condition.
fn should_seed_manager_caps(store_written: bool) -> bool {
!store_written
}
/// Give ruth the `ManageRootAgent` capability on the path that deploys her.
///
/// This is the capability-store half of the grant `roles.json` used to make:
/// `topology::reconcile_roles` seeded `can_manage_top_level_agents` onto
/// `MANAGER_NAME` on every meta sync, and that role is what put the other
/// agents' state/config dirs, `/applied` and `/meta` into her nspawn binds.
/// Collapsing the role into the capability means that without a seed here she
/// loses those recovery mounts at her next container rebuild — silently, and
/// only then, because nspawn bakes bind flags at container start.
///
/// **Seeded once, not re-ensured on every boot.** The role kept an empty-list
/// tombstone so an explicit revoke stuck; the capability store *deletes* an
/// entry that has been emptied rather than tombstoning it, so "the manager
/// has no entry" cannot tell a fresh hive apart from a deliberate revoke.
/// File existence can: every grant and revoke goes through
/// `meta::commit_capabilities` → `capabilities::set_caps`, which
/// writes the file even when the result is an empty `{}`. So while the file
/// is absent nobody has ever had a say, and once it exists this is inert
/// forever — including on the destroy+recreate path, matching
/// [`seed_manager_tool_groups`]'s refusal to reset an operator's choice.
///
/// Written with plain `set_caps` rather than `meta::commit_capabilities`: the
/// role's own seed wrote the file and left it for `stage_generated_meta_files`
/// to `git add` into the next deploy commit, and taking `META_LOCK` on the
/// boot path would be a new ordering constraint for no gain.
fn seed_manager_capabilities() {
if !should_seed_manager_caps(crate::capabilities::capabilities_path().exists()) {
tracing::debug!(
"capabilities.json already written — leaving the manager's capabilities as-is"
);
return;
}
let caps = manager_seed_caps();
match crate::capabilities::set_caps(MANAGER_NAME, &caps) {
Ok(()) => tracing::info!(
caps = ?caps,
"seeded ruth's default capabilities (recovery bind mounts for every agent)"
),
Err(e) => tracing::warn!(
error = ?e,
"failed to seed ruth's capabilities — she will come up without the recovery mounts"
),
}
}
/// Boot reconcile (see the module doc): classify every agent by rev
/// freshness + persisted `wanted` intent, submit one `Boot` DAG
/// (hyperhive lock bump growing an in-DAG rebuild subgraph per stale
@ -572,10 +506,7 @@ fn submit_boot_tree(
#[cfg(test)]
mod tests {
use super::{
BootAction, MANAGER_NAME, RootAgentPlan, boot_action, manager_seed_caps, plan_root_agent,
should_seed_manager_caps,
};
use super::{BootAction, MANAGER_NAME, RootAgentPlan, boot_action, plan_root_agent};
use crate::lifecycle::AGENT_PREFIX;
use crate::power::Wanted;
@ -604,45 +535,6 @@ mod tests {
assert_eq!(plan_root_agent(result), RootAgentPlan::Present);
}
// -----------------------------------------------------------------------
// Root-agent capability seed. `capabilities_path()` resolves under
// `paths::meta_root()`, which is hardcoded to `/var/lib/hyperhive` with no
// test override, so — same as `capabilities.rs`'s own tests — these pin
// the pure decision (`should_seed_manager_caps`, `manager_seed_caps`)
// rather than round-tripping the real file. Together they are the
// replacement for the deleted `reconcile_roles_in_seeds_root_when_absent`
// and `reconcile_roles_in_does_not_reseed_after_explicit_revoke`.
#[test]
fn the_manager_is_seeded_while_the_capability_store_has_never_been_written() {
assert!(should_seed_manager_caps(false));
assert_eq!(manager_seed_caps(), vec!["manage_root_agent".to_owned()]);
}
/// The revoke-sticks half, and the reason the condition is the file rather
/// than the manager's entry: the store deletes an emptied entry instead of
/// keeping a tombstone, but a revoke still writes the file — so once it
/// exists the seed must never fire again.
#[test]
fn a_written_capability_store_is_never_reseeded() {
assert!(!should_seed_manager_caps(true));
}
/// A seed the store would drop as unrecognised grants nothing while
/// logging success, so pin that what we write is a name `prune_unknown`
/// keeps — i.e. one of `Capability::ALL`'s `snake_case` spellings.
#[test]
fn the_seeded_name_is_a_capability_the_store_recognises() {
for cap in manager_seed_caps() {
assert!(
hive_sh4re::permissions::Capability::ALL
.iter()
.any(|known| <&str>::from(*known) == cap),
"{cap} is not a known capability"
);
}
}
// The regression the `Unreadable` variant exists for: the arm used to
// substitute `from_running(running)`, which is `Noop` against BOTH
// observations — so no combination of inputs could reach a reconcile.

4
hive-c0re/src/workers/mod.rs
View file

@ -1,7 +1,7 @@
//! Background tasks and periodic sweeps: crash/login watcher, boot-time
//! auto-update reconcile, the agent-sockets.json writer loop, the MCP socket listener
//! reconcile loop, knowledge-repo sync, the swarm wanted-state pull, and the
//! swarm resource-cap read.
//! swarm resource-cap and permissions reads.
//! Each submodule is re-exported
//! at the crate root, so `crate::crash_watch::…` etc. keep working unchanged.
@ -10,5 +10,7 @@ pub mod auto_update;
pub mod crash_watch;
pub mod knowledge;
pub mod mcp_sockets;
pub mod permissions_import;
pub mod swarm_limits;
pub mod swarm_permissions;
pub mod wanted;

197
hive-c0re/src/workers/permissions_import.rs Normal file
View file

@ -0,0 +1,197 @@
//! The one-shot move of the capabilities this hive's meta repo held
//! (`meta/capabilities.json`) into the swarm's `agent-permissions` bucket.
//!
//! At boot, before any agent starts, [`seed_local`] copies the file's grants
//! into [`crate::capabilities`] when nothing is stored there yet, so agents
//! keep their mounts until the swarm answers. [`send_pending`] then sends the
//! grants to swarm-controller, which writes each agent's key only where there
//! is none, and on its answer deletes the file from the meta repo. Nothing
//! reads the file after that.
use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
use anyhow::{Context, Result};
use hive_sh4re::permissions::Capability;
use swarm_queue_client::agent_permissions::{
self, AgentPermissions, IMPORT_REPLY_DEADLINE, Import, ImportReply,
};
/// The meta repo file the grants are moved out of.
pub const META_FILE: &str = "capabilities.json";
fn meta_path() -> PathBuf {
crate::paths::meta_root().join(META_FILE)
}
/// The grants the meta file holds, by agent, as the swarm stores them. A name
/// that is not a [`Capability`] is dropped, with a warning.
///
/// `ruth` with no entry in a file that exists had her grant revoked: the file
/// was created holding it, and a revoke removed the entry. She is sent an
/// empty list, since unset would give her the default back.
fn grants(file: BTreeMap<String, Vec<String>>) -> BTreeMap<String, Vec<Capability>> {
let mut out: BTreeMap<String, Vec<Capability>> = file
.into_iter()
.map(|(agent, names)| {
let caps = names
.iter()
.filter_map(|name| {
let cap = Capability::ALL
.iter()
.copied()
.find(|c| name.eq_ignore_ascii_case(<&str>::from(*c)));
if cap.is_none() {
tracing::warn!(%agent, capability = %name, "import: dropping unrecognised capability");
}
cap
})
.collect();
(agent, caps)
})
.collect();
out.entry(crate::lifecycle::MANAGER_NAME.to_owned())
.or_default();
out
}
fn read_grants(path: &Path) -> Result<Option<BTreeMap<String, Vec<Capability>>>> {
if !path.exists() {
return Ok(None);
}
let file: BTreeMap<String, Vec<String>> = serde_json::from_slice(
&std::fs::read(path).with_context(|| format!("reading {}", path.display()))?,
)
.with_context(|| format!("parsing {}", path.display()))?;
Ok(Some(grants(file)))
}
/// Store the meta file's grants as the last-read swarm permissions, when the
/// file is still there and nothing has been stored yet.
pub fn seed_local() {
if let Err(e) = seed_local_at(&meta_path(), &crate::capabilities::path()) {
tracing::warn!(error = ?e, "import: seeding the stored permissions from the meta repo failed");
}
}
fn seed_local_at(meta: &Path, stored: &Path) -> Result<()> {
if stored.exists() {
return Ok(());
}
let Some(grants) = read_grants(meta)? else {
return Ok(());
};
let records = grants
.into_iter()
.map(|(agent, caps)| {
(
agent,
AgentPermissions {
capabilities: Some(caps),
},
)
})
.collect();
crate::capabilities::write_at(stored, &records).context("storing the seeded permissions")?;
tracing::info!("import: seeded the stored permissions from the meta repo's capabilities");
Ok(())
}
/// Send the meta file's grants to swarm-controller while the file is there,
/// and delete it once the controller has answered that it handled them.
///
/// # Errors
///
/// The file is still there: the import failed, or deleting the file did. The
/// next pass sends again; the controller keeps a key it already has.
pub async fn send_pending(client: &async_nats::Client, hive: &str) -> Result<()> {
send_at(client, hive, &meta_path())
.await
.context("importing the meta repo's capabilities")
}
async fn send_at(client: &async_nats::Client, hive: &str, path: &Path) -> Result<()> {
let Some(capabilities) = read_grants(path)? else {
return Ok(());
};
swarm_queue_client::ensure_connected(client)?;
let payload = serde_json::to_vec(&Import { capabilities })?;
let reply = tokio::time::timeout(
IMPORT_REPLY_DEADLINE,
client.request(agent_permissions::import_subject(hive), payload.into()),
)
.await
.context("the controller did not answer the import")?
.context("sending the import")?;
match serde_json::from_slice(&reply.payload).context("decoding the controller's answer")? {
ImportReply::Imported {
written,
kept,
skipped,
} => {
tracing::info!(
?written,
?kept,
?skipped,
"import: the controller took the meta repo's capabilities"
);
crate::meta::remove_capabilities_file()
.await
.context("deleting the imported capabilities file")
}
ImportReply::Failed { error } => {
anyhow::bail!("the controller refused the import: {error}")
}
}
}
#[cfg(test)]
mod tests {
use std::collections::BTreeMap;
use super::{grants, seed_local_at};
use hive_sh4re::permissions::Capability;
fn file(json: &str) -> BTreeMap<String, Vec<String>> {
serde_json::from_str(json).expect("decodes")
}
#[test]
fn grants_keep_known_names_and_a_revoked_ruth_stays_revoked() {
let out = grants(file(
r#"{"atlas": ["manage_root_agent", "fly_to_the_moon"], "iris": []}"#,
));
assert_eq!(out["atlas"], vec![Capability::ManageRootAgent]);
assert_eq!(out["iris"], Vec::<Capability>::new());
assert_eq!(out["ruth"], Vec::<Capability>::new());
let out = grants(file(r#"{"ruth": ["manage_root_agent"]}"#));
assert_eq!(out["ruth"], vec![Capability::ManageRootAgent]);
}
/// The seed writes once, from the file, and never over a stored read.
#[test]
fn the_seed_copies_the_file_only_when_nothing_is_stored() {
let dir = tempfile::tempdir().expect("tempdir");
let meta = dir.path().join("capabilities.json");
let stored = dir.path().join("stored.json");
seed_local_at(&meta, &stored).expect("no file is no seed");
assert!(!stored.exists());
std::fs::write(&meta, r#"{"atlas": ["manage_root_agent"]}"#).expect("seed");
seed_local_at(&meta, &stored).expect("seeded");
assert!(
crate::capabilities::holds(&stored, "atlas", Capability::ManageRootAgent)
.expect("read")
);
assert!(
!crate::capabilities::holds(&stored, "ruth", Capability::ManageRootAgent)
.expect("read")
);
std::fs::write(&stored, "{}").expect("a later read");
seed_local_at(&meta, &stored).expect("left alone");
assert_eq!(std::fs::read_to_string(&stored).expect("read"), "{}");
}
}

295
hive-c0re/src/workers/swarm_permissions.rs Normal file
View file

@ -0,0 +1,295 @@
//! Reading the capabilities the swarm grants this hive's agents.
//!
//! swarm-controller keeps one key per agent in the `agent-permissions` bucket
//! ([`swarm_queue_client::agent_permissions`]). Every [`READ_INTERVAL`] this
//! reads the key of each agent this hive has and stores what it finds with
//! [`crate::capabilities`], where the container's bind mounts are worked out.
//! Those are fixed when the container starts, so an agent whose capabilities
//! changed is restarted, after its current turn.
//!
//! A failed read, or a bucket that does not answer, leaves the stored
//! permissions as they are and restarts nothing: dropping them would put an
//! agent back on its default, which can be wider than what the swarm set.
use std::path::Path;
use std::sync::Arc;
use std::time::Duration;
use anyhow::{Context, Result};
use swarm_queue_client::agent_permissions::{self, AgentPermissions, Read};
use crate::capabilities::{self, Stored};
use crate::coordinator::Coordinator;
use crate::stats::sweep_health::{self, SweepHealth};
/// How often the permissions are read.
pub const READ_INTERVAL: Duration = Duration::from_secs(30);
/// At [`READ_INTERVAL`] this is ~90 seconds of failing reads.
const FAILURES_BEFORE_BANNER: u32 = 3;
/// Send the meta repo's grants once, then read the permissions now and every
/// [`READ_INTERVAL`] until shutdown.
pub async fn follow(
client: async_nats::Client,
coord: Arc<Coordinator>,
hive: String,
mut shutdown: tokio::sync::watch::Receiver<bool>,
) {
let mut health = SweepHealth::new("swarm_permissions_read", "warn", FAILURES_BEFORE_BANNER);
loop {
// While the meta repo's grants are not in the swarm yet, an unset key
// is not the swarm's answer, so nothing is read until they are.
let pass = match super::permissions_import::send_pending(&client, &hive).await {
Ok(()) => sync(&client).await,
Err(e) => Err(e),
};
match pass {
Ok(changed) => {
health.record_ok();
restart(&coord, changed).await;
}
Err(e) => {
tracing::warn!(error = ?e, "swarm permissions: the pass failed");
let err = format!("{e:#}");
health.record_err(|ctx| {
let age = ctx.since_last_ok.map_or_else(
|| "no success this session".to_owned(),
|d| format!("last ok {} ago", sweep_health::fmt_age(d)),
);
format!(
"syncing the swarm's agent permissions is failing ({} consecutive, \
{age}) — agents keep the permissions last read: {err}",
ctx.consecutive
)
});
}
}
tokio::select! {
() = tokio::time::sleep(READ_INTERVAL) => {}
_ = shutdown.changed() => {
tracing::info!("swarm permissions: shutdown signal received");
return;
}
}
}
}
/// Read every agent's key and store the result. Returns the agents whose
/// capabilities changed.
async fn sync(client: &async_nats::Client) -> Result<Vec<String>> {
// An unconnected client does not fail a JetStream request, it hangs on it.
swarm_queue_client::ensure_connected(client)?;
let agents: Vec<String> = crate::lifecycle::agents_for_meta_listing()
.await
.context("enumerating this hive's agents")?
.into_iter()
.map(|spec| spec.name)
.collect();
sync_with(&agents, &capabilities::path(), |agent| async move {
agent_permissions::read(client, &agent)
.await
.map_err(|e| anyhow::anyhow!(e))
.with_context(|| format!("reading the swarm's permissions for {agent}"))
})
.await
}
/// [`sync`] with the read and the file given. Nothing is written unless every
/// read succeeds.
async fn sync_with<F, Fut>(agents: &[String], path: &Path, read: F) -> Result<Vec<String>>
where
F: Fn(String) -> Fut,
Fut: std::future::Future<Output = Result<Read>>,
{
let mut read_now = Stored::new();
for agent in agents {
match read(agent.clone()).await? {
Read::Current(record) => {
if record != AgentPermissions::default() {
read_now.insert(agent.clone(), record);
}
}
Read::NoBucket => {
tracing::debug!("swarm permissions: the agent-permissions bucket does not answer");
return Ok(Vec::new());
}
}
}
let stored = capabilities::read_at(path).context("reading the stored swarm permissions")?;
if read_now == stored {
return Ok(Vec::new());
}
capabilities::write_at(path, &read_now).context("storing the swarm permissions")?;
tracing::info!(permissions = ?read_now, "swarm permissions: stored the swarm's permissions");
Ok(agents
.iter()
.filter(|a| {
capabilities::effective(a, stored.get(*a))
!= capabilities::effective(a, read_now.get(*a))
})
.cloned()
.collect())
}
/// Restart each running agent in `changed` once its current turn is done, so
/// its container starts with the bind mounts its new capabilities give. A
/// stopped agent gets them at its next start.
async fn restart(coord: &Arc<Coordinator>, changed: Vec<String>) {
let mut running = Vec::new();
for agent in changed {
if crate::lifecycle::is_running(&agent).await {
running.push(agent);
}
}
if running.is_empty() {
return;
}
tracing::info!(agents = ?running, "swarm permissions: capabilities changed, restarting");
if let Err(e) = crate::job_queue::power::restart_many(coord, &running, true).await {
tracing::error!(agents = ?running, error = ?e, "swarm permissions: restart insert failed");
}
}
#[cfg(test)]
mod tests {
use std::path::Path;
use super::{Read, sync_with};
use swarm_queue_client::agent_permissions::AgentPermissions;
fn record(json: &str) -> AgentPermissions {
serde_json::from_str(json).expect("decodes")
}
fn agents() -> Vec<String> {
vec!["atlas".to_owned(), "ruth".to_owned()]
}
/// Sync `agents()` against `path` with each agent read as `answer` gives.
async fn sync(
path: &Path,
answer: impl Fn(&str) -> anyhow::Result<Read>,
) -> anyhow::Result<Vec<String>> {
sync_with(&agents(), path, |agent| {
let out = answer(&agent);
async move { out }
})
.await
}
/// No key anywhere is everyone on the default: nothing stored, and no
/// restart, since that is what the agents already run with.
#[tokio::test]
async fn missing_keys_are_the_default_and_restart_nothing() {
let dir = tempfile::tempdir().expect("tempdir");
let path = dir.path().join("p.json");
let changed = sync(&path, |_| Ok(Read::Current(AgentPermissions::default())))
.await
.expect("read");
assert!(changed.is_empty());
assert!(
crate::capabilities::read_at(&path)
.expect("read")
.is_empty()
);
assert!(
crate::capabilities::holds(
&path,
"ruth",
hive_sh4re::permissions::Capability::ManageRootAgent
)
.expect("read")
);
}
/// Revoking ruth's default, and granting atlas, restarts both; clearing
/// them again restarts both back.
#[tokio::test]
async fn a_changed_capability_asks_for_a_restart() {
let dir = tempfile::tempdir().expect("tempdir");
let path = dir.path().join("p.json");
let changed = sync(&path, |agent| {
Ok(Read::Current(record(if agent == "ruth" {
r#"{"capabilities":[]}"#
} else {
r#"{"capabilities":["manage_root_agent"]}"#
})))
})
.await
.expect("read");
assert_eq!(changed, agents());
// The same answer again changes nothing.
let unchanged = sync(&path, |agent| {
Ok(Read::Current(record(if agent == "ruth" {
r#"{"capabilities":[]}"#
} else {
r#"{"capabilities":["manage_root_agent"]}"#
})))
})
.await
.expect("read");
assert!(unchanged.is_empty());
let cleared = sync(&path, |_| Ok(Read::Current(AgentPermissions::default())))
.await
.expect("read");
assert_eq!(cleared, agents());
}
/// Setting ruth's default explicitly is a different record but the same
/// mounts, so it is stored and restarts nothing.
#[tokio::test]
async fn an_explicit_default_is_stored_without_a_restart() {
let dir = tempfile::tempdir().expect("tempdir");
let path = dir.path().join("p.json");
let changed = sync(&path, |agent| {
Ok(Read::Current(if agent == "ruth" {
record(r#"{"capabilities":["manage_root_agent"]}"#)
} else {
AgentPermissions::default()
}))
})
.await
.expect("read");
assert!(changed.is_empty());
assert!(
crate::capabilities::read_at(&path)
.expect("read")
.contains_key("ruth")
);
}
/// One failed read, or a bucket that does not answer, keeps what was
/// stored and restarts nothing: never a fall back to the default.
#[tokio::test]
async fn a_failed_read_keeps_the_last_value() {
let dir = tempfile::tempdir().expect("tempdir");
let path = dir.path().join("p.json");
let revoked = || Ok(Read::Current(record(r#"{"capabilities":[]}"#)));
// An empty list is atlas's default already, so only ruth changes.
assert_eq!(
sync(&path, |_| revoked()).await.expect("read"),
vec!["ruth".to_owned()]
);
let before = std::fs::read(&path).expect("stored");
sync(&path, |agent| {
if agent == "ruth" {
Err(anyhow::anyhow!("timed out"))
} else {
Ok(Read::Current(AgentPermissions::default()))
}
})
.await
.expect_err("a failed read fails the pass");
assert_eq!(std::fs::read(&path).expect("stored"), before);
let changed = sync(&path, |_| Ok(Read::NoBucket))
.await
.expect("no bucket");
assert!(changed.is_empty());
assert_eq!(std::fs::read(&path).expect("stored"), before);
}
}