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

package steprun

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"sort"
	"strings"
	"sync"
	"time"

	buildrun "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildrun"
	sdkstep "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildstep"
	stdio "git.novox.be/novox/mesh-sdk/go"
)

// **The pull loop the build modules share** (novox/hq ADR 0525 §4-§6, ADR 0536 §4; issue 1887). A build module on a node
// whose build capacity is above zero pulls the asks of its language's seat: from the delivery worker first, then check,
// then recheck, holding at most its capacity at once. Each ask runs as a step under the ask's own id, so its end is the
// key <node>.<ask id> in the module's steps state, which the asker watches; the ask is acknowledged when that end is
// written, kept alive while it runs by an in-progress signal, and redelivered by the stream to another node if this one
// stops. No part keeps a timer for an ask: the pull's own wait and the stream's acknowledgement wait are the bus's.

// Pulled is one ask a source handed over, with what decides its fate on the bus.
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
	// acknowledgement wait. Nak hands it back to be delivered again at once (to this node or another). Term drops it.
	Ack, InProgress, Nak, Term func() error
	// NakAfter hands it back to be delivered again after delay, not before (a step that ended on a transport fault, ADR 0664
	// §1); nil when the source cannot, and a refusal (a tool runner that does not state the capability) is an error.
	NakAfter func(delay time.Duration) error
}

// AskSource is where a puller takes asks from: the seat's workers on the bus (JetStreamAsks) or a test double
// (FakeAsks).
type AskSource interface {
	// Next takes one ask of kind: with wait 0 only one already waiting, else waiting for one at most wait. It answers
	// nil and no error when none came.
	Next(ctx context.Context, kind string, wait time.Duration) (*Pulled, error)
}

