mesh-tools/node-tools/internal/console/wait_test.go

package console

import (
	"context"
	"encoding/json"
	"net/http"
	"os"
	"reflect"
	"sort"
	"strings"
	"testing"
	"time"

	"github.com/novox/mesh-tools/node-tools/internal/bus"
	mt "github.com/novox/mesh-tools/node-tools/internal/meshtest"
)

// waitingRuntime is the tool runner on a machine, issued the families ADR 0367 §5 names, as the controller
// composes them from its manifest: the surface over its connection, and the mesh to publish on.
func waitingRuntime(t *testing.T) (*Surface, *mt.Mesh) {
	t.Helper()
	mesh := mt.New(t)
	m := mt.MembershipOf("node-tools", "desk", false, nil)
	m.Waits = []bus.WaitIssued{
		{Event: "mesh-knowledge.reviewed", Subject: "mesh.mod.mesh-knowledge.event.reviewed"},
		{Event: "operator-channel.decided.claude-code", Subject: "mesh.seat.operator-channel.event.decided.claude-code"},
	}
	mesh.Issue(t, m)
	conn := connect(t, "node-tools", "desk")
	mt.Until(t, func() error {
		if got := conn.Membership("node-tools"); got == nil || len(got.Waits) == 0 {
			return os.ErrNotExist
		}
		return nil
	})
	return NewSurface(conn, "desk.node-tools"), mesh
}

// callWait is one tools/call of mesh_wait, answered as an agent host reads it.
func callWait(s *Surface, args map[string]any) (text string, isError bool) {
	raw, _ := json.Marshal(map[string]any{"name": verbWait, "arguments": args})
	r := s.Handle(Request{JSONRPC: "2.0", ID: json.RawMessage("1"), Method: "tools/call", Params: raw})
	result, _ := r.Result.(map[string]any)
	content, _ := result["content"].([]map[string]any)
	var parts []string
	for _, c := range content {
		parts = append(parts, c["text"].(string))
	}
	isErr, _ := result["isError"].(bool)
	return strings.Join(parts, "\n"), isErr
}

// novox/hq ADR 0367 §1: mesh_wait answers the first event that matches — by its family and its filter — with
// its subject, body and time, within a second of its publication, and ignores the others.
func TestMeshWaitAnswersTheFirstMatchingEvent(t *testing.T) {
	s, mesh := waitingRuntime(t)
	type answer struct {
		text  string
		isErr bool
		at    time.Time
	}
	done := make(chan answer, 1)
	go func() {
		text, isErr := callWait(s, map[string]any{"subjects": []any{"mesh-knowledge.reviewed"},
			"filter": map[string]any{"kind": "decision", "number": 367}, "timeout": 20})
		done <- answer{text, isErr, time.Now()}
	}()
	// Once it listens: another record's verdict, another family's event, then the one it waits for.
	time.Sleep(300 * time.Millisecond)
	mesh.Emit(t, "mesh.mod.mesh-knowledge.event.reviewed", map[string]any{"kind": "decision", "number": 366, "what": "approved"})
	mesh.Emit(t, "mesh.mod.mesh-knowledge.event.proposed", map[string]any{"kind": "decision", "number": 367})
	published := time.Now()
	mesh.Emit(t, "mesh.mod.mesh-knowledge.event.reviewed", map[string]any{"kind": "decision", "number": 367, "what": "approved"})
	var got answer
	select {
	case got = <-done:
	case <-time.After(10 * time.Second):
		t.Fatal("mesh_wait did not answer the matching event")
	}
	if got.isErr {
		t.Fatalf("mesh_wait failed: %s", got.text)
	}
	if got.at.Sub(published) > time.Second {
		t.Errorf("answered %s after the event was published, want within a second", got.at.Sub(published))
	}
	var heard struct {
		Heard struct {
			Subject string         `json:"subject"`
			Body    map[string]any `json:"body"`
			At      time.Time      `json:"at"`
		} `json:"heard"`
	}
	if err := json.Unmarshal([]byte(got.text), &heard); err != nil {
		t.Fatalf("not an answer: %s", got.text)
	}
	if heard.Heard.Subject != "mesh-knowledge.reviewed" || heard.Heard.Body["number"] != float64(367) || heard.Heard.At.IsZero() {
		t.Errorf("answered %+v, want record 367's verdict with its time", heard.Heard)
	}
	// And the subject it is granted on the asker's own `decided`, by the name a session asks for.
	go func() {
		text, isErr := callWait(s, map[string]any{"subjects": "operator-channel.decided.claude-code", "timeout": 20})
		done <- answer{text, isErr, time.Now()}
	}()
	time.Sleep(300 * time.Millisecond)
	// A seat's event, said by the router on core NATS here: no stream of the test's bus carries the seat.
	if err := connect(t, "messenger", "").Say("mesh.seat.operator-channel.event.decided.claude-code", []byte(`{"ask":"a1"}`)); err != nil {
		t.Fatal(err)
	}
	select {
	case got = <-done:
		if got.isErr || !strings.Contains(got.text, `"subject": "operator-channel.decided.claude-code"`) {
			t.Errorf("the asker's decided: %s", got.text)
		}
	case <-time.After(10 * time.Second):
		t.Fatal("mesh_wait did not answer the asker's decided")
	}
}

