Add semantic reconciliation proposal validation

This commit is contained in:
2026-08-09 16:00:01 +00:00
parent ee71dc4937
commit 297d58f090
5 changed files with 592 additions and 0 deletions

View File

@@ -0,0 +1,33 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "notarius.generic.semantic_reconciliation.llm",
"title": "notarius_semantic_reconciliation_llm_v1",
"type": "object",
"additionalProperties": false,
"required": ["duplicate_groups"],
"properties": {
"duplicate_groups": {
"type": "array",
"items": {
"type": "object",
"additionalProperties": false,
"required": ["candidate_ids", "canonical_candidate_id"],
"properties": {
"candidate_ids": {
"type": "array",
"minItems": 2,
"uniqueItems": true,
"items": {
"type": "integer",
"minimum": 1
}
},
"canonical_candidate_id": {
"type": "integer",
"minimum": 1
}
}
}
}
}
}

View File

@@ -0,0 +1,223 @@
package semanticreconcile
import "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
}
// 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,
}
}

View File

@@ -0,0 +1,188 @@
package semanticreconcile
import (
"reflect"
"testing"
"gitea.maximumdirect.net/eric/notarius/internal/core/source"
)
func TestAssessProducesAStableOriginalPositionPlan(t *testing.T) {
preparation := proposalPreparation(t)
response := ProposalResponse{DuplicateGroups: []DuplicateGroup{
{CandidateIDs: []int{5, 4}, CanonicalCandidateID: 5},
{CandidateIDs: []int{2, 1}, CanonicalCandidateID: 2},
}}
assessment := preparation.Assess(response)
want := []planGroupSnapshot{
{members: []int{0, 2}, canonical: 2},
{members: []int{5, 6}, canonical: 6},
}
if got := snapshotPlan(assessment.Plan()); !reflect.DeepEqual(got, want) {
t.Fatalf("plan = %#v, want %#v", got, want)
}
if assessment.DiscardedGroupCount() != 0 || assessment.RetryRequired() || len(assessment.Issues()) != 0 {
t.Fatalf("assessment diagnostics = discarded %d, retry %t, issues %#v", assessment.DiscardedGroupCount(), assessment.RetryRequired(), assessment.Issues())
}
reordered := preparation.Assess(ProposalResponse{DuplicateGroups: []DuplicateGroup{
{CandidateIDs: []int{1, 2}, CanonicalCandidateID: 2},
{CandidateIDs: []int{4, 5}, CanonicalCandidateID: 5},
}})
if got := snapshotPlan(reordered.Plan()); !reflect.DeepEqual(got, want) {
t.Fatalf("reordered plan = %#v, want %#v", got, want)
}
empty := preparation.Assess(ProposalResponse{})
if len(empty.Plan().Groups()) != 0 || len(empty.Issues()) != 0 || empty.DiscardedGroupCount() != 0 || empty.RetryRequired() {
t.Fatalf("empty assessment = plan %#v, issues %#v, discarded %d, retry %t", empty.Plan().Groups(), empty.Issues(), empty.DiscardedGroupCount(), empty.RetryRequired())
}
}
func TestAssessRejectsEveryUnsafeLocalGroupShape(t *testing.T) {
preparation := proposalPreparation(t)
tests := []struct {
name string
group DuplicateGroup
category IssueCategory
}{
{name: "zero member", group: DuplicateGroup{CandidateIDs: []int{0, 2}, CanonicalCandidateID: 2}, category: IssueMemberNonPositive},
{name: "negative member", group: DuplicateGroup{CandidateIDs: []int{-1, 2}, CanonicalCandidateID: 2}, category: IssueMemberNonPositive},
{name: "unknown member", group: DuplicateGroup{CandidateIDs: []int{99, 2}, CanonicalCandidateID: 2}, category: IssueMemberUnknown},
{name: "repeated member", group: DuplicateGroup{CandidateIDs: []int{1, 1}, CanonicalCandidateID: 1}, category: IssueRepeatedMember},
{name: "too small", group: DuplicateGroup{CandidateIDs: []int{1}, CanonicalCandidateID: 1}, category: IssueFewerThanTwoMembers},
{name: "zero canonical", group: DuplicateGroup{CandidateIDs: []int{1, 2}, CanonicalCandidateID: 0}, category: IssueCanonicalNonPositive},
{name: "negative canonical", group: DuplicateGroup{CandidateIDs: []int{1, 2}, CanonicalCandidateID: -1}, category: IssueCanonicalNonPositive},
{name: "unknown canonical", group: DuplicateGroup{CandidateIDs: []int{1, 2}, CanonicalCandidateID: 99}, category: IssueCanonicalUnknown},
{name: "canonical not member", group: DuplicateGroup{CandidateIDs: []int{1, 2}, CanonicalCandidateID: 3}, category: IssueCanonicalNotMember},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
assessment := preparation.Assess(ProposalResponse{DuplicateGroups: []DuplicateGroup{test.group}})
if len(assessment.Plan().Groups()) != 0 || assessment.DiscardedGroupCount() != 1 || !assessment.RetryRequired() {
t.Fatalf("assessment = plan %#v, discarded %d, retry %t", assessment.Plan().Groups(), assessment.DiscardedGroupCount(), assessment.RetryRequired())
}
if !hasIssue(assessment.Issues(), 0, test.category) {
t.Fatalf("issues = %#v, want category %q", assessment.Issues(), test.category)
}
})
}
}
func TestAssessDiscardsEveryOverlappingGroupAndRetainsIndependentGroups(t *testing.T) {
preparation := proposalPreparation(t)
assessment := preparation.Assess(ProposalResponse{DuplicateGroups: []DuplicateGroup{
{CandidateIDs: []int{1, 2}, CanonicalCandidateID: 1},
{CandidateIDs: []int{2, 3}, CanonicalCandidateID: 2},
{CandidateIDs: []int{4, 5}, CanonicalCandidateID: 5},
}})
wantPlan := []planGroupSnapshot{{members: []int{5, 6}, canonical: 6}}
if got := snapshotPlan(assessment.Plan()); !reflect.DeepEqual(got, wantPlan) {
t.Fatalf("plan = %#v, want %#v", got, wantPlan)
}
if assessment.DiscardedGroupCount() != 2 || !assessment.RetryRequired() {
t.Fatalf("discarded = %d, retry = %t", assessment.DiscardedGroupCount(), assessment.RetryRequired())
}
issues := assessment.Issues()
if len(issues) != 2 || !hasIssue(issues, 0, IssueOverlappingMember) || !hasIssue(issues, 1, IssueOverlappingMember) {
t.Fatalf("issues = %#v, want both conflicting group indexes", issues)
}
invalidAndSafe := preparation.Assess(ProposalResponse{DuplicateGroups: []DuplicateGroup{
{CandidateIDs: []int{1, 99}, CanonicalCandidateID: 1},
{CandidateIDs: []int{1, 2}, CanonicalCandidateID: 2},
}})
wantPlan = []planGroupSnapshot{{members: []int{0, 2}, canonical: 2}}
if got := snapshotPlan(invalidAndSafe.Plan()); !reflect.DeepEqual(got, wantPlan) {
t.Fatalf("plan with independent invalid group = %#v, want %#v", got, wantPlan)
}
if invalidAndSafe.DiscardedGroupCount() != 1 || !invalidAndSafe.RetryRequired() || !hasIssue(invalidAndSafe.Issues(), 0, IssueMemberUnknown) {
t.Fatalf("invalid-and-safe assessment = discarded %d, retry %t, issues %#v", invalidAndSafe.DiscardedGroupCount(), invalidAndSafe.RetryRequired(), invalidAndSafe.Issues())
}
}
func TestAssessmentAccessorsAndInputsDoNotShareRetainedState(t *testing.T) {
preparation := proposalPreparation(t)
response := ProposalResponse{DuplicateGroups: []DuplicateGroup{
{CandidateIDs: []int{1, 2}, CanonicalCandidateID: 2},
}}
assessment := preparation.Assess(response)
want := snapshotPlan(assessment.Plan())
response.DuplicateGroups[0].CandidateIDs[0] = 99
response.DuplicateGroups[0].CanonicalCandidateID = 99
plan := assessment.Plan()
groups := plan.Groups()
members := groups[0].MemberPositions()
members[0] = 99
groups[0] = PlanGroup{}
if got := snapshotPlan(assessment.Plan()); !reflect.DeepEqual(got, want) {
t.Fatalf("assessment plan changed through returned or input data: %#v", got)
}
invalid := preparation.Assess(ProposalResponse{DuplicateGroups: []DuplicateGroup{{
CandidateIDs: []int{1, 99}, CanonicalCandidateID: 1,
}}})
issues := invalid.Issues()
issues[0].GroupIndex = 99
issues[0].Category = IssueOverlappingMember
if retained := invalid.Issues(); retained[0].GroupIndex == 99 || retained[0].Category == IssueOverlappingMember {
t.Fatalf("Issues() exposed retained state: %#v", retained)
}
}
type planGroupSnapshot struct {
members []int
canonical int
}
func snapshotPlan(plan Plan) []planGroupSnapshot {
groups := plan.Groups()
snapshot := make([]planGroupSnapshot, len(groups))
for index, group := range groups {
snapshot[index] = planGroupSnapshot{
members: group.MemberPositions(),
canonical: group.CanonicalPosition(),
}
}
return snapshot
}
func hasIssue(issues []Issue, groupIndex int, category IssueCategory) bool {
for _, issue := range issues {
if issue.GroupIndex == groupIndex && issue.Category == category {
return true
}
}
return false
}
func proposalPreparation(t *testing.T) Preparation {
t.Helper()
document := &source.SourceDocument{ID: "session", Units: make([]source.SourceUnit, 7)}
candidates := make([]Candidate, len(document.Units))
for position := range document.Units {
unitID := position + 1
document.Units[position] = source.SourceUnit{ID: unitID, Text: "unit"}
candidates[position] = Candidate{
Label: "candidate",
SourceRefs: []source.SourceRef{{
SourceID: document.ID,
StartUnitID: unitID,
EndUnitID: unitID,
}},
}
}
for _, position := range []int{1, 4} {
candidates[position].SourceRefs[0].SourceID = "other"
}
preparation, err := Prepare(document, candidates, Limits{
ContextRadius: 0,
MaximumCandidates: len(candidates),
MaximumMaterialBytes: 10000,
})
if err != nil || preparation.Disposition() != Ready {
t.Fatalf("Prepare() disposition = %v, error = %v", preparation.Disposition(), err)
}
return preparation
}

