RN3 (asking half): the runner places a pipeline's steps from the policy and asks each build seat for its step

A step the module's policy does not name is no step (A-91); a bypassed step is asked of no one; an artifact no
build seat names ends its step could not run naming it; an approval due at the start is asked there. The hearer is
named by the runner's own module, the asker the router's decided names.

Builds: design 47 G §22, §24, §26, §44, §54, §103, §323, §476, A-91

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

modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/asks.go

@@ -0,0 +1,42 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"time"
+
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+
+	nodebuildgo "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo"
+	"git.novox.be/novox/mesh-sdk/go/process"
+
+	"mesh-delivery-pipeline/internal/pipeline"
+)
+
+// busStepAsker puts a step's ask on a build seat's work queue at the seat's accept subject, under the runner's own name; its
+// duplicate id is the ask's own, so an ask put twice is kept once (design 47 G §22, §24).
+type busStepAsker struct {
+	ctx  context.Context
+	js   jetstream.JetStream
+	cred process.Account
+}
+
+func (b busStepAsker) AskStep(accept string, a nodebuildgo.StepAsk) error {
+	data, err := json.Marshal(a)
+	if err != nil {
+		return err
+	}
+	msg := nats.NewMsg(accept)
+	msg.Header.Set("x-event-id", a.ID)
+	msg.Header.Set("x-source", b.cred.Module)
+	msg.Header.Set("x-node", b.cred.Node)
+	msg.Header.Set("x-time", time.Now().UTC().Format(time.RFC3339Nano))
+	msg.Header.Set("content-type", "application/json")
+	msg.Header.Set(nats.MsgIdHdr, a.ID)
+	msg.Data = data
+	_, err = b.js.PublishMsg(b.ctx, msg)
+	return err
+}
+
+var _ pipeline.StepAsker = busStepAsker{}

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

@@ -73,10 +73,13 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		}
 		return err
 	}
-	starter := &pipeline.Starter{Pipelines: createdKV{ctx: ctx, kv: pipelines}, Logf: logger.Printf, Say: say}
 	hearer := newHearer(ctx, js, cred, pipelines, kept, say, logger.Printf)
 	hearer.Asker = busAsker{conn: conn, cred: cred}
 	hearer.Where = routerRecordOf(cred.Module, calls.Call, calls.Ask)
+	// Its asking half (§22, §24): a pipeline's steps are placed from the module's registration and each build seat is asked
+	// for its step, an approval due at the start asked as the hearer asks it.
+	starter := &pipeline.Starter{Pipelines: createdKV{ctx: ctx, kv: pipelines}, Logf: logger.Printf, Say: say,
+		Registration: rv.Registrations.Registration, Steps: busStepAsker{ctx: ctx, js: js, cred: cred}, Name: cred.Module, After: hearer}
 	if err := hearStepEnded(hub, hearer, logger.Printf); err != nil {
 		return err
 	}

modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/router_record_test.go

@@ -1,11 +1,13 @@
 package main
 
 import (
+	"context"
 	"encoding/json"
 	"strings"
 	"testing"
 
 	stdio "git.novox.be/novox/mesh-sdk/go"
+	"git.novox.be/novox/mesh-sdk/go/process"
 )
 
 // Design 47 G §700, A-74: the router's record of an ask is read under the indexer's module's name from the state asks of the
@@ -35,3 +37,12 @@ func TestTheRoutersRecordIsReadFromTheBucketOfTheSeatsHolder(t *testing.T) {
 		t.Errorf("with no holder Where answered %v; want the reason", err)
 	}
 }
+
+// The asker the router and the decided event name is the runner's own module (its bus account's), never a constant: a hearer
+// that named another refused every decided of its own asks.
+func TestTheHearerIsNamedByTheRunnersOwnModule(t *testing.T) {
+	h := newHearer(context.Background(), nil, process.Account{Module: "the-runner"}, nil, nil, nil, nil)
+	if h.Name != "the-runner" {
+		t.Errorf("the hearer's name is %q, want its account's module", h.Name)
+	}
+}

modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/steps.go

@@ -74,7 +74,7 @@ func (s pipelineStore) Keys(filter string) ([]string, error) {
 // objects (§27). Each store is opened where it is used, so one the grants part has not made yet fails that write loudly.
 func newHearer(ctx context.Context, js jetstream.JetStream, cred process.Account, pipelines, kept jetstream.KeyValue,
 	say func(string, any) error, logf func(string, ...any)) *pipeline.Hearer {
-	return &pipeline.Hearer{Store: pipelineStore{ctx: ctx, kv: pipelines}, Say: say, Logf: logf, Name: seat.Seat,
+	return &pipeline.Hearer{Store: pipelineStore{ctx: ctx, kv: pipelines}, Say: say, Logf: logf, Name: cred.Module,
 		Log: func(e seat.LogEntry) error {
 			return say(cred.EventSubject("step-log."+keys.EscapeName(e.Pipeline)), e)
 		},

modules/mesh-delivery-pipeline/internal/pipeline/place.go

@@ -0,0 +1,132 @@
+package pipeline
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+	"fmt"
+	"sort"
+	"strings"
+	"time"
+
+	catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
+	seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner"
+	nodebuildgo "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo"
+	nodebuildimage "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildimage"
+	nodebuildnode "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildnode"
+)
+
+// StepAsker puts one step's ask on a build seat's work queue, at the subject accept names (the seat's accepted kind `step`
+// on its stream `check`, design 47 G §22, §24). An ask asked twice under one id is kept once.
+type StepAsker interface {
+	AskStep(accept string, a nodebuildgo.StepAsk) error
+}
+
+// acceptStep is where each build seat is sent its step, from its contract.
+var acceptStep = map[string]string{
+	nodebuildgo.Seat:    nodebuildgo.AcceptSubjectCheckStep,
+	nodebuildnode.Seat:  nodebuildnode.AcceptSubjectCheckStep,
+	nodebuildimage.Seat: nodebuildimage.AcceptSubjectCheckStep,
+}
+
+// buildSteps are the steps an artifact's build seat runs, in the order they are asked (design 47 G §22).
+var buildSteps = []string{seat.StepResultStepLint, seat.StepResultStepUnit, seat.StepResultStepIntegration}
+
+// moduleSteps are the steps of the whole module: composition (the assigner's), review (the runner's own verb) and the
+// approval, asked last (§22, §26).
+var moduleSteps = []string{seat.StepResultStepComposition, seat.StepResultStepReview, seat.StepResultStepApproval}
+
+// kindImage is the form of an acting rule's `on` naming an artifact kind (§324).
+const kindImage = "kind: image"
+
+// AskID is a step's ask id, the same for the same pipeline, artifact and step, so a start asked again asks nothing new.
+func AskID(pipeline, artifact, step string) string {
+	sum := sha256.Sum256([]byte(pipeline + "\n" + artifact + "\n" + step))
+	return "step-" + hex.EncodeToString(sum[:12])
+}
+
+// Place places a pipeline's steps (design 47 G §21, §23, §323; A-91): only the steps the module's registration policy
+// names, each artifact's lint, unit and integration steps beside the module's composition, review and approval. A step the
+// policy does not name is no step. A `bypass` step is placed bypassed and asked of no one; a build step is placed running,
+// asked of the build seat the artifact's language names in the catalog half (§323), or, where no seat names it, ended
+// could not run naming the missing seat; the module's steps wait. It returns the steps and the asks to put on the seats'
+// work queues, one per running step.
+func Place(p seat.Pipeline, reg catalog.Registration, entries map[string]catalog.CatalogEntry, now time.Time, asker string) ([]seat.StepResult, []Asked) {
+	entry := entries[p.Module]
+	arts := append([]catalog.Artifact(nil), entry.Artifacts...)
+	sort.Slice(arts, func(i, j int) bool { return arts[i].Ref < arts[j].Ref })
+	var steps []seat.StepResult
+	var asks []Asked
+	for _, a := range arts {
+		name := a.Ref
+		if _, after, ok := strings.Cut(a.Ref, ":"); ok {
+			name = after
+		}
+		for _, step := range buildSteps {
+			status := Status(reg, name, step)
+			if status == "" {
+				continue
+			}
+			s := seat.StepResult{Step: step, Artifact: a.Ref, Status: status, Fingerprint: p.Fingerprint}
+			if status == seat.StepResultStatusBypass {
+				s.Outcome, s.Why = seat.StepResultOutcomeBypassed, "bypassed by the module's policy"
+				steps = append(steps, s)
+				continue
+			}
+			on := a.Language
+			if on == "" && a.Kind == "image" {
+				on = kindImage
+			}
+			bs, err := BuildSeat(entries, on)
+			if err != nil {
+				s.Outcome, s.Started, s.Ended, s.Why = seat.StepResultOutcomeCouldNotRun, &now, &now, err.Error()
+				steps = append(steps, s)
+				continue
+			}
+			s.Seat, s.Outcome, s.Started = bs, seat.StepResultOutcomeRunning, &now
+			s.Ask = AskID(p.Pipeline, a.Ref, step)
+			steps = append(steps, s)
+			asks = append(asks, Asked{Seat: bs, Ask: stepAsk(p, entry, a, step, s.Ask, asker, now)})
+		}
+	}
+	for _, step := range moduleSteps {
+		status := Status(reg, "", step)
+		if status == "" {
+			continue
+		}
+		s := seat.StepResult{Step: step, Status: status, Outcome: seat.StepResultOutcomeWaiting}
+		if status == seat.StepResultStatusBypass {
+			s.Outcome, s.Why = seat.StepResultOutcomeBypassed, "bypassed by the module's policy"
+		}
+		steps = append(steps, s)
+	}
+	return steps, asks
+}
+
+// Asked is one ask to put on a build seat's work queue.
+type Asked struct {
+	Seat string
+	Ask  nodebuildgo.StepAsk
+}
+
+func stepAsk(p seat.Pipeline, e catalog.CatalogEntry, a catalog.Artifact, step, id, asker string, now time.Time) nodebuildgo.StepAsk {
+	fetch := nodebuildgo.Fetch{Repository: p.Repository, Commit: p.Commit, Dir: e.Path}
+	return nodebuildgo.StepAsk{V: 1, ID: id, Asker: asker, Kind: "step", Language: a.Language, At: &now,
+		Step: &nodebuildgo.StepRun{Artifact: a.Ref, Branch: p.Branch, Commit: p.Commit, Fetch: fetch, Fingerprint: p.Fingerprint,
+			Module: p.Module, Pipeline: p.Pipeline, Revision: p.Source, Step: step}}
+}
+
+// askAll puts each ask on its seat's work queue; the first refusal is named, the rest are still put.
+func askAll(asker StepAsker, key string, asks []Asked) error {
+	var first error
+	for _, a := range asks {
+		accept, ok := acceptStep[a.Seat]
+		err := fmt.Errorf("the seat %s is no build seat the runner asks", a.Seat)
+		if ok {
+			err = asker.AskStep(accept, a.Ask)
+		}
+		if err != nil && first == nil {
+			first = fmt.Errorf("the step %s of %s is not asked of %s: %w", a.Ask.Step.Step, key, a.Seat, err)
+		}
+	}
+	return first
+}

modules/mesh-delivery-pipeline/internal/pipeline/place_test.go

@@ -0,0 +1,117 @@
+package pipeline
+
+import (
+	"encoding/json"
+	"strings"
+	"testing"
+
+	catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
+	seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner"
+	nodebuildgo "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo"
+)
+
+type stepAsks struct {
+	at   []string
+	asks []nodebuildgo.StepAsk
+}
+
+func (s *stepAsks) AskStep(accept string, a nodebuildgo.StepAsk) error {
+	s.at, s.asks = append(s.at, accept), append(s.asks, a)
+	return nil
+}
+
+func placeFixture() (catalog.Registration, map[string]catalog.CatalogEntry) {
+	reg := catalog.Registration{Module: "app", Branches: []string{"main"}, Policy: catalog.Policy{
+		Steps:     catalog.StepStatuses{Lint: "required", Unit: "optional", Approval: "required"},
+		Artifacts: map[string]catalog.StepStatuses{"tools": {Lint: "bypass"}},
+	}}
+	return reg, map[string]catalog.CatalogEntry{
+		"app": {Module: "app", Path: "modules/app", Artifacts: []catalog.Artifact{{Ref: "app:program", Language: "go"}, {Ref: "app:tools", Language: "go"},
+			{Ref: "app:web", Language: "rust"}}},
+		"mesh-build-go": {Module: "mesh-build-go", Acts: []catalog.ActingRule{{Seat: "node-build-go", Step: "build", On: "go"}}},
+	}
+}
+
+// The runner's asking half (design 47 G §22, §24, §323; A-91): a pipeline's steps are the ones the policy names, a bypassed
+// step asked of no one, an artifact no build seat names ended could not run naming it, and each running step asked of its
+// build seat's accepted kind step with the pipeline's commit and fingerprint; a step the policy does not name is no step.
+func TestStartPlacesTheStepsThePolicyNamesAndAsksEachBuildSeat(t *testing.T) {
+	reg, entries := placeFixture()
+	store := created{}
+	asks := &stepAsks{}
+	s := &Starter{Pipelines: store, Say: func(string, any) error { return nil }, Steps: asks, Name: "the-runner",
+		Registration: func(m string) (catalog.Registration, bool) { return reg, m == "app" }}
+	if ok, err := s.Start(Begin{Branch: "main", Source: 4, Module: "app", Commit: "c1", Fp: "src2:aa", Repository: "o/app", Entries: entries}); !ok || err != nil {
+		t.Fatalf("start = %v, %v", ok, err)
+	}
+	var p seat.Pipeline
+	if err := json.Unmarshal(store["main.4.app"], &p); err != nil {
+		t.Fatal(err)
+	}
+	var got []string
+	for _, st := range p.Steps {
+		got = append(got, st.Artifact+" "+st.Step+" "+st.Outcome)
+	}
+	want := []string{"app:program lint running", "app:program unit running", "app:tools lint bypassed", "app:tools unit running",
+		"app:web lint could not run", "app:web unit could not run", " approval waiting"}
+	if strings.Join(got, "|") != strings.Join(want, "|") {
+		t.Errorf("steps = %q, want %q (integration and composition are named by no policy: no step)", got, want)
+	}
+	if !strings.Contains(p.Steps[4].Why, `"rust"`) {
+		t.Errorf("the step of an artifact no seat builds says %q, want it to name the language", p.Steps[4].Why)
+	}
+	if len(asks.asks) != 3 {
+		t.Fatalf("asked %d steps, want the three running ones", len(asks.asks))
+	}
+	a := asks.asks[0]
+	if asks.at[0] != nodebuildgo.AcceptSubjectCheckStep || a.Kind != "step" || a.Asker != "the-runner" || a.ID != p.Steps[0].Ask ||
+		a.Step.Commit != "c1" || a.Step.Fingerprint != "src2:aa" || a.Step.Pipeline != "main.4.app" || a.Step.Artifact != "app:program" ||
+		a.Step.Step != "lint" || a.Step.Fetch.Repository != "o/app" || a.Step.Fetch.Dir != "modules/app" || a.Step.Revision != 4 {
+		t.Errorf("the ask = %+v at %s", a, asks.at[0])
+	}
+	// A second start of the revision asks nothing.
+	if ok, err := s.Start(Begin{Branch: "main", Source: 4, Module: "app", Commit: "c1", Entries: entries}); ok || err != nil || len(asks.asks) != 3 {
+		t.Errorf("second start = %v, %v, %d asks", ok, err, len(asks.asks))
+	}
+}
+
+// A pipeline whose every other step is bypassed asks its approval at its start, and one whose steps all ended ends at once.
+func TestStartAsksTheApprovalDueAndEndsAPipelineWithNothingToWaitFor(t *testing.T) {
+	reg, entries := placeFixture()
+	reg.Policy = catalog.Policy{Steps: catalog.StepStatuses{Lint: "bypass", Unit: "bypass", Approval: "required"}}
+	store := created{}
+	var said []string
+	approval := &askRecorder{}
+	s := &Starter{Pipelines: store, Say: func(subject string, _ any) error {
+		said = append(said, subject[strings.Index(subject, ".event.")+7:])
+		return nil
+	},
+		Steps: &stepAsks{}, Registration: func(string) (catalog.Registration, bool) { return reg, true },
+		After: &Hearer{Name: "the-runner", Asker: approval, Say: func(subject string, _ any) error {
+			said = append(said, subject[strings.Index(subject, ".event.")+7:])
+			return nil
+		}}}
+	if _, err := s.Start(Begin{Branch: "main", Source: 4, Module: "app", Commit: "c1", Fp: "src2:aa", Entries: entries}); err != nil {
+		t.Fatal(err)
+	}
+	if len(approval.asked) != 1 {
+		t.Fatalf("approval asks = %d, want 1 at the start", len(approval.asked))
+	}
+	var p seat.Pipeline
+	_ = json.Unmarshal(store["main.4.app"], &p)
+	if last := p.Steps[len(p.Steps)-1]; last.Step != "approval" || last.Outcome != seat.StepResultOutcomeRunning || last.Ask != approval.asked[0].ID {
+		t.Errorf("approval step = %+v", last)
+	}
+	// Nothing required and nothing named: the pipeline ends at once, passed.
+	reg.Policy = catalog.Policy{Steps: catalog.StepStatuses{Lint: "bypass"}}
+	store2 := created{}
+	s.Pipelines = store2
+	said = nil
+	if _, err := s.Start(Begin{Branch: "main", Source: 5, Module: "app", Entries: entries}); err != nil {
+		t.Fatal(err)
+	}
+	_ = json.Unmarshal(store2["main.5.app"], &p)
+	if p.Outcome != seat.PipelineOutcomePassed || !strings.HasPrefix(said[len(said)-1], seat.EventPipelineEnded) {
+		t.Errorf("pipeline = %s, said %v; want it ended passed and said", p.Outcome, said)
+	}
+}

modules/mesh-delivery-pipeline/internal/pipeline/registrations.go

@@ -12,7 +12,7 @@ import (
 
 // Status is a step's status for one artifact of a module, read from the module's registration and never from a
 // manifest (design 47 G §21, §23): the artifact's own status where its policy names one, else the module's. A step
-// neither names is "" (design question Q-RN8, open).
+// neither names is "" and is no step of the pipeline (A-91): Place places none for it.
 func Status(reg catalog.Registration, artifact, step string) string {
 	if a, ok := reg.Policy.Artifacts[artifact]; ok {
 		if s := statusOf(a, step); s != "" {

modules/mesh-delivery-pipeline/internal/pipeline/start.go

@@ -34,6 +34,8 @@ type Begin struct {
 	Group      string
 	Repository string
 	Revision   int64
+	// Entries are the catalog half's entries of the pipeline's branch: the steps' artifacts and build seats are read from them.
+	Entries map[string]catalog.CatalogEntry
 }
 
 // Key is a pipeline's key in the state pipelines, `<branch>.<source revision>.<module>`, the branch escaped as a key's
@@ -51,6 +53,14 @@ type Starter struct {
 	Say  func(subject string, body any) error
 	Now  func() time.Time
 	Logf func(string, ...any)
+	// Registration is a module's registration, whose policy names the steps of its pipeline (§21, §23); with none, a
+	// pipeline starts with no step and says so.
+	Registration func(module string) (catalog.Registration, bool)
+	// Steps asks each build seat for its step, as the pipeline's asking half (§22, §24); Name is the runner's own name as an
+	// asker. After says what a start wrote (an approval asked, a pipeline ended already) as the hearer does.
+	Steps StepAsker
+	Name  string
+	After *Hearer
 }
 
 // Start writes b's pipeline, running, and says pipeline-started; false, nil when its key was written already (the second
@@ -67,6 +77,23 @@ func (s *Starter) Start(b Begin) (bool, error) {
 	p := seat.Pipeline{V: 1, Pipeline: key, Branch: b.Branch, Source: b.Source, Module: b.Module, Commit: b.Commit, Fingerprint: b.Fp,
 		Group: b.Group, Repository: b.Repository, Revision: b.Revision, Outcome: seat.PipelineOutcomeRunning, Started: &now,
 		Steps: []seat.StepResult{}}
+	var asks []Asked
+	if s.Registration != nil {
+		if reg, ok := s.Registration(b.Module); ok {
+			p.Steps, asks = Place(p, reg, b.Entries, now, s.Name)
+		} else {
+			s.logf("%s has no registration yet: its pipeline starts with no step", key)
+		}
+	}
+	// A pipeline whose steps all ended at once (bypassed, or no seat to run them) ends now; an approval due is asked now.
+	var asked, ended bool
+	if s.After != nil {
+		var err error
+		if asked, err = askApproval(key, &p, now, s.After.Asker); err != nil {
+			return false, err
+		}
+		ended = endPipeline(&p, now)
+	}
 	raw, err := json.Marshal(p)
 	if err != nil {
 		return false, err
@@ -86,6 +113,16 @@ func (s *Starter) Start(b Begin) (bool, error) {
 	if err := s.Say(runnerclient.Subject(seat.EventPipelineStarted, b.Module, b.Branch, commitOrGroup), ev); err != nil {
 		s.logf("pipeline-started of %s is not said: %v", key, err)
 	}
+	if s.After != nil {
+		if err := s.After.said(p, asked, ended); err != nil {
+			return true, err
+		}
+	}
+	if s.Steps != nil && len(asks) > 0 {
+		if err := askAll(s.Steps, key, asks); err != nil {
+			return true, err
+		}
+	}
 	return true, nil
 }
 
@@ -118,7 +155,7 @@ func (s *Starter) Published(ev builds.PackagePublished, outdated func(catalog.Ou
 		if o.Kind != outdatedKind || o.Module == "" {
 			continue
 		}
-		b := Begin{Branch: ev.Branch, Source: ev.Source, Module: o.Module, Revision: half.Revision}
+		b := Begin{Branch: ev.Branch, Source: ev.Source, Module: o.Module, Revision: half.Revision, Entries: half.Entries}
 		if e, ok := half.Entries[o.Module]; ok {
 			b.Commit, b.Fp, b.Repository = e.Commit, e.Fp, e.Repository
 		} else {
@@ -156,7 +193,7 @@ func (s *Starter) TakenIn(ev catalog.TakeInTaken, view func(branch string) (Half
 	var first error
 	for _, e := range ev.Entries {
 		ok, err := s.Start(Begin{Branch: ev.Branch, Source: ev.Source, Module: e.Module, Commit: e.Commit, Fp: e.Fingerprint,
-			Group: ev.Group, Repository: e.Repository, Revision: half.Revision})
+			Group: ev.Group, Repository: e.Repository, Revision: half.Revision, Entries: half.Entries})
 		if err != nil && first == nil {
 			first = err
 		}