mesh-grants: take each user's hash from the vault's bus-accounts and say per user whether one is held, so take-over waits until every user has a credential

ADR 0547 clause 9: the vault writes the hash of each bus account it holds into bus-accounts.<user>. The grants part
watches that state (no timer), carries the hash with each shadow user in memory only, and its user-list tool says per
user whether a hash is held and how its line would differ from the loaded one (same, adds, left out for want of a
credential, loses its login); never the hash, since bcrypt salts make two hashes incomparable. Take-over names the
users with no hash.

Issue: 1945

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

@@ -182,6 +182,42 @@ func TestTheAssignerDoesNotStartWithoutABucketItReads(t *testing.T) {
 			t.Errorf("without %s the assigner started", missing)
 		}
 	}
+	// ADR 0546: without the puts' bucket it starts all the same, and once the bucket is made (by the controller at its start,
+	// ADR 0201) a pass opens it, so a put is taken further without a restart.
+	was := PutsReopenEvery
+	PutsReopenEvery = 0
+	defer func() { PutsReopenEvery = was }()
+	b := startBed(t, delivery.AssignerPutsBucket)
+	done := b.start(t)
+	select {
+	case err := <-done:
+		t.Fatalf("without %s the assigner stopped: %v", delivery.AssignerPutsBucket, err)
+	case <-time.After(2 * time.Second):
+	}
+	puts, err := b.js.CreateKeyValue(b.ctx, jetstream.KeyValueConfig{Bucket: delivery.AssignerPutsBucket, History: 5})
+	if err != nil {
+		t.Fatal(err)
+	}
+	pkey, _ := delivery.PutKey("n1", "web")
+	asked := delivery.Put{FP: delivery.SourcePrefix2 + strings.Repeat("b", 64), Why: "a try", At: time.Now()}.Said(time.Now(), delivery.PutAsked, "asked")
+	raw, _ := delivery.EncodePut(asked)
+	if _, err := puts.Put(b.ctx, pkey, raw); err != nil {
+		t.Fatal(err)
+	}
+	for deadline := time.Now().Add(15 * time.Second); ; time.Sleep(200 * time.Millisecond) {
+		// A change of a bucket it watches starts a pass, which opens the puts' bucket.
+		if _, err := b.buckets[delivery.PresenceBucket].Put(b.ctx, "n1", []byte(time.Now().String())); err != nil {
+			t.Fatal(err)
+		}
+		if e, err := puts.Get(b.ctx, pkey); err == nil {
+			if got, err := delivery.DecodePut(e.Value()); err == nil && got.Step == delivery.PutRefused && strings.Contains(got.Words, "not assigned") {
+				break
+			}
+		}
+		if time.Now().After(deadline) {
+			t.Fatal("the put was not taken further once its bucket was made")
+		}
+	}
 }
 
 // putJSON writes a value into one of the bed's buckets.

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

