Merge pull request 'Teller: teller_end_ask ends one stale ask by hand (issue 774)' (#124) from fix/merge-ask-ends-on-merge into main

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

@@ -0,0 +1,109 @@
+// Command mesh-delivery-teller-tools serves the teller's recovery tool (issue 774): the runtime launches it for a call.
+// teller_end_ask writes one value of the teller's state ends and waits for the teller's answer there. The teller is the
+// other program: it serves no request subject (ADR 0437 §4).
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"os"
+	"strings"
+	"sync"
+	"time"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+
+	"mesh-delivery-teller/internal/ends"
+)
+
+// asker writes an ends value and waits for the teller's answer to it.
+type asker func(key string, value []byte) (ends.End, error)
+
+// Tools are the teller's tools.
+func Tools(ask asker, now func() time.Time) []stdio.Tool {
+	str := func(args map[string]any, k string) string { s, _ := args[k].(string); return strings.TrimSpace(s) }
+	return []stdio.Tool{
+		{Name: ends.Tool,
+			Description: "End one open ask the delivery teller holds at the operator channel, by hand, saying why: the teller " +
+				"cancels it at the router, which frees one of the three places the router gives the teller, or drops it from the " +
+				"teller's wait for room. For an ask whose question is gone but whose key was never taken away (a merge ask of a pull " +
+				"request merged by another path, a backup ask about a backup no longer failing), which otherwise holds its place " +
+				"until it lapses, so a newer ask is refused for want of room. Name the ask by its router id (as " +
+				"operator-channel's asks show it) or by the key of the gate's or the assigner's asks. Answers which ask it found, " +
+				"whether it ended it, and what it did; an ask answered, lapsed or ended already is left and said so.",
+			Input: map[string]any{"type": "object", "required": []string{"ask", "why"}, "properties": map[string]any{
+				"ask": map[string]any{"type": "string", "description": "the ask's router id, or the key of the gate's or the assigner's asks it was asked for"},
+				"why": map[string]any{"type": "string", "description": "why it is ended by hand, in plain words"}}},
+			Run: func(args map[string]any) (any, error) {
+				key, value, err := ends.New(str(args, "ask"), str(args, "why"), stdio.CallerOf(args), now())
+				if err != nil {
+					return nil, err
+				}
+				e, err := ask(key, value)
+				if err != nil {
+					return nil, err
+				}
+				return e.Result, 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) {
+		answered := make(chan ends.End, 1)
+		var rev uint64
+		var mu sync.Mutex
+		take := func(revision uint64, v []byte) {
+			mu.Lock()
+			mine := rev
+			mu.Unlock()
+			if mine == 0 || revision <= mine {
+				return
+			}
+			e, err := ends.Decode(v)
+			if err != nil || !e.Done() {
+				return
+			}
+			select {
+			case answered <- e:
+			default:
+			}
+		}
+		if err := s.Watch(key, func(c stdio.StateChange) error {
+			if c.Op == "put" {
+				take(c.Revision, c.Value)
+			}
+			return nil
+		}); err != nil {
+			return ends.End{}, fmt.Errorf("watching %s.%s: %w", ends.State, key, err)
+		}
+		r, err := s.Put(key, json.RawMessage(value))
+		if err != nil {
+			return ends.End{}, fmt.Errorf("the end could not be written to %s: %w", ends.Bucket, err)
+		}
+		mu.Lock()
+		rev = r
+		mu.Unlock()
+		// The watch may have heard the answer before the revision was known: read it once too.
+		if e, err := s.Get(key); err == nil && e != nil {
+			take(e.Revision, e.Value)
+		}
+		wait, cancel := context.WithTimeout(context.Background(), bound) // a context's bound, no timer (ADR 0437 §8)
+		defer cancel()
+		select {
+		case e := <-answered:
+			return e, nil
+		case <-wait.Done():
+			return ends.End{}, fmt.Errorf("the teller did not answer within %s: is it running? its answer will be in %s.%s", bound, ends.Bucket, key)
+		}
+	}
+}
+
+func main() {
+	if err := stdio.Serve("", Tools(askEnd(stdio.State(ends.State), 25*time.Second), time.Now)); err != nil {
+		fmt.Fprintf(os.Stderr, "[mesh-delivery-teller-tools] %v\n", err)
+		os.Exit(1)
+	}
+}

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

@@ -0,0 +1,92 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"slices"
+	"time"
+
+	"github.com/nats-io/nats.go/jetstream"
+
+	"mesh-delivery-teller/internal/ends"
+)
+
+// EndAsk ends one ask of the teller's by hand (the tool teller_end_ask, issue 774): an ask open at the router is cancelled
+// there and marked ended here, and one waiting for room is dropped from the wait, so it is asked no more under that router
+// id. The ask is named by its router id or by the key of the gate's or the assigner's asks it was asked for (then its latest
+// asking). An ask that is neither open nor waiting is said so and left: it holds no place at the router. The gate's or the
+// assigner's key is not the teller's to take away: a later asking of it, under a new router id, is asked as any.
+func (t *Teller) EndAsk(ask, why, by string) ends.Result {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	id, key := ask, ""
+	if s, rec, err := t.seenOf(ask); err != nil {
+		return ends.Result{Says: err.Error()}
+	} else if rec != nil {
+		key = s.Key
+	} else if g, grec, err := t.seenAt(gateRecordKey(ask)); err != nil {
+		return ends.Result{Says: err.Error()}
+	} else if grec != nil && g.ID != "" {
+		id, key = g.ID, ask
+	} else {
+		return ends.Result{Says: "the teller holds no ask under " + ask + ": name its router id, as operator-channel's asks show it, or the key of the gate's or the assigner's asks"}
+	}
+	r := ends.Result{ID: id, Key: key}
+	waited := false
+	if ws, rec, err := t.waiting(); err == nil && rec != nil {
+		if i := slices.IndexFunc(ws, func(x waitingAsk) bool { return x.ID == id }); i >= 0 {
+			if t.keepWaiting(slices.Delete(ws, i, i+1), rec) != nil {
+				r.Says = "the ask " + id + " waits for room, and the wait could not be written: it is not ended"
+				return r
+			}
+			waited = true
+		}
+	}
+	k, _, err := t.keptOf(id)
+	if err != nil {
+		r.Says = err.Error()
+		return r
+	}
+	if !open(k) {
+		r.Ended = waited
+		if waited {
+			r.Says = "the ask " + id + " waited for room at the router: it is dropped from the wait and asked no more"
+		} else {
+			r.Says = "the ask " + id + " is not open (answered, lapsed or ended already): it holds no place at the router, and nothing is done"
+		}
+		t.logf("%s: %s, by %s: %s", key, r.Says, by, why)
+		return r
+	}
+	t.cancelOpen(key, id)
+	if k, _, err := t.keptOf(id); err != nil || open(k) {
+		r.Says = "the ask " + id + " could not be cancelled at the router (see the teller's journal): it stays open"
+		return r
+	}
+	r.Ended = true
+	r.Says = "the ask " + id + " is cancelled at the router, which frees its place; the next ask waiting for room is asked when the router says it ended"
+	t.logf("%s: the ask %s is ended by hand, by %s: %s", key, id, by, why)
+	return r
+}
+
+// endAsked answers one value of the state ends: one not yet answered is carried out and written back, by compare-and-set at
+// its revision, with its result. One answered already (the teller's own write) is left.
+func endAsked(ctx context.Context, t *Teller, kv jetstream.KeyValue, key string, rev uint64, value []byte) {
+	e, err := ends.Decode(value)
+	if err != nil {
+		t.logf("%s.%s holds what is not an end, left: %v", ends.State, key, err)
+		return
+	}
+	if e.Done() {
+		return
+	}
+	r := t.EndAsk(e.Ask, e.Why, e.By)
+	e.Result, e.Answered = &r, time.Now().UTC()
+	data, err := json.Marshal(e)
+	if err != nil {
+		return
+	}
+	if _, err := kv.Update(ctx, key, data, rev); err != nil && !errors.Is(err, jetstream.ErrKeyExists) && !errors.Is(err, jetstream.ErrKeyRevisionMismatch) {
+		t.logf("%s.%s: the answer could not be written: %v", ends.State, key, err)
+	}
+}

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

@@ -3,7 +3,8 @@
 // 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.
+// subject: its one tool, teller_end_ask (the program mesh-delivery-teller-tools, issue 774), asks through its state ends,
+// which it watches.
 package main
 
 import (
@@ -22,6 +23,8 @@ import (
 
 	"git.novox.be/novox/mesh-sdk/go/asker"
 	"git.novox.be/novox/mesh-sdk/go/delivery"
+
+	"mesh-delivery-teller/internal/ends"
 )
 
 func main() {
@@ -74,7 +77,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	}
 	// 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)
+	failed := make(chan error, 4)
 	assignerKV, viewKV := process.OptionalBucket(ctx, js, delivery.AssignerAsksBucket, teller, logger.Printf), process.OptionalBucket(ctx, js, delivery.ViewAnswersBucket, teller, logger.Printf)
 	// The assigner's asks are raised on the operator channel whether or not the view's answers can be read (issue 1906).
 	if viewKV != nil {
@@ -91,6 +94,15 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 			failed <- watchAsks(ctx, viewKV, func(key string, _ uint64, _ []byte) { t.OnViewAnswer(key) }, func(string) {}, nil)
 		}()
 	}
+	// A person ends one ask by hand with teller_end_ask, which writes the state ends (issue 774): watched as the gate's asks
+	// are, once it is on the bus.
+	process.AwaitBucket(ctx, js, conn, ends.Bucket, "teller_end_ask is not answered; the asks still end when they are answered, lapse or their key is taken away", logger.Printf).Ready(func(kv jetstream.KeyValue) {
+		go func() {
+			if err := watchAsks(ctx, kv, func(key string, rev uint64, value []byte) { endAsked(ctx, t, kv, key, rev, value) }, func(string) {}, nil); err != nil {
+				failed <- fmt.Errorf("watching the ends asked by hand: %w", err)
+			}
+		}()
+	})
 	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)

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

