Test the router whole over the real bus: a held ask is offered when a channel can carry it (issue 926, ADR 0397)

modules/messenger/cmd/messenger/bus_test.go

@@ -0,0 +1,380 @@
+package main
+
+import (
+	"bufio"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/asks"
+	"github.com/nats-io/nats-server/v2/server"
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// The router started whole over a real bus (novox/hq ADR 0391, issue 926). A JetStream server is linked into the
+// test and started for it alone; the module is this test binary started again as the bundle, speaking MCP on its
+// standard streams to a runtime played here, which does what the tool runner does for it: its state is buckets
+// of the bus, an ask submitted on the seat's subject is work handed to the bundle, what the bundle publishes on a
+// seat is published on the bus, an event on the bus is handed to the bundle, and the controller's membership is
+// the kinds this test says. It runs in the forge's merge check, not on the way.
+
+const bundleVar = "MESH_MESSENGER_TEST_BUNDLE"
+
+func TestMain(m *testing.M) {
+	if os.Getenv(bundleVar) == "1" {
+		kindsEvery = 200 * time.Millisecond
+		main()
+		os.Exit(0)
+	}
+	os.Exit(m.Run())
+}
+
+type busRuntime struct {
+	t    *testing.T
+	nc   *nats.Conn
+	js   jetstream.JetStream
+	in   io.WriteCloser
+	mu   sync.Mutex // writes to the bundle, and the fields below
+	next int
+	wait map[string]chan json.RawMessage
+
+	kinds     []map[string]any
+	kindsRead int
+	subscribed,
+	taking bool
+	buckets map[string]jetstream.KeyValue
+}
+
+func newBusRuntime(t *testing.T) *busRuntime {
+	t.Helper()
+	s, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: server.RANDOM_PORT, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	go s.Start()
+	if !s.ReadyForConnections(30 * time.Second) {
+		t.Fatal("no bus within 30s")
+	}
+	t.Cleanup(func() { s.Shutdown(); s.WaitForShutdown() })
+	nc, err := nats.Connect(s.ClientURL())
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(nc.Close)
+	js, _ := jetstream.New(nc)
+	return &busRuntime{t: t, nc: nc, js: js, wait: map[string]chan json.RawMessage{}, buckets: map[string]jetstream.KeyValue{}}
+}
+
+func (r *busRuntime) bucket(name string) jetstream.KeyValue {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if kv, ok := r.buckets[name]; ok {
+		return kv
+	}
+	kv, err := r.js.CreateOrUpdateKeyValue(context.Background(), jetstream.KeyValueConfig{Bucket: name})
+	if err != nil {
+		r.t.Fatal(err)
+	}
+	r.buckets[name] = kv
+	return kv
+}
+
+func (r *busRuntime) setKinds(caps map[string][]string) {
+	var out []map[string]any
+	for kind, c := range caps {
+		out = append(out, map[string]any{"seat": asks.ChannelSeat, "kind": kind, "module": kind, "node": "test", "capabilities": c})
+	}
+	r.mu.Lock()
+	r.kinds = out
+	r.mu.Unlock()
+}
+
+func (r *busRuntime) say(m map[string]any) {
+	m["jsonrpc"] = "2.0"
+	raw, _ := json.Marshal(m)
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	_, _ = r.in.Write(append(raw, '\n'))
+}
+
+// request is a request to the bundle (work, an event, a tool call) and its answer.
+func (r *busRuntime) request(method string, params any) (json.RawMessage, error) {
+	r.mu.Lock()
+	r.next++
+	id := "runtime-" + strconv.Itoa(r.next)
+	ch := make(chan json.RawMessage, 1)
+	r.wait[id] = ch
+	r.mu.Unlock()
+	r.say(map[string]any{"id": id, "method": method, "params": params})
+	select {
+	case raw := <-ch:
+		return raw, nil
+	case <-time.After(60 * time.Second):
+		return nil, fmt.Errorf("%s was not answered in 60 s", method)
+	}
+}
+
+func (r *busRuntime) start(dir string) {
+	r.t.Helper()
+	settings := filepath.Join(dir, "settings.json")
+	_ = os.WriteFile(settings, []byte(`{}`), 0o600)
+	cmd := exec.Command(os.Args[0], "-test.run=^$")
+	cmd.Env = append(os.Environ(), bundleVar+"=1", "MESH_MESSENGER_SETTINGS="+settings, "MESH_MESSENGER_STATE="+dir,
+		"MESH_SERVED_MODULE=messenger", "MESH_MODULE=messenger", "MESH_NODE=test")
+	in, _ := cmd.StdinPipe()
+	out, _ := cmd.StdoutPipe()
+	cmd.Stderr = busLog{r.t}
+	if err := cmd.Start(); err != nil {
+		r.t.Fatal(err)
+	}
+	r.in = in
+	go r.read(out)
+	r.t.Cleanup(func() { _ = in.Close(); _ = cmd.Wait() })
+}
+
+type busLog struct{ t *testing.T }
+
+func (l busLog) Write(p []byte) (int, error) { l.t.Logf("bundle: %s", p); return len(p), nil }
+
+func (r *busRuntime) read(out io.Reader) {
+	sc := bufio.NewScanner(out)
+	sc.Buffer(make([]byte, 1<<20), 1<<24)
+	for sc.Scan() {
+		var m struct {
+			ID     json.RawMessage `json:"id"`
+			Method string          `json:"method"`
+			Params json.RawMessage `json:"params"`
+		}
+		if json.Unmarshal(sc.Bytes(), &m) != nil {
+			continue
+		}
+		if m.Method == "" {
+			var id string
+			_ = json.Unmarshal(m.ID, &id)
+			r.mu.Lock()
+			ch := r.wait[id]
+			delete(r.wait, id)
+			r.mu.Unlock()
+			if ch != nil {
+				ch <- append([]byte(nil), sc.Bytes()...)
+			}
+			continue
+		}
+		go r.runtime(m.ID, m.Method, m.Params)
+	}
+}
+
+// runtime is what the tool runner does for the bundle: only what the router uses.
+func (r *busRuntime) runtime(id json.RawMessage, method string, params json.RawMessage) {
+	var p struct {
+		State    string          `json:"state"`
+		Key      string          `json:"key"`
+		Value    json.RawMessage `json:"value"`
+		Revision uint64          `json:"revision"`
+		Body     json.RawMessage `json:"body"`
+		Subject  string          `json:"subject"`
+		Pattern  string          `json:"pattern"`
+		Worker   string          `json:"worker"`
+	}
+	_ = json.Unmarshal(params, &p)
+	ctx := context.Background()
+	const changed = -32010
+	code := -32000
+	result, err := func() (any, error) {
+		switch method {
+		case "mesh/state.get":
+			e, err := r.bucket(p.State).Get(ctx, p.Key)
+			if errors.Is(err, jetstream.ErrKeyNotFound) {
+				return nil, nil
+			}
+			if err != nil {
+				return nil, err
+			}
+			return map[string]any{"key": p.Key, "value": json.RawMessage(e.Value()), "revision": e.Revision()}, nil
+		case "mesh/state.put":
+			rev, err := r.bucket(p.State).Put(ctx, p.Key, p.Value)
+			return map[string]any{"revision": rev}, err
+		case "mesh/state.create":
+			rev, err := r.bucket(p.State).Create(ctx, p.Key, p.Value)
+			if errors.Is(err, jetstream.ErrKeyExists) {
+				code = changed
+			}
+			return map[string]any{"revision": rev}, err
+		case "mesh/state.update":
+			rev, err := r.bucket(p.State).Update(ctx, p.Key, p.Value, p.Revision)
+			if err != nil {
+				code = changed
+			}
+			return map[string]any{"revision": rev}, err
+		case "mesh/state.delete":
+			err := r.bucket(p.State).Delete(ctx, p.Key)
+			return map[string]any{}, err
+		case "mesh/state.keys":
+			keys, err := r.bucket(p.State).Keys(ctx)
+			if errors.Is(err, jetstream.ErrNoKeysFound) {
+				return []string{}, nil
+			}
+			return keys, err
+		case "mesh/subscribe":
+			if strings.HasPrefix(p.Pattern, asks.IntakeSeat) {
+				r.mu.Lock()
+				r.subscribed = true
+				r.mu.Unlock()
+			}
+			return map[string]any{}, nil
+		case "mesh/seat.take":
+			r.mu.Lock()
+			r.taking = true
+			r.mu.Unlock()
+			return map[string]any{}, nil
+		case "mesh/seat.answer", "mesh/publish":
+			return map[string]any{}, nil
+		case "mesh/seat.kinds":
+			r.mu.Lock()
+			defer r.mu.Unlock()
+			r.kindsRead++
+			return r.kinds, nil
+		case "mesh/seat.publish":
+			return map[string]any{}, r.nc.Publish(p.Subject, p.Body)
+		case "mesh/ask":
+			if p.Key == RootFreeVerb {
+				var a struct {
+					Machines []string `json:"machines"`
+				}
+				_ = json.Unmarshal(p.Body, &a)
+				var out []RootVerdict
+				for _, m := range a.Machines {
+					out = append(out, RootVerdict{Machine: m, Free: true})
+				}
+				return map[string]any{"machines": out}, nil
+			}
+			return nil, fmt.Errorf("the test runtime answers nothing for %s", p.Key)
+		}
+		return nil, fmt.Errorf("the runtime offers no %s", method)
+	}()
+	if err != nil {
+		r.say(map[string]any{"id": id, "error": map[string]any{"code": code, "message": err.Error()}})
+		return
+	}
+	r.say(map[string]any{"id": id, "result": result})
+}
+
+// forward hands the bundle what is published on the bus that it takes: an ask submitted to the seat as work,
+// a channel's word as an event.
+func (r *busRuntime) forward() {
+	if _, err := r.nc.Subscribe("mesh.seat."+asks.Seat+".accept.ask.*", func(m *nats.Msg) {
+		_, _ = r.request("mesh/work", map[string]any{"work": map[string]any{"worker": AsksWorker, "subject": m.Subject, "body": json.RawMessage(m.Data)}})
+	}); err != nil {
+		r.t.Fatal(err)
+	}
+	if _, err := r.nc.Subscribe("mesh.seat."+asks.IntakeSeat+".event.>", func(m *nats.Msg) {
+		key := strings.Replace(strings.TrimPrefix(m.Subject, "mesh.seat."), ".event", "", 1)
+		_, _ = r.request("mesh/event", map[string]any{"envelope": map[string]any{"key": key, "node": "test", "body": json.RawMessage(m.Data)}})
+	}); err != nil {
+		r.t.Fatal(err)
+	}
+}
+
+func (r *busRuntime) until(what string, within time.Duration, ok func() bool) {
+	r.t.Helper()
+	for deadline := time.Now().Add(within); time.Now().Before(deadline); time.Sleep(100 * time.Millisecond) {
+		if ok() {
+			return
+		}
+	}
+	r.t.Fatalf("not within %s: %s", within, what)
+}
+
+func (r *busRuntime) askRecord(key string) *AskRecord {
+	e, err := r.bucket("asks").Get(context.Background(), key)
+	if err != nil {
+		return nil
+	}
+	var rec AskRecord
+	_ = json.Unmarshal(e.Value(), &rec)
+	return &rec
+}
+
+// A held ask over the real bus (ADR 0397, issue 926): taken while the controller's word on Telegram lacks a
+// verified sender, it is held with the reason and its asker is told nothing of an end; once the word gives it a
+// verified sender, it is offered on Telegram.
+func TestAnApprovalAskIsHeldThenOfferedWhenAChannelCanCarryItOverTheBus(t *testing.T) {
+	r := newBusRuntime(t)
+	away := []string{"deliver", "reaches-away", "choice", "edit", "exact-render"}
+	r.setKinds(map[string][]string{"telegram": away})
+	now := time.Now().UTC()
+	// The operator's account, linked long enough ago to approve.
+	id := Identity{Kind: "telegram", Identity: "42", Display: "Jo", Linked: now.Add(-3 * time.Hour), ApproveFrom: now.Add(-2 * time.Hour)}
+	raw, _ := json.Marshal(id)
+	if _, err := r.bucket("identities").Put(context.Background(), identityKey("telegram", "42"), raw); err != nil {
+		t.Fatal(err)
+	}
+	var told, shown []json.RawMessage // shown is kept to say nothing was shown while held
+	var seen sync.Mutex
+	collect := func(into *[]json.RawMessage) nats.MsgHandler {
+		return func(m *nats.Msg) { seen.Lock(); *into = append(*into, append([]byte(nil), m.Data...)); seen.Unlock() }
+	}
+	if _, err := r.nc.Subscribe(asks.DecidedSubject("mesh-delivery"), collect(&told)); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := r.nc.Subscribe(asks.ChannelSubject(asks.Show, "telegram"), collect(&shown)); err != nil {
+		t.Fatal(err)
+	}
+	r.forward()
+	r.start(t.TempDir())
+	r.until("the bundle subscribed, takes asks and read the kinds", 20*time.Second, func() bool {
+		r.mu.Lock()
+		defer r.mu.Unlock()
+		return r.subscribed && r.taking && r.kindsRead > 0
+	})
+	standing, _ := json.Marshal(Standing{Ready: true, At: time.Now().UTC()})
+	if err := r.nc.Publish(asks.IntakeSubject(asks.StandingWhat, "telegram"), standing); err != nil {
+		t.Fatal(err)
+	}
+	time.Sleep(time.Second)
+	c := &clock{t: time.Now().UTC()}
+	submitted, _ := json.Marshal(heldAsk(c, "d-1"))
+	if err := r.nc.Publish(asks.AskSubject("mesh-delivery"), submitted); err != nil {
+		t.Fatal(err)
+	}
+	r.until("the ask is recorded held with the reason", 20*time.Second, func() bool {
+		rec := r.askRecord("mesh-delivery.d-1")
+		return rec != nil && rec.State == StateHeld && strings.Contains(rec.HeldWhy, "Telegram cannot prove who answered")
+	})
+	rec := r.askRecord("mesh-delivery.d-1")
+	if len(rec.Tickets) != 0 {
+		t.Errorf("a held ask has tickets: %+v", rec)
+	}
+	seen.Lock()
+	if len(told) != 0 || len(shown) != 0 {
+		t.Errorf("a held ask was refused or shown: told %s, shown %s", told, shown)
+	}
+	seen.Unlock()
+	// The asks verb shows it held, since when, and why.
+	out, err := r.request("tools/call", map[string]any{"name": "operator-channel.asks", "arguments": map[string]any{}})
+	if err != nil || !strings.Contains(string(out), `held_why`) || !strings.Contains(string(out), `held`) {
+		t.Errorf("the asks verb: %s %v", out, err)
+	}
+	// The controller's word now gives Telegram a verified sender: the ask is offered there.
+	r.setKinds(map[string][]string{"telegram": append(append([]string{}, away...), "verified-sender")})
+	r.until("the ask is offered", 30*time.Second, func() bool {
+		rec := r.askRecord("mesh-delivery.d-1")
+		return rec != nil && rec.State == StateOpen && len(rec.Tickets) == 3 && rec.HeldWhy == ""
+	})
+	// Its message is the router's ordinary one for a new ask, said after the burst window (BurstWindow), which a
+	// unit test covers with a clock; this test waits for no half minute.
+	if rec := r.askRecord("mesh-delivery.d-1"); rec.Tickets == nil {
+		t.Errorf("offered with no tickets: %+v", rec)
+	}
+}

modules/messenger/cmd/messenger/main.go

@@ -543,6 +543,9 @@ func answerProofs(h *Holder) {
 	}
 }
 
