hyperhive/swarm-controller/src/archetypes.rs

//! Agent archetypes: the `agent-archetypes` record this daemon keeps from the
//! swarm builder's evaluation, and `GET /api/archetypes`, which the create
//! screen prefills from.
//!
//! Every [`REFRESH_INTERVAL`], starting once the queue is connected, the pass
//! asks the builder for the `meta` of its hyperhive source's archetypes, checks
//! each one the way `POST /api/agents` checks its settings, and writes the
//! result under [`BUILTIN_SOURCE`] when it differs from what is stored. A
//! failed or unanswered evaluation keeps the stored record and is asked again
//! after [`RETRY_FIRST`], doubling up to [`REFRESH_INTERVAL`].

use std::time::Duration;

use axum::Json;
use axum::extract::State;
use hive_types::Permission;
use problem_details::ProblemDetails;
use serde::{Deserialize, Serialize};
use swarm_controller_api::inference::InferenceSettings;
use swarm_controller_api::limits::Caps;
use swarm_queue_client::archetypes::{
    ArchetypeMeta, BUILTIN_SOURCE, MAX_ICON_BYTES, MAX_PROMPT_BYTES, SourceRecord,
};
use swarm_wire::builder::{
    ARCHETYPES_REPLY_DEADLINE, ARCHETYPES_SUBJECT, ArchetypesReply, EvalArchetypes,
};
use swarm_wire::notify_flows::{Flow, MAX_FLOWS};
use utoipa::ToSchema;

use crate::routes::{AppState, error_problem};
use crate::schedules::{queue, read_or_default, refused, write};
use crate::{agent_limits, agent_permissions, inference};

/// How often the builder is asked. Its source only changes when its host is
/// redeployed, so this bounds how long a redeploy takes to reach the create
/// screen.
const REFRESH_INTERVAL: Duration = Duration::from_mins(10);

/// How soon a failed pass is repeated; each further failure doubles the wait,
/// up to [`REFRESH_INTERVAL`]. The builder connects to the queue on its own
/// schedule, so the first pass can find nobody listening.
const RETRY_FIRST: Duration = Duration::from_secs(10);

/// The wait after a failed pass, given the wait after the failure before it.
fn retry_after(previous: Option<Duration>) -> Duration {
    previous.map_or(RETRY_FIRST, |wait| (wait * 2).min(REFRESH_INTERVAL))
}

/// One archetype's `meta` as a source flake declares it. Keys this daemon
/// does not know, `ui` among them, are ignored, so an archetype written for a
/// newer swarm still lists.
#[derive(Deserialize)]
struct DeclaredMeta {
    description: String,
    #[serde(default)]
    permissions: Option<Vec<Permission>>,
    #[serde(default)]
    limits: Option<Caps>,
    #[serde(default)]
    inference: Option<InferenceSettings>,
    #[serde(default)]
    prompt: Option<String>,
    #[serde(default)]
    icon: Option<String>,
    #[serde(default)]
    flows: Vec<Flow>,
}

/// The flows an archetype carries: each passes [`Flow::validate_archetype`],
/// ids are unique and there are no more than the operator's document may hold.
fn check_flows(flows: &[Flow]) -> Result<(), String> {
    if flows.len() > MAX_FLOWS {
        return Err(format!(
            "flows: {} flows, more than {MAX_FLOWS}",
            flows.len()
        ));
    }
    let mut ids = std::collections::BTreeSet::new();
    for flow in flows {
        if !ids.insert(flow.id.as_str()) {
            return Err(format!("flows: id {} is used twice", flow.id));
        }
        flow.validate_archetype()
            .map_err(|why| format!("flow {}: {why}", flow.id))?;
    }
    Ok(())
}

