mesh-host/internal/selfcompose/selfcompose_test.go

package selfcompose

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"slices"
	"strings"
	"sync"
	"testing"
	"time"

	"git.novox.be/novox/mesh-sdk/go/buscompose"
	"git.novox.be/novox/mesh-sdk/go/compose"
	"git.novox.be/novox/mesh-sdk/go/compose/frombus"
	"git.novox.be/novox/mesh-sdk/go/delivery"
	"github.com/nats-io/nats.go"

	"github.com/novox/mesh-host/internal/agent"
)

const lemurs = `{"module":"lemurs","version":"1","capabilities":["systemd"],"resources":[{"id":"service","type":"service","unit":"lemurs.service","state":"running","boot":"enabled"}]}`

// fakeBus is the buckets the composer watches, each watch a channel that gets the updates a test writes to its bucket
// whose key its pattern matches, as the bus's own watches do, and the keys Get reads. It keeps every watch and get
// made, so a test checks each against the node's grant.
type fakeBus struct {
	mu      sync.Mutex
	watches []patternWatch
	kv      map[string][]byte // bucket/key
	said    []string
	reads   [][2]string // bucket, pattern or key: every watch and get
}

func newFakeBus() *fakeBus {
	return &fakeBus{kv: map[string][]byte{}}
}

func (b *fakeBus) env() Env {
	return Env{Node: "n1",
		Watch: func(_ context.Context, bucket, pattern string) (<-chan agent.Update, error) {
			b.mu.Lock()
			defer b.mu.Unlock()
			w := patternWatch{bucket: bucket, pattern: pattern, ch: make(chan agent.Update, 64)}
			b.watches = append(b.watches, w)
			b.reads = append(b.reads, [2]string{bucket, pattern})
			return w.ch, nil
		},
		Get: func(_ context.Context, bucket, key string) ([]byte, error) {
			b.mu.Lock()
			defer b.mu.Unlock()
			b.reads = append(b.reads, [2]string{bucket, key})
			return b.kv[bucket+"/"+key], nil
		},
		Say: func(s string) { b.mu.Lock(); b.said = append(b.said, s); b.mu.Unlock() }}
}

// send gives u to every watch of bucket whose pattern matches its key; Caught to every watch of bucket.
func (b *fakeBus) send(bucket string, u agent.Update) {
	b.mu.Lock()
	ws := append([]patternWatch(nil), b.watches...)
	b.mu.Unlock()
	for _, w := range ws {
		if w.bucket == bucket && (u.Caught || matches(w.pattern, u.Key)) {
			w.ch <- u
		}
	}
}

func (b *fakeBus) put(t *testing.T, bucket, key string, value []byte) {
	t.Helper()
	b.send(bucket, agent.Update{Key: key, Value: value})
}

// grantAllows says whether the node's grant (buscompose.PermissionsFor, KindNode) lets it watch or read pattern of
// bucket: the consumer a watch or a get makes (`$JS.API.CONSUMER.CREATE.KV_<bucket>.<name>.$KV.<bucket>.<pattern>`)
// and the stream's information, matched as the bus matches a permission against a subject.
func grantAllows(t *testing.T, node, bucket, pattern string) bool {
	t.Helper()
	perms, err := buscompose.PermissionsFor(buscompose.Principal{Kind: buscompose.KindNode, Node: node})
	if err != nil {
		t.Fatal(err)
	}
	allowed := func(subject string) bool {
		for _, deny := range perms.PublishDeny {
			if subjectMatches(deny, subject) {
				return false
			}
		}
		for _, p := range perms.Publish {
			if subjectMatches(p, subject) {
				return true
			}
		}
		return false
	}
	stream := "KV_" + bucket
	return allowed("$JS.API.STREAM.INFO."+stream) && allowed("$JS.API.CONSUMER.CREATE."+stream+".c1.$KV."+bucket+"."+pattern)
}

// subjectMatches is the bus's match of a permission against a subject: `*` any one token, `>` one or more to the end.
func subjectMatches(permission, subject string) bool {
	p, s := strings.Split(permission, "."), strings.Split(subject, ".")
	for i, tok := range p {
		if tok == ">" {
			return len(s) > i
		}
		if i >= len(s) || (tok != "*" && tok != s[i]) {
			return false
		}
	}
	return len(p) == len(s)
}

func next(t *testing.T, out <-chan agent.Update) agent.Update {
	t.Helper()
	select {
	case u := <-out:
		return u
	case <-time.After(5 * time.Second):
		t.Fatal("no update from the composer")
		return agent.Update{}
	}
}

func quiet(t *testing.T, out <-chan agent.Update) {
	t.Helper()
	select {
	case u := <-out:
		t.Fatalf("the composer handed %s, and nothing changed", u.Key)
	case <-time.After(50 * time.Millisecond):
	}
}

