indexer: a source revision is taken in only by the indexer it names (ADR 0633 §1-§5, issue 1585)

A merged message is kept in the indexer's own state pending and the merges there are computed together on overlays of the indexer's states: taken in whole when the indexer's fingerprint is the running build, otherwise only the indexer's own entry, graph key and verdict are written and the rest waits for the named indexer. A group's members are gathered until its end. Waits are said on the stream take-in, read by indexer_pending and steered by indexer_take_in_pending and indexer_drop_pending. Claude-Session: https://claude.ai/code/session_01JG73KG7AfSdh6FP8VrBqCn
2026-11-04 23:07:02 +01:00
parent ae31b66a9e
commit 071a28f218
11 changed files with 1447 additions and 22 deletions
@@ -8,6 +8,7 @@ package main
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"os"
 	"strings"
@@ -191,8 +192,32 @@ func heldCounts(s stdio.KeptState) func() (index.OldSectionsCount, error) {
 	}
 }

+// keptRead is a state the tools bundle reads, as the index package reads a store: Get and Keys; it writes nothing.
+type keptRead struct{ s stdio.KeptState }
+
+func (k keptRead) Get(_ context.Context, key string) ([]byte, uint64, bool, error) {
+	e, err := k.s.Get(key)
+	if err != nil || e == nil {
+		return nil, 0, false, err
+	}
+	return e.Value, e.Revision, true, nil
+}
+
+func (k keptRead) Keys(context.Context) ([]string, error) { return k.s.Keys() }
+
+func (keptRead) Create(context.Context, string, []byte) error { return errReadOnly }
+
+func (keptRead) Update(context.Context, string, []byte, uint64) error { return errReadOnly }
+
+func (keptRead) Delete(context.Context, string, uint64) error { return errReadOnly }
+
+var errReadOnly = errors.New("the tools bundle only reads this state")
+
 func main() {
-	if err := stdio.Serve("", append(Tools(askTakeIn(stdio.State(index.TakeInsState), 25*time.Second), time.Now, stdio.State(index.VerdictsState), heldCounts(stdio.State(index.OldSectionsState))), catalogTools()...)); err != nil {
+	ask := askTakeIn(stdio.State(index.TakeInsState), 25*time.Second)
+	tools := append(Tools(ask, time.Now, stdio.State(index.VerdictsState), heldCounts(stdio.State(index.OldSectionsState))), catalogTools()...)
+	tools = append(tools, index.PendingTools(keptRead{stdio.State(index.PendingState)}, ask, time.Now)...)
+	if err := stdio.Serve("", tools); err != nil {
 		fmt.Fprintf(os.Stderr, "[mesh-delivery-indexer-tools] %v\n", err)
 		os.Exit(1)
 	}
@@ -16,6 +16,8 @@ type takerIn interface {
 	TakeIn(ctx context.Context, module string) index.TakenIn
 	TakeInRepository(ctx context.Context, repository string) index.TakenIn
 	JudgeAgain(ctx context.Context, module, digest string) index.TakenIn
+	TakeInPendingByHand(ctx context.Context, key, by, why string) index.TakenIn
+	DropPending(ctx context.Context, key, by, why string) index.TakenIn
 }

 // takenIn answers one value of the takeins state (the tool indexer_take_in, issue 1945): a value not yet answered takes the
@@ -34,7 +36,13 @@ func takenIn(ctx context.Context, ix takerIn, kv jetstream.KeyValue, e jetstream
 		return
 	}
 	asked, r := "take-in", index.TakenIn{}
-	if a.Repository != "" {
+	if a.Pending != "" && a.Drop {
+		asked, r = "drop of a pending merge", ix.DropPending(ctx, a.Pending, a.By, a.Why)
+		a.Module = a.Pending
+	} else if a.Pending != "" {
+		asked, r = "hand take-in of what waits", ix.TakeInPendingByHand(ctx, a.Pending, a.By, a.Why)
+		a.Module = a.Pending
+	} else if a.Repository != "" {
 		asked, r = "take-in of a repository", ix.TakeInRepository(ctx, a.Repository)
 		a.Module = a.Repository
 	} else if a.Judge {
@@ -121,6 +121,12 @@ type Indexer struct {
 	ownMu        sync.Mutex  // guards ownFP
 	ownFP        string      // the fingerprint of the build this program runs, asked at the start pass (learnOwnBuild)
 	judgePending atomic.Bool // the start's judge pass failed: judged again at the next message (verdicts.go)
+	// Pending is the state of merges held for the indexer their revision names (pending.go, novox/hq ADR 0633 §1); nil
+	// takes every merge in at once, as before the take-in step.
+	Pending Store
+	// SayTakeIn publishes a take-in event on the indexer's own stream take-in (ADR 0633 §5); nil says none.
+	SayTakeIn func(ctx context.Context, t delivery.TakeIn) error
+	pendingMu sync.Mutex // one take-in step at a time: a merge, a group's end, the start and a hand act (pending.go)
 }

 func (ix *Indexer) logf(format string, args ...any) {
@@ -0,0 +1,110 @@
+package index
+
+import (
+	"context"
+	"maps"
+	"slices"
+	"sync"
+)
+
+// overlayStore is a store whose writes stay in memory over the store it reads through (novox/hq ADR 0633 §1): a merge
+// applied to it is computed whole, its every fingerprint and verdict, and nothing reaches the bus. A key it wrote reads
+// as written, a key it deleted as absent, every other as the store under it holds it. Its compare-and-set holds against
+// what it reads: a key it wrote has a revision of its own, counted above every revision under it. A nil store under
+// it holds nothing.
+type overlayStore struct {
+	under Store
+	mu    sync.Mutex
+	vals  map[string]overlaid
+	next  uint64
+}
+
+type overlaid struct {
+	value    []byte
+	revision uint64
+	deleted  bool
+}
+
+func newOverlay(under Store) *overlayStore {
+	return &overlayStore{under: under, vals: map[string]overlaid{}, next: 1 << 62}
+}
+
+func (o *overlayStore) Get(ctx context.Context, key string) ([]byte, uint64, bool, error) {
+	o.mu.Lock()
+	v, ok := o.vals[key]
+	o.mu.Unlock()
+	if ok {
+		if v.deleted {
+			return nil, v.revision, false, nil
+		}
+		return slices.Clone(v.value), v.revision, true, nil
+	}
+	if o.under == nil {
+		return nil, 0, false, nil
+	}
+	return o.under.Get(ctx, key)
+}
+
+// put checks the revision read against what the key holds now and writes it.
+func (o *overlayStore) put(ctx context.Context, key string, value []byte, revision uint64, create, del bool) error {
+	_, rev, found, err := o.Get(ctx, key)
+	if err != nil {
+		return err
+	}
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	switch {
+	case create && found, !create && (!found || rev != revision):
+		return ErrConflict
+	}
+	o.next++
+	o.vals[key] = overlaid{value: slices.Clone(value), revision: o.next, deleted: del}
+	return nil
+}
+
+func (o *overlayStore) Create(ctx context.Context, key string, value []byte) error {
+	return o.put(ctx, key, value, 0, true, false)
+}
+
+func (o *overlayStore) Update(ctx context.Context, key string, value []byte, revision uint64) error {
+	return o.put(ctx, key, value, revision, false, false)
+}
+
+func (o *overlayStore) Delete(ctx context.Context, key string, revision uint64) error {
+	return o.put(ctx, key, nil, revision, false, true)
+}
+
+func (o *overlayStore) Keys(ctx context.Context) ([]string, error) {
+	var keys []string
+	if o.under != nil {
+		var err error
+		if keys, err = o.under.Keys(ctx); err != nil {
+			return nil, err
+		}
+	}
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	keys = slices.DeleteFunc(keys, func(k string) bool { _, ok := o.vals[k]; return ok })
+	for k, v := range o.vals {
+		if !v.deleted {
+			keys = append(keys, k)
+		}
+	}
+	slices.Sort(keys)
+	return keys, nil
+}
+
+// written is what the overlay holds of key: its value, and whether it wrote or deleted it at all.
+func (o *overlayStore) written(key string) (value []byte, deleted, ok bool) {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	v, ok := o.vals[key]
+	return slices.Clone(v.value), v.deleted, ok
+}
+
+// changed is every key the overlay wrote or deleted, in order.
+func (o *overlayStore) changed() []string {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	return slices.Sorted(maps.Keys(o.vals))
+}
@@ -0,0 +1,837 @@
+package index
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"maps"
+	"slices"
+	"strings"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+	"git.novox.be/novox/mesh-sdk/go/process"
+
+	"git.novox.be/novox/mesh-delivery/sdk/graph"
+)
+
+// The take-in step (novox/hq ADR 0633): a source revision is taken in only by the indexer it names. A merged message is
+// kept in the indexer's own state pending, one key per merge, until the take-in that ends it; a revision naming another
+// build of the indexer writes only what building and moving that indexer needs, and the rest stays pending for it.
+const (
+	// PendingState is the indexer's state of held merges: the bucket mesh-delivery-indexer_pending, keyed
+	// delivery.PendingKey, so the key order is the merge order. An old `state` entry (the indexer is foundation, ADR
+	// 0595 §11), history 1, reached by the indexer alone.
+	PendingState  = "pending"
+	PendingBucket = delivery.IndexerModule + "_" + PendingState
+	// PendingTool reads it (ADR 0633 §5); TakeInPendingTool and DropPendingTool steer it.
+	PendingTool       = "indexer_pending"
+	TakeInPendingTool = "indexer_take_in_pending"
+	DropPendingTool   = "indexer_drop_pending"
+)
+
+// errClaimLost says another take-in claimed a pending merge first: this one took nothing in.
+var errClaimLost = errors.New("another take-in claimed the merges first, so this one took nothing in")
+
+// NoBuild opens what a revision waits for when it gives the indexer no fingerprint: the reason follows. No indexer
+// runs such a build, so the revision waits until a later merge gives it one or a hand take-in (indexer_take_in_pending).
+const NoBuild = "no build: "
+
+// ErrOwnBuildUnknown is what a merged message answers while the indexer cannot learn the build it runs (ADR 0633 §1):
+// the message is left unacknowledged, so the bus delivers it again, and nothing is written.
+var ErrOwnBuildUnknown = errors.New("the build of the indexer this program runs is not known")
+
+// Pending is one value of the pending state: the merged message whole, as the forge bridge encoded it, the group it
+// was merged in (Ended once the group's end was heard), since when it is held, the indexer build its revision waits
+// for (absent while it waits for nothing yet, or behind a group being gathered), the build of the indexer that last
+// wrote it, the wait it was last said under on the take-in stream (Said, SaidAt), and Taking, set by a take-in that
+// claimed it: a second indexer that finds it claimed takes nothing in (ADR 0633 §1: never two take-ins of one merge).
+type Pending struct {
+	V          int             `json:"v"`
+	Forge      json.RawMessage `json:"forge"`
+	Group      string          `json:"group,omitempty"`
+	Ended      bool            `json:"ended,omitempty"`
+	Since      time.Time       `json:"since"`
+	WaitsFor   string          `json:"waits-for,omitempty"`
+	LastWriter string          `json:"last-writer"`
+	Said       string          `json:"said,omitempty"`
+	SaidKind   string          `json:"said-kind,omitempty"`
+	SaidAt     time.Time       `json:"said-at,omitzero"`
+	Taking     string          `json:"taking,omitempty"`
+}
+
+// PendingEntry is one key of the pending state as read: its key, revision, value and the merged message decoded.
+type PendingEntry struct {
+	Key      string
+	Revision uint64
+	Value    Pending
+	Forge    delivery.Forge
+}
+
+// DecodePending reads one pending value and its merged message.
+func DecodePending(data []byte) (Pending, delivery.Forge, error) {
+	var p Pending
+	if err := json.Unmarshal(data, &p); err != nil {
+		return p, delivery.Forge{}, fmt.Errorf("not a pending value: %w", err)
+	}
+	if p.V != 1 {
+		return p, delivery.Forge{}, fmt.Errorf("a pending value of version %d, not 1", p.V)
+	}
+	f, err := delivery.DecodeForge(p.Forge)
+	if err != nil {
+		return p, f, fmt.Errorf("its merged message: %w", err)
+	}
+	if f.Kind != delivery.ForgeMerged {
+		return p, f, fmt.Errorf("a pending value holds a %s message, not a merged one", f.Kind)
+	}
+	return p, f, nil
+}
+
+func encodePending(p Pending) ([]byte, error) {
+	p.V = 1
+	return json.Marshal(p)
+}
+
+// ReadPending is every value the pending state holds, in key order (the merge order); a key that does not read is
+// said through logf and left.
+func ReadPending(ctx context.Context, st Store, logf func(string, ...any)) ([]PendingEntry, error) {
+	keys, err := st.Keys(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("listing %s: %w", PendingState, err)
+	}
+	slices.Sort(keys)
+	var out []PendingEntry
+	for _, key := range keys {
+		value, rev, found, err := st.Get(ctx, key)
+		if err != nil {
+			return nil, fmt.Errorf("%s.%s: %w", PendingState, key, err)
+		}
+		if !found {
+			continue
+		}
+		p, f, err := DecodePending(value)
+		if err != nil {
+			if logf != nil {
+				logf("%s.%s is left: %v", PendingState, key, err)
+			}
+			continue
+		}
+		out = append(out, PendingEntry{Key: key, Revision: rev, Value: p, Forge: f})
+	}
+	return out, nil
+}
+
+// Waits is the pending state read as what waits (ADR 0633 §3, §4): Ready are the merges no group still being gathered
+// holds back, in merge order, which are one revision together; Groups are the groups still being gathered, each with
+// its members held so far; Behind are the merges in a gathered group's member repositories that wait behind it, by the
+// group's take-in key.
+type Waits struct {
+	Ready  []PendingEntry
+	Groups map[string][]PendingEntry
+	Behind map[string][]PendingEntry
+}
+
+// WaitsOf works out what waits from the entries, in key order. A member of a group whose end was not heard is
+// gathered; from the first such member on, a later merge in one of the group's members' repositories waits behind it,
+// because its tree holds that member's commit. Every other merge is ready.
+func WaitsOf(entries []PendingEntry) Waits {
+	w := Waits{Groups: map[string][]PendingEntry{}, Behind: map[string][]PendingEntry{}}
+	blocked := map[string]string{} // repository, lower case, to the group key it waits behind
+	for _, e := range entries {
+		if e.Value.Group != "" && !e.Value.Ended {
+			w.Groups[e.Value.Group] = append(w.Groups[e.Value.Group], e)
+			for _, m := range e.Forge.Members {
+				if _, ok := blocked[strings.ToLower(m.Repository)]; !ok {
+					blocked[strings.ToLower(m.Repository)] = delivery.GroupTakeInKey(e.Value.Group)
+				}
+			}
+			continue
+		}
+		if g, ok := blocked[strings.ToLower(e.Forge.Repository)]; ok {
+			w.Behind[g] = append(w.Behind[g], e)
+			continue
+		}
+		w.Ready = append(w.Ready, e)
+	}
+	return w
+}
+
+// mergesOf are the merges of entries, oldest first.
+func mergesOf(entries []PendingEntry) []delivery.GroupMerge {
+	out := make([]delivery.GroupMerge, len(entries))
+	for i, e := range entries {
+		out[i] = delivery.GroupMerge{Repository: e.Forge.Repository, Number: e.Forge.Number, Merge: e.Forge.Merge}
+	}
+	return out
+}
+
+func sinceOf(entries []PendingEntry) time.Time {
+	var at time.Time
+	for _, e := range entries {
+		if at.IsZero() || e.Value.Since.Before(at) {
+			at = e.Value.Since
+		}
+	}
+	return at
+}
+
+// groupWait is the take-in body of a group being gathered.
+func groupWait(kind, group string, held []PendingEntry) delivery.TakeIn {
+	t := delivery.TakeIn{Kind: kind, Key: delivery.GroupTakeInKey(group), What: delivery.TakeInGroup, Group: group,
+		Merges: mergesOf(held), Since: sinceOf(held)}
+	if len(held) > 0 {
+		t.Members = held[len(held)-1].Forge.Members
+	}
+	return t
+}
+
+// revisionWait is the take-in body of a revision: its key is its oldest merge's.
+func revisionWait(kind string, ready []PendingEntry, waitsFor string) delivery.TakeIn {
+	return delivery.TakeIn{Kind: kind, Key: ready[0].Key, What: delivery.TakeInRevision, Merges: mergesOf(ready),
+		WaitsFor: waitsFor, Since: sinceOf(ready)}
+}
+
+// ownBuildNow is the build this program runs: the one learnt, else asked of the node-engine now.
+func (ix *Indexer) ownBuildNow(ctx context.Context) string {
+	ix.ownMu.Lock()
+	fp := ix.ownFP
+	ix.ownMu.Unlock()
+	if fp != "" || ix.OwnBuild == nil {
+		return fp
+	}
+	ix.learnOwnBuild(ctx)
+	ix.ownMu.Lock()
+	defer ix.ownMu.Unlock()
+	return ix.ownFP
+}
+
+// sayTakeIn publishes a take-in event and keeps what it said on the entries, so indexer_pending reads the last event of
+// each wait from the state. A publish that fails is said in the log: the next evaluation says it again.
+func (ix *Indexer) sayTakeIn(ctx context.Context, t delivery.TakeIn) bool {
+	if ix.SayTakeIn == nil {
+		return true
+	}
+	if err := ix.SayTakeIn(ctx, t); err != nil {
+		ix.logf("take-in: %s.%s is not said: %v", t.Kind, t.Key, err)
+		return false
+	}
+	return true
+}
+
+// markSaid writes on each entry the wait it was said under, by compare-and-set; a lost one is left (its next
+// evaluation says it again).
+func (ix *Indexer) markSaid(ctx context.Context, entries []PendingEntry, t *delivery.TakeIn, change func(*Pending)) {
+	for _, e := range entries {
+		p := e.Value
+		if t != nil {
+			p.Said, p.SaidKind, p.SaidAt = t.Key, t.Kind, ix.now()
+		}
+		if change != nil {
+			change(&p)
+		}
+		data, err := encodePending(p)
+		if err != nil {
+			continue
+		}
+		if err := ix.Pending.Update(ctx, e.Key, data, e.Revision); err != nil {
+			ix.logf("take-in: %s.%s is not written again: %v", PendingState, e.Key, err)
+		}
+	}
+}
+
+// Arrive takes one merged message of the forge bridge's stream in under ADR 0633: seq is its stream sequence. It is
+// written to pending first (a redelivered message finds its key there and writes no second), and then what pending
+// holds is evaluated (Evaluate): taken in whole when its revision names the build this program runs, or held for the
+// indexer it names. It answers what was taken in, and a line saying what waits. A message is never written while the
+// build this program runs is not known (ErrOwnBuildUnknown).
+func (ix *Indexer) Arrive(ctx context.Context, seq uint64, f delivery.Forge) (Applied, string, error) {
+	if ix.Pending == nil || ix.Modules == nil {
+		a, err := ix.ApplyMessage(ctx, f)
+		return a, "", err
+	}
+	own := ix.ownBuildNow(ctx)
+	if own == "" {
+		return Applied{}, "", ErrOwnBuildUnknown
+	}
+	ix.pendingMu.Lock()
+	defer ix.pendingMu.Unlock()
+	raw, err := delivery.EncodeForge(f)
+	if err != nil {
+		return Applied{}, "", fmt.Errorf("%w: %v", ErrUnreadable, err)
+	}
+	key := delivery.PendingKey(seq)
+	data, err := encodePending(Pending{Forge: raw, Group: f.Group, Since: ix.now(), LastWriter: own})
+	if err != nil {
+		return Applied{}, "", err
+	}
+	if err := ix.Pending.Create(ctx, key, data); err != nil && !errors.Is(err, ErrConflict) {
+		return Applied{}, "", fmt.Errorf("%s.%s: %w", PendingState, key, err)
+	}
+	if f.Group != "" {
+		entries, err := ReadPending(ctx, ix.Pending, ix.logf)
+		if err != nil {
+			return Applied{}, "", err
+		}
+		if held := WaitsOf(entries).Groups[f.Group]; len(held) > 0 {
+			t := groupWait(delivery.TakeInPending, f.Group, held)
+			if ix.sayTakeIn(ctx, t) {
+				ix.markSaid(ctx, held, &t, nil)
+			}
+		}
+	}
+	return ix.evaluate(ctx, own, false)
+}
+
+// EndGroup takes a group's end in (ADR 0633 §4): once every member it names is held, the group's members are marked
+// ended, the group's wait is said taken (take-in.taken under its group key), and pending is evaluated, the group now
+// one revision. A member named and neither held nor taken in yet answers an error, so the end is delivered again once
+// the member arrived. An end of a group taken in already (every member's merge at or behind its repository's taken
+// merge) changes nothing and is said: it is never delivered again for ever.
+func (ix *Indexer) EndGroup(ctx context.Context, end delivery.ForgeGroupEnd) (Applied, string, error) {
+	if ix.Pending == nil {
+		return Applied{}, "no pending state: the group's merges were taken in one by one", nil
+	}
+	own := ix.ownBuildNow(ctx)
+	if own == "" {
+		return Applied{}, "", ErrOwnBuildUnknown
+	}
+	ix.pendingMu.Lock()
+	defer ix.pendingMu.Unlock()
+	entries, err := ReadPending(ctx, ix.Pending, ix.logf)
+	if err != nil {
+		return Applied{}, "", err
+	}
+	var held []PendingEntry
+	for _, e := range entries {
+		if e.Value.Group == end.Group {
+			held = append(held, e)
+		}
+	}
+	isHeld := func(m delivery.GroupMerge) bool {
+		return slices.ContainsFunc(held, func(e PendingEntry) bool {
+			return delivery.SameRepository(e.Forge.Repository, m.Repository) && e.Forge.Number == m.Number
+		})
+	}
+	var missing []string
+	for _, m := range end.Members {
+		if isHeld(m) {
+			continue
+		}
+		taken, err := ix.takenAlready(ctx, m)
+		if err != nil {
+			return Applied{}, "", err
+		}
+		if !taken {
+			missing = append(missing, fmt.Sprintf("%s#%d", m.Repository, m.Number))
+		}
+	}
+	if len(missing) > 0 {
+		return Applied{}, "", fmt.Errorf("the end of the group %q names %s, not held yet: it waits for their merged messages", end.Group, strings.Join(missing, ", "))
+	}
+	if len(held) == 0 {
+		ix.logf("take-in: the end of the group %q comes after its take-in: every member is taken in already, nothing changes", end.Group)
+		return Applied{}, "the group is taken in already", nil
+	}
+	if !slices.ContainsFunc(held, func(e PendingEntry) bool { return !e.Value.Ended }) {
+		return ix.evaluate(ctx, own, false)
+	}
+	// The group's wait ends first (take-in.taken under its group key); a revision it then is, pending for another
+	// indexer, is said pending under its oldest merge's key by the evaluation that follows.
+	t := groupWait(delivery.TakeInTaken, end.Group, held)
+	said := &t
+	if !ix.sayTakeIn(ctx, t) {
+		said = nil // said again by the take-in or the hold that follows (saidWaits)
+	}
+	ix.markSaid(ctx, held, said, func(p *Pending) { p.Ended = true })
+	ix.logf("take-in: the group %q ended with %d members held: one revision now", end.Group, len(held))
+	return ix.evaluate(ctx, own, false)
+}
+
+// takenAlready says whether a group member's merge is at or behind its repository's taken merge.
+func (ix *Indexer) takenAlready(ctx context.Context, m delivery.GroupMerge) (bool, error) {
+	if ix.Taken == nil {
+		return false, nil
+	}
+	key, err := delivery.RepositoryKey(m.Repository)
+	if err != nil {
+		return false, err
+	}
+	value, _, found, err := ix.Taken.Get(ctx, key)
+	if errors.Is(err, process.ErrNotYet) || (err == nil && !found) {
+		return false, nil
+	}
+	if err != nil {
+		return false, fmt.Errorf("taken.%s: %w", key, err)
+	}
+	cur, err := delivery.DecodeTaken(value)
+	if err != nil || cur.Merge == "" {
+		return false, nil
+	}
+	if cur.Merge == m.Merge {
+		return true, nil
+	}
+	return ix.older(ctx, m.Repository, m.Merge, cur.Merge)
+}
+
+// TakeInAtStart is the take-in at the start (ADR 0633 §2), before the consumer on the forge bridge's stream starts:
+// it asks the build this program runs until it is known (wait says when to ask again, false ends it), then evaluates
+// pending, taking in the whole revisions that name this build; a group still being gathered stays, and a revision
+// naming yet another build is held for that one. A claim left by an indexer that ran before is void here.
+func (ix *Indexer) TakeInAtStart(ctx context.Context, wait func(context.Context) bool) (Applied, string, error) {
+	if ix.Pending == nil {
+		return Applied{}, "", nil
+	}
+	own := ix.ownBuildNow(ctx)
+	for own == "" {
+		ix.logf("take-in: the build this program runs is not known yet, so nothing pending is taken in; asked again")
+		if !wait(ctx) {
+			return Applied{}, "", ErrOwnBuildUnknown
+		}
+		own = ix.ownBuildNow(ctx)
+	}
+	ix.pendingMu.Lock()
+	defer ix.pendingMu.Unlock()
+	return ix.evaluate(ctx, own, true)
+}
+
+// evaluate works out what pending's ready merges name, together (ADR 0633 §3), and takes them in or holds them.
+func (ix *Indexer) evaluate(ctx context.Context, own string, start bool) (Applied, string, error) {
+	entries, err := ReadPending(ctx, ix.Pending, ix.logf)
+	if err != nil {
+		return Applied{}, "", err
+	}
+	w := WaitsOf(entries)
+	if len(w.Ready) == 0 {
+		return Applied{}, ix.waitingLine(w), nil
+	}
+	if !start {
+		for _, e := range w.Ready {
+			if e.Value.Taking != "" {
+				ix.logf("take-in: %s.%s is claimed by the take-in %s, so this indexer takes nothing in", PendingState, e.Key, e.Value.Taking)
+				return Applied{}, "claimed by another take-in", nil
+			}
+		}
+	}
+	names, dry, err := ix.computeOver(ctx, w.Ready)
+	if err != nil {
+		return Applied{}, "", err
+	}
+	if names == own {
+		a, err := ix.takeInWhole(ctx, w.Ready, own, "")
+		if errors.Is(err, errClaimLost) {
+			return a, "claimed by another take-in", nil
+		}
+		return a, ix.waitingLine(w), err
+	}
+	return Applied{}, ix.waitingLine(w), ix.holdFor(ctx, w.Ready, names, own, dry)
+}
+
+// waitingLine says what waits after an evaluation, for the heard state.
+func (ix *Indexer) waitingLine(w Waits) string {
+	var parts []string
+	for _, g := range slices.Sorted(maps.Keys(w.Groups)) {
+		parts = append(parts, fmt.Sprintf("the group %q gathered (%d held)", g, len(w.Groups[g])))
+	}
+	for _, g := range slices.Sorted(maps.Keys(w.Behind)) {
+		parts = append(parts, fmt.Sprintf("%d merges behind %s", len(w.Behind[g]), g))
+	}
+	return strings.Join(parts, "; ")
+}
+
+// dryRun is an indexer computing a take-in without writing it, and the overlays of what it would have written.
+type dryRun struct {
+	ix                                     *Indexer
+	modules, graph, verdicts               *overlayStore
+	candidates, taken, predefined, configs *overlayStore
+	oldSections                            *overlayStore
+}
+
+// dry is an indexer over overlays of this one's states (overlay.go): what a take-in would write stays in memory and
+// nothing reaches the bus but the manifests and the packages the revision publishes, which clause 1 writes all the
+// same. It says no event, kicks nothing in the background, and knows what this one knows of the source, the toolchains
+// and the previews. Every field of Indexer is named here (TestDryNamesEveryField holds it so).
+func (ix *Indexer) dry() *dryRun {
+	over := func(st Store) (Store, *overlayStore) {
+		if st == nil {
+			return nil, nil
+		}
+		o := newOverlay(st)
+		return o, o
+	}
+	d := &dryRun{}
+	x := &Indexer{Manifests: ix.Manifests, Packages: ix.Packages, JudgeID: ix.JudgeID, Files: ix.Files, GoSum: ix.GoSum,
+		GoMod: ix.GoMod, Now: ix.Now, Builds: ix.Builds, Named: ix.Named, Trunk: ix.Trunk, Read: ix.Read, Contains: ix.Contains,
+		ReadOpen: nil, OwnBuild: nil, SayRefused: nil, SayAccepted: nil, SayTakeIn: nil, Pending: nil}
+	x.Logf = func(format string, args ...any) { ix.logf("take-in, computed and not written: "+format, args...) }
+	x.Store, d.candidates = over(ix.Store)
+	x.Modules, d.modules = over(ix.Modules)
+	x.Graph, d.graph = over(ix.Graph)
+	x.Verdicts, d.verdicts = over(ix.Verdicts)
+	x.Taken, d.taken = over(ix.Taken)
+	x.Predefined, d.predefined = over(ix.Predefined)
+	x.ConfigCandidates, d.configs = over(ix.ConfigCandidates)
+	x.OldSections, d.oldSections = over(ix.OldSections)
+	ix.mu.Lock()
+	x.source = sourceNow{rev: ix.source.rev, heads: maps.Clone(ix.source.heads)}
+	x.previews = maps.Clone(ix.previews)
+	x.toolchains = map[string]map[string]delivery.Toolchain{}
+	for l, by := range ix.toolchains {
+		x.toolchains[l] = maps.Clone(by)
+	}
+	x.moving = maps.Clone(ix.moving)
+	x.untaken = maps.Clone(ix.untaken)
+	ix.mu.Unlock()
+	ix.ownMu.Lock()
+	x.ownFP = ix.ownFP
+	ix.ownMu.Unlock()
+	d.ix = x
+	return d
+}
+
+// computeOver applies the ready merges, in merge order, to a dry indexer, as one overlay: each later merge's trees
+// govern (ADR 0633 §3). It answers the fingerprint the indexer's own module has once they are applied (none is "",
+// with its reason), and the dry run, whose overlays hold what the take-in would write. A merge that can never be read
+// is left out, as its take-in will drop it.
+func (ix *Indexer) computeOver(ctx context.Context, ready []PendingEntry) (string, *dryRun, error) {
+	d := ix.dry()
+	for _, e := range ready {
+		if _, err := d.ix.ApplyMessage(ctx, e.Forge); err != nil {
+			if errors.Is(err, ErrUnreadable) {
+				continue
+			}
+			return "", d, fmt.Errorf("computing the take-in of %s (%s#%d): %w", e.Key, e.Forge.Repository, e.Forge.Number, err)
+		}
+	}
+	key, _ := delivery.ModulesKey(delivery.IndexerModule)
+	value, _, found, err := d.ix.Modules.Get(ctx, key)
+	if err != nil {
+		return "", d, err
+	}
+	if !found {
+		return NoBuild + "the indexer holds no record of itself", d, nil
+	}
+	m, err := delivery.DecodeModule(value)
+	if err != nil {
+		return "", d, nil
+	}
+	if m.FP2 == "" {
+		return NoBuild + m.None2, d, nil
+	}
+	return m.FP2, d, nil
+}
+
+// claim marks the entries as taken in by this take-in, by compare-and-set on the revisions read; a lost one leaves
+// every claim made and answers false, and nothing is taken in.
+func (ix *Indexer) claim(ctx context.Context, entries []PendingEntry, own, by string) ([]PendingEntry, bool) {
+	id := fmt.Sprintf("%s at %s", own, ix.now().Format(time.RFC3339Nano))
+	if by != "" {
+		id = by + ": " + id
+	}
+	claimed := make([]PendingEntry, 0, len(entries))
+	for _, e := range entries {
+		p := e.Value
+		p.Taking, p.LastWriter = id, own
+		data, err := encodePending(p)
+		if err == nil {
+			err = ix.Pending.Update(ctx, e.Key, data, e.Revision)
+		}
+		if err != nil {
+			ix.logf("take-in: the claim of %s.%s is lost (%v): another take-in took it, so this one takes nothing in", PendingState, e.Key, err)
+			return nil, false
+		}
+		_, rev, _, err := ix.Pending.Get(ctx, e.Key)
+		if err != nil {
+			return nil, false
+		}
+		e.Revision, e.Value = rev, p
+		claimed = append(claimed, e)
+	}
+	return claimed, true
+}
+
+// takeInWhole takes the entries in with this indexer, as merges are taken in (ApplyMessage), oldest first, then
+// removes them from pending and says each wait they were said under taken. by names a hand act ("" for none).
+func (ix *Indexer) takeInWhole(ctx context.Context, entries []PendingEntry, own, by string) (Applied, error) {
+	return ix.takeInWholeBy(ctx, entries, own, by, "", "")
+}
+
+// takeInWholeBy is takeInWhole; who and why, of a hand act, are carried on the take-in.taken events.
+func (ix *Indexer) takeInWholeBy(ctx context.Context, entries []PendingEntry, own, by, who, why string) (Applied, error) {
+	claimed, ok := ix.claim(ctx, entries, own, by)
+	if !ok {
+		return Applied{}, errClaimLost
+	}
+	var applied Applied
+	for _, e := range claimed {
+		a, err := ix.ApplyMessage(ctx, e.Forge)
+		applied.Written = append(applied.Written, a.Written...)
+		if a.Late != "" {
+			applied.Late = a.Late
+		}
+		if errors.Is(err, ErrUnreadable) {
+			ix.logf("take-in: %s (%s#%d) can never be read, dropped: %v", e.Key, e.Forge.Repository, e.Forge.Number, err)
+			continue
+		}
+		if err != nil {
+			return applied, fmt.Errorf("taking %s (%s#%d) in: %w", e.Key, e.Forge.Repository, e.Forge.Number, err)
+		}
+	}
+	for _, e := range claimed {
+		if err := ix.Pending.Delete(ctx, e.Key, e.Revision); err != nil {
+			ix.logf("take-in: %s.%s is taken in and not removed: %v", PendingState, e.Key, err)
+		}
+	}
+	ix.logf("take-in: %d merges taken in as one revision by %s%s, oldest %s", len(claimed), own, byWords(by), claimed[0].Key)
+	for _, t := range saidWaits(claimed) {
+		t.By, t.Why = who, why
+		ix.sayTakeIn(ctx, t)
+	}
+	return applied, nil
+}
+
+func byWords(by string) string {
+	if by == "" {
+		return ""
+	}
+	return " (a hand act: " + by + ")"
+}
+
+// saidWaits are the taken events of every wait the entries were said under that is not ended yet.
+func saidWaits(entries []PendingEntry) []delivery.TakeIn {
+	by := map[string][]PendingEntry{}
+	for _, e := range entries {
+		if e.Value.Said != "" && e.Value.SaidKind == delivery.TakeInPending {
+			by[e.Value.Said] = append(by[e.Value.Said], e)
+		}
+	}
+	var out []delivery.TakeIn
+	for _, key := range slices.Sorted(maps.Keys(by)) {
+		held := by[key]
+		if strings.HasPrefix(key, "g.") {
+			out = append(out, groupWait(delivery.TakeInTaken, held[0].Value.Group, held))
+			continue
+		}
+		t := revisionWait(delivery.TakeInTaken, held, held[0].Value.WaitsFor)
+		t.Key = key
+		out = append(out, t)
+	}
+	return out
+}
+
+// holdFor keeps the ready merges pending for the indexer build they name and writes only what building and moving
+// that indexer needs (ADR 0633 §1): the indexer's own module entry, its key of the module graph and the verdict on its
+// manifest, from the dry run; the manifests and the packages are written by the dry run itself. The revision is said
+// pending under its oldest merge's key, and a wait it was said under before by another key is said taken.
+func (ix *Indexer) holdFor(ctx context.Context, ready []PendingEntry, names, own string, d *dryRun) error {
+	key, _ := delivery.ModulesKey(delivery.IndexerModule)
+	if err := copyKey(ctx, d.modules, ix.Modules, key); err != nil {
+		return fmt.Errorf("the indexer's own module entry: %w", err)
+	}
+	if gk, err := graph.Key(delivery.IndexerModule); err == nil {
+		if err := copyKey(ctx, d.graph, ix.Graph, gk); err != nil && !errors.Is(err, process.ErrNotYet) {
+			return fmt.Errorf("the indexer's key of the module graph: %w", err)
+		}
+	}
+	if value, _, found, err := d.ix.Modules.Get(ctx, key); err == nil && found {
+		if m, err := delivery.DecodeModule(value); err == nil && m.Manifest != "" {
+			if vk, err := delivery.VerdictsKey(m.Manifest); err == nil {
+				if err := copyKey(ctx, d.verdicts, ix.Verdicts, vk); err != nil && !errors.Is(err, process.ErrNotYet) {
+					return fmt.Errorf("the verdict on the indexer's manifest: %w", err)
+				}
+			}
+		}
+	}
+	t := revisionWait(delivery.TakeInPending, ready, names)
+	var ended, unmarked []PendingEntry
+	said, changed := false, false
+	for _, e := range ready {
+		switch {
+		case e.Value.Said == t.Key && e.Value.SaidKind == delivery.TakeInPending:
+			said = true
+			changed = changed || e.Value.WaitsFor != names
+			if e.Value.WaitsFor != names {
+				unmarked = append(unmarked, e)
+			}
+		case e.Value.Said != "" && e.Value.SaidKind == delivery.TakeInPending:
+			ended = append(ended, e)
+			unmarked = append(unmarked, e)
+		default:
+			unmarked = append(unmarked, e)
+		}
+	}
+	for _, old := range saidWaits(ended) {
+		ix.sayTakeIn(ctx, old)
+	}
+	if !said || changed {
+		ix.logf("take-in: %d merges, oldest %s, name the indexer build %s, not %s, which runs: they wait for it", len(ready), t.Key, names, own)
+		if !ix.sayTakeIn(ctx, t) {
+			ix.markSaid(ctx, unmarked, nil, func(p *Pending) { p.WaitsFor, p.LastWriter = names, own })
+			return nil
+		}
+		unmarked = ready
+	}
+	ix.markSaid(ctx, unmarked, &t, func(p *Pending) { p.WaitsFor, p.LastWriter = names, own })
+	return nil
+}
+
+// copyKey writes what the overlay holds of key to the store, when it wrote or deleted it.
+func copyKey(ctx context.Context, from *overlayStore, to Store, key string) error {
+	if from == nil || to == nil {
+		return nil
+	}
+	value, deleted, ok := from.written(key)
+	if !ok {
+		return nil
+	}
+	return write(ctx, to, key, func(cur []byte, found bool) ([]byte, bool, error) {
+		switch {
+		case deleted:
+			return nil, found, nil
+		case found && string(cur) == string(value):
+			return nil, false, nil
+		}
+		return value, false, nil
+	})
+}
+
+// PendingWaits is what indexer_pending answers of the pending state's entries (ADR 0633 §5): the ready merges as one
+// revision, each group being gathered, and the merges behind a group, each with the modules it holds back and the
+// last take-in event said of it.
+func PendingWaits(entries []PendingEntry) []delivery.PendingWait {
+	w := WaitsOf(entries)
+	var out []delivery.PendingWait
+	if len(w.Ready) > 0 {
+		out = append(out, waitOf(w.Ready, delivery.TakeInRevision, ""))
+	}
+	for _, g := range slices.Sorted(maps.Keys(w.Groups)) {
+		out = append(out, waitOf(w.Groups[g], delivery.TakeInGroup, ""))
+	}
+	for _, g := range slices.Sorted(maps.Keys(w.Behind)) {
+		out = append(out, waitOf(w.Behind[g], delivery.TakeInRevision, g))
+	}
+	slices.SortStableFunc(out, func(a, b delivery.PendingWait) int { return a.Since.Compare(b.Since) })
+	return out
+}
+
+func waitOf(entries []PendingEntry, what, behind string) delivery.PendingWait {
+	pw := delivery.PendingWait{Key: entries[0].Key, What: what, Merges: mergesOf(entries), Behind: behind, Since: sinceOf(entries),
+		WaitsFor: entries[len(entries)-1].Value.WaitsFor, Holds: []string{}}
+	if what == delivery.TakeInGroup {
+		pw.Group = entries[0].Value.Group
+		pw.Key = delivery.GroupTakeInKey(pw.Group)
+		pw.Members = entries[len(entries)-1].Forge.Members
+		pw.WaitsFor = ""
+	}
+	for _, e := range entries {
+		pw.Keys = append(pw.Keys, e.Key)
+		for _, raw := range e.Forge.Manifests {
+			if name := manifestModule(raw); name != "" && !slices.Contains(pw.Holds, name) {
+				pw.Holds = append(pw.Holds, name)
+			}
+		}
+		if e.Value.Said != "" && e.Value.SaidAt.After(pw.SaidAt) {
+			pw.Said, pw.SaidAt = delivery.TakeInStream+"."+e.Value.SaidKind+"."+e.Value.Said, e.Value.SaidAt
+		}
+	}
+	slices.Sort(pw.Holds)
+	return pw
+}
+
+// TakeInPendingByHand is the hand take-in (indexer_take_in_pending, ADR 0633 §5): the wait named by its key, the
+// ready revision (its m. key) or a group gathered so far (its g. key), is taken in with this indexer and its judge,
+// whatever build it names, for when the named indexer can never run or the group's end never comes. by and why are
+// said in the journal and on the take-in.taken event. What waited behind it is evaluated again after.
+func (ix *Indexer) TakeInPendingByHand(ctx context.Context, key, by, why string) TakenIn {
+	r := TakenIn{Module: key}
+	if ix.Pending == nil {
+		r.Answer = "the pending state is not on the bus: nothing waits"
+		return r
+	}
+	own := ix.ownBuildNow(ctx)
+	ix.pendingMu.Lock()
+	defer ix.pendingMu.Unlock()
+	entries, err := ReadPending(ctx, ix.Pending, ix.logf)
+	if err != nil {
+		r.Answer = err.Error()
+		return r
+	}
+	w := WaitsOf(entries)
+	var take []PendingEntry
+	switch {
+	case strings.HasPrefix(key, "g."):
+		for g, held := range w.Groups {
+			if delivery.GroupTakeInKey(g) == key {
+				take = held
+			}
+		}
+	case len(w.Ready) > 0 && w.Ready[0].Key == key:
+		take = w.Ready
+	}
+	if len(take) == 0 {
+		r.Answer = fmt.Sprintf("nothing waits under %s: indexer_pending names the waits", key)
+		return r
+	}
+	ix.logf("take-in: a hand take-in of %s (%d merges) asked by %s: %s", key, len(take), by, why)
+	hand := fmt.Sprintf("%s (%s)", by, why)
+	if _, err := ix.takeInWholeBy(ctx, take, own, hand, by, why); err != nil {
+		r.Answer = err.Error()
+		return r
+	}
+	r.Stored = true
+	r.Answer = fmt.Sprintf("%d merges taken in by the running indexer %s", len(take), own)
+	if _, _, err := ix.evaluate(ctx, own, false); err != nil {
+		r.Answer += "; what waited behind it is not evaluated: " + err.Error()
+	}
+	return r
+}
+
+// DropPending is the hand drop (indexer_drop_pending, ADR 0633 §5): the one pending merge under key, abandoned
+// because a later merge replaced it, is removed and nothing of it is taken in. Its wait is said taken, naming the drop;
+// the merges left under that wait are said pending again by the evaluation that follows.
+func (ix *Indexer) DropPending(ctx context.Context, key, by, why string) TakenIn {
+	r := TakenIn{Module: key}
+	if ix.Pending == nil {
+		r.Answer = "the pending state is not on the bus: nothing waits"
+		return r
+	}
+	own := ix.ownBuildNow(ctx)
+	ix.pendingMu.Lock()
+	defer ix.pendingMu.Unlock()
+	entries, err := ReadPending(ctx, ix.Pending, ix.logf)
+	if err != nil {
+		r.Answer = err.Error()
+		return r
+	}
+	at := slices.IndexFunc(entries, func(e PendingEntry) bool { return e.Key == key })
+	if at < 0 {
+		r.Answer = fmt.Sprintf("%s holds no merge under %s", PendingState, key)
+		return r
+	}
+	e := entries[at]
+	if err := ix.Pending.Delete(ctx, e.Key, e.Revision); err != nil {
+		r.Answer = fmt.Sprintf("%s.%s is not removed: %v", PendingState, key, err)
+		return r
+	}
+	ix.logf("take-in: %s (%s#%d at %s) dropped by hand, asked by %s: %s; nothing of it is taken in", key, e.Forge.Repository, e.Forge.Number, e.Forge.Merge, by, why)
+	if e.Value.Said != "" && e.Value.SaidKind == delivery.TakeInPending {
+		var same []PendingEntry
+		for _, o := range entries {
+			if o.Key != key && o.Value.Said == e.Value.Said {
+				same = append(same, o)
+			}
+		}
+		for _, t := range saidWaits([]PendingEntry{e}) {
+			t.By, t.Why, t.Dropped = by, why, true
+			ix.sayTakeIn(ctx, t)
+		}
+		ix.markSaid(ctx, same, nil, func(p *Pending) { p.Said, p.SaidKind = "", "" })
+	}
+	r.Stored, r.Answer = true, fmt.Sprintf("%s dropped: %s#%d at %s is not taken in", key, e.Forge.Repository, e.Forge.Number, e.Forge.Merge)
+	if own != "" {
+		if _, _, err := ix.evaluate(ctx, own, false); err != nil {
+			r.Answer += "; what waits is not evaluated again: " + err.Error()
+		}
+	}
+	return r
+}
@@ -0,0 +1,321 @@
+package index
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"reflect"
+	"slices"
+	"strings"
+	"testing"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+// takeInBed is an indexer holding the catalogue's avahi and the indexer's own module, both with a fingerprint, its
+// pending state and the take-in events it says; base is the catalogue merge it was built from.
+type takeInBed struct {
+	ix      *Indexer
+	mods    *memStore
+	pending *memStore
+	said    []delivery.TakeIn
+	base    delivery.Forge
+	logs    *strings.Builder
+}
+
+func newTakeInBed(t *testing.T) *takeInBed {
+	t.Helper()
+	ix, mods, f := newFormIndexerUnmerged(t, mod{"avahi", catalogueManifest(t, "avahi")},
+		mod{delivery.IndexerModule, renamed(t, "avahi", delivery.IndexerModule)})
+	if err := ix.Apply(context.Background(), f); err != nil {
+		t.Fatal(err)
+	}
+	reportToolchain(t, ix, "mesh-build-go", "novox", "go", "go1.27.1")
+	reportToolchain(t, ix, "mesh-build-go", "ace", "go", "go1.27.1")
+	b := &takeInBed{ix: ix, mods: mods, pending: newMem(), base: f, logs: &strings.Builder{}}
+	ix.Verdicts = newMem()
+	ix.Pending = b.pending
+	ix.SayTakeIn = func(_ context.Context, t delivery.TakeIn) error { b.said = append(b.said, t); return nil }
+	logf := ix.Logf
+	ix.Logf = func(format string, args ...any) {
+		b.logs.WriteString(format + "\n")
+		if logf != nil {
+			logf(format, args...)
+		}
+	}
+	own := b.fp(t, delivery.IndexerModule)
+	if own == "" || b.fp(t, "avahi") == "" {
+		t.Fatalf("the bed's modules have no fingerprint")
+	}
+	ix.OwnBuild = func(context.Context) (string, error) { return own, nil }
+	return b
+}
+
+func (b *takeInBed) fp(t *testing.T, module string) string {
+	t.Helper()
+	m, _ := moduleOf(t, b.mods, module)
+	return m.FP2
+}
+
+func (b *takeInBed) commit(t *testing.T, module string) string {
+	t.Helper()
+	m, _ := moduleOf(t, b.mods, module)
+	return m.Commit
+}
+
+// merge is a catalogue merge at commit c whose trees of avahi and the indexer are given ("" keeps the base's).
+func (b *takeInBed) merge(number int, c, avahi, indexer string) delivery.Forge {
+	f := b.base
+	f.Number, f.Merge, f.Head = number, c, c
+	f.Trees = map[string]string{}
+	for dir, tree := range b.base.Trees {
+		f.Trees[dir] = tree
+	}
+	if avahi != "" {
+		f.Trees["modules/avahi"] = avahi
+	}
+	if indexer != "" {
+		f.Trees["modules/"+delivery.IndexerModule] = indexer
+	}
+	return f
+}
+
+func (b *takeInBed) arrive(t *testing.T, seq uint64, f delivery.Forge) {
+	t.Helper()
+	if _, _, err := b.ix.Arrive(context.Background(), seq, f); err != nil {
+		t.Fatalf("arrive %d: %v", seq, err)
+	}
+}
+
+func (b *takeInBed) keys() []string {
+	k, _ := b.pending.Keys(context.Background())
+	slices.Sort(k)
+	return k
+}
+
+func (b *takeInBed) events() []string {
+	var out []string
+	for _, s := range b.said {
+		out = append(out, s.Kind+" "+s.Key)
+	}
+	return out
+}
+
+func sha(c byte) string { return strings.Repeat(string(c), 40) }
+
+func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
+	ctx := context.Background()
+	t.Run("T6 a merge while the own build is unknown writes nothing", func(t *testing.T) {
+		b := newTakeInBed(t)
+		b.ix.OwnBuild = func(context.Context) (string, error) { return "", errors.New("the engine does not answer") }
+		_, _, err := b.ix.Arrive(ctx, 1, b.merge(20, sha('b'), "tree-avahi-2", ""))
+		if !errors.Is(err, ErrOwnBuildUnknown) || len(b.keys()) != 0 || b.commit(t, "avahi") == sha('b') {
+			t.Errorf("err %v, pending %v, avahi at %s: want ErrOwnBuildUnknown and nothing written", err, b.keys(), b.commit(t, "avahi"))
+		}
+	})
+	t.Run("T7 and T3 a merge alone, an indexer change, a merge behind it and a revert", func(t *testing.T) {
+		b := newTakeInBed(t)
+		own, avahi0 := b.fp(t, delivery.IndexerModule), b.fp(t, "avahi")
+		// A merge with no change to the indexer and nothing pending is taken in alone, as before.
+		b.arrive(t, 1, b.merge(20, sha('b'), "tree-avahi-2", ""))
+		if avahi1 := b.fp(t, "avahi"); avahi1 == avahi0 || len(b.keys()) != 0 || len(b.said) != 0 {
+			t.Fatalf("alone: avahi %s (was %s), pending %v, said %v: want taken in", avahi1, avahi0, b.keys(), b.events())
+		}
+		avahi1, verdicts := b.fp(t, "avahi"), len(b.ix.Verdicts.(*memStore).vals)
+		// T4: a change to the indexer writes only its entry (with its graph key and verdict); the rest is pending.
+		b.arrive(t, 2, b.merge(21, sha('c'), "tree-avahi-3", "tree-indexer-2"))
+		named := b.fp(t, delivery.IndexerModule)
+		if named == own || b.fp(t, "avahi") != avahi1 || b.commit(t, "avahi") != sha('b') {
+			t.Fatalf("indexer change: indexer %s (own %s), avahi %s at %s: want only the indexer written", named, own, b.fp(t, "avahi"), b.commit(t, "avahi"))
+		}
+		if g := graphOf(t, b.ix, delivery.IndexerModule); g.FP != named {
+			t.Errorf("graph of the indexer names %s, want %s", g.FP, named)
+		}
+		if n := len(b.ix.Verdicts.(*memStore).vals); n > verdicts+1 {
+			t.Errorf("verdicts %d, want at most %d: only the new indexer manifest's", n, verdicts+1)
+		}
+		if want := []string{delivery.PendingKey(2)}; !reflect.DeepEqual(b.keys(), want) || !slices.Equal(b.events(), []string{"pending " + delivery.PendingKey(2)}) || b.said[0].WaitsFor != named {
+			t.Fatalf("pending %v, said %+v: want m.2 pending for %s", b.keys(), b.said, named)
+		}
+		// T5: the merge delivered again writes no second key and says nothing again.
+		b.arrive(t, 2, b.merge(21, sha('c'), "tree-avahi-3", "tree-indexer-2"))
+		if len(b.keys()) != 1 || len(b.said) != 1 {
+			t.Errorf("redelivered: pending %v, said %v", b.keys(), b.events())
+		}
+		// A later merge joins: the newest governs and still names the new indexer.
+		b.arrive(t, 3, b.merge(22, sha('d'), "tree-avahi-4", "tree-indexer-2"))
+		if len(b.keys()) != 2 || b.commit(t, "avahi") != sha('b') || len(b.said) != 1 {
+			t.Fatalf("behind: pending %v, avahi at %s, said %v: want both pending and nothing said again", b.keys(), b.commit(t, "avahi"), b.events())
+		}
+		// A revert brings the indexer back to the running build: everything is taken in, oldest first.
+		b.arrive(t, 4, b.merge(23, sha('e'), "tree-avahi-5", b.base.Trees["modules/"+delivery.IndexerModule]))
+		if len(b.keys()) != 0 || b.fp(t, delivery.IndexerModule) != own || b.commit(t, "avahi") != sha('e') {
+			t.Fatalf("revert: pending %v, indexer %s, avahi at %s: want all taken in", b.keys(), b.fp(t, delivery.IndexerModule), b.commit(t, "avahi"))
+		}
+		if want := []string{"pending " + delivery.PendingKey(2), "taken " + delivery.PendingKey(2)}; !slices.Equal(b.events(), want) {
+			t.Errorf("said %v, want %v", b.events(), want)
+		}
+	})
+	t.Run("T10 the named indexer takes the revision in at its start; a gathered group stays", func(t *testing.T) {
+		b := newTakeInBed(t)
+		b.arrive(t, 2, b.merge(21, sha('c'), "tree-avahi-3", "tree-indexer-2"))
+		g := b.merge(30, sha('f'), "", "")
+		g.Repository, g.Trees, g.Manifests, g.Locks, g.Group = "novox/mesh-sdk", map[string]string{}, nil, nil, "feat/x"
+		g.Members = []delivery.Head{{Repository: "novox/mesh-sdk", Number: 30, Head: g.Head}, {Repository: "novox/mesh-host", Number: 4, Head: sha('9')}}
+		b.arrive(t, 5, g)
+		named := b.fp(t, delivery.IndexerModule)
+		// The new indexer starts; its build is not known at first: a hand ask is answered meanwhile (R5), and the start waits.
+		next := &Indexer{}
+		*next = Indexer{Store: b.ix.Store, Modules: b.ix.Modules, Manifests: b.ix.Manifests, Graph: b.ix.Graph, Verdicts: b.ix.Verdicts,
+			Packages: b.ix.Packages, Builds: b.ix.Builds, Pending: b.pending, SayTakeIn: b.ix.SayTakeIn, Logf: b.ix.Logf, GoSum: b.ix.GoSum}
+		next.toolchains = b.ix.toolchains
+		asked := 0
+		next.OwnBuild = func(context.Context) (string, error) {
+			if asked++; asked < 2 {
+				return "", errors.New("not yet")
+			}
+			return named, nil
+		}
+		if r := next.TakeInPendingByHand(ctx, "m.nothing", "lead@shanks", "a test"); r.Stored || !strings.Contains(r.Answer, "nothing waits") {
+			t.Errorf("a hand ask while the own build is unknown: %+v", r)
+		}
+		if _, _, err := next.TakeInAtStart(ctx, func(context.Context) bool { return true }); err != nil {
+			t.Fatal(err)
+		}
+		if asked < 2 || !reflect.DeepEqual(b.keys(), []string{delivery.PendingKey(5)}) || b.commit(t, "avahi") != sha('c') {
+			t.Errorf("start: asked %d, pending %v, avahi at %s: want the revision taken in, the group left", asked, b.keys(), b.commit(t, "avahi"))
+		}
+	})
+	t.Run("T8 T9 T11 a group gathered until its end, a merge behind it, one alone, an end again", func(t *testing.T) {
+		b := newTakeInBed(t)
+		b.ix.Taken = newMem()
+		// The commits in the order they were made: one is in another when it comes first.
+		order := "a756bc"
+		b.ix.Contains = func(_ context.Context, _, commit, in string) (bool, error) {
+			return strings.IndexByte(order, commit[0]) <= strings.IndexByte(order, in[0]), nil
+		}
+		member := func(repo string, number int, c string) delivery.Forge {
+			f := b.merge(number, c, "", "")
+			f.Group = "feat/take-in"
+			f.Members = []delivery.Head{{Repository: catalogueRepo, Number: 40, Head: sha('1')}, {Repository: "novox/mesh-sdk", Number: 41, Head: sha('2')}}
+			if repo != catalogueRepo {
+				f.Repository, f.Trees, f.Manifests, f.Locks = repo, map[string]string{}, nil, nil
+			}
+			f.Head = map[int]string{40: sha('1'), 41: sha('2')}[number]
+			return f
+		}
+		b.arrive(t, 1, member(catalogueRepo, 40, sha('7')))
+		if b.commit(t, "avahi") == sha('7') || !slices.Equal(b.events(), []string{"pending g.feat/take-in"}) {
+			t.Fatalf("first member: avahi at %s, said %v: want held, the group said pending", b.commit(t, "avahi"), b.events())
+		}
+		// A merge in another repository is taken in alone; one in a member's repository waits behind the group.
+		alone := b.merge(50, sha('5'), "", "")
+		alone.Repository, alone.Trees, alone.Manifests, alone.Locks = "novox/hq", map[string]string{}, nil, nil
+		b.arrive(t, 2, alone)
+		b.arrive(t, 3, b.merge(51, sha('6'), "tree-avahi-9", ""))
+		if !reflect.DeepEqual(b.keys(), []string{delivery.PendingKey(1), delivery.PendingKey(3)}) {
+			t.Fatalf("pending %v: want the member and the merge behind it", b.keys())
+		}
+		end := delivery.ForgeGroupEnd{Group: "feat/take-in", Members: []delivery.GroupMerge{{Repository: catalogueRepo, Number: 40, Merge: sha('7')}, {Repository: "novox/mesh-sdk", Number: 41, Merge: sha('b')}}}
+		if _, _, err := b.ix.EndGroup(ctx, end); err == nil {
+			t.Fatal("an end before its last member is taken in")
+		}
+		b.arrive(t, 4, member("novox/mesh-sdk", 41, sha('b')))
+		if _, _, err := b.ix.EndGroup(ctx, end); err != nil {
+			t.Fatal(err)
+		}
+		if len(b.keys()) != 0 || b.commit(t, "avahi") != sha('6') {
+			t.Fatalf("end: pending %v, avahi at %s: want the group and the merge behind it taken in", b.keys(), b.commit(t, "avahi"))
+		}
+		if want := []string{"pending g.feat/take-in", "pending g.feat/take-in", "taken g.feat/take-in"}; !slices.Equal(b.events(), want) {
+			t.Errorf("said %v, want %v", b.events(), want)
+		}
+		// R1: the end published again after the take-in changes nothing and is not delivered again.
+		if _, words, err := b.ix.EndGroup(ctx, end); err != nil || !strings.Contains(words, "taken in already") {
+			t.Errorf("an end again: %q %v", words, err)
+		}
+		// T11 (the lead's row): a group whose revision changes the indexer: the group's wait is taken, then the revision pending.
+		b.said = nil
+		g := member(catalogueRepo, 40, sha('c'))
+		g.Group, g.Trees = "feat/two", b.merge(0, "", "", "tree-indexer-7").Trees
+		g.Members = g.Members[:1]
+		b.arrive(t, 9, g)
+		if _, _, err := b.ix.EndGroup(ctx, delivery.ForgeGroupEnd{Group: "feat/two", Members: []delivery.GroupMerge{{Repository: catalogueRepo, Number: 40, Merge: sha('c')}}}); err != nil {
+			t.Fatal(err)
+		}
+		if want := []string{"pending g.feat/two", "taken g.feat/two", "pending " + delivery.PendingKey(9)}; !slices.Equal(b.events(), want) {
+			t.Errorf("a group naming another indexer: said %v, want %v", b.events(), want)
+		}
+	})
+	t.Run("T5 two take-ins of one pending revision: the one that loses the claim writes nothing", func(t *testing.T) {
+		b := newTakeInBed(t)
+		b.arrive(t, 2, b.merge(21, sha('c'), "tree-avahi-3", "tree-indexer-2"))
+		named := b.fp(t, delivery.IndexerModule)
+		b.ix.ownMu.Lock()
+		b.ix.ownFP = named
+		b.ix.ownMu.Unlock()
+		// The other indexer claims the merge between this one's read and its claim.
+		b.pending.beforeWrite = func() {
+			v := b.pending.vals[delivery.PendingKey(2)]
+			var p Pending
+			_ = json.Unmarshal(v, &p)
+			p.Taking = "the other indexer"
+			data, _ := json.Marshal(p)
+			b.pending.put(delivery.PendingKey(2), data)
+		}
+		if _, _, err := b.ix.TakeInAtStart(ctx, func(context.Context) bool { return false }); err != nil {
+			t.Fatal(err)
+		}
+		if b.commit(t, "avahi") == sha('c') || len(b.keys()) != 1 || !strings.Contains(b.logs.String(), "is lost") {
+			t.Errorf("avahi at %s, pending %v: want nothing taken in by the loser, and it said", b.commit(t, "avahi"), b.keys())
+		}
+	})
+	t.Run("T12 T13 indexer_pending names the wait; a hand take-in and a drop say what they ended", func(t *testing.T) {
+		b := newTakeInBed(t)
+		b.arrive(t, 2, b.merge(21, sha('c'), "tree-avahi-3", "tree-indexer-2"))
+		b.arrive(t, 3, b.merge(22, sha('d'), "tree-avahi-4", "tree-indexer-2"))
+		var answer delivery.PendingAnswer
+		for _, tool := range PendingTools(b.pending, nil, nil) {
+			if tool.Name == PendingTool {
+				out, err := tool.Run(map[string]any{})
+				if err != nil {
+					t.Fatal(err)
+				}
+				answer = out.(delivery.PendingAnswer)
+			}
+		}
+		if len(answer.Waits) != 1 || answer.Waits[0].Key != delivery.PendingKey(2) || !slices.Equal(answer.Waits[0].Holds, []string{"avahi", delivery.IndexerModule}) ||
+			answer.Waits[0].Said != "take-in.pending."+delivery.PendingKey(2) || len(answer.Waits[0].Keys) != 2 {
+			t.Fatalf("indexer_pending = %+v", answer)
+		}
+		b.said = nil
+		if r := b.ix.DropPending(ctx, delivery.PendingKey(2), "lead@shanks", "replaced by m.3"); !r.Stored {
+			t.Fatalf("drop: %+v", r)
+		}
+		if len(b.said) < 2 || !b.said[0].Dropped || b.said[0].Key != delivery.PendingKey(2) || b.said[0].By != "lead@shanks" || b.said[1].Key != delivery.PendingKey(3) {
+			t.Fatalf("drop said %+v: want m.2 taken (dropped, by) and m.3 pending", b.said)
+		}
+		b.said = nil
+		if r := b.ix.TakeInPendingByHand(ctx, delivery.PendingKey(3), "lead@shanks", "the named indexer can never run"); !r.Stored {
+			t.Fatalf("hand take-in: %+v", r)
+		}
+		if len(b.keys()) != 0 || b.commit(t, "avahi") != sha('d') || len(b.said) != 1 || b.said[0].Kind != delivery.TakeInTaken || b.said[0].By != "lead@shanks" {
+			t.Errorf("hand take-in: pending %v, avahi at %s, said %+v", b.keys(), b.commit(t, "avahi"), b.said)
+		}
+	})
+	t.Run("the dry run names every field of the indexer", func(t *testing.T) {
+		// dry (pending.go) builds an indexer field by field: a field added to Indexer is named there, or this fails.
+		named := map[string]bool{}
+		for _, f := range []string{"Store", "Modules", "Manifests", "Predefined", "ConfigCandidates", "Taken", "Verdicts", "OldSections",
+			"JudgeID", "SayRefused", "SayAccepted", "Packages", "Graph", "Files", "GoSum", "GoMod", "Now", "Builds", "Named", "Trunk",
+			"Read", "Contains", "Logf", "ReadOpen", "OwnBuild", "Pending", "SayTakeIn"} {
+			named[f] = true
+		}
+		ty := reflect.TypeOf(Indexer{})
+		for i := range ty.NumField() {
+			if f := ty.Field(i); f.IsExported() && !named[f.Name] {
+				t.Errorf("Indexer.%s is not named by dry", f.Name)
+			}
+		}
+	})
+}
@@ -0,0 +1,69 @@
+package index
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+// PendingTools are the indexer's tools of what waits for a take-in (novox/hq ADR 0633 §5): indexer_pending reads the
+// pending state, and indexer_take_in_pending and indexer_drop_pending steer it by an ask in takeins, as indexer_take_in
+// does. pending is the pending state as the tools bundle reads it; ask writes an ask and waits for the indexer's answer.
+func PendingTools(pending Store, ask func(key string, value []byte) (TakeInAsk, error), now func() time.Time) []stdio.Tool {
+	str := func(args map[string]any, k string) string { s, _ := args[k].(string); return s }
+	hand := func(drop bool) func(map[string]any) (any, error) {
+		return func(args map[string]any) (any, error) {
+			key, value, err := NewPendingAsk(str(args, "key"), str(args, "why"), stdio.CallerOf(args), drop, now())
+			if err != nil {
+				return nil, err
+			}
+			a, err := ask(key, value)
+			if err != nil {
+				return nil, err
+			}
+			return a.Result, nil
+		}
+	}
+	return []stdio.Tool{
+		{Name: PendingTool,
+			Description: "What waits for a take-in (ADR 0633): merges the indexer holds in its pending state instead of taking them in. " +
+				"A revision (what revision, its key its oldest merge's m.<sequence>) waits for the build of the indexer it names " +
+				"(waits-for): the running indexer wrote only what building and moving that indexer needs, and the new indexer takes " +
+				"it in at its start. A group (what group, key g.<group>) is a delivery group whose members are gathered until the " +
+				"forge bridge says its end; a merge in one of its members' repositories waits behind it (behind). Each names its " +
+				"merges, since when it waits, the modules it holds back (holds) and the last take-in event said of it. Empty waits: " +
+				"every merge is taken in. The watchman raises take-in-waits past 60 minutes for a revision, 30 for a group; " +
+				"indexer_take_in_pending and indexer_drop_pending steer a wait.",
+			Input: map[string]any{"type": "object", "properties": map[string]any{}},
+			Run: func(map[string]any) (any, error) {
+				var unread []string
+				entries, err := ReadPending(context.Background(), pending, func(format string, args ...any) {
+					unread = append(unread, fmt.Sprintf(format, args...))
+				})
+				if err != nil {
+					return nil, err
+				}
+				return delivery.PendingAnswer{Waits: PendingWaits(entries), Unread: unread}, nil
+			}},
+		{Name: TakeInPendingTool,
+			Description: "Take a wait in by hand with the running indexer and its judge (ADR 0633 §5): the revision under its m. key, " +
+				"whatever build of the indexer it names, or a delivery group under its g. key with the members gathered so far. For " +
+				"when the named indexer can never run, or a group's end never comes. A hand act: who and why are said in the " +
+				"indexer's journal and on the take-in.taken event, which clears take-in-waits. indexer_pending names the keys.",
+			Input: map[string]any{"type": "object", "required": []string{"key", "why"}, "properties": map[string]any{
+				"key": map[string]any{"type": "string", "description": "the wait's key, m.<sequence> or g.<group>, as indexer_pending names it"},
+				"why": map[string]any{"type": "string", "description": "why it is taken in by hand, in plain words"}}},
+			Run: hand(false)},
+		{Name: DropPendingTool,
+			Description: "Drop one pending merge by hand, abandoned because a later merge replaced it (ADR 0633 §5): it is removed from " +
+				"the pending state and nothing of it is taken in; what waited with it is worked out again. A hand act: who and why are " +
+				"said in the journal and on the take-in.taken event (dropped).",
+			Input: map[string]any{"type": "object", "required": []string{"key", "why"}, "properties": map[string]any{
+				"key": map[string]any{"type": "string", "description": "the pending merge's key, m.<sequence>, as indexer_pending names its merges' wait"},
+				"why": map[string]any{"type": "string", "description": "why it is dropped, in plain words"}}},
+			Run: hand(true)},
+	}
+}
@@ -51,6 +51,17 @@ func TestASourceRevisionStampsTheModulesOfTheRepositoriesHeadItNames(t *testing.
 }

 func TestAMergeHeardAfterTheSourceNamesItsHeadIsStampedWhenWritten(t *testing.T) { // 1344/AC1 (the source first)
+	// A source revision older than the one held is left (merged here from a test of its own, for the test budget).
+	t.Run("an older source revision is left", func(t *testing.T) {
+		ix, mods := modulesIndexer(t, rec(t, "build-1", "a", "novox/mesh-delivery", "modules/a", "main", "", "2026-10-19T09:00:00Z", aInputs))
+		ctx := context.Background()
+		_ = ix.Apply(ctx, mergedMsg("novox/mesh-delivery", map[string]string{"modules/a": "newtree"}))
+		_ = ix.ApplySource(ctx, 9, sourceOf("novox/mesh-delivery", merge))
+		_ = ix.ApplySource(ctx, 7, sourceOf("novox/mesh-delivery", merge))
+		if m, _ := moduleOf(t, mods, "a"); m.Src != 9 {
+			t.Errorf("module a Src = %d, want 9", m.Src)
+		}
+	})
 	ix, mods := modulesIndexer(t, rec(t, "build-1", "a", "novox/mesh-delivery", "modules/a", "main", "", "2026-10-19T09:00:00Z", aInputs))
 	ctx := context.Background()
 	if err := ix.ApplySource(ctx, 7, sourceOf("novox/mesh-delivery", merge)); err != nil {
@@ -93,14 +104,3 @@ func TestAMergeThatNoSourceRevisionNamesIsWrittenAsToday(t *testing.T) { // 1344
 		t.Errorf("module a = %+v, want the new merge with Src 0", m)
 	}
 }
-
-func TestASourceRevisionOlderThanTheOneHeldIsLeft(t *testing.T) {
-	ix, mods := modulesIndexer(t, rec(t, "build-1", "a", "novox/mesh-delivery", "modules/a", "main", "", "2026-10-19T09:00:00Z", aInputs))
-	ctx := context.Background()
-	_ = ix.Apply(ctx, mergedMsg("novox/mesh-delivery", map[string]string{"modules/a": "newtree"}))
-	_ = ix.ApplySource(ctx, 9, sourceOf("novox/mesh-delivery", merge))
-	_ = ix.ApplySource(ctx, 7, sourceOf("novox/mesh-delivery", merge))
-	if m, _ := moduleOf(t, mods, "a"); m.Src != 9 {
-		t.Errorf("module a Src = %d, want 9", m.Src)
-	}
-}
@@ -33,7 +33,11 @@ type TakeInAsk struct {
 	// Judge asks the manifests to be judged again now (indexer_judge_again) instead of the module to be taken in.
 	Judge bool `json:"judge,omitempty"`
 	// Digest names, with Judge, the one manifest digest of the module to judge (indexer_judge with a digest).
-	Digest   string    `json:"digest,omitempty"`
+	Digest string `json:"digest,omitempty"`
+	// Pending asks the wait under this take-in key to be taken in by hand (indexer_take_in_pending), or with Drop the
+	// one pending merge under it to be dropped (indexer_drop_pending), instead of a module (pending.go, ADR 0633 §5).
+	Pending  string    `json:"pending,omitempty"`
+	Drop     bool      `json:"drop,omitempty"`
 	By       string    `json:"by,omitempty"`
 	At       time.Time `json:"at"`
 	Result   *TakenIn  `json:"result,omitempty"`
@@ -52,8 +56,8 @@ func DecodeTakeIn(data []byte) (TakeInAsk, error) {
 	if a.V != 1 {
 		return a, fmt.Errorf("a takeins value of version %d, not 1", a.V)
 	}
-	if a.Module == "" && a.Repository == "" {
-		return a, errors.New("a takeins value naming no module or repository")
+	if a.Module == "" && a.Repository == "" && a.Pending == "" {
+		return a, errors.New("a takeins value naming no module, repository or pending wait")
 	}
 	return a, nil
 }
@@ -82,3 +86,25 @@ func NewTakeInRepository(repository, why, by string, at time.Time) (string, []by
 	data, err := json.Marshal(TakeInAsk{V: 1, Repository: repository, Why: strings.TrimSpace(why), By: by, At: at.UTC()})
 	return "repository." + key, data, err
 }
+
+// NewPendingAsk is the value indexer_take_in_pending (drop false) or indexer_drop_pending (drop true) writes: its key
+// in takeins, `pending.<take-in|drop>.<key>`, and the value. key is a wait's key as indexer_pending names it, for a
+// drop one pending merge's.
+func NewPendingAsk(key, why, by string, drop bool, at time.Time) (string, []byte, error) {
+	key = strings.TrimSpace(key)
+	if !strings.HasPrefix(key, "m.") && !(strings.HasPrefix(key, "g.") && !drop) {
+		if drop {
+			return "", nil, errors.New("name the pending merge to drop by its key, m.<sequence>, as indexer_pending names it")
+		}
+		return "", nil, errors.New("name the wait to take in by its key, m.<sequence> or g.<group>, as indexer_pending names it")
+	}
+	if strings.TrimSpace(why) == "" {
+		return "", nil, errors.New("say why (why): a hand act on what waits is said in the journal and on its take-in event")
+	}
+	act := "take-in"
+	if drop {
+		act = "drop"
+	}
+	data, err := json.Marshal(TakeInAsk{V: 1, Pending: key, Drop: drop, Why: strings.TrimSpace(why), By: by, At: at.UTC()})
+	return "pending." + act + "." + key, data, err
+}
@@ -16,6 +16,8 @@ import (
 	"git.novox.be/novox/mesh-sdk/go/delivery"

 	"git.novox.be/novox/mesh-delivery/sdk/graph"
+
+	"mesh-delivery-indexer/internal/index"
 )

 // The controller's manifest reader is mesh-controller internal/catalogue (Manifest, manifest.go: Consumes :521,
@@ -28,6 +30,12 @@ type state struct {
 	History int    `json:"history,omitempty"`
 }

+type streamDecl struct {
+	Name       string   `json:"name"`
+	Events     []string `json:"events"`
+	MaxAgeDays int      `json:"max-age-days,omitempty"`
+}
+
 type artifact struct {
 	Name     string   `json:"name"`
 	Kind     string   `json:"kind"`
@@ -51,6 +59,7 @@ type manifest struct {
 	Emits        []string          `json:"emits"`
 	Consumes     []string          `json:"consumes"`
 	State        []state           `json:"state"`
+	Streams      []streamDecl      `json:"streams"`
 	Reads        []string          `json:"reads"`
 	Invokes      []string          `json:"invokes"`
 	OwnSecrets   map[string]string `json:"own-secrets"`
@@ -99,7 +108,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
 	}
 	// 1126 note 76: heard keeps the last 64 FORGE messages the indexer handled, for indexer_heard.
 	if want := []state{{Name: "candidates", History: 10}, {Name: "heard", History: 64}, {Name: "manifests", History: 1}, {Name: "modules", History: 20}, {Name: "predefined", History: 5}, {Name: "config-candidates", History: 5}, {Name: "taken", History: 1}, {Name: "takeins", History: 1}, {Name: "verdicts", History: 5}, {Name: "oldsections", History: 1}, {Name: delivery.PackagesState, History: delivery.PackagesHistory},
-		{Name: graph.State, History: graph.History}}; !reflect.DeepEqual(m.State, want) {
+		{Name: graph.State, History: graph.History}, {Name: index.PendingState, History: 1}}; !reflect.DeepEqual(m.State, want) {
 		t.Errorf("state = %+v, want %+v", m.State, want)
 	}
 	if want := []string{"mesh-build-go.toolchains", "mesh-build-node.toolchains", "mesh-delivery-assigner.placement", "mesh-delivery-builder.build-log", "mesh-delivery-checker.previews", "mesh-delivery-gate.source"}; !reflect.DeepEqual(sorted(m.Reads), want) {
@@ -125,7 +134,9 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
 	// 1945: indexer_take_in, served by its tools bundle through the takeins state; the process serves none. The same bundle
 	// reads the verdicts state (indexer_verdicts) and asks the indexer to judge again through the takeins state
 	// (indexer_judge; issue 774, clause 3). 1966/AC11: indexer_old_sections reads the count of the old sections (state oldsections).
-	if want := []string{"indexer_take_in", "indexer_old_sections", "indexer_verdicts", "indexer_judge"}; !reflect.DeepEqual(m.Tools, want) {
+	// 1585/T12, T13 (ADR 0633 §5): indexer_pending reads the pending state; indexer_take_in_pending and indexer_drop_pending
+	// ask through takeins.
+	if want := []string{"indexer_take_in", "indexer_old_sections", "indexer_verdicts", "indexer_judge", index.PendingTool, index.TakeInPendingTool, index.DropPendingTool}; !reflect.DeepEqual(m.Tools, want) {
 		t.Errorf("tools = %v, want %v", m.Tools, want)
 	}
 	// ADR 0604 §2: it holds the seat mesh-catalog, serving its four verbs from its tools bundle (which stays without a
@@ -134,8 +145,13 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
 		!slices.Equal(m.Claims[0].Serves, []string{"modules", "module", "provides", "dependents"}) {
 		t.Errorf("claims = %+v, want mesh-catalog with its four verbs", m.Claims)
 	}
-	// No streams, no bus, no other seat (ADR 0437 §4); it says no event.
-	for _, k := range []string{"streams", "logs", "provides", "requires", "uses", "defines-seats", "seat", "seats", "bus"} {
+	// 1585/T11 (ADR 0633 §1, §5): its one stream, take-in, an old streams entry (the indexer is foundation, ADR 0595 §11),
+	// carries the take-in events.
+	if want := []streamDecl{{Name: delivery.TakeInStream, Events: []string{"take-in.**"}, MaxAgeDays: 7}}; !reflect.DeepEqual(m.Streams, want) {
+		t.Errorf("streams = %+v, want %+v", m.Streams, want)
+	}
+	// No other bus section, no other seat (ADR 0437 §4).
+	for _, k := range []string{"logs", "provides", "requires", "uses", "defines-seats", "seat", "seats", "bus"} {
 		if _, ok := top[k]; ok {
 			t.Errorf("the manifest declares %q: the indexer has none", k)
 		}
@@ -18,7 +18,10 @@
    "indexer_take_in",
    "indexer_old_sections",
    "indexer_verdicts",
-    "indexer_judge"
+    "indexer_judge",
+    "indexer_pending",
+    "indexer_take_in_pending",
+    "indexer_drop_pending"
  ],
  "consumes": [
    "mesh-build-go.toolchain.reported",
@@ -38,7 +41,11 @@
    { "name": "verdicts", "history": 5 },
    { "name": "oldsections", "history": 1 },
    { "name": "packages", "history": 5 },
-    { "name": "graph", "history": 5 }
+    { "name": "graph", "history": 5 },
+    { "name": "pending", "history": 1 }
+  ],
+  "streams": [
+    { "name": "take-in", "events": ["take-in.**"], "max-age-days": 7 }
  ],
  "emits": [
    "manifest-refused",