package buscompose

import (
	"encoding/json"
	"errors"
	"os"
	"path/filepath"
	"reflect"
	"regexp"
	"slices"
	"strings"
	"testing"
	"time"

	"git.novox.be/novox/mesh-sdk/go/announce"
	"git.novox.be/novox/mesh-sdk/go/catalogue"
	"git.novox.be/novox/mesh-sdk/go/delivery"
	"git.novox.be/novox/mesh-sdk/go/provisions"
	"github.com/nats-io/nats.go/jetstream"
)

// Defends issue 1377 and novox/hq ADR 0547 §5: this is the controller's bus composer copied, and it composes byte for
// byte what the controller's composes. testdata/controller-user-lists.json was written by the controller's own
// internal/broker (mesh-controller branch fix/1945-mesh-grants-publishes-memberships, and fix/1790-grants-create-buckets for the grants part creating declared streams), lists and memberships, from fixtures_test.go compiled there unchanged; composed.conf is the
// controller's golden of the whole server configuration. A row that differs is the copy drifting from the controller,
// which is a fault while both compose (until the controller loses its lease, ADR 0547 §13), but for the grants the copy
// gives beyond it on purpose (beyondTheControllers): the copy grants every user everything the controller's grants, and
// only those more.
// createdAdvisory is one subscription to a bucket's creation in a composed list, as stateCreatedHeard composes it.
var createdAdvisory = regexp.MustCompile(`"\$JS\.EVENT\.ADVISORY\.STREAM\.CREATED\.KV_[^"]*", `)

// beyondTheControllers is what the copy grants beyond the controller's on purpose (novox/hq ADR 0547 §2, ADR 0555 §3,
// issue 1126), by user: a node's account reads every node's own facts, `<node>.node` of the have bucket where the node's
// record rides, and what it composes its own modules from (NodeComposeReads). After the take-over (ADR 0547 §8) the
// grants part's list is the one the bus loads; a hand-back loads the controller's, which lacks these reads, and the nodes
// keep applying without composing their own until the next take-over. Each principal's hearing of its buckets' creation
// (issue 1790) is the other row beyond, matched by createdAdvisory.
//
// And the watchman's dead letters and watchdogs (novox/hq ADR 0564 clauses 4, 5, issue 1952 S5): its own consumer on
// DEAD_LETTER_NOTICES, a read of DEAD_LETTERS, and the words its watchdogs hear, the controller's jobs until the lease
// write.
func beyondTheControllers(user string) map[string]bool {
	out := map[string]bool{}
	if strings.HasSuffix(user, "."+WatchmanModule) {
		for _, s := range append(WatchmanDeadLetterGrants(nil), WatchmanHears()...) {
			out[s] = true
		}
	}
	node, ok := strings.CutPrefix(user, "node.")
	if !ok {
		return out
	}
	for _, s := range append(append(readGrants("node-engine_have", "*.node"), NodeComposeReads(node)...), NodeComposeAdvisories()...) {
		out[s] = true
	}
	return out
}

// mayBeyondTheControllers is what the copy may grant beyond the controller's where the records allow it, and need not
// where they do not: a node's ask of the vault's `own` while the vault serves it (issue 1126, ADR 0550 §5).
func mayBeyondTheControllers(user string) map[string]bool {
	// A runtime's read of its membership from the grants part's state (novox/hq ADR 0571 clause 7): its own key, or the
	// tool runner's node's.
	if parts := strings.Split(user, "."); len(parts) == 2 && parts[0] != "node" {
		key := user
		if parts[1] == RuntimeModule {
			key = parts[0] + ".*"
		}
		out := map[string]bool{}
		for _, s := range readGrants(GrantsMembershipsBucket, key) {
			out[s] = true
		}
		return out
	}
	if _, ok := strings.CutPrefix(user, "node."); !ok {
		return nil
	}
	return map[string]bool{VaultOwnCall(Principal{Kind: KindNode, Node: strings.TrimPrefix(user, "node.")}): true,
		VaultSecretChanged: true}
}

// engineBeyond is a node's engine holding its node seat and asking the operator in its own name (novox/hq ADR 0563 §1,
// §3), beyond the controller's on purpose: its own node's verbs, events and discovery, the bus's discovery asked, and
// the router's seat traffic under node-engine-<node>.
func engineBeyond(user, g string) bool {
	node, ok := strings.CutPrefix(user, "node.")
	if !ok {
		return false
	}
	return strings.HasPrefix(g, "mesh.seat."+EngineSeat+".") && strings.Contains(g, "."+node) ||
		strings.HasPrefix(g, "$SRV.") || strings.Contains(g, ".node-engine-"+node) ||
		strings.Contains(g, "KV_"+EngineActsBucket) || strings.HasPrefix(g, "$KV."+EngineActsBucket+"."+node+".") ||
		// And its node's declared packages (novox/hq ADR 0565 clause 6).
		strings.Contains(g, "KV_"+EnginePackagesBucket) || g == "$KV."+EnginePackagesBucket+"."+node ||
		// And its own health report (issue 774).
		strings.Contains(g, "KV_"+delivery.NodeHealthBucket) || g == "$KV."+delivery.NodeHealthBucket+"."+node ||
		// And the provisions of its node's modules, read (novox/hq ADR 0570 clause 6).
		strings.Contains(g, "KV_"+provisions.ProvisionsBucket)
}

// createsAnAccountsConsumer is the grants part's create and read of a durable consumer an account needs, by its stream
// and name (novox/hq ADR 0561 §6): beyond the controller's on purpose, since the controller made them alone.
func createsAnAccountsConsumer(user, g string) bool {
	return strings.HasSuffix(user, "."+GrantsModule) &&
		(strings.HasPrefix(g, "$JS.API.CONSUMER.CREATE.") || strings.HasPrefix(g, "$JS.API.CONSUMER.INFO."))
}

// quotedToken is a quoted token of a composed list.
var quotedToken = regexp.MustCompile(`"[^"]*"`)

// grantsByUser splits a composed list into the text outside its users and each user's quoted tokens (its name, its
// password's hash, every subject it may publish or subscribe to), by user.
func grantsByUser(list string) (string, map[string]map[string]bool) {
	var outside strings.Builder
	users := map[string]map[string]bool{}
	user := ""
	for _, line := range strings.SplitAfter(list, "\n") {
		trimmed := strings.TrimSpace(line)
		if name, ok := strings.CutPrefix(trimmed, `{ user: "`); ok {
			user, _, _ = strings.Cut(name, `"`)
			users[user] = map[string]bool{}
		} else if user != "" && trimmed == "]" {
			user = ""
		}
		if user == "" {
			outside.WriteString(line)
			continue
		}
		for _, q := range quotedToken.FindAllString(line, -1) {
			users[user][strings.Trim(q, `"`)] = true
		}
	}
	return outside.String(), users
}

