Merge fix/compose-pass-bounded: the composition pass never waits on marking haves

Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i

cmd/mesh-host/marker.go

@@ -0,0 +1,76 @@
+package main
+
+import (
+	"context"
+	"sync"
+	"time"
+)
+
+// markWithin bounds one marking of the haves (what the composer tells its agent: a refusal, the words it keeps). The marks
+// read and write each module's have over the bus one after the other; a bus that answers slowly or not at all must hold
+// the marking, never the composition pass (issue 1965, shanks 2026-11-05 22:22: a pass waited minutes on them).
+var markWithin = 45 * time.Second
+
+// marker runs the composer's marks of the haves off the composition pass: set returns at once, one worker runs what was
+// set last under each name, in the order the names were first set, each within markWithin. A mark set while one runs
+// replaces the pending mark of its name, so the haves are marked once for the newest word, not once for every pass.
+type marker struct {
+	run context.Context
+
+	mu      sync.Mutex
+	pending map[string]func(context.Context)
+	order   []string
+	working bool
+	idle    *sync.Cond
+}
+
+func newMarker(run context.Context) *marker {
+	m := &marker{run: run, pending: map[string]func(context.Context){}}
+	m.idle = sync.NewCond(&m.mu)
+	return m
+}
+
+// set takes the mark name should carry; it never waits on the bus.
+func (m *marker) set(name string, mark func(context.Context)) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if _, queued := m.pending[name]; !queued {
+		m.order = append(m.order, name)
+	}
+	m.pending[name] = mark
+	if m.working {
+		return
+	}
+	m.working = true
+	go m.work()
+}
+
+func (m *marker) work() {
+	for {
+		m.mu.Lock()
+		if len(m.order) == 0 || m.run.Err() != nil {
+			m.order, m.pending = nil, map[string]func(context.Context){}
+			m.working = false
+			m.idle.Broadcast()
+			m.mu.Unlock()
+			return
+		}
+		name := m.order[0]
+		m.order = m.order[1:]
+		mark := m.pending[name]
+		delete(m.pending, name)
+		m.mu.Unlock()
+		ctx, cancel := context.WithTimeout(m.run, markWithin)
+		mark(ctx)
+		cancel()
+	}
+}
+
+// wait returns once no mark is pending or running (for tests).
+func (m *marker) wait() {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	for m.working {
+		m.idle.Wait()
+	}
+}

cmd/mesh-host/marker_test.go

@@ -0,0 +1,47 @@
+package main
+
+import (
+	"context"
+	"testing"
+	"time"
+)
+
+// Issue 1965 (shanks 2026-11-05 22:22): the marks of the haves never hold the composition pass. A mark that stalls on the
+// bus is cut at markWithin, a mark set while one runs replaces the pending one of its name, and set itself never waits.
+func TestMarksOfTheHavesNeverHoldThePass(t *testing.T) {
+	setMarkWithin(t, 30*time.Millisecond)
+	m := newMarker(context.Background())
+	var ran []string
+	stalled := make(chan struct{})
+	began := time.Now()
+	m.set("refused", func(ctx context.Context) { // a have write that stalls until its bound
+		close(stalled)
+		<-ctx.Done()
+		ran = append(ran, "refused-stalled")
+	})
+	<-stalled
+	// Set while the first runs: only the newest of a name is kept.
+	m.set("kept", func(context.Context) { ran = append(ran, "kept-old") })
+	m.set("kept", func(context.Context) { ran = append(ran, "kept-new") })
+	m.set("refused", func(context.Context) { ran = append(ran, "refused-new") })
+	if took := time.Since(began); took > 20*time.Millisecond {
+		t.Errorf("setting marks took %v; it must not wait on the stalled one", took)
+	}
+	m.wait()
+	want := []string{"refused-stalled", "kept-new", "refused-new"}
+	if len(ran) != len(want) {
+		t.Fatalf("ran %v; want %v", ran, want)
+	}
+	for i := range want {
+		if ran[i] != want[i] {
+			t.Fatalf("ran %v; want %v", ran, want)
+		}
+	}
+}
+
+func setMarkWithin(t *testing.T, d time.Duration) {
+	t.Helper()
+	old := markWithin
+	markWithin = d
+	t.Cleanup(func() { markWithin = old })
+}

cmd/mesh-host/nodeagent.go

@@ -596,6 +596,8 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus
 			// No reader is no gate that lets everything through: the node acts on nothing new, as one that waits does.
 			say(err.Error() + "; the node composes nothing new until it is started again and reads the list (novox/hq ADR 0636 §1)")
 		}
+		// The marks of the haves run off the composition pass, each bounded (see marker).
+		marks := newMarker(run)
 		hooks := composerHooks{
 			seatList:       seats,
 			seatListFailed: err != nil,
@@ -621,13 +623,14 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus
 			},
 			refused: func(why string, held map[string]string) {
 				if now := own.get(); now != nil && run.Err() == nil {
-					now.NotComposedEach(run, why, held) // a module held alone marks that module, never the others (issue 1126)
+					// A module held alone marks that module, never the others (issue 1126); the marking never holds the pass.
+					marks.set("refused", func(ctx context.Context) { now.NotComposedEach(ctx, why, held) })
 				}
 			},
 			// And what it keeps of a module as it runs, for a manifest that does not say it yet: on the module's have (ADR 0601 §4).
 			kept: func(kept map[string][]string) {
 				if now := own.get(); now != nil && run.Err() == nil {
-					now.KeptNow(run, kept)
+					marks.set("kept", func(ctx context.Context) { now.KeptNow(ctx, kept) })
 				}
 			},
 			// And the vault's authority beside the controller's, and this node's certificate asked of the vault when a