View File

@@ -0,0 +1,41 @@
package semanticreconcile
import (
"fmt"
"io/fs"
rootassets "gitea.maximumdirect.net/eric/notarius/assets"
"gitea.maximumdirect.net/eric/notarius/internal/framework/llm"
)
const (
ResponseSchemaKey = llm.ResponseSchemaKey("semantic_reconciliation_llm")
ResponseSchemaID = "notarius.generic.semantic_reconciliation.llm"
ResponseSchemaName = "notarius_semantic_reconciliation_llm_v1"
SchemaVersion = "v1"
SchemaAssetPath = "schemas/semantic_reconciliation_llm.v1.json"
)
func schemaAssetFS() (fs.FS, error) {
assets, err := fs.Sub(rootassets.FS(), "generic/normalize/deduplication")
if err != nil {
return nil, fmt.Errorf("scope semantic reconciliation assets: %w", err)
}
return assets, nil
}
// LoadResponseSchema returns the private request-local integer proposal
// contract. It is separate from every durable artifact schema.
func LoadResponseSchema() (llm.ResponseSchema, error) {
assets, err := schemaAssetFS()
if err != nil {
return llm.ResponseSchema{}, err
}
return llm.LoadResponseSchema(assets, llm.ResponseSchemaDefinition{
Key: ResponseSchemaKey,
ID: ResponseSchemaID,
Version: SchemaVersion,
Name: ResponseSchemaName,
AssetPath: SchemaAssetPath,
})
}

