BL12 / G-16 §588: the builder publishes reserved versions again, on an event by stream sequence and by builder_publish_again

The reservation keeps the fingerprint of the catalog entry it was made from (versions.Source, A-76). The state sections
publish-queue (max-mib 4) and acted (max-mib 1); a consumer acting once per message of the node-engine seat's running and
manifest-reads-changed (why returned) by its stream sequence, the running of every subject read at the start with
lastPerSubject; one queue per holder sent one at a time as the registry seat's publish with the digest and no parts, the
seat and the package name read from the entry (A-73, A-76); builder_publish_queue and the new builder_publish_again, a
hand act. The old form goes: build.NewPublishAgain / ModePublishAgain, versions.Acts.PublishAgain and the
Again* decision, and the integration step that refused it.

Builds: design 47 G §588, §369, §695, A-73, A-76 (G-16, G-18, G-20)

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

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

@@ -54,17 +54,9 @@ func Tools(r build.StuckReader, askAgain func(key string, value []byte) error, q
 		if a.Refused {
 			return nil, errors.New(a.Answer)
 		}
-		return map[string]any{"key": key, "name": a.Name, "do": a.Do, "version": a.Version, "answer": a.Answer}, nil
+		return map[string]any{"key": key, "name": a.Name, "answer": a.Answer}, nil
 	}
