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

// Package bus is the node runtime's connection to the mesh bus, on NATS (novox/hq design 25, design
// 29, ADR 0160, ADR 0175). It is the Go port of node-tools' broker-nats.ts, and speaks the same
// wire: the same subjects, the same JSON request and reply bodies, the same event headers.
//
//	mesh.mod.<module>.event.<type>   an event a module emits
//	mesh.mod.<module>.tool.<tool>    a tool a module serves
//	mesh.seat.<seat>.tool.<verb>     a role's verb, answered by whoever holds the seat
//	mesh.mod.<module>.call.<tool>.<caller>, mesh.seat.<seat>.call.<verb>.<caller>
//	                                 the same, naming the bus user that called (caller.go, hq issue 365)
package bus

import (
	"context"
	"crypto/sha256"
	"crypto/tls"
	"crypto/x509"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"regexp"
	"strings"
	"sync"
	"time"

	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"

	"github.com/novox/mesh-tools/node-tools/internal/wire"
)

// RequestTimeout is how long a call waits for its answer — what modules already expect. The
// controller answers its verbs well inside it (its AnswerWithin, ten seconds), with a call id while a
// verb still runs, so a caller is never left guessing past this (novox/hq issue 265).
const RequestTimeout = 30 * time.Second

const assignmentsStream = "ASSIGNMENTS"

// Credential is a broker credential as the mesh delivers it (novox/hq ADR 0120).
type Credential struct {
	URL         string  `json:"url"`
	Fingerprint string  `json:"fingerprint,omitempty"`
	Node        string  `json:"node,omitempty"`
	Module      string  `json:"module,omitempty"`
	User        string  `json:"user,omitempty"`
	Password    string  `json:"password,omitempty"`
	Claims      []Claim `json:"claims,omitempty"`
}

// Claim is a seat a module claims, with the verbs it promises (novox/hq ADR 0159).
type Claim struct {
	Seat   string   `json:"seat"`
	Scope  string   `json:"scope,omitempty"`
	Serves []string `json:"serves,omitempty"`
}

// Membership is what the mesh issued one assignment (novox/hq ADR 0160).
type Membership struct {
	Node    string              `json:"node"`
	Module  string              `json:"module"`
	Serves  []Served            `json:"serves"`
	Seats   []SeatVerb          `json:"seats,omitempty"`
	Emits   string              `json:"emits"`
	Reaches map[string][]string `json:"reaches,omitempty"`
	Tools   string              `json:"tools"`
	// State is every bucket the module's code may reach, by the name it uses (novox/hq ADR 0201).
	State []StateIssued `json:"state,omitempty"`
	// Logs is every log the module keeps, by its local name (novox/hq ADR 0297): its own alone.
	Logs []LogIssued `json:"logs,omitempty"`
	// SeatTraffic is what the module's code may submit, say, take, ask, answer and read on seats that name
	// their caller or their kind (novox/hq ADR 0259): the runtime does for it only what is listed here.
	SeatTraffic *SeatTraffic `json:"seat-traffic,omitempty"`
}

// Served is an address a tool is answered on; `{tool}` stands for the tool's name.
type Served struct {
	Subject string `json:"subject"`
	Queue   string `json:"queue,omitempty"`
}

// SeatVerb is one verb of a seat a module holds, where it is answered.
type SeatVerb struct {
	Seat    string `json:"seat"`
	Verb    string `json:"verb"`
	Subject string `json:"subject"`
}

// Envelope is an event as a module emits it: the body is the payload, the metadata rides as headers
// (novox/hq ADR 0042).
type Envelope struct {
	Key     string            `json:"key"`
	Node    string            `json:"node,omitempty"`
	Body    json.RawMessage   `json:"body"`
	Headers map[string]string `json:"headers,omitempty"`
}

// Answered is a call's result and the machine that gave it (novox/hq ADR 0159).
type Answered struct {
	Result json.RawMessage
	Node   string
}

// Handler answers one request body.
type Handler func(body json.RawMessage) (any, error)

// ErrPin is a bus whose certificate is not the one the mesh pinned: final, never retried.
var ErrPin = errors.New("the bus's certificate does not match the pin")

// Fatal is why a connection failure is final rather than "not yet", or "" when waiting may fix it —
// the same classification the TypeScript runtime makes.
func Fatal(err error) string {
	if err == nil {
		return ""
	}
	msg := err.Error()
	if errors.Is(err, ErrPin) || strings.Contains(msg, ErrPin.Error()) {
		return "the bus's certificate does not match the pin"
	}
	if regexp.MustCompile(`(?i)invalid url|no servers available for connection: .*url`).MatchString(msg) ||
		strings.Contains(msg, "nats: invalid url") {
		return "the bus address is not a usable URL"
	}
	if regexp.MustCompile(`(?i)authorization violation|user authentication expired|permissions violation`).MatchString(msg) {
		return "the bus refused this account"
	}
	return ""
}