// Issue 1126 (ADR 0547 §2): the node composes its wanted modules and its own composition from its inputs on the bus once
// every input was read, hands each, then Caught; an input that changes nothing composed hands nothing; a module no want
// names any more is taken back; a node that cannot compose says so once and hands nothing.
func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
	b := newFakeBus()
	fp := delivery.SourcePrefix2 + strings.Repeat("a", 64)
	digest := delivery.ManifestDigest([]byte(lemurs))
	man, _ := delivery.EncodeManifest(delivery.Manifest{Module: "lemurs", JSON: lemurs})
	key, _ := delivery.ManifestsKey(digest)
	b.kv[delivery.ManifestsBucket+"/"+key] = man

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	env := b.env()
	// The vault's own: keeper's password sealed to n1's key, sealed again (other bytes) at every ask.
	var asks, generation int
	sealedAt := map[string]int{} // each box handed, by the generation it opens to
	env.Own = func(context.Context, map[string]any) (json.RawMessage, error) {
		b.mu.Lock()
		defer b.mu.Unlock()
		asks++
		sealedAt[fmt.Sprintf(`{"ask":%d}`, asks)] = generation + 1
		return json.RawMessage(fmt.Sprintf(`{"node":"n1","key":"sk","own":[{"module":"keeper","name":"password","origin":"made",`+
			`"generation":%d,"sealed":{"ask":%d}}]}`, generation+1, asks)), nil
	}
	secretChanged := make(chan []byte, 1)
	env.Hear = func(subject string) (<-chan []byte, func(), error) {
		if subject != VaultSecretChanged {
			t.Errorf("heard %s", subject)
		}
		return secretChanged, func() {}, nil
	}
	c := New(env)
	out, err := c.Watch(ctx, "n1.>")
	if err != nil {
		t.Fatal(err)
	}
	// oldie is placed and no want names it (a module at a build of the old form, nats on novox): the node runs it inside
	// its own composition, and composes it there again, even before the agent has applied anything (Running empty).
	oldie := `{"module":"oldie","version":"1","capabilities":["systemd"],"resources":[{"id":"service","type":"service","unit":"oldie.service","state":"running","boot":"enabled"}]}`
	odigest := delivery.ManifestDigest([]byte(oldie))
	oman, _ := delivery.EncodeManifest(delivery.Manifest{Module: "oldie", JSON: oldie})
	okey, _ := delivery.ManifestsKey(odigest)
	b.kv[delivery.ManifestsBucket+"/"+okey] = oman
	omod, _ := delivery.EncodeModule(delivery.Module{FP2: delivery.SourcePrefix2 + strings.Repeat("c", 64), Manifest: odigest, Repository: "novox/o", Commit: "c"})
	place, _ := delivery.EncodePlacement(delivery.Placement{Node: "n1", Decided: delivery.NodeDecided{Hub: true, Address: "10.77.0.1"},
		Modules: []delivery.PlacedModule{{Module: "lemurs"}, {Module: "oldie"}}})
	have, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}})
	mod, _ := delivery.EncodeModule(delivery.Module{FP2: fp, Manifest: digest, Repository: "novox/x", Commit: "c"})
	want, _ := delivery.EncodeWant(delivery.Want{FP: fp, Self: true})
	nodeWant, _ := delivery.EncodeNodeWant(delivery.NodeWant{Self: true})
	b.put(t, delivery.PlacementBucket, "placement.n1", place)
	b.put(t, delivery.HaveBucket, "n1.node", have)
	oldHave, _ := delivery.EncodeHave(delivery.Have{Build: "build-1", Phase: delivery.PhaseRunning, Health: delivery.HealthOK})
	b.put(t, delivery.HaveBucket, "n1.oldie", oldHave)
	b.put(t, delivery.ModulesBucket, "lemurs", mod)
	b.put(t, delivery.ModulesBucket, "oldie", omod)
	b.put(t, delivery.AssignerWantsBucket, "want.n1.lemurs", want)
	b.put(t, delivery.AssignerWantsBucket, "want.n1.node", nodeWant)
	for _, bucket := range []string{delivery.AssignerWantsBucket, delivery.PlacementBucket, delivery.HaveBucket, delivery.ModulesBucket, ResolvedBucket} {
		b.send(bucket, agent.Update{Caught: true})
	}
	got := map[string]bool{}
	var own delivery.Composed
	for {
		u := next(t, out)
		if u.Caught {
			break
		}
		c, err := delivery.DecodeComposed(u.Value)
		if err != nil {
			t.Fatalf("%s: %v", u.Key, err)
		}
		if c.Node != "n1" {
			t.Errorf("%s composed for %q", u.Key, c.Node)
		}
		if c.Module == "" {
			own = c
		}
		got[u.Key] = true
	}
	var ownText []string
	for _, r := range own.Resources {
		ownText = append(ownText, string(r))
	}
	if !strings.Contains(strings.Join(ownText, " "), `"id":"oldie.service"`) {
		t.Errorf("the node's own composition, composed before anything was applied, lacks oldie: %v", ownText)
	}
	// Issue 1126, 2026-10-27 09:21: every watch and get the composer makes is one the node's grant allows; a watch of
	// every have key (`>`) was refused by the bus and stopped every node agent.
	b.mu.Lock()
	for _, r := range b.reads {
		if !grantAllows(t, "n1", r[0], r[1]) {
			t.Errorf("the composer reads %s of %s, which the node's grant does not allow", r[1], r[0])
		}
	}
	b.mu.Unlock()
	lemursKey, _ := delivery.ComposedKey("n1", "lemurs", fp)
	nodeKey, _ := delivery.ComposedNodeKey("n1")
	if !got[lemursKey] || !got[nodeKey] || len(got) != 2 {
		b.mu.Lock()
		t.Logf("said: %v", b.said)
		b.mu.Unlock()
		t.Fatalf("handed %v, want %s and %s", got, lemursKey, nodeKey)
	}

	// Issue 1126, the switch made safe: a bus not answering for a moment (a reload of its accounts) is tried again and
	// refuses nothing; a pass whose run ended refuses nothing either.
	composeAgain = []time.Duration{time.Millisecond, time.Millisecond, time.Millisecond}
	fresh := func(get func(context.Context, string, string) ([]byte, error)) (*Composer, *[]string) {
		var refusals []string
		e := b.env()
		e.Get = get
		e.Refused = func(why string) { refusals = append(refusals, why) }
		c2 := New(e)
		c.mu.Lock()
		for k, v := range c.wants {
			c2.wants[k] = v
		}
		for k, v := range c.placements {
			c2.placements[k] = v
		}
		for k, v := range c.records {
			c2.records[k] = v
		}
		for k, v := range c.modules {
			c2.modules[k] = v
		}
		c.mu.Unlock()
		return c2, &refusals
	}
	for name, row := range map[string]struct {
		failures int // reads failing before the bus answers
		cancel   bool
		composes bool
	}{
		"the bus paused for one read":   {failures: 1, composes: true},
		"the bus paused past every try": {failures: 100},
		"the run ended mid-pass":        {failures: 100, cancel: true},
	} {
		failed := 0
		pctx, pcancel := context.WithCancel(ctx)
		c2, refusals := fresh(func(gctx context.Context, bucket, key string) ([]byte, error) {
			if failed < row.failures {
				failed++
				if row.cancel {
					pcancel()
					return nil, fmt.Errorf("the vault's own could not be asked: %w", context.Canceled)
				}
				return nil, fmt.Errorf("the revisions of %s in %s: %w", key, bucket, nats.ErrJetStreamNotEnabled)
			}
			b.mu.Lock()
			defer b.mu.Unlock()
			return b.kv[bucket+"/"+key], nil
		})
		handed := c2.compose(pctx)
		pcancel()
		switch {
		case row.composes && (len(handed) == 0 || len(*refusals) != 0):
			t.Errorf("%s: handed %d, refusals %q; want composed and nothing refused", name, len(handed), *refusals)
		case !row.composes && !row.cancel && (len(*refusals) != 1 || failed != 1+len(composeAgain)):
			t.Errorf("%s: refusals %q after %d reads; want one refusal after %d tries", name, *refusals, failed, 1+len(composeAgain))
		case row.cancel && len(*refusals) != 0:
			t.Errorf("%s: a pass whose run ended refused %q", name, *refusals)
		}
	}
	if !Transient(fmt.Errorf("the manifest of avahi: the revisions of x in mesh-delivery-indexer_manifests: nats: jetstream not enabled")) ||
		!Transient(fmt.Errorf("x: %w", nats.ErrNoResponders)) || Transient(fmt.Errorf("nats says ${dir:mesh-state}, and nats declares no directory")) ||
		Transient(fmt.Errorf("the revisions of it2.mailer.domain: %w", context.DeadlineExceeded)) {
		t.Error("Transient misjudges a bus that does not answer, or a refusal")
	}

	// A config value of another module, and a value changing nothing composed, hand nothing.
	b.put(t, ResolvedBucket, "n1.other.k", []byte(`{"value":1}`))
	quiet(t, out)

	// A want removed takes its composition back.
	removed, _ := delivery.EncodeWant(delivery.Want{Removed: true})
	b.put(t, delivery.AssignerWantsBucket, "want.n1.lemurs", removed)
	if u := next(t, out); u.Key != lemursKey || !u.Deleted {
		t.Fatalf("a removed want handed %+v", u)
	}

	// A want of a module whose manifest cannot be read: said once, nothing handed, what was handed stands.
	gone := delivery.SourcePrefix2 + strings.Repeat("b", 64)
	missing, _ := delivery.EncodeWant(delivery.Want{FP: gone, Self: true})
	b.put(t, delivery.AssignerWantsBucket, "want.n1.lemurs", missing)
	quiet(t, out)
	b.put(t, delivery.AssignerWantsBucket, "want.n1.lemurs", missing)
	quiet(t, out)
	b.mu.Lock()
	n := 0
	for _, s := range b.said {
		if strings.Contains(s, "lemurs is wanted at") {
			n++
		}
	}
	b.mu.Unlock()
	if n != 1 {
		t.Errorf("the fault was said %d times, want once: %v", n, b.said)
	}

	// An own secret comes sealed from the vault into the composition; sealed again at the same generation it is the value
	// first had, so nothing is handed again; a new generation is handed.
	b.put(t, delivery.AssignerWantsBucket, "want.n1.lemurs", want)
	if u := next(t, out); u.Key != lemursKey {
		t.Fatalf("lemurs wanted back handed %s", u.Key)
	}
	keeper := `{"module":"keeper","version":"1","own-secrets":{"password":"/var/lib/mesh/keeper/password"},` +
		`"resources":[{"id":"state","type":"directory","path":"/var/lib/mesh/keeper","mode":"0700"}]}`
	kdigest := delivery.ManifestDigest([]byte(keeper))
	kman, _ := delivery.EncodeManifest(delivery.Manifest{Module: "keeper", JSON: keeper})
	kkey, _ := delivery.ManifestsKey(kdigest)
	b.mu.Lock()
	b.kv[delivery.ManifestsBucket+"/"+kkey] = kman
	b.mu.Unlock()
	kmod, _ := delivery.EncodeModule(delivery.Module{FP2: fp, Manifest: kdigest, Repository: "novox/k", Commit: "c"})
	both, _ := delivery.EncodePlacement(delivery.Placement{Node: "n1", Decided: delivery.NodeDecided{Hub: true, Address: "10.77.0.1"},
		Modules: []delivery.PlacedModule{{Module: "lemurs"}, {Module: "oldie"}, {Module: "keeper"}}})
	b.put(t, delivery.ModulesBucket, "keeper", kmod)
	b.put(t, delivery.PlacementBucket, "placement.n1", both)
	b.put(t, delivery.AssignerWantsBucket, "want.n1.keeper", want)
	keeperKey, _ := delivery.ComposedKey("n1", "keeper", fp)
	var handed string
	for handed == "" {
		if u := next(t, out); u.Key == keeperKey {
			handed = string(u.Value)
		}
	}
	if !strings.Contains(handed, `{\"ask\":`) {
		t.Fatalf("keeper's composition does not carry the sealed value: %s", handed)
	}
	before := handed
	b.put(t, ResolvedBucket, "n1.other.k", []byte(`{"value":2}`))
	quiet(t, out)
	b.mu.Lock()
	generation++
	b.mu.Unlock()
	// The vault says the secret changed: the node composes again on that word, with no other input moving.
	secretChanged <- []byte(`{"name":"own/n1/keeper/password","generation":2}`)
	rotated := next(t, out)
	if rotated.Key != keeperKey {
		t.Fatalf("a new generation handed %s, want keeper again", rotated.Key)
	}

	// Review of L2 (round 2): rotated, then the node-engine restarts. The agent has applied nothing yet, so the box the
	// node runs comes from keeper's have (Held); the vault seals the value again, other bytes, and the box the have holds
	// is kept, since it opens to the same value: the composition does not move. The box from before the rotation opens
	// to the old value, so the vault's is taken.
	boxOf := func(raw string) string {
		c, err := delivery.DecodeComposed([]byte(raw))
		if err != nil {
			t.Fatal(err)
		}
		sealed := agent.SealedIn(c)
		if len(sealed) != 1 {
			t.Fatalf("keeper's composition carries %v", sealed)
		}
		return sealed[0]
	}
	// And the node's own key (review of L2, round 3, finding 1): a core box its own composition carries is held there, and
	// kept after a restart as a module's is, so the node's own composition does not move either.
	for _, row := range []struct {
		name, held, in string
		kept           bool
	}{{"the have holds the rotated value", boxOf(string(rotated.Value)), "keeper", true},
		{"the have holds the value before the rotation", boxOf(before), "keeper", false},
		{"the node's own key holds the rotated value", boxOf(string(rotated.Value)), agent.NodeModule, true}} {
		restarted := env
		restarted.Running = func() []delivery.Composed { return nil }
		restarted.Held = func(module string) []string {
			if module == row.in {
				return []string{row.held}
			}
			return nil
		}
		restarted.Unseal = func(box string) ([]byte, error) {
			b.mu.Lock()
			defer b.mu.Unlock()
			if g, ok := sealedAt[box]; ok {
				return []byte(fmt.Sprintf("password-%d", g)), nil
			}
			return nil, fmt.Errorf("not sealed to n1: %s", box)
		}
		again := New(restarted)
		again.wants = map[string]delivery.Want{"keeper": {FP: fp, Self: true}, agent.NodeModule: {}}
		rows, err := reader{c: again}.OwnSecrets(ctx, "n1")
		if err != nil || len(rows) != 1 {
			t.Fatalf("%s: own secrets %+v, %v", row.name, rows, err)
		}
		if kept := rows[0].Sealed == row.held; kept != row.kept {
			t.Errorf("%s: composed with %s (held %s), kept %v, want %v", row.name, rows[0].Sealed, row.held, kept, row.kept)
		}
	}

	// The pair credentials asked (ADR 0550 §5): the keys go to the vault's own with their role, a consumer's comes back as
	// a credential held, a provider's as the grant's copy, each sealed to n1's key, the value first had kept.
	var sent map[string]any
	pairEnv := b.env()
	pairEnv.Own = func(_ context.Context, args map[string]any) (json.RawMessage, error) {
		sent = args
		return json.RawMessage(`{"node":"n1","key":"sk","credentials":[` +
			`{"role":"consumer","provision":"pg","consumer":"n1","consumer_module":"web","provider":"n2","name":"pair/n2/pg/n1/web","generation":1,"sealed":{"c":1}},` +
			`{"role":"provider","provision":"s3","consumer":"n3","consumer_module":"files","provider":"n1","name":"provision/n1/s3","generation":2,"sealed":{"p":1}}]}`), nil
	}
	// The node runs the consumer's credential already, sealed otherwise (the controller sealed it, or the vault at
	// another ask): that box opens to the same value, so it is kept, and the composition does not move.
	pairEnv.Running = func() []delivery.Composed {
		return []delivery.Composed{{Module: "web", Resources: []json.RawMessage{json.RawMessage(`{"id":"web.secret-pg","sealed":"{\"run\":1}"}`)}}}
	}
	opened := map[string]string{`{"c":1}`: "pg-pw", `{"run":1}`: "pg-pw", `{"p":1}`: "s3-pw"}
	pairEnv.Unseal = func(box string) ([]byte, error) {
		if v, ok := opened[box]; ok {
			return []byte(v), nil
		}
		return nil, fmt.Errorf("not sealed to n1: %s", box)
	}
	var refusals []string
	pairEnv.Refused = func(why string) { refusals = append(refusals, why) }
	pc := New(pairEnv)
	consumer := []compose.CredentialKey{{Provision: "pg", Consumer: "n1", ConsumerModule: "web", Provider: "n2"}}
	provider := []compose.CredentialKey{{Provision: "s3", Consumer: "n3", ConsumerModule: "files", Provider: "n1"}}
	creds, issued, err := reader{c: pc}.Credentials(ctx, "n1", consumer, provider)
	if err != nil || len(creds) != 1 || len(issued) != 1 {
		t.Fatalf("credentials %+v issued %+v, %v", creds, issued, err)
	}
	if creds[0].State != compose.CredentialHeld || creds[0].Provider != "n2" || creds[0].Sealed != `{"run":1}` ||
		issued[0].Consumer != "n3" || issued[0].ForProvider != `{"p":1}` {
		t.Errorf("credentials %+v issued %+v", creds, issued)
	}
	// A node that cannot compose tells the agent once, and once again when it composes (the agent's haves say it).
	pc.refused("a stand-in for web's password")
	pc.refused("a stand-in for web's password")
	pc.refused("")
	if len(refusals) != 2 || refusals[0] == "" || refusals[1] != "" {
		t.Errorf("the agent was told %q", refusals)
	}
	if asked, _ := json.Marshal(sent["credentials"]); !strings.Contains(string(asked), `"role":"consumer"`) || !strings.Contains(string(asked), `"role":"provider"`) {
		t.Errorf("the vault was asked %s", asked)
	}
	// ADR 0559 §4: the node answers placement candidates in shadow, placed or not, applying nothing and handing no
	// composition: the same; a module the group adds, its secret not yet issued; a module removed; a placement it cannot
	// compose; a group whose keys went is no longer answered. What runs is never composed from a candidate: the
	// Composer does not read the state.
	for _, in := range New(env).inputs() {
		if in.bucket == delivery.PlacementCandidatesBucket {
			t.Fatal("the node's own composer reads placement candidates")
		}
	}
	sb := newPatternBus()
	trunk := delivery.Placement{Node: "n1", Decided: delivery.NodeDecided{Hub: true, Address: "10.77.0.1"}, Modules: []delivery.PlacedModule{{Module: "lemurs"}}}
	mods := map[string]delivery.Module{"lemurs": {FP2: fp, Manifest: digest, Repository: "novox/x", Commit: "c"},
		"keeper": {FP2: fp, Manifest: kdigest, Repository: "novox/k", Commit: "c"}}
	readOnly := &fakeBus{kv: map[string][]byte{delivery.ManifestsBucket + "/" + key: man, delivery.ManifestsBucket + "/" + kkey: kman}}
	senv := readOnly.env()
	senv.Watch = sb.watch
	senv.Own = func(context.Context, map[string]any) (json.RawMessage, error) {
		return json.RawMessage(`{"node":"n1","key":"sk","own":[]}`), nil // nothing issued for keeper: it is not placed here
	}
	// What n1 runs: lemurs and its own composition, as its composer composes them from the trunk.
	running, _, err := frombus.Compose(ctx, fixedReader{placements: map[string]delivery.Placement{"n1": trunk},
		records: map[string]delivery.NodeRecord{"n1": {Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}},
		modules: mods, manifests: map[string][]byte{digest: []byte(lemurs), kdigest: []byte(keeper)}}, frombus.Mesh{}, frombus.Scope{Node: "n1"}, map[string]string{"lemurs": fp})
	if err != nil {
		t.Fatal(err)
	}
	senv.Running = func() []delivery.Composed {
		var out []delivery.Composed
		for _, c := range running {
			out = append(out, c)
		}
		return out
	}
	// The controller's compositions at n1's wants, in n1's composition bucket: what a head deleting its placement
	// returns it to (the seed's revert); here the ones it runs.
	composedBucket, _ := delivery.ComposedBucketFor("n1")
	for key, c := range running {
		raw, err := delivery.EncodeComposed(c)
		if err != nil {
			t.Fatal(err)
		}
		readOnly.mu.Lock()
		readOnly.kv[composedBucket+"/"+key] = raw
		readOnly.mu.Unlock()
	}
	answered := make(chan map[string]delivery.CandidateAnswer, 8)
	sh := NewShadow(senv, func(_ context.Context, a map[string]delivery.CandidateAnswer) { answered <- a })
	sctx, scancel := context.WithCancel(ctx)
	defer scancel()
	// Review of L2b, round 1, finding 2: the candidates' state is not there when the node starts (a delivery before the
	// indexer made it): the shadow tries again, and answers once it appears.
	sb.missing(delivery.PlacementCandidatesBucket, true)
	go sh.Keep(sctx, func(int) time.Duration { return 5 * time.Millisecond })
	time.Sleep(30 * time.Millisecond)
	sb.missing(delivery.PlacementCandidatesBucket, false)
	sb.made(7)
	rec, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}})
	tp, _ := delivery.EncodePlacement(trunk)
	sb.put(delivery.PlacementBucket, "placement.n1", tp, 1)
	sb.put(delivery.HaveBucket, "n1.node", rec, 1)
	for name, m := range mods {
		raw, _ := delivery.EncodeModule(m)
		sb.put(delivery.ModulesBucket, name, raw, 1)
	}
	wantLemurs, _ := delivery.WantKey("n1", "lemurs")
	var lemursDigest string
	for _, c := range running {
		if c.Module == "lemurs" {
			lemursDigest, _ = delivery.ComposedDigestOf(c)
		}
	}
	wantRaw, err := delivery.EncodeWant(delivery.Want{FP: fp, Composed: lemursDigest})
	if err != nil {
		t.Fatal(err)
	}
	sb.put(delivery.AssignerWantsBucket, wantLemurs, wantRaw, 1)
	wantNode, _ := delivery.WantNodeKey("n1")
	nodeRaw, _ := delivery.EncodeWant(delivery.Want{})
	sb.put(delivery.AssignerWantsBucket, wantNode, nodeRaw, 1)
	candidate := func(group string, rev uint64, modules ...string) {
		p := trunk
		p.Modules = nil
		for _, m := range modules {
			p.Modules = append(p.Modules, delivery.PlacedModule{Module: m})
		}
		raw, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Group: group,
			Pull: delivery.Head{Repository: delivery.CatalogueRepository, Number: 1, Head: "h-" + group}, Placement: &p})
		if err != nil {
			t.Fatal(err)
		}
		k, _ := delivery.PlacementCandidateKey(group, "n1")
		sb.put(delivery.PlacementCandidatesBucket, k, raw, rev)
	}
	candidate("same", 3, "lemurs")
	candidate("adds", 4, "lemurs", "keeper")
	candidate("drops", 5)
	candidate("ghost", 6, "lemurs", "ghost")
	// A head deleting n1's placement file (the seed's revert, mesh-catalog #526): n1 returns to the controller's
	// compositions, answered as such, not as every module removed.
	if raw, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Group: "deletes",
		Pull: delivery.Head{Repository: delivery.CatalogueRepository, Number: 1, Head: "h-deletes"}}); err != nil {
		t.Fatal(err)
	} else {
		k, _ := delivery.PlacementCandidateKey("deletes", "n1")
		sb.put(delivery.PlacementCandidatesBucket, k, raw, 7)
	}
	sb.caught()
	var answers map[string]delivery.CandidateAnswer
	waitFor := func(why string, ok func(map[string]delivery.CandidateAnswer) bool) {
		t.Helper()
		deadline := time.After(10 * time.Second)
		for answers == nil || !ok(answers) {
			select {
			case answers = <-answered:
			case <-deadline:
				t.Fatalf("%s: last answers %+v", why, answers)
			}
		}
	}
	waitFor("every group answered", func(a map[string]delivery.CandidateAnswer) bool { return len(a) == 5 })
	// Every watch and get the shadow makes is one the node's grant allows (issue 1126, 2026-10-27 09:21).
	sb.mu.Lock()
	for _, w := range sb.watches {
		if !grantAllows(t, "n1", w.bucket, w.pattern) {
			t.Errorf("the shadow watches %s of %s, which the node's grant does not allow", w.pattern, w.bucket)
		}
	}
	sb.mu.Unlock()
	readOnly.mu.Lock()
	for _, r := range readOnly.reads {
		if !grantAllows(t, "n1", r[0], r[1]) {
			t.Errorf("the shadow reads %s of %s, which the node's grant does not allow", r[1], r[0])
		}
	}
	readOnly.mu.Unlock()
	for group, want := range map[string]func(delivery.CandidateAnswer) bool{
		"same": func(a delivery.CandidateAnswer) bool {
			return a.Verdict == delivery.CandidateSame && a.Head == "h-same" && a.Revision == 3 && len(a.Same) == 2
		},
		"adds": func(a delivery.CandidateAnswer) bool {
			return a.Verdict == delivery.CandidateDiffers && len(a.Added) == 1 && a.Added[0].Module == "keeper" &&
				len(a.Added[0].Unissued) == 1 && len(a.Refusals) == 0
		},
		"drops": func(a delivery.CandidateAnswer) bool {
			return a.Verdict == delivery.CandidateDiffers && len(a.Removed) == 1 && a.Removed[0] == "lemurs"
		},
		"ghost": func(a delivery.CandidateAnswer) bool {
			return a.Verdict == delivery.CandidateRefused && a.Refused != ""
		},
		"deletes": func(a delivery.CandidateAnswer) bool {
			return a.Verdict == delivery.CandidateSame && a.Returns == delivery.ReturnsToController && a.Revision == 7 &&
				len(a.Removed) == 0 && fmt.Sprint(a.Same) == "[lemurs node]"
		},
	} {
		if !want(answers[group]) {
			t.Errorf("%s: %+v", group, answers[group])
		}
	}
	// The node's own key changing (its answers written) composes nothing again; a group's key deleted drops its answer.
	sb.put(delivery.HaveBucket, "n1.node", rec, 2)
	select {
	case a := <-answered:
		t.Errorf("answers handed again for nothing changed: %+v", a)
	case <-time.After(100 * time.Millisecond):
	}
	k, _ := delivery.PlacementCandidateKey("same", "n1")
	sb.del(delivery.PlacementCandidatesBucket, k)
	waitFor("the deleted group's answer stays", func(a map[string]delivery.CandidateAnswer) bool { _, ok := a["same"]; return !ok && len(a) == 4 })

	// Review of L2b, round 2: a watch ends while the link stays; a group's key is deleted before the shadow watches
	// again. The new run reads afresh, so that group is no longer answered.
	sb.missing(delivery.PlacementCandidatesBucket, true)
	sb.end(delivery.ModulesBucket)
	dropsKey, _ := delivery.PlacementCandidateKey("drops", "n1")
	sb.del(delivery.PlacementCandidatesBucket, dropsKey)
	time.Sleep(30 * time.Millisecond)
	sb.missing(delivery.PlacementCandidatesBucket, false)
	waitFor("a group deleted while the shadow was down is still answered", func(a map[string]delivery.CandidateAnswer) bool {
		_, ok := a["drops"]
		return !ok && len(a) == 3
	})

	// Whose compositions a node applies at a start (review of the shadow, round 1): only a read that succeeds chooses; a
	// failing read is tried again, said once, and chooses neither, even when the link ends.
	for name, row := range map[string]struct {
		answers []error // per try; nil then the value
		value   []byte
		end     bool
		placed  bool
		tries   int
		quiet   bool // read and known: nothing to say
	}{
		"placed after two failures": {answers: []error{fmt.Errorf("nats: timeout"), fmt.Errorf("nats: timeout")}, value: []byte(`{}`), placed: true, tries: 3},
		"no key":                    {placed: false, tries: 1},
		"no placement state yet":    {answers: []error{fmt.Errorf("the bucket %s: %w", delivery.PlacementBucket, nats.ErrBucketNotFound)}, placed: false, tries: 1, quiet: true},
		"failing until the link ends": {answers: []error{fmt.Errorf("jetstream not ready"), fmt.Errorf("jetstream not ready"),
			fmt.Errorf("jetstream not ready")}, end: true, tries: 3},
	} {
		lctx, lcancel := context.WithCancel(ctx)
		tries := 0
		var said []string
		get := func(context.Context, string, string) ([]byte, error) {
			tries++
			if tries <= len(row.answers) {
				if row.end && tries == len(row.answers) {
					lcancel()
				}
				return nil, row.answers[tries-1]
			}
			return row.value, nil
		}
		placed, err := Placed(lctx, get, "n1", func(int) time.Duration { return time.Millisecond }, func(s string) { said = append(said, s) })
		lcancel()
		if placed != row.placed || tries != row.tries || (err != nil) != row.end || (len(row.answers) > 0 && !row.quiet) != (len(said) == 1) {
			t.Errorf("%s: placed %v, err %v after %d reads, said %q", name, placed, err, tries, said)
		}
	}

	// The way back, row by row (review of the shadow, round 1): the controller's compositions at n1's wants compared
	// with what runs; one differing is answered so; one the controller holds none of refuses, without returning.
	var runsList []delivery.Composed
	for _, c := range running {
		runsList = append(runsList, c)
	}
	lemursKey, _ = delivery.ComposedKey("n1", "lemurs", fp)
	changed := running[lemursKey]
	changed.Resources = []json.RawMessage{json.RawMessage(`{"id":"lemurs.service","type":"service","module":"lemurs","unit":"lemurs.service","state":"stopped","boot":"disabled"}`)}
	changedRaw, _ := delivery.EncodeComposed(changed)
	for name, row := range map[string]struct {
		lemurs []byte
		check  func(delivery.CandidateAnswer) bool
	}{
		"the controller's differs": {changedRaw, func(a delivery.CandidateAnswer) bool {
			return a.Verdict == delivery.CandidateDiffers && a.Returns == delivery.ReturnsToController &&
				len(a.Differs) == 1 && a.Differs[0].Module == "lemurs" && fmt.Sprint(a.Same) == "[node]"
		}},
		"the controller holds none": {nil, func(a delivery.CandidateAnswer) bool {
			return a.Verdict == delivery.CandidateRefused && a.Returns == "" && strings.Contains(a.Refused, "lemurs: the controller holds no composition")
		}},
	} {
		get := func(_ context.Context, bucket, key string) ([]byte, error) {
			if key == lemursKey {
				return row.lemurs, nil
			}
			readOnly.mu.Lock()
			defer readOnly.mu.Unlock()
			return readOnly.kv[bucket+"/"+key], nil
		}
		a := returning(ctx, get, "n1", map[string]delivery.Want{"lemurs": {FP: fp}, agent.NodeModule: {}}, runsList, "h", 1)
		if !row.check(a) {
			t.Errorf("%s: %+v", name, a)
		}
		if _, err := delivery.EncodeHaveNode(delivery.HaveNode{Candidates: map[string]delivery.CandidateAnswer{"g": a}}); err != nil {
			t.Errorf("%s: the answer is not one a have holds: %v", name, err)
		}
	}

	// Issue 1126, the seed's review: a module n1 runs inside its own composition (nats, at a build of the old form) is
	// composed there again, not at the trunk's build as one added; lemurs at the build it runs, keeper (added) at the
	// trunk's.
	inOwnRunning := []delivery.Composed{{Node: "n1", Module: "lemurs", FP: fp}}
	other := "src2:" + strings.Repeat("9", 64)
	gotFps := shadowBuilds(inOwnRunning, delivery.Placement{Node: "n1", Modules: []delivery.PlacedModule{{Module: "lemurs"}, {Module: "nats"}, {Module: "keeper"}}},
		map[string]delivery.Module{"lemurs": {FP2: other}, "nats": {FP2: other}, "keeper": {FP2: other}}, []string{"nats"})
	if fmt.Sprint(gotFps) != fmt.Sprint(map[string]string{"lemurs": fp, "keeper": other}) {
		t.Errorf("shadow builds %v; want lemurs at what runs, keeper at the trunk's, nats at none (folded)", gotFps)
	}

	// Issue 1126, the seed's refusals: what a node composes from its builds and reads of a provider elsewhere. A build
	// is the builder's key; a templates archive is fetched once per digest and refused when it is not that digest; another
	// node's served key is read alone, as the node's grant allows it (buscompose Principal.ServedReads).
	built, _ := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildDone, Manifest: digest})
	buildKey, _ := delivery.BuilderBuildKey("lemurs", fp)
	b.mu.Lock()
	b.kv[delivery.BuilderBuildsBucket+"/"+buildKey] = built
	b.kv[ResolvedBucket+"/n2.mailer.domain"] = []byte(`{"value":"example.org"}`)
	b.reads = nil
	b.mu.Unlock()
	fetched := 0
	archive := []byte("an archive")
	sum := sha256.Sum256(archive)
	env.Fetch = func(context.Context, string) ([]byte, error) { fetched++; return archive, nil }
	r := reader{c: New(env)}
	if got, err := r.Build(ctx, "lemurs", fp); err != nil || got == nil || got.Manifest != digest {
		t.Errorf("the build of lemurs: %+v, %v", got, err)
	}
	if got, err := r.Build(ctx, "lemurs", other); err != nil || got != nil {
		t.Errorf("no build at another fingerprint: %+v, %v", got, err)
	}
	for range 2 {
		if got, err := r.Archive(ctx, "http://store/v2/x/files/blobs/d", "sha256:"+hex.EncodeToString(sum[:])); err != nil || string(got) != "an archive" {
			t.Errorf("the archive: %q, %v", got, err)
		}
	}
	if _, err := r.Archive(ctx, "http://store/v2/x/files/blobs/e", "sha256:"+strings.Repeat("0", 64)); err == nil || fetched != 2 {
		t.Errorf("fetched %d times (want once per digest), and an archive of another digest: %v", fetched, err)
	}
	if got, err := r.Served(ctx, "n2", "mailer", []string{"domain"}); err != nil || fmt.Sprint(got) != "map[domain:example.org]" {
		t.Errorf("what n2's mailer serves: %v, %v", got, err)
	}
	served, err := buscompose.PermissionsFor(buscompose.Principal{Kind: buscompose.KindNode, Node: "n1", ServedReads: []string{"n2.mailer.domain"}})
	if err != nil {
		t.Fatal(err)
	}
	b.mu.Lock()
	for _, read := range b.reads {
		ok := grantAllows(t, "n1", read[0], read[1])
		if read[0] == ResolvedBucket {
			ok = slices.Contains(served.Publish, "$JS.API.CONSUMER.CREATE.KV_"+ResolvedBucket+".*.$KV."+ResolvedBucket+"."+read[1])
		}
		if !ok {
			t.Errorf("the composer reads %s of %s, which the node's grant does not allow", read[1], read[0])
		}
	}
	b.mu.Unlock()
}

