swarm-nats-auth: grant the watch the consumer it needs

A hive's grants for its own `KV_hive-wanted-<hive>` were `STREAM.INFO` +
`DIRECT.GET`, which cover the boot-time read and nothing after it. The
convergence path now opens a KV watch on that bucket, and a watch is a
consumer, so the broker denies it — and the client's `watch()` ends in
`.ok()`, so the denial becomes `None` and the path silently never fires.

The comment four lines above the grant list already argues for this: the
per-hive bucket split exists "so that a watch can be granted without
widening the read". The design was taken and the grant was never written.

Both subject forms, matching the hive-status and agent-status blocks that
grant both for the same documented reason — an ephemeral consumer's
subject carries no name, and `>` never matches zero tokens.

Refs #4006.

swarm-nats-auth/src/policy.rs

@@ -273,6 +273,13 @@ impl Policy {
                 "$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**.
@@ -815,6 +822,30 @@ mod tests {
         );
     }
 
+    #[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.