jschoubben
4fdb047cc8
mesh/merge-gate pass: builds nats, sent nowhere; merging this needs a person's bus upgrade; 1 wait(s) for a person; every machine composes with the change …
mesh/repo-check pass: its merge-check.sh passed, each part in its toolchain: merge-check.sh (go), merge-check-typescript.sh (typescript)
mesh/delivery delivered
nats_user_can read the user list on the machine. A list written is not a list the bus enforces until the server reloads it: on 2026-10-09 the controller asked between the write and the reload's effect. The answer now carries in_force — allowed in the file, and the server's config_load_time after the file's write — with why when it is not, and anything not known is not in force.
152 lines
7.1 KiB
Go
152 lines
7.1 KiB
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), 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 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. 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
|
|
}},
|
|
}
|
|
}
|
|
|
|
func main() {
|
|
if err := stdio.Serve("", tools(LiveMonitor(), ReadUsers, UsersWritten)); err != nil {
|
|
fmt.Fprintln(os.Stderr, err)
|
|
os.Exit(1)
|
|
}
|
|
}
|