// patternBus is a bus whose every watch gets the updates of its bucket matching its pattern, as the bus's own watches
// do: two watches of one bucket each see what matches theirs.
type patternBus struct {
	mu      sync.Mutex
	watches []patternWatch
	absent  map[string]bool
	// kv is each bucket's keys now; once replay is set, a new watch gets those matching it and then Caught, as a watch
	// of the bus's own does.
	kv     map[string]map[string]agent.Update
	replay bool
}

// end ends every watch of a bucket, as a watch the bus drops, and from then on each new watch replays what is there.
func (b *patternBus) end(bucket string) {
	b.mu.Lock()
	defer b.mu.Unlock()
	b.replay = true
	kept := b.watches[:0]
	for _, w := range b.watches {
		if w.bucket == bucket {
			close(w.ch)
			continue
		}
		kept = append(kept, w)
	}
	b.watches = kept
}

// missing makes a bucket's watch fail, as one the bus holds no state for, or work again.
func (b *patternBus) missing(bucket string, gone bool) {
	b.mu.Lock()
	defer b.mu.Unlock()
	if b.absent == nil {
		b.absent = map[string]bool{}
	}
	b.absent[bucket] = gone
}

type patternWatch struct {
	bucket, pattern string
	ch              chan agent.Update
}

func newPatternBus() *patternBus { return &patternBus{} }