// Defaults of a Puller.
var (
	// DefaultPullWait is how long one pull of the delivery worker waits for an ask while every worker is empty: the
	// pull's own wait (as process.Consume's), after which the check and recheck workers are looked at again.
	DefaultPullWait = 5 * time.Second
	// 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.
type Puller struct {
	Source   AskSource
	Language string // the seat's language: sdkstep.AskGo, AskNode or AskImage
	Node     string // the node this holder runs on, the first part of its step keys
	Put      Put    // writes the module's steps state
	// Capacity is how many asks the node runs at once; read before every pull, so a change takes effect at the next
	// one. 0 pulls nothing.
	Capacity func() int
	// Run runs one ask's job; a *Refusal ends its step refused. Its result is the step's result.
	Run func(ctx context.Context, ask sdkstep.StepAsk) (any, error)
	// PullWait and Heartbeat default to DefaultPullWait and DefaultHeartbeat.
	PullWait time.Duration
	// Delay is how long the bus waits before it gives an ask again after its run-th run ended on a transport fault; nil is
	// sdkstep.AgainAfter (a minute, then four). A test's own is short.
	Delay     func(run int) time.Duration
	Heartbeat time.Duration
	// 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
	// KeepOutput keeps a step's answer over what its key holds in the steps state (a check's or a run's, design 47 G A-290)
	// as the object named object, which the key and a run's step-ended name and the seat's verb `output` answers (A-254); a
	// holder keeps it in the object store its manifest requires. Nil keeps none: such an answer ends the ask failed, saying so.
	KeepOutput func(object string, body []byte) error
	// Callers says which seats the build seat's definition lists as callers of an accepted kind (CallersFromDefinitions);
	// found is false while the seat list holds no such kind. Nil checks no named asker (design 47 G §543, A-163).
	Callers       func(kind string) (callers []string, found bool)
	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.
	Load func(key string) ([]byte, error)
	// PausePut writes the node's pause key: the module's pauses state (sdkstep.PausesState), which has no age, so a
	// pause never ends by itself while the steps state drops its old keys. Put when nil.
	PausePut Put
	// ForgetPause deletes one key of the pauses state and PauseKeys lists its keys (nil for none): a dropped ask's key is
	// deleted when the ask is refused at pull, and the start sweeps the keys of asks the stream no longer holds (Drop).
	ForgetPause func(key string) error
	PauseKeys   func() ([]string, error)
	// WatchPauses follows the keys of the pauses state that key names (`**` for the rest): stdio.State(...).Watch in a module. The
	// state is the module's, shared by every node that holds the seat, so a drop another node's holder wrote reaches the node
	// that runs the ask (Drop).
	WatchPauses func(key string, handle func(stdio.StateChange) error) error
	// WatchConfig, when set, follows the module's config value `capacity` (CapacitySetting) as mesh-config resolves it
	// for this node, and wakes the puller on each change (watchCapacity): WatchConfigServed in a module, a test's own
	// double. Until it hands a value the puller judges by Capacity.
	WatchConfig func(key string, handle func(stdio.ProvisionChange) error) error
	// CoresPerStep is what one ask of this module needs, read at every judgement (NodeCoresPerStep); with it the capacity
	// judged by is bounded by the node's build slice: min(the configured capacity, floor(the quota's cores / CoresPerStep))
	// (EffectiveCapacity). Nil bounds nothing by the CPU. Quota reads the quota (Quota when nil; a test's own).
	CoresPerStep func() float64
	Quota        func() (percent int, from string)
	// BuildHours is the node's build window, read at every judgement (NodeBuildHours); with WatchConfig set the value
	// mesh-config resolves for `build-hours` takes its place once handed. Outside the window the capacity judged by is 0:
	// steps running finish, no new one is taken; the puller wakes at the window's next edge. Nil builds at any hour.
	BuildHours func() (*BuildWindow, string)
	// Eligible says why this node cannot run the seat's asks at all now, and "" when it can: what the module proves of
	// its machine and finds missing (mesh-build-image: /dev/net/tun cannot be opened, so buildah's network for a RUN line
	// cannot be made; 2026-10-28 on ace). Read at every judgement, so it is cheap (a module answers its last look).
	// While it says a reason the capacity judged by is 0, said with the reason: steps running finish, no new one is
	// taken, and the asks go to a node that can run them instead of failing here. The module calls Wake when its answer
	// changes. Nil is always eligible.
	Eligible func() string
	// Has says whether this node has a need an ask names (sdkstep.AskNeeds, issue 890). An ask needing what it lacks is
	// declined: its step ends refused as declined (sdkstep.StepDeclined), the message is dropped for its asker to put
	// again, and the worker it came from is not looked at again until the next pull's wait, so a node lacking the need
	// does not take the asker's next copy at once. Nil has nothing.
	Has func(need string) bool
	// SettingsFile is the module's settings file. With it set the puller wakes whenever the file is written (a file event,
	// WatchSettings), and reads Capacity from it itself, keeping why it cannot be read (CapacityNote); a parked puller at
	// capacity 0 therefore wakes when the node-engine refills the file, and a failed config follow is tried again then.
	SettingsFile string
	// watching, when set, is called each time the settings file's watch is (re)armed: a test waits on it, not on a sleep.
	watching func()

	mu      sync.Mutex
	inHand  int
	running map[string]RunningAsk
	// cancels end a running ask's step by its id: the seat's verb drop (Drop).
	cancels map[string]context.CancelCauseFunc
	paused  bool
	wake    chan struct{}
	once    sync.Once
	// watched is the capacity the config watch handed last, used in place of Capacity once hasWatched.
	watched    int
	hasWatched bool
	// capFollow and hoursFollow are the state of the two config follows: followed once WatchConfig took them, trying
	// while a start is in flight; followFault is the last failure said, so each failure is said once.
	capFollowed, capTrying, hoursFollowed, hoursTrying bool
	followFault                                        string
	// fileFault is why the settings file's capacity cannot be read, "" when it can.
	fileFault string
	// capWhy is why the config watch's last value could not be read, "" when it could.
	capWhy string
	// hours is the build window the config watch handed last, and hoursFault why its value cannot be read; used in place
	// of BuildHours once hasHours.
	hours      *BuildWindow
	hoursFault string
	hasHours   bool
	// judged is the capacity said last (log and capacity.<node>), so a change is said once.
	judged     sdkstep.Capacity
	hasJudged  bool
	capWritten bool
	capFault   string
	// declined are the kinds whose worker handed an ask this node declined, left out of the pulls without a wait until
	// the next pull's wait ends.
	declined map[string]bool
}

func (p *Puller) init() {
	p.once.Do(func() { p.wake = make(chan struct{}, 1) })
}

func (p *Puller) logf(format string, args ...any) {
	if p.Logf != nil {
		p.Logf(format, args...)
	}
}

// Wake makes a waiting puller look again: after a capacity change, a resume, or a step's end.
func (p *Puller) Wake() {
	p.init()
	select {
	case p.wake <- struct{}{}:
	default:
	}
}

// Pause stops the puller taking asks; the asks it holds run to their end. by names who paused it. With Load set the
// pause is written on the node's pause key first, and an error there leaves the puller as it was.
func (p *Puller) Pause(by string) error { return p.setPaused(true, by) }

// Resume lets the puller take asks again, written as Pause's is.
func (p *Puller) Resume(by string) error { return p.setPaused(false, by) }

func (p *Puller) setPaused(paused bool, by string) error {
	if p.Load != nil {
		key, err := sdkstep.PauseKey(p.Node)
		if err != nil {
			return err
		}
		raw, err := sdkstep.EncodePause(sdkstep.Pause{Paused: paused, By: by, At: now()})
		if err != nil {
			return err
		}
		put := p.PausePut
		if put == nil {
			put = p.Put
		}
		if _, err := put(key, json.RawMessage(raw)); err != nil {
			return fmt.Errorf("steprun: writing the pause of %s: %w", p.Node, err)
		}
	}
	p.mu.Lock()
	p.paused = paused
	p.mu.Unlock()
	p.Wake()
	return nil
}

// restore reads the node's pause key, when Load is set.
func (p *Puller) restore() error {
	if p.Load == nil {
		return nil
	}
	key, err := sdkstep.PauseKey(p.Node)
	if err != nil {
		return err
	}
	raw, err := p.Load(key)
	if err != nil {
		return fmt.Errorf("steprun: reading the pause of %s: %w", p.Node, err)
	}
	if raw == nil {
		return nil
	}
	v, err := sdkstep.DecodePause(raw)
	if err != nil {
		return fmt.Errorf("steprun: the pause of %s: %w", p.Node, err)
	}
	p.mu.Lock()
	p.paused = v.Paused
	p.mu.Unlock()
	return nil
}

// Paused says whether the puller takes no asks.
func (p *Puller) Paused() bool {
	p.mu.Lock()
	defer p.mu.Unlock()
	return p.paused
}

// InHand is how many asks run now.
func (p *Puller) InHand() int {
	p.mu.Lock()
	defer p.mu.Unlock()
	return p.inHand
}

// RunningAsk is an ask a puller runs now, as its seat's current verb says it.
type RunningAsk struct {
	ID      string    `json:"id"`
	Kind    string    `json:"kind"`
	Module  string    `json:"module,omitempty"`
	Started time.Time `json:"started"`
}

// Running are the asks the puller runs now, oldest first.
func (p *Puller) Running() []RunningAsk {
	p.mu.Lock()
	defer p.mu.Unlock()
	out := make([]RunningAsk, 0, len(p.running))
	for _, r := range p.running {
		out = append(out, r)
	}
	sort.Slice(out, func(i, j int) bool { return out[i].Started.Before(out[j].Started) })
	return out
}

// room says whether the puller may take one more ask now.
func (p *Puller) room() bool {
	ok, _, _ := p.roomNow()
	return ok
}

// roomNow is room, with the capacity and the asks in hand it was judged on.
func (p *Puller) roomNow() (ok bool, capacity, inHand int) {
	capacity = p.CapacityDetail().Effective
	p.mu.Lock()
	defer p.mu.Unlock()
	return !p.paused && p.inHand < capacity, capacity, p.inHand
}

// CapacityNote says when the puller takes nothing for want of a capacity: why, and whether it is a fault (a setting that
// cannot be read or is refused, so a person must look) and not a choice (a capacity set to 0). "" when it pulls.
func (p *Puller) CapacityNote() (why string, fault bool) {
	c := p.CapacityDetail()
	if c.Effective > 0 {
		return "", false
	}
	p.mu.Lock()
	defer p.mu.Unlock()
	if p.fileFault != "" && !p.hasWatched {
		return "the " + p.Language + " build capacity of " + p.Node + " is 0 because it cannot be read: " + p.fileFault, true
	}
	return "the " + p.Language + " build capacity of " + p.Node + " is " + c.Why, p.hasWatched && p.capWhy != ""
}

// CapacityNow is the capacity the puller judges by (CapacityDetail's Effective).
func (p *Puller) CapacityNow() int {
	return p.CapacityDetail().Effective
}

// CapacityDetail is the capacity the puller judges by now and why: the configured one (the config watch's last value,
// else Capacity), bounded by the node's build slice when CoresPerStep is set (EffectiveCapacity), its quota read now. A
// change from what it said last is logged and written to `capacity.<node>` of the pauses state (PausePut), so the
// delivery view says it too.
func (p *Puller) CapacityDetail() sdkstep.Capacity {
	p.mu.Lock()
	watched, has := p.watched, p.hasWatched
	p.mu.Unlock()
	configured, from := 0, "the settings file"
	switch {
	case has:
		configured, from = watched, "mesh-config, as the config watch handed it"
	case p.SettingsFile != "":
		n, err := ReadCapacity(p.SettingsFile)
		configured = n
		p.mu.Lock()
		p.fileFault = ""
		if err != nil {
			p.fileFault = err.Error()
		}
		p.mu.Unlock()
	case p.Capacity != nil:
		configured = p.Capacity()
	}
	var c sdkstep.Capacity
	if p.CoresPerStep == nil {
		c = sdkstep.Capacity{Configured: configured, Effective: configured, QuotaFrom: "not judged: the module names no cores a step needs",
			Why: fmt.Sprintf("its configured capacity %d", configured)}
	} else {
		quota := p.Quota
		if quota == nil {
			quota = Quota
		}
		percent, qfrom := quota()
		c = EffectiveCapacity(configured, percent, qfrom, p.CoresPerStep())
	}
	c.ConfiguredFrom = from
	w, fault := p.window()
	if fault != "" {
		c.BuildHoursFault = fault
		c.Why = fmt.Sprintf("%s; at any hour, because its build hours cannot be read: %s", c.Why, fault)
	}
	if w != nil {
		at := now()
		zone, _ := at.Zone()
		c.BuildHours = w.String() + " " + zone
		if !w.Open(at) && c.Effective > 0 {
			c.OutsideHours = true
			c.Why = fmt.Sprintf("0, outside its build hours %s until %s (otherwise %s)", c.BuildHours, w.NextEdge(at).Format("15:04"), c.Why)
			c.Effective = 0
		} else if c.Effective > 0 {
			c.Why = fmt.Sprintf("%s; within its build hours %s until %s", c.Why, c.BuildHours, w.NextEdge(at).Format("15:04"))
		}
	}
	if p.Eligible != nil {
		if why := p.Eligible(); why != "" {
			c.Ineligible = why
			c.Why = fmt.Sprintf("0, not eligible: %s (otherwise %s)", why, c.Why)
			c.Effective = 0
		}
	}
	p.sayCapacity(c)
	return c
}

// window is the node's build window now (the config watch's last value, else BuildHours, else none), and why its value
// cannot be read when it cannot.
func (p *Puller) window() (*BuildWindow, string) {
	p.mu.Lock()
	w, fault, has := p.hours, p.hoursFault, p.hasHours
	p.mu.Unlock()
	if has {
		return w, fault
	}
	if p.BuildHours != nil {
		return p.BuildHours()
	}
	return nil, ""
}

// untilEdge is how long until the build window's next edge, and false when the node has no window.
func (p *Puller) untilEdge() (time.Duration, bool) {
	w, _ := p.window()
	if w == nil {
		return 0, false
	}
	at := now()
	return w.NextEdge(at).Sub(at), true
}

// sayCapacity logs c when it differs from what was said last, and writes it to the node's capacity key until a write
// lands (a failed write is tried again at the next judgement, its error said once).
func (p *Puller) sayCapacity(c sdkstep.Capacity) {
	p.mu.Lock()
	same := p.hasJudged && p.judged.Effective == c.Effective && p.judged.Configured == c.Configured && p.judged.ByCPU == c.ByCPU &&
		p.judged.CoresPerStep == c.CoresPerStep && p.judged.QuotaCores == c.QuotaCores && p.judged.OutsideHours == c.OutsideHours &&
		p.judged.BuildHours == c.BuildHours && p.judged.BuildHoursFault == c.BuildHoursFault &&
		p.judged.Ineligible == c.Ineligible
	if !same {
		p.judged, p.hasJudged, p.capWritten = c, true, false
	}
	write := !p.capWritten && p.PausePut != nil
	p.mu.Unlock()
	if !same {
		p.logf("the %s build capacity of %s is %s (configured from %s)", p.Language, p.Node, c.Why, c.ConfiguredFrom)
	}
	if !write {
		return
	}
	key, err := sdkstep.CapacityKey(p.Node)
	if err != nil {
		return
	}
	c.At = now()
	raw, err := sdkstep.EncodeCapacity(c)
	if err == nil {
		_, err = p.PausePut(key, json.RawMessage(raw))
	}
	p.mu.Lock()
	defer p.mu.Unlock()
	if err == nil {
		p.capWritten, p.capFault = true, ""
		return
	}
	if err.Error() != p.capFault {
		p.capFault = err.Error()
		p.logf("the %s build capacity of %s could not be written to %s (tried again at the next judgement): %v", p.Language, p.Node, key, err)
	}
}

// Pull takes and runs asks until ctx ends. An ask still running then is neither ended nor acknowledged: the stream
// delivers it again, to another node or to this one restarted.
func (p *Puller) Pull(ctx context.Context) error {
	p.init()
	if p.Source == nil || p.Put == nil || p.Run == nil {
		return errors.New("steprun: a puller needs its source, its steps' put and its run")
	}
	if err := p.restore(); err != nil {
		// An unreadable pause degrades to not paused, said once: a puller that stops silently is the worse failure, and a
		// node kept from building is held by its capacity (0), not by a pause.
		p.logf("the pause of %s cannot be read: %v; pulling as not paused", p.Node, err)
	}
	p.sweepDropped()
	p.watchDrops()
	p.follow() // WatchConfigServed waits for Serve, which a module starts after its pull
	go p.watchSettings(ctx)
	wait := p.PullWait
	if wait <= 0 {
		wait = DefaultPullWait
	}
	for {
		if err := ctx.Err(); err != nil {
			return err
		}
		if ok, capacity, inHand := p.roomNow(); !ok {
			// Parked until something wakes it: a capacity change (the config watch), a resume, or a step's end. Every
			// park logs once, so a puller that sits idle says why (issue 1887).
			p.logf("the %s puller on %s parks: %d of capacity %d in hand, paused %v", p.Language, p.Node, inHand, capacity, p.Paused())
			var edge <-chan time.Time // the build window's next edge, when the node has one: one timer, not a poll
			stop := func() bool { return false }
			if d, ok := p.untilEdge(); ok {
				t := time.NewTimer(d + time.Second)
				edge, stop = t.C, t.Stop
			}
			select {
			case <-ctx.Done():
				stop()
				return ctx.Err()
			case <-p.wake:
			case <-edge:
			}
			stop()
			p.follow() // a follow that failed is tried again at the next event
			_, capacity, inHand = p.roomNow()
			p.logf("the %s puller on %s wakes: %d of capacity %d in hand, paused %v", p.Language, p.Node, inHand, capacity, p.Paused())
			continue
		}
		pullWait := wait
		if d, ok := p.untilEdge(); ok && d < pullWait {
			pullWait = d // a pull does not wait past the window's close
		}
		msg, err := p.next(ctx, pullWait)
		if err != nil {
			if ctx.Err() != nil {
				return ctx.Err()
			}
			p.logf("the %s asks cannot be pulled: %v", p.Language, err)
			select { // the pull's own wait before the next try, as an empty pull's
			case <-ctx.Done():
				return ctx.Err()
			case <-p.wake:
				p.follow()
			case <-time.After(wait):
			}
			continue
		}
		if msg == nil {
			continue
		}
		p.take(ctx, msg)
	}
}

// next is one ask from the workers in their order: one already waiting on delivery, check or recheck, and when none
// waits, a pull of the delivery worker that waits for one.
//
// A kind whose worker handed an ask this node declined is left out until that wait ended (issue 890): the pull's own
// wait, no timer of its own.
func (p *Puller) next(ctx context.Context, wait time.Duration) (*Pulled, error) {
	p.mu.Lock()
	declined := p.declined
	p.mu.Unlock()
	for _, kind := range sdkstep.AskKinds {
		if declined[kind] {
			continue
		}
		msg, err := p.Source.Next(ctx, kind, 0)
		if err != nil || msg != nil {
			return withKind(msg, kind), err
		}
	}
	msg, err := p.Source.Next(ctx, sdkstep.AskDelivery, wait)
	p.mu.Lock()
	p.declined = nil
	p.mu.Unlock()
	return withKind(msg, sdkstep.AskDelivery), err
}

// decline ends the step of an ask this node lacks a need of, refused as declined with what it lacks, and drops the
// message: the asker puts the ask again under a new id (issue 890). A message whose decline cannot be written is handed
// back, as one whose start cannot be.
func (p *Puller) decline(msg *Pulled, ask sdkstep.StepAsk, lacks []string) {
	step, key, err := beginRun(p.Put, keyAsker(msg), p.Node, ask.Kind+" ask", ask.ID, buildrun.AskWhat(askRead(ask)), 0)
	if err == nil {
		why := Refuse("the ask %s needs %s, and %s lacks %s: it is left to a node that has it", ask.ID,
			strings.Join(ask.Needs, ", "), p.Node, strings.Join(lacks, ", "))
		step.Ended = now().UTC()
		step.State, step.Reason = sdkstep.StepRefused, why.Error()
		step.Result, err = json.Marshal(sdkstep.StepDeclined{Declined: true, Needs: ask.Needs, Lacks: lacks})
		if err == nil {
			err = writeEnd(p.Put, key, step)
		}
	}
	if err != nil {
		p.logf("the ask %s cannot be declined on %s, so it is given back: %v", ask.ID, p.Node, err)
		_ = msg.Nak()
		return
	}
	p.logf("the ask %s is declined on %s: it needs %s, and this node lacks %s", ask.ID, p.Node, strings.Join(ask.Needs, ", "), strings.Join(lacks, ", "))
	_ = msg.Term()
	p.mu.Lock()
	if p.declined == nil {
		p.declined = map[string]bool{}
	}
	p.declined[msg.Kind] = true
	p.mu.Unlock()
}

// declineRun ends the step of an ask that began and then found this node should not run it (Declining), as decline does
// before it begins: refused as declined, the message dropped for its asker to put again, and the worker it came from left
// alone until the next pull's wait.
func (p *Puller) declineRun(msg *Pulled, ask sdkstep.StepAsk, key string, step sdkstep.Step, d *Declining) {
	result, _ := json.Marshal(sdkstep.StepDeclined{Declined: true, Needs: ask.Needs, Lacks: d.Lacks})
	step.Ended = now().UTC()
	step.State, step.Reason, step.Result = sdkstep.StepRefused, d.Text, result
	if err := writeEnd(p.Put, key, step); err != nil {
		p.logf("the ask %s cannot be declined on %s, so it is given back: %v", ask.ID, p.Node, err)
		_ = msg.Nak()
		return
	}
	p.logf("the ask %s is declined on %s: %s", ask.ID, p.Node, d.Text)
	_ = msg.Term()
	p.mu.Lock()
	if p.declined == nil {
		p.declined = map[string]bool{}
	}
	p.declined[msg.Kind] = true
	p.mu.Unlock()
}

// refuseUnread ends an ask this module cannot read: when its data holds an id that keys a step, the step is written
// refused with the reason, so the asker learns at once that no node began it; otherwise (no id, not JSON) the ask is only
// dropped and logged. The message is dropped either way.
func (p *Puller) refuseUnread(msg *Pulled, why error) {
	var probe struct {
		ID   string `json:"id"`
		Kind string `json:"kind"`
	}
	if json.Unmarshal(msg.Data, &probe) == nil && probe.ID != "" {
		if key, kerr := sdkstep.AskStepKey(keyAsker(msg), p.Node, probe.ID); kerr == nil {
			tool := "ask"
			if probe.Kind != "" {
				tool = probe.Kind + " ask"
			}
			at := now().UTC()
			step := sdkstep.Step{ID: probe.ID, Tool: tool, Node: p.Node, State: sdkstep.StepRefused, Started: at, Ended: at,
				Reason: fmt.Sprintf("this build module on %s cannot read the ask: %v", p.Node, why), Owner: buildrun.Owner}
			if werr := writeEnd(p.Put, key, step); werr != nil {
				p.logf("the unread ask %s cannot be answered on %s: %v", probe.ID, p.Node, werr)
			} else {
				p.logf("an ask refused: %v", why)
				_ = msg.Term()
				return
			}
		}
	}
	p.logf("an ask dropped: %v", why)
	_ = msg.Term()
}

func withKind(msg *Pulled, kind string) *Pulled {
	if msg != nil {
		msg.Kind = kind
	}
	return msg
}

// take starts one ask: it decodes it, refuses one that is not of this seat, and runs it as a step under its own id.
func (p *Puller) take(ctx context.Context, msg *Pulled) {
	kind := msg.Kind
	ask, err := sdkstep.DecodeStepAsk(msg.Data)
	if err == nil && (ask.Language != p.Language || (kind != "" && !ridesQueue(ask.Kind, kind))) {
		err = fmt.Errorf("the ask %s is a %s ask of %s, pulled from the %s %s worker", ask.ID, ask.Kind, ask.Language, p.Language, kind)
	}
	if err != nil {
		// An ask that is not one runs nowhere: delivering it again would only fail again. But an ask that names an id
		// is answered on its step key as refused, naming why (a job this build module does not know, a version it does
		// not read): a dropped ask must always be answered, or its asker waits to its bound for nothing.
		p.refuseUnread(msg, err)
		return
	}
	if why, judged := p.askerRefusal(msg, ask); !judged {
		p.giveBack(msg, ask)
		return
	} else if why != "" {
		p.refuseAsker(msg, ask, why)
		return
	}
	if d := p.droppedAsk(ask.ID); d != nil {
		p.refuseDropped(msg, ask, d)
		return
	}
	if lacks := sdkstep.Lacks(ask.Needs, p.Has); len(lacks) > 0 {
		p.decline(msg, ask, lacks)
		return
	}
	step, key, err := beginRun(p.Put, keyAsker(msg), p.Node, ask.Kind+" ask", ask.ID, buildrun.AskWhat(askRead(ask)), int(msg.Delivered))
	if err != nil {
		p.logf("the ask %s cannot start on %s: %v", ask.ID, p.Node, err)
		_ = msg.Nak()
		return
	}
	if msg.Delivered > sdkstep.AskRuns {
		// Run AskRuns times and never ended: each run took its node down or died with it. It is run no more, and said
		// (issue 1907): the asker's key fails with why.
		why := fmt.Errorf("the ask was taken %d times and no run of it ended: each took its build node down or died with it, so it is run no more (poison)", msg.Delivered-1)
		p.logf("the ask %s is ended on %s: %v", ask.ID, p.Node, why)
		_, _ = finish(p.Put, p.KeepOutput, key, step, nil, why, p.Logf) // run no more either way: the asker's bound says it then
		_ = msg.Term()
		return
	}
	module := ""
	if ask.Build != nil {
		module = ask.Build.Module
	}
	p.mu.Lock()
	p.inHand++
	if p.running == nil {
		p.running = map[string]RunningAsk{}
	}
	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)
}

