AS5-3: the composition step runs one composer request per machine as a run ask and reads its verdict from step-ended

Each machine placing the module on the step's branch is asked compose-candidate for its input; one composer.Request per machine
(depth compose, the one machine, one case change) carries it with the manifests, verdicts and seat definitions of the step's
catalog revision read through mesh-catalog's revision; the request is kept under its run ask (refused at the inputs state's
64 MiB cap, naming the store); one run ask per machine on node-build-go's check queue under the assigner's name, naming no head,
its program the node-engine build's catalog entry (the build each machine runs as its running says, read through entry, or the
revision's where it differs from the trunk's) with the composer's package. The run's end is heard on the build seat's step-ended
and its verdict read through buildstep.StepEnded.ReadOutput; a machine the watchman said away passes with a warning. The step
is served from main.

Builds: design 47 G §354, §22, §63, §64, §389, A-226, A-238, A-240, A-243, A-252, A-254, A-261

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

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

@@ -18,7 +18,7 @@ import (
 )
 
 // The inputs of a composition step (design 47 G §354, A-243, A-252): one request per machine, each carrying that machine's input,
-// asked of the machine's node-engine by the seat's verb compose-candidate and kept by the digest the run ask names it by.
+// asked of the machine's node-engine by the seat's verb compose-candidate.
 
 // composeCandidateKey is the seat verb's call key, asked of one node.
 const composeCandidateKey = "seat:" + ne.Seat + "." + ne.VerbComposeCandidate
@@ -59,28 +59,28 @@ func (w *world) machinesOf(module, branch string) []string {
 	return machinesFor(w.placedBranches(), module, branch, w.holdsBranch(branch))
 }
 
-// nodeInput is one machine's input: its digest and bytes, or why the node gave none.
+// nodeInput is one machine's input: its bytes, or why the node gave none, or that it is away.
 type nodeInput struct {
-	Node   string
-	Digest string
-	Body   []byte
+	Node string
+	Body []byte
 	// Why is set when the node gave no input: its words, or the call's error.
 	Why string
+	// Away is a machine the watchman said away: it is not asked (§64).
+	Away bool
 }
 
-// compositionInputs asks the machines of a step for their inputs and keeps them.
+// compositionInputs asks the machines of a step for their inputs.
 type compositionInputs struct {
 	machines func(module, branch string) []string
 	// call asks one node's seat verb (process.Conn.CallKeptOn).
-	call   func(node, key string, body any) (json.RawMessage, error)
-	keeper *inputKeeper
-	bound  time.Duration
+	call  func(node, key string, body any) (json.RawMessage, error)
+	bound time.Duration
 }
 
