From fc0cb1873ace3a657ae6ed9853d8ed6638730db7 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 11:37:51 +0100 Subject: [PATCH] RN15: the runner's read verbs show, list, stands, why-carried and kept, from its own states (issue 1966) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The runner's tools bundle serves the seat mesh-pipeline-runner's read verbs as tools named ., answered from its own states pipelines and kept, never another part's: show one pipeline whole with its steps and their outputs' names, list by module, branch, commit or run newest first, stands every step started and not ended with what it waits on and since when, why-carried the pipeline a carried step came from, kept the kept results by level, digest or package. Builds: design 47 G §83, §306, §75 (stands), §25 (why-carried) Not built, waiting on design answers: show's step log (Q-RN6, the step log's section form) and each store's fill against its cap in show and kept (Q-RN2, where the caps are declared). Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- .../cmd/mesh-delivery-pipeline-tools/main.go | 91 +++++++++ .../mesh-delivery-pipeline-tools/main_test.go | 26 +++ .../internal/pipeline/read.go | 191 ++++++++++++++++++ .../internal/pipeline/read_test.go | 129 ++++++++++++ modules/mesh-delivery-pipeline/module.json | 53 ++++- 5 files changed, 482 insertions(+), 8 deletions(-) create mode 100644 modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main.go create mode 100644 modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main_test.go create mode 100644 modules/mesh-delivery-pipeline/internal/pipeline/read.go create mode 100644 modules/mesh-delivery-pipeline/internal/pipeline/read_test.go 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" + ] } ] }