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-sdk/go/buscompose"
)

// 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)))
	}
	var subjects []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())
		}
		for _, subject := range subs {
			if !slices.Contains(subjects, subject) {
				subjects = append(subjects, subject)
			}
		}
	}

	heard, ok, refused, err := s.conn.HearFirst(ctx, subjects, timeout, func(body json.RawMessage) bool { return bodyMatches(body, filter) })
	if err != nil {
		return failure("mesh_wait could not wait: " + err.Error())
	}
	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"}
	} 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}
	}
	// A subscription the bus refused while others stood is said (design 34 every-event §16).
	if len(refused) > 0 {
		answer["refused"] = refused
	}
	return answerText(answer)
}

// 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)
}