// grantsAsTheControllers fails t unless the copy's list is the controller's, user for user, but for the grants it
// names on purpose (beyondTheControllers, the creation advisories): every grant of the controller's is the copy's, and
// the copy's beyond them are exactly those, every one of them there.
func grantsAsTheControllers(t *testing.T, got, want string) {
	t.Helper()
	gotOutside, gotUsers := grantsByUser(got)
	wantOutside, wantUsers := grantsByUser(want)
	if gotOutside != wantOutside {
		t.Errorf("the list differs from the controller's outside its users:\n%s", gotOutside)
	}
	for user, was := range wantUsers {
		now, ok := gotUsers[user]
		if !ok {
			t.Errorf("the copy has no user %s", user)
			continue
		}
		intended := beyondTheControllers(user)
		for g := range was {
			if !now[g] {
				t.Errorf("%s: the copy does not grant %s, which the controller's does", user, g)
			}
		}
		for g := range now {
			if !was[g] && !intended[g] && !mayBeyondTheControllers(user)[g] && !createdAdvisory.MatchString(`"`+g+`", `) &&
				!createsAnAccountsConsumer(user, g) && !engineBeyond(user, g) {
				t.Errorf("%s: the copy grants %s beyond the controller's, and no decision names it", user, g)
			}
		}
		for g := range intended {
			if !now[g] {
				t.Errorf("%s: the copy does not grant %s, which a node composing its own modules reads", user, g)
			}
		}
	}
	for user := range gotUsers {
		if _, ok := wantUsers[user]; !ok {
			t.Errorf("the copy has a user %s the controller's has not", user)
		}
	}
}

const SignersBucketName = "mesh-vault_signers"

