swarm-nats-auth: grant the wanted-state bucket, scoped by direction

The bucket the previous commit adds had no grants, so the controller
could not create or write it and no hive could read its own key.

Measured against nats-server 2.14.4 rather than extended by analogy,
because the shapes are not symmetric:

- controller: `STREAM.INFO` + `STREAM.CREATE` on `KV_hive-wanted` and
  `$KV.hive-wanted.>`. With only today's status grants, `get_key_value`
  timed out and the server named the two missing stream subjects.
- hive: `STREAM.INFO` plus **one** direct-get subject carrying its own
  key. A KV read is a publish — `store.get` is a request — and the
  direct-get subject embeds the key, so the read scopes per hive. By
  analogy with `reader_subjects` this would have been `.>`, handing
  every hive every hive's wanted set.

Both refusals were verified to fire, not assumed: as `alpha`,
`get beta` was refused naming
`$JS.API.DIRECT.GET.KV_hive-wanted.$KV.hive-wanted.beta`, and
`put alpha` was refused naming `$KV.hive-wanted.alpha`.

`a_reader_may_list_and_fetch_but_not_write` asserted a reader holds no
`$KV.` subject at all, which held only while status was the sole
bucket. Narrowed to the status bucket — the invariant it defends is
that the controller cannot forge a hive's own report, and the wanted
bucket runs the other way.

Refs #3124

swarm-nats-auth/src/policy.rs

@@ -138,6 +138,16 @@ impl Policy {
         format!("KV_{}", self.bucket)
     }
 
+    /// The stream backing the wanted-state bucket.
+    ///
+    /// Derived from the crate constant rather than from a second configurable
+    /// bucket name: `wanted::BUCKET` is a `const` precisely so writer and
+    /// reader cannot disagree about it, and a flag here would reintroduce the
+    /// disagreement one layer out.
+    fn wanted_stream() -> String {
+        format!("KV_{}", swarm_queue_client::wanted::BUCKET)
+    }
+
     /// What *any* `JetStream` client must be able to ask before it can do
     /// anything at all, bucket-specific or not.
     ///
@@ -238,6 +248,21 @@ impl Policy {
         // 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, and **only this hive's key**.
+        //
+        // A KV read is a publish: `store.get` is a request, and the direct-get
+        // subject carries the key, so the grant scopes to one hive rather than
+        // to the bucket. `$KV.<wanted>.<hive>` is 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.
+        subjects.extend([
+            format!("$JS.API.STREAM.INFO.{}", Self::wanted_stream()),
+            format!(
+                "$JS.API.DIRECT.GET.{}.$KV.{}.{hive}",
+                Self::wanted_stream(),
+                swarm_queue_client::wanted::BUCKET
+            ),
+        ]);
         subjects.extend(
             self.extra_hive_subjects
                 .iter()
@@ -291,6 +316,13 @@ impl Policy {
             // 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(),
+            // The wanted-state bucket, which the controller creates and writes
+            // for every hive. `.>` here against one key per hive above: this is
+            // the single writer, so scoping it per hive would be a list of
+            // hives to keep, which `hive_name`'s doc argues against.
+            format!("$JS.API.STREAM.INFO.{}", Self::wanted_stream()),
+            format!("$JS.API.STREAM.CREATE.{}", Self::wanted_stream()),
+            format!("$KV.{}.>", swarm_queue_client::wanted::BUCKET),
         ]);
         subjects
     }
@@ -513,7 +545,12 @@ mod tests {
             p.publish
                 .contains(&"$JS.API.DIRECT.GET.KV_hive-status.>".to_owned())
         );
