hyperhive/hive-agent/src/swarm_queue.rs

//! This agent's swarm-queue credentials, resolved once at boot.
//!
//! Three of the four coordinates arrive as environment variables the meta
//! flake renders into the harness unit. The fourth — the OIDC client id —
//! arrives as a *file*, delivered beside the secret as one systemd
//! credential pair (`nix/agent-modules/queue.nix`). That is the whole reason
//! this module exists rather than a bare [`QueueConfig::from_env`] call: the
//! id rides with the secret so a reader never has to spell `hive-<name>-agent`
//! a second time, and `from_env` wants it as a value.
//!
//! Reading the file and assigning the variable would put the same rule back
//! in `from_env`'s hands, but `std::env::set_var` is unsound in a process that
//! has already spawned threads, and this one has. So the rule is restated
//! here over the inputs this consumer actually has, and [`decide`] is the
//! single place it lives.
//!
//! Beside all four sits a fifth coordinate, resolved by [`store_source`] and
//! not part of their group. The four are the *hive's* — one OIDC client
//! shared by every container on it — so at the queue's auth callout they say
//! which hive is connecting and never which agent. The fifth is this agent's
//! own, minted per agent at swarm level. This process reads it from the swarm
//! secret store under the agent's own certificate and holds it in memory only
//! ([`AgentSecret`]).
//!
//! When the store holds one, [`client`] connects with it first, and the agent
//! publishes on its own hive-free subjects. When it holds none, or the queue
//! refuses it, the first connect falls back to the hive's shared client and
//! the hive-scoped subjects. [`Connection::presented`] says which, so a
//! publisher builds the subject that credential is granted.

use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex, OnceLock};
use std::time::Duration;

use anyhow::{Context as _, anyhow};
use tokio::sync::OnceCell;

use swarm_queue_client::{ClientSecret, QueueConfig};
use swarm_secret_client::{
    SecretStore,
    client::{DEFAULT_CERT_MOUNT, ENV_ADDR, ENV_CACERT, Settings},
    policy, queue,
};

/// Variable prefix for this agent's coordinates. Distinct from `HIVE_C0RE`'s
/// on purpose: an agent authenticates as its own client, not as its hive.
const ENV_PREFIX: &str = "HIVE_AGENT";

/// Resolved once at boot, so the publisher that connects to the queue reads
/// one answer rather than re-deriving it per call.
static CONFIG: OnceLock<Option<QueueConfig>> = OnceLock::new();

/// Names the agent whose queue secret this process reads. The cert-auth role
/// and the store path are both built from it by `swarm_secret_client`.
const ENV_AGENT_NAME: &str = "HIVE_AGENT_NAME";

/// How long one read of the store may take. The read runs inside a queue
/// connection attempt, and a store that never answers would otherwise stall
/// every reconnect behind it.
const STORE_READ_TIMEOUT: Duration = Duration::from_secs(10);

/// Where to present this agent's own credential, resolved once at boot.
static AGENT: OnceLock<Option<AgentPath>> = OnceLock::new();

/// The one connection every publisher in this process shares. Separate from
/// [`CONFIG`] because resolving the coordinates is synchronous boot work and
/// connecting is not — see [`client`].
static CLIENT: OnceCell<Option<Connection>> = OnceCell::const_new();

/// What connecting with this agent's own credential needs.
#[derive(Debug, PartialEq, Eq)]
struct AgentPath {
    url: String,
    ca_file: Option<PathBuf>,
    store: Arc<StoreSource>,
}

/// This agent's queue secret as the store holds it: where the store is, the
/// role this agent logs in as, and the path it reads.
#[derive(Debug, PartialEq, Eq)]
struct StoreSource {
    settings: Settings,
    role: String,
    path: String,
}

/// Where [`AgentSecret`] reads the secret from. A trait so the retry rules can
/// be exercised without a store.
trait SecretSource: Send + Sync + 'static {
    /// The store path, for log lines and errors. Never the value.
    fn origin(&self) -> &str;

    /// The value stored now, or `None` when nothing is stored.
    fn fetch(&self) -> impl Future<Output = anyhow::Result<Option<String>>> + Send;
}

impl SecretSource for Arc<StoreSource> {
    fn origin(&self) -> &str {
        &self.path
    }

