Give an asking module a client that acts on each warrant once, and tell a lost compare-and-set by its code

README.md

@@ -99,6 +99,11 @@ compare-and-set. The runtime does each only as far as the module's membership li
 contract on top: the `Ask` an asker publishes and the checks it is held to, the `Warrant` it acts on
 (`For`, `Says`), and the `Message`, `Choice`, `LinkAsked` and `Code` a channel takes and says, with the
 subjects each travels on.
+`go/asker` is an asking module's client (ADR 0259 §10): `Ask` keeps an ask in the module's own state and then
+publishes it under the module's name; `Cancel`; `Where` reads its state from the router's record; `Listen` and
+`Take` hand the module each warrant once — marked acted on by compare-and-set before the module acts, and only
+for an ask it made, an option it offered, at that option's level, before it expired. A lost compare-and-set is
+`stdio.ErrChanged`, told by the runtime's error code.
 
 ### `primitives` — sealing, semver, resolved-env
 Seal/unseal for secrets, semantic version comparison, and the accessor a module uses to read its

go/asker/asker.go

@@ -0,0 +1,196 @@
+// Package asker is a module's client for asking a person (novox/hq ADR 0259 §10): it publishes an ask under
+// the module's own name, cancels it, reads its state from the router's record, and hands the module each
+// warrant once — only one for an ask it made, for an option it offered, at that option's level, before it
+// expired. The module then acts with its own grants.
+//
+// It keeps what it asked, and which warrants it acted on, in a state of the module's own (declared in its
+// manifest, by default `asked`): the ask is written before it is published, and a warrant is marked acted on
+// by compare-and-set before the module's handler runs, so a redelivery, or a second instance, acts once.
+package asker
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"os"
+	"strings"
+	"time"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+	"git.novox.be/novox/mesh-sdk/go/asks"
+)
+
+// Client asks as one module.
+type Client struct {
+	// Module is the asking module's name: the one the bus grants its asks under.
+	Module string
+	// State is the module's own state the asks are kept in.
+	State Store
+	// Now is the clock; nil is time.Now.
+	Now func() time.Time
+	// Publish, Record and Subscribe reach the bus; nil is the stdio SDK's, through the module's runtime.
+	Publish   func(subject string, body any, id string) error
+	Record    func(key string) (*stdio.StateEntry, error)
+	Subscribe func(pattern string, handle func(stdio.Envelope) error) error
+}
+
+// Store is where the asks are kept: stdio.State(name) is one.
+type Store interface {
+	Get(key string) (*stdio.StateEntry, error)
+	Create(key string, value any) (uint64, error)
+	Update(key string, value any, revision uint64) (uint64, error)
+}
+
+// New is a client for the module the runtime serves, keeping its asks in the state named.
+func New(state string) *Client {
+	return &Client{Module: os.Getenv("MESH_MODULE"), State: stdio.State(state)}
+}
+
+func (c *Client) now() time.Time {
+	if c.Now != nil {
+		return c.Now()
+	}
+	return time.Now()
+}
+
+// Kept is one ask as the module keeps it.
+type Kept struct {
+	Ask    asks.Ask      `json:"ask"`
+	Asked  time.Time     `json:"asked"`
+	Acted  bool          `json:"acted,omitempty"`
+	Ended  asks.Outcome  `json:"ended,omitempty"`
+	Answer *asks.Warrant `json:"answer,omitempty"`
+}
+
+func key(id string) string { return "ask." + id }
+
+// Ask checks an ask, keeps it, and publishes it under the module's own name. An ask with an id already kept
+// is refused: a new question is a new id.
+func (c *Client) Ask(a asks.Ask) error {
+	if err := a.Check(c.now()); err != nil {
+		return err
+	}
+	if _, err := c.State.Create(key(a.ID), Kept{Ask: a, Asked: c.now()}); err != nil {
+		if errors.Is(err, stdio.ErrChanged) {
+			return fmt.Errorf("an ask %s was already made: a new question is a new id", a.ID)
+		}
+		return err
+	}
+	return c.publish(asks.AskSubject(c.Module), a, a.ID)
+}
+
+// Cancel takes an ask back: the router ends it and edits every copy.
+func (c *Client) Cancel(id string) error {
+	return c.publish(asks.CancelSubject(c.Module), map[string]string{"id": id}, "cancel."+id)
+}
+
+func (c *Client) publish(subject string, body any, id string) error {
+	if c.Publish != nil {
+		return c.Publish(subject, body, id)
+	}
+	return stdio.SeatPublish(subject, body, id)
+}
+
+// Where is an ask's state as the router keeps it: its state and its warrant, if any — read under the
+// module's own name and no other's.
+func (c *Client) Where(id string) (state string, w *asks.Warrant, err error) {
+	read := c.Record
+	if read == nil {
+		read = func(k string) (*stdio.StateEntry, error) { return stdio.SeatRecord("", k) }
+	}
+	e, err := read(id)
+	if err != nil || e == nil {
+		return "", nil, err
+	}
+	var rec struct {
+		State   string        `json:"state"`
+		Warrant *asks.Warrant `json:"warrant"`
+	}
+	if err := json.Unmarshal(e.Value, &rec); err != nil {
+		return "", nil, fmt.Errorf("the router's record of %s cannot be read: %w", id, err)
+	}
+	return rec.State, rec.Warrant, nil
+}
+
+// Handled is what came of a warrant: acted on, refused (and why), or already done.
+type Handled string
+
+const (
+	Acted    Handled = "acted"
+	Refused  Handled = "refused"
+	Done     Handled = "already acted on"
+	Recorded Handled = "ended without a choice"
+)
+
+// Take hands act a warrant once, after holding it to the ask it answers. act performs the option with the
+// module's own grants; its error is said and the warrant stays acted on (a person's decision is not
+// performed twice; the module says the failure in its own record).
+func (c *Client) Take(w asks.Warrant, act func(asks.Option, asks.Warrant) error) (Handled, error) {
+	if w.Asker != c.Module {
+		return Refused, fmt.Errorf("a warrant for %s reached %s", w.Asker, c.Module)
+	}
+	e, err := c.State.Get(key(w.Ask))
+	if err != nil {
+		return "", err
+	}
+	if e == nil {
+		return Refused, fmt.Errorf("%s holds no ask %s", c.Module, w.Ask)
+	}
+	var k Kept
+	if err := json.Unmarshal(e.Value, &k); err != nil {
+		return "", err
+	}
+	if k.Acted || k.Ended != "" {
+		return Done, nil
+	}
+	if w.Outcome != asks.OutcomeChosen {
+		k.Ended = w.Outcome
+		k.Answer = &w
+		if _, err := c.State.Update(key(w.Ask), k, e.Revision); err != nil && !errors.Is(err, stdio.ErrChanged) {
+			return "", err
+		}
+		return Recorded, nil
+	}
+	o, err := w.For(c.Module, k.Ask)
+	if err != nil {
+		return Refused, err
+	}
+	// Marked before acting: of two deliveries, or two instances, only the one whose write stands acts.
+	k.Acted, k.Ended, k.Answer = true, asks.OutcomeChosen, &w
+	if _, err := c.State.Update(key(w.Ask), k, e.Revision); err != nil {
+		if errors.Is(err, stdio.ErrChanged) {
+			return Done, nil
+		}
+		return "", err
+	}
+	return Acted, act(o, w)
+}
+
+// Listen hands every warrant for this module to Take, with act. It asks the runtime for the module's
+// events matching the router's `decided` subject for this module.
+func (c *Client) Listen(act func(asks.Option, asks.Warrant) error, said func(Handled, asks.Warrant, error)) error {
+	subscribe := c.Subscribe
+	if subscribe == nil {
+		subscribe = stdio.Subscribe
+	}
+	return subscribe(asks.Seat+".decided."+c.Module, func(e stdio.Envelope) error {
+		if !strings.HasPrefix(e.Key, asks.Seat+".decided.") {
+			return nil
+		}
+		var w asks.Warrant
+		if err := json.Unmarshal(e.Body, &w); err != nil {
+			if said != nil {
+				said(Refused, w, fmt.Errorf("a warrant could not be read: %w", err))
+			}
+			return nil
+		}
+		h, err := c.Take(w, act)
+		if h == "" && err != nil {
+			return err // the state could not be reached: offered again
+		}
+		if said != nil {
+			said(h, w, err)
+		}
+		return nil
+	})
+}

