DLV-B6: a tag ask is marked numbered with its versions, and one unanswered is narrated when its bound ends on the bus

A check's build of the ask's head writes the ask numbered in tag-asks by compare-and-set. An unanswered ask has a key in the
new key-ttl state tag-ask-bounds (five minutes); the bus's marker of its end, heard on the watch, narrates the ask with what
it waits on (the check's build not claimed, names waiting on their floor) and writes the bound again; no timer (design 51
*Tagged versions*). A restart arms each unanswered ask whose bound ended meanwhile.

Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig

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

@@ -79,6 +79,13 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 			return fmt.Errorf("MESH_BUILDER_STEP_BOUND: %w", err)
 		}
 	}
+	// An unanswered tag ask is narrated every TagAskBound (design 51: five minutes); a test gives a shorter one.
+	tagAskBound := versions.TagAskBound
+	if v := strings.TrimSpace(getenv("MESH_BUILDER_TAG_ASK_BOUND")); v != "" {
+		if tagAskBound, err = time.ParseDuration(v); err != nil {
+			return fmt.Errorf("MESH_BUILDER_TAG_ASK_BOUND: %w", err)
+		}
+	}
 	tools := &buildrun.ToolchainSet{}
 	// Each build module's part of a build is an ask on its build seat's stream, pulled by a build node with a free slot
 	// (novox/hq ADR 0525): the builder chooses no node, so it reads neither presence nor have and holds no capacity.
@@ -140,11 +147,15 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		Logf:       logger.Printf,
 		Stored:     storedVia(calls.Call),
 		Node:       cred.Node,
-		Numbers:    build.VersionNumbers{N: numberer},
 	}
 	tagAsksKV := process.AwaitBucket(ctx, js, conn, versions.TagAsksBucket, "builder_tag and builder_untag are not answered", logger.Printf)
-	// A check's build of a head a kept tag ask names numbers its packages tagged (design 51 *Tagged versions*).
+	// A check's build of a head a kept tag ask names numbers its packages tagged (design 51 *Tagged versions*), and the ask
+	// is marked numbered with its versions.
 	b.Tags = tagsVia(tagAsksKV.KV, logger.Printf)
+	b.Numbers = build.VersionNumbers{N: numberer, TagNumbered: tagNumbered(tagAsksKV.KV, logger.Printf, time.Now)}
+	// An ask not answered within five minutes is narrated with what it waits on: its bound's end on the bus says when.
+	boundsKV := process.AwaitBucket(ctx, js, conn, versions.TagAskBoundsBucket, "an unanswered tag ask is not narrated", logger.Printf)
+	bounds := &tagBounds{asks: tagAsksKV.KV, bounds: boundsKV.KV, bound: tagAskBound, waits: floorWaits.Waiting, logf: logger.Printf, now: time.Now}
 	// A state bound later numbers the builds looked at without it.
 	versionsKV.Ready(func(jetstream.KeyValue) { b.Reconsider(ctx) })
 	// The versions' hand acts (builder_publish_again, builder_floor_again) come through reasks, as builder_build_again.
@@ -163,7 +174,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	// revision (the gate's source state). The source's revisions are watched with their history (a direct get of one revision
 	// is not what a read grants): a revision heard after the candidate that names it builds that candidate then. Where it
 	// cannot be read the trunk's keys are built and the candidates wait (issue 1945).
