DLV-G2: the bus module's process gets its own account: it defines every module's bus state by its names and an orphan's it created, serves mesh-broker's eight verbs, and no other account holds the acting four (ADR 0595 §4, §5)

The grants are composed for the process of the module holding mesh-broker, from every manifest's bus-state entries and the streams of the bindings kept; the writers table names it beside the grants part as a definer of streams; the acting verbs are denied to every principal whose grant reaches them, and neither the holder's own account nor the tool runner serves the seat's verbs. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
2026-11-04 12:34:50 +01:00
parent 8975c37a0a
commit 8f6562368c
11 changed files with 307 additions and 7 deletions
@@ -50,6 +50,7 @@ func TestTheCopiedComposerComposesAsTheControllers(t *testing.T) {
 	t.Run("a declared stream is created as the controller asserted it", declaredStreamsAreCreatedAsTheControllerAssertsThem)
 	t.Run("every node's compositions are read by their named readers alone and written by their node", everyNodesCompositions)
 	t.Run("a module's ledger is written by its ledger account alone", aLedgerIsWrittenByItsLedgerAccountAlone)
+	t.Run("the bus module's process alone keeps every module's bus state and serves the broker's verbs", theBusModulesProcessKeepsBusState)
 	// ADR 0567 §3, §9: a node reads every key of the seat's placement and its candidates (it composes with the whole
 	// proposal laid over), and none of its open proposals; it writes neither.
 	reads := strings.Join(NodeComposeReads("laptop"), " ")
@@ -1060,3 +1061,72 @@ func aLedgerIsWrittenByItsLedgerAccountAlone(t *testing.T) {
 		}
 	}
 }
