§416, §25, A-104: a step ask carries the kept results of its package and the toolchain of its language

The asking half reads, in one pass for all a pipeline's asks, the kept entries whose package is the artifact's package from
the catalog entry or lies beneath it, and the runner's own toolchain per language, kept in the new state toolchains as
heard from toolchain-agreed.

Builds: design 47 G §25, §416, A-104

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

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

@@ -58,6 +58,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	if err != nil {
 		return err
 	}
+	// The toolchain of each language as heard agreed, for the steps' asks (§416, A-104).
+	toolchains, err := process.EssentialBucket(ctx, js, "pipeline runner", ownBucket(cred.Module, "toolchains"), "name a step's toolchain")
+	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))
@@ -93,7 +98,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	// for its step.
 	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,
-		Arm: armBound(ctx, bounds)}
+		Arm: armBound(ctx, bounds), Kept: keptReader(ctx, kept), Toolchain: toolchainReader(ctx, toolchains)}
 	resume := func(branch string) {
 		half, ok := rv.Catalog.View(branch)
 		if !ok {
@@ -110,6 +115,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	if err := hearStepEnded(hub, hearer, logger.Printf); err != nil {
 		return err
 	}
+	if err := hearToolchain(ctx, hub, toolchains, logger.Printf); err != nil {
+		return err
+	}
 	if err := hearAskTaken(ctx, hub, bounds, logger.Printf); err != nil {
 		return err
 	}

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

@@ -9,6 +9,7 @@ import (
 
 	"github.com/nats-io/nats.go/jetstream"
 
+	builds "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshbuilds"
 	seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner"
 	nodebuildgo "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo"
 	"git.novox.be/novox/mesh-sdk/go/catalogue"
@@ -199,3 +200,71 @@ func watchBounds(ctx context.Context, bounds jetstream.KeyValue, h *pipeline.Hea
 	}()
 	return nil
 }
+
+// keptReader reads the kept results of the packages named (§25, §416): the entries of the state kept whose package is a
+// name or lies beneath it, in one pass over the bucket.
+func keptReader(ctx context.Context, kept jetstream.KeyValue) func([]string) (map[string][]nodebuildgo.KeptResult, error) {
+	return func(pkgs []string) (map[string][]nodebuildgo.KeptResult, error) {
+		w, err := kept.WatchAll(ctx, jetstream.IgnoreDeletes())
+		if err != nil {
+			return nil, err
+		}
+		defer w.Stop()
+		out := map[string][]nodebuildgo.KeptResult{}
+		for {
+			select {
+			case <-ctx.Done():
+				return nil, ctx.Err()
+			case e, ok := <-w.Updates():
+				if !ok {
+					return out, nil
+				}
+				if e == nil { // the end of what the bucket holds now
+					return out, nil
+				}
+				var k nodebuildgo.KeptResult
+				if err := json.Unmarshal(e.Value(), &k); err != nil {
+					continue
+				}
+				for _, name := range pkgs {
+					if pipeline.InPackage(k.Package, name) {
+						out[name] = append(out[name], k)
+					}
+				}
+			}
+		}
+	}
+}
+
+// toolchainReader reads the toolchain of a language as the runner heard it agreed, from its state toolchains (A-104); nil
+// when none was heard.
+func toolchainReader(ctx context.Context, toolchains jetstream.KeyValue) func(string) (*nodebuildgo.ToolchainUsed, error) {
+	return func(language string) (*nodebuildgo.ToolchainUsed, error) {
+		e, err := toolchains.Get(ctx, keys.EscapeName(language))
+		if errors.Is(err, jetstream.ErrKeyNotFound) {
+			return nil, nil
+		}
+		if err != nil {
+			return nil, err
+		}
+		return pipeline.ToolchainOf(e.Value())
+	}
+}
+
+// hearToolchain keeps each toolchain-agreed heard as `<language>` in the state toolchains (A-104): the newest of a language.
+func hearToolchain(ctx context.Context, hub *process.EventHub, toolchains jetstream.KeyValue, logf func(string, ...any)) error {
+	return hub.Handle(builds.SubjectToolchainAgreed, func(m jetstream.Msg) {
+		t, err := pipeline.ToolchainOf(m.Data())
+		if err != nil || t == nil {
+			logf("%s is no toolchain-agreed: %v", m.Subject(), err)
+			_ = m.Term()
+			return
+		}
+		if _, err := toolchains.Put(ctx, keys.EscapeName(t.Language), m.Data()); err != nil {
+			logf("the toolchain of %s is not kept, heard again: %v", t.Language, err)
+			_ = m.Nak()
+			return
+		}
+		_ = m.Ack()
+	})
+}

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

@@ -96,7 +96,11 @@ func Place(p seat.Pipeline, reg catalog.Registration, entries map[string]catalog
 			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, ID: s.Ask, Ask: stepAsk(p, entry, a, step, s.Ask, asker, now)})
+			ask := Asked{Seat: bs, ID: s.Ask, Ask: stepAsk(p, entry, a, step, s.Ask, asker, now)}
+			if a.Package != nil {
+				ask.Package = a.Package.Name
+			}
+			asks = append(asks, ask)
 		}
 	}
 	for _, step := range moduleSteps {
@@ -126,6 +130,9 @@ type Asked struct {
 	Seat string
 	ID   string
 	Ask  any
+	// Package is the name of the artifact's package, from the catalog entry, for a step on a build seat; its kept results
+	// are the ask's (§25, §416).
+	Package string
 }
 
 func stepAsk(p seat.Pipeline, e catalog.CatalogEntry, a catalog.Artifact, step, id, asker string, now time.Time) nodebuildgo.StepAsk {

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

@@ -214,3 +214,55 @@ func TestResumeAsksTheStepsPlacedAndNotEnded(t *testing.T) {
 		}
 	}
 }
