RN25: the runner's health check is its tool runner_health

The process writes, in its state consumers, when its durable consumer's oldest
unacknowledged message was delivered, at each delivery and acknowledgement; the
bundle's runner_health reads it and the pipelines state, and names a consumer
five minutes or more behind. The process resource's health is that tool.

Builds: design 47 G §76, §468, §495; lead answers A-164, A-166

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

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

@@ -98,9 +98,20 @@ func FirstViewTools(st stdio.KeptState, acts views.HandActs, now func() time.Tim
 	return views.StatusTools(Module, Prefix+"_first_views", Prefix+"_first_view_again", st, acts, now)
 }
 
+// HealthTool is the runner's health check, run outside its process (design 47 G §76, §468, §495; A-164): it reads the
+// runner's state pipelines as show does and the ages its process writes in the state consumers, and asks no seat verb.
+func HealthTool(pipelines, consumers pipeline.Kept, now func() time.Time) stdio.Tool {
+	return stdio.Tool{Name: Prefix + "_health",
+		Description: "The pipeline runner's health: healthy when it answers show and each of its durable consumers' oldest " +
+			"unacknowledged message waits less than five minutes; otherwise it names each consumer behind.",
+		Input: map[string]any{"type": "object", "properties": map[string]any{}},
+		Run:   func(map[string]any) (any, error) { return pipeline.Health(pipelines, consumers, now()), nil }}
+}
+
 func main() {
 	r := pipeline.Reads{Pipelines: kept{stdio.State(seat.StatePipelines)}, Kept: kept{stdio.State(seat.StateKept)}}
 	tools := append(Tools(r), FirstViewTools(stdio.State(views.StatusState), meshhandactsclient.Recorder{Ask: stdio.Ask}, time.Now)...)
+	tools = append(tools, HealthTool(r.Pipelines, kept{stdio.State(pipeline.StateConsumers)}, time.Now))
 	if err := stdio.Serve("", tools); err != nil {
 		fmt.Fprintf(os.Stderr, "[mesh-delivery-pipeline-tools] %v\n", err)
 		os.Exit(1)

modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main_test.go

@@ -26,7 +26,9 @@ func TestTools(t *testing.T) {
 	for _, tl := range FirstViewTools(stdio.State(views.StatusState), nil, nil) {
 		names = append(names, tl.Name)
 	}
-	want := "mesh-pipeline-runner.show mesh-pipeline-runner.list mesh-pipeline-runner.stands mesh-pipeline-runner.why-carried mesh-pipeline-runner.kept runner_first_views runner_first_view_again"
+	// RN25 (§468, §76; A-164): the health check is the bundle's runner_health.
+	names = append(names, HealthTool(none{}, none{}, nil).Name)
+	want := "mesh-pipeline-runner.show mesh-pipeline-runner.list mesh-pipeline-runner.stands mesh-pipeline-runner.why-carried mesh-pipeline-runner.kept runner_first_views runner_first_view_again runner_health"
 	if got := strings.Join(names, " "); got != want {
 		t.Errorf("tools %q, want %q", got, want)
 	}

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

@@ -56,7 +56,7 @@ var _ pipeline.Asker = busAsker{}
 
 // hearDecided wakes the runner on the operator-channel's decided for its own asks (§37, A-89): the warrant is read from the
 // router's record, so an event the record does not back yet is heard again.
-func hearDecided(hub *process.EventHub, h *pipeline.Hearer, module string, logf func(string, ...any)) error {
+func hearDecided(hub handlers, h *pipeline.Hearer, module string, logf func(string, ...any)) error {
 	return hub.Handle(oc.SubjectDecided(module), func(m jetstream.Msg) {
 		var ev oc.Warrant
 		if err := json.Unmarshal(m.Data(), &ev); err != nil {

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

@@ -0,0 +1,57 @@
+package main
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"github.com/nats-io/nats.go/jetstream"
+
+	"mesh-delivery-pipeline/internal/pipeline"
+)
+
+// handlers is where a part of the runner registers its handler of the events it hears: the event hub, or agedHub over it.
+type handlers interface {
+	Handle(pattern string, handle func(jetstream.Msg)) error
+}
+
+// agedHub registers each handler on the hub so that every message it is handed is noted delivered, and acknowledged
+// once the handler acknowledges, refuses or ends it, so the runner's state consumers holds when the consumer's oldest
+// unacknowledged message was delivered (design 47 G §76, §468, §495; A-164, A-166).
+type agedHub struct {
+	hub  handlers
+	ages *pipeline.ConsumerAges
+}
+
+func (a agedHub) Handle(pattern string, handle func(jetstream.Msg)) error {
+	return a.hub.Handle(pattern, func(m jetstream.Msg) {
+		handle(&agedMsg{Msg: m, ages: a.ages, mark: a.ages.Delivered()})
+	})
+}
+
+// agedMsg is a message whose acknowledgement, refusal or end is noted on its consumer's ages, once.
+type agedMsg struct {
+	jetstream.Msg
+	ages *pipeline.ConsumerAges
+	mark uint64
+	once sync.Once
+}
+
+func (m *agedMsg) done() { m.once.Do(func() { m.ages.Acknowledged(m.mark) }) }
+
+func (m *agedMsg) Ack() error { defer m.done(); return m.Msg.Ack() }
+
+func (m *agedMsg) DoubleAck(ctx context.Context) error { defer m.done(); return m.Msg.DoubleAck(ctx) }
+
+func (m *agedMsg) Nak() error { defer m.done(); return m.Msg.Nak() }
+
+func (m *agedMsg) NakWithDelay(d time.Duration) error { defer m.done(); return m.Msg.NakWithDelay(d) }
+
+func (m *agedMsg) Term() error { defer m.done(); return m.Msg.Term() }
+
+func (m *agedMsg) TermWithReason(r string) error { defer m.done(); return m.Msg.TermWithReason(r) }
+
+// consumersPut writes a consumer's age in the state consumers.
+func consumersPut(ctx context.Context, kv jetstream.KeyValue) func(string, []byte) error {
+	return func(k string, v []byte) error { _, err := kv.Put(ctx, k, v); return err }
+}

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

@@ -0,0 +1,41 @@
+package main
+
+import (
+	"encoding/json"
+	"testing"
+
+	"github.com/nats-io/nats.go/jetstream"
+
+	"mesh-delivery-pipeline/internal/pipeline"
+)
+
+type oneHub struct{ handle func(jetstream.Msg) }
+
+func (h *oneHub) Handle(_ string, handle func(jetstream.Msg)) error { h.handle = handle; return nil }
+
+type ackedMsg struct{ jetstream.Msg }
+
+func (ackedMsg) Ack() error { return nil }
+
+// TestAgedHub: a message handed to a handler is noted delivered, and acknowledged when the handler acknowledges it, so
+// the state consumers holds the consumer's oldest unacknowledged delivery (§76, §468; RN25, A-166).
+func TestAgedHub(t *testing.T) {
+	var last pipeline.ConsumerAge
+	ages := &pipeline.ConsumerAges{Consumer: "n_runner", Put: func(_ string, v []byte) error { last = pipeline.ConsumerAge{}; return json.Unmarshal(v, &last) }}
+	hub := &oneHub{}
+	var inside *pipeline.ConsumerAge
+	if err := (agedHub{hub: hub, ages: ages}).Handle("x.>", func(m jetstream.Msg) {
+		seen := last
+		inside = &seen
+		_ = m.Ack()
+	}); err != nil {
+		t.Fatal(err)
+	}
+	hub.handle(ackedMsg{})
+	if inside == nil || inside.Oldest == nil {
+		t.Fatalf("inside the handler the consumer's oldest unacknowledged delivery = %+v, want it written", inside)
+	}
+	if last.Oldest != nil {
+		t.Errorf("after the handler acknowledged: oldest %v, want none", last.Oldest)
+	}
+}

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

@@ -71,6 +71,15 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	}
 	defer calls.Close()
 	hub := process.NewEventHub(logger.Printf)
+	// The age of the durable consumer's oldest unacknowledged message, read by the tool runner_health (§76, §468; A-164,
+	// A-166): written when a message is delivered or acknowledged.
+	consumers, err := process.EssentialBucket(ctx, js, "pipeline runner", ownBucket(cred.Module, pipeline.StateConsumers), "say how far its consumer is behind")
+	if err != nil {
+		return err
+	}
+	ages := &pipeline.ConsumerAges{Consumer: cred.EventConsumerName(), Put: consumersPut(ctx, consumers), Logf: logger.Printf}
+	ages.Begin()
+	heard := agedHub{hub: hub, ages: ages}
 	// A branch's view holding its first half asks again the steps placed in its running pipelines that were never asked
 	// (§88); the starter is made after the views, so it is reached through a pointer set once it is.
 	var resumer atomic.Pointer[func(string)]
@@ -112,25 +121,25 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		logger.Printf("the steps placed on %s and not yet asked: %d asked again", branch, n)
 	}
 	resumer.Store(&resume)
-	if err := hearStepEnded(hub, hearer, logger.Printf); err != nil {
+	if err := hearStepEnded(heard, hearer, logger.Printf); err != nil {
 		return err
 	}
-	if err := hearToolchain(ctx, hub, toolchains, logger.Printf); err != nil {
+	if err := hearToolchain(ctx, heard, toolchains, logger.Printf); err != nil {
 		return err
 	}
-	if err := hearAskTaken(ctx, hub, bounds, logger.Printf); err != nil {
+	if err := hearAskTaken(ctx, heard, bounds, logger.Printf); err != nil {
 		return err
 	}
 	if err := watchBounds(ctx, bounds, hearer, logger.Printf); err != nil {
 		return err
 	}
-	if err := hearDecided(hub, hearer, cred.Module, logger.Printf); err != nil {
+	if err := hearDecided(heard, hearer, cred.Module, logger.Printf); err != nil {
 		return err
 	}
-	if err := hearPublished(ctx, hub, starter, calls, rv.Catalog, logger.Printf); err != nil {
+	if err := hearPublished(ctx, heard, starter, calls, rv.Catalog, logger.Printf); err != nil {
 		return err
 	}
-	if err := hearTakenIn(hub, starter, rv.Catalog, logger.Printf); err != nil {
+	if err := hearTakenIn(heard, starter, rv.Catalog, logger.Printf); err != nil {
 		return err
 	}
 	if err := hub.StartUntilBound(ctx, js, cred, nil, nil); err != nil {

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

@@ -6,6 +6,7 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"errors"
 	"os"
 	"path/filepath"
 	"sync"
@@ -20,6 +21,8 @@ import (
 	seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner"
 	"git.novox.be/novox/mesh-sdk/go/process"
 	"git.novox.be/novox/mesh-sdk/go/testbus"
+
+	"mesh-delivery-pipeline/internal/pipeline"
 )
 
 type syncBuffer struct {
@@ -53,7 +56,7 @@ func TestTheRunnerPlacesAPipelineAsksItsBuildTakesItsEndAndAnswersAReview(t *tes
 	defer cancel()
 
 	const module = "mesh-delivery-pipeline"
-	for _, state := range []string{"pipelines", "kept", "toolchains", "findings", "first-views"} {
+	for _, state := range []string{"pipelines", "kept", "toolchains", "findings", "first-views", "consumers"} {
 		if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: module + "_" + state, History: 1}); err != nil {
 			t.Fatal(err)
 		}
@@ -157,6 +160,43 @@ func TestTheRunnerPlacesAPipelineAsksItsBuildTakesItsEndAndAnswersAReview(t *tes
 	})
 
 	waitFor(t, ctx, logs, func() bool { return readPipeline(ctx, pipelines, key, &p) && p.Outcome == seat.PipelineOutcomePassed })
+
+	// RN25 (§76, §468; A-166): the durable consumer's age is in the state consumers, none waiting once every message heard
+	// is acknowledged, and runner_health reads it healthy.
+	consumers, err := js.KeyValue(ctx, module+"_consumers")
+	if err != nil {
+		t.Fatal(err)
+	}
+	var age pipeline.ConsumerAge
+	waitFor(t, ctx, logs, func() bool {
+		e, err := consumers.Get(ctx, "n1_"+module)
+		return err == nil && json.Unmarshal(e.Value(), &age) == nil && age.Oldest == nil && age.Consumer == "n1_"+module
+	})
+	if a := pipeline.Health(kvKept{ctx, pipelines}, kvKept{ctx, consumers}, time.Now()); !a.Healthy {
+		t.Errorf("runner_health over the bus: %+v", a)
+	}
+}
+
+// kvKept reads a bucket as the tools bundle's states do.
+type kvKept struct {
+	ctx context.Context
+	kv  jetstream.KeyValue
+}
+
+func (k kvKept) Keys() ([]string, error) {
+	keys, err := k.kv.Keys(k.ctx)
+	if errors.Is(err, jetstream.ErrNoKeysFound) {
+		return nil, nil
+	}
+	return keys, err
+}
+
+func (k kvKept) Value(key string) (json.RawMessage, error) {
+	e, err := k.kv.Get(k.ctx, key)
+	if err != nil {
+		return nil, err
+	}
+	return e.Value(), nil
 }
 
 func stepOf(p seat.Pipeline, step string) seat.StepResult {

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

@@ -36,7 +36,7 @@ func (c createdKV) Create(key string, value []byte) error {
 // hearPublished starts the pipelines of each package-published heard (§53, §334): the dependents the catalog's outdated
 // names behind, each asked as a kept call. The message is acknowledged once its starts are written, on a goroutine of its
 // own, so a slow catalog holds no other event.
-func hearPublished(ctx context.Context, hub *process.EventHub, starter *pipeline.Starter, calls *process.Conn, v *pipeline.Views, logf func(string, ...any)) error {
+func hearPublished(ctx context.Context, hub handlers, starter *pipeline.Starter, calls *process.Conn, v *pipeline.Views, logf func(string, ...any)) error {
 	outdated := func(in catalog.OutdatedInput) (catalog.OutdatedAnswer, error) {
 		raw, err := calls.CallKept(outdatedVerb, in)
 		if err != nil {
@@ -72,7 +72,7 @@ func hearPublished(ctx context.Context, hub *process.EventHub, starter *pipeline
 
 // hearTakenIn starts the pipelines of each take-in.taken heard (§44, §54): one per entry, the message acknowledged once
 // they are written, and heard again when a write failed, so the keys already written start nothing twice.
-func hearTakenIn(hub *process.EventHub, starter *pipeline.Starter, v *pipeline.Views, logf func(string, ...any)) error {
+func hearTakenIn(hub handlers, starter *pipeline.Starter, v *pipeline.Views, logf func(string, ...any)) error {
 	return hub.Handle(pipeline.TakenInSubject, func(m jetstream.Msg) {
 		var ev catalog.TakeInTaken
 		if err := json.Unmarshal(m.Data(), &ev); err != nil {

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

@@ -99,7 +99,7 @@ func newHearer(ctx context.Context, js jetstream.JetStream, cred process.Account
 
 // 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 {
+func hearStepEnded(hub handlers, h *pipeline.Hearer, logf func(string, ...any)) error {
 	for _, bs := range pipeline.StepSeats {
 		bs := bs
 		err := hub.Handle(bs.Prefix+"*", func(m jetstream.Msg) {
@@ -137,7 +137,7 @@ func hearStepEnded(hub *process.EventHub, h *pipeline.Hearer, logf func(string,
 
 // hearAskTaken deletes the bound of each ask a build seat's holder took (§602, §647): the ask is answered, so its five minutes
 // end with no finding.
-func hearAskTaken(ctx context.Context, hub *process.EventHub, bounds jetstream.KeyValue, logf func(string, ...any)) error {
+func hearAskTaken(ctx context.Context, hub handlers, bounds jetstream.KeyValue, logf func(string, ...any)) error {
 	for _, ts := range pipeline.TakenSeats {
 		ts := ts
 		err := hub.Handle(ts.Subject, func(m jetstream.Msg) {
@@ -252,7 +252,7 @@ func toolchainReader(ctx context.Context, toolchains jetstream.KeyValue) func(st
 }
 
 // hearToolchain keeps each toolchain-agreed heard as `<language>` in the state toolchains (A-104): the newest of a language.
-func hearToolchain(ctx context.Context, hub *process.EventHub, toolchains jetstream.KeyValue, logf func(string, ...any)) error {
+func hearToolchain(ctx context.Context, hub handlers, toolchains jetstream.KeyValue, logf func(string, ...any)) error {
 	return hub.Handle(builds.SubjectToolchainAgreed, func(m jetstream.Msg) {
 		t, err := pipeline.ToolchainOf(m.Data())
 		if err != nil || t == nil {

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

@@ -0,0 +1,142 @@
+package pipeline
+
+import (
+	"encoding/json"
+	"fmt"
+	"sort"
+	"strings"
+	"sync"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/health"
+)
+
+// StateConsumers is the runner's old state section of its durable consumers' ages, a key per consumer (design 47 G
+// §468, §76, §495; A-164, A-166).
+const StateConsumers = "consumers"
+
+// BehindAfter is how long a consumer's oldest unacknowledged message may wait before the runner's health check names it
+// behind (§76).
+const BehindAfter = 5 * time.Minute
+
+// ConsumerAge is one durable consumer's value in the state consumers: when its oldest unacknowledged message was
+// delivered, none when every delivered message is acknowledged (A-166).
+type ConsumerAge struct {
+	Consumer string     `json:"consumer"`
+	Oldest   *time.Time `json:"oldest-unacknowledged,omitempty"`
+}
+
+// ConsumerAges follows one durable consumer's delivered and acknowledged messages and writes its ConsumerAge through
+// Put each time the oldest unacknowledged delivery changes (A-166).
+type ConsumerAges struct {
+	Consumer string
+	Put      func(key string, value []byte) error
+	Now      func() time.Time
+	Logf     func(string, ...any)
+
+	mu      sync.Mutex
+	next    uint64
+	pending map[uint64]time.Time
+	written *time.Time
+	wrote   bool
+}
+
+// Begin writes the consumer's value at the runner's start, before any message is delivered, so its health check reads
+// every consumer the process holds.
+func (c *ConsumerAges) Begin() {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.writeLocked()
+}
+
+// Delivered notes a message delivered now and answers its mark, given to Acknowledged once the message is acknowledged.
+func (c *ConsumerAges) Delivered() uint64 {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.pending == nil {
+		c.pending = map[uint64]time.Time{}
+	}
+	c.next++
+	now := time.Now().UTC()
+	if c.Now != nil {
+		now = c.Now().UTC()
+	}
+	c.pending[c.next] = now
+	c.writeLocked()
+	return c.next
+}
+
+// Acknowledged notes the message of mark acknowledged (or refused for good): it waits no more.
+func (c *ConsumerAges) Acknowledged(mark uint64) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if _, ok := c.pending[mark]; !ok {
+		return
+	}
+	delete(c.pending, mark)
+	c.writeLocked()
+}
+
+// writeLocked writes the consumer's oldest unacknowledged delivery when it differs from the one last written.
+func (c *ConsumerAges) writeLocked() {
+	var oldest *time.Time
+	for _, at := range c.pending {
+		if oldest == nil || at.Before(*oldest) {
+			at := at
+			oldest = &at
+		}
+	}
+	if c.wrote && ((oldest == nil && c.written == nil) || (oldest != nil && c.written != nil && oldest.Equal(*c.written))) {
+		return
+	}
+	raw, err := json.Marshal(ConsumerAge{Consumer: c.Consumer, Oldest: oldest})
+	if err == nil {
+		err = c.Put(c.Consumer, raw)
+	}
+	if err != nil {
+		if c.Logf != nil {
+			c.Logf("the age of the consumer %s is not written in %s: %v", c.Consumer, StateConsumers, err)
+		}
+		return
+	}
+	c.written, c.wrote = oldest, true
+}
+
+// Health is the runner's health as its tool runner_health answers it (§468, §76, §495; A-164): healthy when its state
+// pipelines answers, as show reads it, and each durable consumer's oldest unacknowledged message waits less than
+// BehindAfter; otherwise unhealthy naming each consumer behind and how long.
+func Health(pipelines, consumers Kept, now time.Time) health.Answer {
+	if _, err := pipelines.Keys(); err != nil {
+		return health.Unhealthy(fmt.Sprintf("The runner does not answer show: its state pipelines cannot be listed (%v).", err))
+	}
+	keys, err := consumers.Keys()
+	if err != nil {
+		return health.Unhealthy(fmt.Sprintf("The ages of the runner's consumers cannot be read from its state %s (%v).", StateConsumers, err))
+	}
+	if len(keys) == 0 {
+		return health.Unhealthy(fmt.Sprintf("The runner has written no consumer's age in its state %s: it has heard no event yet.", StateConsumers))
+	}
+	sort.Strings(keys)
+	var behind []string
+	for _, k := range keys {
+		raw, err := consumers.Value(k)
+		if err != nil {
+			return health.Unhealthy(fmt.Sprintf("The age of the consumer %s cannot be read (%v).", k, err))
+		}
+		var a ConsumerAge
+		if raw != nil {
+			if err := json.Unmarshal(raw, &a); err != nil {
+				return health.Unhealthy(fmt.Sprintf("The age of the consumer %s does not read (%v).", k, err))
+			}
+		}
+		if a.Oldest != nil {
+			if age := now.Sub(*a.Oldest); age >= BehindAfter {
+				behind = append(behind, fmt.Sprintf("%s (its oldest unacknowledged message waits %s)", k, age.Round(time.Second)))
+			}
+		}
+	}
+	if len(behind) > 0 {
+		return health.Unhealthy(fmt.Sprintf("The runner's consumers behind by %s or more: %s.", BehindAfter, strings.Join(behind, ", ")))
+	}
+	return health.Healthy(fmt.Sprintf("The runner answers show, and each of its %d consumers is less than %s behind.", len(keys), BehindAfter))
+}

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

@@ -0,0 +1,96 @@
+package pipeline
+
+import (
+	"encoding/json"
+	"errors"
+	"strings"
+	"testing"
+	"time"
+)
+
+type mapKept map[string]json.RawMessage
+
+func (m mapKept) Keys() ([]string, error) {
+	var out []string
+	for k := range m {
+		out = append(out, k)
+	}
+	return out, nil
+}
+func (m mapKept) Value(k string) (json.RawMessage, error) { return m[k], nil }
+
+type failingKept struct{}
+
+func (failingKept) Keys() ([]string, error)               { return nil, errors.New("no bucket") }
+func (failingKept) Value(string) (json.RawMessage, error) { return nil, errors.New("no bucket") }
+
+// TestConsumerAges: the process writes when its consumer's oldest unacknowledged message was delivered, each time it
+// changes, and none once all are acknowledged (§468, §495; A-164, A-166).
+func TestConsumerAges(t *testing.T) {
+	at := time.Date(2026, 11, 12, 10, 0, 0, 0, time.UTC)
+	now := at
+	st := mapKept{}
+	writes := 0
+	c := &ConsumerAges{Consumer: "n_mesh-delivery-pipeline", Now: func() time.Time { return now },
+		Put: func(k string, v []byte) error { writes++; st[k] = v; return nil }}
+	oldest := func() *time.Time {
+		var a ConsumerAge
+		if err := json.Unmarshal(st["n_mesh-delivery-pipeline"], &a); err != nil {
+			t.Fatal(err)
+		}
+		return a.Oldest
+	}
+	c.Begin()
+	if writes != 1 || oldest() != nil {
+		t.Fatalf("at start: %d writes, oldest %v; want one write of none", writes, oldest())
+	}
+	first := c.Delivered()
+	now = at.Add(time.Minute)
+	second := c.Delivered()
+	if writes != 2 || !oldest().Equal(at) {
+		t.Errorf("two delivered: %d writes, oldest %v; want 2 writes, oldest %v (a later delivery leaves the oldest)", writes, oldest(), at)
+	}
+	c.Acknowledged(first)
+	if !oldest().Equal(at.Add(time.Minute)) {
+		t.Errorf("the first acknowledged: oldest %v, want %v", oldest(), at.Add(time.Minute))
+	}
+	c.Acknowledged(second)
+	c.Acknowledged(second)
+	if oldest() != nil || writes != 4 {
+		t.Errorf("all acknowledged: oldest %v after %d writes, want none after 4", oldest(), writes)
+	}
+}
+
+// TestHealth: runner_health is healthy when the runner answers show and each consumer is less than five minutes behind,
+// and otherwise names each consumer behind (§76, §468, §495; RN25, A-164).
+func TestHealth(t *testing.T) {
+	now := time.Date(2026, 11, 12, 10, 0, 0, 0, time.UTC)
+	age := func(name string, since time.Duration) mapKept {
+		a := ConsumerAge{Consumer: name}
+		if since > 0 {
+			t := now.Add(-since)
+			a.Oldest = &t
+		}
+		raw, _ := json.Marshal(a)
+		return mapKept{name: raw}
+	}
+	rows := []struct {
+		name      string
+		pipelines Kept
+		consumers Kept
+		healthy   bool
+		says      string
+	}{
+		{"none waiting", mapKept{}, age("n_runner", 0), true, "less than 5m0s behind"},
+		{"four minutes", mapKept{}, age("n_runner", 4*time.Minute), true, ""},
+		{"six minutes names the consumer", mapKept{}, age("n_runner", 6*time.Minute), false, "n_runner (its oldest unacknowledged message waits 6m0s)"},
+		{"show does not answer", failingKept{}, age("n_runner", 0), false, "does not answer show"},
+		{"no consumer written", mapKept{}, mapKept{}, false, "no consumer's age"},
+	}
+	for _, r := range rows {
+		a := Health(r.pipelines, r.consumers, now)
+		if a.Healthy != r.healthy || !strings.Contains(a.Why, r.says) {
+			t.Errorf("%s: healthy %v %q; want %v saying %q", r.name, a.Healthy, a.Why, r.healthy, r.says)
+		}
+	}
+}

modules/mesh-delivery-pipeline/manifest_test.go

@@ -54,7 +54,7 @@ func TestManifest(t *testing.T) {
 		{"claims the seat serving every verb of its definition", len(m.Claims) == 1 && m.Claims[0].Name == seat.Seat &&
 			equalSets(m.Claims[0].Serves, seat.Verbs())},
 		// §27, §85, §330, §355, §77: the old state sections.
-		{"old state sections", equalSets(states, []string{seat.StatePipelines, seat.StateKept, "aged", "bounds", "toolchains", "findings", "first-views"})},
+		{"old state sections", equalSets(states, []string{seat.StatePipelines, seat.StateKept, "aged", "bounds", "toolchains", "findings", "first-views", "consumers"})},
 		// §535, §647, A-77, A-104: an ask no holder takes is bound by a key with its own time to live.
 		{"bounds is key-ttl", keyTTL["bounds"]},
 		// RN25 (§469, §85): an old state section declares its cap, pipelines 256 MiB and kept 128 MiB.
@@ -91,7 +91,9 @@ func TestManifest(t *testing.T) {
 			"seat:node-build-image.check.step"})},
 		{"uses operator-channel", equalSets(m.Uses, []string{"operator-channel"})},
 		// RN25 (§539): the runner's tools carry the prefix runner_.
-		{"first-view tools", equalSets(m.Tools, []string{"runner_first_views", "runner_first_view_again"})},
+		{"first-view tools", equalSets(m.Tools, []string{"runner_first_views", "runner_first_view_again", "runner_health"})},
+		// RN25 (§468, §76): the health check, run outside the process, is the tool runner_health.
+		{"health is runner_health", healthTool(raw) == "tool runner_health"},
 		// RN1/AC5 (§95, §130): the new assigner is delivered before the runner.
 		{"program needs the assigner's program", slices.Equal(program.Needs, []string{"mesh-delivery-assigner:program"})},
 	}
@@ -127,6 +129,28 @@ func TestNoVerbNamesItsCallers(t *testing.T) {
 	}
 }
 
+// healthTool is the process resource's health kind and tool.
+func healthTool(raw []byte) string {
+	var m struct {
+		Resources []struct {
+			Type   string `json:"type"`
+			Health struct {
+				Kind string `json:"kind"`
+				Tool string `json:"tool"`
+			} `json:"health"`
+		} `json:"resources"`
+	}
+	if err := json.Unmarshal(raw, &m); err != nil {
+		return err.Error()
+	}
+	for _, r := range m.Resources {
+		if r.Type == "process" {
+			return r.Health.Kind + " " + r.Health.Tool
+		}
+	}
+	return ""
+}
+
 func equalSets(a, b []string) bool {
 	x, y := slices.Clone(a), slices.Clone(b)
 	slices.Sort(x)

modules/mesh-delivery-pipeline/module.json

@@ -24,7 +24,8 @@
   ],
   "tools": [
     "runner_first_views",
-    "runner_first_view_again"
+    "runner_first_view_again",
+    "runner_health"
   ],
   "consumes": [
     "mesh-delivery-indexer.take-in.taken",
@@ -89,6 +90,10 @@
     {
       "name": "first-views",
       "history": 1
+    },
+    {
+      "name": "consumers",
+      "history": 1
     }
   ],
   "streams": [
@@ -146,7 +151,8 @@
       },
       "user": "mesh-delivery-pipeline",
       "health": {
-        "kind": "unit"
+        "kind": "tool",
+        "tool": "runner_health"
       }
     }
   ],