-        assert!(!p.publish.iter().any(|s| s.starts_with("$KV.")));
+        // 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]
@@ -570,6 +607,59 @@ mod tests {
         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.$KV.hive-wanted.alpha".to_owned()),
+            "got: {:?}",
+            p.publish
+        );
+    }
+
+    #[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");
+        assert!(!p.publish.iter().any(|s| s.contains("hive-wanted.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");
+        assert!(!p.publish.iter().any(|s| s == "$KV.hive-wanted.alpha"));
+        assert!(!p.publish.iter().any(|s| s == "$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");
+        assert!(p.publish.contains(&"$KV.hive-wanted.>".to_owned()));
+        assert!(
+            p.publish
+                .contains(&"$JS.API.STREAM.CREATE.KV_hive-wanted".to_owned())
+        );
+    }
+
     #[test]
     fn the_notices_stream_grant_is_identical_across_hives() {
         // Unlike `$KV.<bucket>.<hive>` or an `extra_hive_subjects`

swarm-queue-client/src/lib.rs

@@ -164,6 +164,13 @@ pub fn chain(error: &dyn std::error::Error) -> String {
 /// 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 subject the swarm controller publishes on when the hive-wide knowledge
 /// repository has changed. One writer, many readers — every hive subscribes.
 ///

swarm-queue-client/src/wanted.rs

@@ -0,0 +1,80 @@
+//! The hive-wanted KV bucket: the agent set the controller declares for each
+//! hive, keyed by `hiveName`.
+//!
+//! Sibling of [`crate::status`] and deliberately **not** a mirror of it: that
+//! one is **observed** — each hive republishes what it is, so its store losing
+//! everything "degrades to honesty" (`swarm-nats.nix`). This one is
+//! **declared**, and nothing regenerates it; lose it and a reconcile loop has
+//! nothing to converge to, which is the failure such a loop exists to remove.
+//!
+//! # Absence is not a deletion order
+//!
+//! A hive that finds no key for itself — or no bucket at all — has learned
+//! nothing about the agents it is running, not that it should have none.
+//! Swarm-side lifecycle does not yet cover agents that predate it, so callers
+//! converge the agents a value NAMES and leave the rest alone.
+//!
+//! # Only the controller creates it
+//!
+//! Unlike [`crate::status::open_or_create`], where either end may legitimately
+//! arrive first, the writer here is single and known. A hive opens read-only
+//! and treats absence as the case above, so "not published yet" stays quiet
+//! rather than looking like an error a hive could fix.
+
+#[cfg(feature = "kv")]
+use crate::Error;
+
+/// The KV bucket the controller publishes per-hive wanted state into, one key
+/// per hive keyed by `hiveName`.
+///
+/// A constant and not an option, for the reason [`crate::status::BUCKET`]
+/// gives: writer and reader must name the same bucket, and an option is a way
+/// for two deployments to disagree about which one that is.
+pub const BUCKET: &str = "hive-wanted";
+
+/// Open the wanted-state bucket for writing, creating it if nothing has yet.
+///
+/// **Controller-side only.** `history: 1` because a hive converges to the
+/// current declaration and never asks what the previous one was — that
+/// question is answered by the controller's own records, not by replaying a
+/// bucket.
+#[cfg(feature = "kv")]
+pub async fn open_or_create(
+    client: &async_nats::Client,
+) -> Result<async_nats::jetstream::kv::Store, Error> {
+    let js = async_nats::jetstream::new(client.clone());
+    match js.get_key_value(BUCKET).await {
+        Ok(store) => Ok(store),
+        Err(e) => {
+            tracing::info!(
+                bucket = BUCKET,
+                reason = %e,
+                "wanted-state bucket not available, creating it"
+            );
+            js.create_key_value(async_nats::jetstream::kv::Config {
+                bucket: BUCKET.to_owned(),
+                description: "Agent set the swarm controller declares for each hive".to_owned(),
+                history: 1,
+                ..Default::default()
+            })
+            .await
+            .map_err(|source| Error::CreateBucket {
+                bucket: BUCKET,
+                source,
+            })
+        }
+    }
+}
+
+/// Open the wanted-state bucket for reading, or `None` when it does not exist.
+///
+/// **Hive-side.** `None` is the ordinary pre-publication state, not a failure:
+/// a hive holds no grant to create this bucket and must not treat its absence
+/// as a reason to converge to an empty agent set — see the module docs.
+#[cfg(feature = "kv")]
+pub async fn open_read_only(
+    client: &async_nats::Client,
+) -> Option<async_nats::jetstream::kv::Store> {
+    let js = async_nats::jetstream::new(client.clone());
+    js.get_key_value(BUCKET).await.ok()
+}