mesh-host/internal/selfcompose/selfcompose.go

// Package selfcompose is the node agent's own composer (novox/hq ADR 0547 §2, issue 1126): it composes this node's
// modules and its own composition with mesh-sdk's composer, from what is on the bus, and hands the compositions to the
// agent as the agent once read them from the controller's bucket `composed-<node>`.
//
// **It watches, it never polls** (ADR 0547 §2): its inputs are watches of the node's wants, the indexer's placement and
// modules, every node's own have key (where a node's record rides, ADR 0555 §3) and mesh-config's resolved values of this
// node. A change of any of them marks the node to compose again; one loop drains the mark, so a burst of changes composes
// the node once (composing one node takes 0.8 to 4.4 s, measured 2026-10-26). It hands the agent only the compositions that
// differ from the ones it handed before, and takes back one no want names any more. A composition that cannot be made is
// said, once per reason, and the ones handed before stand: a node keeps running what it ran rather than nothing.
package selfcompose

import (
	"context"
	"crypto/ed25519"
	"crypto/sha256"
	"encoding/base64"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"slices"
	"sort"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"git.novox.be/novox/mesh-sdk/go/catalogue"
	"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"
	"git.novox.be/novox/mesh-sdk/go/overlay"

	"github.com/nats-io/nats.go"

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

// ResolvedBucket is mesh-config's state of resolved values, keyed `<node>.<module>.<key>` (mesh-catalog modules/mesh-config
// resolve.go ResolvedKey); a node reads its own slice alone (ADR 0547 §2).
const ResolvedBucket = "mesh-config_resolved"

// Env is what the composer needs of the node.
type Env struct {
	Node string
	// Watch follows one bucket's keys matching pattern, as the agent's watches do.
	Watch func(ctx context.Context, bucket, pattern string) (<-chan agent.Update, error)
	// Get reads one key's latest value; nil and no error when there is none.
	Get  func(ctx context.Context, bucket, key string) ([]byte, error)
	Mesh frombus.Mesh
	Say  func(string)
	// Own asks the vault's `own` in this node's engine's name, with the arguments given: its modules' own secrets and the
	// pair credentials asked, sealed to its key (ADR 0550 §5). Nil asks nothing, and a module with either does not compose.
	Own func(ctx context.Context, args map[string]any) (json.RawMessage, error)
	// Hear subscribes a subject: here the vault's word that a secret changed (VaultSecretChanged), on which the node
	// composes again, so a rotated secret or credential reaches it at once (ADR 0547 §2). Nil hears nothing.
	Hear func(subject string) (<-chan []byte, func(), error)
	// Refused is told why the node cannot compose its modules (why, the node refused whole), or, of a node that composes,
	// each module it holds and why (held: refused alone, the rest composed and applied, frombus.Compose), and both empty
	// once it composes every module again, so each module's have says it (agent.NotComposedEach): a refusal is loud where
	// haves are read, not one journal line, and one module's never marks the others (issue 1126). Nil tells nobody.
	Refused func(why string, held map[string]string)
	// Running is what the node runs now (agent.Everything), and Unseal opens a value sealed to this machine: a secret
	// the vault seals again for each ask is kept as the box the node already runs at the same slot (the same field of
	// the same resource of the same module, delivery.KeepSeals) when that box opens to the same value, so neither the
	// switch to composing its own nor a restart changes a composition (ADR 0547 §3); a box held at another slot is
	// never taken, though it opens to the same value (issue 1126, C1: a provider's grant and its consumer's secret).
	// Either nil keeps the vault's box.
	Running func() []delivery.Composed
	Unseal  func(sealed string) ([]byte, error)
	// Held is the sealed values a module's have says it runs, kept by the agent on the bus: what runs after a restart,
	// before the agent applied anything, and after the switch once the kept declaration is old. By slot
	// (delivery.Have.SealedAt); legacy is the list a have written before slots holds (delivery.Have.Sealed), reused
	// only within its own module.
	// An error is a have the bus did not answer for, never one absent: the pass is then tried again (BusFailed).
	Held func(module string) (slots delivery.Seals, legacy []string, err error)
	// Fetch reads an archive at a URL in the artifact store: a moved module's templates archive, whose texts its built
	// manifest carries (issue 1126). Read once per digest. Nil reads none, and a moved module does not compose.
	Fetch func(ctx context.Context, url string) ([]byte, error)
	// VaultAuthority is the vault's authority as this node read it from the vault's `signers` (ADR 0550 §2), composed
	// beside the controller's as the mesh's authority (frombus.SignersReader); "" or nil composes the controller's alone.
	VaultAuthority func() string
	// Certify asks the vault's `sign` for this node's own certificate, keeps it and writes the node's record again (ADR
	// 0550 §2): asked only when a module placed here asks for a certificate and the record holds none from an authority
	// the node trusts. Nil asks nothing, and such a module stands in, as before.
	Certify func(ctx context.Context) error
}

// VaultSecretChanged is the vault's event that a secret changed: a name and its generation, never a value.
const VaultSecretChanged = "mesh.mod.mesh-vault.event.secret-changed"

// Composer composes the node and is the agent's watcher of its compositions (agent.Env.Composed).
type Composer struct {
	env Env

	mu         sync.Mutex
	wants      map[string]delivery.Want // by module, the node's own under agent.NodeModule
	placements map[string]delivery.Placement
	records    map[string]delivery.NodeRecord
	// inOwn is, by module, whether this node's have of it names no fingerprint of the new form (mesh-wireguard, a module
	// at a build of the old form): the node runs it inside its own composition, and composes it there again (issue 1126).
	inOwn     map[string]bool
	modules   map[string]delivery.Module
	values    map[string]map[string]any // by module, this node's
	manifests map[string][]byte         // by digest, read once
	archives  map[string][]byte         // templates archives by digest, read once
	handed    map[string]string         // composed key to the digest handed
	// sealedOwn is each own secret's sealed value as first handed, by module/name and generation: a value sealed again
	// is other bytes, so the one first had is kept while the generation stands, and the composition stays the same.
	sealedOwn map[string]ownSealed
	// certified is the serving key a certificate was last asked for successfully, so it is asked once per key.
	certified string
	said      string
	refusal   string
	// held is what the composer's last pass held: each module refused alone, and the node's own ("node") when it could
	// compose none (heldNow).
	held []string
	// doing is what the composer waits on now (waiting), for a running answer to say (issue 1126, T2); its own lock, so
	// it is read while c.mu is held.
	doingMu sync.Mutex
	doing   string
	caught  bool
	rev     uint64
	// missed counts the reads the bus did not answer (BusFailed), for a pass or an answer to tell whether it met one:
	// a read that failed on the bus is tried again after a bound, never kept as an answer (issue 1126, 2026-10-28).
	missed atomic.Uint64
	// owed is set when a pass refused for a bus that did not answer: the composer composes again after a bound.
	owed atomic.Bool
}

// note counts err when it is the bus not answering, and returns it.
func (c *Composer) note(err error) error {
	if BusFailed(err) {
		c.missed.Add(1)
	}
	return err
}

// get is env.Get, its failures noted.
func (c *Composer) get(ctx context.Context, bucket, key string) ([]byte, error) {
	raw, err := c.env.Get(ctx, bucket, key)
	return raw, c.note(err)
}

// New is a composer for one node.
func New(env Env) *Composer {
	if env.Say == nil {
		env.Say = func(string) {}
	}
	return &Composer{env: env, wants: map[string]delivery.Want{}, placements: map[string]delivery.Placement{},
		records: map[string]delivery.NodeRecord{}, inOwn: map[string]bool{}, modules: map[string]delivery.Module{}, values: map[string]map[string]any{},
		manifests: map[string][]byte{}, archives: map[string][]byte{}, handed: map[string]string{}, sealedOwn: map[string]ownSealed{}}
}

// ownSealed is one own secret's sealed value at a generation and the key it is sealed to.
type ownSealed struct {
	generation int
	key        string
	sealed     string
}

// input is one watched bucket and how its updates are taken.
type input struct {
	bucket, pattern string
	take            func(c *Composer, u agent.Update)
	// changed, when set, takes the update in take's place and says whether it changed anything composed.
	changed func(c *Composer, u agent.Update) bool
}

func (c *Composer) inputs() []input {
	node := c.env.Node
	return []input{
		{delivery.AssignerWantsBucket, "want." + node + ".>", (*Composer).takeWant, nil},
		{delivery.PlacementBucket, ">", (*Composer).takePlacement, nil},
		// Every node's record, and this node's own haves (what it runs inside its own composition): two watches, each
		// within the node's grant (buscompose: readGrants of the have bucket at `*.node`, NodeKeysGrants at `<node>.>`).
		// A watch of every key is refused by the bus (issue 1126, 2026-10-27 09:21: the fold's `>` stopped every agent).
		{delivery.HaveBucket, "*." + delivery.HaveNodeWord, nil, (*Composer).takeRecordChanged},
		{delivery.HaveBucket, node + ".>", nil, (*Composer).takeModuleHave},
		{delivery.ModulesBucket, ">", (*Composer).takeModule, nil},
		{ResolvedBucket, node + ".>", (*Composer).takeValue, nil},
	}
}

// Watch starts the composer and answers the compositions it makes: each as a key of `composed-<node>` would be, then a
// Caught update once every input was read and the node composed once, then every change. It ends when ctx ends or an
// input's watch ends.
func (c *Composer) Watch(ctx context.Context, _ string) (<-chan agent.Update, error) {
	ins := c.inputs()
	type tagged struct {
		in int
		u  agent.Update
		ok bool
	}
	merged := make(chan tagged, 64)
	for i, in := range ins {
		ch, err := c.env.Watch(ctx, in.bucket, in.pattern)
		if err != nil {
			return nil, fmt.Errorf("the node composes from %s, which cannot be watched: %w", in.bucket, err)
		}
		go func(i int, ch <-chan agent.Update) {
			for u := range ch {
				select {
				case merged <- tagged{in: i, u: u, ok: true}:
				case <-ctx.Done():
					return
				}
			}
			select {
			case merged <- tagged{in: i}:
			case <-ctx.Done():
			}
		}(i, ch)
	}
	if c.env.Hear != nil {
		heard, stop, err := c.env.Hear(VaultSecretChanged)
		if err != nil {
			return nil, fmt.Errorf("the node composes again when a secret changes, and cannot hear the vault say so: %w", err)
		}
		go func() {
			defer stop()
			for {
				select {
				case <-ctx.Done():
					return
				case _, ok := <-heard:
					if !ok {
						return
					}
					select {
					case merged <- tagged{in: -1, ok: true}:
					case <-ctx.Done():
						return
					}
				}
			}
		}()
	}
	out := make(chan agent.Update, 64)
	go func() {
		defer close(out)
		caught := make([]bool, len(ins))
		ready := false
		dirty := false
		// A pass refused for a bus that did not answer is made again after a bound (laterAgain), not kept until another
		// input moves (issue 1126, 2026-10-28).
		var again <-chan time.Time
		tries := 0
		for {
			// Drain whatever came, then compose once for all of it.
			select {
			case <-ctx.Done():
				return
			case <-again:
				again = nil
				dirty = true
			case t := <-merged:
				if !t.ok {
					c.env.Say(fmt.Sprintf("the watch of %s ended; the node composes no more until its agent starts again", ins[t.in].bucket))
					return
				}
				switch {
				case t.in < 0:
					dirty = true // a secret changed: its new generation is asked of the vault at the compose
				case t.u.Caught:
					caught[t.in] = true
				case ins[t.in].changed != nil:
					dirty = ins[t.in].changed(c, t.u) || dirty
				default:
					ins[t.in].take(c, t.u)
					dirty = true
				}
				if len(merged) > 0 {
					continue
				}
			}
			if !ready {
				ready = all(caught)
				if !ready {
					continue
				}
				dirty = true
			}
			if !dirty {
				continue
			}
			dirty = false
			for _, u := range c.compose(ctx) {
				select {
				case out <- u:
				case <-ctx.Done():
					return
				}
			}
			if c.owed.Load() {
				wait := LaterWait(tries)
				tries++
				c.env.Say(fmt.Sprintf("the node composes again in %s: a read on the bus did not answer", wait))
				again = time.After(wait)
			} else {
				tries, again = 0, nil
			}
			if !c.caughtSent() {
				select {
				case out <- agent.Update{Caught: true}:
				case <-ctx.Done():
					return
				}
			}
		}
	}()
	return out, nil
}

func all(b []bool) bool {
	for _, v := range b {
		if !v {
			return false
		}
	}
	return true
}

// caughtSent says the Caught update went out, marking it sent the first time it is asked.
func (c *Composer) caughtSent() bool {
	c.mu.Lock()
	defer c.mu.Unlock()
	sent := c.caught
	c.caught = true
	return sent
}

func (c *Composer) takeWant(u agent.Update) {
	c.mu.Lock()
	defer c.mu.Unlock()
	if _, err := delivery.ParseWantNodeKey(u.Key); err == nil {
		if u.Deleted {
			delete(c.wants, agent.NodeModule)
		} else {
			c.wants[agent.NodeModule] = delivery.Want{}
		}
		return
	}
	node, module, err := delivery.ParseWantKey(u.Key)
	if err != nil || node != c.env.Node {
		return
	}
	w, err := delivery.DecodeWant(u.Value)
	if u.Deleted || err != nil || w.Removed || w.FP == "" {
		delete(c.wants, module)
		return
	}
	c.wants[module] = w
}

func (c *Composer) takePlacement(u agent.Update) {
	node, err := delivery.ParsePlacementKey(u.Key)
	if err != nil {
		return
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	p, err := delivery.DecodePlacement(u.Value)
	if u.Deleted || err != nil || p.Node != node {
		delete(c.placements, node)
		return
	}
	c.placements[node] = p
}

func (c *Composer) takeRecord(u agent.Update) {
	node, _, isNode, err := delivery.ParseHaveKey(u.Key)
	if err != nil || !isNode {
		return
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	h, err := delivery.DecodeHaveNode(u.Value)
	if u.Deleted || err != nil || h.Record == nil {
		delete(c.records, node)
		return
	}
	c.records[node] = *h.Record
}

// takeRecordChanged takes a node's record, which is always composed again.
func (c *Composer) takeRecordChanged(u agent.Update) bool {
	c.takeRecord(u)
	return true
}

// takeModuleHave takes one of this node's modules' haves, and says whether it changed which it runs inside its own
// composition: one whose have names no fingerprint of the new form, is not removed, and runs something (nats on novox: a build-...
// id; mesh-wireguard: none). Read from the haves, which outlast a restart, never from what the agent has applied
// this run: a fresh agent composes before it applies anything (review of the fold, round 1).
func (c *Composer) takeModuleHave(u agent.Update) bool {
	node, module, isNode, err := delivery.ParseHaveKey(u.Key)
	if err != nil || isNode || node != c.env.Node {
		return false
	}
	in := false
	if !u.Deleted {
		if h, err := delivery.DecodeHave(u.Value); err == nil {
			// A have that names nothing it runs, failed (the node could not compose a module it never ran: NotComposed), is
			// no module run inside the node's own composition: taken as one, it was composed from its manifest past its
			// build (issue 1126, review round 1 of the builds).
			runsNothing := h.Phase == delivery.PhaseFailed && h.FP == "" && h.Composed == ""
			in = h.Phase != delivery.PhaseRemoved && !runsNothing && !strings.HasPrefix(h.FP, delivery.SourcePrefix2)
		}
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.inOwn[module] == in {
		return false
	}
	if in {
		c.inOwn[module] = true
	} else {
		delete(c.inOwn, module)
	}
	return true
}

// inOwnOf is, of the modules p places, those the node runs inside its own composition (c.inOwn), sorted; c.mu held.
// Machines is, from the placements this composer holds, each node by its address on the private network, and the hub's
// node: what names the machine a failing network part points at (delivery.NetworkPart.Targets, issue 774). Empty
// before any placement is read.
func (c *Composer) Machines() (byAddress map[string]string, hub string) {
	c.mu.Lock()
	defer c.mu.Unlock()
	byAddress = map[string]string{}
	for node, p := range c.placements {
		if a, _, _ := strings.Cut(p.Decided.Address, "/"); a != "" {
			byAddress[a] = node
		}
		if p.Decided.Hub {
			hub = node
		}
	}
	return byAddress, hub
}

func (c *Composer) inOwnOf(p delivery.Placement) []string {
	var out []string
	for _, m := range p.Modules {
		if c.inOwn[m.Module] {
			out = append(out, m.Module)
		}
	}
	sort.Strings(out)
	return out
}

func (c *Composer) takeModule(u agent.Update) {
	name, err := delivery.ParseModulesKey(u.Key)
	if err != nil {
		return
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	m, err := delivery.DecodeModule(u.Value)
	if u.Deleted || err != nil {
		delete(c.modules, name)
		return
	}
	c.modules[name] = m
}

func (c *Composer) takeValue(u agent.Update) {
	rest, ok := strings.CutPrefix(u.Key, c.env.Node+".")
	if !ok {
		return
	}
	module, key, ok := strings.Cut(rest, ".")
	if !ok {
		return
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	// Only a value its source lays: a default is the composition's already, as the controller reads them
	// (catalogue.LaidValue, issue 1126, C2: Element's links blanked by a laid default).
	v, laid := catalogue.LaidValue(u.Value)
	if u.Deleted || !laid {
		delete(c.values[module], key)
		return
	}
	if c.values[module] == nil {
		c.values[module] = map[string]any{}
	}
	c.values[module][key] = v
}

// compose composes the node once and answers the updates to hand: each composition whose digest differs from the one
// handed, and a deletion for each handed one no want names any more.
func (c *Composer) compose(ctx context.Context) []agent.Update {
	c.mu.Lock()
	fps := map[string]string{}
	kept := map[string]string{}
	var behind []string // modules wanted at another build than the trunk's: composed with that build's manifest
	for module, w := range c.wants {
		if module == agent.NodeModule {
			continue
		}
		fps[module] = w.FP
		if m, ok := c.modules[module]; ok && m.FP2 != w.FP {
			behind = append(behind, module)
		}
	}
	_, nodeWanted := c.wants[agent.NodeModule]
	// What it runs inside its own composition and no want names at a build of the new form stays there, as the
	// controller's composed writer folds it (issue 1126); compose folds none a want names.
	own, placed := c.placements[c.env.Node]
	inOwn := c.inOwnOf(own)
	r := reader{c: c}
	c.mu.Unlock()
	if !placed {
		// The catalogue no longer places it: the agent returns to the controller's compositions (endWhenUnplaced), and
		// this is no refusal of what it runs (issue 1126, the seed's revert).
		c.sayOnce("this node's key in the seat's placement state is gone: it composes nothing of its own, and keeps running what it runs")
		c.owed.Store(false) // nothing of its own to compose, so no pass is owed and no timer armed again
		return nil
	}
	// Composed from the build its want names (frombus.Scope.KeptAt, issue 1126).
	for _, module := range behind {
		kept[module] = fps[module]
	}

	// What it runs: a folded module of an old-form build is carried in its own composition as it runs (issue 1126).
	var running []delivery.Composed
	if c.env.Running != nil {
		running = c.env.Running()
	}
	// Its own certificate, asked of the vault when a module here asks for one and the record holds none it trusts.
	c.certifyIfWanted(ctx, r, own)
	scope := frombus.Scope{Node: c.env.Node, KeptAt: kept, InOwn: inOwn, Running: running}
	missed := c.missed.Load()
	compositions, _, held, err := frombus.Compose(ctx, r, c.env.Mesh, scope, fps)
	err = c.missedIn(missed, err)
	// A bus that does not answer for a moment (a reload of its accounts pauses it a few seconds) is no refusal: the pass is
	// tried again after a growing wait before anything is marked (issue 1126).
	for try := 0; err != nil && Transient(err) && ctx.Err() == nil && try < len(composeAgain); try++ {
		c.env.Say(fmt.Sprintf("the node's composition met a bus that did not answer, and is tried again in %s: %v", composeAgain[try], err))
		select {
		case <-ctx.Done():
		case <-time.After(composeAgain[try]):
		}
		if ctx.Err() == nil {
			missed = c.missed.Load()
			compositions, _, held, err = frombus.Compose(ctx, r, c.env.Mesh, scope, fps)
			err = c.missedIn(missed, err)
		}
	}
	if ctx.Err() != nil {
		// The run ended (the agent begins again, on its own compositions or the controller's): what failed with it is
		// no refusal, and nothing is marked (issue 1126: a pass in flight at the switch marked the next agent's modules).
		return nil
	}
	if err != nil {
		// Said before the agent marks the modules: the marks read each have first, which a paused bus delays.
		c.sayOnce(fmt.Sprintf("the node cannot compose its modules, and keeps running what it runs: %v", err))
		c.refused(fmt.Sprintf("%v", err), nil)
		// A bus that did not answer past every try is no answer: composed again after a bound (issue 1126, 2026-10-28).
		c.owed.Store(BusFailed(err))
		return nil
	}
	// The boxes it runs, slot by slot: read before c.mu is taken, since the agent's lock is taken to read what runs.
	missed = c.missed.Load()
	for key, comp := range compositions {
		compositions[key] = c.keepRunning(comp, running)
	}
	if c.missed.Load() > missed {
		// A have of what runs the bus did not answer for: its boxes would be the vault's, sealed again, and the
		// composition another; nothing is handed, and the pass is made again after a bound (issue 1126, 2026-10-28).
		c.sayOnce("the node composes nothing new for now, and keeps running what it runs: the have of what it runs did not answer on the bus")
		c.owed.Store(true)
		return nil
	}
	c.owed.Store(false)
	// A module refused alone is held: it keeps what it runs, its have says why, and every other module is applied (issue
	// 1126: one module's refusal froze the whole node).
	c.refused("", held)
	if len(held) == 0 {
		c.sayOnce("")
	} else {
		var lines []string
		for _, m := range heldModules(held) {
			lines = append(lines, m+": "+held[m])
		}
		c.sayOnce("the node composes its modules but holds " + strings.Join(heldModules(held), ", ") +
			", which keep running what they run: " + strings.Join(lines, "; "))
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	var out []agent.Update
	now := map[string]bool{}
	keys := make([]string, 0, len(compositions))
	for key := range compositions {
		keys = append(keys, key)
	}
	sort.Strings(keys)
	for _, key := range keys {
		comp := compositions[key]
		if comp.Module == "" && !nodeWanted {
			continue
		}
		now[key] = true
		dig, err := delivery.ComposedDigestOf(comp)
		if err != nil {
			c.env.Say(fmt.Sprintf("the composition %s has no digest, and is not handed: %v", key, err))
			continue
		}
		if c.handed[key] == dig {
			continue
		}
		raw, err := delivery.EncodeComposed(comp)
		if err != nil {
			c.env.Say(fmt.Sprintf("the composition %s cannot be written, and is not handed: %v", key, err))
			continue
		}
		c.rev++
		c.handed[key] = dig
		out = append(out, agent.Update{Key: key, Value: raw, Revision: c.rev})
	}
	for key := range c.handed {
		if _, module, _, _, err := delivery.ParseComposedKey(key); err == nil && module != "" {
			if _, isHeld := held[module]; isHeld {
				continue // held: what was handed of it stays, so the node keeps running it
			}
		}
		if !now[key] {
			c.rev++
			delete(c.handed, key)
			out = append(out, agent.Update{Key: key, Deleted: true, Revision: c.rev})
		}
	}
	return out
}

// heldModules is the modules of held, sorted.
func heldModules(held map[string]string) []string {
	out := make([]string, 0, len(held))
	for m := range held {
		out = append(out, m)
	}
	sort.Strings(out)
	return out
}

// refused tells the agent why the node cannot compose (why), or which modules it holds (held), or both empty that it
// composes every module again, once per change.
func (c *Composer) refused(why string, held map[string]string) {
	said := why
	for _, m := range heldModules(held) {
		said += "\x00" + m + "\x00" + held[m]
	}
	c.mu.Lock()
	same := c.refusal == said
	c.refusal = said
	c.held = heldModules(held)
	if why != "" {
		c.held = append(c.held, delivery.HaveNodeWord)
	}
	c.mu.Unlock()
	if !same && c.env.Refused != nil {
		c.env.Refused(why, held)
	}
}

// heldNow is what the composer's last pass held, sorted.
func (c *Composer) heldNow() []string {
	c.mu.Lock()
	defer c.mu.Unlock()
	return slices.Clone(c.held)
}

// keepRunning is comp with each sealed value the node already runs at the same slot kept, where it opens to the same
// value (delivery.KeepSeals, issue 1126, C1): of what runs (running, the agent's), else of what its module's have says
// it runs (Held). A have written before slots names no slot: its boxes are reused within their own module alone, at a
// slot no box is kept for, and only one opening to the slot's value.
func (c *Composer) keepRunning(comp delivery.Composed, running []delivery.Composed) delivery.Composed {
	if c.env.Unseal == nil {
		return comp
	}
	module := comp.Module
	if module == "" {
		module = agent.NodeModule
	}
	kept := delivery.Seals{}
	var legacy []string
	if c.env.Held != nil {
		slots, old, err := c.env.Held(module)
		c.note(err)
		for slot, box := range slots {
			kept[slot] = box
		}
		legacy = old
	}
	for _, r := range running {
		if r.Module == comp.Module {
			for slot, box := range delivery.SealsIn(r) {
				kept[slot] = box
			}
		}
	}
	for slot, box := range delivery.SealsIn(comp) {
		if _, ok := kept[slot]; ok {
			continue
		}
		for _, h := range legacy {
			if same, known := c.sameValue(box, h); known && same {
				kept[slot] = h
				break
			}
		}
	}
	return delivery.KeepSeals(comp, kept, c.sameValue)
}

// waiting notes what the composer waits on until the returned func is called, which restores what it waited on before.
func (c *Composer) waiting(what string) func() {
	c.doingMu.Lock()
	before := c.doing
	c.doing = what
	c.doingMu.Unlock()
	return func() {
		c.doingMu.Lock()
		c.doing = before
		c.doingMu.Unlock()
	}
}

// waitingOn is what the composer waits on now, in words.
func (c *Composer) waitingOn() string {
	c.doingMu.Lock()
	defer c.doingMu.Unlock()
	if c.doing == "" {
		return "nothing it reads (composing)"
	}
	return c.doing
}

func (c *Composer) sayOnce(line string) {
	c.mu.Lock()
	if c.said == line {
		c.mu.Unlock()
		return
	}
	c.said = line
	c.mu.Unlock()
	if line != "" {
		c.env.Say(line)
	}
}

// Policy is a module's upgrade policy as its manifest at the indexer's trunk derives it (catalogue.DerivedUpgrade), for
// the agent's judgement of a move (ADR 0547 §3); empty when the manifest is not known or cannot be read. The manifest is
// read within ctx, the agent's (review of L2, round 1, finding 5), so a stopping agent stops the read.
func (c *Composer) Policy(ctx context.Context, module string) string {
	c.mu.Lock()
	m, ok := c.modules[module]
	c.mu.Unlock()
	if !ok || m.Manifest == "" {
		return ""
	}
	raw, err := reader{c: c}.Manifest(ctx, m.Manifest)
	if err != nil {
		return ""
	}
	parsed, err := catalogue.ParseManifestForInputs(raw)
	if err != nil {
		return ""
	}
	policy, _, _ := catalogue.DerivedUpgrade(parsed)
	return policy
}

// HoldsData says whether a module's manifest declares data (catalogue.Manifest.DataItems), as a composition's Data says
// it, and whether the manifest could be read: the agent's Data for what runs of a module from the kept declaration
// (agent.Env.HoldsData, issue 924).
func (c *Composer) HoldsData(ctx context.Context, module string) (data, known bool) {
	c.mu.Lock()
	m, ok := c.modules[module]
	c.mu.Unlock()
	if !ok || m.Manifest == "" {
		return false, false
	}
	raw, err := reader{c: c}.Manifest(ctx, m.Manifest)
	if err != nil {
		return false, false
	}
	parsed, err := catalogue.ParseManifestForInputs(raw)
	if err != nil {
		return false, false
	}
	return len(parsed.DataItems()) > 0, true
}

// reader is the composer's view of the bus as frombus reads it.
type reader struct{ c *Composer }

// VaultAuthority is the vault's authority the node follows (frombus.SignersReader).
func (r reader) VaultAuthority(context.Context) string {
	if r.c.env.VaultAuthority == nil {
		return ""
	}
	return r.c.env.VaultAuthority()
}

var _ frombus.SignersReader = reader{}

// certificateWanted says whether a module the placement p puts on this node asks for a certificate for the machine's
// internal name (its manifest's `certificate`), read from the indexer's manifests; a manifest that cannot be read is
// taken as asking none, and the composition says what it misses.
func (c *Composer) certificateWanted(ctx context.Context, r reader, p delivery.Placement) bool {
	for _, m := range p.Modules {
		c.mu.Lock()
		digest := c.modules[m.Module].Manifest
		c.mu.Unlock()
		if digest == "" {
			continue
		}
		raw, err := r.Manifest(ctx, digest)
		if err != nil {
			continue
		}
		if parsed, err := catalogue.ParseManifestForInputs(raw); err == nil && parsed.Certificate != nil {
			return true
		}
	}
	return false
}

// certifyIfWanted asks Certify once per serving key when a module placed here asks for a certificate and this node's
// record holds none for its internal name and serving key from an authority it trusts (the one it holds, or the vault's).
func (c *Composer) certifyIfWanted(ctx context.Context, r reader, p delivery.Placement) {
	if c.env.Certify == nil || !c.certificateWanted(ctx, r, p) {
		return
	}
	c.mu.Lock()
	rec, ok := c.records[c.env.Node]
	asked := c.certified
	c.mu.Unlock()
	if !ok || rec.ServingKey == "" || asked == rec.ServingKey {
		return
	}
	key, err := base64.StdEncoding.DecodeString(rec.ServingKey)
	if err != nil || len(key) != ed25519.PublicKeySize {
		return
	}
	trusted := []string{rec.Authority, r.VaultAuthority(ctx)}
	if rec.Certificate != "" && delivery.CheckNodeCertificate(rec.Certificate, trusted, overlay.InternalName(c.env.Node),
		ed25519.PublicKey(key), time.Now()) == nil {
		return
	}
	if err := c.env.Certify(ctx); err != nil {
		c.sayOnce(fmt.Sprintf("a module here asks for a certificate, and the vault's sign gave none: %v", err))
		return
	}
	c.mu.Lock()
	c.certified = rec.ServingKey
	c.mu.Unlock()
}

func (r reader) Placements(context.Context) (map[string]delivery.Placement, error) {
	r.c.mu.Lock()
	defer r.c.mu.Unlock()
	out := make(map[string]delivery.Placement, len(r.c.placements))
	for k, v := range r.c.placements {
		out[k] = v
	}
	return out, nil
}

func (r reader) Records(context.Context) (map[string]delivery.NodeRecord, error) {
	r.c.mu.Lock()
	defer r.c.mu.Unlock()
	out := make(map[string]delivery.NodeRecord, len(r.c.records))
	for k, v := range r.c.records {
		out[k] = v
	}
	return out, nil
}

func (r reader) Modules(context.Context) (map[string]delivery.Module, error) {
	r.c.mu.Lock()
	defer r.c.mu.Unlock()
	out := make(map[string]delivery.Module, len(r.c.modules))
	for k, v := range r.c.modules {
		out[k] = v
	}
	return out, nil
}

func (r reader) Manifest(ctx context.Context, digest string) ([]byte, error) {
	r.c.mu.Lock()
	raw, ok := r.c.manifests[digest]
	r.c.mu.Unlock()
	if ok {
		return raw, nil
	}
	key, err := delivery.ManifestsKey(digest)
	if err != nil {
		return nil, err
	}
	defer r.c.waiting("the manifest " + digest + " (" + delivery.ManifestsBucket + ")")()
	value, err := r.c.get(ctx, delivery.ManifestsBucket, key)
	if err != nil {
		return nil, err
	}
	if value == nil {
		return nil, fmt.Errorf("the indexer holds no manifest %s", digest)
	}
	m, err := delivery.DecodeManifest(value)
	if err != nil {
		return nil, err
	}
	r.c.mu.Lock()
	r.c.manifests[digest] = []byte(m.JSON)
	r.c.mu.Unlock()
	return []byte(m.JSON), nil
}

func (r reader) Build(ctx context.Context, module, fp string) (*delivery.BuilderBuild, error) {
	key, err := delivery.BuilderBuildKey(module, fp)
	if err != nil {
		return nil, err
	}
	defer r.c.waiting("the build of " + module + " at " + fp + " (" + delivery.BuilderBuildsBucket + ")")()
	raw, err := r.c.get(ctx, delivery.BuilderBuildsBucket, key)
	if err != nil || raw == nil {
		return nil, err
	}
	b, err := delivery.DecodeBuilderBuild(raw)
	if err != nil {
		return nil, fmt.Errorf("the builder's %s: %w", key, err)
	}
	return &b, nil
}

// archiveWithin bounds one read of a templates archive in the artifact store.
const archiveWithin = 30 * time.Second

// Archive reads a templates archive once per digest (an archive is the whole of its digest), checked against it.
func (r reader) Archive(ctx context.Context, url, digest string) ([]byte, error) {
	r.c.mu.Lock()
	body, ok := r.c.archives[digest]
	r.c.mu.Unlock()
	if ok {
		return body, nil
	}
	if r.c.env.Fetch == nil {
		return nil, fmt.Errorf("this node reads no archive (%s)", url)
	}
	// Bounded: a store that does not answer must not hold the composer or the node's answer (issue 1126).
	defer r.c.waiting("the archive " + url)()
	fetching, cancel := context.WithTimeout(ctx, archiveWithin)
	defer cancel()
	body, err := r.c.env.Fetch(fetching, url)
	if r.c.note(err) != nil {
		return nil, err
	}
	sum := sha256.Sum256(body)
	if got := "sha256:" + hex.EncodeToString(sum[:]); got != digest {
		return nil, fmt.Errorf("%s was recorded as %s and what arrived is %s", url, digest, got)
	}
	r.c.mu.Lock()
	r.c.archives[digest] = body
	r.c.mu.Unlock()
	return body, nil
}

// Served reads, of another node's module, the keys named of mesh-config's resolved values, one key each: the node is
// granted those keys alone (buscompose Principal.ServedReads, issue 1126).
func (r reader) Served(ctx context.Context, node, module string, keys []string) (map[string]any, error) {
	defer r.c.waiting("what " + module + " serves on " + node + " (" + ResolvedBucket + ")")()
	out := map[string]any{}
	for _, k := range keys {
		raw, err := r.c.get(ctx, ResolvedBucket, node+"."+module+"."+k)
		if err != nil {
			return nil, err
		}
		value, laid := catalogue.LaidValue(raw)
		if raw == nil || !laid {
			continue
		}
		out[k] = value
	}
	if len(out) == 0 {
		return nil, nil
	}
	return out, nil
}

func (r reader) Values(_ context.Context, node, module string) (map[string]any, error) {
	if node != r.c.env.Node {
		return nil, nil // a node reads its own slice alone (ADR 0547 §2)
	}
	r.c.mu.Lock()
	defer r.c.mu.Unlock()
	if len(r.c.values[module]) == 0 {
		return nil, nil
	}
	out := make(map[string]any, len(r.c.values[module]))
	for k, v := range r.c.values[module] {
		out[k] = v
	}
	return out, nil
}

func (r reader) OwnSecrets(ctx context.Context, node string) ([]compose.OwnSecretRow, error) {
	if node != r.c.env.Node || r.c.env.Own == nil {
		return nil, nil
	}
	done := r.c.waiting("the vault's own secrets")
	raw, err := r.c.env.Own(ctx, map[string]any{})
	r.c.note(err)
	done()
	if err != nil {
		return nil, fmt.Errorf("the vault's own: %w", err)
	}
	var a struct {
		Node string `json:"node"`
		Key  string `json:"key"`
		Own  []struct {
			Module     string          `json:"module"`
			Name       string          `json:"name"`
			Origin     string          `json:"origin"`
			Generation int             `json:"generation"`
			Sealed     json.RawMessage `json:"sealed"`
		} `json:"own"`
	}
	if err := json.Unmarshal(raw, &a); err != nil {
		return nil, fmt.Errorf("the vault's own answer is not one: %w", err)
	}
	if a.Node != node {
		return nil, fmt.Errorf("the vault answered %q's own secrets to %s", a.Node, node)
	}
	r.c.mu.Lock()
	defer r.c.mu.Unlock()
	out := make([]compose.OwnSecretRow, 0, len(a.Own))
	for _, o := range a.Own {
		// The box first had is kept while its generation stands, so sealing it again moves no composition; the box the
		// node runs at the same slot is kept after the composition (keepRunning).
		id := o.Module + "/" + o.Name
		kept, ok := r.c.sealedOwn[id]
		if !ok || kept.generation != o.Generation || kept.key != a.Key {
			kept = ownSealed{generation: o.Generation, key: a.Key, sealed: string(o.Sealed)}
			r.c.sealedOwn[id] = kept
		}
		out = append(out, compose.OwnSecretRow{Node: node, Module: o.Module, Name: o.Name, Origin: o.Origin,
			SealedTo: a.Key, Sealed: kept.sealed})
	}
	return out, nil
}

func (r reader) Credentials(ctx context.Context, node string, asConsumer, asProvider []compose.CredentialKey) ([]compose.Credential, []compose.Issued, error) {
	if node != r.c.env.Node || r.c.env.Own == nil {
		return nil, nil, nil
	}
	type asked struct {
		Role string `json:"role"`
		compose.CredentialKey
	}
	var ask []asked
	for _, k := range asConsumer {
		ask = append(ask, asked{Role: "consumer", CredentialKey: k})
	}
	for _, k := range asProvider {
		ask = append(ask, asked{Role: "provider", CredentialKey: k})
	}
	done := r.c.waiting("the vault's credentials")
	raw, err := r.c.env.Own(ctx, map[string]any{"credentials": ask})
	r.c.note(err)
	done()
	if err != nil {
		return nil, nil, fmt.Errorf("the vault's own: %w", err)
	}
	var a struct {
		Node        string `json:"node"`
		Key         string `json:"key"`
		Credentials []struct {
			Role string `json:"role"`
			compose.CredentialKey
			Name       string          `json:"name"`
			Generation int             `json:"generation"`
			Sealed     json.RawMessage `json:"sealed"`
		} `json:"credentials"`
	}
	if err := json.Unmarshal(raw, &a); err != nil {
		return nil, nil, fmt.Errorf("the vault's own answer is not one: %w", err)
	}
	if a.Node != node {
		return nil, nil, fmt.Errorf("the vault answered %q's credentials to %s", a.Node, node)
	}
	r.c.mu.Lock()
	defer r.c.mu.Unlock()
	var creds []compose.Credential
	var issued []compose.Issued
	for _, c := range a.Credentials {
		// The value first had is kept while its generation stands, so sealing it again moves no composition; the box the
		// node runs at the same slot is kept after the composition (keepRunning).
		id := c.Role + "|" + c.Name + "|" + c.Provision + "|" + c.Consumer + "|" + c.ConsumerModule + "|" + c.Local
		kept, ok := r.c.sealedOwn[id]
		if !ok || kept.generation != c.Generation || kept.key != a.Key {
			kept = ownSealed{generation: c.Generation, key: a.Key, sealed: string(c.Sealed)}
			r.c.sealedOwn[id] = kept
		}
		k := c.CredentialKey
		switch c.Role {
		case "consumer":
			creds = append(creds, compose.Credential{Provision: k.Provision, Consumer: k.Consumer, ConsumerModule: k.ConsumerModule,
				Provider: k.Provider, Local: k.Local, State: compose.CredentialHeld, Sealed: kept.sealed})
		case "provider":
			issued = append(issued, compose.Issued{Provision: k.Provision, Consumer: k.Consumer, ConsumerModule: k.ConsumerModule,
				Local: k.Local, Provider: k.Provider, ForProvider: kept.sealed})
		}
	}
	return creds, issued, nil
}

// Placed reads whether the catalogue places node now, for the node agent choosing whose compositions it applies at
// each start: true when its placement key holds a value, false only when a read succeeds and finds no key or a delete
// marker. A read that fails chooses neither (review of the shadow, round 1: a timeout or a bus not ready at a restart
// flipped a self-composing node to the controller's compositions for a whole run): it is tried again after wait(try),
// each distinct failure said once, until ctx ends, which is then its error.
//
// switched says the node composed from its own placement key before (its have says it composes its own,
// delivery.ComposesOwn, and that is kept on the bus, so it survives a restart). Only a node that never switched takes a
// read that fails (a timeout, or a state its account may not read yet, as the seat's before its grant arrives) as "not
// placed": it stays on the controller's compositions, says why once, and switches on a positive read of its key (the
// watch endWhenPlaced keeps). A node that switched never goes back on a failed read, however long it lasts.
func Placed(ctx context.Context, get func(ctx context.Context, bucket, key string) ([]byte, error), node string,
	wait func(try int) time.Duration, say func(string), switched bool) (bool, error) {
	key, err := delivery.PlacementKey(node)
	if err != nil {
		return false, err
	}
	said := ""
	for try := 1; ; try++ {
		raw, err := get(ctx, delivery.PlacementBucket, key)
		if errors.Is(err, nats.ErrBucketNotFound) {
			return false, nil // the seat has written no placement state yet: nothing places it, read and known
		}
		if err == nil {
			return len(raw) > 0, nil
		}
		if ctx.Err() != nil {
			return false, ctx.Err()
		}
		if !switched {
			say(fmt.Sprintf("this node's key in the seat's placement state cannot be read (%s): it has never composed its own, so it applies the controller's compositions and switches when its key is read: %v", key, err))
			return false, nil
		}
		if why := err.Error(); why != said {
			say(fmt.Sprintf("this node's key in the seat's placement state cannot be read (%s): the node agent waits, choosing neither its own compositions nor the controller's: %v", key, err))
			said = why
		}
		select {
		case <-ctx.Done():
			return false, ctx.Err()
		case <-time.After(wait(try)):
		}
	}
}

// ComposedItsOwn says whether the node's own have, kept on the bus, says it composed from its placement key at its last
// start. A have that cannot be read says yes: the safe side, a node waits and does not fall back.
func ComposedItsOwn(ctx context.Context, get func(ctx context.Context, bucket, key string) ([]byte, error), node string) bool {
	key, err := delivery.HaveNodeKey(node)
	if err != nil {
		return true
	}
	raw, err := get(ctx, delivery.HaveBucket, key)
	if err != nil {
		return true
	}
	if len(raw) == 0 {
		return false
	}
	n, err := delivery.DecodeHaveNode(raw)
	return err != nil || n.Composes == delivery.ComposesOwn
}

// composeAgain are the waits before a pass that met a bus not answering is tried again.
var composeAgain = []time.Duration{time.Second, 2 * time.Second, 4 * time.Second}

// laterAgain are the waits, after a pass or an answer refused for a bus that did not answer past composeAgain, before
// it is made again; the last repeats while the bus does not answer (issue 1126, 2026-10-28: a failed read was kept as
// the answer until another input moved).
var laterAgain = []time.Duration{5 * time.Second, 15 * time.Second, 30 * time.Second, time.Minute, 2 * time.Minute, 5 * time.Minute}

// LaterWait is the wait before the try'th making again (from 0), of a pass, an answer or an ask the bus did not answer.
func LaterWait(try int) time.Duration { return laterAgain[min(try, len(laterAgain)-1)] }

// ErrBusMissed is a pass that met a read the bus did not answer, though the composition was made around it (a have of
// sealed values that could not be read): never handed as made.
var ErrBusMissed = errors.New("a read on the bus did not answer during the pass")

// missedIn is err, or ErrBusMissed when no error came but a read since missed failed on the bus.
func (c *Composer) missedIn(missed uint64, err error) error {
	if err == nil && c.missed.Load() > missed {
		return ErrBusMissed
	}
	return err
}

// BusFailed says err is a read the bus did not answer, for any reason but the run's own end: Transient, a read that
// timed out, or no link to the bus. Such a read is tried again after a bound, never kept as an answer; a key absent, a
// value refused or a verdict is an answer (issue 1126, 2026-10-28). Matched by value and by words, as Transient.
func BusFailed(err error) bool {
	if err == nil || errors.Is(err, context.Canceled) {
		return false
	}
	if Transient(err) || errors.Is(err, context.DeadlineExceeded) || errors.Is(err, nats.ErrTimeout) ||
		errors.Is(err, nats.ErrConnectionClosed) || errors.Is(err, nats.ErrDisconnected) || errors.Is(err, nats.ErrNoServers) {
		return true
	}
	words := err.Error()
	for _, w := range []string{"timeout", "deadline exceeded", "no link to the bus", "did not answer", "connection closed"} {
		if strings.Contains(words, w) {
			return true
		}
	}
	return false
}

// Transient says err is the bus not answering for a moment, as a reload of its accounts makes it for a few seconds:
// JetStream not enabled, or no responders; each fails at once, so a pass tried again costs only its wait. A read that
// timed out is not one: it already waited longer than such a pause, and is most often a read the bus refuses. Matched
// by value and by words, since the vault's and the reads' errors arrive wrapped in words (issue 1126).
func Transient(err error) bool {
	if err == nil {
		return false
	}
	if errors.Is(err, nats.ErrJetStreamNotEnabled) || errors.Is(err, nats.ErrJetStreamNotEnabledForAccount) || errors.Is(err, nats.ErrNoResponders) ||
		errors.Is(err, ErrBusMissed) {
		return true
	}
	words := err.Error()
	for _, w := range []string{"jetstream not enabled", "no responders"} {
		if strings.Contains(words, w) {
			return true
		}
	}
	return false
}