mesh-tools/node-tools/internal/runtime/runtime.go

// Package runtime is the node's tool runtime (novox/hq ADR 0175, ADR 0193), the Go port of
// node-tools' runtime.ts in its launch-only form: it launches every assigned module's bundle,
// serves each module's tools on that module's subjects and each held seat's verbs on the seat's,
// and answers for every module the verb that says what it serves. It imports nothing and knows no
// language.
package runtime

import (
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"sort"
	"strings"
	"sync"
	"time"

	"github.com/novox/mesh-tools/node-tools/internal/announce"
	"github.com/novox/mesh-tools/node-tools/internal/bus"
	"github.com/novox/mesh-tools/node-tools/internal/launch"
)

// ToolsVerb is the verb every module's runtime answers for it (ADR 0152): its tools, from the code
// that answers them.
const ToolsVerb = "tools"

// Words the mesh sets for the runtime (ADR 0175, ADR 0192).
const (
	ToolModules     = "MESH_TOOL_MODULES"
	ToolEnv         = "MESH_TOOL_ENV"
	OperatorAccount = "MESH_OPERATOR_ACCOUNT"
	OperatorHome    = "MESH_OPERATOR_HOME"
)

// Served is one module this runtime serves and its entrypoints.
type Served struct {
	Module      string
	Entrypoints []string
}

// ToolsAnswer is what `tools` answers for one module.
type ToolsAnswer struct {
	Module string       `json:"module"`
	Tools  []ListedTool `json:"tools"`
	Failed string       `json:"failed,omitempty"`
}

// ListedTool is one tool as a module's `tools` answer lists it.
type ListedTool struct {
	Name        string          `json:"name"`
	Description string          `json:"description"`
	Input       json.RawMessage `json:"input"`
	Subjects    []string        `json:"subjects,omitempty"`
	// Effect is the tool's mark where the runtime knows it (novox/hq ADR 0387 §1): its own tools, the ledger repairs.
	Effect string `json:"effect,omitempty"`
}

// NodeRuntimeModule is the module that is a machine's runtime, serving every module on it.
const NodeRuntimeModule = "node-tools"

// runtimeOnly are the words of the runtime's own environment no bundle is given.
var runtimeOnly = map[string]bool{ToolEnv: true, ToolModules: true, "MESH_BROKER_FILE": true,
	"MESH_BROKER_URL": true, "MESH_CONSOLE_LISTEN": true, LedgersFile: true}

// ServedModulesFrom reads MESH_TOOL_MODULES: `<module>=<entrypoint>` entries, comma-separated, several
// per module. The one-module form — a bare path, or the runtime's own module — is the per-module
// containers' (to-be 38 WP4c) and refused here: the node's runtime imports nothing.
func ServedModulesFrom(spec, own string) ([]Served, error) {
	order := []string{}
	by := map[string][]string{}
	for _, raw := range strings.Split(spec, ",") {
		entry := strings.TrimSpace(raw)
		if entry == "" {
			continue
		}
		module, path, ok := strings.Cut(entry, "=")
		module, path = strings.TrimSpace(module), strings.TrimSpace(path)
		if !ok || module == "" || path == "" {
			return nil, fmt.Errorf("%s: %q is not <module>=<entrypoint>; the runtime launches the bundles it "+
				"is given and imports nothing (novox/hq ADR 0193)", ToolModules, entry)
		}
		// The node's runtime serves the machine's modules and never itself. **A runtime on a module's own
		// account serves that module and nothing else** (novox/hq ADR 0259 §8): the router and a channel that
		// proves its sender reach the bus on an account of their own, never the machine's runtime.
		switch {
		case own == NodeRuntimeModule && module == own:
			return nil, fmt.Errorf("%s: %q is the runtime itself; it launches the bundles it is given and "+
				"imports nothing (novox/hq ADR 0193)", ToolModules, entry)
		case own != NodeRuntimeModule && module != own:
			return nil, fmt.Errorf("%s: %q is another module's; a runtime on %s's own account serves %s alone "+
				"(novox/hq ADR 0259)", ToolModules, entry, own, own)
		}
		if _, seen := by[module]; !seen {
			order = append(order, module)
		}
		by[module] = append(by[module], path)
	}
	out := make([]Served, 0, len(order))
	for _, m := range order {
		out = append(out, Served{Module: m, Entrypoints: by[m]})
	}
	return out, nil
}

// TakeToolEnvs reads the composed environments (ADR 0192) and removes them from the process's, so no
// bundle finds another's there.
func TakeToolEnvs() (map[string]map[string]string, error) {
	raw := os.Getenv(ToolEnv)
	os.Unsetenv(ToolEnv)
	out := map[string]map[string]string{}
	if raw == "" {
		return out, nil
	}
	var parsed map[string]map[string]any
	if err := json.Unmarshal([]byte(raw), &parsed); err != nil {
		return nil, fmt.Errorf(`%s is not JSON of the shape {"<module>": {"<word>": "<value>"}}: %w`, ToolEnv, err)
	}
	for module, words := range parsed {
		own := map[string]string{}
		for k, v := range words {
			if s, ok := v.(string); ok {
				own[k] = s
			} else {
				b, _ := json.Marshal(v)
				own[k] = string(b)
			}
		}
		out[module] = own
	}
	return out, nil
}

type registration struct {
	module string // the module's own name, or a seat's
	owner  string // the module whose bundle made it
	tools  []launch.Tool
	// repairs says it is the runtime's own ledger repairs for the module, not a bundle's (ledgers.go).
	repairs bool
}

// withRepairs is regs with the module's ledger repairs as its binding now says: added when it binds a ledger, removed
// when it binds none.
func withRepairs(regs []registration, repairs *registration) []registration {
	out := make([]registration, 0, len(regs)+1)
	for _, r := range regs {
		if !r.repairs {
			out = append(out, r)
		}
	}
	if repairs != nil {
		out = append(out, *repairs)
	}
	return out
}

// repairsOf is a module's ledger repairs as a registration, nil when its membership binds no ledger.
func (r *runner) repairsOf(module string) *registration {
	if !hasLedger(r.conn, module) {
		return nil
	}
	return &registration{module: module, owner: module, tools: r.events.repairTools(module), repairs: true}
}

