mesh-host/cmd/mesh-host/nodeagent.go

package main

import (
	"context"
	"crypto/ed25519"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"sort"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"git.novox.be/novox/mesh-sdk/go/declaration"
	"git.novox.be/novox/mesh-sdk/go/delivery"
	"git.novox.be/novox/mesh-sdk/go/overlay"
	"github.com/novox/mesh-host/internal/agent"
	"github.com/novox/mesh-host/internal/apply"
	"github.com/novox/mesh-host/internal/compose/frombus"
	"github.com/novox/mesh-host/internal/link"
	"github.com/novox/mesh-host/internal/liveness"
	"github.com/novox/mesh-host/internal/selfcompose"
	"github.com/novox/mesh-host/internal/store"
)

// The node agent's wiring (novox/hq ADR 0429 §4, ADR 0539 §9): the buckets it watches, the haves it writes, the backup
// it asks and the apply of one module's composition.

// bucketWatch is a bucket of the bus as the agent watches it, through the session's link.
type bucketWatch struct {
	session *link.Session
	bucket  string
}

func (w bucketWatch) Watch(ctx context.Context, pattern string) (<-chan agent.Update, error) {
	in, err := w.session.KVWatch(ctx, w.bucket, pattern)
	if err != nil {
		return nil, err
	}
	return w.updates(ctx, in), nil
}

// WatchAwait is Watch for a state that its holder makes when it starts (the placement seat's): a bucket not on the bus
// yet is waited for on the bus's word that it was created (link.Session.KVWatchAwait), and waiting is told why.
func (w bucketWatch) WatchAwait(ctx context.Context, pattern string, waiting func(error)) (<-chan agent.Update, error) {
	in, err := w.session.KVWatchAwait(ctx, w.bucket, pattern, waiting)
	if err != nil {
		return nil, err
	}
	return w.updates(ctx, in), nil
}

func (w bucketWatch) updates(ctx context.Context, in <-chan link.KVUpdate) <-chan agent.Update {
	out := make(chan agent.Update, 64)
	go func() {
		defer close(out)
		for u := range in {
			select {
			case out <- agent.Update{Key: u.Key, Value: u.Value, Revision: u.Revision, Deleted: u.Deleted, Caught: u.Caught}:
			case <-ctx.Done():
				return
			}
		}
	}()
	return out
}

// sessionHaves writes the haves the agent says, through the session.
type sessionHaves struct{ session *link.Session }

func (h sessionHaves) Read(ctx context.Context, module string) (agent.Have, bool) {
	if module == agent.NodeModule {
		// The node's own key: the digest of its own composition it runs, and what that carries sealed.
		got, ok := h.session.ReadNodeHave(ctx)
		if !ok || got.Composed == "" {
			return agent.Have{}, false
		}
		return agent.Have{Module: module, Composed: got.Composed, Sealed: got.SealedAt}, true
	}
	got, ok := h.session.ReadHave(ctx, module)
	if !ok {
		return agent.Have{}, false
	}
	return agent.Have{Module: module, FP: got.FP, Composed: got.Composed, Phase: string(got.Phase), Detail: got.Detail,
		Sealed: got.SealedAt}, true
}

// ReadErr is Read telling a have absent from one the bus could not read (agent.HavesReadErr).
func (h sessionHaves) ReadErr(ctx context.Context, module string) (agent.Have, bool, error) {
	if module == agent.NodeModule {
		// The node's own key, its read the bus did not answer told as such (issue 1126, 2026-10-28).
		got, ok, err := h.session.ReadNodeHaveErr(ctx)
		if !ok || err != nil || got.Composed == "" {
			return agent.Have{}, false, err
		}
		return agent.Have{Module: module, Composed: got.Composed, Sealed: got.SealedAt}, true, nil
	}
	got, ok, err := h.session.ReadHaveErr(ctx, module)
	if !ok || err != nil {
		return agent.Have{}, false, err
	}
	return agent.Have{Module: module, FP: got.FP, Composed: got.Composed, Phase: string(got.Phase), Detail: got.Detail,
		Sealed: got.SealedAt}, true, nil
}

func (h sessionHaves) Write(ctx context.Context, x agent.Have) {
	if x.Module == agent.NodeModule {
		// The node's own key holds its facts and the digest of the composition it runs.
		if x.Phase == "" || x.Phase == string(delivery.PhaseRunning) {
			h.session.NoteNode(ctx, x.Composed, x.Sealed)
		}
		return
	}
	phase := x.Phase
	if phase == string(delivery.PhaseRunning) {
		phase = ""
	}
	h.session.NoteModule(ctx, link.ModuleFact{Module: x.Module, FP: x.FP, Composed: x.Composed, Phase: phase,
		Failed: phase == string(delivery.PhaseFailed), Detail: shorten(x.Detail), Judgement: x.Judgement, Waits: x.Waits,
		BuildAlone: x.BuildAlone, Kept: x.Kept, Sealed: x.Sealed})
}

func (h sessionHaves) Removed(ctx context.Context, module string) {
	h.session.NoteRemoved(ctx, module, "")
}

// shorten keeps a have's detail within what a have value holds: its head, which says what failed, and its tail, which is
// the cause (an error wrapped five times names the bus's "no responders" only at its end: 2026-10-25 23:04, mesh-vault on
// the bus node said only which object it read).
func shorten(s string) string {
	const most, head = 300, 100
	if len(s) <= most {
		return s
	}
	const gap = " … "
	tail := s[len(s)-(most-head-len(gap)):]
	return strings.ToValidUTF8(s[:head], "") + gap + strings.ToValidUTF8(tail, "")
}

