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

package console

// mesh_wait: a session waits on the bus for one event instead of polling (novox/hq ADR 0367).
//
// **Any event, by its name** (novox/hq ADR 0670 §1, ADR 0671 §1; design 34 every-event §13, §26): the grants part
// grants this machine's tool runner every event subject, a seat's, a module's and a module stream's, and a name
// becomes its subjects through mesh-sdk's buscompose.EventSubjects; no membership and no manifest names a list.
// **An event only wakes the session** (§4): what it says is read again from the verb that owns it.

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"math"
	"slices"
	"strings"
	"time"

	"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
	"git.novox.be/novox/mesh-sdk/go/buscompose"
	"git.novox.be/novox/mesh-sdk/go/catalogue"

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

// WaitDefault is how long a wait lasts when no timeout is given, and WaitLongest the most it may last
// (ADR 0367 §1): four minutes and ten.
var (
	WaitDefault = 4 * time.Minute
	WaitLongest = 10 * time.Minute
)

// waitMostNames is the most event names one call gives (design 34 every-event §13, §26): each name is six
// subscriptions for the call, so 16 names are 96, far above what a watcher names.
const waitMostNames = 16

func waitTool() map[string]any {
	return map[string]any{"name": verbWait, "inputSchema": map[string]any{"type": "object", "properties": map[string]any{
		"subjects": map[string]any{"type": "array", "items": map[string]any{"type": "string"},
			"description": fmt.Sprintf("one or more event names, `<seat or module>.<event>`, such as "+
				"`mesh-delivery-view.delivered`, at most %d; an event whose subject goes on after the name (an asker, a "+
				"kind, a module) is heard by the name too", waitMostNames)},
		"from": map[string]any{"type": "string", "enum": []string{"seat", "module", "stream"},
			"description": "narrows every name to a seat's events, a module's or a module stream's, for an emitter " +
				"whose name is more than one of them (optional)"},
		"filter": map[string]any{"type": "object",
			"description": "fields of the event's body that must equal the given values, e.g. {\"kind\": \"decision\", \"number\": 367}: the filter narrows by the body's fields (optional)"},
		"timeout": map[string]any{"type": "number",
			"description": fmt.Sprintf("seconds to wait: %d when absent, %d at most", int(WaitDefault.Seconds()), int(WaitLongest.Seconds()))},
	}, "required": []string{"subjects"}},
		"description": "Wait for the first event of the named events that matches the filter, and answer it with its " +
			"name, where it came from, body and times — or say it timed out. Call it in a loop from a background command, " +
			"and after every answer read the state from the verb that owns it: an event only wakes you, and one " +
			"published between two waits is not heard (novox/hq ADR 0367). An operator's answer is read only with " +
			"claude_code_answer, never from the event."}
}

// waitAnswer answers one mesh_wait call.
func (s *Surface) waitAnswer(ctx context.Context, args map[string]any) map[string]any {
	names, err := waitNames(args["subjects"])
	if err != nil {
		return failure(err.Error())
	}
	timeout, err := waitTimeout(args["timeout"])
	if err != nil {
		return failure(err.Error())
	}
	filter, _ := args["filter"].(map[string]any)
	if args["filter"] != nil && filter == nil {
		return failure("mesh_wait's filter is an object of body fields to the values they must equal")
	}

	from, _ := args["from"].(string)
	if args["from"] != nil && from != "seat" && from != "module" && from != "stream" {
		return failure(fmt.Sprintf("mesh_wait's from is seat, module or stream, not %v", args["from"]))
	}
	if len(names) > waitMostNames {
		return failure(fmt.Sprintf("mesh_wait names at most %d events in one call, not %d: each name is six "+
			"subscriptions for the call (design 34 every-event §13)", waitMostNames, len(names)))
	}
	known, unchecked := s.emitters()
	var subjects, undeclared []string
	for _, name := range names {
		subs, err := buscompose.EventSubjects(name, from)
		if err != nil {
			return failure("mesh_wait may not wait on it: " + err.Error())
		}
		// A misspelt emitter is refused, never waited on for ever; a known emitter's undeclared event is waited on and
		// its timed-out answer says so (design 34 every-event §14).
		if unchecked == "" {
			emitter, event, _ := strings.Cut(name, ".")
			seatEvents, isSeat := known.seats[emitter]
			moduleEvents, isModule := known.modules[emitter]
			isSeat = isSeat && from != "module" && from != "stream"
			isModule = isModule && from != "seat"
			if !isSeat && !isModule {
				return failure(fmt.Sprintf("mesh_wait may not wait on %q: %s is neither a seat nor a module of %s",
					name, emitter, known.source))
			}
			if !(isSeat && declares(seatEvents, event)) && !(isModule && declares(moduleEvents, event)) {
				undeclared = append(undeclared, name)
			}
		}
		for _, subject := range subs {
			if !slices.Contains(subjects, subject) {
				subjects = append(subjects, subject)
			}
		}
	}

	listen := func(within time.Duration) (any, error) {
		return s.hear(ctx, names, subjects, undeclared, known.source, unchecked, filter, within)
	}
	// A wait whose subjects overlap a marked event's runs only on the operator's warrant for that one wait (design 34
	// every-event §34 to §36), judged by the runner's approval gate.
	var answer any
	if g, ok := s.Approvals.(waitGate); ok {
		given := map[string]any{}
		for _, k := range []string{"subjects", "from", "filter", "timeout"} {
			if v, has := args[k]; has {
				given[k] = v
			}
		}
		raw, _ := json.Marshal(given)
		answer, err = g.Waited(ctx, approval.Call{Address: verbWait, Name: verbWait, Caller: s.who, Args: raw},
			approval.Wait{Names: names, From: from, Subjects: subjects, Filter: filter, Timeout: timeout}, listen)
	} else {
		answer, err = listen(timeout)
	}
	if err != nil {
		return failure(err.Error())
	}
	return answerText(answer)
}

