F-catalog-B: registry_mirrors reads distribution's mirrors, asks one again and drops one by hand

The state mirrors had no read tool and no ask-again (seams review note 1): a copy a follower holds and
does not end could only wait for a restart. registry_mirrors lists each kept copy and its claim, asks one
again with its claim let go, and drops one, saying refused under the ask's id so its waiter ends.

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

modules/distribution/cmd/distribution-tools/main.go

@@ -230,6 +230,28 @@ func (mirrors) Claim(key string, r holder.MirrorRequest, revision uint64) (bool,
 	return err == nil, err
 }
 func (mirrors) Delete(key string) error { return stdio.State(holder.MirrorsState).Delete(key) }
+func (mirrors) List() (map[string]holder.MirrorRequest, error) {
+	keys, err := stdio.State(holder.MirrorsState).Keys()
+	if err != nil {
+		return nil, err
+	}
+	out := map[string]holder.MirrorRequest{}
+	for _, k := range keys {
+		e, err := stdio.State(holder.MirrorsState).Get(k)
+		if err != nil {
+			return nil, err
+		}
+		var r holder.MirrorRequest
+		if e != nil && json.Unmarshal(e.Value, &r) == nil {
+			out[k] = r
+		}
+	}
+	return out, nil
+}
+func (mirrors) Put(key string, r holder.MirrorRequest) error {
+	_, err := stdio.State(holder.MirrorsState).Put(key, r)
+	return err
+}
 
 // followMirrors runs every copy `mirror` took on, on the state's watch: the asks kept when the bundle starts (a copy a
 // restart cut) and each new one, one at a time. Each is claimed first (Verbs.Take), so a second follower copies none
@@ -447,6 +469,32 @@ func tools(v *holder.Verbs, windows kept[holder.Window], callers func() (holder.
 				used, _ := v.Objects.Used()
 				return map[string]any{"transfers": all, "used": used, "cap": v.CapNow(), "dropped_idle": idle}, err
 			}},
