//! `/api/agents/{name}/permissions` — the permission set the swarm grants one
//! agent — and the one-shot import of the grants a hive's meta repo held.
//!
//! Each agent's permissions are one value under its own key of
//! `agent-permissions` (`swarm_queue_client::agent_permissions`), which this
//! daemon alone writes. The agent's harness and subagent daemon apply the tool
//! permissions at their next turn or start; the hive running the agent
//! restarts it when `manage_root_agent` changes. An unset field leaves the
//! agent on the compiled-in default; an empty list grants nothing.

use std::collections::BTreeSet;
use std::sync::Arc;
use std::time::Duration;

use axum::Json;
use axum::extract::{Path, State};
use axum::http::StatusCode;
use backon::{ConstantBuilder, Retryable as _};
use futures_util::StreamExt as _;
use hive_types::Permission;
use problem_details::ProblemDetails;
use serde::Serialize;
use swarm_controller_api::permissions::Permissions;
use swarm_queue_client::agent_permissions;
use swarm_wire::agent_permissions::{
    AgentPermissions, Import, ImportReply, hive_from_import_subject, import_subject_wildcard,
};
use utoipa::ToSchema;

use crate::routes::{AppState, error_problem, wanted_error_status};
use crate::schedules::{live, queue, read_or_default, refused, write, write_revision};
use crate::wanted::WantedWriter;

/// How many times a losing writer re-reads and re-applies, as in
/// `wanted::MAX_ATTEMPTS`.
const MAX_ATTEMPTS: usize = 5;

/// The permissions to store, or why they are refused.
fn into_stored(req: Permissions) -> Result<AgentPermissions, String> {
    if let Some(perms) = &req.permissions {
        unique(perms)?;
    }
    Ok(AgentPermissions {
        permissions: req.permissions,
    })
}

/// Refuse a list that names a permission twice.
pub(crate) fn unique(perms: &[Permission]) -> Result<(), String> {
    let mut seen = std::collections::HashSet::new();
    for perm in perms {
        if !seen.insert(*perm) {
            let name: &str = (*perm).into();
            return Err(format!("permission {name:?} is listed twice"));
        }
    }
    Ok(())
}

/// The stored permissions as the API shows them.
fn view(v: AgentPermissions) -> Permissions {
    Permissions {
        permissions: v.permissions,
    }
}

fn agent_name(name: &str) -> Result<String, (StatusCode, String)> {
    hive_types::Ident::parse(name)
        .map(hive_types::Ident::into_string)
        .map_err(|reason| (StatusCode::BAD_REQUEST, reason.to_owned()))
}

fn queue_problem(e: &anyhow::Error) -> ProblemDetails {
    tracing::warn!(error = %format!("{e:#}"), "agent permissions: queue failure");
    error_problem(wanted_error_status(e), &format!("{e:#}"))
}

async fn open_or_create(
    client: &async_nats::Client,
) -> Result<async_nats::jetstream::kv::Store, ProblemDetails> {
    agent_permissions::open_or_create(client)
        .await
        .map_err(|e| queue_problem(&anyhow::Error::new(e)))
}

#[utoipa::path(
    get,
    path = "/api/agents/{name}/permissions",
    params(("name" = String, Path, description = "agent name")),
    responses(
        (status = 200, description = "the permissions the swarm sets for the agent; `{}` \
            when it sets none", body = Permissions),
        (status = 400, description = "not an identifier (problem+json)", body = String),
        (status = 503, description = "no swarm queue is configured on this host, or it is \
            not connected (problem+json)", body = String),
        (status = 500, description = "the permissions could not be read (problem+json)",
            body = String),
    ),
    tag = "agents"
)]
pub(crate) async fn get_agent_permissions(
    State(state): State<AppState>,
    Path(name): Path<String>,
) -> Result<Json<Permissions>, ProblemDetails> {
    let agent = agent_name(&name).map_err(refused)?;
    let client = queue(&state).map_err(refused)?;
    let store = open_or_create(&client).await?;
    let (current, _, _) = read_or_default::<AgentPermissions>(&store, &agent)
        .await
        .map_err(|e| queue_problem(&e))?;
    Ok(Json(view(current)))
}