internal/selfcompose/compose_while_passing_test.go

@@ -0,0 +1,82 @@
+package selfcompose
+
+import (
+	"context"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
+	"git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner"
+	nodeengine "git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
+	"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants"
+	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
+	"github.com/novox/mesh-host/internal/agent"
+	"github.com/novox/mesh-host/internal/buckets"
+	"github.com/novox/mesh-host/internal/buckets/bucketstest"
+	"github.com/novox/mesh-host/internal/compositions"
+)
+
+// Issue 1965 (shanks 2026-11-05 22:27: the verb compose gave no answer in 30 s while a pass waited on the bus): an ask
+// that comes while a pass runs is answered at once, says one more pass is queued and what the running one waits on, and
+// the queued pass is made when the running one ends.
+func TestComposeAnsweredWhileAPassRuns(t *testing.T) {
+	b := newFakeBus()
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	env := b.env()
+	hold := make(chan struct{})
+	var passes atomic.Int64
+	reached := make(chan struct{}, 8)
+	env.Running = func() []compositions.Composed {
+		passes.Add(1)
+		reached <- struct{}{}
+		if passes.Load() == 1 {
+			<-hold // the first pass waits on the bus
+		}
+		return nil
+	}
+	c := New(env)
+	out, err := c.Watch(ctx, "n1.>")
+	if err != nil {
+		t.Fatal(err)
+	}
+	place := assignedValues(t, meshmoduleassigner.Assignments{Node: "n1", Decided: meshmoduleassigner.NodeDecided{Hub: true, Address: "10.77.0.1"}})
+	have, _ := nodeengineclient.EncodeHaveNode(nodeengine.HaveByNodeNodeValue{Record: &nodeengine.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}})
+	nodeWant, _ := bucketstest.NodeWant(meshwants.WantsByWantNodeNodeValue{Self: true})
+	b.putAssigned(t, place)
+	b.put(t, buckets.Have, "n1.node", have)
+	b.put(t, buckets.Wants, "want.n1.node", nodeWant)
+	for _, bucket := range []string{buckets.Wants, buckets.Assignments, buckets.Have, buckets.Modules, ResolvedBucket, builds.BuilderBuildsBucket} {
+		b.send(bucket, agent.Update{Caught: true})
+	}
+	go func() {
+		for range out {
+		}
+	}()
+	select {
+	case <-reached:
+	case <-time.After(5 * time.Second):
+		t.Fatal("no pass began")
+	}
+	answered := make(chan Pass, 1)
+	go func() {
+		p, _ := c.ComposeNow(ctx)
+		answered <- p
+	}()
+	select {
+	case p := <-answered:
+		if !strings.Contains(p.Queued, "a pass is running") {
+			t.Errorf("answered %+v; want it to say a pass is running and one is queued", p)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("the verb compose gave no answer while a pass ran")
+	}
+	close(hold)
+	select {
+	case <-reached: // the queued pass
+	case <-time.After(5 * time.Second):
+		t.Fatal("the queued pass was not made when the running one ended")
+	}
+}

internal/selfcompose/selfcompose.go

@@ -191,6 +191,8 @@ type Composer struct {
 	wait  time.Duration
 	// asked are the compose verb's asks (ComposeNow), each answered once the pass it asks is made.
 	asked chan chan Pass
+	// passing is set while a pass is made.
+	passing atomic.Bool
 	// last is how the last pass went, under mu.
 	last Pass
 	// watching says the composer's loop runs (Watch), so an ask of a pass can be answered.
@@ -240,6 +242,9 @@ type Pass struct {
 	Again time.Time `json:"again,omitempty"`
 	// Earlier says the answer is the last pass, made less than askedWithin ago, and no new one was made for the ask.
 	Earlier bool `json:"earlier,omitempty"`
+	// Queued says a pass was running when the ask came: one more is made when it ends, and this answer is not its result.
+	// It says what the running pass waits on and since when.
+	Queued string `json:"queued,omitempty"`
 }
 
 // LastHeld is what the last pass held, each module with why; nil before a pass or when it held none.
@@ -331,6 +336,16 @@ func (c *Composer) ComposeNow(ctx context.Context) (Pass, error) {
 		return last, nil
 	}
 	reply := make(chan Pass, 1)
+	if c.passing.Load() {
+		// A pass runs: the ask joins the one made when it ends (all asks of the time share it), and is answered now, never
+		// held for the pass (issue 1965: the verb gave no answer for minutes while a pass waited on the bus).
+		select {
+		case c.asked <- reply:
+		default: // eight asks already wait for that pass
+		}
+		last.Queued = "a pass is running and waits on " + c.waitingOn() + "; one more is made when it ends"
+		return last, nil
+	}
 	select {
 	case c.asked <- reply:
 	case <-ctx.Done():
@@ -539,7 +554,10 @@ func (c *Composer) Watch(ctx context.Context, _ string) (<-chan agent.Update, er
 				continue
 			}
 			dirty = false
-			for _, u := range c.compose(ctx) {
+			c.passing.Store(true)
+			updates := c.compose(ctx)
+			c.passing.Store(false)
+			for _, u := range updates {
 				select {
 				case out <- u:
 				case <-ctx.Done():