Carry the gate's merge asks to the router and write the operator's answer back, a warrant that does not hold said as a condition (issue 1343)

modules/mesh-delivery-teller/cmd/mesh-delivery-teller/decided.go

@@ -0,0 +1,115 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"time"
+
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+	"git.novox.be/novox/mesh-sdk/go/asks"
+)
+
+// The teller's three bus adapters for mesh-sdk go/asker, which in a bundle reaches the bus through the runtime: here the
+// teller is a process on its own credential. What each may do is the controller's composed grant for a module that uses
+// the seat operator-channel (mesh-control internal/broker seattraffic.go SeatTrafficOf): publish its own ask, cancel and
+// condition, hear its own decided event, and read the router's record of its own asks.
+
+// callTimeout bounds a read of the router's record.
+const callTimeout = 30 * time.Second
+
+// publishVia publishes an accept of the seat, de-duplicated by id: the same id twice is one message.
+func publishVia(ctx context.Context, js jetstream.JetStream) func(subject string, body any, id string) error {
+	return func(subject string, body any, id string) error {
+		data, err := json.Marshal(body)
+		if err != nil {
+			return err
+		}
+		var opts []jetstream.PublishOpt
+		if id != "" {
+			opts = append(opts, jetstream.WithMsgID(id))
+		}
+		_, err = js.Publish(ctx, subject, data, opts...)
+		return err
+	}
+}
+
+// recordVia reads the router's record of one of this module's asks: the key `<module>.<id>` of the router's bucket asks,
+// by a direct get on the subject the controller grants for it, `$JS.API.DIRECT.GET.KV_asks.$KV.asks.<module>.>`. The
+// record is {state, warrant}. An ask the router holds no record of is nil.
+func recordVia(ctx context.Context, conn *nats.Conn, module string) func(id string) (*stdio.StateEntry, error) {
+	return func(id string) (*stdio.StateEntry, error) {
+		ctx, cancel := context.WithTimeout(ctx, callTimeout)
+		defer cancel()
+		msg, err := conn.RequestWithContext(ctx, "$JS.API.DIRECT.GET.KV_asks.$KV.asks."+module+"."+id, nil)
+		if errors.Is(err, nats.ErrNoResponders) {
+			return nil, nil
+		}
+		if err != nil {
+			return nil, err
+		}
+		if status := msg.Header.Get("Status"); status == "404" || len(msg.Data) == 0 {
+			return nil, nil
+		}
+		return &stdio.StateEntry{Key: id, Value: msg.Data}, nil
+	}
+}
+
+// decidedVia hears the warrants the router says to this module alone: its own `decided` event, a subject only the router
+// publishes (the grant to subscribe it names the module). The handler is the asker's: pattern is the key it expects.
+func decidedVia(conn *nats.Conn, module string, logf func(string, ...any)) func(pattern string, handle func(stdio.Envelope) error) error {
+	return func(pattern string, handle func(stdio.Envelope) error) error {
+		_, err := conn.Subscribe(asks.DecidedSubject(module), func(m *nats.Msg) {
+			if err := handle(stdio.Envelope{Key: pattern, Body: m.Data}); err != nil {
+				logf("a warrant could not be handled and is read again from the router's record at the next start: %v", err)
+			}
+		})
+		return err
+	}
+}
+
+// sayVia publishes a condition on the seat's one subject for conditions, under the module's own name.
+func sayVia(ctx context.Context, js jetstream.JetStream, module string) func(c asks.Condition) error {
+	return func(c asks.Condition) error {
+		subject, err := asks.ConditionSubject(module)
+		if err != nil {
+			return err
+		}
+		data, err := asks.EncodeCondition(c)
+		if err != nil {
+			return err
+		}
+		_, err = js.Publish(ctx, subject, data)
+		return err
+	}
+}
+
+// watchAsks hands each key of the gate's bucket asks, with its revision, to put, and each key taken away to gone, until
+// ctx ends or the watch fails.
+func watchAsks(ctx context.Context, kv jetstream.KeyValue, put func(key string, rev uint64, value []byte), gone func(key string)) error {
+	w, err := kv.Watch(ctx, ">")
+	if err != nil {
+		return err
+	}
+	defer w.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case e, ok := <-w.Updates():
+			if !ok {
+				return errors.New("the watch ended")
+			}
+			switch {
+			case e == nil:
+			case e.Operation() == jetstream.KeyValuePut:
+				put(e.Key(), e.Revision(), e.Value())
+			default:
+				gone(e.Key())
+			}
+		}
+	}
+}

modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go

@@ -16,6 +16,9 @@ import (
 	"github.com/nats-io/nats-server/v2/server"
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
+
+	"git.novox.be/novox/mesh-sdk/go/asks"
+	"git.novox.be/novox/mesh-sdk/go/delivery"
 )
 
 // busConf is the shape of the composed accounts file (mesh-controller internal/broker): one account with JetStream, the
@@ -121,3 +124,275 @@ func TestTellerWaitsForTheGate(t *testing.T) { // 1240/AC2
 		t.Errorf("the teller stopped with %v, want a clean stop", err)
 	}
 }
