A-D10, A-D12, A-D13: the runner takes a step's end only as asked, keeps its results, and asks the approval step last, ended by its warrant

The hearer (internal/pipeline/steps.go) takes a build seat's step-ended only from the subject of the node that ran it, for an ask
the runner made of that seat, at the pipeline's commit and fingerprint; a taken end is written by compare-and-set, logged, its
output kept in step-outputs and its passing results in kept. A pipeline ends once every step ended, passed only when every
required step passed or was carried. The approval step is asked as operator-channel's accepted kind ask once every other required
step passed, and ended by the warrant read from the router's record (A-89), the decided event only waking the runner. The program
hears step-ended; the approval's Asker and decided handler wait for the answer on where a process finds the seat holder (A-74).

Builds: design 47 G §24, §26, §27, §37, §700 (A-89, A-74); recovers 18e10200 and 89687404 in today's form

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h

modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go

@@ -48,6 +48,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	if err != nil {
 		return err
 	}
+	kept, err := process.EssentialBucket(ctx, js, "pipeline runner", ownBucket(cred.Module, seat.StateKept), "keep a package's passing result")
+	if err != nil {
+		return err
+	}
 	// The registrations and each watched branch's catalog half, first views then events (§21, §195, §200): until a
 	// branch's view is held the runner acts on nothing new of it, and says so.
 	calls, err := process.Connect(cred, nats.Name(cred.Node+"."+cred.Module+".calls"), nats.RetryOnFailedConnect(true))
@@ -62,14 +66,17 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	}
 	// A pipeline starts by compare-and-set on its key's absence and says pipeline-started (§54, §103); a publish starts
 	// one per dependent behind (§53, §334).
-	starter := &pipeline.Starter{Pipelines: createdKV{ctx: ctx, kv: pipelines}, Logf: logger.Printf,
-		Say: func(subject string, body any) error {
-			data, err := json.Marshal(body)
-			if err == nil {
-				_, err = js.Publish(ctx, subject, data)
-			}
-			return err
-		}}
+	say := func(subject string, body any) error {
+		data, err := json.Marshal(body)
+		if err == nil {
+			_, err = js.Publish(ctx, subject, data)
+		}
+		return err
+	}
+	starter := &pipeline.Starter{Pipelines: createdKV{ctx: ctx, kv: pipelines}, Logf: logger.Printf, Say: say}
+	if err := hearStepEnded(hub, newHearer(ctx, js, cred, pipelines, kept, say, logger.Printf), logger.Printf); err != nil {
+		return err
+	}
 	if err := hearPublished(ctx, hub, starter, calls, rv.Catalog, logger.Printf); err != nil {
 		return err
 	}

modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/steps.go

