The view derives per repository up to which merge every node runs it, keeps it as its state reach and says each reached head, so a fix on the machines reads delivered (issue 1280, ADR 0556 clauses 2 and 3)

2026-10-26 20:56:12 +01:00
parent 6201f58c20
commit f907be2c14
10 changed files with 628 additions and 30 deletions
@@ -19,6 +19,18 @@ import (
 // Tools are the read tools over a modules state, and build_usage over the build modules' states.
 func Tools(r view.Reader, u view.UsageReader) []stdio.Tool {
 	return []stdio.Tool{
+		{Name: view.ReachReadTool,
+			Description: "Whether a merge is on the machines: per repository, the reached head (the newest merge every node runs a build " +
+				"containing, as the view derives it from the indexer's modules, the wants and the haves; a merge at or before it is on the " +
+				"machines), what holds the next one (a module on a node, the indexer's lag), and per module its fingerprint, the nodes " +
+				"that want it, its changes (fingerprint, the merge it first appeared at, its source revision), its floor and the nodes " +
+				"where it is not settled or not healthy (novox/hq ADR 0556). Give a repository (owner/name) for one, none for every one. Read-only.",
+			Input: map[string]any{"type": "object", "properties": map[string]any{
+				"repository": map[string]any{"type": "string", "description": "the repository, owner/name; none for every repository"}}},
+			Run: func(args map[string]any) (any, error) {
+				repository, _ := args["repository"].(string)
+				return r.Reach(repository)
+			}},
 		{Name: view.BuildUsageTool,
 			Description: "How the build nodes are used, and whether a node is eligible to build, per node (every node holding a build module or named by a capacity): each build seat it holds (node-build-go, -node, -image) with its capacity " +
 				"(mesh-config's resolved value, or the modules' default) and pause, and whether it is eligible to build and why not; the asks " +
@@ -180,7 +192,8 @@ func read(state stdio.KeptState) view.StateRead {

 func main() {
 	r := view.Reader{State: read(stdio.State("mesh-delivery-indexer.modules")), Builds: read(stdio.State("mesh-delivery-builder.builds")), CandidatesState: read(stdio.State("mesh-delivery-indexer.candidates")),
-		WantsState: read(stdio.State("mesh-delivery-assigner.wants")), DifferencesState: read(stdio.State("mesh-delivery-assigner.differences")), HavesState: read(stdio.State("node-engine.have"))}
+		WantsState: read(stdio.State("mesh-delivery-assigner.wants")), DifferencesState: read(stdio.State("mesh-delivery-assigner.differences")), HavesState: read(stdio.State("node-engine.have")),
+		ReachState: read(stdio.State("mesh-delivery-view.reach"))}
 	u := view.UsageReader{Haves: r.HavesState, Resolved: read(stdio.State("mesh-config.resolved")), Tries: read(stdio.State("mesh-delivery-checker.tries")),
 		Steps: map[string]view.StateRead{}, Pauses: map[string]view.StateRead{}, Backlog: backlogOf(socket())}
 	for _, lang := range delivery.AskLanguages {
@@ -71,6 +71,19 @@ func TestModuleToolAnswersTheStoredValue(t *testing.T) {
 	if _, err := run(t, r, view.ModuleReadTool, map[string]any{"module": ""}); err == nil {
 		t.Errorf("an empty name was answered")
 	}
+	// 1280 (ADR 0556): view_reach answers the view's reach of one repository as stored, or of every one, and says a
+	// repository it holds none of.
+	const reach = `{"v":1,"repository":"novox/web","reached":"c1","reached_src":12}`
+	r.ReachState = fake{"novox/web": reach}.reader().State
+	if out, err := run(t, r, view.ReachReadTool, map[string]any{"repository": "novox/Web"}); err != nil || string(out.(json.RawMessage)) != reach {
+		t.Errorf("view_reach novox/Web = %v, %v", out, err)
+	}
+	if out, err := run(t, r, view.ReachReadTool, map[string]any{}); err != nil || !strings.Contains(fmt.Sprint(out), "novox/web") {
+		t.Errorf("view_reach of every repository = %v, %v", out, err)
+	}
+	if _, err := run(t, r, view.ReachReadTool, map[string]any{"repository": "novox/gone"}); err == nil || !strings.Contains(err.Error(), "no reach of novox/gone") {
+		t.Errorf("a repository with no reach: %v", err)
+	}
 }

 // 1562/AC1: indexer_modules answers every key as stored, and with none2 only those that say why they have no FP2.
@@ -144,13 +144,27 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		}
 	})
 	model.SetDeliveredEvent(func(e view.DeliveredEvent) {
-		if err := emitEvent(ctx, js, cred, DeliveredEventType, e, fmt.Sprintf("delivered.%d", e.Revision)); err != nil {
+		id := fmt.Sprintf("delivered.%d", e.Revision)
+		if e.Repository != "" { // a repository's reached head (ADR 0556 clause 3)
+			key, _ := delivery.RepositoryKey(e.Repository)
+			id = fmt.Sprintf("reached.%s.%d", key, e.Revision)
+		}
+		if err := emitEvent(ctx, js, cred, DeliveredEventType, e, id); err != nil {
 			logger.Printf("the delivery of source revision %d could not be said: %v", e.Revision, err)
 		}
 	})
+	model.SetReachWriter(func(w view.ReachWrite) {
+		if err := putReach(ctx, js, w); err != nil {
+			logger.Printf("the reach of %s could not be written (written again at its next change): %v", w.Key, err)
+		}
+	})
 	model.SetBacklog(view.BacklogOver(js, 5*time.Second))
-	for _, bucket := range []string{delivery.HaveBucket, view.AssignmentsBucket, view.BuildsBucket, delivery.PresenceBucket, delivery.StuckBucket, delivery.CandidatesBucket, delivery.ModulesBucket, delivery.AsksBucket, delivery.AnswersBucket, delivery.BuilderBuildsBucket, delivery.AssignerWantsBucket, delivery.SourceBucket, delivery.ControllerComposedBucket, delivery.AssignerAsksBucket, delivery.ViewAnswersBucket} {
-		go follow(ctx, js, bucket, model, logger)
+	for _, bucket := range []string{delivery.HaveBucket, view.AssignmentsBucket, view.BuildsBucket, delivery.PresenceBucket, delivery.StuckBucket, delivery.CandidatesBucket, delivery.ModulesBucket, delivery.AsksBucket, delivery.AnswersBucket, delivery.BuilderBuildsBucket, delivery.AssignerWantsBucket, delivery.SourceBucket, delivery.ControllerComposedBucket, delivery.AssignerAsksBucket, delivery.ViewAnswersBucket, delivery.TakenBucket} {
+		go follow(ctx, js, bucket, false, model, logger)
+	}
+	// The modules' and the source's history give each fingerprint's first commit and each head's source revision (ADR 0556).
+	for _, bucket := range []string{delivery.ModulesBucket, delivery.SourceBucket} {
+		go follow(ctx, js, bucket, true, model, logger)
 	}
 	select {
 	case <-ctx.Done():
@@ -181,11 +195,21 @@ func (w kvWriter) Put(key string, value []byte) error {
 	return err
 }

+// putReach writes a reach value under its key in the view's own state reach (ADR 0556 clause 2), its only writer.
+func putReach(ctx context.Context, js jetstream.JetStream, w view.ReachWrite) error {
+	kv, err := js.KeyValue(ctx, delivery.ReachBucket)
+	if err != nil {
+		return fmt.Errorf("the bucket %s: %w", delivery.ReachBucket, err)
+	}
+	_, err = kv.Put(ctx, w.Key, w.Value)
+	return err
+}
+
 // follow watches a bucket with its last values first and then each change, for as long as ctx lives: a bucket that is
 // not there yet, or a watch that ends, is tried again.
-func follow(ctx context.Context, js jetstream.JetStream, bucket string, model *view.Model, logger *log.Logger) {
+func follow(ctx context.Context, js jetstream.JetStream, bucket string, history bool, model *view.Model, logger *log.Logger) {
 	for ctx.Err() == nil {
-		if err := watchOnce(ctx, js, bucket, model); err != nil && ctx.Err() == nil {
+		if err := watchOnce(ctx, js, bucket, history, model); err != nil && ctx.Err() == nil {
 			logger.Printf("cannot watch %s, trying again: %v", bucket, err)
 		}
 		select {
@@ -195,12 +219,17 @@ func follow(ctx context.Context, js jetstream.JetStream, bucket string, model *v
 	}
 }

-func watchOnce(ctx context.Context, js jetstream.JetStream, bucket string, model *view.Model) error {
+func watchOnce(ctx context.Context, js jetstream.JetStream, bucket string, history bool, model *view.Model) error {
 	kv, err := js.KeyValue(ctx, bucket)
 	if err != nil {
 		return err
 	}
-	w, err := kv.WatchAll(ctx)
+	var opts []jetstream.WatchOpt
+	initial := bucket
+	if history {
+		opts, initial = append(opts, jetstream.IncludeHistory()), view.ReachHistory(bucket)
+	}
+	w, err := kv.WatchAll(ctx, opts...)
 	if err != nil {
 		return err
 	}
@@ -214,11 +243,11 @@ func watchOnce(ctx context.Context, js jetstream.JetStream, bucket string, model
 				return errors.New("the watch ended")
 			}
 			if e == nil { // the last values have all been delivered
-				model.Initial(bucket)
+				model.Initial(initial)
 				continue
 			}
 			model.Apply(view.Entry{Bucket: bucket, Key: e.Key(), Value: e.Value(), At: e.Created(),
-				Deleted: e.Operation() != jetstream.KeyValuePut, Revision: e.Revision()})
+				Deleted: e.Operation() != jetstream.KeyValuePut, Revision: e.Revision(), History: history})
 		}
 	}
 }
@@ -0,0 +1,313 @@
+package view
+
+import (
+	"fmt"
+	"maps"
+	"slices"
+	"strings"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+// How far each repository reached (novox/hq ADR 0556 clauses 2 and 3, issue 1280). The view keeps, per repository the gate's
+// source names, how far each of its modules reached on the nodes that want it, and derives the repository's reached head: the
+// newest head of the repository in the source's history, not past the merge the indexer took in, at which every wanted module's
+// floor is at or after its last change at or before that head. It writes that as its state reach and says each new reached
+// head as its event delivered, per repository. Nothing here records a merge as delivered (ADR 0429 §1): a merge at or before
+// the reached head is on the machines, derived by whoever reads it.
+
+// ReachHistory names the watch of a bucket with its history, as Initial is told it: the modules' and the source's history
+// give each fingerprint's first commit and each head's source revision.
+func ReachHistory(bucket string) string { return bucket + "#history" }
+
+// reachBuckets are the watches the derivation waits for before it writes or says anything, so a start does not write a reach
+// derived from half its inputs.
+var reachBuckets = []string{delivery.ModulesBucket, delivery.AssignerWantsBucket, delivery.HaveBucket, delivery.SourceBucket,
+	delivery.TakenBucket, ReachHistory(delivery.ModulesBucket), ReachHistory(delivery.SourceBucket)}
+
+// ReachWrite is a reach value to write under its key in the view's state reach.
+type ReachWrite struct {
+	Key   string
+	Value []byte
+}
+
+// SetReachWriter sets where the view's reach values go; nothing is written until it is set.
+func (m *Model) SetReachWriter(put func(ReachWrite)) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.reachPut = put
+}
+
+// applyHistory takes one revision of a watch with history: of the indexer's modules (each module's values in order) and of
+// the gate's source (each revision's heads).
+func (m *Model) applyHistory(e Entry) {
+	switch e.Bucket {
+	case delivery.ModulesBucket:
+		if e.Deleted {
+			delete(m.modHist, e.Key)
+			return
+		}
+		mod, err := delivery.DecodeModule(e.Value)
+		if err != nil || mod.FP2 == "" || mod.Commit == "" {
+			return
+		}
+		h := m.modHist[e.Key]
+		if n := len(h); n > 0 && h[n-1].FP2 == mod.FP2 && h[n-1].Commit == mod.Commit {
+			if mod.Src != 0 {
+				h[n-1].Src = mod.Src
+			}
+			return
+		}
+		h = append(h, mod)
+		if len(h) > 2*delivery.MostChanges {
+			h = h[len(h)-delivery.MostChanges:]
+		}
+		m.modHist[e.Key] = h
+	case delivery.SourceBucket:
+		if e.Key != delivery.SourceMainKey || e.Deleted || e.Revision == 0 {
+			return
+		}
+		src, err := delivery.DecodeSource(e.Value)
+		if err != nil {
+			return
+		}
+		m.srcHist[e.Revision] = src.Heads
+	}
+}
+
+// headAt is one head of a repository and the first source revision that named it.
+type headAt struct {
+	commit string
+	rev    uint64
+}
+
+// headsOf are the heads of a repository in the source's history, oldest first, each with the first revision naming it.
+func (m *Model) headsOf(repository string) []headAt {
+	var out []headAt
+	for _, rev := range slices.Sorted(maps.Keys(m.srcHist)) {
+		for r, head := range m.srcHist[rev] {
+			if !delivery.SameRepository(r, repository) || head == "" {
+				continue
+			}
+			if n := len(out); n == 0 || !sameCommit(out[n-1].commit, head) {
+				out = append(out, headAt{commit: head, rev: rev})
+			}
+		}
+	}
+	return out
+}
+
+// revOf is the first source revision naming a commit as the repository's head, or 0 when the history holds none.
+func revOf(heads []headAt, commit string) uint64 {
+	for _, h := range heads {
+		if sameCommit(h.commit, commit) {
+			return h.rev
+		}
+	}
+	return 0
+}
+
+// sameCommit is whether two spellings of a commit are one (either may be abbreviated to at least seven characters).
+func sameCommit(a, b string) bool {
+	a, b = strings.ToLower(a), strings.ToLower(b)
+	if len(a) > len(b) {
+		a, b = b, a
+	}
+	return len(a) >= 7 && strings.HasPrefix(b, a)
+}
+
+// noBound is a module's bound when its floor is its newest change: it holds no head back.
+const noBound = ^uint64(0)
+
+// deriveReach is the reach of one repository from the model as it stands.
+func (m *Model) deriveReach(repository string) delivery.Reach {
+	r := delivery.Reach{Repository: delivery.RepositoryName(repository), Modules: map[string]delivery.ModuleReach{}}
+	heads := m.headsOf(repository)
+	key, _ := delivery.RepositoryKey(repository)
+	if t, ok := m.taken[key]; ok {
+		r.Indexed = t.Merge
+	}
+	bound := noBound
+	var holds []string
+	for _, name := range slices.Sorted(maps.Keys(m.modules)) {
+		me := m.modules[name]
+		if me.err != "" || me.value.FP2 == "" || !delivery.SameRepository(me.value.Repository, repository) {
+			continue
+		}
+		mr, b, words := m.moduleReach(name, me.value, heads)
+		r.Modules[name] = mr
+		if b < bound {
+			bound = b
+		}
+		if b != noBound {
+			holds = append(holds, words)
+		}
+	}
+	if len(r.Modules) == 0 {
+		r.Modules = nil
+	}
+	indexed := uint64(0)
+	if r.Indexed != "" {
+		indexed = revOf(heads, r.Indexed)
+	}
+	switch {
+	case r.Indexed == "":
+		holds = append(holds, "the indexer has taken in no merge of "+r.Repository)
+	case indexed == 0:
+		holds = append(holds, fmt.Sprintf("the merge the indexer took in, %.12s, is not a head of %s in the source's history", r.Indexed, r.Repository))
+	}
+	for _, h := range heads {
+		if h.rev <= indexed && h.rev < bound {
+			r.Reached, r.ReachedSrc = h.commit, h.rev
+		}
+	}
+	if n := len(heads); n > 0 && !sameCommit(heads[n-1].commit, r.Reached) {
+		if indexed != 0 && indexed < heads[n-1].rev {
+			holds = append(holds, fmt.Sprintf("the indexer has not yet taken in %.12s", heads[n-1].commit))
+		}
+		slices.Sort(holds)
+		r.Holds = slices.Compact(holds)
+	}
+	return r
+}
+
+// moduleReach is how far one module reached, the source revision it holds the reached head before (noBound when none), and
+// what holds it, in words. A node counts for the fingerprint its running have runs, healthy or not (ADR 0556 clause 2).
+func (m *Model) moduleReach(name string, mod delivery.Module, heads []headAt) (delivery.ModuleReach, uint64, string) {
+	mr := delivery.ModuleReach{FP: mod.FP2}
+	hist := m.modHist[name]
+	if n := len(hist); n == 0 || hist[n-1].FP2 != mod.FP2 || hist[n-1].Commit != mod.Commit {
+		hist = append(slices.Clone(hist), mod) // the value watched now is the newest, whatever the history watch has given
+	}
+	for i := len(hist) - 1; i >= 0 && len(mr.Changes) < delivery.MostChanges; i-- {
+		c := delivery.Change{Commit: hist[i].Commit, FP: hist[i].FP2, Src: hist[i].Src}
+		if c.Src == 0 {
+			c.Src = revOf(heads, c.Commit)
+		}
+		if n := len(mr.Changes); n > 0 && mr.Changes[n-1].FP == c.FP {
+			mr.Changes[n-1] = c // the same fingerprint again, back to back: its first appearance is the older commit
+			continue
+		}
+		mr.Changes = append(mr.Changes, c)
+	}
+	var waiting []string
+	floor, floorNode, unset := -1, "", ""
+	for _, key := range slices.Sorted(maps.Keys(m.wants)) {
+		node, module := splitKey(key)
+		w := m.wants[key]
+		if module != name || w.err != "" || w.value.Removed || w.value.FP == "" {
+			continue
+		}
+		mr.Nodes++
+		h, ok := m.haves[key]
+		switch {
+		case !ok || h.err != "" || h.have.FP == "":
+			waiting = append(waiting, fmt.Sprintf("%s on %s runs nothing yet", name, node))
+			unset = waiting[len(waiting)-1]
+			continue
+		case !runs(h.have.Phase):
+			waiting = append(waiting, fmt.Sprintf("%s on %s is %s at %s, not running", name, node, h.have.Phase, short(h.have.FP)))
+			unset = waiting[len(waiting)-1]
+			continue
+		case h.have.FP != w.value.FP:
+			waiting = append(waiting, fmt.Sprintf("%s on %s runs %s, not yet the %s it wants", name, node, short(h.have.FP), short(w.value.FP)))
+		case h.have.Health == delivery.HealthFailed:
+			waiting = append(waiting, fmt.Sprintf("%s on %s runs %s and its health check failed", name, node, short(h.have.FP)))
+		case h.have.Health != delivery.HealthOK:
+			waiting = append(waiting, fmt.Sprintf("%s on %s runs %s, not yet healthy", name, node, short(h.have.FP)))
+		}
+		i := slices.IndexFunc(mr.Changes, func(c delivery.Change) bool { return c.FP == h.have.FP })
+		if i < 0 {
+			unset = fmt.Sprintf("%s on %s runs %s, which is not among its %d known changes", name, node, short(h.have.FP), len(mr.Changes))
+			waiting = append(waiting, unset)
+			continue
+		}
+		if i > floor {
+			floor, floorNode = i, node
+		}
+	}
+	mr.Waiting = strings.Join(waiting, "; ")
+	switch {
+	case mr.Nodes == 0:
+		return mr, noBound, ""
+	case unset != "":
+		return mr, 0, unset
+	}
+	mr.Floor = mr.Changes[floor].Commit
+	if floor == 0 {
+		return mr, noBound, ""
+	}
+	next := mr.Changes[floor-1]
+	words := fmt.Sprintf("%s on %s runs the build of %.12s, not yet that of %.12s", name, floorNode, mr.Floor, next.Commit)
+	if next.Src == 0 {
+		// The next change's order is not known: the head is held at the floor's own revision, or before every head when that is
+		// not known either.
+		if own := mr.Changes[floor].Src; own != 0 {
+			return mr, own + 1, words
+		}
+		return mr, 0, words
+	}
+	return mr, next.Src, words
+}
+
+// runs is whether a have in this phase runs the fingerprint it names: running, being checked after its apply, or held after a
+// failed backup, which keeps running what it has (ADR 0539 clause 16). A have backing up or applying, failed or removed runs
+// nothing the derivation can count.
+func runs(p delivery.Phase) bool {
+	return p == delivery.PhaseRunning || p == delivery.PhaseChecking || p == delivery.PhaseHeld
+}
+
+// short is a fingerprint's first twelve characters, after its form.
+func short(fp string) string {
+	fp = strings.TrimPrefix(fp, delivery.SourcePrefix2)
+	if len(fp) > 12 {
+		fp = fp[:12]
+	}
+	return fp
+}
+
+// reachReady is whether every watch the derivation reads handed over what stood at its start.
+func (m *Model) reachReady() bool {
+	for _, b := range reachBuckets {
+		if !m.initial[b] {
+			return false
+		}
+	}
+	return m.source != nil && m.source.err == ""
+}
+
+// reachNow derives the reach of every repository the source names, and answers the values that changed and the reached
+// heads not yet said. It is called with the lock held.
+func (m *Model) reachNow() ([]ReachWrite, []DeliveredEvent) {
+	if (m.reachPut == nil && m.deliveredEvent == nil) || !m.reachReady() {
+		return nil, nil
+	}
+	var writes []ReachWrite
+	var events []DeliveredEvent
+	for _, repository := range slices.Sorted(maps.Keys(m.source.heads)) {
+		key, err := delivery.RepositoryKey(repository)
+		if err != nil {
+			continue
+		}
+		r := m.deriveReach(repository)
+		if prev, ok := m.reach[key]; !ok || !delivery.SameReach(prev, r) {
+			if data, err := delivery.EncodeReach(r); err == nil {
+				m.reach[key] = r
+				writes = append(writes, ReachWrite{Key: key, Value: data})
+			}
+		}
+		if r.Reached != "" && r.ReachedSrc > m.reachSaid[key] {
+			m.reachSaid[key] = r.ReachedSrc
+			events = append(events, DeliveredEvent{Revision: r.ReachedSrc, Heads: map[string]string{r.Repository: r.Reached},
+				Repository: r.Repository})
+		}
+	}
+	return writes, events
+}
+
+// Reach is the reach of every repository as the view derived it last, by key.
+func (m *Model) Reach() map[string]delivery.Reach {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return maps.Clone(m.reach)
+}
@@ -0,0 +1,140 @@
+package view
+
+import (
+	"strings"
+	"testing"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+// 1280 (ADR 0556 clauses 2 and 3): the view derives, per repository, how far each module reached and the reached head, writes
+// it when it differs, and says each new reached head once as its event delivered.
+func TestTheViewDerivesHowFarEachRepositoryReached(t *testing.T) {
+	const web, sdk = "novox/web", "novox/sdk"
+	commitA, commitB, commitC := strings.Repeat("a", 40), strings.Repeat("b", 40), strings.Repeat("c", 40)
+	commitS, commitX := strings.Repeat("5", 40), strings.Repeat("e", 40)
+	const fpMail = "src2:3333333333333333333333333333333333333333333333333333333333333333"
+	mod := func(module, fp, commit string, src uint64, history bool) Entry {
+		v, err := delivery.EncodeModule(delivery.Module{FP2: fp, Manifest: strings.Repeat("a", 64), Repository: web, Commit: commit, Src: src, Path: module})
+		if err != nil {
+			t.Fatal(err)
+		}
+		return Entry{Bucket: delivery.ModulesBucket, Key: module, Value: v, At: t0, History: history}
+	}
+	src := func(rev uint64, webHead string) Entry {
+		v, err := delivery.EncodeSource(delivery.Source{Group: "g", Heads: map[string]string{web: webHead, sdk: commitS}})
+		if err != nil {
+			t.Fatal(err)
+		}
+		return Entry{Bucket: delivery.SourceBucket, Key: delivery.SourceMainKey, Value: v, At: t0, Revision: rev, History: true}
+	}
+	taken := func(repository, merge string) Entry {
+		key, _ := delivery.RepositoryKey(repository)
+		v, err := delivery.EncodeTaken(delivery.Taken{Repository: repository, Merge: merge, At: t0})
+		if err != nil {
+			t.Fatal(err)
+		}
+		return Entry{Bucket: delivery.TakenBucket, Key: key, Value: v, At: t0}
+	}
+	haveIn := func(node, module, fp string, phase delivery.Phase, health delivery.Health) Entry {
+		e := haveFP(t, node, module, fp, health)
+		v, _ := delivery.EncodeHave(delivery.Have{FP: fp, Phase: phase, Health: health})
+		e.Value = v
+		return e
+	}
+	// web changed at A (fpOld) and at C (fpNew); mail first appeared at A. The source named A, B, then C as web's head.
+	world := func(t *testing.T) []Entry {
+		cur, _ := delivery.DecodeSource(src(12, commitC).Value)
+		current := Entry{Bucket: delivery.SourceBucket, Key: delivery.SourceMainKey, Revision: 12, At: t0}
+		current.Value, _ = delivery.EncodeSource(cur)
+		return []Entry{src(10, commitA), src(11, commitB), src(12, commitC), current,
+			mod("web", fpOld, commitA, 10, true), mod("web", fpNew, commitC, 12, true), mod("mail", fpMail, commitA, 0, true),
+			mod("web", fpNew, commitC, 12, false), mod("mail", fpMail, commitA, 0, false),
+			taken(web, commitC), taken(sdk, commitS),
+			wantE(t, "ace", "web", delivery.Want{FP: fpNew}), haveFP(t, "ace", "web", fpNew, delivery.HealthOK),
+			wantE(t, "g14", "web", delivery.Want{FP: fpNew}), haveFP(t, "g14", "web", fpNew, delivery.HealthOK),
+			wantE(t, "ace", "mail", delivery.Want{FP: fpMail}), haveFP(t, "ace", "mail", fpMail, delivery.HealthOK)}
+	}
+	for _, tc := range []struct {
+		name    string
+		then    []Entry
+		reached string // web's reached head; "" none
+		holds   string // words one of web's holds carries; "" none
+		waiting string // words web's module web waits with; "" none
+	}{
+		{name: "every node runs the newest build", reached: commitC},
+		{name: "a node lagging stops the head before the change it lacks", reached: commitB, holds: "web on g14",
+			then: []Entry{haveFP(t, "g14", "web", fpOld, delivery.HealthOK)}, waiting: "not yet the"},
+		{name: "a want held back at an older build holds the head as a lag does", reached: commitB, holds: "web on g14 runs the build of",
+			then: []Entry{wantE(t, "g14", "web", delivery.Want{FP: fpOld, Newest: fpNew, Rule: 2}), haveFP(t, "g14", "web", fpOld, delivery.HealthOK)}},
+		{name: "a failing health check is said and holds nothing", reached: commitC, waiting: "health check failed",
+			then: []Entry{haveFP(t, "g14", "web", fpNew, delivery.HealthFailed)}},
+		{name: "a node with no running have leaves no reached head", reached: "", holds: "not running", waiting: "not running",
+			then: []Entry{haveIn("g14", "web", fpNew, delivery.PhaseApplying, delivery.HealthUnknown)}},
+		{name: "a module no node wants holds nothing", reached: commitC,
+			then: []Entry{wantE(t, "ace", "mail", delivery.Want{Removed: true})}},
+		{name: "the indexer not yet having taken the head in", reached: commitB, holds: "has not yet taken in",
+			then: []Entry{taken(web, commitB)}},
+		{name: "a change whose source revision is not known yet holds no head the source named", reached: commitC,
+			then: []Entry{mod("web", "src2:"+strings.Repeat("9", 64), commitX, 0, true), mod("web", "src2:"+strings.Repeat("9", 64), commitX, 0, false)}},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			m := New()
+			var written []ReachWrite
+			var said []DeliveredEvent
+			m.SetReachWriter(func(w ReachWrite) { written = append(written, w) })
+			m.SetDeliveredEvent(func(e DeliveredEvent) {
+				if e.Repository != "" {
+					said = append(said, e)
+				}
+			})
+			fill(m, world(t)...)
+			if len(written) != 0 {
+				t.Fatalf("written before every watch handed over: %v", written)
+			}
+			for _, b := range reachBuckets {
+				m.Initial(b)
+			}
+			fill(m, tc.then...)
+			key, _ := delivery.RepositoryKey(web)
+			r := m.Reach()[key]
+			if r.Reached != tc.reached {
+				t.Errorf("reached = %.8s, want %.8s (holds %v)", r.Reached, tc.reached, r.Holds)
+			}
+			if got := strings.Join(r.Holds, "; "); (tc.holds == "") != (got == "") || !strings.Contains(got, tc.holds) {
+				t.Errorf("holds = %q, want words with %q", got, tc.holds)
+			}
+			if got := r.Modules["web"].Waiting; (tc.waiting == "") != (got == "") || !strings.Contains(got, tc.waiting) {
+				t.Errorf("web waits with %q, want words with %q", got, tc.waiting)
+			}
+			// A library reaches the head the indexer took in, with no module.
+			sk, _ := delivery.RepositoryKey(sdk)
+			if lib := m.Reach()[sk]; lib.Reached != commitS || len(lib.Modules) != 0 {
+				t.Errorf("the library's reach = %+v", lib)
+			}
+			// Each key is written once with its last value; a reached head is said once, for its repository alone.
+			last := map[string]delivery.Reach{}
+			for _, w := range written {
+				v, err := delivery.DecodeReach(w.Value)
+				if err != nil {
+					t.Fatalf("written %s: %v", w.Key, err)
+				}
+				last[w.Key] = v
+			}
+			if !delivery.SameReach(last[key], r) {
+				t.Errorf("the last value written is %+v, not the reach derived %+v", last[key], r)
+			}
+			n := len(said)
+			fill(m, tc.then...)
+			fill(m, world(t)[len(world(t))-1])
+			if len(said) != n && tc.reached == commitC {
+				t.Errorf("said again: %+v", said[n:])
+			}
+			for _, e := range said {
+				if len(e.Heads) != 1 || e.Group != "" || e.Revision == 0 {
+					t.Errorf("a reached head said as %+v", e)
+				}
+			}
+		})
+	}
+}
@@ -39,6 +39,38 @@ type Reader struct {
 	WantsState       StateRead // the assigner's wants and holds
 	DifferencesState StateRead // the assigner's differences
 	HavesState       StateRead // the node-engine's haves
+	ReachState       StateRead // the view's own reach (ADR 0556)
+}
+
+// ReachReadTool reads the view's reach: per repository, how far each module reached and the reached head (ADR 0556).
+const ReachReadTool = "view_reach"
+
+// Reach is the stored reach of one repository, or of every repository when none is named, as the view wrote it.
+func (r Reader) Reach(repository string) (any, error) {
+	if r.ReachState.Get == nil || r.ReachState.Keys == nil {
+		return nil, fmt.Errorf("the reach state cannot be read here")
+	}
+	if repository = strings.TrimSpace(repository); repository != "" {
+		key, err := delivery.RepositoryKey(repository)
+		if err != nil {
+			return nil, err
+		}
+		raw, err := r.ReachState.Get(key)
+		if err != nil {
+			return nil, err
+		}
+		if len(raw) == 0 {
+			return nil, fmt.Errorf("no reach of %s: the view has derived none (the gate's source does not name it, or the view waits for a watch)", repository)
+		}
+		return raw, nil
+	}
+	keys, err := r.ReachState.Keys()
+	if err != nil {
+		return nil, err
+	}
+	sort.Strings(keys)
+	values, err := r.ReachState.values(keys)
+	return map[string]any{"reach": values}, err
 }

 // Module is the stored value of modules.<module>, exactly as the indexer wrote it.
@@ -23,6 +23,8 @@ type Entry struct {
 	Deleted bool
 	// Revision is the bucket revision of the change, when the watch gives one (the source revision is the gate's source key's).
 	Revision uint64
+	// History says the change comes from a watch with history (ReachHistory): it feeds only the history the reach reads.
+	History bool
 }

 // Row is one module on one node.
@@ -240,6 +242,13 @@ type Model struct {
 	// the watchman's buckets, by key
 	presence map[string]presenceEntry
 	stuck    map[string]stuckEntry
+	// how far each repository reached (reach.go, ADR 0556)
+	modHist   map[string][]delivery.Module // the indexer's values of each module, oldest first
+	srcHist   map[uint64]map[string]string // the source's heads, by revision
+	taken     map[string]delivery.Taken    // the indexer's taken, by repository key
+	reach     map[string]delivery.Reach    // the reach derived last, by repository key
+	reachSaid map[string]uint64            // the reached head's source revision said last, by repository key
+	reachPut  func(ReachWrite)
 }

 type presenceEntry struct {
@@ -348,21 +357,41 @@ type nodeEntry struct {
 func New() *Model {
 	return &Model{assigns: map[string]assignment{}, builds: map[string]buildEntry{}, groups: map[string]groupEntry{}, modules: map[string]moduleEntry{}, keys: map[string]keyEntry{}, asks: map[string]askRecord{}, wants: map[string]wantEntry{}, told: map[string]bool{}, initial: map[string]bool{}, composed: map[string]composedEntry{}, aasks: map[string]assignerAskRecord{}, chosen: map[string]delivery.ViewAnswer{}, answers: map[string]string{}, answerOf: map[string]delivery.Answer{}, holds: map[string]heldEntry{}, newest: map[string]string{},
 		haves: map[string]haveEntry{}, nodes: map[string]nodeEntry{},
-		presence: map[string]presenceEntry{}, stuck: map[string]stuckEntry{}}
+		presence: map[string]presenceEntry{}, stuck: map[string]stuckEntry{},
+		modHist: map[string][]delivery.Module{}, srcHist: map[uint64]map[string]string{}, taken: map[string]delivery.Taken{},
+		reach: map[string]delivery.Reach{}, reachSaid: map[string]uint64{}}
 }

 // Apply takes one change of a key. A value it cannot read is kept as an error on its row, never guessed. When the change makes
 // the source revision delivered, the event of it is said once (SetDeliveredEvent), after the model's lock is released.
 func (m *Model) Apply(e Entry) {
 	m.mu.Lock()
-	m.apply(e)
-	event := m.deliveredNow()
-	say := m.deliveredEvent
-	motion, sayMotion := m.motionNow(e), m.motionEvent
+	var event *DeliveredEvent
+	var motion []MotionEvent
+	if e.History {
+		m.applyHistory(e)
+	} else {
+		m.apply(e)
+		event = m.deliveredNow()
+		motion = m.motionNow(e)
+	}
+	writes, reached := m.reachNow()
+	say, put := m.deliveredEvent, m.reachPut
+	sayMotion := m.motionEvent
 	m.mu.Unlock()
 	if event != nil && say != nil {
 		say(*event)
 	}
+	if put != nil {
+		for _, w := range writes {
+			put(w)
+		}
+	}
+	if say != nil {
+		for _, ev := range reached {
+			say(ev)
+		}
+	}
 	if sayMotion != nil {
 		for _, ev := range motion {
 			sayMotion(ev)
@@ -373,10 +402,15 @@ func (m *Model) Apply(e Entry) {
 // DeliveredEvent is what the view says when the source revision is delivered (event `delivered` of the view, issue 1906 D9, ADR
 // 0539 §17): the revision of the gate's source key, the group that moved it and every repository's merged head at it. A
 // consumer takes every earlier revision as covered by it, and takes the same revision twice as once (after a restart of the view).
+//
+// The same event says a repository's reached head (ADR 0556 clause 3): Repository names it, Heads holds its head alone, Revision
+// is the source revision that named that head and Group is empty; a consumer takes every earlier merge of that repository as on
+// the machines.
 type DeliveredEvent struct {
-	Revision uint64            `json:"revision"`
-	Group    string            `json:"group"`
-	Heads    map[string]string `json:"heads"`
+	Revision   uint64            `json:"revision"`
+	Group      string            `json:"group"`
+	Heads      map[string]string `json:"heads"`
+	Repository string            `json:"repository,omitempty"`
 }

 // SetDeliveredEvent sets where the event of a delivered source revision goes; nothing is said until it is set.
@@ -512,6 +546,16 @@ func (m *Model) apply(e Entry) {
 		}
 		d, err := delivery.DecodeComposedDigest(e.Value)
 		m.composed[e.Key] = composedEntry{value: d, err: errText(err)}
+	case delivery.TakenBucket:
+		if e.Deleted {
+			delete(m.taken, e.Key)
+			return
+		}
+		if t, err := delivery.DecodeTaken(e.Value); err == nil {
+			m.taken[e.Key] = t
+		} else {
+			delete(m.taken, e.Key)
+		}
 	case delivery.SourceBucket:
 		if e.Key != delivery.SourceMainKey {
 			return
@@ -1214,10 +1258,22 @@ func (m *Model) Initial(bucket string) {
 		}
 	}
 	say := m.motionEvent
+	writes, reached := m.reachNow()
+	put, sayDelivered := m.reachPut, m.deliveredEvent
 	m.mu.Unlock()
 	for _, ev := range out {
 		say(ev)
 	}
+	if put != nil {
+		for _, w := range writes {
+			put(w)
+		}
+	}
+	if sayDelivered != nil {
+		for _, ev := range reached {
+			sayDelivered(ev)
+		}
+	}
 }

 // motionOf is the events to say now about the node and module of a model key, from their want, have and stuck key.
@@ -110,7 +110,7 @@ func TestWaitsForTheNodeAndForANewerBuild(t *testing.T) { // 1134/AC2
 	}
 }

-func TestLeftBehindIsNotRunning(t *testing.T) { // 1134/AC3
+func TestLeftBehindIsNotRunningAndPulledInIs(t *testing.T) { // 1134/AC3 (both halves)
 	for name, extra := range map[string]string{"ended": `,"ended":"2026-10-18T11:30:00Z"`, "state": `,"state":"pending"`} {
 		m := New()
 		fill(m, base(t)...)
@@ -131,13 +131,11 @@ func TestLeftBehindIsNotRunning(t *testing.T) { // 1134/AC3
 	if find(t, s, "a", "w").Status != StatusLeftBehind || s.Counted() != 4 {
 		t.Fatalf("a have with no assignment: %+v counted %d", find(t, s, "a", "w"), s.Counted())
 	}
-}
-
-func TestPulledInIsRunning(t *testing.T) { // 1134/AC3
-	m := New()
+	// A module pulled in by another stands: shown and counted, with what required it.
+	m = New()
 	fill(m, base(t)...)
 	fill(m, assign("b", "z", "b3", `,"pulled-in":true,"required-by":"x"`))
-	s := m.Snapshot()
+	s = m.Snapshot()
 	r := find(t, s, "b", "z")
 	if r.Status != StatusAssigned || r.RequiredBy != "x" || r.Have != "b3" || s.Counted() != 4 {
 		t.Fatalf("pulled-in row = %+v counted %d, want shown and counted with required-by x", r, s.Counted())
@@ -84,7 +84,7 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7
 	if m.Module != "mesh-delivery-view" {
 		t.Errorf("module = %q", m.Module)
 	}
-	want := []string{"mesh-build-go.pauses", "mesh-build-go.steps", "mesh-build-image.pauses", "mesh-build-image.steps", "mesh-build-node.pauses", "mesh-build-node.steps", "mesh-config.resolved", "mesh-controller.assignments", "mesh-controller.builds", "mesh-controller.composed", "mesh-delivery-assigner.asks", "mesh-delivery-assigner.differences", "mesh-delivery-assigner.wants", "mesh-delivery-builder.builds", "mesh-delivery-checker.tries", "mesh-delivery-gate.asks", "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.modules", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "mesh-delivery-watchman.stuck", "node-engine.have"} // build_usage reads the build modules' steps and pauses, mesh-config's capacities and the checker's tries; 1168/AC1, 1241/AC4, 1345/AC3, 1297/B8, B9
+	want := []string{"mesh-build-go.pauses", "mesh-build-go.steps", "mesh-build-image.pauses", "mesh-build-image.steps", "mesh-build-node.pauses", "mesh-build-node.steps", "mesh-config.resolved", "mesh-controller.assignments", "mesh-controller.builds", "mesh-controller.composed", "mesh-delivery-assigner.asks", "mesh-delivery-assigner.differences", "mesh-delivery-assigner.wants", "mesh-delivery-builder.builds", "mesh-delivery-checker.tries", "mesh-delivery-gate.asks", "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.modules", "mesh-delivery-indexer.taken", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "mesh-delivery-watchman.stuck", "node-engine.have"} // build_usage reads the build modules' steps and pauses, mesh-config's capacities and the checker's tries; 1168/AC1, 1241/AC4, 1345/AC3, 1297/B8, B9
 	got := slices.Clone(m.Reads)
 	slices.Sort(got)
 	if !reflect.DeepEqual(got, want) {
@@ -103,8 +103,9 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7
 	if len(m.Uses) != 0 {
 		t.Errorf("uses = %v, want none", m.Uses)
 	}
-	// The one state the view writes: the operator's answers to the asks that need no proof (ADR 0528 clause 6, 1297/B9), which the teller reads.
-	if want := []state{{Name: "answers", History: 5}}; !reflect.DeepEqual(m.State, want) {
+	// The states the view writes: the operator's answers to the asks that need no proof (ADR 0528 clause 6, 1297/B9), which the
+	// teller reads, and how far each repository reached (ADR 0556 clause 2, issue 1280), which the gate reads.
+	if want := []state{{Name: "answers", History: 5}, {Name: "reach", History: 1}}; !reflect.DeepEqual(m.State, want) {
 		t.Errorf("state = %+v, want %+v", m.State, want)
 	}
 	if _, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 {
@@ -115,7 +116,7 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7
 	}
 	// The two read tools of the indexer's modules (issue 1562), carried by a bundle the runtime launches; the indexer
 	// itself serves no tool (ADR 0437 §4). Carrying them gives the tool runner read access to every bucket the view reads.
-	if want := []string{"indexer_module", "indexer_modules", "indexer_candidate", "indexer_candidates", "builder_build", "builder_builds", "assigner_wants", "assigner_want", "assigner_differences", "node_haves", "build_usage"}; !reflect.DeepEqual(m.Tools, want) {
+	if want := []string{"indexer_module", "indexer_modules", "indexer_candidate", "indexer_candidates", "builder_build", "builder_builds", "assigner_wants", "assigner_want", "assigner_differences", "node_haves", "build_usage", "view_reach"}; !reflect.DeepEqual(m.Tools, want) {
 		t.Errorf("tools = %v, want %v", m.Tools, want)
 	}
 	for _, a := range m.Build.Artifacts {
@@ -14,7 +14,8 @@
    "assigner_want",
    "assigner_differences",
    "node_haves",
-    "build_usage"
+    "build_usage",
+    "view_reach"
  ],
  "emits": [
    "delivered",
@@ -23,7 +24,8 @@
    "stuck"
  ],
  "state": [
-    { "name": "answers", "history": 5 }
+    { "name": "answers", "history": 5 },
+    { "name": "reach", "history": 1 }
  ],
  "reads": [
    "mesh-controller.assignments",
@@ -45,6 +47,7 @@
    "mesh-delivery-gate.source",
    "mesh-delivery-indexer.candidates",
    "mesh-delivery-indexer.modules",
+    "mesh-delivery-indexer.taken",
    "mesh-delivery-teller.answers",
    "mesh-delivery-watchman.presence",
    "mesh-delivery-watchman.stuck",