Watch
0
0
Fork
You've already forked hyperhive
0
hyperhive/swarm-controller/src/routes/agents.rs

2371 lines
95 KiB
Rust

//! Creating agents, declaring their wanted state, and reading the roster,
//! their status and their icons.
use std::sync::Arc;
use anyhow::Result;
use axum::{
Json,
extract::{Path, State},
};
use serde::{Deserialize, Serialize};
use swarm_controller_api::agents::{CreateAgentRequest, CreateAgentResponse};
use utoipa::ToSchema;
use crate::jobq::{SwarmNodeKind, SwarmResourceKind, declare_agent_job};
use crate::routes::hives::HiveEntry;
use crate::routes::{AppState, StatusUnavailable, error_problem, wanted_error_status};
use crate::{
agent_archetype, agent_identity, agent_limits, agent_permissions, agent_status,
agents_md_initial, auth, inference, schedules, wanted,
};
/// The swarm's agent roster — every agent the swarm holds an identity for.
///
/// The identity store **is** the roster rather than one source to assemble
/// one from: an agent without a swarm identity is not a swarm agent, so
/// there is no hive-side list to merge in and no reconciliation to do.
///
/// Deliberately says nothing about health. A roster is the set other views
/// are *complete against* — it is what makes "this agent has never reported"
/// expressible at all, and that distinction only survives while the declared
/// set and the reported set stay separate.
///
/// A hive-less swarm answers `[]`; a swarm with no bridge answers 503. Those
/// are different facts and collapsing them would render an unreadable store
/// as an empty one.
#[utoipa::path(
get,
path = "/api/agents",
responses(
(status = 200, description = "every agent the swarm holds an identity for", body = Vec<String>),
(status = 503, description = "no identity bridge is configured here, or its store could not be read", body = String),
),
tag = "agents"
)]
pub(crate) async fn get_agents(
State(state): State<AppState>,
) -> Result<Json<Vec<String>>, StatusUnavailable> {
let Some(bridge) = state.auth.as_ref() else {
return Err(StatusUnavailable(
"no identity bridge is configured on this host".to_owned(),
));
};
match bridge.list_agent_identities().await {
Ok(agents) => Ok(Json(agents)),
Err(e) => {
let detail = format!("{e:#}");
tracing::warn!(error = %detail, "reading the agent roster failed");
Err(StatusUnavailable(detail))
}
}
}
/// The state to declare for one agent.
#[derive(Debug, Deserialize, ToSchema)]
pub(crate) struct SetAgentStateRequest {
/// Typed as a string in the schema only — the parse is the real enum, so
/// a value this build does not know is a 400 rather than a field that
/// silently does nothing.
#[schema(value_type = String, example = "up")]
pub(crate) state: swarm_wire::wanted::AgentState,
}
/// One agent's line in a hive's declaration.
#[derive(Clone, Debug, Serialize, ToSchema)]
pub(crate) struct AgentDeclaration {
agent: String,
state: String,
}
pub(crate) fn render(declaration: &swarm_wire::wanted::HiveWanted) -> Vec<AgentDeclaration> {
declaration
.agents
.iter()
.map(|(agent, wanted)| AgentDeclaration {
agent: agent.clone(),
state: <&str>::from(wanted.state).to_owned(),
})
.collect()
}
/// A hive name that is shaped like one and names a hive this swarm has.
///
/// Reports the status and the detail rather than a rendered
/// `ProblemDetails`: that type is 232 bytes, which makes every `Result` in
/// this path pay for the error case it usually does not take. The handlers
/// render at the boundary, where the body is actually needed.
pub(crate) fn swarm_hive(
state: &AppState,
hive: &str,
) -> Result<String, (axum::http::StatusCode, String)> {
let hive = hive_types::Ident::parse(hive)
.map_err(|reason| (axum::http::StatusCode::BAD_REQUEST, reason.to_owned()))?
.into_string();
// Shaped like a hive name, and actually one. Same two checks
// `create_agent` makes, for the same reason: a typo otherwise publishes a
// declaration under a key no hive will ever read.
if !state.hives.iter().any(|h| h.name == hive) {
return Err((
axum::http::StatusCode::BAD_REQUEST,
format!("{hive:?} is not a hive in this swarm"),
));
}
Ok(hive)
}
/// Both declaration handlers below take the same hive name and reject it the
/// same way, and need the writer besides.
fn declaration_target(
state: &AppState,
hive: &str,
) -> Result<(Arc<wanted::WantedWriter>, String), (axum::http::StatusCode, String)> {
// Before the name checks, so a deployment with no queue answers 503
// whatever the caller spelled.
let writer = state.wanted.clone().ok_or_else(|| {
(
axum::http::StatusCode::SERVICE_UNAVAILABLE,
"this deployment wired up no swarm queue, so there is nowhere to publish a declaration"
.to_owned(),
)
})?;
Ok((writer, swarm_hive(state, hive)?))
}
/// Declare what this swarm wants of one agent on one hive.
///
/// The whole declaration is returned as published, because the value is the
/// hive's entire agent map and a caller that changed one agent still wants to
/// render the rest.
#[utoipa::path(
put,
path = "/api/hives/{hive}/agents/{agent}/state",
params(
("hive" = String, Path, description = "hive whose declaration this is"),
("agent" = String, Path, description = "agent to declare"),
),
request_body = SetAgentStateRequest,
responses(
(status = 200, description = "the declaration as now published", body = Vec<AgentDeclaration>),
(status = 400, description = "a name is not an identifier, the hive is not in this swarm, the state is unknown, or `agent` is new and reserved or refused by the forge (problem+json)", body = String),
(status = 409, description = "the swarm has already placed `agent` on a different hive (problem+json)", body = String),
(status = 503, description = "no swarm queue is wired up, or it is not connected, or `agent` is new here and breaks a naming rule but whether it already exists could not be checked (problem+json)", body = String),
(status = 500, description = "the declaration could not be published, or another hive's wanted state could not be read (problem+json)", body = String),
),
tag = "agents"
)]
pub(crate) async fn set_agent_state(
State(state): State<AppState>,
axum::extract::Path((hive, agent)): axum::extract::Path<(String, String)>,
Json(req): Json<SetAgentStateRequest>,
) -> Result<Json<Vec<AgentDeclaration>>, problem_details::ProblemDetails> {
let (writer, hive) =
declaration_target(&state, &hive).map_err(|(s, d)| error_problem(s, &d))?;
let agent = hive_types::Ident::parse(&agent)
.map_err(|reason| error_problem(axum::http::StatusCode::BAD_REQUEST, reason))?
.into_string();
// Only a first declaration of the name on this hive is checked, against
// what `create_agent` refuses: an agent already declared here changes
// state whatever its name. An undeclared name that passes is accepted:
// that is how the swarm adopts an agent that predates swarm-level
// creation. Deliberate, pending the operator's decision on whether a name
// the swarm never placed should be refused here instead.
//
// The gate is held from the checks through the write. A destroy places
// nothing, so it is not checked and does not wait on creations.
let _gate = if places_agent(req.state) {
let gate = state.create_gate.lock().await;
let here = writer.view(&hive).await.map_err(|e| {
let detail = format!(
"cannot tell whether {agent:?} is already declared on hive {hive:?}, so it was \
not placed: {e:#}"
);
error_problem(wanted_error_status(&e), &detail)
})?;
refuse_placement(
&agent,
&hive,
here.as_ref(),
hive_types::reserved_names_raw().as_deref(),
|| in_roster(state.auth.as_deref(), &agent),
|| queued_placements_now(&state),
|| declarations_elsewhere(&state, &agent, &hive),
)
.await?;
Some(gate)
} else {
None
};
let declaration = writer.set(&hive, &agent, req.state).await.map_err(|e| {
tracing::warn!(hive = %hive, agent = %agent, error = %format!("{e:#}"), "declaring agent state failed");
error_problem(wanted_error_status(&e), &format!("{e:#}"))
})?;
// After the declaration and never before it: the write above is the
// destroy order, and until it lands the agent is not being torn down, so
// its credential, permissions and archetype have to keep working. Sequenced
// rather than spawned so every attempt is finished — and their outcomes
// logged — by the time the operator's call returns.
if revokes_queue_credential(req.state) {
revoke_or_complain(&hive, &agent).await;
agent_permissions::clear_on_destroy(&state, &hive, &agent).await;
agent_archetype::clear_on_destroy(&state, &hive, &agent).await;
agents_md_initial::clear_on_destroy(&state, &hive, &agent).await;
}
Ok(Json(render(&declaration)))
}
/// Whether a declaration in `state` places the agent on its hive.
///
/// Exhaustive for the reason [`revokes_queue_credential`] is. A `Destroyed`
/// agent is gone from that hive, and the hive deploys nothing for it.
fn places_agent(state: swarm_wire::wanted::AgentState) -> bool {
use swarm_wire::wanted::AgentState;
match state {
AgentState::Up | AgentState::Offline | AgentState::Paused => true,
AgentState::Destroyed => false,
}
}
/// Refuse a first placement of `agent` on `hive` that `create_agent`
/// refuses: 400 for a new name that breaks a naming rule, 409 for a name the
/// swarm has placed on another hive. An agent `here`, the hive's current
/// declaration, already declares in a placing state is placed on this hive,
/// and passes.
///
/// The roster is read only for a name that breaks a rule. Unlike in
/// `create_agent`, no later step refuses a bad name, so a roster that cannot
/// be read refuses (503) rather than warns.
async fn refuse_placement<R, D>(
agent: &str,
hive: &str,
here: Option<&swarm_wire::wanted::HiveWanted>,
reserved: Option<&str>,
read_roster: impl FnOnce() -> R,
snapshot_queued: impl FnOnce() -> Vec<(String, String)>,
read_declared: impl FnOnce() -> D,
) -> Result<(), problem_details::ProblemDetails>
where
R: std::future::Future<Output = Result<bool, String>>,
D: std::future::Future<
Output = Result<
Vec<(String, swarm_wire::wanted::HiveWanted)>,
problem_details::ProblemDetails,
>,
>,
{
let declared_here = here
.and_then(|declaration| declaration.agents.get(agent))
.is_some_and(|wanted| places_agent(wanted.state));
if declared_here {
return Ok(());
}
// Both calls log what they push here; this response has nowhere to
// carry warnings.
let mut warnings = Vec::new();
let broken = broken_name_rules(agent, reserved, &mut warnings);
if !broken.is_empty() {
let existing = read_roster().await.map_err(|why| {
let detail = format!(
"{}; whether {agent:?} is an existing agent, which would keep its name, could \
not be checked, so it was not placed: {why}",
broken.join("; ")
);
error_problem(axum::http::StatusCode::SERVICE_UNAVAILABLE, &detail)
})?;
name_verdict(agent, &broken, Ok(existing), &mut warnings)
.map_err(|detail| error_problem(axum::http::StatusCode::BAD_REQUEST, &detail))?;
}
let elsewhere = placements_elsewhere(agent, hive, snapshot_queued, read_declared).await?;
if !elsewhere.is_empty() {
let detail = format!(
"agent {agent:?} is placed on hive {}, not {hive:?} — agent names are unique across \
the swarm; declare its state there, or destroy it there first to move it",
elsewhere.join(", ")
);
return Err(error_problem(axum::http::StatusCode::CONFLICT, &detail));
}
Ok(())
}
/// Whether declaring an agent into `state` ends its queue credential's life.
///
/// Exhaustive on purpose, like every other match on
/// [`AgentState`][swarm_wire::wanted::AgentState] in this
/// tree: a state added later has to say whether it revokes rather than
/// inheriting "no" from a catch-all. `Offline` and `Paused` deliberately do
/// not — both are states an agent comes back from, and revoking on either
/// would mean an agent that could be stopped but never restarted.
fn revokes_queue_credential(state: swarm_wire::wanted::AgentState) -> bool {
use swarm_wire::wanted::AgentState;
match state {
AgentState::Destroyed => true,
AgentState::Up | AgentState::Offline | AgentState::Paused => false,
}
}
/// Revoke the credential, and make a failure impossible to miss without
/// letting it stop the teardown.
///
/// **The destroy is not blocked by this.** The declaration is already
/// published, the hive will converge on it, and refusing the operator's call
/// at this point would leave them with an agent they cannot destroy — the
/// larger of the two harms. What is left is that the failure must not be
/// quiet, or the outcome is the silent orphan this whole path exists to
/// remove: hence `error`, naming the agent, with the cause chain attached.
async fn revoke_or_complain(hive: &str, agent: &str) {
if let Err(e) = agent_identity::revoke_queue_credential(agent).await {
tracing::error!(
hive,
agent,
error = %format!("{e:#}"),
"revoking this agent's queue credential FAILED; it was destroyed anyway, so \
the credential outlives it and may still authenticate. Declaring the agent \
destroyed again re-runs this revocation."
);
}
}
/// What this swarm currently declares for a hive.
///
/// Read back from the bucket rather than from a second copy kept here: the
/// published value is the record.
#[utoipa::path(
get,
path = "/api/hives/{hive}/wanted",
params(("hive" = String, Path, description = "hive whose declaration to read")),
responses(
(status = 200, description = "the declaration, empty when nothing is published yet", body = Vec<AgentDeclaration>),
(status = 400, description = "not an identifier, or not a hive in this swarm (problem+json)", body = String),
(status = 503, description = "no swarm queue is wired up, or it is not connected (problem+json)", body = String),
(status = 500, description = "the declaration could not be read (problem+json)", body = String),
),
tag = "agents"
)]
pub(crate) async fn get_hive_wanted(
State(state): State<AppState>,
axum::extract::Path(hive): axum::extract::Path<String>,
) -> Result<Json<Vec<AgentDeclaration>>, problem_details::ProblemDetails> {
let (writer, hive) =
declaration_target(&state, &hive).map_err(|(s, d)| error_problem(s, &d))?;
let declaration = writer.view(&hive).await.map_err(|e| {
tracing::warn!(hive = %hive, error = %format!("{e:#}"), "reading the declaration failed");
error_problem(wanted_error_status(&e), &format!("{e:#}"))
})?;
Ok(Json(declaration.as_ref().map(render).unwrap_or_default()))
}
/// What each agent last said about itself, its open config PR if any, and
/// its declared wanted state if any, read at request time — the single
/// call swarm-ui's agent roster page fills its whole table from (mara,
/// 2026-09-02: "the view should be filled by a single backend call").
///
/// Every agent in the roster gets a row whether or not it has ever
/// reported — see the `agent_status` module for why, and for why its hive
/// comes from its own bucket key rather than a second lookup. `config_pr`
/// and `wanted` are both merged in here rather than inside
/// `agent_status::AgentStatusReader`: that module has no forge client or
/// wanted-state writer and is not the place to add one just for a single
/// field, whereas this handler already holds `AppState::config_prs` and
/// `AppState::wanted` alongside the roster-shaped status view — the one
/// spot that has all three without adding an avoidable dependency to a
/// reader that stays deliberately narrow. Both fields are absent on rows
/// where the underlying store isn't configured or has nothing on record,
/// same "absence is the answer" shape `get_config_prs` uses — not a
/// reason to fail the whole response over a field that already knows how
/// to be missing.
#[utoipa::path(
get,
path = "/api/agents/status",
responses(
(status = 200, description = "a row per agent, freshness + config PR derived now", body = Vec<agent_status::AgentStatusRow>),
(status = 503, description = "no swarm queue is configured here, no identity bridge is configured, or either store could not be read", body = String),
),
tag = "agents"
)]
pub(crate) async fn get_agents_status(
State(state): State<AppState>,
) -> Result<Json<Vec<agent_status::AgentStatusRow>>, StatusUnavailable> {
let Some(reader) = state.agent_status.as_ref() else {
return Err(StatusUnavailable(
"no swarm queue is configured on this host".to_owned(),
));
};
let Some(bridge) = state.auth.as_ref() else {
return Err(StatusUnavailable(
"no identity bridge is configured on this host".to_owned(),
));
};
let roster = bridge.list_agent_identities().await.map_err(|e| {
let detail = format!("{e:#}");
tracing::warn!(error = %detail, "reading the agent roster failed");
StatusUnavailable(detail)
})?;
match reader.view(&roster, std::time::SystemTime::now()).await {
Ok(mut rows) => {
if let Some(cache) = state.config_prs.get() {
let mut config_prs = cache.snapshot();
for row in &mut rows {
row.config_pr = config_prs.remove(&row.name);
}
}
// `wanted` merges in the same way and for the same reason as
// `config_pr` above — one read per distinct hive, not one per
// agent, since a declaration is already a hive's whole agent
// map.
if let Some(writer) = state.wanted.as_ref() {
// Owned `String` keys, not `&str` borrowed from `rows` —
// `declarations` outlives the loop below that needs `rows`
// mutably, so it cannot hold a live borrow into it.
let hives: std::collections::BTreeSet<String> =
rows.iter().filter_map(|r| r.hive.clone()).collect();
let mut declarations: std::collections::HashMap<
String,
swarm_wire::wanted::HiveWanted,
> = std::collections::HashMap::new();
for hive in hives {
match writer.view(&hive).await {
Ok(Some(declaration)) => {
declarations.insert(hive, declaration);
}
Ok(None) => {}
Err(e) => {
// Missing wanted state for one hive shouldn't
// fail the whole roster — same "absence is
// survivable" rule `config_pr` follows above.
tracing::warn!(
hive, error = %format!("{e:#}"),
"reading the wanted declaration failed; its agents show no wanted state"
);
}
}
}
for row in &mut rows {
row.wanted = row.hive.as_deref().and_then(|hive| {
declarations
.get(hive)?
.agents
.get(&row.name)
.map(|w| <&str>::from(w.state).to_owned())
});
}
}
if let Some(login) = state.agent_login.as_ref() {
for row in &mut rows {
row.needs_login = login.get(&row.name);
}
}
Ok(Json(rows))
}
Err(e) => {
let detail = format!("{e:#}");
tracing::warn!(error = %detail, "reading the agent-status bucket failed");
Err(StatusUnavailable(detail))
}
}
}
/// Queue the whole agent-creation job graph for `name` — two independent
/// roots, `CreateIdentity` and `CreateForgeUser`, since an authelia
/// subject and a forge account need nothing from each other. `CreateRepo`
/// runs once `CreateIdentity` succeeds; once THAT succeeds,
/// `AddRepoMember` and `InitAgentConfigRepo` both run off it — except
/// `AddRepoMember` also waits on `CreateForgeUser`, because adding a
/// collaborator that does not yet exist as a forge user is a Forgejo
/// validation error, not an idempotent no-op. `InitAgentConfigRepo` has no
/// such second dependency: seeding files uses this daemon's own forge
/// token, never the agent's, so it only ever needed the repo (mara, design
/// review: it's a jobq graph, not a linear chain). An agent created from an
/// `archetype` has no config repo, so its graph has none of the three repo
/// nodes. Returns as soon as the graph is inserted — **not** once any of it
/// has run; `run_swarm_node` does that asynchronously off the scheduler loop
/// already running
/// (`spawn_jobq_worker`). The response reports `CreateIdentity`'s id, one
/// of the graph's two roots — a caller watches the whole thing settle via
/// `/api/jobq/graph`, which serves every root, not just this one.
///
/// `name` is validated with [`hive_types::Ident::parse`] before it becomes
/// `agent`/`repo` anywhere downstream — not the *only* gate (`CreateIdentity`
/// validates server-side too), but a few hops removed from where an
/// unvalidated name would do damage (`forge::seed_agent_config`
/// interpolates `agent` into a nix comment line). Fragile if a second
/// caller of these nodes ever appears; validating here closes it locally.
#[utoipa::path(
post,
path = "/api/agents",
request_body = CreateAgentRequest,
responses(
(status = 200, description = "job chain queued", body = CreateAgentResponse),
(status = 400, description = "`name` or `hive` is not a valid identifier, `name` is new and reserved or refused by the forge, `hive` is not in this swarm, `limits` or `inference` holds a value the matching `PUT` refuses, a permission is listed twice, `archetype` is not one the stored archetype list holds, or `agents_md` is over the size limit (problem+json)", body = String),
(status = 409, description = "the swarm has already placed `name` on a different hive (problem+json)", body = String),
(status = 422, description = "an unknown field in the body, in `limits`, or in `inference`, an `effort` that is not one of the five levels, or a `permissions` entry that is not a permission name"),
(status = 503, description = "the swarm queue is not connected, so whether `name` is placed on another hive is unknown, or `limits`, `inference`, `permissions`, `archetype` or `agents_md` is set and cannot be stored (problem+json)", body = String),
(status = 500, description = "the job chain could not be queued, or another hive's wanted state could not be read (problem+json)", body = String),
),
tag = "agents"
)]
pub(crate) async fn create_agent(
State(state): State<AppState>,
Json(req): Json<CreateAgentRequest>,
) -> Result<Json<CreateAgentResponse>, problem_details::ProblemDetails> {
let agent = hive_types::Ident::parse(&req.name)
.map_err(|reason| error_problem(axum::http::StatusCode::BAD_REQUEST, reason))?
.into_string();
// Settings are checked before anything is created, so a bad value stops
// the request rather than leaving an agent without them.
let settings = CreateSettings::parse(&req)
.map_err(|why| error_problem(axum::http::StatusCode::BAD_REQUEST, &why))?;
// Two checks, and the second is the one that makes the field worth
// having: `Ident::parse` says the string is *shaped* like a hive name,
// and the roster says it *is* one. Without the roster check a typo is
// accepted, the agent gets created, and nothing notices until a deploy
// message is addressed to a hive that does not exist — by which point
// the operator who could have corrected it in one keystroke is gone.
//
// `state.hives` is the swarm directory loaded from `SWARM_CONTROLLER_HIVES`
// at startup and never mutated, so this is a scan of a handful of
// entries against a value an operator just chose.
// The name is shaped like an identifier; the checks below ask whether it
// is *available*. An operator with agents already created under a
// colliding name must still be able to re-run creation, so a name that
// breaks a naming rule is refused only when it is new — see
// `name_verdict`. The hive collision further down still only warns, on
// its grace period, once the warning has had time to be seen.
//
// Collected rather than logged-and-dropped — see `CreateAgentResponse`.
let mut warnings = Vec::new();
let broken = broken_name_rules(
&agent,
hive_types::reserved_names_raw().as_deref(),
&mut warnings,
);
if !broken.is_empty() {
let existing = in_roster(state.auth.as_deref(), &agent).await;
name_verdict(&agent, &broken, existing, &mut warnings)
.map_err(|detail| error_problem(axum::http::StatusCode::BAD_REQUEST, &detail))?;
}
let hive = hive_types::Ident::parse(&req.hive)
.map_err(|reason| error_problem(axum::http::StatusCode::BAD_REQUEST, reason))?
.into_string();
// The roster is the same source the `hive` field is validated against
// immediately below — one scan, asked of the other field. A hive and an
// agent sharing a name collide in the swarm's own directory keys, which
// is what makes a pipeline and a label vanish with no failed assertion.
if state.hives.iter().any(|h| h.name == agent) {
let detail = format!(
"agent name {agent:?} is also a hive in this swarm — they share the swarm's directory \
keys; this will become an error"
);
tracing::warn!(agent = %agent, "create_agent: name collides with a hive");
warnings.push(detail);
}
if !state.hives.iter().any(|h| h.name == hive) {
let known: Vec<&str> = state.hives.iter().map(|h| h.name.as_str()).collect();
// Name the hives that *would* work: the caller is an operator who
// just picked one, and "not in this swarm" without the roster
// leaves them guessing at a typo they cannot see.
let known = if known.is_empty() {
"(none configured)".to_owned()
} else {
known.join(", ")
};
let detail = format!("hive {hive:?} is not in this swarm — known hives: {known}");
return Err(error_problem(axum::http::StatusCode::BAD_REQUEST, &detail));
}
// Checked after the roster, so a mistyped hive stays a 400 while the queue
// is down.
let settings_queue = settings.queue(&state).await?;
// Agent names are one swarm-wide namespace: identity, forge user, store
// path and matrix id all carry the bare name. The same name on the same
// hive is that agent being re-created, and goes through.
let _gate = state.create_gate.lock().await;
let elsewhere = placements_elsewhere(
&agent,
&hive,
|| queued_placements_now(&state),
|| declarations_elsewhere(&state, &agent, &hive),
)
.await?;
if !elsewhere.is_empty() {
let detail = format!(
"agent name {agent:?} is already placed on hive {} — agent names are unique across \
the swarm; destroy it there first to move it, or choose another name",
elsewhere.join(", ")
);
return Err(error_problem(axum::http::StatusCode::CONFLICT, &detail));
}
// Stored ahead of the job graph, not once the agent is declared: the hive
// reads these by agent name when it first runs the agent. A failure from
// here on puts back what was stored, so a refused creation leaves no
// settings behind for the name.
let stored = match &settings_queue {
Some(client) => {
StoredSettings::store(
client,
&agent,
settings.limits.as_ref(),
settings.inference.as_ref(),
settings.permissions.as_ref(),
settings.archetype.as_ref(),
settings.agents_md.as_deref(),
)
.await?
}
None => StoredSettings::default(),
};
let queued = {
let mut sched = state
.jobq
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
sched.insert_job(None, |b| {
declare_agent_job(b, &agent, &hive, settings.archetype.is_some())
})
};
let ids = match queued {
Ok(ids) => ids,
Err(e) => {
if let Some(client) = &settings_queue {
stored.roll_back(client, &agent).await;
}
return Err(error_problem(
axum::http::StatusCode::INTERNAL_SERVER_ERROR,
&e.to_string(),
));
}
};
let [id] = ids[..] else {
unreachable!("exactly one handle was asked for");
};
Ok(Json(CreateAgentResponse {
node_id: id.get(),
warnings,
}))
}
/// The settings a `POST /api/agents` body gives, as they are stored.
struct CreateSettings {
limits: Option<swarm_wire::agent_limits::AgentCaps>,
inference: Option<swarm_wire::agent_inference::AgentInference>,
permissions: Option<swarm_wire::agent_permissions::AgentPermissions>,
archetype: Option<swarm_queue_client::agent_archetype::ArchetypeRef>,
agents_md: Option<String>,
}
impl CreateSettings {
/// Each setting given, or why its `PUT` would refuse the first bad one.
/// An archetype is checked here for its shape only; whether the swarm
/// lists it is [`agent_archetype::check`].
fn parse(req: &CreateAgentRequest) -> Result<Self, String> {
let CreateAgentRequest {
limits,
inference,
permissions,
archetype,
agents_md,
..
} = req.clone();
if let Some(text) = &agents_md {
agents_md_initial::check(text)?;
}
if let Some(a) = &archetype {
for part in [&a.source, &a.name] {
hive_types::Ident::parse(part)
.map_err(|why| format!("archetype {part:?}: {why}"))?;
}
}
Ok(Self {
limits: limits.as_ref().map(agent_limits::into_stored).transpose()?,
inference: inference.map(inference::into_stored).transpose()?,
permissions: permissions.map(agent_permissions::at_create).transpose()?,
archetype,
agents_md,
})
}
/// Whether any setting is given.
fn any(&self) -> bool {
self.limits.is_some()
|| self.inference.is_some()
|| self.permissions.is_some()
|| self.archetype.is_some()
|| self.agents_md.is_some()
}
/// The queue to store the settings through, `None` when none is given.
/// An archetype is refused here unless the stored list holds it.
async fn queue(
&self,
state: &AppState,
) -> Result<Option<async_nats::Client>, problem_details::ProblemDetails> {
if !self.any() {
return Ok(None);
}
let client = schedules::queue(state).map_err(schedules::refused)?;
if let Some(archetype) = &self.archetype {
agent_archetype::check(&client, archetype).await?;
}
Ok(Some(client))
}
}
/// What `create_agent` replaced in the settings stores, to put back if the
/// creation is refused after the write.
#[derive(Default)]
struct StoredSettings {
limits: agent_limits::Replaced,
inference: inference::Replaced,
permissions: agent_permissions::Replaced,
archetype: agent_archetype::Replaced,
agents_md: agents_md_initial::Replaced,
}
impl StoredSettings {
/// Store whichever of `limits`, `permissions`, `inference`, `archetype` and `agents_md` are
/// given, in that order. A failure on one puts back the ones stored before it.
async fn store(
client: &async_nats::Client,
agent: &str,
limits: Option<&swarm_wire::agent_limits::AgentCaps>,
inference: Option<&swarm_wire::agent_inference::AgentInference>,
permissions: Option<&swarm_wire::agent_permissions::AgentPermissions>,
archetype: Option<&swarm_queue_client::agent_archetype::ArchetypeRef>,
agents_md: Option<&str>,
) -> Result<Self, problem_details::ProblemDetails> {
let mut stored = Self::default();
if let Some(limits) = limits {
stored.limits = agent_limits::store(client, agent, limits).await?;
}
if let Some(permissions) = permissions {
match agent_permissions::store(client, agent, permissions).await {
Ok(replaced) => stored.permissions = replaced,
Err(e) => {
stored.roll_back(client, agent).await;
return Err(e);
}
}
}
if let Some(inference) = inference {
match inference::store(client, agent, inference).await {
Ok(replaced) => stored.inference = replaced,
Err(e) => {
stored.roll_back(client, agent).await;
return Err(e);
}
}
}
if let Some(archetype) = archetype {
match agent_archetype::store(client, agent, archetype).await {
Ok(replaced) => stored.archetype = replaced,
Err(e) => {
stored.roll_back(client, agent).await;
return Err(e);
}
}
}
if let Some(text) = agents_md {
match agents_md_initial::store(client, agent, text).await {
Ok(replaced) => stored.agents_md = replaced,
Err(e) => {
stored.roll_back(client, agent).await;
return Err(e);
}
}
}
Ok(stored)
}
async fn roll_back(self, client: &async_nats::Client, agent: &str) {
agent_limits::restore(client, agent, self.limits).await;
inference::restore(client, agent, self.inference).await;
agent_permissions::restore(client, agent, self.permissions).await;
agent_archetype::restore(client, agent, self.archetype).await;
agents_md_initial::restore(client, agent, self.agents_md).await;
}
}
/// [`placed_elsewhere`] over a snapshot of the queued placements and a read
/// of the published ones, **in that order**.
///
/// A `SetAgentWanted` node writes its declaration before it turns terminal.
/// Snapshotting the queue first means a node already terminal there has
/// written, so the read that follows sees its declaration; a node still
/// queued is in the snapshot. Reading first leaves a gap: a node finishing
/// between the two is in neither.
async fn placements_elsewhere<E, Fut>(
agent: &str,
hive: &str,
snapshot_queued: impl FnOnce() -> Vec<(String, String)>,
read_declared: impl FnOnce() -> Fut,
) -> Result<Vec<String>, E>
where
Fut: std::future::Future<Output = Result<Vec<(String, swarm_wire::wanted::HiveWanted)>, E>>,
{
let queued = snapshot_queued();
let declared = read_declared().await?;
Ok(placed_elsewhere(agent, hive, &declared, &queued))
}
/// Every hive but `hive`'s published wanted state, for [`placed_elsewhere`].
///
/// No queue wired up means no hive has been sent a declaration, so there is
/// nothing to collide with. A queue that cannot be read refuses: an unread
/// declaration and an absent one look the same, and reading "absent" is how
/// a duplicate would get through.
async fn declarations_elsewhere(
state: &AppState,
agent: &str,
hive: &str,
) -> Result<Vec<(String, swarm_wire::wanted::HiveWanted)>, problem_details::ProblemDetails> {
read_declarations(state, state.hives.iter().filter(|h| h.name != hive))
.await
.map_err(|(other, e)| {
let detail = format!(
"cannot tell whether {agent:?} already exists on hive {other:?}, so it was not \
placed: {e:#}"
);
error_problem(wanted_error_status(&e), &detail)
})
}
/// The published wanted state of each of `hives` that has one. No queue wired
/// up reads as nothing declared anywhere. The error names the hive whose read
/// failed.
pub(crate) async fn read_declarations<'a>(
state: &AppState,
hives: impl Iterator<Item = &'a HiveEntry>,
) -> Result<Vec<(String, swarm_wire::wanted::HiveWanted)>, (String, anyhow::Error)> {
let Some(writer) = state.wanted.as_deref() else {
return Ok(Vec::new());
};
let mut declared = Vec::new();
for hive in hives {
match writer.view(&hive.name).await {
Ok(Some(declaration)) => declared.push((hive.name.clone(), declaration)),
Ok(None) => {}
Err(e) => return Err((hive.name.clone(), e)),
}
}
Ok(declared)
}
/// Every hive whose declaration places `agent` there.
pub(crate) fn claiming_hives(
agent: &str,
declared: &[(String, swarm_wire::wanted::HiveWanted)],
) -> Vec<String> {
declared
.iter()
.filter(|(_, declaration)| {
declaration
.agents
.get(agent)
.is_some_and(|wanted| places_agent(wanted.state))
})
.map(|(hive, _)| hive.clone())
.collect()
}
/// [`queued_placements`] of the swarm's own queue, as it is now.
fn queued_placements_now(state: &AppState) -> Vec<(String, String)> {
queued_placements(
state
.jobq
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.graph(),
)
}
/// `(hive, agent)` of every `SetAgentWanted` node not yet settled: a creation
/// accepted but not yet visible in the wanted state it is about to write.
fn queued_placements(
graph: &hive_jobq::Graph<SwarmNodeKind, SwarmResourceKind>,
) -> Vec<(String, String)> {
graph
.nodes()
.filter(|node| !node.state.is_terminal())
.filter_map(|node| match &node.payload {
SwarmNodeKind::SetAgentWanted { hive, agent } => Some((hive.clone(), agent.clone())),
_ => None,
})
.collect()
}
/// The hives other than `hive` the swarm has placed `agent` on, sorted.
///
/// A placement is a declaration in a state that [places][places_agent] the
/// agent, or a queued one. A `Destroyed` agent is gone from that hive, so
/// placing it on another moves it.
fn placed_elsewhere(
agent: &str,
hive: &str,
declared: &[(String, swarm_wire::wanted::HiveWanted)],
queued: &[(String, String)],
) -> Vec<String> {
let declared = declared.iter().filter(|(_, declaration)| {
declaration
.agents
.get(agent)
.is_some_and(|wanted| places_agent(wanted.state))
});
let queued = queued.iter().filter(|(_, queued)| queued == agent);
let hives: std::collections::BTreeSet<&str> = declared
.map(|(h, _)| h.as_str())
.chain(queued.map(|(h, _)| h.as_str()))
.filter(|h| *h != hive)
.collect();
hives.into_iter().map(str::to_owned).collect()
}
/// Every naming rule `agent` breaks, each as a sentence naming the rule.
///
/// `reserved` is [`hive_types::reserved_names_raw`]: the list nix owns in
/// `nix/reserved-names.nix`, read by this daemon, by hive-c0re and by the
/// swarm collector's own owner assertion. `None` means this process was never
/// told, which is not the same as "no name is reserved" — staying quiet there
/// would be a check that reports clean because it could not run, so it goes
/// onto `warnings` instead.
fn broken_name_rules(
agent: &str,
reserved: Option<&str>,
warnings: &mut Vec<String>,
) -> Vec<String> {
let mut broken = Vec::new();
match reserved.map(hive_types::parse_reserved_names) {
None => {
tracing::error!(
var = hive_types::RESERVED_NAMES_ENV,
"reserved-name check could not run — variable not set"
);
warnings.push(format!(
"the reserved-name check did not run: {} is unset, so {agent:?} was accepted \
without being checked against the reserved names",
hive_types::RESERVED_NAMES_ENV
));
}
Some(reserved) if hive_types::is_reserved_name(agent, &reserved) => {
broken.push(format!(
"agent name {agent:?} is reserved (nix/reserved-names.nix): hyperhive's message \
layer already uses it, or the forge refuses it as a username"
));
}
Some(_) => {}
}
if let Some(rule) = hive_types::forge_username_violation(agent) {
broken.push(format!(
"agent name {agent:?} {rule}: the forge refuses it as a username"
));
}
if let Some(rule) = hive_types::reserved_agent_prefix_violation(agent) {
broken.push(format!("agent name {agent:?} {rule}"));
}
broken
}
/// Whether `agent` is already in the swarm roster, or why that could not be
/// told. `create_agent` asks only of a name that breaks a rule, so an
/// ordinary creation still never waits on the bridge; `make_forge_admin`
/// asks of every name.
pub(crate) async fn in_roster(
auth: Option<&auth::AuthBridge>,
agent: &str,
) -> Result<bool, String> {
let Some(auth) = auth else {
return Err("no identity bridge is configured".to_owned());
};
auth.list_agent_identities()
.await
.map(|roster| roster.iter().any(|name| name == agent))
.map_err(|e| format!("{e:#}"))
}
/// Refuse a NEW agent whose name breaks a rule; warn for one that exists.
///
/// Re-creating an agent under its own name is a migration path (mara, on
/// the duplicate-name refusal), so an agent already in the roster keeps its
/// name with a warning. When the roster cannot be read, a new name and an
/// existing one look the same, and this warns rather than refuse: refusing
/// would block that re-creation, and the forge still refuses the account one
/// node in, as it did before.
///
/// # Errors
/// Every rule `agent` breaks, as one caller-ready sentence, when `agent` is
/// not in the roster.
pub(crate) fn name_verdict(
agent: &str,
broken: &[String],
existing: Result<bool, String>,
warnings: &mut Vec<String>,
) -> Result<(), String> {
match existing {
Ok(false) => Err(format!("{}; choose another name", broken.join("; "))),
Ok(true) => {
tracing::warn!(agent, "existing agent breaks a naming rule");
warnings.extend(
broken
.iter()
.map(|rule| format!("{rule} — accepted only because {agent:?} already exists")),
);
Ok(())
}
Err(why) => {
tracing::warn!(agent, why, "naming rule broken, roster unreadable");
warnings.extend(broken.iter().map(|rule| {
format!(
"{rule} — not refused, because whether {agent:?} already exists could not \
be checked: {why}"
)
}));
Ok(())
}
}
}
/// How long a served icon may be reused without asking again.
///
/// An agent roster asks once per row, and an icon changes about as often
/// as an agent is reconfigured — so a few minutes of browser caching is
/// what keeps a page render from being one bucket read per agent, and
/// costs nothing anyone will notice.
const ICON_MAX_AGE: u32 = 300;
/// `agent`'s icon — the swarm-level answer to "what does this agent look
/// like".
///
/// **Nothing on this path names a hive.** The icon is a property of the
/// agent, held at swarm scope under the agent's own key, so it is
/// answerable for an agent that has moved hives or that no hive is
/// currently running. That is the whole point of it living here rather
/// than being fetched from wherever the agent happens to run.
///
/// **404 means this agent has no icon**, which is the expected
/// unconfigured state and not an error — the same contract the per-agent
/// harness's own `GET /icon` already has, so a caller falls back
/// client-side on a failed load rather than probing first.
///
/// 🩸 **The body is an untrusted document served from this daemon's own
/// origin**, and an SVG can carry script. The response headers say it may
/// not run any: `Content-Security-Policy: sandbox` with no `allow-scripts`
/// keeps a direct navigation to this URL from executing it, and `nosniff`
/// keeps a browser from deciding the bytes are something else. An `<img>`
/// load — the one consumer — never runs script regardless; the headers are
/// for every other way a URL gets opened.
#[utoipa::path(
get,
path = "/api/agents/{name}/icon",
params(("name" = String, Path, description = "agent name")),
responses(
(status = 200, description = "the agent's icon, an SVG", content_type = "image/svg+xml"),
(status = 400, description = "not an identifier (problem+json)", body = String),
(status = 404, description = "this agent has no icon (problem+json)", body = String),
(status = 500, description = "the icon bucket could not be read (problem+json)", body = String),
(status = 503, description = "no swarm queue is wired up, or it is not connected (problem+json)", body = String),
),
tag = "agents"
)]
pub(crate) async fn get_agent_icon(
State(state): State<AppState>,
Path(name): Path<String>,
) -> Result<axum::response::Response, problem_details::ProblemDetails> {
use axum::http::{StatusCode, header};
use axum::response::IntoResponse as _;
let Some(reader) = state.agent_icons.as_ref() else {
return Err(error_problem(
StatusCode::SERVICE_UNAVAILABLE,
"no swarm queue is wired up on this host",
));
};
let agent = hive_types::Ident::parse(&name)
.map_err(|reason| error_problem(StatusCode::BAD_REQUEST, reason))?
.into_string();
let icon = reader.get(&agent).await.map_err(|e| {
tracing::warn!(agent = %agent, error = %format!("{e:#}"), "reading the agent icon failed");
error_problem(wanted_error_status(&e), &format!("{e:#}"))
})?;
let Some(icon) = icon else {
return Err(error_problem(
StatusCode::NOT_FOUND,
"this agent has no icon",
));
};
Ok((
[
(
header::CONTENT_TYPE,
swarm_wire::agent_icon::MEDIA_TYPE.to_owned(),
),
(
header::CACHE_CONTROL,
format!("public, max-age={ICON_MAX_AGE}"),
),
(header::CONTENT_SECURITY_POLICY, "sandbox".to_owned()),
(header::X_CONTENT_TYPE_OPTIONS, "nosniff".to_owned()),
],
icon,
)
.into_response())
}
#[cfg(test)]
mod tests {
use super::*;
use crate::routes::tests::{SharedSched, state_with_roster};
/// A swarm with no identity bridge cannot answer the roster, and must
/// say so rather than answer `[]`.
///
/// The distinction is the whole reason this endpoint exists: an empty
/// roster means *no agents*, and a UI that renders "no agents" for a
/// store it could not read is showing a fact nobody established. 503
/// rather than 500 for the same reason the status route uses it — a
/// bridge is wired up per deployment, so a caller retrying is right.
#[tokio::test]
async fn a_roster_with_no_bridge_refuses_rather_than_answering_empty() {
use axum::response::IntoResponse as _;
let (state, _sched) = state_with_roster();
assert!(state.auth.is_none(), "the fixture must have no bridge");
let err = super::get_agents(axum::extract::State(state))
.await
.expect_err("a bridge-less controller cannot produce a roster");
let resp = err.into_response();
assert_eq!(resp.status(), axum::http::StatusCode::SERVICE_UNAVAILABLE);
let bytes = axum::body::to_bytes(resp.into_body(), 64 * 1024)
.await
.expect("body reads");
let v: serde_json::Value = serde_json::from_slice(&bytes).expect("problem+json parses");
// Asserted on the rendered body rather than the error value: what a
// consumer can distinguish is what matters, and `[]` and this share
// a type on the Rust side.
assert_eq!(v["status"], 503);
assert!(
v["detail"]
.as_str()
.is_some_and(|d| d.contains("identity bridge")),
"the cause must name what is missing, got {:?}",
v["detail"]
);
}
/// An icon route with no bucket behind it must say it could not ask,
/// not that the agent has no icon.
///
/// The two are one status code apart and read identically to an
/// `<img>` — both end in the fallback glyph — which is exactly why the
/// distinction has to be asserted here: nothing downstream would ever
/// notice the 503 silently becoming a 404.
#[tokio::test]
async fn an_icon_with_no_queue_refuses_rather_than_answering_no_icon() {
use axum::response::IntoResponse as _;
let (state, _sched) = state_with_roster();
assert!(
state.agent_icons.is_none(),
"the fixture must have no queue"
);
let err = super::get_agent_icon(
axum::extract::State(state),
axum::extract::Path("iris".to_owned()),
)
.await
.expect_err("a queue-less controller cannot answer an icon");
let resp = err.into_response();
assert_eq!(resp.status(), axum::http::StatusCode::SERVICE_UNAVAILABLE);
}
/// The roster check is the half that makes the recorded hive worth
/// having, so assert it by EFFECT rather than by the message: a hive
/// that is not in this swarm must be refused **before anything is
/// queued**. A version that queued the graph and then complained would
/// satisfy an assertion on the status code alone while still creating
/// the agent — which is the failure this exists to stop.
#[tokio::test]
async fn a_hive_outside_the_roster_is_refused_before_anything_is_queued() {
let (state, sched) = state_with_roster();
let err = super::create_agent(
axum::extract::State(state),
axum::Json(CreateAgentRequest {
name: "atlas".to_owned(),
hive: "pr1maa".to_owned(),
limits: None,
inference: None,
permissions: None,
archetype: None,
agents_md: None,
}),
)
.await
.expect_err("a hive outside the roster must be refused");
// Names what would have worked: an operator who just mistyped a
// hive cannot see the roster from the error otherwise.
let rendered = format!("{err:?}");
assert!(
rendered.contains("pr1ma"),
"the refusal should name the known hives, got: {rendered}"
);
let queued = sched
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.graph()
.nodes()
.count();
assert_eq!(queued, 0, "a refused creation must queue no work");
}
/// The control for the arm above: the same call with a hive that IS in
/// the roster gets through and queues the graph. Without this, "refused"
/// could equally mean the endpoint refuses everything.
#[tokio::test]
async fn a_hive_in_the_roster_is_accepted_and_queues_the_graph() {
let (state, sched) = state_with_roster();
// The response is bound rather than asserted on: node ids start at
// zero, so every property I reached for first ("> 0") was a claim
// about the id allocator rather than about this endpoint. What the
// accept arm is actually for is the queue count below.
let _queued = super::create_agent(
axum::extract::State(state),
axum::Json(CreateAgentRequest {
name: "atlas".to_owned(),
hive: "pr1ma".to_owned(),
limits: None,
inference: None,
permissions: None,
archetype: None,
agents_md: None,
}),
)
.await
.expect("a hive in the roster must be accepted");
let queued = sched
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.graph()
.nodes()
.count();
assert!(queued > 0, "an accepted creation must queue work");
}
/// A request with no settings is the one callers sent before the fields
/// existed, and each field is checked by the same rules as its `PUT`.
#[test]
fn a_creation_request_takes_optional_settings() {
let plain: CreateAgentRequest =
serde_json::from_str(r#"{"name":"atlas","hive":"pr1ma"}"#).expect("plain body");
assert!(plain.limits.is_none() && plain.inference.is_none());
let full: CreateAgentRequest = serde_json::from_str(
r#"{"name":"atlas","hive":"pr1ma","limits":{"cpu_quota":"400%"},
"inference":{"model":"opus","effort":"high"}}"#,
)
.expect("body with settings");
assert!(full.limits.is_some() && full.inference.is_some());
let typo = serde_json::from_str::<CreateAgentRequest>(
r#"{"name":"atlas","hive":"pr1ma","limits":{"cpu":"400%"}}"#,
);
assert!(typo.is_err(), "an unknown cap field must be refused");
}
/// A bad cap or model must stop the creation before the graph is queued:
/// the point of taking settings in the creating request is that a refusal
/// leaves no agent without them.
#[tokio::test]
async fn bad_settings_are_refused_before_anything_is_queued() {
use axum::http::StatusCode;
let bad: [(&str, &str); 2] = [
(r#"{"cpu_quota":"lots"}"#, "limits"),
(r#"{"model":"-opus"}"#, "inference"),
];
for (value, field) in bad {
let (state, sched) = state_with_roster();
let body = format!(r#"{{"name":"atlas","hive":"pr1ma","{field}":{value}}}"#);
let req: CreateAgentRequest = serde_json::from_str(&body).expect("body parses");
let err = super::create_agent(axum::extract::State(state), axum::Json(req))
.await
.expect_err("a bad setting must be refused");
assert_eq!(
axum::response::IntoResponse::into_response(err).status(),
StatusCode::BAD_REQUEST,
"{field}"
);
assert_eq!(node_count(&sched), 0, "{field}: nothing may be queued");
}
}
/// Settings that cannot be stored, here because there is no queue, refuse
/// the creation rather than leaving an agent without them.
#[tokio::test]
async fn settings_that_cannot_be_stored_refuse_the_creation() {
let (state, sched) = state_with_roster();
let req: CreateAgentRequest = serde_json::from_str(
r#"{"name":"atlas","hive":"pr1ma","limits":{"cpu_quota":"400%"}}"#,
)
.expect("body parses");
let err = super::create_agent(axum::extract::State(state), axum::Json(req))
.await
.expect_err("no queue means the caps cannot be stored");
assert_eq!(
axum::response::IntoResponse::into_response(err).status(),
axum::http::StatusCode::SERVICE_UNAVAILABLE
);
assert_eq!(node_count(&sched), 0, "nothing may be queued");
}
/// Through the real handler and a real `nats-server`: both settings land
/// under the agent's key as the hive reads them, and the graph is queued.
#[tokio::test]
async fn a_creation_with_settings_stores_them_and_queues_the_graph() {
let nats = crate::test_nats::TestNats::start();
let client = nats.client().await;
let state = crate::agent_limits::tests::state(&client);
let req: CreateAgentRequest = serde_json::from_str(
r#"{"name":"atlas","hive":"pr1ma","limits":{"cpu_quota":"400%"},
"inference":{"model":"opus","max_subagents":3}}"#,
)
.expect("body parses");
let _queued = super::create_agent(axum::extract::State(state.clone()), axum::Json(req))
.await
.expect("a valid creation with settings goes through");
assert_eq!(
crate::agent_limits::tests::published(&client, "atlas")
.await
.as_deref(),
Some(r#"{"cpu_quota":"400%"}"#)
);
let inference = swarm_queue_client::agent_inference::open_or_create(&client)
.await
.expect("bucket")
.get("atlas")
.await
.expect("read")
.expect("a value");
assert_eq!(&inference[..], br#"{"model":"opus","max_subagents":3}"#);
assert!(
state
.jobq
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.graph()
.nodes()
.count()
> 0,
"the graph must be queued"
);
}
/// When the inference store fails after the caps are stored, the caps go
/// back to what they were: the earlier value, or no key. Without the
/// roll-back this leaves a cap behind for an agent that was never created.
/// The inference bucket is made with a one-byte value limit, so any write
/// to it fails.
#[tokio::test]
async fn a_failed_second_store_puts_the_first_back() {
let nats = crate::test_nats::TestNats::start();
let client = nats.client().await;
async_nats::jetstream::new(client.clone())
.create_key_value(async_nats::jetstream::kv::Config {
bucket: swarm_wire::agent_inference::BUCKET.to_owned(),
history: 1,
max_value_size: 1,
..Default::default()
})
.await
.expect("a bucket that refuses any value");
let caps = |json: &str| {
crate::agent_limits::into_stored(
&serde_json::from_str::<swarm_controller_api::limits::Caps>(json).expect("caps"),
)
.expect("valid")
};
let inference = crate::inference::into_stored(
serde_json::from_str::<swarm_controller_api::inference::InferenceSettings>(
r#"{"model":"opus"}"#,
)
.expect("settings"),
)
.expect("valid");
crate::agent_limits::store(&client, "atlas", &caps(r#"{"cpu_quota":"100%"}"#))
.await
.expect("an earlier cap");
for agent in ["atlas", "iris"] {
let stored = super::StoredSettings::store(
&client,
agent,
Some(&caps(r#"{"cpu_quota":"400%"}"#)),
Some(&inference),
None,
None,
None,
)
.await;
assert!(stored.is_err(), "{agent}: the inference write must fail");
}
assert_eq!(
crate::agent_limits::tests::published(&client, "atlas")
.await
.as_deref(),
Some(r#"{"cpu_quota":"100%"}"#),
"the earlier cap comes back"
);
assert_eq!(
crate::agent_limits::tests::published(&client, "iris").await,
None,
"a cap stored for a name with none is removed"
);
}
/// Through the real handler: when the inference store fails after the
/// permissions are stored, the creation is refused and the permissions go
/// back to what they were, the earlier value or no key. The inference
/// bucket refuses any value, as above, and the refusal is that bucket's.
#[tokio::test]
async fn a_failed_creation_leaves_no_permissions_behind() {
let nats = crate::test_nats::TestNats::start();
let client = nats.client().await;
async_nats::jetstream::new(client.clone())
.create_key_value(async_nats::jetstream::kv::Config {
bucket: swarm_wire::agent_inference::BUCKET.to_owned(),
history: 1,
max_value_size: 1,
..Default::default()
})
.await
.expect("a bucket that refuses any value");
let state = crate::agent_limits::tests::state(&client);
let earlier = crate::agent_permissions::at_create(vec![hive_types::Permission::Messaging])
.expect("valid");
crate::agent_permissions::store(&client, "atlas", &earlier)
.await
.expect("an earlier grant");
for agent in ["atlas", "iris"] {
let req: CreateAgentRequest = serde_json::from_str(&format!(
r#"{{"name":"{agent}","hive":"pr1ma","permissions":["web_tools"],
"inference":{{"model":"opus"}}}}"#
))
.expect("body parses");
let err = super::create_agent(axum::extract::State(state.clone()), axum::Json(req))
.await
.expect_err("the inference write must fail");
assert_eq!(
err.status,
Some(axum::http::StatusCode::INTERNAL_SERVER_ERROR),
"{agent}: {err:?}"
);
assert!(
err.detail
.as_deref()
.is_some_and(|d| d.contains("message size exceeds maximum allowed")),
"{agent}: refused by the inference bucket, not an earlier check: {err:?}"
);
}
let bucket = swarm_queue_client::agent_permissions::open_or_create(&client)
.await
.expect("bucket");
let held =
|value: Option<bytes::Bytes>| value.map(|v| String::from_utf8_lossy(&v).into_owned());
assert_eq!(
held(bucket.get("atlas").await.expect("read")).as_deref(),
Some(r#"{"permissions":["messaging"]}"#),
"the earlier grant comes back"
);
assert_eq!(
held(bucket.get("iris").await.expect("read")),
None,
"a grant stored for a name with none is removed"
);
assert_eq!(node_count(&state.jobq), 0, "nothing may be queued");
}
/// A valid cap beside an invalid model stores neither and queues nothing,
/// on a server that could have stored the cap.
#[tokio::test]
async fn a_bad_setting_stores_none_of_them() {
let nats = crate::test_nats::TestNats::start();
let client = nats.client().await;
let state = crate::agent_limits::tests::state(&client);
let req: CreateAgentRequest = serde_json::from_str(
r#"{"name":"atlas","hive":"pr1ma","limits":{"cpu_quota":"400%"},
"inference":{"model":"-opus"}}"#,
)
.expect("body parses");
let err = super::create_agent(axum::extract::State(state.clone()), axum::Json(req))
.await
.expect_err("a bad model must be refused");
assert_eq!(
axum::response::IntoResponse::into_response(err).status(),
axum::http::StatusCode::BAD_REQUEST
);
assert_eq!(
crate::agent_limits::tests::published(&client, "atlas").await,
None
);
assert_eq!(node_count(&state.jobq), 0, "nothing may be queued");
}
/// `state_with_roster` with a second hive, `sec0nd`.
fn state_with_two_hives() -> (crate::routes::AppState, SharedSched) {
let (state, sched) = state_with_roster();
let mut hives = (*state.hives).clone();
hives.push(HiveEntry {
name: "sec0nd".to_owned(),
domain: "sec0nd.example".to_owned(),
dashboard: "sec0nd.hives.example".to_owned(),
});
let state = crate::routes::AppState {
hives: std::sync::Arc::new(hives),
..state
};
(state, sched)
}
pub(crate) async fn create(
state: &crate::routes::AppState,
name: &str,
hive: &str,
) -> Result<(), problem_details::ProblemDetails> {
super::create_agent(
axum::extract::State(state.clone()),
axum::Json(CreateAgentRequest {
name: name.to_owned(),
hive: hive.to_owned(),
limits: None,
inference: None,
permissions: None,
archetype: None,
agents_md: None,
}),
)
.await
.map(|_| ())
}
fn node_count(sched: &SharedSched) -> usize {
sched
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.graph()
.nodes()
.count()
}
/// A name queued for one hive is refused on another, by effect: 409 and
/// nothing more queued. The queued-but-unwritten placement is the one a
/// wanted-state read alone would miss.
#[tokio::test]
async fn a_name_placed_on_another_hive_is_refused_before_anything_is_queued() {
let (state, sched) = state_with_two_hives();
create(&state, "atlas", "pr1ma")
.await
.expect("the first creation of a name must be accepted");
let before = node_count(&sched);
let err = create(&state, "atlas", "sec0nd")
.await
.expect_err("the same name on another hive must be refused");
assert_eq!(
err.status,
Some(axum::http::StatusCode::CONFLICT),
"{err:?}"
);
assert!(
format!("{err:?}").contains("pr1ma"),
"the refusal should name the hive holding the name, got: {err:?}"
);
assert_eq!(
node_count(&sched),
before,
"a refused creation must queue no work"
);
}
/// The control for the refusal above: the same name on the SAME hive is
/// that agent being re-created, and queues again.
#[tokio::test]
async fn re_creating_an_agent_on_its_own_hive_is_accepted() {
let (state, sched) = state_with_two_hives();
create(&state, "atlas", "pr1ma")
.await
.expect("the first creation of a name must be accepted");
let before = node_count(&sched);
create(&state, "atlas", "pr1ma")
.await
.expect("re-creating an agent on its own hive must be accepted");
assert!(node_count(&sched) > before, "a re-creation must queue work");
}
/// A wanted state that cannot be read refuses rather than reading as
/// "placed nowhere", and queues nothing.
#[tokio::test]
async fn an_unreadable_wanted_state_refuses_creation() {
let (state, sched) = state_with_two_hives();
let state = crate::routes::AppState {
wanted: Some(std::sync::Arc::new(wanted::WantedWriter::new(
disconnected_client().await,
))),
..state
};
let err = create(&state, "atlas", "pr1ma")
.await
.expect_err("an unreadable placement must refuse");
assert_eq!(
err.status,
Some(axum::http::StatusCode::SERVICE_UNAVAILABLE),
"{err:?}"
);
assert_eq!(
node_count(&sched),
0,
"a refused creation must queue no work"
);
}
pub(crate) fn declaring(
hive: &str,
agent: &str,
state: swarm_wire::wanted::AgentState,
) -> (String, swarm_wire::wanted::HiveWanted) {
let mut declaration = swarm_wire::wanted::HiveWanted::default();
declaration
.agents
.insert(agent.to_owned(), swarm_wire::wanted::AgentWanted { state });
(hive.to_owned(), declaration)
}
/// A `SetAgentWanted` node that finishes between the two reads: whichever
/// read runs first sees it queued and undeclared, the second sees it
/// terminal and declared. Only the queue-first order catches it.
#[tokio::test]
async fn a_placement_finishing_between_the_two_reads_is_still_caught() {
use swarm_wire::wanted::AgentState;
let reads = std::cell::Cell::new(0);
let snapshot_queued = || {
reads.set(reads.get() + 1);
if reads.get() == 1 {
vec![("sec0nd".to_owned(), "atlas".to_owned())]
} else {
Vec::new()
}
};
let read_declared = || {
reads.set(reads.get() + 1);
let declared = if reads.get() == 1 {
Vec::new()
} else {
vec![declaring("sec0nd", "atlas", AgentState::Paused)]
};
std::future::ready(Ok::<_, ()>(declared))
};
let elsewhere =
super::placements_elsewhere("atlas", "pr1ma", snapshot_queued, read_declared)
.await
.expect("both reads succeed");
assert_eq!(elsewhere, ["sec0nd"]);
}
#[test]
fn a_live_declaration_on_another_hive_is_a_placement() {
use swarm_wire::wanted::AgentState;
for state in [AgentState::Up, AgentState::Offline, AgentState::Paused] {
let declared = [declaring("sec0nd", "atlas", state)];
assert_eq!(
super::placed_elsewhere("atlas", "pr1ma", &declared, &[]),
["sec0nd"],
"{state:?}"
);
}
}
/// A destroyed agent has left its hive, so creating it elsewhere moves it.
#[test]
fn a_destroyed_declaration_is_not_a_placement() {
let declared = [declaring(
"sec0nd",
"atlas",
swarm_wire::wanted::AgentState::Destroyed,
)];
assert!(super::placed_elsewhere("atlas", "pr1ma", &declared, &[]).is_empty());
}
#[test]
fn placements_on_the_requested_hive_or_of_other_names_do_not_count() {
use swarm_wire::wanted::AgentState;
let declared = [
declaring("pr1ma", "atlas", AgentState::Up),
declaring("sec0nd", "argus", AgentState::Up),
];
let queued = [
("pr1ma".to_owned(), "atlas".to_owned()),
("sec0nd".to_owned(), "argus".to_owned()),
];
assert!(super::placed_elsewhere("atlas", "pr1ma", &declared, &queued).is_empty());
}
#[test]
fn the_hive_placing_an_agent_claims_it_and_no_other_does() {
use swarm_wire::wanted::AgentState;
let declared = [
declaring("pr1ma", "atlas", AgentState::Paused),
declaring("sec0nd", "argus", AgentState::Up),
declaring("th1rd", "atlas", AgentState::Destroyed),
];
assert_eq!(super::claiming_hives("atlas", &declared), ["pr1ma"]);
assert!(super::claiming_hives("iris", &declared).is_empty());
}
#[test]
fn two_hives_placing_one_agent_both_claim_it() {
use swarm_wire::wanted::AgentState;
let declared = [
declaring("pr1ma", "atlas", AgentState::Up),
declaring("sec0nd", "atlas", AgentState::Up),
];
assert_eq!(
super::claiming_hives("atlas", &declared),
["pr1ma", "sec0nd"]
);
}
/// `refuse_placement` of `agent` on `pr1ma`, with `admin` reserved,
/// `roster` as the roster read, and `declared` as every hive's wanted
/// state, `pr1ma`'s included.
async fn refuse_on_pr1ma_with(
agent: &str,
declared: Vec<(String, swarm_wire::wanted::HiveWanted)>,
roster: Result<bool, String>,
) -> Result<(), problem_details::ProblemDetails> {
let here = declared
.iter()
.find(|(hive, _)| hive == "pr1ma")
.map(|(_, declaration)| declaration.clone());
super::refuse_placement(
agent,
"pr1ma",
here.as_ref(),
Some("admin"),
|| std::future::ready(roster),
Vec::new,
|| std::future::ready(Ok(declared)),
)
.await
}
/// [`refuse_on_pr1ma_with`] for an `agent` the roster does not hold.
async fn refuse_on_pr1ma(
agent: &str,
declared: Vec<(String, swarm_wire::wanted::HiveWanted)>,
) -> Result<(), problem_details::ProblemDetails> {
refuse_on_pr1ma_with(agent, declared, Ok(false)).await
}
const PLACING: [swarm_wire::wanted::AgentState; 3] = [
swarm_wire::wanted::AgentState::Up,
swarm_wire::wanted::AgentState::Offline,
swarm_wire::wanted::AgentState::Paused,
];
/// The handler checks exactly the placing states; a destroy is never
/// checked.
#[test]
fn only_a_destroy_places_nothing() {
for state in PLACING {
assert!(super::places_agent(state), "{state:?}");
}
assert!(!super::places_agent(
swarm_wire::wanted::AgentState::Destroyed
));
}
#[tokio::test]
async fn placing_a_new_reserved_name_is_refused() {
let err = refuse_on_pr1ma("admin", Vec::new())
.await
.expect_err("a new reserved name must be refused");
assert_eq!(
err.status,
Some(axum::http::StatusCode::BAD_REQUEST),
"{err:?}"
);
}
/// `offline` first, then `up`: the first step is a first declaration, so
/// it is refused and leaves nothing on the hive for `up` to pass as.
/// `refuse_placement` takes no state: every placing state reaches it the
/// same way, which `only_a_destroy_places_nothing` pins.
#[tokio::test]
async fn offline_first_cannot_carry_a_reserved_name_into_up() {
let err = refuse_on_pr1ma("admin", Vec::new())
.await
.expect_err("a first `offline` of a reserved name must be refused");
assert_eq!(
err.status,
Some(axum::http::StatusCode::BAD_REQUEST),
"{err:?}"
);
// Nothing was declared, so `up` is a first declaration as well.
let err = refuse_on_pr1ma("admin", Vec::new())
.await
.expect_err("`up` after a refused `offline` must be refused");
assert_eq!(
err.status,
Some(axum::http::StatusCode::BAD_REQUEST),
"{err:?}"
);
}
/// No later step refuses a bad name, so an unreadable roster refuses
/// rather than lets it through. An ordinary name never reads it.
#[tokio::test]
async fn an_unreadable_roster_refuses_a_rule_breaking_name() {
let unreadable = || Err("no identity bridge is configured".to_owned());
let err = refuse_on_pr1ma_with("admin", Vec::new(), unreadable())
.await
.expect_err("an unreadable roster must not let a reserved name through");
assert_eq!(
err.status,
Some(axum::http::StatusCode::SERVICE_UNAVAILABLE),
"{err:?}"
);
refuse_on_pr1ma_with("atlas", Vec::new(), unreadable())
.await
.expect("an ordinary name does not depend on the roster");
}
/// A destroyed agent is gone from its hive, so declaring it again is a
/// first declaration and its name is checked.
#[tokio::test]
async fn a_destroyed_declaration_here_does_not_exempt_the_name() {
use swarm_wire::wanted::AgentState;
let err = refuse_on_pr1ma(
"admin",
vec![declaring("pr1ma", "admin", AgentState::Destroyed)],
)
.await
.expect_err("re-placing a destroyed reserved name must be refused");
assert_eq!(
err.status,
Some(axum::http::StatusCode::BAD_REQUEST),
"{err:?}"
);
}
/// An agent already declared here keeps a rule-breaking name through any
/// state change, whether or not the roster holds it or can be read.
#[tokio::test]
async fn an_agent_declared_here_is_not_refused_over_its_name() {
for current in PLACING {
for roster in [Ok(false), Err("bridge down".to_owned())] {
refuse_on_pr1ma_with(
"admin",
vec![declaring("pr1ma", "admin", current)],
roster.clone(),
)
.await
.unwrap_or_else(|err| panic!("{current:?}, {roster:?}: {err:?}"));
}
}
}
#[tokio::test]
async fn a_first_declaration_of_a_name_placed_on_another_hive_is_refused() {
use swarm_wire::wanted::AgentState;
let err = refuse_on_pr1ma("atlas", vec![declaring("sec0nd", "atlas", AgentState::Up)])
.await
.expect_err("a name placed on another hive must be refused");
assert_eq!(
err.status,
Some(axum::http::StatusCode::CONFLICT),
"{err:?}"
);
assert!(format!("{err:?}").contains("sec0nd"), "{err:?}");
}
/// An agent that runs on `pr1ma` but was never declared anywhere, and is
/// not in the roster, is one the swarm adopts by declaring it.
#[tokio::test]
async fn adopting_an_undeclared_agent_on_its_own_hive_is_accepted() {
refuse_on_pr1ma("legacy", Vec::new())
.await
.expect("an undeclared agent must still be adoptable");
}
#[tokio::test]
async fn changing_the_state_of_an_agent_on_its_own_hive_is_accepted() {
use swarm_wire::wanted::AgentState;
refuse_on_pr1ma("atlas", vec![declaring("pr1ma", "atlas", AgentState::Up)])
.await
.expect("an agent's own hive must accept a new state for it");
}
/// `set_agent_state` on `pr1ma`, in a two-hive swarm whose queue never
/// connected and with no identity bridge.
async fn set_on_disconnected_pr1ma(
agent: &str,
wanted_state: swarm_wire::wanted::AgentState,
) -> problem_details::ProblemDetails {
let (state, _sched) = state_with_two_hives();
let state = crate::routes::AppState {
wanted: Some(std::sync::Arc::new(wanted::WantedWriter::new(
disconnected_client().await,
))),
..state
};
super::set_agent_state(
axum::extract::State(state),
axum::extract::Path(("pr1ma".to_owned(), agent.to_owned())),
axum::Json(SetAgentStateRequest {
state: wanted_state,
}),
)
.await
.expect_err("a queue that never connected cannot publish")
}
/// Through the handler: a wanted state that cannot be read refuses the
/// declaration before it is written, as it refuses creation.
#[tokio::test]
async fn set_agent_state_refuses_when_the_wanted_state_is_unreadable() {
let err = set_on_disconnected_pr1ma("atlas", swarm_wire::wanted::AgentState::Up).await;
assert_eq!(
err.status,
Some(axum::http::StatusCode::SERVICE_UNAVAILABLE),
"{err:?}"
);
assert!(
format!("{err:?}").contains("was not placed"),
"the refusal must come from the placement check, got: {err:?}"
);
}
/// Through the handler: a destroy of a reserved name reaches the write
/// without being checked. It fails there only because the queue never
/// connected; the placing control above fails in the check instead.
#[tokio::test]
async fn set_agent_state_does_not_check_a_destroy() {
let err =
set_on_disconnected_pr1ma("admin", swarm_wire::wanted::AgentState::Destroyed).await;
assert_eq!(
err.status,
Some(axum::http::StatusCode::SERVICE_UNAVAILABLE),
"{err:?}"
);
assert!(
!format!("{err:?}").contains("not placed"),
"a destroy must not be refused by the placement check, got: {err:?}"
);
}
/// Forgejo v16.0.5's reserved usernames our charset can spell, plus the
/// bare `-` — the forge section of `nix/reserved-names.nix`. Read from
/// the list nix exports, so dropping one from the file reds this.
const FORGE_REFUSED_LITERALS: [&str; 23] = [
"-",
"admin",
"api",
"assets",
"attachments",
"avatar",
"avatars",
"captcha",
"explore",
"forgejo-actions",
"ghost",
"gitea-actions",
"issues",
"login",
"metrics",
"milestones",
"notifications",
"org",
"pulls",
"repo",
"repo-avatars",
"user",
"v2",
];
/// The list nix renders, panicking when absent for the reason
/// `hive-sh4re`'s drift test gives: a skip would read as green.
fn nix_reserved() -> String {
hive_types::reserved_names_raw()
.expect("HIVE_RESERVED_NAMES is exported by nix/checks.nix and nix/devshell.nix")
}
#[test]
fn a_new_agent_named_a_forge_refused_literal_is_refused() {
let raw = nix_reserved();
for name in FORGE_REFUSED_LITERALS {
let mut warnings = Vec::new();
let broken = super::broken_name_rules(name, Some(&raw), &mut warnings);
let err = super::name_verdict(name, &broken, Ok(false), &mut warnings).expect_err(name);
assert!(err.contains("nix/reserved-names.nix"), "{name:?}: {err}");
}
}
#[test]
fn a_new_agent_named_like_a_service_forge_account_is_refused() {
let raw = nix_reserved();
for name in [
crate::forge::builder_account::LOGIN,
crate::forge::classifier_account::LOGIN,
crate::forge::ci_metrics_account::LOGIN,
] {
let mut warnings = Vec::new();
let broken = super::broken_name_rules(name, Some(&raw), &mut warnings);
let err = super::name_verdict(name, &broken, Ok(false), &mut warnings).expect_err(name);
assert!(err.contains("nix/reserved-names.nix"), "{name:?}: {err}");
}
}
/// One case per shape rule, each named in the refusal. Checked against a
/// list without them, so the literal half cannot carry this test.
#[test]
fn a_new_agent_breaking_a_forge_shape_rule_is_refused() {
let over = "a".repeat(hive_types::FORGE_USERNAME_MAX_LEN + 1);
for (name, rule) in [
("-agent", "must not start with '-'"),
("my--agent", "must not contain '--'"),
("agent-", "must not end with '-'"),
(over.as_str(), "must be 40 characters or fewer"),
] {
let mut warnings = Vec::new();
let broken = super::broken_name_rules(name, Some("operator"), &mut warnings);
let err = super::name_verdict(name, &broken, Ok(false), &mut warnings).expect_err(name);
assert!(err.contains(rule), "{name:?}: {err}");
}
}
/// The control for both tests above, against the real list.
#[test]
fn an_ordinary_name_and_a_40_char_name_break_no_rule() {
let raw = nix_reserved();
let at_cap = "a".repeat(hive_types::FORGE_USERNAME_MAX_LEN);
for name in ["my-agent", at_cap.as_str()] {
let mut warnings = Vec::new();
let broken = super::broken_name_rules(name, Some(&raw), &mut warnings);
assert!(broken.is_empty(), "{name:?}: {broken:?}");
assert!(warnings.is_empty(), "{name:?}: {warnings:?}");
}
}
/// The `hive-` prefix belongs to hives' forge accounts, so a new agent
/// may not take it. `hivemind` and `xhive-y` are the control.
#[test]
fn a_new_agent_with_the_hive_prefix_is_refused() {
let raw = nix_reserved();
let mut warnings = Vec::new();
let broken = super::broken_name_rules("hive-x", Some(&raw), &mut warnings);
let err =
super::name_verdict("hive-x", &broken, Ok(false), &mut warnings).expect_err("hive-x");
assert!(err.contains("reserved for hive forge accounts"), "{err}");
for ok in ["hivemind", "xhive-y"] {
let broken = super::broken_name_rules(ok, Some(&raw), &mut warnings);
assert!(broken.is_empty(), "{ok:?}: {broken:?}");
}
assert!(warnings.is_empty(), "{warnings:?}");
}
/// The prefix the rule reserves is the one hive accounts are named with.
#[test]
fn the_reserved_prefix_is_the_hive_account_prefix() {
assert!(
swarm_secret_client::forge::hive_login("alpha")
.starts_with(hive_types::RESERVED_AGENT_PREFIX)
);
}
/// Re-creating an agent under its own name must keep working: a name
/// already in the roster is warned about, never refused.
#[test]
fn an_existing_agent_keeps_a_rule_breaking_name_with_a_warning() {
let mut warnings = Vec::new();
let broken = super::broken_name_rules("admin", Some("admin"), &mut warnings);
super::name_verdict("admin", &broken, Ok(true), &mut warnings)
.expect("an existing agent must not be refused");
assert_eq!(warnings.len(), 1, "{warnings:?}");
assert!(warnings[0].contains("already exists"), "{warnings:?}");
}
/// A roster that cannot be read cannot tell new from existing, so it
/// warns — and the warning says the check did not decide.
#[test]
fn an_unreadable_roster_warns_rather_than_refusing() {
let mut warnings = Vec::new();
let broken = super::broken_name_rules("agent-", Some("operator"), &mut warnings);
super::name_verdict(
"agent-",
&broken,
Err("bridge down".to_owned()),
&mut warnings,
)
.expect("an unreadable roster must not refuse");
assert_eq!(warnings.len(), 1, "{warnings:?}");
assert!(warnings[0].contains("bridge down"), "{warnings:?}");
}
/// `Destroyed` is the one declaration that ends an agent's life, so it is
/// the one that revokes. The credential is a bearer secret: left behind,
/// anything that recovers it still authenticates as an agent that no
/// longer exists.
#[test]
fn declaring_an_agent_destroyed_revokes_its_queue_credential() {
assert!(super::revokes_queue_credential(
swarm_wire::wanted::AgentState::Destroyed
));
}
/// The control, and the more expensive half to get wrong: a revocation on
/// `Offline` or `Paused` would give an agent that can be stopped and never
/// restarted. Every non-terminal state is listed rather than sampled, so
/// a state added later shows up here as well as in the match itself.
#[test]
fn no_state_an_agent_comes_back_from_revokes() {
use swarm_wire::wanted::AgentState;
for state in [AgentState::Up, AgentState::Offline, AgentState::Paused] {
assert!(
!super::revokes_queue_credential(state),
"{state:?} is a state an agent returns from"
);
}
}
/// Seed the stored archetype list with `reviewer` under `hyperhive`.
async fn seed_archetypes(client: &async_nats::Client) {
let record = swarm_queue_client::archetypes::SourceRecord {
source: "/nix/store/h".to_owned(),
archetypes: std::collections::BTreeMap::from([(
"reviewer".to_owned(),
swarm_queue_client::archetypes::ArchetypeMeta {
description: "reviews".to_owned(),
..Default::default()
},
)]),
refused: std::collections::BTreeMap::new(),
};
swarm_queue_client::archetypes::open_or_create(client)
.await
.expect("bucket")
.put(
swarm_queue_client::archetypes::BUILTIN_SOURCE,
serde_json::to_vec(&record).expect("encodes").into(),
)
.await
.expect("seeded");
}
/// Through the real handler and a real `nats-server`: a listed archetype
/// lands under the agent's key as the hive reads it, and the queued graph
/// has no config repo.
#[tokio::test]
async fn a_creation_from_an_archetype_records_it_and_queues_no_repo() {
use hive_jobq_wire::WireNode as _;
let nats = crate::test_nats::TestNats::start();
let client = nats.client().await;
seed_archetypes(&client).await;
let state = crate::agent_limits::tests::state(&client);
let req: CreateAgentRequest = serde_json::from_str(
r#"{"name":"atlas","hive":"pr1ma",
"archetype":{"source":"hyperhive","name":"reviewer"}}"#,
)
.expect("body parses");
let _queued = super::create_agent(axum::extract::State(state.clone()), axum::Json(req))
.await
.expect("a listed archetype goes through");
let stored = swarm_queue_client::agent_archetype::open_or_create(&client)
.await
.expect("bucket")
.get("atlas")
.await
.expect("read")
.expect("a value");
assert_eq!(&stored[..], br#"{"source":"hyperhive","name":"reviewer"}"#);
let labels: Vec<String> = state
.jobq
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.graph()
.nodes()
.map(|n| n.payload.label())
.collect();
assert!(labels.iter().any(|l| l == "trigger_deploy"), "{labels:?}");
assert!(!labels.iter().any(|l| l == "create_repo"), "{labels:?}");
}
/// An archetype the stored list does not hold is a 400 that stores and
/// queues nothing, and so is a name that is not an identifier.
#[tokio::test]
async fn an_unlisted_archetype_is_refused_before_anything_is_stored() {
let nats = crate::test_nats::TestNats::start();
let client = nats.client().await;
seed_archetypes(&client).await;
let state = crate::agent_limits::tests::state(&client);
for archetype in [
r#"{"source":"hyperhive","name":"nope"}"#,
r#"{"source":"elsewhere","name":"reviewer"}"#,
r#"{"source":"hyperhive","name":"a.b"}"#,
] {
let req: CreateAgentRequest = serde_json::from_str(&format!(
r#"{{"name":"atlas","hive":"pr1ma","permissions":["messaging"],
"archetype":{archetype}}}"#
))
.expect("body parses");
let err = super::create_agent(axum::extract::State(state.clone()), axum::Json(req))
.await
.expect_err("an unlisted archetype must be refused");
assert_eq!(
err.status,
Some(axum::http::StatusCode::BAD_REQUEST),
"{archetype}: {err:?}"
);
}
assert_eq!(node_count(&state.jobq), 0, "nothing may be queued");
let permissions = swarm_queue_client::agent_permissions::open_or_create(&client)
.await
.expect("bucket");
assert_eq!(permissions.get("atlas").await.expect("read"), None);
}
/// A client that exists but has never connected — same shape
/// `matrix_account`'s tests build, duplicated here rather than shared:
/// both are a handful of lines and neither crate has a test-support
/// module to put a shared helper in.
async fn disconnected_client() -> async_nats::Client {
async_nats::ConnectOptions::new()
.retry_on_initial_connect()
.connect("127.0.0.1:1")
.await
.expect("retry_on_initial_connect returns without waiting for a real connection")
}
/// `state_with_roster` plus a `wanted` writer over a client that never
/// connected — the shape `set_agent_state`/`get_hive_wanted` see when
/// the queue is mid-reconnect.
async fn state_with_disconnected_wanted() -> crate::routes::AppState {
let (state, _sched) = state_with_roster();
crate::routes::AppState {
wanted: Some(std::sync::Arc::new(wanted::WantedWriter::new(
disconnected_client().await,
))),
..state
}
}
/// End to end through the real handler: `PUT .../state` against a
/// disconnected queue answers 503, not the 500 every other writer
/// error still gets.
#[tokio::test]
async fn set_agent_state_answers_503_on_a_disconnected_queue() {
let state = state_with_disconnected_wanted().await;
let result = super::set_agent_state(
axum::extract::State(state),
axum::extract::Path(("pr1ma".to_owned(), "atlas".to_owned())),
axum::Json(SetAgentStateRequest {
state: swarm_wire::wanted::AgentState::Paused,
}),
)
.await;
let problem = result.expect_err("a disconnected queue must not read as success");
assert_eq!(
problem.status,
Some(axum::http::StatusCode::SERVICE_UNAVAILABLE),
"{problem:?}"
);
}
/// Same fix, the read side: `GET .../wanted` against a disconnected
/// queue also answers 503.
#[tokio::test]
async fn get_hive_wanted_answers_503_on_a_disconnected_queue() {
let state = state_with_disconnected_wanted().await;
let result = super::get_hive_wanted(
axum::extract::State(state),
axum::extract::Path("pr1ma".to_owned()),
)
.await;
let problem = result.expect_err("a disconnected queue must not read as success");
assert_eq!(
problem.status,
Some(axum::http::StatusCode::SERVICE_UNAVAILABLE),
"{problem:?}"
);
}
/// Control for both tests above: a hive that is not in the roster is
/// still a plain 400, even with the same disconnected writer wired up —
/// so the 503 above is `ensure_connected` firing, not every error this
/// handler can produce collapsing to 503.
#[tokio::test]
async fn set_agent_state_still_answers_400_for_an_unknown_hive() {
let state = state_with_disconnected_wanted().await;
let result = super::set_agent_state(
axum::extract::State(state),
axum::extract::Path(("not-a-hive".to_owned(), "atlas".to_owned())),
axum::Json(SetAgentStateRequest {
state: swarm_wire::wanted::AgentState::Paused,
}),
)
.await;
let problem = result.expect_err("an unknown hive is still a 400");
assert_eq!(
problem.status,
Some(axum::http::StatusCode::BAD_REQUEST),
"{problem:?}"
);
}
/// `GET /api/agents/{name}/icon` against a disconnected queue answers
/// 503 like its siblings, not the 404 an `<img>` would show identically
/// or the 500 of a failed read.
#[tokio::test]
async fn get_agent_icon_answers_503_on_a_disconnected_queue() {
let (state, _sched) = state_with_roster();
let state = crate::routes::AppState {
agent_icons: Some(std::sync::Arc::new(
crate::agent_icon::AgentIconReader::new(disconnected_client().await),
)),
..state
};
let problem = super::get_agent_icon(
axum::extract::State(state),
axum::extract::Path("iris".to_owned()),
)
.await
.expect_err("a disconnected queue must not read as an icon");
assert_eq!(
problem.status,
Some(axum::http::StatusCode::SERVICE_UNAVAILABLE),
"{problem:?}"
);
}
}