Merge pull request 'The build machine takes work on the bus its credential names, and the work queue has a taker' (#107) from feat/the-build-machine-takes-work-on-nats into main

Two halves of one gap the first build over the new bus met. The machine decided
its bus from a variable its container never received, so the credential the mesh
sealed to it went unread; a credential for the new bus names the bus by scheme and
carries user, password and fingerprint beside the address, and that is enough to
dial it, pinned. And the roles' work queues were raised with no holders, so the
consumer a machine binds to take work was never created: the holders are read
from the catalogue and the handover record, as the resolver reads them.

cmd/mesh-builder/main.go

@@ -135,6 +135,16 @@ func run() error {
 // machine told about both would take work from one and answer on the other, and every log line would
 // say it was fine.
 func takeWorkFrom(credential Credential, on string) (link.BuildMachine, error) {
+	// **The credential decides, before any variable does.** A machine moved to the new bus was
+	// handed a credential for it and nothing else changed in its environment; that credential
+	// names the bus by scheme, so it is enough to know which bus to take work from.
+	if credential.onTheNewBus() {
+		js, err := broker.DialPinned(credential.natsURL(), credential.Fingerprint)
+		if err != nil {
+			return nil, err
+		}
+		return link.MachineOverNATS(js, on), nil
+	}
 	address, onNATS, err := broker.OnNATS()
 	if err != nil {
 		return nil, err
@@ -461,9 +471,25 @@ func brokerFrom() (Credential, error) {
 // out of band — here, sealed with the credential — and the endpoint is verified once at connect.
 type Credential struct {
 	URL         string `json:"url"`
-	// Fingerprint is SHA-256 over the broker certificate's DER bytes, or empty to verify the
-	// ordinary way.
 	Fingerprint string `json:"fingerprint,omitempty"`
+	// User and Password ride beside the address on the bus being built (design 25): a credential
+	// embedded in a URL leaks into every log line that prints a connection, so the mesh seals them
+	// as two fields and this machine joins them once, here, to dial.
+	User     string `json:"user,omitempty"`
+	Password string `json:"password,omitempty"`
+}
+
+// onTheNewBus is whether a credential is for the bus being built: its address says so, and the
+// mesh only ever seals such a credential with the user and password beside it.
+func (c Credential) onTheNewBus() bool { return strings.HasPrefix(strings.TrimSpace(c.URL), "nats://") }
+
+// natsURL is the address with this machine's credential in it, for the one dial that needs it.
+func (c Credential) natsURL() string {
+	rest := strings.TrimPrefix(strings.TrimSpace(c.URL), "nats://")
+	if c.User == "" {
+		return "nats://" + rest
+	}
+	return "nats://" + c.User + ":" + c.Password + "@" + rest
 }
 
 // dial opens the connection, pinning the broker's certificate when there is one to pin.

cmd/mesh-controller/push.go

@@ -761,10 +761,52 @@ func raiseTheBus(ctx context.Context, inv *inventory.Inventory, address string)
 	// The work queues of the mesh's own roles (novox/hq ADR 0121). The queue before the holder,
 	// deliberately: work queues until somebody arrives to do it, so assigning a build machine a week
 	// after something started asking for builds flushes the backlog instead of having lost it.
-	if err := broker.RaiseSeats(js, inventory.MeshSeats(), nil); err != nil {
+	// With the seats' holders, so each role's work queue gets the consumer its holder takes
+	// work from. Passed as nil until the first live raise, which left the build machine bound to a
+	// consumer nothing had created (2026-09-28).
+	holders, err := seatHolders(ctx, inv)
+	if err != nil {
+		return err
+	}
+	if err := broker.RaiseSeats(js, inventory.MeshSeats(), holders); err != nil {
 		return err
 	}
 	fmt.Printf("the bus at %s has its streams, and %d machine(s) can hear a declaration\n",
-		address, len(names))
+		broker.BareAddress(address), len(names))
 	return nil
 }
+
+// seatHolders is who holds each of the mesh's seats, by seat name: the record where a handover
+// wrote one, and the assigned module claiming the seat otherwise — the same derivation the
+// resolver makes, read from the catalogue rather than re-resolved.
+func seatHolders(ctx context.Context, inv *inventory.Inventory) (map[string]broker.Holder, error) {
+	out := map[string]broker.Holder{}
+	entries, err := inv.Catalogued(ctx)
+	if err != nil {
+		return nil, err
+	}
+	for _, e := range entries {
+		if len(e.On) == 0 {
+			continue
+		}
+		for _, c := range e.Manifest.Claims {
+			seat, known := catalogue.SeatNamed(c.Name)
+			if !known {
+				continue
+			}
+			if _, taken := out[seat.Name]; !taken {
+				out[seat.Name] = broker.Holder{Node: e.On[0], Module: e.Manifest.Module}
+			}
+		}
+	}
+	recorded, err := inv.Holdings(ctx)
+	if err != nil {
+		return nil, err
+	}
+	for _, h := range recorded {
+		if seat, known := catalogue.SeatNamed(h.Claim); known {
+			out[seat.Name] = broker.Holder{Node: h.Node, Module: h.Module}
+		}
+	}
+	return out, nil
+}

internal/broker/jetstream.go

@@ -42,6 +42,12 @@ func Dial(url string, opts ...nats.Option) (*JetStream, error) {
 		}
 		opts = append(opts, nats.Secure(pinned))
 	}
+	// **Its own inbox, and nothing wider.** Every principal is granted `_INBOX.<its user>.>` and
+	// no other inbox; the client's default prefix is random, and the server refused the first
+	// subscription to it (2026-09-28). The user is in the URL, so the prefix follows from it.
+	if user, _, _ := CredentialIn(url); user != "" {
+		opts = append(opts, nats.CustomInboxPrefix("_INBOX."+user))
+	}
 	// The address in an error is the address alone. The URL carries this controller's password,
 	// and an error here is written on the assumption it will be logged.
 	where := BareAddress(url)
@@ -65,6 +71,20 @@ func pinnedTo(path string) (*tls.Config, error) {
 	if err != nil {
 		return nil, err
 	}
+	return PinnedToFingerprint(want), nil
+}
+
+// DialPinned is Dial with the server's certificate pinned by a fingerprint the caller already holds
+// — a module or a build machine that was handed one beside its credential, and has no file.
+func DialPinned(url, fingerprint string, opts ...nats.Option) (*JetStream, error) {
+	if strings.TrimSpace(fingerprint) != "" {
+		opts = append(opts, nats.Secure(PinnedToFingerprint(fingerprint)))
+	}
+	return Dial(url, opts...)
+}
+
+// PinnedToFingerprint accepts exactly the certificate with this SHA-256 and no other.
+func PinnedToFingerprint(want string) *tls.Config {
 	return &tls.Config{
 		InsecureSkipVerify: true, //nolint:gosec // replaced by the pin below, which is stricter
 		MinVersion:         tls.VersionTLS12,
@@ -79,7 +99,7 @@ func pinnedTo(path string) (*tls.Config, error) {
 			}
 			return nil
 		},
-	}, nil
+	}
 }
 
 // Conn is the connection itself, for what the mesh keeps off JetStream on purpose — a heartbeat,

internal/broker/nats.go

@@ -169,7 +169,11 @@ func PermissionsFor(p Principal) (Permissions, error) {
 		// The controller owns the mesh's own traffic and the streams. It is the only writer of
 		// stream definitions (design 25 §3), so it alone reaches the JetStream API.
 		pub = []string{"mesh.control.>", "mesh.node.>", "$JS.API.>"}
-		sub = []string{"mesh.control.>", "$JS.API.>"}
+		// **And where its consumers deliver.** A push consumer delivers on `_DELIVER.<its name>`,
+		// and a client bound to it subscribes exactly that; the server refused it for every
+		// principal the first time one bound a consumer (2026-09-28). Each kind below is granted
+		// its own consumers' delivery subjects and no other's.
+		sub = []string{"mesh.control.>", "$JS.API.>", "_DELIVER." + ControllerName, "_DELIVER." + ControllerName + ".>"}
 
 		// Work the mesh's own flows submit to a role, and the outcomes they wait on (ADR 0121). A
 		// build is the one today: the controller asks, and reads the answer from the seat's event
@@ -252,7 +256,7 @@ func PermissionsFor(p Principal) (Permissions, error) {
 			"mesh.control." + p.Node + ".>",
 			"$JS.API.CONSUMER.INFO.NODES." + p.Node,
 		}
-		sub = []string{"mesh.node." + p.Node + ".declare"}
+		sub = []string{"mesh.node." + p.Node + ".declare", "_DELIVER." + p.Node}
 
 	case KindModule:
 		// 1. Its own namespace: it publishes its events there and serves its tools there. Nothing
@@ -285,7 +289,12 @@ func PermissionsFor(p Principal) (Permissions, error) {
 		}
 
 		// 3. Seats it holds: full participation.
+		// Its consumer's name, not ConsumerFor: that asks for these permissions to build the
+		// consumer, and would ask forever. A subject for a consumer that turns out not to exist
+		// grants nothing anybody can use.
+		sub = append(sub, "_DELIVER."+consumerDurable(p))
 		for _, s := range p.Holds {
+			sub = append(sub, "_DELIVER.SEAT_"+upperSnake(s.Name)+"_worker")
 			for _, a := range s.Accepts {
 				sub = append(sub, seatSubject(s, "accept", a))
 			}
@@ -521,7 +530,10 @@ func ComposeAccounts(principals []Principal) (string, error) {
 	// One account for the mesh: accounts in NATS isolate subject spaces entirely, and the mesh is
 	// one space (design 25 §4). The cost of that — that permissions are the only isolation — is
 	// paid in the scoping of every inbox and every ack subject.
-	b.WriteString("accounts {\n  MESH {\n    users = [\n")
+	// JetStream is enabled per account once accounts exist at all: with only the global block set,
+	// a user in MESH is told "JetStream not enabled for account" the first time it binds a
+	// consumer, which is the first thing every host does (2026-09-28).
+	b.WriteString("accounts {\n  MESH {\n    jetstream: enabled\n    users = [\n")
 	for _, p := range sorted {
 		perms, err := PermissionsFor(p)
 		if err != nil {

internal/broker/testdata/composed.conf

@@ -21,10 +21,11 @@ jetstream {
 
 accounts {
   MESH {
+    jetstream: enabled
     users = [
       { user: "controller", password: "$2a$11$cccccccccccccccccccccc", permissions: {
         publish: { allow: ["$JS.ACK.CONTROL.controller.>", "$JS.ACK.EVENTS.controller.>", "$JS.API.>", "_INBOX.enrol.>", "mesh.control.>", "mesh.node.>", "mesh.seat.mesh-build-machine.accept.>"] }
-        subscribe: { allow: ["$JS.API.>", "_INBOX.controller.>", "mesh.control.>", "mesh.mod.mesh-catalog.event.catching-up", "mesh.mod.mesh-catalog.event.upgraded", "mesh.seat.mesh-build-machine.event.built"] }
+        subscribe: { allow: ["$JS.API.>", "_DELIVER.controller", "_DELIVER.controller.>", "_INBOX.controller.>", "mesh.control.>", "mesh.mod.mesh-catalog.event.catching-up", "mesh.mod.mesh-catalog.event.upgraded", "mesh.seat.mesh-build-machine.event.built"] }
         allow_responses: { max: 1, ttl: "1m" }
       } }
       { user: "enrol.one", password: "$2a$11$eeeeeeeeeeeeeeeeeeeeee", permissions: {
@@ -33,20 +34,20 @@ accounts {
       } }
       { user: "node.one", password: "$2a$11$nnnnnnnnnnnnnnnnnnnnnn", permissions: {
         publish: { allow: ["$JS.ACK.NODES.one.>", "$JS.API.CONSUMER.INFO.NODES.one", "mesh.control.one.>"] }
-        subscribe: { allow: ["_INBOX.node.one.>", "mesh.node.one.declare"] }
+        subscribe: { allow: ["_DELIVER.one", "_INBOX.node.one.>", "mesh.node.one.declare"] }
       } }
       { user: "one.telegram", password: "$2a$11$tttttttttttttttttttttt", permissions: {
         publish: { allow: ["$JS.ACK.EVENTS.one_telegram.>", "mesh.seat.telegram-sender.event.delivered", "mesh.seat.telegram-sender.event.failed"] }
-        subscribe: { allow: ["_INBOX.one.telegram.>", "mesh.mod.telegram.tool.status", "mesh.seat.telegram-sender.accept.send"] }
+        subscribe: { allow: ["_DELIVER.SEAT_TELEGRAM_SENDER_worker", "_DELIVER.one_telegram", "_INBOX.one.telegram.>", "mesh.mod.telegram.tool.status", "mesh.seat.telegram-sender.accept.send"] }
         allow_responses: { max: 1, ttl: "1m" }
       } }
       { user: "two.audit", password: "$2a$11$aaaaaaaaaaaaaaaaaaaaaa", permissions: {
         publish: { allow: ["$JS.ACK.EVENTS.two_audit.>"] }
-        subscribe: { allow: ["_INBOX.two.audit.>", "mesh.mod.shop.event.order.placed"] }
+        subscribe: { allow: ["_DELIVER.two_audit", "_INBOX.two.audit.>", "mesh.mod.shop.event.order.placed"] }
       } }
       { user: "two.shop", password: "$2a$11$ssssssssssssssssssssss", permissions: {
         publish: { allow: ["$JS.ACK.EVENTS.two_shop.>", "mesh.mod.shop.event.order.placed", "mesh.seat.telegram-sender.accept.send"] }
-        subscribe: { allow: ["_INBOX.two.shop.>"] }
+        subscribe: { allow: ["_DELIVER.two_shop", "_INBOX.two.shop.>"] }
       } }
     ]
   }

internal/broker/users_test.go

@@ -186,7 +186,13 @@ func TestWhatTheMeshWritesIsUsersAndNothingAboutTheServer(t *testing.T) {
 	}
 	// None of the server's own settings. Each of these in the mesh's file is a value the controller
 	// would then own, and the module could no longer change its own image without the mesh agreeing.
-	for _, absent := range []string{"port:", "http:", "jetstream", "tls {", "store_dir", "cert_file"} {
+	// `jetstream {` is the server's block (its store, its limits); `jetstream: enabled` inside the
+	// account is the account's, and the mesh owns the account — a user in it is told "JetStream
+	// not enabled for account" without it (2026-09-28).
+	if !strings.Contains(got, "jetstream: enabled") {
+		t.Errorf("the account does not enable JetStream, so no user in it can bind a consumer")
+	}
+	for _, absent := range []string{"port:", "http:", "jetstream {", "tls {", "store_dir", "cert_file"} {
 		if strings.Contains(got, absent) {
 			t.Errorf("the accounts file contains %q, which belongs to the module that raises the "+
 				"server, not to the mesh", absent)