RN10 (widens): a pipeline's end names what its change widens, from the catalog's revisions and the grants part

The candidate (the pipeline's own revision) against the branch's newest through mesh-catalog's revision; serving parts by the holder of the runner's, catalog's, steps' and composer's seats; acting rules, credentials; grants through grants-for. A read that fails says could not run. Touches steps.go (Hearer.Widens field, say's pipeline-ended), main.go (hearer.Widens), module.json invokes.

Builds: design 47 G §71, §275, §335, §441, A-207, A-215, A-220

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
2026-11-12 06:11:47 +01:00
parent 50f3dde9f5
commit a8314676c9
9 changed files with 417 additions and 1 deletions
@@ -116,6 +116,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
fills := &pipeline.Fills{Read: storeFills(ctx, js, cred, time.Now), Logf: logger.Printf,
Say: func(event string, f seat.Fill) error { return say("mesh.seat."+seat.Seat+".event."+event, f) }}
hearer.Fills = fills
// A pipeline's end names what its change widens (§71, A-207), computed by this running runner from the catalog and the grants part.
hearer.Widens = widenerOf(calls.Call).Widens
hearer.LogLast = stepLog{ctx: ctx, js: js, module: cred.Module}.Last
hearer.Where = routerRecordOf(cred.Module, calls.Call, calls.Ask)
hearer.Failures = &pipeline.Failures{Findings: rv.Findings, Observer: cred.Module, Author: commitAuthor(calls.Call), Logf: logger.Printf}
@@ -0,0 +1,81 @@
package main
import (
"encoding/json"
"fmt"
"strconv"
"git.novox.be/novox/mesh-contracts/identity-and-access/go/seats/meshgrants"
catalog "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-sdk/go/buscalls"
"git.novox.be/novox/mesh-sdk/go/keys"
"git.novox.be/novox/mesh-sdk/go/seatlist"
"mesh-delivery-pipeline/internal/pipeline"
)
// widenerOf is the runner's computation of `widens` (design 47 G §71, A-207, A-215, A-220) over its calls: the catalog seat's
// `revision` for the candidate and the branch's newest, the holders from the seat holders' one reading, and the grants part's
// `grants-for`; no other verb.
func widenerOf(call func(key string, body any) (json.RawMessage, error)) *pipeline.Widener {
return &pipeline.Widener{
Revision: catalogRevision(call),
Holder: func(s string) (string, error) { return seatlist.HolderOf(call, s) },
GrantsFor: func(branch string, source int64) (meshgrants.GrantsFor, error) {
return grantsFor(call, branch, source)
},
}
}
// catalogRevision reads one revision of a branch whole, page by page; revision 0 asks the newest (§195).
func catalogRevision(call func(string, any) (json.RawMessage, error)) func(string, int64) (pipeline.Half, error) {
return func(branch string, revision int64) (pipeline.Half, error) {
in := catalog.RevisionInput{Branch: keys.EscapeName(branch), Revision: revision}
var half pipeline.Half
for {
raw, err := call("seat:"+catalog.Seat+"."+catalog.VerbRevision, in)
if err != nil {
return pipeline.Half{}, err
}
answer, said, ok := buscalls.Unwrap(raw)
if !ok {
return pipeline.Half{}, fmt.Errorf("%s answered: %s", catalog.VerbRevision, said)
}
var a catalog.RevisionAnswer
if err := json.Unmarshal(answer, &a); err != nil {
return pipeline.Half{}, fmt.Errorf("%s answered no revision: %w", catalog.VerbRevision, err)
}
if a.NoneYet || a.Revision == nil {
return pipeline.Half{}, fmt.Errorf("%s holds no revision %d of %s", catalog.VerbRevision, revision, branch)
}
if half.Entries == nil {
half = pipeline.Half{Revision: a.Revision.Revision, Source: a.Revision.Source, Entries: map[string]catalog.CatalogEntry{}}
in.Revision = a.Revision.Revision // the pages of one revision, though a newer one is said between them
}
for _, e := range a.Entries {
half.Entries[e.Module] = e
}
if a.Next == "" {
return half, nil
}
in.After = a.Next
}
}
}
// grantsFor asks the grants part what it would compose for the candidate's source revision on its branch.
func grantsFor(call func(string, any) (json.RawMessage, error), branch string, source int64) (meshgrants.GrantsFor, error) {
raw, err := call("seat:"+meshgrants.Seat+"."+meshgrants.VerbGrantsFor, meshgrants.GrantsForInput{Branch: branch, Source: strconv.FormatInt(source, 10)})
if err != nil {
return meshgrants.GrantsFor{}, err
}
answer, said, ok := buscalls.Unwrap(raw)
if !ok {
return meshgrants.GrantsFor{}, fmt.Errorf("%s answered: %s", meshgrants.VerbGrantsFor, said)
}
var g meshgrants.GrantsFor
if err := json.Unmarshal(answer, &g); err != nil {
return meshgrants.GrantsFor{}, fmt.Errorf("%s answered no grants: %w", meshgrants.VerbGrantsFor, err)
}
return g, nil
}
@@ -0,0 +1,35 @@
package main
import (
"encoding/json"
"strings"
"testing"
)
// The runner reads a revision whole, page by page, of the revision the first page named, and the candidate by its number; the grants
// part is asked with the source revision (design 47 G §195, §441, A-207).
func TestWidenerReadsARevisionWholeAndAsksGrantsFor(t *testing.T) {
var asked []string
call := func(key string, body any) (json.RawMessage, error) {
raw, _ := json.Marshal(body)
asked = append(asked, key+" "+string(raw))
switch {
case strings.HasSuffix(key, ".grants-for"):
return json.RawMessage(`{"modules":[{"module":"app","added":[{"subject":"s","verb":"v"}],"removed":[]}]}`), nil
case strings.Contains(string(raw), `"after":"a"`):
return json.RawMessage(`{"revision":{"revision":8,"kind":"trunk","entries":[]},"entries":[{"module":"b","commit":"c","repository":"r","manifest":{}}]}`), nil
}
return json.RawMessage(`{"revision":{"revision":8,"kind":"trunk","entries":[]},"next":"a","entries":[{"module":"a","commit":"c","repository":"r","manifest":{}}]}`), nil
}
half, err := catalogRevision(call)("feat/x", 0)
if err != nil || half.Revision != 8 || len(half.Entries) != 2 {
t.Fatalf("half %+v, %v", half, err)
}
if !strings.Contains(asked[0], `"branch":"feat`) || !strings.Contains(asked[1], `"revision":8`) || strings.Contains(asked[0], `"revision"`) {
t.Errorf("asked %v: the first page names no revision, the next the one the first answered", asked)
}
g, err := grantsFor(call, "main", 31)
if err != nil || len(g.Modules) != 1 || !strings.Contains(asked[len(asked)-1], `"source":"31"`) {
t.Errorf("grants %+v, %v, asked %v", g, err, asked)
}
}
+2
View File
@@ -32,3 +32,5 @@ require (
require git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants v0.0.0-00010101000000-000000000000
require git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery v0.0.0-00010101000000-000000000000
require git.novox.be/novox/mesh-contracts/identity-and-access/go/seats/meshgrants v0.0.0-00010101000000-000000000000
@@ -115,6 +115,8 @@ type Hearer struct {
Ager *Ager
// Fills says the stores' fill at each step's end (§366); nil says none.
Fills *Fills
// Widens computes the widens a pipeline-ended carries (§71, A-207); nil carries none.
Widens func(seat.Pipeline) []seat.Widening
}
func (h *Hearer) now() time.Time {
@@ -254,7 +256,11 @@ func (h *Hearer) say(event string, p seat.Pipeline, step *seat.StepResult) error
return h.Say(subject, seat.StepEvent{V: 1, At: &at, Pipeline: p.Pipeline, Branch: p.Branch, Module: p.Module, Commit: p.Commit,
Fingerprint: p.Fingerprint, Group: p.Group, Revision: p.Revision, Source: p.Source, Result: *step})
case seat.EventPipelineEnded:
return h.Say(subject, pipelineEnded(p))
e := pipelineEnded(p)
if h.Widens != nil {
e.Widens = h.Widens(p)
}
return h.Say(subject, e)
default:
return h.Say(subject, seat.Wait{V: 1, Pipeline: p.Pipeline, Branch: p.Branch, Module: p.Module, Commit: p.Commit, Fingerprint: p.Fingerprint,
Group: p.Group, Revision: p.Revision, Source: p.Source, Step: seat.StepResultStepApproval, On: "operator", Since: at})
@@ -0,0 +1,179 @@
package pipeline
import (
"fmt"
"sort"
"time"
"git.novox.be/novox/mesh-contracts/identity-and-access/go/seats/meshgrants"
catalog "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner"
)
// ComposerSeat is the seat of the composer, the node-engine's mesh-compose (design 47 G A-167, A-220): a part serving a step.
const ComposerSeat = "node-engine"
// Widener computes a pipeline's `widens` (design 47 G §71, §275, §335, §441, A-207, A-215): the pipeline's own revision, the
// candidate, against the branch's newest revision, both read through the catalog seat's `revision`, and the permissions from the
// grants part's `grants-for`; no other verb. A widening is said, never refused: a read that fails says `could not run`.
type Widener struct {
// Revision reads one catalog revision of a watched branch whole; revision 0 is the branch's newest.
Revision func(branch string, revision int64) (Half, error)
// Holder is the module that holds a seat, from the seat holders' one reading; an error when it cannot be read or none holds it.
Holder func(seat string) (string, error)
// GrantsFor asks the seat mesh-grants' verb grants-for for the candidate's source revision on the branch.
GrantsFor func(branch string, source int64) (meshgrants.GrantsFor, error)
}
// Widens names each part serving a step the change alters (the holders of the runner's seat, the catalog's, each of the pipeline's
// steps' and the composer's, whose entry's fingerprint moved), each acting rule of the pipeline's module that is gained, matches
// more or is held longer, each own secret or credential provision it names that it did not, and each grant the grants part adds.
func (w *Widener) Widens(p seat.Pipeline) []seat.Widening {
if w == nil || w.Revision == nil {
return nil
}
could := func(name, said string) []seat.Widening {
return []seat.Widening{{What: seat.WideningWhatCouldNotRun, Name: name, Said: said}}
}
cand, err := w.Revision(p.Branch, p.Revision)
if err != nil {
return could("mesh-catalog", fmt.Sprintf("the candidate's revision %d is not read: %v", p.Revision, err))
}
newest, err := w.Revision(p.Branch, 0)
if err != nil {
return could("mesh-catalog", fmt.Sprintf("the branch's newest revision is not read: %v", err))
}
var out []seat.Widening
for _, s := range servingSeats(p) {
holder, err := w.Holder(s)
if err != nil {
out = append(out, seat.Widening{What: seat.WideningWhatCouldNotRun, Name: s, Said: fmt.Sprintf("the holder of %s is not read: %v", s, err)})
continue
}
before, after := newest.Entries[holder], cand.Entries[holder]
if after.Fp != "" && after.Fp != before.Fp {
out = append(out, seat.Widening{What: seat.WideningWhatServingPart, Name: holder,
Said: fmt.Sprintf("serves %s; its fingerprint is %s, was %s", s, after.Fp, before.Fp)})
}
}
out = append(out, ruleWidens(p.Module, newest.Entries[p.Module], cand.Entries[p.Module])...)
out = append(out, credentialWidens(p.Module, newest.Entries[p.Module], cand.Entries[p.Module])...)
if w.GrantsFor != nil {
g, err := w.GrantsFor(p.Branch, p.Source)
switch {
case err != nil:
out = append(out, seat.Widening{What: seat.WideningWhatCouldNotRun, Name: "mesh-grants", Said: err.Error()})
case g.Error != "":
out = append(out, seat.Widening{What: seat.WideningWhatCouldNotRun, Name: "mesh-grants", Said: g.Error})
default:
for _, m := range g.Modules {
if m.Module != p.Module {
continue
}
for _, a := range m.Added {
out = append(out, seat.Widening{What: seat.WideningWhatGrant, Name: m.Module + " " + a.Subject + " " + a.Verb,
Said: "the grants part would compose it, and the branch's newest does not hold it"})
}
}
}
}
order := map[string]int{seat.WideningWhatServingPart: 0, seat.WideningWhatActingRule: 1, seat.WideningWhatCredential: 2,
seat.WideningWhatGrant: 3, seat.WideningWhatCouldNotRun: 4}
sort.SliceStable(out, func(i, j int) bool {
if order[out[i].What] != order[out[j].What] {
return order[out[i].What] < order[out[j].What]
}
return out[i].Name < out[j].Name
})
return out
}
// servingSeats are the seats whose holders serve a step of p (A-220): the runner's, the catalog's, the composer's and each step's.
func servingSeats(p seat.Pipeline) []string {
set := map[string]bool{seat.Seat: true, catalog.Seat: true, ComposerSeat: true}
for _, s := range p.Steps {
if s.Seat != "" {
set[s.Seat] = true
}
}
var out []string
for s := range set {
out = append(out, s)
}
sort.Strings(out)
return out
}
// ruleWidens names each acting rule of the candidate's entry that the newest entry has not, or that now names more entries
// (`every` where it named a language) or holds longer (A-215).
func ruleWidens(module string, before, after catalog.CatalogEntry) []seat.Widening {
on := func(s string) string {
if s == "" {
return "every"
}
return s
}
var out []seat.Widening
for _, r := range after.Acts {
name := module + " " + r.Seat + "." + r.Step
var was *catalog.ActingRule
for i := range before.Acts {
if before.Acts[i].Seat == r.Seat && before.Acts[i].Step == r.Step {
was = &before.Acts[i]
}
}
switch {
case was == nil:
out = append(out, seat.Widening{What: seat.WideningWhatActingRule, Name: name, Said: "the rule is gained"})
case on(r.On) == "every" && on(was.On) != "every":
out = append(out, seat.Widening{What: seat.WideningWhatActingRule, Name: name, Said: "the rule acts on every entry, it acted on " + was.On})
case longer(r.Bound, was.Bound):
out = append(out, seat.Widening{What: seat.WideningWhatActingRule, Name: name, Said: "the bound is " + r.Bound + ", it was " + was.Bound})
}
}
return out
}
func longer(a, b string) bool {
x, err1 := time.ParseDuration(a)
y, err2 := time.ParseDuration(b)
return err1 == nil && err2 == nil && x > y
}
// credentialWidens names each own secret and each provision with a credential of its own that the candidate's manifest names and
// the newest entry's did not (A-215).
func credentialWidens(module string, before, after catalog.CatalogEntry) []seat.Widening {
was := map[string]bool{}
for _, n := range credentialsOf(before.Manifest) {
was[n] = true
}
var out []seat.Widening
for _, n := range credentialsOf(after.Manifest) {
if !was[n] {
out = append(out, seat.Widening{What: seat.WideningWhatCredential, Name: module + " " + n, Said: "the manifest names it, and the branch's newest does not"})
}
}
return out
}
// credentialsOf are the names of the own secrets a manifest declares and of the provisions that carry a credential: an own secret
// of the provider, one issued by software or one only a person gives.
func credentialsOf(manifest map[string]any) []string {
var out []string
if secrets, ok := manifest["secrets"].(map[string]any); ok {
for n := range secrets {
out = append(out, n)
}
}
if provides, ok := manifest["provides"].([]any); ok {
for _, o := range provides {
if m, ok := o.(map[string]any); ok && m["credential"] != nil {
if n, ok := m["name"].(string); ok {
out = append(out, n)
}
}
}
}
sort.Strings(out)
return out
}
@@ -0,0 +1,108 @@
package pipeline
import (
"errors"
"strings"
"testing"
"git.novox.be/novox/mesh-contracts/identity-and-access/go/seats/meshgrants"
catalog "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner"
)
func entry(module, fp string, acts []catalog.ActingRule, manifest map[string]any) catalog.CatalogEntry {
return catalog.CatalogEntry{Module: module, Fp: fp, Acts: acts, Manifest: manifest}
}
func halfOf(rev int64, es ...catalog.CatalogEntry) Half {
h := Half{Revision: rev, Entries: map[string]catalog.CatalogEntry{}}
for _, e := range es {
h.Entries[e.Module] = e
}
return h
}
// widens compares the pipeline's own revision with the branch's newest: the serving parts found by the holder of the seats
// (the runner, the catalog, each step's seat, the node-engine) by fingerprint, an acting rule gained, matching more or held longer, a
// credential named that was not, and the grants the grants part adds; each is said by name, a grants part that does not answer says
// `could not run` and nothing is refused (design 47 G §71, §275, §441, A-207, A-215, A-220).
func TestWidensIsTheCandidateAgainstTheNewest(t *testing.T) {
p := seat.Pipeline{Pipeline: "main.8.app", Module: "app", Branch: "main", Source: 31, Revision: 8,
Steps: []seat.StepResult{{Step: "unit", Seat: "node-build-go"}, {Step: "lint", Seat: "node-build-go"}, {Step: "approval"}}}
holders := map[string]string{seat.Seat: "runner", catalog.Seat: "indexer", "node-build-go": "build-go", "node-engine": "host"}
rule := func(on, bound string) []catalog.ActingRule {
return []catalog.ActingRule{{Seat: "node-build-go", Step: "check", On: on, Bound: bound}}
}
newest := halfOf(9,
entry("runner", "fp-1", nil, nil), entry("indexer", "fp-1", nil, nil), entry("build-go", "fp-1", nil, nil), entry("host", "fp-1", nil, nil),
entry("app", "fp-1", rule("go", "10m"), map[string]any{"secrets": map[string]any{"token": "/x"}, "provides": []any{"db"}}))
cases := []struct {
name string
cand Half
grants meshgrants.GrantsFor
grantsErr error
want []string
}{
{"nothing changes", newest, meshgrants.GrantsFor{}, nil, nil},
{"a serving part's fingerprint moves", halfOf(8, entry("build-go", "fp-2", nil, nil), entry("host", "fp-1", nil, nil), entry("indexer", "fp-1", nil, nil),
entry("runner", "fp-1", nil, nil), newest.Entries["app"]), meshgrants.GrantsFor{}, nil, []string{"serving part build-go"}},
{"an acting rule is gained", halfOf(8, entry("app", "fp-1", append(rule("go", "10m"), catalog.ActingRule{Seat: "x", Step: "grant"}), newest.Entries["app"].Manifest)),
meshgrants.GrantsFor{}, nil, []string{"acting rule app x.grant"}},
{"an acting rule matches more", halfOf(8, entry("app", "fp-1", rule("every", "10m"), newest.Entries["app"].Manifest)), meshgrants.GrantsFor{}, nil,
[]string{"acting rule app node-build-go.check"}},
{"an acting rule is held longer", halfOf(8, entry("app", "fp-1", rule("go", "1h"), newest.Entries["app"].Manifest)), meshgrants.GrantsFor{}, nil,
[]string{"acting rule app node-build-go.check"}},
{"an acting rule is held shorter", halfOf(8, entry("app", "fp-1", rule("go", "5m"), newest.Entries["app"].Manifest)), meshgrants.GrantsFor{}, nil, nil},
{"a secret and a credential provision are named", halfOf(8, entry("app", "fp-1", rule("go", "10m"), map[string]any{
"secrets": map[string]any{"token": "/x", "key": map[string]any{"path": "/k"}},
"provides": []any{"db", map[string]any{"name": "vault", "credential": map[string]any{"own": "key"}}}})),
meshgrants.GrantsFor{}, nil, []string{"credential app key", "credential app vault"}},
{"a grant is added", newest, meshgrants.GrantsFor{Modules: []meshgrants.GrantsForModulesItem{
{Module: "app", Added: []meshgrants.Grant{{Subject: "mesh.seat.x", Verb: "do"}}}, {Module: "other", Added: []meshgrants.Grant{{Subject: "s", Verb: "v"}}}}},
nil, []string{"grant app mesh.seat.x do"}},
{"the grants part does not answer", newest, meshgrants.GrantsFor{}, errors.New("no responders"), []string{"could not run mesh-grants"}},
{"the grants part cannot read the candidate", newest, meshgrants.GrantsFor{Error: "the catalog half is not held"}, nil, []string{"could not run mesh-grants"}},
}
for _, c := range cases {
w := &Widener{
Revision: func(_ string, rev int64) (Half, error) {
if rev == 0 {
return newest, nil
}
if rev != 8 {
t.Errorf("%s: the candidate asked as revision %d", c.name, rev)
}
return c.cand, nil
},
Holder: func(s string) (string, error) { return holders[s], nil },
GrantsFor: func(string, int64) (meshgrants.GrantsFor, error) { return c.grants, c.grantsErr },
}
var got []string
for _, x := range w.Widens(p) {
got = append(got, x.What+" "+x.Name)
}
if strings.Join(got, ";") != strings.Join(c.want, ";") {
t.Errorf("%s: widens %q, want %q", c.name, got, c.want)
}
}
}
// pipeline-ended carries the widens the runner computed, and says it whether or not anything widens (design 47 G §71, A-207).
func TestPipelineEndedCarriesWidens(t *testing.T) {
var body any
h := &Hearer{Say: func(_ string, b any) error { body = b; return nil },
Widens: func(p seat.Pipeline) []seat.Widening {
return []seat.Widening{{What: seat.WideningWhatGrant, Name: p.Module + " s v"}}
}}
if err := h.say(seat.EventPipelineEnded, seat.Pipeline{Module: "app", Outcome: seat.PipelineOutcomePassed}, nil); err != nil {
t.Fatal(err)
}
if e, ok := body.(seat.PipelineEnded); !ok || len(e.Widens) != 1 || e.Widens[0].Name != "app s v" {
t.Errorf("pipeline-ended says %+v", body)
}
h.Widens = nil
_ = h.say(seat.EventPipelineEnded, seat.Pipeline{Module: "app"}, nil)
if e := body.(seat.PipelineEnded); len(e.Widens) != 0 {
t.Errorf("a runner with no widener said %v", e.Widens)
}
}
@@ -84,6 +84,8 @@ func TestManifest(t *testing.T) {
{"invokes", equalSets(m.Invokes, []string{"seat:mesh-catalog.registrations", "seat:mesh-catalog.revision", "seat:mesh-hand-acts.hand-act",
// RN3 (§53, §334): a publication's dependents behind.
"seat:mesh-catalog.outdated",
// RN10 (§71, §441, A-207): the permissions a change widens, asked of the grants part.
"seat:mesh-grants.grants-for",
// RN11 (§33, A-191): the author of a failed head, read as the forge records it.
"seat:git.pull",
// A-74, A-89: who holds the seat operator-channel, whose state asks holds the router's record of the approval ask.
@@ -54,6 +54,7 @@
"seat:mesh-catalog.outdated",
"seat:git.pull",
"seat:mesh-module-assigner.assignments",
"seat:mesh-grants.grants-for",
"seat:mesh-builds.step",
"seat:mesh-wants.step",
"seat:node-build-go.check.step",