// refreshRepairs serves each module's ledger repairs as its membership binds a ledger now, re-serving its tools where
// that changed (a binding is written when the bus module created the ledger, design 25 § 12.2).
func (r *runner) refreshRepairs() {
	r.mu.Lock()
	defer r.mu.Unlock()
	changed := false
	for _, m := range r.modules {
		l := r.of[m]
		served := false
		for _, reg := range l.own {
			served = served || reg.repairs
		}
		repairs := r.repairsOf(m)
		if served == (repairs != nil) {
			continue
		}
		for i := len(l.serving) - 1; i >= 0; i-- {
			l.serving[i]()
		}
		l.serving, l.names = nil, nil
		l.own, l.regs = withRepairs(l.own, repairs), withRepairs(l.regs, repairs)
		r.serve(l)
		changed = true
	}
	if changed {
		r.bindVerbs()
	}
}

// loadedModule is one served module as its bundles were last loaded: what they registered, what of it
// is served, and each entrypoint's signature at that load, which tells the watch that it changed.
type loadedModule struct {
	served   Served
	seen     map[string]string // entrypoint → its signature when it was loaded
	children []func()
	regs     []registration // every registration its bundles made
	own      []registration // those under the name of a module served here
	failed   string
	names    []string // the tools it serves, as module.tool
	serving  []func() // its tools' subscriptions
}

// unload stops serving the module's tools and stops its bundles.
func (l *loadedModule) unload() {
	for i := len(l.serving) - 1; i >= 0; i-- {
		l.serving[i]()
	}
	l.serving, l.names = nil, nil
	for i := len(l.children) - 1; i >= 0; i-- {
		l.children[i]()
	}
	l.children = nil
}

// runner is the runtime at work: every served module as last loaded, loaded again when its bundle
// appears, changes or goes (issue 1945, note 23).
type runner struct {
	conn     *bus.Conn
	logf     func(string, ...any)
	envFor   func(module string) []string
	events   *consumers
	modules  []string
	isServed map[string]bool

	mu    sync.Mutex
	of    map[string]*loadedModule
	verbs map[string]func() // each module's `tools` verb, where it is answered

	// fingerprintOf is the build a module's bundle serves: the fingerprint its composed environment carries (design 47 G
	// §424, §618), never the runner's own.
	fingerprintOf func(module string) string

	seatsAgain func()
	stops      []func() // the watch, the announcement and the seats, in the order they began
}

// Option sets how Run serves, beside what it is given.
type Option func(*consumers)

// WithLedgerWriters gives the runtime the connection each module's ledger history is published on (ledger_credentials.go).
func WithLedgerWriters(writer func(module string) (*bus.Conn, error)) Option {
	return func(c *consumers) { c.ledgers.writer = writer }
}

// Run launches, binds and serves, and serves each module's bundle again when it appears, changes or
// goes, with no restart. It answers a stop function.
func Run(conn *bus.Conn, served []Served, envs map[string]map[string]string, logf func(string, ...any), opts ...Option) (func(), error) {
	if account := os.Getenv(OperatorAccount); account != "" {
		home := ""
		if h := os.Getenv(OperatorHome); h != "" {
			home = " (home " + h + ")"
		}
		logf("[mesh-tools] the operator's account here is %s%s", account, home)
	}
	r := &runner{conn: conn, logf: logf, isServed: map[string]bool{}, of: map[string]*loadedModule{},
		verbs: map[string]func(){}}
	for _, s := range served {
		r.modules = append(r.modules, s.Module)
		r.isServed[s.Module] = true
		conn.Follow(s.Module)
	}
	node := conn.Node()

	base := os.Environ()
	r.envFor = func(module string) []string { return BundleEnv(base, envs[module], module, node) }
	r.fingerprintOf = func(module string) string { return envs[module][FingerprintWord] }

	// The module's events, for every child of it that subscribes (ADR 0198): one consumer per module,
	// bound the first time any of its children subscribes, each event handed to every child that did.
	r.events = &consumers{conn: conn, logf: logf, of: map[string]*moduleEvents{}}
	for _, opt := range opts {
		opt(r.events)
	}

	for _, s := range served {
		r.of[s.Module] = r.load(s)
	}
	for _, m := range r.modules {
		if err := refusal(r.of[m].own); err != nil {
			r.stop()
			return nil, err
		}
	}
	r.mu.Lock()
	var names, failedNames []string
	for _, m := range r.modules {
		r.serve(r.of[m])
		names = append(names, r.of[m].names...)
		if r.of[m].failed != "" {
			failedNames = append(failedNames, m)
		}
	}
	r.bindVerbs()
	r.mu.Unlock()

	line := fmt.Sprintf("[mesh-tools] serving %d tool(s) for %d module(s): %s", len(names), len(r.modules), orNone(names))
	if len(failedNames) > 0 {
		line += fmt.Sprintf("; not serving %s, whose bundle(s) failed to load", strings.Join(failedNames, ", "))
	}
	logf("%s", line)
	conn.OnMembership(func(bus.Membership) { go func() { r.refreshRepairs(); conn.Flush() }() })
	stopSeats, again := serveSeats(conn, r.modules, r.registrations, logf)
	r.seatsAgain = again
	r.stops = append(r.stops, stopSeats)

	// **What it serves, it announces** (novox/hq ADR 0197): the NATS services protocol's discovery,
	// answered with what is served at the moment it is asked — re-served memberships and bundles loaded
	// again included, so a bundle served after start is announced as one served at start.
	announced, err := announce.Serve(conn, announce.Service{
		Name: conn.Module(), ID: instanceOf(conn),
		Description: "the mesh's tool runtime on " + node + ": every assigned module's tools and the seats they hold",
		Metadata:    map[string]string{"node": node, announce.Capabilities: strings.Join(CallCapabilities(), ",")},
	}, func() []announce.Endpoint { return endpointsOf(conn, r.own(), r.registrations(), r.modules) })
	if err != nil {
		logf("[mesh-tools] cannot announce what it serves: %v", err)
	} else {
		r.stops = append(r.stops, announced)
	}

	// **A bundle is served as soon as it is in place** (issue 1945, note 23): the entrypoints'
	// directories are watched, and a module whose bundle appears, changes or goes is loaded again.
	var paths []string
	for _, s := range served {
		for _, entry := range s.Entrypoints {
			path, _ := filepath.Abs(entry)
			paths = append(paths, path)
		}
	}
	if len(paths) > 0 {
		stopWatch, err := watchBundles(paths, r.reconcile, logf)
		if err != nil {
			logf("[mesh-tools] cannot watch the bundles: %v; a bundle placed or changed later is served only after a restart", err)
		} else {
			r.stops = append(r.stops, stopWatch)
		}
	}
	conn.Flush()
	return r.stop, nil
}

