Design 47's last milestones: the keeper, self-check, healers, hand-acts, provisioner, vault, artifact sweep, disk readings, credential forms (issues 774, 1952, 1953, 1954, 1955)

One commit with exactly the tree of the reviewed head 35b4509c0a, on main 12c2574c9d, so two compiled programs added and removed on the branch leave no trace in history (issue 1577).

modules/nats/cmd/mesh-nats-snapshot/certificate.go

@@ -0,0 +1,174 @@
+package main
+
+import (
+	"crypto/rand"
+	"crypto/rsa"
+	"crypto/x509"
+	"crypto/x509/pkix"
+	"encoding/pem"
+	"errors"
+	"fmt"
+	"math/big"
+	"net"
+	"os"
+	"path/filepath"
+	"time"
+)
+
+// The bus's own certificate, made by the mesh rather than borrowed from an image. Moved here from mesh-controller's
+// `broker certificate` with the same output (design 47's L3 s11): the command belongs to the bus module, and the
+// controller's binary is retired after its lease is handed away (ADR 0547 §11). Nothing changes at the lease write: the
+// certificate in /var/lib/mesh-broker-tls stays, and every pin stays valid.
+//
+// **The foundation asked a third-party image for a tool it never said must be there** (novox/hq
+// 04-ISSUES/146). The bootstrap made this certificate by running `openssl` inside the broker's
+// image, which worked while the broker was one that happened to carry it and stopped the day the
+// bus changed: the new one has a shell and no openssl, so the step exited 127 and no mesh could be
+// raised. Substituting another image the bundle names does not help — none of them carry it
+// either.
+//
+// So the program that needs a certificate makes one. It is the mesh's own binary, already on the
+// machine at this point in the bootstrap (the schema step ran it), and it needs nothing from the
+// image it writes into but a mounted directory.
+//
+// **Self-signed, and that is the design** — a host pins this server's exact certificate and
+// authenticates with a password (novox/hq ADR 0004). There is no authority above it to ask, and at
+// this moment in a bootstrap there is no mesh to ask one of.
+//
+// Idempotent, because the step is applied again on every reconcile and a second certificate would
+// be one the hosts that pinned the first no longer believe.
+
+// busCertificateNames is what the bus is reached by: the container name on a mesh network, and the
+// loopback address the machine's own foundation dials.
+var busCertificateNames = []string{"mesh-broker"}
+
+const busCertificateLife = 10 * 365 * 24 * time.Hour
+
+// busCertificate makes the bus's certificate in a directory, or says whether one is there.
+//
+//	mesh-nats-snapshot certificate --into /tls           make it if it is not there
+//	mesh-nats-snapshot certificate --check --into /tls   exit non-zero unless a usable pair is
+func busCertificate(args []string) error {
+	into, check := "", false
+	for i := 0; i < len(args); i++ {
+		switch args[i] {
+		case "--check":
+			check = true
+		case "--into":
+			if i+1 >= len(args) {
+				return errors.New("--into needs a directory")
+			}
+			into = args[i+1]
+			i++
+		default:
+			return fmt.Errorf("certificate [--check] --into <directory>: %q", args[i])
+		}
+	}
+	if into == "" {
+		return errors.New("certificate [--check] --into <directory>")
+	}
+	crt, key := filepath.Join(into, "tls.crt"), filepath.Join(into, "tls.key")
+
+	if usable, err := busCertificateUsable(crt, key); err != nil {
+		return err
+	} else if usable {
+		fmt.Printf("the bus already has a certificate at %s, and it was left alone\n", crt)
+		return nil
+	}
+	if check {
+		// Said as a failure, because that is what the caller asked: a bootstrap's verify runs
+		// this and a false answer is what makes the step run.
+		return fmt.Errorf("no usable certificate and key at %s", into)
+	}
+	return writeBusCertificate(crt, key)
+}
+
+// busCertificateUsable says whether a certificate and its key are both there and parse.
+//
+// Both, and parsed rather than stat'ed: a half-written pair is the state a bootstrap interrupted
+// between the two files leaves behind, and a step that treated it as done would hand the server a
+// certificate with no key and report success.
+func busCertificateUsable(crt, key string) (bool, error) {
+	certPEM, err := os.ReadFile(crt)
+	if errors.Is(err, os.ErrNotExist) {
+		return false, nil
+	}
+	if err != nil {
+		return false, err
+	}
+	keyPEM, err := os.ReadFile(key)
+	if errors.Is(err, os.ErrNotExist) {
+		return false, nil
+	}
+	if err != nil {
+		return false, err
+	}
+	if _, err := tlsPairParses(certPEM, keyPEM); err != nil {
+		return false, nil
+	}
+	return true, nil
+}
+
+func tlsPairParses(certPEM, keyPEM []byte) (*x509.Certificate, error) {
+	block, _ := pem.Decode(certPEM)
+	if block == nil || block.Type != "CERTIFICATE" {
+		return nil, errors.New("not a certificate")
+	}
+	certificate, err := x509.ParseCertificate(block.Bytes)
+	if err != nil {
+		return nil, err
+	}
+	keyBlock, _ := pem.Decode(keyPEM)
+	if keyBlock == nil {
+		return nil, errors.New("not a key")
+	}
+	if _, err := x509.ParsePKCS8PrivateKey(keyBlock.Bytes); err != nil {
+		if _, err := x509.ParsePKCS1PrivateKey(keyBlock.Bytes); err != nil {
+			return nil, err
+		}
+	}
+	return certificate, nil
+}
+
+func writeBusCertificate(crt, key string) error {
+	private, err := rsa.GenerateKey(rand.Reader, 2048)
+	if err != nil {
+		return err
+	}
+	serial, err := rand.Int(rand.Reader, new(big.Int).Lsh(big.NewInt(1), 128))
+	if err != nil {
+		return err
+	}
+	template := &x509.Certificate{
+		SerialNumber:          serial,
+		Subject:               pkix.Name{CommonName: busCertificateNames[0]},
+		DNSNames:              busCertificateNames,
+		IPAddresses:           []net.IP{net.ParseIP("127.0.0.1")},
+		NotBefore:             time.Now().Add(-time.Hour),
+		NotAfter:              time.Now().Add(busCertificateLife),
+		KeyUsage:              x509.KeyUsageDigitalSignature | x509.KeyUsageKeyEncipherment,
+		ExtKeyUsage:           []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
+		BasicConstraintsValid: true,
+	}
+	der, err := x509.CreateCertificate(rand.Reader, template, template, &private.PublicKey, private)
+	if err != nil {
+		return err
+	}
+	pkcs8, err := x509.MarshalPKCS8PrivateKey(private)
+	if err != nil {
+		return err
+	}
+
+	// **The key first, and only then the certificate**, so the pair a reader finds is never a
+	// certificate whose key has not been written yet — the one order in which an interruption
+	// leaves something that looks finished (novox/hq 04-ISSUES/014, a key present and unusable).
+	if err := os.WriteFile(key, pem.EncodeToMemory(&pem.Block{Type: "PRIVATE KEY", Bytes: pkcs8}), 0o600); err != nil {
+		return err
+	}
+	if err := os.WriteFile(crt, pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}), 0o644); err != nil {
+		return err
+	}
+	fmt.Printf("made the bus a certificate for %v, valid until %s\n  %s\n  %s\n",
+		busCertificateNames, template.NotAfter.Format(time.RFC3339), crt, key)
+	return nil
+}

