node-tools pins mesh-sdk at 8311ecb

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

node-tools/go.mod

@@ -8,7 +8,7 @@ require (
 )
 
 require (
-	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030100935-c2ec8b6b2fde
+	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101023317-8311ecb84911
 	github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op // indirect
 	github.com/google/go-tpm v0.9.8 // indirect
 	github.com/klauspost/compress v1.20.0 // indirect

node-tools/go.sum

@@ -1,11 +1,5 @@
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468 h1:UcMee+IV2ii5E4dRZdQUaSaeh9IedTFYumuQbBiB0k0=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101023317-8311ecb84911 h1:+82ygf2hiY5Mhwfmc1xfCe9e/KY/5ObBuE2LecM37qI=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101023317-8311ecb84911/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030095131-16bc0dcf3e9d h1:jFOz6wUfZvIayd4DctwlaFKKZHQNME6NLs2OS1rNv30=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030095131-16bc0dcf3e9d/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030095816-530d8d3007dc h1:Wdjl2gomooHswJ5qpkwZPq/ureZgXVw6FMCsgzPJhPo=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030095816-530d8d3007dc/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030100935-c2ec8b6b2fde h1:eKXSllHCS31GgMazfvZ7BeiLoXIUibnUmzxWVkj+Vcw=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030100935-c2ec8b6b2fde/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
 github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
 github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
 github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=

node-tools/internal/bus/bus.go

@@ -10,6 +10,7 @@
 package bus
 
 import (
+	"context"
 	"crypto/sha256"
 	"crypto/tls"
 	"crypto/x509"
@@ -17,6 +18,8 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"git.novox.be/novox/mesh-sdk/go/buscompose"
+	"git.novox.be/novox/mesh-sdk/go/membership"
 	"log"
 	"regexp"
 	"strings"
@@ -180,9 +183,12 @@ type Conn struct {
 	cred   Credential
 	mu     sync.Mutex
 	issued map[string]*Membership // module → membership; present with nil = followed, none issued
-	onNew  []*func(Membership)
+	// fromGrants is every module whose membership the grants part's state memberships has given (novox/hq ADR 0571
-	subs   []*nats.Subscription
+	// clause 7): its assignments stream copy is then a fallback no longer applied.
-	Logf   func(format string, args ...any)
+	fromGrants map[string]bool
+	onNew      []*func(Membership)
+	subs       []*nats.Subscription
+	Logf       func(format string, args ...any)
 	// answering is every subject this runtime answers on, so a subscription the bus refused can be
 	// asked for again (novox/hq issue 222).
 	answering map[string]*answered
@@ -326,7 +332,7 @@ func Connect(cred Credential) (*Conn, error) {
 		return nil, err
 	}
 	c = &Conn{nc: nc, js: js, self: cred.Module, node: cred.Node, cred: cred,
-		issued: map[string]*Membership{}, Logf: log.Printf, answering: map[string]*answered{}}
+		issued: map[string]*Membership{}, fromGrants: map[string]bool{}, Logf: log.Printf, answering: map[string]*answered{}}
 	c.Follow(cred.Module)
 	return c, nil
 }
@@ -346,7 +352,10 @@ func MembershipSubject(node, module string) string {
 }
 
 // Follow reads a module's membership on this machine once and follows it live, so its tools are
-// served where the mesh issued them (ADR 0175).
+// served where the mesh issued them (ADR 0175). It is read from the grants part's state memberships, its key
+// `<node>.<module>` (novox/hq ADR 0571 clause 7, mesh-sdk buscompose.GrantsMembershipsBucket); the controller's
+// assignments stream is read beside it as the fallback until the lease write, and applied only while the state has given
+// the module no membership.
 func (c *Conn) Follow(module string) {
 	c.mu.Lock()
 	if c.node == "" {
@@ -359,16 +368,21 @@ func (c *Conn) Follow(module string) {
 	}
 	c.issued[module] = nil
 	c.mu.Unlock()
+	c.followGrants(module)
 	subject := MembershipSubject(c.node, module)
 	// The subject-addressed direct get: the one address the mesh grants this account on the
 	// stream's API.
-	if got, err := c.nc.Request("$JS.API.DIRECT.GET."+assignmentsStream+"."+subject, nil, 5*time.Second); err == nil {
+	if c.Membership(module) == nil {
-		if got.Header.Get("Status") == "" && len(got.Data) > 0 {
+		if got, err := c.nc.Request("$JS.API.DIRECT.GET."+assignmentsStream+"."+subject, nil, 5*time.Second); err == nil {
-			var m Membership
+			if got.Header.Get("Status") == "" && len(got.Data) > 0 {
-			if json.Unmarshal(got.Data, &m) == nil {
+				var m Membership
-				c.mu.Lock()
+				if json.Unmarshal(got.Data, &m) == nil {
-				c.issued[module] = &m
+					c.mu.Lock()
-				c.mu.Unlock()
+					if !c.fromGrants[module] {
+						c.issued[module] = &m
+					}
+					c.mu.Unlock()
+				}
 			}
 		}
 	}
@@ -381,24 +395,124 @@ func (c *Conn) Follow(module string) {
 			c.Logf("[mesh-tools] a membership arrived that is not one: %v", err)
 			return
 		}
-		c.mu.Lock()
+		c.issue(module, m, false)
-		c.issued[module] = &m
-		handlers := make([]func(Membership), 0, len(c.onNew))
-		for _, h := range c.onNew {
-			handlers = append(handlers, *h)
-		}
-		c.mu.Unlock()
-		c.Logf("[mesh-tools] %s on %s was issued a new membership; re-serving on it", module, c.node)
-		for _, h := range handlers {
-			h(m)
-		}
-		_ = c.nc.Flush()
 	})
 	if err == nil {
 		c.track(sub)
 	}
 }
 
+// GrantsPatience bounds the first read of a membership from the grants part's state.
+var GrantsPatience = 2 * time.Second
+
+// issue applies a membership that arrived for module, live, and re-serves on it. One from the assignments stream is
+// passed over once the grants part's state has given the module one.
+func (c *Conn) issue(module string, m Membership, fromGrants bool) {
+	c.mu.Lock()
+	if !fromGrants && c.fromGrants[module] {
+		c.mu.Unlock()
+		return
+	}
+	if fromGrants {
+		c.fromGrants[module] = true
+	}
+	c.issued[module] = &m
+	handlers := make([]func(Membership), 0, len(c.onNew))
+	for _, h := range c.onNew {
+		handlers = append(handlers, *h)
+	}
+	c.mu.Unlock()
+	c.Logf("[mesh-tools] %s on %s was issued a new membership; re-serving on it", module, c.node)
+	for _, h := range handlers {
+		h(m)
+	}
+	_ = c.nc.Flush()
+}
+
+// followGrants reads module's key of the grants part's state memberships and watches it: the value there now is applied
+// before Follow returns, each change after it live. A state that is not there yet, or not readable, is said once, and
+// the assignments stream answers alone.
+func (c *Conn) followGrants(module string) {
+	key := buscompose.MembershipKey(c.node, module)
+	// First the key's direct get, which the account is granted alone: a bus that refuses it, or a state not there yet,
+	// answers at once, where binding the bucket would wait out its bound on a refusal.
+	bucket := buscompose.GrantsMembershipsBucket
+	probe, err := c.nc.Request("$JS.API.DIRECT.GET.KV_"+bucket+".$KV."+bucket+"."+key, nil, GrantsPatience)
+	if err != nil || probe.Header.Get("Status") == "503" {
+		why := "it did not answer"
+		if err != nil {
+			why = err.Error()
+		}
+		c.Logf("[mesh-tools] the grants part's memberships (%s) cannot be read (%s); %s's membership is read from the "+
+			"assignments stream alone", bucket, why, module)
+		return
+	}
+	js, err := jetstream.New(c.nc)
+	if err != nil {
+		return
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
+	defer cancel()
+	kv, err := js.KeyValue(ctx, bucket)
+	if err != nil {
+		c.Logf("[mesh-tools] the grants part's memberships (%s) cannot be read (%v); %s's membership is read from the "+
+			"assignments stream alone", buscompose.GrantsMembershipsBucket, err, module)
+		return
+	}
+	w, err := kv.Watch(context.Background(), key)
+	if err != nil {
+		c.Logf("[mesh-tools] %s's key %s of the grants part's memberships cannot be watched (%v); read from the "+
+			"assignments stream alone", module, key, err)
+		return
+	}
+	apply := func(e jetstream.KeyValueEntry) {
+		if e.Operation() != jetstream.KeyValuePut {
+			return // a membership withdrawn is the stream's, or the module's unassignment's, to say
+		}
+		// Read through mesh-sdk's one codec of this state, the grants part's writer's (issue 774): it takes the bare
+		// membership and a wrapper the part wrote before of one published or unchanged, and refuses a held-back one's
+		// wrapper or a value naming no node or module, which then leaves the assignments stream's copy in force instead.
+		// A refused value arriving after a good one keeps the last good membership: nothing here clears it.
+		var m Membership
+		body, err := membership.BareGrantsMembership(e.Value())
+		if err == nil {
+			err = json.Unmarshal(body, &m)
+		}
+		if err != nil {
+			c.Logf("[mesh-tools] the membership at %s is not one: %v", key, err)
+			return
+		}
+		c.issue(module, m, true)
+	}
+	timer := time.NewTimer(StateTimeout)
+	defer timer.Stop()
+	for current := true; current; {
+		select {
+		case e, ok := <-w.Updates():
+			if !ok {
+				return
+			}
+			if e == nil {
+				current = false
+				continue
+			}
+			apply(e)
+		case <-timer.C:
+			_ = w.Stop()
+			c.Logf("[mesh-tools] %s's key %s of the grants part's memberships did not answer in %s; read from the "+
+				"assignments stream alone", module, key, StateTimeout)
+			return
+		}
+	}
+	go func() {
+		for e := range w.Updates() {
+			if e != nil {
+				apply(e)
+			}
+		}
+	}()
+}
+
 // Membership is what the mesh issued a module here, or nil when nothing has been issued.
 func (c *Conn) Membership(module string) *Membership {
 	c.mu.Lock()

node-tools/internal/bus/log.go

@@ -1,6 +1,7 @@
 package bus
 
 import (
+	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -8,6 +9,7 @@ import (
 	"sort"
 	"strconv"
 	"strings"
+	"sync/atomic"
 	"time"
 
 	"github.com/nats-io/nats.go"
@@ -94,6 +96,10 @@ func logFilter(l LogIssued, pattern string) (string, error) {
 	}
 	parts := strings.Split(pattern, ".")
 	for i, p := range parts {
+		if p == "**" && i == len(parts)-1 {
+			// `**` for the rest, as a state's key pattern and mesh-sdk's log reader say it.
+			parts[i], p = ">", ">"
+		}
 		switch {
 		case p == ">" && i == len(parts)-1, p == "*", logToken.MatchString(p):
 		default:
@@ -104,7 +110,7 @@ func logFilter(l LogIssued, pattern string) (string, error) {
 	if len(pattern) > LogKeyMax {
 		return "", fmt.Errorf("%q is longer than a key can be (%d bytes)", pattern, LogKeyMax)
 	}
-	return l.Subject + "." + pattern, nil
+	return l.Subject + "." + strings.Join(parts, "."), nil
 }
 
 // issuedLog is the log a module may reach by a name, or the reason it may not.
@@ -126,8 +132,9 @@ func (c *Conn) issuedLog(module, name string) (LogIssued, error) {
 	if len(names) > 0 {
 		issued = strings.Join(names, ", ")
 	}
-	return LogIssued{}, fmt.Errorf("%s keeps no log called %q: a module declares the logs it keeps under `logs`, "+
+	return LogIssued{}, fmt.Errorf("%s keeps no log called %q: a module declares the logs it keeps under `logs`, and "+
-		"and a log is its owner's alone; it was issued: %s (novox/hq ADR 0297)", module, name, issued)
+		"another module's it reads under `reads` as <module>.<log>; it was issued: %s (novox/hq ADR 0297, ADR 0571)",
+		module, name, issued)
 }
 
 // notOnTheBus says a log is issued and its stream is not there yet.
@@ -200,9 +207,10 @@ func (c *Conn) LogAppend(module, name, key string, value json.RawMessage, expect
 	return ack.Seq, nil
 }
 
-// LogRead is the entries from a sequence on, of the keys matching a pattern, at most limit of them and
+// LogRead is the entries from a sequence on, or from a time on (since, when not zero), of the keys matching a pattern, at
-// about 4 MiB, by one batch direct get.
+// most limit of them and about 4 MiB, by one batch direct get. A log another module keeps is read as its own is, by the
-func (c *Conn) LogRead(module, name string, from uint64, keys string, limit int) (*LogPage, error) {
+// name its reads declare, `<module>.<log>` (novox/hq ADR 0571).
+func (c *Conn) LogRead(module, name string, from uint64, since time.Time, keys string, limit int) (*LogPage, error) {
 	l, err := c.issuedLog(module, name)
 	if err != nil {
 		return nil, err
@@ -218,20 +226,24 @@ func (c *Conn) LogRead(module, name string, from uint64, keys string, limit int)
 	if err != nil {
 		return nil, err
 	}
+	if from != 0 && !since.IsZero() {
+		return nil, fmt.Errorf("a read of %s starts from a sequence or from a time, not both", name)
+	}
 	if from == 0 {
 		from = 1
 	}
-	msgs, next, err := c.directBatch(l.Stream, filter, from, limit, name, func() error { return notOnTheBus(l) })
+	req := map[string]any{"seq": from, "next_by_subj": filter, "batch": limit, "max_bytes": LogPageBytes}
+	if !since.IsZero() {
+		delete(req, "seq")
+		req["start_time"] = since.UTC()
+	}
+	entries, more, err := c.logBatch(l, req)
 	if err != nil {
 		return nil, err
 	}
-	page := &LogPage{Entries: []LogEntry{}, Next: next}
+	page := &LogPage{Entries: entries}
-	for _, msg := range msgs {
+	if more && len(entries) > 0 {
-		e, err := logEntryOf(l, msg)
+		page.Next = entries[len(entries)-1].Sequence + 1
-		if err != nil {
-			return nil, err
-		}
-		page.Entries = append(page.Entries, *e)
 	}
 	info, err := c.logInfo(l)
 	if err != nil {
@@ -241,56 +253,148 @@ func (c *Conn) LogRead(module, name string, from uint64, keys string, limit int)
 	return page, nil
 }
 
-// directBatch is the messages of a stream from a sequence on, on the subjects matching filter, at most limit and about
+// LogNewest is the newest entry of each key matching a pattern, in the log's order, by one batch direct get (mesh-sdk
-// LogPageBytes, by one batch direct get (NATS 2.11), and the sequence to read on from, 0 when nothing follows. name is
+// process.LogReader.Newest).
-// what is read, for its errors; missing is the error of a stream that is not on the bus.
+func (c *Conn) LogNewest(module, name, keys string) ([]LogEntry, error) {
-func (c *Conn) directBatch(stream, filter string, from uint64, limit int, name string, missing func() error) ([]*nats.Msg, uint64, error) {
+	l, err := c.issuedLog(module, name)
-	if from == 0 {
+	if err != nil {
-		from = 1
+		return nil, err
 	}
-	req, _ := json.Marshal(map[string]any{"seq": from, "next_by_subj": filter, "batch": limit, "max_bytes": LogPageBytes})
+	filter, err := logFilter(l, keys)
+	if err != nil {
+		return nil, err
+	}
+	entries, more, err := c.logBatch(l, map[string]any{"multi_last": []string{filter}, "batch": LogReadMax})
+	if err != nil {
+		return nil, err
+	}
+	if more {
+		return nil, fmt.Errorf("%s holds more than %d keys matching %q: name fewer", name, LogReadMax, keys)
+	}
+	return entries, nil
+}
+
+// logBatch is one batch direct get of a log: its entries, and whether more are pending after them.
+func (c *Conn) logBatch(l LogIssued, request map[string]any) ([]LogEntry, bool, error) {
+	req, _ := json.Marshal(request)
 	inbox := c.nc.NewInbox()
 	sub, err := c.nc.SubscribeSync(inbox)
 	if err != nil {
-		return nil, 0, err
+		return nil, false, err
 	}
 	defer func() { _ = sub.Unsubscribe() }()
-	if err := c.nc.PublishRequest("$JS.API.DIRECT.GET."+stream, inbox, req); err != nil {
+	if err := c.nc.PublishRequest("$JS.API.DIRECT.GET."+l.Stream, inbox, req); err != nil {
-		return nil, 0, err
+		return nil, false, err
 	}
-	var msgs []*nats.Msg
+	entries := []LogEntry{}
-	var next uint64
 	deadline := time.Now().Add(StateTimeout)
-	for done := false; !done; {
+	for {
 		msg, err := sub.NextMsg(time.Until(deadline))
 		if err != nil {
 			if errors.Is(err, nats.ErrNoResponders) {
-				return nil, 0, missing()
+				return nil, false, notOnTheBus(l)
 			}
 			if errors.Is(err, nats.ErrTimeout) {
-				return nil, 0, fmt.Errorf("the read of %s did not end in %s", name, StateTimeout)
+				return nil, false, fmt.Errorf("the read of %s did not end in %s", l.Name, StateTimeout)
 			}
-			return nil, 0, err
+			return nil, false, err
 		}
 		switch status := msg.Header.Get("Status"); status {
 		case "":
-			msgs = append(msgs, msg)
+			e, err := logEntryOf(l, msg)
-		case "204": // the end of the batch: what is still pending after the last message sent
+			if err != nil {
-			pending, _ := strconv.ParseUint(msg.Header.Get("Nats-Num-Pending"), 10, 64)
+				return nil, false, err
-			if pending > 0 && len(msgs) > 0 {
-				last, _ := strconv.ParseUint(msgs[len(msgs)-1].Header.Get("Nats-Sequence"), 10, 64)
-				next = last + 1
 			}
-			done = true
+			entries = append(entries, *e)
-		case "404": // nothing from that sequence on
+		case "204": // the end of the batch: what is still pending after the last entry sent
-			done = true
+			pending, _ := strconv.ParseUint(msg.Header.Get("Nats-Num-Pending"), 10, 64)
+			return entries, pending > 0, nil
+		case "404": // nothing from there on
+			return entries, false, nil
 		case "503":
-			return nil, 0, missing()
+			return nil, false, notOnTheBus(l)
 		default:
-			return nil, 0, fmt.Errorf("the read of %s was refused by the bus: %s %s", name, status, msg.Header.Get("Description"))
+			return nil, false, fmt.Errorf("the read of %s was refused by the bus: %s %s", l.Name, status, msg.Header.Get("Description"))
 		}
 	}
-	return msgs, next, nil
+}
+
+// LogWatch hands deliver every entry of the keys matching a pattern from a sequence on (from), from a time on (since),
+// or, with neither, each new one, in the log's order, one at a time, through an ordered consumer it lets go of when
+// stopped (mesh-sdk process.LogReader.Watch). It returns once the entries there at its start are handed over; an error
+// from deliver is said, and the watch goes on with the next entry.
+func (c *Conn) LogWatch(module, name, keys string, from uint64, since time.Time, deliver func(LogEntry) error) (func(), error) {
+	l, err := c.issuedLog(module, name)
+	if err != nil {
+		return nil, err
+	}
+	filter, err := logFilter(l, keys)
+	if err != nil {
+		return nil, err
+	}
+	cfg := jetstream.OrderedConsumerConfig{FilterSubjects: []string{filter}, DeliverPolicy: jetstream.DeliverNewPolicy}
+	switch {
+	case from != 0 && !since.IsZero():
+		return nil, fmt.Errorf("a watch of %s starts from a sequence or from a time, not both", name)
+	case from != 0:
+		cfg.DeliverPolicy, cfg.OptStartSeq = jetstream.DeliverByStartSequencePolicy, from
+	case !since.IsZero():
+		at := since.UTC()
+		cfg.DeliverPolicy, cfg.OptStartTime = jetstream.DeliverByStartTimePolicy, &at
+	}
+	js, err := jetstream.New(c.nc)
+	if err != nil {
+		return nil, err
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
+	defer cancel()
+	cons, err := js.OrderedConsumer(ctx, l.Stream, cfg)
+	if errors.Is(err, jetstream.ErrStreamNotFound) {
+		return nil, notOnTheBus(l)
+	}
+	if err != nil {
+		return nil, fmt.Errorf("the watch of %s (stream %s): %w", name, l.Stream, err)
+	}
+	info, err := cons.Info(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("the watch of %s (stream %s): %w", name, l.Stream, err)
+	}
+	it, err := cons.Messages()
+	if err != nil {
+		return nil, fmt.Errorf("the watch of %s (stream %s): %w", name, l.Stream, err)
+	}
+	hand := func(m jetstream.Msg) error {
+		meta, err := m.Metadata()
+		if err != nil {
+			return err
+		}
+		e := LogEntry{Sequence: meta.Sequence.Stream, Key: strings.TrimPrefix(m.Subject(), l.Subject+"."),
+			Value: valueOf(m.Data()), At: meta.Timestamp.UTC()}
+		return deliver(e)
+	}
+	for pending := info.NumPending; pending > 0; pending-- {
+		m, err := it.Next()
+		if err == nil {
+			err = hand(m)
+		}
+		if err != nil {
+			it.Stop()
+			return nil, fmt.Errorf("the watch of %s: %w", name, err)
+		}
+	}
+	var stopped atomic.Bool
+	go func() {
+		for {
+			m, err := it.Next()
+			if err != nil || stopped.Load() {
+				return // stopped, or the connection closed; a message fetched as it stopped is not handed over
+			}
+			if err := hand(m); err != nil {
+				c.Logf("[mesh-tools] %s did not take entry of %s: %v; it is in the log, a read finds it", module, name, err)
+			}
+		}
+	}()
+	return func() { stopped.Store(true); it.Stop() }, nil
 }
 
 // LogLast is a key's last entry, or nil when it has none.

node-tools/internal/bus/state.go

@@ -37,7 +37,8 @@ type StateEntry struct {
 	Revision uint64          `json:"revision"`
 }
 
-// StateChange is one change a watch delivers: a key put or deleted.
+// StateChange is one change a watch delivers: a key put, deleted, or expired (its own time to live, or the
+// bucket's age, ended; issue 1952).
 type StateChange struct {
 	State    string          `json:"state"`
 	Key      string          `json:"key"`
@@ -173,6 +174,51 @@ func (c *Conn) StateHistory(module, name, key string) ([]StateKept, error) {
 	return kept, nil
 }
 
+// StateInfo is how full a state's bucket is against its caps (tools-are-how-a-module-is-spoken-to): a bundle's tools
+// say fill against cap from it.
+type StateInfo struct {
+	State  string `json:"state"`
+	Bucket string `json:"bucket"`
+	// Bytes is what the bucket holds now, every kept value and marker counted; Keys how many keys it holds, a key
+	// whose last value is a delete or expiry marker counted until the marker goes.
+	Bytes uint64 `json:"bytes"`
+	Keys  uint64 `json:"keys"`
+	// MaxBytes and MaxValueBytes are the bucket's caps; -1 is none.
+	MaxBytes      int64 `json:"max_bytes"`
+	MaxValueBytes int32 `json:"max_value_bytes"`
+	// History is how many values a key keeps; TTLSeconds the bucket's age for a value (0 none); KeyTTL whether a key may
+	// carry its own time to live (issue 1952).
+	History    int64 `json:"history"`
+	TTLSeconds int64 `json:"ttl_seconds"`
+	KeyTTL     bool  `json:"key_ttl"`
+}
+
+// StateInfo reads a state's bucket's fill and caps. It is a read like StateGet: served to the owner and every reader.
+func (c *Conn) StateInfo(module, name string) (*StateInfo, error) {
+	s, err := c.issuedState(module, name)
+	if err != nil {
+		return nil, err
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
+	defer cancel()
+	kv, err := c.bucket(ctx, s)
+	if err != nil {
+		return nil, err
+	}
+	status, err := kv.Status(ctx)
+	if err != nil {
+		return nil, err
+	}
+	b, ok := status.(*jetstream.KeyValueBucketStatus)
+	if !ok || b.StreamInfo() == nil {
+		return nil, fmt.Errorf("the bus answered no stream behind %s's state %s", module, name)
+	}
+	info := b.StreamInfo()
+	return &StateInfo{State: name, Bucket: status.Bucket(), Bytes: info.State.Bytes, Keys: uint64(info.State.NumSubjects),
+		MaxBytes: info.Config.MaxBytes, MaxValueBytes: info.Config.MaxMsgSize, History: status.History(),
+		TTLSeconds: int64(status.TTL() / time.Second), KeyTTL: info.Config.AllowMsgTTL && status.LimitMarkerTTL() > 0}, nil
+}
+
 // StatePut writes one key, as the module, where the module keeps the state. It answers the revision.
 func (c *Conn) StatePut(module, name, key string, value json.RawMessage) (uint64, error) {
 	s, err := c.writable(module, name)
@@ -199,6 +245,61 @@ func (c *Conn) StatePut(module, name, key string, value json.RawMessage) (uint64
 	return kv.Put(ctx, key, value)
 }
 
+// MinKeyTTL is the shortest time to live a key may be given: the bus counts a key's age in whole seconds (mesh-sdk
+// process.MinTTL).
+const MinKeyTTL = time.Second
+
+// StateCreateTTL writes one key only when it has no value, with its own time to live (novox/hq ADR 0564 clauses 4 and
+// 7, issue 1952; mesh-sdk process.State.CreateTTL): the bus removes it when ttl passes and every watch hears the
+// removal as an expiry, so a bundle's wake needs no timer. A key that has a value is refused as ErrStateChanged. The
+// bucket must take keys with their own time to live (a state declared `key-ttl`); one that does not is refused, naming
+// it, before anything is written.
+func (c *Conn) StateCreateTTL(module, name, key string, value json.RawMessage, ttl time.Duration) (uint64, error) {
+	if ttl < MinKeyTTL {
+		return 0, fmt.Errorf("%s's state %s, key %s: a time to live of %s is shorter than the bus keeps (%s)",
+			module, name, key, ttl, MinKeyTTL)
+	}
+	return c.stateWrite(module, name, key, value, func(ctx context.Context, kv jetstream.KeyValue) (uint64, error) {
+		if err := takesKeyTTL(ctx, kv, module, name); err != nil {
+			return 0, err
+		}
+		return kv.Create(ctx, key, value, jetstream.KeyTTL(ttl))
+	})
+}
+
+// StateRenewTTL writes one key whole with a fresh time to live, whatever it held (mesh-sdk process.State.RenewTTL): a
+// time to live is set only when a key is created, so a live key is deleted first, and its watches hear the delete and
+// then the new value.
+func (c *Conn) StateRenewTTL(module, name, key string, value json.RawMessage, ttl time.Duration) (uint64, error) {
+	rev, err := c.StateCreateTTL(module, name, key, value, ttl)
+	if !errors.Is(err, ErrStateChanged) {
+		return rev, err
+	}
+	if err := c.StateDelete(module, name, key); err != nil {
+		return 0, err
+	}
+	return c.StateCreateTTL(module, name, key, value, ttl)
+}
+
+// takesKeyTTL refuses a bucket that does not allow a time to live per key, or leaves no marker when one ends: there the
+// server would refuse the write, or the key would vanish and no watch would hear it.
+func takesKeyTTL(ctx context.Context, kv jetstream.KeyValue, module, name string) error {
+	status, err := kv.Status(ctx)
+	if err != nil {
+		return err
+	}
+	allows := status.LimitMarkerTTL() > 0
+	if b, ok := status.(*jetstream.KeyValueBucketStatus); ok && b.StreamInfo() != nil {
+		allows = allows && b.StreamInfo().Config.AllowMsgTTL
+	}
+	if !allows {
+		return fmt.Errorf("%s's state %s does not take keys with their own time to live: its bucket %s allows none, or "+
+			"leaves no marker a watch hears when one ends; %s declares the state `key-ttl` for it (novox/hq ADR 0564, "+
+			"issue 1952)", module, name, status.Bucket(), module)
+	}
+	return nil
+}
+
 // StateDelete removes one key, as the module, where the module keeps the state. A key that was not
 // there is not an error: what is asked for is that it is gone.
 func (c *Conn) StateDelete(module, name, key string) error {
@@ -322,6 +423,12 @@ func (c *Conn) StateWatch(module, name, pattern string, deliver func(StateChange
 	go func() {
 		initial := true
 		for e := range w.Updates() {
+			// A watch given up, or stopped, hands over nothing more, though values stay buffered in the
+			// watcher: they would reach a watch id the bundle no longer holds (issue 774: 1257 orphan
+			// changes after 51 failed watches).
+			if watching.Err() != nil {
+				break
+			}
 			if e == nil {
 				// The end of what was there when the watch began.
 				if initial {
@@ -330,6 +437,11 @@ func (c *Conn) StateWatch(module, name, pattern string, deliver func(StateChange
 				}
 				continue
 			}
+			if initial {
+				// Arrival from the bus is progress before the bundle is asked: the bound measures the current
+				// values coming in, and restarts when the bundle has taken one.
+				progress()
+			}
 			change := StateChange{State: name, Key: e.Key(), Revision: e.Revision(), Current: initial}
 			switch e.Operation() {
 			case jetstream.KeyValuePut:
@@ -343,6 +455,15 @@ func (c *Conn) StateWatch(module, name, pattern string, deliver func(StateChange
 					continue
 				}
 				change.Op = "delete"
+				if e.Operation() == jetstream.KeyValuePurge {
+					// The marker the server leaves when a key's time to live ends (Nats-Marker-Reason MaxAge,
+					// read by the client as a purge): nothing else purges a module's state, since neither the
+					// runtime nor mesh-sdk's State offers a purge (issue 1952).
+					change.Op = "expire"
+				}
+			}
+			if watching.Err() != nil {
+				break
 			}
 			if err := deliver(change); err != nil {
 				c.Logf("[mesh-tools] %s did not take %s %s of %s: %v; its next change, or its next start, reads it again",

node-tools/internal/bus/state_watch_test.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -116,6 +117,56 @@ func TestAWatchIsRefusedWhenNoCurrentValueIsTakenWithinTheBound(t *testing.T) {
 	}
 }
 
+// **A membership the grants part wrote in its old wrapper is served whole** (issue 774). Until 2026-10-30 the part wrote
+// its pass record, the membership inside, into its state memberships; the tool runner read the value as a bare
+// membership, took it over the assignments stream's copy, and served no tool and no state. This is what the live mesh
+// held: read through mesh-sdk's codec, its tools are answered where the membership says and its state is reached.
+func TestAMembershipTheGrantsPartWroteWrappedIsServed(t *testing.T) {
+	mesh := mt.New(t)
+	mesh.Bucket(t, "gitea_repos")
+	gitea := mt.MembershipOf("gitea", "anchor", false, nil)
+	gitea.State = []bus.StateIssued{{Name: "repos", Bucket: "gitea_repos", Writes: true}}
+	mesh.IssueInGrantsWrapped(t, gitea, "published")
+	// The old wrapper of a membership the part held back carries its unpublished composition: refused, so the
+	// assignments stream's copy is served, never what the hold kept back.
+	pages := mt.MembershipOf("pages", "anchor", false, nil)
+	mesh.Issue(t, pages)
+	heldBack := mt.MembershipOf("pages", "anchor", true, nil)
+	mesh.IssueInGrantsWrapped(t, heldBack, "not published, a shadow difference: publishing would lose the last holder of x")
+	conn, err := bus.Connect(bus.Credential{URL: mt.URL(t), Module: "node-tools", Node: "anchor"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	conn.Logf = func(string, ...any) {}
+	t.Cleanup(conn.Close)
+	conn.Follow("gitea")
+	mt.Until(t, func() error {
+		if m := conn.Membership("gitea"); m == nil || len(m.Serves) == 0 {
+			return fmt.Errorf("gitea's membership is %+v, want the one inside the wrapper", m)
+		}
+		return nil
+	})
+	if got, want := fmt.Sprint(conn.ServedOn("gitea", "gitea_list")), "[{mesh.mod.gitea.tool.gitea_list.anchor }]"; got != want {
+		t.Errorf("gitea_list is served on %s, want %s", got, want)
+	}
+	if _, err := conn.StatePut("gitea", "repos", "novox.hq", []byte(`{"v":1}`)); err != nil {
+		t.Fatalf("the state the membership issues is refused: %v", err)
+	}
+	if e, err := conn.StateGet("gitea", "repos", "novox.hq"); err != nil || e == nil || string(e.Value) != `{"v":1}` {
+		t.Errorf("the state reads back %+v, %v", e, err)
+	}
+	conn.Follow("pages")
+	mt.Until(t, func() error {
+		if m := conn.Membership("pages"); m == nil || len(m.Serves) == 0 {
+			return fmt.Errorf("pages' membership is %+v, want the assignments stream's", m)
+		}
+		return nil
+	})
+	if got, want := fmt.Sprint(conn.ServedOn("pages", "pages_list")), "[{mesh.mod.pages.tool.pages_list.anchor }]"; got != want {
+		t.Errorf("a held-back membership's wrapper is served: pages_list on %s, want the stream's %s", got, want)
+	}
+}
+
 // owedWatcher is the gitea module's view of the checker's owed: the bucket, and a runtime connection
 // that serves gitea with the owed issued for reading.
 func owedWatcher(t *testing.T) (*bus.Conn, interface {
@@ -142,3 +193,72 @@ func owedWatcher(t *testing.T) (*bus.Conn, interface {
 	})
 	return conn, kv
 }
+
+// **A watch given up, or stopped, hands over nothing more** (issue 774). The runtime's watch stopped its watcher
+// when the current values did not arrive within the bound, but the watcher still held values, and its goroutine
+// went on handing them over: to a watch id the bundle no longer held (on the provisioner, 51 failed watches were
+// followed by 1257 orphan changes). Each row blocks the bundle on its first value, gives the watch up, lets the
+// bundle go and counts what it is still handed.
+func TestAWatchGivenUpHandsOverNothingMore(t *testing.T) {
+	for _, tc := range []struct {
+		name    string
+		refused bool // the bound gives it up, or the caller stops it
+	}{
+		{"refused at the bound", true},
+		{"stopped by its caller", false},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			was := bus.StateTimeout
+			bus.StateTimeout = 300 * time.Millisecond
+			t.Cleanup(func() { bus.StateTimeout = was })
+			conn, kv := owedWatcher(t)
+			put := func(i int) {
+				if _, err := kv.Put(fmt.Sprintf("novox/hq.%d.%040x.status=mesh/check", i+1, i), []byte(`{"v":1}`)); err != nil {
+					t.Fatal(err)
+				}
+			}
+			if tc.refused {
+				for i := 0; i < 4; i++ {
+					put(i)
+				}
+			}
+			var calls atomic.Int32
+			first := make(chan struct{}, 1)
+			release := make(chan struct{})
+			deliver := func(ch bus.StateChange) error {
+				if calls.Add(1) == 1 {
+					first <- struct{}{}
+					<-release
+				}
+				return nil
+			}
+			var stop func()
+			if tc.refused {
+				if _, err := conn.StateWatch("gitea", "mesh-delivery-checker.owed", "", deliver); err == nil {
+					t.Fatal("the watch whose bundle took nothing was not refused")
+				}
+			} else {
+				var err error
+				if stop, err = conn.StateWatch("gitea", "mesh-delivery-checker.owed", "", deliver); err != nil {
+					t.Fatal(err)
+				}
+				for i := 0; i < 4; i++ {
+					put(i)
+				}
+				select {
+				case <-first:
+				case <-time.After(5 * time.Second):
+					t.Fatal("the bundle was handed no change")
+				}
+				stop()
+			}
+			// Let the watcher's server side deliver what it held, then free the bundle.
+			time.Sleep(300 * time.Millisecond)
+			close(release)
+			time.Sleep(500 * time.Millisecond)
+			if n := calls.Load(); n != 1 {
+				t.Fatalf("the bundle was handed %d changes of a watch given up, want 1 (the one in hand)", n)
+			}
+		})
+	}
+}

node-tools/internal/bus/streamread.go

@@ -1,126 +0,0 @@
-package bus
-
-import (
-	"encoding/json"
-	"fmt"
-	"sort"
-	"strconv"
-	"strings"
-	"time"
-
-	"github.com/nats-io/nats.go"
-)
-
-// A module reading its own stream back (novox/hq ADR 0363; the operator, 2026-10-30: a build step's whole failure output
-// is said on its build module's stream, and the module's tools read it there). As a log's read: one batch direct get,
-// no consumer, and only a stream the module's own membership names (`streams`); the bus grants its owner that stream's
-// info and direct get and no other's (mesh-sdk buscompose ownStreamGrants).
-
-// StreamEntry is one event of a module's own stream: its sequence, its event (the subject after the stream's
-// `…event.`), its body and when the bus took it.
-type StreamEntry struct {
-	Sequence uint64          `json:"sequence"`
-	Event    string          `json:"event"`
-	Body     json.RawMessage `json:"body"`
-	At       time.Time       `json:"at"`
-}
-
-// StreamPage is one page of a stream's read: Next is the sequence to continue from, 0 when nothing follows; Last the
-// stream's last sequence.
-type StreamPage struct {
-	Entries []StreamEntry `json:"entries"`
-	Next    uint64        `json:"next"`
-	Last    uint64        `json:"last"`
-}
-
-// streamBusName is a stream's local name as the bus holds it (mesh-sdk catalogue.StreamBusName): capitals, a hyphen an
-// underscore; a bus name is itself.
-func streamBusName(name string) string { return strings.ToUpper(strings.ReplaceAll(name, "-", "_")) }
-
-// issuedStream is the stream of its own a module may read by a name, local or the bus's, or the reason it may not.
-func (c *Conn) issuedStream(module, name string) (StreamIssued, error) {
-	m := c.Membership(module)
-	if m == nil {
-		return StreamIssued{}, fmt.Errorf("%s has no membership issued on %s yet, so no stream of it is readable "+
-			"until the mesh issues one (novox/hq ADR 0363)", module, c.node)
-	}
-	var names []string
-	for _, s := range m.Streams {
-		if s.Stream == streamBusName(name) && strings.Contains(s.Subject, "{event}") {
-			return s, nil
-		}
-		names = append(names, s.Stream)
-	}
-	sort.Strings(names)
-	issued := "none"
-	if len(names) > 0 {
-		issued = strings.Join(names, ", ")
-	}
-	return StreamIssued{}, fmt.Errorf("%s keeps no stream called %q: a module reads the streams it declares under `streams`, "+
-		"and no other's; it was issued: %s (novox/hq ADR 0363)", module, name, issued)
-}
-
-// StreamRead is a module's own stream's events from a sequence on, whose event names match a pattern (`*` one name, `>`
-// the rest, last; empty is every event), at most limit (LogReadDefault when 0, at most LogReadMax) and about LogPageBytes.
-func (c *Conn) StreamRead(module, name, events string, from uint64, limit int) (*StreamPage, error) {
-	s, err := c.issuedStream(module, name)
-	if err != nil {
-		return nil, err
-	}
-	switch {
-	case limit == 0:
-		limit = LogReadDefault
-	case limit < 0 || limit > LogReadMax:
-		return nil, fmt.Errorf("a read's limit is 1 to %d events, %d when none is given, not %d", LogReadMax, LogReadDefault, limit)
-	}
-	if events == "" {
-		events = ">"
-	}
-	parts := strings.Split(events, ".")
-	for i, p := range parts {
-		if !(p == ">" && i == len(parts)-1) && p != "*" && !logToken.MatchString(p) {
-			return nil, fmt.Errorf("%q is not an event pattern: names, `*` for one and `>` for the rest, last", events)
-		}
-	}
-	prefix, _, _ := strings.Cut(s.Subject, "{event}")
-	missing := func() error {
-		return fmt.Errorf("the stream %s is issued and not on the bus yet: the controller creates it from the catalogue "+
-			"on its next raise, and reads it directly once it allows that (novox/hq ADR 0363)", s.Stream)
-	}
-	msgs, next, err := c.directBatch(s.Stream, prefix+events, from, limit, s.Stream, missing)
-	if err != nil {
-		return nil, err
-	}
-	page := &StreamPage{Entries: []StreamEntry{}, Next: next}
-	for _, msg := range msgs {
-		e, err := streamEntryOf(s.Stream, prefix, msg)
-		if err != nil {
-			return nil, err
-		}
-		page.Entries = append(page.Entries, *e)
-	}
-	info, err := c.logInfo(LogIssued{Name: s.Stream, Stream: s.Stream})
-	if err != nil {
-		return nil, err
-	}
-	page.Last = info.Last
-	return page, nil
-}
-
-// streamEntryOf is an event a direct get answered: its event is its subject after prefix.
-func streamEntryOf(stream, prefix string, msg *nats.Msg) (*StreamEntry, error) {
-	subject := msg.Header.Get("Nats-Subject")
-	event, ok := strings.CutPrefix(subject, prefix)
-	if !ok {
-		return nil, fmt.Errorf("the bus answered an event of %s on %q, which is not under %s", stream, subject, prefix)
-	}
-	seq, err := strconv.ParseUint(msg.Header.Get("Nats-Sequence"), 10, 64)
-	if err != nil {
-		return nil, fmt.Errorf("the bus answered an event of %s with no sequence", stream)
-	}
-	at, err := time.Parse(time.RFC3339Nano, msg.Header.Get("Nats-Time-Stamp"))
-	if err != nil {
-		return nil, fmt.Errorf("the bus answered event %d of %s with no time: %v", seq, stream, err)
-	}
-	return &StreamEntry{Sequence: seq, Event: event, Body: valueOf(msg.Data), At: at.UTC()}, nil
-}

node-tools/internal/bus/streams_test.go

@@ -160,6 +160,69 @@ func TestAModuleHearsEveryStreamItsMembershipNames(t *testing.T) {
 		}
 		return nil
 	})
+
+	// The membership as the grants part keeps it (ADR 0571 clause 7): read first, applied live, and the assignments
+	// stream's copy passed over once the state has given one.
+	granted := reader
+	granted.Hears = []bus.HeardIssued{{Stream: "MAIL", Consumer: "from-grants"}}
+	mesh.IssueInGrants(t, granted)
+	again, err := bus.Connect(bus.Credential{URL: mt.URL(t), Module: "node-tools", Node: "anchor"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	again.Logf = func(string, ...any) {}
+	t.Cleanup(again.Close)
+	again.Follow("reader")
+	consumerOf := func() string {
+		if m := again.Membership("reader"); m != nil && len(m.Hears) == 1 {
+			return m.Hears[0].Consumer
+		}
+		return ""
+	}
+	if got := consumerOf(); got != "from-grants" {
+		t.Fatalf("the membership read at the follow names %q, want the grants part's", got)
+	}
+	granted.Hears[0].Consumer = "from-grants-2"
+	mesh.IssueInGrants(t, granted)
+	mt.Until(t, func() error {
+		if got := consumerOf(); got != "from-grants-2" {
+			return fmt.Errorf("the membership names %q, want the grants part's change", got)
+		}
+		return nil
+	})
+	// A later, different copy on the assignments stream changes nothing once the state has given one.
+	arrived := make(chan struct{}, 1)
+	probe, err := bus.Connect(bus.Credential{URL: mt.URL(t), Module: "probe", Node: "elsewhere"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	probe.Logf = func(string, ...any) {}
+	t.Cleanup(probe.Close)
+	stopProbe, err := probe.Raw(bus.MembershipSubject("anchor", "reader"), func(string, []byte) []byte {
+		select {
+		case arrived <- struct{}{}:
+		default:
+		}
+		return nil
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(stopProbe)
+	probe.Flush()
+	streamed := reader
+	streamed.Hears = []bus.HeardIssued{{Stream: "MAIL", Consumer: "from-stream"}}
+	mesh.Issue(t, streamed)
+	select {
+	case <-arrived:
+	case <-time.After(5 * time.Second):
+		t.Fatal("the stream's copy was not published")
+	}
+	again.Flush()
+	time.Sleep(200 * time.Millisecond)
+	if got := consumerOf(); got != "from-grants-2" {
+		t.Fatalf("the stream's copy was applied over the grants part's: the membership names %q", got)
+	}
 }
 
 // novox/hq ADR 0363 §6 (issue 725): a consumer the membership names before the controller made it is bound once

node-tools/internal/console/address.go

@@ -17,6 +17,9 @@ package console
 // session — so a tool that arrived a minute ago is found without the client reconnecting.
 
 import (
+	"git.novox.be/novox/mesh-sdk/go/catalogue"
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+
 	"context"
 	"encoding/json"
 	"fmt"
@@ -128,13 +131,17 @@ type index struct {
 	NotAnswering []string
 	Listing      *Listing // the flat catalogue the call path resolves subjects with
 	Discovery    announce.Discovery
-	// Recorded is where the controller's records place each module that declares tools.
+	// Recorded is where placement places each module that declares tools.
 	Recorded map[string][]string
 	// Unheard is every runtime whose answer this index lacks: it said it was there and did not say
 	// what it serves in time, or it answered before and not now. An address it might answer is never
 	// called missing while it is here (2026-10-05).
 	Unheard []unheard
-	// Replaces is what each verb and tool replaces, from the controller's records (novox/hq ADR 0245):
+	// Placed is the state seat-holders as the seat mesh-placement's holder wrote it (novox/hq ADR 0571 clause 4), by its
+	// keys `<seat>` and `<seat>.<node>`, read with delivery.HoldersOf; PlacedWhy says why it could not be read.
+	Placed    map[string][]byte
+	PlacedWhy string
+	// Replaces is what each verb and tool replaces (novox/hq ADR 0245):
 	// `seat:<seat>.<verb>` and `<module>.<tool>`.
 	Replaces map[string][]string
 	// Earlier is every runtime not heard this round whose last answer, within RuntimeForgotten, this
@@ -224,42 +231,8 @@ func findTool(tools []Tool, name string) *Tool {
 	return nil
 }
 
-// controllerOutput is a controller seat verb's answer: the command's printed output.
+// recordedModule is a module as the bus holds it: where the placement seat places it, and whether its manifest declares
-func controllerOutput(conn *bus.Conn, verb string) (string, error) {
+// tools — what should announce itself, and where (novox/hq ADR 0571).
-	got, err := conn.Ask("seat:mesh-controller."+verb, map[string]any{}, "")
-	if err != nil {
-		return "", err
-	}
-	var r struct {
-		Output string `json:"output"`
-		OK     *bool  `json:"ok"`
-	}
-	if json.Unmarshal(got.Result, &r) != nil {
-		return "", fmt.Errorf("mesh-controller.%s answered something that is not its output", verb)
-	}
-	if r.OK != nil && !*r.OK {
-		return "", fmt.Errorf("mesh-controller.%s: %s", verb, strings.TrimSpace(r.Output))
-	}
-	return r.Output, nil
-}
-
-// jsonIn is the JSON document a command printed, after any lines it said first: a seat verb runs
-// the controller's command, and a command may warn before it answers.
-func jsonIn(output string) string {
-	t := strings.TrimSpace(output)
-	if strings.HasPrefix(t, "{") || strings.HasPrefix(t, "[") {
-		return t
-	}
-	for _, open := range []string{"\n{", "\n["} {
-		if i := strings.Index(output, open); i >= 0 {
-			return strings.TrimSpace(output[i+1:])
-		}
-	}
-	return ""
-}
-
-// recordedModule is a module as the controller's records hold it (`module list --json`): where the
-// mesh assigned it, and whether it declares tools — what should announce itself, and where.
 type recordedModule struct {
 	Module string   `json:"module"`
 	On     []string `json:"on"`
@@ -268,11 +241,6 @@ type recordedModule struct {
 	Replaces map[string][]string `json:"replaces,omitempty"`
 }
 
-// recordedMachine is a machine as the controller's records hold it (`node list --json`).
-type recordedMachine struct {
-	Name string `json:"name"`
-}
-
 // indexOn asks the mesh what it holds (novox/hq ADR 0197): what answers, from every runtime's own
 // announcement on the bus — one `$SRV.INFO` request — and what should, from the controller's records
 // read as JSON. Nothing is inferred from a roster and nothing is parsed from print.
@@ -282,19 +250,17 @@ type recordedMachine struct {
 // index holds instead, marked with its age (novox/hq issue 554).
 func indexOn(conn *bus.Conn, held announce.Known, before func(announce.Discovery) []earlier) (*index, error) {
 	var wg sync.WaitGroup
-	var nodesOut, modulesOut string
+	var placed map[string][]byte
-	var toolsAnswer json.RawMessage
+	var placedErr error
-	wg.Add(3)
+	var recorded []recordedModule
-	go func() { defer wg.Done(); nodesOut, _ = controllerOutput(conn, "nodes") }()
+	var known []string
-	go func() { defer wg.Done(); modulesOut, _ = controllerOutput(conn, "modules") }()
+	var recordsWhy []string
-	// The seats' verbs as the records define them, for what each replaces (novox/hq ADR 0245): answered
+	wg.Add(2)
-	// in-process from the records, never a command.
+	// Which modules there are and where they are placed, and which machines the mesh has, from the bus (ADR 0571
-	go func() {
+	// clauses 4 and 5): the indexer's modules and manifests and the placement seat's placements, never the controller.
-		defer wg.Done()
+	go func() { defer wg.Done(); recorded, known, recordsWhy = recordsOn(conn) }()
-		if got, err := conn.Ask("seat:mesh-controller.tools", map[string]any{}, ""); err == nil {
+	// Who holds which seat on which node, from placement (ADR 0571 clause 4): no call of the controller.
-			toolsAnswer = got.Result
+	go func() { defer wg.Done(); placed, placedErr = seatHoldersOn(conn) }()
-		}
-	}()
 	d, err := announce.Gather(conn, held)
 	wg.Wait()
 	if err != nil {
@@ -333,6 +299,10 @@ func indexOn(conn *bus.Conn, held announce.Known, before func(announce.Discovery
 	l := x.Listing
 	x.Discovery = d
 	x.Stale, x.Earlier = stale, earlierAt
+	x.Placed = placed
+	if placedErr != nil {
+		x.PlacedWhy = placedErr.Error()
+	}
 	for _, s := range d.Silent {
 		x.Unheard = append(x.Unheard, unheard{Runtime: s.Name, Machine: s.Machine,
 			Why: fmt.Sprintf("it answered PING; what it serves did not arrive within %s", announce.Patience)})
@@ -340,25 +310,24 @@ func indexOn(conn *bus.Conn, held announce.Known, before func(announce.Discovery
 
 	// What should have answered: every assignment of a module that declares tools. Silence is named;
 	// a module with no tools is never a name here.
-	var recorded []recordedModule
+	x.Replaces = replacesOf(catalogue.DefaultSeats(), recorded)
-	recordedOK := json.Unmarshal([]byte(jsonIn(modulesOut)), &recorded) == nil
+	for _, m := range recorded {
-	x.Replaces = replacesOf(toolsAnswer, recorded)
+		if !m.Tools {
-	if recordedOK {
+			continue
-		for _, m := range recorded {
+		}
-			if !m.Tools {
+		x.Recorded[m.Module] = append([]string{}, m.On...)
-				continue
+		for _, n := range m.On {
-			}
+			// A holder of a seat held once for the mesh announces no machine — the seat is the mesh's,
-			x.Recorded[m.Module] = append([]string{}, m.On...)
+			// not a machine's — so what it announced without one answers for wherever it is assigned.
-			for _, n := range m.On {
+			if !announced[m.Module][n] && !announced[m.Module][""] {
-				// A holder of a seat held once for the mesh announces no machine — the seat is the mesh's,
+				l.NotAnswering = append(l.NotAnswering, m.Module+" on "+n)
-				// not a machine's — so what it announced without one answers for wherever it is assigned.
-				if !announced[m.Module][n] && !announced[m.Module][""] {
-					l.NotAnswering = append(l.NotAnswering, m.Module+" on "+n)
-				}
 			}
 		}
-	} else {
+	}
-		l.NotAnswering = append(l.NotAnswering, "mesh-controller (its records of the modules did not answer, so what is missing cannot be said)")
+	l.NotAnswering = append(l.NotAnswering, recordsWhy...)
+	if x.PlacedWhy != "" {
+		l.NotAnswering = append(l.NotAnswering, delivery.SeatHoldersRead+" (who holds which seat by placement could not be read: "+
+			x.PlacedWhy+")")
 	}
 	for _, u := range x.Unheard {
 		l.NotAnswering = append(l.NotAnswering, "the runtime "+sayUnheard([]unheard{u}))
@@ -366,13 +335,8 @@ func indexOn(conn *bus.Conn, held announce.Known, before func(announce.Discovery
 	sort.Strings(l.NotAnswering)
 	x.NotAnswering = l.NotAnswering
 
-	var known []recordedMachine
+	for _, n := range known {
-	if json.Unmarshal([]byte(jsonIn(nodesOut)), &known) == nil {
+		machines[n] = true
-		for _, n := range known {
-			if n.Name != "" {
-				machines[n.Name] = true
-			}
-		}
 	}
 	for n := range machines {
 		x.Machines = append(x.Machines, n)
@@ -381,6 +345,138 @@ func indexOn(conn *bus.Conn, held announce.Known, before func(announce.Discovery
 	return x, nil
 }
 
+// stateOn reads a state the runtime's membership reads, whole: its current values by key, a deleted key absent.
+func stateOn(conn *bus.Conn, name string) (map[string][]byte, error) {
+	var mu sync.Mutex
+	keys := map[string][]byte{}
+	stop, err := conn.StateWatch(conn.Credential().Module, name, "", func(c bus.StateChange) error {
+		if c.Current && c.Op == "put" {
+			mu.Lock()
+			keys[c.Key] = c.Value
+			mu.Unlock()
+		}
+		return nil
+	})
+	if err != nil {
+		return nil, err
+	}
+	stop()
+	mu.Lock()
+	defer mu.Unlock()
+	return keys, nil
+}
+
+// seatHoldersOn reads the state seat-holders whole (novox/hq ADR 0571 clause 4), as delivery.SeatHoldersRead.
+func seatHoldersOn(conn *bus.Conn) (map[string][]byte, error) {
+	return stateOn(conn, delivery.SeatHoldersRead)
+}
+
+// The indexer's states the console reads (novox/hq ADR 0571 clauses 4 and 5): every module, and each manifest by its
+// digest, by the names a reader's manifest declares.
+var (
+	modulesRead   = delivery.IndexerModule + ".modules"
+	manifestsRead = delivery.IndexerModule + ".manifests"
+)
+
+// manifestFacts is what the console takes of a manifest: the tools it declares and what each replaces. A manifest's
+// digest never names another, so what is read once is kept.
+type manifestFacts struct {
+	Tools    []string            `json:"tools"`
+	Replaces map[string][]string `json:"replaces"`
+}
+
+var manifestsKept sync.Map // digest → manifestFacts
+
+// recordsOn is every module the indexer holds, with where the placement seat places it and whether it declares tools,
+// and every machine placement names (delivery.ModuleNames, delivery.MachineNames). What cannot be read is said, one line
+// per state, for the listing's not answering.
+func recordsOn(conn *bus.Conn) (recorded []recordedModule, machines []string, why []string) {
+	placements, err := stateOn(conn, delivery.PlacementRead)
+	if err != nil {
+		why = append(why, delivery.PlacementRead+" (where modules are placed could not be read, so what is missing cannot "+
+			"be said: "+err.Error()+")")
+	}
+	machines = delivery.MachineNames(placements)
+	on := map[string][]string{}
+	// The placement's keys in either form, a node key and its module keys or the form before (research 072 row 37).
+	assembled, err := delivery.PlacementsOf(placements)
+	if err != nil {
+		why = append(why, delivery.PlacementRead+" (not every key is a placement: "+err.Error()+")")
+	}
+	for _, node := range machines {
+		for _, m := range assembled[node].Modules {
+			on[m.Module] = append(on[m.Module], node)
+		}
+	}
+	modules, err := stateOn(conn, modulesRead)
+	if err != nil {
+		return nil, machines, append(why, modulesRead+" (which modules there are could not be read: "+err.Error()+")")
+	}
+	names, err := delivery.ModuleNames(modules)
+	if err != nil {
+		return nil, machines, append(why, modulesRead+" ("+err.Error()+")")
+	}
+	for _, name := range names {
+		r := recordedModule{Module: name, On: on[name]}
+		if m, err := delivery.DecodeModule(modules[name]); err == nil && m.Manifest != "" {
+			facts, err := manifestOf(conn, m.Manifest)
+			if err != nil {
+				why = append(why, name+" (its manifest could not be read, so whether it answers tools is not known: "+err.Error()+")")
+			} else {
+				r.Tools, r.Replaces = len(facts.Tools) > 0, facts.Replaces
+			}
+		}
+		recorded = append(recorded, r)
+	}
+	return recorded, machines, why
+}
+
+// manifestOf is what the console takes of the manifest of a digest, read from the indexer's manifests once.
+func manifestOf(conn *bus.Conn, digest string) (manifestFacts, error) {
+	if kept, ok := manifestsKept.Load(digest); ok {
+		return kept.(manifestFacts), nil
+	}
+	entry, err := conn.StateGet(conn.Credential().Module, manifestsRead, digest)
+	if err != nil {
+		return manifestFacts{}, err
+	}
+	if entry == nil {
+		return manifestFacts{}, fmt.Errorf("the indexer holds no manifest %s", digest)
+	}
+	m, err := delivery.DecodeManifest(entry.Value)
+	if err != nil {
+		return manifestFacts{}, err
+	}
+	var facts manifestFacts
+	if err := json.Unmarshal([]byte(m.JSON), &facts); err != nil {
+		return manifestFacts{}, err
+	}
+	manifestsKept.Store(digest, facts)
+	return facts, nil
+}
+
+// placedHolders is who holds seat (on node, for a seat held per machine) by placement, or none.
+func (x *index) placedHolders(seat, node string) []delivery.SeatHolder {
+	if len(x.Placed) == 0 {
+		return nil
+	}
+	hs, err := delivery.HoldersOf(x.Placed, seat, node)
+	if err != nil || len(hs) > 0 || node != "" {
+		return hs
+	}
+	// A seat held per machine, asked without one: every node's key of it.
+	var all []delivery.SeatHolder
+	for key := range x.Placed {
+		if s, n, ok := strings.Cut(key, "."); ok && s == seat {
+			if more, err := delivery.HoldersOf(x.Placed, seat, n); err == nil {
+				all = append(all, more...)
+			}
+		}
+	}
+	sort.Slice(all, func(i, j int) bool { return all[i].Node+"/"+all[i].Module < all[j].Node+"/"+all[j].Module })
+	return all
+}
+
 func containsHolder(hs []holder, h holder) bool {
 	for _, x := range hs {
 		if x == h {
@@ -606,6 +702,14 @@ func resolve(x *index, address string) (target, error) {
 			return target{}, fmt.Errorf("%s is assigned to %s in the mesh's records, but nothing that answered discovery serves it: %s. "+
 				"Ask again in a moment; mesh_runtimes shows which runtimes answered", prefix, strings.Join(on, ", "), why)
 		}
+		if hs := x.placedHolders(prefix, node); len(hs) > 0 {
+			var held []string
+			for _, h := range hs {
+				held = append(held, h.Module+" on "+h.Node)
+			}
+			return target{}, fmt.Errorf("the seat %s is held by %s by placement, but no holder of it announced itself on the bus. "+
+				"Ask again in a moment; mesh_runtimes shows which runtimes answered", prefix, strings.Join(held, ", "))
+		}
 		if len(x.Unheard) > 0 {
 			return target{}, fmt.Errorf("nothing that answered discovery is called %s, but not every runtime answered: %s — "+
 				"%s may be theirs. Ask again in a moment; mesh_runtimes shows which runtimes answered", prefix, sayUnheard(x.Unheard), prefix)

node-tools/internal/console/address_test.go

@@ -70,29 +70,28 @@ func TestTheMeshsToolsAreFoundByAddress(t *testing.T) {
 		t.Cleanup(stop)
 	}
 
-	// The controller: its records as JSON, as its seat verbs answer them, and its own seat's verbs
+	// The controller: its own seat's verbs announced on the bus like every runtime's; its records are asked by no one.
-	// announced on the bus like every runtime's.
 	controller := connect(t, "mesh-controller", "bench")
-	out := func(s string) map[string]any { return map[string]any{"output": s, "ok": true} }
+	// Its verb nodes, called through the console as any seat's verb is.
-	serve := func(verb string, answer func() any) {
+	stopNodes, err := controller.HandleSubject("mesh.seat.mesh-controller.tool.nodes", func(json.RawMessage) (any, error) {
-		stop, err := controller.HandleSubject("mesh.seat.mesh-controller.tool."+verb, func(json.RawMessage) (any, error) {
+		return map[string]any{"ok": true, "output": `[{"name":"bench","mode":"converged"},{"name":"desk","mode":"converged"}]`}, nil
-			return answer(), nil
-		})
-		if err != nil {
-			t.Fatal(err)
-		}
-		t.Cleanup(stop)
-	}
-	serve("nodes", func() any {
-		return out("the bus's user list leaves out 2 user(s)\n" +
-			`[{"name":"bench","heard":"3m ago","mode":"converged","id":"1f2e"},{"name":"desk","heard":"here","mode":"converged","id":"9a8b"}]`)
 	})
-	var gammaOn atomic.Value
+	if err != nil {
-	gammaOn.Store("[]")
+		t.Fatal(err)
-	serve("modules", func() any {
+	}
-		return out(fmt.Sprintf(`[{"module":"alpha","on":["desk"],"tools":true},{"module":"beta","on":["desk"],"tools":true},`+
+	t.Cleanup(stopNodes)
-			`{"module":"gamma","on":%s,"tools":true},{"module":"delta","on":["desk"],"tools":false},`+
+	// What the mesh holds, as the indexer and placement write it (ADR 0571 clauses 4, 5): never the controller's records.
-			`{"module":"epsilon","on":["bench"],"tools":true},{"module":"mesh-controller","on":["bench"],"tools":true}]`, gammaOn.Load().(string)))
+	tools := map[string][]string{"alpha": {"one", "two"}, "beta": {"three"}, "gamma": {"given"}, "delta": nil,
+		"epsilon": {"seven"}, "mesh-controller": {"nodes"}}
+	mesh.Placed(t, tools, map[string][]string{"desk": {"alpha", "beta", "delta"}, "bench": {"epsilon", "mesh-controller"}})
+	console := mt.MembershipOf("node-tools", "desk", false, nil)
+	console.State = mt.ConsoleReads()
+	mesh.Issue(t, console)
+	mt.Until(t, func() error {
+		if m := nodeTools.Membership("node-tools"); m == nil || len(m.State) == 0 {
+			return fmt.Errorf("the console's reads are not issued yet")
+		}
+		return nil
 	})
 	stopAnn, err := announce.Serve(controller, announce.Service{Name: "mesh-controller", ID: "bench"}, func() []announce.Endpoint {
 		return []announce.Endpoint{{Kind: announce.KindSeat, Module: "mesh-controller", Seat: "mesh-controller", Scope: "mesh",
@@ -202,7 +201,7 @@ func TestTheMeshsToolsAreFoundByAddress(t *testing.T) {
 		t.Fatalf("gamma was found before it served: %s", got)
 	}
 	mesh.Issue(t, mt.MembershipOf("gamma", "desk", false, nil))
-	gammaOn.Store(`["desk"]`)
+	mesh.Placed(t, tools, map[string][]string{"desk": {"alpha", "beta", "delta", "gamma"}, "bench": {"epsilon", "mesh-controller"}})
 	late := connect(t, "node-tools", "desk")
 	stopLate, err := runtime.Run(late, []runtime.Served{{Module: "gamma", Entrypoints: []string{mt.Fixture("env-gamma.serve.mjs")}}},
 		nil, (&mt.Logs{}).Logf)
@@ -233,12 +232,3 @@ func TestTheMeshsToolsAreFoundByAddress(t *testing.T) {
 		t.Errorf("an old name: %s", got)
 	}
 }
-
-func TestTheControllersRecordsAreReadAsJSON(t *testing.T) {
-	if got := jsonIn("a warning printed first\n[{\"name\":\"ace\"}]"); got != `[{"name":"ace"}]` {
-		t.Errorf("an array after a warning: %q", got)
-	}
-	if got := jsonIn(`{"seats":[]}`); got != `{"seats":[]}` {
-		t.Errorf("an object: %q", got)
-	}
-}

node-tools/internal/console/console_test.go

@@ -3,6 +3,7 @@ package console
 import (
 	"bytes"
 	"encoding/json"
+	"fmt"
 	"io"
 	"net/http"
 	"strings"
@@ -56,14 +57,19 @@ func TestTheConsoleListsAndCallsOverHTTP(t *testing.T) {
 	}
 	defer stop()
 
-	// The controller's records (ADR 0197): ghost declares tools on desk and nothing announces it.
+	// What the mesh holds, as the indexer and placement write it (ADR 0571): ghost declares tools on desk and nothing
-	controller := connect(t, "mesh-controller", "")
+	// announces it.
-	stopSeat, _ := controller.HandleSubject("mesh.seat.mesh-controller.tool.modules", func(json.RawMessage) (any, error) {
+	mesh.Placed(t, map[string][]string{"alpha": {"one", "two"}, "beta": {"three"}, "ghost": {"boo"}},
-		return map[string]any{"ok": true, "output": `[{"module":"alpha","on":["desk"],"tools":true},` +
+		map[string][]string{"desk": {"alpha", "beta", "ghost"}})
-			`{"module":"beta","on":["desk"],"tools":true},{"module":"ghost","on":["desk"],"tools":true}]`}, nil
+	console := mt.MembershipOf("node-tools", "desk", false, nil)
+	console.State = mt.ConsoleReads()
+	mesh.Issue(t, console)
+	mt.Until(t, func() error {
+		if m := nodeTools.Membership("node-tools"); m == nil || len(m.State) == 0 {
+			return fmt.Errorf("the console's reads are not issued yet")
+		}
+		return nil
 	})
-	defer stopSeat()
-	controller.Flush()
 
 	flat := NewSurface(nodeTools, "desk.node-tools")
 	flat.Flat = true // the whole catalogue, as before ADR 0195: still reachable, no longer announced

node-tools/internal/console/manifest_waits_test.go

@@ -26,7 +26,7 @@ func TestTheManifestNamesTheFirstFamiliesOneByOne(t *testing.T) {
 		t.Fatal(err)
 	}
 	want := []string{"mesh-knowledge.reviewed", "mesh-delivery-gate.transition", "gitea.forge.status",
-		"mesh-controller.checked", "operator-channel.decided.claude-code"}
+		"mesh-self-check.checked", "operator-channel.decided.claude-code"}
 	if m.Module != "node-tools" || !reflect.DeepEqual(m.Waits, want) {
 		t.Errorf("%s waits on %v, want %v", m.Module, m.Waits, want)
 	}

node-tools/internal/console/mcp.go

@@ -6,6 +6,8 @@ package console
 import (
 	"context"
 	"encoding/json"
+	"git.novox.be/novox/mesh-sdk/go/catalogue"
+	"git.novox.be/novox/mesh-sdk/go/delivery"
 	"os"
 	"strings"
 	"sync"
@@ -60,6 +62,34 @@ type Surface struct {
 	Flat bool
 }
 
+// MethodSeats is the console's JSON-RPC method beside MCP's that answers every seat's verbs and holders, in the shape
+// the controller's `tools` answered: `{seats: [{seat, scope, tools: [{name, description, input, replaces}], holders:
+// [{module, node}]}], holders_why}`.
+const MethodSeats = "mesh/seats"
+
+// seatsAnswer is MethodSeats' answer: mesh-sdk's seat definitions, with each seat's holders by placement.
+func (s *Surface) seatsAnswer() map[string]any {
+	placed, err := seatHoldersOn(s.conn)
+	x := &index{Placed: placed}
+	seats := []map[string]any{}
+	for _, st := range catalogue.DefaultSeats() {
+		holders := x.placedHolders(st.Name, "")
+		if holders == nil {
+			holders = []delivery.SeatHolder{}
+		}
+		tools := st.Serves
+		if tools == nil {
+			tools = []catalogue.Verb{}
+		}
+		seats = append(seats, map[string]any{"seat": st.Name, "scope": st.Scope, "tools": tools, "holders": holders})
+	}
+	out := map[string]any{"seats": seats}
+	if err != nil {
+		out["holders_why"] = "who holds which seat by placement could not be read: " + err.Error()
+	}
+	return out
+}
+
 // NewSurface is the surface over a connection, as `who`.
 func NewSurface(conn *bus.Conn, who string) *Surface {
 	return &Surface{conn: conn, who: who, Flat: os.Getenv("MESH_CONSOLE_FLAT") == "1"}
@@ -125,6 +155,11 @@ func (s *Surface) HandleContext(ctx context.Context, r Request) *Reply {
 			return nil
 		}
 		return answer(r.ID, map[string]any{})
+	case MethodSeats:
+		// The seats' contract and who holds each, for a client on this machine that has no seat definitions of its own
+		// (the TypeScript `mesh`): read in-process from mesh-sdk and from placement, never asked of the controller
+		// (novox/hq ADR 0571 clauses 4 and 5).
+		return answer(r.ID, s.seatsAnswer())
 	case "tools/list":
 		if !s.Flat {
 			return answer(r.ID, map[string]any{"tools": discovery()})

node-tools/internal/console/runtimes_test.go

@@ -2,10 +2,13 @@ package console
 
 import (
 	"encoding/json"
+	"fmt"
 	"strings"
+	"sync/atomic"
 	"testing"
 	"time"
 
+	"git.novox.be/novox/mesh-sdk/go/delivery"
 	"github.com/nats-io/nats.go/micro"
 
 	"github.com/novox/mesh-tools/node-tools/internal/announce"
@@ -52,7 +55,23 @@ func TestARuntimeThatAnswersLateIsFoundAndOneThatNeverDoesIsNamed(t *testing.T)
 	// Kept for the test: one round of discovery answers every question but mesh_runtimes, which asks anew.
 	IndexKept, announce.Patience = time.Minute, 2500*time.Millisecond
 	t.Cleanup(func() { IndexKept, announce.Patience = keptWas, patienceWas })
-	mt.New(t)
+	mesh := mt.New(t)
+	// What the mesh holds, as the indexer and placement write it (ADR 0571): never the controller's records.
+	mesh.Placed(t, map[string][]string{"slack": {"slack_check"}, "nap": {"doze"}, "nextcloud-client": {"nextcloud_check"}},
+		map[string][]string{"laptop": {"slack", "nextcloud-client"}, "sleeper": {"nap", "nextcloud-client"},
+			"desktop": {"nextcloud-client"}, "bench": nil})
+	// Who holds which seat, by placement (ADR 0571 clause 4), as the seat mesh-placement's holder composes it.
+	placedKV := mesh.Bucket(t, delivery.SeatHoldersBucket)
+	for key, sh := range delivery.SeatHoldersOf(map[string]delivery.Placement{"bench": {Node: "bench",
+		Holds: []delivery.Holding{{Claim: "mesh-wants", Scope: "mesh", Module: "mesh-delivery-assigner"}}}}) {
+		raw, _ := json.Marshal(sh)
+		if _, err := placedKV.Put(key, raw); err != nil {
+			t.Fatal(err)
+		}
+	}
+	consoleMembership := mt.MembershipOf("node-tools", "laptop-console", false, nil)
+	consoleMembership.State = mt.ConsoleReads()
+	mesh.Issue(t, consoleMembership)
 
 	laptop := connect(t, "node-tools", "laptop")
 	stopTool, err := laptop.HandleSubject("mesh.mod.slack.tool.slack_check.laptop", func(json.RawMessage) (any, error) {
@@ -77,13 +96,16 @@ func TestARuntimeThatAnswersLateIsFoundAndOneThatNeverDoesIsNamed(t *testing.T)
 	})
 
 	controller := connect(t, "mesh-controller", "bench")
+	var toolsAsked atomic.Int32
 	for verb, output := range map[string]string{
-		"nodes": `[{"name":"laptop"},{"name":"sleeper"},{"name":"desktop"},{"name":"bench"}]`,
+		// The controller's records are served here only to see that none is asked (ADR 0571 clauses 4 and 5).
-		"modules": `[{"module":"slack","on":["laptop"],"tools":true},{"module":"nap","on":["sleeper"],"tools":true},` +
+		"nodes":   `[{"name":"laptop"}]`,
-			`{"module":"nextcloud-client","on":["laptop","desktop","sleeper"],"tools":true}]`,
+		"modules": `[]`,
+		"tools":   `{"seats":[]}`,
 	} {
-		output := output
+		output, verb := output, verb
 		stop, err := controller.HandleSubject("mesh.seat.mesh-controller.tool."+verb, func(json.RawMessage) (any, error) {
+			toolsAsked.Add(1)
 			return map[string]any{"output": output, "ok": true}, nil
 		})
 		if err != nil {
@@ -94,6 +116,13 @@ func TestARuntimeThatAnswersLateIsFoundAndOneThatNeverDoesIsNamed(t *testing.T)
 	controller.Flush()
 
 	console := connect(t, "node-tools", "laptop-console")
+	console.Follow("node-tools")
+	mt.Until(t, func() error {
+		if console.Membership("node-tools") == nil {
+			return fmt.Errorf("the console's membership is not issued yet")
+		}
+		return nil
+	})
 	up, err := Serve(NewSurface(console, "laptop.node-tools"), "127.0.0.1:0")
 	if err != nil {
 		t.Fatal(err)
@@ -113,8 +142,20 @@ func TestARuntimeThatAnswersLateIsFoundAndOneThatNeverDoesIsNamed(t *testing.T)
 		t.Errorf("the late runtime's machine lists nothing: %s", machine)
 	}
 
+	// The seats' contract and holders, for a client on this machine with no seat definitions (the TypeScript `mesh`).
+	seats := post(t, endpoint, `{"jsonrpc":"2.0","id":9,"method":"mesh/seats"}`)
+	if raw, _ := json.Marshal(seats["result"]); !strings.Contains(string(raw), `"holders":[{"module":"mesh-delivery-assigner","node":"bench"}],"scope":"mesh","seat":"mesh-wants"`) ||
+		!strings.Contains(string(raw), `"replaces":["journalctl"]`) {
+		t.Errorf("mesh/seats answered %s", raw)
+	}
+	if n := toolsAsked.Load(); n != 0 {
+		t.Errorf("the console asked the controller's records %d times", n)
+	}
+
 	// The silent one is named, never called missing.
 	for address, want := range map[string]string{
+		// A seat whose holder announced nothing is named with its holder by placement, never called missing.
+		"mesh-wants.read":                          "the seat mesh-wants is held by mesh-delivery-assigner on bench by placement",
 		"sleeper/nap.doze":                         "nap is assigned to sleeper in the mesh's records, but nothing that answered discovery serves it: discovery did not hear in full from node-tools on sleeper (it answered PING",
 		"sleeper/nextcloud-client.nextcloud_check": "nextcloud-client on sleeper did not answer discovery: discovery did not hear in full from node-tools on sleeper",
 	} {

node-tools/internal/console/search.go

@@ -19,7 +19,8 @@ package console
 // Ties go to a seat — the place tools belong — then to the shorter name, then by address.
 
 import (
-	"encoding/json"
+	"git.novox.be/novox/mesh-sdk/go/catalogue"
+
 	"path"
 	"sort"
 	"strings"
@@ -265,27 +266,16 @@ func (x *index) search(query string) []hit {
 	return hits
 }
 
-// replacesOf is what every verb and tool replaces, from the controller's records (novox/hq ADR 0245):
+// replacesOf is what every verb and tool replaces (novox/hq ADR 0245): a seat's verbs from the seat's contract in
-// a seat's verbs from its `tools` answer, keyed `seat:<seat>.<verb>`, and a module's own tools from
+// mesh-sdk, keyed `seat:<seat>.<verb>`, read in-process and asked of no part (ADR 0571 clause 5: no part keeps a
-// its `modules` answer, keyed `<module>.<tool>`. A controller older than the field answers neither,
+// central record of tools); and a module's own tools from its records, keyed `<module>.<tool>`. A verb or a tool with
-// and search matches names and descriptions as before.
+// no `replaces` is matched by its name and description as before.
-func replacesOf(toolsAnswer json.RawMessage, modules []recordedModule) map[string][]string {
+func replacesOf(seats []catalogue.Seat, modules []recordedModule) map[string][]string {
 	out := map[string][]string{}
-	var seats struct {
+	for _, s := range seats {
-		Seats []struct {
+		for _, v := range s.Serves {
-			Seat  string `json:"seat"`
+			if len(v.Replaces) > 0 {
-			Tools []struct {
+				out["seat:"+s.Name+"."+v.Name] = append([]string(nil), v.Replaces...)
-				Name     string   `json:"name"`
-				Replaces []string `json:"replaces"`
-			} `json:"tools"`
-		} `json:"seats"`
-	}
-	if json.Unmarshal(toolsAnswer, &seats) == nil {
-		for _, s := range seats.Seats {
-			for _, t := range s.Tools {
-				if len(t.Replaces) > 0 {
-					out["seat:"+s.Seat+"."+t.Name] = t.Replaces
-				}
 			}
 		}
 	}

node-tools/internal/console/search_test.go

@@ -1,7 +1,8 @@
 package console
 
 import (
-	"encoding/json"
+	"git.novox.be/novox/mesh-sdk/go/catalogue"
+
 	"strings"
 	"testing"
 
@@ -19,6 +20,7 @@ func meshToSearch(t *testing.T) *index {
 		"status":  "One unit as the service manager sees it now: its states, whether it starts at boot.",
 		"restart": "Restart one unit.",
 		"journal": "The last lines of one unit's journal (at most 2000).",
+		"failed":  "The units that failed.",
 	}
 	for _, node := range []string{"anchor", "laptop"} {
 		for verb, does := range verbs {
@@ -44,13 +46,7 @@ func meshToSearch(t *testing.T) *index {
 			Description: does, Subject: "mesh.mod.docker.tool." + tool + ".anchor"})
 	}
 	x, _, _ := indexOfEndpoints(endpoints)
-	tools := json.RawMessage(`{"seats":[{"seat":"node-service-manager","scope":"node","tools":[
+	x.Replaces = replacesOf(catalogue.DefaultSeats(), []recordedModule{{Module: "docker", On: []string{"anchor"}, Tools: true,
-		{"name":"units","replaces":["systemctl list-units","systemctl --failed"]},
-		{"name":"status","replaces":["systemctl status","systemctl is-active"]},
-		{"name":"restart","replaces":["systemctl restart"]},
-		{"name":"journal","replaces":["journalctl"]}]},
-		{"seat":"mesh-controller","scope":"mesh","tools":[{"name":"node","replaces":["hostnamectl","uptime"]},{"name":"nodes"}]}]}`)
-	x.Replaces = replacesOf(tools, []recordedModule{{Module: "docker", On: []string{"anchor"}, Tools: true,
 		Replaces: map[string][]string{"docker_list": {"docker ps"}, "docker_logs": {"docker logs"}, "docker_restart": {"docker restart"}}}})
 	return x
 }
@@ -68,7 +64,7 @@ func TestSearchFindsTheToolThatReplacesACommandFirst(t *testing.T) {
 		"systemctl status":                    "anchor/node-service-manager.status",
 		"systemctl status nats.service":       "anchor/node-service-manager.status",
 		"systemctl restart mesh-controller":   "anchor/node-service-manager.restart",
-		"systemctl --failed":                  "anchor/node-service-manager.units",
+		"systemctl --failed":                  "anchor/node-service-manager.failed",
 		"docker ps":                           "anchor/docker.docker_list",
 		"docker ps -a":                        "anchor/docker.docker_list",
 		"docker logs":                         "anchor/docker.docker_logs",
@@ -134,10 +130,10 @@ func TestSearchStillMatchesEveryWord(t *testing.T) {
 	}
 }
 
-// A controller older than `replaces` answers without it: search matches names and descriptions as before.
+// A seat's contract and a module's records without `replaces`: search matches names and descriptions as before.
 func TestSearchWithoutReplacesMatchesAsBefore(t *testing.T) {
 	x := meshToSearch(t)
-	x.Replaces = replacesOf(json.RawMessage(`{"seats":[{"seat":"node-service-manager","tools":[{"name":"journal"}]}]}`),
+	x.Replaces = replacesOf([]catalogue.Seat{{Name: "node-service-manager", Serves: []catalogue.Verb{{Name: "journal"}}}},
 		[]recordedModule{{Module: "docker"}})
 	if len(x.Replaces) != 0 {
 		t.Fatalf("replaces from records that have none: %v", x.Replaces)

node-tools/internal/launch/event_answer_test.go

@@ -38,7 +38,6 @@ func (b *subscribing) Publish(json.RawMessage) error                          {
 func (b *subscribing) Ask(json.RawMessage) (json.RawMessage, error)           { return nil, nil }
 func (b *subscribing) State(string, json.RawMessage) (json.RawMessage, error) { return nil, nil }
 func (b *subscribing) Log(string, json.RawMessage) (json.RawMessage, error)   { return nil, nil }
-func (b *subscribing) StreamRead(json.RawMessage) (json.RawMessage, error)    { return nil, nil }
 func (b *subscribing) Watch(json.RawMessage, func(json.RawMessage) error) (func(), error) {
 	return func() {}, nil
 }
@@ -96,3 +95,7 @@ func TestAnEventIsTakenByAnyAnswerThatIsNotAnError(t *testing.T) {
 		t.Fatalf("an error answer was not the bundle's refusal of mesh/event: %#v", err)
 	}
 }
+
+func (b *subscribing) LogWatch(json.RawMessage, func(json.RawMessage) error) (func(), error) {
+	return func() {}, nil
+}

node-tools/internal/launch/launch.go

@@ -80,9 +80,8 @@ type Bus interface {
 	Watch(params json.RawMessage, deliver func(change json.RawMessage) error) (stop func(), err error)
 	// Log answers a bundle's `append`, `read`, `last` and `info` on its module's log (novox/hq ADR 0297).
 	Log(verb string, params json.RawMessage) (json.RawMessage, error)
-	// StreamRead answers a bundle's `read` of one of its module's own streams (novox/hq ADR 0363): {stream, subject,
+	// LogWatch hands deliver each entry of a log from where the bundle asked, one at a time (novox/hq ADR 0571).
-	// from, limit}, as a log's read.
+	LogWatch(params json.RawMessage, deliver func(entry json.RawMessage) error) (stop func(), err error)
-	StreamRead(params json.RawMessage) (json.RawMessage, error)
 	// Seat answers a bundle's seat traffic (novox/hq ADR 0259 §3): `publish`, `prove` and `record` at once;
 	// `take` and `answer` bind once and hand each message or proof on through handOn, to whichever child
 	// of the module asked last — a restarted child asks again as its code runs again.
@@ -326,21 +325,16 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an
 								result = answered
 							}
 						case "mesh/state.get", "mesh/state.put", "mesh/state.delete", "mesh/state.keys",
-							"mesh/state.create", "mesh/state.update", "mesh/state.history":
+							"mesh/state.create", "mesh/state.update", "mesh/state.history", "mesh/state.info":
 							var answered json.RawMessage
 							if answered, err = mesh.State(strings.TrimPrefix(m.Method, "mesh/state."), m.Params); err == nil {
 								result = answered
 							}
-						case "mesh/log.append", "mesh/log.read", "mesh/log.last", "mesh/log.info":
+						case "mesh/log.append", "mesh/log.read", "mesh/log.newest", "mesh/log.last", "mesh/log.info":
 							var answered json.RawMessage
 							if answered, err = mesh.Log(strings.TrimPrefix(m.Method, "mesh/log."), m.Params); err == nil {
 								result = answered
 							}
-						case "mesh/stream.read":
-							var answered json.RawMessage
-							if answered, err = mesh.StreamRead(m.Params); err == nil {
-								result = answered
-							}
 						case "mesh/provision.get":
 							var answered json.RawMessage
 							if answered, err = mesh.Provision("get", m.Params); err == nil {
@@ -365,6 +359,25 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an
 									c.mu.Unlock()
 								}
 							}
+						case "mesh/log.watch":
+							// As a state's watch: each entry asked of this child in order, `mesh/log {watch, entry}`,
+							// naming the watch by the id the child asked it with (novox/hq ADR 0571).
+							var stop func()
+							watch := m.ID
+							stop, err = mesh.LogWatch(m.Params, func(entry json.RawMessage) error {
+								_, err := c.ask(module, "mesh/log", map[string]any{"watch": watch, "entry": entry}, EventTimeout)
+								return err
+							})
+							if err == nil {
+								c.mu.Lock()
+								if c.why != nil {
+									c.mu.Unlock()
+									stop()
+								} else {
+									c.watches = append(c.watches, stop)
+									c.mu.Unlock()
+								}
+							}
 						case "mesh/state.watch":
 							// Each change is asked of this child, in order, naming the watch it is for by
 							// the id the child asked it with — two watches of one state are two handlers;

node-tools/internal/launch/provision_test.go

@@ -105,3 +105,7 @@ func TestABundleAsksForItsValuesAndIsHandedItsWatchsChanges(t *testing.T) {
 		t.Errorf("the runtime's bus was asked %q", b.asked)
 	}
 }
+
+func (b *provisioning) LogWatch(json.RawMessage, func(json.RawMessage) error) (func(), error) {
+	return func() {}, nil
+}

node-tools/internal/meshtest/meshtest.go

@@ -11,7 +11,11 @@
 package meshtest
 
 import (
+	"context"
 	"encoding/json"
+	"git.novox.be/novox/mesh-sdk/go/buscompose"
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+	"git.novox.be/novox/mesh-sdk/go/membership"
 	"os"
 	"path/filepath"
 	"runtime"
@@ -22,6 +26,7 @@ import (
 
 	"github.com/nats-io/nats-server/v2/server"
 	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
 
 	"github.com/novox/mesh-tools/node-tools/internal/bus"
 )
@@ -104,6 +109,111 @@ func (m *Mesh) Issue(t *testing.T, mem bus.Membership) {
 	}
 }
 
+// IssueInGrants writes a membership where the grants part keeps it, its state memberships, key `<node>.<module>`
+// (novox/hq ADR 0571 clause 7; mesh-sdk buscompose.GrantsMembershipsBucket), making the bucket when it is not there.
+// It writes with mesh-sdk's encoder, the grants part's own, never json of this package's type: a value built here by
+// hand agreed with the reader's mistake and hid that the part wrote another shape (issue 774).
+func (m *Mesh) IssueInGrants(t *testing.T, mem bus.Membership) {
+	t.Helper()
+	m.putGrants(t, mem, membership.EncodeGrantsMembership(sdkMembership(t, mem)))
+}
+
+// IssueInGrantsWrapped writes a membership as the grants part wrote it before mesh-sdk's codec (until 2026-10-30, what
+// the live mesh holds until the part writes again): its pass record, the membership inside, published saying what
+// became of it ("published", "unchanged", or why it was held back).
+func (m *Mesh) IssueInGrantsWrapped(t *testing.T, mem bus.Membership, published string) {
+	t.Helper()
+	body, _ := json.Marshal(map[string]any{"node": mem.Node, "module": mem.Module, "digest": "sha256:0",
+		"membership": sdkMembership(t, mem), "published": published})
+	m.putGrants(t, mem, body)
+}
+
+func (m *Mesh) putGrants(t *testing.T, mem bus.Membership, body []byte) {
+	t.Helper()
+	kv, err := m.js.KeyValue(buscompose.GrantsMembershipsBucket)
+	if err != nil {
+		if kv, err = m.js.CreateKeyValue(&nats.KeyValueConfig{Bucket: buscompose.GrantsMembershipsBucket, History: 1}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if _, err := kv.Put(buscompose.MembershipKey(mem.Node, mem.Module), body); err != nil {
+		t.Fatal(err)
+	}
+}
+
+// sdkMembership is a membership of this package's type as mesh-sdk's, which the grants part composes and writes.
+func sdkMembership(t *testing.T, mem bus.Membership) membership.Membership {
+	t.Helper()
+	raw, _ := json.Marshal(mem)
+	var out membership.Membership
+	if err := json.Unmarshal(raw, &out); err != nil {
+		t.Fatal(err)
+	}
+	return out
+}
+
+// ConsoleReads are the states a console's runtime reads to say what the mesh holds (novox/hq ADR 0571 clauses 4, 5):
+// placement, the indexer's modules and manifests, and the seat holders.
+func ConsoleReads() []bus.StateIssued {
+	return []bus.StateIssued{
+		{Name: delivery.PlacementRead, Bucket: delivery.PlacementBucket},
+		{Name: delivery.IndexerModule + ".modules", Bucket: delivery.ModulesBucket},
+		{Name: delivery.IndexerModule + ".manifests", Bucket: delivery.ManifestsBucket},
+		{Name: delivery.SeatHoldersRead, Bucket: delivery.SeatHoldersBucket},
+	}
+}
+
+// Placed writes what the mesh holds as its parts write it, with mesh-sdk's encoders: each module the indexer holds with
+// its manifest (tools, the tools it declares; none for a module without), and each node's placement of modules.
+func (m *Mesh) Placed(t *testing.T, tools map[string][]string, on map[string][]string) {
+	t.Helper()
+	kv := func(bucket string) nats.KeyValue {
+		_ = m.js.DeleteKeyValue(bucket)
+		b, err := m.js.CreateKeyValue(&nats.KeyValueConfig{Bucket: bucket, History: 1, MaxValueSize: 256 * 1024})
+		if err != nil {
+			t.Fatal(err)
+		}
+		return b
+	}
+	modules, manifests, placement := kv(delivery.ModulesBucket), kv(delivery.ManifestsBucket), kv(delivery.PlacementBucket)
+	kv(delivery.SeatHoldersBucket)
+	for module, declared := range tools {
+		manifest, _ := json.Marshal(map[string]any{"module": module, "tools": declared})
+		digest := delivery.ManifestDigest(manifest)
+		raw, err := delivery.EncodeManifest(delivery.Manifest{Module: module, JSON: string(manifest)})
+		if err != nil {
+			t.Fatal(err)
+		}
+		if _, err := manifests.Put(digest, raw); err != nil {
+			t.Fatal(err)
+		}
+		raw, err = delivery.EncodeModule(delivery.Module{None: "not built here", Manifest: digest, Repository: "novox/" + module,
+			Commit: strings.Repeat("a", 40)})
+		if err != nil {
+			t.Fatal(err)
+		}
+		if _, err := modules.Put(module, raw); err != nil {
+			t.Fatal(err)
+		}
+	}
+	for node, placed := range on {
+		p := delivery.Placement{V: delivery.Version, Node: node}
+		for _, module := range placed {
+			p.Modules = append(p.Modules, delivery.PlacedModule{Module: module})
+		}
+		// As the seat writes it: its node key and a key per module (research 072 row 37).
+		values, err := delivery.PlacementValues(p)
+		if err != nil {
+			t.Fatal(err)
+		}
+		for key, raw := range values {
+			if _, err := placement.Put(key, raw); err != nil {
+				t.Fatal(err)
+			}
+		}
+	}
+}
+
 // NextEvent is the subject the next event under a pattern lands on.
 func (m *Mesh) NextEvent(t *testing.T, pattern string) <-chan string {
 	t.Helper()
@@ -243,6 +353,23 @@ func (m *Mesh) Bucket(t *testing.T, name string) nats.KeyValue {
 	return kv
 }
 
+// BucketKeyTTL makes a module's state afresh as the controller does for one declared `key-ttl` (mesh-sdk buscompose:
+// a time to live per key allowed, its end leaving a marker kept a minute; issue 1952).
+func (m *Mesh) BucketKeyTTL(t *testing.T, name string) {
+	t.Helper()
+	_ = m.js.DeleteKeyValue(name)
+	js, err := jetstream.New(m.nc)
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: 1, MaxValueSize: 256 * 1024,
+		LimitMarkerTTL: time.Minute}); err != nil {
+		t.Fatal(err)
+	}
+}
+
 // Log makes a module's log afresh as the controller does from the catalogue (novox/hq ADR 0297): a file
 // stream named LOG_<module>_<name> on mesh.log.<module>.<name>.>, limits retention, no age, no cap per
 // key, the declared cap, 260 KiB a message (a 256 KiB value and its headers), refusing new entries when full, direct get, no delete or
@@ -278,7 +405,7 @@ func (m *Mesh) ModuleStream(t *testing.T, module, name string) {
 	t.Helper()
 	_ = m.js.DeleteStream(name)
 	if _, err := m.js.AddStream(&nats.StreamConfig{Name: name, Subjects: []string{"mesh.stream." + module + ".event.>"},
-		Retention: nats.LimitsPolicy, Discard: nats.DiscardOld, AllowDirect: true}); err != nil {
+		Retention: nats.LimitsPolicy, Discard: nats.DiscardOld}); err != nil {
 		t.Fatal(err)
 	}
 }

node-tools/internal/provide/provide.go

@@ -16,6 +16,7 @@ import (
 	"time"
 
 	"git.novox.be/novox/mesh-sdk/go/provider"
+	"git.novox.be/novox/mesh-sdk/go/provisions"
 
 	"github.com/novox/mesh-tools/node-tools/internal/bus"
 )
@@ -38,15 +39,18 @@ func Offers() map[string]provider.Offer {
 }
 
 // busOf is the helper's bus over the runtime's own state: the provider's bucket is its state `bound`, and the
-// controller's assignments are the state it reads, `mesh-controller.assignments` (both declared in the manifest).
+// provisions and plans of the seat mesh-provisioner are the states it reads, by mesh-sdk's names (novox/hq ADR 0570,
+// issue 1954 L6-T1). The controller's assignments are read no more (mesh-sdk 692dbf9, L6-P3).
 type busOf struct{ conn *bus.Conn }
 
 func (b busOf) state(bucket string) (string, error) {
 	switch bucket {
 	case provider.BucketOf(Module):
 		return "bound", nil
-	case provider.AssignmentsBucket:
+	case provisions.ProvisionsBucket:
-		return "mesh-controller.assignments", nil
+		return provisions.ProvisionsRead, nil
+	case provisions.PlansBucket:
+		return provisions.PlansRead, nil
 	}
 	return "", fmt.Errorf("%s is not a bucket this provider keeps or reads", bucket)
 }
@@ -57,7 +61,8 @@ func (b busOf) Watch(ctx context.Context, bucket string, handle func(key string,
 		return err
 	}
 	stop, err := b.conn.StateWatch(Module, name, "", func(ch bus.StateChange) error {
-		if ch.Op == "delete" {
+		if ch.Op != "put" {
+			// A delete, or an expiry (issue 1952): the key is gone.
 			handle(ch.Key, nil)
 		} else {
 			handle(ch.Key, ch.Value)

node-tools/internal/provide/provide_test.go

@@ -7,6 +7,10 @@ import (
 	"testing"
 	"time"
 
+	"git.novox.be/novox/mesh-sdk/go/catalogue"
+	"git.novox.be/novox/mesh-sdk/go/provider"
+	"git.novox.be/novox/mesh-sdk/go/provisions"
+
 	"github.com/novox/mesh-tools/node-tools/internal/bus"
 	mt "github.com/novox/mesh-tools/node-tools/internal/meshtest"
 )
@@ -41,19 +45,39 @@ func fill(t *testing.T, template string, entry map[string]any) string {
 func TestTheRuntimeAnswersAHandedConsumersMcpEndpointThroughTheHelper(t *testing.T) {
 	mesh := mt.New(t)
 	bound := mesh.Bucket(t, "node-tools_bound")
-	assignments := mesh.Bucket(t, "mesh-controller_assignments")
+	plans := mesh.Bucket(t, provisions.PlansBucket)
+	provs := mesh.Bucket(t, provisions.ProvisionsBucket)
 
 	mem := mt.MembershipOf("node-tools", "anchor", false, nil)
 	mem.State = []bus.StateIssued{
 		{Name: "bound", Bucket: "node-tools_bound", Writes: true},
-		{Name: "mesh-controller.assignments", Bucket: "mesh-controller_assignments"},
+		{Name: provisions.PlansRead, Bucket: provisions.PlansBucket},
+		{Name: provisions.ProvisionsRead, Bucket: provisions.ProvisionsBucket},
 	}
 	mesh.Issue(t, mem)
 
-	// claude-code, handed on this machine, is bound to node-tools here for mcp-endpoint (issue 788's plan entry).
+	// claude-code, handed on this machine, is bound to node-tools here for mcp-endpoint: its plan and its provision
-	plan := `{"node":"anchor","module":"claude-code","provisioning":{"provisions":[{"provision":"mcp-endpoint",` +
+	// key, written as mesh-provisioner writes them, with mesh-sdk's encoders (ADR 0570 clauses 1-3, issue 1954).
-		`"provider":"anchor","provider-module":"node-tools","at":"127.0.0.1","from":"anchor","as":"mesh_anchor_claude_code"}]}}`
+	plan, err := provisions.EncodePlan(provisions.Plan{Node: "anchor", Module: "claude-code", Plan: catalogue.Plan{
-	if _, err := assignments.Put("anchor.claude-code", []byte(plan)); err != nil {
+		Provisions: []catalogue.PlanProvision{{Provision: "mcp-endpoint", Provider: "anchor", ProviderModule: "node-tools",
+			At: "127.0.0.1", From: "anchor", As: "mesh_anchor_claude_code"}}}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := plans.Put("anchor.claude-code", plan); err != nil {
+		t.Fatal(err)
+	}
+	key, err := provisions.Key("anchor", "claude-code", "mcp-endpoint", "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	asked, err := provisions.Encode(provisions.Provision{Provision: "mcp-endpoint", State: provisions.Asked,
+		Consumer: provisions.End{Node: "anchor", Module: "claude-code"}, Provider: provisions.End{Node: "anchor", Module: "node-tools"},
+		At: time.Now()})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := provs.Put(key, asked); err != nil {
 		t.Fatal(err)
 	}
 
@@ -102,6 +126,8 @@ func TestTheOfferServesThePortTheManifestServes(t *testing.T) {
 	var m struct {
 		Serves   map[string]map[string]any `json:"serves"`
 		Moved    bool                      `json:"moved"`
+		State    []string                  `json:"state"`
+		Reads    []string                  `json:"reads"`
 		Provides []struct {
 			Name       string `json:"name"`
 			AnsweredBy string `json:"answered-by"`
@@ -133,11 +159,32 @@ func TestTheOfferServesThePortTheManifestServes(t *testing.T) {
 			t.Error("the manifest still carries a consumers file the helper does not need")
 		}
 	}
+	for _, r := range m.Reads {
+		if r == "mesh-controller.assignments" {
+			t.Error("the manifest still reads mesh-controller.assignments, which the sdk's helper no longer reads (issue 1954)")
+		}
+	}
 	for _, a := range m.Build.Artifacts {
 		if a.Name == "files" || len(a.Env) > 0 {
 			t.Errorf("the build still carries %+v, which only a moved module needs", a)
 		}
 	}
+	// Every bucket the sdk's helper reaches is a state the manifest keeps or reads, by the sdk's names: the provider's
+	// own and the seat mesh-provisioner's provisions and plans (issue 1954 L6-T1); the controller's assignments no more.
+	declared := map[string]bool{}
+	for _, n := range append(m.State, m.Reads...) {
+		declared[n] = true
+	}
+	for bucket, want := range map[string]string{
+		provider.BucketOf(Module):   "bound",
+		provisions.ProvisionsBucket: "mesh-provisioner.provisions",
+		provisions.PlansBucket:      "mesh-provisioner.plans",
+	} {
+		got, err := (busOf{}).state(bucket)
+		if err != nil || got != want || !declared[got] {
+			t.Errorf("the helper's bucket %s reaches state %q (%v), want %q declared in the manifest", bucket, got, err, want)
+		}
+	}
 }
 
 // Only the machine's runtime is the provider: a process on another credential, or with no machine, answers nobody.

node-tools/internal/runtime/log_test.go

@@ -20,11 +20,13 @@ func logBuses(t *testing.T, maxBytes int64) (issues, other launch.Bus, mesh *mt.
 	t.Helper()
 	mesh = mt.New(t)
 	m := mt.MembershipOf("issues", "anchor", false, nil)
-	m.Logs = []bus.LogIssued{mesh.Log(t, "issues", "changes", maxBytes)}
+	changes := mesh.Log(t, "issues", "changes", maxBytes)
-	mesh.ModuleStream(t, "issues", "ISSUE_MOVES")
+	m.Logs = []bus.LogIssued{changes}
-	m.Streams = []bus.StreamIssued{{Stream: "ISSUE_MOVES", Events: []string{"moved.>"}, Subject: "mesh.stream.issues.event.{event}"}}
 	mesh.Issue(t, m)
-	mesh.Issue(t, mt.MembershipOf("other", "anchor", false, nil))
+	// other reads issues' log, as its manifest's reads declare it (ADR 0571): `<module>.<log>`, without writing it.
+	o := mt.MembershipOf("other", "anchor", false, nil)
+	o.Logs = []bus.LogIssued{{Name: "issues.changes", Stream: changes.Stream, Subject: changes.Subject}}
+	mesh.Issue(t, o)
 	conn := connect(t, "node-tools", "anchor")
 	conn.Follow("issues")
 	conn.Follow("other")
@@ -192,44 +194,63 @@ func TestALogAnswersEachVerbOnTheBus(t *testing.T) {
 		t.Fatalf("the runtime made %d consumers of the log; it uses none", n)
 	}
 
-	// A module reads its own stream back as it reads its log (mesh/stream.read): by its local name, an event pattern,
+	// From a time (ADR 0571): `since` in place of `from`, RFC 3339; both at once are refused.
-	// in pages; another module's stream is refused with the reason.
+	cut := all.Entries[2].At
-	for i, ev := range []string{"moved.a.1", "moved.b.1", "moved.a.1"} {
+	if p := readOf(t, b, fmt.Sprintf(`{"log":"changes","since":%q}`, cut)); len(p.Entries) == 0 || p.Entries[0].Sequence > seqs[2] ||
-		if err := b.Publish(json.RawMessage(fmt.Sprintf(`{"key":%q,"body":{"n":%d}}`, ev, i))); err != nil {
+		p.Entries[len(p.Entries)-1].Sequence != seqs[4] {
-			t.Fatal(err)
+		t.Fatalf("a read since %s: %+v", cut, p)
+	}
+	if p := readOf(t, b, fmt.Sprintf(`{"log":"changes","since":%q}`, time.Now().Add(time.Hour).Format(time.RFC3339))); len(p.Entries) != 0 {
+		t.Fatalf("a read since an hour ahead: %+v", p)
+	}
+	if _, err := logAsk(b, "read", `{"log":"changes","from":2,"since":"2026-10-29T00:00:00Z"}`); err == nil {
+		t.Fatal("a read from a sequence and a time was not refused")
+	}
+	// The newest of each key, in the log's order; `**` for the rest, as mesh-sdk says it.
+	raw, err = logAsk(b, "newest", `{"log":"changes"}`)
+	var newest []struct {
+		Sequence uint64 `json:"sequence"`
+		Key      string `json:"key"`
+	}
+	if err != nil || json.Unmarshal(raw, &newest) != nil || len(newest) != 3 || newest[0].Key != "13" || newest[1].Key != "14.notes" ||
+		newest[2].Sequence != seqs[4] {
+		t.Fatalf("newest: %s %v", raw, err)
+	}
+	if raw, err := logAsk(b, "newest", `{"log":"changes","keys":"14.**"}`); err != nil || !strings.Contains(string(raw), `"14.notes"`) ||
+		strings.Contains(string(raw), `"12"`) {
+		t.Fatalf("newest of 14.**: %s %v", raw, err)
+	}
+	// Another module's log, read by the name its reads declare; never written.
+	if p := readOf(t, other, `{"log":"issues.changes","from":1}`); len(p.Entries) != 5 {
+		t.Fatalf("other's read of issues.changes: %+v", p)
+	}
+	if _, err := logAsk(other, "append", `{"log":"issues.changes","key":"1","value":{}}`); err == nil ||
+		!strings.Contains(err.Error(), "without writing it") {
+		t.Fatalf("other's append to the log it reads: %v", err)
+	}
+	// A watch hands each entry from where it is asked, then each new one, one at a time.
+	heard := make(chan bus.LogEntry, 16)
+	stop, err := b.LogWatch(json.RawMessage(fmt.Sprintf(`{"log":"changes","keys":"12","from":%d}`, seqs[2])), func(raw json.RawMessage) error {
+		var e bus.LogEntry
+		_ = json.Unmarshal(raw, &e)
+		heard <- e
+		return nil
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer stop()
+	next := appendOf(t, b, `{"log":"changes","key":"12","value":{"op":"late"}}`)
+	for _, want := range []uint64{seqs[2], seqs[4], next} {
+		select {
+		case e := <-heard:
+			if e.Sequence != want || e.Key != "12" {
+				t.Fatalf("the watch handed %d %s, want %d", e.Sequence, e.Key, want)
+			}
+		case <-time.After(5 * time.Second):
+			t.Fatalf("the watch did not hand entry %d", want)
 		}
 	}
-	streamOf := func(b launch.Bus, params string) (wireStream, error) {
-		var p wireStream
-		raw, err := b.StreamRead(json.RawMessage(params))
-		if err == nil {
-			err = json.Unmarshal(raw, &p)
-		}
-		return p, err
-	}
-	page, err := streamOf(b, `{"stream":"issue-moves","subject":"moved.a.>","limit":1}`)
-	if err != nil || len(page.Entries) != 1 || page.Entries[0].Event != "moved.a.1" || string(page.Entries[0].Body) != `{"n":0}` || page.Next == 0 || page.Last != 3 {
-		t.Fatalf("the first page of moved.a: %+v, %v", page, err)
-	}
-	page, err = streamOf(b, fmt.Sprintf(`{"stream":"ISSUE_MOVES","subject":"moved.a.>","from":%d}`, page.Next))
-	if err != nil || len(page.Entries) != 1 || string(page.Entries[0].Body) != `{"n":2}` || page.Next != 0 {
-		t.Fatalf("the second page of moved.a: %+v, %v", page, err)
-	}
-	if _, err := streamOf(other, `{"stream":"issue-moves"}`); err == nil || !strings.Contains(err.Error(), "keeps no stream") {
-		t.Fatalf("another module's stream: %v", err)
-	}
-}
-
-// wireStream is a stream read's answer with exactly the contract's field names (mesh-sdk stdio.StreamPage).
-type wireStream struct {
-	Entries []struct {
-		Sequence uint64          `json:"sequence"`
-		Event    string          `json:"event"`
-		Body     json.RawMessage `json:"body"`
-		At       string          `json:"at"`
-	} `json:"entries"`
-	Next uint64 `json:"next"`
-	Last uint64 `json:"last"`
 }
 
 // A page holds about 4 MiB at most, whatever its limit, and says where to continue.
@@ -308,7 +329,7 @@ func TestALogNotIssuedIsRefusedWithTheReason(t *testing.T) {
 	} {
 		_, err := logAsk(other, verb, params)
 		if err == nil || !strings.Contains(err.Error(), `other keeps no log called "changes"`) ||
-			!strings.Contains(err.Error(), "issued: none") {
+			!strings.Contains(err.Error(), "issued: issues.changes") {
 			t.Errorf("%s of a log not issued: %v", verb, err)
 		}
 	}

node-tools/internal/runtime/runtime.go

@@ -868,10 +868,14 @@ type stateAsked struct {
 	Key      string          `json:"key"`
 	Value    json.RawMessage `json:"value"`
 	Revision uint64          `json:"revision"`
+	// TTLSeconds gives a created or put key its own time to live (issue 1952): create makes it only when it has no
+	// value, put renews it whole. Zero is none.
+	TTLSeconds int `json:"ttl_seconds"`
 }
 
-// State answers a bundle's `get`, `put`, `delete`, `keys`, `create`, `update` and `history` on its
+// State answers a bundle's `get`, `put`, `delete`, `keys`, `create`, `update`, `history` and `info` on its
-// module's state (ADR 0201, ADR 0259, ADR 0307). The runtime refuses, with the reason, a state the module was not issued and a write to one it only reads.
+// module's state (ADR 0201, ADR 0259, ADR 0307); `create` and `put` take `ttl_seconds`, a key's own time to live,
+// whose end every watch hears as an `expire` (issue 1952). The runtime refuses, with the reason, a state the module was not issued and a write to one it only reads.
 func (b *moduleBus) State(verb string, params json.RawMessage) (json.RawMessage, error) {
 	var asked stateAsked
 	if err := json.Unmarshal(params, &asked); err != nil || asked.State == "" {
@@ -897,7 +901,13 @@ func (b *moduleBus) State(verb string, params json.RawMessage) (json.RawMessage,
 		}
 		answer = kept
 	case "put":
-		revision, err := conn.StatePut(b.module, asked.State, asked.Key, asked.Value)
+		put := conn.StatePut
+		if asked.TTLSeconds != 0 {
+			put = func(module, name, key string, value json.RawMessage) (uint64, error) {
+				return conn.StateRenewTTL(module, name, key, value, time.Duration(asked.TTLSeconds)*time.Second)
+			}
+		}
+		revision, err := put(b.module, asked.State, asked.Key, asked.Value)
 		if err != nil {
 			return nil, err
 		}
@@ -907,6 +917,13 @@ func (b *moduleBus) State(verb string, params json.RawMessage) (json.RawMessage,
 			return nil, err
 		}
 		answer = map[string]any{}
+	case "info":
+		// The bucket's fill against its caps, so a bundle's tools can say how full it is.
+		info, err := conn.StateInfo(b.module, asked.State)
+		if err != nil {
+			return nil, err
+		}
+		answer = info
 	case "keys":
 		keys, err := conn.StateKeys(b.module, asked.State)
 		if err != nil {
@@ -915,13 +932,23 @@ func (b *moduleBus) State(verb string, params json.RawMessage) (json.RawMessage,
 		answer = keys
 	case "create":
 		// Only when the key has no value: of two writers making it, one wins (novox/hq ADR 0259).
-		revision, err := conn.StateCreate(b.module, asked.State, asked.Key, asked.Value)
+		create := conn.StateCreate
+		if asked.TTLSeconds != 0 {
+			create = func(module, name, key string, value json.RawMessage) (uint64, error) {
+				return conn.StateCreateTTL(module, name, key, value, time.Duration(asked.TTLSeconds)*time.Second)
+			}
+		}
+		revision, err := create(b.module, asked.State, asked.Key, asked.Value)
 		if err != nil {
 			return nil, err
 		}
 		answer = map[string]any{"revision": revision}
 	case "update":
 		// Only when the key is still at the revision read: a second answer to one ask loses (ADR 0259).
+		if asked.TTLSeconds != 0 {
+			return nil, fmt.Errorf("mesh/state.update gives no time to live: the bus sets one only when a key is " +
+				"created; put the key with ttl_seconds to renew it whole (issue 1952)")
+		}
 		revision, err := conn.StateUpdate(b.module, asked.State, asked.Key, asked.Value, asked.Revision)
 		if err != nil {
 			return nil, err
@@ -942,9 +969,24 @@ type logAsked struct {
 	From   uint64          `json:"from"`
 	Keys   string          `json:"keys"`
 	Limit  int             `json:"limit"`
+	// Since is a read's or a watch's start as an RFC 3339 time, in place of From (novox/hq ADR 0571).
+	Since string `json:"since"`
 }
 
-// Log answers a bundle's `append`, `read`, `last` and `info` on its module's log (ADR 0297). The runtime
+// since is the asked start time, zero when none.
+func (a logAsked) since() (time.Time, error) {
+	if a.Since == "" {
+		return time.Time{}, nil
+	}
+	at, err := time.Parse(time.RFC3339Nano, a.Since)
+	if err != nil {
+		return time.Time{}, fmt.Errorf("since is an RFC 3339 time (2026-10-29T09:30:00Z), not %q", a.Since)
+	}
+	return at, nil
+}
+
+// Log answers a bundle's `append`, `read`, `newest`, `last` and `info` on its module's log and on a log its module reads
+// as `<module>.<log>` (ADR 0297, ADR 0571). The runtime
 // refuses, with the reason, a log the module was not issued; an append whose `expect` is stale is
 // refused with the code of a lost state update.
 func (b *moduleBus) Log(verb string, params json.RawMessage) (json.RawMessage, error) {
@@ -962,11 +1004,22 @@ func (b *moduleBus) Log(verb string, params json.RawMessage) (json.RawMessage, e
 		}
 		answer = map[string]any{"sequence": seq}
 	case "read":
-		page, err := conn.LogRead(b.module, asked.Log, asked.From, asked.Keys, asked.Limit)
+		since, err := asked.since()
+		if err != nil {
+			return nil, err
+		}
+		page, err := conn.LogRead(b.module, asked.Log, asked.From, since, asked.Keys, asked.Limit)
 		if err != nil {
 			return nil, err
 		}
 		answer = page
+	case "newest":
+		// The newest entry of each key the pattern names (mesh-sdk process.LogReader.Newest).
+		entries, err := conn.LogNewest(b.module, asked.Log, asked.Keys)
+		if err != nil {
+			return nil, err
+		}
+		answer = entries
 	case "last":
 		entry, err := conn.LogLast(b.module, asked.Log, asked.Key)
 		if err != nil {
@@ -988,24 +1041,24 @@ func (b *moduleBus) Log(verb string, params json.RawMessage) (json.RawMessage, e
 	return json.Marshal(answer)
 }
 
-// StreamRead answers a bundle's read of one of its module's own streams (ADR 0363): {stream, subject, from, limit}, the
+// LogWatch hands a bundle each entry of a log, `{log, keys, from, since}`: from a sequence, from a time, or with neither
-// stream by its local name, the subject an event pattern. The runtime refuses, with the reason, a stream the module's
+// each new one, one at a time (novox/hq ADR 0571; mesh-sdk process.LogReader.Watch).
-// membership does not name.
+func (b *moduleBus) LogWatch(params json.RawMessage, deliver func(json.RawMessage) error) (func(), error) {
-func (b *moduleBus) StreamRead(params json.RawMessage) (json.RawMessage, error) {
+	var asked logAsked
-	var asked struct {
+	if err := json.Unmarshal(params, &asked); err != nil || asked.Log == "" {
-		Stream  string `json:"stream"`
+		return nil, fmt.Errorf("mesh/log.watch names no log: {log, keys, from, since}")
-		Subject string `json:"subject"`
-		From    uint64 `json:"from"`
-		Limit   int    `json:"limit"`
 	}
-	if err := json.Unmarshal(params, &asked); err != nil || asked.Stream == "" {
+	since, err := asked.since()
-		return nil, fmt.Errorf("mesh/stream.read names no stream: {stream, subject, from, limit}")
-	}
-	page, err := b.all.conn.StreamRead(b.module, asked.Stream, asked.Subject, asked.From, asked.Limit)
 	if err != nil {
 		return nil, err
 	}
-	return json.Marshal(page)
+	return b.all.conn.LogWatch(b.module, asked.Log, asked.Keys, asked.From, since, func(e bus.LogEntry) error {
+		raw, err := json.Marshal(e)
+		if err != nil {
+			return err
+		}
+		return deliver(raw)
+	})
 }
 
 // Watch hands a bundle its module's state as it is and as it changes (ADR 0201): `{state, key}`, the

node-tools/internal/runtime/state_counter_test.go

@@ -4,8 +4,10 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"strings"
 	"sync"
 	"testing"
+	"time"
 
 	"github.com/novox/mesh-tools/node-tools/internal/bus"
 	"github.com/novox/mesh-tools/node-tools/internal/launch"
@@ -18,8 +20,10 @@ func counterBus(t *testing.T) launch.Bus {
 	t.Helper()
 	mesh := mt.New(t)
 	mesh.Bucket(t, "issues_numbers")
+	mesh.BucketKeyTTL(t, "issues_wakes")
 	m := mt.MembershipOf("issues", "anchor", false, nil)
-	m.State = []bus.StateIssued{{Name: "numbers", Bucket: "issues_numbers", Writes: true}}
+	m.State = []bus.StateIssued{{Name: "numbers", Bucket: "issues_numbers", Writes: true},
+		{Name: "wakes", Bucket: "issues_wakes", Writes: true}}
 	mesh.Issue(t, m)
 	conn := connect(t, "node-tools", "anchor")
 	conn.Follow("issues")
@@ -131,6 +135,9 @@ func TestTwoWritersIncrementingOneCounterOnTheBusLoseNoUpdate(t *testing.T) {
 // A write at a revision that is no longer the key's is refused with the lost-write code and writes
 // nothing; create is refused for a key with a value and taken for one deleted; an update names the
 // revision it read, never zero.
+// A key's own time to live (issue 1952, mesh-sdk process.State.CreateTTL and RenewTTL): created once, renewed whole by a
+// put, its end heard by a watch as an expiry; refused, writing nothing, on a bucket that takes none, under a second, and
+// on an update.
 func TestAStaleWriteIsRefusedAndWritesNothing(t *testing.T) {
 	b := counterBus(t)
 	state := func(verb string, extra map[string]any) (uint64, error) {
@@ -200,4 +207,63 @@ func TestAStaleWriteIsRefusedAndWritesNothing(t *testing.T) {
 	if v := value(); v != "1" {
 		t.Fatalf("after delete and create: %s", v)
 	}
+
+	// A time to live per key: refused where it cannot be kept, writing nothing.
+	for _, c := range []struct {
+		verb, state, why string
+		ttl              int
+	}{
+		{"create", "numbers", "does not take keys with their own time to live", 60},
+		{"put", "numbers", "does not take keys with their own time to live", 60},
+		{"create", "wakes", "shorter than the bus keeps", -1},
+		{"update", "wakes", "gives no time to live", 60},
+	} {
+		_, err := ask(b, c.verb, map[string]any{"state": c.state, "key": "late", "value": 1, "revision": 1, "ttl_seconds": c.ttl})
+		if err == nil || !strings.Contains(err.Error(), c.why) {
+			t.Errorf("%s with ttl %d on %s: want refused for %q, got %v", c.verb, c.ttl, c.state, c.why, err)
+		}
+		if raw, _ := ask(b, "get", map[string]any{"state": c.state, "key": "late"}); string(raw) != "null" {
+			t.Errorf("%s with ttl %d on %s wrote %s", c.verb, c.ttl, c.state, raw)
+		}
+	}
+
+	// The bucket's fill against its caps, for a bundle's tools.
+	if raw, err := ask(b, "info", map[string]any{"state": "wakes"}); err != nil || !strings.Contains(string(raw),
+		`"bucket":"issues_wakes","bytes":0,"keys":0,"max_bytes":-1,"max_value_bytes":262144,"history":1,"ttl_seconds":0,"key_ttl":true`) {
+		t.Errorf("info of a key-ttl bucket answered %s, %v", raw, err)
+	}
+
+	// Created once, renewed whole, and its end heard: a wake with no timer.
+	heard := make(chan bus.StateChange, 8)
+	stop, err := b.Watch(json.RawMessage(`{"state":"wakes"}`), func(raw json.RawMessage) error {
+		var c bus.StateChange
+		_ = json.Unmarshal(raw, &c)
+		heard <- c
+		return nil
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer stop()
+	if _, err := ask(b, "create", map[string]any{"state": "wakes", "key": "settle", "value": 1, "ttl_seconds": 60}); err != nil {
+		t.Fatalf("create with a time to live: %v", err)
+	}
+	if _, err := ask(b, "create", map[string]any{"state": "wakes", "key": "settle", "value": 2, "ttl_seconds": 60}); !isLost(err) {
+		t.Fatalf("a second create with a time to live: %v", err)
+	}
+	if _, err := ask(b, "put", map[string]any{"state": "wakes", "key": "settle", "value": 3, "ttl_seconds": 1}); err != nil {
+		t.Fatalf("a put renewing the time to live: %v", err)
+	}
+	var ops []string
+	for want := "put delete put expire"; strings.Join(ops, " ") != want; {
+		select {
+		case c := <-heard:
+			ops = append(ops, c.Op)
+			if c.Op == "put" && len(ops) == 3 && string(c.Value) != "3" {
+				t.Fatalf("the renewed key holds %s", c.Value)
+			}
+		case <-time.After(5 * time.Second):
+			t.Fatalf("the watch heard %q, want %q", strings.Join(ops, " "), want)
+		}
+	}
 }

node-tools/module.json

@@ -9,13 +9,16 @@
     "mesh-knowledge.reviewed",
     "mesh-delivery-gate.transition",
     "gitea.forge.status",
-    "mesh-controller.checked",
+    "mesh-self-check.checked",
     "operator-channel.decided.claude-code"
   ],
   "provides": [
     {
       "name": "mcp-endpoint",
       "scope": "node",
+      "credential": {
+        "form": "none"
+      },
       "answered-by": "helper"
     }
   ],
@@ -65,6 +68,11 @@
     "bound"
   ],
   "reads": [
-    "mesh-controller.assignments"
+    "mesh-provisioner.provisions",
+    "mesh-provisioner.plans",
+    "mesh-delivery-assigner.seat-holders",
+    "mesh-delivery-assigner.placement",
+    "mesh-delivery-indexer.modules",
+    "mesh-delivery-indexer.manifests"
   ]
 }

node-tools/src/client.ts

@@ -26,9 +26,28 @@ import { TOOLS_VERB, type ToolsAnswer } from "./runtime.js";
 /** Where the catalogue answers which modules the mesh holds. */
 const CATALOGUE_MODULES = "mesh-catalog.catalog_modules";
 
-/** Where the mesh answers every role's tools, from its records: the mesh-controller seat's own
+/** Where every role's tools are answered: the console on this machine, the Go runtime, which holds mesh-sdk's seat
- *  `tools` verb (novox/hq ADR 0154, design 33 §5). */
+ *  definitions and reads who holds each seat from placement (novox/hq ADR 0571 clauses 4 and 5). Asked over the loopback
-const SEAT_TOOLS = "seat:mesh-controller.tools";
+ *  channel `mesh --console` already uses, by the method `mesh/seats` beside MCP's; never the controller. */
+export const SEATS_METHOD = "mesh/seats";
+const DEFAULT_CONSOLE = "http://127.0.0.1:4270";
+
+type SeatsAnswer = { seats?: { seat: string; scope?: string; tools?: ToolsAnswer["tools"] }[]; holders_why?: string };
+
+/** Every role's tools from the console on this machine. */
+async function seatsFromConsole(consoleUrl: string): Promise<SeatsAnswer> {
+  const endpoint = consoleUrl.endsWith("/mcp") ? consoleUrl : `${consoleUrl.replace(/\/$/, "")}/mcp`;
+  const res = await fetch(endpoint, {
+    method: "POST",
+    headers: { "content-type": "application/json", accept: "application/json" },
+    body: JSON.stringify({ jsonrpc: "2.0", id: 1, method: SEATS_METHOD }),
+    signal: AbortSignal.timeout(5_000),
+  });
+  if (!res.ok) throw new Error(`the console at ${endpoint} answered ${res.status}`);
+  const reply = (await res.json()) as { result?: SeatsAnswer; error?: { message: string } };
+  if (reply.error) throw new Error(reply.error.message);
+  return reply.result ?? {};
+}
 
 /** A tool as its module describes it. */
 export interface Tool {
@@ -58,8 +77,8 @@ export interface Tool {
 export interface Listing {
   tools: Tool[];
   /** Modules the catalogue holds whose runtime did not answer `tools`: not assigned, not up, or built
-   *  before the runtime answered it. Each may still be called by name. The mesh's own records are
+   *  before the runtime answered it. Each may still be called by name. The roles' tools are listed here, with the
-   *  listed here as `mesh-controller (seat)` when the control plane did not answer. */
+   *  reason, when the console on this machine did not answer them. */
   notAnswering: string[];
 }
 
@@ -159,17 +178,16 @@ export async function connectAsTheConsole(path: string): Promise<{ bus: Broker;
  * list that goes stale the first time a module is assigned. Every module is asked at once, and the bus
  * refuses at once a request nothing serves, so the cost is bounded by the modules that are up.
  */
-export async function toolsOn(bus: Broker): Promise<Listing> {
+export async function toolsOn(bus: Broker, consoleUrl: string = process.env.MESH_CONSOLE || DEFAULT_CONSOLE): Promise<Listing> {
+  let rolesWhy = "";
   const [answered, roles] = await Promise.all([
     bus.request<Record<string, never>, { modules?: { module: string }[] }>(CATALOGUE_MODULES, {}),
-    // The roles' tools, from the mesh's records (design 33 §5). Asked beside the modules rather
+    // The roles' tools, from the console on this machine (ADR 0571). Asked beside the modules rather than first: a
-    // than first: a control plane that is restarting must not hide every module's tools with it.
+    // console that is restarting must not hide every module's tools with it.
-    bus
+    seatsFromConsole(consoleUrl).catch((e: unknown) => {
-      .request<Record<string, never>, { seats?: { seat: string; scope?: string; tools?: ToolsAnswer["tools"] }[] }>(
+      rolesWhy = e instanceof Error ? e.message : String(e);
-        SEAT_TOOLS,
+      return undefined;
-        {},
+    }),
-      )
-      .catch(() => undefined),
   ]);
   const names = (answered.modules ?? []).map((m) => m.module).filter((m) => typeof m === "string");
 
@@ -188,7 +206,7 @@ export async function toolsOn(bus: Broker): Promise<Listing> {
       }
     }
   } else {
-    notAnswering.push("mesh-controller (seat)");
+    notAnswering.push(`the roles' tools (the console at ${consoleUrl} did not answer ${SEATS_METHOD}: ${rolesWhy})`);
   }
   asked.forEach((outcome, i) => {
     const module = names[i]!;

node-tools/test/client.test.ts

@@ -16,6 +16,8 @@ import { test } from "node:test";
 import { connectNats } from "../dist/broker-nats.js";
 import { callTool, seatsIn, toolKey, toolsOn, whyItFailed } from "../dist/client.js";
 
+import { aConsoleWithSeats } from "./console-seats.ts";
+
 const url = process.env.MESH_TEST_NATS;
 
 /** A mesh as discovery sees it (design 34 §3): the catalogue holding three modules, two of them up
@@ -36,21 +38,26 @@ async function aMeshWithTools() {
     tools: [{ name: "price", description: "what something costs", input: { type: "object" } }],
   }));
   await shop.handle("price", async (body: { of?: string }) => ({ of: body.of ?? "nothing", cost: 12 }));
-  // And the mesh's own records, served by the holder of the mesh-controller seat (ADR 0154): one
+  // Every role's tools, from the console on this machine (ADR 0571), the method beside MCP's; the controller's
-  // role's tool, and the verb that lists every role's.
+  // verb answers a call of its seat and is never asked for the roles.
   const controller = await connectNats({ url: url!, module: "mesh-controller" });
-  await controller.handle("seat:mesh-controller.tools", async () => ({
+  let controllerToolsAsked = 0;
-    seats: [
+  await controller.handle("seat:mesh-controller.tools", async () => {
-      { seat: "mesh-controller", scope: "mesh", tools: [{ name: "status", description: "what is wrong", input: {} }] },
+    controllerToolsAsked++;
-      { seat: "node-dns-resolver", scope: "node", tools: [{ name: "lookup", description: "one machine's", input: {} }] },
+    return { seats: [] };
-    ],
+  });
-  }));
   await controller.handle("seat:mesh-controller.status", async () => ({ output: "all quiet", ok: true }));
+  const local = await aConsoleWithSeats([
+    { seat: "mesh-controller", scope: "mesh", tools: [{ name: "status", description: "what is wrong", input: {} }] },
+    { seat: "node-dns-resolver", scope: "node", tools: [{ name: "lookup", description: "one machine's", input: {} }] },
+  ]);
   return {
     async close() {
       await catalogue.close();
       await shop.close();
       await controller.close();
+      await local.close();
+      assert.equal(controllerToolsAsked, 0, "the controller was asked for the roles' tools");
     },
   };
 }

node-tools/test/console-seats.ts

@@ -0,0 +1,31 @@
+/**
+ * The console on this machine as a client asks it for every role's tools (novox/hq ADR 0571): the method `mesh/seats`
+ * beside MCP's, answered in the shape the Go console's `seatsAnswer` gives (internal/console/mcp.go). Sets
+ * MESH_CONSOLE to it, so a client, and a `mesh` it spawns, asks it.
+ */
+import { createServer } from "node:http";
+import type { AddressInfo } from "node:net";
+
+import { SEATS_METHOD } from "../dist/client.js";
+
+export type SeatOfConsole = { seat: string; scope: string; tools: { name: string; description?: string; input?: unknown }[] };
+
+export async function aConsoleWithSeats(seats: SeatOfConsole[]): Promise<{ url: string; close: () => Promise<void> }> {
+  const local = createServer((req, res) => {
+    let body = "";
+    req.on("data", (c) => (body += c));
+    req.on("end", () => {
+      const asked = JSON.parse(body) as { id: number; method: string };
+      res.setHeader("content-type", "application/json");
+      if (req.url !== "/mcp" || asked.method !== SEATS_METHOD) {
+        res.end(JSON.stringify({ jsonrpc: "2.0", id: asked.id, error: { code: -32601, message: "no such method" } }));
+        return;
+      }
+      res.end(JSON.stringify({ jsonrpc: "2.0", id: asked.id, result: { seats: seats.map((s) => ({ ...s, holders: [] })) } }));
+    });
+  });
+  await new Promise<void>((done) => local.listen(0, "127.0.0.1", done));
+  const url = `http://127.0.0.1:${(local.address() as AddressInfo).port}`;
+  process.env.MESH_CONSOLE = url;
+  return { url, close: () => new Promise((done) => local.close(() => done())) };
+}

node-tools/test/mcp.test.ts

@@ -12,6 +12,8 @@ import assert from "node:assert/strict";
 import { test } from "node:test";
 import { spawn } from "node:child_process";
 
+
+import { aConsoleWithSeats } from "./console-seats.ts";
 import { connectNats } from "../dist/broker-nats.js";
 
 const url = process.env.MESH_TEST_NATS;
@@ -29,16 +31,15 @@ async function aMeshAndACredential(t: { after: (fn: () => Promise<void> | void)
   }));
   await shop.handle("price", async (body: { of?: string }) => ({ of: body.of ?? "nothing", cost: 12 }));
   const controller = await connectNats({ url: url!, module: "mesh-controller" });
-  await controller.handle("seat:mesh-controller.tools", async () => ({
+  // Every role's tools, from the console on this machine (ADR 0571), which the spawned `mesh` asks through MESH_CONSOLE.
-    seats: [
+  const local = await aConsoleWithSeats([
-      { seat: "mesh-controller", scope: "mesh", tools: [
+    { seat: "mesh-controller", scope: "mesh", tools: [
-        { name: "status", description: "what is wrong", input: {} },
+      { name: "status", description: "what is wrong", input: {} },
-        { name: "push", description: "tell a machine", input: { node: { type: "string" } } },
+      { name: "push", description: "tell a machine", input: { node: { type: "string" } } },
-      ] },
+    ] },
-      // A seat held once per machine (design 33 §4, ADR 0170): its verb is asked of one.
+    // A seat held once per machine (design 33 §4, ADR 0170): its verb is asked of one.
-      { seat: "node-dns-resolver", scope: "node", tools: [{ name: "lookup", description: "one machine's", input: {} }] },
+    { seat: "node-dns-resolver", scope: "node", tools: [{ name: "lookup", description: "one machine's", input: {} }] },
-    ],
+  ]);
-  }));
   await controller.handle("seat:node-dns-resolver.lookup@anchor", async () => ({ machine: "anchor", answered: true }));
   await controller.handle("seat:mesh-controller.status", async () => ({ output: "all quiet", ok: true }));
   await controller.handle("seat:mesh-controller.push", async (body: { node?: string }) => ({ told: body.node ?? "nobody" }));
@@ -46,6 +47,7 @@ async function aMeshAndACredential(t: { after: (fn: () => Promise<void> | void)
     await catalogue.close();
     await shop.close();
     await controller.close();
+    await local.close();
   });
 
   const { mkdtemp, writeFile } = await import("node:fs/promises");

node-tools/test/node-runtime.test.ts

@@ -172,9 +172,12 @@ test("the node's runtime serves five modules' bundles on one credential — two
     const catalogue = await connectNats({ url, module: "mesh-catalog" });
     await catalogue.handle("catalog_modules", async () => ({ modules: [{ module: "alpha" }, { module: "beta" }, { module: "gamma" }, { module: "delta" }, { module: "epsilon" }] }));
     try {
-      const have = await toolsOn(asker);
+      // No console on this port: the roles' tools are said missing with the reason, never asked of the controller.
+      const have = await toolsOn(asker, "http://127.0.0.1:9");
       assert.deepEqual(have.tools.map((x) => `${x.module}.${x.name}`), ["alpha.one", "alpha.two", "beta.five", "beta.four", "beta.three", "delta.die", "delta.greet", "epsilon.seven"]);
-      assert.deepEqual(have.notAnswering, ["gamma (its tools bundle failed to load: gamma's bundle exited (1): Error: gamma's bundle cannot find its client)", "mesh-controller (seat)"]);
+      assert.equal(have.notAnswering[0], "gamma (its tools bundle failed to load: gamma's bundle exited (1): Error: gamma's bundle cannot find its client)");
+      assert.match(have.notAnswering[1] ?? "", /^the roles' tools \(the console at http:\/\/127\.0\.0\.1:9 did not answer mesh\/seats: /);
+      assert.equal(have.notAnswering.length, 2);
     } finally {
       await catalogue.close();
     }