Fill an assigner ask's answer from the view's answer under its view key, once and never for an ask that needs proof (issue 1297 B10)

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

@@ -165,6 +165,8 @@ const (
 	tellerAnswers = "mesh-delivery-teller_answers"
 	tellerAsked   = "mesh-delivery-teller_asked"
 	gateAsksKV    = "mesh-delivery-gate_asks"
+	assignerAsks  = "mesh-delivery-assigner_asks"
+	viewAnswers   = "mesh-delivery-view_answers"
 	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"
@@ -174,7 +176,9 @@ const (
 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)...)
+	teller = append(append(teller, readOf(assignerAsks)...), readOf(viewAnswers)...) // issue 1297 B10: the manifest's reads
+	view := append(append(append(readOf(tellerAnswers), readOf(gateAsksKV)...), readOf(assignerAsks)...), keepsOf(viewAnswers)...)
+	assigner := keepsOf(assignerAsks)
 	tools := append(keepsOf(gateAsksKV), readOf(tellerAnswers)...)
 	return fmt.Sprintf(`
 listen: 127.0.0.1:-1
@@ -184,12 +188,13 @@ accounts {
     jetstream: enabled
     users = [
       { user: "ctl", password: "pw" }
-%s%s%s    ]
+%s%s%s%s    ]
   }
 }
 `, filepath.Join(dir, "js"),
 		user("n1.mesh-delivery-teller", teller, []string{seatDecided}),
 		user("n1.mesh-delivery-view", view, nil),
+		user("n1.mesh-delivery-assigner", assigner, nil),
 		user("n1.node-tools", tools, nil))
 }
 
@@ -238,7 +243,7 @@ func startFull(t *testing.T) *fullBed {
 	}
 	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} {
+	for name, history := range map[string]uint8{gateAsksKV: delivery.AsksHistory, tellerAnswers: delivery.AnswersHistory, tellerAsked: 1, assignerAsks: delivery.AssignerAsksHistory, viewAnswers: delivery.ViewAnswersHistory} {
 		if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: history}); err != nil {
 			t.Fatal(err)
 		}
@@ -468,3 +473,75 @@ func TestTellerAsksAgainAnAskRefusedForWantOfRoom(t *testing.T) { // 1610/AC1
 	say(asks.Warrant{Ask: ga2.Ask.ID, Asker: delivery.TellerModule, Outcome: asks.OutcomeExpired, At: now, AskDigest: ga2.Ask.Digest()})
 	lastAsk(held.Ask.ID, second)
 }
+
+// TestTellerFillsAnAskFromTheViewsAnswerOnce starts the program against the bus with the permissions above: the assigner's
+// fixture hold ask is written, then the view's answer under `<ask key>.view`; the answer is under the ask's own key, once
+// (rewriting the view's answer does not write a second), and a view answer under the merge's key fills nothing
+// (1297/B10 AC: a view answer fills the ask once; an ask needing proof is not answered from a view key).
+func TestTellerFillsAnAskFromTheViewsAnswerOnce(t *testing.T) { // 1297/B10 AC
+	now = time.Now().UTC()
+	b := startFull(t)
+	assigner, view := b.as(t, "n1.mesh-delivery-assigner"), b.as(t, "n1.mesh-delivery-view")
+	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, ask := holdAsk(t)
+	askKV, err := assigner.KeyValue(b.ctx, assignerAsks)
+	if err != nil {
+		t.Fatal(err)
+	}
+	viewKV, err := view.KeyValue(b.ctx, viewAnswers)
+	if err != nil {
+		t.Fatal(err)
+	}
+	answers, err := b.js.KeyValue(b.ctx, tellerAnswers)
+	if err != nil {
+		t.Fatal(err)
+	}
+	time.Sleep(500 * time.Millisecond) // the program is watching
+	if _, err := askKV.Put(b.ctx, key, value); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := viewKV.Put(b.ctx, key+".view", viewAnswer(t, "release", "the check was a flake")); 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 != "release" || answeredAsking(e.Value()) != ask.ID {
+		t.Fatalf("answer %+v err %v", a, err)
+	}
+	// The view's answer written again changes nothing under the ask.
+	if _, err := viewKV.Put(b.ctx, key+".view", viewAnswer(t, "keep", "")); err != nil {
+		t.Fatal(err)
+	}
+	time.Sleep(time.Second)
+	if again, err := answers.Get(b.ctx, delivery.AnswerKey(key)); err != nil || again.Revision() != e.Revision() {
+		t.Errorf("the answer was written a second time: %v", err)
+	}
+	// An answer under the merge's key fills nothing.
+	mkey, mvalue, _ := gateAsk(t, twoHeads(), twoHeads())
+	if _, err := askKV.Put(b.ctx, mkey, mvalue); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := viewKV.Put(b.ctx, mkey+".view", viewAnswer(t, "approve", "")); err != nil {
+		t.Fatal(err)
+	}
+	time.Sleep(time.Second)
+	if _, err := answers.Get(b.ctx, delivery.AnswerKey(mkey)); err == nil {
+		t.Error("the merge's ask was answered from a view key")
+	}
+}

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

