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. Almost everything here reads — the server's monitoring API and the user list the
// controller composed. The two that change the bus (removal.go) are the only ones that connect to it, on the
// module's own credential, and say what they did. stdout is the MCP channel; this says nothing else.
package main

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

	stdio "git.novox.be/novox/mesh-sdk/go"
	"nats-tools/internal/busarchive"
)

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), written func(context.Context) (time.Time, 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, size and size cap (max_bytes, when it has one) — 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. Also whether the running server holds it: `in_force` is true only when the user list on the machine allows it and the server reloaded its configuration after that list was written (novox/hq issue 353: a list sent is not a list the bus enforces until it reloads).",
			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
				}
				answer, err := UserCan(c, arg(a, "user"), arg(a, "action"), arg(a, "subject"))
				if err != nil {
					return nil, err
				}
				ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
				defer cancel()
				at, werr := written(ctx)
				varz, verr := get("/varz")
				return InForce(answer, at, werr, varz, verr), nil
			}},
		{Name: "nats_snapshot_verify",
			Description: "Verifies the bus's snapshot (novox/hq ADR 0235) before an upgrade or a restore: the newest bus.tar in the module's snapshots directory, or the one named, with every file held to the manifest's checksum and every archive read to its end. Answers the file, its time and size, each stream's name and last sequence, `passed`, and the first fault when it did not. Reads only; nothing outside the snapshots directory.",
			Input:       map[string]any{"file": str("a snapshot file inside the snapshots directory, e.g. bus.tar (default: the newest .tar there)")},
			Run: func(a map[string]any) (any, error) {
				return busarchive.VerifyFile(env("MESH_NATS_SNAPSHOTS_DIR", ""), arg(a, "file"))
			}},
	}
}

func main() {
	if err := stdio.Serve("", append(append(tools(LiveMonitor(), ReadUsers, UsersWritten), removalTools(LiveBus(), SaidByTheRuntime)...), expiryTools(LiveBus(), SaidByTheRuntime, RecordedBySeat)...)); err != nil {
		fmt.Fprintln(os.Stderr, err)
		os.Exit(1)
	}
}

// The seat mesh-broker's reading verbs, which this bundle serves (design 25 § 12.3): every entry, one entry, what the
// bus holds against what was asked, and what no manifest asks for. Its acting verbs (create, apply, recreate, drop) are
// the keeper's own work, on its own account alone (keeper.go), so no other account, the tool runner's included, holds
// them.
var ReadingVerbs = []string{"list", "show", "check", "orphans"}

// Events are the bus module's own events (ADR 0595 §6, ADR 0620 §2), on its own stream broker-events (BROKER_EVENTS).
var Events = []string{"created", "refused", "drifted", "drift-cleared", "full", "full-cleared", "orphaned", "recreated", "dropped"}