Merge pull request 'Hold an ask no channel can carry now and offer it when one can (issue 926, ADR 0397)' (#333) from fix/926-a-held-ask-waits-for-a-channel into main
This commit was merged in pull request #333.
This commit is contained in:
@@ -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())
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 != "" {
|
||||
|
||||
@@ -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).
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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"))
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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 `<asker>.<id>` 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
|
||||
})
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
@@ -542,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) {
|
||||
@@ -561,9 +565,10 @@ 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)
|
||||
time.Sleep(kindsEvery)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -673,7 +678,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 }},
|
||||
|
||||
@@ -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
|
||||
)
|
||||
|
||||
@@ -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=
|
||||
|
||||
Reference in New Issue
Block a user