// failedWith is a failed apply that says why, transient when the words are the bus not answering for a moment
// (link.TransientWords): the agent then tries it again without a new revision.
func failedWith(why string) agent.Result {
	return agent.Result{Failed: true, Transient: link.TransientWords(why), Detail: shorten(why)}
}

// backupWithin is how long a backup is given from its ask to say it is done (ADR 0429 §14).
var backupWithin = 10 * time.Minute

// backedUpEvent is where the holder of node-backup says a backup taken by hand ended, also one that could not start.
const backedUpEvent = "mesh.mod.restic.event.backed-up"

// backupNode asks the holder of node-backup on this node to back a module up now, and waits for its event to say how it
// went: the holder answers `now` when it has started, and says `backed-up` when the backup ends.
func backupNode(session *link.Session) func(ctx context.Context, module string) agent.Backup {
	return func(ctx context.Context, module string) agent.Backup {
		// Heard before the ask, so an event that comes at once is not lost.
		events, stop, err := session.Hear(backedUpEvent)
		if err != nil {
			return agent.Backup{Words: "the end of the backup cannot be heard: " + err.Error()}
		}
		defer stop()
		// A module this machine backs nothing of up moves without a backup: never held for one that cannot run (issue
		// 1945). An answer that cannot be had or read asks for the backup as before.
		ask := func(verb string) func(context.Context) (json.RawMessage, error) {
			return func(ctx context.Context) (json.RawMessage, error) {
				return session.AskSeat(ctx, "node-backup", verb, map[string]any{"module": module}, 30*time.Second)
			}
		}
		if said, err := askUntilHeard(ctx, ask("backed-up"), backupAskFor, link.Backoff); err == nil {
			if nothing, read := declaresNothingToBackUp(said, module); read && nothing {
				return agent.Backup{Done: true, Words: module + " declares nothing to back up on this machine"}
			}
		}
		asked := time.Now()
		if _, err := askUntilHeard(ctx, ask("now"), backupAskFor, link.Backoff); err != nil {
			return agent.Backup{Words: err.Error()}
		}
		return awaitBackup(ctx, events, session.Membership.Node, module, asked, backupWithin)
	}
}

// backupAskFor is how long the backup's ask is asked again while it is known never to have reached the holder of
// node-backup: nothing on this machine answers it (the tool runner restarting, its subscriptions not made yet) or no
// link is open. The sweep of 2026-10-30 met it on ace's redis, postgres and restic (14:03), step-ca (13:55) and
// mesh-vault on novox (15:22): each was held on one unanswered ask. Past the bound the backup failed, said with how
// often and how long it was asked: the module is held, its have says so, and the assigner asks the operator.
var backupAskFor = 3 * time.Minute

// askUntilHeard asks, and asks again after wait(try) while the answer says the ask was never delivered (link.ErrNotAsked),
// for at most within: an ask that may have reached its holder (a timeout, a refusal in words) is never asked twice, so a
// backup is never started twice.
func askUntilHeard(ctx context.Context, ask func(context.Context) (json.RawMessage, error), within time.Duration,
	wait func(int) time.Duration) (json.RawMessage, error) {
	began := time.Now()
	for try := 1; ; try++ {
		said, err := ask(ctx)
		if err == nil || !errors.Is(err, link.ErrNotAsked) {
			return said, err
		}
		pause := wait(try)
		if time.Since(began)+pause > within {
			return nil, fmt.Errorf("asked %d times over %s and never heard: %w", try, time.Since(began).Round(time.Second), err)
		}
		select {
		case <-ctx.Done():
			return nil, fmt.Errorf("the node agent was stopped while it asked again: %w", err)
		case <-time.After(pause):
		}
	}
}

// awaitBackup waits for the event that says the backup of a module on this node ended, within the bound: done when it
// says ok, held on its words when it does not, and held when it says nothing within the bound. An event of another node
// or module, or from before the ask, is not the answer.
func awaitBackup(ctx context.Context, events <-chan []byte, node, module string, since time.Time, within time.Duration) agent.Backup {
	bound := time.NewTimer(within)
	defer bound.Stop()
	for {
		select {
		case <-ctx.Done():
			return agent.Backup{Words: agent.BackupInterrupted, Interrupted: true}
		case <-bound.C:
			return agent.Backup{Words: fmt.Sprintf("the backup did not say it was done within %s", within)}
		case raw := <-events:
			var e struct {
				Node   string    `json:"node"`
				Module string    `json:"module"`
				OK     bool      `json:"ok"`
				At     time.Time `json:"at"`
				Error  string    `json:"error"`
			}
			if json.Unmarshal(raw, &e) != nil || e.Node != node || e.Module != module || e.At.Before(since.Add(-time.Second)) {
				continue
			}
			if e.OK {
				return agent.Backup{Done: true}
			}
			words := "the backup failed"
			if e.Error != "" {
				words += ": " + e.Error
			}
			return agent.Backup{Words: words}
		}
	}
}