// waitGate is the runner's approval gate as mesh_wait goes through it (design 34 every-event §34 to §36).
type waitGate interface {
	Waited(ctx context.Context, c approval.Call, w approval.Wait, hear func(within time.Duration) (any, error)) (any, error)
}

// hear subscribes the wait's subjects for within and answers the first matching event, or that it timed out.
func (s *Surface) hear(ctx context.Context, names, subjects, undeclared []string, source, unchecked string, filter map[string]any,
	timeout time.Duration) (map[string]any, error) {
	heard, ok, refused, err := s.conn.HearFirst(ctx, subjects, timeout, func(body json.RawMessage) bool { return bodyMatches(body, filter) })
	if err != nil {
		return nil, fmt.Errorf("mesh_wait could not wait: %w", err)
	}
	var answer map[string]any
	if !ok {
		answer = map[string]any{"timed-out": true, "waited": timeout.String(), "subjects": names,
			"read": "nothing matching was heard; read the state once from the verb that owns it, then wait again"}
		if len(undeclared) > 0 {
			answer["undeclared"] = fmt.Sprintf("no declaration names %s: no seat definition and no module's entry in %s "+
				"declares that event", strings.Join(undeclared, ", "), source)
		}
	} else {
		name, from, _ := buscompose.EventName(heard.Subject)
		h := map[string]any{"subject": name, "from": from, "body": heard.Body, "at": heard.At.Format(time.RFC3339Nano)}
		if heard.Published != "" {
			h["published"] = heard.Published
		}
		read := "an event only wakes you: read the state from the verb that owns it before acting (novox/hq ADR 0367 §4)"
		if strings.HasPrefix(name, "operator-channel.decided") {
			read = "an event only wakes you: the operator's answer is read only with claude_code_answer, never from " +
				"the event's body (novox/hq ADR 0670 §5)"
		}
		answer = map[string]any{"heard": h, "read": read}
	}
	if unchecked != "" {
		answer["unchecked"] = "the names were not checked: " + unchecked
	}
	// A subscription the bus refused while others stood is said (design 34 every-event §16).
	if len(refused) > 0 {
		answer["refused"] = refused
	}
	return answer, nil
}

// halfReader is a Source that also holds the catalog half of each followed branch (firstview.Views.Halves): mesh_wait
// checks an event name's emitter against it (design 34 every-event §14).
type halfReader interface {
	Halves() (map[string]meshcatalogclient.Half, error)
}

// knownEmitters is what the catalog half says emits what: each seat's declared events and each module's, and the
// source a refusal names, with its revisions.
type knownEmitters struct {
	seats, modules map[string][]string
	source         string
}

