The grants part adds a manifest's new events to a held stream and its consumers' filters, only ever adding (issue 1921)

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

modules/mesh-grants/cmd/mesh-grants/create.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"slices"
 
 	"github.com/nats-io/nats.go/jetstream"
 
@@ -67,3 +68,45 @@ func (c creating) CreateDeclared(cfg jetstream.StreamConfig) error {
 	}
 	return err
 }
+
+// AddSubjects adds subjects to a held stream and changes nothing else: it reads the stream's configuration, sets its
+// subjects to the union and updates with the rest as it was.
+func (c creating) AddSubjects(stream string, add []string) error {
+	ctx, cancel := context.WithTimeout(c.ctx, requestWithin)
+	defer cancel()
+	s, err := c.js.Stream(ctx, stream)
+	if err != nil {
+		return err
+	}
+	cfg := s.CachedInfo().Config
+	for _, a := range add {
+		if !slices.Contains(cfg.Subjects, a) {
+			cfg.Subjects = append(cfg.Subjects, a)
+		}
+	}
+	_, err = c.js.UpdateStream(ctx, cfg)
+	return err
+}
+
+// AddFilters adds filter subjects to a held consumer and changes nothing else.
+func (c creating) AddFilters(stream, name string, add []string) error {
+	ctx, cancel := context.WithTimeout(c.ctx, requestWithin)
+	defer cancel()
+	held, err := c.js.Consumer(ctx, stream, name)
+	if err != nil {
+		return err
+	}
+	cfg := held.CachedInfo().Config
+	filters := slices.Clone(cfg.FilterSubjects)
+	if cfg.FilterSubject != "" {
+		filters = append(filters, cfg.FilterSubject)
+	}
+	for _, a := range add {
+		if !slices.Contains(filters, a) {
+			filters = append(filters, a)
+		}
+	}
+	cfg.FilterSubject, cfg.FilterSubjects = "", filters
+	_, err = c.js.UpdateConsumer(ctx, stream, cfg)
+	return err
+}

modules/mesh-grants/grants_test.go

@@ -52,9 +52,11 @@ type fakeBus struct {
 	// hold, when set, keeps every nats_users call waiting until it is closed: a pass waiting on the bus.
 	hold chan struct{}
 	// creates are the streams the part asked to create, refused when refuseCreate.
-	creates      []string
-	createdAge   map[string]time.Duration
-	refuseCreate bool
+	creates []string
+	// subjectAdds and filterAdds are the additions the part asked of a held stream and a held consumer.
+	subjectAdds, filterAdds []string
+	createdAge              map[string]time.Duration
+	refuseCreate            bool
 	// refuseUsers refuses the bus module's nats_users: the loaded list cannot be read.
 	refuseUsers bool
 	// rootFree answers a machine's engine's root-free verb: nil is no answerer at all (no responders).
@@ -102,6 +104,41 @@ func (b *fakeBus) CreateConsumer(stream string, cfg jetstream.ConsumerConfig) er
 	return nil
 }
 
