Let a process call a tool or a seat verb on its own account, with the tool runner's caller fallback (issue 1191)

2026-10-19 04:41:08 +02:00
parent f73ec7457a
commit 80d3975762
3 changed files with 362 additions and 0 deletions
@@ -0,0 +1,84 @@
+package process
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/nats-io/nats.go"
+)
+
+// CallTimeout bounds a call to a tool or a seat verb (mesh-tools node-tools/internal/bus/bus.go:35,
+// RequestTimeout): what a module already expects.
+var CallTimeout = 30 * time.Second
+
+// Call invokes a tool or a seat verb as this process's own account (novox/hq ADR 0436 clause 4), on the
+// caller-named subject, falling back to the subject that names no caller when nothing answers it —
+// exactly the provisioner client's own pattern (mesh-catalog modules/mesh-provisioner-client/cmd/
+// mesh-provisioner-client/secrets.go:59-61, 93-97, 119, 125-133). key is "seat:<seat>.<verb>" for a seat's
+// verb, or "<module>.<tool>" for a module's own tool, as a process's bus.invokes names it. The reply is
+// {result, error, node} (mesh-tools node-tools/internal/bus/bus.go answerOf); a non-empty error is the
+// call's failure. It offers no way to answer a call — a process reaches the bus only as itself, and
+// never serves anything.
+func (c *Conn) Call(key string, body any) (json.RawMessage, error) {
+	called, fallback, err := callSubjects(key, c.callerToken())
+	if err != nil {
+		return nil, err
+	}
+	data, err := json.Marshal(body)
+	if err != nil {
+		return nil, err
+	}
+	result, err := c.request(called, data)
+	if errors.Is(err, nats.ErrNoResponders) {
+		result, err = c.request(fallback, data)
+	}
+	return result, err
+}
+
+// callerToken is this account's user name as the last token of a call it publishes, each dot written ~
+// (mesh-tools node-tools/internal/bus/caller.go:43, CallerToken).
+func (c *Conn) callerToken() string { return strings.ReplaceAll(c.account.User, ".", "~") }
+
+// callSubjects is the caller-named subject key is called on, and its plain fallback (mesh-tools
+// node-tools/internal/bus/caller.go CalledSubject, toolShape, calledPattern).
+func callSubjects(key, token string) (called, fallback string, err error) {
+	if rest, ok := strings.CutPrefix(key, "seat:"); ok {
+		seat, verb, ok := strings.Cut(rest, ".")
+		if !ok || seat == "" || verb == "" {
+			return "", "", fmt.Errorf("%q names no seat and verb (seat:<seat>.<verb>)", key)
+		}
+		return "mesh.seat." + seat + ".call." + verb + "." + token, "mesh.seat." + seat + ".tool." + verb, nil
+	}
+	module, tool, ok := strings.Cut(key, ".")
+	if !ok || module == "" || tool == "" {
+		return "", "", fmt.Errorf("%q is not <module>.<tool>, and does not start seat:<seat>.<verb>", key)
+	}
+	return "mesh.mod." + module + ".call." + tool + "." + token, "mesh.mod." + module + ".tool." + tool, nil
+}
+
+// request asks one subject once and reads its reply ({result, error, node}); a reply naming an error is
+// the call's failure, not a transport one.
+func (c *Conn) request(subject string, data []byte) (json.RawMessage, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), CallTimeout)
+	defer cancel()
+	msg, err := c.nc.RequestWithContext(ctx, subject, data)
+	if err != nil {
+		return nil, err
+	}
+	var r struct {
+		Result json.RawMessage `json:"result"`
+		Error  string          `json:"error"`
+		Node   string          `json:"node"`
+	}
+	if err := json.Unmarshal(msg.Data, &r); err != nil {
+		return nil, fmt.Errorf("%s answered something that is not a seat's answer: %w", subject, err)
+	}
+	if r.Error != "" {
+		return nil, errors.New(r.Error)
+	}
+	return r.Result, nil
+}
@@ -0,0 +1,237 @@
+//go:build integration
+
+package process
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+)
+
+// 1191/AC1: given a stand-in holder answering seat operator-channel verb tell on a test bus, when the
+// process connection of novox.mesh-mail.send calls seat:operator-channel.tell with {headline, lines,
+// subject}, then the request goes to mesh.seat.operator-channel.call.tell.novox~mesh-mail~send with the
+// arguments as the body, and the holder's {result} comes back; a reply with error set comes back as an
+// error with those words.
+func TestCallReachesTheCallerNamedSubjectAndReadsTheReply(t *testing.T) {
+	url, account := grantedSendProcess(t)
+	adminNC, _ := adminJS(t, url)
+
+	heard := make(chan *nats.Msg, 1)
+	sub, err := adminNC.Subscribe("mesh.seat.operator-channel.call.tell.novox~mesh-mail~send", func(m *nats.Msg) {
+		heard <- m
+		_ = m.Respond([]byte(`{"result":{"sent":true}}`))
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = sub.Unsubscribe() }()
+	if err := adminNC.Flush(); err != nil {
+		t.Fatal(err)
+	}
+
+	conn, err := Connect(account)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+
+	result, err := conn.Call("seat:operator-channel.tell",
+		map[string]any{"headline": "h", "lines": []string{"l1"}, "subject": "s"})
+	if err != nil {
+		t.Fatalf("Call: %v", err)
+	}
+	if string(result) != `{"sent":true}` {
+		t.Fatalf("result: %s", result)
+	}
+
+	select {
+	case msg := <-heard:
+		var body map[string]any
+		if err := json.Unmarshal(msg.Data, &body); err != nil {
+			t.Fatal(err)
+		}
+		if body["headline"] != "h" || body["subject"] != "s" {
+			t.Errorf("the arguments were not the call's body: %v", body)
+		}
+	case <-time.After(3 * time.Second):
+		t.Fatal("the holder never heard the call")
+	}
+}
+
+// 1191/AC1: a reply with error set comes back as an error with those words.
+func TestCallReadsAnErrorReplyAsAnError(t *testing.T) {
+	url, account := grantedSendProcess(t)
+	adminNC, _ := adminJS(t, url)
+	sub, err := adminNC.Subscribe("mesh.seat.operator-channel.call.tell.novox~mesh-mail~send", func(m *nats.Msg) {
+		_ = m.Respond([]byte(`{"error":"refused: no such warrant"}`))
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = sub.Unsubscribe() }()
+	if err := adminNC.Flush(); err != nil {
+		t.Fatal(err)
+	}
+
+	conn, err := Connect(account)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+
+	_, err = conn.Call("seat:operator-channel.tell", map[string]any{})
+	if err == nil || err.Error() != "refused: no such warrant" {
+		t.Fatalf("Call with an error reply: %v, want the words of the error", err)
+	}
+}
+
+// 1191/AC2: given no responder on the caller-named subject, when the call is made, then it is asked
+// again on mesh.seat.operator-channel.tool.tell (the tool runner's fallback).
+func TestCallFallsBackToThePlainSubjectWhenNothingAnswersTheCallerNamedOne(t *testing.T) {
+	url, account := grantedSendProcess(t)
+	adminNC, _ := adminJS(t, url)
+	sub, err := adminNC.Subscribe("mesh.seat.operator-channel.tool.tell", func(m *nats.Msg) {
+		_ = m.Respond([]byte(`{"result":{"via":"fallback"}}`))
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = sub.Unsubscribe() }()
+	if err := adminNC.Flush(); err != nil {
+		t.Fatal(err)
+	}
+
+	conn, err := Connect(account)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+
+	result, err := conn.Call("seat:operator-channel.tell", map[string]any{})
+	if err != nil {
+		t.Fatalf("Call: %v", err)
+	}
+	if string(result) != `{"via":"fallback"}` {
+		t.Fatalf("result: %s, want the fallback holder's answer", result)
+	}
+}
+
+// 1191/AC4, release 2's acceptance test: given the test bus with accounts composed as the controller
+// composes a process account (the subjects of mesh-controller internal/broker/process_grants.go), when
+// one process connection emits, updates a state, reads an own ask's record, consumes and acknowledges its
+// queue and calls tell, then each succeeds; and its attempt to subscribe a tool-call subject is refused
+// by the bus.
+func TestReleaseAcceptanceAProcessDoesExactlyItsBusDeclaration(t *testing.T) {
+	url, account := grantedSendProcess(t)
+	_, adminJetStream := adminJS(t, url)
+
+	if _, err := adminJetStream.CreateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: "mesh-mail_sends"}); err != nil {
+		t.Fatalf("the admin could not make the state bucket: %v", err)
+	}
+	asksKV, err := adminJetStream.CreateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: "messenger_asks"})
+	if err != nil {
+		t.Fatalf("the admin could not make the records bucket: %v", err)
+	}
+	if _, err := asksKV.Put(ctxFor(t), "mesh-mail.ask-1", []byte(`{"state":"decided"}`)); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := adminJetStream.CreateStream(ctxFor(t), jetstream.StreamConfig{
+		Name: "MAIL_EVENTS", Subjects: []string{"mesh.mod.mesh-mail.event.>"}, Duplicates: time.Minute,
+	}); err != nil {
+		t.Fatalf("the admin could not make the events stream: %v", err)
+	}
+	makeQueue(t, url, account)
+	publishApproved(t, url, `{"send":"s1"}`)
+
+	adminNC, _ := adminJS(t, url)
+	holderSub, err := adminNC.Subscribe("mesh.seat.operator-channel.call.tell.novox~mesh-mail~send", func(m *nats.Msg) {
+		_ = m.Respond([]byte(`{"result":{}}`))
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = holderSub.Unsubscribe() }()
+	if err := adminNC.Flush(); err != nil {
+		t.Fatal(err)
+	}
+
+	conn, err := Connect(account)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+
+	// 1. Emits.
+	if err := conn.Emit("mail.send.sent", map[string]any{"send": "s1"}, stdio.EmitOptions{MessageID: "ev-1"}); err != nil {
+		t.Fatalf("Emit: %v", err)
+	}
+
+	// 2. Updates a state.
+	s := conn.State("sends")
+	rev, err := s.Create("k", map[string]any{"v": 1})
+	if err != nil {
+		t.Fatalf("State.Create: %v", err)
+	}
+	if _, err := s.Update("k", map[string]any{"v": 2}, rev); err != nil {
+		t.Fatalf("State.Update: %v", err)
+	}
+
+	// 3. Reads an own ask's record.
+	if e, err := conn.Ask("messenger_asks", "ask-1"); err != nil || e == nil {
+		t.Fatalf("Ask: %+v, %v", e, err)
+	}
+
+	// 4. Consumes and acknowledges its queue.
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	consumed := make(chan error, 1)
+	go func() {
+		consumed <- conn.Consume(ctx, "MAIL_SENDS", func(m *Message) {
+			if err := m.Ack(); err != nil {
+				t.Errorf("Ack: %v", err)
+			}
+			cancel()
+		})
+	}()
+	select {
+	case <-consumed:
+	case <-time.After(8 * time.Second):
+		t.Fatal("the process did not consume its own queue")
+	}
+
+	// 5. Calls tell.
+	if _, err := conn.Call("seat:operator-channel.tell", map[string]any{}); err != nil {
+		t.Fatalf("Call: %v", err)
+	}
+
+	// 6. Refused: subscribing a tool-call subject — a fresh connection, so the error handler catches
+	// exactly this one refusal and nothing earlier.
+	refusals := make(chan error, 1)
+	watched, err := Connect(account, nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) { refusals <- err }))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer watched.Close()
+	if _, err := watched.nc.Subscribe("mesh.mod.mesh-mail.tool.>", func(*nats.Msg) {}); err != nil {
+		t.Fatal(err)
+	}
+	if err := watched.nc.Flush(); err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case err := <-refusals:
+		if !errors.Is(err, nats.ErrPermissionViolation) {
+			t.Errorf("subscribing a tool-call subject was not refused as a permissions violation: %v", err)
+		}
+	case <-time.After(3 * time.Second):
+		t.Error("subscribing a tool-call subject the process does not serve was not refused")
+	}
+}
@@ -0,0 +1,41 @@
+package process
+
+import "testing"
+
+// 1191/AC3: given a module tool key <module>.<tool>, when called, then the subject is
+// mesh.mod.<module>.call.<tool>.<token> with the same fallback (mesh.mod.<module>.tool.<tool>).
+func TestCallSubjectsForAModuleTool(t *testing.T) {
+	called, fallback, err := callSubjects("mesh-mail.send", "novox~mesh-mail~send")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if called != "mesh.mod.mesh-mail.call.send.novox~mesh-mail~send" {
+		t.Errorf("called: %q", called)
+	}
+	if fallback != "mesh.mod.mesh-mail.tool.send" {
+		t.Errorf("fallback: %q", fallback)
+	}
+}
+
+// 1191/AC1's subject (the seat form): seat:<seat>.<verb> calls mesh.seat.<seat>.call.<verb>.<token>,
+// falling back to mesh.seat.<seat>.tool.<verb>.
+func TestCallSubjectsForASeatVerb(t *testing.T) {
+	called, fallback, err := callSubjects("seat:operator-channel.tell", "novox~mesh-mail~send")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if called != "mesh.seat.operator-channel.call.tell.novox~mesh-mail~send" {
+		t.Errorf("called: %q", called)
+	}
+	if fallback != "mesh.seat.operator-channel.tool.tell" {
+		t.Errorf("fallback: %q", fallback)
+	}
+}
+
+func TestCallSubjectsRefusesAKeyNamingNeither(t *testing.T) {
+	for _, key := range []string{"", "nodot", "seat:", "seat:noverb", "."} {
+		if _, _, err := callSubjects(key, "token"); err == nil {
+			t.Errorf("callSubjects(%q) was accepted", key)
+		}
+	}
+}