+
+// DLV-G2, 1966/AC35 (novox/hq ADR 0595 §4, §5; design 25 § 12.2, § 12.3): the process of the module holding mesh-broker
+// may create, update and delete every module's bus state by its names (a bucket's KV_, an object store's OBJ_, a ledger's
+// four parts) and an orphan's the bus module said it created, never a stream of the foundation; it serves the seat's eight
+// verbs; and no other principal, the tool runner and the bus module's own account included, holds the acting four.
+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,
+				Removes: []string{catalogue.RemovesEngineActs}}}},
+		{Module: "mod-x"},
+		{Module: "admin-ish", Invokes: []string{"*"}},
+	}},
+		BusState: map[string][]catalogue.BusStateEntry{"mod-x": {
+			{Name: "records", Storage: catalogue.StorageKeyValue, Template: catalogue.TemplateState},
+			{Name: "audit", Storage: catalogue.StorageLedger, Template: catalogue.TemplateLedger,
+				Types: []catalogue.LedgerType{{Name: "decision", Indexes: []catalogue.LedgerIndex{{Name: "by-status", On: []string{"status"}}}}}},
+			{Name: "parts", Storage: catalogue.StorageObjectStore, Template: catalogue.TemplateObjects}}},
+		Bound: map[string][]string{"gone": {"KV_gone_kept"}}}
+	users, err := Users(r)
+	if err != nil {
+		t.Fatal(err)
+	}
+	perms := map[string]Permissions{}
+	for _, u := range users {
+		p, err := PermissionsFor(u)
+		if err != nil {
+			t.Fatalf("%s: %v", u.Username(), err)
+		}
+		perms[u.Username()] = p
+	}
+	keeper := perms["a.nats.keeper"]
+	for _, s := range []string{"KV_mod-x_records", "mod-x_audit", "mod-x_audit_snapshot", "KV_mod-x_audit_state",
+		"KV_mod-x_audit_index-by-status", "OBJ_mod-x_parts", "KV_gone_kept"} {
+		for _, op := range []string{"CREATE", "UPDATE", "DELETE"} {
+			if !MayPublish(keeper, "$JS.API.STREAM."+op+"."+s) {
+				t.Errorf("the bus module's process may not %s %s", op, s)
+			}
+		}
+	}
+	for _, s := range []string{"EVENTS", "KV_mesh-grants_memberships", "KV_mod-x_other"} {
+		if MayPublish(keeper, "$JS.API.STREAM.CREATE."+s) || MayPublish(keeper, "$JS.API.STREAM.DELETE."+s) {
+			t.Errorf("the bus module's process may define %s", s)
+		}
+	}
+	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 !keeper.AllowResponses {
+		t.Error("the bus module's process cannot answer what it serves")
+	}
+	for user, p := range perms {
+		if user == "a.nats.keeper" {
+			continue
+		}
+		for _, v := range BrokerActingVerbs {
+			subject := "mesh.seat.mesh-broker.tool." + v
+			if MayPublish(p, subject) || MaySubscribe(p, subject) {
+				t.Errorf("%s holds %s", user, subject)
+			}
+		}
+		if MayPublish(p, "$JS.API.STREAM.CREATE.KV_mod-x_records") && user != "a.mesh-grants" {
+			t.Errorf("%s may create mod-x's bus state", user)
+		}
+	}
+}
@@ -0,0 +1,128 @@
+package buscompose
+
+import (
+	"slices"
+	"sort"
+	"strings"
+
+	"git.novox.be/novox/mesh-sdk/go/catalogue"
+)
+
+// **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.
+//
+// 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.
+
+// The verbs of the seat mesh-broker (design 25 § 12.3), until its definition is generated in mesh-contracts: the four
+// reading verbs any caller its role grants may ask, and the four acting verbs the bus module's process alone serves.
+var (
+	BrokerReadingVerbs = []string{"list", "show", "check", "orphans"}
+	BrokerActingVerbs  = []string{"create", "apply", "recreate", "drop"}
+)
+
+// BrokerVerbs are the eight.
+func BrokerVerbs() []string {
+	return append(append([]string(nil), BrokerReadingVerbs...), BrokerActingVerbs...)
+}
+
+// brokerSeat is the seat mesh-broker as its verbs' subjects name it: held once, for the mesh.
+var brokerSeat = Seat{Name: catalogue.BrokerSeat}
+
+// BrokerVerbSubjects are where the seat's verbs are asked: each verb's tool subject, and its caller-named form (issue
+// 365).
+func BrokerVerbSubjects(verbs []string) []string {
+	var out []string
+	for _, v := range verbs {
+		out = append(out, seatToolSubject(brokerSeat, v, ""), seatSubject(brokerSeat, CallKind, v)+".>")
+	}
+	sort.Strings(out)
+	return out
+}
+
+// BrokerActingOnly are the subjects of the seat's acting verbs, which no principal publishes on, however wide its
+// grant: the bus module's process serves them, and a person's ask reaches it on its own tool subject instead
+// (design 25 § 12.3).
+func BrokerActingOnly() []string { return BrokerVerbSubjects(BrokerActingVerbs) }
+
+// BusStateStreams are the streams one bus-state entry of module is kept in on the bus, by the mesh's names
+// `<module>_<local>` (design 25 § 12.1): a stream's own; a key-value bucket's KV_ stream; an object store's OBJ_
+// stream; a ledger's history stream, its snapshot stream, and its state and index buckets' KV_ streams. A name that is
+// not one plain subject token names nothing.
+func BusStateStreams(module string, e catalogue.BusStateEntry) []string {
+	base := module + "_" + e.Name
+	if !safeSubject.MatchString(base) || strings.Contains(e.Name, "_") || strings.Contains(module, "_") {
+		return nil
+	}
+	var out []string
+	switch e.Storage {
+	case catalogue.StorageKeyValue:
+		out = []string{"KV_" + base}
+	case catalogue.StorageObjectStore:
+		out = []string{"OBJ_" + base}
+	case catalogue.StorageLedger:
+		out = []string{base, base + "_snapshot", "KV_" + base + "_state"}
+		for _, t := range e.Types {
+			for _, ix := range t.Indexes {
+				if safeSubject.MatchString(ix.Name) {
+					out = append(out, "KV_"+base+"_index-"+ix.Name)
+				}
+			}
+		}
+	default:
+		out = []string{base}
+	}
+	sort.Strings(out)
+	return slices.Compact(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.
+func BusStateNames(asked map[string][]catalogue.BusStateEntry, bound map[string][]string) []string {
+	var out []string
+	for module, entries := range asked {
+		for _, e := range entries {
+			out = append(out, BusStateStreams(module, e)...)
+		}
+	}
+	for _, names := range bound {
+		for _, n := range names {
+			if safeSubject.MatchString(n) {
+				out = append(out, n)
+			}
+		}
+	}
+	sort.Strings(out)
+	return slices.Compact(out)
+}
+
+// busStateKeeperGrants are what the bus module's process publishes to keep the bus state named: each stream's create,
+// update, information, deletion and snapshot, and the stream list it finds orphans in.
+func busStateKeeperGrants(names []string) []string {
+	out := []string{"$JS.API.STREAM.NAMES", "$JS.API.STREAM.LIST"}
+	for _, n := range names {
+		for _, op := range []string{"CREATE", "UPDATE", "INFO", "DELETE", "SNAPSHOT"} {
+			out = append(out, "$JS.API.STREAM."+op+"."+n)
+		}
+	}
+	return out
+}
+
+// keepsBusState is the bus module's process defining one of the bus-state streams it keeps: the writers table's
+// exemption from stream definitions being the grants part's (ADR 0595 §4).
+func keepsBusState(p Principal, pattern string) bool {
+	if p.Kind != KindProcess || len(p.KeepsBusState) == 0 {
+		return false
+	}
+	for _, op := range []string{"CREATE", "UPDATE", "DELETE"} {
+		if name, ok := strings.CutPrefix(pattern, "$JS.API.STREAM."+op+"."); ok {
+			return slices.Contains(p.KeepsBusState, name)
+		}
+	}
+	return false
+}
@@ -110,6 +110,11 @@ 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
+	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>.
 	Process string
@@ -894,8 +899,11 @@ 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.
+		// 3. Seats it holds: full participation. But mesh-broker's verbs, which its module's process serves (busstate.go).
 		for _, s := range p.Holds {
+			if s.Name == catalogue.BrokerSeat {
+				s.Serves = nil
+			}
 			if s.isNewTraffic() {
 				// Composed by SeatTrafficOf below, worker and all; only its tools are served here.
 				for _, t := range s.Serves {
@@ -991,6 +999,10 @@ 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).
+				if s.Name == catalogue.BrokerSeat {
+					continue
+				}
 				for _, t := range s.Serves {
 					sub = append(sub, seatToolSubject(s, t, p.Node))
 				}
@@ -1157,6 +1169,13 @@ func PermissionsFor(p Principal) (Permissions, error) {
 			}
 		}
 	}
+	// And the seat mesh-broker's acting verbs to every principal whose grant reaches them (ADR 0595 §5): served by the bus
+	// module's process alone, asked by nobody on the bus.
+	for _, only := range BrokerActingOnly() {
+		if slices.ContainsFunc(pub, func(a string) bool { return SubjectsOverlap(a, only) }) {
+			deny = append(deny, only)
+		}
+	}
 	// And the vault's enrolment verbs to every principal but mesh-enrolment's (novox/hq ADR 0587 §1).
 	if p.VaultEnrols && !((p.Kind == KindModule || p.Kind == KindProcess) && p.Module == delivery.EnrolmentModule) {
 		for _, only := range EnrolmentOnly() {
@@ -1177,7 +1196,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 == KindProvisioner || (p.Kind == KindProcess && p.ServesBroker),
 	}, nil
 }

@@ -52,5 +52,12 @@ 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 {
+		pub = append(pub, busStateKeeperGrants(p.KeepsBusState)...)
+		sub = append(sub, BrokerVerbSubjects(BrokerVerbs())...)
+	}
+
 	return pub, sub, nil
 }
@@ -131,6 +131,8 @@ type ProcessPrincipal struct {
 	// Invokes are the tools and seat verbs this process calls, each <module>.<tool> or a role's
 	// seat:<seat>.<verb>, or * for every tool on the mesh.
 	Invokes []string
+	// Tools says it is a tools bundle's account, run by the node's tool runtime, not a process the node runs.
+	Tools bool
 }

 // Records is what composing a user list needs to know about the mesh, and nothing more.
@@ -164,6 +166,11 @@ type Records struct {
 	// Announced is each module's last announcement of the states it keeps (novox/hq ADR 0495, issue 1790), as the
 	// controller keeps it; a module absent has announced nothing yet. Nil where none was read.
 	Announced map[string]announce.States
+	// BusState is every bus-state entry each manifest asks for in `requires`, by module, and Bound the streams of each
+	// entry the bus module said it created and has not dropped, by module: what the bus module's process may define
+	// (busstate.go, ADR 0595 §4).
+	BusState map[string][]catalogue.BusStateEntry
+	Bound    map[string][]string
 }

 // Users is every user the composed file should contain, in the order it will be written.
@@ -232,6 +239,11 @@ 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).
+				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)
+				}
 				// The ended controller's leftovers include each node's declaration files' bucket (ADR 0590 clause 6).
 				if slices.Contains(pr.Removes, catalogue.RemovesControllerLeftovers) {
 					out[len(out)-1].FilesFor = sortedCopy(r.Nodes)
@@ -40,7 +40,7 @@ type WriterRow struct {
 // 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
 // the ended controller's leftover streams (ADR 0590 clause 6).
 func createsDeclared(p Principal, pattern string) bool {
-	if removesLeftover(p, pattern) {
+	if removesLeftover(p, pattern) || keepsBusState(p, pattern) {
 		return true
 	}
 	name, ok := strings.CutPrefix(pattern, "$JS.API.STREAM.CREATE.")
@@ -151,7 +151,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, 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); the bus module's process creates, updates and deletes every module's bus state, each stream by its name (novox/hq ADR 0595 §4)"},
 	{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"},
@@ -196,3 +196,52 @@ func OwnHears(s Store, node string) ([]membership.HeardIssued, error) {
 	}
 	return m.Hears, nil
 }
+
+// boundStreams are, by module, the streams on the bus of every binding kept (BoundKey): what the bus module's process
+// may still define, an orphan's among them (buscompose.BusStateNames).
+func (p *Part) boundStreams() (map[string][]string, error) {
+	keys, err := p.Store.Keys(StateMemberships)
+	if err != nil {
+		return nil, fmt.Errorf("its own %s cannot be listed: %w", StateMemberships, err)
+	}
+	out := map[string][]string{}
+	for _, key := range keys {
+		module, ok := strings.CutPrefix(key, BoundPrefix)
+		if !ok {
+			continue
+		}
+		b, err := p.bound(module)
+		if err != nil {
+			return nil, err
+		}
+		for _, bind := range b.Bindings {
+			out[module] = append(out[module], bindingStreams(bind)...)
+		}
+	}
+	return out, nil
+}
+
+// bindingStreams are the streams on the bus a binding names: a stream's, a bucket's KV_, an object store's OBJ_, a ledger's
+// history, its snapshot stream and its state and index buckets.
+func bindingStreams(b membership.BusBinding) []string {
+	var out []string
+	if b.Stream != "" {
+		out = append(out, b.Stream)
+	}
+	if b.Bucket != "" {
+		out = append(out, "KV_"+b.Bucket)
+	}
+	if b.ObjectStore != "" {
+		out = append(out, "OBJ_"+b.ObjectStore)
+	}
+	if b.History != "" {
+		out = append(out, b.History, b.History+"_snapshot")
+	}
+	if b.State != "" {
+		out = append(out, "KV_"+b.State)
+	}
+	for _, ix := range b.Indexes {
+		out = append(out, "KV_"+ix)
+	}
+	return out
+}
@@ -107,6 +107,9 @@ type Inputs struct {
 	Placed map[string]delivery.Placement
 	// Manifests is each module's manifest at the indexer's trunk, by module.
 	Manifests map[string]catalogue.Manifest
+	// Bound is, by module, the streams of each bus-state entry the bus module said it created and has not dropped, from
+	// the part's own state memberships (bindings.go): set by the pass, never read from another part.
+	Bound map[string][]string
 	// Digests is each module's manifest digest, by module.
 	Digests map[string]string
 	// Fingerprints is the fingerprint of the new form the indexer's trunk value names for each module, by module: what a
@@ -268,6 +268,11 @@ func (p *Part) Compose() Result {
 	}()
 	in := ReadInputs(p.Read, &p.cache)
 	p.keepBrokers(in)
+	if bound, err := p.boundStreams(); err != nil {
+		res.Errors = append(res.Errors, err.Error())
+	} else {
+		in.Bound = bound
+	}
 	for _, g := range in.Refused {
 		res.Refused = append(res.Refused, g.Input+": "+g.Why)
 	}
@@ -41,7 +41,14 @@ func records(in Inputs, union bool) (buscompose.Records, []Gap) {
 	out := buscompose.Records{Assigned: map[string][]buscompose.Declared{}, People: map[string][]string{},
 		Interchangeable: map[string]bool{}, Enrolling: slices.Clone(in.Enrolling),
 		// And each node's logins (ADR 0583 §1): both while it rotates.
-		NodeLogins: in.NodeLogins}
+		NodeLogins: in.NodeLogins,
+		// What the bus module's process may define: every manifest's bus state, and what it said it created (DLV-G2).
+		BusState: map[string][]catalogue.BusStateEntry{}, Bound: in.Bound}
+	for module, m := range in.Manifests {
+		if len(m.BusState) > 0 {
+			out.BusState[module] = m.BusState
+		}
+	}
 	values := func(provider, setting string) []any {
 		var vs []any
 		for _, node := range in.Nodes {
@@ -370,7 +377,7 @@ func processesOf(m catalogue.Manifest) []buscompose.ProcessPrincipal {
 	// what it may remove from the bus (ADR 0564 Limits 4, issue 1952 S5).
 	for _, a := range accounts {
 		pr := buscompose.ProcessPrincipal{Process: a.ID, Emits: a.Bus.Emits, State: a.Bus.State, Asks: a.Bus.Asks,
-			Invokes: a.Bus.Invokes, Removes: a.Bus.Removes}
+			Invokes: a.Bus.Invokes, Removes: a.Bus.Removes, Tools: a.Tools}
 		for _, q := range queues {
 			if q.Process == a.ID {
 				pr.Queues = append(pr.Queues, q)