+
+// The principals of the composed accounts file for the teller's tests, with the permissions mesh-controller composes for
+// them (internal/broker): a module's own state is the read grants of its bucket and `$KV.<bucket>.>` to write it
+// (state.go stateGrants, readGrants), a read of another module's bucket is the read grants alone, and a module that uses
+// the seat operator-channel submits its own `accept.ask.<module>`, `accept.cancel.<module>` and `accept.condition.<module>`,
+// hears its own `event.decided.<module>` and reads `$KV.asks.<module>.>` of the router's record (seattraffic.go
+// SeatTrafficOf, with the messenger's seat declaring accepts ask, cancel, condition, emits decided, by-caller and records
+// asks). The gate is a bundle on the tool runner's account (ADR 0437 §3), so the gate's writes are the account node-tools'.
+// The permissions are these lists written out, not the controller's function: the broker tests of issue 1243 prove that.
+func readOf(bucket string) []string {
+	stream := "KV_" + bucket
+	return []string{"$JS.API.STREAM.INFO." + stream, "$JS.API.DIRECT.GET." + stream + ".>", "$JS.API.CONSUMER.CREATE." + stream + ".>",
+		"$JS.API.CONSUMER.DELETE." + stream + ".>", "$JS.FC." + stream + ".>"}
+}
+
+func keepsOf(bucket string) []string { return append(readOf(bucket), "$KV."+bucket+".>") }
+
+func user(name string, pub, sub []string) string {
+	q := func(l []string) string {
+		var b strings.Builder
+		for _, s := range l {
+			b.WriteString(fmt.Sprintf("            %q\n", s))
+		}
+		return b.String()
+	}
+	return fmt.Sprintf(`      { user: %q, password: "pw"
+        permissions {
+          publish { allow: [
+%s          ] }
+          subscribe { allow: [
+%s          ] }
+          allow_responses: true
+        }
+      }
+`, name, q(pub), q(append([]string{"_INBOX." + name + ".>"}, sub...)))
+}
+
+const (
+	tellerAnswers = "mesh-delivery-teller_answers"
+	tellerAsked   = "mesh-delivery-teller_asked"
+	gateAsksKV    = "mesh-delivery-gate_asks"
+	seatAsk       = "mesh.seat.operator-channel.accept.ask.mesh-delivery-teller"
+	seatCancel    = "mesh.seat.operator-channel.accept.cancel.mesh-delivery-teller"
+	seatCondition = "mesh.seat.operator-channel.accept.condition.mesh-delivery-teller"
+	seatDecided   = "mesh.seat.operator-channel.event.decided.mesh-delivery-teller"
+)
+
+func fullConf(dir string) string {
+	teller := append(append(append(keepsOf(tellerAnswers), keepsOf(tellerAsked)...), readOf(gateAsksKV)...),
+		seatAsk, seatCancel, seatCondition, "$JS.API.DIRECT.GET.KV_asks.$KV.asks.mesh-delivery-teller.>")
+	view := append(readOf(tellerAnswers), readOf(gateAsksKV)...)
+	tools := append(keepsOf(gateAsksKV), readOf(tellerAnswers)...)
+	return fmt.Sprintf(`
+listen: 127.0.0.1:-1
+jetstream { store_dir: %q }
+accounts {
+  MESH {
+    jetstream: enabled
+    users = [
+      { user: "ctl", password: "pw" }
+%s%s%s    ]
+  }
+}
+`, filepath.Join(dir, "js"),
+		user("n1.mesh-delivery-teller", teller, []string{seatDecided}),
+		user("n1.mesh-delivery-view", view, nil),
+		user("n1.node-tools", tools, nil))
+}
+
+type fullBed struct {
+	srv *server.Server
+	ctl *nats.Conn
+	js  jetstream.JetStream
+	ctx context.Context
+}
+
+// startFull starts the bus with the principals above and what the controller makes before any module starts: the buckets
+// of the gate's asks, the teller's answers and asked, the router's work queue SEAT_OPERATOR_CHANNEL.
+func startFull(t *testing.T) *fullBed {
+	t.Helper()
+	dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "t")
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { os.RemoveAll(dir) })
+	conf := filepath.Join(dir, "bus.conf")
+	if err := os.WriteFile(conf, []byte(fullConf(dir)), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	opts, err := server.ProcessConfigFile(conf)
+	if err != nil {
+		t.Fatal(err)
+	}
+	opts.NoLog, opts.NoSigs = true, true
+	srv, err := server.NewServer(opts)
+	if err != nil {
+		t.Fatal(err)
+	}
+	go srv.Start()
+	if !srv.ReadyForConnections(10 * time.Second) {
+		t.Fatal("the bus did not come up")
+	}
+	t.Cleanup(srv.Shutdown)
+	ctl, err := nats.Connect(srv.ClientURL(), nats.UserInfo("ctl", "pw"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(ctl.Close)
+	js, err := jetstream.New(ctl)
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
+	t.Cleanup(cancel)
+	for name, history := range map[string]uint8{gateAsksKV: delivery.AsksHistory, tellerAnswers: delivery.AnswersHistory, tellerAsked: 1} {
+		if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: history}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "SEAT_OPERATOR_CHANNEL", Subjects: []string{"mesh.seat.operator-channel.accept.>"},
+		Retention: jetstream.WorkQueuePolicy}); err != nil {
+		t.Fatal(err)
+	}
+	return &fullBed{srv: srv, ctl: ctl, js: js, ctx: ctx}
+}
+
+func (b *fullBed) as(t *testing.T, name string) jetstream.JetStream {
+	t.Helper()
+	c, err := nats.Connect(b.srv.ClientURL(), nats.UserInfo(name, "pw"), nats.CustomInboxPrefix("_INBOX."+name))
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(c.Close)
+	js, err := jetstream.New(c)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return js
+}
+
+// put writes a key as a principal, within a bounded time: a write the permissions refuse reaches the writer as a timeout.
+func (b *fullBed) put(t *testing.T, js jetstream.JetStream, bucket, key string) error {
+	t.Helper()
+	ctx, cancel := context.WithTimeout(b.ctx, 3*time.Second)
+	defer cancel()
+	kv, err := js.KeyValue(ctx, bucket)
+	if err != nil {
+		t.Fatalf("%s cannot be bound: %v", bucket, err)
+	}
+	_, err = kv.Put(ctx, key, []byte(`{}`))
+	return err
+}
+
+// TestCredentialsWriteOnlyTheirOwnBuckets: the teller's credential is refused writing the gate's asks, and the gate's
+// (the tool runner's) and the view's are refused writing the teller's answers; each writes where it may, so a refusal is the
+// permissions' and not a bus that does not take writes (1343/AC4, ADR 0437 §6's check).
+func TestCredentialsWriteOnlyTheirOwnBuckets(t *testing.T) { // 1343/AC4
+	b := startFull(t)
+	teller, view, tools := b.as(t, "n1.mesh-delivery-teller"), b.as(t, "n1.mesh-delivery-view"), b.as(t, "n1.node-tools")
+	if err := b.put(t, teller, tellerAnswers, "merge.g.0123456789ab"); err != nil {
+		t.Errorf("the teller cannot write its own answers: %v", err)
+	}
+	if err := b.put(t, tools, gateAsksKV, "merge.g.0123456789ab"); err != nil {
+		t.Errorf("the gate's account cannot write the gate's asks: %v", err)
+	}
+	if err := b.put(t, teller, gateAsksKV, "merge.g.ffffffffffff"); err == nil {
+		t.Error("the teller's credential wrote the gate's asks")
+	}
+	if err := b.put(t, tools, tellerAnswers, "merge.g.ffffffffffff"); err == nil {
+		t.Error("the gate's account wrote the teller's answers")
+	}
+	if err := b.put(t, view, tellerAnswers, "merge.g.ffffffffffff"); err == nil {
+		t.Error("the view's credential wrote the teller's answers")
+	}
+	// What was refused is not there.
+	for _, c := range []struct{ bucket, key string }{{gateAsksKV, "merge.g.ffffffffffff"}, {tellerAnswers, "merge.g.ffffffffffff"}} {
+		kv, _ := b.js.KeyValue(b.ctx, c.bucket)
+		if _, err := kv.Get(b.ctx, c.key); err == nil {
+			t.Errorf("%s holds %s though its write was refused", c.bucket, c.key)
+		}
+	}
+}
+
+// TestTellerCarriesAnAskAndWritesTheAnswer starts the program on the composed principal against the bus: the gate's account
+// writes an ask, the teller publishes it to the seat's work queue under its own name, the router (the test) says the
+// warrant on the teller's decided subject, and the answer is under the ask's key in the teller's bucket; withdrawing a
+// second ask cancels it (1343/AC1, AC2).
+func TestTellerCarriesAnAskAndWritesTheAnswer(t *testing.T) { // 1343/AC1, AC2
+	now = time.Now().UTC()
+	b := startFull(t)
+	tools := b.as(t, "n1.node-tools")
+	broker := filepath.Join(t.TempDir(), "broker")
+	cred, _ := json.Marshal(credential{URL: b.srv.ClientURL(), Node: "n1", Module: "mesh-delivery-teller", User: "n1.mesh-delivery-teller", Password: "pw"})
+	if err := os.WriteFile(broker, cred, 0o600); err != nil {
+		t.Fatal(err)
+	}
+	env := map[string]string{"MESH_BROKER_FILE": broker}
+	logs := &syncBuffer{}
+	rn := startRunner(b.ctx, func(c context.Context) error { return run(c, func(k string) string { return env[k] }, logs) })
+	t.Cleanup(func() { rn.finish(t, 10*time.Second) })
+
+	key, value, ga := gateAsk(t, twoHeads(), twoHeads())
+	kv, err := tools.KeyValue(b.ctx, gateAsksKV)
+	if err != nil {
+		t.Fatal(err)
+	}
+	// The program hears warrants before it reads the asks; give it its subscription before the ask is written.
+	time.Sleep(500 * time.Millisecond)
+	if _, err := kv.Put(b.ctx, key, value); err != nil {
+		t.Fatal(err)
+	}
+	seat, err := b.js.Stream(b.ctx, "SEAT_OPERATOR_CHANNEL")
+	if err != nil {
+		t.Fatal(err)
+	}
+	var raw *jetstream.RawStreamMsg
+	for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
+		if raw, err = seat.GetLastMsgForSubject(b.ctx, seatAsk); err == nil {
+			break
+		}
+	}
+	if raw == nil {
+		t.Fatalf("no ask on the seat's queue: %v\n%s", err, logs.String())
+	}
+	var asked asks.Ask
+	if err := json.Unmarshal(raw.Data, &asked); err != nil || asked.Digest() != ga.Ask.Digest() {
+		t.Fatalf("the ask on the queue is %+v (%v), want the gate's", asked, err)
+	}
+	// The router says the warrant.
+	w, _ := json.Marshal(warrantFor(ga.Ask, "approve"))
+	if err := b.ctl.Publish(seatDecided, w); err != nil {
+		t.Fatal(err)
+	}
+	answers, err := b.js.KeyValue(b.ctx, tellerAnswers)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var e jetstream.KeyValueEntry
+	for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
+		if e, err = answers.Get(b.ctx, delivery.AnswerKey(key)); err == nil {
+			break
+		}
+	}
+	if e == nil {
+		t.Fatalf("no answer under %s: %v\n%s", key, err, logs.String())
+	}
+	if a, err := delivery.DecodeAnswer(e.Value()); err != nil || a.Option != "approve" {
+		t.Errorf("answer %+v err %v", a, err)
+	}
+	// A second ask, withdrawn by the gate, is cancelled at the router.
+	other := []delivery.Head{{Repository: "novox/mesh-delivery", Number: 13, Head: headB}}
+	key2, value2, ga2 := gateAsk(t, other, other)
+	if _, err := kv.Put(b.ctx, key2, value2); err != nil {
+		t.Fatal(err)
+	}
+	for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
+		if _, err := seat.GetLastMsgForSubject(b.ctx, seatAsk); err == nil && strings.Contains(logs.String(), ga2.Ask.ID) {
+			break
+		}
+	}
+	if err := kv.Delete(b.ctx, key2); err != nil {
+		t.Fatal(err)
+	}
+	for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
+		if raw, err = seat.GetLastMsgForSubject(b.ctx, seatCancel); err == nil {
+			break
+		}
+	}
+	if err != nil || string(raw.Data) != `{"id":"`+ga2.Ask.ID+`"}` {
+		t.Errorf("the withdrawn ask was not cancelled: %v %v\n%s", raw, err, logs.String())
+	}
+}

