Migrate item reconciliation to shared engine

internal/modules/dnd/normalize/itemregistry/normalizer.go

@@ -3,7 +3,6 @@ package itemregistry
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"reflect"
 	"sort"
@@ -13,20 +12,19 @@ import (
 	"gitea.maximumdirect.net/eric/notarius/internal/core/source"
 	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
 	"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/semanticreconcile"
 	"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd"
 	"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/items/identity"
 	"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/shared"
 	"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/shared/diagnostics"
-	"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/shared/entityreconcile"
 )
 
 const (
-	Key                   = "dnd/item-registry"
-	PromptID              = "dnd.item_registry.normalize"
-	normalizationPolicy   = "dnd.item_registry.normalize.v2"
-	semanticContextPolicy = "dnd.entity_reconcile.context.v1"
-	semanticContextRadius = 2
-	NormalizationPolicy   = normalizationPolicy
+	Key                 = "dnd/item-registry"
+	PromptID            = "dnd.item_registry.normalize"
+	PromptVersion       = "v1"
+	normalizationPolicy = "dnd.item_registry.normalize.v3"
+	NormalizationPolicy = normalizationPolicy
 
 	ReasonCodeItemFieldsNormalized                = "item_fields_normalized"
 	ReasonCodeItemIDRecomputed                    = "item_id_recomputed"
@@ -47,9 +45,7 @@ var _ pipeline.CheckpointFingerprintProvider = (*Normalizer)(nil)
 type Options struct{}
 
 type Normalizer struct {
-	llm               contracts.StructuredLLMClient
-	promptSHA         string
-	responseSchemaSHA string
+	engine *semanticreconcile.Engine
 }
 
 func New(llmClient contracts.StructuredLLMClient, _ Options) (*Normalizer, error) {
@@ -60,46 +56,45 @@ func New(llmClient contracts.StructuredLLMClient, _ Options) (*Normalizer, error
 	if err != nil {
 		return nil, normalizerErrorf("load prompt metadata: %w", err)
 	}
-	responseSchema, err := entityreconcile.LoadResponseSchema()
+	engine, err := semanticreconcile.NewEngine(llmClient, semanticreconcile.PromptSpec{
+		ID: PromptID, Version: PromptVersion, SHA256: promptSHA,
+	}, semanticreconcile.DefaultLimits())
 	if err != nil {
-		return nil, normalizerErrorf("load response schema: %w", err)
+		return nil, normalizerErrorf("construct semantic reconciliation engine: %w", err)
 	}
-	return &Normalizer{llm: llmClient, promptSHA: promptSHA, responseSchemaSHA: responseSchema.SHA256}, nil
+	return &Normalizer{engine: engine}, nil
 }
 
 func (n *Normalizer) Key() string                               { return Key }
 func (n *Normalizer) ReferenceSlots() []contracts.ReferenceSlot { return nil }
 
 func (n *Normalizer) ManifestMetadata() map[string]any {
-	if n == nil {
+	if n == nil || n.engine == nil {
 		return nil
 	}
-	return map[string]any{
-		"prompt_id": PromptID, "prompt_version": entityreconcile.SchemaVersion, "prompt_sha256": n.promptSHA,
-		"response_schema_key": string(entityreconcile.ResponseSchemaKey), "response_schema_id": entityreconcile.ResponseSchemaID,
-		"response_schema_name": entityreconcile.ResponseSchemaName, "response_schema_version": entityreconcile.SchemaVersion,
-		"response_schema_sha256": n.responseSchemaSHA, "identity_policy": identity.Policy,
-		"normalization_policy": normalizationPolicy, "semantic_context_policy": semanticContextPolicy, "semantic_context_radius": semanticContextRadius,
-	}
+	metadata := n.engine.ManifestMetadata()
+	metadata["identity_policy"] = identity.Policy
+	metadata["normalization_policy"] = normalizationPolicy
+	return metadata
 }
 
 func (n *Normalizer) CheckpointFingerprints() []pipeline.CheckpointFingerprint {
-	if n == nil {
+	if n == nil || n.engine == nil {
 		return nil
 	}
-	return []pipeline.CheckpointFingerprint{
-		{Name: "prompt", Value: n.promptSHA}, {Name: "response_schema", Value: n.responseSchemaSHA},
-		{Name: "identity_policy", Value: identity.Policy}, {Name: "normalization_policy", Value: normalizationPolicy},
-		{Name: "semantic_context_policy", Value: fmt.Sprintf("%s:%d", semanticContextPolicy, semanticContextRadius)},
-	}
+	fingerprints := n.engine.CheckpointFingerprints()
+	return append(fingerprints,
+		pipeline.CheckpointFingerprint{Name: "identity_policy", Value: identity.Policy},
+		pipeline.CheckpointFingerprint{Name: "normalization_policy", Value: normalizationPolicy},
+	)
 }
 
 func (n *Normalizer) Normalize(ctx context.Context, req contracts.TypedNormalizeRequest[dnd.ItemRegistry]) (contracts.TypedNormalizeResult[dnd.ItemRegistry], error) {
 	if n == nil {
 		return contracts.TypedNormalizeResult[dnd.ItemRegistry]{}, normalizerErrorf("normalizer must not be nil")
 	}
-	if n.llm == nil {
-		return contracts.TypedNormalizeResult[dnd.ItemRegistry]{}, normalizerErrorf("LLM client must not be nil")
+	if n.engine == nil {
+		return contracts.TypedNormalizeResult[dnd.ItemRegistry]{}, normalizerErrorf("semantic reconciliation engine must not be nil")
 	}
 	if ctx == nil {
 		return contracts.TypedNormalizeResult[dnd.ItemRegistry]{}, normalizerErrorf("context must not be nil")
@@ -111,35 +106,44 @@ func (n *Normalizer) Normalize(ctx context.Context, req contracts.TypedNormalize
 	order := shared.NewSourceRefOrder(req.Source)
 	records, warnings := preprocessRecords(req.MergeOutput.Value, order)
 	deterministic := recordList(records)
-	materials, ready, err := entityreconcile.BuildContext(req.Source, reconciliationCandidates(records), semanticContextRadius)
-	if err != nil {
-		return contracts.TypedNormalizeResult[dnd.ItemRegistry]{}, normalizerErrorf("build semantic context: %w", err)
-	}
-	if !ready {
+	if len(records) < 2 || req.Source == nil {
 		return contracts.TypedNormalizeResult[dnd.ItemRegistry]{Value: deterministic, Warnings: limitWarnings(warnings)}, nil
 	}
 
-	var response entityreconcile.ProposalResponse
-	if _, err := n.llm.CompleteStructured(ctx, contracts.StructuredCompletionRequest{
-		StageName: Key, PromptID: PromptID, PromptVersion: entityreconcile.SchemaVersion,
+	candidates, envelopes, err := reconciliationInputs(records)
+	if err != nil {
+		return contracts.TypedNormalizeResult[dnd.ItemRegistry]{}, normalizerErrorf("prepare semantic reconciliation inputs: %w", err)
+	}
+	reconciliation, err := n.engine.Reconcile(ctx, semanticreconcile.Request{
+		StageName: Key, Source: req.Source, Candidates: candidates,
 		ProfileID: req.LLMProfile, SessionID: req.SessionID,
-		Inputs: contracts.LLMInputSet{"candidates": materials.Candidates, "transcript": materials.Transcript},
-	}, &response); err != nil {
-		if errors.Is(err, contracts.ErrInvalidStructuredOutput) {
-			return n.invalidStructuredResult(deterministic, warnings), nil
-		}
-		return contracts.TypedNormalizeResult[dnd.ItemRegistry]{}, normalizerErrorf("complete structured output: %w", err)
+	})
+	if err != nil {
+		return contracts.TypedNormalizeResult[dnd.ItemRegistry]{}, normalizerErrorf("reconcile semantic duplicates: %w", err)
+	}
+
+	switch reconciliation.Disposition() {
+	case semanticreconcile.SkippedInsufficientCandidates:
+		return contracts.TypedNormalizeResult[dnd.ItemRegistry]{Value: deterministic, Warnings: limitWarnings(warnings)}, nil
+	case semanticreconcile.SkippedLimitExceeded:
+		return contracts.TypedNormalizeResult[dnd.ItemRegistry]{Value: deterministic, Warnings: limitWarningsWithSemanticFallback(warnings)}, nil
+	case semanticreconcile.RetryableInvalidStructuredOutput:
+		return n.invalidStructuredResult(deterministic, warnings), nil
+	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)
+	if err != nil {
+		return contracts.TypedNormalizeResult[dnd.ItemRegistry]{}, normalizerErrorf("apply semantic reconciliation plan: %w", err)
 	}
-	assessment := materials.Assess(response)
-	applied, semanticWarnings, rejectedGroups := applySafeGroups(records, reconciliationGroups(assessment, materials.CandidateKeys()), order)
 	warnings = append(warnings, semanticWarnings...)
-	if rejectedGroups > 0 {
-		return currencyRetryResult(recordList(applied), warnings, rejectedGroups), nil
-	}
-	if assessment.DiscardedGroups() == 0 {
+	discardedGroups := reconciliation.DiscardedGroupCount() + rejectedGroups
+	if discardedGroups == 0 {
 		return contracts.TypedNormalizeResult[dnd.ItemRegistry]{Value: recordList(applied), Warnings: limitWarnings(warnings)}, nil
 	}
-	return retryResult(recordList(applied), warnings, assessment), nil
+	return retryResult(recordList(applied), warnings, reconciliation, rejectedGroups), nil
 }
 
 func (n *Normalizer) invalidStructuredResult(value dnd.ItemRegistry, warnings []contracts.Warning) contracts.TypedNormalizeResult[dnd.ItemRegistry] {
@@ -149,17 +153,15 @@ func (n *Normalizer) invalidStructuredResult(value dnd.ItemRegistry, warnings []
 	}}
 }
 
-func retryResult(value dnd.ItemRegistry, warnings []contracts.Warning, assessment entityreconcile.Assessment) contracts.TypedNormalizeResult[dnd.ItemRegistry] {
+func retryResult(value dnd.ItemRegistry, warnings []contracts.Warning, reconciliation semanticreconcile.Result, rejectedGroups int) contracts.TypedNormalizeResult[dnd.ItemRegistry] {
+	details := reconciliationIssues(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), Retry: &contracts.NormalizeRetry{
-		ReasonCode: ReasonCodeItemSemanticProposalInvalid, Message: diagnostics.Aggregate("semantic proposal requires retry", reconciliationIssues(assessment)),
-		FallbackWarnings: []contracts.Warning{semanticFallbackWarning(assessment.DiscardedGroups())},
-	}}
-}
-
-func currencyRetryResult(value dnd.ItemRegistry, warnings []contracts.Warning, rejectedGroups int) contracts.TypedNormalizeResult[dnd.ItemRegistry] {
-	return contracts.TypedNormalizeResult[dnd.ItemRegistry]{Value: value, Warnings: limitWarningsForRetry(warnings), Retry: &contracts.NormalizeRetry{
-		ReasonCode: ReasonCodeItemSemanticProposalInvalid, Message: "semantic proposal requires retry: currency may only be consolidated with aliases of one denomination",
-		FallbackWarnings: []contracts.Warning{semanticFallbackWarning(rejectedGroups)},
+		ReasonCode: ReasonCodeItemSemanticProposalInvalid, Message: diagnostics.Aggregate("semantic proposal requires retry", details),
+		FallbackWarnings: []contracts.Warning{semanticFallbackWarning(discardedGroups)},
 	}}
 }
 
@@ -187,6 +189,10 @@ func limitWarningsForRetry(warnings []contracts.Warning) []contracts.Warning {
 	return append(bounded, contracts.Warning{Scope: "items", ReasonCode: ReasonCodeItemNormalizationWarningsOmitted, Message: fmt.Sprintf("%d additional warning(s) omitted", len(warnings)-displayed)})
 }
 
+func limitWarningsWithSemanticFallback(warnings []contracts.Warning) []contracts.Warning {
+	return append(limitWarningsForRetry(warnings), semanticFallbackWarning(-1))
+}
+
 type normalizedRecord struct {
 	item         dnd.Item
 	inputIndexes []int