func (b *patternBus) watch(_ context.Context, bucket, pattern string) (<-chan agent.Update, error) {
	b.mu.Lock()
	defer b.mu.Unlock()
	if b.absent[bucket] {
		return nil, fmt.Errorf("nats: bucket not found")
	}
	w := patternWatch{bucket: bucket, pattern: pattern, ch: make(chan agent.Update, 64)}
	b.watches = append(b.watches, w)
	if b.replay {
		for key, u := range b.kv[bucket] {
			if matches(pattern, key) {
				w.ch <- u
			}
		}
		w.ch <- agent.Update{Caught: true}
	}
	return w.ch, nil
}

func matches(pattern, key string) bool {
	p, k := strings.Split(pattern, "."), strings.Split(key, ".")
	for i, tok := range p {
		if tok == ">" {
			return len(k) > i
		}
		if i >= len(k) || (tok != "*" && tok != k[i]) {
			return false
		}
	}
	return len(p) == len(k)
}

func (b *patternBus) send(bucket string, u agent.Update) {
	b.mu.Lock()
	defer b.mu.Unlock()
	if !u.Caught {
		if b.kv == nil {
			b.kv = map[string]map[string]agent.Update{}
		}
		if b.kv[bucket] == nil {
			b.kv[bucket] = map[string]agent.Update{}
		}
		if u.Deleted {
			delete(b.kv[bucket], u.Key)
		} else {
			b.kv[bucket][u.Key] = u
		}
	}
	for _, w := range b.watches {
		if w.bucket == bucket && (u.Caught || matches(w.pattern, u.Key)) {
			select {
			case w.ch <- u:
			default: // a watch of a run that ended reads no more
			}
		}
	}
}

