mesh-grants: create a declared bucket, log or stream the bus lacks, so a new module state no longer waits for a controller start

A module's new state reached the bus only when the controller started (issue 1790, met three
times: puts, queues, mesh-vault's bus-accounts). The operator allowed one more grant in the
controller's composer on 2026-10-26 08:54, creation only: the part's process now creates each
bucket, log or stream a placed module's manifest declares and the bus lacks, with the same
configuration the controller asserts (mesh-sdk BucketStreamConfig, Log.Config,
ModuleStreamConfig), on the event that brought the manifest, never on a timer. It says asserted
and logs one line per module; a stream the bus holds is never updated or deleted, its difference
only said; a create refused is said and tried at the next event; one created at this manifest
and gone since is made again only by hand. grants_assert asks the process to do the same for one
module now. The user list, the grants and any change to a held stream stay in shadow. mesh-sdk
moves to fix/1790-grants-create-buckets, whose composer grants this, so the shadow list is
composed as the controller's.

Issue: 1790

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

@@ -0,0 +1,34 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/nats-io/nats.go/jetstream"
+
+	"mesh-grants/internal/grants"
+)
+
+// creating creates a stream a manifest declares that the bus lacks (issue 1790; the operator, 2026-10-26 08:54): the
+// create alone, which the account is granted by the stream's name, never an update or a delete. The one place the part
+// creates on the bus.
+type creating struct {
+	ctx context.Context
+	js  jetstream.JetStream
+}
+
+func (c creating) CreateDeclared(cfg jetstream.StreamConfig) error {
+	ctx, cancel := context.WithTimeout(c.ctx, requestWithin)
+	defer cancel()
+	_, err := c.js.CreateStream(ctx, cfg)
+	switch {
+	case err == nil:
+		return nil
+	case errors.Is(err, jetstream.ErrStreamNameAlreadyInUse):
+		return grants.ErrHeld
+	case errors.Is(err, context.DeadlineExceeded):
+		return fmt.Errorf("no answer in %s: a create the bus refuses is answered by none (%w)", requestWithin, err)
+	}
+	return err
+}

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

@@ -5,7 +5,10 @@
 //
 // **Memberships are published for real** (the operator, 2026-10-25 22:31): since release 6 nobody publishes them, so a
 // module placed since cannot serve its seat or reach a new state. Each is published on the assignments stream only when
-// it differs from the one last published there. **The user list, the assertions and the grants stay in shadow** until
+// it differs from the one last published there. **A bucket, log or stream a manifest declares that the bus lacks is
+// created** (the operator, 2026-10-26 08:54; issue 1790), once per manifest, with the manifest's limits, on the event
+// that brought the manifest; one the bus holds is never changed. **The user list, the rest of the assertions and the
+// grants stay in shadow** until
 // the operator's take-over, which this milestone always refuses: they are composed, compared with what the bus loaded
 // and holds (through the bus module's own read tools) and kept in the part's own state, and take no effect.
 //
@@ -112,7 +115,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	}
 	bus := newBusReader(ctx, js)
 	f.Part = &grants.Part{Read: bus, Call: conn, Store: ownStore{conn}, Emit: emitting{conn},
-		Publish: memberships{ctx: ctx, nc: nc, js: js}}
+		Publish: memberships{ctx: ctx, nc: nc, js: js}, Create: creating{ctx: ctx, js: js}}
 	f.Watch = bus
 	f.Asked = func(wake func()) error {
 		return conn.State(grants.StateUserList).Watch(grants.KeyAsked, func(c stdio.StateChange) error {
@@ -122,7 +125,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 			return nil
 		})
 	}
-	logger.Printf("started on the account %s: publishing memberships; the user list, assertions and grants in shadow", account.User)
+	logger.Printf("started on the account %s: publishing memberships and creating the declared buckets, logs and streams the bus lacks; the user list, grants and any change to a held stream in shadow", account.User)
 	f.Run(ctx)
 	return nil
 }

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

