assigner: the test drives the reads-changed handler (issue 1975 note 6)

Builds: issue 1975 note 6 (the lead's review round 1; no design line)

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

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

@@ -552,13 +552,10 @@ func sayVia(ctx context.Context, js jetstream.JetStream) func(oc.Condition) erro
 	}
 }
 
-// hearReadsChanged hears the node-engine seat's event manifest-reads-changed on the assigner's one hub of its durable
-// consumer on EVENTS, and asks the node it names again; the hub also hands the indexer's views to the followers of
-// indexerViews, and starts once both are registered.
-func hearReadsChanged(ctx context.Context, js jetstream.JetStream, hub *process.EventHub, reads *enginereads.Keeper, findings assign.Findings, live func(), logf func(string, ...any)) {
-	account := process.Account{Node: thisNode, Module: assignment.Module}
-	w := notHeardWriter(findings, logf)
-	handle := func(m jetstream.Msg) {
+// readsChangedHandler is what the assigner does with each event manifest-reads-changed: acknowledges it, keeps what it says
+// of the node (a statement older than it does not stand) and asks the node again.
+func readsChangedHandler(reads *enginereads.Keeper, logf func(string, ...any)) func(jetstream.Msg) {
+	return func(m jetstream.Msg) {
 		if err := m.Ack(); err != nil {
 			logf("acknowledging %s: %v", m.Subject(), err)
 		}
@@ -576,6 +573,15 @@ func hearReadsChanged(ctx context.Context, js jetstream.JetStream, hub *process.
 			reads.AskNode(node, enginereads.Event)
 		}
 	}
+}
+
+// hearReadsChanged hears the node-engine seat's event manifest-reads-changed on the assigner's one hub of its durable
+// consumer on EVENTS, and asks the node it names again; the hub also hands the indexer's views to the followers of
+// indexerViews, and starts once both are registered.
+func hearReadsChanged(ctx context.Context, js jetstream.JetStream, hub *process.EventHub, reads *enginereads.Keeper, findings assign.Findings, live func(), logf func(string, ...any)) {
+	account := process.Account{Node: thisNode, Module: assignment.Module}
+	w := notHeardWriter(findings, logf)
+	handle := readsChangedHandler(reads, logf)
 	err := hub.Handle(enginereads.SubjectPrefix+"*", handle)
 	// At each bind every node the assigner holds a statement of is asked again: an event said before its consumer existed
 	// is not delivered to it (issue 1966 note 155: ace's bootstrapped statement stood; ADR 0668 §4).

modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go

@@ -319,17 +319,15 @@ func TestAnAnswerCountsOnlyForTheAskingItAnswered(t *testing.T) {
 		}
 		// Issue 1975: ace's engine answered its bootstrapped seats, then stated the full list and said so; an answer older
 		// than that event does not stand: the rules read what the event says, and the node is asked until it is as new.
-		stale := true
+		// The engine's verb keeps answering the bootstrapped statement (its stated-at is older than the event); the event is
+		// handed to the assigner's own handler (readsChangedHandler, what hearReadsChanged registers) as the bus delivers it.
 		bootstrapped := func(node, _ string, _ any) (json.RawMessage, error) {
-			if stale {
-				stale = false
-				return json.Marshal(nodeengine.ManifestReadsAnswer{V: 1, Node: node, ManifestReads: []string{"seat:a"}, StatedAt: "2026-11-12T01:14:36.498Z"})
-			}
-			return json.Marshal(nodeengine.ManifestReadsAnswer{V: 1, Node: node, ManifestReads: []string{"seat:a", "seat:b"}, StatedAt: "2026-11-12T01:14:39.001Z"})
+			return json.Marshal(nodeengine.ManifestReadsAnswer{V: 1, Node: node, ManifestReads: []string{"seat:a"}, StatedAt: "2026-11-12T01:14:36.498Z"})
 		}
 		rk := &enginereads.Keeper{Call: bootstrapped, Sync: true}
 		rk.AskNode("ace", "the assigner started")
-		rk.HeardChanged(nodeengine.ManifestReadsChangedEvent{V: 1, Node: "ace", Why: enginereads.ReadsStarted, At: "2026-11-12T01:14:39.002Z", ManifestReads: []string{"seat:a", "seat:b"}})
+		body, _ := json.Marshal(nodeengine.ManifestReadsChangedEvent{V: 1, Node: "ace", Why: enginereads.ReadsStarted, At: "2026-11-12T01:14:39.002Z", ManifestReads: []string{"seat:a", "seat:b"}})
+		readsChangedHandler(rk, t.Logf)(readsMsg{subject: enginereads.SubjectPrefix + "ace", data: body})
 		rp := &passer{w: w, reads: rk, online: map[string]bool{"ace": true}}
 		rn := w.nodes(nil, nil)
 		rp.engineReads(rn)
@@ -1510,3 +1508,14 @@ func readAskID(t *testing.T, p *passer, key string) string {
 	}
 	return d.Ask.ID
 }
+
+// readsMsg is a message of the events stream as the assigner's handler reads it.
+type readsMsg struct {
+	jetstream.Msg
+	subject string
+	data    []byte
+}
+
+func (m readsMsg) Ack() error      { return nil }
+func (m readsMsg) Subject() string { return m.subject }
+func (m readsMsg) Data() []byte    { return m.data }