Take-in events are the indexer's ordinary events: no stream of its own, which the manifest validator refuses (issue 1585, ADR 0633 §5)

The group's check failed at 7b7096f3: the group's own manifest validator refuses the indexer's stream "take-in", outside
the bus's predefined forms. The lead's ruling: the indexer says take-in.pending and take-in.taken as its ordinary events
(its emits, mesh.mod.<module>.event.take-in.<kind>), as manifest-accepted is said, and keeps no stream. The wait's key is
in the body only: an emitted event names one event (no wildcard), and a key such as g.<group> is no event name. The
watchman consumes the two events by name on the shared events stream; it needs no replay, since it seeds its waits from
indexer_pending at its start. indexer_pending says the last event of each wait as take-in.<kind>.

Run locally, as the check runs it: cmd/mesh-judge over every manifest of mesh-catalog, mesh-delivery, mesh-host,
mesh-tools, mesh-networking and mesh-media-catalog, with this branch's and mesh-catalog's feat/take-in manifests in the
change case and origin/main's in the trunk case: 171 of 171 accepted in both. The same run with the indexer's manifest at
7b7096f3 refuses it with the check's words.

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

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

@@ -74,8 +74,8 @@ func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10; 1585/T14, T15
 				t.Fatal(err)
 			}
 		}
-		for name, subject := range map[string]string{"TAKE_IN": "mesh.stream.mesh-delivery-indexer.event.take-in.>",
-			"INDEXER_EVENTS": "mesh.mod.mesh-delivery-indexer.event.>"} {
+		// The take-in events are the indexer's ordinary events (ADR 0633 §5), on the path manifest-accepted takes.
+		for name, subject := range map[string]string{"INDEXER_EVENTS": "mesh.mod.mesh-delivery-indexer.event.>"} {
 			if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: name, Subjects: []string{subject}}); err != nil {
 				t.Fatal(err)
 			}
@@ -164,18 +164,34 @@ func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10; 1585/T14, T15
 		return ack.Sequence
 	}
 	m1Key := takein.PendingKey(merged(m1))
-	takeIns, err := js.Stream(ctx, "TAKE_IN")
+	events, err := js.Stream(ctx, "INDEXER_EVENTS")
 	if err != nil {
 		t.Fatal(err)
 	}
+	// saidAll is every take-in event of kind said under key so far, oldest first.
+	saidAll := func(kind, key string) []takein.TakeIn {
+		event, _ := takein.TakeInEvent(kind)
+		info, err := events.Info(ctx)
+		if err != nil {
+			t.Fatal(err)
+		}
+		var out []takein.TakeIn
+		for seq := info.State.FirstSeq; seq <= info.State.LastSeq && seq > 0; seq++ {
+			msg, err := events.GetMsg(ctx, seq)
+			if err != nil || msg.Subject != "mesh.mod."+ix+".event."+event {
+				continue
+			}
+			if ti, err := takein.DecodeTakeIn(msg.Data); err == nil && ti.Key == key {
+				out = append(out, ti)
+			}
+		}
+		return out
+	}
 	// said is the last take-in event under kind and key, within 20 s.
 	said := func(kind, key string) (takein.TakeIn, bool) {
-		event, _ := takein.TakeInEvent(kind, key)
 		for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
-			if msg, err := takeIns.GetLastMsgForSubject(ctx, process.ModuleStreamSubject(ix, event)); err == nil {
-				if ti, err := takein.DecodeTakeIn(msg.Data); err == nil {
-					return ti, true
-				}
+			if all := saidAll(kind, key); len(all) > 0 {
+				return all[len(all)-1], true
 			}
 		}
 		return takein.TakeIn{}, false
@@ -258,7 +274,7 @@ func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10; 1585/T14, T15
 		}
 	}
 	if len(w) != 2 || w[0].Key != m1Key || w[0].WaitsFor != named || !slices.Contains(w[0].Holds, "a") || !slices.Contains(w[0].Holds, ix) ||