// emitters is the catalog half's emitters, or why they are not checked: the half not yet read (its first view,
// mesh_first_views), or no half held at all.
func (s *Surface) emitters() (known knownEmitters, unchecked string) {
	hr, ok := s.src.(halfReader)
	if !ok {
		return known, "this tool runner holds no catalog half"
	}
	halves, err := hr.Halves()
	if err != nil {
		return known, "the catalog half is not yet read (" + err.Error() + "); its first view is in mesh_first_views"
	}
	seatEvents := map[string][]string{} // a seat's events, as its definition in the seat list declares them
	for _, st := range catalogue.Seats() {
		seatEvents[st.Name] = st.Emits
	}
	known.seats, known.modules = map[string][]string{}, map[string][]string{}
	branches := make([]string, 0, len(halves))
	for b := range halves {
		branches = append(branches, b)
	}
	slices.Sort(branches)
	var revisions []string
	for _, b := range branches {
		half := halves[b]
		revisions = append(revisions, fmt.Sprintf("revision %d of %s", half.Revision.Revision, b))
		for _, d := range half.Revision.Definitions {
			known.seats[d.Seat] = seatEvents[d.Seat]
		}
		for m, entry := range half.Entries {
			if _, taken := known.modules[m]; !taken {
				known.modules[m] = eventsDeclared(entry.Manifest)
			}
		}
	}
	known.source = "the catalog half (mesh-catalog.revision, " + strings.Join(revisions, "; ") + ")"
	return known, ""
}

// eventsDeclared is the events a module's manifest declares: its `emits` and its streams' `events`.
func eventsDeclared(manifest map[string]any) []string {
	var out []string
	add := func(list any) {
		items, _ := list.([]any)
		for _, x := range items {
			if e, ok := x.(string); ok {
				out = append(out, e)
			}
		}
	}
	add(manifest["emits"])
	streams, _ := manifest["streams"].([]any)
	for _, st := range streams {
		if m, ok := st.(map[string]any); ok {
			add(m["events"])
		}
	}
	return out
}

// declares says whether one of the declared events is the event a name gives, or one its name is followed by.
func declares(declared []string, event string) bool {
	for _, d := range declared {
		if buscompose.Overlaps(d, event) || buscompose.Overlaps(d, event+".>") {
			return true
		}
	}
	return false
}

// waitNames is the event names a call gives: a list of names, or one name.
func waitNames(given any) ([]string, error) {
	var names []string
	switch v := given.(type) {
	case string:
		names = []string{v}
	case []any:
		for _, x := range v {
			name, ok := x.(string)
			if !ok {
				return nil, fmt.Errorf("mesh_wait's subjects are event names, such as mesh-delivery-view.delivered; %v is not one", x)
			}
			names = append(names, name)
		}
	}
	for i := range names {
		names[i] = strings.TrimSpace(names[i])
	}
	if len(names) == 0 {
		return nil, fmt.Errorf("mesh_wait names at least one event in `subjects`, such as mesh-delivery-view.delivered")
	}
	return names, nil
}

// waitTimeout is the call's timeout: WaitDefault when absent, refused beyond WaitLongest or below a second.
func waitTimeout(given any) (time.Duration, error) {
	if given == nil {
		return WaitDefault, nil
	}
	longest := int(WaitLongest.Seconds())
	seconds, ok := given.(float64)
	if !ok {
		if n, isInt := given.(int); isInt {
			seconds, ok = float64(n), true
		}
	}
	if !ok || math.IsNaN(seconds) || seconds < 1 || seconds > float64(longest) {
		return 0, fmt.Errorf("mesh_wait's timeout is a number of seconds from 1 to %d (novox/hq ADR 0367 §1), not %v", longest, given)
	}
	return time.Duration(seconds * float64(time.Second)), nil
}

// bodyMatches says whether every field the filter names equals, in the event's body, the value given.
func bodyMatches(body json.RawMessage, filter map[string]any) bool {
	if len(filter) == 0 {
		return true
	}
	var fields map[string]json.RawMessage
	if json.Unmarshal(body, &fields) != nil {
		return false
	}
	for k, want := range filter {
		got, has := fields[k]
		if !has {
			return false
		}
		wanted, _ := json.Marshal(want)
		if !sameJSON(got, wanted) {
			return false
		}
	}
	return true
}

// sameJSON compares two JSON values as values: 367 and 367.0 are one number, key order does not matter.
func sameJSON(a, b []byte) bool {
	var x, y any
	da, db := json.NewDecoder(bytes.NewReader(a)), json.NewDecoder(bytes.NewReader(b))
	if da.Decode(&x) != nil || db.Decode(&y) != nil {
		return false
	}
	ja, _ := json.Marshal(x)
	jb, _ := json.Marshal(y)
	return bytes.Equal(ja, jb)
}