    /// A fresh login per read: reads happen once per connection attempt, and a
    /// token held between them would be one more secret in memory to expire.
    async fn fetch(&self) -> anyhow::Result<Option<String>> {
        let store = SecretStore::connect(&self.settings, &self.role, DEFAULT_CERT_MOUNT)
            .await
            .context("logging in to the swarm secret store as this agent")?;
        // `read_optional`: only a 404 is "nothing stored". A 403 stays an
        // error, so a policy that stopped covering the path is not reported
        // as a secret that was deleted.
        let credential: Option<queue::AgentCredential> = store
            .read_optional(&self.path)
            .await
            .with_context(|| format!("reading {} from the store", self.path))?;
        Ok(credential.map(|c| c.value))
    }
}

/// This agent's queue secret, held in memory between connection attempts.
///
/// No `Debug`: [`Held`] carries the secret.
struct AgentSecret<S> {
    source: S,
    timeout: Duration,
    held: Mutex<Held>,
}

#[derive(Default)]
struct Held {
    value: Option<String>,
    /// `value` was read by [`AgentSecret::prime`] and no attempt has
    /// presented it yet.
    unpresented: bool,
}

impl<S: SecretSource> AgentSecret<S> {
    fn new(source: S) -> Self {
        Self {
            source,
            timeout: STORE_READ_TIMEOUT,
            held: Mutex::default(),
        }
    }

    fn held(&self) -> std::sync::MutexGuard<'_, Held> {
        // A poisoned lock only means another attempt panicked mid-update; the
        // worst it left is a stale value, which the next read replaces.
        self.held
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
    }

    /// Read the secret before the first connection attempt. An `Err` here is
    /// what sends [`connect_once`] to the hive's shared client.
    async fn prime(&self) -> anyhow::Result<()> {
        let value = self
            .read()
            .await?
            .with_context(|| format!("nothing is stored at {}", self.source.origin()))?;
        *self.held() = Held {
            value: Some(value),
            unpresented: true,
        };
        Ok(())
    }

    /// The secret to present on one connection attempt.
    ///
    /// The value [`prime`](Self::prime) read serves the first attempt. Every
    /// later attempt reads the store again, so after the queue refuses a
    /// re-minted secret the next attempt presents the current one. A store
    /// that cannot be read leaves the held value in use; a store that holds
    /// nothing drops it. Either way the attempt is one read, and the next one
    /// waits out the connection's reconnect backoff.
    async fn for_attempt(&self) -> anyhow::Result<String> {
        {
            let mut held = self.held();
            if std::mem::take(&mut held.unpresented)
                && let Some(value) = &held.value
            {
                return Ok(value.clone());
            }
        }
        match self.read().await {
            Ok(Some(value)) => {
                self.held().value = Some(value.clone());
                Ok(value)
            }
            Ok(None) => {
                self.held().value = None;
                tracing::warn!(
                    path = %self.source.origin(),
                    "nothing is stored at this agent's queue credential path; its queue \
                     connection retries under backoff"
                );
                Err(anyhow!("nothing is stored at {}", self.source.origin()))
            }
            Err(e) => {
                let held = self.held().value.clone();
                tracing::warn!(
                    error = format!("{e:#}"),
                    presenting_held = held.is_some(),
                    "re-reading this agent's queue credential from the store failed"
                );
                held.ok_or(e)
            }
        }
    }

    /// One bounded read. An empty stored value is no value.
    async fn read(&self) -> anyhow::Result<Option<String>> {
        let value = tokio::time::timeout(self.timeout, self.source.fetch())
            .await
            .map_err(|_| anyhow!("the store did not answer within {:?}", self.timeout))??;
        Ok(value.map(|v| v.trim().to_owned()).filter(|v| !v.is_empty()))
    }
}

/// Which credential the shared connection presented.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Presented {
    /// This agent's own, granted `<prefix>.<agent>`.
    Agent,
    /// The hive's shared client, granted `<prefix>.<hive>.>` for the hive the
    /// client id names.
    Hive { client_id: String },
}

/// The shared queue connection and the credential it was made with.
#[derive(Clone)]
pub struct Connection {
    pub client: async_nats::Client,
    pub presented: Presented,
}

/// The four variables the harness unit sets, before the client-id file is
/// read. Collected into a struct so [`decide`] is pure over them and the
/// process env is touched in exactly one place.
struct QueueEnv {
    nats_url: Option<String>,
    token_endpoint: Option<String>,
    client_id_file: Option<String>,
    client_secret_file: Option<String>,
    /// Independent of the all-or-none group below, exactly as in
    /// `QueueConfig::from_env`: a swarm behind a publicly-trusted
    /// certificate needs no extra anchor. Nothing in this tree sets it for
    /// an agent — a container already trusts the swarm root, which
    /// `hive_c0re::meta` embeds at build time — so it is here for a
    /// deployment that needs a different one, not for ours.
    ca_file: Option<String>,
}