@@ -0,0 +1,127 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+
+	"github.com/nats-io/nats.go/jetstream"
+
+	seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner"
+	"git.novox.be/novox/mesh-sdk/go/catalogue"
+	"git.novox.be/novox/mesh-sdk/go/keys"
+	"git.novox.be/novox/mesh-sdk/go/process"
+
+	"mesh-delivery-pipeline/internal/pipeline"
+)
+
+// pipelineStore is the state pipelines as the hearer reads and writes it: a write by compare-and-set on the revision read (§27).
+type pipelineStore struct {
+	ctx context.Context
+	kv  jetstream.KeyValue
+}
+
+func (s pipelineStore) Get(key string) (seat.Pipeline, uint64, error) {
+	e, err := s.kv.Get(s.ctx, key)
+	if err != nil {
+		return seat.Pipeline{}, 0, err
+	}
+	var p seat.Pipeline
+	if err := json.Unmarshal(e.Value(), &p); err != nil {
+		return seat.Pipeline{}, 0, fmt.Errorf("the pipeline %s is no pipeline: %w", key, err)
+	}
+	return p, e.Revision(), nil
+}
+
+func (s pipelineStore) Update(key string, p seat.Pipeline, revision uint64) (uint64, error) {
+	raw, err := json.Marshal(p)
+	if err != nil {
+		return 0, err
+	}
+	rev, err := s.kv.Update(s.ctx, key, raw, revision)
+	if err != nil && (errors.Is(err, jetstream.ErrKeyExists) || strings.Contains(err.Error(), "wrong last sequence")) {
+		return 0, pipeline.ErrChanged
+	}
+	return rev, err
+}
+
+func (s pipelineStore) Keys(filter string) ([]string, error) {
+	all, err := s.kv.Keys(s.ctx)
+	if errors.Is(err, jetstream.ErrNoKeysFound) {
+		return nil, nil
+	}
+	var out []string
+	want := strings.Split(filter, ".")
+	for _, k := range all {
+		got := strings.Split(k, ".")
+		if len(got) != len(want) {
+			continue
+		}
+		ok := true
+		for i := range want {
+			ok = ok && (want[i] == "*" || want[i] == got[i])
+		}
+		if ok {
+			out = append(out, k)
+		}
+	}
+	return out, err
+}
+
+// newHearer is the hearer over the runner's own states: pipelines, kept, the step log on its stream and step-outputs on its
+// objects (§27). Each store is opened where it is used, so one the grants part has not made yet fails that write loudly.
+func newHearer(ctx context.Context, js jetstream.JetStream, cred process.Account, pipelines, kept jetstream.KeyValue,
+	say func(string, any) error, logf func(string, ...any)) *pipeline.Hearer {
+	return &pipeline.Hearer{Store: pipelineStore{ctx: ctx, kv: pipelines}, Say: say, Logf: logf, Name: seat.Seat,
+		Log: func(e seat.LogEntry) error {
+			return say(cred.EventSubject("step-log."+keys.EscapeName(e.Pipeline)), e)
+		},
+		Keep: func(key string, k seat.KeptResult) error {
+			raw, err := json.Marshal(k)
+			if err == nil {
+				_, err = kept.Put(ctx, key, raw)
+			}
+			return err
+		},
+		Output: func(key string, data []byte) error {
+			o, err := js.ObjectStore(ctx, catalogue.ObjectsBucket(cred.Module, seat.StateStepOutputs))
+			if err != nil {
+				return fmt.Errorf("the objects %s are not open: %w", seat.StateStepOutputs, err)
+			}
+			_, err = o.PutBytes(ctx, key, data)
+			return err
+		}}
+}
+
+// hearStepEnded takes each build seat's step-ended into its pipeline (§24; ADR 0666 §21): refused and said, never taken,
+// when it is not as the runner asked it, and heard again when a write failed.
+func hearStepEnded(hub *process.EventHub, h *pipeline.Hearer, logf func(string, ...any)) error {
+	for _, bs := range pipeline.StepSeats {
+		bs := bs
+		err := hub.Handle(bs.Prefix+"*", func(m jetstream.Msg) {
+			var ev pipeline.StepEnded
+			if err := json.Unmarshal(m.Data(), &ev); err != nil {
+				logf("%s is no step-ended: %v", m.Subject(), err)
+				_ = m.Term()
+				return
+			}
+			_, err := h.OnStepEnded(bs, m.Subject(), ev)
+			switch {
+			case errors.Is(err, pipeline.ErrRefused):
+				logf("%v", err)
+				_ = m.Term()
+			case err != nil:
+				logf("the step-ended of ask %s is not taken yet, heard again: %v", ev.ID, err)
+				_ = m.Nak()
+			default:
+				_ = m.Ack()
+			}
+		})
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}

modules/mesh-delivery-pipeline/go.mod

@@ -13,6 +13,11 @@ require (
 	git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-00010101000000-000000000000
 	git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient v0.0.0-00010101000000-000000000000
 	git.novox.be/novox/mesh-delivery/contracts/go/clients/meshpipelinerunnerclient v0.0.0-00010101000000-000000000000
+	git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient v0.0.0-00010101000000-000000000000
+	git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel v0.0.0-00010101000000-000000000000
+	git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo v0.0.0-00010101000000-000000000000
+	git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildimage v0.0.0-00010101000000-000000000000
+	git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildnode v0.0.0-00010101000000-000000000000
 	git.novox.be/novox/mesh-delivery/contracts/go/seats/meshbuilds v0.0.0-00010101000000-000000000000
 	git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog v0.0.0-00010101000000-000000000000
 	github.com/klauspost/compress v1.20.1 // indirect

modules/mesh-delivery-pipeline/internal/pipeline/approval.go

@@ -0,0 +1,218 @@
+package pipeline
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	"git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient"
+	oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
+	seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner"
+	"git.novox.be/novox/mesh-sdk/go/keys"
+)
+
+// The approval step is an authorising ask of tier approve, the seat operator-channel's accepted kind ask under the
+// runner's own name, its about the module and the watched branch only, so a newer commit's ask replaces the older one's
+// (design 47 G §26, §37; ADR 0666 §10). It is asked last, once every other required step passed.
+const (
+	OptionApprove = "approve"
+	OptionDecline = "decline"
+)
+
+// About is what an approval ask is about: the module and the watched branch, each a key's name, joined by a dot.
+func About(module, branch string) string {
+	return keys.EscapeName(module) + "." + keys.EscapeName(branch)
+}
+
+// approvalDue is the index of the approval step to ask now (§26): waiting, not bypassed, every other step ended and every
+// other required step passed or carried; -1 when none is due.
+func approvalDue(p seat.Pipeline) int {
+	at := -1
+	for i, s := range p.Steps {
+		if s.Step == seat.StepResultStepApproval {
+			if s.Status != seat.StepResultStatusBypass && s.Outcome == seat.StepResultOutcomeWaiting {
+				at = i
+			}
+			continue
+		}
+		if s.Outcome == seat.StepResultOutcomeWaiting || s.Outcome == seat.StepResultOutcomeRunning {
+			return -1
+		}
+		if s.Status == seat.StepResultStatusRequired && s.Outcome != seat.StepResultOutcomePassed && s.Outcome != seat.StepResultOutcomeCarried {
+			return -1
+		}
+	}
+	return at
+}
+
+// ApprovalAsk is the ask of a pipeline's approval step. Its id and its expiry follow from the pipeline and from when the
+// step was asked, so the ask built again for the same pipeline is the same ask, digest included: a warrant is held to it.
+func ApprovalAsk(p seat.Pipeline, from time.Time) (oc.Ask, error) {
+	sum := sha256.Sum256([]byte(p.Pipeline + "\n" + p.Commit))
+	act := func(do string) (string, error) {
+		return operatorchannelclient.ActDigest(operatorchannelclient.Act{"pipeline": p.Pipeline, "commit": p.Commit,
+			"fingerprint": p.Fingerprint, "do": do})
+	}
+	approve, err := act(OptionApprove)
+	if err != nil {
+		return oc.Ask{}, err
+	}
+	decline, err := act(OptionDecline)
+	if err != nil {
+		return oc.Ask{}, err
+	}
+	a := oc.Ask{ID: "approval-" + hex.EncodeToString(sum[:12]), About: About(p.Module, p.Branch), Who: operatorchannelclient.Operator,
+		Headline:    trimRunes("Deliver "+p.Module+" from "+p.Branch+"?", operatorchannelclient.HeadlineLength),
+		Explanation: fmt.Sprintf("Every required step of %s's pipeline at commit %s on %s passed; its policy asks your approval before it is delivered.", p.Module, p.Commit, p.Branch),
+		Details:     "pipeline " + p.Pipeline + ", fingerprint " + p.Fingerprint,
+		Expires:     operatorchannelclient.FormatTime(from.Add(operatorchannelclient.ApproveLasts)),
+		OnExpiry:    "the step ends declined and the commit is not delivered; ask-again asks once more",
+		Options: []oc.Option{
+			{ID: OptionApprove, Label: "Approve", Does: "delivers this commit of " + p.Module, Level: operatorchannelclient.Approve, Binds: approve},
+			{ID: OptionDecline, Label: "Decline", Does: "ends the step declined; nothing is delivered", Level: operatorchannelclient.Approve, Binds: decline},
+		}}
+	return a, operatorchannelclient.Check(a, from)
+}
+
+func trimRunes(s string, n int) string {
+	r := []rune(s)
+	if len(r) <= n {
+		return s
+	}
+	return string(r[:n-1]) + "…"
+}
+
+// Asker is the runner's side of the operator-channel: Ask publishes an ask under the runner's name, its duplicate id the
+// ask's own, so an ask published twice is kept once.
+type Asker interface {
+	Ask(oc.Ask) error
+}
+
+// askApproval asks p's approval step when it is due, before the pipeline is written: an ask published and then not
+// written is published again, the same ask, when its step's end is heard again.
+func askApproval(key string, p *seat.Pipeline, from time.Time, asker Asker) (bool, error) {
+	i := approvalDue(*p)
+	if i < 0 || asker == nil {
+		return false, nil
+	}
+	a, err := ApprovalAsk(*p, from)
+	if err != nil {
+		return false, fmt.Errorf("the approval of %s cannot be asked: %w", key, err)
+	}
+	if err := asker.Ask(a); err != nil {
+		return false, fmt.Errorf("the approval of %s is not asked: %w", key, err)
+	}
+	s := &p.Steps[i]
+	s.Outcome, s.Ask, s.Seat, s.Started = seat.StepResultOutcomeRunning, a.ID, seat.Seat, &from
+	s.Why = "waiting on the operator"
+	return true, nil
+}
+
+// ErrNoAsk is a warrant naming no approval step the runner keeps for its about.
+var ErrNoAsk = errors.New("no approval step asked under the warrant's ask")
+
+// decide ends an approval step on its warrant: a chosen approve, held to the ask rebuilt from the step, ends it passed; a
+// chosen decline, an expiry, a cancel or a refusal ends it declined; a replacement by a newer commit's ask ends it
+// superseded (design 47 G §37). It says whether the step changed.
+func decide(p *seat.Pipeline, w oc.Warrant, asker string, now time.Time) (bool, error) {
+	for i := range p.Steps {
+		s := &p.Steps[i]
+		if s.Step != seat.StepResultStepApproval || s.Ask != w.Ask || s.Started == nil {
+			continue
+		}
+		if s.Outcome != seat.StepResultOutcomeRunning {
+			return false, nil
+		}
+		switch w.Outcome {
+		case oc.WarrantOutcomeChosen:
+			asked, err := ApprovalAsk(*p, *s.Started)
+			if err != nil {
+				return false, err
+			}
+			o, err := operatorchannelclient.For(w, asker, asked)
+			if err != nil {
+				return false, fmt.Errorf("the warrant for %s is not taken: %w", w.Ask, err)
+			}
+			s.Outcome = seat.StepResultOutcomeDeclined
+			if o.ID == OptionApprove {
+				s.Outcome = seat.StepResultOutcomePassed
+			}
+			if w.By != nil {
+				s.By = w.By.Identity
+			}
+			s.Why = "the operator chose " + o.ID
+		case oc.WarrantOutcomeReplaced:
+			s.Outcome, s.Why = seat.StepResultOutcomeSuperseded, "a newer commit's ask replaced it"
+		default:
+			s.Outcome, s.Why = seat.StepResultOutcomeDeclined, "the ask ended "+w.Outcome+" "+w.Words
+		}
+		s.Ended = &now
+		return true, nil
+	}
+	return false, ErrNoAsk
+}
+
+// ErrNoRecord is a decided event whose ask the router's record holds no warrant for yet: the event is heard again.
+var ErrNoRecord = errors.New("the router's record of the ask holds no warrant yet")
+
+// OnDecided takes the operator-channel's decided for the runner's own asks: it only wakes the runner (design 47 G A-89), which
+// reads the warrant from the router's record of the ask through Where, never from the event's body (§37, §700). The approval
+// step asked under that ask, among the pipelines of the module and watched branch the ask's about names, ends as decide says.
+func (h *Hearer) OnDecided(event oc.Warrant) (seat.Pipeline, error) {
+	if event.Asker != h.Name {
+		return seat.Pipeline{}, fmt.Errorf("%w: the decided event is %s's, not %s's", ErrRefused, event.Asker, h.Name)
+	}
+	if h.Where == nil {
+		return seat.Pipeline{}, fmt.Errorf("the runner reads no router's record, so the ask %s is not taken", event.Ask)
+	}
+	_, rec, err := h.Where(event.Ask)
+	if err != nil {
+		return seat.Pipeline{}, fmt.Errorf("the router's record of the ask %s cannot be read: %w", event.Ask, err)
+	}
+	if rec == nil {
+		return seat.Pipeline{}, fmt.Errorf("%w: %s", ErrNoRecord, event.Ask)
+	}
+	w := *rec
+	module, branch, ok := strings.Cut(w.About, ".")
+	if !ok {
+		return seat.Pipeline{}, fmt.Errorf("%w: the warrant for %s is about %q, not a module and a branch", ErrNoAsk, w.Ask, w.About)
+	}
+	found, err := h.Store.Keys(branch + ".*." + module)
+	if err != nil {
+		return seat.Pipeline{}, err
+	}
+	for _, key := range found {
+		for {
+			p, rev, err := h.Store.Get(key)
+			if err != nil {
+				return seat.Pipeline{}, err
+			}
+			now := h.now()
+			changed, err := decide(&p, w, h.Name, now)
+			if errors.Is(err, ErrNoAsk) {
+				break
+			}
+			if err != nil || !changed {
+				return p, err
+			}
+			ended := endPipeline(&p, now)
+			if _, err := h.Store.Update(key, p, rev); errors.Is(err, ErrChanged) {
+				continue
+			} else if err != nil {
+				return p, err
+			}
+			i := stepOfAsk(p, w.Ask)
+			h.logEnd(key, p.Steps[i], now)
+			for _, ev := range []string{seat.EventWaitEnded, seat.EventStepEnded} {
+				if err := h.say(ev, p, &p.Steps[i]); err != nil {
+					return p, fmt.Errorf("pipeline %s's %s is written but not said: %w", key, ev, err)
+				}
+			}
+			return p, h.said(p, false, ended)
+		}
+	}
+	return seat.Pipeline{}, fmt.Errorf("%w: %s about %s", ErrNoAsk, w.Ask, w.About)
+}

modules/mesh-delivery-pipeline/internal/pipeline/steps.go

@@ -0,0 +1,269 @@
+package pipeline
+
+import (
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
+	seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner"
+	nodebuildgo "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo"
+	nodebuildimage "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildimage"
+	nodebuildnode "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildnode"
+	"git.novox.be/novox/mesh-sdk/go/keys"
+)
+
+// StepSeat is a seat whose holder runs the lint and test steps (design 47 G §22, §24): where the runner hears its
+// step-ended, on the subject of the node that ran it.
+type StepSeat struct {
+	Seat string
+	// Prefix is the subject before <node>.
+	Prefix string
+}
+
+// StepSeats are the three build seats, from their contracts.
+var StepSeats = []StepSeat{
+	{Seat: nodebuildgo.Seat, Prefix: strings.TrimSuffix(nodebuildgo.SubjectStepEndedAll, "*")},
+	{Seat: nodebuildnode.Seat, Prefix: strings.TrimSuffix(nodebuildnode.SubjectStepEndedAll, "*")},
+	{Seat: nodebuildimage.Seat, Prefix: strings.TrimSuffix(nodebuildimage.SubjectStepEndedAll, "*")},
+}
+
+// StepEnded is a build seat's step-ended. The three seats' bodies are one form, the part build-step's, held equal by
+// their definitions, so the runner reads each in this one type.
+type StepEnded = nodebuildgo.StepEnded
+
+// ErrRefused is a step-ended the runner does not take: not the node's holder's subject, or naming an ask, a commit or a
+// fingerprint the runner did not ask (ADR 0666 §21; design 47 G §24). It is said, never taken.
+var ErrRefused = errors.New("a step's end the runner does not take")
+
+// ErrChanged is an Updater's answer to an Update whose revision is no longer the key's: another write came between.
+var ErrChanged = errors.New("the pipeline was written since it was read")
+
+// Updater is the state pipelines read and written by compare-and-set on the revision read (§27).
+type Updater interface {
+	Get(key string) (seat.Pipeline, uint64, error)
+	Update(key string, p seat.Pipeline, revision uint64) (uint64, error)
+	// Keys are the keys matching a filter, `*` one key part.
+	Keys(filter string) ([]string, error)
+}
+
+// Hearer takes the steps' ends into their pipelines and ends a pipeline whose every step ended (design 47 G §24, §27).
+type Hearer struct {
+	Store Updater
+	// Say says one of the runner's events of a pipeline, on its subject; Log appends to the step log.
+	Say func(subject string, body any) error
+	Log func(seat.LogEntry) error
+	// Keep writes one package's passing result into the state kept, under its key (§27): only from a step's end the
+	// runner took, so a kept result is its seat holder's alone.
+	Keep func(key string, k seat.KeptResult) error
+	// Output keeps a step's output under key in the objects step-outputs.
+	Output func(key string, data []byte) error
+	// Asker publishes an approval step's ask (§37); Name is the runner's own name, the asker a warrant names.
+	Asker Asker
+	Name  string
+	// Where reads the router's record of one of the runner's asks, as operatorchannelclient.Client.Where does: its state and
+	// its warrant (design 47 G A-89, §700).
+	Where func(ask string) (string, *oc.Warrant, error)
+	Logf  func(string, ...any)
+	Now   func() time.Time
+}
+
+func (h *Hearer) now() time.Time {
+	if h.Now != nil {
+		return h.Now().UTC()
+	}
+	return time.Now().UTC()
+}
+
+// OnStepEnded takes the step-ended heard on subject, from seat's holders. It is refused, and said, when the subject's
+// node is not the one the body names, when no pipeline the runner keeps names the ask, or when the ask's step names
+// another seat, commit, fingerprint, module or step (ADR 0666 §21; the bus grants step-ended.<node> to that node's
+// holder alone, so the subject's node is the account's). A step already ended takes nothing (the bus gives an event again).
+func (h *Hearer) OnStepEnded(bs StepSeat, subject string, ev StepEnded) (seat.Pipeline, error) {
+	node := strings.TrimPrefix(subject, bs.Prefix)
+	if node == subject || node == "" || strings.Contains(node, ".") {
+		return seat.Pipeline{}, fmt.Errorf("%w: %s is not a subject of %s's step-ended", ErrRefused, subject, bs.Seat)
+	}
+	if ev.Node != node {
+		return seat.Pipeline{}, fmt.Errorf("%w: ask %s ended on %s's subject names node %q", ErrRefused, ev.ID, node, ev.Node)
+	}
+	for {
+		p, rev, err := h.Store.Get(ev.Pipeline)
+		if err != nil {
+			return seat.Pipeline{}, fmt.Errorf("%w: ask %s names pipeline %q, which the runner cannot read: %v", ErrRefused, ev.ID, ev.Pipeline, err)
+		}
+		i := stepOfAsk(p, ev.ID)
+		switch {
+		case i < 0:
+			return p, fmt.Errorf("%w: pipeline %s asked no step under ask %s", ErrRefused, ev.Pipeline, ev.ID)
+		case p.Steps[i].Seat != bs.Seat:
+			return p, fmt.Errorf("%w: ask %s was asked of %s, not %s", ErrRefused, ev.ID, p.Steps[i].Seat, bs.Seat)
+		case ev.Commit != p.Commit || ev.Fingerprint != p.Fingerprint || ev.Module != p.Module || ev.Step != p.Steps[i].Step:
+			return p, fmt.Errorf("%w: ask %s ended %s of %s at %s (%s), asked as %s at %s (%s)", ErrRefused, ev.ID, ev.Step,
+				ev.Module, ev.Commit, ev.Fingerprint, p.Steps[i].Step, p.Commit, p.Fingerprint)
+		case p.Steps[i].Outcome != seat.StepResultOutcomeRunning:
+			return p, nil
+		case !endOutcome(ev.Outcome):
+			return p, fmt.Errorf("%w: ask %s ended %q, not an outcome of step-ended", ErrRefused, ev.ID, ev.Outcome)
+		}
+		now := h.now()
+		s := &p.Steps[i]
+		s.Outcome, s.Node, s.Why, s.Rerun, s.By, s.Ended = ev.Outcome, node, ev.Said, ev.Rerun, ev.By, &now
+		if ev.Tests != nil {
+			s.Tests = &seat.TestCounts{Unit: ev.Tests.Unit, Integration: ev.Tests.Integration}
+		}
+		if ev.Output != "" && h.Output != nil {
+			name := seat.StepOutputsByPipelineStepArtifactKey(keys.EscapeName(ev.Pipeline), s.Step, keys.EscapeName(s.Artifact))
+			if err := h.Output(name, []byte(ev.Output)); err != nil {
+				return p, fmt.Errorf("the output of ask %s cannot be kept: %w", ev.ID, err)
+			}
+			s.Output = name
+		}
+		from := now
+		if ev.Ended != nil {
+			from = ev.Ended.UTC()
+		}
+		asked, err := askApproval(ev.Pipeline, &p, from, h.Asker)
+		if err != nil {
+			return p, err
+		}
+		ended := endPipeline(&p, now)
+		if _, err := h.Store.Update(ev.Pipeline, p, rev); errors.Is(err, ErrChanged) {
+			continue
+		} else if err != nil {
+			return p, fmt.Errorf("pipeline %s cannot be written: %w", ev.Pipeline, err)
+		}
+		if h.Keep != nil {
+			for _, k := range ev.Kept {
+				kept := seat.KeptResult{Digest: k.Digest, Level: k.Level, Package: k.Package, Pipeline: k.Pipeline}
+				if k.Toolchain != nil {
+					kept.Toolchain = &seat.ToolchainUsed{Language: k.Toolchain.Language, Recipe: k.Toolchain.Recipe, Version: k.Toolchain.Version}
+				}
+				if kept.Pipeline == "" {
+					kept.Pipeline = ev.Pipeline
+				}
+				if err := h.Keep(seat.KeptByLevelDigestKey(keys.EscapeName(k.Level), keys.EscapeName(k.Digest)), kept); err != nil {
+					h.logf("the passing result of %s (%s) at %s is not kept, so it runs again next time: %v", k.Package, k.Level, k.Digest, err)
+				}
+			}
+		}
+		h.logEnd(ev.Pipeline, *s, now)
+		if err := h.say(seat.EventStepEnded, p, s); err != nil {
+			return p, fmt.Errorf("pipeline %s's %s is written but not said: %w", ev.Pipeline, seat.EventStepEnded, err)
+		}
+		return p, h.said(p, asked, ended)
+	}
+}
+
+// said says what a written pipeline became: an approval asked is a wait held on the operator (ADR 0283), never a stall.
+func (h *Hearer) said(p seat.Pipeline, asked, ended bool) error {
+	if asked {
+		if err := h.say(seat.EventWaitHeld, p, nil); err != nil {
+			return fmt.Errorf("pipeline %s waits on the operator but %s is not said: %w", p.Pipeline, seat.EventWaitHeld, err)
+		}
+	}
+	if ended {
+		if err := h.say(seat.EventPipelineEnded, p, nil); err != nil {
+			return fmt.Errorf("pipeline %s ended but %s is not said: %w", p.Pipeline, seat.EventPipelineEnded, err)
+		}
+	}
+	return nil
+}
+
+func (h *Hearer) logEnd(key string, s seat.StepResult, at time.Time) {
+	if h.Log == nil {
+		return
+	}
+	if err := h.Log(seat.LogEntry{Pipeline: key, Step: s.Step, Artifact: s.Artifact, What: seat.LogEntryWhatEnded, Outcome: s.Outcome,
+		Output: s.Output, At: at}); err != nil {
+		h.logf("the step log misses the end of %s of %s: %v", s.Step, key, err)
+	}
+}
+
+func (h *Hearer) logf(format string, args ...any) {
+	if h.Logf != nil {
+		h.Logf(format, args...)
+	}
+}
+
+// say says one event of p on the runner's subject for it; step, when not nil, is the result a step-ended carries.
+func (h *Hearer) say(event string, p seat.Pipeline, step *seat.StepResult) error {
+	last := p.Commit
+	if p.Group != "" {
+		last = p.Group
+	}
+	subject := "mesh.seat." + seat.Seat + ".event." + event + "." + keys.EscapeName(p.Module) + "." + keys.EscapeName(p.Branch) + "." + keys.EscapeName(last)
+	at := h.now()
+	switch event {
+	case seat.EventStepEnded:
+		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))
+	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})
+	}
+}
+
+func pipelineEnded(p seat.Pipeline) seat.PipelineEnded {
+	return seat.PipelineEnded{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, Run: p.Run, Outcome: p.Outcome, Why: p.Why, Started: p.Started,
+		Ended: p.Ended, EndCount: p.EndCount, Steps: p.Steps}
+}
+
+func endOutcome(o string) bool {
+	switch o {
+	case nodebuildgo.StepEndedOutcomePassed, nodebuildgo.StepEndedOutcomeFailed, nodebuildgo.StepEndedOutcomeNothingToRun,
+		nodebuildgo.StepEndedOutcomeCouldNotRun:
+		return true
+	}
+	return false
+}
+
+func stepOfAsk(p seat.Pipeline, id string) int {
+	for i, s := range p.Steps {
+		if id != "" && s.Ask == id {
+			return i
+		}
+	}
+	return -1
+}
+
+// endPipeline ends a pipeline whose every step ended: passed when every required step passed or was carried, else
+// failed, naming each required step that held it (ADR 0666 §7, §9; design 47 G §24, §26). A failed optional step holds
+// nothing. An approval step still waiting when a required step did not pass is `not asked` (§26). It says whether the
+// pipeline ended now, its end count one higher.
+func endPipeline(p *seat.Pipeline, now time.Time) bool {
+	if p.Outcome != seat.PipelineOutcomeRunning {
+		return false
+	}
+	var held []string
+	notAsked := -1
+	for i, s := range p.Steps {
+		switch {
+		case s.Step == seat.StepResultStepApproval && s.Outcome == seat.StepResultOutcomeWaiting:
+			notAsked = i
+			continue
+		case s.Outcome == seat.StepResultOutcomeWaiting, s.Outcome == seat.StepResultOutcomeRunning:
+			return false
+		}
+		if s.Status == seat.StepResultStatusRequired && s.Outcome != seat.StepResultOutcomePassed && s.Outcome != seat.StepResultOutcomeCarried {
+			held = append(held, s.Artifact+" "+s.Step+" "+s.Outcome)
+		}
+	}
+	if notAsked >= 0 {
+		if len(held) == 0 {
+			return false
+		}
+		p.Steps[notAsked].Outcome = seat.StepResultOutcomeNotAsked
+	}
+	p.Ended = &now
+	p.EndCount++
+	p.Outcome, p.Why = seat.PipelineOutcomePassed, ""
+	if len(held) > 0 {
+		p.Outcome, p.Why = seat.PipelineOutcomeFailed, "required steps not passed: "+strings.Join(held, ", ")
+	}
+	return true
+}

