mesh-catalog/modules/claude-licence-manager/cmd/manager/whole_integration_test.go

//go:build integration

package main

// The manager whole against a real bus (design 39 bus-and-switch §9, §23, §26, §28, §29, §46, B41): the
// built binary, played over its real stdin and stdout as the node's tool runtime plays it (tools/call with the caller in
// _meta; mesh/state.* from a real nats-server's key-value buckets, mesh/log.* from the logs' real streams, mesh/publish
// and mesh/seat.publish onto their subjects). What the manager asks of
// other parts is played beside it: the vault's licence-put and get, a node's claude_code_grant and its holdings-changed,
// the operator channel's tell, and the vendor's token and usage endpoints. A condition is taken off the operator
// channel's subject and decoded by its client, the contract the router reads (B41); the router itself is not run.

import (
	"bufio"
	"context"
	"crypto/rand"
	"encoding/base64"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/http/httptest"
	"os"
	"os/exec"
	"path/filepath"
	"sort"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"
	"golang.org/x/crypto/nacl/box"

	occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient"
	oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
	"git.novox.be/novox/mesh-sdk/go/testbus"

	. "claude-licence-manager/internal/licences"
)

const (
	licenceA = "first@example.org"
	licenceB = "second@example.org"
)

// consumers are the four nodes bound to licenceA.
var consumers = []string{"anchor", "home-server", "laptop", "workstation"}

// seed keeps what a running manager kept before this start: two subscriptions, their grants and access tokens in the
// vault, and four consumers bound to the first.
func (m *mesh) seed() {
	m.t.Helper()
	now := time.Now()
	far := now.Add(30 * 24 * time.Hour).UnixMilli()
	licences, err := m.js.KeyValue(m.ctx, Publisher+"_licences")
	if err != nil {
		m.t.Fatal(err)
	}
	bindings, err := m.js.KeyValue(m.ctx, Publisher+"_bindings")
	if err != nil {
		m.t.Fatal(err)
	}
	for i, l := range []struct{ name, account, refresh, access string }{
		{licenceA, "acct-a", "rt-a", "at-a"}, {licenceB, "acct-b", "rt-b", "at-b"}} {
		grant, _ := json.Marshal(FullGrant{AccessToken: l.access, RefreshToken: l.refresh, ExpiresAt: far})
		m.secrets[SecretName(KeyFor(l.name), PartGrant)] = string(grant)
		m.secrets[SecretName(KeyFor(l.name), PartAccess)] = l.access
		v, _ := json.Marshal(Licence{Name: l.name, Kind: "subscription", AccountUUID: l.account, Email: l.name,
			RefreshFingerprint: Fingerprint(l.refresh), AccessExpiresAt: far, RefreshExpiresAt: far,
			AdoptedAt: int64(i + 1), RotatedAt: now.UnixMilli(), Generation: 1})
		if _, err := licences.Put(m.ctx, KeyFor(l.name), v); err != nil {
			m.t.Fatal(err)
		}
	}
	for i, c := range consumers {
		v, _ := json.Marshal(BindingState{Licence: licenceA, Kind: "subscription", Generation: int64(1013 + i),
			Since: now.UTC().Format(time.RFC3339), Why: "bound", By: Itself})
		if _, err := bindings.Put(m.ctx, c, v); err != nil {
			m.t.Fatal(err)
		}
	}
}

// ---- the vendor ----------------------------------------------------------------------------------------------

// vendor answers the token endpoint by refresh token and the usage endpoint by access token, and counts its calls.
type vendor struct {
	mu     sync.Mutex
	calls  int
	logins map[string][2]string // refresh token: the next access token and the account
	usage  map[string]float64   // access token: the weekly percentage; the session window is answered empty (C4)
	server *httptest.Server
}

func newVendor(t *testing.T) *vendor {
	v := &vendor{logins: map[string][2]string{}, usage: map[string]float64{}}
	v.server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		v.mu.Lock()
		defer v.mu.Unlock()
		v.calls++
		if r.URL.Path == "/token" {
			_ = r.ParseForm()
			l, ok := v.logins[r.PostForm.Get("refresh_token")]
			if !ok {
				http.Error(w, `{"error":"invalid_grant"}`, http.StatusBadRequest)
				return
			}
			in := int64(8 * 3600)
			_ = json.NewEncoder(w).Encode(map[string]any{"access_token": l[0], "expires_in": in,
				"account": map[string]string{"uuid": l[1]}})
			return
		}
		u, ok := v.usage[strings.TrimPrefix(r.Header.Get("authorization"), "Bearer ")]
		if !ok {
			http.Error(w, `{"error":"unknown token"}`, http.StatusUnauthorized)
			return
		}
		resets := time.Now().Add(3 * time.Hour).UTC().Format(time.RFC3339)
		// As the vendor answers since it retired the session limit: the session window with no utilization (C4).
		_ = json.NewEncoder(w).Encode(map[string]any{"five_hour": map[string]any{"utilization": nil, "resets_at": nil},
			"seven_day": map[string]any{"utilization": u, "resets_at": resets}})
	}))
	t.Cleanup(v.server.Close)
	return v
}

