Run a pulled ask's whole job with a build module's own step tools, kept verdicts taken after its digest, so both build modules run asks one way and their tools stay the hand path (issue 1887 A3, ADR 0525 §3, §8)

2026-10-23 15:22:25 +02:00
parent c2967432e6
commit 1f6fc027c0
3 changed files with 497 additions and 0 deletions
@@ -0,0 +1,345 @@
+package buildstep
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"path"
+	"sort"
+	"strings"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+// **One ask, one whole job** (novox/hq ADR 0525 §3). A build module runs a pulled ask with its own step tools' bodies,
+// the same ones its request and reply tools run (ADR 0525 §8), in the order the job names: a build ask fetches, then
+// builds or copies each artifact; a check ask fetches, digests the level, takes the kept verdicts and tests the rest,
+// the packages whose tests the change touches first. The answers are the tools' own, whole, so the asker judges them as
+// it judged the tools' answers.
+
+// StepFunc is one step tool's body: its arguments in, its answer out.
+type StepFunc func(ctx context.Context, args map[string]any) (any, error)
+
+// AskSteps are the step tools a build module runs an ask with. Copy is the image module's alone; Digest the Go module's.
+type AskSteps struct {
+	Fetch, Build, Copy, Digest, Test StepFunc
+	// Toolchain is the toolchain the module builds with now; a build ask naming another is refused. Nil checks nothing
+	// (an image or an archive names none).
+	Toolchain func(ctx context.Context) (delivery.ToolchainUsed, error)
+	// Kept looks up a kept verdict by level and digest (the checker's kept bucket, delivery.KeptKey): whether one is
+	// kept and how many tests it counted. Nil takes none.
+	Kept func(ctx context.Context, level, digest string) (tests int, found bool, err error)
+	Logf func(format string, args ...any)
+}
+
+// RunAsk runs ask's job with steps and answers a delivery.BuildAnswer or a delivery.CheckAnswer.
+func RunAsk(ctx context.Context, steps AskSteps, ask delivery.StepAsk) (any, error) {
+	if ask.Build != nil {
+		return steps.build(ctx, *ask.Build)
+	}
+	if ask.Check != nil {
+		return steps.check(ctx, *ask.Check)
+	}
+	return nil, Refuse("the ask %s holds no job", ask.ID)
+}
+
+func raw(v any) (json.RawMessage, error) {
+	if r, ok := v.(json.RawMessage); ok {
+		return r, nil
+	}
+	return json.Marshal(v)
+}
+
+// fetched is what an ask reads of a fetch's answer.
+type fetched struct {
+	Workspace string   `json:"workspace"`
+	Merge     string   `json:"merge"`
+	Conflicts []string `json:"conflicts"`
+	Changed   []string `json:"changed"`
+	Modules   []string `json:"modules"`
+}
+
+func (s AskSteps) fetch(ctx context.Context, args map[string]any) (json.RawMessage, fetched, error) {
+	if s.Fetch == nil {
+		return nil, fetched{}, Refuse("this build module fetches nothing")
+	}
+	out, err := s.Fetch(ctx, args)
+	if err != nil {
+		return nil, fetched{}, err
+	}
+	r, err := raw(out)
+	if err != nil {
+		return nil, fetched{}, err
+	}
+	var f fetched
+	if err := json.Unmarshal(r, &f); err != nil || f.Workspace == "" {
+		return r, f, fmt.Errorf("the fetch answered no workspace: %s", r)
+	}
+	return r, f, nil
+}
+
+func (s AskSteps) build(ctx context.Context, job delivery.BuildJob) (any, error) {
+	answer := delivery.BuildAnswer{}
+	if job.Toolchain != nil && s.Toolchain != nil {
+		tc, err := s.Toolchain(ctx)
+		if err != nil {
+			return nil, err
+		}
+		if tc.Version != job.Toolchain.Version || tc.Recipe != job.Toolchain.Recipe {
+			return nil, Refuse("the ask is for %s (recipe %s) and this node builds with %s (recipe %s)",
+				job.Toolchain.Version, job.Toolchain.Recipe, tc.Version, tc.Recipe)
+		}
+		answer.Toolchain = &tc
+	}
+	var f fetched
+	for _, a := range job.Artifacts {
+		if a.From == "" && f.Workspace == "" {
+			r, got, err := s.fetch(ctx, job.Fetch)
+			if err != nil {
+				return nil, err
+			}
+			if _, candidate := job.Fetch["base"]; candidate {
+				// A candidate is built from its head merged onto the trunk; a head that does not merge leaves the tree at the
+				// trunk, and building that would store the trunk's build under the candidate's fingerprint.
+				if len(got.Conflicts) > 0 {
+					return nil, fmt.Errorf("the head does not merge into the trunk: files in conflict: %s", strings.Join(got.Conflicts, ", "))
+				}
+				if got.Merge == "" {
+					return nil, fmt.Errorf("the fetch with a base answered no merge: %s", r)
+				}
+			}
+			f = got
+		}
+		var (
+			out any
+			err error
+		)
+		if a.From != "" {
+			if s.Copy == nil {
+				return nil, Refuse("this build module copies no artifact (%s)", a.Artifact)
+			}
+			out, err = s.Copy(ctx, map[string]any{"module": job.Module, "artifact": a.Artifact, "from": a.From})
+		} else {
+			if s.Build == nil {
+				return nil, Refuse("this build module builds no artifact (%s)", a.Artifact)
+			}
+			args := map[string]any{"workspace": f.Workspace, "module": job.Path, "artifact": a.Artifact}
+			if job.Publish {
+				args["publish"] = true
+			}
+			out, err = s.Build(ctx, args)
+		}
+		if err != nil {
+			return nil, fmt.Errorf("%s: %w", a.Artifact, err)
+		}
+		r, err := raw(out)
+		if err != nil {
+			return nil, err
+		}
+		var built struct {
+			Digest    string `json:"digest"`
+			Reference string `json:"reference"`
+		}
+		if err := json.Unmarshal(r, &built); err != nil || built.Digest == "" {
+			return nil, fmt.Errorf("the build of %s answered no digest: %s", a.Artifact, r)
+		}
+		answer.Artifacts = append(answer.Artifacts, delivery.AskBuilt{Artifact: a.Artifact, Digest: built.Digest, Reference: built.Reference})
+	}
+	return answer, nil
+}
+
+func (s AskSteps) check(ctx context.Context, job delivery.CheckJob) (any, error) {
+	r, f, err := s.fetch(ctx, job.Fetch)
+	if err != nil {
+		return nil, err
+	}
+	answer := delivery.CheckAnswer{Fetch: r}
+	if len(f.Conflicts) > 0 {
+		return answer, nil // the asker reads the conflict from the fetch's answer: a verdict, not an error
+	}
+	runner := ""
+	if job.Runner != "" && (job.RunnerWhen == "" || changesUnder(f.Changed, job.RunnerWhen)) {
+		runner = job.Runner
+	}
+	withRunner := func(args map[string]any) map[string]any {
+		if runner != "" {
+			args["runner"] = runner
+		}
+		return args
+	}
+	if s.Test == nil {
+		return nil, Refuse("this build module tests nothing")
+	}
+	test := func(args map[string]any) (bool, error) {
+		out, err := s.Test(ctx, args)
+		if err != nil {
+			return false, err
+		}
+		r, err := raw(out)
+		if err != nil {
+			return false, err
+		}
+		answer.Tests = append(answer.Tests, r)
+		var t struct {
+			Passed bool `json:"passed"`
+		}
+		_ = json.Unmarshal(r, &t)
+		return t.Passed, nil
+	}
+	if s.Digest == nil {
+		// A module with no digest (Node) tests each module directory of the tree, those whose tests changed first.
+		for _, priority := range []bool{true, false} {
+			for _, dir := range f.Modules {
+				if changesTestIn(f.Changed, dir) != priority {
+					continue
+				}
+				passed, err := test(withRunner(map[string]any{"workspace": f.Workspace, "module": dir, "level": job.Level}))
+				if err != nil {
+					return nil, fmt.Errorf("the %s level of %s: %w", job.Level, dir, err)
+				}
+				if !passed {
+					return answer, nil
+				}
+			}
+		}
+		return answer, nil
+	}
+	todo, all, err := s.plan(ctx, job, f, withRunner, &answer)
+	if err != nil {
+		return nil, err
+	}
+	for _, priority := range []bool{true, false} {
+		args := withRunner(map[string]any{"workspace": f.Workspace, "level": job.Level})
+		if job.Run != "" {
+			args["run"] = job.Run
+		}
+		if all {
+			if priority {
+				continue // nothing to put first: the level runs whole after
+			}
+		} else {
+			var run []string
+			for _, p := range todo {
+				if changesTestOf(f.Changed, p) == priority {
+					run = append(run, p)
+				}
+			}
+			if len(run) == 0 {
+				continue
+			}
+			args["packages"] = run
+		}
+		passed, err := test(args)
+		if err != nil {
+			return nil, fmt.Errorf("the %s level: %w", job.Level, err)
+		}
+		if !passed {
+			return answer, nil
+		}
+	}
+	return answer, nil
+}
+
+// plan digests the level and takes the kept verdicts: the packages left to test, or all when the level runs whole (a
+// try of some tests, or a digest that named nothing).
+func (s AskSteps) plan(ctx context.Context, job delivery.CheckJob, f fetched, withRunner func(map[string]any) map[string]any, answer *delivery.CheckAnswer) ([]string, bool, error) {
+	if job.Run != "" {
+		return nil, true, nil // a partial run proves no package: nothing is taken or kept
+	}
+	out, err := s.Digest(ctx, withRunner(map[string]any{"workspace": f.Workspace, "level": job.Level}))
+	if err != nil {
+		if s.Logf != nil {
+			s.Logf("no digest of the %s level, so it runs whole: %v", job.Level, err)
+		}
+		return nil, true, nil
+	}
+	r, err := raw(out)
+	if err != nil {
+		return nil, true, nil
+	}
+	answer.Digest = r
+	var d struct {
+		Packages map[string]string `json:"packages"`
+	}
+	if err := json.Unmarshal(r, &d); err != nil || len(d.Packages) == 0 {
+		return nil, true, nil
+	}
+	pkgs := make([]string, 0, len(d.Packages))
+	for p := range d.Packages {
+		pkgs = append(pkgs, p)
+	}
+	sort.Strings(pkgs)
+	var todo []string
+	for _, p := range pkgs {
+		if job.Kept && s.Kept != nil {
+			tests, found, err := s.Kept(ctx, job.Level, d.Packages[p])
+			if err != nil && s.Logf != nil {
+				s.Logf("the kept verdict of %s at %s cannot be read, so it runs: %v", p, job.Level, err)
+			}
+			if err == nil && found {
+				answer.Kept = append(answer.Kept, delivery.KeptTaken{Package: p, Digest: d.Packages[p], Tests: tests})
+				continue
+			}
+		}
+		todo = append(todo, p)
+	}
+	if len(todo) == 0 {
+		return nil, false, nil
+	}
+	return todo, false, nil
+}
+
+// changesUnder says whether a path the merge changes is under prefix.
+func changesUnder(changed []string, prefix string) bool {
+	for _, p := range changed {
+		if strings.HasPrefix(p, prefix) {
+			return true
+		}
+	}
+	return false
+}
+
+// isTestFile says whether a repository-relative path is a test file: Go's _test.go, and a TypeScript module's .test.
+// and .spec. files and test directories.
+func isTestFile(p string) bool {
+	switch {
+	case strings.HasSuffix(p, "_test.go"):
+		return true
+	case strings.Contains(p, ".test.") || strings.Contains(p, ".spec."):
+		return true
+	}
+	for _, seg := range strings.Split(path.Dir(p), "/") {
+		if seg == "__tests__" || seg == "test" || seg == "tests" {
+			return true
+		}
+	}
+	return false
+}
+
+// changesTestOf says whether the merge changes a test file of the Go package with the import path pkg: the file's
+// directory and the import path agree on their last segments.
+func changesTestOf(changed []string, pkg string) bool {
+	for _, f := range changed {
+		if !strings.HasSuffix(f, "_test.go") {
+			continue
+		}
+		dir := path.Dir(f)
+		if dir == pkg || strings.HasSuffix(pkg, "/"+dir) || strings.HasSuffix(dir, "/"+pkg) {
+			return true
+		}
+	}
+	return false
+}
+
+// changesTestIn says whether the merge changes a test file inside the directory of a Node module.
+func changesTestIn(changed []string, dir string) bool {
+	for _, f := range changed {
+		if !isTestFile(f) || strings.HasSuffix(f, "_test.go") {
+			continue
+		}
+		if dir == "." || dir == "" || strings.HasPrefix(f, strings.TrimSuffix(dir, "/")+"/") {
+			return true
+		}
+	}
+	return false
+}
@@ -0,0 +1,147 @@
+package buildstep
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+	"testing"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+const runCommit = "0123456789abcdef0123456789abcdef01234567"
+
+// recorder is a build module's step tools that record each call.
+type recorder struct {
+	calls   []string
+	fetch   map[string]any
+	tests   map[string]bool // package or module directory -> passed
+	digests map[string]string
+}
+
+func (r *recorder) steps() AskSteps {
+	call := func(name string, answer func(args map[string]any) any) StepFunc {
+		return func(ctx context.Context, args map[string]any) (any, error) {
+			b, _ := json.Marshal(args)
+			r.calls = append(r.calls, name+" "+string(b))
+			return answer(args), nil
+		}
+	}
+	return AskSteps{
+		Fetch: call("fetch", func(map[string]any) any { return r.fetch }),
+		Build: call("build", func(args map[string]any) any {
+			return map[string]any{"digest": "sha256:" + args["artifact"].(string), "reference": "artifact-store://" + args["artifact"].(string)}
+		}),
+		Copy:   call("copy", func(args map[string]any) any { return map[string]any{"digest": "sha256:copied"} }),
+		Digest: call("digest", func(map[string]any) any { return map[string]any{"level": "unit", "packages": r.digests} }),
+		Test: call("test", func(args map[string]any) any {
+			passed := true
+			if pkgs, ok := args["packages"].([]string); ok {
+				for _, p := range pkgs {
+					passed = passed && r.tests[p]
+				}
+			}
+			if dir, ok := args["module"].(string); ok {
+				passed = r.tests[dir]
+			}
+			return map[string]any{"passed": passed, "tests": 1}
+		}),
+		Kept: func(ctx context.Context, level, digest string) (int, bool, error) { return 7, digest == "d-kept", nil },
+	}
+}
+
+func ask(job any) delivery.StepAsk {
+	a := delivery.StepAsk{ID: NewID(), Kind: delivery.AskCheck, Language: delivery.AskGo, Asker: "t", At: time.Now()}
+	switch j := job.(type) {
+	case delivery.BuildJob:
+		a.Build = &j
+	case delivery.CheckJob:
+		a.Check = &j
+	}
+	return a
+}
+
+// 1887/A3: a build ask fetches once, builds each artifact from the workspace with publish, copies a pinned one without a
+// fetch, and answers each artifact's digest and reference.
+func TestRunAskBuilds(t *testing.T) {
+	r := &recorder{fetch: map[string]any{"workspace": "/w/1"}}
+	out, err := RunAsk(context.Background(), r.steps(), ask(delivery.BuildJob{Module: "m", Path: "modules/m", Publish: true,
+		Fetch:     map[string]any{"repository": "novox/x", "commit": runCommit},
+		Artifacts: []delivery.AskArtifact{{Artifact: "base", From: "registry/b@sha256:00"}, {Artifact: "bundle"}, {Artifact: "lib"}}}))
+	if err != nil {
+		t.Fatal(err)
+	}
+	got := strings.Join(r.calls, "\n")
+	want := `copy {"artifact":"base","from":"registry/b@sha256:00","module":"m"}
+fetch {"commit":"` + runCommit + `","repository":"novox/x"}
+build {"artifact":"bundle","module":"modules/m","publish":true,"workspace":"/w/1"}
+build {"artifact":"lib","module":"modules/m","publish":true,"workspace":"/w/1"}`
+	if got != want {
+		t.Fatalf("calls:\n%s\nwant:\n%s", got, want)
+	}
+	a := out.(delivery.BuildAnswer)
+	if len(a.Artifacts) != 3 || a.Artifacts[1].Reference != "artifact-store://bundle" {
+		t.Fatalf("answer %+v", a)
+	}
+}
+
+// A build ask for another toolchain is refused, so its step ends refused and the asker reads why.
+func TestRunAskRefusesAnotherToolchain(t *testing.T) {
+	r := &recorder{fetch: map[string]any{"workspace": "/w"}}
+	s := r.steps()
+	s.Toolchain = func(context.Context) (delivery.ToolchainUsed, error) { return delivery.ToolchainUsed{Version: "go1.27", Recipe: "3"}, nil }
+	_, err := RunAsk(context.Background(), s, ask(delivery.BuildJob{Module: "m", Path: ".", Fetch: map[string]any{"repository": "novox/x", "commit": runCommit},
+		Artifacts: []delivery.AskArtifact{{Artifact: "bundle"}}, Toolchain: &delivery.ToolchainUsed{Version: "go1.26", Recipe: "3"}}))
+	var ref *Refusal
+	if !errors.As(err, &ref) || len(r.calls) != 0 {
+		t.Fatalf("err %v, calls %v", err, r.calls)
+	}
+}
+
+// 1887/A3, issue 1887 design limit 1: a check ask digests the level, takes the kept verdicts, tests the packages whose
+// tests changed first, and stops at the first failure.
+func TestRunAskChecksAGoLevel(t *testing.T) {
+	r := &recorder{
+		fetch:   map[string]any{"workspace": "/w", "changed": []string{"pkg/b/b_test.go"}},
+		digests: map[string]string{"x/pkg/a": "d-a", "x/pkg/b": "d-b", "x/pkg/k": "d-kept"},
+		tests:   map[string]bool{"x/pkg/a": true, "x/pkg/b": false},
+	}
+	out, err := RunAsk(context.Background(), r.steps(), ask(delivery.CheckJob{Level: delivery.LevelUnit, Kept: true,
+		Runner: "workspace", RunnerWhen: "pkg/b/", Fetch: map[string]any{"repository": "novox/x", "commit": runCommit, "base": runCommit}}))
+	if err != nil {
+		t.Fatal(err)
+	}
+	got := strings.Join(r.calls[1:], "\n")
+	want := `digest {"level":"unit","runner":"workspace","workspace":"/w"}
+test {"level":"unit","packages":["x/pkg/b"],"runner":"workspace","workspace":"/w"}`
+	if got != want {
+		t.Fatalf("calls:\n%s\nwant:\n%s", got, want)
+	}
+	a := out.(delivery.CheckAnswer)
+	if len(a.Kept) != 1 || a.Kept[0].Package != "x/pkg/k" || a.Kept[0].Tests != 7 || len(a.Tests) != 1 || len(a.Digest) == 0 {
+		t.Fatalf("answer %+v", a)
+	}
+}
+
+// A Node check tests each module directory, the one whose tests changed first; a conflict ends the ask with the fetch
+// alone, a verdict for the asker.
+func TestRunAskChecksNodeModulesAndConflicts(t *testing.T) {
+	r := &recorder{fetch: map[string]any{"workspace": "/w", "modules": []string{"a", "b"}, "changed": []string{"b/test/x.test.ts"}},
+		tests: map[string]bool{"a": true, "b": true}}
+	s := r.steps()
+	s.Digest = nil
+	if _, err := RunAsk(context.Background(), s, ask(delivery.CheckJob{Level: delivery.LevelUnit, Fetch: map[string]any{"repository": "novox/x", "commit": runCommit}})); err != nil {
+		t.Fatal(err)
+	}
+	if fmt.Sprint(r.calls[1:]) != `[test {"level":"unit","module":"b","workspace":"/w"} test {"level":"unit","module":"a","workspace":"/w"}]` {
+		t.Fatalf("calls %v", r.calls)
+	}
+	r = &recorder{fetch: map[string]any{"workspace": "/w", "conflicts": []string{"go.mod"}}}
+	out, err := RunAsk(context.Background(), r.steps(), ask(delivery.CheckJob{Level: delivery.LevelUnit, Fetch: map[string]any{"repository": "novox/x", "commit": runCommit}}))
+	if err != nil || len(r.calls) != 1 || len(out.(delivery.CheckAnswer).Tests) != 0 {
+		t.Fatalf("a conflict: %v %v", err, r.calls)
+	}
+}
@@ -1,5 +1,7 @@
 package buildstep

+import "fmt"
+
 // Refusal is a step that is not run on purpose, as against one that failed: the toolchain installed is not the
 // recipe's, the integration level's postgres major is not the one it is built for (novox/hq ADR 0469 §1, ADR 0470
 // §4; issue 1473). A step's run returns one and the runner ends the step refused with its text.
@@ -11,3 +13,6 @@ type Refusal struct {
 }

 func (r *Refusal) Error() string { return r.Text }
+
+// Refuse is a Refusal with its text formatted.
+func Refuse(format string, args ...any) *Refusal { return &Refusal{Text: fmt.Sprintf(format, args...)} }