go/asker/asker_test.go

@@ -0,0 +1,130 @@
+package asker
+
+import (
+	"encoding/json"
+	"strings"
+	"testing"
+	"time"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+	"git.novox.be/novox/mesh-sdk/go/asks"
+)
+
+type mem struct {
+	vals map[string]json.RawMessage
+	revs map[string]uint64
+	n    uint64
+}
+
+func (m *mem) Get(k string) (*stdio.StateEntry, error) {
+	v, ok := m.vals[k]
+	if !ok {
+		return nil, nil
+	}
+	return &stdio.StateEntry{Key: k, Value: v, Revision: m.revs[k]}, nil
+}
+func (m *mem) Create(k string, v any) (uint64, error) {
+	if _, ok := m.vals[k]; ok {
+		return 0, stdio.ErrChanged
+	}
+	return m.put(k, v)
+}
+func (m *mem) Update(k string, v any, rev uint64) (uint64, error) {
+	if rev == 0 || m.revs[k] != rev {
+		return 0, stdio.ErrChanged
+	}
+	return m.put(k, v)
+}
+func (m *mem) put(k string, v any) (uint64, error) {
+	raw, _ := json.Marshal(v)
+	m.n++
+	m.vals[k], m.revs[k] = raw, m.n
+	return m.n, nil
+}
+
+var now = time.Date(2026, 10, 8, 14, 0, 0, 0, time.UTC)
+
+func client() (*Client, map[string][]string) {
+	sent := map[string][]string{}
+	c := &Client{Module: "mesh-delivery", State: &mem{vals: map[string]json.RawMessage{}, revs: map[string]uint64{}},
+		Now: func() time.Time { return now },
+		Publish: func(subject string, body any, id string) error {
+			sent[subject] = append(sent[subject], id)
+			return nil
+		}}
+	return c, sent
+}
+
+func held() asks.Ask {
+	return asks.Ask{ID: "d-1", Headline: "Delivery of hq held", Explanation: "Needs you: release it, or stop it.",
+		Who: asks.Operator, Expires: now.Add(time.Hour), OnExpiry: "the delivery stays held",
+		Options: []asks.Option{{ID: "release", Label: "Release", Does: "it goes on", Level: asks.Approve}}}
+}
+
+func warrant() asks.Warrant {
+	return asks.Warrant{Ask: "d-1", Asker: "mesh-delivery", Outcome: asks.OutcomeChosen, Option: "release",
+		Label: "Release", Level: asks.Approve, At: now.Add(time.Minute),
+		By: &asks.Person{Who: asks.Operator, Kind: "telegram", Identity: "42", Verified: "user id verified"}}
+}
+
+func TestAnAskIsKeptThenPublishedUnderTheModulesOwnName(t *testing.T) {
+	c, sent := client()
+	if err := c.Ask(held()); err != nil {
+		t.Fatal(err)
+	}
+	if got := sent[asks.AskSubject("mesh-delivery")]; len(got) != 1 || got[0] != "d-1" {
+		t.Fatalf("%v", sent)
+	}
+	if err := c.Ask(held()); err == nil || !strings.Contains(err.Error(), "already made") {
+		t.Errorf("the same id was asked twice: %v", err)
+	}
+	bad := held()
+	bad.Who = "agent"
+	if err := c.Ask(bad); err == nil {
+		t.Error("an ask the contract refuses was published")
+	}
+}
+
+func TestAWarrantIsActedOnOnceAndOnlyForItsOwnAsk(t *testing.T) {
+	c, _ := client()
+	_ = c.Ask(held())
+	acted := 0
+	act := func(asks.Option, asks.Warrant) error { acted++; return nil }
+	if h, err := c.Take(warrant(), act); h != Acted || err != nil || acted != 1 {
+		t.Fatalf("%s %v %d", h, err, acted)
+	}
+	if h, _ := c.Take(warrant(), act); h != Done || acted != 1 {
+		t.Errorf("a redelivery was acted on: %s", h)
+	}
+	for why, change := range map[string]func(*asks.Warrant){
+		"another asker":         func(w *asks.Warrant) { w.Asker = "mesh-controller" },
+		"an ask not made":       func(w *asks.Warrant) { w.Ask = "d-9" },
+		"another level":         func(w *asks.Warrant) { w.Level = asks.Acknowledge },
+		"an option not offered": func(w *asks.Warrant) { w.Option = "delete" },
+		"after its expiry":      func(w *asks.Warrant) { w.At = now.Add(2 * time.Hour) },
+	} {
+		c2, _ := client()
+		_ = c2.Ask(held())
+		w := warrant()
+		change(&w)
+		if h, _ := c2.Take(w, act); h == Acted {
+			t.Errorf("a warrant from %s was acted on", why)
+		}
+	}
+	if acted != 1 {
+		t.Errorf("acted %d times", acted)
+	}
+}
+
+func TestAnAskEndedWithoutAChoiceIsRecordedAndNothingDone(t *testing.T) {
+	c, _ := client()
+	_ = c.Ask(held())
+	w := warrant()
+	w.Outcome, w.Option, w.By = asks.OutcomeExpired, "", nil
+	if h, err := c.Take(w, func(asks.Option, asks.Warrant) error { t.Fatal("acted"); return nil }); h != Recorded || err != nil {
+		t.Fatalf("%s %v", h, err)
+	}
+	if h, _ := c.Take(warrant(), func(asks.Option, asks.Warrant) error { t.Fatal("acted after it ended"); return nil }); h != Done {
+		t.Errorf("%s", h)
+	}
+}