@@ -16,6 +16,7 @@ import (
 	"os/signal"
 	"strings"
 	"syscall"
+	"time"
 
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
@@ -85,19 +86,20 @@ var keptStates = []announce.State{
 // buckets is every bucket the assigner opened, by name.
 type buckets map[string]jetstream.KeyValue
 
-// open opens every bucket the assigner reads and writes; a bucket it cannot open stops the start, named, since a pass
-// without it would write wants from half the mesh.
-func open(ctx context.Context, js jetstream.JetStream) (buckets, error) {
+// open opens every bucket the assigner reads and writes; a bucket it cannot open stops the start, named with its grant (issue 1945),
+// since a pass without it would write wants from half the mesh. The puts' bucket is the exception: whatever keeps it from opening (it
+// does not exist yet, the credential is not granted it yet, the bus does not answer in time), the assigner writes wants
+// without it and takes no put further until it starts again (ADR 0546).
+func open(ctx context.Context, js process.BucketOpener, logf func(string, ...any)) (buckets, error) {
 	out := buckets{}
 	for _, name := range append(append([]string{}, readBuckets...), ownBuckets...) {
 		kv, err := js.KeyValue(ctx, name)
-		if name == delivery.AssignerPutsBucket && errors.Is(err, jetstream.ErrBucketNotFound) {
-			// The puts (ADR 0546) are taken further only once their bucket exists: an assigner never stops writing wants
-			// for want of it. The verb's first put is refused by the runtime until then.
+		if name == delivery.AssignerPutsBucket && err != nil {
+			logf("the bucket %s cannot be opened (%v): no put is taken further until the assigner starts again", name, err)
 			continue
 		}
 		if err != nil {
-			return nil, fmt.Errorf("the bucket %s cannot be opened: %w", name, err)
+			return nil, process.EssentialErr("assigner", name, "write a want from the whole mesh", err)
 		}
 		out[name] = kv
 	}
@@ -134,13 +136,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	if err != nil {
 		return err
 	}
-	kvs, err := open(ctx, js)
+	kvs, err := open(ctx, js, logger.Printf)
 	if err != nil {
 		return err
 	}
-	if kvs[delivery.AssignerPutsBucket] == nil {
-		logger.Printf("the bucket %s does not exist yet, so no put is taken further until the assigner starts again", delivery.AssignerPutsBucket)
-	}
 	// What the assigner keeps, said on the bus (ADR 0495 §2-§3): a reader is granted from it. Not fatal when it cannot
 	// be said now: process.Conn says it again at every reconnect.
 	said, err := process.Connect(account, nats.Name(account.Node+"."+account.Module+".keeps"), nats.RetryOnFailedConnect(true))
@@ -196,6 +195,8 @@ func follow(ctx context.Context, js jetstream.JetStream, kvs buckets, say func(a
 		if err != nil {
 			return fmt.Errorf("the bucket %s cannot be watched: %w", name, err)
 		}
+		// The puts' watch never stops the wants (ADR 0546): when it ends, the puts are no longer heard, said in the journal.
+		optional := name == delivery.AssignerPutsBucket
 		go func() {
 			defer kw.Stop()
 			for {
@@ -205,11 +206,17 @@ func follow(ctx context.Context, js jetstream.JetStream, kvs buckets, say func(a
 					return
 				case e, ok := <-kw.Updates():
 					if !ok {
+						if optional {
+							logger.Printf("the watch of %s ended: no put is taken further until the assigner starts again", name)
+							return
+						}
 						errs <- fmt.Errorf("the watch of %s ended", name)
 						return
 					}
 					if e == nil {
-						settled <- struct{}{}
+						if !optional { // the puts never gate the first pass, so never the wants (ADR 0546)
+							settled <- struct{}{}
+						}
 						continue
 					}
 					var value []byte
@@ -223,11 +230,40 @@ func follow(ctx context.Context, js jetstream.JetStream, kvs buckets, say func(a
 		}()
 		return nil
 	}
-	followed := watched
+	// The puts' watch is not waited for before the first pass: a watch of them that ends before it handed what stands only
+	// stops the puts, never the wants.
 	if kvs[delivery.AssignerPutsBucket] != nil {
-		followed = append(append([]string{}, watched...), delivery.AssignerPutsBucket)
+		if err := watch(delivery.AssignerPutsBucket); err != nil {
+			logger.Printf("%v: no put is taken further until it is opened again", err)
+			delete(kvs, delivery.AssignerPutsBucket)
+		}
 	}
-	for _, name := range followed {
+	// The puts' bucket is made by the controller from the manifest's states at its start (ADR 0201), and the assigner's
+	// credential reads it once the bus's user list carries the grant: neither waits for the assigner. So while it is not open,
+	// a pass started by a change opens it again, at most once a minute, and from then on the puts are watched and taken
+	// further, without a restart.
+	var triedPuts time.Time
+	reopenPuts := func(p *passer) {
+		if kvs[delivery.AssignerPutsBucket] != nil || time.Since(triedPuts) < PutsReopenEvery {
+			return
+		}
+		triedPuts = time.Now()
+		kv, err := js.KeyValue(ctx, delivery.AssignerPutsBucket)
+		if err != nil {
+			return
+		}
+		kvs[delivery.AssignerPutsBucket] = kv
+		if err := watch(delivery.AssignerPutsBucket); err != nil {
+			logger.Printf("%v: no put is taken further until it is opened again", err)
+			delete(kvs, delivery.AssignerPutsBucket)
+			return
+		}
+		if bus != nil {
+			p.acts = newPutActs(bus, wake, logger.Printf)
+		}
+		logger.Printf("the bucket %s is open: puts are taken further", delivery.AssignerPutsBucket)
+	}
+	for _, name := range watched {
 		var opts []jetstream.WatchOpt
 		if name == delivery.HaveBucket {
 			opts = append(opts, jetstream.IncludeHistory()) // what each node ran healthy before, for a set-back
@@ -239,7 +275,7 @@ func follow(ctx context.Context, js jetstream.JetStream, kvs buckets, say func(a
 	if err := watch(delivery.ManifestsBucket, jetstream.MetaOnly()); err != nil {
 		return err
 	}
-	for range len(followed) + 1 {
+	for range len(watched) + 1 {
 		select {
 		case <-ctx.Done():
 			return nil
@@ -262,11 +298,15 @@ func follow(ctx context.Context, js jetstream.JetStream, kvs buckets, say func(a
 			return err
 		case <-settled:
 		case <-dirty:
+			reopenPuts(p)
 			p.pass(ctx)
 		}
 	}
 }
 
+// PutsReopenEvery is how often, at most, a pass opens the puts' bucket again while it is not open.
+var PutsReopenEvery = time.Minute
+
 // sayVia publishes a condition on the operator-channel seat's one subject for conditions, under the assigner's own name
 // (ADR 0437 clause 9).
 func sayVia(ctx context.Context, js jetstream.JetStream) func(asks.Condition) error {

modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go

@@ -2,15 +2,23 @@ package main
 
 import (
 	"context"
+	"errors"
+	"fmt"
 	"io"
+	"log"
 	"os"
 	"path/filepath"
 	"reflect"
 	"slices"
 	"strings"
 	"testing"
+	"time"
 
+	"github.com/nats-io/nats.go/jetstream"
+
+	"git.novox.be/novox/mesh-sdk/go/asks"
 	"git.novox.be/novox/mesh-sdk/go/catalogue"
+	"git.novox.be/novox/mesh-sdk/go/delivery"
 )
 
 // 1297/B2a: the buckets the program opens are exactly the manifest's reads and states, as the broker names them
@@ -48,6 +56,53 @@ func TestOpensWhatItsManifestNames(t *testing.T) {
 	if got, want := slices.Sorted(slices.Values(kept)), slices.Sorted(slices.Values(ownBuckets)); !reflect.DeepEqual(got, want) {
 		t.Errorf("the program announces %v, keeps %v", got, want)
 	}
+	// ADR 0546: whatever keeps the puts' bucket from opening (not there, not granted, no answer in time), the assigner opens
+	// the rest and starts, writing wants; any other bucket it cannot open stops the start, named.
+	for _, c := range []struct {
+		missing string
+		err     error
+		starts  bool
+	}{
+		{delivery.AssignerPutsBucket, jetstream.ErrBucketNotFound, true},
+		{delivery.AssignerPutsBucket, errors.New("nats: permissions violation for subscription to \"$JS.API.STREAM.INFO.KV_mesh-delivery-assigner_puts\""), true},
+		{delivery.AssignerPutsBucket, context.DeadlineExceeded, true},
+		{delivery.AssignerWantsBucket, errors.New("nats: permissions violation"), false},
+	} {
+		var logged string
+		kvs, err := open(context.Background(), refusingJS{missing: c.missing, err: c.err}, func(f string, a ...any) { logged = fmt.Sprintf(f, a...) })
+		switch {
+		case c.starts && (err != nil || kvs[c.missing] != nil || len(kvs) != len(readBuckets)+len(ownBuckets)-1 || !strings.Contains(logged, "no put is taken further")):
+			t.Errorf("%s refused with %v: the start answered %v, %d buckets, logged %q", c.missing, c.err, err, len(kvs), logged)
+		case !c.starts && (err == nil || !strings.Contains(err.Error(), "grant of "+c.missing)): // 1945: named with its grant
+			t.Errorf("%s refused with %v: the start answered %v", c.missing, c.err, err)
+		}
+	}
+	// And a watch of the puts that ends before it handed what stands (its subscription closed, its grant withdrawn) does not
+	// keep the first pass from being made: the wants are written all the same.
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	wrote := make(chan string, 8)
+	kvs := buckets{}
+	for _, name := range append(append([]string{}, readBuckets...), ownBuckets...) {
+		kvs[name] = &watchKV{memKV: &memKV{values: map[string][]byte{}}}
+	}
+	ckey, _ := delivery.ComposedNodeKey("ace")
+	cd, err := delivery.EncodeComposedDigest(delivery.ComposedDigest{Digest: "sha256:" + strings.Repeat("d", 64), Judgement: delivery.JudgedUnchanged})
+	if err != nil {
+		t.Fatal(err)
+	}
+	kvs[delivery.ControllerComposedBucket].(*watchKV).entries = []kvEntry{{key: ckey, value: cd}}
+	kvs[delivery.AssignerPutsBucket].(*watchKV).closed = true
+	kvs[delivery.AssignerWantsBucket].(*watchKV).wrote = wrote
+	go follow(ctx, nil, kvs, func(asks.Condition) error { return nil }, nil, log.New(io.Discard, "", 0))
+	select {
+	case key := <-wrote:
+		if key != "want.ace.node" {
+			t.Errorf("the first pass wrote %s", key)
+		}
+	case <-ctx.Done():
+		t.Error("a puts watch that ended before its first marker kept the first pass from being made: no want was written")
+	}
 }
 
 // 1297/B2a: without a credential the program does not start, and says why.
@@ -57,3 +112,61 @@ func TestNoCredential(t *testing.T) {
 		t.Errorf("run without a credential = %v", err)
 	}
 }
+
+// refusingJS opens every bucket but one, which it refuses with err.
+type refusingJS struct {
+	jetstream.JetStream
+	missing string
+	err     error
+}
+
+func (r refusingJS) KeyValue(_ context.Context, name string) (jetstream.KeyValue, error) {
+	if name == r.missing {
+		return nil, r.err
+	}
+	return &memKV{values: map[string][]byte{}}, nil
+}
+
+// watchKV is a memKV whose watch hands its entries and then the marker that all stands was handed, or ends at once when
+// closed; a put on it is said on wrote.
+type watchKV struct {
+	*memKV
+	entries []kvEntry
+	closed  bool
+	wrote   chan string
+}
+
+func (w *watchKV) WatchAll(context.Context, ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) {
+	ch := make(chan jetstream.KeyValueEntry, len(w.entries)+1)
+	if w.closed {
+		close(ch)
+		return keyWatcher{ch}, nil
+	}
+	for _, e := range w.entries {
+		ch <- e
+	}
+	ch <- nil
+	return keyWatcher{ch}, nil
+}
+
+func (w *watchKV) Put(ctx context.Context, key string, value []byte) (uint64, error) {
+	if w.wrote != nil {
+		w.wrote <- key
+	}
+	return 1, nil
+}
+
+type keyWatcher struct{ ch chan jetstream.KeyValueEntry }
+
+func (k keyWatcher) Updates() <-chan jetstream.KeyValueEntry { return k.ch }
+func (k keyWatcher) Stop() error                             { return nil }
+
+type kvEntry struct {
+	jetstream.KeyValueEntry
+	key   string
+	value []byte
+}
+
+func (e kvEntry) Key() string                     { return e.key }
+func (e kvEntry) Value() []byte                   { return e.value }
+func (e kvEntry) Operation() jetstream.KeyValueOp { return jetstream.KeyValuePut }

modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go

@@ -3,6 +3,7 @@ package main
 import (
 	"context"
 	"errors"
+	"fmt"
 	"strings"
 	"time"
 
@@ -35,6 +36,9 @@ type passer struct {
 	neither map[string]neitherOne
 	// acts are the puts' acts (ADR 0546); nil: the puts are not taken further (no bus to call, or no puts bucket).
 	acts *putActs
+	// noWant is, by node and module, the reason last said why the node gets no want for the module (issue 1945): a reason is
+	// said in the journal when it appears or changes, and its clearing when the node has a want again; one line per change.
+	noWant map[string]string
 }
 
 func (p *passer) clock() time.Time {
@@ -148,6 +152,7 @@ func (p *passer) wants(ctx context.Context) {
 		}
 		out := assign.Wants(m, nodes, wanted[name], last)
 		gatherNeither(neither, name, out)
+		p.sayNoWant(name, m.Nodes, wanted[name], out)
 		if first {
 			startClears(stale, m, out)
 			if fp != "" && !m.Held && out.Held == nil && !out.Lifted {
@@ -186,6 +191,42 @@ func (p *passer) wants(ctx context.Context) {
 	}
 }
 
+// sayNoWant says in the journal why a node of module gets no want after this pass, from the rule the outcome names for it
+// (issue 1945: a new assignment whose trunk build was not done was held back with nothing said), once per reason, and says
+// when the node has a want again. A node with a live want, written before and kept or written in this pass, has one.
+func (p *passer) sayNoWant(module string, nodes []string, before map[string]delivery.Want, out assign.Outcome) {
+	if p.noWant == nil {
+		p.noWant = map[string]string{}
+	}
+	assigned := map[string]bool{}
+	for _, node := range nodes {
+		assigned[node] = true
+		key := node + "." + module
+		w, written := out.Wants[node]
+		if !written {
+			w, written = before[node]
+		}
+		reason := ""
+		if h, held := out.Rules[node]; held && (!written || w.Removed) {
+			reason = fmt.Sprintf("rule %d: %s", h.Rule, h.Words)
+		}
+		last, said := p.noWant[key]
+		switch {
+		case reason != "" && reason != last:
+			p.logf("%s on %s gets no want: %s", module, node, reason)
+			p.noWant[key] = reason
+		case reason == "" && said && written && !w.Removed:
+			p.logf("%s on %s has a want", module, node)
+			delete(p.noWant, key)
+		}
+	}
+	for key := range p.noWant {
+		if node, m, _ := strings.Cut(key, "."); m == module && !assigned[node] {
+			delete(p.noWant, key) // no longer assigned there: its removed want says so
+		}
+	}
+}
+
 // neitherOne is a module assigned to a node that holds no want for it after a pass, and the rule holding its first want back.
 type neitherOne struct {
 	node, module string

modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go

@@ -380,6 +380,31 @@ func TestThePassGathersTheModulesOnNeitherList(t *testing.T) {
 	if len(got) != 1 || got["g14.screen-lock"].node != "g14" || got["g14.screen-lock"].module != "screen-lock" || got["g14.screen-lock"].holding.Rule != delivery.RuleGrant {
 		t.Errorf("gathered = %+v", got)
 	}
+	// 1945 (a first want held back silently): the journal says why a node gets no want when the reason appears or changes, and
+	// when it has a want again: one line per change, not per pass.
+	p := newPasser(buckets{}, time.Date(2026, 10, 25, 21, 30, 0, 0, time.UTC))
+	var logged []string
+	p.logf = func(f string, a ...any) { logged = append(logged, fmt.Sprintf(f, a...)) }
+	notDone := assign.Outcome{Rules: map[string]assign.Holding{"shanks": {Rule: delivery.RuleBuildDone, Words: "the build of rofi at 4e7b118e is not done"}}}
+	wanted := assign.Outcome{Wants: map[string]delivery.Want{"shanks": {FP: fpTest, Composed: "sha256:" + strings.Repeat("c", 64)}}}
+	for i, c := range []struct {
+		before map[string]delivery.Want
+		out    assign.Outcome
+		line   string // the line said in this pass, "" for none
+	}{
+		{nil, notDone, "rofi on shanks gets no want: rule 1: the build of rofi at 4e7b118e is not done"},
+		{nil, notDone, ""},
+		{map[string]delivery.Want{"shanks": {Removed: true}}, notDone, ""},
+		{nil, assign.Outcome{Rules: map[string]assign.Holding{"shanks": {Rule: delivery.RuleGrant, Words: "no credential"}}}, "rofi on shanks gets no want: rule 6: no credential"},
+		{nil, wanted, "rofi on shanks has a want"},
+		{map[string]delivery.Want{"shanks": wanted.Wants["shanks"]}, assign.Outcome{}, ""},
+	} {
+		logged = nil
+		p.sayNoWant("rofi", []string{"shanks"}, c.before, c.out)
+		if got := strings.Join(logged, "\n"); got != c.line {
+			t.Errorf("pass %d said %q, want %q", i, got, c.line)
+		}
+	}
 }
 
 // A neither or held condition a previous process raised is cleared by the first pass after a start: a fresh pass remembers

modules/mesh-delivery-assigner/go.mod

@@ -3,7 +3,7 @@ module mesh-delivery-assigner
 go 1.26.0
 
 require (
-	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025203528-db26f7e8b6f5
+	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585
 	github.com/nats-io/nats-server/v2 v2.11.17
 	github.com/nats-io/nats.go v1.54.0
 )

modules/mesh-delivery-assigner/go.sum

@@ -6,6 +6,8 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025200810-d4ae4b6790b8 h1:vhYk/iv4
 git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025200810-d4ae4b6790b8/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
 git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025203528-db26f7e8b6f5 h1:gE80bR0tPME7MTMBLcdP62v+F/9soVXmdcBYNMOye44=
 git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025203528-db26f7e8b6f5/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585 h1:7QoFH5qjXRpx5M4uPdYsVJ5qKl2082gq8n0lFxhoNIg=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585/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=

modules/mesh-delivery-assigner/internal/assign/wants.go

@@ -226,7 +226,15 @@ func Wants(m Module, n Nodes, wants map[string]delivery.Want, lastHealthy map[st
 		}
 		before, had := wants[node]
 		var next delivery.Want
-		if p, put := m.Puts[node]; put {
+		p, put := m.Puts[node]
+		if !put && had && before.Put && before.Hold {
+			// The want is the truth (ADR 0546 clause 5): a want marked put and held stands while its put is not heard (the
+			// puts not open, or not yet handed after a start); only a put heard released gives it back to the rules. A put
+			// without hold is not kept this way: its put takes it further once heard.
+			p, put = delivery.Put{FP: before.FP, Composed: before.Composed, Hold: true, Why: strings.TrimPrefix(before.Words, "put: "),
+				Step: delivery.PutHolding}, true
+		}
+		if put {
 			if want, owns := PutOwns(p); owns {
 				if want != nil {
 					w := *want
@@ -325,8 +333,18 @@ func Wants(m Module, n Nodes, wants map[string]delivery.Want, lastHealthy map[st
 				// the pass says so. A new assignment, with no have, waits for the rules and is on no list yet by design. A held module (rule 2,
 				// held before or by this pass) is on the held list instead, whose hold is said and answered on its own: it is never on neither
 				// list (issue 1945: systemd-resolved, held on shanks by the operator, was said as urgent every hour).
-				if h, there := n.Haves[node][m.Name]; there && h.Phase != delivery.PhaseRemoved && !held {
+				h, there := n.Haves[node][m.Name]
+				switch {
+				case there && h.Phase != delivery.PhaseRemoved && !held:
 					out.Neither[node] = Holding{Rule: rule, Words: words}
+				case had && (!there || h.Phase == delivery.PhaseRemoved):
+					// A removed want of a node that does not run the module says why no want follows it (issue 1945), so the view's
+					// assigner_want answers it; it stays removed. A node with no key gets none: a removed want written where none was
+					// would have its node agent remove the module and stop holding its composition's resources back.
+					next = delivery.Want{Removed: true, Words: NoWantWords(rule, words)}
+					if !delivery.SameWant(normal(before), normal(next)) {
+						out.Wants[node] = next
+					}
 				}
 				continue
 			}
@@ -362,6 +380,11 @@ func Wants(m Module, n Nodes, wants map[string]delivery.Want, lastHealthy map[st
 	return out
 }
 
+// NoWantWords say why a node gets no want for a module: the rule that holds its first want back and why (issue 1945).
+func NoWantWords(rule int, words string) string {
+	return fmt.Sprintf("no want: rule %d: %s", rule, words)
+}
+
 // failed says whether a have's health check failed once the node is done moving: phase running or failed (never removed, held, or
 // moving). A failed health while it
 // backs up, applies or checks is a step of the move, not its outcome (on novox at 15:27:36 on 2026-10-25, "health failed, phase

modules/mesh-delivery-assigner/internal/assign/wants_test.go

@@ -397,10 +397,15 @@ func TestTheBusIsNotMovedButItsUserListRevisionIsWrittenAtOnce(t *testing.T) {
 		{"wanted", delivery.PutWanted, wantedOld(), &delivery.Want{FP: fNew, Composed: dig(fNew), Put: true, Hold: true, Words: "put: the bus's user list"}},
 		{"holding, written already", delivery.PutHolding, map[string]delivery.Want{"ace": PutWant(put)}, nil},
 		{"released: the rules again", delivery.PutReleased, map[string]delivery.Want{"ace": PutWant(put)}, &delivery.Want{FP: fNew, Composed: dig(fNew)}},
+		{"not heard: a held want at an older fingerprint stands", "", map[string]delivery.Want{"ace": {FP: fOld, Composed: dig(fOld), Put: true, Hold: true, Words: "put: back"}}, nil},
+		{"not heard: a held want the rules would rewrite stands", "", map[string]delivery.Want{"ace": PutWant(put)}, nil},
 	} {
 		p := put
 		p.Step = c.step
 		m.Puts = map[string]delivery.Put{"ace": p}
+		if c.step == "" {
+			m.Puts = nil
+		}
 		out := Wants(m, n, c.wanted, nil)
 		got, wrote := out.Wants["ace"]
 		switch {
@@ -862,6 +867,33 @@ func TestANewAssignmentWaitingIsNotOnNeitherList(t *testing.T) {
 	if out := Wants(m, n, nil, nil); len(out.Neither) != 1 || out.Neither["g14"].Rule != delivery.RuleBuildDone {
 		t.Errorf("with no build: neither = %+v, want g14 alone (it runs the module), by rule 1", out.Neither)
 	}
+	// 1945 (the assigner holds a first want back silently): with no build, a node whose want was removed and that does not run
+	// the module has the reason written into its removed want, once; a node with no key gets no want written, and a node that
+	// runs the module is on neither list, its removed want left as it is.
+	gone := delivery.Want{Removed: true, Words: "screen-lock is no longer assigned to shanks"}
+	for _, c := range []struct {
+		name   string
+		before map[string]delivery.Want
+		want   map[string]delivery.Want
+	}{
+		{"no key: nothing written", nil, map[string]delivery.Want{}},
+		{"a removed want: it says why", map[string]delivery.Want{"shanks": gone},
+			map[string]delivery.Want{"shanks": {Removed: true, Words: "no want: rule 1: the build of screen-lock at " + short(m.FP) + " is not done"}}},
+		{"a removed want that says why already: nothing written", map[string]delivery.Want{"shanks": {Removed: true, Words: "no want: rule 1: the build of screen-lock at " + short(m.FP) + " is not done"}},
+			map[string]delivery.Want{}},
+		{"a node that runs it: on neither list, its removed want left", map[string]delivery.Want{"g14": gone}, map[string]delivery.Want{}},
+	} {
+		out := Wants(m, n, c.before, nil)
+		if len(out.Wants) != len(c.want) {
+			t.Errorf("%s: wants = %+v, want %+v", c.name, out.Wants, c.want)
+			continue
+		}
+		for node, w := range c.want {
+			if !delivery.SameWant(normal(out.Wants[node]), w) {
+				t.Errorf("%s: want of %s = %+v, want %+v", c.name, node, out.Wants[node], w)
+			}
+		}
+	}
 	m.FP = ""
 	if out := Wants(m, n, nil, nil); len(out.Neither) != 0 {
 		t.Errorf("a module of the node's own composition named on neither list: %+v", out.Neither)

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

@@ -0,0 +1,178 @@
+// Command mesh-delivery-builder-tools serves the builder's tools (novox/hq issue 1945): the runtime launches it for a call.
+// builder_stuck reads the builder's builds and reasks states and the build modules' steps through the builder's state and
+// reads; builder_ask_again writes one value of the reasks state, which the builder watches and answers. The builder is the
+// other program: it serves no tool (ADR 0437 §4).
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"os"
+	"strings"
+	"sync"
+	"time"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+
+	"mesh-delivery-builder/internal/build"
+)
+
+// queueAsker writes a queues value and waits for the builder's answer to it.
+type queueAsker func(key string, value []byte) (json.RawMessage, error)
+
+// Tools are the builder's tools: stuck reads, askAgain writes the reasks value, queue writes a queues value and waits for
+// the builder's read.
+func Tools(r build.StuckReader, askAgain func(key string, value []byte) error, queue queueAsker, now func() time.Time) []stdio.Tool {
+	str := func(args map[string]any, k string) string { s, _ := args[k].(string); return strings.TrimSpace(s) }
+	return []stdio.Tool{
+		{Name: build.StuckTool,
+			Description: "Every build key that waits (building or queued), oldest first: its module, fingerprint, step words and since when, and each ask it " +
+				"waits on by language with its id, when it was put, its asking (1 for the first; the builder asks an ask again when it can no longer be " +
+				"answered, at most 3 times) and what its build module's steps hold of it (none, or its state on its node since its start). The " +
+				"seat workers' delivery counts are not readable by the builder. With module, that module's keys only. Read-only.",
+			Input: map[string]any{"type": "object", "properties": map[string]any{
+				"module": map[string]any{"type": "string", "description": "only this module's keys"}}},
+			Run: func(args map[string]any) (any, error) { return r.Stuck(str(args, "module")) }},
+		{Name: build.AskAgainTool,
+			Description: "Ask a waiting build key's asks again now, under new ids the key then names, as the builder does by itself for an ask its build " +
+				"seat's stream gave up on. The builder answers in the reasks state, which " + build.StuckTool + " shows beside the key; a key no ask of " +
+				"which waits is answered so and nothing is asked. An ask whose step already runs on a node is asked again too: it is then " +
+				"built twice, the first run's work wasted and its end read by nobody, never a wrong key; builder_stuck shows whether a step runs.",
+			Input: map[string]any{"type": "object", "required": []string{"module", "fp", "why"}, "properties": map[string]any{
+				"module": map[string]any{"type": "string", "description": "the module's name"},
+				"fp":     map[string]any{"type": "string", "description": "its new-form fingerprint, src2:<hex>, as builder_stuck shows it"},
+				"why":    map[string]any{"type": "string", "description": "why it is asked again, in plain words"}}},
+			Run: func(args map[string]any) (any, error) {
+				key, value, err := build.NewReAsk(str(args, "module"), str(args, "fp"), str(args, "why"), stdio.CallerOf(args), now())
+				if err != nil {
+					return nil, err
+				}
+				if err := askAgain(key, value); err != nil {
+					return nil, fmt.Errorf("the ask could not be written to %s: %w", build.ReAsksBucket, err)
+				}
+				return map[string]any{"key": key, "asked": true,
+					"next": "the builder answers in " + build.ReAsksState + "." + key + "; " + build.StuckTool + " shows the answer and the new ask"}, nil
+			}},
+		{Name: build.SeatQueueTool,
+			Description: "What waits on a build seat's stream (SEAT_NODE_BUILD_<LANGUAGE>), as the builder reads it: the stream's message count and " +
+				"sequences, and per kind (delivery, check, recheck) its messages oldest first, at most limit each, with sequence, ask id, when asked " +
+				"and put, age, the state of the step a build node wrote for it, and whether the builder counts it as still waiting or as given up " +
+				"on (its worker delivered it its most times) or taken; and the ask asked last that a node took. The workers' delivery counts are " +
+				"not readable by the builder (nats.nats_streams shows them). Read by the builder, which alone may read the stream; read-only.",
+			Input: map[string]any{"type": "object", "properties": map[string]any{
+				"language": map[string]any{"type": "string", "description": "go, node or image (default go)"},
+				"seat":     map[string]any{"type": "string", "description": "or the seat: node-build-go, node-build-node, node-build-image"},
+				"limit":    map[string]any{"type": "integer", "description": fmt.Sprintf("messages read of each kind (default %d, at most %d)", build.DefaultQueueLimit, build.MaxQueueLimit)}}},
+			Run: func(args map[string]any) (any, error) {
+				language := build.LanguageOf(str(args, "language"))
+				if language == "" {
+					language = build.LanguageOf(str(args, "seat"))
+				}
+				if language == "" {
+					language = delivery.AskGo
+				}
+				limit, _ := args["limit"].(float64)
+				key, value, err := build.NewQueueAsk(language, int(limit), stdio.CallerOf(args), now())
+				if err != nil {
+					return nil, err
+				}
+				return queue(key, value)
+			}},
+	}
+}
+
+// askQueue writes the queues value at key and waits, on a watch of the key, at most bound for the builder's answer.
+func askQueue(s stdio.KeptState, bound time.Duration) queueAsker {
+	return func(key string, value []byte) (json.RawMessage, error) {
+		answered := make(chan json.RawMessage, 1)
+		var rev uint64
+		var mu sync.Mutex
+		if err := s.Watch(key, func(c stdio.StateChange) error {
+			mu.Lock()
+			mine := rev
+			mu.Unlock()
+			if c.Op != "put" || mine == 0 || c.Revision <= mine {
+				return nil
+			}
+			if a, err := build.DecodeQueueAsk(c.Value); err == nil && !a.Answered.IsZero() {
+				select {
+				case answered <- c.Value:
+				default:
+				}
+			}
+			return nil
+		}); err != nil {
+			return nil, fmt.Errorf("watching %s.%s: %w", build.QueuesState, key, err)
+		}
+		r, err := s.Put(key, json.RawMessage(value))
+		if err != nil {
+			return nil, fmt.Errorf("the read could not be written to %s: %w", build.QueuesBucket, err)
+		}
+		mu.Lock()
+		rev = r
+		mu.Unlock()
+		// The watch may have heard the answer before the revision was known: read it once too.
+		if e, err := s.Get(key); err == nil && e != nil && e.Revision > r {
+			if a, err := build.DecodeQueueAsk(e.Value); err == nil && !a.Answered.IsZero() {
+				return e.Value, nil
+			}
+		}
+		wait, cancel := context.WithTimeout(context.Background(), bound) // a context's bound, no timer (ADR 0437 §8)
+		defer cancel()
+		select {
+		case v := <-answered:
+			return v, nil
+		case <-wait.Done():
+			return nil, fmt.Errorf("the builder did not answer %s.%s within %s: it is not running, or cannot read %s", build.QueuesState, key, bound, build.QueuesBucket)
+		}
+	}
+}
+
+func read(s stdio.KeptState) build.StateRead {
+	return build.StateRead{Keys: s.Keys, Get: func(key string) (json.RawMessage, time.Time, error) {
+		e, err := s.Get(key)
+		if err != nil || e == nil {
+			return nil, time.Time{}, err
+		}
+		return e.Value, time.Time{}, nil
+	}, Written: func(key string) time.Time {
+		e, err := s.Get(key)
+		if err != nil || e == nil {
+			return time.Time{}
+		}
+		return written(s, key, e.Revision)
+	}}
+}
+
+// written is when the key's revision was written, from its history, or zero.
+func written(s stdio.KeptState, key string, revision uint64) time.Time {
+	h, err := s.History(key)
+	if err != nil {
+		return time.Time{}
+	}
+	for _, r := range h {
+		if r.Revision == revision {
+			return r.At
+		}
+	}
+	return time.Time{}
+}
+
+func main() {
+	steps := map[string]build.StateRead{}
+	for _, bm := range build.StuckBuildModules {
+		steps[bm] = read(stdio.State(bm + ".steps"))
+	}
+	reasks := stdio.State(build.ReAsksState)
+	r := build.StuckReader{Logf: func(f string, a ...any) { fmt.Fprintf(os.Stderr, "[mesh-delivery-builder-tools] "+f+"\n", a...) }, Builds: read(stdio.State("builds")), ReAsks: read(reasks), Steps: steps}
+	put := func(key string, value []byte) error {
+		_, err := reasks.Put(key, json.RawMessage(value))
+		return err
+	}
+	if err := stdio.Serve("", Tools(r, put, askQueue(stdio.State(build.QueuesState), 25*time.Second), time.Now)); err != nil {
+		fmt.Fprintf(os.Stderr, "[mesh-delivery-builder-tools] %v\n", err)
+		os.Exit(1)
+	}
+}

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

@@ -4,73 +4,303 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"sort"
+	"strings"
+	"time"
 
 	"github.com/nats-io/nats.go/jetstream"
 
 	"git.novox.be/novox/mesh-sdk/go/delivery"
+
+	"mesh-delivery-builder/internal/build"
 )
 
 // heldOnTheBus says whether an ask a build key names is on its build seat's stream, or a build node wrote a step for it
-// (issue 1910): read once, at the take-over of a key at start, never polled.
+// (issue 1910), read once at the take-over of a key at start; and, at the builder's check bound, whether an ask can still
+// be answered (issue 1945). Never polled.
 type heldOnTheBus struct{ js jetstream.JetStream }
 
 // Held is build.Builder's Held: a step key `<node>.<ask id>` in the build module's steps, or a message of the stream under
 // the ask's subject whose Nats-Msg-Id is the ask's id.
 func (h heldOnTheBus) Held(ctx context.Context, a delivery.StepAsk) (bool, error) {
-	module, err := delivery.BuildModuleOf(a.Language)
+	seen, err := h.seen(ctx, a, false)
 	if err != nil {
 		return false, err
 	}
-	pattern, err := delivery.AskStepWatch(a.ID)
+	return seen.Step || seen.OnStream, nil
+}
+
+// Gone is build.Builder's Gone: why the ask can no longer be answered (build.AskSeen.Why), or "".
+func (h heldOnTheBus) Gone(ctx context.Context, a delivery.StepAsk) (string, error) {
+	seen, err := h.seen(ctx, a, true)
 	if err != nil {
-		return false, err
+		return "", err
+	}
+	return seen.Why(), nil
+}
+
+// maxOnSubject bounds the messages of one subject a read walks (each a read by sequence, next by subject), and maxSteps
+// the steps it reads whole to find the last ask taken: past them the read says it cannot tell, and the ask is waited for.
+const (
+	maxOnSubject = 5000
+	maxSteps     = 200
+)
+
+// seen reads what the bus holds of the ask; with later, the ask asked last that a build node took too (AskSeen.Last), when
+// the ask is on its stream.
+func (h heldOnTheBus) seen(ctx context.Context, a delivery.StepAsk, later bool) (build.AskSeen, error) {
+	seen := build.AskSeen{Kind: a.Kind, At: build.AskedAt(a.ID)}
+	module, err := delivery.BuildModuleOf(a.Language)
+	if err != nil {
+		return seen, err
 	}
 	steps, err := h.js.KeyValue(ctx, delivery.StepsBucket(module))
 	if err != nil {
-		return false, fmt.Errorf("the bucket %s: %w", delivery.StepsBucket(module), err)
+		return seen, fmt.Errorf("the bucket %s: %w", delivery.StepsBucket(module), err)
+	}
+	pattern, err := delivery.AskStepWatch(a.ID)
+	if err != nil {
+		return seen, err
 	}
 	keys, err := steps.ListKeysFiltered(ctx, pattern)
 	if err != nil {
-		return false, err
+		return seen, err
 	}
 	for range keys.Keys() {
 		keys.Stop()
-		return true, nil // a node took it
+		seen.Step = true // a node took it
+		return seen, nil
+	}
+	stream, name, subject, err := h.stream(ctx, a.Language, a.Kind)
+	seen.Stream = name
+	if errors.Is(err, jetstream.ErrStreamNotFound) {
+		return seen, nil
 	}
-	subject, err := delivery.AskSubject(a.Language, a.Kind)
 	if err != nil {
-		return false, err
+		return seen, err
+	}
+	var ahead []build.Queued
+	err = onSubject(ctx, stream, subject, func(q build.Queued) bool {
+		if q.Ask == a.ID {
+			seen.OnStream, seen.PutAt = true, q.PutAt
+			return false
+		}
+		ahead = append(ahead, q)
+		return true
+	})
+	if err != nil || !seen.OnStream || !later {
+		return seen, err
+	}
+	taken, err := stepKeys(ctx, steps)
+	if err != nil {
+		return seen, err
+	}
+	for i := range ahead {
+		ahead[i].Step = stepState(ctx, steps, taken[ahead[i].Ask])
+	}
+	seen.Ahead = ahead
+	since := seen.At
+	if since.IsZero() {
+		since = seen.PutAt
+	}
+	for _, q := range ahead {
+		if q.At.Before(since) {
+			since = q.At
+		}
+	}
+	seen.Last = lastTaken(ctx, steps, taken, a.Kind, since)
+	return seen, nil
+}
+
+// stream is the build seat's stream of language, its name and the subject of kind on it.
+func (h heldOnTheBus) stream(ctx context.Context, language, kind string) (jetstream.Stream, string, string, error) {
+	subject, err := delivery.AskSubject(language, kind)
+	if err != nil {
+		return nil, "", "", err
 	}
 	// The seat's stream by its name (delivery.AskStream), as the broker grants the builder its information and its
 	// messages by sequence and nothing wider (issue 1910).
-	name, err := delivery.AskStream(a.Language)
+	name, err := delivery.AskStream(language)
 	if err != nil {
-		return false, err
+		return nil, "", "", err
 	}
 	stream, err := h.js.Stream(ctx, name)
-	if errors.Is(err, jetstream.ErrStreamNotFound) {
-		return false, nil
-	}
-	if err != nil {
-		return false, err
-	}
-	// A seat's stream is a work queue, which takes no second consumer over its subjects: its messages are read by sequence
-	// instead, each a direct read that consumes nothing.
-	info, err := stream.Info(ctx)
-	if err != nil {
-		return false, err
-	}
-	for seq := info.State.FirstSeq; seq != 0 && seq <= info.State.LastSeq; seq++ {
-		m, err := stream.GetMsg(ctx, seq)
+	return stream, name, subject, err
+}
+
+// onSubject hands each message of subject on stream to each, in order, until each answers false or the subject's
+// messages end. A seat's stream is a work queue, which takes no second consumer over its subjects: its messages are read
+// by sequence instead, each the next of the subject (one read per message, none for the sequences taken between), and
+// none consumed.
+func onSubject(ctx context.Context, stream jetstream.Stream, subject string, each func(build.Queued) bool) error {
+	for seq, n := uint64(1), 0; ; n++ {
+		if n >= maxOnSubject {
+			return fmt.Errorf("%s holds more than %d messages before the one asked after: not read further", subject, maxOnSubject)
+		}
+		m, err := stream.GetMsg(ctx, seq, jetstream.WithGetMsgSubject(subject))
 		if errors.Is(err, jetstream.ErrMsgNotFound) {
-			continue // taken and acknowledged, or removed
+			return nil
 		}
 		if err != nil {
-			return false, err
+			return err
 		}
-		if m.Subject == subject && m.Header.Get(jetstream.MsgIDHeader) == a.ID {
-			return true, nil
+		q := build.Queued{Seq: m.Sequence, Ask: m.Header.Get(jetstream.MsgIDHeader), PutAt: m.Time}
+		if _, kind, err := delivery.ParseAskSubject(m.Subject); err == nil {
+			q.Kind = kind
+		}
+		if q.At = build.AskedAt(q.Ask); q.At.IsZero() {
+			q.At = q.PutAt
+		}
+		if !each(q) {
+			return nil
+		}
+		seq = m.Sequence + 1
+	}
+}
+
+// stepKeys are the build module's step keys by ask id: `<node>.<ask id>` (a node's own key, such as its pause, has no
+// ask id and is left out).
+func stepKeys(ctx context.Context, steps jetstream.KeyValue) (map[string][]string, error) {
+	all, err := steps.ListKeys(ctx)
+	if err != nil {
+		if errors.Is(err, jetstream.ErrNoKeysFound) {
+			return map[string][]string{}, nil
+		}
+		return nil, err
+	}
+	defer all.Stop()
+	out := map[string][]string{}
+	for k := range all.Keys() {
+		if _, id, ok := strings.Cut(k, "."); ok && !build.AskedAt(id).IsZero() {
+			out[id] = append(out[id], k)
 		}
 	}
-	return false, nil
+	return out, nil
+}
+
+// stepState is the state of the first of keys a node wrote, "" for none.
+func stepState(ctx context.Context, steps jetstream.KeyValue, keys []string) string {
+	for _, k := range keys {
+		e, err := steps.Get(ctx, k)
+		if err != nil {
+			continue
+		}
+		if s, err := delivery.DecodeStep(e.Value()); err == nil {
+			return s.State
+		}
+	}
+	return ""
+}
+
+// lastTaken is the ask asked last after since that a build node took, of a kind whose taking proves kind passed
+// (build.Proves), read newest first and at most maxSteps of them; nil for none. A step's tool is "<kind> ask"
+// (mesh-sdk buildstep).
+func lastTaken(ctx context.Context, steps jetstream.KeyValue, taken map[string][]string, kind string, since time.Time) *build.Taken {
+	var ids []string
+	for id := range taken {
+		if build.AskedAt(id).After(since) {
+			ids = append(ids, id)
+		}
+	}
+	sort.Slice(ids, func(i, j int) bool { return build.AskedAt(ids[i]).After(build.AskedAt(ids[j])) })
+	read := 0
+	for _, id := range ids {
+		for _, k := range taken[id] {
+			if read >= maxSteps {
+				return nil
+			}
+			read++
+			e, err := steps.Get(ctx, k)
+			if err != nil {
+				continue
+			}
+			s, err := delivery.DecodeStep(e.Value())
+			if err != nil {
+				continue
+			}
+			stepKind, ok := strings.CutSuffix(s.Tool, " ask")
+			if !ok || !build.Proves(stepKind, kind) {
+				continue
+			}
+			return &build.Taken{Ask: id, Kind: stepKind, At: build.AskedAt(id), Node: s.Node, Started: s.Started}
+		}
+	}
+	return nil
+}
+
+// Drop is build.Dropper's Drop: it removes the ask's message from its build seat's stream, found by its Nats-Msg-Id on its
+// subject, when the stream still holds it. The builder's account must be granted $JS.API.STREAM.MSG.DELETE.<stream>.
+func (h heldOnTheBus) Drop(ctx context.Context, a delivery.StepAsk) error {
+	stream, name, subject, err := h.stream(ctx, a.Language, a.Kind)
+	if errors.Is(err, jetstream.ErrStreamNotFound) {
+		return nil
+	}
+	if err != nil {
+		return err
+	}
+	var seq uint64
+	if err := onSubject(ctx, stream, subject, func(q build.Queued) bool {
+		if q.Ask == a.ID {
+			seq = q.Seq
+			return false
+		}
+		return true
+	}); err != nil || seq == 0 {
+		return err
+	}
+	if err := stream.DeleteMsg(ctx, seq); err != nil && !errors.Is(err, jetstream.ErrMsgNotFound) {
+		return fmt.Errorf("deleting message %d of %s (is $JS.API.STREAM.MSG.DELETE.%s granted to the builder?): %w", seq, name, name, err)
+	}
+	return nil
+}
+
+// Queue is what the tool builder_seat_queue answers (build.SeatQueue): the messages of language's build seat's stream, of
+// every kind, oldest first and at most limit of each kind, each with its step's state and whether it still waits.
+func (h heldOnTheBus) Queue(ctx context.Context, language string, limit int) (build.SeatQueue, error) {
+	q := build.SeatQueue{Language: language, Note: build.QueueNote}
+	module, err := delivery.BuildModuleOf(language)
+	if err != nil {
+		return q, err
+	}
+	stream, name, _, err := h.stream(ctx, language, delivery.AskDelivery)
+	q.Stream = name
+	if errors.Is(err, jetstream.ErrStreamNotFound) {
+		q.Note = name + " does not exist"
+		return q, nil
+	}
+	if err != nil {
+		return q, err
+	}
+	info, err := stream.Info(ctx)
+	if err != nil {
+		return q, err
+	}
+	q.Messages, q.FirstSeq, q.LastSeq = info.State.Msgs, info.State.FirstSeq, info.State.LastSeq
+	steps, err := h.js.KeyValue(ctx, delivery.StepsBucket(module))
+	if err != nil {
+		return q, fmt.Errorf("the bucket %s: %w", delivery.StepsBucket(module), err)
+	}
+	taken, err := stepKeys(ctx, steps)
+	if err != nil {
+		return q, err
+	}
+	for _, kind := range delivery.AskKinds {
+		subject, _ := delivery.AskSubject(language, kind)
+		var ahead []build.Queued
+		if err := onSubject(ctx, stream, subject, func(m build.Queued) bool {
+			ahead = append(ahead, m)
+			return len(ahead) < limit
+		}); err != nil {
+			return q, err
+		}
+		if len(ahead) == 0 {
+			continue
+		}
+		for i := range ahead {
+			ahead[i].Step = stepState(ctx, steps, taken[ahead[i].Ask])
+		}
+		seen := build.AskSeen{Kind: kind, Ahead: ahead, Last: lastTaken(ctx, steps, taken, kind, ahead[0].At)}
+		q.Kinds = append(q.Kinds, build.NewKindQueue(kind, seen, time.Now()))
+	}
+	return q, nil
 }

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

@@ -24,6 +24,8 @@ import (
 
 	"git.novox.be/novox/mesh-sdk/go/buildstep"
 	"git.novox.be/novox/mesh-sdk/go/delivery"
+
+	"mesh-delivery-builder/internal/build"
 )
 
 // busConf is the shape of the composed accounts file (mesh-controller internal/broker): one account with JetStream, the
@@ -822,6 +824,34 @@ func TestARestartedBuilderPutsNoSecondAsk(t *testing.T) {
 			t.Fatalf("asks put = %+v within 20 s, want the two the key names", asked())
 		}
 	}
+	// 1945 note 62, on the real bus: the stream read as builder_seat_queue reads it shows the Go ask waiting; the ask can
+	// still be answered (nothing was taken since); and a message dropped as an ask asked again is gone from the stream.
+	held := heldOnTheBus{b.js}
+	goAsk := delivery.StepAsk{ID: v.Asks[delivery.AskGo], Language: delivery.AskGo, Kind: delivery.AskDelivery}
+	stale := delivery.StepAsk{ID: "mvo53jcm70b5f7", Language: delivery.AskGo, Kind: delivery.AskDelivery}
+	subject, _ := delivery.AskSubject(delivery.AskGo, delivery.AskDelivery)
+	if _, err := b.js.Publish(b.ctx, subject, []byte("{}"), jetstream.WithMsgID(stale.ID)); err != nil {
+		t.Fatal(err)
+	}
+	queued := func() []build.QueuedState {
+		q, err := held.Queue(b.ctx, delivery.AskGo, 10)
+		if err != nil || len(q.Kinds) != 1 || q.Kinds[0].Kind != delivery.AskDelivery {
+			t.Fatalf("the Go seat's queue = %+v, %v; want its delivery asks", q, err)
+		}
+		return q.Kinds[0].Messages
+	}
+	if m := queued(); len(m) != 2 || m[0].Ask != goAsk.ID || !m[0].Waits || m[1].Ask != stale.ID {
+		t.Fatalf("the Go delivery queue = %+v, want the key's ask waiting, then the stale one", m)
+	}
+	if why, err := held.Gone(b.ctx, goAsk); why != "" || err != nil {
+		t.Errorf("Gone(the waiting ask) = %q, %v; want it answerable", why, err)
+	}
+	if err := held.Drop(b.ctx, stale); err != nil {
+		t.Fatal(err)
+	}
+	if m := queued(); len(m) != 1 || m[0].Ask != goAsk.ID {
+		t.Fatalf("after the drop the Go delivery queue = %+v, want the key's ask alone", m)
+	}
 	b.stop()
 	b.start(t, "n1", nil) // the same builder, started again, while both asks wait
 	b.fake.open()

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

@@ -13,7 +13,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
 	"io"
 	"log"
 	"maps"
@@ -25,6 +24,8 @@ import (
 	"syscall"
 	"time"
 
+	"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
+
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
 
@@ -124,29 +125,20 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	if err != nil {
 		return err
 	}
-	bucket := func(name string) (jetstream.KeyValue, error) {
-		kv, err := js.KeyValue(ctx, name)
-		if err != nil {
-			return nil, fmt.Errorf("the bucket %s: %w", name, err)
-		}
-		return kv, nil
-	}
-	builds, err := bucket(delivery.BuilderBuildsBucket)
+	builds, err := process.EssentialBucket(ctx, js, "builder", delivery.BuilderBuildsBucket, "claim or write a build key")
 	if err != nil {
 		return err
 	}
-	modules, err := bucket(delivery.ModulesBucket)
+	modules, err := process.EssentialBucket(ctx, js, "builder", delivery.ModulesBucket, "know any fingerprint to build")
 	if err != nil {
 		return err
 	}
-	manifests, err := bucket(delivery.ManifestsBucket)
-	if err != nil {
-		return err
-	}
-	candidates, err := bucket(delivery.CandidatesBucket)
+	manifests, err := process.EssentialBucket(ctx, js, "builder", delivery.ManifestsBucket, "read what a module builds")
 	if err != nil {
 		return err
 	}
+	// The candidates' fingerprints are built too (issue 1684): a builder that cannot read them builds the trunk's.
+	candidates := process.OptionalBucket(ctx, js, delivery.CandidatesBucket, "no candidate's fingerprint is built until the builder starts again; the trunk's are", logger.Printf)
 	bound := defaultStepBound
 	if v := strings.TrimSpace(getenv("MESH_BUILDER_STEP_BOUND")); v != "" {
 		if bound, err = time.ParseDuration(v); err != nil {
@@ -169,44 +161,36 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	}
 	defer calls.Close()
 	b := &build.Builder{
-		Claimer:    build.Claimer{Store: kvStore{builds}, By: cred.Node + "." + cred.Module},
-		Asks:       buildrun.Asker{Publisher: asks, Waiter: asks, Bound: bound},
-		Held:       heldOnTheBus{js},
+		Claimer: build.Claimer{Store: kvStore{builds}, By: cred.Node + "." + cred.Module},
+		Asks:    buildrun.Asker{Publisher: asks, Waiter: asks, Bound: bound},
+		Held:    heldOnTheBus{js},
+		// An ask that can no longer be answered is asked again (issue 1945), read at the step bound: a context's bound.
+		Gone:       heldOnTheBus{js},
+		CheckAfter: bound,
 		Toolchains: tools,
 		Manifests:  manifestReader{manifests},
 		Logf:       logger.Printf,
 		Stored:     storedVia(calls.Call),
 	}
 	// A candidate's preview is merged onto the trunk head the indexer previewed it onto: source.main at the candidate's base
-	// revision (the gate's source state). Where the bucket is not there yet no candidate's fingerprint is built, and said.
-	// The source's revisions are watched with their history (a direct get of one revision is not what a read grants): a
-	// revision heard after the candidate that names it builds that candidate then.
-	trunks := &sourceHeads{}
-	var sw jetstream.KeyWatcher
-	switch kv, err := js.KeyValue(ctx, delivery.SourceBucket); {
-	case errors.Is(err, jetstream.ErrBucketNotFound):
-		logger.Printf("the bucket %s does not exist yet: no candidate's fingerprint is built until the builder starts again", delivery.SourceBucket)
-	case err != nil:
-		return fmt.Errorf("the bucket %s: %w", delivery.SourceBucket, err)
-	default:
-		w, err := kv.Watch(ctx, delivery.SourceMainKey, jetstream.IncludeHistory())
-		if err != nil {
-			return fmt.Errorf("the bucket %s cannot be watched: %w", delivery.SourceBucket, err)
+	// revision (the gate's source state). The source's revisions are watched with their history (a direct get of one revision
+	// is not what a read grants): a revision heard after the candidate that names it builds that candidate then. Where it
+	// cannot be read the trunk's keys are built and the candidates wait (issue 1945).
+	errs := make(chan error, 6)
+	trunks := &trunkSource{ctx: ctx, js: js, heads: &sourceHeads{}, moved: b.TrunkMoved, errs: errs, logf: logger.Printf}
+	trunks.start()
+	b.Trunk = trunks.head
+	b.Said = func(e build.AskedAgain) {
+		if err := emitAskedAgain(context.WithoutCancel(ctx), js, cred, e); err != nil {
+			logger.Printf("%v", err)
 		}
-		if sw, err = settled(ctx, w, func(e jetstream.KeyValueEntry) { trunks.put(e) }); err != nil {
-			return err
-		}
-		b.Trunk = trunks.head
 	}
 	// **Said once every build module answered a reference, taken away when stopped** (issue 1886, ADR 0522): the
 	// controller delivers from this builder's keys only while an instance says it writes references. A bucket not there
 	// yet (the manifest's state not made) is said, and the controller's plan waits for the keys, named (no build-agent
 	// is asked since R4g, ADR 0524 §7).
-	switch kv, err := js.KeyValue(ctx, instancesBucket); {
-	case errors.Is(err, jetstream.ErrBucketNotFound):
-		logger.Printf("the bucket %s does not exist yet: the controller is not told that this builder writes references until it starts again", instancesBucket)
-	case err != nil:
-		return fmt.Errorf("the bucket %s: %w", instancesBucket, err)
+	switch kv := process.OptionalBucket(ctx, js, instancesBucket, "the controller is not told that this builder writes references until it starts again", logger.Printf); {
+	case kv == nil:
 	default:
 		says := newSayer(ctx, kv, cred.Node, []string{buildrun.GoBuildModule, buildrun.NodeBuildModule, build.ImageBuildModule}, logger.Printf)
 		b.Answered = says.Answered
@@ -286,23 +270,36 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 			b.ForgetCandidate(e.Key())
 		}
 	}
-	cw, err := settle(ctx, candidates, eachCandidate)
-	if err != nil {
-		return err
+	if candidates != nil {
+		cw, err := settle(ctx, candidates, eachCandidate)
+		if err != nil {
+			process.NotWatched(logger.Printf, delivery.CandidatesBucket, err, "no candidate's fingerprint is built until the builder starts again")
+		} else {
+			go func() { errs <- follow(ctx, cw, eachCandidate) }()
+		}
+	}
+	// A person asks a key's build again with builder_ask_again, which writes the reasks state (issue 1945): watched, as
+	// every other input. The builder asks again by itself without it.
+	if kv := process.OptionalBucket(ctx, js, build.ReAsksBucket, "builder_ask_again is not heard until the builder starts again; an ask that can no longer be answered is still asked again by itself", logger.Printf); kv != nil {
+		eachReAsk := func(e jetstream.KeyValueEntry) { reAsked(ctx, b, kv, e, logger.Printf) }
+		if rw, err := settle(ctx, kv, eachReAsk); err != nil {
+			process.NotWatched(logger.Printf, build.ReAsksBucket, err, "builder_ask_again is not heard until the builder starts again")
+		} else {
+			go func() { errs <- follow(ctx, rw, eachReAsk) }()
+		}
+	}
+	// A person reads a build seat's stream with builder_seat_queue, which writes the queues state (issue 1945 note 62): the
+	// builder alone may read the stream, so it answers there.
+	if kv := process.OptionalBucket(ctx, js, build.QueuesBucket, "builder_seat_queue is not answered until the builder starts again", logger.Printf); kv != nil {
+		eachRead := func(e jetstream.KeyValueEntry) { queueRead(ctx, heldOnTheBus{js}, kv, e, logger.Printf) }
+		if qw, err := settle(ctx, kv, eachRead); err != nil {
+			process.NotWatched(logger.Printf, build.QueuesBucket, err, "builder_seat_queue is not answered until the builder starts again")
+		} else {
+			go func() { errs <- follow(ctx, qw, eachRead) }()
+		}
 	}
 	logger.Printf("started on %s", cred.Node)
-	errs := make(chan error, 3)
 	go func() { errs <- follow(ctx, w, each) }()
-	go func() { errs <- follow(ctx, cw, eachCandidate) }()
-	if sw != nil {
-		go func() {
-			errs <- follow(ctx, sw, func(e jetstream.KeyValueEntry) {
-				if trunks.put(e) {
-					b.TrunkMoved(ctx)
-				}
-			})
-		}()
-	}
 	select {
 	case <-ctx.Done():
 		b.Wait()

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

@@ -0,0 +1,78 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"sync"
+
+	"github.com/nats-io/nats.go/jetstream"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+// **A part does not stop for a bucket it can work without** (mesh-sdk process.EssentialBucket, OptionalBucket) (novox/hq issue 1945, met twice on 2026-10-25: the builder of
+// mesh-delivery #48 exited on a refused read of the gate's source, and its reasks bucket would have done the same). A grant
+// reaches the bus with a delivery, so a bucket the bus refuses, or times out on, at start is often only not granted yet:
+// exiting for it stops every build until a person's hand. A bucket the builder cannot build at all without (its builds,
+// the indexer's modules and manifests) still stops the start, naming the grant to look at; every other is said and left,
+// and opened again at the next event that needs it where there is one.
+
+// trunkSource is the gate's source history the candidates' previews are merged onto (sourceHeads), opened at start and,
+// while it cannot be, again at the next candidate that asks a trunk head: a builder without it builds the trunk's keys,
+// and a candidate's preview waits, saying why.
+type trunkSource struct {
+	ctx   context.Context
+	js    bucketOpener
+	heads *sourceHeads
+	moved func(context.Context)
+	errs  chan<- error
+	logf  func(string, ...any)
+
+	mu     sync.Mutex
+	opened bool
+}
+
+// ensure opens and settles the source's watch once, and follows it; an error leaves it closed for the next ask.
+func (t *trunkSource) ensure() error {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	if t.opened {
+		return nil
+	}
+	kv, err := t.js.KeyValue(t.ctx, delivery.SourceBucket)
+	if err != nil {
+		return fmt.Errorf("the bucket %s cannot be read (%v)", delivery.SourceBucket, err)
+	}
+	w, err := kv.Watch(t.ctx, delivery.SourceMainKey, jetstream.IncludeHistory())
+	if err != nil {
+		return fmt.Errorf("the bucket %s cannot be watched (%v)", delivery.SourceBucket, err)
+	}
+	sw, err := settled(t.ctx, w, func(e jetstream.KeyValueEntry) { t.heads.put(e) })
+	if err != nil {
+		return fmt.Errorf("the bucket %s: %w", delivery.SourceBucket, err)
+	}
+	t.opened = true
+	go func() {
+		t.errs <- follow(t.ctx, sw, func(e jetstream.KeyValueEntry) {
+			if t.heads.put(e) && t.moved != nil {
+				t.moved(t.ctx)
+			}
+		})
+	}()
+	return nil
+}
+
+// start opens the source at start; one that cannot be is said, and the builder goes on.
+func (t *trunkSource) start() {
+	if err := t.ensure(); err != nil {
+		t.logf("%v: the trunk's keys are built, and a candidate's preview waits until the source can be read, tried again at its next candidate", err)
+	}
+}
+
+// head is Builder.Trunk: the head of repository at source revision rev, opening the source first if it is not yet.
+func (t *trunkSource) head(ctx context.Context, rev uint64, repository string) (string, error) {
+	if err := t.ensure(); err != nil {
+		return "", err
+	}
+	return t.heads.head(ctx, rev, repository)
+}

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

@@ -0,0 +1,49 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"time"
+
+	"github.com/nats-io/nats.go/jetstream"
+
+	"mesh-delivery-builder/internal/build"
+)
+
+// queueReader is the part of the builder a queues value reads through: heldOnTheBus.
+type queueReader interface {
+	Queue(ctx context.Context, language string, limit int) (build.SeatQueue, error)
+}
+
+// queueRead answers one value of the queues state (the tool builder_seat_queue, issue 1945 note 62): a value not yet
+// answered is read now, from the stream the builder's account may read, and written back by compare-and-set at its
+// revision with what was read or why it could not be. A value answered already (this builder's own write, or another
+// instance's) or deleted is left.
+func queueRead(ctx context.Context, r queueReader, kv jetstream.KeyValue, e jetstream.KeyValueEntry, logf func(string, ...any)) {
+	if e.Operation() != jetstream.KeyValuePut {
+		return
+	}
+	a, err := build.DecodeQueueAsk(e.Value())
+	if err != nil {
+		logf("%s.%s holds what is not a queue read, left: %v", build.QueuesState, e.Key(), err)
+		return
+	}
+	if !a.Answered.IsZero() {
+		return
+	}
+	read, cancel := context.WithTimeout(ctx, 20*time.Second)
+	q, err := r.Queue(read, e.Key(), max(1, min(a.Limit, build.MaxQueueLimit)))
+	cancel()
+	if err != nil {
+		a.Error = err.Error()
+	}
+	a.Queue, a.Answered = &q, time.Now().UTC()
+	data, err := json.Marshal(a)
+	if err != nil {
+		return
+	}
+	if _, err := kv.Update(ctx, e.Key(), data, e.Revision()); err != nil && !errors.Is(err, jetstream.ErrKeyExists) && !errors.Is(err, jetstream.ErrKeyRevisionMismatch) {
+		logf("%s.%s: the read could not be written: %v", build.QueuesState, e.Key(), err)
+	}
+}

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

@@ -0,0 +1,75 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+
+	"mesh-delivery-builder/internal/build"
+)
+
+// askAgainer is the part of the builder a reasks value asks through.
+type askAgainer interface {
+	AskAgain(key, why string) []string
+}
+
+// reAsked answers one value of the reasks state (the tool builder_ask_again, issue 1945): a value not yet answered asks the
+// key's waiting lanes again and is written back, by compare-and-set at its revision, with what the builder did. A value
+// answered already (this builder's own write) or deleted is left.
+func reAsked(ctx context.Context, b askAgainer, kv jetstream.KeyValue, e jetstream.KeyValueEntry, logf func(string, ...any)) {
+	if e.Operation() != jetstream.KeyValuePut {
+		return
+	}
+	r, err := build.DecodeReAsk(e.Value())
+	if err != nil {
+		logf("%s.%s holds what is not an ask to ask again, left: %v", build.ReAsksState, e.Key(), err)
+		return
+	}
+	if r.Answer != "" {
+		return
+	}
+	asked := b.AskAgain(e.Key(), r.Why)
+	if len(asked) == 0 {
+		r.Answer = "no ask of this key waits in this builder now (it is not building, or its build has not reached its ask yet): nothing was asked; builder_stuck shows where it stands"
+	} else {
+		r.Answer = "asked again: the " + strings.Join(asked, ", ") + " ask, under a new id the key names"
+	}
+	r.Answered = time.Now().UTC()
+	logf("%s: asked again by %s (%s): %s", e.Key(), r.By, r.Why, r.Answer)
+	data, err := json.Marshal(r)
+	if err != nil {
+		return
+	}
+	if _, err := kv.Update(ctx, e.Key(), data, e.Revision()); err != nil && !errors.Is(err, jetstream.ErrKeyExists) && !errors.Is(err, jetstream.ErrKeyRevisionMismatch) {
+		logf("%s.%s: the answer could not be written: %v", build.ReAsksState, e.Key(), err)
+	}
+}
+
+// emitAskedAgain says the builder's event build.asked-again (issue 1945): mesh.mod.<module>.event.build.asked-again,
+// with the SDK's envelope headers and the new ask's id as the bus's dedup id.
+func emitAskedAgain(ctx context.Context, js jetstream.JetStream, cred credential, e build.AskedAgain) error {
+	data, err := json.Marshal(e)
+	if err != nil {
+		return err
+	}
+	msg := nats.NewMsg("mesh.mod." + cred.Module + ".event." + build.AskedAgainEvent)
+	id := "asked-again-" + e.New
+	msg.Header.Set("x-event-id", id)
+	msg.Header.Set("x-source", cred.Module)
+	msg.Header.Set("x-node", cred.Node)
+	msg.Header.Set("x-time", time.Now().UTC().Format(time.RFC3339Nano))
+	msg.Header.Set("content-type", "application/json")
+	msg.Data = data
+	pub, cancel := context.WithTimeout(ctx, 10*time.Second)
+	defer cancel()
+	if _, err := js.PublishMsg(pub, msg, jetstream.WithMsgID(id)); err != nil {
+		return fmt.Errorf("saying %s for %s: %w", build.AskedAgainEvent, e.Key, err)
+	}
+	return nil
+}

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

@@ -193,7 +193,9 @@ func watchAssignments(ctx context.Context, js bucketOpener, gate *toolchainGate,
 	each := func(e jetstream.KeyValueEntry) { gate.Assignment(e.Key(), e.Value(), !opOf(e), e.Created()) }
 	w, err := settle(ctx, kv, each)
 	if err != nil {
-		return nil, fmt.Errorf("the bucket %s: %w", provider.AssignmentsBucket, err)
+		// Not fatal either (issue 1945): a watch the bus refuses is the same missing grant as a refused read.
+		logger.Printf("the bucket %s cannot be watched, so every node's toolchain report counts, an unassigned build node's too (issue 1940): %v", provider.AssignmentsBucket, err)
+		return errs, nil
 	}
 	if gate.Settled() {
 		changed(ctx)

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

@@ -3,11 +3,19 @@ package main
 // The same in mesh-delivery-indexer, mesh-delivery-builder and mesh-delivery-checker, as standing.go is (issue 1940).
 
 import (
+	"bytes"
+	"context"
+	"errors"
 	"fmt"
+	"log"
 	"strings"
 	"testing"
 	"time"
 
+	"git.novox.be/novox/mesh-sdk/go/process"
+
+	"github.com/nats-io/nats.go/jetstream"
+
 	"git.novox.be/novox/mesh-sdk/go/delivery"
 	"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
 )
@@ -89,6 +97,61 @@ func TestAnUnassignedBuildNodesReportIsNoBuilder(t *testing.T) {
 	if n := len(set.Entries("typescript")); n != 2 {
 		t.Fatalf("novox and shanks are left; got %d entries", n)
 	}
+
+	// 1945 (a part exits at start for a bucket it works without): a bus that refuses the assignments' watch, or every
+	// toolchains bucket, leaves the watches unstarted and the part running, said; a refused bucket it cannot work without
+	// is an error naming the bucket and its grant.
+	var logs bytes.Buffer
+	logger := log.New(&logs, "", 0)
+	refused := refusingBus{err: errors.New("nats: permissions violation for publish to $JS.API.CONSUMER.CREATE.KV_mesh-controller_assignments")}
+	if _, err := watchAssignments(context.Background(), refused, newToolchainGate(nil, set.Put, t.Logf), func(context.Context) {}, logger); err != nil {
+		t.Errorf("a refused watch of the assignments stopped the start: %v", err)
+	}
+	if err := newToolchainWatch(context.Background(), refusingBus{refuseOpen: true, err: refused.err}, set, func(context.Context) {}, logger).start(); err != nil {
+		t.Errorf("a refused toolchains bucket stopped the start: %v", err)
+	}
+	if !strings.Contains(logs.String(), "cannot be watched") || !strings.Contains(logs.String(), "cannot be read") {
+		t.Errorf("the refusals were not said: %q", logs.String())
+	}
+	if _, err := process.EssentialBucket(context.Background(), refusingBus{refuseOpen: true, err: refused.err}, "builder", "mesh-delivery-indexer_modules", "work"); err == nil ||
+		!strings.Contains(err.Error(), "grant of mesh-delivery-indexer_modules") {
+		t.Errorf("a refused bucket the part needs: %v, want an error naming its grant", err)
+	}
+	// The gate's source (the previews' trunk heads): refused at start, the builder goes on, and a candidate asking a trunk
+	// head is told why and opens it again then.
+	src := &trunkSource{ctx: context.Background(), js: refusingBus{refuseOpen: true, err: refused.err}, heads: &sourceHeads{}, errs: make(chan error, 1), logf: logger.Printf}
+	src.start()
+	if _, err := src.head(context.Background(), 7, "novox/mesh-catalog"); err == nil || !strings.Contains(err.Error(), delivery.SourceBucket) {
+		t.Errorf("a trunk head with the source refused: %v, want why", err)
+	}
+	if src.opened {
+		t.Error("a refused source counts as opened, so it is never opened again")
+	}
+}
+
+// refusingBus is a bus that refuses: every open (refuseOpen), or every watch of a bucket it opens, with err.
+type refusingBus struct {
+	refuseOpen bool
+	err        error
+}
+
+func (r refusingBus) KeyValue(context.Context, string) (jetstream.KeyValue, error) {
+	if r.refuseOpen {
+		return nil, r.err
+	}
+	return refusedWatch{err: r.err}, nil
+}
+
+type refusedWatch struct {
+	jetstream.KeyValue
+	err error
+}
+
+func (r refusedWatch) WatchAll(context.Context, ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) {
+	return nil, r.err
+}
+func (r refusedWatch) Watch(context.Context, string, ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) {
+	return nil, r.err
 }
 
 // A build node's report counts from the first one it writes since its assignment: a newly assigned node that has not

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

@@ -4,11 +4,13 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
 	"log"
 	"strings"
 	"sync"
 
+	"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
+	"git.novox.be/novox/mesh-sdk/go/process"
+
 	"github.com/nats-io/nats.go/jetstream"
 
 	"git.novox.be/novox/mesh-sdk/go/delivery"
@@ -62,11 +64,14 @@ func (w *toolchainWatch) ensure(module string) error {
 		w.logger.Printf("the bucket %s does not exist yet: it is watched when %s reports a toolchain", name, module)
 		return nil
 	case err != nil:
-		return fmt.Errorf("the bucket %s: %w", name, err)
+		// Refused or not answered (its grant not on the bus yet, issue 1945): said, and opened again at the module's next report.
+		w.logger.Printf("the bucket %s cannot be read (%v): %s's toolchains are not heard until its next report", name, err, module)
+		return nil
 	}
 	watcher, err := settle(w.ctx, kv, func(e jetstream.KeyValueEntry) { w.put(module, e.Key(), e.Value(), !opOf(e)) })
 	if err != nil {
-		return err
+		process.NotWatched(w.logger.Printf, name, err, fmt.Sprintf("%s's toolchains are not heard until its next report", module))
+		return nil
 	}
 	w.watched[module] = true
 	go func() {

modules/mesh-delivery-builder/go.mod

@@ -3,7 +3,7 @@ module mesh-delivery-builder
 go 1.26.0
 
 require (
-	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024020722-34f210e5eba1
+	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585
 	github.com/nats-io/nats-server/v2 v2.11.17
 	github.com/nats-io/nats.go v1.54.0
 )

modules/mesh-delivery-builder/go.sum

@@ -1,5 +1,7 @@
 git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024020722-34f210e5eba1 h1:fzOkrz9CSp3mEjeV7NfZx8yoivI1sD1AOtIPtyyF8RI=
 git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024020722-34f210e5eba1/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585 h1:7QoFH5qjXRpx5M4uPdYsVJ5qKl2082gq8n0lFxhoNIg=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585/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=

modules/mesh-delivery-builder/internal/build/askagain.go

@@ -0,0 +1,337 @@
+package build
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"regexp"
+	"slices"
+	"strconv"
+	"strings"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/buildstep"
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+// **An ask that can no longer be answered is asked again** (novox/hq issue 1945). A build seat's stream gives an ask up
+// once it was delivered as often as its worker may (delivery.AskMaxDeliver), and a message given up on, dropped or
+// terminated with no step written is answered by no build node: the lane waiting for its step would wait for ever, and on
+// restart take over the same id and wait again. The builder notices it at the check bound (CheckAfter, the step bound),
+// a context's bound and no timer of its own: it asks Gone, which reads the bus, and asks such an ask again under a new
+// id, written into the key, said in its log and as the event build.asked-again, at most Askings times; then the key is
+// failed with why. A person asks the same by hand with the tool builder_ask_again (AskAgain).
+
+// DefaultAskings is how many times a build module's part of a build is asked, the first included, when Builder.Askings is 0.
+const DefaultAskings = 3
+
+// AskedAgainEvent is the event the builder says when it asked an ask again: mesh.mod.mesh-delivery-builder.event.<this>.
+const AskedAgainEvent = "build.asked-again"
+
+// AskedAgain is the body of AskedAgainEvent.
+type AskedAgain struct {
+	Key         string    `json:"key"`
+	Module      string    `json:"module"`
+	FP          string    `json:"fp"`
+	BuildModule string    `json:"build_module"`
+	Language    string    `json:"language"`
+	Old         string    `json:"old"`
+	New         string    `json:"new"`
+	Asking      int       `json:"asking"`
+	Askings     int       `json:"askings"`
+	Why         string    `json:"why"`
+	At          time.Time `json:"at"`
+}
+
+// AskSeen is what the bus says of an ask whose step has not ended, as the builder may read it: its build seat's stream
+// (information and messages by sequence) and its build module's steps. The builder may not read the seat workers'
+// delivery counts or ack floors (granted to the delivery view alone: $JS.API.CONSUMER.INFO.<stream>.<worker>), so a
+// message given up on is told by what the workers did after it: a build node took an ask asked later, of the same kind or
+// of one pulled after it (issue 1945 note 62: seven messages SEAT_NODE_BUILD_GO had given up on sat ahead of a Go ask for
+// five hours, each counted as waiting, so the ask was never asked again).
+type AskSeen struct {
+	Stream string
+	// Kind is the ask's kind, At when it was asked (its id's time).
+	Kind string
+	At   time.Time
+	// Step is whether a build node wrote a step for the ask (running: the step bound of buildrun.Asker judges it).
+	Step bool
+	// OnStream is whether the stream still holds the ask's message; PutAt is when it was put.
+	OnStream bool
+	PutAt    time.Time
+	// Ahead are the messages of its subject before it on the stream.
+	Ahead []Queued
+	// Last is the ask asked last that a build node took (wrote a step for) of the ask's kind or of a kind pulled after
+	// it, when one was taken: every message of the kind asked before it was then delivered already, or given up on.
+	Last *Taken
+}
+
+// Queued is one message of a build seat's stream: its sequence, its ask's id and kind, when the ask was asked (its id's
+// time, the put time when the id says none) and put, and the state of the step a build node wrote for it ("" for none).
+type Queued struct {
+	Seq   uint64    `json:"seq"`
+	Ask   string    `json:"ask"`
+	Kind  string    `json:"kind"`
+	At    time.Time `json:"asked"`
+	PutAt time.Time `json:"put"`
+	Step  string    `json:"step,omitempty"`
+}
+
+// Taken is an ask a build node took: its id, kind and when it was asked, and the step's node and start.
+type Taken struct {
+	Ask     string    `json:"ask"`
+	Kind    string    `json:"kind"`
+	At      time.Time `json:"asked"`
+	Node    string    `json:"node"`
+	Started time.Time `json:"started"`
+}
+
+// Proves is whether a taken ask of kind taken shows that a message of kind waiting asked before it is no longer
+// delivered: a holder pulls the kinds in delivery.AskKinds' order, so taking one of the same kind or of a later one means
+// its worker had none of waiting's kind left to give.
+func Proves(taken, waiting string) bool {
+	t, w := slices.Index(delivery.AskKinds, taken), slices.Index(delivery.AskKinds, waiting)
+	return t >= 0 && w >= 0 && t >= w
+}
+
+// Waits is whether a message ahead still waits to be delivered: no build node wrote a step for it, and no ask of its
+// kind asked after it was taken. A message its worker gave up on (delivered its most times) stays on a work queue until
+// its age, and is not counted.
+func (s AskSeen) Waits(q Queued) bool {
+	return q.Step == "" && !s.passed(q.At)
+}
+
+func (s AskSeen) passed(at time.Time) bool {
+	return s.Last != nil && Proves(s.Last.Kind, s.Kind) && s.Last.At.After(at)
+}
+
+// Waiting is how many messages ahead still wait to be delivered (Waits).
+func (s AskSeen) Waiting() int {
+	n := 0
+	for _, q := range s.Ahead {
+		if s.Waits(q) {
+			n++
+		}
+	}
+	return n
+}
+
+// Why is why the ask can no longer be answered, or "" when it can: it has a step, waits behind others of its kind still
+// deliverable, or no build node took a later ask of its kind (the nodes are busy or paused: a backlog, never a fault,
+// ADR 0525 §9).
+func (s AskSeen) Why() string {
+	switch {
+	case s.Step:
+		return ""
+	case !s.OnStream:
+		return fmt.Sprintf("no build node wrote a step for it and %s no longer holds it: it was given up on or dropped", s.Stream)
+	case s.Waiting() > 0 || !s.passed(s.at()):
+		return ""
+	default:
+		behind := ""
+		if n := len(s.Ahead); n > 0 {
+			behind = fmt.Sprintf(" behind %d given up on or taken", n)
+		}
+		return fmt.Sprintf("it waits on %s since %s%s while %s took the %s ask %s asked later at %s: its stream no longer delivers it (given up after its most deliveries)",
+			s.Stream, s.PutAt.UTC().Format(time.RFC3339), behind, s.Last.Node, s.Last.Kind, s.Last.Ask, s.Last.At.UTC().Format(time.RFC3339))
+	}
+}
+
+func (s AskSeen) at() time.Time {
+	if s.At.IsZero() {
+		return s.PutAt
+	}
+	return s.At
+}
+
+// AskedAt is when an ask id was made (buildstep.NewID: its milliseconds in base 36 before six hex digits), or zero.
+func AskedAt(id string) time.Time {
+	if len(id) <= 6 {
+		return time.Time{}
+	}
+	ms, err := strconv.ParseInt(id[:len(id)-6], 36, 64)
+	if err != nil || ms <= 0 {
+		return time.Time{}
+	}
+	return time.UnixMilli(ms)
+}
+
+// Dropper removes an ask's message from its build seat's stream (Builder.Gone may be one): the old message of an ask asked
+// again, so a message given up on does not sit ahead of every later ask of its kind until its age (issue 1945 note 62).
+type Dropper interface {
+	Drop(ctx context.Context, a delivery.StepAsk) error
+}
+
+// drop removes the old ask's message when Gone can; a refusal is said and changes nothing else: the new ask is put.
+func (b *Builder) drop(ctx context.Context, ask delivery.StepAsk, old string) {
+	d, ok := b.Gone.(Dropper)
+	if !ok {
+		return
+	}
+	prior := ask
+	prior.ID = old
+	if err := d.Drop(ctx, prior); err != nil {
+		b.logf("the message of the ask %s asked again as %s stays on its stream: %v", old, ask.ID, err)
+	}
+}
+
+func (b *Builder) askings() int {
+	if b.Askings > 0 {
+		return b.Askings
+	}
+	return DefaultAskings
+}
+
+var askingWords = regexp.MustCompile(`asked (\S+) again \(asking (\d+)\)`)
+
+// askingOf is the asking a key's step words name for a build module's ask, 1 when none.
+func askingOf(step, buildModule string) int {
+	for _, m := range askingWords.FindAllStringSubmatch(step, -1) {
+		if m[1] == buildModule {
+			if n, err := strconv.Atoi(m[2]); err == nil && n > 0 {
+				return n
+			}
+		}
+	}
+	return 1
+}
+
+// AskAgain asks the waiting asks of a build key again now, under new ids (the tool builder_ask_again): it answers the
+// languages whose lane waits, none when no lane of this builder waits on the key.
+func (b *Builder) AskAgain(key, why string) []string {
+	b.init()
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	var asked []string
+	for language, c := range b.waiting[key] {
+		select {
+		case c <- why:
+		default: // one is already asked
+		}
+		asked = append(asked, language)
+	}
+	return asked
+}
+
+// Waiting are the build keys whose lanes wait for a step now, each with its languages.
+func (b *Builder) Waiting() map[string][]string {
+	b.init()
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	out := map[string][]string{}
+	for key, ls := range b.waiting {
+		for l := range ls {
+			out[key] = append(out[key], l)
+		}
+	}
+	return out
+}
+
+func (b *Builder) listen(key, language string) chan string {
+	b.init()
+	c := make(chan string, 1)
+	b.mu.Lock()
+	if b.waiting[key] == nil {
+		b.waiting[key] = map[string]chan string{}
+	}
+	b.waiting[key][language] = c
+	b.mu.Unlock()
+	return c
+}
+
+func (b *Builder) unlisten(key, language string, c chan string) {
+	b.mu.Lock()
+	if b.waiting[key][language] == c {
+		delete(b.waiting[key], language)
+		if len(b.waiting[key]) == 0 {
+			delete(b.waiting, key)
+		}
+	}
+	b.mu.Unlock()
+}
+
+var (
+	errCheck = errors.New("the check bound passed")
+	errAgain = errors.New("asked again")
+)
+
+type byHand struct{ why string }
+
+func (h byHand) Error() string { return "asked again by hand: " + h.why }
+
+// answer waits for the step of the lane's ask, and asks it again while it can no longer be answered (Gone) or a person
+// asked it again (AskAgain); it answers the step and the ask that has it.
+func (b *Builder) answer(ctx context.Context, l *lane, ask delivery.StepAsk, asked *askNotes, module, fp string) (delivery.Step, delivery.StepAsk, error) {
+	key, _ := delivery.BuilderBuildKey(module, fp)
+	again := b.listen(key, ask.Language)
+	defer b.unlisten(key, ask.Language, again)
+	asking := askingOf(asked.step, l.module)
+	for {
+		step, why, err := b.waitOnce(ctx, ask, again)
+		switch {
+		case errors.Is(err, errCheck):
+			continue
+		case errors.Is(err, errAgain):
+			if asking >= b.askings() {
+				return step, ask, fmt.Errorf("the ask %s could not be answered and was asked %d times, the most it is: %s", ask.ID, asking, why)
+			}
+			asking++
+			old := ask.ID
+			ask.ID, ask.At = buildstep.NewID(), time.Now()
+			if err := asked.note(ctx, ask.Language, ask.ID, fmt.Sprintf("asked %s again (asking %d): %s", l.module, asking, why)); err != nil {
+				return step, ask, fmt.Errorf("the ask of %s asked again could not be written to the key: %w", l.module, err)
+			}
+			if err := b.Asks.Put(ctx, ask); err != nil {
+				return step, ask, err
+			}
+			b.drop(ctx, ask, old)
+			b.logf("%s: the ask %s of %s is asked again as %s (asking %d of %d): %s", key, old, l.module, ask.ID, asking, b.askings(), why)
+			if b.Said != nil {
+				b.Said(AskedAgain{Key: key, Module: module, FP: fp, BuildModule: l.module, Language: ask.Language, Old: old, New: ask.ID,
+					Asking: asking, Askings: b.askings(), Why: why, At: ask.At})
+			}
+		default:
+			return step, ask, err
+		}
+	}
+}
+
+// waitOnce waits for the ask's step until it ends, a person asks it again, or the check bound passes: then it asks Gone
+// and answers errAgain with why when the ask can no longer be answered, errCheck when it waits on.
+func (b *Builder) waitOnce(ctx context.Context, ask delivery.StepAsk, again <-chan string) (delivery.Step, string, error) {
+	hand, cancel := context.WithCancelCause(ctx)
+	defer cancel(nil)
+	go func() {
+		select {
+		case why := <-again:
+			cancel(byHand{why})
+		case <-hand.Done():
+		}
+	}()
+	wait := hand
+	if b.Gone != nil && b.CheckAfter > 0 {
+		var stop context.CancelFunc
+		wait, stop = context.WithTimeoutCause(hand, b.CheckAfter, errCheck)
+		defer stop()
+	}
+	step, err := b.Asks.Wait(wait, ask)
+	if err == nil || ctx.Err() != nil {
+		return step, "", err
+	}
+	var h byHand
+	if errors.As(context.Cause(hand), &h) {
+		return step, strings.TrimSpace("asked again by hand: " + h.why), errAgain
+	}
+	if errors.Is(context.Cause(wait), errCheck) {
+		why, gerr := b.Gone.Gone(ctx, ask)
+		if gerr != nil {
+			b.logf("whether the ask %s can still be answered cannot be read, so it is waited for: %v", ask.ID, gerr)
+			return step, "", errCheck
+		}
+		if why == "" {
+			return step, "", errCheck
+		}
+		return step, why, errAgain
+	}
+	return step, "", err
+}

modules/mesh-delivery-builder/internal/build/queue.go

@@ -0,0 +1,115 @@
+package build
+
+import (
+	"encoding/json"
+	"fmt"
+	"strings"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+// **The tool builder_seat_queue reads a build seat's stream as the builder reads it** (novox/hq issue 1945 note 62; the
+// instruction "tools are how a module is spoken to"): the messages waiting on SEAT_NODE_BUILD_<LANGUAGE>, bounded, each
+// with its ask, its age, the state of the step a build node wrote for it, and whether the builder counts it as still
+// waiting (AskSeen.Waits) or as given up on. Only the builder's own account may read the stream (issue 1910): the tools
+// bundle writes the read into the builder's state queues, keyed by the language, which the builder watches and answers
+// there, as builder_ask_again does through reasks.
+const (
+	SeatQueueTool = "builder_seat_queue"
+	// QueuesState is the builder's state of queue reads: the bucket mesh-delivery-builder_queues, keyed by the language.
+	QueuesState  = "queues"
+	QueuesBucket = delivery.BuilderModule + "_" + QueuesState
+	// DefaultQueueLimit and MaxQueueLimit bound the messages read of each kind.
+	DefaultQueueLimit = 50
+	MaxQueueLimit     = 500
+)
+
+// QueueNote says what the read cannot show.
+const QueueNote = "the builder may not read the seat workers' delivery counts or ack floors ($JS.API.CONSUMER.INFO.<stream>.<worker> is " +
+	"granted to the delivery view alone): a message counts as given up on when no node wrote its step and a node took an ask of its kind " +
+	"asked after it; nats.nats_streams shows the workers' pending and waiting pulls"
+
+// QueueAsk is a value of the queues state: a read asked (limit, who, when) and, once the builder read the stream, its answer.
+type QueueAsk struct {
+	V        int        `json:"v"`
+	Limit    int        `json:"limit"`
+	By       string     `json:"by,omitempty"`
+	At       time.Time  `json:"at"`
+	Answered time.Time  `json:"answered,omitzero"`
+	Error    string     `json:"error,omitempty"`
+	Queue    *SeatQueue `json:"queue,omitempty"`
+}
+
+// SeatQueue is what the builder read of one build seat's stream.
+type SeatQueue struct {
+	Language string      `json:"language"`
+	Stream   string      `json:"stream"`
+	Messages uint64      `json:"messages"`
+	FirstSeq uint64      `json:"first_seq"`
+	LastSeq  uint64      `json:"last_seq"`
+	Kinds    []KindQueue `json:"kinds"`
+	Note     string      `json:"note"`
+}
+
+// KindQueue is one kind's messages on the stream, oldest first: how many the builder counts as waiting, the ask asked
+// last a build node took of the kind (or of one pulled after it), and each message.
+type KindQueue struct {
+	Kind     string        `json:"kind"`
+	Read     int           `json:"read"`
+	Waiting  int           `json:"waiting"`
+	GivenUp  int           `json:"given_up_or_taken"`
+	Last     *Taken        `json:"last_taken,omitempty"`
+	Messages []QueuedState `json:"messages"`
+}
+
+// QueuedState is one message as the tool shows it.
+type QueuedState struct {
+	Queued
+	AgeSeconds int64 `json:"age_s"`
+	Waits      bool  `json:"waits"`
+}
+
+// NewKindQueue is seen's messages ahead as one kind's queue at now.
+func NewKindQueue(kind string, seen AskSeen, now time.Time) KindQueue {
+	k := KindQueue{Kind: kind, Read: len(seen.Ahead), Last: seen.Last}
+	for _, q := range seen.Ahead {
+		w := seen.Waits(q)
+		if w {
+			k.Waiting++
+		} else {
+			k.GivenUp++
+		}
+		k.Messages = append(k.Messages, QueuedState{Queued: q, AgeSeconds: int64(now.Sub(q.PutAt) / time.Second), Waits: w})
+	}
+	return k
+}
+
+// NewQueueAsk is the value builder_seat_queue writes for a language: its key and the value.
+func NewQueueAsk(language string, limit int, by string, at time.Time) (string, []byte, error) {
+	if _, err := delivery.AskStream(language); err != nil {
+		return "", nil, fmt.Errorf("language: %w", err)
+	}
+	if limit <= 0 {
+		limit = DefaultQueueLimit
+	}
+	data, err := json.Marshal(QueueAsk{V: 1, Limit: min(limit, MaxQueueLimit), By: by, At: at.UTC()})
+	return language, data, err
+}
+
+// DecodeQueueAsk reads a queues value.
+func DecodeQueueAsk(data []byte) (QueueAsk, error) {
+	var a QueueAsk
+	if err := json.Unmarshal(data, &a); err != nil {
+		return a, err
+	}
+	if a.V != 1 {
+		return a, fmt.Errorf("a queues value of version %d, not 1", a.V)
+	}
+	return a, nil
+}
+
+// LanguageOf is the language a tool's seat or language argument names: go, node or image, or node-build-<language>.
+func LanguageOf(arg string) string {
+	return strings.TrimPrefix(strings.TrimSpace(arg), "node-build-")
+}

modules/mesh-delivery-builder/internal/build/run.go

@@ -56,6 +56,18 @@ type Builder struct {
 	// Trunk answers the head of a repository in the gate's source revision rev (source.main at that revision): the trunk a
 	// candidate's preview is merged onto (Candidate.Base). Nil builds no candidate's fingerprint, and says why.
 	Trunk func(ctx context.Context, rev uint64, repository string) (string, error)
+	// Gone says why an ask with no ended step can no longer be answered, or "" when it can (issue 1945): asked once the
+	// ask has waited CheckAfter, and again each CheckAfter it goes on waiting. An ask that cannot be answered is asked
+	// again under a new id, at most Askings times. Nil, or CheckAfter 0, never asks it.
+	Gone interface {
+		Gone(ctx context.Context, a delivery.StepAsk) (string, error)
+	}
+	CheckAfter time.Duration
+	// Askings is how many times one build module's part of a build is asked, the first included, before its key is failed
+	// with why; DefaultAskings when 0.
+	Askings int
+	// Said is told each ask asked again (the builder's event build.asked-again). Nil tells nobody.
+	Said func(AskedAgain)
 
 	once     sync.Once
 	sem      chan struct{}
@@ -75,6 +87,8 @@ type Builder struct {
 	// held are the done keys' references the store answered held, by `<key> <references>`, asked once per process: a
 	// blob collected later is found by the take-in, whose artifacts-gone forgets the key here (issue 1937).
 	held map[string]bool
+	// waiting are the lanes that wait for an ask's step, by build key and language: AskAgain asks through them.
+	waiting map[string]map[string]chan string
 }
 
 type planned struct {
@@ -91,7 +105,7 @@ func (b *Builder) init() {
 		b.sem = make(chan struct{}, limit)
 		b.cands, b.via, b.trunkFP = map[string]delivery.Candidate{}, map[string]via{}, map[string]bool{}
 		b.known, b.inflight, b.again, b.plans, b.tried = map[string]delivery.Module{}, map[string]bool{}, map[string]delivery.Module{}, map[string]planned{}, map[string]bool{}
-		b.published, b.held = map[string]bool{}, map[string]bool{}
+		b.published, b.held, b.waiting = map[string]bool{}, map[string]bool{}, map[string]map[string]chan string{}
 	})
 }
 
@@ -673,12 +687,14 @@ type askNotes struct {
 	b       *Builder
 	cl      *Claim
 	resumed map[string]string
+	// step is the step words of a key taken over at start: an ask asked again says its asking there (askingOf).
+	step string
 }
 
 func (b *Builder) notes(cl *Claim) *askNotes {
 	n := &askNotes{b: b, cl: cl}
 	if cl.Build.State == delivery.BuildBuilding && len(cl.Build.Asks) > 0 {
-		n.resumed = maps.Clone(cl.Build.Asks)
+		n.resumed, n.step = maps.Clone(cl.Build.Asks), cl.Build.Step
 	}
 	return n
 }
@@ -776,7 +792,7 @@ func (b *Builder) runLane(ctx context.Context, l *lane, repository string, m del
 			return
 		}
 	}
-	step, err := b.Asks.Wait(ctx, ask)
+	step, ask, err := b.answer(ctx, l, ask, asked, module, m.FP2)
 	if err != nil {
 		l.err = fmt.Errorf("%s: %w", l.module, err)
 		return

modules/mesh-delivery-builder/internal/build/run_test.go

@@ -6,8 +6,10 @@ import (
 	"errors"
 	"fmt"
 	"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
+	"strconv"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -616,27 +618,286 @@ func TestAFailedPublishOfACandidatesImageIsWrittenToTheKey(t *testing.T) { // re
 	}
 }
 
-// Review round 2 item 1: a build ask that waits in its stream longer than the step bound is not failed and not asked
-// again; the key is done when a node takes it.
-func TestABuildAskWaitingInItsStreamIsNotBounded(t *testing.T) {
-	r := newRig(t, goBundle)
-	r.asks.Answer = nil
-	r.b.Asks = buildrun.Asker{Publisher: r.asks, Waiter: r.asks, Bound: 20 * time.Millisecond}
-	r.b.Module(context.Background(), "m", moduleValue(t, "modules/m"))
-	for deadline := time.Now().Add(2 * time.Second); len(r.asks.Asks()) == 0; time.Sleep(time.Millisecond) {
+// goneAsks answers Gone from a function of the ask, and counts how often it was asked; it keeps the ids it was asked to
+// drop from their stream (Dropper).
+type goneAsks struct {
+	mu      sync.Mutex
+	n       int
+	why     func(a delivery.StepAsk) string
+	dropped []string
+}
+
+func (g *goneAsks) Drop(_ context.Context, a delivery.StepAsk) error {
+	g.mu.Lock()
+	defer g.mu.Unlock()
+	g.dropped = append(g.dropped, a.ID)
+	return nil
+}
+
+func (g *goneAsks) Gone(_ context.Context, a delivery.StepAsk) (string, error) {
+	g.mu.Lock()
+	g.n++
+	g.mu.Unlock()
+	return g.why(a), nil
+}
+
+func (g *goneAsks) asked() int {
+	g.mu.Lock()
+	defer g.mu.Unlock()
+	return g.n
+}
+
+// waitForAsks waits until n asks were put.
+func waitForAsks(t *testing.T, r *rig, n int) []delivery.StepAsk {
+	t.Helper()
+	for deadline := time.Now().Add(2 * time.Second); len(r.asks.Asks()) < n; time.Sleep(time.Millisecond) {
 		if time.Now().After(deadline) {
-			t.Fatal("no ask was put")
+			t.Fatalf("%d asks put, want %d", len(r.asks.Asks()), n)
 		}
 	}
-	time.Sleep(100 * time.Millisecond)
-	if v := valueOf(t, r.st, r.key); v.State != delivery.BuildBuilding {
-		t.Fatalf("the key = %+v while its ask waits, want building", v)
-	}
-	r.asks.End(r.answer(r.asks.Asks()[0]))
-	r.b.Wait()
-	if v := valueOf(t, r.st, r.key); v.State != delivery.BuildDone || len(r.asks.Asks()) != 1 {
-		t.Fatalf("the key = %+v after %d asks, want done from the one ask", v, len(r.asks.Asks()))
+	return r.asks.Asks()
+}
+
+// Review round 2 item 1: a build ask that waits in its stream longer than the step bound is not failed and not asked
+// again; the key is done when a node takes it. Issue 1945: an ask that can no longer be answered (its stream gave it up)
+// is asked again under a new id, said, at most Askings times; a person asks it again by hand the same way.
+func TestABuildAskWaitingInItsStreamIsNotBounded(t *testing.T) {
+	t.Run("waits in its stream: never asked again", func(t *testing.T) {
+		r := newRig(t, goBundle)
+		r.asks.Answer = nil
+		r.b.Asks = buildrun.Asker{Publisher: r.asks, Waiter: r.asks, Bound: 20 * time.Millisecond}
+		gone := &goneAsks{why: func(delivery.StepAsk) string { return "" }}
+		r.b.Gone, r.b.CheckAfter = gone, 5*time.Millisecond
+		r.b.Module(context.Background(), "m", moduleValue(t, "modules/m"))
+		waitForAsks(t, r, 1)
+		time.Sleep(100 * time.Millisecond)
+		if v := valueOf(t, r.st, r.key); v.State != delivery.BuildBuilding {
+			t.Fatalf("the key = %+v while its ask waits, want building", v)
+		}
+		if gone.asked() == 0 {
+			t.Fatal("whether the ask can still be answered was never read at the check bound") // 1945
+		}
+		r.asks.End(r.answer(r.asks.Asks()[0]))
+		r.b.Wait()
+		if v := valueOf(t, r.st, r.key); v.State != delivery.BuildDone || len(r.asks.Asks()) != 1 {
+			t.Fatalf("the key = %+v after %d asks, want done from the one ask", v, len(r.asks.Asks()))
+		}
+	})
+	for _, c := range []struct {
+		name    string
+		askings int
+		// step is the key's step words at start when it names an ask (a key taken over), "" for a fresh build.
+		step   string
+		byHand bool
+		always bool // Gone says every ask cannot be answered, not only the first
+		puts   int
+		done   bool
+		asking int // the asking the event says, 0 when none is said
+	}{
+		{name: "given up by its stream: asked again under a new id and done from it", puts: 2, done: true, asking: 2},
+		{name: "asked again by hand: the same, with no check bound", byHand: true, puts: 2, done: true, asking: 2},
+		{name: "never answerable: failed after its askings", askings: 2, always: true, puts: 2},
+		{name: "taken over at its last asking: failed, not asked again", askings: 3, always: true, step: "asked mesh-build-go again (asking 3): x", puts: 0},
+	} {
+		t.Run(c.name, func(t *testing.T) { // 1945
+			r := newRig(t, goBundle)
+			r.asks.Answer = nil
+			r.b.Askings = c.askings
+			var said []AskedAgain
+			var mu sync.Mutex
+			r.b.Said = func(e AskedAgain) { mu.Lock(); said = append(said, e); mu.Unlock() }
+			first := ""
+			if c.step != "" {
+				first = "mvl0frv18d1960"
+				data, _ := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildBuilding, By: "novox.builder", Manifest: manifestD,
+					Step: c.step, Asks: map[string]string{delivery.AskGo: first}})
+				r.st.vals[r.key] = data
+			}
+			gone := &goneAsks{}
+			if !c.byHand {
+				gone.why = func(a delivery.StepAsk) string {
+					if c.always || a.ID == first || first == "" {
+						return "it waits first in line while a node took a later ask"
+					}
+					return ""
+				}
+				r.b.Gone, r.b.CheckAfter = gone, 5*time.Millisecond
+			}
+			r.b.Module(context.Background(), "m", moduleValue(t, "modules/m"))
+			if c.byHand {
+				waitForAsks(t, r, 1)
+				for deadline := time.Now().Add(2 * time.Second); len(r.b.AskAgain(r.key, "stuck since 19:06")) == 0; time.Sleep(time.Millisecond) {
+					if time.Now().After(deadline) {
+						t.Fatal("no lane of the key waits")
+					}
+				}
+			}
+			if c.done {
+				asks := waitForAsks(t, r, c.puts)
+				if first == "" && !c.byHand {
+					first = asks[0].ID
+				}
+				last := asks[len(asks)-1]
+				if v := valueOf(t, r.st, r.key); v.Asks[delivery.AskGo] != last.ID || !strings.Contains(v.Step, "again (asking 2)") {
+					t.Fatalf("the key = %+v, want building naming the ask asked again %s", v, last.ID)
+				}
+				r.asks.End(r.answer(last))
+			}
+			r.b.Wait()
+			v := valueOf(t, r.st, r.key)
+			if n := len(r.asks.Asks()); n != c.puts {
+				t.Errorf("%d asks put, want %d", n, c.puts)
+			}
+			if c.done && v.State != delivery.BuildDone {
+				t.Errorf("the key = %+v, want done from the ask asked again", v)
+			}
+			if !c.done && (v.State != delivery.BuildFailed || !strings.Contains(v.Error, "could not be answered and was asked")) {
+				t.Errorf("the key = %+v, want failed saying the ask could not be answered", v)
+			}
+			mu.Lock()
+			defer mu.Unlock()
+			// Note 62: the old message of an ask asked again because its stream gave it up is dropped from the stream.
+			if gone.mu.Lock(); c.done && !c.byHand && (len(gone.dropped) != 1 || gone.dropped[0] != first) {
+				t.Errorf("dropped %v from the stream, want the old ask %s", gone.dropped, first)
+			}
+			gone.mu.Unlock()
+			if c.asking > 0 && (len(said) != 1 || said[0].Asking != c.asking || said[0].New == said[0].Old || said[0].Key != r.key || said[0].Why == "") {
+				t.Errorf("said %+v, want once at asking %d", said, c.asking)
+			}
+		})
 	}
+	t.Run("what the bus says of an ask", func(t *testing.T) { // 1945
+		put := time.Date(2026, 10, 25, 17, 6, 0, 0, time.UTC)
+		taken := func(kind string, at time.Time) *Taken {
+			return &Taken{Ask: "later", Kind: kind, At: at, Node: "g14", Started: at}
+		}
+		later := taken(delivery.AskDelivery, put.Add(time.Minute))
+		// ahead is n messages put before the ask, a minute apart, each with step (its state, "" for none).
+		ahead := func(n int, step string) []Queued {
+			var qs []Queued
+			for i := range n {
+				at := put.Add(-time.Duration(n-i) * time.Minute)
+				qs = append(qs, Queued{Seq: uint64(i + 1), Ask: fmt.Sprint("a", i), Kind: delivery.AskDelivery, At: at, PutAt: at, Step: step})
+			}
+			return qs
+		}
+		seen := func(qs []Queued, last *Taken) AskSeen {
+			return AskSeen{Stream: "S", Kind: delivery.AskDelivery, At: put, OnStream: true, PutAt: put, Ahead: qs, Last: last}
+		}
+		for _, c := range []struct {
+			name    string
+			seen    AskSeen
+			want    string // a part of the why, "" when it can be answered
+			waiting int
+		}{
+			{"a node took it", AskSeen{Stream: "S", Step: true}, "", 0},
+			{"no longer on its stream", AskSeen{Stream: "S"}, "no longer holds it", 0},
+			{"first in line, a later ask taken", seen(nil, later), "no longer delivers it", 0},
+			{"first in line, nothing taken since", seen(nil, nil), "", 0},
+			{"first in line, the ask taken last was asked before it", seen(nil, taken(delivery.AskDelivery, put.Add(-time.Second))), "", 0},
+			{"two wait ahead, nothing taken since", seen(ahead(2, ""), nil), "", 2},
+			{"two ahead passed over, the ask not yet", seen(ahead(2, ""), taken(delivery.AskDelivery, put.Add(-time.Second))), "", 0},
+			// Note 62: seven messages SEAT_NODE_BUILD_GO gave up on sat ahead of a Go ask for five hours.
+			{"seven given up on ahead, a later ask taken", seen(ahead(7, ""), later), "behind 7 given up on or taken", 0},
+			{"seven taken ahead (failed as poison), a later ask taken", seen(ahead(7, delivery.StepFailed), later), "no longer delivers it", 0},
+			{"a check ask taken later proves a delivery ask passed", seen(ahead(7, ""), taken(delivery.AskCheck, put.Add(time.Minute))), "no longer delivers it", 0},
+			{"a delivery ask taken later does not prove a check ask passed", func() AskSeen { s := seen(ahead(3, ""), later); s.Kind = delivery.AskCheck; return s }(), "", 3},
+		} {
+			got := c.seen.Why()
+			if (c.want == "") != (got == "") || !strings.Contains(got, c.want) || c.seen.Waiting() != c.waiting {
+				t.Errorf("%s: why %q and %d waiting, want %q and %d", c.name, got, c.seen.Waiting(), c.want, c.waiting)
+			}
+		}
+		if at := time.UnixMilli(1793120000123); !AskedAt(strconv.FormatInt(at.UnixMilli(), 36)+"0a1b2c").Equal(at) || !AskedAt("x").IsZero() {
+			t.Errorf("AskedAt does not read an ask id's time")
+		}
+	})
+	t.Run("the tool builder_stuck", func(t *testing.T) { // 1945
+		at := time.Date(2026, 10, 25, 17, 6, 0, 0, time.UTC)
+		state := func(vals map[string]string) StateRead {
+			return StateRead{Keys: func() ([]string, error) {
+				var ks []string
+				for k := range vals {
+					ks = append(ks, k)
+				}
+				return ks, nil
+			}, Get: func(k string) (json.RawMessage, time.Time, error) {
+				if v, ok := vals[k]; ok {
+					return json.RawMessage(v), at, nil
+				}
+				return nil, time.Time{}, nil
+			}}
+		}
+		key, _ := delivery.BuilderBuildKey("m", fp1)
+		done, _ := delivery.BuilderBuildKey("d", fp1)
+		building, _ := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildBuilding, By: "novox.builder", Manifest: manifestD,
+			Step: "asked mesh-build-go again (asking 2): x", Asks: map[string]string{delivery.AskGo: "mvo53jcm70b5f7", delivery.AskImage: "mvo53jcm70b5f8"}})
+		finished, _ := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildDone, Manifest: manifestD, Step: "built"})
+		running, _ := json.Marshal(delivery.Step{V: delivery.Version, ID: "mvo53jcm70b5f8", Tool: "delivery ask", Node: "g14", State: delivery.StepRunning, Started: at})
+		_, reask, _ := NewReAsk("m", fp1, "stuck", "shanks.lead", at)
+		r := StuckReader{Builds: state(map[string]string{key: string(building), done: string(finished)}), ReAsks: state(map[string]string{key: string(reask)}),
+			Steps: map[string]StateRead{buildrun.GoBuildModule: state(map[string]string{}), ImageBuildModule: state(map[string]string{"g14.mvo53jcm70b5f8": string(running)})},
+			Now:   func() time.Time { return at.Add(time.Hour) }}
+		out, err := r.Stuck("")
+		if err != nil {
+			t.Fatal(err)
+		}
+		stuck := out["stuck"].([]Stuck)
+		if len(stuck) != 1 || stuck[0].Key != key || stuck[0].Age != "1h0m0s" || stuck[0].NoneStep != 1 || stuck[0].ReAsk == nil || stuck[0].ReAsk.Why != "stuck" {
+			t.Fatalf("stuck = %+v, want the building key alone, an hour old, one ask with no step, with its ask again", stuck)
+		}
+		a := stuck[0].Asks
+		if len(a) != 2 || a[0].Language != delivery.AskGo || a[0].Asking != 2 || a[0].Step != "none" || a[0].Asked.IsZero() ||
+			a[1].Asking != 1 || !strings.HasPrefix(a[1].Step, "running on g14") {
+			t.Fatalf("asks = %+v, want go at asking 2 with no step and image at asking 1 running on g14", a)
+		}
+
+		// 1827: on the live shape (3000 build keys over 100 modules, 1 ms a Get, three keys of each module building) a module reads only its
+		// own keys, the whole answer reads MaxReads values in parallel and says it was cut, and the history (Written) is asked
+		// only for the rows kept.
+		var gets, written, inFlight, maxFlight atomic.Int64
+		vals := map[string]string{}
+		for i := 0; i < 3000; i++ {
+			k, _ := delivery.BuilderBuildKey(fmt.Sprintf("mod%02d", i%100), fmt.Sprintf("src2:%064x", i))
+			vals[k] = string(finished)
+			if (i/100)%10 == 0 {
+				vals[k] = string(building)
+			}
+		}
+		slow := state(vals)
+		get := slow.Get
+		slow.Get = func(k string) (json.RawMessage, time.Time, error) {
+			gets.Add(1)
+			n := inFlight.Add(1)
+			for m := maxFlight.Load(); n > m && !maxFlight.CompareAndSwap(m, n); m = maxFlight.Load() {
+			}
+			time.Sleep(time.Millisecond)
+			inFlight.Add(-1)
+			raw, _, err := get(k)
+			return raw, time.Time{}, err
+		}
+		slow.Written = func(string) time.Time { written.Add(1); return at }
+		for _, row := range []struct {
+			module string
+			stuck  int
+			gets   int64
+			cut    bool
+		}{{"mod10", 3, 30, false}, {"", 100, MaxReads, true}} {
+			gets.Store(0)
+			written.Store(0)
+			out, err := StuckReader{Builds: slow, ReAsks: state(map[string]string{}), Steps: map[string]StateRead{}, Now: func() time.Time { return at }}.Stuck(row.module)
+			if err != nil {
+				t.Fatal(err)
+			}
+			_, cut := out["cut"]
+			if n := len(out["stuck"].([]Stuck)); n != row.stuck || cut != row.cut || written.Load() != int64(n) || gets.Load() != row.gets {
+				t.Errorf("%q: %d stuck, %d gets, %d written, cut %v; want %d stuck and gets, written for each, cut %v", row.module, n, gets.Load(), written.Load(), cut, row.stuck, row.cut)
+			}
+		}
+		if maxFlight.Load() < 2 {
+			t.Errorf("the values were read one at a time")
+		}
+	})
 }
 
 // heldAsks says an ask is held when its id is among held.

modules/mesh-delivery-builder/internal/build/stuck.go

@@ -0,0 +1,290 @@
+package build
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"sort"
+	"strconv"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+	"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
+)
+
+// The builder's tools (novox/hq issue 1945; the instruction "tools are how a module is spoken to"): builder_stuck reads
+// every key that waits on an ask, builder_ask_again asks a key's build again now. The tools bundle serves them through
+// the tool runner; the builder's process serves no tool (ADR 0437 §4), so builder_ask_again writes the ask into the
+// builder's state reasks, which the builder watches and answers there.
+const (
+	StuckTool    = "builder_stuck"
+	AskAgainTool = "builder_ask_again"
+	// ReAsksState is the builder's state of asks to ask again: the bucket mesh-delivery-builder_reasks, keyed as the build key.
+	ReAsksState  = "reasks"
+	ReAsksBucket = delivery.BuilderModule + "_" + ReAsksState
+)
+
+// ReAsk is a value of the reasks state: why a person asked the key's build again, who and when, and, once the builder
+// read it, its answer.
+type ReAsk struct {
+	V        int       `json:"v"`
+	Why      string    `json:"why"`
+	By       string    `json:"by,omitempty"`
+	At       time.Time `json:"at"`
+	Answer   string    `json:"answer,omitempty"`
+	Answered time.Time `json:"answered,omitzero"`
+}
+
+// DecodeReAsk reads a reasks value.
+func DecodeReAsk(data []byte) (ReAsk, error) {
+	var r ReAsk
+	if err := json.Unmarshal(data, &r); err != nil {
+		return r, err
+	}
+	if r.V != 1 {
+		return r, fmt.Errorf("a reasks value of version %d, not 1", r.V)
+	}
+	return r, nil
+}
+
+// NewReAsk is the value builder_ask_again writes for a module's fingerprint: its key and the value.
+func NewReAsk(module, fp, why, by string, at time.Time) (string, []byte, error) {
+	key, err := delivery.BuilderBuildKey(module, fp)
+	if err != nil {
+		return "", nil, err
+	}
+	if strings.TrimSpace(why) == "" {
+		return "", nil, errors.New("say why the build is asked again (why)")
+	}
+	data, err := json.Marshal(ReAsk{V: 1, Why: strings.TrimSpace(why), By: by, At: at.UTC()})
+	return key, data, err
+}
+
+// StateRead is a state the tools read: its keys, and a key's value with when it was written (zero when not known).
+type StateRead struct {
+	Keys func() ([]string, error)
+	Get  func(key string) (json.RawMessage, time.Time, error)
+	// Written is when the key's current value was written, from the state's history: one more round trip, so the tools ask it
+	// only for a row they keep, and only when Get did not say.
+	Written func(key string) time.Time
+}
+
+// StuckAsk is one ask a key waits on.
+type StuckAsk struct {
+	Language    string `json:"language"`
+	BuildModule string `json:"build_module"`
+	ID          string `json:"id"`
+	Asking      int    `json:"asking"`
+	// Asked is when the ask was put, read from its id; Step is what its build module's steps hold of it: none, or its
+	// state on its node since its start.
+	Asked time.Time `json:"asked,omitzero"`
+	Step  string    `json:"step"`
+}
+
+// Stuck is one build key that waits on an ask.
+type Stuck struct {
+	Key      string     `json:"key"`
+	Module   string     `json:"module"`
+	FP       string     `json:"fp"`
+	State    string     `json:"state"`
+	Step     string     `json:"step"`
+	By       string     `json:"by"`
+	Since    time.Time  `json:"since,omitzero"`
+	Age      string     `json:"age,omitempty"`
+	Asks     []StuckAsk `json:"asks"`
+	ReAsk    *ReAsk     `json:"reask,omitempty"`
+	Unasked  bool       `json:"unasked,omitempty"`
+	NoneStep int        `json:"asks_with_no_step"`
+}
+
+// StuckReader reads what builder_stuck answers.
+type StuckReader struct {
+	Builds StateRead
+	ReAsks StateRead
+	// Steps are the build modules' steps states, by build module.
+	Steps map[string]StateRead
+	Now   func() time.Time
+	// Logf, when set, gets the one line a call logs: the values it read and the time it took.
+	Logf func(format string, args ...any)
+}
+
+// ReadWorkers is how many values builder_stuck asks the bus for at once, and MaxReads how many it reads in a call (issue 1827:
+// every value is a round trip through the runtime, so one after the other a few thousand keys outlast the call's 30 s).
+// A copy of the delivery view's and the checker's: a few lines are not worth a group delivery through mesh-sdk.
+const (
+	ReadWorkers = 16
+	MaxReads    = 1000
+)
+
+// parallel runs read(i) for i in [0, n), at most ReadWorkers at a time, and answers the first error.
+func parallel(n int, read func(i int) error) error {
+	var (
+		next   atomic.Int64
+		failed atomic.Bool
+		once   sync.Once
+		first  error
+		wg     sync.WaitGroup
+	)
+	for w := 0; w < min(ReadWorkers, n); w++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for !failed.Load() {
+				i := int(next.Add(1) - 1)
+				if i >= n {
+					return
+				}
+				if err := read(i); err != nil {
+					once.Do(func() { first = err })
+					failed.Store(true)
+				}
+			}
+		}()
+	}
+	wg.Wait()
+	return first
+}
+
+// Stuck is every key building or queued with its asks, oldest first; with module, that module's only.
+func (r StuckReader) Stuck(module string) (map[string]any, error) {
+	started := time.Now()
+	reads := 0
+	defer func() {
+		if r.Logf != nil {
+			r.Logf("builder_stuck read %d values in %s", reads, time.Since(started).Round(time.Millisecond))
+		}
+	}()
+	keys, err := r.Builds.Keys()
+	if err != nil {
+		return nil, err
+	}
+	now := time.Now
+	if r.Now != nil {
+		now = r.Now
+	}
+	// Steps are found by the ask's id, which their key ends in: one index per build module, no value read until an ask names it.
+	stepKeys := map[string]map[string]string{}
+	for bm, s := range r.Steps {
+		ks, err := s.Keys()
+		if err != nil {
+			return nil, fmt.Errorf("the steps of %s: %w", bm, err)
+		}
+		byID := make(map[string]string, len(ks))
+		for _, k := range ks {
+			if _, id, err := delivery.ParseStepKey(k); err == nil {
+				byID[id] = k
+			}
+		}
+		stepKeys[bm] = byID
+	}
+	// Narrowed by the key (the module) before any value is read; the state is in the value, so every matching key is read, in
+	// parallel, at most MaxReads of them.
+	type parsed struct{ key, mod, fp string }
+	var matching []parsed
+	for _, key := range keys {
+		mod, fp, err := delivery.ParseBuilderBuildKey(key)
+		if err != nil || (module != "" && mod != module) {
+			continue
+		}
+		matching = append(matching, parsed{key, mod, fp})
+	}
+	sort.Slice(matching, func(i, j int) bool { return matching[i].key < matching[j].key })
+	cut := len(matching) > MaxReads
+	total := len(matching)
+	if cut {
+		matching = matching[:MaxReads]
+	}
+	values := make([]json.RawMessage, len(matching))
+	ats := make([]time.Time, len(matching))
+	if err := parallel(len(matching), func(i int) error {
+		raw, at, err := r.Builds.Get(matching[i].key)
+		if err == nil {
+			values[i], ats[i] = raw, at
+		}
+		return nil
+	}); err != nil {
+		return nil, err
+	}
+	reads += len(matching)
+	out := []Stuck{}
+	for i, m := range matching {
+		key, mod, fp := m.key, m.mod, m.fp
+		raw, at := values[i], ats[i]
+		if raw == nil {
+			continue
+		}
+		v, err := delivery.DecodeBuilderBuild(raw)
+		if err != nil || (v.State != delivery.BuildBuilding && v.State != delivery.BuildQueued) {
+			continue
+		}
+		if at.IsZero() && r.Builds.Written != nil {
+			at = r.Builds.Written(key)
+		}
+		s := Stuck{Key: key, Module: mod, FP: fp, State: v.State, Step: v.Step, By: v.By, Since: at, Asks: []StuckAsk{}}
+		if !at.IsZero() {
+			s.Age = now().Sub(at).Round(time.Second).String()
+		}
+		for language, id := range v.Asks {
+			bm, _ := delivery.BuildModuleOf(language)
+			a := StuckAsk{Language: language, BuildModule: bm, ID: id, Asking: askingOf(v.Step, bm), Step: "none"}
+			if t, ok := idTime(id); ok {
+				a.Asked = t
+			}
+			if st, ok := r.Steps[bm]; ok {
+				if k, ok := stepKeys[bm][id]; ok {
+					reads++
+					if raw, _, err := st.Get(k); err == nil && raw != nil {
+						if step, err := delivery.DecodeStep(raw); err == nil {
+							a.Step = fmt.Sprintf("%s on %s since %s", step.State, step.Node, step.Started.UTC().Format(time.RFC3339))
+						}
+					}
+				}
+			}
+			if a.Step == "none" {
+				s.NoneStep++
+			}
+			s.Asks = append(s.Asks, a)
+		}
+		sort.Slice(s.Asks, func(i, j int) bool { return s.Asks[i].Language < s.Asks[j].Language })
+		s.Unasked = v.State == delivery.BuildBuilding && len(v.Asks) == 0
+		if r.ReAsks.Get != nil {
+			reads++
+			if raw, _, err := r.ReAsks.Get(key); err == nil && raw != nil {
+				if ra, err := DecodeReAsk(raw); err == nil {
+					s.ReAsk = &ra
+				}
+			}
+		}
+		out = append(out, s)
+	}
+	sort.Slice(out, func(i, j int) bool {
+		if !out[i].Since.Equal(out[j].Since) {
+			return out[i].Since.Before(out[j].Since)
+		}
+		return out[i].Key < out[j].Key
+	})
+	answer := map[string]any{"stuck": out, "count": len(out),
+		"delivery_counts": "not readable by the builder: the build seats' workers' information is granted to the delivery view alone (its backlog); an ask given up on is told by the builder at its check bound and asked again"}
+	if cut {
+		answer["cut"], answer["note"] = true, fmt.Sprintf("read %d of %d build keys in key order and stopped; give a module to read its keys alone", MaxReads, total)
+	}
+	return answer, nil
+}
+
+// StuckBuildModules are the build modules whose steps builder_stuck reads.
+var StuckBuildModules = []string{buildrun.GoBuildModule, buildrun.NodeBuildModule, ImageBuildModule}
+
+// idTime is when a step id was made (buildstep.NewID: the time in milliseconds in base 36, then six hex digits).
+func idTime(id string) (time.Time, bool) {
+	if len(id) <= 6 {
+		return time.Time{}, false
+	}
+	ms, err := strconv.ParseInt(id[:len(id)-6], 36, 64)
+	if err != nil || ms <= 0 {
+		return time.Time{}, false
+	}
+	return time.UnixMilli(ms).UTC(), true
+}

modules/mesh-delivery-builder/manifest_test.go

@@ -23,18 +23,21 @@ type state struct {
 }
 
 type artifact struct {
-	Name     string `json:"name"`
-	Kind     string `json:"kind"`
-	Language string `json:"language"`
-	System   string `json:"system"`
-	From     string `json:"from"`
-	Binary   string `json:"binary"`
+	Name     string   `json:"name"`
+	Kind     string   `json:"kind"`
+	Language string   `json:"language"`
+	System   string   `json:"system"`
+	From     string   `json:"from"`
+	Binary   string   `json:"binary"`
+	Loads    []string `json:"loads"`
 }
 
 type manifest struct {
 	Module   string   `json:"module"`
 	Version  string   `json:"version"`
 	Slug     string   `json:"slug"`
+	Tools    []string `json:"tools"`
+	Emits    []string `json:"emits"`
 	Consumes []string `json:"consumes"`
 	State    []state  `json:"state"`
 	Reads    []string `json:"reads"`
@@ -84,7 +87,7 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
 	if m.Module != "mesh-delivery-builder" {
 		t.Errorf("module = %q", m.Module)
 	}
-	if want := []state{{Name: "builds", History: 5}, {Name: "instances", History: 1}}; !reflect.DeepEqual(m.State, want) { // delivery.BuilderBuildsHistory
+	if want := []state{{Name: "builds", History: 5}, {Name: "instances", History: 1}, {Name: "queues", History: 1}, {Name: "reasks", History: 1}}; !reflect.DeepEqual(m.State, want) { // delivery.BuilderBuildsHistory
 		t.Errorf("state = %+v, want %+v", m.State, want)
 	}
 	// 1887/A4c: the builder chooses no node, so it reads neither presence nor have, invokes no build module's tool and
@@ -112,7 +115,14 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
 	if m.SecretsOwner != "mesh-delivery-builder" {
 		t.Errorf("secrets-owner = %q", m.SecretsOwner)
 	}
-	for _, k := range []string{"streams", "logs", "emits", "claims", "provides", "requires", "defines-seats", "tools", "seat", "seats", "bus", "config"} {
+	// 1945: the builder's tools (builder_seat_queue: note 62), served by its tools bundle, and the event it says when it asks an ask again.
+	if want := []string{"builder_ask_again", "builder_seat_queue", "builder_stuck"}; !reflect.DeepEqual(sorted(m.Tools), want) {
+		t.Errorf("tools = %v, want %v", m.Tools, want)
+	}
+	if want := []string{"build.asked-again"}; !reflect.DeepEqual(m.Emits, want) {
+		t.Errorf("emits = %v, want %v", m.Emits, want)
+	}
+	for _, k := range []string{"streams", "logs", "claims", "provides", "requires", "defines-seats", "seat", "seats", "bus", "config"} {
 		if _, ok := top[k]; ok {
 			t.Errorf("the manifest declares %q: the builder has none", k)
 		}
@@ -127,7 +137,7 @@ func TestManifestProcess(t *testing.T) { // 1607/AC1
 			procs = append(procs, r)
 		}
 		if r["type"] == "tool" || r["type"] == "seat" {
-			t.Errorf("resource %v: the builder serves no tool and holds no seat", r)
+			t.Errorf("resource %v: the builder's tools are its bundle's, and it holds no seat", r)
 		}
 	}
 	if len(procs) != 1 {
@@ -149,8 +159,13 @@ func TestManifestProcess(t *testing.T) { // 1607/AC1
 	if _, bus := p["bus"]; bus {
 		t.Errorf("the process declares a bus: it runs on its own broker credential")
 	}
-	if len(m.Build.Artifacts) != 1 {
-		t.Fatalf("artifacts = %+v, want the program alone", m.Build.Artifacts)
+	if len(m.Build.Artifacts) != 2 {
+		t.Fatalf("artifacts = %+v, want the program and the tools", m.Build.Artifacts)
+	}
+	if a := m.Build.Artifacts[1]; a.Name != "tools" || a.Kind != "bundle" || a.Binary != "mesh-delivery-builder-tools" || !reflect.DeepEqual(a.Loads, []string{"mesh-delivery-builder-tools"}) {
+		t.Errorf("tools artifact = %+v", a) // 1945
+	} else if st, err := os.Stat(a.From); err != nil || !st.IsDir() {
+		t.Errorf("the tools' source %s: %v", a.From, err)
 	}
 	a := m.Build.Artifacts[0]
 	if st, err := os.Stat(a.From); err != nil || !st.IsDir() || a.Binary != "mesh-delivery-builder" || !strings.HasSuffix(a.From, a.Binary) || a.Kind != "bundle" || a.Language != "go" {

modules/mesh-delivery-builder/module.json

@@ -2,6 +2,14 @@
   "module": "mesh-delivery-builder",
   "version": "1",
   "slug": "builder",
+  "tools": [
+    "builder_ask_again",
+    "builder_seat_queue",
+    "builder_stuck"
+  ],
+  "emits": [
+    "build.asked-again"
+  ],
   "consumes": [
     "mesh-build-go.toolchain.reported",
     "mesh-build-node.toolchain.reported",
@@ -12,7 +20,9 @@
   ],
   "state": [
     { "name": "builds", "history": 5 },
-    { "name": "instances", "history": 1 }
+    { "name": "instances", "history": 1 },
+    { "name": "queues", "history": 1 },
+    { "name": "reasks", "history": 1 }
   ],
   "reads": [
     "mesh-build-go.steps",
@@ -78,6 +88,17 @@
         "system": "arch",
         "from": "cmd/mesh-delivery-builder",
         "binary": "mesh-delivery-builder"
+      },
+      {
+        "name": "tools",
+        "kind": "bundle",
+        "language": "go",
+        "system": "arch",
+        "from": "cmd/mesh-delivery-builder-tools",
+        "binary": "mesh-delivery-builder-tools",
+        "loads": [
+          "mesh-delivery-builder-tools"
+        ]
       }
     ]
   }

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

@@ -7,6 +7,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"os"
+	"sync/atomic"
 	"time"
 
 	stdio "git.novox.be/novox/mesh-sdk/go"
@@ -15,8 +16,35 @@ import (
 	"mesh-delivery-checker/internal/check"
 )
 
+// reads counts the values this call read from the bus: the runtime launches the tools for one call, so the count is the call's.
+var reads atomic.Int64
+
+// cutAnswer says in the answer that it was cut, and how to narrow it.
+func cutAnswer(answer map[string]any, cut string) map[string]any {
+	if cut != "" {
+		answer["cut"], answer["note"] = true, cut
+	}
+	return answer
+}
+
+// logged runs every tool with one line to stderr per call: the tool, the values it read and the time it took.
+func logged(tools []stdio.Tool) []stdio.Tool {
+	for i := range tools {
+		name, run := tools[i].Name, tools[i].Run
+		tools[i].Run = func(args map[string]any) (any, error) {
+			reads.Store(0)
+			start := time.Now()
+			out, err := run(args)
+			fmt.Fprintf(os.Stderr, "[mesh-delivery-checker-tools] %s read %d values in %s (error: %v)\n", name, reads.Load(), time.Since(start).Round(time.Millisecond), err)
+			return out, err
+		}
+	}
+	return tools
+}
+
 func read(s stdio.KeptState) check.StateRead {
 	return check.StateRead{Keys: s.Keys, Get: func(key string) (json.RawMessage, error) {
+		reads.Add(1)
 		e, err := s.Get(key)
 		if err != nil || e == nil {
 			return nil, err
@@ -100,8 +128,8 @@ func Tools(r check.Reader) []stdio.Tool {
 				"group": map[string]any{"type": "string", "description": "the delivery group's branch name"}}},
 			Run: func(args map[string]any) (any, error) {
 				group, _ := args["group"].(string)
-				vs, err := r.Verdicts(group)
-				return map[string]any{"verdicts": vs}, err
+				vs, cut, err := r.Verdicts(group)
+				return cutAnswer(map[string]any{"verdicts": vs}, cut), err
 			}},
 		{Name: check.PullReadTool,
 			Description: "The verdict that judges one pull request and the full report of its status mesh/check. Read-only.",
@@ -122,8 +150,8 @@ func Tools(r check.Reader) []stdio.Tool {
 			Run: func(args map[string]any) (any, error) {
 				level, _ := args["level"].(string)
 				digest, _ := args["digest"].(string)
-				kept, err := r.Kept(level, digest)
-				return map[string]any{"kept": kept}, err
+				kept, cut, err := r.Kept(level, digest)
+				return cutAnswer(map[string]any{"kept": kept}, cut), err
 			}},
 		{Name: check.PreviewsReadTool,
 			Description: "The merge previews the checker holds, each as stored under previews.<repository>.<head>.<trunk>: the merge commit, per module directory " +
@@ -138,8 +166,8 @@ func Tools(r check.Reader) []stdio.Tool {
 				if head != "" && len(head) < 7 {
 					return nil, fmt.Errorf("the head %q is too short: give at least 7 characters", head)
 				}
-				ps, bad, err := r.Previews(repository, head)
-				return map[string]any{"previews": ps, "unreadable": bad}, err
+				ps, bad, cut, err := r.Previews(repository, head)
+				return cutAnswer(map[string]any{"previews": ps, "unreadable": bad}, cut), err
 			}},
 		{Name: check.StatusReadTool,
 			Description: "What runs: the steps in flight on the build modules by node, the candidates without a verdict for their heads, " +
@@ -184,7 +212,7 @@ func main() {
 		}},
 		Trunks: trunks,
 	}
-	if err := stdio.Serve("", append(Tools(r), TryTools(trying)...)); err != nil {
+	if err := stdio.Serve("", logged(append(Tools(r), TryTools(trying)...))); err != nil {
 		fmt.Fprintf(os.Stderr, "[mesh-delivery-checker-tools] %v\n", err)
 		os.Exit(1)
 	}

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

@@ -152,17 +152,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	if err != nil {
 		return err
 	}
-	bucket := func(name string) (jetstream.KeyValue, error) {
-		kv, err := js.KeyValue(ctx, name)
-		if err != nil {
-			return nil, fmt.Errorf("the bucket %s: %w", name, err)
-		}
-		return kv, nil
-	}
+	// Its own states, the candidates it checks, and the modules and manifests that say which build modules judge a repository:
+	// a check without one of them would be wrong or unwritten, so each stops the start, naming its grant (issue 1945).
 	buckets := map[string]jetstream.KeyValue{}
 	for _, name := range []string{delivery.CheckerVerdictsBucket, delivery.CheckerKeptBucket, delivery.CheckerOwedBucket, delivery.PreviewsBucket,
 		delivery.CandidatesBucket, delivery.ModulesBucket, delivery.ManifestsBucket, check.TriesBucket} {
-		if buckets[name], err = bucket(name); err != nil {
+		if buckets[name], err = process.EssentialBucket(ctx, js, "checker", name, "judge or keep a check"); err != nil {
 			return err
 		}
 	}
@@ -212,12 +207,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	// The builder's builds say how the candidates' fingerprints stand (issue 1684); where its bucket is not there yet the line
 	// says nothing of the builds, until the checker starts again.
 	var builds check.BuildStates
-	switch kv, err := js.KeyValue(ctx, delivery.BuilderBuildsBucket); {
-	case errors.Is(err, jetstream.ErrBucketNotFound):
-		logger.Printf("the bucket %s does not exist yet: the line says nothing of the builds until the checker starts again", delivery.BuilderBuildsBucket)
-	case err != nil:
-		return fmt.Errorf("the bucket %s: %w", delivery.BuilderBuildsBucket, err)
-	default:
+	if kv := process.OptionalBucket(ctx, js, delivery.BuilderBuildsBucket, "the line says nothing of the builds until the checker starts again", logger.Printf); kv != nil {
 		builds = buildReader{kvStore{kv}}
 	}
 	// The assigner's compositions (ADR 0528 clause 5, issue 1297): their bucket is looked up at each check until it is
@@ -290,47 +280,38 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	// the assigner's own presence says whether a composition is being made (compositionWatch, ADR 0537 §4). A check is no
 	// longer held for a node: its asks wait in the build seats' streams, which redeliver a lost node's ask (ADR 0525 §6,
 	// issue 1912), so the node-engine's have is not watched here.
-	switch kv, err := js.KeyValue(ctx, delivery.PresenceBucket); {
-	case errors.Is(err, jetstream.ErrBucketNotFound):
-		logger.Printf("the bucket %s does not exist yet: no node counts as online until the checker starts again", delivery.PresenceBucket)
-	case err != nil:
-		return fmt.Errorf("the bucket %s: %w", delivery.PresenceBucket, err)
-	default:
-		pw, err := settle(ctx, kv, func(e jetstream.KeyValueEntry) { presence.Put(e.Key(), e.Value(), !opOf(e)) })
-		if err != nil {
-			return err
+	if kv := process.OptionalBucket(ctx, js, delivery.PresenceBucket, "no node counts as online until the checker starts again", logger.Printf); kv != nil {
+		if pw, err := settle(ctx, kv, func(e jetstream.KeyValueEntry) { presence.Put(e.Key(), e.Value(), !opOf(e)) }); err != nil {
+			process.NotWatched(logger.Printf, delivery.PresenceBucket, err, "no node counts as online until the checker starts again")
+		} else {
+			go func() {
+				errs <- follow(ctx, pw, func(e jetstream.KeyValueEntry) {
+					if presence.Put(e.Key(), e.Value(), !opOf(e)) {
+						ch.NodesChanged(ctx) // a node came online or went: a check whose forge was not heard may now be heard
+					}
+				})
+			}()
 		}
-		go func() {
-			errs <- follow(ctx, pw, func(e jetstream.KeyValueEntry) {
-				if presence.Put(e.Key(), e.Value(), !opOf(e)) {
-					ch.NodesChanged(ctx) // a node came online or went: a check whose forge was not heard may now be heard
-				}
-			})
-		}()
 	}
 	// The gate's source says the trunk head of each repository, which a pull request is merged onto. Where it is not there
 	// yet no trunk is known, so nothing is checked.
-	switch kv, err := js.KeyValue(ctx, delivery.SourceBucket); {
-	case errors.Is(err, jetstream.ErrBucketNotFound):
-		logger.Printf("the bucket %s does not exist yet: no trunk is known until the checker starts again", delivery.SourceBucket)
-	case err != nil:
-		return fmt.Errorf("the bucket %s: %w", delivery.SourceBucket, err)
-	default:
+	if kv := process.OptionalBucket(ctx, js, delivery.SourceBucket, "no trunk is known, so nothing is checked until the checker starts again", logger.Printf); kv != nil {
 		sw, err := settle(ctx, kv, func(e jetstream.KeyValueEntry) {
 			if e.Key() == delivery.SourceMainKey {
 				trunks.Put(e.Value(), !opOf(e))
 			}
 		})
 		if err != nil {
-			return err
+			process.NotWatched(logger.Printf, delivery.SourceBucket, err, "no trunk is known, so nothing is checked until the checker starts again")
+		} else {
+			go func() {
+				errs <- follow(ctx, sw, func(e jetstream.KeyValueEntry) {
+					if e.Key() == delivery.SourceMainKey && trunks.Put(e.Value(), !opOf(e)) {
+						ch.TrunkMoved(ctx) // the trunk moved: its statuses go pending, and a verdict for the old trunk is judged again
+					}
+				})
+			}()
 		}
-		go func() {
-			errs <- follow(ctx, sw, func(e jetstream.KeyValueEntry) {
-				if e.Key() == delivery.SourceMainKey && trunks.Put(e.Value(), !opOf(e)) {
-					ch.TrunkMoved(ctx) // the trunk moved: its statuses go pending, and a verdict for the old trunk is judged again
-				}
-			})
-		}()
 	}
 	// The modules say which repository holds a TypeScript module, so the Node build module judges it too.
 	mw, err := settle(ctx, buckets[delivery.ModulesBucket], func(e jetstream.KeyValueEntry) { repos.Put(ctx, e.Key(), e.Value(), !opOf(e)) })

modules/mesh-delivery-checker/cmd/mesh-delivery-checker/standing.go

@@ -193,7 +193,9 @@ func watchAssignments(ctx context.Context, js bucketOpener, gate *toolchainGate,
 	each := func(e jetstream.KeyValueEntry) { gate.Assignment(e.Key(), e.Value(), !opOf(e), e.Created()) }
 	w, err := settle(ctx, kv, each)
 	if err != nil {
-		return nil, fmt.Errorf("the bucket %s: %w", provider.AssignmentsBucket, err)
+		// Not fatal either (issue 1945): a watch the bus refuses is the same missing grant as a refused read.
+		logger.Printf("the bucket %s cannot be watched, so every node's toolchain report counts, an unassigned build node's too (issue 1940): %v", provider.AssignmentsBucket, err)
+		return errs, nil
 	}
 	if gate.Settled() {
 		changed(ctx)

modules/mesh-delivery-checker/cmd/mesh-delivery-checker/standing_test.go

@@ -3,11 +3,19 @@ package main
 // The same in mesh-delivery-indexer, mesh-delivery-builder and mesh-delivery-checker, as standing.go is (issue 1940).
 
 import (
+	"bytes"
+	"context"
+	"errors"
 	"fmt"
+	"log"
 	"strings"
 	"testing"
 	"time"
 
+	"git.novox.be/novox/mesh-sdk/go/process"
+
+	"github.com/nats-io/nats.go/jetstream"
+
 	"git.novox.be/novox/mesh-sdk/go/delivery"
 	"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
 )
@@ -89,6 +97,51 @@ func TestAnUnassignedBuildNodesReportIsNoBuilder(t *testing.T) {
 	if n := len(set.Entries("typescript")); n != 2 {
 		t.Fatalf("novox and shanks are left; got %d entries", n)
 	}
+
+	// 1945 (a part exits at start for a bucket it works without): a bus that refuses the assignments' watch, or every
+	// toolchains bucket, leaves the watches unstarted and the part running, said; a refused bucket it cannot work without
+	// is an error naming the bucket and its grant.
+	var logs bytes.Buffer
+	logger := log.New(&logs, "", 0)
+	refused := refusingBus{err: errors.New("nats: permissions violation for publish to $JS.API.CONSUMER.CREATE.KV_mesh-controller_assignments")}
+	if _, err := watchAssignments(context.Background(), refused, newToolchainGate(nil, set.Put, t.Logf), func(context.Context) {}, logger); err != nil {
+		t.Errorf("a refused watch of the assignments stopped the start: %v", err)
+	}
+	if err := newToolchainWatch(context.Background(), refusingBus{refuseOpen: true, err: refused.err}, set, func(context.Context) {}, logger).start(); err != nil {
+		t.Errorf("a refused toolchains bucket stopped the start: %v", err)
+	}
+	if !strings.Contains(logs.String(), "cannot be watched") || !strings.Contains(logs.String(), "cannot be read") {
+		t.Errorf("the refusals were not said: %q", logs.String())
+	}
+	if _, err := process.EssentialBucket(context.Background(), refusingBus{refuseOpen: true, err: refused.err}, "checker", "mesh-delivery-indexer_modules", "work"); err == nil ||
+		!strings.Contains(err.Error(), "grant of mesh-delivery-indexer_modules") {
+		t.Errorf("a refused bucket the part needs: %v, want an error naming its grant", err)
+	}
+}
+
+// refusingBus is a bus that refuses: every open (refuseOpen), or every watch of a bucket it opens, with err.
+type refusingBus struct {
+	refuseOpen bool
+	err        error
+}
+
+func (r refusingBus) KeyValue(context.Context, string) (jetstream.KeyValue, error) {
+	if r.refuseOpen {
+		return nil, r.err
+	}
+	return refusedWatch{err: r.err}, nil
+}
+
+type refusedWatch struct {
+	jetstream.KeyValue
+	err error
+}
+
+func (r refusedWatch) WatchAll(context.Context, ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) {
+	return nil, r.err
+}
+func (r refusedWatch) Watch(context.Context, string, ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) {
+	return nil, r.err
 }
 
 // A build node's report counts from the first one it writes since its assignment: a newly assigned node that has not

modules/mesh-delivery-checker/cmd/mesh-delivery-checker/toolchains.go

@@ -8,6 +8,8 @@ import (
 	"strings"
 	"sync"
 
+	"git.novox.be/novox/mesh-sdk/go/process"
+
 	"github.com/nats-io/nats.go/jetstream"
 
 	"git.novox.be/novox/mesh-sdk/go/delivery"
@@ -63,11 +65,14 @@ func (w *toolchainWatch) ensure(module string) error {
 		w.logger.Printf("the bucket %s does not exist yet: it is watched when %s reports a toolchain", name, module)
 		return nil
 	case err != nil:
-		return fmt.Errorf("the bucket %s: %w", name, err)
+		// Refused or not answered (its grant not on the bus yet, issue 1945): said, and opened again at the module's next report.
+		w.logger.Printf("the bucket %s cannot be read (%v): %s's toolchains are not heard until its next report", name, err, module)
+		return nil
 	}
 	watcher, err := settle(w.ctx, kv, func(e jetstream.KeyValueEntry) { w.put(module, e.Key(), e.Value(), !opOf(e)) })
 	if err != nil {
-		return err
+		process.NotWatched(w.logger.Printf, name, err, fmt.Sprintf("%s's toolchains are not heard until its next report", module))
+		return nil
 	}
 	w.watched[module] = true
 	go func() {

modules/mesh-delivery-checker/go.mod

@@ -3,7 +3,7 @@ module mesh-delivery-checker
 go 1.26.0
 
 require (
-	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023213200-403a5d72ab93
+	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585
 	github.com/nats-io/nats-server/v2 v2.11.17
 	github.com/nats-io/nats.go v1.54.0
 )

modules/mesh-delivery-checker/go.sum

@@ -1,5 +1,7 @@
 git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023213200-403a5d72ab93 h1:FAiFY2KBKEoZSH49OZTt7/zRBIQxyl5g/KhYrpggbN4=
 git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023213200-403a5d72ab93/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585 h1:7QoFH5qjXRpx5M4uPdYsVJ5qKl2082gq8n0lFxhoNIg=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585/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=

modules/mesh-delivery-checker/internal/check/bounded.go

@@ -0,0 +1,79 @@
+package check
+
+import (
+	"fmt"
+	"sort"
+	"sync"
+	"sync/atomic"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+// ReadWorkers is how many values a read tool asks the bus for at once. Each value is a round trip from the tool through the
+// runtime to the bus, so reading a bucket of thousands of keys one after the other outlasts the 30 s a tool call is given
+// (issue 1827, as the delivery view's issue 1945). It is a copy of the view's: a few lines are not worth a group delivery
+// through mesh-sdk (rules-are-weighed).
+const ReadWorkers = 16
+
+// MaxReads bounds the values one call of a read tool reads; past it the answer says it was cut and how to narrow it.
+const MaxReads = 1000
+
+// parallel runs read(i) for i in [0, n), at most ReadWorkers at a time, and answers the first error; once one failed no
+// further read starts.
+func parallel(n int, read func(i int) error) error {
+	var (
+		next   atomic.Int64
+		failed atomic.Bool
+		once   sync.Once
+		first  error
+		wg     sync.WaitGroup
+	)
+	for w := 0; w < min(ReadWorkers, n); w++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for !failed.Load() {
+				i := int(next.Add(1) - 1)
+				if i >= n {
+					return
+				}
+				if err := read(i); err != nil {
+					once.Do(func() { first = err })
+					failed.Store(true)
+				}
+			}
+		}()
+	}
+	wg.Wait()
+	return first
+}
+
+// capKeys keeps the first MaxReads keys and answers a note when it cut the rest, "" otherwise.
+func capKeys(keys []string, narrow string) ([]string, string) {
+	if len(keys) <= MaxReads {
+		return keys, ""
+	}
+	return keys[:MaxReads], fmt.Sprintf("read %d of %d matching keys and stopped; %s", MaxReads, len(keys), narrow)
+}
+
+// newestFirst sorts step keys (<node>.<id>) by id, newest first: an id is the time in base 36, then six hex digits, so the
+// longer id is the later and equal lengths compare as text. A key that is no step key goes last.
+func newestFirst(keys []string) {
+	id := func(k string) (string, bool) {
+		_, id, err := delivery.ParseStepKey(k)
+		return id, err == nil
+	}
+	sort.Slice(keys, func(i, j int) bool {
+		a, aok := id(keys[i])
+		b, bok := id(keys[j])
+		switch {
+		case aok != bok:
+			return aok
+		case len(a) != len(b):
+			return len(a) > len(b)
+		case a != b:
+			return a > b
+		}
+		return keys[i] > keys[j]
+	})
+}

modules/mesh-delivery-checker/internal/check/read.go

@@ -89,37 +89,46 @@ func (r Reader) read(key string, raw json.RawMessage) (VerdictRead, error) {
 	return out, nil
 }
 
-// Verdicts are the verdicts the checker holds, by key; with a group only that group's.
-func (r Reader) Verdicts(group string) ([]VerdictRead, error) {
+// Verdicts are the verdicts the checker holds, by key; with a group only that group's. The keys are narrowed by the group
+// before any value is read, at most MaxReads are read (sixteen at a time), and cut says when it stopped there.
+func (r Reader) Verdicts(group string) (out []VerdictRead, cut string, err error) {
 	if r.VerdictState.Keys == nil || r.VerdictState.Get == nil {
-		return nil, fmt.Errorf("the verdicts state cannot be read here")
+		return nil, "", fmt.Errorf("the verdicts state cannot be read here")
 	}
 	keys, err := r.VerdictState.Keys()
 	if err != nil {
-		return nil, err
+		return nil, "", err
 	}
 	sort.Strings(keys)
-	var out []VerdictRead
+	var matching []string
 	for _, k := range keys {
 		if group != "" {
 			if g, _, err := delivery.ParseCheckerVerdictKey(k); err != nil || g != group {
 				continue
 			}
 		}
-		raw, err := r.VerdictState.Get(k)
-		if err != nil {
-			return nil, err
-		}
-		if len(raw) == 0 {
-			continue
-		}
-		vr, err := r.read(k, raw)
-		if err != nil {
-			return nil, err
-		}
-		out = append(out, vr)
+		matching = append(matching, k)
 	}
-	return out, nil
+	matching, cut = capKeys(matching, "give a group to narrow the answer")
+	rows := make([]VerdictRead, len(matching))
+	found := make([]bool, len(matching))
+	if err := parallel(len(matching), func(i int) error {
+		raw, err := r.VerdictState.Get(matching[i])
+		if err != nil || len(raw) == 0 {
+			return err
+		}
+		rows[i], err = r.read(matching[i], raw)
+		found[i] = err == nil
+		return err
+	}); err != nil {
+		return nil, "", err
+	}
+	for i, row := range rows {
+		if found[i] {
+			out = append(out, row)
+		}
+	}
+	return out, cut, nil
 }
 
 // Pull answers for one pull request: the verdict of the candidate that holds it, with its report; for one the indexer holds
@@ -133,74 +142,95 @@ func (r Reader) Pull(repository string, number int) (VerdictRead, error) {
 		return VerdictRead{}, err
 	}
 	sort.Strings(keys)
-	for _, k := range keys {
-		raw, err := r.CandidateState.Get(k)
-		if err != nil || len(raw) == 0 {
-			continue
-		}
-		c, err := delivery.DecodeCandidate(raw)
-		if err != nil {
-			continue
-		}
-		for _, p := range c.Pulls {
-			if p.Repository != repository || p.Number != number {
+	keys, cut := capKeys(keys, "")
+	// The candidates are read in batches, each in parallel and taken in key order, so the read stops at the batch that holds it.
+	for from := 0; from < len(keys); from += ReadWorkers * 4 {
+		batch := keys[from:min(from+ReadWorkers*4, len(keys))]
+		cands := make([]delivery.Candidate, len(batch))
+		ok := make([]bool, len(batch))
+		_ = parallel(len(batch), func(i int) error {
+			raw, err := r.CandidateState.Get(batch[i])
+			if err != nil || len(raw) == 0 {
+				return nil
+			}
+			c, err := delivery.DecodeCandidate(raw)
+			cands[i], ok[i] = c, err == nil
+			return nil
+		})
+		for i, c := range cands {
+			if !ok[i] {
 				continue
 			}
-			vkey, err := delivery.CheckerVerdictKey(c.Group, c.HeadsHash)
-			if err != nil {
-				return VerdictRead{}, err
+			for _, p := range c.Pulls {
+				if p.Repository != repository || p.Number != number {
+					continue
+				}
+				vkey, err := delivery.CheckerVerdictKey(c.Group, c.HeadsHash)
+				if err != nil {
+					return VerdictRead{}, err
+				}
+				vraw, err := r.VerdictState.Get(vkey)
+				if err != nil {
+					return VerdictRead{}, err
+				}
+				if len(vraw) == 0 {
+					return VerdictRead{Key: vkey, Note: "no verdict yet for the heads of its candidate"}, nil
+				}
+				return r.read(vkey, vraw)
 			}
-			vraw, err := r.VerdictState.Get(vkey)
-			if err != nil {
-				return VerdictRead{}, err
-			}
-			if len(vraw) == 0 {
-				return VerdictRead{Key: vkey, Note: "no verdict yet for the heads of its candidate"}, nil
-			}
-			return r.read(vkey, vraw)
 		}
 	}
+	if cut != "" {
+		return VerdictRead{}, fmt.Errorf("no candidate among the first %d holds %s#%d", MaxReads, repository, number)
+	}
 	return VerdictRead{}, fmt.Errorf("no candidate holds %s#%d", repository, number)
 }
 
 // Kept is the kept verdicts: one key's value as stored, or with an empty key the keys of a level (or every level) and their values.
-func (r Reader) Kept(level, digest string) (map[string]json.RawMessage, error) {
+func (r Reader) Kept(level, digest string) (map[string]json.RawMessage, string, error) {
 	if r.KeptState.Keys == nil || r.KeptState.Get == nil {
-		return nil, fmt.Errorf("the kept state cannot be read here")
+		return nil, "", fmt.Errorf("the kept state cannot be read here")
 	}
 	if digest != "" {
 		key, err := delivery.KeptKey(level, digest)
 		if err != nil {
-			return nil, err
+			return nil, "", err
 		}
 		raw, err := r.KeptState.Get(key)
 		if err != nil {
-			return nil, err
+			return nil, "", err
 		}
 		if len(raw) == 0 {
-			return nil, fmt.Errorf("no kept verdict %s", key)
+			return nil, "", fmt.Errorf("no kept verdict %s", key)
 		}
-		return map[string]json.RawMessage{key: raw}, nil
+		return map[string]json.RawMessage{key: raw}, "", nil
 	}
 	keys, err := r.KeptState.Keys()
 	if err != nil {
-		return nil, err
+		return nil, "", err
 	}
 	sort.Strings(keys)
-	out := map[string]json.RawMessage{}
+	var matching []string
 	for _, k := range keys {
-		if l, _, err := delivery.ParseKeptKey(k); err != nil || (level != "" && l != level) {
-			continue
-		}
-		raw, err := r.KeptState.Get(k)
-		if err != nil {
-			return nil, err
-		}
-		if len(raw) > 0 {
-			out[k] = raw
+		if l, _, err := delivery.ParseKeptKey(k); err == nil && (level == "" || l == level) {
+			matching = append(matching, k)
 		}
 	}
-	return out, nil
+	matching, cut := capKeys(matching, "give a level and a digest to read one")
+	values := make([]json.RawMessage, len(matching))
+	if err := parallel(len(matching), func(i int) (err error) {
+		values[i], err = r.KeptState.Get(matching[i])
+		return err
+	}); err != nil {
+		return nil, "", err
+	}
+	out := map[string]json.RawMessage{}
+	for i, k := range matching {
+		if len(values[i]) > 0 {
+			out[k] = values[i]
+		}
+	}
+	return out, cut, nil
 }
 
 // PreviewRead is one preview as stored, with the parts of its key.
@@ -214,17 +244,18 @@ type PreviewRead struct {
 
 // Previews are the previews the checker holds, each as stored. With a repository only that repository's; with a head (a
 // prefix of at least 7 characters is enough) only previews of that head. A key that is no preview key is listed as unreadable.
-func (r Reader) Previews(repository, head string) ([]PreviewRead, []string, error) {
+func (r Reader) Previews(repository, head string) ([]PreviewRead, []string, string, error) {
 	if r.PreviewState.Keys == nil || r.PreviewState.Get == nil {
-		return nil, nil, fmt.Errorf("the previews state cannot be read here")
+		return nil, nil, "", fmt.Errorf("the previews state cannot be read here")
 	}
 	keys, err := r.PreviewState.Keys()
 	if err != nil {
-		return nil, nil, err
+		return nil, nil, "", err
 	}
 	sort.Strings(keys)
-	out := []PreviewRead{}
-	var bad []string
+	type part struct{ repo, head, trunk string }
+	parts := map[string]part{}
+	var matching, bad []string
 	for _, k := range keys {
 		repo, h, trunk, err := delivery.ParsePreviewKey(k)
 		if err != nil {
@@ -234,15 +265,25 @@ func (r Reader) Previews(repository, head string) ([]PreviewRead, []string, erro
 		if (repository != "" && repo != repository) || (head != "" && !strings.HasPrefix(h, head)) {
 			continue
 		}
-		raw, err := r.PreviewState.Get(k)
-		if err != nil {
-			return nil, nil, err
-		}
-		if len(raw) > 0 {
-			out = append(out, PreviewRead{Key: k, Repository: repo, Head: h, Trunk: trunk, Preview: raw})
+		parts[k] = part{repo, h, trunk}
+		matching = append(matching, k)
+	}
+	matching, cut := capKeys(matching, "give a repository and a head to narrow the answer")
+	values := make([]json.RawMessage, len(matching))
+	if err := parallel(len(matching), func(i int) (err error) {
+		values[i], err = r.PreviewState.Get(matching[i])
+		return err
+	}); err != nil {
+		return nil, nil, "", err
+	}
+	out := []PreviewRead{}
+	for i, k := range matching {
+		if len(values[i]) > 0 {
+			p := parts[k]
+			out = append(out, PreviewRead{Key: k, Repository: p.repo, Head: p.head, Trunk: p.trunk, Preview: values[i]})
 		}
 	}
-	return out, bad, nil
+	return out, bad, cut, nil
 }
 
 // RunningStep is a step in flight on a build module.
@@ -300,16 +341,28 @@ func (r Reader) Status() (Status, error) {
 			st.Notes = append(st.Notes, "the steps of "+m+" cannot be listed: "+err.Error())
 			continue
 		}
-		sort.Strings(keys)
-		for _, k := range keys {
-			raw, err := sr.Get(k)
+		// A step that runs started lately: the newest ids are read first, at most MaxReads of them.
+		newestFirst(keys)
+		keys, cut := capKeys(keys, "")
+		if cut != "" {
+			st.Notes = append(st.Notes, "the steps of "+m+": read the newest "+fmt.Sprint(MaxReads)+" of its keys; a step older than those is not shown")
+		}
+		steps := make([]delivery.Step, len(keys))
+		isRunning := make([]bool, len(keys))
+		_ = parallel(len(keys), func(i int) error {
+			raw, err := sr.Get(keys[i])
 			if err != nil || len(raw) == 0 {
-				continue
+				return nil
 			}
 			step, err := delivery.DecodeStep(raw)
-			if err != nil || step.State != delivery.StepRunning {
+			steps[i], isRunning[i] = step, err == nil && step.State == delivery.StepRunning
+			return nil
+		})
+		for i := range keys {
+			if !isRunning[i] {
 				continue
 			}
+			step := steps[i]
 			rs := RunningStep{Module: m, Node: step.Node, ID: step.ID, Tool: step.Tool}
 			if !step.Started.IsZero() {
 				rs.Since = step.Started.UTC().Format("2006-01-02T15:04:05Z")
@@ -344,23 +397,41 @@ func (r Reader) Status() (Status, error) {
 			return st, err
 		}
 		sort.Strings(keys)
-		for _, k := range keys {
-			raw, err := r.CandidateState.Get(k)
+		total := len(keys)
+		keys, cut := capKeys(keys, "")
+		if cut != "" {
+			st.Notes = append(st.Notes, fmt.Sprintf("read the first %d candidates of %d", MaxReads, total))
+		}
+		type judged struct {
+			c   delivery.Candidate
+			raw json.RawMessage
+			ok  bool
+		}
+		rows := make([]judged, len(keys))
+		_ = parallel(len(keys), func(i int) error {
+			raw, err := r.CandidateState.Get(keys[i])
 			if err != nil || len(raw) == 0 {
-				continue
+				return nil
 			}
 			c, err := delivery.DecodeCandidate(raw)
 			if err != nil {
-				continue
+				return nil
 			}
 			vkey, err := delivery.CheckerVerdictKey(c.Group, c.HeadsHash)
 			if err != nil {
+				return nil
+			}
+			rows[i].c = c
+			if vraw, err := r.VerdictState.Get(vkey); err == nil {
+				rows[i].raw, rows[i].ok = vraw, true
+			}
+			return nil
+		})
+		for _, row := range rows {
+			if !row.ok {
 				continue
 			}
-			raw, err = r.VerdictState.Get(vkey)
-			if err != nil {
-				continue
-			}
+			c, raw := row.c, row.raw
 			if len(raw) == 0 {
 				st.Unjudged = append(st.Unjudged, c.Group)
 				st.Waiting[KindNew] = append(st.Waiting[KindNew], c.Group)

modules/mesh-delivery-checker/internal/check/read_test.go

@@ -3,7 +3,9 @@ package check
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"strings"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -51,7 +53,7 @@ func readerOf(t *testing.T) (Reader, string) {
 
 func TestTheVerdictToolsAnswerTheStoredVerdictWithTheFullReportOfEachPullRequest(t *testing.T) { // N9/AC1, 1686
 	rd, key := readerOf(t)
-	vs, err := rd.Verdicts("")
+	vs, _, err := rd.Verdicts("")
 	if err != nil || len(vs) != 1 || vs[0].Key != key {
 		t.Fatalf("verdicts = %+v, %v", vs, err)
 	}
@@ -61,7 +63,7 @@ func TestTheVerdictToolsAnswerTheStoredVerdictWithTheFullReportOfEachPullRequest
 	if got := vs[0].Reports["novox/mesh-delivery#7"]; !strings.HasPrefix(got, "pass 111111111111 on aaaaaaaaaaaa") || !strings.Contains(got, "composition not yet run") {
 		t.Errorf("reports = %v", vs[0].Reports)
 	}
-	if only, _ := rd.Verdicts("feat/other"); len(only) != 0 {
+	if only, _, _ := rd.Verdicts("feat/other"); len(only) != 0 {
 		t.Errorf("another group's verdicts = %+v", only)
 	}
 }
@@ -85,11 +87,11 @@ func TestAnHqPullRequestIsAnsweredAsAnyOther(t *testing.T) {
 
 func TestTheKeptToolAnswersTheKeysAsStoredByLevelAndByDigest(t *testing.T) { // N9/AC1
 	rd, _ := readerOf(t)
-	all, err := rd.Kept("", "")
+	all, _, err := rd.Kept("", "")
 	if err != nil || len(all) != 1 {
 		t.Fatalf("kept = %v, %v", all, err)
 	}
-	one, err := rd.Kept("unit", digA)
+	one, _, err := rd.Kept("unit", digA)
 	if err != nil || len(one) != 1 {
 		t.Fatalf("one kept = %v, %v", one, err)
 	}
@@ -98,10 +100,10 @@ func TestTheKeptToolAnswersTheKeysAsStoredByLevelAndByDigest(t *testing.T) { //
 			t.Errorf("kept value = %+v, %v", k, err)
 		}
 	}
-	if none, _ := rd.Kept("integration", ""); len(none) != 0 {
+	if none, _, _ := rd.Kept("integration", ""); len(none) != 0 {
 		t.Errorf("integration = %v", none)
 	}
-	if _, err := rd.Kept("unit", digB); err == nil {
+	if _, _, err := rd.Kept("unit", digB); err == nil {
 		t.Error("a digest nothing kept was answered")
 	}
 }
@@ -115,17 +117,17 @@ func TestThePreviewsToolShowsTheStoredPreviewOfAPullRequest(t *testing.T) {
 	}
 	key, _ := delivery.PreviewKey(repoA, head1, head2)
 	rd := Reader{PreviewState: fakeState{key: data, "junk": []byte("{}")}.read()}
-	got, bad, err := rd.Previews(repoA, head1[:8])
+	got, bad, _, err := rd.Previews(repoA, head1[:8])
 	if err != nil || len(got) != 1 || got[0].Key != key || got[0].Head != head1 || got[0].Trunk != head2 || string(got[0].Preview) != string(data) {
 		t.Fatalf("previews = %+v, %v", got, err)
 	}
 	if len(bad) != 1 || bad[0] != "junk" {
 		t.Errorf("unreadable = %v", bad)
 	}
-	if got, _, _ := rd.Previews("novox/other", ""); len(got) != 0 {
+	if got, _, _, _ := rd.Previews("novox/other", ""); len(got) != 0 {
 		t.Errorf("another repository's = %+v", got)
 	}
-	if _, _, err := (Reader{}).Previews("", ""); err == nil {
+	if _, _, _, err := (Reader{}).Previews("", ""); err == nil {
 		t.Error("answered with no state to read")
 	}
 }
@@ -233,3 +235,121 @@ func TestTheStatusCountsAnHQOnlyGroupAsUnjudged(t *testing.T) {
 		t.Errorf("unjudged = %v, want pb-x", st.Unjudged)
 	}
 }
+
+// slowBucket wraps a state with a Get that takes delay and counts the values read, the lists and the most Gets in flight.
+type slowBucket struct {
+	vals                fakeState
+	reads, lists        atomic.Int64
+	inFlight, maxFlight atomic.Int64
+}
+
+func (b *slowBucket) get(key string) (json.RawMessage, error) {
+	b.reads.Add(1)
+	now := b.inFlight.Add(1)
+	for m := b.maxFlight.Load(); now > m && !b.maxFlight.CompareAndSwap(m, now); m = b.maxFlight.Load() {
+	}
+	time.Sleep(time.Millisecond)
+	b.inFlight.Add(-1)
+	return b.vals.read().Get(key)
+}
+
+func (b *slowBucket) read() StateRead {
+	return StateRead{Keys: func() ([]string, error) { b.lists.Add(1); return b.vals.read().Keys() }, Get: b.get}
+}
+
+func (b *slowBucket) Keys() ([]string, error)                 { return b.read().Keys() }
+func (b *slowBucket) Get(key string) (json.RawMessage, error) { return b.get(key) }
+func (b *slowBucket) Create(string, []byte) error             { return nil }
+
+// 1827: a read tool lists the keys of a bucket of thousands and reads only the values its answer needs, sixteen at a time, at
+// most MaxReads, saying when it cut (on the live shape of 2026-10-26: kept 7990 keys, steps 10034, tries 1000s).
+func TestReadToolsReadOnlyWhatTheyNeedInParallelAndSayWhenCut(t *testing.T) { // 1827
+	kept := fakeState{}
+	for i := 0; i < 1800; i++ {
+		level := map[bool]string{true: "unit", false: "integration"}[i%9 != 0]
+		k, _ := delivery.KeptKey(level, fmt.Sprintf("%064x", i))
+		kept[k] = []byte(`{"v":1}`)
+	}
+	previews := fakeState{}
+	for i := 0; i < 1500; i++ {
+		repo := map[bool]string{true: repoA, false: repoSDK}[i < 300]
+		k, _ := delivery.PreviewKey(repo, fmt.Sprintf("%08x%032x", i, 0), trunkA)
+		previews[k] = []byte(`{"v":1}`)
+	}
+	for _, row := range []struct {
+		name         string
+		call         func(*slowBucket) (rows int, cut string, err error)
+		rows         int
+		reads, lists int64
+		cut          bool
+	}{
+		{"kept of one key", func(b *slowBucket) (int, string, error) {
+			m, c, err := Reader{KeptState: b.read()}.Kept("unit", fmt.Sprintf("%064x", 1))
+			return len(m), c, err
+		}, 1, 1, 0, false},
+		{"kept of a level", func(b *slowBucket) (int, string, error) {
+			m, c, err := Reader{KeptState: b.read()}.Kept("integration", "")
+			return len(m), c, err
+		}, 200, 200, 1, false},
+		{"kept of everything", func(b *slowBucket) (int, string, error) {
+			m, c, err := Reader{KeptState: b.read()}.Kept("", "")
+			return len(m), c, err
+		}, MaxReads, MaxReads, 1, true},
+		{"previews of a repository", func(b *slowBucket) (int, string, error) {
+			p, _, c, err := Reader{PreviewState: b.read()}.Previews(repoSDK, "")
+			return len(p), c, err
+		}, 1200, 1000, 1, true},
+		{"previews of a repository and a head", func(b *slowBucket) (int, string, error) {
+			p, _, c, err := Reader{PreviewState: b.read()}.Previews(repoA, fmt.Sprintf("%08x", 7))
+			return len(p), c, err
+		}, 1, 1, 1, false},
+	} {
+		b := &slowBucket{}
+		b.vals = kept
+		if strings.HasPrefix(row.name, "previews") {
+			b.vals = previews
+		}
+		start := time.Now()
+		rows, cut, err := row.call(b)
+		if err != nil {
+			t.Fatalf("%s: %v", row.name, err)
+		}
+		if rows != min(row.rows, MaxReads) || b.reads.Load() != row.reads || b.lists.Load() != row.lists || (cut != "") != row.cut {
+			t.Errorf("%s: %d rows, %d reads, %d lists, cut %q; want %d, %d, %d, cut %v", row.name, rows, b.reads.Load(), b.lists.Load(), cut, min(row.rows, MaxReads), row.reads, row.lists, row.cut)
+		}
+		if row.reads > 20 && b.maxFlight.Load() < 2 {
+			t.Errorf("%s: values were read one at a time", row.name)
+		}
+		t.Logf("%s: %d values in %v (one after the other: %v)", row.name, b.reads.Load(), time.Since(start), time.Duration(b.reads.Load())*time.Millisecond)
+	}
+
+	// The status reads the newest steps of a bucket of 1500 first, at most MaxReads, and finds the one that runs.
+	now := time.Date(2026, 10, 26, 7, 0, 0, 0, time.UTC)
+	steps := fakeState{}
+	for i := 0; i < 1500; i++ {
+		id := fmt.Sprintf("mvo5%04dabcdef", i)
+		step := delivery.Step{ID: id, Tool: "t", Node: "g14", State: delivery.StepPassed}
+		if i == 1499 {
+			step.State, step.Started = delivery.StepRunning, now.Add(-time.Minute)
+		}
+		raw, _ := delivery.EncodeStep(step)
+		steps["g14."+id] = raw
+	}
+	sb := &slowBucket{vals: steps}
+	st, err := Reader{StepStates: map[string]StateRead{"mesh-build-go": sb.read()}, HaveState: fakeState{"g14.mesh-build-go": haveOf(t, delivery.PhaseRunning)}.read(),
+		Now: func() time.Time { return now }}.Status()
+	if err != nil || sb.reads.Load() != MaxReads || len(st.Running) != 1 || sb.maxFlight.Load() < 2 {
+		t.Errorf("status: %d reads (want %d), running %+v, in flight %d, %v", sb.reads.Load(), MaxReads, st.Running, sb.maxFlight.Load(), err)
+	}
+
+	// A new try reads the newest TryScan of 1500 ended tries, not all, and still finds the same head asked twice.
+	tries := fakeState{}
+	for i := 0; i < 1500; i++ {
+		raw, _ := encodeTry(Try{ID: fmt.Sprintf("try-%d", 1700000000000000000+i), Repository: repoA, Head: head1, Trunk: trunkA, State: TryPass})
+		tries[fmt.Sprintf("try-%d", 1700000000000000000+i)] = raw
+	}
+	tb := &slowBucket{vals: tries}
+	if _, err := AskTry(tb, tryTrunks, TryAsk{Repository: repoA, Head: head2, Trunk: trunkA}, now); err != nil || tb.reads.Load() != TryScan {
+		t.Errorf("AskTry read %d of 1500 tries (want %d), %v", tb.reads.Load(), TryScan, err)
+	}
+}

modules/mesh-delivery-checker/internal/check/try.go

@@ -29,6 +29,8 @@ const (
 	TriesBucket = delivery.CheckerModule + "_" + TriesState
 	// MaxTries is how many tries may wait or run at once, mesh-wide: each holds a build node for minutes.
 	MaxTries = 4
+	// TryScan is how many of the newest tries a new ask reads to find the ones that wait or run (issue 1827).
+	TryScan = 64
 	// TryKeep is how long an ended try's key is kept before the checker removes it.
 	TryKeep = 7 * 24 * time.Hour
 )
@@ -155,15 +157,33 @@ func AskTry(store TryStore, trunks map[string]string, a TryAsk, now time.Time) (
 	if err != nil {
 		return "", fmt.Errorf("the tries cannot be listed: %w", err)
 	}
-	sort.Strings(keys)
-	var open []string
-	for _, k := range keys {
-		raw, err := store.Get(k)
+	// An ended try's key stays TryKeep, so the bucket holds thousands; one that waits or runs was asked lately. Only the
+	// newest TryScan keys are read, in parallel (a try id is the time in nanoseconds, so the newest sorts last).
+	sort.Slice(keys, func(i, j int) bool {
+		if len(keys[i]) != len(keys[j]) {
+			return len(keys[i]) > len(keys[j])
+		}
+		return keys[i] > keys[j]
+	})
+	if len(keys) > TryScan {
+		keys = keys[:TryScan]
+	}
+	tries := make([]Try, len(keys))
+	live := make([]bool, len(keys))
+	if err := parallel(len(keys), func(i int) error {
+		raw, err := store.Get(keys[i])
 		if err != nil || len(raw) == 0 {
-			continue
+			return nil
 		}
 		cur, err := DecodeTry(raw)
-		if err != nil || cur.Over() {
+		tries[i], live[i] = cur, err == nil && !cur.Over()
+		return nil
+	}); err != nil {
+		return "", err
+	}
+	var open []string
+	for i, cur := range tries {
+		if !live[i] {
 			continue
 		}
 		if cur.Repository == t.Repository && cur.Head == t.Head && cur.Trunk == t.Trunk && cur.Tests == t.Tests && cur.Level == t.Level {

modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go

@@ -22,6 +22,8 @@ import (
 	"syscall"
 	"time"
 
+	"git.novox.be/novox/mesh-sdk/go/process"
+
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
 
@@ -188,18 +190,18 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	call := callerOn(conn, cred.User)
 	// No forge event is heard while the open pull requests are read and said (ADR 0443 §5): events and syncs take turns.
 	var turn sync.Mutex
-	kv, err := js.KeyValue(ctx, headsBucket)
+	kv, err := process.EssentialBucket(ctx, js, "bridge", headsBucket, "keep the heads it wrote on the forge stream")
 	if err != nil {
-		return fmt.Errorf("the bucket %s: %w", headsBucket, err)
+		return err
 	}
 	heads := kvHeads{kv}
 	// Everything written on FORGE, an event's and a sync's, keeps the head in the bucket.
 	rec := &recorder{pub: js, heads: heads, logger: logger}
 	said := func(ctx context.Context) error { return syncOpen(ctx, call, rec, heads, &turn, logger) }
 	// What the gate asks of the forge (ADR 0437 §8): its owed acts and its merges, each performed once per revision.
-	performedKV, err := js.KeyValue(ctx, performedBucket)
+	performedKV, err := process.EssentialBucket(ctx, js, "bridge", performedBucket, "perform an act of the gate only once")
 	if err != nil {
-		return fmt.Errorf("the bucket %s: %w", performedBucket, err)
+		return err
 	}
 	perf := &performer{call: call, pub: rec, done: kvRevisions{performedKV}, logger: logger}
 	watching := make(chan error, 3)
@@ -207,16 +209,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		bucket string
 		react  func(ctx context.Context, key string, rev uint64, value []byte)
 	}{{delivery.OwedBucket, perf.owed}, {delivery.MergesBucket, perf.merges}} {
-		kv, err := js.KeyValue(ctx, w.bucket)
-		if errors.Is(err, jetstream.ErrBucketNotFound) {
-			// The gate's buckets come with the gate; the bridge does not wait for them (no timer), and is started again
-			// by its delivery when the gate arrives.
-			logger.Printf("the bucket %s does not exist yet: nothing of it is performed until the bridge starts again", w.bucket)
+		// The gate's buckets come with the gate and its grant; the bridge does not wait for them (no timer), and is started
+		// again by its delivery when the gate arrives. A bucket refused or not answered is the same (issue 1945).
+		kv := process.OptionalBucket(ctx, js, w.bucket, "nothing of it is performed until the bridge starts again", logger.Printf)
+		if kv == nil {
 			continue
 		}
-		if err != nil {
-			return fmt.Errorf("the bucket %s: %w", w.bucket, err)
-		}
 		go func() {
 			watching <- watch(ctx, kv, func(key string, rev uint64, value []byte) {
 				hctx, cancel := context.WithTimeout(ctx, syncDeadline)
@@ -231,7 +229,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	go onReconnect(ctx, again, said, logger)
 	cons, err := js.Consumer(ctx, eventsStream, cred.Node+"_"+cred.Module)
 	if err != nil {
-		return fmt.Errorf("the consumer %s_%s on %s: %w", cred.Node, cred.Module, eventsStream, err)
+		// Essential: the forge's events are what the bridge carries.
+		return fmt.Errorf("the consumer %s_%s on %s cannot be read (%v): the bridge hears no forge event without it, so it stops; where the bus refused it or did not answer, this module's consumes (made by the controller with its grant) have not reached the bus yet", cred.Node, cred.Module, eventsStream, err)
 	}
 	cc, err := cons.Consume(func(m jetstream.Msg) {
 		hear(&turn, func() {

modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main_test.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"log"
 	"slices"
 	"strings"
@@ -12,6 +13,8 @@ import (
 	"testing"
 	"time"
 
+	"git.novox.be/novox/mesh-sdk/go/process"
+
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
 
@@ -66,6 +69,16 @@ func body(m map[string]any) []byte {
 }
 
 func TestReadableEventIsWrittenOnForgeAndAcked(t *testing.T) { // 1235/AC2 (unit part; the bus is the integration test)
+	t.Run("1945: a refused bucket is said, or stops the start naming its grant", func(t *testing.T) {
+		refusal := errors.New("nats: permissions violation for publish to $JS.API.STREAM.INFO.KV_x")
+		var said []string
+		if kv := process.OptionalBucket(context.Background(), refusingBus{refusal}, "x", "x is not heard", func(f string, a ...any) { said = append(said, fmt.Sprintf(f, a...)) }); kv != nil || len(said) != 1 || !strings.Contains(said[0], "permissions violation") {
+			t.Errorf("an optional bucket refused: %v, said %v", kv, said)
+		}
+		if _, err := process.EssentialBucket(context.Background(), refusingBus{refusal}, "bridge", "x", "work"); err == nil || !strings.Contains(err.Error(), "grant of x") {
+			t.Errorf("an essential bucket refused: %v, want its grant named", err)
+		}
+	})
 	var logs bytes.Buffer
 	m, pub := &fakeMsg{subject: "mesh.mod.gitea.event.pull.updated", data: body(nil)}, &fakePub{}
 	handle(context.Background(), m, pub, nil, someFiles, log.New(&logs, "", 0))
@@ -230,3 +243,8 @@ func someFiles(_, _, path, _ string) ([]byte, error) {
 	}
 	return nil, bridge.ErrNotFound
 }
+
+// refusingBus refuses every bucket with err (issue 1945).
+type refusingBus struct{ err error }
+
+func (r refusingBus) KeyValue(context.Context, string) (jetstream.KeyValue, error) { return nil, r.err }

modules/mesh-delivery-forge-bridge/go.mod

@@ -3,7 +3,7 @@ module mesh-delivery-forge-bridge
 go 1.26.0
 
 require (
-	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9
+	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585
 	github.com/nats-io/nats-server/v2 v2.11.17
 	github.com/nats-io/nats.go v1.54.0
 )

modules/mesh-delivery-forge-bridge/go.sum

@@ -1,5 +1,7 @@
 git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9 h1:Ufmd2cYEXICQ817IGflccP9bexABE4WhIO49HPbaoSo=
 git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585 h1:7QoFH5qjXRpx5M4uPdYsVJ5qKl2082gq8n0lFxhoNIg=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585/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=

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

@@ -240,29 +240,21 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	if err != nil {
 		return err
 	}
-	bucket := func(name string) (jetstream.KeyValue, error) {
-		kv, err := js.KeyValue(ctx, name)
-		if err != nil {
-			return nil, fmt.Errorf("the bucket %s: %w", name, err)
-		}
-		return kv, nil
-	}
-	candidates, err := bucket(delivery.CandidatesBucket)
+	candidates, err := process.EssentialBucket(ctx, js, "indexer", delivery.CandidatesBucket, "write a candidate's fingerprints")
 	if err != nil {
 		return err
 	}
-	modules, err := bucket(delivery.ModulesBucket)
+	modules, err := process.EssentialBucket(ctx, js, "indexer", delivery.ModulesBucket, "write a module's fingerprint")
 	if err != nil {
 		return err
 	}
-	manifests, err := bucket(delivery.ManifestsBucket)
-	if err != nil {
-		return err
-	}
-	builds, err := bucket(delivery.BuildsBucket)
+	manifests, err := process.EssentialBucket(ctx, js, "indexer", delivery.ManifestsBucket, "keep a module's manifest")
 	if err != nil {
 		return err
 	}
+	// The controller's build records only name the directory of a module whose key names none (fillPaths): without them
+	// such a module waits for its path, every other is indexed (issue 1945).
+	builds := process.OptionalBucket(ctx, js, delivery.BuildsBucket, "a module whose key names no directory is not given one from the controller's build records until the indexer starts again", logger.Printf)
 	known := index.NewBuilds()
 	// The controller's build records name the directory of a module whose key names none (fillPaths); no build of any
 	// module moves a fingerprint since no fingerprint takes the builder (ADR 0530 §1).
@@ -273,12 +265,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 			known.Delete(e.Key())
 		}
 	}
-	w, err := settle(ctx, builds, each)
-	if err != nil {
-		return err
-	}
 	errs := make(chan error, 1)
-	go func() { errs <- follow(ctx, w, each) }()
+	if builds != nil {
+		if w, err := settle(ctx, builds, each); err != nil {
+			process.NotWatched(logger.Printf, delivery.BuildsBucket, err, "a module whose key names no directory is not given one until the indexer starts again")
+		} else {
+			go func() { errs <- follow(ctx, w, each) }()
+		}
+	}
 
 	// The indexer asks the forge for the objects of a manifest's named paths as its own account (ADR 0499 clause 3).
 	account, err := process.ReadAccount(path)
@@ -296,22 +290,17 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	// main is the mesh's source revision, which stamps the modules (issue 1344). Where it does not exist yet the indexer does
 	// not wait for it (that needs a timer, ADR 0437 §8): it says so, and is started again with the gate.
 	sourceErrs := make(chan error, 1)
-	switch source, err := js.KeyValue(ctx, delivery.SourceBucket); {
-	case errors.Is(err, jetstream.ErrBucketNotFound):
-		logger.Printf("the bucket %s does not exist yet: no source revision is read until the indexer starts again", delivery.SourceBucket)
-	case err != nil:
-		return fmt.Errorf("the bucket %s: %w", delivery.SourceBucket, err)
-	default:
+	if source := process.OptionalBucket(ctx, js, delivery.SourceBucket, "no source revision is read until the indexer starts again", logger.Printf); source != nil {
 		each := func(e jetstream.KeyValueEntry) {
 			if opOf(e) {
 				applySource(ctx, ix, e.Key(), e.Revision(), e.Value(), logger)
 			}
 		}
-		sw, err := settle(ctx, source, each)
-		if err != nil {
-			return err
+		if sw, err := settle(ctx, source, each); err != nil {
+			process.NotWatched(logger.Printf, delivery.SourceBucket, err, "no source revision is read until the indexer starts again")
+		} else {
+			go func() { sourceErrs <- follow(ctx, sw, each) }()
 		}
-		go func() { sourceErrs <- follow(ctx, sw, each) }()
 	}
 	// Only the toolchain reports of nodes the build module is assigned to count (issue 1940): the controller's
 	// assignments are settled first, so the toolchains settle through the gate already knowing them.
@@ -381,7 +370,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	}
 	cons, err := js.Consumer(ctx, forgeStream, cred.Node+"_"+cred.Module)
 	if err != nil {
-		return fmt.Errorf("the consumer %s_%s on %s: %w", cred.Node, cred.Module, forgeStream, err)
+		// Essential: the forge's merges are what the indexer indexes.
+		return fmt.Errorf("the consumer %s_%s on %s cannot be read (%v): the indexer hears no merge without it, so it stops; where the bus refused it or did not answer, this module's consumes (made by the controller with its grant) have not reached the bus yet", cred.Node, cred.Module, forgeStream, err)
 	}
 	cc, err := cons.Consume(func(m jetstream.Msg) { handle(ctx, m, ix, logger) },
 		jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) }))

modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/previews.go

@@ -3,7 +3,7 @@ package main
 import (
 	"context"
 	"errors"
-	"fmt"
+	"git.novox.be/novox/mesh-sdk/go/process"
 	"log"
 	"strings"
 	"sync"
@@ -55,11 +55,14 @@ func (w *previewWatch) ensure() error {
 		w.logger.Printf("the bucket %s does not exist yet: it is watched when the checker says it wrote a preview", delivery.PreviewsBucket)
 		return nil
 	case err != nil:
-		return fmt.Errorf("the bucket %s: %w", delivery.PreviewsBucket, err)
+		// Refused or not answered (its grant not on the bus yet, issue 1945): said, and opened again at the checker's next preview.
+		w.logger.Printf("the bucket %s cannot be read (%v): it is watched when the checker says it wrote a preview", delivery.PreviewsBucket, err)
+		return nil
 	}
 	watcher, err := settle(w.ctx, kv, func(e jetstream.KeyValueEntry) { w.ix.PutPreview(e.Key(), e.Value(), !opOf(e)) })
 	if err != nil {
-		return err
+		process.NotWatched(w.logger.Printf, delivery.PreviewsBucket, err, "it is watched when the checker says it wrote a preview")
+		return nil
 	}
 	w.watched = true
 	go func() {

modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/standing.go

@@ -193,7 +193,9 @@ func watchAssignments(ctx context.Context, js bucketOpener, gate *toolchainGate,
 	each := func(e jetstream.KeyValueEntry) { gate.Assignment(e.Key(), e.Value(), !opOf(e), e.Created()) }
 	w, err := settle(ctx, kv, each)
 	if err != nil {
-		return nil, fmt.Errorf("the bucket %s: %w", provider.AssignmentsBucket, err)
+		// Not fatal either (issue 1945): a watch the bus refuses is the same missing grant as a refused read.
+		logger.Printf("the bucket %s cannot be watched, so every node's toolchain report counts, an unassigned build node's too (issue 1940): %v", provider.AssignmentsBucket, err)
+		return errs, nil
 	}
 	if gate.Settled() {
 		changed(ctx)

modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/toolchains.go

@@ -10,6 +10,8 @@ import (
 	"strings"
 	"sync"
 
+	"git.novox.be/novox/mesh-sdk/go/process"
+
 	"github.com/nats-io/nats.go/jetstream"
 
 	"git.novox.be/novox/mesh-sdk/go/delivery"
@@ -73,11 +75,14 @@ func (w *toolchainWatch) ensure(module string) error {
 			w.logger.Printf("the bucket %s does not exist yet: it is watched when %s reports a toolchain", name, module)
 			return nil
 		case err != nil:
-			return fmt.Errorf("the bucket %s: %w", name, err)
+			// Refused or not answered (its grant not on the bus yet, issue 1945): said, and opened again at the module's next report.
+			w.logger.Printf("the bucket %s cannot be read (%v): %s's toolchains are not heard until its next report", name, err, module)
+			return nil
 		}
 		watcher, err := settle(w.ctx, kv, func(e jetstream.KeyValueEntry) { w.sink.put(module, e.Key(), e.Value(), !opOf(e)) })
 		if err != nil {
-			return err
+			process.NotWatched(w.logger.Printf, name, err, fmt.Sprintf("%s's toolchains are not heard until its next report", module))
+			return nil
 		}
 		w.watched[module] = true
 		go func() {

modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/toolchains_test.go

@@ -3,6 +3,7 @@ package main
 import (
 	"bytes"
 	"context"
+	"errors"
 	"log"
 	"strings"
 	"sync"
@@ -21,6 +22,7 @@ type laterBuckets struct {
 	mu      sync.Mutex
 	appears map[string]jetstream.KeyValue // buckets that exist now
 	asked   map[string]int
+	refuse  error // answered for a bucket that does not exist, when set: a bus that refuses it (issue 1945)
 }
 
 func (l *laterBuckets) KeyValue(_ context.Context, name string) (jetstream.KeyValue, error) {
@@ -30,6 +32,9 @@ func (l *laterBuckets) KeyValue(_ context.Context, name string) (jetstream.KeyVa
 	if kv, ok := l.appears[name]; ok {
 		return kv, nil
 	}
+	if l.refuse != nil {
+		return nil, l.refuse
+	}
 	return nil, jetstream.ErrBucketNotFound
 }
 
@@ -100,6 +105,12 @@ func TestABuildModulesBucketThatAppearsLaterIsWatchedWhenItsEventArrives(t *test
 		recompute: func(context.Context) error { mu.Lock(); recomputes++; mu.Unlock(); return nil },
 	}
 	var logs bytes.Buffer
+	// 1945: a bus that refuses both buckets, or does not answer, leaves them unwatched, said, and the indexer running.
+	refusing := &laterBuckets{appears: map[string]jetstream.KeyValue{}, asked: map[string]int{}, refuse: errors.New("context deadline exceeded")}
+	if err := newToolchainWatch(ctx, refusing, sink, log.New(&logs, "", 0)).start(); err != nil || !strings.Contains(logs.String(), "cannot be read") {
+		t.Fatalf("a refused toolchains bucket: start %v, log %q; want it said and the start going on", err, logs.String())
+	}
+	logs.Reset()
 	w := newToolchainWatch(ctx, buckets, sink, log.New(&logs, "", 0))
 	if err := w.start(); err != nil { // neither bucket exists: nothing is watched, nothing is written, nobody waits
 		t.Fatal(err)

modules/mesh-delivery-indexer/go.mod

@@ -3,7 +3,7 @@ module mesh-delivery-indexer
 go 1.26.0
 
 require (
-	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468
+	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585
 	github.com/nats-io/nats-server/v2 v2.11.17
 	github.com/nats-io/nats.go v1.54.0
 )

modules/mesh-delivery-indexer/go.sum

@@ -1,5 +1,7 @@
 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.20261023150321-a2c813a6b468/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585 h1:7QoFH5qjXRpx5M4uPdYsVJ5qKl2082gq8n0lFxhoNIg=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585/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=

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

@@ -22,6 +22,8 @@ import (
 	"strings"
 	"syscall"
 
+	"git.novox.be/novox/mesh-sdk/go/process"
+
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
 
@@ -119,12 +121,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		return nil
 	}
 	if err != nil {
-		return fmt.Errorf("the bucket %s: %w", delivery.AsksBucket, err)
+		// Essential: the gate's asks are what the teller tells (issue 1945: the error names the grant).
+		return process.EssentialErr("teller", delivery.AsksBucket, "tell any of the gate's asks", err)
 	}
+	// The answers it writes and what it asked: its own work, so each stops the start, naming its grant.
 	var kvs [2]jetstream.KeyValue
 	for i, name := range []string{delivery.AnswersBucket, askedBucket} {
-		if kvs[i], err = js.KeyValue(ctx, name); err != nil {
-			return fmt.Errorf("the bucket %s: %w", name, err)
+		if kvs[i], err = process.EssentialBucket(ctx, js, "teller", name, "write an answer or remember what it asked"); err != nil {
+			return err
 		}
 	}
 	state := kvStore{ctx: ctx, kv: kvs[1]}
@@ -140,7 +144,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	// The view's answers to the assigner's asks (issue 1297 B10): both buckets are the controller's to create, and a bucket not
 	// there is said and leaves the teller without that path rather than without the gate's asks (a wait for it needs a timer).
 	failed := make(chan error, 2)
-	assignerKV, viewKV := optionalBucket(ctx, js, delivery.AssignerAsksBucket, logger), optionalBucket(ctx, js, delivery.ViewAnswersBucket, logger)
+	assignerKV, viewKV := process.OptionalBucket(ctx, js, delivery.AssignerAsksBucket, teller, logger.Printf), process.OptionalBucket(ctx, js, delivery.ViewAnswersBucket, teller, logger.Printf)
 	// The assigner's asks are raised on the operator channel whether or not the view's answers can be read (issue 1906).
 	if viewKV != nil {
 		t.View = kvGate{ctx: ctx, kv: viewKV}
@@ -172,15 +176,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	}
 }
 
-// optionalBucket is a bucket the teller reads when it exists, and nil, said once, when it does not.
-func optionalBucket(ctx context.Context, js jetstream.JetStream, name string, logger *log.Logger) jetstream.KeyValue {
-	kv, err := js.KeyValue(ctx, name)
-	if err != nil {
-		logger.Printf("the bucket %s cannot be read (%v): the asks and answers it holds are not carried until the teller is started again with it", name, err)
-		return nil
-	}
-	return kv
-}
+// teller is what an optional bucket of the teller leaves out.
+const teller = "the asks and answers it holds are not carried until the teller is started again with it"
 
 // askedBucket is the teller's state bucket asked, `<module>_<name>`: what it asked of the router and which condition it
 // raised (manifest state entry asked, novox/hq ADR 0437 §4).

modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller_test.go

@@ -1,12 +1,17 @@
 package main
 
 import (
+	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"strings"
 	"testing"
 	"time"
 
+	"git.novox.be/novox/mesh-sdk/go/process"
+	"github.com/nats-io/nats.go/jetstream"
+
 	stdio "git.novox.be/novox/mesh-sdk/go"
 	"git.novox.be/novox/mesh-sdk/go/asker"
 	"git.novox.be/novox/mesh-sdk/go/asks"
@@ -236,6 +241,16 @@ func (b *bed) askedOf(subject string) []published {
 }
 
 func TestAnAsksKeyBecomesOneRouterAskAndDeletingItCancelsIt(t *testing.T) { // 1343/AC1
+	t.Run("1945: a refused bucket is said, or stops the start naming its grant", func(t *testing.T) {
+		refusal := errors.New("nats: permissions violation for publish to $JS.API.STREAM.INFO.KV_x")
+		var said []string
+		if kv := process.OptionalBucket(context.Background(), refusingBus{refusal}, "x", "x is not heard", func(f string, a ...any) { said = append(said, fmt.Sprintf(f, a...)) }); kv != nil || len(said) != 1 || !strings.Contains(said[0], "permissions violation") {
+			t.Errorf("an optional bucket refused: %v, said %v", kv, said)
+		}
+		if _, err := process.EssentialBucket(context.Background(), refusingBus{refusal}, "teller", "x", "work"); err == nil || !strings.Contains(err.Error(), "grant of x") {
+			t.Errorf("an essential bucket refused: %v, want its grant named", err)
+		}
+	})
 	b := newBed(t)
 	key, value, ga := gateAsk(t, twoHeads(), twoHeads())
 	b.gate[key] = value
@@ -518,3 +533,8 @@ func TestAnAnswerToAnEarlierAskingDoesNotBlockTheNextAsking(t *testing.T) { // R
 		t.Errorf("writes %d asks %d dropped %v", b.answers.writes, len(b.askedOf(asks.AskSubject(delivery.TellerModule))), b.dropped)
 	}
 }
+
+// refusingBus refuses every bucket with err (issue 1945).
+type refusingBus struct{ err error }
+
+func (r refusingBus) KeyValue(context.Context, string) (jetstream.KeyValue, error) { return nil, r.err }

modules/mesh-delivery-teller/go.mod

@@ -3,7 +3,7 @@ module mesh-delivery-teller
 go 1.26.0
 
 require (
-	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025203528-db26f7e8b6f5
+	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585
 	github.com/nats-io/nats-server/v2 v2.11.17
 	github.com/nats-io/nats.go v1.54.0
 )

modules/mesh-delivery-teller/go.sum

@@ -6,6 +6,8 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025200810-d4ae4b6790b8 h1:vhYk/iv4
 git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025200810-d4ae4b6790b8/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
 git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025203528-db26f7e8b6f5 h1:gE80bR0tPME7MTMBLcdP62v+F/9soVXmdcBYNMOye44=
 git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025203528-db26f7e8b6f5/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585 h1:7QoFH5qjXRpx5M4uPdYsVJ5qKl2082gq8n0lFxhoNIg=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585/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=

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

@@ -9,8 +9,9 @@ import (
 	"os"
 	"strings"
 
-	stdio "git.novox.be/novox/mesh-sdk/go"
 	"mesh-delivery-view/internal/view"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
 )
 
 // Tools are the read tools over a modules state.
@@ -85,7 +86,8 @@ func Tools(r view.Reader) []stdio.Tool {
 			}},
 		{Name: view.WantsReadTool,
 			Description: "The assigner's wants and holds as stored in mesh-delivery-assigner_wants: wants by key (want.<node>.<module>) and " +
-				"held fingerprints by key (held.<module>.<hex>). Narrow by node and module (a hold has no node and is shown when no node is given). Read-only.",
+				"held fingerprints by key (held.<module>.<hex>). Narrow by node and module (a hold has no node and is shown when no node is given); a node and a module read that one key. " +
+				"At most 1000 values a call; the answer says when it was cut (cut, note). Read-only.",
 			Input: map[string]any{"type": "object", "properties": nodeModule},
 			Run: func(args map[string]any) (any, error) {
 				node, module := nodeModuleOf(args)
@@ -100,7 +102,8 @@ func Tools(r view.Reader) []stdio.Tool {
 			}},
 		{Name: view.DifferencesReadTool,
 			Description: "The assigner's differences between a want and what the controller sends, as stored in mesh-delivery-assigner_differences, " +
-				"by key (<node>.<module>), with the rule that explains each or unexplained. Narrow by node and module. Read-only.",
+				"by key (<node>.<module>), with the rule that explains each or unexplained. Narrow by node and module (both: that one key). " +
+				"At most 1000 values a call; the answer says when it was cut (cut, note). Read-only.",
 			Input: map[string]any{"type": "object", "properties": nodeModule},
 			Run: func(args map[string]any) (any, error) {
 				node, module := nodeModuleOf(args)
@@ -108,7 +111,8 @@ func Tools(r view.Reader) []stdio.Tool {
 			}},
 		{Name: view.HavesReadTool,
 			Description: "What each node has, as the node-engine stored it in node-engine_have: by key <node>.<module> (the fingerprint it runs, its phase and health), " +
-				"and <node>.node (the node's own facts). Narrow by node and module. Read-only.",
+				"and <node>.node (the node's own facts). Narrow by node and module (both: that one key). " +
+				"At most 1000 values a call; the answer says when it was cut (cut, note). Read-only.",
 			Input: map[string]any{"type": "object", "properties": nodeModule},
 			Run: func(args map[string]any) (any, error) {
 				node, module := nodeModuleOf(args)

modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main_test.go

@@ -4,7 +4,9 @@ import (
 	"encoding/json"
 	"fmt"
 	"strings"
+	"sync/atomic"
 	"testing"
+	"time"
 
 	"mesh-delivery-view/internal/view"
 )
@@ -164,7 +166,7 @@ func TestBuilderBuildJoinsTheIndexersFingerprint(t *testing.T) {
 // bigBuilds is a bucket of n keys over 100 modules that counts the values read.
 type bigBuilds struct {
 	fake  fake
-	reads *int
+	reads *atomic.Int64
 }
 
 func newBig(n int) bigBuilds {
@@ -172,15 +174,15 @@ func newBig(n int) bigBuilds {
 	for i := 0; i < n; i++ {
 		f[fmt.Sprintf("mod%02d.%064x", i%100, i)] = doneBuild
 	}
-	return bigBuilds{fake: f, reads: new(int)}
+	return bigBuilds{fake: f, reads: new(atomic.Int64)}
 }
 
 func (b bigBuilds) reader() view.Reader {
 	r := b.fake.reader()
 	get := r.State.Get
-	r.Builds = view.StateRead{Keys: r.State.Keys, Get: func(k string) (json.RawMessage, error) { *b.reads++; return get(k) },
+	r.Builds = view.StateRead{Keys: r.State.Keys, Get: func(k string) (json.RawMessage, error) { b.reads.Add(1); return get(k) },
 		GetEntry: func(k string) (json.RawMessage, uint64, error) {
-			*b.reads++
+			b.reads.Add(1)
 			raw, err := get(k)
 			var rev uint64
 			fmt.Sscanf(k[strings.Index(k, ".")+1:], "%x", &rev) // the fake's revision is the key's number
@@ -198,8 +200,8 @@ func TestBuilderBuildsFiltersByPrefixBeforeAnyGet(t *testing.T) {
 		t.Fatal(err)
 	}
 	a := out.(view.BuildsAnswer)
-	if *b.reads != 20 {
-		t.Errorf("read %d values for a module of 20 keys, want 20", *b.reads)
+	if b.reads.Load() != 20 {
+		t.Errorf("read %d values for a module of 20 keys, want 20", b.reads.Load())
 	}
 	if len(a.Builds) != 5 || !a.Cut || !strings.Contains(a.Note, "5") {
 		t.Errorf("answer = %d rows, cut %v, note %q", len(a.Builds), a.Cut, a.Note)
@@ -222,13 +224,13 @@ func TestBuilderBuildsWithoutAModuleIsBounded(t *testing.T) {
 		t.Fatal(err)
 	}
 	a := out.(view.BuildsAnswer)
-	if len(a.Builds) != view.DefaultBuildsLimit || !a.Cut || *b.reads > view.DefaultBuildsLimit {
-		t.Errorf("%d rows, cut %v, %d reads", len(a.Builds), a.Cut, *b.reads)
+	if len(a.Builds) != view.DefaultBuildsLimit || !a.Cut || b.reads.Load() > view.DefaultBuildsLimit {
+		t.Errorf("%d rows, cut %v, %d reads", len(a.Builds), a.Cut, b.reads.Load())
 	}
 	b = newBig(2000)
 	out, _ = run(t, b.reader(), view.BuilderBuildsReadTool, map[string]any{"state": "failed"})
-	if a := out.(view.BuildsAnswer); *b.reads != view.MaxBuildReads || !a.Cut || len(a.Builds) != 0 {
-		t.Errorf("a state matching nothing: %d reads (want the bound %d), cut %v", *b.reads, view.MaxBuildReads, a.Cut)
+	if a := out.(view.BuildsAnswer); b.reads.Load() != view.MaxBuildReads || !a.Cut || len(a.Builds) != 0 {
+		t.Errorf("a state matching nothing: %d reads (want the bound %d), cut %v", b.reads.Load(), view.MaxBuildReads, a.Cut)
 	}
 }
 
@@ -280,6 +282,41 @@ func keptReader() view.Reader {
 	}
 }
 
+// slowBucket is a bucket of the live shape (2026-10-25: node_haves and assigner_wants timed out at 30 s): keys over 4 nodes
+// and perNode modules each, written by key(node, module), whose Get takes delay and counts the values read, the lists of
+// the bucket and the most Gets in flight at once.
+type slowBucket struct {
+	fake                fake
+	delay               time.Duration
+	reads, lists        atomic.Int64
+	inFlight, maxFlight atomic.Int64
+}
+
+func newSlow(perNode int, delay time.Duration, key func(node string, i int) string) *slowBucket {
+	b := &slowBucket{fake: fake{}, delay: delay}
+	for _, n := range []string{"ace", "g14", "novox", "shanks"} {
+		for i := 0; i < perNode; i++ {
+			b.fake[key(n, i)] = `{"fp":"x"}`
+		}
+	}
+	return b
+}
+
+func (b *slowBucket) state() view.StateRead {
+	f := b.fake.reader().State
+	return view.StateRead{
+		Keys: func() ([]string, error) { b.lists.Add(1); return f.Keys() },
+		Get: func(k string) (json.RawMessage, error) {
+			b.reads.Add(1)
+			now := b.inFlight.Add(1)
+			for m := b.maxFlight.Load(); now > m && !b.maxFlight.CompareAndSwap(m, now); m = b.maxFlight.Load() {
+			}
+			time.Sleep(b.delay)
+			b.inFlight.Add(-1)
+			return f.Get(k)
+		}}
+}
+
 func TestAssignerWantsToolAnswersAsStoredAndNarrows(t *testing.T) {
 	r := keptReader()
 	out, err := run(t, r, view.WantsReadTool, map[string]any{})
@@ -298,6 +335,39 @@ func TestAssignerWantsToolAnswersAsStoredAndNarrows(t *testing.T) {
 	if got := out.(view.WantsAnswer).Wants; len(got) != 2 {
 		t.Errorf("module zsh = %v", got)
 	}
+
+	// On the live shape (4 nodes, 150 wants each, 1 ms a Get): a node and a module read that one key and list nothing;
+	// a node reads only its own keys; the whole answer reads its values in parallel.
+	slow := newSlow(150, time.Millisecond, func(n string, i int) string { return fmt.Sprintf("want.%s.m%03d", n, i) })
+	r = view.Reader{WantsState: slow.state()}
+	for _, row := range []struct {
+		args         map[string]any
+		wants        int
+		reads, lists int64
+	}{
+		{map[string]any{"node": "ace", "module": "m007"}, 1, 1, 0},
+		{map[string]any{"node": "ace", "module": "gone"}, 0, 1, 0},
+		{map[string]any{"node": "g14"}, 150, 150, 1},
+		{map[string]any{"module": "m007"}, 4, 4, 1},
+		{map[string]any{}, 600, 600, 1},
+	} {
+		slow.reads.Store(0)
+		slow.lists.Store(0)
+		start := time.Now()
+		out, err := run(t, r, view.WantsReadTool, row.args)
+		took := time.Since(start)
+		if err != nil {
+			t.Fatalf("%v: %v", row.args, err)
+		}
+		a := out.(view.WantsAnswer)
+		if len(a.Wants) != row.wants || slow.reads.Load() != row.reads || slow.lists.Load() != row.lists || a.Cut {
+			t.Errorf("%v: %d wants, %d reads, %d lists, cut %v; want %d, %d, %d", row.args, len(a.Wants), slow.reads.Load(), slow.lists.Load(), a.Cut, row.wants, row.reads, row.lists)
+		}
+		t.Logf("assigner_wants %v: %d values in %v (one after the other: %v)", row.args, slow.reads.Load(), took, time.Duration(slow.reads.Load())*slow.delay)
+	}
+	if slow.maxFlight.Load() < 2 {
+		t.Errorf("the values were read one at a time (at most %d in flight)", slow.maxFlight.Load())
+	}
 }
 
 func TestAssignerWantToolAnswersOneAndSaysWhenThereIsNone(t *testing.T) {
@@ -328,4 +398,39 @@ func TestAssignerDifferencesAndNodeHavesToolsAnswerAsStored(t *testing.T) {
 	if got := out.(view.KeptAnswer).Keys; len(got) != 2 || got["ace.node"] == nil {
 		t.Errorf("haves of ace (the node's own key too) = %v", got)
 	}
+
+	// More haves than one call reads (4 nodes, 300 each, 1 ms a Get): a node and a module read that one key and list
+	// nothing; the whole answer reads MaxStateReads values in parallel and says it was cut.
+	slow := newSlow(300, time.Millisecond, func(n string, i int) string { return fmt.Sprintf("%s.m%03d", n, i) })
+	r = view.Reader{HavesState: slow.state(), DifferencesState: slow.state()}
+	for _, row := range []struct {
+		tool         string
+		args         map[string]any
+		keys         int
+		reads, lists int64
+		cut          bool
+	}{
+		{view.HavesReadTool, map[string]any{"node": "shanks", "module": "m042"}, 1, 1, 0, false},
+		{view.DifferencesReadTool, map[string]any{"node": "shanks", "module": "m042"}, 1, 1, 0, false},
+		{view.HavesReadTool, map[string]any{"node": "ace"}, 300, 300, 1, false},
+		{view.HavesReadTool, map[string]any{"module": "m042"}, 4, 4, 1, false},
+		{view.HavesReadTool, map[string]any{}, view.MaxStateReads, view.MaxStateReads, 1, true},
+	} {
+		slow.reads.Store(0)
+		slow.lists.Store(0)
+		start := time.Now()
+		out, err := run(t, r, row.tool, row.args)
+		took := time.Since(start)
+		if err != nil {
+			t.Fatalf("%s %v: %v", row.tool, row.args, err)
+		}
+		a := out.(view.KeptAnswer)
+		if len(a.Keys) != row.keys || slow.reads.Load() != row.reads || slow.lists.Load() != row.lists || a.Cut != row.cut || (a.Cut && !strings.Contains(a.Note, "narrow")) {
+			t.Errorf("%s %v: %d keys, %d reads, %d lists, cut %v %q; want %d, %d, %d, %v", row.tool, row.args, len(a.Keys), slow.reads.Load(), slow.lists.Load(), a.Cut, a.Note, row.keys, row.reads, row.lists, row.cut)
+		}
+		t.Logf("%s %v: %d values in %v (one after the other: %v)", row.tool, row.args, slow.reads.Load(), took, time.Duration(slow.reads.Load())*slow.delay)
+	}
+	if slow.maxFlight.Load() < 2 {
+		t.Errorf("the values were read one at a time (at most %d in flight)", slow.maxFlight.Load())
+	}
 }

modules/mesh-delivery-view/go.mod

@@ -3,7 +3,7 @@ module mesh-delivery-view
 go 1.26.0
 
 require (
-	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024210118-748c564cdf7c
+	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025224357-095d41fa3e81
 	github.com/nats-io/nats-server/v2 v2.11.17
 	github.com/nats-io/nats.go v1.54.0
 )

modules/mesh-delivery-view/go.sum

@@ -1,5 +1,5 @@
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024210118-748c564cdf7c h1:pY2YpHSSsfDPBrOEiWTNSIPn0zdnAayRmtmmyQ3hCHY=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024210118-748c564cdf7c/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025224357-095d41fa3e81 h1:LhljEtt7FbNpVn67eOQbdsFDS8Qj6WPUF6l9IXr9mo4=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025224357-095d41fa3e81/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=

modules/mesh-delivery-view/internal/view/read.go

@@ -10,6 +10,8 @@ import (
 	"slices"
 	"sort"
 	"strings"
+	"sync"
+	"sync/atomic"
 
 	"git.novox.be/novox/mesh-sdk/go/delivery"
 )
@@ -70,12 +72,13 @@ func (r Reader) Modules(none2 bool) (map[string]json.RawMessage, error) {
 		return nil, err
 	}
 	sort.Strings(keys)
+	values, err := r.State.values(keys)
+	if err != nil {
+		return nil, err
+	}
 	out := map[string]json.RawMessage{}
-	for _, k := range keys {
-		raw, err := r.State.Get(k)
-		if err != nil {
-			return nil, err
-		}
+	for i, k := range keys {
+		raw := values[i]
 		if len(raw) == 0 {
 			continue
 		}
@@ -129,13 +132,14 @@ func (r Reader) Candidates(module string) (map[string]json.RawMessage, []string,
 		return nil, nil, err
 	}
 	sort.Strings(keys)
+	values, err := r.CandidatesState.values(keys)
+	if err != nil {
+		return nil, nil, err
+	}
 	out := map[string]json.RawMessage{}
 	var unreadable []string
-	for _, k := range keys {
-		raw, err := r.CandidatesState.Get(k)
-		if err != nil {
-			return nil, nil, err
-		}
+	for i, k := range keys {
+		raw := values[i]
 		if len(raw) == 0 {
 			continue
 		}
@@ -242,35 +246,51 @@ func (r Reader) BuilderBuilds(module, state string, limit int) (BuildsAnswer, er
 		}
 		candidates = append(candidates, k)
 	}
+	// The values are read in batches, each batch in parallel and taken in key order: without a module or a state a batch
+	// is only the rows still missing, so no more is read than the answer needs; otherwise up to readWorkers*4.
 	reads := 0
-	for _, k := range candidates {
+walk:
+	for reads < len(candidates) {
 		if reads >= MaxBuildReads {
 			out.Cut, out.Note = true, fmt.Sprintf("read %d values and stopped; give a module (and a state) to narrow the answer", MaxBuildReads)
 			break
 		}
-		reads++
-		row, ok, err := r.builderRow(k)
-		if err != nil {
+		size := readWorkers * 4
+		if prefix == "" && state == "" {
+			size = limit - len(out.Builds)
+		}
+		size = min(size, MaxBuildReads-reads, len(candidates)-reads)
+		batch := candidates[reads : reads+size]
+		rows := make([]BuilderRow, len(batch))
+		found := make([]bool, len(batch))
+		if err := parallel(len(batch), func(i int) error {
+			var err error
+			rows[i], found[i], err = r.builderRow(batch[i])
+			return err
+		}); err != nil {
 			return BuildsAnswer{}, err
 		}
-		if !ok {
-			continue
-		}
-		if state != "" {
-			var b struct {
-				State string `json:"state"`
-			}
-			if json.Unmarshal(row.Build, &b) != nil || b.State != state {
+		reads += len(batch)
+		for i, row := range rows {
+			if !found[i] {
 				continue
 			}
-		}
-		out.Builds = append(out.Builds, row)
-		// Without a module the walk is in key order and stops at the limit; with one, all its (few) keys are read to order them.
-		if prefix == "" && len(out.Builds) >= limit {
-			if reads < len(candidates) {
-				out.Cut, out.Note = true, fmt.Sprintf("stopped at the limit of %d builds in key order, not newest first; give a module for its newest", limit)
+			if state != "" {
+				var b struct {
+					State string `json:"state"`
+				}
+				if json.Unmarshal(row.Build, &b) != nil || b.State != state {
+					continue
+				}
+			}
+			out.Builds = append(out.Builds, row)
+			// Without a module the walk is in key order and stops at the limit; with one, all its (few) keys are read to order them.
+			if prefix == "" && len(out.Builds) >= limit {
+				if reads-len(batch)+i+1 < len(candidates) {
+					out.Cut, out.Note = true, fmt.Sprintf("stopped at the limit of %d builds in key order, not newest first; give a module for its newest", limit)
+				}
+				break walk
 			}
-			break
 		}
 	}
 	sort.SliceStable(out.Builds, func(i, j int) bool {
@@ -352,6 +372,9 @@ type WantsAnswer struct {
 	Wants      map[string]json.RawMessage `json:"wants"`
 	Held       map[string]json.RawMessage `json:"held"`
 	Unreadable []string                   `json:"unreadable,omitempty"`
+	// Cut says the answer stopped at MaxStateReads values, and Note how to narrow it.
+	Cut  bool   `json:"cut,omitempty"`
+	Note string `json:"note,omitempty"`
 }
 
 // KeptAnswer is the answer of assigner_differences and node_haves: the keys of the bucket as stored, and the keys that
@@ -359,60 +382,158 @@ type WantsAnswer struct {
 type KeptAnswer struct {
 	Keys       map[string]json.RawMessage `json:"keys"`
 	Unreadable []string                   `json:"unreadable,omitempty"`
+	Cut        bool                       `json:"cut,omitempty"`
+	Note       string                     `json:"note,omitempty"`
 }
 
-func (s StateRead) all(what string) (map[string]json.RawMessage, error) {
-	if s.Keys == nil || s.Get == nil {
-		return nil, fmt.Errorf("%s cannot be read here", what)
-	}
-	keys, err := s.Keys()
-	if err != nil {
-		return nil, err
+// readWorkers is how many values a read tool asks the bus for at once. Each value is a round trip from the tool through the
+// runtime to the bus (the runtime opens the bucket and gets the key: two bus requests), so reading a bucket of a few
+// hundred keys one after the other outlasted the 30 s a tool call is given on 2026-10-25 (node_haves, assigner_wants).
+const readWorkers = 16
+
+// MaxStateReads bounds the values one call of assigner_wants, assigner_differences or node_haves reads; past it the
+// answer says it was cut and how to narrow it.
+const MaxStateReads = 1000
+
+// parallel runs read(i) for i in [0, n), at most readWorkers at a time, and answers the first error; once one failed no
+// further read starts.
+func parallel(n int, read func(i int) error) error {
+	var (
+		next   atomic.Int64
+		failed atomic.Bool
+		once   sync.Once
+		first  error
+		wg     sync.WaitGroup
+	)
+	workers := min(readWorkers, n)
+	for w := 0; w < workers; w++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for !failed.Load() {
+				i := int(next.Add(1) - 1)
+				if i >= n {
+					return
+				}
+				if err := read(i); err != nil {
+					once.Do(func() { first = err })
+					failed.Store(true)
+				}
+			}
+		}()
 	}
+	wg.Wait()
+	return first
+}
+
+// values reads the stored value of each key, in parallel, in the order of keys (nil for a key with none).
+func (s StateRead) values(keys []string) ([]json.RawMessage, error) {
+	out := make([]json.RawMessage, len(keys))
+	err := parallel(len(keys), func(i int) error {
+		raw, err := s.Get(keys[i])
+		out[i] = raw
+		return err
+	})
+	return out, err
+}
+
+// read reads the values of keys (sorted first) into a map, at most MaxStateReads of them; cut says it stopped there.
+func (s StateRead) read(keys []string) (map[string]json.RawMessage, bool, error) {
 	sort.Strings(keys)
+	cut := len(keys) > MaxStateReads
+	if cut {
+		keys = keys[:MaxStateReads]
+	}
+	values, err := s.values(keys)
+	if err != nil {
+		return nil, false, err
+	}
 	out := map[string]json.RawMessage{}
-	for _, k := range keys {
-		raw, err := s.Get(k)
-		if err != nil {
-			return nil, err
-		}
-		if len(raw) > 0 {
-			out[k] = raw
+	for i, k := range keys {
+		if len(values[i]) > 0 {
+			out[k] = values[i]
 		}
 	}
-	return out, nil
+	return out, cut, nil
+}
+
+func cutNote(cut bool, n int) string {
+	if !cut {
+		return ""
+	}
+	return fmt.Sprintf("read the first %d of %d matching keys in key order and stopped; give a node and a module to narrow the answer", MaxStateReads, n)
 }
 
 // Wants is every want and hold of the assigner, narrowed by node and module (a hold has no node and is kept when the module
 // matches).
 func (r Reader) Wants(node, module string) (WantsAnswer, error) {
-	all, err := r.WantsState.all("the assigner's wants")
+	s := r.WantsState
+	if s.Get == nil {
+		return WantsAnswer{}, fmt.Errorf("the assigner's wants cannot be read here")
+	}
+	out := WantsAnswer{Wants: map[string]json.RawMessage{}, Held: map[string]json.RawMessage{}}
+	if node != "" && module != "" {
+		// One key, read directly: no list of the bucket (a hold has no node, so none is shown).
+		key, err := delivery.WantKey(node, module)
+		if module == "node" {
+			key, err = delivery.WantNodeKey(node)
+		}
+		if err != nil {
+			return WantsAnswer{}, err
+		}
+		raw, err := s.Get(key)
+		if err != nil {
+			return WantsAnswer{}, err
+		}
+		if len(raw) > 0 {
+			out.Wants[key] = raw
+		}
+		return out, nil
+	}
+	if s.Keys == nil {
+		return WantsAnswer{}, fmt.Errorf("the assigner's wants cannot be read here")
+	}
+	keys, err := s.Keys()
 	if err != nil {
 		return WantsAnswer{}, err
 	}
-	out := WantsAnswer{Wants: map[string]json.RawMessage{}, Held: map[string]json.RawMessage{}}
-	for k, raw := range all {
+	// Narrowed by the key before any value is read.
+	var matching []string
+	for _, k := range keys {
 		if n, m, err := delivery.ParseWantKey(k); err == nil {
 			if (node == "" || n == node) && (module == "" || m == module) {
-				out.Wants[k] = raw
+				matching = append(matching, k)
 			}
 			continue
 		}
 		// A node's own want, want.<node>.node (ADR 0539 clause 6): listed with the wants, under the module name "node".
 		if n, err := delivery.ParseWantNodeKey(k); err == nil {
 			if (node == "" || n == node) && (module == "" || module == "node") {
-				out.Wants[k] = raw
+				matching = append(matching, k)
 			}
 			continue
 		}
 		if m, _, err := delivery.ParseHeldKey(k); err == nil {
 			if node == "" && (module == "" || m == module) {
-				out.Held[k] = raw
+				matching = append(matching, k)
 			}
 			continue
 		}
 		out.Unreadable = append(out.Unreadable, k)
 	}
+	sort.Strings(out.Unreadable)
+	values, cut, err := s.read(matching)
+	if err != nil {
+		return WantsAnswer{}, err
+	}
+	out.Cut, out.Note = cut, cutNote(cut, len(matching))
+	for k, raw := range values {
+		if _, _, err := delivery.ParseHeldKey(k); err == nil {
+			out.Held[k] = raw
+		} else {
+			out.Wants[k] = raw
+		}
+	}
 	return out, nil
 }
 
@@ -436,21 +557,50 @@ func (r Reader) Want(node, module string) (json.RawMessage, error) {
 }
 
 func (r Reader) kept(s StateRead, what, node, module string) (KeptAnswer, error) {
-	all, err := s.all(what)
+	if s.Get == nil {
+		return KeptAnswer{}, fmt.Errorf("%s cannot be read here", what)
+	}
+	out := KeptAnswer{Keys: map[string]json.RawMessage{}}
+	if node != "" && module != "" {
+		// One key, read directly: no list of the bucket.
+		key, err := delivery.HaveKey(node, module)
+		if err != nil {
+			return KeptAnswer{}, err
+		}
+		raw, err := s.Get(key)
+		if err != nil {
+			return KeptAnswer{}, err
+		}
+		if len(raw) > 0 {
+			out.Keys[key] = raw
+		}
+		return out, nil
+	}
+	if s.Keys == nil {
+		return KeptAnswer{}, fmt.Errorf("%s cannot be read here", what)
+	}
+	keys, err := s.Keys()
 	if err != nil {
 		return KeptAnswer{}, err
 	}
-	out := KeptAnswer{Keys: map[string]json.RawMessage{}}
-	for k, raw := range all {
+	// Narrowed by the key before any value is read.
+	var matching []string
+	for _, k := range keys {
 		n, m, isNode, err := delivery.ParseHaveKey(k)
 		if err != nil {
 			out.Unreadable = append(out.Unreadable, k)
 			continue
 		}
 		if (node == "" || n == node) && (module == "" || (!isNode && m == module)) {
-			out.Keys[k] = raw
+			matching = append(matching, k)
 		}
 	}
+	sort.Strings(out.Unreadable)
+	values, cut, err := s.read(matching)
+	if err != nil {
+		return KeptAnswer{}, err
+	}
+	out.Keys, out.Cut, out.Note = values, cut, cutNote(cut, len(matching))
 	return out, nil
 }
 

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

@@ -22,6 +22,8 @@ import (
 	"syscall"
 	"time"
 
+	"git.novox.be/novox/mesh-sdk/go/process"
+
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
 
@@ -222,23 +224,23 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	if err != nil {
 		return err
 	}
-	bucket := func(name string) (jetstream.KeyValue, error) {
-		kv, err := js.KeyValue(ctx, name)
-		if err != nil {
-			return nil, fmt.Errorf("the bucket %s: %w", name, err)
-		}
-		return kv, nil
-	}
+	// Its own presence, pulses and stuck rows are what it keeps: each stops the start, naming its grant. The assigner's wants
+	// and the node-engine's haves only make the pulses of a move: without one the watchman watches presence and the build
+	// seats' steps, and says that moves are not watched (issue 1945).
 	var kvs struct{ presence, pulse, stuck, wants, haves jetstream.KeyValue }
 	for _, b := range []struct {
 		name string
 		into *jetstream.KeyValue
-	}{{delivery.PresenceBucket, &kvs.presence}, {delivery.PulseBucket, &kvs.pulse}, {delivery.StuckBucket, &kvs.stuck},
-		{delivery.AssignerWantsBucket, &kvs.wants}, {delivery.HaveBucket, &kvs.haves}} {
-		if *b.into, err = bucket(b.name); err != nil {
+	}{{delivery.PresenceBucket, &kvs.presence}, {delivery.PulseBucket, &kvs.pulse}, {delivery.StuckBucket, &kvs.stuck}} {
+		if *b.into, err = process.EssentialBucket(ctx, js, "watchman", b.name, "say who is online or what stopped moving"); err != nil {
 			return err
 		}
 	}
+	kvs.wants = process.OptionalBucket(ctx, js, delivery.AssignerWantsBucket, "a want makes no pulse, so a move that stops is not seen, until the watchman starts again", logger.Printf)
+	kvs.haves = process.OptionalBucket(ctx, js, delivery.HaveBucket, "a want makes no pulse, so a move that stops is not seen, until the watchman starts again", logger.Printf)
+	if kvs.wants == nil || kvs.haves == nil {
+		kvs.wants, kvs.haves = nil, nil // a want is a pulse only against the have that ends it
+	}
 
 	w := watch.New(&kvOut{ctx: ctx, presence: kvs.presence, pulse: kvs.pulse, stuk: kvs.stuck}, time.Now)
 	w.Logf(logger.Printf)
@@ -283,13 +285,15 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	}
 	// What the node agents run, then the assigner's wants (issue 1906, ADR 0539 §19): a want is a pulse owed by its node's agent
 	// until the have names its fingerprint, so the haves are read first and a want already applied makes no pulse.
-	haveSeen, err := settle(ctx, kvs.haves, func(e jetstream.KeyValueEntry) { w.Have(e.Key(), e.Value(), opOf(e)) })
-	if err != nil {
-		return err
-	}
-	wantSeen, err := settle(ctx, kvs.wants, func(e jetstream.KeyValueEntry) { w.Want(e.Key(), e.Value(), opOf(e), e.Created()) })
-	if err != nil {
-		return err
+	var haveSeen, wantSeen jetstream.KeyWatcher
+	if kvs.haves != nil {
+		if haveSeen, err = settle(ctx, kvs.haves, func(e jetstream.KeyValueEntry) { w.Have(e.Key(), e.Value(), opOf(e)) }); err != nil {
+			process.NotWatched(logger.Printf, delivery.HaveBucket, err, "moves are not watched until the watchman starts again")
+		} else if wantSeen, err = settle(ctx, kvs.wants, func(e jetstream.KeyValueEntry) { w.Want(e.Key(), e.Value(), opOf(e), e.Created()) }); err != nil {
+			process.NotWatched(logger.Printf, delivery.AssignerWantsBucket, err, "moves are not watched until the watchman starts again")
+			haveSeen.Stop()
+			haveSeen = nil
+		}
 	}
 	// The build seats' steps: a step key `<node>.<ask id>` written is the ask taken (issue 1887 A8b). A seat whose build module
 	// keeps no steps bucket (yet) is said and left; the asks of its stream are mirrored all the same.
@@ -304,7 +308,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		lang := lang
 		sw, err := settle(ctx, kv, func(e jetstream.KeyValueEntry) { stepTaken(w, lang, e) })
 		if err != nil {
-			return err
+			logger.Printf("the steps of %s cannot be watched (%v): its asks are mirrored and never seen taken", module, err)
+			continue
 		}
 		taken[lang] = sw
 	}
@@ -348,12 +353,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	go func() {
 		errs <- follow(ctx, pulses, func(e jetstream.KeyValueEntry) { w.Pulse(e.Key(), opOf(e), e.Created()) })
 	}()
-	go func() {
-		errs <- follow(ctx, haveSeen, func(e jetstream.KeyValueEntry) { w.Have(e.Key(), e.Value(), opOf(e)) })
-	}()
-	go func() {
-		errs <- follow(ctx, wantSeen, func(e jetstream.KeyValueEntry) { w.Want(e.Key(), e.Value(), opOf(e), e.Created()) })
-	}()
+	if haveSeen != nil && wantSeen != nil {
+		go func() {
+			errs <- follow(ctx, haveSeen, func(e jetstream.KeyValueEntry) { w.Have(e.Key(), e.Value(), opOf(e)) })
+		}()
+		go func() {
+			errs <- follow(ctx, wantSeen, func(e jetstream.KeyValueEntry) { w.Want(e.Key(), e.Value(), opOf(e), e.Created()) })
+		}()
+	}
 	for lang, sw := range taken {
 		lang, sw := lang, sw
 		go func() { errs <- follow(ctx, sw, func(e jetstream.KeyValueEntry) { stepTaken(w, lang, e) }) }()

modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go

@@ -2,11 +2,17 @@ package main
 
 import (
 	"context"
+	"errors"
+	"fmt"
 	"io"
 	"os"
 	"path/filepath"
 	"strings"
 	"testing"
+
+	"git.novox.be/novox/mesh-sdk/go/process"
+
+	"github.com/nats-io/nats.go/jetstream"
 )
 
 func TestCredentialIsTheBrokerFile(t *testing.T) { // 1135/AC6
@@ -27,8 +33,23 @@ func TestCredentialIsTheBrokerFile(t *testing.T) { // 1135/AC6
 }
 
 func TestRunNeedsItsCredential(t *testing.T) { // 1135/AC6
+	t.Run("1945: a refused bucket is said, or stops the start naming its grant", func(t *testing.T) {
+		refusal := errors.New("nats: permissions violation for publish to $JS.API.STREAM.INFO.KV_x")
+		var said []string
+		if kv := process.OptionalBucket(context.Background(), refusingBus{refusal}, "x", "x is not heard", func(f string, a ...any) { said = append(said, fmt.Sprintf(f, a...)) }); kv != nil || len(said) != 1 || !strings.Contains(said[0], "permissions violation") {
+			t.Errorf("an optional bucket refused: %v, said %v", kv, said)
+		}
+		if _, err := process.EssentialBucket(context.Background(), refusingBus{refusal}, "watchman", "x", "work"); err == nil || !strings.Contains(err.Error(), "grant of x") {
+			t.Errorf("an essential bucket refused: %v, want its grant named", err)
+		}
+	})
 	err := run(context.Background(), func(string) string { return "" }, io.Discard)
 	if err == nil || !strings.Contains(err.Error(), "MESH_BROKER_FILE") {
 		t.Fatalf("run = %v, want a refusal naming MESH_BROKER_FILE", err)
 	}
 }
+
+// refusingBus refuses every bucket with err (issue 1945).
+type refusingBus struct{ err error }
+
+func (r refusingBus) KeyValue(context.Context, string) (jetstream.KeyValue, error) { return nil, r.err }

modules/mesh-delivery-watchman/go.mod

@@ -3,7 +3,7 @@ module mesh-delivery-watchman
 go 1.26.0
 
 require (
-	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024210118-748c564cdf7c
+	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585
 	github.com/nats-io/nats-server/v2 v2.11.17
 	github.com/nats-io/nats.go v1.54.0
 )

modules/mesh-delivery-watchman/go.sum

@@ -1,5 +1,7 @@
 git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024210118-748c564cdf7c h1:pY2YpHSSsfDPBrOEiWTNSIPn0zdnAayRmtmmyQ3hCHY=
 git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024210118-748c564cdf7c/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585 h1:7QoFH5qjXRpx5M4uPdYsVJ5qKl2082gq8n0lFxhoNIg=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
 github.com/antithesishq/antithesis-sdk-go v0.8.0-default-no-op h1:1BOWQJweNyvZMlpAHXGLiZQn9S+QXGcz3xh94lC0w6E=
 github.com/antithesishq/antithesis-sdk-go v0.8.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
 github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=

modules/mesh-grants/cmd/mesh-grants/tools.go

@@ -76,7 +76,9 @@ func toolsOf(p *grants.Part) []stdio.Tool {
 		{Name: grants.ToolUserList, Description: "The bus's user list in shadow (ADR 0547 clauses 6 and 8): the list the grants part " +
 			"composed (digest, revision, users), what the bus node says it loaded (source, revision, digest, users, error), and how " +
 			"they differ: users added, users dropped (and of those, gone from placement), each user's grant differences, the " +
-			"assignments left out, and every input the bus does not carry, named. Read-only.",
+			"assignments left out, every input the bus does not carry, named; and per user whether the vault holds its hash " +
+			"(mesh-vault.bus-accounts, ADR 0547 clause 9) and how its line would differ from the loaded one (same, adds, left-out " +
+			"for want of a credential, loses-login), never a hash. Read-only.",
 			Run: func(map[string]any) (any, error) { return p.UserList() }},
 		{Name: grants.ToolComposeNow, Description: "Compose again now (ADR 0547 clause 6), as the part does on every change it hears: asks " +
 			"the part's process (which alone holds the grant to publish a membership) to compose the memberships, the user list and " +

modules/mesh-grants/grants_test.go

@@ -385,6 +385,57 @@ func TestGrants(t *testing.T) {
 		}
 	})
 
+	t.Run("credentials", func(t *testing.T) {
+		shadow := []grants.User{{User: "a.held"}, {User: "a.new"}, {User: "a.bare"}}
+		loaded := []grants.User{{User: "a.held"}, {User: "a.bare"}, {User: "enrol.x"}, {User: "person.op"}}
+		hashes := map[string]grants.HeldHash{"a.held": {Hash: "$2a$10$h1", Generation: 2}, "a.new": {Hash: "$2a$10$h2", Generation: 1},
+			"person.op": {Hash: "$2a$10$h3", Adopted: true}}
+		c := grants.CredentialsOf(shadow, loaded, hashes)
+		for _, row := range []struct {
+			user, would   string
+			held, adopted bool
+		}{
+			{"a.bare", grants.WouldLeaveOut, false, false},
+			{"a.held", grants.WouldSame, true, false},
+			{"a.new", grants.WouldAdd, true, false},
+			{"enrol.x", grants.WouldLoseLogin, false, false},
+			{"person.op", grants.WouldNotComposed, true, true},
+		} {
+			i := slices.IndexFunc(c.Users, func(u grants.Credential) bool { return u.User == row.user })
+			if i < 0 || c.Users[i].Would != row.would || c.Users[i].HashHeld != row.held || c.Users[i].Adopted != row.adopted {
+				t.Errorf("%s: %+v, want %s held %v", row.user, c.Users, row.would, row.held)
+			}
+		}
+		if c.HashesHeld != 3 || !slices.Equal(c.LeftOut, []string{"a.bare"}) || !slices.Equal(c.LosesLogin, []string{"enrol.x"}) ||
+			!slices.Equal(c.Adds, []string{"a.new"}) {
+			t.Errorf("credentials = %+v", c)
+		}
+		// Composed from the vault's state: a hash changes the list's state (written on the event), and no hash is ever
+		// in what the part writes or says.
+		b := fixture()
+		b.loaded = shadowOf(t, b)
+		p := partOn(b)
+		before := p.Compose()
+		b.put(grants.ReadAccounts, "a.mod-x", map[string]any{"user": "a.mod-x", "hash": "$2a$10$secrethashbytes", "generation": 1})
+		after := p.Compose()
+		cr := after.UserList.Credentials
+		if !after.Written || after.Revision != before.Revision+1 || cr.HashesHeld != 1 || slices.Contains(cr.LeftOut, "a.mod-x") ||
+			!slices.Contains(cr.LeftOut, "b.mod-y") {
+			t.Errorf("a hash heard = %+v", after.UserList.Credentials)
+		}
+		err := p.TakeOver("person.operator", "hashes")
+		if err == nil || !strings.Contains(err.Error(), "no hash for") || strings.Contains(err.Error(), "a.mod-x,") {
+			t.Errorf("take-over = %v, want the users with no hash named", err)
+		}
+		for state, keys := range b.own {
+			for k, v := range keys {
+				if strings.Contains(string(v), "secrethashbytes") {
+					t.Errorf("the hash is written in %s %s", state, k)
+				}
+			}
+		}
+	})
+
 	t.Run("compare", func(t *testing.T) {
 		b := fixture()
 		shadow := shadowOf(t, b)

modules/mesh-grants/internal/grants/follow.go

@@ -16,8 +16,8 @@ type Watcher interface {
 
 // Followed are the states whose change composes again: placement, the indexer's manifests by module, the resolved
 // config values, the node agents' haves (a machine's first have, and the bus node's loaded list) and the assigner's
-// wants (the manifest each machine is wanted to run).
-var Followed = []string{ReadAssignments, ReadModules, ReadResolved, ReadHave, ReadWants}
+// wants (the manifest each machine is wanted to run), and the vault's hashes of the bus accounts.
+var Followed = []string{ReadAssignments, ReadModules, ReadResolved, ReadHave, ReadWants, ReadAccounts}
 
 // Follower composes again on every change it hears, never on a timer. It never stops for a refusal: a watch it cannot
 // open is said, composed without, and opened again at the next wake (any change it does hear, or a reconnect).

modules/mesh-grants/internal/grants/inputs.go

@@ -28,10 +28,13 @@ const (
 	ReadResolved        = "mesh-config.resolved"
 	ReadWants           = "mesh-delivery-assigner.wants"
 	ReadBuilds          = "mesh-delivery-builder.builds"
+	// ReadAccounts is the vault's bcrypt hashes of the bus accounts it holds, by bus user (ADR 0547 clause 9).
+	ReadAccounts = "mesh-vault.bus-accounts"
 )
 
 // Reads is every state the part reads, in the manifest's order.
-var Reads = []string{ReadAssignments, ReadControllerGrant, ReadModules, ReadManifests, ReadResolved, ReadHave, ReadWants, ReadBuilds}
+var Reads = []string{ReadAssignments, ReadControllerGrant, ReadModules, ReadManifests, ReadResolved, ReadHave, ReadWants, ReadBuilds,
+	ReadAccounts}
 
 // The bus module's tools the part calls, as its manifest's invokes name them: read-only, each.
 const (
@@ -67,7 +70,6 @@ var standingGaps = []Gap{
 	{Input: "people", Why: "the people and the tools they may invoke are in the controller's store alone: no person user is composed"},
 	{Input: "seat holdings", Why: "who holds each seat held once for the mesh is in the controller's store alone: a mesh seat claimed on two machines is composed as held on both"},
 	{Input: "announced states", Why: "the controller does not announce its states-announced readable, so every read is composed from `reads`, as when announcements cannot be read"},
-	{Input: "credentials", Why: "the shadow holds no hash, so a user it composes that the loaded list lacks may be one left out for want of a credential"},
 }
 
 // assignment is a key of the controller's assignments as composing needs it (mesh-controller internal/chain Assignment):
@@ -108,7 +110,32 @@ type Inputs struct {
 	// BusNode is the machine running the bus's module, and Loaded what its node agent says it loaded.
 	BusNode string
 	Loaded  *delivery.UserListWritten
-	Gaps    []Gap
+	// Hashes are the vault's hashes of the bus accounts, by bus user (ReadAccounts); kept in memory, never written.
+	Hashes map[string]HeldHash
+	Gaps   []Gap
+}
+
+// HeldHash is one key of the vault's bus-accounts: the hash, and what it is of. The hash is never said or written by
+// the part in shadow.
+type HeldHash struct {
+	Hash       string    `json:"hash"`
+	Generation int       `json:"generation,omitempty"`
+	At         time.Time `json:"at"`
+	Adopted    bool      `json:"adopted,omitempty"`
+}
+
+// readHashes reads the vault's bus-accounts; a refusal is a gap, and every user then has no hash held.
+func readHashes(r Reader, gaps *[]Gap) map[string]HeldHash {
+	out := map[string]HeldHash{}
+	for user, raw := range each(r, ReadAccounts, gaps) {
+		var h HeldHash
+		if json.Unmarshal(raw, &h) != nil || h.Hash == "" {
+			*gaps = append(*gaps, Gap{Input: ReadAccounts + " " + user, Why: "holds no hash"})
+			continue
+		}
+		out[user] = h
+	}
+	return out
 }
 
 // BusModule is the bus's module: it holds mesh-broker.
@@ -301,6 +328,7 @@ func ReadInputs(r Reader, cache *Manifests) Inputs {
 		sort.Strings(fromTrunk)
 		in.Gaps = append(in.Gaps, Gap{Input: "wanted manifests", Why: "no build record names the manifest at the wanted fingerprint, so the trunk's is used: " + strings.Join(fromTrunk, ", ")})
 	}
+	in.Hashes = readHashes(r, &in.Gaps)
 	for _, raw := range each(r, ReadResolved, &in.Gaps) {
 		var v resolved
 		if json.Unmarshal(raw, &v) != nil || v.Node == "" || v.Module == "" || v.Key == "" || len(v.Value) == 0 {

modules/mesh-grants/internal/grants/part.go

@@ -6,6 +6,7 @@ import (
 	"encoding/hex"
 	"encoding/json"
 	"fmt"
+	"slices"
 	"sort"
 	"strings"
 	"sync"
@@ -66,9 +67,11 @@ type UserListState struct {
 	LeftOut []Gap                     `json:"left_out"`
 	Compare Comparison                `json:"compare"`
 	Loaded  *delivery.UserListWritten `json:"loaded,omitempty"`
-	BusNode string                    `json:"bus_node,omitempty"`
-	Gaps    []Gap                     `json:"gaps"`
-	At      time.Time                 `json:"at"`
+	// Credentials says per user whether the vault holds its hash and how its line would differ (ADR 0547 clause 9).
+	Credentials Credentials `json:"credentials"`
+	BusNode     string      `json:"bus_node,omitempty"`
+	Gaps        []Gap       `json:"gaps"`
+	At          time.Time   `json:"at"`
 }
 
 func (s UserListState) inputs() string {
@@ -141,12 +144,14 @@ func (p *Part) Compose() Result {
 	if err != nil {
 		res.Errors = append(res.Errors, "the shadow list could not be composed: "+err.Error())
 	}
+	users = WithHashes(users, in.Hashes)
 	st.Digest, st.Users = Digest(users), len(users)
 	loaded, err := LoadedUsers(p.Call)
 	st.Compare = Compare(users, loaded, Placed(in))
 	if err != nil {
 		st.Compare.Unread = err.Error()
 	}
+	st.Credentials = CredentialsOf(users, loaded, in.Hashes)
 	st.Inputs = st.inputs()
 	res.UserList = st
 	if raw, rev, found, err := p.Store.Get(StateUserList, KeyList); err != nil {
@@ -324,6 +329,18 @@ func (p *Part) TakeOver(caller, why string) error {
 			reasons = append(reasons, fmt.Sprintf("%d users differ between the shadow list and the loaded list: %s", len(d), strings.Join(d, ", ")))
 		}
 	}
+	cr, gone := res.UserList.Credentials, res.UserList.Compare.Gone
+	var unhashed []string
+	for _, u := range append(append([]string(nil), cr.LeftOut...), cr.LosesLogin...) {
+		if !slices.Contains(gone, u) {
+			unhashed = append(unhashed, u)
+		}
+	}
+	if len(unhashed) > 0 {
+		sort.Strings(unhashed)
+		reasons = append(reasons, fmt.Sprintf("the vault holds no hash for %d users (ADR 0547 clause 9: handed or adopted before take-over): %s",
+			len(unhashed), strings.Join(unhashed, ", ")))
+	}
 	reasons = append(reasons, NotComposable)
 	a.Refused = "take-over refused: " + strings.Join(reasons, "; ")
 	return fmt.Errorf("%s%s", a.Refused, p.record(a))

modules/mesh-grants/internal/grants/userlist.go

@@ -18,6 +18,96 @@ type User struct {
 	Publish        []string `json:"publish"`
 	Subscribe      []string `json:"subscribe"`
 	AllowResponses bool     `json:"allow_responses"`
+	// Hash is the bcrypt hash the vault holds for the user (WithHashes): in memory alone, never in a digest, a compare
+	// or the part's state, and empty for a loaded user.
+	Hash string `json:"-"`
+}
+
+// WithHashes gives each user of the shadow list the hash the vault holds for it, when it holds one.
+func WithHashes(users []User, hashes map[string]HeldHash) []User {
+	for i := range users {
+		users[i].Hash = hashes[users[i].User].Hash
+	}
+	return users
+}
+
+// Credential is one user's hash as the list would carry it: whether the vault holds one, and how the user stands in the
+// shadow list and the loaded list. Never the hash: a bcrypt hash of the same password differs by its salt, so two hashes
+// say nothing compared.
+type Credential struct {
+	User     string `json:"user"`
+	InShadow bool   `json:"in_shadow"`
+	Loaded   bool   `json:"loaded"`
+	HashHeld bool   `json:"hash_held"`
+	Adopted  bool   `json:"adopted,omitempty"`
+	// Generation is the vault's generation of the account the hash is of; zero for an adopted one.
+	Generation int `json:"generation,omitempty"`
+	// Would is how the list with the vault's hashes would differ from the loaded one for this user: same (a hash held,
+	// loaded), adds (a hash held, not loaded), left-out (in the shadow list, no hash: left out for want of a
+	// credential), loses-login (loaded, no hash held: its login would be lost) or not-composed (a hash held, loaded,
+	// and not in the shadow list).
+	Would string `json:"would"`
+}
+
+// The ways a user's line would differ.
+const (
+	WouldSame        = "same"
+	WouldAdd         = "adds"
+	WouldLeaveOut    = "left-out"
+	WouldLoseLogin   = "loses-login"
+	WouldNotComposed = "not-composed"
+)
+
+// Credentials is every user's Credential, and the users of each way that differs.
+type Credentials struct {
+	Users      []Credential `json:"users"`
+	HashesHeld int          `json:"hashes_held"`
+	LeftOut    []string     `json:"left_out"`
+	LosesLogin []string     `json:"loses_login"`
+	Adds       []string     `json:"adds"`
+}
+
+// CredentialsOf says, of every user of the shadow and the loaded list, whether the vault holds its hash and how its line
+// would differ. Pure.
+func CredentialsOf(shadow, loaded []User, hashes map[string]HeldHash) Credentials {
+	c := Credentials{Users: []Credential{}, LeftOut: []string{}, LosesLogin: []string{}, Adds: []string{}}
+	rows := map[string]*Credential{}
+	row := func(user string) *Credential {
+		if rows[user] == nil {
+			h, held := hashes[user]
+			rows[user] = &Credential{User: user, HashHeld: held, Adopted: h.Adopted, Generation: h.Generation}
+		}
+		return rows[user]
+	}
+	for _, u := range shadow {
+		row(u.User).InShadow = true
+	}
+	for _, u := range loaded {
+		row(u.User).Loaded = true
+	}
+	for _, user := range sortedKeys(rows) {
+		r := rows[user]
+		switch {
+		case r.HashHeld && r.InShadow && r.Loaded:
+			r.Would = WouldSame
+		case r.HashHeld && r.InShadow:
+			r.Would = WouldAdd
+			c.Adds = append(c.Adds, user)
+		case r.InShadow:
+			r.Would = WouldLeaveOut
+			c.LeftOut = append(c.LeftOut, user)
+		case !r.HashHeld:
+			r.Would = WouldLoseLogin
+			c.LosesLogin = append(c.LosesLogin, user)
+		default:
+			r.Would = WouldNotComposed
+		}
+		if r.HashHeld {
+			c.HashesHeld++
+		}
+		c.Users = append(c.Users, *r)
+	}
+	return c
 }
 
 // ComposeUsers is the shadow list: every user the composer derives from the records, with its grants. Pure.

modules/mesh-grants/module.json

@@ -10,7 +10,8 @@
     "mesh-delivery-builder.builds",
     "mesh-delivery-indexer.manifests",
     "mesh-delivery-indexer.modules",
-    "node-engine.have"
+    "node-engine.have",
+    "mesh-vault.bus-accounts"
   ],
   "invokes": [
     "nats.nats_streams",