notarius/internal/framework/semanticreconcile/proposal.go

package semanticreconcile

import (
	"fmt"
	"sort"
)

// ProposalResponse is the complete private structured response contract.
type ProposalResponse struct {
	DuplicateGroups []DuplicateGroup `json:"duplicate_groups"`
}

// DuplicateGroup proposes supplied request-local candidate IDs that may denote
// one entity and identifies one supplied member as canonical.
type DuplicateGroup struct {
	CandidateIDs         []int `json:"candidate_ids"`
	CanonicalCandidateID int   `json:"canonical_candidate_id"`
}

// IssueCategory identifies one stable proposal safety failure.
type IssueCategory string

const (
	IssueMemberNonPositive    IssueCategory = "member_non_positive"
	IssueMemberUnknown        IssueCategory = "member_unknown"
	IssueRepeatedMember       IssueCategory = "repeated_member"
	IssueFewerThanTwoMembers  IssueCategory = "fewer_than_two_members"
	IssueCanonicalNonPositive IssueCategory = "canonical_non_positive"
	IssueCanonicalUnknown     IssueCategory = "canonical_unknown"
	IssueCanonicalNotMember   IssueCategory = "canonical_not_member"
	IssueOverlappingMember    IssueCategory = "overlapping_member"
)

// Issue identifies an unsafe proposal category at its original response group
// index without prescribing caller warning text.
type Issue struct {
	GroupIndex int
	Category   IssueCategory
}

// IssueDetails renders stable, domain-neutral proposal diagnostics for an
// adapter's retry message.
func IssueDetails(issues []Issue) []string {
	details := make([]string, len(issues))
	for index, issue := range issues {
		details[index] = fmt.Sprintf("group %d: %s", issue.GroupIndex, issue.Category)
	}
	return details
}

// PlanGroup identifies one validated group using original candidate positions.
type PlanGroup struct {
	memberPositions   []int
	canonicalPosition int
}

// MemberPositions returns an owned, ascending list of original candidate
// positions.
func (group PlanGroup) MemberPositions() []int {
	return append([]int(nil), group.memberPositions...)
}

// CanonicalPosition returns the original position of the selected canonical
// candidate.
func (group PlanGroup) CanonicalPosition() int {
	return group.canonicalPosition
}

// Plan contains deterministic, non-overlapping reconciliation groups.
type Plan struct {
	groups []PlanGroup
}

// Groups returns a deeply owned copy ordered by each group's earliest member.
func (plan Plan) Groups() []PlanGroup {
	groups := make([]PlanGroup, len(plan.groups))
	for index, group := range plan.groups {
		groups[index] = clonePlanGroup(group)
	}
	return groups
}

// Assessment contains the safe plan and stable diagnostics for discarded
// response groups.
type Assessment struct {
	plan                Plan
	discardedGroupCount int
	issues              []Issue
}

// Plan returns an independently owned reconciliation plan.
func (assessment Assessment) Plan() Plan {
	groups := assessment.plan.Groups()
	return Plan{groups: groups}
}

// Issues returns an owned copy ordered by original response group index.
func (assessment Assessment) Issues() []Issue {
	return append([]Issue(nil), assessment.issues...)
}

// DiscardedGroupCount returns the number of response groups excluded from the
// safe plan.
func (assessment Assessment) DiscardedGroupCount() int {
	return assessment.discardedGroupCount
}

// RetryRequired reports whether any response group was discarded.
func (assessment Assessment) RetryRequired() bool {
	return assessment.discardedGroupCount > 0
}

type assessedGroup struct {
	memberPositions   []int
	canonicalPosition int
	issues            []IssueCategory
	locallyValid      bool
	conflicting       bool
}

