mesh-host/cmd/mesh-host/nodeagent_integration_test.go

//go:build integration

package main

import (
	"context"
	"crypto/ed25519"
	"encoding/json"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"testing"
	"time"

	"git.novox.be/novox/mesh-sdk/go/buscompose"
	"git.novox.be/novox/mesh-sdk/go/delivery"
	"github.com/nats-io/nats-server/v2/server"
	"github.com/nats-io/nats.go"
	"github.com/novox/mesh-host/internal/agent"
	"github.com/novox/mesh-host/internal/link"
	"github.com/novox/mesh-host/internal/nodegrant"
	"github.com/novox/mesh-host/internal/selfcompose"
)

// The node agent whole, against a real bus with JetStream started in the test: the wants bucket, the node's composition
// bucket and the have bucket, a session on the bus, and the real apply of a module's composition into a directory of the
// test. R6-A AC1, AC5 and AC9 (novox/hq ADR 0539).

const itNode = "it1"

type itBed struct {
	t    *testing.T
	conn *nats.Conn
	js   nats.JetStreamContext
	// nodeConn and nodeJS are the node's own, under its real grant (the grants part's, nodegrant.For): the agent runs
	// on them, so a watch or a read the grant does not allow fails here as it fails on the bus (issue 1126, 2026-10-27
	// 09:21: a watch of every have key stopped every node agent). conn is the test's, unrestricted, for setting up.
	nodeConn *nats.Conn
	nodeJS   nats.JetStreamContext
	wants    nats.KeyValue
	composed nats.KeyValue
	haves    nats.KeyValue
	dir      string
	opts     options
	session  *link.Session
	controll ed25519.PrivateKey
	said     []string
	// refusals are the bus's asynchronous errors on the node's connection: each permissions violation it met.
	refusals []string
	mu       sync.Mutex
}

// aServer is an embedded bus with JetStream and two users: the test's, unrestricted, and the node's, granted exactly what
// the grants part gives a node (nodegrant.For); it answers a connection of each.
func aServer(t *testing.T, perms nodegrant.Permissions, refused func(error)) (admin, node *nats.Conn) {
	t.Helper()
	granted := &server.Permissions{Publish: &server.SubjectPermission{Allow: perms.Publish, Deny: perms.PublishDeny},
		Subscribe: &server.SubjectPermission{Allow: perms.Subscribe}}
	if perms.AllowResponses {
		granted.Response = &server.ResponsePermission{MaxMsgs: 1, Expires: buscompose.ResponseTTL}
	}
	nodeUser := buscompose.Principal{Kind: buscompose.KindNode, Node: itNode}.Username()
	srv, err := server.NewServer(&server.Options{Port: -1, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true,
		Users: []*server.User{{Username: "test", Password: "test"}, {Username: nodeUser, Password: "node", Permissions: granted}}})
	if err != nil {
		t.Fatal(err)
	}
	go srv.Start()
	if !srv.ReadyForConnections(10 * time.Second) {
		t.Fatal("the embedded server did not start")
	}
	t.Cleanup(srv.Shutdown)
	if admin, err = nats.Connect(srv.ClientURL(), nats.UserInfo("test", "test")); err != nil {
		t.Fatal(err)
	}
	t.Cleanup(admin.Close)
	if node, err = nats.Connect(srv.ClientURL(), nats.UserInfo(nodeUser, "node"), nats.CustomInboxPrefix("_INBOX."+nodeUser),
		nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) { refused(err) })); err != nil {
		t.Fatal(err)
	}
	t.Cleanup(node.Close)
	return admin, node
}

func newItBed(t *testing.T) *itBed {
	t.Helper()
	return newItBedAs(t, nodegrant.For(itNode))
}