impl QueueEnv {
    fn from_env() -> Self {
        let var = |suffix: &str| std::env::var(format!("{ENV_PREFIX}_{suffix}")).ok();
        Self {
            nats_url: var("NATS_URL"),
            token_endpoint: var("OIDC_TOKEN_ENDPOINT"),
            client_id_file: var("OIDC_CLIENT_ID_FILE"),
            client_secret_file: var("OIDC_CLIENT_SECRET_FILE"),
            ca_file: var("OIDC_CA_FILE"),
        }
    }
}

/// What the environment plus the client-id file add up to.
enum Resolution {
    /// Everything is here; the agent can reach the queue.
    Configured(Box<QueueConfig>),
    /// No queue coordinates for this agent, and that is a legal state — carries why.
    Absent(&'static str),
    /// Some of the environment, not all of it. A deployment bug rather than
    /// an absent integration, so it is reported and then survived.
    Partial,
}

/// Read the client id out of the file the credential landed at.
///
/// `None` for a missing or empty file, which is the ordinary state of a hive
/// whose secret store has nothing published yet — the credential simply is
/// not there, and nspawn forwards nothing. Trailing newline stripped: the
/// reader unit writes one and an id with a newline in it authenticates as
/// nobody.
fn read_client_id(path: &Path) -> Option<String> {
    match std::fs::read_to_string(path) {
        Ok(raw) => {
            let id = raw.trim();
            (!id.is_empty()).then(|| id.to_owned())
        }
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => None,
        Err(e) => {
            tracing::warn!(path = %path.display(), error = %e, "reading the queue client id failed");
            None
        }
    }
}

/// Where this agent reads its own queue secret from, or `None` when this
/// container was given no store.
///
/// Taken through a lookup so it can be asserted without mutating the process
/// environment. The shape follows `hive_matrix_mcp::credential`'s
/// `store_settings`, which reads the same variables for the same identity.
///
/// # Errors
/// A store address without the agent name or the identity beside it, which
/// is a half-delivered container rather than one without a store.
fn store_source(get: impl Fn(&str) -> Option<String>) -> anyhow::Result<Option<StoreSource>> {
    if get(ENV_ADDR).is_none_or(|v| v.is_empty()) {
        return Ok(None);
    }
    let agent = get(ENV_AGENT_NAME)
        .filter(|v| !v.is_empty())
        .with_context(|| format!("{ENV_AGENT_NAME} is unset or empty"))?;
    let settings = Settings::from_lookup(|k| get(k).filter(|v| k != ENV_CACERT || ca_is_usable(v)))
        .context("reading the swarm secret store's coordinates from the environment")?;
    Ok(Some(StoreSource {
        settings,
        role: policy::agent_object_name(&agent)?,
        path: queue::agent_queue_path(&agent)?,
    }))
}

/// Whether the CA bundle at `path` is a file with bytes in it. `BAO_CACERT`
/// names a systemd credential that may not have been delivered; absent means
/// the container's own trust store.
fn ca_is_usable(path: &str) -> bool {
    std::fs::metadata(path).is_ok_and(|m| m.len() > 0)
}

/// Decide whether this agent can connect with its own credential: it needs the
/// queue's address and a store to read its secret from, and nothing of the
/// hive's client.
fn decide_agent_path(env: &QueueEnv, store: Option<StoreSource>) -> Option<AgentPath> {
    Some(AgentPath {
        url: env.nats_url.clone()?,
        ca_file: env.ca_file.as_ref().map(Into::into),
        store: Arc::new(store?),
    })
}

/// Decide what this agent's queue configuration is, given the environment and
/// whatever the client-id file held.
///
/// All four variables or none, for the reason `QueueConfig::from_env` gives:
/// a half-set environment produces a process that comes up fine and never
/// connects. The client id is graded separately from the four because its
/// file has a legal absence the variables do not — the unit names the path a
/// credential *would* arrive at whether or not one has been published yet.
fn decide(env: &QueueEnv, client_id: Option<String>) -> Resolution {
    match (
        env.nats_url.as_ref(),
        env.token_endpoint.as_ref(),
        env.client_id_file.as_ref(),
        env.client_secret_file.as_ref(),
    ) {
        (None, None, None, None) => Resolution::Absent("this hive has no swarm queue configured"),
        (Some(url), Some(token_endpoint), Some(_), Some(secret)) => match client_id {
            Some(client_id) => Resolution::Configured(Box::new(QueueConfig {
                url: url.clone(),
                token_endpoint: token_endpoint.clone(),
                client_id,
                client_secret: ClientSecret::File(secret.into()),
                ca_file: env.ca_file.as_ref().map(Into::into),
            })),
            None => {
                Resolution::Absent("the queue credential has not been published to this hive yet")
            }
        },
        _ => Resolution::Partial,
    }
}

/// Resolve this agent's queue configuration and record it for later use.
///
/// Never fails: a harness that cannot reach the queue still serves its
/// operator, its web UI and its turn loop, so every outcome here is a log
/// line and not an exit.
pub fn init() {
    let env = QueueEnv::from_env();
    let client_id = env
        .client_id_file
        .as_deref()
        .map(Path::new)
        .and_then(read_client_id);
    let resolved = match decide(&env, client_id) {
        Resolution::Configured(cfg) => {
            // url + client id only. The secret is a path in this struct and
            // stays one: neither it nor its contents belong in a log.
            tracing::info!(url = %cfg.url, client_id = %cfg.client_id, "swarm queue configured");
            Some(*cfg)
        }
        Resolution::Absent(why) => {
            tracing::info!(why, "no swarm queue coordinates for this agent");
            None
        }
        Resolution::Partial => {
            tracing::error!(
                prefix = ENV_PREFIX,
                "swarm queue half-configured: {ENV_PREFIX}_NATS_URL, \
                 {ENV_PREFIX}_OIDC_TOKEN_ENDPOINT, {ENV_PREFIX}_OIDC_CLIENT_ID_FILE and \
                 {ENV_PREFIX}_OIDC_CLIENT_SECRET_FILE are set together or not at all — \
                 this agent will not connect"
            );
            None
        }
    };
    let _ = CONFIG.set(resolved);

    // Independent of everything above: this agent may hold its own secret on
    // a hive whose shared client is not published yet, or the shared client
    // and no secret of its own.
    let store = store_source(|k| std::env::var(k).ok()).unwrap_or_else(|e| {
        tracing::warn!(
            error = format!("{e:#}"),
            "this agent's secret store coordinates are incomplete, so it cannot read \
             its own swarm queue credential"
        );
        None
    });
    if let Some(store) = &store {
        tracing::info!(
            path = %store.path,
            "this agent reads its own swarm queue credential from the secret store"
        );
    } else {
        tracing::info!(
            "no secret store for this agent; it is known to the queue by its hive's \
             shared client"
        );
    }
    let _ = AGENT.set(decide_agent_path(&env, store));
}

/// Whether this agent has any credential to reach the queue with. `false`
/// before [`init`] has run.
pub fn configured() -> bool {
    config().is_some() || AGENT.get().is_some_and(Option::is_some)
}

/// What [`init`] resolved, or `None` when this agent has no queue.
///
/// Borrowed from the `OnceLock` rather than cloned: a caller wants the client
/// id to derive its subject from, and handing out an owned copy of a struct
/// holding a credential path invites it being stored somewhere with a
/// different lifetime than the one place that owns it.
///
/// `None` before [`init`] has run, which is the same answer as "no queue" and
/// deliberately not a panic — the ordering is a boot detail, and a harness
/// that reordered its boot should lose the queue, not die.
pub fn config() -> Option<&'static QueueConfig> {
    CONFIG.get()?.as_ref()
}