@@ -137,10 +137,43 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	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, t.OnCurrent); err != nil {
-		return fmt.Errorf("watching the gate's asks: %w", err)
+	// The view's answers to the assigner's asks (issue 1297 B10): both buckets are the controller's to create, and a bucket not
+	// there is said and leaves the teller without that path rather than without the gate's asks (a wait for it needs a timer).
+	failed := make(chan error, 2)
+	assignerKV, viewKV := optionalBucket(ctx, js, delivery.AssignerAsksBucket, logger), optionalBucket(ctx, js, delivery.ViewAnswersBucket, logger)
+	if assignerKV != nil && viewKV != nil {
+		t.Assigner, t.View = kvGate{ctx: ctx, kv: assignerKV}, kvGate{ctx: ctx, kv: viewKV}
+		go func() {
+			failed <- watchAsks(ctx, viewKV, func(key string, _ uint64, _ []byte) { t.OnViewAnswer(key) }, func(string) {}, nil)
+		}()
+		go func() {
+			failed <- watchAsks(ctx, assignerKV, func(key string, _ uint64, _ []byte) { t.OnAssignerAsk(key) }, func(string) {}, nil)
+		}()
 	}
-	return nil
+	go func() {
+		if err := watchAsks(ctx, gate, t.OnAsk, t.OnAskGone, t.OnCurrent); err != nil {
+			failed <- fmt.Errorf("watching the gate's asks: %w", err)
+		}
+	}()
+	select {
+	case <-ctx.Done():
+		return nil
+	case err := <-failed:
+		if err != nil {
+			return fmt.Errorf("a watch ended: %w", err)
+		}
+		return nil
+	}
+}
+
+// optionalBucket is a bucket the teller reads when it exists, and nil, said once, when it does not.
+func optionalBucket(ctx context.Context, js jetstream.JetStream, name string, logger *log.Logger) jetstream.KeyValue {
+	kv, err := js.KeyValue(ctx, name)
+	if err != nil {
+		logger.Printf("the bucket %s cannot be read (%v): the view's answers fill no ask until the teller is started again with it", name, err)
+		return nil
+	}
+	return kv
 }
 
 // askedBucket is the teller's state bucket asked, `<module>_<name>`: what it asked of the router and which condition it

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

@@ -58,6 +58,10 @@ type Teller struct {
 	Store   store
 	Answers answers
 	Gate    gateAsks
+	// Assigner is the assigner's bucket asks and View the view's bucket answers (issue 1297 B10); with either nil the teller
+	// answers nothing from the view.
+	Assigner gateAsks
+	View     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)

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

