From 07a8f04e65a34781da07b8ae41ee54664de28e0f Mon Sep 17 00:00:00 2001
From: jochen <jschoubben@gmail.com>
Date: Mon, 19 Oct 2026 15:31:13 +0200
Subject: [PATCH] Count a module's own stream events beside its emits in the
 catalogue agreement (issue 1249)

---
 internal/broker/agreement_catalogue_test.go | 62 +++++++++++++++------
 internal/broker/agreement_streams_test.go   | 38 +++++++++++++
 2 files changed, 83 insertions(+), 17 deletions(-)
 create mode 100644 internal/broker/agreement_streams_test.go

diff --git a/internal/broker/agreement_catalogue_test.go b/internal/broker/agreement_catalogue_test.go
index 3bba76be..805ab90f 100644
--- a/internal/broker/agreement_catalogue_test.go
+++ b/internal/broker/agreement_catalogue_test.go
@@ -97,27 +97,17 @@ func theCataloguesEvents(t *testing.T) ([]AnEmitter, []AConsumer, []DeclaredSeat
 		if err != nil {
 			continue
 		}
-		var m struct {
-			Module   string   `json:"module"`
-			Emits    []string `json:"emits"`
-			Consumes []string `json:"consumes"`
-			Seats    []struct {
-				Name  string   `json:"name"`
-				Emits []string `json:"emits"`
-			} `json:"seats"`
-		}
-		if err := json.Unmarshal(raw, &m); err != nil {
+		emitter, consumer, declared, err := eventsOfManifest(raw)
+		if err != nil {
 			t.Fatalf("%s: %v", e.Name(), err)
 		}
-		if len(m.Emits) > 0 {
-			emitters = append(emitters, AnEmitter{Module: m.Module, Emits: m.Emits})
+		if emitter != nil {
+			emitters = append(emitters, *emitter)
 		}
-		if len(m.Consumes) > 0 {
-			consumers = append(consumers, AConsumer{Module: m.Module, Consumes: m.Consumes})
-		}
-		for _, s := range m.Seats {
-			seats = append(seats, DeclaredSeat{Name: s.Name, Emits: s.Emits})
+		if consumer != nil {
+			consumers = append(consumers, *consumer)
 		}
+		seats = append(seats, declared...)
 	}
 	if len(emitters) == 0 {
 		t.Fatalf("no module under %s emits anything, so this proved nothing", root)
@@ -245,3 +235,41 @@ func TestTheBrokerAndTheCatalogueAgreeOnTheRuntimeModule(t *testing.T) {
 		t.Fatalf("the broker calls the runtime %q and the catalogue %q", RuntimeModule, catalogue.RuntimeModule)
 	}
 }
+
+// eventsOfManifest is what one manifest says it emits and consumes for the agreement: its `emits`, and the events of
+// its own streams, which a consumer names as <module>.<event> as it names the others (novox/hq issue 1249: once
+// mesh-mail declared emits beside its stream, the stream's events were no longer counted).
+func eventsOfManifest(raw []byte) (*AnEmitter, *AConsumer, []DeclaredSeat, error) {
+	var m struct {
+		Module   string   `json:"module"`
+		Emits    []string `json:"emits"`
+		Consumes []string `json:"consumes"`
+		Streams  []struct {
+			Events []string `json:"events"`
+		} `json:"streams"`
+		Seats []struct {
+			Name  string   `json:"name"`
+			Emits []string `json:"emits"`
+		} `json:"seats"`
+	}
+	if err := json.Unmarshal(raw, &m); err != nil {
+		return nil, nil, nil, err
+	}
+	emits := append([]string{}, m.Emits...)
+	for _, s := range m.Streams {
+		emits = append(emits, s.Events...)
+	}
+	var emitter *AnEmitter
+	if len(emits) > 0 {
+		emitter = &AnEmitter{Module: m.Module, Emits: emits}
+	}
+	var consumer *AConsumer
+	if len(m.Consumes) > 0 {
+		consumer = &AConsumer{Module: m.Module, Consumes: m.Consumes}
+	}
+	var seats []DeclaredSeat
+	for _, s := range m.Seats {
+		seats = append(seats, DeclaredSeat{Name: s.Name, Emits: s.Emits})
+	}
+	return emitter, consumer, seats, nil
+}
diff --git a/internal/broker/agreement_streams_test.go b/internal/broker/agreement_streams_test.go
new file mode 100644
index 00000000..255fecc7
--- /dev/null
+++ b/internal/broker/agreement_streams_test.go
@@ -0,0 +1,38 @@
+package broker
+
+import (
+	"strings"
+	"testing"
+)
+
+// mailShaped is mesh-mail's manifest as mesh-catalog main (b767aac8) declares its events: emits beside two streams of
+// its own (modules/mesh-mail/module.json), trimmed to the keys the agreement reads.
+const mailShaped = `{"module": "mesh-mail",
+ "emits": ["mail.send.sent", "mail.send.failed"],
+ "streams": [
+  {"name": "mail", "events": ["mail.*.message.added", "mail.*.message.changed", "mail.*.message.removed"]},
+  {"name": "mail-sends", "events": ["mail.send.approved"]}]}`
+
+func TestAModulesStreamEventsCountBesideItsEmits(t *testing.T) { // 1249/AC1
+	emitter, _, _, err := eventsOfManifest([]byte(mailShaped))
+	if err != nil || emitter == nil {
+		t.Fatalf("emitter %v, err %v", emitter, err)
+	}
+	bad := Disagreements([]AnEmitter{*emitter},
+		[]AConsumer{{Module: "mesh-mail-notifier", Consumes: []string{"mesh-mail.mail.*.message.added"}}}, nil)
+	if len(bad) != 0 {
+		t.Fatalf("a consumer of mesh-mail's stream event was reported: %v", bad)
+	}
+}
+
+func TestAnEventNeitherEmittedNorOnTheStreamIsStillCaught(t *testing.T) { // 1249/AC2
+	emitter, _, _, err := eventsOfManifest([]byte(mailShaped))
+	if err != nil || emitter == nil {
+		t.Fatalf("emitter %v, err %v", emitter, err)
+	}
+	bad := Disagreements([]AnEmitter{*emitter},
+		[]AConsumer{{Module: "mesh-mail-notifier", Consumes: []string{"mesh-mail.mail.*.message.renamed"}}}, nil)
+	if len(bad) != 1 || !strings.Contains(bad[0], "mail.*.message.added") {
+		t.Fatalf("want one disagreement naming the stream's events among what mesh-mail emits, got %v", bad)
+	}
+}
-- 
2.54.0