// stop ends the watch, the announcement and the seats, unbinds the modules' consumers, and stops every
// module's tools and bundles.
func (r *runner) stop() {
	for i := len(r.stops) - 1; i >= 0; i-- {
		r.stops[i]()
	}
	r.stops = nil
	r.events.stopAll()
	r.mu.Lock()
	for _, stop := range r.verbs {
		stop()
	}
	r.verbs = map[string]func(){}
	loaded := make([]*loadedModule, 0, len(r.modules))
	for _, m := range r.modules {
		if l := r.of[m]; l != nil {
			loaded = append(loaded, l)
		}
	}
	r.mu.Unlock()
	for i := len(loaded) - 1; i >= 0; i-- {
		loaded[i].unload()
	}
}

// claimed is every seat a served module's membership names.
func (r *runner) claimed() map[string]bool {
	out := map[string]bool{}
	for _, m := range r.modules {
		if mem := r.conn.Membership(m); mem != nil {
			for _, s := range mem.Seats {
				out[s.Seat] = true
			}
		}
	}
	return out
}

// load launches one module's bundles and learns what they register; it serves nothing yet.
func (r *runner) load(s Served) *loadedModule {
	l := &loadedModule{served: s, seen: map[string]string{}}
	claimed := r.claimed()
	for _, entry := range s.Entrypoints {
		path, _ := filepath.Abs(entry)
		module := s.Module
		l.seen[path] = signatureOf(path)
		fail := func(why string) {
			l.failed = why
			r.logf("[mesh-tools] %s's bundle %s failed to load: %s; its tools are not served here", module, entry, why)
		}
		if why := notStartable(path); why != "" {
			fail(why)
			continue
		}
		child, err := launch.Start(module, path, r.envFor(module), r.events.forModule(module), r.logf)
		if err != nil {
			fail(err.Error())
			continue
		}
		// A tool named as one of the runner's ledger repairs would be taken for it, or hide it: refused at the load, naming both
		// (ADR 0597 §10).
		clashed := ""
		for _, reg := range child.Registrations {
			if clashed == "" && r.isServed[reg.Module] {
				clashed = repairClash(reg.Module, reg.Tools)
			}
		}
		if clashed != "" {
			child.Stop()
			fail(clashed)
			continue
		}
		// A bundle whose marks name a tool it does not register is not loaded: no tool of its module is served here, rather
		// than one served unmarked that the manifest meant to mark (design 46 7-approval-mark §4).
		if unserved := r.unservedMarks(module, child.Registrations); len(unserved) > 0 {
			child.Stop()
			fail(fmt.Sprintf("its marks name %s, which it does not register (design 46 7-approval-mark §4)", strings.Join(unserved, ", ")))
			continue
		}
		l.children = append(l.children, child.Stop)
		for _, reg := range child.Registrations {
			got := registration{module: reg.Module, owner: module, tools: reg.Tools}
			l.regs = append(l.regs, got)
			switch {
			case r.isServed[got.module]:
				l.own = append(l.own, got)
			case claimed[got.module]:
			default:
				r.logf(`[mesh-tools] %s registers tools under "%s", which is neither a module served here nor a seat one of them claims; not served until the mesh issues the claim`, got.owner, got.module)
			}
		}
	}
	if repairs := r.repairsOf(s.Module); repairs != nil {
		l.regs, l.own = append(l.regs, *repairs), append(l.own, *repairs)
	}
	return l
}

// refusal is why a module's registrations cannot be served: a tool named as the verb the runtime
// answers for every module, or two tools of one name.
func refusal(own []registration) error {
	for _, r := range own {
		seen := map[string]bool{}
		for _, t := range r.tools {
			if t.Name == ToolsVerb {
				return fmt.Errorf(`%s names a tool "%s", which is the verb the runtime answers for every module with what it serves (novox/hq ADR 0152) — refused, rename it`, r.module, ToolsVerb)
			}
			if seen[t.Name] {
				return fmt.Errorf("%s exposes two tools named %s — refused", r.module, t.Name)
			}
			seen[t.Name] = true
		}
	}
	return nil
}

// serve binds a loaded module's tools. r.mu is held.
func (r *runner) serve(l *loadedModule) {
	for _, reg := range l.own {
		for _, t := range reg.tools {
			t := t
			// Told who called, from the subject the call arrived on, never from its body (novox/hq issue 365).
			stop, err := r.conn.HandleCalled(reg.module+"."+t.Name, func(caller string, body json.RawMessage) (any, error) {
				return t.Run(caller, argsOf(body))
			})
			if err != nil {
				r.logf("[mesh-tools] cannot serve %s.%s: %v", reg.module, t.Name, err)
				continue
			}
			l.names = append(l.names, reg.module+"."+t.Name)
			l.serving = append(l.serving, stop)
		}
	}
}

// toolsOf is every tool served under a module's name, and why its bundle failed, if it did. r.mu is held.
func (r *runner) toolsOf(module string) ([]launch.Tool, string) {
	var tools []launch.Tool
	for _, m := range r.modules {
		for _, reg := range r.of[m].own {
			if reg.module == module {
				tools = append(tools, reg.tools...)
			}
		}
	}
	return tools, r.of[module].failed
}

// bindVerbs answers each module's `tools` verb where it serves a tool or its bundle failed, and not
// where neither is so (a pure-events module is silent, as it always was). r.mu is held.
func (r *runner) bindVerbs() {
	for _, module := range r.modules {
		module := module
		tools, why := r.toolsOf(module)
		want := len(tools) > 0 || why != ""
		stop, bound := r.verbs[module]
		switch {
		case want && !bound:
			stop, err := r.conn.Handle(module+"."+ToolsVerb, func(json.RawMessage) (any, error) {
				r.mu.Lock()
				tools, why := r.toolsOf(module)
				r.mu.Unlock()
				answer := ToolsAnswer{Module: module, Tools: []ListedTool{}, Failed: why}
				for _, t := range tools {
					answer.Tools = append(answer.Tools, ListedTool{Name: t.Name, Description: t.Description,
						Input: t.Input, Subjects: subjectsOf(r.conn, module, t.Name), Effect: t.Effect})
				}
				return answer, nil
			})
			if err == nil {
				r.verbs[module] = stop
			}
		case !want && bound:
			stop()
			delete(r.verbs, module)
		}
	}
}

