Merge branch 'feat/r4g-view' into feat/r4g

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

@@ -602,3 +602,97 @@ func TestProcessWholeWritesTheAnswerToAnAskNeedingNoProof(t *testing.T) { // 129
 	}
 	wait(func(s view.Snapshot) bool { return len(s.Asks) == 1 && s.Asks[0].Key == mkey }, "the answered ask never left the list")
 }
+
+// TestProcessWholeShowsTheBacklogOfBuildAsks starts the program whole against a bus holding the go seat's work-queue stream
+// with its three shared workers, three check asks waiting and one held by a node: the socket's backlog row names three
+// waiting, one held, and the oldest's age (1887/A8a AC; compiled in the unit run, run by the merge check).
+func TestProcessWholeShowsTheBacklogOfBuildAsks(t *testing.T) { // 1887/A8a AC
+	srv, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: -1, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	go srv.Start()
+	if !srv.ReadyForConnections(10 * time.Second) {
+		t.Fatal("the bus did not come up")
+	}
+	t.Cleanup(srv.Shutdown)
+	nc, err := nats.Connect(srv.ClientURL())
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(nc.Close)
+	js, err := jetstream.New(nc)
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
+	defer cancel()
+	stream, _ := delivery.AskStream(delivery.AskGo)
+	st, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: stream, Subjects: []string{"mesh.seat.node-build-go.accept.>"}, Retention: jetstream.WorkQueuePolicy})
+	if err != nil {
+		t.Fatal(err)
+	}
+	var checkWorker jetstream.Consumer
+	for _, kind := range delivery.AskKinds {
+		name, _ := delivery.AskWorker(delivery.AskGo, kind)
+		subject, _ := delivery.AskSubject(delivery.AskGo, kind)
+		c, err := st.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{Durable: name, FilterSubject: subject, AckPolicy: jetstream.AckExplicitPolicy})
+		if err != nil {
+			t.Fatal(err)
+		}
+		if kind == delivery.AskCheck {
+			checkWorker = c
+		}
+	}
+	subject, _ := delivery.AskSubject(delivery.AskGo, delivery.AskCheck)
+	oldest := time.Now().UTC().Add(-2 * time.Minute)
+	for i, id := range []string{"m0000001aaaaaa", "m0000002bbbbbb", "m0000003cccccc", "m0000004dddddd"} {
+		body, err := delivery.EncodeStepAsk(delivery.StepAsk{ID: id, Kind: delivery.AskCheck, Language: delivery.AskGo, Asker: "mesh-delivery-checker", At: oldest.Add(time.Duration(i) * time.Second),
+			Check: &delivery.CheckJob{Fetch: map[string]any{"repository": "novox/x", "commit": "0123456789abcdef0123456789abcdef01234567"}, Levels: []string{delivery.LevelUnit}}})
+		if err != nil {
+			t.Fatal(err)
+		}
+		if _, err := js.Publish(ctx, subject, body); err != nil {
+			t.Fatal(err)
+		}
+	}
+	// A node pulls one and holds it unacknowledged.
+	held, err := checkWorker.Fetch(1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for range held.Messages() {
+	}
+
+	dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "i")
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { os.RemoveAll(dir) })
+	broker := filepath.Join(dir, "broker")
+	cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"})
+	if err := os.WriteFile(broker, cred, 0o600); err != nil {
+		t.Fatal(err)
+	}
+	socket := filepath.Join(dir, "run", "cli.sock")
+	env := map[string]string{"MESH_BROKER_FILE": broker, "MESH_DELIVERY_SOCKET": socket, "MESH_DELIVERY_WEB": "127.0.0.1:0"}
+	runCtx, stop := context.WithCancel(ctx)
+	done := make(chan error, 1)
+	go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }()
+	t.Cleanup(func() { stop(); <-done })
+
+	var last view.Snapshot
+	for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(100 * time.Millisecond) {
+		s, err := view.Ask(socket)
+		if err != nil {
+			continue
+		}
+		last = s
+		for _, b := range s.Backlog {
+			if b.Language == "go" && b.Kind == "check" && b.Waiting == 3 && b.Held == 1 && b.OldestSecs >= 118 {
+				return
+			}
+		}
+	}
+	t.Fatalf("the backlog never showed three waiting asks with the oldest's age; last: %+v", last.Backlog)
+}

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

@@ -138,6 +138,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		return err
 	}
 	model.SetWriter(kvWriter{ctx: ctx, js: js})
+	model.SetBacklog(view.BacklogOver(js, 5*time.Second))
 	for _, bucket := range []string{delivery.HaveBucket, view.AssignmentsBucket, view.BuildsBucket, delivery.PresenceBucket, delivery.StuckBucket, delivery.CandidatesBucket, delivery.ModulesBucket, delivery.AsksBucket, delivery.AnswersBucket, delivery.BuilderBuildsBucket, delivery.AssignerDifferencesBucket, delivery.AssignerAsksBucket, delivery.ViewAnswersBucket} {
 		go follow(ctx, js, bucket, model, logger)
 	}

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

@@ -0,0 +1,116 @@
+package view
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/nats-io/nats.go/jetstream"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+// The backlog of the build seats (novox/hq ADR 0525 §4, issue 1887 A8a): for each language and kind of ask, how many asks wait
+// on the seat's stream for a node to pull them, how many a node holds without having finished them, and how old the oldest
+// waiting ask is. It is read from the stream's shared pull worker (delivery.AskWorker) when the view is asked: NumPending
+// waits, NumAckPending is held by a node, and the oldest waiting ask is the stream's message at the worker's next sequence.
+
+// BacklogRow is one language's kind of ask: Waiting asks not yet pulled, Held asks pulled by a node and not yet ended, and the
+// oldest waiting ask's time and age in seconds. Error says why the row could not be read, never dropping it silently.
+type BacklogRow struct {
+	Language   string    `json:"language"`
+	Kind       string    `json:"kind"`
+	Waiting    uint64    `json:"waiting"`
+	Held       uint64    `json:"held"`
+	Oldest     time.Time `json:"oldest,omitzero"`
+	OldestSecs int64     `json:"oldest-secs,omitempty"`
+	Error      string    `json:"error,omitempty"`
+}
+
+// Backlog reads the seats' workers. Info is the worker's counts and the stream sequence it delivers next; Message is the body
+// of the first message of the stream at or after seq on the subject.
+type Backlog struct {
+	Info    func(stream, worker string) (waiting, held, next uint64, err error)
+	Message func(stream, subject string, seq uint64) ([]byte, error)
+	Now     func() time.Time
+}
+
+// Rows are the rows of every language and kind that has an ask waiting or held, or that could not be read.
+func (b *Backlog) Rows() []BacklogRow {
+	now := time.Now
+	if b.Now != nil {
+		now = b.Now
+	}
+	out := []BacklogRow{}
+	for _, lang := range delivery.AskLanguages {
+		stream, _ := delivery.AskStream(lang)
+		for _, kind := range delivery.AskKinds {
+			worker, _ := delivery.AskWorker(lang, kind)
+			subject, _ := delivery.AskSubject(lang, kind)
+			row := BacklogRow{Language: lang, Kind: kind}
+			waiting, held, next, err := b.Info(stream, worker)
+			if err != nil {
+				row.Error = err.Error()
+				out = append(out, row)
+				continue
+			}
+			row.Waiting, row.Held = waiting, held
+			if waiting > 0 {
+				body, err := b.Message(stream, subject, next)
+				if err == nil {
+					var a delivery.StepAsk
+					a, err = delivery.DecodeStepAsk(body)
+					if err == nil {
+						row.Oldest, row.OldestSecs = a.At, int64(now().Sub(a.At).Seconds())
+					}
+				}
+				if err != nil {
+					row.Error = fmt.Sprintf("the oldest waiting ask cannot be read: %v", err)
+				}
+			}
+			if row.Waiting > 0 || row.Held > 0 || row.Error != "" {
+				out = append(out, row)
+			}
+		}
+	}
+	return out
+}
+
+// BacklogOver reads the backlog from the bus: the worker's consumer info, and the stream's message by sequence.
+func BacklogOver(js jetstream.JetStream, timeout time.Duration) *Backlog {
+	ctx := func() (context.Context, context.CancelFunc) {
+		return context.WithTimeout(context.Background(), timeout)
+	}
+	return &Backlog{
+		Info: func(stream, worker string) (uint64, uint64, uint64, error) {
+			c, cancel := ctx()
+			defer cancel()
+			s, err := js.Stream(c, stream)
+			if err != nil {
+				return 0, 0, 0, fmt.Errorf("the stream %s: %w", stream, err)
+			}
+			cons, err := s.Consumer(c, worker)
+			if err != nil {
+				return 0, 0, 0, fmt.Errorf("the worker %s: %w", worker, err)
+			}
+			info := cons.CachedInfo()
+			if i, err := cons.Info(c); err == nil {
+				info = i
+			}
+			return uint64(info.NumPending), uint64(info.NumAckPending), info.Delivered.Stream + 1, nil
+		},
+		Message: func(stream, subject string, seq uint64) ([]byte, error) {
+			c, cancel := ctx()
+			defer cancel()
+			s, err := js.Stream(c, stream)
+			if err != nil {
+				return nil, err
+			}
+			m, err := s.GetMsg(c, seq, jetstream.WithGetMsgSubject(subject))
+			if err != nil {
+				return nil, err
+			}
+			return m.Data, nil
+		},
+	}
+}

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

