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

controller: turn an archetype agent into a custom one

Browse source
This commit is contained in:
atlas 2026-11-01 12:02:38 +01:00 • committed by mara
commit 0737798d78
1 changed files with 1008 additions and 44 deletions
Download patch file Download diff file Expand all files Collapse all files

611
swarm-controller/src/agent_custom.rs Normal file
View file

@ -0,0 +1,611 @@
//! `POST /api/agents/{name}/custom`: turn an archetype agent into a custom
//! one, with its own `agent-configs/<name>` repo.
//!
//! The repo's `flake.nix` declares the archetype's source flake as an input, at
//! the URL `internal/meta` holds for that source, and imports the archetype's
//! module from it. Nothing is copied: any source `internal/meta` has a pinned
//! input for works. The URL's `rev` is the only pin, since the seeded repo has
//! no lock. The meta flake points the agent's `hyperhive` input at its own, so
//! the lock holds one hyperhive. From then on the repo's owner keeps the config
//! current. The job graph is [`crate::jobq::declare_custom_job`]: repo and seed
//! first, the archetype key deleted only after the seed succeeded, then the
//! meta update and the deploy. A failure before the delete leaves the agent
//! archetype-built and working, and the same request can be sent again. A
//! failure after it leaves the agent custom, and the same request then only
//! re-queues the meta update and the deploy.
use axum::Json;
use axum::extract::{Path, State};
use axum::http::StatusCode;
use problem_details::ProblemDetails;
use serde::Serialize;
use swarm_queue_client::agent_archetype::ArchetypeRef;
use utoipa::ToSchema;
use crate::routes::agents::{claiming_hives, read_declarations};
use crate::routes::{AppState, error_problem};
use swarm_wire::hive_status::{Capabilities, NO_CONFIG_CHECKOUT};
use crate::status::{Freshness, HiveStatus};
use crate::{forge, meta_flake, schedules};
/// Answer of a queued conversion.
#[derive(Serialize, ToSchema)]
pub struct ConvertResponse {
/// A node of the queued graph; `/api/jobq/graph` serves the rest.
node_id: u64,
}
/// Turn an archetype agent into a custom agent.
#[utoipa::path(
post,
path = "/api/agents/{name}/custom",
params(("name" = String, Path, description = "archetype agent to turn into a custom one")),
responses(
(status = 200, description = "the conversion is queued", body = ConvertResponse),
(status = 400, description = "the agent name is not shaped like an identifier (problem+json)", body = String),
(status = 409, description = "the agent is placed on no hive or on several, or its hive has not reported the capability a converted agent's rebuild needs (problem+json)", body = String),
(status = 503, description = "no swarm queue is configured here or it is not connected (problem+json)", body = String),
(status = 500, description = "the queue or the job graph failed (problem+json)", body = String),
),
tag = "agents"
)]
pub(crate) async fn post_agent_custom(
State(state): State<AppState>,
Path(agent): Path<String>,
) -> Result<Json<ConvertResponse>, ProblemDetails> {
let agent = hive_types::Ident::parse(&agent)
.map_err(|reason| error_problem(StatusCode::BAD_REQUEST, reason))?
.into_string();
let client = schedules::queue(&state).map_err(schedules::refused)?;
let internal =
|e: anyhow::Error| error_problem(StatusCode::INTERNAL_SERVER_ERROR, &format!("{e:#}"));
let store = swarm_queue_client::agent_archetype::open_or_create(&client)
.await
.map_err(|e| internal(e.into()))?;
let (archetype, _, _) = schedules::read_or_default::<Option<ArchetypeRef>>(&store, &agent)
.await
.map_err(internal)?;
let declared = read_declarations(&state, state.hives.iter())
.await
.map_err(|(hive, e)| {
internal(e.context(format!("read where hive {hive} places agents")))
})?;
let Some(status) = state.status.as_ref() else {
return Err(error_problem(
StatusCode::SERVICE_UNAVAILABLE,
"no swarm queue is configured on this host",
));
};
let rows = status
.view(&state.hives, std::time::SystemTime::now())
.await
.map_err(internal)?;
let (archetype, hive) = plan(&agent, archetype, &declared, &rows)
.map_err(|why| error_problem(StatusCode::CONFLICT, &why))?;
let queued = {
let mut sched = state
.jobq
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
sched.insert_job(None, |b| {
crate::jobq::declare_custom_job(b, &agent, &hive, archetype.as_ref())
})
};
let ids = queued.map_err(|e| internal(anyhow::anyhow!("{e}")))?;
let [id] = ids[..] else {
unreachable!("exactly one handle was asked for");
};
tracing::info!(%agent, %hive, ?archetype, "turn into custom queued");
Ok(Json(ConvertResponse { node_id: id.get() }))
}
/// What the request does, or why it is refused: `archetype` is the agent's
/// stored one, and `None` for a custom agent, which a request only re-deploys
/// (the retry of a conversion whose build failed after the key was deleted).
/// The agent's hive is the one hive that places it, and must pass
/// [`offers_no_config_checkout`].
fn plan(
agent: &str,
archetype: Option<ArchetypeRef>,
declared: &[(String, swarm_wire::wanted::HiveWanted)],
rows: &[HiveStatus],
) -> Result<(Option<ArchetypeRef>, String), String> {
let hive = match claiming_hives(agent, declared).as_slice() {
[hive] => hive.clone(),
other => {
return Err(format!(
"agent {agent:?} is placed on {} hives, so its hive is unknown",
other.len()
));
}
};
offers_no_config_checkout(rows, &hive)?;
Ok((archetype, hive))
}
/// Whether `hive`'s latest status advertises [`NO_CONFIG_CHECKOUT`]. An older
/// hive keeps a checkout of a custom agent's config and would create it for a
/// converted agent under the running container's mount, so a conversion waits
/// for this.
///
/// # Errors
/// The reason, naming the hive: it has no recent status, or its status does
/// not list it. An older hive lists no capabilities at all.
fn offers_no_config_checkout(rows: &[HiveStatus], hive: &str) -> Result<(), String> {
let capability = NO_CONFIG_CHECKOUT;
let fresh = rows
.iter()
.find(|row| row.name == hive && row.freshness == Freshness::Fresh);
let Some(snapshot) = fresh.and_then(|row| row.snapshot.as_ref()) else {
return Err(format!(
"hive {hive:?} has no recent status, so whether it can rebuild a converted agent is \
unknown; retry once it reports"
));
};
if Capabilities::of_snapshot(snapshot).has(capability) {
return Ok(());
}
Err(format!(
"hive {hive:?} does not advertise the {capability:?} capability, so a converted agent \
would get a config checkout on it; deploy this release to hive {hive:?} first"
))
}
/// What [`seed`] reads and writes on the forge: a fake in the tests.
pub(crate) trait SeedForge {
/// `internal/meta`'s `flake.nix` at its head.
fn meta_flake(&self) -> impl Future<Output = anyhow::Result<Option<String>>> + Send;
/// `path` on the default branch of `agent`'s config repo; `None` when the
/// repo is empty or has no such file.
fn config_file(
&self,
agent: &str,
path: &str,
) -> impl Future<Output = anyhow::Result<Option<String>>> + Send;
/// Commit `files` to the default branch of `agent`'s config repo.
fn create_files(
&self,
agent: &str,
files: &[(&str, String)],
) -> impl Future<Output = anyhow::Result<()>> + Send;
}
impl SeedForge for forge::Client {
async fn meta_flake(&self) -> anyhow::Result<Option<String>> {
let head = self
.default_branch_files(
forge::objects::SHARED_ORG,
meta_flake::META_REPO,
&["flake.nix"],
)
.await?;
Ok(head
.and_then(|h| h.files.into_iter().next())
.map(|f| String::from_utf8_lossy(&f.bytes).into_owned()))
}
async fn config_file(&self, agent: &str, path: &str) -> anyhow::Result<Option<String>> {
let head = self
.default_branch_files(forge::CONFIG_ORG, agent, &[path])
.await?;
Ok(head
.and_then(|h| h.files.into_iter().next())
.map(|f| String::from_utf8_lossy(&f.bytes).into_owned()))
}
async fn create_files(&self, agent: &str, files: &[(&str, String)]) -> anyhow::Result<()> {
self.create_config_files(agent, files, "swarm-controller: turn into custom")
.await
}
}
/// A nix identifier, so the names can be written as attribute names unquoted.
fn is_nix_ident(name: &str) -> bool {
let mut chars = name.chars();
chars
.next()
.is_some_and(|c| c.is_ascii_lowercase() || c == '_')
&& chars.all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '_' || c == '-')
}
/// The `flake.nix` of a converted agent: the shape every config repo has, plus
/// the archetype's source as an input at `url`, and the archetype's module
/// among the imports. The module is imported here, from `inputs`, because a
/// module cannot take its own imports from `_module.args`.
fn converted_flake_nix(archetype: &ArchetypeRef, url: &str) -> String {
let ArchetypeRef { source, name } = archetype;
format!(
"{{\n description = \"hyperhive agent\";\n inputs = {{\n {source}.url = \"{url}\";\n }};\n outputs =\n {{ self, ... }}@inputs:\n {{\n nixosModules.default = {{\n imports = [\n ./agent.nix\n inputs.{source}.hyperhiveArchetypes.{name}.module\n ];\n _module.args.flakeInputs = builtins.removeAttrs inputs [ \"self\" ];\n }};\n }};\n}}\n"
)
}
/// The `agent.nix` of a converted agent: an empty module for the owner's edits.
fn converted_agent_nix(agent: &str, archetype: &ArchetypeRef) -> String {
let ArchetypeRef { source, name } = archetype;
format!(
"{{ config, pkgs, lib, ... }}:\n{{\n # Per-agent overrides for {agent}. flake.nix imports the {source} archetype\n # {name}; keeping that input current is up to this repo's owner.\n}}\n"
)
}
/// Seed `agent`'s config repo so it builds what the archetype built: its
/// `flake.nix` declares the archetype's source as an input at the URL
/// `internal/meta` holds for it today (a pinned rev), and imports the module.
/// The job behind `SeedArchetypeConfig`.
///
/// Safe to run again: a repo whose `flake.nix` is what this writes, whatever
/// URL it holds, is left as it is, since both files go in one commit. A
/// `flake.nix` that differs belongs to someone, and is an error.
///
/// # Errors
/// A forge failure, a name that is not a nix identifier, `internal/meta`
/// holding no input for the source or one not pinned to a commit, or a different
/// `flake.nix` already in the repo.
pub(crate) async fn seed(
forge: &impl SeedForge,
agent: &str,
archetype: &ArchetypeRef,
) -> anyhow::Result<()> {
use anyhow::Context as _;
let ArchetypeRef { source, name } = archetype;
for part in [source, name] {
anyhow::ensure!(
is_nix_ident(part),
"archetype name {part:?} is not a nix identifier"
);
}
let meta = forge
.meta_flake()
.await?
.context("internal/meta has no flake.nix to read the archetype source from")?;
let url = hive_meta_flake::input_url(&meta, source).with_context(|| {
format!("internal/meta has no input {source:?}, so the archetype source has no URL to pin")
})?;
// The seeded repo has no lock, so the URL's `rev` is the only pin the input has.
anyhow::ensure!(
hive_meta_flake::pinned_rev(&meta, source).is_some(),
"internal/meta's input {source:?} is not pinned to a commit, so the converted agent would float"
);
if let Some(have) = forge.config_file(agent, "flake.nix").await? {
let ours = hive_meta_flake::input_url(&have, source)
.is_some_and(|held| have == converted_flake_nix(archetype, held));
anyhow::ensure!(
ours,
"agent-configs/{agent} already holds a different flake.nix; the conversion does not \
overwrite it"
);
tracing::debug!(%agent, "turn into custom: config already seeded");
return Ok(());
}
forge
.create_files(
agent,
&[
("agent.nix", converted_agent_nix(agent, archetype)),
("flake.nix", converted_flake_nix(archetype, url)),
],
)
.await?;
tracing::info!(%agent, %url, "turn into custom: seeded the config repo");
Ok(())
}
/// Delete `agent`'s archetype key, so the next render and rebuild treat it as
/// custom. The job behind `DropArchetype`; deleting an absent key succeeds.
///
/// # Errors
/// The bucket cannot be opened or written.
pub(crate) async fn drop_archetype(client: &async_nats::Client, agent: &str) -> anyhow::Result<()> {
swarm_queue_client::agent_archetype::open_or_create(client)
.await?
.delete(agent)
.await?;
tracing::info!(%agent, "turn into custom: archetype key deleted");
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
use std::collections::BTreeMap;
use std::sync::Mutex;
const URL: &str =
"git+http://forge/hyperhive/hyperhive.git?rev=1111111111111111111111111111111111111111";
fn reviewer() -> ArchetypeRef {
ArchetypeRef {
source: "hyperhive".to_owned(),
name: "reviewer".to_owned(),
}
}
/// An in-memory forge: `internal/meta`'s flake and the files of one config
/// repo. `fail_create` makes the next `create_files` fail once, without
/// writing.
#[derive(Default)]
struct FakeForge {
meta_flake: Option<String>,
repo: Mutex<BTreeMap<String, String>>,
fail_create: Mutex<bool>,
creates: Mutex<usize>,
}
impl FakeForge {
fn with_inputs(inputs: &[(&str, &str)]) -> Self {
use std::fmt::Write as _;
let mut lines = String::new();
for (name, url) in inputs {
let _ = writeln!(lines, " {name}.url = \"{url}\";");
}
Self {
meta_flake: Some(lines),
..Self::default()
}
}
fn pinned() -> Self {
Self::with_inputs(&[("hyperhive", URL)])
}
}
impl SeedForge for FakeForge {
async fn meta_flake(&self) -> anyhow::Result<Option<String>> {
Ok(self.meta_flake.clone())
}
async fn config_file(&self, _agent: &str, path: &str) -> anyhow::Result<Option<String>> {
Ok(self.repo.lock().unwrap().get(path).cloned())
}
async fn create_files(&self, _agent: &str, files: &[(&str, String)]) -> anyhow::Result<()> {
if std::mem::take(&mut *self.fail_create.lock().unwrap()) {
anyhow::bail!("the forge refused the commit");
}
*self.creates.lock().unwrap() += 1;
let mut repo = self.repo.lock().unwrap();
for (path, content) in files {
repo.insert((*path).to_owned(), content.clone());
}
Ok(())
}
}
/// The seeded flake declares the source as an input at meta's URL and
/// imports the module from it; nothing is copied into the repo.
#[tokio::test]
async fn the_seed_declares_the_source_as_an_input_and_imports_its_module() {
let forge = FakeForge::pinned();
seed(&forge, "atlas", &reviewer()).await.expect("seeds");
let repo = forge.repo.lock().unwrap();
let flake = &repo["flake.nix"];
assert!(
flake.contains(&format!(" hyperhive.url = \"{URL}\";")),
"{flake}"
);
assert!(
flake.contains("inputs.hyperhive.hyperhiveArchetypes.reviewer.module"),
"{flake}"
);
assert!(flake.contains("_module.args.flakeInputs"), "{flake}");
assert_eq!(
repo.keys().map(String::as_str).collect::<Vec<_>>(),
["agent.nix", "flake.nix"]
);
}
/// A source other than hyperhive takes the URL `internal/meta` holds for
/// that name, and one meta has no input for is an error.
#[tokio::test]
async fn another_source_takes_its_url_from_meta_and_an_unknown_one_fails() {
let other = ArchetypeRef {
source: "extra".to_owned(),
name: "builder".to_owned(),
};
let forge = FakeForge::with_inputs(&[
("hyperhive", URL),
(
"extra",
"git+http://x/y.git?rev=2222222222222222222222222222222222222222",
),
]);
seed(&forge, "atlas", &other).await.expect("seeds");
let flake = forge.repo.lock().unwrap()["flake.nix"].clone();
assert!(
flake.contains(" extra.url = \"git+http://x/y.git?rev=2222"),
"{flake}"
);
assert!(
flake.contains("inputs.extra.hyperhiveArchetypes.builder.module"),
"{flake}"
);
let why = format!(
"{:#}",
seed(&FakeForge::pinned(), "atlas", &other)
.await
.unwrap_err()
);
assert!(why.contains("no input \"extra\""), "{why}");
}
/// A seed that failed wrote nothing, and sending it again completes; a
/// third run, after the repo is seeded, commits nothing more — even when
/// meta's pin has moved since. The presence control: the second run did
/// write.
#[tokio::test]
async fn a_seed_that_failed_is_retried_and_a_seeded_repo_is_left_alone() {
let forge = FakeForge::pinned();
*forge.fail_create.lock().unwrap() = true;
seed(&forge, "atlas", &reviewer())
.await
.expect_err("refused");
assert!(forge.repo.lock().unwrap().is_empty(), "a failed seed wrote");
seed(&forge, "atlas", &reviewer()).await.expect("retry");
assert_eq!(*forge.creates.lock().unwrap(), 1);
let moved = FakeForge {
repo: Mutex::new(forge.repo.lock().unwrap().clone()),
..FakeForge::with_inputs(&[(
"hyperhive",
"git+http://forge/h.git?rev=3333333333333333333333333333333333333333",
)])
};
seed(&moved, "atlas", &reviewer()).await.expect("again");
assert_eq!(
*moved.creates.lock().unwrap(),
0,
"a seeded repo was committed to again"
);
}
/// Someone else's `flake.nix` is never overwritten.
#[tokio::test]
async fn a_different_flake_nix_is_an_error() {
let forge = FakeForge::pinned();
forge
.repo
.lock()
.unwrap()
.insert("flake.nix".to_owned(), "{ }".to_owned());
let why = format!(
"{:#}",
seed(&forge, "atlas", &reviewer()).await.unwrap_err()
);
assert!(why.contains("different flake.nix"), "{why}");
assert_eq!(*forge.creates.lock().unwrap(), 0);
}
#[tokio::test]
async fn a_missing_meta_flake_or_a_bad_name_is_an_error() {
let forge = FakeForge::default();
seed(&forge, "atlas", &reviewer())
.await
.expect_err("no meta flake");
let bad = ArchetypeRef {
source: "hyperhive".to_owned(),
name: "9; evil".to_owned(),
};
seed(&FakeForge::pinned(), "atlas", &bad)
.await
.expect_err("bad name");
}
fn row(name: &str, freshness: Freshness, snapshot: Option<serde_json::Value>) -> HiveStatus {
HiveStatus {
name: name.to_owned(),
domain: None,
dashboard: None,
freshness,
last_seen_unix: None,
age_seconds: None,
snapshot,
}
}
/// The rollout gate: a hive listing the capability is allowed; one that
/// lists others, lists none (an older hive's payload), has gone quiet or
/// never reported is refused, naming the hive.
#[test]
fn only_a_fresh_hive_listing_the_capability_may_convert() {
let with = serde_json::json!({"status": "ok", "warnings": [], "capabilities": ["no-config-checkout"]});
let old = serde_json::json!({"status": "ok", "warnings": []});
let other = serde_json::json!({"status": "ok", "warnings": [], "capabilities": ["x"]});
let rows = |hive: Freshness, snapshot| {
vec![
row("alpha", hive, Some(snapshot)),
row("beta", Freshness::Fresh, Some(with.clone())),
]
};
assert_eq!(
offers_no_config_checkout(&rows(Freshness::Fresh, with.clone()), "alpha"),
Ok(())
);
for refused in [old.clone(), other] {
let why =
offers_no_config_checkout(&rows(Freshness::Fresh, refused), "alpha").unwrap_err();
assert!(
why.contains("\"alpha\"") && why.contains("no-config-checkout"),
"{why}"
);
}
let stale =
offers_no_config_checkout(&rows(Freshness::Stale, with.clone()), "alpha").unwrap_err();
assert!(stale.contains("no recent status"), "{stale}");
let never = vec![row("alpha", Freshness::NeverReported, None)];
offers_no_config_checkout(&never, "alpha").unwrap_err();
offers_no_config_checkout(&[], "alpha").unwrap_err();
}
/// An old-shaped payload, without `capabilities`, still parses as the
/// opaque snapshot the status reader serves.
#[test]
fn an_old_shaped_payload_still_parses_and_lists_nothing() {
let old: serde_json::Value =
serde_json::from_str(r#"{"status":"ok","warnings":[]}"#).unwrap();
assert!(old["capabilities"].as_array().is_none());
let why = offers_no_config_checkout(&[row("alpha", Freshness::Fresh, Some(old))], "alpha")
.unwrap_err();
assert!(why.contains("alpha"), "{why}");
}
fn placing(hive: &str, agent: &str) -> Vec<(String, swarm_wire::wanted::HiveWanted)> {
let wanted = swarm_wire::wanted::HiveWanted {
agents: BTreeMap::from([(
agent.to_owned(),
swarm_wire::wanted::AgentWanted {
state: swarm_wire::wanted::AgentState::Up,
},
)]),
};
vec![(hive.to_owned(), wanted)]
}
fn reporting(hive: &str, capabilities: &[&str]) -> Vec<HiveStatus> {
let snapshot =
serde_json::json!({"status": "ok", "warnings": [], "capabilities": capabilities});
vec![row(hive, Freshness::Fresh, Some(snapshot))]
}
/// The refusals and the outcomes of a request, decided before anything is
/// queued: an agent no hive places and a hive that doesn't list the
/// capability are refused; an archetype agent converts, and a custom agent
/// (its key already gone) is only re-deployed. The conversion is the
/// control that the refusals' inputs get through otherwise.
#[test]
fn a_request_is_refused_for_an_unplaced_agent_or_an_old_hive() {
let placed = placing("alpha", "atlas");
let ready = reporting("alpha", &["no-config-checkout"]);
assert_eq!(
plan("atlas", Some(reviewer()), &placed, &ready),
Ok((Some(reviewer()), "alpha".to_owned()))
);
assert_eq!(
plan("atlas", None, &placed, &ready),
Ok((None, "alpha".to_owned()))
);
let unplaced = plan("atlas", Some(reviewer()), &[], &ready).unwrap_err();
assert!(unplaced.contains("0 hives"), "{unplaced}");
let old = plan("atlas", Some(reviewer()), &placed, &reporting("alpha", &[])).unwrap_err();
assert!(old.contains("\"alpha\""), "{old}");
// The retry of a custom agent waits for the capability too.
plan("atlas", None, &placed, &reporting("alpha", &[])).unwrap_err();
}
/// A source `internal/meta` does not pin to a commit would float in a repo
/// with no lock, so the seed refuses it and writes nothing.
#[tokio::test]
async fn a_source_that_meta_does_not_pin_is_not_seeded() {
let other = ArchetypeRef {
source: "extra".to_owned(),
name: "builder".to_owned(),
};
let forge = FakeForge::with_inputs(&[("hyperhive", URL), ("extra", "git+http://x/y.git")]);
let why = format!("{:#}", seed(&forge, "atlas", &other).await.unwrap_err());
assert!(why.contains("not pinned"), "{why}");
assert!(forge.repo.lock().unwrap().is_empty());
}
}