Ask a refused subscription again on a new link and announce the node seat as discovery reads it, so node-engine.root-free is answered again (issue 774)

Every engine's node-seat subscriptions were refused at 12:10 by a user list composed without them. nats
asks a refused subscription again only on a new connection, and a user-list reload makes none, so the
seat answered nobody all afternoon and the messenger held every Telegram ask. The endpoints also lacked
the kind and scope discovery reads, so the console said nothing in the mesh is called node-engine.

Quick fix: restart the engines. Thorough fix (chosen, the second time a stale user list left an engine
deaf): a refused subscription of any kind is said and the link is opened again after a pause (1 min,
doubling to 30 min), so every subscription the engine makes is asked again; the seat's endpoints carry
kind, seat, scope, module, description and schema like every tool runner's.

internal/link/engineseat.go

@@ -111,7 +111,8 @@ func (l *natsLink) ServeEngineSeat(handle EngineFunc) {
 		return
 	}
 	svc, err := micro.AddService(l.conn, micro.Config{Name: EngineSeat, Version: "1.0.0",
-		Description: "the node seat node-engine of " + l.node + " (novox/hq ADR 0563)", QueueGroupDisabled: true})
+		Description: "the node seat node-engine of " + l.node + " (novox/hq ADR 0563)", QueueGroupDisabled: true,
+		Metadata: map[string]string{"seat": EngineSeat, "scope": "node", "node": l.node}})
 	if err != nil {
 		return
 	}