// own is every registration served under a served module's name, now.
func (r *runner) own() []registration {
	r.mu.Lock()
	defer r.mu.Unlock()
	var out []registration
	for _, m := range r.modules {
		out = append(out, r.of[m].own...)
	}
	return out
}

// registrations is every registration the bundles made, now.
func (r *runner) registrations() []registration {
	r.mu.Lock()
	defer r.mu.Unlock()
	var out []registration
	for _, m := range r.modules {
		out = append(out, r.of[m].regs...)
	}
	return out
}

// reconcile loads again every module whose entrypoint appeared, changed or went since it was loaded.
// The watch calls it once a burst of changes settles; it runs on the watch's one goroutine.
func (r *runner) reconcile() {
	for _, m := range r.modules {
		r.mu.Lock()
		l := r.of[m]
		r.mu.Unlock()
		for _, entry := range l.served.Entrypoints {
			path, _ := filepath.Abs(entry)
			was, now := l.seen[path], signatureOf(path)
			if was == now {
				continue
			}
			what := "changed"
			switch {
			case was == "":
				what = "appeared"
			case now == "":
				what = "is gone"
			}
			r.reload(m, entry+" "+what)
			break
		}
	}
}

// reload stops a module's tools and bundles, loads them again, serves what they register now, and
// serves its seats' verbs again; discovery answers with it from then on.
func (r *runner) reload(module, what string) {
	r.mu.Lock()
	old := r.of[module]
	r.of[module] = &loadedModule{served: old.served, seen: old.seen}
	r.mu.Unlock()
	// Its events are refused first, so one heard while the old bundle stops is refused at once.
	r.events.dropModule(module)
	old.unload()

	r.logf("[mesh-tools] %s's bundle %s; loading it again", module, what)
	fresh := r.load(old.served)
	if err := refusal(fresh.own); err != nil {
		fresh.unload()
		fresh.regs, fresh.own, fresh.failed = nil, nil, err.Error()
		r.events.dropModule(module)
		r.logf("[mesh-tools] %s's bundle is refused: %v; its tools are not served here", module, err)
	}
	r.mu.Lock()
	r.of[module] = fresh
	r.serve(fresh)
	r.bindVerbs()
	r.mu.Unlock()
	if fresh.failed != "" {
		r.logf("[mesh-tools] not serving %s, whose bundle failed to load", module)
	} else {
		r.logf("[mesh-tools] serving %d tool(s) for %s: %s", len(fresh.names), module, orNone(fresh.names))
	}
	if r.seatsAgain != nil {
		r.seatsAgain()
	}
	r.conn.Flush()
}

// instanceOf is this runtime's instance on the bus: its machine, which is what tells two instances of
// one service apart; the connection's module where it has no machine.
func instanceOf(conn *bus.Conn) string {
	if n := conn.Node(); n != "" {
		return n
	}
	return conn.Module()
}

// endpointsOf is everything this runtime serves now: each served module's tools on every subject the
// mesh issued for them, and each held seat's verbs on the seat's subject (ADR 0197).
func endpointsOf(conn *bus.Conn, own []registration, registrations []registration, modules []string) []announce.Endpoint {
	node := conn.Node()
	var out []announce.Endpoint
	for _, r := range own {
		for _, t := range r.tools {
			served := conn.ServedOn(r.module, t.Name)
			interchangeable := false
			for _, s := range served {
				interchangeable = interchangeable || s.Subject == "mesh.mod."+r.module+".tool."+t.Name
			}
			for _, s := range served {
				out = append(out, announce.Endpoint{Kind: announce.KindTool, Module: r.module, Tool: t.Name,
					Node: node, Description: t.Description, Schema: t.Input, Interchangeable: interchangeable,
					Subject: s.Subject, Queue: s.Queue})
			}
		}
	}
	impl := map[string]map[string]launch.Tool{}
	for _, r := range registrations {
		if impl[r.module] == nil {
			impl[r.module] = map[string]launch.Tool{}
		}
		for _, t := range r.tools {
			impl[r.module][t.Name] = t
		}
	}
	have := map[string]bool{}
	for _, module := range modules {
		m := conn.Membership(module)
		if m == nil {
			continue
		}
		for _, v := range m.Seats {
			t, ok := impl[v.Seat][v.Verb]
			if !ok || have[v.Subject] {
				continue
			}
			have[v.Subject] = true
			scope := "mesh"
			if node != "" && strings.HasSuffix(v.Subject, "."+node) {
				scope = "node"
			}
			out = append(out, announce.Endpoint{Kind: announce.KindSeat, Module: module, Tool: v.Verb, Seat: v.Seat,
				Scope: scope, Node: node, Description: t.Description, Schema: t.Input, Subject: v.Subject})
		}
	}
	return out
}

func orNone(names []string) string {
	if len(names) == 0 {
		return "(none)"
	}
	return strings.Join(names, ", ")
}

// argsOf is a call's arguments as the tool receives them: an object, `{}` for none.
func argsOf(body json.RawMessage) json.RawMessage {
	trimmed := strings.TrimSpace(string(body))
	if trimmed == "" || trimmed == "null" {
		return json.RawMessage("{}")
	}
	return body
}

// subjectsOf is where a tool is answered as the mesh issued it: the plain subject first, then this
// machine's; nothing before a membership is issued.
func subjectsOf(conn *bus.Conn, module, tool string) []string {
	m := conn.Membership(module)
	if m == nil {
		return nil
	}
	var plain, mine []string
	for _, s := range m.Serves {
		subject := strings.ReplaceAll(s.Subject, "{tool}", tool)
		if s.Queue != "" {
			plain = append(plain, subject)
		} else {
			mine = append(mine, subject)
		}
	}
	return append(plain, mine...)
}