View File

@@ -0,0 +1,107 @@
package semanticreconcile
import (
"bytes"
"encoding/json"
"reflect"
"strings"
"testing"
"github.com/santhosh-tekuri/jsonschema/v6"
)
func TestResponseSchemaMetadataAndOwnership(t *testing.T) {
schema, err := LoadResponseSchema()
if err != nil {
t.Fatal(err)
}
if schema.Key != ResponseSchemaKey || schema.ID != ResponseSchemaID || schema.Version != SchemaVersion || schema.Name != ResponseSchemaName {
t.Fatalf("schema metadata = %#v", schema)
}
if !json.Valid(schema.JSONSchema) || !strings.HasPrefix(schema.SHA256, "sha256:") {
t.Fatalf("schema content metadata = %#v", schema)
}
first := append([]byte(nil), schema.JSONSchema...)
schema.JSONSchema[0] = '['
loadedAgain, err := LoadResponseSchema()
if err != nil {
t.Fatal(err)
}
if !bytes.Equal(loadedAgain.JSONSchema, first) || !json.Valid(loadedAgain.JSONSchema) {
t.Fatal("LoadResponseSchema() exposed shared schema content")
}
}
func TestResponseSchemaAcceptsOnlyTheIntegerProposalShape(t *testing.T) {
schema, err := LoadResponseSchema()
if err != nil {
t.Fatal(err)
}
tests := []struct {
name string
value any
valid bool
}{
{name: "empty proposal", value: map[string]any{"duplicate_groups": []any{}}, valid: true},
{name: "valid group", value: map[string]any{"duplicate_groups": []any{map[string]any{"candidate_ids": []any{1, 2}, "canonical_candidate_id": 1}}}, valid: true},
{name: "semantic canonical mismatch", value: map[string]any{"duplicate_groups": []any{map[string]any{"candidate_ids": []any{1, 2}, "canonical_candidate_id": 3}}}, valid: true},
{name: "missing proposal", value: map[string]any{}, valid: false},
{name: "unknown top-level field", value: map[string]any{"duplicate_groups": []any{}, "extra": true}, valid: false},
{name: "missing members", value: map[string]any{"duplicate_groups": []any{map[string]any{"canonical_candidate_id": 1}}}, valid: false},
{name: "missing canonical", value: map[string]any{"duplicate_groups": []any{map[string]any{"candidate_ids": []any{1, 2}}}}, valid: false},
{name: "unknown group field", value: map[string]any{"duplicate_groups": []any{map[string]any{"candidate_ids": []any{1, 2}, "canonical_candidate_id": 1, "name": "replacement"}}}, valid: false},
{name: "too few members", value: map[string]any{"duplicate_groups": []any{map[string]any{"candidate_ids": []any{1}, "canonical_candidate_id": 1}}}, valid: false},
{name: "repeated members", value: map[string]any{"duplicate_groups": []any{map[string]any{"candidate_ids": []any{1, 1}, "canonical_candidate_id": 1}}}, valid: false},
{name: "zero member", value: map[string]any{"duplicate_groups": []any{map[string]any{"candidate_ids": []any{0, 1}, "canonical_candidate_id": 1}}}, valid: false},
{name: "negative member", value: map[string]any{"duplicate_groups": []any{map[string]any{"candidate_ids": []any{-1, 1}, "canonical_candidate_id": 1}}}, valid: false},
{name: "non-integer member", value: map[string]any{"duplicate_groups": []any{map[string]any{"candidate_ids": []any{1, 2.5}, "canonical_candidate_id": 1}}}, valid: false},
{name: "zero canonical", value: map[string]any{"duplicate_groups": []any{map[string]any{"candidate_ids": []any{1, 2}, "canonical_candidate_id": 0}}}, valid: false},
{name: "contextual selectors", value: map[string]any{"duplicate_groups": []any{map[string]any{"candidate_ids": []any{1, 2}, "canonical_candidate_id": 1, "source_refs": []any{}}}}, valid: false},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
content, err := json.Marshal(test.value)
if err != nil {
t.Fatal(err)
}
err = validateAgainstSchema(content, schema.JSONSchema)
if (err == nil) != test.valid {
t.Fatalf("schema validation error = %v, want valid=%t", err, test.valid)
}
})
}
content := []byte(`{"duplicate_groups":[{"candidate_ids":[1,2],"canonical_candidate_id":2}]}`)
if err := validateAgainstSchema(content, schema.JSONSchema); err != nil {
t.Fatal(err)
}
var response ProposalResponse
if err := json.Unmarshal(content, &response); err != nil {
t.Fatal(err)
}
want := ProposalResponse{DuplicateGroups: []DuplicateGroup{{CandidateIDs: []int{1, 2}, CanonicalCandidateID: 2}}}
if !reflect.DeepEqual(response, want) {
t.Fatalf("decoded response = %#v, want %#v", response, want)
}
}
func validateAgainstSchema(instanceContent, schemaContent []byte) error {
instance, err := jsonschema.UnmarshalJSON(bytes.NewReader(instanceContent))
if err != nil {
return err
}
document, err := jsonschema.UnmarshalJSON(bytes.NewReader(schemaContent))
if err != nil {
return err
}
compiler := jsonschema.NewCompiler()
if err := compiler.AddResource("schema.json", document); err != nil {
return err
}
compiled, err := compiler.Compile("schema.json")
if err != nil {
return err
}
return compiled.Validate(instance)
}