From c2117628139cb297a00b3615a88130783aa7ac5c Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 16 Oct 2026 17:51:26 +0200 Subject: [PATCH 1/2] Hold an ask no channel can carry now and offer it when one can; show held and why (issue 926, ADR 0397) The router refused such an ask with 'no channel', so the asker waited for an Approve nobody saw. It is now held as an open ask with the per-kind reason, judged again when the kinds or a channel's standing change and every minute, and the knowledge base says held and why. --- .../cmd/mesh-knowledge/answers.go | 4 + .../cmd/mesh-knowledge/kit_test.go | 18 ++ .../cmd/mesh-knowledge/operator.go | 43 +++- .../cmd/mesh-knowledge/record.go | 3 + .../cmd/mesh-knowledge/review_test.go | 51 ++++ .../messenger/cmd/messenger/answers_test.go | 18 +- modules/messenger/cmd/messenger/asks.go | 217 +++++++++++++++--- modules/messenger/cmd/messenger/held_test.go | 142 ++++++++++++ modules/messenger/cmd/messenger/holder.go | 1 + modules/messenger/cmd/messenger/main.go | 4 +- 10 files changed, 453 insertions(+), 48 deletions(-) create mode 100644 modules/messenger/cmd/messenger/held_test.go diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/answers.go b/modules/mesh-knowledge/cmd/mesh-knowledge/answers.go index 6245228e..ae9f7b4d 100644 --- a/modules/mesh-knowledge/cmd/mesh-knowledge/answers.go +++ b/modules/mesh-knowledge/cmd/mesh-knowledge/answers.go @@ -361,6 +361,10 @@ func (k *KB) Get(a GetAsk) (*Got, error) { } if r.Ask != nil { g.Waiting = "the operator's ask: " + r.Ask.Said + if r.Ask.Said == AskWaiting && r.Ask.Held != "" { + // Held by the router, not shown to the operator yet: said so, with why (ADR 0397, issue 926). + g.Waiting = "the operator's ask is held by the router, not yet shown to the operator: " + r.Ask.Held + } } if a.Version == 0 { g.Review = reviewSays(e.Working()) diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/kit_test.go b/modules/mesh-knowledge/cmd/mesh-knowledge/kit_test.go index 422a5597..07915e59 100644 --- a/modules/mesh-knowledge/cmd/mesh-knowledge/kit_test.go +++ b/modules/mesh-knowledge/cmd/mesh-knowledge/kit_test.go @@ -367,6 +367,24 @@ func (r *router) answer(id, option string, at time.Time) asks.Warrant { return w } +// hold marks an ask as the router does when no channel can carry it yet (ADR 0397): state held, with why. +func (r *router) hold(id, why string) { + r.mu.Lock() + defer r.mu.Unlock() + rec := r.records[id] + rec.State, rec.Held = "held", why + r.records[id] = rec +} + +// offer marks a held ask open again, as the router does when a channel can carry it. +func (r *router) offer(id string) { + r.mu.Lock() + defer r.mu.Unlock() + rec := r.records[id] + rec.State, rec.Held = "open", "" + r.records[id] = rec +} + // said is the events said. type said struct { mu sync.Mutex diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/operator.go b/modules/mesh-knowledge/cmd/mesh-knowledge/operator.go index 5f05e26d..6f136a3c 100644 --- a/modules/mesh-knowledge/cmd/mesh-knowledge/operator.go +++ b/modules/mesh-knowledge/cmd/mesh-knowledge/operator.go @@ -222,7 +222,9 @@ func (k *KB) askOperator(ctx context.Context, held *Entry, w *Record, by string, // routerRecord is the router's record of one ask, as far as this module reads it. type routerRecord struct { - State string `json:"state"` + State string `json:"state"` + // Held is why no channel can carry the ask yet, while its state is held (ADR 0397). + Held string `json:"held"` Ask *asks.Ask `json:"ask"` Warrant *asks.Warrant `json:"warrant"` } @@ -255,7 +257,15 @@ func (k *KB) Reconcile() ([]string, error) { continue } var rec routerRecord - if json.Unmarshal(entry.Value, &rec) != nil || rec.State == "open" || rec.Warrant == nil { + if json.Unmarshal(entry.Value, &rec) != nil { + continue + } + // The router holds an ask no channel can carry yet (ADR 0397, issue 926): the record says so, and why, + // until it is offered, rather than saying it waits for an answer nobody was shown. + if held := heldSays(rec); held != w.Ask.Held && (rec.State == "open" || rec.State == "held") { + said = append(said, k.keepHeld(ctx, e, w, held)) + } + if rec.State == "open" || rec.State == "held" || rec.Warrant == nil { continue } said = append(said, k.decide(ctx, e, w, *rec.Warrant)) @@ -263,6 +273,33 @@ func (k *KB) Reconcile() ([]string, error) { return said, nil } +// heldSays is why the router holds an ask, in words, or "" when it does not. +func heldSays(rec routerRecord) string { + if rec.State != "held" { + return "" + } + if rec.Held == "" { + return "the router holds it; no channel can carry it yet" + } + return rec.Held +} + +// keepHeld records, on the record that waits for the operator, that the router holds its ask and why (or that it +// no longer does). The ask stays waiting. +func (k *KB) keepHeld(ctx context.Context, e *Entry, w *Record, held string) string { + w = w.Clone() + w.Ask.Held = held + why := "the operator's ask is offered by the router" + if held != "" { + why = "the operator's ask is held by the router: " + held + } + if err := k.Store.Save(ctx, Version{By: "the knowledge base", At: k.now(), Why: why, Record: w, Insight: e.Pending != nil && e.Pending.IsInsight()}, + e.Pending == nil, false, e.Revision); err != nil { + return fmt.Sprintf("%s: what became of its ask could not be kept (%v)", w.Key(), err) + } + return fmt.Sprintf("%s: %s", w.Key(), why) +} + // endAsk records an ask ended without the operator's choice: the record stays proposed. func (k *KB) endAsk(ctx context.Context, e *Entry, w *Record, why string) string { w = w.Clone() @@ -277,6 +314,8 @@ func (k *KB) endAsk(ctx context.Context, e *Entry, w *Record, why string) string // decide holds one closed ask's warrant to the record, and accepts the record when it is the operator's Approve. func (k *KB) decide(ctx context.Context, e *Entry, w *Record, warrant asks.Warrant) string { + w = w.Clone() + w.Ask.Held = "" // an ask that ended is held no longer if warrant.Outcome != asks.OutcomeChosen { why := "ended " + string(warrant.Outcome) if warrant.Words != "" { diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/record.go b/modules/mesh-knowledge/cmd/mesh-knowledge/record.go index 4ad61e16..cf4f3d69 100644 --- a/modules/mesh-knowledge/cmd/mesh-knowledge/record.go +++ b/modules/mesh-knowledge/cmd/mesh-knowledge/record.go @@ -254,6 +254,9 @@ type OperatorAsk struct { // Said is what came of it: waiting, approved, declined, expired, refused (and why). Said string `json:"said"` Warrant string `json:"warrant,omitempty"` + // Held is why the router holds the ask while it waits: no channel can carry it yet (ADR 0397, issue 926). + // Empty once the router offers it, or when it ended. + Held string `json:"held,omitempty"` } // Drafting is where a prose record stands in the drafting and confirming of its clauses (to-be 55 §7). diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/review_test.go b/modules/mesh-knowledge/cmd/mesh-knowledge/review_test.go index 2590512a..ec3be005 100644 --- a/modules/mesh-knowledge/cmd/mesh-knowledge/review_test.go +++ b/modules/mesh-knowledge/cmd/mesh-knowledge/review_test.go @@ -418,3 +418,54 @@ func TestReserveAsksTheTrackerAndANumberNeverProposedIsWithdrawn(t *testing.T) { t.Fatal("a withdrawn number took a proposal") } } + +// An ask the router holds because no channel can carry it yet is shown as held, with the reason, never as waiting +// for an answer nobody was shown (ADR 0397, issue 926); offered, it waits again; answered, it is decided as ever. +func TestAnAskTheRouterHoldsIsShownHeldWithTheReason(t *testing.T) { + w := imported(t, newMemStore()) + n := proposed(t, w, "author@g14", true) + w.reviewAll(t) + id := w.router.asked[0].ID + key := Key{KindDecision, n} + waiting := func() string { + g, err := w.k.Get(GetAsk{Kind: key.Kind, Number: key.Number}) + if err != nil { + t.Fatal(err) + } + return g.Waiting + } + if got := waiting(); !strings.Contains(got, AskWaiting) { + t.Fatalf("before it is held: %q", got) + } + w.router.hold(id, "Telegram cannot prove who answered") + if said, err := w.k.Reconcile(); err != nil || len(said) != 1 { + t.Fatalf("%v %v", said, err) + } + got := waiting() + if !strings.Contains(got, "held by the router") || !strings.Contains(got, "Telegram cannot prove who answered") || strings.Contains(got, AskWaiting) { + t.Fatalf("held: %q", got) + } + if e := mustGet(t, w.k.Store, key); e.Current.Status != StatusProposed || e.Current.Ask.Said != AskWaiting { + t.Fatalf("a held ask changed the record: %+v", e.Current.Ask) + } + if said, _ := w.k.Reconcile(); len(said) != 0 { + t.Fatalf("said again while nothing changed: %v", said) + } + w.router.offer(id) + if _, err := w.k.Reconcile(); err != nil { + t.Fatal(err) + } + if got := waiting(); got != "the operator's ask: "+AskWaiting { + t.Fatalf("offered: %q", got) + } + // Held, then answered: decided on the warrant, with nothing of the hold left. + w.router.hold(id, "Telegram cannot prove who answered") + _, _ = w.k.Reconcile() + w.router.answer(id, OptionApprove, w.clock.Add(time.Minute)) + if _, err := w.k.Reconcile(); err != nil { + t.Fatal(err) + } + if e := mustGet(t, w.k.Store, key); e.Current.Status != StatusAccepted || e.Current.Ask.Held != "" { + t.Fatalf("%s %+v", e.Current.Status, e.Current.Ask) + } +} diff --git a/modules/messenger/cmd/messenger/answers_test.go b/modules/messenger/cmd/messenger/answers_test.go index 3410abb8..cdffa473 100644 --- a/modules/messenger/cmd/messenger/answers_test.go +++ b/modules/messenger/cmd/messenger/answers_test.go @@ -310,7 +310,7 @@ func TestAnAskIsOfferedOnEachKindAsFarAsItCanCarryIt(t *testing.T) { } func TestWithoutALinkedAccountOnlyTheMeshMCPServerApproves(t *testing.T) { - h, tg, dt, c, _, w := newRouter(t) + h, tg, dt, c, store, w := newRouter(t) submit(t, h, "mesh-delivery", heldAsk(c, "d-1")) settle(h, c) // Telegram says only to the account linked as the operator: none linked, it is not used at all. @@ -320,10 +320,10 @@ func TestWithoutALinkedAccountOnlyTheMeshMCPServerApproves(t *testing.T) { if err := h.Bench.Ready("telegram"); err == nil || !strings.Contains(err.Error(), "send /start to the bot") { t.Errorf("an unlinked Telegram is ready: %v", err) } - // The ask offers an approval no channel can carry: refused to its asker, in words. - if got := w.of("mesh-delivery"); len(got) != 1 || got[0].Outcome != asks.OutcomeRefused || - !strings.Contains(got[0].Words, "no channel can carry") { - t.Fatalf("not refused: %+v", got) + // The ask offers an approval no channel can carry: held, not refused, with the reason (ADR 0397, issue 926). + if rec := store.asks["mesh-delivery.d-1"]; len(w.of("mesh-delivery")) != 0 || rec.State != StateHeld || + !strings.Contains(rec.HeldWhy, "no Telegram account is linked") { + t.Fatalf("not held: %+v %+v", rec, w.of("mesh-delivery")) } if len(dt.sends) != 0 { t.Errorf("the desk was offered an ask with an approval: %+v", dt.sends) @@ -600,13 +600,13 @@ func TestAnAskIsRefusedInWordsToItsAsker(t *testing.T) { if _, kept := store.asks["mesh-delivery.d-1"]; kept { t.Error("a refused ask was recorded") } - // No channel can carry anything: refused, never queued in silence. - h2, _, _, c2, _, w2 := newRouter(t) + // No channel can carry anything: held with the reason (ADR 0397), never queued in silence. + h2, _, _, c2, store2, w2 := newRouter(t) h2.Bench.Stand("telegram", Standing{Ready: false, Why: "no bot token", At: c2.now()}) h2.Bench.Stand("desktop", Standing{Ready: false, Why: "no session", At: c2.now()}) submit(t, h2, "mesh-delivery", heldAsk(c2, "d-1")) - if got := w2.of("mesh-delivery"); len(got) != 1 || !strings.Contains(got[0].Words, "no channel can carry any of its answers now") { - t.Errorf("%+v", got) + if rec := store2.asks["mesh-delivery.d-1"]; len(w2.of("mesh-delivery")) != 0 || rec.State != StateHeld || !strings.Contains(rec.HeldWhy, "no bot token") { + t.Errorf("%+v %+v", rec, w2.of("mesh-delivery")) } } diff --git a/modules/messenger/cmd/messenger/asks.go b/modules/messenger/cmd/messenger/asks.go index c98c7ace..c80c56bc 100644 --- a/modules/messenger/cmd/messenger/asks.go +++ b/modules/messenger/cmd/messenger/asks.go @@ -7,8 +7,8 @@ package main // bus grants that asker alone; the name is read from the subject, never from the body. It is recorded // in the bucket `asks` under `.` before the work is acknowledged, so a restart loses none. // - **It is refused at once, in words, on its asker's `decided` subject** when its words break a rule, -// when no ready channel can carry any of its answers, or when its asker already has three open. An -// option no channel can carry is named in the words, with where it can still be given. +// or when its asker already has three open. An ask no ready channel can carry any answer of now is held, +// not refused (ADR 0397): kept open, its reason in its record, offered when a channel can carry it. // - **Each option gets a one-time ticket per channel kind that can carry it**: a button carries only the // ticket. Bound to the ask, the option and the kind; dead when the ask closes or expires. // - **An answer is judged in order, refusing at the first failure and saying so on the channel**: the ticket @@ -38,9 +38,13 @@ var ErrChanged = errors.New("written by another writer first") // operator on 2026-10-08, until an authenticator-app code exists). const LinkedApproveAfter = time.Hour -// StateOpen is an ask not yet ended; every other state is the asks.Outcome it ended with. +// StateOpen is an ask not yet ended; every other state is the asks.Outcome it ended with, but one. const StateOpen = "open" +// StateHeld is an ask that broke no rule and that no ready channel can carry now: kept as an open ask, up to its +// expiry, and offered the moment a channel can carry it (novox/hq ADR 0397, issue 926). It counts as open. +const StateHeld = "held" + // Ticket is one option's one-time ticket on one channel kind. type Ticket struct { Option string `json:"option"` @@ -60,6 +64,11 @@ type AskRecord struct { Warrant *asks.Warrant `json:"warrant,omitempty"` // Told: the warrant, or the end, was published to the asker. Told bool `json:"told,omitempty"` + // HeldSince and HeldWhy say, while the ask is held, since when and why no channel can carry it: per kind, the + // reason from the judgement that judges an answer. The asker reads them in this record, as it reads + // the state (the asks protocol has no word for a held ask on `decided`, which means an ask ended). + HeldSince time.Time `json:"held-since,omitempty"` + HeldWhy string `json:"held,omitempty"` // Record is the message the ask is shown on: a condition's key, or its own. Record string `json:"record,omitempty"` // Only is the one account that may answer it, on its kind: an ask of the router's own, put to the account @@ -498,7 +507,7 @@ func (h *Holder) TakeAsk(verb, asker string, body json.RawMessage) error { open, replaced := 0, "" h.mu.Lock() for k, e := range h.asks { - if e.rec.Asker != asker || e.rec.State != StateOpen { + if e.rec.Asker != asker || (e.rec.State != StateOpen && e.rec.State != StateHeld) { continue } if a.About != "" && e.rec.Ask.About == a.About { @@ -512,38 +521,18 @@ func (h *Holder) TakeAsk(verb, asker string, body json.RawMessage) error { h.refuseAsk(asker, a, fmt.Sprintf("its asker already has %d open asks", open)) return nil } - rec := AskRecord{Asker: asker, Ask: a, State: StateOpen, Opened: now, Tickets: map[string]Ticket{}} - // Every option on the kinds that carry the ask's highest level: an ask with an approval is answered only - // where it could be approved (answerLevel). - for _, kind := range h.carriers(answerLevel(a)) { - for _, o := range a.Options { - rec.Tickets[ticket()] = Ticket{Option: o.ID, Kind: kind} - } - } + rec := AskRecord{Asker: asker, Ask: a, State: StateOpen, Opened: now, Tickets: h.ticketsFor(a)} if len(rec.Tickets) == 0 { - h.refuseAsk(asker, a, "no channel can carry any of its answers now: "+h.whyNot()) - return nil + // Held, not refused (ADR 0397): the ask breaks no rule, and the operator is asked when a channel can carry it. + rec.State, rec.HeldSince, rec.HeldWhy = StateHeld, now, h.holdWhy(a) } if replaced != "" { if err := h.closeAsk(replaced, asks.OutcomeReplaced, "a newer ask about the same thing replaced it", nil, "", nil); err != nil && !errors.Is(err, ErrChanged) { return err } } - // Shown on the condition it is about, when that is open here; otherwise as a message of its own. - h.mu.Lock() - target := h.open["ask."+key] - // Shown on a condition's message only when asked by the condition's owner: another asker naming a - // condition's key does not take its message over. - if a.About != "" && asker == ConditionOwner { - if r := h.open[a.About]; r != nil && r.Cleared.IsZero() && r.Origin == "condition" { - target = r - } - } - h.mu.Unlock() - if target != nil { - rec.Record = target.Key - } else { - rec.Record = "ask." + key + if rec.State == StateOpen { + rec.Record = h.recordFor(asker, key, a) } if h.Asks != nil { rev, err := h.Asks.CreateAsk(key, rec) @@ -562,6 +551,78 @@ func (h *Holder) TakeAsk(verb, asker string, body json.RawMessage) error { } else { h.keepAsk(key, rec, 0) } + if rec.State == StateHeld { + h.Logf("[messenger] %s asks %q (%s): held, not shown: %s", asker, a.Headline, a.ID, rec.HeldWhy) + return nil + } + h.showAsk(key, rec, now) + return nil +} + +// ticketsFor is every option's ticket on every kind that can carry the ask's highest level now: an ask with an +// approval is answered only where it could be approved (answerLevel). None when no ready channel can. +func (h *Holder) ticketsFor(a asks.Ask) map[string]Ticket { + out := map[string]Ticket{} + for _, kind := range h.carriers(answerLevel(a)) { + for _, o := range a.Options { + out[ticket()] = Ticket{Option: o.ID, Kind: kind} + } + } + return out +} + +// holdWhy is why no kind can carry an ask now, per kind and in the words of the judgement that judges an answer +// (Bench.Ready, the capabilities carriers asks for, approveHolds): never only "no channel" (ADR 0397 §3, issue 926). +func (h *Holder) holdWhy(a asks.Ask) string { + if h.Bench == nil || len(h.Bench.Kinds()) == 0 { + return "the mesh has not said which channels exist for answers" + } + var parts []string + for _, k := range h.Bench.Kinds() { + switch { + case h.Bench.Ready(k.Kind) != nil: + parts = append(parts, Display(k.Kind)+": "+h.Bench.Ready(k.Kind).Error()) + case !h.Bench.Promises(k.Kind, "choice") || !h.Bench.Promises(k.Kind, "exact-render"): + parts = append(parts, Display(k.Kind)+" cannot offer a choice shown exactly as the router rendered it") + case answerLevel(a) == asks.Approve: + if why := h.approveHolds(k.Kind, true); why != "" { + if !strings.HasPrefix(why, Display(k.Kind)) { + why = Display(k.Kind) + ": " + why + } + parts = append(parts, why) + } + } + } + if len(parts) == 0 { + return "no channel can carry it now" + } + return strings.Join(parts, "; ") +} + +// recordFor is the message an ask is shown on: the condition it is about when that is open here, else its own. +func (h *Holder) recordFor(asker, key string, a asks.Ask) string { + h.mu.Lock() + defer h.mu.Unlock() + // Shown on a condition's message only when asked by the condition's owner: another asker naming a + // condition's key does not take its message over. + if a.About != "" && asker == ConditionOwner { + if r := h.open[a.About]; r != nil && r.Cleared.IsZero() && r.Origin == "condition" { + return r.Key + } + } + if r := h.open["ask."+key]; r != nil { + return r.Key + } + return "ask." + key +} + +// showAsk shows an ask that was just taken or just left held: on the condition it is about, or as a message of +// its own. rec is already recorded. +func (h *Holder) showAsk(key string, rec AskRecord, now time.Time) { + asker, a := rec.Asker, rec.Ask + h.mu.Lock() + target := h.open[rec.Record] + h.mu.Unlock() h.Logf("[messenger] %s asks %q (%s), shown on %s", asker, a.Headline, a.ID, rec.Record) if target != nil { h.mu.Lock() @@ -571,7 +632,7 @@ func (h *Holder) TakeAsk(verb, asker string, body json.RawMessage) error { h.offerAsk(target) } h.persist(target) - return nil + return } severity := Warning if a.Urgent { @@ -585,7 +646,83 @@ func (h *Holder) TakeAsk(verb, asker string, body json.RawMessage) error { h.mu.Unlock() h.deliver(&r, "raised") h.persist(&r) - return nil +} + +// releaseHeld judges every held ask again: one a channel can carry now is offered as a new ask would be, its +// tickets made and its message shown; one still held has its reason kept current. Called whenever the +// controller's word on the kinds or a channel's standing changes, and every minute (ADR 0397 §2). +func (h *Holder) releaseHeld() { + h.mu.Lock() + var held []string + for k, e := range h.asks { + if e.rec.State == StateHeld { + held = append(held, k) + } + } + h.mu.Unlock() + sort.Strings(held) + now := h.Now() + for _, key := range held { + e := h.askOf(key) + if e == nil { + continue + } + rec, rev := e.rec, e.rev + if h.Asks != nil { + if got, r, err := h.Asks.GetAsk(key); err == nil && got != nil { + rec, rev = *got, r + } + } + if rec.State != StateHeld || !now.Before(rec.Ask.Expires) { + continue // ended meanwhile; expiry is the sweep's + } + rec.Tickets = h.ticketsFor(rec.Ask) + if len(rec.Tickets) == 0 { + rec.Tickets = nil + if why := h.holdWhy(rec.Ask); why != rec.HeldWhy { + rec.HeldWhy = why + h.saveHeld(key, rec, rev) + } + continue + } + rec.State, rec.HeldWhy = StateOpen, "" + rec.Record = h.recordFor(rec.Asker, key, rec.Ask) + if h.saveHeld(key, rec, rev) { + h.showAsk(key, rec, now) + } + } +} + +// saveHeld records a held ask's change by compare-and-set, and keeps it; false when another writer was first. +func (h *Holder) saveHeld(key string, rec AskRecord, rev uint64) bool { + if h.Asks != nil { + newRev, err := h.Asks.UpdateAsk(key, rec, rev) + if err != nil { + if errors.Is(err, ErrChanged) { + if got, r, gerr := h.Asks.GetAsk(key); gerr == nil && got != nil { + h.keepAsk(key, *got, r) + } + } else { + h.Logf("[messenger] could not change the held ask %s: %v", key, err) + } + return false + } + rev = newRev + } + h.keepAsk(key, rec, rev) + return true +} + +// ReleaseHeld judges the held asks again at once, outside the work's own turn: the controller's word on the +// kinds or a channel's standing changed. +func (h *Holder) ReleaseHeld() { + h.warmRootFree(false) + h.work.Lock() + defer h.work.Unlock() + h.mu.Lock() + h.init() + h.mu.Unlock() + h.releaseHeld() } // askKinds are the kinds an open ask about a record is offered on. @@ -670,12 +807,13 @@ func (h *Holder) closeAsk(key string, outcome asks.Outcome, words string, w *ask rec, rev = *got, r } } - if rec.State != StateOpen { + if rec.State != StateOpen && rec.State != StateHeld { h.keepAsk(key, rec, rev) return ErrChanged } now := h.Now() rec.State = string(outcome) + rec.HeldWhy = "" rec.Closed = now if t, ok := rec.Tickets[used]; ok { t.Used = true @@ -852,9 +990,9 @@ func (h *Holder) sweepAsks(now time.Time) { var expired, untold []string for k, e := range h.asks { switch { - case e.rec.State == StateOpen && !now.Before(e.rec.Ask.Expires): + case (e.rec.State == StateOpen || e.rec.State == StateHeld) && !now.Before(e.rec.Ask.Expires): expired = append(expired, k) - case e.rec.State != StateOpen && !e.rec.Told && e.rec.Warrant != nil: + case e.rec.State != StateOpen && e.rec.State != StateHeld && !e.rec.Told && e.rec.Warrant != nil: untold = append(untold, k) } } @@ -1154,6 +1292,9 @@ type AskView struct { Offered []string `json:"offered_on,omitempty"` Answer string `json:"answer,omitempty"` Told bool `json:"asker_told"` + // HeldSince and HeldWhy: while the ask is held, since when and why no channel can carry it yet (ADR 0397 §4). + HeldSince string `json:"held_since,omitempty"` + HeldWhy string `json:"held_why,omitempty"` } // AsksNow is every ask, open first, newest first. @@ -1165,6 +1306,9 @@ func (h *Holder) AsksNow() []AskView { v := AskView{Key: k, Asker: e.rec.Asker, Headline: e.rec.Ask.Headline, State: e.rec.State, Opened: e.rec.Opened.UTC().Format(time.RFC3339), Expires: e.rec.Ask.Expires.UTC().Format(time.RFC3339), Told: e.rec.Told} + if e.rec.State == StateHeld { + v.HeldSince, v.HeldWhy = e.rec.HeldSince.UTC().Format(time.RFC3339), e.rec.HeldWhy + } kinds := map[string]bool{} for _, t := range e.rec.Tickets { kinds[t.Kind] = true @@ -1183,8 +1327,9 @@ func (h *Holder) AsksNow() []AskView { out = append(out, v) } sort.Slice(out, func(i, j int) bool { - if (out[i].State == StateOpen) != (out[j].State == StateOpen) { - return out[i].State == StateOpen + oi, oj := out[i].State == StateOpen || out[i].State == StateHeld, out[j].State == StateOpen || out[j].State == StateHeld + if oi != oj { + return oi } return out[i].Opened > out[j].Opened }) diff --git a/modules/messenger/cmd/messenger/held_test.go b/modules/messenger/cmd/messenger/held_test.go new file mode 100644 index 00000000..1efae214 --- /dev/null +++ b/modules/messenger/cmd/messenger/held_test.go @@ -0,0 +1,142 @@ +package main + +import ( + "encoding/json" + "strings" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/asks" +) + +// A held ask (novox/hq ADR 0397, issue 926): an ask that breaks no rule but that no ready channel can carry now +// is kept, not refused, and offered the moment a channel can carry it. + +// withoutVerifiedSender is the controller's word on the kinds when Telegram's claim lacks verified-sender, as +// on 2026-10-16 (issue 930). +func withoutVerifiedSender(h *Holder) { + h.Bench.SetKinds([]KindInfo{ + {Kind: "telegram", Module: "telegram", Node: "anchor", Capabilities: []string{"deliver", "reaches-away", "choice", "edit", "exact-render"}}, + {Kind: "desktop", Module: "desk-channel", Node: "anchor", Capabilities: []string{"deliver", "choice", "edit", "exact-render", "private"}}, + }) +} + +func withVerifiedSender(h *Holder) { + h.Bench.SetKinds([]KindInfo{ + {Kind: "telegram", Module: "telegram", Node: "anchor", Capabilities: []string{"deliver", "reaches-away", "choice", "edit", "verified-sender", "exact-render"}}, + {Kind: "desktop", Module: "desk-channel", Node: "anchor", Capabilities: []string{"deliver", "choice", "edit", "exact-render", "private"}}, + }) +} + +func recordJSON(t *testing.T, store *memAsks, key string) string { + t.Helper() + raw, _ := json.Marshal(store.asks[key]) + return string(raw) +} + +func TestAnApprovalAskNoChannelCanCarryIsHeldAndItsAskerToldWhy(t *testing.T) { + h, tg, dt, c, store, w := newRouter(t) + link(h, store, c, "42", c.now().Add(-2*time.Hour)) + withoutVerifiedSender(h) + submit(t, h, "mesh-delivery", heldAsk(c, "d-1")) + settle(h, c) + for _, g := range w.of("mesh-delivery") { + t.Errorf("the asker was told it ended: %+v", g) + } + rec, kept := store.asks["mesh-delivery.d-1"] + if !kept || rec.State != StateHeld || len(rec.Tickets) != 0 { + t.Fatalf("not held: %+v", rec) + } + for _, want := range []string{"Telegram cannot prove who answered", "held"} { + if !strings.Contains(recordJSON(t, store, "mesh-delivery.d-1"), want) { + t.Errorf("the record does not say %q: %s", want, recordJSON(t, store, "mesh-delivery.d-1")) + } + } + if strings.Contains(rec.HeldWhy, "no channel") && !strings.Contains(rec.HeldWhy, "Telegram") { + t.Errorf("only no channel: %q", rec.HeldWhy) + } + if len(tg.sends) != 0 || len(dt.sends) != 0 { + t.Errorf("a held ask was shown: %d %d", len(tg.sends), len(dt.sends)) + } +} + +func TestAHeldAskIsOfferedWhenTheKindsChange(t *testing.T) { + h, tg, _, c, store, w := newRouter(t) + link(h, store, c, "42", c.now().Add(-2*time.Hour)) + withoutVerifiedSender(h) + submit(t, h, "mesh-delivery", heldAsk(c, "d-1")) + c.pass(time.Minute) + stand(h, c) + h.Tick() // still held: the reason is unchanged + if rec := store.asks["mesh-delivery.d-1"]; rec.State != StateHeld || len(tg.sends) != 0 { + t.Fatalf("offered too soon: %+v", rec) + } + withVerifiedSender(h) + h.Tick() + settle(h, c) + rec := store.asks["mesh-delivery.d-1"] + if rec.State != StateOpen || len(rec.Tickets) != 3 || rec.HeldWhy != "" { + t.Fatalf("not offered: %+v", rec) + } + if len(tg.sends) != 1 || labelsOf(tg.sends[0]) != "Release, Stop, Silence for a week" { + t.Fatalf("telegram was sent %+v", tg.sends) + } + if got := w.of("mesh-delivery"); len(got) != 0 { + t.Errorf("an offer is not an end: %+v", got) + } + // Answered as any ask is. + out, err := h.Chosen("telegram", choiceOf(ticketFor(tg.sends[0], "Release"), "42", c.now())) + if err != nil || out != "chosen" { + t.Fatalf("%s %v", out, err) + } +} + +func TestAHeldAskExpiresAsAnyOpenAskDoes(t *testing.T) { + h, _, _, c, store, w := newRouter(t) + withoutVerifiedSender(h) + submit(t, h, "mesh-delivery", heldAsk(c, "d-1")) + c.pass(3 * time.Hour) + stand(h, c) + h.Tick() + got := w.of("mesh-delivery") + if len(got) != 1 || got[0].Outcome != asks.OutcomeExpired || store.asks["mesh-delivery.d-1"].State != string(asks.OutcomeExpired) { + t.Fatalf("%+v %+v", got, store.asks["mesh-delivery.d-1"]) + } +} + +func TestAHeldAskCountsTowardTheOpenAsksOfItsAsker(t *testing.T) { + h, _, _, c, _, w := newRouter(t) + withoutVerifiedSender(h) + for i := 1; i <= 4; i++ { + a := heldAsk(c, "d-"+string(rune('0'+i))) + a.About = "about-" + string(rune('0'+i)) + submit(t, h, "mesh-delivery", a) + } + got := w.of("mesh-delivery") + if len(got) != 1 || got[0].Outcome != asks.OutcomeRefused || !strings.Contains(got[0].Words, "already has 3 open asks") { + t.Fatalf("%+v", got) + } +} + +func TestTheAsksVerbShowsAHeldAskWithSinceAndWhy(t *testing.T) { + h, _, _, c, _, _ := newRouter(t) + withoutVerifiedSender(h) + submit(t, h, "mesh-delivery", heldAsk(c, "d-1")) + raw, _ := json.Marshal(h.AsksNow()) + for _, want := range []string{`"state":"held"`, `"held_since"`, `"held_why":"`, "Telegram cannot prove who answered"} { + if !strings.Contains(string(raw), want) { + t.Errorf("the asks verb lacks %q: %s", want, raw) + } + } +} + +func TestAHeldAskNamesEveryKindAndWhyWhenNoChannelIsReady(t *testing.T) { + h, _, _, c, store, _ := newRouter(t) + h.Bench.Stand("telegram", Standing{Ready: false, Why: "no bot token", At: c.now()}) + h.Bench.Stand("desktop", Standing{Ready: false, Why: "no session", At: c.now()}) + submit(t, h, "mesh-delivery", heldAsk(c, "d-1")) + why := store.asks["mesh-delivery.d-1"].HeldWhy + if store.asks["mesh-delivery.d-1"].State != StateHeld || !strings.Contains(why, "Telegram: no bot token") || !strings.Contains(why, "the desk: no session") { + t.Fatalf("%q", why) + } +} diff --git a/modules/messenger/cmd/messenger/holder.go b/modules/messenger/cmd/messenger/holder.go index 9182672c..f24ae6d7 100644 --- a/modules/messenger/cmd/messenger/holder.go +++ b/modules/messenger/cmd/messenger/holder.go @@ -759,6 +759,7 @@ func (h *Holder) Tick() { h.mu.Unlock() sort.Slice(recs, func(i, j int) bool { return recs[i].Raised.Before(recs[j].Raised) }) h.sweepAsks(now) + h.releaseHeld() h.breakGlass() h.boundWalks(now) h.duePasses(now) diff --git a/modules/messenger/cmd/messenger/main.go b/modules/messenger/cmd/messenger/main.go index c50dcecd..484db587 100644 --- a/modules/messenger/cmd/messenger/main.go +++ b/modules/messenger/cmd/messenger/main.go @@ -481,6 +481,7 @@ func listenForChannels(h *Holder) { var s Standing if json.Unmarshal(e.Body, &s) == nil { h.Stand(kind, s) + h.ReleaseHeld() // a channel's standing changed: held asks are judged again (ADR 0397 §2) } } return nil @@ -561,6 +562,7 @@ func followKinds(h *Holder) { } } h.Bench.SetKinds(out) + h.ReleaseHeld() // a held ask is judged again when the kinds change (ADR 0397 §2) said = "" } time.Sleep(30 * time.Second) @@ -673,7 +675,7 @@ func tools(h *Holder, l *listening) []stdio.Tool { "whether condition events arrive, and the rules it applies.", Run: func(map[string]any) (any, error) { return h.Status(l.get()), nil }}, {Name: "operator-channel.asks", - Description: "The asks this router holds (novox/hq ADR 0259): open first, then the last month's — each " + + Description: "The asks this router holds (novox/hq ADR 0259): open (or held, with since when and why no channel can carry it yet) first, then the last month's — each " + "one's asker, headline, state, when it opened and expires, the channels it was offered on, how it " + "ended (the warrant: who chose what, through which channel), and whether its asker was told.", Run: func(map[string]any) (any, error) { return h.AsksNow(), nil }}, -- 2.54.0 From 77d751fe830d4358728714572bf8869f196aaf9d Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 16 Oct 2026 17:57:52 +0200 Subject: [PATCH 2/2] Test the router whole over the real bus: a held ask is offered when a channel can carry it (issue 926, ADR 0397) --- modules/messenger/cmd/messenger/bus_test.go | 380 ++++++++++++++++++++ modules/messenger/cmd/messenger/main.go | 5 +- modules/messenger/go.mod | 21 +- modules/messenger/go.sum | 31 +- 4 files changed, 428 insertions(+), 9 deletions(-) create mode 100644 modules/messenger/cmd/messenger/bus_test.go diff --git a/modules/messenger/cmd/messenger/bus_test.go b/modules/messenger/cmd/messenger/bus_test.go new file mode 100644 index 00000000..28c8135e --- /dev/null +++ b/modules/messenger/cmd/messenger/bus_test.go @@ -0,0 +1,380 @@ +package main + +import ( + "bufio" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "os" + "os/exec" + "path/filepath" + "strconv" + "strings" + "sync" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/asks" + "github.com/nats-io/nats-server/v2/server" + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" +) + +// The router started whole over a real bus (novox/hq ADR 0391, issue 926). A JetStream server is linked into the +// test and started for it alone; the module is this test binary started again as the bundle, speaking MCP on its +// standard streams to a runtime played here, which does what the tool runner does for it: its state is buckets +// of the bus, an ask submitted on the seat's subject is work handed to the bundle, what the bundle publishes on a +// seat is published on the bus, an event on the bus is handed to the bundle, and the controller's membership is +// the kinds this test says. It runs in the forge's merge check, not on the way. + +const bundleVar = "MESH_MESSENGER_TEST_BUNDLE" + +func TestMain(m *testing.M) { + if os.Getenv(bundleVar) == "1" { + kindsEvery = 200 * time.Millisecond + main() + os.Exit(0) + } + os.Exit(m.Run()) +} + +type busRuntime struct { + t *testing.T + nc *nats.Conn + js jetstream.JetStream + in io.WriteCloser + mu sync.Mutex // writes to the bundle, and the fields below + next int + wait map[string]chan json.RawMessage + + kinds []map[string]any + kindsRead int + subscribed, + taking bool + buckets map[string]jetstream.KeyValue +} + +func newBusRuntime(t *testing.T) *busRuntime { + t.Helper() + s, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: server.RANDOM_PORT, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true}) + if err != nil { + t.Fatal(err) + } + go s.Start() + if !s.ReadyForConnections(30 * time.Second) { + t.Fatal("no bus within 30s") + } + t.Cleanup(func() { s.Shutdown(); s.WaitForShutdown() }) + nc, err := nats.Connect(s.ClientURL()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(nc.Close) + js, _ := jetstream.New(nc) + return &busRuntime{t: t, nc: nc, js: js, wait: map[string]chan json.RawMessage{}, buckets: map[string]jetstream.KeyValue{}} +} + +func (r *busRuntime) bucket(name string) jetstream.KeyValue { + r.mu.Lock() + defer r.mu.Unlock() + if kv, ok := r.buckets[name]; ok { + return kv + } + kv, err := r.js.CreateOrUpdateKeyValue(context.Background(), jetstream.KeyValueConfig{Bucket: name}) + if err != nil { + r.t.Fatal(err) + } + r.buckets[name] = kv + return kv +} + +func (r *busRuntime) setKinds(caps map[string][]string) { + var out []map[string]any + for kind, c := range caps { + out = append(out, map[string]any{"seat": asks.ChannelSeat, "kind": kind, "module": kind, "node": "test", "capabilities": c}) + } + r.mu.Lock() + r.kinds = out + r.mu.Unlock() +} + +func (r *busRuntime) say(m map[string]any) { + m["jsonrpc"] = "2.0" + raw, _ := json.Marshal(m) + r.mu.Lock() + defer r.mu.Unlock() + _, _ = r.in.Write(append(raw, '\n')) +} + +// request is a request to the bundle (work, an event, a tool call) and its answer. +func (r *busRuntime) request(method string, params any) (json.RawMessage, error) { + r.mu.Lock() + r.next++ + id := "runtime-" + strconv.Itoa(r.next) + ch := make(chan json.RawMessage, 1) + r.wait[id] = ch + r.mu.Unlock() + r.say(map[string]any{"id": id, "method": method, "params": params}) + select { + case raw := <-ch: + return raw, nil + case <-time.After(60 * time.Second): + return nil, fmt.Errorf("%s was not answered in 60 s", method) + } +} + +func (r *busRuntime) start(dir string) { + r.t.Helper() + settings := filepath.Join(dir, "settings.json") + _ = os.WriteFile(settings, []byte(`{}`), 0o600) + cmd := exec.Command(os.Args[0], "-test.run=^$") + cmd.Env = append(os.Environ(), bundleVar+"=1", "MESH_MESSENGER_SETTINGS="+settings, "MESH_MESSENGER_STATE="+dir, + "MESH_SERVED_MODULE=messenger", "MESH_MODULE=messenger", "MESH_NODE=test") + in, _ := cmd.StdinPipe() + out, _ := cmd.StdoutPipe() + cmd.Stderr = busLog{r.t} + if err := cmd.Start(); err != nil { + r.t.Fatal(err) + } + r.in = in + go r.read(out) + r.t.Cleanup(func() { _ = in.Close(); _ = cmd.Wait() }) +} + +type busLog struct{ t *testing.T } + +func (l busLog) Write(p []byte) (int, error) { l.t.Logf("bundle: %s", p); return len(p), nil } + +func (r *busRuntime) read(out io.Reader) { + sc := bufio.NewScanner(out) + sc.Buffer(make([]byte, 1<<20), 1<<24) + for sc.Scan() { + var m struct { + ID json.RawMessage `json:"id"` + Method string `json:"method"` + Params json.RawMessage `json:"params"` + } + if json.Unmarshal(sc.Bytes(), &m) != nil { + continue + } + if m.Method == "" { + var id string + _ = json.Unmarshal(m.ID, &id) + r.mu.Lock() + ch := r.wait[id] + delete(r.wait, id) + r.mu.Unlock() + if ch != nil { + ch <- append([]byte(nil), sc.Bytes()...) + } + continue + } + go r.runtime(m.ID, m.Method, m.Params) + } +} + +// runtime is what the tool runner does for the bundle: only what the router uses. +func (r *busRuntime) runtime(id json.RawMessage, method string, params json.RawMessage) { + var p struct { + State string `json:"state"` + Key string `json:"key"` + Value json.RawMessage `json:"value"` + Revision uint64 `json:"revision"` + Body json.RawMessage `json:"body"` + Subject string `json:"subject"` + Pattern string `json:"pattern"` + Worker string `json:"worker"` + } + _ = json.Unmarshal(params, &p) + ctx := context.Background() + const changed = -32010 + code := -32000 + result, err := func() (any, error) { + switch method { + case "mesh/state.get": + e, err := r.bucket(p.State).Get(ctx, p.Key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return nil, nil + } + if err != nil { + return nil, err + } + return map[string]any{"key": p.Key, "value": json.RawMessage(e.Value()), "revision": e.Revision()}, nil + case "mesh/state.put": + rev, err := r.bucket(p.State).Put(ctx, p.Key, p.Value) + return map[string]any{"revision": rev}, err + case "mesh/state.create": + rev, err := r.bucket(p.State).Create(ctx, p.Key, p.Value) + if errors.Is(err, jetstream.ErrKeyExists) { + code = changed + } + return map[string]any{"revision": rev}, err + case "mesh/state.update": + rev, err := r.bucket(p.State).Update(ctx, p.Key, p.Value, p.Revision) + if err != nil { + code = changed + } + return map[string]any{"revision": rev}, err + case "mesh/state.delete": + err := r.bucket(p.State).Delete(ctx, p.Key) + return map[string]any{}, err + case "mesh/state.keys": + keys, err := r.bucket(p.State).Keys(ctx) + if errors.Is(err, jetstream.ErrNoKeysFound) { + return []string{}, nil + } + return keys, err + case "mesh/subscribe": + if strings.HasPrefix(p.Pattern, asks.IntakeSeat) { + r.mu.Lock() + r.subscribed = true + r.mu.Unlock() + } + return map[string]any{}, nil + case "mesh/seat.take": + r.mu.Lock() + r.taking = true + r.mu.Unlock() + return map[string]any{}, nil + case "mesh/seat.answer", "mesh/publish": + return map[string]any{}, nil + case "mesh/seat.kinds": + r.mu.Lock() + defer r.mu.Unlock() + r.kindsRead++ + return r.kinds, nil + case "mesh/seat.publish": + return map[string]any{}, r.nc.Publish(p.Subject, p.Body) + case "mesh/ask": + if p.Key == RootFreeVerb { + var a struct { + Machines []string `json:"machines"` + } + _ = json.Unmarshal(p.Body, &a) + var out []RootVerdict + for _, m := range a.Machines { + out = append(out, RootVerdict{Machine: m, Free: true}) + } + return map[string]any{"machines": out}, nil + } + return nil, fmt.Errorf("the test runtime answers nothing for %s", p.Key) + } + return nil, fmt.Errorf("the runtime offers no %s", method) + }() + if err != nil { + r.say(map[string]any{"id": id, "error": map[string]any{"code": code, "message": err.Error()}}) + return + } + r.say(map[string]any{"id": id, "result": result}) +} + +// forward hands the bundle what is published on the bus that it takes: an ask submitted to the seat as work, +// a channel's word as an event. +func (r *busRuntime) forward() { + if _, err := r.nc.Subscribe("mesh.seat."+asks.Seat+".accept.ask.*", func(m *nats.Msg) { + _, _ = r.request("mesh/work", map[string]any{"work": map[string]any{"worker": AsksWorker, "subject": m.Subject, "body": json.RawMessage(m.Data)}}) + }); err != nil { + r.t.Fatal(err) + } + if _, err := r.nc.Subscribe("mesh.seat."+asks.IntakeSeat+".event.>", func(m *nats.Msg) { + key := strings.Replace(strings.TrimPrefix(m.Subject, "mesh.seat."), ".event", "", 1) + _, _ = r.request("mesh/event", map[string]any{"envelope": map[string]any{"key": key, "node": "test", "body": json.RawMessage(m.Data)}}) + }); err != nil { + r.t.Fatal(err) + } +} + +func (r *busRuntime) until(what string, within time.Duration, ok func() bool) { + r.t.Helper() + for deadline := time.Now().Add(within); time.Now().Before(deadline); time.Sleep(100 * time.Millisecond) { + if ok() { + return + } + } + r.t.Fatalf("not within %s: %s", within, what) +} + +func (r *busRuntime) askRecord(key string) *AskRecord { + e, err := r.bucket("asks").Get(context.Background(), key) + if err != nil { + return nil + } + var rec AskRecord + _ = json.Unmarshal(e.Value(), &rec) + return &rec +} + +// A held ask over the real bus (ADR 0397, issue 926): taken while the controller's word on Telegram lacks a +// verified sender, it is held with the reason and its asker is told nothing of an end; once the word gives it a +// verified sender, it is offered on Telegram. +func TestAnApprovalAskIsHeldThenOfferedWhenAChannelCanCarryItOverTheBus(t *testing.T) { + r := newBusRuntime(t) + away := []string{"deliver", "reaches-away", "choice", "edit", "exact-render"} + r.setKinds(map[string][]string{"telegram": away}) + now := time.Now().UTC() + // The operator's account, linked long enough ago to approve. + id := Identity{Kind: "telegram", Identity: "42", Display: "Jo", Linked: now.Add(-3 * time.Hour), ApproveFrom: now.Add(-2 * time.Hour)} + raw, _ := json.Marshal(id) + if _, err := r.bucket("identities").Put(context.Background(), identityKey("telegram", "42"), raw); err != nil { + t.Fatal(err) + } + var told, shown []json.RawMessage // shown is kept to say nothing was shown while held + var seen sync.Mutex + collect := func(into *[]json.RawMessage) nats.MsgHandler { + return func(m *nats.Msg) { seen.Lock(); *into = append(*into, append([]byte(nil), m.Data...)); seen.Unlock() } + } + if _, err := r.nc.Subscribe(asks.DecidedSubject("mesh-delivery"), collect(&told)); err != nil { + t.Fatal(err) + } + if _, err := r.nc.Subscribe(asks.ChannelSubject(asks.Show, "telegram"), collect(&shown)); err != nil { + t.Fatal(err) + } + r.forward() + r.start(t.TempDir()) + r.until("the bundle subscribed, takes asks and read the kinds", 20*time.Second, func() bool { + r.mu.Lock() + defer r.mu.Unlock() + return r.subscribed && r.taking && r.kindsRead > 0 + }) + standing, _ := json.Marshal(Standing{Ready: true, At: time.Now().UTC()}) + if err := r.nc.Publish(asks.IntakeSubject(asks.StandingWhat, "telegram"), standing); err != nil { + t.Fatal(err) + } + time.Sleep(time.Second) + c := &clock{t: time.Now().UTC()} + submitted, _ := json.Marshal(heldAsk(c, "d-1")) + if err := r.nc.Publish(asks.AskSubject("mesh-delivery"), submitted); err != nil { + t.Fatal(err) + } + r.until("the ask is recorded held with the reason", 20*time.Second, func() bool { + rec := r.askRecord("mesh-delivery.d-1") + return rec != nil && rec.State == StateHeld && strings.Contains(rec.HeldWhy, "Telegram cannot prove who answered") + }) + rec := r.askRecord("mesh-delivery.d-1") + if len(rec.Tickets) != 0 { + t.Errorf("a held ask has tickets: %+v", rec) + } + seen.Lock() + if len(told) != 0 || len(shown) != 0 { + t.Errorf("a held ask was refused or shown: told %s, shown %s", told, shown) + } + seen.Unlock() + // The asks verb shows it held, since when, and why. + out, err := r.request("tools/call", map[string]any{"name": "operator-channel.asks", "arguments": map[string]any{}}) + if err != nil || !strings.Contains(string(out), `held_why`) || !strings.Contains(string(out), `held`) { + t.Errorf("the asks verb: %s %v", out, err) + } + // The controller's word now gives Telegram a verified sender: the ask is offered there. + r.setKinds(map[string][]string{"telegram": append(append([]string{}, away...), "verified-sender")}) + r.until("the ask is offered", 30*time.Second, func() bool { + rec := r.askRecord("mesh-delivery.d-1") + return rec != nil && rec.State == StateOpen && len(rec.Tickets) == 3 && rec.HeldWhy == "" + }) + // Its message is the router's ordinary one for a new ask, said after the burst window (BurstWindow), which a + // unit test covers with a clock; this test waits for no half minute. + if rec := r.askRecord("mesh-delivery.d-1"); rec.Tickets == nil { + t.Errorf("offered with no tickets: %+v", rec) + } +} diff --git a/modules/messenger/cmd/messenger/main.go b/modules/messenger/cmd/messenger/main.go index 484db587..649e90a5 100644 --- a/modules/messenger/cmd/messenger/main.go +++ b/modules/messenger/cmd/messenger/main.go @@ -543,6 +543,9 @@ func answerProofs(h *Holder) { } } +// kindsEvery is how often the controller's word on the kinds is read again (a test bundle reads it faster). +var kindsEvery = 30 * time.Second + // followKinds reads, every half minute, which channel kinds the controller says exist and what each // promises: never from a channel. func followKinds(h *Holder) { @@ -565,7 +568,7 @@ func followKinds(h *Holder) { h.ReleaseHeld() // a held ask is judged again when the kinds change (ADR 0397 §2) said = "" } - time.Sleep(30 * time.Second) + time.Sleep(kindsEvery) } } diff --git a/modules/messenger/go.mod b/modules/messenger/go.mod index 2f961edb..4b62f792 100644 --- a/modules/messenger/go.mod +++ b/modules/messenger/go.mod @@ -1,5 +1,22 @@ module messenger -go 1.22 +go 1.25.0 -require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261013182924-5933eb757d44 +require ( + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261013182924-5933eb757d44 + github.com/nats-io/nats-server/v2 v2.11.17 + github.com/nats-io/nats.go v1.51.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/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/nuid v1.0.1 // indirect + golang.org/x/crypto v0.50.0 // indirect + golang.org/x/sys v0.43.0 // indirect + golang.org/x/time v0.15.0 // indirect +) diff --git a/modules/messenger/go.sum b/modules/messenger/go.sum index 3d45e21c..f84463ba 100644 --- a/modules/messenger/go.sum +++ b/modules/messenger/go.sum @@ -1,8 +1,27 @@ -git.novox.be/novox/mesh-sdk/go v0.1.13 h1:Su4JYpZ+zhNovkGA2DgxiZdcuHpjw2tPJzc/7PljhXg= -git.novox.be/novox/mesh-sdk/go v0.1.13/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261013141520-9588871d6629 h1:7TYFV1Q6Mcz+lsLP/sds87tysET4rKCn3rRIQV8gNmA= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261013141520-9588871d6629/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261013145622-f4745447ba6d h1:HEJyqorKktrkJPeeJdwEo0kdrJjH3fm/YVS9MGftCqY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261013145622-f4745447ba6d/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY= 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= +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/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/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/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/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= +golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= -- 2.54.0