The node agent runs in the node-engine: watches the wants and compositions on the link, applies one module's composition, asks the backup, writes the haves (R6-A A1, A5, A6)

cmd/mesh-host/applywhole_test.go

@@ -0,0 +1,41 @@
+package main
+
+import (
+	"context"
+
+	"git.novox.be/novox/mesh-sdk/go/declaration"
+	"github.com/novox/mesh-host/internal/apply"
+	"github.com/novox/mesh-host/internal/link"
+	"github.com/novox/mesh-host/internal/store"
+)
+
+// applyAndKeep applies a whole declaration the way the carried bundle is applied, and says what came of it as the facts
+// the node says: what the tests of the apply's steps read. The node-engine applies no whole declaration on its own now
+// (novox/hq ADR 0539 §9); this is the test's way to put a machine in a state.
+func applyAndKeep(ctx context.Context, opts options, raw []byte, signed *store.Declared, sched *apply.Scheduler,
+	say link.Announce) link.Report {
+	declared, err := declaration.Parse(raw)
+	if err != nil {
+		return link.Report{Refused: err.Error()}
+	}
+	got := applyModuleHeld(ctx, opts, declared, nil, sched, say)
+	if got.Refused != "" {
+		return link.Report{Refused: got.Refused}
+	}
+	held := map[string]bool{}
+	for _, h := range got.Updated.Held {
+		held[h.ID] = true
+	}
+	report := factsOf(ctx, opts, declared, got.Updated, got.Outcome, held)
+	if signed != nil {
+		if err := store.SaveDeclared(store.DeclaredPath(opts.state), *signed); err != nil {
+			report.Failed = map[string]string{"keeping the declaration": err.Error()}
+		}
+	}
+	if got.Err != nil {
+		report.Failed = map[string]string{"apply": got.Err.Error()}
+	}
+	report.Failed = withUnanswered(report.Failed, got.Probe)
+	report.Unapplied = unapplied(declared, got.Outcome, got.Err)
+	return report
+}

cmd/mesh-host/main.go

