From 3f89c96d205773877a8ddf0765bd001703d09493 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 05:00:03 +0100 Subject: [PATCH 1/4] E-1: the tool runner's manifest names no waits and keeps its approvals (issue 1966) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit module.json drops waits and its five families, uses operator-channel for the asks of marked waits, and keeps the per-machine state approvals (48 hours) in its old state section; the manifest test holds each change. Builds: design 34 every-event §11, §19 (the manifest test); design 46 7-approval-mark §7, §13 (manifest entry) Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../internal/console/manifest_waits_test.go | 58 +++++++++++++------ node-tools/internal/provide/provide_test.go | 13 ++++- node-tools/module.json | 15 ++--- 3 files changed, 59 insertions(+), 27 deletions(-) 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/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", From cb47de85176cca12e56c0ec0ecdd3131341c3b9b Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 05:03:13 +0100 Subject: [PATCH 2/4] E-2: a heard event keeps its own time, and a refused wait is never a hot loop (issue 1966) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit HearFirst reads the event's x-time as Published, notes a subscription the bus refuses while the others go on waiting, and answers a wait refused whole only after its timeout, in words naming the grants part's grant of every event subject; mesh_wait lists the refused subjects in its answer. Builds: design 34 every-event §12, §16, §18, §19 (bus/wait_test.go) Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- node-tools/internal/bus/wait.go | 74 ++++++++++++++++--------- node-tools/internal/bus/wait_test.go | 82 ++++++++++++++++++++-------- node-tools/internal/console/wait.go | 21 ++++--- 3 files changed, 119 insertions(+), 58 deletions(-) diff --git a/node-tools/internal/bus/wait.go b/node-tools/internal/bus/wait.go index 4dda03b..aa2e961 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" @@ -23,31 +24,42 @@ type WaitIssued struct { 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 +70,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 +88,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/wait.go b/node-tools/internal/console/wait.go index 9ee1a41..1194b5b 100644 --- a/node-tools/internal/console/wait.go +++ b/node-tools/internal/console/wait.go @@ -86,18 +86,25 @@ func (s *Surface) waitAnswer(ctx context.Context, args map[string]any) map[strin 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"} + } else { + answer = 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(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)", - }) + // 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. From 662ab2a3b1f54c11a8c8311d75af86ee3ca64216 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 05:06:36 +0100 Subject: [PATCH 3/4] E-3: mesh_wait takes any event name (issue 1966) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit waitAnswer no longer reads the membership: each name becomes its subjects through buscompose.EventSubjects, optionally narrowed by from (seat, module or stream), at most 16 names, a subject once; the heard event is named by buscompose.EventName with its namespace and its own time, and a decided event says the operator's answer is read only with claude_code_answer. WaitIssued and Membership.Waits go; the refusals are one table. Builds: design 34 every-event §13, §15, §17, §19, §26 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- node-tools/internal/bus/bus.go | 3 - node-tools/internal/bus/wait.go | 7 - node-tools/internal/console/wait.go | 87 +++++++----- node-tools/internal/console/wait_test.go | 172 +++++++++++++++-------- 4 files changed, 163 insertions(+), 106 deletions(-) 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 aa2e961..c42a667 100644 --- a/node-tools/internal/bus/wait.go +++ b/node-tools/internal/bus/wait.go @@ -17,13 +17,6 @@ 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, 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 { diff --git a/node-tools/internal/console/wait.go b/node-tools/internal/console/wait.go index 1194b5b..c35af73 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,11 @@ import ( "encoding/json" "fmt" "math" - "sort" + "slices" "strings" "time" + + "git.novox.be/novox/mesh-sdk/go/buscompose" ) // WaitDefault is how long a wait lasts when no timeout is given, and WaitLongest the most it may last @@ -25,20 +27,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,34 +67,25 @@ 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 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) + for _, subject := range subs { + if !slices.Contains(subjects, subject) { + subjects = append(subjects, subject) + } } - family[subject] = name } heard, ok, refused, err := s.conn.HearFirst(ctx, subjects, timeout, func(body json.RawMessage) bool { return bodyMatches(body, filter) }) @@ -95,10 +97,17 @@ func (s *Surface) waitAnswer(ctx context.Context, args map[string]any) map[strin answer = map[string]any{"timed-out": true, "waited": timeout.String(), "subjects": names, "read": "nothing matching was heard; read the state once from the verb that owns it, then wait again"} } else { - answer = 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)", + name, from, _ := buscompose.EventName(heard.Subject) + h := map[string]any{"subject": name, "from": from, "body": heard.Body, "at": heard.At.Format(time.RFC3339Nano)} + if heard.Published != "" { + h["published"] = heard.Published } + read := "an event only wakes you: read the state from the verb that owns it before acting (novox/hq ADR 0367 §4)" + if strings.HasPrefix(name, "operator-channel.decided") { + read = "an event only wakes you: the operator's answer is read only with claude_code_answer, never from " + + "the event's body (novox/hq ADR 0670 §5)" + } + answer = map[string]any{"heard": h, "read": read} } // A subscription the bus refused while others stood is said (design 34 every-event §16). if len(refused) > 0 { @@ -107,7 +116,7 @@ func (s *Surface) waitAnswer(ctx context.Context, args map[string]any) map[strin return answerText(answer) } -// waitNames is the families a call names: a list of names, or one name. +// 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) { @@ -117,7 +126,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) } @@ -126,7 +135,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 862517f..a19522e 100644 --- a/node-tools/internal/console/wait_test.go +++ b/node-tools/internal/console/wait_test.go @@ -3,6 +3,7 @@ package console import ( "context" "encoding/json" + "fmt" "net/http" "os" "reflect" @@ -11,24 +12,22 @@ import ( "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" ) -// 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 +35,17 @@ func waitingRuntime(t *testing.T) (*Surface, *mt.Mesh) { return NewSurface(conn, busSource{conn}, "desk.node-tools"), mesh } +// 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 +60,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 +70,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 +115,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): @@ -148,38 +180,64 @@ 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) + // 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) } } From 0b0f879e86f1225e7d0a8428989fdc5f31a90308 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 05:11:52 +0100 Subject: [PATCH 4/4] E-4: a misspelt event name is refused, not waited on for ever (issue 1966) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Before it subscribes, mesh_wait checks each name's emitter against the catalog half the tool runner holds (its seats' definitions and its modules' entries), refusing one neither knows and naming the half and its revisions; a known emitter's undeclared event is waited on and its timed-out answer says so; while the half is not yet read the answer says the names were unchecked. The first views answer the halves by branch for it. Builds: design 34 every-event §14, §19 (rows of the table of refusals) Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- node-tools/internal/console/wait.go | 110 +++++++++++++++++- node-tools/internal/console/wait_test.go | 35 +++++- node-tools/internal/firstview/firstview.go | 19 +++ .../internal/firstview/firstview_test.go | 4 + 4 files changed, 166 insertions(+), 2 deletions(-) diff --git a/node-tools/internal/console/wait.go b/node-tools/internal/console/wait.go index c35af73..a93a16d 100644 --- a/node-tools/internal/console/wait.go +++ b/node-tools/internal/console/wait.go @@ -17,7 +17,9 @@ import ( "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 @@ -75,12 +77,29 @@ func (s *Surface) waitAnswer(ctx context.Context, args map[string]any) map[strin return failure(fmt.Sprintf("mesh_wait names at most %d events in one call, not %d: each name is six "+ "subscriptions for the call (design 34 every-event §13)", waitMostNames, len(names))) } - var subjects []string + known, unchecked := s.emitters() + var subjects, undeclared []string for _, name := range names { subs, err := buscompose.EventSubjects(name, from) if err != nil { return failure("mesh_wait may not wait on it: " + err.Error()) } + // A misspelt emitter is refused, never waited on for ever; a known emitter's undeclared event is waited on and + // its timed-out answer says so (design 34 every-event §14). + if unchecked == "" { + emitter, event, _ := strings.Cut(name, ".") + seatEvents, isSeat := known.seats[emitter] + moduleEvents, isModule := known.modules[emitter] + isSeat = isSeat && from != "module" && from != "stream" + isModule = isModule && from != "seat" + if !isSeat && !isModule { + return failure(fmt.Sprintf("mesh_wait may not wait on %q: %s is neither a seat nor a module of %s", + name, emitter, known.source)) + } + if !(isSeat && declares(seatEvents, event)) && !(isModule && declares(moduleEvents, event)) { + undeclared = append(undeclared, name) + } + } for _, subject := range subs { if !slices.Contains(subjects, subject) { subjects = append(subjects, subject) @@ -96,6 +115,10 @@ func (s *Surface) waitAnswer(ctx context.Context, args map[string]any) map[strin 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, ", "), 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)} @@ -109,6 +132,9 @@ func (s *Surface) waitAnswer(ctx context.Context, args map[string]any) map[strin } 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 @@ -116,6 +142,88 @@ func (s *Surface) waitAnswer(ctx context.Context, args map[string]any) map[strin return answerText(answer) } +// 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 diff --git a/node-tools/internal/console/wait_test.go b/node-tools/internal/console/wait_test.go index a19522e..21bd105 100644 --- a/node-tools/internal/console/wait_test.go +++ b/node-tools/internal/console/wait_test.go @@ -12,6 +12,8 @@ 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" @@ -35,6 +37,14 @@ 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() @@ -157,7 +167,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 { @@ -211,6 +222,28 @@ func TestMeshWaitRefusesLoudlyAndNeverLoops(t *testing.T) { } } + // 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{ diff --git a/node-tools/internal/firstview/firstview.go b/node-tools/internal/firstview/firstview.go index 9e60efd..ff5b5eb 100644 --- a/node-tools/internal/firstview/firstview.go +++ b/node-tools/internal/firstview/firstview.go @@ -437,6 +437,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] }