modules/mesh-delivery-pipeline/internal/pipeline/steps_test.go

@@ -0,0 +1,278 @@
+package pipeline
+
+import (
+	"errors"
+	"strings"
+	"testing"
+	"time"
+
+	"git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient"
+	oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
+	seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner"
+	nodebuildgo "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo"
+	"git.novox.be/novox/mesh-sdk/go/keys"
+)
+
+type memStore map[string]struct {
+	p   seat.Pipeline
+	rev uint64
+}
+
+func (m memStore) Get(key string) (seat.Pipeline, uint64, error) {
+	e, ok := m[key]
+	if !ok {
+		return seat.Pipeline{}, 0, errors.New("no such key")
+	}
+	return e.p, e.rev, nil
+}
+
+func (m memStore) Update(key string, p seat.Pipeline, rev uint64) (uint64, error) {
+	if m[key].rev != rev {
+		return 0, ErrChanged
+	}
+	m[key] = struct {
+		p   seat.Pipeline
+		rev uint64
+	}{p, rev + 1}
+	return rev + 1, nil
+}
+
+func (m memStore) Keys(filter string) ([]string, error) {
+	var out []string
+	for k := range m {
+		if ok, _ := matchKey(filter, k); ok {
+			out = append(out, k)
+		}
+	}
+	return out, nil
+}
+
+func matchKey(filter, key string) (bool, error) {
+	f, k := strings.Split(filter, "."), strings.Split(key, ".")
+	if len(f) != len(k) {
+		return false, nil
+	}
+	for i := range f {
+		if f[i] != "*" && f[i] != k[i] {
+			return false, nil
+		}
+	}
+	return true, nil
+}
+
+type askRecorder struct{ asked []oc.Ask }
+
+func (a *askRecorder) Ask(x oc.Ask) error { a.asked = append(a.asked, x); return nil }
+
+const pkey = "main.5.app"
+
+// fixture is a pipeline of app with a required lint asked of node-build-go, an optional unit asked of the same seat and an
+// approval step waiting, and a hearer over it.
+func fixture() (*Hearer, memStore, *[]string, *askRecorder, map[string]seat.KeptResult, map[string]string) {
+	store := memStore{pkey: {p: seat.Pipeline{V: 1, Pipeline: pkey, Branch: "main", Source: 5, Module: "app", Commit: "c1", Fingerprint: "src2:aa",
+		Outcome: seat.PipelineOutcomeRunning, Steps: []seat.StepResult{
+			{Artifact: "program", Step: seat.StepResultStepLint, Status: seat.StepResultStatusRequired, Outcome: seat.StepResultOutcomeRunning, Ask: "a1", Seat: nodebuildgo.Seat},
+			{Artifact: "program", Step: seat.StepResultStepUnit, Status: seat.StepResultStatusOptional, Outcome: seat.StepResultOutcomeRunning, Ask: "a2", Seat: nodebuildgo.Seat},
+			{Artifact: "program", Step: seat.StepResultStepApproval, Status: seat.StepResultStatusRequired, Outcome: seat.StepResultOutcomeWaiting},
+		}}, rev: 1}}
+	said := &[]string{}
+	asker := &askRecorder{}
+	kept := map[string]seat.KeptResult{}
+	outs := map[string]string{}
+	h := &Hearer{Store: store, Name: "mesh-pipeline-runner", Asker: asker,
+		Say: func(subject string, _ any) error {
+			*said = append(*said, subject[strings.Index(subject, ".event.")+7:])
+			return nil
+		},
+		Log:    func(seat.LogEntry) error { return nil },
+		Keep:   func(k string, v seat.KeptResult) error { kept[k] = v; return nil },
+		Output: func(k string, d []byte) error { outs[k] = string(d); return nil },
+		Now:    func() time.Time { return time.Date(2026, 11, 11, 12, 0, 0, 0, time.UTC) }}
+	return h, store, said, asker, kept, outs
+}
+
+func ended(ask, step, outcome string) StepEnded {
+	return StepEnded{ID: ask, Ask: ask, Node: "n1", Pipeline: pkey, Module: "app", Commit: "c1", Fingerprint: "src2:aa", Step: step, Outcome: outcome}
+}
+
+var goSeat = StepSeat{Seat: nodebuildgo.Seat, Prefix: "mesh.seat.node-build-go.event.step-ended."}
+
+// A step's end is taken only as the runner asked it (design 47 G §24; ADR 0666 §21).
+func TestStepEndedIsTakenOnlyAsAsked(t *testing.T) {
+	h, store, _, _, _, _ := fixture()
+	other := ended("a1", "lint", "passed")
+	other.Fingerprint = "src2:bb"
+	wrongNode := ended("a1", "lint", "passed")
+	wrongNode.Node = "n2"
+	unknownPipeline := ended("a1", "lint", "passed")
+	unknownPipeline.Pipeline = "main.9.app"
+	for name, c := range map[string]struct {
+		subject string
+		ev      StepEnded
+	}{
+		"a subject of another seat":   {"mesh.seat.node-build-node.event.step-ended.n1", ended("a1", "lint", "passed")},
+		"a body naming another node":  {goSeat.Prefix + "n1", wrongNode},
+		"an unknown pipeline":         {goSeat.Prefix + "n1", unknownPipeline},
+		"an ask nobody made":          {goSeat.Prefix + "n1", ended("zz", "lint", "passed")},
+		"another fingerprint":         {goSeat.Prefix + "n1", other},
+		"another step":                {goSeat.Prefix + "n1", ended("a1", "unit", "passed")},
+		"an outcome step-ended lacks": {goSeat.Prefix + "n1", ended("a1", "lint", "waiting")},
+	} {
+		if _, err := h.OnStepEnded(goSeat, c.subject, c.ev); !errors.Is(err, ErrRefused) {
+			t.Errorf("%s: err %v, want a refusal", name, err)
+		}
+	}
+	if store[pkey].p.Steps[0].Outcome != seat.StepResultOutcomeRunning {
+		t.Error("a refused end changed the pipeline")
+	}
+	// The seat the step was asked of is the only one that ends it.
+	if _, err := h.OnStepEnded(StepSeat{Seat: "node-build-node", Prefix: "mesh.seat.node-build-node.event.step-ended."}, "mesh.seat.node-build-node.event.step-ended.n1", ended("a1", "lint", "passed")); !errors.Is(err, ErrRefused) {
+		t.Errorf("another seat's holder ended the step: %v", err)
+	}
+}
+
+// A taken end is written, kept and said; the pipeline ends once every step ended (§24, §27).
+func TestStepEndedEndsThePipeline(t *testing.T) {
+	h, store, said, _, kept, outs := fixture()
+	h.Asker = nil
+	lint := ended("a1", "lint", "passed")
+	lint.Output = "all clean"
+	unit := ended("a2", "unit", "failed")
+	unit.Kept = []nodebuildgo.KeptResult{{Digest: "sha256:d1", Level: "unit", Package: "app/x"}}
+	if _, err := h.OnStepEnded(goSeat, goSeat.Prefix+"n1", lint); err != nil {
+		t.Fatal(err)
+	}
+	if p := store[pkey].p; p.Outcome != seat.PipelineOutcomeRunning || p.Steps[0].Node != "n1" || p.Steps[0].Output == "" {
+		t.Fatalf("after the lint: %+v", p)
+	}
+	if len(outs) != 1 {
+		t.Errorf("outputs kept %v", outs)
+	}
+	// Approval waits with no asker, so the pipeline does not end on the last test step alone.
+	if _, err := h.OnStepEnded(goSeat, goSeat.Prefix+"n1", unit); err != nil {
+		t.Fatal(err)
+	}
+	if p := store[pkey].p; p.Outcome != seat.PipelineOutcomeRunning {
+		t.Fatalf("a pipeline with its approval waiting ended: %+v", p)
+	}
+	if k := kept[seat.KeptByLevelDigestKey("unit", keys.EscapeName("sha256:d1"))]; k.Package != "app/x" || k.Pipeline != pkey {
+		t.Errorf("kept %+v", kept)
+	}
+	// A step heard again takes nothing.
+	n := len(*said)
+	if _, err := h.OnStepEnded(goSeat, goSeat.Prefix+"n1", unit); err != nil || len(*said) != n {
+		t.Errorf("a second end said %v (%v)", (*said)[n:], err)
+	}
+}
+
+// The approval step is asked last, only when every other required step passed, and ended by its warrant (§26, §37).
+func TestApprovalIsAskedLastAndEndedByItsWarrant(t *testing.T) {
+	h, store, said, asker, _, _ := fixture()
+	var record *oc.Warrant
+	h.Where = func(string) (string, *oc.Warrant, error) { return "chosen", record, nil }
+	if _, err := h.OnStepEnded(goSeat, goSeat.Prefix+"n1", ended("a1", "lint", "passed")); err != nil || len(asker.asked) != 0 {
+		t.Fatalf("asked %d before the unit step ended (%v)", len(asker.asked), err)
+	}
+	if _, err := h.OnStepEnded(goSeat, goSeat.Prefix+"n1", ended("a2", "unit", "failed")); err != nil {
+		t.Fatal(err)
+	}
+	// An optional step that failed holds nothing: the approval is asked.
+	if len(asker.asked) != 1 || asker.asked[0].About != "app.main" {
+		t.Fatalf("asked %+v", asker.asked)
+	}
+	a := asker.asked[0]
+	if got := (*said)[len(*said)-1]; !strings.HasPrefix(got, "wait-held.app.main.c1") {
+		t.Errorf("the wait was said as %q", got)
+	}
+	warrant := func(option, outcome string) oc.Warrant {
+		return oc.Warrant{Asker: "mesh-pipeline-runner", Ask: a.ID, About: a.About, Outcome: outcome, Option: option, Level: operatorchannelclient.Approve,
+			AskDigest: operatorchannelclient.Digest(a), At: operatorchannelclient.FormatTime(time.Date(2026, 11, 11, 13, 0, 0, 0, time.UTC)),
+			By: &oc.Person{Who: operatorchannelclient.Operator, Identity: "op", Kind: "telegram", Verified: "linked"}}
+	}
+	// The event only wakes the runner (A-89): an event that says approve while the router's record holds none takes nothing.
+	event := warrant(OptionApprove, oc.WarrantOutcomeChosen)
+	if _, err := h.OnDecided(event); !errors.Is(err, ErrNoRecord) || store[pkey].p.Steps[2].Outcome != seat.StepResultOutcomeRunning {
+		t.Errorf("an event with no record was taken: %v", err)
+	}
+	// A record whose warrant is for another ask's digest is not taken; an event of another asker neither.
+	bad := warrant(OptionApprove, oc.WarrantOutcomeChosen)
+	bad.AskDigest = "sha256:other"
+	record = &bad
+	if _, err := h.OnDecided(event); err == nil || store[pkey].p.Steps[2].Outcome != seat.StepResultOutcomeRunning {
+		t.Errorf("a warrant for another ask was taken: %v", err)
+	}
+	notMine := event
+	notMine.Asker = "someone"
+	if _, err := h.OnDecided(notMine); !errors.Is(err, ErrRefused) {
+		t.Errorf("another asker's event: %v", err)
+	}
+	// The record says decline although the event body says approve: the record decides.
+	decline := warrant(OptionDecline, oc.WarrantOutcomeChosen)
+	record = &decline
+	if p, err := h.OnDecided(event); err != nil || p.Steps[2].Outcome != seat.StepResultOutcomeDeclined {
+		t.Fatalf("the record said decline: %+v (%v)", p.Steps[2], err)
+	}
+	// (the declined pipeline above is a fresh fixture's: approve it on a new one)
+	h, store, said, asker, _, _ = fixture()
+	h.Where = func(string) (string, *oc.Warrant, error) { return "chosen", record, nil }
+	for _, s := range []StepEnded{ended("a1", "lint", "passed"), ended("a2", "unit", "failed")} {
+		if _, err := h.OnStepEnded(goSeat, goSeat.Prefix+"n1", s); err != nil {
+			t.Fatal(err)
+		}
+	}
+	a = asker.asked[0]
+	approve := warrant(OptionApprove, oc.WarrantOutcomeChosen)
+	record = &approve
+	p, err := h.OnDecided(event)
+	if err != nil || p.Steps[2].Outcome != seat.StepResultOutcomePassed || p.Outcome != seat.PipelineOutcomePassed || p.EndCount != 1 {
+		t.Fatalf("approved: %+v (%v)", p, err)
+	}
+	if !strings.HasPrefix((*said)[len(*said)-1], "pipeline-ended.app.main.c1") {
+		t.Errorf("said %v", *said)
+	}
+	// A pipeline that ended takes no second decision.
+	record = &decline
+	if p, err := h.OnDecided(event); err != nil || p.Steps[2].Outcome != seat.StepResultOutcomePassed {
+		t.Errorf("a second warrant changed it: %+v (%v)", p.Steps[2], err)
+	}
+}
+
+// A decline, an expiry or a replacement ends the approval step declined or superseded, and a failed required step leaves it
+// not asked with the pipeline failed (§26, §37).
+func TestApprovalEnds(t *testing.T) {
+	for name, c := range map[string]struct {
+		outcome, option, want string
+	}{
+		"declined":   {oc.WarrantOutcomeChosen, OptionDecline, seat.StepResultOutcomeDeclined},
+		"expired":    {oc.WarrantOutcomeExpired, "", seat.StepResultOutcomeDeclined},
+		"refused":    {oc.WarrantOutcomeRefused, "", seat.StepResultOutcomeDeclined},
+		"superseded": {oc.WarrantOutcomeReplaced, "", seat.StepResultOutcomeSuperseded},
+	} {
+		h, _, _, asker, _, _ := fixture()
+		for _, s := range []StepEnded{ended("a1", "lint", "passed"), ended("a2", "unit", "passed")} {
+			if _, err := h.OnStepEnded(goSeat, goSeat.Prefix+"n1", s); err != nil {
+				t.Fatal(err)
+			}
+		}
+		a := asker.asked[0]
+		rec := oc.Warrant{Asker: h.Name, Ask: a.ID, About: a.About, Outcome: c.outcome, Option: c.option, Level: operatorchannelclient.Approve,
+			AskDigest: operatorchannelclient.Digest(a), At: operatorchannelclient.FormatTime(time.Date(2026, 11, 11, 13, 0, 0, 0, time.UTC)),
+			By: &oc.Person{Who: operatorchannelclient.Operator, Identity: "op", Kind: "telegram", Verified: "linked"}}
+		h.Where = func(string) (string, *oc.Warrant, error) { return c.outcome, &rec, nil }
+		p, err := h.OnDecided(oc.Warrant{Asker: h.Name, Ask: a.ID, About: a.About})
+		if err != nil || p.Steps[2].Outcome != c.want || p.Outcome != seat.PipelineOutcomeFailed {
+			t.Errorf("%s: step %q, pipeline %q (%v)", name, p.Steps[2].Outcome, p.Outcome, err)
+		}
+	}
+	h, store, _, asker, _, _ := fixture()
+	if _, err := h.OnStepEnded(goSeat, goSeat.Prefix+"n1", ended("a1", "lint", "failed")); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := h.OnStepEnded(goSeat, goSeat.Prefix+"n1", ended("a2", "unit", "passed")); err != nil {
+		t.Fatal(err)
+	}
+	p := store[pkey].p
+	if len(asker.asked) != 0 || p.Steps[2].Outcome != seat.StepResultOutcomeNotAsked || p.Outcome != seat.PipelineOutcomeFailed || !strings.Contains(p.Why, "lint") {
+		t.Errorf("asked %d; %+v", len(asker.asked), p)
+	}
+}