/// `meta` as stored, or why a create would refuse it.
fn check(name: &str, meta: serde_json::Value) -> Result<ArchetypeMeta, String> {
    hive_types::Ident::parse(name).map_err(|reason| format!("name: {reason}"))?;
    let declared: DeclaredMeta = serde_json::from_value(meta).map_err(|e| e.to_string())?;
    if declared.description.trim().is_empty() {
        return Err("description is empty".to_owned());
    }
    if let Some(prompt) = &declared.prompt
        && prompt.len() > MAX_PROMPT_BYTES
    {
        return Err(format!(
            "prompt is {} bytes; the limit is {MAX_PROMPT_BYTES}",
            prompt.len()
        ));
    }
    let icon = declared
        .icon
        .filter(|i| !i.trim().is_empty())
        .map(|i| check_icon(&i).map(|()| i))
        .transpose()
        .map_err(|e| format!("icon: {e}"))?;
    check_flows(&declared.flows)?;
    let permissions = match declared.permissions {
        Some(perms) => {
            agent_permissions::unique(&perms)?;
            Some(
                perms
                    .into_iter()
                    .map(|p| <&str>::from(p).to_owned())
                    .collect(),
            )
        }
        None => None,
    };
    Ok(ArchetypeMeta {
        description: declared.description,
        permissions,
        limits: declared
            .limits
            .as_ref()
            .map(agent_limits::into_stored)
            .transpose()
            .map_err(|e| format!("limits: {e}"))?,
        inference: declared
            .inference
            .map(inference::into_stored)
            .transpose()
            .map_err(|e| format!("inference: {e}"))?,
        prompt: declared.prompt.filter(|p| !p.trim().is_empty()),
        icon,
        flows: declared.flows,
    })
}

/// Whether `icon` can be shown as an image: SVG text within [`MAX_ICON_BYTES`].
/// The create form loads it through an `<img>` as a data URL, so a script in it
/// never runs.
fn check_icon(icon: &str) -> Result<(), String> {
    if icon.len() > MAX_ICON_BYTES {
        return Err(format!(
            "is {} bytes; the limit is {MAX_ICON_BYTES}",
            icon.len()
        ));
    }
    if !icon.trim_start().starts_with("<svg") && !icon.trim_start().starts_with("<?xml") {
        return Err("is not SVG text".to_owned());
    }
    Ok(())
}

/// The record to store for one evaluation: every archetype that passes
/// [`check`], and every one that does not with its reason.
fn record(source: String, archetypes: serde_json::Map<String, serde_json::Value>) -> SourceRecord {
    let mut out = SourceRecord {
        source,
        ..SourceRecord::default()
    };
    for (name, meta) in archetypes {
        match check(&name, meta) {
            Ok(meta) => {
                out.archetypes.insert(name, meta);
            }
            Err(why) => {
                out.refused.insert(name, why);
            }
        }
    }
    out
}

/// Ask the builder once and answer what to store, or why there is nothing to.
async fn evaluate(client: &async_nats::Client) -> Result<SourceRecord, String> {
    use async_nats::RequestErrorKind;

    let body = serde_json::to_vec(&EvalArchetypes {}).map_err(|e| e.to_string())?;
    let sent = async_nats::Request::new()
        .payload(body.into())
        .timeout(Some(ARCHETYPES_REPLY_DEADLINE));
    let msg = client
        .send_request(ARCHETYPES_SUBJECT, sent)
        .await
        .map_err(|e| match e.kind() {
            RequestErrorKind::NoResponders => "no swarm builder is listening: \
                 deploy.builder.enable is set on no host, or its service is not connected to \
                 the swarm queue"
                .to_owned(),
            _ => format!("asking the swarm builder failed: {e}"),
        })?;
    match serde_json::from_slice(&msg.payload) {
        Ok(ArchetypesReply::Evaluated { source, archetypes }) => Ok(record(source, archetypes)),
        Ok(ArchetypesReply::Failed { error_tail }) => {
            Err(format!("the evaluation failed: {error_tail}"))
        }
        Err(e) => Err(format!("the builder's reply did not decode: {e}")),
    }
}

/// Store `value` under [`BUILTIN_SOURCE`] unless it is already there.
/// Answers whether it wrote.
async fn store(client: &async_nats::Client, value: &SourceRecord) -> anyhow::Result<bool> {
    let store = swarm_queue_client::archetypes::open_or_create(client).await?;
    let (current, revision, undecodable) =
        read_or_default::<SourceRecord>(&store, BUILTIN_SOURCE).await?;
    if current == *value && !undecodable {
        return Ok(false);
    }
    // A lost race is another write of the same evaluation; the next pass
    // compares again.
    write(&store, BUILTIN_SOURCE, value, revision).await
}