// declarationOf is the declaration the compositions make together: their resources, and the node's own parts when the
// node's composition is among them. It is what the node judges and says its health by, and what one module's apply is
// made of (with a single composition).
//
// **A resource the node's own composition shares with a module's is the module's** (issue 1945): from 2026-10-25 12:43
// the controller put modules it knew no fingerprint of the new form for into the node's own composition while their
// wants named their own, and the declaration joined of both declared each such file twice, which the filling refuses
// ("2 resources declare X") on every node. The node's own copy is left out, and which were left out is said once.
func declarationOf(cs ...delivery.Composed) (*declaration.Declaration, error) {
	d := &declaration.Declaration{Version: declaration.Version}
	var modules []delivery.Composed
	for _, c := range cs {
		if c.Module != "" {
			modules = append(modules, c)
		}
	}
	if len(modules) > 0 {
		joined := make([]delivery.Composed, 0, len(cs))
		for _, c := range cs {
			if c.Module == "" {
				var dropped []string
				c, dropped = ownWithout(c, modules)
				sayDroppedOnce(dropped)
			}
			joined = append(joined, c)
		}
		cs = joined
	}
	for _, c := range cs {
		for _, raw := range c.Resources {
			r, err := declaration.ParseResource(raw)
			if err != nil {
				return nil, fmt.Errorf("a resource of %q: %w", c.Module, err)
			}
			d.Resources = append(d.Resources, r)
		}
		d.Filled = append(d.Filled, c.Filled...)
		if c.Module == "" {
			d.Adoption = c.Own.Adoption
			d.Identities = append(d.Identities, c.Own.Identities...)
			d.Groups = append(d.Groups, c.Own.Groups...)
			d.Libraries = append(d.Libraries, c.Own.Libraries...)
			d.Provisioner = c.Own.Provisioner
		}
	}
	return d, nil
}

// ownWithout is the node's own composition without the resources a module's composition among modules also declares:
// the same kind and id. A resource that shares only its target (a unit or container of the same name, another id) is
// not the module's, and stays: dropped, its record would be an orphan the apply removes, stopping the module's unit. It
// answers what it left out, as "<id> (<module>)".
func ownWithout(own delivery.Composed, modules []delivery.Composed) (delivery.Composed, []string) {
	byID := map[string]string{}
	for _, c := range modules {
		if c.Module == "" {
			continue
		}
		for _, raw := range c.Resources {
			r, err := declaration.ParseResource(raw)
			if err != nil {
				continue
			}
			byID[string(r.Kind())+"\x00"+r.Identity()] = c.Module
		}
	}
	if len(byID) == 0 {
		return own, nil
	}
	var dropped []string
	kept := make([]json.RawMessage, 0, len(own.Resources))
	for _, raw := range own.Resources {
		r, err := declaration.ParseResource(raw)
		if err != nil {
			kept = append(kept, raw) // refused where it is parsed, with its words
			continue
		}
		module, dup := byID[string(r.Kind())+"\x00"+r.Identity()]
		if dup {
			dropped = append(dropped, fmt.Sprintf("%s (%s)", r.Identity(), module))
			continue
		}
		kept = append(kept, raw)
	}
	if len(dropped) == 0 {
		return own, nil
	}
	own.Resources = kept
	return own, dropped
}

// droppedSaid is every set of the node's own resources left out that was said, so each is said once.
var droppedSaid sync.Map

// sayDroppedOnce says once which of the node's own resources were left out because a module's composition declares them.
func sayDroppedOnce(dropped []string) {
	if len(dropped) == 0 {
		return
	}
	sort.Strings(dropped)
	key := strings.Join(dropped, ", ")
	if _, said := droppedSaid.LoadOrStore(key, true); said {
		return
	}
	fmt.Fprintf(os.Stderr, "mesh-host: the node's own composition also declares %d resource(s) a module's composition "+
		"declares, and they are left out of it: %s (issue 1945)\n", len(dropped), key)
}

func scopeName(c delivery.Composed) string {
	if c.Module == "" {
		return agent.NodeModule
	}
	return c.Module
}

func resourcesOf(cs []delivery.Composed) []declaration.Resource {
	var out []declaration.Resource
	for _, c := range cs {
		for _, raw := range c.Resources {
			if r, err := declaration.ParseResource(raw); err == nil {
				out = append(out, r)
			}
		}
	}
	return out
}

// keptOnce reads the kept declaration once and splits it by module: its resources (the node's own are those that say
// none, ADR 0539 §9) and its filled paths. Once the read is retired (keptRetired) nothing is read.
func keptOnce(opts options, signer ed25519.PublicKey) func() (map[string][]json.RawMessage, map[string][]string) {
	var once sync.Once
	var by map[string][]json.RawMessage
	var filled map[string][]string
	return func() (map[string][]json.RawMessage, map[string][]string) {
		once.Do(func() {
			if keptRetired.Load() {
				return
			}
			raw, err := store.LoadDeclared(store.DeclaredPath(opts.state), signer)
			if err != nil {
				return
			}
			by, filled = splitKept(raw)
		})
		return by, filled
	}
}

