Merge pull request 'S9 raises every bus user's slow consumer and closed connection, from the bus module's events (hq issue 270, ADR 0357)' (#326) from fix/270-the-bus-module-says-every-users-slow-consumer into main

A storm of closures must not spend a watchdog tick, so the read stops at 2000
events and says the count is a floor. A failed read is its own condition
rather than blinding S9, so the controller's own signals are still said. The
controller's fall behind, told to its client and closed by the server, is one
fault, counted once.

cmd/mesh-controller/bus_users_closed_test.go

@@ -0,0 +1,92 @@
+package main
+
+import (
+	"errors"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/novox/mesh-controller/internal/conditions"
+	"github.com/novox/mesh-controller/internal/link"
+)
+
+// **Every user's slow consumer is S9's, not the controller's alone** (novox/hq ADR 0357, issue 270). On
+// 2026-10-07 the bus closed two clients on the laptop as slow consumers and nothing raised a condition:
+// the controller heard only its own connection. Said by the bus module and read back, a user the bus
+// closes is one condition naming the module and its machine, raised at the first look and cleared an hour
+// after the last closure.
+func TestAUserTheBusClosedAsASlowConsumerIsOneConditionNamingItsModuleAndMachine(t *testing.T) {
+	closed := time.Date(2026, 10, 14, 12, 0, 0, 0, time.UTC)
+	now := closed.Add(20 * time.Second)
+	store := conditions.NewInMemory()
+	k := conditions.NewKeeper(t.Context(), conditions.Options{Store: store, History: store,
+		Teller: &conditions.Told{}, Now: func() time.Time { return now }})
+	t.Cleanup(func() { k.Close(t.Context()) })
+
+	advisories := link.BusClosures([]link.BusClosure{
+		{User: "laptop.claude-code", CID: 10, Reason: "Slow Consumer (Write Deadline)", Opened: closed.Add(-time.Hour),
+			Closed: closed.Add(-10 * time.Minute)},
+		{User: "laptop.claude-code", CID: 44, Reason: "Slow Consumer (Write Deadline)", Opened: closed.Add(-time.Minute),
+			Closed: closed},
+		{User: "anchor.desk", CID: 7, Reason: "Maximum Message Payload Exceeded", Opened: closed.Add(-time.Hour),
+			Closed: closed},
+	}, []string{link.AdvisorySlowConsumer, link.AdvisorySlowConsumer, link.AdvisoryConnectionClosed}, closed.Add(-time.Hour))
+	look := func() {
+		t.Helper()
+		f := &signalFacts{now: now, host: "anchor", advisories: advisories}
+		if err := k.Reconcile(t.Context(), "S9", watchAdvisories(f)); err != nil {
+			t.Fatal(err)
+		}
+	}
+	look()
+
+	slow, found, err := conditions.ReadOne(t.Context(), store, "bus.laptop.claude-code.slow-consumer")
+	if err != nil || !found {
+		t.Fatalf("no slow-consumer condition for the user: %v %v", found, err)
+	}
+	if slow.Subject.Machine != "laptop" || slow.Observations != 2 || !slow.LastObserved.Equal(closed) {
+		t.Fatalf("the slow consumer is said on %q, %d closure(s), last %s: %+v", slow.Subject.Machine, slow.Observations,
+			slow.LastObserved, slow)
+	}
+	if !strings.Contains(slow.Summary, "claude-code on laptop") || strings.Contains(slow.Summary, "laptop.claude-code") {
+		t.Fatalf("the summary does not name the module in words, or carries the dotted user: %q", slow.Summary)
+	}
+	if !strings.Contains(slow.Headline, "Claude-code on laptop") {
+		t.Fatalf("the headline is %q", slow.Headline)
+	}
+	cut, found, err := conditions.ReadOne(t.Context(), store, "bus.anchor.desk.connection-closed")
+	if err != nil || !found || cut.Subject.Machine != "anchor" || !strings.Contains(cut.Summary, "maximum message payload exceeded") {
+		t.Fatalf("the user cut off for breaking the bus's rules: %v %v %+v", found, err, cut)
+	}
+
+	// An hour after the newest closure without another, both clear.
+	now = closed.Add(time.Hour + time.Minute)
+	look()
+	for _, key := range []string{"bus.laptop.claude-code.slow-consumer", "bus.anchor.desk.connection-closed"} {
+		if _, found, _ := k.Get(t.Context(), key); found {
+			t.Fatalf("%s is still open an hour after the last closure", key)
+		}
+	}
+}
+
+// **A read of the bus module's events that fails does not blind S9** (review of #326): the controller's own
+// signals are still said, and the failed read is a condition of its own.
+func TestABusModuleReadThatFailsKeepsTheControllersOwnSignals(t *testing.T) {
+	at := time.Date(2026, 10, 14, 12, 0, 0, 0, time.UTC)
+	f := &signalFacts{now: at.Add(time.Minute), host: "anchor",
+		advisories: []link.Advisory{{Kind: link.AdvisorySlowConsumer, ID: "controller", Said: "the controller fell behind",
+			First: at, Last: at, Count: 1}},
+		busFaultsErr: errors.New("nats: timeout")}
+	keys := map[string]bool{}
+	for _, o := range watchAdvisories(f) {
+		keys[o.Key()] = true
+	}
+	if !keys["bus.controller.slow-consumer"] || !keys["bus.bus-module.bus-faults-unread"] {
+		t.Fatalf("observed %v: the controller's own slow consumer and the unread events, both", keys)
+	}
+	for _, row := range signalsTable {
+		if row.Row == "S9" && row.needs(f) != nil {
+			t.Fatal("the S9 row went blind for a read of the bus module's events")
+		}
+	}
+}

cmd/mesh-controller/plain_words.go

@@ -620,6 +620,18 @@ var plainWordings = map[string]func(conditions.Observation) words{
 			Explanation: "The bus reports a listener too slow to keep up, so messages to it are late.",
 			Resolved:    "The listener keeps up again"}
 	}),