func normalizeFingerprint(f string) string {
	f = strings.TrimSpace(f)
	if len(f) > 7 && strings.EqualFold(f[:7], "sha256:") {
		f = f[7:]
	}
	return strings.ToLower(strings.ReplaceAll(f, ":", ""))
}

// pinned accepts exactly the certificate with this SHA-256 and no other. The pin is the only check:
// the bus's certificate names the seat, not the address a machine dials it by.
func pinned(want string) *tls.Config {
	want = normalizeFingerprint(want)
	return &tls.Config{
		InsecureSkipVerify: true, //nolint:gosec // replaced by the pin, which is stricter
		MinVersion:         tls.VersionTLS12,
		VerifyPeerCertificate: func(raw [][]byte, _ [][]*x509.Certificate) error {
			if len(raw) == 0 {
				return fmt.Errorf("%w: it presented none", ErrPin)
			}
			sum := sha256.Sum256(raw[0])
			if got := hex.EncodeToString(sum[:]); got != want {
				return fmt.Errorf("%w: it presented %s, not the pinned %s", ErrPin, got, want)
			}
			return nil
		},
	}
}

// Conn is the runtime's connection: the module it is, the memberships it follows, the subjects it
// answers.
type Conn struct {
	nc     *nats.Conn
	js     nats.JetStreamContext
	self   string
	node   string
	cred   Credential
	mu     sync.Mutex
	issued map[string]*Membership // module → membership; present with nil = followed, none issued
	onNew  []func(Membership)
	subs   []*nats.Subscription
	Logf   func(format string, args ...any)
	// answering is every subject this runtime answers on, so a subscription the bus refused can be
	// asked for again (novox/hq issue 222).
	answering map[string]*answered
	// RetryAfter is how long after a refusal a subscription is asked for again, by attempt; past the
	// last, it is given up and said. A variable so a test need not wait minutes.
	RetryAfter []time.Duration
	// calls are the caller-named calls in flight (caller.go, novox/hq issue 365).
	calls callers
}

// answered is one subject the runtime answers on: what to subscribe again with, and how often it was
// refused.
type answered struct {
	subject, queue string
	cb             nats.MsgHandler
	sub            *nats.Subscription
	refusals       int
	stopped        bool
}

// retryAfter is the default wait before each new attempt: a push sends the bus's user list in the
// same breath as the machine that needs it, and the bus reloads it a moment later — five minutes
// covers a slow push, and a grant that never comes is said and left.
var retryAfter = []time.Duration{2 * time.Second, 5 * time.Second, 10 * time.Second, 20 * time.Second,
	30 * time.Second, 60 * time.Second, 60 * time.Second, 120 * time.Second}

var (
	refusedSubject = regexp.MustCompile(`Subscription to "(\S+)"`)
	refusedQueue   = regexp.MustCompile(`using queue "(\S+)"`)
)

func answeringKey(subject, queue string) string { return subject + "\x00" + queue }

// refused is the bus saying no to a subscription. **Asked again, not given up** (novox/hq issue 222):
// what an account may answer is the bus's user list, written on the bus's machine, and a push that
// assigns a module somewhere sends that machine and the bus's in the same breath — the runtime can
// subscribe in the moment before the bus has reloaded. Refused once, the subscription stayed dead
// until some later membership happened to re-serve it, and the module ran unreachable meanwhile.
func (c *Conn) refused(err error) {
	if !errors.Is(err, nats.ErrPermissionViolation) {
		return
	}
	if c.publishRefused(err) {
		return
	}
	m := refusedSubject.FindStringSubmatch(err.Error())
	if len(m) < 2 {
		return
	}
	queue := ""
	if q := refusedQueue.FindStringSubmatch(err.Error()); len(q) >= 2 {
		queue = q[1]
	}
	c.mu.Lock()
	a := c.answering[answeringKey(m[1], queue)]
	if a == nil || a.stopped {
		c.mu.Unlock()
		return
	}
	waits := c.RetryAfter
	if waits == nil {
		waits = retryAfter
	}
	if a.refusals >= len(waits) {
		c.mu.Unlock()
		c.Logf("[mesh-tools] the bus still refuses %s after %d attempts; not served here until the mesh issues it again", a.subject, a.refusals)
		return
	}
	wait := waits[a.refusals]
	a.refusals++
	attempt := a.refusals
	c.mu.Unlock()
	c.Logf("[mesh-tools] the bus refused %s; asking again in %s (attempt %d)", a.subject, wait, attempt)
	time.AfterFunc(wait, func() {
		c.mu.Lock()
		defer c.mu.Unlock()
		if a.stopped {
			return
		}
		if a.sub != nil {
			_ = a.sub.Unsubscribe()
		}
		var sub *nats.Subscription
		var err error
		if a.queue != "" {
			sub, err = c.nc.QueueSubscribe(a.subject, a.queue, a.cb)
		} else {
			sub, err = c.nc.Subscribe(a.subject, a.cb)
		}
		if err != nil {
			c.Logf("[mesh-tools] asking again for %s failed: %v", a.subject, err)
			return
		}
		a.sub = sub
		c.subs = append(c.subs, sub)
	})
}