@@ -91,8 +91,10 @@ func toolsOf(p *grants.Part) []stdio.Tool {
 			"verb, and the answer says so.",
 			Run: func(a map[string]any) (any, error) { return p.AskCompose(stdio.CallerOf(a)) }},
 		{Name: grants.ToolAssert, Description: "One module's buckets, logs and streams (ADR 0547 clause 6), composed from its manifest at " +
-			"the indexer's trunk with their limits, beside what the bus holds (through the bus module's nats_streams). In shadow it " +
+			"the indexer's trunk with their limits, beside what the bus holds (through the bus module's nats_streams). When the " +
-			"asserts nothing, and says so.",
+			"bus lacks one, asks the part's process (which alone holds the grant) to create it now with the manifest's limits, " +
+			"as it does by itself when a manifest is registered (issue 1790), and says asserted; one the bus holds is never " +
+			"changed, a difference is only said. Answers the comparison and the ask; call again to see what the bus holds.",
 			Input: schema(map[string]string{"module": "the module"}, "module"),
 			Run:   func(a map[string]any) (any, error) { return p.AssertOne(grants.Answer(a, "module")) }},
 		{Name: grants.ToolTakeOver, Description: "The operator's switch to the grants part's list (ADR 0547 clause 8), refused without the " +

modules/mesh-grants/go.mod

@@ -3,7 +3,7 @@ module mesh-grants
 go 1.26.0
 
 require (
-	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025215021-b7b5c6ef27a0
+	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026080443-136fa8b2a896
 	github.com/nats-io/nats.go v1.54.0
 )
 

modules/mesh-grants/go.sum

@@ -1,5 +1,5 @@
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025215021-b7b5c6ef27a0 h1:8dKGrZzPthx/jZlV9Bjm2lWGNORUTETmq2vxKN+MRV0=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026080443-136fa8b2a896 h1:5rHfru5NJ7SK20QOt8YUJSglcn1gsjtLc9AGmWYGnRQ=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025215021-b7b5c6ef27a0/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026080443-136fa8b2a896/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
 github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
 github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
 github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=

modules/mesh-grants/grants_test.go

@@ -14,6 +14,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/nats-io/nats.go/jetstream"
+
 	"git.novox.be/novox/mesh-sdk/go/catalogue"
 	"git.novox.be/novox/mesh-sdk/go/delivery"
 
@@ -43,6 +45,9 @@ type fakeBus struct {
 	refusePub bool
 	// 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
+	refuseCreate bool
 }
 
 func (b *fakeBus) LastMembership(node, module string) ([]byte, uint64, bool, error) {
@@ -72,6 +77,25 @@ func (b *fakeBus) PublishMembership(node, module string, body []byte, after uint
 	return nil
 }
 
+// CreateDeclared is the bus's create of a stream: refused when refuseCreate, held when the bus holds it, and otherwise
+// held from now on, as nats_streams then reads it.
+func (b *fakeBus) CreateDeclared(cfg jetstream.StreamConfig) error {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	b.creates = append(b.creates, cfg.Name)
+	if b.refuseCreate {
+		return errors.New("Permissions Violation for Publish to $JS.API.STREAM.CREATE." + cfg.Name)
+	}
+	for _, s := range b.streams {
+		if s["name"] == cfg.Name {
+			return grants.ErrHeld
+		}
+	}
+	b.streams = append(b.streams, map[string]any{"name": cfg.Name, "subjects": cfg.Subjects, "retention": "limits",
+		"max_age_hours": int(cfg.MaxAge.Hours()), "max_per_subject": int(cfg.MaxMsgsPerSubject)})
+	return nil
+}
+
 func newFakeBus() *fakeBus {
 	return &fakeBus{states: map[string]map[string][]byte{}, refused: map[string]bool{}, own: map[string]map[string][]byte{},
 		revs: map[string]uint64{}, refuseTo: map[string]bool{}}
@@ -295,12 +319,14 @@ func TestGrants(t *testing.T) {
 		}
 	})
 
-	t.Run("no timer, no create, a publish only of memberships", func(t *testing.T) {
+	t.Run("no timer, no update or delete, a publish only of memberships, 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", "CreateStream", "CreateKeyValue", "CreateOrUpdate", "UpdateStream", "UpdateKeyValue",
+			"PublishMsg", "CreateKeyValue", "CreateOrUpdate", "UpdateStream", "UpdateKeyValue", "DeleteStream", "PurgeStream",
-			"CreateObjectStore", "EnsureStream", "RaiseBuckets", "RaiseLogs", ".Create("}
+			"DeleteKeyValue", "CreateObjectStore", "EnsureStream", "RaiseBuckets", "RaiseLogs", ".Create("}
-		// The one publish: a membership, on its own subject, in publish.go.
+		// The one publish: a membership, on its own subject, in publish.go; the one create: a declared stream the bus lacks,
+		// in create.go (issue 1790).
 		publishOnly := filepath.Join("cmd", "mesh-grants", "publish.go")
+		createOnly := filepath.Join("cmd", "mesh-grants", "create.go")
 		files := 0
 		for _, root := range []string{"cmd", "internal"} {
 			err := filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error {
@@ -315,6 +341,9 @@ func TestGrants(t *testing.T) {
 				if path == publishOnly && strings.Count(string(src), ".Publish(") != 1 {
 					t.Errorf("publish.go publishes other than one membership")
 				}
+				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)
+				}
 				for _, f := range forbidden {
 					if strings.Contains(string(src), f) {
 						t.Errorf("%s holds %s: the part holds no timer and creates or asserts nothing", path, f)
@@ -513,8 +542,9 @@ func TestGrants(t *testing.T) {
 		}
 
 		a, err := p.AssertOne("mod-x")
-		as, _ := a.(grants.Assertion)
+		am, _ := a.(map[string]any)
-		if err != nil || as.Asserted || as.Says != grants.ShadowAsserts || len(as.Items) != 3 {
+		as, _ := am["assertion"].(grants.Assertion)
+		if err != nil || as.Asserted || as.Says != grants.ShadowAsserts || len(as.Items) != 3 || am["asked"] == nil {
 			t.Fatalf("assert = %+v, %v", a, err)
 		}
 		for _, it := range as.Items {
@@ -721,6 +751,95 @@ func TestGrants(t *testing.T) {
 		})
 	})
 
+	t.Run("assertions: what the bus lacks is created once, what it holds is never touched", func(t *testing.T) {
+		b := fixture()
+		b.loaded = shadowOf(t, b)
+		// The bus holds mod-x's bucket with another history, and neither its log nor its stream.
+		b.streams = []map[string]any{{"name": "KV_mod-x_kept", "subjects": []string{"$KV.mod-x_kept.>"}, "retention": "limits", "max_per_subject": 2}}
+		p := partOn(b)
+		p.Create = b
+		assertion := func() grants.Assertion {
+			raw, _, _, _ := b.ownGet(grants.StateAssertions, "mod-x")
+			var a grants.Assertion
+			_ = json.Unmarshal(raw, &a)
+			return a
+		}
+		first := p.Compose()
+		if !slices.Equal(b.creates, []string{"LOG_mod-x_history", "THINGS"}) || len(first.Asserted) != 1 ||
+			!strings.HasPrefix(first.Asserted[0], "mod-x at ") || !slices.Contains(b.events, grants.EventAsserted) {
+			t.Fatalf("the first pass created %v, said %v, events %v", b.creates, first.Asserted, b.events)
+		}
+		a := assertion()
+		if !a.Asserted || a.Says != grants.CreatesMissing {
+			t.Errorf("the assertion kept = %+v", a)
+		}
+		for _, it := range a.Items {
+			switch it.Stream {
+			case "KV_mod-x_kept":
+				if it.Created != "" || !slices.Contains(it.Differences, "history 2 held, 3 composed") {
+					t.Errorf("a bucket the bus holds was touched, or its difference not said: %+v", it)
+				}
+			default:
+				if it.Created != grants.ItemCreated {
+					t.Errorf("%s: %q, want created", it.Stream, it.Created)
+				}
+			}
+		}
+		if again := p.Compose(); len(again.Asserted) != 0 || len(b.creates) != 2 {
+			t.Errorf("a pass with everything held created again: %v, %v", again.Asserted, b.creates)
+		}
+		// THINGS deleted since: not made again at the same manifest, said; a hand assert (the tool, asking the process)
+		// makes it again.
+		b.mu.Lock()
+		b.streams = slices.DeleteFunc(b.streams, func(s map[string]any) bool { return s["name"] == "THINGS" })
+		b.mu.Unlock()
+		gone := p.Compose()
+		if len(b.creates) != 2 || !strings.Contains(strings.Join(gone.NotAsserted, " "), "THINGS: created at this manifest and gone since") {
+			t.Errorf("a stream gone since was made again by itself, or not said: %v, %v", b.creates, gone.NotAsserted)
+		}
+		tool := partOn(b)
+		if got, err := tool.AssertOne("mod-x"); err != nil || fmt.Sprint(got) == "" || len(b.creates) != 2 {
+			t.Fatalf("the tool = %v, %v, creates %v: it asks, and creates nothing itself", got, err, b.creates)
+		}
+		if hand := p.Compose(); len(hand.Asserted) != 1 || b.creates[len(b.creates)-1] != "THINGS" {
+			t.Errorf("the hand assert did not create THINGS again: %v, %v", hand.Asserted, b.creates)
+		}
+		if after := p.Compose(); len(after.Asserted) != 0 {
+			t.Errorf("the hand ask was acted on twice: %v", after.Asserted)
+		}
+		for _, c := range b.creates {
+			if c == "KV_mod-x_kept" {
+				t.Error("the bucket the bus holds was asked to be created")
+			}
+		}
+	})
+
+	t.Run("assertions: a refused create is said, never stops it, and is tried at the next event", func(t *testing.T) {
+		b := fixture()
+		b.loaded = shadowOf(t, b)
+		b.refuseCreate = true
+		p := partOn(b)
+		p.Create = b
+		refused := p.Compose()
+		if len(refused.Asserted) != 0 || !strings.Contains(strings.Join(refused.NotAsserted, " "), "Permissions Violation") ||
+			slices.Contains(b.events, grants.EventAsserted) {
+			t.Fatalf("a refused create = %+v", refused)
+		}
+		b.refuseCreate = false
+		if next := p.Compose(); len(next.Asserted) != 1 || !strings.Contains(next.Asserted[0], "KV_mod-x_kept") {
+			t.Errorf("the next event created %v", next.Asserted)
+		}
+		// What the bus holds unread: nothing is created.
+		c := fixture()
+		c.loaded = shadowOf(t, c)
+		q := partOn(c)
+		q.Create = c
+		q.Call = refusingStreams{c}
+		if r := q.Compose(); len(c.creates) != 0 || len(r.Asserted) != 0 {
+			t.Errorf("created %v with what the bus holds unread", c.creates)
+		}
+	})
+
 	t.Run("memberships held back: every one is said and listed", func(t *testing.T) {
 		b := fixture()
 		b.loaded = shadowOf(t, b)
@@ -813,6 +932,16 @@ func nonNil(s []string) []string {
 }
 
 // fakeWatcher refuses a state's first watches as told, and hands changes to the watches it opened.
+// refusingStreams is the bus with nats_streams refused: what the bus holds cannot be read.
+type refusingStreams struct{ *fakeBus }
+
+func (r refusingStreams) Call(key string, body any) (json.RawMessage, error) {
+	if key == grants.CallStreams {
+		return nil, errors.New("refused")
+	}
+	return r.fakeBus.Call(key, body)
+}
+
 type fakeWatcher struct {
 	mu      sync.Mutex
 	refuse  map[string]int

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

@@ -2,9 +2,14 @@ package grants
 
 import (
 	"encoding/json"
+	"errors"
 	"fmt"
 	"slices"
 	"sort"
+	"strings"
+	"sync"
+
+	"github.com/nats-io/nats.go/jetstream"
 
 	"git.novox.be/novox/mesh-sdk/go/buscompose"
 	"git.novox.be/novox/mesh-sdk/go/catalogue"
@@ -19,6 +24,9 @@ type Item struct {
 	Held     *Limits `json:"held,omitempty"`
 	// Differences says how they differ, in words; empty when they agree in what can be read from here.
 	Differences []string `json:"differences,omitempty"`
+	// Created says what became of creating it when the bus lacked it: "created", or why not (issue 1790). Empty when
+	// the bus holds it or no creation was asked here.
+	Created string `json:"created,omitempty"`
 }
 
 // Limits are the facts of a stream an assertion sets and the bus module's nats_streams reads back. Its size cap is not
@@ -31,8 +39,8 @@ type Limits struct {
 	MaxBytes      int64    `json:"max_bytes,omitempty"`
 }
 
-// Assertion is one module's declared buckets, logs and streams, composed and compared. Asserted is always false in
+// Assertion is one module's declared buckets, logs and streams, composed and compared. Asserted says this pass created
-// shadow: nothing is created or changed.
+// one the bus lacked (the operator, 2026-10-26 08:54); one the bus holds is never changed.
 type Assertion struct {
 	Module   string `json:"module"`
 	Manifest string `json:"manifest"`
@@ -41,9 +49,119 @@ type Assertion struct {
 	Says     string `json:"says"`
 }
 
-// ShadowAsserts is what every assertion says in this milestone.
+// ShadowAsserts is what an assertion says where nothing may be created: in the tools, which ask the part's process.
 const ShadowAsserts = "in shadow: composed and compared, nothing asserted (ADR 0547 clause 5 asserts from take-over)"
 
+// CreatesMissing is what an assertion of the part's process says (the operator, 2026-10-26 08:54; issue 1790).
+const CreatesMissing = "a bucket, log or stream the manifest declares and the bus lacks is created with the manifest's limits, once per " +
+	"manifest; one the bus holds is never changed or deleted, a difference is only said (ADR 0547 clause 5 asserts the rest from take-over)"
+
+// Creator creates one stream the bus does not hold, with the configuration given, and never updates or deletes one: the
+// part's process alone, whose account is granted the create of each declared stream by name (the operator, 2026-10-26
+// 08:54). A stream the bus holds already answers ErrHeld.
+type Creator interface {
+	CreateDeclared(cfg jetstream.StreamConfig) error
+}
+
+// ErrHeld is a create refused because the bus holds the stream already (made meanwhile): left as it is.
+var ErrHeld = errors.New("the bus holds it already, so it is left as it is")
+
+// Created words of an item.
+const (
+	ItemCreated    = "created"
+	ItemNotCreated = "not created: "
+)
+
+// Configs are the streams a manifest declares, as the controller asserts them at its start (mesh-sdk buscompose), by
+// stream name. Pure.
+func Configs(m catalogue.Manifest) map[string]jetstream.StreamConfig {
+	out := map[string]jetstream.StreamConfig{}
+	for _, b := range bucketsOf(m) {
+		c := buscompose.BucketStreamConfig(b)
+		out[c.Name] = c
+	}
+	for _, l := range logsOf(m) {
+		c := l.Config()
+		out[c.Name] = c
+	}
+	for _, s := range streamsOf(m) {
+		c := buscompose.ModuleStreamConfig(s)
+		out[c.Name] = c
+	}
+	return out
+}
+
+// creations remembers each stream created, by the manifest digest it was created at: created once per manifest, so a
+// stream deleted since is not made again until a new manifest or a hand assert (grants_assert).
+type creations struct {
+	mu   sync.Mutex
+	made map[string]string
+}
+
+func (c *creations) madeAt(stream string) string {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.made[stream]
+}
+
+func (c *creations) keep(stream, digest string) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.made == nil {
+		c.made = map[string]string{}
+	}
+	c.made[stream] = digest
+}
+
+// createMissing creates each item of a the bus lacks, with the manifest's limits, and says what became of it. It never
+// touches an item the bus holds, and does nothing when what the bus holds was not read. force asks again a stream
+// created at this manifest and gone since (a hand assert). It answers the streams created.
+func createMissing(c Creator, done *creations, a *Assertion, m catalogue.Manifest, force bool) []string {
+	cfgs := Configs(m)
+	var made []string
+	a.Says = CreatesMissing
+	for i := range a.Items {
+		it := &a.Items[i]
+		if it.Held != nil || !slices.Contains(it.Differences, "the bus holds no stream "+it.Stream) {
+			continue
+		}
+		if !force && done.madeAt(it.Stream) == a.Manifest {
+			it.Created = ItemNotCreated + "created at this manifest and gone since; grants_assert creates it again"
+			continue
+		}
+		cfg, ok := cfgs[it.Stream]
+		if !ok {
+			it.Created = ItemNotCreated + "no configuration composed for it"
+			continue
+		}
+		switch err := c.CreateDeclared(cfg); {
+		case err == nil:
+			it.Created = ItemCreated
+			done.keep(it.Stream, a.Manifest)
+			made = append(made, it.Kind+" "+it.Stream)
+		case errors.Is(err, ErrHeld):
+			it.Created = ItemNotCreated + ErrHeld.Error()
+		default:
+			it.Created = ItemNotCreated + err.Error() + " (tried again at the next event)"
+		}
+	}
+	if len(made) > 0 {
+		a.Asserted = true
+	}
+	return made
+}
+
+// notCreated is each item of a creation refused, with why, as `<stream>: <why>`.
+func notCreated(a Assertion) []string {
+	var out []string
+	for _, it := range a.Items {
+		if why, ok := strings.CutPrefix(it.Created, ItemNotCreated); ok && it.Created != "" {
+			out = append(out, it.Stream+": "+why)
+		}
+	}
+	return out
+}
+
 // Composed is a module's buckets, logs and streams as an assertion would ask for them. Pure.
 func Composed(m catalogue.Manifest) []Item {
 	var out []Item

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

@@ -36,8 +36,9 @@ type Follower struct {
 	watched map[string]bool
 	refused map[string]string
 	have    map[string]string
-	// held is what the last pass held back, said again only when it changes.
+	// held and notAsserted are what the last pass held back and could not create, said again only when they change.
-	held string
+	held        string
+	notAsserted string
 }
 
 func (f *Follower) init() {
@@ -145,8 +146,9 @@ func (f *Follower) open() {
 	}
 }
 
-// said logs every membership the pass held back, each with what it would lose (issue 1945), when the set changes from
+// said logs every membership the pass held back, each with what it would lose (issue 1945), and every declared stream
-// the last pass's.
+// the pass could not create, each with why (issue 1790), when the set changes from the last pass's; and every stream it
+// created, one line per module.
 func (f *Follower) said(res Result) {
 	if held := strings.Join(res.NotPublished, "\n"); held != f.held {
 		f.held = held
@@ -160,6 +162,15 @@ func (f *Follower) said(res Result) {
 			}
 		}
 	}
+	for _, a := range res.Asserted {
+		f.logf("asserted %s", a)
+	}
+	if not := strings.Join(res.NotAsserted, "\n"); not != f.notAsserted {
+		f.notAsserted = not
+		for _, n := range res.NotAsserted {
+			f.logf("not asserted %s", n)
+		}
+	}
 }
 
 // Run composes at start and after every change until ctx ends. It returns only when ctx ends.

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

@@ -1,8 +1,10 @@
 // Package grants is the grants part in shadow (novox/hq ADR 0547 clauses 5 to 8, issue 1945): it composes the bus's user
 // list, each module's membership and each manifest's buckets, logs and streams from what the bus holds, with the bus
-// composer the controller uses (mesh-sdk buscompose), and compares them with what the bus loaded and holds. It writes only
+// composer the controller uses (mesh-sdk buscompose), and compares them with what the bus loaded and holds. Beyond its own
-// its own state and takes no effect: it publishes no membership, asserts nothing, writes no grant and no user list the
+// state it takes effect in two things alone, each by a grant the operator allowed: it publishes memberships (2026-10-25
-// bus node reads, until the operator's take-over, which this milestone always refuses.
+// 22:31), and creates a bucket, log or stream a manifest declares that the bus lacks (2026-10-26 08:54, issue 1790). It
+// changes no stream the bus holds, writes no grant and no user list the bus node reads, until the operator's take-over,
+// which this milestone always refuses.
 package grants
 
 import (

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

@@ -35,10 +35,12 @@ var States = []string{StateAssertions, StateMemberships, StateTakeover, StateUse
 const (
 	EventUserList   = "user-list-changed"
 	EventMembership = "membership-changed"
+	// EventAsserted is said when the part created a bucket, log or stream a manifest declares (issue 1790).
+	EventAsserted = "asserted"
 )
 
 // Emits is every event the part says.
-var Emits = []string{EventMembership, EventUserList}
+var Emits = []string{EventAsserted, EventMembership, EventUserList}
 
 // Shadow is the switch's one state in this milestone: the bus node follows the controller's list.
 const Shadow = "shadow"
@@ -83,21 +85,25 @@ func (s UserListState) inputs() string {
 
 // Result is what one composition did.
 type Result struct {
-	UserList            UserListState `json:"user_list"`
+	UserList           UserListState `json:"user_list"`
-	Revision            uint64        `json:"revision"`
+	Revision           uint64        `json:"revision"`
-	Written             bool          `json:"written"`
+	Written            bool          `json:"written"`
-	Memberships         int           `json:"memberships"`
+	Memberships        int           `json:"memberships"`
-	MembershipsWritten  int           `json:"memberships_written"`
+	MembershipsWritten int           `json:"memberships_written"`
-	Published           []string      `json:"published,omitempty"`
+	Published          []string      `json:"published,omitempty"`
-	NotPublished        []string      `json:"not_published,omitempty"`
+	NotPublished       []string      `json:"not_published,omitempty"`
-	Assertions          int           `json:"assertions"`
+	Assertions         int           `json:"assertions"`
-	AssertionsWritten   int           `json:"assertions_written"`
+	AssertionsWritten  int           `json:"assertions_written"`
-	Errors              []string      `json:"errors,omitempty"`
+	// Asserted is each module whose declared streams this pass created, `<module> at <digest>: created …`, and
-	PublishedOrAsserted string        `json:"published_or_asserted"`
+	// NotAsserted each creation refused, `<module> <stream>: <why>` (issue 1790).
+	Asserted            []string `json:"asserted,omitempty"`
+	NotAsserted         []string `json:"not_asserted,omitempty"`
+	Errors              []string `json:"errors,omitempty"`
+	PublishedOrAsserted string   `json:"published_or_asserted"`
 }
 
 // NothingTakesEffect is what every composition says it did on the bus beyond its own state and the memberships.
-const NothingTakesEffect = "memberships only (on the assignments stream, a first one or one that only adds to the last published or drops only a verb the seat catalogue retired; any other loss stays a shadow difference); the user list, assertions and grants stay in shadow until take-over"
+const NothingTakesEffect = "memberships (on the assignments stream, a first one or one that only adds to the last published or drops only a verb the seat catalogue retired; any other loss stays a shadow difference), and each bucket, log or stream a manifest declares that the bus lacks, created once with the manifest's limits (the operator, 2026-10-26 08:54); the user list, the grants and any change to a stream the bus holds stay in shadow until take-over"
 
 // Part is the grants part: it reads, calls and writes through these, and nothing else.
 type Part struct {
@@ -107,11 +113,36 @@ type Part struct {
 	Emit  Emitter
 	// Publish publishes memberships: the process's connection; nil in the tools, which ask the process instead.
 	Publish Publisher
-	Now     func() time.Time
+	// Create creates a declared stream the bus lacks: the process's connection; nil in the tools, which ask the process.
+	Create Creator
+	Now    func() time.Time
 
 	cache Manifests
 	mu    sync.Mutex
 	last  Result
+	made  creations
+	// askSeen is the revision of the last ask to compose (KeyAsked) this part acted on.
+	askSeen uint64
+}
+
+// Ask is the value of KeyAsked: a tool's ask to compose now, and, from grants_assert, to create one module's missing
+// streams again even where they were created at this manifest before.
+type Ask struct {
+	By     string    `json:"by"`
+	At     time.Time `json:"at"`
+	Assert string    `json:"assert,omitempty"`
+}
+
+// askedAssert is the module a new ask names to assert by hand, read once per ask; empty when none.
+func (p *Part) askedAssert() string {
+	raw, rev, found, err := p.Store.Get(StateUserList, KeyAsked)
+	if err != nil || !found || rev == p.askSeen {
+		return ""
+	}
+	p.askSeen = rev
+	var a Ask
+	_ = json.Unmarshal(raw, &a)
+	return a.Assert
 }
 
 func (p *Part) now() time.Time {
@@ -187,11 +218,25 @@ func (p *Part) Compose() Result {
 		}
 	}
 	held, heldErr := HeldStreams(p.Call)
+	forced := ""
+	if p.Create != nil {
+		forced = p.askedAssert()
+	}
 	for _, module := range sortedKeys(in.Manifests) {
 		if !assigned(in, module) {
 			continue
 		}
 		a := Assert(module, in.Digests[module], in.Manifests[module], held, heldErr)
+		// Created only where the bus was read and lacks it, never changed where it holds it (issue 1790).
+		if p.Create != nil && heldErr == nil {
+			if made := createMissing(p.Create, &p.made, &a, in.Manifests[module], forced == module); len(made) > 0 {
+				res.Asserted = append(res.Asserted, fmt.Sprintf("%s at %s: created %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) {
+				res.NotAsserted = append(res.NotAsserted, module+" "+why)
+			}
+		}
 		res.Assertions++
 		data, _ := json.Marshal(a)
 		if p.changed(&res, StateAssertions, module, data) {

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

@@ -137,7 +137,7 @@ const NoGrantAgain = "missing: a verb to ask the controller to write a module's
 // AskCompose is the tool compose-now: it asks the part's process to compose and publish now (only the process holds the
 // grant to publish a membership), and answers the last pass, whose revision moves when the process answered.
 func (p *Part) AskCompose(caller string) (any, error) {
-	ask, _ := json.Marshal(map[string]any{"by": caller, "at": p.now()})
+	ask, _ := json.Marshal(Ask{By: caller, At: p.now()})
 	rev, err := p.Store.Put(StateUserList, KeyAsked, ask)
 	if err != nil {
 		return nil, fmt.Errorf("the ask could not be written: %w", err)
@@ -180,7 +180,9 @@ func (p *Part) UserList() (any, error) {
 }
 
 // AssertOne answers the tool assert: one module's buckets, logs and streams as composed from its manifest at the
-// indexer's trunk, beside what the bus holds. It asserts nothing.
+// indexer's trunk, beside what the bus holds, and asks the part's process (which alone holds the grant to create) to
+// create what the bus lacks now, even one it created at this manifest and that is gone since (issue 1790). It changes no
+// stream the bus holds. Here, in the tools, it creates nothing itself: the answer names the ask.
 func (p *Part) AssertOne(module string) (any, error) {
 	if module == "" {
 		return nil, errors.New("assert names a module")
@@ -201,7 +203,27 @@ func (p *Part) AssertOne(module string) (any, error) {
 		return nil, err
 	}
 	held, heldErr := HeldStreams(p.Call)
-	return Assert(module, v.Manifest, m, held, heldErr), nil
+	a := Assert(module, v.Manifest, m, held, heldErr)
+	out := map[string]any{"assertion": a}
+	missing := 0
+	for _, it := range a.Items {
+		if it.Held == nil {
+			missing++
+		}
+	}
+	if missing == 0 {
+		out["says"] = "the bus holds every bucket, log and stream the manifest declares: nothing to create; a difference is only said"
+		return out, nil
+	}
+	ask, _ := json.Marshal(Ask{By: "grants_assert", At: p.now(), Assert: module})
+	rev, err := p.Store.Put(StateUserList, KeyAsked, ask)
+	if err != nil {
+		return nil, fmt.Errorf("the ask to create could not be written: %w", err)
+	}
+	out["asked"] = rev
+	out["says"] = fmt.Sprintf("%d the bus lacks: the part's process creates them at this ask if %s is placed (its grant names only placed "+
+		"modules' streams) and says asserted; grants_assert again shows what the bus holds then", missing, module)
+	return out, nil
 }
 
 // Answer is a tool's arguments read as text.

modules/mesh-grants/module.json

@@ -19,6 +19,7 @@
     "nats.nats_users"
   ],
   "emits": [
+    "asserted",
     "membership-changed",
     "user-list-changed"
   ],