Merge lead-47's L3 mesh-delivery head 710c4032 into the registry builder

The builder keeps both: the registry's versions, floor, tag asks and findings, and ADR 0618's engine-reads state,
its tool builder_engine_reads, the node-engine seat's manifest-reads verb and manifest-reads-changed event.

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

modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go

@@ -18,10 +18,15 @@ import (
 	stdio "git.novox.be/novox/mesh-sdk/go"
 	"git.novox.be/novox/mesh-sdk/go/delivery"
 
+	"git.novox.be/novox/mesh-delivery/sdk/enginereads"
+
 	"mesh-delivery-builder/internal/build"
 	"mesh-delivery-builder/internal/versions"
 )
 
+// EngineReadsTool lists, asks again and drops what each counted node's engine reads, as the builder keeps it.
+const EngineReadsTool = "builder_engine_reads"
+
 // queueAsker writes a queues value and waits for the builder's answer to it.
 type queueAsker func(key string, value []byte) (json.RawMessage, error)
 
@@ -373,7 +378,11 @@ func main() {
 		}
 		return e.Value, true, nil
 	}}
-	if err := stdio.Serve("", Tools(r, put, askQueue(stdio.State(build.QueuesState), 25*time.Second), buildAgain, stdio.Log(delivery.BuilderLog).Since, time.Now, vr, tags, tr, fr)); err != nil {
+	tools := Tools(r, put, askQueue(stdio.State(build.QueuesState), 25*time.Second), buildAgain, stdio.Log(delivery.BuilderLog).Since, time.Now, vr, tags, tr, fr)
+	// What each counted node's engine reads, as the builder keeps it (ADR 0618 §2, ADR 0624 §1): listed, one node asked
+	// again, or one node's key dropped.
+	tools = append(tools, enginereads.Tool(EngineReadsTool, "the builder", enginereads.KeptState(stdio.State("engine-reads")), true, time.Now))
+	if err := stdio.Serve("", tools); err != nil {
 		fmt.Fprintf(os.Stderr, "[mesh-delivery-builder-tools] %v\n", err)
 		os.Exit(1)
 	}

modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go

@@ -439,6 +439,10 @@ func TestBuilderMakesADoneKeyWithTheDigests(t *testing.T) { // 1607/AC7
 	if v.Toolchains["go"].Version != "go1.27.1" || v.Toolchains["typescript"].Version != "tsc5" {
 		t.Errorf("toolchains = %+v", v.Toolchains)
 	}
+	// ADR 0618 §1: the done key names the reference forms it is written in.
+	if len(v.Needs) != 1 || v.Needs[0] != "form:artifact-store" {
+		t.Errorf("needs = %v, want the first form", v.Needs)
+	}
 	// 1955/L7 (ADR 0571 §1): the build's start and end are in the module log build-log under the module's key, read as a
 	// reader with the log's read reads them, the end naming the manifest and what was made.
 	ended := b.logged(t, "netcheck")

modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go

@@ -159,7 +159,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	// revision (the gate's source state). The source's revisions are watched with their history (a direct get of one revision
 	// is not what a read grants): a revision heard after the candidate that names it builds that candidate then. Where it
 	// cannot be read the trunk's keys are built and the candidates wait (issue 1945).
-	errs := make(chan error, 8)
+	errs := make(chan error, 9)
 	trunks := &trunkSource{ctx: ctx, js: js, heads: &sourceHeads{}, moved: func(ctx context.Context) {
 		// A source revision heard is the forge's merge: it wakes a floor whose bridge_tags was silent (design 51 *The floor*).
 		(&floorWake{b: b, waits: floorWaits, logf: logger.Printf}).wake(ctx, versions.WakeForge, "", "")
@@ -204,6 +204,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	// placement is settled first, so the toolchains settle through the gate already knowing it.
 	gate := buildrun.NewToolchainGate([]string{buildrun.GoBuildModule, buildrun.NodeBuildModule}, tools.Put, logger.Printf)
 	watch.put = gate.Toolchain
+	// What each counted node's engine reads (ADR 0618 §2, ADR 0624 §1): asked through the node-engine seat's verb as the
+	// builder's own account, for every node the placement names, and kept in the state engine-reads once it is bound.
+	reads := newEngineReads(calls.CallOn, gate, logger.Printf)
+	gate.OnPlaced = reads.placed
+	b.Forms = reads.forms
 	placementErrs, err := buildrun.FollowPlacement(ctx, js, gate, b.NodesChanged, logger.Printf)
 	if err != nil {
 		return err
@@ -241,7 +246,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		// And the four registry seats' published, refused and removed (DLV-B4), which move a numbered version.
 		seats := seatHearer{n: numberer, b: b, logf: logger.Printf}
 		cc, err := process.HearEvents(cons, func(m jetstream.Msg) {
-			if floor.heard(ctx, m) || seats.heard(ctx, m) {
+			if reads.heard(m, logger.Printf) || floor.heard(ctx, m) || seats.heard(ctx, m) { // reads: the node-engine seat's manifest-reads-changed (ADR 0618 §4)
 				return
 			}
 			heardGone(ctx, b, m, logger.Printf)
@@ -341,6 +346,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		}
 		defer cc.Stop()
 	}
+	// The state engine-reads, new with ADR 0618 (issue 1790's kind): until it is on the bus each statement is kept in memory
+	// only, and builder_engine_reads answers nothing.
+	process.AwaitBucket(ctx, js, conn, engineReadsBucket, "what each node's engine reads is kept in memory only", logger.Printf).Ready(func(kv jetstream.KeyValue) {
+		reads.bind(ctx, kv, errs, logger.Printf)
+	})
 	logger.Printf("started on %s", cred.Node)
 	go func() { errs <- follow(ctx, w, each) }()
 	select {

modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reads.go

@@ -0,0 +1,105 @@
+package main
+
+import (
+	"context"
+	"slices"
+	"sync"
+
+	"github.com/nats-io/nats.go/jetstream"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+	"git.novox.be/novox/mesh-sdk/go/process"
+
+	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun"
+	"git.novox.be/novox/mesh-delivery/sdk/enginereads"
+)
+
+// EngineReadsState is the builder's state of what each counted node's engine states it reads (novox/hq ADR 0618 §2, ADR
+// 0624 §1): one key `<node>`, written from the node-engine seat's verb manifest-reads at the builder's start, on the
+// seat's event manifest-reads-changed for that node, and when a person asks again (builder_engine_reads); a node that
+// is away keeps its last statement, and its key is removed when the placement no longer names the node. A counted node
+// with no key reads as stating the baseline: its first forms. No node's record is read in its place.
+const EngineReadsState = "engine-reads"
+
+// engineReadsBucket is that state's bucket.
+const engineReadsBucket = delivery.BuilderModule + "_" + EngineReadsState
+
+// engineReads keeps the statements and the counted nodes for the builder's forms (build.Builder.Forms).
+type engineReads struct {
+	keeper  *enginereads.Keeper
+	gate    *buildrun.ToolchainGate
+	mu      sync.Mutex
+	counted []string
+}
+
+// newEngineReads asks through call (process.Conn.CallOn) and keeps each answer in the state once it is bound.
+func newEngineReads(call enginereads.Caller, gate *buildrun.ToolchainGate, logf func(string, ...any)) *engineReads {
+	return &engineReads{gate: gate, keeper: &enginereads.Keeper{Call: call, Logf: logf}}
+}
+
+// bind keeps the statements in kv and hears it: every node's last statement at the start, a person's ask again and drop
+// later. Bound once the state is on the bus (process.AwaitBucket).
+func (r *engineReads) bind(ctx context.Context, kv jetstream.KeyValue, errs chan<- error, logf func(string, ...any)) {
+	r.keeper.Put = func(node string, v []byte) error { _, err := kv.Put(ctx, node, v); return err }
+	r.keeper.Delete = func(node string) error { return kv.Delete(ctx, node) }
+	each := func(e jetstream.KeyValueEntry) {
+		r.keeper.Heard(e.Key(), e.Value(), e.Operation() != jetstream.KeyValuePut)
+	}
+	w, err := settle(ctx, kv, each)
+	if err != nil {
+		process.NotWatched(logf, engineReadsBucket, err, "builder_engine_reads is not answered until the builder starts again")
+		return
+	}
+	go func() { errs <- follow(ctx, w, each) }()
+	// Every counted node asked again, so what was asked before the state was bound is kept in it too.
+	r.mu.Lock()
+	nodes := slices.Clone(r.counted)
+	r.mu.Unlock()
+	for _, node := range nodes {
+		r.keeper.AskNode(node, "the state engine-reads is bound")
+	}
+}
+
+// placed takes the placement in: a node it names newly is asked, one it no longer names is forgotten, key and all. The
+// first call asks every counted node (the builder's start).
+func (r *engineReads) placed() {
+	nodes := r.gate.Nodes()
+	r.mu.Lock()
+	before, first := r.counted, r.counted == nil
+	r.counted = nodes
+	r.mu.Unlock()
+	for _, node := range nodes {
+		if first || !slices.Contains(before, node) {
+			r.keeper.AskNode(node, "counted: the placement names it")
+		}
+	}
+	for _, node := range before {
+		if !slices.Contains(nodes, node) {
+			r.keeper.Forget(node)
+		}
+	}
+}
+
+// heard takes an event of the builder's consumer on EVENTS: manifest-reads-changed asks its node again, and answers true.
+func (r *engineReads) heard(m jetstream.Msg, logf func(string, ...any)) bool {
+	node, ours, err := enginereads.NodeOf(m.Subject(), m.Data())
+	if !ours {
+		return false
+	}
+	if ackErr := m.Ack(); ackErr != nil {
+		logf("acknowledging %s: %v", m.Subject(), ackErr)
+	}
+	if err != nil {
+		logf("%s holds what is not %s, and %s is asked again all the same: %v", m.Subject(), enginereads.Event, node, err)
+	}
+	r.keeper.AskNode(node, enginereads.Event)
+	return true
+}
+
+// forms is build.Builder.Forms: the counted nodes' statements, and those of the nodes whose engine answered.
+func (r *engineReads) forms() (present, counted map[string][]string) {
+	r.mu.Lock()
+	nodes := slices.Clone(r.counted)
+	r.mu.Unlock()
+	return r.keeper.Present(nodes), r.keeper.Reads(nodes)
+}

modules/mesh-delivery-builder/cmd/mesh-delivery-builder/refused_test.go

@@ -3,8 +3,11 @@ package main
 import (
 	"bytes"
 	"context"
+	"encoding/json"
 	"errors"
 	"log"
+	"maps"
+	"slices"
 	"strings"
 	"testing"
 	"time"
@@ -16,6 +19,8 @@ import (
 	"github.com/nats-io/nats.go/jetstream"
 
 	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun"
+	"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
+	"git.novox.be/novox/mesh-delivery/sdk/enginereads"
 	"git.novox.be/novox/mesh-sdk/go/delivery"
 
 	"mesh-delivery-builder/internal/build"
@@ -54,6 +59,100 @@ func TestARefusedBucketIsSaidAndThePartGoesOn(t *testing.T) {
 	if src.opened {
 		t.Error("a refused source counts as opened, so it is never opened again")
 	}
+	t.Run("ADR 0624: the counted nodes' engine reads", theCountedNodesEngineReads)
+}
+
+// theCountedNodesEngineReads: the builder asks the node-engine seat's verb of every node the placement names; a node away
+// and never asked has no statement and reads as the baseline's first forms, with no node record read in its place; an
+// away node keeps its last statement; a node the placement no longer names leaves the count, its key removed; and
+// builder_engine_reads drops one node's key.
+func theCountedNodesEngineReads(t *testing.T) {
+	gate := buildrun.NewToolchainGate(nil, func(string, string, []byte, bool) bool { return false }, t.Logf)
+	place := func(nodes ...string) {
+		for _, node := range []string{"n1", "n2"} {
+			pl := delivery.Placement{Node: node}
+			if slices.Contains(nodes, node) {
+				pl.Modules = []delivery.PlacedModule{{Module: "web"}}
+			}
+			values, err := delivery.PlacementValues(pl)
+			if err != nil {
+				t.Fatal(err)
+			}
+			for key, raw := range values {
+				gate.Placement(key, raw, false, time.Now())
+			}
+			if !slices.Contains(nodes, node) {
+				key, _ := delivery.PlacementModuleKey(node, "web")
+				gate.Placement(key, nil, true, time.Now())
+			}
+		}
+	}
+	away := map[string]bool{"n2": true}
+	asked := map[string]int{}
+	r := newEngineReads(func(node, key string, _ any) (json.RawMessage, error) {
+		asked[node]++
+		if away[node] {
+			return nil, nats.ErrNoResponders
+		}
+		return json.Marshal(nodeengine.ManifestReadsAnswer{V: 1, Node: node, ManifestReads: []string{"form:new"}, StatedAt: time.Now().Format(time.RFC3339Nano)})
+	}, gate, t.Logf)
+	r.keeper.Sync = true
+	kept := map[string][]byte{}
+	r.keeper.Put = func(node string, v []byte) error { kept[node] = v; return nil }
+	r.keeper.Delete = func(node string) error { delete(kept, node); return nil }
+	place("n1", "n2")
+	r.placed()
+	present, counted := r.forms()
+	if !slices.Equal(present["n1"], []string{"form:new"}) || len(present) != 1 || len(counted) != 2 || counted["n2"] != nil {
+		t.Errorf("away and never asked: present %v, counted %v", present, counted)
+	}
+	if s := r.keeper.Statements()["n2"]; s.From != enginereads.FromNone || s.Answered || asked["n2"] != 1 {
+		t.Errorf("n2, away and never asked, is %+v (asked %d)", s, asked["n2"])
+	}
+	// n2 back, stating {word:module}, then away again: it keeps that statement.
+	away["n2"] = false
+	r.keeper.Heard("n2", mustStatement(t, enginereads.Statement{Node: "n2", From: enginereads.FromVerb, Answered: true,
+		AskAgain: &enginereads.Again{By: "n1.ada", Why: "back"}}), false)
+	away["n2"] = true
+	r.keeper.AskNode("n2", "its event")
+	if _, counted = r.forms(); !slices.Equal(counted["n2"], []string{"form:new"}) || r.keeper.Statements()["n2"].Answered {
+		t.Errorf("away again, n2 reads %v (%+v)", counted["n2"], r.keeper.Statements()["n2"])
+	}
+	// n2 retired: it leaves the placement, and its key goes.
+	place("n1")
+	r.placed()
+	if _, counted = r.forms(); len(counted) != 1 || kept["n2"] != nil {
+		t.Errorf("retired n2: counted %v, its key %s", counted, kept["n2"])
+	}
+	// builder_engine_reads drops one key.
+	st := &memReads{v: map[string][]byte{"n1": kept["n1"]}}
+	tool := enginereads.Tool("builder_engine_reads", "the builder", st, true, time.Now)
+	if _, err := tool.Run(map[string]any{"node": "n1", "drop": "true", "why": "a test"}); err != nil || st.v["n1"] != nil {
+		t.Errorf("drop: %v, the key %s", err, st.v["n1"])
+	}
+}
+
+func mustStatement(t *testing.T, s enginereads.Statement) []byte {
+	t.Helper()
+	raw, err := enginereads.Encode(s)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return raw
+}
+
+// memReads is the builder's state of engine reads in memory, as the tool reads it.
+type memReads struct{ v map[string][]byte }
+
+func (m *memReads) Keys() ([]string, error) { return slices.Collect(maps.Keys(m.v)), nil }
+func (m *memReads) Get(key string) ([]byte, uint64, bool, error) {
+	v, ok := m.v[key]
+	return v, 1, ok, nil
+}
+func (m *memReads) Put(key string, value []byte) (uint64, error) { m.v[key] = value; return 2, nil }
+func (m *memReads) Delete(key string) error                     { delete(m.v, key); return nil }
+func (m *memReads) Await(string, uint64, func([]byte) bool, time.Duration) ([]byte, error) {
+	return nil, errors.New("no part answers here")
 }
 
 // refusingBus is a bus that refuses: every open (refuseOpen), or every watch of a bucket it opens, with err.