@@ -0,0 +1,115 @@
+package main
+
+import (
+	"encoding/json"
+	"slices"
+	"strings"
+
+	"git.novox.be/novox/mesh-sdk/go/asks"
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+// The view's answers (novox/hq ADR 0528 clause 6, ADR 0437 §10, issue 1297 B10). The operator may answer, in the view, the
+// asks of the assigner that need no proof of who answered (the hold of a fingerprint and the restore of a data module set back,
+// mesh-sdk go/delivery AskNeedsProof); the view writes its answer under `<ask key>.view` in its bucket answers, and the teller,
+// which alone fills an ask's answer, takes it from there as it takes a warrant from the phone: under the ask's own key, once,
+// by compare-and-set. The merge's ask needs proof (ADR 0421 §1, issue 953) and is never answered from a view key.
+
+// An ask of the assigner as the teller reads it: the router's ask under "ask". Its other fields are the assigner's.
+type assignerAsk struct {
+	Ask asks.Ask `json:"ask"`
+}
+
+// OnViewAnswer handles a key put in the view's bucket answers. A key that is no view answer key, or is the answer to an ask that
+// needs proof of who answered, is left and said.
+func (t *Teller) OnViewAnswer(key string) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	askKey, err := delivery.ParseViewAnswerKey(key)
+	if err != nil {
+		t.logf("the view's key %s is left: %v", key, err)
+		return
+	}
+	t.fillFromView(askKey)
+}
+
+// OnAssignerAsk handles a key put in the assigner's bucket asks: a view answer that came before its ask fills it now.
+func (t *Teller) OnAssignerAsk(key string) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	if _, err := delivery.ViewAnswerKey(key); err != nil {
+		t.logf("the assigner's key %s is left and never answered from the view: %v", key, err)
+		return
+	}
+	t.fillFromView(key)
+}
+
+// fillFromView writes the answer to the assigner's ask at askKey from the view's answer, when both are there and readable, the
+// ask is of a kind that needs no proof, the option is one the ask offers and the asking is not answered yet.
+func (t *Teller) fillFromView(askKey string) {
+	if t.Assigner == nil || t.View == nil {
+		return
+	}
+	viewKey, err := delivery.ViewAnswerKey(askKey) // refuses an ask that needs proof, naming the phone
+	if err != nil {
+		t.logf("%s is not answered from the view: %v", askKey, err)
+		return
+	}
+	rawAnswer, found, err := t.View.Get(viewKey)
+	if err != nil {
+		t.logf("%s: the view's answer could not be read, so the ask is left: %v", askKey, err)
+		return
+	}
+	if !found {
+		return
+	}
+	rawAsk, found, err := t.Assigner.Get(askKey)
+	if err != nil {
+		t.logf("%s: the assigner's ask could not be read, so the view's answer is left: %v", askKey, err)
+		return
+	}
+	if !found {
+		return // the answer waits for its ask: OnAssignerAsk fills it when it comes
+	}
+	va, err := delivery.DecodeViewAnswer(rawAnswer)
+	if err != nil {
+		t.logf("%s: the view's answer is not readable and fills nothing: %v", viewKey, err)
+		return
+	}
+	var ask assignerAsk
+	if err := json.Unmarshal(rawAsk, &ask); err != nil || ask.Ask.ID == "" {
+		t.logf("%s: the assigner's ask is not readable and takes no answer (%v)", askKey, err)
+		return
+	}
+	if !slices.ContainsFunc(ask.Ask.Options, func(o asks.Option) bool { return o.ID == va.Option }) {
+		t.logf("%s: the view answered %q and the ask offers no option %q: nothing is written", askKey, va.Option, va.Option)
+		return
+	}
+	akey := delivery.AnswerKey(askKey)
+	cur, crev, has, err := t.Answers.Get(akey)
+	if err != nil {
+		t.logf("%s: the answer could not be read: %v", askKey, err)
+		return
+	}
+	if has && answeredAsking(cur) == ask.Ask.ID {
+		return // this asking is answered, from the view or elsewhere
+	}
+	words := "answered in the view (mesh-delivery-view): " + va.Option
+	if w := strings.TrimSpace(va.Words); w != "" {
+		words += ": " + w
+	}
+	body, err := delivery.EncodeAnswer(delivery.Answer{Option: va.Option, Warrant: ask.Ask.ID + "@" + t.now().UTC().Format("2006-01-02T15:04:05.999999999Z07:00"), Words: words})
+	if err != nil {
+		t.logf("%s: the answer could not be made: %v", askKey, err)
+		return
+	}
+	written, err := t.Answers.Write(akey, body, crev, has)
+	switch {
+	case err != nil:
+		t.logf("%s: the answer could not be written: %v", askKey, err)
+	case !written:
+		t.logf("%s: another writer answered first, and the view's answer is not written", askKey)
+	default:
+		t.logf("%s: %s", askKey, words)
+	}
+}

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

