Merge branch 'feat/47-G-fix11-a3' into feat/47-G-join11
This commit is contained in:
@@ -24,6 +24,7 @@ import (
|
||||
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
@@ -103,12 +104,13 @@ func discovery() []map[string]any {
|
||||
}), "description": "Ask the tool runner's four first views again, now, and say what each answers. A hand act, recorded in " +
|
||||
"the runner's log with its reason; for when a picture is doubted, or after an owner's state was made again."},
|
||||
waitTool(),
|
||||
approvalsTool(),
|
||||
}
|
||||
}
|
||||
|
||||
func isDiscovery(name string) bool {
|
||||
switch name {
|
||||
case verbOverview, verbMachine, verbSearch, verbDescribe, verbCall, verbRuntimes, verbFirstViews, verbFirstAgain, verbWait:
|
||||
case verbOverview, verbMachine, verbSearch, verbDescribe, verbCall, verbRuntimes, verbFirstViews, verbFirstAgain, verbWait, verbApprovals:
|
||||
return true
|
||||
}
|
||||
return false
|
||||
@@ -925,6 +927,21 @@ func schemaWithoutNode(raw json.RawMessage) map[string]any {
|
||||
}
|
||||
|
||||
// answerText is a discovery verb's answer as MCP content: JSON, indented.
|
||||
// callFailed is why a call failed, as mesh_call says it: a marked tool or verb that reaches no responder names `withheld`
|
||||
// and the grants part's read (design 46 7-approval-mark §6), and a target known only from an earlier round says so.
|
||||
func (s *Surface) callFailed(t target, err error) string {
|
||||
why := whyItFailed(t.Address, err)
|
||||
if ok, _ := s.markedTool(t); ok && errors.Is(err, bus.ErrNoResponders) {
|
||||
// A marked subject withheld from a machine whose tool runner does not honour the mark (§6).
|
||||
why += "; it needs the operator's approval, so it may be withheld where the machine's tool runner does not state " +
|
||||
"approval-mark/1: mesh-grants.grants_user_list names it under withheld"
|
||||
}
|
||||
if t.Stale != nil {
|
||||
why += " (" + t.Stale.say() + ")"
|
||||
}
|
||||
return why
|
||||
}
|
||||
|
||||
func answerText(v any) map[string]any {
|
||||
var b strings.Builder
|
||||
enc := json.NewEncoder(&b)
|
||||
@@ -949,6 +966,9 @@ func (s *Surface) discover(ctx context.Context, name string, args map[string]any
|
||||
if name == verbWait {
|
||||
return s.waitAnswer(ctx, args)
|
||||
}
|
||||
if name == verbApprovals {
|
||||
return s.approvalsAnswer(args)
|
||||
}
|
||||
x, err := s.index()
|
||||
if err != nil {
|
||||
return failure("the mesh's discovery failed: " + err.Error())
|
||||
@@ -1109,6 +1129,10 @@ func (s *Surface) discover(ctx context.Context, name string, args map[string]any
|
||||
if t.Stale != nil {
|
||||
described["found"] = t.Stale.say()
|
||||
}
|
||||
if ok, source := s.markedTool(t); ok {
|
||||
described["needs-approval"] = map[string]any{"source": source,
|
||||
"says": "the first call asks the operator and runs nothing; the same call again within 30 minutes of the Approve runs once"}
|
||||
}
|
||||
return answerText(described)
|
||||
|
||||
case verbCall:
|
||||
@@ -1128,11 +1152,7 @@ func (s *Surface) discover(ctx context.Context, name string, args map[string]any
|
||||
got, err = callTool(s.conn, t.Key, callArgs, nil, x.Listing)
|
||||
}
|
||||
if err != nil {
|
||||
why := whyItFailed(t.Address, err)
|
||||
if t.Stale != nil {
|
||||
why += " (" + t.Stale.say() + ")"
|
||||
}
|
||||
return failure(why)
|
||||
return failure(s.callFailed(t, err))
|
||||
}
|
||||
result, unread := resolveSent(s.conn, got.Result)
|
||||
content := []map[string]any{{"type": "text", "text": pretty(result)}}
|
||||
|
||||
@@ -0,0 +1,99 @@
|
||||
package console
|
||||
|
||||
import (
|
||||
"log"
|
||||
"strings"
|
||||
|
||||
"github.com/novox/mesh-tools/node-tools/internal/approval"
|
||||
)
|
||||
|
||||
// verbApprovals is the mesh MCP server's tool that reads and steers what the gate asked on this machine (design 46
|
||||
// 7-approval-mark §18).
|
||||
const verbApprovals = "mesh_approvals"
|
||||
|
||||
// Approvals is the gate of this machine's tool runner, as mesh_approvals and mesh_describe read it.
|
||||
type Approvals interface {
|
||||
List() ([]approval.Listed, error)
|
||||
Read(id string) (approval.Whole, error)
|
||||
Applied() []approval.Marks
|
||||
Describe(module, seat, name string) (bool, string)
|
||||
Again(id string) error
|
||||
Cancel(id string) error
|
||||
// Fill is the state approvals' bytes against its cap, every machine's keys together.
|
||||
Fill() (bytes uint64, max int64, err error)
|
||||
}
|
||||
|
||||
func approvalsTool() map[string]any {
|
||||
str := func(desc string) map[string]any { return map[string]any{"type": "string", "description": desc} }
|
||||
return map[string]any{"name": verbApprovals, "inputSchema": map[string]any{"type": "object", "properties": map[string]any{
|
||||
"ask": str("an ask's id: its whole arguments and the act's digest"),
|
||||
"record": str("an ask's id: ask mesh-hand-acts.warranted again for its run, recorded in the hand-act log first"),
|
||||
"cancel": str("an ask's id: take that open ask back, recorded in the hand-act log first"),
|
||||
"why": str("why, for record and cancel, kept in the hand-act log"),
|
||||
}}, "description": "The calls that needed the operator's approval on this machine: with no argument, those asked in " +
|
||||
"the last day, newest first, with their state and fingerprint, the marks the tool runner applies (each module's " +
|
||||
"marked tools and verbs with their source and the catalog revision read) and its state's fill; `ask` reads one " +
|
||||
"whole, `record` asks its hand-act again, `cancel` takes an open ask back."}
|
||||
}
|
||||
|
||||
// approvalsAnswer is mesh_approvals.
|
||||
func (s *Surface) approvalsAnswer(args map[string]any) map[string]any {
|
||||
if s.Approvals == nil {
|
||||
return failure("this tool runner has no approval gate")
|
||||
}
|
||||
str := func(k string) string { v, _ := args[k].(string); return strings.TrimSpace(v) }
|
||||
why := str("why")
|
||||
if why == "" {
|
||||
why = "asked by hand with no reason given"
|
||||
}
|
||||
act := func(id, what, cause string, do func(string) error) map[string]any {
|
||||
// A hand act, recorded through the seat mesh-hand-acts before it is done (§18); one that cannot be recorded is refused.
|
||||
log.Printf("[mesh-tools] hand act: %s %s ask %s: %s", s.who, what, id, why)
|
||||
if err := s.recordAct(map[string]any{"what": what + " ask " + id, "why": why, "cause": cause, "verb": verbApprovals}); err != nil {
|
||||
return failure("the hand act was not recorded through mesh-hand-acts, so nothing was done: " + err.Error())
|
||||
}
|
||||
if err := do(id); err != nil {
|
||||
return failure(err.Error())
|
||||
}
|
||||
return answerText(map[string]any{"done": what + " ask " + id, "recorded": "mesh-hand-acts.hand-act"})
|
||||
}
|
||||
switch {
|
||||
case str("record") != "":
|
||||
return act(str("record"), "asked the hand-act of", "approval-record", s.Approvals.Again)
|
||||
case str("cancel") != "":
|
||||
return act(str("cancel"), "cancelled", "approval-cancel", s.Approvals.Cancel)
|
||||
case str("ask") != "":
|
||||
w, err := s.Approvals.Read(str("ask"))
|
||||
if err != nil {
|
||||
return failure(err.Error())
|
||||
}
|
||||
return answerText(w)
|
||||
}
|
||||
asks, err := s.Approvals.List()
|
||||
out := map[string]any{"asks": asks, "marks": s.Approvals.Applied()}
|
||||
if err != nil {
|
||||
out["asks"], out["asks_why"] = []approval.Listed{}, "the state approvals cannot be read: "+err.Error()
|
||||
}
|
||||
if b, max, err := s.Approvals.Fill(); err != nil {
|
||||
out["fill_why"] = err.Error()
|
||||
} else {
|
||||
out["fill"] = map[string]any{"bytes": b, "max_bytes": max, "of": "every machine's keys together"}
|
||||
}
|
||||
return answerText(out)
|
||||
}
|
||||
|
||||
// marked is what mesh_describe and a call that reached no one say of a marked tool or verb (§6, §18).
|
||||
func (s *Surface) marked(module, seat, name string) (bool, string) {
|
||||
if s.Approvals == nil {
|
||||
return false, ""
|
||||
}
|
||||
return s.Approvals.Describe(module, seat, name)
|
||||
}
|
||||
|
||||
// markedTool is marked for a resolved address.
|
||||
func (s *Surface) markedTool(t target) (bool, string) {
|
||||
if t.Seat {
|
||||
return s.marked("", t.Tool.Module, t.Tool.Name)
|
||||
}
|
||||
return s.marked(t.Tool.Module, "", t.Tool.Name)
|
||||
}
|
||||
@@ -8,19 +8,28 @@ import (
|
||||
"testing"
|
||||
)
|
||||
|
||||
// novox/hq ADR 0367 §5: the tool runner's manifest names the families a session waits on today, one by one,
|
||||
// and the asker's `decided` for the claude-code module alone. Adding one is a reviewed change to this file
|
||||
// (§6), so this list changes with it. The delivery's transitions are waited on under the gate's name alone (ADR 0450 §8, issues 1350
|
||||
// and 1406): today's module is forgotten.
|
||||
func TestTheManifestNamesTheFirstFamiliesOneByOne(t *testing.T) {
|
||||
// The tool runner's manifest names no events to wait on (novox/hq ADR 0670 §2): a session waits on any event by its
|
||||
// name. It holds each change of design 34 every-event §11 (E-1): no `waits`, `operator-channel` in `uses` for the asks of
|
||||
// marked waits, the per-machine state `approvals` in the old `state` section with its 48 hours and none in `requires`
|
||||
// (design 46 7-approval-mark §13), and the finding stream (design 47 G §387).
|
||||
func TestTheManifestNamesNoWaitsAndKeepsItsApprovals(t *testing.T) {
|
||||
raw, err := os.ReadFile("../../module.json")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var keys map[string]json.RawMessage
|
||||
if err := json.Unmarshal(raw, &keys); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, ok := keys["waits"]; ok {
|
||||
t.Error("the manifest still carries waits (novox/hq ADR 0670 §2)")
|
||||
}
|
||||
var m struct {
|
||||
Module string `json:"module"`
|
||||
Waits []string `json:"waits"`
|
||||
Consumes []string `json:"consumes"`
|
||||
Module string `json:"module"`
|
||||
Uses []string `json:"uses"`
|
||||
State []json.RawMessage `json:"state"`
|
||||
Requires []json.RawMessage `json:"requires"`
|
||||
Consumes []string `json:"consumes"`
|
||||
Streams []struct {
|
||||
Name string `json:"name"`
|
||||
Events []string `json:"events"`
|
||||
@@ -29,20 +38,33 @@ func TestTheManifestNamesTheFirstFamiliesOneByOne(t *testing.T) {
|
||||
if err := json.Unmarshal(raw, &m); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if m.Module != "node-tools" || !reflect.DeepEqual(m.Uses, []string{"operator-channel"}) {
|
||||
t.Errorf("%s uses %v, want operator-channel (design 46 7-approval-mark §7)", m.Module, m.Uses)
|
||||
}
|
||||
type state struct {
|
||||
Name string `json:"name"`
|
||||
PerMachine bool `json:"per-machine"`
|
||||
TTLSeconds int `json:"ttl-seconds"`
|
||||
}
|
||||
var approvals []state
|
||||
for _, r := range m.State {
|
||||
var s state
|
||||
if json.Unmarshal(r, &s) == nil && s.Name == "approvals" {
|
||||
approvals = append(approvals, s)
|
||||
}
|
||||
}
|
||||
if want := []state{{Name: "approvals", PerMachine: true, TTLSeconds: 172800}}; !reflect.DeepEqual(approvals, want) {
|
||||
t.Errorf("the old state section's approvals is %+v, want %+v (design 46 7-approval-mark §13)", approvals, want)
|
||||
}
|
||||
for _, r := range m.Requires {
|
||||
if strings.Contains(string(r), `"approvals"`) {
|
||||
t.Errorf("approvals is asked in requires too: %s", r)
|
||||
}
|
||||
}
|
||||
if len(m.Streams) != 1 || m.Streams[0].Name != "node-tools-findings" ||
|
||||
!reflect.DeepEqual(m.Streams[0].Events, []string{"finding-written.*", "finding-cleared.*", "findings-stated.*"}) {
|
||||
t.Errorf("the finding stream is %+v, want node-tools-findings with the three finding events (design 47 G §387)", m.Streams)
|
||||
}
|
||||
want := []string{"mesh-knowledge.reviewed", "mesh-delivery-gate.transition", "gitea.forge.status",
|
||||
"mesh-self-check.checked", "operator-channel.decided.claude-code"}
|
||||
if m.Module != "node-tools" || !reflect.DeepEqual(m.Waits, want) {
|
||||
t.Errorf("%s waits on %v, want %v", m.Module, m.Waits, want)
|
||||
}
|
||||
for _, w := range m.Waits {
|
||||
if strings.ContainsAny(w, "*>") {
|
||||
t.Errorf("a family named by a wildcard: %s", w)
|
||||
}
|
||||
}
|
||||
// A wait is no consumer (ADR 0367 §2): what it consumes is the assigner's two views families (ADR 0615 §5, issue 1966
|
||||
// TLS-5), the events of the catalog half's branch follower (design 47 G §196, §614: `catalog-revision`,
|
||||
// `registration-changed`, and the gate's `source-moved`; never the indexer's `views.modules` or `views.manifests`), the two
|
||||
|
||||
@@ -63,6 +63,8 @@ type Surface struct {
|
||||
Flat bool
|
||||
// Record, when set, records a hand act in place of the seat mesh-hand-acts' verb (a test's).
|
||||
Record func(args map[string]any) error
|
||||
// Approvals is this machine's approval gate, read by mesh_approvals and mesh_describe; nil when the runner has none.
|
||||
Approvals Approvals
|
||||
}
|
||||
|
||||
// MethodSeats is the console's JSON-RPC method beside MCP's that answers every seat's verbs and holders, in the shape
|
||||
|
||||
@@ -11,18 +11,33 @@ import (
|
||||
"github.com/novox/mesh-tools/node-tools/internal/bus"
|
||||
)
|
||||
|
||||
// A tool's own error that mentions 503 is the tool's answer, not a routing fault (issue 1529).
|
||||
func TestAToolsOwn503IsNotNothingServes(t *testing.T) {
|
||||
// A marked tool or verb that reaches no responder names withheld and the grants part's read (design 46 7-approval-mark §6);
|
||||
// an unmarked one, or a marked one failing otherwise, does not. A tool's own error that mentions 503 is the tool's answer, not a routing fault (issue 1529); a real no-responders still says
|
||||
// nothing serves.
|
||||
func TestARealNoRespondersStillSaysNothingServes(t *testing.T) {
|
||||
got := whyItFailed("scanner.scan", errors.New("the scanner refused the scan job: 503 Service Unavailable"))
|
||||
if strings.Contains(got, "nothing serves") || !strings.HasPrefix(got, "scanner.scan failed: ") || !strings.Contains(got, "503") {
|
||||
t.Fatalf("got %q", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestARealNoRespondersStillSaysNothingServes(t *testing.T) {
|
||||
for _, err := range []error{bus.ErrNoResponders, fmt.Errorf("ask: %w", bus.ErrNoResponders), nats.ErrNoResponders} {
|
||||
if got := whyItFailed("a.b", err); !strings.HasPrefix(got, "nothing serves a.b") {
|
||||
t.Fatalf("%v: got %q", err, got)
|
||||
}
|
||||
}
|
||||
s := &Surface{Approvals: &fakeApprovals{marked: "shell.wipe"}}
|
||||
for _, c := range []struct {
|
||||
target target
|
||||
err error
|
||||
hinted bool
|
||||
}{
|
||||
{target{Address: "shell.wipe", Tool: Tool{Module: "shell", Name: "wipe"}}, bus.ErrNoResponders, true},
|
||||
{target{Address: "node-login-shell.execute", Seat: true, Tool: Tool{Module: "node-login-shell", Name: "execute"}}, bus.ErrNoResponders, false},
|
||||
{target{Address: "shell.tidy", Tool: Tool{Module: "shell", Name: "tidy"}}, bus.ErrNoResponders, false},
|
||||
{target{Address: "shell.wipe", Tool: Tool{Module: "shell", Name: "wipe"}}, errors.New("its own failure"), false},
|
||||
} {
|
||||
got := s.callFailed(c.target, c.err)
|
||||
if hinted := strings.Contains(got, "approval-mark/1: mesh-grants.grants_user_list names it under withheld"); hinted != c.hinted {
|
||||
t.Errorf("%s (%v): %q", c.target.Address, c.err, got)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2,9 +2,9 @@ package console
|
||||
|
||||
// mesh_wait: a session waits on the bus for one event instead of polling (novox/hq ADR 0367).
|
||||
//
|
||||
// **Only the families the tool runner was issued** (§2, §6): its membership names them, composed by the
|
||||
// mesh from the tool runner's manifest together with the grant that lets this account subscribe them.
|
||||
// A name not listed there is refused here, before the bus would refuse it; no argument or setting adds one.
|
||||
// **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 (
|
||||
@@ -13,9 +13,15 @@ import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"math"
|
||||
"sort"
|
||||
"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"
|
||||
|
||||
"github.com/novox/mesh-tools/node-tools/internal/approval"
|
||||
)
|
||||
|
||||
// WaitDefault is how long a wait lasts when no timeout is given, and WaitLongest the most it may last
|
||||
@@ -25,20 +31,29 @@ var (
|
||||
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": "one or more event families to wait on, by name, e.g. `mesh-knowledge.reviewed`; only " +
|
||||
"those the mesh granted this machine's tool runner — a refusal names them"},
|
||||
"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} (optional)"},
|
||||
"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 families that matches the filter, and answer it with its " +
|
||||
"family, body and time — 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)."}
|
||||
"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.
|
||||
@@ -56,51 +71,195 @@ func (s *Surface) waitAnswer(ctx context.Context, args map[string]any) map[strin
|
||||
return failure("mesh_wait's filter is an object of body fields to the values they must equal")
|
||||
}
|
||||
|
||||
issued := map[string]string{} // name → subject
|
||||
if m := s.conn.Membership(s.conn.Module()); m != nil {
|
||||
for _, w := range m.Waits {
|
||||
issued[w.Event] = w.Subject
|
||||
}
|
||||
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(issued) == 0 {
|
||||
return failure("this machine's tool runner was issued no event families to wait on: the mesh grants them from " +
|
||||
"the tool runner's manifest (novox/hq ADR 0367 §2), and its membership names none yet")
|
||||
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)))
|
||||
}
|
||||
byName := make([]string, 0, len(issued))
|
||||
for name := range issued {
|
||||
byName = append(byName, name)
|
||||
}
|
||||
sort.Strings(byName)
|
||||
family := map[string]string{} // subject → name
|
||||
var subjects []string
|
||||
known, unchecked := s.emitters()
|
||||
var subjects, undeclared []string
|
||||
for _, name := range names {
|
||||
subject, granted := issued[name]
|
||||
if !granted {
|
||||
return failure(fmt.Sprintf("mesh_wait may not wait on %q: this machine's tool runner is granted %s alone. "+
|
||||
"A family is added only by a reviewed change to the tool runner's manifest (node-tools' module.json, "+
|
||||
"novox/hq ADR 0367 §6); no argument, call or setting adds one", name, strings.Join(byName, ", ")))
|
||||
subs, err := buscompose.EventSubjects(name, from)
|
||||
if err != nil {
|
||||
return failure("mesh_wait may not wait on it: " + err.Error())
|
||||
}
|
||||
if _, twice := family[subject]; !twice {
|
||||
subjects = append(subjects, subject)
|
||||
// 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)
|
||||
}
|
||||
}
|
||||
family[subject] = name
|
||||
}
|
||||
|
||||
heard, ok, err := s.conn.HearFirst(ctx, subjects, timeout, func(body json.RawMessage) bool { return bodyMatches(body, filter) })
|
||||
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("mesh_wait could not wait: " + err.Error())
|
||||
return failure(err.Error())
|
||||
}
|
||||
if !ok {
|
||||
return answerText(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"})
|
||||
}
|
||||
return answerText(map[string]any{
|
||||
"heard": map[string]any{"subject": family[heard.Subject], "body": heard.Body, "at": heard.At.Format(time.RFC3339Nano)},
|
||||
"read": "an event only wakes you: read the state from the verb that owns it before acting (novox/hq ADR 0367 §4)",
|
||||
})
|
||||
return answerText(answer)
|
||||
}
|
||||
|
||||
// waitNames is the families a call names: a list of names, or one name.
|
||||
// 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) {
|
||||
@@ -110,7 +269,7 @@ func waitNames(given any) ([]string, error) {
|
||||
for _, x := range v {
|
||||
name, ok := x.(string)
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("mesh_wait's subjects are event names, such as mesh-knowledge.reviewed; %v is not one", x)
|
||||
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)
|
||||
}
|
||||
@@ -119,7 +278,7 @@ func waitNames(given any) ([]string, error) {
|
||||
names[i] = strings.TrimSpace(names[i])
|
||||
}
|
||||
if len(names) == 0 {
|
||||
return nil, fmt.Errorf("mesh_wait names at least one event family in `subjects`, such as mesh-knowledge.reviewed")
|
||||
return nil, fmt.Errorf("mesh_wait names at least one event in `subjects`, such as mesh-delivery-view.delivered")
|
||||
}
|
||||
return names, nil
|
||||
}
|
||||
|
||||
@@ -3,32 +3,40 @@ package console
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"os"
|
||||
"reflect"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient"
|
||||
opseat "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
|
||||
catalogseat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/seatdef"
|
||||
"github.com/nats-io/nats-server/v2/server"
|
||||
"github.com/nats-io/nats.go"
|
||||
|
||||
"github.com/novox/mesh-tools/node-tools/internal/approval"
|
||||
"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.
|
||||
// waitingRuntime is the tool runner on a machine, issued no waits (novox/hq ADR 0670 §2): 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)
|
||||
mesh.Issue(t, mt.MembershipOf("node-tools", "desk", false, nil))
|
||||
conn := connect(t, "node-tools", "desk")
|
||||
mt.Until(t, func() error {
|
||||
if got := conn.Membership("node-tools"); got == nil || len(got.Waits) == 0 {
|
||||
if conn.Membership("node-tools") == nil {
|
||||
return os.ErrNotExist
|
||||
}
|
||||
return nil
|
||||
@@ -36,6 +44,25 @@ func waitingRuntime(t *testing.T) (*Surface, *mt.Mesh) {
|
||||
return NewSurface(conn, busSource{conn}, "desk.node-tools"), mesh
|
||||
}
|
||||
|
||||
// halfSource is a Source that also holds a catalog half, as the tool runner's first views do.
|
||||
type halfSource struct {
|
||||
Source
|
||||
halves map[string]meshcatalogclient.Half
|
||||
}
|
||||
|
||||
func (h halfSource) Halves() (map[string]meshcatalogclient.Half, error) { return h.halves, nil }
|
||||
|
||||
// listening waits until the surface's connection holds n more subscriptions than before: a wait has subscribed.
|
||||
func listening(t *testing.T, s *Surface, before, n int) {
|
||||
t.Helper()
|
||||
mt.Until(t, func() error {
|
||||
if s.conn.Subscriptions() < before+n {
|
||||
return os.ErrNotExist
|
||||
}
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
// 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})
|
||||
@@ -50,8 +77,9 @@ func callWait(s *Surface, args map[string]any) (text string, 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.
|
||||
// novox/hq ADR 0367 §1, ADR 0670 §1 (design 34 every-event §13, §15, §19): mesh_wait answers the first event that
|
||||
// matches — by its name and its filter — with no membership naming it: its whole name, the namespace it arrived on, its
|
||||
// body, when it arrived and when it was published, within a second of its publication, and ignores the others.
|
||||
func TestMeshWaitAnswersTheFirstMatchingEvent(t *testing.T) {
|
||||
s, mesh := waitingRuntime(t)
|
||||
type answer struct {
|
||||
@@ -59,18 +87,39 @@ func TestMeshWaitAnswersTheFirstMatchingEvent(t *testing.T) {
|
||||
isErr bool
|
||||
at time.Time
|
||||
}
|
||||
type heardAnswer struct {
|
||||
Heard struct {
|
||||
Subject string `json:"subject"`
|
||||
From string `json:"from"`
|
||||
Body map[string]any `json:"body"`
|
||||
At time.Time `json:"at"`
|
||||
Published *string `json:"published"`
|
||||
} `json:"heard"`
|
||||
Read string `json:"read"`
|
||||
}
|
||||
nc, err := nats.Connect(mt.URL(t))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(nc.Close)
|
||||
done := make(chan answer, 1)
|
||||
before := s.conn.Subscriptions()
|
||||
go func() {
|
||||
text, isErr := callWait(s, map[string]any{"subjects": []any{"mesh-knowledge.reviewed"},
|
||||
"filter": map[string]any{"kind": "decision", "number": 367}, "timeout": 20})
|
||||
text, isErr := callWait(s, map[string]any{"subjects": []any{"mesh-delivery-view.delivered"},
|
||||
"filter": map[string]any{"module": "netcheck"}, "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})
|
||||
// Once it listens: another module's delivery, another event, then the one it waits for, with its own time.
|
||||
listening(t, s, before, 6)
|
||||
mesh.Emit(t, "mesh.mod.mesh-delivery-view.event.delivered", map[string]any{"module": "grafana"})
|
||||
mesh.Emit(t, "mesh.mod.mesh-delivery-view.event.planned", map[string]any{"module": "netcheck"})
|
||||
msg := nats.NewMsg("mesh.mod.mesh-delivery-view.event.delivered")
|
||||
msg.Header.Set("x-time", "2026-11-12T03:00:00.5Z")
|
||||
msg.Data = []byte(`{"module":"netcheck"}`)
|
||||
published := time.Now()
|
||||
mesh.Emit(t, "mesh.mod.mesh-knowledge.event.reviewed", map[string]any{"kind": "decision", "number": 367, "what": "approved"})
|
||||
if err := nc.PublishMsg(msg); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var got answer
|
||||
select {
|
||||
case got = <-done:
|
||||
@@ -83,32 +132,32 @@ func TestMeshWaitAnswersTheFirstMatchingEvent(t *testing.T) {
|
||||
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"`
|
||||
}
|
||||
var heard heardAnswer
|
||||
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)
|
||||
if h := heard.Heard; h.Subject != "mesh-delivery-view.delivered" || h.From != "module" || h.Body["module"] != "netcheck" ||
|
||||
h.At.IsZero() || h.Published == nil || *h.Published != "2026-11-12T03:00:00.5Z" || !strings.Contains(heard.Read, "only wakes you") {
|
||||
t.Errorf("answered %s, want netcheck's delivery from a module with its times", got.text)
|
||||
}
|
||||
// And the subject it is granted on the asker's own `decided`, by the name a session asks for.
|
||||
// And a seat's `decided` of one asker, heard by `operator-channel.decided`: its whole name, from the seat, with no
|
||||
// time of its own, and the answer read only with claude_code_answer.
|
||||
before = s.conn.Subscriptions()
|
||||
go func() {
|
||||
text, isErr := callWait(s, map[string]any{"subjects": "operator-channel.decided.claude-code", "timeout": 20})
|
||||
text, isErr := callWait(s, map[string]any{"subjects": "operator-channel.decided", "timeout": 20})
|
||||
done <- answer{text, isErr, time.Now()}
|
||||
}()
|
||||
time.Sleep(300 * time.Millisecond)
|
||||
listening(t, s, before, 6)
|
||||
// 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"`) {
|
||||
heard = heardAnswer{}
|
||||
_ = json.Unmarshal([]byte(got.text), &heard)
|
||||
if h := heard.Heard; got.isErr || h.Subject != "operator-channel.decided.claude-code" || h.From != "seat" ||
|
||||
h.Published != nil || !strings.Contains(heard.Read, "claude_code_answer") {
|
||||
t.Errorf("the asker's decided: %s", got.text)
|
||||
}
|
||||
case <-time.After(10 * time.Second):
|
||||
@@ -125,7 +174,8 @@ func TestMeshWaitSaysItTimedOutAndKeepsNothing(t *testing.T) {
|
||||
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`) {
|
||||
// Before the catalog half is read, names are not checked, and the answer says so (design 34 every-event §14).
|
||||
if isErr || !strings.Contains(text, `"timed-out": true`) || !strings.Contains(text, `"unchecked"`) {
|
||||
t.Errorf("a wait that heard nothing: %s", text)
|
||||
}
|
||||
if after := s.conn.Subscriptions(); after != before {
|
||||
@@ -148,41 +198,313 @@ func TestMeshWaitsTimeoutIsBounded(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// 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{},
|
||||
// novox/hq ADR 0670 §1, §2 (design 34 every-event §13, §16, §19): mesh_wait refuses loudly and never loops: a name with
|
||||
// a wildcard, of one token, more than 16 names, none, or a from of another word is refused at once naming the rule; a
|
||||
// wait whose every subscription the bus refuses answers after its timeout, in words naming no wait grant.
|
||||
func TestMeshWaitRefusesLoudlyAndNeverLoops(t *testing.T) {
|
||||
s, mesh := waitingRuntime(t)
|
||||
var seventeen []any
|
||||
for i := 0; i < 17; i++ {
|
||||
seventeen = append(seventeen, fmt.Sprintf("mesh-knowledge.reviewed%d", i))
|
||||
}
|
||||
for _, c := range []struct {
|
||||
args map[string]any
|
||||
says string
|
||||
}{
|
||||
{map[string]any{"subjects": []any{"mesh-knowledge.*"}}, "[A-Za-z0-9_-]+"},
|
||||
{map[string]any{"subjects": []any{"operator-channel.decided.>"}}, "[A-Za-z0-9_-]+"},
|
||||
{map[string]any{"subjects": []any{"mesh-knowledge"}}, "fewer than two tokens"},
|
||||
{map[string]any{"subjects": seventeen}, "at most 16"},
|
||||
{map[string]any{"subjects": []any{}}, "at least one event"},
|
||||
{map[string]any{"subjects": []any{"mesh-knowledge.reviewed"}, "from": "bench"}, "seat, module or stream"},
|
||||
} {
|
||||
text, isErr := callWait(s, map[string]any{"subjects": subjects, "timeout": 1})
|
||||
if !isErr {
|
||||
t.Errorf("%v was waited on: %s", subjects, text)
|
||||
continue
|
||||
c.args["timeout"] = 1
|
||||
start := time.Now()
|
||||
text, isErr := callWait(s, c.args)
|
||||
if !isErr || !strings.Contains(text, c.says) || strings.Contains(text, "manifest") {
|
||||
t.Errorf("%v: %s, want refused saying %q", c.args, text, c.says)
|
||||
}
|
||||
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)
|
||||
if took := time.Since(start); took > 500*time.Millisecond {
|
||||
t.Errorf("%v was refused after %s, want at once", c.args, took)
|
||||
}
|
||||
}
|
||||
|
||||
// E §14: against the catalog half, a misspelt emitter is refused naming the half and its revision, a known emitter's
|
||||
// undeclared event is waited on and its timeout says so, and a declared one says nothing of it.
|
||||
half := NewSurface(s.conn, halfSource{busSource{s.conn}, map[string]meshcatalogclient.Half{"main": {
|
||||
Revision: catalogseat.CatalogRevision{Revision: 7, Definitions: []catalogseat.CatalogRevisionDefinitionsItem{{Seat: "mesh-catalog"}}},
|
||||
Entries: map[string]catalogseat.CatalogEntry{"mesh-delivery-view": {Module: "mesh-delivery-view",
|
||||
Manifest: map[string]any{"emits": []any{"delivered"}}}},
|
||||
}}}, "desk.node-tools")
|
||||
for _, c := range []struct {
|
||||
name, says string
|
||||
refused bool
|
||||
}{
|
||||
{"mesh-delivry-view.delivered", "mesh-delivry-view is neither a seat nor a module of the catalog half (mesh-catalog.revision, revision 7 of main)", true},
|
||||
{"mesh-delivery-view.planned", "no declaration names mesh-delivery-view.planned", false},
|
||||
{"mesh-catalog.catalog-revision", "", false},
|
||||
} {
|
||||
text, isErr := callWait(half, map[string]any{"subjects": []any{c.name}, "timeout": 1})
|
||||
if isErr != c.refused || c.says != "" && !strings.Contains(text, c.says) ||
|
||||
c.says == "" && (strings.Contains(text, "undeclared") || strings.Contains(text, "unchecked")) {
|
||||
t.Errorf("%s against the catalog half: %s", c.name, text)
|
||||
}
|
||||
}
|
||||
|
||||
// The bus refusing every subscription of the call: answered after the timeout, in §16's words.
|
||||
b, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: server.RANDOM_PORT, NoLog: true, NoSigs: true,
|
||||
Users: []*server.User{{Username: "desk.node-tools", Password: "pw", Permissions: &server.Permissions{
|
||||
Subscribe: &server.SubjectPermission{Allow: []string{"_INBOX.desk.node-tools.>", "mesh.assignment.desk.*"}},
|
||||
}}}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
go b.Start()
|
||||
if !b.ReadyForConnections(10 * time.Second) {
|
||||
t.Fatal("the bus did not come up")
|
||||
}
|
||||
t.Cleanup(b.Shutdown)
|
||||
conn, err := bus.Connect(bus.Credential{URL: b.ClientURL(), Module: "node-tools", Node: "desk", User: "desk.node-tools", Password: "pw"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
conn.Logf = func(string, ...any) {}
|
||||
t.Cleanup(conn.Close)
|
||||
start := time.Now()
|
||||
text, isErr := callWait(NewSurface(conn, nil, "desk.node-tools"), map[string]any{"subjects": []any{"mesh-delivery-view.delivered"}, "timeout": 1})
|
||||
if took := time.Since(start); took < time.Second || took > 3*time.Second {
|
||||
t.Errorf("a wait the bus refused whole answered after %s, want its timeout of 1s", took)
|
||||
}
|
||||
if !isErr || !strings.Contains(text, "the bus refused this account a subscription on mesh.seat.mesh-delivery-view.event.delivered") ||
|
||||
!strings.Contains(text, "novox/hq ADR 0670 §1") || strings.Contains(text, "wait grant") || strings.Contains(text, "0367 §2") {
|
||||
t.Errorf("a wait the bus refused whole: %s", text)
|
||||
}
|
||||
// Design 34 every-event §37: a wait whose subjects overlap a marked event's, or a stream's with a `grant`, runs only on the
|
||||
// operator's warrant for that one wait; with none it subscribes nothing, asks once and answers at its timeout or the choice.
|
||||
markedWaits(t, s, mesh)
|
||||
}
|
||||
|
||||
// markedWaits are §37's rows, against a gate on the test bus, the router's records written there as the router writes them.
|
||||
func markedWaits(t *testing.T, s *Surface, mesh *mt.Mesh) {
|
||||
gw := newGatedWaits(t, s, mesh, "src2:runner")
|
||||
g := gw.surface
|
||||
took := func(args map[string]any) (string, bool, time.Duration) {
|
||||
start := time.Now()
|
||||
text, isErr := callWait(g, args)
|
||||
return text, isErr, time.Since(start)
|
||||
}
|
||||
// A fixture module's marked event, no warrant: nothing of it subscribed (only its ask's decided), asked once, §12's
|
||||
// words at its timeout; the same wait again asks nothing new.
|
||||
before := g.conn.Subscriptions()
|
||||
go func() {
|
||||
listening(t, g, before, 1)
|
||||
mesh.Emit(t, "mesh.mod.mail.event.message", map[string]any{"n": 1})
|
||||
}()
|
||||
text, isErr, d := took(map[string]any{"subjects": []any{"mail.message"}, "timeout": 1})
|
||||
if !isErr || !strings.Contains(text, "this wait needs the operator's approval. Asked on telegram") || gw.asks() != 1 ||
|
||||
d < time.Second || strings.Contains(text, "heard") {
|
||||
t.Errorf("a marked event's wait, no warrant: %s after %s, %d asks", text, d, gw.asks())
|
||||
}
|
||||
if text, isErr, _ = took(map[string]any{"subjects": []any{"mail.message"}, "timeout": 1}); !isErr ||
|
||||
!strings.Contains(text, "still waiting") || gw.asks() != 1 {
|
||||
t.Errorf("the same marked wait again: %s, %d asks", text, gw.asks())
|
||||
}
|
||||
// A stream with a `grant` is marked whole; a name with the node after a marked event's name is marked.
|
||||
for i, name := range []string{"mail.inbox", "mail.message.desk"} {
|
||||
if text, isErr, _ := took(map[string]any{"subjects": []any{name}, "timeout": 1}); !isErr || !strings.Contains(text, "Not run") || gw.asks() != 2+i {
|
||||
t.Errorf("%s: %s, %d asks", name, text, gw.asks())
|
||||
}
|
||||
}
|
||||
// The same names from a seat ask nothing.
|
||||
if text, isErr, _ := took(map[string]any{"subjects": []any{"mail.message"}, "from": "seat", "timeout": 1}); isErr ||
|
||||
!strings.Contains(text, `"timed-out": true`) || gw.asks() != 3 {
|
||||
t.Errorf("from a seat: %s, %d asks", text, gw.asks())
|
||||
}
|
||||
// The most timeout a wait takes, its ask chosen while it waits: answered at the choice; the same call made then
|
||||
// subscribes and hears the event published after it, answered with its approval; a warrant for other names, another
|
||||
// filter or another timeout is refused, asked anew.
|
||||
long := map[string]any{"subjects": []any{"mail.message"}, "filter": map[string]any{"n": 2}, "timeout": 600}
|
||||
done := make(chan string, 1)
|
||||
go func() { text, _, d := took(long); done <- fmt.Sprintf("%s after %s", text, d) }()
|
||||
mt.Until(t, func() error {
|
||||
if gw.asks() < 4 {
|
||||
return os.ErrNotExist
|
||||
}
|
||||
return nil
|
||||
})
|
||||
gw.approve(t)
|
||||
select {
|
||||
case got := <-done:
|
||||
if !strings.Contains(got, "make this wait again within 30 minutes") {
|
||||
t.Errorf("a long wait chosen while it waits: %s", got)
|
||||
}
|
||||
case <-time.After(10 * time.Second):
|
||||
t.Fatal("a long marked wait was not answered at the choice")
|
||||
}
|
||||
before = g.conn.Subscriptions()
|
||||
go func() {
|
||||
listening(t, g, before, 6)
|
||||
mesh.Emit(t, "mesh.mod.mail.event.message", map[string]any{"n": 2})
|
||||
}()
|
||||
if text, isErr, _ := took(long); isErr || !strings.Contains(text, `"approval": "run on the operator's approval`) ||
|
||||
!strings.Contains(text, `"subject": "mail.message"`) {
|
||||
t.Errorf("the same wait on its warrant: %s", text)
|
||||
}
|
||||
for _, other := range []map[string]any{
|
||||
{"subjects": []any{"mail.message", "mail.inbox"}, "filter": map[string]any{"n": 2}, "timeout": 1},
|
||||
{"subjects": []any{"mail.message"}, "filter": map[string]any{"n": 3}, "timeout": 1},
|
||||
{"subjects": []any{"mail.message"}, "filter": map[string]any{"n": 2}, "timeout": 2},
|
||||
} {
|
||||
asked := gw.asks()
|
||||
if text, isErr, _ := took(other); !isErr || !strings.Contains(text, "Not run") || gw.asks() != asked+1 {
|
||||
t.Errorf("another wait than the warrant's %v: %s", other, text)
|
||||
}
|
||||
}
|
||||
// Before the catalog half is read nothing is asked: judged when its first view comes within the timeout, refused at
|
||||
// the timeout when it does not.
|
||||
gw.cat.mu.Lock()
|
||||
gw.cat.ready = errors.New("no first view yet")
|
||||
gw.cat.mu.Unlock()
|
||||
asked := gw.asks()
|
||||
if text, isErr, d := took(map[string]any{"subjects": []any{"mail.message"}, "timeout": 1}); !isErr ||
|
||||
!strings.Contains(text, "the marks are not yet read") || gw.asks() != asked || d < time.Second {
|
||||
t.Errorf("before the catalog half is read: %s after %s", text, d)
|
||||
}
|
||||
gw.cat.came = make(chan chan struct{})
|
||||
go func() { close(<-gw.cat.came) }()
|
||||
if text, isErr, _ := took(map[string]any{"subjects": []any{"mail.message"}, "filter": map[string]any{"n": 9}, "timeout": 1}); !isErr ||
|
||||
!strings.Contains(text, "this wait needs the operator's approval") || gw.asks() != asked+1 {
|
||||
t.Errorf("the catalog half read within the timeout: %s", text)
|
||||
}
|
||||
// While the runner's own fingerprint is unknown a marked wait is refused, saying so.
|
||||
unknown := newGatedWaits(t, s, mesh, "")
|
||||
if text, isErr := callWait(unknown.surface, map[string]any{"subjects": []any{"mail.message"}, "timeout": 1}); !isErr ||
|
||||
!strings.Contains(text, "own fingerprint is unknown") || unknown.asks() != 0 {
|
||||
t.Errorf("the runner's fingerprint unknown: %s", 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"), nil, "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)
|
||||
// gatedWaits is the surface with the runner's approval gate: the fixture module mail marks its event message and declares
|
||||
// the stream inbox with a grant; the router takes each ask as it is published, its records on the test bus.
|
||||
type gatedWaits struct {
|
||||
surface *Surface
|
||||
cat *waitCatalog
|
||||
records nats.KeyValue
|
||||
mu sync.Mutex
|
||||
asked []opseat.Ask
|
||||
say *bus.Conn
|
||||
}
|
||||
|
||||
func newGatedWaits(t *testing.T, s *Surface, mesh *mt.Mesh, build string) *gatedWaits {
|
||||
gw := &gatedWaits{cat: &waitCatalog{entries: []catalogseat.CatalogEntry{{Module: "mail", Fp: "src2:mail", Manifest: map[string]any{
|
||||
"emits": []any{"message"}, "needs-approval": []any{"message"},
|
||||
"streams": []any{map[string]any{"name": "inbox", "events": []any{"inbox.*"}, "grant": map[string]any{"subscribe": []any{"x"}}}}}}}},
|
||||
records: mesh.Bucket(t, "operator-channel-asks"), say: connect(t, "messenger", "")}
|
||||
gate := approval.New(approval.Deps{Node: "desk", Runner: "node-tools", Build: build, Catalog: gw.cat, Wait: gw.cat.wait,
|
||||
State: mt.NewState(), Holder: func(string) (bool, error) { return true, nil },
|
||||
Hear: func(ctx context.Context, subjects []string, within time.Duration, match func(json.RawMessage) bool) (bool, error) {
|
||||
_, ok, _, err := s.conn.HearFirst(ctx, subjects, within, match)
|
||||
return ok, err
|
||||
},
|
||||
Publish: func(module, subject string, body any, id string) error {
|
||||
a, ok := body.(opseat.Ask)
|
||||
if subject != opseat.AcceptSubjectAsk(module) || !ok {
|
||||
return nil
|
||||
}
|
||||
gw.mu.Lock()
|
||||
gw.asked = append(gw.asked, a)
|
||||
gw.mu.Unlock()
|
||||
gw.write(t, opseat.AsksByAskerIDValue{Asker: module, Key: module + "." + a.ID, State: "open", Headline: a.Headline,
|
||||
Expires: a.Expires, Opened: operatorchannelclient.FormatTime(time.Now()), OfferedOn: []string{"telegram"}})
|
||||
return nil
|
||||
},
|
||||
Record: func(module, id string) (*stdio.StateEntry, error) {
|
||||
e, err := gw.records.Get(module + "." + id)
|
||||
if errors.Is(err, nats.ErrKeyNotFound) {
|
||||
return nil, nil
|
||||
} else if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &stdio.StateEntry{Key: e.Key(), Value: e.Value(), Revision: e.Revision()}, nil
|
||||
}})
|
||||
gw.surface = NewSurface(s.conn, busSource{s.conn}, "desk.node-tools")
|
||||
gw.surface.Approvals = gateApprovals{gate}
|
||||
return gw
|
||||
}
|
||||
|
||||
func (gw *gatedWaits) write(t *testing.T, rec opseat.AsksByAskerIDValue) {
|
||||
raw, _ := json.Marshal(rec)
|
||||
if _, err := gw.records.Put(rec.Key, raw); err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
}
|
||||
|
||||
func (gw *gatedWaits) asks() int {
|
||||
gw.mu.Lock()
|
||||
defer gw.mu.Unlock()
|
||||
return len(gw.asked)
|
||||
}
|
||||
|
||||
// approve writes the router's record of the newest ask as a person's Approve, and says its decided.
|
||||
func (gw *gatedWaits) approve(t *testing.T) {
|
||||
gw.mu.Lock()
|
||||
a := gw.asked[len(gw.asked)-1]
|
||||
gw.mu.Unlock()
|
||||
gw.write(t, opseat.AsksByAskerIDValue{Asker: "node-tools", Key: "node-tools." + a.ID, State: "chosen", Expires: a.Expires,
|
||||
OfferedOn: []string{"telegram"}, Warrant: &opseat.Warrant{Ask: a.ID, Asker: "node-tools", Outcome: operatorchannelclient.OutcomeChosen,
|
||||
Option: approval.OptionApprove, Level: operatorchannelclient.Approve, At: operatorchannelclient.FormatTime(time.Now()),
|
||||
AskDigest: operatorchannelclient.Digest(a), Channel: "telegram",
|
||||
By: &opseat.Person{Who: operatorchannelclient.Operator, Kind: "telegram", Identity: "42", Verified: "user id verified"}}})
|
||||
if err := gw.say.Say(opseat.SubjectDecided("node-tools"), []byte(`{"ask":"`+a.ID+`"}`)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
// gateApprovals is the gate as the surface holds it: mesh_approvals' reads unused here, mesh_wait's gate.
|
||||
type gateApprovals struct{ *approval.Gate }
|
||||
|
||||
func (gateApprovals) Fill() (uint64, int64, error) { return 0, 0, nil }
|
||||
|
||||
// waitCatalog is a catalog half as a fixture: its entries, read or not; with came set, its first view comes once a wait
|
||||
// waits for it (waiting is told on it, then it closes).
|
||||
type waitCatalog struct {
|
||||
entries []catalogseat.CatalogEntry
|
||||
mu sync.Mutex
|
||||
ready error
|
||||
came chan chan struct{}
|
||||
}
|
||||
|
||||
func (c *waitCatalog) Ready() error {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
return c.ready
|
||||
}
|
||||
func (c *waitCatalog) wait(within time.Duration) error {
|
||||
if c.Ready() == nil {
|
||||
return nil
|
||||
}
|
||||
if c.came != nil {
|
||||
view := make(chan struct{})
|
||||
c.came <- view
|
||||
<-view
|
||||
c.mu.Lock()
|
||||
c.ready = nil
|
||||
c.mu.Unlock()
|
||||
return nil
|
||||
}
|
||||
<-time.After(within) // the bound of the runner's own wait for the first view
|
||||
return c.Ready()
|
||||
}
|
||||
func (c *waitCatalog) Revision() string { return "7" }
|
||||
func (c *waitCatalog) Served(string) (catalogseat.CatalogEntry, bool, string) {
|
||||
return catalogseat.CatalogEntry{}, false, "not-found"
|
||||
}
|
||||
func (c *waitCatalog) Newest(string) (catalogseat.CatalogEntry, bool) {
|
||||
return catalogseat.CatalogEntry{}, false
|
||||
}
|
||||
func (c *waitCatalog) Definitions() map[string]seatdef.Definition { return nil }
|
||||
func (c *waitCatalog) Entries() []catalogseat.CatalogEntry { return c.entries }
|
||||
|
||||
// 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
|
||||
@@ -194,7 +516,7 @@ func TestMeshWaitIsAnnouncedAmongSeven(t *testing.T) {
|
||||
}
|
||||
}
|
||||
sort.Strings(names)
|
||||
want := []string{"mesh_call", "mesh_describe", "mesh_first_view_again", "mesh_first_views", "mesh_machine", "mesh_overview", "mesh_runtimes", "mesh_search", "mesh_wait"}
|
||||
want := []string{"mesh_approvals", "mesh_call", "mesh_describe", "mesh_first_view_again", "mesh_first_views", "mesh_machine", "mesh_overview", "mesh_runtimes", "mesh_search", "mesh_wait"}
|
||||
if !reflect.DeepEqual(names, want) {
|
||||
t.Errorf("announced %v, want %v", names, want)
|
||||
}
|
||||
@@ -207,8 +529,49 @@ func TestMeshWaitIsAnnouncedAmongSeven(t *testing.T) {
|
||||
t.Errorf("mesh_wait's schema has no %s: %s", word, schema)
|
||||
}
|
||||
}
|
||||
// Design 46 7-approval-mark §18: mesh_approvals lists, and records a hand act before it cancels; refused without a record.
|
||||
gate := &fakeApprovals{}
|
||||
var recorded []string
|
||||
s := &Surface{who: "n1.node-tools", Approvals: gate, Record: func(a map[string]any) error {
|
||||
if a["why"] == "unrecordable" {
|
||||
return errors.New("mesh-hand-acts does not answer")
|
||||
}
|
||||
recorded = append(recorded, a["what"].(string))
|
||||
return nil
|
||||
}}
|
||||
if out := s.discover(context.Background(), verbApprovals, map[string]any{}); out["isError"] == true || !strings.Contains(fmtAnswer(out), "every machine") {
|
||||
t.Errorf("the list %v", out)
|
||||
}
|
||||
if out := s.discover(context.Background(), verbApprovals, map[string]any{"cancel": "a1", "why": "no longer needed"}); out["isError"] == true ||
|
||||
gate.cancelled != "a1" || len(recorded) != 1 || recorded[0] != "cancelled ask a1" {
|
||||
t.Errorf("a cancel %v, recorded %v", out, recorded)
|
||||
}
|
||||
if out := s.discover(context.Background(), verbApprovals, map[string]any{"cancel": "a2", "why": "unrecordable"}); out["isError"] != true || gate.cancelled != "a1" {
|
||||
t.Errorf("a cancel without a record %v", out)
|
||||
}
|
||||
}
|
||||
|
||||
func fmtAnswer(out map[string]any) string { raw, _ := json.Marshal(out); return string(raw) }
|
||||
|
||||
// fakeApprovals is a gate as mesh_approvals reads it.
|
||||
type fakeApprovals struct {
|
||||
cancelled string
|
||||
// marked is the one `<module>.<tool>` its Describe says is marked.
|
||||
marked string
|
||||
}
|
||||
|
||||
func (f *fakeApprovals) List() ([]approval.Listed, error) {
|
||||
return []approval.Listed{{Ask: "a1", State: "asked"}}, nil
|
||||
}
|
||||
func (f *fakeApprovals) Read(id string) (approval.Whole, error) { return approval.Whole{}, nil }
|
||||
func (f *fakeApprovals) Applied() []approval.Marks { return nil }
|
||||
func (f *fakeApprovals) Describe(module, _, name string) (bool, string) {
|
||||
return f.marked != "" && f.marked == module+"."+name, "entry " + module
|
||||
}
|
||||
func (f *fakeApprovals) Again(string) error { return nil }
|
||||
func (f *fakeApprovals) Cancel(id string) error { f.cancelled = id; return nil }
|
||||
func (f *fakeApprovals) Fill() (uint64, int64, error) { return 10, 64 << 20, nil }
|
||||
|
||||
// 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) {
|
||||
|
||||
Reference in New Issue
Block a user