// newItBedAs is newItBed with the node's connection under the grant perms, a node's.
func newItBedAs(t *testing.T, perms nodegrant.Permissions) *itBed {
	t.Helper()
	// Not skipped off Arch: a package manager that answers is put on PATH, so the apply of files runs on any build node.
	bin := t.TempDir()
	if err := os.WriteFile(filepath.Join(bin, "pacman"), []byte("#!/bin/sh\nexit 0\n"), 0o755); err != nil {
		t.Fatal(err)
	}
	t.Setenv("PATH", bin+string(os.PathListSeparator)+os.Getenv("PATH"))
	was := builtFor
	builtFor = "arch"
	t.Cleanup(func() { builtFor = was })

	b := &itBed{t: t, dir: t.TempDir()}
	b.conn, b.nodeConn = aServer(t, perms, func(err error) {
		b.mu.Lock()
		b.refusals = append(b.refusals, err.Error())
		b.mu.Unlock()
	})
	var err error
	if b.js, err = b.conn.JetStream(); err != nil {
		t.Fatal(err)
	}
	if b.nodeJS, err = b.nodeConn.JetStream(); err != nil {
		t.Fatal(err)
	}
	composedBucket, _ := delivery.ComposedBucketFor(itNode)
	for _, c := range []struct {
		name    string
		into    *nats.KeyValue
		history uint8
	}{{delivery.AssignerWantsBucket, &b.wants, 8}, {composedBucket, &b.composed, delivery.ComposedHistory},
		{delivery.HaveBucket, &b.haves, delivery.HaveHistory}} {
		kv, err := b.js.CreateKeyValue(&nats.KeyValueConfig{Bucket: c.name, History: c.history})
		if err != nil {
			t.Fatal(err)
		}
		*c.into = kv
	}
	b.opts = options{state: filepath.Join(b.dir, "state", "state.json"), timeout: 2 * time.Second}
	if err := os.MkdirAll(filepath.Dir(b.opts.state), 0o755); err != nil {
		t.Fatal(err)
	}
	_, b.controll, _ = ed25519.GenerateKey(nil)
	return b
}

// start runs the node agent under a session on the bus.
func (b *itBed) start() (stop func()) {
	b.t.Helper()
	b.session = &link.Session{Membership: link.Membership{Node: itNode}, Timeout: 2 * time.Second,
		Say: b.say, Keys: func() link.RecordKeys { return link.RecordKeys{} }}
	b.session.Attach(context.Background(), link.OverNATS{Conn: b.nodeConn, JS: b.nodeJS, Node: itNode})
	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan struct{})
	go func() {
		defer close(done)
		controller := b.controll.Public().(ed25519.PublicKey)
		nodeAgent(ctx, b.opts, controller, newMeshTrust(), b.session, nil, b.say, func() {},
			&agentNow{}, func(context.Context, *agent.Agent) {})
	}()
	return func() { cancel(); <-done }
}

func (b *itBed) say(line string) {
	b.mu.Lock()
	b.said = append(b.said, line)
	b.mu.Unlock()
	b.t.Log(line)
}

func fpIt(c string) string { return delivery.SourcePrefix2 + strings.Repeat(c, 64) }

func fpDigest(c string) string { return "sha256:" + strings.Repeat(c, 64) }

func (b *itBed) want(module, fp, digest, backup string) {
	b.t.Helper()
	raw, err := delivery.EncodeWant(delivery.Want{FP: fp, Composed: digest, Backup: backup})
	if err != nil {
		b.t.Fatal(err)
	}
	key, _ := delivery.WantKey(itNode, module)
	if _, err := b.wants.Put(key, raw); err != nil {
		b.t.Fatal(err)
	}
}

func (b *itBed) file(module string) string {
	got, _ := os.ReadFile(filepath.Join(b.dir, module+".conf"))
	return string(got)
}

func (b *itBed) have(module string) delivery.Have {
	key, _ := delivery.HaveKey(itNode, module)
	e, err := b.haves.Get(key)
	if err != nil {
		return delivery.Have{}
	}
	h, _ := delivery.DecodeHave(e.Value())
	return h
}

func within(t *testing.T, d time.Duration, what string, ok func() bool) {
	t.Helper()
	deadline := time.Now().Add(d)
	for !ok() {
		if time.Now().After(deadline) {
			t.Fatal(what)
		}
		time.Sleep(10 * time.Millisecond)
	}
}