// Connect dials the bus as the credential's module. A module's subjects come from its credential,
// never from its calls (ADR 0074).
func Connect(cred Credential) (*Conn, error) {
	if cred.Module == "" {
		return nil, errors.New("a broker credential with no module: the runtime derives its subjects " +
			"from the account the mesh issued, and cannot guess which module it is")
	}
	node := cred.Node
	if node == "" {
		node = "?"
	}
	var c *Conn
	opts := []nats.Option{
		nats.Name(node + "." + cred.Module),
		nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) {
			if c != nil {
				c.refused(err)
			}
		}),
		// Reconnect forever: the bus restarting is an upgrade, not a reason to exit.
		nats.MaxReconnects(-1),
	}
	if cred.User != "" {
		opts = append(opts, nats.UserInfo(cred.User, cred.Password),
			// Its own inbox: every user's inbox is private to it (design 25 §4).
			nats.CustomInboxPrefix("_INBOX."+cred.User))
	}
	if strings.TrimSpace(cred.Fingerprint) != "" {
		opts = append(opts, nats.Secure(pinned(cred.Fingerprint)))
	}
	nc, err := nats.Connect(cred.URL, opts...)
	if err != nil {
		return nil, err
	}
	js, err := nc.JetStream()
	if err != nil {
		nc.Close()
		return nil, err
	}
	c = &Conn{nc: nc, js: js, self: cred.Module, node: cred.Node, cred: cred,
		issued: map[string]*Membership{}, Logf: log.Printf, answering: map[string]*answered{}}
	c.Follow(cred.Module)
	return c, nil
}

// Module is what this connection is.
func (c *Conn) Module() string { return c.self }

// Node is the machine this connection's account is scoped to.
func (c *Conn) Node() string { return c.node }

// Credential is what this connection was opened with.
func (c *Conn) Credential() Credential { return c.cred }

// MembershipSubject is the one address a runtime derives for an assignment (ADR 0160).
func MembershipSubject(node, module string) string {
	return "mesh.assignment." + node + "." + module
}

// Follow reads a module's membership on this machine once and follows it live, so its tools are
// served where the mesh issued them (ADR 0175).
func (c *Conn) Follow(module string) {
	c.mu.Lock()
	if c.node == "" {
		c.mu.Unlock()
		return
	}
	if _, has := c.issued[module]; has {
		c.mu.Unlock()
		return
	}
	c.issued[module] = nil
	c.mu.Unlock()
	subject := MembershipSubject(c.node, module)
	// The subject-addressed direct get: the one address the mesh grants this account on the
	// stream's API.
	if got, err := c.nc.Request("$JS.API.DIRECT.GET."+assignmentsStream+"."+subject, nil, 5*time.Second); err == nil {
		if got.Header.Get("Status") == "" && len(got.Data) > 0 {
			var m Membership
			if json.Unmarshal(got.Data, &m) == nil {
				c.mu.Lock()
				c.issued[module] = &m
				c.mu.Unlock()
			}
		}
	}
	if c.Membership(module) == nil {
		c.Logf("[mesh-tools] no membership issued for %s on %s yet; serving the derived shape until one arrives", module, c.node)
	}
	sub, err := c.nc.Subscribe(subject, func(msg *nats.Msg) {
		var m Membership
		if err := json.Unmarshal(msg.Data, &m); err != nil {
			c.Logf("[mesh-tools] a membership arrived that is not one: %v", err)
			return
		}
		c.mu.Lock()
		c.issued[module] = &m
		handlers := append([]func(Membership){}, c.onNew...)
		c.mu.Unlock()
		c.Logf("[mesh-tools] %s on %s was issued a new membership; re-serving on it", module, c.node)
		for _, h := range handlers {
			h(m)
		}
		_ = c.nc.Flush()
	})
	if err == nil {
		c.track(sub)
	}
}

// Membership is what the mesh issued a module here, or nil when nothing has been issued.
func (c *Conn) Membership(module string) *Membership {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.issued[module]
}

// Following says whether this connection follows a module's membership.
func (c *Conn) Following(module string) bool {
	c.mu.Lock()
	defer c.mu.Unlock()
	_, has := c.issued[module]
	return has
}

