hyperhive/swarm-queue-client/src/lib.rs

//! Connecting to the swarm message queue as an authenticated client.
//!
//! Shared by every process that needs the queue — the swarm controller reads
//! hive status out of it, a hive publishes its own status into it — because
//! the *connect* is identical for all of them and only the use differs.
//! Duplicating it per binary would put credential handling in two places, and
//! a token-refresh fix would then have to be found twice.
//!
//! The queue admits every non-responder client through `auth_callout`: a
//! client presents a token at CONNECT, the callout responder introspects it
//! against authelia and mints a user JWT if it is good. So each participant is
//! an ordinary client that needs an identity of its own — the controller is
//! not a hive, and the per-hive clients issued from the roster are not its to
//! use.
//!
//! Two things about that shape drive everything here:
//!
//! - **A token expires.** Authelia issues `client_credentials` access tokens
//!   with `expires_in: 3599`. Authentication happens at CONNECT, and the user
//!   JWT `swarm-nats-auth` issues expires no later than the token does, so
//!   the server closes the connection within the hour and the *reconnect*
//!   needs a token minted after it.
//! - **`async-nats` re-runs an auth callback per connection attempt** (it is
//!   handed that attempt's nonce). So the refresh belongs in the callback and
//!   not in a timer: there is no window in which the client presents a token
//!   that has expired since it was minted.
//!
//! ⚠️ That second point is also a trap that cost a production incident —
//! **anything in a retry path runs at the FAILURE rate.** See `CachedToken`
//! and `MAX_RECONNECT_DELAY`.

use std::path::PathBuf;

use swarm_secret_client::stored_secret::{self, StoredSecret};

/// Everything that can go wrong reaching the swarm queue.
///
/// `thiserror` and not `anyhow` because this is a library: a caller gets a
/// type it can match on, and the binaries that consume it keep using
/// `anyhow` — `?` converts for free, so nothing downstream is more verbose
/// for it. The same split `hive-claude` uses.
///
/// The variants are the failures an operator acts on differently: an
/// unreadable secret file is a deployment bug, a refused token is an
/// identity-provider config problem, an unreachable queue is a network
/// one. Collapsing them into one string would make that distinction a
/// matter of reading prose.
#[derive(Debug, thiserror::Error)]
pub enum Error {
    #[error("reading the queue client secret from {path}")]
    ClientSecret {
        path: String,
        #[source]
        source: std::io::Error,
    },