// Issue 1126 (ADR 0547 §2, the milestone's H8): the node agent started whole against a real bus, once the catalogue
// places its node (the placement written after it started), composes a module from its want, the indexer's manifest, its record and its mesh-config value, and applies it,
// with no key in the controller's compositions; a changed config value is composed again on its event and applied as a
// revision of what runs. And a catalogue pull request's placement candidate equal to its placement is answered "same"
// in the node's own key, and nothing is applied for it (ADR 0559 §4, subtask 10 of the milestone).
//
// And (issue 1126, ADR 0560) it reads, through its own session under the grant the grants part composes for it
// (nodegrant.For with the served key, as the grants part grants it), the build of a placed module that declares one and the one config value
// a provider on another node serves from (a mail domain); under a grant without that served key the bus refuses the
// read and the node composes nothing.
func TestNatsANodeThePlacementNamesComposesAndAppliesItsOwn(t *testing.T) {
	t.Run("without the served key the bus refuses the read", func(t *testing.T) {
		b := newItBed(t)
		stop, _ := h8(t, b, false)
		defer stop()
		// 15 s, above link.historyWithin (10 s): the bus's refusal of a publish reaches only the connection's error
		// handler (nats.go fails no request on it), so the refused read ends at its own bound, then the node says it.
		within(t, 15*time.Second, "the bus did not refuse the served key's read, or the node did not say it", func() bool {
			b.mu.Lock()
			defer b.mu.Unlock()
			return strings.Contains(strings.Join(b.refusals, "\n"), "Permissions Violation for Publish to \"$JS.API.CONSUMER.CREATE.KV_"+
				selfcompose.ResolvedBucket) && strings.Contains(strings.Join(b.refusals, "\n"), ".it2.mailer.domain") &&
				strings.Contains(strings.Join(b.said, "\n"), "what mailer serves on it2 (domain)")
		})
		if b.file("web") != "" {
			t.Errorf("composed and applied under a grant without the served key: %q", b.file("web"))
		}
	})
	b := newItBedAs(t, nodegrant.For(itNode, "it2.mailer.domain"))
	stop, st := h8(t, b, true)
	defer stop()
	placement, resolved, candidates := st.placement, st.resolved, st.candidates
	hkey, place, fp := st.hkey, st.place, st.fp
	within(t, 5*time.Second, "the node did not compose and apply its own module once placed", func() bool { return b.file("web") == "greeting=bonjour\n" })
	// ADR 0565 §6 (1953): the node's engine writes its own key of the state packages, under the grant the grants part
	// composes for it, with each package its compositions declare and the module declaring it; once, while nothing changes.
	wantPkgs := `{"node":"` + itNode + `","packages":[{"package":"web-tools","modules":["web"],"absent":false}]}`
	within(t, 5*time.Second, "the node did not write its key of the state packages", func() bool {
		e, err := st.packages.Get(itNode)
		return err == nil && string(e.Value()) == wantPkgs
	})
	written, _ := st.packages.Get(itNode)
	h8rest(t, b, resolved, candidates, hkey, place, fp)
	// web composed again for its changed config value declares the same packages: the key is not written again.
	if e, _ := st.packages.Get(itNode); e == nil || written == nil || e.Revision() != written.Revision() {
		t.Errorf("the node's packages key was written again with no package changed: %v, was %v", e, written)
	}
	t.Run("one answer per candidate revision", func(t *testing.T) { h8once(t, b, resolved, candidates, place) })
	h8back(t, b, placement, resolved, candidates, hkey)
}

// h8once is issue 1126, 2026-10-28 (shanks answered probe/1126-seed-7 at revision 84 every 13-17 s, its bus flooded):
// a candidate revision is answered once. The node's own answer written, a module's have whose health words alone move
// (each look rewrites them) and a config value written again with the value it held answer nothing again; the
// candidate's next revision is answered once.
func h8once(t *testing.T, b *itBed, resolved, candidates nats.KeyValue, place []byte) {
	answering := func() int {
		b.mu.Lock()
		defer b.mu.Unlock()
		n := 0
		for _, l := range b.said {
			if strings.HasPrefix(l, "answering placement candidate feat-x ") {
				n++
			}
		}
		return n
	}
	time.Sleep(time.Second)
	settled := answering()
	if settled == 0 {
		t.Fatal("feat-x was never answered")
	}
	hkey, _ := delivery.HaveKey(itNode, "web")
	for _, detail := range []string{"its tool web_health: no link to the bus is open", "its tool web_health: no answer within 10s"} {
		h := b.have("web")
		h.Health, h.Detail = delivery.HealthFailed, detail
		raw, err := delivery.EncodeHave(h)
		if err != nil {
			t.Fatal(err)
		}
		if _, err := b.haves.Put(hkey, raw); err != nil {
			t.Fatal(err)
		}
		if _, err := resolved.Put(itNode+".web.greeting", []byte(`{"value":"salut","source":"module"}`)); err != nil {
			t.Fatal(err)
		}
	}
	time.Sleep(1500 * time.Millisecond)
	if got := answering(); got != settled {
		t.Errorf("feat-x answered %d more time(s) at the same revision, with nothing it is composed from changed", got-settled)
	}
	same, _ := delivery.DecodePlacement(place)
	craw, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Proposal: "feat-x", Placement: &same})
	if err != nil {
		t.Fatal(err)
	}
	ckey, _ := delivery.PlacementCandidateKey("feat-x", itNode)
	if _, err := candidates.Put(ckey, craw); err != nil {
		t.Fatal(err)
	}
	within(t, 10*time.Second, "the candidate's next revision was not answered", func() bool { return answering() > settled })
	time.Sleep(1500 * time.Millisecond)
	if got := answering(); got != settled+1 {
		t.Errorf("the candidate's next revision was answered %d time(s), want 1", got-settled)
	}
}