/// Set the permissions the swarm grants one agent.
///
/// The value replaces the stored one whole. The hive running the agent reads
/// it within 30 seconds and restarts the agent, after its current turn, when
/// `manage_root_agent` changed; the agent applies the rest at its next turn.
#[utoipa::path(
    put,
    path = "/api/agents/{name}/permissions",
    params(("name" = String, Path, description = "agent name")),
    request_body = Permissions,
    responses(
        (status = 200, description = "the permissions as stored", body = Permissions),
        (status = 400, description = "not an identifier, not declared on any hive, or a \
            permission listed twice (problem+json)", body = String),
        (status = 422, description = "an unknown field or permission name"),
        (status = 503, description = "no swarm queue is configured on this host, or it is \
            not connected (problem+json)", body = String),
        (status = 500, description = "the permissions could not be written (problem+json)",
            body = String),
    ),
    tag = "agents"
)]
pub(crate) async fn put_agent_permissions(
    State(state): State<AppState>,
    Path(name): Path<String>,
    Json(req): Json<Permissions>,
) -> Result<Json<Permissions>, ProblemDetails> {
    let agent = agent_name(&name).map_err(refused)?;
    let value = into_stored(req).map_err(|why| error_problem(StatusCode::BAD_REQUEST, &why))?;
    let client = queue(&state).map_err(refused)?;
    if !live(&state).await?.contains(&agent) {
        return Err(error_problem(
            StatusCode::BAD_REQUEST,
            &format!("no hive's wanted state declares an agent {agent:?} that is not destroyed"),
        ));
    }
    let store = open_or_create(&client).await?;
    for _ in 0..MAX_ATTEMPTS {
        let (current, revision, undecodable) = read_or_default::<AgentPermissions>(&store, &agent)
            .await
            .map_err(|e| queue_problem(&e))?;
        if (current == value && !undecodable)
            || write(&store, &agent, &value, revision)
                .await
                .map_err(|e| queue_problem(&e))?
        {
            tracing::info!(%agent, ?value, "set agent permissions");
            return Ok(Json(view(value)));
        }
    }
    Err(queue_problem(&anyhow::anyhow!(
        "gave up writing the permissions of {agent} after {MAX_ATTEMPTS} conflicting writes"
    )))
}

#[utoipa::path(
    delete,
    path = "/api/agents/{name}/permissions",
    params(("name" = String, Path, description = "agent name")),
    responses(
        (status = 204, description = "cleared; the hive's default applies again"),
        (status = 400, description = "not an identifier (problem+json)", body = String),
        (status = 503, description = "no swarm queue is configured on this host, or it is \
            not connected (problem+json)", body = String),
        (status = 500, description = "the permissions could not be cleared (problem+json)",
            body = String),
    ),
    tag = "agents"
)]
pub(crate) async fn delete_agent_permissions(
    State(state): State<AppState>,
    Path(name): Path<String>,
) -> Result<StatusCode, ProblemDetails> {
    let agent = agent_name(&name).map_err(refused)?;
    let client = queue(&state).map_err(refused)?;
    let store = open_or_create(&client).await?;
    store
        .delete(&agent)
        .await
        .map_err(|e| queue_problem(&anyhow::Error::new(e)))?;
    tracing::info!(%agent, "cleared agent permissions");
    Ok(StatusCode::NO_CONTENT)
}