@@ -0,0 +1,155 @@
+package main
+
+import (
+	"encoding/json"
+	"strings"
+	"testing"
+
+	"git.novox.be/novox/mesh-sdk/go/asks"
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+const fpHeld = "src2:2222222222222222222222222222222222222222222222222222222222222222"
+
+// holdAsk is a key of the assigner's asks: the ask about a held fingerprint, as a fixture (in release 5 the assigner writes
+// none yet; the value is `{"v":1,"ask":<the router's ask>}`).
+func holdAsk(t *testing.T) (key string, value []byte, ask asks.Ask) {
+	t.Helper()
+	key, err := delivery.HoldAskKey("plex", fpHeld)
+	if err != nil {
+		t.Fatal(err)
+	}
+	ask = asks.Ask{ID: delivery.AskID(key), Headline: "plex is held after a failed health check", Explanation: "ace failed the check.", Who: asks.Operator,
+		Expires: now.Add(asks.ApproveLasts), OnExpiry: "the fingerprint stays held",
+		Options: []asks.Option{{ID: "release", Label: "Release", Does: "lets the move go on", Level: asks.Approve}, {ID: "keep", Label: "Keep held", Does: "keeps it held", Level: asks.Approve}}}
+	value, err = json.Marshal(map[string]any{"v": 1, "ask": ask})
+	if err != nil {
+		t.Fatal(err)
+	}
+	return key, value, ask
+}
+
+func viewAnswer(t *testing.T, option, words string) []byte {
+	t.Helper()
+	v, err := delivery.EncodeViewAnswer(delivery.ViewAnswer{Option: option, Words: words})
+	if err != nil {
+		t.Fatal(err)
+	}
+	return v
+}
+
+// withView gives the bed's teller the assigner's asks and the view's answers, both empty.
+func withView(b *bed) (assigner, view memGate) {
+	assigner, view = memGate{}, memGate{}
+	b.t.Assigner, b.t.View = assigner, view
+	return assigner, view
+}
+
+func TestAViewAnswerFillsTheAskOnce(t *testing.T) { // 1297/B10 AC
+	b := newBed(t)
+	assigner, view := withView(b)
+	key, value, ask := holdAsk(t)
+	assigner[key] = value
+	view[key+".view"] = viewAnswer(t, "release", "the check was a flake")
+	b.t.OnViewAnswer(key + ".view")
+	raw, ok := b.answers.written[key]
+	if !ok {
+		t.Fatalf("no answer under %s; log:\n%s", key, b.logs.String())
+	}
+	a, err := delivery.DecodeAnswer(raw)
+	if err != nil || a.Option != "release" || answeredAsking(raw) != ask.ID || !strings.Contains(a.Words, "view") || !strings.Contains(a.Words, "the check was a flake") {
+		t.Errorf("answer = %+v, %v", a, err)
+	}
+	// The same answer heard again (the watch replays, the ask is rewritten) writes nothing more.
+	b.t.OnViewAnswer(key + ".view")
+	b.t.OnAssignerAsk(key)
+	if b.answers.writes != 1 {
+		t.Errorf("%d writes, want one", b.answers.writes)
+	}
+}
+
+func TestAViewAnswerHeardBeforeItsAskFillsItWhenTheAskComes(t *testing.T) { // 1297/B10
+	b := newBed(t)
+	assigner, view := withView(b)
+	key, value, _ := holdAsk(t)
+	view[key+".view"] = viewAnswer(t, "keep", "")
+	b.t.OnViewAnswer(key + ".view")
+	if b.answers.writes != 0 {
+		t.Fatalf("answered an ask the teller does not hold")
+	}
+	assigner[key] = value
+	b.t.OnAssignerAsk(key)
+	if a, err := delivery.DecodeAnswer(b.answers.written[key]); err != nil || a.Option != "keep" {
+		t.Errorf("answer = %+v, %v", a, err)
+	}
+}
+
+func TestAnAskNeedingProofIsNotAnsweredFromAViewKey(t *testing.T) { // 1297/B10 AC
+	b := newBed(t)
+	assigner, view := withView(b)
+	key, value, _ := gateAsk(t, twoHeads(), twoHeads())
+	// Even with the merge's ask under the assigner's bucket and a view answer under its key, nothing is written.
+	assigner[key] = value
+	view[key+".view"] = viewAnswer(t, "approve", "")
+	b.t.OnViewAnswer(key + ".view")
+	b.t.OnAssignerAsk(key)
+	if b.answers.writes != 0 || len(b.answers.written) != 0 {
+		t.Fatalf("the merge's ask was answered from the view: %v", b.answers.written)
+	}
+	if !strings.Contains(b.logs.String(), "issue 953") {
+		t.Errorf("the refusal does not say why:\n%s", b.logs.String())
+	}
+}
+
+func TestAViewAnswerOfAnOptionTheAskDoesNotOfferFillsNothing(t *testing.T) { // 1297/B10
+	b := newBed(t)
+	assigner, view := withView(b)
+	key, value, _ := holdAsk(t)
+	assigner[key] = value
+	view[key+".view"] = viewAnswer(t, "delete", "")
+	b.t.OnViewAnswer(key + ".view")
+	if b.answers.writes != 0 || !strings.Contains(b.logs.String(), `offers no option "delete"`) {
+		t.Errorf("writes %d, log:\n%s", b.answers.writes, b.logs.String())
+	}
+}
+
+func TestAViewAnswerNeverReplacesTheAnswerOfTheSameAsking(t *testing.T) { // 1297/B10
+	b := newBed(t)
+	assigner, view := withView(b)
+	key, value, ask := holdAsk(t)
+	assigner[key] = value
+	view[key+".view"] = viewAnswer(t, "keep", "")
+	early, err := delivery.EncodeAnswer(delivery.Answer{Option: "release", Warrant: ask.ID + "@2026-10-20T09:00:00Z", Words: "answered elsewhere"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	b.answers.written[key], b.answers.revs[key] = early, 7
+	b.t.OnViewAnswer(key + ".view")
+	if b.answers.writes != 0 || string(b.answers.written[key]) != string(early) {
+		t.Errorf("the answer of the same asking was replaced")
+	}
+	// An earlier asking's answer (another router id) does not stop this one.
+	older, _ := delivery.EncodeAnswer(delivery.Answer{Option: "release", Warrant: "an-older-id@2026-10-19T09:00:00Z", Words: "old"})
+	b.answers.written[key], b.answers.revs[key] = older, 8
+	b.t.OnViewAnswer(key + ".view")
+	if b.answers.writes != 1 || answeredAsking(b.answers.written[key]) != ask.ID {
+		t.Errorf("writes %d, answer %s", b.answers.writes, b.answers.written[key])
+	}
+}
+
+func TestAnUnreadableViewAnswerOrAskFillsNothing(t *testing.T) { // 1297/B10
+	b := newBed(t)
+	assigner, view := withView(b)
+	key, value, _ := holdAsk(t)
+	assigner[key] = []byte(`{"v":1}`)
+	view[key+".view"] = viewAnswer(t, "release", "")
+	b.t.OnViewAnswer(key + ".view")
+	assigner[key] = value
+	view[key+".view"] = []byte(`{"v":99}`)
+	b.t.OnViewAnswer(key + ".view")
+	b.t.OnViewAnswer("not-a-view-key")
+	b.t.OnViewAnswer("hold.x.view")
+	if b.answers.writes != 0 {
+		t.Errorf("filled from something unreadable; log:\n%s", b.logs.String())
+	}
+}

modules/mesh-delivery-teller/manifest_test.go

@@ -81,7 +81,8 @@ func TestManifestDeclarations(t *testing.T) { // 1240/AC1
 	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) {
+	// The assigner's asks and the view's answers to them are read so the view's answer fills an ask that needs no proof (1297/B10).
+	if want := []string{"mesh-delivery-assigner.asks", "mesh-delivery-gate.asks", "mesh-delivery-view.answers"}; !reflect.DeepEqual(sorted(m.Reads), want) {
 		t.Errorf("reads = %v, want %v", m.Reads, want)
 	}
 	if want := []string{"operator-channel"}; !reflect.DeepEqual(sorted(m.Uses), want) {

modules/mesh-delivery-teller/module.json

@@ -7,7 +7,9 @@
     { "name": "asked", "history": 1 }
   ],
   "reads": [
-    "mesh-delivery-gate.asks"
+    "mesh-delivery-assigner.asks",
+    "mesh-delivery-gate.asks",
+    "mesh-delivery-view.answers"
   ],
   "uses": [
     "operator-channel"