// splitKept is a kept declaration's resources by the module they say (the node's own under agent.NodeModule) and its
// filled paths by the module whose file is at each path; nil for a declaration that does not read.
func splitKept(raw []byte) (map[string][]json.RawMessage, map[string][]string) {
	var kept struct {
		Resources []json.RawMessage `json:"resources"`
		Filled    []string          `json:"filled_files"`
	}
	if json.Unmarshal(raw, &kept) != nil {
		return nil, nil
	}
	isFilled := make(map[string]bool, len(kept.Filled))
	for _, p := range kept.Filled {
		isFilled[p] = true
	}
	by, filled := map[string][]json.RawMessage{}, map[string][]string{}
	named := map[string]bool{}
	for _, r := range kept.Resources {
		var head struct {
			Module string `json:"module"`
			Type   string `json:"type"`
			Path   string `json:"path"`
		}
		_ = json.Unmarshal(r, &head)
		name := head.Module
		if name == "" {
			name = agent.NodeModule
		}
		by[name] = append(by[name], r)
		if head.Type == string(declaration.TypeFile) && isFilled[head.Path] && !named[name+"\x00"+head.Path] {
			named[name+"\x00"+head.Path] = true
			filled[name] = append(filled[name], head.Path)
		}
	}
	return by, filled
}

