mesh-catalog/modules/nats/cmd/nats-tools/main.go

// nats-tools: the mesh bus's own tools (novox/hq design 25), a Go bundle the node's runtime launches on the
// machine that runs the bus. Everything here reads — the server's monitoring API and the user list the
// controller composed — and nothing changes the bus. stdout is the MCP channel; this says nothing else.
package main

import (
	"context"
	"fmt"
	"os"
	"time"

	stdio "git.novox.be/novox/mesh-sdk/go"
)

func str(description string) map[string]any {
	return map[string]any{"type": "string", "description": description}
}

func num(description string) map[string]any {
	return map[string]any{"type": "number", "description": description}
}

func arg(a map[string]any, k string) string { s, _ := a[k].(string); return s }

func limitOf(a map[string]any, fallback int) int {
	if v, ok := a["limit"].(float64); ok && v > 0 {
		return int(v)
	}
	return fallback
}

func tools(monitor Monitor, users func(context.Context) ([]byte, error)) []stdio.Tool {
	get := func(path string) ([]byte, error) {
		ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
		defer cancel()
		return monitor(ctx, path)
	}
	conf := func() ([]byte, error) {
		ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
		defer cancel()
		return users(ctx)
	}
	jsz := "/jsz?accounts=1&streams=1&consumers=1&config=1"
	return []stdio.Tool{
		{Name: "nats_server",
			Description: "The bus server as it runs: version, uptime, limits (max_payload included), connections, subscriptions, slow consumers, traffic, memory and CPU, JetStream totals, and its health.",
			Run: func(map[string]any) (any, error) {
				varz, err := get("/varz")
				if err != nil {
					return nil, err
				}
				health, _ := get("/healthz?js-enabled-only=true")
				return Server(varz, health)
			}},
		{Name: "nats_connections",
			Description: "Every client connected to the bus, named by the mesh user it is (a machine's host `node.<machine>`, a machine's runtime `<machine>.node-tools`, the controller, a person): address, uptime, round trip, traffic and pending bytes. Narrowed by a user or connection name, sorted by out_bytes (default), in_bytes, pending, msgs or subscriptions.",
			Input:       map[string]any{"user": str("part of a user or connection name, e.g. g14"), "sort": str("out_bytes, in_bytes, pending, msgs or subscriptions"), "limit": num("at most this many (default 50)")},
			Run: func(a map[string]any) (any, error) {
				connz, err := get("/connz?auth=1&limit=1024")
				if err != nil {
					return nil, err
				}
				return Connections(connz, arg(a, "user"), arg(a, "sort"), limitOf(a, 50))
			}},
		{Name: "nats_closed_connections",
			Description: "Who opened and closed connections to the bus lately (novox/hq issue 327): per user and connection name, how many closed in the last hours, how many were dropped rather than closed by their client and why, how long each lived, and the rate per hour — the question behind a connection total that keeps rising while the open count stays flat. Narrowed by part of a user or connection name.",
			Input:       map[string]any{"hours": num("how many hours back (default 1)"), "user": str("part of a user or connection name, e.g. controller")},
			Run: func(a map[string]any) (any, error) {
				connz, err := get(fmt.Sprintf("/connz?state=closed&auth=1&limit=%d", maxClosedKept))
				if err != nil {
					return nil, err
				}
				hours, _ := a["hours"].(float64)
				return ClosedConnections(connz, time.Now(), hours, arg(a, "user"))
			}},
		{Name: "nats_subscriptions",
			Description: "Who listens on what: every subscription whose subject contains the text given, with its queue group, the messages it received and the user holding it, the busiest first.",
			Input:       map[string]any{"subject": str("part of a subject, e.g. mesh.seat.anthropic-licence-manager or $SRV"), "limit": num("at most this many (default 100)")},
			Run: func(a map[string]any) (any, error) {
				subsz, err := get("/subsz?subs=1&limit=100000")
				if err != nil {
					return nil, err
				}
				connz, _ := get("/connz?auth=1&limit=1024")
				return Subscriptions(subsz, connz, arg(a, "subject"), limitOf(a, 100))
			}},
		{Name: "nats_streams",
			Description: "JetStream: every stream with its subjects, retention, message count and size — or, given one stream, every consumer on it with what it has pending, unacknowledged and redelivered.",
			Input:       map[string]any{"stream": str("one stream, e.g. EVENTS, CONTROL, ASSIGNMENTS, SEAT_NODE_BUILD_AGENT or KV_<bucket>")},
			Run: func(a map[string]any) (any, error) {
				doc, err := get(jsz)
				if err != nil {
					return nil, err
				}
				return Streams(doc, arg(a, "stream"))
			}},
		{Name: "nats_backlog",
			Description: "Every consumer that is behind — messages pending or delivered and not yet acknowledged — the furthest behind first. Empty when every module has handled what it was sent.",
			Run: func(map[string]any) (any, error) {
				doc, err := get(jsz)
				if err != nil {
					return nil, err
				}
				return Backlog(doc)
			}},
		{Name: "nats_buckets",
			Description: "Every module's state on the bus (novox/hq ADR 0201): each key-value bucket with the module and state it belongs to, its keys, size, history kept, last change and live watchers.",
			Run: func(map[string]any) (any, error) {
				doc, err := get(jsz)
				if err != nil {
					return nil, err
				}
				return Buckets(doc)
			}},
		{Name: "nats_users",
			Description: "The bus's users as the controller composed them — never a password hash. Without a user, each with how many grants it holds; with part of a user's name, its publish and subscribe grants in full.",
			Input:       map[string]any{"user": str("part of a user name, e.g. g14.node-tools or controller")},
			Run: func(a map[string]any) (any, error) {
				c, err := conf()
				if err != nil {
					return nil, err
				}
				return Users(c, arg(a, "user")), nil
			}},
		{Name: "nats_user_can",
			Description: "Whether a bus user may publish or subscribe to a subject, and the grant that allows it — the question behind a request that timed out because the bus refused it.",
			Input:       map[string]any{"user": str("the user exactly, e.g. g14.node-tools"), "action": str("publish or subscribe"), "subject": str("a concrete subject, e.g. $KV.claude-code_servers.all.x")},
			Run: func(a map[string]any) (any, error) {
				c, err := conf()
				if err != nil {
					return nil, err
				}
				return UserCan(c, arg(a, "user"), arg(a, "action"), arg(a, "subject"))
			}},
	}
}

func main() {
	if err := stdio.Serve("", tools(LiveMonitor(), ReadUsers)); err != nil {
		fmt.Fprintln(os.Stderr, err)
		os.Exit(1)
	}
}