Watchman: its one pull of its durable on the shared events stream is the module's event hub: the take-in events, the seats' events and the views reader are three handlers of it (B7c finding 1; issue 1966)

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

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

@@ -59,7 +59,15 @@ func consumed(w *watch.Watchman, subject string, body []byte, logf func(string,
 
 // hearConsumed opens the watchman's durable consumers: on EVENTS now, and on each module stream its membership names as
 // that membership arrives. A consumer the bus does not hold yet is said and tried again at the membership's next change.
-func hearConsumed(ctx context.Context, js jetstream.JetStream, cred process.Account, w *watch.Watchman, logf func(string, ...any)) {
+func hearConsumed(ctx context.Context, js jetstream.JetStream, cred process.Account, hub *process.EventHub, w *watch.Watchman, logf func(string, ...any)) {
+	// The seats' events (manifest-accepted, the bus module's created and refused) are one handler of the module's one
+	// pull of EVENTS, beside the take-in events and the views reader.
+	if err := hub.Handle("mesh.seat.*.event.>", func(m jetstream.Msg) {
+		consumed(w, m.Subject(), m.Data(), logf)
+		_ = m.Ack()
+	}); err != nil {
+		w.NotHeard("EVENTS", "the seats' events cannot be handled", err)
+	}
 	var mu sync.Mutex
 	opened := map[string]bool{}
 	open := func(stream, name string) {
@@ -70,6 +78,15 @@ func hearConsumed(ctx context.Context, js jetstream.JetStream, cred process.Acco
 		}
 		// Not bound, the stream is the finding family-not-heard (ADR 0615 §5) until it is, at a later change of the
 		// membership or a later start.
+		if stream == "EVENTS" {
+			if err := hub.Start(ctx, js, cred); err != nil {
+				w.NotHeard(stream, "what it carries is heard once the grants part has made the consumer "+name, err)
+				return
+			}
+			w.Heard(stream)
+			opened[stream] = true
+			return
+		}
 		cons, err := js.Consumer(ctx, stream, name)
 		if err != nil {
 			w.NotHeard(stream, "what it carries is heard once the grants part has made the consumer "+name, err)

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

@@ -234,7 +234,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		return fmt.Errorf("the watchman's connection for its tool calls: %w", err)
 	}
 	defer calls.Close()
-	reader := &views.ProcessReader{Module: delivery.WatchmanModule, Call: calls.Call, Logf: logger.Printf,
+	// The module has one durable consumer of EVENTS: one hub pulls it for the views reader, the indexer's take-in events
+	// and the seats' events, started once every part is registered.
+	hub := process.NewEventHub(logger.Printf)
+	reader := &views.ProcessReader{Module: delivery.WatchmanModule, Call: calls.Call, Logf: logger.Printf, Hub: hub,
 		Findings: func() views.FindingWriter {
 			return sdkfindings.Writer(ctx, func() jetstream.KeyValue { return kvs.findings })
 		}}
@@ -450,17 +453,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 				w.TakeInCaughtUp()
 			}
 		}
-		hc, err := process.HearEvents(tc, func(m jetstream.Msg) { takeInHeard(w, m, logger.Printf) }, logger.Printf)
-		if err != nil {
+		if err := hub.Handle("mesh.mod."+delivery.IndexerModule+".event.take-in.*", func(m jetstream.Msg) { takeInHeard(w, m, logger.Printf) }); err != nil {
 			return err
 		}
-		defer hc.Stop()
 	}
 	w.Reconcile()
 	if err := hear(conn, w, dead, logger.Printf); err != nil {
 		return err
 	}
-	hearConsumed(ctx, js, cred, w, logger.Printf)
 
 	errs := make(chan error, 16)
 	if healthWatch != nil {
@@ -468,6 +468,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	}
 	// 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))
+	// Every part has registered: the hub pulls the module's durable, and the watchman's other consumers open.
+	hearConsumed(ctx, js, cred, hub, w, logger.Printf)
+	defer hub.Stop()
 	// 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) }()