// h8back is A7 of issue 1126: no composition made elsewhere is applied. A candidate deleting the node's placement is
// refused, never answered as a return to the controller's compositions; the placement key deleted leaves the agent on
// its own composer, composing nothing and keeping what runs, said in the journal, the node's have still saying it
// composes its own.
func h8back(t *testing.T, b *itBed, placement, resolved, candidates nats.KeyValue, hkey string) {
	t.Helper()
	node := func() delivery.HaveNode {
		e, err := b.haves.Get(hkey)
		if err != nil {
			return delivery.HaveNode{}
		}
		n, _ := delivery.DecodeHaveNode(e.Value())
		return n
	}
	if n := node(); n.Composes != delivery.ComposesOwn {
		t.Errorf("a node composing its own says it composes %q", n.Composes)
	}
	craw, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Proposal: "revert-x"})
	if err != nil {
		t.Fatal(err)
	}
	ckey, _ := delivery.PlacementCandidateKey("revert-x", itNode)
	if _, err := candidates.Put(ckey, craw); err != nil {
		t.Fatal(err)
	}
	within(t, 10*time.Second, "the node did not refuse a candidate taking its placement away", func() bool {
		a := node().Candidates["revert-x"]
		return a.Proposal == "revert-x" && a.Returns == "" && a.Verdict == delivery.CandidateRefused &&
			strings.Contains(a.Refused, "composes nothing")
	})
	ran := b.have("web").Composed
	pkey, _ := delivery.PlacementNodeKey(itNode)
	// A pass in flight when the key goes (issue 1126, 2026-10-28): a config value changed starts one, and the key goes at once.
	if _, err := resolved.Put(itNode+".web.greeting", []byte(`{"value":"salut","source":"module"}`)); err != nil {
		t.Fatal(err)
	}
	if err := placement.Delete(pkey); err != nil {
		t.Fatal(err)
	}
	within(t, 15*time.Second, "the node did not say it composes nothing once its placement went", func() bool {
		b.mu.Lock()
		said := strings.Join(b.said, "\n")
		b.mu.Unlock()
		return strings.Contains(said, "key in the seat's placement state is gone: it composes nothing of its own")
	})
	time.Sleep(3 * time.Second) // what a pass in flight would mark lands within it
	b.mu.Lock()
	said := strings.Join(b.said, "\n")
	b.mu.Unlock()
	if n := node(); n.Composes != delivery.ComposesOwn || strings.Contains(said, "applying the controller's compositions") {
		t.Errorf("without its placement the node turned to the controller's compositions: %q", n.Composes)
	}
	if h := b.have("web"); h.Composed != ran || h.Phase == delivery.PhaseFailed || b.file("web") != "greeting=salut\n" {
		t.Errorf("without its placement what runs moved: %+v, %q", h, b.file("web"))
	}
}

// h8state is what h8 set up that the rest of H8 writes to.
type h8state struct {
	placement, resolved, candidates nats.KeyValue
	// packages is the engine's state packages (ADR 0565 §6), as the grants part creates it from the engine's manifest.
	packages nats.KeyValue
	hkey     string
	place    []byte
	fp       string
}

