Pin the group's mesh-sdk head, whose step log goes quiet after one failed publish

node-tools/go.mod

@@ -8,7 +8,7 @@ require (
 )
 
 require (
-	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468
+	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030100935-c2ec8b6b2fde
 	github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op // indirect
 	github.com/google/go-tpm v0.9.8 // indirect
 	github.com/klauspost/compress v1.20.0 // indirect

node-tools/go.sum

@@ -1,5 +1,11 @@
 git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468 h1:UcMee+IV2ii5E4dRZdQUaSaeh9IedTFYumuQbBiB0k0=
 git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030095131-16bc0dcf3e9d h1:jFOz6wUfZvIayd4DctwlaFKKZHQNME6NLs2OS1rNv30=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030095131-16bc0dcf3e9d/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030095816-530d8d3007dc h1:Wdjl2gomooHswJ5qpkwZPq/ureZgXVw6FMCsgzPJhPo=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030095816-530d8d3007dc/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030100935-c2ec8b6b2fde h1:eKXSllHCS31GgMazfvZ7BeiLoXIUibnUmzxWVkj+Vcw=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261030100935-c2ec8b6b2fde/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
 github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
 github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
 github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=

node-tools/internal/bus/log.go

@@ -221,49 +221,17 @@ func (c *Conn) LogRead(module, name string, from uint64, keys string, limit int)
 	if from == 0 {
 		from = 1
 	}
-	req, _ := json.Marshal(map[string]any{"seq": from, "next_by_subj": filter, "batch": limit, "max_bytes": LogPageBytes})
-	inbox := c.nc.NewInbox()
-	sub, err := c.nc.SubscribeSync(inbox)
+	msgs, next, err := c.directBatch(l.Stream, filter, from, limit, name, func() error { return notOnTheBus(l) })
 	if err != nil {
 		return nil, err
 	}
-	defer func() { _ = sub.Unsubscribe() }()
-	if err := c.nc.PublishRequest("$JS.API.DIRECT.GET."+l.Stream, inbox, req); err != nil {
-		return nil, err
-	}
-	page := &LogPage{Entries: []LogEntry{}}
-	deadline := time.Now().Add(StateTimeout)
-	for done := false; !done; {
-		msg, err := sub.NextMsg(time.Until(deadline))
+	page := &LogPage{Entries: []LogEntry{}, Next: next}
+	for _, msg := range msgs {
+		e, err := logEntryOf(l, msg)
 		if err != nil {
-			if errors.Is(err, nats.ErrNoResponders) {
-				return nil, notOnTheBus(l)
-			}
-			if errors.Is(err, nats.ErrTimeout) {
-				return nil, fmt.Errorf("the read of %s did not end in %s", name, StateTimeout)
-			}
 			return nil, err
 		}
-		switch status := msg.Header.Get("Status"); status {
-		case "":
-			e, err := logEntryOf(l, msg)
-			if err != nil {
-				return nil, err
-			}
-			page.Entries = append(page.Entries, *e)
-		case "204": // the end of the batch: what is still pending after the last entry sent
-			pending, _ := strconv.ParseUint(msg.Header.Get("Nats-Num-Pending"), 10, 64)
-			if pending > 0 && len(page.Entries) > 0 {
-				page.Next = page.Entries[len(page.Entries)-1].Sequence + 1
-			}
-			done = true
-		case "404": // nothing from that sequence on
-			done = true
-		case "503":
-			return nil, notOnTheBus(l)
-		default:
-			return nil, fmt.Errorf("the read of %s was refused by the bus: %s %s", name, status, msg.Header.Get("Description"))
-		}
+		page.Entries = append(page.Entries, *e)
 	}
 	info, err := c.logInfo(l)
 	if err != nil {
@@ -273,6 +241,58 @@ func (c *Conn) LogRead(module, name string, from uint64, keys string, limit int)
 	return page, nil
 }
 
+// directBatch is the messages of a stream from a sequence on, on the subjects matching filter, at most limit and about
+// LogPageBytes, by one batch direct get (NATS 2.11), and the sequence to read on from, 0 when nothing follows. name is
+// what is read, for its errors; missing is the error of a stream that is not on the bus.
+func (c *Conn) directBatch(stream, filter string, from uint64, limit int, name string, missing func() error) ([]*nats.Msg, uint64, error) {
+	if from == 0 {
+		from = 1
+	}
+	req, _ := json.Marshal(map[string]any{"seq": from, "next_by_subj": filter, "batch": limit, "max_bytes": LogPageBytes})
+	inbox := c.nc.NewInbox()
+	sub, err := c.nc.SubscribeSync(inbox)
+	if err != nil {
+		return nil, 0, err
+	}
+	defer func() { _ = sub.Unsubscribe() }()
+	if err := c.nc.PublishRequest("$JS.API.DIRECT.GET."+stream, inbox, req); err != nil {
+		return nil, 0, err
+	}
+	var msgs []*nats.Msg
+	var next uint64
+	deadline := time.Now().Add(StateTimeout)
+	for done := false; !done; {
+		msg, err := sub.NextMsg(time.Until(deadline))
+		if err != nil {
+			if errors.Is(err, nats.ErrNoResponders) {
+				return nil, 0, missing()
+			}
+			if errors.Is(err, nats.ErrTimeout) {
+				return nil, 0, fmt.Errorf("the read of %s did not end in %s", name, StateTimeout)
+			}
+			return nil, 0, err
+		}
+		switch status := msg.Header.Get("Status"); status {
+		case "":
+			msgs = append(msgs, msg)
+		case "204": // the end of the batch: what is still pending after the last message sent
+			pending, _ := strconv.ParseUint(msg.Header.Get("Nats-Num-Pending"), 10, 64)
+			if pending > 0 && len(msgs) > 0 {
+				last, _ := strconv.ParseUint(msgs[len(msgs)-1].Header.Get("Nats-Sequence"), 10, 64)
+				next = last + 1
+			}
+			done = true
+		case "404": // nothing from that sequence on
+			done = true
+		case "503":
+			return nil, 0, missing()
+		default:
+			return nil, 0, fmt.Errorf("the read of %s was refused by the bus: %s %s", name, status, msg.Header.Get("Description"))
+		}
+	}
+	return msgs, next, nil
+}
+
 // LogLast is a key's last entry, or nil when it has none.
 func (c *Conn) LogLast(module, name, key string) (*LogEntry, error) {
 	l, err := c.issuedLog(module, name)

node-tools/internal/bus/streamread.go

@@ -0,0 +1,126 @@
+package bus
+
+import (
+	"encoding/json"
+	"fmt"
+	"sort"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/nats-io/nats.go"
+)
+
+// A module reading its own stream back (novox/hq ADR 0363; the operator, 2026-10-30: a build step's whole failure output
+// is said on its build module's stream, and the module's tools read it there). As a log's read: one batch direct get,
+// no consumer, and only a stream the module's own membership names (`streams`); the bus grants its owner that stream's
+// info and direct get and no other's (mesh-sdk buscompose ownStreamGrants).
+
+// StreamEntry is one event of a module's own stream: its sequence, its event (the subject after the stream's
+// `…event.`), its body and when the bus took it.
+type StreamEntry struct {
+	Sequence uint64          `json:"sequence"`
+	Event    string          `json:"event"`
+	Body     json.RawMessage `json:"body"`
+	At       time.Time       `json:"at"`
+}
+
+// StreamPage is one page of a stream's read: Next is the sequence to continue from, 0 when nothing follows; Last the
+// stream's last sequence.
+type StreamPage struct {
+	Entries []StreamEntry `json:"entries"`
+	Next    uint64        `json:"next"`
+	Last    uint64        `json:"last"`
+}
+
+// streamBusName is a stream's local name as the bus holds it (mesh-sdk catalogue.StreamBusName): capitals, a hyphen an
+// underscore; a bus name is itself.
+func streamBusName(name string) string { return strings.ToUpper(strings.ReplaceAll(name, "-", "_")) }
+
+// issuedStream is the stream of its own a module may read by a name, local or the bus's, or the reason it may not.
+func (c *Conn) issuedStream(module, name string) (StreamIssued, error) {
+	m := c.Membership(module)
+	if m == nil {
+		return StreamIssued{}, fmt.Errorf("%s has no membership issued on %s yet, so no stream of it is readable "+
+			"until the mesh issues one (novox/hq ADR 0363)", module, c.node)
+	}
+	var names []string
+	for _, s := range m.Streams {
+		if s.Stream == streamBusName(name) && strings.Contains(s.Subject, "{event}") {
+			return s, nil
+		}
+		names = append(names, s.Stream)
+	}
+	sort.Strings(names)
+	issued := "none"
+	if len(names) > 0 {
+		issued = strings.Join(names, ", ")
+	}
+	return StreamIssued{}, fmt.Errorf("%s keeps no stream called %q: a module reads the streams it declares under `streams`, "+
+		"and no other's; it was issued: %s (novox/hq ADR 0363)", module, name, issued)
+}
+
+// StreamRead is a module's own stream's events from a sequence on, whose event names match a pattern (`*` one name, `>`
+// the rest, last; empty is every event), at most limit (LogReadDefault when 0, at most LogReadMax) and about LogPageBytes.
+func (c *Conn) StreamRead(module, name, events string, from uint64, limit int) (*StreamPage, error) {
+	s, err := c.issuedStream(module, name)
+	if err != nil {
+		return nil, err
+	}
+	switch {
+	case limit == 0:
+		limit = LogReadDefault
+	case limit < 0 || limit > LogReadMax:
+		return nil, fmt.Errorf("a read's limit is 1 to %d events, %d when none is given, not %d", LogReadMax, LogReadDefault, limit)
+	}
+	if events == "" {
+		events = ">"
+	}
+	parts := strings.Split(events, ".")
+	for i, p := range parts {
+		if !(p == ">" && i == len(parts)-1) && p != "*" && !logToken.MatchString(p) {
+			return nil, fmt.Errorf("%q is not an event pattern: names, `*` for one and `>` for the rest, last", events)
+		}
+	}
+	prefix, _, _ := strings.Cut(s.Subject, "{event}")
+	missing := func() error {
+		return fmt.Errorf("the stream %s is issued and not on the bus yet: the controller creates it from the catalogue "+
+			"on its next raise, and reads it directly once it allows that (novox/hq ADR 0363)", s.Stream)
+	}
+	msgs, next, err := c.directBatch(s.Stream, prefix+events, from, limit, s.Stream, missing)
+	if err != nil {
+		return nil, err
+	}
+	page := &StreamPage{Entries: []StreamEntry{}, Next: next}
+	for _, msg := range msgs {
+		e, err := streamEntryOf(s.Stream, prefix, msg)
+		if err != nil {
+			return nil, err
+		}
+		page.Entries = append(page.Entries, *e)
+	}
+	info, err := c.logInfo(LogIssued{Name: s.Stream, Stream: s.Stream})
+	if err != nil {
+		return nil, err
+	}
+	page.Last = info.Last
+	return page, nil
+}
+
+// streamEntryOf is an event a direct get answered: its event is its subject after prefix.
+func streamEntryOf(stream, prefix string, msg *nats.Msg) (*StreamEntry, error) {
+	subject := msg.Header.Get("Nats-Subject")
+	event, ok := strings.CutPrefix(subject, prefix)
+	if !ok {
+		return nil, fmt.Errorf("the bus answered an event of %s on %q, which is not under %s", stream, subject, prefix)
+	}
+	seq, err := strconv.ParseUint(msg.Header.Get("Nats-Sequence"), 10, 64)
+	if err != nil {
+		return nil, fmt.Errorf("the bus answered an event of %s with no sequence", stream)
+	}
+	at, err := time.Parse(time.RFC3339Nano, msg.Header.Get("Nats-Time-Stamp"))
+	if err != nil {
+		return nil, fmt.Errorf("the bus answered event %d of %s with no time: %v", seq, stream, err)
+	}
+	return &StreamEntry{Sequence: seq, Event: event, Body: valueOf(msg.Data), At: at.UTC()}, nil
+}

node-tools/internal/launch/event_answer_test.go

@@ -38,6 +38,7 @@ func (b *subscribing) Publish(json.RawMessage) error                          {
 func (b *subscribing) Ask(json.RawMessage) (json.RawMessage, error)           { return nil, nil }
 func (b *subscribing) State(string, json.RawMessage) (json.RawMessage, error) { return nil, nil }
 func (b *subscribing) Log(string, json.RawMessage) (json.RawMessage, error)   { return nil, nil }
+func (b *subscribing) StreamRead(json.RawMessage) (json.RawMessage, error)    { return nil, nil }
 func (b *subscribing) Watch(json.RawMessage, func(json.RawMessage) error) (func(), error) {
 	return func() {}, nil
 }

node-tools/internal/launch/launch.go

@@ -80,6 +80,9 @@ type Bus interface {
 	Watch(params json.RawMessage, deliver func(change json.RawMessage) error) (stop func(), err error)
 	// Log answers a bundle's `append`, `read`, `last` and `info` on its module's log (novox/hq ADR 0297).
 	Log(verb string, params json.RawMessage) (json.RawMessage, error)
+	// StreamRead answers a bundle's `read` of one of its module's own streams (novox/hq ADR 0363): {stream, subject,
+	// from, limit}, as a log's read.
+	StreamRead(params json.RawMessage) (json.RawMessage, error)
 	// Seat answers a bundle's seat traffic (novox/hq ADR 0259 §3): `publish`, `prove` and `record` at once;
 	// `take` and `answer` bind once and hand each message or proof on through handOn, to whichever child
 	// of the module asked last — a restarted child asks again as its code runs again.
@@ -333,6 +336,11 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an
 							if answered, err = mesh.Log(strings.TrimPrefix(m.Method, "mesh/log."), m.Params); err == nil {
 								result = answered
 							}
+						case "mesh/stream.read":
+							var answered json.RawMessage
+							if answered, err = mesh.StreamRead(m.Params); err == nil {
+								result = answered
+							}
 						case "mesh/provision.get":
 							var answered json.RawMessage
 							if answered, err = mesh.Provision("get", m.Params); err == nil {

node-tools/internal/meshtest/meshtest.go

@@ -278,7 +278,7 @@ func (m *Mesh) ModuleStream(t *testing.T, module, name string) {
 	t.Helper()
 	_ = m.js.DeleteStream(name)
 	if _, err := m.js.AddStream(&nats.StreamConfig{Name: name, Subjects: []string{"mesh.stream." + module + ".event.>"},
-		Retention: nats.LimitsPolicy, Discard: nats.DiscardOld}); err != nil {
+		Retention: nats.LimitsPolicy, Discard: nats.DiscardOld, AllowDirect: true}); err != nil {
 		t.Fatal(err)
 	}
 }

node-tools/internal/runtime/log_test.go

@@ -21,6 +21,8 @@ func logBuses(t *testing.T, maxBytes int64) (issues, other launch.Bus, mesh *mt.
 	mesh = mt.New(t)
 	m := mt.MembershipOf("issues", "anchor", false, nil)
 	m.Logs = []bus.LogIssued{mesh.Log(t, "issues", "changes", maxBytes)}
+	mesh.ModuleStream(t, "issues", "ISSUE_MOVES")
+	m.Streams = []bus.StreamIssued{{Stream: "ISSUE_MOVES", Events: []string{"moved.>"}, Subject: "mesh.stream.issues.event.{event}"}}
 	mesh.Issue(t, m)
 	mesh.Issue(t, mt.MembershipOf("other", "anchor", false, nil))
 	conn := connect(t, "node-tools", "anchor")
@@ -111,7 +113,7 @@ func infoOf(t *testing.T, b launch.Bus) map[string]uint64 {
 // RFC 3339; last is a key's last entry or null; info says entries, bytes, cap, first and last — and no
 // consumer is made for any of it.
 func TestALogAnswersEachVerbOnTheBus(t *testing.T) {
-	b, _, mesh := logBuses(t, 1<<20)
+	b, other, mesh := logBuses(t, 1<<20)
 	before := time.Now().Add(-time.Second)
 
 	var seqs []uint64
@@ -189,6 +191,45 @@ func TestALogAnswersEachVerbOnTheBus(t *testing.T) {
 	if n := mesh.Consumers(t, "LOG_issues_changes"); n != 0 {
 		t.Fatalf("the runtime made %d consumers of the log; it uses none", n)
 	}
+
+	// A module reads its own stream back as it reads its log (mesh/stream.read): by its local name, an event pattern,
+	// in pages; another module's stream is refused with the reason.
+	for i, ev := range []string{"moved.a.1", "moved.b.1", "moved.a.1"} {
+		if err := b.Publish(json.RawMessage(fmt.Sprintf(`{"key":%q,"body":{"n":%d}}`, ev, i))); err != nil {
+			t.Fatal(err)
+		}
+	}
+	streamOf := func(b launch.Bus, params string) (wireStream, error) {
+		var p wireStream
+		raw, err := b.StreamRead(json.RawMessage(params))
+		if err == nil {
+			err = json.Unmarshal(raw, &p)
+		}
+		return p, err
+	}
+	page, err := streamOf(b, `{"stream":"issue-moves","subject":"moved.a.>","limit":1}`)
+	if err != nil || len(page.Entries) != 1 || page.Entries[0].Event != "moved.a.1" || string(page.Entries[0].Body) != `{"n":0}` || page.Next == 0 || page.Last != 3 {
+		t.Fatalf("the first page of moved.a: %+v, %v", page, err)
+	}
+	page, err = streamOf(b, fmt.Sprintf(`{"stream":"ISSUE_MOVES","subject":"moved.a.>","from":%d}`, page.Next))
+	if err != nil || len(page.Entries) != 1 || string(page.Entries[0].Body) != `{"n":2}` || page.Next != 0 {
+		t.Fatalf("the second page of moved.a: %+v, %v", page, err)
+	}
+	if _, err := streamOf(other, `{"stream":"issue-moves"}`); err == nil || !strings.Contains(err.Error(), "keeps no stream") {
+		t.Fatalf("another module's stream: %v", err)
+	}
+}
+
+// wireStream is a stream read's answer with exactly the contract's field names (mesh-sdk stdio.StreamPage).
+type wireStream struct {
+	Entries []struct {
+		Sequence uint64          `json:"sequence"`
+		Event    string          `json:"event"`
+		Body     json.RawMessage `json:"body"`
+		At       string          `json:"at"`
+	} `json:"entries"`
+	Next uint64 `json:"next"`
+	Last uint64 `json:"last"`
 }
 
 // A page holds about 4 MiB at most, whatever its limit, and says where to continue.

node-tools/internal/runtime/runtime.go

@@ -988,6 +988,26 @@ func (b *moduleBus) Log(verb string, params json.RawMessage) (json.RawMessage, e
 	return json.Marshal(answer)
 }
 
+// StreamRead answers a bundle's read of one of its module's own streams (ADR 0363): {stream, subject, from, limit}, the
+// stream by its local name, the subject an event pattern. The runtime refuses, with the reason, a stream the module's
+// membership does not name.
+func (b *moduleBus) StreamRead(params json.RawMessage) (json.RawMessage, error) {
+	var asked struct {
+		Stream  string `json:"stream"`
+		Subject string `json:"subject"`
+		From    uint64 `json:"from"`
+		Limit   int    `json:"limit"`
+	}
+	if err := json.Unmarshal(params, &asked); err != nil || asked.Stream == "" {
+		return nil, fmt.Errorf("mesh/stream.read names no stream: {stream, subject, from, limit}")
+	}
+	page, err := b.all.conn.StreamRead(b.module, asked.Stream, asked.Subject, asked.From, asked.Limit)
+	if err != nil {
+		return nil, err
+	}
+	return json.Marshal(page)
+}
+
 // Watch hands a bundle its module's state as it is and as it changes (ADR 0201): `{state, key}`, the
 // key a pattern with `*` and `**`, empty for every key.
 func (b *moduleBus) Watch(params json.RawMessage, deliver func(json.RawMessage) error) (func(), error) {