func (b *patternBus) put(bucket, key string, value []byte, rev uint64) {
	b.send(bucket, agent.Update{Key: key, Value: value, Revision: rev})
}

func (b *patternBus) del(bucket, key string) { b.send(bucket, agent.Update{Key: key, Deleted: true}) }

// made waits until n watches are made, so nothing put is lost.
func (b *patternBus) made(n int) {
	for {
		b.mu.Lock()
		got := len(b.watches)
		b.mu.Unlock()
		if got >= n {
			return
		}
		time.Sleep(5 * time.Millisecond)
	}
}

// caught says every watch caught up.
func (b *patternBus) caught() {
	b.mu.Lock()
	defer b.mu.Unlock()
	for _, w := range b.watches {
		w.ch <- agent.Update{Caught: true}
	}
}

// fixedReader is frombus's Reader over values held, for composing what a node runs.
type fixedReader struct {
	placements map[string]delivery.Placement
	records    map[string]delivery.NodeRecord
	modules    map[string]delivery.Module
	manifests  map[string][]byte
}

func (r fixedReader) Placements(context.Context) (map[string]delivery.Placement, error) {
	return r.placements, nil
}
func (r fixedReader) Records(context.Context) (map[string]delivery.NodeRecord, error) {
	return r.records, nil
}
func (r fixedReader) Modules(context.Context) (map[string]delivery.Module, error) {
	return r.modules, nil
}
func (r fixedReader) Manifest(_ context.Context, d string) ([]byte, error) {
	if raw, ok := r.manifests[d]; ok {
		return raw, nil
	}
	return nil, fmt.Errorf("no manifest %s", d)
}
func (r fixedReader) Values(context.Context, string, string) (map[string]any, error) { return nil, nil }
func (r fixedReader) OwnSecrets(context.Context, string) ([]compose.OwnSecretRow, error) {
	return nil, nil
}
func (r fixedReader) Credentials(context.Context, string, []compose.CredentialKey, []compose.CredentialKey) ([]compose.Credential, []compose.Issued, error) {
	return nil, nil, nil
}
func (r fixedReader) Build(context.Context, string, string) (*delivery.BuilderBuild, error) {
	return nil, nil
}
func (r fixedReader) Archive(_ context.Context, url, _ string) ([]byte, error) {
	return nil, fmt.Errorf("no archive %s", url)
}
func (r fixedReader) Served(context.Context, string, string, []string) (map[string]any, error) {
	return nil, nil
}