/// Refresh the record once the queue is connected, then every
/// [`REFRESH_INTERVAL`]; a failed pass is repeated sooner, see [`retry_after`].
pub(crate) fn spawn(client: async_nats::Client) {
    tokio::spawn(async move {
        const CONNECT_POLL: Duration = Duration::from_secs(5);
        let mut retry = None;
        loop {
            while swarm_queue_client::ensure_connected(&client).is_err() {
                tokio::time::sleep(CONNECT_POLL).await;
            }
            tokio::time::sleep(pass(&client, &mut retry).await).await;
        }
    });
}

/// One refresh, and how long to wait before the next. `retry` is the wait
/// after the last failed pass, `None` after a pass that succeeded.
async fn pass(client: &async_nats::Client, retry: &mut Option<Duration>) -> Duration {
    if refresh(client).await {
        *retry = None;
        return REFRESH_INTERVAL;
    }
    let wait = retry_after(*retry);
    *retry = Some(wait);
    wait
}

/// One evaluation, stored when it changed anything. Answers whether the
/// stored record is now what the builder evaluated.
async fn refresh(client: &async_nats::Client) -> bool {
    let value = match evaluate(client).await {
        Ok(value) => value,
        Err(error) => {
            tracing::warn!(%error, "archetypes: kept the stored list");
            return false;
        }
    };
    for (name, why) in &value.refused {
        tracing::warn!(archetype = %name, %why, "archetypes: refused");
    }
    match store(client, &value).await {
        Ok(true) => tracing::info!(
            count = value.archetypes.len(),
            source = %value.source,
            "archetypes: stored"
        ),
        Ok(false) => {}
        Err(e) => {
            tracing::warn!(error = %format!("{e:#}"), "archetypes: storing failed");
            return false;
        }
    }
    true
}

/// One archetype the create screen can offer, with the settings it prefills.
#[derive(Debug, PartialEq, Eq, Serialize, ToSchema)]
pub(crate) struct ArchetypeView {
    /// The source flake's key, `hyperhive` for the built-ins.
    source: String,
    name: String,
    description: String,
    /// Permission names; absent leaves the default set.
    #[serde(skip_serializing_if = "Option::is_none")]
    permissions: Option<Vec<String>>,
    /// As `PUT /api/agents/{name}/limits` takes them.
    #[serde(skip_serializing_if = "Option::is_none")]
    #[schema(value_type = Option<Object>)]
    limits: Option<swarm_wire::agent_limits::AgentCaps>,
    /// As `PUT /api/agents/{name}/inference` takes them.
    #[serde(skip_serializing_if = "Option::is_none")]
    #[schema(value_type = Option<Object>)]
    inference: Option<swarm_wire::agent_inference::AgentInference>,
    /// The text the create form prefills the agent's `AGENTS.md` with.
    #[serde(skip_serializing_if = "Option::is_none")]
    prompt: Option<String>,
    /// The archetype's icon as SVG text, for the create form to show.
    #[serde(skip_serializing_if = "Option::is_none")]
    icon: Option<String>,
}

/// One archetype a source declared and this daemon refused.
#[derive(Debug, PartialEq, Eq, Serialize, ToSchema)]
pub(crate) struct RefusedView {
    source: String,
    name: String,
    reason: String,
}

#[derive(Debug, Default, PartialEq, Eq, Serialize, ToSchema)]
pub(crate) struct Archetypes {
    archetypes: Vec<ArchetypeView>,
    refused: Vec<RefusedView>,
}

fn view(key: &str, record: SourceRecord) -> Archetypes {
    let mut out = Archetypes::default();
    for (name, meta) in record.archetypes {
        out.archetypes.push(ArchetypeView {
            source: key.to_owned(),
            name,
            description: meta.description,
            permissions: meta.permissions,
            limits: meta.limits,
            inference: meta.inference,
            prompt: meta.prompt,
            icon: meta.icon,
        });
    }
    for (name, reason) in record.refused {
        out.refused.push(RefusedView {
            source: key.to_owned(),
            name,
            reason,
        });
    }
    out
}