func (v *vendor) set(access string, weekly float64) {
	v.mu.Lock()
	v.usage[access] = weekly
	v.mu.Unlock()
}

func (v *vendor) called() int {
	v.mu.Lock()
	defer v.mu.Unlock()
	return v.calls
}

// ---- the runtime -----------------------------------------------------------------------------------------------

type wire struct {
	ID     json.RawMessage `json:"id,omitempty"`
	Method string          `json:"method,omitempty"`
	Params json.RawMessage `json:"params,omitempty"`
	Result json.RawMessage `json:"result,omitempty"`
	Error  *struct {
		Message string `json:"message"`
	} `json:"error,omitempty"`
}

// said is one event the manager said, or one call it asked of another part.
type said struct {
	key  string
	body json.RawMessage
}

// mesh is the bus, the state directory and the parts beside the manager, kept across its starts.
type mesh struct {
	t   *testing.T
	ctx context.Context
	nc  *nats.Conn
	js  jetstream.JetStream
	bin string
	// toolsBin is the tools bundle's binary.
	toolsBin string
	dir      string
	vendor   *vendor
	own      issued // what the manager's own runtime is issued from its manifest (C1)

	mu         sync.Mutex
	said       []said
	news       chan struct{}
	secrets    map[string]string   // the vault: a secret's name and its value
	calls      map[string][32]byte // the vault: licence-put's open calls and the private key made for each
	callArgs   map[string]map[string]any
	holdings   []Holdings           // the agent modules: what each node says it holds when asked
	logins     map[string]FullGrant // the agent modules: the login each node hands over, by node
	conditions []oc.Condition
}

func newWholeMesh(t *testing.T) *mesh {
	t.Helper()
	// The manager runs as its own account, never in the machine's tool runner, whose account is issued none of its
	// logs or states (C1): this player plays that runtime of its own, issuing what the manifest names and nothing else.
	own := manifestIssued(t)
	if own.RunsAs != Publisher {
		t.Fatalf("the manager declares runs-as %q: it would run in the machine's tool runner, not as its own account", own.RunsAs)
	}
	ctx, cancel := context.WithCancel(context.Background())
	t.Cleanup(cancel)
	dir := t.TempDir()
	// The module's two bundles, each loaded by the same runtime (design 39 parts §2, §3).
	bin, toolsBin := filepath.Join(dir, "claude-licence-manager"), filepath.Join(dir, "claude-licence-manager-tools")
	for out, from := range map[string]string{bin: ".", toolsBin: "../tools"} {
		build := exec.Command("go", "build", "-o", out, from)
		build.Stderr = os.Stderr
		if err := build.Run(); err != nil {
			t.Fatalf("%s does not build: %v", from, err)
		}
	}
	nc, err := nats.Connect(testbus.URL(t))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(nc.Close)
	js, err := jetstream.New(nc)
	if err != nil {
		t.Fatal(err)
	}
	// What the bus module creates from the manifest's templates (§1 to §9) before the manager's want moves (§50).
	for s := range own.States {
		if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: Publisher + "_" + s, History: 8}); err != nil {
			t.Fatal(err)
		}
	}
	// Each log with the cap its logs section names (C3), as the bus module creates it.
	raw, err := os.ReadFile(filepath.Join(moduleDir, "module.json"))
	if err != nil {
		t.Fatal(err)
	}
	var manifest struct {
		Logs []json.RawMessage `json:"logs"`
	}
	_ = json.Unmarshal(raw, &manifest)
	bounds := logsSection(manifest.Logs)
	for l := range own.Logs {
		mib, _ := bounds[l]["max-mib"].(float64)
		if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "LOG_" + Publisher + "_" + l,
			Subjects: []string{"mesh.log." + Publisher + "." + l + ".>"}, MaxBytes: int64(mib) << 20}); err != nil {
			t.Fatal(err)
		}
	}
	_ = os.MkdirAll(filepath.Join(dir, "state"), 0o700)
	if err := os.WriteFile(filepath.Join(dir, "settings.json"), []byte("{}"), 0o600); err != nil {
		t.Fatal(err)
	}
	m := &mesh{t: t, ctx: ctx, nc: nc, js: js, bin: bin, toolsBin: toolsBin, dir: dir, vendor: newVendor(t), own: own,
		news: make(chan struct{}, 1), secrets: map[string]string{}, calls: map[string][32]byte{},
		callArgs: map[string]map[string]any{}, logins: map[string]FullGrant{}}
	sub, err := nc.Subscribe("mesh.seat.operator-channel.accept.condition.>", func(msg *nats.Msg) {
		c, err := occ.DecodeCondition(msg.Data)
		if err != nil {
			t.Errorf("a condition the operator channel's client does not decode: %v: %s", err, msg.Data)
			return
		}
		if want, _ := occ.ConditionSubject(Publisher); msg.Subject != want {
			t.Errorf("a condition said on %s, not %s", msg.Subject, want)
		}
		m.mu.Lock()
		m.conditions = append(m.conditions, c)
		m.mu.Unlock()
		m.notify()
	})
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = sub.Unsubscribe() })
	return m
}