/// Delete a destroyed agent's key. A name is free again once its agent is
/// destroyed, and a create left on the default stores nothing, so a key left
/// here would hand the next agent of that name this one's grant.
///
/// Like the credential revoke it runs beside, a failure is logged and does
/// not stop the destroy; declaring the agent destroyed again re-runs it.
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_permissions::open_or_create(&client)
            .await?
            .delete(agent)
            .await?;
        anyhow::Ok(())
    }
    .await;
    match cleared {
        Ok(()) => tracing::info!(hive, agent, "destroyed agent's permissions cleared"),
        Err(e) => tracing::error!(
            hive,
            agent,
            error = %format!("{e:#}"),
            "clearing this agent's permissions FAILED; it was destroyed anyway, so a new \
             agent of the same name created on the default gets its grant. Declaring the \
             agent destroyed again re-runs this."
        ),
    }
}

/// Every permission name, and the set an agent the swarm sets nothing for
/// holds — so a client lists them without a copy of its own.
#[derive(Debug, Serialize, ToSchema)]
pub(crate) struct Catalogue {
    #[schema(value_type = Vec<String>)]
    all: Vec<Permission>,
    #[schema(value_type = Vec<String>)]
    default: Vec<Permission>,
}

#[utoipa::path(
    get,
    path = "/api/permissions",
    responses((status = 200, description = "every permission name, and the default set",
        body = Catalogue)),
    tag = "agents"
)]
pub(crate) async fn get_permission_catalogue() -> Json<Catalogue> {
    Json(Catalogue {
        all: Permission::ALL.to_vec(),
        default: Permission::DEFAULT.to_vec(),
    })
}

/// The value `POST /api/agents` stores for `perms`, or why it is refused.
/// The agent is not declared on its hive yet, so the "declared by a hive"
/// check of [`put_agent_permissions`] does not apply.
pub(crate) fn at_create(perms: Vec<Permission>) -> Result<AgentPermissions, String> {
    unique(&perms)?;
    Ok(AgentPermissions {
        permissions: Some(perms),
    })
}

/// What [`store`] replaced, for [`restore`]; `None` when it wrote nothing.
pub(crate) type Replaced = Option<crate::schedules::Replaced<AgentPermissions>>;

/// Replace `agent`'s stored permissions with `value`.
pub(crate) async fn store(
    client: &async_nats::Client,
    agent: &str,
    value: &AgentPermissions,
) -> Result<Replaced, ProblemDetails> {
    let store = open_or_create(client).await?;
    for _ in 0..MAX_ATTEMPTS {
        let (current, revision, undecodable) = read_or_default::<AgentPermissions>(&store, agent)
            .await
            .map_err(|e| queue_problem(&e))?;
        if current == *value && !undecodable {
            tracing::info!(%agent, ?value, "set agent permissions at create");
            return Ok(None);
        }
        // A deleted key reads as present at its marker's revision, so ask
        // whether it holds a value before keeping one to put back.
        let held = store
            .get(agent)
            .await
            .map_err(|e| queue_problem(&anyhow::Error::new(e)))?
            .is_some();
        if let Some(written) = write_revision(&store, agent, value, revision)
            .await
            .map_err(|e| queue_problem(&e))?
        {
            tracing::info!(%agent, ?value, "set agent permissions at create");
            return Ok(Some(crate::schedules::Replaced::new(
                held.then_some(current),
                written,
            )));
        }
    }
    Err(queue_problem(&anyhow::anyhow!(
        "gave up writing the permissions 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 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 permissions: another write landed since");
        }
        Err(e) => {
            tracing::warn!(%agent, error = %format!("{e:#}"), "could not put back the agent's permissions");
        }
    }
}