// nodeAgent runs the node agent under the session's links for as long as ctx lasts: it waits for a link, follows the
// node's wants and compositions on it, and begins again with the next link. A watch it cannot make (a bucket this node's
// account may not read) is said, and tried again with a growing wait, as the link is.
func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trust *meshTrust, session *link.Session, sched *apply.Scheduler,
	say link.Announce, afterApply func(), current *agentNow, sayFacts func(ctx context.Context, ag *agent.Agent)) {
	wait := 2 * time.Second
	for ctx.Err() == nil {
		linked, cancel, err := session.WhenLinked(ctx)
		if err != nil {
			return
		}
		began := time.Now()
		node := session.Membership.Node
		composedBucket, err := delivery.ComposedBucketFor(node)
		if err != nil {
			cancel()
			say("the node agent cannot name this node's composition bucket: " + err.Error())
			return
		}
		var ag *agent.Agent
		// The kept release-5 declaration is no longer read once every module the seat's placement puts here has a have of its
		// own (ADR 0550 §2: its signature is the controller's, trusted no more after the lease write).
		// Nor before this engine keeps the node's own as it runs (applied/node.json): until then the kept declaration is the one
		// record of what the node's own composition carries, its folded modules among them (issue 774, review of the
		// re-land, round 1: novox's first start would compose its own with nats running nothing).
		if retired, why := placedAllHave(linked, session, node); retired && !keptRetired.Load() {
			if !appliedStoreFor(opts.state).holds(agent.NodeModule) {
				say("the kept declaration is still read: " + why + ", but this engine keeps no node's own composition yet")
			} else if !keptRetired.Swap(true) {
				say("the kept declaration is no longer read: " + why)
			}
		}
		kept := keptOnce(opts, signer)
		// The mesh's own certificate the declaration kept from before carries, kept for the record before this machine
		// composes its own (ADR 0555 §3): without it a module that asks for a certificate would be stood in for.
		if byModule, _ := kept(); byModule != nil {
			for module, resources := range byModule {
				if err := keepAuthority(opts.state, delivery.Composed{Module: module, Resources: resources}); err != nil {
					say("the mesh's certificate the kept declaration carries could not be kept for this machine's record: " + err.Error())
				}
			}
		}
		// The bus's user list, the grants part's alone (ADR 0547 §8, clause 11; issue 1126), written into the file the
		// bus module this node runs declares, as its agent holds what runs (its own composition, or the declaration kept
		// before the switch): never read from the node's composition bucket, which the controller wrote. On every node;
		// a node that runs no list writes nothing. Looked at again whenever what runs changed (RunsChanged, Settled).
		users := userListFollower(opts, session, func() []delivery.Composed {
			if now := current.get(); now != nil {
				return now.Everything()
			}
			return nil
		}, say, func() {
			// This engine wrote the list, so the bus reloads its accounts and answers no JetStream request for a few
			// seconds: what failed in them is tried again once that is over, not at its next wait.
			time.AfterFunc(reloadSettles, func() {
				if now := current.get(); now != nil {
					now.RetryTransient()
				}
			})
		})
		go func() {
			if err := users.Run(linked); err != nil && linked.Err() == nil {
				say("the bus's user list is not followed on this link: " + err.Error())
			}
		}()
		env := agent.Env{
			Node:     node,
			Wants:    bucketWatch{session, delivery.AssignerWantsBucket},
			Composed: bucketWatch{session, composedBucket},
			Haves:    sessionHaves{session},
			Kept:     func() map[string][]json.RawMessage { by, _ := kept(); return by },
			// The same kept declaration's filled paths, so a module still running from it keeps its files filled.
			KeptFilled: func() map[string][]string { _, filled := kept(); return filled },
			Recorded:   recordedIn(opts.state, say),
			// What this engine applied, kept on the machine and read at the start, so what runs is known before the node
			// composes its own (issue 774).
			Applied:  appliedStoreFor(opts.state),
			Inactive: inactiveIn(&liveness.Exec{Run: liveness.Runner(opts.run()), System: builtFor}, heldBy(apply.WindowsIn(filepath.Dir(opts.state)))),
			Held: func() map[string]string {
				if c := composerNow.Load(); c != nil {
					return c.LastHeld()
				}
				return nil
			},
			Revisions: func(ctx context.Context, key string) ([]agent.Update, error) {
				got, err := session.KVHistory(ctx, composedBucket, key)
				if err != nil {
					return nil, err
				}
				out := make([]agent.Update, 0, len(got))
				for _, u := range got {
					out = append(out, agent.Update{Key: u.Key, Value: u.Value, Revision: u.Revision, Deleted: u.Deleted})
				}
				return out, nil
			},
			Backup: backupNode(session),
			// What the filling refused before this apply may be named now: the provisioner's client is told to fill
			// again (ace, 2026-10-25 21:17).
			Settled: func(module string) {
				knockFilling(module+" settled", say)
				judgeWhatRunsNow(opts.state, signer)
				users.RunsChanged()
			},
			// And once the haves are read at the start, and whenever a module is taken as running without an apply, so
			// a refusal from the engine's start does not wait for some module to settle (issue 924).
			RunsChanged: func(why string) {
				knockFilling(why, say)
				judgeWhatRunsNow(opts.state, signer)
				users.RunsChanged()
			},
			Say: say,
			// A failure of the bus not answering for a moment is tried again: 2 s, 5 s, 15 s, 60 s, then every 5 min.
			Backoff: link.Backoff,
			Apply: func(ctx context.Context, c delivery.Composed, others []delivery.Composed) agent.Result {
				actx, done := applyContext(session.WithLink(ctx))
				// While the grants part's list is followed, the controller's list in the bus module's composition is not
				// written: the apply keeps the list on the machine (ADR 0547 clause 8, never both).
				held, herr := users.Hold(c)
				var res agent.Result
				if herr != nil {
					res = failedWith(herr.Error())
				} else {
					res = applyComposition(actx, opts, held, others, ag.Placed(scopeName(c)), ag.RemovedByWant(), sched, say)
				}
				done()
				// The mesh's own certificate it carries is kept for this machine's record (ADR 0555 §3).
				if !res.Failed {
					if err := keepAuthority(opts.state, c); err != nil {
						say("the mesh's certificate in " + c.Module + "'s composition could not be kept for this machine's record: " + err.Error())
					}
				}
				// An apply of the bus module may write an older list than the newest composition holds: looked at again.
				if module, _ := users.State(); module != "" && module == c.Module {
					users.Again()
				}
				afterApply()
				sayFacts(ctx, ag)
				return res
			},
			Remove: func(ctx context.Context, module string, others []delivery.Composed) agent.Result {
				actx, done := applyContext(session.WithLink(ctx))
				res := removeComposition(actx, opts, module, others, ag.Placed(module), sched, say)
				done()
				afterApply()
				sayFacts(ctx, ag)
				return res
			},
		}
		// The node composes its own modules (novox/hq ADR 0547 §2, ADR 0549, issue 1126).
		run, stopRun := context.WithCancel(linked)
		// What the node runs and how it opens a sealed value, for the composer to keep the boxes it runs; and the agent
		// told when the node cannot compose, so each wanted module's have says it.
		// The agent of this run alone: its composer tells it, never the agent of a run begun since (issue 1126: a pass in
		// flight at the switch marked the next agent's modules failed).
		own := &agentNow{}
		hooks := composerHooks{
			running: func() []delivery.Composed {
				if now := own.get(); now != nil {
					return runsAsWritten(now.Everything(), opts.state)
				}
				return nil
			},
			unseal: sealOpener(opts.state),
			// And what each module's have says it runs sealed: after a restart the agent has applied nothing yet.
			// A have the bus did not answer for is said as such, never as one holding no seals (issue 1126, 2026-10-28).
			held: func(module string) (delivery.Seals, []string, error) {
				if module == agent.NodeModule {
					got, _, err := session.ReadNodeHaveErr(linked)
					return got.SealedAt, got.Sealed, err
				}
				got, ok, err := session.ReadHaveErr(linked, module)
				if !ok {
					return nil, nil, err
				}
				return got.SealedAt, got.Sealed, nil
			},
			refused: func(why string, held map[string]string) {
				if now := own.get(); now != nil && run.Err() == nil {
					now.NotComposedEach(run, why, held) // a module held alone marks that module, never the others (issue 1126)
				}
			},
			// And what it keeps of a module as it runs, for a manifest that does not say it yet: on the module's have (ADR 0601 §4).
			kept: func(kept map[string][]string) {
				if now := own.get(); now != nil && run.Err() == nil {
					now.KeptNow(run, kept)
				}
			},
			// And the vault's authority beside the controller's, and this node's certificate asked of the vault when a
			// module here asks for one (ADR 0550 §2).
			vaultAuthority: trust.VaultAuthority,
			certify: func(ctx context.Context) error {
				return certifyThisNode(ctx, opts.state, node, trust, session.AskVault, func() { sayFacts(ctx, ag) })
			},
		}
		// And its answer to every placement candidate, placed or not, applying nothing (novox/hq ADR 0559 §4): written into
		// its own key, `<node>.node`, under candidates.
		shadowHooks := hooks
		shadowHooks.refused, shadowHooks.kept, shadowHooks.certify = nil, nil, nil
		shadowEnv := composerEnv(session, node, say, shadowHooks)
		shadow := selfcompose.NewShadow(shadowEnv,
			func(ctx context.Context, answers map[string]delivery.CandidateAnswer) {
				session.NoteCandidates(ctx, delivery.WithValuesDiffer(answers))
			})
		go shadow.Keep(run, link.Backoff) // this run's alone: a switch begins a run, and its shadow, again
		shadowNow.Store(shadow)           // what the verb compose-candidate composes a check's candidate with
		// Its own composer, always (issue 1126, A7): a composition made elsewhere (the controller's, in the node's
		// composition bucket) is never applied. While the seat's placement holds no key for this node it composes
		// nothing and keeps running what it runs; a placement it cannot watch stops the agent, said, and it begins again.
		composer := selfComposer(session, node, hooks, say)
		machinesNow.Store(composer) // the machines a failing network part points at, by the placements (issue 774)
		composerNow.Store(composer) // what the verb compose asks a pass of, this run's alone
		env.Composed, env.SelfComposed, env.Policy, env.ComposedToday = composer, true, composer.Policy, composer.ComposedToday
		env.HoldsData = composer.HoldsData // what runs from the kept declaration says Data as its manifest does
		session.NoteComposes(linked, delivery.ComposesOwn, "it composes its own modules, and applies no composition made elsewhere")
		ag = agent.New(env)
		own.set(ag)
		current.set(ag)
		err = ag.Run(run)
		stopRun()
		cancel()
		current.set(nil)
		composerNow.Store(nil)
		shadowNow.Store(nil)
		if ctx.Err() != nil {
			return
		}
		if time.Since(began) > 30*time.Second {
			wait = 2 * time.Second
		}
		if err != nil {
			say(fmt.Sprintf("the node agent stopped: %v — beginning again in %s", err, wait))
		}
		select {
		case <-ctx.Done():
			return
		case <-time.After(wait):
		}
		if wait *= 2; wait > 2*time.Minute {
			wait = 2 * time.Minute
		}
	}
}