+// kindsEvery is how often the controller's word on the kinds is read again (a test bundle reads it faster).
+var kindsEvery = 30 * time.Second
+
 // followKinds reads, every half minute, which channel kinds the controller says exist and what each
 // promises: never from a channel.
 func followKinds(h *Holder) {
@@ -565,7 +568,7 @@ func followKinds(h *Holder) {
 			h.ReleaseHeld() // a held ask is judged again when the kinds change (ADR 0397 §2)
 			said = ""
 		}
-		time.Sleep(30 * time.Second)
+		time.Sleep(kindsEvery)
 	}
 }
 

modules/messenger/go.mod

@@ -1,5 +1,22 @@
 module messenger
 
-go 1.22
+go 1.25.0
 
-require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261013182924-5933eb757d44
+require (
+	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261013182924-5933eb757d44
+	github.com/nats-io/nats-server/v2 v2.11.17
+	github.com/nats-io/nats.go v1.51.0
+)
+
+require (
+	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.18.5 // indirect
+	github.com/minio/highwayhash v1.0.4 // indirect
+	github.com/nats-io/jwt/v2 v2.8.1 // indirect
+	github.com/nats-io/nkeys v0.4.15 // indirect
+	github.com/nats-io/nuid v1.0.1 // indirect
+	golang.org/x/crypto v0.50.0 // indirect
+	golang.org/x/sys v0.43.0 // indirect
+	golang.org/x/time v0.15.0 // indirect
+)