@@ -0,0 +1,87 @@
+package view
+
+import (
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+func stepAsk(t *testing.T, id, kind string, at time.Time) []byte {
+	t.Helper()
+	b, err := delivery.EncodeStepAsk(delivery.StepAsk{ID: id, Kind: kind, Language: delivery.AskGo, Asker: "mesh-delivery-checker", At: at,
+		Check: &delivery.CheckJob{Fetch: map[string]any{"repository": "novox/x", "commit": "0123456789abcdef0123456789abcdef01234567"}, Levels: []string{delivery.LevelUnit}}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	return b
+}
+
+// fakeSeats is the check worker of the go seat with three asks waiting and one held, the oldest at t0.
+func fakeSeats(t *testing.T, now time.Time) *Backlog {
+	worker, _ := delivery.AskWorker(delivery.AskGo, delivery.AskCheck)
+	subject, _ := delivery.AskSubject(delivery.AskGo, delivery.AskCheck)
+	oldest := stepAsk(t, "m0000001aaaaaa", delivery.AskCheck, now.Add(-90*time.Second))
+	return &Backlog{Now: func() time.Time { return now },
+		Info: func(stream, w string) (uint64, uint64, uint64, error) {
+			switch w {
+			case worker:
+				return 3, 1, 42, nil
+			case "SEAT_NODE_BUILD_IMAGE_DELIVERY_worker":
+				return 0, 0, 1, fmt.Errorf("the worker is not there")
+			}
+			return 0, 0, 1, nil
+		},
+		Message: func(stream, s string, seq uint64) ([]byte, error) {
+			if s != subject || seq != 42 {
+				return nil, fmt.Errorf("asked %s at %d", s, seq)
+			}
+			return oldest, nil
+		}}
+}
+
+func TestThreeWaitingAsksAreShownWithTheOldestsAge(t *testing.T) { // 1887/A8a AC
+	now := time.Date(2026, 10, 23, 12, 0, 0, 0, time.UTC)
+	m := New()
+	m.SetBacklog(fakeSeats(t, now))
+	rows := m.Snapshot().Backlog
+	var check, image *BacklogRow
+	for i := range rows {
+		switch {
+		case rows[i].Language == "go" && rows[i].Kind == "check":
+			check = &rows[i]
+		case rows[i].Language == "image" && rows[i].Kind == "delivery":
+			image = &rows[i]
+		default:
+			t.Errorf("a quiet seat is listed: %+v", rows[i])
+		}
+	}
+	if check == nil || check.Waiting != 3 || check.Held != 1 || check.OldestSecs != 90 || !check.Oldest.Equal(now.Add(-90*time.Second)) || check.Error != "" {
+		t.Fatalf("check row = %+v", check)
+	}
+	if image == nil || image.Error == "" {
+		t.Errorf("a worker that cannot be read is dropped: %+v", rows)
+	}
+	var out strings.Builder
+	Table(&out, m.Snapshot())
+	for _, w := range []string{"BACKLOG OF BUILD ASKS", "go", "check", "3", "1m30s", "unreadable"} {
+		if !strings.Contains(out.String(), w) {
+			t.Errorf("table lacks %q:\n%s", w, out.String())
+		}
+	}
+	rec := httptest.NewRecorder()
+	Web(m).ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/", nil))
+	if !strings.Contains(rec.Body.String(), "BACKLOG OF BUILD ASKS") || !strings.Contains(rec.Body.String(), "1m30s") {
+		t.Errorf("page lacks the backlog:\n%s", rec.Body.String())
+	}
+}
+
+func TestNoBacklogReaderGivesNoRows(t *testing.T) {
+	if got := New().Snapshot().Backlog; got == nil || len(got) != 0 {
+		t.Errorf("backlog = %#v", got)
+	}
+}

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

@@ -236,6 +236,24 @@ func differenceText(s Snapshot) [][]string {
 	return out
 }
 
+// backlogText is one line per language and kind: its waiting asks, the asks held by a node, the oldest waiting ask's age.
+func backlogText(s Snapshot) [][]string {
+	var out [][]string
+	for _, b := range s.Backlog {
+		age := ""
+		if !b.Oldest.IsZero() {
+			age = (time.Duration(b.OldestSecs) * time.Second).String() + " (since " + since(b.Oldest) + ")"
+		}
+		if b.Error != "" {
+			age = "unreadable: " + b.Error
+		}
+		out = append(out, []string{b.Language, b.Kind, fmt.Sprint(b.Waiting), fmt.Sprint(b.Held), age})
+	}
+	return out
+}
+
+var backlogHead = []string{"LANGUAGE", "KIND", "WAITING", "HELD", "OLDEST WAITING"}
+
 var differenceHead = []string{"NODE", "MODULE", "WANT", "SENT", "RULE", "WORDS"}
 
 var head = []string{"NODE", "MODULE", "ASSIGNED", "HAVE", "PHASE", "HEALTH", "SINCE", "DETAIL", "WAITS"}
@@ -277,6 +295,14 @@ func Table(w io.Writer, s Snapshot) {
 			row(cells)
 		}
 	}
+	if b := backlogText(s); len(b) > 0 {
+		fmt.Fprintln(tw)
+		row([]string{"BACKLOG OF BUILD ASKS"})
+		row(backlogHead)
+		for _, cells := range b {
+			row(cells)
+		}
+	}
 	if d := differenceText(s); len(d) > 0 {
 		fmt.Fprintln(tw)
 		row([]string{"WANTS AGAINST SENDS"})
@@ -329,6 +355,7 @@ var page = template.Must(template.New("p").Parse(`<!doctype html><html><head><me
 {{range .Rows}}<tr>{{range .}}<td>{{.}}</td>{{end}}</tr>{{end}}</table>
 {{if .Groups}}<h2>CANDIDATES</h2><table><tr><th>GROUP</th><th>PULL REQUEST</th><th>HEAD</th><th>MODULES</th></tr>{{range .Groups}}<tr>{{range .}}<td>{{.}}</td>{{end}}</tr>{{end}}</table>{{end}}
 {{if .Asks}}<h2>ASKS</h2><table><tr><th>ASK</th><th>HEADLINE</th><th>WORDS</th><th>PULL REQUESTS</th><th>ANSWER</th></tr>{{range .Asks}}<tr>{{range .Cells}}<td>{{.}}</td>{{end}}<td>{{if .Options}}<form method="post" action="/answer"><input type="hidden" name="key" value="{{.Key}}"><input name="words" placeholder="words (optional)">{{range .Options}} <button name="option" value="{{.}}">{{.}}</button>{{end}}</form>{{end}}</td></tr>{{end}}</table>{{end}}
+{{if .Backlog}}<h2>BACKLOG OF BUILD ASKS</h2><table><tr>{{range .BacklogHead}}<th>{{.}}</th>{{end}}</tr>{{range .Backlog}}<tr>{{range .}}<td>{{.}}</td>{{end}}</tr>{{end}}</table>{{end}}
 {{if .Differences}}<h2>WANTS AGAINST SENDS</h2><table><tr>{{range .DifferenceHead}}<th>{{.}}</th>{{end}}</tr>{{range .Differences}}<tr>{{range .}}<td>{{.}}</td>{{end}}</tr>{{end}}</table>{{end}}
 {{if .Fingerprints}}<h2>FINGERPRINTS</h2><table><tr><th>MODULE</th><th>STATE</th><th>INDEXER AGAINST BUILD</th></tr>{{range .Fingerprints}}<tr>{{range .}}<td>{{.}}</td>{{end}}</tr>{{end}}</table>{{end}}
 {{if .Offline}}<h2>OFFLINE</h2><table>{{range .Offline}}<tr>{{range .}}<td>{{.}}</td>{{end}}</tr>{{end}}</table>{{end}}
@@ -383,7 +410,7 @@ func Web(m *Model) http.Handler {
 		for _, u := range snap.Unreadable {
 			bad = append(bad, []string{u.Bucket, u.Key, u.Error})
 		}
-		page.Execute(w, map[string]any{"Head": head, "Rows": rows, "Groups": groupText(snap), "Asks": askPage(snap), "Fingerprints": fingerprintText(snap), "Differences": differenceText(snap), "DifferenceHead": differenceHead, "Offline": off, "Stuck": stuck, "Unreadable": bad})
+		page.Execute(w, map[string]any{"Head": head, "Rows": rows, "Groups": groupText(snap), "Asks": askPage(snap), "Fingerprints": fingerprintText(snap), "Differences": differenceText(snap), "Backlog": backlogText(snap), "BacklogHead": backlogHead, "DifferenceHead": differenceHead, "Offline": off, "Stuck": stuck, "Unreadable": bad})
 	})
 }
 

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

@@ -69,6 +69,9 @@ type Snapshot struct {
 	// Differences is every difference between what the assigner wants a node to run and what the controller sends, with the rule that
 	// explains it or unexplained (novox/hq ADR 0528 clause 4, issue 1297 B8).
 	Differences []DifferenceRow `json:"differences"`
+	// Backlog is, for each language and kind of build ask with one waiting or held, the asks waiting for a node, the asks a node
+	// holds and the oldest waiting ask's age (issue 1887 A8a).
+	Backlog []BacklogRow `json:"backlog"`
 }
 
 // DifferenceRow is a module on a node whose want and send differ: the fingerprint wanted (empty when the node has no want,
@@ -186,6 +189,7 @@ type Model struct {
 	aasks   map[string]assignerAskRecord   // the assigner's asks (hold, restore), by key
 	chosen  map[string]delivery.ViewAnswer // the answers the view wrote, by ask key
 	writer  Writer
+	backlog *Backlog
 	answers map[string]string // by key: the router id of the asking the teller's answer is for ("" when unreadable)
 	newest  map[string]string // module -> id
 	haves   map[string]haveEntry
@@ -452,6 +456,25 @@ func errText(err error) string {
 // Snapshot derives one row per module and node with an open assignment, a row marked left behind for each have with
 // none, and a row per node's own key. Rows are ordered by node and module.
 func (m *Model) Snapshot() Snapshot {
+	s := m.snapshot()
+	m.mu.Lock()
+	b := m.backlog
+	m.mu.Unlock()
+	s.Backlog = []BacklogRow{}
+	if b != nil {
+		s.Backlog = b.Rows() // read from the bus outside the model's lock
+	}
+	return s
+}
+
+// SetBacklog gives the model the seats' workers to read the backlog of asks from.
+func (m *Model) SetBacklog(b *Backlog) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.backlog = b
+}
+
+func (m *Model) snapshot() Snapshot {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 	rows := []Row{}

modules/mesh-delivery-view/manifest_test.go

@@ -48,6 +48,7 @@ type manifest struct {
 	Version      string            `json:"version"`
 	Slug         string            `json:"slug"`
 	Reads        []string          `json:"reads"`
+	Uses         []string          `json:"uses"`
 	Tools        []string          `json:"tools"`
 	OwnSecrets   map[string]string `json:"own-secrets"`
 	SecretsOwner string            `json:"secrets-owner"`
@@ -93,6 +94,10 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7
 			t.Errorf("the manifest declares %q: the view writes only its answers", k)
 		}
 	}
+	// The view reads the build seats' streams for the backlog of asks (issue 1887 A8a): the three seats it uses.
+	if want := []string{"node-build-go", "node-build-image", "node-build-node"}; !reflect.DeepEqual(m.Uses, want) {
+		t.Errorf("uses = %v, want %v", m.Uses, want)
+	}
 	// The one state the view writes: the operator's answers to the asks that need no proof (ADR 0528 clause 6, 1297/B9), which the teller reads.
 	if want := []state{{Name: "answers", History: 5}}; !reflect.DeepEqual(m.State, want) {
 		t.Errorf("state = %+v, want %+v", m.State, want)

modules/mesh-delivery-view/module.json

@@ -28,6 +28,11 @@
     "mesh-delivery-watchman.stuck",
     "node-engine.have"
   ],
+  "uses": [
+    "node-build-go",
+    "node-build-image",
+    "node-build-node"
+  ],
   "own-secrets": {
     "broker": "${dir:state}/broker"
   },

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

@@ -17,8 +17,6 @@ import (
 	"github.com/nats-io/nats.go/jetstream"
 
 	"git.novox.be/novox/mesh-sdk/go/delivery"
-
-	"mesh-delivery-watchman/internal/watch"
 )
 
 // busConf is the composed accounts file's shape (mesh-controller internal/broker): the account MESH imports the system
@@ -100,7 +98,7 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8
 		{Bucket: delivery.PresenceBucket, History: 20},
 		{Bucket: delivery.StuckBucket, History: 20},
 		{Bucket: delivery.PulseBucket, LimitMarkerTTL: time.Second},
-		{Bucket: watch.ChecksBucket, History: 5},
+		{Bucket: delivery.StepsBucket("mesh-build-go"), History: 5},
 	} {
 		kv, err := js.CreateKeyValue(ctx, c)
 		if err != nil {
@@ -167,19 +165,26 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8
 	n1.Close()
 	until("node n1 offline", func() bool { on, ok := online("n1"); return ok && !on })
 
-	// A check running gets a pulse, as its builder puts the key (mesh-controller CheckRunning); a builder that stops
-	// heartbeating lets the pulse expire, and that writes a stuck key.
-	running := func() []byte {
-		b, _ := json.Marshal(map[string]any{"head": "abc", "trunk": "def", "builder": "n1", "build": "b9", "started": time.Now().UTC()})
-		return b
+	// An ask put on a build seat gets a pulse; one no node takes lets the pulse expire, and that writes a stuck key.
+	askStream, _ := delivery.AskStream(delivery.AskGo)
+	askSubject, _ := delivery.AskSubject(delivery.AskGo, delivery.AskCheck)
+	putAsk := func(id string) {
+		t.Helper()
+		body, err := delivery.EncodeStepAsk(delivery.StepAsk{ID: id, Kind: delivery.AskCheck, Language: delivery.AskGo, Asker: "mesh-delivery-checker", At: time.Now().UTC(),
+			Check: &delivery.CheckJob{Fetch: map[string]any{"repository": "novox/x", "commit": "0123456789abcdef0123456789abcdef01234567"}, Levels: []string{delivery.LevelUnit}}})
+		if err != nil {
+			t.Fatal(err)
+		}
+		if err := ctl.Publish(askSubject, body); err != nil {
+			t.Fatal(err)
+		}
 	}
-	if _, err := kvs[watch.ChecksBucket].Put(ctx, "o.r.7", running()); err != nil {
-		t.Fatal(err)
-	}
-	until("a pulse for the check running", func() bool { _, ok := has(delivery.PulseBucket, pulseOf(watch.ChecksBucket, "o.r.7")); return ok })
+	const ask1, ask2 = "m0000001aaaaaa", "m0000002bbbbbb"
+	putAsk(ask1)
+	until("a pulse for the ask waiting", func() bool { _, ok := has(delivery.PulseBucket, pulseOf(askStream, ask1)); return ok })
 	var stuck delivery.Stuck
-	until("a stuck key for the check whose builder stopped", func() bool {
-		e, ok := has(delivery.StuckBucket, stuckOf(watch.ChecksBucket, "o.r.7"))
+	until("a stuck key for the ask no node took", func() bool {
+		e, ok := has(delivery.StuckBucket, stuckOf(askStream, ask1))
 		if !ok {
 			return false
 		}
@@ -187,7 +192,7 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8
 		stuck, err = delivery.DecodeStuck(e.Value())
 		return err == nil
 	})
-	if stuck.Moved || stuck.Waited != "the check of o.r.7 to end" || stuck.Bucket != watch.ChecksBucket || stuck.Key != "o.r.7" {
+	if stuck.Moved || stuck.Bucket != askStream || stuck.Key != ask1 {
 		t.Fatalf("stuck = %+v", stuck)
 	}
 	// The node n1, gone offline and never back, is stuck as well.
@@ -204,17 +209,15 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8
 	connect("node.n2")
 	until("node n2 online again", func() bool { on, ok := online("n2"); return ok && on })
 	until("the pulse of node n2 deleted", func() bool { _, ok := has(delivery.PulseBucket, pulseOf(delivery.PresenceBucket, "n2")); return !ok })
-	// A check that ends at once is the same: no stuck key, however long the bound passes.
-	if _, err := kvs[watch.ChecksBucket].Put(ctx, "o.r.8", running()); err != nil {
+	// An ask a node takes at once is the same: no stuck key, however long the bound passes.
+	putAsk(ask2)
+	until("a pulse for the second ask", func() bool { _, ok := has(delivery.PulseBucket, pulseOf(askStream, ask2)); return ok })
+	if _, err := kvs[delivery.StepsBucket("mesh-build-go")].Put(ctx, "n1."+ask2, []byte(`{}`)); err != nil {
 		t.Fatal(err)
 	}
-	until("a pulse for the second check", func() bool { _, ok := has(delivery.PulseBucket, pulseOf(watch.ChecksBucket, "o.r.8")); return ok })
-	if err := kvs[watch.ChecksBucket].Delete(ctx, "o.r.8"); err != nil {
-		t.Fatal(err)
-	}
-	until("the pulse of the second check deleted", func() bool { _, ok := has(delivery.PulseBucket, pulseOf(watch.ChecksBucket, "o.r.8")); return !ok })
+	until("the pulse of the second ask deleted", func() bool { _, ok := has(delivery.PulseBucket, pulseOf(askStream, ask2)); return !ok })
 	time.Sleep(6 * time.Second) // twice the bound and the marker time
-	for _, k := range []string{stuckOf(delivery.PresenceBucket, "n2"), stuckOf(watch.ChecksBucket, "o.r.8")} {
+	for _, k := range []string{stuckOf(delivery.PresenceBucket, "n2"), stuckOf(askStream, ask2)} {
 		if _, ok := has(delivery.StuckBucket, k); ok {
 			t.Errorf("a stuck key %s was written for a move that happened before the bound", k)
 		}

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

@@ -1,5 +1,5 @@
 // Command mesh-delivery-watchman is delivery's watchman (novox/hq ADR 0429 §1, ADR 0431, ADR 0433, issue 1135): it
-// hears the bus's connect and disconnect events and writes presence, mirrors every move a machine owes (a check running,
+// hears the bus's connect and disconnect events and writes presence, mirrors every move a machine owes (a build ask waiting for a node,
 // a node or part gone offline) into the pulse bucket with a time to live, and writes
 // a stuck key when the server expires one. It holds no timer: every reaction is an event, a watch or a request at
 // start or at a reconnect, and the server's expiry is the only clock.
@@ -179,6 +179,17 @@ func follow(ctx context.Context, w jetstream.KeyWatcher, each func(jetstream.Key
 	}
 }
 
+// stepTaken is a change of a build module's steps: a step key written for an ask is that ask taken by a node. The node's
+// `paused.<node>` key and a key that is no step key are left.
+func stepTaken(w *watch.Watchman, language string, e jetstream.KeyValueEntry) {
+	if opOf(e) != watch.Put {
+		return
+	}
+	if _, id, err := delivery.ParseStepKey(e.Key()); err == nil {
+		w.Taken(language, id)
+	}
+}
+
 // run is the program: it returns when ctx ends or the bus cannot be used.
 func run(ctx context.Context, getenv func(string) string, logOut io.Writer) error {
 	logger := log.New(logOut, "mesh-delivery-watchman: ", log.LstdFlags)
@@ -223,12 +234,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		}
 		return kv, nil
 	}
-	var kvs struct{ presence, pulse, stuck, checks jetstream.KeyValue }
+	var kvs struct{ presence, pulse, stuck jetstream.KeyValue }
 	for _, b := range []struct {
 		name string
 		into *jetstream.KeyValue
-	}{{delivery.PresenceBucket, &kvs.presence}, {delivery.PulseBucket, &kvs.pulse}, {delivery.StuckBucket, &kvs.stuck},
-		{watch.ChecksBucket, &kvs.checks}} {
+	}{{delivery.PresenceBucket, &kvs.presence}, {delivery.PulseBucket, &kvs.pulse}, {delivery.StuckBucket, &kvs.stuck}} {
 		if *b.into, err = bucket(b.name); err != nil {
 			return err
 		}
@@ -275,11 +285,22 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	if err != nil {
 		return err
 	}
-	checks, err := settle(ctx, kvs.checks, func(e jetstream.KeyValueEntry) {
-		w.Check(e.Key(), e.Value(), e.Created(), opOf(e))
-	})
-	if err != nil {
-		return err
+	// 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.
+	taken := map[string]jetstream.KeyWatcher{}
+	for _, lang := range delivery.AskLanguages {
+		module, _ := delivery.BuildModuleOf(lang)
+		kv, err := js.KeyValue(ctx, delivery.StepsBucket(module))
+		if err != nil {
+			logger.Printf("the steps of %s cannot be read (%v): its asks are mirrored and never seen taken", module, err)
+			continue
+		}
+		lang := lang
+		sw, err := settle(ctx, kv, func(e jetstream.KeyValueEntry) { stepTaken(w, lang, e) })
+		if err != nil {
+			return err
+		}
+		taken[lang] = sw
 	}
 
 	// The events first, then the one request of the open connections, so no connection falls between them.
@@ -297,6 +318,23 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 			}
 		}
 	}
+	// The asks put on the build seats' streams: a plain subscription hears each as it is published, as the stream captures it.
+	for _, lang := range delivery.AskLanguages {
+		for _, kind := range delivery.AskKinds {
+			subject, _ := delivery.AskSubject(lang, kind)
+			lang := lang
+			if _, err := conn.Subscribe(subject, func(m *nats.Msg) {
+				a, err := delivery.DecodeStepAsk(m.Data)
+				if err != nil {
+					logger.Printf("an ask on %s is not readable: %v", m.Subject, err)
+					return
+				}
+				w.Asked(lang, a.ID, a.At)
+			}); err != nil {
+				return err
+			}
+		}
+	}
 	for subject, open := range map[string]bool{connectSubject: true, disconnectSubject: false} {
 		if _, err := conn.Subscribe(subject, onEvent(open)); err != nil {
 			return err
@@ -317,13 +355,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	}
 	w.Reconcile()
 
-	errs := make(chan error, 4)
+	errs := make(chan error, 8)
 	go func() {
 		errs <- follow(ctx, pulses, func(e jetstream.KeyValueEntry) { w.Pulse(e.Key(), opOf(e), e.Created()) })
 	}()
-	go func() {
-		errs <- follow(ctx, checks, func(e jetstream.KeyValueEntry) { w.Check(e.Key(), e.Value(), e.Created(), opOf(e)) })
-	}()
+	for lang, sw := range taken {
+		lang, sw := lang, sw
+		go func() { errs <- follow(ctx, sw, func(e jetstream.KeyValueEntry) { stepTaken(w, lang, e) }) }()
+	}
 	go func() {
 		for {
 			select {

modules/mesh-delivery-watchman/internal/watch/mirror.go

@@ -1,42 +1,42 @@
 package watch
 
 import (
-	"encoding/json"
+	"git.novox.be/novox/mesh-sdk/go/delivery"
 	"time"
 )
 
-// ChecksBucket is the bucket of the checks the builders are running (mesh-controller internal/broker
-// CheckRunningBucket).
-const ChecksBucket = "mesh-controller_checks-running"
+// A waiting ask is a move (novox/hq ADR 0525 §4, issue 1887 A8b): a message on a build seat's stream (mesh-sdk go/delivery
+// AskStream) that a node of the seat must pull. Its pulse is the pulse of the bucket named as the stream, under the ask's id;
+// the ask put on the seat is the move beginning at its time, and the ask taken (its step key `<node>.<ask id>` written in
+// the build module's steps, delivery.StepsBucket) is the move done. An ask no node takes within the bound is stuck, as any
+// move. The watchman reads no bucket of the controller for it.
 
-// checkRecord is the part of a running check's key the watchman reads, as the controller's builder writes it
-// (mesh-controller internal/link/checks_nats.go CheckRunning): the head being checked.
-type checkRecord struct {
-	Head string `json:"head"`
-}
+// AskedKey names a waiting ask for the pulse and stuck keys: the seat's stream and the ask's id.
+func askedBucket(language string) (string, error) { return delivery.AskStream(language) }
 
-// Check is a change in the checks-running bucket. The builder puts a check's key when the check starts and puts it
-// again at every heartbeat while the check lives, and deletes it when the check ends. Each new revision of the key is
-// the check moving: the pulse is deleted and written again with the full bound from that revision's time, so only a
-// check whose builder stops heartbeating for the whole bound is stuck; the key deleted is the check ended. An aged-out
-// key is the bucket's own expiry of a builder that died: the pulse is left to run out.
-func (w *Watchman) Check(key string, value []byte, at time.Time, op Op) {
-	w.mu.Lock()
-	defer w.mu.Unlock()
-	switch op {
-	case Expired:
-		return
-	case Delete:
-		w.moved(ChecksBucket, key)
+// Asked is an ask of a build seat of language put on its stream at time at (the ask's own time).
+func (w *Watchman) Asked(language, id string, at time.Time) {
+	bucket, err := askedBucket(language)
+	if err != nil {
+		w.say("no pulse for the ask %s: %v", id, err)
 		return
 	}
-	var rec checkRecord
-	_ = json.Unmarshal(value, &rec)
+	w.mu.Lock()
+	defer w.mu.Unlock()
 	if at.IsZero() {
 		at = w.now()
 	}
-	if _, stalled := w.ttl(w.now().Sub(at)); !stalled {
-		w.stuckMoved(ChecksBucket, key)
+	w.mirror(bucket, id, describe("the ask %s to be taken by a node", id), "", at, 0)
+}
+
+// Taken is the ask taken by a node: its step key written. The pulse is deleted, so it never expires, and a stuck key it left is
+// written as moved.
+func (w *Watchman) Taken(language, id string) {
+	bucket, err := askedBucket(language)
+	if err != nil {
+		return
 	}
-	w.mirror(ChecksBucket, key, describe("the check of %s to end", key), rec.Head, at, 0)
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	w.moved(bucket, id)
 }

modules/mesh-delivery-watchman/internal/watch/watch_test.go

@@ -74,7 +74,7 @@ func checkValue(started time.Time, head string) []byte {
 }
 
 const (
-	checksKey  = "mesh-controller_checks-running"
+	checksKey  = "SEAT_NODE_BUILD_GO"
 	presenceNS = delivery.PresenceBucket
 )
 
@@ -207,13 +207,13 @@ func TestRebuildTakesThePresenceOfWhoIsGone(t *testing.T) { // 1135/AC2
 func TestRebuildPulsesFromCurrentKeys(t *testing.T) { // 1135/AC2
 	w, o, c := setup()
 	c.t = t0.Add(30 * time.Second)
-	w.Check("o.r.7", checkValue(t0, "abc"), t0.Add(10*time.Second), Put)
+	w.Asked("go", "askid1", t0.Add(10*time.Second))
 	w.SeedPresence("ace", false, t0)
 	w.Reconcile()
 	if err := w.Rebuild(func([]byte) ([]byte, error) { return connzAnswer(), nil }); err != nil {
 		t.Fatal(err)
 	}
-	for _, k := range []string{pulseKeyOf(t, checksKey, "o.r.7"), pulseKeyOf(t, presenceNS, "ace")} {
+	for _, k := range []string{pulseKeyOf(t, checksKey, "askid1"), pulseKeyOf(t, presenceNS, "ace")} {
 		if _, ok := o.pulses[k]; !ok {
 			t.Errorf("no pulse %s; have %v", k, o.pulses)
 		}
@@ -222,11 +222,11 @@ func TestRebuildPulsesFromCurrentKeys(t *testing.T) { // 1135/AC2
 
 func TestReconcileDropsPulsesNothingMirrorsAnyMore(t *testing.T) { // 1135/AC2
 	w, o, _ := setup()
-	stale := pulseKeyOf(t, checksKey, "o.r.gone")
-	live := pulseKeyOf(t, checksKey, "o.r.7")
+	stale := pulseKeyOf(t, checksKey, "gone")
+	live := pulseKeyOf(t, checksKey, "askid1")
 	w.SeedPulse(stale)
 	w.SeedPulse(live)
-	w.Check("o.r.7", checkValue(t0, "abc"), t0, Put)
+	w.Asked("go", "askid1", t0)
 	w.Reconcile()
 	if !slices.Equal(o.pulseDeletes, []string{stale}) {
 		t.Fatalf("deleted = %v, want only %s", o.pulseDeletes, stale)
@@ -239,12 +239,12 @@ func TestReconcileDropsPulsesNothingMirrorsAnyMore(t *testing.T) { // 1135/AC2
 func TestPulseTimeToLiveIsTheBoundMinusTheAge(t *testing.T) { // 1135/AC3
 	w, o, c := setup()
 	c.t = t0.Add(30 * time.Second)
-	w.Check("o.r.7", checkValue(t0, "abc"), t0, Put) // the key found at start, last renewed 30s ago
-	k := pulseKeyOf(t, checksKey, "o.r.7")
+	w.Asked("go", "askid1", t0) // the key found at start, last renewed 30s ago
+	k := pulseKeyOf(t, checksKey, "askid1")
 	if o.ttls[k] != 90*time.Second {
 		t.Fatalf("ttl = %v, want 90s (last renewed 30s ago)", o.ttls[k])
 	}
-	if p := o.pulses[k]; p.Bucket != checksKey || p.Key != "o.r.7" {
+	if p := o.pulses[k]; p.Bucket != checksKey || p.Key != "askid1" {
 		t.Fatalf("pulse = %+v", p)
 	}
 	w.Connect("node.ace")
@@ -257,15 +257,15 @@ func TestPulseTimeToLiveIsTheBoundMinusTheAge(t *testing.T) { // 1135/AC3
 func TestAMoveAtTheBoundIsStuckAtOnce(t *testing.T) { // 1135/AC3
 	w, o, c := setup()
 	c.t = t0.Add(120 * time.Second)
-	w.Check("o.r.7", checkValue(t0, "abc"), t0, Put)
+	w.Asked("go", "askid1", t0)
 	if len(o.pulses) != 0 {
 		t.Fatalf("a pulse was written for a heartbeat 120s old: %v", o.pulses)
 	}
-	s, ok := o.stuck[stuckKeyOf(t, checksKey, "o.r.7")]
+	s, ok := o.stuck[stuckKeyOf(t, checksKey, "askid1")]
 	if !ok {
 		t.Fatalf("no stuck key; have %v", o.stuck)
 	}
-	if s.Moved || !s.Since.Equal(t0) || s.Waited != "the check of o.r.7 to end" {
+	if s.Moved || !s.Since.Equal(t0) || s.Waited != "the ask askid1 to be taken by a node" {
 		t.Fatalf("stuck = %+v", s)
 	}
 }
@@ -273,13 +273,13 @@ func TestAMoveAtTheBoundIsStuckAtOnce(t *testing.T) { // 1135/AC3
 func TestWaitedInWords(t *testing.T) { // 1135/AC3
 	w, o, c := setup()
 	c.t = t0.Add(200 * time.Second)
-	w.Check("o.r.7", checkValue(t0, "abc"), t0, Put)
+	w.Asked("go", "askid1", t0)
 	w.SeedPresence("ace", false, t0)
 	w.SeedPresence("part.mesh-delivery-view.ace", false, t0)
 	w.Reconcile()
 	_ = w.Rebuild(func([]byte) ([]byte, error) { return connzAnswer(), nil })
 	want := map[string]string{
-		stuckKeyOf(t, checksKey, "o.r.7"):                        "the check of o.r.7 to end",
+		stuckKeyOf(t, checksKey, "askid1"):                       "the ask askid1 to be taken by a node",
 		stuckKeyOf(t, presenceNS, "ace"):                         "ace to come back online",
 		stuckKeyOf(t, presenceNS, "part.mesh-delivery-view.ace"): "mesh-delivery-view on ace to come back online",
 	}
@@ -292,14 +292,14 @@ func TestWaitedInWords(t *testing.T) { // 1135/AC3
 
 func TestEveryHeartbeatRenewsThePulse(t *testing.T) { // 1135/AC3
 	w, o, c := setup()
-	k := pulseKeyOf(t, checksKey, "o.r.7")
-	w.Check("o.r.7", checkValue(t0, "abc"), t0, Put)
+	k := pulseKeyOf(t, checksKey, "askid1")
+	w.Asked("go", "askid1", t0)
 	if o.ttls[k] != 120*time.Second {
 		t.Fatalf("ttl = %v, want the full 120s", o.ttls[k])
 	}
 	o.ttls[k] = 0
 	c.t = t0.Add(100 * time.Second)
-	w.Check("o.r.7", checkValue(t0, "abc"), c.t, Put) // the builder re-puts the key: same check, started long ago
+	w.Asked("go", "askid1", c.t) // the builder re-puts the key: same check, started long ago
 	if o.ttls[k] != 120*time.Second {
 		t.Fatalf("a heartbeat left the pulse as it was: ttl = %v, want the full 120s again", o.ttls[k])
 	}
@@ -308,7 +308,7 @@ func TestEveryHeartbeatRenewsThePulse(t *testing.T) { // 1135/AC3
 	}
 	// The same revision heard twice changes nothing.
 	o.ttls[k] = 0
-	w.Check("o.r.7", checkValue(t0, "abc"), c.t, Put)
+	w.Asked("go", "askid1", c.t)
 	if o.ttls[k] != 0 {
 		t.Fatal("one revision rewrote the pulse twice")
 	}
@@ -317,10 +317,10 @@ func TestEveryHeartbeatRenewsThePulse(t *testing.T) { // 1135/AC3
 func TestMirroredMovesDeleteThePulse(t *testing.T) { // 1135/AC4
 	w, o, c := setup()
 	c.t = t0.Add(10 * time.Second)
-	w.Check("o.r.7", checkValue(t0, "abc"), t0, Put)
+	w.Asked("go", "askid1", t0)
 	w.Connect("node.ace")
 	w.Disconnect("node.ace")
-	w.Check("o.r.7", nil, c.t, Delete) // the check ended
+	w.Taken("go", "askid1") // the check ended
 	w.Connect("node.ace")
 	if len(o.pulses) != 0 {
 		t.Fatalf("pulses left: %v", o.pulses)
@@ -335,12 +335,12 @@ func TestMirroredMovesDeleteThePulse(t *testing.T) { // 1135/AC4
 
 func TestACheckWhoseBuilderStopsIsStuckFromItsLastHeartbeat(t *testing.T) { // 1135/AC4
 	w, o, c := setup()
-	w.Check("o.r.7", checkValue(t0, "abc"), t0, Put)
+	w.Asked("go", "askid1", t0)
 	c.t = t0.Add(50 * time.Second)
-	w.Check("o.r.7", checkValue(t0, "abc"), c.t, Put) // the last heartbeat
+	w.Asked("go", "askid1", c.t) // the last heartbeat
 	c.t = t0.Add(170 * time.Second)
-	w.Pulse(pulseKeyOf(t, checksKey, "o.r.7"), Expired, c.t)
-	s := o.stuck[stuckKeyOf(t, checksKey, "o.r.7")]
+	w.Pulse(pulseKeyOf(t, checksKey, "askid1"), Expired, c.t)
+	s := o.stuck[stuckKeyOf(t, checksKey, "askid1")]
 	if !s.Since.Equal(t0.Add(50 * time.Second)) {
 		t.Fatalf("stuck = %+v, want since the last heartbeat", s)
 	}
@@ -348,35 +348,35 @@ func TestACheckWhoseBuilderStopsIsStuckFromItsLastHeartbeat(t *testing.T) { // 1
 
 func TestAStuckKeyIsWrittenMovedNeverDeleted(t *testing.T) { // 1135/AC4
 	w, o, c := setup()
-	w.Check("o.r.7", checkValue(t0, "abc"), t0, Put)
-	pk := pulseKeyOf(t, checksKey, "o.r.7")
+	w.Asked("go", "askid1", t0)
+	pk := pulseKeyOf(t, checksKey, "askid1")
 	c.t = t0.Add(120 * time.Second)
 	w.Pulse(pk, Expired, c.t)
-	sk := stuckKeyOf(t, checksKey, "o.r.7")
+	sk := stuckKeyOf(t, checksKey, "askid1")
 	if o.stuck[sk].Moved {
 		t.Fatal("stuck written as moved at expiry")
 	}
 	c.t = t0.Add(300 * time.Second)
-	w.Check("o.r.7", nil, c.t, Delete)
+	w.Taken("go", "askid1")
 	s := o.stuck[sk]
-	if !s.Moved || !s.Since.Equal(t0) || s.Waited != "the check of o.r.7 to end" {
+	if !s.Moved || !s.Since.Equal(t0) || s.Waited != "the ask askid1 to be taken by a node" {
 		t.Fatalf("stuck after the move = %+v, want the same key moved", s)
 	}
 	n := o.stuckWrites
-	w.Check("o.r.7", nil, c.t, Delete)
+	w.Taken("go", "askid1")
 	if o.stuckWrites != n {
 		t.Fatal("a stuck key already moved was written again")
 	}
 }
 
-func TestAHeartbeatAfterStuckMovesIt(t *testing.T) { // 1135/AC4
+func TestAnAskTakenAfterStuckMovesIt(t *testing.T) { // 1135/AC4
 	w, o, c := setup()
-	w.Check("o.r.7", checkValue(t0, "abc"), t0, Put)
+	w.Asked("go", "askid1", t0)
 	c.t = t0.Add(120 * time.Second)
-	w.Pulse(pulseKeyOf(t, checksKey, "o.r.7"), Expired, c.t)
-	w.Check("o.r.7", checkValue(t0, "abc"), c.t, Put) // the builder is back
-	if !o.stuck[stuckKeyOf(t, checksKey, "o.r.7")].Moved {
-		t.Fatal("a check heartbeating again is still stuck")
+	w.Pulse(pulseKeyOf(t, checksKey, "askid1"), Expired, c.t)
+	w.Taken("go", "askid1") // a node took it at last
+	if !o.stuck[stuckKeyOf(t, checksKey, "askid1")].Moved {
+		t.Fatal("an ask taken after it was stuck is still stuck")
 	}
 }
 
@@ -398,23 +398,23 @@ func TestPresenceStuckThenBackOnline(t *testing.T) { // 1135/AC4
 
 func TestExpiryMarkerWritesStuck(t *testing.T) { // 1135/AC5
 	w, o, c := setup()
-	w.Check("o.r.7", checkValue(t0, "abc"), t0, Put)
+	w.Asked("go", "askid1", t0)
 	c.t = t0.Add(120 * time.Second)
-	w.Pulse(pulseKeyOf(t, checksKey, "o.r.7"), Expired, c.t)
-	s, ok := o.stuck[stuckKeyOf(t, checksKey, "o.r.7")]
+	w.Pulse(pulseKeyOf(t, checksKey, "askid1"), Expired, c.t)
+	s, ok := o.stuck[stuckKeyOf(t, checksKey, "askid1")]
 	if !ok {
 		t.Fatalf("no stuck key; have %v", o.stuck)
 	}
-	if s.Bucket != checksKey || s.Key != "o.r.7" || s.Waited != "the check of o.r.7 to end" || !s.Since.Equal(t0) || s.Moved {
+	if s.Bucket != checksKey || s.Key != "askid1" || s.Waited != "the ask askid1 to be taken by a node" || !s.Since.Equal(t0) || s.Moved {
 		t.Fatalf("stuck = %+v", s)
 	}
 }
 
 func TestOwnDeleteOfAPulseWritesNothing(t *testing.T) { // 1135/AC5
 	w, o, c := setup()
-	w.Check("o.r.7", checkValue(t0, "abc"), t0, Put)
-	pk := pulseKeyOf(t, checksKey, "o.r.7")
-	w.Check("o.r.7", nil, c.t, Delete)
+	w.Asked("go", "askid1", t0)
+	pk := pulseKeyOf(t, checksKey, "askid1")
+	w.Taken("go", "askid1")
 	w.Pulse(pk, Delete, c.t) // the delete the watchman made, heard on its own watch
 	w.Pulse(pk, Expired, c.t)
 	w.Pulse("mesh-delivery-watchman_pulse.unknown.key", Expired, c.t)
@@ -427,13 +427,13 @@ func TestTheBoundCanBeShortenedForATest(t *testing.T) { // 1135/AC8
 	w, o, c := setup()
 	w.SetBound(2 * time.Second)
 	c.t = t0.Add(500 * time.Millisecond)
-	w.Check("o.r.7", checkValue(t0, "abc"), t0, Put)
-	if got := o.ttls[pulseKeyOf(t, checksKey, "o.r.7")]; got != 1500*time.Millisecond {
+	w.Asked("go", "askid1", t0)
+	if got := o.ttls[pulseKeyOf(t, checksKey, "askid1")]; got != 1500*time.Millisecond {
 		t.Fatalf("ttl = %v, want 1.5s", got)
 	}
 	c.t = t0.Add(3 * time.Second)
-	w.Check("o.r.8", checkValue(t0, "abc"), t0, Put)
-	if _, ok := o.stuck[stuckKeyOf(t, checksKey, "o.r.8")]; !ok {
+	w.Asked("go", "askid2", t0)
+	if _, ok := o.stuck[stuckKeyOf(t, checksKey, "askid2")]; !ok {
 		t.Fatalf("a heartbeat older than the bound is stuck at once; have %v", o.stuck)
 	}
 }

modules/mesh-delivery-watchman/manifest_test.go

@@ -81,7 +81,7 @@ func TestManifestStateAndReads(t *testing.T) { // 1135/AC6
 	}
 	reads := slices.Clone(m.Reads)
 	slices.Sort(reads)
-	if !reflect.DeepEqual(reads, []string{"mesh-controller.checks-running"}) {
+	if !reflect.DeepEqual(reads, []string{"mesh-build-go.steps", "mesh-build-image.steps", "mesh-build-node.steps"}) {
 		t.Errorf("reads = %v", reads)
 	}
 	for _, k := range []string{"emits", "context", "items", "consumes", "provides", "claims"} {

modules/mesh-delivery-watchman/module.json

@@ -3,7 +3,9 @@
   "version": "1",
   "slug": "watchman",
   "reads": [
-    "mesh-controller.checks-running"
+    "mesh-build-go.steps",
+    "mesh-build-image.steps",
+    "mesh-build-node.steps"
   ],
   "state": [
     { "name": "presence", "history": 20 },