func (m *mesh) notify() {
	select {
	case m.news <- struct{}{}:
	default:
	}
}

// until waits for what to hold, woken by each event, call and condition the manager says, never by a timer of its own.
func (m *mesh) until(what string, ok func() bool) {
	m.t.Helper()
	deadline := time.After(60 * time.Second)
	for {
		m.mu.Lock()
		done := ok()
		m.mu.Unlock()
		if done {
			return
		}
		select {
		case <-m.news:
		case <-deadline:
			m.t.Fatalf("never: %s", what)
		}
	}
}

// count is how many of what was said have key; under mu.
func (m *mesh) count(key string) int {
	n := 0
	for _, s := range m.said {
		if s.key == key {
			n++
		}
	}
	return n
}

// of is the bodies said under key; under mu.
func (m *mesh) of(key string) []map[string]any {
	var out []map[string]any
	for _, s := range m.said {
		if s.key == key {
			var b map[string]any
			_ = json.Unmarshal(s.body, &b)
			out = append(out, b)
		}
	}
	return out
}

// conditionsOf is each condition said with key, in order; under mu.
func (m *mesh) conditionsOf(key string) []oc.Condition {
	var out []oc.Condition
	for _, c := range m.conditions {
		if c.Key == key {
			out = append(out, c)
		}
	}
	return out
}

// started is one start of the binary.
type started struct {
	*mesh
	in      io.WriteCloser
	cmd     *exec.Cmd
	next    int
	pending map[string]chan wire
	pmu     sync.Mutex
	subs    []string
}

// start starts the manager as its runtime launches it, with the module's words (ADR 0192).
func (m *mesh) start() *started {
	m.t.Helper()
	return m.load(m.bin)
}

// load starts one bundle's binary as the module's runtime loads it, with the module's words, which the runtime hands
// to every bundle of the module (ADR 0192).
func (m *mesh) load(bin string) *started {
	m.t.Helper()
	cmd := exec.CommandContext(m.ctx, bin)
	cmd.Env = append(os.Environ(), "MESH_LICENCE_STATE="+filepath.Join(m.dir, "state"),
		"MESH_LICENCE_SETTINGS="+filepath.Join(m.dir, "settings.json"), "MESH_NODE=novox", "MESH_MODULE="+Publisher,
		"MESH_ANTHROPIC_TOKEN_ENDPOINT="+m.vendor.server.URL+"/token", "MESH_ANTHROPIC_USAGE_ENDPOINT="+m.vendor.server.URL+"/usage")
	in, _ := cmd.StdinPipe()
	out, _ := cmd.StdoutPipe()
	cmd.Stderr = os.Stderr
	if err := cmd.Start(); err != nil {
		m.t.Fatal(err)
	}
	p := &started{mesh: m, in: in, cmd: cmd, pending: map[string]chan wire{}}
	m.t.Cleanup(p.stop)
	go p.serve(out)
	return p
}

// stop ends this start, as the runtime does when the module moves.
func (p *started) stop() {
	_ = p.in.Close()
	_ = p.cmd.Process.Kill()
	_ = p.cmd.Wait()
}

func (p *started) send(v map[string]any) {
	raw, _ := json.Marshal(v)
	p.pmu.Lock()
	defer p.pmu.Unlock()
	_, _ = p.in.Write(append(raw, '\n'))
}

func (p *started) answer(id json.RawMessage, result any, err error) {
	if err != nil {
		code := -32000
		if errors.Is(err, jetstream.ErrKeyExists) {
			code = -32010 // a lost compare-and-set, as the runtime says it
		}
		p.send(map[string]any{"id": id, "error": map[string]any{"code": code, "message": err.Error()}})
		return
	}
	p.send(map[string]any{"id": id, "result": result})
}

func (p *started) serve(out io.Reader) {
	scanner := bufio.NewScanner(out)
	scanner.Buffer(make([]byte, 1<<20), 1<<24)
	for scanner.Scan() {
		var w wire
		if json.Unmarshal(scanner.Bytes(), &w) != nil {
			continue
		}
		if w.Method == "" {
			var key string
			_ = json.Unmarshal(w.ID, &key)
			p.pmu.Lock()
			ch := p.pending[key]
			delete(p.pending, key)
			p.pmu.Unlock()
			if ch != nil {
				ch <- w
			}
			continue
		}
		go p.handle(w)
	}
}

func (p *started) kv(state string) (jetstream.KeyValue, error) {
	return p.js.KeyValue(p.ctx, Publisher+"_"+state)
}