// Serving is every module whose membership this connection follows.
func (c *Conn) Serving() []string {
	c.mu.Lock()
	defer c.mu.Unlock()
	out := make([]string, 0, len(c.issued))
	for m := range c.issued {
		out = append(out, m)
	}
	return out
}

// OnMembership is called with every new membership any followed module is issued.
func (c *Conn) OnMembership(h func(Membership)) {
	c.mu.Lock()
	c.onNew = append(c.onNew, h)
	c.mu.Unlock()
}

func (c *Conn) track(s *nats.Subscription) {
	c.mu.Lock()
	c.subs = append(c.subs, s)
	c.mu.Unlock()
}

// servedOn is where a served module's tool is answered: its membership's subjects when issued, the
// derived shape otherwise (the shape the mesh issues on day one).
func (c *Conn) servedOn(module, tool string) []Served {
	if m := c.Membership(module); m != nil {
		out := make([]Served, 0, len(m.Serves)+1)
		for _, s := range m.Serves {
			out = append(out, Served{Subject: strings.ReplaceAll(s.Subject, "{tool}", tool), Queue: s.Queue})
		}
		if tool == "tools" && m.Tools != "" {
			found := false
			for _, s := range out {
				found = found || s.Subject == m.Tools
			}
			if !found {
				out = append([]Served{{Subject: m.Tools, Queue: "serve." + module}}, out...)
			}
		}
		return out
	}
	base := "mesh.mod." + module + ".tool." + tool
	out := []Served{{Subject: base, Queue: "serve." + module}}
	if c.node != "" {
		out = append(out, Served{Subject: base + "." + c.node})
	}
	return out
}

type reply struct {
	Result any    `json:"result,omitempty"`
	Error  string `json:"error,omitempty"`
	Node   string `json:"node,omitempty"`
}

// answerOn answers one subject with one handler, and says which machine answered (ADR 0159).
func (c *Conn) answerOn(subject, queue string, h Handler) (func(), error) {
	return c.answerCalledOn(subject, queue, func(_ string, body json.RawMessage) (any, error) { return h(body) })
}

// answerCalledOn answers one subject, handing the handler the caller the subject a call arrived on names:
// under a pattern calledPattern made, its last token; on any other subject, "" — none was proven (novox/hq
// issue 365). Never from the body: a body is the caller's to write.
func (c *Conn) answerCalledOn(subject, queue string, h CalledHandler) (func(), error) {
	cb := func(msg *nats.Msg) {
		go func() {
			var r reply
			result, err := h(callerOn(subject, msg.Subject), json.RawMessage(msg.Data))
			if err != nil {
				r.Error = err.Error()
			} else {
				r.Result = nullable(result)
			}
			r.Node = c.node
			body, _ := wire.Marshal(r)
			// Within what the bus carries, or said to be too large — never a send the library
			// refuses and the caller waits out (novox/hq issue 314).
			if err := msg.Respond(c.fits(subject, body)); err != nil {
				c.Logf("[mesh-tools] could not answer on %s: %v", subject, err)
			}
		}()
	}
	var sub *nats.Subscription
	var err error
	if queue != "" {
		sub, err = c.nc.QueueSubscribe(subject, queue, cb)
	} else {
		sub, err = c.nc.Subscribe(subject, cb)
	}
	if err != nil {
		return func() {}, err
	}
	c.track(sub)
	a := &answered{subject: subject, queue: queue, cb: cb, sub: sub}
	c.mu.Lock()
	c.answering[answeringKey(subject, queue)] = a
	c.mu.Unlock()
	return func() {
		c.mu.Lock()
		a.stopped = true
		if c.answering[answeringKey(subject, queue)] == a {
			delete(c.answering, answeringKey(subject, queue))
		}
		current := a.sub
		c.mu.Unlock()
		_ = current.Unsubscribe()
	}, nil
}

// nullable keeps a nil result as JSON null rather than dropping the key: the TypeScript reply always
// carries `result` when the handler did not throw.
func nullable(v any) any {
	if v == nil {
		return json.RawMessage("null")
	}
	return v
}

// HandleSubject answers a subject outright: a seat's verb where the mesh issued it.
func (c *Conn) HandleSubject(subject string, h Handler) (func(), error) {
	return c.HandleSubjectCalled(subject, func(_ string, body json.RawMessage) (any, error) { return h(body) })
}

// HandleSubjectCalled answers a tool subject and its caller-named counterpart (novox/hq issue 365): a call on
// the counterpart is handed the caller its subject names, one on the subject itself "" — none proven. The
// subject that names no caller is served beside the other for one release, so callers move without a gap
// (its retirement: hq issue 464).
func (c *Conn) HandleSubjectCalled(subject string, h CalledHandler) (func(), error) {
	return c.answerBoth(subject, "", h)
}