#[utoipa::path(
    get,
    path = "/api/archetypes",
    responses(
        (status = 200, description = "every archetype the swarm builder's evaluation passed, \
            and every one it refused; both empty before the first evaluation", body = Archetypes),
        (status = 503, description = "no swarm queue is configured on this host, or it is \
            not connected (problem+json)", body = String),
        (status = 500, description = "the archetypes could not be read (problem+json)",
            body = String),
    ),
    tag = "agents"
)]
pub(crate) async fn get_archetypes(
    State(state): State<AppState>,
) -> Result<Json<Archetypes>, ProblemDetails> {
    let client = queue(&state).map_err(refused)?;
    let failed = |e: anyhow::Error| {
        tracing::warn!(error = %format!("{e:#}"), "archetypes: read failed");
        error_problem(
            axum::http::StatusCode::INTERNAL_SERVER_ERROR,
            &format!("{e:#}"),
        )
    };
    let Some(store) = swarm_queue_client::archetypes::open(&client)
        .await
        .map_err(|e| failed(e.into()))?
    else {
        return Ok(Json(Archetypes::default()));
    };
    let (record, _, _) = read_or_default::<SourceRecord>(&store, BUILTIN_SOURCE)
        .await
        .map_err(failed)?;
    Ok(Json(view(BUILTIN_SOURCE, record)))
}

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

    fn evaluated(archetypes: serde_json::Value) -> SourceRecord {
        let serde_json::Value::Object(map) = archetypes else {
            panic!("an object")
        };
        record("/nix/store/h".to_owned(), map)
    }

    #[test]
    fn a_valid_archetype_is_stored_with_its_settings() {
        let r = evaluated(json!({
            "reviewer": {
                "description": "reviews",
                "permissions": ["messaging", "meta", "inbox"],
                "limits": { "memory_max": "4G" },
                "inference": { "effort": "high" },
                "ui": { "anything": true }
            }
        }));
        assert!(r.refused.is_empty(), "{:?}", r.refused);
        let meta = &r.archetypes["reviewer"];
        assert_eq!(meta.description, "reviews");
        assert_eq!(
            meta.permissions.as_deref(),
            Some(
                &[
                    "messaging".to_owned(),
                    "meta".to_owned(),
                    "inbox".to_owned()
                ][..]
            )
        );
        assert_eq!(
            meta.limits.as_ref().and_then(|l| l.memory_max.as_deref()),
            Some("4G")
        );
        assert!(meta.inference.as_ref().is_some_and(|i| i.effort.is_some()));
    }

    /// The prompt is stored as declared; one past the cap refuses the
    /// archetype by name and the cap itself passes; a blank one is no prompt.
    #[test]
    fn the_prompt_is_stored_and_capped() {
        let r = evaluated(json!({
            "plain": { "description": "x" },
            "with": { "description": "x", "prompt": "Review only." },
            "blank": { "description": "x", "prompt": "  \n" },
            "at-cap": { "description": "x", "prompt": "a".repeat(MAX_PROMPT_BYTES) },
            "over-cap": { "description": "x", "prompt": "a".repeat(MAX_PROMPT_BYTES + 1) },
            "not-text": { "description": "x", "prompt": 3 }
        }));
        assert_eq!(r.archetypes["plain"].prompt, None);
        assert_eq!(r.archetypes["with"].prompt.as_deref(), Some("Review only."));
        assert_eq!(r.archetypes["blank"].prompt, None);
        assert_eq!(
            r.archetypes["at-cap"].prompt.as_deref().map(str::len),
            Some(MAX_PROMPT_BYTES)
        );
        assert_eq!(
            r.refused.keys().map(String::as_str).collect::<Vec<_>>(),
            ["not-text", "over-cap"]
        );
        assert!(
            r.refused["over-cap"].contains(&MAX_PROMPT_BYTES.to_string()),
            "{}",
            r.refused["over-cap"]
        );
    }

    /// `GET /api/archetypes` carries the prompt, and leaves it out when
    /// there is none.
    #[test]
    fn the_view_carries_the_prompt() {
        let r = evaluated(json!({
            "with": { "description": "x", "prompt": "Review only." },
            "plain": { "description": "x" }
        }));
        let wire = serde_json::to_value(view(BUILTIN_SOURCE, r)).unwrap();
        let by_name = |name: &str| {
            wire["archetypes"]
                .as_array()
                .unwrap()
                .iter()
                .find(|a| a["name"] == name)
                .unwrap()
                .clone()
        };
        assert_eq!(by_name("with")["prompt"], "Review only.");
        assert!(by_name("plain").get("prompt").is_none());
    }

    /// An icon is stored as declared when it is SVG text within the cap; one
    /// that is not SVG, is over the cap or is not text refuses the archetype
    /// by name, and a blank one is no icon.
    #[test]
    fn the_icon_is_stored_and_checked() {
        let svg = "<svg xmlns=\"http://www.w3.org/2000/svg\"/>";
        let r = evaluated(json!({
            "plain": { "description": "x" },
            "with": { "description": "x", "icon": svg },
            "blank": { "description": "x", "icon": " \n" },
            "not-svg": { "description": "x", "icon": "hello" },
            "over-cap": { "description": "x", "icon": format!("<svg>{}</svg>", "a".repeat(MAX_ICON_BYTES)) },
            "not-text": { "description": "x", "icon": 3 }
        }));
        assert_eq!(r.archetypes["plain"].icon, None);
        assert_eq!(r.archetypes["with"].icon.as_deref(), Some(svg));
        assert_eq!(r.archetypes["blank"].icon, None);
        assert_eq!(
            r.refused.keys().map(String::as_str).collect::<Vec<_>>(),
            ["not-svg", "not-text", "over-cap"]
        );
        assert!(
            r.refused["not-svg"].starts_with("icon:"),
            "{}",
            r.refused["not-svg"]
        );
    }

    /// `GET /api/archetypes` carries the icon, and leaves it out when there is
    /// none.
    #[test]
    fn the_view_carries_the_icon() {
        let r = evaluated(json!({
            "with": { "description": "x", "icon": "<svg/>" },
            "plain": { "description": "x" }
        }));
        let wire = serde_json::to_value(view(BUILTIN_SOURCE, r)).unwrap();
        let by_name = |name: &str| {
            wire["archetypes"]
                .as_array()
                .unwrap()
                .iter()
                .find(|a| a["name"] == name)
                .unwrap()
                .clone()
        };
        assert_eq!(by_name("with")["icon"], "<svg/>");
        assert!(by_name("plain").get("icon").is_none());
    }

    /// `ruth`'s `meta` as `nix/archetypes/default.nix` evaluates it: the check
    /// stores it, and it keeps `manage_root_agent` and the model.
    #[test]
    fn the_ruth_archetype_passes_the_check_with_manage_root_agent_and_sonnet() {
        let r = evaluated(json!({
            "ruth": {
                "description": "manager",
                "inference": { "model": "sonnet" },
                "permissions": [
                    "messaging", "meta", "inbox", "execution",
                    "manage_root_agent", "logs", "metrics"
                ]
            }
        }));
        assert!(r.refused.is_empty(), "{:?}", r.refused);
        let meta = &r.archetypes["ruth"];
        let perms = meta.permissions.as_deref().expect("permissions");
        assert!(perms.contains(&"manage_root_agent".to_owned()));
        assert!(perms.contains(&"logs".to_owned()));
        assert!(perms.contains(&"metrics".to_owned()));
        assert_eq!(
            meta.inference.as_ref().and_then(|i| i.model.as_deref()),
            Some("sonnet")
        );
    }

    #[test]
    fn a_bad_archetype_is_refused_by_name_and_the_rest_stay() {
        let r = evaluated(json!({
            "ok": { "description": "fine" },
            "no-description": { "description": " " },
            "bad-permission": { "description": "x", "permissions": ["root"] },
            "twice": { "description": "x", "permissions": ["meta", "meta"] },
            "bad-limits": { "description": "x", "limits": { "cpu_quota": "lots" } },
            "typo-limits": { "description": "x", "limits": { "memory": "4G" } },
            "bad-inference": { "description": "x", "inference": { "effort": "huge" } },
            "Bad_Name": { "description": "x" }
        }));
        assert_eq!(r.archetypes.keys().collect::<Vec<_>>(), ["ok"]);
        let refused: Vec<&str> = r.refused.keys().map(String::as_str).collect();
        assert_eq!(
            refused,
            [
                "Bad_Name",
                "bad-inference",
                "bad-limits",
                "bad-permission",
                "no-description",
                "twice",
                "typo-limits"
            ]
        );
        assert!(r.refused["bad-limits"].starts_with("limits: "));
        assert!(r.refused["twice"].contains("listed twice"));
    }

    /// A `todo` node to `to`, below a source of `issues`, as an archetype's
    /// `meta.flows` entry in the JSON a source flake declares.
    fn flow_to(id: &str, to: &serde_json::Value) -> serde_json::Value {
        json!({
            "id": id,
            "name": "new issues",
            "nodes": [
                { "id": "src", "kind": "source", "events": ["issues"] },
                { "id": "tell", "kind": "todo", "to": to, "op": "upsert",
                  "key": "forge:{{ repo }}#{{ number }}:new", "text": "new issue {{ title }}" }
            ],
            "edges": [{ "from": "src", "to": "tell" }]
        })
    }

    #[test]
    fn an_archetype_flow_to_members_is_stored_and_any_other_recipient_is_refused() {
        let r = evaluated(json!({
            "ok": { "description": "x", "flows": [flow_to("new", &json!("members"))] },
            "agent": { "description": "x", "flows": [flow_to("new", &json!({ "agent": "triage" }))] },
            "field": { "description": "x", "flows": [flow_to("new", &json!({ "field": "author" }))] },
            "twice": { "description": "x", "flows": [
                flow_to("new", &json!("members")), flow_to("new", &json!("members"))
            ] },
            "review": { "description": "x", "flows": [{
                "id": "rev",
                "name": "review",
                "nodes": [
                    { "id": "src", "kind": "source", "events": ["pull_request"] },
                    { "id": "ask", "kind": "request_reviewer", "reviewer": "argus" }
                ],
                "edges": [{ "from": "src", "to": "ask" }]
            }] },
            "broken": { "description": "x", "flows": [{ "id": "x" }] }
        }));
        assert_eq!(r.archetypes.keys().collect::<Vec<_>>(), ["ok"]);
        assert_eq!(r.archetypes["ok"].flows.len(), 1);
        assert_eq!(r.archetypes["ok"].flows[0].id, "new");
        for refused in ["agent", "field"] {
            assert!(
                r.refused[refused].contains("operator's flows document"),
                "{refused}: {}",
                r.refused[refused]
            );
        }
        assert!(r.refused["twice"].contains("used twice"));
        assert!(r.refused["review"].contains("cannot request a review"));
        assert!(r.refused.contains_key("broken"));
        assert_eq!(r.refused.len(), 5);
    }

    /// Against a real server: no builder listening names the option; a
    /// listener's evaluation is checked and stored, and storing the same
    /// record again writes nothing.
    #[tokio::test]
    async fn an_evaluation_is_stored_and_rewritten_only_when_it_changes() {
        use futures_util::StreamExt as _;

        let nats = crate::test_nats::TestNats::start();
        let client = nats.client().await;
        let unanswered = evaluate(&client).await;
        assert!(
            matches!(&unanswered, Err(e) if e.contains("deploy.builder.enable")),
            "{unanswered:?}"
        );

        let builder = nats.client().await;
        let mut requests = builder
            .queue_subscribe(
                ARCHETYPES_SUBJECT,
                swarm_wire::builder::QUEUE_GROUP.to_owned(),
            )
            .await
            .expect("subscribe");
        // A message the builder sends itself and receives back proves the
        // server processed the subscription before it.
        let mut probe = builder
            .subscribe("archetypes-test.probe")
            .await
            .expect("subscribe");
        builder
            .publish("archetypes-test.probe", "".into())
            .await
            .expect("publish");
        probe.next().await.expect("the probe comes back");
        tokio::spawn(async move {
            let msg = requests.next().await.expect("a request");
            let serde_json::Value::Object(archetypes) = json!({
                "reviewer": { "description": "reviews", "permissions": ["meta"] },
                "broken": { "description": "" }
            }) else {
                unreachable!()
            };
            let reply = serde_json::to_vec(&ArchetypesReply::Evaluated {
                source: "/nix/store/h".to_owned(),
                archetypes,
            })
            .expect("encodes");
            builder
                .publish(msg.reply.expect("a reply subject"), reply.into())
                .await
                .expect("publish");
        });
        let value = evaluate(&client).await.expect("evaluated");
        assert_eq!(value.archetypes.keys().collect::<Vec<_>>(), ["reviewer"]);
        assert_eq!(value.refused.keys().collect::<Vec<_>>(), ["broken"]);

        assert!(store(&client, &value).await.expect("stored"));
        assert!(!store(&client, &value).await.expect("compared"));
        let changed = SourceRecord {
            source: "/nix/store/h2".to_owned(),
            ..value
        };
        assert!(store(&client, &changed).await.expect("stored"));
        let bucket = swarm_queue_client::archetypes::open(&client)
            .await
            .expect("opens")
            .expect("exists");
        let (stored, _, _) = read_or_default::<SourceRecord>(&bucket, BUILTIN_SOURCE)
            .await
            .expect("reads");
        assert_eq!(stored, changed);
    }

    #[test]
    fn a_failed_pass_is_repeated_sooner_each_time_up_to_the_interval() {
        let mut previous = None;
        let mut waits = Vec::new();
        for _ in 0..8 {
            let wait = retry_after(previous);
            waits.push(wait.as_secs());
            previous = Some(wait);
        }
        assert_eq!(waits, [10, 20, 40, 80, 160, 320, 600, 600]);
    }

    /// Against a real server: with no builder listening a pass keeps failing
    /// and asks again after a longer wait each time; the first answer is
    /// stored and puts the wait back to the interval.
    #[tokio::test]
    async fn a_pass_asks_again_until_the_builder_answers() {
        use futures_util::StreamExt as _;

        let nats = crate::test_nats::TestNats::start();
        let client = nats.client().await;
        let mut retry = None;
        assert_eq!(pass(&client, &mut retry).await, Duration::from_secs(10));
        assert_eq!(pass(&client, &mut retry).await, Duration::from_secs(20));
        assert_eq!(retry, Some(Duration::from_secs(20)));

        let builder = nats.client().await;
        let mut requests = builder
            .queue_subscribe(
                ARCHETYPES_SUBJECT,
                swarm_wire::builder::QUEUE_GROUP.to_owned(),
            )
            .await
            .expect("subscribe");
        // A message the builder sends itself and receives back proves the
        // server processed the subscription before it.
        let mut probe = builder
            .subscribe("archetypes-test.probe")
            .await
            .expect("subscribe");
        builder
            .publish("archetypes-test.probe", "".into())
            .await
            .expect("publish");
        probe.next().await.expect("the probe comes back");
        tokio::spawn(async move {
            let msg = requests.next().await.expect("a request");
            let serde_json::Value::Object(archetypes) = json!({
                "reviewer": { "description": "reviews" }
            }) else {
                unreachable!()
            };
            let reply = serde_json::to_vec(&ArchetypesReply::Evaluated {
                source: "/nix/store/h".to_owned(),
                archetypes,
            })
            .expect("encodes");
            builder
                .publish(msg.reply.expect("a reply subject"), reply.into())
                .await
                .expect("publish");
        });
        assert_eq!(pass(&client, &mut retry).await, REFRESH_INTERVAL);
        assert_eq!(retry, None);
        let bucket = swarm_queue_client::archetypes::open(&client)
            .await
            .expect("opens")
            .expect("exists");
        let (stored, _, _) = read_or_default::<SourceRecord>(&bucket, BUILTIN_SOURCE)
            .await
            .expect("reads");
        assert_eq!(stored.archetypes.keys().collect::<Vec<_>>(), ["reviewer"]);
    }

    #[test]
    fn the_view_lists_each_archetype_under_its_source() {
        let r = evaluated(json!({
            "reviewer": { "description": "reviews" },
            "broken": { "description": "" }
        }));
        let v = view(BUILTIN_SOURCE, r);
        assert_eq!(v.archetypes.len(), 1);
        assert_eq!(v.archetypes[0].source, "hyperhive");
        assert_eq!(v.archetypes[0].name, "reviewer");
        assert_eq!(v.archetypes[0].permissions, None);
        assert_eq!(v.refused.len(), 1);
        assert_eq!(v.refused[0].name, "broken");
    }
}