func (p *started) handle(w wire) {
	var a struct {
		State    string          `json:"state"`
		Key      string          `json:"key"`
		Value    json.RawMessage `json:"value"`
		Revision uint64          `json:"revision"`
		Subject  string          `json:"subject"`
		Body     json.RawMessage `json:"body"`
		Pattern  string          `json:"pattern"`
	}
	_ = json.Unmarshal(w.Params, &a)
	if why := p.refused(w.Method, w.Params); why != nil {
		p.answer(w.ID, nil, why)
		return
	}
	if strings.HasPrefix(w.Method, "mesh/log.") {
		p.log(w)
		return
	}
	switch w.Method {
	case "mesh/state.get":
		kv, err := p.kv(a.State)
		if err != nil {
			p.answer(w.ID, nil, err)
			return
		}
		e, err := kv.Get(p.ctx, a.Key)
		if errors.Is(err, jetstream.ErrKeyNotFound) {
			p.answer(w.ID, nil, nil)
			return
		}
		if err != nil {
			p.answer(w.ID, nil, err)
			return
		}
		p.answer(w.ID, map[string]any{"key": e.Key(), "value": json.RawMessage(e.Value()), "revision": e.Revision(), "at": e.Created()}, nil)
	case "mesh/state.put", "mesh/state.create", "mesh/state.update":
		kv, err := p.kv(a.State)
		var rev uint64
		if err == nil {
			// A key's own time to live is the bus's; no test here lives as long as one (a lease's two minutes).
			switch w.Method {
			case "mesh/state.put":
				rev, err = kv.Put(p.ctx, a.Key, a.Value)
			case "mesh/state.create":
				rev, err = kv.Create(p.ctx, a.Key, a.Value)
			default:
				rev, err = kv.Update(p.ctx, a.Key, a.Value, a.Revision)
				if err != nil {
					err = fmt.Errorf("%w: %v", jetstream.ErrKeyExists, err)
				}
			}
		}
		p.answer(w.ID, map[string]any{"revision": rev}, err)
	case "mesh/state.keys":
		kv, err := p.kv(a.State)
		if err != nil {
			p.answer(w.ID, nil, err)
			return
		}
		keys, err := kv.Keys(p.ctx)
		if errors.Is(err, jetstream.ErrNoKeysFound) {
			keys, err = []string{}, nil
		}
		sort.Strings(keys)
		p.answer(w.ID, keys, err)
	case "mesh/state.delete":
		kv, err := p.kv(a.State)
		if err == nil {
			err = kv.Delete(p.ctx, a.Key)
		}
		p.answer(w.ID, map[string]any{}, err)
	case "mesh/state.history":
		kv, err := p.kv(a.State)
		if err != nil {
			p.answer(w.ID, nil, err)
			return
		}
		out := []map[string]any{}
		h, err := kv.History(p.ctx, a.Key)
		if errors.Is(err, jetstream.ErrKeyNotFound) {
			p.answer(w.ID, out, nil)
			return
		}
		for _, e := range h {
			op, v := "put", json.RawMessage(e.Value())
			if e.Operation() != jetstream.KeyValuePut {
				op, v = "delete", nil
			}
			out = append(out, map[string]any{"key": e.Key(), "value": v, "revision": e.Revision(), "op": op, "at": e.Created()})
		}
		p.answer(w.ID, out, err)
	case "mesh/publish":
		p.answer(w.ID, map[string]any{}, p.nc.Publish("mesh.mod."+Publisher+".event."+a.Key, a.Body))
		p.say(a.Key, a.Body)
		if a.Key == HoldingsAsked {
			p.answerHoldings()
		}
	case "mesh/seat.publish":
		p.answer(w.ID, map[string]any{}, p.nc.Publish(a.Subject, a.Body))
	case "mesh/subscribe":
		p.pmu.Lock()
		p.subs = append(p.subs, a.Pattern)
		p.pmu.Unlock()
		p.answer(w.ID, map[string]any{}, nil)
	case "mesh/ask":
		result, err := p.asked(a.Key, a.Body)
		p.say("ask "+a.Key, a.Body)
		p.answer(w.ID, result, err)
	default:
		p.answer(w.ID, nil, errors.New("nats: no responders available for request"))
	}
}

// refused is the runtime's refusal of what the manager's own account was not issued (C1): a state or a log its
// manifest does not ask for, an ask its invokes do not name, an event its emits do not name; nil when it was issued.
func (p *started) refused(method string, params json.RawMessage) error {
	var a struct {
		State string `json:"state"`
		Log   string `json:"log"`
		Key   string `json:"key"`
	}
	_ = json.Unmarshal(params, &a)
	switch {
	case strings.HasPrefix(method, "mesh/state.") && !p.own.States[a.State]:
		return fmt.Errorf("%s keeps no state called %q", Publisher, a.State)
	case strings.HasPrefix(method, "mesh/log.") && !p.own.Logs[a.Log]:
		return fmt.Errorf("%s keeps no log called %q", Publisher, a.Log)
	case method == "mesh/ask" && !p.own.mayAsk(a.Key):
		return fmt.Errorf("this account may not call %s", a.Key)
	case method == "mesh/publish" && !p.own.Emits[a.Key]:
		return fmt.Errorf("%s does not emit %s", Publisher, a.Key)
	}
	return nil
}

