Work slower than the window says so, and one address is the bus's

Three faults the mesh's own logs showed this morning. A handler that outlives the acknowledgement
window was handed its message again while it was still working: acting on a merge builds modules,
minutes against a thirty-second window, so one merge ran the whole catalogue five times over. The
transport now says the work is in progress while it runs, which is where the window belongs.

Everything the mesh hands out — a token, a membership, a person's credential — took its address
from the enrolment setting, which on a mesh that has moved still names the broker it moved from:
the first person issued after the move was handed the retired broker's port. There is one bus, and
its address is the one the control plane is connected to.

And `operator issue` documented an argument order its parser refused.

cmd/mesh-controller/operator.go

@@ -189,13 +189,17 @@ func readPrivateKey(path string) (string, error) {
 func personIssue(ctx context.Context, args []string) error {
 	set := flag.NewFlagSet("operator issue", flag.ContinueOnError)
 	invokes := set.String("invokes", "", "the tools this person may call, comma-separated, or * for every one")
-	if err := set.Parse(args); err != nil {
+	// Flags on either side of the name, because the usage this command prints puts them after it —
+	// and the standard parser stops at the first thing that is not a flag, so the order the command
+	// documents was the one order it refused (2026-09-28).
+	positionals, err := parseAround(set, args)
+	if err != nil {
 		return err
 	}
-	if set.NArg() != 1 {
+	if len(positionals) != 1 {
 		return errors.New("operator issue <name> --invokes <tool,tool|*>")
 	}
-	name := set.Arg(0)
+	name := positionals[0]
 	if *invokes == "" {
 		return errors.New(
 			"say what this person may call: --invokes mesh-catalog.catalog_tools,gitea.repo_create, " +

cmd/mesh-controller/upgrades.go

@@ -245,10 +245,12 @@ func (f following) SourceMoved(ctx context.Context, m link.SourceMoved) error {
 	// commit it already records — so it is rebuilt and its record left alone. Writing this commit as
 	// its source would make it permanently behind a repository its manifest does not come from.
 	var from, packaging []inventory.Entry
+	already := 0
 	for _, e := range entries {
 		switch {
 		case sourceIs(e.Source, m):
 			if e.Source.BuiltFrom == m.Commit {
+				already++
 				continue
 			}
 			// **A merge older than the last look at the source is history, not a move.** The forge
@@ -264,6 +266,13 @@ func (f following) SourceMoved(ctx context.Context, m link.SourceMoved) error {
 		}
 	}
 	if len(from) == 0 && len(packaging) == 0 {
+		// "Already built from it" and "nothing reads it" are different facts, and reading the first
+		// as the second sends somebody looking for a broken trigger when the mesh is up to date.
+		if already > 0 {
+			fmt.Printf("%s/%s merged into %s (%.8s); %d module(s) the mesh holds are already built "+
+				"from it\n", m.Owner, m.Repo, m.Base, m.Commit, already)
+			return nil
+		}
 		fmt.Printf("%s/%s merged into %s (%.8s); nothing the mesh holds reads it\n",
 			m.Owner, m.Repo, m.Base, m.Commit)
 		return nil

internal/broker/broker.go

@@ -66,6 +66,19 @@ func FromEnvironment() (Broker, error) {
 	if err != nil {
 		return Broker{}, err
 	}
+	// **One bus, one address** (novox/hq ADR 0131). Everything the mesh hands out — a token, a
+	// machine's membership, a person's credential — must name the bus the control plane itself is
+	// connected to; the setting above predates the move and, on a mesh that has moved, still names
+	// the broker it moved from. The first person issued after the move was handed the retired
+	// broker's port and could not connect to anything (2026-09-28).
+	//
+	// Read from the credential rather than from a second setting somebody keeps in step: the
+	// control plane cannot be wrong about where it is connected.
+	if bus, on, err := OnNATS(); err == nil && on {
+		if where := strings.TrimPrefix(BareAddress(bus), "nats://"); where != "" {
+			address = where
+		}
+	}
 	return Broker{Address: address, Fingerprint: fingerprint}, nil
 }
 

internal/link/receive_nats.go

@@ -154,10 +154,47 @@ func (n *natsInbound) deliver(ctx context.Context, act func(context.Context, Con
 			m.seq = meta.Sequence.Stream
 			m.delivered = meta.NumDelivered
 		}
+		// **Work that outlives the acknowledgement window says so while it runs.**
+		//
+		// The bus waits a fixed time to be told a message was taken, and then hands it to whoever
+		// consumes next — which is right for a consumer that died and wrong for one that is busy.
+		// Acting on a merge builds every module the merge changed: minutes of work against a
+		// thirty-second window. So the same merge was handed over again while the first build was
+		// still running, and again after that — on 2026-09-28 one merge ran the mesh's whole
+		// catalogue five times over and exhausted a public registry's pull limit.
+		//
+		// Here rather than in each handler, because the window belongs to the transport and every
+		// handler would otherwise have to remember it. It changes nothing about a handler that
+		// dies: a message is kept alive only while this goroutine is, so a controller that stops
+		// stops saying so, and the bus redelivers exactly as it should.
+		working := make(chan struct{})
+		defer close(working)
+		go stillWorking(msg, working)
 	}
 	act(ctx, m)
 }
 
+// heartbeatWhileWorking is how often a handler still running tells the bus so — comfortably inside
+// the shortest acknowledgement window the mesh gives any of its consumers.
+const heartbeatWhileWorking = 10 * time.Second
+
+// stillWorking keeps one message alive until the work on it returns.
+//
+// An error is not worth reporting: what the bus does when it is not told is redeliver, which is
+// exactly what happens if this fails, and the handler's own outcome is the thing worth logging.
+func stillWorking(msg *nats.Msg, done <-chan struct{}) {
+	tick := time.NewTicker(heartbeatWhileWorking)
+	defer tick.Stop()
+	for {
+		select {
+		case <-done:
+			return
+		case <-tick.C:
+			_ = msg.InProgress()
+		}
+	}
+}
+
 // kindOfSubject is how this transport's addressing becomes what the mesh calls a message.
 //
 // By subject, which is the only thing the server enforces: a body claiming to be a report does not

internal/link/receive_nats_test.go

@@ -395,3 +395,60 @@ func TestEverySubjectTheControllerFollowsDecodesToAKind(t *testing.T) {
 		t.Error("a subject nobody follows decoded to a kind")
 	}
 }
+
+// **A handler slower than the acknowledgement window is not handed its message again.**
+//
+// The bus waits a fixed time to be told a message was taken and then redelivers, which is right for
+// a consumer that died and wrong for one that is busy. Acting on a merge builds modules — minutes
+// against a thirty-second window — and the same merge was handed over five times while the first
+// build was still running (2026-09-28). Here the window is two seconds and the work takes six.
+func TestNatsWorkSlowerThanTheWindowIsNotHandedOverAgain(t *testing.T) {
+	js := aBus(t)
+	// The controller's own consumer, with a window short enough to outlive in a test.
+	if err := js.EnsureConsumer(broker.Consumer{
+		Name: broker.ControllerName, Stream: "CONTROL", Push: true, AckWaitSeconds: 2,
+		Why: "a window short enough to outlive in a test",
+	}); err != nil {
+		t.Fatal(err)
+	}
+	var mu sync.Mutex
+	handled := 0
+	slow := make(chan struct{})
+	s, stop := servingOn(t, js, nil)
+	defer stop()
+	s.listener = slowly{func() {
+		mu.Lock()
+		handled++
+		first := handled == 1
+		mu.Unlock()
+		if first {
+			time.Sleep(6 * time.Second)
+			close(slow)
+		}
+	}}
+
+	body, err := json.Marshal(Report{Node: "anchor", Declared: "d1", Applied: []string{"store"}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := js.Context().Publish(ReportSubject("anchor"), body); err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case <-slow:
+	case <-time.After(30 * time.Second):
+		t.Fatal("the slow work never finished")
+	}
+	// A moment for a redelivery to arrive, if the bus were going to send one.
+	time.Sleep(3 * time.Second)
+	mu.Lock()
+	defer mu.Unlock()
+	if handled != 1 {
+		t.Fatalf("one report was handled %d times, so slow work is run again while it is running", handled)
+	}
+}
+
+// slowly is a listener that runs whatever it was given.
+type slowly struct{ work func() }
+
+func (s slowly) Heard(context.Context, Report) error { s.work(); return nil }