    #[error("building the token-endpoint HTTP client")]
    HttpClient(#[source] reqwest::Error),

    /// Distinct from `HttpClient` because the operator's next move differs:
    /// this one names a path they configured, and it fires before any
    /// network call.
    #[error("reading the token-endpoint CA certificate from {path}")]
    CaFile {
        path: String,
        #[source]
        source: std::io::Error,
    },

    #[error("parsing the token-endpoint CA certificate from {path} as PEM")]
    CaParse {
        path: String,
        #[source]
        source: reqwest::Error,
    },

    #[error("requesting an access token from authelia")]
    TokenRequest(#[source] reqwest::Error),

    /// The body carries authelia's own `error_description`, and it is far
    /// more useful than the status alone: a wrong grant says
    /// `unauthorized_client`, a wrong secret says `invalid_client`, and
    /// those point at different config.
    #[error("authelia refused the token request ({status}): {body}")]
    TokenRefused {
        status: reqwest::StatusCode,
        body: String,
    },

    #[error("parsing authelia's token response")]
    TokenResponse(#[source] serde_json::Error),

    #[error("connecting to the swarm queue at {url}")]
    Connect {
        url: String,
        #[source]
        source: async_nats::ConnectError,
    },

    /// Distinct from [`Error::Connect`] on purpose: that one is a connect
    /// that was attempted and refused, this one is a request made before
    /// any connection exists. The second is a caller-ordering problem and
    /// the first is a deployment one.
    #[error("not connected to the swarm queue (client state: {0:?})")]
    NotConnected(async_nats::connection::State),

    #[cfg(feature = "kv")]
    #[error("creating the {bucket} bucket")]
    CreateBucket {
        // `String` rather than `&'static str`: a per-hive bucket name is built
        // at runtime (`wanted::bucket`), so the error has to own it.
        bucket: String,
        #[source]
        source: async_nats::jetstream::context::CreateKeyValueError,
    },

    #[cfg(feature = "kv")]
    #[error("opening the {bucket} bucket")]
    OpenBucket {
        bucket: String,
        #[source]
        source: async_nats::jetstream::context::KeyValueError,
    },

    #[cfg(feature = "kv")]
    #[error("watching key {key}")]
    WatchKey {
        key: String,
        #[source]
        source: async_nats::jetstream::kv::WatchError,
    },

    #[cfg(feature = "notices")]
    #[error("creating the {stream} stream")]
    CreateStream {
        stream: &'static str,
        #[source]
        source: async_nats::jetstream::context::CreateStreamError,
    },

    #[cfg(any(feature = "subagent-term", feature = "agent-term"))]
    #[error("creating the {stream} stream")]
    CreateAgentStream {
        // Owned: the name carries the agent's (`subagent_term::stream_name`,
        // `agent_term::stream_name`).
        stream: String,
        #[source]
        source: async_nats::jetstream::context::CreateStreamError,
    },
}

/// Render an error and its source chain on one line.
///
/// **Use this instead of `{:#}` anywhere an [`Error`] is rendered without
/// first being `?`-converted into an `anyhow::Error`.** `anyhow`'s
/// `Display` special-cases `f.alternate()` to walk the source chain;
/// `thiserror`'s derive does not, so `{e:#}` and `{e}` render
/// identically for this type. A call site that formatted an
/// `anyhow::Error` with `{:#}` and now holds an [`Error`] therefore keeps
/// compiling, keeps looking right, and silently drops the cause — which
/// is the half that says *why*, and is exactly what a one-shot boot
/// warning with no retry needs most.
///
/// Public for that reason: the fix cannot live only inside this crate's
/// own auth callback while the callers it was written for reach for
/// `{:#}` and get nothing.
pub fn chain(error: &dyn std::error::Error) -> String {
    let mut rendered = error.to_string();
    let mut source = error.source();
    while let Some(cause) = source {
        rendered.push_str(": ");
        rendered.push_str(&cause.to_string());
        source = cause.source();
    }
    rendered
}

/// The hive-status KV bucket, shared by the hive that writes it and the
/// controller that reads it. See the module doc for why a bucket name and
/// its config belong to neither end alone.
///
/// The module itself is unconditional; only the parts that *open* the bucket
/// need the `kv` feature. The name is a `&str` with no dependencies, and a
/// third end names it too — the auth-callout responder, which derives the
/// subjects a hive may publish to from it without ever speaking `jetstream`.
/// Gating the name behind `kv` would have forced that consumer to choose
/// between pulling a JetStream stack it does not use and copying the literal,
/// which is the disagreement this module exists to prevent.
pub mod status;

/// The bucket the controller declares each hive's wanted agent set in — the
/// opposite direction from [`status`], and deliberately not a mirror of it.
/// Status is *observed* and republishes itself; this is *declared* and nothing
/// recreates it, which is why the module documents its own durability and its
/// "absence is not a deletion order" rule rather than inheriting either.
pub mod wanted;

/// The bucket per-agent status snapshots are published into — one key per
/// `(hive, agent)`, deliberately a sibling of [`status`] rather than a field
/// inside it. See the module doc for why the two must not merge.
pub mod agent_status;

/// The token an agent presents its own queue secret in. Store-free, so an agent
/// formats it without linking the secret-store client.
pub mod agent_token;

/// The bucket agent icons are published into — one key per agent, with no
/// hive in it, unlike [`agent_status`]. See the module doc for why the
/// placement stays out of the key, and for who may write it.
pub mod agent_icon;

/// The bucket each agent mirrors its `AGENTS.md` into, and the edit command
/// for it. The names and types are unconditional; opening the bucket needs
/// `kv`.
pub mod agents_md;

/// The bucket swarm-controller declares each agent's scheduled prompts in —
/// one key per agent, read by that agent alone. The names are unconditional;
/// writing needs `kv` and reading needs `agent-schedules`.
pub mod agent_schedules;

/// The bucket swarm-controller sets each agent's model, effort and subagent
/// models in — one key per agent, read by that agent alone. The names are
/// unconditional; writing needs `kv` and reading needs `agent-inference`.
pub mod agent_inference;

/// The bucket swarm-controller sets each agent's CPU and memory caps in — one
/// key per agent, read by the hive that runs it. The names are unconditional;
/// writing needs `kv` and reading needs `agent-limits`.
pub mod agent_limits;

/// The bucket swarm-controller sets each agent's permissions in — one key
/// per agent, read by the hive that runs it, plus the one-shot import a hive
/// sends. The names are unconditional; writing needs `kv` and the stored type
/// and reading need `agent-permissions`.
pub mod agent_permissions;

/// The bucket swarm-controller publishes each operator's name and forge-profile
/// pronouns in — one key holding the whole list, read by every agent. The names are
/// unconditional; writing needs `kv` and reading needs `operator-profiles`.
pub mod operator_profiles;

/// The bucket swarm-controller records each archetype agent's archetype in
/// — one key per agent, read by the hive that runs it. The names and the
/// stored type are unconditional; writing needs `kv` and reading needs
/// `agent-archetype`.
pub mod agent_archetype;

/// The login flag in an agent's turn-state header. No bucket: it travels
/// on [`agent_state_subject`] only.
pub mod agent_login;

/// The bucket swarm-controller keeps the evaluated agent archetypes in — one
/// key per source flake. The names and the stored type are unconditional;
/// opening needs `kv`.
pub mod archetypes;

#[cfg(feature = "kv")]
mod kv;

/// The subject the swarm controller publishes on when the hive-wide knowledge
/// repository has changed. One writer, many readers — every hive subscribes.
///
/// Here rather than in a module of its own for the same reason as the bucket
/// name above: three crates must agree on the string, and the one that agrees
/// hardest — the auth-callout responder, which decides whether a publish is
/// permitted at all — speaks neither `jetstream` nor `kv`.
pub const KNOWLEDGE_SUBJECT: &str = "$SWARM.knowledge";

/// The subject the swarm controller publishes on to ask `hive` to rebuild one
/// of its agents. Lives here for the same three-crate reason as
/// [`KNOWLEDGE_SUBJECT`], but unlike it this is a **family, one subject per
/// hive** — a hive subscribes to its own and is never woken by a deploy meant
/// for someone else. The queue grants each hive subscribe on its own subject
/// only, so it cannot read another hive's deploys either.
#[must_use]
pub fn deploy_subject(hive: &str) -> String {
    format!("{DEPLOY_SUBJECT_PREFIX}.{hive}")
}

/// The publish grant covering every [`deploy_subject`], for the one client
/// that may send them. A wildcard rather than a subject per hive because the
/// responder has no roster — it cannot enumerate hives, and a grant it had to
/// keep in step with one would be a second place to get the list wrong.
pub const DEPLOY_SUBJECT_WILDCARD: &str = "$SWARM.deploy.*";

/// Shared by [`deploy_subject`] and [`DEPLOY_SUBJECT_WILDCARD`] so the two
/// cannot drift into naming different families.
const DEPLOY_SUBJECT_PREFIX: &str = "$SWARM.deploy";

/// What a [`deploy_subject`] message carries.
///
/// No hive: the subject already names it, and repeating it here would be two
/// places stating one fact, free to disagree.
///
/// With [`Self::rev`] set it names the commit of the agent's config repo to
/// deploy. The config itself stays in git: the hive fetches that commit from
/// the forge, so a hive that misses a message is behind, never wrong.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct DeployRequest {
    /// The agent to rebuild, as the swarm knows it.
    pub agent: String,
    /// The commit on the agent's config repo `main` to deploy. `None`
    /// rebuilds whatever the hive already has applied.
    ///
    /// `serde(default)` so a payload without the field still decodes: the
    /// controller and the hives are deployed independently.
    #[serde(default)]
    pub rev: Option<String>,
}

/// The subject the swarm controller sends an operator's input for `agent`
/// on. That agent's harness subscribes to its own and turns each message into
/// an inbox turn.
///
/// Its own family rather than a leaf under `$SWARM.term.`, so no agent's
/// terminal grant covers it: the harness trusts [`AgentInput::from`] only
/// because the controller is the one client granted publish here.
#[must_use]
pub fn agent_input_subject(agent: &str) -> String {
    format!("{AGENT_INPUT_SUBJECT_PREFIX}.{agent}")
}

/// The publish grant covering every [`agent_input_subject`], for the one
/// client that may send them. A wildcard for the same no-roster reason as
/// [`DEPLOY_SUBJECT_WILDCARD`].
pub const AGENT_INPUT_SUBJECT_WILDCARD: &str = "$SWARM.input.*";

/// Shared by [`agent_input_subject`] and [`AGENT_INPUT_SUBJECT_WILDCARD`].
const AGENT_INPUT_SUBJECT_PREFIX: &str = "$SWARM.input";

/// The subject the swarm controller sends an operator's control command for
/// `agent` on (see [`AgentControl`]).
///
/// A family of its own, apart from [`agent_input_subject`]: a grant to send
/// text never covers a command that stops a turn, and the other way round.
/// Like the input subject it runs controller to agent over the swarm queue
/// with no hive in the path.
#[must_use]
pub fn agent_control_subject(agent: &str) -> String {
    format!("{AGENT_CONTROL_SUBJECT_PREFIX}.{agent}")
}

/// The publish grant covering every [`agent_control_subject`], for the one
/// client that may send them. A wildcard for the same no-roster reason as
/// [`AGENT_INPUT_SUBJECT_WILDCARD`].
pub const AGENT_CONTROL_SUBJECT_WILDCARD: &str = "$SWARM.control.*";

/// Shared by [`agent_control_subject`] and [`AGENT_CONTROL_SUBJECT_WILDCARD`].
const AGENT_CONTROL_SUBJECT_PREFIX: &str = "$SWARM.control";

/// The subject `agent` publishes its terminal rows on. One leaf per agent, so
/// a subscriber can follow one agent without filtering the swarm's whole
/// terminal traffic.
///
/// Shared by the publisher (`hive-agent`), the relay that streams it to
/// swarm-ui (`swarm-controller`) and the queue's subscribe grant for readers
/// (`swarm-nats-auth`), which spells [`term_subject_wildcard`] — so a rename
/// moves all three together.
#[must_use]
pub fn term_subject(agent: &str) -> String {
    format!("{TERM_SUBJECT_PREFIX}.{agent}")
}

/// Every [`term_subject`] at once: the reader's subscribe grant, which has no
/// roster to enumerate agents from.
#[must_use]
pub fn term_subject_wildcard() -> String {
    term_subject("*")
}

/// Shared by [`term_subject`] and [`subagent_term`], whose subjects sit under
/// the same prefix.
const TERM_SUBJECT_PREFIX: &str = "$SWARM.term";

/// The subject `agent` publishes its turn-state header on. Same shape and
/// same three users as [`term_subject`].
#[must_use]
pub fn agent_state_subject(agent: &str) -> String {
    format!("{AGENT_STATE_SUBJECT_PREFIX}.{agent}")
}

/// Every [`agent_state_subject`] at once, for the same reason as
/// [`term_subject_wildcard`].
#[must_use]
pub fn agent_state_subject_wildcard() -> String {
    agent_state_subject("*")
}

/// The agent an [`agent_state_subject`] names, or `None` for any other
/// subject.
#[must_use]
pub fn agent_from_state_subject(subject: &str) -> Option<&str> {
    let agent = subject
        .strip_prefix(AGENT_STATE_SUBJECT_PREFIX)?
        .strip_prefix('.')?;
    (!agent.is_empty() && !agent.contains('.')).then_some(agent)
}

/// Shared by [`agent_state_subject`] and its wildcard.
const AGENT_STATE_SUBJECT_PREFIX: &str = "$SWARM.agent-state";

/// The reply-inbox prefix a client presenting `agent`'s own credential
/// connects with.
///
/// The queue grants that credential subscribe on this prefix's `.>` and its
/// [`agent_input_subject`], nothing else, so a request made under the
/// library's default `_INBOX.<nuid>` is refused its reply and times out. The
/// `agent` token keeps it disjoint from the default, whose second token is a
/// 22-character nuid.
#[must_use]
pub fn agent_inbox_prefix(agent: &str) -> String {
    format!("_INBOX.agent.{agent}")
}

/// The reply-inbox prefix [`connect`] uses for the client `client_id`: a hive
/// or the swarm controller.
///
/// The queue grants that client subscribe on this prefix's `.>`, so a request
/// made under the library's default `_INBOX.<nuid>` is refused its reply and
/// times out, as is a KV watch or key listing, whose consumer delivers there
/// too. The `client` token keeps it disjoint from [`agent_inbox_prefix`] and
/// from the default.
#[must_use]
pub fn client_inbox_prefix(client_id: &str) -> String {
    format!("_INBOX.client.{client_id}")
}

/// What an [`agent_control_subject`] message carries: one command for the
/// agent's harness.
///
/// The harness acts on a command only because the controller is the one
/// client granted publish on the subject, so there is no sender field to
/// trust or forge here, unlike [`AgentInput::from`].
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum AgentControl {
    /// Interrupt the turn in flight, as the agent's own cancel button does.
    /// Answered with an [`AgentControlReply`].
    Cancel,
    /// Apply a diff to the agent's `AGENTS.md`. Answered with an
    /// [`agents_md::ApplyReply`].
    ApplyAgentsMd(agents_md::Apply),
}

/// What an agent answers an [`AgentControl::Cancel`] request with.
///
/// The controller sends a command as a request, so a reply is the only
/// confirmation the operator gets that the harness took it. No reply within
/// [`AGENT_CONTROL_REPLY_DEADLINE`] means "not confirmed", which is not the
/// same as "refused".
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct AgentControlReply {
    /// Whether the command changed anything: `false` for a cancel that
    /// found no turn in flight.
    pub acted: bool,
}

/// How long the queue lets an agent answer an [`AgentControl`] request, and
/// the controller waits for it. The harness answers as soon as it has acted,
/// so this only bounds a harness that is down or wedged.
pub const AGENT_CONTROL_REPLY_DEADLINE: std::time::Duration = std::time::Duration::from_secs(10);

/// What an [`agent_input_subject`] message carries.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct AgentInput {
    /// The operator's authelia username, as the swarm gateway reported it.
    pub from: String,
    /// The text, verbatim.
    pub body: String,
}

/// The hive-notices stream, shared by the hive that publishes and
/// whatever eventually consumes it. Behind the `notices` feature, same
/// reason `status` is behind `kv` — see the module doc.
#[cfg(feature = "notices")]
pub mod notices;

/// The stream that keeps an agent's own terminal rows, and the consumer a
/// reader replays them through. Only the `JetStream` calls need the
/// `agent-term` feature; the names are unconditional.
pub mod agent_term;

/// The subject an agent's subagents' terminal rows go to, and the per-agent
/// stream that lists them. Only opening the stream needs the `subagent-term`
/// feature; the names are unconditional, like [`agent_icon`]'s.
pub mod subagent_term;

/// The review classifier's try-out subject and the request and reply it
/// carries. The `schema` feature adds `ToSchema` for an HTTP API serving the
/// same shapes.
pub mod review_classifier;

mod startup_read;
pub use startup_read::read_with_startup_retry;

/// Only the fields this needs; authelia returns several.
#[derive(serde::Deserialize)]
struct TokenResponse {
    access_token: String,
    /// Seconds the token stays valid — authelia says 3599 for
    /// `client_credentials`. Read rather than assumed, because it is what
    /// decides when the cache below stops handing the same token back, and a
    /// hardcoded hour would silently become wrong the day the identity
    /// provider is retuned.
    expires_in: u64,
}

/// A minted token and the moment it stops being usable.
///
/// Exists because the auth callback runs **per connection attempt**, not per
/// hour: without a cache, every reconnect mints, and a queue that cannot
/// connect turns its retry loop into a token-request loop against the
/// identity provider. That is not hypothetical — the first version of this
/// crate did exactly that, and authelia answered `429 Too Many Requests`
/// continuously until the client was changed.
struct CachedToken {
    token: String,
    expires_at: std::time::Instant,
}

/// Re-mint this long before expiry rather than at it. A token that expires
/// mid-CONNECT is refused by the server, which costs a whole reconnect cycle
/// to discover — far more than minting slightly early costs.
const TOKEN_REFRESH_SKEW: std::time::Duration = std::time::Duration::from_mins(2);

/// Cap on how long `async-nats` waits between reconnect attempts.
///
/// ⚠️ Deliberately far above the library's default, and the default is the
/// bug: `reconnect_delay_callback_default` backs off exponentially but clamps
/// at **4 seconds** (`connector.rs`), forever. Every attempt runs the auth
/// callback, so an unreachable queue means one token request every 4s
/// indefinitely — measured in production as a continuous
/// `429 Too Many Requests / temporarily_unavailable` from authelia, which then
/// keeps *itself* alive: the rate limit outlives whatever first broke the
/// connection, and its log volume buries the original cause.
///
/// The cost of raising it is slower recovery from a brief outage. That is the
/// right trade for an infrastructure service: a minute of staleness is a
/// dashboard reading `stale`, whereas a hot loop against the identity provider
/// degrades every other client of it.
const MAX_RECONNECT_DELAY: std::time::Duration = std::time::Duration::from_mins(1);

/// Exponential from 500ms, capped at [`MAX_RECONNECT_DELAY`].
fn reconnect_delay(attempts: usize) -> std::time::Duration {
    let exp = u32::try_from(attempts.saturating_sub(1)).unwrap_or(u32::MAX);
    std::cmp::min(
        std::time::Duration::from_millis(500u64.saturating_mul(2u64.saturating_pow(exp.min(8)))),
        MAX_RECONNECT_DELAY,
    )
}

/// Where the controller finds the queue and what it authenticates with.
///
/// Every field comes from an environment variable the NixOS module sets, the
/// same way `load_hives` takes the roster — a config change is a redeploy, and
/// this process reads no file it was not pointed at.
#[derive(Debug, Clone)]
pub struct QueueConfig {
    /// `tls://<name>:<port>` for the swarm queue.
    pub url: String,
    /// Authelia's token endpoint, e.g. `https://auth.<swarm>/api/oidc/token`.
    pub token_endpoint: String,
    /// The controller's own `OAuth2` client id.
    pub client_id: String,
    /// Where the client secret's plaintext comes from. Never the environment,
    /// which would publish it to anything that can read `/proc/<pid>/environ`.
    pub client_secret: ClientSecret,
    /// Extra trust anchor for the token endpoint and the queue itself, when
    /// they are not signed by a publicly-trusted CA.
    ///
    /// Optional, and deliberately NOT part of the all-or-none group below: a
    /// swarm fronted by a public certificate needs no extra anchor, and
    /// making this required would break that deployment to fix ours. Absent
    /// means "the platform's roots are enough", which is the correct default
    /// for a client that might talk to anything.
    ///
    /// ⚠️ Without it, a swarm using its own CA fails at TLS with
    /// `invalid peer certificate: UnknownIssuer` — the anchor exists on the
    /// host and this client simply never looked at it.
    pub ca_file: Option<PathBuf>,
}

/// Where a [`QueueConfig`]'s client secret comes from. `Debug` never prints
/// a held secret.
#[derive(Clone, Debug)]
pub enum ClientSecret {
    /// A file holding the plaintext, read at every token mint so a rotated
    /// file takes effect without a restart.
    File(PathBuf),
    /// The plaintext, held only in this process's memory and read again
    /// when the token endpoint refuses it.
    Stored(StoredSecret),
}

impl ClientSecret {
    /// The secret to read again after `error`: a held one the token endpoint
    /// refused as a credential ([`stored_secret::is_credential_refusal`]).
    fn refused(&self, error: &Error) -> Option<&StoredSecret> {
        match (self, error) {
            (Self::Stored(stored), Error::TokenRefused { status, .. })
                if stored_secret::is_credential_refusal(*status) =>
            {
                Some(stored)
            }
            _ => None,
        }
    }