@@ -30,7 +30,9 @@ import (
 	"time"
 
 	"git.novox.be/novox/mesh-sdk/go/declaration"
+	"git.novox.be/novox/mesh-sdk/go/delivery"
 	"github.com/novox/mesh-host/internal/accounts"
+	"github.com/novox/mesh-host/internal/agent"
 	"github.com/novox/mesh-host/internal/apply"
 	"github.com/novox/mesh-host/internal/bundle"
 	"github.com/novox/mesh-host/internal/firewall"
@@ -1158,8 +1160,12 @@ func runLink(ctx context.Context, opts options) error {
 		Say:       say,
 		Timeout:   opts.timeout,
 	}
+	current := &agentNow{}
+	factsOfNow := func(ctx context.Context, ag *agent.Agent) {
+		sayFacts(ctx, opts, mine.Membership.Signer, queue, say, ag)
+	}
 	queue.OnLink = func(ctx context.Context) {
-		sayFacts(ctx, opts, mine.Membership.Signer, queue, say)
+		factsOfNow(ctx, current.get())
 		afterApply()
 	}
 	// And the move of a value sealed to this machine into the vault, asked by the controller (design 57's V3,
@@ -1186,6 +1192,14 @@ func runLink(ctx context.Context, opts options) error {
 		go judging.Probe(aside)
 		go judgeWhatRuns(aside, judging, queue, say)
 	}
+	// The node follows its wants: one module at a time from the composition its want names (novox/hq ADR 0539), and the
+	// busy containers are asked for again soon (ADR 0360 §11).
+	go nodeAgent(aside, opts, mine.Membership.Signer, queue, sched, say, afterApply, current, factsOfNow)
+	go busyPasses(aside, func() {
+		if ag := current.get(); ag != nil {
+			ag.Retry(busyModules()...)
+		}
+	}, BusyAskEvery)
 	// And the core builds this host placed are judged, whichever host placed them (to-be 45 §8).
 	go watchWhatThisHostPlaced(aside, mine.Node, queue, say)
 	// And mesh-cli is answered on this machine: the account from the kernel, the line asked of the controller
@@ -1563,6 +1577,23 @@ func orNothing(s, unless string) string {
 // busyWaiting is whether the last apply left a container running because it said busy (novox/hq ADR 0335).
 var busyWaiting atomic.Bool
 
+// busyBy is, per module, whether its last apply left a container running because it said busy; busyWaiting is any of them.
+var (
+	busyMu sync.Mutex
+	busyBy = map[string]bool{}
+)
+
+func noteBusy(module string, waiting bool) {
+	busyMu.Lock()
+	defer busyMu.Unlock()
+	busyBy[module] = waiting
+	any := false
+	for _, w := range busyBy {
+		any = any || w
+	}
+	busyWaiting.Store(any)
+}
+
 // BusyAskEvery is how often a container waiting because it said busy is asked again (novox/hq ADR 0335 §2):
 // each ask is a pass of its own, so a build that ends is followed by its replacement within this, not
 // within the reconcile's five minutes — and the bound of 15 minutes is met within this of its end.
@@ -1611,100 +1642,86 @@ func announceOr(say link.Announce) func(string) {
 // run by hand beside this service — the lock beside the state does the same (store.Lock).
 var applying sync.Mutex
 
-// applyAndKeep applies a declaration and, when it came from the mesh, keeps it so this node can
-// go on obeying it while disconnected. One at a time, whoever asks. Given unsigned, the bytes are
-// what this node kept, already verified against the mesh's key on load.
-func applyAndKeep(ctx context.Context, opts options, raw []byte, signed *store.Declared,
-	sched *apply.Scheduler, say link.Announce) link.Report {
-	applying.Lock()
-	defer applying.Unlock()
-	return applyAndKeepHeld(ctx, opts, raw, signed, sched, say)
+// appliedModule is how the apply of one module's declaration came out.
+type appliedModule struct {
+	// Refused is why nothing was applied, when nothing was.
+	Refused string
+	Outcome apply.Report
+	Updated store.State
+	// Err is the apply's own error, when it failed part way.
+	Err   error
+	Probe *unanswered
 }
 
-// applyAndKeepHeld is applyAndKeep for a caller already holding applying.
-func applyAndKeepHeld(ctx context.Context, opts options, raw []byte, signed *store.Declared,
-	sched *apply.Scheduler, say link.Announce) link.Report {
+// applyModuleHeld applies one module's declaration (novox/hq ADR 0539 §9): the resources of its composition, with the
+// compositions the node's other modules run held out of any removal. A nil scope is a whole declaration, as the carried
+// bundle is. One at a time, and the state is read and saved under the lock of the node's state.
+func applyModuleHeld(ctx context.Context, opts options, declared *declaration.Declaration, scope *apply.Scope,
+	sched *apply.Scheduler, say link.Announce) appliedModule {
+	applying.Lock()
+	defer applying.Unlock()
 	unlock, err := store.Lock(opts.state, nil)
 	if err != nil {
-		return link.Report{Refused: err.Error()}
+		return appliedModule{Refused: err.Error()}
 	}
 	defer unlock()
 
-	declared, err := declaration.Parse(raw)
-	if err != nil {
-		return link.Report{Refused: err.Error()}
-	}
 	known, err := store.Load(opts.state)
 	if err != nil {
-		return link.Report{Refused: err.Error()}
+		return appliedModule{Refused: err.Error()}
 	}
-	// A declaration from the mesh is the controller's word on the node's mode — the flip arrives
-	// as exactly that, the first converged declaration after adopted ones — and becomes the
-	// record. So does what this node kept: it is signed by the mesh and verified on load, where
-	// the state's note of the mode is this host's own. Where they disagree the note is wrong — a
-	// genesis re-run over a node the mesh converged since, a state saved when the kept declaration
-	// could not be — and it is repaired and said, not obeyed: refusing would hold this node off what
-	// the mesh said until a delivery that comes only when something changes (novox/hq issue 104).
-	if signed == nil && known.Mode != "" && known.Mode != apply.ModeOf(declared) {
-		if say != nil {
-			say(fmt.Sprintf("this node's record said %s; what the mesh last said, signed, says %s — the record is repaired",
-				known.Mode, apply.ModeOf(declared)))
-		}
+	node := scope == nil || scope.Module == store.NodeModule
+	// The node's own composition is the controller's word on the node's mode — the flip arrives as exactly that, the
+	// first converged composition after adopted ones — and becomes the record (novox/hq issue 104).
+	if node && known.Mode != "" && known.Mode != apply.ModeOf(declared) {
+		announceOr(say)(fmt.Sprintf("this node's record said %s; what the mesh last said, signed, says %s — the record is repaired",
+			known.Mode, apply.ModeOf(declared)))
 		known.Mode = apply.ModeOf(declared)
 	}
 
-	// **A file given by digest is fetched and verified before anything of the declaration is applied** (novox/hq ADR
-	// 0486 §1, issue 1695): one object missing or differing refuses the declaration whole. The signed bytes are what is
-	// kept; only what is applied has the content in place of the reference.
+	// **A file given by digest is fetched and verified before anything of the composition is applied** (novox/hq ADR
+	// 0486 §1, issue 1695): one object missing or differing refuses the module whole.
 	if err := apply.ResolveFileReferences(ctx, declared, link.FilesFrom(ctx)); err != nil {
-		return link.Report{Refused: "a file of this declaration could not be had by its digest: " + err.Error()}
+		return appliedModule{Refused: "a file of this composition could not be had by its digest: " + err.Error()}
 	}
 
 	built, err := system.For(builtFor)
 	if err != nil {
-		return link.Report{Refused: err.Error()}
+		return appliedModule{Refused: err.Error()}
 	}
 	if err := system.Check(built, declared); err != nil {
-		return link.Report{Refused: err.Error()}
+		return appliedModule{Refused: err.Error()}
 	}
 
-	// A package manager that does not answer fails only what needs it, and the report says so (novox/hq
-	// issue 233, probe.go).
+	// A package manager that does not answer fails only what needs it, and the report says so (novox/hq issue 233).
 	probe, err := probePackageManager(ctx, built)
 	if err != nil {
-		return link.Report{Refused: err.Error()}
+		return appliedModule{Refused: err.Error()}
 	}
 	if probe != nil {
 		announceOr(say)(probe.said())
 	}
 
-	// Declared, not carried. A declaration from the mesh removes only what the mesh previously
-	// declared — never what this machine raised for itself from its bundle (04-ISSUES/010).
-	// **What the apply says goes to the console, which is the journal when this runs as a service.**
-	//
-	// It was nil, and nil is silence. The one-shot path has always passed a real one, so every detail
-	// the apply produces — a file held, a container replaced, the found firewall retired — was visible
-	// when a person ran it by hand and discarded in the way the host actually runs. Measured: after a
-	// machine was converged and its found firewall was not retired, what the host decided was
-	// unrecoverable, because it had said it to nobody (novox/hq 04-ISSUES/143).
-	//
-	// `say` already reaches stdout, and the launcher's unit sends that to the journal, so this needs
-	// no new mechanism — only for the argument to be passed.
-	// Once more when the package manager did not answer and this apply wrote something (issue 233, probe.go).
+	// **What the apply says goes to the console, which is the journal when this runs as a service** (novox/hq
+	// 04-ISSUES/143). Once more when the package manager did not answer and this apply wrote something (issue 233).
 	outcome, updated, applyErr, probe, firstPass := applyMindingProbe(ctx, built, probe, known,
 		func(known store.State) (apply.Report, store.State, error) {
-			return apply.ApplyMindingWindows(ctx, built, declared, known, store.OriginDeclared,
+			if scope == nil {
+				return apply.ApplyMindingWindows(ctx, built, declared, known, store.OriginDeclared,
+					apply.ExecRunner, announceOr(say), sealOpener(opts.state), apply.KeepIn(filepath.Dir(opts.state)),
+					apply.WindowsIn(filepath.Dir(opts.state)))
+			}
+			return apply.ApplyModule(ctx, built, declared, *scope, known, store.OriginDeclared,
 				apply.ExecRunner, announceOr(say), sealOpener(opts.state), apply.KeepIn(filepath.Dir(opts.state)),
 				apply.WindowsIn(filepath.Dir(opts.state)))
 		}, announceOr(say))
 
-	// The mode the mesh said, recorded whichever way the apply went: the declaration is kept
-	// either way, and the node is held to it from the next reconcile (novox/hq ADR 0100).
-	updated.Mode = apply.ModeOf(declared)
+	if node {
+		updated.Mode = apply.ModeOf(declared)
+	}
 
-	// Root acted on what an agent account's judgement reads — an account, a package, an action, sudo's
-	// rules: the search for setuid programs runs again (novox/hq issue 361). A failed apply counts what it
-	// did before it stopped.
+	// Root acted on what an agent account's judgement reads — an account, a package, an action, sudo's rules: the search
+	// for setuid programs runs again (novox/hq issue 361). A failed apply counts what it did before it stopped.
 	moved := append(append([]apply.Outcome(nil), firstPass...), outcome.Outcomes...)
 	var partial *apply.Error
 	if errors.As(applyErr, &partial) {
@@ -1714,55 +1731,34 @@ func applyAndKeepHeld(ctx context.Context, opts options, raw []byte, signed *sto
 		setuidSearch.SearchAgain()
 	}
 
-	// Saved whichever way it went. Recording only on success would lose the footprint of a
-	// failed apply, and that footprint is on the machine either way.
+	// Saved whichever way it went: the footprint of a failed apply is on the machine either way.
 	if saveErr := store.Save(opts.state, updated); saveErr != nil {
-		return link.Report{Refused: "applied, and the node's state could not be saved: " +
-			saveErr.Error()}
+		return appliedModule{Refused: "applied, and the node's state could not be saved: " + saveErr.Error()}
 	}
 
-	// Re-establish the scheduled steps from the declaration just applied (novox/hq ADR 0053). Done
-	// from the parsed declaration, which is the source of truth (ADR 0018): a schedule newly declared
-	// is armed, one whose image, environment or cadence changed is re-armed, and one no longer
-	// declared is forgotten — and after a host restart the first apply rebuilds them all. A nil
-	// scheduler is the one-shot CLI path, which exits rather than staying up to fire anything.
+	// Re-establish the scheduled steps of this declaration (novox/hq ADR 0053): a schedule newly declared is armed, one
+	// whose image, environment or cadence changed is re-armed, one no longer declared is forgotten, and the other
+	// modules' are left as they are. A nil scheduler is a test.
 	held := map[string]bool{}
 	for _, h := range updated.Held {
 		held[h.ID] = true
 	}
 	if sched != nil {
-		sched.Sync(declared, held)
-	}
-
-	report := factsOf(ctx, opts, declared, updated, outcome, held)
-	// Kept whichever way it went, so a node that is disconnected next minute still knows what it
-	// was told. Saved after applying rather than before: what is kept is what this node acted on.
-	var before []string
-	if signed != nil {
-		// The modules the declaration kept until now held, so a have key written for one the new declaration no longer
-		// holds is written removed even by a node-engine that started since (issue 1369).
-		if kept, keptErr := store.ReadDeclared(store.DeclaredPath(opts.state)); keptErr == nil {
-			if old, parseErr := declaration.ParseTrusted(kept.Declaration); parseErr == nil {
-				for _, f := range moduleFacts(old, apply.Report{}, nil) {
-					before = append(before, f.Module)
-				}
-			}
-		}
-		if err := store.SaveDeclared(store.DeclaredPath(opts.state), *signed); err != nil {
-			report.Failed = map[string]string{"keeping the declaration": err.Error()}
+		if scope == nil {
+			sched.Sync(declared, held)
+		} else {
+			sched.SyncModule(declared, held, func(id string) bool {
+				m, ok := declaration.ModuleGuess(id)
+				return ok && m == scope.Module || (!ok && scope.Module == store.NodeModule)
+			})
 		}
 	}
-	if applyErr != nil {
-		report.Failed = map[string]string{"apply": applyErr.Error()}
+	name := ""
+	if scope != nil {
+		name = scope.Module
 	}
-	report.Failed = withUnanswered(report.Failed, probe)
-	// And how each module of the declaration came out, for the have keys written beside this report (novox/hq ADR 0429
-	// §4, issue 1132): not on the wire.
-	report.Modules = moduleFacts(declared, outcome, applyErr)
-	report.Before = before
-	// Which resources did not apply, for whether a reconcile's account is news (issue 233): not said on the wire.
-	report.Unapplied = unapplied(declared, outcome, applyErr)
-	return report
+	noteBusy(name, outcome.Waiting())
+	return appliedModule{Outcome: outcome, Updated: updated, Err: applyErr, Probe: probe}
 }
 
 // factsOf is what this node says of itself: its ports, its profile, how what runs here is, which links face outside,
@@ -1896,16 +1892,22 @@ func factsOf(ctx context.Context, opts options, declared *declaration.Declaratio
 	return report
 }
 
-// sayFacts says this node's facts on the link open now (novox/hq ADR 0539 §9): what the node is made of, how what runs here
-// is, and the haves of the modules the declaration it last applied holds. It applies nothing.
-func sayFacts(ctx context.Context, opts options, signer ed25519.PublicKey, session *link.Session, say link.Announce) {
+// sayFacts says this node's facts on the link open now (novox/hq ADR 0539 §9): what the node is made of, how what runs
+// here is, and its filters and ports. What runs is the compositions the node agent holds running (ag), or, before it
+// has run, the declaration the node kept before the switch. It applies nothing; the haves are the agent's.
+func sayFacts(ctx context.Context, opts options, signer ed25519.PublicKey, session *link.Session, say link.Announce,
+	ag *agent.Agent) {
 	known, err := store.Load(opts.state)
 	if err != nil {
 		say("cannot say this node's facts: " + err.Error())
 		return
 	}
 	declared := &declaration.Declaration{}
-	if raw, err := store.LoadDeclared(store.DeclaredPath(opts.state), signer); err == nil {
+	if running := runningOf(ag); len(running) > 0 {
+		if d, err := declarationOf(running...); err == nil {
+			declared = d
+		}
+	} else if raw, err := store.LoadDeclared(store.DeclaredPath(opts.state), signer); err == nil {
 		if d, err := declaration.Parse(raw); err == nil {
 			declared = d
 		}
@@ -1918,6 +1920,13 @@ func sayFacts(ctx context.Context, opts options, signer ed25519.PublicKey, sessi
 	session.SayFacts(ctx, report)
 }
 
+func runningOf(ag *agent.Agent) []delivery.Composed {
+	if ag == nil {
+		return nil
+	}
+	return ag.Running()
+}
+
 // waitForEnrolment returns this node's identity, waiting for one if it has none.
 //
 // It applies the carried bundle first, if there is one, because that is what a first node does

cmd/mesh-host/nodeagent.go

@@ -0,0 +1,353 @@
+package main
+
+import (
+	"context"
+	"crypto/ed25519"
+	"encoding/json"
+	"fmt"
+	"sort"
+	"sync"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/declaration"
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+	"github.com/novox/mesh-host/internal/agent"
+	"github.com/novox/mesh-host/internal/apply"
+	"github.com/novox/mesh-host/internal/link"
+	"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
+	}
+	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, nil
+}
+
+// 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) {
+	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}, true
+}
+
+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)
+		}
+		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)})
+}
+
+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.
+func shorten(s string) string {
+	const most = 300
+	if len(s) <= most {
+		return s
+	}
+	return s[:most-1] + "…"
+}
+
+// backupWithin is how long a backup is given from its ask to say it is done (ADR 0429 §14), and backupPollEvery how
+// soon the holder is asked again how it went while it works.
+var (
+	backupWithin    = 10 * time.Minute
+	backupPollEvery = 5 * time.Second
+)
+
+// backupNode asks the holder of node-backup on this node to back a module up now, and waits for it to say how it went:
+// the holder answers when it has started, and its `backed-up` says the module's last night.
+func backupNode(session *link.Session) func(ctx context.Context, module string) agent.Backup {
+	return func(ctx context.Context, module string) agent.Backup {
+		asked := time.Now()
+		if _, err := session.AskSeat(ctx, "node-backup", "now", map[string]any{"module": module}, 30*time.Second); err != nil {
+			return agent.Backup{Words: err.Error()}
+		}
+		deadline := asked.Add(backupWithin)
+		for {
+			select {
+			case <-ctx.Done():
+				return agent.Backup{Words: "the node agent was stopped while the backup ran"}
+			case <-time.After(backupPollEvery):
+			}
+			raw, err := session.AskSeat(ctx, "node-backup", "backed-up", map[string]any{"module": module}, 10*time.Second)
+			if err == nil {
+				var read []struct {
+					Module    string `json:"module"`
+					LastNight *struct {
+						OK    bool      `json:"ok"`
+						At    time.Time `json:"at"`
+						Error string    `json:"error"`
+					} `json:"lastNight"`
+				}
+				if json.Unmarshal(raw, &read) == nil {
+					for _, m := range read {
+						if m.Module == module && m.LastNight != nil && m.LastNight.At.After(asked) {
+							if m.LastNight.OK {
+								return agent.Backup{Done: true}
+							}
+							return agent.Backup{Words: "the backup failed: " + m.LastNight.Error}
+						}
+					}
+				}
+			}
+			if time.Now().After(deadline) {
+				return agent.Backup{Words: fmt.Sprintf("the backup did not say it was done within %s", backupWithin)}
+			}
+		}
+	}
+}
+
+// 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).
+func declarationOf(cs ...delivery.Composed) (*declaration.Declaration, error) {
+	d := &declaration.Declaration{Version: declaration.Version}
+	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
+}
+
+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
+}
+
+// keptByModule is the resources of the declaration the node kept before the switch, by the module they say; the node's
+// own are those that say none (ADR 0539 §9).
+func keptByModule(opts options, signer ed25519.PublicKey) func(module string) ([]json.RawMessage, bool) {
+	var once sync.Once
+	var by map[string][]json.RawMessage
+	return func(module string) ([]json.RawMessage, bool) {
+		once.Do(func() {
+			raw, err := store.LoadDeclared(store.DeclaredPath(opts.state), signer)
+			if err != nil {
+				return
+			}
+			var kept struct {
+				Resources []json.RawMessage `json:"resources"`
+			}
+			if json.Unmarshal(raw, &kept) != nil {
+				return
+			}
+			by = map[string][]json.RawMessage{}
+			for _, r := range kept.Resources {
+				var head struct {
+					Module string `json:"module"`
+				}
+				_ = json.Unmarshal(r, &head)
+				name := head.Module
+				if name == "" {
+					name = agent.NodeModule
+				}
+				by[name] = append(by[name], r)
+			}
+		})
+		r, ok := by[module]
+		return r, ok
+	}
+}
+
+// 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, 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
+		env := agent.Env{
+			Node:     node,
+			Wants:    bucketWatch{session, delivery.AssignerWantsBucket},
+			Composed: bucketWatch{session, composedBucket},
+			Haves:    sessionHaves{session},
+			Kept:     keptByModule(opts, signer),
+			Backup:   backupNode(session),
+			Say:      say,
+			Apply: func(ctx context.Context, c delivery.Composed, others []delivery.Composed) agent.Result {
+				res := applyComposition(session.WithLink(ctx), opts, c, others, sched, say)
+				afterApply()
+				sayFacts(ctx, ag)
+				return res
+			},
+			Remove: func(ctx context.Context, module string, others []delivery.Composed) agent.Result {
+				res := removeComposition(session.WithLink(ctx), opts, module, others, sched, say)
+				afterApply()
+				sayFacts(ctx, ag)
+				return res
+			},
+		}
+		ag = agent.New(env)
+		current.set(ag)
+		err = ag.Run(linked)
+		cancel()
+		current.set(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
+		}
+	}
+}
+
+// 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.
+func applyComposition(ctx context.Context, opts options, c delivery.Composed, others []delivery.Composed,
+	sched *apply.Scheduler, say link.Announce) agent.Result {
+	name := scopeName(c)
+	d, err := declarationOf(c)
+	if err != nil {
+		return agent.Result{Failed: true, Detail: shorten(err.Error())}
+	}
+	scope := apply.Scope{Module: name, Others: resourcesOf(others)}
+	got := applyModuleHeld(ctx, opts, d, &scope, sched, say)
+	if got.Refused != "" {
+		return agent.Result{Failed: true, Detail: shorten(got.Refused)}
+	}
+	results := apply.ModuleResults(d, got.Outcome, got.Err)
+	if r, ok := results[name]; ok && r.Failed {
+		return agent.Result{Failed: true, Detail: shorten(r.Detail)}
+	}
+	if got.Err != nil {
+		return agent.Result{Failed: true, Detail: shorten(got.Err.Error())}
+	}
+	if got.Probe != nil {
+		return agent.Result{Failed: true, Detail: shorten(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,
+	sched *apply.Scheduler, say link.Announce) agent.Result {
+	scope := apply.Scope{Module: module, Others: resourcesOf(others)}
+	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
+}

internal/agent/agent.go

@@ -107,6 +107,7 @@ type Agent struct {
 	// prior is a module the node ran before this agent started, so it has something to back up.
 	prior map[string]bool
 	said  map[string]string
+	retry chan []string
 }
 
 type wantNow struct {
@@ -137,7 +138,8 @@ func New(env Env) *Agent {
 		env.Say = func(string) {}
 	}
 	return &Agent{env: env, wants: map[string]wantNow{}, comps: map[string]compNow{}, running: map[string]delivery.Have{},
-		attempt: map[string]attempt{}, checked: map[string]bool{}, prior: map[string]bool{}, said: map[string]string{}}
+		attempt: map[string]attempt{}, checked: map[string]bool{}, prior: map[string]bool{}, said: map[string]string{},
+		retry: make(chan []string, 8)}
 }
 
 // Run follows the wants until ctx ends. The keys there when it starts are taken first, the node's own composition
@@ -157,6 +159,16 @@ func (a *Agent) Run(ctx context.Context) error {
 		select {
 		case <-ctx.Done():
 			return nil
+		case modules := <-a.retry:
+			// Applied again though the keys did not change: a container that waited because it said busy (ADR 0360
+			// §11) is replaced once it is idle.
+			a.mu.Lock()
+			for _, m := range modules {
+				delete(a.running, m)
+				delete(a.attempt, m)
+				dirty[m] = true
+			}
+			a.mu.Unlock()
 		case u, ok := <-wants:
 			if !ok {
 				return fmt.Errorf("the watch of the wants ended")
@@ -186,6 +198,39 @@ func (a *Agent) Run(ctx context.Context) error {
 	}
 }
 
+// Retry asks for the modules to be applied again from the keys they have, though nothing changed: what ADR 0360 §11
+// needs for a container that waits because it said busy. It never blocks; a retry asked while eight wait is dropped, and
+// asked again by its caller's next look.
+func (a *Agent) Retry(modules ...string) {
+	select {
+	case a.retry <- modules:
+	default:
+	}
+}
+
+// Running is the compositions the node runs now: each module's that the agent applied or found already running, and
+// the node's own. What the node judges its health by and says of itself is built from them.
+func (a *Agent) Running() []delivery.Composed {
+	a.mu.Lock()
+	defer a.mu.Unlock()
+	mods := make([]string, 0, len(a.running))
+	for m := range a.running {
+		mods = append(mods, m)
+	}
+	sort.Strings(mods)
+	var out []delivery.Composed
+	for _, m := range mods {
+		h := a.running[m]
+		if h.Phase == delivery.PhaseRemoved {
+			continue
+		}
+		if c, ok := a.compositionFor(m, h.FP); ok && c.Err == nil && c.Dig == h.Composed {
+			out = append(out, c.C)
+		}
+	}
+	return out
+}
+
 // takeWant takes a change of a want key and returns the module it concerns, or "" for a key that is not a want.
 func (a *Agent) takeWant(u Update) string {
 	a.mu.Lock()

internal/apply/schedule.go

@@ -127,6 +127,17 @@ func (s *Scheduler) RecordWindowsIn(w *Windows) {
 // work on the predecessor's data, under the predecessor's service, which is the one thing an
 // adopted node must not do. Nil on a converged node, where nothing is held.
 func (s *Scheduler) Sync(d *declaration.Declaration, held map[string]bool) {
+	s.sync(d, held, nil)
+}
+
+// SyncModule is Sync for one module's declaration (novox/hq ADR 0539 §9): the schedules of that module are armed,
+// re-armed or forgotten, and the schedules of every other module are left as they are. owns says whether a scheduled
+// step's id is the module's.
+func (s *Scheduler) SyncModule(d *declaration.Declaration, held map[string]bool, owns func(id string) bool) {
+	s.sync(d, held, owns)
+}
+
+func (s *Scheduler) sync(d *declaration.Declaration, held map[string]bool, owns func(id string) bool) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
@@ -177,7 +188,7 @@ func (s *Scheduler) Sync(d *declaration.Declaration, held map[string]bool) {
 	}
 
 	for id := range s.jobs {
-		if !seen[id] {
+		if !seen[id] && (owns == nil || owns(id)) {
 			delete(s.jobs, id)
 		}
 	}

internal/link/have.go

@@ -420,3 +420,35 @@ func (q *Session) NoteNode(ctx context.Context, composed string) {
 	q.mu.Unlock()
 	q.writeHaves(ctx, bus)
 }
+
+// ReadHave is the have this node-engine last wrote for a module, and whether the bucket holds one. A bucket that cannot
+// be read answers as none: the agent then compares with the declaration it kept, and applies only what differs.
+func (q *Session) ReadHave(ctx context.Context, module string) (delivery.Have, bool) {
+	q.init()
+	q.mu.Lock()
+	bus := q.bus
+	q.mu.Unlock()
+	hb, ok := bus.(HaveBus)
+	if !ok || bus == nil {
+		return delivery.Have{}, false
+	}
+	key, err := delivery.HaveKey(q.Membership.Node, module)
+	if err != nil {
+		return delivery.Have{}, false
+	}
+	read, cancel := context.WithTimeout(ctx, q.timeout())
+	defer cancel()
+	bucket, err := hb.HaveBind(read)
+	if err != nil {
+		return delivery.Have{}, false
+	}
+	value, _, err := bucket.Get(read, key)
+	if err != nil {
+		return delivery.Have{}, false
+	}
+	h, err := delivery.DecodeHave(value)
+	if err != nil {
+		return delivery.Have{}, false
+	}
+	return h, true
+}

internal/link/session.go

@@ -54,6 +54,9 @@ type Session struct {
 
 	mu  sync.Mutex
 	bus Bus // the link open now; nil while there is none
+	// gone is closed when the link open now closes, and changed whenever a link opens or closes (WhenLinked).
+	gone    chan struct{}
+	changed chan struct{}
 
 	// saying is held while a facts event is made and said, so they leave in the order they are made.
 	saying sync.Mutex
@@ -72,6 +75,7 @@ func (q *Session) init() {
 			q.Say = func(string) {}
 		}
 		q.epoch = time.Now().UnixMilli()
+		q.changed = make(chan struct{})
 	})
 }
 
@@ -117,6 +121,9 @@ func (q *Session) attach(ctx context.Context, bus Bus) {
 	q.init()
 	q.mu.Lock()
 	q.bus = bus
+	q.gone = make(chan struct{})
+	close(q.changed)
+	q.changed = make(chan struct{})
 	q.mu.Unlock()
 	q.FactsAsked(ctx)
 }
@@ -141,6 +148,9 @@ func (q *Session) detach(bus Bus) {
 	q.mu.Lock()
 	if q.bus == bus {
 		q.bus = nil
+		close(q.gone)
+		close(q.changed)
+		q.changed = make(chan struct{})
 	}
 	q.mu.Unlock()
 }
@@ -255,3 +265,46 @@ func ReadToolAnswer(reply []byte) (health.Answer, error) {
 	}
 	return a, nil
 }
+
+// SeatToolSubject is where a node seat's verb is asked of one machine's holder (design 33 §4): the seat's verb subject
+// with the machine as its last token.
+func SeatToolSubject(seat, verb, node string) string {
+	return "mesh.seat." + seat + ".tool." + verb + "." + node
+}
+
+// AskSeat asks the holder of a seat on this node a verb, over the link open now, and answers its result, or its refusal
+// as an error. The one call the node agent makes of another part: the backup before a data module moves (ADR 0429 §14).
+func (q *Session) AskSeat(ctx context.Context, seat, verb string, args any, within time.Duration) (json.RawMessage, error) {
+	q.init()
+	q.mu.Lock()
+	bus := q.bus
+	q.mu.Unlock()
+	if bus == nil {
+		return nil, fmt.Errorf("%w: this machine has no link to the bus open now", ErrNotAsked)
+	}
+	asker, ok := bus.(ToolBus)
+	if !ok {
+		return nil, fmt.Errorf("%w: the link to the bus open now cannot ask a seat", ErrNotAsked)
+	}
+	body, err := json.Marshal(args)
+	if err != nil {
+		return nil, err
+	}
+	asking, cancel := context.WithTimeout(ctx, within)
+	defer cancel()
+	reply, err := asker.Ask(asking, SeatToolSubject(seat, verb, q.Membership.Node), body)
+	if err != nil {
+		return nil, fmt.Errorf("%s.%s could not be asked: %w", seat, verb, err)
+	}
+	var env struct {
+		Result json.RawMessage `json:"result"`
+		Error  string          `json:"error"`
+	}
+	if err := json.Unmarshal(reply, &env); err != nil {
+		return nil, fmt.Errorf("%s.%s answered with something that is not a tool's answer", seat, verb)
+	}
+	if env.Error != "" {
+		return nil, errors.New(env.Error)
+	}
+	return env.Result, nil
+}

internal/link/watch.go

@@ -0,0 +1,129 @@
+package link
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/nats-io/nats.go"
+)
+
+// KVUpdate is one change of a key in a bucket the node-engine watches (novox/hq ADR 0539 §9). Caught marks the end of
+// the keys that were there when the watch began.
+type KVUpdate struct {
+	Key      string
+	Value    []byte
+	Revision uint64
+	Deleted  bool
+	Caught   bool
+}
+
+// KVBus is a link that can watch a bucket's keys: the node agent follows its wants and its compositions with it, and
+// never polls.
+type KVBus interface {
+	// KVWatch yields the keys of the bucket matching the pattern now, a Caught update, then every change, until ctx ends
+	// or the link does (the channel is then closed). A bucket the node's account may not read is an error.
+	KVWatch(ctx context.Context, bucket, pattern string) (<-chan KVUpdate, error)
+}
+
+// KVWatch is the watch on the link open now.
+func (b OverNATS) KVWatch(ctx context.Context, bucket, pattern string) (<-chan KVUpdate, error) {
+	js := b.JS
+	if b.Conn != nil {
+		bound, err := b.Conn.JetStream()
+		if err != nil {
+			return nil, err
+		}
+		js = bound
+	}
+	kv, err := js.KeyValue(bucket)
+	if err != nil {
+		return nil, fmt.Errorf("the bucket %s: %w", bucket, err)
+	}
+	w, err := kv.Watch(pattern, nats.Context(ctx))
+	if err != nil {
+		return nil, fmt.Errorf("watching %s in %s: %w", pattern, bucket, err)
+	}
+	out := make(chan KVUpdate, 64)
+	go func() {
+		defer close(out)
+		defer func() { _ = w.Stop() }()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case e, ok := <-w.Updates():
+				if !ok {
+					return
+				}
+				var u KVUpdate
+				if e == nil {
+					u.Caught = true
+				} else {
+					u = KVUpdate{Key: e.Key(), Value: e.Value(), Revision: e.Revision(),
+						Deleted: e.Operation() == nats.KeyValueDelete || e.Operation() == nats.KeyValuePurge}
+				}
+				select {
+				case out <- u:
+				case <-ctx.Done():
+					return
+				}
+			}
+		}
+	}()
+	return out, nil
+}
+
+func (l *natsLink) KVWatch(ctx context.Context, bucket, pattern string) (<-chan KVUpdate, error) {
+	return OverNATS{Conn: l.conn, JS: l.js}.KVWatch(ctx, bucket, pattern)
+}
+
+var (
+	_ KVBus = OverNATS{}
+	_ KVBus = (*natsLink)(nil)
+)
+
+// ErrNoLink is what a watch asks for with no link open.
+var ErrNoLink = errors.New("no link to the bus is open")
+
+// KVWatch watches a bucket on the link open now. It ends when ctx does or the link closes.
+func (q *Session) KVWatch(ctx context.Context, bucket, pattern string) (<-chan KVUpdate, error) {
+	q.init()
+	q.mu.Lock()
+	bus := q.bus
+	q.mu.Unlock()
+	w, ok := bus.(KVBus)
+	if bus == nil || !ok {
+		return nil, ErrNoLink
+	}
+	return w.KVWatch(ctx, bucket, pattern)
+}
+
+// WhenLinked waits for a link to be open and returns a context that ends when that link does (or ctx does): what the
+// node agent runs under, so its watches end with the link and begin again with the next.
+func (q *Session) WhenLinked(ctx context.Context) (context.Context, context.CancelFunc, error) {
+	q.init()
+	for {
+		q.mu.Lock()
+		if q.bus != nil {
+			gone := q.gone
+			q.mu.Unlock()
+			linked, cancel := context.WithCancel(ctx)
+			go func() {
+				select {
+				case <-gone:
+					cancel()
+				case <-linked.Done():
+				}
+			}()
+			return linked, cancel, nil
+		}
+		changed := q.changed
+		q.mu.Unlock()
+		select {
+		case <-ctx.Done():
+			return nil, nil, ctx.Err()
+		case <-changed:
+		}
+	}
+}