GATE A-33, A-34: a take-in.taken opens one delivery record and moves its branch's source, saying source-moved

Builds: design 47 G §151, §291, §592, A-33, A-34 (G-3, G-5, G-16, G-20)

modules/mesh-delivery-gate/cmd/mesh-delivery-gate/consumer.go

@@ -120,6 +120,8 @@ type StepConsumer struct {
 	Families []stepFamily
 	Now      func() time.Time
 	Logf     func(format string, a ...any)
+	// Taken opens the record of a take-in.taken and moves its branch's source (takenin.go), before its outcomes are applied.
+	Taken func(body json.RawMessage) error
 
 	mu sync.Mutex
 	// heard counts, per family, the events applied, refused and the last applied, since this owner started.
@@ -180,6 +182,11 @@ func (c *StepConsumer) Hear(e stdio.Envelope) error {
 		c.logf("[mesh-delivery] the record does not take %s: %s", e.Key, why)
 		return nil
 	}
+	if f.Event == takein.TakeInTakenEvent && c.Taken != nil {
+		if err := c.Taken(e.Body); err != nil {
+			return fmt.Errorf("the record of %s is not opened yet: %w", e.Key, err)
+		}
+	}
 	id := e.Headers["x-event-id"]
 	for _, o := range outcomes {
 		if f.NodeScoped && o.Node == "" {
@@ -267,10 +274,13 @@ func holds(r *Record, o StepOutcome) bool {
 
 // mergedAs are the commits the record's members merged as.
 func (r *Record) mergedAs() []string {
-	if r.Group == nil {
-		return nil
-	}
 	var out []string
+	for _, c := range r.Heads {
+		out = append(out, c)
+	}
+	if r.Group == nil {
+		return out
+	}
 	for _, p := range r.Group.Pulls {
 		if p.Commit != "" {
 			out = append(out, p.Commit)

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

@@ -64,7 +64,7 @@ func main() {
 	h.SaySeat = stdio.SeatPublish
 	h.Records = &Records{Ledger: sdkLedger{}, Now: h.Now, Retention: h.recordRetention, Say: stdio.SeatPublish, Logf: logf}
 	h.Old = kvOld{}
-	h.Consumer = &StepConsumer{Records: h.Records, Families: stepFamilies, Now: h.Now, Logf: logf}
+	h.Consumer = &StepConsumer{Records: h.Records, Families: stepFamilies, Now: h.Now, Logf: logf, Taken: h.TakenIn}
 	l := &listening{now: "not yet: starting"}
 	go run(h, l)
 	if err := announceKept(stdio.Keeps); err != nil {

modules/mesh-delivery-gate/cmd/mesh-delivery-gate/merges.go

@@ -49,9 +49,9 @@ func (kvOld) Get(state, key string) ([]byte, uint64, error) {
 
 // SourceStore is the gate's bucket source, one key per watched branch (design 47 G §15), moved by compare-and-set.
 type SourceStore interface {
-	// Move reads the source key of the gate's merge branch (MergeBranch) and writes what update makes of it, unless update
-	// says nothing is to be written; a write another writer won is read again and tried again.
-	Move(update func(prev *Source) (Source, bool)) error
+	// Set reads a branch's source key and writes what update makes of it, unless update says nothing is to be written, and answers
+	// the key's revision; a write another writer won is read again and tried again.
+	Set(branch string, update func(prev *Source) (Source, bool)) (uint64, error)
 	// Read is a branch's source key as held and its revision on the bus; nil and 0 before it is first written.
 	Read(branch string) (*Source, uint64, error)
 	// Branches are the branch names of the source keys the gate keeps (§614, §690).
@@ -80,42 +80,44 @@ func (kvMerges) Get(group string) (*Merge, error) {
 // kvSource is source over the runtime.
 type kvSource struct{}
 
-func (kvSource) Move(update func(prev *Source) (Source, bool)) error {
+func (kvSource) Set(branch string, update func(prev *Source) (Source, bool)) (uint64, error) {
 	state := stdio.State("source")
+	key := mdc.SourceKey(keys.EscapeName(branch))
 	for range changeAttempts {
-		e, err := state.Get(mdc.SourceKey(MergeBranch))
+		e, err := state.Get(key)
 		if err != nil {
-			return err
+			return 0, err
 		}
 		var prev *Source
+		var rev uint64
 		if e != nil {
 			p, err := mdc.DecodeSource(e.Value)
 			if err != nil {
-				return fmt.Errorf("source.main is not readable: %w", err)
+				return 0, fmt.Errorf("source.%s is not readable: %w", branch, err)
 			}
-			prev = &p
+			prev, rev = &p, e.Revision
 		}
 		next, write := update(prev)
 		if !write {
-			return nil
+			return rev, nil
 		}
 		raw, err := mdc.EncodeSource(next)
 		if err != nil {
-			return err
+			return 0, err
 		}
 		if e == nil {
-			_, err = state.Create(mdc.SourceKey(MergeBranch), json.RawMessage(raw))
+			rev, err = state.Create(key, json.RawMessage(raw))
 		} else {
-			_, err = state.Update(mdc.SourceKey(MergeBranch), json.RawMessage(raw), e.Revision)
+			rev, err = state.Update(key, json.RawMessage(raw), e.Revision)
 		}
 		if err == nil {
-			return nil
+			return rev, nil
 		}
 		if !errors.Is(err, stdio.ErrChanged) {
-			return err
+			return 0, err
 		}
 	}
-	return fmt.Errorf("source.main was written by another writer first %d times: %w", changeAttempts, stdio.ErrChanged)
+	return 0, fmt.Errorf("source.%s was written by another writer first %d times: %w", branch, changeAttempts, stdio.ErrChanged)
 }
 
 func (kvSource) Read(branch string) (*Source, uint64, error) {
@@ -189,11 +191,7 @@ func (h *Holder) SourceMain() map[string]any {
 	if h.Source == nil {
 		return map[string]any{"error": "no source bucket"}
 	}
-	var held *Source
-	err := h.Source.Move(func(prev *Source) (Source, bool) {
-		held = prev
-		return Source{}, false
-	})
+	held, _, err := h.Source.Read(MergeBranch)
 	switch {
 	case err != nil:
 		return map[string]any{"error": err.Error()}

modules/mesh-delivery-gate/cmd/mesh-delivery-gate/takenin.go

@@ -0,0 +1,98 @@
+package main
+
+import (
+	"encoding/json"
+	"fmt"
+	"sort"
+	"strconv"
+
+	mdc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient"
+	meshcatalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
+	seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery"
+	"git.novox.be/novox/mesh-sdk/go/keys"
+)
+
+// What a take-in.taken does at the gate (design 47 G A-33, A-34; §151, §291, §592): it opens one delivery record, keyed by the
+// taken body's branch and source revision, and moves that branch's source revision, saying source-moved. Nothing else opens a
+// record or moves a source key.
+
+// recordMilestone is the key of a taken revision's record: `<branch>.<source>`, each escaped as a name (A-33).
+func recordMilestone(branch string, source int64) string {
+	return keys.EscapeName(branch) + "." + keys.EscapeName(strconv.FormatInt(source, 10))
+}
+
+// TakenIn takes one take-in.taken body: the record of its source revision is opened (safe to repeat: a record there already
+// is left as it is), its members are the entries' pull requests, its heads the merge commits the take-in names; then the
+// branch's source key is written and source-moved said. A hand drop took nothing in, so it opens nothing.
+func (h *Holder) TakenIn(body json.RawMessage) error {
+	var t meshcatalog.TakeInTaken
+	if err := json.Unmarshal(body, &t); err != nil {
+		return fmt.Errorf("a take-in.taken is not readable: %w", err)
+	}
+	if t.Dropped || len(t.Merges) == 0 || t.Branch == "" {
+		return nil
+	}
+	if h.Records != nil {
+		heads := map[string]string{}
+		for _, m := range t.Merges {
+			heads[mdc.ApprovedKey(m.Repository, int(m.Number))] = m.Merge
+		}
+		milestone := recordMilestone(t.Branch, t.Source)
+		r, _, err := h.Records.Open(milestone, heads, false)
+		if err != nil {
+			return fmt.Errorf("the record of %s cannot be opened: %w", milestone, err)
+		}
+		g := RecordGroup{Group: t.Group}
+		seen := map[string]bool{}
+		for _, e := range t.Entries {
+			if e.Repository == "" || seen[mdc.ApprovedKey(e.Repository, int(e.Pull))] {
+				continue
+			}
+			seen[mdc.ApprovedKey(e.Repository, int(e.Pull))] = true
+			g.Members = append(g.Members, mdc.ApprovedKey(e.Repository, int(e.Pull)))
+		}
+		sort.Strings(g.Members)
+		for _, m := range t.Merges {
+			g.Pulls = append(g.Pulls, RecordPull{Repository: m.Repository, Number: int(m.Number), Head: heads[mdc.ApprovedKey(m.Repository, int(m.Number))],
+				State: string(mdc.MergeMerged), Commit: m.Merge})
+		}
+		if err := h.Records.SetGroup(r, g); err != nil {
+			return err
+		}
+	}
+	return h.moveSource(t)
+}
+
+// moveSource writes the taken source revision of a branch: the merge commit of each repository it took in, the group that
+// moved it, and says source-moved with the key's revision.
+func (h *Holder) moveSource(t meshcatalog.TakeInTaken) error {
+	if h.Source == nil {
+		return nil
+	}
+	heads := map[string]string{}
+	for _, m := range t.Merges {
+		heads[m.Repository] = m.Merge
+	}
+	rev, err := h.Source.Set(t.Branch, func(prev *Source) (Source, bool) {
+		next := Source{Group: t.Group, Heads: heads}
+		if prev != nil {
+			next.Heads = map[string]string{}
+			for r, c := range prev.Heads {
+				next.Heads[r] = c
+			}
+			for r, c := range heads {
+				next.Heads[r] = c
+			}
+		}
+		return next, true
+	})
+	if err != nil {
+		return fmt.Errorf("source.%s cannot be moved: %w", t.Branch, err)
+	}
+	if h.SaySeat == nil {
+		return nil
+	}
+	at := h.Now().UTC()
+	return h.SaySeat(seat.SubjectSourceMoved(keys.EscapeName(t.Branch)),
+		seat.SourceMovedEvent{Branch: t.Branch, Group: t.Group, Revision: int64(rev), At: &at}, "source-moved."+keys.EscapeName(t.Branch)+"."+strconv.FormatUint(rev, 10))
+}

modules/mesh-delivery-gate/cmd/mesh-delivery-gate/takenin_test.go

@@ -0,0 +1,94 @@
+package main
+
+import (
+	"encoding/json"
+	"strings"
+	"testing"
+	"time"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+	"git.novox.be/novox/mesh-sdk/go/keys"
+)
+
+// memSource is the gate's source bucket in memory: one value and revision per branch key.
+type memSource struct {
+	vals map[string]Source
+	rev  uint64
+}
+
+func (m *memSource) Set(branch string, update func(prev *Source) (Source, bool)) (uint64, error) {
+	var prev *Source
+	if p, ok := m.vals[branch]; ok {
+		prev = &p
+	}
+	next, write := update(prev)
+	if !write {
+		return m.rev, nil
+	}
+	if m.vals == nil {
+		m.vals = map[string]Source{}
+	}
+	m.vals[branch] = next
+	m.rev++
+	return m.rev, nil
+}
+func (m *memSource) Read(branch string) (*Source, uint64, error) {
+	v, ok := m.vals[branch]
+	if !ok {
+		return nil, 0, nil
+	}
+	return &v, m.rev, nil
+}
+func (m *memSource) Branches() ([]string, error) { return nil, nil }
+
+// Design 47 G A-33, A-34: a take-in.taken opens one record keyed `<branch>.<source>`, whose members are the entries' pull
+// requests and whose heads are the merge commits it names, moves that branch's source key and says source-moved; heard twice
+// it opens no second record; a hand drop opens nothing and moves nothing.
+func TestATakeInTakenOpensOneRecordAndMovesItsBranchsSource(t *testing.T) {
+	now := time.Date(2026, 11, 11, 20, 0, 0, 0, time.UTC)
+	type said struct{ subject, id string }
+	var events []said
+	src := &memSource{}
+	h := &Holder{Now: func() time.Time { return now }, Source: src,
+		SaySeat: func(subject string, body any, id string) error {
+			events = append(events, said{subject, id})
+			return nil
+		}}
+	h.Records = &Records{Ledger: newLedgerFake(t), Now: h.Now}
+	merge := strings.Repeat("a", 40)
+	body := json.RawMessage(`{"v":1,"kind":"taken","key":"g.feat","what":"group","group":"feat","branch":"main","source":7,` +
+		`"merges":[{"repository":"novox/app","number":3,"merge":"` + merge + `"}],` +
+		`"entries":[{"module":"app","commit":"` + merge + `","fingerprint":"f","repository":"novox/app","pull":3}],"since":"2026-11-11T19:00:00Z"}`)
+	c := &StepConsumer{Records: h.Records, Families: stepFamilies, Now: h.Now, Taken: h.TakenIn}
+	for i := 0; i < 2; i++ {
+		if err := c.Hear(stdio.Envelope{Key: "mesh-delivery-indexer.take-in.taken", Body: body}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	kept, err := h.Records.Kept()
+	if err != nil || len(kept) != 1 {
+		t.Fatalf("records %v %v: one per take-in", kept, err)
+	}
+	r := kept[0]
+	if r.Milestone != keys.EscapeName("main")+".7" || r.Heads["novox/app@3"] != merge || r.Group == nil ||
+		len(r.Group.Members) != 1 || r.Group.Members[0] != "novox/app@3" || r.Group.Group != "feat" {
+		t.Fatalf("the record %+v group %+v", r, r.Group)
+	}
+	if got := r.Outcomes[OutcomeKey("", "", "", "take-in")]; got.Outcome != OutcomeTakenIn {
+		t.Fatalf("the take-in is not the record's outcome: %+v", r.Outcomes)
+	}
+	if s := src.vals["main"]; s.Group != "feat" || s.Heads["novox/app"] != merge {
+		t.Fatalf("source.main %+v", s)
+	}
+	if len(events) != 2 || !strings.HasSuffix(events[0].subject, ".event.source-moved.main") {
+		t.Fatalf("said %+v", events)
+	}
+	drop := json.RawMessage(`{"v":1,"kind":"taken","key":"m.00000000000000000002","what":"revision","dropped":true,"branch":"main","source":8,` +
+		`"merges":[{"repository":"novox/lib","number":4,"merge":"` + merge + `"}],"since":"2026-11-11T19:00:00Z"}`)
+	if err := c.Hear(stdio.Envelope{Key: "mesh-delivery-indexer.take-in.taken", Body: drop}); err != nil {
+		t.Fatal(err)
+	}
+	if kept, _ := h.Records.Kept(); len(kept) != 1 || len(events) != 2 {
+		t.Fatalf("a hand drop opened %d records and said %d events", len(kept), len(events))
+	}
+}