// h8 sets H8's bus up and starts the agent, then places the node. With checkBuild, the placed module's build key is
// first one that does not decode, which the node must read through its session (and say), then a done one.
func h8(t *testing.T, b *itBed, checkBuild bool) (func(), h8state) {
	t.Helper()
	kvOf := func(bucket string) nats.KeyValue {
		kv, err := b.js.CreateKeyValue(&nats.KeyValueConfig{Bucket: bucket, History: 5})
		if err != nil {
			t.Fatal(err)
		}
		return kv
	}
	placement, modules, manifests := kvOf(delivery.PlacementBucket), kvOf(delivery.ModulesBucket), kvOf(delivery.ManifestsBucket)
	resolved := kvOf(selfcompose.ResolvedBucket)
	builds := kvOf(delivery.BuilderBuildsBucket)
	candidates := kvOf(delivery.PlacementCandidatesBucket)
	packages := kvOf(link.PackagesBucket)
	path := filepath.Join(b.dir, "web.conf")
	man := `{"module":"web","version":"1","config":{"greeting":{"type":"string","description":"what the file says","default":"hello"}},` +
		`"resources":[{"id":"conf","type":"file","path":"` + path + `","mode":"0644","content":"greeting=${config:greeting}\n"},` +
		`{"id":"tools","type":"package","package":"web-tools"}],` +
		`"build":{"artifacts":[{"name":"files","kind":"archive","from":"files"}]}}`
	digest := delivery.ManifestDigest([]byte(man))
	mraw, _ := delivery.EncodeManifest(delivery.Manifest{Module: "web", JSON: man})
	mkey, _ := delivery.ManifestsKey(digest)
	fp := fpIt("a")
	modv, _ := delivery.EncodeModule(delivery.Module{FP2: fp, Manifest: digest, Repository: "novox/web", Commit: strings.Repeat("1", 40)})
	place, _ := delivery.EncodePlacement(delivery.Placement{Node: itNode, Modules: []delivery.PlacedModule{{Module: "web"}}})
	rec, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: itNode, Capabilities: []string{"service-manager", "package-manager"}}})
	// it1 in one key per module and node (research 072 row 37), it2 in the form before: a node reads both.
	pkey, _ := delivery.PlacementNodeKey(itNode)
	places, _ := delivery.PlacementValues(delivery.Placement{Node: itNode, Modules: []delivery.PlacedModule{{Module: "web"}}})
	place = places[pkey]
	hkey, _ := delivery.HaveNodeKey(itNode)
	// A provider on it2 serving from its own config value: a mail domain, which it1's composition settles.
	mailer := `{"module":"mailer","version":"1","config":{"domain":{"type":"string","description":"the mail domain","required":true}},` +
		`"provides":[{"name":"smtp","scope":"mesh"}],"serves":{"smtp":{"port":587,"domain":"${config:domain}"}}}`
	mdigest := delivery.ManifestDigest([]byte(mailer))
	mlraw, _ := delivery.EncodeManifest(delivery.Manifest{Module: "mailer", JSON: mailer})
	mlkey, _ := delivery.ManifestsKey(mdigest)
	mlmod, _ := delivery.EncodeModule(delivery.Module{FP2: fpIt("b"), Manifest: mdigest, Repository: "novox/mailer", Commit: strings.Repeat("2", 40)})
	place2, _ := delivery.EncodePlacement(delivery.Placement{Node: "it2", Modules: []delivery.PlacedModule{{Module: "mailer"}}})
	rec2, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: "it2", Capabilities: []string{"service-manager"}}})
	pkey2, _ := delivery.LegacyPlacementKey("it2")
	hkey2, _ := delivery.HaveNodeKey("it2")
	buildKey, _ := delivery.BuilderBuildKey("web", fp)
	done, _ := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildDone, Manifest: digest,
		Artifacts: []delivery.BuiltArtifact{{Name: "files", Kind: "archive", Reference: "artifact-store://web/files/blobs/" + fpDigest("f"), Digest: fpDigest("f")}}})
	firstBuild := done
	if checkBuild {
		firstBuild = []byte(`{"v":1,"state":"no such state"}`)
	}
	for _, w := range []struct {
		kv    nats.KeyValue
		key   string
		value []byte
	}{{manifests, mkey, mraw}, {modules, "web", modv}, {b.haves, hkey, rec},
		{resolved, itNode + ".web.greeting", []byte(`{"value":"bonjour","source":"module"}`)},
		{manifests, mlkey, mlraw}, {modules, "mailer", mlmod}, {b.haves, hkey2, rec2}, {placement, pkey2, place2},
		{resolved, "it2.mailer.domain", []byte(`{"value":"example.org","source":"module"}`)}, {builds, buildKey, firstBuild}} {
		if _, err := w.kv.Put(w.key, w.value); err != nil {
			t.Fatal(err)
		}
	}
	want, _ := delivery.EncodeWant(delivery.Want{FP: fp, Self: true})
	wkey, _ := delivery.WantKey(itNode, "web")
	if _, err := b.wants.Put(wkey, want); err != nil {
		t.Fatal(err)
	}

	stop := b.start()
	// Not placed yet: the agent applies the controller's compositions, of which there are none. The catalogue's seed then
	// places the node, and the agent begins again composing its own (ADR 0549 §4), with no restart of the engine.
	time.Sleep(300 * time.Millisecond)
	if b.file("web") != "" {
		t.Fatal("a node the catalogue does not place composed its own")
	}
	if _, err := placement.Put(itNode+".web", places[itNode+".web"]); err != nil {
		t.Fatal(err)
	}
	if _, err := placement.Put(pkey, place); err != nil {
		t.Fatal(err)
	}
	if checkBuild {
		// The build of the placed module is read through the node's session: one that does not decode is said, and nothing
		// is applied; a done one, then a config event, and the node composes.
		within(t, 5*time.Second, "the node did not read web's build key through its session", func() bool {
			b.mu.Lock()
			defer b.mu.Unlock()
			return strings.Contains(strings.Join(b.said, "\n"), "the builder's "+buildKey)
		})
		if b.file("web") != "" {
			t.Fatal("composed with a build key that does not decode")
		}
		if _, err := builds.Put(buildKey, done); err != nil {
			t.Fatal(err)
		}
		if _, err := resolved.Put(itNode+".web.greeting", []byte(`{"value":"bonjour","source":"module"}`)); err != nil {
			t.Fatal(err)
		}
	}
	return stop, h8state{placement: placement, resolved: resolved, candidates: candidates, packages: packages, hkey: hkey, place: place, fp: fp}
}