func TestTheCopiedComposerComposesAsTheControllers(t *testing.T) {
	// ADR 0567 §3, §9: a node reads every key of the seat's placement and its candidates (it composes with the whole
	// proposal laid over), and none of its open proposals; it writes neither.
	reads := strings.Join(NodeComposeReads("laptop"), " ")
	for bucket, granted := range map[string]bool{delivery.PlacementBucket: true, delivery.PlacementCandidatesBucket: true,
		delivery.PlacementProposalsBucket: false} {
		if strings.Contains(reads, "$JS.API.DIRECT.GET.KV_"+bucket+".>") != granted || strings.Contains(reads, "$KV."+bucket+".laptop") {
			t.Errorf("a node is granted %s: %v, want %v", bucket, !granted, granted)
		}
	}
	// It hears the creation of those two buckets, and of no other.
	if got := NodeComposeAdvisories(); len(got) != 2 || got[0] != "$JS.EVENT.ADVISORY.STREAM.CREATED.KV_mesh-delivery-assigner_placement" ||
		got[1] != "$JS.EVENT.ADVISORY.STREAM.CREATED.KV_mesh-delivery-assigner_candidates" {
		t.Errorf("a node hears the creation of %v", got)
	}
	raw, err := os.ReadFile(filepath.Join("testdata", "controller-user-lists.json"))
	if err != nil {
		t.Fatal(err)
	}
	var want []struct {
		Name, List  string
		Missing     []string
		Err         string
		Memberships json.RawMessage
	}
	if err := json.Unmarshal(raw, &want); err != nil {
		t.Fatal(err)
	}
	fixtures := UserListFixtures()
	if len(want) != len(fixtures) {
		t.Fatalf("%d fixtures, %d lists the controller composed", len(fixtures), len(want))
	}
	for i, f := range fixtures {
		t.Run(f.Name, func(t *testing.T) {
			if want[i].Name != f.Name {
				t.Fatalf("fixture %q against the controller's %q", f.Name, want[i].Name)
			}
			list, missing, err := ComposedFor(f)
			got := ""
			if err != nil {
				got = err.Error()
			}
			if got != want[i].Err {
				t.Fatalf("error %q, the controller's %q", got, want[i].Err)
			}
			if strings.Join(missing, ",") != strings.Join(want[i].Missing, ",") {
				t.Errorf("left out %v, the controller %v", missing, want[i].Missing)
			}
			// The one row the copy composes beyond the controller's, on purpose (novox/hq issue 1790, 2026-10-26, after
			// the take-over of ADR 0547 milestone 2): each principal hears its own and read buckets created, so a state
			// missing at its start is bound when it appears. The controller's composer is frozen and its list no longer
			// loaded; while it is not, this row is left out of the comparison and nothing else.
			grantsAsTheControllers(t, list, want[i].List)
			// And every membership the records issue, as the controller's MembershipFor makes it (ADR 0547 §5): the
			// grants part publishes them.
			raw, err := json.Marshal(MembershipsFor(f))
			if err != nil {
				t.Fatal(err)
			}
			var gotM, wasM any
			if err := json.Unmarshal(raw, &gotM); err != nil {
				t.Fatal(err)
			}
			if err := json.Unmarshal(want[i].Memberships, &wasM); err != nil {
				t.Fatal(err)
			}
			if !reflect.DeepEqual(gotM, wasM) {
				t.Errorf("the memberships differ from the controller's:\n%s", raw)
			}
		})
	}
	// An asker on a seat pulled by kind hears what that seat's workers gave up on, and no other stream's dead letters
	// (novox/hq issue 734: a check ask given up during a bus reload left its try running for ever).
	t.Run("an asker reads its seat's dead letters", func(t *testing.T) {
		build := Seat{Name: "node-build-go", Scope: "node", Accepts: []string{"delivery", "check", "recheck"}, PulledByKind: true}
		perms, err := PermissionsFor(Principal{Kind: KindModule, Node: "one", Module: "mesh-delivery-checker", Uses: []Seat{build}})
		if err != nil {
			t.Fatal(err)
		}
		for subject, want := range map[string]bool{
			"$JS.API.CONSUMER.CREATE.DEAD_LETTERS.x1.mesh.events.dead.SEAT_NODE_BUILD_GO.>":      true,
			"$JS.API.CONSUMER.CREATE.DEAD_LETTERS.x1.mesh.events.dead.SEAT_NODE_BUILD_GO.w.next": true,
			"$JS.API.CONSUMER.DELETE.DEAD_LETTERS.x1":                                            true,
			"$JS.API.CONSUMER.MSG.NEXT.DEAD_LETTERS.x1":                                          true,
			"$JS.API.CONSUMER.CREATE.DEAD_LETTERS.x1.mesh.events.dead.EVENTS.>":                  false,
			"$JS.API.CONSUMER.CREATE.DEAD_LETTERS.x1":                                            false,
			"$JS.API.STREAM.PURGE.DEAD_LETTERS":                                                  false,
			"$JS.FC.DEAD_LETTERS.x1":                                                             false,
		} {
			if got := MayPublish(perms, subject); got != want {
				t.Errorf("may publish %s: %v, want %v", subject, got, want)
			}
		}
		plain, err := PermissionsFor(Principal{Kind: KindModule, Node: "one", Module: "shop",
			Uses: []Seat{{Name: "telegram-sender", Accepts: []string{"send"}}}})
		if err != nil {
			t.Fatal(err)
		}
		if MayPublish(plain, "$JS.API.CONSUMER.DELETE.DEAD_LETTERS.x1") {
			t.Error("a user of a seat not pulled by kind reads DEAD_LETTERS")
		}
	})
	// ADR 0563 §1, §3, §6: each node's engine alone answers its own node's node-engine verbs and says its events, any
	// runtime may call them, and it asks the operator under node-engine-<node> alone.
	t.Run("a node's engine holds its node seat and asks in its own name", func(t *testing.T) {
		router := Seat{Name: AsksSeat, Accepts: []string{"ask", "cancel"}, Emits: []string{"decided"},
			ByCaller: []string{"ask", "cancel", "decided"}, Records: []string{"messenger_asks"}}
		users, err := Users(Records{Nodes: []string{"one", "two"}, Assigned: map[string][]Declared{
			"one": {{Module: "messenger", Holds: []Seat{router}, Invokes: []string{"seat:" + EngineSeat + ".root-free"}}},
			"two": {{Module: RuntimeModule}, {Module: "shop", Invokes: []string{"*"}}}}})
		if err != nil {
			t.Fatal(err)
		}
		for _, u := range users {
			perms, err := PermissionsFor(u)
			if err != nil {
				t.Fatal(err)
			}
			answers := MaySubscribe(perms, "mesh.seat."+EngineSeat+".tool.root-free.one") ||
				MaySubscribe(perms, "mesh.seat."+EngineSeat+".call.root-free.one.messenger~one")
			if engineOne := u.Kind == KindNode && u.Node == "one"; answers != engineOne {
				t.Errorf("%s answers one's root-free: %v", u.Username(), answers)
			}
			switch {
			case u.Kind == KindNode && u.Node == "one":
				for subject, want := range map[string]bool{
					"mesh.seat.operator-channel.accept.ask.node-engine-one":                               true,
					"mesh.seat.operator-channel.accept.cancel.node-engine-one":                            true,
					"mesh.seat.operator-channel.accept.ask.node-engine-two":                               false,
					"$JS.API.DIRECT.GET.KV_messenger_asks.$KV.messenger_asks.node-engine-one.hand-over-1": true,
					"$JS.API.DIRECT.GET.KV_messenger_asks.$KV.messenger_asks.node-engine-two.hand-over-1": false,
					"mesh.seat." + EngineSeat + ".event.done.one":                                         true,
					"mesh.seat." + EngineSeat + ".event.done.two":                                         false,
					"$SRV.INFO": true,
				} {
					if got := MayPublish(perms, subject); got != want {
						t.Errorf("node one may publish %s: %v, want %v", subject, got, want)
					}
				}
				if !MaySubscribe(perms, "mesh.seat.operator-channel.event.decided.node-engine-one") ||
					MaySubscribe(perms, "mesh.seat.operator-channel.event.decided.node-engine-two") ||
					!MaySubscribe(perms, "$SRV.INFO."+EngineSeat) || MaySubscribe(perms, "mesh.seat."+EngineSeat+".tool.hand-over.two") {
					t.Errorf("node one hears %v", perms.Subscribe)
				}
			case u.Kind == KindNodeTools || u.Module == "messenger":
				if !MayPublish(perms, "mesh.seat."+EngineSeat+".tool.root-free.one") {
					t.Errorf("%s cannot call one's root-free", u.Username())
				}
			}
		}
	})
	// Issue 1126 (ADR 0550 §5): a node's engine asks the vault's `own` in its own name while the vault serves it, and no
	// other node's.
	t.Run("a node asks the vault its own secrets", func(t *testing.T) {
		one := Principal{Kind: KindNode, Node: "one", AsksItsOwnSecrets: true}
		perms, err := PermissionsFor(one)
		if err != nil {
			t.Fatal(err)
		}
		if !MayPublish(perms, VaultOwnCall(one)) || MayPublish(perms, VaultOwnCall(Principal{Kind: KindNode, Node: "two"})) {
			t.Errorf("node one asks own as %v, and as two %v", MayPublish(perms, VaultOwnCall(one)),
				MayPublish(perms, VaultOwnCall(Principal{Kind: KindNode, Node: "two"})))
		}
		if plain, _ := PermissionsFor(Principal{Kind: KindNode, Node: "one"}); MayPublish(plain, VaultOwnCall(one)) {
			t.Error("a node asks own while the vault does not serve it")
		}
		if !slices.Contains(perms.Subscribe, VaultSecretChanged) {
			t.Errorf("node one does not hear the vault's secret-changed: %v", perms.Subscribe)
		}
		// L3 s6 (ADR 0550 §2): while the vault serves `sign`, a node asks it in its own name alone, and reads the vault's
		// `signers` and the lease key, never writing either.
		signs, _ := PermissionsFor(Principal{Kind: KindNode, Node: "one", AsksTheVaultToSign: true})
		signersGet := "$JS.API.DIRECT.GET.KV_" + SignersBucketName + ".$KV." + SignersBucketName + ".key"
		leaseGet := "$JS.API.DIRECT.GET.KV_" + LeaseBucket + ".$KV." + LeaseBucket + ".holder"
		for subject, want := range map[string]bool{VaultSignCall(one): true, signersGet: true, leaseGet: true,
			VaultSignCall(Principal{Kind: KindNode, Node: "two"}): false, "$KV." + SignersBucketName + ".key": false,
			"$KV." + LeaseBucket + ".holder": false} {
			if MayPublish(signs, subject) != want {
				t.Errorf("a node asking sign may publish %s: %v, want %v", subject, !want, want)
			}
		}
		if MayPublish(perms, VaultSignCall(one)) || MayPublish(perms, signersGet) {
			t.Error("a node asks sign or reads signers while the vault does not serve sign")
		}
		// And what a module on another node serves from its settings, that key alone (mailu's smtp domain, issue 1126).
		users, err := Users(Records{Nodes: []string{"one", "two"}, Assigned: map[string][]Declared{
			"two": {{Module: "mailer", Served: []string{"domain", "not.a.token"}}}}})
		if err != nil {
			t.Fatal(err)
		}
		get := func(key string) string {
			return "$JS.API.DIRECT.GET.KV_" + ResolvedBucket + ".$KV." + ResolvedBucket + "." + key
		}
		for _, u := range users {
			if u.Kind != KindNode {
				continue
			}
			perms, err := PermissionsFor(u)
			if err != nil {
				t.Fatal(err)
			}
			if u.Node == "one" && (!reflect.DeepEqual(u.ServedReads, []string{"two.mailer.domain"}) || !MayPublish(perms, get("two.mailer.domain")) ||
				MayPublish(perms, get("two.mailer.sitename"))) {
				t.Errorf("node one reads %v of two's resolved values", u.ServedReads)
			}
			if u.Node == "two" && len(u.ServedReads) != 0 {
				t.Errorf("node two is granted its own served keys apart: %v", u.ServedReads)
			}
		}
	})
	t.Run("the controller's golden server configuration", func(t *testing.T) {
		seat := Seat{Name: "telegram-sender", Accepts: []string{"send"}, Emits: []string{"delivered", "failed"}}
		got, err := Compose(
			Server{ClientPort: 4222, MonitoringPort: 8222, StoreDir: "/data",
				TLSCert: "/tls/tls.crt", TLSKey: "/tls/tls.key", TLSCA: "/tls/ca.crt"},
			[]Principal{
				{Kind: KindController, PasswordHash: "$2a$11$cccccccccccccccccccccc"},
				{Kind: KindEnrolment, Node: "one", PasswordHash: "$2a$11$eeeeeeeeeeeeeeeeeeeeee"},
				{Kind: KindNode, Node: "one", PasswordHash: "$2a$11$nnnnnnnnnnnnnnnnnnnnnn"},
				{Kind: KindModule, Node: "one", Module: "telegram", Holds: []Seat{seat},
					Serves: []string{"status"}, PasswordHash: "$2a$11$tttttttttttttttttttttt"},
				{Kind: KindModule, Node: "two", Module: "shop", Uses: []Seat{seat},
					Emits: []string{"order.placed"}, PasswordHash: "$2a$11$ssssssssssssssssssssss"},
				{Kind: KindModule, Node: "two", Module: "audit",
					Consumes: []string{"shop.order.placed"}, PasswordHash: "$2a$11$aaaaaaaaaaaaaaaaaaaaaa"},
				{Kind: KindModule, Node: "one", Module: WatchmanModule, PasswordHash: "$2a$11$wwwwwwwwwwwwwwwwwwwwww"},
				{Kind: KindModule, Node: "one", Module: "nats", SnapshotsTheBus: true,
					Serves: []string{"nats_streams"}, PasswordHash: "$2a$11$bbbbbbbbbbbbbbbbbbbbbb"},
			})
		if err != nil {
			t.Fatal(err)
		}
		want, err := os.ReadFile(filepath.Join("testdata", "composed.conf"))
		if err != nil {
			t.Fatal(err)
		}
		grantsAsTheControllers(t, got, string(want))
	})
	// 774/AC2 (novox/hq ADR 0561 §2): a wanted account round-trips through the grants part's state `accounts`, keyed by
	// its user, at the subject `$KV.mesh-grants_accounts.<user>`, read by the vault alone; and the vault's account-issued
	// names the account, never a value.
	t.Run("a wanted account is kept by its user and read by the vault alone", func(t *testing.T) {
		for _, w := range []WantedAccount{
			{User: "ace.mesh-build-go", Kind: KindModule, Node: "ace", Module: "mesh-build-go", URL: "nats://novox.mesh:4222",
				Fingerprint: "ab", Claims: []ClaimedSeat{{Seat: "node-build-go", Scope: "node", Serves: []string{"build"}}}},
			{User: "novox.mesh-mail.send", Kind: KindProcess, Node: "novox", Module: "mesh-mail", Process: "send",
				URL: "nats://novox.mesh:4222"},
			{User: "provisioner.g14", Kind: KindProvisioner, Node: "g14", Module: "mesh-provisioner-client",
				URL: "nats://novox.mesh:4222", Fingerprint: "ab"},
		} {
			if err := w.Valid(); err != nil {
				t.Fatalf("%s: %v", w.User, err)
			}
			raw, err := json.Marshal(w)
			if err != nil {
				t.Fatal(err)
			}
			back, err := ReadWantedAccount(w.User, raw)
			if err != nil || !back.Same(w) {
				t.Errorf("%s does not round-trip: %+v, %v", w.User, back, err)
			}
			if got := AccountsSubject(w.User); got != "$KV.mesh-grants_accounts."+w.User {
				t.Errorf("%s is kept at %s", w.User, got)
			}
			if _, err := ReadWantedAccount("someone.else", raw); err == nil {
				t.Errorf("%s was read under another user's key", w.User)
			}
		}
		for _, bad := range []WantedAccount{
			{User: "ace.x", Kind: KindModule, Node: "ace", Module: "y", URL: "nats://a:1"},
			{User: "ace.x.p", Kind: KindProcess, Node: "ace", Module: "x", URL: "nats://a:1"},
			{User: "node.ace", Kind: KindNode, Node: "ace", URL: "nats://a:1"},
			{User: "controller", Kind: KindController, URL: "nats://a:1"},
			{User: "ace.x", Kind: KindModule, Node: "ace", Module: "x"},
		} {
			if bad.Valid() == nil {
				t.Errorf("%+v is taken as a wanted account", bad)
			}
		}
		if ReaderMay("mesh-delivery-assigner", GrantsAccounts) || !ReaderMay(VaultSeat, GrantsAccounts) {
			t.Errorf("%s is read by %v alone, want the vault's", GrantsAccounts, NamedReaders[GrantsAccounts])
		}
		e := AccountIssued{User: "ace.x", Kind: KindModule, Node: "ace", Module: "x", Generation: 1}
		raw, _ := json.Marshal(e)
		if strings.Contains(string(raw), "password") || strings.Contains(string(raw), "hash") ||
			AccountIssuedSubject != "mesh.mod.mesh-vault.event.account-issued" {
			t.Errorf("account-issued says %s at %s", raw, AccountIssuedSubject)
		}
	})
	// 774/AC3 (novox/hq ADR 0561 §2): the account the vault seals is, byte for byte, what the controller hands for each of
	// its three kinds: testdata/controller-credentials.json was written by the controller's own code (its "how").
	t.Run("a held account is the controller's shape byte for byte", func(t *testing.T) {
		raw, err := os.ReadFile(filepath.Join("testdata", "controller-credentials.json"))
		if err != nil {
			t.Fatal(err)
		}
		var captured struct {
			Rows []struct{ Name, Held string }
		}
		if err := json.Unmarshal(raw, &captured); err != nil {
			t.Fatal(err)
		}
		vault := ClaimedSeat{Seat: "mesh-vault", Scope: "mesh", Serves: []string{"get", "accept", "secret-ask", "issued",
			"rotate", "move-in", "node-key", "own", "move-asks"}}
		controller := ClaimedSeat{Seat: "mesh-controller", Scope: "mesh", Serves: strings.Split("tools,calls,status,nodes,"+
			"node,placement-export,modules,seats,builds,plans,delivery-plan,delivery-order,delivery-check,deliver,"+
			"delivery-stop,delivery-backups-again,delivery-walks,plan,assign,unassign,pin,unpin,push,rotate,give,issued,"+
			"secret-move,secret-ask,issue,token,settings,command,queue,cancel,clear,replay,try,kill,pause,resume,hand-act,"+
			"drill,warranted,hand-acts,durations,conditions,healers,doctor,upgrade,bus,retire,cleanup,judge-directories,"+
			"root-free,dead-letters,data,build,artifacts,held,collect,move-placement,move-config,mirrors,images", ",")}
		at := "nats://10.0.0.1:4222"
		wanted := map[string]struct {
			w        WantedAccount
			password string
		}{
			"module": {WantedAccount{User: "laptop.cap-bot", Kind: KindModule, Node: "laptop", Module: "cap-bot", URL: at,
				Fingerprint: "ab:cd", Claims: []ClaimedSeat{vault, {Seat: "node-service-manager", Scope: "node",
					Serves: []string{"status"}}, controller, {Seat: "no-such-seat", Scope: "node"}}}, "pw-module"},
			"module-no-claims-no-fingerprint": {WantedAccount{User: "laptop.cap-plain", Kind: KindModule, Node: "laptop",
				Module: "cap-plain", URL: at}, "pw-plain"},
			"provisioner": {WantedAccount{User: "provisioner.laptop", Kind: KindProvisioner, Node: "laptop", Module: "nats",
				URL: at, Fingerprint: "ab:cd"}, "pw-prov"},
			"process": {WantedAccount{User: "laptop.mesh-mail.send", Kind: KindProcess, Node: "laptop", Module: "mesh-mail",
				Process: "send", URL: at, Fingerprint: "ab:cd"}, "pw-proc"},
		}
		if len(captured.Rows) != len(wanted) {
			t.Fatalf("%d rows captured, %d wanted", len(captured.Rows), len(wanted))
		}
		for _, row := range captured.Rows {
			c, ok := wanted[row.Name]
			if !ok {
				t.Fatalf("no wanted account for the captured %s", row.Name)
			}
			got, err := HeldCredential(c.w, c.password)
			if err != nil || string(got) != row.Held {
				t.Errorf("%s:\n got %s (%v)\nwant %s", row.Name, got, err, row.Held)
			}
		}
		if _, err := HeldCredential(wanted["module"].w, ""); err == nil {
			t.Error("an account was held with no password")
		}
	})
	// 774 subtasks 6 and 8 (novox/hq ADR 0561 §3, §6): minting and the grants part's consumers turn on only while the
	// lease is handed to a holder that is no controller; the vault and the grants part alone read the lease, whatever the
	// controller announced; a module's consumer is created as the controller's EnsureConsumer creates it.
	t.Run("the lease turns minting on only when handed away; a consumer is created as the controller's", func(t *testing.T) {
		for _, row := range []struct {
			name  string
			raw   string
			found bool
			err   error
			on    bool
		}{
			{"a controller holds it", `{"instance":"novox/1234/2026-10-27T20:00:00Z","epoch":57}`, true, nil, false},
			{"handed away", `{"handed-to":"the operator's hand-over","handed-by":"person.jochen"}`, true, nil, true},
			{"expired or deleted", ``, false, nil, false},
			{"unreadable", ``, false, errors.New("refused"), false},
			{"not a holder", `[1]`, true, nil, false},
			{"both a controller and a handed holder", `{"instance":"novox/1","handed-to":"x"}`, true, nil, false},
		} {
			if on, why := LeaseHandedAway([]byte(row.raw), row.found, row.err); on != row.on || why == "" {
				t.Errorf("%s: on %v (%s), want %v", row.name, on, why, row.on)
			}
			// The assigner's grants source follows it: the grants part's bucket only for a handed-away lease (issue 774).
			on, _ := LeaseHandedAway([]byte(row.raw), row.found, row.err)
			if bucket, _ := delivery.GrantSource(on); (bucket == delivery.PartGrantsBucket) != row.on {
				t.Errorf("%s: the grants source is %s", row.name, bucket)
			}
		}
		announced := map[string]announce.States{ControllerSeat: ControllerAnnouncement("novox")}
		for reader, want := range map[string]bool{VaultSeat: true, GrantsModule: true, "mesh-delivery-assigner": true, "mesh-delivery-view": false} {
			granted, _ := ReadsAnnouncedBy(reader, []string{ControllerLease}, announced)
			if (len(granted) == 1) != want {
				t.Errorf("%s reads the lease: %v, want %v", reader, granted, want)
			}
		}
		c, _ := ConsumerFor(Principal{Kind: KindModule, Node: "a", Module: "m", Consumes: []string{"x.happened"}})
		cfg := ConsumerConfig(c)
		if cfg.Durable != "a_m" || cfg.AckPolicy != jetstream.AckExplicitPolicy || cfg.AckWait != 30*time.Second ||
			cfg.MaxDeliver != 5 || cfg.DeliverPolicy != jetstream.DeliverNewPolicy || len(cfg.FilterSubjects) != 2 ||
			cfg.Description == "" {
			t.Errorf("a module's consumer is created as %+v", cfg)
		}
		// The grants part is granted the create of each account's consumer by its stream and name, and no other.
		users, err := Users(Records{Nodes: []string{"a"}, Assigned: map[string][]Declared{"a": {
			{Module: GrantsModule}, {Module: "m", Consumes: []string{"x.happened"}}, {Module: "quiet"}}}})
		if err != nil {
			t.Fatal(err)
		}
		for _, u := range users {
			if u.Module != GrantsModule || u.Kind != KindModule {
				continue
			}
			perms, err := PermissionsFor(u)
			if err != nil {
				t.Fatal(err)
			}
			if !MayPublish(perms, "$JS.API.CONSUMER.CREATE.EVENTS.a_m") || MayPublish(perms, "$JS.API.CONSUMER.CREATE.EVENTS.a_quiet") ||
				MayPublish(perms, "$JS.API.CONSUMER.DELETE.EVENTS.a_m") || MayPublish(perms, "$JS.API.CONSUMER.CREATE.EVENTS.a_m.mesh.mod.x.event.happened") {
				t.Errorf("the grants part creates consumers as %v", u.CreatesConsumers)
			}
		}
		// 1952 S5 (novox/hq ADR 0564 clauses 2, 5, 8; Limits 2, 4, 8): the health holders' credentials.
		healthChecks(t)
		q := ConsumerConfig(QueueConsumer(Principal{Kind: KindProcess, Node: "a", Module: "m", Process: "p"},
			QueueOwner{Process: "p", Stream: "M_WORK", MaxDeliver: 3}))
		if q.Durable != "a_m_p" || q.FilterSubject != "" || q.MaxDeliver != 3 || q.DeliverPolicy != jetstream.DeliverAllPolicy {
			t.Errorf("a process's queue consumer is created as %+v", q)
		}
	})
}

