Merge branch 'feat/mail-hardening' into mh-785

cmd/mesh-controller/busobjects.go

@@ -49,10 +49,12 @@ func assertBusObjects(ctx context.Context, inv *inventory.Inventory, r broker.Ra
 	// composed from, so a module the mesh grants a consumer's subjects has that consumer waiting.
 	// Done on every raise, not only when a credential is issued: every module moved onto this bus
 	// by the rollout was issued on the old one, and came up with nothing to bind to (2026-09-28).
-	consumers, err := moduleConsumers(ctx, inv)
+	consumers, records, err := moduleConsumersAndRecords(ctx, inv)
 	if err != nil {
 		return nil, err
 	}
+	// And a grant of a restricted stream that composes nothing, said on every raise and send (novox/hq ADR 0376 §2).
+	sayWithheld(records, "  ")
 	// And the work queues of seats that name their caller or their kind, with each holder's worker
 	// (novox/hq ADR 0259 §3): an ask queues until the router takes it, a channel's work until that kind
 	// takes it.
@@ -165,15 +167,21 @@ func unassertedObservation(err error) conditions.Observation {
 
 // moduleConsumers is every module's durable consumer, from the records the user list is composed from.
 func moduleConsumers(ctx context.Context, inv *inventory.Inventory) ([]broker.ModuleConsumer, error) {
+	consumers, _, err := moduleConsumersAndRecords(ctx, inv)
+	return consumers, err
+}
+
+// moduleConsumersAndRecords is moduleConsumers with the records they were composed from.
+func moduleConsumersAndRecords(ctx context.Context, inv *inventory.Inventory) ([]broker.ModuleConsumer, broker.Records, error) {
 	records, err := inv.BusRecords(ctx)
 	if err != nil {
-		return nil, err
+		return nil, broker.Records{}, err
 	}
 	users, err := broker.Users(records)
 	if err != nil {
-		return nil, err
+		return nil, broker.Records{}, err
 	}
-	return broker.ConsumersOf(users), nil
+	return broker.ConsumersOf(users), records, nil
 }
 
 // seatTrafficObjects is the work queues and workers of seats that name their caller or their kind, from

cmd/mesh-controller/deadletters_aging.go

@@ -0,0 +1,117 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"sort"
+	"strings"
+	"time"
+
+	"github.com/novox/mesh-controller/internal/conditions"
+	"github.com/novox/mesh-controller/internal/link"
+)
+
+// A dead letter from a restricted stream ages out with its stream (novox/hq ADR 0376 §3, §6).
+//
+// DEAD_LETTERS lets such a message go when its stream's maximum age runs out, so a consumer's `max-deliveries`
+// condition can clear with nobody having delivered it again or dropped it. That is said: the watchdogs remember,
+// from one look to the next, when each restricted dead letter ages out, count those gone once their time passed,
+// say each in the controller's log as it goes, and clear the condition with words saying how many aged out
+// undelivered and from which stream.
+
+// agingTally is what the watchdogs remember of restricted dead letters between looks.
+type agingTally struct {
+	// seen is each consumer's restricted dead letters at the last look, by id, with when each ages out.
+	seen map[string]map[uint64]time.Time
+	// agedOut is how many of each consumer's aged out undelivered while its condition was open.
+	agedOut map[string]int
+}
+
+// look compares this look's restricted dead letters with the last: newly is how many of each consumer's aged out
+// since, and cleared how many aged out in all for each consumer DEAD_LETTERS no longer holds anything of, when any
+// did. held is every consumer's count this look; aging its restricted dead letters.
+func (t *agingTally) look(now time.Time, held map[string]int, aging map[string][]link.Aging) (newly, cleared map[string]int) {
+	if t.seen == nil {
+		t.seen, t.agedOut = map[string]map[uint64]time.Time{}, map[string]int{}
+	}
+	newly, cleared = map[string]int{}, map[string]int{}
+	for key, before := range t.seen {
+		still := map[uint64]bool{}
+		for _, a := range aging[key] {
+			still[a.ID] = true
+		}
+		for id, agesOut := range before {
+			if !still[id] && !agesOut.After(now) {
+				newly[key]++
+			}
+		}
+		if newly[key] == 0 {
+			delete(newly, key)
+		}
+		t.agedOut[key] += newly[key]
+	}
+	for key, n := range t.agedOut {
+		if held[key] > 0 {
+			continue
+		}
+		if n > 0 {
+			cleared[key] = n
+		}
+		delete(t.agedOut, key)
+	}
+	t.seen = map[string]map[uint64]time.Time{}
+	for key, letters := range aging {
+		t.seen[key] = map[uint64]time.Time{}
+		for _, a := range letters {
+			t.seen[key][a.ID] = a.AgesOut
+		}
+	}
+	return newly, cleared
+}
+
+// agedOutResolved is the line a consumer's condition clears with when messages it gave up on aged out undelivered.
+func agedOutResolved(key string, n int) string {
+	stream, _, _ := strings.Cut(key, ".")
+	messages := "a message"
+	if n > 1 {
+		messages = fmt.Sprintf("%d messages", n)
+	}
+	return fmt.Sprintf("Resolved: %s of the stream %s aged out undelivered", messages, stream)
+}
+
+// sayAgedOut says, in the log, what aged out since the last look, and clears the condition of every consumer whose
+// dead letters are all gone with the words saying how many aged out. Before the rows are reconciled, so the
+// condition is cleared with these words and not its own.
+func (w *watchdogs) sayAgedOut(ctx context.Context, f *signalFacts) []string {
+	if f.advisoriesErr != nil || f.deadLettersAging == nil {
+		return nil
+	}
+	newly, cleared := w.aging.look(f.now, f.deadLetters, f.deadLettersAging)
+	var problems []string
+	for _, key := range countedKeys(newly) {
+		stream, consumer, _ := strings.Cut(key, ".")
+		fmt.Printf("%d message(s) %s gave up on aged out of %s undelivered: %s keeps them only for its maximum age "+
+			"(novox/hq ADR 0376)\n", newly[key], link.ConsumerInWords(stream, consumer), "DEAD_LETTERS", stream)
+	}
+	for _, key := range countedKeys(cleared) {
+		stream, consumer, _ := strings.Cut(key, ".")
+		fmt.Printf("%s no longer has a message kept: %d aged out of %s undelivered, so its condition clears "+
+			"(novox/hq ADR 0376)\n", link.ConsumerInWords(stream, consumer), cleared[key], stream)
+		if _, err := w.keeper.ClearSaying(ctx, conditions.Key(conditions.ScopeBus, key, link.AdvisoryMaxDeliveries),
+			fmt.Sprintf("its %d dead letter(s) of %s aged out undelivered", cleared[key], stream),
+			agedOutResolved(key, cleared[key])); err != nil {
+			problems = append(problems, err.Error())
+		}
+	}
+	return problems
+}
+
+// countedKeys is a count's keys in order.
+func countedKeys(m map[string]int) []string {
+	out := make([]string, 0, len(m))
+	for k := range m {
+		out = append(out, k)
+	}
+	sort.Strings(out)
+	return out
+}

cmd/mesh-controller/deadletters_aging_test.go

@@ -0,0 +1,98 @@
+package main
+
+import (
+	"context"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/novox/mesh-controller/internal/conditions"
+	"github.com/novox/mesh-controller/internal/link"
+)
+
+// **A dead letter that ages out clears its condition and says so** (novox/hq ADR 0376 §6), on a fake clock: while
+// DEAD_LETTERS holds a restricted stream's messages for a consumer its max-deliveries condition is open; once they
+// have aged out it clears, with words saying how many aged out undelivered and from which stream. One dropped before
+// its age is not counted as aged out, and a consumer of EVENTS clears as before.
+func TestADeadLetterThatAgesOutClearsItsConditionAndSaysSo(t *testing.T) {
+	now := time.Date(2026, 10, 15, 12, 0, 0, 0, time.UTC)
+	store := conditions.NewInMemory()
+	told := &conditions.Told{}
+	k := conditions.NewKeeper(t.Context(), conditions.Options{Store: store, History: store, Teller: told,
+		Now: func() time.Time { return now }})
+	w := &watchdogs{keeper: k, started: now.Add(-time.Hour)}
+
+	look := func(at time.Time, held map[string]int, aging map[string][]link.Aging) {
+		f := calm(at)
+		f.deadLetters, f.deadLettersAging = held, aging
+		w.see(t.Context(), f)
+	}
+	mail := "MAIL.ace_mailer"
+	look(now, map[string]int{mail: 3, "EVENTS.media_sonarr": 1}, map[string][]link.Aging{mail: {
+		{ID: 1, AgesOut: now.Add(time.Hour)}, {ID: 2, AgesOut: now.Add(2 * time.Hour)}, {ID: 3, AgesOut: now.Add(3 * time.Hour)}}})
+	if open, _ := k.Open(t.Context()); len(open) != 2 {
+		t.Fatalf("open: %+v", open)
+	}
+	// One dropped by a person before its age: not aged out.
+	look(now.Add(30*time.Minute), map[string]int{mail: 2, "EVENTS.media_sonarr": 1}, map[string][]link.Aging{mail: {
+		{ID: 2, AgesOut: now.Add(2 * time.Hour)}, {ID: 3, AgesOut: now.Add(3 * time.Hour)}}})
+	look(now.Add(150*time.Minute), map[string]int{mail: 1}, map[string][]link.Aging{mail: {{ID: 3, AgesOut: now.Add(3 * time.Hour)}}})
+	open, _ := k.Open(t.Context())
+	if len(open) != 1 || open[0].Key != conditions.Key(conditions.ScopeBus, mail, link.AdvisoryMaxDeliveries) {
+		t.Fatalf("after one aged out: %+v", open)
+	}
+	look(now.Add(4*time.Hour), map[string]int{}, map[string][]link.Aging{})
+	if open, _ := k.Open(t.Context()); len(open) != 0 {
+		t.Fatalf("after all aged out: %+v", open)
+	}
+	k.Close(context.Background()) // everything said is said once it closes
+	var mailCleared, sonarrCleared string
+	for _, e := range told.Said() {
+		if e.Change != conditions.ChangeCleared {
+			continue
+		}
+		switch e.Condition.Key {
+		case conditions.Key(conditions.ScopeBus, mail, link.AdvisoryMaxDeliveries):
+			mailCleared = e.Condition.Resolved
+		case conditions.Key(conditions.ScopeBus, "EVENTS.media_sonarr", link.AdvisoryMaxDeliveries):
+			sonarrCleared = e.Condition.Resolved
+		}
+	}
+	if mailCleared != "Resolved: 2 messages of the stream MAIL aged out undelivered" {
+		t.Fatalf("MAIL's condition cleared with %q", mailCleared)
+	}
+	if sonarrCleared != "Resolved: the messages it gave up on were delivered again or dropped" {
+		t.Fatalf("an EVENTS consumer's condition cleared with %q", sonarrCleared)
+	}
+	if why, ok := conditions.Plain(mailCleared); !ok {
+		t.Fatalf("not plain: %s", why)
+	}
+}
+
+// **A message past its restricted stream's age is not kept, only counted** (novox/hq ADR 0376, as the feature's lead
+// decided): its advisory is said as a condition whose words give how many aged out undelivered and from which stream.
+func TestMessagesNotKeptPastTheirAgeAreCounted(t *testing.T) {
+	now := time.Date(2026, 10, 15, 12, 0, 0, 0, time.UTC)
+	f := calm(now)
+	f.advisories = []link.Advisory{{Kind: link.AdvisoryMaxDeliveries, ID: "MAIL.ace_mailer", Stream: "MAIL",
+		Consumer: "ace_mailer", Token: link.AdvisoryAgedOut, Said: "the consumer gave up after the stream's age",
+		First: now.Add(-10 * time.Minute), Last: now.Add(-time.Minute), Count: 2}}
+	var found []conditions.Observation
+	for _, o := range watchAdvisories(f) {
+		if o.Token == link.AdvisoryAgedOut {
+			found = append(found, o)
+		}
+	}
+	if len(found) != 1 {
+		t.Fatalf("said %+v", found)
+	}
+	o := found[0]
+	if o.Headline != "2 messages of the stream MAIL aged out undelivered" || o.Times != 2 ||
+		!strings.Contains(o.Summary, "2 messages of the stream MAIL aged out undelivered") {
+		t.Fatalf("%+v", o)
+	}
+	if why, ok := conditions.PlainWords(conditions.Words{Headline: o.Headline, Explanation: o.Explanation,
+		Resolved: o.Resolved, Needs: o.Needs}); !ok {
+		t.Fatalf("not plain: %s", why)
+	}
+}

cmd/mesh-controller/doctor.go

@@ -146,6 +146,11 @@ var probeRegistry = []probe{
 	{ID: probeFillID, Asserts: "every module's bucket and log holds less than three quarters of its cap; one " +
 		"said filling is cleared once it holds less than seven tenths", From: "ADR 0297, issue 501",
 		Kind: kindBucketOrLogFilling, Raises: []string{kindBucketOrLogUnread}, Phase: 1, run: probeFill},
+	// A grant of a restricted stream that composes nothing (novox/hq ADR 0376 §2): the module is carried by the
+	// machine's tool runner, which never hears a restricted stream, so it hears nothing it was granted.
+	{ID: probeWithheldID, Asserts: "no grant of a restricted stream names a module that composes nothing for it: every " +
+		"module granted one runs as an account of its own", From: "ADR 0376 §2", Kind: kindGrantComposesNothing,
+		Phase: 1, run: probeWithheldHearing},
 	{ID: "DW", Asserts: "the watchdogs of the signals table ran within three of their intervals",
 		From: "ADR 0227 rule 6: the watchers are watched", Kind: "watchdogs-silent", Phase: 1, run: probeWatchdogs},
 	// The core's health definitions (novox/hq to-be 45 §8, ADR 0236): what a core component's new build is

cmd/mesh-controller/heard_withheld.go

@@ -0,0 +1,66 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"github.com/novox/mesh-controller/internal/broker"
+	"github.com/novox/mesh-controller/internal/conditions"
+)
+
+// A grant of a restricted stream that composes nothing (novox/hq ADR 0376 §2).
+//
+// A restricted stream is heard only by a module that runs as an account of its own (`runs-as`, ADR 0259): the
+// tool runner's bus account is the one every agent session on its machine reaches the bus as, so the controller
+// composes no consumer, grant or bind word for a module it carries (broker.ComposedHears). Where a grant still
+// names such a module, the module hears nothing it was granted, and that is said — in `doctor` and in the
+// controller's log on every raise and send — naming the module, the stream and the remedy. Never silent.
+
+const (
+	// probeWithheldID is the self-check's probe of it, and kindGrantComposesNothing the condition it raises.
+	probeWithheldID          = "D-heard"
+	kindGrantComposesNothing = "grant-composes-nothing"
+)
+
+// probeWithheldHearing says every granted hearing of a restricted stream composed for nothing.
+func probeWithheldHearing(ctx context.Context, d *doctor) ([]conditions.Observation, error) {
+	records, err := d.open.inventory.BusRecords(ctx)
+	if err != nil {
+		return nil, err
+	}
+	return withheldObservations(broker.WithheldOf(records)), nil
+}
+
+// withheldObservations is each withheld hearing as a condition of its module on its machine.
+func withheldObservations(withheld []broker.Withheld) []conditions.Observation {
+	var out []conditions.Observation
+	for _, w := range withheld {
+		out = append(out, conditions.Observation{Scope: conditions.ScopeModule,
+			ID:   w.Module + "." + strings.ToLower(w.Stream) + "." + w.Node,
+			Kind: kindGrantComposesNothing, Machine: w.Node, Severity: conditions.Warning,
+			Summary: w.Said(), Said: fmt.Sprintf("%s on %s granted %s; composed nothing", w.Module, w.Node, w.Stream)})
+	}
+	return sortedFound(out)
+}
+
+// sayWithheld puts every withheld hearing in the controller's log, once per raise or send that composes it.
+func sayWithheld(records broker.Records, indent string) {
+	for _, w := range broker.WithheldOf(records) {
+		fmt.Printf("%sA GRANT COMPOSES NOTHING: %s\n", indent, w.Said())
+	}
+}
+
+// withheldWords are the plain words of a withheld hearing: the module, the machine and what to do.
+func withheldWords(o conditions.Observation) words {
+	module, m := "a module", machineOr(o, "a machine")
+	if o.Scope == conditions.ScopeModule && idPart(o, 0) != "" {
+		module = idPart(o, 0)
+	}
+	return words{Headline: clip(conditions.Capital(fmt.Sprintf("%s on %s hears nothing it was granted", module, m)), 60),
+		Needs: fmt.Sprintf("give %s an account of its own in its manifest; the details say which stream.", module),
+		Explanation: fmt.Sprintf("You granted %s part of a stream that only modules running under an account of "+
+			"their own may hear. It runs inside the shared tool runner on %s, which every assistant session there "+
+			"also uses, so the mesh gives it none of that stream.", module, m),
+		Resolved: fmt.Sprintf("%s on %s hears what it was granted, or is no longer granted it", module, m)}
+}

cmd/mesh-controller/heard_withheld_test.go

@@ -0,0 +1,43 @@
+package main
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/novox/mesh-controller/internal/broker"
+	"github.com/novox/mesh-controller/internal/conditions"
+)
+
+// A grant naming a module the tool runner carries composes nothing, and doctor says so in plain words, naming the
+// module, the stream and the remedy (novox/hq ADR 0376 §2).
+func TestAGrantThatComposesNothingIsSaid(t *testing.T) {
+	records := broker.Records{Nodes: []string{"ace"}, Assigned: map[string][]broker.Declared{"ace": {
+		{Module: broker.RuntimeModule},
+		{Module: "reader", Hears: []broker.Heard{{Stream: "MAIL", Emitter: "mesh-mail", Restricted: true,
+			Filters: []string{"mesh.stream.mesh-mail.event.mail.mesh.>"}}}},
+		{Module: "mailer", RunsAs: "mailer", Hears: []broker.Heard{{Stream: "MAIL", Emitter: "mesh-mail", Restricted: true,
+			Filters: []string{"mesh.stream.mesh-mail.event.mail.mesh.>"}}}},
+	}}}
+	found := withheldObservations(broker.WithheldOf(records))
+	if len(found) != 1 {
+		t.Fatalf("said %+v", found)
+	}
+	o := found[0]
+	if o.Kind != kindGrantComposesNothing || o.Machine != "ace" || !strings.Contains(o.Summary, "reader on ace") ||
+		!strings.Contains(o.Summary, "MAIL") || !strings.Contains(o.Summary, "runs-as") {
+		t.Fatalf("%+v", o)
+	}
+	w := plainWordings[kindGrantComposesNothing](o)
+	if why, ok := conditions.PlainWords(w, "ace"); !ok {
+		t.Fatalf("not plain: %s (%+v)", why, w)
+	}
+	registered := false
+	for _, p := range probeRegistry {
+		if p.ID == probeWithheldID && p.Kind == kindGrantComposesNothing && p.run != nil {
+			registered = true
+		}
+	}
+	if !registered {
+		t.Fatal("doctor does not run the probe of a grant that composes nothing")
+	}
+}