modules/mesh-delivery-teller/cmd/mesh-delivery-teller/kv.go

@@ -0,0 +1,101 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+
+	"github.com/nats-io/nats.go/jetstream"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+)
+
+// kvStore is a state bucket of the teller as the asker's store (mesh-sdk go/asker Store): a compare-and-set that lost is
+// stdio.ErrChanged, as the runtime's state reports it.
+type kvStore struct {
+	ctx context.Context
+	kv  jetstream.KeyValue
+}
+
+func (s kvStore) Get(key string) (*stdio.StateEntry, error) {
+	e, err := s.kv.Get(s.ctx, key)
+	if errors.Is(err, jetstream.ErrKeyNotFound) {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, err
+	}
+	return &stdio.StateEntry{Key: key, Value: e.Value(), Revision: e.Revision()}, nil
+}
+
+func (s kvStore) Create(key string, value any) (uint64, error) {
+	b, err := json.Marshal(value)
+	if err != nil {
+		return 0, err
+	}
+	rev, err := s.kv.Create(s.ctx, key, b)
+	if errors.Is(err, jetstream.ErrKeyExists) {
+		return 0, fmt.Errorf("%s: %w", key, stdio.ErrChanged)
+	}
+	return rev, err
+}
+
+func (s kvStore) Update(key string, value any, revision uint64) (uint64, error) {
+	b, err := json.Marshal(value)
+	if err != nil {
+		return 0, err
+	}
+	rev, err := s.kv.Update(s.ctx, key, b, revision)
+	if errors.Is(err, jetstream.ErrKeyExists) {
+		return 0, fmt.Errorf("%s: %w", key, stdio.ErrChanged)
+	}
+	return rev, err
+}
+
+func (s kvStore) Delete(key string) error {
+	err := s.kv.Delete(s.ctx, key)
+	if errors.Is(err, jetstream.ErrKeyNotFound) {
+		return nil
+	}
+	return err
+}
+
+// kvAnswers is the bucket answers: the teller's, written by compare-and-set only when the key has no answer.
+type kvAnswers struct {
+	ctx context.Context
+	kv  jetstream.KeyValue
+}
+
+func (a kvAnswers) Create(key string, value []byte) (bool, error) {
+	_, err := a.kv.Create(a.ctx, key, value)
+	if errors.Is(err, jetstream.ErrKeyExists) {
+		return false, nil
+	}
+	return err == nil, err
+}
+
+func (a kvAnswers) Has(key string) (bool, error) {
+	_, err := a.kv.Get(a.ctx, key)
+	if errors.Is(err, jetstream.ErrKeyNotFound) {
+		return false, nil
+	}
+	return err == nil, err
+}
+
+// kvGate reads the gate's bucket asks, which the teller may watch and read and never write.
+type kvGate struct {
+	ctx context.Context
+	kv  jetstream.KeyValue
+}
+
+func (g kvGate) Get(key string) ([]byte, bool, error) {
+	e, err := g.kv.Get(g.ctx, key)
+	if errors.Is(err, jetstream.ErrKeyNotFound) {
+		return nil, false, nil
+	}
+	if err != nil {
+		return nil, false, err
+	}
+	return e.Value(), true, nil
+}

modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go

@@ -1,8 +1,9 @@
-// Command mesh-delivery-teller is delivery's teller (novox/hq ADR 0437 §4, issue 1240). It will carry the gate's asks to
-// the operator and write the answers back (release 3d). In release 3c nothing writes asks, so it only runs: it connects
-// as itself, says once that it waits for the gate's asks, and writes and publishes nothing. It does not open the asks
-// bucket, for waiting for an absent bucket needs a timer (ADR 0437 §8 lets the parts hold none); the watch comes with the
-// gate. It serves no request subject.
+// Command mesh-delivery-teller is delivery's teller (novox/hq ADR 0437 §4, §6, §9; issues 1240, 1343). It carries the
+// gate's merge asks to the operator and writes the answers back: it watches the gate's asks, asks each of the router as
+// itself, and on the router's warrant, once it holds, writes the answer under the ask's key. Where the gate's asks
+// bucket does not exist yet it says once that it waits for the gate's asks and does nothing: waiting for an absent bucket
+// needs a timer (ADR 0437 §8 lets the parts hold none), so it is started again with the gate. It serves no request
+// subject.
 package main
 
 import (
@@ -22,6 +23,10 @@ import (
 	"syscall"
 
 	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+
+	"git.novox.be/novox/mesh-sdk/go/asker"
+	"git.novox.be/novox/mesh-sdk/go/delivery"
 )
 
 func main() {
@@ -103,10 +108,43 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		return fmt.Errorf("connecting to the bus: %w", err)
 	}
 	defer conn.Close()
-	logger.Print(waiting)
-	<-ctx.Done()
+	js, err := jetstream.New(conn)
+	if err != nil {
+		return err
+	}
+	gate, err := js.KeyValue(ctx, delivery.AsksBucket)
+	if errors.Is(err, jetstream.ErrBucketNotFound) {
+		logger.Print(waiting)
+		<-ctx.Done()
+		return nil
+	}
+	if err != nil {
+		return fmt.Errorf("the bucket %s: %w", delivery.AsksBucket, err)
+	}
+	var kvs [2]jetstream.KeyValue
+	for i, name := range []string{delivery.AnswersBucket, askedBucket} {
+		if kvs[i], err = js.KeyValue(ctx, name); err != nil {
+			return fmt.Errorf("the bucket %s: %w", name, err)
+		}
+	}
+	state := kvStore{ctx: ctx, kv: kvs[1]}
+	client := &asker.Client{Module: delivery.TellerModule, State: state, Publish: publishVia(ctx, js),
+		Record: recordVia(ctx, conn, delivery.TellerModule), Subscribe: decidedVia(conn, delivery.TellerModule, logger.Printf)}
+	t := &Teller{Asker: client, Store: state, Answers: kvAnswers{ctx: ctx, kv: kvs[0]}, Gate: kvGate{ctx: ctx, kv: gate},
+		Say: sayVia(ctx, js, delivery.TellerModule), Logf: logger.Printf}
+	// Hear the warrants first, so none that comes while the asks are read is missed.
+	if err := client.Listen(t.Answer, t.Said); err != nil {
+		return fmt.Errorf("hearing the router's warrants: %w", err)
+	}
+	if err := watchAsks(ctx, gate, t.OnAsk, t.OnAskGone); err != nil {
+		return fmt.Errorf("watching the gate's asks: %w", err)
+	}
 	return nil
 }
 