// serveSeats serves every verb of every seat a served module holds, where the mesh issued it, by
// the tool of the same name registered under the seat's name (ADR 0159, 0160) — and serves again
// whenever a membership changes, and when again is called because a bundle was loaded again. Whether
// this machine holds the seat is the bus's to decide. It answers its stop and again.
func serveSeats(conn *bus.Conn, modules []string, current func() []registration, logf func(string, ...any)) (func(), func()) {
	var mu sync.Mutex
	var stops []func()
	stopped := false
	serve := func() {
		mu.Lock()
		defer mu.Unlock()
		if stopped {
			return
		}
		for _, s := range stops {
			s()
		}
		stops = nil
		impl := map[string]map[string]launch.Tool{}
		for _, r := range current() {
			if impl[r.module] == nil {
				impl[r.module] = map[string]launch.Tool{}
			}
			for _, t := range r.tools {
				impl[r.module][t.Name] = t
			}
		}
		have := map[string]bool{}
		for _, module := range modules {
			m := conn.Membership(module)
			if m == nil {
				continue
			}
			for _, v := range m.Seats {
				if have[v.Subject] {
					continue
				}
				have[v.Subject] = true
				t, ok := impl[v.Seat][v.Verb]
				if !ok {
					logf("[mesh-tools] %s claims %s and implements no %s, which that seat promises; not served", module, v.Seat, v.Verb)
					continue
				}
				stop, err := conn.HandleSubjectCalled(v.Subject, func(caller string, body json.RawMessage) (any, error) {
					return t.Run(caller, argsOf(body))
				})
				if err != nil {
					logf("[mesh-tools] cannot serve %s's %s on %s: %v", v.Seat, v.Verb, v.Subject, err)
					continue
				}
				stops = append(stops, stop)
				logf("[mesh-tools] serving %s's %s on %s, admitted where %s holds the seat", v.Seat, v.Verb, v.Subject, module)
			}
		}
	}
	serve()
	conn.OnMembership(func(bus.Membership) { go func() { serve(); conn.Flush() }() })
	return func() {
		mu.Lock()
		defer mu.Unlock()
		for _, s := range stops {
			s()
		}
		stops = nil
		stopped = true
	}, serve
}

// consumers holds, per module the runtime serves, the one durable consumer its events arrive on and
// the children its events are handed to (novox/hq ADR 0198).
type consumers struct {
	conn *bus.Conn
	logf func(string, ...any)
	mu   sync.Mutex
	of   map[string]*moduleEvents
	// seatBound is each module's seat traffic bound once: a worker taken, a proof subject answered.
	seatBound map[string]func()
	// provisioned is every watch a module's code holds of its values at runtime (provision.go).
	provisioned provisionWatches
	// ledgers is how a module's ledger is opened (ledgers.go).
	ledgers ledgerSource
	// calls are the tool calls in flight, by the id each bundle was handed (ledgers.go, novox/hq ADR 0597 §7).
	calls callsInFlight
	// sessions are the bundles' put and get sessions (sessions.go).
	sessions sessionTable
	// gate is the approval gate every call is served through (gate.go); nil serves every call as before.
	gate Gate
}

type moduleEvents struct {
	stop     func()
	delivers []*deliverSlot
	beats    []*beatSlot
}

// slot is where a bundle's handler is kept (its event handler, its heartbeat handler); a bundle loaded again sets it
// while the consumer's loop reads it, so both go through its own lock.
type slot[F any] struct {
	mu sync.Mutex
	f  F
}

func (s *slot[F]) get() F {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.f
}

func (s *slot[F]) set(f F) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.f = f
}

type (
	deliverSlot = slot[func(json.RawMessage) error]
	beatSlot    = slot[func(json.RawMessage)]
)

// beatAsked says whether any bundle of the module asked for the heartbeat; the caller holds c.mu.
func (m *moduleEvents) beatAsked() bool {
	for _, t := range m.beats {
		if t.get() != nil {
			return true
		}
	}
	return false
}

// stopAll unbinds every module's consumer.
func (c *consumers) stopAll() {
	c.mu.Lock()
	defer c.mu.Unlock()
	for _, m := range c.of {
		if m.stop != nil {
			m.stop()
		}
	}
	c.of = map[string]*moduleEvents{}
	for _, stop := range c.seatBound {
		stop()
	}
	c.seatBound = map[string]func(){}
	c.provisioned.stopAll()
}

// dropModule unbinds one module's consumer, its seat traffic and its provision watches, so the bundle
// loaded again in its place binds them anew and no event is handed to a child that was stopped.
func (c *consumers) dropModule(module string) {
	c.mu.Lock()
	defer c.mu.Unlock()
	if m := c.of[module]; m != nil && m.stop != nil {
		m.stop()
	}
	delete(c.of, module)
	for key, stop := range c.seatBound {
		if strings.HasPrefix(key, module+" ") {
			stop()
			delete(c.seatBound, key)
		}
	}
	c.provisioned.stopModule(module)
	c.dropModuleSessions(module)
}

// forModule is the bus one launched bundle of a module reaches the mesh through.
func (c *consumers) forModule(module string) launch.Bus {
	return &moduleBus{all: c, module: module}
}

type moduleBus struct {
	all     *consumers
	module  string
	mu      sync.Mutex
	deliver *deliverSlot
	beat    *beatSlot
}

// Publish emits an event as the module (ADR 0193).
func (b *moduleBus) Publish(params json.RawMessage) error {
	var env bus.Envelope
	if err := json.Unmarshal(params, &env); err != nil {
		return fmt.Errorf("not an event envelope: %w", err)
	}
	return b.all.conn.PublishAs(b.module, env)
}

// Ask calls a tool as the module: `{key, body}`, answered with the tool's result (ADR 0198).
func (b *moduleBus) Ask(params json.RawMessage) (json.RawMessage, error) {
	var asked struct {
		Key  string          `json:"key"`
		Body json.RawMessage `json:"body"`
	}
	if err := json.Unmarshal(params, &asked); err != nil || asked.Key == "" {
		return nil, fmt.Errorf("mesh/ask names no tool: {key, body}")
	}
	body := any(asked.Body)
	if len(asked.Body) == 0 {
		body = map[string]any{}
	}
	answered, err := b.all.conn.AskAs(b.module, asked.Key, body)
	if err != nil {
		return nil, err
	}
	if len(answered.Result) == 0 {
		return json.RawMessage("null"), nil
	}
	return answered.Result, nil
}