cmd/mesh-controller/plain_words.go

@@ -117,6 +117,8 @@ var plainWordings = map[string]func(conditions.Observation) words{
 				"reaches it. It keeps running what it has.", m),
 			Resolved: m + " can get new instructions again"}
 	}),
+	// A grant of a restricted stream that composes nothing (novox/hq ADR 0376 §2).
+	kindGrantComposesNothing: worded(withheldWords),
 	kindAgentCanBecomeRoot: worded(func(o conditions.Observation) words {
 		m := machineOr(o, "a machine")
 		return words{Headline: "Sessions on " + m + " could become root",

cmd/mesh-controller/probes.go

@@ -929,6 +929,12 @@ func streamDiffers(want broker.Stream, have nats.StreamConfig) string {
 		differs = append(differs, fmt.Sprintf("keeps one of a message id for %s, defined %s", have.Duplicates,
 			time.Duration(want.DuplicatesSeconds)*time.Second))
 	}
+	if want.AllowMsgTTL && !have.AllowMsgTTL {
+		differs = append(differs, "takes no time to live of a message's own, defined to (novox/hq ADR 0376)")
+	}
+	if want.Restricted && have.Metadata[broker.RestrictedMetadata] != "true" {
+		differs = append(differs, "does not say it is restricted, so a message kept from it would not age out with it")
+	}
 	if want.DiscardNew && have.Discard != nats.DiscardNew {
 		differs = append(differs, "drops what it holds when full, defined to refuse what comes next")
 	}

cmd/mesh-controller/signals.go

@@ -584,6 +584,18 @@ func watchAdvisories(f *signalFacts) []conditions.Observation {
 				"for you yet. It tries again every minute."
 			o.Resolved = "Resolved: the message is kept"
 		}
+		if a.Kind == link.AdvisoryMaxDeliveries && a.Token == link.AdvisoryAgedOut {
+			// Given up on after its restricted stream's age, so not kept at all, only counted (novox/hq ADR 0376).
+			counted := link.AgedOutWords(a.Count, a.Stream)
+			o.Machine = consumerMachine(a.Stream, a.Consumer)
+			o.Summary = fmt.Sprintf("%s in the last hour: %s", counted, a.Said)
+			o.Headline = clip(conditions.Capital(counted), 60)
+			o.Explanation = conditions.Capital(fmt.Sprintf("%s could not handle %s, and by the time it gave up the "+
+				"stream's time limit had passed, so the mesh did not keep them. Nothing to do; this clears an hour "+
+				"after the last one.", consumerWho(a.Stream, a.Consumer), map[bool]string{true: "some messages",
+				false: "a message"}[a.Count > 1]))
+			o.Resolved = "Resolved: no more messages of the stream " + a.Stream + " aged out undelivered"
+		}
 		out = append(out, o)
 	}
 	if f.busFaultsErr != nil {

cmd/mesh-controller/watchdogs.go

@@ -91,7 +91,10 @@ type signalFacts struct {
 	// deadLettersNewest is when DEAD_LETTERS kept the newest message it holds for each of those
 	// consumers: the condition counts the messages given up on, not the looks (novox/hq issue 440).
 	deadLettersNewest map[string]time.Time
-	advisoriesErr     error
+	// deadLettersAging is, for each consumer whose dead letters came from a restricted stream, when each ages out
+	// (novox/hq ADR 0376 §6); nil when it was not read.
+	deadLettersAging map[string][]link.Aging
+	advisoriesErr    error
 	// busFaultsErr is why the bus module's events could not be read this tick (novox/hq ADR 0357).
 	busFaultsErr error
 
@@ -211,6 +214,9 @@ type watchdogs struct {
 	// confirm holds back a row blind for one tick: a read that timed out once on a loaded store.
 	confirm confirming
 
+	// aging is what is remembered of restricted dead letters between looks (novox/hq ADR 0376 §6).
+	aging agingTally
+
 	mu     sync.Mutex
 	ticked time.Time
 	last   *signalFacts
@@ -264,6 +270,8 @@ func (w *watchdogs) tick(ctx context.Context) {
 func (w *watchdogs) see(running context.Context, f *signalFacts) {
 	var problems []string
 	var blind []conditions.Observation
+	// What aged out of DEAD_LETTERS, said and cleared in its own words before the rows clear it in theirs.
+	problems = append(problems, w.sayAgedOut(running, f)...)
 	for _, row := range watchedRows() {
 		if err := row.needs(f); err != nil {
 			blind = append(blind, blindRow(row, err))
@@ -382,6 +390,9 @@ func (w *watchdogs) gather(ctx context.Context) *signalFacts {
 			}
 		}
 	}
+	if f.advisoriesErr == nil && w.js != nil {
+		f.deadLettersAging, f.advisoriesErr = link.AgingDeadLetters(w.js.Context(), f.deadLetters)
+	}
 	f.handActs, f.handActsErr = w.gatherHandActs(ctx, now)
 	f.refusedSends, f.refusedSendsErr = inv.RefusedSendsSince(ctx, now.Add(-generationRefusalsSaid))
 	f.doing, f.doingErr = inv.NotDoingWhatTheyWereTold(ctx)