/// The shared queue connection, made on first call and memoized for the rest
/// of the process.
///
/// One connection per process, not per publisher: the agent authenticates as
/// one client, so a second `connect` would be a second token mint and a second
/// live connection for the same identity rather than a second credential.
///
/// `None` covers both "no queue coordinates" and "configured but the connect
/// failed" — a caller does nothing differently between them, since either way
/// there is nothing to publish onto. Connecting is lazy so that an agent on a
/// hive with no queue pays nothing at boot.
pub async fn client() -> Option<Connection> {
    CLIENT.get_or_init(connect_once).await.clone()
}

/// This agent's own credential first, then the hive's. Each path taken is
/// logged once, here.
async fn connect_once() -> Option<Connection> {
    if let Some(agent) = AGENT.get().and_then(Option::as_ref) {
        let secret = AgentSecret::new(Arc::clone(&agent.store));
        let name = crate::identity::label();
        match connect_as_agent(&agent.url, agent.ca_file.as_deref(), name, secret).await {
            Ok(client) => {
                tracing::info!(
                    url = %agent.url,
                    "connected to the swarm queue with this agent's own credential"
                );
                return Some(Connection {
                    client,
                    presented: Presented::Agent,
                });
            }
            Err(e) => tracing::warn!(
                error = format!("{e:#}"),
                "connecting with this agent's own credential failed; falling back to \
                 its hive's shared client"
            ),
        }
    }
    let cfg = config()?;
    let inbox = swarm_queue_client::client_inbox_prefix(&cfg.client_id);
    match swarm_queue_client::connect_with_inbox_prefix(cfg.clone(), inbox).await {
        Ok(client) => {
            tracing::info!(
                url = %cfg.url,
                client_id = %cfg.client_id,
                "connecting to the swarm queue with the hive's shared client; the \
                 client retries in the background until the queue accepts it"
            );
            Some(Connection {
                client,
                presented: Presented::Hive {
                    client_id: cfg.client_id.clone(),
                },
            })
        }
        Err(e) => {
            // `chain`, not `{:#}`: this is `swarm_queue_client::Error`, whose
            // `Display` ignores the alternate flag, so `{:#}` renders the
            // headline and drops the cause that says which half failed.
            tracing::warn!(
                error = %swarm_queue_client::chain(&e),
                "swarm queue connect failed; this agent publishes nothing upward"
            );
            None
        }
    }
}