// h8rest is H8 once the node composes its own: the have, a config revision, and the candidate answered "same".
func h8rest(t *testing.T, b *itBed, resolved, candidates nats.KeyValue, hkey string, place []byte, fp string) {
	t.Helper()
	within(t, 2*time.Second, "the have does not name the build", func() bool {
		h := b.have("web")
		return h.FP == fp && h.Composed != "" && h.Phase == delivery.PhaseRunning
	})
	// ADR 0593 §6: the node publishes what it composed in its composition state, under its own grant: web at the build
	// its want names, with the digest its have names (the plan left out of it), and the node's own.
	first := b.have("web").Composed
	webKey, _ := delivery.ComposedKey(itNode, "web", fp)
	within(t, 5*time.Second, "the node did not publish web's composition in its composition state", func() bool {
		e, err := b.composed.Get(webKey)
		if err != nil {
			return false
		}
		c, err := delivery.DecodeComposed(e.Value())
		if err != nil {
			return false
		}
		d, err := delivery.ComposedDigestOf(c)
		return err == nil && d == first
	})
	if _, err := resolved.Put(itNode+".web.greeting", []byte(`{"value":"salut","source":"module"}`)); err != nil {
		t.Fatal(err)
	}
	within(t, 3*time.Second, "a changed config value was not composed and applied again", func() bool { return b.file("web") == "greeting=salut\n" })
	within(t, 2*time.Second, "the have does not name the revision", func() bool {
		h := b.have("web")
		return h.FP == fp && h.Composed != first && h.Phase == delivery.PhaseRunning
	})

	// The node says its facts (its record, as the test wrote it), then a candidate equal to its placement is written: the
	// node answers "same" at the candidate's head and revision in its own key, and applies nothing.
	b.session.SayFacts(context.Background(), link.Report{Profile: map[string]any{"capabilities": []any{
		map[string]any{"name": "service-manager", "present": true}}}})
	second := b.have("web").Composed
	same, _ := delivery.DecodePlacement(place)
	craw, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Proposal: "feat-x", Placement: &same})
	if err != nil {
		t.Fatal(err)
	}
	ckey, _ := delivery.PlacementCandidateKey("feat-x", itNode)
	crev, err := candidates.Put(ckey, craw)
	if err != nil {
		t.Fatal(err)
	}
	var answer delivery.CandidateAnswer
	within(t, 10*time.Second, "the node did not answer its candidate", func() bool {
		e, err := b.haves.Get(hkey)
		if err != nil {
			return false
		}
		n, err := delivery.DecodeHaveNode(e.Value())
		answer = n.Candidates["feat-x"]
		return err == nil && answer.Proposal == "feat-x" && n.Record != nil
	})
	if answer.Verdict != delivery.CandidateSame || answer.Revision != crev || len(answer.Same) != 1 || answer.Same[0] != "web" {
		t.Errorf("the answer to a candidate equal to the placement: %+v", answer)
	}
	if h := b.have("web"); h.Composed != second || b.file("web") != "greeting=salut\n" {
		t.Errorf("a candidate moved what runs: %+v, %q", h, b.file("web"))
	}
	h8candidate(t, b, fp, second)
}

