mesh-catalog/modules/mesh-hand-acts/cmd/mesh-hand-acts/whole_integration_test.go

//go:build integration

package main

// mesh-hand-acts whole against a real bus (novox/hq ADR 0565 §10; issue 1953 L5-C2): 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.* answered
// from a real nats-server's key-value buckets, mesh/log.* from the log's real stream, mesh/publish onto its subjects). A recorder's hand act is kept in the
// log on the bus under the proven caller and heard as act-recorded; a warrant the router recorded is recorded once.

import (
	"bufio"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"os"
	"os/exec"
	"path/filepath"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"

	"git.novox.be/novox/mesh-sdk/go/testbus"
)

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"`
}

// player is the node's tool runtime, as far as the self-check needs it, over a real bus.
type player struct {
	ctx     context.Context
	nc      *nats.Conn
	js      jetstream.JetStream
	mu      sync.Mutex
	in      io.Writer
	next    int
	pending map[string]chan wire
}

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

func (p *player) answer(id json.RawMessage, result any, err error) {
	if err != nil {
		p.send(map[string]any{"id": id, "error": map[string]any{"code": -32000, "message": err.Error()}})
		return
	}
	p.send(map[string]any{"id": id, "result": result})
}

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

// kv is the bucket a state names: this module's own by its local name, another's as `<owner>.<name>`.
func (p *player) kv(state string) (jetstream.KeyValue, error) {
	if owner, name, ok := strings.Cut(state, "."); ok {
		return p.js.KeyValue(p.ctx, owner+"_"+name)
	}
	return p.js.KeyValue(p.ctx, Module+"_"+state)
}

