TL: the teller says answered and reads its answers through teller_answers; the view says reached with each reach it writes (abdf11be)

Builds: design 47 G §431

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

@@ -8,6 +8,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"os"
+	"sort"
 	"strings"
 	"sync"
 	"time"
@@ -51,6 +52,55 @@ func Tools(ask asker, now func() time.Time) []stdio.Tool {
 	}
 }
 
+// AnswersTool is the tool that reads the teller's state answers (ADR 0596 §3: the owner's state is read through its owner's
+// tool): one answer by the key of the ask it answers, or every answer when none is named. It reads and writes nothing else.
+const AnswersTool = "teller_answers"
+
+// answers is the teller's state answers as the tool reads it.
+type answers interface {
+	Get(key string) (*stdio.StateEntry, error)
+	Keys() ([]string, error)
+}
+
+// AnswerTools is the tool teller_answers over the state answers.
+func AnswerTools(s answers) []stdio.Tool {
+	return []stdio.Tool{{Name: AnswersTool,
+		Description: "Read the answers the delivery teller kept to the gate's asks, as it wrote them: the answer under one ask's " +
+			"key (found false when the ask has none), or every answer when no key is named. An answer is the operator's chosen " +
+			"option, the warrant it came by and the words of it. What the gate reads to merge on the Approve.",
+		Input: map[string]any{"type": "object", "properties": map[string]any{
+			"key": map[string]any{"type": "string", "description": "the key of the gate's ask the answer is under; none for every answer"}}},
+		Run: func(args map[string]any) (any, error) {
+			key, _ := args["key"].(string)
+			if key = strings.TrimSpace(key); key != "" {
+				e, err := s.Get(key)
+				if err != nil {
+					return nil, fmt.Errorf("the answer under %s cannot be read: %w", key, err)
+				}
+				if e == nil {
+					return map[string]any{"key": key, "found": false}, nil
+				}
+				return map[string]any{"key": key, "found": true, "answer": json.RawMessage(e.Value)}, nil
+			}
+			keys, err := s.Keys()
+			if err != nil {
+				return nil, fmt.Errorf("the answers cannot be listed: %w", err)
+			}
+			sort.Strings(keys)
+			out := make([]map[string]any, 0, len(keys))
+			for _, k := range keys {
+				e, err := s.Get(k)
+				if err != nil {
+					return nil, fmt.Errorf("the answer under %s cannot be read: %w", k, err)
+				}
+				if e != nil {
+					out = append(out, map[string]any{"key": k, "answer": json.RawMessage(e.Value)})
+				}
+			}
+			return map[string]any{"answers": out}, nil
+		}}}
+}
+
 // askEnd writes the ends value at key and waits, on a watch of the key, at most bound for the teller's answer.
 func askEnd(s stdio.KeptState, bound time.Duration) asker {
 	return func(key string, value []byte) (ends.End, error) {
@@ -108,6 +158,7 @@ func main() {
 	// asking it again by hand (ADR 0615, design 25 § 12.2).
 	tools := append(Tools(askEnd(stdio.State(ends.State), 25*time.Second), time.Now),
 		views.StatusTools("mesh-delivery-teller", "teller_first_views", "teller_first_view_again", stdio.State(views.StatusState), meshhandactsclient.Recorder{Ask: stdio.Ask}, time.Now)...)
+	tools = append(tools, AnswerTools(stdio.State("answers"))...)
 	if err := stdio.Serve("", tools); err != nil {
 		fmt.Fprintf(os.Stderr, "[mesh-delivery-teller-tools] %v\n", err)
 		os.Exit(1)

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

@@ -0,0 +1,49 @@
+package main
+
+import (
+	"encoding/json"
+	"fmt"
+	"testing"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+)
+
+type fakeAnswers map[string]string
+
+func (f fakeAnswers) Get(key string) (*stdio.StateEntry, error) {
+	v, ok := f[key]
+	if !ok {
+		return nil, nil
+	}
+	return &stdio.StateEntry{Key: key, Value: json.RawMessage(v)}, nil
+}
+
+func (f fakeAnswers) Keys() ([]string, error) {
+	var out []string
+	for k := range f {
+		out = append(out, k)
+	}
+	return out, nil
+}
+
+// teller_answers reads the teller's answers: one by the ask's key, an ask with none said as not found, every answer when no key
+// is named (ADR 0596 §3: the gate asks this tool, never the bucket).
+func TestTellerAnswersReadsOneAnswerOrEveryAnswer(t *testing.T) {
+	tool := AnswerTools(fakeAnswers{"k1": `{"option":"approve"}`, "k2": `{"option":"decline"}`})[0]
+	if tool.Name != "teller_answers" {
+		t.Fatalf("the tool is %s", tool.Name)
+	}
+	out, err := tool.Run(map[string]any{"key": "k1"})
+	if err != nil || fmt.Sprint(out) != `map[answer:{"option":"approve"} found:true key:k1]` {
+		t.Errorf("one answer: %v, %v", out, err)
+	}
+	out, err = tool.Run(map[string]any{"key": "gone"})
+	if err != nil || fmt.Sprint(out) != `map[found:false key:gone]` {
+		t.Errorf("no answer: %v, %v", out, err)
+	}
+	out, err = tool.Run(map[string]any{})
+	raw, _ := json.Marshal(out)
+	if err != nil || string(raw) != `{"answers":[{"answer":{"option":"approve"},"key":"k1"},{"answer":{"option":"decline"},"key":"k2"}]}` {
+		t.Errorf("every answer: %s, %v", raw, err)
+	}
+}

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

@@ -187,7 +187,8 @@ func fullConf(dir string) string {
 		seatAsk, seatCancel, seatCondition, "$JS.API.DIRECT.GET.KV_"+messengerAsks+".$KV."+messengerAsks+".mesh-delivery-teller.>")
 	// The durable consumer on EVENTS the controller makes for the seat it uses, pulled and acknowledged (issue 1126).
 	teller = append(teller, "$JS.API.CONSUMER.INFO.EVENTS."+tellerConsumer, "$JS.API.CONSUMER.MSG.NEXT.EVENTS."+tellerConsumer, "$JS.ACK.EVENTS."+tellerConsumer+".>")
-	teller = append(teller, readOf(viewAnswers)...) // issue 1297 B10: the manifest's reads
+	teller = append(teller, "mesh.mod.mesh-delivery-teller.event.answered") // the manifest's emits (ADR 0596 §2)
+	teller = append(teller, readOf(viewAnswers)...)                         // issue 1297 B10: the manifest's reads
 	// The assigner's asks by its read tool and its views family (ADR 0615, DLV-M3), and the teller's own first-views and
 	// findings; the bucket itself is no longer the teller's to read.
 	teller = append(append(append(teller, viewstest.Grants(assignerModule, assignerModule+".assigner_read_asks", tellerConsumer)...), keepsOf(firstViewsBucket)...), keepsOf(findingsBucket)...)
@@ -434,6 +435,23 @@ func TestTellerCarriesAnAskAndWritesTheAnswer(t *testing.T) { // 1343/AC1, AC2
 	if a, err := meshdeliveryclient.DecodeAnswer(e.Value()); err != nil || a.Option != "approve" {
 		t.Errorf("answer %+v err %v", a, err)
 	}
+	// The answer is said as the event answered, which the gate hears (ADR 0596 §2).
+	events, err := b.js.Stream(b.ctx, process.EventsStream)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var said *jetstream.RawStreamMsg
+	for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
+		if said, err = events.GetLastMsgForSubject(b.ctx, "mesh.mod.mesh-delivery-teller.event.answered"); err == nil {
+			break
+		}
+	}
+	if said == nil {
+		t.Fatalf("the answer was not said as the event answered: %v\n%s", err, logs.String())
+	}
+	if got := string(said.Data); !strings.Contains(got, `"key":"`+key+`"`) || !strings.Contains(got, `"option":"approve"`) {
+		t.Errorf("the event answered is %s", got)
+	}
 	// A second ask, withdrawn by the gate, is cancelled at the router.
 	other := []meshdeliveryclient.Head{{Repository: "novox/mesh-delivery", Number: 13, Head: headB}}
 	key2, value2, ga2 := gateAsk(t, other, other)

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

@@ -19,6 +19,7 @@ import (
 	"os/signal"
 	"syscall"
 
+	stdio "git.novox.be/novox/mesh-sdk/go"
 	"git.novox.be/novox/mesh-sdk/go/process"
 	"git.novox.be/novox/mesh-sdk/go/views"
 
@@ -99,6 +100,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		return fmt.Errorf("the teller's connection for its tool calls: %w", err)
 	}
 	defer calls.Close()
+	t.Answered = func(key, askID string, answer []byte) error {
+		return calls.Emit(answeredEvent, Answered{Key: key, Ask: askID, Answer: answer}, stdio.EmitOptions{MessageID: answeredID(key, askID)})
+	}
 	reader := &views.ProcessReader{Module: tellerModule, Call: calls.Call, Logf: logger.Printf, Hub: hub}
 	findings := process.AwaitBucket(ctx, js, conn, findingsBucket, "a first view unanswered past its bound is said in the log alone", logger.Printf)
 	reader.Findings = func() views.FindingWriter { return sdkfindings.Writer(ctx, findings.KV) }

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

@@ -1,5 +1,7 @@
 package main
 
+import "encoding/json"
+
 // The holders and buckets the teller names, taken from its own manifest (module.json: its name, its `reads`) and the
 // states it keeps itself; a unit row checks them against it (TestTheNamesAreTheManifests). No contract names a holder
 // (novox/hq ADR 0625 §4, ADR 0634 §1).
@@ -27,3 +29,16 @@ const (
 
 // answerKey is the key, in the teller's answers, of its answer to the ask at askKey: the same string.
 func answerKey(askKey string) string { return askKey }
+
+// answeredEvent is the event the teller says after each answer it writes (manifest emits): the key of the ask and the answer.
+const answeredEvent = "answered"
+
+// Answered is the body of the event answered.
+type Answered struct {
+	Key    string          `json:"key"`
+	Ask    string          `json:"ask"`
+	Answer json.RawMessage `json:"answer"`
+}
+
+// answeredID is the bus message id of the event for one ask's answer: said twice, it is kept once.
+func answeredID(key, askID string) string { return "answered." + key + "." + askID }

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

@@ -100,6 +100,9 @@ type Teller struct {
 	// Publish is the bus the asker publishes on, used to ask again an ask the router refused for want of room under a bus
 	// message id of its own (the bus drops a second message under an id it holds).
 	Publish func(subject string, body any, id string) error
+	// Answered says an answer written under an ask's key as the event answered (ADR 0596 §2: the owner says what it
+	// wrote; a reader hears it and asks the tool teller_answers, never the bucket). Nil says nothing.
+	Answered func(key, askID string, answer []byte) error
 	// Now is the clock; nil is time.Now.
 	Now func() time.Time
 	// Emit says one of the teller's own events (its manifest's emits, tellerevents) under a bus message id of its own; nil
@@ -427,7 +430,9 @@ func (t *Teller) writeAnswer(key, askID string, body []byte, words string) error
 	}
 	if found && answeredAsking(cur) == askID {
 		t.logf("%s: an answer is there already, and is not written again", key)
-		return nil
+		// The event is said again under the same message id, so an answer written whose event was not said is said by the
+		// next warrant, and a second time inside the bus's window is kept once.
+		return t.sayAnswered(key, askID, cur)
 	}
 	written, err := t.Answers.Write(akey, body, crev, found)
 	if err != nil {
@@ -438,6 +443,17 @@ func (t *Teller) writeAnswer(key, askID string, body []byte, words string) error
 		return nil
 	}
 	t.logf("%s: %s", key, words)
+	return t.sayAnswered(key, askID, body)
+}
+
+// sayAnswered says the event answered for the answer under key to the asking askID: the key and the answer itself.
+func (t *Teller) sayAnswered(key, askID string, answer []byte) error {
+	if t.Answered == nil {
+		return nil
+	}
+	if err := t.Answered(key, askID, answer); err != nil {
+		return fmt.Errorf("the answer under %s is written and its event could not be said: %w", key, err)
+	}
 	return nil
 }
 

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

@@ -425,6 +425,40 @@ func TestAWarrantThatHoldsWritesOneAnswerAndASecondWritesNone(t *testing.T) { //
 	}
 }
 
+// ADR 0596 §2: the teller says answered after each answer it writes, under one message id per ask, and says it again for an
+// answer written whose event was not said.
+func TestEveryAnswerWrittenIsSaidAsTheEventAnswered(t *testing.T) {
+	b := newBed(t)
+	var said []string
+	fail := true
+	b.t.Answered = func(key, askID string, answer []byte) error {
+		if fail {
+			return fmt.Errorf("the bus did not answer")
+		}
+		said = append(said, answeredID(key, askID)+" "+string(answer))
+		return nil
+	}
+	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)
+	if len(said) != 0 || b.answers.writes != 1 {
+		t.Fatalf("said %v with the bus down, written %d times", said, b.answers.writes)
+	}
+	fail = false
+	if err := b.t.writeAnswer(key, ga.Ask.ID, b.answers.written[answerKey(key)], "again"); err != nil { // the answer is there; its event was not said
+		t.Fatal(err)
+	}
+	if len(said) != 1 || b.answers.writes != 1 {
+		t.Fatalf("said %v, written %d times; want the event said once and the answer written once", said, b.answers.writes)
+	}
+	raw := b.answers.written[answerKey(key)]
+	if want := answeredID(key, ga.Ask.ID) + " " + string(raw); said[0] != want {
+		t.Errorf("said %q, want %q", said[0], want)
+	}
+}
+
 func TestAWarrantThatDoesNotHoldWritesNoAnswerAndRaisesTheCondition(t *testing.T) { // 1343/AC3
 	other := []meshdeliveryclient.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, oc.Warrant){

modules/mesh-delivery-teller/manifest_test.go

@@ -45,12 +45,12 @@ type manifest struct {
 	Version      string            `json:"version"`
 	Slug         string            `json:"slug"`
 	Tools        []string          `json:"tools"`
+	Emits        []string          `json:"emits"`
 	Uses         []string          `json:"uses"`
 	State        []state           `json:"state"`
 	Reads        []string          `json:"reads"`
 	Invokes      []string          `json:"invokes"`
 	Consumes     []string          `json:"consumes"`
-	Emits        []string          `json:"emits"`
 	Streams      []stream          `json:"streams"`
 	OwnSecrets   map[string]string `json:"own-secrets"`
 	SecretsOwner string            `json:"secrets-owner"`
@@ -124,11 +124,12 @@ func TestManifestDeclarations(t *testing.T) { // 1240/AC1
 		t.Errorf("secrets-owner = %q", m.SecretsOwner)
 	}
 	// 774: teller_end_ask, served by its tools bundle through the state ends; the process serves none.
-	if want := []string{"teller_end_ask", "teller_first_views", "teller_first_view_again"}; !reflect.DeepEqual(m.Tools, want) {
+	if want := []string{"teller_end_ask", "teller_answers", "teller_first_views", "teller_first_view_again"}; !reflect.DeepEqual(m.Tools, want) {
 		t.Errorf("tools = %v, want %v", m.Tools, want)
 	}
-	// Its one event: an assigner ask ended at the channel without an answer, heard by the assigner (issue 1966, tellerevents).
-	if want := []string{tellerevents.AskEnded}; !reflect.DeepEqual(m.Emits, want) {
+	// Its two events: each answer it writes (ADR 0596 §2), and an assigner ask ended at the channel without an answer, heard by
+	// the assigner (issue 1966, tellerevents).
+	if want := []string{"answered", tellerevents.AskEnded}; !reflect.DeepEqual(m.Emits, want) {
 		t.Errorf("emits = %v, want %v", m.Emits, want)
 	}
 	// design 47 G §383, §387: the stream <module>-findings, 7 days and 16 MiB; §368: version is MAJOR.MINOR.

modules/mesh-delivery-teller/module.json

@@ -4,6 +4,7 @@
   "slug": "teller",
   "tools": [
     "teller_end_ask",
+    "teller_answers",
     "teller_first_views",
     "teller_first_view_again"
   ],
@@ -24,6 +25,7 @@
     }
   ],
   "emits": [
+    "answered",
     "ask-ended"
   ],
   "state": [

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

@@ -827,7 +827,7 @@ func TestProcessWholeSaysTheSourceRevisionIsDelivered(t *testing.T) {
 		t.Fatal(err)
 	}
 	kvs := map[string]jetstream.KeyValue{}
-	for _, name := range []string{view.SourceBucket, view.AssignerWantsBucket, view.HaveBucket} {
+	for _, name := range []string{view.SourceBucket, view.AssignerWantsBucket, view.HaveBucket, viewstate.ReachBucket} {
 		kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: 5})
 		if err != nil {
 			t.Fatal(err)
@@ -869,7 +869,7 @@ func TestProcessWholeSaysTheSourceRevisionIsDelivered(t *testing.T) {
 		t.Fatal(err)
 	}
 	time.Sleep(time.Second)
-	if info, err := events.Info(ctx, jetstream.WithSubjectFilter("mesh.mod.mesh-delivery-view.event.>")); err != nil || len(info.State.Subjects) != 0 {
+	if info, err := events.Info(ctx, jetstream.WithSubjectFilter("mesh.mod.mesh-delivery-view.event."+DeliveredEventType)); err != nil || len(info.State.Subjects) != 0 {
 		t.Fatalf("an event was said before the revision was delivered: %v %+v", err, info.State)
 	}
 	if _, err := kvs[view.HaveBucket].Put(ctx, "ace.web", good); err != nil {
@@ -889,6 +889,21 @@ func TestProcessWholeSaysTheSourceRevisionIsDelivered(t *testing.T) {
 	if msg.Header.Get("x-source") != "mesh-delivery-view" || msg.Header.Get("x-event-id") == "" {
 		t.Errorf("the headers = %v", msg.Header)
 	}
+	// The reach it wrote is said as the event reached, with the reach as written (ADR 0596 §2).
+	var rm *jetstream.RawStreamMsg
+	for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline) && rm == nil; time.Sleep(100 * time.Millisecond) {
+		rm, _ = events.GetLastMsgForSubject(ctx, "mesh.mod.mesh-delivery-view.event."+ReachedEventType)
+	}
+	if rm == nil {
+		t.Fatal("the reach of novox/web was written and never said as the event reached")
+	}
+	var re ReachedEvent
+	if err := json.Unmarshal(rm.Data, &re); err != nil || re.Key != "novox/web" {
+		t.Fatalf("the event reached = %s, %v", rm.Data, err)
+	}
+	if stored, gerr := kvs[viewstate.ReachBucket].Get(ctx, re.Key); gerr != nil || string(stored.Value()) != string(re.Reach) {
+		t.Errorf("the event carries %s, the state reach holds %v (%v)", re.Reach, stored, gerr)
+	}
 }
 
 // 1906/D10, whole on the bus (through run): a want that differs from its have is said `moving`, the have reaching it running and

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

@@ -102,9 +102,15 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		return err
 	})
 	model.SetReachWriter(func(w view.ReachWrite) error {
-		err := putReach(ctx, js, w)
+		rev, err := putReach(ctx, js, w)
 		if err != nil {
 			logger.Printf("the reach of %s could not be written (written again at the next change the view takes in): %v", w.Key, err)
+			return err
+		}
+		// What the view wrote it says (ADR 0596 §2): a reader hears the reach change and asks view_reach, never the bucket.
+		err = emitEvent(ctx, js, cred, ReachedEventType, ReachedEvent{Key: w.Key, Reach: w.Value}, fmt.Sprintf("reached.%s.%d", w.Key, rev))
+		if err != nil {
+			logger.Printf("the reach of %s is written and could not be said (said again at the next change the view takes in): %v", w.Key, err)
 		}
 		return err
 	})
@@ -172,13 +178,22 @@ func (w kvWriter) Put(key string, value []byte) error {
 }
 
 // putReach writes a reach value under its key in the view's own state reach (ADR 0556 clause 2), its only writer.
-func putReach(ctx context.Context, js jetstream.JetStream, w view.ReachWrite) error {
+func putReach(ctx context.Context, js jetstream.JetStream, w view.ReachWrite) (uint64, error) {
 	kv, err := js.KeyValue(ctx, viewstate.ReachBucket)
 	if err != nil {
-		return fmt.Errorf("the bucket %s: %w", viewstate.ReachBucket, err)
+		return 0, fmt.Errorf("the bucket %s: %w", viewstate.ReachBucket, err)
 	}
-	_, err = kv.Put(ctx, w.Key, w.Value)
-	return err
+	return kv.Put(ctx, w.Key, w.Value)
+}
+
+// ReachedEventType is the event the view says after it writes a repository's reach: mesh.mod.mesh-delivery-view.event.reached
+// (ADR 0596 §2). Its body is ReachedEvent.
+const ReachedEventType = "reached"
+
+// ReachedEvent is the body of the event reached: the key of the reach (the repository's) and the reach as the view wrote it.
+type ReachedEvent struct {
+	Key   string          `json:"key"`
+	Reach json.RawMessage `json:"reach"`
 }
 
 // The view's own states of its first views (views.StatusState) and of its findings, where the condition keeper raises a

modules/mesh-delivery-view/manifest_test.go

@@ -139,8 +139,8 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7
 		t.Errorf("consumes = %v, want %v", m.Consumes, want)
 	}
 	// The one event the view says: a source revision is delivered (1906/D9, ADR 0539 §17), derived by it from the wants and the haves.
-	if !reflect.DeepEqual(m.Emits, []string{"delivered", "moving", "settled", "stuck"}) {
-		t.Errorf("emits = %v, want exactly delivered, moving, settled and stuck (1906/D9, D10)", m.Emits)
+	if !reflect.DeepEqual(m.Emits, []string{"delivered", "reached", "moving", "settled", "stuck"}) {
+		t.Errorf("emits = %v, want exactly delivered, reached (ADR 0596 §2), moving, settled and stuck (1906/D9, D10)", m.Emits)
 	}
 	// The view uses no build seat: it reads their workers by the broker's named grant and publishes no ask (ADR 0525 §2).
 	if len(m.Uses) != 0 {

modules/mesh-delivery-view/module.json

@@ -31,6 +31,7 @@
   ],
   "emits": [
     "delivered",
+    "reached",
     "moving",
     "settled",
     "stuck"