/// Create the bucket as soon as the queue is connected, then answer every
/// hive's import. Creating it first means a hive that finds no bucket knows
/// the queue is not serving it, rather than that nothing is set.
pub(crate) fn spawn(client: async_nats::Client, wanted: Arc<WantedWriter>, hives: Vec<String>) {
    tokio::spawn(async move {
        const RETRY: Duration = Duration::from_secs(5);
        // The first two retry until they work: the queue is the only thing either waits on.
        let every_five_seconds = || {
            ConstantBuilder::default()
                .with_delay(RETRY)
                .without_max_times()
        };
        let Ok(store) = (|| async {
            swarm_queue_client::ensure_connected(&client)?;
            agent_permissions::open_or_create(&client).await
        })
        .retry(every_five_seconds())
        .notify(|e: &swarm_queue_client::Error, _| {
            // Not connected is the queue being late, and it says so itself.
            if !matches!(e, swarm_queue_client::Error::NotConnected(_)) {
                tracing::warn!(
                    error = swarm_queue_client::chain(e),
                    "creating the agent-permissions bucket failed; retrying"
                );
            }
        })
        .await
        else {
            return;
        };
        tokio::spawn({
            let (store, wanted, hives) = (store.clone(), Arc::clone(&wanted), hives.clone());
            async move {
                // Retries without end, so it answers only once it has worked.
                let _ = (|| seed_ruth_grant(&store, &wanted, &hives))
                    .retry(every_five_seconds())
                    .notify(|e: &anyhow::Error, _| {
                        tracing::warn!(
                            error = %format!("{e:#}"),
                            "agent permissions: seeding ruth's grant failed; retrying"
                        );
                    })
                    .await;
            }
        });
        let subject = import_subject_wildcard();
        loop {
            match client.subscribe(subject.clone()).await {
                Ok(mut requests) => {
                    tracing::info!(%subject, "agent permissions: answering imports");
                    while let Some(msg) = requests.next().await {
                        let Some(reply) = msg.reply.clone() else {
                            continue;
                        };
                        let answer = answer_import(&store, &wanted, &hives, &msg).await;
                        let encoded = serde_json::to_vec(&answer).unwrap_or_default();
                        if let Err(e) = client.publish(reply, encoded.into()).await {
                            tracing::warn!(error = %e, "agent permissions: answering an import failed");
                        }
                    }
                    tracing::warn!(%subject, "agent permissions: import subscription ended");
                }
                Err(e) => {
                    tracing::warn!(%subject, error = %e, "agent permissions: subscribe failed");
                }
            }
            tokio::time::sleep(RETRY).await;
        }
    });
}

async fn answer_import(
    store: &async_nats::jetstream::kv::Store,
    wanted: &WantedWriter,
    hives: &[String],
    msg: &async_nats::Message,
) -> ImportReply {
    let failed = |error: String| {
        tracing::warn!(subject = %msg.subject, %error, "agent permissions: import refused");
        ImportReply::Failed { error }
    };
    let Some(hive) = hive_from_import_subject(&msg.subject) else {
        return failed(format!("{} is not an import subject", msg.subject));
    };
    if !hives.iter().any(|h| h == hive) {
        return failed(format!("{hive} is not a hive this controller serves"));
    }
    let import: Import = match serde_json::from_slice(&msg.payload) {
        Ok(import) => import,
        Err(e) => return failed(format!("the import does not decode: {e}")),
    };
    let declared = match wanted.view(hive).await {
        Ok(d) => crate::agent_renewal::live_agents(d.as_slice()),
        Err(e) => return failed(format!("reading {hive}'s wanted state: {e:#}")),
    };
    match import_into(store, &declared, import).await {
        Ok(reply) => {
            tracing::info!(%hive, ?reply, "agent permissions: imported a hive's grants");
            reply
        }
        Err(e) => failed(format!("{e:#}")),
    }
}

/// The agent [`seed_into`] seeds.
const SEEDED_AGENT: &str = "ruth";

/// One-shot migration, run at every start; expect to delete it (with
/// [`seed_into`] and its call in [`spawn`]) once every swarm has passed it.
/// Builds before it granted `manage_root_agent` by name to an agent called
/// `ruth` whose permissions were unset; this writes that grant into her key
/// so she keeps her recovery mounts. Idempotent: a set key is never touched.
async fn seed_ruth_grant(
    store: &async_nats::jetstream::kv::Store,
    wanted: &WantedWriter,
    hives: &[String],
) -> anyhow::Result<()> {
    let mut declarations = Vec::with_capacity(hives.len());
    for hive in hives {
        if let Some(d) = wanted.view(hive).await? {
            declarations.push(d);
        }
    }
    seed_into(store, &crate::agent_renewal::live_agents(&declarations)).await?;
    Ok(())
}

