Tell the operator at once when sends wait for the bus's planned step, and say it before the merge (hq issue 336)

For 28 minutes on 2026-10-08 every send to the control-node was refused for a new bus build that only a
person's bus upgrade moves, and no condition said so: the refusal lived only in each walk's note, and S3
would have called it lateness after half an hour, in words that named neither the bus nor the verb.

- Row S17, bus.<module>.step-waiting: raised on the first watchdog tick after a walk's send is refused for
  the bus, for the operator, naming the machines, what waits, the bus build from and to, since when and
  mesh-controller.bus upgrade. It clears once the bus's machine runs the build the mesh holds.
- S3 leaves out a walk held only by the bus's step.
- A change that builds the bus says in its delivery plan and summary (which mesh/merge-gate carries) that
  merging it needs a person's bus upgrade, and that nothing else reaches its machine until then.
- TestReplay336 fails on the commit before and passes on this one.

cmd/mesh-controller/bus_waits_test.go

@@ -0,0 +1,159 @@
+package main
+
+import (
+	"context"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/novox/mesh-controller/internal/conditions"
+	"github.com/novox/mesh-controller/internal/inventory"
+)
+
+// novox/hq issue 336: a send held because it would replace the bus outside its planned step waits for a
+// person, so a person is told at once — what waits, behind which bus build, since when, and the verb that
+// ends it — and the wait is not read as a walk running late.
+
+// aBusOnAnchor is a new bus build waiting for its step on anchor: nats 88135ad0 there, 32307bd1 held.
+func aBusOnAnchor() busPending {
+	return busPending{module: "nats", machines: []string{"anchor"}, from: map[string]string{"anchor": "88135ad0aaaa"},
+		to: "32307bd1bbbb", same: map[string]bool{}}
+}
+
+// refusedNote is the note a walk keeps when its send was refused for the bus, as advanceHeld writes it.
+func refusedNote(b busPending, tier int) string {
+	held := "sending anchor would replace the bus (nats " + short(b.from["anchor"]) + " → " + short(b.to) +
+		"), which is a planned step: `bus upgrade --why …` snapshots its streams first and checks them after (novox/hq ADR 0236)"
+	return "tier " + string(rune('0'+tier)) + ": " + errBusWaits.Error() + ": " + held + " — tried again"
+}
+
+func TestASendHeldForTheBusStepIsFoundFromTheWalksItHolds(t *testing.T) {
+	now := time.Date(2026, 10, 8, 18, 30, 0, 0, time.UTC)
+	b := aBusOnAnchor()
+	walk := func(id, repo, note string, updated time.Time) inventory.Plan {
+		return inventory.Plan{ID: id, Repository: repo, Commit: "c0ffee001122", State: inventory.PlanRolling, Tier: 0,
+			Tiers: [][]string{{"mesh-host"}}, Modules: map[string]*inventory.PlanModule{"mesh-host": {}}, Note: note,
+			Updated: updated}
+	}
+	for _, c := range []struct {
+		name  string
+		bus   busPending
+		plans []inventory.Plan
+		first map[string]time.Time
+		want  []string
+		since time.Time
+	}{
+		{"a walk refused for the bus", b,
+			[]inventory.Plan{walk("plan-1", "novox/mesh-host", refusedNote(b, 0), now.Add(-27*time.Minute))}, nil,
+			[]string{"plan-1"}, now.Add(-27 * time.Minute)},
+		{"refused earlier than its last save, as this controller saw it", b,
+			[]inventory.Plan{walk("plan-1", "novox/mesh-host", refusedNote(b, 0), now.Add(-5*time.Minute))},
+			map[string]time.Time{"plan-1": now.Add(-28 * time.Minute)}, []string{"plan-1"}, now.Add(-28 * time.Minute)},
+		{"a walk refused for another reason", b,
+			[]inventory.Plan{walk("plan-1", "novox/mesh-host", "tier 0: the build seat is paused — tried again", now)}, nil,
+			nil, time.Time{}},
+		{"refused for an older bus build than the one held now", func() busPending { o := b; o.to = "99999999cccc"; return o }(),
+			[]inventory.Plan{walk("plan-1", "novox/mesh-host", refusedNote(b, 0), now)}, nil, nil, time.Time{}},
+		{"the bus already runs the build held: its step started", func() busPending {
+			o := aBusOnAnchor()
+			o.from = map[string]string{"anchor": o.to}
+			return o
+		}(), []inventory.Plan{walk("plan-1", "novox/mesh-host", refusedNote(b, 0), now)}, nil, nil, time.Time{}},
+		{"a walk waiting for its delivery's word asks no send", b, func() []inventory.Plan {
+			p := walk("plan-1", "novox/mesh-host", refusedNote(b, 0), now)
+			p.Delivery = &inventory.PlanDelivery{Awaits: "mesh-delivery"}
+			return []inventory.Plan{p}
+		}(), nil, nil, time.Time{}},
+	} {
+		t.Run(c.name, func(t *testing.T) {
+			got := busWaitsOf(c.plans, c.bus, func(id string) time.Time { return c.first[id] })
+			var ids []string
+			for _, w := range got.waits {
+				ids = append(ids, w.plan)
+			}
+			if strings.Join(ids, ",") != strings.Join(c.want, ",") {
+				t.Fatalf("held for the bus: %v, want %v", ids, c.want)
+			}
+			if len(c.want) > 0 {
+				if !got.waits[0].since.Equal(c.since) {
+					t.Errorf("waiting since %s, want %s", got.waits[0].since, c.since)
+				}
+				if strings.Join(got.machines, ",") != "anchor" || got.to != b.to || got.module != "nats" {
+					t.Errorf("behind %+v", got)
+				}
+			}
+		})
+	}
+}
+
+// **Said at once, not as lateness, and cleared when the step starts**: the walk held only for the bus raises
+// the bus's condition on the first tick, before any tier bound, naming what waits, the bus build from and to,
+// since when and `bus upgrade`, for the operator; S3 says nothing of it even past its bound; and the
+// condition clears once the bus's machine runs the new build.
+func TestASendHeldForTheBusStepIsSaidAtOnceAndNotAsLateness(t *testing.T) {
+	now := time.Date(2026, 10, 8, 18, 30, 0, 0, time.UTC)
+	b := aBusOnAnchor()
+	held := func(entered time.Duration) *signalFacts {
+		f := calm(now)
+		f.plans = []planFacts{{id: "plan-1", repository: "novox/mesh-host", commit: "c0ffee00", tier: 0, tiers: 1,
+			entered: now.Add(-entered), bound: 30 * time.Minute, waiting: refusedNote(b, 0), bus: true}}
+		f.bus = busFacts{module: "nats", to: b.to, from: b.from, machines: []string{"anchor"},
+			waits: []busWaitFacts{{plan: "plan-1", repository: "novox/mesh-host", commit: "c0ffee001122",
+				modules: []string{"mesh-host"}, since: now.Add(-time.Minute)}}}
+		return f
+	}
+
+	f := held(time.Minute)
+	got := watchBusWaits(f)
+	if len(got) != 1 {
+		t.Fatalf("a send held for the bus a minute ago raised %+v", got)
+	}
+	o := got[0]
+	if o.Key() != "bus.nats.step-waiting" || o.Kind != kindBusStepWaiting || o.Resolver != conditions.ResolverOperator {
+		t.Fatalf("raised %s (%s), resolver %q", o.Key(), o.Kind, o.Resolver)
+	}
+	for _, want := range []string{"anchor", "mesh-host", "88135ad0", "32307bd1", "mesh-controller.bus upgrade --why",
+		"--reversible", "--irreversible", "2026-10-08T18:29:00Z", "plan-1"} {
+		if !strings.Contains(o.Summary, want) {
+			t.Errorf("its summary does not say %q: %s", want, o.Summary)
+		}
+	}
+	if !strings.Contains(o.Explanation, "mesh-host") || !strings.Contains(o.Headline, "bus upgrade") || o.Needs == "" {
+		t.Errorf("its words do not say what waits and what the operator does: %+v", o)
+	}
+	if why, ok := conditions.PlainWords(conditions.Words{Headline: o.Headline, Explanation: o.Explanation,
+		Resolved: o.Resolved, Needs: o.Needs}, "anchor"); !ok {
+		t.Errorf("its words are not plain: %s", why)
+	}
+
+	// Past S3's bound: still the bus's wait, never a stalled walk.
+	late := held(45 * time.Minute)
+	if s3 := watchPlans(late); len(s3) != 0 {
+		t.Fatalf("a walk held only by the bus step was said stalled: %+v", s3)
+	}
+	// A walk held for something else past its bound is still stalled.
+	other := held(45 * time.Minute)
+	other.plans[0].bus = false
+	if s3 := watchPlans(other); len(s3) != 1 {
+		t.Fatalf("a walk held for something else past its bound raised %+v", s3)
+	}
+
+	// Through the keeper: open at the first tick, cleared when the step started.
+	store := conditions.NewInMemory()
+	k := conditions.NewKeeper(t.Context(), conditions.Options{Store: store, History: store, Teller: &conditions.Told{},
+		Now: func() time.Time { return now }})
+	defer k.Close(context.Background())
+	w := &watchdogs{keeper: k, started: now.Add(-time.Hour)}
+	w.see(t.Context(), held(time.Minute))
+	open, err := k.Open(t.Context())
+	if err != nil || len(open) != 1 || open[0].Key != "bus.nats.step-waiting" {
+		t.Fatalf("after the first tick, open: %+v (%v)", open, err)
+	}
+	started := held(time.Minute)
+	started.bus = busFacts{module: "nats", to: b.to}
+	started.plans[0].bus = false
+	w.see(t.Context(), started)
+	if open, _ := k.Open(t.Context()); len(open) != 0 {
+		t.Fatalf("the step started, and open: %+v", open)
+	}
+}