// It says it timed out at its timeout, and leaves no subscription behind (ADR 0367 §1, §2).
func TestMeshWaitSaysItTimedOutAndKeepsNothing(t *testing.T) {
	s, _ := waitingRuntime(t)
	before := s.conn.Subscriptions()
	start := time.Now()
	text, isErr := callWait(s, map[string]any{"subjects": []any{"mesh-knowledge.reviewed"}, "timeout": 1})
	if took := time.Since(start); took < time.Second || took > 3*time.Second {
		t.Errorf("a one-second wait took %s", took)
	}
	if isErr || !strings.Contains(text, `"timed-out": true`) {
		t.Errorf("a wait that heard nothing: %s", text)
	}
	if after := s.conn.Subscriptions(); after != before {
		t.Errorf("%d subscriptions before the wait, %d after it", before, after)
	}
}

// Its timeout is four minutes when none is given and ten at most: a longer one, or none at all, is refused
// with the bound, never shortened in silence.
func TestMeshWaitsTimeoutIsBounded(t *testing.T) {
	s, _ := waitingRuntime(t)
	if WaitDefault != 4*time.Minute || WaitLongest != 10*time.Minute {
		t.Errorf("the bounds are %s and %s, want 4m and 10m (ADR 0367 §1)", WaitDefault, WaitLongest)
	}
	for _, timeout := range []any{601, 0, -5, "soon"} {
		text, isErr := callWait(s, map[string]any{"subjects": []any{"mesh-knowledge.reviewed"}, "timeout": timeout})
		if !isErr || !strings.Contains(text, "600") {
			t.Errorf("a timeout of %v: %s", timeout, text)
		}
	}
}

// novox/hq ADR 0367 §2 and §6: an event outside the granted families is refused, naming the families and how
// one is added; no argument — a wildcard, another asker's `decided`, a subject spelled out — widens it.
func TestMeshWaitRefusesAnEventOutsideTheGrantedFamilies(t *testing.T) {
	s, _ := waitingRuntime(t)
	for _, subjects := range []any{
		[]any{"mesh-knowledge.proposed"},
		[]any{"mesh-knowledge.reviewed", "gitea.forge.push"},
		[]any{"mesh-knowledge.*"},
		[]any{"operator-channel.decided.shop"},
		[]any{"operator-channel.decided.*"},
		[]any{"mesh.mod.mesh-knowledge.event.reviewed"},
		[]any{">"},
		[]any{},
	} {
		text, isErr := callWait(s, map[string]any{"subjects": subjects, "timeout": 1})
		if !isErr {
			t.Errorf("%v was waited on: %s", subjects, text)
			continue
		}
		if len(subjects.([]any)) > 0 && (!strings.Contains(text, "mesh-knowledge.reviewed") || !strings.Contains(text, "manifest")) {
			t.Errorf("%v refused without naming the families and how one is added: %s", subjects, text)
		}
	}
}

