archetypes: create an agent from an archetype with no config repo
POST /api/agents takes an archetype the stored list holds, records it in the new agent-archetype bucket, and queues no CreateRepo, AddRepoMember or InitAgentConfigRepo for it. A destroy clears the key. Hives may read the bucket. Refs #5386
This commit is contained in:
parent
6b93839290
commit
b199c83899
7 changed files with 645 additions and 37 deletions
171
swarm-controller/src/agent_archetype.rs
Normal file
171
swarm-controller/src/agent_archetype.rs
Normal file
|
|
@ -0,0 +1,171 @@
|
||||||
|
//! The archetype an agent was created from: one key per archetype agent in
|
||||||
|
//! `agent-archetype` (`swarm_queue_client::agent_archetype`), which this
|
||||||
|
//! daemon alone writes. `POST /api/agents` checks the choice against the
|
||||||
|
//! `agent-archetypes` record and stores it before it queues the creation; a
|
||||||
|
//! destroy deletes it. The hive reads it to build the agent from the
|
||||||
|
//! archetype's module, with no config repo.
|
||||||
|
|
||||||
|
use problem_details::ProblemDetails;
|
||||||
|
use swarm_queue_client::agent_archetype::{self, ArchetypeRef};
|
||||||
|
use swarm_queue_client::archetypes::SourceRecord;
|
||||||
|
|
||||||
|
use crate::schedules::{queue, read_or_default, write_revision};
|
||||||
|
use crate::{AppState, error_problem};
|
||||||
|
|
||||||
|
/// How many times a losing writer re-reads and re-applies, as in
|
||||||
|
/// `wanted::MAX_ATTEMPTS`.
|
||||||
|
const MAX_ATTEMPTS: usize = 5;
|
||||||
|
|
||||||
|
fn queue_problem(e: &anyhow::Error) -> ProblemDetails {
|
||||||
|
tracing::warn!(error = %format!("{e:#}"), "agent archetype: queue operation failed");
|
||||||
|
error_problem(
|
||||||
|
axum::http::StatusCode::INTERNAL_SERVER_ERROR,
|
||||||
|
&format!("{e:#}"),
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Refuse `choice` unless the stored `agent-archetypes` record lists it.
|
||||||
|
pub(crate) async fn check(
|
||||||
|
client: &async_nats::Client,
|
||||||
|
choice: &ArchetypeRef,
|
||||||
|
) -> Result<(), ProblemDetails> {
|
||||||
|
let record = match swarm_queue_client::archetypes::open(client)
|
||||||
|
.await
|
||||||
|
.map_err(|e| queue_problem(&e.into()))?
|
||||||
|
{
|
||||||
|
Some(store) => {
|
||||||
|
read_or_default::<SourceRecord>(&store, &choice.source)
|
||||||
|
.await
|
||||||
|
.map_err(|e| queue_problem(&e))?
|
||||||
|
.0
|
||||||
|
}
|
||||||
|
None => SourceRecord::default(),
|
||||||
|
};
|
||||||
|
listed(&record, choice)
|
||||||
|
.map_err(|why| error_problem(axum::http::StatusCode::BAD_REQUEST, &why))
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Whether `record`, the stored record of `choice.source`, lists `choice`.
|
||||||
|
fn listed(record: &SourceRecord, choice: &ArchetypeRef) -> Result<(), String> {
|
||||||
|
if record.archetypes.contains_key(&choice.name) {
|
||||||
|
return Ok(());
|
||||||
|
}
|
||||||
|
let ArchetypeRef { source, name } = choice;
|
||||||
|
Err(match record.refused.get(name) {
|
||||||
|
Some(why) => format!("archetype {name:?} of {source:?} was refused: {why}"),
|
||||||
|
None => format!("{source:?} has no archetype {name:?}; GET /api/archetypes lists them"),
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
/// What [`store`] replaced, for [`restore`]; `None` when it wrote nothing.
|
||||||
|
pub(crate) type Replaced = Option<crate::schedules::Replaced<Option<ArchetypeRef>>>;
|
||||||
|
|
||||||
|
/// Record that `agent` is built from `choice`.
|
||||||
|
pub(crate) async fn store(
|
||||||
|
client: &async_nats::Client,
|
||||||
|
agent: &str,
|
||||||
|
choice: &ArchetypeRef,
|
||||||
|
) -> Result<Replaced, ProblemDetails> {
|
||||||
|
let store = agent_archetype::open_or_create(client)
|
||||||
|
.await
|
||||||
|
.map_err(|e| queue_problem(&e.into()))?;
|
||||||
|
let value = Some(choice.clone());
|
||||||
|
for _ in 0..MAX_ATTEMPTS {
|
||||||
|
let (current, revision, undecodable) =
|
||||||
|
read_or_default::<Option<ArchetypeRef>>(&store, agent)
|
||||||
|
.await
|
||||||
|
.map_err(|e| queue_problem(&e))?;
|
||||||
|
if current == value && !undecodable {
|
||||||
|
tracing::info!(%agent, ?choice, "set agent archetype at create");
|
||||||
|
return Ok(None);
|
||||||
|
}
|
||||||
|
if let Some(written) = write_revision(&store, agent, &value, revision)
|
||||||
|
.await
|
||||||
|
.map_err(|e| queue_problem(&e))?
|
||||||
|
{
|
||||||
|
tracing::info!(%agent, ?choice, "set agent archetype at create");
|
||||||
|
return Ok(Some(crate::schedules::Replaced::new(
|
||||||
|
current.map(Some),
|
||||||
|
written,
|
||||||
|
)));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
Err(queue_problem(&anyhow::anyhow!(
|
||||||
|
"gave up writing the archetype of {agent} after {MAX_ATTEMPTS} conflicting writes"
|
||||||
|
)))
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Undo what [`store`] did, unless another write has landed since. Best
|
||||||
|
/// effort: a failure is logged, since the caller is already on its way to an
|
||||||
|
/// error of its own.
|
||||||
|
pub(crate) async fn restore(client: &async_nats::Client, agent: &str, replaced: Replaced) {
|
||||||
|
let Some(replaced) = replaced else { return };
|
||||||
|
let undone = match agent_archetype::open_or_create(client).await {
|
||||||
|
Ok(store) => replaced.undo(&store, agent).await,
|
||||||
|
Err(e) => Err(anyhow::anyhow!("{e:?}")),
|
||||||
|
};
|
||||||
|
match undone {
|
||||||
|
Ok(true) => {}
|
||||||
|
Ok(false) => {
|
||||||
|
tracing::info!(%agent, "left the agent's archetype: another write landed since");
|
||||||
|
}
|
||||||
|
Err(e) => {
|
||||||
|
tracing::warn!(%agent, error = %format!("{e:#}"), "could not put back the agent's archetype");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Delete a destroyed agent's key, so the next agent of that name is not
|
||||||
|
/// built from this one's archetype. A failure is logged and does not stop the
|
||||||
|
/// destroy, as in `agent_permissions::clear_on_destroy`.
|
||||||
|
pub(crate) async fn clear_on_destroy(state: &AppState, hive: &str, agent: &str) {
|
||||||
|
let cleared = async {
|
||||||
|
let client = queue(state).map_err(|(_, why)| anyhow::anyhow!(why))?;
|
||||||
|
agent_archetype::open_or_create(&client)
|
||||||
|
.await?
|
||||||
|
.delete(agent)
|
||||||
|
.await?;
|
||||||
|
anyhow::Ok(())
|
||||||
|
}
|
||||||
|
.await;
|
||||||
|
match cleared {
|
||||||
|
Ok(()) => tracing::info!(hive, agent, "destroyed agent's archetype cleared"),
|
||||||
|
Err(e) => tracing::error!(
|
||||||
|
hive,
|
||||||
|
agent,
|
||||||
|
error = %format!("{e:#}"),
|
||||||
|
"clearing this agent's archetype FAILED; it was destroyed anyway, so a new agent \
|
||||||
|
of the same name is built from this archetype. Declaring the agent destroyed \
|
||||||
|
again re-runs this."
|
||||||
|
),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
use std::collections::BTreeMap;
|
||||||
|
use swarm_queue_client::archetypes::ArchetypeMeta;
|
||||||
|
|
||||||
|
fn choice(name: &str) -> ArchetypeRef {
|
||||||
|
ArchetypeRef {
|
||||||
|
source: "hyperhive".to_owned(),
|
||||||
|
name: name.to_owned(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn only_a_listed_archetype_passes() {
|
||||||
|
let record = SourceRecord {
|
||||||
|
source: "/nix/store/h".to_owned(),
|
||||||
|
archetypes: BTreeMap::from([("reviewer".to_owned(), ArchetypeMeta::default())]),
|
||||||
|
refused: BTreeMap::from([("broken".to_owned(), "limits: bad".to_owned())]),
|
||||||
|
};
|
||||||
|
assert_eq!(listed(&record, &choice("reviewer")), Ok(()));
|
||||||
|
let refused = listed(&record, &choice("broken")).unwrap_err();
|
||||||
|
assert!(refused.contains("limits: bad"), "{refused}");
|
||||||
|
let unknown = listed(&record, &choice("nope")).unwrap_err();
|
||||||
|
assert!(unknown.contains("no archetype \"nope\""), "{unknown}");
|
||||||
|
listed(&SourceRecord::default(), &choice("reviewer")).unwrap_err();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
@ -797,6 +797,7 @@ mod tests {
|
||||||
limits: None,
|
limits: None,
|
||||||
inference: None,
|
inference: None,
|
||||||
permissions,
|
permissions,
|
||||||
|
archetype: None,
|
||||||
}),
|
}),
|
||||||
)
|
)
|
||||||
};
|
};
|
||||||
|
|
@ -857,6 +858,7 @@ mod tests {
|
||||||
limits: None,
|
limits: None,
|
||||||
inference: None,
|
inference: None,
|
||||||
permissions: None,
|
permissions: None,
|
||||||
|
archetype: None,
|
||||||
}),
|
}),
|
||||||
)
|
)
|
||||||
.await
|
.await
|
||||||
|
|
|
||||||
|
|
@ -42,6 +42,7 @@ use swarm_authelia_bridge_sock::BridgeResponse;
|
||||||
use utoipa::{OpenApi, ToSchema};
|
use utoipa::{OpenApi, ToSchema};
|
||||||
use utoipa_axum::{router::OpenApiRouter, routes};
|
use utoipa_axum::{router::OpenApiRouter, routes};
|
||||||
|
|
||||||
|
mod agent_archetype;
|
||||||
mod agent_control;
|
mod agent_control;
|
||||||
mod agent_icon;
|
mod agent_icon;
|
||||||
mod agent_identity;
|
mod agent_identity;
|
||||||
|
|
@ -1455,12 +1456,13 @@ async fn set_agent_state(
|
||||||
|
|
||||||
// After the declaration and never before it: the write above is the
|
// 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
|
// destroy order, and until it lands the agent is not being torn down, so
|
||||||
// its credential and permissions have to keep working. Sequenced rather
|
// its credential, permissions and archetype have to keep working. Sequenced
|
||||||
// than spawned so both attempts are finished — and their outcomes
|
// rather than spawned so every attempt is finished — and their outcomes
|
||||||
// logged — by the time the operator's call returns.
|
// logged — by the time the operator's call returns.
|
||||||
if revokes_queue_credential(req.state) {
|
if revokes_queue_credential(req.state) {
|
||||||
revoke_or_complain(&hive, &agent).await;
|
revoke_or_complain(&hive, &agent).await;
|
||||||
agent_permissions::clear_on_destroy(&state, &hive, &agent).await;
|
agent_permissions::clear_on_destroy(&state, &hive, &agent).await;
|
||||||
|
agent_archetype::clear_on_destroy(&state, &hive, &agent).await;
|
||||||
}
|
}
|
||||||
Ok(Json(render(&declaration)))
|
Ok(Json(render(&declaration)))
|
||||||
}
|
}
|
||||||
|
|
@ -1793,6 +1795,12 @@ struct CreateAgentRequest {
|
||||||
#[serde(default)]
|
#[serde(default)]
|
||||||
#[schema(value_type = Option<Vec<String>>)]
|
#[schema(value_type = Option<Vec<String>>)]
|
||||||
permissions: Option<Vec<hive_sh4re::permissions::Permission>>,
|
permissions: Option<Vec<hive_sh4re::permissions::Permission>>,
|
||||||
|
/// The archetype to build the agent from, one `GET /api/archetypes`
|
||||||
|
/// lists. The agent then gets no `agent-configs/<name>` repo: its hive
|
||||||
|
/// builds it from the archetype's module. Absent is a custom agent.
|
||||||
|
#[serde(default)]
|
||||||
|
#[schema(value_type = Option<Object>, example = json!({"source": "hyperhive", "name": "reviewer"}))]
|
||||||
|
archetype: Option<swarm_queue_client::agent_archetype::ArchetypeRef>,
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Where the queued job landed — a caller polls `/api/jobq/graph` (or
|
/// Where the queued job landed — a caller polls `/api/jobq/graph` (or
|
||||||
|
|
@ -1827,9 +1835,11 @@ struct CreateAgentResponse {
|
||||||
/// validation error, not an idempotent no-op. `InitAgentConfigRepo` has no
|
/// validation error, not an idempotent no-op. `InitAgentConfigRepo` has no
|
||||||
/// such second dependency: seeding files uses this daemon's own forge
|
/// 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
|
/// token, never the agent's, so it only ever needed the repo (mara, design
|
||||||
/// review: it's a jobq graph, not a linear chain). Returns as soon as the
|
/// review: it's a jobq graph, not a linear chain). An agent created from an
|
||||||
/// graph is inserted — **not** once any of it has run; `run_swarm_node`
|
/// `archetype` has no config repo, so its graph has none of the three repo
|
||||||
/// does that asynchronously off the scheduler loop already running
|
/// 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
|
/// (`spawn_jobq_worker`). The response reports `CreateIdentity`'s id, one
|
||||||
/// of the graph's two roots — a caller watches the whole thing settle via
|
/// 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.
|
/// `/api/jobq/graph`, which serves every root, not just this one.
|
||||||
|
|
@ -1846,10 +1856,10 @@ struct CreateAgentResponse {
|
||||||
request_body = CreateAgentRequest,
|
request_body = CreateAgentRequest,
|
||||||
responses(
|
responses(
|
||||||
(status = 200, description = "job chain queued", body = CreateAgentResponse),
|
(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, or a permission is listed twice (problem+json)", body = String),
|
(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, or `archetype` is not one the stored archetype list holds (problem+json)", body = String),
|
||||||
(status = 409, description = "the swarm has already placed `name` on a different hive (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 = 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` or `permissions` is set and cannot be stored (problem+json)", body = String),
|
(status = 503, description = "the swarm queue is not connected, so whether `name` is placed on another hive is unknown, or `limits`, `inference`, `permissions` or `archetype` 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),
|
(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"
|
tag = "agents"
|
||||||
|
|
@ -1864,7 +1874,7 @@ async fn create_agent(
|
||||||
|
|
||||||
// Settings are checked before anything is created, so a bad value stops
|
// Settings are checked before anything is created, so a bad value stops
|
||||||
// the request rather than leaving an agent without them.
|
// the request rather than leaving an agent without them.
|
||||||
let settings = CreateSettings::parse(req.limits, req.inference, req.permissions)
|
let settings = CreateSettings::parse(req.limits, req.inference, req.permissions, req.archetype)
|
||||||
.map_err(|why| error_problem(axum::http::StatusCode::BAD_REQUEST, &why))?;
|
.map_err(|why| error_problem(axum::http::StatusCode::BAD_REQUEST, &why))?;
|
||||||
|
|
||||||
// Two checks, and the second is the one that makes the field worth
|
// Two checks, and the second is the one that makes the field worth
|
||||||
|
|
@ -1928,11 +1938,7 @@ async fn create_agent(
|
||||||
}
|
}
|
||||||
// Checked after the roster, so a mistyped hive stays a 400 while the queue
|
// Checked after the roster, so a mistyped hive stays a 400 while the queue
|
||||||
// is down.
|
// is down.
|
||||||
let settings_queue = if settings.any() {
|
let settings_queue = settings.queue(&state).await?;
|
||||||
Some(schedules::queue(&state).map_err(schedules::refused)?)
|
|
||||||
} else {
|
|
||||||
None
|
|
||||||
};
|
|
||||||
|
|
||||||
// Agent names are one swarm-wide namespace: identity, forge user, store
|
// 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
|
// path and matrix id all carry the bare name. The same name on the same
|
||||||
|
|
@ -1965,6 +1971,7 @@ async fn create_agent(
|
||||||
settings.limits.as_ref(),
|
settings.limits.as_ref(),
|
||||||
settings.inference.as_ref(),
|
settings.inference.as_ref(),
|
||||||
settings.permissions.as_ref(),
|
settings.permissions.as_ref(),
|
||||||
|
settings.archetype.as_ref(),
|
||||||
)
|
)
|
||||||
.await?
|
.await?
|
||||||
}
|
}
|
||||||
|
|
@ -1975,7 +1982,9 @@ async fn create_agent(
|
||||||
.jobq
|
.jobq
|
||||||
.lock()
|
.lock()
|
||||||
.unwrap_or_else(std::sync::PoisonError::into_inner);
|
.unwrap_or_else(std::sync::PoisonError::into_inner);
|
||||||
sched.insert_job(None, |b| declare_agent_job(b, &agent, &hive))
|
sched.insert_job(None, |b| {
|
||||||
|
declare_agent_job(b, &agent, &hive, settings.archetype.is_some())
|
||||||
|
})
|
||||||
};
|
};
|
||||||
let ids = match queued {
|
let ids = match queued {
|
||||||
Ok(ids) => ids,
|
Ok(ids) => ids,
|
||||||
|
|
@ -2003,27 +2012,55 @@ struct CreateSettings {
|
||||||
limits: Option<swarm_queue_client::agent_limits::AgentCaps>,
|
limits: Option<swarm_queue_client::agent_limits::AgentCaps>,
|
||||||
inference: Option<swarm_queue_client::agent_inference::AgentInference>,
|
inference: Option<swarm_queue_client::agent_inference::AgentInference>,
|
||||||
permissions: Option<swarm_queue_client::agent_permissions::AgentPermissions>,
|
permissions: Option<swarm_queue_client::agent_permissions::AgentPermissions>,
|
||||||
|
archetype: Option<swarm_queue_client::agent_archetype::ArchetypeRef>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl CreateSettings {
|
impl CreateSettings {
|
||||||
/// Each setting given, or why its `PUT` would refuse the first bad one.
|
/// 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(
|
fn parse(
|
||||||
limits: Option<agent_limits::Caps>,
|
limits: Option<agent_limits::Caps>,
|
||||||
inference: Option<inference::InferenceSettings>,
|
inference: Option<inference::InferenceSettings>,
|
||||||
permissions: Option<Vec<hive_sh4re::permissions::Permission>>,
|
permissions: Option<Vec<hive_sh4re::permissions::Permission>>,
|
||||||
|
archetype: Option<swarm_queue_client::agent_archetype::ArchetypeRef>,
|
||||||
) -> Result<Self, String> {
|
) -> Result<Self, String> {
|
||||||
|
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 {
|
Ok(Self {
|
||||||
limits: limits.map(agent_limits::Caps::into_stored).transpose()?,
|
limits: limits.map(agent_limits::Caps::into_stored).transpose()?,
|
||||||
inference: inference
|
inference: inference
|
||||||
.map(inference::InferenceSettings::into_stored)
|
.map(inference::InferenceSettings::into_stored)
|
||||||
.transpose()?,
|
.transpose()?,
|
||||||
permissions: permissions.map(agent_permissions::at_create).transpose()?,
|
permissions: permissions.map(agent_permissions::at_create).transpose()?,
|
||||||
|
archetype,
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Whether any setting is given.
|
/// Whether any setting is given.
|
||||||
fn any(&self) -> bool {
|
fn any(&self) -> bool {
|
||||||
self.limits.is_some() || self.inference.is_some() || self.permissions.is_some()
|
self.limits.is_some() || self.inference.is_some() || self.permissions.is_some()
|
||||||
|
|| self.archetype.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))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -2034,17 +2071,19 @@ struct StoredSettings {
|
||||||
limits: agent_limits::Replaced,
|
limits: agent_limits::Replaced,
|
||||||
inference: inference::Replaced,
|
inference: inference::Replaced,
|
||||||
permissions: agent_permissions::Replaced,
|
permissions: agent_permissions::Replaced,
|
||||||
|
archetype: agent_archetype::Replaced,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl StoredSettings {
|
impl StoredSettings {
|
||||||
/// Store whichever of `limits`, `permissions` and `inference` are given, in
|
/// Store whichever of `limits`, `permissions`, `inference` and `archetype` are
|
||||||
/// that order. A failure on one puts back the ones stored before it.
|
/// given, in that order. A failure on one puts back the ones stored before it.
|
||||||
async fn store(
|
async fn store(
|
||||||
client: &async_nats::Client,
|
client: &async_nats::Client,
|
||||||
agent: &str,
|
agent: &str,
|
||||||
limits: Option<&swarm_queue_client::agent_limits::AgentCaps>,
|
limits: Option<&swarm_queue_client::agent_limits::AgentCaps>,
|
||||||
inference: Option<&swarm_queue_client::agent_inference::AgentInference>,
|
inference: Option<&swarm_queue_client::agent_inference::AgentInference>,
|
||||||
permissions: Option<&swarm_queue_client::agent_permissions::AgentPermissions>,
|
permissions: Option<&swarm_queue_client::agent_permissions::AgentPermissions>,
|
||||||
|
archetype: Option<&swarm_queue_client::agent_archetype::ArchetypeRef>,
|
||||||
) -> Result<Self, problem_details::ProblemDetails> {
|
) -> Result<Self, problem_details::ProblemDetails> {
|
||||||
let mut stored = Self::default();
|
let mut stored = Self::default();
|
||||||
if let Some(limits) = limits {
|
if let Some(limits) = limits {
|
||||||
|
|
@ -2068,6 +2107,15 @@ impl StoredSettings {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
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);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
Ok(stored)
|
Ok(stored)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -2075,6 +2123,7 @@ impl StoredSettings {
|
||||||
agent_limits::restore(client, agent, self.limits).await;
|
agent_limits::restore(client, agent, self.limits).await;
|
||||||
inference::restore(client, agent, self.inference).await;
|
inference::restore(client, agent, self.inference).await;
|
||||||
agent_permissions::restore(client, agent, self.permissions).await;
|
agent_permissions::restore(client, agent, self.permissions).await;
|
||||||
|
agent_archetype::restore(client, agent, self.archetype).await;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -2430,6 +2479,7 @@ fn declare_agent_job(
|
||||||
b: &hive_jobq::builder::JobBuilder<SwarmNodeKind, SwarmResourceKind>,
|
b: &hive_jobq::builder::JobBuilder<SwarmNodeKind, SwarmResourceKind>,
|
||||||
agent: &str,
|
agent: &str,
|
||||||
hive: &str,
|
hive: &str,
|
||||||
|
archetype: bool,
|
||||||
) -> Vec<hive_jobq::builder::NodeGuid> {
|
) -> Vec<hive_jobq::builder::NodeGuid> {
|
||||||
let create_identity = b.node(SwarmNodeKind::CreateIdentity {
|
let create_identity = b.node(SwarmNodeKind::CreateIdentity {
|
||||||
agent: agent.to_owned(),
|
agent: agent.to_owned(),
|
||||||
|
|
@ -2440,22 +2490,31 @@ fn declare_agent_job(
|
||||||
let create_forge_user = b.node(SwarmNodeKind::CreateForgeUser {
|
let create_forge_user = b.node(SwarmNodeKind::CreateForgeUser {
|
||||||
agent: agent.to_owned(),
|
agent: agent.to_owned(),
|
||||||
});
|
});
|
||||||
let create_repo = b
|
// An archetype agent has no config repo: its hive builds it from the
|
||||||
.node(SwarmNodeKind::CreateRepo {
|
// archetype's module, so there is no repo to create, share or seed.
|
||||||
agent: agent.to_owned(),
|
let init_config = (!archetype).then(|| {
|
||||||
})
|
let create_repo = b
|
||||||
.after_ok(create_identity);
|
.node(SwarmNodeKind::CreateRepo {
|
||||||
// `AddRepoMember` needs both parents: the repo to add a collaborator to,
|
agent: agent.to_owned(),
|
||||||
// and the forge user to add as one — adding a nonexistent user is a
|
})
|
||||||
// Forgejo validation error, not an idempotent no-op.
|
.after_ok(create_identity);
|
||||||
// `InitAgentConfigRepo` needs only the repo — see the doc comment above
|
// `AddRepoMember` needs both parents: the repo to add a collaborator
|
||||||
// for why.
|
// to, and the forge user to add as one — adding a nonexistent user is
|
||||||
let _add_repo_member = b
|
// a Forgejo validation error, not an idempotent no-op.
|
||||||
.node(SwarmNodeKind::AddRepoMember {
|
// `InitAgentConfigRepo` needs only the repo — see the doc comment
|
||||||
|
// above for why.
|
||||||
|
let _add_repo_member = b
|
||||||
|
.node(SwarmNodeKind::AddRepoMember {
|
||||||
|
agent: agent.to_owned(),
|
||||||
|
})
|
||||||
|
.after_ok(create_repo)
|
||||||
|
.after_ok(create_forge_user);
|
||||||
|
b.node(SwarmNodeKind::InitAgentConfigRepo {
|
||||||
agent: agent.to_owned(),
|
agent: agent.to_owned(),
|
||||||
})
|
})
|
||||||
.after_ok(create_repo)
|
.after_ok(create_repo)
|
||||||
.after_ok(create_forge_user);
|
.guid()
|
||||||
|
});
|
||||||
// Read on `internal/docs`, a repo that exists whatever this agent is, so
|
// Read on `internal/docs`, a repo that exists whatever this agent is, so
|
||||||
// it needs only the user. Not a parent of the deploy: the agent reads docs
|
// it needs only the user. Not a parent of the deploy: the agent reads docs
|
||||||
// once running, and a failed grant is retried by `forge::docs_grant::spawn`.
|
// once running, and a failed grant is retried by `forge::docs_grant::spawn`.
|
||||||
|
|
@ -2464,11 +2523,6 @@ fn declare_agent_job(
|
||||||
agent: agent.to_owned(),
|
agent: agent.to_owned(),
|
||||||
})
|
})
|
||||||
.after_ok(create_forge_user);
|
.after_ok(create_forge_user);
|
||||||
let init_config = b
|
|
||||||
.node(SwarmNodeKind::InitAgentConfigRepo {
|
|
||||||
agent: agent.to_owned(),
|
|
||||||
})
|
|
||||||
.after_ok(create_repo);
|
|
||||||
// The agent's own identity at the swarm secret store, minted and
|
// The agent's own identity at the swarm secret store, minted and
|
||||||
// published at swarm level so no hive ever needs the capability to mint
|
// published at swarm level so no hive ever needs the capability to mint
|
||||||
// one. After `create_identity` because the certificate names a principal
|
// one. After `create_identity` because the certificate names a principal
|
||||||
|
|
@ -2507,7 +2561,8 @@ fn declare_agent_job(
|
||||||
// deploys by reading that repo, so a deploy asked for any earlier would
|
// deploys by reading that repo, so a deploy asked for any earlier would
|
||||||
// find nothing to build. This is the edge that makes creating an agent at
|
// find nothing to build. This is the edge that makes creating an agent at
|
||||||
// swarm level actually put it on a hive, rather than leaving a
|
// swarm level actually put it on a hive, rather than leaving a
|
||||||
// provisioned name nobody runs.
|
// provisioned name nobody runs. An archetype agent has no repo, so its
|
||||||
|
// deploy waits on the mints and the declaration alone.
|
||||||
//
|
//
|
||||||
// `after_any` on the mint, not `after_ok`: a hive cannot pass down a
|
// `after_any` on the mint, not `after_ok`: a hive cannot pass down a
|
||||||
// certificate the swarm has not published, so the deploy must not
|
// certificate the swarm has not published, so the deploy must not
|
||||||
|
|
@ -2526,13 +2581,12 @@ fn declare_agent_job(
|
||||||
// queue at all fails this node — but `TriggerDeploy` has nothing to
|
// queue at all fails this node — but `TriggerDeploy` has nothing to
|
||||||
// publish to on such a host either, so the strict edge cancels a node
|
// publish to on such a host either, so the strict edge cancels a node
|
||||||
// that could not have succeeded anyway.
|
// that could not have succeeded anyway.
|
||||||
let _trigger_deploy = b
|
let trigger_deploy = b
|
||||||
.node(SwarmNodeKind::TriggerDeploy {
|
.node(SwarmNodeKind::TriggerDeploy {
|
||||||
hive: hive.to_owned(),
|
hive: hive.to_owned(),
|
||||||
agent: agent.to_owned(),
|
agent: agent.to_owned(),
|
||||||
rev: None,
|
rev: None,
|
||||||
})
|
})
|
||||||
.after_ok(init_config)
|
|
||||||
.after_any(mint_identity)
|
.after_any(mint_identity)
|
||||||
.after_any(mint_forge_token)
|
.after_any(mint_forge_token)
|
||||||
// `after_any` for the same reason: the container reads this token
|
// `after_any` for the same reason: the container reads this token
|
||||||
|
|
@ -2541,6 +2595,9 @@ fn declare_agent_job(
|
||||||
// and only this node fails, by name.
|
// and only this node fails, by name.
|
||||||
.after_any(mint_matrix)
|
.after_any(mint_matrix)
|
||||||
.after_ok(set_wanted);
|
.after_ok(set_wanted);
|
||||||
|
if let Some(init_config) = init_config {
|
||||||
|
let _ = trigger_deploy.after_ok(init_config);
|
||||||
|
}
|
||||||
vec![create_identity.guid()]
|
vec![create_identity.guid()]
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -3904,6 +3961,7 @@ mod tests {
|
||||||
limits: None,
|
limits: None,
|
||||||
inference: None,
|
inference: None,
|
||||||
permissions: None,
|
permissions: None,
|
||||||
|
archetype: None,
|
||||||
}),
|
}),
|
||||||
)
|
)
|
||||||
.await
|
.await
|
||||||
|
|
@ -3945,6 +4003,7 @@ mod tests {
|
||||||
limits: None,
|
limits: None,
|
||||||
inference: None,
|
inference: None,
|
||||||
permissions: None,
|
permissions: None,
|
||||||
|
archetype: None,
|
||||||
}),
|
}),
|
||||||
)
|
)
|
||||||
.await
|
.await
|
||||||
|
|
@ -4108,6 +4167,7 @@ mod tests {
|
||||||
Some(&caps(r#"{"cpu_quota":"400%"}"#)),
|
Some(&caps(r#"{"cpu_quota":"400%"}"#)),
|
||||||
Some(&inference),
|
Some(&inference),
|
||||||
None,
|
None,
|
||||||
|
None,
|
||||||
)
|
)
|
||||||
.await;
|
.await;
|
||||||
assert!(stored.is_err(), "{agent}: the inference write must fail");
|
assert!(stored.is_err(), "{agent}: the inference write must fail");
|
||||||
|
|
@ -4246,6 +4306,7 @@ mod tests {
|
||||||
limits: None,
|
limits: None,
|
||||||
inference: None,
|
inference: None,
|
||||||
permissions: None,
|
permissions: None,
|
||||||
|
archetype: None,
|
||||||
}),
|
}),
|
||||||
)
|
)
|
||||||
.await
|
.await
|
||||||
|
|
@ -5374,6 +5435,7 @@ mod tests {
|
||||||
limits: None,
|
limits: None,
|
||||||
inference: None,
|
inference: None,
|
||||||
permissions: None,
|
permissions: None,
|
||||||
|
archetype: None,
|
||||||
}),
|
}),
|
||||||
)
|
)
|
||||||
.await
|
.await
|
||||||
|
|
@ -5408,6 +5470,168 @@ mod tests {
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Node labels of the graph [`super::declare_agent_job`] builds, and the
|
||||||
|
/// labels the deploy node waits on.
|
||||||
|
fn creation_graph(archetype: bool) -> (Vec<String>, Vec<String>) {
|
||||||
|
use hive_jobq_wire::WireNode as _;
|
||||||
|
|
||||||
|
let mut sched = hive_jobq::scheduler::Scheduler::new(
|
||||||
|
hive_jobq::Graph::new(),
|
||||||
|
hive_jobq::resources::ResourceTable::new(),
|
||||||
|
);
|
||||||
|
sched
|
||||||
|
.insert_job(None, |b| {
|
||||||
|
super::declare_agent_job(b, "atlas", "pr1ma", archetype)
|
||||||
|
})
|
||||||
|
.expect("the graph inserts");
|
||||||
|
let graph = sched.graph();
|
||||||
|
let labels = graph.nodes().map(|n| n.payload.label()).collect();
|
||||||
|
let deploy = graph
|
||||||
|
.nodes()
|
||||||
|
.find(|n| n.payload.label() == "trigger_deploy")
|
||||||
|
.expect("every creation deploys");
|
||||||
|
let waits = deploy
|
||||||
|
.deps
|
||||||
|
.iter()
|
||||||
|
.filter_map(|d| match d {
|
||||||
|
hive_jobq::Dep::Node { id, .. } => graph.node(*id).map(|n| n.payload.label()),
|
||||||
|
hive_jobq::Dep::Resource { .. } => None,
|
||||||
|
})
|
||||||
|
.collect();
|
||||||
|
(labels, waits)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// An archetype agent's graph has no repo nodes and its deploy waits on
|
||||||
|
/// no repo; a custom agent's has all three, and its deploy waits on the
|
||||||
|
/// seeded repo.
|
||||||
|
#[test]
|
||||||
|
fn an_archetype_agent_is_created_without_a_config_repo() {
|
||||||
|
const REPO_NODES: [&str; 3] = ["create_repo", "add_repo_member", "init_agent_config_repo"];
|
||||||
|
|
||||||
|
let (custom, custom_waits) = creation_graph(false);
|
||||||
|
for repo in REPO_NODES {
|
||||||
|
assert!(custom.iter().any(|l| l == repo), "custom lacks {repo}");
|
||||||
|
}
|
||||||
|
assert!(custom_waits.iter().any(|l| l == "init_agent_config_repo"));
|
||||||
|
|
||||||
|
let (archetype, archetype_waits) = creation_graph(true);
|
||||||
|
for repo in REPO_NODES {
|
||||||
|
assert!(!archetype.iter().any(|l| l == repo), "archetype has {repo}");
|
||||||
|
}
|
||||||
|
for kept in [
|
||||||
|
"create_identity",
|
||||||
|
"create_forge_user",
|
||||||
|
"grant_docs_read",
|
||||||
|
"mint_agent_identity",
|
||||||
|
"mint_agent_forge_token",
|
||||||
|
"mint_agent_matrix_account",
|
||||||
|
"set_agent_wanted",
|
||||||
|
] {
|
||||||
|
assert!(archetype.iter().any(|l| l == kept), "archetype lacks {kept}");
|
||||||
|
assert!(custom.iter().any(|l| l == kept), "custom lacks {kept}");
|
||||||
|
}
|
||||||
|
assert!(archetype_waits.iter().any(|l| l == "set_agent_wanted"));
|
||||||
|
assert_eq!(archetype.len() + REPO_NODES.len(), custom.len());
|
||||||
|
}
|
||||||
|
|
||||||
|
/// 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: super::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: super::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);
|
||||||
|
}
|
||||||
|
|
||||||
/// The ordering the operator's ruling requires: a hive cannot pass down
|
/// The ordering the operator's ruling requires: a hive cannot pass down
|
||||||
/// a certificate the swarm has not published, so the deploy message must
|
/// a certificate the swarm has not published, so the deploy message must
|
||||||
/// not leave before the mint is terminal.
|
/// not leave before the mint is terminal.
|
||||||
|
|
@ -5428,6 +5652,7 @@ mod tests {
|
||||||
limits: None,
|
limits: None,
|
||||||
inference: None,
|
inference: None,
|
||||||
permissions: None,
|
permissions: None,
|
||||||
|
archetype: None,
|
||||||
}),
|
}),
|
||||||
)
|
)
|
||||||
.await
|
.await
|
||||||
|
|
@ -5478,6 +5703,7 @@ mod tests {
|
||||||
limits: None,
|
limits: None,
|
||||||
inference: None,
|
inference: None,
|
||||||
permissions: None,
|
permissions: None,
|
||||||
|
archetype: None,
|
||||||
}),
|
}),
|
||||||
)
|
)
|
||||||
.await
|
.await
|
||||||
|
|
@ -5536,6 +5762,7 @@ mod tests {
|
||||||
limits: None,
|
limits: None,
|
||||||
inference: None,
|
inference: None,
|
||||||
permissions: None,
|
permissions: None,
|
||||||
|
archetype: None,
|
||||||
}),
|
}),
|
||||||
)
|
)
|
||||||
.await
|
.await
|
||||||
|
|
@ -5741,6 +5968,7 @@ mod tests {
|
||||||
limits: None,
|
limits: None,
|
||||||
inference: None,
|
inference: None,
|
||||||
permissions: None,
|
permissions: None,
|
||||||
|
archetype: None,
|
||||||
}),
|
}),
|
||||||
)
|
)
|
||||||
.await
|
.await
|
||||||
|
|
|
||||||
|
|
@ -583,6 +583,13 @@ impl Policy {
|
||||||
"$JS.API.DIRECT.GET.{}.>",
|
"$JS.API.DIRECT.GET.{}.>",
|
||||||
swarm_queue_client::agent_permissions::STREAM
|
swarm_queue_client::agent_permissions::STREAM
|
||||||
));
|
));
|
||||||
|
// The archetype an agent was created from
|
||||||
|
// (`swarm_queue_client::agent_archetype`), read the same way and every
|
||||||
|
// key for the same reason.
|
||||||
|
subjects.push(format!(
|
||||||
|
"$JS.API.DIRECT.GET.{}.>",
|
||||||
|
swarm_queue_client::agent_archetype::STREAM
|
||||||
|
));
|
||||||
// The one-shot import of the grants this hive's meta repo held, sent
|
// The one-shot import of the grants this hive's meta repo held, sent
|
||||||
// as a request on this hive's own subject: the controller takes the
|
// as a request on this hive's own subject: the controller takes the
|
||||||
// hive from the subject, so one hive cannot import for another.
|
// hive from the subject, so one hive cannot import for another.
|
||||||
|
|
@ -877,6 +884,38 @@ mod tests {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// A hive reads any agent's archetype and can change none of them, and the
|
||||||
|
/// grant reaches no other bucket.
|
||||||
|
#[test]
|
||||||
|
fn a_hive_may_read_every_agents_archetype_and_write_none() {
|
||||||
|
use swarm_queue_client::agent_archetype::{STREAM, direct_get_subject, key_subject};
|
||||||
|
|
||||||
|
let p = policy()
|
||||||
|
.permissions("hive-alpha")
|
||||||
|
.expect("a hive is admitted");
|
||||||
|
let may = |subject: &str| p.publish.iter().any(|g| covers(g, subject));
|
||||||
|
|
||||||
|
for agent in ["atlas", "iris"] {
|
||||||
|
assert!(may(&direct_get_subject(agent)), "cannot read {agent}");
|
||||||
|
}
|
||||||
|
for forbidden in [
|
||||||
|
key_subject("atlas"),
|
||||||
|
format!("$JS.API.STREAM.CREATE.{STREAM}"),
|
||||||
|
format!("$JS.API.STREAM.UPDATE.{STREAM}"),
|
||||||
|
format!("$JS.API.STREAM.DELETE.{STREAM}"),
|
||||||
|
format!("$JS.API.STREAM.PURGE.{STREAM}"),
|
||||||
|
format!("$JS.API.STREAM.MSG.DELETE.{STREAM}"),
|
||||||
|
format!("$JS.API.CONSUMER.CREATE.{STREAM}"),
|
||||||
|
] {
|
||||||
|
assert!(!may(&forbidden), "hive alpha may publish {forbidden}");
|
||||||
|
}
|
||||||
|
// Not the per-source archetype list, which only the controller reads.
|
||||||
|
assert!(!may(&format!(
|
||||||
|
"$JS.API.DIRECT.GET.KV_{0}.$KV.{0}.hyperhive",
|
||||||
|
swarm_queue_client::archetypes::BUCKET
|
||||||
|
)));
|
||||||
|
}
|
||||||
|
|
||||||
/// A hive reads any agent's permissions and can change none of them; an
|
/// A hive reads any agent's permissions and can change none of them; an
|
||||||
/// agent, with the template `swarm-nats.nix` passes, reads its own key and
|
/// agent, with the template `swarm-nats.nix` passes, reads its own key and
|
||||||
/// no other, and writes none.
|
/// no other, and writes none.
|
||||||
|
|
|
||||||
|
|
@ -59,6 +59,9 @@ agent-limits = ["async-nats/jetstream"]
|
||||||
# `Permission` names, so the crates that only need the bucket name do not link
|
# `Permission` names, so the crates that only need the bucket name do not link
|
||||||
# it.
|
# it.
|
||||||
agent-permissions = ["dep:hive-sh4re", "async-nats/jetstream"]
|
agent-permissions = ["dep:hive-sh4re", "async-nats/jetstream"]
|
||||||
|
# `agent_archetype::read`, for hive-c0re, which reads the archetype of each
|
||||||
|
# agent it provisions or rebuilds. Explicit for the same reason as `notices`.
|
||||||
|
agent-archetype = ["async-nats/jetstream"]
|
||||||
|
|
||||||
[dependencies]
|
[dependencies]
|
||||||
# Bare (no `kv`/`jetstream`) unless a consumer opts into the `kv` feature
|
# Bare (no `kv`/`jetstream`) unless a consumer opts into the `kv` feature
|
||||||
|
|
|
||||||
159
swarm-queue-client/src/agent_archetype.rs
Normal file
159
swarm-queue-client/src/agent_archetype.rs
Normal file
|
|
@ -0,0 +1,159 @@
|
||||||
|
//! The per-agent archetype KV bucket: one key per agent created from an
|
||||||
|
//! archetype, holding which archetype that is.
|
||||||
|
//!
|
||||||
|
//! **swarm-controller is the only writer**: `POST /api/agents` stores the key
|
||||||
|
//! before it queues the agent's creation, and a destroy deletes it. An agent
|
||||||
|
//! with no key is a custom agent, built from its own `agent-configs/<name>`
|
||||||
|
//! repo.
|
||||||
|
//!
|
||||||
|
//! **The hive running the agent reads the key** when it provisions or
|
||||||
|
//! rebuilds the agent, and renders the archetype's module into its meta flake
|
||||||
|
//! in place of a config repo. Its grant covers every key of this bucket,
|
||||||
|
//! read-only, for the reason [`crate::agent_limits`] gives.
|
||||||
|
//!
|
||||||
|
//! Keyed by the agent alone, like [`crate::agent_limits`]: an agent can move
|
||||||
|
//! between hives, and its archetype goes with it.
|
||||||
|
|
||||||
|
/// The KV bucket archetype choices are published into, keyed by agent name.
|
||||||
|
pub const BUCKET: &str = "agent-archetype";
|
||||||
|
|
||||||
|
/// The stream backing [`BUCKET`]. `create_key_value` prefixes `KV_`.
|
||||||
|
pub const STREAM: &str = "KV_agent-archetype";
|
||||||
|
|
||||||
|
/// The subject `agent`'s entry is stored under.
|
||||||
|
#[must_use]
|
||||||
|
pub fn key_subject(agent: &str) -> String {
|
||||||
|
format!("$KV.{BUCKET}.{agent}")
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The request subject that reads `agent`'s current entry.
|
||||||
|
#[must_use]
|
||||||
|
pub fn direct_get_subject(agent: &str) -> String {
|
||||||
|
format!("$JS.API.DIRECT.GET.{STREAM}.{}", key_subject(agent))
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Which archetype an agent was created from: the value under
|
||||||
|
/// `BUCKET`/`<agent>`. `source` is a key of [`crate::archetypes::BUCKET`]
|
||||||
|
/// and `name` an archetype in that source's record.
|
||||||
|
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
|
||||||
|
pub struct ArchetypeRef {
|
||||||
|
pub source: String,
|
||||||
|
pub name: String,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Open the archetype-choice bucket for writing, creating it if nothing has
|
||||||
|
/// yet.
|
||||||
|
///
|
||||||
|
/// **Controller-side only.** `history: 1`: a hive wants the current value.
|
||||||
|
#[cfg(feature = "kv")]
|
||||||
|
pub async fn open_or_create(
|
||||||
|
client: &async_nats::Client,
|
||||||
|
) -> Result<async_nats::jetstream::kv::Store, crate::Error> {
|
||||||
|
let js = async_nats::jetstream::new(client.clone());
|
||||||
|
crate::kv::open_or_create(
|
||||||
|
&js,
|
||||||
|
BUCKET,
|
||||||
|
"agent-archetype",
|
||||||
|
"The archetype each agent was created from, set by the swarm controller".to_owned(),
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
}
|
||||||
|
|
||||||
|
/// What reading one agent's key found.
|
||||||
|
#[cfg(feature = "agent-archetype")]
|
||||||
|
#[derive(Debug)]
|
||||||
|
pub enum Read {
|
||||||
|
/// The bucket answered: the agent's archetype, or `None` for an absent or
|
||||||
|
/// deleted key.
|
||||||
|
Current(Option<ArchetypeRef>),
|
||||||
|
/// Nothing answers for the bucket's stream: no archetype agent has been
|
||||||
|
/// created yet, or `JetStream` is not serving it right now.
|
||||||
|
NoBucket,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Read `agent`'s current archetype with one direct get.
|
||||||
|
///
|
||||||
|
/// # Errors
|
||||||
|
///
|
||||||
|
/// A failed request or an undecodable value. A request the queue refused for
|
||||||
|
/// permissions reaches the client as a timeout, not as a refusal.
|
||||||
|
#[cfg(feature = "agent-archetype")]
|
||||||
|
pub async fn read(
|
||||||
|
client: &async_nats::Client,
|
||||||
|
agent: &str,
|
||||||
|
) -> Result<Read, Box<dyn std::error::Error + Send + Sync>> {
|
||||||
|
use async_nats::jetstream::stream::DirectGetErrorKind;
|
||||||
|
|
||||||
|
let js = async_nats::jetstream::new(client.clone());
|
||||||
|
let stream = js.get_stream_no_info(STREAM).await?;
|
||||||
|
let message = match stream.direct_get_last_for_subject(key_subject(agent)).await {
|
||||||
|
Ok(message) => message,
|
||||||
|
Err(e) if matches!(e.kind(), DirectGetErrorKind::NotFound) => {
|
||||||
|
return Ok(Read::Current(None));
|
||||||
|
}
|
||||||
|
Err(e)
|
||||||
|
if matches!(
|
||||||
|
e.kind(),
|
||||||
|
DirectGetErrorKind::ErrorResponse(status, _)
|
||||||
|
if status == async_nats::StatusCode::NO_RESPONDERS
|
||||||
|
) =>
|
||||||
|
{
|
||||||
|
return Ok(Read::NoBucket);
|
||||||
|
}
|
||||||
|
Err(e) => return Err(e.into()),
|
||||||
|
};
|
||||||
|
Ok(Read::Current(decode(
|
||||||
|
message
|
||||||
|
.headers
|
||||||
|
.get("KV-Operation")
|
||||||
|
.map(async_nats::HeaderValue::as_str),
|
||||||
|
&message.payload,
|
||||||
|
)?))
|
||||||
|
}
|
||||||
|
|
||||||
|
/// One stored message as a value: a KV delete or purge is stored as an empty
|
||||||
|
/// marker carrying the `KV-Operation` header, and reads as no archetype.
|
||||||
|
#[cfg(feature = "agent-archetype")]
|
||||||
|
fn decode(operation: Option<&str>, payload: &[u8]) -> serde_json::Result<Option<ArchetypeRef>> {
|
||||||
|
if matches!(operation, Some("DEL" | "PURGE")) {
|
||||||
|
return Ok(None);
|
||||||
|
}
|
||||||
|
serde_json::from_slice(payload).map(Some)
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
|
||||||
|
/// The names the grant in `swarm-nats-auth` is built from.
|
||||||
|
#[test]
|
||||||
|
fn names_match_the_queue_grant() {
|
||||||
|
assert_eq!(STREAM, format!("KV_{BUCKET}"));
|
||||||
|
assert_eq!(key_subject("iris"), "$KV.agent-archetype.iris");
|
||||||
|
assert_eq!(
|
||||||
|
direct_get_subject("iris"),
|
||||||
|
"$JS.API.DIRECT.GET.KV_agent-archetype.$KV.agent-archetype.iris"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// A stored choice reads back, a field a newer controller adds is
|
||||||
|
/// ignored, and a deleted key is no archetype.
|
||||||
|
#[cfg(feature = "agent-archetype")]
|
||||||
|
#[test]
|
||||||
|
fn decoding_reads_the_choice_and_a_delete_as_none() {
|
||||||
|
let android = ArchetypeRef {
|
||||||
|
source: "hyperhive".to_owned(),
|
||||||
|
name: "android-dev".to_owned(),
|
||||||
|
};
|
||||||
|
assert_eq!(
|
||||||
|
decode(
|
||||||
|
None,
|
||||||
|
br#"{"source":"hyperhive","name":"android-dev","later":1}"#
|
||||||
|
)
|
||||||
|
.unwrap(),
|
||||||
|
Some(android)
|
||||||
|
);
|
||||||
|
assert_eq!(decode(Some("DEL"), b"").unwrap(), None);
|
||||||
|
assert_eq!(decode(Some("PURGE"), b"").unwrap(), None);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
@ -232,6 +232,12 @@ pub mod agent_permissions;
|
||||||
/// unconditional; writing needs `kv` and reading needs `operator-profiles`.
|
/// unconditional; writing needs `kv` and reading needs `operator-profiles`.
|
||||||
pub mod operator_profiles;
|
pub mod operator_profiles;
|
||||||
|
|
||||||
|
/// The bucket swarm-controller records each archetype agent's archetype in
|
||||||
|
/// — one key per agent, read by the hive that runs it. The names and the
|
||||||
|
/// stored type are unconditional; writing needs `kv` and reading needs
|
||||||
|
/// `agent-archetype`.
|
||||||
|
pub mod agent_archetype;
|
||||||
|
|
||||||
/// The login flag in an agent's turn-state header. No bucket: it travels
|
/// The login flag in an agent's turn-state header. No bucket: it travels
|
||||||
/// on [`agent_state_subject`] only.
|
/// on [`agent_state_subject`] only.
|
||||||
pub mod agent_login;
|
pub mod agent_login;
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue