hyperhive/swarm-nats-auth/src/policy.rs

//! What an admitted client is allowed to do.
//!
//! [`crate::introspect`] answers *whether* to admit and *as whom*. This
//! answers *what that identity may touch* — the two are deliberately separate
//! decisions: admission is the `IdP`'s, authorisation is the swarm's.
//!
//! # Deny is the default, and that is a choice with a measurement behind it
//!
//! A client whose id matches no rule gets **no grant at all**, not an
//! unrestricted one. Every other shape here fails open: an unmatched client
//! that kept today's unscoped grant would make the whole policy advisory, and
//! the failure would be silent. Denying costs a loud refusal the first time a
//! new consumer appears, which is a config line to fix.
//!
//! Each subject set below is *minimal by removal*, and carries its own note
//! saying what breaks without it — the reasoning lives next to the list it
//! constrains rather than in one block here, because that is where a reader
//! about to edit the list will meet it.

/// The subjects an admitted client may publish and subscribe to.
#[derive(Debug, PartialEq, Eq)]
pub struct Permissions {
    /// Subjects allowed for publish. Never empty: an empty allow-list is a
    /// grant that can do nothing, which is a denial wearing a grant's shape.
    pub publish: Vec<String>,
    /// Subjects allowed for subscribe, or `None` for every subject in the
    /// account.
    ///
    /// `Some` only for agents, the least trusted clients: their own reply
    /// inbox and their own input subject.
    /// Hives and readers get `None`. Never `Some(vec![])`: NATS reads an empty
    /// allow-list as no restriction at all.
    pub subscribe: Option<Vec<String>>,
}

/// Which client ids get which permissions.
///
/// Constructed from configuration so that adding a subject a hive may publish
/// — a second event stream, say — is a deployment change rather than a change
/// to this responder.
#[derive(Debug, Clone)]
pub struct Policy {
    hive_prefix: String,
    agent_suffix: String,
    bucket: String,
    readers: Vec<String>,
    extra_hive_subjects: Vec<String>,
    agent_token_subjects: Vec<String>,
}

/// Placeholder replaced with the hive's own name in `extra_hive_subjects`.
const HIVE_PLACEHOLDER: &str = "{hive}";

/// Placeholder replaced with the agent's own name in `agent_token_subjects`.
const AGENT_PLACEHOLDER: &str = "{agent}";

impl Policy {
    /// `hive_prefix` is the client-id prefix that marks a hive and
    /// `agent_suffix` what a hive's agent containers carry **on top of** it —
    /// `hive-<name>` and `hive-<name>-agent`. `bucket` is the KV bucket hives
    /// report status in, `readers` the client ids allowed to read every hive's
    /// key, and `extra_hive_subjects` additional subjects a hive may publish to
    /// (with `{hive}` standing for the hive's own name).
    ///
    /// The agent id is deliberately the hive's id extended, not a second
    /// top-level prefix: `agent-<name>` reads as *the agent called `<name>`*,
    /// which is the one thing this identity does not carry.
    ///
    /// # Errors
    ///
    /// An extra subject with no `{hive}` in it is refused. Such a template
    /// expands to the **same** subject for every hive, which is not a per-hive
    /// namespace — it is the absence of one, arrived at through the option
    /// whose only purpose is to provide one. Returning a `Result` rather than
    /// checking at the call site is deliberate: it makes an unscoped policy
    /// unconstructible instead of merely unlikely.
    ///
    /// An empty `agent_suffix` is refused for the same reason: every hive id
    /// would also parse as its own agent id, and the arms are tried in order,
    /// so the overlap does not error at match time — it silently hands one
    /// principal the other's grant.
    pub fn new(
        hive_prefix: String,
        agent_suffix: String,
        bucket: String,
        readers: Vec<String>,
        extra_hive_subjects: Vec<String>,
    ) -> anyhow::Result<Self> {
        if let Some(bad) = extra_hive_subjects
            .iter()
            .find(|s| !s.contains(HIVE_PLACEHOLDER))
        {
            anyhow::bail!(
                "--hive-publish-subject {bad:?} contains no {HIVE_PLACEHOLDER}: every hive \
                 would be granted that exact subject, so it is a shared channel rather than \
                 a per-hive namespace"
            );
        }
        // Checked here rather than trusted from nix, because these are two
        // flags and nothing downstream compares them. An agent id is a hive id
        // plus this suffix, so an empty one makes `strip_suffix` succeed on
        // every hive id and the two principals become the same string. The
        // agent arm runs first and refuses, so every hive would be refused as
        // its own agents' client.
        if agent_suffix.is_empty() {
            anyhow::bail!(
                "--agent-client-suffix is empty: an agent id is a hive id plus this suffix, \
                 so every hive would also parse as its own agents' client and be refused"
            );
        }
        Ok(Self {
            hive_prefix,
            agent_suffix,
            bucket,
            readers,
            extra_hive_subjects,
            agent_token_subjects: Vec::new(),
        })
    }

    /// Set the subjects an agent that proved its own credential may publish
    /// to, with `{agent}` standing for its name.
    ///
    /// # Errors
    ///
    /// A template with no `{agent}` in it is refused: it would be one subject
    /// shared by every agent in the swarm rather than the agent's own.
    pub fn with_agent_token_subjects(mut self, subjects: Vec<String>) -> anyhow::Result<Self> {
        if let Some(bad) = subjects.iter().find(|s| !s.contains(AGENT_PLACEHOLDER)) {
            anyhow::bail!(
                "--agent-token-publish-subject {bad:?} contains no {AGENT_PLACEHOLDER}: every \
                 agent in the swarm would be granted that exact subject"
            );
        }
        self.agent_token_subjects = subjects;
        Ok(self)
    }

    /// The permissions for an agent whose own credential was verified, or
    /// `None` when none are configured.
    ///
    /// Keyed on the agent alone: its identity is not tied to a hive. `agent`
    /// must already be a single `[A-Za-z0-9_-]` segment, which the token parse
    /// guarantees, so it cannot widen a subject with `.`, `*` or `>`.
    pub fn agent_token_permissions(&self, agent: &str) -> Option<Permissions> {
        let publish: Vec<String> = self
            .agent_token_subjects
            .iter()
            .map(|s| s.replace(AGENT_PLACEHOLDER, agent))
            .collect();
        (!publish.is_empty()).then(|| Permissions {
            publish,
            subscribe: Some(vec![
                format!("{}.>", swarm_queue_client::agent_inbox_prefix(agent)),
                swarm_queue_client::agent_input_subject(agent),
            ]),
        })
    }

    /// The permissions for `client_id`, or `None` when no rule matches.
    ///
    /// `None` is a denial. It is not "grant nothing and let them connect":
    /// a connected client with no permissions still holds a slot and still
    /// looks admitted in the logs, which is a worse answer than a refusal.
    pub fn permissions(&self, client_id: &str) -> Option<Permissions> {
        // ⚠️ THE AGENT ARM RUNS FIRST, AND THE ORDER IS LOAD-BEARING. An agent
        // id is a hive id with a suffix, so `hive-foo-agent` satisfies the hive
        // arm too — it strips to the hive `foo-agent`. Hive-first would
        // therefore hand every agent its hive's grant, including writing that
        // hive's status key, and nothing would report it: the client
        // authenticates, connects, and is simply able to do more than it
        // should. Most specific wins, so the more specific test is the one
        // that has to be asked first.
        //
        // A hive's shared agent client is refused outright. Agents connect with
        // their own credential (`Self::agent_token_permissions`); this identity
        // names the hive and never the agent (see `Self::agent_hive`), so any
        // grant to it would let one agent publish as its siblings.
        if self.agent_hive(client_id).is_some() {
            return None;
        }
        if let Some(hive) = self.hive_name(client_id) {
            return Some(Permissions {
                publish: self.hive_subjects(hive),
                subscribe: None,
            });
        }
        if self.readers.iter().any(|r| r == client_id) {
            return Some(Permissions {
                publish: self.reader_subjects(),
                subscribe: None,
            });
        }
        None
    }

    /// The hive a client id names, when it names one.
    ///
    /// The prefix is configuration, not a pattern this module guesses at: the
    /// authelia client is `hive-<name>` while the KV key is the bare `<name>`,
    /// and stripping by eye is how a client id that merely *looks*
    /// hive-shaped ends up with a hive's permissions. An id that is the prefix
    /// and nothing else names no hive and is refused — `$KV.<bucket>.` is not
    /// a narrower subject than `$KV.<bucket>.<name>`, it is a different one.
    ///
    /// # This matches the prefix; it does not verify the roster
    ///
    /// A client id is a hive here because it *starts with the prefix*, not
    /// because it appears in `services.hyperhive.swarm.hives`. This responder
    /// runs inside a container and has no view of the roster, so an
    /// operator-declared client called `hive-anything` would be granted
    /// `$KV.<bucket>.anything` — a key no real hive owns, which the controller
    /// renders as a hive it does not recognise.
    ///
    /// Passing the roster in would close that, and would also be a **second
    /// place deciding who may connect as what**, which `crate::introspect`'s
    /// docs argue against for the same reason admission lives in one place.
    /// The prefix is a contract with `swarm-authelia.nix`, and this is the end
    /// of it that can be checked from in here.
    fn hive_name<'a>(&self, client_id: &'a str) -> Option<&'a str> {
        client_id
            .strip_prefix(&self.hive_prefix)
            .filter(|name| !name.is_empty())
    }

    /// The hive whose agents a client id names, when it names one.
    ///
    /// [`Self::hive_name`]'s caveats all apply — this is that parse with the
    /// suffix stripped as well, so `hive-foo-agent` answers `foo`. Two more on
    /// top:
    ///
    /// - The agent client is minted per **hive**, so this answers *whose
    ///   agents*, never *which agent*. Two agents on one hive are
    ///   indistinguishable here — not a gap in the parsing but the identity
    ///   itself, which does not carry the agent.
    /// - A hive genuinely named `foo-agent` mints `hive-foo-agent` and is
    ///   therefore **indistinguishable from hive `foo`'s agents** right here.
    ///   Unguardable from inside this responder, which has no roster (see
    ///   [`Self::hive_name`]) — `swarm-authelia.nix` asserts it at eval, where
    ///   the roster is known, so the deployment fails to build rather than
    ///   minting two principals with one id.
    fn agent_hive<'a>(&self, client_id: &'a str) -> Option<&'a str> {
        client_id
            .strip_prefix(&self.hive_prefix)?
            .strip_suffix(&self.agent_suffix)
            .filter(|name| !name.is_empty())
    }

    fn stream(&self) -> String {
        format!("KV_{}", self.bucket)
    }

    /// The stream backing one hive's wanted-state bucket.
    ///
    /// Derived from the crate's own name function rather than from a
    /// configurable bucket name: `wanted::bucket` is what the writer and the
    /// reader both call, so a flag here would let two deployments disagree
    /// about a name they must share.
    fn wanted_stream(hive: &str) -> String {
        format!("KV_{}", swarm_queue_client::wanted::bucket(hive))
    }

    /// The stream backing the per-agent status bucket.
    ///
    /// Same shape and same reason as [`Self::wanted_stream`]: derived from
    /// the crate constant rather than a third configurable bucket name.
    fn agent_status_stream() -> String {
        format!("KV_{}", swarm_queue_client::agent_status::BUCKET)
    }

    /// What *any* `JetStream` client must be able to ask before it can do
    /// anything at all, bucket-specific or not.
    ///
    /// Both were measured from the server's own refusals, not reasoned about:
    /// a grant carrying every bucket-specific subject and neither of these
    /// cannot even create the bucket — the client times out on `$JS.API.INFO`
    /// long before it reaches a subject that was granted.
    ///
    /// - `$JS.API.INFO` — account-level `JetStream` info, requested on connect.
    /// - `$JS.API.STREAM.NAMES` — how a client finds the stream backing a
    ///   bucket. It lets a client enumerate stream names in the account, which
    ///   in an account holding one bucket discloses a name both ends already
    ///   share.
    ///
    /// 🩸 Earlier measurements missed both, because they either granted
    /// `$JS.API.>` wholesale or ran against a bucket the *setup* had already
    /// created while unscoped. A minimum established against an existing
    /// bucket is not the minimum for making one.
    fn jetstream_minimum() -> [String; 2] {
        ["$JS.API.INFO".to_owned(), "$JS.API.STREAM.NAMES".to_owned()]
    }

    /// Creating the bucket, which **both** ends need.
    ///
    /// `swarm_queue_client::status::open_or_create` is called by the hive that
    /// writes and the controller that reads, because either may arrive first
    /// on a fresh swarm — a bucket that must pre-exist turns "deployed in the
    /// wrong order" into a permanent, silent absence of data. Whichever
    /// connects first therefore has to be able to create it.
    ///
    /// Narrower than it looks: this is `CREATE` on one named stream, not
    /// `UPDATE` and not the `$JS.API.>` wildcard. Measured rather than
    /// assumed — a hive holding this grant and running `stream edit` leaves
    /// the stream's config untouched, and no `$JS.API.STREAM.UPDATE` is ever
    /// published. Worth stating because the failure it would hide is quiet: a
    /// hive able to reshape the shared bucket could set `MaxMsgs: 1` and evict
    /// every other hive's status without ever touching `STREAM.DELETE`, and
    /// per-key scoping would still look intact.
    fn create(&self) -> String {
        format!("$JS.API.STREAM.CREATE.{}", self.stream())
    }

    /// What every hive needs to open the shared lifecycle-notices stream:
    /// look it up, and create it if this hive is the first to arrive.
    ///
    /// **Not per-hive templated, unlike `extra_hive_subjects`.** The KV
    /// bucket case above namespaces a shared resource *within* itself
    /// (`$KV.<bucket>.<hive>`, one key per hive); the notices stream has
    /// no such per-hive split at the `STREAM.INFO`/`STREAM.CREATE` layer
    /// — the stream itself, not a slice of it, is what every hive's
    /// `open_or_create` needs to reach before it can publish to its own
    /// `hive-notices.<hive>` subject, which [`Self::hive_subjects`] grants
    /// beside these from the same crate constant. Granting these two subjects
    /// to every hive is therefore correct, not a widening: it is
    /// `CREATE`/`INFO` on one named stream, the same shape already
    /// measured safe for the hive-status bucket in [`Self::create`] —
    /// create-if-absent, never `STREAM.UPDATE`, so no hive can reshape a
    /// stream another hive (or the swarm-controller, once it reads from
    /// this stream) already created.
    fn notices_subjects() -> [String; 2] {
        // Unlike `Self::stream()` above, no `KV_` prefix: the notices
        // stream is a plain `JetStream` stream, not a KV bucket, so its
        // NATS stream name *is* `swarm_queue_client::notices::STREAM`
        // verbatim — the `KV_` prefix is `create_key_value`'s own
        // convention, not something every stream carries.
        let stream = swarm_queue_client::notices::STREAM;
        [
            format!("$JS.API.STREAM.INFO.{stream}"),
            format!("$JS.API.STREAM.CREATE.{stream}"),
        ]
    }

    /// What one hive may publish: the account minimum, the bucket lookup,
    /// creation, and its **own** key.
    ///
    /// Two things a reader would reasonably assume, both false and both
    /// measured (`state/attack-3297-js-api-door.sh` in atlas's notes):
    ///
    /// - `$KV.<bucket>.<key>` alone does **not** let a client write that key.
    ///   The client resolves the bucket first, so the `STREAM.INFO` subject is
    ///   part of the minimum for a plain write.
    /// - `$JS.API.>` is not "the `JetStream` permission". It also covers
    ///   `$JS.API.STREAM.DELETE`, with which a hive correctly refused on a
    ///   neighbour's individual key can destroy the whole bucket — every
    ///   hive's data. Granting it would make per-key scoping decorative, which
    ///   is why these are named one at a time and a test asserts the wildcard
    ///   never returns as a convenience.
    fn hive_subjects(&self, hive: &str) -> Vec<String> {
        let mut subjects = Self::jetstream_minimum().to_vec();
        subjects.extend([
            format!("$JS.API.STREAM.INFO.{}", self.stream()),
            self.create(),
            format!("$KV.{}.{hive}", self.bucket),
        ]);
        subjects.extend(Self::notices_subjects());
        // The hive's own slice of the stream the subjects above let it open.
        // Named from the defining crate so a rename cannot leave the grant
        // pointing at a subject nobody publishes to. Not `extra_hive_subjects`:
        // that is for streams this crate does not know about.
        subjects.push(swarm_queue_client::notices::subject(hive));
        // Reading the wanted-state bucket, which is **this hive's own**.
        //
        // One bucket per hive rather than one keyed by hive, so that a *watch*
        // can be granted without widening the read: a consumer's filter travels
        // in the request payload, so `CONSUMER.CREATE` grants a whole stream and
        // cannot be scoped to a key the way `DIRECT.GET` can.
        //
        // `$KV.<bucket>.<hive>` stays deliberately absent — the controller
        // declares this and a hive converges to it, so a hive that could write
        // its own key could declare its own desired state.
        let wanted = Self::wanted_stream(hive);
        subjects.extend([
            format!("$JS.API.STREAM.INFO.{wanted}"),
            format!(
                "$JS.API.DIRECT.GET.{wanted}.$KV.{}.{hive}",
                swarm_queue_client::wanted::bucket(hive)
            ),
            // The watch the paragraph above buys: a KV watch is a consumer, so
            // `DIRECT.GET` covers the boot-time read and nothing after it.
            // Both spellings for the same reason the status buckets grant both
            // — an ephemeral consumer's subject carries no name, and `>` never
            // matches zero tokens.
            format!("$JS.API.CONSUMER.CREATE.{wanted}"),
            format!("$JS.API.CONSUMER.CREATE.{wanted}.>"),
        ]);
        // Publishing this hive's agents into the per-agent status bucket, and
        // **only its own agents**.
        //
        // `.*` matches exactly one token, so this is every agent of one hive
        // and nothing deeper — `.>` would also admit `<hive>.<agent>.<more>`,
        // which `agent_status::split_key` rejects on read anyway; the grant has
        // no reason to be wider than the format.
        //
        // The scoping is only expressible because the key is `.`-joined: a KV
        // entry publishes to `$KV.<bucket>.<key>`, and a NATS wildcard matches
        // whole tokens, so a key joined by anything else is one token and
        // leaves only "one exact subject per agent" (needs a roster in here,
        // which `hive_name`'s doc argues against) or a bucket-wide wildcard
        // (any hive overwrites any other hive's agents).
        //
        // `STREAM.INFO` + `CREATE` come along for the same reason they do
        // above: `open_or_create` resolves the bucket before it writes, and
        // either end may be first on a fresh swarm. Without them the publish
        // grant is unreachable — the client times out at bucket open.
        let agent_status = Self::agent_status_stream();
        subjects.extend([
            format!("$JS.API.STREAM.INFO.{agent_status}"),
            format!("$JS.API.STREAM.CREATE.{agent_status}"),
            format!("$KV.{}.{hive}.*", swarm_queue_client::agent_status::BUCKET),
        ]);
        subjects.extend(
            self.extra_hive_subjects
                .iter()
                .map(|s| s.replace(HIVE_PLACEHOLDER, hive)),
        );
        subjects
    }

    fn reader_subjects(&self) -> Vec<String> {
        let stream = self.stream();
        let mut subjects = Self::jetstream_minimum().to_vec();
        subjects.extend([
            format!("$JS.API.STREAM.INFO.{stream}"),
            self.create(),
            // The `.>` form specifically: the bare `$JS.API.DIRECT.GET.<stream>`
            // is not the subject the client uses, and granting it was measured
            // to make no difference.
            format!("$JS.API.DIRECT.GET.{stream}.>"),
            // `store.keys()` — the controller lists before it fetches, so a
            // reader without this can get a key it already knows and discover
            // nothing.
            //
            // BOTH forms, and the bare one is the one that matters. `keys()`
            // creates an **ephemeral** ordered consumer, whose create subject
            // carries no consumer name — and `>` matches one or more tokens,
            // never zero, so the `.>` form alone does not cover it. Granting
            // only that produced `Permissions Violation for Publish to
            // "$JS.API.CONSUMER.CREATE.KV_hive-status"`, which reaches the
            // client as a **timeout** and the operator as a 503. The `.>` form
            // stays for a named/durable consumer.
            format!("$JS.API.CONSUMER.CREATE.{stream}"),
            format!("$JS.API.CONSUMER.CREATE.{stream}.>"),
            // The knowledge event. One writer, many readers: the controller is
            // the only publisher and every hive subscribes, so this is one
            // literal subject rather than a per-hive family — named from the
            // crate the publisher and the subscriber also name, so a rename
            // cannot leave the grant pointing at a subject nobody uses.
            //
            // This is the reader's only non-JetStream subject, and without it
            // the controller cannot emit the event at all. Worth stating because
            // the symptom is unhelpful: a refused publish reaches the client as
            // a **timeout**, so the visible failure is a hive that never hears
            // about a change, with nothing in the controller's log to say a
            // permission was the reason.
            swarm_queue_client::KNOWLEDGE_SUBJECT.to_owned(),
            // The deploy events: one subject per hive, so a hive is not woken
            // by a deploy meant for another. Granted as a wildcard because
            // this responder has no roster — it cannot enumerate hives, and a
            // grant that had to track one would be a second place to get the
            // list wrong (see `hive_name`'s doc for the same argument about
            // admission). Same failure mode as the knowledge event above: a
            // refused publish reaches the client as a timeout.
            swarm_queue_client::DEPLOY_SUBJECT_WILDCARD.to_owned(),
            // Operator input to an agent, one subject per agent, wildcard for
            // the same no-roster reason. The harness takes the message's
            // `from` as given, which holds only while this client is the
            // family's sole publisher: no hive or agent grant may name it.
            swarm_queue_client::AGENT_INPUT_SUBJECT_WILDCARD.to_owned(),
            // The wanted-state buckets — one per hive, all created and written
            // by this single client.
            //
            // Wildcards rather than a name per hive: a bucket name is one
            // subject token and subjects have no prefix matching, so
            // `KV_hive-wanted-*` is not expressible. The choice is an exact
            // name per hive — which needs a hive roster in this responder,
            // which `hive_name`'s doc argues against — or `*`.
            //
            // ⚠️ `*` therefore reaches every stream in the account, including
            // the buckets hives publish themselves, so this client *can*
            // overwrite a hive's self-reported status. That is an accepted
            // operator decision, not an oversight: the controller is the
            // swarm's writer of record, and a swarm that cannot trust it has a
            // larger problem than this grant.
            "$JS.API.STREAM.INFO.*".to_owned(),
            "$JS.API.STREAM.CREATE.*".to_owned(),
            "$JS.API.DIRECT.GET.*.>".to_owned(),
            // The newest row of one subagent in `term-sub-<agent>`, which the
            // subagent list reads by subject. Those streams do not allow
            // direct get, so `DIRECT.GET` above does not reach them. `*` is a
            // whole token: `term-sub-*` would name one stream literally, and
            // the agent is not known here. Also reaches every other stream's
            // `MSG.GET`, the same width as `STREAM.INFO.*` above. A refused
            // publish reaches the client as a timeout.
            "$JS.API.STREAM.MSG.GET.*".to_owned(),
            "$KV.*.>".to_owned(),
        ]);
        // The per-agent status bucket, read-side only. Same five subjects as
        // the hive-status block above and for the same measured reasons —
        // including BOTH `CONSUMER.CREATE` forms, since `keys()` builds an
        // ephemeral consumer whose subject carries no name.
        //
        // Deliberately **no `$KV.<bucket>.…` subject**: that is the write
        // side, and the controller only reads this bucket. The hive gets a
        // write grant scoped to its own agents (see `hive_subjects`); a
        // reader holding one could forge any agent's status on any hive.
        // Reading needs none of it — a KV read is a `DIRECT.GET`.
        let agent_status = Self::agent_status_stream();
        subjects.extend([
            format!("$JS.API.STREAM.INFO.{agent_status}"),
            format!("$JS.API.STREAM.CREATE.{agent_status}"),
            format!("$JS.API.DIRECT.GET.{agent_status}.>"),
            format!("$JS.API.CONSUMER.CREATE.{agent_status}"),
            format!("$JS.API.CONSUMER.CREATE.{agent_status}.>"),
        ]);
        subjects
    }
}

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

    fn policy() -> Policy {
        Policy::new(
            "hive-".to_owned(),
            "-agent".to_owned(),
            "hive-status".to_owned(),
            vec!["swarm-controller".to_owned()],
            Vec::new(),
        )
        .expect("the default policy is valid")
    }

    #[test]
    fn the_reader_can_open_and_list_the_agent_status_bucket() {
        let subjects = policy().reader_subjects();
        let s = Policy::agent_status_stream();
        for want in [
            format!("$JS.API.STREAM.INFO.{s}"),
            format!("$JS.API.STREAM.CREATE.{s}"),
            format!("$JS.API.DIRECT.GET.{s}.>"),
            // The bare form is the one `keys()` needs: its ephemeral consumer
            // has no name, and `>` never matches zero tokens.
            format!("$JS.API.CONSUMER.CREATE.{s}"),
            format!("$JS.API.CONSUMER.CREATE.{s}.>"),
        ] {
            assert!(subjects.contains(&want), "reader is missing {want}");
        }
        // Control: the same reader really can be missing a subject, so the
        // assertions above are not vacuously true of any string.
        assert!(!subjects.contains(&format!("$JS.API.STREAM.DELETE.{s}")));
    }

    #[test]
    fn no_hive_may_write_another_role_s_agent_status() {
        // A hive publishes its *own* agents' status under
        // `$KV.agent-status.<hive>.*` and must reach no further. This is the
        // arm of the old reader-side guard that still holds.
        let bucket = swarm_queue_client::agent_status::BUCKET;
        let subjects = policy().hive_subjects("alpha");
        for forbidden in [
            format!("$KV.{bucket}.>"),
            format!("$KV.{bucket}.*"),
            format!("$KV.{bucket}.beta.*"),
        ] {
            assert!(!subjects.contains(&forbidden), "hive alpha got {forbidden}");
        }
        // Control: the containment being asserted is not vacuous — the hive
        // does hold its own slice.
        assert!(subjects.contains(&format!("$KV.{bucket}.alpha.*")));
    }

    #[test]
    fn the_readers_grant_is_deliberately_account_wide() {
        // ⚠️ This pins a DECISION, not a safety property, and it replaced a
        // guard that asserted the opposite — the operator's call was that the
        // controller being able to override a hive is acceptable.
        //
        // A bucket name is one subject token and subjects have no prefix
        // matching, so per-hive wanted-state buckets cannot be covered by any
        // wildcard narrower than `*`. Scoping the controller would mean a hive
        // roster inside this responder. If that trade is ever revisited, this
        // test is the thing to change — it exists so the width reads as chosen.
        let subjects = policy().reader_subjects();
        assert!(subjects.contains(&"$KV.*.>".to_owned()));
        assert!(subjects.contains(&"$JS.API.STREAM.INFO.*".to_owned()));
    }

    #[test]
    fn a_hive_may_write_only_its_own_agents_status_keys() {
        let bucket = swarm_queue_client::agent_status::BUCKET;
        let subjects = policy().hive_subjects("alpha");
        let s = Policy::agent_status_stream();
        for want in [
            // Without these the write subject is unreachable: `open_or_create`
            // resolves the bucket first and times out before it publishes.
            format!("$JS.API.STREAM.INFO.{s}"),
            format!("$JS.API.STREAM.CREATE.{s}"),
            format!("$KV.{bucket}.alpha.*"),
        ] {
            assert!(subjects.contains(&want), "hive alpha is missing {want}");
        }
        // The scoping is the point, so pin what must NOT be there: a
        // bucket-wide wildcard in any of its spellings, and another hive's
        // slice. `.*` matches one token; `.>` would match deeper than the key
        // format ever produces.
        for forbidden in [
            format!("$KV.{bucket}.>"),
            format!("$KV.{bucket}.*"),
            format!("$KV.{bucket}.alpha.>"),
            format!("$KV.{bucket}.beta.*"),
        ] {
            assert!(
                !subjects.contains(&forbidden),
                "hive alpha was granted {forbidden}, which reaches past its own agents"
            );
        }
    }

    #[test]
    fn an_extra_subject_without_the_placeholder_is_refused() {
        // 🩸 The option exists to put a second stream inside ONE hive's
        // namespace. A template with no `{hive}` expands to the same subject
        // for all of them, so the flag whose purpose is scoping becomes the
        // way to remove it - silently, and only in the deployment that set it.
        let err = Policy::new(
            "hive-".to_owned(),
            "-agent".to_owned(),
            "hive-status".to_owned(),
            Vec::new(),
            vec!["$SWARM.events.all".to_owned()],
        )
        .expect_err("a subject shared by every hive must not be accepted");
        let msg = format!("{err}");
        assert!(
            msg.contains("$SWARM.events.all"),
            "the error must name the offending template, got: {msg}"
        );
    }

    #[test]
    fn an_extra_subject_with_the_placeholder_is_accepted() {
        // The other half: a check that only ever rejects would be indistinguishable
        // from the option being broken.
        Policy::new(
            "hive-".to_owned(),
            "-agent".to_owned(),
            "hive-status".to_owned(),
            Vec::new(),
            vec!["$SWARM.events.{hive}.>".to_owned()],
        )
        .expect("a per-hive template is the shape this option is for");
    }

    #[test]
    fn a_hive_may_publish_its_own_key_and_nothing_elses() {
        let p = policy()
            .permissions("hive-alpha")
            .expect("a hive is admitted");
        assert!(p.publish.contains(&"$KV.hive-status.alpha".to_owned()));
        assert!(!p.publish.iter().any(|s| s.contains("beta")));
    }

    #[test]
    fn a_reader_may_publish_the_knowledge_event() {
        let p = policy().permissions("swarm-controller").expect("a reader");
        assert!(
            p.publish
                .contains(&swarm_queue_client::KNOWLEDGE_SUBJECT.to_owned()),
            "the controller is the only publisher of this event; without the \
             grant its publish is refused, and a refusal arrives as a timeout"
        );
    }

    #[test]
    fn a_hive_may_not_publish_the_knowledge_event_to_anyone_including_itself() {
        // The controller *interprets* what a delivery means; a hive receives
        // that verdict. A hive able to publish here could tell every other hive
        // in the swarm — or itself — that the knowledge repo changed when it did
        // not, which is an unauthenticated write into someone else's control
        // path wearing an event's shape.
        //
        // One writer and many readers makes this arm matter MORE, not less: with
        // a single shared subject a forged event reaches the whole swarm, where
        // a per-hive subject would have reached one.
        let p = policy()
            .permissions("hive-alpha")
            .expect("a hive is admitted");
        assert!(
            !p.publish
                .iter()
                .any(|s| s == swarm_queue_client::KNOWLEDGE_SUBJECT),
            "a hive must not publish the knowledge event: {:?}",
            p.publish
        );
    }

    #[test]
    fn a_reader_may_publish_a_deploy_event_to_any_hive() {
        let p = policy().permissions("swarm-controller").expect("a reader");
        assert!(
            p.publish
                .contains(&swarm_queue_client::DEPLOY_SUBJECT_WILDCARD.to_owned()),
            "the controller is the only publisher of these events; without the \
             grant its publish is refused, and a refusal arrives as a timeout"
        );
    }

    #[test]
    fn a_hive_may_not_publish_a_deploy_event_to_anyone_including_itself() {
        // Same arm as the knowledge event's, and it matters more here: a forged
        // knowledge event makes a hive re-read a repo, while a forged deploy
        // event makes it rebuild and restart a named agent.
        //
        // Both directions asserted, because splitting the subject per hive
        // makes them separate strings for the first time: a hive must reach
        // neither another hive's deploy subject nor its own. Nothing in the
        // grant should mention this family at all.
        let p = policy()
            .permissions("hive-alpha")
            .expect("a hive is admitted");
        for forbidden in [
            swarm_queue_client::deploy_subject("beta"),
            swarm_queue_client::deploy_subject("alpha"),
            swarm_queue_client::DEPLOY_SUBJECT_WILDCARD.to_owned(),
        ] {
            assert!(
                !p.publish.contains(&forbidden),
                "a hive must not publish {forbidden}: {:?}",
                p.publish
            );
        }
    }

    #[test]
    fn a_reader_may_publish_input_to_any_agent() {
        let p = policy().permissions("swarm-controller").expect("a reader");
        assert!(
            p.publish
                .contains(&swarm_queue_client::AGENT_INPUT_SUBJECT_WILDCARD.to_owned()),
            "the controller is the only publisher of operator input; without \
             the grant its request is refused, and a refusal arrives as a timeout"
        );
    }

    #[test]
    fn a_hive_may_not_publish_input_to_an_agent() {
        // The harness turns this message into a turn from the operator named in
        // it. A hive able to publish here could speak as the operator to any
        // agent in the swarm.
        let p = policy().permissions("hive-alpha").expect("admitted");
        for forbidden in [
            swarm_queue_client::agent_input_subject("alpha"),
            swarm_queue_client::AGENT_INPUT_SUBJECT_WILDCARD.to_owned(),
        ] {
            assert!(
                !p.publish.contains(&forbidden),
                "a hive must not publish {forbidden}: {:?}",
                p.publish
            );
        }
        assert!(
            !p.publish.iter().any(|s| s.starts_with("$SWARM.input")),
            "{:?}",
            p.publish
        );
    }

    #[test]
    fn a_hive_grant_never_includes_the_jetstream_wildcard() {
        // `$JS.API.>` also covers `$JS.API.STREAM.DELETE.KV_hive-status`, with
        // which a hive refused on a neighbour's key can delete the entire
        // bucket. Measured, not theorised - this test is the guard against it
        // coming back as a convenience.
        let p = policy()
            .permissions("hive-alpha")
            .expect("a hive is admitted");
        assert!(!p.publish.iter().any(|s| s.contains("$JS.API.>")));
        assert!(!p.publish.iter().any(|s| s.contains("STREAM.DELETE")));
        assert!(!p.publish.iter().any(|s| s.contains("STREAM.PURGE")));
    }

    #[test]
    fn the_bucket_lookup_is_part_of_a_publishers_minimum() {
        // Without it the put is refused: the client resolves the bucket before
        // it writes. Dropping this line looks like tightening and is breaking.
        let p = policy()
            .permissions("hive-alpha")
            .expect("a hive is admitted");
        assert!(
            p.publish
                .contains(&"$JS.API.STREAM.INFO.KV_hive-status".to_owned())
        );
    }

    #[test]
    fn every_grant_carries_the_jetstream_minimum() {
        // 🩸 Found by the shipping gate, not by any unit test: a grant with
        // every bucket-specific subject and neither of these cannot create the
        // bucket at all. The client times out on `$JS.API.INFO` before it
        // reaches anything that was granted, and a NATS denial looks like a
        // hang from the client side — the server log is what named them.
        for client in ["hive-alpha", "swarm-controller"] {
            let p = policy().permissions(client).expect("admitted");
            for required in ["$JS.API.INFO", "$JS.API.STREAM.NAMES"] {
                assert!(
                    p.publish.iter().any(|s| s == required),
                    "{client} is missing {required}, so it cannot use JetStream at all"
                );
            }
        }
    }

    #[test]
    fn both_ends_may_create_the_bucket_but_not_reshape_it() {
        // 🩸 The bug the subject measurements could not see: they ran against
        // a bucket the *setup* had already created while unscoped, so
        // "minimal" meant minimal-given-a-bucket-that-exists. `open_or_create`
        // is called by both ends by design — whichever arrives first on a
        // fresh swarm makes the bucket — so without CREATE a new swarm never
        // gets one, and every other test still passes.
        for client in ["hive-alpha", "swarm-controller"] {
            let p = policy().permissions(client).expect("admitted");
            assert!(
                p.publish
                    .contains(&"$JS.API.STREAM.CREATE.KV_hive-status".to_owned()),
                "{client} cannot create the bucket on a fresh swarm"
            );
            // CREATE is not UPDATE: a second arrival must not be able to
            // reshape the bucket the first one made.
            assert!(!p.publish.iter().any(|s| s.contains("STREAM.UPDATE")));
        }
    }

    #[test]
    fn a_reader_may_list_and_fetch_but_not_write() {
        let p = policy()
            .permissions("swarm-controller")
            .expect("the reader is admitted");
        // The BARE subject, asserted separately and first: `keys()` creates an
        // ephemeral consumer, so its create subject ends at the stream name,
        // and `>` matches one or more tokens rather than zero. This assertion
        // used to name only the `.>` form below — which reads as covering the
        // bare one and does not, so the suite was green while every list timed
        // out in production.
        assert!(
            p.publish
                .contains(&"$JS.API.CONSUMER.CREATE.KV_hive-status".to_owned()),
            "an ephemeral consumer create carries no name token"
        );
        assert!(
            p.publish
                .contains(&"$JS.API.CONSUMER.CREATE.KV_hive-status.>".to_owned())
        );
        assert!(
            p.publish
                .contains(&"$JS.API.DIRECT.GET.KV_hive-status.>".to_owned())
        );
        // Scoped to the status bucket, not to `$KV.` as a whole: the hives
        // write that one and a controller able to write it could forge a
        // hive's own report. The wanted bucket runs the other way and the
        // controller is its writer, so a blanket assertion here would forbid
        // the grant that bucket exists for.
        assert!(!p.publish.iter().any(|s| s.starts_with("$KV.hive-status")));
    }

    #[test]
    fn an_unmatched_client_is_denied_not_granted_everything() {
        // The whole policy is advisory if this returns a grant.
        assert_eq!(policy().permissions("some-other-service"), None);
        assert_eq!(policy().permissions(""), None);
    }

    /// Whether the grant `grant` covers the concrete subject `subject`, by
    /// NATS' token rules: `*` is one token, `>` one or more.
    fn covers(grant: &str, subject: &str) -> bool {
        let g: Vec<&str> = grant.split('.').collect();
        let s: Vec<&str> = subject.split('.').collect();
        for (i, t) in g.iter().enumerate() {
            match *t {
                ">" => return s.len() > i,
                "*" if i < s.len() => {}
                t if s.get(i) == Some(&t) => {}
                _ => return false,
            }
        }
        g.len() == s.len()
    }

    fn may_subscribe(p: &Permissions, subject: &str) -> bool {
        let grants = p
            .subscribe
            .as_ref()
            .expect("an agent's subscribe is an allow-list, never unrestricted");
        assert!(!grants.is_empty(), "an empty allow-list is unrestricted");
        grants.iter().any(|g| covers(g, subject))
    }

    /// The reply inbox each agent client connects with, as a subject a
    /// request under it would receive on.
    fn reply(prefix: &str) -> String {
        format!("{prefix}.nuid123.token456")
    }

    #[test]
    fn an_agent_token_may_subscribe_to_its_own_inbox_and_input_only() {
        use swarm_queue_client::{agent_inbox_prefix, agent_input_subject};

        let p = policy()
            .with_agent_token_subjects(vec!["$SWARM.term.{agent}".to_owned()])
            .expect("a per-agent template is valid");
        let atlas = p.agent_token_permissions("atlas").expect("configured");
        assert!(may_subscribe(&atlas, &reply(&agent_inbox_prefix("atlas"))));
        assert!(may_subscribe(&atlas, &agent_input_subject("atlas")));
        for refused in [
            ">".to_owned(),
            "$SWARM.term.>".to_owned(),
            "$SWARM.term.argus".to_owned(),
            "$SWARM.term.atlas".to_owned(),
            "$SWARM.input.*".to_owned(),
            agent_input_subject("argus"),
            reply(&agent_inbox_prefix("argus")),
            "_INBOX.nuid123.token456".to_owned(),
        ] {
            assert!(
                !may_subscribe(&atlas, &refused),
                "atlas may subscribe to {refused}"
            );
        }
    }

    #[test]
    fn hives_and_readers_keep_an_unrestricted_subscribe() {
        let p = policy();
        assert_eq!(p.permissions("hive-alpha").expect("a hive").subscribe, None);
        assert_eq!(
            p.permissions("swarm-controller")
                .expect("the reader")
                .subscribe,
            None
        );
    }

    #[test]
    fn the_prefix_alone_names_no_hive() {
        // `hive-` would otherwise produce `$KV.hive-status.`, a subject nobody
        // reviewed and that no hive owns.
        assert_eq!(policy().permissions("hive-"), None);
    }

    #[test]
    fn a_hive_may_open_the_shared_notices_stream() {
        // 🩸 The actual defect this closes: `open_or_create`'s `get_stream`
        // call needs these two before a hive ever reaches its own
        // `hive-notices.<hive>` publish subject, and neither has a
        // `{hive}` to go through `extra_hive_subjects`.
        let p = policy()
            .permissions("hive-alpha")
            .expect("a hive is admitted");
        assert!(
            p.publish
                .contains(&"$JS.API.STREAM.INFO.hive-notices".to_owned())
        );
        assert!(
            p.publish
                .contains(&"$JS.API.STREAM.CREATE.hive-notices".to_owned())
        );
    }

    #[test]
    fn a_hive_may_publish_to_its_own_notices_subject() {
        // Opening the stream and writing into it are separate grants.
        let p = policy()
            .permissions("hive-alpha")
            .expect("a hive is admitted");
        assert!(p.publish.contains(&"hive-notices.alpha".to_owned()));
    }

    #[test]
    fn a_hive_may_not_publish_to_another_hives_notices_subject() {
        // The stream-level subjects are identical for every hive; this one
        // must not be, or reaching the stream would carry writing to every
        // other hive's slice of it.
        let p = policy()
            .permissions("hive-alpha")
            .expect("a hive is admitted");
        assert!(!p.publish.contains(&"hive-notices.beta".to_owned()));
        assert!(!p.publish.iter().any(|s| s == "hive-notices.>"));
    }

    #[test]
    fn a_hive_may_read_its_own_wanted_key() {
        // A KV read is a publish, and the direct-get subject carries the key.
        let p = policy()
            .permissions("hive-alpha")
            .expect("a hive is admitted");
        assert!(
            p.publish.contains(
                &"$JS.API.DIRECT.GET.KV_hive-wanted-alpha.$KV.hive-wanted-alpha.alpha".to_owned()
            ),
            "got: {:?}",
            p.publish
        );
    }

    #[test]
    fn a_hive_may_watch_its_own_wanted_bucket() {
        // A watch is a consumer, so `DIRECT.GET` covers the boot-time read and
        // nothing else. Both spellings, for the reason the status buckets grant
        // both: an ephemeral consumer's subject carries no name, and `>` never
        // matches zero tokens.
        let p = policy()
            .permissions("hive-alpha")
            .expect("a hive is admitted");
        for want in [
            "$JS.API.CONSUMER.CREATE.KV_hive-wanted-alpha",
            "$JS.API.CONSUMER.CREATE.KV_hive-wanted-alpha.>",
        ] {
            assert!(p.publish.contains(&want.to_owned()), "got: {:?}", p.publish);
        }
        // Control: the grant's width is one stream, not the account — this is
        // the property the per-hive bucket split bought.
        assert!(
            !p.publish
                .iter()
                .any(|s| s.starts_with("$JS.API.CONSUMER.CREATE.KV_hive-wanted-beta"))
        );
    }

    #[test]
    fn a_hive_may_not_read_another_hives_wanted_key() {
        // The whole reason the read is scoped by key rather than by bucket.
        let p = policy()
            .permissions("hive-alpha")
            .expect("a hive is admitted");
        // ⚠️ Matched against the CURRENT bucket name. The per-hive split moved
        // this from `hive-wanted.beta` to `hive-wanted-beta`, and the old
        // spelling kept passing against a subject set that no longer contains
        // it — a pass that had stopped meaning anything.
        assert!(!p.publish.iter().any(|s| s.contains("hive-wanted-beta")));
        assert!(!p.publish.iter().any(|s| s.contains("beta")));
        assert!(
            !p.publish
                .iter()
                .any(|s| s.starts_with("$JS.API.DIRECT.GET.KV_hive-wanted") && s.ends_with('>'))
        );
    }

    #[test]
    fn a_hive_may_not_write_its_own_wanted_key() {
        // The controller declares wanted state; a hive that could write its own
        // key could declare its own.
        let p = policy()
            .permissions("hive-alpha")
            .expect("a hive is admitted");
        // Current names, for the reason the sibling test above records: the
        // pre-split strings `$KV.hive-wanted.{alpha,>}` still "passed" after
        // the rename because nothing produces them any more.
        assert!(!p.publish.iter().any(|s| s == "$KV.hive-wanted-alpha.alpha"));
        assert!(!p.publish.iter().any(|s| s == "$KV.hive-wanted-alpha.>"));
        assert!(!p.publish.iter().any(|s| s.starts_with("$KV.hive-wanted")));
    }

    #[test]
    fn the_controller_may_create_and_write_the_wanted_bucket() {
        // Presence control for the three above: the subjects exist, on the one
        // identity that is meant to hold them.
        let p = policy()
            .permissions("swarm-controller")
            .expect("a reader is admitted");
        // Account-wide since the per-hive split: a bucket name is one subject
        // token, so no wildcard narrower than `*` covers N per-hive buckets.
        // See `the_readers_grant_is_deliberately_account_wide`.
        assert!(p.publish.contains(&"$KV.*.>".to_owned()));
        assert!(p.publish.contains(&"$JS.API.STREAM.CREATE.*".to_owned()));
    }

    #[test]
    fn the_controller_may_read_a_subagents_newest_row() {
        let p = policy()
            .permissions("swarm-controller")
            .expect("a reader is admitted");
        assert!(p.publish.contains(&"$JS.API.STREAM.MSG.GET.*".to_owned()));
        // `*` only matches a whole token, so a prefix spelling grants one
        // stream literally named that and reads nothing.
        assert!(!p.publish.iter().any(|s| s.contains("term-sub-*")));
        // Control: a hive's grant holds no `MSG.GET`, so the subject above is
        // the controller's alone.
        assert!(
            !policy()
                .hive_subjects("alpha")
                .iter()
                .any(|s| s.starts_with("$JS.API.STREAM.MSG.GET"))
        );
    }

    #[test]
    fn the_notices_stream_grant_is_identical_across_hives() {
        // Unlike `$KV.<bucket>.<hive>` or an `extra_hive_subjects`
        // template, these two subjects name the stream itself, not a
        // per-hive slice of it — every hive gets the exact same two
        // strings, and that is the correct shape, not an oversight.
        let alpha = policy()
            .permissions("hive-alpha")
            .expect("a hive is admitted");
        let beta = policy()
            .permissions("hive-beta")
            .expect("a hive is admitted");
        for subject in [
            "$JS.API.STREAM.INFO.hive-notices",
            "$JS.API.STREAM.CREATE.hive-notices",
        ] {
            assert!(alpha.publish.contains(&subject.to_owned()));
            assert!(beta.publish.contains(&subject.to_owned()));
        }
    }

    #[test]
    fn the_notices_grant_never_includes_stream_update_or_delete() {
        // Same invariant `a_hive_grant_never_includes_the_jetstream_wildcard`
        // holds for the status bucket, restated for the stream this fold
        // grants CREATE/INFO on: create-if-absent must not become
        // reshape-or-destroy.
        let p = policy()
            .permissions("hive-alpha")
            .expect("a hive is admitted");
        assert!(
            !p.publish
                .iter()
                .any(|s| s.contains("hive-notices") && s.contains("STREAM.UPDATE"))
        );
        assert!(
            !p.publish
                .iter()
                .any(|s| s.contains("hive-notices") && s.contains("STREAM.DELETE"))
        );
    }

    #[test]
    fn extra_subjects_are_scoped_to_the_hive_that_publishes_them() {
        // The extension point: a second stream (lifecycle notices, say) is
        // published by the same `hive-<name>` identity on a different subject.
        // It has to land inside that hive's namespace, or one hive can write
        // another's events even though its status key is scoped.
        let p = Policy::new(
            "hive-".to_owned(),
            "-agent".to_owned(),
            "hive-status".to_owned(),
            Vec::new(),
            vec!["$SWARM.events.{hive}.>".to_owned()],
        )
        .expect("a per-hive template is valid");
        let g = p.permissions("hive-alpha").expect("a hive is admitted");
        assert!(g.publish.contains(&"$SWARM.events.alpha.>".to_owned()));
        assert!(!g.publish.iter().any(|s| s.contains("{hive}")));
    }

    #[test]
    fn the_hives_shared_agent_client_is_refused() {
        // Agents connect with their own credential. The shared client names
        // the hive and never the agent, so any grant to it would let one agent
        // publish as its siblings.
        //
        // 🩸 The agent id satisfies the HIVE parse too (`hive-alpha-agent`
        // strips to a hive named `alpha-agent`), so a hive-first arm order
        // would hand this client its hive's grant, status key included, and
        // nothing would report it. `is_none` here is what holds the order.
        assert!(policy().permissions("hive-alpha-agent").is_none());
        // Control: the hive itself is admitted, so the refusal above is the
        // agent arm answering rather than the prefix failing to match.
        assert!(policy().permissions("hive-alpha").is_some());
    }

    #[test]
    fn an_empty_agent_suffix_is_refused() {
        // 🩸 An agent id is a hive id plus the suffix, so an empty suffix makes
        // `strip_suffix` succeed on every hive id: the two principals become
        // one string, and the agent arm, which runs first and refuses, would
        // refuse every hive. That fails at construction rather than at connect.
        Policy::new(
            "hive-".to_owned(),
            String::new(),
            "hive-status".to_owned(),
            Vec::new(),
            Vec::new(),
        )
        .expect_err("an empty agent suffix must not construct");
    }

    #[test]
    fn an_agent_token_grant_is_the_agents_own_subjects_and_nothing_else() {
        let p = policy()
            .with_agent_token_subjects(vec![
                "$SWARM.term.{agent}".to_owned(),
                "$SWARM.agent-state.{agent}".to_owned(),
            ])
            .expect("per-agent templates are valid");
        let g = p.agent_token_permissions("atlas").expect("configured");
        assert_eq!(
            g.publish,
            vec![
                "$SWARM.term.atlas".to_owned(),
                "$SWARM.agent-state.atlas".to_owned(),
            ]
        );
    }

    /// A verified agent may write its own icon key and no other agent's.
    #[test]
    fn an_agent_may_write_only_its_own_icon_key() {
        use swarm_queue_client::agent_icon::subject;

        let p = policy()
            .with_agent_token_subjects(vec!["$KV.agent-icons.{agent}".to_owned()])
            .expect("a per-agent template is valid");
        let atlas = p.agent_token_permissions("atlas").expect("configured");
        assert_eq!(atlas.publish, vec![subject("atlas")]);
        assert!(!atlas.publish.contains(&subject("argus")));
        // Control: the same template does grant argus its own key, so atlas
        // lacking it is atlas's grant being scoped.
        let argus = p.agent_token_permissions("argus").expect("configured");
        assert_eq!(argus.publish, vec![subject("argus")]);
    }

    /// The subagent template as `swarm-nats.nix` spells it expands to the
    /// subjects the subagent daemon publishes on, publish only. The stream is
    /// the controller's to create, which its own grant covers.
    #[test]
    fn an_agents_subagent_grant_is_publish_on_its_own_family_only() {
        use swarm_queue_client::subagent_term::{stream_name, stream_subjects, subject};

        let p = policy()
            .with_agent_token_subjects(vec!["$SWARM.term.{agent}.sub.>".to_owned()])
            .expect("a per-agent template is valid");
        let g = p.agent_token_permissions("atlas").expect("configured");
        assert_eq!(g.publish, vec![stream_subjects("atlas")]);
        assert!(subject("atlas", "scout").starts_with(g.publish[0].trim_end_matches('>')));
        let argus = p.agent_token_permissions("argus").expect("configured");
        assert_eq!(argus.publish, vec![stream_subjects("argus")]);
        assert_ne!(g.publish, argus.publish);

        let stream = stream_name("atlas");
        assert!(
            !stream.contains('.'),
            "{stream} is one token, so `*` covers it"
        );
        assert!(
            policy()
                .reader_subjects()
                .contains(&"$JS.API.STREAM.CREATE.*".to_owned())
        );
    }

    /// The schedule template as `swarm-nats.nix` spells it reads one key: the
    /// agent's own. It covers no other agent's key, no other stream or bucket,
    /// and no way to open, watch or write the bucket.
    #[test]
    fn an_agent_may_read_only_its_own_schedule_key() {
        use swarm_queue_client::agent_schedules::{STREAM, direct_get_subject, key_subject};

        let p = policy()
            .with_agent_token_subjects(vec![
                "$JS.API.DIRECT.GET.KV_agent-schedules.$KV.agent-schedules.{agent}".to_owned(),
            ])
            .expect("a per-agent template is valid");
        let atlas = p.agent_token_permissions("atlas").expect("configured");
        assert_eq!(atlas.publish, vec![direct_get_subject("atlas")]);
        // Control: argus's own expansion is a different subject, so atlas
        // lacking it is the grant being scoped, not the template being empty.
        let argus = p.agent_token_permissions("argus").expect("configured");
        assert_eq!(argus.publish, vec![direct_get_subject("argus")]);
        assert!(!atlas.publish.contains(&direct_get_subject("argus")));

        for refused in [
            direct_get_subject("argus"),
            format!("$JS.API.DIRECT.GET.{STREAM}"),
            format!("$JS.API.STREAM.INFO.{STREAM}"),
            format!("$JS.API.STREAM.CREATE.{STREAM}"),
            format!("$JS.API.STREAM.DELETE.{STREAM}"),
            format!("$JS.API.STREAM.MSG.GET.{STREAM}"),
            format!("$JS.API.CONSUMER.CREATE.{STREAM}"),
            format!(
                "$JS.API.CONSUMER.CREATE.{STREAM}.w.{}",
                key_subject("atlas")
            ),
            key_subject("atlas"),
            key_subject("argus"),
            "$JS.API.DIRECT.GET.KV_agent-icons.$KV.agent-icons.atlas".to_owned(),
            "$JS.API.DIRECT.GET.KV_hive-wanted-alpha.$KV.hive-wanted-alpha.alpha".to_owned(),
            "$JS.API.INFO".to_owned(),
        ] {
            assert!(
                !atlas.publish.iter().any(|g| covers(g, &refused)),
                "atlas's grant covers {refused}"
            );
        }
        // Control: the matcher does match, so the refusals above are not
        // vacuous.
        assert!(covers(&atlas.publish[0], &direct_get_subject("atlas")));
        assert!(covers("$KV.*.>", &key_subject("argus")));
    }

    #[test]
    fn an_agent_token_subject_without_the_placeholder_is_refused() {
        let err = policy()
            .with_agent_token_subjects(vec!["$SWARM.term.all".to_owned()])
            .expect_err("a subject shared by every agent is not the agent's own");
        assert!(format!("{err}").contains("$SWARM.term.all"), "{err}");
    }

    #[test]
    fn with_no_agent_token_subject_configured_an_agent_token_is_refused() {
        assert!(policy().agent_token_permissions("atlas").is_none());
    }
}