// A tool runner issued no families waits on nothing, and says so.
func TestMeshWaitWithNoFamiliesIssuedWaitsOnNothing(t *testing.T) {
	mt.New(t)
	s := NewSurface(connect(t, "node-tools", "desk"), "desk.node-tools")
	text, isErr := callWait(s, map[string]any{"subjects": []any{"mesh-knowledge.reviewed"}, "timeout": 1})
	if !isErr || !strings.Contains(text, "no event families") {
		t.Errorf("a runtime issued no families: %s", text)
	}
}

// The mesh MCP server announces seven tools (ADR 0367 v13): ADR 0195's five, mesh_runtimes and mesh_wait, its timeout bounded.
func TestMeshWaitIsAnnouncedAmongSeven(t *testing.T) {
	var names []string
	var wait map[string]any
	for _, tool := range discovery() {
		names = append(names, tool["name"].(string))
		if tool["name"] == verbWait {
			wait = tool
		}
	}
	sort.Strings(names)
	want := []string{"mesh_call", "mesh_describe", "mesh_machine", "mesh_overview", "mesh_runtimes", "mesh_search", "mesh_wait"}
	if !reflect.DeepEqual(names, want) {
		t.Errorf("announced %v, want %v", names, want)
	}
	if wait == nil || !isDiscovery(verbWait) {
		t.Fatal("mesh_wait is not announced")
	}
	schema, _ := json.Marshal(wait["inputSchema"])
	for _, word := range []string{`"subjects"`, `"filter"`, `"timeout"`} {
		if !strings.Contains(string(schema), word) {
			t.Errorf("mesh_wait's schema has no %s: %s", word, schema)
		}
	}
}

// A client that hangs up or cancels ends its wait at once: the subscription is closed and nothing is left
// waiting for the timeout (the review of issue 745). Over the console's own HTTP, as an agent host calls it.
func TestMeshWaitEndsWhenItsClientHangsUp(t *testing.T) {
	s, _ := waitingRuntime(t)
	up, err := Serve(s, "127.0.0.1:0")
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = up.Close() })
	before := s.conn.Subscriptions()
	ctx, cancel := context.WithCancel(context.Background())
	body := `{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"mesh_wait",` +
		`"arguments":{"subjects":["mesh-knowledge.reviewed"],"timeout":600}}}`
	req, _ := http.NewRequestWithContext(ctx, http.MethodPost, "http://"+up.Address+"/mcp", strings.NewReader(body))
	req.Header.Set("content-type", "application/json")
	go func() { _, _ = http.DefaultClient.Do(req) }()
	mt.Until(t, func() error {
		if s.conn.Subscriptions() <= before {
			return os.ErrNotExist
		}
		return nil
	})
	cancel()
	start := time.Now()
	for s.conn.Subscriptions() > before {
		if time.Since(start) > 3*time.Second {
			t.Fatalf("%d subscriptions 3s after the client hung up, %d before the wait", s.conn.Subscriptions(), before)
		}
		time.Sleep(20 * time.Millisecond)
	}
}

// The tool runner stopping ends every wait in flight at once (the second review of issue 745): Close cancels
// the requests' context rather than waiting up to ten minutes for a wait to time out.
func TestMeshWaitEndsWhenTheConsoleCloses(t *testing.T) {
	s, _ := waitingRuntime(t)
	up, err := Serve(s, "127.0.0.1:0")
	if err != nil {
		t.Fatal(err)
	}
	before := s.conn.Subscriptions()
	body := `{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"mesh_wait",` +
		`"arguments":{"subjects":["mesh-knowledge.reviewed"],"timeout":600}}}`
	go func() { _, _ = http.Post("http://"+up.Address+"/mcp", "application/json", strings.NewReader(body)) }()
	mt.Until(t, func() error {
		if s.conn.Subscriptions() <= before {
			return os.ErrNotExist
		}
		return nil
	})
	closed := make(chan error, 1)
	start := time.Now()
	go func() { closed <- up.Close() }()
	select {
	case <-closed:
	case <-time.After(5 * time.Second):
		t.Fatal("closing the console waited on a mesh_wait in flight")
	}
	if took := time.Since(start); took > 3*time.Second {
		t.Errorf("closing took %s", took)
	}
	mt.Until(t, func() error {
		if s.conn.Subscriptions() > before {
			return os.ErrExist
		}
		return nil
	})
}