// Subscribe hands this bundle the module's events. The consumer is bound once per module; each of
// the module's children that subscribed is handed every event, and the event is acknowledged only
// when all of them took it — one consumer split between two readers would give each half.
func (b *moduleBus) Subscribe(deliver func(json.RawMessage) error, beat func(json.RawMessage)) error {
	b.mu.Lock()
	if b.deliver == nil {
		b.deliver = &deliverSlot{}
	}
	b.deliver.set(deliver)
	mine := b.deliver
	if b.beat == nil {
		b.beat = &beatSlot{}
	}
	b.beat.set(beat)
	mineBeat := b.beat
	b.mu.Unlock()

	c := b.all
	c.mu.Lock()
	defer c.mu.Unlock()
	m := c.of[b.module]
	if m == nil {
		m = &moduleEvents{}
		c.of[b.module] = m
	}
	listed := false
	for _, d := range m.delivers {
		if d == mine {
			listed = true
		}
	}
	if !listed {
		m.delivers = append(m.delivers, mine)
		m.beats = append(m.beats, mineBeat)
	}
	if m.stop != nil {
		return nil
	}
	module := b.module
	// The bus is asked about the consumer only while a bundle of the module asked for the heartbeat.
	stop, err := c.conn.ConsumeAsBeatingWhen(module, func(env bus.Envelope) error { return c.handOut(module, m, env) },
		func(b bus.Beat) { c.handBeat(m, b) }, func() bool {
			c.mu.Lock()
			defer c.mu.Unlock()
			return m.beatAsked()
		})
	if err != nil {
		return err
	}
	m.stop = stop
	c.logf("[mesh-tools] %s's events arrive on its consumer %s", module, bus.ConsumerOf(c.conn.Node(), module))
	return nil
}

// handBeat hands the word on a module's consumer to every child of the module that subscribed, as an event is.
func (c *consumers) handBeat(m *moduleEvents, b bus.Beat) {
	raw, err := json.Marshal(b)
	if err != nil {
		return
	}
	c.mu.Lock()
	targets := append([]*beatSlot(nil), m.beats...)
	c.mu.Unlock()
	for _, t := range targets {
		if f := t.get(); f != nil {
			f(raw)
		}
	}
}

// errModuleReloading refuses an event heard by a module's consumer that was stopped for a reload.
var errModuleReloading = errors.New("the module's bundle is being loaded again; offered again")

// handOut hands one event of a module's consumer m to every child of the module that subscribed; an
// event of a consumer that is no longer the module's is refused.
func (c *consumers) handOut(module string, m *moduleEvents, env bus.Envelope) error {
	raw, err := json.Marshal(env)
	if err != nil {
		return err
	}
	c.mu.Lock()
	if c.of[module] != m {
		// Heard by a consumer stopped while the module is loaded again (issue 1945): refused, so the bus
		// offers it again, to the bundle loaded in its place once it subscribes.
		c.mu.Unlock()
		return errModuleReloading
	}
	targets := append([]*deliverSlot(nil), m.delivers...)
	c.mu.Unlock()
	for _, d := range targets {
		if err := d.get()(raw); err != nil {
			c.logf("[mesh-tools] %s did not take %s%s: %s; offered again in %s", module, env.Key,
				eventRef(env), whyNotTaken(err), bus.NakDelay)
			return err
		}
	}
	return nil
}

// whyNotTaken says why a bundle did not take an event, so the line names who failed: the module's
// handler, in its own words — an error it answered — or the runtime not reaching it. A bundle's
// handler error read bare ("Unexpected end of JSON input") looked like the runtime's (issue 276).
func whyNotTaken(err error) string {
	var refused *launch.Refused
	if errors.As(err, &refused) {
		return fmt.Sprintf("its handler answered an error, in its own words: %q (an event is taken by any answer that is not an error)", refused.Message)
	}
	return err.Error()
}

// eventRef names the event a line is about by its id, when it has one, so its offers can be told
// apart from another event's.
func eventRef(env bus.Envelope) string {
	if id := env.Headers["x-event-id"]; id != "" {
		return " (event " + id + ")"
	}
	return ""
}

// stateAsked is what a bundle names when it reaches its state (ADR 0201): the state by the name its
// module uses, a key, and for a put the value.
type stateAsked struct {
	State    string          `json:"state"`
	Key      string          `json:"key"`
	Value    json.RawMessage `json:"value"`
	Revision uint64          `json:"revision"`
	// TTLSeconds gives a created or put key its own time to live (issue 1952): create makes it only when it has no
	// value, put renews it whole. Zero is none.
	TTLSeconds int `json:"ttl_seconds"`
	// After and MaxBytes are a `page`'s start key and its size in keys and values (zero is a megabyte).
	After    string `json:"after"`
	MaxBytes int    `json:"max_bytes"`
	// BudgetMS is the time a `page` may spend reading, in milliseconds (zero is the runtime's own, well under 30 s).
	BudgetMS int `json:"budget_ms"`
}

// unknownVerb is an answer with the JSON-RPC code of a method not found, which a bundle tells by its code.
type unknownVerb struct{ error }

func (unknownVerb) ErrorCode() int { return -32601 }