func (p *Puller) run(ctx context.Context, msg *Pulled, ask sdkstep.StepAsk, key string, step sdkstep.Step) {
	defer func() {
		p.mu.Lock()
		p.inHand--
		delete(p.running, ask.ID)
		p.mu.Unlock()
		p.Wake()
	}()
	beat := p.Heartbeat
	if beat <= 0 {
		beat = DefaultHeartbeat
	}
	done := make(chan struct{})
	go func() {
		t := time.NewTicker(beat)
		defer t.Stop()
		for {
			select {
			case <-done:
				return
			case <-t.C:
				if msg.InProgress != nil {
					_ = msg.InProgress()
				}
			}
		}
	}()
	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)
	})
	step = ph.end() // its end keeps the phase it ended in
	close(done)
	if d := droppedCause(runCtx); d != nil && ctx.Err() == nil {
		// The seat's verb drop ended it (ADR 0664 §4, §5): dropped, naming who dropped it and why; acknowledged, never given again.
		err = &Dropped{Text: d.Error()}
		defer p.forgetDrop(ask.ID)
	}
	p.mu.Lock()
	delete(p.cancels, ask.ID)
	p.mu.Unlock()
	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, nil, "")
		return
	}
	if ctx.Err() != nil {
		// The process stops: the ask is neither ended nor acknowledged here, so the stream gives it to a node that runs.
		return
	}
	if run := int(msg.Delivered); err != nil && sdkstep.IsPassing(err) && run < sdkstep.AskRuns && msg.NakAfter != nil {
		// A transport fault (ADR 0664 §1): the step says it is given again and the bus gives it after a delay, to this node or
		// another; it is not acknowledged, so its asker sees no end. Run AskRuns times it ends failed, said as a transport fault.
		delay := sdkstep.AgainAfter(run)
		if p.Delay != nil {
			delay = p.Delay(run)
		}
		if aerr := again(p.Put, key, step, err, delay); aerr == nil {
			p.logf("the %s ask %s ended on %s on a transport fault in its run %d (%v), and is given again in %s", ask.Kind, ask.ID, p.Node, run, err, delay)
			if nerr := msg.NakAfter(delay); nerr != nil {
				p.logf("the ask %s could not be handed back with a delay (%v), so it is handed back at once", ask.ID, nerr)
				_ = msg.Nak()
			}
			return
		} else {
			p.logf("the ask %s ended on a transport fault and its step could not be written again (%v): it ends failed", ask.ID, aerr)
		}
	}
	written, ferr := finish(p.Put, p.KeepOutput, key, step, result, err, p.Logf)
	if ferr != nil {
		// No end was written: the ask is not acknowledged as done, so the stream gives it again and a run that can
		// write its end (or the poison rule, issue 1907) ends it where its asker sees it (issue 1930).
		p.logf("the ask %s ran on %s and no end of it could be written, so it is given back: %v", ask.ID, p.Node, ferr)
		_ = msg.Nak()
		return
	}
	// step-ended says the end the key holds (design 47 G A-290): its outcome, and the answer only when the key kept it, so no
	// step reads passed in one place and failed in another.
	if len(written.Result) == 0 && written.OutputObject == "" {
		result = nil
	}
	outcome := "passed"
	if written.State != sdkstep.StepPassed {
		outcome = "did not pass: " + written.Reason
		if len(outcome) > 300 {
			outcome = outcome[:300] + "..."
		}
		p.sayEnded(msg, ask, sdkstep.OutcomeFailed, written.Reason, result, written.OutputObject)
	} else {
		p.sayEnded(msg, ask, sdkstep.OutcomePassed, "", result, written.OutputObject)
	}
	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)
	if aerr := msg.Ack(); aerr != nil {
		p.logf("the ask %s ended on %s and its acknowledgement failed: %v", ask.ID, p.Node, aerr)
	}
}

