diff --git a/node-tools/internal/bus/bus.go b/node-tools/internal/bus/bus.go index 920189e..f430db6 100644 --- a/node-tools/internal/bus/bus.go +++ b/node-tools/internal/bus/bus.go @@ -75,9 +75,6 @@ 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"` // Bindings are the module's bus state in the new form, by local name (novox/hq ADR 0595 §1, mesh-sdk // membership.BusBinding): its object stores, streams, work queues and ledgers, its own alone (bindings.go). Bindings []membership.BusBinding `json:"bindings,omitempty"` diff --git a/node-tools/internal/bus/wait.go b/node-tools/internal/bus/wait.go index 4dda03b..c42a667 100644 --- a/node-tools/internal/bus/wait.go +++ b/node-tools/internal/bus/wait.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "fmt" + "slices" "strings" "time" @@ -16,38 +17,42 @@ import ( // 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. +// Heard is one event a wait heard: the subject it arrived on, its body, when it arrived, and when it was published, the +// event's own header `x-time` as its emitter wrote it, empty for an event that carries none (design 34 every-event §12). type Heard struct { - Subject string - Body json.RawMessage - At time.Time + Subject string + Body json.RawMessage + At time.Time + Published string } // 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) { +// closed before it returns. A subscription the bus refuses is noted and the others go on waiting; refused lists +// those noted by the time it answers (design 34 every-event §16). When the bus refused every subscription of the +// call, it still waits out `within` and then answers the refusal, so a caller looping on it asks once a timeout and +// never at once again. 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, refused []string, err error) { if len(subjects) == 0 { - return Heard{}, false, fmt.Errorf("a wait names at least one subject") + return Heard{}, false, nil, fmt.Errorf("a wait names at least one subject") + } + var unique []string + for _, subject := range subjects { + if !slices.Contains(unique, subject) { + unique = append(unique, subject) + } } got := make(chan Heard, 1) - refused := make(chan string, len(subjects)) + refusals := make(chan string, len(unique)) var subs []*nats.Subscription defer func() { for _, s := range subs { _ = s.Unsubscribe() } c.mu.Lock() - for _, subject := range subjects { - delete(c.waits[subject], refused) + for _, subject := range unique { + delete(c.waits[subject], refusals) if len(c.waits[subject]) == 0 { delete(c.waits, subject) } @@ -58,14 +63,14 @@ func (c *Conn) HearFirst(ctx context.Context, subjects []string, within time.Dur if c.waits == nil { c.waits = map[string]map[chan string]bool{} } - for _, subject := range subjects { + for _, subject := range unique { if c.waits[subject] == nil { c.waits[subject] = map[chan string]bool{} } - c.waits[subject][refused] = true + c.waits[subject][refusals] = true } c.mu.Unlock() - for _, subject := range subjects { + for _, subject := range unique { sub, err := c.nc.Subscribe(subject, func(msg *nats.Msg) { body := json.RawMessage(msg.Data) if !json.Valid(body) { @@ -76,31 +81,39 @@ func (c *Conn) HearFirst(ctx context.Context, subjects []string, within time.Dur return } select { - case got <- Heard{Subject: msg.Subject, Body: body, At: time.Now().UTC()}: + case got <- Heard{Subject: msg.Subject, Body: body, At: time.Now().UTC(), Published: msg.Header.Get("x-time")}: default: } }) if err != nil { - return Heard{}, false, fmt.Errorf("could not wait on %s: %w", subject, err) + return Heard{}, false, nil, 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) + return Heard{}, false, nil, 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() + for { + select { + case h := <-got: + return h, true, refused, nil + case subject := <-refusals: + if !slices.Contains(refused, subject) { + refused = append(refused, subject) + } + case <-timer.C: + if len(refused) == len(unique) { + return Heard{}, false, refused, fmt.Errorf("the bus refused this account a subscription on %s: the "+ + "grants part grants this machine's tool runner every event subject (novox/hq ADR 0670 §1), so its "+ + "grant is not composed yet; read the state from its owner meanwhile", strings.Join(refused, ", ")) + } + return Heard{}, false, refused, nil + case <-ctx.Done(): + return Heard{}, false, refused, ctx.Err() + } } } diff --git a/node-tools/internal/bus/wait_test.go b/node-tools/internal/bus/wait_test.go index 45d3b5a..c12d8d7 100644 --- a/node-tools/internal/bus/wait_test.go +++ b/node-tools/internal/bus/wait_test.go @@ -2,22 +2,24 @@ package bus_test import ( "context" - "encoding/json" "errors" "strings" "testing" "time" "github.com/nats-io/nats-server/v2/server" + "github.com/nats-io/nats.go" "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. +// novox/hq ADR 0670 §1 (design 34 every-event §12, §16): a subscription the bus refuses is noted and the others go on +// waiting, a heard event carries its own time; a wait whose every subscription is refused answers only after its +// timeout, in words naming the grants part's grant of every event subject, never a wait grant (E-2). func TestTheBusRefusesAWaitOutsideTheGrant(t *testing.T) { granted := "mesh.seat.operator-channel.event.decided.claude-code" + other := "mesh.seat.operator-channel.event.decided.shop" 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.>"}}, @@ -33,33 +35,65 @@ func TestTheBusRefusesAWaitOutsideTheGrant(t *testing.T) { t.Cleanup(s.Shutdown) c := dial(t, s.ClientURL(), "node-tools", "desk", "desk.node-tools") + // Every subscription refused: answered after the timeout, never at once, in §16's words. 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) + _, ok, refused, err := c.HearFirst(context.Background(), []string{other}, 300*time.Millisecond, nil) + if took := time.Since(start); took < 300*time.Millisecond || took > 3*time.Second { + t.Errorf("a wait refused whole answered after %s, want its timeout of 300ms", took) } - if time.Since(start) > 3*time.Second { - t.Errorf("the refusal took %s to be said", time.Since(start)) + if ok || err == nil || !strings.Contains(err.Error(), "the bus refused this account a subscription on "+other) || + !strings.Contains(err.Error(), "every event subject (novox/hq ADR 0670 §1)") || + strings.Contains(err.Error(), "wait grant") || strings.Contains(err.Error(), "0367") || + len(refused) != 1 || refused[0] != other { + t.Errorf("a wait on another asker's decided: heard %v, refused %v, %v", ok, refused, err) } - 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 { + // One refused beside one granted: the granted one hears, with the event's own time, and the refused one is listed. + pub, err := nats.Connect(s.ClientURL(), nats.UserInfo("desk.node-tools", "pw")) + if err != nil { t.Fatal(err) } - select { - case h := <-done: - if h.Subject != granted || string(h.Body) != `{"ask":"a1"}` { - t.Errorf("heard %+v", h) + t.Cleanup(pub.Close) + for _, published := range []string{"2026-11-12T03:00:00.5Z", ""} { + type result struct { + h bus.Heard + refused []string + err error + } + done := make(chan result, 1) + go func() { + h, ok, refused, err := c.HearFirst(context.Background(), []string{other, granted}, 5*time.Second, nil) + if !ok && err == nil { + err = errors.New("timed out") + } + done <- result{h, refused, err} + }() + // The event is said again until the wait, subscribed in its own time, hears it. + say := time.NewTicker(20 * time.Millisecond) + var r result + heard: + for { + select { + case r = <-done: + break heard + case <-say.C: + msg := nats.NewMsg(granted) + msg.Data = []byte(`{"ask":"a1"}`) + if published != "" { + msg.Header.Set("x-time", published) + } + if err := pub.PublishMsg(msg); err != nil { + t.Fatal(err) + } + } + } + say.Stop() + if r.err != nil || r.h.Subject != granted || string(r.h.Body) != `{"ask":"a1"}` || r.h.Published != published { + t.Errorf("heard %+v, %v; want published %q", r.h, r.err, published) + } + if len(r.refused) > 1 || len(r.refused) == 1 && r.refused[0] != other { + t.Errorf("refused %v, want at most %s", r.refused, other) } - case <-time.After(5 * time.Second): - t.Fatal("a wait on the granted subject heard nothing") } } @@ -70,7 +104,7 @@ func TestAWaitEndsWhenItsCallerIsGone(t *testing.T) { 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) + _, 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) } diff --git a/node-tools/internal/console/manifest_waits_test.go b/node-tools/internal/console/manifest_waits_test.go index f80ec5f..da628ff 100644 --- a/node-tools/internal/console/manifest_waits_test.go +++ b/node-tools/internal/console/manifest_waits_test.go @@ -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 diff --git a/node-tools/internal/console/wait.go b/node-tools/internal/console/wait.go index 9ee1a41..a93a16d 100644 --- a/node-tools/internal/console/wait.go +++ b/node-tools/internal/console/wait.go @@ -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,13 @@ 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" ) // WaitDefault is how long a wait lasts when no timeout is given, and WaitLongest the most it may last @@ -25,20 +29,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, `.`, 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 +69,162 @@ 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) }) + heard, ok, refused, err := s.conn.HearFirst(ctx, subjects, timeout, func(body json.RawMessage) bool { return bodyMatches(body, filter) }) if err != nil { return failure("mesh_wait could not wait: " + err.Error()) } + var answer map[string]any if !ok { - 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"}) + 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, ", "), known.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} } - 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)", - }) + 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 answerText(answer) } -// waitNames is the families a call names: a list of names, or one name. +// 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 +234,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 +243,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 } diff --git a/node-tools/internal/console/wait_test.go b/node-tools/internal/console/wait_test.go index 3cfc486..54e2aaf 100644 --- a/node-tools/internal/console/wait_test.go +++ b/node-tools/internal/console/wait_test.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "errors" + "fmt" "net/http" "os" "reflect" @@ -12,25 +13,25 @@ import ( "testing" "time" + "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" + catalogseat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + "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 @@ -38,6 +39,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}) @@ -52,8 +72,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 { @@ -61,18 +82,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: @@ -85,32 +127,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): @@ -127,7 +169,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 { @@ -150,38 +193,86 @@ 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) { +// 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, _ := 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{}, + 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) } } -} -// 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) + // 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) } } diff --git a/node-tools/internal/firstview/firstview.go b/node-tools/internal/firstview/firstview.go index 8cc6eaa..8f0d522 100644 --- a/node-tools/internal/firstview/firstview.go +++ b/node-tools/internal/firstview/firstview.go @@ -445,6 +445,25 @@ func (w wiring) Drop(owner, state string) error { return nil } +// Halves is the catalog half of each followed branch, by branch: its revision, with the seat definitions it was judged +// with, and its entries; mesh_wait checks an event name's emitter against them (design 34 every-event §14). An error says +// the half has no first view yet, never that no module exists. +func (v *Views) Halves() (map[string]meshcatalogclient.Half, error) { + if v.branches == nil { + return nil, fmt.Errorf("the tool runner does not follow the catalog half") + } + if err := v.branches.Ready(); err != nil { + return nil, err + } + out := map[string]meshcatalogclient.Half{} + for _, branch := range v.branches.Followed() { + if c, ok := v.branches.Catalog(branch); ok { + out[branch], _ = c.View() // a branch at none-yet is ready and empty (§614) + } + } + return out, nil +} + // Modules is every module the catalog half holds, by name, with its entry: the entries of the watched branches' newest trunk // revisions, the branch followed first naming a module more than one hold (design 47 G §196, §594). An error says the half has // no first view yet, never that no module exists. diff --git a/node-tools/internal/firstview/firstview_test.go b/node-tools/internal/firstview/firstview_test.go index 997203b..18c7698 100644 --- a/node-tools/internal/firstview/firstview_test.go +++ b/node-tools/internal/firstview/firstview_test.go @@ -158,6 +158,10 @@ func TestTheRunnerHoldsTheFourStatesFromTheOwnersFirstViewsAndTheirEvents(t *tes t.Fatal(err) } eventually(t, "the catalog-revision was not taken", func() bool { m, err := v.Modules(); return err == nil && len(m) == 2 }) + // And the halves by branch, with their revision, which mesh_wait names checking an emitter (design 34 every-event §14). + if h, err := v.Halves(); err != nil || h["main"].Revision.Revision != 4 || len(h["main"].Entries) != 2 { + t.Errorf("the halves: %+v, %v", h, err) + } // Events: the next revision is applied; an old one passed over; a deletion removes; none of it asks again. asked := func(tool string) int { o.mu.Lock(); defer o.mu.Unlock(); return o.asked[tool] } diff --git a/node-tools/internal/provide/provide_test.go b/node-tools/internal/provide/provide_test.go index eb35808..0d41e7a 100644 --- a/node-tools/internal/provide/provide_test.go +++ b/node-tools/internal/provide/provide_test.go @@ -126,7 +126,7 @@ func TestTheOfferServesThePortTheManifestServes(t *testing.T) { var m struct { Serves map[string]map[string]any `json:"serves"` Moved bool `json:"moved"` - State []string `json:"state"` + State []json.RawMessage `json:"state"` Reads []string `json:"reads"` Provides []struct { Name string `json:"name"` @@ -172,9 +172,18 @@ func TestTheOfferServesThePortTheManifestServes(t *testing.T) { // Every bucket the sdk's helper reaches is a state the manifest keeps or reads, by the sdk's names: the provider's // own and the seat mesh-provisioner's provisions and plans (issue 1954 L6-T1); the controller's assignments no more. declared := map[string]bool{} - for _, n := range append(m.State, m.Reads...) { + for _, n := range m.Reads { declared[n] = true } + for _, r := range m.State { // an old state entry is its name, or an object naming it (approvals, per machine) + var e struct { + Name string `json:"name"` + } + if json.Unmarshal(r, &e.Name) != nil { + _ = json.Unmarshal(r, &e) + } + declared[e.Name] = true + } for bucket, want := range map[string]string{ provider.BucketOf(Module): "bound", provisions.ProvisionsBucketOf(Provisioner): "mesh-provisioner.provisions", diff --git a/node-tools/module.json b/node-tools/module.json index f206a4d..28fae51 100644 --- a/node-tools/module.json +++ b/node-tools/module.json @@ -6,12 +6,8 @@ "invokes": [ "*" ], - "waits": [ - "mesh-knowledge.reviewed", - "mesh-delivery-gate.transition", - "gitea.forge.status", - "mesh-self-check.checked", - "operator-channel.decided.claude-code" + "uses": [ + "operator-channel" ], "provides": [ { @@ -68,7 +64,12 @@ }, "state": [ "bound", - "findings" + "findings", + { + "name": "approvals", + "per-machine": true, + "ttl-seconds": 172800 + } ], "consumes": [ "mesh-catalog.catalog-revision",