// Assess resolves request-local IDs through the retained preparation mapping
// and returns only deterministic, non-overlapping groups.
func (preparation Preparation) Assess(response ProposalResponse) Assessment {
	positionsByID := make(map[int]int, len(preparation.mappings))
	for _, mapping := range preparation.mappings {
		positionsByID[mapping.CandidateID] = mapping.CandidatePosition
	}

	groups := make([]assessedGroup, len(response.DuplicateGroups))
	owners := make(map[int][]int)
	for groupIndex, proposal := range response.DuplicateGroups {
		groups[groupIndex] = assessGroup(proposal, positionsByID)
		if !groups[groupIndex].locallyValid {
			continue
		}
		for _, position := range groups[groupIndex].memberPositions {
			owners[position] = append(owners[position], groupIndex)
		}
	}
	for _, groupIndexes := range owners {
		if len(groupIndexes) < 2 {
			continue
		}
		for _, groupIndex := range groupIndexes {
			groups[groupIndex].conflicting = true
		}
	}

	assessment := Assessment{}
	for groupIndex, group := range groups {
		for _, category := range group.issues {
			assessment.issues = append(assessment.issues, Issue{GroupIndex: groupIndex, Category: category})
		}
		if group.conflicting {
			assessment.issues = append(assessment.issues, Issue{GroupIndex: groupIndex, Category: IssueOverlappingMember})
		}
		if !group.locallyValid || group.conflicting {
			assessment.discardedGroupCount++
			continue
		}
		assessment.plan.groups = append(assessment.plan.groups, PlanGroup{
			memberPositions:   append([]int(nil), group.memberPositions...),
			canonicalPosition: group.canonicalPosition,
		})
	}
	sort.Slice(assessment.plan.groups, func(left, right int) bool {
		return assessment.plan.groups[left].memberPositions[0] < assessment.plan.groups[right].memberPositions[0]
	})
	return assessment
}

func assessGroup(proposal DuplicateGroup, positionsByID map[int]int) assessedGroup {
	group := assessedGroup{}
	seenIDs := make(map[int]struct{}, len(proposal.CandidateIDs))
	memberPositions := make(map[int]struct{}, len(proposal.CandidateIDs))
	for _, candidateID := range proposal.CandidateIDs {
		if _, repeated := seenIDs[candidateID]; repeated {
			group.issues = append(group.issues, IssueRepeatedMember)
			continue
		}
		seenIDs[candidateID] = struct{}{}
		position, category := resolveMember(candidateID, positionsByID)
		if category != "" {
			group.issues = append(group.issues, category)
			continue
		}
		memberPositions[position] = struct{}{}
		group.memberPositions = append(group.memberPositions, position)
	}
	if len(memberPositions) < 2 {
		group.issues = append(group.issues, IssueFewerThanTwoMembers)
	}

	canonicalPosition, canonicalCategory := resolveCanonical(proposal.CanonicalCandidateID, positionsByID)
	if canonicalCategory != "" {
		group.issues = append(group.issues, canonicalCategory)
	} else {
		group.canonicalPosition = canonicalPosition
		if _, member := memberPositions[canonicalPosition]; !member {
			group.issues = append(group.issues, IssueCanonicalNotMember)
		}
	}

	sort.Ints(group.memberPositions)
	group.locallyValid = len(group.issues) == 0
	return group
}

func resolveMember(candidateID int, positionsByID map[int]int) (int, IssueCategory) {
	if candidateID <= 0 {
		return 0, IssueMemberNonPositive
	}
	position, exists := positionsByID[candidateID]
	if !exists {
		return 0, IssueMemberUnknown
	}
	return position, ""
}

func resolveCanonical(candidateID int, positionsByID map[int]int) (int, IssueCategory) {
	if candidateID <= 0 {
		return 0, IssueCanonicalNonPositive
	}
	position, exists := positionsByID[candidateID]
	if !exists {
		return 0, IssueCanonicalUnknown
	}
	return position, ""
}

func clonePlanGroup(group PlanGroup) PlanGroup {
	return PlanGroup{
		memberPositions:   append([]int(nil), group.memberPositions...),
		canonicalPosition: group.canonicalPosition,
	}
}