/// Connect presenting this agent's own token as `name`.
///
/// The secret is read before the first attempt, and any failure of that read
/// or of the first attempt, a refusal included, is returned for
/// [`connect_once`] to fall back on. Once connected, each reconnect attempt
/// takes its secret from [`AgentSecret::for_attempt`].
async fn connect_as_agent<S: SecretSource>(
    url: &str,
    ca_file: Option<&Path>,
    name: String,
    secret: AgentSecret<S>,
) -> anyhow::Result<async_nats::Client> {
    secret.prime().await?;
    let secret = Arc::new(secret);
    let agent = name.clone();
    swarm_queue_client::connect_with_token(url, ca_file, &agent, move || {
        let secret = Arc::clone(&secret);
        let name = name.clone();
        // Spawned so the future handed back is only a `JoinHandle`, which is
        // `Sync` as async-nats's auth callback requires; the store read is not.
        let attempt = tokio::spawn(async move {
            let value = secret.for_attempt().await.map_err(|e| format!("{e:#}"))?;
            swarm_queue_client::agent_token::format_agent_token(&name, &value)
                .map_err(|e| e.to_string())
        });
        async move { attempt.await.map_err(|e| e.to_string())? }
    })
    .await
    .map_err(|e| anyhow!(swarm_queue_client::chain(&e)))
}
#[cfg(test)]
mod tests {
    use std::collections::VecDeque;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::sync::{Arc, Mutex};
    use std::time::Duration;

    use anyhow::anyhow;

    use super::{
        AgentPath, AgentSecret, ClientSecret, QueueEnv, Resolution, SecretSource, StoreSource,
        connect_as_agent, decide, decide_agent_path, read_client_id, store_source,
    };

    fn env(parts: [Option<&str>; 4]) -> QueueEnv {
        let [nats_url, token_endpoint, client_id_file, client_secret_file] = parts;
        QueueEnv {
            nats_url: nats_url.map(str::to_owned),
            token_endpoint: token_endpoint.map(str::to_owned),
            client_id_file: client_id_file.map(str::to_owned),
            client_secret_file: client_secret_file.map(str::to_owned),
            ca_file: None,
        }
    }

    fn full() -> QueueEnv {
        env([
            Some("nats://10.42.0.1:4222"),
            Some("https://auth.t.local/api/oidc/token"),
            Some("/run/credentials/hive-agent.service/hive-queue-agent-client-id"),
            Some("/run/credentials/hive-agent.service/hive-queue-agent-secret"),
        ])
    }