-	errs := make(chan error, 9)
+	errs := make(chan error, 10)
 	trunks := &trunkSource{ctx: ctx, js: js, heads: &sourceHeads{}, moved: func(ctx context.Context) {
 		// A source revision heard is the forge's merge: it wakes a floor whose bridge_tags was silent (design 51 *The floor*).
 		(&floorWake{b: b, waits: floorWaits, logf: logger.Printf}).wake(ctx, versions.WakeForge, "", "")
@@ -326,13 +337,28 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	// A tag ask or a branch's removal by hand (builder_tag, builder_untag) is written to tag-asks and answered there
 	// (design 51 *Tagged versions*); the checker's tag-wanted and the branch's end join it on their consumers (DLV-B6).
 	tagAsksKV.Ready(func(kv jetstream.KeyValue) {
-		each := func(e jetstream.KeyValueEntry) { tagAsked(ctx, acts, kv, e, logger.Printf) }
+		each := func(e jetstream.KeyValueEntry) {
+			tagAsked(ctx, acts, kv, e, logger.Printf)
+			if !opOf(e) {
+				bounds.disarm(ctx, e.Key())
+			} else if t, err := versions.DecodeTagAsk(e.Value()); err == nil {
+				bounds.asked(ctx, e.Key(), t)
+			}
+		}
 		if tw, err := settle(ctx, kv, each); err != nil {
 			process.NotWatched(logger.Printf, versions.TagAsksBucket, err, "builder_tag and builder_untag are not answered until the builder starts again")
 		} else {
 			go func() { errs <- follow(ctx, tw, each) }()
 		}
 	})
+	boundsKV.Ready(func(kv jetstream.KeyValue) {
+		each := func(e jetstream.KeyValueEntry) { bounds.ended(ctx, e) }
+		if bw, err := settle(ctx, kv, each); err != nil {
+			process.NotWatched(logger.Printf, versions.TagAskBoundsBucket, err, "an unanswered tag ask is not narrated until the builder starts again")
+		} else {
+			go func() { errs <- follow(ctx, bw, each) }()
+		}
+	})
 	// A person reads a build seat's stream with builder_seat_queue, which writes the queues state (issue 1945 note 62): the
 	// builder alone may read the stream, so it answers there.
 	process.AwaitBucket(ctx, js, conn, build.QueuesBucket, "builder_seat_queue is not answered", logger.Printf).Ready(func(kv jetstream.KeyValue) {

modules/mesh-delivery-builder/cmd/mesh-delivery-builder/tagbounds.go

@@ -0,0 +1,121 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"time"
+
+	"github.com/nats-io/nats.go/jetstream"
+
+	"mesh-delivery-builder/internal/versions"
+)
+
+// tagBounds narrates a tag ask the builder has not answered within five minutes, with what it waits on (design 51
+// *Tagged versions*): each unanswered ask has a key in the state tag-ask-bounds written with its own time to live, and
+// the bus's marker of its end, heard on the watch, is what narrates it and writes the bound again. No timer.
+type tagBounds struct {
+	asks   func() jetstream.KeyValue
+	bounds func() jetstream.KeyValue
+	waits  func() []versions.FloorWait
+	logf   func(string, ...any)
+	now    func() time.Time
+	bound  time.Duration
+}
+
+// arm writes key's bound when none stands: a Create, so a standing bound keeps its time and a restart arms each
+// unanswered ask whose bound ended while the builder was away.
+func (tb *tagBounds) arm(ctx context.Context, key string) {
+	kv := tb.bounds()
+	if kv == nil {
+		return
+	}
+	if _, err := kv.Create(ctx, key, []byte(`{}`), jetstream.KeyTTL(tb.bound)); err != nil && !errors.Is(err, jetstream.ErrKeyExists) {
+		tb.logf("the bound of the tag ask %s is not written, so it is not narrated past five minutes: %v", key, err)
+	}
+}
+
+// disarm removes key's bound: the ask was answered or ended.
+func (tb *tagBounds) disarm(ctx context.Context, key string) {
+	if kv := tb.bounds(); kv != nil {
+		_ = kv.Delete(ctx, key)
+	}
+}
+
+// asked is told each value of tag-asks: an unanswered ask is armed, any other disarmed.
+func (tb *tagBounds) asked(ctx context.Context, key string, t versions.TagAsk) {
+	if t.Unanswered() {
+		tb.arm(ctx, key)
+		return
+	}
+	tb.disarm(ctx, key)
+}
+
+// ended is told each change of tag-ask-bounds: a bound whose time ended (the bus's marker, a purge on the watch) on an
+// ask still unanswered narrates it and arms it again.
+func (tb *tagBounds) ended(ctx context.Context, e jetstream.KeyValueEntry) {
+	if e.Operation() != jetstream.KeyValuePurge {
+		return
+	}
+	kv := tb.asks()
+	if kv == nil {
+		return
+	}
+	got, err := kv.Get(ctx, e.Key())
+	if err != nil {
+		return // the ask ended with its branch
+	}
+	t, err := versions.DecodeTagAsk(got.Value())
+	if err != nil || !t.Unanswered() {
+		return
+	}
+	var waits []versions.FloorWait
+	if tb.waits != nil {
+		waits = tb.waits()
+	}
+	tb.logf("%s", versions.NarrateUnanswered(t, waits, tb.now()))
+	tb.arm(ctx, e.Key())
+}
+
+// tagNumbered marks the tag ask of a check's build's head numbered with its versions in tag-asks, by compare-and-set.
+func tagNumbered(kv func() jetstream.KeyValue, logf func(string, ...any), now func() time.Time) func(context.Context, versions.BuildClaim, versions.Numbering) {
+	return func(ctx context.Context, c versions.BuildClaim, got versions.Numbering) {
+		store := kv()
+		if store == nil {
+			return
+		}
+		lister, err := store.ListKeysFiltered(ctx, versions.EscapeToken(c.Repository)+".>")
+		if err != nil {
+			logf("the tag ask of %s at %s is not marked numbered: %v", c.Repository, c.Head, err)
+			return
+		}
+		for key := range lister.Keys() {
+			for range 3 {
+				e, err := store.Get(ctx, key)
+				if err != nil {
+					break
+				}
+				t, err := versions.DecodeTagAsk(e.Value())
+				if err != nil {
+					break
+				}
+				t, changed := t.WithNumbered(c.Module, c.Head, got.Versions, now())
+				if !changed {
+					break
+				}
+				data, err := json.Marshal(t)
+				if err != nil {
+					break
+				}
+				if _, err = store.Update(ctx, key, data, e.Revision()); err == nil {
+					logf("%s: tag ask of %s %s: %s", key, t.Repository, t.Branch, t.Answer)
+					break
+				}
+				if !errors.Is(err, jetstream.ErrKeyExists) && !errors.Is(err, jetstream.ErrKeyRevisionMismatch) {
+					logf("the tag ask %s is not marked numbered: %v", key, err)
+					break
+				}
+			}
+		}
+	}
+}

modules/mesh-delivery-builder/internal/build/numbers.go

@@ -19,7 +19,12 @@ type Numbers interface {
 }
 
 // VersionNumbers is Numbers over the builder's state versions.
-type VersionNumbers struct{ N *versions.Numberer }
+type VersionNumbers struct {
+	N *versions.Numberer
+	// TagNumbered is told the versions a check's build of a tag ask's head was numbered (the ask is then marked numbered
+	// in tag-asks, design 51 *Tagged versions*); nil tells nobody.
+	TagNumbered func(ctx context.Context, c versions.BuildClaim, got versions.Numbering)
+}
 
 // ClaimOf is the claim a module value's trunk build is numbered by: a package per source revision (Src), the rest per
 // fingerprint, in merge order (Src).
@@ -41,7 +46,11 @@ func (v VersionNumbers) Number(ctx context.Context, module string, m delivery.Mo
 		// A check's build of a head a tag ask names: its packages tagged, below the next version (design 51 *Tagged versions*).
 		c.Tag, c.Head, c.Revision, c.Merge = tag, m.Commit, "", 0
 	}
-	return v.N.Number(ctx, c)
+	got, err := v.N.Number(ctx, c)
+	if err == nil && tag != "" && len(got.Versions) > 0 && v.TagNumbered != nil {
+		v.TagNumbered(ctx, c, got)
+	}
+	return got, err
 }
 
 func (v VersionNumbers) Built(ctx context.Context, module string, m delivery.Module, plan Plan, got versions.Numbering, made []delivery.BuiltArtifact) error {

modules/mesh-delivery-builder/internal/versions/tagbound.go

@@ -0,0 +1,74 @@
+package versions
+
+import (
+	"fmt"
+	"maps"
+	"slices"
+	"strings"
+	"time"
+)
+
+// The bound of an unanswered tag ask (design 51 *Tagged versions*: "an ask the builder has not answered within five
+// minutes is narrated in the build log with what it waits on"): a key of the state tag-ask-bounds, written with its own
+// time to live, whose expiry the builder hears on its watch of the state. No timer: the bus ends the bound.
+const (
+	TagAskBoundsState  = "tag-ask-bounds"
+	TagAskBoundsBucket = "mesh-delivery-builder_tag-ask-bounds"
+	TagAskBound        = 5 * time.Minute
+)
+
+// Unanswered says whether a tag ask still waits for its versions: kept, not an untag, none numbered yet.
+func (t TagAsk) Unanswered() bool {
+	return !t.Untag && t.State == TagAsked && len(t.Versions) == 0
+}
+
+// WithNumbered is the tag ask once a check's build of its head numbered module's packages under its tag: numbered, each
+// version by its name, the answer naming them. An untag, or an ask of another head, is left as it is.
+func (t TagAsk) WithNumbered(module, head string, got map[string]string, at time.Time) (TagAsk, bool) {
+	if t.Untag || t.Head != head || len(got) == 0 || t.State == TagRemoving || t.State == TagRemoved {
+		return t, false
+	}
+	if t.Versions == nil {
+		t.Versions = map[string]string{}
+	}
+	changed := false
+	for artifact, v := range got {
+		name := NameOf(module, artifact)
+		if t.Versions[name] != v {
+			t.Versions[name], changed = v, true
+		}
+	}
+	if !changed && t.State == TagNumbered {
+		return t, false
+	}
+	var said []string
+	for _, name := range slices.Sorted(maps.Keys(t.Versions)) {
+		said = append(said, name+" "+t.Versions[name])
+	}
+	t.State, t.Answered = TagNumbered, at.UTC()
+	t.Answer = "numbered: " + strings.Join(said, ", ")
+	return t, true
+}
+
+// NarrateUnanswered is the line said for a tag ask its bound found unanswered: since when, and what it waits on, the
+// names waiting on their floor named with their silent sources.
+func NarrateUnanswered(t TagAsk, waits []FloorWait, now time.Time) string {
+	on := "the check's build of " + short(t.Head) + " is not claimed yet (the checker builds a group member's head once it asks)"
+	var floors []string
+	for _, w := range waits {
+		floors = append(floors, w.Name+" waits on its floor ("+strings.Join(w.Silent, ", ")+" silent)")
+	}
+	if len(floors) > 0 {
+		slices.Sort(floors)
+		on += "; " + strings.Join(floors, "; ")
+	}
+	return fmt.Sprintf("the tag ask of %s %s at %s (%s) is not answered after %s: it waits on %s",
+		t.Repository, t.Branch, short(t.Head), t.Tag, now.Sub(t.At).Round(time.Second), on)
+}
+
+func short(head string) string {
+	if len(head) > 12 {
+		return head[:12]
+	}
+	return head
+}

modules/mesh-delivery-builder/internal/versions/versions_test.go

@@ -639,6 +639,29 @@ func TestTheBuildersVersions(t *testing.T) {
 		if end, _ := acts.BranchEnded(context.Background(), "o/r", "feat/x"); !end.Done {
 			t.Errorf("every seat said removed, and the branch's end is not done: %+v", end)
 		}
+		// B6: a check's build of the ask's head marks it numbered with its versions; another head, or an untag, is left.
+		if !a.Unanswered() {
+			t.Errorf("a kept ask with no version is answered: %+v", a)
+		}
+		if _, changed := a.WithNumbered("mesh-sdk", "other", map[string]string{"go": "0.1.15-feat-x-y.1"}, at); changed {
+			t.Error("an ask is marked numbered by a build of another head")
+		}
+		if _, changed := u.WithNumbered("o", "", map[string]string{"go": "1"}, at); changed {
+			t.Error("an untag is marked numbered")
+		}
+		n, changed := a.WithNumbered("mesh-sdk", "abc", map[string]string{"go": "0.1.15-feat-x-y.1"}, at.Add(time.Minute))
+		if !changed || n.State != TagNumbered || n.Versions["mesh-sdk/go"] != "0.1.15-feat-x-y.1" || n.Unanswered() || !strings.Contains(n.Answer, "mesh-sdk/go 0.1.15-feat-x-y.1") {
+			t.Errorf("WithNumbered = %+v, %v", n, changed)
+		}
+		if _, again := n.WithNumbered("mesh-sdk", "abc", map[string]string{"go": "0.1.15-feat-x-y.1"}, at); again {
+			t.Error("the same versions mark the ask again")
+		}
+		said := NarrateUnanswered(a, []FloorWait{{Name: "mesh-sdk/go", Silent: []string{"bridge_tags"}}}, at.Add(6*time.Minute))
+		for _, w := range []string{"novox/mesh-sdk feat/X y", "after 6m0s", "not claimed yet", "mesh-sdk/go waits on its floor (bridge_tags silent)"} {
+			if !strings.Contains(said, w) {
+				t.Errorf("NarrateUnanswered = %q, want %q in it", said, w)
+			}
+		}
 		if k, v, err := TagWantedAsk(TagWanted{Repository: "o/r", Branch: "feat/x", Head: "abc"}, "mesh-delivery-checker", at); err != nil || k != TagAskKey("o/r", "feat/x") || !strings.Contains(string(v), `"by":"mesh-delivery-checker"`) {
 			t.Errorf("TagWantedAsk = %q %s %v", k, v, err)
 		}

modules/mesh-delivery-builder/manifest_test.go

@@ -22,6 +22,7 @@ import (
 type state struct {
 	Name    string `json:"name"`
 	History int    `json:"history,omitempty"`
+	KeyTTL  bool   `json:"key-ttl,omitempty"`
 }
 
 type artifact struct {
@@ -93,7 +94,7 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
 	if m.Module != "mesh-delivery-builder" {
 		t.Errorf("module = %q", m.Module)
 	}
-	if want := []state{{Name: "builds", History: 5}, {Name: "engine-reads", History: 1}, {Name: "findings", History: 5}, {Name: "floor-waits", History: 1}, {Name: "hearing", History: 1}, {Name: "instances", History: 1}, {Name: "old-form", History: 1}, {Name: "queues", History: 1}, {Name: "reasks", History: 1}, {Name: "tag-asks", History: 1}, {Name: "versions", History: 1}}; !reflect.DeepEqual(m.State, want) { // delivery.BuilderBuildsHistory
+	if want := []state{{Name: "builds", History: 5}, {Name: "engine-reads", History: 1}, {Name: "findings", History: 5}, {Name: "floor-waits", History: 1}, {Name: "hearing", History: 1}, {Name: "instances", History: 1}, {Name: "old-form", History: 1}, {Name: "queues", History: 1}, {Name: "reasks", History: 1}, {Name: "tag-ask-bounds", History: 1, KeyTTL: true}, {Name: "tag-asks", History: 1}, {Name: "versions", History: 1}}; !reflect.DeepEqual(m.State, want) { // delivery.BuilderBuildsHistory
 		t.Errorf("state = %+v, want %+v", m.State, want)
 	}
 	// 1887/A4c: the builder chooses no node, so it reads neither presence nor have, invokes no build module's tool and

modules/mesh-delivery-builder/module.json

@@ -70,6 +70,7 @@
     { "name": "old-form", "history": 1 },
     { "name": "queues", "history": 1 },
     { "name": "reasks", "history": 1 },
+    { "name": "tag-ask-bounds", "history": 1, "key-ttl": true },
     { "name": "tag-asks", "history": 1 },
     { "name": "versions", "history": 1 }
   ],