+
+// A step ask carries the kept results of its artifact's package and the toolchain of its language (design 47 G §25, §416,
+// A-104): the results whose package is the artifact's package or lies beneath it, read once for all the asks; a failed
+// read asks the steps with none.
+func TestAStepAskCarriesTheKeptResultsAndTheToolchain(t *testing.T) {
+	reg, entries := placeFixture()
+	e := entries["app"]
+	e.Artifacts[0].Package = &catalog.ArtifactPackage{Ecosystem: "go", Name: "git.example/o/app"}
+	entries["app"] = e
+	store, asks := created{}, &stepAsks{}
+	reads := 0
+	s := &Starter{Pipelines: store, Say: func(string, any) error { return nil }, Steps: asks, Name: "the-runner",
+		Registration: func(string) (catalog.Registration, bool) { return reg, true },
+		Kept: func(pkgs []string) (map[string][]nodebuildgo.KeptResult, error) {
+			reads++
+			if len(pkgs) != 1 || pkgs[0] != "git.example/o/app" {
+				t.Errorf("kept read for %v, want the one package of the entry", pkgs)
+			}
+			return map[string][]nodebuildgo.KeptResult{pkgs[0]: {{Digest: "sha256:aa", Level: "unit", Package: "git.example/o/app/sub"}}}, nil
+		},
+		Toolchain: func(l string) (*nodebuildgo.ToolchainUsed, error) {
+			return &nodebuildgo.ToolchainUsed{Language: l, Version: "1.26"}, nil
+		}}
+	if _, err := s.Start(Begin{Branch: "main", Source: 4, Module: "app", Commit: "c1", Fp: "src2:aa", Repository: "o/app", Entries: entries}); err != nil {
+		t.Fatal(err)
+	}
+	var program, tools *nodebuildgo.StepRun
+	for _, a := range asks.asks {
+		if ask, ok := a.(nodebuildgo.StepAsk); ok {
+			switch {
+			case ask.Step.Artifact == "app:program" && ask.Step.Step == "lint":
+				program = ask.Step
+			case ask.Step.Artifact == "app:tools" && ask.Step.Step == "unit":
+				tools = ask.Step
+			}
+		}
+	}
+	if reads != 1 || program == nil || len(program.Kept) != 1 || program.Kept[0].Digest != "sha256:aa" || program.Toolchain == nil || program.Toolchain.Version != "1.26" {
+		t.Errorf("the program's ask = %+v after %d reads", program, reads)
+	}
+	if tools == nil || len(tools.Kept) != 0 || tools.Toolchain == nil {
+		t.Errorf("an artifact with no package carries no kept result and its language's toolchain: %+v", tools)
+	}
+	for pkg, want := range map[string]bool{"a/b": true, "a/b/c": true, "a/bc": false, "a": false} {
+		if InPackage(pkg, "a/b") != want {
+			t.Errorf("InPackage(%s, a/b) = %v, want %v", pkg, !want, want)
+		}
+	}
+	if tc, err := ToolchainOf([]byte(`{"language":"go","version":"1.26.1","digests":[]}`)); err != nil || tc.Language != "go" || tc.Version != "1.26.1" {
+		t.Errorf("ToolchainOf = %+v, %v", tc, err)
+	}
+}

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

@@ -11,6 +11,7 @@ import (
 	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"
+	nodebuildgo "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo"
 	"git.novox.be/novox/mesh-sdk/go/keys"
 )
 
@@ -60,6 +61,13 @@ type Starter struct {
 	// asker.
 	Steps StepAsker
 	Name  string
+	// Kept reads the kept results of each named package, by the package name an artifact's catalog entry carries (§25, §416): the
+	// results whose package is that name or one beneath it. A step ask carries them, so a build seat runs only the packages
+	// with no result for their digest.
+	Kept func(packages []string) (map[string][]nodebuildgo.KeptResult, error)
+	// Toolchain is the toolchain of a language as the runner heard it agreed (§416, A-104): its language and version; nil when
+	// none was heard, which asks the step with none.
+	Toolchain func(language string) (*nodebuildgo.ToolchainUsed, error)
 	// Arm arms the bound of an ask put on a build seat's work queue (§535, §647); nil arms none.
 	Arm Arm
 }
