Merge pull request 'A module may declare a stream of its own, heard only by the consumers its trusted config value grants (hq issue 718, ADR 0363)' (#338) from fix-718-module-streams-and-grants into main

cmd/mesh-controller/busobjects.go

@@ -60,9 +60,19 @@ func assertBusObjects(ctx context.Context, inv *inventory.Inventory, r broker.Ra
 	if err != nil {
 		return nil, err
 	}
+	// And every module's own stream (novox/hq ADR 0363), from the catalogue, before the consumers on it.
+	modStreams, err := inv.DeclaredStreams(ctx)
+	if err != nil {
+		return nil, err
+	}
 	// Every one tried, and every failure named: one module's consumer the bus refuses is no reason
 	// the modules after it in the list hear nothing (novox/hq issue 208, where this runs on each send).
 	var failed []error
+	for _, s := range broker.ModuleStreams(modStreams) {
+		if err := r.EnsureStream(s); err != nil {
+			failed = append(failed, fmt.Errorf("the module stream %s: %w", s.Name, err))
+		}
+	}
 	for _, s := range trafficStreams {
 		if err := r.EnsureStream(s); err != nil {
 			failed = append(failed, fmt.Errorf("the work queue %s: %w", s.Name, err))
@@ -73,11 +83,30 @@ func assertBusObjects(ctx context.Context, inv *inventory.Inventory, r broker.Ra
 			failed = append(failed, fmt.Errorf("the worker %s on %s: %w", c.Name, c.Stream, err))
 		}
 	}
+	restricted := map[string]bool{}
+	for _, s := range modStreams {
+		restricted[s.Name] = s.Restricted
+	}
 	for _, c := range consumers {
-		if err := r.EnsureConsumer(c.Consumer); err != nil {
+		if err := broker.EnsureHeard(r, c.Consumer, restricted[c.Consumer.Stream]); err != nil {
 			failed = append(failed, fmt.Errorf("how %s on %s hears what it consumes: %w", c.Module, c.Node, err))
 		}
 	}
+	// **And a consumer no longer granted loses what it reads through** (novox/hq ADR 0363 §6): every consumer
+	// on a module's stream the composition no longer asks for is removed, on this very send.
+	if p, prunes := r.(broker.Pruner); prunes && len(modStreams) > 0 {
+		var wanted []broker.Consumer
+		for _, c := range consumers {
+			wanted = append(wanted, c.Consumer)
+		}
+		removed, err := broker.PruneHeard(p, modStreams, wanted)
+		if err != nil {
+			failed = append(failed, err)
+		}
+		for _, gone := range removed {
+			fmt.Printf("  removed the consumer %s: its module is no longer granted that stream (novox/hq ADR 0363)\n", gone)
+		}
+	}
 	if len(failed) > 0 {
 		return nil, errors.Join(failed...)
 	}

cmd/mesh-controller/push.go

@@ -17,6 +17,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/nats-io/nats.go"
 	"github.com/novox/mesh-controller/internal/broker"
 	"github.com/novox/mesh-controller/internal/catalogue"
 	"github.com/novox/mesh-controller/internal/inventory"
@@ -1386,6 +1387,13 @@ func issueMemberships(ctx context.Context, open *stores, server *link.Server, se
 	if issued > 0 {
 		fmt.Printf("  issued %d membership(s)\n", issued)
 	}
+	// **And what every other machine hears of a module's stream** (novox/hq ADR 0363 §6): a grant the provider's
+	// setting changed reaches a consumer on a machine this send does not reach. Its membership is issued again
+	// as last issued, with only what it hears changed, so nothing else of it moves before its own send.
+	reheard := reissueHearing(ctx, bus, records, where, sent)
+	if reheard > 0 {
+		fmt.Printf("  issued %d membership(s) again for what they hear of a module's stream\n", reheard)
+	}
 	if len(refused) > 0 {
 		// Returned, never passed over (novox/hq issue 249): the declarations of the machines they are
 		// for are not sent, and the rollout that asked is tried again rather than waiting for a push.
@@ -1744,3 +1752,78 @@ func schemaReach() (int, error) {
 	}
 	return reach, nil
 }
+
+// reissueHearing issues again, on every machine not just sent, the membership of each assignment whose
+// `consumes` names a module owning a stream, when what it hears changed: read as last issued from the bus,
+// its `hears` replaced, and published. One that was never issued is left to its own send. Failures are said
+// and passed over: the send that carries the grant is not the consumer's, and its own next send issues it.
+func reissueHearing(ctx context.Context, bus link.OverNATS, records broker.Records, where broker.Placements,
+	sent []readyNode) int {
+	reached := map[string]bool{}
+	for _, s := range sent {
+		reached[s.node] = true
+	}
+	issued := 0
+	for node, declared := range records.Assigned {
+		if reached[node] {
+			continue
+		}
+		for _, d := range declared {
+			if !d.HearsAStream {
+				continue
+			}
+			subject := broker.MembershipSubject(node, d.Module)
+			last, err := bus.JS.GetLastMsg(broker.AssignmentsStream, subject)
+			if err != nil {
+				if !errors.Is(err, nats.ErrMsgNotFound) {
+					fmt.Printf("  %s on %s: its membership could not be read to issue what it hears again: %v\n",
+						d.Module, node, err)
+				}
+				continue
+			}
+			body, changed, err := withHearing(last.Data, broker.HeardIssuedFor(node, d))
+			if err != nil {
+				fmt.Printf("  %s on %s: its membership could not be read to issue what it hears again: %v\n",
+					d.Module, node, err)
+				continue
+			}
+			if !changed {
+				continue
+			}
+			if err := bus.PublishMembership(ctx, node, d.Module, body); err != nil {
+				fmt.Printf("  %s on %s: what it hears could not be issued again: %v\n", d.Module, node, err)
+				continue
+			}
+			issued++
+		}
+	}
+	return issued
+}
+
+// withHearing is a membership as issued with its `hears` replaced, and whether that changed it. Pure.
+func withHearing(issued []byte, hears []broker.HeardIssued) ([]byte, bool, error) {
+	var fields map[string]json.RawMessage
+	if err := json.Unmarshal(issued, &fields); err != nil {
+		return nil, false, err
+	}
+	var had []broker.HeardIssued
+	if raw, ok := fields["hears"]; ok {
+		if err := json.Unmarshal(raw, &had); err != nil {
+			return nil, false, err
+		}
+	}
+	if slices.Equal(had, hears) {
+		return issued, false, nil
+	}
+	if len(hears) == 0 {
+		delete(fields, "hears")
+	} else {
+		raw, err := json.Marshal(hears)
+		if err != nil {
+			return nil, false, err
+		}
+		fields["hears"] = raw
+	}
+	body, err := json.Marshal(fields)
+	return body, true, err
+}

cmd/mesh-controller/reissue_hearing_test.go

@@ -0,0 +1,39 @@
+package main
+
+import (
+	"encoding/json"
+	"testing"
+
+	"github.com/novox/mesh-controller/internal/broker"
+)
+
+// A membership issued again for what it hears changes nothing else of it (novox/hq ADR 0363 §6).
+func TestAMembershipIssuedAgainChangesOnlyWhatItHears(t *testing.T) {
+	issued := []byte(`{"node":"ace","module":"reader","receives":{"x":[1]},"emits":"mesh.mod.reader.event.{event}"}`)
+	hears := []broker.HeardIssued{{Stream: "MAIL", Consumer: "ace_reader"}}
+	body, changed, err := withHearing(issued, hears)
+	if err != nil || !changed {
+		t.Fatalf("changed %v, %v", changed, err)
+	}
+	var got map[string]any
+	if err := json.Unmarshal(body, &got); err != nil {
+		t.Fatal(err)
+	}
+	if got["receives"] == nil || got["emits"] != "mesh.mod.reader.event.{event}" || got["hears"] == nil {
+		t.Fatalf("issued again as %s", body)
+	}
+	if _, again, _ := withHearing(body, hears); again {
+		t.Fatal("the same hearing issued again counts as a change")
+	}
+	gone, changed, err := withHearing(body, nil)
+	if err != nil || !changed {
+		t.Fatalf("a withdrawal: %v, %v", changed, err)
+	}
+	got = nil
+	if err := json.Unmarshal(gone, &got); err != nil {
+		t.Fatal(err)
+	}
+	if _, still := got["hears"]; still {
+		t.Fatalf("a withdrawn hearing is still issued: %s", gone)
+	}
+}

internal/broker/derived.go

@@ -128,7 +128,13 @@ func ConsumerFor(p Principal) (Consumer, bool) {
 	// connected, and its graph stayed empty with nothing anywhere reporting why.
 	// And one that hears its own answers on a seat it uses (novox/hq ADR 0259 §3): an asker's warrants.
 	hearsItsOwn := len(SeatTrafficOf(p.Module, nil, p.Uses, nil).Subscribe) > 0
-	if p.Kind != KindModule || (len(p.Consumes) == 0 && len(p.Watches) == 0 && !hearsItsOwn) {
+	hearsAStream := false
+	for _, h := range p.Hears {
+		if len(h.Filters) > 0 {
+			hearsAStream = true
+		}
+	}
+	if p.Kind != KindModule || (len(p.Consumes) == 0 && len(p.Watches) == 0 && !hearsItsOwn && !hearsAStream) {
 		return Consumer{}, false
 	}
 	perms, err := PermissionsFor(p)
@@ -140,11 +146,14 @@ func ConsumerFor(p Principal) (Consumer, bool) {
 	// business, which is why this is a filter and not the whole list.
 	var filters []string
 	for _, s := range perms.Subscribe {
-		if strings.Contains(s, ".event.") {
+		// Not a module stream's, which its own consumer on that stream hears (novox/hq ADR 0363).
+		if strings.Contains(s, ".event.") && !strings.HasPrefix(s, moduleStreamPrefix) {
 			filters = append(filters, s)
 		}
 	}
-	if len(filters) == 0 {
+	// **A module that hears only a module's stream still has one here** (novox/hq ADR 0363 §7): what it
+	// gives up on there is delivered again under `mesh.again.<consumer>.` on EVENTS, and only this filters it.
+	if len(filters) == 0 && !hearsAStream {
 		return Consumer{}, false
 	}
 	sort.Strings(filters)
@@ -181,6 +190,13 @@ func ConsumersOf(users []Principal) []ModuleConsumer {
 		}
 		seen[p.Node+"/"+p.Module] = true
 		out = append(out, ModuleConsumer{Node: p.Node, Module: p.Module, Consumer: c})
+		// And one on each module stream it hears (novox/hq ADR 0363 §5).
+		for _, h := range p.Hears {
+			if len(h.Filters) == 0 {
+				continue
+			}
+			out = append(out, ModuleConsumer{Node: p.Node, Module: p.Module, Consumer: heardConsumer(p, h)})
+		}
 	}
 	for _, p := range users {
 		if p.Kind == KindModule {
@@ -193,7 +209,8 @@ func ConsumersOf(users []Principal) []ModuleConsumer {
 		}
 		for _, d := range p.Carries {
 			add(Principal{Kind: KindModule, Node: p.Node, Module: d.Module, Emits: d.Emits,
-				Consumes: d.Consumes, Serves: d.Serves, Holds: d.Holds, Uses: d.Uses, Watches: d.Watches})
+				Consumes: d.Consumes, Serves: d.Serves, Holds: d.Holds, Uses: d.Uses, Watches: d.Watches,
+				Streams: d.Streams, Hears: d.Hears})
 		}
 	}
 	return out

internal/broker/jetstream.go

@@ -486,3 +486,25 @@ func (j *JetStream) LogStreams() ([]string, error) {
 	}
 	return out, lister.Err()
 }
+
+// ConsumerNames is every durable consumer on a stream; none for a stream that does not exist yet.
+func (j *JetStream) ConsumerNames(stream string) ([]string, error) {
+	if _, err := j.js.StreamInfo(stream); errors.Is(err, nats.ErrStreamNotFound) {
+		return nil, nil
+	} else if err != nil {
+		return nil, err
+	}
+	var out []string
+	for name := range j.js.ConsumerNames(stream) {
+		out = append(out, name)
+	}
+	return out, nil
+}
+
+// RemoveConsumer deletes one durable consumer (novox/hq ADR 0363 §6): one the composition no longer asks for.
+func (j *JetStream) RemoveConsumer(stream, name string) error {
+	if err := j.js.DeleteConsumer(stream, name); err != nil && !errors.Is(err, nats.ErrConsumerNotFound) {
+		return err
+	}
+	return nil
+}

internal/broker/membership.go

@@ -61,6 +61,11 @@ type Membership struct {
 	// over every module on the machine, so one module's code reaching another's name or kind through
 	// the runtime is the runtime's to refuse.
 	SeatTraffic *SeatTraffic `json:"seat-traffic,omitempty"`
+	// Streams are this module's own streams (novox/hq ADR 0363 §2): an event one of them carries is published
+	// on that stream's subject, never on Emits. Hears are the module streams it hears, each through a durable
+	// consumer on that stream the runtime binds beside the one on EVENTS (§8).
+	Streams []StreamIssued `json:"streams,omitempty"`
+	Hears   []HeardIssued  `json:"hears,omitempty"`
 }
 
 // Served is one address a tool is answered on.
@@ -126,6 +131,8 @@ func MembershipFor(node string, d Declared, where Placements) Membership {
 	}
 	m.State = stateIssuedFor(d, node)
 	m.Logs = logsIssuedFor(d)
+	m.Streams = streamsIssuedFor(d)
+	m.Hears = HeardIssuedFor(node, d)
 	t := SeatTrafficOf(d.Module, d.Holds, d.Uses, d.Watches)
 	for _, s := range append(append([]Seat{}, d.Uses...), d.Watches...) {
 		if !s.Kinded {

internal/broker/modstreams.go

@@ -0,0 +1,243 @@
+package broker
+
+import (
+	"fmt"
+	"sort"
+	"strings"
+)
+
+// A module's own stream on the bus (novox/hq ADR 0363).
+//
+// A module declares a stream as it declares a log: named locally, created by the controller from the
+// catalogue on every raise and every send, never removed by the mesh. Its events are published on
+// `mesh.stream.<module>.event.<event>`, a subject space the shared EVENTS stream (`mesh.mod.*.event.>`)
+// does not take — the bus refuses two streams claiming one subject, so a module's stream cannot share
+// EVENTS' — and a consumer hears it through a durable consumer of its own on that stream, named as on
+// EVENTS. Pure; the controller's raise asserts what is composed here.
+
+// moduleStreamPrefix starts every subject a module's own stream takes.
+const moduleStreamPrefix = "mesh.stream."
+
+// ModuleStreamSubject is where a module's event on its own stream lands: `mesh.stream.<module>.event.<event>`.
+// The token before `.event.` is the emitter, as on EVENTS, so a runtime reads the same key from it.
+func ModuleStreamSubject(module, event string) string {
+	return moduleStreamPrefix + module + ".event." + event
+}
+
+// OnAModuleStream says a subject is one a module's own stream takes.
+func OnAModuleStream(subject string) bool { return strings.HasPrefix(subject, moduleStreamPrefix) }
+
+// WirePattern is a local event pattern as the bus spells it: `**` is `>`.
+func WirePattern(local string) string {
+	parts := strings.Split(local, ".")
+	for i, p := range parts {
+		if p == "**" {
+			parts[i] = ">"
+		}
+	}
+	return strings.Join(parts, ".")
+}
+
+// A ModuleStream is one module's declared stream as the bus holds it.
+type ModuleStream struct {
+	Module string
+	// Name is its name on the bus, `MAIL`.
+	Name string
+	// Events are the events it carries, as the bus spells them (WirePattern).
+	Events     []string
+	MaxAgeDays int
+	MaxMiB     int
+	// Restricted says a trusted config value of the module grants who hears it (ADR 0363 §4).
+	Restricted bool
+}
+
+// Subjects are the subjects the stream takes.
+func (s ModuleStream) Subjects() []string {
+	var out []string
+	for _, e := range s.Events {
+		out = append(out, ModuleStreamSubject(s.Module, e))
+	}
+	sort.Strings(out)
+	return out
+}
+
+// Stream is the stream the controller asserts: kept by limits, its age and size the module's, dropping
+// its oldest when full — a consumer that was away catches up within the age, and the module's own record
+// answers for anything older.
+func (s ModuleStream) Stream() Stream {
+	why := fmt.Sprintf("%s's own stream (novox/hq ADR 0363): its events, kept %d days and at most %d MiB so a "+
+		"consumer that was away catches up", s.Module, s.MaxAgeDays, s.MaxMiB)
+	if s.Restricted {
+		why += "; heard only by the consumers its trusted config value grants"
+	}
+	return Stream{
+		Name:      s.Name,
+		Subjects:  s.Subjects(),
+		Retention: RetentionLimits,
+		MaxAge:    s.MaxAgeDays * 24 * 60 * 60,
+		MaxBytes:  int64(s.MaxMiB) * 1024 * 1024,
+		Why:       why,
+	}
+}
+
+// ModuleStreams is every declared module stream as the controller asserts it, in name order.
+func ModuleStreams(declared []ModuleStream) []Stream {
+	sorted := append([]ModuleStream(nil), declared...)
+	sort.Slice(sorted, func(i, j int) bool { return sorted[i].Name < sorted[j].Name })
+	out := make([]Stream, 0, len(sorted))
+	for _, s := range sorted {
+		out = append(out, s.Stream())
+	}
+	return out
+}
+
+// Heard is a module's stream another module hears, and the subjects it hears of it: its `consumes`
+// within the stream and, for a restricted stream, within what the provider's trusted config value grants it
+// (ADR 0363 §5). Composed only with at least one subject.
+type Heard struct {
+	Stream  string
+	Emitter string
+	Filters []string
+}
+
+// heardConsumer is the durable consumer a module hears a module's stream through: named as its consumer
+// on EVENTS, on that stream, with the subjects it was granted as its filters.
+func heardConsumer(p Principal, h Heard) Consumer {
+	filters := append([]string(nil), h.Filters...)
+	sort.Strings(filters)
+	return Consumer{
+		Name:           consumerDurable(p),
+		Stream:         h.Stream,
+		Filters:        filters,
+		AckWaitSeconds: 30,
+		MaxDeliver:     5,
+		Why: "what " + p.Module + " consumes of " + h.Emitter + "'s stream " + h.Stream + " and is granted " +
+			"(novox/hq ADR 0363); after max-deliver it gives an event up, and the controller keeps it in " +
+			"DEAD_LETTERS until a person delivers it again or drops it",
+	}
+}
+
+// heardGrants are what a principal publishes to read and acknowledge the consumers it hears module streams
+// through: asking about each, pulling from it, acknowledging it — its own name on each stream, no other's.
+func heardGrants(p Principal, hears []Heard) []string {
+	var out []string
+	durable := consumerDurable(p)
+	for _, h := range hears {
+		if !safeSubject.MatchString(h.Stream) || len(h.Filters) == 0 {
+			continue
+		}
+		out = append(out,
+			"$JS.API.CONSUMER.INFO."+h.Stream+"."+durable,
+			"$JS.API.CONSUMER.MSG.NEXT."+h.Stream+"."+durable,
+			"$JS.ACK."+h.Stream+"."+durable+".>")
+	}
+	return out
+}
+
+// ownStreamGrants are the subjects a module publishes its own streams' events on, and no other's.
+func ownStreamGrants(module string, streams []ModuleStream) []string {
+	var out []string
+	for _, s := range streams {
+		if s.Module != module {
+			continue
+		}
+		out = append(out, s.Subjects()...)
+	}
+	return out
+}
+
+// StreamIssued is one of a module's own streams as its membership names it (ADR 0363 §2): the events it
+// carries, as the bus spells them, and the subject an event of them is published on, `{event}` the event.
+type StreamIssued struct {
+	Stream  string   `json:"stream"`
+	Events  []string `json:"events"`
+	Subject string   `json:"subject"`
+}
+
+// HeardIssued is one module stream an assignment hears, as its membership names it (ADR 0363 §8): the
+// stream and the durable consumer on it the runtime binds.
+type HeardIssued struct {
+	Stream   string `json:"stream"`
+	Consumer string `json:"consumer"`
+}
+
+// streamsIssuedFor is every stream of its own a module's membership names.
+func streamsIssuedFor(d Declared) []StreamIssued {
+	var out []StreamIssued
+	for _, s := range d.Streams {
+		if s.Module != d.Module || !safeSubject.MatchString(d.Module) {
+			continue
+		}
+		out = append(out, StreamIssued{Stream: s.Name, Events: append([]string(nil), s.Events...),
+			Subject: ModuleStreamSubject(d.Module, "{event}")})
+	}
+	return out
+}
+
+// HeardIssuedFor is every module stream an assignment hears, as its membership names it.
+func HeardIssuedFor(node string, d Declared) []HeardIssued {
+	var out []HeardIssued
+	for _, h := range d.Hears {
+		if len(h.Filters) == 0 {
+			continue
+		}
+		out = append(out, HeardIssued{Stream: h.Stream, Consumer: node + "_" + d.Module})
+	}
+	sort.Slice(out, func(i, j int) bool { return out[i].Stream < out[j].Stream })
+	return out
+}
+
+// Pruner is the part of a JetStream connection that removes a module's consumer on a module's stream it
+// may no longer hear (ADR 0363 §6). Optional: what only records what it is asked to assert has none.
+type Pruner interface {
+	ConsumerNames(stream string) ([]string, error)
+	RemoveConsumer(stream, name string) error
+}
+
+// PruneHeard removes, from each module stream, every durable consumer the composition no longer asks for,
+// and answers what it removed as `<stream>/<consumer>`. A consumer that loses its grant loses what it reads
+// through, whatever its runtime still holds; a stream that does not exist yet has nothing to remove.
+func PruneHeard(p Pruner, streams []ModuleStream, wanted []Consumer) ([]string, error) {
+	keep := map[string]bool{}
+	for _, c := range wanted {
+		keep[c.Stream+"/"+c.Name] = true
+	}
+	var removed []string
+	for _, s := range streams {
+		names, err := p.ConsumerNames(s.Name)
+		if err != nil {
+			return removed, fmt.Errorf("listing the consumers on %s: %w", s.Name, err)
+		}
+		for _, n := range names {
+			if keep[s.Name+"/"+n] {
+				continue
+			}
+			if err := p.RemoveConsumer(s.Name, n); err != nil {
+				return removed, fmt.Errorf("removing %s's consumer on %s, which is no longer granted: %w", n, s.Name, err)
+			}
+			removed = append(removed, s.Name+"/"+n)
+		}
+	}
+	return removed, nil
+}
+
+// EnsureHeard asserts a consumer on a module's stream (ADR 0363 §5, §6). **On a restricted stream, a consumer
+// the bus will not bring to match is removed**: a pull consumer delivers to whoever pulls it, so one left with
+// its wider filters after a narrowing was refused would keep handing over what the grant no longer names,
+// whatever the user list now says. The next assertion makes it again with the granted filters. The refusal is
+// returned either way, saying what was done, so the send says it and raises it.
+func EnsureHeard(r Raiser, c Consumer, restricted bool) error {
+	err := r.EnsureConsumer(c)
+	if err == nil || !restricted {
+		return err
+	}
+	p, prunes := r.(Pruner)
+	if !prunes {
+		return fmt.Errorf("%w; it could not be removed here, and may still deliver more than it is granted", err)
+	}
+	if rerr := p.RemoveConsumer(c.Stream, c.Name); rerr != nil {
+		return fmt.Errorf("%w; REMOVING IT FAILED TOO, so it may still deliver more than it is granted: %v", err, rerr)
+	}
+	return fmt.Errorf("%w; REMOVED, so it delivers nothing it is no longer granted: the next assertion makes it "+
+		"again with its granted filters (novox/hq ADR 0363 §6)", err)
+}

internal/broker/modstreams_live_test.go

@@ -0,0 +1,68 @@
+package broker
+
+import (
+	"slices"
+	"testing"
+	"time"
+
+	"github.com/novox/mesh-controller/internal/testbus"
+)
+
+// On a real bus: a module's stream is made beside EVENTS without a clash, keeps what the module publishes on
+// its own subjects, a granted consumer hears only its part, a narrowed grant narrows it, and a withdrawn one
+// is removed (novox/hq ADR 0363).
+//
+//	go test ./internal/broker/ -run TestAModuleStreamOnARealBus
+func TestAModuleStreamOnARealBus(t *testing.T) {
+	js, err := Dial(testbus.URL(t))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer js.Close()
+	for _, s := range []Stream{MeshStreams()[3], mailStream.Stream()} {
+		_ = js.js.DeleteStream(s.Name)
+		if err := js.EnsureStream(s); err != nil {
+			t.Fatalf("%s: %v", s.Name, err)
+		}
+		defer func(name string) { _ = js.js.DeleteStream(name) }(s.Name)
+	}
+	for _, account := range []string{"mesh", "work"} {
+		if _, err := js.js.Publish(ModuleStreamSubject("mesh-mail", "mail."+account+".message.added"), []byte("{}")); err != nil {
+			t.Fatalf("publishing %s's mail: %v", account, err)
+		}
+	}
+	if info, _ := js.js.StreamInfo(EventsStream); info.State.Msgs != 0 {
+		t.Fatalf("EVENTS kept %d of the module stream's events", info.State.Msgs)
+	}
+
+	reader := Principal{Kind: KindModule, Node: "ace", Module: "reader", Hears: []Heard{mailHeard}}
+	c := heardConsumer(reader, mailHeard)
+	if err := js.EnsureConsumer(c); err != nil {
+		t.Fatal(err)
+	}
+	time.Sleep(50 * time.Millisecond)
+	if info, _ := js.js.ConsumerInfo("MAIL", c.Name); info.NumPending != 1 {
+		t.Fatalf("the reader granted `mesh` has %d pending, want 1", info.NumPending)
+	}
+	narrowed := c
+	narrowed.Filters = []string{"mesh.stream.mesh-mail.event.mail.mesh.message.changed"}
+	if err := EnsureHeard(js, narrowed, true); err != nil {
+		t.Fatalf("narrowing the grant: %v", err)
+	}
+	if info, _ := js.js.ConsumerInfo("MAIL", c.Name); info.Config.FilterSubject != narrowed.Filters[0] || len(info.Config.FilterSubjects) != 0 {
+		t.Fatalf("narrowed to %q %v, want only %s", info.Config.FilterSubject, info.Config.FilterSubjects, narrowed.Filters[0])
+	}
+	if info, _ := js.js.ConsumerInfo("MAIL", c.Name); info.NumPending != 0 {
+		t.Fatalf("after narrowing, %d of the reader's former grant are still pending", info.NumPending)
+	}
+	removed, err := PruneHeard(js, []ModuleStream{mailStream}, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !slices.Equal(removed, []string{"MAIL/ace_reader"}) {
+		t.Fatalf("removed %v", removed)
+	}
+	if names, _ := js.ConsumerNames("MAIL"); len(names) != 0 {
+		t.Fatalf("still on MAIL: %v", names)
+	}
+}

internal/broker/modstreams_test.go

@@ -0,0 +1,210 @@
+package broker
+
+import (
+	"errors"
+	"reflect"
+	"slices"
+	"strings"
+	"testing"
+)
+
+// mailStream is mesh-mail's own stream as the bus holds it (novox/hq ADR 0363).
+var mailStream = ModuleStream{Module: "mesh-mail", Name: "MAIL",
+	Events:     []string{"mail.*.message.added", "mail.*.message.changed", "mail.*.message.removed"},
+	MaxAgeDays: 7, MaxMiB: 1024, Restricted: true}
+
+// mailHeard is what a reader granted the account `mesh` hears of it.
+var mailHeard = Heard{Stream: "MAIL", Emitter: "mesh-mail", Filters: []string{
+	"mesh.stream.mesh-mail.event.mail.mesh.message.added", "mesh.stream.mesh-mail.event.mail.mesh.message.changed"}}
+
+// A module's stream is its own: subjects EVENTS does not claim, its age and size, kept by limits (ADR 0363 §1, §2).
+func TestAModuleStreamTakesSubjectsEventsDoesNot(t *testing.T) {
+	s := mailStream.Stream()
+	if s.Name != "MAIL" || s.Retention != RetentionLimits || s.MaxAge != 7*24*60*60 || s.MaxBytes != 1024<<20 || s.DiscardNew {
+		t.Fatalf("composed as %+v", s)
+	}
+	want := []string{"mesh.stream.mesh-mail.event.mail.*.message.added", "mesh.stream.mesh-mail.event.mail.*.message.changed",
+		"mesh.stream.mesh-mail.event.mail.*.message.removed"}
+	if !reflect.DeepEqual(s.Subjects, want) {
+		t.Fatalf("subjects %v", s.Subjects)
+	}
+	for _, own := range MeshStreams() {
+		for _, a := range own.Subjects {
+			for _, b := range s.Subjects {
+				if SubjectsOverlap(a, b) {
+					t.Fatalf("%s claims %s, which overlaps %s of %s: the bus would refuse the stream", s.Name, b, a, own.Name)
+				}
+			}
+		}
+	}
+	if WirePattern("mail.**") != "mail.>" {
+		t.Fatal("`**` is spelled `>` on the wire")
+	}
+}
+
+// The provider publishes its stream's events on their subjects and nowhere on EVENTS for them; a consumer
+// granted a part hears it through its own consumer on the stream, and one not granted has nothing there
+// (ADR 0363 §2, §5).
+func TestTheProviderPublishesItsStreamAndOnlyAGrantedConsumerHearsIt(t *testing.T) {
+	provider, err := PermissionsFor(Principal{Kind: KindModule, Node: "novox", Module: "mesh-mail",
+		Streams: []ModuleStream{mailStream}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !MayPublish(provider, "mesh.stream.mesh-mail.event.mail.work.message.added") {
+		t.Fatal("the provider may not publish its own stream's event")
+	}
+	if MayPublish(provider, "mesh.mod.mesh-mail.event.mail.work.message.added") {
+		t.Fatal("the provider may publish its stream's event on EVENTS")
+	}
+	if MayPublish(provider, "mesh.stream.other.event.mail.work.message.added") {
+		t.Fatal("the provider may publish on another module's stream")
+	}
+
+	reader := Principal{Kind: KindModule, Node: "ace", Module: "reader", Hears: []Heard{mailHeard}}
+	perms, err := PermissionsFor(reader)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, s := range []string{"$JS.API.CONSUMER.MSG.NEXT.MAIL.ace_reader", "$JS.API.CONSUMER.INFO.MAIL.ace_reader",
+		"$JS.ACK.MAIL.ace_reader.1"} {
+		if !MayPublish(perms, s) {
+			t.Errorf("the granted reader may not publish %s", s)
+		}
+	}
+	if !MaySubscribe(perms, "mesh.stream.mesh-mail.event.mail.mesh.message.added") ||
+		MaySubscribe(perms, "mesh.stream.mesh-mail.event.mail.work.message.added") {
+		t.Fatal("the reader's subscription is not what it was granted")
+	}
+	if MayPublish(perms, "$JS.API.CONSUMER.MSG.NEXT.MAIL.ace_other") {
+		t.Fatal("the reader may pull another's consumer")
+	}
+
+	stranger, err := PermissionsFor(Principal{Kind: KindModule, Node: "ace", Module: "stranger",
+		Consumes: []string{"billing.paid"}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if MayPublish(stranger, "$JS.API.CONSUMER.MSG.NEXT.MAIL.ace_stranger") ||
+		MaySubscribe(stranger, "mesh.stream.mesh-mail.event.mail.mesh.message.added") {
+		t.Fatal("a module not granted reaches the stream")
+	}
+}
+
+// A granted consumer has a durable consumer on the stream, with its granted subjects as filters, and an EVENTS
+// consumer that hears what it gives up on delivered again (ADR 0363 §5, §7); the runtime carrying it may read both.
+func TestAGrantedConsumerHasAConsumerOnTheStream(t *testing.T) {
+	records := Records{Nodes: []string{"ace", "novox"}, Assigned: map[string][]Declared{
+		"novox": {{Module: "mesh-mail", Streams: []ModuleStream{mailStream}}},
+		"ace":   {{Module: RuntimeModule}, {Module: "reader", Hears: []Heard{mailHeard}, HearsAStream: true}, {Module: "stranger", HearsAStream: true}},
+	}}
+	users, err := Users(records)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var onMail, onEvents []Consumer
+	for _, c := range ConsumersOf(users) {
+		switch c.Consumer.Stream {
+		case "MAIL":
+			onMail = append(onMail, c.Consumer)
+		case EventsStream:
+			onEvents = append(onEvents, c.Consumer)
+		}
+	}
+	if len(onMail) != 1 || onMail[0].Name != "ace_reader" || !slices.Equal(onMail[0].Filters, mailHeard.Filters) {
+		t.Fatalf("on MAIL: %+v", onMail)
+	}
+	if len(onEvents) != 1 || onEvents[0].Name != "ace_reader" || !slices.Equal(onEvents[0].Filters, []string{AgainFilter("ace_reader")}) {
+		t.Fatalf("on EVENTS: %+v", onEvents)
+	}
+	for _, u := range users {
+		if u.Kind != KindNodeTools || u.Node != "ace" {
+			continue
+		}
+		perms, err := PermissionsFor(u)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if !MayPublish(perms, "$JS.API.CONSUMER.MSG.NEXT.MAIL.ace_reader") || !MayPublish(perms, "$JS.ACK.MAIL.ace_reader.1") {
+			t.Fatal("the runtime carrying the reader may not read its consumer on MAIL")
+		}
+		if MayPublish(perms, "$JS.API.CONSUMER.MSG.NEXT.MAIL.ace_stranger") {
+			t.Fatal("the runtime may read a consumer no module it carries is granted")
+		}
+	}
+}
+
+// The membership says where a module's stream events go and which consumers on module streams to bind
+// (ADR 0363 §2, §8).
+func TestTheMembershipNamesTheStreamAndWhatIsHeard(t *testing.T) {
+	where := Placements{Nodes: map[string][]string{}}
+	provider := MembershipFor("novox", Declared{Module: "mesh-mail", Streams: []ModuleStream{mailStream}}, where)
+	if len(provider.Streams) != 1 || provider.Streams[0].Stream != "MAIL" ||
+		provider.Streams[0].Subject != "mesh.stream.mesh-mail.event.{event}" ||
+		!slices.Equal(provider.Streams[0].Events, mailStream.Events) {
+		t.Fatalf("the provider's membership: %+v", provider.Streams)
+	}
+	reader := MembershipFor("ace", Declared{Module: "reader", Hears: []Heard{mailHeard}}, where)
+	if !reflect.DeepEqual(reader.Hears, []HeardIssued{{Stream: "MAIL", Consumer: "ace_reader"}}) {
+		t.Fatalf("the reader's membership: %+v", reader.Hears)
+	}
+	if none := MembershipFor("ace", Declared{Module: "stranger", HearsAStream: true}, where); none.Hears != nil {
+		t.Fatalf("a module not granted is told to bind %+v", none.Hears)
+	}
+}
+
+type fakePruner struct {
+	on      map[string][]string
+	removed []string
+}
+
+func (f *fakePruner) ConsumerNames(stream string) ([]string, error) { return f.on[stream], nil }
+func (f *fakePruner) RemoveConsumer(stream, name string) error {
+	f.removed = append(f.removed, stream+"/"+name)
+	return nil
+}
+
+// A consumer that lost its grant has its consumer on the stream removed by the next assertion (ADR 0363 §6).
+func TestAConsumerThatLostItsGrantIsRemoved(t *testing.T) {
+	f := &fakePruner{on: map[string][]string{"MAIL": {"ace_reader", "ace_former"}}}
+	wanted := []Consumer{{Name: "ace_reader", Stream: "MAIL"}, {Name: "ace_former", Stream: EventsStream}}
+	removed, err := PruneHeard(f, []ModuleStream{mailStream}, wanted)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !slices.Equal(removed, []string{"MAIL/ace_former"}) || !slices.Equal(f.removed, removed) {
+		t.Fatalf("removed %v", removed)
+	}
+}
+
+type refusingRaiser struct {
+	fakePruner
+	refuse bool
+}
+
+func (r *refusingRaiser) EnsureStream(Stream) error { return nil }
+func (r *refusingRaiser) EnsureConsumer(Consumer) error {
+	if r.refuse {
+		return errors.New("consumer update refused")
+	}
+	return nil
+}
+
+// A narrowing the bus refuses on a restricted stream removes the consumer, so its wider filters deliver nothing
+// more; on an unrestricted stream the refusal is only said (review of PR 338, ADR 0363 §6).
+func TestARefusedNarrowingOnARestrictedStreamRemovesTheConsumer(t *testing.T) {
+	c := Consumer{Name: "ace_reader", Stream: "MAIL", Filters: mailHeard.Filters[:1]}
+	r := &refusingRaiser{refuse: true}
+	err := EnsureHeard(r, c, true)
+	if err == nil || !strings.Contains(err.Error(), "REMOVED") || !slices.Equal(r.removed, []string{"MAIL/ace_reader"}) {
+		t.Fatalf("restricted: %v, removed %v", err, r.removed)
+	}
+	r = &refusingRaiser{refuse: true}
+	if err := EnsureHeard(r, c, false); err == nil || len(r.removed) != 0 {
+		t.Fatalf("unrestricted: %v, removed %v", err, r.removed)
+	}
+	r = &refusingRaiser{}
+	if err := EnsureHeard(r, c, true); err != nil || len(r.removed) != 0 {
+		t.Fatalf("accepted: %v, removed %v", err, r.removed)
+	}
+}

internal/broker/nats.go

@@ -141,6 +141,10 @@ type Principal struct {
 	KeyedReads []KeyedRead
 	// Logs is the local names of the logs this principal's module keeps (novox/hq ADR 0297).
 	Logs []string
+	// Streams are the streams this principal's module owns, and Hears the module streams it hears with the
+	// subjects it was granted (novox/hq ADR 0363).
+	Streams []ModuleStream
+	Hears   []Heard
 
 	// SnapshotsTheBus is the bus's own module, the one holding mesh-broker (novox/hq ADR 0235). Its
 	// whole authority is BusSnapshotGrants: it copies the streams for the night's backup and nothing
@@ -637,6 +641,8 @@ func PermissionsFor(p Principal) (Permissions, error) {
 		for _, e := range p.Emits {
 			pub = append(pub, own+".event."+e)
 		}
+		// And its own streams' events, on their own subjects (novox/hq ADR 0363 §2).
+		pub = append(pub, ownStreamGrants(p.Module, p.Streams)...)
 		// Every tool under its own name, not a list: the tools a module serves are what its code
 		// answers, and a second copy of that list in the manifest would be a second source of
 		// truth for the mesh to keep in step (2026-09-28: every module that served a tool was
@@ -678,6 +684,16 @@ func PermissionsFor(p Principal) (Permissions, error) {
 			sub = append(sub, subject)
 		}
 
+		// 2a. What it hears of a module's own stream, as it was granted (novox/hq ADR 0363 §5): the subjects,
+		//     and its own consumer on that stream, asked about, pulled from and acknowledged.
+		for _, h := range p.Hears {
+			if len(h.Filters) == 0 {
+				continue
+			}
+			sub = append(sub, h.Filters...)
+		}
+		pub = append(pub, heardGrants(p, p.Hears)...)
+
 		// 2b. Events of a role it watches, under the seat's own namespace. Subscribe only: watching a
 		//     role is hearing what it announced, not taking part in it.
 		for _, w := range p.Watches {
@@ -800,6 +816,8 @@ func PermissionsFor(p Principal) (Permissions, error) {
 			for _, e := range d.Emits {
 				pub = append(pub, own+".event."+e)
 			}
+			// And its own streams' events (novox/hq ADR 0363 §2).
+			pub = append(pub, ownStreamGrants(d.Module, d.Streams)...)
 			for _, s := range d.Holds {
 				if servedOnlyByTheController(s) {
 					continue
@@ -838,7 +856,10 @@ func PermissionsFor(p Principal) (Permissions, error) {
 		// consumer is still the controller's to make, from the module's own principal.
 		for _, d := range p.Carries {
 			own := Principal{Kind: KindModule, Node: p.Node, Module: d.Module, Emits: d.Emits,
-				Consumes: d.Consumes, Serves: d.Serves, Holds: d.Holds, Uses: d.Uses, Watches: d.Watches}
+				Consumes: d.Consumes, Serves: d.Serves, Holds: d.Holds, Uses: d.Uses, Watches: d.Watches,
+				Streams: d.Streams, Hears: d.Hears}
+			// And the module streams it hears, each through its own consumer there (novox/hq ADR 0363 §8).
+			pub = append(pub, heardGrants(own, d.Hears)...)
 			if _, consumes := ConsumerFor(own); !consumes {
 				continue
 			}

internal/broker/stream_names_agreement_test.go

@@ -0,0 +1,23 @@
+package broker
+
+import (
+	"slices"
+	"testing"
+
+	"github.com/novox/mesh-controller/internal/catalogue"
+)
+
+// The catalogue's list of the mesh's own stream names, which a module's stream may not take, is the broker's
+// own list (novox/hq ADR 0363 §1): read from MeshStreams, so a stream added there cannot be forgotten there.
+func TestTheCataloguesReservedStreamNamesAreTheMeshsOwn(t *testing.T) {
+	var own []string
+	for _, s := range MeshStreams() {
+		own = append(own, s.Name)
+	}
+	reserved := append([]string(nil), catalogue.MeshStreamNames...)
+	slices.Sort(own)
+	slices.Sort(reserved)
+	if !slices.Equal(own, reserved) {
+		t.Fatalf("the broker's own streams are %v, the catalogue reserves %v", own, reserved)
+	}
+}

internal/broker/users.go

@@ -40,6 +40,13 @@ type Declared struct {
 	State []Bucket
 	// Logs are the logs it keeps, each a stream its instances append to (novox/hq ADR 0297).
 	Logs []Log
+	// Streams are the streams it owns, each its events' one place (novox/hq ADR 0363).
+	Streams []ModuleStream
+	// Hears are the module streams it hears on this machine, each with the subjects it was granted
+	// (novox/hq ADR 0363 §5); HearsAStream says its `consumes` names a module that owns a stream, granted
+	// or not, so its membership is issued again when a grant changes.
+	Hears        []Heard
+	HearsAStream bool
 	// Reads are other modules' state it reads, each `<module>.<name>` (novox/hq ADR 0201).
 	Reads []string
 	// KeyedReads are keys of other modules' state it reads, each one key alone: what a seat's holder is
@@ -129,7 +136,7 @@ func Users(r Records) ([]Principal, error) {
 				Holds: d.Holds, Uses: d.Uses, Watches: d.Watches, Invokes: d.Invokes,
 				State: stateNames(d.State), Reads: d.Reads, SnapshotsTheBus: d.SnapshotsTheBus,
 				PerMachine: perMachineNames(d.State), KeyedReads: d.KeyedReads,
-				Logs: logNames(d.Logs),
+				Logs: logNames(d.Logs), Streams: d.Streams, Hears: d.Hears,
 			})
 		}
 		if runtimeHere {

internal/catalogue/manifest.go

@@ -523,6 +523,12 @@ type Manifest struct {
 	// (novox/hq ADR 0297). A log is its owner's alone: no other module reads it.
 	Logs []LogDeclaration `json:"logs,omitempty"`
 
+	// Streams are the streams this module owns on the bus (novox/hq ADR 0363): each named locally, the
+	// events it carries named as `emits` names them but with `*` and `**`, and bounded in age and size; the
+	// controller creates each. One may be restricted by a trusted setting of this module, so a consumer
+	// hears a part of it only where that setting's value names it.
+	Streams []StreamDeclaration `json:"streams,omitempty"`
+
 	// Settings are the defaults this module gives its settings (novox/hq ADR 0262): each key a file,
 	// a contribution or a served fact asks for as `${setting:<key>}`, its default, and why that
 	// default. Only a preference has one — a font size, a width, a number of workers — and a value
@@ -1648,7 +1654,8 @@ func ParseManifest(raw []byte) (Manifest, error) {
 			name string
 			n    int
 		}{{"consumes", len(m.Consumes)}, {"uses", len(m.Uses)}, {"invokes", len(m.Invokes)},
-			{"state", len(m.State)}, {"logs", len(m.Logs)}, {"reads", len(m.Reads)}} {
+			{"state", len(m.State)}, {"logs", len(m.Logs)}, {"reads", len(m.Reads)},
+			{"streams", len(m.Streams)}} {
 			if f.n > 0 {
 				said = append(said, f.name)
 			}
@@ -1792,6 +1799,8 @@ func ParseManifest(raw []byte) (Manifest, error) {
 	problems = append(problems, StateProblems(m)...)
 	// And what it may call its logs, and how large it may ask them to grow (logs.go, novox/hq ADR 0297).
 	problems = append(problems, LogProblems(m)...)
+	// And the streams it owns, and who may hear them (streams.go, novox/hq ADR 0363).
+	problems = append(problems, StreamProblems(m)...)
 	// And the defaults it gives its settings (setting_defaults.go, novox/hq ADR 0262).
 	problems = append(problems, SettingProblems(m)...)
 	// And what it requires optionally, and the files it delivers only while bound (novox/hq ADR 0330).

internal/catalogue/seats_declared.go

@@ -245,6 +245,8 @@ func IsUndeclaredSeat(problem string) bool { return strings.HasSuffix(problem, U
 
 func CatalogueProblems(shelf Shelf) []string {
 	var problems []string
+	// A stream name has one owner (novox/hq ADR 0363).
+	problems = append(problems, StreamNameProblems(shelf)...)
 
 	// Who declares what, and who declared it first.
 	declaredBy := map[string]string{}

internal/catalogue/streams.go

@@ -0,0 +1,479 @@
+package catalogue
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"slices"
+	"strings"
+)
+
+// A module's own stream (novox/hq ADR 0363).
+//
+// A module declares a stream as it declares a log (ADR 0297) or a bucket (ADR 0201): named locally, the
+// events it carries named locally as `emits` names them, its age and its size within the mesh's bounds.
+// The controller creates it and stays the only writer of stream definitions. A stream may be restricted
+// by a trusted config value of its own module: then a consumer hears a part of it only where its own
+// `consumes` asks and that value names it.
+
+// The mesh's bounds on a module's stream: its age in days and its size in MiB, and what it gets when it
+// says nothing.
+const (
+	StreamLeastDays   = 1
+	StreamMostDays    = 30
+	StreamDefaultDays = 7
+	StreamLeastMiB    = 1
+	StreamMostMiB     = 8192
+	StreamDefaultMiB  = 1024
+)
+
+// GrantKey is the name in a grant's pattern that each key of the config value fills.
+const GrantKey = "{key}"
+
+// ReservedStreamPrefixes are the starts of the stream names the mesh derives, which no module's stream
+// may take: a seat's work queue, a log, a bucket, an object store, and the dead letters.
+var ReservedStreamPrefixes = []string{"SEAT_", "LOG_", "KV_", "OBJ_", "DEAD_"}
+
+// MeshStreamNames are the mesh's own streams by name. Written here as well as in the broker, which this
+// package cannot import; a test keeps them agreeing.
+var MeshStreamNames = []string{"CONTROL", "NODES", "ASSIGNMENTS", "EVENTS", "DEAD_LETTER_NOTICES", "DEAD_LETTERS"}
+
+// StreamDeclaration is one stream a module owns.
+type StreamDeclaration struct {
+	Name string `json:"name"`
+	// Events are the events it carries, named locally; `*` is one name and `**`, last only, the rest.
+	Events []string `json:"events"`
+	// MaxAgeDays and MaxMiB bound it; zero is the default.
+	MaxAgeDays int `json:"max-age-days,omitempty"`
+	MaxMiB     int `json:"max-mib,omitempty"`
+	// Grant restricts it, when said.
+	Grant *StreamGrant `json:"grant,omitempty"`
+}
+
+// StreamGrant says which trusted config value of the module grants a consumer a part of the stream, and
+// which part each of its keys names.
+type StreamGrant struct {
+	// Config is the trusted config value of the module that grants it (ADR 0348 §8): set only at the
+	// controller's terminal or on the operator's Approve.
+	Config string `json:"config"`
+	// Events is a pattern holding GrantKey exactly once, which each key of the config value fills.
+	Events string `json:"events"`
+}
+
+// UnmarshalJSON reads a stream strictly: an unknown key is refused here, not passed over.
+func (s *StreamDeclaration) UnmarshalJSON(raw []byte) error {
+	type plain StreamDeclaration
+	var p plain
+	dec := json.NewDecoder(bytes.NewReader(raw))
+	dec.DisallowUnknownFields()
+	if err := dec.Decode(&p); err != nil {
+		return fmt.Errorf("a stream is {name, events, max-age-days, max-mib, grant}: %w", typedUnknown(err))
+	}
+	*s = StreamDeclaration(p)
+	return nil
+}
+
+// UnmarshalJSON reads a grant strictly.
+func (g *StreamGrant) UnmarshalJSON(raw []byte) error {
+	type plain StreamGrant
+	var p plain
+	dec := json.NewDecoder(bytes.NewReader(raw))
+	dec.DisallowUnknownFields()
+	if err := dec.Decode(&p); err != nil {
+		return fmt.Errorf("a stream's grant is {config, events}: %w", typedUnknown(err))
+	}
+	*g = StreamGrant(p)
+	return nil
+}
+
+// Days is the stream's age in days, the default where none is said.
+func (s StreamDeclaration) Days() int {
+	if s.MaxAgeDays == 0 {
+		return StreamDefaultDays
+	}
+	return s.MaxAgeDays
+}
+
+// Cap is the stream's size in MiB, the default where none is said.
+func (s StreamDeclaration) Cap() int {
+	if s.MaxMiB == 0 {
+		return StreamDefaultMiB
+	}
+	return s.MaxMiB
+}
+
+// BusName is the stream's name on the bus: its local name in capitals, a hyphen an underscore.
+func (s StreamDeclaration) BusName() string { return StreamBusName(s.Name) }
+
+// StreamBusName is a local stream name as the bus holds it.
+func StreamBusName(local string) string {
+	return strings.ToUpper(strings.ReplaceAll(local, "-", "_"))
+}
+
+// Carries says whether an event, or every event a pattern names, is one this stream carries: within one
+// of its patterns.
+func (s StreamDeclaration) Carries(pattern string) bool {
+	for _, p := range s.Events {
+		if within(pattern, p) {
+			return true
+		}
+	}
+	return false
+}
+
+// StreamProblems is what is wrong with a manifest's streams, judged on the one manifest.
+func StreamProblems(m Manifest) []string {
+	if len(m.Streams) == 0 {
+		return nil
+	}
+	var problems []string
+	if !stateName.MatchString(m.Module) {
+		problems = append(problems, fmt.Sprintf(
+			"%s declares a stream, and a module's name is part of its stream's subjects, which take one plain "+
+				"name — no dot (novox/hq ADR 0363)", m.Module))
+	}
+	seen := map[string]bool{}
+	terminal := TerminalKeys(m)
+	for i, s := range m.Streams {
+		switch {
+		case !stateName.MatchString(s.Name):
+			problems = append(problems, fmt.Sprintf(
+				"%s declares stream %q: a stream is named locally — lower-case letters, digits and hyphens "+
+					"(novox/hq ADR 0363)", m.Module, s.Name))
+		case seen[s.BusName()]:
+			problems = append(problems, fmt.Sprintf("%s declares stream %q twice", m.Module, s.Name))
+		case slices.Contains(MeshStreamNames, s.BusName()):
+			problems = append(problems, fmt.Sprintf(
+				"%s declares stream %q, which is the mesh's own stream %s (novox/hq ADR 0363)", m.Module, s.Name, s.BusName()))
+		default:
+			for _, prefix := range ReservedStreamPrefixes {
+				if strings.HasPrefix(s.BusName(), prefix) {
+					problems = append(problems, fmt.Sprintf(
+						"%s declares stream %q, whose name on the bus starts with %s, which the mesh derives names "+
+							"with (novox/hq ADR 0363)", m.Module, s.Name, prefix))
+				}
+			}
+		}
+		seen[s.BusName()] = true
+		if s.MaxAgeDays != 0 && (s.MaxAgeDays < StreamLeastDays || s.MaxAgeDays > StreamMostDays) {
+			problems = append(problems, fmt.Sprintf("%s keeps stream %q %d days; a stream keeps between %d and %d "+
+				"(novox/hq ADR 0363)", m.Module, s.Name, s.MaxAgeDays, StreamLeastDays, StreamMostDays))
+		}
+		if s.MaxMiB != 0 && (s.MaxMiB < StreamLeastMiB || s.MaxMiB > StreamMostMiB) {
+			problems = append(problems, fmt.Sprintf("%s caps stream %q at %d MiB; a stream holds between %d and %d "+
+				"(novox/hq ADR 0363)", m.Module, s.Name, s.MaxMiB, StreamLeastMiB, StreamMostMiB))
+		}
+		if len(s.Events) == 0 {
+			problems = append(problems, fmt.Sprintf("%s declares stream %q with no events", m.Module, s.Name))
+		}
+		for _, e := range s.Events {
+			if err := eventPattern(e, false); err != nil {
+				problems = append(problems, fmt.Sprintf("%s's stream %q carries %q: %v", m.Module, s.Name, e, err))
+			}
+		}
+		// **An event goes to one place** (ADR 0363 §2): an emitted event a stream carries, or two streams
+		// carrying one event, would leave the module's runtime to choose.
+		for _, e := range m.EmitsAll() {
+			if s.Carries(e) {
+				problems = append(problems, fmt.Sprintf(
+					"%s emits %q, which its stream %q carries: an event goes to EVENTS or to the module's stream, "+
+						"so name it in one place (novox/hq ADR 0363)", m.Module, e, s.Name))
+			}
+		}
+		for _, other := range m.Streams[i+1:] {
+			for _, a := range s.Events {
+				for _, b := range other.Events {
+					if patternsOverlap(a, b) {
+						problems = append(problems, fmt.Sprintf(
+							"%s's streams %q and %q both carry %q and %q: an event goes to one stream (novox/hq "+
+								"ADR 0363)", m.Module, s.Name, other.Name, a, b))
+					}
+				}
+			}
+		}
+		if g := s.Grant; g != nil {
+			if !slices.Contains(terminal, g.Config) {
+				problems = append(problems, fmt.Sprintf(
+					"%s's stream %q is granted by %q, which is not a trusted config value of the module, set only at "+
+						"the controller's terminal or on the operator's Approve: a grant of what a module hears is the "+
+						"operator's (novox/hq ADR 0348 §8, ADR 0363)", m.Module, s.Name, g.Config))
+			}
+			if strings.Count(g.Events, GrantKey) != 1 {
+				problems = append(problems, fmt.Sprintf(
+					"%s's stream %q grants %q, which must hold %s exactly once, as a name of its own (novox/hq ADR 0363)",
+					m.Module, s.Name, g.Events, GrantKey))
+			} else if err := eventPattern(g.Events, true); err != nil {
+				problems = append(problems, fmt.Sprintf("%s's stream %q grants %q: %v", m.Module, s.Name, g.Events, err))
+			}
+		}
+	}
+	return problems
+}
+
+// eventPattern checks one local event pattern: names, `*` for one, `**` last for the rest; with key,
+// GrantKey is a name too.
+func eventPattern(pattern string, key bool) error {
+	if pattern == "" {
+		return fmt.Errorf("a pattern needs a name")
+	}
+	parts := strings.Split(pattern, ".")
+	for i, part := range parts {
+		switch {
+		case part == TheRest:
+			if i != len(parts)-1 {
+				return fmt.Errorf("%q stands for the rest of a name, so nothing may follow it", TheRest)
+			}
+		case part == OneName:
+		case key && part == GrantKey:
+		case !eventName.MatchString(part):
+			return fmt.Errorf("%q is not a usable name: lower-case letters, digits and dashes", part)
+		}
+	}
+	return nil
+}
+
+// StreamsHearing is every stream of a module that carries some of a consumed pattern's events, and
+// whether one of them carries all of them.
+func (m Manifest) StreamsHearing(pattern string) (hearing []StreamDeclaration, whole bool) {
+	for _, s := range m.Streams {
+		for _, p := range s.Events {
+			if patternsOverlap(pattern, p) {
+				hearing = append(hearing, s)
+				break
+			}
+		}
+		if s.Carries(pattern) {
+			whole = true
+		}
+	}
+	return hearing, whole
+}
+
+// within says every event pattern a names is one b names too.
+func within(a, b string) bool {
+	x, y := strings.Split(a, "."), strings.Split(b, ".")
+	for i := 0; ; i++ {
+		switch {
+		case i == len(y):
+			return i == len(x)
+		case y[i] == TheRest:
+			return i < len(x)
+		case i == len(x):
+			return false
+		case x[i] == TheRest:
+			return false
+		case y[i] == OneName:
+			continue
+		case x[i] == OneName:
+			return false
+		case x[i] != y[i]:
+			return false
+		}
+	}
+}
+
+// patternsOverlap says some event both patterns name.
+func patternsOverlap(a, b string) bool {
+	x, y := strings.Split(a, "."), strings.Split(b, ".")
+	for i := 0; ; i++ {
+		switch {
+		case i == len(x) && i == len(y):
+			return true
+		case i == len(x) || i == len(y):
+			return false
+		case x[i] == TheRest || y[i] == TheRest:
+			return true
+		case x[i] == OneName || y[i] == OneName || x[i] == y[i]:
+			continue
+		default:
+			return false
+		}
+	}
+}
+
+// StreamNameProblems are the rules about stream names no single manifest can be judged against: two
+// modules declaring one name.
+func StreamNameProblems(shelf Shelf) []string {
+	var problems []string
+	declaredBy := map[string]string{}
+	for _, module := range shelfOrder(shelf) {
+		for _, s := range shelf[module].Streams {
+			name := s.BusName()
+			if first, taken := declaredBy[name]; taken && first != module {
+				problems = append(problems, fmt.Sprintf(
+					"%s declares the stream %s, which %s already declares; a stream has one owner (novox/hq ADR 0363)",
+					module, name, first))
+				continue
+			}
+			declaredBy[name] = module
+		}
+	}
+	return problems
+}
+
+// GrantedKeys reads a restricting config value (novox/hq ADR 0363 §4): each key whose list names the
+// module wherever it runs, or the module on this machine as `<machine>.<module>`. A value of another shape
+// grants nothing; a key that is not one usable name grants nothing.
+func GrantedKeys(value any, module, machine string) []string {
+	if text, isText := value.(string); isText {
+		// A layer may hold the object as its JSON text.
+		var parsed any
+		if json.Unmarshal([]byte(text), &parsed) != nil {
+			return nil
+		}
+		value = parsed
+	}
+	entries, ok := value.(map[string]any)
+	if !ok {
+		return nil
+	}
+	var out []string
+	for key, listed := range entries {
+		if !eventName.MatchString(key) {
+			continue
+		}
+		names, ok := listed.([]any)
+		if !ok {
+			continue
+		}
+		for _, n := range names {
+			s, _ := n.(string)
+			if s == module || (machine != "" && s == machine+"."+module) {
+				out = append(out, key)
+				break
+			}
+		}
+	}
+	slices.Sort(out)
+	return out
+}
+
+// Granted is a grant's pattern with its key filled.
+func (g StreamGrant) Granted(key string) string { return strings.Replace(g.Events, GrantKey, key, 1) }
+
+// narrow is the events both patterns name, as one pattern; false when they name none in common.
+func narrow(a, b string) (string, bool) {
+	x, y := strings.Split(a, "."), strings.Split(b, ".")
+	var out []string
+	for i := 0; ; i++ {
+		switch {
+		case i == len(x) && i == len(y):
+			return strings.Join(out, "."), true
+		case i == len(x) || i == len(y):
+			return "", false
+		case x[i] == TheRest:
+			return strings.Join(append(out, y[i:]...), "."), true
+		case y[i] == TheRest:
+			return strings.Join(append(out, x[i:]...), "."), true
+		case x[i] == OneName:
+			out = append(out, y[i])
+		case y[i] == OneName || x[i] == y[i]:
+			out = append(out, x[i])
+		default:
+			return "", false
+		}
+	}
+}
+
+// HeardPart is what a consumer hears of one module's stream: the patterns, local, it is granted.
+type HeardPart struct {
+	Emitter  string
+	Stream   StreamDeclaration
+	Patterns []string
+}
+
+// Hearing is what a consumer's `consumes` hears of the module streams the shelf declares, on one machine
+// (novox/hq ADR 0363 §5): for each stream, its `consumes` narrowed to the stream's events and, for a
+// restricted stream, to the parts the provider's trusted config value grants it. values answers a restricting config
+// value on each machine running the provider. Answered too: whether any `consumes` names a module owning a
+// stream, granted or not, and the `consumes` left for EVENTS — those naming something the provider emits
+// there.
+//
+// Only an entry naming its emitter literally reaches a module's stream: a wildcard emitter, `**` among them,
+// is EVENTS' alone, so nothing that hears every event hears a restricted stream by saying so.
+func Hearing(consumer Manifest, shelf map[string]Manifest, machine string,
+	values func(provider, setting string) []any) (parts []HeardPart, touches bool, onEvents []string) {
+	byStream := map[string]*HeardPart{}
+	var order []string
+	for _, c := range consumer.Consumes {
+		emitter, rest, named := strings.Cut(c, ".")
+		provider, known := shelf[emitter]
+		if !named || emitter == OneName || emitter == TheRest || !known || len(provider.Streams) == 0 {
+			onEvents = append(onEvents, c)
+			continue
+		}
+		hearing, whole := provider.StreamsHearing(rest)
+		if len(hearing) > 0 {
+			touches = true
+		}
+		// Kept for EVENTS while the provider says something there it names: a stream carrying it whole, or
+		// a provider that says nothing of it on EVENTS, leaves it nothing to hear there.
+		if !whole && slices.ContainsFunc(provider.EmitsAll(), func(e string) bool { return patternsOverlap(rest, e) }) {
+			onEvents = append(onEvents, c)
+		}
+		for _, s := range hearing {
+			var granted []string
+			if s.Grant != nil {
+				for _, v := range values(emitter, s.Grant.Config) {
+					for _, key := range GrantedKeys(v, consumer.Module, machine) {
+						granted = append(granted, s.Grant.Granted(key))
+					}
+				}
+			}
+			var got []string
+			for _, p := range s.Events {
+				x, ok := narrow(rest, p)
+				if !ok {
+					continue
+				}
+				if s.Grant == nil {
+					got = append(got, x)
+					continue
+				}
+				for _, g := range granted {
+					if y, ok := narrow(x, g); ok {
+						got = append(got, y)
+					}
+				}
+			}
+			if len(got) == 0 {
+				continue
+			}
+			id := emitter + "/" + s.BusName()
+			if byStream[id] == nil {
+				byStream[id] = &HeardPart{Emitter: emitter, Stream: s}
+				order = append(order, id)
+			}
+			byStream[id].Patterns = append(byStream[id].Patterns, got...)
+		}
+	}
+	for _, id := range order {
+		h := byStream[id]
+		h.Patterns = uncovered(h.Patterns)
+		parts = append(parts, *h)
+	}
+	return parts, touches, onEvents
+}
+
+// uncovered is a list of patterns without those another names whole, and without repeats: the bus refuses
+// a consumer whose filters overlap.
+func uncovered(patterns []string) []string {
+	var out []string
+	for i, p := range patterns {
+		covered := false
+		for j, q := range patterns {
+			if i == j {
+				continue
+			}
+			if within(p, q) && (!within(q, p) || j < i) {
+				covered = true
+				break
+			}
+		}
+		if !covered {
+			out = append(out, p)
+		}
+	}
+	slices.Sort(out)
+	return out
+}

internal/catalogue/streams_test.go

@@ -0,0 +1,206 @@
+package catalogue
+
+import (
+	"encoding/json"
+	"slices"
+	"strings"
+	"testing"
+)
+
+// mailManifest is a module owning a restricted stream, as mesh-mail declares it (novox/hq ADR 0363), with
+// the trusted mergeable file that takes its grants setting; extra is spliced in at the top.
+func mailManifest(t *testing.T, streams, extra string) string {
+	t.Helper()
+	return `{"module":"mesh-mail","version":"1",` + extra + `"streams":` + streams + `,` +
+		`"resources":[{"id":"state","type":"directory","mode":"0700","place":"."},` +
+		`{"id":"grants","trusted":true,"type":"file","path":"${dir:state}/grants.json","mode":"0600",` +
+		`"merge":"json","takes-settings":["grants"],"content":"{\"grants\": {}}"}]}`
+}
+
+const mailStream = `[{"name":"mail","events":["mail.*.message.added","mail.*.message.changed",` +
+	`"mail.*.message.removed"],"max-age-days":7,"grant":{"config":"grants","events":"mail.{key}.**"}}]`
+
+// A module declares a stream of its own (ADR 0363 §1): its events with a variable name (§3), its age and
+// size with defaults, and a grant by a trusted setting of its own (§4).
+func TestAModuleDeclaresAStreamOfItsOwn(t *testing.T) {
+	m, err := ParseManifest([]byte(mailManifest(t, mailStream, "")))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(m.Streams) != 1 {
+		t.Fatalf("streams not read: %+v", m.Streams)
+	}
+	s := m.Streams[0]
+	if s.BusName() != "MAIL" || s.Days() != 7 || s.Cap() != StreamDefaultMiB || s.Grant == nil ||
+		s.Grant.Config != "grants" || s.Grant.Granted("mesh") != "mail.mesh.**" {
+		t.Fatalf("stream read as %+v", s)
+	}
+	if !s.Carries("mail.work.message.added") || s.Carries("mail.work.body") || !s.Carries("mail.*.message.removed") {
+		t.Fatal("the stream's patterns do not say what it carries")
+	}
+	plain, err := ParseManifest([]byte(`{"module":"feed","version":"1","streams":[{"name":"feed-items","events":["item.**"]}]}`))
+	if err != nil {
+		t.Fatalf("a stream without a grant was refused: %v", err)
+	}
+	if got := plain.Streams[0]; got.BusName() != "FEED_ITEMS" || got.Days() != StreamDefaultDays || got.Grant != nil {
+		t.Fatalf("defaults read as %+v", got)
+	}
+}
+
+// Registration refuses a stream the mesh could not hold or would leave ambiguous (ADR 0363 §1, §2, §4).
+func TestAStreamIsNamedBoundedAndGrantedByATrustedConfigValue(t *testing.T) {
+	grants := `"grant":{"config":"grants","events":"mail.{key}.**"}`
+	for _, c := range []struct{ manifest, says string }{
+		{mailManifest(t, `[{"name":"Mail","events":["mail.**"]}]`, ""), `a stream is named locally`},
+		{mailManifest(t, `[{"name":"mail.x","events":["mail.**"]}]`, ""), `a stream is named locally`},
+		{mailManifest(t, `[{"name":"events","events":["mail.**"]}]`, ""), `the mesh's own stream EVENTS`},
+		{mailManifest(t, `[{"name":"log-x","events":["mail.**"]}]`, ""), `starts with LOG_`},
+		{mailManifest(t, `[{"name":"mail","events":["mail.**"]},{"name":"mail","events":["x.**"]}]`, ""), `twice`},
+		{mailManifest(t, `[{"name":"mail","events":[]}]`, ""), `with no events`},
+		{mailManifest(t, `[{"name":"mail","events":["mail.**.x"]}]`, ""), `nothing may follow it`},
+		{mailManifest(t, `[{"name":"mail","events":["mail.**"],"max-age-days":31}]`, ""), `between 1 and 30`},
+		{mailManifest(t, `[{"name":"mail","events":["mail.**"],"max-mib":9000}]`, ""), `between 1 and 8192`},
+		{mailManifest(t, `[{"name":"mail","events":["mail.**"],"stream":"MAIL"}]`, ""), `{name, events`},
+		{mailManifest(t, `[{"name":"mail","events":["mail.**"]}]`, `"emits":["mail.mesh.message.added"],`),
+			`which its stream "mail" carries`},
+		{mailManifest(t, `[{"name":"mail","events":["mail.**"]},{"name":"more","events":["mail.x.*"]}]`, ""),
+			`both carry`},
+		{mailManifest(t, `[{"name":"mail","events":["mail.**"],"grant":{"config":"tags","events":"mail.{key}.**"}}]`, ""),
+			`is not a trusted config value`},
+		{mailManifest(t, `[{"name":"mail","events":["mail.**"],"grant":{"config":"grants","events":"mail.**"}}]`, ""),
+			`exactly once`},
+		{mailManifest(t, `[{"name":"mail","events":["mail.**"],"grant":{"config":"grants","events":"{key}.{key}"}}]`, ""),
+			`exactly once`},
+		{mailManifest(t, `[{"name":"mail","events":["mail.**"],`+grants+`,"why":1}]`, ""), `{name, events`},
+		{`{"module":"a.b","version":"1","streams":[{"name":"x","events":["x"]}]}`, `no dot`},
+	} {
+		_, err := ParseManifest([]byte(c.manifest))
+		if err == nil {
+			t.Errorf("%s was accepted", c.manifest)
+			continue
+		}
+		if !strings.Contains(err.Error(), c.says) {
+			t.Errorf("%s refused for the wrong reason: %v", c.manifest, err)
+		}
+	}
+}
+
+// A stream name has one owner across the catalogue (ADR 0363 §1).
+func TestTwoModulesCannotDeclareOneStream(t *testing.T) {
+	a := Manifest{Module: "a", Streams: []StreamDeclaration{{Name: "mail", Events: []string{"x"}}}}
+	b := Manifest{Module: "b", Streams: []StreamDeclaration{{Name: "mail", Events: []string{"y"}}}}
+	got := strings.Join(CatalogueProblems(Shelf{"a": a, "b": b}), "; ")
+	if !strings.Contains(got, "b declares the stream MAIL, which a already declares") {
+		t.Fatalf("a second owner of MAIL was not refused: %s", got)
+	}
+	if got := StreamNameProblems(Shelf{"a": a}); len(got) != 0 {
+		t.Fatalf("one owner refused: %v", got)
+	}
+}
+
+// The setting's value names who hears each key: a module wherever it runs, or on one machine (ADR 0363 §4).
+func TestARestrictingSettingNamesAModuleOrAModuleOnAMachine(t *testing.T) {
+	var value any
+	if err := json.Unmarshal([]byte(`{"mesh":["reader","person.jochen"],"work":["ace.reader"],"home":["other"],`+
+		`"Bad.Key":["reader"]}`), &value); err != nil {
+		t.Fatal(err)
+	}
+	if got := GrantedKeys(value, "reader", "ace"); !slices.Equal(got, []string{"mesh", "work"}) {
+		t.Fatalf("on ace reader is granted %v", got)
+	}
+	if got := GrantedKeys(value, "reader", "shanks"); !slices.Equal(got, []string{"mesh"}) {
+		t.Fatalf("on shanks reader is granted %v", got)
+	}
+	if got := GrantedKeys(`{"mesh":["reader"]}`, "reader", "ace"); !slices.Equal(got, []string{"mesh"}) {
+		t.Fatalf("a value held as its JSON text grants %v", got)
+	}
+	if got := GrantedKeys([]any{"reader"}, "reader", "ace"); len(got) != 0 {
+		t.Fatalf("a value of another shape granted %v", got)
+	}
+}
+
+// A consumer hears a restricted stream only where its consumes asks and the setting grants it, and only
+// by naming the emitter (ADR 0363 §5); what a stream carries whole leaves EVENTS.
+func TestAConsumerHearsOnlyWhatItAsksAndIsGranted(t *testing.T) {
+	provider, err := ParseManifest([]byte(mailManifest(t, mailStream, `"emits":["account.added"],`)))
+	if err != nil {
+		t.Fatal(err)
+	}
+	shelf := map[string]Manifest{"mesh-mail": provider}
+	var value any
+	_ = json.Unmarshal([]byte(`{"mesh":["reader"],"work":["shanks.reader"]}`), &value)
+	values := func(module, setting string) []any {
+		if module == "mesh-mail" && setting == "grants" {
+			return []any{value}
+		}
+		return nil
+	}
+	reader := Manifest{Module: "reader", Consumes: []string{"mesh-mail.mail.**", "mesh-mail.account.added", "billing.paid"}}
+
+	parts, touches, onEvents := Hearing(reader, shelf, "ace", values)
+	if !touches || len(parts) != 1 || parts[0].Stream.BusName() != "MAIL" {
+		t.Fatalf("heard %+v", parts)
+	}
+	want := []string{"mail.mesh.message.added", "mail.mesh.message.changed", "mail.mesh.message.removed"}
+	if !slices.Equal(parts[0].Patterns, want) {
+		t.Fatalf("on ace reader hears %v, want %v", parts[0].Patterns, want)
+	}
+	if !slices.Equal(onEvents, []string{"mesh-mail.account.added", "billing.paid"}) {
+		t.Fatalf("left on EVENTS: %v", onEvents)
+	}
+	parts, _, _ = Hearing(reader, shelf, "shanks", values)
+	if len(parts) != 1 || len(parts[0].Patterns) != 6 {
+		t.Fatalf("on shanks reader hears %+v", parts)
+	}
+
+	narrow := Manifest{Module: "reader", Consumes: []string{"mesh-mail.mail.*.message.added"}}
+	parts, _, _ = Hearing(narrow, shelf, "ace", values)
+	if len(parts) != 1 || !slices.Equal(parts[0].Patterns, []string{"mail.mesh.message.added"}) {
+		t.Fatalf("a narrower consumes heard %+v", parts)
+	}
+
+	stranger := Manifest{Module: "stranger", Consumes: []string{"mesh-mail.mail.**"}}
+	parts, touches, onEvents = Hearing(stranger, shelf, "ace", values)
+	if len(parts) != 0 || !touches || len(onEvents) != 0 {
+		t.Fatalf("a module not granted heard %+v (touches %v, on EVENTS %v)", parts, touches, onEvents)
+	}
+	everything := Manifest{Module: "audit", Consumes: []string{"**", "*.mail.**"}}
+	parts, touches, _ = Hearing(everything, shelf, "ace", values)
+	if len(parts) != 0 || touches {
+		t.Fatalf("a wildcard emitter reached a module's stream: %+v", parts)
+	}
+	parts, _, _ = Hearing(reader, shelf, "ace", func(string, string) []any { return nil })
+	if len(parts) != 0 {
+		t.Fatalf("with no setting reader heard %+v", parts)
+	}
+}
+
+// A stream without a grant is heard by whoever asks, within its patterns.
+func TestAnUnrestrictedStreamIsHeardByWhoeverAsks(t *testing.T) {
+	feed := Manifest{Module: "feed", Streams: []StreamDeclaration{{Name: "items", Events: []string{"item.*.added", "item.*.gone"}}}}
+	reader := Manifest{Module: "reader", Consumes: []string{"feed.item.**", "feed.item.x.added"}}
+	parts, touches, onEvents := Hearing(reader, map[string]Manifest{"feed": feed}, "ace", nil)
+	if !touches || len(onEvents) != 0 || len(parts) != 1 ||
+		!slices.Equal(parts[0].Patterns, []string{"item.*.added", "item.*.gone"}) {
+		t.Fatalf("heard %+v, on EVENTS %v", parts, onEvents)
+	}
+}
+
+func TestNarrowingTwoPatterns(t *testing.T) {
+	for _, c := range []struct{ a, b, want string }{
+		{"mail.**", "mail.*.message.added", "mail.*.message.added"},
+		{"mail.*.message.added", "mail.mesh.**", "mail.mesh.message.added"},
+		{"a.*", "*.b", "a.b"},
+		{"**", "a.**", "a.**"},
+		{"a.b", "a.c", ""},
+		{"a", "a.**", ""},
+	} {
+		got, ok := narrow(c.a, c.b)
+		if (c.want == "") == ok || got != c.want {
+			t.Errorf("narrow(%s, %s) = %q, %v; want %q", c.a, c.b, got, ok, c.want)
+		}
+	}
+	if got := uncovered([]string{"a.**", "a.b", "a.b", "c"}); !slices.Equal(got, []string{"a.**", "c"}) {
+		t.Fatalf("uncovered: %v", got)
+	}
+}

internal/inventory/busrecords.go

@@ -61,12 +61,23 @@ func (i *Inventory) BusRecords(ctx context.Context) (broker.Records, error) {
 
 	out := broker.Records{Assigned: map[string][]broker.Declared{}, People: map[string][]string{},
 		Interchangeable: map[string]bool{}}
+	assigned := map[string][]string{}
 	for _, n := range nodes {
-		out.Nodes = append(out.Nodes, n.Name)
 		modules, err := i.Assigned(ctx, n.Name)
 		if err != nil {
 			return broker.Records{}, fmt.Errorf("cannot read what %s runs: %w", n.Name, err)
 		}
+		assigned[n.Name] = modules
+	}
+	// What each restricting config value says on each machine running its module (novox/hq ADR 0363 §5): read
+	// once, for every consumer composed below.
+	grants, err := i.streamGrantValues(ctx, declared, nodes, assigned)
+	if err != nil {
+		return broker.Records{}, err
+	}
+	for _, n := range nodes {
+		out.Nodes = append(out.Nodes, n.Name)
+		modules := assigned[n.Name]
 		for _, module := range modules {
 			m, known := declared[module]
 			if !known {
@@ -89,6 +100,9 @@ func (i *Inventory) BusRecords(ctx context.Context) (broker.Records, error) {
 				d.KeyedReads = append(d.KeyedReads, broker.KeyedRead{Read: sr.Read, Key: sr.Key})
 			}
 			d.Holds = heldHere(d.Holds, holdings, n.Name, module)
+			// And the module streams it hears on this machine, as granted (novox/hq ADR 0363 §5): what a
+			// stream carries whole leaves EVENTS' consumes.
+			d.Hears, d.HearsAStream, d.Consumes = hearsOf(m, declared, n.Name, grants, d.Consumes)
 			out.Assigned[n.Name] = append(out.Assigned[n.Name], d)
 			if m.Instances == catalogue.InstancesInterchangeable {
 				out.Interchangeable[m.Module] = true
@@ -169,6 +183,8 @@ func declaredFor(m catalogue.Manifest, seats map[string]catalogue.SeatDeclaratio
 		Reads: m.Reads,
 		// And the logs it keeps (novox/hq ADR 0297).
 		Logs: logsOf(m),
+		// And the streams it owns (novox/hq ADR 0363).
+		Streams: streamsOf(m),
 		// And the tools its health asks (novox/hq ADR 0240): the machine's node-engine is granted them.
 		Checks: catalogue.HealthChecks(m),
 		// And whether it runs as an account of its own (novox/hq ADR 0259 §8).
@@ -362,3 +378,101 @@ func (i *Inventory) DeclaredTrafficSeats(ctx context.Context) ([]broker.Seat, er
 	}
 	return out, nil
 }
+
+// streamsOf is the streams a module owns, as the bus holds them (novox/hq ADR 0363).
+func streamsOf(m catalogue.Manifest) []broker.ModuleStream {
+	var out []broker.ModuleStream
+	for _, s := range m.Streams {
+		events := make([]string, 0, len(s.Events))
+		for _, e := range s.Events {
+			events = append(events, broker.WirePattern(e))
+		}
+		out = append(out, broker.ModuleStream{Module: m.Module, Name: s.BusName(), Events: events,
+			MaxAgeDays: s.Days(), MaxMiB: s.Cap(), Restricted: s.Grant != nil})
+	}
+	return out
+}
+
+// DeclaredStreams is every module stream the catalogue declares, registered modules assigned or not: a
+// stream exists from registration, as a log does (novox/hq ADR 0363 §1).
+func (i *Inventory) DeclaredStreams(ctx context.Context) ([]broker.ModuleStream, error) {
+	declared, err := i.Catalogue(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("cannot read the catalogue: %w", err)
+	}
+	var out []broker.ModuleStream
+	for _, m := range declared {
+		out = append(out, streamsOf(m)...)
+	}
+	return out, nil
+}
+
+// grantValues is each restricting config value on each machine running its module, by `<module>/<key>`.
+type grantValues map[string][]any
+
+// streamGrantValues reads, for every module owning a restricted stream, the config value that restricts it on every
+// machine the module runs on: the mesh's layer under that machine's, the machine's winning (novox/hq ADR 0363 §5).
+func (i *Inventory) streamGrantValues(ctx context.Context, declared map[string]catalogue.Manifest,
+	nodes []Node, assigned map[string][]string) (grantValues, error) {
+	out := grantValues{}
+	for _, n := range nodes {
+		for _, module := range assigned[n.Name] {
+			m, known := declared[module]
+			if !known {
+				continue
+			}
+			var settings []string
+			for _, s := range m.Streams {
+				if s.Grant != nil && !slices.Contains(settings, s.Grant.Config) {
+					settings = append(settings, s.Grant.Config)
+				}
+			}
+			if len(settings) == 0 {
+				continue
+			}
+			layers, err := i.SettingsFor(ctx, n.Name, module)
+			if err != nil {
+				return nil, fmt.Errorf("cannot read %s's settings on %s, which grant who hears its stream: %w",
+					module, n.Name, err)
+			}
+			for _, setting := range settings {
+				if v, set := settingValue(layers, setting); set {
+					out[module+"/"+setting] = append(out[module+"/"+setting], v)
+				}
+			}
+		}
+	}
+	return out, nil
+}
+
+// settingValue is one key's value over a machine's layers, the last that sets it winning. Pure.
+func settingValue(layers []catalogue.Layer, key string) (any, bool) {
+	var value any
+	set := false
+	for _, l := range layers {
+		if v, ok := l.Values[key]; ok {
+			value, set = v, true
+		}
+	}
+	return value, set
+}
+
+// hearsOf is what a module hears of the module streams on one machine, and the consumes left for EVENTS
+// (novox/hq ADR 0363 §5). Pure.
+func hearsOf(m catalogue.Manifest, declared map[string]catalogue.Manifest, node string, grants grantValues,
+	consumes []string) ([]broker.Heard, bool, []string) {
+	reading := m
+	reading.Consumes = consumes
+	parts, touches, onEvents := catalogue.Hearing(reading, declared, node, func(provider, setting string) []any {
+		return grants[provider+"/"+setting]
+	})
+	var out []broker.Heard
+	for _, p := range parts {
+		h := broker.Heard{Stream: p.Stream.BusName(), Emitter: p.Emitter}
+		for _, pattern := range p.Patterns {
+			h.Filters = append(h.Filters, broker.ModuleStreamSubject(p.Emitter, broker.WirePattern(pattern)))
+		}
+		out = append(out, h)
+	}
+	return out, touches, onEvents
+}

internal/inventory/streams_test.go

@@ -0,0 +1,45 @@
+package inventory
+
+import (
+	"encoding/json"
+	"slices"
+	"testing"
+
+	"github.com/novox/mesh-controller/internal/catalogue"
+)
+
+// A machine's layer wins over the mesh's for the setting that restricts a stream (novox/hq ADR 0363 §5).
+func TestARestrictingSettingIsReadMachineOverMesh(t *testing.T) {
+	layers := []catalogue.Layer{
+		{From: catalogue.MeshWideLayer, Values: map[string]any{"grants": "mesh", "other": 1}},
+		{From: "novox", Values: map[string]any{"grants": "novox"}},
+	}
+	if v, set := settingValue(layers, "grants"); !set || v != "novox" {
+		t.Fatalf("read %v, %v", v, set)
+	}
+	if _, set := settingValue(layers[:1], "absent"); set {
+		t.Fatal("a key no layer sets read as set")
+	}
+}
+
+// What a consumer hears is composed into the bus's subjects, and what a stream carries leaves EVENTS.
+func TestWhatAModuleHearsIsComposedOntoTheStreamsSubjects(t *testing.T) {
+	provider := catalogue.Manifest{Module: "mesh-mail", Streams: []catalogue.StreamDeclaration{{Name: "mail",
+		Events: []string{"mail.*.message.added"}, Grant: &catalogue.StreamGrant{Config: "grants", Events: "mail.{key}.**"}}}}
+	var value any
+	_ = json.Unmarshal([]byte(`{"mesh":["reader"]}`), &value)
+	grants := grantValues{"mesh-mail/grants": {value}}
+	reader := catalogue.Manifest{Module: "reader"}
+	hears, touches, onEvents := hearsOf(reader, map[string]catalogue.Manifest{"mesh-mail": provider}, "ace", grants,
+		[]string{"mesh-mail.mail.**", "billing.paid"})
+	if !touches || len(hears) != 1 || hears[0].Stream != "MAIL" ||
+		!slices.Equal(hears[0].Filters, []string{"mesh.stream.mesh-mail.event.mail.mesh.message.added"}) {
+		t.Fatalf("heard %+v", hears)
+	}
+	if !slices.Equal(onEvents, []string{"billing.paid"}) {
+		t.Fatalf("left on EVENTS: %v", onEvents)
+	}
+	if got := streamsOf(provider); len(got) != 1 || got[0].Name != "MAIL" || got[0].MaxAgeDays != 7 || !got[0].Restricted {
+		t.Fatalf("the provider's stream: %+v", got)
+	}
+}

internal/link/deadletters.go

@@ -259,7 +259,8 @@ func DeadLetterNamed(js nats.JetStreamContext, id uint64) (DeadLetter, error) {
 // work queue has one worker per subject, so only the worker that gave it up takes it, and DeliverAgain
 // removes the original from the queue first, so there are never two (novox/hq issue 334); only while the
 // worker that gave it up is on the bus. Any other stream's
-// message is refused, with why: an ask to a seat the controller does not ask would need a publish it is
+// message is refused, with why — but a module's own stream's, which goes back as an event while its consumer is
+// still granted it (novox/hq ADR 0363 §7): an ask to a seat the controller does not ask would need a publish it is
 // not granted, in its asker's name (ADR 0264's consequences, ADR 0259 §3), and a stream that is neither
 // would reach every consumer of its subject.
 func AgainTo(js nats.JetStreamContext, d DeadLetter) (string, error) {
@@ -280,6 +281,29 @@ func AgainTo(js nats.JetStreamContext, d DeadLetter) (string, error) {
 			return "", fmt.Errorf("whether %s can receive dead letter %d cannot be read: %w", d.Who, d.ID, err)
 		}
 		return d.Subject, nil
+	case d.Stream != broker.EventsStream && broker.OnAModuleStream(d.Subject):
+		// **A module stream's event goes back to its consumer only while it is still granted** (novox/hq ADR
+		// 0363 §7): its consumer on that stream must still be there and still filter the subject, or delivering
+		// it again would hand a module what its grant no longer names. Then it is delivered as an event is, on
+		// EVENTS under the consumer's own again-subject, which the module's consumer there filters.
+		info, err := js.ConsumerInfo(d.Stream, d.Consumer)
+		if errors.Is(err, nats.ErrConsumerNotFound) {
+			return "", fmt.Errorf("dead letter %d was given up by %s, which no longer hears %s, so it would reach a "+
+				"module no longer granted it. Drop it", d.ID, d.Who, d.Stream)
+		}
+		if err != nil {
+			return "", fmt.Errorf("whether %s still hears %s cannot be read: %w", d.Who, d.Stream, err)
+		}
+		granted := false
+		for _, f := range append([]string{info.Config.FilterSubject}, info.Config.FilterSubjects...) {
+			if f != "" && broker.SubjectsOverlap(f, d.Subject) {
+				granted = true
+			}
+		}
+		if !granted {
+			return "", fmt.Errorf("dead letter %d is on %s, which %s is no longer granted. Drop it", d.ID, d.Subject, d.Who)
+		}
+		d.Stream = broker.EventsStream
 	case d.Stream != broker.EventsStream:
 		return "", fmt.Errorf("dead letter %d is from %s, and only an event or an ask the controller made is "+
 			"delivered again: publishing it again would reach every consumer of its subject, or need a grant the "+

internal/link/deadletters_module_stream_test.go

@@ -0,0 +1,48 @@
+package link
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/novox/mesh-controller/internal/broker"
+)
+
+// A message given up on a module's own stream is delivered again on EVENTS to the consumer that gave it up,
+// and only while that consumer is still granted it (novox/hq ADR 0363 §7).
+func TestAModuleStreamsDeadLetterGoesBackOnlyWhileGranted(t *testing.T) {
+	js := aBus(t)
+	mail := broker.ModuleStream{Module: "mesh-mail", Name: "MAIL", Events: []string{"mail.*.message.added"},
+		MaxAgeDays: 7, MaxMiB: 16, Restricted: true}
+	if err := js.EnsureStream(mail.Stream()); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = js.Context().DeleteStream("MAIL") })
+	subject := broker.ModuleStreamSubject("mesh-mail", "mail.mesh.message.added")
+	heard := broker.Heard{Stream: "MAIL", Emitter: "mesh-mail", Filters: []string{subject}}
+	reader := broker.Principal{Kind: broker.KindModule, Node: "ace", Module: "reader", Hears: []broker.Heard{heard},
+		PasswordHash: "x"}
+	users := broker.ConsumersOf([]broker.Principal{reader})
+	if len(users) != 2 {
+		t.Fatalf("the reader's consumers: %+v", users)
+	}
+	for _, c := range users {
+		if err := js.EnsureConsumer(c.Consumer); err != nil {
+			t.Fatal(err)
+		}
+	}
+	d := DeadLetter{ID: 9, Stream: "MAIL", Consumer: "ace_reader", Subject: subject}
+	got, err := AgainTo(js.Context(), d)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got != broker.AgainSubject("ace_reader", subject) {
+		t.Fatalf("delivered again on %s", got)
+	}
+
+	if err := js.RemoveConsumer("MAIL", "ace_reader"); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := AgainTo(js.Context(), d); err == nil || !strings.Contains(err.Error(), "no longer hears MAIL") {
+		t.Fatalf("delivered again to a consumer no longer granted: %v", err)
+	}
+}