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

This commit is contained in:
2026-11-11 11:41:25 +01:00
5 changed files with 482 additions and 8 deletions
@@ -0,0 +1,91 @@
// Command mesh-delivery-pipeline-tools serves the seat mesh-pipeline-runner's verbs as tools named <seat>.<verb>
// (design 47 G §83, §306), answered from the runner's own states; the tool runner describes each from the seat's
// definition, so the tools carry no description or input.
package main
import (
"encoding/json"
"fmt"
"os"
seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner"
stdio "git.novox.be/novox/mesh-sdk/go"
"mesh-delivery-pipeline/internal/pipeline"
)
// kept reads one of the runner's states through the runtime.
type kept struct{ s stdio.KeptState }
func (k kept) Keys() ([]string, error) { return k.s.Keys() }
func (k kept) Value(key string) (json.RawMessage, error) {
e, err := k.s.Get(key)
if err != nil || e == nil {
return nil, err
}
return e.Value, nil
}
// as decodes a call's arguments into a verb's input, refusing one that does not fit it.
func as[T any](args map[string]any) (T, error) {
var in T
raw, err := json.Marshal(args)
if err == nil {
err = json.Unmarshal(raw, &in)
}
if err != nil {
return in, fmt.Errorf("the arguments do not fit the verb's input: %w", err)
}
return in, nil
}
// Tools are the runner's read verbs over its states pipelines and kept.
func Tools(r pipeline.Reads) []stdio.Tool {
runs := map[string]func(map[string]any) (any, error){
seat.VerbShow: func(a map[string]any) (any, error) {
in, err := as[seat.ShowInput](a)
if err != nil {
return nil, err
}
return r.Show(in)
},
seat.VerbList: func(a map[string]any) (any, error) {
in, err := as[seat.ListInput](a)
if err != nil {
return nil, err
}
return r.List(in)
},
seat.VerbStands: func(map[string]any) (any, error) { return r.Stands() },
seat.VerbWhyCarried: func(a map[string]any) (any, error) {
in, err := as[seat.WhyCarriedInput](a)
if err != nil {
return nil, err
}
return r.WhyCarried(in)
},
seat.VerbKept: func(a map[string]any) (any, error) {
in, err := as[seat.KeptInput](a)
if err != nil {
return nil, err
}
return r.KeptResults(in)
},
}
var out []stdio.Tool
for _, v := range seat.Verbs() {
if run, ok := runs[v]; ok {
out = append(out, stdio.Tool{Name: seat.Seat + "." + v, Run: run})
}
}
return out
}
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 {
fmt.Fprintf(os.Stderr, "[mesh-delivery-pipeline-tools] %v\n", err)
os.Exit(1)
}
}
@@ -0,0 +1,26 @@
package main
import (
"encoding/json"
"strings"
"testing"
"mesh-delivery-pipeline/internal/pipeline"
)
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).
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"
if got := strings.Join(names, " "); got != want {
t.Errorf("tools %q, want %q", got, want)
}
}
@@ -0,0 +1,191 @@
// Package pipeline is the pipeline runner's own logic, holder of the seat mesh-pipeline-runner (design 47 G §21). It names
// no module, branch or repository (§7): every name comes from what it reads.
package pipeline
import (
"encoding/json"
"fmt"
"sort"
"strings"
"time"
seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner"
)
// Kept is one of the runner's old state sections as it reads them: every key, and one key's value (nil when absent).
type Kept interface {
Keys() ([]string, error)
Value(key string) (json.RawMessage, error)
}
// Reads answers the runner's read verbs (§83, §306) from its own states pipelines and kept, never another part's.
type Reads struct {
Pipelines Kept
Kept Kept
}
// Show is one pipeline whole, with its steps and the names of their outputs in step-outputs (§83).
func (r Reads) Show(in seat.ShowInput) (seat.ShowAnswer, error) {
p, ok, err := r.pipeline(in.Pipeline)
if err != nil {
return seat.ShowAnswer{}, err
}
if !ok {
return seat.ShowAnswer{}, fmt.Errorf("no pipeline %q in %s", in.Pipeline, seat.StatePipelines)
}
a := seat.ShowAnswer{Pipeline: p}
for _, s := range p.Steps {
if s.Output != "" {
a.Outputs = append(a.Outputs, s.Output)
}
}
return a, nil
}
// List is the pipelines and runs of a module, a branch, a commit or a run, newest started first, at most Limit when it
// names one (§83).
func (r Reads) List(in seat.ListInput) (seat.ListAnswer, error) {
all, err := r.pipelines()
if err != nil {
return seat.ListAnswer{}, err
}
out := []seat.Pipeline{}
for _, p := range all {
if (in.Module == "" || p.Module == in.Module) && (in.Branch == "" || p.Branch == in.Branch) &&
(in.Commit == "" || p.Commit == in.Commit) && (in.Run == "" || p.Run == in.Run) {
out = append(out, p)
}
}
sort.SliceStable(out, func(i, j int) bool { return startedOf(out[i]).After(startedOf(out[j])) })
if in.Limit > 0 && int64(len(out)) > in.Limit {
out = out[:in.Limit]
}
return seat.ListAnswer{Pipelines: out}, nil
}
// Stands is every step that has started and not ended, with what it waits on and since when (§75, §83): the watchman
// seeds its open waits from it.
func (r Reads) Stands() (seat.StandsAnswer, error) {
all, err := r.pipelines()
if err != nil {
return seat.StandsAnswer{}, err
}
out := []seat.Wait{}
for _, p := range all {
for _, s := range p.Steps {
if s.Ended != nil || s.Started == nil {
continue
}
if s.Outcome != seat.StepResultOutcomeRunning && s.Outcome != seat.StepResultOutcomeWaiting {
continue
}
out = append(out, seat.Wait{Pipeline: p.Pipeline, Module: p.Module, Branch: p.Branch, Commit: p.Commit,
Group: p.Group, Run: p.Run, Source: p.Source, Revision: p.Revision, Fp: p.Fp, Step: s.Step,
Artifact: s.Artifact, On: s.On, Since: *s.Started, V: 1})
}
}
sort.SliceStable(out, func(i, j int) bool { return out[i].Since.Before(out[j].Since) })
return seat.StandsAnswer{Stands: out}, nil
}
// WhyCarried is where a step's result was carried from (§25, §83): the pipeline it names.
func (r Reads) WhyCarried(in seat.WhyCarriedInput) (seat.WhyCarriedAnswer, error) {
p, ok, err := r.pipeline(in.Pipeline)
if err != nil {
return seat.WhyCarriedAnswer{}, err
}
if !ok {
return seat.WhyCarriedAnswer{}, fmt.Errorf("no pipeline %q in %s", in.Pipeline, seat.StatePipelines)
}
for _, s := range p.Steps {
if s.Step != in.Step || (in.Artifact != "" && s.Artifact != in.Artifact) {
continue
}
if s.Outcome != seat.StepResultOutcomeCarried {
return seat.WhyCarriedAnswer{}, fmt.Errorf("step %s of %s %s was not carried: it ended %q", s.Step,
in.Pipeline, s.Artifact, s.Outcome)
}
return seat.WhyCarriedAnswer{CarriedFrom: s.CarriedFrom}, nil
}
return seat.WhyCarriedAnswer{}, fmt.Errorf("pipeline %q has no step %s %s", in.Pipeline, in.Step, in.Artifact)
}
// KeptResults is the kept passing results, by level, digest or package (§83).
func (r Reads) KeptResults(in seat.KeptInput) (seat.KeptAnswer, error) {
keys, err := r.Kept.Keys()
if err != nil {
return seat.KeptAnswer{}, fmt.Errorf("%s cannot be read: %w", seat.StateKept, err)
}
sort.Strings(keys)
out := []seat.KeptResult{}
for _, k := range keys {
level, digest, ok := seat.ParseKeptByLevelDigestKey(k)
if !ok || (in.Level != "" && level != in.Level) || (in.Digest != "" && digest != in.Digest) {
continue
}
raw, err := r.Kept.Value(k)
if err != nil {
return seat.KeptAnswer{}, fmt.Errorf("%s %s cannot be read: %w", seat.StateKept, k, err)
}
if raw == nil {
continue
}
v, err := seat.DecodeKeptByLevelDigestValue(raw)
if err != nil {
return seat.KeptAnswer{}, fmt.Errorf("%s %s: %w", seat.StateKept, k, err)
}
if in.Package != "" && v.Package != in.Package {
continue
}
out = append(out, v)
}
return seat.KeptAnswer{Kept: out}, nil
}
func (r Reads) pipeline(key string) (seat.Pipeline, bool, error) {
key = strings.TrimSpace(key)
if key == "" {
return seat.Pipeline{}, false, fmt.Errorf("name the pipeline's key, <branch>.<revision>.<module>")
}
raw, err := r.Pipelines.Value(key)
if err != nil {
return seat.Pipeline{}, false, fmt.Errorf("%s %s cannot be read: %w", seat.StatePipelines, key, err)
}
if raw == nil {
return seat.Pipeline{}, false, nil
}
p, err := seat.DecodePipelinesByBranchRevisionModuleValue(raw)
if err != nil {
return seat.Pipeline{}, false, fmt.Errorf("%s %s: %w", seat.StatePipelines, key, err)
}
if p.Pipeline == "" {
p.Pipeline = key
}
return p, true, nil
}
func (r Reads) pipelines() ([]seat.Pipeline, error) {
keys, err := r.Pipelines.Keys()
if err != nil {
return nil, fmt.Errorf("%s cannot be read: %w", seat.StatePipelines, err)
}
sort.Strings(keys)
var out []seat.Pipeline
for _, k := range keys {
p, ok, err := r.pipeline(k)
if err != nil {
return nil, err
}
if ok {
out = append(out, p)
}
}
return out, nil
}
func startedOf(p seat.Pipeline) time.Time {
if p.Started == nil {
return time.Time{}
}
return *p.Started
}
@@ -0,0 +1,129 @@
package pipeline
import (
"encoding/json"
"strings"
"testing"
"time"
seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner"
)
type memKept map[string]json.RawMessage
func (m memKept) Keys() ([]string, error) {
var out []string
for k := range m {
out = append(out, k)
}
return out, nil
}
func (m memKept) Value(key string) (json.RawMessage, error) { return m[key], nil }
func put(t *testing.T, m memKept, key string, v any) {
t.Helper()
raw, err := json.Marshal(v)
if err != nil {
t.Fatal(err)
}
m[key] = raw
}
// TestReads holds the runner's read verbs to §83 and §306, one row per case (RN15/AC1, test 93's runner case).
func TestReads(t *testing.T) {
t0 := time.Date(2026, 11, 11, 10, 0, 0, 0, time.UTC)
t1, t2 := t0.Add(time.Minute), t0.Add(2*time.Minute)
pipes, kept := memKept{}, memKept{}
put(t, pipes, "main.7.alpha", seat.Pipeline{Pipeline: "main.7.alpha", Module: "alpha", Branch: "main", Commit: "c7",
Source: 7, Outcome: seat.PipelineOutcomeRunning, Started: &t0, Steps: []seat.StepResult{
{Step: seat.StepResultStepLint, Artifact: "alpha:program", Status: "required", Outcome: seat.StepResultOutcomePassed,
Started: &t0, Ended: &t1, Output: "main.7.alpha.lint.alpha:program"},
{Step: seat.StepResultStepUnit, Artifact: "alpha:program", Status: "required", Outcome: seat.StepResultOutcomeCarried,
CarriedFrom: "main.6.alpha"},
{Step: seat.StepResultStepApproval, Status: "required", Outcome: seat.StepResultOutcomeWaiting, On: seat.WaitOnPerson,
Started: &t1},
}})
put(t, pipes, "beta.3.alpha", seat.Pipeline{Pipeline: "beta.3.alpha", Module: "alpha", Branch: "beta", Commit: "b3",
Outcome: seat.PipelineOutcomePassed, Started: &t2})
put(t, pipes, "main.7.gamma", seat.Pipeline{Pipeline: "main.7.gamma", Module: "gamma", Branch: "main", Commit: "c7",
Outcome: seat.PipelineOutcomeRunning, Started: &t1})
put(t, kept, "unit.sha256-aa", seat.KeptResult{Level: "unit", Digest: "sha256-aa", Package: "alpha/x"})
put(t, kept, "integration.sha256-bb", seat.KeptResult{Level: "integration", Digest: "sha256-bb", Package: "alpha/y"})
r := Reads{Pipelines: pipes, Kept: kept}
names := func(ps []seat.Pipeline) string {
var s []string
for _, p := range ps {
s = append(s, p.Pipeline)
}
return strings.Join(s, " ")
}
rows := []struct {
name string
got func() (string, error)
want string
}{
{"show answers the pipeline whole and its outputs", func() (string, error) {
a, err := r.Show(seat.ShowInput{Pipeline: "main.7.alpha"})
return a.Pipeline.Module + " " + strings.Join(a.Outputs, ","), err
}, "alpha main.7.alpha.lint.alpha:program"},
{"show refuses a pipeline it does not hold", func() (string, error) {
_, err := r.Show(seat.ShowInput{Pipeline: "main.9.alpha"})
return "", err
}, `error: no pipeline "main.9.alpha" in pipelines`},
{"list by module, newest first", func() (string, error) {
a, err := r.List(seat.ListInput{Module: "alpha"})
return names(a.Pipelines), err
}, "beta.3.alpha main.7.alpha"},
{"list by commit", func() (string, error) {
a, err := r.List(seat.ListInput{Commit: "c7"})
return names(a.Pipelines), err
}, "main.7.gamma main.7.alpha"},
{"list by branch with a limit", func() (string, error) {
a, err := r.List(seat.ListInput{Branch: "main", Limit: 1})
return names(a.Pipelines), err
}, "main.7.gamma"},
{"stands names a waiting step and since when", func() (string, error) {
a, err := r.Stands()
var s []string
for _, w := range a.Stands {
s = append(s, w.Pipeline+" "+w.Step+" "+w.On+" "+w.Since.Format("15:04"))
}
return strings.Join(s, ";"), err
}, "main.7.alpha approval person 10:01"},
{"why-carried names where a carried step came from", func() (string, error) {
a, err := r.WhyCarried(seat.WhyCarriedInput{Pipeline: "main.7.alpha", Step: "unit", Artifact: "alpha:program"})
return a.CarriedFrom, err
}, "main.6.alpha"},
{"why-carried refuses a step that ran", func() (string, error) {
_, err := r.WhyCarried(seat.WhyCarriedInput{Pipeline: "main.7.alpha", Step: "lint"})
return "", err
}, `error: step lint of main.7.alpha alpha:program was not carried: it ended "passed"`},
{"kept by level", func() (string, error) {
a, err := r.KeptResults(seat.KeptInput{Level: "unit"})
var s []string
for _, k := range a.Kept {
s = append(s, k.Package)
}
return strings.Join(s, " "), err
}, "alpha/x"},
{"kept by package", func() (string, error) {
a, err := r.KeptResults(seat.KeptInput{Package: "alpha/y"})
var s []string
for _, k := range a.Kept {
s = append(s, k.Digest)
}
return strings.Join(s, " "), err
}, "sha256-bb"},
}
for _, row := range rows {
got, err := row.got()
if err != nil {
got = "error: " + err.Error()
}
if got != row.want {
t.Errorf("%s: got %q, want %q", row.name, got, row.want)
}
}
}
+45 -8
View File
@@ -35,17 +35,41 @@
"mesh-catalog.catalog-revision"
],
"state": [
{ "name": "pipelines", "history": 1 },
{ "name": "kept", "history": 1 },
{ "name": "aged", "history": 1 },
{ "name": "findings", "history": 1 },
{ "name": "first-views", "history": 1 }
{
"name": "pipelines",
"history": 1
},
{
"name": "kept",
"history": 1
},
{
"name": "aged",
"history": 1
},
{
"name": "findings",
"history": 1
},
{
"name": "first-views",
"history": 1
}
],
"streams": [
{ "name": "step-log", "events": ["step-log.**"], "max-mib": 128 }
{
"name": "step-log",
"events": [
"step-log.**"
],
"max-mib": 128
}
],
"objects": [
{ "name": "step-outputs", "max-mib": 1024 }
{
"name": "step-outputs",
"max-mib": 1024
}
],
"own-secrets": {
"broker": "${dir:state}/broker"
@@ -93,7 +117,20 @@
"system": "arch",
"from": "cmd/mesh-delivery-pipeline",
"binary": "mesh-delivery-pipeline",
"needs": ["mesh-delivery-assigner:program"]
"needs": [
"mesh-delivery-assigner:program"
]
},
{
"name": "tools",
"kind": "bundle",
"language": "go",
"system": "arch",
"from": "cmd/mesh-delivery-pipeline-tools",
"binary": "mesh-delivery-pipeline-tools",
"loads": [
"mesh-delivery-pipeline-tools"
]
}
]
}