@@ -106,6 +114,7 @@ func (s *Starter) Start(b Begin) (bool, error) {
 		s.logf("pipeline-started of %s is not said: %v", key, err)
 	}
 	if s.Steps != nil && len(asks) > 0 {
+		s.carry(key, asks)
 		if err := askAll(s.Steps, s.Arm, key, asks); err != nil {
 			return true, err
 		}
@@ -163,6 +172,7 @@ func (s *Starter) Resume(store Updater, branch string, half Half) (int, error) {
 		if len(due) == 0 {
 			continue
 		}
+		s.carry(key, due)
 		if err := askAll(s.Steps, s.Arm, key, due); err != nil && first == nil {
 			first = err
 		}
@@ -171,6 +181,48 @@ func (s *Starter) Resume(store Updater, branch string, half Half) (int, error) {
 	return asked, first
 }
 
+// carry puts the kept results of each ask's package and the toolchain of its language on its step ask (§25, §416, A-104). A
+// read that fails asks the steps with none, so every package runs: slower, never wrong; it is said.
+func (s *Starter) carry(key string, asks []Asked) {
+	var pkgs []string
+	seen := map[string]bool{}
+	for _, a := range asks {
+		if a.Package != "" && !seen[a.Package] {
+			seen[a.Package] = true
+			pkgs = append(pkgs, a.Package)
+		}
+	}
+	var kept map[string][]nodebuildgo.KeptResult
+	if s.Kept != nil && len(pkgs) > 0 {
+		var err error
+		if kept, err = s.Kept(pkgs); err != nil {
+			s.logf("the kept results for the steps of %s are not read, so each package runs: %v", key, err)
+		}
+	}
+	chains := map[string]*nodebuildgo.ToolchainUsed{}
+	for _, a := range asks {
+		ask, ok := a.Ask.(nodebuildgo.StepAsk)
+		if !ok || ask.Step == nil {
+			continue
+		}
+		if a.Package != "" {
+			ask.Step.Kept = kept[a.Package]
+		}
+		if s.Toolchain == nil || ask.Language == "" {
+			continue
+		}
+		t, done := chains[ask.Language]
+		if !done {
+			var err error
+			if t, err = s.Toolchain(ask.Language); err != nil {
+				s.logf("the toolchain of %s for the steps of %s is not read: %v", ask.Language, key, err)
+			}
+			chains[ask.Language] = t
+		}
+		ask.Step.Toolchain = t
+	}
+}
+
 func (s *Starter) logf(format string, args ...any) {
 	if s.Logf != nil {
 		s.Logf(format, args...)

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

@@ -0,0 +1,28 @@
+package pipeline
+
+import (
+	"encoding/json"
+	"strings"
+
+	builds "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshbuilds"
+	nodebuildgo "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo"
+)
+
+// ToolchainOf reads a toolchain-agreed as the toolchain a step ask names: its language and version (design 47 G §416, A-104).
+// The recipe and the building node's digest are the seat's to report, not the runner's to know.
+func ToolchainOf(raw []byte) (*nodebuildgo.ToolchainUsed, error) {
+	var t builds.ToolchainAgreed
+	if err := json.Unmarshal(raw, &t); err != nil {
+		return nil, err
+	}
+	if t.Language == "" {
+		return nil, nil
+	}
+	return &nodebuildgo.ToolchainUsed{Language: t.Language, Version: t.Version}, nil
+}
+
+// InPackage says whether a kept result's package is the artifact's package name or lies beneath it (A-104: the kept entries
+// whose package is the artifact's package, a Go module's packages being the module path and the paths beneath it).
+func InPackage(pkg, name string) bool {
+	return name != "" && (pkg == name || strings.HasPrefix(pkg, name+"/"))
+}

modules/mesh-delivery-pipeline/manifest_test.go

@@ -50,7 +50,7 @@ func TestManifest(t *testing.T) {
 		{"claims the seat serving every verb of its definition", len(m.Claims) == 1 && m.Claims[0].Name == seat.Seat &&
 			equalSets(m.Claims[0].Serves, seat.Verbs())},
 		// §27, §85, §330, §355, §77: the old state sections.
-		{"old state sections", equalSets(states, []string{seat.StatePipelines, seat.StateKept, "aged", "bounds", "findings", "first-views"})},
+		{"old state sections", equalSets(states, []string{seat.StatePipelines, seat.StateKept, "aged", "bounds", "toolchains", "findings", "first-views"})},
 		// §535, §647, A-77, A-104: an ask no holder takes is bound by a key with its own time to live.
 		{"bounds is key-ttl", keyTTL["bounds"]},
 		// §27, §85: the step log on an old stream, its outputs in an old objects section.

modules/mesh-delivery-pipeline/module.json

@@ -71,6 +71,10 @@
       "name": "aged",
       "history": 1
     },
+    {
+      "name": "toolchains",
+      "history": 1
+    },
     {
       "name": "bounds",
       "history": 1,