modules/messenger/go.sum

@@ -1,8 +1,27 @@
-git.novox.be/novox/mesh-sdk/go v0.1.13 h1:Su4JYpZ+zhNovkGA2DgxiZdcuHpjw2tPJzc/7PljhXg=
-git.novox.be/novox/mesh-sdk/go v0.1.13/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261013141520-9588871d6629 h1:7TYFV1Q6Mcz+lsLP/sds87tysET4rKCn3rRIQV8gNmA=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261013141520-9588871d6629/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261013145622-f4745447ba6d h1:HEJyqorKktrkJPeeJdwEo0kdrJjH3fm/YVS9MGftCqY=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261013145622-f4745447ba6d/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY=
 git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261013182924-5933eb757d44 h1:r5xIo5qWOXQdCSMxvqJy8kV4U+P/QWXFK6971Ha6wU8=
 git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261013182924-5933eb757d44/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY=
+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=
+github.com/google/go-tpm v0.9.8/go.mod h1:h9jEsEECg7gtLis0upRBQU+GhYVH6jMjrFxI8u6bVUY=
+github.com/klauspost/compress v1.18.5 h1:/h1gH5Ce+VWNLSWqPzOVn6XBO+vJbCNGvjoaGBFW2IE=
+github.com/klauspost/compress v1.18.5/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ=
+github.com/minio/highwayhash v1.0.4 h1:asJizugGgchQod2ja9NJlGOWq4s7KsAWr5XUc9Clgl4=
+github.com/minio/highwayhash v1.0.4/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ=
+github.com/nats-io/jwt/v2 v2.8.1 h1:V0xpGuD/N8Mi+fQNDynXohVvp7ZztevW5io8CUWlPmU=
+github.com/nats-io/jwt/v2 v2.8.1/go.mod h1:nWnOEEiVMiKHQpnAy4eXlizVEtSfzacZ1Q43LIRavZg=
+github.com/nats-io/nats-server/v2 v2.11.17 h1:GKEghcFK6A+aFx11Yf1LjgLC3txAwvyhnYzhBIQZA8I=
+github.com/nats-io/nats-server/v2 v2.11.17/go.mod h1:B1sFVz4StNosQ903ak4N1G01Fl/9f8e06mXpFIE2K24=
+github.com/nats-io/nats.go v1.51.0 h1:ByW84XTz6W03GSSsygsZcA+xgKK8vPGaa/FCAAEHnAI=
+github.com/nats-io/nats.go v1.51.0/go.mod h1:26HypzazeOkyO3/mqd1zZd53STJN0EjCYF9Uy2ZOBno=
+github.com/nats-io/nkeys v0.4.15 h1:JACV5jRVO9V856KOapQ7x+EY8Jo3qw1vJt/9Jpwzkk4=
+github.com/nats-io/nkeys v0.4.15/go.mod h1:CpMchTXC9fxA5zrMo4KpySxNjiDVvr8ANOSZdiNfUrs=
+github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
+github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
+golang.org/x/crypto v0.50.0 h1:zO47/JPrL6vsNkINmLoo/PH1gcxpls50DNogFvB5ZGI=
+golang.org/x/crypto v0.50.0/go.mod h1:3muZ7vA7PBCE6xgPX7nkzzjiUq87kRItoJQM1Yo8S+Q=
+golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/sys v0.43.0 h1:Rlag2XtaFTxp19wS8MXlJwTvoh8ArU6ezoyFsMyCTNI=
+golang.org/x/sys v0.43.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
+golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=
+golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno=