Builder: its one pull of its durable on the shared events stream is the module's event hub: the seats' events and the condition keeper's are one handler, the views reader another, the hub started last (B7c finding 1; issue 1966)

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

modules/mesh-delivery-builder/cmd/mesh-delivery-builder/gone_integration_test.go

@@ -65,7 +65,7 @@ func (h *heldVerb) asks() int {
 // from the manifest's consumes.
 func eventsConsumer(t *testing.T, b *bed, node string) {
 	t.Helper()
-	if _, err := b.js.CreateStream(b.ctx, jetstream.StreamConfig{Name: eventsStream, Subjects: []string{delivery.ArtifactsGoneSubject}}); err != nil {
+	if _, err := b.js.CreateOrUpdateStream(b.ctx, jetstream.StreamConfig{Name: eventsStream, Subjects: []string{"mesh.mod.*.event.>", "mesh.seat.*.event.>"}}); err != nil {
 		t.Fatal(err)
 	}
 	if _, err := b.js.CreateOrUpdateConsumer(b.ctx, eventsStream, jetstream.ConsumerConfig{Durable: node + "_mesh-delivery-builder",

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

@@ -1191,11 +1191,10 @@ func (b *bed) registry(t *testing.T, node string, seatVersions ...string) *regis
 	mk(versions.FloorWaitsBucket)
 	// EVENTS with two seats' outcomes, and the forge bridge's stream (a branch's end): each with the builder's durable consumer.
 	for stream, subjects := range map[string][]string{
-		eventsStream: {delivery.ArtifactsGoneSubject, ociartifactregistry.SubjectPublished, ociartifactregistry.SubjectRefused, ociartifactregistry.SubjectRemoved,
-			gopackageregistry.SubjectPublished, gopackageregistry.SubjectRefused, gopackageregistry.SubjectRemoved},
+		eventsStream: {"mesh.mod.*.event.>", "mesh.seat.*.event.>"},
 		forgeStream: {process.ModuleStreamSubject(delivery.ForgeBridgeModule, ">")},
 	} {
-		if _, err := b.js.CreateStream(b.ctx, jetstream.StreamConfig{Name: stream, Subjects: subjects}); err != nil {
+		if _, err := b.js.CreateOrUpdateStream(b.ctx, jetstream.StreamConfig{Name: stream, Subjects: subjects}); err != nil {
 			t.Fatal(err)
 		}
 		if _, err := b.js.CreateOrUpdateConsumer(b.ctx, stream, jetstream.ConsumerConfig{Durable: node + "_" + delivery.BuilderModule,
@@ -1216,10 +1215,6 @@ func (b *bed) registry(t *testing.T, node string, seatVersions ...string) *regis
 		t.Fatal(err)
 	}
 	t.Cleanup(func() { _ = rm.Unsubscribe() })
-	own := "mesh.mod." + delivery.BuilderModule + ".event.>"
-	if _, err := b.js.CreateStream(b.ctx, jetstream.StreamConfig{Name: "BUILDER_EVENTS", Subjects: []string{own}}); err != nil {
-		t.Fatal(err)
-	}
 	nc := b.js.Conn()
 	// As a build module hears it: on the contract's subject, read by the contract's decoder (BG2's late publish).
 	sub, err := nc.Subscribe(delivery.VersionNumberedSubject, func(m *nats.Msg) {

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

@@ -75,7 +75,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		return err
 	}
 	defer calls.Close()
-	reader := &views.ProcessReader{Module: delivery.BuilderModule, Call: calls.Call, Logf: logger.Printf}
+	// The module has one durable consumer of EVENTS: one hub pulls it and hands each message to the part that hears it (the
+	// views reader and the builder's own events), started once every part is registered.
+	hub := process.NewEventHub(logger.Printf)
+	reader := &views.ProcessReader{Module: delivery.BuilderModule, Call: calls.Call, Logf: logger.Printf, Hub: hub}
 	findingsKV := process.AwaitBucket(ctx, js, conn, findingsBucket, "the builder's findings, a first view unanswered past its bound among them, are said in its log only", logger.Printf)
 	reader.Findings = func() views.FindingWriter { return sdkfindings.Writer(ctx, findingsKV.KV) }
 	modules, err := reader.Follow(delivery.IndexerModule, modulesState, delivery.IndexerModule+".indexer_read_modules")
@@ -330,16 +333,29 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	floor := floorWake{b: b, waits: floorWaits, logf: logger.Printf}
 	// And the four registry seats' published, refused and removed (DLV-B4), which move a numbered version.
 	seats := seatHearer{n: numberer, b: b, logf: logger.Printf, pub: pubs}
-	hear(eventsStream, delivery.ArtifactsGoneEvent+", what wakes a floor, the registry seats' words and manifest-reads-changed are not heard until the builder starts again", func(m jetstream.Msg) {
+	events := func(m jetstream.Msg) {
 		if reads.heard(m, logger.Printf) || floor.heard(ctx, m) || seats.heard(ctx, m) { // reads: the node-engine seat's manifest-reads-changed (ADR 0618 §4)
 			return
 		}
 		heardGone(ctx, b, m, logger.Printf)
-	})
+	}
+	// The seats' events (artifacts-gone, the registry seats' words, manifest-accepted and manifest-reads-changed) and the
+	// condition keeper's condition-cleared; the owners' views events are the views reader's.
+	for _, pattern := range []string{"mesh.seat.*.event.>", conditionClearedSubject} {
+		if err := hub.Handle(pattern, events); err != nil {
+			return err
+		}
+	}
 	if stopped != nil {
 		return stopped
 	}
 	reader.Start(ctx, js, cred, process.AwaitBucket(ctx, js, conn, firstViewsBucket, "builder_first_views reads nothing and builder_first_view_again is not heard", logger.Printf))
+	// Every part has registered: the hub pulls the module's durable. Where the consumer is not there yet the builder says so
+	// and binds it at a later change (heard.Writer; ADR 0615 §5).
+	_ = notHeard.Bind(eventsStream, delivery.ArtifactsGoneEvent+", what wakes a floor, the registry seats' words, the views and manifest-reads-changed are not heard until the builder starts again", func() error {
+		return hub.Start(ctx, js, cred)
+	})
+	go func() { <-ctx.Done(); hub.Stop() }()
 	// The modules, once their first views and the manifests' are answered: what a module builds is read from the manifests.
 	go func() { errs <- buildFromModules(ctx, b, modules, manifests, candidates) }()
 	// A package's name heard in the indexer's packages looks at its module again, so a done build is numbered then.