// watchCapacity follows the module's config value `capacity` when WatchConfig is set (issue 1887, 2026-10-25: the laptop's
// puller started at capacity 0 sat parked with capacity 2 set, because nothing woke it to read its settings file
// again). The value comes as mesh-config resolved it for this node, the event of the change itself: the runtime hands
// the value now and then each change its machine's provisioner client says moved, so nothing polls, and the puller does
// not wait for the node-engine to fill its settings file again. Each value handed is kept, used in place of Capacity,
// and wakes the puller. A watch that cannot start leaves the puller on Capacity, said once.
func (p *Puller) watchCapacity() {
	if p.WatchConfig == nil {
		return
	}
	p.mu.Lock()
	if p.capFollowed || p.capTrying {
		p.mu.Unlock()
		return
	}
	p.capTrying = true
	p.mu.Unlock()
	err := p.WatchConfig(sdkstep.CapacitySetting, func(c stdio.ProvisionChange) error {
		n, why := p.capacityFrom(c)
		p.mu.Lock()
		before, had := p.watched, p.hasWatched
		p.watched, p.hasWatched, p.capWhy = n, true, why
		p.mu.Unlock()
		switch {
		case why != "":
			p.logf("the %s puller on %s pulls nothing: %s", p.Language, p.Node, why)
		case !had || before != n:
			p.logf("the %s build capacity of %s is %d, as mesh-config resolved it", p.Language, p.Node, n)
		}
		p.Wake()
		return nil
	})
	p.followed(&p.capFollowed, &p.capTrying, err, "build capacity")
}