@@ -287,6 +287,55 @@ func TestAnAsksKeyBecomesOneRouterAskAndDeletingItCancelsIt(t *testing.T) { // 1
 	if n := len(b.askedOf(asks.CancelSubject(delivery.TellerModule))); n != 1 {
 		t.Errorf("the ask was cancelled %d times", n)
 	}
+	// Issue 774 (2026-10-31): two merge asks of pull requests merged by another path held two of the router's three places,
+	// their keys never taken away, and a fourth ask was refused for want of room. teller_end_ask ends one by hand, named by its
+	// gate key or its router id; the router's word that it ended asks the one waiting for room again.
+	for _, c := range []struct {
+		name string
+		by   func(key string, ga delivery.Ask) string
+	}{
+		{"774: ended by hand, named by its gate key", func(key string, _ delivery.Ask) string { return key }},
+		{"774: ended by hand, named by its router id", func(_ string, ga delivery.Ask) string { return ga.Ask.ID }},
+	} {
+		t.Run(c.name, func(t *testing.T) {
+			b := newBed(t)
+			k1, first := b.askOf(t, 1, 1)
+			b.askOf(t, 2, 2)
+			b.askOf(t, 3, 3)
+			_, fourth := b.askOf(t, 4, 4)
+			b.refuse(t, fourth.Ask.ID, roomRefusal)
+			r := b.t.EndAsk(c.by(k1, first), "its pull request merged without the Approve", "lead@shanks")
+			cancels := b.askedOf(asks.CancelSubject(delivery.TellerModule))
+			if !r.Ended || r.ID != first.Ask.ID || r.Key != k1 || len(cancels) != 1 || string(cancels[0].body) != `{"id":"`+first.Ask.ID+`"}` {
+				t.Fatalf("end = %+v, cancels %+v", r, cancels)
+			}
+			b.deliver(t, asks.Warrant{Ask: first.Ask.ID, Asker: delivery.TellerModule, Outcome: asks.OutcomeCancelled, At: now, AskDigest: first.Ask.Digest()})
+			if got := b.askedIDs(); len(got) != 5 || got[4] != fourth.Ask.ID {
+				t.Errorf("asks = %v, want the refused %s asked again once the ask ended by hand left", got, fourth.Ask.ID)
+			}
+			if again := b.t.EndAsk(k1, "again", "lead@shanks"); again.Ended || len(b.askedOf(asks.CancelSubject(delivery.TellerModule))) != 1 {
+				t.Errorf("a second end = %+v: an ended ask is left", again)
+			}
+		})
+	}
+	t.Run("774: an ask waiting for room is dropped from the wait; a name the teller holds no ask under is said", func(t *testing.T) {
+		b := newBed(t)
+		_, first := b.askOf(t, 1, 1)
+		b.askOf(t, 2, 2)
+		b.askOf(t, 3, 3)
+		_, fourth := b.askOf(t, 4, 4)
+		b.refuse(t, fourth.Ask.ID, roomRefusal)
+		if r := b.t.EndAsk(fourth.Ask.ID, "asked about what no longer holds", "lead@shanks"); !r.Ended || len(b.askedOf(asks.CancelSubject(delivery.TellerModule))) != 0 {
+			t.Fatalf("end of a waiting ask = %+v", r)
+		}
+		b.deliver(t, warrantFor(first.Ask, "approve"))
+		if got := b.askedIDs(); len(got) != 4 {
+			t.Errorf("asks = %v: the ask ended while it waited was asked again", got)
+		}
+		if r := b.t.EndAsk("merge.nothing.here", "a typo", "lead@shanks"); r.Ended || !strings.Contains(r.Says, "holds no ask") {
+			t.Errorf("end of an unknown ask = %+v", r)
+		}
+	})
 }
 
 func TestAWarrantThatHoldsWritesOneAnswerAndASecondWritesNone(t *testing.T) { // 1343/AC2

modules/mesh-delivery-teller/internal/ends/ends.go

@@ -0,0 +1,79 @@
+// Package ends is the teller's recovery tool teller_end_ask and the value it writes (issue 774; the instruction "tools are
+// how a module is spoken to"). An ask the teller holds open at the router keeps one of the three places the router gives
+// its asker (asks.MostOpen) until it is answered, it lapses, or the gate or the assigner takes its key away. When that last
+// never comes (on 2026-10-31 two merge asks of pull requests merged by another path held two places for a day, and a
+// placement ask was refused for want of room), a person ends the ask by hand. The teller's process serves no request
+// subject (ADR 0437 §4), so the tool writes an end into the teller's state ends, which the teller watches, carries out
+// and answers there, as the indexer answers indexer_take_in.
+package ends
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+const (
+	Tool = "teller_end_ask"
+	// State is the teller's state of ends asked by hand: the bucket mesh-delivery-teller_ends.
+	State  = "ends"
+	Bucket = delivery.TellerModule + "_" + State
+)
+
+// Result is the teller's answer to an end: which ask it found (its router id and the gate's or the assigner's key it was
+// asked for), whether this end ended it, and what it did in words.
+type Result struct {
+	ID    string `json:"id,omitempty"`
+	Key   string `json:"key,omitempty"`
+	Ended bool   `json:"ended"`
+	Says  string `json:"says"`
+}
+
+// End is a value of the state ends: the ask named (a router id, as operator-channel's asks show it, or the key of the
+// gate's or the assigner's asks it was asked for), why, who and when, and, once the teller carried it out, its answer.
+type End struct {
+	V        int       `json:"v"`
+	Ask      string    `json:"ask"`
+	Why      string    `json:"why"`
+	By       string    `json:"by,omitempty"`
+	At       time.Time `json:"at"`
+	Result   *Result   `json:"result,omitempty"`
+	Answered time.Time `json:"answered,omitzero"`
+}
+
+// Done says the end is answered.
+func (e End) Done() bool { return !e.Answered.IsZero() && e.Result != nil }
+
+// Decode reads a value of the state ends.
+func Decode(data []byte) (End, error) {
+	var e End
+	if err := json.Unmarshal(data, &e); err != nil {
+		return e, err
+	}
+	if e.V != 1 {
+		return e, fmt.Errorf("an ends value of version %d, not 1", e.V)
+	}
+	if e.Ask == "" || e.Why == "" {
+		return e, errors.New("an ends value naming no ask or no why")
+	}
+	return e, nil
+}
+
+// New is the value teller_end_ask writes: its key and the value. The key is the router's spelling of the name given
+// (delivery.AskID), which a bucket key takes whatever the name holds, and is the same for the same name, so a second end
+// of one ask overwrites the first.
+func New(ask, why, by string, at time.Time) (string, []byte, error) {
+	ask, why = strings.TrimSpace(ask), strings.TrimSpace(why)
+	if ask == "" {
+		return "", nil, errors.New("name the ask to end (ask): its router id, as operator-channel's asks show it, or the key of the gate's or the assigner's asks")
+	}
+	if why == "" {
+		return "", nil, errors.New("say why the ask is ended (why), in plain words: it is logged with who ended it")
+	}
+	data, err := json.Marshal(End{V: 1, Ask: ask, Why: why, By: by, At: at.UTC()})
+	return delivery.AskID(ask), data, err
+}

modules/mesh-delivery-teller/manifest_test.go

@@ -23,18 +23,20 @@ type state struct {
 }
 
 type artifact struct {
-	Name     string `json:"name"`
+	Name     string   `json:"name"`
-	Kind     string `json:"kind"`
+	Kind     string   `json:"kind"`
-	Language string `json:"language"`
+	Language string   `json:"language"`
-	System   string `json:"system"`
+	System   string   `json:"system"`
-	From     string `json:"from"`
+	From     string   `json:"from"`
-	Binary   string `json:"binary"`
+	Binary   string   `json:"binary"`
+	Loads    []string `json:"loads"`
 }
 
 type manifest struct {
 	Module       string            `json:"module"`
 	Version      string            `json:"version"`
 	Slug         string            `json:"slug"`
+	Tools        []string          `json:"tools"`
 	Uses         []string          `json:"uses"`
 	State        []state           `json:"state"`
 	Reads        []string          `json:"reads"`
@@ -77,8 +79,8 @@ func TestManifestDeclarations(t *testing.T) { // 1240/AC1
 		t.Errorf("module = %q", m.Module)
 	}
 	// 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).
+	// gate key and revision each ask was made at (issue 1343); ends is the ends asked by hand with teller_end_ask (issue 774).
-	if want := []state{{Name: "answers", History: 5}, {Name: "asked", History: 1}}; !reflect.DeepEqual(m.State, want) {
+	if want := []state{{Name: "answers", History: 5}, {Name: "asked", History: 1}, {Name: "ends", History: 1}}; !reflect.DeepEqual(m.State, want) {
 		t.Errorf("state = %+v, want %+v", m.State, 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).
@@ -94,8 +96,12 @@ func TestManifestDeclarations(t *testing.T) { // 1240/AC1
 	if m.SecretsOwner != "mesh-delivery-teller" {
 		t.Errorf("secrets-owner = %q", m.SecretsOwner)
 	}
-	// No consumes, invokes, emits, streams, bus, seat or tool (ADR 0437 §4).
+	// 774: teller_end_ask, served by its tools bundle through the state ends; the process serves none.
-	for _, k := range []string{"consumes", "invokes", "streams", "logs", "emits", "claims", "provides", "requires", "defines-seats", "tools", "seat", "seats", "bus"} {
+	if want := []string{"teller_end_ask"}; !reflect.DeepEqual(m.Tools, want) {
+		t.Errorf("tools = %v, want %v", m.Tools, want)
+	}
+	// No consumes, invokes, emits, streams, bus or seat (ADR 0437 §4).
+	for _, k := range []string{"consumes", "invokes", "streams", "logs", "emits", "claims", "provides", "requires", "defines-seats", "seat", "seats", "bus"} {
 		if _, ok := top[k]; ok {
 			t.Errorf("the manifest declares %q: the teller has none", k)
 		}
@@ -127,13 +133,18 @@ func TestManifestProcess(t *testing.T) { // 1240/AC1
 	if _, bus := p["bus"]; bus {
 		t.Errorf("the process declares a bus: it runs on its own broker credential")
 	}
-	if len(m.Build.Artifacts) != 1 {
+	if len(m.Build.Artifacts) != 2 {
-		t.Fatalf("artifacts = %+v", m.Build.Artifacts)
+		t.Fatalf("artifacts = %+v, want the program and the tools", m.Build.Artifacts)
 	}
 	a := m.Build.Artifacts[0]
 	if st, err := os.Stat(a.From); err != nil || !st.IsDir() || a.Binary != "mesh-delivery-teller" || !strings.HasSuffix(a.From, a.Binary) {
 		t.Errorf("artifact = %+v", a)
 	}
+	if a := m.Build.Artifacts[1]; a.Name != "tools" || a.Kind != "bundle" || a.Binary != "mesh-delivery-teller-tools" || !reflect.DeepEqual(a.Loads, []string{"mesh-delivery-teller-tools"}) {
+		t.Errorf("tools artifact = %+v", a) // 774
+	} else if st, err := os.Stat(a.From); err != nil || !st.IsDir() {
+		t.Errorf("the tools' source %s: %v", a.From, err)
+	}
 }
 
 func TestServesNoRequestSubjectAndHoldsNoTicker(t *testing.T) { // 1240/AC3 (ADR 0437 §8)

modules/mesh-delivery-teller/module.json

@@ -2,9 +2,13 @@
   "module": "mesh-delivery-teller",
   "version": "1",
   "slug": "teller",
+  "tools": [
+    "teller_end_ask"
+  ],
   "state": [
     { "name": "answers", "history": 5 },
-    { "name": "asked", "history": 1 }
+    { "name": "asked", "history": 1 },
+    { "name": "ends", "history": 1 }
   ],
   "reads": [
     "mesh-delivery-assigner.asks",
@@ -60,6 +64,17 @@
         "system": "arch",
         "from": "cmd/mesh-delivery-teller",
         "binary": "mesh-delivery-teller"
+      },
+      {
+        "name": "tools",
+        "kind": "bundle",
+        "language": "go",
+        "system": "arch",
+        "from": "cmd/mesh-delivery-teller-tools",
+        "binary": "mesh-delivery-teller-tools",
+        "loads": [
+          "mesh-delivery-teller-tools"
+        ]
       }
     ]
   }