+// AddSubjects is the bus's addition of subjects to a held stream: the union, every other setting kept.
+func (b *fakeBus) AddSubjects(stream string, add []string) error {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	b.subjectAdds = append(b.subjectAdds, stream+" +"+strings.Join(add, ","))
+	for _, s := range b.streams {
+		if s["name"] == stream {
+			have, _ := s["subjects"].([]string)
+			for _, a := range add {
+				if !slices.Contains(have, a) {
+					have = append(have, a)
+				}
+			}
+			s["subjects"] = have
+			return nil
+		}
+	}
+	return errors.New("stream not found")
+}
+
+// AddFilters is the bus's addition of filters to a held consumer: the union, max deliver and the rest kept.
+func (b *fakeBus) AddFilters(stream, name string, add []string) error {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	b.filterAdds = append(b.filterAdds, stream+"."+name+" +"+strings.Join(add, ","))
+	for _, c := range b.consumers[stream] {
+		if c["name"] == name {
+			have, _ := c["filters"].([]string)
+			c["filters"] = append(slices.Clone(have), add...)
+			return nil
+		}
+	}
+	return errors.New("consumer not found")
+}
+
 func (b *fakeBus) PublishMembership(node, module string, body []byte) error {
 	b.mu.Lock()
 	defer b.mu.Unlock()
@@ -474,7 +511,7 @@ func TestGrants(t *testing.T) {
 
 	t.Run("no timer, no update or delete, a publish only of conditions, a create only of declared streams", func(t *testing.T) {
 		forbidden := []string{"time.Tick", "time.NewTicker", "time.Sleep", "time.After", "time.NewTimer", "time.AfterFunc",
-			"PublishMsg", "CreateKeyValue", "CreateOrUpdate", "UpdateStream", "UpdateKeyValue", "DeleteStream", "PurgeStream",
+			"PublishMsg", "CreateKeyValue", "CreateOrUpdate", "UpdateKeyValue", "DeleteStream", "PurgeStream",
 			"DeleteKeyValue", "CreateObjectStore", "EnsureStream", "RaiseBuckets", "RaiseLogs", ".Create("}
 		// The one publish: a condition, on the seat operator-channel's subject, in asking.go; a membership is written to
 		// the part's state alone, never said on its subject (issue 1126). The one create: a declared stream the bus lacks,
@@ -509,6 +546,10 @@ func TestGrants(t *testing.T) {
 				if n := strings.Count(string(src), "CreateStream"); (path == createOnly && n != 1) || (path != createOnly && n != 0) {
 					t.Errorf("%s holds CreateStream %d times: only create.go creates, once", path, n)
 				}
+				// The one update of a stream: adding subjects to a held one, in create.go (the operator, 2026-11-05).
+				if n := strings.Count(string(src), "UpdateStream("); (path == createOnly && n != 1) || (path != createOnly && n != 0) {
+					t.Errorf("%s holds UpdateStream %d times: only create.go updates a stream, once, to add subjects", path, n)
+				}
 				// The one object bucket: the part's own, of the user list sealed to the bus node, in sealed.go (ADR 0547 clause 7).
 				if path == sealedOnly {
 					if strings.Count(string(src), "CreateObjectStore(") != 1 || !strings.Contains(string(src), "delivery.GrantsUserListObjects") {
@@ -1830,6 +1871,53 @@ func TestGrants(t *testing.T) {
 
 	// 774 (ADR 0563 §2, §3): the node-engine's manifest (mesh-host) is placed on every node though no node is assigned it,
 	// so its declared state acts is created as a placed module's, and composed for the user list's create right.
+	t.Run("assertions: a held stream gains the subjects its manifest names, and loses, changes nothing else", func(t *testing.T) {
+		b := fixture()
+		b.loaded = shadowOf(t, b)
+		b.streams = []map[string]any{{"name": "THINGS", "subjects": []string{"dropped.subject"}, "retention": "limits", "max_age_hours": 48}}
+		p := partOn(b)
+		p.Create = b
+		var logged []string
+		p.Logf = func(f string, a ...any) { logged = append(logged, fmt.Sprintf(f, a...)) }
+		res := p.Compose()
+		if len(b.subjectAdds) != 1 || !strings.HasPrefix(b.subjectAdds[0], "THINGS +") || strings.Contains(b.subjectAdds[0], "dropped") ||
+			!slices.Contains(b.creates, "LOG_mod-x_history") || slices.Contains(b.creates, "THINGS") || !slices.Contains(b.events, grants.EventAsserted) ||
+			!strings.Contains(strings.Join(res.Asserted, " "), "gained") || !strings.Contains(strings.Join(logged, " "), "THINGS gained") {
+			t.Fatalf("the pass added %v, created %v, said %v, journal %v", b.subjectAdds, b.creates, res.Asserted, logged)
+		}
+		for _, s := range b.streams {
+			if s["name"] == "THINGS" {
+				if got := s["subjects"].([]string); !slices.Contains(got, "dropped.subject") || len(got) < 2 || s["max_age_hours"] != 48 {
+					t.Errorf("THINGS after the addition = %v: a subject the manifest dropped must stay, the limits as they were", s)
+				}
+			}
+		}
+		if again := p.Compose(); len(b.subjectAdds) != 1 || len(again.Asserted) != 0 {
+			t.Errorf("a stream that holds every subject was added to again: %v, %v", b.subjectAdds, again.Asserted)
+		}
+	})
+
+	t.Run("account consumers: a held consumer gains the filters of a new event, never loses one", func(t *testing.T) {
+		const cManifest = `{"module":"mod-c","version":"1","own-secrets":{"broker":"/var/lib/c/broker"},"consumes":["mod-y.done","mod-y.more"]}`
+		b := fixture()
+		b.manifestOf("mod-c", cManifest)
+		b.assign("b", "mod-c")
+		b.loaded = shadowOf(t, b)
+		p := partOn(b)
+		p.Consumers = b
+		b.consumers = map[string][]map[string]any{"EVENTS": {{"name": "b_mod-c", "filters": []string{"old.filter", "mesh.mod.mod-y.event.done"}, "max_deliver": 9}}}
+		res := p.Compose()
+		if len(b.filterAdds) != 1 || !strings.Contains(b.filterAdds[0], "mesh.mod.mod-y.event.more") || strings.Contains(b.filterAdds[0], "done") ||
+			!slices.Contains(b.consumers["EVENTS"][0]["filters"].([]string), "old.filter") || b.consumers["EVENTS"][0]["max_deliver"] != 9 ||
+			len(res.Consumers.Matched) != 1 {
+			t.Fatalf("filters added %v, now %v, said %+v", b.filterAdds, b.consumers["EVENTS"], res.Consumers)
+		}
+		p.Compose()
+		if len(b.filterAdds) != 1 {
+			t.Errorf("a consumer that holds every filter was added to again: %v", b.filterAdds)
+		}
+	})
+
 	t.Run("assertions: the node-engine's declared state is created though no node is assigned it", func(t *testing.T) {
 		b := fixture()
 		b.manifestOf(buscompose.EngineModule, `{"module":"mesh-host","version":"1","state":[{"name":"acts"}]}`)

modules/mesh-grants/internal/buscompose/create.go

@@ -9,7 +9,7 @@ import (
 // The streams a manifest declares, as the server is asked to create them (novox/hq ADR 0547 §5, issue 1790): the
 // grants part creates each one the bus does not hold yet, with what the controller asserts at its start, so the two
 // make the same stream. Pure. Only a create is ever asked with these: a stream the bus holds is never updated by the
-// grants part (CreateStreamGrants grants no update).
+// grants part (CreateStreamGrants grants its update only to add subjects).
 
 // BucketStreamConfig is a module's bucket as the stream the controller's CreateOrUpdateKeyValue makes of its options
 // (mesh-controller internal/broker BucketConfig, with no marker age held): the client library's key-value shape

modules/mesh-grants/internal/buscompose/creates_streams_test.go

@@ -65,7 +65,8 @@ func theGrantsPartCreatesOnlyDeclaredStreams(t *testing.T) {
 		{GrantsModule, "$JS.API.STREAM.CREATE.LOG_mesh-delivery-builder_builds", true},
 		{GrantsModule, "$JS.API.STREAM.CREATE.BUILDASKS", true},
 		{GrantsModule, "$JS.API.STREAM.CREATE.KV_mesh-grants_assertions", true},
-		{GrantsModule, "$JS.API.STREAM.UPDATE.KV_mesh-delivery-builder_steps", false},
+		{GrantsModule, "$JS.API.STREAM.UPDATE.KV_mesh-delivery-builder_steps", true},
+		{GrantsModule, "$JS.API.STREAM.UPDATE.KV_mesh-delivery-builder_undeclared", false},
 		{GrantsModule, "$JS.API.STREAM.DELETE.KV_mesh-delivery-builder_steps", false},
 		{GrantsModule, "$JS.API.STREAM.PURGE.KV_mesh-delivery-builder_steps", false},
 		{GrantsModule, "$JS.API.STREAM.CREATE.KV_mesh-delivery-builder_undeclared", false},
@@ -132,7 +133,7 @@ func theGrantsPartCreatesOnlyDeclaredStreams(t *testing.T) {
 		}
 	}
 	// A name that is no plain token is never granted, and a module other than the grants part is never granted one.
-	if got := CreateStreamGrants([]string{"*", ">", "KV_a.b", "KV_a_b"}); len(got) != 2 {
+	if got := CreateStreamGrants([]string{"*", ">", "KV_a.b", "KV_a_b"}); len(got) != 3 {
 		t.Errorf("grants for wildcard names: %v", got)
 	}
 	other, err := PermissionsFor(Principal{Kind: KindModule, Node: "novox", Module: "shop", CreatesStreams: []string{"KV_shop_x"}})
@@ -142,12 +143,15 @@ func theGrantsPartCreatesOnlyDeclaredStreams(t *testing.T) {
 	if MayPublish(other, "$JS.API.STREAM.CREATE.KV_shop_x") {
 		t.Error("a module other than the grants part was granted a create")
 	}
-	// The writers table refuses the grants part an update, and a create of a stream it was not given.
+	// The writers table refuses the grants part an update of a stream not declared, a delete, and a create of a stream it was not given.
 	g := Principal{Kind: KindModule, Node: "novox", Module: GrantsModule, CreatesStreams: []string{"KV_a_b"}}
 	if err := CheckWriters(g, []string{"$JS.API.STREAM.CREATE.KV_a_b"}); err != nil {
 		t.Errorf("the declared create refused: %v", err)
 	}
-	for _, refused := range []string{"$JS.API.STREAM.UPDATE.KV_a_b", "$JS.API.STREAM.CREATE.KV_a_c", "$JS.API.STREAM.CREATE.*"} {
+	if err := CheckWriters(g, []string{"$JS.API.STREAM.UPDATE.KV_a_b"}); err != nil { // the update that adds subjects
+		t.Errorf("the declared update refused: %v", err)
+	}
+	for _, refused := range []string{"$JS.API.STREAM.UPDATE.KV_a_c", "$JS.API.STREAM.UPDATE.*", "$JS.API.STREAM.DELETE.KV_a_b", "$JS.API.STREAM.CREATE.KV_a_c", "$JS.API.STREAM.CREATE.*"} {
 		if CheckWriters(g, []string{refused}) == nil {
 			t.Errorf("%s was not refused", refused)
 		}

modules/mesh-grants/internal/buscompose/nats.go

@@ -1444,14 +1444,16 @@ const WatchmanModule = "mesh-delivery-watchman"
 
 // CreateStreamGrants is what creating each named stream needs, and nothing wider (novox/hq ADR 0547 §5; the operator,
 // 2026-10-26 08:54): its create, and its information to read back what it holds. Each by its own name: a name that
-// is not one plain subject token is left out, since a wildcard there would be every stream.
+// is not one plain subject token is left out, since a wildcard there would be every stream. Its update is granted for the one
+// act the part does on a held stream: adding the subjects a manifest's new events need (grants.Creator.AddSubjects); it
+// never removes a subject, changes a limit or recreates the stream, and never deletes or purges one.
 func CreateStreamGrants(streams []string) []string {
 	var out []string
 	for _, s := range streams {
 		if !safeSubject.MatchString(s) {
 			continue
 		}
-		out = append(out, "$JS.API.STREAM.CREATE."+s, "$JS.API.STREAM.INFO."+s)
+		out = append(out, "$JS.API.STREAM.CREATE."+s, "$JS.API.STREAM.INFO."+s, "$JS.API.STREAM.UPDATE."+s)
 	}
 	return out
 }

modules/mesh-grants/internal/buscompose/writers.go

@@ -36,14 +36,19 @@ type WriterRow struct {
 
 // createsDeclared, and the other writers' tests, are who a row's writer is as a principal.
 //
-// createsDeclared is the grants part creating one stream a manifest declares, by its own name: the create alone, never an
-// update or a delete (novox/hq ADR 0547 §5; the operator, 2026-10-26 08:54); or the bus module's tools deleting one of
+// createsDeclared is the grants part creating one stream a manifest declares, by its own name: the create, and the update
+// that only adds subjects, never a delete (novox/hq ADR 0547 §5; the operator, 2026-10-26 08:54, 2026-11-05); or the bus module's tools deleting one of
 // the ended controller's leftover streams (ADR 0590 clause 6).
 func createsDeclared(p Principal, pattern string) bool {
 	if removesLeftover(p, pattern) {
 		return true
 	}
 	name, ok := strings.CutPrefix(pattern, "$JS.API.STREAM.CREATE.")
+	if upd, isUpdate := strings.CutPrefix(pattern, "$JS.API.STREAM.UPDATE."); isUpdate {
+		// The update of a stream a manifest declares, by its name, to add the subjects a new event needs (the operator,
+		// 2026-11-05); the part's process adds and never removes a subject or changes a limit.
+		return p.Kind == KindModule && p.Module == GrantsModule && slices.Contains(p.CreatesStreams, upd) && safeSubject.MatchString(upd)
+	}
 	if ok && p.Kind == KindModule && p.Module == GrantsModule && name == GrantsUserListObjectStream {
 		return true
 	}
@@ -151,7 +156,7 @@ var WritersTable = []WriterRow{
 		Subjects: []string{"$JS.API.STREAM.CREATE.>", "$JS.API.STREAM.UPDATE.>", "$JS.API.STREAM.DELETE.>",
 			"$JS.API.CONSUMER.DURABLE.CREATE.>"},
 		Writes: createsDeclared,
-		Shared: "the grants part creates a stream a manifest declares, by its name, and never updates or deletes one (novox/hq ADR 0547 §5; the operator, 2026-10-26 08:54); the bus module's tools account deletes the ended controller's leftover streams, each by its name, and no other (novox/hq ADR 0590 clause 6)"},
+		Shared: "the grants part creates a stream a manifest declares, by its name, adds to a held one the subjects its manifest names, and never removes a subject or deletes a stream (novox/hq ADR 0547 §5; the operator, 2026-10-26 08:54, 2026-11-05); the bus module's tools account deletes the ended controller's leftover streams, each by its name, and no other (novox/hq ADR 0590 clause 6)"},
 	{State: "builds and their outcomes", Writer: "the build seat's holder", KeptIn: "its own state",
 		Others:   "the builder asks",
 		Subjects: []string{"mesh.seat.mesh-build-machine.event.built"},

modules/mesh-grants/internal/grants/assertions.go

@@ -61,6 +61,9 @@ const CreatesMissing = "a bucket, log or stream the manifest declares and the bu
 // 08:54). A stream the bus holds already answers ErrHeld.
 type Creator interface {
 	CreateDeclared(cfg jetstream.StreamConfig) error
+	// AddSubjects adds subjects to a stream the bus holds and nothing else: the stream's subjects become the union of what
+	// it holds and add; no subject is removed, no limit, retention or other setting changed, the stream not recreated.
+	AddSubjects(stream string, add []string) error
 }
 
 // ErrHeld is a create refused because the bus holds the stream already (made meanwhile): left as it is.
@@ -149,6 +152,29 @@ func createMissing(c Creator, done *creations, a *Assertion, m catalogue.Manifes
 			it.Created = ItemNotCreated + err.Error() + " (tried again at the next event)"
 		}
 	}
+	// A stream the bus holds whose manifest names events it has no subject for gains those subjects, and only ever those: a
+	// subject the manifest dropped stays, limits and retention are left as they are (the operator, 2026-11-05).
+	for i := range a.Items {
+		it := &a.Items[i]
+		if it.Held == nil || it.Kind != "stream" {
+			continue
+		}
+		var add []string
+		for _, s := range it.Composed.Subjects {
+			if !slices.Contains(it.Held.Subjects, s) {
+				add = append(add, s)
+			}
+		}
+		if len(add) == 0 {
+			continue
+		}
+		if err := c.AddSubjects(it.Stream, add); err != nil {
+			it.Created = ItemNotCreated + "subjects " + strings.Join(add, ", ") + " not added: " + err.Error() + " (tried again at the next event)"
+			continue
+		}
+		it.Created = "subjects added: " + strings.Join(add, ", ")
+		made = append(made, "stream "+it.Stream+" gained "+strings.Join(add, ", "))
+	}
 	if len(made) > 0 {
 		a.Asserted = true
 	}

modules/mesh-grants/internal/grants/consumers.go

@@ -26,6 +26,9 @@ type ConsumerCreator interface {
 	// MatchConsumer brings a held consumer to cfg, keeping where it delivers from: only on a person's ask
 	// (grants_consumer_match), never by a pass.
 	MatchConsumer(stream string, cfg jetstream.ConsumerConfig) error
+	// AddFilters adds filter subjects to a held consumer and nothing else: its filters become the union of what it holds and
+	// add, every other setting and its delivery position kept.
+	AddFilters(stream, name string, add []string) error
 }
 
 // ConsumersState is what one pass found and did of the account consumers.
@@ -85,6 +88,21 @@ func heldConsumers(c Caller, stream string) (map[string]heldConsumer, error) {
 	return out, nil
 }
 
+// filtersToAdd are the composed filters of a consumer the bus's one lacks, in the composed order; none when the consumer
+// filters the whole stream (no filters) or holds all.
+func filtersToAdd(c buscompose.Consumer, h heldConsumer) []string {
+	if len(h.Filters) == 0 {
+		return nil
+	}
+	var add []string
+	for _, f := range c.Filters {
+		if !slices.Contains(h.Filters, f) && !slices.Contains(add, f) {
+			add = append(add, f)
+		}
+	}
+	return add
+}
+
 // ConsumersOn is why the part creates and matches the account consumers: the controller ended by its unassignment, so
 // the part is their only maker (novox/hq ADR 0590, issue 1126).
 const ConsumersOn = "the grants part makes and matches the account consumers: the controller ended by its unassignment (issue 1126)"
@@ -124,6 +142,20 @@ func (p *Part) consumers(res *Result, recs buscompose.Records, match string) {
 					}
 					seen[key] = true
 				}
+				// A consumer whose module now consumes more events of the stream gains those filters by itself, only ever added:
+				// a filter it holds that is no longer composed stays, and nothing else about it changes.
+				if add := filtersToAdd(c, h); len(add) > 0 && st.On && p.Consumers != nil {
+					if err := p.Consumers.AddFilters(stream, c.Name, add); err != nil {
+						st.Matched = append(st.Matched, key+": filters "+strings.Join(add, ", ")+" not added: "+err.Error())
+					} else {
+						st.Matched = append(st.Matched, key+": filters added: "+strings.Join(add, ", "))
+						if p.Logf != nil {
+							p.Logf("consumer %s gained the filters %s", key, strings.Join(add, ", "))
+						}
+						p.say(res, EventAsserted, map[string]any{"consumer": c.Name, "stream": stream, "filters-added": add})
+						h.Filters = append(slices.Clone(h.Filters), add...)
+					}
+				}
 				why := consumerDiffers(c, h)
 				if why != "" && key == match && st.On && p.Consumers != nil {
 					if err := p.Consumers.MatchConsumer(stream, buscompose.ConsumerConfig(c)); err != nil {

modules/mesh-grants/internal/grants/part.go

@@ -399,6 +399,9 @@ func (p *Part) Compose() Result {
 		if p.Create != nil && heldErr == nil {
 			if made := createMissing(p.Create, &p.made, &a, in.Manifests[module], in.Nodes, forced == module); len(made) > 0 {
 				res.Asserted = append(res.Asserted, fmt.Sprintf("%s at %s: created %s", module, a.Manifest, strings.Join(made, ", ")))
+				if p.Logf != nil {
+					p.Logf("%s at %s: asserted on the bus: %s", module, a.Manifest, strings.Join(made, ", "))
+				}
 				p.say(&res, EventAsserted, map[string]any{"module": module, "manifest": a.Manifest, "created": made})
 			}
 			for _, why := range notCreated(a) {