// answerBoth answers a subject and, when it is a tool's, the caller-named pattern beside it, in one queue.
func (c *Conn) answerBoth(subject, queue string, h CalledHandler) (func(), error) {
	stop, err := c.answerCalledOn(subject, queue, h)
	if err != nil {
		return func() {}, err
	}
	pattern := calledPattern(subject)
	if pattern == "" {
		return stop, nil
	}
	stopCalled, err := c.answerCalledOn(pattern, queue, h)
	if err != nil {
		c.Logf("[mesh-tools] cannot serve the caller-named %s: %v; its callers are asked on %s alone", pattern, err, subject)
		return stop, nil
	}
	return func() { stop(); stopCalled() }, nil
}

// Handle serves `<module>.<tool>` where the mesh issued that module, and follows its membership:
// when a new one arrives, it serves where it now says and stops where it no longer does.
func (c *Conn) Handle(key string, h Handler) (func(), error) {
	return c.HandleCalled(key, func(_ string, body json.RawMessage) (any, error) { return h(body) })
}

// HandleCalled is Handle telling the handler who called: the bus user a caller-named subject names, or "" for
// a call on the subject that names none (novox/hq issue 365).
func (c *Conn) HandleCalled(key string, h CalledHandler) (func(), error) {
	if strings.HasPrefix(key, "seat:") {
		subject, err := ToolSubject(key, c.self)
		if err != nil {
			return func() {}, err
		}
		return c.answerBoth(subject, "", h)
	}
	module, tool := c.self, key
	if dot := strings.Index(key, "."); dot >= 0 {
		module, tool = key[:dot], key[dot+1:]
	}
	if module != c.self && !c.Following(module) {
		return func() {}, fmt.Errorf("%s cannot serve %s: a module serves its own tools, and a runtime "+
			"those of the modules it follows", c.self, key)
	}
	var mu sync.Mutex
	var stops []func()
	serve := func() {
		mu.Lock()
		defer mu.Unlock()
		for _, s := range stops {
			s()
		}
		stops = nil
		for _, s := range c.servedOn(module, tool) {
			if stop, err := c.answerBoth(s.Subject, s.Queue, h); err == nil {
				stops = append(stops, stop)
			} else {
				c.Logf("[mesh-tools] cannot serve %s on %s: %v", key, s.Subject, err)
			}
		}
	}
	serve()
	c.OnMembership(func(m Membership) {
		if m.Module == module {
			serve()
		}
	})
	return func() {
		mu.Lock()
		defer mu.Unlock()
		for _, s := range stops {
			s()
		}
		stops = nil
	}, nil
}

// reachedAt is where a call by key goes: a subject this connection's own membership says it
// reaches — the machine's when named — else the derived shape.
func (c *Conn) reachedAt(key string) (string, error) {
	name, wanted, _ := strings.Cut(key, "@")
	if m := c.Membership(c.self); m != nil {
		if reach := m.Reaches[name]; len(reach) > 0 {
			if wanted == "" {
				return reach[0], nil
			}
			for _, s := range reach {
				if strings.HasSuffix(s, "."+wanted) {
					return s, nil
				}
			}
		}
	}
	return ToolSubject(key, c.self)
}

// Ask calls a tool by key — or on a subject the mesh listed for it — and learns which machine
// answered. Core request/reply: a tool call is never persisted (design 25 §3).
func (c *Conn) Ask(key string, body any, on string) (Answered, error) {
	subject := on
	if subject == "" {
		s, err := c.reachedAt(key)
		if err != nil {
			return Answered{}, err
		}
		subject = s
	}
	data, err := wire.Marshal(body)
	if err != nil {
		return Answered{}, err
	}
	// **Named by its caller first** (novox/hq issue 365): on the caller-named subject, which the bus lets this
	// account publish with its own name alone; asked again on the subject that names no caller only when the
	// first reached nobody — nothing serves it yet, or the bus does not grant it yet — so no call is done twice.
	if called := CalledSubject(subject, c.cred.User); called != "" {
		answered, handingOver, fallBack, err := c.askCalled(called, data)
		if !fallBack {
			if !handingOver {
				return answered, err
			}
			time.Sleep(HandoverPause)
			answered, again, _, err := c.askCalled(called, data)
			if err != nil && again {
				return Answered{}, fmt.Errorf("asked twice, %s apart, and both times refused as a handover: %w", HandoverPause, err)
			}
			return answered, err
		}
	}
	answered, handingOver, err := c.askOnce(subject, data)
	if !handingOver {
		return answered, err
	}
	// **Refused because the controller is handing over: asked once more** (novox/hq issue 289). The
	// controller being replaced did nothing and said so; the one after it answers the same call. Once:
	// a second refusal is a controller that is not coming back, and the caller is told.
	time.Sleep(HandoverPause)
	answered, again, err := c.askOnce(subject, data)
	if err != nil && again {
		return Answered{}, fmt.Errorf("asked twice, %s apart, and both times refused as a handover: %w", HandoverPause, err)
	}
	return answered, err
}