func (p *started) say(key string, body json.RawMessage) {
	p.mu.Lock()
	p.said = append(p.said, said{key: key, body: append(json.RawMessage(nil), body...)})
	p.mu.Unlock()
	p.notify()
}

// answerHoldings is each node's agent module hearing holdings.asked and saying what it holds (§23).
func (p *started) answerHoldings() {
	p.mu.Lock()
	reports := append([]Holdings(nil), p.holdings...)
	p.mu.Unlock()
	for _, r := range reports {
		body, _ := json.Marshal(r)
		p.mu.Lock()
		p.next++
		id := fmt.Sprintf("e-%d", p.next)
		p.mu.Unlock()
		p.send(map[string]any{"id": id, "method": "mesh/event", "params": map[string]any{"envelope": map[string]any{
			"key": HoldingsChanged, "node": r.Node, "body": json.RawMessage(body)}}})
	}
}

// asked answers what the manager asks of another part: the vault, a node's agent module, the operator channel.
func (p *started) asked(key string, body json.RawMessage) (any, error) {
	var args map[string]any
	_ = json.Unmarshal(body, &args)
	p.mu.Lock()
	defer p.mu.Unlock()
	switch {
	case key == "seat:mesh-vault.licence-put":
		if call, _ := args["call"].(string); call != "" {
			priv, ok := p.calls[call]
			if !ok {
				return nil, errors.New("no such call")
			}
			sealed, _ := base64.StdEncoding.DecodeString(fmt.Sprint(args["sealed"]))
			var pub [32]byte
			copy(pub[:], p.callArgs[call]["pub"].([]byte))
			value, ok := box.OpenAnonymous(nil, sealed, &pub, &priv)
			if !ok {
				return nil, errors.New("the sealed value does not open")
			}
			a := p.callArgs[call]
			p.secrets[SecretName(fmt.Sprint(a["licence"]), fmt.Sprint(a["part"]))] = string(value)
			delete(p.calls, call)
			return map[string]any{"put": true}, nil
		}
		pub, priv, _ := box.GenerateKey(rand.Reader)
		call := fmt.Sprintf("call-%d", len(p.callArgs)+1)
		p.calls[call] = *priv
		args["pub"] = pub[:]
		p.callArgs[call] = args
		return map[string]any{"call": call, "seal_to": base64.StdEncoding.EncodeToString(pub[:])}, nil
	case key == "seat:mesh-vault.get":
		name, _ := args["name"].(string)
		v, ok := p.secrets[name]
		if !ok {
			return nil, fmt.Errorf("the vault holds no %s", name)
		}
		sealed, err := Seal(v, fmt.Sprint(args["public_key"]))
		return map[string]any{"sealed": sealed}, err
	case strings.HasPrefix(key, "claude-code.claude_code_grant@"):
		node := strings.TrimPrefix(key, "claude-code.claude_code_grant@")
		g, ok := p.logins[node]
		if !ok {
			return GrantAnswer{}, nil
		}
		raw, _ := json.Marshal(g)
		sealed, err := Seal(string(raw), fmt.Sprint(args["public_key"]))
		return GrantAnswer{Sealed: &sealed, Fingerprint: Fingerprint(g.RefreshToken)}, err
	case key == "seat:operator-channel.tell":
		return map[string]any{"told": true}, nil
	}
	return nil, errors.New("nats: no responders available for request")
}

// keyMatches is a log's key against a pattern: `*` one name, `>` or `**` the rest, empty every key.
func keyMatches(pattern, key string) bool {
	if pattern == "" {
		return true
	}
	return namesMatch(strings.Split(pattern, "."), strings.Split(key, "."))
}

func namesMatch(p, k []string) bool {
	if len(p) == 0 {
		return len(k) == 0
	}
	if p[0] == ">" || p[0] == "**" {
		return len(k) > 0
	}
	if len(k) == 0 || (p[0] != "*" && p[0] != k[0]) {
		return false
	}
	return namesMatch(p[1:], k[1:])
}

