BG4, BN4, BI4: the puller says ask-taken, step-progress and the end of any ask

A holder says ask-taken when an ask starts, step-progress every minute while it runs (phase fetch, build or run) and
step-ended for an ask outside a pipeline, each on the seat's subject of its node (SayOn); the asker is the last token of
the subject the ask came on (Pulled.Subject).

Builds: design 47 G §496, §602, §603, §481

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

modules/mesh-delivery-sdk/steprun/phase.go

@@ -65,3 +65,10 @@ func (r *phases) end() sdkstep.Step {
 	r.ended = true
 	return r.step
 }
+
+// phase is the phase the ask is in now, "" before its first.
+func (r *phases) phase() string {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	return r.step.Phase
+}

modules/mesh-delivery-sdk/steprun/pull.go

@@ -27,6 +27,9 @@ type Pulled struct {
 	// Kind is the worker it came from, filled by the puller.
 	Kind string
 	Data []byte
+	// Subject is the subject the ask was published on: its last token is the asker the bus proved (design 47 G §481); empty
+	// for a source that has none, and for main's form, which names none.
+	Subject string
 	// Delivered is how many times the bus has handed it over, this time included.
 	Delivered uint64
 	// Ack ends it: it is never delivered again. InProgress tells the bus it is still being worked, restarting its
@@ -53,6 +56,8 @@ var (
 	// DefaultHeartbeat is how often a running ask tells the bus it is still being worked; under the seat worker's
 	// acknowledgement wait of 60 s (mesh-controller internal/broker HolderConsumerFor).
 	DefaultHeartbeat = 20 * time.Second
+	// DefaultProgressEvery is how often a running ask says step-progress (design 47 G §496: once a minute).
+	DefaultProgressEvery = time.Minute
 )
 
 // Puller pulls a build seat's asks and runs them.
@@ -72,7 +77,12 @@ type Puller struct {
 	// sdkstep.AgainAfter (a minute, then four). A test's own is short.
 	Delay     func(run int) time.Duration
 	Heartbeat time.Duration
-	Logf      func(format string, args ...any)
+	// Say publishes one event of the seat on this node's subject (SayOn): ask-taken when an ask starts, step-progress once
+	// every ProgressEvery while it runs, step-ended when an ask outside a pipeline ends (design 47 G §496, §602). Nil says
+	// none, a test's. ProgressEvery defaults to DefaultProgressEvery.
+	Say           func(event string, body any, id string) error
+	ProgressEvery time.Duration
+	Logf          func(format string, args ...any)
 	// Load reads one key of the state the node's pause is kept in (nil and no error when it is absent). When set, Pull
 	// reads the node's pause key (sdkstep.PauseKey) before it pulls, and Pause and Resume write it with PausePut, so a
 	// pause outlives a restart.
@@ -665,6 +675,7 @@ func (p *Puller) take(ctx context.Context, msg *Pulled) {
 	p.running[ask.ID] = RunningAsk{ID: ask.ID, Kind: ask.Kind, Module: module, Started: step.Started}
 	p.mu.Unlock()
 	p.logf("the %s ask %s starts on %s: %s", ask.Kind, ask.ID, p.Node, step.What)
+	p.say("ask-taken", askTaken{Ask: ask.ID, Asker: askerOf(msg, ask), Node: p.Node, At: now().UTC()}, "ask-taken."+ask.ID)
 	go p.run(ctx, msg, ask, key, step)
 }
 
@@ -696,6 +707,25 @@ func (p *Puller) run(ctx context.Context, msg *Pulled, ask sdkstep.StepAsk, key
 		}
 	}()
 	ph := &phases{put: p.Put, key: key, step: step, logf: p.Logf, prefix: fmt.Sprintf("the %s ask %s on %s", ask.Kind, ask.ID, p.Node)}
+	if p.Say != nil {
+		every := p.ProgressEvery
+		if every <= 0 {
+			every = DefaultProgressEvery
+		}
+		go func() {
+			t := time.NewTicker(every)
+			defer t.Stop()
+			for {
+				select {
+				case <-done:
+					return
+				case <-t.C:
+					p.say("step-progress", stepProgress{Ask: ask.ID, Asker: askerOf(msg, ask), Phase: progressPhase(ask, ph.phase()), At: now().UTC()},
+						fmt.Sprintf("step-progress.%s.%d", ask.ID, now().UnixNano()))
+				}
+			}
+		}()
+	}
 	runCtx := p.cancellable(ctx, ask.ID)
 	result, err := protect(runCtx, func(ctx context.Context) (any, error) {
 		return p.Run(context.WithValue(ctx, phaseKey{}, ph), ask)
@@ -713,6 +743,7 @@ func (p *Puller) run(ctx context.Context, msg *Pulled, ask sdkstep.StepAsk, key
 	var dec *Declining
 	if errors.As(err, &dec) && ctx.Err() == nil {
 		p.declineRun(msg, ask, key, step, dec)
+		p.sayEnded(msg, ask, sdkstep.OutcomeCouldNotRun, dec.Text)
 		return
 	}
 	if ctx.Err() != nil {
@@ -750,6 +781,9 @@ func (p *Puller) run(ctx context.Context, msg *Pulled, ask sdkstep.StepAsk, key
 		if len(outcome) > 300 {
 			outcome = outcome[:300] + "..."
 		}
+		p.sayEnded(msg, ask, sdkstep.OutcomeFailed, err.Error())
+	} else {
+		p.sayEnded(msg, ask, sdkstep.OutcomePassed, "")
 	}
 	p.logf("the %s ask %s ended on %s after %s, in its phase %q; %s", ask.Kind, ask.ID, p.Node,
 		now().Sub(step.Started).Round(time.Second), step.Phase, outcome)
@@ -884,3 +918,70 @@ func askRead(ask sdkstep.StepAsk) sdkstep.Ask {
 	out, _ := sdkstep.AskAs[sdkstep.Ask](ask)
 	return out
 }
+
+// askTaken and stepProgress are the bodies of the seat's events ask-taken and step-progress (design 47 G §602, §496).
+type askTaken struct {
+	Ask   string    `json:"ask"`
+	Asker string    `json:"asker"`
+	Node  string    `json:"node"`
+	At    time.Time `json:"at"`
+}
+
+type stepProgress struct {
+	Ask   string    `json:"ask"`
+	Asker string    `json:"asker"`
+	Phase string    `json:"phase"`
+	At    time.Time `json:"at"`
+}
+
+// askerOf is the asker the bus proved: the last token of the subject the ask came on (design 47 G §481), the ask's own
+// field only for a source that gives no subject and for main's form.
+func askerOf(msg *Pulled, ask sdkstep.StepAsk) string {
+	if _, _, asker, err := sdkstep.ParseAskSubjectBy(msg.Subject); err == nil && asker != "" {
+		return asker
+	}
+	return ask.Asker
+}
+
+// progressPhase is the phase step-progress says (fetch, build, run or report): fetch until the ask has entered a phase of
+// its own or while it names the fetch, then the ask's kind (build for a build, run for any other).
+func progressPhase(ask sdkstep.StepAsk, phase string) string {
+	switch {
+	case phase == "" || strings.HasPrefix(phase, "fetch"):
+		return "fetch"
+	case ask.Build != nil:
+		return "build"
+	}
+	return "run"
+}
+
+// say publishes one event of the seat; a failure is logged and never fails the ask.
+func (p *Puller) say(event string, body any, id string) {
+	if p.Say == nil {
+		return
+	}
+	if err := p.Say(event, body, id); err != nil {
+		p.logf("the event %s of the ask %s could not be said on %s: %v", event, id, p.Node, err)
+	}
+}
+
+// sayEnded says step-ended for an ask outside a pipeline, whatever its asker (design 47 G §602): a pipeline step's end is
+// its run's own (AskSteps.SayStep).
+func (p *Puller) sayEnded(msg *Pulled, ask sdkstep.StepAsk, outcome, said string) {
+	if p.Say == nil || ask.Step != nil {
+		return
+	}
+	if len(said) > 300 {
+		said = said[:300] + "..."
+	}
+	if said == "" && outcome != sdkstep.OutcomePassed {
+		said = "the ask did not pass"
+	}
+	end := sdkstep.StepEnded{ID: ask.ID, Asker: askerOf(msg, ask), Ask: ask.ID, Node: p.Node, Outcome: outcome, Said: said, At: now().UTC()}
+	body, err := sdkstep.EncodeStepEnded(end)
+	if err != nil {
+		p.logf("the end of the ask %s cannot be said on %s: %v", ask.ID, p.Node, err)
+		return
+	}
+	p.say("step-ended", json.RawMessage(body), "step-ended."+ask.ID)
+}

modules/mesh-delivery-sdk/steprun/pull_bus.go

@@ -69,7 +69,7 @@ func (s *JetStreamAsks) Next(ctx context.Context, kind string, wait time.Duratio
 		if meta, err := msg.Metadata(); err == nil && meta != nil {
 			delivered = meta.NumDelivered
 		}
-		return &Pulled{Data: append([]byte(nil), msg.Data()...), Delivered: delivered,
+		return &Pulled{Data: append([]byte(nil), msg.Data()...), Subject: msg.Subject(), Delivered: delivered,
 			Ack: msg.Ack, InProgress: msg.InProgress, Nak: msg.Nak, Term: msg.Term,
 			NakAfter: func(delay time.Duration) error { return msg.NakWithDelay(delay) }}, nil
 	case <-ctx.Done():

modules/mesh-delivery-sdk/steprun/pull_events_test.go

@@ -0,0 +1,93 @@
+package steprun
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+
+	sdkstep "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildstep"
+)
+
+type said struct {
+	event, id string
+	body      map[string]any
+}
+
+type sayer struct {
+	mu  sync.Mutex
+	got []said
+}
+
+func (s *sayer) say(event string, body any, id string) error {
+	raw, _ := json.Marshal(body)
+	var m map[string]any
+	_ = json.Unmarshal(raw, &m)
+	s.mu.Lock()
+	s.got = append(s.got, said{event, id, m})
+	s.mu.Unlock()
+	return nil
+}
+
+func (s *sayer) of(event string) []said {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	var out []said
+	for _, g := range s.got {
+		if g.event == event {
+			out = append(out, g)
+		}
+	}
+	return out
+}
+
+// Design 47 G §496, §602 (BG4, BN4, BI4): a holder says ask-taken when an ask starts, step-progress while it runs and
+// step-ended when an ask outside a pipeline ends, whatever its asker.
+func TestPullSaysAskTakenProgressAndEnd(t *testing.T) {
+	for _, c := range []struct {
+		name    string
+		runErr  error
+		outcome string
+	}{{"passed", nil, sdkstep.OutcomePassed}, {"failed", errors.New("the build broke"), sdkstep.OutcomeFailed}} {
+		t.Run(c.name, func(t *testing.T) {
+			src, st, sy := NewFakeAsks(), newSteps(), &sayer{}
+			a := testAsk(sdkstep.AskDelivery)
+			_ = src.PublishAsk(a)
+			release, entered := make(chan struct{}), make(chan struct{})
+			p := &Puller{Source: src, Language: sdkstep.AskGo, Node: "shanks", Put: st.put, Capacity: func() int { return 1 },
+				PullWait: 10 * time.Millisecond, Heartbeat: time.Hour, Say: sy.say, ProgressEvery: 5 * time.Millisecond,
+				Run: func(ctx context.Context, a sdkstep.StepAsk) (any, error) {
+					Phase(ctx, "build m")
+					close(entered)
+					<-release
+					return nil, c.runErr
+				}}
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+			go p.Pull(ctx)
+			<-entered
+			waitFor(t, "a step-progress in the build", func() bool {
+				pr := sy.of("step-progress")
+				return len(pr) > 0 && pr[len(pr)-1].body["phase"] == "build"
+			})
+			close(release)
+			waitFor(t, "step-ended", func() bool { return len(sy.of("step-ended")) == 1 })
+			taken := sy.of("ask-taken")
+			if len(taken) != 1 || taken[0].body["ask"] != a.ID || taken[0].body["node"] != "shanks" || taken[0].id != "ask-taken."+a.ID {
+				t.Fatalf("ask-taken = %+v, want one for %s on shanks", taken, a.ID)
+			}
+			if pr := sy.of("step-progress")[len(sy.of("step-progress"))-1].body; pr["ask"] != a.ID || pr["phase"] != "build" || pr["asker"] != a.Asker {
+				t.Fatalf("step-progress = %v, want the ask, its asker and the phase build", pr)
+			}
+			end := sy.of("step-ended")[0].body
+			if end["outcome"] != c.outcome || end["ask"] != a.ID || end["asker"] != a.Asker || end["node"] != "shanks" || end["at"] == nil {
+				t.Fatalf("step-ended = %v, want outcome %s of ask %s", end, c.outcome, a.ID)
+			}
+			if c.runErr != nil && end["said"] != "the build broke" {
+				t.Fatalf("step-ended said %v, want why", end["said"])
+			}
+		})
+	}
+}

modules/mesh-delivery-sdk/steprun/steprun.go

@@ -250,3 +250,15 @@ func SayStepOn(language, node string, publish func(subject string, body any, id
 		return publish(subject, json.RawMessage(body), "step-ended."+end.ID)
 	}
 }
+
+// SayOn is a holder's Puller.Say: an event of the language's build seat published on that seat's subject of node with
+// publish (stdio.SeatPublish), `mesh.seat.node-build-<language>.event.<event>.<node>`.
+func SayOn(language, node string, publish func(subject string, body any, id string) error) func(event string, body any, id string) error {
+	return func(event string, body any, id string) error {
+		seat, err := sdkstep.BuildSeat(language)
+		if err != nil {
+			return err
+		}
+		return publish("mesh.seat."+seat+".event."+event+"."+node, body, id)
+	}
+}