-	nameFP := map[string]any{"type": "object", "required": []string{"name", "fp", "why"}, "properties": map[string]any{
-		"name": map[string]any{"type": "string", "description": "the numbered name, <module>/<artifact>, as builder_versions lists it"},
-		"fp":   map[string]any{"type": "string", "description": "the fingerprint, src2:<hex>"},
-		"why":  map[string]any{"type": "string", "description": "why, in plain words"}}}
 	return []stdio.Tool{
-		{Name: versions.PublishAgainTool, Description: versions.PublishAgainDescription, Input: nameFP,
-			Run: func(args map[string]any) (any, error) {
-				return handAct(build.NewPublishAgain(str(args, "name"), str(args, "fp"), str(args, "why"), stdio.CallerOf(args), now()))
-			}},
 		{Name: versions.FloorAgainTool, Description: versions.FloorAgainDescription,
 			Input: map[string]any{"type": "object", "required": []string{"name", "why"}, "properties": map[string]any{
 				"name": map[string]any{"type": "string", "description": "the numbered name, <module>/<artifact>"},
@@ -372,6 +364,8 @@ func main() {
 	}}
 	tools := Tools(r, put, askQueue(stdio.State(build.QueuesState), 25*time.Second), buildAgain, stdio.Log(builds.BuilderLog).Since, time.Now, vr, fr, ofr)
 	tools = append(tools, reservedTools(rr, buildAgain, time.Now, meshhandactsclient.Recorder{Ask: stdio.Ask})...)
+	qs, as := stdio.State(versions.PublishQueueState), stdio.State(versions.ActedState)
+	tools = append(tools, republishTools(readerOf(qs), readerOf(as), buildAgain, time.Now, meshhandactsclient.Recorder{Ask: stdio.Ask})...)
 	ws := stdio.State(build.WaitsState)
 	wr := versions.Reader{Keys: ws.Keys, Get: func(key string) (json.RawMessage, bool, error) {
 		e, err := ws.Get(key)

modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/republish.go

@@ -0,0 +1,65 @@
+package main
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient"
+	stdio "git.novox.be/novox/mesh-sdk/go"
+
+	"mesh-delivery-builder/internal/build"
+	"mesh-delivery-builder/internal/versions"
+)
+
+// readerOf is a kept state as the versions package's reader.
+func readerOf(s stdio.KeptState) versions.Reader {
+	return versions.Reader{Keys: s.Keys, Get: func(key string) (json.RawMessage, bool, error) {
+		e, err := s.Get(key)
+		if err != nil || e == nil {
+			return nil, false, err
+		}
+		return e.Value, true, nil
+	}}
+}
+
+// republishTools are builder_publish_again and builder_publish_queue (design 47 G §588). The publish again is a hand act:
+// it is recorded in the hand-act log first, then written to reasks for the builder, which owns the queue, to answer
+// with the versions it queued.
+func republishTools(queue, acted versions.Reader, ask queueAsker, now func() time.Time, acts meshhandactsclient.Recorder) []stdio.Tool {
+	str := func(args map[string]any, k string) string { s, _ := args[k].(string); return strings.TrimSpace(s) }
+	return []stdio.Tool{
+		{Name: versions.PublishAgainTool, Description: versions.PublishAgainDescription,
+			Input: map[string]any{"type": "object", "required": []string{"why"}, "properties": map[string]any{
+				"module":  map[string]any{"type": "string", "description": "only this module's reserved versions"},
+				"version": map[string]any{"type": "string", "description": "with module: only this reserved version, MAJOR.MINOR.PATCH"},
+				"why":     map[string]any{"type": "string", "description": "why, in plain words"}}},
+			Run: func(args map[string]any) (any, error) {
+				key, value, err := build.NewRepublish(str(args, "module"), str(args, "version"), str(args, "why"), stdio.CallerOf(args), now())
+				if err != nil {
+					return nil, err
+				}
+				if _, err := acts.Record(meshhandactsclient.Request{What: fmt.Sprintf("published reserved versions again by hand (module %q, version %q)", str(args, "module"), str(args, "version")),
+					Why: str(args, "why"), Cause: versions.PublishAgainTool, Verb: "publish-again"}); err != nil {
+					return nil, fmt.Errorf("the hand act was not recorded through mesh-hand-acts, so nothing was queued: %w", err)
+				}
+				raw, err := ask(key, value)
+				if err != nil {
+					return nil, err
+				}
+				a, err := build.DecodeReAsk(raw)
+				if err != nil {
+					return nil, err
+				}
+				if a.Refused {
+					return nil, errors.New(a.Answer)
+				}
+				return map[string]any{"key": key, "queued": a.Queued, "answer": a.Answer}, nil
+			}},
+		{Name: versions.PublishQueueTool, Description: versions.PublishQueueDescription,
+			Input: map[string]any{"type": "object", "properties": map[string]any{}},
+			Run:   func(map[string]any) (any, error) { return versions.AnswerQueue(queue, acted) }},
+	}
+}

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

@@ -582,18 +582,10 @@ func TestBuilderMakesADoneKeyWithTheDigests(t *testing.T) { // 1607/AC7
 	if got := reg.numberedOf("netcheck/tools-go"); len(got) != 0 {
 		t.Errorf("version.numbered of netcheck/tools-go said %v, want none", got)
 	}
-	// DLV-B10: builder_publish_again of a published, held build is refused: nothing says a version numbered (§349).
 	for _, a := range []string{"tools-go", "tools-typescript"} {
 		reg.published(t, "netcheck/"+a, "0.1.5", digestOf(a))
 	}
 	reg.awaitOf(t, "netcheck/tools-go", func(es []versions.Entry) bool { return len(es) == 1 && es[0].State == versions.Published })
-	key, again, err := build.NewPublishAgain("netcheck/tools-go", fp, "the holder lost it", "me@shanks", time.Now())
-	if err != nil {
-		t.Fatal(err)
-	}
-	if a := reg.answered(t, key, again); !a.Refused {
-		t.Errorf("builder_publish_again = %+v, want a refusal", a)
-	}
 	// A numbered version a holder will refuse below, written to the state as the builder numbers one.
 	refusable, _ := json.Marshal(versions.Entry{Name: "netcheck/tools-go", Kind: versions.KindBundle, Series: "0.1", Version: "0.1.6", Module: "netcheck",
 		Fingerprint: "src2:" + strings.Repeat("6", 64), Seat: ociartifactregistry.Seat, State: versions.Numbered, At: time.Now().UTC()})

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

@@ -219,17 +219,22 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	// 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.
-	acts := versions.Acts{Numberer: numberer, DropReserved: reserved.dropped, Done: doneVia(b), Reconsider: b.Reconsider,
-		BuildAgain: func(ctx context.Context, module, fp, why, by string) (string, bool) {
-			r := b.BuildAgain(ctx, module, fp, why, by, buildAgainBound)
-			if r.Refused != "" {
-				return r.Refused, true
-			}
-			if r.Failed != "" {
-				return "built again, and it failed before its asks were put: " + r.Failed, false
-			}
-			return fmt.Sprintf("built again as revision %d", r.Revision), false
-		}}
+	acts := versions.Acts{Numberer: numberer, DropReserved: reserved.dropped, Reconsider: b.Reconsider}
+	// A reserved version left unpublished is published again, by the builder on an event or by a person's tool (design 47 G
+	// §588): its states publish-queue and acted, bound once the bus holds them.
+	republishMax, err := republishMaxOf(getenv("MESH_BUILDER_REPUBLISH_MAX_FILE"))
+	if err != nil {
+		return err
+	}
+	queueKV := process.AwaitBucket(ctx, js, conn, versions.PublishQueueBucket, "no version is published again on an event or by builder_publish_again", logger.Printf)
+	actedKV := process.AwaitBucket(ctx, js, conn, versions.ActedBucket, "no version is published again on an event or by builder_publish_again", logger.Printf)
+	republish := &versions.Republisher{Reserved: reserved.r, Queue: awaitedQueue{awaitedReserved{awaitedVersions{b: queueKV, bucket: versions.PublishQueueBucket}}},
+		Acted: awaitedReserved{awaitedVersions{b: actedKV, bucket: versions.ActedBucket}}, Entry: entryVia(calls.Call), Send: sendVia(calls.Call),
+		Unreleased: reserved.unreleasedWords, Max: republishMax, Logf: logger.Printf}
+	acts.Republish = republish.Republish
+	again := &republishing{rp: republish}
+	queueKV.Ready(again.bound(ctx, js, logger.Printf))
+	actedKV.Ready(again.bound(ctx, js, logger.Printf))
 	// A candidate's preview is merged onto the trunk head the indexer previewed it onto: source.main at the candidate's base
 	// 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
@@ -371,6 +376,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	// And the four registry seats' published, refused and removed (DLV-B4), which move a numbered version.
 	seats := seatHearer{n: numberer, b: b, logf: logger.Printf, pub: pubs, res: reserved}
 	events := func(m jetstream.Msg) {
+		if again.heard(ctx, m, logger.Printf) { // the publish again's own subjects, left for the bus to give again when its state cannot take them
+			return
+		}
 		if reads.heard(m, logger.Printf) || heardTools.Reported(ctx, m) || heardTools.Running(ctx, m) || floor.heard(ctx, m) || reserved.heard(ctx, m) || seats.heard(ctx, m) { // reads: the node-engine seat's manifest-reads-changed (ADR 0618 §4)
 			return
 		}

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

@@ -58,7 +58,7 @@ func reAsked(ctx context.Context, b askAgainer, acts versions.Acts, kv jetstream
 		released(ctx, b, kv, e, r, logf)
 		return
 	}
-	if r.Mode == build.ModePublishAgain || r.Mode == build.ModeFloorAgain || r.Mode == build.ModeOldFormDrop || r.Mode == build.ModeReservedDrop {
+	if r.Mode == build.ModeRepublish || r.Mode == build.ModeFloorAgain || r.Mode == build.ModeOldFormDrop || r.Mode == build.ModeReservedDrop {
 		// A version's hand act (design 51 *Tools of the builder*) may build again, so it is answered beside the watch.
 		go versionActed(ctx, acts, kv, e, r, logf)
 		return
@@ -164,10 +164,15 @@ func appendLog(ctx context.Context, js jetstream.JetStream, cred process.Account
 // versionActed answers a reasks value of builder_publish_again, builder_floor_again or builder_old_form's drop, a hand act said in the log.
 func versionActed(ctx context.Context, acts versions.Acts, kv jetstream.KeyValue, e jetstream.KeyValueEntry, r build.ReAsk, logf func(string, ...any)) {
 	var err error
-	if r.Mode == build.ModePublishAgain {
-		var ag versions.Again
-		ag, r.Answer, err = acts.PublishAgain(ctx, r.Name, r.FP, r.Why, r.By, r.At)
-		r.Do, r.Version = ag.Do, ag.Entry.Version
+	if r.Mode == build.ModeRepublish {
+		if acts.Republish == nil {
+			err = errors.New("this builder publishes nothing again")
+		} else {
+			r.Queued, err = acts.Republish(ctx, r.Module, r.Version)
+			if err == nil {
+				r.Answer = "queued " + fmt.Sprint(len(r.Queued)) + " versions to publish again: " + strings.Join(r.Queued, ", ") + "; builder_publish_queue shows them"
+			}
+		}
 	} else if r.Mode == build.ModeReservedDrop {
 		if acts.DropReserved == nil {
 			err = errors.New("this builder keeps no reservations")

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

@@ -0,0 +1,146 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io/fs"
+	"os"
+	"strconv"
+	"strings"
+	"sync/atomic"
+
+	"github.com/nats-io/nats.go/jetstream"
+
+	"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
+	"git.novox.be/novox/mesh-delivery/contracts/go/seats/gopackageregistry"
+	"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
+
+	"mesh-delivery-builder/internal/versions"
+)
+
+// The builder's publish again (design 47 G §588, A-73, A-76): the wiring of versions.Republisher to the bus.
+
+// awaitedQueue is the state publish-queue as versions.QueueStore: bound once the bus holds it; its bytes are the cap's.
+type awaitedQueue struct{ awaitedReserved }
+
+func (s awaitedQueue) Bytes(ctx context.Context) (uint64, error) {
+	kv, err := s.kv()
+	if err != nil {
+		return 0, err
+	}
+	st, err := kv.Status(ctx)
+	if err != nil {
+		return 0, err
+	}
+	return st.Bytes(), nil
+}
+
+// entryVia reads a catalog graph entry by its fingerprint through the seat mesh-catalog's verb entry.
+func entryVia(call toolCaller) func(ctx context.Context, fingerprint string) (meshcatalog.CatalogEntry, error) {
+	return func(_ context.Context, fingerprint string) (meshcatalog.CatalogEntry, error) {
+		raw, err := call("seat:"+meshcatalog.Seat+"."+meshcatalog.VerbEntry, meshcatalog.EntryInput{Fingerprint: fingerprint})
+		if err != nil {
+			return meshcatalog.CatalogEntry{}, err
+		}
+		var a meshcatalog.EntryAnswer
+		if err := json.Unmarshal(raw, &a); err != nil {
+			return meshcatalog.CatalogEntry{}, fmt.Errorf("the catalog answered what is no entry: %w", err)
+		}
+		return a.Entry, nil
+	}
+}
+
+// sendVia sends a publish to the registry seat with the digest and no parts (A-73): the holder takes the bytes from the
+// artifact store, addressed by digest. The four seats' publish take one wire form; an answer that lacks parts, or a fault of
+// the call, is the holder's refusal, said in its words (not stored when the holder lacks the bytes).
+func sendVia(call toolCaller) func(ctx context.Context, p versions.Publish) error {
+	return func(_ context.Context, p versions.Publish) error {
+		raw, err := call("seat:"+p.Seat+"."+versions.VerbPublish, publishBody(p))
+		if err != nil {
+			return fmt.Errorf("%s's publish of %s %s: %w", p.Seat, p.Name, p.Version, err)
+		}
+		var a gopackageregistry.PublishAnswer
+		if err := json.Unmarshal(raw, &a); err != nil {
+			return fmt.Errorf("%s answered what is no publish answer: %w", p.Seat, err)
+		}
+		if len(a.Lacks) > 0 {
+			return fmt.Errorf("%s%s holds %d of the parts of %s %s", versions.NotStoredWords, p.Seat, len(a.Lacks), p.Name, p.Version)
+		}
+		return nil
+	}
+}
+
+// publishBody is the seat's publish input for p: the digest, its kind and name and version, no parts.
+func publishBody(p versions.Publish) gopackageregistry.PublishInput {
+	return gopackageregistry.PublishInput{Commit: p.Commit, Digest: p.Digest, Fingerprint: p.Fingerprint, Kind: p.Kind, Name: p.Name, Transfer: p.Transfer, Version: p.Version}
+}
+
+// republishMaxOf reads the setting republish-max-versions from the file the manifest declares (design 47 G §633, A-71's
+// form): no file or an empty one is the default; a content that is no positive number is an error naming the file.
+func republishMaxOf(path string) (int, error) {
+	if path == "" {
+		return versions.DefaultRepublishMax, nil
+	}
+	raw, err := os.ReadFile(path)
+	if errors.Is(err, fs.ErrNotExist) {
+		return versions.DefaultRepublishMax, nil
+	}
+	if err != nil {
+		return 0, err
+	}
+	v := strings.TrimSpace(string(raw))
+	if v == "" {
+		return versions.DefaultRepublishMax, nil
+	}
+	n, err := strconv.Atoi(v)
+	if err != nil || n < 1 {
+		return 0, fmt.Errorf("%s holds %q, which is not a number of versions", path, v)
+	}
+	return n, nil
+}
+
+// republishing starts the publish again once both its states are bound: the queue is taken up again and the last running
+// of every node's module is read (§556), acted on where its sequence is newer than the kept one.
+type republishing struct {
+	rp    *versions.Republisher
+	ready atomic.Int32
+}
+
+func (g *republishing) bound(ctx context.Context, js jetstream.JetStream, logf func(string, ...any)) func(jetstream.KeyValue) {
+	return func(jetstream.KeyValue) {
+		if g.ready.Add(1) != 2 {
+			return
+		}
+		if err := g.rp.Resume(ctx); err != nil {
+			logf("the queued publishes are not taken up again: %v", err)
+		}
+		err := lastPerSubject(ctx, js, nodeengine.StreamRunning, nodeengine.SubjectRunningAll, func(subject string, data []byte, seq uint64) {
+			if _, err := g.rp.Heard(ctx, subject, data, seq); err != nil {
+				logf("the running %s is not acted on at the start: %v", subject, err)
+			}
+		})
+		if err != nil {
+			logf("the running of the node-engine seat is not read for the publish again at the start: %v", err)
+		}
+	}
+}
+
+// heard hands a message of the events consumer to the publish again: one of its subjects whose state cannot take it is left
+// for the bus to give again (Nak); every message goes on to the other parts.
+func (g *republishing) heard(ctx context.Context, m jetstream.Msg, logf func(string, ...any)) (stop bool) {
+	meta, err := m.Metadata()
+	if err != nil {
+		return false
+	}
+	handled, err := g.rp.Heard(ctx, m.Subject(), m.Data(), meta.Sequence.Stream)
+	if handled && err != nil {
+		logf("%v; heard again", err)
+		if nakErr := m.Nak(); nakErr != nil {
+			logf("leaving %s for the bus to give again: %v", m.Subject(), nakErr)
+		}
+		return true
+	}
+	return false
+}

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

@@ -0,0 +1,79 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"git.novox.be/novox/mesh-delivery/contracts/go/seats/containerimageregistry"
+	"git.novox.be/novox/mesh-delivery/contracts/go/seats/gopackageregistry"
+	"git.novox.be/novox/mesh-delivery/contracts/go/seats/npmpackageregistry"
+	"git.novox.be/novox/mesh-delivery/contracts/go/seats/ociartifactregistry"
+
+	"mesh-delivery-builder/internal/build"
+	"mesh-delivery-builder/internal/versions"
+)
+
+// TestRepublishSend: design 47 G §588, A-73: a publish again is the registry seat's publish with the digest and no parts; the
+// four seats take one wire form, a call's fault or an answer that lacks parts is the holder's refusal in its words, and its
+// setting is read from the file the manifest declares.
+func TestRepublishSend(t *testing.T) {
+	p := versions.Publish{Seat: versions.SeatArtifacts, Kind: "bundle", Name: "m/prog", Version: "0.1.3", Digest: "sha256:aa", Fingerprint: "src2:m", Commit: "c", Transfer: "t"}
+	want, _ := json.Marshal(publishBody(p))
+	if strings.Contains(string(want), "parts") {
+		t.Errorf("the publish names parts: %s", want)
+	}
+	for name, v := range map[string]any{"container image": containerimageregistry.PublishInput{Digest: "sha256:aa"}, "oci": ociartifactregistry.PublishInput{Digest: "sha256:aa"},
+		"go": gopackageregistry.PublishInput{Digest: "sha256:aa"}, "npm": npmpackageregistry.PublishInput{Digest: "sha256:aa"}} {
+		var ref, got map[string]any
+		a, _ := json.Marshal(gopackageregistry.PublishInput{Digest: "sha256:aa"})
+		b, _ := json.Marshal(v)
+		_ = json.Unmarshal(a, &ref)
+		_ = json.Unmarshal(b, &got)
+		if len(ref) != len(got) {
+			t.Errorf("the %s seat's publish input is %s, the go seat's %s: one wire form is sent to all four", name, b, a)
+		}
+	}
+	var asked string
+	for name, c := range map[string]struct {
+		answer string
+		err    error
+		want   string
+	}{
+		"held":     {`{"held":true,"transfer":"t"}`, nil, ""},
+		"lacks":    {`{"lacks":["sha256:bb"],"transfer":"t"}`, nil, versions.NotStoredWords},
+		"a fault":  {"", errors.New("no responders"), "no responders"},
+		"no sense": {`[`, nil, "no publish answer"},
+	} {
+		send := sendVia(func(key string, body any) (json.RawMessage, error) {
+			asked = key
+			return json.RawMessage(c.answer), c.err
+		})
+		err := send(context.Background(), p)
+		if asked != "seat:oci-artifact-registry.publish" || (c.want == "") != (err == nil) || (err != nil && !strings.Contains(err.Error(), c.want)) {
+			t.Errorf("%s: asked %q, error %v, want %q", name, asked, err, c.want)
+		}
+	}
+	dir := t.TempDir()
+	for content, want := range map[string]int{"": 50, "7": 7} {
+		path := filepath.Join(dir, "max")
+		_ = os.WriteFile(path, []byte(content), 0o600)
+		if got, err := republishMaxOf(path); err != nil || got != want {
+			t.Errorf("republishMaxOf(%q) = %d, %v, want %d", content, got, err, want)
+		}
+	}
+	_ = os.WriteFile(filepath.Join(dir, "bad"), []byte("0"), 0o600)
+	if _, err := republishMaxOf(filepath.Join(dir, "bad")); err == nil {
+		t.Error("a setting of 0 is accepted")
+	}
+	if _, _, err := build.NewRepublish("", "0.1.3", "why", "me", versionsNow()); err == nil {
+		t.Error("a version with no module is accepted")
+	}
+}
+
+func versionsNow() time.Time { return time.Unix(1, 0) }

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

@@ -125,6 +125,11 @@ func (rv *reservations) unreleased(ctx context.Context, v versions.ReservedVersi
 		needs:       "builder_reserved shows the reservation; builder_publish_again publishes the missing artifacts under it; builder_reserved_drop drops it."})
 }
 
+// unreleasedWords raises version-unreleased about a version with the words of a publish again that ended badly.
+func (rv *reservations) unreleasedWords(ctx context.Context, v versions.ReservedVersion, why string) {
+	rv.unreleased(ctx, v, why)
+}
+
 // ended is told each change of the state bounds: a reservation's bound whose time ended raises version-unreleased while
 // the reservation stands built.
 func (rv *reservations) ended(ctx context.Context, e jetstream.KeyValueEntry) bool {

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

@@ -254,8 +254,8 @@ func (h *toolchainsHeard) agree(ctx context.Context) {
 }
 
 // lastPerSubject reads the last message of every subject of a stream kept last per subject, through an ordered consumer
-// made for that read (design 47 G §556), and hands each to each; it answers when the last of them was handed.
-func lastPerSubject(ctx context.Context, js jetstream.JetStream, stream, filter string, each func(subject string, data []byte)) error {
+// made for that read (design 47 G §556), and hands each to each with its stream sequence; it answers when the last of them was handed.
+func lastPerSubject(ctx context.Context, js jetstream.JetStream, stream, filter string, each func(subject string, data []byte, sequence uint64)) error {
 	cons, err := js.OrderedConsumer(ctx, stream, jetstream.OrderedConsumerConfig{FilterSubjects: []string{filter}, DeliverPolicy: jetstream.DeliverLastPerSubjectPolicy})
 	if err != nil {
 		return fmt.Errorf("reading %s: %w", stream, err)
@@ -278,14 +278,18 @@ func lastPerSubject(ctx context.Context, js jetstream.JetStream, stream, filter
 		if err != nil {
 			return fmt.Errorf("reading %s: %w", stream, err)
 		}
-		each(m.Subject(), m.Data())
+		seq := uint64(0)
+		if meta, err := m.Metadata(); err == nil {
+			seq = meta.Sequence.Stream
+		}
+		each(m.Subject(), m.Data(), seq)
 	}
 	return nil
 }
 
 // readRunning is the start's read of every node's module's last running (§556), before the consumer's.
 func (h *toolchainsHeard) readRunning(ctx context.Context, js jetstream.JetStream) {
-	if err := lastPerSubject(ctx, js, nodeengine.StreamRunning, nodeengine.SubjectRunningAll, func(_ string, data []byte) { h.running(ctx, data) }); err != nil {
+	if err := lastPerSubject(ctx, js, nodeengine.StreamRunning, nodeengine.SubjectRunningAll, func(_ string, data []byte, _ uint64) { h.running(ctx, data) }); err != nil {
 		h.logf("the running of the node-engine seat is not read at the start, so a build module stopped while this builder was away is forgotten at its next running: %v", err)
 	}
 }

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

@@ -13,7 +13,6 @@ import (
 
 	"git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
 	"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
-	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
 	"git.novox.be/novox/mesh-sdk/go/process"
 
 	"mesh-delivery-builder/internal/build"
@@ -306,22 +305,3 @@ func (f *floorWake) wake(ctx context.Context, kind, seat, module string) {
 func conditionModules(id string) []string {
 	return strings.FieldsFunc(id, func(r rune) bool { return r == '/' || r == '.' || r == ':' })
 }
-
-// doneVia says whether a build key is done, read from the builder's own builds state: the build's file is held.
-func doneVia(b *build.Builder) func(ctx context.Context, module, fp string) (bool, error) {
-	return func(ctx context.Context, module, fp string) (bool, error) {
-		key, err := builds.BuilderBuildKey(module, fp)
-		if err != nil {
-			return false, err
-		}
-		raw, _, found, err := b.Claimer.Store.Get(ctx, key)
-		if err != nil || !found {
-			return false, err
-		}
-		cur, err := builds.DecodeBuilderBuild(raw)
-		if err != nil {
-			return false, err
-		}
-		return cur.State == builds.BuildDone, nil
-	}
-}

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

@@ -177,7 +177,7 @@ func (b *Builder) reserve(ctx context.Context, module string, m meshcatalog.Modu
 	for _, t := range plan.Tasks {
 		refs = append(refs, module+":"+t.Artifact)
 	}
-	src := versions.Source{Repository: m.Repository, Commit: m.Commit}
+	src := versions.Source{Repository: m.Repository, Commit: m.Commit, Fingerprint: m.Fp2}
 	for _, t := range plan.Tasks {
 		src.GoPackage = src.GoPackage || (t.Kind == catalogue.ArtifactPackage && t.Language == "go")
 	}

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

@@ -57,35 +57,32 @@ type ReAsk struct {
 	Failed   string            `json:"failed,omitempty"`
 	Revision uint64            `json:"revision,omitempty"`
 	Asks     map[string]string `json:"asks,omitempty"`
-	// Name is the numbered name (`<module>/<artifact>`) of a ModePublishAgain or ModeFloorAgain value; Do and Version the
-	// builder's answer: what it did (versions.Again*) and the version published.
-	Name    string `json:"name,omitempty"`
-	Do      string `json:"do,omitempty"`
-	Version string `json:"version,omitempty"`
+	// Name is the numbered name (`<module>/<artifact>`) of a ModeFloorAgain value; Version is a ModeRepublish value's
+	// request and Queued the builder's answer to it: the versions it queued, `<module> <version>`.
+	Name    string   `json:"name,omitempty"`
+	Version string   `json:"version,omitempty"`
+	Queued  []string `json:"queued,omitempty"`
 }
 
-// ModePublishAgain is the Mode of a reasks value of builder_publish_again (design 51 *Tools of the builder*).
-const ModePublishAgain = "publish-again"
+// ModeRepublish is the Mode of a reasks value of builder_publish_again (design 47 G §588): Module and Version, each
+// optional, name what to publish again; the builder answers with the versions it queued (Queued).
+const ModeRepublish = "republish"
 
 // ModeFloorAgain is the Mode of a reasks value of builder_floor_again (design 51 *The floor*).
 const ModeFloorAgain = "floor-again"
 
-// NewPublishAgain is the value builder_publish_again writes for a name's fingerprint: the key of the name's module and
-// fingerprint, and the value.
-func NewPublishAgain(name, fp, why, by string, at time.Time) (string, []byte, error) {
-	module, artifact, ok := strings.Cut(strings.TrimSpace(name), "/")
-	if !ok || module == "" || artifact == "" {
-		return "", nil, fmt.Errorf("name %q is not a numbered name, <module>/<artifact>, as builder_versions lists it", name)
-	}
-	key, err := builds.BuilderBuildKey(module, fp)
-	if err != nil {
-		return "", nil, err
+// NewRepublish is the value builder_publish_again writes (design 47 G §588): its key, `republish.<unix nanoseconds>`
+// (a call is one value, never a build key), and the value. Module and version are each optional.
+func NewRepublish(module, version, why, by string, at time.Time) (string, []byte, error) {
+	module, version = strings.TrimSpace(module), strings.TrimSpace(version)
+	if version != "" && module == "" {
+		return "", nil, errors.New("a version is published again under its module: give module with version")
 	}
 	if strings.TrimSpace(why) == "" {
 		return "", nil, errors.New("say why it is published again (why)")
 	}
-	data, err := json.Marshal(ReAsk{V: 1, Why: strings.TrimSpace(why), By: by, At: at.UTC(), Mode: ModePublishAgain, Module: module, FP: fp, Name: strings.TrimSpace(name)})
-	return key, data, err
+	data, err := json.Marshal(ReAsk{V: 1, Why: strings.TrimSpace(why), By: by, At: at.UTC(), Mode: ModeRepublish, Module: module, Version: version})
+	return "republish." + strconv.FormatInt(at.UnixNano(), 10), data, err
 }
 
 // NewFloorAgain is the value builder_floor_again writes for a name: the key `floor.<name escaped>`, never a build key,

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

@@ -2,63 +2,13 @@ package versions
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"strings"
 	"time"
 )
 
-// What builder_publish_again does with one fingerprint of a name (design 51 *Tools of the builder*), after a holder failed.
-const (
-	// AgainPublish is a held build whose recorded version is published again: refused since design 47 G §349 (no version.numbered is said).
-	AgainPublish = "publish"
-	// AgainBuild builds the fingerprint again under the same number: the build's file is gone and no published was said
-	// for the number, so the new digest replaces the recorded one (Numberer.Built).
-	AgainBuild = "build"
-	// AgainRenumber takes a new number: a published was said for the number, so other bytes never go under it.
-	AgainRenumber = "renumber"
-)
-
-// Again is what builder_publish_again does, with the entry it acts on and why.
-type Again struct {
-	Do    string `json:"do"`
-	Entry Entry  `json:"entry"`
-	Why   string `json:"why"`
-}
-
-// PublishAgain decides builder_publish_again for fingerprint among a name's entries; built says whether the builder still
-// holds the build's file (its done key). It refuses a fingerprint not numbered, and a version refused for other bytes
-// than the ones recorded under it.
-func PublishAgain(entries []Entry, fingerprint string, built bool) (Again, error) {
-	fingerprint = strings.TrimSpace(fingerprint)
-	if fingerprint == "" {
-		return Again{}, errors.New("fp is required: the fingerprint whose version is published again")
-	}
-	for _, e := range latestFirst(entries) {
-		if e.Tag != "" || e.Fingerprint != fingerprint {
-			continue
-		}
-		switch {
-		case e.State == Refused && e.RefusedDigest != "" && e.Digest != "" && e.RefusedDigest != e.Digest:
-			return Again{}, fmt.Errorf("%s@%s was refused for other bytes (%s, recorded %s): it is not published again; a new build takes a new number",
-				e.Name, e.Version, e.RefusedDigest, e.Digest)
-		case e.State == Published || e.State == Removed:
-			return Again{Do: AgainRenumber, Entry: e, Why: fmt.Sprintf("%s@%s was %s, so the fingerprint takes a new number", e.Name, e.Version, e.State)}, nil
-		case !built:
-			return Again{Do: AgainBuild, Entry: e, Why: fmt.Sprintf("the build's file is gone and no published was said for %s@%s: built again under the same number", e.Name, e.Version)}, nil
-		}
-		return Again{Do: AgainPublish, Entry: e, Why: fmt.Sprintf("%s@%s (%s) is held", e.Name, e.Version, e.State)}, nil
-	}
-	return Again{}, fmt.Errorf("fingerprint %s is not numbered for this name: builder_versions lists what is", fingerprint)
-}
-
 // The tools' descriptions.
 const (
-	PublishAgainDescription = "Publish one fingerprint's version again after its holder failed (design 51): with the build's file held, the " +
-		"nothing says a version numbered since design 47 G §349, so a held build is refused; with the file gone, it builds the fingerprint " +
-		"again under the same number, the new digest replacing the recorded one, since no published was said for it; after a published (or " +
-		"a removal) it takes a new number. It refuses a fingerprint not numbered and a version the holder refused for other bytes. A hand " +
-		"act, recorded with its caller and why in the builder's log."
 	FloorAgainDescription = "Ask the floor of one name again now, by hand (design 51 *The floor*): the seat's versions, the forge bridge's " +
 		"bridge_tags and the indexer's indexer_requirements, as the builder asks them by itself when a silent source says something. " +
 		"Answers which sources were silent and whether the floor is now known. A name the builder's versions already hold needs no floor: " +
@@ -68,56 +18,16 @@ const (
 // Acts are the builder's hand acts on its versions, answered by its program (the tools bundle writes them to reasks).
 type Acts struct {
 	Numberer *Numberer
-	// Done says whether the build key of module and fingerprint is done: the build's file is held.
-	Done func(ctx context.Context, module, fp string) (bool, error)
-	// BuildAgain builds a failed key again (builder_build_again's path): its answer, or why it refused.
-	BuildAgain func(ctx context.Context, module, fp, why, by string) (answer string, refused bool)
 	// Reconsider looks at every module again, so a claim asks a name's floor.
 	Reconsider func(ctx context.Context)
+	// Republish queues the reserved versions builder_publish_again names (design 47 G §588): the versions it queued.
+	Republish func(ctx context.Context, module, version string) ([]string, error)
 	// DropOldForm drops a retired node from the old form's count by hand (builder_old_form): its answer, or why not.
 	DropOldForm func(ctx context.Context, node, by, why string, at time.Time) (string, error)
 	// DropReserved drops a module's reservation of one MAJOR.MINOR by hand (builder_reserved_drop): its answer, or why not.
 	DropReserved func(ctx context.Context, module, series, by, why string, at time.Time) (string, error)
 }
 
-// PublishAgain is builder_publish_again of name's fingerprint: what it did, the version, and its answer; an error refuses.
-func (a Acts) PublishAgain(ctx context.Context, name, fp, why, by string, at time.Time) (Again, string, error) {
-	module, _, _ := strings.Cut(name, "/")
-	entries, err := a.Numberer.Versions.Of(ctx, name)
-	if err != nil {
-		return Again{}, "", err
-	}
-	done, err := a.Done(ctx, module, fp)
-	if err != nil {
-		return Again{}, "", fmt.Errorf("the build key of %s at %s cannot be read: %w", module, fp, err)
-	}
-	ag, err := PublishAgain(entries, fp, done)
-	if err != nil {
-		return Again{}, "", err
-	}
-	if ag.Do == AgainPublish || (ag.Do == AgainRenumber && done) {
-		return Again{}, "", errors.New("nothing says a version numbered since design 47 G §349: the version is published under its reservation (§369)")
-	}
-	e := ag.Entry
-	if ag.Do == AgainRenumber {
-		_, ne, err := a.Numberer.Versions.Number(ctx, Claim{Name: e.Name, Kind: e.Kind, Series: e.Series, Module: e.Module, Fingerprint: e.Fingerprint,
-			Commit: e.Commit, Revision: e.Revision, Seat: e.Seat, Merge: e.Merge, Fresh: true})
-		if err != nil {
-			return Again{}, "", err
-		}
-		ag.Entry = ne
-	}
-	answer := ag.Why
-	if ag.Do == AgainBuild || (ag.Do == AgainRenumber && !done) {
-		said, refused := a.BuildAgain(ctx, module, fp, why, by)
-		if refused {
-			return Again{}, "", fmt.Errorf("%s; and the build again was refused: %s", ag.Why, said)
-		}
-		answer += "; " + said
-	}
-	return ag, answer, nil
-}
-
 // FloorAgain is builder_floor_again of name: the floor counts only for a name the versions hold nothing of.
 func (a Acts) FloorAgain(ctx context.Context, name string) (string, error) {
 	entries, err := a.Numberer.Versions.Of(ctx, name)

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

@@ -0,0 +1,646 @@
+package versions
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"slices"
+	"sort"
+	"strings"
+	"sync"
+	"time"
+
+	"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
+	"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
+)
+
+// **A reserved version left unpublished is published again by the builder, on an event or by a person's tool** (design 47
+// G §588, ADR 0674): on the node-engine seat's `running` of a module whose catalog graph entry claims the registry seat
+// of the missing artifacts' kind, and on its `manifest-reads-changed` with `why` `returned` of the node such a holder
+// runs on. Each message is acted on once, by its stream sequence (state `acted`); each holder has one queue of versions,
+// kept in the state `publish-queue`, sent one at a time. A send is the registry seat's `publish` with the digest and no
+// parts (A-73): the holder takes the bytes from the artifact store by digest, and refuses what no store holds.
+
+const (
+	// PublishQueueState is the builder's old state section of the queues, capped at PublishQueueMiB.
+	PublishQueueState  = "publish-queue"
+	PublishQueueBucket = "mesh-delivery-builder_" + PublishQueueState
+	PublishQueueMiB    = 4
+	// ActedState holds, one key per subject, the last stream sequence the builder acted on; capped at ActedMiB.
+	ActedState  = "acted"
+	ActedBucket = "mesh-delivery-builder_" + ActedState
+	ActedMiB    = 1
+	// PublishQueueTool is builder_publish_queue.
+	PublishQueueTool = "builder_publish_queue"
+	// DefaultRepublishMax is the setting republish-max-versions unless set (§633).
+	DefaultRepublishMax = 50
+	// QueueFull is what version-unreleased says of a version the queue's cap refused.
+	QueueFull = "queue full"
+	// NotStoredWords opens what a refused artifact is said as.
+	NotStoredWords = "not stored: "
+	// VerbPublish is the registry seats' verb a publish again sends.
+	VerbPublish = "publish"
+)
+
+// RegistrySeats are the four registry seats a version is published to.
+var RegistrySeats = []string{SeatImages, SeatArtifacts, SeatGoPackages, SeatNPM}
+
+// SeatOfArtifact is the registry seat of one artifact of the catalog graph entry: from its package's ecosystem when the
+// entry names one (go, npm, image), else from its kind (SeatOf).
+func SeatOfArtifact(a meshcatalog.Artifact) (string, bool) {
+	if a.Package != nil {
+		switch a.Package.Ecosystem {
+		case "go":
+			return SeatGoPackages, true
+		case "npm":
+			return SeatNPM, true
+		case "image":
+			return SeatImages, true
+		}
+	}
+	seat, _, ok := SeatOf(a.Kind, a.Language)
+	return seat, ok
+}
+
+// QueuedPublish is the value of a publish-queue key `<holder module>.<module>.<version>` (each token escaped, §73): the
+// artifacts of the version this holder publishes, by reference with their digests, when it was queued and whether its
+// publish has begun.
+type QueuedPublish struct {
+	V         int               `json:"v"`
+	Holder    string            `json:"holder"`
+	Module    string            `json:"module"`
+	Version   string            `json:"version"`
+	Artifacts map[string]string `json:"artifacts"`
+	Queued    time.Time         `json:"queued"`
+	Begun     bool              `json:"begun,omitempty"`
+}
+
+// QueueKey is the key of holder's queued version of module.
+func QueueKey(holder, module, version string) string {
+	return EscapeToken(holder) + "." + EscapeToken(module) + "." + EscapeToken(version)
+}
+
+// Acted is the value of an acted key: the last stream sequence acted on for its subject.
+type Acted struct {
+	Sequence uint64 `json:"sequence"`
+}
+
+// QueueStore is the queue's bucket: compare-and-set, deletion and how many bytes it holds, for its cap.
+type QueueStore interface {
+	ReserveStore
+	Bytes(ctx context.Context) (uint64, error)
+}
+
+// Publish is one artifact's publish again: what the registry seat's publish takes, without parts.
+type Publish struct {
+	Seat, Kind, Name, Version, Digest, Fingerprint, Commit, Transfer string
+}
+
+// Republisher publishes reserved versions again (design 47 G §588).
+type Republisher struct {
+	Reserved *Reserver
+	Queue    QueueStore
+	Acted    ReserveStore
+	// Entry reads the catalog graph entry of a fingerprint (the seat mesh-catalog's verb entry).
+	Entry func(ctx context.Context, fingerprint string) (meshcatalog.CatalogEntry, error)
+	// Send sends one publish to a registry seat; an error is the holder's refusal or the call's fault, in its words.
+	Send func(ctx context.Context, p Publish) error
+	// Unreleased says version-unreleased about a version with the holder's words.
+	Unreleased func(ctx context.Context, v ReservedVersion, why string)
+	// Max is the setting republish-max-versions: the most versions one event queues.
+	Max  int
+	Now  func() time.Time
+	Logf func(string, ...any)
+	// Go starts the sending of a holder's queue; nil starts a goroutine.
+	Go func(func())
+
+	mu   sync.Mutex
+	runs map[string]map[string][]string // node -> holder module -> the registry seats it claims
+	busy map[string]bool
+}
+
+func (r *Republisher) now() time.Time {
+	if r.Now != nil {
+		return r.Now().UTC()
+	}
+	return time.Now().UTC()
+}
+
+func (r *Republisher) logf(f string, a ...any) {
+	if r.Logf != nil {
+		r.Logf(f, a...)
+	}
+}
+
+// ActedKey is the acted key of a subject.
+func ActedKey(subject string) string { return EscapeToken(subject) }
+
+// Heard takes one message of the node-engine seat's running or manifest-reads-changed in, at stream sequence seq: it acts
+// on it only when its sequence is newer than the one kept for its subject, and keeps it after. A subject that is neither is
+// not its; an error leaves the message not acted on, so it is read again.
+func (r *Republisher) Heard(ctx context.Context, subject string, data []byte, seq uint64) (bool, error) {
+	running := strings.HasPrefix(subject, strings.TrimSuffix(nodeengine.SubjectRunningAll, "*.*"))
+	changed := strings.HasPrefix(subject, "mesh.seat.node-engine.event."+nodeengine.EventManifestReadsChanged+".")
+	if !running && !changed {
+		return false, nil
+	}
+	key := ActedKey(subject)
+	raw, rev, found, err := r.Acted.Get(ctx, key)
+	if err != nil {
+		return true, fmt.Errorf("the acted sequence of %s cannot be read: %w", subject, err)
+	}
+	if found {
+		var a Acted
+		if json.Unmarshal(raw, &a) == nil && a.Sequence >= seq {
+			return true, nil
+		}
+	}
+	if running {
+		err = r.running(ctx, data)
+	} else {
+		err = r.returned(ctx, data)
+	}
+	if err != nil {
+		return true, err
+	}
+	val, _ := json.Marshal(Acted{Sequence: seq})
+	if found {
+		_, err = r.Acted.Update(ctx, key, val, rev)
+	} else {
+		_, err = r.Acted.Create(ctx, key, val)
+	}
+	if err != nil && !errors.Is(err, ErrConflict) {
+		return true, fmt.Errorf("the acted sequence of %s cannot be kept: %w", subject, err)
+	}
+	return true, nil
+}
+
+// registrySeatsOf are the registry seats the module of the catalog entry fingerprint claims, by their publish verb.
+func (r *Republisher) registrySeatsOf(ctx context.Context, fingerprint string) ([]string, error) {
+	e, err := r.Entry(ctx, fingerprint)
+	if err != nil {
+		return nil, err
+	}
+	var out []string
+	for _, seat := range RegistrySeats {
+		if slices.Contains(e.Serves, seat+"."+VerbPublish) {
+			out = append(out, seat)
+		}
+	}
+	return out, nil
+}
+
+// running takes a running in: a module stopped is forgotten; one running whose entry claims a registry seat is a holder on
+// its node, and its versions are queued.
+func (r *Republisher) running(ctx context.Context, data []byte) error {
+	var ev nodeengine.Running
+	if err := json.Unmarshal(data, &ev); err != nil || ev.Module == "" || ev.Node == "" {
+		r.logf("a running that does not read is not acted on: %v", err)
+		return nil
+	}
+	if ev.Stopped || ev.Fingerprint == "" {
+		r.mu.Lock()
+		delete(r.runs[ev.Node], ev.Module)
+		r.mu.Unlock()
+		return nil
+	}
+	seats, err := r.registrySeatsOf(ctx, ev.Fingerprint)
+	if err != nil {
+		return fmt.Errorf("the catalog entry %s of %s cannot be read: %w", ev.Fingerprint, ev.Module, err)
+	}
+	r.mu.Lock()
+	if r.runs == nil {
+		r.runs = map[string]map[string][]string{}
+	}
+	if r.runs[ev.Node] == nil {
+		r.runs[ev.Node] = map[string][]string{}
+	}
+	if len(seats) == 0 {
+		delete(r.runs[ev.Node], ev.Module)
+	} else {
+		r.runs[ev.Node][ev.Module] = seats
+	}
+	r.mu.Unlock()
+	if len(seats) == 0 {
+		return nil
+	}
+	_, err = r.queueEvent(ctx, map[string][]string{ev.Module: seats})
+	return err
+}
+
+// returned takes a manifest-reads-changed in: only one with why returned acts, for every holder known to run on its node.
+func (r *Republisher) returned(ctx context.Context, data []byte) error {
+	var ev nodeengine.ManifestReadsChanged
+	if err := json.Unmarshal(data, &ev); err != nil || ev.Node == "" {
+		r.logf("a manifest-reads-changed that does not read is not acted on: %v", err)
+		return nil
+	}
+	if ev.Why != nodeengine.ManifestReadsChangedWhyReturned {
+		return nil
+	}
+	r.mu.Lock()
+	holders := map[string][]string{}
+	for m, seats := range r.runs[ev.Node] {
+		holders[m] = seats
+	}
+	r.mu.Unlock()
+	if len(holders) == 0 {
+		return nil
+	}
+	_, err := r.queueEvent(ctx, holders)
+	return err
+}
+
+// view is a reserved version with the artifacts of it a holder would publish.
+type view struct {
+	v    ReservedVersion
+	arts map[string]meshcatalog.Artifact // reference -> its catalog artifact
+}
+
+// queueEvent queues, for each holder (module -> the seats it claims), the unreleased versions whose missing artifacts
+// that holder's seats take: newest keys first, at most Max versions in all (one key per module and MAJOR.MINOR is the
+// reservation's own).
+func (r *Republisher) queueEvent(ctx context.Context, holders map[string][]string) ([]string, error) {
+	over, err := r.Reserved.Overdue(ctx, nil)
+	if err != nil {
+		return nil, err
+	}
+	var vs []ReservedVersion
+	for _, o := range over {
+		if v, ok, err := r.Reserved.Of(ctx, o.Module, o.Version); err == nil && ok {
+			vs = append(vs, v)
+		}
+	}
+	return r.queue(ctx, vs, func(seat string) []string { return holdersOf(holders, seat) }, r.Max)
+}
+
+func holdersOf(holders map[string][]string, seat string) []string {
+	var out []string
+	for m, seats := range holders {
+		if slices.Contains(seats, seat) {
+			out = append(out, m)
+		}
+	}
+	sort.Strings(out)
+	return out
+}
+
+// queue queues vs newest first, at most max of them (0 is no limit), each artifact to the first holder holdersOf names for
+// its seat. It answers the versions queued, `<module> <version>`.
+func (r *Republisher) queue(ctx context.Context, vs []ReservedVersion, holdersOf func(seat string) []string, max int) ([]string, error) {
+	sort.SliceStable(vs, func(i, j int) bool { return vs[i].At.After(vs[j].At) })
+	var queued []string
+	started := map[string]bool{}
+	for _, v := range vs {
+		if max > 0 && len(queued) >= max {
+			break
+		}
+		if v.Fingerprint == "" || v.BuiltAt.IsZero() {
+			continue
+		}
+		e, err := r.Entry(ctx, v.Fingerprint)
+		if err != nil {
+			return queued, fmt.Errorf("the catalog entry %s of %s %s cannot be read: %w", v.Fingerprint, v.Module, v.Version, err)
+		}
+		byHolder := map[string]map[string]string{}
+		for _, ref := range v.Missing() {
+			if _, no := v.Unstored[ref]; no {
+				continue
+			}
+			for _, a := range e.Artifacts {
+				seat, ok := SeatOfArtifact(a)
+				if a.Ref != ref || !ok {
+					continue
+				}
+				if hs := holdersOf(seat); len(hs) > 0 {
+					if byHolder[hs[0]] == nil {
+						byHolder[hs[0]] = map[string]string{}
+					}
+					byHolder[hs[0]][ref] = v.Artifacts[ref]
+				}
+			}
+		}
+		did := false
+		for holder, arts := range byHolder {
+			ok, err := r.enqueue(ctx, v, holder, arts)
+			if err != nil {
+				return queued, err
+			}
+			did = did || ok
+			started[holder] = true
+		}
+		if did {
+			queued = append(queued, v.Module+" "+v.Version)
+		}
+	}
+	for holder := range started {
+		r.drainLater(ctx, holder)
+	}
+	return queued, nil
+}
+
+// enqueue puts one holder's artifacts of a version in its queue: not again when it is queued already, and not past the
+// section's cap (said as queue full).
+func (r *Republisher) enqueue(ctx context.Context, v ReservedVersion, holder string, arts map[string]string) (bool, error) {
+	key := QueueKey(holder, v.Module, v.Version)
+	if _, _, found, err := r.Queue.Get(ctx, key); err != nil {
+		return false, err
+	} else if found {
+		return false, nil
+	}
+	raw, _ := json.Marshal(QueuedPublish{V: 1, Holder: holder, Module: v.Module, Version: v.Version, Artifacts: arts, Queued: r.now()})
+	used, err := r.Queue.Bytes(ctx)
+	if err != nil {
+		return false, fmt.Errorf("the size of %s cannot be read: %w", PublishQueueBucket, err)
+	}
+	if used+uint64(len(raw)) > PublishQueueMiB<<20 {
+		if r.Unreleased != nil {
+			r.Unreleased(ctx, v, QueueFull)
+		}
+		return false, nil
+	}
+	if _, err := r.Queue.Create(ctx, key, raw); err != nil {
+		if errors.Is(err, ErrConflict) {
+			return false, nil
+		}
+		return false, fmt.Errorf("%s cannot be queued: %w", key, err)
+	}
+	return true, nil
+}
+
+// Republish is builder_publish_again: with module and version, that reserved version's missing artifacts; with module
+// alone, each reserved version of it whose artifacts are not all published; with neither, every version the finding
+// version-unreleased names. It queues as the events do (no limit but the cap), for the holders known to run, and answers
+// the versions it queued.
+func (r *Republisher) Republish(ctx context.Context, module, version string) ([]string, error) {
+	var vs []ReservedVersion
+	if module == "" {
+		over, err := r.Reserved.Overdue(ctx, nil)
+		if err != nil {
+			return nil, err
+		}
+		for _, o := range over {
+			if v, ok, err := r.Reserved.Of(ctx, o.Module, o.Version); err == nil && ok {
+				vs = append(vs, v)
+			}
+		}
+	} else {
+		all, err := r.Reserved.All(ctx)
+		if err != nil {
+			return nil, err
+		}
+		for _, v := range all {
+			if v.Module == module && (version == "" || v.Version == version) && len(v.Missing()) > 0 {
+				vs = append(vs, v)
+			}
+		}
+		if len(vs) == 0 {
+			return nil, fmt.Errorf("%s holds no reserved version%s with an artifact unpublished", module, map[bool]string{true: " " + version, false: ""}[version != ""])
+		}
+	}
+	r.mu.Lock()
+	known := map[string][]string{}
+	for _, mods := range r.runs {
+		for m, seats := range mods {
+			known[m] = seats
+		}
+	}
+	r.mu.Unlock()
+	return r.queue(ctx, vs, func(seat string) []string { return holdersOf(known, seat) }, 0)
+}
+
+// Queued are the queued versions, per holder in the order they are sent, and the acted keys with their sequences:
+// builder_publish_queue.
+type Queued struct {
+	Queue []QueuedPublish   `json:"queue"`
+	Acted map[string]uint64 `json:"acted"`
+}
+
+// AnswerQueue is builder_publish_queue over the two sections as the tools bundle reads them.
+func AnswerQueue(queue, acted Reader) (Queued, error) {
+	out := Queued{Queue: []QueuedPublish{}, Acted: map[string]uint64{}}
+	keys, err := queue.Keys()
+	if err != nil {
+		return out, fmt.Errorf("the state %s cannot be listed: %w", PublishQueueState, err)
+	}
+	for _, k := range keys {
+		raw, ok, err := queue.Get(k)
+		if err != nil || !ok {
+			continue
+		}
+		var q QueuedPublish
+		if json.Unmarshal(raw, &q) == nil {
+			out.Queue = append(out.Queue, q)
+		}
+	}
+	sort.SliceStable(out.Queue, func(i, j int) bool {
+		if out.Queue[i].Holder != out.Queue[j].Holder {
+			return out.Queue[i].Holder < out.Queue[j].Holder
+		}
+		return out.Queue[i].Queued.Before(out.Queue[j].Queued)
+	})
+	akeys, err := acted.Keys()
+	if err != nil {
+		return out, fmt.Errorf("the state %s cannot be listed: %w", ActedState, err)
+	}
+	for _, k := range akeys {
+		raw, ok, err := acted.Get(k)
+		if err != nil || !ok {
+			continue
+		}
+		var a Acted
+		if json.Unmarshal(raw, &a) == nil {
+			out.Acted[k] = a.Sequence
+		}
+	}
+	return out, nil
+}
+
+// PublishQueueDescription is builder_publish_queue's description.
+const PublishQueueDescription = "The versions the builder will publish again (its state " + PublishQueueState + ", design 47 G section 588): per registry " +
+	"seat holder, in the order they are sent, each with its artifacts' digests, when it was queued and whether its publish has begun; and the " +
+	"keys of the state " + ActedState + ", the last stream sequence the builder acted on for each running and manifest-reads-changed subject. Read-only."
+
+// PublishAgainDescription is builder_publish_again's description.
+const PublishAgainDescription = "Publish reserved versions again (design 47 G section 588), which builds nothing: with module and version, that reserved " +
+	"version's missing artifacts; with module alone, each reserved version of that module whose artifacts are not all published; with neither, every " +
+	"version the finding version-unreleased names, newest first. It queues them as the builder does on a registry holder's running, and answers at once " +
+	"with the versions it queued; each outcome is said as package-published, or left in version-unreleased with the holder's words. A hand act, " +
+	"recorded in the hand-act log."
+
+// Resume takes up, at the builder's start, every queued version again, a begun one sent anew (§588).
+func (r *Republisher) Resume(ctx context.Context) error {
+	keys, err := r.Queue.KeysOf(ctx, AllFilter)
+	if err != nil {
+		return fmt.Errorf("the state %s cannot be listed: %w", PublishQueueState, err)
+	}
+	holders := map[string]bool{}
+	for _, k := range keys {
+		raw, _, found, err := r.Queue.Get(ctx, k)
+		if err != nil || !found {
+			continue
+		}
+		var q QueuedPublish
+		if json.Unmarshal(raw, &q) == nil {
+			holders[q.Holder] = true
+		}
+	}
+	for h := range holders {
+		r.drainLater(ctx, h)
+	}
+	return nil
+}
+
+func (r *Republisher) drainLater(ctx context.Context, holder string) {
+	r.mu.Lock()
+	if r.busy == nil {
+		r.busy = map[string]bool{}
+	}
+	if r.busy[holder] {
+		r.mu.Unlock()
+		return
+	}
+	r.busy[holder] = true
+	r.mu.Unlock()
+	run := func() {
+		defer func() { r.mu.Lock(); delete(r.busy, holder); r.mu.Unlock() }()
+		r.Drain(ctx, holder)
+	}
+	if r.Go != nil {
+		r.Go(run)
+		return
+	}
+	go run()
+}
+
+// Drain sends holder's queued versions one at a time, oldest first, each bounded by UnreleasedBound; a key is deleted when
+// its outcome is said.
+func (r *Republisher) Drain(ctx context.Context, holder string) {
+	for ctx.Err() == nil {
+		keys, err := r.Queue.KeysOf(ctx, EscapeToken(holder)+".*.*")
+		if err != nil {
+			r.logf("the queue of %s cannot be listed: %v", holder, err)
+			return
+		}
+		var items []QueuedPublish
+		for _, k := range keys {
+			if raw, _, found, err := r.Queue.Get(ctx, k); err == nil && found {
+				var q QueuedPublish
+				if json.Unmarshal(raw, &q) == nil {
+					items = append(items, q)
+				}
+			}
+		}
+		if len(items) == 0 {
+			return
+		}
+		sort.SliceStable(items, func(i, j int) bool { return items[i].Queued.Before(items[j].Queued) })
+		r.send(ctx, items[0])
+	}
+}
+
+// send publishes one queued version's artifacts and says the outcome, then deletes its key.
+func (r *Republisher) send(ctx context.Context, q QueuedPublish) {
+	key := QueueKey(q.Holder, q.Module, q.Version)
+	defer func() {
+		if err := r.Queue.Delete(context.WithoutCancel(ctx), key); err != nil {
+			r.logf("%s cannot be deleted from %s: %v", key, PublishQueueState, err)
+		}
+	}()
+	if !q.Begun {
+		q.Begun = true
+		raw, _ := json.Marshal(q)
+		if _, rev, found, err := r.Queue.Get(ctx, key); err == nil && found {
+			_, _ = r.Queue.Update(ctx, key, raw, rev)
+		}
+	}
+	v, found, err := r.Reserved.Of(ctx, q.Module, q.Version)
+	if err != nil || !found {
+		return // released meanwhile, or dropped
+	}
+	e, err := r.Entry(ctx, v.Fingerprint)
+	if err != nil {
+		r.say(ctx, v, fmt.Sprintf("the catalog entry %s cannot be read: %v", v.Fingerprint, err))
+		return
+	}
+	refs := make([]string, 0, len(q.Artifacts))
+	for ref := range q.Artifacts {
+		refs = append(refs, ref)
+	}
+	sort.Strings(refs)
+	for _, ref := range refs {
+		var art *meshcatalog.Artifact
+		for i := range e.Artifacts {
+			if e.Artifacts[i].Ref == ref {
+				art = &e.Artifacts[i]
+			}
+		}
+		seat, ok := "", false
+		if art != nil {
+			seat, ok = SeatOfArtifact(*art)
+		}
+		if !ok {
+			r.say(ctx, v, ref+": the catalog entry names no registry seat for it")
+			return
+		}
+		name := NameOf(q.Module, strings.TrimPrefix(ref, q.Module+":"))
+		if art.Package != nil && art.Package.Name != "" {
+			name = art.Package.Name
+		}
+		kind := art.Kind
+		bound, cancel := context.WithTimeout(ctx, UnreleasedBound) // a context's bound, no timer (ADR 0437 §8)
+		err := r.Send(bound, Publish{Seat: seat, Kind: kind, Name: name, Version: v.Version, Digest: q.Artifacts[ref], Fingerprint: v.Fingerprint,
+			Commit: v.Commit, Transfer: "republish-" + q.Module + "-" + v.Version + "-" + strings.TrimPrefix(ref, q.Module+":")})
+		cancel()
+		if err != nil {
+			words := err.Error()
+			if strings.Contains(words, strings.TrimSuffix(NotStoredWords, ": ")) {
+				if ok, merr := r.Reserved.MarkUnstored(ctx, q.Module, q.Version, ref, words); merr != nil {
+					r.logf("%s %s: the unstored %s is not kept: %v", q.Module, q.Version, ref, merr)
+				} else if ok {
+					r.say(ctx, v, ref+" "+NotStoredWords+words)
+				}
+				continue
+			}
+			r.say(ctx, v, ref+": "+words)
+		}
+	}
+}
+
+func (r *Republisher) say(ctx context.Context, v ReservedVersion, why string) {
+	r.logf("%s %s is not published again: %s", v.Module, v.Version, why)
+	if r.Unreleased != nil {
+		r.Unreleased(ctx, v, why)
+	}
+}
+
+// MarkUnstored records that a holder refused artifact of module's reserved version as not stored; it answers whether that
+// was new (said once).
+func (r *Reserver) MarkUnstored(ctx context.Context, module, version, artifact, words string) (bool, error) {
+	series, _, err := SplitVersion(version)
+	if err != nil {
+		return false, err
+	}
+	key := ReservedKey(module, series)
+	for {
+		cur, rev, found, err := r.get(ctx, key)
+		if err != nil || !found || cur.Version != version {
+			return false, err
+		}
+		if _, was := cur.Unstored[artifact]; was {
+			return false, nil
+		}
+		if cur.Unstored == nil {
+			cur.Unstored = map[string]string{}
+		}
+		cur.Unstored[artifact] = words
+		raw, _ := json.Marshal(cur)
+		if _, err := r.Store.Update(ctx, key, raw, rev); errors.Is(err, ErrConflict) {
+			continue
+		} else if err != nil {
+			return false, fmt.Errorf("%s cannot be written: %w", key, err)
+		}
+		return true, nil
+	}
+}

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

@@ -0,0 +1,220 @@
+package versions
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+	"testing"
+	"time"
+
+	"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
+	"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
+)
+
+type queueMem struct {
+	*reserveMem
+	bytes uint64
+}
+
+func (q *queueMem) Bytes(context.Context) (uint64, error) { return q.bytes, nil }
+
+// republishBed is a Republisher over memory states: reservations of module m (a bundle by the oci seat and a Go package by
+// the Go seat), one holder module `oci` that claims the oci seat, and a Send and a Unreleased that record.
+type republishBed struct {
+	*Republisher
+	reserved *reserveMem
+	queue    *queueMem
+	acted    *reserveMem
+	now      time.Time
+	sent     []Publish
+	said     []string
+	refuse   string // the words Send fails with, empty for none
+}
+
+func newRepublishBed(t *testing.T, max int) *republishBed {
+	t.Helper()
+	b := &republishBed{reserved: &reserveMem{mem: newMem()}, queue: &queueMem{reserveMem: &reserveMem{mem: newMem()}}, acted: &reserveMem{mem: newMem()}, now: time.Date(2026, 11, 11, 12, 0, 0, 0, time.UTC)}
+	res := &Reserver{Store: b.reserved, Now: func() time.Time { return b.now }, ReleasedOf: func(context.Context, string) ([]string, error) { return nil, nil }}
+	b.Republisher = &Republisher{Reserved: res, Queue: b.queue, Acted: b.acted, Max: max, Now: func() time.Time { return b.now }, Go: func(f func()) { f() },
+		Entry: func(_ context.Context, fp string) (meshcatalog.CatalogEntry, error) {
+			if fp == "src2:holder" {
+				return meshcatalog.CatalogEntry{Serves: []string{SeatArtifacts + ".publish", SeatArtifacts + ".versions"}}, nil
+			}
+			m := strings.TrimPrefix(fp, "src2:")
+			return meshcatalog.CatalogEntry{Artifacts: []meshcatalog.Artifact{
+				{Ref: m + ":prog", Kind: "bundle", Language: "go"},
+				{Ref: m + ":lib", Kind: "package", Language: "go", Package: &meshcatalog.ArtifactPackage{Ecosystem: "go", Name: "git.novox.be/novox/" + m + "/go"}}}}, nil
+		},
+		Send: func(_ context.Context, p Publish) error {
+			b.sent = append(b.sent, p)
+			if b.refuse != "" {
+				return errors.New(b.refuse)
+			}
+			return nil
+		},
+		Unreleased: func(_ context.Context, v ReservedVersion, why string) {
+			b.said = append(b.said, v.Module+" "+v.Version+": "+why)
+		},
+		Logf: t.Logf}
+	return b
+}
+
+// reserve reserves and builds a version of module m (its patch given by the reservation order), its build ended long ago.
+func (b *republishBed) reserve(t *testing.T, module string) ReservedVersion {
+	t.Helper()
+	b.now = b.now.Add(time.Minute)
+	ctx := context.Background()
+	v, err := b.Reserved.Reserve(ctx, module, "0.1", []string{module + ":prog", module + ":lib"}, Source{Fingerprint: "src2:" + module})
+	if err != nil {
+		t.Fatal(err)
+	}
+	for ref, d := range map[string]string{module + ":prog": "sha256:aa", module + ":lib": "sha256:bb"} {
+		if _, err := b.Reserved.Built(ctx, module, v.Version, ref, d); err != nil {
+			t.Fatal(err)
+		}
+	}
+	b.now = b.now.Add(UnreleasedBound + time.Minute)
+	return v
+}
+
+func running(node, module, fp string, stopped bool) (string, []byte) {
+	data, _ := json.Marshal(nodeengine.Running{Node: node, Module: module, Fingerprint: fp, Stopped: stopped, At: time.Unix(1, 0)})
+	return nodeengine.SubjectRunning(node, module), data
+}
+
+// TestRepublish: design 47 G §588 (A-73, A-76): a reserved version left unpublished is published again, on a registry
+// holder's running and on its node's returned, once per message by its stream sequence, as a publish by digest with no parts,
+// at most the setting's number per event, newest first; an artifact no store holds is said once and not tried again.
+func TestRepublish(t *testing.T) {
+	ctx := context.Background()
+	t.Run("running of a holder, once per sequence", func(t *testing.T) {
+		b := newRepublishBed(t, 50)
+		v := b.reserve(t, "m")
+		if v.Fingerprint != "src2:m" {
+			t.Fatalf("the reservation keeps the entry's fingerprint: %q", v.Fingerprint)
+		}
+		subject, data := running("n1", "oci", "src2:holder", false)
+		for _, seq := range []uint64{5, 4} { // 4 is older than the acted 5: it starts nothing
+			if _, err := b.Heard(ctx, subject, data, seq); err != nil || len(b.sent) != 1 {
+				t.Fatalf("seq %d: sent %d (%v), want 1", seq, len(b.sent), err)
+			}
+		}
+		p := b.sent[0]
+		if p.Seat != SeatArtifacts || p.Name != "m/prog" || p.Kind != "bundle" || p.Version != v.Version || p.Digest != "sha256:aa" || p.Fingerprint != "src2:m" {
+			t.Errorf("sent %+v, want the bundle by digest to the oci seat, the Go package left to its own seat's holder", p)
+		}
+		if _, err := b.Heard(ctx, subject, data, 5); err != nil || len(b.sent) != 1 {
+			t.Errorf("the same sequence read again sent %d, want still 1", len(b.sent))
+		}
+		if _, err := b.Heard(ctx, subject, data, 6); err != nil || len(b.sent) != 2 {
+			t.Errorf("a newer sequence sent %d, want 2", len(b.sent))
+		}
+		if keys, _ := b.queue.KeysOf(ctx, AllFilter); len(keys) != 0 {
+			t.Errorf("a key stays in the queue after its outcome: %v", keys)
+		}
+	})
+	t.Run("returned acts, started does not", func(t *testing.T) {
+		b := newRepublishBed(t, 50)
+		b.reserve(t, "m")
+		subject, data := running("n1", "oci", "src2:holder", true) // stopped: not a holder yet
+		_, _ = b.Heard(ctx, subject, data, 1)
+		subject, data = running("n1", "oci", "src2:holder", false)
+		_, _ = b.Heard(ctx, subject, data, 2)
+		b.sent = nil
+		for i, why := range []string{nodeengine.ManifestReadsChangedWhyStarted, "", nodeengine.ManifestReadsChangedWhyReturned} {
+			ev, _ := json.Marshal(nodeengine.ManifestReadsChanged{Node: "n1", Why: why})
+			handled, err := b.Heard(ctx, nodeengine.SubjectManifestReadsChanged("n1"), ev, uint64(10+i))
+			if !handled || err != nil {
+				t.Fatalf("%q: handled %v, %v", why, handled, err)
+			}
+			if want := map[string]int{"started": 0, "": 0, "returned": 1}[why]; len(b.sent) != want {
+				t.Errorf("why %q: sent %d, want %d", why, len(b.sent), want)
+			}
+		}
+		if handled, _ := b.Heard(ctx, "mesh.seat.node-engine.event.composed.n1", nil, 1); handled {
+			t.Error("a subject that is neither is taken as its own")
+		}
+	})
+	t.Run("the setting limits an event, newest first", func(t *testing.T) {
+		b := newRepublishBed(t, 2)
+		var versions []string
+		for _, m := range []string{"m1", "m2", "m3"} {
+			versions = append(versions, b.reserve(t, m).Module)
+		}
+		subject, data := running("n1", "oci", "src2:holder", false)
+		if _, err := b.Heard(ctx, subject, data, 1); err != nil {
+			t.Fatal(err)
+		}
+		if len(b.sent) != 2 || !strings.HasPrefix(b.sent[0].Name, "m3/") || !strings.HasPrefix(b.sent[1].Name, "m2/") {
+			t.Errorf("sent %+v, want the two newest, m3 then m2", b.sent)
+		}
+	})
+	t.Run("an artifact no store holds is said once and not tried again", func(t *testing.T) {
+		b := newRepublishBed(t, 50)
+		b.reserve(t, "m")
+		b.refuse = "oci-artifact-registry's publish of m/prog: not stored: no blob sha256:aa"
+		subject, data := running("n1", "oci", "src2:holder", false)
+		_, _ = b.Heard(ctx, subject, data, 1)
+		_, _ = b.Heard(ctx, subject, data, 2)
+		if len(b.sent) != 1 || len(b.said) != 1 || !strings.Contains(b.said[0], "not stored: ") {
+			t.Errorf("sent %d, said %v: want one try and one saying of not stored", len(b.sent), b.said)
+		}
+	})
+	t.Run("a refusal in other words is said each time", func(t *testing.T) {
+		b := newRepublishBed(t, 50)
+		b.reserve(t, "m")
+		b.refuse = "the holder is full"
+		subject, data := running("n1", "oci", "src2:holder", false)
+		_, _ = b.Heard(ctx, subject, data, 1)
+		_, _ = b.Heard(ctx, subject, data, 2)
+		if len(b.sent) != 2 || len(b.said) != 2 {
+			t.Errorf("sent %d, said %v: want both events tried and said", len(b.sent), b.said)
+		}
+	})
+	t.Run("a queue past its cap says queue full", func(t *testing.T) {
+		b := newRepublishBed(t, 50)
+		b.reserve(t, "m")
+		b.queue.bytes = PublishQueueMiB << 20
+		subject, data := running("n1", "oci", "src2:holder", false)
+		_, _ = b.Heard(ctx, subject, data, 1)
+		if len(b.sent) != 0 || len(b.said) != 1 || !strings.HasSuffix(b.said[0], QueueFull) {
+			t.Errorf("sent %d, said %v, want nothing sent and queue full said", len(b.sent), b.said)
+		}
+	})
+	t.Run("a restart sends the queued again, a begun one anew", func(t *testing.T) {
+		b := newRepublishBed(t, 50)
+		v := b.reserve(t, "m")
+		raw, _ := json.Marshal(QueuedPublish{V: 1, Holder: "oci", Module: "m", Version: v.Version, Artifacts: map[string]string{"m:prog": "sha256:aa"}, Queued: b.now, Begun: true})
+		if _, err := b.queue.Create(ctx, QueueKey("oci", "m", v.Version), raw); err != nil {
+			t.Fatal(err)
+		}
+		if err := b.Resume(ctx); err != nil || len(b.sent) != 1 {
+			t.Errorf("resume sent %d (%v), want the begun version sent anew", len(b.sent), err)
+		}
+	})
+	t.Run("the tool queues for a holder known to run", func(t *testing.T) {
+		b := newRepublishBed(t, 1)
+		b.reserve(t, "m1")
+		b.reserve(t, "m2")
+		if q, err := b.Republish(ctx, "", ""); err != nil || len(q) != 0 {
+			t.Errorf("no holder known: queued %v (%v), want none", q, err)
+		}
+		subject, data := running("n1", "oci", "src2:holder", false)
+		_, _ = b.Heard(ctx, subject, data, 1) // Max 1: only m2 sent
+		b.sent = nil
+		q, err := b.Republish(ctx, "", "")
+		if err != nil || fmt.Sprint(q) != "[m2 0.1.0 m1 0.1.0]" || len(b.sent) != 2 {
+			t.Errorf("the tool with neither queued %v (%v), sent %d: want both versions, with no limit", q, err, len(b.sent))
+		}
+		if _, err := b.Republish(ctx, "m1", "0.1.9"); err == nil {
+			t.Error("a version no reservation holds is not refused")
+		}
+		answer, err := AnswerQueue(Reader{Keys: func() ([]string, error) { return nil, nil }}, Reader{Keys: func() ([]string, error) { return b.acted.KeysOf(ctx, AllFilter) },
+			Get: func(k string) (json.RawMessage, bool, error) { v, _, ok, err := b.acted.Get(ctx, k); return v, ok, err }})
+		if err != nil || answer.Acted[ActedKey(subject)] != 1 {
+			t.Errorf("builder_publish_queue acted = %v (%v), want the running's sequence 1", answer.Acted, err)
+		}
+	})
+}

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

@@ -54,13 +54,21 @@ type ReservedVersion struct {
 	Repository string `json:"repository,omitempty"`
 	Commit     string `json:"commit,omitempty"`
 	Tag        string `json:"tag,omitempty"`
+	// Fingerprint is the fingerprint of the catalog entry the reservation was made from, so a publish again reads that
+	// entry (design 47 G A-76).
+	Fingerprint string `json:"fingerprint,omitempty"`
+	// Unstored are the artifacts a holder refused as not stored, with its words: said once, and not published again on
+	// events (design 47 G §588); builder_build_again builds the key again, which reserves anew.
+	Unstored map[string]string `json:"unstored,omitempty"`
 }
 
 // Source is what a build reserving a version is made from (design 47 G §370): the repository and commit the git tag is
 // written at, and whether the module has a Go package, whose version is tagged.
 type Source struct {
 	Repository, Commit string
-	GoPackage          bool
+	// Fingerprint is the fingerprint of the catalog entry the build is made from (design 47 G A-76).
+	Fingerprint string
+	GoPackage   bool
 }
 
 // ReservedKey is the key of module's reservation of series (MAJOR.MINOR).
@@ -135,7 +143,7 @@ func (r *Reserver) Reserve(ctx context.Context, module, series string, artifacts
 			}
 		}
 		out := ReservedVersion{V: 1, Module: module, Series: series, Version: fmt.Sprintf("%s.%d", series, next+1), Artifacts: map[string]string{}, At: r.now(),
-			Repository: src.Repository, Commit: src.Commit}
+			Repository: src.Repository, Commit: src.Commit, Fingerprint: src.Fingerprint}
 		if src.GoPackage {
 			out.Tag = GoTagOf(out.Version)
 		}

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

@@ -73,6 +73,13 @@ func (m *mem) KeysOf(_ context.Context, filter string) ([]string, error) {
 			out = append(out, k)
 			continue
 		}
+		if strings.Count(filter, ".") == 2 { // a three-token filter, `a.*.*` as the bus reads it
+			ft, kt := strings.Split(filter, "."), strings.Split(k, ".")
+			if len(kt) == 3 && kt[0] == ft[0] {
+				out = append(out, k)
+			}
+			continue
+		}
 		if strings.HasPrefix(k, prefix) && !strings.Contains(k[len(prefix):], ".") {
 			out = append(out, k)
 		}
@@ -540,39 +547,7 @@ func TestTheBuildersVersions(t *testing.T) {
 			t.Error("80 % is not full")
 		}
 	})
-	t.Run("publish again", func(t *testing.T) {
-		num := Entry{Name: "m/p", Kind: KindBundle, Version: "0.1.3", Fingerprint: "f1", Digest: "sha256:aa", State: Numbered}
-		ref := num
-		ref.State, ref.RefusedDigest = Refused, "sha256:aa"
-		other := ref
-		other.RefusedDigest = "sha256:bb"
-		pub := num
-		pub.State = Published
-		gone := pub
-		gone.State = Removed
-		tagged := num
-		tagged.Tag, tagged.Version = "feat-x", "0.1.3-feat-x.1"
-		for _, c := range []struct {
-			name    string
-			entries []Entry
-			fp      string
-			built   bool
-			do      string
-		}{
-			{"a held build is said numbered again", []Entry{num}, "f1", true, AgainPublish},
-			{"refused for its own bytes is published again", []Entry{ref}, "f1", true, AgainPublish},
-			{"a gone file is built again under the same number", []Entry{ref}, "f1", false, AgainBuild},
-			{"after a published it takes a new number", []Entry{pub}, "f1", true, AgainRenumber},
-			{"after a removal it takes a new number", []Entry{gone}, "f1", false, AgainRenumber},
-			{"refused for other bytes is refused", []Entry{other}, "f1", true, ""},
-			{"a fingerprint not numbered is refused", []Entry{num}, "f2", true, ""},
-			{"a tagged version is not a trunk fingerprint's", []Entry{tagged}, "f1", true, ""},
-		} {
-			a, err := PublishAgain(c.entries, c.fp, c.built)
-			if (err != nil) != (c.do == "") || a.Do != c.do {
-				t.Errorf("%s: PublishAgain = %+v, %v; want %q", c.name, a, err, c.do)
-			}
-		}
+	t.Run("fresh claims and the floor", func(t *testing.T) {
 		ctx := context.Background()
 		// A fresh claim of a published fingerprint takes the next number; the old one stays.
 		st := newMem()
@@ -603,32 +578,8 @@ func TestTheBuildersVersions(t *testing.T) {
 		if all, err := v.All(ctx); err != nil || len(all) != 2 {
 			t.Errorf("All = %d entries, %v", len(all), err)
 		}
-		// The hand acts: a held build is said numbered once more under a fresh dedup id; a gone one is built again; a
-		// published one takes a new number; a name with versions asks no floor.
-		if _, _, err := (Acts{Numberer: &Numberer{Versions: v}, Done: func(context.Context, string, string) (bool, error) { return true, nil }}).PublishAgain(ctx, "m/p", "f1", "w", "", time.Unix(4, 0)); err == nil {
-			t.Error("a version refused for other bytes is published again")
-		}
 		_, _ = v.Mark(ctx, "m/p", "0.1.4", func(e *Entry) { e.State = Published })
-		var rebuilt []string
-		done := true
-		acts := Acts{Numberer: &Numberer{Versions: v, Waits: &FloorWaits{}},
-			Done: func(context.Context, string, string) (bool, error) { return done, nil },
-			BuildAgain: func(_ context.Context, m, fp, _, _ string) (string, bool) {
-				rebuilt = append(rebuilt, m+" "+fp)
-				return "built again", false
-			},
-			Reconsider: func(context.Context) {}}
-		// Nothing says a version numbered (design 47 G §349): a held build is refused before any number is taken.
-		if _, _, err := acts.PublishAgain(ctx, "m/p", "f1", "w", "me", time.Unix(5, 0)); err == nil || len(rebuilt) != 0 {
-			t.Errorf("publish again of a published, held build answered %v, rebuilt %v; want a refusal", err, rebuilt)
-		}
-		if es, _ := v.Of(ctx, "m/p"); len(es) != 2 {
-			t.Errorf("the refused publish again took a number: %+v", es)
-		}
-		done = false
-		if ag, _, err := acts.PublishAgain(ctx, "m/p", "f1", "w", "me", time.Unix(6, 0)); err != nil || ag.Do != AgainRenumber || ag.Entry.Version != "0.1.5" || len(rebuilt) != 1 {
-			t.Errorf("publish again of a gone build: %+v %v rebuilt %v", ag, err, rebuilt)
-		}
+		acts := Acts{Numberer: &Numberer{Versions: v, Waits: &FloorWaits{}}, Reconsider: func(context.Context) {}}
 		if _, err := acts.FloorAgain(ctx, "m/p"); err == nil {
 			t.Error("a name with versions asks its floor again")
 		}

modules/mesh-delivery-builder/manifest_test.go

@@ -113,7 +113,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: "bounds", History: 1, KeyTTL: true}, {Name: "queues", History: 1}, {Name: "toolchains-heard", History: 1}, {Name: "waits", History: 1, MaxMiB: 4}, {Name: "reserved", History: 1}, {Name: "reasks", History: 1}, {Name: "versions", History: 1}, {Name: "first-views", History: 1}}; !reflect.DeepEqual(m.State, want) { // builds.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: "bounds", History: 1, KeyTTL: true}, {Name: "queues", History: 1}, {Name: "toolchains-heard", History: 1}, {Name: "waits", History: 1, MaxMiB: 4}, {Name: "reserved", History: 1}, {Name: "publish-queue", History: 1, MaxMiB: 4}, {Name: "acted", History: 1, MaxMiB: 1}, {Name: "reasks", History: 1}, {Name: "versions", History: 1}, {Name: "first-views", History: 1}}; !reflect.DeepEqual(m.State, want) { // builds.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
@@ -142,10 +142,10 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
 	// Design 51 *The floor*: the floor's three sources, the registry seats' versions, bridge_tags and indexer_requirements.
 	// The watchdogs seat's presence says a node's machine is heard (issue 1966, ADR 0596 §2, ADR 0668 §3).
 	if want := []string{"mesh-delivery-assigner.assigner_read_assignments", "mesh-delivery-forge-bridge.bridge_tags", "mesh-delivery-indexer.indexer_read_candidates", "mesh-delivery-indexer.indexer_read_manifests",
-		"mesh-delivery-indexer.indexer_read_modules", "mesh-delivery-indexer.indexer_read_packages", "mesh-delivery-indexer.indexer_requirements", "seat:container-image-registry.remove", "seat:container-image-registry.versions",
+		"mesh-delivery-indexer.indexer_read_modules", "mesh-delivery-indexer.indexer_read_packages", "mesh-delivery-indexer.indexer_requirements", "seat:container-image-registry.publish", "seat:container-image-registry.remove", "seat:container-image-registry.versions",
 		"seat:git.tag",
-		"seat:go-package-registry.remove", "seat:go-package-registry.versions", "seat:mesh-artifact-sweep.held", "seat:mesh-catalog.versions", "seat:mesh-hand-acts.hand-act", "seat:mesh-watchdogs.presence", "seat:node-build-go.drop", "seat:node-build-image.drop", "seat:node-build-node.drop", "seat:node-engine.manifest-reads", "seat:npm-package-registry.remove",
-		"seat:npm-package-registry.versions", "seat:oci-artifact-registry.remove", "seat:oci-artifact-registry.versions"}; !reflect.DeepEqual(sorted(m.Invokes), want) {
+		"seat:go-package-registry.publish", "seat:go-package-registry.remove", "seat:go-package-registry.versions", "seat:mesh-artifact-sweep.held", "seat:mesh-catalog.entry", "seat:mesh-catalog.versions", "seat:mesh-hand-acts.hand-act", "seat:mesh-watchdogs.presence", "seat:node-build-go.drop", "seat:node-build-image.drop", "seat:node-build-node.drop", "seat:node-engine.manifest-reads", "seat:npm-package-registry.publish", "seat:npm-package-registry.remove",
+		"seat:npm-package-registry.versions", "seat:oci-artifact-registry.publish", "seat:oci-artifact-registry.remove", "seat:oci-artifact-registry.versions"}; !reflect.DeepEqual(sorted(m.Invokes), want) {
 		t.Errorf("invokes = %v, want %v", m.Invokes, want)
 	}
 	// The builder hears the build seats' toolchain-reported and the node-engine seat's running (design 47 G §492, §546) and reads no
@@ -172,7 +172,7 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
 	}
 	// 1945: the builder's tools (builder_seat_queue: note 62), served by its tools bundle, and the event it says when it asks an ask again.
 	if want := []string{"builder_announce_again", "builder_ask_again", "builder_bounds", "builder_build_again", "builder_builds_since", "builder_engine_reads", "builder_first_view_again", "builder_first_views",
-		"builder_floor_again", "builder_floor_waits", "builder_old_form", "builder_publish_again", "builder_release", "builder_reserved", "builder_reserved_drop", "builder_seat_queue", "builder_stuck", "builder_toolchains", "builder_toolchains_drop", "builder_versions", "builder_waits", "reserved"}; !reflect.DeepEqual(sorted(m.Tools), want) {
+		"builder_floor_again", "builder_floor_waits", "builder_old_form", "builder_publish_again", "builder_publish_queue", "builder_release", "builder_reserved", "builder_reserved_drop", "builder_seat_queue", "builder_stuck", "builder_toolchains", "builder_toolchains_drop", "builder_versions", "builder_waits", "reserved"}; !reflect.DeepEqual(sorted(m.Tools), want) {
 		t.Errorf("tools = %v, want %v", m.Tools, want)
 	}
 	// Design 47 G §672: the toolchain waits' stream, kept 7 days within 16 MiB, and the bound's setting, 60 minutes.
@@ -181,8 +181,8 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
 		m.Streams[1].Name != "mesh-delivery-builder-findings" || !reflect.DeepEqual(m.Streams[1].Events, []string{"finding-written.*", "finding-cleared.*", "findings-stated.*"}) {
 		t.Errorf("streams = %+v, want waits with wait-held.** and wait-ended.**, 7 days, 16 MiB, and the findings' stream", m.Streams)
 	}
-	if c, ok := m.Config["toolchain-wait-minutes"]; !ok || c.Default != float64(60) || len(m.Config) != 1 {
-		t.Errorf("config = %+v, want toolchain-wait-minutes of 60", m.Config)
+	if c, ok := m.Config["toolchain-wait-minutes"]; !ok || c.Default != float64(60) || len(m.Config) != 2 || m.Config["republish-max-versions"].Default != float64(50) {
+		t.Errorf("config = %+v, want toolchain-wait-minutes of 60 and republish-max-versions of 50 (design 47 G §588, §633)", m.Config)
 	}
 	// Design 47 G §349: version.numbered is no longer said; old-form-closed on the family version, apart from the builds' ends.
 	if want := []string{"build.asked-again", "version.old-form-closed"}; !reflect.DeepEqual(m.Emits, want) {
@@ -222,7 +222,7 @@ func TestManifestProcess(t *testing.T) { // 1607/AC1
 		t.Errorf("process = %v", p)
 	}
 	env, _ := p["env"].(map[string]any)
-	if len(env) != 2 || env["MESH_BROKER_FILE"] != "${dir:state}/broker" || env["MESH_BUILDER_TOOLCHAIN_WAIT_FILE"] != "${dir:state}/toolchain-wait-minutes" {
+	if len(env) != 3 || env["MESH_BROKER_FILE"] != "${dir:state}/broker" || env["MESH_BUILDER_REPUBLISH_MAX_FILE"] != "${dir:state}/republish-max-versions" || env["MESH_BUILDER_TOOLCHAIN_WAIT_FILE"] != "${dir:state}/toolchain-wait-minutes" {
 		t.Errorf("env = %v, want the broker file and the path of the toolchain wait's file, never the setting itself (design 47 G §672, A-71)", p["env"])
 	}
 	for k, v := range env {

modules/mesh-delivery-builder/module.json

@@ -22,6 +22,7 @@
     "builder_floor_waits",
     "builder_old_form",
     "builder_publish_again",
+    "builder_publish_queue",
     "builder_reserved",
     "builder_release",
     "builder_reserved_drop",
@@ -82,6 +83,7 @@
     "seat:go-package-registry.remove",
     "seat:go-package-registry.versions",
     "seat:mesh-artifact-sweep.held",
+    "seat:mesh-catalog.entry",
     "seat:mesh-catalog.versions",
     "seat:mesh-hand-acts.hand-act",
     "seat:mesh-watchdogs.presence",
@@ -97,7 +99,11 @@
     "mesh-delivery-indexer.indexer_read_candidates",
     "mesh-delivery-indexer.indexer_read_manifests",
     "mesh-delivery-indexer.indexer_read_modules",
-    "mesh-delivery-indexer.indexer_read_packages"
+    "mesh-delivery-indexer.indexer_read_packages",
+    "seat:container-image-registry.publish",
+    "seat:go-package-registry.publish",
+    "seat:npm-package-registry.publish",
+    "seat:oci-artifact-registry.publish"
   ],
   "state": [
     { "name": "builds", "history": 5 },
@@ -112,6 +118,8 @@
     { "name": "toolchains-heard", "history": 1 },
     { "name": "waits", "history": 1, "max-mib": 4 },
     { "name": "reserved", "history": 1 },
+    { "name": "publish-queue", "history": 1, "max-mib": 4 },
+    { "name": "acted", "history": 1, "max-mib": 1 },
     { "name": "reasks", "history": 1 },
     { "name": "versions", "history": 1 },
     { "name": "first-views", "history": 1 }
@@ -128,6 +136,14 @@
       "maximum": 10080,
       "trusted": false,
       "default": 60
+    },
+    "republish-max-versions": {
+      "type": "number",
+      "description": "the most versions one event of a registry holder queues to publish again, newest first (design 47 G section 588); the rest wait for the next event or builder_publish_again.",
+      "minimum": 1,
+      "maximum": 10000,
+      "trusted": false,
+      "default": 50
     }
   },
   "logs": [
@@ -173,6 +189,14 @@
       "mode": "0600",
       "content": "${config:toolchain-wait-minutes}"
     },
+    {
+      "id": "republish-max",
+      "type": "file",
+      "path": "${dir:state}/republish-max-versions",
+      "owner": "mesh-delivery-builder",
+      "mode": "0600",
+      "content": "${config:republish-max-versions}"
+    },
     {
       "id": "builder",
       "type": "process",
@@ -183,7 +207,8 @@
       ],
       "env": {
         "MESH_BROKER_FILE": "${dir:state}/broker",
-        "MESH_BUILDER_TOOLCHAIN_WAIT_FILE": "${dir:state}/toolchain-wait-minutes"
+        "MESH_BUILDER_TOOLCHAIN_WAIT_FILE": "${dir:state}/toolchain-wait-minutes",
+        "MESH_BUILDER_REPUBLISH_MAX_FILE": "${dir:state}/republish-max-versions"
       },
       "user": "mesh-delivery-builder",
       "health": {