// log answers the runtime's log verbs from the log's real stream, LOG_<module>_<name> on mesh.log.<module>.<name>.<key>.
func (p *started) log(w wire) {
	var a struct {
		Log   string          `json:"log"`
		Key   string          `json:"key"`
		Keys  string          `json:"keys"`
		Value json.RawMessage `json:"value"`
		From  uint64          `json:"from"`
		Since string          `json:"since"`
		Limit int             `json:"limit"`
	}
	_ = json.Unmarshal(w.Params, &a)
	subject := "mesh.log." + Publisher + "." + a.Log
	s, err := p.js.Stream(p.ctx, "LOG_"+Publisher+"_"+a.Log)
	if err != nil {
		p.answer(w.ID, nil, err)
		return
	}
	info, err := s.Info(p.ctx)
	if err != nil {
		p.answer(w.ID, nil, err)
		return
	}
	entry := func(r *jetstream.RawStreamMsg) map[string]any {
		return map[string]any{"sequence": r.Sequence, "key": strings.TrimPrefix(r.Subject, subject+"."),
			"value": json.RawMessage(r.Data), "at": r.Time}
	}
	all := func(keys string, since time.Time) []map[string]any {
		out := []map[string]any{}
		for seq := max(a.From, info.State.FirstSeq, 1); seq <= info.State.LastSeq; seq++ {
			r, err := s.GetMsg(p.ctx, seq)
			if err != nil || !keyMatches(keys, strings.TrimPrefix(r.Subject, subject+".")) || r.Time.Before(since) {
				continue
			}
			out = append(out, entry(r))
		}
		return out
	}
	switch w.Method {
	case "mesh/log.append":
		ack, err := p.js.Publish(p.ctx, subject+"."+a.Key, a.Value)
		if err != nil {
			p.answer(w.ID, nil, err)
			return
		}
		p.answer(w.ID, map[string]any{"sequence": ack.Sequence}, nil)
	case "mesh/log.last":
		r, err := s.GetLastMsgForSubject(p.ctx, subject+"."+a.Key)
		if errors.Is(err, jetstream.ErrMsgNotFound) {
			p.answer(w.ID, nil, nil)
			return
		}
		if err != nil {
			p.answer(w.ID, nil, err)
			return
		}
		p.answer(w.ID, entry(r), nil)
	case "mesh/log.read":
		var since time.Time
		if a.Since != "" {
			since, _ = time.Parse(time.RFC3339Nano, a.Since)
		}
		entries := all(a.Keys, since)
		limit := a.Limit
		if limit <= 0 {
			limit = 100
		}
		next := uint64(0)
		if len(entries) > limit {
			next = entries[limit]["sequence"].(uint64)
			entries = entries[:limit]
		}
		p.answer(w.ID, map[string]any{"entries": entries, "next": next, "last": info.State.LastSeq}, nil)
	case "mesh/log.newest":
		newest := map[string]int{}
		var out []map[string]any
		for _, e := range all(a.Keys, time.Time{}) {
			k := e["key"].(string)
			if i, ok := newest[k]; ok {
				out[i] = e
				continue
			}
			newest[k] = len(out)
			out = append(out, e)
		}
		if out == nil {
			out = []map[string]any{}
		}
		p.answer(w.ID, out, nil)
	case "mesh/log.info":
		p.answer(w.ID, map[string]any{"entries": info.State.Msgs, "bytes": info.State.Bytes,
			"max_bytes": info.Config.MaxBytes, "first": info.State.FirstSeq, "last": info.State.LastSeq}, nil)
	default:
		p.answer(w.ID, nil, fmt.Errorf("no log verb %s", w.Method))
	}
}

// call is a tools/call of the manager by a person, answered as the tool's JSON.
func (p *started) call(name string, args map[string]any) (json.RawMessage, error) {
	p.mu.Lock()
	p.next++
	id := fmt.Sprintf("c-%d", p.next)
	p.mu.Unlock()
	ch := make(chan wire, 1)
	p.pmu.Lock()
	p.pending[id] = ch
	p.pmu.Unlock()
	p.send(map[string]any{"id": id, "method": "tools/call", "params": map[string]any{"name": name, "arguments": args,
		"_meta": map[string]any{"mesh/caller": "person.operator"}}})
	select {
	case w := <-ch:
		var r struct {
			Content []struct {
				Text string `json:"text"`
			} `json:"content"`
			IsError bool `json:"isError"`
		}
		if w.Error != nil {
			return nil, errors.New(w.Error.Message)
		}
		if err := json.Unmarshal(w.Result, &r); err != nil || len(r.Content) != 1 {
			return nil, fmt.Errorf("%s answered %s", name, w.Result)
		}
		if r.IsError {
			return nil, errors.New(r.Content[0].Text)
		}
		return json.RawMessage(r.Content[0].Text), nil
	case <-time.After(60 * time.Second):
		return nil, fmt.Errorf("%s did not answer in 60s", name)
	}
}

func (p *started) must(name string, args map[string]any) json.RawMessage {
	p.t.Helper()
	raw, err := p.call(name, args)
	if err != nil {
		p.t.Fatalf("%s: %v", name, err)
	}
	return raw
}

// bindings are the consumers' bindings, as the seat's verb answers them.
func (p *started) bindings() map[string]Binding {
	p.t.Helper()
	var list []Binding
	if err := json.Unmarshal(p.must(Seat+".bindings", nil), &list); err != nil {
		p.t.Fatal(err)
	}
	out := map[string]Binding{}
	for _, b := range list {
		out[b.Consumer] = b
	}
	return out
}

