H-3: the node-engine says composed, running and capabilities-changed on the seat's subjects

Builds: design 47 G §260, §227, §226, §177, §309, §404

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ

cmd/mesh-host/nodeagent.go

@@ -656,8 +656,9 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus
 		// composition bucket) is never applied. While the seat's assignment holds no key for this node it composes
 		// nothing and keeps running what it runs; a assignment 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 assignments (issue 774)
-		composerNow.Store(composer) // what the verb compose asks a pass of, this run's alone
+		machinesNow.Store(composer)                // the machines a failing network part points at, by the assignments (issue 774)
+		composerNow.Store(composer)                // what the verb compose asks a pass of, this run's alone
+		session.SetManifestOf(composer.ManifestOf) // the event running names the manifest of the build that runs
 		env.Composed, env.SelfComposed, env.Policy, env.ComposedToday = composer, true, composer.Policy, composer.ComposedToday
 		env.WhyUnmade, env.Waiting = composer.WhyUnmade, composer.Waiting
 		env.HoldsData = composer.HoldsData // what runs from the kept declaration says Data as its manifest does
@@ -670,6 +671,7 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus
 		cancel()
 		current.set(nil)
 		composerNow.Store(nil)
+		session.SetManifestOf(nil)
 		shadowNow.Store(nil)
 		if ctx.Err() != nil {
 			return
@@ -1076,6 +1078,8 @@ func selfComposer(session *link.Session, node string, hooks composerHooks, say f
 	// 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}
+	// And each composition it writes there said as the seat's event composed, which the assigner waits on.
+	env.SayComposed = session.SayComposed
 	return selfcompose.New(env)
 }
 

internal/link/engineseat.go

@@ -204,11 +204,15 @@ func EngineEndpointMetadata(v catalogue.Verb, node string) map[string]string {
 // SayEngine says one of the seat's events about this node, on the events stream, said again through a transport fault
 // (sayOnce).
 func (l *natsLink) SayEngine(event string, body any) error {
+	return l.SayEngineOn(EngineEventSubject(event, l.node), body)
+}
+
+// SayEngineOn says one of the seat's events on its subject, which names the node and, for some, the module.
+func (l *natsLink) SayEngineOn(subject string, body any) error {
 	raw, err := json.Marshal(body)
 	if err != nil {
 		return err
 	}
-	subject := EngineEventSubject(event, l.node)
 	return sayOnce(func(id string) error {
 		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 		defer cancel()

internal/link/have.go

@@ -9,6 +9,7 @@ import (
 	nodeengine "git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
 	sdkkeys "git.novox.be/novox/mesh-sdk/go/keys"
 	"github.com/novox/mesh-host/internal/compositions"
+	"slices"
 	"sort"
 	"strings"
 	"sync"
@@ -67,6 +68,11 @@ type haveState struct {
 	userList map[string]nodeengine.UserListWritten
 	// userListNoKey is each module whose user list was said to have no have key to ride on, so it is said once.
 	userListNoKey map[string]bool
+	// runningSaid is, by module, the fingerprint and manifest its event running last said, so each is said once; and
+	// capsSaid the capabilities the event capabilities-changed last said (nil: none said yet this run).
+	runningSaid map[string]string
+	capsSaid    []string
+	capsKnown   bool
 	// unjudged is each module whose composition moved (an apply began, or the fact names another build) and that no look
 	// has judged since it stopped moving: its health is not yet judged, never the last look's of what ran before.
 	unjudged map[string]bool
@@ -360,6 +366,14 @@ func (q *Session) writeHaves(ctx context.Context, bus Bus) {
 		return
 	}
 	s := &q.haves
+	// What the pass says as the seat's events (running, capabilities-changed) is said once the haves' lock is let go: a
+	// publish never holds it.
+	var said []func()
+	defer func() {
+		for _, say := range said {
+			say()
+		}
+	}()
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	if s.modules == nil && s.node == nil && len(s.removed) == 0 && len(s.userList) == 0 {
@@ -426,6 +440,9 @@ func (q *Session) writeHaves(ctx context.Context, bus Bus) {
 			s.written = map[string]string{}
 		}
 		s.written[module] = f.Build
+		if ev, ok := s.runningToSay(module, h, q.manifestOf()); ok {
+			said = append(said, func() { q.sayRunning(ev) })
+		}
 	}
 	if !q.writeRemoved(write, bucket, node) {
 		return
@@ -450,6 +467,9 @@ func (q *Session) writeHaves(ctx context.Context, bus Bus) {
 				}) {
 					return
 				}
+				if ev, ok := s.capabilitiesToSay(node, facts.Record); ok {
+					said = append(said, func() { q.sayCapabilities(ev) })
+				}
 			}
 		}
 		if err != nil {
@@ -885,3 +905,75 @@ func nothingJudged(f ModuleFact, looked bool) string {
 	}
 	return nodeengineclient.HealthUnknown
 }
+
+// runningToSay is the event running for a module whose have was just written running and healthy at a fingerprint whose
+// manifest is known, when it has not been said for that fingerprint and manifest (design 47, *What each step says*): the
+// module, the fingerprint and the digest of the manifest of that build. A module not yet healthy, or one whose manifest
+// the node does not know, says nothing, and is said at the write that knows. Under s.mu.
+func (s *haveState) runningToSay(module string, h nodeengine.HaveByNodeModuleValue, manifest func(module, fp string) string) (nodeengine.Running, bool) {
+	if h.Phase != nodeengineclient.PhaseRunning || h.Health != nodeengineclient.HealthOK || h.Fp == "" || manifest == nil {
+		return nodeengine.Running{}, false
+	}
+	digest := manifest(module, h.Fp)
+	if digest == "" {
+		return nodeengine.Running{}, false
+	}
+	if s.runningSaid[module] == h.Fp+"\x00"+digest {
+		return nodeengine.Running{}, false
+	}
+	if s.runningSaid == nil {
+		s.runningSaid = map[string]string{}
+	}
+	s.runningSaid[module] = h.Fp + "\x00" + digest
+	return nodeengine.Running{V: sdkkeys.Version, Module: module, Fingerprint: h.Fp, Manifest: digest}, true
+}
+
+// capabilitiesToSay is the event capabilities-changed when the node's record says other capabilities than the event last
+// said (or none was said this run). Under s.mu.
+func (s *haveState) capabilitiesToSay(node string, rec *nodeengine.NodeRecord) (nodeengine.CapabilitiesChanged, bool) {
+	if rec == nil {
+		return nodeengine.CapabilitiesChanged{}, false
+	}
+	if s.capsKnown && slices.Equal(s.capsSaid, rec.Capabilities) {
+		return nodeengine.CapabilitiesChanged{}, false
+	}
+	s.capsKnown, s.capsSaid = true, slices.Clone(rec.Capabilities)
+	caps := slices.Clone(rec.Capabilities)
+	if caps == nil {
+		caps = []string{}
+	}
+	return nodeengine.CapabilitiesChanged{V: sdkkeys.Version, Node: node, Capabilities: caps}, true
+}
+
+// sayRunning says the event running, on its subject `running.<node>.<module>`.
+func (q *Session) sayRunning(ev nodeengine.Running) {
+	ev.Node, ev.At = q.Membership.Node, time.Now().UTC()
+	q.sayEngineOn(nodeengine.SubjectRunning(ev.Node, ev.Module), "running", ev)
+}
+
+// sayCapabilities says the event capabilities-changed, on its subject `capabilities-changed.<node>`.
+func (q *Session) sayCapabilities(ev nodeengine.CapabilitiesChanged) {
+	ev.At = time.Now().UTC()
+	q.sayEngineOn(nodeengine.SubjectCapabilitiesChanged(ev.Node), "capabilities-changed", ev)
+}
+
+// SetManifestOf gives the session what names the manifest of the build a module runs (the composer's picture of the
+// catalog half): its digest, or "" when the node does not know it.
+func (q *Session) SetManifestOf(f func(module, fp string) string) {
+	q.manifest.Lock()
+	q.manifestFn = f
+	q.manifest.Unlock()
+}
+
+func (q *Session) manifestOf() func(module, fp string) string {
+	q.manifest.Lock()
+	defer q.manifest.Unlock()
+	return q.manifestFn
+}
+
+// SayComposed says the event composed for a composition written at a fingerprint, on `composed.<node>`.
+func (q *Session) SayComposed(module, fingerprint, digest string) {
+	node := q.Membership.Node
+	q.sayEngineOn(nodeengine.SubjectComposed(node), "composed", nodeengine.Composed{V: sdkkeys.Version, Node: node,
+		Module: module, Fingerprint: fingerprint, Digest: digest, At: time.Now().UTC()})
+}

internal/link/have_test.go

@@ -44,6 +44,29 @@ type haveLink struct {
 	// sayFails fails that many of the node seat's says, as a publish past the bus's retry bound; said is every event said.
 	sayFails int
 	said     []string
+	// onSaid are the events said on a subject of their own (running, capabilities-changed), by subject, in order.
+	onSaid []string
+}
+
+func (l *haveLink) SayEngineOn(subject string, body any) error {
+	raw, _ := json.Marshal(body)
+	l.hmu.Lock()
+	l.onSaid = append(l.onSaid, subject+" "+string(raw))
+	l.hmu.Unlock()
+	return nil
+}
+
+// saidOn is the events said on a subject, as the bus would carry them.
+func (l *haveLink) saidOn(prefix string) []string {
+	l.hmu.Lock()
+	defer l.hmu.Unlock()
+	var out []string
+	for _, e := range l.onSaid {
+		if strings.HasPrefix(e, prefix) {
+			out = append(out, e)
+		}
+	}
+	return out
 }
 
 func (l *haveLink) SayEngine(event string, body any) error {
@@ -326,6 +349,10 @@ func TestTheNodeKeyCarriesTheNodesFacts(t *testing.T) {
 	if l.kv["n1.node"].rev != at {
 		t.Errorf("a report changing nothing wrote the node key again")
 	}
+	// Design 47: the node's capabilities, said when they change, on capabilities-changed.<node>.
+	if got := l.saidOn("mesh.seat.node-engine.event.capabilities-changed.n1 "); len(got) != 1 || !strings.Contains(got[0], `"capabilities":["apt","bus-state-calls","ledger-calls","systemd"]`) {
+		t.Errorf("capabilities-changed said %v, want once for the record's capabilities", got)
+	}
 	// A runner that does not answer states what it stated before; one answering without any states none (an older runner).
 	l.statedErr = fmt.Errorf("%w: nothing answered", ErrNoAnswer)
 	l.stated = nil
@@ -659,6 +686,20 @@ func TestAnAppliedModulesHaveNamesTheFingerprintAndTheDigestAndFollowsItsHealth(
 	if h.Fp != fp || h.Composed != dig || h.Phase != nodeengineclient.PhaseRunning || h.Health != nodeengineclient.HealthUnknown {
 		t.Fatalf("the have after the apply: %+v", h)
 	}
+	// Design 47, *What each step says*: running is said once, when the build is healthy and its manifest is known, on
+	// running.<node>.<module>, with the module, the fingerprint and the manifest's digest; never while it is unknown.
+	q.SetManifestOf(func(module, f string) string {
+		if module == "web" && f == fp {
+			return "sha256:" + strings.Repeat("c", 64)
+		}
+		return ""
+	})
+	q.SayHealth(context.Background(), *healthOf(map[string]string{"web": StateHealthy}))
+	q.SayHealth(context.Background(), *healthOf(map[string]string{"web": StateHealthy}))
+	if got := l.saidOn("mesh.seat.node-engine.event.running.n1.web "); len(got) != 1 ||
+		!strings.Contains(got[0], `"fingerprint":"`+fp+`"`) || !strings.Contains(got[0], `"manifest":"sha256:`+strings.Repeat("c", 64)+`"`) {
+		t.Fatalf("running said %v, want one for the healthy build", got)
+	}
 	// Starting: it waits. Past its start window the check fails: failed.
 	q.SayHealth(context.Background(), *healthOf(map[string]string{"web": StateStarting}))
 	if got := l.have(t, "n1.web"); got.Health != nodeengineclient.HealthWaits || got.Fp != fp || got.Composed != dig {

internal/link/session.go

@@ -103,6 +103,9 @@ type Session struct {
 
 	// haves is what this node-engine writes beside its facts (novox/hq ADR 0429 §4, issue 1132).
 	haves haveState
+	// manifestFn names the manifest of the build a module runs, for the event running (SetManifestOf), under manifest.
+	manifest   sync.Mutex
+	manifestFn func(module, fp string) string
 	// healthWritten is this node's health report as last written or read (WriteNodeHealth): the health loop alone
 	// writes it, one look at a time.
 	healthWritten *NodeHealth
@@ -463,6 +466,20 @@ func (q *Session) SayReads() {
 		Node: q.Membership.Node, ManifestReads: manifestReads(), At: time.Now().UTC().Format(time.RFC3339Nano), Why: why})
 }
 
+// sayEngineOn says one of the node seat's events on a subject of its own (running, composed: a module in it), over the link
+// open now; with none open it is not said.
+func (q *Session) sayEngineOn(subject, event string, body any) {
+	q.init()
+	q.mu.Lock()
+	bus := q.bus
+	q.mu.Unlock()
+	if sayer, ok := bus.(interface{ SayEngineOn(string, any) error }); ok {
+		if err := sayer.SayEngineOn(subject, body); err != nil {
+			q.Say("could not say the node seat's " + event + ": " + err.Error())
+		}
+	}
+}
+
 // SayEngine says one of the node seat's events over the link open now. With none open, or a publish that failed past the
 // bus's retry bound, it is kept and said at the next link's opening (SayReads; novox/hq ADR 0668 §4, issue 1966): an
 // event a reader waits on is never only a line in this engine's log.

internal/selfcompose/composedstate.go

@@ -137,6 +137,7 @@ func (c *Composer) writeComposed(ctx context.Context, at *published) (wrote, del
 		}
 		c.pubKept[key] = ComposedEntry{Value: values[key], Revision: rev}
 		wrote++
+		c.saidComposed(key, values[key])
 	}
 	for _, key := range c.pastKeeping() {
 		if err := c.env.ComposedState.Delete(ctx, key); err != nil {
@@ -319,3 +320,24 @@ func ranOf(running []engcomp.Composed) map[string]string {
 	}
 	return out
 }
+
+// saidComposed says the event composed for a key just written that holds a module's composition at a fingerprint and was
+// not refused: the node's own composition and a refusal say nothing.
+func (c *Composer) saidComposed(key string, value []byte) {
+	if c.env.SayComposed == nil {
+		return
+	}
+	_, module, fp, isNode, err := engcomp.ParseKey(key)
+	if err != nil || isNode {
+		return
+	}
+	comp, err := engcomp.Decode(value)
+	if err != nil || comp.Refused != "" {
+		return
+	}
+	digest, err := engcomp.DigestOf(comp)
+	if err != nil {
+		return
+	}
+	c.env.SayComposed(module, fp, digest)
+}

internal/selfcompose/composedstate_test.go

@@ -84,6 +84,14 @@ func TestTheNodePublishesItsCompositions(t *testing.T) {
 	state := &fakeComposed{kv: map[string]ComposedEntry{}}
 	env := b.env()
 	env.ComposedState = state
+	// Design 47, *What each step says*: each module composition written says composed, with its fingerprint and digest.
+	var saidMu sync.Mutex
+	var saidComposed []string
+	env.SayComposed = func(module, fingerprint, digest string) {
+		saidMu.Lock()
+		defer saidMu.Unlock()
+		saidComposed = append(saidComposed, module+" "+fingerprint+" "+digest)
+	}
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 	c := New(env)
@@ -136,6 +144,14 @@ func TestTheNodePublishesItsCompositions(t *testing.T) {
 			t.Errorf("the published composition of lemurs: %+v, %v", comp, err)
 		}
 	}
+	if comp, err := engcomp.Decode(state.kv[lemursKey].Value); err == nil {
+		dig, _ := engcomp.DigestOf(comp)
+		saidMu.Lock()
+		if len(saidComposed) != 1 || saidComposed[0] != "lemurs "+fp+" "+dig || dig == "" {
+			t.Errorf("composed said %v, want once for lemurs at %s with its digest %q (the node's own says none)", saidComposed, fp, dig)
+		}
+		saidMu.Unlock()
+	}
 	// A pass that changes nothing writes nothing.
 	setFor(t, &askedWithin, 0)
 	if _, err := c.ComposeNow(ctx); err != nil {

internal/selfcompose/selfcompose.go

@@ -118,11 +118,26 @@ type Env struct {
 	// SeatList is the node's reader of the seat list (novox/hq ADR 0636 §1): until the list was read, the composer hands
 	// nothing new of a module whose seats are not all bootstrapped, and keeps what runs of it. Nil gates nothing.
 	SeatList SeatGate
+	// SayComposed says the seat's event composed for a composition of a module written into the composition state at a
+	// fingerprint (design 47, *What each step says*): the module, the fingerprint and the digest of the composition. Nil says
+	// nothing.
+	SayComposed func(module, fingerprint, digest string)
 	// SeatListChanged says the reader's held set changed: the node composes again, so what waited for the list is
 	// handed at the read and not on a timer. Nil hears nothing.
 	SeatListChanged <-chan struct{}
 }
 
+// ManifestOf is the digest of the manifest of the build a module runs at fp, from the composer's picture of the catalog half;
+// "" when the picture holds no such build (the event running names it, so one not known is not said).
+func (c *Composer) ManifestOf(module, fp string) string {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if m, ok := c.modules[module]; ok && m.Fp2 == fp {
+		return m.Manifest
+	}
+	return ""
+}
+
 // SeatGate is what the composer asks of the reader of the seat list: whether it may compose a module that claims,
 // invokes or provides these seats (seatlist.Reader.MayAct).
 type SeatGate interface {