// follow starts each config follow that is not running: at the start, and at every wake after (a file event, a resume, a
// step's end), so a follow that failed is tried again by the next event and not given up for good.
func (p *Puller) follow() {
	go p.watchCapacity()
	go p.watchHours()
}

// followed records the end of one follow's start. A failure is said once per distinct error, with that it is tried again.
func (p *Puller) followed(done, trying *bool, err error, what string) {
	p.mu.Lock()
	*trying = false
	*done = err == nil
	said := err != nil && err.Error() == p.followFault
	if err != nil {
		p.followFault = err.Error()
	} else {
		p.followFault = ""
	}
	p.mu.Unlock()
	if err != nil && !said {
		p.logf("the %s %s of %s cannot be followed on the bus, so it is read from the settings file; the follow is tried "+
			"again at the next event: %v", p.Language, what, p.Node, err)
	}
}

// watchHours follows the module's config value `build-hours` when WatchConfig and BuildHours are set, as watchCapacity
// follows `capacity`: each value handed is kept, used in place of BuildHours, and wakes the puller. A value that cannot
// be read is no window, said; a watch that cannot start leaves the puller on BuildHours, said once.
func (p *Puller) watchHours() {
	if p.WatchConfig == nil || p.BuildHours == nil {
		return
	}
	p.mu.Lock()
	if p.hoursFollowed || p.hoursTrying {
		p.mu.Unlock()
		return
	}
	p.hoursTrying = true
	p.mu.Unlock()
	err := p.WatchConfig(BuildHoursSetting, func(c stdio.ProvisionChange) error {
		setting := ""
		var w *BuildWindow
		var why string
		switch {
		case c.Refused != "":
			why = "its build hours are no longer answered: " + c.Refused
		case len(c.Value) > 0 && string(c.Value) != "null" && json.Unmarshal(c.Value, &setting) != nil:
			why = fmt.Sprintf("the setting %s %s is not a string", BuildHoursSetting, c.Value)
		default:
			var err error
			if w, err = ParseBuildWindow(setting); err != nil {
				why = err.Error()
			}
		}
		p.mu.Lock()
		p.hours, p.hoursFault, p.hasHours = w, why, true
		p.mu.Unlock()
		if why != "" {
			p.logf("the %s puller on %s builds at any hour: %s", p.Language, p.Node, why)
		}
		p.Wake()
		return nil
	})
	p.followed(&p.hoursFollowed, &p.hoursTrying, err, "build hours")
}