// TestAHaveSaysWhetherItsModuleRunsInsideTheNodesOwn feeds takeModuleHave a module's have and reads whether the node
// counts the module as run inside its own composition (issue 1126, review round 2): a failed have that names nothing
// it runs (NotComposed before the module ever ran) is not, so the next composition does not take the module from its
// manifest past its build.
func TestAHaveSaysWhetherItsModuleRunsInsideTheNodesOwn(t *testing.T) {
	for _, row := range []struct {
		name string
		have delivery.Have
		in   bool
	}{
		{"failed, naming nothing it runs", delivery.Have{Phase: delivery.PhaseFailed, Health: delivery.HealthFailed,
			Detail: "the node cannot compose it, and keeps running what it ran: no build"}, false},
		{"failed at a build fingerprint", delivery.Have{FP: "build-0123abcd", Composed: "sha256:" + strings.Repeat("c", 64), Phase: delivery.PhaseFailed,
			Health: delivery.HealthFailed, Detail: "the node cannot compose it, and keeps running what it ran: no build"}, true},
		{"running with no fingerprint (mesh-wireguard)", delivery.Have{Phase: delivery.PhaseRunning, Health: delivery.HealthOK}, true},
	} {
		t.Run(row.name, func(t *testing.T) {
			c := New(Env{Node: "n1"})
			key, err := delivery.HaveKey("n1", "web")
			if err != nil {
				t.Fatal(err)
			}
			value, err := delivery.EncodeHave(row.have)
			if err != nil {
				t.Fatal(err)
			}
			c.takeModuleHave(agent.Update{Key: key, Value: value})
			if c.inOwn["web"] != row.in {
				t.Errorf("web counted inside the node's own composition: %v, want %v", c.inOwn["web"], row.in)
			}
		})
	}
}