R-11: mesh/publish with ack answers its sequence or {error}, mesh/subscribe with deliver new makes an ordered consumer; finding-calls is stated

The subscription's key resolves through mesh-sdk's buscompose.EventSubjects (lead answer A-153; the sdk commit EE-S1 is
cherry-picked into this change's sdk placement), a wildcard is refused.

Builds: design 47 G §559, §488, §576, §622

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

node-tools/internal/bus/bus.go

@@ -852,6 +852,12 @@ func answerOf(data []byte) (Answered, error) {
 // PublishAs emits an event as a module: published into JetStream and awaited, de-duplicated by its
 // own id (ADR 0042). The body is the payload; the metadata rides as headers.
 func (c *Conn) PublishAs(module string, env Envelope) error {
+	_, err := c.PublishAsAcked(module, env)
+	return err
+}
+
+// PublishAsAcked is PublishAs that also answers the sequence the bus acknowledged the event at.
+func (c *Conn) PublishAsAcked(module string, env Envelope) (uint64, error) {
 	// On the module's own stream when its membership names one carrying the event (ADR 0363 §2), on EVENTS
 	// otherwise — and only there, as before, for a membership that names no stream.
 	msg := nats.NewMsg(eventSubject(c.Membership(module), module, env.Key))
@@ -873,8 +879,11 @@ func (c *Conn) PublishAs(module string, env Envelope) error {
 	if id := env.Headers["x-event-id"]; id != "" {
 		opts = append(opts, nats.MsgId(id))
 	}
-	_, err := c.js.PublishMsg(msg, opts...)
-	return err
+	ack, err := c.js.PublishMsg(msg, opts...)
+	if err != nil {
+		return 0, err
+	}
+	return ack.Sequence, nil
 }
 
 // Say publishes one message on core NATS, kept by nobody: a heartbeat, whose loss is the next one.

node-tools/internal/bus/subscribenew.go

@@ -0,0 +1,87 @@
+package bus
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"slices"
+	"strings"
+
+	"git.novox.be/novox/mesh-sdk/go/buscompose"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// SubscribeNew hears one event, named `<emitter>.<event>`, from the next message on, on an ordered consumer of its own on each
+// stream that carries it, and hands every message to deliver as the envelope a module sees. The subjects the name lands on are
+// mesh-sdk's one rule (buscompose.EventSubjects, design 47 G §559, A-153), so a name with a wildcard is refused, naming the rule;
+// a name no stream on the bus carries is an error, never a subscription that hears nothing. stop ends every consumer.
+func (c *Conn) SubscribeNew(name string, deliver func(json.RawMessage) error) (func(), error) {
+	subjects, err := buscompose.EventSubjects(name, "")
+	if err != nil {
+		return nil, err
+	}
+	js, err := c.JetStream()
+	if err != nil {
+		return nil, err
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
+	defer cancel()
+	// The streams that carry each subject, found by the bus: a namespace no stream carries hears nothing and is skipped.
+	byStream := map[string][]string{}
+	for _, subject := range subjects {
+		if strings.HasSuffix(subject, ".>") {
+			continue
+		}
+		stream, err := js.StreamNameBySubject(ctx, subject)
+		if err != nil || stream == "" {
+			continue
+		}
+		byStream[stream] = append(byStream[stream], subject, subject+".>")
+	}
+	if len(byStream) == 0 {
+		return nil, fmt.Errorf("no stream on the bus carries %s (%s): nothing would be heard", name, strings.Join(subjects, ", "))
+	}
+	var stops []func()
+	stopAll := func() {
+		for _, s := range stops {
+			s()
+		}
+	}
+	for _, stream := range slices.Sorted(mapKeys(byStream)) {
+		cons, err := js.OrderedConsumer(ctx, stream, jetstream.OrderedConsumerConfig{FilterSubjects: byStream[stream],
+			DeliverPolicy: jetstream.DeliverNewPolicy})
+		if err != nil {
+			stopAll()
+			return nil, fmt.Errorf("an ordered consumer of %s on %s: %w", name, stream, err)
+		}
+		cc, err := cons.Consume(func(m jetstream.Msg) {
+			env, ok := envelopeOf(m.Subject(), m.Headers(), m.Data())
+			if !ok {
+				return
+			}
+			raw, err := json.Marshal(env)
+			if err != nil {
+				return
+			}
+			if err := deliver(raw); err != nil && c.Logf != nil {
+				c.Logf("[mesh-tools] a subscription for new %s messages did not take one: %v", name, err)
+			}
+		})
+		if err != nil {
+			stopAll()
+			return nil, fmt.Errorf("consuming %s on %s: %w", name, stream, err)
+		}
+		stops = append(stops, cc.Stop)
+	}
+	return stopAll, nil
+}
+
+func mapKeys(m map[string][]string) func(func(string) bool) {
+	return func(yield func(string) bool) {
+		for k := range m {
+			if !yield(k) {
+				return
+			}
+		}
+	}
+}

node-tools/internal/launch/launch.go

@@ -387,7 +387,29 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an
 						asked := time.Now()
 						switch m.Method {
 						case "mesh/publish":
-							err = mesh.Publish(m.Params)
+							var said struct {
+								Ack bool `json:"ack"`
+							}
+							_ = json.Unmarshal(m.Params, &said)
+							if !said.Ack {
+								err = mesh.Publish(m.Params)
+								break
+							}
+							// With ack the runtime answers only once the bus acknowledged the event: its sequence, or {error}
+							// (design 47 G §559). A refusal is an answer, so a writer says it as a finding's resend.
+							acker, ok := mesh.(interface {
+								PublishAcked(params json.RawMessage) (uint64, error)
+							})
+							switch {
+							case !ok:
+								result = map[string]any{"error": "this tool runner's bus publishes no event with an acknowledgement"}
+							default:
+								if seq, perr := acker.PublishAcked(m.Params); perr != nil {
+									result = map[string]any{"error": perr.Error()}
+								} else {
+									result = map[string]any{"sequence": seq}
+								}
+							}
 						case "mesh/ask":
 							var asked json.RawMessage
 							if asked, err = mesh.Ask(m.Params); err == nil {
@@ -404,6 +426,38 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an
 								result = json.RawMessage(raw)
 							}
 						case "mesh/subscribe":
+							var said struct {
+								Pattern string `json:"pattern"`
+								Deliver string `json:"deliver"`
+							}
+							_ = json.Unmarshal(m.Params, &said)
+							if said.Deliver == "new" {
+								// An ordered consumer of its own from the next message on (design 47 G §559, A-153): each message
+								// is handed to this child as an event is, and the consumer ends with the child.
+								newer, ok := mesh.(interface {
+									SubscribeNew(pattern string, deliver func(json.RawMessage) error) (func(), error)
+								})
+								if !ok {
+									err = errors.New("this tool runner's bus makes no subscription for new messages")
+									break
+								}
+								var stop func()
+								stop, err = newer.SubscribeNew(said.Pattern, func(envelope json.RawMessage) error {
+									_, err := c.ask(module, "mesh/event", map[string]any{"envelope": envelope}, EventTimeout)
+									return err
+								})
+								if err == nil {
+									c.mu.Lock()
+									if c.why != nil {
+										c.mu.Unlock()
+										stop()
+									} else {
+										c.watches = append(c.watches, stop)
+										c.mu.Unlock()
+									}
+								}
+								break
+							}
 							mu.Lock()
 							subscriber = c
 							mu.Unlock()

node-tools/internal/launch/sessions_test.go

@@ -2,6 +2,7 @@ package launch
 
 import (
 	"encoding/json"
+	"errors"
 	"os"
 	"path/filepath"
 	"strings"
@@ -66,3 +67,77 @@ func TestABundleReadsTheCapabilityWordsRunsAPutSessionAndItsChannelsClosingDrops
 		t.Fatalf("the runner took %v", b.chunks)
 	}
 }
+
+// findingBundle asks, as mesh-sdk's finding writer does, a publish with ack that the bus takes, one it refuses, and a
+// subscription for new messages; each answer it gets is written to $FINDING_OUT, and it ends after the subscription's.
+const findingBundle = `#!/bin/sh
+while IFS= read -r line; do
+  id=$(printf '%s' "$line" | sed -n 's/.*"id":\([0-9]*\)[,}].*/\1/p')
+  case "$line" in
+    *'"method":"initialize"'*) printf '{"jsonrpc":"2.0","id":%s,"result":{}}\n' "$id" ;;
+    *'"method":"tools/list"'*) printf '{"jsonrpc":"2.0","id":%s,"result":{"tools":[]}}\n' "$id"
+       printf '{"jsonrpc":"2.0","id":"a1","method":"mesh/publish","params":{"key":"taken","ack":true,"body":{}}}\n' ;;
+    *'"id":"a1"'*) printf '%s\n' "$line" >>"$FINDING_OUT"
+       printf '{"jsonrpc":"2.0","id":"a2","method":"mesh/publish","params":{"key":"refused","ack":true,"body":{}}}\n' ;;
+    *'"id":"a2"'*) printf '%s\n' "$line" >>"$FINDING_OUT"
+       printf '{"jsonrpc":"2.0","id":"s1","method":"mesh/subscribe","params":{"pattern":"mesh-conditions.findings-asked","deliver":"new"}}\n' ;;
+    *'"id":"s1"'*) printf '%s\n' "$line" >>"$FINDING_OUT"; exit 0 ;;
+  esac
+done
+`
+
+type findingBus struct {
+	subscribing
+	mu      sync.Mutex
+	pattern string
+	stopped chan struct{}
+}
+
+func (b *findingBus) PublishAcked(params json.RawMessage) (uint64, error) {
+	if strings.Contains(string(params), `"refused"`) {
+		return 0, errors.New("the bus refused it")
+	}
+	return 42, nil
+}
+
+func (b *findingBus) SubscribeNew(pattern string, _ func(json.RawMessage) error) (func(), error) {
+	b.mu.Lock()
+	b.pattern = pattern
+	b.mu.Unlock()
+	return func() { close(b.stopped) }, nil
+}
+
+// Design 47 G §559: `mesh/publish` with `ack` is answered with its sequence once the bus acknowledged it, or `{error}`;
+// `mesh/subscribe` with `deliver: new` makes a subscription for new messages that ends with the bundle's channel.
+func TestAPublishWithAckIsAnsweredWithItsSequenceAndASubscriptionForNewMessagesEndsWithTheChannel(t *testing.T) {
+	dir := t.TempDir()
+	entry, out := filepath.Join(dir, "bundle"), filepath.Join(dir, "out")
+	if err := os.WriteFile(entry, []byte(findingBundle), 0o755); err != nil {
+		t.Fatal(err)
+	}
+	b := &findingBus{stopped: make(chan struct{})}
+	l, err := Start("writer", entry, append(os.Environ(), "FINDING_OUT="+out), b, func(string, ...any) {})
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(l.Stop)
+	select {
+	case <-b.stopped:
+	case <-time.After(5 * time.Second):
+		t.Fatal("the subscription for new messages was not ended with the bundle's channel")
+	}
+	got, _ := os.ReadFile(out)
+	for _, want := range []string{`"sequence":42`, `"error":"the bus refused it"`} {
+		if !strings.Contains(string(got), want) {
+			t.Errorf("the bundle was answered %s, want it to hold %s", got, want)
+		}
+	}
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	if b.pattern != "mesh-conditions.findings-asked" {
+		t.Errorf("the subscription was made for %q", b.pattern)
+	}
+	if len(b.published) != 0 {
+		t.Errorf("a publish with ack went the unacknowledged way: %v", b.published)
+	}
+}

node-tools/internal/runtime/announce_test.go

@@ -142,7 +142,7 @@ func TestTheRunnerStatesTheCapabilitiesOfItsCallsInItsPing(t *testing.T) {
 		t.Fatalf("ping of the instance: %v", err)
 	}
 	var ping micro.Ping
-	if err := json.Unmarshal(msg.Data, &ping); err != nil || ping.Metadata[announce.Capabilities] != "bus-state-calls,ledger-calls,ledger-upkeep-calls,put-sessions,seat-settle-delay,sent-objects-calls" {
+	if err := json.Unmarshal(msg.Data, &ping); err != nil || ping.Metadata[announce.Capabilities] != "bus-state-calls,finding-calls,ledger-calls,ledger-upkeep-calls,put-sessions,seat-settle-delay,sent-objects-calls" {
 		t.Errorf("the runner states %q (%v), want its bus-state and ledger calls", ping.Metadata[announce.Capabilities], err)
 	}
 }

node-tools/internal/runtime/busstate.go

@@ -40,9 +40,10 @@ var callFamilies = map[string][]string{
 }
 
 // CallCapabilities is every capability of the calls this runner answers, sorted. The delayed settle of a pulled seat
-// message names a fate, not a method, so it stands beside the method families (novox/hq ADR 0664 §5).
+// message names a fate, not a method, so it stands beside the method families (novox/hq ADR 0664 §5); so does `finding-calls`,
+// the acknowledged `mesh/publish` and the `mesh/subscribe` for new messages the launch answers (design 47 G §559).
 func CallCapabilities() []string {
-	out := []string{stdio.CapSeatSettleDelay}
+	out := []string{stdio.CapSeatSettleDelay, stdio.CapFindingCalls}
 	for c := range callFamilies {
 		out = append(out, c)
 	}

node-tools/internal/runtime/ledgers_test.go

@@ -204,7 +204,7 @@ func TestABundleToolNamedAsALedgerRepairIsRefused(t *testing.T) {
 // no other name; none answers "the runtime answers no ...".
 func TestEveryCallOfAStatedCapabilityIsAnswered(t *testing.T) {
 	m, _, _ := aLedger(t)
-	if got := strings.Join(CallCapabilities(), ","); got != "bus-state-calls,ledger-calls,ledger-upkeep-calls,put-sessions,seat-settle-delay,sent-objects-calls" {
+	if got := strings.Join(CallCapabilities(), ","); got != "bus-state-calls,finding-calls,ledger-calls,ledger-upkeep-calls,put-sessions,seat-settle-delay,sent-objects-calls" {
 		t.Fatalf("the runner states %s", got)
 	}
 	for capability, methods := range callFamilies {

node-tools/internal/runtime/runtime.go

@@ -876,6 +876,21 @@ func (b *moduleBus) Publish(params json.RawMessage) error {
 	return b.all.conn.PublishAs(b.module, env)
 }
 
+// PublishAcked is Publish that answers the sequence the bus acknowledged the event at (`mesh/publish` with `ack`, design 47 G §559).
+func (b *moduleBus) PublishAcked(params json.RawMessage) (uint64, error) {
+	var env bus.Envelope
+	if err := json.Unmarshal(params, &env); err != nil {
+		return 0, fmt.Errorf("not an event envelope: %w", err)
+	}
+	return b.all.conn.PublishAsAcked(b.module, env)
+}
+
+// SubscribeNew hands the bundle one event's messages published from now on, on an ordered consumer of its own (`mesh/subscribe`
+// with `deliver: new`, design 47 G §559); the consumer ends with the bundle's channel.
+func (b *moduleBus) SubscribeNew(pattern string, deliver func(json.RawMessage) error) (func(), error) {
+	return b.all.conn.SubscribeNew(pattern, deliver)
+}
+
 // Ask calls a tool as the module: `{key, body}`, answered with the tool's result (ADR 0198).
 func (b *moduleBus) Ask(params json.RawMessage) (json.RawMessage, error) {
 	var asked struct {

node-tools/internal/runtime/subscribenew_integration_test.go

@@ -0,0 +1,84 @@
+//go:build integration
+
+package runtime
+
+import (
+	"context"
+	"encoding/json"
+	"strings"
+	"sync"
+	"testing"
+
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+
+	"github.com/novox/mesh-tools/node-tools/internal/bus"
+	mt "github.com/novox/mesh-tools/node-tools/internal/meshtest"
+)
+
+// Design 47 G §559, A-153: against the real bus a publish with ack answers the sequence the bus gave it, and a subscription
+// for new messages hears only what is published after it was made, on the subject mesh-sdk's one rule names for the event;
+// a wildcard is refused.
+func TestAnAckedPublishAnswersItsSequenceAndANewSubscriptionHearsOnlyWhatComesAfter(t *testing.T) {
+	mt.New(t)
+	nc, err := nats.Connect(mt.URL(t))
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(nc.Close)
+	js, _ := jetstream.New(nc)
+	ctx := context.Background()
+	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "SEATEVENTS", Subjects: []string{"mesh.seat.mesh-conditions.event.>"}}); err != nil {
+		t.Fatal(err)
+	}
+	conn := connect(t, "node-tools", "anchor")
+	publish := func(body string) {
+		t.Helper()
+		if _, err := js.Publish(ctx, "mesh.seat.mesh-conditions.event.findings-asked", []byte(body)); err != nil {
+			t.Fatal(err)
+		}
+	}
+	publish(`{"before":true}`)
+
+	var mu sync.Mutex
+	var heard []string
+	stop, err := conn.SubscribeNew("mesh-conditions.findings-asked", func(raw json.RawMessage) error {
+		var env bus.Envelope
+		_ = json.Unmarshal(raw, &env)
+		mu.Lock()
+		heard = append(heard, env.Key+" "+string(env.Body))
+		mu.Unlock()
+		return nil
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer stop()
+	publish(`{"after":true}`)
+	publish(`{"after":2}`)
+	mt.Until(t, func() error {
+		mu.Lock()
+		defer mu.Unlock()
+		if len(heard) < 2 {
+			return context.DeadlineExceeded
+		}
+		return nil
+	})
+	mu.Lock()
+	if len(heard) != 2 || heard[0] != `mesh-conditions.findings-asked {"after":true}` {
+		t.Errorf("heard %v, want the messages after the subscription alone, in order", heard)
+	}
+	mu.Unlock()
+
+	for name, want := range map[string]string{"mesh-conditions.*": "[A-Za-z0-9_-]"} {
+		if _, err := conn.SubscribeNew(name, func(json.RawMessage) error { return nil }); err == nil || !strings.Contains(err.Error(), want) {
+			t.Errorf("a subscription for %s answered %v, want one naming %q", name, err, want)
+		}
+	}
+
+	first, err := conn.PublishAsAcked("node-tools", bus.Envelope{Key: "finding-written", Body: json.RawMessage(`{}`)})
+	second, err2 := conn.PublishAsAcked("node-tools", bus.Envelope{Key: "finding-written", Body: json.RawMessage(`{}`)})
+	if err != nil || err2 != nil || first == 0 || second != first+1 {
+		t.Errorf("acknowledged publishes answered %d (%v) and %d (%v), want consecutive sequences", first, err, second, err2)
+	}
+}

node-tools/module.json

@@ -65,6 +65,7 @@
         "binary": "node-tools",
         "states": [
           "bus-state-calls",
+          "finding-calls",
           "ledger-calls",
           "ledger-upkeep-calls",
           "put-sessions",