func (p *player) handle(m wire) {
	var a struct {
		State string          `json:"state"`
		Key   string          `json:"key"`
		Value json.RawMessage `json:"value"`
		Body  json.RawMessage `json:"body"`
	}
	_ = json.Unmarshal(m.Params, &a)
	switch m.Method {
	case "mesh/state.get":
		kv, err := p.kv(a.State)
		if err != nil {
			p.answer(m.ID, nil, err)
			return
		}
		e, err := kv.Get(p.ctx, a.Key)
		if errors.Is(err, jetstream.ErrKeyNotFound) {
			p.answer(m.ID, nil, nil)
			return
		}
		if err != nil {
			p.answer(m.ID, nil, err)
			return
		}
		p.answer(m.ID, map[string]any{"key": e.Key(), "value": json.RawMessage(e.Value()), "revision": e.Revision()}, nil)
	case "mesh/state.put":
		kv, err := p.kv(a.State)
		if err == nil {
			var rev uint64
			if rev, err = kv.Put(p.ctx, a.Key, a.Value); err == nil {
				p.answer(m.ID, map[string]any{"revision": rev}, nil)
				return
			}
		}
		p.answer(m.ID, nil, err)
	case "mesh/state.create":
		kv, err := p.kv(a.State)
		if err == nil {
			var rev uint64
			if rev, err = kv.Create(p.ctx, a.Key, a.Value); err == nil {
				p.answer(m.ID, map[string]any{"revision": rev}, nil)
				return
			}
			if errors.Is(err, jetstream.ErrKeyExists) {
				p.send(map[string]any{"id": m.ID, "error": map[string]any{"code": -32001, "message": "the key was written by another writer first"}})
				return
			}
		}
		p.answer(m.ID, nil, err)
	case "mesh/state.watch":
		kv, err := p.kv(a.State)
		if err != nil {
			p.answer(m.ID, nil, err)
			return
		}
		key := a.Key
		if key == "" {
			key = ">"
		}
		w, err := kv.Watch(p.ctx, key)
		if err != nil {
			p.answer(m.ID, nil, err)
			return
		}
		var watch string
		_ = json.Unmarshal(m.ID, &watch)
		current := true
		for e := range w.Updates() {
			if e == nil {
				current = false
				p.answer(m.ID, map[string]any{}, nil)
				continue
			}
			op := "put"
			if e.Operation() != jetstream.KeyValuePut {
				op = "delete"
			}
			change := map[string]any{"state": a.State, "key": e.Key(), "op": op, "revision": e.Revision(), "current": current}
			if op == "put" {
				change["value"] = json.RawMessage(e.Value())
			}
			p.mu.Lock()
			p.next++
			id := fmt.Sprintf("w-%d", p.next)
			p.mu.Unlock()
			p.send(map[string]any{"id": id, "method": "mesh/state", "params": map[string]any{"watch": watch, "change": change}})
		}
	case "mesh/subscribe":
		p.answer(m.ID, map[string]any{}, nil)
	case "mesh/log.append", "mesh/log.read", "mesh/log.last", "mesh/log.info":
		p.log(m)
	case "mesh/state.keys":
		kv, err := p.kv(a.State)
		if err != nil {
			p.answer(m.ID, nil, err)
			return
		}
		keys, err := kv.Keys(p.ctx)
		if errors.Is(err, jetstream.ErrNoKeysFound) {
			keys, err = []string{}, nil
		}
		p.answer(m.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(m.ID, map[string]any{}, err)
	case "mesh/seat.publish":
		var e struct {
			Subject string          `json:"subject"`
			Body    json.RawMessage `json:"body"`
		}
		_ = json.Unmarshal(m.Params, &e)
		p.answer(m.ID, map[string]any{}, p.nc.Publish(e.Subject, e.Body))
	case "mesh/publish":
		var e struct {
			Key  string          `json:"key"`
			Body json.RawMessage `json:"body"`
		}
		_ = json.Unmarshal(m.Params, &e)
		p.answer(m.ID, map[string]any{}, p.nc.Publish("mesh.mod."+Module+".event."+e.Key, e.Body))
	default:
		p.answer(m.ID, nil, errors.New("nats: no responders available for request"))
	}
}

// log answers the runtime's log verbs from the log's real stream, LOG_<module>_<name> on mesh.log.<module>.<name>.<key>.
func (p *player) log(m wire) {
	var a struct {
		Log   string          `json:"log"`
		Key   string          `json:"key"`
		Value json.RawMessage `json:"value"`
		From  uint64          `json:"from"`
	}
	_ = json.Unmarshal(m.Params, &a)
	stream := "LOG_" + Module + "_" + a.Log
	subject := "mesh.log." + Module + "." + a.Log
	s, err := p.js.Stream(p.ctx, stream)
	if err != nil {
		p.answer(m.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}
	}
	switch m.Method {
	case "mesh/log.append":
		ack, err := p.js.Publish(p.ctx, subject+"."+a.Key, a.Value)
		if err != nil {
			p.answer(m.ID, nil, err)
			return
		}
		p.answer(m.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(m.ID, nil, nil)
			return
		}
		if err != nil {
			p.answer(m.ID, nil, err)
			return
		}
		p.answer(m.ID, entry(r), nil)
	case "mesh/log.read":
		info, err := s.Info(p.ctx)
		if err != nil {
			p.answer(m.ID, nil, err)
			return
		}
		entries := []map[string]any{}
		for seq := max(a.From, 1); seq <= info.State.LastSeq; seq++ {
			if r, err := s.GetMsg(p.ctx, seq); err == nil {
				entries = append(entries, entry(r))
			}
		}
		p.answer(m.ID, map[string]any{"entries": entries, "next": 0, "last": info.State.LastSeq}, nil)
	case "mesh/log.info":
		info, err := s.Info(p.ctx)
		if err != nil {
			p.answer(m.ID, nil, err)
			return
		}
		p.answer(m.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)
	}
}

func (p *player) call(name, caller string, args map[string]any) (json.RawMessage, error) {
	p.mu.Lock()
	p.next++
	id := fmt.Sprintf("c-%d", p.next)
	ch := make(chan wire, 1)
	p.pending[id] = ch
	p.mu.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": caller}}})
	select {
	case m := <-ch:
		if m.Error != nil {
			return nil, errors.New(m.Error.Message)
		}
		var r struct {
			Content []struct {
				Text string `json:"text"`
			} `json:"content"`
			IsError bool `json:"isError"`
		}
		if err := json.Unmarshal(m.Result, &r); err != nil || len(r.Content) != 1 {
			return nil, fmt.Errorf("%s answered %s", name, m.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)
	}
}

// 1953/L5-C2: an act recorded through the seat is kept on the bus and heard as act-recorded.
func TestTheHandActLogWholeRecordsOnTheBus(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	t.Cleanup(cancel)
	dir := t.TempDir()
	bin := filepath.Join(dir, Module)
	build := exec.Command("go", "build", "-o", bin, ".")
	build.Stderr = os.Stderr
	if err := build.Run(); err != nil {
		t.Fatalf("mesh-hand-acts does not build: %v", 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)
	}
	for _, b := range []string{Module + "_" + WarrantsState, Module + "_" + CopiedState, "mesh-controller_hand-acts", "messenger_asks"} {
		if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: b}); err != nil {
			t.Fatal(err)
		}
	}
	// The log as the mesh makes it (ADR 0297): a stream kept by limits, refusing when full, never deleting.
	logStream, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "LOG_" + Module + "_" + ActsLog,
		Subjects: []string{"mesh.log." + Module + "." + ActsLog + ".>"}, MaxBytes: 64 << 20, Discard: jetstream.DiscardNew,
		DenyDelete: true, DenyPurge: true, AllowDirect: true})
	if err != nil {
		t.Fatal(err)
	}
	recorded, err := nc.SubscribeSync("mesh.seat." + Seat + ".event." + RecordedEvent)
	if err != nil {
		t.Fatal(err)
	}
	// One act of the controller's, kept before the module starts: the module no longer reads that bucket, so it never copies it.
	oldActs, _ := js.KeyValue(ctx, "mesh-controller_hand-acts")
	oldKey := fmt.Sprintf("act-%d-1", time.Now().Add(-time.Hour).UnixNano())
	if _, err := oldActs.Put(ctx, oldKey, []byte(`{"id":"`+oldKey+`","at":"`+time.Now().Add(-time.Hour).UTC().Format(time.RFC3339)+
		`","by":"ace.old","verb":"hand-act","why":"w","cause":"c"}`)); err != nil {
		t.Fatal(err)
	}
	cmd := exec.CommandContext(ctx, bin)
	cmd.Env = append(os.Environ(), "MESH_NODE=novox", "MESH_MODULE="+Module)
	stdin, _ := cmd.StdinPipe()
	stdout, _ := cmd.StdoutPipe()
	cmd.Stderr = os.Stderr
	if err := cmd.Start(); err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = stdin.Close(); cancel(); _ = cmd.Wait() })
	p := &player{ctx: ctx, nc: nc, js: js, in: stdin, pending: map[string]chan wire{}}
	go p.serve(stdout)

	raw, err := p.call(HandActTool, "ace.mesh-config", map[string]any{"what": "set dns on g14", "why": "it was wrong",
		"cause": "config", "by": "someone else"})
	if err != nil {
		t.Fatal(err)
	}
	var r Recorded
	if err := json.Unmarshal(raw, &r); err != nil || r.Act.By != "ace.mesh-config" {
		t.Fatalf("hand-act answered %s", raw)
	}
	e, err := logStream.GetLastMsgForSubject(ctx, "mesh.log."+Module+"."+ActsLog+"."+r.Act.ID)
	if err != nil || !strings.Contains(string(e.Data), `"by":"ace.mesh-config"`) {
		t.Fatalf("the act is not kept in the log on the bus: %v", err)
	}
	msg, err := recorded.NextMsg(10 * time.Second)
	if err != nil || !strings.Contains(string(msg.Data), r.Act.ID) {
		t.Fatalf("act-recorded was not said: %v", err)
	}
	// A warrant the router recorded, asked twice: one act.
	asks, _ := js.KeyValue(ctx, "messenger_asks")
	if _, err := asks.Put(ctx, "claude-code.a1", []byte(`{"state":"closed","warrant":{"ask":"a1","asker":"claude-code",
		"outcome":"chosen","label":"Approve","by":{"who":"operator","kind":"telegram","identity":"42"},"channel":"telegram",
		"at":"2026-10-29T21:00:00Z","ask-digest":"sha256:x"}}`)); err != nil {
		t.Fatal(err)
	}
	for _, caller := range []string{"ace.claude-code", "g14.claude-code"} {
		if _, err := p.call(WarrantedTool, caller, map[string]any{"asker": "claude-code", "ask": "a1"}); err != nil {
			t.Fatal(err)
		}
	}
	if c, err := p.call(CopyTool, "person.jochen", map[string]any{}); err != nil || !strings.Contains(string(c), "no longer read") || strings.Contains(string(c), `"done":true`) {
		t.Errorf("the copy's tool answered %s, %v", c, err)
	}
	if _, err := logStream.GetLastMsgForSubject(ctx, "mesh.log."+Module+"."+ActsLog+"."+oldKey); err == nil {
		t.Errorf("the controller's act was copied although its bucket is no longer read")
	}
	if info, err := logStream.Info(ctx); err != nil || info.State.Msgs != 2 {
		t.Errorf("the log keeps %v, want the act and one warrant (%v)", info, err)
	}
	listed, err := p.call(HandActsTool, "person.jochen", map[string]any{})
	if err != nil || !strings.Contains(string(listed), "warrant-claude-code-a1") || !strings.Contains(string(listed), r.Act.ID) || strings.Contains(string(listed), oldKey) {
		t.Errorf("hand-acts answered %s, %v", listed, err)
	}
}