// healthChecks are the health holders' credential cases (ADR 0564 clauses 2, 5, 8; Limits 2, 4, 8; issue 1952 S5).
func healthChecks(t *testing.T) {
	t.Helper()
	announced := map[string]announce.States{ControllerSeat: ControllerAnnouncement("novox"),
		"mesh-healers": {Module: "mesh-healers", States: []announce.State{{Name: "attempts", Readable: true}, {Name: "would", Readable: true}}}}
	for _, c := range []struct {
		reader, read string
		may          bool
	}{
		{"mesh-conditions", ControllerLease, true}, {"mesh-healers", ControllerLease, true}, {WatchmanModule, ControllerLease, true},
		{"mesh-self-check", ControllerLease, false}, {"messenger", ControllerLease, false}, {ArtifactStoreTools, ControllerLease, true},
		{"mesh-conditions", ControllerConditions, true}, {"mesh-healers", ControllerConditions, false}, {"messenger", ControllerConditions, false},
		{"mesh-healers", ControllerHandActs, true}, {"mesh-conditions", ControllerHandActs, false},
		// ADR 0565 clauses 10, 11 (issue 1953 L5-S4): mesh-hand-acts reads the controller's log beside its own.
		{"mesh-hand-acts", ControllerHandActs, true},
		{"mesh-conditions", "mesh-healers.attempts", true}, {"mesh-conditions", HealersWould, false}, {"mesh-delivery-view", HealersWould, false},
	} {
		granted, _ := ReadsAnnouncedBy(c.reader, []string{c.read}, announced)
		if (len(granted) == 1) != c.may {
			t.Errorf("%s reads %s: %v, want %v", c.reader, c.read, granted, c.may)
		}
	}
	bucket := func(m, n string) Bucket { return Bucket{Module: m, Name: n} }
	users, err := Users(Records{Nodes: []string{"a"}, Assigned: map[string][]Declared{"a": {
		{Module: "mesh-conditions", State: []Bucket{bucket("mesh-conditions", "conditions"), bucket("mesh-conditions", "takeover")},
			Reads: []string{"mesh-self-check.findings", "mesh-healers.attempts", ControllerConditions}},
		{Module: "mesh-self-check", State: []Bucket{bucket("mesh-self-check", "findings")}, Reads: []string{"restic.good"}},
		// Issue 1955 L7: the grants part keeps the hearings it withholds; the self-check reads them.
		{Module: GrantsModule, State: []Bucket{bucket(GrantsModule, GrantsWithheldState)}},
		{Module: "mesh-healers", State: []Bucket{bucket("mesh-healers", "findings"), bucket("mesh-healers", "attempts"), bucket("mesh-healers", "would")}},
		{Module: WatchmanModule, State: []Bucket{bucket(WatchmanModule, "findings")}, Reads: []string{delivery.NodeHealthRead}},
		{Module: "mesh-hand-acts", State: []Bucket{bucket("mesh-hand-acts", "warrants")}, Logs: []Log{{Module: "mesh-hand-acts", Name: "acts"}}},
		// ADR 0570 clauses 1, 8 (issue 1954 L6-P5): mesh-provisioner keeps provisions and plans; the vault and a provider
		// read them.
		{Module: "mesh-provisioner", State: []Bucket{bucket("mesh-provisioner", "provisions"), bucket("mesh-provisioner", "plans"),
			bucket("mesh-provisioner", "registry")}},
		{Module: "mesh-vault", Reads: []string{"mesh-provisioner.provisions", "mesh-provisioner.plans"}},
		{Module: "postgres", Reads: []string{"mesh-provisioner.provisions", "mesh-provisioner.plans"}},
		// The node's provisioner client reads its machine's plans and records no needs (ADR 0570 clause 2; its V3).
		{Module: "mesh-provisioner-client", ProvisionsNode: true},
		// ADR 0571 clauses 2, 4 (issue 1955): the seat mesh-artifact-sweep says artifacts-gone; the assigner keeps seat-holders.
		{Module: "artifact-store-tools", Holds: []Seat{{Name: "mesh-artifact-sweep", Emits: []string{"artifacts-gone"}}},
			Reads: []string{delivery.SeatHoldersRead}},
		{Module: delivery.AssignerModule, State: []Bucket{bucket(delivery.AssignerModule, delivery.SeatHoldersState)}},
		// ADR 0571 clause 3: disk-load keeps the readings; the backup holder reads them.
		{Module: "disk-load", State: []Bucket{bucket("disk-load", "readings")}},
		{Module: "restic", Reads: []string{"disk-load.readings"}, State: []Bucket{bucket("restic", "good")}},
		// ADR 0571 clauses 1, 9: the indexer reads the builder's log through its own consumer; the catalogue module asks
		// the builder's tool.
		{Module: delivery.BuilderModule, Logs: []Log{{Module: delivery.BuilderModule, Name: delivery.BuilderLog, MaxAgeDays: 30}}},
		{Module: delivery.IndexerModule, Reads: []string{delivery.BuildLogRead}},
		{Module: "mesh-catalog", Invokes: []string{delivery.BuilderModule + "." + delivery.BuildsSinceTool}},
		{Module: "telegram", Holds: []Seat{{Name: "telegram-sender", Accepts: []string{"send"}}},
			Processes: []ProcessPrincipal{{Process: "w", Queues: []QueueOwner{{Process: "w", Stream: "M_WORK"}, {Process: "w", Stream: "KV_x"}}}}},
		{Module: "nats", SnapshotsTheBus: true, Processes: []ProcessPrincipal{{Process: "nats-tools",
			Removes: []string{catalogue.RemovesControllerDeadLetterConsumer, catalogue.RemovesEngineActs}}, {Process: "other"}}},
		{Module: "shop", Processes: []ProcessPrincipal{{Process: "nats-tools", Removes: []string{catalogue.RemovesEngineActs}}}},
	}}})
	if err != nil {
		t.Fatal(err)
	}
	may := map[string][]string{
		"a.mesh-conditions":  {"$KV.mesh-conditions_conditions.machine.g14.silent"},
		"a.mesh-provisioner": {"$KV.mesh-provisioner_provisions.a.keycloak.postgres-database", "$KV.mesh-provisioner_plans.a.keycloak"},
		"a.mesh-vault": {"$JS.API.DIRECT.GET.KV_mesh-provisioner_provisions.$KV.mesh-provisioner_provisions.a.keycloak.postgres-database",
			"$JS.API.DIRECT.GET.KV_mesh-provisioner_plans.$KV.mesh-provisioner_plans.a.keycloak"},
		"a.mesh-hand-acts": {"$KV.mesh-hand-acts_warrants.mesh-config.ask-1", "mesh.log.mesh-hand-acts.acts.act-1"},
		"a.disk-load":      {"$KV.disk-load_readings.a.pg.data"},
		"provisioner.a": {"$JS.API.DIRECT.GET.KV_mesh-provisioner_plans.$KV.mesh-provisioner_plans.a.keycloak",
			"$JS.API.CONSUMER.CREATE.KV_mesh-provisioner_plans.c1.$KV.mesh-provisioner_plans.a.keycloak",
			// The needs, for the client before V3, until the lease write (review round 3).
			"$KV." + ProvisionerNeedsBucket + ".a.keycloak.db.x"},
		// ADR 0571 clause 3: restic's bundle writes each item's last good night as its own state; the self-check reads it.
		"a.mesh-self-check": {"$JS.API.DIRECT.GET.KV_restic_good.$KV.restic_good.a.pg.data"},
		"a.mesh-grants":     {"$KV.mesh-grants_withheld.reader.mail.a"},
		"a.restic": {"$JS.API.DIRECT.GET.KV_disk-load_readings.$KV.disk-load_readings.a.pg.data", "$KV.restic_good.a.pg.data",
			// ADR 0571 clause 7: its own membership from the grants part's state.
			"$JS.API.DIRECT.GET.KV_mesh-grants_memberships.$KV.mesh-grants_memberships.a.restic"},
		"a.mesh-delivery-indexer": {"$JS.API.CONSUMER.CREATE.LOG_mesh-delivery-builder_build-log.r1.mesh.log.mesh-delivery-builder.build-log.letta",
			"$JS.API.CONSUMER.MSG.NEXT.LOG_mesh-delivery-builder_build-log.r1", "$JS.API.CONSUMER.DELETE.LOG_mesh-delivery-builder_build-log.r1",
			"$JS.API.DIRECT.GET.LOG_mesh-delivery-builder_build-log.mesh.log.mesh-delivery-builder.build-log.letta"},
		"a.mesh-delivery-builder": {"mesh.log.mesh-delivery-builder.build-log.letta"},
		"a.mesh-catalog":          {"mesh.mod.mesh-delivery-builder.tool.builder_builds_since"},
		WatchmanModule: {"$JS.API.CONSUMER.CREATE.DEAD_LETTER_NOTICES." + WatchmanModule, "$JS.API.CONSUMER.MSG.NEXT.DEAD_LETTER_NOTICES." + WatchmanModule,
			"$JS.API.CONSUMER.INFO.DEAD_LETTER_NOTICES.controller", "$JS.API.STREAM.MSG.GET.DEAD_LETTERS"},
		WatchmanModule + " sources": {"$JS.API.STREAM.MSG.GET.EVENTS", "$JS.API.STREAM.INFO.SEAT_TELEGRAM_SENDER",
			"$JS.API.STREAM.MSG.GET.M_WORK", "$JS.API.STREAM.INFO.EVENTS", "mesh.events.dead.EVENTS.a_m"},
		"a.nats.nats-tools": {"$JS.API.CONSUMER.DELETE.DEAD_LETTER_NOTICES.controller", "$JS.API.CONSUMER.INFO.DEAD_LETTER_NOTICES.controller",
			"$KV.mesh-host_acts.g14.a1", "$JS.API.STREAM.INFO.KV_mesh-host_acts"},
	}
	mayNot := map[string][]string{
		// The keeper writes no findings and no attempts; no observer writes the keeper's buckets (clause 2).
		"a.mesh-conditions": {"$KV.mesh-self-check_findings.x", "$KV.mesh-healers_attempts.x", "$KV.mesh-controller_conditions.x"},
		"a.mesh-self-check": {"$KV.mesh-conditions_conditions.x", "$KV.mesh-conditions_takeover.switch", "$KV.restic_good.a.pg.data",
			"$KV.mesh-grants_withheld.reader.mail.a", "$SRV.INFO", "$SRV.PING.node-tools",
			"$JS.API.DIRECT.GET.KV_mesh-grants_withheld.$KV.mesh-grants_withheld.reader.mail.a"},
		// Only mesh-provisioner writes provisions and plans, and it writes none of the grants part's states (ADR 0570
		// clauses 1, 8).
		"a.mesh-provisioner": {"$KV.mesh-grants_accounts.a.m", "$KV.mesh-grants_user-list.sealed"},
		"a.mesh-vault": {"$KV.mesh-provisioner_provisions.a.keycloak.postgres-database", "$KV.mesh-provisioner_plans.a.keycloak",
			"mesh.seat.mesh-artifact-sweep.event.artifacts-gone"},
		// Each writes its own state alone, and these users are composed.
		"a.disk-load":             {"$KV.restic_good.a.pg.data"},
		"a.mesh-delivery-builder": {"$KV.disk-load_readings.a.pg.data"},
		// Another machine's plans are not its to read, nor another machine's needs, which it keeps for the client before
		// V3 until the lease write.
		"provisioner.a": {"$JS.API.DIRECT.GET.KV_mesh-provisioner_plans.$KV.mesh-provisioner_plans.b.keycloak",
			"$KV.mesh-provisioner_plans.a.keycloak", "$KV." + ProvisionerNeedsBucket + ".b.keycloak.db.x"},
		// Only the assigner writes seat-holders.
		"a.artifact-store-tools": {"$KV.mesh-delivery-assigner_seat-holders.mesh-broker"},
		// Only disk-load writes its readings.
		"a.restic":   {"$KV.disk-load_readings.a.pg.data"},
		"a.postgres": {"$KV.mesh-provisioner_provisions.a.keycloak.postgres-database", "$KV.mesh-provisioner_plans.a.keycloak"},
		// Nor mesh-hand-acts' log and warrants, which it alone writes (ADR 0565 clause 10).
		"a.mesh-healers": {"$KV.mesh-conditions_conditions.x", "$KV.mesh-conditions_takeover.switch",
			"$KV.mesh-hand-acts_warrants.x", "mesh.log.mesh-hand-acts.acts.x"},
		// The watchman never deletes or changes the controller's consumer (clause 5).
		// Its reads and deletes of a message are on the streams a dead letter comes from, named, never a bucket's
		// (the lead, 2026-10-29).
		WatchmanModule + " sources": {"$JS.API.STREAM.MSG.GET.KV_mesh-vault_bus-accounts", "$JS.API.STREAM.MSG.GET.KV_x",
			"$JS.API.STREAM.MSG.DELETE.KV_mesh-host_acts", "$JS.API.STREAM.MSG.GET.OBJ_mesh-grants-user-list", "$JS.API.STREAM.MSG.GET.OTHER",
			// In L4 it neither delivers again nor drops (clause 5, the lead 2026-10-29).
			"$JS.API.STREAM.MSG.DELETE.EVENTS", "$JS.API.STREAM.MSG.DELETE.DEAD_LETTERS", "$JS.API.CONSUMER.CREATE.DEAD_LETTERS.x",
			"mesh.again.a_m.x"},
		WatchmanModule: {"$KV.mesh-conditions_conditions.x", "$JS.API.CONSUMER.DELETE.DEAD_LETTER_NOTICES.controller",
			"$JS.API.CONSUMER.CREATE.DEAD_LETTER_NOTICES.controller", "$JS.API.CONSUMER.DELETE.DEAD_LETTER_NOTICES." + WatchmanModule},
		// The bus module's tools remove the one consumer and the acts' keys, nothing else (Limits 4); its other accounts
		// remove nothing; an account is granted the removals its bus declares and no other (the catalogue refuses them
		// outside the bus module).
		"a.nats.nats-tools": {"$JS.API.CONSUMER.DELETE.DEAD_LETTER_NOTICES.other", "$JS.API.CONSUMER.DELETE.EVENTS.a_m",
			"$KV.mesh-conditions_conditions.x", "$JS.API.STREAM.DELETE.KV_mesh-host_acts", "$JS.API.STREAM.PURGE.KV_mesh-host_acts"},
		"a.nats.other":      {"$JS.API.CONSUMER.DELETE.DEAD_LETTER_NOTICES.controller", "$KV.mesh-host_acts.g14.a1"},
		"a.nats":            {"$JS.API.CONSUMER.DELETE.DEAD_LETTER_NOTICES.controller", "$KV.mesh-host_acts.g14.a1"},
		"a.shop.nats-tools": {"$JS.API.CONSUMER.DELETE.DEAD_LETTER_NOTICES.controller", "$KV.mesh-conditions_conditions.x"},
	}
	// ADR 0571 clause 1: a read naming another module's log is issued to the reader as a log it reads, never writes.
	if m := MembershipFor("a", Declared{Module: delivery.IndexerModule, Reads: []string{delivery.BuildLogRead, "x.y"}},
		Placements{Logs: map[string]bool{delivery.BuildLogRead: true}}); len(m.Logs) != 1 || m.Logs[0].Writes ||
		m.Logs[0].Stream != "LOG_mesh-delivery-builder_build-log" || m.Logs[0].Name != delivery.BuildLogRead {
		t.Errorf("the indexer's membership issues the logs %+v", m.Logs)
	}
	// ADR 0565 clauses 6, 7 (issue 1953 L5-S4): a node's engine writes its own packages key and no other node's; only the
	// assigner writes its policies.
	for _, u := range users {
		perms, err := PermissionsFor(u)
		if err != nil {
			continue
		}
		// ADR 0570 clause 6: a node's engine reads its own node's provision keys, and no other node's, and writes none.
		if u.Kind == KindNode && (!MayPublish(perms, "$JS.API.DIRECT.GET.KV_mesh-provisioner_provisions.$KV.mesh-provisioner_provisions.a.keycloak.db") ||
			MayPublish(perms, "$JS.API.DIRECT.GET.KV_mesh-provisioner_provisions.$KV.mesh-provisioner_provisions.b.keycloak.db") ||
			MayPublish(perms, "$KV.mesh-provisioner_provisions.a.keycloak.db")) {
			t.Errorf("node %s reads the provisions %v", u.Node, perms.Publish)
		}
		// Issue 774: a node's engine writes its own health report's key and no other node's.
		if u.Kind == KindNode && (!MayPublish(perms, "$KV.mesh-host_health.a") || MayPublish(perms, "$KV.mesh-host_health.b")) {
			t.Errorf("node %s writes the health reports %v", u.Node, perms.Publish)
		}
		if u.Kind == KindModule && u.Module == WatchmanModule && (!MayPublish(perms, "$JS.API.DIRECT.GET.KV_mesh-host_health.$KV.mesh-host_health.b") ||
			MayPublish(perms, "$KV.mesh-host_health.b")) {
			t.Errorf("the watchman reads the health reports %v, %v", MayPublish(perms, "$JS.API.DIRECT.GET.KV_mesh-host_health.$KV.mesh-host_health.b"), perms.Publish)
		}
		if u.Kind == KindNode && (!MayPublish(perms, "$KV.mesh-host_packages.a") || MayPublish(perms, "$KV.mesh-host_packages.b")) {
			t.Errorf("node %s writes the packages keys %v", u.Node, perms.Publish)
		}
		if u.Kind == KindModule && MayPublish(perms, "$KV.mesh-delivery-assigner_policies.x") {
			t.Errorf("%s writes the assigner's policies", u.Username())
		}
	}
	// 1953 S-age (ADR 0565 §10): a log that says an age keeps its entries that long and no longer; one that says none, as
	// long as the log (ADR 0297).
	if aged, kept := (Log{Module: "mesh-hand-acts", Name: "acts", MaxAgeDays: 400}).Config(), (Log{Module: "m", Name: "l"}).Config(); aged.MaxAge != 400*24*time.Hour || kept.MaxAge != 0 || !aged.DenyDelete {
		t.Errorf("an aged log's stream keeps %s, an unaged one %s", aged.MaxAge, kept.MaxAge)
	}
	// The tools account's secret is the module's own broker-<binary>, which its manifest's own-secrets names and the
	// vault keeps by that name (AccountSecret).
	if w := (WantedAccount{User: "a.nats.nats-tools", Kind: KindProcess, Node: "a", Module: "nats", Process: "nats-tools",
		URL: "nats://a:4222"}); w.Valid() != nil || w.OwnSecret() != "broker-nats-tools" {
		t.Errorf("the tools account is held in %q (%v)", w.OwnSecret(), w.Valid())
	}
	seen := map[string]bool{}
	for _, u := range users {
		name := u.Username()
		if u.Module == WatchmanModule && u.Kind == KindModule {
			name = WatchmanModule
		}
		perms, err := PermissionsFor(u)
		if err != nil {
			t.Fatalf("%s: %v", name, err)
		}
		if name == WatchmanModule {
			// It hears its watchdogs' words on core subjects (Limits 3, 11).
			for _, subject := range []string{"mesh.mod.power.event.sleeping", "mesh.control.g14.tools-alive",
				"mesh.seat.mesh-self-check.event.checked", "mesh.mod.keycloak.event.provisioner.failing",
				"mesh.mod.nats.event.bus.slow-consumer", "prov.applied.g14.keycloak"} {
				if !MaySubscribe(perms, subject) {
					t.Errorf("the watchman may not hear %s", subject)
				}
			}
			// Never the bus's API audit, which carries every API answer in the account (the lead, 2026-10-29).
			if MaySubscribe(perms, "$JS.EVENT.ADVISORY.API") || MaySubscribe(perms, "$JS.EVENT.ADVISORY.API.x") {
				t.Error("the watchman hears the bus's API audit")
			}
			if !MayPublish(perms, "$JS.API.DIRECT.GET.PROV_APPLIED.prov.applied.g14.keycloak") {
				t.Error("the watchman may not read the newest provisioner report")
			}
		}
		subjects := append(append([]string(nil), may[name]...), may[name+" sources"]...)
		denied := append(append([]string(nil), mayNot[name]...), mayNot[name+" sources"]...)
		if len(may[name+" sources"]) > 0 {
			seen[name+" sources"] = true
		}
		for _, subject := range subjects {
			seen[name] = true
			if !MayPublish(perms, subject) {
				t.Errorf("%s may not publish %s", name, subject)
			}
		}
		for _, subject := range denied {
			seen[name] = true
			if MayPublish(perms, subject) {
				t.Errorf("%s may publish %s", name, subject)
			}
		}
	}
	// Every row names a user that was composed, in either table: a row for a user absent would pass for nothing.
	for _, table := range []map[string][]string{may, mayNot} {
		for name := range table {
			if !seen[name] {
				t.Errorf("no user %s was composed", name)
			}
		}
	}
}