Review round 2: publish a module stream's event as the tool runner carrying it, not the controller (issue 1181)

Step 3 published mesh.stream.mesh-mail.event.mail.send.approved on the
controller's connection, which the controller is never granted. Added
the runtime module to the fixture's assigned set, gave mesh-mail its
Streams so the carried-module path grants ownStreamGrants, minted the
runtime's password through aMesh like the other principals, and publish
over that connection with its own inbox prefix.

cmd/mesh-controller/process_live_test.go

@@ -58,9 +58,12 @@ func TestAProcessAccountOnARealBusAndStore(t *testing.T) {
 
 	m := mailSenderLiveManifest(t)
 	operatorChannel := operatorChannelWithTell()
+	mailSends := broker.ModuleStream{Module: "mesh-mail", Name: "MAIL_SENDS", Events: []string{"mail.send.approved"},
+		MaxAgeDays: 1, MaxMiB: 8, WorkQueue: true, Duplicates: 1}
 	records := broker.Records{Nodes: []string{"laptop"}, Assigned: map[string][]broker.Declared{"laptop": {
+		{Module: broker.RuntimeModule},
 		{Module: "messenger", Holds: []broker.Seat{operatorChannel}},
-		{Module: "mesh-mail", NoAccount: true, Uses: []broker.Seat{operatorChannel},
+		{Module: "mesh-mail", NoAccount: true, Uses: []broker.Seat{operatorChannel}, Streams: []broker.ModuleStream{mailSends},
 			Processes: inventory.ProcessesOf(m)},
 	}}}
 	users, err := broker.Users(records)
@@ -68,21 +71,26 @@ func TestAProcessAccountOnARealBusAndStore(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	// The three principals this test drives, each minted the password module issue mints — the same
+	// The four principals this test drives, each minted the password module issue mints — the same
-	// inv.MintBusPassword call issueProcess makes for the process account.
+	// inv.MintBusPassword call issueProcess makes for the process account. mesh-mail publishes its own
-	var admin, messenger, sender broker.Principal
+	// stream's events as the tool runner carrying it (review round 2's finding): the controller's own
+	// grants do not cover a module's stream, only the runtime that carries the module does.
+	var admin, runtime, messenger, sender broker.Principal
 	for _, p := range users {
 		switch {
 		case p.Kind == broker.KindController:
 			admin = p
+		case p.Kind == broker.KindNodeTools:
+			runtime = p
 		case p.Module == "messenger":
 			messenger = p
 		case p.Kind == broker.KindProcess:
 			sender = p
 		}
 	}
-	if admin.Kind == "" || messenger.Kind == "" || sender.Kind == "" {
+	if admin.Kind == "" || runtime.Kind == "" || messenger.Kind == "" || sender.Kind == "" {
-		t.Fatalf("not every principal was composed: admin=%+v messenger=%+v sender=%+v", admin, messenger, sender)
+		t.Fatalf("not every principal was composed: admin=%+v runtime=%+v messenger=%+v sender=%+v",
+			admin, runtime, messenger, sender)
 	}
 	if sender.Username() != "laptop.mesh-mail.send" {
 		t.Fatalf("the process's username: %q", sender.Username())
@@ -93,7 +101,8 @@ func TestAProcessAccountOnARealBusAndStore(t *testing.T) {
 	for _, p := range []struct {
 		principal broker.Principal
 		kind      string
-	}{{admin, inventory.BusController}, {messenger, inventory.BusModule}, {sender, inventory.BusProcess}} {
+	}{{admin, inventory.BusController}, {runtime, inventory.BusNodeTools}, {messenger, inventory.BusModule},
+		{sender, inventory.BusProcess}} {
 		pw, err := inv.MintBusPassword(ctx, inventory.BusUser{Username: p.principal.Username(), Kind: p.kind,
 			Node: p.principal.Node, Module: p.principal.Module})
 		if err != nil {
@@ -106,7 +115,7 @@ func TestAProcessAccountOnARealBusAndStore(t *testing.T) {
 		}
 		hashes[p.principal.Username()] = hash
 	}
-	filled, missing := broker.WithPasswords([]broker.Principal{admin, messenger, sender}, hashes)
+	filled, missing := broker.WithPasswords([]broker.Principal{admin, runtime, messenger, sender}, hashes)
 	if len(missing) > 0 {
 		t.Fatalf("no password for %v", missing)
 	}
@@ -143,8 +152,6 @@ func TestAProcessAccountOnARealBusAndStore(t *testing.T) {
 		t.Fatal(err)
 	}
 	defer adminConn.Close()
-	mailSends := broker.ModuleStream{Module: "mesh-mail", Name: "MAIL_SENDS", Events: []string{"mail.send.approved"},
-		MaxAgeDays: 1, MaxMiB: 8, WorkQueue: true, Duplicates: 1}
 	if err := adminConn.EnsureStream(mailSends.Stream()); err != nil {
 		t.Fatal(err)
 	}
@@ -162,6 +169,15 @@ func TestAProcessAccountOnARealBusAndStore(t *testing.T) {
 		t.Fatal(err)
 	}
 
+	// The machine's tool runner, which carries mesh-mail (it has no account of its own) and so is the
+	// principal that publishes its own stream's events on its behalf.
+	runtimeConn, err := nats.Connect(s.ClientURL(), nats.UserInfo(runtime.Username(), passwords[runtime.Username()]),
+		nats.CustomInboxPrefix("_INBOX."+runtime.Username()))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer runtimeConn.Close()
+
 	// messenger, the router, answers exactly its granted verbs — never a wildcard the grant does not give it.
 	holder, err := nats.Connect(s.ClientURL(), nats.UserInfo(messenger.Username(), passwords[messenger.Username()]),
 		nats.CustomInboxPrefix("_INBOX."+messenger.Username()))
@@ -210,8 +226,14 @@ func TestAProcessAccountOnARealBusAndStore(t *testing.T) {
 		t.Fatalf("the process may not read its own state back: %v, %v", got, err)
 	}
 
-	// 3. Its queue: pulls and acknowledges MAIL_SENDS through its own consumer.
+	// 3. Its queue: pulls and acknowledges MAIL_SENDS through its own consumer. Published as the tool
-	if _, err := js.Publish(ctx, broker.ModuleStreamSubject("mesh-mail", "mail.send.approved"), []byte("{}")); err != nil {
+	// runner carrying mesh-mail — the live path a module's own stream's event takes — never as the
+	// controller, whose own grants do not cover a module's stream at all.
+	runtimeJS, err := jetstream.New(runtimeConn)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := runtimeJS.Publish(ctx, broker.ModuleStreamSubject("mesh-mail", "mail.send.approved"), []byte("{}")); err != nil {
 		t.Fatal(err)
 	}
 	reading, err := jetstream.New(sender2)