Merge branch 'feat/47-G-build' into feat/47-G-build-assigner

2026-11-11 12:37:06 +01:00
parent 846340c553 38a2492394
commit 33d9e942f2
12 changed files with 795 additions and 9 deletions
@@ -7,6 +7,10 @@ import (
 	"encoding/json"
 	"fmt"
 	"os"
+	"time"
+
+	"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient"
+	"git.novox.be/novox/mesh-sdk/go/views"

 	seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner"
 	stdio "git.novox.be/novox/mesh-sdk/go"
@@ -82,9 +86,22 @@ func Tools(r pipeline.Reads) []stdio.Tool {
 	return out
 }

+// Module and Slug are the runner's module and its tools' prefix, as its manifest names them.
+const (
+	Module = "mesh-delivery-pipeline"
+	Slug   = "pipeline"
+)
+
+// FirstViewTools are the runner's `<module>_first_views` and `<module>_first_view_again` over its state first-views
+// (design 47 G §77, §200): each view it holds or waits for, since when, and asked again by hand, a hand act.
+func FirstViewTools(st stdio.KeptState, acts views.HandActs, now func() time.Time) []stdio.Tool {
+	return views.StatusTools(Module, Slug+"_first_views", Slug+"_first_view_again", st, acts, now)
+}
+
 func main() {
 	r := pipeline.Reads{Pipelines: kept{stdio.State(seat.StatePipelines)}, Kept: kept{stdio.State(seat.StateKept)}}
-	if err := stdio.Serve("", Tools(r)); err != nil {
+	tools := append(Tools(r), FirstViewTools(stdio.State(views.StatusState), meshhandactsclient.Recorder{Ask: stdio.Ask}, time.Now)...)
+	if err := stdio.Serve("", tools); err != nil {
 		fmt.Fprintf(os.Stderr, "[mesh-delivery-pipeline-tools] %v\n", err)
 		os.Exit(1)
 	}
@@ -5,6 +5,9 @@ import (
 	"strings"
 	"testing"

+	stdio "git.novox.be/novox/mesh-sdk/go"
+	"git.novox.be/novox/mesh-sdk/go/views"
+
 	"mesh-delivery-pipeline/internal/pipeline"
 )

@@ -13,13 +16,17 @@ type none struct{}
 func (none) Keys() ([]string, error)               { return nil, nil }
 func (none) Value(string) (json.RawMessage, error) { return nil, nil }

-// TestTools: the bundle serves the five read verbs as <seat>.<verb> (§83, §306; RN15/AC1).
+// TestTools: the bundle serves the five read verbs as <seat>.<verb> (§83, §306; RN15/AC1), and the first-view tools
+// (§77, §200; RN2/AC4).
 func TestTools(t *testing.T) {
 	var names []string
 	for _, tl := range Tools(pipeline.Reads{Pipelines: none{}, Kept: none{}}) {
 		names = append(names, tl.Name)
 	}
-	want := "mesh-pipeline-runner.show mesh-pipeline-runner.list mesh-pipeline-runner.stands mesh-pipeline-runner.why-carried mesh-pipeline-runner.kept"
+	for _, tl := range FirstViewTools(stdio.State(views.StatusState), nil, nil) {
+		names = append(names, tl.Name)
+	}
+	want := "mesh-pipeline-runner.show mesh-pipeline-runner.list mesh-pipeline-runner.stands mesh-pipeline-runner.why-carried mesh-pipeline-runner.kept pipeline_first_views pipeline_first_view_again"
 	if got := strings.Join(names, " "); got != want {
 		t.Errorf("tools %q, want %q", got, want)
 	}
@@ -5,6 +5,7 @@ package main

 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"io"
 	"log"
@@ -12,8 +13,12 @@ import (
 	"os/signal"
 	"syscall"

+	"github.com/nats-io/nats.go"
+
 	seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner"
 	"git.novox.be/novox/mesh-sdk/go/process"
+
+	"mesh-delivery-pipeline/internal/pipeline"
 )

 func main() {
@@ -39,9 +44,39 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	defer conn.Close()
 	// The state pipelines, `<module>_pipelines`, holds every pipeline by compare-and-set (§27): without it the
 	// runner keeps nothing, so it stops.
-	if _, err := process.EssentialBucket(ctx, js, "pipeline runner", ownBucket(cred.Module, seat.StatePipelines), "keep a pipeline"); err != nil {
+	pipelines, err := process.EssentialBucket(ctx, js, "pipeline runner", ownBucket(cred.Module, seat.StatePipelines), "keep a pipeline")
+	if err != nil {
 		return err
 	}
+	// The registrations and each watched branch's catalog half, first views then events (§21, §195, §200): until a
+	// branch's view is held the runner acts on nothing new of it, and says so.
+	calls, err := process.Connect(cred, nats.Name(cred.Node+"."+cred.Module+".calls"), nats.RetryOnFailedConnect(true))
+	if err != nil {
+		return err
+	}
+	defer calls.Close()
+	hub := process.NewEventHub(logger.Printf)
+	rv, err := startViews(ctx, js, conn, cred, hub, calls.Call, logger.Printf)
+	if err != nil {
+		return err
+	}
+	// A pipeline starts by compare-and-set on its key's absence and says pipeline-started (§54, §103); a publish starts
+	// one per dependent behind (§53, §334).
+	starter := &pipeline.Starter{Pipelines: createdKV{ctx: ctx, kv: pipelines}, Logf: logger.Printf,
+		Say: func(subject string, body any) error {
+			data, err := json.Marshal(body)
+			if err == nil {
+				_, err = js.Publish(ctx, subject, data)
+			}
+			return err
+		}}
+	if err := hearPublished(ctx, hub, starter, calls, rv.Catalog, logger.Printf); err != nil {
+		return err
+	}
+	if err := hub.StartUntilBound(ctx, js, cred, nil, nil); err != nil {
+		logger.Printf("the consumer %s is not bound yet (%v): bound once the bus makes it", cred.EventConsumerName(), err)
+	}
+	go func() { <-ctx.Done(); hub.Stop() }()
 	logger.Printf("holding %s", seat.Seat)
 	<-ctx.Done()
 	return nil
@@ -0,0 +1,71 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+
+	"github.com/nats-io/nats.go/jetstream"
+
+	builds "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshbuilds"
+	catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
+	"git.novox.be/novox/mesh-sdk/go/buscalls"
+	"git.novox.be/novox/mesh-sdk/go/process"
+
+	"mesh-delivery-pipeline/internal/pipeline"
+)
+
+// outdatedVerb is the catalog's verb a publication's dependents behind are asked by, as the manifest's invokes names it.
+const outdatedVerb = "seat:" + catalog.Seat + "." + catalog.VerbOutdated
+
+// createdKV is the state pipelines written by compare-and-set on a key's absence (§54).
+type createdKV struct {
+	ctx context.Context
+	kv  jetstream.KeyValue
+}
+
+func (c createdKV) Create(key string, value []byte) error {
+	_, err := c.kv.Create(c.ctx, key, value)
+	if errors.Is(err, jetstream.ErrKeyExists) {
+		return pipeline.ErrExists
+	}
+	return err
+}
+
+// hearPublished starts the pipelines of each package-published heard (§53, §334): the dependents the catalog's outdated
+// names behind, each asked as a kept call. The message is acknowledged once its starts are written, on a goroutine of its
+// own, so a slow catalog holds no other event.
+func hearPublished(ctx context.Context, hub *process.EventHub, starter *pipeline.Starter, calls *process.Conn, v *pipeline.Views, logf func(string, ...any)) error {
+	outdated := func(in catalog.OutdatedInput) (catalog.OutdatedAnswer, error) {
+		raw, err := calls.CallKept(outdatedVerb, in)
+		if err != nil {
+			return catalog.OutdatedAnswer{}, err
+		}
+		answer, said, ok := buscalls.Unwrap(raw)
+		if !ok {
+			return catalog.OutdatedAnswer{}, fmt.Errorf("%s answered: %s", outdatedVerb, said)
+		}
+		var a catalog.OutdatedAnswer
+		err = json.Unmarshal(answer, &a)
+		return a, err
+	}
+	return hub.Handle(builds.SubjectPackagePublished, func(m jetstream.Msg) {
+		var ev builds.PackagePublished
+		if err := json.Unmarshal(m.Data(), &ev); err != nil {
+			logf("%s is no package-published: %v", m.Subject(), err)
+			_ = m.Ack()
+			return
+		}
+		go func() {
+			n, err := starter.Published(ev, outdated, v.View)
+			if err != nil {
+				logf("the publish of %s %s on %s: %d pipelines started, then %v; heard again", ev.Artifact, ev.Version, ev.Branch, n, err)
+				_ = m.Nak()
+				return
+			}
+			logf("the publish of %s %s on %s started %d pipelines", ev.Artifact, ev.Version, ev.Branch, n)
+			_ = m.Ack()
+		}()
+	})
+}
@@ -0,0 +1,140 @@
+package main
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"github.com/nats-io/nats.go/jetstream"
+
+	mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
+	sdkfindings "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient/findings"
+	"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
+	catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
+	"git.novox.be/novox/mesh-sdk/go/process"
+	"git.novox.be/novox/mesh-sdk/go/views"
+
+	"mesh-delivery-pipeline/internal/pipeline"
+)
+
+// catalogRevisions is every watched branch's catalog-revision, `mesh.seat.mesh-catalog.event.catalog-revision.<branch>`.
+const catalogRevisions = "mesh.seat." + catalog.Seat + ".event." + catalog.EventCatalogRevision + ".>"
+
+// runnerViews are the runner's first views (design 47 G §21, §195, §200): the registrations, then the catalog half of
+// each watched branch they name, each said in `<module>_first_views` (its state first-views) and, past five minutes
+// without a view, as view-waits in its findings.
+type runnerViews struct {
+	Registrations *meshcatalogclient.Registrations
+	Catalog       *pipeline.Views
+}
+
+// startViews wires the runner's views on hub and asks each first view; register every other handler on hub before or
+// after, then start hub once.
+func startViews(ctx context.Context, js jetstream.JetStream, conn process.Subscriber, cred process.Account, hub *process.EventHub,
+	call meshcatalogclient.Call, logf func(string, ...any)) (*runnerViews, error) {
+	findings := process.AwaitBucket(ctx, js, conn, mcc.FindingsBucket(cred.Module), "a view waited for past five minutes is said in the log alone", logf)
+	status := process.AwaitBucket(ctx, js, conn, cred.Module+"_"+views.StatusState, "the runner's first views are not said in its first-view tool, nor asked again by hand", logf)
+	writer := func() views.FindingWriter { return sdkfindings.Writer(ctx, findings.KV) }
+	reader := &views.ProcessReader{Module: cred.Module, Call: call, Logf: logf, Hub: hub, Findings: writer}
+
+	// A watched branch first named by a registration gets its view of the catalog half (CT10), its conditions and its
+	// status kept as every first view's, asked at once.
+	wire := func(f *views.Follower) {
+		f.Logf = logf
+		f.Raise = func(s views.Status) {
+			putFinding(writer(), views.UnansweredFinding(cred.Module, s, time.Now()), true, logf)
+		}
+		f.Clear = func(s views.Status) {
+			putFinding(writer(), views.UnansweredFinding(cred.Module, s, time.Now()), false, logf)
+		}
+		f.RaiseWaits = func(s views.Status) {
+			logf("waits for %s's view %s since %s", s.Owner, s.WaitsFor, s.WaitingSince.Format(time.RFC3339))
+			putFinding(writer(), views.ViewWaitsFinding(cred.Module, s, time.Now()), true, logf)
+		}
+		f.ClearWaits = func(s views.Status) {
+			putFinding(writer(), views.ViewWaitsFinding(cred.Module, s, time.Now()), false, logf)
+		}
+		f.Said = func(s views.Status) {
+			if kv := status.KV(); kv != nil {
+				views.KeepStatus(ctx, kv, logf)(s)
+			}
+		}
+		status.Ready(func(kv jetstream.KeyValue) {
+			if err := views.HearAgain(ctx, kv, []*views.Follower{f}, logf); err != nil {
+				logf("the hand asks of the first view %s are not heard (%v) until the runner starts again", f.Tool, err)
+			}
+		})
+	}
+	var (
+		mu   sync.Mutex
+		made []*views.Follower
+	)
+	cv := &pipeline.Views{Logf: logf, Open: func(branch string) (pipeline.BranchView, error) {
+		c, err := meshcatalogclient.NewCatalog(call, branch)
+		if err != nil {
+			return nil, err
+		}
+		wire(c.Follower)
+		mu.Lock()
+		made = append(made, c.Follower)
+		mu.Unlock()
+		go func() {
+			if err := c.Start(ctx); err != nil {
+				logf("the first view of the catalog half of %s was not answered: %v", branch, err)
+			}
+		}()
+		return catalogView{c}, nil
+	}}
+	regs, _, err := meshcatalogclient.WireRegistrationsProcess(reader, call, func(reg catalog.Registration, gone bool) {
+		if !gone {
+			cv.Registered(reg)
+		}
+	})
+	if err != nil {
+		return nil, err
+	}
+	if err := hub.Handle(catalogRevisions, func(m jetstream.Msg) {
+		if err := cv.HearRevision(ctx, m.Data()); err != nil {
+			logf("%s is not taken: %v", m.Subject(), err)
+		}
+		_ = m.Ack()
+	}); err != nil {
+		return nil, err
+	}
+	// A consumer made after a view was asked does not deliver what was said before it: each branch's view asks again.
+	hub.OnBound(func(created time.Time) {
+		mu.Lock()
+		fs := append([]*views.Follower(nil), made...)
+		mu.Unlock()
+		for _, f := range fs {
+			f.ConsumerMade(ctx, created)
+		}
+	})
+	reader.Start(ctx, js, cred, status)
+	return &runnerViews{Registrations: regs, Catalog: cv}, nil
+}
+
+func putFinding(w views.FindingWriter, f views.Finding, raised bool, logf func(string, ...any)) {
+	var err error
+	if raised {
+		err = w.Put(f)
+	} else {
+		err = w.Delete(f)
+	}
+	if err != nil {
+		logf("the finding %s is not written: %v", f.Kind, err)
+	}
+}
+
+// catalogView is a branch's meshcatalogclient.Catalog as the runner reads it.
+type catalogView struct{ c *meshcatalogclient.Catalog }
+
+func (v catalogView) Hear(ctx context.Context, data []byte) error { return v.c.Hear(ctx, data) }
+
+func (v catalogView) View() (pipeline.Half, bool) {
+	h, ok := v.c.View()
+	if !ok {
+		return pipeline.Half{}, false
+	}
+	return pipeline.Half{Revision: h.Revision.Revision, Source: h.Revision.Source, Entries: h.Entries}, true
+}
@@ -9,6 +9,12 @@ require (
 )

 require (
+	git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient v0.0.0-00010101000000-000000000000
+	git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient v0.0.0-00010101000000-000000000000
+	git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient v0.0.0-00010101000000-000000000000
+	git.novox.be/novox/mesh-delivery/contracts/go/clients/meshpipelinerunnerclient v0.0.0-00010101000000-000000000000
+	git.novox.be/novox/mesh-delivery/contracts/go/seats/meshbuilds v0.0.0-00010101000000-000000000000
+	git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog v0.0.0-00010101000000-000000000000
 	github.com/klauspost/compress v1.20.1 // indirect
 	github.com/nats-io/nkeys v0.4.16 // indirect
 	github.com/nats-io/nuid v1.0.1 // indirect
@@ -0,0 +1,170 @@
+package pipeline
+
+import (
+	"context"
+	"fmt"
+	"sort"
+	"sync"
+
+	catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
+	seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner"
+)
+
+// 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).
+func Status(reg catalog.Registration, artifact, step string) string {
+	if a, ok := reg.Policy.Artifacts[artifact]; ok {
+		if s := statusOf(a, step); s != "" {
+			return s
+		}
+	}
+	return statusOf(reg.Policy.Steps, step)
+}
+
+func statusOf(s catalog.StepStatuses, step string) string {
+	switch step {
+	case seat.StepResultStepLint:
+		return s.Lint
+	case seat.StepResultStepUnit:
+		return s.Unit
+	case seat.StepResultStepIntegration:
+		return s.Integration
+	case seat.StepResultStepComposition:
+		return s.Composition
+	case seat.StepResultStepReview:
+		return s.Review
+	case seat.StepResultStepApproval:
+		return s.Approval
+	}
+	return ""
+}
+
+// buildStep is the acting rule's step a build seat names for the languages it builds (design 47 G §323).
+const buildStep = "build"
+
+// BuildSeat is the build seat that runs lint and the test steps of an artifact of language, read from the catalog half
+// of the pipeline's revision view (design 47 G §323): the seat whose acting rule `build` has `on` naming that language,
+// as an entry of the half that claims it carries it in `acts`. Two seats naming one language are an error naming both.
+func BuildSeat(entries map[string]catalog.CatalogEntry, language string) (string, error) {
+	var seats []string
+	for _, e := range entries {
+		for _, r := range e.Acts {
+			if r.Step == buildStep && r.On == language && r.Seat != "" && !contains(seats, r.Seat) {
+				seats = append(seats, r.Seat)
+			}
+		}
+	}
+	sort.Strings(seats)
+	switch len(seats) {
+	case 1:
+		return seats[0], nil
+	case 0:
+		return "", fmt.Errorf("no build seat's acting rule %s names the language %q in the catalog half", buildStep, language)
+	}
+	return "", fmt.Errorf("the build seats %v each name the language %q in their acting rule %s", seats, language, buildStep)
+}
+
+func contains(list []string, s string) bool {
+	for _, x := range list {
+		if x == s {
+			return true
+		}
+	}
+	return false
+}
+
+// BranchView is the runner's view of one watched branch's catalog half (design 47 G §195, §200, §479): a first view, then
+// that branch's catalog-revision events. meshcatalogclient.Catalog is one.
+type BranchView interface {
+	Hear(ctx context.Context, data []byte) error
+	View() (Half, bool)
+}
+
+// Half is the catalog half a branch view holds, as the runner reads it: its revision and its entries by module.
+type Half struct {
+	Revision int64
+	Source   int64
+	Entries  map[string]catalog.CatalogEntry
+}
+
+// Views is the runner's views of the catalog half, one per watched branch the registrations name (design 47 G §21,
+// §195): a branch first named by a registration gets its view made and started by Open, and every catalog-revision
+// heard goes to every branch's view, each taking its own branch's alone.
+type Views struct {
+	// Open makes and starts the view of a branch; its error is said in the journal and the branch asked again at the
+	// next registration heard.
+	Open func(branch string) (BranchView, error)
+	Logf func(string, ...any)
+
+	mu       sync.Mutex
+	branches map[string]BranchView
+}
+
+// Registered takes the watched branches of a module's registration (design 47 G §21): each branch the views do not hold
+// yet is opened.
+func (v *Views) Registered(reg catalog.Registration) {
+	for _, b := range reg.Branches {
+		v.ensure(b)
+	}
+}
+
+func (v *Views) ensure(branch string) {
+	v.mu.Lock()
+	defer v.mu.Unlock()
+	if v.branches == nil {
+		v.branches = map[string]BranchView{}
+	}
+	if _, ok := v.branches[branch]; ok || branch == "" {
+		return
+	}
+	bv, err := v.Open(branch)
+	if err != nil {
+		if v.Logf != nil {
+			v.Logf("the catalog half of the watched branch %s is not followed (asked again at the next registration heard): %v", branch, err)
+		}
+		return
+	}
+	v.branches[branch] = bv
+}
+
+// Branches are the watched branches the views hold, sorted.
+func (v *Views) Branches() []string {
+	v.mu.Lock()
+	defer v.mu.Unlock()
+	out := make([]string, 0, len(v.branches))
+	for b := range v.branches {
+		out = append(out, b)
+	}
+	sort.Strings(out)
+	return out
+}
+
+// View is the catalog half the view of branch holds; false while it has none, when the runner acts on nothing new of
+// that branch (design 47 G §200).
+func (v *Views) View(branch string) (Half, bool) {
+	v.mu.Lock()
+	bv := v.branches[branch]
+	v.mu.Unlock()
+	if bv == nil {
+		return Half{}, false
+	}
+	return bv.View()
+}
+
+// HearRevision hands one catalog-revision event to every branch's view.
+func (v *Views) HearRevision(ctx context.Context, data []byte) error {
+	v.mu.Lock()
+	all := make([]BranchView, 0, len(v.branches))
+	for _, bv := range v.branches {
+		all = append(all, bv)
+	}
+	v.mu.Unlock()
+	var first error
+	for _, bv := range all {
+		if err := bv.Hear(ctx, data); err != nil && first == nil {
+			first = err
+		}
+	}
+	return first
+}
@@ -0,0 +1,85 @@
+package pipeline
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"testing"
+
+	catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
+)
+
+// The policy, the build seat by language and the views per watched branch (design 47 G §21, §23, §323, §195).
+func TestPolicyAndBuildSeat(t *testing.T) {
+	reg := catalog.Registration{Module: "m", Branches: []string{"main"}, Policy: catalog.Policy{
+		Steps:     catalog.StepStatuses{Lint: "required", Unit: "required", Integration: "optional", Approval: "required"},
+		Artifacts: map[string]catalog.StepStatuses{"tools": {Unit: "bypass"}},
+	}}
+	for _, c := range []struct{ artifact, step, want string }{
+		{"program", "unit", "required"},
+		{"tools", "unit", "bypass"},
+		{"tools", "integration", "optional"},
+		{"tools", "approval", "required"},
+		{"program", "composition", ""},
+	} {
+		if got := Status(reg, c.artifact, c.step); got != c.want {
+			t.Errorf("Status(%s, %s) = %q, want %q", c.artifact, c.step, got, c.want)
+		}
+	}
+
+	entries := map[string]catalog.CatalogEntry{
+		"mesh-build-go":   {Module: "mesh-build-go", Acts: []catalog.ActingRule{{Seat: "node-build-go", Step: "build", On: "go"}}},
+		"mesh-build-node": {Module: "mesh-build-node", Acts: []catalog.ActingRule{{Seat: "node-build-node", Step: "build", On: "typescript"}, {Seat: "node-build-node", Step: "check", On: "go"}}},
+		"app":             {Module: "app", Artifacts: []catalog.Artifact{{Ref: "app:program", Language: "go"}}},
+	}
+	for _, c := range []struct{ language, want, err string }{
+		{"go", "node-build-go", ""},
+		{"typescript", "node-build-node", ""},
+		{"rust", "", `"rust"`},
+	} {
+		got, err := BuildSeat(entries, c.language)
+		if got != c.want || (c.err == "") != (err == nil) || (err != nil && !strings.Contains(err.Error(), c.err)) {
+			t.Errorf("BuildSeat(%s) = %q, %v; want %q, an error naming %s", c.language, got, err, c.want, c.err)
+		}
+	}
+	entries["other"] = catalog.CatalogEntry{Acts: []catalog.ActingRule{{Seat: "node-build-other", Step: "build", On: "go"}}}
+	if _, err := BuildSeat(entries, "go"); err == nil || !strings.Contains(err.Error(), "node-build-other") {
+		t.Errorf("two seats naming go: %v, want an error naming both", err)
+	}
+
+	heard := map[string]int{}
+	opened := 0
+	v := &Views{Open: func(b string) (BranchView, error) {
+		opened++
+		if b == "broken" {
+			return nil, errors.New("refused")
+		}
+		return fakeView{branch: b, heard: heard}, nil
+	}}
+	v.Registered(reg)
+	v.Registered(catalog.Registration{Module: "n", Branches: []string{"main", "feature/x", "broken"}})
+	if got := v.Branches(); strings.Join(got, ",") != "feature/x,main" || opened != 3 {
+		t.Fatalf("branches %v opened %d, want feature/x and main, three opens", got, opened)
+	}
+	v.Registered(catalog.Registration{Module: "n", Branches: []string{"broken"}})
+	if opened != 4 {
+		t.Errorf("a branch whose view was refused is opened again at the next registration: %d opens", opened)
+	}
+	if err := v.HearRevision(context.Background(), []byte(`{}`)); err != nil || heard["main"] != 1 || heard["feature/x"] != 1 {
+		t.Errorf("a revision heard reaches every branch's view: %v %v", err, heard)
+	}
+	if h, ok := v.View("main"); !ok || h.Revision != 7 {
+		t.Errorf("View(main) = %v %v", h, ok)
+	}
+	if _, ok := v.View("gone"); ok {
+		t.Error("a branch not watched has no view")
+	}
+}
+
+type fakeView struct {
+	branch string
+	heard  map[string]int
+}
+
+func (f fakeView) Hear(context.Context, []byte) error { f.heard[f.branch]++; return nil }
+func (f fakeView) View() (Half, bool)                 { return Half{Revision: 7}, true }
@@ -0,0 +1,136 @@
+package pipeline
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strconv"
+	"time"
+
+	runnerclient "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshpipelinerunnerclient"
+	builds "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshbuilds"
+	catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
+	seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner"
+	"git.novox.be/novox/mesh-sdk/go/keys"
+)
+
+// ErrExists is a Create of a pipeline key that is written already: the second start, which writes nothing (§54).
+var ErrExists = errors.New("the pipeline's key is written already")
+
+// Created is the runner's state pipelines as a start writes it: a key written only when it is absent (compare-and-set on
+// its absence), ErrExists when it is not.
+type Created interface {
+	Create(key string, value []byte) error
+}
+
+// Begin is what a start names: the pipeline's key (watched branch, source revision, module; design 47 G §27, §54) and
+// what the catalog half of that branch's view says of the module.
+type Begin struct {
+	Branch     string
+	Source     int64
+	Module     string
+	Commit     string
+	Fp         string
+	Group      string
+	Repository string
+	Revision   int64
+}
+
+// Key is a pipeline's key in the state pipelines, `<branch>.<source revision>.<module>`, the branch escaped as a key's
+// name is (§27).
+func Key(branch string, source int64, module string) string {
+	return seat.PipelinesByBranchRevisionModuleKey(keys.EscapeName(branch), strconv.FormatInt(source, 10), module)
+}
+
+// Starter starts pipelines (design 47 G §54, §103): it writes a pipeline's key by compare-and-set on its absence, so one
+// source revision and module has one pipeline whichever event starts it first, and says pipeline-started for the one it
+// wrote.
+type Starter struct {
+	Pipelines Created
+	// Say says one of the seat's events on its subject.
+	Say  func(subject string, body any) error
+	Now  func() time.Time
+	Logf func(string, ...any)
+}
+
+// Start writes b's pipeline, running, and says pipeline-started; false, nil when its key was written already (the second
+// start writes nothing, §54).
+func (s *Starter) Start(b Begin) (bool, error) {
+	if b.Branch == "" || b.Module == "" || b.Source <= 0 {
+		return false, fmt.Errorf("a pipeline names its watched branch, source revision and module: %+v", b)
+	}
+	now := time.Now().UTC()
+	if s.Now != nil {
+		now = s.Now().UTC()
+	}
+	key := Key(b.Branch, b.Source, b.Module)
+	p := seat.Pipeline{V: 1, Pipeline: key, Branch: b.Branch, Source: b.Source, Module: b.Module, Commit: b.Commit, Fp: b.Fp,
+		Group: b.Group, Repository: b.Repository, Revision: b.Revision, Outcome: seat.PipelineOutcomeRunning, Started: &now,
+		Steps: []seat.StepResult{}}
+	raw, err := json.Marshal(p)
+	if err != nil {
+		return false, err
+	}
+	if err := s.Pipelines.Create(key, raw); errors.Is(err, ErrExists) {
+		s.logf("%s is started already: this start writes nothing", key)
+		return false, nil
+	} else if err != nil {
+		return false, fmt.Errorf("the pipeline %s is not written: %w", key, err)
+	}
+	ev := seat.PipelineStarted{V: 1, At: &now, Pipeline: key, Branch: b.Branch, Source: b.Source, Module: b.Module,
+		Commit: b.Commit, Fp: b.Fp, Group: b.Group, Revision: b.Revision}
+	commitOrGroup := b.Commit
+	if b.Group != "" {
+		commitOrGroup = b.Group
+	}
+	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)
+	}
+	return true, nil
+}
+
+func (s *Starter) logf(format string, args ...any) {
+	if s.Logf != nil {
+		s.Logf(format, args...)
+	}
+}
+
+// outdatedKind is the edge `outdated` names a dependent behind by for a rebuild on a publish: B published before A builds,
+// a package A imports (§53).
+const outdatedKind = 1
+
+// Published starts, for a package-published, a pipeline for each dependent the catalog's `outdated` names behind for that
+// package on that branch by kind 1, keyed by the publication's watched branch and source revision and the dependent
+// (design 47 G §53, §334). outdated is the kept call of the verb; view the catalog half of the publication's branch. A
+// dependent the half does not hold is said and started with what the event names.
+func (s *Starter) Published(ev builds.PackagePublished, outdated func(catalog.OutdatedInput) (catalog.OutdatedAnswer, error), view func(branch string) (Half, bool)) (int, error) {
+	if ev.Artifact == "" || ev.Branch == "" || ev.Source <= 0 {
+		return 0, fmt.Errorf("a package-published names its artifact, watched branch and source revision: %+v", ev)
+	}
+	a, err := outdated(catalog.OutdatedInput{Branch: ev.Branch, Dependency: ev.Artifact})
+	if err != nil {
+		return 0, fmt.Errorf("outdated of %s on %s: %w", ev.Artifact, ev.Branch, err)
+	}
+	half, _ := view(ev.Branch)
+	started := 0
+	var first error
+	for _, o := range a.Behind {
+		if o.Kind != outdatedKind || o.Module == "" {
+			continue
+		}
+		b := Begin{Branch: ev.Branch, Source: ev.Source, Module: o.Module, Revision: half.Revision}
+		if e, ok := half.Entries[o.Module]; ok {
+			b.Commit, b.Fp, b.Repository = e.Commit, e.Fp, e.Repository
+		} else {
+			s.logf("the catalog half of %s holds no entry of %s, behind %s: its pipeline starts without its commit", ev.Branch, o.Module, ev.Artifact)
+		}
+		ok, err := s.Start(b)
+		if err != nil && first == nil {
+			first = err
+		}
+		if ok {
+			started++
+		}
+	}
+	return started, first
+}
@@ -0,0 +1,87 @@
+package pipeline
+
+import (
+	"encoding/json"
+	"errors"
+	"strings"
+	"testing"
+
+	builds "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshbuilds"
+	catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
+	seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner"
+)
+
+type created map[string][]byte
+
+func (c created) Create(key string, value []byte) error {
+	if _, ok := c[key]; ok {
+		return ErrExists
+	}
+	c[key] = value
+	return nil
+}
+
+// The starts (design 47 G §54, §103, §334): a pipeline's key written by compare-and-set on its absence, so a second start
+// of the same source revision and module writes nothing; a publish starts one pipeline per dependent outdated names
+// behind by kind 1, keyed by the publishing branch and revision.
+func TestStarts(t *testing.T) {
+	store := created{}
+	var said []string
+	s := &Starter{Pipelines: store, Say: func(subject string, body any) error {
+		ev := body.(seat.PipelineStarted)
+		said = append(said, subject+" "+ev.Pipeline)
+		return nil
+	}}
+	view := func(branch string) (Half, bool) {
+		return Half{Revision: 9, Entries: map[string]catalog.CatalogEntry{
+			"app": {Module: "app", Commit: "c1", Fp: "src2:aa", Repository: "o/app"}}}, true
+	}
+	var asked catalog.OutdatedInput
+	outdated := func(in catalog.OutdatedInput) (catalog.OutdatedAnswer, error) {
+		asked = in
+		return catalog.OutdatedAnswer{Behind: []catalog.Outdated{
+			{Module: "app", Kind: 1}, {Module: "tool", Kind: 1}, {Module: "ordered", Kind: 4}}}, nil
+	}
+	n, err := s.Published(builds.PackagePublished{Artifact: "mesh-sdk:go", Branch: "main", Source: 41, Module: "mesh-sdk", Version: "0.1.16"}, outdated, view)
+	if err != nil || n != 2 {
+		t.Fatalf("a publish started %d (%v), want the two dependents behind by kind 1", n, err)
+	}
+	if asked.Dependency != "mesh-sdk:go" || asked.Branch != "main" {
+		t.Errorf("outdated asked %+v, want the package on its branch", asked)
+	}
+	var p seat.Pipeline
+	if err := json.Unmarshal(store["main.41.app"], &p); err != nil || p.Commit != "c1" || p.Fp != "src2:aa" || p.Revision != 9 ||
+		p.Outcome != seat.PipelineOutcomeRunning || p.Source != 41 {
+		t.Errorf("the pipeline main.41.app = %+v (%v)", p, err)
+	}
+	if _, ok := store["main.41.ordered"]; ok {
+		t.Error("an ordering edge starts no pipeline")
+	}
+	// The take-in of the same source revision and module is the same pipeline: the second start writes nothing.
+	before := string(store["main.41.app"])
+	ok, err := s.Start(Begin{Branch: "main", Source: 41, Module: "app", Commit: "c2"})
+	if ok || err != nil || string(store["main.41.app"]) != before || len(said) != 2 {
+		t.Errorf("a second start wrote %v (%v); said %v", ok, err, said)
+	}
+	if !strings.HasPrefix(said[0], "mesh.seat.mesh-pipeline-runner.event.pipeline-started.app.main.c1 main.41.app") {
+		t.Errorf("pipeline-started said on %q", said[0])
+	}
+	// A branch holding a dot is escaped in the key, so the key reads back; a start naming no source revision is refused.
+	if _, err := s.Start(Begin{Branch: "release.1", Source: 3, Module: "app"}); err != nil {
+		t.Fatal(err)
+	}
+	if b, r, m, ok := seat.ParsePipelinesByBranchRevisionModuleKey(Key("release.1", 3, "app")); !ok || r != "3" || m != "app" || b == "release" {
+		t.Errorf("the key of release.1 is %q", Key("release.1", 3, "app"))
+	}
+	if _, err := s.Start(Begin{Branch: "main", Module: "app"}); err == nil {
+		t.Error("a start with no source revision is refused")
+	}
+	s.Pipelines = failing{}
+	if _, err := s.Start(Begin{Branch: "main", Source: 5, Module: "app"}); err == nil {
+		t.Error("a write the bus refuses is an error")
+	}
+}
+
+type failing struct{}
+
+func (failing) Create(string, []byte) error { return errors.New("bus away") }
@@ -49,8 +49,13 @@ func TestManifest(t *testing.T) {
 		// §27, §85, §330, §355, §77: the old state sections.
 		{"old state sections", equalSets(states, []string{seat.StatePipelines, seat.StateKept, "aged", "findings", "first-views"})},
 		// §27, §85: the step log on an old stream, its outputs in an old objects section.
-		{"old streams section", equalSets(streams, []string{seat.StateStepLog})},
-		{"step-log 128 MiB", len(m.Streams) == 1 && m.Streams[0].MaxMiB == 128},
+		// RN24 (§383, §387): beside it, the finding stream its findings are said on.
+		{"old streams section", equalSets(streams, []string{seat.StateStepLog, "mesh-delivery-pipeline-findings"})},
+		{"step-log 128 MiB", len(m.Streams) == 2 && m.Streams[0].MaxMiB == 128},
+		{"finding stream 7 days 16 MiB", len(m.Streams) == 2 && m.Streams[1].MaxAgeDays == 7 && m.Streams[1].MaxMiB == 16 &&
+			equalSets(m.Streams[1].Events, []string{"finding-written.*", "finding-cleared.*", "findings-stated.*"})},
+		// RN24 (§368): the version as MAJOR.MINOR, 0.1 with none published.
+		{"version 0.1", m.Version == "0.1"},
 		{"old objects section", equalSets(objects, []string{seat.StateStepOutputs})},
 		{"step-outputs 1 GiB", len(m.Objects) == 1 && m.Objects[0].MaxMiB == 1024},
 		// §333, §54, §44, §53, §376, §24, §37, §4, §77: what the runner hears.
@@ -58,7 +63,13 @@ func TestManifest(t *testing.T) {
 			"mesh-delivery-indexer.take-in.taken", "mesh-builds.package-published", "mesh-builds.toolchain-agreed",
 			"operator-channel.decided", "node-build-go.step-ended", "node-build-node.step-ended",
 			"node-build-image.step-ended", "mesh-wants.step-ended", "mesh-catalog.registration-changed",
-			"mesh-catalog.catalog-revision"})},
+			"mesh-catalog.catalog-revision", "mesh-conditions.findings-asked"})},
+		// RN2 (§21, §195, §200, §77): the registrations and the catalog half asked by their verbs, the first views read and
+		// asked again by hand (a hand act).
+		{"invokes", equalSets(m.Invokes, []string{"seat:mesh-catalog.registrations", "seat:mesh-catalog.revision", "seat:mesh-hand-acts.hand-act",
+			// RN3 (§53, §334): a publication's dependents behind.
+			"seat:mesh-catalog.outdated"})},
+		{"first-view tools", equalSets(m.Tools, []string{"pipeline_first_views", "pipeline_first_view_again"})},
 		// RN1/AC5 (§95, §130): the new assigner is delivered before the runner.
 		{"program needs the assigner's program", slices.Equal(program.Needs, []string{"mesh-delivery-assigner:program"})},
 	}
