swarm-nats-auth: let the reader open the agent-status bucket

The swarm-controller had no grant for the per-agent status bucket, so
resolving it failed and the dashboard answered 503.

Grants the reader the same five subjects the hive-status bucket already
has, for the same measured reasons -- including BOTH `CONSUMER.CREATE`
forms. `keys()` builds an ephemeral consumer whose create subject carries
no consumer name, and `>` matches one or more tokens, never zero, so the
`.>` form alone does not cover it; that omission surfaces as a client
timeout and an operator-visible 503, which is the shape reported here.

Deliberately grants NO `$KV.agent-status.*` subject, to anyone. That is
the write side, and writing is what picks a key layout -- whether agent
status is published by the agent or forwarded by its hive, and under
which key, is still open. A write grant here would answer that by
implication. Reading needs none of it: a KV read is a DIRECT.GET, and the
`.>` form is bucket-wide rather than per-key, so nothing added here
encodes a layout. A test asserts that, for the reader and for a hive,
with a control proving the same matcher fires on a bucket that does have
write grants.

The stream name is derived from the crate constant, like the wanted-state
bucket beside it, so writer and reader cannot disagree about it.

swarm-nats-auth/src/policy.rs

@@ -148,6 +148,14 @@ impl Policy {
         format!("KV_{}", swarm_queue_client::wanted::BUCKET)
     }
 
+    /// 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.
     ///
@@ -324,6 +332,26 @@ impl Policy {
             format!("$JS.API.STREAM.CREATE.{}", Self::wanted_stream()),
             format!("$KV.{}.>", swarm_queue_client::wanted::BUCKET),
         ]);
+        // 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 writing is what picks a key layout. Who publishes agent
+        // status, and under which key, is still open — so granting a write
+        // subject here would answer a question this change has no business
+        // answering. Reading needs none of it: a KV read is a `DIRECT.GET`,
+        // and the `.>` form is bucket-wide rather than per-key, so nothing
+        // below encodes a layout.
+        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
     }
 }
@@ -342,6 +370,49 @@ mod tests {
         .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_grant_encodes_an_agent_status_key_layout() {
+        // Who writes agent status, and under which key, is undecided. A `$KV.`
+        // subject is the write side and would answer that by implication, so
+        // NOBODY gets one for this bucket until the layout is settled.
+        let bucket = swarm_queue_client::agent_status::BUCKET;
+        let prefix = format!("$KV.{bucket}.");
+        let p = policy();
+        for (who, subjects) in [
+            ("reader", p.reader_subjects()),
+            ("hive", p.hive_subjects("alpha")),
+        ] {
+            assert!(
+                !subjects.iter().any(|s| s.starts_with(&prefix)),
+                "{who} was granted a {prefix}* subject, which picks a key layout"
+            );
+        }
+        // Control: the prefix test does fire on a bucket that legitimately has
+        // write grants, so a pass above is about agent-status, not the matcher.
+        let wanted = format!("$KV.{}.", swarm_queue_client::wanted::BUCKET);
+        assert!(p.reader_subjects().iter().any(|s| s.starts_with(&wanted)));
+    }
+
     #[test]
     fn an_extra_subject_without_the_placeholder_is_refused() {
         // 🩸 The option exists to put a second stream inside ONE hive's