+// askedBucket is the teller's state bucket asked, `<module>_<name>`: what it asked of the router and which condition it
+// raised (manifest state entry asked, novox/hq ADR 0437 §4).
+const askedBucket = delivery.TellerModule + "_asked"
+
 // waiting is said once at start.
-const waiting = "waiting for the gate's asks, which come with the gate in release 3d; nothing is read, written or published until then"
+const waiting = "waiting for the gate's asks, whose bucket does not exist yet; the teller is started again with the gate, and nothing is read, written or published until then"

modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller.go

@@ -0,0 +1,383 @@
+package main
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"sort"
+	"strings"
+	"sync"
+	"time"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+	"git.novox.be/novox/mesh-sdk/go/asker"
+	"git.novox.be/novox/mesh-sdk/go/asks"
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+// The teller carries the gate's merge asks to the operator and writes the operator's answer back (novox/hq ADR 0437
+// §6, §9). It reads the gate's asks, publishes each to the router unchanged as the asker mesh-delivery-teller
+// (mesh-sdk go/asker), withdraws it when the gate takes its key away, and on the warrant the router says to it
+// holds the warrant to the ask (ADR 0259 §6) and writes the answer under the ask's own key by compare-and-set. A
+// warrant that does not hold writes no answer and is said as a condition. It holds no timer and serves no request:
+// every reaction is a revision of the gate's key or a warrant.
+
+// store is where the teller keeps what it asked: the asker's own records (`ask.<id>`) and, beside them, which gate key
+// each ask was made for and at which revision of it (`key.<id>`), and which conditions it has raised (`cond.<id>`).
+type store interface {
+	asker.Store
+	Delete(key string) error
+}
+
+// routerAsks is the asker.Client as the teller uses it.
+type routerAsks interface {
+	Ask(a asks.Ask) error
+	Cancel(id string) error
+	Where(id string) (state string, w *asks.Warrant, err error)
+	Take(w asks.Warrant, act func(asks.Option, asks.Warrant) error) (asker.Handled, error)
+}
+
+// answers is the teller's bucket answers.
+type answers interface {
+	// Create writes the answer under the ask's key only when there is none; created is false when there is one.
+	Create(key string, value []byte) (created bool, err error)
+	// Has says whether there is an answer under the key.
+	Has(key string) (bool, error)
+}
+
+// gateAsks reads the gate's bucket asks.
+type gateAsks interface {
+	Get(key string) (value []byte, found bool, err error)
+}
+
+// Teller is the teller's logic; the program's wiring is main.go and decided.go.
+type Teller struct {
+	Asker   routerAsks
+	Store   store
+	Answers answers
+	Gate    gateAsks
+	// Say publishes a condition to the operator (mesh.seat.operator-channel.accept.condition.<module>).
+	Say  func(c asks.Condition) error
+	Logf func(format string, args ...any)
+
+	mu sync.Mutex
+}
+
+func (t *Teller) logf(format string, args ...any) {
+	if t.Logf != nil {
+		t.Logf(format, args...)
+	}
+}
+
+// seen is what the teller keeps beside an ask it made: the gate key it was made for and the revision of that key it was
+// made at, so a replay of the revision asks nothing again.
+type seen struct {
+	Key string `json:"key"`
+	Rev uint64 `json:"rev"`
+}
+
+func recordKey(id string) string { return "key." + id }
+func keptKey(id string) string   { return "ask." + id }
+func condKey(id string) string   { return "cond." + id }
+
+func (t *Teller) seenOf(id string) (seen, *stdio.StateEntry, error) {
+	e, err := t.Store.Get(recordKey(id))
+	if err != nil || e == nil {
+		return seen{}, nil, err
+	}
+	var s seen
+	if err := json.Unmarshal(e.Value, &s); err != nil {
+		return seen{}, nil, fmt.Errorf("the record %s cannot be read: %w", recordKey(id), err)
+	}
+	return s, e, nil
+}
+
+func (t *Teller) keptOf(id string) (*asker.Kept, uint64, error) {
+	e, err := t.Store.Get(keptKey(id))
+	if err != nil || e == nil {
+		return nil, 0, err
+	}
+	var k asker.Kept
+	if err := json.Unmarshal(e.Value, &k); err != nil {
+		return nil, 0, fmt.Errorf("the record %s cannot be read: %w", keptKey(id), err)
+	}
+	return &k, e.Revision, nil
+}
+
+// open says whether an ask is made and neither answered nor ended.
+func open(k *asker.Kept) bool { return k != nil && !k.Acted && k.Ended == "" }
+
+func (t *Teller) note(id, key string, rev uint64, prior *stdio.StateEntry) error {
+	s := seen{Key: key, Rev: rev}
+	if prior == nil {
+		_, err := t.Store.Create(recordKey(id), s)
+		return err
+	}
+	_, err := t.Store.Update(recordKey(id), s, prior.Revision)
+	return err
+}
+
+// OnAsk handles the gate's asks key put at revision rev: the ask it holds is published to the router once per
+// revision of its key. A key whose value is not the ask its name says, one already answered, and a revision the teller
+// has handled are left. An ask still open and unchanged is only noted (and its warrant read from the router, in case it
+// came while the teller was away); one whose words changed is cancelled and asked afresh; one that ended, or that the
+// teller cancelled, is asked again under the same id when the gate writes its key again (ADR 0441 §1).
+func (t *Teller) OnAsk(key string, rev uint64, value []byte) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	ga, err := delivery.DecodeAsk(value)
+	if err != nil {
+		t.logf("the gate's key %s at revision %d is not an ask, and is left: %v", key, rev, err)
+		return
+	}
+	if want, err := delivery.MergeAskKey(ga.Group, ga.Heads); err != nil || want != key {
+		t.logf("the gate's key %s at revision %d holds the ask of %q, not its own, and is left (%v)", key, rev, want, err)
+		return
+	}
+	id := ga.Ask.ID
+	if has, err := t.Answers.Has(key); err != nil {
+		t.logf("%s: the answers could not be read, so the ask is left: %v", key, err)
+		return
+	} else if has {
+		return
+	}
+	prior, srec, err := t.seenOf(id)
+	if err != nil {
+		t.logf("%s: %v", key, err)
+		return
+	}
+	k, krev, err := t.keptOf(id)
+	if err != nil {
+		t.logf("%s: %v", key, err)
+		return
+	}
+	if srec != nil && rev <= prior.Rev {
+		if open(k) {
+			t.catchUp(id)
+		}
+		return
+	}
+	if k != nil {
+		if open(k) && k.Ask.Digest() == ga.Ask.Digest() {
+			_ = t.note(id, key, rev, srec)
+			t.catchUp(id)
+			return
+		}
+		if open(k) {
+			if err := t.Asker.Cancel(id); err != nil {
+				t.logf("%s: the ask %s changed and could not be cancelled, so it is left: %v", key, id, err)
+				return
+			}
+			t.logf("%s: the ask %s changed at revision %d: the old one is cancelled and the new one asked", key, id, rev)
+		}
+		if err := t.Store.Delete(keptKey(id)); err != nil {
+			t.logf("%s: the earlier ask %s (revision %d of its record) could not be forgotten: %v", key, id, krev, err)
+			return
+		}
+	}
+	if err := t.note(id, key, rev, srec); err != nil {
+		t.logf("%s: the record of the ask %s could not be kept, so it is left: %v", key, id, err)
+		return
+	}
+	if err := t.Asker.Ask(ga.Ask); err != nil {
+		t.logf("%s at revision %d: the ask %s was not asked: %v", key, rev, id, err)
+		if kept, _, _ := t.keptOf(id); open(kept) {
+			// Kept but not published: forgotten, so the next revision of the key asks it.
+			_ = t.Store.Delete(keptKey(id))
+		}
+		return
+	}
+	t.logf("%s at revision %d: the ask %s is asked of the operator", key, rev, id)
+}
+
+// OnAskGone handles the gate's asks key taken away (a withdrawn ask): the ask is cancelled at the router, and an answer
+// to it merges nothing.
+func (t *Teller) OnAskGone(key string) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	if _, _, _, err := delivery.ParseAskKey(key); err != nil {
+		t.logf("the gate's key %s was taken away and is no ask key, and is left: %v", key, err)
+		return
+	}
+	id := delivery.AskID(key)
+	k, krev, err := t.keptOf(id)
+	if err != nil {
+		t.logf("%s: %v", key, err)
+		return
+	}
+	if !open(k) {
+		return
+	}
+	if err := t.Asker.Cancel(id); err != nil {
+		t.logf("%s: the withdrawn ask %s could not be cancelled: %v", key, id, err)
+		return
+	}
+	// The router's word that it ended may come later: until it does, the ask is not open here.
+	k.Ended = asks.OutcomeCancelled
+	if _, err := t.Store.Update(keptKey(id), k, krev); err != nil && !errors.Is(err, stdio.ErrChanged) {
+		t.logf("%s: the cancelled ask %s could not be marked: %v", key, id, err)
+	}
+	t.logf("%s: the gate withdrew the ask %s: cancelled", key, id)
+}
+
+// catchUp reads the router's record of an ask the teller made and hands its warrant on, as the router's event would:
+// a warrant that came while the teller was away.
+func (t *Teller) catchUp(id string) {
+	state, w, err := t.Asker.Where(id)
+	if err != nil {
+		t.logf("the router's record of the ask %s could not be read: %v", id, err)
+		return
+	}
+	if w == nil {
+		return
+	}
+	t.logf("the ask %s is %s at the router: its warrant is handled now", id, state)
+	h, err := t.Asker.Take(*w, t.Answer)
+	t.Said(h, *w, err)
+}
+
+// mergeAct is what the approve option of the gate's ask binds: this ask, the answer, and the heads it covers. It is
+// spelled as the gate composes it (mesh-catalog modules/mesh-delivery-gate approval.go mergeAct, from
+// modules/mesh-delivery approval.go:63-65) and as mesh-controller verifies it (cmd/mesh-controller/moved.go:98-106).
+func mergeAct(ask, option string, heads map[string]string) asks.Act {
+	var out []string
+	for id, sha := range heads {
+		out = append(out, id+"="+sha)
+	}
+	sort.Strings(out)
+	return asks.Act{"verb": "mesh-delivery.merge", "ask": ask, "option": option, "heads": strings.Join(out, ",")}
+}
+
+// headsOf is the heads an ask covers as an act names them: a delivery's id, `owner/repository@<twelve characters of its
+// commit>` (mesh-catalog modules/mesh-delivery model.go Delivery.ID), and its whole commit.
+func headsOf(heads []delivery.Head) map[string]string {
+	out := map[string]string{}
+	for _, h := range heads {
+		short := h.Head
+		if len(short) > 12 {
+			short = short[:12]
+		}
+		out[h.Repository+"@"+short] = h.Head
+	}
+	return out
+}
+
+// warrantID names a warrant, which carries no id of its own: the ask and the time the router gave it.
+func warrantID(w asks.Warrant) string {
+	return w.Ask + "@" + w.At.UTC().Format(time.RFC3339Nano)
+}
+
+// Answer is the act of a warrant that holds the ask (asker.Client.Take has checked its asker, the ask's digest, the
+// option and its level, and the time): the option's bound act must be the act of this ask and the heads it covers,
+// and then the answer is written under the ask's key, once. An error is a warrant that does not hold here.
+func (t *Teller) Answer(o asks.Option, w asks.Warrant) error {
+	s, srec, err := t.seenOf(w.Ask)
+	if err != nil {
+		return err
+	}
+	if srec == nil {
+		return fmt.Errorf("the teller holds no gate key for the ask %s", w.Ask)
+	}
+	raw, found, err := t.Gate.Get(s.Key)
+	if err != nil {
+		return fmt.Errorf("the gate's ask %s could not be read: %w", s.Key, err)
+	}
+	if !found {
+		t.logf("%s: the ask %s was answered and the gate has taken it away: an answer to a withdrawn ask merges nothing", s.Key, w.Ask)
+		return nil
+	}
+	ga, err := delivery.DecodeAsk(raw)
+	if err != nil {
+		return fmt.Errorf("the gate's ask %s is not an ask: %w", s.Key, err)
+	}
+	if ga.Ask.ID != w.Ask || ga.Ask.Digest() != w.AskDigest {
+		return fmt.Errorf("the gate's ask %s is no longer the ask the operator answered (its digest is %s, the warrant's %s)",
+			s.Key, ga.Ask.Digest(), w.AskDigest)
+	}
+	if err := o.Performs(mergeAct(w.Ask, o.ID, headsOf(ga.Heads))); err != nil {
+		return err
+	}
+	body, err := delivery.EncodeAnswer(delivery.Answer{Option: o.ID, Warrant: warrantID(w), Words: w.Says()})
+	if err != nil {
+		return err
+	}
+	created, err := t.Answers.Create(delivery.AnswerKey(s.Key), body)
+	if err != nil {
+		return fmt.Errorf("the answer under %s could not be written: %w", s.Key, err)
+	}
+	if !created {
+		t.logf("%s: an answer is there already, and is not written again", s.Key)
+		return nil
+	}
+	t.logf("%s: %s", s.Key, w.Says())
+	return nil
+}
+
+// Said is what the teller does with what became of a warrant: a good answer clears the condition of its ask; a warrant
+// that does not hold, or that the router refused, is said as a condition (ADR 0437 §9). Nothing is written for either.
+func (t *Teller) Said(h asker.Handled, w asks.Warrant, err error) {
+	switch {
+	case h == asker.Acted && err == nil:
+		t.logf("the warrant of %s was %s", w.Ask, h)
+		t.clear(w.Ask)
+	case h == asker.Refused || (h == asker.Acted && err != nil):
+		t.logf("the warrant of %s does not hold, and no answer is written: %v", w.Ask, err)
+		t.raise(w.Ask, "A merge Approve did not hold", "its answer did not hold here: "+oneLine(err))
+	case h == asker.Recorded && w.Outcome == asks.OutcomeRefused:
+		t.logf("the router refused the ask %s: %s", w.Ask, w.Words)
+		t.raise(w.Ask, "A merge ask was refused", "the router refused it: "+w.Words)
+	case err != nil:
+		t.logf("the warrant of %s: %v", w.Ask, err)
+	}
+}
+
+func oneLine(err error) string {
+	if err == nil {
+		return "no reason was given"
+	}
+	s := strings.Join(strings.Fields(err.Error()), " ")
+	if r := []rune(s); len(r) > 200 {
+		return string(r[:199]) + "…"
+	}
+	return s
+}
+
+// conditionKey is the key of the condition about an ask.
+func conditionKey(id string) string { return delivery.TellerModule + ".merge-ask." + id }
+
+func (t *Teller) raise(id, headline, why string) {
+	if t.Say == nil {
+		return
+	}
+	if _, err := t.Store.Create(condKey(id), seen{}); err != nil {
+		if errors.Is(err, stdio.ErrChanged) {
+			return // raised already
+		}
+		t.logf("the condition of %s could not be kept: %v", id, err)
+	}
+	needs := "to ask for the Approve again"
+	c := asks.Condition{Key: conditionKey(id), Severity: asks.SeverityWarning, Subject: asks.Scope{Scope: "merge", ID: id}, State: asks.StateRaised,
+		Summary: "The merge ask " + id + ": " + why, Headline: headline, Needs: needs,
+		Explanation: asks.NeedsYou + needs + ". The merge ask " + id + ", " + why + ". Nothing was merged.",
+		Resolved:    "A good answer to the merge ask was written."}
+	if err := t.Say(c); err != nil {
+		t.logf("the condition of %s could not be said: %v", id, err)
+		_ = t.Store.Delete(condKey(id))
+	}
+}
+
+func (t *Teller) clear(id string) {
+	if t.Say == nil {
+		return
+	}
+	e, err := t.Store.Get(condKey(id))
+	if err != nil || e == nil {
+		return
+	}
+	c := asks.Condition{Key: conditionKey(id), Subject: asks.Scope{Scope: "merge", ID: id}, State: asks.StateCleared}
+	if err := t.Say(c); err != nil {
+		t.logf("the condition of %s could not be cleared: %v", id, err)
+		return
+	}
+	_ = t.Store.Delete(condKey(id))
+}

modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller_test.go

@@ -0,0 +1,375 @@
+package main
+
+import (
+	"encoding/json"
+	"fmt"
+	"strings"
+	"testing"
+	"time"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+	"git.novox.be/novox/mesh-sdk/go/asker"
+	"git.novox.be/novox/mesh-sdk/go/asks"
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+const (
+	headA = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+	headB = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
+)
+
+var now = time.Date(2026, 10, 20, 9, 0, 0, 0, time.UTC)
+
+// memStore is a state bucket in memory with the revisions and the compare-and-set of the runtime's state.
+type memStore struct {
+	entries map[string]*stdio.StateEntry
+	seq     uint64
+}
+
+func (m *memStore) Get(key string) (*stdio.StateEntry, error) {
+	if e, ok := m.entries[key]; ok {
+		c := *e
+		return &c, nil
+	}
+	return nil, nil
+}
+
+func (m *memStore) put(key string, value any) (uint64, error) {
+	b, err := json.Marshal(value)
+	if err != nil {
+		return 0, err
+	}
+	m.seq++
+	m.entries[key] = &stdio.StateEntry{Key: key, Value: b, Revision: m.seq}
+	return m.seq, nil
+}
+
+func (m *memStore) Create(key string, value any) (uint64, error) {
+	if _, ok := m.entries[key]; ok {
+		return 0, stdio.ErrChanged
+	}
+	return m.put(key, value)
+}
+
+func (m *memStore) Update(key string, value any, revision uint64) (uint64, error) {
+	if e, ok := m.entries[key]; !ok || e.Revision != revision {
+		return 0, stdio.ErrChanged
+	}
+	return m.put(key, value)
+}
+
+func (m *memStore) Delete(key string) error { delete(m.entries, key); return nil }
+
+type memAnswers struct {
+	written map[string][]byte
+	creates int
+}
+
+func (a *memAnswers) Create(key string, value []byte) (bool, error) {
+	a.creates++
+	if _, ok := a.written[key]; ok {
+		return false, nil
+	}
+	a.written[key] = value
+	return true, nil
+}
+
+func (a *memAnswers) Has(key string) (bool, error) { _, ok := a.written[key]; return ok, nil }
+
+type memGate map[string][]byte
+
+func (g memGate) Get(key string) ([]byte, bool, error) { v, ok := g[key]; return v, ok, nil }
+
+type published struct {
+	subject string
+	body    json.RawMessage
+	id      string
+}
+
+// bed is a teller with the asker.Client of the SDK on in-memory ends: what is published, the router's records, the
+// warrants it would hear, and the conditions it says.
+type bed struct {
+	t       *Teller
+	client  *asker.Client
+	store   *memStore
+	answers *memAnswers
+	gate    memGate
+	pubs    []published
+	conds   []asks.Condition
+	records map[string]string
+	hear    func(stdio.Envelope) error
+	logs    strings.Builder
+}
+
+func newBed(t *testing.T) *bed {
+	t.Helper()
+	b := &bed{store: &memStore{entries: map[string]*stdio.StateEntry{}}, answers: &memAnswers{written: map[string][]byte{}},
+		gate: memGate{}, records: map[string]string{}}
+	b.client = &asker.Client{Module: delivery.TellerModule, State: b.store, Now: func() time.Time { return now },
+		Publish: func(subject string, body any, id string) error {
+			raw, err := json.Marshal(body)
+			b.pubs = append(b.pubs, published{subject, raw, id})
+			return err
+		},
+		Record: func(id string) (*stdio.StateEntry, error) {
+			if v, ok := b.records[id]; ok {
+				return &stdio.StateEntry{Key: id, Value: json.RawMessage(v)}, nil
+			}
+			return nil, nil
+		},
+		Subscribe: func(_ string, handle func(stdio.Envelope) error) error { b.hear = handle; return nil },
+	}
+	b.t = &Teller{Asker: b.client, Store: b.store, Answers: b.answers, Gate: b.gate,
+		Say: func(c asks.Condition) error {
+			if _, err := asks.EncodeCondition(c); err != nil {
+				t.Errorf("a condition the router would refuse: %v", err)
+			}
+			b.conds = append(b.conds, c)
+			return nil
+		},
+		Logf: func(f string, a ...any) { fmt.Fprintf(&b.logs, f+"\n", a...) }}
+	if err := b.client.Listen(b.t.Answer, b.t.Said); err != nil {
+		t.Fatal(err)
+	}
+	return b
+}
+
+// gateAsk is an ask as the gate composes it: the router's ask with the approve and decline options binding the act
+// `mesh-delivery.merge` for this ask and the heads (mesh-catalog modules/mesh-delivery approval.go:123-191 composeMergeAsk
+// and :63-65 mergeAct), under the id its key gives (mesh-sdk go/delivery AskID). boundTo are the heads the options bind,
+// which are the ask's own unless a test says another.
+func gateAsk(t *testing.T, heads, boundTo []delivery.Head) (key string, value []byte, ask delivery.Ask) {
+	t.Helper()
+	key, err := delivery.MergeAskKey("g", heads)
+	if err != nil {
+		t.Fatal(err)
+	}
+	id := delivery.AskID(key)
+	a := asks.Ask{ID: id, Headline: "Merge into main: novox/mesh-delivery#12", Explanation: "Merges two pull requests.\nReview: approved.",
+		Who: asks.Operator, Expires: now.Add(asks.ApproveLasts - 10*time.Minute), OnExpiry: "the merge stays held", About: "delivery.mesh-delivery.12"}
+	for _, o := range []struct{ id, label, does string }{{"approve", "Approve", "merges"}, {"decline", "Decline", "merges nothing"}} {
+		binds, err := asks.ActDigest(mergeAct(id, o.id, headsOf(boundTo)))
+		if err != nil {
+			t.Fatal(err)
+		}
+		a.Options = append(a.Options, asks.Option{ID: o.id, Label: o.label, Does: o.does, Level: asks.Approve, Binds: binds})
+	}
+	ask = delivery.Ask{Group: "g", Heads: heads, Ask: a}
+	value, err = delivery.EncodeAsk(ask)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return key, value, ask
+}
+
+func twoHeads() []delivery.Head {
+	return []delivery.Head{{Repository: "novox/mesh-delivery", Number: 12, Head: headA}, {Repository: "novox/mesh-sdk", Number: 4, Head: headB}}
+}
+
+// warrantFor is the router's warrant for a chosen option of an ask, as the router verified the operator.
+func warrantFor(a asks.Ask, option string) asks.Warrant {
+	return asks.Warrant{Ask: a.ID, Asker: delivery.TellerModule, Outcome: asks.OutcomeChosen, Option: option, Label: "Approve", Level: asks.Approve,
+		By: &asks.Person{Who: asks.Operator, Kind: "telegram", Identity: "42", Verified: "user id"}, Channel: "telegram",
+		At: now.Add(time.Hour), AskDigest: a.Digest()}
+}
+
+func (b *bed) deliver(t *testing.T, w asks.Warrant) {
+	t.Helper()
+	raw, _ := json.Marshal(w)
+	if err := b.hear(stdio.Envelope{Key: asks.Seat + ".decided." + delivery.TellerModule, Body: raw}); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func (b *bed) askedOf(subject string) []published {
+	var out []published
+	for _, p := range b.pubs {
+		if p.subject == subject {
+			out = append(out, p)
+		}
+	}
+	return out
+}
+
+func TestAnAsksKeyBecomesOneRouterAskAndDeletingItCancelsIt(t *testing.T) { // 1343/AC1
+	b := newBed(t)
+	key, value, ga := gateAsk(t, twoHeads(), twoHeads())
+	b.gate[key] = value
+	b.t.OnAsk(key, 3, value)
+	asked := b.askedOf(asks.AskSubject(delivery.TellerModule))
+	if len(asked) != 1 || asked[0].id != ga.Ask.ID {
+		t.Fatalf("published %+v, want one ask under %s with id %s", b.pubs, delivery.TellerModule, ga.Ask.ID)
+	}
+	var got asks.Ask
+	if err := json.Unmarshal(asked[0].body, &got); err != nil {
+		t.Fatal(err)
+	}
+	if got.Digest() != ga.Ask.Digest() || got.Headline != ga.Ask.Headline || got.Explanation != ga.Ask.Explanation || got.ID != delivery.AskID(key) {
+		t.Errorf("the router's ask differs from the gate's: %+v", got)
+	}
+	b.t.OnAsk(key, 3, value) // the revision seen again, as at a restart
+	b.t.OnAsk(key, 4, value) // the key written again with the same ask
+	if n := len(b.askedOf(asks.AskSubject(delivery.TellerModule))); n != 1 {
+		t.Errorf("the ask was published %d times", n)
+	}
+	delete(b.gate, key)
+	b.t.OnAskGone(key)
+	cancels := b.askedOf(asks.CancelSubject(delivery.TellerModule))
+	if len(cancels) != 1 || string(cancels[0].body) != `{"id":"`+ga.Ask.ID+`"}` {
+		t.Fatalf("cancels = %+v", cancels)
+	}
+	b.t.OnAskGone(key)
+	if n := len(b.askedOf(asks.CancelSubject(delivery.TellerModule))); n != 1 {
+		t.Errorf("the ask was cancelled %d times", n)
+	}
+}
+
+func TestAWarrantThatHoldsWritesOneAnswerAndASecondWritesNone(t *testing.T) { // 1343/AC2
+	b := newBed(t)
+	key, value, ga := gateAsk(t, twoHeads(), twoHeads())
+	b.gate[key] = value
+	b.t.OnAsk(key, 1, value)
+	w := warrantFor(ga.Ask, "approve")
+	b.deliver(t, w)
+	raw, ok := b.answers.written[delivery.AnswerKey(key)]
+	if !ok || len(b.answers.written) != 1 {
+		t.Fatalf("answers = %v", b.answers.written)
+	}
+	a, err := delivery.DecodeAnswer(raw)
+	if err != nil || a.Option != "approve" || a.Warrant != w.Ask+"@"+w.At.UTC().Format(time.RFC3339Nano) || !strings.Contains(a.Words, "chose Approve") {
+		t.Errorf("answer %+v err %v", a, err)
+	}
+	b.deliver(t, w) // the same warrant again
+	w2 := w
+	w2.At = w.At.Add(time.Minute) // a second warrant for the same ask
+	b.deliver(t, w2)
+	if b.answers.creates != 1 {
+		t.Errorf("the answer was written %d times, want once", b.answers.creates)
+	}
+	if len(b.conds) != 0 {
+		t.Errorf("conditions said: %+v", b.conds)
+	}
+}
+
+func TestAWarrantThatDoesNotHoldWritesNoAnswerAndRaisesTheCondition(t *testing.T) { // 1343/AC3
+	other := []delivery.Head{{Repository: "novox/mesh-delivery", Number: 12, Head: headB}, {Repository: "novox/mesh-sdk", Number: 4, Head: headB}}
+	cases := map[string]func(t *testing.T) (*bed, string, asks.Warrant){
+		"heads that differ from the ones the option bound": func(t *testing.T) (*bed, string, asks.Warrant) {
+			b := newBed(t)
+			key, value, ga := gateAsk(t, twoHeads(), other) // the options bind other heads than the ask covers
+			b.gate[key] = value
+			b.t.OnAsk(key, 1, value)
+			return b, key, warrantFor(ga.Ask, "approve")
+		},
+		"an asker that differs": func(t *testing.T) (*bed, string, asks.Warrant) {
+			b := newBed(t)
+			key, value, ga := gateAsk(t, twoHeads(), twoHeads())
+			b.gate[key] = value
+			b.t.OnAsk(key, 1, value)
+			w := warrantFor(ga.Ask, "approve")
+			w.Asker = "mesh-delivery"
+			return b, key, w
+		},
+		"a digest that differs": func(t *testing.T) (*bed, string, asks.Warrant) {
+			b := newBed(t)
+			key, value, ga := gateAsk(t, twoHeads(), twoHeads())
+			b.gate[key] = value
+			b.t.OnAsk(key, 1, value)
+			w := warrantFor(ga.Ask, "approve")
+			w.AskDigest = "sha256:0000"
+			return b, key, w
+		},
+	}
+	for name, mk := range cases {
+		t.Run(name, func(t *testing.T) {
+			b, key, w := mk(t)
+			b.deliver(t, w)
+			if len(b.answers.written) != 0 {
+				t.Errorf("an answer was written: %v", b.answers.written)
+			}
+			if len(b.conds) != 1 || b.conds[0].State != asks.StateRaised || b.conds[0].Subject.ID != delivery.AskID(key) ||
+				!strings.HasPrefix(b.conds[0].Explanation, asks.NeedsYou) {
+				t.Fatalf("conditions = %+v", b.conds)
+			}
+		})
+	}
+}
+
+func TestAGoodAnswerClearsTheConditionOfAnAskAskedAgain(t *testing.T) { // 1343/AC3 (cleared when a good answer is written)
+	b := newBed(t)
+	other := []delivery.Head{{Repository: "novox/mesh-delivery", Number: 12, Head: headB}, {Repository: "novox/mesh-sdk", Number: 4, Head: headB}}
+	key, bad, gaBad := gateAsk(t, twoHeads(), other)
+	b.gate[key] = bad
+	b.t.OnAsk(key, 1, bad)
+	b.deliver(t, warrantFor(gaBad.Ask, "approve"))
+	if len(b.conds) != 1 || b.conds[0].State != asks.StateRaised {
+		t.Fatalf("conditions = %+v", b.conds)
+	}
+	// The gate writes the ask again, right this time, at a new revision of its key: asked again under the same id.
+	_, good, gaGood := gateAsk(t, twoHeads(), twoHeads())
+	b.gate[key] = good
+	b.t.OnAsk(key, 2, good)
+	if n := len(b.askedOf(asks.AskSubject(delivery.TellerModule))); n != 2 {
+		t.Fatalf("the ask was published %d times, want again at the new revision", n)
+	}
+	b.deliver(t, warrantFor(gaGood.Ask, "approve"))
+	if _, ok := b.answers.written[delivery.AnswerKey(key)]; !ok {
+		t.Fatalf("no answer: %v\n%s", b.answers.written, b.logs.String())
+	}
+	if len(b.conds) != 2 || b.conds[1].State != asks.StateCleared || b.conds[1].Key != b.conds[0].Key {
+		t.Errorf("conditions = %+v", b.conds)
+	}
+}
+
+func TestAWarrantComingWhileTheTellerWasAwayIsReadFromTheRouter(t *testing.T) {
+	b := newBed(t)
+	key, value, ga := gateAsk(t, twoHeads(), twoHeads())
+	b.gate[key] = value
+	b.t.OnAsk(key, 1, value)
+	w, _ := json.Marshal(warrantFor(ga.Ask, "approve"))
+	b.records[ga.Ask.ID] = `{"state":"chosen","warrant":` + string(w) + `}`
+	b.t.OnAsk(key, 1, value) // the teller starts again and reads the key at the same revision
+	if _, ok := b.answers.written[delivery.AnswerKey(key)]; !ok {
+		t.Errorf("no answer: %s", b.logs.String())
+	}
+}
+
+func TestAnAskWhoseWordsChangedIsCancelledAndAskedAgain(t *testing.T) {
+	b := newBed(t)
+	key, value, ga := gateAsk(t, twoHeads(), twoHeads())
+	b.t.OnAsk(key, 1, value)
+	changed := ga
+	changed.Ask.Explanation = "Merges two pull requests, now with a third point.\nReview: approved."
+	v2, err := delivery.EncodeAsk(changed)
+	if err != nil {
+		t.Fatal(err)
+	}
+	b.t.OnAsk(key, 2, v2)
+	if n := len(b.askedOf(asks.CancelSubject(delivery.TellerModule))); n != 1 {
+		t.Errorf("cancels = %d", n)
+	}
+	if n := len(b.askedOf(asks.AskSubject(delivery.TellerModule))); n != 2 {
+		t.Errorf("asks = %d", n)
+	}
+}
+
+func TestAKeyWhoseValueIsNotItsOwnAskIsLeft(t *testing.T) {
+	b := newBed(t)
+	_, value, _ := gateAsk(t, twoHeads(), twoHeads())
+	b.t.OnAsk("merge.g.000000000000", 1, value)
+	b.t.OnAsk("merge.g.000000000000", 2, []byte(`{"v":1`))
+	if len(b.pubs) != 0 {
+		t.Errorf("published %+v", b.pubs)
+	}
+}
+
+func TestARefusedAskIsSaidAsACondition(t *testing.T) { // ADR 0437 §9
+	b := newBed(t)
+	key, value, ga := gateAsk(t, twoHeads(), twoHeads())
+	b.t.OnAsk(key, 1, value)
+	w := asks.Warrant{Ask: ga.Ask.ID, Asker: delivery.TellerModule, Outcome: asks.OutcomeRefused, Words: "no channel can send", At: now}
+	b.deliver(t, w)
+	if len(b.conds) != 1 || b.conds[0].State != asks.StateRaised || !strings.Contains(b.conds[0].Summary, "no channel can send") {
+		t.Errorf("conditions = %+v", b.conds)
+	}
+}

modules/mesh-delivery-teller/go.mod

@@ -3,6 +3,7 @@ module mesh-delivery-teller
 go 1.26.0
 
 require (
+	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019230201-5786ba190c69
 	github.com/nats-io/nats-server/v2 v2.11.17
 	github.com/nats-io/nats.go v1.54.0
 )

modules/mesh-delivery-teller/go.sum

@@ -1,3 +1,5 @@
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019230201-5786ba190c69 h1:kvtV/SPsdRVUL/BZ3xxGS6U8YrXy/t7TRBjVrD6VLw8=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019230201-5786ba190c69/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
 github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
 github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
 github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=

modules/mesh-delivery-teller/manifest_test.go

@@ -76,7 +76,9 @@ func TestManifestDeclarations(t *testing.T) { // 1240/AC1
 	if m.Module != "mesh-delivery-teller" {
 		t.Errorf("module = %q", m.Module)
 	}
-	if want := []state{{Name: "answers", History: 5}}; !reflect.DeepEqual(m.State, want) {
+	// answers is the teller's answer to each ask; asked is what it asked of the router, which condition it raised and which
+	// gate key and revision each ask was made at (issue 1343).
+	if want := []state{{Name: "answers", History: 5}, {Name: "asked", History: 1}}; !reflect.DeepEqual(m.State, want) {
 		t.Errorf("state = %+v, want %+v", m.State, want)
 	}
 	if want := []string{"mesh-delivery-gate.asks"}; !reflect.DeepEqual(sorted(m.Reads), want) {
@@ -134,7 +136,10 @@ func TestManifestProcess(t *testing.T) { // 1240/AC1
 }
 
 func TestServesNoRequestSubjectAndHoldsNoTicker(t *testing.T) { // 1240/AC3 (ADR 0437 §8)
-	forbidden := []string{"Subscribe(", "QueueSubscribe", "ChanSubscribe", "time.NewTicker", "time.Tick", "time.NewTimer", "time.AfterFunc", "time.After(", "time.Sleep"}
+	forbidden := []string{"Subscribe(", "QueueSubscribe", "ChanSubscribe", "Respond(", "RespondMsg(", "time.NewTicker", "time.Tick", "time.NewTimer", "time.AfterFunc", "time.After(", "time.Sleep"}
+	// The one subscription the teller makes is to its own `decided` event, the router's word on its asks (ADR 0259, ADR 0437
+	// §6): an event it hears, not a request it answers. It is allowed in decided.go and there only on that subject.
+	const decidedFile = "decided.go"
 	var files int
 	for _, root := range []string{"cmd", "internal"} {
 		err := filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error {
@@ -150,6 +155,13 @@ func TestServesNoRequestSubjectAndHoldsNoTicker(t *testing.T) { // 1240/AC3 (ADR
 				return err
 			}
 			for _, f := range forbidden {
+				if f == "Subscribe(" && filepath.Base(path) == decidedFile {
+					if n := strings.Count(string(src), "conn.Subscribe("); n != 1 || !strings.Contains(string(src), "conn.Subscribe(asks.DecidedSubject(module),") ||
+						strings.Count(string(src), f) != n {
+						t.Errorf("%s holds %d subscriptions, and not just one to asks.DecidedSubject(module)", path, strings.Count(string(src), f))
+					}
+					continue
+				}
 				if strings.Contains(string(src), f) {
 					t.Errorf("%s holds %s: the teller serves no request subject and holds no timer", path, f)
 				}

modules/mesh-delivery-teller/module.json

@@ -3,7 +3,8 @@
   "version": "1",
   "slug": "teller",
   "state": [
-    { "name": "answers", "history": 5 }
+    { "name": "answers", "history": 5 },
+    { "name": "asked", "history": 1 }
   ],
   "reads": [
     "mesh-delivery-gate.asks"