+	kindBusFaultsUnread: worded(func(o conditions.Observation) words {
+		return words{Headline: "Slow clients of the bus cannot be checked",
+			Explanation: "The controller could not read what the bus said about its clients' connections, so a " +
+				"client the bus cut off would go unnoticed. It tries again at every check.",
+			Resolved: "Resolved: the bus's clients are checked again"}
+	}),
+	"connection-closed": worded(func(o conditions.Observation) words {
+		return words{Headline: "The bus cut a client off",
+			Explanation: "The bus closed a client's connection because it broke the bus's rules, so what it was " +
+				"sending was lost.",
+			Resolved: "Resolved: the client keeps to the bus's rules"}
+	}),
 	kindBusReconnects: worded(func(o conditions.Observation) words {
 		return words{Headline: "A client keeps losing the bus",
 			Explanation: "One of the mesh's clients lost its connection to the bus again and again in the last hour. " +

cmd/mesh-controller/signals.go

@@ -153,11 +153,15 @@ var signalsTable = []signalRow{
 			return newestOf(f.standings, func(c conditions.Condition) time.Time { return c.LastObserved })
 		}},
 	{Row: "S9", Signal: "bus advisories: maximum deliveries, consumer deleted; the controller's own slow " +
-		"consumer and refused subjects", Emitter: "bus server's advisory subjects; the controller's connection",
+		"consumer and refused subjects; every user's connection the server closed as a slow consumer or for " +
+		"breaking the bus's rules (ADR 0357)",
+		Emitter: "bus server's advisory subjects; the controller's connection; the bus module's watch of its " +
+			"server (bus.slow-consumer, bus.connection-closed)",
 		Trigger: "any", Bound: "any occurrence; clears after an hour without another, a deleted consumer " +
 			"once it exists again or the mesh no longer expects it, and a message given up on once DEAD_LETTERS " +
 			"no longer holds it (novox/hq issue 330)",
-		Kind: "slow-consumer, max-deliveries, refused, consumer-lost", Severity: conditions.Warning, Phase: 1,
+		Kind: "slow-consumer, connection-closed, max-deliveries, refused, consumer-lost", Severity: conditions.Warning,
+		Phase: 1,
 		needs: func(f *signalFacts) error { return f.advisoriesErr }, watch: watchAdvisories,
 		newest: func(f *signalFacts) time.Time {
 			return newestOf(f.advisories, func(a link.Advisory) time.Time { return a.Last })
@@ -569,6 +573,9 @@ func watchAdvisories(f *signalFacts) []conditions.Observation {
 			// The advisory is remembered and read again on every look for an hour: the condition counts
 			// what the bus said and when, not the looks (novox/hq issue 402).
 			Happened: a.Last, Times: a.Count}
+		if a.User != "" && a.Reason != "" {
+			busUserClosed(&o, a, f.host)
+		}
 		if a.Kind == link.AdvisoryMaxDeliveries && a.Token == link.AdvisoryNotKept {
 			o.Machine = consumerMachine(a.Stream, a.Consumer)
 			o.Headline = clip(conditions.Capital(fmt.Sprintf("%s gave up on a message, not kept",
@@ -579,9 +586,58 @@ func watchAdvisories(f *signalFacts) []conditions.Observation {
 		}
 		out = append(out, o)
 	}
+	if f.busFaultsErr != nil {
+		out = append(out, busFaultsUnread(f.busFaultsErr))
+	}
 	return append(out, watchDeadLetters(f)...)
 }
 
+// kindBusFaultsUnread is the bus module's events not read back (novox/hq ADR 0357): every other user's
+// slow consumer is unheard while it lasts, and the controller's own signals are still said.
+const kindBusFaultsUnread = "bus-faults-unread"
+
+func busFaultsUnread(err error) conditions.Observation {
+	return conditions.Observation{Scope: conditions.ScopeBus, ID: "bus-module", Kind: kindBusFaultsUnread,
+		Severity: conditions.Warning, Confirm: true,
+		Summary: "what the bus module said of its users' connections could not be read back, so a slow consumer " +
+			"or a connection the bus closed is not said until it can",
+		Said: firstLine(err.Error())}
+}
+
+// busUserClosed words an advisory about a bus user's connection the server closed (novox/hq ADR 0357): the
+// user as the operator says it and the machine it is on, the dotted user name kept in the evidence only.
+func busUserClosed(o *conditions.Observation, a link.Advisory, host string) {
+	who, machine := busUserWords(a.User)
+	if machine == "" && a.User == "controller" {
+		machine = host
+	}
+	o.Machine = machine
+	times, count := "once", "once"
+	if a.Count > 1 {
+		times = fmt.Sprintf("%d times since %s", a.Count, a.First.UTC().Format("15:04 MST"))
+		count = fmt.Sprintf("%d times", a.Count)
+	}
+	switch a.Kind {
+	case link.AdvisorySlowConsumer:
+		o.Summary = fmt.Sprintf("the bus closed the connection of %s as a slow consumer %s in the last hour, "+
+			"newest at %s: it could not take what it was sent fast enough, and what was on its way to it was lost",
+			who, times, a.Last.UTC().Format("15:04 MST"))
+		o.Headline = clip(conditions.Capital(who)+" was too slow for the bus", 60)
+		o.Explanation = conditions.Capital(fmt.Sprintf("%s could not keep up with what the bus sent it, so the bus "+
+			"closed its connection (%s in the last hour) and what was on its way was lost. It connects again "+
+			"by itself; this clears an hour after the last time.", who, count))
+		o.Resolved = "Resolved: " + who + " kept up for an hour"
+	default:
+		o.Summary = fmt.Sprintf("the bus closed the connection of %s for breaking its rules %s in the last hour, "+
+			"newest at %s: %s", who, times, a.Last.UTC().Format("15:04 MST"), strings.ToLower(a.Reason))
+		o.Headline = clip(conditions.Capital(who)+" was cut off by the bus", 60)
+		o.Explanation = conditions.Capital(fmt.Sprintf("The bus closed the connection of %s because it broke "+
+			"the bus's rules (%s, %s in the last hour), so what it was sending was lost. This clears an hour "+
+			"after the last time.", who, strings.ToLower(a.Reason), count))
+		o.Resolved = "Resolved: " + who + " kept to the bus's rules for an hour"
+	}
+}
+
 // watchDeadLetters says each consumer that DEAD_LETTERS holds a message for (novox/hq issue 330): open
 // while it holds any, so it clears when they are delivered again or dropped, never because the server
 // stopped saying it.

cmd/mesh-controller/watchdogs.go

@@ -92,6 +92,8 @@ type signalFacts struct {
 	// consumers: the condition counts the messages given up on, not the looks (novox/hq issue 440).
 	deadLettersNewest map[string]time.Time
 	advisoriesErr     error
+	// busFaultsErr is why the bus module's events could not be read this tick (novox/hq ADR 0357).
+	busFaultsErr error
 
 	selfCheck selfCheckFacts
 
@@ -358,6 +360,15 @@ func (w *watchdogs) gather(ctx context.Context) *signalFacts {
 	}
 	f.advisories = link.Advisories.Since(now.Add(-advisoryQuiet))
 	f.lostConsumers, f.advisoriesErr = w.lostConsumers(ctx, f.advisories)
+	// And what the bus did to every other user's connection, as the bus module said it (novox/hq ADR 0357):
+	// read back for the same hour, and one advisory with the controller's own where both are its.
+	// A read that fails is its own condition, never S9 blind: the controller's own signals are still said.
+	if w.server != nil && w.js != nil {
+		var faults []link.Advisory
+		if faults, f.busFaultsErr = w.server.BusFaults(ctx, now.Add(-advisoryQuiet)); f.busFaultsErr == nil {
+			f.advisories = link.MergeAdvisories(f.advisories, faults)
+		}
+	}
 	if f.advisoriesErr == nil && w.js != nil {
 		f.deadLetters, f.advisoriesErr = link.HeldDeadLetters(w.js.Context())
 	}

internal/broker/bus_module_emits_test.go

@@ -0,0 +1,34 @@
+package broker
+
+import (
+	"slices"
+	"testing"
+)
+
+// **The bus module's watch speaks through the runtime that carries it** (novox/hq ADR 0357): the node's
+// runtime is granted the events the bus module declares, under the module's name, while the bus's own
+// account keeps the snapshot API and nothing else (ADR 0235).
+func TestTheRuntimeCarryingTheBusModuleMayPublishItsEventsAndTheBussAccountMayNot(t *testing.T) {
+	events := []string{"bus.slow-consumer", "bus.connection-closed"}
+	runtime := Principal{Kind: KindNodeTools, Node: "anchor", Module: RuntimeModule,
+		Carries: []Declared{{Module: "nats", Emits: events}}}
+	p, err := PermissionsFor(runtime)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, e := range events {
+		if !slices.Contains(p.Publish, "mesh.mod.nats.event."+e) {
+			t.Errorf("the runtime carrying the bus module may not publish %s: %v", e, p.Publish)
+		}
+	}
+	account := Principal{Kind: KindModule, Node: "anchor", Module: "nats", Emits: events, SnapshotsTheBus: true}
+	a, err := PermissionsFor(account)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, s := range a.Publish {
+		if s == "mesh.mod.nats.event.bus.slow-consumer" || s == "mesh.mod.nats.event.bus.connection-closed" {
+			t.Errorf("the bus's own account is granted %s", s)
+		}
+	}
+}

internal/catalogue/bus_snapshot_test.go

@@ -21,6 +21,16 @@ func TestTheBussAccountSaysNothingOnTheBus(t *testing.T) {
 		t.Fatalf("a bus module declaring what it emits was not refused, or not for the reason: %v", err)
 	}
 
+	// Its tools bundle says what it emits (novox/hq ADR 0357): the node's runtime carries the bundle and is
+	// granted the module's events, so a bus module that ships one may declare them.
+	bundled := []byte(`{"module":"bus","version":"1","provides":[{"name":"mesh-bus","scope":"mesh"}],
+		"claims":[{"name":"mesh-broker","scope":"mesh"}],"own-secrets":{"broker":"/run/broker"},
+		"emits":["bus.slow-consumer"],"tools":["bus_streams"],
+		"build":{"artifacts":[{"name":"tools","kind":"bundle","language":"go","system":"arch","from":"cmd/t","binary":"t"}]}}`)
+	if _, err := ParseManifest(bundled); err != nil {
+		t.Fatalf("a bus module whose tools bundle declares what it emits was refused: %v", err)
+	}
+
 	quiet := []byte(`{"module":"bus","version":"1","provides":[{"name":"mesh-bus","scope":"mesh"}],
 		"claims":[{"name":"mesh-broker","scope":"mesh"}],"own-secrets":{"broker":"/run/broker"},
 		"tools":["bus_streams"]}`)

internal/catalogue/manifest.go

@@ -1630,9 +1630,14 @@ func ParseManifest(raw []byte) (Manifest, error) {
 	// **The bus's own account copies the bus and says nothing on it** (novox/hq ADR 0235). The module
 	// holding mesh-broker is granted the snapshot API and nothing else, so anything it declared to say
 	// or hear on the bus would be granted nothing — refused here rather than silently dropped.
+	//
+	// **But its tools bundle may say what it declares** (novox/hq ADR 0357): the node's runtime carries
+	// the bundle and is granted the events the module emits, under the module's name, as for any module it
+	// carries — the bus's account is not who publishes them. So emits are refused only for a bus module
+	// with no bundle to say them.
 	if _, account := m.OwnSecrets["broker"]; account && m.ClaimsSeat(BrokerSeat) {
 		var said []string
-		if len(m.EmitsAll()) > 0 {
+		if len(m.EmitsAll()) > 0 && !m.shipsABundle() {
 			said = append(said, "emits")
 		}
 		for _, f := range []struct {
@@ -2897,3 +2902,16 @@ func whileStoppedProblems(m Manifest, r map[string]any, scheduled bool) []string
 	}
 	return problems
 }
+
+// shipsABundle says the module builds a tools bundle, which the node's runtime carries and speaks for.
+func (m Manifest) shipsABundle() bool {
+	if m.Build == nil {
+		return false
+	}
+	for _, a := range m.Build.Artifacts {
+		if a.Kind == ArtifactBundle {
+			return true
+		}
+	}
+	return false
+}

internal/link/advisories.go

@@ -24,9 +24,10 @@ import (
 // which consumer of whose, which subject — for the watchdog to say as a condition, as the refused
 // reply of issue 265 is said today.
 //
-// What the bus says about **other** principals' connections — a module's slow consumer, a module
-// refused a subject — the server publishes only to a system account, which the mesh's bus does not
-// have; that half is not heard yet (to-be 45 S9, recorded as deferred).
+// What the bus does to **other** principals' connections the server does not publish in the mesh's
+// account. A module's slow consumer it closes and records, and the bus module says that record on the
+// bus (busfaults.go, novox/hq ADR 0357). A module refused a subject is told and written in the server's
+// log, and nowhere else: not heard yet (novox/hq issue 702).
 
 // The advisory kinds, as conditions name them.
 const (
@@ -48,6 +49,11 @@ type Advisory struct {
 	Stream, Consumer string
 	// Token is the condition key's last part, when it is not Kind.
 	Token string
+	// User is the bus user whose connection it is about, when it is about one (novox/hq ADR 0357): the
+	// controller's own, or another user's the bus module said.
+	User string
+	// Reason is the server's reason for closing that connection, as it gave it.
+	Reason string
 	// Said is the newest saying, in the mesh's words.
 	Said        string
 	First, Last time.Time

internal/link/busfaults.go

@@ -0,0 +1,235 @@
+package link
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"sort"
+	"time"
+
+	"github.com/nats-io/nats.go"
+
+	"github.com/novox/mesh-controller/internal/broker"
+)
+
+// What the bus did to other users' connections, as the bus module says it (novox/hq ADR 0357, issue 270).
+//
+// **The server says another user's slow consumer only in its record of closed connections**, which the
+// bus module reads on its own machine and says as events: `bus.slow-consumer` for a connection the server
+// closed as a slow consumer, `bus.connection-closed` for one it closed for breaking the bus's rules. They
+// are read back from EVENTS on every watchdog tick for the hour S9 says an advisory, rather than taken by
+// the controller's durable consumer: that one takes a message at a time while a merge builds for
+// minutes, and read back, a controller that restarts says the last hour again without keeping anything.
+//
+// **Said at least once, counted once.** The module says the last hour again whenever it starts; a closure
+// is its user, its connection id and when it opened, and is counted once however often it is said.
+
+// BusModule is the module whose events say what the bus did to its users' connections.
+const BusModule = "nats"
+
+// The bus module's events, by their local names (its manifest's `emits`).
+const (
+	BusSlowConsumerEvent     = "bus.slow-consumer"
+	BusConnectionClosedEvent = "bus.connection-closed"
+)
+
+// AdvisoryConnectionClosed is a connection the bus closed for breaking its rules, as conditions name it.
+const AdvisoryConnectionClosed = "connection-closed"
+
+// BusClosure is one closed connection as the bus module says it: the user, never a credential.
+type BusClosure 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"`
+	Pending int64     `json:"pending_bytes,omitempty"`
+}
+
+func (c BusClosure) identity() string {
+	return fmt.Sprintf("%s/%d/%s", c.User, c.CID, c.Opened.UTC().Format(time.RFC3339Nano))
+}
+
+// BusClosures are the advisories the bus module's closures make, closed at or after since: one per user
+// and kind, each closure counted once, the newest said.
+func BusClosures(closures []BusClosure, kinds []string, since time.Time) []Advisory {
+	type entry struct {
+		a      *Advisory
+		seen   map[string]bool
+		newest BusClosure
+	}
+	by := map[string]*entry{}
+	var order []string
+	for i, c := range closures {
+		if c.User == "" || c.Closed.Before(since) {
+			continue
+		}
+		kind := kinds[i]
+		key := kind + "/" + c.User
+		e := by[key]
+		if e == nil {
+			e = &entry{a: &Advisory{Kind: kind, ID: c.User, User: c.User}, seen: map[string]bool{}}
+			by[key] = e
+			order = append(order, key)
+		}
+		if e.seen[c.identity()] {
+			continue
+		}
+		e.seen[c.identity()] = true
+		e.a.Count++
+		if e.a.First.IsZero() || c.Closed.Before(e.a.First) {
+			e.a.First = c.Closed
+		}
+		if !c.Closed.Before(e.a.Last) {
+			e.a.Last, e.newest = c.Closed, c
+		}
+	}
+	sort.Strings(order)
+	out := make([]Advisory, 0, len(order))
+	for _, key := range order {
+		e := by[key]
+		c := e.newest
+		name := ""
+		if c.Name != "" && c.Name != c.User {
+			name = fmt.Sprintf(" (connection %q)", c.Name)
+		}
+		pending := ""
+		if c.Pending > 0 {
+			pending = fmt.Sprintf(", %d bytes still to be written to it", c.Pending)
+		}
+		switch e.a.Kind {
+		case AdvisorySlowConsumer:
+			e.a.Said = fmt.Sprintf("the bus closed %s's connection %d%s as a slow consumer: %s%s", c.User, c.CID,
+				name, c.Reason, pending)
+		default:
+			e.a.Said = fmt.Sprintf("the bus closed %s's connection %d%s: %s", c.User, c.CID, name, c.Reason)
+		}
+		e.a.Reason = c.Reason
+		out = append(out, *e.a)
+	}
+	return out
+}
+
+// busEventKind is the advisory kind of one of the bus module's events, by its subject.
+func busEventKind(subject string) (string, bool) {
+	switch subject {
+	case EventSubject(BusModule, BusSlowConsumerEvent):
+		return AdvisorySlowConsumer, true
+	case EventSubject(BusModule, BusConnectionClosedEvent):
+		return AdvisoryConnectionClosed, true
+	}
+	return "", false
+}
+
+// BusFaults reads back what the bus module said of its users' connections since a moment, from the events
+// stream, on a consumer of its own that acknowledges nothing (as PowerStates reads), and says it as
+// advisories. Nothing to read is no advisory, and costs two direct reads: the newest of each event.
+func (s *Server) BusFaults(ctx context.Context, since time.Time) ([]Advisory, error) {
+	if s.js == nil {
+		return nil, errors.New("this control plane is not on the bus, so it cannot read what the bus module said")
+	}
+	js := s.js.Context()
+	recent := false
+	for _, event := range []string{BusSlowConsumerEvent, BusConnectionClosedEvent} {
+		last, err := js.GetLastMsg(broker.EventsStream, EventSubject(BusModule, event), nats.Context(ctx))
+		if errors.Is(err, nats.ErrMsgNotFound) {
+			continue
+		}
+		if err != nil {
+			return nil, fmt.Errorf("reading what the bus module said of its users' connections: %w", err)
+		}
+		if !last.Time.Before(since) {
+			recent = true
+		}
+	}
+	if !recent {
+		return nil, nil
+	}
+	sub, err := js.SubscribeSync(EventSubject(BusModule, "bus.>"), nats.OrderedConsumer(), nats.StartTime(since))
+	if err != nil {
+		return nil, fmt.Errorf("reading what the bus module said of its users' connections: %w", err)
+	}
+	defer func() { _ = sub.Unsubscribe() }()
+	var closures []BusClosure
+	var kinds []string
+	read, cut := 0, false
+	for {
+		if read >= MaxBusEventsRead {
+			cut = true
+			break
+		}
+		read++
+		wait, cancel := context.WithTimeout(ctx, readQuiet)
+		msg, err := sub.NextMsgWithContext(wait)
+		cancel()
+		if err != nil {
+			if ctx.Err() != nil {
+				return nil, ctx.Err()
+			}
+			break
+		}
+		if kind, ok := busEventKind(msg.Subject); ok {
+			var c BusClosure
+			if json.Unmarshal(msg.Data, &c) == nil && c.User != "" && !c.Closed.IsZero() {
+				closures, kinds = append(closures, c), append(kinds, kind)
+			}
+		}
+		if meta, err := msg.Metadata(); err != nil || meta.NumPending == 0 {
+			break
+		}
+	}
+	out := BusClosures(closures, kinds, since)
+	if cut {
+		// Said, never silent: a busy hour is read up to the bound, and each advisory says it is a floor.
+		for i := range out {
+			out[i].Said += fmt.Sprintf(" (at least: the read stopped at %d of the bus module's events this hour)",
+				MaxBusEventsRead)
+		}
+	}
+	return out, nil
+}
+
+// MaxBusEventsRead bounds one tick's read of the bus module's events: a watchdog tick must not spend itself
+// reading a storm of closures back. A read cut there says so in every advisory it makes.
+var MaxBusEventsRead = 2000
+
+// MergeAdvisories is two lists of advisories as one: the same kind about the same thing is one advisory,
+// counted once (the larger count of the two), said by the newer. The controller's own slow consumer as its client library sees it and
+// as the server closed it are the same user's (`controller`).
+func MergeAdvisories(a, b []Advisory) []Advisory {
+	out := append([]Advisory(nil), a...)
+	at := map[string]int{}
+	for i, x := range out {
+		at[x.Kind+"/"+x.ID+"/"+x.Token] = i
+	}
+	for _, y := range b {
+		key := y.Kind + "/" + y.ID + "/" + y.Token
+		i, had := at[key]
+		if !had {
+			at[key] = len(out)
+			out = append(out, y)
+			continue
+		}
+		x := &out[i]
+		// The same fault seen twice, not two faults: the controller's own slow consumer is told to its client
+		// library and closed by the server, and is counted once — by whichever saw more of it.
+		if y.Count > x.Count {
+			x.Count = y.Count
+		}
+		if y.First.Before(x.First) {
+			x.First = y.First
+		}
+		if y.Last.After(x.Last) {
+			x.Last, x.Said = y.Last, y.Said
+			if y.Reason != "" {
+				x.Reason = y.Reason
+			}
+		}
+		if x.User == "" {
+			x.User = y.User
+		}
+	}
+	return out
+}

internal/link/busfaults_test.go

@@ -0,0 +1,121 @@
+package link
+
+import (
+	"encoding/json"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/novox/mesh-controller/internal/broker"
+	"github.com/novox/mesh-controller/internal/testbus"
+)
+
+// What the bus module says of its users' connections, read back as advisories (novox/hq ADR 0357, issue
+// 270): one per user and kind, each closure counted once however often it is said.
+//
+//	go test ./internal/link/ -run TestNatsBusFaults   (each test on a bus of its own: internal/testbus)
+
+func aBusWithEvents(t *testing.T) *Server {
+	t.Helper()
+	js, err := broker.Dial(testbus.URL(t))
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(js.Close)
+	if err := broker.AssertMeshStreams(js); err != nil {
+		t.Fatal(err)
+	}
+	return ConnectNats(js, nil, nil)
+}
+
+func sayClosure(t *testing.T, s *Server, event string, c BusClosure) {
+	t.Helper()
+	body, _ := json.Marshal(c)
+	if _, err := s.js.Context().Publish(EventSubject(BusModule, event), body); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestNatsBusFaultsAreOnePerUserAndKindEachClosureCountedOnce(t *testing.T) {
+	s := aBusWithEvents(t)
+	now := time.Now().UTC()
+	slow := BusClosure{User: "laptop.claude-code", Name: "laptop.claude-code", CID: 10, Reason: "Slow Consumer (Write Deadline)",
+		Opened: now.Add(-20 * time.Minute), Closed: now.Add(-2 * time.Minute), Pending: 334000}
+	sayClosure(t, s, BusSlowConsumerEvent, slow)
+	sayClosure(t, s, BusSlowConsumerEvent, slow) // said again: the module started again
+	again := slow
+	again.CID, again.Opened, again.Closed = 31, now.Add(-90*time.Second), now.Add(-time.Minute)
+	sayClosure(t, s, BusSlowConsumerEvent, again)
+	sayClosure(t, s, BusConnectionClosedEvent, BusClosure{User: "anchor.desk", CID: 7,
+		Reason: "Maximum Message Payload Exceeded", Opened: now.Add(-time.Hour / 2), Closed: now.Add(-30 * time.Second)})
+	// Older than the hour: S9 no longer says it.
+	sayClosure(t, s, BusSlowConsumerEvent, BusClosure{User: "anchor.media", CID: 3, Reason: "Slow Consumer (Pending Bytes)",
+		Opened: now.Add(-3 * time.Hour), Closed: now.Add(-2 * time.Hour)})
+
+	got, err := s.BusFaults(t.Context(), now.Add(-time.Hour))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(got) != 2 {
+		t.Fatalf("read %d advisories, want one for each user within the hour: %+v", len(got), got)
+	}
+	for _, a := range got {
+		switch a.User {
+		case "laptop.claude-code":
+			if a.Kind != AdvisorySlowConsumer || a.ID != "laptop.claude-code" || a.Count != 2 ||
+				!a.Last.Equal(again.Closed) || !a.First.Equal(slow.Closed) || !strings.Contains(a.Said, "connection 31") {
+				t.Errorf("the slow consumer read as %+v: two closures, the one said twice counted once", a)
+			}
+		case "anchor.desk":
+			if a.Kind != AdvisoryConnectionClosed || a.Count != 1 || a.Reason != "Maximum Message Payload Exceeded" {
+				t.Errorf("the closed connection read as %+v", a)
+			}
+		default:
+			t.Errorf("an advisory for %s", a.User)
+		}
+	}
+}
+
+func TestNatsBusFaultsReadNothingWhenTheBusModuleSaidNothing(t *testing.T) {
+	s := aBusWithEvents(t)
+	started := time.Now()
+	got, err := s.BusFaults(t.Context(), time.Now().Add(-time.Hour))
+	if err != nil || len(got) != 0 {
+		t.Fatalf("read %v, %v from a stream the bus module never wrote", got, err)
+	}
+	if took := time.Since(started); took > time.Second {
+		t.Fatalf("reading nothing took %s: a tick must not wait on an empty stream", took)
+	}
+}
+
+func TestTheControllersOwnSlowConsumerAndItsClosedConnectionAreOneAdvisory(t *testing.T) {
+	at := time.Date(2026, 10, 14, 12, 0, 0, 0, time.UTC)
+	own := []Advisory{{Kind: AdvisorySlowConsumer, ID: "controller", Said: "the controller fell behind", First: at, Last: at, Count: 1}}
+	closed := BusClosures([]BusClosure{{User: "controller", CID: 5, Reason: "Slow Consumer (Pending Bytes)",
+		Opened: at.Add(-time.Hour), Closed: at.Add(time.Minute)}}, []string{AdvisorySlowConsumer}, at.Add(-time.Hour))
+	got := MergeAdvisories(own, closed)
+	if len(got) != 1 || got[0].Count != 1 || !got[0].Last.Equal(at.Add(time.Minute)) || got[0].User != "controller" {
+		t.Fatalf("merged as %+v: one fall behind, told to the client and closed by the server, is counted once", got)
+	}
+}
+
+// **A busy hour is read up to a bound, and says it was cut** (review of #326): a storm of closures must not
+// spend a watchdog tick, and a count cut short is a floor, said as one.
+func TestNatsBusFaultsReadAtMostTheBoundAndSaySo(t *testing.T) {
+	s := aBusWithEvents(t)
+	was := MaxBusEventsRead
+	MaxBusEventsRead = 2
+	t.Cleanup(func() { MaxBusEventsRead = was })
+	now := time.Now().UTC()
+	for i := range 3 {
+		sayClosure(t, s, BusSlowConsumerEvent, BusClosure{User: "laptop.claude-code", CID: uint64(10 + i),
+			Reason: "Slow Consumer (Pending Bytes)", Opened: now.Add(-time.Hour / 2), Closed: now.Add(-time.Duration(3-i) * time.Minute)})
+	}
+	got, err := s.BusFaults(t.Context(), now.Add(-time.Hour))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(got) != 1 || got[0].Count != 2 || !strings.Contains(got[0].Said, "at least: the read stopped at 2") {
+		t.Fatalf("a read past its bound read as %+v", got)
+	}
+}