From 21d038d3397c20ab74074a07b5f35f934831678a Mon Sep 17 00:00:00 2001 From: mesh-tools Date: Thu, 15 Oct 2026 00:47:59 +0000 Subject: [PATCH] End every wait in flight when the tool runner stops (issue 745) Closing the console shut its server down with a context that never ends, so a stop or restart of the tool runner waited on any mesh_wait in flight for up to ten minutes. The server's base context is now cancelled first, ending every wait at once, with a five-second shutdown as a backstop (second review of issue 745). Co-Authored-By: Claude Opus 5.5 --- merge-check.sh | 4 +- node-tools/internal/bus/bus.go | 8 + node-tools/internal/bus/wait.go | 132 +++++++++ node-tools/internal/bus/wait_test.go | 83 ++++++ node-tools/internal/console/address.go | 18 +- node-tools/internal/console/address_test.go | 6 +- node-tools/internal/console/http.go | 27 +- .../internal/console/manifest_waits_test.go | 40 +++ node-tools/internal/console/mcp.go | 12 +- node-tools/internal/console/wait.go | 177 +++++++++++ node-tools/internal/console/wait_test.go | 279 ++++++++++++++++++ node-tools/module.json | 7 + 12 files changed, 775 insertions(+), 18 deletions(-) create mode 100644 node-tools/internal/bus/wait.go create mode 100644 node-tools/internal/bus/wait_test.go create mode 100644 node-tools/internal/console/manifest_waits_test.go create mode 100644 node-tools/internal/console/wait.go create mode 100644 node-tools/internal/console/wait_test.go diff --git a/merge-check.sh b/merge-check.sh index 873b60a..5d7e7e3 100644 --- a/merge-check.sh +++ b/merge-check.sh @@ -32,8 +32,8 @@ if ! command -v node >/dev/null 2>&1 || [ ! -d node_modules/@novox/mesh-sdk ]; t packages=$(printf '%s\n' "$packages" | grep -v -e '/internal/runtime$' -e '/internal/console$') # What of the console launches no bundle runs anyway: how an address resolves (novox/hq issue 287), # and how a search ranks (novox/hq ADR 0245), and what discovery heard and how a round that heard nothing is - # answered (novox/hq issue 554). - bundleless='^(TestASeatAndAModuleOfOneNameAreEachReached|TestSearch.*|TestASearchByCommandSaysWhatItReplaces|TestARuntimeThatAnswersLateIsFoundAndOneThatNeverDoesIsNamed|TestARuntimeHeardBeforeAndNotNowIsNamed|TestACallAfterARoundThatHeardNothingUsesWhatWasHeardBefore|TestAnAddressKnownOnlyFromAnEarlierAnswerSaysHowOld)$' + # answered (novox/hq issue 554), and how a session waits on the bus (novox/hq ADR 0367). + bundleless='^(TestMeshWait.*|TestTheManifestNamesTheFirstFamiliesOneByOne|TestASeatAndAModuleOfOneNameAreEachReached|TestSearch.*|TestASearchByCommandSaysWhatItReplaces|TestARuntimeThatAnswersLateIsFoundAndOneThatNeverDoesIsNamed|TestARuntimeHeardBeforeAndNotNowIsNamed|TestACallAfterARoundThatHeardNothingUsesWhatWasHeardBefore|TestAnAddressKnownOnlyFromAnEarlierAnswerSaysHowOld)$' # And what of the runtime launches no bundle: how a bundle's seat traffic reaches the bus, what its # environment holds, what a runtime on a module's own account serves (novox/hq ADR 0259), and that a write # by compare-and-set loses no update under contention (novox/hq research 042), and a module's log verbs diff --git a/node-tools/internal/bus/bus.go b/node-tools/internal/bus/bus.go index 76b4183..84b3aa6 100644 --- a/node-tools/internal/bus/bus.go +++ b/node-tools/internal/bus/bus.go @@ -76,6 +76,9 @@ type Membership struct { // Hears are the module streams the module hears, each through a durable consumer on that stream the // runtime binds beside the one on EVENTS (ADR 0363 §8). Hears []HeardIssued `json:"hears,omitempty"` + // Waits are the event families a session may wait on through the mesh MCP server (novox/hq ADR 0367 §2): + // the tool runner's alone, composed from its manifest with the grant that lets it subscribe them. + Waits []WaitIssued `json:"waits,omitempty"` } // Served is an address a tool is answered on; `{tool}` stands for the tool's name. @@ -181,6 +184,8 @@ type Conn struct { RetryAfter []time.Duration // calls are the caller-named calls in flight (caller.go, novox/hq issue 365). calls callers + // waits are the waits in flight by subject, each told when the bus refuses its subscription (wait.go). + waits map[string]map[chan string]bool } // answered is one subject the runtime answers on: what to subscribe again with, and how often it was @@ -218,6 +223,9 @@ func (c *Conn) refused(err error) { if c.publishRefused(err) { return } + if subject := subjectRefused(err); subject != "" && c.waitRefused(subject) { + return + } m := refusedSubject.FindStringSubmatch(err.Error()) if len(m) < 2 { return diff --git a/node-tools/internal/bus/wait.go b/node-tools/internal/bus/wait.go new file mode 100644 index 0000000..4dda03b --- /dev/null +++ b/node-tools/internal/bus/wait.go @@ -0,0 +1,132 @@ +package bus + +import ( + "context" + "encoding/json" + "fmt" + "strings" + "time" + + "github.com/nats-io/nats.go" +) + +// Waiting on the bus for one event (novox/hq ADR 0367): what the mesh MCP server's `mesh_wait` does. +// +// **A subscription for the call, and nothing kept** (§2): no durable consumer, no queue, no stream. It +// subscribes the subjects when called and unsubscribes them before it answers, so an event published +// between two calls is not heard — a session reads the state from its owner after every wake (§4). + +// WaitIssued is one event family the mesh issued this account to wait on (ADR 0367 §2): the name a session +// asks for, and the subject it lands on. Only the tool runner's membership carries any. +type WaitIssued struct { + Event string `json:"event"` + Subject string `json:"subject"` +} + +// Heard is one event a wait heard: the subject it arrived on, its body and when it arrived. +type Heard struct { + Subject string + Body json.RawMessage + At time.Time +} + +// HearFirst subscribes the subjects for this call alone and answers the first message whose body match +// accepts (every one, when match is nil), or ok false once `within` has passed. Every subscription is +// closed before it returns. A subscription the bus refuses is said as an error naming the subject, at once; +// a caller gone — ctx done — ends the wait at once with ctx's error, so nothing waits on for nobody. +func (c *Conn) HearFirst(ctx context.Context, subjects []string, within time.Duration, match func(json.RawMessage) bool) (heard Heard, ok bool, err error) { + if len(subjects) == 0 { + return Heard{}, false, fmt.Errorf("a wait names at least one subject") + } + got := make(chan Heard, 1) + refused := make(chan string, len(subjects)) + var subs []*nats.Subscription + defer func() { + for _, s := range subs { + _ = s.Unsubscribe() + } + c.mu.Lock() + for _, subject := range subjects { + delete(c.waits[subject], refused) + if len(c.waits[subject]) == 0 { + delete(c.waits, subject) + } + } + c.mu.Unlock() + }() + c.mu.Lock() + if c.waits == nil { + c.waits = map[string]map[chan string]bool{} + } + for _, subject := range subjects { + if c.waits[subject] == nil { + c.waits[subject] = map[chan string]bool{} + } + c.waits[subject][refused] = true + } + c.mu.Unlock() + for _, subject := range subjects { + sub, err := c.nc.Subscribe(subject, func(msg *nats.Msg) { + body := json.RawMessage(msg.Data) + if !json.Valid(body) { + quoted, _ := json.Marshal(string(msg.Data)) + body = quoted + } + if match != nil && !match(body) { + return + } + select { + case got <- Heard{Subject: msg.Subject, Body: body, At: time.Now().UTC()}: + default: + } + }) + if err != nil { + return Heard{}, false, fmt.Errorf("could not wait on %s: %w", subject, err) + } + subs = append(subs, sub) + } + // The bus has every subscription — or has said which it refused — once this returns. + if err := c.nc.FlushTimeout(5 * time.Second); err != nil { + return Heard{}, false, fmt.Errorf("the bus did not take the wait: %w", err) + } + timer := time.NewTimer(within) + defer timer.Stop() + select { + case h := <-got: + return h, true, nil + case subject := <-refused: + return Heard{}, false, fmt.Errorf("the bus refused this account a subscription on %s: it is not in the "+ + "account's wait grant (novox/hq ADR 0367 §2)", subject) + case <-timer.C: + return Heard{}, false, nil + case <-ctx.Done(): + return Heard{}, false, ctx.Err() + } +} + +// waitRefused tells every wait on a subject the bus refused that it was refused; true when one was waiting. +func (c *Conn) waitRefused(subject string) bool { + c.mu.Lock() + defer c.mu.Unlock() + told := false + for ch := range c.waits[subject] { + select { + case ch <- subject: + default: + } + told = true + } + return told +} + +// Subscriptions is how many subscriptions this connection holds now: what a wait leaves behind is none. +func (c *Conn) Subscriptions() int { return c.nc.NumSubscriptions() } + +// subjectRefused is the subject a subscription refusal names, or "". +func subjectRefused(err error) string { + m := refusedSubject.FindStringSubmatch(err.Error()) + if len(m) < 2 { + return "" + } + return strings.Trim(m[1], `"`) +} diff --git a/node-tools/internal/bus/wait_test.go b/node-tools/internal/bus/wait_test.go new file mode 100644 index 0000000..45d3b5a --- /dev/null +++ b/node-tools/internal/bus/wait_test.go @@ -0,0 +1,83 @@ +package bus_test + +import ( + "context" + "encoding/json" + "errors" + "strings" + "testing" + "time" + + "github.com/nats-io/nats-server/v2/server" + + "github.com/novox/mesh-tools/node-tools/internal/bus" + mt "github.com/novox/mesh-tools/node-tools/internal/meshtest" +) + +// novox/hq ADR 0367 §2, §5: the bus refuses the tool runner's account any subject its wait grant does not +// name — another asker's `decided` among them — and a wait on one says the bus refused it, at once. +func TestTheBusRefusesAWaitOutsideTheGrant(t *testing.T) { + granted := "mesh.seat.operator-channel.event.decided.claude-code" + s, 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{ + Publish: &server.SubjectPermission{Allow: []string{"$JS.API.DIRECT.GET.>", "mesh.>"}}, + Subscribe: &server.SubjectPermission{Allow: []string{granted, "_INBOX.desk.node-tools.>", "mesh.assignment.desk.*"}}, + }}}}) + if err != nil { + t.Fatal(err) + } + go s.Start() + if !s.ReadyForConnections(10 * time.Second) { + t.Fatal("the bus did not come up") + } + t.Cleanup(s.Shutdown) + c := dial(t, s.ClientURL(), "node-tools", "desk", "desk.node-tools") + + start := time.Now() + _, ok, err := c.HearFirst(context.Background(), []string{"mesh.seat.operator-channel.event.decided.shop"}, 5*time.Second, nil) + if ok || err == nil || !strings.Contains(err.Error(), "refused") { + t.Errorf("a wait on another asker's decided: heard %v, %v", ok, err) + } + if time.Since(start) > 3*time.Second { + t.Errorf("the refusal took %s to be said", time.Since(start)) + } + + done := make(chan bus.Heard, 1) + go func() { + h, ok, err := c.HearFirst(context.Background(), []string{granted}, 5*time.Second, func(body json.RawMessage) bool { return true }) + if ok && err == nil { + done <- h + } + }() + time.Sleep(300 * time.Millisecond) + if err := c.Say(granted, []byte(`{"ask":"a1"}`)); err != nil { + t.Fatal(err) + } + select { + case h := <-done: + if h.Subject != granted || string(h.Body) != `{"ask":"a1"}` { + t.Errorf("heard %+v", h) + } + case <-time.After(5 * time.Second): + t.Fatal("a wait on the granted subject heard nothing") + } +} + +// A wait whose caller is gone ends at once, its subscription closed (the review of issue 745). +func TestAWaitEndsWhenItsCallerIsGone(t *testing.T) { + c := dial(t, mt.URL(t), "node-tools", "desk", "") + before := c.Subscriptions() + ctx, cancel := context.WithCancel(context.Background()) + time.AfterFunc(200*time.Millisecond, cancel) + start := time.Now() + _, ok, err := c.HearFirst(ctx, []string{"mesh.mod.mesh-knowledge.event.reviewed"}, time.Minute, nil) + if ok || !errors.Is(err, context.Canceled) { + t.Errorf("a cancelled wait answered %v, %v", ok, err) + } + if took := time.Since(start); took > 2*time.Second { + t.Errorf("a cancelled wait ended after %s", took) + } + if after := c.Subscriptions(); after != before { + t.Errorf("%d subscriptions before, %d after", before, after) + } +} diff --git a/node-tools/internal/console/address.go b/node-tools/internal/console/address.go index 73ef7c7..584d648 100644 --- a/node-tools/internal/console/address.go +++ b/node-tools/internal/console/address.go @@ -2,8 +2,8 @@ package console // The mesh's tools found by address, not announced whole (novox/hq ADR 0195, to-be 34 §3a). // -// The console announces six tools — five to find and call, and mesh_runtimes to say which runtimes -// discovery heard. Everything the mesh answers is reached through them by one address per layer: +// The console announces seven tools — five to find and call, mesh_runtimes to say which runtimes +// discovery heard, and mesh_wait to wait on the bus for an event (novox/hq ADR 0367). Everything the mesh answers is reached through them by one address per layer: // // . a seat held once for the mesh — its holder answers // /. a seat held once per machine — that machine's holder answers @@ -17,6 +17,7 @@ package console // session — so a tool that arrived a minute ago is found without the client reconnecting. import ( + "context" "encoding/json" "fmt" "sort" @@ -32,7 +33,7 @@ import ( // one agent turn's search, describe and call ask once, short enough that nothing goes stale. var IndexKept = 5 * time.Second -// The six tools the console announces. Names of the API's kind — letters, digits, `_`, `-`. +// The seven tools the console announces. Names of the API's kind — letters, digits, `_`, `-`. const ( verbOverview = "mesh_overview" verbMachine = "mesh_machine" @@ -40,6 +41,7 @@ const ( verbDescribe = "mesh_describe" verbCall = "mesh_call" verbRuntimes = "mesh_runtimes" + verbWait = "mesh_wait" ) // searchCap is how many matches a search answers before it says how many more there were. @@ -50,7 +52,7 @@ const grammar = "Addresses: `.` for a seat held once for the mesh (e "`/.` for a module on one machine (e.g. `novox/postgres.postgres_list_databases`); " + "and `.` also for a module whose instances are interchangeable." -// discovery is the six tools as tools/list announces them. +// discovery is the seven tools as tools/list announces them. func discovery() []map[string]any { str := func(desc string) map[string]any { return map[string]any{"type": "string", "description": desc} } obj := func(props map[string]any, required ...string) map[string]any { @@ -84,12 +86,13 @@ func discovery() []map[string]any { "arrive, its size in bytes, how many modules and tools it announced, whether it was shortened to fit the " + "bus, and when it was last heard — and every runtime or machine expected that was not heard. Read-only; " + "for when an address is said to be missing or on another machine."}, + waitTool(), } } func isDiscovery(name string) bool { switch name { - case verbOverview, verbMachine, verbSearch, verbDescribe, verbCall, verbRuntimes: + case verbOverview, verbMachine, verbSearch, verbDescribe, verbCall, verbRuntimes, verbWait: return true } return false @@ -815,10 +818,13 @@ func failure(text string) map[string]any { } // discover answers one of the discovery verbs. -func (s *Surface) discover(name string, args map[string]any) map[string]any { +func (s *Surface) discover(ctx context.Context, name string, args map[string]any) map[string]any { if name == verbRuntimes { return s.runtimesAnswer() } + if name == verbWait { + return s.waitAnswer(ctx, args) + } x, err := s.index() if err != nil { return failure("the mesh's discovery failed: " + err.Error()) diff --git a/node-tools/internal/console/address_test.go b/node-tools/internal/console/address_test.go index 60b6903..6a333b7 100644 --- a/node-tools/internal/console/address_test.go +++ b/node-tools/internal/console/address_test.go @@ -36,7 +36,7 @@ func call(t *testing.T, endpoint, tool string, args map[string]any) (string, boo return text(t, post(t, endpoint, string(body))) } -// novox/hq ADR 0195: the console announces six tools, and everything the mesh answers is reached +// novox/hq ADR 0195 and ADR 0367: the console announces seven tools, and everything the mesh answers is reached // through them by one address per layer. func TestTheMeshsToolsAreFoundByAddress(t *testing.T) { was := IndexKept @@ -113,7 +113,7 @@ func TestTheMeshsToolsAreFoundByAddress(t *testing.T) { defer up.Close() endpoint := "http://" + up.Address + "/mcp" - // Six tools, nothing else. + // Seven tools, nothing else (ADR 0367). listed := post(t, endpoint, `{"jsonrpc":"2.0","id":2,"method":"tools/list"}`)["result"].(map[string]any) var names []string for _, x := range listed["tools"].([]any) { @@ -125,7 +125,7 @@ func TestTheMeshsToolsAreFoundByAddress(t *testing.T) { } } } - if got := strings.Join(names, ","); got != "mesh_overview,mesh_machine,mesh_search,mesh_describe,mesh_call,mesh_runtimes" { + if got := strings.Join(names, ","); got != "mesh_overview,mesh_machine,mesh_search,mesh_describe,mesh_call,mesh_runtimes,mesh_wait" { t.Errorf("announced %s", got) } diff --git a/node-tools/internal/console/http.go b/node-tools/internal/console/http.go index 40e19d0..e8be4c5 100644 --- a/node-tools/internal/console/http.go +++ b/node-tools/internal/console/http.go @@ -10,6 +10,7 @@ import ( "net/http" "strconv" "strings" + "time" "github.com/novox/mesh-tools/node-tools/internal/wire" ) @@ -45,12 +46,30 @@ func Serve(s *Surface, listen string) (*Listening, error) { if err != nil { return nil, err } - server := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { route(w, r, s) })} + // Every request's context ends when the console is closed, so a mesh_wait in flight ends at once rather + // than holding a stop or a restart for up to its ten minutes (the review of issue 745). + base, cancel := context.WithCancel(context.Background()) + server := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { route(w, r, s) }), + BaseContext: func(net.Listener) context.Context { return base }} go func() { _ = server.Serve(ln) }() bound := ln.Addr().(*net.TCPAddr).Port - return &Listening{Address: host + ":" + strconv.Itoa(bound), Close: func() error { return server.Shutdown(context.Background()) }}, nil + return &Listening{Address: host + ":" + strconv.Itoa(bound), Close: func() error { + cancel() + // A backstop: what is still answering after the cancel is cut off, never waited on for long. + ctx, done := context.WithTimeout(context.Background(), ShutdownWithin) + defer done() + err := server.Shutdown(ctx) + if errors.Is(err, context.DeadlineExceeded) { + return server.Close() + } + return err + }}, nil } +// ShutdownWithin is how long closing the console waits for the requests still being answered after their +// context was cancelled. +var ShutdownWithin = 5 * time.Second + func route(w http.ResponseWriter, r *http.Request, s *Surface) { switch r.URL.Path { case "/": @@ -89,7 +108,7 @@ func route(w http.ResponseWriter, r *http.Request, s *Surface) { } replies := []*Reply{} for _, req := range batch { - if reply := s.Handle(req); reply != nil { + if reply := s.HandleContext(r.Context(), req); reply != nil { replies = append(replies, reply) } } @@ -105,7 +124,7 @@ func route(w http.ResponseWriter, r *http.Request, s *Surface) { writeJSON(w, 400, map[string]any{"jsonrpc": "2.0", "id": nil, "error": map[string]any{"code": -32700, "message": "the body is not JSON"}}) return } - reply := s.Handle(req) + reply := s.HandleContext(r.Context(), req) if reply == nil { w.WriteHeader(202) return diff --git a/node-tools/internal/console/manifest_waits_test.go b/node-tools/internal/console/manifest_waits_test.go new file mode 100644 index 0000000..2426644 --- /dev/null +++ b/node-tools/internal/console/manifest_waits_test.go @@ -0,0 +1,40 @@ +package console + +import ( + "encoding/json" + "os" + "reflect" + "strings" + "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. +func TestTheManifestNamesTheFirstFamiliesOneByOne(t *testing.T) { + raw, err := os.ReadFile("../../module.json") + if err != nil { + t.Fatal(err) + } + var m struct { + Module string `json:"module"` + Waits []string `json:"waits"` + Consumes []string `json:"consumes"` + } + if err := json.Unmarshal(raw, &m); err != nil { + t.Fatal(err) + } + want := []string{"mesh-knowledge.reviewed", "mesh-delivery.transition", "gitea.forge.status", + "mesh-controller.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) + } + } + if len(m.Consumes) > 0 { + t.Errorf("the tool runner consumes %v: a wait is no consumer (ADR 0367 §2)", m.Consumes) + } +} diff --git a/node-tools/internal/console/mcp.go b/node-tools/internal/console/mcp.go index 062d52c..70c5c76 100644 --- a/node-tools/internal/console/mcp.go +++ b/node-tools/internal/console/mcp.go @@ -4,6 +4,7 @@ package console import ( + "context" "encoding/json" "os" "strings" @@ -103,7 +104,11 @@ func idOrNull(id json.RawMessage) json.RawMessage { } // Handle answers one request; nil for a notification, which expects none. -func (s *Surface) Handle(r Request) *Reply { +func (s *Surface) Handle(r Request) *Reply { return s.HandleContext(context.Background(), r) } + +// HandleContext answers one request for as long as ctx lasts: a wait whose client hung up or cancelled +// ends at once, its subscription closed (the review of issue 745). +func (s *Surface) HandleContext(ctx context.Context, r Request) *Reply { notification := isNotification(r.ID) switch r.Method { case "initialize": @@ -164,7 +169,7 @@ func (s *Surface) Handle(r Request) *Reply { if p.Arguments == nil { p.Arguments = map[string]any{} } - return answer(r.ID, s.discover(p.Name, p.Arguments)) + return answer(r.ID, s.discover(ctx, p.Name, p.Arguments)) } args := map[string]any{} for k, v := range p.Arguments { @@ -305,7 +310,8 @@ func (s *Surface) instructions() string { "whole (novox/hq ADR 0195). mesh_overview shows the mesh's seats and machines; mesh_machine one " + "machine's seats and modules; mesh_search finds a tool by words; mesh_describe gives one tool's " + "arguments; mesh_call calls it; mesh_runtimes says which machines' runtimes answered discovery, how fast " + - "and how much — ask it when an address is said to be missing or on another machine. " + grammar + " What may be called was fixed when this account " + + "and how much — ask it when an address is said to be missing or on another machine; mesh_wait waits on the " + + "bus for an event instead of polling (novox/hq ADR 0367). " + grammar + " What may be called was fixed when this account " + "was issued, so a refusal means the account, not the tool." } diff --git a/node-tools/internal/console/wait.go b/node-tools/internal/console/wait.go new file mode 100644 index 0000000..40449ec --- /dev/null +++ b/node-tools/internal/console/wait.go @@ -0,0 +1,177 @@ +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 +// controller 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. +// **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" + "sort" + "strings" + "time" +) + +// 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 +) + +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"}, + "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)"}, + "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)."} +} + +// 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") + } + + 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 + } + } + 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") + } + 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 + 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, ", "))) + } + if _, twice := family[subject]; !twice { + 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) }) + if err != nil { + return failure("mesh_wait could not wait: " + 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)", + }) +} + +// waitNames is the families a call names: 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-knowledge.reviewed; %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 family in `subjects`, such as mesh-knowledge.reviewed") + } + 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) +} diff --git a/node-tools/internal/console/wait_test.go b/node-tools/internal/console/wait_test.go new file mode 100644 index 0000000..40c5a0c --- /dev/null +++ b/node-tools/internal/console/wait_test.go @@ -0,0 +1,279 @@ +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 + }) +} diff --git a/node-tools/module.json b/node-tools/module.json index 1f5737b..8c02041 100644 --- a/node-tools/module.json +++ b/node-tools/module.json @@ -5,6 +5,13 @@ "invokes": [ "*" ], + "waits": [ + "mesh-knowledge.reviewed", + "mesh-delivery.transition", + "gitea.forge.status", + "mesh-controller.checked", + "operator-channel.decided.claude-code" + ], "provides": [ { "name": "mcp-endpoint",