// HandoverPause is how long a call refused as a handover waits before it is asked once more: the
// supervisor's restart of the controller, which stops the one being replaced and starts the next. A
// variable so a test need not wait.
var HandoverPause = 3 * time.Second

// RetryHandingOver is the mark the controller puts on a call it refused because it is being replaced
// (mesh-controller internal/link, `"retry": "handing-over"`).
const RetryHandingOver = "handing-over"

// askOnce asks once, and says whether the answer was a refusal marked to be asked again.
func (c *Conn) askOnce(subject string, data []byte) (Answered, bool, error) {
	msg, err := c.nc.Request(subject, data, RequestTimeout)
	if err != nil {
		if errors.Is(err, nats.ErrNoResponders) {
			return Answered{}, false, errors.New("503 no responders")
		}
		if errors.Is(err, nats.ErrTimeout) {
			return Answered{}, false, errors.New("timeout")
		}
		return Answered{}, false, err
	}
	return c.answerOf(subject, msg.Data)
}

// answerOf reads one answer, asking the rest of a paged one on the subject it was asked on, and says whether it
// was a refusal marked to be asked again.
func (c *Conn) answerOf(subject string, data []byte) (Answered, bool, error) {
	// An answer larger than one message comes in parts (pages.go, novox/hq issue 314).
	body, err := c.whole(subject, data)
	if err != nil {
		return Answered{}, false, err
	}
	var r struct {
		Result json.RawMessage `json:"result"`
		Error  string          `json:"error"`
		Node   string          `json:"node"`
		Retry  string          `json:"retry"`
	}
	if err := json.Unmarshal(body, &r); err != nil {
		return Answered{}, false, err
	}
	if r.Error != "" {
		return Answered{}, r.Retry == RetryHandingOver, errors.New(r.Error)
	}
	return Answered{Result: r.Result, Node: r.Node}, false, nil
}

// PublishAs emits an event as a module: published into JetStream and awaited, de-duplicated by its
// own id (ADR 0042). The body is the payload; the metadata rides as headers.
func (c *Conn) PublishAs(module string, env Envelope) error {
	msg := nats.NewMsg("mesh.mod." + module + ".event." + env.Key)
	for k, v := range env.Headers {
		msg.Header.Set(k, v)
	}
	if env.Headers["content-type"] == "" {
		msg.Header.Set("content-type", "application/json")
	}
	if env.Node != "" {
		msg.Header.Set("x-node", env.Node)
	}
	body := env.Body
	if len(body) == 0 {
		body = json.RawMessage("null")
	}
	msg.Data = body
	var opts []nats.PubOpt
	if id := env.Headers["x-event-id"]; id != "" {
		opts = append(opts, nats.MsgId(id))
	}
	_, err := c.js.PublishMsg(msg, opts...)
	return err
}

// Say publishes one message on core NATS, kept by nobody: a heartbeat, whose loss is the next one.
func (c *Conn) Say(subject string, body []byte) error { return c.nc.Publish(subject, body) }

// ServedOn is where a served module's tool is answered right now: the membership's subjects when
// issued, the derived shape otherwise — what Handle subscribes, for what announces it (ADR 0197).
func (c *Conn) ServedOn(module, tool string) []Served { return c.servedOn(module, tool) }

// Raw answers one subject with a function of the request, not a tool's reply envelope: the NATS
// services protocol's discovery subjects answer in their own format (novox/hq ADR 0197). A nil
// answer is no reply — the request was for another service.
func (c *Conn) Raw(subject string, answer func(subject string, data []byte) []byte) (func(), error) {
	sub, err := c.nc.Subscribe(subject, func(msg *nats.Msg) {
		if body := answer(msg.Subject, msg.Data); body != nil {
			// Said, never dropped: an answer the bus will not carry was silence before (2026-10-04).
			if err := msg.Respond(body); err != nil {
				c.Logf("[mesh-tools] could not answer %s (%d bytes): %v", msg.Subject, len(body), err)
			}
		}
	})
	if err != nil {
		return func() {}, err
	}
	c.track(sub)
	return func() { _ = sub.Unsubscribe() }, nil
}

// Arrival is one answer to a request, and when it arrived.
type Arrival struct {
	Data []byte
	At   time.Time
}