// State answers a bundle's `get`, `put`, `delete`, `keys`, `page`, `create`, `update`, `history` and `info` on its
// module's state (ADR 0201, ADR 0259, ADR 0307); `create` and `put` take `ttl_seconds`, a key's own time to live,
// whose end every watch hears as an `expire` (issue 1952). The runtime refuses, with the reason, a state the module was not issued and a write to one it only reads.
func (b *moduleBus) State(verb string, params json.RawMessage) (json.RawMessage, error) {
	var asked stateAsked
	if err := json.Unmarshal(params, &asked); err != nil || asked.State == "" {
		return nil, fmt.Errorf("mesh/state.%s names no state: {state, key, value}", verb)
	}
	conn := b.all.conn
	var answer any
	switch verb {
	case "get":
		entry, err := conn.StateGet(b.module, asked.State, asked.Key)
		if err != nil {
			return nil, err
		}
		if entry == nil {
			return json.RawMessage("null"), nil
		}
		answer = entry
	case "page":
		// The values after a key in key order, one request for many (a first view's page of a slow state).
		if asked.MaxBytes <= 0 {
			asked.MaxBytes = 1 << 20
		}
		page, err := conn.StatePage(b.module, asked.State, asked.After, asked.MaxBytes, time.Duration(asked.BudgetMS)*time.Millisecond)
		if err != nil {
			return nil, err
		}
		answer = page
	case "history":
		// Every value the bucket still keeps for the key, oldest first (novox/hq ADR 0307).
		kept, err := conn.StateHistory(b.module, asked.State, asked.Key)
		if err != nil {
			return nil, err
		}
		answer = kept
	case "put":
		put := conn.StatePut
		if asked.TTLSeconds != 0 {
			put = func(module, name, key string, value json.RawMessage) (uint64, error) {
				return conn.StateRenewTTL(module, name, key, value, time.Duration(asked.TTLSeconds)*time.Second)
			}
		}
		revision, err := put(b.module, asked.State, asked.Key, asked.Value)
		if err != nil {
			return nil, err
		}
		answer = map[string]any{"revision": revision}
	case "delete":
		if err := conn.StateDelete(b.module, asked.State, asked.Key); err != nil {
			return nil, err
		}
		answer = map[string]any{}
	case "info":
		// The bucket's fill against its caps, so a bundle's tools can say how full it is.
		info, err := conn.StateInfo(b.module, asked.State)
		if err != nil {
			return nil, err
		}
		answer = info
	case "keys":
		keys, err := conn.StateKeys(b.module, asked.State)
		if err != nil {
			return nil, err
		}
		answer = keys
	case "create":
		// Only when the key has no value: of two writers making it, one wins (novox/hq ADR 0259).
		create := conn.StateCreate
		if asked.TTLSeconds != 0 {
			create = func(module, name, key string, value json.RawMessage) (uint64, error) {
				return conn.StateCreateTTL(module, name, key, value, time.Duration(asked.TTLSeconds)*time.Second)
			}
		}
		revision, err := create(b.module, asked.State, asked.Key, asked.Value)
		if err != nil {
			return nil, err
		}
		answer = map[string]any{"revision": revision}
	case "update":
		// Only when the key is still at the revision read: a second answer to one ask loses (ADR 0259).
		if asked.TTLSeconds != 0 {
			return nil, fmt.Errorf("mesh/state.update gives no time to live: the bus sets one only when a key is " +
				"created; put the key with ttl_seconds to renew it whole (issue 1952)")
		}
		revision, err := conn.StateUpdate(b.module, asked.State, asked.Key, asked.Value, asked.Revision)
		if err != nil {
			return nil, err
		}
		answer = map[string]any{"revision": revision}
	default:
		return nil, unknownVerb{fmt.Errorf("the runtime answers no mesh/state.%s", verb)}
	}
	return json.Marshal(answer)
}

// logAsked is what a bundle names when it reaches its log (novox/hq ADR 0297).
type logAsked struct {
	Log    string          `json:"log"`
	Key    string          `json:"key"`
	Value  json.RawMessage `json:"value"`
	Expect *uint64         `json:"expect"`
	From   uint64          `json:"from"`
	Keys   string          `json:"keys"`
	Limit  int             `json:"limit"`
	// Since is a read's or a watch's start as an RFC 3339 time, in place of From (novox/hq ADR 0571).
	Since string `json:"since"`
}

// since is the asked start time, zero when none.
func (a logAsked) since() (time.Time, error) {
	if a.Since == "" {
		return time.Time{}, nil
	}
	at, err := time.Parse(time.RFC3339Nano, a.Since)
	if err != nil {
		return time.Time{}, fmt.Errorf("since is an RFC 3339 time (2026-10-29T09:30:00Z), not %q", a.Since)
	}
	return at, nil
}

// Log answers a bundle's `append`, `read`, `newest`, `last` and `info` on its module's log and on a log its module reads
// as `<module>.<log>` (ADR 0297, ADR 0571). The runtime
// refuses, with the reason, a log the module was not issued; an append whose `expect` is stale is
// refused with the code of a lost state update.
func (b *moduleBus) Log(verb string, params json.RawMessage) (json.RawMessage, error) {
	var asked logAsked
	if err := json.Unmarshal(params, &asked); err != nil || asked.Log == "" {
		return nil, fmt.Errorf("mesh/log.%s names no log: {log, ...}", verb)
	}
	conn := b.all.conn
	var answer any
	switch verb {
	case "append":
		seq, err := conn.LogAppend(b.module, asked.Log, asked.Key, asked.Value, asked.Expect)
		if err != nil {
			return nil, err
		}
		answer = map[string]any{"sequence": seq}
	case "read":
		since, err := asked.since()
		if err != nil {
			return nil, err
		}
		page, err := conn.LogRead(b.module, asked.Log, asked.From, since, asked.Keys, asked.Limit)
		if err != nil {
			return nil, err
		}
		answer = page
	case "newest":
		// The newest entry of each key the pattern names (mesh-sdk process.LogReader.Newest).
		entries, err := conn.LogNewest(b.module, asked.Log, asked.Keys)
		if err != nil {
			return nil, err
		}
		answer = entries
	case "last":
		entry, err := conn.LogLast(b.module, asked.Log, asked.Key)
		if err != nil {
			return nil, err
		}
		if entry == nil {
			return json.RawMessage("null"), nil
		}
		answer = entry
	case "info":
		info, err := conn.LogInfo(b.module, asked.Log)
		if err != nil {
			return nil, err
		}
		answer = info
	default:
		return nil, fmt.Errorf("the runtime answers no mesh/log.%s", verb)
	}
	return json.Marshal(answer)
}

// LogWatch hands a bundle each entry of a log, `{log, keys, from, since}`: from a sequence, from a time, or with neither
// each new one, one at a time (novox/hq ADR 0571; mesh-sdk process.LogReader.Watch).
func (b *moduleBus) LogWatch(params json.RawMessage, deliver func(json.RawMessage) error) (func(), error) {
	var asked logAsked
	if err := json.Unmarshal(params, &asked); err != nil || asked.Log == "" {
		return nil, fmt.Errorf("mesh/log.watch names no log: {log, keys, from, since}")
	}
	since, err := asked.since()
	if err != nil {
		return nil, err
	}
	return b.all.conn.LogWatch(b.module, asked.Log, asked.Keys, asked.From, since, func(e bus.LogEntry) error {
		raw, err := json.Marshal(e)
		if err != nil {
			return err
		}
		return deliver(raw)
	})
}