// WatchConfigServed is stdio.WatchConfig once Serve has the runtime's channel (stdio.Served): a build module starts its
// pull before it serves, and a watch asked before that would be refused.
func WatchConfigServed(key string, handle func(stdio.ProvisionChange) error) error {
	<-stdio.Served()
	return stdio.WatchConfig(key, handle)
}

// capacityFrom is the node's capacity in one value the watch handed, and why it is 0 when it cannot be read.
func (p *Puller) capacityFrom(c stdio.ProvisionChange) (int, string) {
	if c.Refused != "" {
		return 0, "its capacity is no longer answered: " + c.Refused
	}
	n, err := capacityOf(c.Value)
	if err != nil {
		return 0, err.Error()
	}
	return n, ""
}

// askRead is the part of an ask that buildrun.AskWhat reads, taken from the ask's own layout by the one conversion the
// neutral form has (sdkstep.AskAs); an ask that does not convert reads as empty and is described by its kind alone.
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"`
}

// keyAsker is the asker an ask's step is keyed under: the last token of the subject it came on, none for main's form or a
// source that gives no subject (design 47 G §481, §483), never the ask's own field.
func keyAsker(msg *Pulled) string {
	if _, _, asker, err := sdkstep.ParseAskSubjectBy(msg.Subject); err == nil {
		return asker
	}
	return ""
}