+		stdio.Tool{Name: "registry_mirrors", Description: "The copies of outside images mirror took on and has not ended (the state mirrors): each by its key, its source, digest, who asked and the follower that claimed it (none: it waits). With again, asks one again by hand, its claim let go so a follower takes it on; with drop, drops one by hand, saying refused on the image seat's events so its waiter ends (each a hand act, why required).",
+			Input: obj(map[string]any{"again": s("a kept copy's key to ask again"), "drop": s("a kept copy's key to drop"), "why": s("why (a hand act)")}),
+			Run: func(args map[string]any) (any, error) {
+				by, why := stdio.CallerOf(args), str(args, "why")
+				switch {
+				case str(args, "again") != "":
+					if why == "" {
+						return nil, errors.New("asking a copy again is a hand act: say why")
+					}
+					r, err := v.MirrorAgain(str(args, "again"))
+					if err != nil {
+						return nil, err
+					}
+					logf("the copy %s asked again by hand by %s: %s", str(args, "again"), by, why)
+					return map[string]any{"asked_again": str(args, "again"), "copy": r}, nil
+				case str(args, "drop") != "":
+					r, err := v.MirrorDrop(str(args, "drop"), by, why)
+					if err != nil {
+						return nil, err
+					}
+					logf("the copy %s dropped by hand by %s: %s", str(args, "drop"), by, why)
+					return map[string]any{"dropped": str(args, "drop"), "copy": r}, nil
+				}
+				all, err := v.MirrorsNow()
+				return map[string]any{"mirrors": all}, err
+			}},
 		stdio.Tool{Name: "registry_holds", Description: "What is held, since when and for whom; with release, lets one go through the seat's release as the operator's session (a hand act).",
 			Input: obj(map[string]any{"seat": s("the seat"), "release": s("the name to release"), "digest": s("its digest"), "why": s("why")}),
 			Run: func(args map[string]any) (any, error) {

modules/distribution/internal/holder/holder_test.go

@@ -159,6 +159,14 @@ func (m memMirrors) Claim(k string, r MirrorRequest, rev uint64) (bool, error) {
 	return true, nil
 }
 func (m memMirrors) Delete(k string) error { delete(m, k); return nil }
+func (m memMirrors) List() (map[string]MirrorRequest, error) {
+	out := map[string]MirrorRequest{}
+	for k, r := range m {
+		out[k] = r
+	}
+	return out, nil
+}
+func (m memMirrors) Put(k string, r MirrorRequest) error { m[k] = r; return nil }
 
 // mirrorRev is a kept ask's revision in memMirrors: 1 unclaimed, 2 once claimed.
 func mirrorRev(r MirrorRequest) uint64 {
@@ -646,6 +654,24 @@ func TestTheHolderOfTheTwoImageSeats(t *testing.T) {
 			t.Fatalf("two followers on one ask said %+v (ask id %q), asks %v: want one copy, said once under the ask's id", b.said[copies:], ask.ID, asked)
 		}
 		b.said = b.said[:copies]
+		// registry_mirrors (seams review note 1): a kept copy is read, asked again with its claim let go, and dropped
+		// by hand, which says refused under the ask's id with who and why.
+		_, _ = b.v.Mirror(ImageSeat, "novox.mesh-build-image", MirrorRequest{Source: "nats", Digest: d})
+		asked[key] = MirrorRequest{Source: asked[key].Source, Digest: d, ID: asked[key].ID, Claim: "cut-follower"}
+		if now, err := b.v.MirrorsNow(); err != nil || now[key].Claim != "cut-follower" {
+			t.Fatalf("MirrorsNow = %v, %v; want the claimed ask", now, err)
+		}
+		if r, err := b.v.MirrorAgain(key); err != nil || r.Claim != "" || asked[key].Claim != "" {
+			t.Errorf("MirrorAgain = %+v, %v (kept %+v); want the claim let go", r, err, asked[key])
+		}
+		if _, err := b.v.MirrorDrop(key, "person.jochen", ""); err == nil {
+			t.Error("a drop with no why is taken")
+		}
+		if _, err := b.v.MirrorDrop(key, "person.jochen", "upstream gone"); err != nil || len(asked) != 0 ||
+			b.said[copies].event != EventRefused || b.said[copies].body.ID == "" || !strings.Contains(b.said[copies].body.Reason, "dropped by hand by person.jochen: upstream gone") {
+			t.Errorf("MirrorDrop = %v, said %+v, asks %v; want refused said under the ask's id and the ask gone", err, b.said[copies:], asked)
+		}
+		b.said = b.said[:copies]
 		b.cat.mirrorErr = errors.New("the upstream said 404")
 		_, _ = b.v.Mirror(ImageSeat, "novox.mesh-build-image", MirrorRequest{Source: "nats", Digest: d})
 		if err := b.v.Mirrored(asked[key]); err != nil || b.said[1].event != EventRefused || !strings.Contains(b.said[1].body.Reason, "404") || len(asked) != 0 {

modules/distribution/internal/holder/verbs.go

@@ -46,6 +46,10 @@ type Mirrors interface {
 	// writer wrote the key first.
 	Claim(key string, r MirrorRequest, revision uint64) (bool, error)
 	Delete(key string) error
+	// List is every ask kept, by its key: what registry_mirrors reads.
+	List() (map[string]MirrorRequest, error)
+	// Put writes r under key whatever it held: an ask asked again by hand, its claim let go.
+	Put(key string, r MirrorRequest) error
 }
 
 // HoldState is distribution's own state holds (bus state, template state; ADR 0617 §2).
@@ -353,6 +357,55 @@ func (v *Verbs) Take(key string, r MirrorRequest, revision uint64, current bool,
 	return v.Mirrored(r)
 }
 
+// MirrorsNow lists every copy `mirror` took on and has not ended, by its key, for registry_mirrors: what waits (no
+// claim) and what a follower took on (its claim).
+func (v *Verbs) MirrorsNow() (map[string]MirrorRequest, error) { return v.Mirrors.List() }
+
+// MirrorAgain asks a kept copy again by hand (registry_mirrors, a hand act): its claim is let go, so the state's watch
+// hands it to a follower, which takes it on as a new ask. A copy whose follower was cut, or that a follower holds and
+// does not end, goes on this way without a restart.
+func (v *Verbs) MirrorAgain(key string) (MirrorRequest, error) {
+	all, err := v.Mirrors.List()
+	if err != nil {
+		return MirrorRequest{}, fmt.Errorf("mirror: the asks cannot be read: %w", err)
+	}
+	r, ok := all[key]
+	if !ok {
+		return MirrorRequest{}, fmt.Errorf("mirror: no copy is kept under %s", key)
+	}
+	r.Claim = ""
+	if err := v.Mirrors.Put(key, r); err != nil {
+		return MirrorRequest{}, fmt.Errorf("mirror: %s is not asked again: %w", key, err)
+	}
+	return r, nil
+}
+
+// MirrorDrop drops a kept copy by hand (registry_mirrors, a hand act with who and why): it says refused on the image
+// seat's events with that reason, under the ask's id, so a waiter on the copy ends, and the ask leaves the state.
+func (v *Verbs) MirrorDrop(key, by, why string) (MirrorRequest, error) {
+	if strings.TrimSpace(why) == "" {
+		return MirrorRequest{}, errors.New("mirror: a drop is a hand act: say why")
+	}
+	all, err := v.Mirrors.List()
+	if err != nil {
+		return MirrorRequest{}, fmt.Errorf("mirror: the asks cannot be read: %w", err)
+	}
+	r, ok := all[key]
+	if !ok {
+		return MirrorRequest{}, fmt.Errorf("mirror: no copy is kept under %s", key)
+	}
+	src, err := ParseSource(r.Source)
+	if err != nil {
+		return MirrorRequest{}, err
+	}
+	e := Event{Name: src.Name(), Kind: KindUpstream, Digest: r.Digest, Holder: Self, By: by, ID: r.ID,
+		Reason: fmt.Sprintf("%s@%s was not copied: dropped by hand by %s: %s", r.Source, r.Digest, by, why)}
+	if err := v.Say(ImageSeat, EventRefused, e); err != nil {
+		return MirrorRequest{}, fmt.Errorf("mirror: the drop of %s cannot be said, so the ask stays: %w", key, err)
+	}
+	return r, v.Mirrors.Delete(key)
+}
+
 // HeldNow lists the holds, for registry_holds.
 func (v *Verbs) HeldNow() ([]Hold, error) {
 	all, err := v.Holds.List()

modules/distribution/module.json

@@ -48,6 +48,7 @@
     "registry_health",
     "registry_holds",
     "registry_incoming",
+    "registry_mirrors",
     "registry_old_name"
   ],
   "invokes": [