    async fn read(&self) -> Result<String, Error> {
        match self {
            Self::File(path) => {
                tokio::fs::read_to_string(path)
                    .await
                    .map_err(|source| Error::ClientSecret {
                        path: path.display().to_string(),
                        source,
                    })
            }
            Self::Stored(stored) => Ok(stored.current()),
        }
    }

    fn read_blocking(&self) -> Result<String, Error> {
        match self {
            Self::File(path) => {
                std::fs::read_to_string(path).map_err(|source| Error::ClientSecret {
                    path: path.display().to_string(),
                    source,
                })
            }
            Self::Stored(stored) => Ok(stored.current()),
        }
    }
}

/// Mint a fresh access token for the controller's own client.
///
/// `client_credentials`, because there is no user here: the controller
/// authenticates as itself. Authelia refuses the `openid` scope for this grant
/// (a machine client receives an access token and never an id-token), so a
/// caller passing `None` for `scope` requests none at all — see
/// [`token_request`] for the one scope a caller does have to ask for.
///
/// A [`ClientSecret::Stored`] the endpoint refuses is read again, and when
/// that lands a different secret the request is sent once more with it; see
/// [`stored_secret`] for how often a re-read may run.
async fn mint_token(
    http: &reqwest::Client,
    cfg: &QueueConfig,
    audience: Option<&str>,
    scope: Option<&str>,
) -> Result<CachedToken, Error> {
    // Read per call rather than caching: the file is small, and a cached
    // secret would survive a rotation that the operator believes took effect.
    let secret = cfg.client_secret.read().await?;
    let refused = match mint_with(http, cfg, &secret, audience, scope).await {
        Err(e) => e,
        minted => return minted,
    };
    let Some(stored) = cfg.client_secret.refused(&refused) else {
        return Err(refused);
    };
    tracing::warn!(
        client_id = %cfg.client_id,
        error = %chain(&refused),
        "the token endpoint refused this client's secret; re-reading it"
    );
    match stored.after_refusal(&secret).await {
        Some(fresh) => mint_with(http, cfg, &fresh, audience, scope).await,
        None => Err(refused),
    }
}

/// One token request presenting `secret`.
async fn mint_with(
    http: &reqwest::Client,
    cfg: &QueueConfig,
    secret: &str,
    audience: Option<&str>,
    scope: Option<&str>,
) -> Result<CachedToken, Error> {
    let response = token_request(http, cfg, secret.trim(), audience, scope)
        .send()
        .await
        .map_err(Error::TokenRequest)?;

    let status = response.status();
    let body = response.text().await.unwrap_or_default();
    parse_token_response(status, &body)
}

/// Turn a raw token-endpoint response into a [`CachedToken`] — the half of
/// [`mint_token`] that is pure and therefore shared with the blocking sibling
/// below: neither the success/failure check nor the JSON shape cares which
/// HTTP client fetched the bytes.
///
/// Takes the body already read out, not the response itself: `reqwest`'s
/// blocking and async response types have no common trait for that, and
/// forcing one here would mean this function is not actually callable from
/// both callers, which defeats the reason it exists.
fn parse_token_response(status: reqwest::StatusCode, body: &str) -> Result<CachedToken, Error> {
    // Why the body and not just the status: see `Error::TokenRefused`.
    if !status.is_success() {
        return Err(Error::TokenRefused {
            status,
            body: body.to_owned(),
        });
    }

    let parsed: TokenResponse = serde_json::from_str(body).map_err(Error::TokenResponse)?;
    Ok(CachedToken {
        token: parsed.access_token,
        expires_at: std::time::Instant::now() + std::time::Duration::from_secs(parsed.expires_in),
    })
}

/// Build the token request: `client_credentials`, authenticated with HTTP
/// Basic.
///
/// 🩸 **The credentials go in the `Authorization` header, not the form
/// body.** A client registration names one flavor, and authelia's default
/// (and ours) is Basic — the other one once produced an hour of misleading
/// `429`s before the real cause (every request refused at the auth step)
/// surfaced. `audience` and `scope` are both opt-in, RFC 8707 /
/// registered-scope territory; `None` for either reproduces the request
/// this crate sent before the parameter existed. See the crate README's
/// "Token request shape" section for the incident, the RFC citations, and
/// why each parameter exists.
///
/// Split out of [`mint_token`] so the request's *shape* is testable without a
/// running identity provider — see the tests at the bottom of this file.
fn token_request(
    http: &reqwest::Client,
    cfg: &QueueConfig,
    secret: &str,
    audience: Option<&str>,
    scope: Option<&str>,
) -> reqwest::RequestBuilder {
    let mut form = vec![("grant_type", "client_credentials")];
    if let Some(audience) = audience {
        form.push(("audience", audience));
    }
    if let Some(scope) = scope {
        form.push(("scope", scope));
    }
    http.post(&cfg.token_endpoint)
        .basic_auth(&cfg.client_id, Some(secret))
        .form(&form)
}

/// Read + parse `path` as a PEM root certificate. Shared by the async and
/// blocking HTTP client builders below — `reqwest::Certificate::from_pem`
/// takes the same bytes regardless of which client type ends up trusting
/// it, so the CA loading has no reason to exist twice.
fn load_ca_cert(path: &std::path::Path) -> Result<reqwest::Certificate, Error> {
    let pem = std::fs::read(path).map_err(|source| Error::CaFile {
        path: path.display().to_string(),
        source,
    })?;
    reqwest::Certificate::from_pem(&pem).map_err(|source| Error::CaParse {
        path: path.display().to_string(),
        source,
    })
}

/// Build the HTTP client used to reach `cfg.token_endpoint`, trusting
/// `cfg.ca_file` when set. Shared by [`connect`]'s auth callback and by
/// [`mint_token_for`] — anything presenting this identity's credentials to
/// the token endpoint needs the swarm's own CA trusted the same way, so the
/// trust-anchor logic lives here once rather than once per caller.
fn build_http_client(cfg: &QueueConfig) -> Result<reqwest::Client, Error> {
    let mut builder = reqwest::Client::builder().timeout(std::time::Duration::from_secs(10));
    if let Some(path) = &cfg.ca_file {
        builder = builder.add_root_certificate(load_ca_cert(path)?);
    }
    builder.build().map_err(Error::HttpClient)
}

/// Mint a fresh token for `cfg`'s identity and hand back just the string —
/// no caching, a fresh HTTP client per call.
///
/// Public because the queue connection is not the only thing this identity
/// authenticates: "one identity per principal" means a caller that already
/// holds a [`QueueConfig`] for its queue connection authenticates anywhere
/// else it needs to prove who it is from the exact same client, rather than
/// a second identity being provisioned per destination. No caching here
/// unlike [`connect`]'s callback: that one exists because `async-nats` reruns
/// its callback per reconnect *attempt*, a hot path this isn't — a caller
/// outside that loop (e.g. `swarm-controller::auth`'s bridge client) mints
/// per call, same as this crate did before the reconnect-storm fix added the
/// cache.
///
/// `audience` and `scope` are passed straight to the private `token_request`
/// helper — see its doc for why each is optional and when a caller needs it.
/// `None` for both reproduces this
/// function's behaviour before the parameters existed, so every caller from
/// before they were added (the queue connect path, `auth.rs`'s bridge client)
/// is unaffected.
pub async fn mint_token_for(
    cfg: &QueueConfig,
    audience: Option<&str>,
    scope: Option<&str>,
) -> Result<String, Error> {
    let http = build_http_client(cfg)?;
    Ok(mint_token(&http, cfg, audience, scope).await?.token)
}

/// Blocking sibling of [`mint_token_for`], for a caller with no tokio
/// reactor to `.await` on.
///
/// That caller is real, not hypothetical: an OTLP exporter's `HttpClient`
/// implementation runs on the `PeriodicReader`'s own thread, which this
/// workspace deliberately keeps reactor-free (`reqwest-blocking-client` was
/// chosen over the async client for exactly that reason — see
/// `swarm-controller::vcs_metrics`'s module doc). Calling the async
/// [`mint_token_for`] from there would need a runtime that thread does not
/// have; this function uses `reqwest::blocking::Client` end to end instead,
/// so it never needs one.
///
/// Same request shape and same no-caching behaviour as [`mint_token_for`] —
/// see that function's doc for why both of those are the right call here.
pub fn mint_token_for_blocking(
    cfg: &QueueConfig,
    audience: Option<&str>,
    scope: Option<&str>,
) -> Result<String, Error> {
    mint_token_blocking(&build_blocking_http_client(cfg)?, cfg, audience, scope)
}

/// [`mint_token`] on the blocking client.
fn mint_token_blocking(
    http: &reqwest::blocking::Client,
    cfg: &QueueConfig,
    audience: Option<&str>,
    scope: Option<&str>,
) -> Result<String, Error> {
    // Read per call rather than caching — see `mint_token`'s identical
    // comment on the async path; the reasoning does not change with the
    // client type.
    let secret = cfg.client_secret.read_blocking()?;
    let refused = match mint_with_blocking(http, cfg, &secret, audience, scope) {
        Err(e) => e,
        minted => return minted,
    };
    let Some(stored) = cfg.client_secret.refused(&refused) else {
        return Err(refused);
    };
    tracing::warn!(
        client_id = %cfg.client_id,
        error = %chain(&refused),
        "the token endpoint refused this client's secret; re-reading it"
    );
    match stored.after_refusal_blocking(&secret) {
        Some(fresh) => mint_with_blocking(http, cfg, &fresh, audience, scope),
        None => Err(refused),
    }
}

/// [`mint_with`] on the blocking client.
fn mint_with_blocking(
    http: &reqwest::blocking::Client,
    cfg: &QueueConfig,
    secret: &str,
    audience: Option<&str>,
    scope: Option<&str>,
) -> Result<String, Error> {
    let mut form = vec![("grant_type", "client_credentials")];
    if let Some(audience) = audience {
        form.push(("audience", audience));
    }
    if let Some(scope) = scope {
        form.push(("scope", scope));
    }
    let response = http
        .post(&cfg.token_endpoint)
        .basic_auth(&cfg.client_id, Some(secret.trim()))
        .form(&form)
        .send()
        .map_err(Error::TokenRequest)?;

    let status = response.status();
    let body = response.text().unwrap_or_default();
    Ok(parse_token_response(status, &body)?.token)
}

/// Blocking counterpart to [`build_http_client`] — same CA-trust logic
/// (via the shared [`load_ca_cert`]), a `reqwest::blocking::Client` instead
/// of the async one. See [`mint_token_for_blocking`] for why a caller needs
/// this rather than the async builder.
fn build_blocking_http_client(cfg: &QueueConfig) -> Result<reqwest::blocking::Client, Error> {
    let mut builder =
        reqwest::blocking::Client::builder().timeout(std::time::Duration::from_secs(10));
    if let Some(path) = &cfg.ca_file {
        builder = builder.add_root_certificate(load_ca_cert(path)?);
    }
    builder.build().map_err(Error::HttpClient)
}

/// Fail fast unless the client is actually connected.
///
/// **Call this before every `JetStream` request.** `retry_on_initial_connect`
/// means the client exists before it is usable, and a `JetStream` request made
/// in that window does not fail — it *waits*, for longer than any caller
/// should (measured: still going at 15s against a queue that simply refuses
/// the credential). On a request path that hangs a poll; on a periodic task it
/// hangs the task, including the shutdown branch it never reaches.
///
/// Testing for `!= Connected` rather than `== Disconnected` is the whole
/// point: a client that has never connected once sits in `Pending`, so the
/// `Disconnected` test passes it straight through to the hang it was written
/// to prevent. That is exactly the case that matters — a process whose
/// credential is wrong from boot never reaches `Disconnected`, because it was
/// never connected to begin with.
///
/// Naming the state is also the better error: "not connected" is actionable,
/// a timeout is not.
pub fn ensure_connected(client: &async_nats::Client) -> Result<(), Error> {
    let state = client.connection_state();
    if state != async_nats::connection::State::Connected {
        return Err(Error::NotConnected(state));
    }
    Ok(())
}

/// The largest payload this connection's server will accept, as the server
/// itself announced it at CONNECT.
///
/// Read it from here rather than from a constant: the limit is set in the
/// queue's own NixOS module, so a client that hardcoded it would be asserting
/// a number it does not own and would keep asserting it after the operator
/// raised it. A publisher that exceeds the limit does not get a shortened
/// message — the server answers `Maximum Payload Violation` and closes the
/// connection, costing the row and a reconnect — so a publisher of
/// variable-size rows must size them against this before handing them over.
///
/// ⚠️ The value is per connection. Before the first successful CONNECT the
/// client reports the library's own default, which is smaller than any
/// deployment sets; call this after [`ensure_connected`], and the answer is
/// the one this server actually announced.
#[must_use]
pub fn max_payload(client: &async_nats::Client) -> usize {
    client.server_info().max_payload
}

/// Connect to the swarm queue, presenting a token on each connection attempt
/// and minting a fresh one only when the cached one is near expiry. Reply
/// inboxes are under [`client_inbox_prefix`] of `cfg.client_id`.
///
/// The rejected alternative — mint once, pass a static `auth_token`, own the
/// reconnect loop — fails in the way this subsystem exists to prevent: the
/// token expires, the controller keeps serving, its status data quietly stops
/// updating, and nothing says so until someone reads a dashboard.
pub async fn connect(cfg: QueueConfig) -> Result<async_nats::Client, Error> {
    swarm_secret_client::install_crypto_provider();
    // Trusts `cfg.ca_file` when set (the swarm's own CA, when the token
    // endpoint is signed by it) — see `build_http_client`. A timeout too,
    // because this client runs INSIDE the auth callback: a token endpoint
    // that accepts the connection and then never answers would hang the
    // callback, and with it the connection attempt that invoked it, with no
    // retry and nothing in the log to say why. Failing fast lets
    // `async-nats` do what it already does well — back off and try again.
    let http = build_http_client(&cfg)?;
    let url = cfg.url.clone();
    let ca_file = cfg.ca_file.clone();
    let inbox_prefix = client_inbox_prefix(&cfg.client_id);

    // Shared across every invocation of the callback below, which is the
    // whole point: the callback fires per connection ATTEMPT, so without
    // somewhere to remember the last token, "mint on demand" and "mint on
    // every retry" are the same code.
    let cache: std::sync::Arc<tokio::sync::Mutex<Option<CachedToken>>> =
        std::sync::Arc::new(tokio::sync::Mutex::new(None));

    let mut options = async_nats::ConnectOptions::with_auth_callback(move |_nonce| {
        let http = http.clone();
        let cfg = cfg.clone();
        let cache = cache.clone();
        async move {
            let mut slot = cache.lock().await;

            // Reuse while there is comfortably more than the skew left. The
            // freshness property this callback exists for is preserved — a
            // reconnect an hour later finds an expired entry and re-mints —
            // but a reconnect ten seconds later does NOT ask the IdP again.
            let reuse = slot
                .as_ref()
                .filter(|t| {
                    t.expires_at
                        .saturating_duration_since(std::time::Instant::now())
                        > TOKEN_REFRESH_SKEW
                })
                .map(|t| t.token.clone());

            let token = if let Some(token) = reuse {
                token
            } else {
                let minted = mint_token(&http, &cfg, None, None)
                    .await
                    // The callback's error type carries a string, so the
                    // source chain would be lost; flatten it rather than
                    // dropping it.
                    .map_err(|e| async_nats::AuthError::new(chain(&e)))?;
                let token = minted.token.clone();
                *slot = Some(minted);
                token
            };

            let mut auth = async_nats::Auth::new();
            auth.token = Some(token);
            Ok(auth)
        }
    })
    // See `MAX_RECONNECT_DELAY`. The library's default clamps at 4s forever,
    // which turns an unreachable queue into a permanent 4s poll — and, before
    // the cache above, a permanent 4s token-request loop against authelia.
    // Exponential from 500ms so a momentary blip still reconnects promptly.
    .reconnect_delay_callback(reconnect_delay)
    // The controller and the queue are separate units on (possibly)
    // separate hosts, and nothing orders them. Without this, a queue that
    // comes up one second later leaves the controller permanently
    // queue-less until someone restarts it — a boot-order race that
    // presents as "status has been unavailable since Tuesday".
    //
    // It also composes with the callback above rather than fighting it:
    // each background attempt runs the callback, which hands over a token
    // that is still valid — minting a new one only when the cached one is
    // near expiry. (Before the cache, "runs the callback" meant "mints",
    // and this retry loop was a token-request loop. See
    // `MAX_RECONNECT_DELAY`.)
    .retry_on_initial_connect()
    .custom_inbox_prefix(inbox_prefix);

    // The queue's leaf chains to the swarm's own PKI root, which a host
    // process's platform store does not hold: the same anchor the token
    // endpoint needs. async-nats then trusts this file INSTEAD of the platform
    // roots, which is right here, because the queue has no public certificate.
    if let Some(path) = ca_file {
        options = options.add_root_certificates(path);
    }

    let client = options
        .connect(&url)
        .await
        .map_err(|source| Error::Connect {
            url: url.clone(),
            source,
        })?;

    Ok(client)
}

/// Connect presenting the token `token` resolves to, as `agent` presents its
/// own queue credential, with reply inboxes under [`agent_inbox_prefix`].
///
/// `token` runs once per connection attempt, reconnects included, so a
/// credential replaced at its source is presented on the next attempt. An
/// `Err` from it fails that attempt, and the next one waits out the same
/// backoff as a refused connect.
///
/// Unlike [`connect`], the first attempt must succeed: a refused or failed
/// connect is returned rather than retried in the background, so the caller
/// sees it and decides how to retry. `ca_file` is the queue's trust anchor,
/// as [`QueueConfig::ca_file`].
pub async fn connect_with_token<F, Fut>(
    url: &str,
    ca_file: Option<&std::path::Path>,
    agent: &str,
    token: F,
) -> Result<async_nats::Client, Error>
where
    F: Fn() -> Fut + Send + Sync + 'static,
    Fut: Future<Output = Result<String, String>> + Send + Sync + 'static,
{
    let mut options = async_nats::ConnectOptions::with_auth_callback(move |_nonce| {
        let token = token();
        async move {
            let mut auth = async_nats::Auth::new();
            auth.token = Some(token.await.map_err(async_nats::AuthError::new)?);
            Ok(auth)
        }
    })
    .reconnect_delay_callback(reconnect_delay)
    .custom_inbox_prefix(agent_inbox_prefix(agent));
    if let Some(path) = ca_file {
        options = options.add_root_certificates(path.to_path_buf());
    }
    options.connect(url).await.map_err(|source| Error::Connect {
        url: url.to_owned(),
        source,
    })
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;
    use std::sync::atomic::{AtomicUsize, Ordering};

    use super::*;

    /// A store holding `secret`, counting how often it is read.
    struct Store {
        secret: &'static str,
        reads: Arc<AtomicUsize>,
    }

    impl Store {
        fn holding(secret: &'static str) -> Self {
            Self {
                secret,
                reads: Arc::default(),
            }
        }

        /// `client_id`'s secret, `initial` in use, re-read from this store.
        fn secret(&self, client_id: &str, initial: &str) -> StoredSecret {
            let (secret, reads) = (self.secret, Arc::clone(&self.reads));
            StoredSecret::new(client_id, initial.to_owned(), move || {
                reads.fetch_add(1, Ordering::SeqCst);
                async move { Ok(secret.to_owned()) }
            })
        }

        fn reads(&self) -> usize {
            self.reads.load(Ordering::SeqCst)
        }
    }

    /// The names the publishers, the relays and the queue's grant all spell.
    #[test]
    fn term_and_agent_state_subjects_are_the_agreed_names() {
        assert_eq!(term_subject("mara"), "$SWARM.term.mara");
        assert_eq!(term_subject_wildcard(), "$SWARM.term.*");
        assert_eq!(agent_state_subject("mara"), "$SWARM.agent-state.mara");
        assert_eq!(agent_state_subject_wildcard(), "$SWARM.agent-state.*");
        assert_eq!(subagent_term::subject_wildcard(), "$SWARM.term.*.sub.*");
    }

    #[test]
    fn the_agent_is_read_back_out_of_its_state_subject() {
        assert_eq!(
            agent_from_state_subject(&agent_state_subject("iris")),
            Some("iris")
        );
        for other in [
            "$SWARM.agent-state.",
            "$SWARM.agent-state.iris.sub.scout",
            "$SWARM.agent-stateiris",
            "$SWARM.term.iris",
        ] {
            assert_eq!(agent_from_state_subject(other), None, "{other}");
        }
    }

    #[tokio::test]
    async fn a_debug_print_of_a_config_does_not_carry_a_held_secret() {
        let cfg = QueueConfig {
            client_secret: ClientSecret::Stored(
                Store::holding("new").secret("hive-alpha", "s3cret"),
            ),
            ..token_cfg()
        };
        let printed = format!("{cfg:?}");
        assert!(!printed.contains("s3cret"), "{printed}");
        assert!(printed.contains("StoredSecret"), "{printed}");
    }

    /// A token endpoint on a loopback port that mints for
    /// `hive-alpha:{registered}` and answers any other secret with a 401,
    /// counting its requests.
    async fn stub_idp(registered: &'static str, calls: Arc<AtomicUsize>) -> String {
        use axum::http::{HeaderMap, StatusCode, header::AUTHORIZATION};

        let expected = format!(
            "Basic {}",
            data_encoding::BASE64.encode(format!("hive-alpha:{registered}").as_bytes())
        );
        let app = axum::Router::new().fallback(move |headers: HeaderMap| async move {
            calls.fetch_add(1, Ordering::SeqCst);
            if headers.get(AUTHORIZATION).and_then(|v| v.to_str().ok()) == Some(&*expected) {
                (
                    StatusCode::OK,
                    r#"{"access_token":"a-token","expires_in":3599}"#,
                )
            } else {
                (StatusCode::UNAUTHORIZED, r#"{"error":"invalid_client"}"#)
            }
        });
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("bind");
        let url = format!("http://{}/token", listener.local_addr().expect("addr"));
        tokio::spawn(async move { axum::serve(listener, app).await });
        url
    }

    /// A plain-HTTP loopback client: no proxy, and no platform trust store,
    /// which the nix build sandbox does not have (see [`offline_client`]).
    fn loopback_client() -> reqwest::Client {
        reqwest::Client::builder()
            .no_proxy()
            .danger_accept_invalid_certs(true)
            .build()
            .expect("a client that verifies nothing needs no system trust store")
    }

    /// The rotation the re-read exists for: the token endpoint refuses the
    /// secret held since start, the store holds the new one, and the same
    /// mint succeeds with it.
    #[tokio::test]
    async fn a_refused_secret_is_re_read_and_the_mint_retried_with_the_new_one() {
        let calls = Arc::new(AtomicUsize::new(0));
        let store = Store::holding("secret-b");
        let cfg = QueueConfig {
            token_endpoint: stub_idp("secret-b", Arc::clone(&calls)).await,
            client_secret: ClientSecret::Stored(store.secret("hive-alpha", "secret-a")),
            ..token_cfg()
        };
        let minted = mint_token(&loopback_client(), &cfg, None, None)
            .await
            .expect("minted with the re-read secret");
        assert_eq!(minted.token, "a-token");
        assert_eq!(calls.load(Ordering::SeqCst), 2);
        assert_eq!(store.reads(), 1);
    }

    /// A secret the store still holds as refused: the mint fails, and the
    /// mints after it inside the backoff ask the token endpoint once each
    /// and read nothing.
    #[tokio::test]
    async fn a_permanently_refused_secret_is_not_re_read_on_every_mint() {
        let calls = Arc::new(AtomicUsize::new(0));
        let store = Store::holding("secret-a");
        let cfg = QueueConfig {
            token_endpoint: stub_idp("secret-b", Arc::clone(&calls)).await,
            client_secret: ClientSecret::Stored(store.secret("hive-alpha", "secret-a")),
            ..token_cfg()
        };
        let http = loopback_client();
        for _ in 0..5 {
            let err = mint_token(&http, &cfg, None, None)
                .await
                .err()
                .expect("refused");
            assert!(
                matches!(err, Error::TokenRefused { status, .. } if status == 401),
                "{err}"
            );
        }
        assert_eq!(calls.load(Ordering::SeqCst), 5);
        assert_eq!(store.reads(), 1);
    }

    /// A refusal that is not about the credential reads nothing.
    #[tokio::test]
    async fn a_rate_limited_or_failing_mint_does_not_re_read() {
        use axum::http::StatusCode;

        let http = loopback_client();
        for status in [
            StatusCode::TOO_MANY_REQUESTS,
            StatusCode::SERVICE_UNAVAILABLE,
        ] {
            let app = axum::Router::new().fallback(move || async move { (status, "") });
            let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
                .await
                .expect("bind");
            let url = format!("http://{}/token", listener.local_addr().expect("addr"));
            tokio::spawn(async move { axum::serve(listener, app).await });

            let store = Store::holding("secret-b");
            let cfg = QueueConfig {
                token_endpoint: url,
                client_secret: ClientSecret::Stored(store.secret("hive-alpha", "secret-a")),
                ..token_cfg()
            };
            let err = mint_token(&http, &cfg, None, None)
                .await
                .err()
                .expect("refused");
            assert!(
                matches!(err, Error::TokenRefused { status: s, .. } if s == status.as_u16()),
                "{err}"
            );
            assert_eq!(store.reads(), 0, "{status}");
        }
    }

    /// The blocking mint re-reads the same way, from a thread with no
    /// runtime of its own.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn the_blocking_mint_re_reads_a_refused_secret() {
        let calls = Arc::new(AtomicUsize::new(0));
        let store = Store::holding("secret-b");
        let cfg = QueueConfig {
            token_endpoint: stub_idp("secret-b", Arc::clone(&calls)).await,
            client_secret: ClientSecret::Stored(store.secret("hive-alpha", "secret-a")),
            ..token_cfg()
        };
        let token = tokio::task::spawn_blocking(move || {
            let http = reqwest::blocking::Client::builder()
                .no_proxy()
                .danger_accept_invalid_certs(true)
                .build()
                .expect("client");
            mint_token_blocking(&http, &cfg, None, None)
        })
        .await
        .expect("joined")
        .expect("minted with the re-read secret");
        assert_eq!(token, "a-token");
        assert_eq!(store.reads(), 1);
    }

    fn token_cfg() -> QueueConfig {
        QueueConfig {
            url: "nats://127.0.0.1:4222".to_owned(),
            token_endpoint: "https://auth.example.com/api/oidc/token".to_owned(),
            client_id: "hive-alpha".to_owned(),
            client_secret: ClientSecret::File(PathBuf::from("/nonexistent")),
            ca_file: None,
        }
    }

    /// 🩸 THE REGRESSION TEST FOR AN OUTAGE THAT RAN FOR WEEKS.
    ///
    /// The credentials used to go in the form body (`client_secret_post`).
    /// Authelia's client registration allows only `client_secret_basic`, so
    /// every token request was refused — and the refusals tripped a rate
    /// limiter whose 429 then arrived *before* the credentials were evaluated,
    /// so the error naming the cause appeared roughly once an hour inside a
    /// continuous storm of a different error.
    ///
    /// This asserts the *shape of the request* rather than a server's reply,
    /// which is the whole point: it fails on the old code with no identity
    /// provider, no deployment and no network.
    /// A client for inspecting a request, never for sending one.
    ///
    /// 🩸 `reqwest::Client::new()` **panics in the nix build sandbox**, which
    /// has no system CA store: `ClientBuilder::build()` reaches
    /// `rustls_platform_verifier::Verifier::new()` and fails with "No CA
    /// certificates were loaded from the system", and `new()` is
    /// `build().expect(..)`. The test passed locally — a devshell has
    /// `/etc/ssl/certs` — and failed in CI.
    ///
    /// Turning verification off takes the `!certs_verification` branch, which
    /// installs a no-op verifier and never consults the platform store, so
    /// this builds anywhere. It is sound *here specifically* because nothing
    /// is ever sent: the request is built and its bytes inspected.
    fn offline_client() -> reqwest::Client {
        reqwest::Client::builder()
            .danger_accept_invalid_certs(true)
            .build()
            .expect("a client that verifies nothing needs no system trust store")
    }

    #[test]
    fn the_token_request_authenticates_with_http_basic() {
        let req = token_request(&offline_client(), &token_cfg(), "s3cret", None, None)
            .build()
            .expect("the token request must build");

        let auth = req
            .headers()
            .get(reqwest::header::AUTHORIZATION)
            .expect("credentials must travel in the Authorization header")
            .to_str()
            .expect("the header is ascii");
        assert!(
            auth.starts_with("Basic "),
            "must be client_secret_basic, got: {auth}"
        );

        let body = std::str::from_utf8(
            req.body()
                .and_then(reqwest::Body::as_bytes)
                .expect("the request has an in-memory body"),
        )
        .expect("the body is utf-8");
        assert!(
            body.contains("grant_type=client_credentials"),
            "the grant type still belongs in the body, got: {body}"
        );
        // The half that was actually broken: a secret in the body is both the
        // wrong auth method for our registration and a value proxies log.
        assert!(
            !body.contains("client_secret"),
            "the secret must not be in the request body, got: {body}"
        );
        assert!(
            !body.contains("client_id"),
            "the client id belongs in the Basic credentials, got: {body}"
        );
    }

    /// The queue/bridge shape (`audience: None`, `scope: None`) must stay
    /// unchanged by the parameters' addition — no `audience` field appears
    /// in the body at all, not even empty.
    #[test]
    fn no_audience_means_no_audience_field() {
        let req = token_request(&offline_client(), &token_cfg(), "s3cret", None, None)
            .build()
            .expect("the token request must build");
        let body = std::str::from_utf8(
            req.body()
                .and_then(reqwest::Body::as_bytes)
                .expect("the request has an in-memory body"),
        )
        .expect("the body is utf-8");
        assert!(
            !body.contains("audience"),
            "omitting the audience must not even send an empty field, got: {body}"
        );
    }

    /// A caller that DOES ask for an audience gets it in the form body,
    /// verbatim — this is the half `swarm-otel`'s `oidc/<owner>`
    /// authenticators actually check.
    #[test]
    fn an_audience_is_sent_verbatim() {
        let req = token_request(
            &offline_client(),
            &token_cfg(),
            "s3cret",
            Some("https://otel.example/swarm"),
            None,
        )
        .build()
        .expect("the token request must build");
        let body = std::str::from_utf8(
            req.body()
                .and_then(reqwest::Body::as_bytes)
                .expect("the request has an in-memory body"),
        )
        .expect("the body is utf-8");
        assert!(
            body.contains("audience=https%3A%2F%2Fotel.example%2Fswarm"),
            "the requested audience must reach the form body, got: {body}"
        );
    }

    /// No scope asked for, no `scope` field — the queue connection and the
    /// OTLP push both mint this way, and a `scope=` authelia has no mapping
    /// for is refused rather than ignored.
    #[test]
    fn no_scope_means_no_scope_field() {
        let req = token_request(&offline_client(), &token_cfg(), "s3cret", None, None)
            .build()
            .expect("the token request must build");
        let body = std::str::from_utf8(
            req.body()
                .and_then(reqwest::Body::as_bytes)
                .expect("the request has an in-memory body"),
        )
        .expect("the body is utf-8");
        assert!(
            !body.contains("scope"),
            "omitting the scope must not even send an empty field, got: {body}"
        );
    }

    #[test]
    fn a_deploy_request_without_a_rev_decodes() {
        let request: super::DeployRequest =
            serde_json::from_str(r#"{"agent":"damocles"}"#).expect("a rev-less payload decodes");
        assert_eq!(
            request,
            super::DeployRequest {
                agent: "damocles".to_owned(),
                rev: None,
            }
        );
    }

    #[test]
    fn a_deploy_request_carries_its_rev() {
        let request: super::DeployRequest =
            serde_json::from_str(r#"{"agent":"damocles","rev":"abc123"}"#).expect("decodes");
        assert_eq!(request.rev.as_deref(), Some("abc123"));
    }

    /// The bug behind the log store's 401: a client REGISTERED for
    /// `authelia.bearer.authz` still receives a token carrying no scope
    /// unless the request asks, and authelia's authz endpoint refuses that.
    #[test]
    fn a_scope_is_sent_verbatim() {
        let req = token_request(
            &offline_client(),
            &token_cfg(),
            "s3cret",
            Some("https://logs.example/select/logsql/query"),
            Some("authelia.bearer.authz"),
        )
        .build()
        .expect("the token request must build");
        let body = std::str::from_utf8(
            req.body()
                .and_then(reqwest::Body::as_bytes)
                .expect("the request has an in-memory body"),
        )
        .expect("the body is utf-8");
        assert!(
            body.contains("scope=authelia.bearer.authz"),
            "the requested scope must reach the form body, got: {body}"
        );
    }
}