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

@@ -1,6 +1,8 @@
 // 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.
+// controller composed — and nothing changes the bus. Beside the tools runs the watch (watch.go, novox/hq
+// ADR 0357), which says on the bus every user's slow consumer the server closed. stdout is the MCP
+// channel; this says nothing else.
 package main
 
 import (
@@ -29,7 +31,7 @@ func limitOf(a map[string]any, fallback int) int {
 	return fallback
 }
 
-func tools(monitor Monitor, users func(context.Context) ([]byte, error), written func(context.Context) (time.Time, error)) []stdio.Tool {
+func tools(monitor Monitor, users func(context.Context) ([]byte, error), written func(context.Context) (time.Time, error), watch *Watch) []stdio.Tool {
 	get := func(path string) ([]byte, error) {
 		ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
 		defer cancel()
@@ -112,6 +114,9 @@ func tools(monitor Monitor, users func(context.Context) ([]byte, error), written
 				}
 				return Buckets(doc)
 			}},
+		{Name: "nats_bus_watch",
+			Description: "The watch that says every bus user's slow consumer and every connection the bus closed for breaking its rules, as the events bus.slow-consumer and bus.connection-closed the controller raises as S9 (novox/hq ADR 0357): when it last read the server's record of closed connections, what it said, the newest closures said, the reasons it leaves to D15 and any it could not publish.",
+			Run:         func(map[string]any) (any, error) { return watch.Report(), nil }},
 		{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")},
@@ -144,7 +149,14 @@ func tools(monitor Monitor, users func(context.Context) ([]byte, error), written
 }
 
 func main() {
-	if err := stdio.Serve("", tools(LiveMonitor(), ReadUsers, UsersWritten)); err != nil {
+	monitor := LiveMonitor()
+	// The watch publishes through the runtime that launched this bundle, under the module's name: the
+	// events the manifest declares, and nothing else (novox/hq ADR 0357).
+	watch := NewWatch(monitor, func(event string, body any) error { return stdio.Emit(event, body) })
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go watch.Run(ctx)
+	if err := stdio.Serve("", tools(monitor, ReadUsers, UsersWritten, watch)); err != nil {
 		fmt.Fprintln(os.Stderr, err)
 		os.Exit(1)
 	}

modules/nats/cmd/nats-tools/watch.go

@@ -0,0 +1,433 @@
+package main
+
+// The watch: every user's slow consumer and every connection the bus closed for breaking its rules,
+// said on the bus (novox/hq ADR 0357, issue 270).
+//
+// **The server says these of other users only in its record of closed connections.** It closes every
+// client it judges a slow consumer, and keeps the last 10,000 closed connections with the user and the
+// reason (`/connz?state=closed&auth=1`) — about two days of the mesh's traffic. It publishes neither in
+// the mesh's account; a system account would hear them, and the bus has none. So the bundle that already
+// reads the record for `nats_closed_connections` looks at it every WatchEvery and publishes each closure
+// that matters as an event the controller raises as S9.
+//
+// **Said at least once, counted once.** A closure is its user, its connection id and when it opened —
+// the id alone is reused after the server restarts. Each one said is remembered while this process runs;
+// one whose publish failed is tried again at the next look. On start the last hour (S9's window) is said
+// again, so a restart repeats closures rather than losing them; the controller counts each by its
+// identity, so a repeat changes no count.
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"os"
+	"sort"
+	"strings"
+	"sync"
+	"time"
+)
+
+const (
+	// WatchEvery is how often the watch reads the record of closed connections. With the controller's
+	// watchdogs ticking every 30 s too, a closure is a condition about a minute after the server closed it.
+	WatchEvery = 30 * time.Second
+	// WatchSaysAgain is how far back the first look says closures again: S9's window, an hour.
+	WatchSaysAgain = time.Hour
+	// EventSlowConsumer and EventConnectionClosed are the events the watch publishes (module.json `emits`).
+	EventSlowConsumer     = "bus.slow-consumer"
+	EventConnectionClosed = "bus.connection-closed"
+	// EventWatch is the watch's heartbeat, after every look: whether it is healthy and why not. The controller
+	// raises a watch that stops saying it, or says it is not healthy (novox/hq ADR 0357, review 274): a dead
+	// watch must never read as no slow consumers.
+	EventWatch = "bus.watch"
+)
+
+// notSaid are the reasons a closure is routine: the client's own choice, the network or a machine asleep
+// (D15 says a user dropped in a loop), the server's own shutdown, and a refused login, which names no user.
+// Every other reason is said — one the module does not know included — because a new way for the bus to
+// throw a client out is exactly what must not pass in silence.
+var notSaid = map[string]bool{
+	ClientClosed:             true,
+	"Stale Connection":       true,
+	"Read Error":             true,
+	"Write Error":            true,
+	"TLS Handshake Failure":  true,
+	"Authentication Timeout": true,
+	"Authentication Failure": true,
+	"Server Shutdown":        true,
+}
+
+// isSlowConsumer says a closure's reason is the server judging the client a slow consumer: "Slow Consumer
+// (Pending Bytes)" or "Slow Consumer (Write Deadline)".
+func isSlowConsumer(reason string) bool { return strings.HasPrefix(reason, "Slow Consumer") }
+
+// Closure is one closed connection as the watch says it. The user and the connection's own name, never a
+// credential: the record the server keeps names who logged in, and nothing they logged in with.
+type Closure struct {
+	User   string    `json:"user"`
+	Name   string    `json:"name,omitempty"`
+	CID    uint64    `json:"cid"`
+	Reason string    `json:"reason"`
+	Opened time.Time `json:"opened"`
+	Closed time.Time `json:"closed"`
+	// Lived is how long the connection was open, in seconds.
+	Lived float64 `json:"lived_seconds"`
+	// Pending is what the server still had to write to it when it closed, in bytes, where it says.
+	Pending int64 `json:"pending_bytes,omitempty"`
+}
+
+// identity is what tells one closure from another across a server restart.
+func (c Closure) identity() string {
+	return fmt.Sprintf("%s/%d/%s", c.User, c.CID, c.Opened.UTC().Format(time.RFC3339Nano))
+}
+
+// event is the event a closure is said as.
+func (c Closure) event() string {
+	if isSlowConsumer(c.Reason) {
+		return EventSlowConsumer
+	}
+	return EventConnectionClosed
+}
+
+// Closures are the closed connections in the record that are to be said, closed at or after since, oldest
+// first.
+func Closures(connz []byte, since time.Time) ([]Closure, error) {
+	var doc struct {
+		Connections []struct {
+			CID            uint64     `json:"cid"`
+			Name           string     `json:"name"`
+			AuthorizedUser string     `json:"authorized_user"`
+			Start          time.Time  `json:"start"`
+			Stop           *time.Time `json:"stop"`
+			Reason         string     `json:"reason"`
+			Pending        int64      `json:"pending_bytes"`
+		} `json:"connections"`
+	}
+	if err := json.Unmarshal(connz, &doc); err != nil {
+		return nil, fmt.Errorf("the bus's record of closed connections cannot be read: %w", err)
+	}
+	var out []Closure
+	for _, c := range doc.Connections {
+		if c.Stop == nil || c.Stop.Before(since) || c.AuthorizedUser == "" || notSaid[c.Reason] {
+			continue
+		}
+		reason := c.Reason
+		if reason == "" {
+			reason = "unsaid"
+		}
+		out = append(out, Closure{User: c.AuthorizedUser, Name: c.Name, CID: c.CID, Reason: reason,
+			Opened: c.Start.UTC(), Closed: c.Stop.UTC(), Lived: c.Stop.Sub(c.Start).Seconds(), Pending: c.Pending})
+	}
+	sort.SliceStable(out, func(i, j int) bool { return out[i].Closed.Before(out[j].Closed) })
+	return out, nil
+}
+
+// Watch is the watch's state, shared with the tools that report it.
+type Watch struct {
+	monitor Monitor
+	emit    func(string, any) error
+	now     func() time.Time
+	started time.Time
+
+	mu sync.Mutex
+	// said is every closure said, by identity, with when it closed: forgotten once older than the hour, since
+	// the record is never read that far back again.
+	said map[string]time.Time
+	// failed are the closures whose publish failed, by identity, tried again at every look until said or
+	// older than the hour.
+	failed map[string]failedClosure
+	// lastLook is when the watch last looked; lastRead when a look last read the record whole.
+	lastLook, lastRead time.Time
+	// failingSince is when the reads began to fail, zero while they succeed.
+	failingSince time.Time
+	lastErr      string
+	saidN        map[string]int
+	newest       []Closure
+	// faults are the looks that panicked and were recovered: the loop goes on, and each is said.
+	faults    int
+	lastFault string
+}
+
+type failedClosure struct {
+	closure Closure
+	err     string
+	since   time.Time
+}
+
+// NewWatch is a watch that has not looked yet.
+func NewWatch(monitor Monitor, emit func(string, any) error) *Watch {
+	return &Watch{monitor: monitor, emit: emit, now: time.Now, started: time.Now(), said: map[string]time.Time{},
+		saidN: map[string]int{}, failed: map[string]failedClosure{}}
+}
+
+// readPage is how many closed connections one request asks for, newest first.
+const readPage = 256
+
+// readSince reads the record newest first (`sort=stop`), a page at a time, until a page reaches back past since
+// or the record ends: a look reads only what closed since the last one, not all 10,000 kept.
+func (w *Watch) readSince(ctx context.Context, since time.Time) ([]Closure, error) {
+	var out []Closure
+	for offset := 0; offset < maxClosedKept; offset += readPage {
+		page, err := w.monitor(ctx, fmt.Sprintf("/connz?state=closed&auth=1&sort=stop&limit=%d&offset=%d", readPage, offset))
+		if err != nil {
+			return nil, err
+		}
+		closures, oldest, n, err := closuresOfPage(page, since)
+		if err != nil {
+			return nil, err
+		}
+		out = append(out, closures...)
+		if n < readPage || (!oldest.IsZero() && oldest.Before(since)) {
+			break
+		}
+	}
+	sort.SliceStable(out, func(i, j int) bool { return out[i].Closed.Before(out[j].Closed) })
+	return out, nil
+}
+
+// closuresOfPage is one page's closures to say, its oldest stop and how many connections it held.
+func closuresOfPage(page []byte, since time.Time) ([]Closure, time.Time, int, error) {
+	var doc struct {
+		Connections []struct {
+			Stop *time.Time `json:"stop"`
+		} `json:"connections"`
+	}
+	if err := json.Unmarshal(page, &doc); err != nil {
+		return nil, time.Time{}, 0, fmt.Errorf("the bus's record of closed connections cannot be read: %w", err)
+	}
+	var oldest time.Time
+	for _, c := range doc.Connections {
+		if c.Stop != nil && (oldest.IsZero() || c.Stop.Before(oldest)) {
+			oldest = *c.Stop
+		}
+	}
+	closures, err := Closures(page, since)
+	return closures, oldest, len(doc.Connections), err
+}
+
+// Look reads what closed since the last look and says every closure not said yet, and tries again every
+// closure whose publish failed. A fault in a look is recovered and said: the watch goes on.
+func (w *Watch) Look(ctx context.Context) {
+	defer func() {
+		if r := recover(); r != nil {
+			w.mu.Lock()
+			w.faults++
+			w.lastFault = fmt.Sprintf("a look stopped on a fault at %s: %v", w.now().UTC().Format(time.RFC3339), r)
+			w.mu.Unlock()
+			fmt.Fprintln(os.Stderr, w.lastFault)
+		}
+	}()
+	now := w.now()
+	hour := now.Add(-WatchSaysAgain)
+	w.mu.Lock()
+	// The first look says the last hour again; a later one reads from one interval before the last whole
+	// read, so a closure the server recorded late is not missed — the said set tells what was said.
+	since := hour
+	if !w.lastRead.IsZero() && w.lastRead.Add(-WatchEvery).After(hour) {
+		since = w.lastRead.Add(-WatchEvery)
+	}
+	w.mu.Unlock()
+	read, cancel := context.WithTimeout(ctx, 20*time.Second)
+	defer cancel()
+	closures, err := w.readSince(read, since)
+
+	w.mu.Lock()
+	w.lastLook = now
+	for id, closed := range w.said {
+		if closed.Before(hour) {
+			delete(w.said, id)
+		}
+	}
+	for id, f := range w.failed {
+		if f.closure.Closed.Before(hour) {
+			delete(w.failed, id) // S9 would no longer say it
+		}
+	}
+	if err != nil {
+		w.lastErr = err.Error()
+		if w.failingSince.IsZero() {
+			w.failingSince = now
+		}
+		fmt.Fprintf(os.Stderr, "the watch could not read the bus's closed connections: %v\n", err)
+	} else {
+		w.lastErr, w.failingSince, w.lastRead = "", time.Time{}, now
+	}
+	todo := map[string]Closure{}
+	for _, c := range closures {
+		if _, done := w.said[c.identity()]; !done {
+			todo[c.identity()] = c
+		}
+	}
+	for id, f := range w.failed {
+		if _, done := w.said[id]; !done {
+			todo[id] = f.closure
+		}
+	}
+	w.mu.Unlock()
+
+	order := make([]Closure, 0, len(todo))
+	for _, c := range todo {
+		order = append(order, c)
+	}
+	sort.SliceStable(order, func(i, j int) bool { return order[i].Closed.Before(order[j].Closed) })
+	for _, c := range order {
+		if w.emit == nil {
+			continue
+		}
+		err := w.emit(c.event(), c)
+		w.mu.Lock()
+		if err != nil {
+			// Tried again at the next look: it stays out of the said set.
+			f := w.failed[c.identity()]
+			if f.since.IsZero() {
+				f.since = now
+			}
+			f.closure, f.err = c, err.Error()
+			w.failed[c.identity()] = f
+			w.mu.Unlock()
+			fmt.Fprintf(os.Stderr, "%s for %s not published, tried again at the next look: %v\n", c.event(), c.User, err)
+			continue
+		}
+		delete(w.failed, c.identity())
+		w.said[c.identity()] = c.Closed
+		w.saidN[c.event()]++
+		w.newest = append(w.newest, c)
+		if len(w.newest) > 10 {
+			w.newest = w.newest[len(w.newest)-10:]
+		}
+		w.mu.Unlock()
+	}
+	w.beat()
+}
+
+// Beat is the watch's heartbeat as it publishes it.
+type Beat struct {
+	At       time.Time  `json:"at"`
+	Healthy  bool       `json:"healthy"`
+	Why      string     `json:"why,omitempty"`
+	LastRead *time.Time `json:"last_read,omitempty"`
+	Every    string     `json:"every"`
+}
+
+// beat says the watch is there, and whether it is healthy. One that cannot be published is the controller's to
+// notice by its absence.
+func (w *Watch) beat() {
+	if w.emit == nil {
+		return
+	}
+	w.mu.Lock()
+	healthy, why := w.healthLocked()
+	b := Beat{At: w.now().UTC(), Healthy: healthy, Why: why, Every: WatchEvery.String()}
+	if !w.lastRead.IsZero() {
+		at := w.lastRead.UTC()
+		b.LastRead = &at
+	}
+	w.mu.Unlock()
+	if err := w.emit(EventWatch, b); err != nil {
+		fmt.Fprintf(os.Stderr, "the watch's heartbeat not published: %v\n", err)
+	}
+}
+
+// Run is the watch's life: a look every WatchEvery until the context ends. A look that faults is recovered in
+// Look, so the loop never ends but with the context.
+func (w *Watch) Run(ctx context.Context) {
+	// The first look one interval in, not at once: the runtime that publishes for this bundle answers only
+	// once the bundle serves, and the first look reads the last hour back anyway.
+	t := time.NewTicker(WatchEvery)
+	defer t.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-t.C:
+		}
+		w.Look(ctx)
+	}
+}
+
+// WatchStale is how long the watch may go without reading the record whole, or keep a closure unpublished,
+// before it is not healthy: three looks.
+const WatchStale = 3 * WatchEvery
+
+// Health is the watch's own verdict: healthy while it has read the record whole within WatchStale (after the
+// same grace from its start) and holds no closure unpublished for longer. Said in every heartbeat, so the
+// controller raises a watch that cannot read (novox/hq ADR 0357, review 274).
+func (w *Watch) Health() (bool, string) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	return w.healthLocked()
+}
+
+func (w *Watch) healthLocked() (bool, string) {
+	now := w.now()
+	switch {
+	case w.lastRead.IsZero() && now.Sub(w.started) > WatchStale:
+		why := fmt.Sprintf("the watch of slow consumers has not read the bus's closed connections since it started %s ago",
+			now.Sub(w.started).Round(time.Second))
+		if w.lastErr != "" {
+			why += ": " + w.lastErr
+		}
+		return false, why
+	case !w.lastRead.IsZero() && now.Sub(w.lastRead) > WatchStale:
+		why := fmt.Sprintf("the watch of slow consumers last read the bus's closed connections %s ago",
+			now.Sub(w.lastRead).Round(time.Second))
+		if w.lastErr != "" {
+			why += ": " + w.lastErr
+		}
+		return false, why
+	}
+	for _, f := range w.failed {
+		if now.Sub(f.since) > WatchStale {
+			return false, fmt.Sprintf("the watch could not publish %s for %s for %s: %s", f.closure.event(),
+				f.closure.User, now.Sub(f.since).Round(time.Second), f.err)
+		}
+	}
+	return true, ""
+}
+
+// Report is the watch as the tool `nats_bus_watch` shows it.
+func (w *Watch) Report() map[string]any {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	healthy, why := w.healthLocked()
+	out := map[string]any{
+		"healthy":     healthy,
+		"every":       WatchEvery.String(),
+		"reads_back":  WatchSaysAgain.String(),
+		"events":      []string{EventSlowConsumer, EventConnectionClosed},
+		"said":        w.saidN,
+		"remembered":  len(w.said),
+		"newest_said": w.newest,
+		"not_said_reasons": func() []string {
+			var r []string
+			for k := range notSaid {
+				r = append(r, k)
+			}
+			sort.Strings(r)
+			return r
+		}(),
+	}
+	if why != "" {
+		out["why"] = why
+	}
+	if w.faults > 0 {
+		out["faults_recovered"], out["last_fault"] = w.faults, w.lastFault
+	}
+	if !w.lastRead.IsZero() {
+		out["last_read"] = w.lastRead.UTC()
+	}
+	if !w.lastLook.IsZero() {
+		out["last_look"] = w.lastLook.UTC()
+	}
+	if w.lastErr != "" {
+		out["last_error"] = w.lastErr
+	}
+	if len(w.failed) > 0 {
+		failed := map[string]string{}
+		for id, f := range w.failed {
+			failed[id] = f.err
+		}
+		out["not_published"] = failed
+	}
+	return out
+}

modules/nats/cmd/nats-tools/watch_test.go

@@ -0,0 +1,336 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"os"
+	"sort"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+)
+
+// The watch (novox/hq ADR 0357, issue 270): every user's slow consumer and every connection the bus closed
+// for breaking its rules, said once as an event the controller raises as S9.
+
+var watchNow = time.Date(2026, 10, 14, 12, 0, 0, 0, time.UTC)
+
+// closedRecord is a record of closed connections as the server serves it, with a password-shaped field the
+// server never serves, to prove nothing but the user is carried.
+const closedRecord = `{"connections":[
+ {"cid":10,"name":"g14.claude-code","authorized_user":"g14.claude-code","start":"2026-10-14T11:40:00Z","stop":"2026-10-14T11:58:00Z","reason":"Slow Consumer (Write Deadline)","pending_bytes":334000},
+ {"cid":11,"name":"mesh-controller verb node","authorized_user":"controller","start":"2026-10-14T11:58:00Z","stop":"2026-10-14T11:58:01Z","reason":"Client Closed"},
+ {"cid":12,"name":"mesh-host/g14","authorized_user":"node.g14","start":"2026-10-14T10:00:00Z","stop":"2026-10-14T11:59:00Z","reason":"Stale Connection"},
+ {"cid":13,"name":"g14.node-tools","authorized_user":"g14.node-tools","start":"2026-10-14T10:00:00Z","stop":"2026-10-14T11:59:10Z","reason":"Read Error"},
+ {"cid":14,"name":"shanks.desk","authorized_user":"shanks.desk","start":"2026-10-14T11:00:00Z","stop":"2026-10-14T11:59:20Z","reason":"Maximum Message Payload Exceeded"},
+ {"cid":15,"name":"","authorized_user":"","start":"2026-10-14T11:59:00Z","stop":"2026-10-14T11:59:30Z","reason":"Authentication Failure"},
+ {"cid":16,"name":"ace.media","authorized_user":"ace.media","start":"2026-10-14T09:00:00Z","stop":"2026-10-14T10:30:00Z","reason":"Slow Consumer (Pending Bytes)"},
+ {"cid":17,"name":"ace.future","authorized_user":"ace.future","start":"2026-10-14T11:00:00Z","stop":"2026-10-14T11:59:40Z","reason":"A Reason Not Yet Invented","password":"never-carried"},
+ {"cid":18,"name":"ace.route-proxy","authorized_user":"ace.route-proxy","start":"2026-10-14T11:00:00Z","stop":"2026-10-14T11:59:50Z","reason":"Server Shutdown"}
+]}`
+
+type published struct {
+	event string
+	body  Closure
+}
+
+type recorder struct {
+	mu    sync.Mutex
+	got   []published
+	beats []Beat
+	fail  error
+}
+
+func (r *recorder) emit(event string, body any) error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if r.fail != nil {
+		return r.fail
+	}
+	raw, _ := json.Marshal(body)
+	if event == EventWatch {
+		var b Beat
+		_ = json.Unmarshal(raw, &b)
+		r.beats = append(r.beats, b)
+		return nil
+	}
+	if strings.Contains(string(raw), "never-carried") || strings.Contains(strings.ToLower(string(raw)), "password") {
+		panic("a credential was published: " + string(raw))
+	}
+	var c Closure
+	_ = json.Unmarshal(raw, &c)
+	r.got = append(r.got, published{event, c})
+	return nil
+}
+
+func (r *recorder) said() []string {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	var out []string
+	for _, p := range r.got {
+		out = append(out, p.event+" "+p.body.User)
+	}
+	sort.Strings(out)
+	return out
+}
+
+func recordOf(doc *string) Monitor {
+	return func(_ context.Context, path string) ([]byte, error) {
+		if !strings.HasPrefix(path, "/connz?state=closed&auth=1") {
+			return nil, errors.New("the watch asked " + path)
+		}
+		return []byte(*doc), nil
+	}
+}
+
+func watchAt(doc *string, r *recorder) *Watch {
+	w := NewWatch(recordOf(doc), r.emit)
+	w.now = func() time.Time { return watchNow }
+	return w
+}
+
+func TestTheWatchSaysEachClosureTheServerDroppedNamingItsUserAndNothingRoutine(t *testing.T) {
+	doc := closedRecord
+	r := &recorder{}
+	watchAt(&doc, r).Look(context.Background())
+	want := []string{
+		"bus.connection-closed ace.future",  // a reason the module does not know is said
+		"bus.connection-closed shanks.desk", // a limit broken
+		"bus.slow-consumer g14.claude-code", // the slow consumer research 036 could not name
+	}
+	if got := r.said(); strings.Join(got, "|") != strings.Join(want, "|") {
+		t.Fatalf("said %v, want %v (none for the client's own close, the network, a refused login, the "+
+			"server's shutdown, or a closure older than the hour)", got, want)
+	}
+	for _, p := range r.got {
+		if p.body.User == "g14.claude-code" {
+			if p.body.CID != 10 || p.body.Reason != "Slow Consumer (Write Deadline)" || p.body.Pending != 334000 ||
+				!p.body.Closed.Equal(time.Date(2026, 10, 14, 11, 58, 0, 0, time.UTC)) || p.body.Lived != 18*60 {
+				t.Fatalf("the slow consumer said as %+v", p.body)
+			}
+		}
+	}
+}
+
+func TestTheWatchSaysAClosureOnceWhileItRuns(t *testing.T) {
+	doc := closedRecord
+	r := &recorder{}
+	w := watchAt(&doc, r)
+	w.Look(context.Background())
+	w.Look(context.Background())
+	if got := len(r.said()); got != 3 {
+		t.Fatalf("said %d events over two looks at the same record, want 3: %v", got, r.said())
+	}
+}
+
+func TestAClosureWhosePublishFailedIsTriedAgainAtTheNextLook(t *testing.T) {
+	doc := closedRecord
+	r := &recorder{fail: errors.New("the runtime did not answer")}
+	w := watchAt(&doc, r)
+	w.Look(context.Background())
+	if len(r.said()) != 0 {
+		t.Fatal("a failed publish was counted as said")
+	}
+	if report := w.Report(); len(report["not_published"].(map[string]string)) != 3 {
+		t.Fatalf("the watch does not say what it could not publish: %v", report)
+	}
+	r.fail = nil
+	w.Look(context.Background())
+	if got := len(r.said()); got != 3 {
+		t.Fatalf("said %d after the runtime came back, want the 3 that failed", got)
+	}
+	if _, left := w.Report()["not_published"]; left {
+		t.Fatal("a closure published at last is still said not published")
+	}
+}
+
+func TestAWatchStartedAgainSaysTheLastHourAgain(t *testing.T) {
+	doc := closedRecord
+	first := &recorder{}
+	watchAt(&doc, first).Look(context.Background())
+	again := &recorder{}
+	watchAt(&doc, again).Look(context.Background())
+	if strings.Join(first.said(), "|") != strings.Join(again.said(), "|") || len(again.said()) != 3 {
+		t.Fatalf("a restarted watch said %v, the first %v: the last hour must be said again, for the controller "+
+			"to count once", again.said(), first.said())
+	}
+}
+
+func TestAReusedConnectionIDWithANewOpeningIsANewClosure(t *testing.T) {
+	doc := `{"connections":[{"cid":10,"authorized_user":"g14.claude-code","start":"2026-10-14T11:40:00Z","stop":"2026-10-14T11:50:00Z","reason":"Slow Consumer (Pending Bytes)"}]}`
+	r := &recorder{}
+	w := watchAt(&doc, r)
+	w.Look(context.Background())
+	// The server restarted and numbered connections from one again.
+	doc = `{"connections":[
+ {"cid":10,"authorized_user":"g14.claude-code","start":"2026-10-14T11:40:00Z","stop":"2026-10-14T11:50:00Z","reason":"Slow Consumer (Pending Bytes)"},
+ {"cid":10,"authorized_user":"g14.claude-code","start":"2026-10-14T12:00:05Z","stop":"2026-10-14T12:00:20Z","reason":"Slow Consumer (Pending Bytes)"}]}`
+	w.now = func() time.Time { return watchNow.Add(WatchEvery) }
+	w.Look(context.Background())
+	if got := len(r.said()); got != 2 {
+		t.Fatalf("said %d, want 2: the same connection id opened again is another closure", got)
+	}
+}
+
+func TestTheWatchSaysItCannotReadTheRecord(t *testing.T) {
+	r := &recorder{}
+	w := NewWatch(func(context.Context, string) ([]byte, error) { return nil, errors.New("connection refused") }, r.emit)
+	w.Look(context.Background())
+	if report := w.Report(); !strings.Contains(report["last_error"].(string), "connection refused") {
+		t.Fatalf("a record that could not be read is not said: %v", report)
+	}
+}
+
+// **A dead watch is heard** (review 274): every look ends with a heartbeat saying whether the watch is
+// healthy; a record it cannot read for longer than three looks is said not healthy, with why, rather than read
+// as no slow consumers; and it recovers with a read.
+func TestAWatchThatCannotReadForThreeLooksSaysSoInItsHeartbeat(t *testing.T) {
+	broken := true
+	doc := closedRecord
+	clock := watchNow
+	r := &recorder{}
+	w := NewWatch(func(ctx context.Context, path string) ([]byte, error) {
+		if broken {
+			return nil, errors.New("connection refused")
+		}
+		return recordOf(&doc)(ctx, path)
+	}, r.emit)
+	w.now, w.started = func() time.Time { return clock }, clock
+	w.Look(context.Background())
+	if len(r.beats) != 1 || !r.beats[0].Healthy {
+		t.Fatalf("a watch that just started beat %+v", r.beats)
+	}
+	clock = clock.Add(WatchStale + time.Second)
+	w.Look(context.Background())
+	last := r.beats[len(r.beats)-1]
+	if last.Healthy || !strings.Contains(last.Why, "connection refused") || !last.At.Equal(clock) {
+		t.Fatalf("a watch that has not read for %s beat %+v", WatchStale, last)
+	}
+	broken = false
+	w.Look(context.Background())
+	if last := r.beats[len(r.beats)-1]; !last.Healthy || last.LastRead == nil {
+		t.Fatalf("a watch that reads again beat %+v", last)
+	}
+}
+
+// A closure that cannot be published for longer than three looks is not healthy either.
+func TestAClosureUnpublishedForThreeLooksIsNotHealthy(t *testing.T) {
+	doc := closedRecord
+	r := &recorder{fail: errors.New("the runtime did not answer")}
+	clock := watchNow
+	w := watchAt(&doc, r)
+	w.now, w.started = func() time.Time { return clock }, clock
+	w.Look(context.Background())
+	clock = clock.Add(WatchStale + time.Second)
+	w.Look(context.Background())
+	if ok, why := w.Health(); ok || !strings.Contains(why, "could not publish") {
+		t.Fatalf("closures unpublished for %s read healthy %v: %q", WatchStale, ok, why)
+	}
+}
+
+// **A look that panics does not end the watch** (review 274): it is recovered and said, and the next look runs.
+func TestALookThatPanicsIsRecoveredAndTheWatchGoesOn(t *testing.T) {
+	calls := 0
+	doc := closedRecord
+	r := &recorder{}
+	w := NewWatch(func(ctx context.Context, path string) ([]byte, error) {
+		calls++
+		if calls == 1 {
+			panic("a nil map somewhere")
+		}
+		return recordOf(&doc)(ctx, path)
+	}, r.emit)
+	w.now = func() time.Time { return watchNow }
+	w.Look(context.Background())
+	w.Look(context.Background())
+	if len(r.said()) != 3 {
+		t.Fatalf("after a look that panicked the next said %v", r.said())
+	}
+	if report := w.Report(); report["faults_recovered"] != 1 {
+		t.Fatalf("the recovered fault is not said: %v", report)
+	}
+}
+
+// **The record is read newest first, and only back to the last look** (review 274), never all 10,000 kept.
+func TestALaterLookReadsOnlyWhatClosedSinceTheLastOne(t *testing.T) {
+	var asked []string
+	page := func(n int, newest time.Time) []byte {
+		var conns []string
+		for i := range n {
+			stop := newest.Add(-time.Duration(i) * time.Second)
+			conns = append(conns, fmt.Sprintf(`{"cid":%d,"authorized_user":"u","start":"%s","stop":"%s","reason":"Client Closed"}`,
+				1000-i, stop.Add(-time.Minute).Format(time.RFC3339), stop.Format(time.RFC3339)))
+		}
+		return []byte(`{"connections":[` + strings.Join(conns, ",") + `]}`)
+	}
+	clock := watchNow
+	w := NewWatch(func(_ context.Context, path string) ([]byte, error) {
+		asked = append(asked, path)
+		if !strings.Contains(path, "sort=stop") {
+			return nil, errors.New("not read newest first: " + path)
+		}
+		if strings.Contains(path, "offset=0") {
+			return page(readPage, clock), nil // a full page of the last few minutes
+		}
+		return page(readPage, clock.Add(-2*time.Hour)), nil // older than any look reaches
+	}, (&recorder{}).emit)
+	w.now = func() time.Time { return clock }
+	w.Look(context.Background())
+	first := len(asked)
+	asked = nil
+	clock = clock.Add(WatchEvery)
+	w.Look(context.Background())
+	if len(asked) != 1 || first > 2 {
+		t.Fatalf("the first look asked %d pages and a later one %d (%v): a later look reads back only to the last", first,
+			len(asked), asked)
+	}
+}
+
+// The failed-publish record forgets a closure older than the hour: S9 would no longer say it.
+func TestAClosureThatCouldNotBePublishedIsForgottenAfterTheHour(t *testing.T) {
+	doc := closedRecord
+	r := &recorder{fail: errors.New("the runtime did not answer")}
+	clock := watchNow
+	w := watchAt(&doc, r)
+	w.now = func() time.Time { return clock }
+	w.Look(context.Background())
+	doc = `{"connections":[]}`
+	clock = clock.Add(2 * time.Hour)
+	w.Look(context.Background())
+	if _, left := w.Report()["not_published"]; left {
+		t.Fatal("closures older than the hour are still kept as not published")
+	}
+}
+
+func TestTheManifestDeclaresTheToolsServedTheWatchesEventsAndItsInterval(t *testing.T) {
+	raw, err := os.ReadFile("../../module.json")
+	if err != nil {
+		t.Fatal(err)
+	}
+	var m struct {
+		Tools []string `json:"tools"`
+		Emits []string `json:"emits"`
+	}
+	if err := json.Unmarshal(raw, &m); err != nil {
+		t.Fatal(err)
+	}
+	var served []string
+	for _, tool := range tools(nil, nil, nil, NewWatch(nil, nil)) {
+		served = append(served, tool.Name)
+	}
+	sort.Strings(served)
+	sort.Strings(m.Tools)
+	if strings.Join(served, ",") != strings.Join(m.Tools, ",") {
+		t.Fatalf("served %v, the manifest lists %v", served, m.Tools)
+	}
+	sort.Strings(m.Emits)
+	if strings.Join(m.Emits, ",") != EventConnectionClosed+","+EventSlowConsumer+","+EventWatch {
+		t.Fatalf("the manifest emits %v: the runtime may publish only what is declared", m.Emits)
+	}
+	if WatchEvery != 30*time.Second || WatchSaysAgain != time.Hour {
+		t.Fatalf("the watch looks every %s and reads back %s; ADR 0357 says 30s and an hour", WatchEvery, WatchSaysAgain)
+	}
+}

modules/nats/module.json

@@ -24,7 +24,11 @@
   "capabilities": [
     "container-runtime"
   ],
-  "emits": [],
+  "emits": [
+    "bus.slow-consumer",
+    "bus.connection-closed",
+    "bus.watch"
+  ],
   "consumes": [],
   "listens": [
     {
@@ -39,6 +43,7 @@
     "nats_server",
     "nats_connections",
     "nats_closed_connections",
+    "nats_bus_watch",
     "nats_subscriptions",
     "nats_streams",
     "nats_backlog",