DLV-G2: the bus module's process serves nothing, as every process account (ADR 0436 §3): it hears its module's consumes on a consumer of its own, says its stream's events and raises its conditions; the seat's reading verbs stay its module's tools'

The keeper is asked by a person through its own state requests (CAT-N8); the acting verbs
are served by nobody on the bus and denied to every principal whose grant reaches them. The
bus module's tools bundle may read its module's own stream back (nats_history).

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

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

@@ -1070,7 +1070,9 @@ func theBusModulesProcessKeepsBusState(t *testing.T) {
 	r := Records{Nodes: []string{"a"}, Assigned: map[string][]Declared{"a": {
 		{Module: "node-tools", ToolRunner: true},
 		{Module: "nats", SnapshotsTheBus: true, Holds: []Seat{{Name: catalogue.BrokerSeat, Serves: BrokerVerbs()}},
-			Processes: []ProcessPrincipal{{Process: "keeper", Emits: []string{"created"}}, {Process: "nats-tools", Tools: true,
+			Consumes: []string{"mesh-delivery-indexer.manifest-accepted"}, Uses: []Seat{{Name: "operator-channel", Accepts: []string{"condition"}, ByCaller: []string{"condition"}}},
+			Streams: []ModuleStream{{Module: "nats", Name: "BROKER_EVENTS", Events: []string{"created", "refused"}, MaxAgeDays: 30, MaxMiB: 256}},
+			Processes: []ProcessPrincipal{{Process: "keeper", State: []string{"asked"}}, {Process: "nats-tools", Tools: true,
 				Removes: []string{catalogue.RemovesEngineActs}}}},
 		{Module: "mod-x"},
 		{Module: "admin-ish", Invokes: []string{"*"}},
@@ -1107,13 +1109,38 @@ func theBusModulesProcessKeepsBusState(t *testing.T) {
 			t.Errorf("the bus module's process may define %s", s)
 		}
 	}
+	// A process serves nothing (ADR 0436 §3): it hears its module's consumes on its own consumer, says its module's own
+	// stream's events and raises its conditions through the operator channel.
 	for _, v := range BrokerVerbs() {
-		if !MaySubscribe(keeper, "mesh.seat.mesh-broker.tool."+v) {
-			t.Errorf("the bus module's process does not serve %s", v)
+		if MaySubscribe(keeper, "mesh.seat.mesh-broker.tool."+v) {
+			t.Errorf("the bus module's process serves %s", v)
 		}
 	}
-	if !keeper.AllowResponses {
-		t.Error("the bus module's process cannot answer what it serves")
+	if keeper.AllowResponses {
+		t.Error("the bus module's process answers what it was asked")
+	}
+	for _, s := range []string{"$JS.API.CONSUMER.MSG.NEXT.EVENTS.a_nats_keeper", "$JS.ACK.EVENTS.a_nats_keeper.x",
+		"mesh.stream.nats.event.created", "mesh.seat.operator-channel.accept.condition.nats"} {
+		if !MayPublish(keeper, s) {
+			t.Errorf("the bus module's process may not publish %s", s)
+		}
+	}
+	made := false
+	for _, mc := range AccountConsumers(users) {
+		if mc.Consumer.Stream == "EVENTS" && mc.Consumer.Name == "a_nats_keeper" &&
+			slices.Equal(mc.Consumer.Filters, []string{"mesh.mod.mesh-delivery-indexer.event.manifest-accepted"}) {
+			made = true
+		}
+	}
+	if !made {
+		t.Errorf("the bus module's process's consumer is not made: %+v", AccountConsumers(users))
+	}
+	// Its module's tools serve the reading verbs, and its tools bundle reads the module's own stream back.
+	if !MaySubscribe(perms["a.node-tools"], "mesh.seat.mesh-broker.tool.list") {
+		t.Error("nothing serves the seat's list")
+	}
+	if !MayPublish(perms["a.nats.nats-tools"], "$JS.API.CONSUMER.CREATE.BROKER_EVENTS.x.y") {
+		t.Error("the bus module's tools cannot read its events back")
 	}
 	for user, p := range perms {
 		if user == "a.nats.keeper" {

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

@@ -11,9 +11,12 @@ import (
 // **The bus module's process has a bus account of its own** (novox/hq ADR 0595 §4, §5; design 25 § 12.2, § 12.3;
 // DLV-G2): composed for the process of the module holding the seat mesh-broker (a seat, never a module's name), it
 // creates, updates, reads, snapshots and deletes every module's bus state under that module's names only, each stream
-// by its name (a key-value bucket's and an object store's by the stream the bus keeps it in), and serves the seat's
-// eight verbs. No other account holds the four acting verbs' subjects, the tool runner's included: each is denied to
-// every principal whose publish would reach it, and only the bus module's process subscribes to them.
+// by its name (a key-value bucket's and an object store's by the stream the bus keeps it in); it hears what its module
+// consumes (the indexer's manifest-accepted) on a durable consumer of its own, says its module's own stream's events,
+// and raises its conditions through the operator channel. As every process account it serves nothing (ADR 0436 §3): the
+// seat's four reading verbs are served by its module's tools, and the four acting verbs by nobody on the bus, a
+// person's ask reaching the process through its own state. No account holds the acting verbs' subjects, the tool
+// runner's included: each is denied to every principal whose publish would reach it.
 //
 // The names are one rule (BusStateStreams), read by the bus module where it creates them: until mesh-sdk carries it
 // as one function (asked of the mesh-sdk member of the milestone), this is the grants part's side of it.
@@ -80,6 +83,45 @@ func BusStateStreams(module string, e catalogue.BusStateEntry) []string {
 	return slices.Compact(out)
 }
 
+// keeperConsumer is the durable consumer the bus module's process hears its module's consumes on, on EVENTS:
+// `<node>_<module>_<process>`, as a process's own consumer is named (QueueConsumer), filtered on the consumed events.
+func keeperConsumer(p Principal) (Consumer, bool) {
+	var filters []string
+	for _, c := range p.Consumes {
+		if s, err := consumedSubject(c); err == nil && !strings.HasPrefix(s, moduleStreamPrefix) {
+			filters = append(filters, s)
+		}
+	}
+	if len(filters) == 0 {
+		return Consumer{}, false
+	}
+	sort.Strings(filters)
+	return Consumer{Name: p.Node + "_" + p.Module + "_" + p.Process, Stream: "EVENTS", Filters: slices.Compact(filters),
+		AckWaitSeconds: 30, MaxDeliver: 5, FromNow: true,
+		Why: "what the bus module's process " + p.Process + " hears of its module's consumes (novox/hq ADR 0595 §4): the " +
+			"indexer's accepted verdicts, from now, its first view at start holding what came before"}, true
+}
+
+// keeperConsumerGrants are the subjects the process reads and acknowledges its consumer through.
+func keeperConsumerGrants(c Consumer) []string {
+	return []string{"$JS.API.CONSUMER.INFO." + c.Stream + "." + c.Name, "$JS.API.CONSUMER.MSG.NEXT." + c.Stream + "." + c.Name,
+		"$JS.ACK." + c.Stream + "." + c.Name + ".>"}
+}
+
+// ownStreamReadGrants are what reading a module's own streams back needs (the bus module's tool history): each stream's
+// information and an ordered consumer of its own, as a log's reader has (readLogGrants).
+func ownStreamReadGrants(module string, streams []ModuleStream) []string {
+	var out []string
+	for _, s := range streams {
+		if s.Module != module || !safeSubject.MatchString(s.Name) {
+			continue
+		}
+		out = append(out, "$JS.API.STREAM.INFO."+s.Name, "$JS.API.CONSUMER.CREATE."+s.Name+".>",
+			"$JS.API.CONSUMER.MSG.NEXT."+s.Name+".*", "$JS.API.CONSUMER.INFO."+s.Name+".*", "$JS.API.CONSUMER.DELETE."+s.Name+".*")
+	}
+	return out
+}
+
 // BusStateNames are every stream the bus module's process may define: those of every entry a manifest asks for now
 // (asked), and those the bus module said it created and has not dropped (bound, by module: names of streams), so an
 // orphan whose manifest left the trunk can still be dropped. Sorted, once each.
@@ -126,3 +168,14 @@ func keepsBusState(p Principal, pattern string) bool {
 	}
 	return false
 }
+
+// withoutActing are a holder's served verbs of mesh-broker without the acting four.
+func withoutActing(served []string) []string {
+	var out []string
+	for _, v := range served {
+		if !slices.Contains(BrokerActingVerbs, v) {
+			out = append(out, v)
+		}
+	}
+	return out
+}

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

@@ -229,6 +229,12 @@ func ConsumersOf(users []Principal) []ModuleConsumer {
 		if p.Kind == KindModule {
 			add(p)
 		}
+		// The bus module's process hears its module's consumes on a consumer of its own (busstate.go).
+		if p.Kind == KindProcess && p.KeepsBus {
+			if c, ok := keeperConsumer(p); ok {
+				out = append(out, ModuleConsumer{Node: p.Node, Module: p.Module, Consumer: c})
+			}
+		}
 	}
 	for _, p := range users {
 		if p.Kind != KindNodeTools {

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

@@ -110,10 +110,9 @@ type Principal struct {
 	// Ledgers are the local names of the ledgers its module asks for in `requires`, for a KindLedger principal alone
 	// (issue 1966 D5): its whole authority is publishing on those ledgers' histories and snapshots (LedgerPublishGrants).
 	Ledgers []string
-	// ServesBroker says this is the process of the module holding mesh-broker, the bus module's own (busstate.go, ADR
-	// 0595 §4, §5): it serves the seat's eight verbs, and KeepsBusState are the streams it defines, every module's bus
-	// state by its name.
-	ServesBroker  bool
+	// KeepsBus says this is the process of the module holding mesh-broker, the bus module's own (busstate.go, ADR 0595
+	// §4, §5), and KeepsBusState are the streams it defines, every module's bus state by its name.
+	KeepsBus      bool
 	KeepsBusState []string
 	// Process is the process resource's id, for a KindProcess principal alone (novox/hq ADR 0436 §2):
 	// Username and inbox are <node>.<module>.<process id>.
@@ -597,6 +596,8 @@ func PermissionsFor(p Principal) (Permissions, error) {
 		if err != nil {
 			return Permissions{}, err
 		}
+		// The bus module's tools bundle reads its module's own stream back (its tool history, design 25 § 12.3).
+		pub = append(pub, ownStreamReadGrants(p.Module, p.Streams)...)
 		sort.Strings(pub)
 		if err := CheckWriters(p, pub); err != nil {
 			return Permissions{}, err
@@ -899,10 +900,10 @@ func PermissionsFor(p Principal) (Permissions, error) {
 			"$JS.API.CONSUMER.INFO."+consumerStream(p)+"."+consumerDurable(p),
 			"$JS.API.CONSUMER.MSG.NEXT."+consumerStream(p)+"."+consumerDurable(p))
 
-		// 3. Seats it holds: full participation. But mesh-broker's verbs, which its module's process serves (busstate.go).
+		// 3. Seats it holds: full participation. But mesh-broker's acting verbs, which nobody serves on the bus (busstate.go).
 		for _, s := range p.Holds {
 			if s.Name == catalogue.BrokerSeat {
-				s.Serves = nil
+				s.Serves = withoutActing(s.Serves)
 			}
 			if s.isNewTraffic() {
 				// Composed by SeatTrafficOf below, worker and all; only its tools are served here.
@@ -999,9 +1000,9 @@ func PermissionsFor(p Principal) (Permissions, error) {
 			// And its own streams' events (novox/hq ADR 0363 §2).
 			pub = append(pub, ownStreamGrants(d.Module, d.Streams)...)
 			for _, s := range d.Holds {
-				// mesh-broker's verbs are its holder's process's, never the runtime's (busstate.go).
+				// mesh-broker's acting verbs are served by nobody on the bus (busstate.go).
 				if s.Name == catalogue.BrokerSeat {
-					continue
+					s.Serves = withoutActing(s.Serves)
 				}
 				for _, t := range s.Serves {
 					sub = append(sub, seatToolSubject(s, t, p.Node))
@@ -1196,7 +1197,7 @@ func PermissionsFor(p Principal) (Permissions, error) {
 		// provisioner client is asked by its own machine's tool runner alone, on its own machine's subject (novox/hq
 		// ADR 0370 §3), and answers that. A person is never asked anything, and is granted nothing here.
 		AllowResponses: p.Kind == KindModule || p.Kind == KindNodeTools || p.Kind == KindNode ||
-			p.Kind == KindProvisioner || (p.Kind == KindProcess && p.ServesBroker),
+			p.Kind == KindProvisioner,
 	}, nil
 }
 

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

@@ -52,11 +52,17 @@ func processGrants(p Principal) (pub, sub []string, err error) {
 	}
 	pub = append(pub, invoked...)
 
-	// 6. The bus module's process (busstate.go, ADR 0595 §4, §5): every module's bus state it keeps, by name, and the
-	// seat mesh-broker's eight verbs it serves.
-	if p.ServesBroker {
+	// 6. The bus module's process (busstate.go, ADR 0595 §4, §5): every module's bus state it keeps, by name; its own
+	// consumer of what its module consumes; its module's own stream's events; and its conditions through the seats its
+	// module uses (the operator channel's accept.condition in its module's name).
+	if p.KeepsBus {
 		pub = append(pub, busStateKeeperGrants(p.KeepsBusState)...)
-		sub = append(sub, BrokerVerbSubjects(BrokerVerbs())...)
+		if c, ok := keeperConsumer(p); ok {
+			pub = append(pub, keeperConsumerGrants(c)...)
+		}
+		pub = append(pub, ownStreamGrants(p.Module, p.Streams)...)
+		tp, ts := seatTrafficGrants(SeatTrafficOf(p.Module, nil, p.Uses, nil))
+		pub, sub = append(pub, tp...), append(sub, ts...)
 	}
 
 	return pub, sub, nil

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

@@ -239,10 +239,15 @@ func Users(r Records) ([]Principal, error) {
 					Emits: pr.Emits, Queues: pr.Queues, State: pr.State, Uses: d.Uses, Asks: pr.Asks,
 					Invokes: pr.Invokes, Removes: pr.Removes,
 				})
-				// The process of the module holding mesh-broker keeps every module's bus state (busstate.go, ADR 0595 §4).
+				// The process of the module holding mesh-broker keeps every module's bus state (busstate.go, ADR 0595 §4),
+				// and its tools bundle reads its module's own stream back.
 				if d.SnapshotsTheBus && !pr.Tools && len(pr.Removes) == 0 {
-					out[len(out)-1].ServesBroker = true
-					out[len(out)-1].KeepsBusState = BusStateNames(r.BusState, r.Bound)
+					k := &out[len(out)-1]
+					k.KeepsBus, k.KeepsBusState = true, BusStateNames(r.BusState, r.Bound)
+					k.Consumes, k.Streams = d.Consumes, d.Streams
+				}
+				if d.SnapshotsTheBus && len(pr.Removes) > 0 {
+					out[len(out)-1].Streams = d.Streams
 				}
 				// The ended controller's leftovers include each node's declaration files' bucket (ADR 0590 clause 6).
 				if slices.Contains(pr.Removes, catalogue.RemovesControllerLeftovers) {