// reloadSettles is how long after this engine wrote the bus's user list the bus is taken to have reloaded its accounts:
// the bus module's entrypoint looks at the list every 5 s and then signals the server, whose reload of the accounts
// answers no JetStream request for about three seconds (on the bus node, 2026-10-25).
var reloadSettles = 10 * time.Second

// agentNow is the agent running now, which the node's facts are built from.
type agentNow struct {
	mu sync.Mutex
	ag *agent.Agent
}

func (c *agentNow) set(a *agent.Agent) {
	c.mu.Lock()
	c.ag = a
	c.mu.Unlock()
}

func (c *agentNow) get() *agent.Agent {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.ag
}

// applyComposition applies one composition: a module's, or the node's own.
//
// placed is the modules a want still places here besides this one: none of their resources is removed by this apply,
// which is refused instead (issue 774).
//
// removed is the modules whose want says removed: in the node's own apply, a resource of any other module goes only when
// its module is among them (issue 774).
func applyComposition(ctx context.Context, opts options, c delivery.Composed, others []delivery.Composed, placed, removed map[string]bool,
	sched *apply.Scheduler, say link.Announce) agent.Result {
	name := scopeName(c)
	if c.Module == "" {
		// What another module's composition declares is that module's, never the node's own apply's (issue 1945).
		var dropped []string
		c, dropped = ownWithout(c, others)
		sayDroppedOnce(dropped)
	}
	d, err := declarationOf(c)
	if err != nil {
		return failedWith(err.Error())
	}
	if err := refuseEmptyWrites(d); err != nil {
		return failedWith(err.Error())
	}
	scope := apply.Scope{Module: name, Others: resourcesOf(others), Placed: placed}
	if name == agent.NodeModule {
		scope.RemovedByWant = removed
		if scope.RemovedByWant == nil {
			scope.RemovedByWant = map[string]bool{}
		}
	}
	got := applyModuleHeld(ctx, opts, d, &scope, sched, say)
	if got.Refused != "" {
		return failedWith(got.Refused)
	}
	results := apply.ModuleResults(d, got.Outcome, got.Err)
	if r, ok := results[name]; ok && r.Failed {
		return failedWith(r.Detail)
	}
	if got.Err != nil {
		return failedWith(got.Err.Error())
	}
	if got.Probe != nil {
		return failedWith(got.Probe.partial().Error())
	}
	return agent.Result{}
}

// removeComposition removes one module's own resources and nothing of another's.
func removeComposition(ctx context.Context, opts options, module string, others []delivery.Composed, placed map[string]bool,
	sched *apply.Scheduler, say link.Announce) agent.Result {
	scope := apply.Scope{Module: module, Others: resourcesOf(others), Placed: placed}
	got := applyModuleHeld(ctx, opts, &declaration.Declaration{Version: declaration.Version}, &scope, sched, say)
	switch {
	case got.Refused != "":
		return agent.Result{Failed: true, Detail: shorten(got.Refused)}
	case got.Err != nil:
		return agent.Result{Failed: true, Detail: shorten(got.Err.Error())}
	}
	return agent.Result{}
}

// busyModules are the modules whose last apply left a container running because it said busy.
func busyModules() []string {
	busyMu.Lock()
	defer busyMu.Unlock()
	var out []string
	for m, w := range busyBy {
		if w && m != "" {
			out = append(out, m)
		}
	}
	sort.Strings(out)
	return out
}

// machinesNow is the composer whose placements name the mesh's machines by address, for the health report's targets
// (writeHealthReport); nil on a node that does not compose its own, whose parts then name no target.
var machinesNow atomic.Pointer[selfcompose.Composer]

// composerNow is the composer of the run that composes the node's own now, nil while the node applies the controller's
// compositions or no run is going: what the verb compose asks a pass of (issue 774).
var composerNow atomic.Pointer[selfcompose.Composer]

// shadowNow is the shadow of the run now, which answers placement candidates and the verb compose-candidate; nil
// between runs.
var shadowNow atomic.Pointer[selfcompose.Shadow]