@@ -1,6 +1,6 @@
 {
  "module": "mesh-delivery-pipeline",
-  "version": "1",
+  "version": "0.1",
  "slug": "pipeline",
  "claims": [
    {
@@ -22,6 +22,10 @@
      ]
    }
  ],
+  "tools": [
+    "pipeline_first_views",
+    "pipeline_first_view_again"
+  ],
  "consumes": [
    "mesh-delivery-indexer.take-in.taken",
    "mesh-builds.package-published",
@@ -32,7 +36,14 @@
    "node-build-image.step-ended",
    "mesh-wants.step-ended",
    "mesh-catalog.registration-changed",
-    "mesh-catalog.catalog-revision"
+    "mesh-catalog.catalog-revision",
+    "mesh-conditions.findings-asked"
+  ],
+  "invokes": [
+    "seat:mesh-catalog.registrations",
+    "seat:mesh-catalog.revision",
+    "seat:mesh-hand-acts.hand-act",
+    "seat:mesh-catalog.outdated"
  ],
  "state": [
    {
@@ -63,6 +74,16 @@
        "step-log.**"
      ],
      "max-mib": 128
+    },
+    {
+      "name": "mesh-delivery-pipeline-findings",
+      "events": [
+        "finding-written.*",
+        "finding-cleared.*",
+        "findings-stated.*"
+      ],
+      "max-age-days": 7,
+      "max-mib": 16
    }
  ],
  "objects": [