Add generic reconciliation plan application

2026-08-09 16:27:36 +00:00
parent 5b6eb591b2
commit 569e12c6f4
2 changed files with 770 additions and 0 deletions
--- a/internal/framework/semanticreconcile/application.go
+++ b/internal/framework/semanticreconcile/application.go
@@ -0,0 +1,394 @@
+package semanticreconcile
+
+import (
+	"fmt"
+	"sort"
+	"strings"
+)
+
+// CloneValueFunc returns a value that shares no caller-owned mutable state with
+// its input.
+type CloneValueFunc[T any] func(T) T
+
+// Record owns one typed value and its deterministic input provenance.
+type Record[T any] struct {
+	value                 T
+	originalInputIndexes  []int
+	earliestInputPosition int
+	cloneValue            CloneValueFunc[T]
+}
+
+// NewRecord constructs an owned typed record. Original input indexes are
+// normalized into ascending unique order.
+func NewRecord[T any](value T, originalInputIndexes []int, earliestInputPosition int, cloneValue CloneValueFunc[T]) (Record[T], error) {
+	if cloneValue == nil {
+		return Record[T]{}, fmt.Errorf("construct semantic reconciliation record: clone value function must not be nil")
+	}
+	if earliestInputPosition < 0 {
+		return Record[T]{}, fmt.Errorf("construct semantic reconciliation record: earliest input position must not be negative")
+	}
+	indexes, err := normalizeInputIndexes(originalInputIndexes)
+	if err != nil {
+		return Record[T]{}, fmt.Errorf("construct semantic reconciliation record: %w", err)
+	}
+	return Record[T]{
+		value:                 cloneValue(value),
+		originalInputIndexes:  indexes,
+		earliestInputPosition: earliestInputPosition,
+		cloneValue:            cloneValue,
+	}, nil
+}
+
+// Value returns an independently owned typed value.
+func (record Record[T]) Value() T {
+	if record.cloneValue == nil {
+		var zero T
+		return zero
+	}
+	return record.cloneValue(record.value)
+}
+
+// OriginalInputIndexes returns an owned ascending unique index list.
+func (record Record[T]) OriginalInputIndexes() []int {
+	return cloneSlice(record.originalInputIndexes)
+}
+
+// EarliestInputPosition returns the earliest deterministic input position
+// contributing to this record.
+func (record Record[T]) EarliestInputPosition() int { return record.earliestInputPosition }
+
+// RejectionCategory is a stable, domain-neutral reason that an otherwise safe
+// semantic group was not applied.
+type RejectionCategory string
+
+// ApplicationPolicy supplies only the typed behavior needed to apply a safe
+// reconciliation plan. An empty category from RejectGroup accepts the group.
+type ApplicationPolicy[T any] struct {
+	CloneValue       CloneValueFunc[T]
+	RejectGroup      func(members []T, canonical T) RejectionCategory
+	ConsolidateGroup func(members []T, canonical T) (T, error)
+}
+
+// GroupProvenance identifies the complete input contribution of one plan
+// group without prescribing domain warning or retry policy.
+type GroupProvenance struct {
+	memberPositions      []int
+	canonicalPosition    int
+	originalInputIndexes []int
+	earliestPosition     int
+}
+
+// MemberPositions returns owned record positions in ascending order.
+func (provenance GroupProvenance) MemberPositions() []int {
+	return cloneSlice(provenance.memberPositions)
+}
+
+// CanonicalPosition returns the plan-selected canonical record position.
+func (provenance GroupProvenance) CanonicalPosition() int {
+	return provenance.canonicalPosition
+}
+
+// OriginalInputIndexes returns the sorted union contributed by all members.
+func (provenance GroupProvenance) OriginalInputIndexes() []int {
+	return cloneSlice(provenance.originalInputIndexes)
+}
+
+// EarliestInputPosition returns the earliest member input position.
+func (provenance GroupProvenance) EarliestInputPosition() int {
+	return provenance.earliestPosition
+}
+
+// AppliedGroup records the provenance of one successfully consolidated group.
+type AppliedGroup struct {
+	provenance GroupProvenance
+}
+
+// Provenance returns an independently owned provenance snapshot.
+func (event AppliedGroup) Provenance() GroupProvenance {
+	return cloneGroupProvenance(event.provenance)
+}
+
+// RejectedGroup records a typed guard decision while leaving warning and retry
+// construction to the consuming domain.
+type RejectedGroup struct {
+	category   RejectionCategory
+	provenance GroupProvenance
+}
+
+// Category returns the stable neutral rejection category.
+func (event RejectedGroup) Category() RejectionCategory { return event.category }
+
+// Provenance returns an independently owned provenance snapshot.
+func (event RejectedGroup) Provenance() GroupProvenance {
+	return cloneGroupProvenance(event.provenance)
+}
+
+// ApplicationResult owns the ordered records and neutral events from one plan
+// application.
+type ApplicationResult[T any] struct {
+	records        []Record[T]
+	appliedGroups  []AppliedGroup
+	rejectedGroups []RejectedGroup
+}
+
+// Records returns independently owned records in earliest-contribution order.
+func (result ApplicationResult[T]) Records() []Record[T] {
+	return cloneRecords(result.records)
+}
+
+// AppliedGroups returns independently owned applied-group events.
+func (result ApplicationResult[T]) AppliedGroups() []AppliedGroup {
+	events := make([]AppliedGroup, len(result.appliedGroups))
+	for index, event := range result.appliedGroups {
+		events[index] = AppliedGroup{provenance: cloneGroupProvenance(event.provenance)}
+	}
+	return preserveEmptySlice(result.appliedGroups, events)
+}
+
+// RejectedGroups returns independently owned rejected-group events.
+func (result ApplicationResult[T]) RejectedGroups() []RejectedGroup {
+	events := make([]RejectedGroup, len(result.rejectedGroups))
+	for index, event := range result.rejectedGroups {
+		events[index] = RejectedGroup{category: event.category, provenance: cloneGroupProvenance(event.provenance)}
+	}
+	return preserveEmptySlice(result.rejectedGroups, events)
+}
+
+type applicationEntry[T any] struct {
+	record Record[T]
+	order  int
+}
+
+// ApplyPlan applies safe, non-overlapping groups without mutating the plan,
+// records, or values supplied to policy callbacks.
+func ApplyPlan[T any](plan Plan, records []Record[T], policy ApplicationPolicy[T]) (ApplicationResult[T], error) {
+	if err := validateApplicationPolicy(policy); err != nil {
+		return ApplicationResult[T]{}, err
+	}
+	for index, record := range records {
+		if err := validateRecord(record); err != nil {
+			return ApplicationResult[T]{}, fmt.Errorf("apply semantic reconciliation plan: record %d: %w", index, err)
+		}
+	}
+
+	groups := plan.Groups()
+	groupByFirstMember, err := validateApplicationPlan(groups, len(records))
+	if err != nil {
+		return ApplicationResult[T]{}, err
+	}
+	groupedPositions := make(map[int]struct{}, len(records))
+	for _, group := range groups {
+		for _, position := range group.memberPositions {
+			groupedPositions[position] = struct{}{}
+		}
+	}
+
+	entries := make([]applicationEntry[T], 0, len(records))
+	result := ApplicationResult[T]{}
+	for position, record := range records {
+		group, firstMember := groupByFirstMember[position]
+		if !firstMember {
+			if _, grouped := groupedPositions[position]; grouped {
+				continue
+			}
+			entries = append(entries, applicationEntry[T]{record: cloneRecordWith(record, policy.CloneValue), order: position})
+			continue
+		}
+
+		provenance := groupProvenance(group, records)
+		guardMembers, guardCanonical := policyInputs(group, records, policy.CloneValue)
+		category := RejectionCategory("")
+		if policy.RejectGroup != nil {
+			category = policy.RejectGroup(guardMembers, guardCanonical)
+		}
+		if category != "" && strings.TrimSpace(string(category)) == "" {
+			return ApplicationResult[T]{}, fmt.Errorf("apply semantic reconciliation group beginning at position %d: rejection category must not be blank", position)
+		}
+		if category != "" {
+			for _, memberPosition := range group.memberPositions {
+				entries = append(entries, applicationEntry[T]{record: cloneRecordWith(records[memberPosition], policy.CloneValue), order: memberPosition})
+			}
+			result.rejectedGroups = append(result.rejectedGroups, RejectedGroup{category: category, provenance: provenance})
+			continue
+		}
+
+		members, canonical := policyInputs(group, records, policy.CloneValue)
+		consolidated, err := policy.ConsolidateGroup(members, canonical)
+		if err != nil {
+			return ApplicationResult[T]{}, fmt.Errorf("apply semantic reconciliation group beginning at position %d: consolidate: %w", position, err)
+		}
+		owned, err := NewRecord(consolidated, provenance.originalInputIndexes, provenance.earliestPosition, policy.CloneValue)
+		if err != nil {
+			return ApplicationResult[T]{}, fmt.Errorf("apply semantic reconciliation group beginning at position %d: own consolidated record: %w", position, err)
+		}
+		entries = append(entries, applicationEntry[T]{record: owned, order: position})
+		result.appliedGroups = append(result.appliedGroups, AppliedGroup{provenance: provenance})
+	}
+
+	sort.SliceStable(entries, func(left, right int) bool {
+		if entries[left].record.earliestInputPosition == entries[right].record.earliestInputPosition {
+			return entries[left].order < entries[right].order
+		}
+		return entries[left].record.earliestInputPosition < entries[right].record.earliestInputPosition
+	})
+	if records != nil {
+		result.records = make([]Record[T], len(entries))
+		for index, entry := range entries {
+			result.records[index] = cloneRecord(entry.record)
+		}
+	}
+	return result, nil
+}
+
+func validateApplicationPolicy[T any](policy ApplicationPolicy[T]) error {
+	if policy.CloneValue == nil {
+		return fmt.Errorf("apply semantic reconciliation plan: clone value function must not be nil")
+	}
+	if policy.ConsolidateGroup == nil {
+		return fmt.Errorf("apply semantic reconciliation plan: consolidate group function must not be nil")
+	}
+	return nil
+}
+
+func validateRecord[T any](record Record[T]) error {
+	if record.cloneValue == nil {
+		return fmt.Errorf("invalid construction state: clone value function must not be nil")
+	}
+	if record.earliestInputPosition < 0 {
+		return fmt.Errorf("invalid construction state: earliest input position must not be negative")
+	}
+	for index, value := range record.originalInputIndexes {
+		if value < 0 {
+			return fmt.Errorf("invalid construction state: original input index must not be negative")
+		}
+		if index > 0 && record.originalInputIndexes[index-1] >= value {
+			return fmt.Errorf("invalid construction state: original input indexes must be ascending and unique")
+		}
+	}
+	return nil
+}
+
+func validateApplicationPlan(groups []PlanGroup, recordCount int) (map[int]PlanGroup, error) {
+	groupByFirstMember := make(map[int]PlanGroup, len(groups))
+	used := make(map[int]struct{})
+	for groupIndex, group := range groups {
+		if len(group.memberPositions) < 2 {
+			return nil, fmt.Errorf("apply semantic reconciliation plan: group %d must contain at least two member positions", groupIndex)
+		}
+		canonicalMember := false
+		for memberIndex, position := range group.memberPositions {
+			if position < 0 || position >= recordCount {
+				return nil, fmt.Errorf("apply semantic reconciliation plan: group %d member position %d is outside record range [0,%d)", groupIndex, position, recordCount)
+			}
+			if memberIndex > 0 && group.memberPositions[memberIndex-1] >= position {
+				return nil, fmt.Errorf("apply semantic reconciliation plan: group %d member positions must be ascending and unique", groupIndex)
+			}
+			if _, exists := used[position]; exists {
+				return nil, fmt.Errorf("apply semantic reconciliation plan: record position %d belongs to multiple groups", position)
+			}
+			used[position] = struct{}{}
+			canonicalMember = canonicalMember || position == group.canonicalPosition
+		}
+		if group.canonicalPosition < 0 || group.canonicalPosition >= recordCount {
+			return nil, fmt.Errorf("apply semantic reconciliation plan: group %d canonical position %d is outside record range [0,%d)", groupIndex, group.canonicalPosition, recordCount)
+		}
+		if !canonicalMember {
+			return nil, fmt.Errorf("apply semantic reconciliation plan: group %d canonical position %d is not a member", groupIndex, group.canonicalPosition)
+		}
+		groupByFirstMember[group.memberPositions[0]] = group
+	}
+	return groupByFirstMember, nil
+}
+
+func groupProvenance[T any](group PlanGroup, records []Record[T]) GroupProvenance {
+	provenance := GroupProvenance{
+		memberPositions:   cloneSlice(group.memberPositions),
+		canonicalPosition: group.canonicalPosition,
+		earliestPosition:  records[group.memberPositions[0]].earliestInputPosition,
+	}
+	for _, position := range group.memberPositions {
+		provenance.originalInputIndexes = append(provenance.originalInputIndexes, records[position].originalInputIndexes...)
+		if records[position].earliestInputPosition < provenance.earliestPosition {
+			provenance.earliestPosition = records[position].earliestInputPosition
+		}
+	}
+	provenance.originalInputIndexes, _ = normalizeInputIndexes(provenance.originalInputIndexes)
+	return provenance
+}
+
+func policyInputs[T any](group PlanGroup, records []Record[T], cloneValue CloneValueFunc[T]) ([]T, T) {
+	members := make([]T, len(group.memberPositions))
+	for index, position := range group.memberPositions {
+		members[index] = cloneValue(records[position].value)
+	}
+	return members, cloneValue(records[group.canonicalPosition].value)
+}
+
+func normalizeInputIndexes(indexes []int) ([]int, error) {
+	if indexes == nil {
+		return nil, nil
+	}
+	normalized := append([]int{}, indexes...)
+	for _, index := range normalized {
+		if index < 0 {
+			return nil, fmt.Errorf("original input index must not be negative")
+		}
+	}
+	sort.Ints(normalized)
+	write := 0
+	for _, index := range normalized {
+		if write > 0 && normalized[write-1] == index {
+			continue
+		}
+		normalized[write] = index
+		write++
+	}
+	return normalized[:write], nil
+}
+
+func cloneRecord[T any](record Record[T]) Record[T] {
+	return cloneRecordWith(record, record.cloneValue)
+}
+
+func cloneRecordWith[T any](record Record[T], cloneValue CloneValueFunc[T]) Record[T] {
+	return Record[T]{
+		value:                 cloneValue(record.value),
+		originalInputIndexes:  cloneSlice(record.originalInputIndexes),
+		earliestInputPosition: record.earliestInputPosition,
+		cloneValue:            cloneValue,
+	}
+}
+
+func cloneRecords[T any](records []Record[T]) []Record[T] {
+	if records == nil {
+		return nil
+	}
+	cloned := make([]Record[T], len(records))
+	for index, record := range records {
+		cloned[index] = cloneRecord(record)
+	}
+	return cloned
+}
+
+func cloneGroupProvenance(provenance GroupProvenance) GroupProvenance {
+	return GroupProvenance{
+		memberPositions:      cloneSlice(provenance.memberPositions),
+		canonicalPosition:    provenance.canonicalPosition,
+		originalInputIndexes: cloneSlice(provenance.originalInputIndexes),
+		earliestPosition:     provenance.earliestPosition,
+	}
+}
+
+func cloneSlice[T any](values []T) []T {
+	if values == nil {
+		return nil
+	}
+	return append([]T{}, values...)
+}
+
+func preserveEmptySlice[S ~[]E, E any](source S, cloned []E) []E {
+	if source == nil {
+		return nil
+	}
+	return cloned
+}
--- a/internal/framework/semanticreconcile/application_test.go
+++ b/internal/framework/semanticreconcile/application_test.go
@@ -0,0 +1,376 @@
+package semanticreconcile
+
+import (
+	"errors"
+	"reflect"
+	"strings"
+	"testing"
+)
+
+type syntheticRecord struct {
+	Name  string
+	Notes []string
+}
+
+func cloneSyntheticRecord(record syntheticRecord) syntheticRecord {
+	record.Notes = cloneSlice(record.Notes)
+	return record
+}
+
+func TestNewRecordOwnsValueAndNormalizesProvenance(t *testing.T) {
+	value := syntheticRecord{Name: "alpha", Notes: []string{"owned"}}
+	indexes := []int{3, 1, 3}
+	record, err := NewRecord(value, indexes, 4, cloneSyntheticRecord)
+	if err != nil {
+		t.Fatalf("NewRecord() error = %v", err)
+	}
+
+	value.Notes[0] = "caller mutation"
+	indexes[0] = 99
+	gotValue := record.Value()
+	gotIndexes := record.OriginalInputIndexes()
+	if gotValue.Name != "alpha" || !reflect.DeepEqual(gotValue.Notes, []string{"owned"}) {
+		t.Fatalf("Value() = %#v, want owned original", gotValue)
+	}
+	if !reflect.DeepEqual(gotIndexes, []int{1, 3}) {
+		t.Fatalf("OriginalInputIndexes() = %v, want [1 3]", gotIndexes)
+	}
+	if record.EarliestInputPosition() != 4 {
+		t.Fatalf("EarliestInputPosition() = %d, want 4", record.EarliestInputPosition())
+	}
+
+	gotValue.Notes[0] = "accessor mutation"
+	gotIndexes[0] = 88
+	if again := record.Value(); again.Notes[0] != "owned" {
+		t.Fatalf("Value() retained accessor mutation: %#v", again)
+	}
+	if again := record.OriginalInputIndexes(); !reflect.DeepEqual(again, []int{1, 3}) {
+		t.Fatalf("OriginalInputIndexes() retained accessor mutation: %v", again)
+	}
+}
+
+func TestNewRecordPreservesNilAndEmptyIndexOwnership(t *testing.T) {
+	nilIndexes, err := NewRecord(syntheticRecord{}, nil, 0, cloneSyntheticRecord)
+	if err != nil {
+		t.Fatalf("NewRecord(nil) error = %v", err)
+	}
+	emptyIndexes, err := NewRecord(syntheticRecord{}, []int{}, 0, cloneSyntheticRecord)
+	if err != nil {
+		t.Fatalf("NewRecord(empty) error = %v", err)
+	}
+	if nilIndexes.OriginalInputIndexes() != nil {
+		t.Fatal("nil original indexes became non-nil")
+	}
+	if got := emptyIndexes.OriginalInputIndexes(); got == nil || len(got) != 0 {
+		t.Fatalf("empty original indexes = %#v, want non-nil empty", got)
+	}
+
+	if _, err := NewRecord(syntheticRecord{}, nil, 0, CloneValueFunc[syntheticRecord](nil)); err == nil {
+		t.Fatal("NewRecord() accepted nil clone function")
+	}
+	if _, err := NewRecord(syntheticRecord{}, nil, -1, cloneSyntheticRecord); err == nil {
+		t.Fatal("NewRecord() accepted negative earliest position")
+	}
+	if _, err := NewRecord(syntheticRecord{}, []int{-1}, 0, cloneSyntheticRecord); err == nil {
+		t.Fatal("NewRecord() accepted negative original input index")
+	}
+}
+
+func TestApplyPlanConsolidatesGroupsAndOrdersByEarliestContribution(t *testing.T) {
+	records := syntheticRecords(t,
+		recordFixture{"alpha", []int{4}, 4},
+		recordFixture{"bravo", []int{3, 1}, 1},
+		recordFixture{"charlie", []int{2}, 2},
+		recordFixture{"delta", []int{2, 0}, 0},
+	)
+	plan := Plan{groups: []PlanGroup{
+		{memberPositions: []int{0, 2}, canonicalPosition: 2},
+		{memberPositions: []int{1, 3}, canonicalPosition: 1},
+	}}
+	var canonicalNames []string
+	result, err := ApplyPlan(plan, records, ApplicationPolicy[syntheticRecord]{
+		CloneValue: cloneSyntheticRecord,
+		ConsolidateGroup: func(members []syntheticRecord, canonical syntheticRecord) (syntheticRecord, error) {
+			canonicalNames = append(canonicalNames, canonical.Name)
+			canonical.Notes = []string{members[0].Name, members[1].Name}
+			return canonical, nil
+		},
+	})
+	if err != nil {
+		t.Fatalf("ApplyPlan() error = %v", err)
+	}
+
+	got := result.Records()
+	if names := recordNames(got); !reflect.DeepEqual(names, []string{"bravo", "charlie"}) {
+		t.Fatalf("record names = %v, want [bravo charlie]", names)
+	}
+	if !reflect.DeepEqual(canonicalNames, []string{"charlie", "bravo"}) {
+		t.Fatalf("canonical values = %v, want [charlie bravo]", canonicalNames)
+	}
+	if indexes := got[0].OriginalInputIndexes(); !reflect.DeepEqual(indexes, []int{0, 1, 2, 3}) {
+		t.Fatalf("first provenance indexes = %v, want [0 1 2 3]", indexes)
+	}
+	if indexes := got[1].OriginalInputIndexes(); !reflect.DeepEqual(indexes, []int{2, 4}) {
+		t.Fatalf("second provenance indexes = %v, want [2 4]", indexes)
+	}
+	if got[0].EarliestInputPosition() != 0 || got[1].EarliestInputPosition() != 2 {
+		t.Fatalf("earliest positions = [%d %d], want [0 2]", got[0].EarliestInputPosition(), got[1].EarliestInputPosition())
+	}
+
+	events := result.AppliedGroups()
+	if len(events) != 2 || len(result.RejectedGroups()) != 0 {
+		t.Fatalf("event counts = applied %d rejected %d, want 2 and 0", len(events), len(result.RejectedGroups()))
+	}
+	first := events[0].Provenance()
+	if !reflect.DeepEqual(first.MemberPositions(), []int{0, 2}) || first.CanonicalPosition() != 2 || !reflect.DeepEqual(first.OriginalInputIndexes(), []int{2, 4}) || first.EarliestInputPosition() != 2 {
+		t.Fatalf("first applied provenance = members %v canonical %d indexes %v earliest %d", first.MemberPositions(), first.CanonicalPosition(), first.OriginalInputIndexes(), first.EarliestInputPosition())
+	}
+}
+
+func TestApplyPlanWithoutGroupsReturnsOwnedRecordsInProvenanceOrder(t *testing.T) {
+	records := syntheticRecords(t,
+		recordFixture{"alpha", []int{2}, 2},
+		recordFixture{"bravo", []int{0}, 0},
+		recordFixture{"charlie", []int{1}, 1},
+	)
+	result, err := ApplyPlan(Plan{}, records, syntheticPolicy())
+	if err != nil {
+		t.Fatalf("ApplyPlan() error = %v", err)
+	}
+	if names := recordNames(result.Records()); !reflect.DeepEqual(names, []string{"bravo", "charlie", "alpha"}) {
+		t.Fatalf("record names = %v, want [bravo charlie alpha]", names)
+	}
+	if result.AppliedGroups() != nil || result.RejectedGroups() != nil {
+		t.Fatalf("events = applied %#v rejected %#v, want nil", result.AppliedGroups(), result.RejectedGroups())
+	}
+
+	value := result.Records()[0].Value()
+	value.Notes[0] = "changed"
+	if records[1].Value().Notes[0] != "bravo" || result.Records()[0].Value().Notes[0] != "bravo" {
+		t.Fatal("no-group result shares mutable value state")
+	}
+}
+
+func TestApplyPlanPreservesUngroupedRecordsAndOwnsResults(t *testing.T) {
+	records := syntheticRecords(t,
+		recordFixture{"alpha", []int{4}, 4},
+		recordFixture{"bravo", []int{1}, 1},
+		recordFixture{"charlie", []int{2}, 2},
+	)
+	result, err := ApplyPlan(Plan{groups: []PlanGroup{{memberPositions: []int{0, 2}, canonicalPosition: 0}}}, records, syntheticPolicy())
+	if err != nil {
+		t.Fatalf("ApplyPlan() error = %v", err)
+	}
+	if names := recordNames(result.Records()); !reflect.DeepEqual(names, []string{"bravo", "alpha"}) {
+		t.Fatalf("record names = %v, want ungrouped bravo then consolidated alpha", names)
+	}
+
+	firstRead := result.Records()
+	firstValue := firstRead[0].Value()
+	firstValue.Notes[0] = "changed"
+	firstRead[0].originalInputIndexes[0] = 99
+	if again := result.Records(); again[0].Value().Notes[0] != "bravo" || !reflect.DeepEqual(again[0].OriginalInputIndexes(), []int{1}) {
+		t.Fatalf("result retained accessor mutations: %#v", again[0])
+	}
+	if original := records[1].Value(); original.Notes[0] != "bravo" {
+		t.Fatalf("input record was mutated: %#v", original)
+	}
+}
+
+func TestApplyPlanGuardRejectionPreservesEveryMemberOnce(t *testing.T) {
+	records := syntheticRecords(t,
+		recordFixture{"alpha", []int{3}, 3},
+		recordFixture{"bravo", []int{1}, 1},
+		recordFixture{"charlie", []int{2}, 2},
+	)
+	consolidations := 0
+	result, err := ApplyPlan(Plan{groups: []PlanGroup{{memberPositions: []int{0, 2}, canonicalPosition: 2}}}, records, ApplicationPolicy[syntheticRecord]{
+		CloneValue: cloneSyntheticRecord,
+		RejectGroup: func(members []syntheticRecord, canonical syntheticRecord) RejectionCategory {
+			members[0].Notes[0] = "guard mutation"
+			canonical.Notes[0] = "canonical mutation"
+			return "typed_constraint"
+		},
+		ConsolidateGroup: func([]syntheticRecord, syntheticRecord) (syntheticRecord, error) {
+			consolidations++
+			return syntheticRecord{}, nil
+		},
+	})
+	if err != nil {
+		t.Fatalf("ApplyPlan() error = %v", err)
+	}
+	if consolidations != 0 {
+		t.Fatalf("consolidation calls = %d, want 0", consolidations)
+	}
+	if names := recordNames(result.Records()); !reflect.DeepEqual(names, []string{"bravo", "charlie", "alpha"}) {
+		t.Fatalf("preserved record names = %v, want [bravo charlie alpha]", names)
+	}
+	for index, record := range records {
+		if got := record.Value().Notes[0]; got != record.Value().Name {
+			t.Fatalf("input record %d note = %q after guard, want original", index, got)
+		}
+	}
+	rejected := result.RejectedGroups()
+	if len(rejected) != 1 || rejected[0].Category() != "typed_constraint" {
+		t.Fatalf("rejected events = %#v, want typed_constraint", rejected)
+	}
+	provenance := rejected[0].Provenance()
+	if !reflect.DeepEqual(provenance.MemberPositions(), []int{0, 2}) || !reflect.DeepEqual(provenance.OriginalInputIndexes(), []int{2, 3}) || provenance.EarliestInputPosition() != 2 {
+		t.Fatalf("rejected provenance = members %v indexes %v earliest %d", provenance.MemberPositions(), provenance.OriginalInputIndexes(), provenance.EarliestInputPosition())
+	}
+}
+
+func TestApplyPlanSuppliesFreshPolicyValuesAndDoesNotMutateInputs(t *testing.T) {
+	records := syntheticRecords(t,
+		recordFixture{"alpha", []int{0}, 0},
+		recordFixture{"bravo", []int{1}, 1},
+	)
+	result, err := ApplyPlan(Plan{groups: []PlanGroup{{memberPositions: []int{0, 1}, canonicalPosition: 1}}}, records, ApplicationPolicy[syntheticRecord]{
+		CloneValue: cloneSyntheticRecord,
+		RejectGroup: func(members []syntheticRecord, canonical syntheticRecord) RejectionCategory {
+			members[0].Name = "guard mutation"
+			canonical.Name = "guard canonical mutation"
+			return ""
+		},
+		ConsolidateGroup: func(members []syntheticRecord, canonical syntheticRecord) (syntheticRecord, error) {
+			if members[0].Name != "alpha" || canonical.Name != "bravo" {
+				t.Fatalf("consolidation observed guard mutations: members %#v canonical %#v", members, canonical)
+			}
+			members[0].Notes[0] = "consolidator mutation"
+			canonical.Notes = []string{"result"}
+			return canonical, nil
+		},
+	})
+	if err != nil {
+		t.Fatalf("ApplyPlan() error = %v", err)
+	}
+	if got := result.Records()[0].Value(); got.Name != "bravo" || !reflect.DeepEqual(got.Notes, []string{"result"}) {
+		t.Fatalf("consolidated value = %#v", got)
+	}
+	if records[0].Value().Notes[0] != "alpha" || records[1].Value().Notes[0] != "bravo" {
+		t.Fatalf("input records changed: %#v %#v", records[0].Value(), records[1].Value())
+	}
+}
+
+func TestApplyPlanRejectsMalformedPlans(t *testing.T) {
+	records := syntheticRecords(t,
+		recordFixture{"alpha", nil, 0},
+		recordFixture{"bravo", nil, 1},
+		recordFixture{"charlie", nil, 2},
+	)
+	tests := []struct {
+		name string
+		plan Plan
+		want string
+	}{
+		{name: "member out of range", plan: Plan{groups: []PlanGroup{{memberPositions: []int{0, 3}, canonicalPosition: 0}}}, want: "outside record range"},
+		{name: "canonical out of range", plan: Plan{groups: []PlanGroup{{memberPositions: []int{0, 1}, canonicalPosition: 3}}}, want: "canonical position"},
+		{name: "canonical not a member", plan: Plan{groups: []PlanGroup{{memberPositions: []int{0, 1}, canonicalPosition: 2}}}, want: "is not a member"},
+		{name: "duplicate member", plan: Plan{groups: []PlanGroup{{memberPositions: []int{0, 0}, canonicalPosition: 0}}}, want: "ascending and unique"},
+		{name: "overlapping groups", plan: Plan{groups: []PlanGroup{{memberPositions: []int{0, 1}, canonicalPosition: 0}, {memberPositions: []int{1, 2}, canonicalPosition: 1}}}, want: "multiple groups"},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			_, err := ApplyPlan(test.plan, records, syntheticPolicy())
+			if err == nil || !strings.Contains(err.Error(), test.want) {
+				t.Fatalf("ApplyPlan() error = %v, want containing %q", err, test.want)
+			}
+		})
+	}
+}
+
+func TestApplyPlanReturnsConsolidationFailures(t *testing.T) {
+	records := syntheticRecords(t,
+		recordFixture{"alpha", nil, 0},
+		recordFixture{"bravo", nil, 1},
+	)
+	want := errors.New("cannot consolidate")
+	result, err := ApplyPlan(Plan{groups: []PlanGroup{{memberPositions: []int{0, 1}, canonicalPosition: 0}}}, records, ApplicationPolicy[syntheticRecord]{
+		CloneValue: cloneSyntheticRecord,
+		ConsolidateGroup: func([]syntheticRecord, syntheticRecord) (syntheticRecord, error) {
+			return syntheticRecord{}, want
+		},
+	})
+	if !errors.Is(err, want) || !strings.Contains(err.Error(), "position 0") {
+		t.Fatalf("ApplyPlan() error = %v, want contextual wrapped failure", err)
+	}
+	if result.Records() != nil || result.AppliedGroups() != nil || result.RejectedGroups() != nil {
+		t.Fatalf("ApplyPlan() partial result = %#v, want zero result", result)
+	}
+}
+
+func TestApplyPlanPreservesNilAndEmptyRecordCollections(t *testing.T) {
+	policy := syntheticPolicy()
+	nilResult, err := ApplyPlan(Plan{}, []Record[syntheticRecord](nil), policy)
+	if err != nil {
+		t.Fatalf("ApplyPlan(nil) error = %v", err)
+	}
+	emptyResult, err := ApplyPlan(Plan{}, []Record[syntheticRecord]{}, policy)
+	if err != nil {
+		t.Fatalf("ApplyPlan(empty) error = %v", err)
+	}
+	if nilResult.Records() != nil {
+		t.Fatal("nil records became non-nil")
+	}
+	if got := emptyResult.Records(); got == nil || len(got) != 0 {
+		t.Fatalf("empty records = %#v, want non-nil empty", got)
+	}
+}
+
+func TestApplyPlanValidatesPolicyAndRecordState(t *testing.T) {
+	valid := syntheticPolicy()
+	if _, err := ApplyPlan(Plan{}, []Record[syntheticRecord]{}, ApplicationPolicy[syntheticRecord]{ConsolidateGroup: valid.ConsolidateGroup}); err == nil {
+		t.Fatal("ApplyPlan() accepted nil clone function")
+	}
+	if _, err := ApplyPlan(Plan{}, []Record[syntheticRecord]{}, ApplicationPolicy[syntheticRecord]{CloneValue: cloneSyntheticRecord}); err == nil {
+		t.Fatal("ApplyPlan() accepted nil consolidate function")
+	}
+	if _, err := ApplyPlan(Plan{}, []Record[syntheticRecord]{{}}, valid); err == nil || !strings.Contains(err.Error(), "record 0") {
+		t.Fatalf("ApplyPlan() invalid record error = %v", err)
+	}
+
+	records := syntheticRecords(t,
+		recordFixture{"alpha", nil, 0},
+		recordFixture{"bravo", nil, 1},
+	)
+	valid.RejectGroup = func([]syntheticRecord, syntheticRecord) RejectionCategory { return " " }
+	if _, err := ApplyPlan(Plan{groups: []PlanGroup{{memberPositions: []int{0, 1}, canonicalPosition: 0}}}, records, valid); err == nil || !strings.Contains(err.Error(), "category") {
+		t.Fatalf("ApplyPlan() blank category error = %v", err)
+	}
+}
+
+type recordFixture struct {
+	name     string
+	indexes  []int
+	earliest int
+}
+
+func syntheticRecords(t *testing.T, fixtures ...recordFixture) []Record[syntheticRecord] {
+	t.Helper()
+	records := make([]Record[syntheticRecord], len(fixtures))
+	for index, fixture := range fixtures {
+		record, err := NewRecord(syntheticRecord{Name: fixture.name, Notes: []string{fixture.name}}, fixture.indexes, fixture.earliest, cloneSyntheticRecord)
+		if err != nil {
+			t.Fatalf("NewRecord(%d) error = %v", index, err)
+		}
+		records[index] = record
+	}
+	return records
+}
+
+func syntheticPolicy() ApplicationPolicy[syntheticRecord] {
+	return ApplicationPolicy[syntheticRecord]{
+		CloneValue: cloneSyntheticRecord,
+		ConsolidateGroup: func(_ []syntheticRecord, canonical syntheticRecord) (syntheticRecord, error) {
+			return canonical, nil
+		},
+	}
+}
+
+func recordNames(records []Record[syntheticRecord]) []string {
+	names := make([]string, len(records))
+	for index, record := range records {
+		names[index] = record.Value().Name
+	}
+	return names
+}