// keptRetired says the kept declaration is read no more: every module the seat's placement puts on this node has a have of
// its own (placedAllHave), so what runs is the agent's alone (ADR 0550 §2, design 47's L3 s9).
var keptRetired atomic.Bool

// errKeptRetired is keptDeclaration's answer once the read is retired.
var errKeptRetired = errors.New("the kept declaration is read no more: every module placed on this node has a have of its own")

// keptDeclaration is the declaration the node kept before the switch, verified with the mesh's key when it is given;
// once the read is retired, nothing is read.
func keptDeclaration(statePath string, signer ed25519.PublicKey) (*declaration.Declaration, error) {
	if keptRetired.Load() {
		return nil, errKeptRetired
	}
	var raw []byte
	if len(signer) > 0 {
		var err error
		if raw, err = store.LoadDeclared(store.DeclaredPath(statePath), signer); err != nil {
			return nil, err
		}
	} else {
		kept, err := store.ReadDeclared(store.DeclaredPath(statePath))
		if err != nil {
			return nil, err
		}
		raw = kept.Declaration
	}
	return declaration.Parse(raw)
}

// whatRuns is what runs on this node as the declaration of it (novox/hq ADR 0539 §9): the compositions the agent runs
// and, for every module the agent does not hold yet, what the kept declaration has for it, with the node's own parts
// (adoption, identities, groups, libraries) from the node's composition once the agent holds it and from the kept
// declaration until then. What the node judges its health by, and what the filling, the take act and the judgement of
// directories read.
func whatRuns(everything []delivery.Composed, holdsNode bool, kept func() (*declaration.Declaration, error)) (*declaration.Declaration, error) {
	d, err := declarationOf(everything...)
	if err != nil {
		return nil, err
	}
	if holdsNode {
		return d, nil
	}
	old, err := kept()
	if err != nil {
		if len(everything) > 0 {
			return d, nil
		}
		return nil, err
	}
	d.Adoption, d.Identities, d.Groups, d.Libraries = old.Adoption, old.Identities, old.Groups, old.Libraries
	d.Provisioner, d.Builds = old.Provisioner, old.Builds
	d.Filled = append(d.Filled, old.Filled...)
	return d, nil
}

// runsHere says what runs on this node; the running serve sets it to the agent's view, and a command run by hand reads
// the kept declaration.
var runsHere struct {
	mu sync.Mutex
	fn func(statePath string, signer ed25519.PublicKey) (*declaration.Declaration, error)
}

func setRunsHere(fn func(statePath string, signer ed25519.PublicKey) (*declaration.Declaration, error)) {
	runsHere.mu.Lock()
	runsHere.fn = fn
	runsHere.mu.Unlock()
}

// declaredNow is the declaration of what runs on this node now.
func declaredNow(statePath string, signer ed25519.PublicKey) (*declaration.Declaration, error) {
	runsHere.mu.Lock()
	fn := runsHere.fn
	runsHere.mu.Unlock()
	if fn != nil {
		return fn(statePath, signer)
	}
	return keptDeclaration(statePath, signer)
}

// runsFromAgent is what runs now as the agent in hand sees it.
func runsFromAgent(current *agentNow) func(string, ed25519.PublicKey) (*declaration.Declaration, error) {
	return func(statePath string, signer ed25519.PublicKey) (*declaration.Declaration, error) {
		kept := func() (*declaration.Declaration, error) { return keptDeclaration(statePath, signer) }
		ag := current.get()
		if ag == nil {
			return kept()
		}
		return whatRuns(ag.Everything(), ag.HoldsNode(), kept)
	}
}

// refuseEmptyWrites refuses a declaration that names a file the provisioner fills (owned at runtime, no content, no
// bytes, no digest, no sealed value) without saying it is filled: applied, the file would be written empty over what the
// provisioner wrote (issue 1945, live on 2026-10-25 at 01:22 on ace, g14 and shanks). The apply fails, naming the file,
// and nothing is written.
func refuseEmptyWrites(d *declaration.Declaration) error {
	filled := make(map[string]bool, len(d.Filled))
	for _, p := range d.Filled {
		filled[p] = true
	}
	var refused []string
	for _, r := range d.Resources {
		f, ok := r.(*declaration.File)
		if !ok || f.OwnerAtRuntime != declaration.RuntimeOwned || filled[f.Path] {
			continue
		}
		if f.Content == "" && f.Bytes == "" && f.Digest == "" && f.Sealed == "" && f.VaultSealed == nil {
			refused = append(refused, fmt.Sprintf("%s (%s)", f.ID, f.Path))
		}
	}
	if len(refused) > 0 {
		return fmt.Errorf("the composition names %s with no content and does not say the provisioner fills it, so it "+
			"would be written empty; nothing was applied", strings.Join(refused, ", "))
	}
	return nil
}

// recordedIn says the engine's applied state holds a record of every resource of a composition: what the agent believes
// of a have or the kept declaration only when it does, so a resource removed since is made again (issue 1945). A state
// that cannot be read says yes: the agent then believes what it was told, as before.
func recordedIn(statePath string, say func(string)) func(ctx context.Context, c delivery.Composed) bool {
	return func(_ context.Context, c delivery.Composed) bool {
		known, err := store.Load(statePath)
		if err != nil {
			if say != nil {
				say(fmt.Sprintf("the applied state cannot be read, so what runs is believed as told: %v", err))
			}
			return true
		}
		filled := make(map[string]bool, len(c.Filled))
		for _, p := range c.Filled {
			filled[p] = true
		}
		for _, raw := range c.Resources {
			r, err := declaration.ParseResource(raw)
			if err != nil {
				return true
			}
			// A file the provisioner client fills is left to it, and has no record until it writes it.
			if f, ok := r.(*declaration.File); ok && filled[f.Path] {
				continue
			}
			if _, ok := known.Find(r.Identity()); !ok {
				return false
			}
		}
		return true
	}
}