// h8candidate is ADR 0593 §7: a check's candidate is composed on the node through its seat's verb compose-candidate,
// under its own grant: the indexer's candidate changing web's manifest is answered composed, web changed at the
// candidate's build, within the answer bound; nothing is applied and the composition state does not move.
func h8candidate(t *testing.T, b *itBed, fp, running string) {
	t.Helper()
	kvOf := func(bucket string) nats.KeyValue {
		kv, err := b.js.CreateKeyValue(&nats.KeyValueConfig{Bucket: bucket, History: 5})
		if err != nil {
			t.Fatal(err)
		}
		return kv
	}
	cands, _ := kvOf(delivery.CandidatesBucket), kvOf(delivery.VerdictsBucket)
	manifests, err := b.js.KeyValue(delivery.ManifestsBucket)
	if err != nil {
		t.Fatal(err)
	}
	path := filepath.Join(b.dir, "web.conf")
	man := `{"module":"web","version":"2","config":{"greeting":{"type":"string","description":"what the file says","default":"hello"}},` +
		`"resources":[{"id":"conf","type":"file","path":"` + path + `","mode":"0640","content":"greeting=${config:greeting}\n"},` +
		`{"id":"tools","type":"package","package":"web-tools"}]}`
	digest := delivery.ManifestDigest([]byte(man))
	mraw, _ := delivery.EncodeManifest(delivery.Manifest{Module: "web", JSON: man})
	mkey, _ := delivery.ManifestsKey(digest)
	if _, err := manifests.Put(mkey, mraw); err != nil {
		t.Fatal(err)
	}
	head := delivery.Head{Repository: "novox/web", Number: 3, Head: strings.Repeat("3", 40)}
	heads, _ := delivery.HeadsHash([]delivery.Head{head})
	pulls := []delivery.CandidatePull{{Head: head, Modules: []string{"web"},
		Fps: map[string]delivery.CandidateModule{"web": {FP2: fpIt("c"), Manifest: digest}}}}
	craw, err := delivery.EncodeCandidate(delivery.Candidate{Group: "novox/web#3", HeadsHash: heads, Pulls: pulls,
		FPSet: delivery.FingerprintSet(pulls)})
	if err != nil {
		t.Fatal(err)
	}
	key, _ := delivery.CompositionKey("novox/web#3", heads)
	if _, err := cands.Put(key, craw); err != nil {
		t.Fatal(err)
	}
	before, _ := b.composed.Status()
	got, err := composeCandidateFrom(itNode, &shadowNow)(key, false)
	if err != nil {
		t.Fatal(err)
	}
	raw, _ := got.(json.RawMessage)
	a, err := delivery.DecodeComposeCandidateAnswer(raw)
	if err != nil {
		t.Fatalf("the answer %s: %v", raw, err)
	}
	if a.Verdict != delivery.ComposeCandidateComposed || len(a.Modules) != 1 || a.Modules[0].Module != "web" ||
		a.Modules[0].FP != fpIt("c") || !a.Modules[0].Changed || a.Took > delivery.ComposeAnswerBound.Milliseconds() {
		t.Errorf("the node's answer to a candidate changing web: %+v", a)
	}
	after, _ := b.composed.Status()
	if h := b.have("web"); h.Composed != running || b.file("web") != "greeting=salut\n" || before == nil || after == nil ||
		after.Values() != before.Values() {
		t.Errorf("composing a candidate moved what runs or the composition state: %+v, %q", h, b.file("web"))
	}
	if _, err := b.composed.Get(mustComposedKey(t, "web", fpIt("c"))); err == nil {
		t.Error("the candidate's composition was written into the composition state")
	}
}

func mustComposedKey(t *testing.T, module, fp string) string {
	t.Helper()
	key, err := delivery.ComposedKey(itNode, module, fp)
	if err != nil {
		t.Fatal(err)
	}
	return key
}