// Collect publishes one request and hands every answer to the channel as it arrives, until stopped:
// for a caller that decides for itself when it has heard enough — discovery waits for every service
// that said it is there, not for a fixed moment (2026-10-05). Answers beyond what the caller has
// taken are kept up to the channel's room; an empty answer is no answer.
func (c *Conn) Collect(subject string, body []byte) (<-chan Arrival, func(), error) {
	out, ask, stop, err := c.Collector(subject)
	if err != nil {
		return nil, func() {}, err
	}
	if err := ask(subject, body); err != nil {
		stop()
		return nil, func() {}, err
	}
	return out, stop, nil
}

// Collector is one inbox for many requests: every answer to every request `ask` publishes arrives on
// the one channel, until stopped — for discovery, which asks each service that said it is there for
// what it serves, as each says so (novox/hq issue 554). `what` names the requests in a log line.
func (c *Conn) Collector(what string) (<-chan Arrival, func(subject string, body []byte) error, func(), error) {
	inbox := c.nc.NewRespInbox()
	out := make(chan Arrival, 1024)
	var mu sync.Mutex
	stopped := false
	sub, err := c.nc.Subscribe(inbox, func(msg *nats.Msg) {
		if len(msg.Data) == 0 {
			return
		}
		a := Arrival{Data: msg.Data, At: time.Now()}
		mu.Lock()
		defer mu.Unlock()
		if stopped {
			return
		}
		select {
		case out <- a:
		default:
			c.Logf("[mesh-tools] more answers to %s than are kept; one (%d bytes) dropped", what, len(msg.Data))
		}
	})
	if err != nil {
		return nil, nil, func() {}, err
	}
	stop := func() {
		_ = sub.Unsubscribe()
		mu.Lock()
		stopped = true
		mu.Unlock()
	}
	ask := func(subject string, body []byte) error { return c.nc.PublishRequest(subject, inbox, body) }
	return out, ask, stop, nil
}

// MaxPayload is the largest message the bus carries, as the server told this connection.
func (c *Conn) MaxPayload() int64 { return c.nc.MaxPayload() }

// Flush waits until the bus has every subscription made so far, so what is served is answerable
// when this returns.
func (c *Conn) Flush() { _ = c.nc.Flush() }

// Close unsubscribes everything and drains, so an in-flight reply is finished rather than dropped.
func (c *Conn) Close() {
	c.mu.Lock()
	subs := c.subs
	c.subs = nil
	c.mu.Unlock()
	for _, s := range subs {
		_ = s.Unsubscribe()
	}
	_ = c.nc.Drain()
}

// ToolSubject is a tool's subject. A bare name is this module's own; `<module>.<tool>` another's;
// `seat:<seat>.<verb>` a role's, with `@<node>` for a node-scoped seat (design 33 §4).
func ToolSubject(key, self string) (string, error) {
	if strings.HasPrefix(key, "seat:") {
		rest := strings.TrimPrefix(key, "seat:")
		dot := strings.Index(rest, ".")
		if dot < 0 {
			return "", fmt.Errorf("%q names a seat and no verb: seat:<seat>.<verb>", key)
		}
		seat := rest[:dot]
		verb, node, _ := strings.Cut(rest[dot+1:], "@")
		if node != "" {
			return "mesh.seat." + seat + ".tool." + verb + "." + node, nil
		}
		return "mesh.seat." + seat + ".tool." + verb, nil
	}
	name, node, _ := strings.Cut(key, "@")
	var base string
	if dot := strings.Index(name, "."); dot < 0 {
		base = "mesh.mod." + self + ".tool." + name
	} else {
		base = "mesh.mod." + name[:dot] + ".tool." + name[dot+1:]
	}
	if node != "" {
		return base + "." + node, nil
	}
	return base, nil
}

// SeatToolSubject is a seat's verb as its holder serves it: flat for a mesh seat, carrying the
// machine for a node-scoped one.
func SeatToolSubject(seat, verb, scope, node string) string {
	base := "mesh.seat." + seat + ".tool." + verb
	if scope == "node" && node != "" {
		return base + "." + node
	}
	return base
}

// AskAs calls a tool on a module's behalf (novox/hq ADR 0198): a bare key is that module's own tool,
// `<module>.<tool>` another's, `seat:<seat>.<verb>[@<node>]` a role's — resolved through what the
// mesh issued that module to reach, as its own runtime resolved it, else the derived shape.
func (c *Conn) AskAs(module, key string, body any) (Answered, error) {
	name, wanted, _ := strings.Cut(key, "@")
	subject := ""
	if m := c.Membership(module); m != nil {
		if reach := m.Reaches[name]; len(reach) > 0 {
			subject = reach[0]
			if wanted != "" {
				subject = ""
				for _, s := range reach {
					if strings.HasSuffix(s, "."+wanted) {
						subject = s
					}
				}
			}
		}
	}
	if subject == "" {
		s, err := ToolSubject(key, module)
		if err != nil {
			return Answered{}, err
		}
		subject = s
	}
	return c.Ask(key, body, subject)
}