// ---- the tests -------------------------------------------------------------------------------------------------

// design 39 bus-and-switch §26/AC2, §28, §29/AC2, §9/AC3, B41: a reading of 99 % moves the four consumers bound to a
// licence to the one with the most room in one act, saying one binding.changed each, one licence.switched and one tell;
// with every licence at its threshold no-room is raised on the operator channel's subject, and a reading with room
// clears it.
func TestAReadingAtTheThresholdMovesEveryConsumer(t *testing.T) {
	m := newWholeMesh(t)
	m.seed()
	p := m.start()
	m.until("holdings.asked at the start", func() bool { return m.count(HoldingsAsked) > 0 })
	m.mu.Lock()
	carried := m.count("binding.changed")
	m.mu.Unlock()
	m.vendor.set("at-b", 20)
	m.vendor.set("at-a", 99)
	p.must("licences_read_usage", map[string]any{"licence": licenceB})
	p.must("licences_read_usage", map[string]any{"licence": licenceA})
	m.until("the four consumers moved", func() bool { return m.count("binding.changed") >= carried+4 })
	m.mu.Lock()
	switched, told := m.of("licence.switched"), m.of("ask seat:operator-channel.tell")
	changed := m.of("binding.changed")[carried:]
	m.mu.Unlock()
	if len(switched) != 1 || switched[0]["from"] != licenceA || switched[0]["to"] != licenceB || switched[0]["by"] != "usage" ||
		len(switched[0]["consumers"].([]any)) != 4 {
		t.Errorf("licence.switched said %v", switched)
	}
	if len(told) != 1 {
		t.Errorf("the operator was told %d times: %v", len(told), told)
	}
	moved := map[string]bool{}
	for _, c := range changed {
		if c["licence"] == licenceB && c["from"] == licenceA && c["by"] == "usage" {
			moved[fmt.Sprint(c["consumer"])] = true
		}
	}
	if len(moved) != 4 || len(changed) != 4 {
		t.Errorf("binding.changed said %v", changed)
	}
	for c, b := range p.bindings() {
		if b.Licence != licenceB || b.By != "usage" {
			t.Errorf("%s is bound %+v", c, b)
		}
	}

	// Every licence at its threshold: the consumers have nowhere to go.
	noRoom := "licence." + KeyFor(licenceB) + ".no-room"
	m.vendor.set("at-a", 99)
	m.vendor.set("at-b", 99)
	p.must("licences_read_usage", map[string]any{"licence": licenceA})
	p.must("licences_read_usage", map[string]any{"licence": licenceB})
	m.until("no-room is raised", func() bool { return len(m.conditionsOf(noRoom)) > 0 })
	m.mu.Lock()
	r := m.conditionsOf(noRoom)[0]
	m.mu.Unlock()
	if r.State != "raised" || r.Severity != "urgent" || r.Subject.ID != Publisher {
		t.Errorf("no-room said %+v", r)
	}
	// A reading with room: the next reading of the full licence moves its consumers, and no-room is cleared.
	m.vendor.set("at-a", 30)
	p.must("licences_read_usage", map[string]any{"licence": licenceA})
	p.must("licences_read_usage", map[string]any{"licence": licenceB})
	m.until("no-room is cleared", func() bool {
		c := m.conditionsOf(noRoom)
		return len(c) > 1 && c[len(c)-1].State == "cleared"
	})
	m.mu.Lock()
	defer m.mu.Unlock()
	if c := m.conditionsOf(noRoom); len(c) != 2 {
		t.Errorf("no-room was said %d times, not once raised and once cleared: %+v", len(c), c)
	}
	if n := m.count("licence.switched"); n != 2 {
		t.Errorf("licence.switched was said %d times, not once per act", n)
	}
}