modules/nats/cmd/mesh-nats-snapshot/certificate_test.go

@@ -0,0 +1,138 @@
+package main
+
+import (
+	"crypto/tls"
+	"crypto/x509"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+)
+
+// L3 s11 (design 47): the bus's certificate command, moved from mesh-controller with the same output, made once, and
+// its check false without a usable pair. Its cases, as the controller's tests had them, are rows of one test.
+func TestTheBusCertificate(t *testing.T) {
+	for _, row := range []struct {
+		name string
+		run  func(*testing.T)
+	}{
+		{"it loads as a server loads it", loadsAsAServersWould},
+		{"it is made once", isMadeOnce},
+		{"the check is false until there is a pair", checkIsFalseUntilAPair},
+		{"a certificate without its key is not usable", withoutItsKeyIsNotUsable},
+		{"where to write is required", whereToWriteIsRequired},
+	} {
+		t.Run(row.name, row.run)
+	}
+}
+
+// novox/hq 04-ISSUES/146. The bootstrap could not make the bus a certificate: it asked an image for
+// `openssl` and the image it asks has none. What replaces it is this command, so what is checked is
+// what the bootstrap needs from it — a pair a TLS server can actually load, made once and only once.
+
+func loadsAsAServersWould(t *testing.T) {
+	into := t.TempDir()
+	if err := busCertificate([]string{"--into", into}); err != nil {
+		t.Fatalf("the bus could not be given a certificate: %v", err)
+	}
+
+	// The check a cheaper test would not make. The key was present and valid and the server could
+	// not start, once, because nothing loaded the pair the way a server loads it
+	// (novox/hq 04-ISSUES/014).
+	pair, err := tls.LoadX509KeyPair(filepath.Join(into, "tls.crt"), filepath.Join(into, "tls.key"))
+	if err != nil {
+		t.Fatalf("a TLS server cannot load what was written: %v", err)
+	}
+	leaf := pair.Leaf
+	if leaf == nil {
+		if leaf, err = x509.ParseCertificate(pair.Certificate[0]); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if err := leaf.VerifyHostname("mesh-broker"); err != nil {
+		t.Errorf("the certificate is not for the name the bus is reached by: %v", err)
+	}
+	if len(leaf.IPAddresses) == 0 || leaf.IPAddresses[0].String() != "127.0.0.1" {
+		t.Errorf("the certificate does not cover the loopback address the foundation dials: %v", leaf.IPAddresses)
+	}
+
+	// The key is not readable by anything else on the machine; the certificate is public and is.
+	key, err := os.Stat(filepath.Join(into, "tls.key"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if key.Mode().Perm() != 0o600 {
+		t.Errorf("the key is %v", key.Mode().Perm())
+	}
+	crt, err := os.Stat(filepath.Join(into, "tls.crt"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if crt.Mode().Perm() != 0o644 {
+		t.Errorf("the certificate is %v, which the server runs as another user cannot read", crt.Mode().Perm())
+	}
+}
+
+// **Made once.** The step is applied again on every reconcile, and a second certificate is one the
+// hosts that pinned the first no longer believe (novox/hq ADR 0004).
+func isMadeOnce(t *testing.T) {
+	into := t.TempDir()
+	if err := busCertificate([]string{"--into", into}); err != nil {
+		t.Fatal(err)
+	}
+	first, err := os.ReadFile(filepath.Join(into, "tls.crt"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := busCertificate([]string{"--into", into}); err != nil {
+		t.Fatal(err)
+	}
+	again, err := os.ReadFile(filepath.Join(into, "tls.crt"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if string(first) != string(again) {
+		t.Fatal("running it twice replaced the certificate every host had pinned")
+	}
+}
+
+// The verify half: false before, true after, which is what makes the bootstrap run the step at all.
+func checkIsFalseUntilAPair(t *testing.T) {
+	into := t.TempDir()
+	if err := busCertificate([]string{"--check", "--into", into}); err == nil {
+		t.Fatal("an empty directory reported a usable certificate")
+	}
+	if err := busCertificate([]string{"--into", into}); err != nil {
+		t.Fatal(err)
+	}
+	if err := busCertificate([]string{"--check", "--into", into}); err != nil {
+		t.Fatalf("the certificate it just made does not satisfy its own check: %v", err)
+	}
+}
+
+// A half-written pair is not a pair. An interrupted bootstrap leaves exactly this, and a step that
+// called it done would hand the server a certificate with no key and report success.
+func withoutItsKeyIsNotUsable(t *testing.T) {
+	into := t.TempDir()
+	if err := busCertificate([]string{"--into", into}); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.Remove(filepath.Join(into, "tls.key")); err != nil {
+		t.Fatal(err)
+	}
+	if err := busCertificate([]string{"--check", "--into", into}); err == nil {
+		t.Fatal("a certificate with no key passed the check")
+	}
+	if err := busCertificate([]string{"--into", into}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := tls.LoadX509KeyPair(filepath.Join(into, "tls.crt"), filepath.Join(into, "tls.key")); err != nil {
+		t.Fatalf("it did not replace the unusable pair: %v", err)
+	}
+}
+
+func whereToWriteIsRequired(t *testing.T) {
+	if err := busCertificate(nil); err == nil || !strings.Contains(err.Error(), "--into") {
+		t.Fatalf("it did not ask where to write: %v", err)
+	}
+}

modules/nats/cmd/mesh-nats-snapshot/connect.go

@@ -1,105 +1,16 @@
 package main
 
 import (
-	"crypto/sha256"
-	"crypto/tls"
-	"crypto/x509"
-	"encoding/hex"
-	"encoding/json"
-	"errors"
-	"fmt"
-	"io"
-	"os"
-	"strings"
-	"time"
+	"nats-tools/internal/busconn"
 
 	"github.com/nats-io/nats.go"
 )
 
-// Credential is what the mesh seals to a module so it can reach the bus (mesh-controller
-// cmd/mesh-controller issueWith): where, the server's certificate by fingerprint, who, and the
-// password. The address is not used here — this runs beside the server, and the server is reached on
-// loopback — but the fingerprint is: loopback or not, the certificate is the mesh's own.
-type Credential struct {
-	URL         string `json:"url"`
-	Fingerprint string `json:"fingerprint,omitempty"`
-	User        string `json:"user"`
-	Password    string `json:"password"`
-}
+// Credential is what the mesh seals to a module so it can reach the bus; see busconn.
+type Credential = busconn.Credential
 
-func readCredential(path string) (Credential, error) {
-	var raw []byte
-	var err error
-	if path == "-" {
-		raw, err = io.ReadAll(io.LimitReader(os.Stdin, 1<<20))
-	} else {
-		raw, err = os.ReadFile(path)
-	}
-	if err != nil {
-		return Credential{}, fmt.Errorf("cannot read the credential: %w", err)
-	}
-	var c Credential
-	if err := json.Unmarshal(raw, &c); err != nil {
-		return Credential{}, errors.New("the credential is not the JSON the mesh seals to a module (url, fingerprint, user, password)")
-	}
-	if c.User == "" || c.Password == "" {
-		return Credential{}, errors.New("the credential names no user or password, so the bus would refuse it")
-	}
-	return c, nil
-}
+func readCredential(path string) (Credential, error) { return busconn.Read(path) }
 
-// connect dials the bus as the credential's user. A `tls://` server is held to the credential's
-// fingerprint and nothing else — the mesh's bus presents a certificate of its own, in no trust store,
-// so a name check could only fail or be skipped. A plain `nats://` server is accepted only on
-// loopback, and is what a restore's own private server is.
 func connect(server string, c Credential) (*nats.Conn, error) {
-	opts := []nats.Option{
-		nats.UserInfo(c.User, c.Password),
-		nats.Name("mesh-nats-snapshot"),
-		// Answers come to the user's own inbox and no other: the bus grants every user its own
-		// `_INBOX.<user>.>` and nothing wider (novox/hq design 25 §4).
-		nats.CustomInboxPrefix("_INBOX." + c.User),
-		nats.Timeout(10 * time.Second),
-		// One attempt: a snapshot that cannot reach the bus fails the night, loudly, rather than
-		// waiting for it.
-		nats.NoReconnect(),
-	}
-	switch {
-	case strings.HasPrefix(server, "tls://"):
-		if c.Fingerprint == "" {
-			return nil, errors.New("the credential carries no fingerprint, so the bus's certificate cannot be checked")
-		}
-		opts = append(opts, nats.Secure(pinnedTo(c.Fingerprint)))
-	case strings.HasPrefix(server, "nats://127.0.0.1:"), strings.HasPrefix(server, "nats://localhost:"):
-		if c.Fingerprint != "" {
-			// The mesh's bus requires TLS; a plain connection to it would be refused anyway.
-			opts = append(opts, nats.Secure(pinnedTo(c.Fingerprint)))
-		}
-	default:
-		return nil, fmt.Errorf("%s: the bus is reached over tls:// (or plain nats:// on loopback, for a restore's own server)", server)
-	}
-	nc, err := nats.Connect(server, opts...)
-	if err != nil {
-		return nil, fmt.Errorf("cannot reach the bus at %s as %s: %w", server, c.User, err)
-	}
-	return nc, nil
-}
-
-// pinnedTo accepts exactly the certificate with this SHA-256 — the same pin every machine holds
-// (mesh-controller internal/broker PinnedToFingerprint).
-func pinnedTo(want string) *tls.Config {
-	return &tls.Config{
-		InsecureSkipVerify: true, //nolint:gosec // replaced by the pin below, which is stricter
-		MinVersion:         tls.VersionTLS12,
-		VerifyPeerCertificate: func(rawCerts [][]byte, _ [][]*x509.Certificate) error {
-			if len(rawCerts) == 0 {
-				return errors.New("the bus presented no certificate")
-			}
-			sum := sha256.Sum256(rawCerts[0])
-			if got := "sha256:" + hex.EncodeToString(sum[:]); got != want {
-				return fmt.Errorf("the bus presented a certificate this mesh does not know (%s)", got)
-			}
-			return nil
-		},
-	}
+	return busconn.Connect(server, c, "mesh-nats-snapshot")
 }

modules/nats/cmd/mesh-nats-snapshot/main.go

@@ -17,6 +17,7 @@
 //	mesh-nats-snapshot verify    [--from FILE]
 //	mesh-nats-snapshot restore   --into DIR [--account MESH] [--from FILE]
 //	mesh-nats-snapshot restore   --server URL --credential FILE [--from FILE]
+//	mesh-nats-snapshot certificate [--check] --into DIR
 //
 // **Read-only on the live bus.** A snapshot neither pauses nor reconfigures a stream; the server
 // keeps accepting writes while it streams the copy out. The credential it uses is the bus's own
@@ -47,6 +48,8 @@ func main() {
 		err = runVerify(os.Args[2:])
 	case "restore":
 		err = runRestore(os.Args[2:])
+	case "certificate":
+		err = busCertificate(os.Args[2:])
 	case "-h", "--help", "help":
 		usage()
 		return
@@ -67,6 +70,7 @@ func usage() {
   verify    [--from bus.tar]                                       every archive against the manifest
   restore   --into DIR [--account MESH] [--from bus.tar]           into a new store directory, by a server of its own
   restore   --server URL --credential FILE [--from bus.tar]        into a running server that holds none of them
+  certificate [--check] --into DIR                                 the bus's self-signed certificate, made once
 
 The credential is the JSON the mesh seals to the bus's own module: url, fingerprint, user, password.
 `)

modules/nats/cmd/nats-tools/bus.go

@@ -0,0 +1,41 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+	"github.com/nats-io/nats.go/jetstream"
+	"nats-tools/internal/busconn"
+)
+
+// LiveBus is the bus as the module's own credential reaches it. Nothing connects until a tool that changes the
+// bus is called, so the tools that read (through the monitoring endpoint) answer as they did with no credential.
+// NATS_TOOLS_CREDENTIAL names the credential (a word of the bundle's own: the runtime strips MESH_BROKER_FILE) (the process nats-tools's own secret `broker-nats-tools`); MESH_NATS_SERVER where the bus is as
+// seen from this machine (default: its own port on loopback).
+func LiveBus() Bus {
+	return func(ctx context.Context) (jetstream.JetStream, func(), error) {
+		file := env("NATS_TOOLS_CREDENTIAL", "")
+		if file == "" {
+			return nil, nil, errors.New("NATS_TOOLS_CREDENTIAL is not set, so this tool has no credential to reach the bus with (the module's manifest sets it)")
+		}
+		c, err := busconn.Read(file)
+		if err != nil {
+			return nil, nil, fmt.Errorf("the credential at %s (NATS_TOOLS_CREDENTIAL) cannot be read: %w", file, err)
+		}
+		nc, err := busconn.Connect(env("MESH_NATS_SERVER", "tls://127.0.0.1:4222"), c, "nats-tools")
+		if err != nil {
+			return nil, nil, err
+		}
+		js, err := jetstream.New(nc)
+		if err != nil {
+			nc.Close()
+			return nil, nil, fmt.Errorf("JetStream on the bus: %w", err)
+		}
+		return js, nc.Close, nil
+	}
+}
+
+// SaidByTheRuntime is the audit line: an event of this module, published through the runtime that launched it.
+func SaidByTheRuntime(event string, body any) error { return stdio.Emit(event, body) }

modules/nats/cmd/nats-tools/main.go

@@ -1,6 +1,7 @@
 // nats-tools: the mesh bus's own tools (novox/hq design 25), a Go bundle the node's runtime launches on the
-// machine that runs the bus. Everything here reads — the server's monitoring API and the user list the
-// controller composed — and nothing changes the bus. stdout is the MCP channel; this says nothing else.
+// machine that runs the bus. Almost everything here reads — the server's monitoring API and the user list the
+// controller composed. The two that change the bus (removal.go) are the only ones that connect to it, on the
+// module's own credential, and say what they did. stdout is the MCP channel; this says nothing else.
 package main
 
 import (
@@ -86,7 +87,7 @@ func tools(monitor Monitor, users func(context.Context) ([]byte, error), written
 				return Subscriptions(subsz, connz, arg(a, "subject"), limitOf(a, 100))
 			}},
 		{Name: "nats_streams",
-			Description: "JetStream: every stream with its subjects, retention, message count and size — or, given one stream, every consumer on it with what it has pending, unacknowledged and redelivered.",
+			Description: "JetStream: every stream with its subjects, retention, message count, size and size cap (max_bytes, when it has one) — or, given one stream, every consumer on it with what it has pending, unacknowledged and redelivered.",
 			Input:       map[string]any{"stream": str("one stream, e.g. EVENTS, CONTROL, ASSIGNMENTS, SEAT_NODE_BUILD_AGENT or KV_<bucket>")},
 			Run: func(a map[string]any) (any, error) {
 				doc, err := get(jsz)
@@ -151,7 +152,7 @@ func tools(monitor Monitor, users func(context.Context) ([]byte, error), written
 }
 
 func main() {
-	if err := stdio.Serve("", tools(LiveMonitor(), ReadUsers, UsersWritten)); err != nil {
+	if err := stdio.Serve("", append(tools(LiveMonitor(), ReadUsers, UsersWritten), removalTools(LiveBus(), SaidByTheRuntime)...)); err != nil {
 		fmt.Fprintln(os.Stderr, err)
 		os.Exit(1)
 	}

modules/nats/cmd/nats-tools/manifest_test.go

@@ -3,6 +3,7 @@ package main
 import (
 	"encoding/json"
 	"os"
+	"slices"
 	"strings"
 	"testing"
 )
@@ -24,11 +25,14 @@ func TestTheBussManifestMountsItsProgramsIntoTheUpstreamImage(t *testing.T) {
 				} `json:"backup"`
 			} `json:"own"`
 		} `json:"data"`
+		Tools     []string         `json:"tools"`
+		Emits     []string         `json:"emits"`
 		Resources []map[string]any `json:"resources"`
 		Build     struct {
 			On        any `json:"on"`
 			Artifacts []struct {
 				Name, Kind, From, Language, System, Binary string
+				Env                                        map[string]string
 			} `json:"artifacts"`
 		} `json:"build"`
 	}
@@ -38,7 +42,10 @@ func TestTheBussManifestMountsItsProgramsIntoTheUpstreamImage(t *testing.T) {
 	if m.Build.On != nil {
 		t.Error("build.on is declared: nothing is built on a base image any more")
 	}
-	artifacts := map[string]struct{ Name, Kind, From, Language, System, Binary string }{}
+	artifacts := map[string]struct {
+		Name, Kind, From, Language, System, Binary string
+		Env                                        map[string]string
+	}{}
 	for _, a := range m.Build.Artifacts {
 		artifacts[a.Name] = a
 		if a.Kind == "image" {
@@ -103,6 +110,23 @@ func TestTheBussManifestMountsItsProgramsIntoTheUpstreamImage(t *testing.T) {
 			t.Errorf("%s is not in the container's restart-on: a new program would leave the old one running", p.resource)
 		}
 	}
+	// The tools that change the bus (issue 1952 N1-N3): listed, their audit events declared, and the credential
+	// they dial with the module's own secret `broker`.
+	for _, want := range []string{"nats_consumer_remove", "nats_bucket_remove"} {
+		if !slices.Contains(m.Tools, want) {
+			t.Errorf("tools do not list %s", want)
+		}
+	}
+	for _, want := range []string{"consumer-removed", "bucket-keys-removed"} {
+		if !slices.Contains(m.Emits, want) {
+			t.Errorf("emits do not list %s: the removal tools say it as the audit line", want)
+		}
+	}
+	for _, a := range m.Build.Artifacts {
+		if a.Name == "tools" && a.Env["NATS_TOOLS_CREDENTIAL"] != "${dir:tools-secret}/broker-nats-tools" {
+			t.Errorf("the tools' credential is %q: the process nats-tools' own secret broker-nats-tools, under a word of the bundle's own (the runtime strips MESH_BROKER_FILE)", a.Env["NATS_TOOLS_CREDENTIAL"])
+		}
+	}
 	args, _ := server["args"].([]any)
 	if len(args) != 2 || args[0] != "sh" || args[1] != "/opt/mesh-entrypoint/entrypoint.sh" {
 		t.Errorf("args are %v: the entrypoint run through sh, so no execute bit is relied on", args)

modules/nats/cmd/nats-tools/removal.go

@@ -0,0 +1,246 @@
+package main
+
+// The two tools of this module that change the bus (novox/hq ADR 0564 clause 5, issue 1952 notes 6, 7 and 9):
+// remove one named consumer, and remove one node's keys under a prefix of a key-value bucket. Everything else
+// here reads.
+//
+// **Narrow by what the tool needs, and narrower by what the bus grants.** The tools reach the bus on the
+// credential the module's manifest names (NATS_TOOLS_CREDENTIAL); the bus refuses whatever that credential is not
+// granted, so a wrong stream, consumer or bucket is refused by the server and said here, not trusted to the
+// arguments. Each tool refuses without `why`, reads what it would remove first (and says it, so a person can ask
+// with dry_run and then for real), removes, reads that it is gone, and says what it did as an event of the module
+// (the audit trail: who asked is in the call the controller audits, why and what are in the event).
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"regexp"
+	"sort"
+	"strings"
+	"time"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// Bus opens JetStream on the bus as the module's credential for one call; close ends the connection.
+type Bus func(ctx context.Context) (js jetstream.JetStream, close func(), err error)
+
+// Said keeps the audit line: an event of this module.
+type Said func(event string, body any) error
+
+const (
+	consumerRemoved   = "consumer-removed"
+	bucketKeysRemoved = "bucket-keys-removed"
+	// maxKeysShown bounds the keys an answer lists; the count is always whole.
+	maxKeysShown = 200
+)
+
+var (
+	// A stream, consumer or bucket is one subject token: no dot, no wildcard, no space.
+	name = regexp.MustCompile(`^[A-Za-z0-9_-]+$`)
+	// A key prefix is subject tokens and their dots; no wildcard, so it names keys and not a pattern.
+	keyPrefix = regexp.MustCompile(`^[A-Za-z0-9_./=-]+$`)
+)
+
+func why(a map[string]any) (string, error) {
+	w := strings.TrimSpace(arg(a, "why"))
+	if w == "" {
+		return "", errors.New("why is required: say why this is removed, for the audit trail (novox/hq ADR 0564 clause 5); nothing was removed")
+	}
+	return w, nil
+}
+
+// answerWithin is how long the bus is given to answer one request. **The bus does not refuse a request its
+// credential is not granted: it drops it** (a permissions violation is told to the connection's error handler,
+// never answered), so an unanswered request is how a refusal shows here.
+var answerWithin = 10 * time.Second
+
+// unanswered says what a request with no answer most likely is.
+func unanswered(err error, what string) error {
+	if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, nats.ErrTimeout) {
+		return fmt.Errorf("%s: the bus did not answer within %s; it drops, without an answer, what the module's credential is not granted, "+
+			"so this is most likely outside its grant (or the bus is not reachable): %w", what, answerWithin, err)
+	}
+	return fmt.Errorf("%s: %w", what, err)
+}
+
+func step(ctx context.Context) (context.Context, context.CancelFunc) {
+	return context.WithTimeout(ctx, answerWithin)
+}
+
+// ConsumerRemoved is what nats_consumer_remove answers and says.
+type ConsumerRemoved struct {
+	Stream   string         `json:"stream"`
+	Consumer string         `json:"consumer"`
+	Why      string         `json:"why"`
+	Was      map[string]any `json:"was"`
+	Removed  bool           `json:"removed"`
+	At       time.Time      `json:"at"`
+}
+
+// RemoveConsumer removes one named consumer of one stream, after reading it, and says so.
+func RemoveConsumer(ctx context.Context, bus Bus, said Said, a map[string]any) (any, error) {
+	w, err := why(a)
+	if err != nil {
+		return nil, err
+	}
+	stream, consumer := arg(a, "stream"), arg(a, "consumer")
+	if !name.MatchString(stream) || !name.MatchString(consumer) {
+		return nil, errors.New("stream and consumer are required, each one name (letters, digits, - and _): nothing was removed")
+	}
+	js, closeBus, err := bus(ctx)
+	if err != nil {
+		return nil, err
+	}
+	defer closeBus()
+	sctx, cancel := step(ctx)
+	c, err := js.Consumer(sctx, stream, consumer)
+	cancel()
+	if err != nil {
+		if errors.Is(err, jetstream.ErrConsumerNotFound) || errors.Is(err, jetstream.ErrStreamNotFound) {
+			return nil, fmt.Errorf("%s has no consumer %s (%v): nothing was removed", stream, consumer, err)
+		}
+		return nil, unanswered(err, fmt.Sprintf("the consumer %s of %s could not be read, so it was not removed", consumer, stream))
+	}
+	info := c.CachedInfo()
+	was := map[string]any{
+		"durable": info.Config.Durable, "filter_subject": info.Config.FilterSubject, "filter_subjects": info.Config.FilterSubjects,
+		"deliver_subject": info.Config.DeliverSubject, "pending": info.NumPending, "ack_pending": info.NumAckPending,
+		"redelivered": info.NumRedelivered, "created": info.Created,
+	}
+	sctx, cancel = step(ctx)
+	err = js.DeleteConsumer(sctx, stream, consumer)
+	cancel()
+	if err != nil {
+		return nil, unanswered(err, fmt.Sprintf("consumer %s of %s was not removed", consumer, stream))
+	}
+	sctx, cancel = step(ctx)
+	_, err = js.Consumer(sctx, stream, consumer)
+	cancel()
+	if err == nil {
+		return nil, fmt.Errorf("consumer %s of %s is still there after the removal was accepted", consumer, stream)
+	}
+	out := ConsumerRemoved{Stream: stream, Consumer: consumer, Why: w, Was: was, Removed: true, At: time.Now().UTC()}
+	if err := said(consumerRemoved, out); err != nil {
+		return nil, fmt.Errorf("consumer %s of %s WAS removed (%s) but the audit event could not be said: %w", consumer, stream, w, err)
+	}
+	return out, nil
+}
+
+// KeysRemoved is what nats_bucket_remove answers and says.
+type KeysRemoved struct {
+	Bucket  string    `json:"bucket"`
+	Prefix  string    `json:"prefix"`
+	Why     string    `json:"why"`
+	DryRun  bool      `json:"dry_run"`
+	Count   int       `json:"count"`
+	Keys    []string  `json:"keys"`
+	Cut     bool      `json:"keys_cut,omitempty"`
+	Removed int       `json:"removed"`
+	At      time.Time `json:"at"`
+}
+
+// RemoveKeys removes the keys of a key-value bucket under a prefix: counted first, and only those.
+func RemoveKeys(ctx context.Context, bus Bus, said Said, a map[string]any) (any, error) {
+	w, err := why(a)
+	if err != nil {
+		return nil, err
+	}
+	bucket, node, prefix := arg(a, "bucket"), arg(a, "node"), arg(a, "prefix")
+	if !name.MatchString(bucket) {
+		return nil, errors.New("bucket is required, a bucket's name as nats_buckets lists it: nothing was removed")
+	}
+	if (node == "") == (prefix == "") {
+		return nil, errors.New("give exactly one of node (that node's keys: those starting `<node>.`) or prefix (keys starting so): " +
+			"a removal of a whole bucket is not this tool's; nothing was removed")
+	}
+	if node != "" {
+		if !name.MatchString(node) {
+			return nil, errors.New("node is one machine's name: nothing was removed")
+		}
+		prefix = node + "."
+	}
+	if !keyPrefix.MatchString(prefix) {
+		return nil, errors.New("prefix is key text without wildcards or spaces: nothing was removed")
+	}
+	dry, _ := a["dry_run"].(bool)
+	js, closeBus, err := bus(ctx)
+	if err != nil {
+		return nil, err
+	}
+	defer closeBus()
+	sctx, cancel := step(ctx)
+	kv, err := js.KeyValue(sctx, bucket)
+	cancel()
+	if err != nil {
+		return nil, unanswered(err, fmt.Sprintf("bucket %s could not be opened, nothing was removed", bucket))
+	}
+	sctx, cancel = step(ctx)
+	stream, err := js.Stream(sctx, "KV_"+bucket)
+	cancel()
+	if err != nil {
+		return nil, unanswered(err, fmt.Sprintf("bucket %s has no stream to read its keys from", bucket))
+	}
+	// The server narrows to the prefix's tokens when the prefix ends on a token; any other prefix is cut here.
+	filter := "$KV." + bucket + ".>"
+	if strings.HasSuffix(prefix, ".") {
+		filter = "$KV." + bucket + "." + prefix + ">"
+	}
+	sctx, cancel = step(ctx)
+	info, err := stream.Info(sctx, jetstream.WithSubjectFilter(filter))
+	cancel()
+	if err != nil {
+		return nil, unanswered(err, fmt.Sprintf("the keys of %s could not be read, so none were removed", bucket))
+	}
+	var keys []string
+	for subject := range info.State.Subjects {
+		if key := strings.TrimPrefix(subject, "$KV."+bucket+"."); strings.HasPrefix(key, prefix) {
+			keys = append(keys, key)
+		}
+	}
+	sort.Strings(keys)
+	out := KeysRemoved{Bucket: bucket, Prefix: prefix, Why: w, DryRun: dry, Count: len(keys), Keys: keys, At: time.Now().UTC()}
+	if len(keys) > maxKeysShown {
+		out.Keys, out.Cut = keys[:maxKeysShown], true
+	}
+	if want, ok := a["expect"].(float64); ok && int(want) != len(keys) {
+		return nil, fmt.Errorf("%d keys of %s start with %q, not the %d expected: nothing was removed", len(keys), bucket, prefix, int(want))
+	}
+	if dry || len(keys) == 0 {
+		return out, nil
+	}
+	for _, k := range keys {
+		sctx, cancel := step(ctx)
+		err := kv.Purge(sctx, k)
+		cancel()
+		if err != nil {
+			return nil, unanswered(err, fmt.Sprintf("removing %q from %s failed after %d of %d keys, the rest were left", k, bucket, out.Removed, len(keys)))
+		}
+		out.Removed++
+	}
+	if err := said(bucketKeysRemoved, out); err != nil {
+		return nil, fmt.Errorf("%d keys of %s under %q WERE removed (%s) but the audit event could not be said: %w", out.Removed, bucket, prefix, w, err)
+	}
+	return out, nil
+}
+
+// removalTools are the module's two tools that change the bus; every other tool here reads.
+func removalTools(bus Bus, said Said) []stdio.Tool {
+	run := func(f func(context.Context, Bus, Said, map[string]any) (any, error)) func(map[string]any) (any, error) {
+		return func(a map[string]any) (any, error) { return f(context.Background(), bus, said, a) }
+	}
+	return []stdio.Tool{
+		{Name: "nats_consumer_remove",
+			Description: "Removes (deletes) one named JetStream consumer of one stream, e.g. the consumer `controller` of DEAD_LETTER_NOTICES at the lease write (novox/hq ADR 0564 clause 5). Needs `why`, kept in the audit trail; refuses an unknown stream or consumer; reads the consumer first and answers what it was (filters, pending, unacknowledged); the stream and its other consumers are not touched. The bus refuses any consumer the module's credential is not granted.",
+			Input:       map[string]any{"stream": str("the stream, e.g. DEAD_LETTER_NOTICES"), "consumer": str("the consumer's name, e.g. controller"), "why": str("why it is removed, for the audit trail (required)")},
+			Run:         run(RemoveConsumer)},
+		{Name: "nats_bucket_remove",
+			Description: "Removes (purges) keys from a key-value bucket: one node's keys (`node`, the keys starting `<node>.`) or the keys under a `prefix`, e.g. a full acts bucket of the node-engine. Needs `why`, kept in the audit trail; never the whole bucket. Counts the keys first and lists them (`dry_run` only counts; `expect` refuses when the count differs); removes only those keys, and their history. See the buckets and their keys with nats_buckets. The bus refuses any bucket the module's credential is not granted.",
+			Input: map[string]any{"bucket": str("the bucket as nats_buckets lists it, e.g. mesh-host_acts"), "node": str("a node whose keys (`<node>.`) go"), "prefix": str("key prefix instead of node"),
+				"why": str("why they are removed, for the audit trail (required)"), "dry_run": map[string]any{"type": "boolean", "description": "count and list the keys, remove none"}, "expect": num("the count of keys expected; refuses when it differs")},
+			Run: run(RemoveKeys)},
+	}
+}

modules/nats/cmd/nats-tools/removal_test.go

@@ -0,0 +1,173 @@
+//go:build integration
+
+package main
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"path/filepath"
+	"sort"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats-server/v2/server"
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// The two tools that change the bus, against a server shaped like the mesh's, on the credential the module's
+// grant proposal gives them and nothing wider (issue 1952 N2, N3): the bus itself refuses what the grant leaves out,
+// so this is also the proof that the proposed grant is enough and no wider than it needs to be.
+func TestTheRemovalToolsChangeOnlyWhatTheyAreAskedAndTheGrantAllows(t *testing.T) {
+	dir := t.TempDir()
+	conf := filepath.Join(dir, "nats.conf")
+	if err := os.WriteFile(conf, []byte(fmt.Sprintf(`listen: "127.0.0.1:-1"
+jetstream { store_dir: %q }
+accounts { MESH { jetstream: enabled, users: [
+  { user: "admin", password: "admin" },
+  { user: "nats-tools", password: "tools", permissions: {
+      publish: { allow: [
+        "$JS.API.CONSUMER.INFO.DEAD_LETTER_NOTICES.controller",
+        "$JS.API.CONSUMER.DELETE.DEAD_LETTER_NOTICES.controller",
+        "$JS.API.STREAM.INFO.KV_mesh-host_acts",
+        "$KV.mesh-host_acts.>" ] },
+      subscribe: { allow: ["_INBOX.nats-tools.>"] } } }
+] } }
+`, filepath.Join(dir, "js"))), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	opts, err := server.ProcessConfigFile(conf)
+	if err != nil {
+		t.Fatal(err)
+	}
+	opts.NoLog, opts.NoSigs = true, true
+	s, err := server.NewServer(opts)
+	if err != nil {
+		t.Fatal(err)
+	}
+	go s.Start()
+	if !s.ReadyForConnections(10 * time.Second) {
+		t.Fatal("the server did not start")
+	}
+	defer s.Shutdown()
+
+	answerWithin = 2 * time.Second
+	ctx := context.Background()
+	admin, err := nats.Connect(s.ClientURL(), nats.UserInfo("admin", "admin"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer admin.Close()
+	ajs, _ := jetstream.New(admin)
+	if _, err := ajs.CreateStream(ctx, jetstream.StreamConfig{Name: "DEAD_LETTER_NOTICES", Subjects: []string{"dl.>"}, Retention: jetstream.WorkQueuePolicy}); err != nil {
+		t.Fatal(err)
+	}
+	for _, c := range []struct{ name, filter string }{{"controller", "dl.a"}, {"watchman", "dl.b"}} {
+		if _, err := ajs.CreateConsumer(ctx, "DEAD_LETTER_NOTICES", jetstream.ConsumerConfig{Durable: c.name, FilterSubject: c.filter, AckPolicy: jetstream.AckExplicitPolicy}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	acts, err := ajs.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "mesh-host_acts"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	other, _ := ajs.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "other_state"})
+	for _, k := range []string{"g14.1", "g14.2", "g14.sub.3", "g14x.4", "ace.5"} {
+		acts.Put(ctx, k, []byte("v"))
+	}
+	other.Put(ctx, "g14.1", []byte("v"))
+
+	bus := func(context.Context) (jetstream.JetStream, func(), error) {
+		nc, err := nats.Connect(s.ClientURL(), nats.UserInfo("nats-tools", "tools"), nats.CustomInboxPrefix("_INBOX.nats-tools"))
+		if err != nil {
+			return nil, nil, err
+		}
+		js, err := jetstream.New(nc)
+		return js, nc.Close, err
+	}
+	var said []string
+	say := func(event string, body any) error { said = append(said, event); return nil }
+	keys := func(kv jetstream.KeyValue) string {
+		ks, _ := kv.Keys(ctx)
+		sort.Strings(ks)
+		return strings.Join(ks, ",")
+	}
+
+	// N2: the consumer goes, the stream and its other consumer stay; refused without why and for an unknown one.
+	if _, err := RemoveConsumer(ctx, bus, say, map[string]any{"stream": "DEAD_LETTER_NOTICES", "consumer": "controller"}); err == nil || !strings.Contains(err.Error(), "why is required") {
+		t.Fatalf("without why: %v", err)
+	}
+	if _, err := ajs.Consumer(ctx, "DEAD_LETTER_NOTICES", "controller"); err != nil {
+		t.Fatal("refused without why, but the consumer is gone")
+	}
+	// An unknown consumer: the granted name that does not exist is answered "not found"; an ungranted one is dropped.
+	if _, err := RemoveConsumer(ctx, bus, say, map[string]any{"stream": "DEAD_LETTER_NOTICES", "consumer": "nobody", "why": "x"}); err == nil || !strings.Contains(err.Error(), "not grant") && !strings.Contains(err.Error(), "no consumer") {
+		t.Fatalf("an unknown consumer: %v", err)
+	}
+	if _, err := RemoveConsumer(ctx, bus, say, map[string]any{"stream": "DEAD_LETTER_NOTICES", "consumer": "watchman", "why": "x"}); err == nil {
+		t.Fatal("a consumer outside the grant was removed")
+	} else if !strings.Contains(err.Error(), "grant") {
+		t.Fatalf("a consumer outside the grant: the refusal does not say so: %v", err)
+	}
+	if _, err := ajs.Consumer(ctx, "DEAD_LETTER_NOTICES", "watchman"); err != nil {
+		t.Fatalf("the grant did not hold, the watchman's consumer: %v", err)
+	}
+	got, err := RemoveConsumer(ctx, bus, say, map[string]any{"stream": "DEAD_LETTER_NOTICES", "consumer": "controller", "why": "the lease write"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if r := got.(ConsumerRemoved); !r.Removed || r.Was["filter_subject"] != "dl.a" || r.Why != "the lease write" {
+		t.Fatalf("%+v", r)
+	}
+	if _, err := ajs.Consumer(ctx, "DEAD_LETTER_NOTICES", "controller"); err == nil {
+		t.Fatal("the consumer is still there")
+	}
+	if _, err := ajs.Consumer(ctx, "DEAD_LETTER_NOTICES", "watchman"); err != nil {
+		t.Fatalf("the other consumer went too: %v", err)
+	}
+	if _, err := RemoveConsumer(ctx, bus, say, map[string]any{"stream": "DEAD_LETTER_NOTICES", "consumer": "controller", "why": "again"}); err == nil || !strings.Contains(err.Error(), "no consumer") {
+		t.Fatalf("a consumer already gone: %v", err)
+	}
+
+	// N3: only the node's keys go; counted first; refused without why, for a whole bucket, and outside the grant.
+	if _, err := RemoveKeys(ctx, bus, say, map[string]any{"bucket": "mesh-host_acts", "node": "g14"}); err == nil || !strings.Contains(err.Error(), "why is required") {
+		t.Fatalf("without why: %v", err)
+	}
+	if _, err := RemoveKeys(ctx, bus, say, map[string]any{"bucket": "mesh-host_acts", "why": "x"}); err == nil {
+		t.Fatal("a removal naming neither node nor prefix was not refused")
+	}
+	if _, err := RemoveKeys(ctx, bus, say, map[string]any{"bucket": "other_state", "node": "g14", "why": "x"}); err == nil {
+		t.Fatal("a bucket outside the grant was opened")
+	}
+	if k := keys(acts); k != "ace.5,g14.1,g14.2,g14.sub.3,g14x.4" {
+		t.Fatalf("keys after the refusals: %s", k)
+	}
+	dry, err := RemoveKeys(ctx, bus, say, map[string]any{"bucket": "mesh-host_acts", "node": "g14", "why": "look", "dry_run": true})
+	if err != nil || dry.(KeysRemoved).Count != 3 || dry.(KeysRemoved).Removed != 0 {
+		t.Fatalf("dry run: %+v %v", dry, err)
+	}
+	if p, err := RemoveKeys(ctx, bus, say, map[string]any{"bucket": "mesh-host_acts", "prefix": "g14.s", "why": "look", "dry_run": true}); err != nil || p.(KeysRemoved).Count != 1 {
+		t.Fatalf("a prefix that ends inside a token: %+v %v", p, err)
+	}
+	if _, err := RemoveKeys(ctx, bus, say, map[string]any{"bucket": "mesh-host_acts", "node": "g14", "why": "x", "expect": 2.0}); err == nil {
+		t.Fatal("a count that differs from the expected was not refused")
+	}
+	got, err = RemoveKeys(ctx, bus, say, map[string]any{"bucket": "mesh-host_acts", "node": "g14", "why": "the acts bucket is full", "expect": 3.0})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if r := got.(KeysRemoved); r.Removed != 3 || r.Count != 3 {
+		t.Fatalf("%+v", r)
+	}
+	if k := keys(acts); k != "ace.5,g14x.4" {
+		t.Fatalf("keys left: %s", k)
+	}
+	if k := keys(other); k != "g14.1" {
+		t.Fatalf("another bucket changed: %s", k)
+	}
+	if strings.Join(said, ",") != "consumer-removed,bucket-keys-removed" {
+		t.Fatalf("the audit events said: %v", said)
+	}
+}

modules/nats/cmd/nats-tools/tools.go

@@ -351,6 +351,10 @@ func Streams(jsz []byte, stream string) (map[string]any, error) {
 			if s.Config.MaxMsgsPerSubject > 0 {
 				row["max_per_subject"] = s.Config.MaxMsgsPerSubject
 			}
+			// The cap on the stream's size, so how full it is reads against it (-1 and 0 are no cap).
+			if s.Config.MaxBytes > 0 {
+				row["max_bytes"] = s.Config.MaxBytes
+			}
 			if stream != "" {
 				row["description"] = s.Config.Description
 				var consumers []map[string]any

modules/nats/cmd/nats-tools/tools_test.go

@@ -1,12 +1,15 @@
 package main
 
 import (
+	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
 	"strings"
 	"testing"
 	"time"
+
+	"github.com/nats-io/nats.go/jetstream"
 )
 
 const connz = `{"connections":[
@@ -19,7 +22,7 @@ const subsz = `{"num_subscriptions":3,"subscriptions_list":[
  {"subject":"mesh.node.anchor.declare","msgs":1,"cid":1}]}`
 
 const jsz = `{"streams":2,"consumers":2,"messages":12,"bytes":300,"account_details":[{"stream_detail":[
- {"name":"EVENTS","config":{"subjects":["mesh.mod.*.event.>"],"retention":"limits","max_age":604800000000000,"max_msgs_per_subject":10000},
+ {"name":"EVENTS","config":{"subjects":["mesh.mod.*.event.>"],"retention":"limits","max_age":604800000000000,"max_msgs_per_subject":10000,"max_bytes":1048576},
   "state":{"messages":10,"bytes":200,"num_subjects":3,"consumer_count":2,"last_ts":"2026-10-04T10:00:00Z"},
   "consumer_detail":[
    {"name":"anchor_audit","config":{"filter_subject":"mesh.mod.*.event.>"},"num_pending":5,"num_ack_pending":1,"num_redelivered":2,"delivered":{"last_active":"2026-10-04T10:00:00Z"}},
@@ -77,7 +80,7 @@ func TestStreamsBacklogAndBuckets(t *testing.T) {
 	}
 	one, _ := Streams([]byte(jsz), "EVENTS")
 	row := one["detail"].([]map[string]any)[0]
-	if row["max_age_hours"].(int64) != 168 || len(row["consumer_detail"].([]map[string]any)) != 2 {
+	if row["max_age_hours"].(int64) != 168 || row["max_bytes"].(int64) != 1048576 || len(row["consumer_detail"].([]map[string]any)) != 2 {
 		t.Fatalf("%v", row)
 	}
 	if _, err := Streams([]byte(jsz), "NOPE"); err == nil {
@@ -208,3 +211,47 @@ func TestAGrantIsInForceOnlyWhenTheServerReloadedAfterTheListWasWritten(t *testi
 		t.Error("a write time that is not one was read")
 	}
 }
+
+// **The tools that change the bus are two, each refuses before it reaches the bus, and every other tool reads**
+// (issue 1952 N1-N3). A row per refusal: the bus is never opened for them.
+func TestTheRemovalToolsRefuseBeforeTheyReachTheBus(t *testing.T) {
+	bus := func(context.Context) (jetstream.JetStream, func(), error) {
+		t.Fatal("the bus was opened for a call that must be refused")
+		return nil, nil, nil
+	}
+	said := func(string, any) error { t.Fatal("an audit event was said for a refused call"); return nil }
+	for _, c := range []struct {
+		tool string
+		args map[string]any
+		want string
+	}{
+		{"nats_consumer_remove", map[string]any{"stream": "DEAD_LETTER_NOTICES", "consumer": "controller"}, "why is required"},
+		{"nats_consumer_remove", map[string]any{"stream": "DEAD_LETTER_NOTICES", "consumer": "controller", "why": "  "}, "why is required"},
+		{"nats_consumer_remove", map[string]any{"stream": "A.B", "consumer": "c", "why": "x"}, "stream and consumer are required"},
+		{"nats_consumer_remove", map[string]any{"stream": "S", "consumer": ">", "why": "x"}, "stream and consumer are required"},
+		{"nats_bucket_remove", map[string]any{"bucket": "mesh-host_acts", "node": "g14"}, "why is required"},
+		{"nats_bucket_remove", map[string]any{"bucket": "mesh-host_acts", "why": "x"}, "exactly one of node"},
+		{"nats_bucket_remove", map[string]any{"bucket": "mesh-host_acts", "node": "g14", "prefix": "g", "why": "x"}, "exactly one of node"},
+		{"nats_bucket_remove", map[string]any{"bucket": "mesh-host_acts", "prefix": "g*", "why": "x"}, "without wildcards"},
+		{"nats_bucket_remove", map[string]any{"bucket": "", "node": "g14", "why": "x"}, "bucket is required"},
+	} {
+		var run func(map[string]any) (any, error)
+		for _, tl := range removalTools(bus, said) {
+			if tl.Name == c.tool {
+				run = tl.Run
+			}
+		}
+		if run == nil {
+			t.Fatalf("no tool %s", c.tool)
+		}
+		if _, err := run(c.args); err == nil || !strings.Contains(err.Error(), c.want) {
+			t.Errorf("%s %v: %v, want a refusal saying %q", c.tool, c.args, err, c.want)
+		}
+	}
+	// The read tools stay read tools: none of the first ten takes a `why`, and the list is these ten.
+	for _, tl := range tools(nil, nil, nil) {
+		if _, ok := tl.Input["why"]; ok || strings.Contains(tl.Name, "remove") {
+			t.Errorf("%s changes the bus; only the removal tools may", tl.Name)
+		}
+	}
+}