/// Write [`Permission::DEFAULT`] plus `manage_root_agent` under
/// [`SEEDED_AGENT`]'s key when `declared` holds her and her `permissions`
/// are unset (no key, a deleted key, or `{}`). `true` when it wrote.
async fn seed_into(
    store: &async_nats::jetstream::kv::Store,
    declared: &BTreeSet<String>,
) -> anyhow::Result<bool> {
    if !declared.contains(SEEDED_AGENT) {
        return Ok(false);
    }
    let (current, revision, undecodable) =
        read_or_default::<AgentPermissions>(store, SEEDED_AGENT).await?;
    if current.permissions.is_some() || undecodable {
        return Ok(false);
    }
    let mut perms = Permission::DEFAULT.to_vec();
    perms.push(Permission::ManageRootAgent);
    let value = AgentPermissions {
        permissions: Some(perms),
    };
    let written = write(store, SEEDED_AGENT, &value, revision).await?;
    if written {
        tracing::info!(
            agent = SEEDED_AGENT,
            ?value,
            "agent permissions: one-shot seed of the grant ruth held by name"
        );
    }
    Ok(written)
}

/// Write each agent's imported permissions under its key when it has none,
/// for the agents in `declared` only.
async fn import_into(
    store: &async_nats::jetstream::kv::Store,
    declared: &BTreeSet<String>,
    import: Import,
) -> anyhow::Result<ImportReply> {
    let (mut written, mut kept, mut skipped) = (Vec::new(), Vec::new(), Vec::new());
    for (agent, caps) in import.permissions {
        if !declared.contains(&agent) {
            skipped.push(agent);
            continue;
        }
        let value = AgentPermissions {
            permissions: Some(caps),
        };
        if write(store, &agent, &value, None).await? {
            written.push(agent);
        } else {
            kept.push(agent);
        }
    }
    Ok(ImportReply::Imported {
        written,
        kept,
        skipped,
    })
}

#[cfg(test)]
mod tests {
    use super::{
        Permissions, delete_agent_permissions, get_agent_permissions, import_into,
        put_agent_permissions, seed_into,
    };
    use crate::test_nats::TestNats;
    use axum::extract::{Path, State};
    use axum::http::StatusCode;
    use hive_types::Permission;
    use swarm_wire::agent_permissions::{Import, ImportReply};

    async fn published(client: &async_nats::Client, agent: &str) -> Option<String> {
        let store = swarm_queue_client::agent_permissions::open_or_create(client)
            .await
            .expect("bucket");
        store
            .get(agent)
            .await
            .expect("read")
            .map(|v| String::from_utf8(v.to_vec()).expect("utf-8"))
    }

    fn granted() -> Permissions {
        Permissions {
            permissions: Some(vec![Permission::ManageRootAgent]),
        }
    }

    async fn put(
        state: &crate::routes::AppState,
        agent: &str,
        value: Permissions,
    ) -> Result<axum::Json<Permissions>, problem_details::ProblemDetails> {
        put_agent_permissions(
            State(state.clone()),
            Path(agent.to_owned()),
            axum::Json(value),
        )
        .await
    }

    async fn get(state: &crate::routes::AppState, agent: &str) -> Permissions {
        get_agent_permissions(State(state.clone()), Path(agent.to_owned()))
            .await
            .expect("read")
            .0
    }