go/asks/asks.go

@@ -289,6 +289,11 @@ func (w Warrant) For(asker string, a Ask) (Option, error) {
 	if w.Level != o.Level {
 		return Option{}, fmt.Errorf("the option %s is %s, and the warrant says %s", o.ID, o.Level, w.Level)
 	}
+	// A choice made after the ask expired is no answer to it, whatever the router said.
+	if !w.At.IsZero() && w.At.After(a.Expires) {
+		return Option{}, fmt.Errorf("the warrant was given at %s, after the ask %s expired at %s",
+			w.At.UTC().Format(time.RFC3339), a.ID, a.Expires.UTC().Format(time.RFC3339))
+	}
 	return o, nil
 }
 

go/stdio.go

@@ -48,6 +48,7 @@ type message struct {
 	Params json.RawMessage  `json:"params,omitempty"`
 	Result json.RawMessage  `json:"result,omitempty"`
 	Error  *struct {
+		Code    int    `json:"code"`
 		Message string `json:"message"`
 	} `json:"error,omitempty"`
 }
@@ -114,7 +115,7 @@ func (c *channel) askAs(id, method string, params any, timeout time.Duration) (j
 	select {
 	case m := <-answer:
 		if m.Error != nil {
-			return nil, &refused{m.Error.Message}
+			return nil, &refused{message: m.Error.Message, code: m.Error.Code}
 		}
 		return m.Result, nil
 	case <-time.After(timeout):
@@ -126,10 +127,23 @@ func (c *channel) askAs(id, method string, params any, timeout time.Duration) (j
 }
 
 // refused is the runtime's error answer, as opposed to no answer: what it refused, it did not do.
-type refused struct{ message string }
+type refused struct {
+	message string
+	code    int
+}
 
 func (r *refused) Error() string { return r.message }
 
+// Is makes a compare-and-set the runtime refused because another writer was first ErrChanged.
+func (r *refused) Is(target error) bool { return target == ErrChanged && r.code == CodeChanged }
+
+// ErrChanged is a compare-and-set that lost: the key was made or changed by another writer first (novox/hq
+// ADR 0259). Told apart by the runtime's error code, never by its words.
+var ErrChanged = errors.New("the key was written by another writer first")
+
+// CodeChanged is the error code the runtime answers a lost compare-and-set with.
+const CodeChanged = -32010
+
 var (
 	current   *channel
 	currentMu sync.Mutex