@@ -120,7 +121,7 @@ func (l *natsLink) ServeEngineSeat(handle EngineFunc) {
 		answer := micro.HandlerFunc(func(r micro.Request) {
 			_ = r.Respond(AnswerEngineCall(handle, verb, l.node, r.Subject(), r.Data()))
 		})
-		meta := map[string]string{"seat": EngineSeat, "tool": verb, "node": l.node}
+		meta := EngineEndpointMetadata(v, l.node)
 		_ = svc.AddEndpoint(verb, answer, micro.WithEndpointSubject(EngineToolSubject(verb, l.node)),
 			micro.WithEndpointMetadata(meta))
 		_ = svc.AddEndpoint(verb+"-called", answer, micro.WithEndpointSubject(EngineCallPattern(verb, l.node)),
@@ -129,6 +130,23 @@ func (l *natsLink) ServeEngineSeat(handle EngineFunc) {
 	l.engine = svc
 }
 
+// EngineModule is the module whose code answers the seat, as discovery names it: the node-engine's manifest's.
+const EngineModule = "mesh-host"
+
+// EngineEndpointMetadata is what discovery reads of one verb's endpoint, in the shape every tool runner and the
+// controller announce a seat's verb in (mesh-tools node-tools' announce package): its kind, the seat, its scope, the
+// module whose code answers, the verb, the node, its description and its schema. Without the kind and the scope the
+// console listed none of it, and `<node>/node-engine.<verb>` answered that nothing in the mesh is called node-engine
+// (the sweep of 2026-10-30).
+func EngineEndpointMetadata(v catalogue.Verb, node string) map[string]string {
+	schema, err := json.Marshal(v.Input)
+	if err != nil || v.Input == nil {
+		schema = []byte("{}")
+	}
+	return map[string]string{"kind": "seat", "seat": EngineSeat, "scope": "node", "module": EngineModule,
+		"tool": v.Name, "node": node, "interchangeable": "false", "description": v.Description, "schema": string(schema)}
+}
+
 // SayEngine says one of the seat's events about this node, on the events stream.
 func (l *natsLink) SayEngine(event string, body any) error {
 	raw, err := json.Marshal(body)

internal/link/handover_test.go

@@ -342,6 +342,35 @@ func TestNatsAHandOverIsAskedOfTheNode(t *testing.T) {
 		if _, err := conn.Request(EngineToolSubject("acts", "another"), []byte(`{}`), 300*time.Millisecond); err == nil {
 			t.Fatal("another node's verb was answered here")
 		}
+		// Announced in the shape the console reads a seat's verb in (sweep 2026-10-30): without kind and scope it listed
+		// none of it, and <node>/node-engine.root-free answered that nothing in the mesh is called node-engine.
+		info, err := conn.Request("$SRV.INFO."+EngineSeat, nil, 5*time.Second)
+		if err != nil {
+			t.Fatal(err)
+		}
+		var said struct {
+			Endpoints []struct {
+				Subject  string            `json:"subject"`
+				Metadata map[string]string `json:"metadata"`
+			} `json:"endpoints"`
+		}
+		if err := json.Unmarshal(info.Data, &said); err != nil {
+			t.Fatal(err)
+		}
+		announced := 0
+		for _, e := range said.Endpoints {
+			md := e.Metadata
+			if md["kind"] != "seat" || md["scope"] != "node" || md["seat"] != EngineSeat || md["module"] != "mesh-host" ||
+				md["node"] != m.Node || md["description"] == "" || !strings.HasPrefix(md["schema"], "{") {
+				t.Errorf("the endpoint on %s announces %v", e.Subject, md)
+			}
+			if e.Subject == EngineToolSubject("root-free", m.Node) {
+				announced++
+			}
+		}
+		if announced != 1 {
+			t.Errorf("root-free is announced on its tool subject %d times", announced)
+		}
 		for seat, want := range map[string]bool{EngineSeat: true, "node-login-shell": false} {
 			verb := "acts"
 			if !want {

internal/link/hearing_nats.go

@@ -5,6 +5,8 @@ import (
 	"crypto/ed25519"
 	"errors"
 	"fmt"
+	"os"
+	"regexp"
 	"strings"
 	"sync"
 	"time"
@@ -57,6 +59,9 @@ type natsLink struct {
 	engine   micro.Service
 	warrants *nats.Subscription
 	node     string
+	// refused carries each subject the bus refused this link a subscription to (RefusedSubscription), which serve drops
+	// the link for after a pause, so every subscription is asked again on the next.
+	refused chan string
 	// signer is the mesh's key, which a hand-over ask is verified with as a declaration is (issue 356).
 	signer ed25519.PublicKey
 	lost   chan error
@@ -108,9 +113,13 @@ func dialNats(ctx context.Context, m Membership, timeout time.Duration) (Link, e
 
 	l := &natsLink{
 		conn: conn, js: js, node: m.Node, signer: m.Signer,
-		lost:  make(chan error, 1),
-		asked: make(chan struct{}, 1),
+		lost:    make(chan error, 1),
+		asked:   make(chan struct{}, 1),
+		refused: make(chan string, 64),
 	}
+	// Set before anything subscribes, so the node seat's service (micro), which wraps the handler it finds, hands every
+	// refusal of its own subjects on to it too.
+	conn.SetErrorHandler(l.asyncError)
 
 	l.hearAsks()
 
@@ -259,6 +268,23 @@ func (l *natsLink) HearMoveSecrets(handle MoveSecretFunc) {
 	}
 }
 
+// asyncError is said by the client as before, and a subscription the bus refused is handed to serve (Refused).
+func (l *natsLink) asyncError(_ *nats.Conn, _ *nats.Subscription, err error) {
+	if err == nil {
+		return
+	}
+	fmt.Fprintf(os.Stderr, "%v\n", err)
+	if subject, ok := RefusedSubscription(err); ok && l.refused != nil {
+		select {
+		case l.refused <- subject:
+		default:
+		}
+	}
+}
+
+// Refused says each subject the bus refused this link a subscription to.
+func (l *natsLink) Refused() <-chan string { return l.refused }
+
 func (l *natsLink) AskedToReport() <-chan struct{} { return l.asked }
 func (l *natsLink) Lost() <-chan error             { return l.lost }
 
@@ -339,3 +365,19 @@ var (
 	_ EngineSeats  = (*natsLink)(nil)
 	_ OperatorAsks = (*natsLink)(nil)
 )
+
+// refusedSubscription is how the server words a subscription its user list does not grant.
+var refusedSubscription = regexp.MustCompile(`Permissions Violation for Subscription to "([^"]+)"`)
+
+// RefusedSubscription is the subject a permissions violation names when it refused a subscription; false for any other
+// error, a refused publish among them (a publish is refused each time it is tried, and is said where it is tried).
+func RefusedSubscription(err error) (string, bool) {
+	if err == nil || !errors.Is(err, nats.ErrPermissionViolation) && !strings.Contains(err.Error(), "permissions violation") {
+		return "", false
+	}
+	m := refusedSubscription.FindStringSubmatch(err.Error())
+	if m == nil {
+		return "", false
+	}
+	return m[1], true
+}

internal/link/run.go

@@ -150,6 +150,15 @@ func holdWith(ctx context.Context, run attempt, say Announce, roused Roused) err
 	}
 }
 
+// Refusals is a link that says each subject the bus refused it a subscription to. Optional.
+type Refusals interface {
+	Refused() <-chan string
+}
+
+// refusalsSettle is how long a link runs with no subscription refused before it counts as holding them: refusals arrive
+// within moments of subscribing.
+const refusalsSettle = 30 * time.Second
+
 // Run holds the link open once, hearing what the controller asks and saying the node's facts.
 //
 // Outbound only, and nothing listens on this machine. Returns when the link ends, for any reason;
@@ -226,6 +235,21 @@ func serve(ctx context.Context, link Link, m Membership, queue *Session, say Ann
 		s.HearMoveSecrets(queue.MoveSecret)
 	}
 
+	// And every subscription the bus refused, because the user list it loaded does not grant it (yet): nats asks a
+	// refused subscription again only on a new connection, and a reload of the user list does not make one, so the node
+	// seat, the hand-over and every other subject heard here stayed deaf until the engine restarted (the sweep of
+	// 2026-10-30: every engine was refused its node seat at 12:10 and answered nobody all afternoon). The link is kept
+	// for a pause, said, and then dropped, so Hold opens it again and every subscription is asked anew.
+	var refused <-chan string
+	if r, ok := link.(Refusals); ok {
+		refused = r.Refused()
+	}
+	var again <-chan time.Time
+	var first string
+	refusedCount := 0
+	held := time.NewTimer(refusalsSettle)
+	defer held.Stop()
+
 	for {
 		if ctx.Err() != nil {
 			return nil
@@ -237,6 +261,23 @@ func serve(ctx context.Context, link Link, m Membership, queue *Session, say Ann
 			return reason
 		case <-asks:
 			queue.FactsAsked(ctx)
+		case subject := <-refused:
+			refusedCount++
+			if again == nil {
+				first = subject
+				pause := queue.refusedPause()
+				say(fmt.Sprintf("the bus refused this node a subscription to %s: the user list it loaded does not grant "+
+					"it, so nothing here answers there; the link is opened again in %s so every subscription is asked "+
+					"again", subject, pause))
+				again = time.After(pause)
+			}
+		case <-held.C:
+			if refusedCount == 0 {
+				queue.refusalsHeld()
+			}
+		case <-again:
+			return fmt.Errorf("the bus refused %d of this node's subscriptions, %s first; opened again so each is "+
+				"asked again", refusedCount, first)
 		}
 	}
 }

internal/link/session.go

@@ -60,6 +60,12 @@ type Session struct {
 	Keys    func() RecordKeys
 	Say     Announce
 	Timeout time.Duration
+	// RefusedPause is how long a link keeps running after the bus refused it a subscription before it is dropped and
+	// opened again (zero is DefaultRefusedPause); doubled for each link in a row that is refused again, to RefusedPauseMost.
+	RefusedPause time.Duration
+
+	// refusedInARow is how many links in a row the bus refused a subscription to.
+	refusedInARow int
 
 	once sync.Once
 
@@ -91,6 +97,36 @@ type Session struct {
 	healthWritten *NodeHealth
 }
 
+// DefaultRefusedPause and RefusedPauseMost bound how soon a link refused a subscription is opened again: the grants
+// part composes the user list again within a minute of what it is composed from changing, so the first pause waits for
+// that; one that is still refused waits longer each time, never more than half an hour, and always says so.
+const (
+	DefaultRefusedPause = time.Minute
+	RefusedPauseMost    = 30 * time.Minute
+)
+
+// refusedPause is the pause before a link the bus refused a subscription to is opened again, counting it.
+func (q *Session) refusedPause() time.Duration {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+	pause := q.RefusedPause
+	if pause <= 0 {
+		pause = DefaultRefusedPause
+	}
+	for i := 0; i < q.refusedInARow && pause < RefusedPauseMost; i++ {
+		pause *= 2
+	}
+	q.refusedInARow++
+	return min(pause, RefusedPauseMost)
+}
+
+// refusalsHeld says a link has held its subscriptions: the next refusal starts from the first pause again.
+func (q *Session) refusalsHeld() {
+	q.mu.Lock()
+	q.refusedInARow = 0
+	q.mu.Unlock()
+}
+
 func (q *Session) init() {
 	q.once.Do(func() {
 		if q.Say == nil {

internal/link/session_test.go

@@ -5,7 +5,9 @@ import (
 	"crypto/ed25519"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"os"
+	"strings"
 	"sync"
 	"testing"
 	"time"
@@ -162,3 +164,78 @@ func eventuallyHeld(t *testing.T, what string, ok func() bool) {
 		time.Sleep(5 * time.Millisecond)
 	}
 }
+
+// refusingLink is a quiet link the bus refused subscriptions to.
+type refusingLink struct {
+	*quietLink
+	refused chan string
+}
+
+func (l *refusingLink) Refused() <-chan string { return l.refused }
+
+// The sweep of 2026-10-30: every engine was refused its node seat's subscriptions at 12:10, by a user list composed
+// without them, and nats asks a refused subscription again only on a new connection; the seat answered nobody all
+// afternoon. A refused subscription is read from the server's words, said, and the link opened again after a pause
+// that grows while it is refused again; a refused publish is not one.
+func TestASubscriptionTheBusRefusedIsAskedAgainOnANewLink(t *testing.T) {
+	// The server's words, as the client hands them on (journal of nox-mesh-host on novox, 2026-10-30 12:10:44).
+	for _, row := range []struct {
+		err  error
+		want string
+	}{
+		{fmt.Errorf("%w: %s", nats.ErrPermissionViolation,
+			`Permissions Violation for Subscription to "mesh.seat.node-engine.tool.root-free.novox"`),
+			"mesh.seat.node-engine.tool.root-free.novox"},
+		{errors.Join(fmt.Errorf("%w: %s", nats.ErrPermissionViolation,
+			`Permissions Violation for Subscription to "$SRV.PING.node-engine"`), errors.New("stopping service: x")),
+			"$SRV.PING.node-engine"},
+		{fmt.Errorf("%w: %s", nats.ErrPermissionViolation,
+			`Permissions Violation for Publish to "$JS.API.STREAM.INFO.KV_mesh-host_health"`), ""},
+		{errors.New("nats: timeout"), ""},
+	} {
+		got, ok := RefusedSubscription(row.err)
+		if got != row.want || ok != (row.want != "") {
+			t.Errorf("%v reads as %q, %v", row.err, got, ok)
+		}
+		l := &natsLink{refused: make(chan string, 1)}
+		l.asyncError(nil, nil, row.err)
+		if heard := len(l.refused); heard != map[bool]int{true: 1}[row.want != ""] {
+			t.Errorf("%v handed %d refusals on", row.err, heard)
+		}
+	}
+
+	m, _ := aMember(t)
+	q := &Session{Membership: m, Timeout: time.Second, RefusedPause: 50 * time.Millisecond}
+	var said []string
+	var mu sync.Mutex
+	say := func(s string) { mu.Lock(); said = append(said, s); mu.Unlock() }
+	for _, want := range []time.Duration{50 * time.Millisecond, 100 * time.Millisecond} {
+		l := &refusingLink{quietLink: newQuietLink(), refused: make(chan string, 2)}
+		l.refused <- "mesh.seat.node-engine.tool.root-free.n1"
+		l.refused <- "$SRV.PING"
+		began := time.Now()
+		done := make(chan error, 1)
+		go func() { done <- serve(context.Background(), l, m, q, say, time.Second) }()
+		select {
+		case err := <-done:
+			if err == nil || !strings.Contains(err.Error(), "refused 2 of this node's subscriptions, "+
+				"mesh.seat.node-engine.tool.root-free.n1 first") {
+				t.Fatalf("the link ended with %v", err)
+			}
+			if took := time.Since(began); took < want {
+				t.Errorf("the link was dropped after %s, before its pause of %s", took, want)
+			}
+		case <-time.After(5 * time.Second):
+			t.Fatal("a link the bus refused a subscription was kept, and the subscription never asked again")
+		}
+	}
+	mu.Lock()
+	defer mu.Unlock()
+	if n := strings.Count(strings.Join(said, "\n"), "the bus refused this node a subscription to "+
+		"mesh.seat.node-engine.tool.root-free.n1"); n != 2 {
+		t.Errorf("the refusal was said %d times in two links: %q", n, said)
+	}
+	if q.refusalsHeld(); q.refusedPause() != 50*time.Millisecond {
+		t.Error("a link that held its subscriptions did not start from the first pause again")
+	}
+}