-		w[1].Key != groupKey || w[1].What != takein.TakeInGroup || len(w[1].Merges) != 1 || !strings.HasSuffix(w[1].Said, "pending."+groupKey) {
+		w[1].Key != groupKey || w[1].What != takein.TakeInGroup || len(w[1].Merges) != 1 || w[1].Said != takein.TakeInPendingEvent {
 		t.Fatalf("indexer_pending = %+v, want the revision %s waiting for %s holding a and the indexer, then the group gathered", w, m1Key, named)
 	}
 
@@ -309,8 +325,7 @@ func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10; 1585/T14, T15
 		t.Errorf("after the group's end: modules.c %v, modules.k %v, indexer_pending %+v: want both taken in and nothing waiting", hasC, hasK, w)
 	}
 	// One take-in: neither member was said as a revision of its own.
-	single, _ := takein.TakeInEvent(takein.TakeInPending, g1Key)
-	if _, err := takeIns.GetLastMsgForSubject(ctx, process.ModuleStreamSubject(ix, single)); err == nil {
+	if len(saidAll(takein.TakeInPending, g1Key)) != 0 {
 		t.Errorf("the group's first member %s was said pending as a revision of its own", g1Key)
 	}
 }

modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/verdicts.go

@@ -64,11 +64,11 @@ func sayAccepted(js jetstream.JetStream, cred process.Account, logf func(string,
 	}
 }
 
-// sayTakeIn publishes a take-in event on the indexer's own stream take-in (novox/hq ADR 0633 §5):
-// `mesh.stream.<module>.event.take-in.<kind>.<key>`. A publish that fails is the caller's to say.
+// sayTakeIn says a take-in event (novox/hq ADR 0633 §5) as the indexer's ordinary event, on the path its other events
+// take: mesh.mod.<module>.event.take-in.<kind>, the wait's key in the body. A publish that fails is the caller's to say.
 func sayTakeIn(js jetstream.JetStream, cred process.Account) func(context.Context, takein.TakeIn) error {
 	return func(ctx context.Context, t takein.TakeIn) error {
-		event, err := takein.TakeInEvent(t.Kind, t.Key)
+		event, err := takein.TakeInEvent(t.Kind)
 		if err != nil {
 			return err
 		}
@@ -76,7 +76,7 @@ func sayTakeIn(js jetstream.JetStream, cred process.Account) func(context.Contex
 		if err != nil {
 			return err
 		}
-		msg := nats.NewMsg(process.ModuleStreamSubject(cred.Module, event))
+		msg := nats.NewMsg(cred.EventSubject(event))
 		msg.Header.Set("x-source", cred.Module)
 		msg.Header.Set("x-node", cred.Node)
 		msg.Header.Set("x-time", time.Now().UTC().Format(time.RFC3339Nano))

modules/mesh-delivery-indexer/internal/index/pending.go

@@ -815,7 +815,7 @@ func waitOf(entries []PendingEntry, what, behind string) takein.PendingWait {
 			}
 		}
 		if e.Value.Said != "" && e.Value.SaidAt.After(pw.SaidAt) {
-			pw.Said, pw.SaidAt = takein.TakeInStream+"."+e.Value.SaidKind+"."+e.Value.Said, e.Value.SaidAt
+			pw.Said, pw.SaidAt = "take-in."+e.Value.SaidKind, e.Value.SaidAt
 		}
 	}
 	slices.Sort(pw.Holds)

modules/mesh-delivery-indexer/internal/index/pending_test.go

@@ -385,7 +385,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
 			}
 		}
 		if len(answer.Waits) != 1 || answer.Waits[0].Key != takein.PendingKey(2) || !slices.Equal(answer.Waits[0].Holds, []string{"avahi", delivery.IndexerModule}) ||
