RN15: the runner's read verbs show, list, stands, why-carried and kept, from its own states (issue 1966)

The runner's tools bundle serves the seat mesh-pipeline-runner's read verbs as tools named
<seat>.<verb>, 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

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 <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)
+	}
+}

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 <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)
+	}
+}

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, <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
+}

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)
+		}
+	}
+}

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"
+        ]
       }
     ]
   }