// EventsStream is where every module's events land, and ConsumerOf the durable consumer the
// controller makes for a module on a machine: the same names the module's own runtime bound
// (node-tools/src/broker-nats.ts), so moving the module into the node's runtime neither loses an
// event nor sees one twice.
const EventsStream = "EVENTS"

// ConsumerOf is a module's durable consumer on a machine: `<node>_<module>`.
func ConsumerOf(node, module string) string {
	if node == "" {
		node = "?"
	}
	return node + "_" + module
}

// NakDelay is how long an event a handler failed waits before it is offered again: a transient cause
// gets another attempt, a permanent one exhausts the consumer's max-deliver rather than spinning. The
// consumer then gives the event up, and the controller keeps it in DEAD_LETTERS until a person delivers
// it again — under `mesh.again.<consumer>.…`, from which keyOf reads the same key — or drops it
// (novox/hq issue 330).
var NakDelay = 5 * time.Second

// EventHeartbeat is how often an event a bundle is still handling is said to be in progress, well inside the
// consumer's ack wait. The client holds one event at a time (PullMaxMessages(1)): events are handed over in
// order, one at a time, so an event waiting in the client's buffer would have its ack wait run out before it is
// handed over, and a burst would end in max-deliver with nothing failed (novox/hq issue 387).
var EventHeartbeat = 10 * time.Second

// ConsumeAs reads a module's durable consumer and hands each event to deliver (novox/hq ADR 0198):
// acknowledged when deliver returns nil, negatively acknowledged after NakDelay when it returns an
// error, terminated when it is not an event at all. The consumer is the controller's to create; this
// binds to it and never makes one. Runs until stopped.
func (c *Conn) ConsumeAs(module string, deliver func(Envelope) error) (func(), error) {
	durable := ConsumerOf(c.node, module)
	js, err := jetstream.New(c.nc)
	if err != nil {
		return nil, err
	}
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	// Bound by name, never created and never matched against a subject: the consumer and its
	// filters are the controller's (design 29 §3), exactly as the module's own runtime bound it.
	consumer, err := js.Consumer(ctx, EventsStream, durable)
	if err != nil {
		return nil, fmt.Errorf("binding %s's consumer %s: %w", module, durable, err)
	}
	reading, err := consumer.Consume(func(msg jetstream.Msg) {
		env, ok := envelopeOf(msg.Subject(), msg.Headers(), msg.Data())
		if !ok {
			// Unparseable: redelivering bytes no version of a handler can read is a loop.
			_ = msg.Term()
			return
		}
		// Kept alive while the bundle works on it, as a worker's work is (seat.go): the consumer's ack wait
		// (30 seconds) is for a runtime that died, not a handler that is slow — the runtime gives a handler
		// launch.EventTimeout. Without it an event handled longer than the ack wait was offered again while
		// it was handled, and counted toward max-deliver (novox/hq issue 387).
		done, stopped := make(chan struct{}), make(chan struct{})
		go func() {
			defer close(stopped)
			tick := time.NewTicker(EventHeartbeat)
			defer tick.Stop()
			for {
				select {
				case <-done:
					return
				case <-tick.C:
					_ = msg.InProgress()
				}
			}
		}()
		err := deliver(env)
		close(done)
		// No in-progress after the answer: one after a Nak would stretch its delay to the whole ack wait.
		<-stopped
		if err != nil {
			_ = msg.NakWithDelay(NakDelay)
			return
		}
		_ = msg.Ack()
	}, jetstream.PullMaxMessages(1))
	if err != nil {
		return nil, fmt.Errorf("reading %s's consumer %s: %w", module, durable, err)
	}
	var once sync.Once
	return func() { once.Do(reading.Stop) }, nil
}

// envelopeOf rebuilds the envelope a module sees from a delivered event: its key is the emitter and
// the event, recovered from the subject; its metadata rides as headers (novox/hq ADR 0042).
func envelopeOf(subject string, header nats.Header, data []byte) (Envelope, bool) {
	if !json.Valid(data) {
		return Envelope{}, false
	}
	headers := map[string]string{}
	for k := range header {
		headers[k] = header.Get(k)
	}
	return Envelope{Key: keyOf(subject), Node: headers["x-node"], Body: json.RawMessage(data), Headers: headers}, true
}

// keyOf is the key a module sees for an event subject: `mesh.mod.<emitter>.event.<event>` is
// `<emitter>.<event>` — the vocabulary its manifest names what it consumes in.
func keyOf(subject string) string {
	before, event, found := strings.Cut(subject, ".event.")
	if !found {
		return subject
	}
	parts := strings.Split(before, ".")
	if emitter := parts[len(parts)-1]; emitter != "" {
		return emitter + "." + event
	}
	return event
}