// design 39 bus-and-switch §23/AC1, §46/AC1, design 39 parts §11: on a fresh install the manager waits for no carry-over
// and says holdings.asked; each node's agent module answers with one holdings-changed; a report carrying an
// unknown refresh token's fingerprint makes a candidate, which is adopted, listed by licences with its organisation,
// and chosen as a target at its first reading with room.
func TestANewLoginIsAdoptedListedAndATarget(t *testing.T) {
	m := newWholeMesh(t)
	changed := time.Now().UTC().Format(time.RFC3339Nano)
	report := func(node, account, email, org, refresh string) Holdings {
		h := Holdings{Node: node, Identity: &Identity{AccountUUID: account, EmailAddress: email, OrganizationUUID: org},
			Kind: "subscription", ChangedAt: changed}
		h.Refresh.Present, h.Refresh.Fingerprint = true, Fingerprint(refresh)
		return h
	}
	m.holdings = []Holdings{report("laptop", "acct-old", licenceA, "", "rt-old"), report("workstation", "acct-team", "team@example.org", "org-1", "rt-team")}
	m.logins["laptop"] = FullGrant{AccessToken: "at-old-0", RefreshToken: "rt-old"}
	m.logins["workstation"] = FullGrant{AccessToken: "at-team-0", RefreshToken: "rt-team"}
	m.vendor.logins["rt-old"] = [2]string{"at-old", "acct-old"}
	m.vendor.logins["rt-team"] = [2]string{"at-team", "acct-team"}
	p := m.start()
	m.until("both logins adopted", func() bool { return m.count("licence.adopted") >= 2 })
	m.mu.Lock()
	asked := m.count("ask claude-code.claude_code_grant@laptop") + m.count("ask claude-code.claude_code_grant@workstation")
	m.mu.Unlock()
	if asked != 2 {
		t.Errorf("the nodes were asked for their logins %d times, not once each", asked)
	}

	// The full licence is read first: nothing has room yet, so its consumer stays.
	m.vendor.set("at-old", 99)
	m.vendor.set("at-team", 5)
	p.must("licences_read_usage", map[string]any{"licence": licenceA})
	p.must("licences_read_usage", map[string]any{"licence": "team@example.org"})
	var listed []map[string]any
	if err := json.Unmarshal(p.must(Seat+".licences", nil), &listed); err != nil {
		t.Fatal(err)
	}
	var team map[string]any
	for _, l := range listed {
		if l["name"] == "team@example.org" {
			team = l
		}
	}
	if team == nil || team["organisation"] != "org-1" || team["target"] != true || team["usage"] == nil {
		t.Fatalf("licences lists the new login as %v (all: %v)", team, listed)
	}
	p.must("licences_read_usage", map[string]any{"licence": licenceA})
	m.until("the full licence's consumer moved to the new one", func() bool {
		for _, s := range m.of("licence.switched") {
			if s["to"] == "team@example.org" && s["from"] == licenceA {
				return true
			}
		}
		return false
	})
	if b := p.bindings()["laptop"]; b.Licence != "team@example.org" {
		t.Errorf("laptop is bound %+v", b)
	}
}

// design 39 parts §3, §10, §18: with the manager bundle not loaded (as when it refuses to start), the tools bundle,
// loaded by the same runtime, answers licences_kept with every entry the module keeps and licences_conditions with
// each condition kept, and says, writes and asks nothing while it does.
func TestTheReadingToolsAnswerWithoutTheManager(t *testing.T) {
	m := newWholeMesh(t)
	m.seed()
	carried, err := m.js.KeyValue(m.ctx, Publisher+"_carried")
	if err != nil {
		t.Fatal(err)
	}
	if _, err := carried.Put(m.ctx, "postgres", []byte(`{"at":"2026-11-12T09:44:59Z"}`)); err != nil {
		t.Fatal(err)
	}
	seqs := func() map[string]uint64 {
		out := map[string]uint64{}
		for s := range m.own.States {
			st, err := m.js.Stream(m.ctx, "KV_"+Publisher+"_"+s)
			if err != nil {
				t.Fatal(err)
			}
			i, _ := st.Info(m.ctx)
			out[s] = i.State.LastSeq
		}
		for l := range m.own.Logs {
			st, err := m.js.Stream(m.ctx, "LOG_"+Publisher+"_"+l)
			if err != nil {
				t.Fatal(err)
			}
			i, _ := st.Info(m.ctx)
			out["log "+l] = i.State.LastSeq
		}
		return out
	}
	before := seqs()
	tools := m.load(m.toolsBin)
	var all map[string]map[string]any
	if err := json.Unmarshal(tools.must("licences_kept", nil), &all); err != nil {
		t.Fatal(err)
	}
	if len(all) != len(m.own.States)+len(m.own.Logs) || len(all) != 8 {
		t.Fatalf("licences_kept answers %d entries without the manager, not 8: %v", len(all), all)
	}
	for name, e := range all {
		if e["fill"] == nil {
			t.Errorf("licences_kept answers %s without its fill: %v", name, e)
		}
	}
	var one map[string]map[string]any
	if err := json.Unmarshal(tools.must("licences_kept", map[string]any{"entry": "carried", "key": "postgres"}), &one); err != nil {
		t.Fatal(err)
	}
	if v, _ := one["carried"]["value"].(map[string]any); v["at"] != "2026-11-12T09:44:59Z" {
		t.Errorf("the carry-over's record is answered as %v", one)
	}
	if keys, _ := all["bindings"]["keys"].([]any); len(keys) != len(consumers) {
		t.Errorf("bindings answers %v", all["bindings"])
	}
	tools.must("licences_conditions", nil)
	if after := seqs(); fmt.Sprint(after) != fmt.Sprint(before) {
		t.Errorf("the tools bundle wrote: before %v, after %v", before, after)
	}
	m.mu.Lock()
	defer m.mu.Unlock()
	if len(m.said) != 0 || len(m.conditions) != 0 {
		t.Errorf("the tools bundle said or asked %v and %v", m.said, m.conditions)
	}
}