// Watch hands a bundle its module's state as it is and as it changes (ADR 0201): `{state, key}`, the
// key a pattern with `*` and `**`, empty for every key.
func (b *moduleBus) Watch(params json.RawMessage, deliver func(json.RawMessage) error) (func(), error) {
	var asked stateAsked
	if err := json.Unmarshal(params, &asked); err != nil || asked.State == "" {
		return nil, fmt.Errorf("mesh/state.watch names no state: {state, key}")
	}
	return b.all.conn.StateWatch(b.module, asked.State, asked.Key, func(change bus.StateChange) error {
		raw, err := json.Marshal(change)
		if err != nil {
			return err
		}
		return deliver(raw)
	})
}

// seatAsked is what a bundle names for its seat traffic (novox/hq ADR 0259 §3).
type seatAsked struct {
	Subject string          `json:"subject"`
	Body    json.RawMessage `json:"body"`
	ID      string          `json:"id"`
	Worker  string          `json:"worker"`
	Bucket  string          `json:"bucket"`
	Key     string          `json:"key"`
	// WaitMS is how long a pull waits for a message (none when 0); Fate what a settle gives one (issue 1887 A11).
	WaitMS int `json:"wait_ms"`
	// DelayMS is how long a settle with the fate nak-after waits before the bus gives the message again.
	DelayMS int64  `json:"delay_ms"`
	Fate    string `json:"fate"`
}

// Seat answers a bundle's seat traffic, each as the module and only as far as its membership lists:
// `publish` an accept or an event, `prove` (ask a proof), `record` (read a record kept for it), `take` a
// worker's work and `answer` a proof subject — the last two bound once per module and subject, each
// message or proof handed to the module's child that asked last (novox/hq ADR 0259 §3).
func (b *moduleBus) Seat(verb string, params json.RawMessage, handOn func(string, any) (json.RawMessage, error)) (json.RawMessage, error) {
	var asked seatAsked
	if err := json.Unmarshal(params, &asked); err != nil {
		return nil, fmt.Errorf("mesh/seat.%s: %w", verb, err)
	}
	conn := b.all.conn
	switch verb {
	case "publish":
		seq, duplicate, err := conn.SeatPublishSaid(b.module, asked.Subject, asked.Body, asked.ID)
		if err != nil {
			return nil, err
		}
		return json.Marshal(map[string]any{"sequence": seq, "duplicate": duplicate})
	case "prove":
		return conn.SeatProve(b.module, asked.Subject, asked.Body)
	case "kinds":
		kinds, err := conn.SeatKinds(b.module)
		if err != nil {
			return nil, err
		}
		return json.Marshal(kinds)
	case "record":
		entry, err := conn.SeatRecord(b.module, asked.Bucket, asked.Key)
		if err != nil || entry == nil {
			return json.RawMessage("null"), err
		}
		return json.Marshal(entry)
	case "pull":
		// One message from a worker the membership lists, never a prefetch: a build module pulls an ask only when it
		// has room, from its workers in their order (novox/hq ADR 0525 §4; issue 1887 A11).
		p, err := conn.SeatPull(b.module, asked.Worker, time.Duration(asked.WaitMS)*time.Millisecond)
		if err != nil || p == nil {
			return json.RawMessage("null"), err
		}
		return json.Marshal(p)
	case "settle":
		settle := func() error { return conn.SeatSettle(b.module, asked.ID, asked.Fate) }
		if asked.Fate == bus.SettleNakAfter {
			settle = func() error {
				return conn.SeatSettleAfter(b.module, asked.ID, time.Duration(asked.DelayMS)*time.Millisecond)
			}
		}
		if err := settle(); err != nil {
			return nil, err
		}
		return json.RawMessage(`{}`), nil
	case "take":
		key := "take " + asked.Worker
		if b.bound(key) {
			return json.RawMessage(`{}`), nil
		}
		stop, err := conn.SeatTake(b.module, asked.Worker, func(w bus.Work) error {
			_, err := handOn("mesh/work", map[string]any{"work": w})
			return err
		})
		if err != nil {
			return nil, err
		}
		b.bind(key, stop)
		b.all.logf("[mesh-tools] %s takes its work from %s", b.module, asked.Worker)
		return json.RawMessage(`{}`), nil
	case "answer":
		key := "answer " + asked.Subject
		if b.bound(key) {
			return json.RawMessage(`{}`), nil
		}
		stop, err := conn.SeatAnswer(b.module, asked.Subject, func(subject string, body json.RawMessage) (json.RawMessage, error) {
			return handOn("mesh/proof", map[string]any{"subject": subject, "body": body})
		})
		if err != nil {
			return nil, err
		}
		b.bind(key, stop)
		b.all.logf("[mesh-tools] %s answers the proofs on %s", b.module, asked.Subject)
		return json.RawMessage(`{}`), nil
	}
	return nil, fmt.Errorf("the runtime answers no mesh/seat.%s", verb)
}

// bound and bind keep, per module, what its seat traffic is bound to, so a child that starts again and
// asks again is handed what is already bound instead of a second reader of it.
func (b *moduleBus) bound(key string) bool {
	b.all.mu.Lock()
	defer b.all.mu.Unlock()
	_, ok := b.all.seatBound[b.module+" "+key]
	return ok
}

func (b *moduleBus) bind(key string, stop func()) {
	b.all.mu.Lock()
	defer b.all.mu.Unlock()
	if b.all.seatBound == nil {
		b.all.seatBound = map[string]func(){}
	}
	b.all.seatBound[b.module+" "+key] = stop
}

// BundleEnv is what one module's bundle is started with: the runtime's environment without its own words,
// the module's composed words over it, and the module's and machine's names. **No bus word reaches a
// bundle** (security review of 2026-10-08): the credential and the bus's address are the runtime's, and a
// bundle reaches the bus only through the runtime.
func BundleEnv(base []string, given map[string]string, module, node string) []string {
	words := map[string]string{}
	for _, kv := range base {
		if k, v, ok := strings.Cut(kv, "="); ok && !runtimeOnly[k] {
			words[k] = v
		}
	}
	for k, v := range given {
		if !runtimeOnly[k] {
			words[k] = v
		}
	}
	words["MESH_SERVED_MODULE"] = module
	words["MESH_MODULE"] = module
	if node != "" {
		words["MESH_NODE"] = node
	}
	out := make([]string, 0, len(words))
	for k, v := range words {
		out = append(out, k+"="+v)
	}
	sort.Strings(out)
	return out
}