    /// Through the real handlers: a grant lands as the hive reads it, an empty
    /// list is stored as an empty list, and `DELETE` takes the key out.
    #[tokio::test]
    async fn the_route_writes_what_a_hive_reads_and_delete_clears_it() {
        let nats = TestNats::start();
        let client = nats.client().await;
        let state = crate::agent_limits::tests::declared(&client).await;

        assert_eq!(
            put(&state, "atlas", granted()).await.expect("set").0,
            granted()
        );
        assert_eq!(
            published(&client, "atlas").await.as_deref(),
            Some(r#"{"permissions":["manage_root_agent"]}"#)
        );
        assert_eq!(get(&state, "atlas").await, granted());

        let none = Permissions {
            permissions: Some(Vec::new()),
        };
        let _ = put(&state, "atlas", none.clone()).await.expect("set");
        assert_eq!(
            published(&client, "atlas").await.as_deref(),
            Some(r#"{"permissions":[]}"#)
        );

        assert_eq!(
            delete_agent_permissions(State(state.clone()), Path("atlas".to_owned())).await,
            Ok(StatusCode::NO_CONTENT)
        );
        assert_eq!(published(&client, "atlas").await, None);
        assert_eq!(get(&state, "atlas").await, Permissions::default());
    }

    /// A name no hive declares, or a name that is not an identifier, is a 400
    /// and writes nothing.
    #[tokio::test]
    async fn an_undeclared_or_malformed_agent_is_a_400() {
        let nats = TestNats::start();
        let client = nats.client().await;
        let state = crate::agent_limits::tests::declared(&client).await;

        for agent in ["iris", "Not_An_Ident"] {
            let problem = put(&state, agent, granted()).await.expect_err("refused");
            assert_eq!(problem.status, Some(StatusCode::BAD_REQUEST), "{agent}");
        }
        assert_eq!(published(&client, "iris").await, None);
    }

    async fn put_raw(app: &axum::Router, body: &'static str) -> StatusCode {
        use tower::ServiceExt as _;
        app.clone()
            .oneshot(
                axum::http::Request::put("/api/agents/atlas/permissions")
                    .header(axum::http::header::CONTENT_TYPE, "application/json")
                    .body(axum::body::Body::from(body))
                    .expect("request"),
            )
            .await
            .expect("router answers")
            .status()
    }

    /// Behind the real router: an unknown field or permission name is a 422,
    /// a duplicate is a 400, and none of them changes what is stored.
    #[tokio::test]
    async fn a_bad_body_is_refused_and_the_grant_stays() {
        const GRANTED: &str = r#"{"permissions":["manage_root_agent"]}"#;
        let nats = TestNats::start();
        let client = nats.client().await;
        let state = crate::agent_limits::tests::declared(&client).await;
        let _ = put(&state, "atlas", granted()).await.expect("set");
        let app = crate::startup::build_app(state);

        for (body, status) in [
            (
                r#"{"permissions":["manage_root_agent"],"tool_groups":["meta"]}"#,
                StatusCode::UNPROCESSABLE_ENTITY,
            ),
            (r#"{"permisions":[]}"#, StatusCode::UNPROCESSABLE_ENTITY),
            (
                r#"{"permissions":["lifecycle"]}"#,
                StatusCode::UNPROCESSABLE_ENTITY,
            ),
            (
                r#"{"permissions":["fly_to_the_moon"]}"#,
                StatusCode::UNPROCESSABLE_ENTITY,
            ),
            (
                r#"{"permissions":["manage_root_agent","manage_root_agent"]}"#,
                StatusCode::BAD_REQUEST,
            ),
        ] {
            assert_eq!(put_raw(&app, body).await, status, "{body}");
            assert_eq!(
                published(&client, "atlas").await.as_deref(),
                Some(GRANTED),
                "{body}"
            );
        }
        // Control: the same router and request do reach the bucket.
        assert_eq!(put_raw(&app, "{}").await, StatusCode::OK);
        assert_eq!(published(&client, "atlas").await.as_deref(), Some("{}"));
    }

    /// The import writes a key only where there is none, and only for the
    /// agents the sending hive declares.
    #[tokio::test]
    async fn an_import_writes_only_missing_keys_of_the_hives_own_agents() {
        let nats = TestNats::start();
        let client = nats.client().await;
        let store = swarm_queue_client::agent_permissions::open_or_create(&client)
            .await
            .expect("bucket");
        store
            .put("atlas", br#"{"permissions":[]}"#.to_vec().into())
            .await
            .expect("seed");
        let declared = ["atlas".to_owned(), "ruth".to_owned()].into();
        let import = Import {
            permissions: [
                ("atlas".to_owned(), vec![Permission::ManageRootAgent]),
                ("ruth".to_owned(), vec![Permission::ManageRootAgent]),
                ("iris".to_owned(), vec![Permission::ManageRootAgent]),
            ]
            .into(),
        };

        let reply = import_into(&store, &declared, import)
            .await
            .expect("imported");
        assert_eq!(
            reply,
            ImportReply::Imported {
                written: vec!["ruth".to_owned()],
                kept: vec!["atlas".to_owned()],
                skipped: vec!["iris".to_owned()],
            }
        );
        assert_eq!(
            published(&client, "atlas").await.as_deref(),
            Some(r#"{"permissions":[]}"#)
        );
        assert_eq!(
            published(&client, "ruth").await.as_deref(),
            Some(r#"{"permissions":["manage_root_agent"]}"#)
        );
        assert_eq!(published(&client, "iris").await, None);
    }

    const SEEDED: &str =
        r#"{"permissions":["messaging","meta","inbox","execution","manage_root_agent"]}"#;

    /// A declared `ruth` whose permissions are unset (no key, or a deleted
    /// one, or `{}`) gets the default plus `manage_root_agent`, once.
    #[tokio::test]
    async fn the_seed_writes_an_unset_ruth_grant() {
        let nats = TestNats::start();
        let client = nats.client().await;
        let store = swarm_queue_client::agent_permissions::open_or_create(&client)
            .await
            .expect("bucket");
        let declared = ["atlas".to_owned(), "ruth".to_owned()].into();

        assert!(seed_into(&store, &declared).await.expect("seeded"));
        assert_eq!(published(&client, "ruth").await.as_deref(), Some(SEEDED));
        assert_eq!(published(&client, "atlas").await, None, "only ruth");
        assert!(!seed_into(&store, &declared).await.expect("idempotent"));

        store.delete("ruth").await.expect("delete");
        assert!(seed_into(&store, &declared).await.expect("seeded"));
        assert_eq!(published(&client, "ruth").await.as_deref(), Some(SEEDED));

        store
            .put("ruth", b"{}".to_vec().into())
            .await
            .expect("unset");
        assert!(seed_into(&store, &declared).await.expect("seeded"));
        assert_eq!(published(&client, "ruth").await.as_deref(), Some(SEEDED));
    }

    /// A ruth key the swarm already set, even to nothing, is left as it is;
    /// so is one this build cannot decode.
    #[tokio::test]
    async fn the_seed_leaves_a_set_ruth_key_alone() {
        let nats = TestNats::start();
        let client = nats.client().await;
        let store = swarm_queue_client::agent_permissions::open_or_create(&client)
            .await
            .expect("bucket");
        let declared = ["ruth".to_owned()].into();
        for set in [
            r#"{"permissions":[]}"#,
            r#"{"permissions":["web_tools"]}"#,
            r#"{"permissions":["fly_to_the_moon"]}"#,
        ] {
            store
                .put("ruth", set.as_bytes().to_vec().into())
                .await
                .expect("set");
            assert!(!seed_into(&store, &declared).await.expect("read"), "{set}");
            assert_eq!(published(&client, "ruth").await.as_deref(), Some(set));
        }
    }

    /// With no `ruth` declared nothing is written; declaring her is what
    /// makes the same call write.
    #[tokio::test]
    async fn the_seed_writes_nothing_without_a_declared_ruth() {
        let nats = TestNats::start();
        let client = nats.client().await;
        let store = swarm_queue_client::agent_permissions::open_or_create(&client)
            .await
            .expect("bucket");

        let declared = ["atlas".to_owned()].into();
        assert!(!seed_into(&store, &declared).await.expect("read"));
        assert_eq!(published(&client, "ruth").await, None);
        assert_eq!(published(&client, "atlas").await, None);

        let declared = ["atlas".to_owned(), "ruth".to_owned()].into();
        assert!(seed_into(&store, &declared).await.expect("seeded"));
        assert_eq!(published(&client, "ruth").await.as_deref(), Some(SEEDED));
    }

    /// `POST /api/agents` with `permissions` stores them under the new agent's
    /// key in the same call; a list naming one twice is refused before
    /// anything is stored or queued.
    #[tokio::test]
    async fn create_writes_the_permissions_it_is_given() {
        let nats = TestNats::start();
        let client = nats.client().await;
        let state = crate::agent_limits::tests::declared(&client).await;
        let create = |name: &str, permissions| {
            crate::routes::agents::create_agent(
                State(state.clone()),
                axum::Json(swarm_controller_api::agents::CreateAgentRequest {
                    name: name.to_owned(),
                    hive: "pr1ma".to_owned(),
                    limits: None,
                    inference: None,
                    permissions,
                    archetype: None,
                    agents_md: None,
                }),
            )
        };

        let problem = create(
            "iris",
            Some(vec![Permission::Messaging, Permission::Messaging]),
        )
        .await
        .expect_err("a duplicate is refused");
        assert_eq!(problem.status, Some(StatusCode::BAD_REQUEST), "{problem:?}");
        assert_eq!(published(&client, "iris").await, None);
        let queued = state
            .jobq
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .graph()
            .nodes()
            .count();
        assert_eq!(queued, 0, "nothing is queued for a refused create");

        let _ = create("iris", Some(vec![Permission::Messaging]))
            .await
            .expect("created");
        assert_eq!(
            published(&client, "iris").await.as_deref(),
            Some(r#"{"permissions":["messaging"]}"#)
        );

        // Control: no `permissions` writes no key, so the default applies.
        let _ = create("damocles", None).await.expect("created");
        assert_eq!(published(&client, "damocles").await, None);
    }

    /// Destroying an agent deletes its key, so a new agent of the same name
    /// created on the default gets the default, not the old grant.
    #[tokio::test]
    async fn a_recreated_agent_does_not_inherit_the_destroyed_ones_grant() {
        let nats = TestNats::start();
        let client = nats.client().await;
        let state = crate::agent_limits::tests::declared(&client).await;
        let _ = put(&state, "atlas", granted()).await.expect("set");

        let _ = crate::routes::agents::set_agent_state(
            State(state.clone()),
            Path(("pr1ma".to_owned(), "atlas".to_owned())),
            axum::Json(crate::routes::agents::SetAgentStateRequest {
                state: swarm_wire::wanted::AgentState::Destroyed,
            }),
        )
        .await
        .expect("destroyed");
        let _ = crate::routes::agents::create_agent(
            State(state.clone()),
            axum::Json(swarm_controller_api::agents::CreateAgentRequest {
                name: "atlas".to_owned(),
                hive: "pr1ma".to_owned(),
                limits: None,
                inference: None,
                permissions: None,
                archetype: None,
                agents_md: None,
            }),
        )
        .await
        .expect("recreated");

        assert_eq!(published(&client, "atlas").await, None);
        assert_eq!(get(&state, "atlas").await, Permissions::default());
    }

    #[tokio::test]
    async fn the_catalogue_lists_every_permission_and_the_default() {
        let json = serde_json::to_value(super::get_permission_catalogue().await.0).unwrap();
        assert_eq!(
            json["all"].as_array().map(Vec::len),
            Some(Permission::ALL.len())
        );
        assert_eq!(
            json["default"],
            serde_json::json!(["messaging", "meta", "inbox", "execution"])
        );
    }
}