// ridesQueue says an ask of kind may come from the worker of queue: its own, or for a step check or delivery (design 47 G
// §415, §600, A-170).
func ridesQueue(kind, queue string) bool {
	return kind == queue || (kind == sdkstep.AskStep && (queue == sdkstep.AskCheck || queue == sdkstep.AskDelivery))
}

// 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). The end of a run carries the run's answer (A-254): the last line of its program's
// standard output, inline as `output` up to sdkstep.StepEndedInlineMost, else as `output-object`, the object kept, the step's
// key naming it (design 47 G A-290: finish kept it with KeepOutput, so it is not put twice); a program that exits other than 0
// ends failed with its answer carried.
func (p *Puller) sayEnded(msg *Pulled, ask sdkstep.StepAsk, outcome, said string, result any, kept string) {
	if p.Say == nil || ask.Step != nil {
		return
	}
	end := sdkstep.StepEnded{ID: ask.ID, Asker: askerOf(msg, ask), Ask: ask.ID, Node: p.Node, Outcome: outcome, At: now().UTC()}
	if ans, ok := runAnswerOf(ask, result); ok {
		if ans.Exit != 0 && outcome == sdkstep.OutcomePassed {
			outcome, said = sdkstep.OutcomeFailed, exitedWhy(ans)
		}
		if len(ans.Output) <= sdkstep.StepEndedInlineMost {
			end.Output = string(ans.Output)
		} else if kept != "" {
			end.OutputObject = kept
		} else {
			outcome, said = sdkstep.OutcomeFailed, fmt.Sprintf("the run's answer is %d bytes, over what the end carries, and no object of it was kept", len(ans.Output))
		}
	}
	if len(said) > 300 {
		said = said[:300] + "..."
	}
	if said == "" && outcome != sdkstep.OutcomePassed {
		said = "the ask did not pass"
	}
	end.Outcome, end.Said = outcome, said
	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)
}

// runAnswerOf is the run answer a run ask's step handed back, whatever its form (the value or a pointer).
func runAnswerOf(ask sdkstep.StepAsk, result any) (sdkstep.RunAnswer, bool) {
	if ask.Run == nil {
		return sdkstep.RunAnswer{}, false
	}
	return runAnswerIn(result)
}