-			answer.Waits[0].Said != "take-in.pending."+takein.PendingKey(2) || len(answer.Waits[0].Keys) != 2 {
+			answer.Waits[0].Said != takein.TakeInPendingEvent || len(answer.Waits[0].Keys) != 2 {
 			t.Fatalf("indexer_pending = %+v", answer)
 		}
 		b.said = nil

modules/mesh-delivery-indexer/manifest_test.go

@@ -104,8 +104,8 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
 	// 1126: placement is the seat mesh-placement's (ADR 0567): the indexer holds no placement state. ADR 0578 §1: it says
 	// a refused manifest as manifest-refused, and (issue 1966 D3, 1966/AC10) a newly accepted one asking for bus state as
 	// manifest-accepted.
-	if !reflect.DeepEqual(m.Emits, []string{"manifest-refused", "manifest-accepted"}) {
-		t.Errorf("emits = %v, want manifest-refused and manifest-accepted", m.Emits)
+	if !reflect.DeepEqual(m.Emits, []string{"manifest-refused", "manifest-accepted", takein.TakeInPendingEvent, takein.TakeInTakenEvent}) {
+		t.Errorf("emits = %v, want manifest-refused, manifest-accepted and the two take-in events", m.Emits)
 	}
 	// 1126 note 76: heard keeps the last 64 FORGE messages the indexer handled, for indexer_heard.
 	if want := []state{{Name: "candidates", History: 10}, {Name: "heard", History: 64}, {Name: "manifests", History: 1}, {Name: "modules", History: 20}, {Name: "predefined", History: 5}, {Name: "config-candidates", History: 5}, {Name: "taken", History: 1}, {Name: "takeins", History: 1}, {Name: "verdicts", History: 5}, {Name: "oldsections", History: 1}, {Name: delivery.PackagesState, History: delivery.PackagesHistory},
@@ -146,10 +146,10 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
 		!slices.Equal(m.Claims[0].Serves, []string{"modules", "module", "provides", "dependents"}) {
 		t.Errorf("claims = %+v, want mesh-catalog with its four verbs", m.Claims)
 	}
-	// 1585/T11 (ADR 0633 §1, §5): its one stream, take-in, an old streams entry (the indexer is foundation, ADR 0595 §11),
-	// carries the take-in events.
-	if want := []streamDecl{{Name: takein.TakeInStream, Events: []string{"take-in.**"}, MaxAgeDays: 7}}; !reflect.DeepEqual(m.Streams, want) {
-		t.Errorf("streams = %+v, want %+v", m.Streams, want)
+	// 1585/T11 (ADR 0633 §1, §5): the take-in events are its ordinary events, with no stream of its own: a stream is
+	// outside the bus's predefined forms, and the manifest validator refuses it.
+	if len(m.Streams) != 0 || !slices.Contains(m.Emits, takein.TakeInPendingEvent) || !slices.Contains(m.Emits, takein.TakeInTakenEvent) {
+		t.Errorf("streams = %+v, emits = %v: want no stream, and both take-in events emitted", m.Streams, m.Emits)
 	}
 	// No other bus section, no other seat (ADR 0437 §4).
 	for _, k := range []string{"logs", "provides", "requires", "uses", "defines-seats", "seat", "seats", "bus"} {

modules/mesh-delivery-indexer/module.json

@@ -44,12 +44,11 @@
     { "name": "graph", "history": 5 },
     { "name": "pending", "history": 1 }
   ],
-  "streams": [
-    { "name": "take-in", "events": ["take-in.**"], "max-age-days": 7 }
-  ],
   "emits": [
     "manifest-refused",
-    "manifest-accepted"
+    "manifest-accepted",
+    "take-in.pending",
+    "take-in.taken"
   ],
   "reads": [
     "mesh-build-go.toolchains",

modules/mesh-delivery-indexer/takein/takein.go

@@ -49,14 +49,17 @@ func decode(data []byte, into any) error {
 	return nil
 }
 
-// The take-in events (novox/hq ADR 0633 §1, §5): the indexer publishes them on its own stream, which keeps the last
-// message per subject, `take-in.<kind>.<key>`. A wait starts with `take-in.pending` and ends with `take-in.taken`,
-// whoever ended it (the named indexer, a revert, a hand take-in or a drop). The stream's name is the indexer's manifest's;
-// these names are its events, relative to it.
+// The take-in events (novox/hq ADR 0633 §1, §5): the indexer's ordinary events, `take-in.pending` and `take-in.taken`, on
+// the path its other events take (the module's event subject; its manifest's emits), with no stream of its own. A wait
+// starts with `take-in.pending` and ends with `take-in.taken`, whoever ended it (the named indexer, a revert, a hand
+// take-in or a drop); the wait's key is in the body (TakeIn.Key), since an emitted event names one event and a key is
+// no event name. A reader that starts later asks the waits of indexer_pending, not a replay.
 const (
-	TakeInStream  = "take-in"
 	TakeInPending = "pending"
 	TakeInTaken   = "taken"
+	// TakeInPendingEvent and TakeInTakenEvent are the two events' names, relative to the indexer.
+	TakeInPendingEvent = "take-in." + TakeInPending
+	TakeInTakenEvent   = "take-in." + TakeInTaken
 )
 
 // What waits in a take-in event: a source revision for the indexer build it names, or a delivery group for its end.
@@ -80,27 +83,24 @@ func GroupTakeInKey(group string) string {
 	}, group)
 }
 
-// TakeInEvent is the name of a take-in event of the given kind and key: `take-in.<pending|taken>.<key>`.
-func TakeInEvent(kind, key string) (string, error) {
+// TakeInEvent is the name of the take-in event of the given kind: `take-in.<pending|taken>`.
+func TakeInEvent(kind string) (string, error) {
 	if kind != TakeInPending && kind != TakeInTaken {
 		return "", fmt.Errorf("delivery: the take-in kind %q is not pending or taken", kind)
 	}
-	if err := checkTakeInKey(key); err != nil {
-		return "", err
-	}
-	return TakeInStream + "." + kind + "." + key, nil
+	return "take-in." + kind, nil
 }
 
-// ParseTakeInEvent is the inverse of TakeInEvent.
-func ParseTakeInEvent(event string) (kind, key string, err error) {
-	parts := strings.SplitN(event, ".", 3)
-	if len(parts) != 3 || parts[0] != TakeInStream {
-		return "", "", fmt.Errorf("delivery: %q is not a take-in event, take-in.<pending|taken>.<key>", event)
+// ParseTakeInEvent is the inverse of TakeInEvent: the kind of a take-in event's name.
+func ParseTakeInEvent(event string) (kind string, err error) {
+	k, ok := strings.CutPrefix(event, "take-in.")
+	if !ok {
+		return "", fmt.Errorf("delivery: %q is not a take-in event, take-in.<pending|taken>", event)
 	}
-	if _, err := TakeInEvent(parts[1], parts[2]); err != nil {
-		return "", "", fmt.Errorf("delivery: %q is not a take-in event: %w", event, err)
+	if _, err := TakeInEvent(k); err != nil {
+		return "", fmt.Errorf("delivery: %q is not a take-in event: %w", event, err)
 	}
-	return parts[1], parts[2], nil
+	return k, nil
 }
 
 func checkTakeInKey(key string) error {
@@ -136,7 +136,10 @@ type TakeIn struct {
 }
 
 func (t TakeIn) check() error {
-	if _, err := TakeInEvent(t.Kind, t.Key); err != nil {
+	if _, err := TakeInEvent(t.Kind); err != nil {
+		return err
+	}
+	if err := checkTakeInKey(t.Key); err != nil {
 		return err
 	}
 	if len(t.Merges) == 0 {
@@ -213,7 +216,7 @@ func SameTakeIn(a, b TakeIn) bool { return reflect.DeepEqual(a, b) }
 // merge's key, waiting for the indexer build WaitsFor (empty while it is worked out), or a group being gathered under
 // its group key, with Members every approved member and Merges those held so far. A revision that waits behind a
 // group being gathered names that group's key in Behind. Keys are the pending keys of its merges, in Merges' order (indexer_drop_pending drops one). Holds are the modules it holds back: every module of its
-// merges' trees. Said is the last take-in event said of it (`take-in.<kind>.<key>`) and SaidAt when.
+// merges' trees. Said is the last take-in event said of it (`take-in.<kind>`, of its Key) and SaidAt when.
 type PendingWait struct {
 	Key      string          `json:"key"`
 	What     string          `json:"what"`

modules/mesh-delivery-indexer/takein/takein_test.go

@@ -25,21 +25,21 @@ func sampleTakeIns(since time.Time) []TakeIn {
 
 // TestTakeInEvents holds the indexer's take-in contract, moved from mesh-sdk go/delivery with its tests (1585/T2, T12).
 func TestTakeInEvents(t *testing.T) {
-	// The take-in events, `take-in.<pending|taken>.<key>`, a group's key with its dots and no subject wildcard.
+	// The take-in events, `take-in.<pending|taken>`; a group's key with its dots and no subject wildcard, in the body.
 	if PendingKey(42) != "m.00000000000000000042" || GroupTakeInKey("feat/a.b *>") != "g.feat/a.b___" {
 		t.Errorf("take-in keys %q %q", PendingKey(42), GroupTakeInKey("feat/a.b *>"))
 	}
-	for _, c := range []struct{ kind, key string }{{TakeInPending, PendingKey(7)}, {TakeInTaken, GroupTakeInKey("feat/a.b")}} {
-		ev, err := TakeInEvent(c.kind, c.key)
-		if err != nil || ev != "take-in."+c.kind+"."+c.key {
+	for kind, want := range map[string]string{TakeInPending: TakeInPendingEvent, TakeInTaken: TakeInTakenEvent} {
+		ev, err := TakeInEvent(kind)
+		if err != nil || ev != want || ev != "take-in."+kind {
 			t.Fatalf("take-in event = %q, %v", ev, err)
 		}
-		if k, key, err := ParseTakeInEvent(ev); err != nil || k != c.kind || key != c.key {
-			t.Errorf("parse take-in event %q = %q %q %v", ev, k, key, err)
+		if k, err := ParseTakeInEvent(ev); err != nil || k != kind {
+			t.Errorf("parse take-in event %q = %q %v", ev, k, err)
 		}
 	}
-	for _, bad := range []string{"", "take-in.pending", "take-in.held.m.1", "take-in.taken.x.1", "take-in.taken.m..1", "forge.taken.m.1"} {
-		if _, _, err := ParseTakeInEvent(bad); err == nil {
+	for _, bad := range []string{"", "take-in", "take-in.held", "take-in.pending.m.1", "forge.taken"} {
+		if _, err := ParseTakeInEvent(bad); err == nil {
 			t.Errorf("parse take-in event %q accepted", bad)
 		}
 	}
@@ -74,6 +74,7 @@ func TestTakeInEvents(t *testing.T) {
 		{"take-in without merges", "no merge", 0, func(t *TakeIn) { t.Merges = nil }},
 		{"take-in without since", "since", 0, func(t *TakeIn) { t.Since = time.Time{} }},
 		{"a merge of no repository", "owner", 0, func(t *TakeIn) { t.Merges = []Merge{{Repository: "nope", Merge: "m"}} }},
+		{"a key that is neither a merge's nor a group's", "neither", 0, func(t *TakeIn) { t.Key = "x.1" }},
 	} {
 		ti := sampleTakeIns(since)[c.i]
 		c.mut(&ti)

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

@@ -24,7 +24,6 @@ import (
 	"github.com/nats-io/nats.go/jetstream"
 
 	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein"
-	"git.novox.be/novox/mesh-sdk/go/catalogue"
 	"git.novox.be/novox/mesh-sdk/go/delivery"
 	"git.novox.be/novox/mesh-sdk/go/provisions"
 
@@ -378,7 +377,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	dead.ages()
 	dead.check()
 	provisioned(ctx, js, w, logger.Printf)
-	// The indexer's stream take-in (ADR 0633 §5): a revision or a group it holds back is a wait, ended by take-in.taken. The
+	// The indexer's take-in events (ADR 0633 §5): a revision or a group it holds back is a wait, ended by take-in.taken. The
 	// durable consumer <node>_<module> is made by the controller from this module's consumes, and it is durable: what it
 	// acknowledged before a restart is not heard again. So the waits the indexer holds now are asked of it first, through
 	// its tool indexer_pending (this module's invokes), each with its own since; the consumer then hears what follows.
@@ -388,10 +387,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		seedTakeIn(w, calls.Call, logger.Printf)
 		calls.Close()
 	}
-	switch tc, err := cred.EventConsumer(ctx, js, catalogue.StreamBusName(takein.TakeInStream)); {
+	// The take-in events are the indexer's ordinary events (its emits), heard on the shared events stream through this
+	// module's durable consumer, which its consumes filter to them.
+	switch tc, err := cred.EventConsumer(ctx, js, eventsStream); {
 	case err != nil:
 		logger.Printf("the consumer %s_%s on %s does not exist yet, so no take-in wait is said until the watchman starts again: %v",
-			cred.Node, cred.Module, catalogue.StreamBusName(takein.TakeInStream), err)
+			cred.Node, cred.Module, eventsStream, err)
 	default:
 		// The replay has caught up when a message says none is pending after it, or when the consumer had none pending at its
 		// bind (the stream holds nothing it has not acknowledged): the waits past their bound are said then, not before.
@@ -473,6 +474,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	}
 }
 
+// eventsStream is the shared stream of the modules' ordinary events, as the bus names it.
+const eventsStream = "EVENTS"
+
 // pendingTool is the indexer's tool that answers the waits it holds (ADR 0633 §5), as the view asks it.
 const pendingTool = delivery.IndexerModule + ".indexer_pending"
 
@@ -491,7 +495,7 @@ func seedTakeIn(w *watch.Watchman, call func(key string, body any) (json.RawMess
 	logf("take-in: %s cannot be asked (%v), so a wait the indexer held before this start is said only once a take-in event of it is heard", pendingTool, err)
 }
 
-// takeInHeard hands one event of a take-in stream to the watchman, by its name relative to the stream, and acknowledges it;
+// takeInHeard hands one take-in event of the indexer to the watchman, by its name relative to the indexer, and acknowledges it;
 // one that is not a take-in event is said and acknowledged, since hearing it again would not read it.
 func takeInHeard(w *watch.Watchman, m jetstream.Msg, logf func(string, ...any)) {
 	_, event, ok := strings.Cut(m.Subject(), ".event.")

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

@@ -622,7 +622,7 @@ const (
 
 func takeInEvent(t *testing.T, kind, key string) string {
 	t.Helper()
-	e, err := takein.TakeInEvent(kind, key)
+	e, err := takein.TakeInEvent(kind)
 	if err != nil {
 		t.Fatal(err)
 	}

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

@@ -24,11 +24,11 @@ const (
 // takeInRow is the signals table's row of a take-in wait.
 const takeInRow = "take-in"
 
-// TakeIn hears one event of the indexer's stream `take-in`, by its name relative to the stream (`take-in.<kind>.<key>`):
-// a pending one starts the wait for its key, whose bound runs from the event's since, and a taken one ends it. A group's
+// TakeIn hears one take-in event of the indexer, by its name relative to the indexer (`take-in.<kind>`), the wait's key
+// in its body: a pending one starts the wait for its key, whose bound runs from the event's since, and a taken one ends it. A group's
 // taken event ends the group's wait only: the revision it becomes is its own key and starts its own, longer wait.
 func (w *Watchman) TakeIn(event string, body []byte) {
-	kind, key, err := takein.ParseTakeInEvent(event)
+	kind, err := takein.ParseTakeInEvent(event)
 	if err != nil {
 		w.say("a take-in event is left: %v", err)
 		return
@@ -38,6 +38,11 @@ func (w *Watchman) TakeIn(event string, body []byte) {
 		w.say("the body of %s cannot be read, left: %v", event, err)
 		return
 	}
+	if t.Kind != kind {
+		w.say("the body of %s says the kind %q, left", event, t.Kind)
+		return
+	}
+	key := t.Key
 	w.mu.Lock()
 	defer w.mu.Unlock()
 	w.takeIn(kind, key, t)

modules/mesh-delivery-watchman/manifest_test.go

@@ -97,7 +97,7 @@ func TestManifestStateAndReads(t *testing.T) { // 1135/AC6
 		t.Errorf("reads = %v", reads)
 	}
 	// 1585/T20: it hears the indexer's stream take-in, whole, through a durable consumer the controller makes from consumes.
-	if !reflect.DeepEqual(m.Consumes, []string{"mesh-delivery-indexer.take-in.**"}) {
+	if !reflect.DeepEqual(m.Consumes, []string{"mesh-delivery-indexer.take-in.pending", "mesh-delivery-indexer.take-in.taken"}) {
 		t.Errorf("consumes = %v", m.Consumes)
 	}
 	// The waits the indexer holds at the watchman's start are asked of its tool (ADR 0633 §5), a verb edge as the view's.

modules/mesh-delivery-watchman/module.json

@@ -27,7 +27,8 @@
     "node-engine.have"
   ],
   "consumes": [
-    "mesh-delivery-indexer.take-in.**"
+    "mesh-delivery-indexer.take-in.pending",
+    "mesh-delivery-indexer.take-in.taken"
   ],
   "invokes": [
     "mesh-delivery-indexer.indexer_pending"