eric/notarius
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Classify registry normalization diagnostics

Browse Source
This commit is contained in:
Eric Rakestraw
2026-08-27 15:57:29 +00:00
parent ccba2ce3f9
commit 6a1fd7bdb6
12 changed files with 247 additions and 206 deletions
Download Patch File Download Diff File Expand all files Collapse all files

91
internal/modules/dnd/normalize/itemregistry/normalizer.go
View File

@@ -31,8 +31,8 @@ const (
ReasonCodeSourceReferencesNormalized = "source_references_normalized"
ReasonCodeDuplicateItemCollapsed = "duplicate_item_collapsed"
ReasonCodeItemSemanticProposalInvalid = "item_semantic_proposal_invalid"
ReasonCodeItemSemanticRetryProposalInvalid = "item_semantic_retry_proposal_invalid"
ReasonCodeItemSemanticReconciliationExhausted = "item_semantic_reconciliation_exhausted"
ReasonCodeItemNormalizationWarningsOmitted = "item_normalization_warnings_omitted"
)
var requiredCapabilities = []string{"merged"}
@@ -107,10 +107,10 @@ func (n *Normalizer) Normalize(ctx context.Context, req contracts.TypedNormalize
}
order := shared.NewSourceRefOrder(req.Source)
records, warnings := preprocessRecords(req.MergeOutput.Value, order)
records, findings := preprocessRecords(req.MergeOutput.Value, order)
deterministic := recordList(records)
if len(records) < 2 {
return contracts.TypedNormalizeResult[dnd.ItemRegistry]{Value: deterministic, Warnings: limitWarnings(warnings)}, nil
return normalizationResult(deterministic, findings, nil, nil)
}
candidates, envelopes, err := reconciliationInputs(records)
@@ -127,48 +127,42 @@ func (n *Normalizer) Normalize(ctx context.Context, req contracts.TypedNormalize
switch reconciliation.Disposition() {
case semanticreconcile.SkippedInsufficientCandidates:
return contracts.TypedNormalizeResult[dnd.ItemRegistry]{Value: deterministic, Warnings: limitWarnings(warnings)}, nil
return normalizationResult(deterministic, findings, nil, nil)
case semanticreconcile.SkippedLimitExceeded:
return contracts.TypedNormalizeResult[dnd.ItemRegistry]{Value: deterministic, Warnings: limitWarningsWithSemanticFallback(warnings)}, nil
return fallbackResult(deterministic, findings, nil, semanticFallbackFinding(-1))
case semanticreconcile.RetryableInvalidStructuredOutput:
return n.invalidStructuredResult(deterministic, warnings), nil
return n.invalidStructuredResult(deterministic, findings)
case semanticreconcile.Complete, semanticreconcile.RetryableDiscardedProposalGroups:
default:
return contracts.TypedNormalizeResult[dnd.ItemRegistry]{}, normalizerErrorf("unknown semantic reconciliation disposition %d", reconciliation.Disposition())
}
applied, semanticWarnings, rejectedGroups, err := applyReconciliationPlan(reconciliation.Plan(), records, envelopes, order)
applied, semanticFindings, advisoryFindings, rejectedGroups, err := applyReconciliationPlan(reconciliation.Plan(), records, envelopes, order)
if err != nil {
return contracts.TypedNormalizeResult[dnd.ItemRegistry]{}, normalizerErrorf("apply semantic reconciliation plan: %w", err)
}
warnings = append(warnings, semanticWarnings...)
findings = append(findings, semanticFindings...)
discardedGroups := reconciliation.DiscardedGroupCount() + rejectedGroups
if discardedGroups == 0 {
return contracts.TypedNormalizeResult[dnd.ItemRegistry]{Value: recordList(applied), Warnings: limitWarnings(warnings), ModelCandidate: reconciliation.ModelCandidate()}, nil
return normalizationResult(recordList(applied), findings, advisoryFindings, reconciliation.ModelCandidate())
}
return retryResult(recordList(applied), warnings, reconciliation, rejectedGroups), nil
return retryResult(recordList(applied), findings, advisoryFindings, reconciliation, rejectedGroups)
}
func (n *Normalizer) invalidStructuredResult(value dnd.ItemRegistry, warnings []contracts.Warning) contracts.TypedNormalizeResult[dnd.ItemRegistry] {
return contracts.TypedNormalizeResult[dnd.ItemRegistry]{Value: value, Warnings: limitWarningsForRetry(warnings), Retry: &contracts.NormalizeRetry{
ReasonCode: ReasonCodeItemSemanticProposalInvalid, Message: "semantic proposal requires retry: invalid structured output",
FallbackWarnings: []contracts.Warning{semanticFallbackWarning(-1)},
}}
func (n *Normalizer) invalidStructuredResult(value dnd.ItemRegistry, findings []contracts.Warning) (contracts.TypedNormalizeResult[dnd.ItemRegistry], error) {
return retryResultWithFallback(value, findings, nil, nil, ReasonCodeItemSemanticRetryProposalInvalid, "semantic proposal requires retry: invalid structured output", semanticFallbackFinding(-1))
}
func retryResult(value dnd.ItemRegistry, warnings []contracts.Warning, reconciliation semanticreconcile.Result, rejectedGroups int) contracts.TypedNormalizeResult[dnd.ItemRegistry] {
func retryResult(value dnd.ItemRegistry, findings, advisoryFindings []contracts.Warning, reconciliation semanticreconcile.Result, rejectedGroups int) (contracts.TypedNormalizeResult[dnd.ItemRegistry], error) {
details := semanticreconcile.IssueDetails(reconciliation.Issues())
if rejectedGroups > 0 {
details = append(details, "currency may only be consolidated with aliases of one denomination")
}
discardedGroups := reconciliation.DiscardedGroupCount() + rejectedGroups
return contracts.TypedNormalizeResult[dnd.ItemRegistry]{Value: value, Warnings: limitWarningsForRetry(warnings), ModelCandidate: reconciliation.ModelCandidate(), Retry: &contracts.NormalizeRetry{
ReasonCode: ReasonCodeItemSemanticProposalInvalid, Message: diagnostics.Aggregate("semantic proposal requires retry", details),
FallbackWarnings: []contracts.Warning{semanticFallbackWarning(discardedGroups)},
}}
return retryResultWithFallback(value, findings, advisoryFindings, reconciliation.ModelCandidate(), ReasonCodeItemSemanticRetryProposalInvalid, diagnostics.Aggregate("semantic proposal requires retry", details), semanticFallbackFinding(discardedGroups))
}
func semanticFallbackWarning(discarded int) contracts.Warning {
func semanticFallbackFinding(discarded int) contracts.Warning {
message := "semantic proposal could not be applied"
if discarded >= 0 {
message = fmt.Sprintf("%d proposal group(s) omitted after semantic proposal retry exhaustion", discarded)
@@ -176,24 +170,43 @@ func semanticFallbackWarning(discarded int) contracts.Warning {
return contracts.Warning{Scope: "items", ReasonCode: ReasonCodeItemSemanticReconciliationExhausted, Message: message}
}
func limitWarnings(warnings []contracts.Warning) []contracts.Warning {
return diagnostics.LimitWarnings(warnings, "items", ReasonCodeItemNormalizationWarningsOmitted)
func normalizationResult(value dnd.ItemRegistry, findings, advisoryFindings []contracts.Warning, candidate *contracts.ModelCandidate) (contracts.TypedNormalizeResult[dnd.ItemRegistry], error) {
diagnosticGroups, err := diagnostics.Collect(findings, contracts.DiagnosticDispositionObservation, contracts.DiagnosticCategoryNormalization)
if err != nil {
return contracts.TypedNormalizeResult[dnd.ItemRegistry]{}, normalizerErrorf("collect normalization diagnostics: %w", err)
}
advisoryGroups, err := diagnostics.Collect(advisoryFindings, contracts.DiagnosticDispositionAdvisory, contracts.DiagnosticCategoryDataQuality)
if err != nil {
return contracts.TypedNormalizeResult[dnd.ItemRegistry]{}, normalizerErrorf("collect data-quality diagnostics: %w", err)
}
diagnosticGroups = append(diagnosticGroups, advisoryGroups...)
return contracts.TypedNormalizeResult[dnd.ItemRegistry]{Value: value, Diagnostics: diagnosticGroups, ModelCandidate: candidate}, nil
}
func limitWarningsForRetry(warnings []contracts.Warning) []contracts.Warning {
if warnings == nil {
return nil
func fallbackResult(value dnd.ItemRegistry, findings, advisoryFindings []contracts.Warning, fallback contracts.Warning) (contracts.TypedNormalizeResult[dnd.ItemRegistry], error) {
result, err := normalizationResult(value, findings, advisoryFindings, nil)
if err != nil {
return contracts.TypedNormalizeResult[dnd.ItemRegistry]{}, err
}
if len(warnings) < diagnostics.MaxWarnings {
return append([]contracts.Warning(nil), warnings...)
fallbackGroups, err := diagnostics.Collect([]contracts.Warning{fallback}, contracts.DiagnosticDispositionWarning, contracts.DiagnosticCategoryFallback)
if err != nil {
return contracts.TypedNormalizeResult[dnd.ItemRegistry]{}, normalizerErrorf("collect fallback diagnostic: %w", err)
}
displayed := diagnostics.MaxWarnings - 2
bounded := append([]contracts.Warning(nil), warnings[:displayed]...)
return append(bounded, contracts.Warning{Scope: "items", ReasonCode: ReasonCodeItemNormalizationWarningsOmitted, Message: fmt.Sprintf("%d additional warning(s) omitted", len(warnings)-displayed)})
result.Diagnostics = append(result.Diagnostics, fallbackGroups...)
return result, nil
}
func limitWarningsWithSemanticFallback(warnings []contracts.Warning) []contracts.Warning {
return append(limitWarningsForRetry(warnings), semanticFallbackWarning(-1))
func retryResultWithFallback(value dnd.ItemRegistry, findings, advisoryFindings []contracts.Warning, candidate *contracts.ModelCandidate, reasonCode, message string, fallback contracts.Warning) (contracts.TypedNormalizeResult[dnd.ItemRegistry], error) {
result, err := normalizationResult(value, findings, advisoryFindings, candidate)
if err != nil {
return contracts.TypedNormalizeResult[dnd.ItemRegistry]{}, err
}
fallbackGroups, err := diagnostics.Collect([]contracts.Warning{fallback}, contracts.DiagnosticDispositionWarning, contracts.DiagnosticCategoryFallback)
if err != nil {
return contracts.TypedNormalizeResult[dnd.ItemRegistry]{}, normalizerErrorf("collect fallback diagnostic: %w", err)
}
result.Retry = &contracts.NormalizeRetry{ReasonCode: reasonCode, Message: message, FallbackDiagnostics: fallbackGroups}
return result, nil
}
type normalizedRecord struct {
@@ -207,18 +220,18 @@ func preprocessRecords(input dnd.ItemRegistry, order shared.SourceRefOrder) ([]n
return nil, nil
}
records := make([]normalizedRecord, len(input.Items))
warnings := make([]contracts.Warning, 0)
findings := make([]contracts.Warning, 0)
for index, inputItem := range input.Items {
item, fieldsChanged, refsChanged := normalizeRecord(inputItem, order)
records[index] = normalizedRecord{item: item, inputIndexes: []int{index}, earliest: index}
if fieldsChanged {
warnings = append(warnings, contracts.Warning{Scope: itemScope(index), ReasonCode: ReasonCodeItemFieldsNormalized, Message: fmt.Sprintf("input index %d: item name normalized for %s", index, diagnostics.Quote(inputItem.Name))})
findings = append(findings, contracts.Warning{Scope: itemScope(index), ReasonCode: ReasonCodeItemFieldsNormalized, Message: fmt.Sprintf("input index %d: item name normalized for %s", index, diagnostics.Quote(inputItem.Name))})
}
if refsChanged {
warnings = append(warnings, contracts.Warning{Scope: itemScope(index), ReasonCode: ReasonCodeSourceReferencesNormalized, Message: fmt.Sprintf("input index %d: source references normalized (original count %d, final count %d)", index, len(inputItem.SourceRefs), len(item.SourceRefs))})
findings = append(findings, contracts.Warning{Scope: itemScope(index), ReasonCode: ReasonCodeSourceReferencesNormalized, Message: fmt.Sprintf("input index %d: source references normalized (original count %d, final count %d)", index, len(inputItem.SourceRefs), len(item.SourceRefs))})
}
if inputItem.ID != item.ID {
warnings = append(warnings, contracts.Warning{Scope: itemScope(index), ReasonCode: ReasonCodeItemIDRecomputed, Message: fmt.Sprintf("input index %d: item ID recomputed from %s", index, diagnostics.Quote(item.Name))})
findings = append(findings, contracts.Warning{Scope: itemScope(index), ReasonCode: ReasonCodeItemIDRecomputed, Message: fmt.Sprintf("input index %d: item ID recomputed from %s", index, diagnostics.Quote(item.Name))})
}
}
groups := comparisonNameGroups(records)
@@ -234,10 +247,10 @@ func preprocessRecords(input dnd.ItemRegistry, order shared.SourceRefOrder) ([]n
retained.inputIndexes = sortedUniqueIndexes(retained.inputIndexes)
output = append(output, retained)
if len(members) > 1 {
warnings = append(warnings, duplicateWarning(retained.earliest, memberInputIndexes(records, members[1:])))
findings = append(findings, duplicateWarning(retained.earliest, memberInputIndexes(records, members[1:])))
}
}
return output, warnings
return output, findings
}
func normalizeRecord(input dnd.Item, order shared.SourceRefOrder) (dnd.Item, bool, bool) {