    /// The reader unit writes the id with a trailing newline; an id carrying
    /// one is a client authelia has never heard of.
    #[test]
    fn a_client_id_file_is_read_without_its_newline() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("client_id");
        std::fs::write(&path, "hive-h1-agent\n").expect("write");
        assert_eq!(read_client_id(&path).as_deref(), Some("hive-h1-agent"));
    }

    /// Both shapes of "no credential here": never delivered, or delivered
    /// empty. Neither is an id, and treating an empty string as one would
    /// authenticate as the anonymous client rather than failing.
    #[test]
    fn a_missing_or_empty_client_id_file_reads_as_no_id() {
        let dir = tempfile::tempdir().expect("tempdir");
        let missing = dir.path().join("client_id");
        assert_eq!(read_client_id(&missing), None);
        std::fs::write(&missing, "\n").expect("write");
        assert_eq!(read_client_id(&missing), None);
    }

    #[test]
    fn a_complete_environment_with_a_published_id_configures_the_queue() {
        let e = full();
        let Resolution::Configured(cfg) = decide(&e, Some("hive-h1-agent".to_owned())) else {
            panic!("expected a configured queue");
        };
        assert_eq!(cfg.url, "nats://10.42.0.1:4222");
        assert_eq!(cfg.client_id, "hive-h1-agent");
        let ClientSecret::File(path) = &cfg.client_secret else {
            panic!("the secret stays a path: {:?}", cfg.client_secret);
        };
        assert!(
            path.ends_with("hive-queue-agent-secret"),
            "the secret stays a path: {}",
            path.display()
        );
    }

    /// A hive that has not been given its queue's coordinates at all. Silence
    /// here is correct, and it has to be distinguishable from the half-set
    /// case below — that distinction is the only thing that makes the error
    /// branch worth logging.
    #[test]
    fn an_empty_environment_is_no_queue_rather_than_an_error() {
        let e = env([None, None, None, None]);
        assert!(matches!(decide(&e, None), Resolution::Absent(_)));
    }

    /// The credential's own absence. The unit names the path unconditionally
    /// once the hive has a queue, so this is the state of every agent on a
    /// swarm whose publisher has not run — legal, and not the error branch.
    #[test]
    fn a_complete_environment_with_no_published_id_is_no_queue() {
        let e = full();
        assert!(matches!(decide(&e, None), Resolution::Absent(_)));
    }

    /// Each single-variable omission, because the failure a partial set
    /// produces is a harness that looks healthy and publishes nothing.
    #[test]
    fn any_missing_variable_is_a_partial_configuration() {
        for drop in 0..4 {
            let mut parts = [
                Some("nats://10.42.0.1:4222"),
                Some("https://auth.t.local/api/oidc/token"),
                Some("/run/credentials/hive-agent.service/hive-queue-agent-client-id"),
                Some("/run/credentials/hive-agent.service/hive-queue-agent-secret"),
            ];
            parts[drop] = None;
            let e = env(parts);
            assert!(
                matches!(
                    decide(&e, Some("hive-h1-agent".to_owned())),
                    Resolution::Partial
                ),
                "dropping variable {drop} must report a partial configuration"
            );
        }
    }

    /// A lookup standing in for a container that was handed a store.
    fn with_store(k: &str) -> Option<String> {
        match k {
            "BAO_ADDR" => Some("https://bao.t.local:8200".to_owned()),
            "BAO_CLIENT_CERT" => Some("/run/credentials/x/cert".to_owned()),
            "BAO_CLIENT_KEY" => Some("/run/credentials/x/key".to_owned()),
            "HIVE_AGENT_NAME" => Some("a1".to_owned()),
            _ => None,
        }
    }

    fn source() -> StoreSource {
        store_source(with_store)
            .expect("a complete store environment")
            .expect("a store address was given")
    }

    #[test]
    fn a_store_address_reads_the_agents_own_path_as_its_own_role() {
        let s = source();
        assert_eq!(s.path, "swarm/agents/a1/queue");
        assert_eq!(s.role, "hive-agent-a1");
    }

    /// No address: this container was given no store, so there is nothing to
    /// read from and nothing wrong.
    #[test]
    fn no_store_address_is_no_store() {
        let none = store_source(|k| if k == "BAO_ADDR" { None } else { with_store(k) });
        assert!(none.expect("absent is not an error").is_none());
        let empty = store_source(|k| {
            if k == "BAO_ADDR" {
                Some(String::new())
            } else {
                with_store(k)
            }
        });
        assert!(empty.expect("empty is absent").is_none());
    }

    #[test]
    fn a_store_address_without_the_agent_or_its_identity_is_an_error() {
        for var in ["HIVE_AGENT_NAME", "BAO_CLIENT_CERT", "BAO_CLIENT_KEY"] {
            let e = store_source(|k| if k == var { None } else { with_store(k) })
                .expect_err("a half-delivered store");
            assert!(format!("{e:#}").contains(var), "dropping {var}: {e:#}");
        }
    }

    /// `BAO_CACERT` is always named and may point at nothing; that is the
    /// container's own trust store, not a CA path that fails the handshake.
    #[test]
    fn an_undelivered_ca_is_dropped() {
        let with_missing_ca = store_source(|k| {
            if k == "BAO_CACERT" {
                Some("/nonexistent/ca".to_owned())
            } else {
                with_store(k)
            }
        })
        .expect("complete")
        .expect("given a store");
        assert_eq!(with_missing_ca, source());
    }

    /// The per-agent secret is resolved by a separate switch from the hive's
    /// client: an agent whose hive has no queue client can still read its own
    /// swarm-minted secret, and needs only the queue's address for it.
    #[test]
    fn an_agent_with_a_store_and_the_queue_address_connects_as_itself() {
        let url_only = env([Some("nats://10.42.0.1:4222"), None, None, None]);
        assert!(!matches!(
            decide(&url_only, None),
            Resolution::Configured(_)
        ));
        assert_eq!(
            decide_agent_path(&url_only, Some(source())),
            Some(AgentPath {
                url: "nats://10.42.0.1:4222".to_owned(),
                ca_file: None,
                store: Arc::new(source()),
            })
        );
    }

    #[test]
    fn without_a_store_or_the_queue_address_an_agent_does_not_connect_as_itself() {
        assert_eq!(decide_agent_path(&full(), None), None);
        assert_eq!(
            decide_agent_path(&env([None, None, None, None]), Some(source())),
            None
        );
    }

    enum Answer {
        Value(&'static str),
        Absent,
        Fails,
        Hangs,
    }

    /// A store that answers each read with the next scripted answer and
    /// counts the reads.
    struct Scripted {
        answers: Mutex<VecDeque<Answer>>,
        reads: AtomicUsize,
    }

    fn scripted(answers: impl IntoIterator<Item = Answer>) -> Arc<Scripted> {
        Arc::new(Scripted {
            answers: Mutex::new(answers.into_iter().collect()),
            reads: AtomicUsize::new(0),
        })
    }

    impl SecretSource for Arc<Scripted> {
        fn origin(&self) -> &'static str {
            "swarm/agents/a1/queue"
        }

        async fn fetch(&self) -> anyhow::Result<Option<String>> {
            self.reads.fetch_add(1, Ordering::SeqCst);
            let next = self.answers.lock().expect("lock").pop_front();
            match next.expect("a read nobody scripted") {
                Answer::Value(v) => Ok(Some(v.to_owned())),
                Answer::Absent => Ok(None),
                Answer::Fails => Err(anyhow!("the store is unreachable")),
                Answer::Hangs => std::future::pending().await,
            }
        }
    }

    fn reads(s: &Arc<Scripted>) -> usize {
        s.reads.load(Ordering::SeqCst)
    }

    /// (a) The first attempt presents what was just read, and does not read
    /// the store a second time for it.
    #[tokio::test]
    async fn the_first_attempt_presents_the_value_read_before_it() {
        let store = scripted([Answer::Value("v1\n")]);
        let secret = AgentSecret::new(Arc::clone(&store));
        secret.prime().await.expect("the store holds a secret");
        assert_eq!(secret.for_attempt().await.expect("primed"), "v1");
        assert_eq!(reads(&store), 1);
    }

    /// (b) After the queue refuses a re-minted secret, the next attempt reads
    /// the store and presents what it holds now: one read per attempt.
    #[tokio::test]
    async fn each_later_attempt_presents_what_the_store_holds_now() {
        let store = scripted([
            Answer::Value("v1"),
            Answer::Value("v1"),
            Answer::Value("v2"),
        ]);
        let secret = AgentSecret::new(Arc::clone(&store));
        secret.prime().await.expect("the store holds a secret");
        assert_eq!(secret.for_attempt().await.expect("primed"), "v1");
        assert_eq!(secret.for_attempt().await.expect("re-read"), "v1");
        assert_eq!(secret.for_attempt().await.expect("re-read"), "v2");
        assert_eq!(reads(&store), 3);
    }

    /// (b) Bounded: a store that never answers costs one attempt the read
    /// timeout, and the attempt goes ahead with the value it already holds.
    #[tokio::test]
    async fn a_store_that_does_not_answer_is_given_up_on() {
        let store = scripted([Answer::Value("v1"), Answer::Hangs, Answer::Hangs]);
        let mut secret = AgentSecret::new(Arc::clone(&store));
        secret.timeout = Duration::from_millis(20);
        secret.prime().await.expect("the store holds a secret");
        assert_eq!(secret.for_attempt().await.expect("primed"), "v1");
        assert_eq!(
            secret
                .for_attempt()
                .await
                .expect("the held value stands in"),
            "v1"
        );
        assert_eq!(reads(&store), 2);

        let cold = AgentSecret::new(scripted([Answer::Hangs]));
        let mut cold = cold;
        cold.timeout = Duration::from_millis(20);
        let e = cold.prime().await.expect_err("nothing to stand in");
        assert!(format!("{e:#}").contains("did not answer"), "{e:#}");
    }

    /// (b) A secret the store no longer holds (the agent was destroyed) is not
    /// presented again, even when a later read fails.
    #[tokio::test]
    async fn a_secret_removed_from_the_store_is_dropped() {
        let store = scripted([Answer::Value("v1"), Answer::Absent, Answer::Fails]);
        let secret = AgentSecret::new(Arc::clone(&store));
        secret.prime().await.expect("the store holds a secret");
        assert_eq!(secret.for_attempt().await.expect("primed"), "v1");
        let gone = secret.for_attempt().await.expect_err("nothing stored");
        assert!(
            format!("{gone:#}").contains("swarm/agents/a1/queue"),
            "{gone:#}"
        );
        secret
            .for_attempt()
            .await
            .expect_err("the dropped value does not come back");
    }

    /// (c) Nothing to connect with before the first attempt is an error, which
    /// is what sends `connect_once` to the hive's shared client. The queue is
    /// never dialled: nothing listens at the address given.
    #[tokio::test]
    async fn no_secret_before_the_first_connect_falls_back() {
        for (answer, says) in [
            (Answer::Fails, "unreachable"),
            (Answer::Absent, "nothing is stored"),
            (Answer::Value("  \n"), "nothing is stored"),
        ] {
            let secret = AgentSecret::new(scripted([answer]));
            let e = connect_as_agent("nats://127.0.0.1:9", None, "a1".to_owned(), secret)
                .await
                .expect_err("no secret to present");
            assert!(format!("{e:#}").contains(says), "{e:#}");
        }
    }

    /// Serve one NATS handshake on `sock`: send INFO, read CONNECT up to the
    /// PING, and return the token presented.
    async fn handshake(sock: &mut tokio::net::TcpStream, port: u16) -> String {
        use tokio::io::{AsyncBufReadExt as _, AsyncWriteExt as _};
        let info = format!(
            "INFO {{\"server_id\":\"fake\",\"version\":\"2.10.0\",\"go\":\"go\",\
             \"host\":\"127.0.0.1\",\"port\":{port},\"headers\":true,\
             \"max_payload\":1048576,\"proto\":1,\"auth_required\":true,\
             \"nonce\":\"n0nce\"}}\r\n"
        );
        sock.write_all(info.as_bytes()).await.expect("send INFO");
        let mut lines = tokio::io::BufReader::new(sock).lines();
        let mut token = String::new();
        while let Some(line) = lines.next_line().await.expect("read") {
            if let Some(json) = line.strip_prefix("CONNECT ") {
                let v: serde_json::Value = serde_json::from_str(json).expect("CONNECT is JSON");
                token = v["auth_token"].as_str().unwrap_or_default().to_owned();
            } else if line.starts_with("PING") {
                break;
            }
        }
        token
    }

    /// (b) end to end through async-nats: connected with `v1`, the connection
    /// drops, the reconnect presenting `v1` is refused, and the next attempt
    /// reads `v2` and is accepted.
    #[tokio::test]
    async fn a_refused_reconnect_reconnects_with_the_secret_read_after_it() {
        use tokio::io::AsyncWriteExt as _;

        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("bind");
        let port = listener.local_addr().expect("addr").port();
        let (seen_tx, mut seen_rx) = tokio::sync::mpsc::unbounded_channel();
        tokio::spawn(async move {
            let mut kept = Vec::new();
            for n in 0.. {
                let (mut sock, _) = listener.accept().await.expect("accept");
                let token = handshake(&mut sock, port).await;
                let accept = n == 0 || token.rsplit_once('.').is_some_and(|(_, s)| s == "v2");
                let reply: &[u8] = if accept {
                    b"PONG\r\n"
                } else {
                    b"-ERR 'Authorization Violation'\r\n"
                };
                let _ = sock.write_all(reply).await;
                let _ = seen_tx.send(token);
                // The first connection is dropped to force a reconnect; the
                // one accepted after the refusal stays open.
                if accept && n > 0 {
                    kept.push(sock);
                }
            }
        });

        let store = scripted([
            Answer::Value("v1"),
            Answer::Value("v1"),
            Answer::Value("v2"),
        ]);
        let secret = AgentSecret::new(Arc::clone(&store));
        let url = format!("nats://127.0.0.1:{port}");
        let _client = connect_as_agent(&url, None, "a1".to_owned(), secret)
            .await
            .expect("the first connect is accepted");

        let mut seen = Vec::new();
        while seen.len() < 3 {
            let token = tokio::time::timeout(Duration::from_secs(20), seen_rx.recv())
                .await
                .expect("the client keeps reconnecting")
                .expect("the server is running");
            seen.push(token);
        }
        let secrets: Vec<&str> = seen
            .iter()
            .map(|t| t.rsplit_once('.').map_or("", |(_, s)| s))
            .collect();
        assert_eq!(secrets, ["v1", "v1", "v2"]);
        assert_eq!(reads(&store), 3);
    }
}