DLV-M5: the watchman follows the assigner's wants and the indexer's modules by their read tools' first views, read once answered

ADR 0615 §1: the watchman invokes assigner_read_wants and indexer_read_modules, consumes their views
families and never reads either bucket; it keeps first-views and writes an unanswered first view as a
finding of its own. The lead's ruling: an owner that is down never holds the start; each state is read
on a goroutine once its first view is answered, and what it holds (seeded pulses, stuck keys, the
findings it is the source of) is reconciled then (watch.Awaits, ReadLate), never dropped at the start's
end. Tools watchman_first_views and watchman_first_view_again.

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

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

@@ -10,8 +10,10 @@ import (
 	"os"
 	"sort"
 	"strings"
+	"time"
 
 	stdio "git.novox.be/novox/mesh-sdk/go"
+	"git.novox.be/novox/mesh-sdk/go/views"
 
 	"mesh-delivery-watchman/internal/socket"
 	"mesh-delivery-watchman/internal/watch"
@@ -28,18 +30,22 @@ const (
 	VerbDeadLetters = "dead-letters"
 	VerbResync      = "resync"
 	ToolFindings    = "watchman_findings"
+	// ToolFirstViews and ToolFirstViewAgain are where the watchman's first views of the assigner's wants and the indexer's
+	// modules stand, and asking one again by hand (novox/hq ADR 0615, DLV-M5).
+	ToolFirstViews     = "watchman_first_views"
+	ToolFirstViewAgain = "watchman_first_view_again"
 )
 
 // Verbs are the seat's verbs, in the manifest's order.
 var Verbs = []string{VerbSignals, VerbStuck, VerbPresence, VerbDeadLetters, VerbResync}
 
 // ModuleTools are the module's own tools, in the manifest's order.
-var ModuleTools = []string{ToolFindings}
+var ModuleTools = []string{ToolFindings, ToolFirstViews, ToolFirstViewAgain}
 
 // Reads names, per tool, the states it reads whole; Acts, per tool, the program's act it starts. The clause-12 test holds
 // that every state the manifest keeps is read by one, and every act the program takes by hand is started by one.
 var (
-	Reads = map[string][]string{VerbSignals: {"pulse"}, VerbStuck: {"stuck"}, VerbPresence: {"presence"}, ToolFindings: {"findings"}}
+	Reads = map[string][]string{VerbSignals: {"pulse"}, VerbStuck: {"stuck"}, VerbPresence: {"presence"}, ToolFindings: {"findings"}, ToolFirstViews: {views.StatusState}}
 	Acts  = map[string][]string{VerbDeadLetters: {socket.AskDeadLetters}, VerbResync: {socket.AskResync}}
 )
 
@@ -124,7 +130,7 @@ func Tools(states func(string) state, ask func(socket.Ask) (json.RawMessage, err
 	}
 	resync := func(map[string]any) (any, error) { return ask(socket.Ask{Ask: socket.AskResync}) }
 	verb := func(v string) string { return Seat + "." + v }
-	return []stdio.Tool{
+	return append([]stdio.Tool{
 		{Name: verb(VerbSignals), Input: keyArgs, Run: signals,
 			Description: "The signals table's rows the watchman holds (S1, S8, S9, S10, S11, S21 and the haves' health), each a pulse key " +
 				"whose time to live is its bound and whose expiry is its finding (novox/hq ADR 0564 §4), and every pulse the watchman " +
@@ -157,7 +163,7 @@ func Tools(states func(string) state, ask func(socket.Ask) (json.RawMessage, err
 			Description: "The watchman's findings (novox/hq ADR 0564 §2): each thing it observes wrong now, <scope>.<id>.<kind>, with its " +
 				"severity, subject, evidence and the row that observed it; the condition keeper raises and clears the conditions from " +
 				"them. Give key for one, prefix to narrow. Read-only."},
-	}
+	}, views.StatusTools(watch.Observer, ToolFirstViews, ToolFirstViewAgain, stdio.State(views.StatusState), time.Now)...)
 }
 
 func main() {

modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go

@@ -17,6 +17,7 @@ import (
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
 
+	"git.novox.be/novox/mesh-delivery/sdk/firstview/firstviewtest"
 	"git.novox.be/novox/mesh-sdk/go/buscompose"
 	"git.novox.be/novox/mesh-sdk/go/delivery"
 	"git.novox.be/novox/mesh-sdk/go/process"
@@ -57,6 +58,14 @@ accounts {
 }
 `
 
+// serveOwners serves the assigner's wants and the indexer's modules as their owners do, by their read tools and views
+// families (DLV-M5, ADR 0615): the watchman (n1) follows both, never their buckets.
+func serveOwners(t *testing.T, ctx context.Context, js jetstream.JetStream, nc *nats.Conn) {
+	t.Helper()
+	firstviewtest.Serve(t, ctx, js, nc, delivery.AssignerModule, wantsState, "n1_"+delivery.WatchmanModule)
+	firstviewtest.Serve(t, ctx, js, nc, delivery.IndexerModule, modulesState, "n1_"+delivery.WatchmanModule)
+}
+
 // TestWatchmanWhole starts the program the way the node does (its entry, run, with its environment and a broker file)
 // against an embedded bus with the system exports and JetStream, and proves the four things the issue names.
 func TestWatchmanWhole(t *testing.T) { // 1135/AC8
@@ -109,6 +118,7 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8
 		{Bucket: delivery.AssignerWantsBucket, History: 20},
 		{Bucket: delivery.HaveBucket, History: 20},
 		{Bucket: watch.FindingsBucket, History: 5},
+		{Bucket: firstViewsBucket, History: 1},
 	} {
 		kv, err := js.CreateKeyValue(ctx, c)
 		if err != nil {
@@ -116,6 +126,7 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8
 		}
 		kvs[c.Bucket] = kv
 	}
+	serveOwners(t, ctx, js, ctl)
 
 	// The controller's consumer on DEAD_LETTER_NOTICES, the dead letters' streams and a source stream with one message a
 	// consumer will give up on (1952/WM5); no lease bucket at all (issue 1126).
@@ -355,6 +366,7 @@ func TestAnAskRunningBeforeTheStartGetsItsPulse(t *testing.T) { // 1909
 		{Bucket: delivery.AssignerWantsBucket, History: 20},
 		{Bucket: delivery.HaveBucket, History: 20},
 		{Bucket: watch.FindingsBucket, History: 5},
+		{Bucket: firstViewsBucket, History: 1},
 	} {
 		kv, err := js.CreateKeyValue(ctx, c)
 		if err != nil {
@@ -362,6 +374,7 @@ func TestAnAskRunningBeforeTheStartGetsItsPulse(t *testing.T) { // 1909
 		}
 		kvs[c.Bucket] = kv
 	}
+	serveOwners(t, ctx, js, ctl)
 	const id = "m0000009ffffff"
 	body, err := delivery.EncodeStep(delivery.Step{ID: id, Tool: "build_go_test", Node: "n1", State: delivery.StepRunning, Started: time.Now().UTC()})
 	if err != nil {
@@ -440,6 +453,7 @@ func TestAWantIsMirroredAsAPulseAndTheHaveNamingItEndsIt(t *testing.T) {
 		{Bucket: delivery.AssignerWantsBucket, History: 20},
 		{Bucket: delivery.HaveBucket, History: 20},
 		{Bucket: watch.FindingsBucket, History: 5},
+		{Bucket: firstViewsBucket, History: 1},
 	} {
 		kv, err := js.CreateKeyValue(ctx, c)
 		if err != nil {
@@ -447,6 +461,7 @@ func TestAWantIsMirroredAsAPulseAndTheHaveNamingItEndsIt(t *testing.T) {
 		}
 		kvs[c.Bucket] = kv
 	}
+	serveOwners(t, ctx, js, ctl)
 	fpNew := delivery.SourcePrefix2 + strings.Repeat("b", 64)
 	fpOld := delivery.SourcePrefix2 + strings.Repeat("a", 64)
 	putWant := func(node, module, fp string) {

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

@@ -18,6 +18,9 @@ import (
 	"time"
 
 	"git.novox.be/novox/mesh-sdk/go/process"
+	"git.novox.be/novox/mesh-sdk/go/views"
+
+	"git.novox.be/novox/mesh-delivery/sdk/firstview"
 
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
@@ -122,6 +125,26 @@ func settle(ctx context.Context, kv jetstream.KeyValue, each func(jetstream.KeyV
 	}
 }
 
+// readLate reads an owner's state the watchman follows by its first view (pic), on a goroutine of its own once that is
+// answered, so an owner that is down never holds the start: every key it holds now, then the start's end for its bucket
+// (watch.ReadLate), then each change until ctx ends.
+func readLate(ctx context.Context, w *watch.Watchman, pic *firstview.Picture, bucket string, each func(jetstream.KeyValueEntry)) error {
+	select {
+	case <-ctx.Done():
+		return nil
+	case <-pic.Follower().Ready():
+	}
+	sw, err := settle(ctx, pic, each)
+	if err != nil {
+		if ctx.Err() != nil {
+			return nil
+		}
+		return err
+	}
+	w.ReadLate(bucket)
+	return follow(ctx, sw, each)
+}
+
 // follow hands each change of a watcher to each until ctx ends; a watch that ends is an error, and the process
 // manager starts the program again.
 func follow(ctx context.Context, w jetstream.KeyWatcher, each func(jetstream.KeyValueEntry)) error {
@@ -147,6 +170,25 @@ func stepTaken(w *watch.Watchman, language string, e jetstream.KeyValueEntry) {
 	w.Step(language, e.Key(), e.Value(), opOf(e))
 }
 
+// createdOr is when an entry was written, or at when its picture does not say: the first view of an owner's state names
+// no key's time (novox/hq ADR 0615), so a want that stood before the start waits from the start.
+func createdOr(e jetstream.KeyValueEntry, at time.Time) time.Time {
+	if t := e.Created(); !t.IsZero() {
+		return t
+	}
+	return at
+}
+
+// The watchman's own states of its first views (views.StatusState); its findings of a first view unanswered past its
+// bound go to its findings bucket with the others (ADR 0615 §6). The owners' states it follows are the assigner's wants
+// and the indexer's modules (DLV-M5).
+var firstViewsBucket = delivery.WatchmanModule + "_" + views.StatusState
+
+const (
+	wantsState   = "wants"
+	modulesState = "modules"
+)
+
 // run is the program: it returns when ctx ends or the bus cannot be used.
 func run(ctx context.Context, getenv func(string) string, logOut io.Writer) error {
 	logger := log.New(logOut, "mesh-delivery-watchman: ", log.LstdFlags)
@@ -168,9 +210,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	}
 	defer conn.Close()
 	// Its own presence, pulses and stuck rows are what it keeps: each stops the start, naming its grant. The assigner's wants
-	// and the node-engine's haves only make the pulses of a move: without one the watchman watches presence and the build
-	// seats' steps, and says that moves are not watched (issue 1945).
-	var kvs struct{ presence, pulse, stuck, findings, wants, haves jetstream.KeyValue }
+	// and the node-engine's haves only make the pulses of a move: without the haves the watchman watches presence and the
+	// build seats' steps, and says that moves are not watched (issue 1945).
+	var kvs struct{ presence, pulse, stuck, findings, haves jetstream.KeyValue }
 	for _, b := range []struct {
 		name string
 		into *jetstream.KeyValue
@@ -179,14 +221,26 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 			return err
 		}
 	}
-	kvs.wants = process.OptionalBucket(ctx, js, delivery.AssignerWantsBucket, "a want makes no pulse, so a move that stops is not seen, until the watchman starts again", logger.Printf)
 	kvs.haves = process.OptionalBucket(ctx, js, delivery.HaveBucket, "a want makes no pulse, so a move that stops is not seen, until the watchman starts again", logger.Printf)
-	if kvs.wants == nil || kvs.haves == nil {
-		kvs.wants, kvs.haves = nil, nil // a want is a pulse only against the have that ends it
+	// The assigner's wants and the indexer's modules are their owners' states, read by the first view of their read tools and
+	// then their views families, never their buckets (ADR 0615 §1, DLV-M5). An owner that is down never holds the start: each
+	// is read on a goroutine once its first view is answered, and until then the unanswered first view is the watchman's
+	// finding.
+	calls, err := process.Connect(cred, nats.Name(cred.Node+"."+cred.Module+".calls"), nats.RetryOnFailedConnect(true))
+	if err != nil {
+		return fmt.Errorf("the watchman's connection for its tool calls: %w", err)
 	}
+	defer calls.Close()
+	reader := &firstview.Reader{Module: delivery.WatchmanModule, Call: calls.Call, Logf: logger.Printf,
+		Findings: func() jetstream.KeyValue { return kvs.findings }}
+	wants := reader.Follow(delivery.AssignerModule, wantsState)
+	modules := reader.Follow(delivery.IndexerModule, modulesState)
 
 	w := watch.New(&kvOut{ctx: ctx, presence: kvs.presence, pulse: kvs.pulse, stuk: kvs.stuck, findings: kvs.findings}, time.Now)
 	w.Logf(logger.Printf)
+	// What the wants and the modules hold is ended by their own late read (watch.ReadLate), never by the start's end.
+	w.Awaits(delivery.AssignerWantsBucket)
+	w.Awaits(delivery.ModulesBucket)
 	// A test shortens how long a move may take; the mesh never sets it.
 	if v := strings.TrimSpace(getenv("MESH_DELIVERY_PULSE_BOUND")); v != "" {
 		d, err := time.ParseDuration(v)
@@ -232,18 +286,13 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	if err != nil {
 		return err
 	}
-	// What the node agents run, then the assigner's wants (issue 1906, ADR 0539 §19): a want is a pulse owed by its node's agent
-	// until the have names its fingerprint, so the haves are read first and a want already applied makes no pulse.
-	var haveSeen, wantSeen jetstream.KeyWatcher
+	// What the node agents run (issue 1906, ADR 0539 §19): a want is a pulse owed by its node's agent until the have names its
+	// fingerprint, so the haves are read at the start and the wants after them, once their first view is answered.
+	var haveSeen jetstream.KeyWatcher
 	if kvs.haves != nil {
 		if haveSeen, err = settle(ctx, kvs.haves, func(e jetstream.KeyValueEntry) { w.Have(e.Key(), e.Value(), opOf(e)) }); err != nil {
 			process.NotWatched(logger.Printf, delivery.HaveBucket, err, "moves are not watched until the watchman starts again")
-		} else if wantSeen, err = settle(ctx, kvs.wants, func(e jetstream.KeyValueEntry) { w.Want(e.Key(), e.Value(), opOf(e), e.Created()) }); err != nil {
-			process.NotWatched(logger.Printf, delivery.AssignerWantsBucket, err, "moves are not watched until the watchman starts again")
-			haveSeen.Stop()
 			haveSeen = nil
-		} else {
-			w.ReadWhole(delivery.AssignerWantsBucket)
 		}
 	}
 	// The builder's builds (ADR 0603 §8): a key queued behind a package is a wait, mirrored from the time it was written.
@@ -358,17 +407,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 			go func() { _ = follow(ctx, hw, healthEach) }()
 		})
 	}
-	// The indexer's modules (ADR 0578 §1): a module whose value carries a refusal of its manifest is a finding; read
-	// before the findings are reconciled, each change heard on the watch.
-	verdictEach := func(e jetstream.KeyValueEntry) { w.ModuleVerdict(e.Key(), e.Value(), opOf(e)) }
-	var verdictWatch jetstream.KeyWatcher
-	if kv, err := js.KeyValue(ctx, delivery.ModulesBucket); err == nil {
-		if verdictWatch, err = settle(ctx, kv, verdictEach); err != nil {
-			process.NotWatched(logger.Printf, delivery.ModulesBucket, err, "no refused manifest is said until the watchman starts again")
-		}
-	} else {
-		process.NotWatched(logger.Printf, delivery.ModulesBucket, err, "no refused manifest is said until the watchman starts again")
-	}
 	dead := &deadKeeper{ctx: ctx, js: js, w: w, logf: logger.Printf}
 	w.SetRecount(dead.count)
 	dead.count()
@@ -384,8 +422,20 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	if healthWatch != nil {
 		go func() { errs <- follow(ctx, healthWatch, healthEach) }()
 	}
-	if verdictWatch != nil {
-		go func() { errs <- follow(ctx, verdictWatch, verdictEach) }()
+	// The first views of the wants and the modules, asked now and read each on its own goroutine once answered (ADR 0615).
+	reader.Start(ctx, js, cred, process.AwaitBucket(ctx, js, conn, firstViewsBucket, "watchman_first_views reads nothing and watchman_first_view_again is not heard", logger.Printf))
+	// The indexer's modules (ADR 0578 §1): a module whose value carries a refusal of its manifest is a finding.
+	verdictEach := func(e jetstream.KeyValueEntry) { w.ModuleVerdict(e.Key(), e.Value(), opOf(e)) }
+	go func() { errs <- readLate(ctx, w, modules, delivery.ModulesBucket, verdictEach) }()
+	if haveSeen != nil {
+		go func() {
+			errs <- follow(ctx, haveSeen, func(e jetstream.KeyValueEntry) { w.Have(e.Key(), e.Value(), opOf(e)) })
+		}()
+		go func() {
+			errs <- readLate(ctx, w, wants, delivery.AssignerWantsBucket, func(e jetstream.KeyValueEntry) {
+				w.Want(e.Key(), e.Value(), opOf(e), createdOr(e, time.Now()))
+			})
+		}()
 	}
 	if provWatch != nil {
 		go func() {
@@ -402,14 +452,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	go func() {
 		errs <- follow(ctx, pulses, func(e jetstream.KeyValueEntry) { w.Pulse(e.Key(), opOf(e), e.Created()) })
 	}()
-	if haveSeen != nil && wantSeen != nil {
-		go func() {
-			errs <- follow(ctx, haveSeen, func(e jetstream.KeyValueEntry) { w.Have(e.Key(), e.Value(), opOf(e)) })
-		}()
-		go func() {
-			errs <- follow(ctx, wantSeen, func(e jetstream.KeyValueEntry) { w.Want(e.Key(), e.Value(), opOf(e), e.Created()) })
-		}()
-	}
 	if buildSeen != nil {
 		go func() {
 			errs <- follow(ctx, buildSeen, func(e jetstream.KeyValueEntry) { w.Build(e.Key(), e.Value(), opOf(e), e.Created()) })

modules/mesh-delivery-watchman/go.mod

@@ -22,3 +22,7 @@ require (
 )
 
 replace git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun => ../mesh-delivery-builder/buildrun
+
+require git.novox.be/novox/mesh-delivery/sdk v0.0.0-00010101000000-000000000000
+
+replace git.novox.be/novox/mesh-delivery/sdk => ../../sdk

modules/mesh-delivery-watchman/internal/watch/watch.go

@@ -101,6 +101,10 @@ type Watchman struct {
 	standing   map[string]map[string]bool
 	readWhole  map[string]bool
 	reconciled bool
+	// awaited is each bucket read only once its owner's first view is answered, after the start's end (Awaits): its seeded
+	// pulses, stuck keys and findings wait for its own end (ReadLate), so a restart while its owner is down neither drops
+	// nor raises again what it holds.
+	awaited map[string]bool
 }
 
 // New is a watchman writing to out, reading the time from now.
@@ -111,7 +115,7 @@ func New(out Out, now func() time.Time) *Watchman {
 		haveState: map[string]delivery.Have{}, unhealthy: map[string]int{}, advised: map[string]int{}, advisedSince: map[string]time.Time{},
 		control: map[string]bool{}, failedSince: map[string]time.Time{}, provFailed: map[string]map[string]ProvReport{}, closures: map[string]map[string]time.Time{}, waiting: map[string]map[string]string{},
 		health: map[string]delivery.NodeHealth{}, tunnels: map[string]string{},
-		standing: map[string]map[string]bool{}, readWhole: map[string]bool{}}
+		standing: map[string]map[string]bool{}, readWhole: map[string]bool{}, awaited: map[string]bool{}}
 }
 
 // Logf sets where a write that failed is said; it is silent until set.
@@ -294,7 +298,7 @@ func (w *Watchman) Reconcile() {
 	w.mu.Lock()
 	defer w.mu.Unlock()
 	for pk, p := range w.pulses {
-		if p.seeded {
+		if p.seeded && !w.awaited[p.bucket] {
 			w.deletePulse(pk)
 		}
 	}
@@ -319,7 +323,7 @@ func (w *Watchman) ReadWhole(bucket string) {
 
 // stands records a mirrored key read standing during the start.
 func (w *Watchman) stands(bucket, key string) {
-	if w.reconciled {
+	if w.reconciled && !w.awaited[bucket] {
 		return
 	}
 	if w.standing[bucket] == nil {
@@ -330,15 +334,61 @@ func (w *Watchman) stands(bucket, key string) {
 
 func (w *Watchman) closeGone() {
 	for sk, s := range w.stuck {
-		if s.Moved || !w.readWhole[s.Bucket] || w.standing[s.Bucket][s.Key] {
+		if s.Moved || w.awaited[s.Bucket] || !w.readWhole[s.Bucket] || w.standing[s.Bucket][s.Key] {
 			continue
 		}
-		s.Moved = true
-		w.say("the stuck key %s is closed: its key %s is no longer in %s (it aged out of the bucket or was deleted while nobody heard), so nothing waits for it since %s",
-			sk, s.Key, s.Bucket, s.Since.UTC().Format(time.RFC3339))
-		w.putStuck(sk, s)
+		w.closeStuck(sk, s)
 	}
-	w.standing, w.readWhole, w.reconciled = map[string]map[string]bool{}, map[string]bool{}, true
+	standing := map[string]map[string]bool{}
+	for b := range w.awaited {
+		standing[b] = w.standing[b]
+	}
+	w.standing, w.readWhole, w.reconciled = standing, map[string]bool{}, true
+}
+
+func (w *Watchman) closeStuck(sk string, s delivery.Stuck) {
+	s.Moved = true
+	w.say("the stuck key %s is closed: its key %s is no longer in %s (it aged out of the bucket or was deleted while nobody heard), so nothing waits for it since %s",
+		sk, s.Key, s.Bucket, s.Since.UTC().Format(time.RFC3339))
+	w.putStuck(sk, s)
+}
+
+// Awaits says bucket (an owner's state the watchman reads through its first view: the assigner's wants, the indexer's
+// modules; novox/hq ADR 0615) is read only once that first view is answered, which may be after the start's end: Reconcile
+// leaves its seeded pulses, its stuck keys and the seeded findings it is the source of to ReadLate. Called before Reconcile.
+func (w *Watchman) Awaits(bucket string) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	w.awaited[bucket] = true
+}
+
+// ReadLate ends the start for an awaited bucket, called once every key it holds was read standing: as Reconcile does for
+// the others, a seeded pulse of it no key made again is deleted, a stuck key of it whose key is gone is closed, and a
+// seeded finding it is the source of and nothing observed again is deleted.
+func (w *Watchman) ReadLate(bucket string) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	if !w.awaited[bucket] {
+		return
+	}
+	for pk, p := range w.pulses {
+		if p.seeded && p.bucket == bucket {
+			w.deletePulse(pk)
+		}
+	}
+	for sk, s := range w.stuck {
+		if !s.Moved && s.Bucket == bucket && !w.standing[bucket][s.Key] {
+			w.closeStuck(sk, s)
+		}
+	}
+	for key := range w.seededFindings {
+		if f, err := DecodeFinding(w.findings[key]); err == nil && f.Source == bucket {
+			delete(w.seededFindings, key)
+			w.unfind(key)
+		}
+	}
+	delete(w.awaited, bucket)
+	delete(w.standing, bucket)
 }
 
 // reconcileFindings ends the start for the findings: one read at the start that what the watchman read since did not
@@ -350,6 +400,9 @@ func (w *Watchman) reconcileFindings() {
 		kept[due.Finding.Key()] = true
 	}
 	for key := range w.seededFindings {
+		if f, err := DecodeFinding(w.findings[key]); err == nil && w.awaited[f.Source] {
+			continue // ended by ReadLate
+		}
 		delete(w.seededFindings, key)
 		if kept[key] || signalKind(w.findings[key]) {
 			continue

modules/mesh-delivery-watchman/internal/watch/watch_test.go

@@ -668,12 +668,17 @@ func TestAStuckKeyWhoseKeyIsGoneIsClosedAtTheStart(t *testing.T) {
 		standing  func(w *Watchman) // what the start reads standing
 		readWhole bool
 		closed    bool
+		late      bool // read after the start's end (Awaits, ReadLate)
 	}{
-		{"an ask's step aged out of the steps bucket", checksKey, "askold1", func(*Watchman) {}, true, true},
-		{"an ask's step still running", checksKey, "askold1", func(w *Watchman) { w.Step("go", "n1.askold1", runningStep("askold1", t0), Put) }, true, false},
-		{"the steps bucket could not be read", checksKey, "askold1", func(*Watchman) {}, false, false},
-		{"a want deleted while nobody heard", delivery.AssignerWantsBucket, wantKey, func(*Watchman) {}, true, true},
-		{"a want still standing", delivery.AssignerWantsBucket, wantKey, func(w *Watchman) { w.Want(wantKey, wantValue(t, fpNew, false), Put, t0) }, true, false},
+		{"an ask's step aged out of the steps bucket", checksKey, "askold1", func(*Watchman) {}, true, true, false},
+		{"an ask's step still running", checksKey, "askold1", func(w *Watchman) { w.Step("go", "n1.askold1", runningStep("askold1", t0), Put) }, true, false, false},
+		{"the steps bucket could not be read", checksKey, "askold1", func(*Watchman) {}, false, false, false},
+		{"a want deleted while nobody heard", delivery.AssignerWantsBucket, wantKey, func(*Watchman) {}, true, true, false},
+		{"a want still standing", delivery.AssignerWantsBucket, wantKey, func(w *Watchman) { w.Want(wantKey, wantValue(t, fpNew, false), Put, t0) }, true, false, false},
+		// DLV-M5 (ADR 0615): the wants are read from the assigner's first view, which may come after the start's end; what
+		// they hold waits for it, and is then reconciled as at the start.
+		{"a want deleted while nobody heard, read from a late first view", delivery.AssignerWantsBucket, wantKey, func(*Watchman) {}, true, true, true},
+		{"a want still standing, read from a late first view", delivery.AssignerWantsBucket, wantKey, func(w *Watchman) { w.Want(wantKey, wantValue(t, fpNew, false), Put, t0) }, true, false, true},
 	} {
 		t.Run(row.name, func(t *testing.T) {
 			w, o, c := setup()
@@ -684,11 +689,36 @@ func TestAStuckKeyWhoseKeyIsGoneIsClosedAtTheStart(t *testing.T) {
 			if _, ok := o.findings[FindingKey(ScopeDelivery, sk, KindStuck)]; !ok {
 				t.Fatal("the seeded stuck key raised no finding")
 			}
-			row.standing(w)
-			if row.readWhole {
-				w.ReadWhole(row.bucket)
+			if row.late {
+				// A pulse of another key of the bucket, and a finding the bucket is the source of, stood at the start.
+				pk := pulseKeyOf(t, row.bucket, "want.n9.gone")
+				w.SeedPulse(pk)
+				fv, err := EncodeFinding(Finding{Scope: ScopeModule, ID: "gone", Kind: KindManifestRefused, Severity: Warning, Source: row.bucket, Summary: "refused", Since: t0})
+				if err != nil {
+					t.Fatal(err)
+				}
+				fk := FindingKey(ScopeModule, "gone", KindManifestRefused)
+				o.findings[fk], _ = DecodeFinding(fv)
+				w.SeedFinding(fk, fv)
+				w.Awaits(row.bucket)
+				w.Reconcile()
+				_, open := o.findings[FindingKey(ScopeDelivery, sk, KindStuck)]
+				_, seeded := o.findings[fk]
+				if o.stuck[sk].Moved || !open || !seeded || slices.Contains(o.pulseDeletes, pk) {
+					t.Fatalf("the start's end took what an awaited bucket holds: moved %v, stuck open %v, finding %v, pulses deleted %v", o.stuck[sk].Moved, open, seeded, o.pulseDeletes)
+				}
+				row.standing(w)
+				w.ReadLate(row.bucket)
+				if _, seeded := o.findings[fk]; seeded || !slices.Contains(o.pulseDeletes, pk) {
+					t.Errorf("the late read kept the finding (%v) or the pulse (deleted %v) nothing made again", seeded, o.pulseDeletes)
+				}
+			} else {
+				row.standing(w)
+				if row.readWhole {
+					w.ReadWhole(row.bucket)
+				}
+				w.Reconcile()
 			}
-			w.Reconcile()
 			_, open := o.findings[FindingKey(ScopeDelivery, sk, KindStuck)]
 			if got := o.stuck[sk].Moved; got != row.closed || open == row.closed {
 				t.Errorf("moved %v, finding open %v; want closed %v", got, open, row.closed)

modules/mesh-delivery-watchman/manifest_test.go

@@ -51,6 +51,8 @@ type manifest struct {
 	Version      string            `json:"version"`
 	Slug         string            `json:"slug"`
 	Reads        []string          `json:"reads"`
+	Invokes      []string          `json:"invokes"`
+	Consumes     []string          `json:"consumes"`
 	State        []state           `json:"state"`
 	OwnSecrets   map[string]string `json:"own-secrets"`
 	SecretsOwner string            `json:"secrets-owner"`
@@ -84,24 +86,37 @@ func TestManifestStateAndReads(t *testing.T) { // 1135/AC6
 	if m.Module != "mesh-delivery-watchman" || m.Moved || m.Slug != "watchman" {
 		t.Errorf("module %q moved %v (it has no template, so it is not moved) slug %q", m.Module, m.Moved, m.Slug)
 	}
-	want := []state{{Name: "presence", History: 20}, {Name: "pulse", KeyTTL: true}, {Name: "stuck", History: 20}, {Name: "findings", History: 5}}
+	want := []state{{Name: "presence", History: 20}, {Name: "pulse", KeyTTL: true}, {Name: "stuck", History: 20}, {Name: "findings", History: 5}, {Name: "first-views", History: 1}}
 	if !reflect.DeepEqual(m.State, want) {
 		t.Errorf("state = %+v, want %+v", m.State, want)
 	}
 	reads := slices.Clone(m.Reads)
 	slices.Sort(reads)
 	if !reflect.DeepEqual(reads, []string{"mesh-build-go.steps", "mesh-build-image.steps", "mesh-build-node.steps",
-		"mesh-delivery-assigner.wants", "mesh-delivery-builder.builds", "mesh-delivery-indexer.modules", "mesh-host.health", "mesh-provisioner.provisions", "node-engine.have"}) {
+		"mesh-delivery-builder.builds", "mesh-host.health", "mesh-provisioner.provisions", "node-engine.have"}) {
 		t.Errorf("reads = %v", reads)
 	}
+	// DLV-M5 (ADR 0615 §1): the assigner's wants and the indexer's modules are their owners' states, asked by their read
+	// tools and followed by their views families, never read.
+	for _, r := range m.Reads {
+		if strings.HasPrefix(r, "mesh-delivery-indexer.") || strings.HasPrefix(r, "mesh-delivery-assigner.") {
+			t.Errorf("reads %s: the indexer's and the assigner's state is read by their tools and views families alone", r)
+		}
+	}
+	if want := []string{"mesh-delivery-assigner.assigner_read_wants", "mesh-delivery-indexer.indexer_read_modules"}; !reflect.DeepEqual(m.Invokes, want) {
+		t.Errorf("invokes = %v, want %v", m.Invokes, want)
+	}
+	if want := []string{"mesh-delivery-assigner.views.wants", "mesh-delivery-indexer.views.modules"}; !reflect.DeepEqual(m.Consumes, want) {
+		t.Errorf("consumes = %v, want %v", m.Consumes, want)
+	}
 	// 1952/WM1: the watchman holds the seat mesh-watchdogs at mesh scope, its first seat (novox/hq ADR 0564 §1, §4).
 	if !reflect.DeepEqual(m.Claims, []claim{{Name: "mesh-watchdogs", Scope: "mesh", Serves: []string{"signals", "stuck", "presence", "dead-letters", "resync"}}}) {
 		t.Errorf("claims = %+v", m.Claims)
 	}
-	if !reflect.DeepEqual(m.Tools, []string{"watchman_findings"}) {
+	if !reflect.DeepEqual(m.Tools, []string{"watchman_findings", "watchman_first_views", "watchman_first_view_again"}) {
 		t.Errorf("tools = %v", m.Tools)
 	}
-	for _, k := range []string{"emits", "context", "items", "consumes", "provides"} {
+	for _, k := range []string{"emits", "context", "items", "provides"} {
 		if _, ok := top[k]; ok {
 			t.Errorf("the manifest declares %q: the watchman has none", k)
 		}

modules/mesh-delivery-watchman/module.json

@@ -19,21 +19,30 @@
     "mesh-build-go.steps",
     "mesh-build-image.steps",
     "mesh-build-node.steps",
-    "mesh-delivery-assigner.wants",
     "mesh-delivery-builder.builds",
-    "mesh-delivery-indexer.modules",
     "mesh-host.health",
     "mesh-provisioner.provisions",
     "node-engine.have"
   ],
+  "invokes": [
+    "mesh-delivery-assigner.assigner_read_wants",
+    "mesh-delivery-indexer.indexer_read_modules"
+  ],
+  "consumes": [
+    "mesh-delivery-assigner.views.wants",
+    "mesh-delivery-indexer.views.modules"
+  ],
   "tools": [
-    "watchman_findings"
+    "watchman_findings",
+    "watchman_first_views",
+    "watchman_first_view_again"
   ],
   "state": [
     { "name": "presence", "history": 20 },
     { "name": "pulse", "key-ttl": true },
     { "name": "stuck", "history": 20 },
-    { "name": "findings", "history": 5 }
+    { "name": "findings", "history": 5 },
+    { "name": "first-views", "history": 1 }
   ],
   "own-secrets": {
     "broker": "${dir:state}/broker"