-// gather asks each machine of the step's module and branch for its input with the step's candidate key, keeps each input under the
-// ask askOf names for that node (the build seat's step-ended of that ask deletes it), and answers every machine in name order. A node
-// that is silent, unready or answers no input has Why set and keeps none; no machine at all is an error.
-func (c *compositionInputs) gather(ctx context.Context, run meshwants.StepRun, askOf func(node string) string) ([]nodeInput, error) {
+// gather asks each machine of the step's module and branch for its input with the step's candidate key, and answers every machine in
+// name order: one away (away) is not asked; a node that is silent, unready or answers no input has Why set; no machine at all is an
+// error. The run keeps each request built from an input under its run ask (compositionRun.machine).
+func (c *compositionInputs) gather(ctx context.Context, run meshwants.StepRun, away func(node string) bool) ([]nodeInput, error) {
 	key, err := candidateKeyOf(run)
 	if err != nil {
 		return nil, err
@@ -96,18 +96,22 @@ func (c *compositionInputs) gather(ctx context.Context, run meshwants.StepRun, a
 	out := make([]nodeInput, len(nodes))
 	var wg sync.WaitGroup
 	for i, node := range nodes {
+		if away != nil && away(node) {
+			out[i] = nodeInput{Node: node, Away: true}
+			continue
+		}
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
-			out[i] = c.one(ctx, bound, node, key, askOf(node))
+			out[i] = c.one(ctx, bound, node, key)
 		}()
 	}
 	wg.Wait()
 	return out, nil
 }
 
-// one asks one node and keeps its input.
-func (c *compositionInputs) one(ctx context.Context, bound time.Duration, node, key, ask string) nodeInput {
+// one asks one node for its input.
+func (c *compositionInputs) one(ctx context.Context, bound time.Duration, node, key string) nodeInput {
 	in := nodeInput{Node: node}
 	at, cancel := context.WithTimeout(ctx, bound)
 	defer cancel()
@@ -156,11 +160,6 @@ func (c *compositionInputs) one(ctx context.Context, bound time.Duration, node,
 		in.Why = "its composition input cannot be encoded again: " + err.Error()
 		return in
 	}
-	digest, err := c.keeper.keep(ask, body)
-	if err != nil {
-		in.Why = "its input could not be kept: " + err.Error()
-		return in
-	}
-	in.Digest, in.Body = digest, body
+	in.Body = body
 	return in
 }

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

@@ -52,9 +52,9 @@ func TestTheMachinesOfAStep(t *testing.T) {
 	}
 }
 
-// §354, A-243, A-252: each machine is asked compose-candidate with the step's key and input true, and its input is kept under the ask
-// named for it; a silent or unready node keeps none and says why.
-func TestAStepsInputsAreAskedOfEachMachineAndKept(t *testing.T) {
+// §354, A-243, A-252, §64: each machine is asked compose-candidate with the step's key and input true, its numbers as written; a
+// silent or unready node says why, and a machine away is not asked.
+func TestAStepsInputsAreAskedOfEachMachine(t *testing.T) {
 	asked := map[string]string{}
 	call := func(node, key string, body any) (json.RawMessage, error) {
 		raw, _ := json.Marshal(body)
@@ -67,10 +67,9 @@ func TestAStepsInputsAreAskedOfEachMachineAndKept(t *testing.T) {
 		}
 		return json.RawMessage(`{"candidate":"run-r1","node":"` + node + `","verdict":"composed","input":{"n":12345678901234567890,"node":"` + node + `"},"modules":[],"own":{},"took":1,"v":1}`), nil
 	}
-	keeper := testKeeper(&objectsMem{objs: map[string][]byte{}}, time.Now)
-	c := &compositionInputs{machines: func(module, branch string) []string { return []string{"ace", "g14", "novox"} }, call: call, keeper: keeper, bound: time.Second}
-	got, err := c.gather(context.Background(), meshwants.StepRun{Module: "m", Branch: "main", Run: "r1"}, func(node string) string { return "ask-" + node })
-	if err != nil || len(got) != 3 {
+	c := &compositionInputs{machines: func(module, branch string) []string { return []string{"ace", "g14", "novox", "shanks"} }, call: call, bound: time.Second}
+	got, err := c.gather(context.Background(), meshwants.StepRun{Module: "m", Branch: "main", Run: "r1"}, func(node string) bool { return node == "shanks" })
+	if err != nil || len(got) != 4 {
 		t.Fatalf("%+v %v", got, err)
 	}
 	if want := `seat:node-engine.compose-candidate {"candidate":"run-r1","input":true}`; asked["ace"] != want {
@@ -79,19 +78,13 @@ func TestAStepsInputsAreAskedOfEachMachineAndKept(t *testing.T) {
 	if got[0].Node != "ace" || got[0].Why != "" || !strings.Contains(string(got[0].Body), "12345678901234567890") {
 		t.Errorf("ace: %+v", got[0])
 	}
-	if b, err := keeper.body(got[0].Digest); err != nil || string(b) != string(got[0].Body) {
-		t.Errorf("the input of ace is not kept: %v", err)
-	}
-	if !strings.Contains(got[1].Why, "no responders") || !strings.Contains(got[2].Why, "unready") || got[1].Digest != "" || got[2].Digest != "" {
+	if !strings.Contains(got[1].Why, "no responders") || !strings.Contains(got[2].Why, "unready") || got[1].Body != nil || got[2].Body != nil {
 		t.Errorf("g14 and novox: %+v %+v", got[1], got[2])
 	}
-	if err := keeper.ended("ask-ace"); err != nil {
-		t.Fatal(err)
+	if _, was := asked["shanks"]; was || !got[3].Away {
+		t.Errorf("the machine away is asked or not said away: %+v", got[3])
 	}
-	if _, err := keeper.body(got[0].Digest); err == nil {
-		t.Error("the input is kept past its step-ended")
-	}
-	none := &compositionInputs{machines: func(string, string) []string { return nil }, call: call, keeper: keeper}
+	none := &compositionInputs{machines: func(string, string) []string { return nil }, call: call}
 	if _, err := none.gather(context.Background(), meshwants.StepRun{Module: "m", Run: "r1"}, nil); err == nil {
 		t.Error("a step with no machine is gathered")
 	}

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

@@ -0,0 +1,498 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"slices"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/nats-io/nats.go/jetstream"
+
+	ne "git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
+	mcc "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient"
+	meshcatalog "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
+	meshwants "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants"
+	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildstep"
+	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/composer"
+	"git.novox.be/novox/mesh-sdk/go/buscalls"
+	"git.novox.be/novox/mesh-sdk/go/keys"
+)
+
+// The run of a composition step (design 47 G §354, §63, §64; A-226, A-238, A-240, A-243, A-252, A-254, A-261): one
+// composer.Request per machine, depth compose, the one machine and one case `change`, its manifests, verdicts and seats from the
+// step's catalog revision read through mesh-catalog's `revision`; one run ask per machine of the seat node-build-go, naming no
+// head, its program the node-engine build's catalog entry with the composer's package; the run's end heard on the build seat's
+// step-ended of that ask, its answer (a composer.Verdict) read inline from `output`. A machine the watchman said away
+// (machine-away) is not composed and passes with a warning (§64).
+
+const (
+	// runAsker is the assigner's name as an asker of a build seat: the last token of the subject its run asks are put on and
+	// the first of their step keys (§467, §481, §600).
+	runAsker = "assigner"
+	// changeCase is the one case each request composes.
+	changeCase = "change"
+	// inputVerbKey is the verb a run ask's input is served on: the seat mesh-wants' input (§354).
+	inputVerbKey = "seat:" + meshwants.Seat + "." + meshwants.VerbInput
+	// entryVerbKey and seatDefinitionsKey are the mesh-catalog seat verbs the step reads.
+	entryVerbKey       = "seat:" + meshcatalog.Seat + "." + meshcatalog.VerbEntry
+	seatDefinitionsKey = "seat:" + meshcatalog.Seat + "." + meshcatalog.VerbSeatDefinitions
+)
+
+// runEnds hands each build seat's step-ended of an ask to the composition run that waits for it (A-240), by the ask's id.
+type runEnds struct {
+	mu    sync.Mutex
+	waits map[string]chan buildstep.StepEnded
+}
+
+// expect is the end of the ask id, once heard; called before the ask is put, so no end is missed.
+func (r *runEnds) expect(id string) <-chan buildstep.StepEnded {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if r.waits == nil {
+		r.waits = map[string]chan buildstep.StepEnded{}
+	}
+	ch := make(chan buildstep.StepEnded, 1)
+	r.waits[id] = ch
+	return ch
+}
+
+// forget stops waiting for the ask id.
+func (r *runEnds) forget(id string) {
+	r.mu.Lock()
+	delete(r.waits, id)
+	r.mu.Unlock()
+}
+
+// heard hands an end to the run that waits for its ask; one nobody waits for is another asker's, left.
+func (r *runEnds) heard(e buildstep.StepEnded) {
+	r.mu.Lock()
+	ch := r.waits[e.Ask]
+	delete(r.waits, e.Ask)
+	r.mu.Unlock()
+	if ch != nil {
+		ch <- e
+	}
+}
+
+// awayNodes are the machines the watchman said away (machine-away of the seat mesh-watchdogs, §389) and not back since.
+type awayNodes struct {
+	mu   sync.Mutex
+	away map[string]bool
+}
+
+// set keeps machine away (true) or back (false).
+func (a *awayNodes) set(machine string, away bool) {
+	a.mu.Lock()
+	defer a.mu.Unlock()
+	if a.away == nil {
+		a.away = map[string]bool{}
+	}
+	if away {
+		a.away[machine] = true
+	} else {
+		delete(a.away, machine)
+	}
+}
+
+// is says machine is away.
+func (a *awayNodes) is(machine string) bool {
+	if a == nil {
+		return false
+	}
+	a.mu.Lock()
+	defer a.mu.Unlock()
+	return a.away[machine]
+}
+
+// compositionRun runs a composition step.
+type compositionRun struct {
+	inputs *compositionInputs
+	keeper *inputKeeper
+	// call asks a seat verb held once for the mesh (process.Conn.CallKept).
+	call func(key string, body any) (json.RawMessage, error)
+	// trunk is the trunk's catalog half as the assigner holds it.
+	trunk func() (mcc.Half, bool)
+	// running is the last `running` heard of a module on a node (world.runningOf), never the node-engine's bucket.
+	running func(node, module string) (ne.Running, bool)
+	away    *awayNodes
+	// put puts a run ask on the build seat's check queue under the assigner's name.
+	put  func(ctx context.Context, a buildstep.StepAsk) error
+	ends *runEnds
+	// ask asks a seat verb on one node, for the run's answer kept as the build seat's object (buildstep.StepEnded.ReadOutput).
+	ask  buildstep.AskVerb
+	now  func() time.Time
+	logf func(string, ...any)
+}
+
+// machineEnd is what one machine came to.
+type machineEnd struct {
+	failed, couldNot, warning string
+}
+
+// run runs one composition step and answers what it came to: failed when a machine does not compose, could not run when a machine
+// could not be composed, else passed with a warning per machine away or not judged.
+func (r *compositionRun) run(ctx context.Context, _ string, run meshwants.StepRun) stepOutcome {
+	half, err := r.revision(run)
+	if err != nil {
+		return couldNotRun("the step's catalog revision is not read: " + err.Error())
+	}
+	seats, err := r.seats(half.Revision.Definitions)
+	if err != nil {
+		return couldNotRun(err.Error())
+	}
+	manifests, verdicts, err := caseOf(half)
+	if err != nil {
+		return couldNotRun(err.Error())
+	}
+	programOf, err := r.programs(half)
+	if err != nil {
+		return couldNotRun(err.Error())
+	}
+	inputs, err := r.inputs.gather(ctx, run, r.away.is)
+	if err != nil {
+		return couldNotRun(err.Error())
+	}
+	ends := make([]machineEnd, len(inputs))
+	var wg sync.WaitGroup
+	for i, in := range inputs {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			ends[i] = r.machine(ctx, in, buildstep.NewID(), programOf, manifests, verdicts, seats)
+		}()
+	}
+	wg.Wait()
+	var out stepOutcome
+	var failed, couldNot []string
+	for _, e := range ends {
+		failed = appendSome(failed, e.failed)
+		couldNot = appendSome(couldNot, e.couldNot)
+		out.Warnings = appendSome(out.Warnings, e.warning)
+	}
+	switch {
+	case len(failed) > 0:
+		out.Outcome, out.Said = meshwants.StepEndedOutcomeFailed, strings.Join(failed, "; ")
+	case len(couldNot) > 0:
+		out.Outcome, out.Said = meshwants.StepEndedOutcomeCouldNotRun, strings.Join(couldNot, "; ")
+	default:
+		out.Outcome = meshwants.StepEndedOutcomePassed
+	}
+	return out
+}
+
+func appendSome(to []string, s string) []string {
+	if s == "" {
+		return to
+	}
+	return append(to, s)
+}
+
+func couldNotRun(said string) stepOutcome {
+	return stepOutcome{Outcome: meshwants.StepEndedOutcomeCouldNotRun, Said: said}
+}
+
+// machine composes one machine: its request kept under its ask, the ask put, its end heard and its verdict read.
+func (r *compositionRun) machine(ctx context.Context, in nodeInput, ask string, programOf func(node string) (buildstep.RunProgram, error),
+	manifests map[string]composer.Manifest, verdicts map[string]json.RawMessage, seats []json.RawMessage) machineEnd {
+	node := in.Node
+	switch {
+	case in.Away:
+		return machineEnd{warning: node + " is away (the watchman's machine-away), so it is not composed"}
+	case in.Why != "":
+		return machineEnd{couldNot: node + " gave no composition input: " + in.Why}
+	}
+	program, err := programOf(node)
+	if err != nil {
+		return machineEnd{couldNot: node + ": " + err.Error()}
+	}
+	req := composer.Request{V: composer.Version, Input: in.Body, Depth: composer.DepthCompose, Machines: []string{node},
+		Cases: []composer.Case{{Name: changeCase, Manifests: manifests, Verdicts: verdicts}}, Seats: seats}
+	raw, err := json.Marshal(req)
+	if err != nil {
+		return machineEnd{couldNot: node + ": its request cannot be encoded: " + err.Error()}
+	}
+	if _, err := r.keeper.keep(ask, raw); err != nil {
+		r.logf("the composition request of %s is refused: %v", node, err)
+		return machineEnd{couldNot: node + ": its request is not kept: " + err.Error()}
+	}
+	ref, err := buildstep.RunInputOf(inputVerbKey, raw)
+	if err != nil {
+		return machineEnd{couldNot: node + ": " + err.Error()}
+	}
+	a := buildstep.StepAsk{ID: ask, Kind: buildstep.AskCheck, Language: buildstep.AskGo, Asker: runAsker, At: r.now().UTC(),
+		Run: &buildstep.RunJob{Program: program, Heads: []buildstep.RunHead{}, Input: ref, Reader: "the composition of " + node + " by the node-engine at " + program.Commit}}
+	ended := r.ends.expect(ask)
+	defer r.ends.forget(ask)
+	if err := r.put(ctx, a); err != nil {
+		return machineEnd{couldNot: node + ": its run ask " + ask + " is not put: " + err.Error()}
+	}
+	var e buildstep.StepEnded
+	select {
+	case <-ctx.Done():
+		return machineEnd{couldNot: node + ": the step stopped before its run ask " + ask + " ended"}
+	case e = <-ended:
+	}
+	if e.Outcome != buildstep.OutcomePassed {
+		return machineEnd{couldNot: fmt.Sprintf("%s: its run ask %s ended %s on %s: %s", node, ask, e.Outcome, e.Node, e.Said)}
+	}
+	answer, err := e.ReadOutput(buildstep.AskGo, r.ask) // the program's last line of standard output (A-254)
+	if err != nil {
+		return machineEnd{couldNot: node + ": " + err.Error()}
+	}
+	return verdictOf(node, answer)
+}
+
+// verdictOf reads the composer's verdict of one machine: not composing is failed, naming why; not judged passes with a warning.
+func verdictOf(node string, answer []byte) machineEnd {
+	var v composer.Verdict
+	if err := json.Unmarshal(answer, &v); err != nil {
+		return machineEnd{couldNot: node + ": the composer answered no verdict: " + err.Error()}
+	}
+	if v.V != composer.Version {
+		return machineEnd{couldNot: fmt.Sprintf("%s: the composer answered a verdict of version %d, the assigner reads %d", node, v.V, composer.Version)}
+	}
+	for _, c := range v.Cases {
+		if c.Name != changeCase {
+			continue
+		}
+		m, ok := c.Machines[node]
+		switch {
+		case !ok:
+			return machineEnd{couldNot: node + ": the composer's verdict names no such machine"}
+		case m.NotJudged != "":
+			return machineEnd{warning: node + " was not judged: " + m.NotJudged}
+		case !m.Composes:
+			return machineEnd{failed: node + " does not compose: " + strings.Join(appendSome(append(slices.Clone(m.Problems), m.Refusals...), m.Reason), "; ")}
+		}
+		return machineEnd{}
+	}
+	return machineEnd{couldNot: node + ": the composer's verdict has no case " + changeCase}
+}
+
+// revision reads the step's catalog revision whole, page by page, through mesh-catalog's `revision` (A-252).
+func (r *compositionRun) revision(run meshwants.StepRun) (mcc.Half, error) {
+	if run.Revision <= 0 {
+		return mcc.Half{}, errors.New("the step names no catalog revision")
+	}
+	half := mcc.Half{Entries: map[string]meshcatalog.CatalogEntry{}}
+	in := meshcatalog.RevisionInput{Branch: keys.EscapeName(run.Branch), Revision: run.Revision}
+	for {
+		raw, err := r.call(mcc.RevisionVerb, in)
+		if err != nil {
+			return mcc.Half{}, err
+		}
+		body, said, ok := buscalls.Unwrap(raw)
+		if !ok {
+			return mcc.Half{}, fmt.Errorf("%s answered: %s", mcc.RevisionVerb, said)
+		}
+		var a meshcatalog.RevisionAnswer
+		if err := json.Unmarshal(body, &a); err != nil {
+			return mcc.Half{}, fmt.Errorf("%s answered no revision: %w", mcc.RevisionVerb, err)
+		}
+		if a.NoneYet || a.Revision == nil || a.Revision.Revision != run.Revision {
+			return mcc.Half{}, fmt.Errorf("%s answered no revision %d of %q", mcc.RevisionVerb, run.Revision, run.Branch)
+		}
+		if in.After == "" {
+			half.Revision = *a.Revision // the first page names what the revision was read from, its definitions included
+		}
+		for _, e := range a.Entries {
+			half.Entries[e.Module] = e
+		}
+		if a.Next == "" {
+			return half, nil
+		}
+		in.After = a.Next
+	}
+}
+
+// seats are the definitions of the revision's (seat, digest) pairs, raw (A-261): the revision's definitions carry no definition
+// yet (A-234), so each is mesh-catalog's seat-definitions answer for the seat where the trunk's digest is the revision's; a digest
+// neither holds is an error naming the seat and the digest.
+func (r *compositionRun) seats(defs []meshcatalog.CatalogRevisionDefinitionsItem) ([]json.RawMessage, error) {
+	out := make([]json.RawMessage, 0, len(defs))
+	for _, d := range defs {
+		raw, err := r.call(seatDefinitionsKey, meshcatalog.SeatDefinitionsInput{Seat: d.Seat})
+		if err != nil {
+			return nil, fmt.Errorf("the definition of the seat %s is not read: %w", d.Seat, err)
+		}
+		body, said, ok := buscalls.Unwrap(raw)
+		if !ok {
+			return nil, fmt.Errorf("the definition of the seat %s is not read: %s", d.Seat, said)
+		}
+		var a meshcatalog.SeatDefinitionsAnswer
+		if err := json.Unmarshal(body, &a); err != nil {
+			return nil, fmt.Errorf("the definition of the seat %s is not read: %w", d.Seat, err)
+		}
+		if a.Entry == nil || a.Entry.Digest != d.Digest || a.Entry.Definition == nil {
+			return nil, fmt.Errorf("no definition of the seat %s with the digest %s is held: neither the revision nor the trunk holds it", d.Seat, d.Digest)
+		}
+		def, err := json.Marshal(a.Entry.Definition)
+		if err != nil {
+			return nil, err
+		}
+		out = append(out, def)
+	}
+	return out, nil
+}
+
+// caseOf is the case's manifests and verdicts from the revision's entries: each manifest whole, and the verdict of each one not
+// accepted (a module without one is accepted).
+func caseOf(half mcc.Half) (map[string]composer.Manifest, map[string]json.RawMessage, error) {
+	manifests := map[string]composer.Manifest{}
+	verdicts := map[string]json.RawMessage{}
+	for name, e := range half.Entries {
+		if len(e.Manifest) == 0 {
+			continue
+		}
+		raw, err := json.Marshal(e.Manifest)
+		if err != nil {
+			return nil, nil, fmt.Errorf("the manifest of %s cannot be encoded: %w", name, err)
+		}
+		manifests[name] = composer.Manifest{Raw: raw}
+		if e.Verdict == "" || e.Verdict == mcc.VerdictAccepted {
+			continue
+		}
+		v := meshcatalog.VerdictsByDigestValue{V: 1, Module: name, State: e.Verdict}
+		if e.Verdict != mcc.VerdictRefused {
+			v.State = mcc.VerdictManifestWaits
+		}
+		if e.Inputs != nil {
+			v.Problems, v.Unknown, v.Warnings = e.Inputs.Problems, e.Inputs.Unknown, e.Inputs.Warnings
+		}
+		raw, err = json.Marshal(v)
+		if err != nil {
+			return nil, nil, err
+		}
+		verdicts[name] = raw
+	}
+	return manifests, verdicts, nil
+}
+
+// nodeEngineEntry is the entry of the module that claims the seat node-engine, or false.
+func nodeEngineEntry(entries map[string]meshcatalog.CatalogEntry) (meshcatalog.CatalogEntry, bool) {
+	for _, name := range slices.Sorted(mapsKeys(entries)) {
+		for _, s := range entries[name].Serves {
+			if strings.HasPrefix(s, ne.Seat+".") {
+				return entries[name], true
+			}
+		}
+	}
+	return meshcatalog.CatalogEntry{}, false
+}
+
+func mapsKeys(m map[string]meshcatalog.CatalogEntry) func(func(string) bool) {
+	return func(yield func(string) bool) {
+		for k := range m {
+			if !yield(k) {
+				return
+			}
+		}
+	}
+}
+
+// programOfEntry is the composer built from a node-engine entry: its repository, commit and directory, the composer's package.
+func programOfEntry(e meshcatalog.CatalogEntry) (buildstep.RunProgram, error) {
+	dir := e.Path
+	if dir == "" {
+		dir = "."
+	}
+	if e.Repository == "" || e.Commit == "" {
+		return buildstep.RunProgram{}, fmt.Errorf("the node-engine's entry %s names no repository or commit", e.Fp)
+	}
+	return buildstep.RunProgram{Repository: e.Repository, Commit: e.Commit, Dir: dir, Package: composer.Package}, nil
+}
+
+// programs answers, per machine, the composer its run builds (A-226, A-243, A-252): where the revision's node-engine entry differs
+// from the trunk's, that build for every machine; else the build the machine's node-engine runs, as its `running` says, its entry
+// read through mesh-catalog's `entry`.
+func (r *compositionRun) programs(half mcc.Half) (func(node string) (buildstep.RunProgram, error), error) {
+	engine, ok := nodeEngineEntry(half.Entries)
+	if !ok {
+		return nil, fmt.Errorf("the catalog revision %d holds no module claiming the seat %s", half.Revision.Revision, ne.Seat)
+	}
+	if trunk, ok := r.trunk(); !ok || func() bool { t, ok := nodeEngineEntry(trunk.Entries); return !ok || t.Fp != engine.Fp }() {
+		p, err := programOfEntry(engine)
+		return func(string) (buildstep.RunProgram, error) { return p, err }, nil
+	}
+	var mu sync.Mutex
+	byFp := map[string]buildstep.RunProgram{}
+	return func(node string) (buildstep.RunProgram, error) {
+		running, ok := r.running(node, engine.Module)
+		if !ok || running.Fingerprint == "" {
+			return buildstep.RunProgram{}, fmt.Errorf("no build of %s is heard running there", engine.Module)
+		}
+		mu.Lock()
+		p, ok := byFp[running.Fingerprint]
+		mu.Unlock()
+		if ok {
+			return p, nil
+		}
+		raw, err := r.call(entryVerbKey, meshcatalog.EntryInput{Fingerprint: running.Fingerprint})
+		if err != nil {
+			return buildstep.RunProgram{}, fmt.Errorf("the entry of the node-engine's build %s is not read: %w", running.Fingerprint, err)
+		}
+		body, said, ok := buscalls.Unwrap(raw)
+		if !ok {
+			return buildstep.RunProgram{}, fmt.Errorf("the entry of the node-engine's build %s is not read: %s", running.Fingerprint, said)
+		}
+		var a meshcatalog.EntryAnswer
+		if err := json.Unmarshal(body, &a); err != nil {
+			return buildstep.RunProgram{}, fmt.Errorf("the entry of the node-engine's build %s is not read: %w", running.Fingerprint, err)
+		}
+		p, err = programOfEntry(a.Entry)
+		if err != nil {
+			return p, err
+		}
+		mu.Lock()
+		byFp[running.Fingerprint] = p
+		mu.Unlock()
+		return p, nil
+	}, nil
+}
+
+// heardPresence keeps a machine-away (away true) or machine-back of the seat mesh-watchdogs (§389, §436): the machine it names.
+func heardPresence(a *awayNodes, away bool, m jetstream.Msg, logf func(string, ...any)) {
+	var e struct {
+		Machine string `json:"machine"`
+	}
+	if err := json.Unmarshal(m.Data(), &e); err != nil || e.Machine == "" {
+		logf("%s holds no machine, left: %v", m.Subject(), err)
+	} else {
+		a.set(e.Machine, away)
+	}
+	if err := m.Ack(); err != nil {
+		logf("acknowledging %s: %v", m.Subject(), err)
+	}
+}
+
+// trunkOf is the trunk's catalog half the assigner's views hold, or false while none is held.
+func trunkOf(iv *catalogViews) func() (mcc.Half, bool) {
+	return func() (mcc.Half, bool) {
+		if iv == nil || iv.h == nil {
+			return mcc.Half{}, false
+		}
+		h, ok := iv.h.views()[trunkBranch]
+		return h, ok
+	}
+}
+
+// runAskPublisher puts a run ask on the Go build seat's check queue under the assigner's name (`accept.check.assigner`, §600), its
+// id the message id, so the stream takes it once.
+func runAskPublisher(js jetstream.JetStream) func(ctx context.Context, a buildstep.StepAsk) error {
+	return func(ctx context.Context, a buildstep.StepAsk) error {
+		raw, err := buildstep.EncodeStepAsk(a)
+		if err != nil {
+			return err
+		}
+		subject, err := buildstep.AskSubjectBy(a.Language, a.Kind, a.Asker)
+		if err != nil {
+			return err
+		}
+		_, err = js.Publish(ctx, subject, raw, jetstream.WithMsgID(a.ID))
+		return err
+	}
+}

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

@@ -0,0 +1,176 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	ne "git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
+	mcc "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient"
+	meshcatalog "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
+	meshwants "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants"
+	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildstep"
+	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/composer"
+)
+
+const (
+	commitE1 = "1111111111111111111111111111111111111111"
+	commitE2 = "2222222222222222222222222222222222222222"
+	commitE3 = "3333333333333333333333333333333333333333"
+)
+
+// runStand is a composition run over stand-ins of the mesh-catalog seat, the machines' engines, the build seat and its step-ended.
+type runStand struct {
+	mu       sync.Mutex
+	asks     map[string]buildstep.StepAsk // by node
+	requests map[string]composer.Request  // by node
+	entries  []string                     // the fingerprints `entry` was asked
+	r        *compositionRun
+}
+
+func newRunStand(t *testing.T, revisionEngine, trunkEngine, seatDigest string) *runStand {
+	s := &runStand{asks: map[string]buildstep.StepAsk{}, requests: map[string]composer.Request{}}
+	engine := func(fp, commit string) meshcatalog.CatalogEntry {
+		return meshcatalog.CatalogEntry{Module: "engine", Fp: fp, Repository: "novox/mesh-host", Commit: commit, Path: ".", Serves: []string{ne.Seat + ".composition"},
+			Manifest: map[string]any{"module": "engine"}}
+	}
+	revisionCommit := commitE1
+	if revisionEngine != "src2:e1" {
+		revisionCommit = commitE3
+	}
+	pages := map[string]meshcatalog.RevisionAnswer{
+		"": {Revision: &meshcatalog.CatalogRevision{Revision: 7, Branch: "main", Definitions: []meshcatalog.CatalogRevisionDefinitionsItem{{Seat: "s1", Digest: "d1"}}},
+			Entries: []meshcatalog.CatalogEntry{engine(revisionEngine, revisionCommit), {Module: "bad", Verdict: mcc.VerdictRefused, Inputs: &meshcatalog.EntryInputs{Problems: []string{"no seat x"}},
+				Manifest: map[string]any{"module": "bad"}}}, Next: "bad"},
+		"bad": {Revision: &meshcatalog.CatalogRevision{Revision: 7}, Entries: []meshcatalog.CatalogEntry{{Module: "m", Verdict: mcc.VerdictAccepted, Manifest: map[string]any{"module": "m"}}}},
+	}
+	call := func(key string, body any) (json.RawMessage, error) {
+		raw, _ := json.Marshal(body)
+		switch key {
+		case mcc.RevisionVerb:
+			var in meshcatalog.RevisionInput
+			_ = json.Unmarshal(raw, &in)
+			if in.Revision != 7 || in.Branch != "main" {
+				return nil, errors.New("asked another revision")
+			}
+			return json.Marshal(pages[in.After])
+		case seatDefinitionsKey:
+			return json.Marshal(meshcatalog.SeatDefinitionsAnswer{Found: true, Entry: &meshcatalog.SeatDefinitionsAnswerEntry{Seat: "s1", Digest: seatDigest, Definition: map[string]any{"seat": "s1"}}})
+		case entryVerbKey:
+			var in meshcatalog.EntryInput
+			_ = json.Unmarshal(raw, &in)
+			s.mu.Lock()
+			s.entries = append(s.entries, in.Fingerprint)
+			s.mu.Unlock()
+			commit := map[string]string{"src2:e1": commitE1, "src2:e2": commitE2}[in.Fingerprint]
+			return json.Marshal(meshcatalog.EntryAnswer{Entry: engine(in.Fingerprint, commit)})
+		}
+		return nil, errors.New("no such verb " + key)
+	}
+	engines := func(node, _ string, _ any) (json.RawMessage, error) {
+		return json.RawMessage(`{"candidate":"main.7","node":"` + node + `","verdict":"composed","input":{"node":"` + node + `"},"modules":[],"own":{},"took":1,"v":1}`), nil
+	}
+	running := map[string]string{"ace": "src2:e1", "g14": "src2:e2"}
+	away := &awayNodes{}
+	away.set("shanks", true)
+	keeper := testKeeper(&objectsMem{objs: map[string][]byte{}}, time.Now)
+	ends := &runEnds{}
+	s.r = &compositionRun{
+		inputs: &compositionInputs{machines: func(string, string) []string { return []string{"ace", "g14", "shanks"} }, call: engines, bound: time.Second},
+		keeper: keeper, call: call, ends: ends, away: away, now: time.Now, logf: t.Logf,
+		trunk: func() (mcc.Half, bool) {
+			return mcc.Half{Entries: map[string]meshcatalog.CatalogEntry{"engine": engine(trunkEngine, commitE1)}}, true
+		},
+		running: func(node, module string) (ne.Running, bool) {
+			fp, ok := running[node]
+			return ne.Running{Node: node, Module: module, Fingerprint: fp}, ok && module == "engine"
+		},
+		put: func(_ context.Context, a buildstep.StepAsk) error {
+			body, err := keeper.body(a.Run.Input.Digest)
+			if err != nil {
+				return err
+			}
+			req, err := composer.Decode(body)
+			if err != nil {
+				return err
+			}
+			node := req.Machines[0]
+			s.mu.Lock()
+			s.asks[node], s.requests[node] = a, req
+			s.mu.Unlock()
+			m := composer.Machine{Composes: true}
+			if node == "g14" {
+				m = composer.Machine{Reason: "two modules own one path", Problems: []string{"/etc/x is m's and bad's"}}
+			}
+			v, _ := json.Marshal(composer.Verdict{V: composer.Version, Cases: []composer.CaseVerdict{{Name: changeCase, Machines: map[string]composer.Machine{node: m}}}})
+			go ends.heard(buildstep.StepEnded{Ask: a.ID, Asker: runAsker, Node: "builder1", Outcome: buildstep.OutcomePassed, Output: string(v)})
+			return nil
+		},
+	}
+	return s
+}
+
+// Design 47 G §354, §63, §64, A-226, A-238, A-240, A-243, A-252, A-254, A-261: one request and one run ask per machine, the request
+// depth compose with the one machine and one case `change`, its manifests, verdicts and seats from the step's revision; the
+// composer is the node-engine build each machine runs, its entry read through `entry`, or the revision's where it differs from the
+// trunk's; the run's verdict is read from its step-ended; a machine away passes with a warning, one not composing fails the step.
+func TestTheCompositionStepRunsOneAskPerMachine(t *testing.T) {
+	step := meshwants.StepRun{Module: "m", Branch: "main", Revision: 7, Step: meshwants.StepRunStepComposition}
+	s := newRunStand(t, "src2:e1", "src2:e1", "d1")
+	out := s.r.run(context.Background(), "ask-step", step)
+	if out.Outcome != meshwants.StepEndedOutcomeFailed || !strings.Contains(out.Said, "g14 does not compose") || !strings.Contains(out.Said, "/etc/x") ||
+		strings.Contains(out.Said, "ace") || len(out.Warnings) != 1 || !strings.Contains(out.Warnings[0], "shanks is away") {
+		t.Fatalf("the step came to %+v", out)
+	}
+	if len(s.asks) != 2 {
+		t.Fatalf("asked %d runs, want one per machine not away", len(s.asks))
+	}
+	a, req := s.asks["ace"], s.requests["ace"]
+	if a.Kind != buildstep.AskCheck || a.Language != buildstep.AskGo || a.Asker != runAsker || a.Run == nil || len(a.Run.Heads) != 0 ||
+		a.Run.Input.Verb != "seat:mesh-wants.input" || a.Run.Program != (buildstep.RunProgram{Repository: "novox/mesh-host", Commit: commitE1, Dir: ".", Package: composer.Package}) {
+		t.Errorf("the run ask of ace %+v %+v", a, a.Run)
+	}
+	if s.asks["g14"].Run.Program.Commit != commitE2 {
+		t.Errorf("g14's composer is built at %s, want the build its node-engine runs", s.asks["g14"].Run.Program.Commit)
+	}
+	if req.Depth != composer.DepthCompose || len(req.Machines) != 1 || req.Machines[0] != "ace" || len(req.Cases) != 1 || req.Cases[0].Name != changeCase ||
+		string(req.Input) != `{"node":"ace"}` || len(req.Seats) != 1 || string(req.Seats[0]) != `{"seat":"s1"}` {
+		t.Errorf("the request of ace %+v", req)
+	}
+	c := req.Cases[0]
+	if string(c.Manifests["m"].Raw) != `{"module":"m"}` || len(c.Manifests) != 3 || len(c.Verdicts) != 1 || !strings.Contains(string(c.Verdicts["bad"]), `"refused"`) ||
+		!strings.Contains(string(c.Verdicts["bad"]), "no seat x") {
+		t.Errorf("the case %+v %s", c.Manifests, c.Verdicts)
+	}
+	if strings.Join(s.entries, " ") == "" {
+		t.Error("no node-engine entry is read through entry")
+	}
+
+	// The revision's node-engine differs from the trunk's: that build composes every machine, and no entry is asked.
+	changed := newRunStand(t, "src2:e3", "src2:e1", "d1")
+	if out := changed.r.run(context.Background(), "ask-step", step); out.Outcome != meshwants.StepEndedOutcomeFailed {
+		t.Fatalf("the step came to %+v", out)
+	}
+	if changed.asks["ace"].Run.Program.Commit != commitE3 || changed.asks["g14"].Run.Program.Commit != commitE3 || len(changed.entries) != 0 {
+		t.Errorf("the change's node-engine does not compose every machine: %+v %+v %v", changed.asks["ace"].Run.Program, changed.asks["g14"].Run.Program, changed.entries)
+	}
+
+	// A seat digest neither the revision nor the trunk holds: could not run, naming the seat and the digest (A-261).
+	other := newRunStand(t, "src2:e1", "src2:e1", "d2")
+	if out := other.r.run(context.Background(), "ask-step", step); out.Outcome != meshwants.StepEndedOutcomeCouldNotRun ||
+		!strings.Contains(out.Said, "s1") || !strings.Contains(out.Said, "d1") || len(other.asks) != 0 {
+		t.Errorf("a seat digest not held came to %+v, %d asks", out, len(other.asks))
+	}
+
+	// At the inputs state's cap the ask is refused, naming the store (§354).
+	full := newRunStand(t, "src2:e1", "src2:e1", "d1")
+	full.r.keeper.used = func() (int64, error) { return inputsCap, nil }
+	if out := full.r.run(context.Background(), "ask-step", step); out.Outcome != meshwants.StepEndedOutcomeCouldNotRun ||
+		!strings.Contains(out.Said, "the state inputs is at its cap") || len(full.asks) != 0 {
+		t.Errorf("an ask at the cap came to %+v, %d asks", out, len(full.asks))
+	}
+}

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

@@ -8,6 +8,7 @@ import (
 	"github.com/nats-io/nats.go"
 
 	meshwants "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants"
+	"git.novox.be/novox/mesh-sdk/go/process"
 )
 
 // The composition step (design 47 G §22, §354, §63): the assigner takes the seat mesh-wants' accepted kind `step`, the
@@ -66,7 +67,7 @@ func stepAskID(m *nats.Msg) string {
 }
 
 // serveCompositionSteps hears the composition step asks on the seat mesh-wants' accept subject.
-func serveCompositionSteps(ctx context.Context, conn *nats.Conn, c *compositionSteps) error {
+func serveCompositionSteps(ctx context.Context, conn process.Subscriber, c *compositionSteps) error {
 	sub, err := conn.Subscribe(meshwants.AcceptSubjectStep, func(m *nats.Msg) { c.take(ctx, stepAskID(m), m.Data) })
 	if err != nil {
 		return err

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

@@ -17,7 +17,7 @@ import (
 	"github.com/nats-io/nats.go/jetstream"
 
 	meshwants "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants"
-	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/nodebuildgo"
+	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildstep"
 	stdio "git.novox.be/novox/mesh-sdk/go"
 	"git.novox.be/novox/mesh-sdk/go/buscalls"
 	"git.novox.be/novox/mesh-sdk/go/catalogue"
@@ -51,11 +51,16 @@ const (
 // digestOf is an input's digest as the ask names it: its sha256 in hex.
 func digestOf(b []byte) string { sum := sha256.Sum256(b); return hex.EncodeToString(sum[:]) }
 
+// inputsCap is the state inputs' cap (design 47 G §354's kept-data row: 64 MiB, the manifest's max-mib): at it, an ask is refused.
+const inputsCap = 64 << 20
+
 // inputKeeper keeps the inputs over the state and its object store.
 type inputKeeper struct {
 	vals stateobjects.Values
 	now  func() time.Time
 	logf func(string, ...any)
+	// used is how many bytes the state and its object store hold; nil judges no cap.
+	used func() (int64, error)
 }
 
 // keep keeps the exact bytes of the input of the ask under their write time and sha256, after sweeping the keys more than a day old
@@ -67,6 +72,16 @@ func (k *inputKeeper) keep(ask string, body []byte) (string, error) {
 	if err := k.sweep(); err != nil {
 		k.logf("inputs: the keys older than a day are not all deleted: %v", err)
 	}
+	if k.used != nil {
+		used, err := k.used()
+		if err != nil {
+			return "", fmt.Errorf("the fill of the state %s is not read, so no input is kept: %w", meshwants.StateInputs, err)
+		}
+		if used+int64(len(body)) > inputsCap {
+			return "", fmt.Errorf("the state %s is at its cap of %d MiB (%d bytes kept), so the ask's input of %d bytes is refused",
+				meshwants.StateInputs, inputsCap>>20, used, len(body))
+		}
+	}
 	digest := digestOf(body)
 	key := meshwants.InputsByTimeDigestKey(k.now().UTC().Format(inputsTime), digest)
 	if err := k.vals.Put(key, base64.StdEncoding.EncodeToString(body)); err != nil && !errors.Is(err, stateobjects.ErrChanged) {
@@ -276,8 +291,26 @@ func inputsOver(kv jetstream.KeyValue, js jetstream.JetStream, module string, lo
 	if kv == nil {
 		return nil
 	}
-	v := stateobjects.Values{KV: stateobjects.BusKV{KV: kv}, Objects: &inputObjects{js: js, module: module}, Store: catalogue.StateObjectsName(meshwants.StateInputs)}
-	return &inputKeeper{vals: v, now: time.Now, logf: logf}
+	objects := &inputObjects{js: js, module: module}
+	v := stateobjects.Values{KV: stateobjects.BusKV{KV: kv}, Objects: objects, Store: catalogue.StateObjectsName(meshwants.StateInputs)}
+	used := func() (int64, error) {
+		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+		defer cancel()
+		st, err := kv.Status(ctx)
+		if err != nil {
+			return 0, err
+		}
+		n := int64(st.Bytes())
+		objs, err := objects.List()
+		if err != nil && !errors.Is(err, process.ErrNotYet) {
+			return 0, err
+		}
+		for _, o := range objs {
+			n += o.Size
+		}
+		return n, nil
+	}
+	return &inputKeeper{vals: v, now: time.Now, logf: logf, used: used}
 }
 
 // inputSaid is one kept input as assigner_inputs lists it.
@@ -323,10 +356,10 @@ func inputsRead(k *inputKeeper) stdio.Tool {
 		}}
 }
 
-// heardStepEnded deletes the input kept for the ask a build seat's step-ended names (design 47 G §354); one that cannot be deleted is
-// given again by the bus.
-func heardStepEnded(k *inputKeeper, m jetstream.Msg, logf func(string, ...any)) {
-	var e nodebuildgo.StepEnded
+// heardStepEnded deletes the input kept for the ask a build seat's step-ended names (design 47 G §354), and hands the end to the
+// composition run that waits for that ask (A-240); one whose input cannot be deleted is given again by the bus.
+func heardStepEnded(k *inputKeeper, ends *runEnds, m jetstream.Msg, logf func(string, ...any)) {
+	var e buildstep.StepEnded
 	if err := json.Unmarshal(m.Data(), &e); err != nil {
 		logf("%s holds what is no step-ended, left: %v", m.Subject(), err)
 		_ = m.Ack()
@@ -337,5 +370,8 @@ func heardStepEnded(k *inputKeeper, m jetstream.Msg, logf func(string, ...any))
 		_ = m.Nak()
 		return
 	}
+	if ends != nil {
+		ends.heard(e)
+	}
 	_ = m.Ack()
 }

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

@@ -30,6 +30,7 @@ import (
 
 	mma "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner"
 	mcseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
+	"git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshwatchdogs"
 	occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient"
 	oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
 	"git.novox.be/novox/mesh-sdk/go/announce"
@@ -491,12 +492,20 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
 	if bus != nil {
 		gate.Follow(seatlist.ProcessBus(bus.Call, hub), wake, logger.Printf)
 	}
-	// The build seat's step-ended of an ask deletes the input kept for it (design 47 G §354); registered before the hub starts.
+	// The build seat's step-ended of an ask deletes the input kept for it (design 47 G §354) and ends the composition run that waits
+	// for it (A-240); the watchman's machine-away and machine-back say which machines a composition step does not compose (§64,
+	// §389). Registered before the hub starts.
+	ends, away := &runEnds{}, &awayNodes{}
 	if inputs != nil {
-		if err := hub.Handle(nodebuildgo.SubjectStepEndedAll, func(m jetstream.Msg) { heardStepEnded(inputs, m, logger.Printf) }); err != nil {
+		if err := hub.Handle(nodebuildgo.SubjectStepEndedAll, func(m jetstream.Msg) { heardStepEnded(inputs, ends, m, logger.Printf) }); err != nil {
 			return err
 		}
 	}
+	for subject, isAway := range map[string]bool{meshwatchdogs.SubjectMachineAwayAll: true, meshwatchdogs.SubjectMachineBackAll: false} {
+		if err := hub.Handle(subject, func(m jetstream.Msg) { heardPresence(away, isAway, m, logger.Printf) }); err != nil {
+			logger.Printf("%s is not heard, so a composition step asks a machine that sleeps: %v", subject, err)
+		}
+	}
 	// The indexer's manifests, modules and module graph, through its first views and views family, never its buckets
 	// (DLV-F2r, ADR 0631): the first pass waits for the three answers, as it waits for every watch to hand what stands.
 	if iv != nil {
@@ -628,6 +637,15 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
 	if bus != nil && kvs[buckets.Puts] != nil {
 		p.acts = newPutActs(bus, wake, logger.Printf)
 	}
+	// The composition step (design 47 G §22, §354): the seat mesh-wants' accepted kind step, run by one run ask per machine.
+	if bus != nil && js != nil && inputs != nil && p.publishWants != nil {
+		run := &compositionRun{inputs: &compositionInputs{machines: w.machinesOf, call: bus.CallKeptOn}, keeper: inputs, call: bus.CallKept,
+			trunk: trunkOf(iv), running: w.runningOf, away: away, put: runAskPublisher(js), ends: ends, ask: bus.Call, now: time.Now, logf: logger.Printf}
+		steps := &compositionSteps{node: thisNode, say: p.publishWants, run: run.run, now: time.Now, logf: logger.Printf}
+		if err := serveCompositionSteps(ctx, nc, steps); err != nil {
+			logger.Printf("the composition step is not heard, so no composition step runs until the assigner starts again: %v", err)
+		}
+	}
 	if bus != nil {
 		p.seat = newSeatActs(bus, wake, logger.Printf)
 	}

modules/mesh-delivery-assigner/manifest_test.go

@@ -130,14 +130,18 @@ func TestManifest(t *testing.T) {
 	}
 	// ADR 0618 §4: and it hears each node's engine say its reads changed; ADR 0636 §4 (issue 1966): and the seat list's two events;
 	// issue 1966 note 151: and the teller's ask-ended, its subject the one the assigner hears.
-	if !slices.Equal(m.Consumes, []string{"mesh-conditions.findings-asked", "mesh-conditions.condition-raised", "mesh-conditions.condition-changed", "mesh-conditions.condition-cleared", "mesh-catalog.catalog-revision", "mesh-catalog.registration-changed", "mesh-catalog.version-released", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", builds.BuilderModule + "." + builds.BuildEndedEvent, buckets.TellerModule + "." + tellerevents.AskEnded, "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.group-ended", "node-build-go.step-ended", "node-engine.manifest-reads-changed", "node-engine.running", "mesh-delivery.pinned", "mesh-delivery.freed", "mesh-delivery.heads-replaced", "mesh-delivery.pins-stated"}) {
+	if !slices.Equal(m.Consumes, []string{"mesh-conditions.findings-asked", "mesh-conditions.condition-raised", "mesh-conditions.condition-changed", "mesh-conditions.condition-cleared", "mesh-catalog.catalog-revision", "mesh-catalog.registration-changed", "mesh-catalog.version-released", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", builds.BuilderModule + "." + builds.BuildEndedEvent, buckets.TellerModule + "." + tellerevents.AskEnded, "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.group-ended", "node-build-go.step-ended", "node-engine.manifest-reads-changed", "node-engine.running", "mesh-delivery.pinned", "mesh-delivery.freed", "mesh-delivery.heads-replaced", "mesh-delivery.pins-stated",
+		"mesh-watchdogs.machine-away", "mesh-watchdogs.machine-back"}) { // design 47 G §64, §389: a composition step passes a machine away
 		t.Errorf("consumes = %v, want the catalog half's revisions (design 47 G §195), the teller's ask-ended, the engines' reads changed and what they run", m.Consumes)
 	}
 	if want := []string{"mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules",
+		"seat:mesh-catalog.entry",                                                                             // design 47 G A-243, A-252: the node-engine build's entry a composition run builds
 		"seat:mesh-catalog.registrations", "seat:mesh-catalog.revision", "seat:mesh-catalog.seat-definitions", // ADR 0636 §4: the seat list
 		"seat:mesh-conditions.conditions",                                                                // design 47 G §706: the conditions read at each judging of an outside wait
 		"seat:mesh-hand-acts.hand-act", "seat:mesh-hand-acts.hand-acts", "seat:mesh-hand-acts.warranted", // ADR 0565 §9, §11: the seat alone
 		"seat:node-backup.backed-up", "seat:node-backup.now",
+		"seat:node-build-go.output",                                                                  // A-254: a composition run's answer kept as the build seat's object
+		"seat:node-engine.compose-candidate",                                                         // §354, A-226: each machine's composition input
 		"seat:node-engine.manifest-reads",                                                            // ADR 0618 §3
 		"seat:operator-channel.asks"}; !slices.Equal(slices.Sorted(slices.Values(m.Invokes)), want) { // ADR 0668 §4: the ended asks at bind
 		t.Errorf("invokes = %v, want %v", m.Invokes, want)

modules/mesh-delivery-assigner/module.json

@@ -63,11 +63,14 @@
     "mesh-delivery.pinned",
     "mesh-delivery.freed",
     "mesh-delivery.heads-replaced",
-    "mesh-delivery.pins-stated"
+    "mesh-delivery.pins-stated",
+    "mesh-watchdogs.machine-away",
+    "mesh-watchdogs.machine-back"
   ],
   "invokes": [
     "mesh-delivery-indexer.manifests",
     "mesh-delivery-indexer.modules",
+    "seat:mesh-catalog.entry",
     "seat:mesh-catalog.registrations",
     {"verb": "seat:mesh-catalog.revision", "fields": ["none-yet"]},
     "seat:mesh-catalog.seat-definitions",
@@ -77,6 +80,8 @@
     "seat:mesh-hand-acts.warranted",
     "seat:node-backup.backed-up",
     "seat:node-backup.now",
+    "seat:node-build-go.output",
+    "seat:node-engine.compose-candidate",
     "seat:node-engine.manifest-reads",
     "seat:operator-channel.asks"
   ],