diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main.go new file mode 100644 index 00000000..bebc4d19 --- /dev/null +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main.go @@ -0,0 +1,91 @@ +// Command mesh-delivery-pipeline-tools serves the seat mesh-pipeline-runner's verbs as tools named . +// (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) + } +} diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main_test.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main_test.go new file mode 100644 index 00000000..0565790a --- /dev/null +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main_test.go @@ -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 . (§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) + } +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/read.go b/modules/mesh-delivery-pipeline/internal/pipeline/read.go new file mode 100644 index 00000000..00163110 --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/read.go @@ -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, ..") + } + 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 +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/read_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/read_test.go new file mode 100644 index 00000000..af516211 --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/read_test.go @@ -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) + } + } +} diff --git a/modules/mesh-delivery-pipeline/module.json b/modules/mesh-delivery-pipeline/module.json index 02802158..3a618fe0 100644 --- a/modules/mesh-delivery-pipeline/module.json +++ b/modules/mesh-delivery-pipeline/module.json @@ -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" + ] } ] }