// declaresNothingToBackUp reads node-backup.backed-up's answer for a module: true when this machine backs nothing of it
// up (no runs, no paths), and whether the answer could be read at all.
func declaresNothingToBackUp(said json.RawMessage, module string) (nothing bool, read bool) {
	var list []struct {
		Module string   `json:"module"`
		Runs   int      `json:"runs"`
		Paths  []string `json:"paths"`
	}
	if err := json.Unmarshal(said, &list); err != nil {
		return false, false
	}
	for _, m := range list {
		if m.Module == module && (m.Runs > 0 || len(m.Paths) > 0) {
			return false, true
		}
	}
	return true, true
}

// applyBound is how long one apply or removal may take: long enough for the slowest image pull and backup step, short
// enough that a step that hangs cannot hold the node agent for ever.
const applyBound = 30 * time.Minute

// applyContext is the context an apply or a removal runs in: the link's values, never its end, and a bound of its own.
// The link's context ends when the bus goes, and the bus may go because of the apply itself (it replaces the bus's own
// container), which must then finish: a container removed and never made again took the mesh down (issue 1945, live on
// 2026-10-25 at 11:23). applyBound keeps a hung step from blocking the agent for ever.
func applyContext(ctx context.Context) (context.Context, context.CancelFunc) {
	return context.WithTimeout(context.WithoutCancel(ctx), applyBound)
}

// composerHooks are what the composer reads of the agent and the machine (selfcompose.Env Running, Unseal, Refused).
type composerHooks struct {
	running func() []delivery.Composed
	unseal  func(string) ([]byte, error)
	refused func(why string, held map[string]string)
	kept    func(kept map[string][]string)
	held    func(module string) (delivery.Seals, []string, error)
	// vaultAuthority and certify are selfcompose.Env's VaultAuthority and Certify.
	vaultAuthority func() string
	certify        func(ctx context.Context) error
}

// selfComposer is the node's own composer: it composes what the seat's placement places on the node, and nothing while
// the placement holds no key for it (issue 1126).
func selfComposer(session *link.Session, node string, hooks composerHooks, say func(string)) *selfcompose.Composer {
	env := composerEnv(session, node, say, hooks)
	// And it keeps each placed module's data items in its own state data, which the node's disk-load watches (novox/hq
	// ADR 0577 clause 1); the shadow composes for candidates and keeps none.
	env.Data = link.EngineData{Session: session}
	// And the packages its compositions declare in its own key of the state packages, which pacman reads (novox/hq ADR
	// 0565 §6).
	env.Packages = link.EnginePackages{Session: session}
	// And every composition it makes in its composition state composed-<node>, a record the provisioner, the assigner
	// and the view read and nothing applies (novox/hq ADR 0593 §6).
	env.ComposedState = link.EngineComposed{Session: session, Node: node}
	return selfcompose.New(env)
}

// composerEnv is what the node's composers read of the bus, the vault and the agent: its own composer and its answer
// to placement candidates (selfcompose.Shadow) alike.
func composerEnv(session *link.Session, node string, say func(string), hooks composerHooks) selfcompose.Env {
	return selfcompose.Env{Node: node,
		Watch: func(ctx context.Context, bucket, pattern string) (<-chan agent.Update, error) {
			if bucket == delivery.PlacementBucket || bucket == delivery.PlacementCandidatesBucket {
				// The seat's states are made when their holder starts: a node-engine that starts first waits for the
				// bus's word that the bucket was created, and the shadow and the composer say they wait.
				return bucketWatch{session, bucket}.WatchAwait(ctx, pattern, func(err error) {
					say("the seat's state " + bucket + " is not on the bus yet (" + err.Error() + "): it is watched when the bus says it was created")
				})
			}
			return bucketWatch{session, bucket}.Watch(ctx, pattern)
		},
		Get: func(ctx context.Context, bucket, key string) ([]byte, error) {
			h, err := session.KVHistory(ctx, bucket, key)
			if err != nil || len(h) == 0 || h[len(h)-1].Deleted {
				return nil, err
			}
			return h[len(h)-1].Value, nil
		},
		Mesh: frombus.Mesh{InternalSuffix: overlay.Suffix()},
		Say:  say,
		// Its modules' own secrets, sealed by the vault to this machine's key (ADR 0550 §5).
		Own: func(ctx context.Context, args map[string]any) (json.RawMessage, error) {
			return session.AskVault(ctx, "own", args)
		},
		// And the vault's word that a secret changed, on which it composes again (ADR 0547 §2).
		Hear: session.Hear,
		// And a moved module's templates archive, in the artifact store, as every archive is fetched (issue 1126).
		Fetch:   apply.Fetch,
		Running: hooks.running, Unseal: hooks.unseal, Refused: hooks.refused, Held: hooks.held, Kept: hooks.kept,
		VaultAuthority: hooks.vaultAuthority, Certify: hooks.certify}
}