From de6741a335c78c218bb544804457d9550ff45ded Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 19 Oct 2026 11:58:10 +0200 Subject: [PATCH 1/5] Take a module's condition on the operator channel's one subject (issue 1217) --- modules/messenger/cmd/messenger/asks.go | 9 ++ modules/messenger/cmd/messenger/condition.go | 32 +++++ .../cmd/messenger/condition_intake_test.go | 134 ++++++++++++++++++ .../messenger/cmd/messenger/manifest_test.go | 6 +- modules/messenger/go.mod | 14 +- modules/messenger/go.sum | 24 ++-- modules/messenger/module.json | 4 +- 7 files changed, 200 insertions(+), 23 deletions(-) create mode 100644 modules/messenger/cmd/messenger/condition_intake_test.go diff --git a/modules/messenger/cmd/messenger/asks.go b/modules/messenger/cmd/messenger/asks.go index c80c56bc..008f264b 100644 --- a/modules/messenger/cmd/messenger/asks.go +++ b/modules/messenger/cmd/messenger/asks.go @@ -450,6 +450,15 @@ const ConditionOwner = "mesh-controller" // grants that module alone. An error is returned only when the ask could not be recorded, so the queue // offers it again; a refusal is said to the asker and the work is taken. func (h *Holder) TakeAsk(verb, asker string, body json.RawMessage) error { + if verb == ConditionVerb { + // Before the work lock: Condition takes it, and decides as for the controller's conditions. + if !asks.UsableID(strings.ReplaceAll(asker, "-", "_")) { + h.Logf("[messenger] a condition from %q was taken and dropped: not a module's name", asker) + return nil + } + h.takeCondition(asker, body) + return nil + } h.warmRootFree(true) // fresh, before the work: an ask refused for no channel stays refused with its asker h.work.Lock() defer h.work.Unlock() diff --git a/modules/messenger/cmd/messenger/condition.go b/modules/messenger/cmd/messenger/condition.go index cdd55003..c646089a 100644 --- a/modules/messenger/cmd/messenger/condition.go +++ b/modules/messenger/cmd/messenger/condition.go @@ -38,6 +38,8 @@ import ( "fmt" "strings" "time" + + "git.novox.be/novox/mesh-sdk/go/asks" ) // The events, by their local names under the controller's seat. @@ -341,3 +343,33 @@ func DecodeOpen(raw json.RawMessage) ([]Condition, error) { } return nil, fmt.Errorf("the conditions answer lists no conditions") } + +// ConditionVerb is the verb a module says a condition with: `mesh.seat.operator-channel.accept.condition.` +// (novox/hq ADR 0437 §9). +const ConditionVerb = "condition" + +// takeCondition takes one condition a module said on the seat's one subject. It is decided exactly as the +// controller's conditions are, by Condition: there is no second path. The publisher is the subject's last +// token, which the bus grants that module alone, so it is not compared with the condition's key. A body that is +// not a condition in the SDK's layout is logged by publisher, taken and dropped. +func (h *Holder) takeCondition(publisher string, body []byte) { + sc, err := asks.DecodeCondition(body) + if err != nil { + h.Logf("[messenger] a condition from %s was taken and dropped: %v", publisher, err) + return + } + c := Condition{Key: sc.Key, Scope: sc.Subject.Scope, ID: sc.Subject.ID, Machine: sc.Subject.Machine, + Severity: sc.Severity, Summary: sc.Summary, Headline: sc.Headline, Explanation: sc.Explanation, + Resolved: sc.Resolved, Needs: sc.Needs, Source: publisher} + if i := strings.LastIndex(sc.Key, "."); i >= 0 { + c.Kind = sc.Key[i+1:] + } + for _, a := range sc.Actions { + c.Actions = append(c.Actions, Action{Label: a.Label, Verb: a.Verb, Machine: a.Machine, Arguments: a.Arguments}) + } + if sc.State == asks.StateCleared { + h.Condition(EventCleared, c) + return + } + h.Condition(EventRaised, c) +} diff --git a/modules/messenger/cmd/messenger/condition_intake_test.go b/modules/messenger/cmd/messenger/condition_intake_test.go new file mode 100644 index 00000000..3312b1fa --- /dev/null +++ b/modules/messenger/cmd/messenger/condition_intake_test.go @@ -0,0 +1,134 @@ +package main + +import ( + "encoding/json" + "errors" + "fmt" + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/asks" +) + +// sdkCondition is a raised warning in mesh-sdk's layout, as a module says it on the operator channel's one subject. +func sdkCondition(needs string) asks.Condition { + verdict := asks.NothingToDo + if needs != "" { + verdict = asks.NeedsYou + needs + } + return asks.Condition{Key: "merge.d-7.stalled", Severity: asks.SeverityWarning, + Subject: asks.Scope{Scope: "merge", ID: "d-7"}, State: asks.StateRaised, Summary: "the merge waits on a check", + Headline: "A merge waits on a check", Needs: needs, Explanation: verdict + " It waits on a check."} +} + +func conditionBody(t *testing.T, c asks.Condition) json.RawMessage { + t.Helper() + b, err := asks.EncodeCondition(c) + if err != nil { + t.Fatal(err) + } + return b +} + +func TestARaisedConditionSaidOnTheOneSubjectIsToldAsTheControllersIs(t *testing.T) { // 1217/AC1 + h, tg, dt, c, _ := newHolder(t) + if err := h.TakeAsk("condition", "mesh-delivery-teller", conditionBody(t, sdkCondition("look at it."))); err != nil { + t.Fatalf("not acknowledged: %v", err) // 1217/AC1: an error would offer the work again + } + open := h.Open() + if len(open) != 1 || open[0].Key != "merge.d-7.stalled" || open[0].Subject != "merge d-7" || open[0].Origin != "condition" { + t.Fatalf("open: %+v", open) + } + settle(h, c) + if len(dt.sends) != 1 || len(tg.sends) != 0 { + t.Fatalf("a warning goes to the desk when a session answers: desktop %d telegram %d", len(dt.sends), len(tg.sends)) + } + dt.fail = errors.New("the account is not logged in") + c.pass(DesktopWarningEvery) + k := sdkCondition("look at it.") + k.Key, k.Subject.ID = "merge.d-8.stalled", "d-8" + if err := h.TakeAsk("condition", "mesh-delivery-teller", conditionBody(t, k)); err != nil { + t.Fatal(err) + } + settle(h, c) // the desktop fails: handed to Telegram + h.Flush() + if len(tg.sends) != 1 { + t.Fatalf("no session: telegram %d", len(tg.sends)) + } +} + +func TestTheSameConditionTakenTwiceIsSaidOnce(t *testing.T) { // 1217/AC2 + h, tg, dt, c, _ := newHolder(t) + body := conditionBody(t, sdkCondition("look at it.")) + for range 2 { + if err := h.TakeAsk("condition", "mesh-delivery-teller", body); err != nil { + t.Fatal(err) + } + } + settle(h, c) + if len(dt.sends)+len(tg.sends) != 1 { + t.Fatalf("said %d times, want once", len(dt.sends)+len(tg.sends)) + } + if open := h.Open(); len(open) != 1 { + t.Fatalf("open: %+v", open) + } +} + +func TestAClearedConditionIsClearedAsTheControllersIs(t *testing.T) { // 1217/AC3 + h, _, _, c, _ := newHolder(t) + if err := h.TakeAsk("condition", "mesh-delivery-teller", conditionBody(t, sdkCondition("look at it."))); err != nil { + t.Fatal(err) + } + settle(h, c) + cleared := asks.Condition{Key: "merge.d-7.stalled", Subject: asks.Scope{Scope: "merge", ID: "d-7"}, State: asks.StateCleared} + if err := h.TakeAsk("condition", "mesh-delivery-teller", conditionBody(t, cleared)); err != nil { + t.Fatal(err) + } + if open := h.Open(); len(open) != 0 { + t.Fatalf("still open: %+v", open) + } + // The same as the controller's clearing of the same key. + h2, _, _, c2, _ := newHolder(t) + h2.Condition(EventRaised, cond("merge.d-7.stalled", Warning, "the merge waits on a check")) + settle(h2, c2) + h2.Condition(EventCleared, Condition{Key: "merge.d-7.stalled"}) + if len(h2.Open()) != 0 { + t.Fatal("the controller's clearing leaves it open") + } +} + +func TestAWarningThatNeedsNothingIsKeptQuiet(t *testing.T) { // 1217/AC4 + h, tg, dt, c, _ := newHolder(t) + if err := h.TakeAsk("condition", "mesh-delivery-teller", conditionBody(t, sdkCondition(""))); err != nil { + t.Fatal(err) + } + settle(h, c) + if len(dt.sends)+len(tg.sends) != 0 { + t.Fatalf("a quiet warning was told: desktop %d telegram %d", len(dt.sends), len(tg.sends)) + } + if len(h.Open()) != 1 { + t.Fatalf("it is still recorded: %+v", h.Open()) + } +} + +func TestABodyThatIsNotAConditionIsDroppedAndTheLogNamesThePublisher(t *testing.T) { // 1217/AC5 + for name, body := range map[string]string{ + "bad JSON": `{"v":1,`, + "unknown version": `{"v":9,"key":"merge.d-7.stalled","subject":{"scope":"merge","id":"d-7"},"state":"cleared"}`, + "no version": `{"key":"merge.d-7.stalled","subject":{"scope":"merge","id":"d-7"},"state":"cleared"}`, + } { + h, tg, dt, c, _ := newHolder(t) + var logged []string + h.Logf = func(f string, a ...any) { logged = append(logged, strings.TrimSpace(fmt.Sprintf(f, a...))) } + if err := h.TakeAsk("condition", "mesh-delivery-teller", json.RawMessage(body)); err != nil { + t.Fatalf("%s: the work is taken and dropped, got %v", name, err) + } + settle(h, c) + if len(h.Open()) != 0 || len(dt.sends)+len(tg.sends) != 0 { + t.Fatalf("%s: something was recorded or told", name) + } + if !strings.Contains(strings.Join(logged, "\n"), "mesh-delivery-teller") { + t.Fatalf("%s: the log does not name the publisher: %q", name, logged) + } + } +} diff --git a/modules/messenger/cmd/messenger/manifest_test.go b/modules/messenger/cmd/messenger/manifest_test.go index 4e94d8f3..8fc2f9e5 100644 --- a/modules/messenger/cmd/messenger/manifest_test.go +++ b/modules/messenger/cmd/messenger/manifest_test.go @@ -60,12 +60,12 @@ func readManifest(t *testing.T) (manifest, string) { return m, string(raw) } -func TestItDeclaresTheSeatsOfAnAskAndHoldsTheOperatorChannel(t *testing.T) { +func TestItDeclaresTheSeatsOfAnAskAndHoldsTheOperatorChannel(t *testing.T) { // 1217/AC6: condition beside ask and cancel m, _ := readManifest(t) verbs := []string{"open", "history", "notify", "tell", "asks"} want := []seat{ - {Name: asks.Seat, Scope: "mesh", Serves: verbs, Accepts: []string{"ask", "cancel"}, Emits: []string{"decided"}, - ByCaller: []string{"ask", "cancel", "decided"}, Records: []string{"asks"}}, + {Name: asks.Seat, Scope: "mesh", Serves: verbs, Accepts: []string{"ask", "cancel", "condition"}, Emits: []string{"decided"}, + ByCaller: []string{"ask", "cancel", "condition", "decided"}, Records: []string{"asks"}}, {Name: asks.ChannelSeat, Scope: "mesh", Kinded: true, Accepts: []string{asks.Show, asks.Edit, asks.Send}}, {Name: asks.IntakeSeat, Scope: "mesh", Kinded: true, Emits: []string{asks.Chosen, asks.Link, asks.FailedWhat, asks.StandingWhat}, Proofs: []string{asks.CodeProof}}, diff --git a/modules/messenger/go.mod b/modules/messenger/go.mod index 4b62f792..e70013a7 100644 --- a/modules/messenger/go.mod +++ b/modules/messenger/go.mod @@ -1,22 +1,22 @@ module messenger -go 1.25.0 +go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261013182924-5933eb757d44 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019091011-aa68b67389c9 github.com/nats-io/nats-server/v2 v2.11.17 - github.com/nats-io/nats.go v1.51.0 + github.com/nats-io/nats.go v1.54.0 ) require ( github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op // indirect github.com/google/go-tpm v0.9.8 // indirect - github.com/klauspost/compress v1.18.5 // indirect + github.com/klauspost/compress v1.20.0 // indirect github.com/minio/highwayhash v1.0.4 // indirect github.com/nats-io/jwt/v2 v2.8.1 // indirect - github.com/nats-io/nkeys v0.4.15 // indirect + github.com/nats-io/nkeys v0.4.16 // indirect github.com/nats-io/nuid v1.0.1 // indirect - golang.org/x/crypto v0.50.0 // indirect - golang.org/x/sys v0.43.0 // indirect + golang.org/x/crypto v0.57.0 // indirect + golang.org/x/sys v0.48.0 // indirect golang.org/x/time v0.15.0 // indirect ) diff --git a/modules/messenger/go.sum b/modules/messenger/go.sum index f84463ba..aa13456d 100644 --- a/modules/messenger/go.sum +++ b/modules/messenger/go.sum @@ -1,27 +1,27 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261013182924-5933eb757d44 h1:r5xIo5qWOXQdCSMxvqJy8kV4U+P/QWXFK6971Ha6wU8= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261013182924-5933eb757d44/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019091011-aa68b67389c9 h1:qHpN/fpTDqijfMVoepxUT6mPkfcydb4LiPKsIokv6iM= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019091011-aa68b67389c9/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= github.com/google/go-tpm v0.9.8/go.mod h1:h9jEsEECg7gtLis0upRBQU+GhYVH6jMjrFxI8u6bVUY= -github.com/klauspost/compress v1.18.5 h1:/h1gH5Ce+VWNLSWqPzOVn6XBO+vJbCNGvjoaGBFW2IE= -github.com/klauspost/compress v1.18.5/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= +github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA= +github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= github.com/minio/highwayhash v1.0.4 h1:asJizugGgchQod2ja9NJlGOWq4s7KsAWr5XUc9Clgl4= github.com/minio/highwayhash v1.0.4/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ= github.com/nats-io/jwt/v2 v2.8.1 h1:V0xpGuD/N8Mi+fQNDynXohVvp7ZztevW5io8CUWlPmU= github.com/nats-io/jwt/v2 v2.8.1/go.mod h1:nWnOEEiVMiKHQpnAy4eXlizVEtSfzacZ1Q43LIRavZg= github.com/nats-io/nats-server/v2 v2.11.17 h1:GKEghcFK6A+aFx11Yf1LjgLC3txAwvyhnYzhBIQZA8I= github.com/nats-io/nats-server/v2 v2.11.17/go.mod h1:B1sFVz4StNosQ903ak4N1G01Fl/9f8e06mXpFIE2K24= -github.com/nats-io/nats.go v1.51.0 h1:ByW84XTz6W03GSSsygsZcA+xgKK8vPGaa/FCAAEHnAI= -github.com/nats-io/nats.go v1.51.0/go.mod h1:26HypzazeOkyO3/mqd1zZd53STJN0EjCYF9Uy2ZOBno= -github.com/nats-io/nkeys v0.4.15 h1:JACV5jRVO9V856KOapQ7x+EY8Jo3qw1vJt/9Jpwzkk4= -github.com/nats-io/nkeys v0.4.15/go.mod h1:CpMchTXC9fxA5zrMo4KpySxNjiDVvr8ANOSZdiNfUrs= +github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA= +github.com/nats-io/nats.go v1.54.0/go.mod h1:y+DZoD1oBOYfZTU681eTUiUjI0vbqYGixNVFHcjHJ0k= +github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg= +github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs= github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw= github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= -golang.org/x/crypto v0.50.0 h1:zO47/JPrL6vsNkINmLoo/PH1gcxpls50DNogFvB5ZGI= -golang.org/x/crypto v0.50.0/go.mod h1:3muZ7vA7PBCE6xgPX7nkzzjiUq87kRItoJQM1Yo8S+Q= +golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M= +golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA= golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= -golang.org/x/sys v0.43.0 h1:Rlag2XtaFTxp19wS8MXlJwTvoh8ArU6ezoyFsMyCTNI= -golang.org/x/sys v0.43.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo= +golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= diff --git a/modules/messenger/module.json b/modules/messenger/module.json index 7af1c177..8501bd4c 100644 --- a/modules/messenger/module.json +++ b/modules/messenger/module.json @@ -16,7 +16,8 @@ ], "accepts": [ "ask", - "cancel" + "cancel", + "condition" ], "emits": [ "decided" @@ -24,6 +25,7 @@ "by-caller": [ "ask", "cancel", + "condition", "decided" ], "records": [ From 5b235a6e802a08c231befdc517244fc59b2f51b3 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 19 Oct 2026 11:59:49 +0200 Subject: [PATCH 2/5] Prove over the bus that a condition on the one subject reaches the desk (issue 1218) --- .../cmd/messenger/condition_bus_test.go | 209 ++++++++++++++++++ 1 file changed, 209 insertions(+) create mode 100644 modules/messenger/cmd/messenger/condition_bus_test.go diff --git a/modules/messenger/cmd/messenger/condition_bus_test.go b/modules/messenger/cmd/messenger/condition_bus_test.go new file mode 100644 index 00000000..95c7bb48 --- /dev/null +++ b/modules/messenger/cmd/messenger/condition_bus_test.go @@ -0,0 +1,209 @@ +//go:build integration + +package main + +import ( + "context" + "encoding/json" + "strings" + "sync" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/asks" + "github.com/nats-io/nats.go" +) + +// A condition said on the operator channel's one subject over the real bus (novox/hq ADR 0437 §9, issue 1218), +// in the bed of bus_test.go. The fake runtime enforces no grant: that a module may say a condition only under its +// own name is the broker's, proven in the controller (issue 1216). + +// conditionBed starts the router with the desk able to send, hands it the conditions said on the one subject as +// work, and keeps what the desk is sent and whether each work item was acknowledged. +type conditionBed struct { + r *busRuntime + mu sync.Mutex + desk []asks.Message + acks []string // the answer of the router to each work item: "" when acknowledged +} + +func newConditionBed(t *testing.T) *conditionBed { + t.Helper() + b := &conditionBed{r: newBusRuntime(t)} + b.r.setKinds(map[string][]string{"desktop": {"choice", "edit"}}) + if _, err := b.r.nc.Subscribe("mesh.seat."+asks.ChannelSeat+".accept.*.desktop", func(m *nats.Msg) { + var msg asks.Message + if json.Unmarshal(m.Data, &msg) == nil { + b.mu.Lock() + b.desk = append(b.desk, msg) + b.mu.Unlock() + } + }); err != nil { + t.Fatal(err) + } + if _, err := b.r.nc.Subscribe("mesh.seat."+asks.Seat+".accept.condition.*", func(m *nats.Msg) { + answer, err := b.r.request("mesh/work", map[string]any{"work": map[string]any{"worker": AsksWorker, "subject": m.Subject, "body": json.RawMessage(m.Data)}}) + var a struct { + Error *struct { + Message string `json:"message"` + } `json:"error"` + } + _ = json.Unmarshal(answer, &a) + res := "" + if err != nil { + res = err.Error() + } else if a.Error != nil { + res = a.Error.Message + } + b.mu.Lock() + b.acks = append(b.acks, res) + b.mu.Unlock() + }); err != nil { + t.Fatal(err) + } + b.forwardEvents(t) + b.r.start(t.TempDir()) + b.r.until("the bundle takes work and read the kinds", 20*time.Second, func() bool { + b.r.mu.Lock() + defer b.r.mu.Unlock() + return b.r.taking && b.r.kindsRead > 0 + }) + b.r.until("the bundle subscribed to the channels' word", 20*time.Second, func() bool { + b.r.mu.Lock() + defer b.r.mu.Unlock() + return b.r.subscribed + }) + standing, _ := json.Marshal(Standing{Ready: true, At: time.Now().UTC()}) + if err := b.r.nc.Publish(asks.IntakeSubject(asks.StandingWhat, "desktop"), standing); err != nil { + t.Fatal(err) + } + time.Sleep(time.Second) + return b +} + +// forwardEvents hands the bundle the channels' word, as bus_test.go's forward does for it. +func (b *conditionBed) forwardEvents(t *testing.T) { + t.Helper() + if _, err := b.r.nc.Subscribe("mesh.seat."+asks.IntakeSeat+".event.>", func(m *nats.Msg) { + key := strings.Replace(strings.TrimPrefix(m.Subject, "mesh.seat."), ".event", "", 1) + _, _ = b.r.request("mesh/event", map[string]any{"envelope": map[string]any{"key": key, "node": "test", "body": json.RawMessage(m.Data)}}) + }); err != nil { + t.Fatal(err) + } +} + +func (b *conditionBed) say(t *testing.T, publisher string, c asks.Condition) { + t.Helper() + body, err := asks.EncodeCondition(c) + if err != nil { + t.Fatal(err) + } + subject, err := asks.ConditionSubject(publisher) + if err != nil { + t.Fatal(err) + } + if err := b.r.nc.Publish(subject, body); err != nil { + t.Fatal(err) + } +} + +func (b *conditionBed) record(key string) *Record { + e, err := b.r.bucket("open").Get(context.Background(), kvKey(key)) + if err != nil { + return nil + } + var rec Record + if json.Unmarshal(e.Value(), &rec) != nil { + return nil + } + return &rec +} + +func (b *conditionBed) told() []asks.Message { + b.mu.Lock() + defer b.mu.Unlock() + return append([]asks.Message(nil), b.desk...) +} + +func (b *conditionBed) acknowledged(n int) bool { + b.mu.Lock() + defer b.mu.Unlock() + if len(b.acks) != n { + return false + } + for _, a := range b.acks { + if a != "" { + return false + } + } + return true +} + +func busCondition(key, id, headline, needs string) asks.Condition { + verdict := asks.NothingToDo + if needs != "" { + verdict = asks.NeedsYou + needs + } + return asks.Condition{Key: key, Severity: asks.SeverityWarning, Subject: asks.Scope{Scope: "merge", ID: id}, + State: asks.StateRaised, Summary: "the merge waits on a check", Headline: headline, Needs: needs, + Explanation: verdict + " It waits on a check."} +} + +// 1218/AC1 and 1218/AC2: a raised condition said on the one subject reaches the desk, is recorded and its work item +// is acknowledged; its cleared form ends it. The router holds a warning for its start grace and burst window. +func TestAConditionSaidOnTheOneSubjectReachesTheDeskAndIsClearedOverTheBus(t *testing.T) { + b := newConditionBed(t) + b.say(t, "mesh-delivery-teller", busCondition("merge.d-7.stalled", "d-7", "A merge waits on a check", "release it, or stop it.")) + // 1218/AC1: recorded, acknowledged, and told to the desk. + b.r.until("the condition is recorded open", 30*time.Second, func() bool { + rec := b.record("merge.d-7.stalled") + return rec != nil && rec.Origin == "condition" && rec.Cleared.IsZero() + }) + b.r.until("the work item is acknowledged", 30*time.Second, func() bool { return b.acknowledged(1) }) + b.r.until("the desk is sent the condition", 3*time.Minute, func() bool { + for _, m := range b.told() { + if strings.Contains(m.Title+m.Body, "A merge waits on a check") { + return true + } + } + return false + }) + time.Sleep(3 * time.Second) + if n := b.nAcks(); n != 1 { + t.Errorf("the work item was offered %d times, want once: it was not acknowledged", n) + } + // 1218/AC2: the cleared form, on the same subject, ends it. + b.say(t, "mesh-delivery-teller", asks.Condition{Key: "merge.d-7.stalled", Subject: asks.Scope{Scope: "merge", ID: "d-7"}, State: asks.StateCleared}) + b.r.until("the condition is recorded cleared", 30*time.Second, func() bool { + rec := b.record("merge.d-7.stalled") + return rec == nil || !rec.Cleared.IsZero() + }) +} + +func (b *conditionBed) nAcks() int { + b.mu.Lock() + defer b.mu.Unlock() + return len(b.acks) +} + +// 1218/AC3: a warning whose needs is empty is recorded and reaches no channel. A second warning that needs the +// operator, said after it, is told: when it is, the quiet one's turn has passed and it was not sent. +func TestAWarningThatNeedsNothingReachesNoChannelOverTheBus(t *testing.T) { + b := newConditionBed(t) + b.say(t, "mesh-delivery-teller", busCondition("merge.d-8.quiet", "d-8", "A quiet merge note", "")) + b.r.until("the quiet warning is recorded", 30*time.Second, func() bool { return b.record("merge.d-8.quiet") != nil }) + b.say(t, "mesh-delivery-teller", busCondition("merge.d-9.loud", "d-9", "A loud merge note", "release it, or stop it.")) + b.r.until("the warning that needs the operator is sent to the desk", 3*time.Minute, func() bool { + for _, m := range b.told() { + if strings.Contains(m.Title+m.Body, "A loud merge note") { + return true + } + } + return false + }) + for _, m := range b.told() { + if strings.Contains(m.Title+m.Body, "A quiet merge note") { + t.Errorf("a warning that needs nothing was sent: %+v", m) + } + } +} From f737418bb3d79cc62c654ada914f1ae9a688fdf1 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 19 Oct 2026 12:04:58 +0200 Subject: [PATCH 3/5] Keep a module's condition open when the controller's conditions are read (issue 1217, review round 1 finding 1) --- modules/messenger/cmd/messenger/condition.go | 31 ++++++++++--------- .../cmd/messenger/condition_intake_test.go | 22 +++++++++++++ modules/messenger/cmd/messenger/holder.go | 10 +++--- modules/messenger/cmd/messenger/sync.go | 3 +- 4 files changed, 47 insertions(+), 19 deletions(-) diff --git a/modules/messenger/cmd/messenger/condition.go b/modules/messenger/cmd/messenger/condition.go index c646089a..1378d596 100644 --- a/modules/messenger/cmd/messenger/condition.go +++ b/modules/messenger/cmd/messenger/condition.go @@ -59,19 +59,22 @@ const ( // Condition is what this holder needs of one. type Condition struct { - Key string - Kind string - Scope string - ID string - Machine string - Severity string - Summary string - Headline string - Explanation string - Resolved string - Needs string - Actions []Action - Source string + Key string + Kind string + Scope string + ID string + Machine string + Severity string + Summary string + Headline string + Explanation string + Resolved string + Needs string + Actions []Action + Source string + // Publisher is the module that said the condition on the operator channel's one subject; empty for the + // controller's own, which a reading of the controller's conditions may end. + Publisher string Resolver string Raised time.Time LastObserved time.Time @@ -360,7 +363,7 @@ func (h *Holder) takeCondition(publisher string, body []byte) { } c := Condition{Key: sc.Key, Scope: sc.Subject.Scope, ID: sc.Subject.ID, Machine: sc.Subject.Machine, Severity: sc.Severity, Summary: sc.Summary, Headline: sc.Headline, Explanation: sc.Explanation, - Resolved: sc.Resolved, Needs: sc.Needs, Source: publisher} + Resolved: sc.Resolved, Needs: sc.Needs, Source: publisher, Publisher: publisher} if i := strings.LastIndex(sc.Key, "."); i >= 0 { c.Kind = sc.Key[i+1:] } diff --git a/modules/messenger/cmd/messenger/condition_intake_test.go b/modules/messenger/cmd/messenger/condition_intake_test.go index 3312b1fa..e3a1899d 100644 --- a/modules/messenger/cmd/messenger/condition_intake_test.go +++ b/modules/messenger/cmd/messenger/condition_intake_test.go @@ -6,6 +6,7 @@ import ( "fmt" "strings" "testing" + "time" "git.novox.be/novox/mesh-sdk/go/asks" ) @@ -132,3 +133,24 @@ func TestABodyThatIsNotAConditionIsDroppedAndTheLogNamesThePublisher(t *testing. } } } + +func TestAModulesConditionStaysOpenWhenTheControllersConditionsAreRead(t *testing.T) { // 1217/AC7 + h, tg, dt, c, _ := newHolder(t) + body := conditionBody(t, sdkCondition("look at it.")) + if err := h.TakeAsk("condition", "mesh-delivery-teller", body); err != nil { + t.Fatal(err) + } + settle(h, c) + c.pass(time.Second) + res := h.Sync(nil, c.now()) // the controller lists nothing: the teller's condition is not its to end + if len(res.Gone) != 0 || len(h.Open()) != 1 { + t.Fatalf("the reading ended a module's condition: gone %v, open %+v", res.Gone, h.Open()) + } + if err := h.TakeAsk("condition", "mesh-delivery-teller", body); err != nil { // raised again: the same message + t.Fatal(err) + } + settle(h, c) + if len(dt.sends)+len(tg.sends) != 1 { + t.Fatalf("said %d times, want once", len(dt.sends)+len(tg.sends)) + } +} diff --git a/modules/messenger/cmd/messenger/holder.go b/modules/messenger/cmd/messenger/holder.go index f24ae6d7..8fe66c20 100644 --- a/modules/messenger/cmd/messenger/holder.go +++ b/modules/messenger/cmd/messenger/holder.go @@ -184,8 +184,10 @@ type Record struct { Answered string `json:"answered,omitempty"` // Ask is the open ask about this record, `.`: its options are this record's answers // (novox/hq ADR 0259). - Ask string `json:"ask,omitempty"` - Origin string `json:"origin"` + Ask string `json:"ask,omitempty"` + Origin string `json:"origin"` + // Publisher is the module that said it on the operator channel's one subject; empty for the controller's. + Publisher string `json:"publisher,omitempty"` More string `json:"more"` Raised time.Time `json:"raised"` SilencedTill time.Time `json:"silenced_till,omitempty"` @@ -228,8 +230,8 @@ func recordOf(c Condition) Record { return Record{Key: c.Key, Kind: c.Kind, Subject: c.SubjectWords(), Scope: c.Scope, SubjectID: c.ID, Machine: c.Machine, Severity: c.Severity, Summary: c.Summary, Headline: c.Headline, Explanation: c.Explanation, Resolved: c.Resolved, Needs: c.Needs, Actions: c.Actions, - Origin: "condition", - More: "conditions show " + c.Key, Raised: c.Raised, SilencedTill: c.SilencedTill, SilencedWhy: c.SilencedWhy} + Origin: "condition", Publisher: c.Publisher, + More: "conditions show " + c.Key, Raised: c.Raised, SilencedTill: c.SilencedTill, SilencedWhy: c.SilencedWhy} } // takeWords keeps the plain words current, as the summary is: a condition's words follow what is seen. diff --git a/modules/messenger/cmd/messenger/sync.go b/modules/messenger/cmd/messenger/sync.go index 35587be1..efeaa6ee 100644 --- a/modules/messenger/cmd/messenger/sync.go +++ b/modules/messenger/cmd/messenger/sync.go @@ -33,13 +33,14 @@ func (h *Holder) Sync(open []Condition, asked time.Time) SyncResult { for _, c := range open { in[c.Key] = true } + // A module's condition (Publisher set) is not the controller's to end: only what it said is ended here. // What is open here and no longer there ended while this holder was not told: forgotten without // a message — said only where an edit notifies nobody. What this holder heard of after the // reading was asked for is newer than it, and stands. h.mu.Lock() var gone []*Record for _, r := range h.open { - if r.Origin == "condition" && r.Cleared.IsZero() && !in[r.Key] && !r.Seen.After(asked) { + if r.Origin == "condition" && r.Publisher == "" && r.Cleared.IsZero() && !in[r.Key] && !r.Seen.After(asked) { gone = append(gone, r) } } From 9f9e9fc9d3e8dabafceb401a12bf796df505cd71 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 19 Oct 2026 12:05:07 +0200 Subject: [PATCH 4/5] Shorten the router's start grace in the test bundle so the bus tests stay in seconds (issue 1218, review round 1 finding 2) --- modules/messenger/cmd/messenger/bus_test.go | 1 + modules/messenger/cmd/messenger/condition_bus_test.go | 6 +++--- modules/messenger/cmd/messenger/holder.go | 6 ++++-- 3 files changed, 8 insertions(+), 5 deletions(-) diff --git a/modules/messenger/cmd/messenger/bus_test.go b/modules/messenger/cmd/messenger/bus_test.go index 28c8135e..dc6ada66 100644 --- a/modules/messenger/cmd/messenger/bus_test.go +++ b/modules/messenger/cmd/messenger/bus_test.go @@ -34,6 +34,7 @@ const bundleVar = "MESH_MESSENGER_TEST_BUNDLE" func TestMain(m *testing.M) { if os.Getenv(bundleVar) == "1" { kindsEvery = 200 * time.Millisecond + StartGrace = 500 * time.Millisecond main() os.Exit(0) } diff --git a/modules/messenger/cmd/messenger/condition_bus_test.go b/modules/messenger/cmd/messenger/condition_bus_test.go index 95c7bb48..69dd1739 100644 --- a/modules/messenger/cmd/messenger/condition_bus_test.go +++ b/modules/messenger/cmd/messenger/condition_bus_test.go @@ -150,7 +150,7 @@ func busCondition(key, id, headline, needs string) asks.Condition { } // 1218/AC1 and 1218/AC2: a raised condition said on the one subject reaches the desk, is recorded and its work item -// is acknowledged; its cleared form ends it. The router holds a warning for its start grace and burst window. +// is acknowledged; its cleared form ends it. The bundle starts with a short grace (bus_test.go). func TestAConditionSaidOnTheOneSubjectReachesTheDeskAndIsClearedOverTheBus(t *testing.T) { b := newConditionBed(t) b.say(t, "mesh-delivery-teller", busCondition("merge.d-7.stalled", "d-7", "A merge waits on a check", "release it, or stop it.")) @@ -160,7 +160,7 @@ func TestAConditionSaidOnTheOneSubjectReachesTheDeskAndIsClearedOverTheBus(t *te return rec != nil && rec.Origin == "condition" && rec.Cleared.IsZero() }) b.r.until("the work item is acknowledged", 30*time.Second, func() bool { return b.acknowledged(1) }) - b.r.until("the desk is sent the condition", 3*time.Minute, func() bool { + b.r.until("the desk is sent the condition", 15*time.Second, func() bool { for _, m := range b.told() { if strings.Contains(m.Title+m.Body, "A merge waits on a check") { return true @@ -193,7 +193,7 @@ func TestAWarningThatNeedsNothingReachesNoChannelOverTheBus(t *testing.T) { b.say(t, "mesh-delivery-teller", busCondition("merge.d-8.quiet", "d-8", "A quiet merge note", "")) b.r.until("the quiet warning is recorded", 30*time.Second, func() bool { return b.record("merge.d-8.quiet") != nil }) b.say(t, "mesh-delivery-teller", busCondition("merge.d-9.loud", "d-9", "A loud merge note", "release it, or stop it.")) - b.r.until("the warning that needs the operator is sent to the desk", 3*time.Minute, func() bool { + b.r.until("the warning that needs the operator is sent to the desk", 15*time.Second, func() bool { for _, m := range b.told() { if strings.Contains(m.Title+m.Body, "A loud merge note") { return true diff --git a/modules/messenger/cmd/messenger/holder.go b/modules/messenger/cmd/messenger/holder.go index 8fe66c20..9afc0e30 100644 --- a/modules/messenger/cmd/messenger/holder.go +++ b/modules/messenger/cmd/messenger/holder.go @@ -55,6 +55,10 @@ import ( "time" ) +// StartGrace: after start, nothing but the urgent conditions open in the controller's state. A variable, as +// kindsEvery is, so the test bundle shortens it. +var StartGrace = time.Minute + const ( RemindUrgent = time.Hour RemindWarning = 12 * time.Hour @@ -75,8 +79,6 @@ const ( BurstWindow = 30 * time.Second // DesktopWarningEvery: warnings reach the desktop at most this often. DesktopWarningEvery = 15 * time.Minute - // StartGrace: after start, nothing but the urgent conditions open in the controller's state. - StartGrace = time.Minute // ResyncEvery: how often the open set is read from the controller again. ResyncEvery = 10 * time.Minute // RetryEvery: a channel that failed is tried again after this. From 5aca005cb741306fbe2caece689c3330fa124e02 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 19 Oct 2026 12:15:47 +0200 Subject: [PATCH 5/5] Shorten the router's burst window in the test bundle so the desk is sent within seconds (issue 1218, check finding 1) --- modules/messenger/cmd/messenger/bus_test.go | 1 + modules/messenger/cmd/messenger/holder.go | 6 ++++-- 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/modules/messenger/cmd/messenger/bus_test.go b/modules/messenger/cmd/messenger/bus_test.go index dc6ada66..6e34d5eb 100644 --- a/modules/messenger/cmd/messenger/bus_test.go +++ b/modules/messenger/cmd/messenger/bus_test.go @@ -35,6 +35,7 @@ func TestMain(m *testing.M) { if os.Getenv(bundleVar) == "1" { kindsEvery = 200 * time.Millisecond StartGrace = 500 * time.Millisecond + BurstWindow = 200 * time.Millisecond main() os.Exit(0) } diff --git a/modules/messenger/cmd/messenger/holder.go b/modules/messenger/cmd/messenger/holder.go index 9afc0e30..79cbb1b8 100644 --- a/modules/messenger/cmd/messenger/holder.go +++ b/modules/messenger/cmd/messenger/holder.go @@ -55,6 +55,10 @@ import ( "time" ) +// BurstWindow: what arrives for a channel within it goes out as one message. A variable, so the test bundle +// shortens it. +var BurstWindow = 30 * time.Second + // StartGrace: after start, nothing but the urgent conditions open in the controller's state. A variable, as // kindsEvery is, so the test bundle shortens it. var StartGrace = time.Minute @@ -75,8 +79,6 @@ const ( // FreshFor: an event older than this by its own time is history — state, never a message. FreshFor = 10 * time.Minute - // BurstWindow: what arrives for a channel within it goes out as one message. - BurstWindow = 30 * time.Second // DesktopWarningEvery: warnings reach the desktop at most this often. DesktopWarningEvery = 15 * time.Minute // ResyncEvery: how often the open set is read from the controller again.