Register item registry normalization

2026-08-05 19:39:06 +00:00
parent ad9ee076b5
commit f91237e9c0
19 changed files with 1165 additions and 15 deletions
--- a/internal/modules/dnd/normalize/itemregistry/normalizer.go
+++ b/internal/modules/dnd/normalize/itemregistry/normalizer.go
@@ -0,0 +1,350 @@
+// Package itemregistry normalizes merged D&D item-registry candidates conservatively.
+package itemregistry
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"reflect"
+	"sort"
+	"strconv"
+	"strings"
+
+	"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/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.v1"
+	semanticContextPolicy = "dnd.entity_reconcile.context.v1"
+	semanticContextRadius = 2
+	NormalizationPolicy   = normalizationPolicy
+
+	ReasonCodeItemFieldsNormalized                = "item_fields_normalized"
+	ReasonCodeItemIDRecomputed                    = "item_id_recomputed"
+	ReasonCodeSourceReferencesNormalized          = "source_references_normalized"
+	ReasonCodeDuplicateItemCollapsed              = "duplicate_item_collapsed"
+	ReasonCodeItemSemanticProposalInvalid         = "item_semantic_proposal_invalid"
+	ReasonCodeItemSemanticReconciliationExhausted = "item_semantic_reconciliation_exhausted"
+	ReasonCodeItemNormalizationWarningsOmitted    = "item_normalization_warnings_omitted"
+)
+
+var requiredCapabilities = []string{"merged"}
+var providedCapabilities = []string{"normalized"}
+
+var _ contracts.Normalizer[dnd.ItemRegistry] = (*Normalizer)(nil)
+var _ contracts.ManifestMetadataProvider = (*Normalizer)(nil)
+var _ pipeline.CheckpointFingerprintProvider = (*Normalizer)(nil)
+
+type Options struct{}
+
+type Normalizer struct {
+	llm               contracts.StructuredLLMClient
+	promptSHA         string
+	responseSchemaSHA string
+}
+
+func New(llmClient contracts.StructuredLLMClient, _ Options) (*Normalizer, error) {
+	if llmClient == nil {
+		return nil, normalizerErrorf("LLM client must not be nil")
+	}
+	promptSHA, err := promptAssetMetadata()
+	if err != nil {
+		return nil, normalizerErrorf("load prompt metadata: %w", err)
+	}
+	responseSchema, err := entityreconcile.LoadResponseSchema()
+	if err != nil {
+		return nil, normalizerErrorf("load response schema: %w", err)
+	}
+	return &Normalizer{llm: llmClient, promptSHA: promptSHA, responseSchemaSHA: responseSchema.SHA256}, 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 {
+		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,
+	}
+}
+
+func (n *Normalizer) CheckpointFingerprints() []pipeline.CheckpointFingerprint {
+	if n == 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)},
+	}
+}
+
+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 ctx == nil {
+		return contracts.TypedNormalizeResult[dnd.ItemRegistry]{}, normalizerErrorf("context must not be nil")
+	}
+	if err := ctx.Err(); err != nil {
+		return contracts.TypedNormalizeResult[dnd.ItemRegistry]{}, normalizerErrorf("context error before normalize: %w", err)
+	}
+
+	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 {
+		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,
+		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)
+	}
+	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 {
+		return contracts.TypedNormalizeResult[dnd.ItemRegistry]{Value: recordList(applied), Warnings: limitWarnings(warnings)}, nil
+	}
+	return retryResult(recordList(applied), warnings, assessment), nil
+}
+
+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 retryResult(value dnd.ItemRegistry, warnings []contracts.Warning, assessment entityreconcile.Assessment) contracts.TypedNormalizeResult[dnd.ItemRegistry] {
+	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 denominations must remain distinct",
+		FallbackWarnings: []contracts.Warning{semanticFallbackWarning(rejectedGroups)},
+	}}
+}
+
+func semanticFallbackWarning(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)
+	}
+	return contracts.Warning{Scope: "items", ReasonCode: ReasonCodeItemSemanticReconciliationExhausted, Message: message}
+}
+
+func limitWarnings(warnings []contracts.Warning) []contracts.Warning {
+	return diagnostics.LimitWarnings(warnings, "items", ReasonCodeItemNormalizationWarningsOmitted)
+}
+
+func limitWarningsForRetry(warnings []contracts.Warning) []contracts.Warning {
+	if warnings == nil {
+		return nil
+	}
+	if len(warnings) < diagnostics.MaxWarnings {
+		return append([]contracts.Warning(nil), warnings...)
+	}
+	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)})
+}
+
+type normalizedRecord struct {
+	item         dnd.Item
+	inputIndexes []int
+	earliest     int
+}
+
+func preprocessRecords(input dnd.ItemRegistry, order shared.SourceRefOrder) ([]normalizedRecord, []contracts.Warning) {
+	if input.Items == nil {
+		return nil, nil
+	}
+	records := make([]normalizedRecord, len(input.Items))
+	warnings := 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))})
+		}
+		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))})
+		}
+		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))})
+		}
+	}
+	groups := exactDuplicateGroups(records)
+	output := make([]normalizedRecord, 0, len(groups))
+	for _, members := range groups {
+		retained := cloneRecord(records[members[0]])
+		for _, member := range members[1:] {
+			retained.inputIndexes = append(retained.inputIndexes, records[member].inputIndexes...)
+		}
+		retained.inputIndexes = sortedUniqueIndexes(retained.inputIndexes)
+		output = append(output, retained)
+		if len(members) > 1 {
+			warnings = append(warnings, duplicateWarning(retained.earliest, memberInputIndexes(records, members[1:])))
+		}
+	}
+	return output, warnings
+}
+
+func normalizeRecord(input dnd.Item, order shared.SourceRefOrder) (dnd.Item, bool, bool) {
+	output := cloneItem(input)
+	output.Name = identity.NormalizeDisplay(input.Name)
+	output.SourceRefs = order.Canonicalize(input.SourceRefs)
+	output.ID = identity.DeriveID(output.Name)
+	return output, input.Name != output.Name, !reflect.DeepEqual(input.SourceRefs, output.SourceRefs)
+}
+
+func exactDuplicateGroups(records []normalizedRecord) [][]int {
+	groups := make([][]int, 0, len(records))
+	for index, record := range records {
+		key := identity.ComparisonKey(record.item.Name)
+		found := false
+		for groupIndex, members := range groups {
+			first := records[members[0]]
+			if identity.ComparisonKey(first.item.Name) == key && reflect.DeepEqual(first.item.SourceRefs, record.item.SourceRefs) {
+				groups[groupIndex] = append(groups[groupIndex], index)
+				found = true
+				break
+			}
+		}
+		if !found {
+			groups = append(groups, []int{index})
+		}
+	}
+	return groups
+}
+
+func cloneItem(input dnd.Item) dnd.Item {
+	input.SourceRefs = cloneSourceRefs(input.SourceRefs)
+	return input
+}
+func cloneRecord(input normalizedRecord) normalizedRecord {
+	input.item = cloneItem(input.item)
+	input.inputIndexes = append([]int(nil), input.inputIndexes...)
+	return input
+}
+func cloneSourceRefs(input []source.SourceRef) []source.SourceRef {
+	return append([]source.SourceRef(nil), input...)
+}
+
+func sortedUniqueIndexes(indexes []int) []int {
+	if len(indexes) == 0 {
+		return nil
+	}
+	out := append([]int(nil), indexes...)
+	sort.Ints(out)
+	write := 1
+	for _, index := range out[1:] {
+		if index != out[write-1] {
+			out[write] = index
+			write++
+		}
+	}
+	return out[:write]
+}
+func memberInputIndexes(records []normalizedRecord, members []int) []int {
+	indexes := make([]int, 0, len(members))
+	for _, member := range members {
+		indexes = append(indexes, records[member].inputIndexes...)
+	}
+	return sortedUniqueIndexes(indexes)
+}
+func recordValues(records []normalizedRecord) []dnd.Item {
+	if records == nil {
+		return nil
+	}
+	values := make([]dnd.Item, len(records))
+	for index, record := range records {
+		values[index] = cloneItem(record.item)
+	}
+	return values
+}
+func recordList(records []normalizedRecord) dnd.ItemRegistry {
+	if records == nil {
+		return dnd.ItemRegistry{}
+	}
+	return dnd.ItemRegistry{Items: recordValues(records)}
+}
+
+func duplicateWarning(retainedIndex int, removed []int) contracts.Warning {
+	const maxDisplayedIndices = 20
+	displayed := removed
+	if len(displayed) > maxDisplayedIndices {
+		displayed = displayed[:maxDisplayedIndices]
+	}
+	indices := make([]string, len(displayed))
+	for index, removedIndex := range displayed {
+		indices[index] = strconv.Itoa(removedIndex)
+	}
+	message := fmt.Sprintf("retained input index %d; removed input indices [%s]", retainedIndex, strings.Join(indices, ", "))
+	if omitted := len(removed) - len(displayed); omitted > 0 {
+		message += fmt.Sprintf("; %d additional removed input indices omitted", omitted)
+	}
+	return contracts.Warning{Scope: itemScope(retainedIndex), ReasonCode: ReasonCodeDuplicateItemCollapsed, Message: message}
+}
+func itemScope(index int) string { return fmt.Sprintf("items[%d]", index) }
+
+func ModuleSpec() pipeline.ModuleSpec {
+	return pipeline.ModuleSpec{Key: Key, Stage: pipeline.StageNormalize, ExecutionClass: contracts.ExecutionClassLLMBacked, Requires: append([]string(nil), requiredCapabilities...), Provides: append([]string(nil), providedCapabilities...), ArtifactKind: dnd.ItemRegistryKind}
+}
+func Register(registry *pipeline.NormalizerRegistry) error {
+	return pipeline.RegisterNormalizerBuilder(registry, ModuleSpec(), validateOptions, func(request pipeline.BuildRequest) (contracts.Normalizer[dnd.ItemRegistry], error) {
+		options, err := DecodeOptions(request.Options)
+		if err != nil {
+			return nil, err
+		}
+		return New(request.Dependencies.LLM, options)
+	})
+}
+func validateOptions(options map[string]any) error { _, err := DecodeOptions(options); return err }
+func DecodeOptions(options map[string]any) (Options, error) {
+	if err := pipeline.RejectUnknownOptions(options); err != nil {
+		return Options{}, normalizerErrorf("%w", err)
+	}
+	return Options{}, nil
+}
+func normalizerErrorf(format string, args ...any) error {
+	return fmt.Errorf("dnd item registry normalizer: "+format, args...)
+}
--- a/internal/modules/dnd/normalize/itemregistry/normalizer_test.go
+++ b/internal/modules/dnd/normalize/itemregistry/normalizer_test.go
@@ -0,0 +1,193 @@
+package itemregistry
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"reflect"
+	"strings"
+	"testing"
+	"time"
+
+	"gitea.maximumdirect.net/eric/notarius/internal/core/source"
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/llm"
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
+	"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/diagnostics"
+	"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/shared/entityreconcile"
+	"gitea.maximumdirect.net/eric/promptkit"
+)
+
+func TestModuleContractAndMetadata(t *testing.T) {
+	want := pipeline.ModuleSpec{Key: Key, Stage: pipeline.StageNormalize, ExecutionClass: contracts.ExecutionClassLLMBacked, Requires: []string{"merged"}, Provides: []string{"normalized"}, ArtifactKind: dnd.ItemRegistryKind}
+	if got := ModuleSpec(); !reflect.DeepEqual(got, want) {
+		t.Fatalf("ModuleSpec() = %#v, want %#v", got, want)
+	}
+	registry := pipeline.NewNormalizerRegistry()
+	if err := Register(registry); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := New(nil, Options{}); err == nil {
+		t.Fatal("New() accepted nil client")
+	}
+	metadata := newNormalizer(t, &recordingNormalizerClient{}).ManifestMetadata()
+	if metadata["identity_policy"] != identity.Policy || metadata["response_schema_id"] != entityreconcile.ResponseSchemaID || metadata["normalization_policy"] != normalizationPolicy || metadata["semantic_context_radius"] != semanticContextRadius {
+		t.Fatalf("metadata = %#v", metadata)
+	}
+}
+
+func TestNormalizePreprocessesExactDuplicatesWithoutMutation(t *testing.T) {
+	input := dnd.ItemRegistry{Items: []dnd.Item{
+		{Name: "  Rope  ", SourceRefs: []source.SourceRef{{SourceID: "session", StartUnitID: 1, EndUnitID: 1}}},
+		{Name: "rope", SourceRefs: []source.SourceRef{{SourceID: "session", StartUnitID: 1, EndUnitID: 1}}},
+		{Name: "Rope", SourceRefs: []source.SourceRef{{SourceID: "session", StartUnitID: 2, EndUnitID: 2}}},
+	}}
+	before := dnd.ItemRegistry{Items: append([]dnd.Item(nil), input.Items...)}
+	result, err := newNormalizer(t, &recordingNormalizerClient{}).Normalize(context.Background(), normalizeRequest(input))
+	if err != nil || len(result.Value.Items) != 2 || !reflect.DeepEqual(input, before) {
+		t.Fatalf("Normalize() = %#v, %v; want non-mutating exact deduplication", result, err)
+	}
+	if result.Value.Items[0].ID != result.Value.Items[1].ID || !hasWarning(result.Warnings, ReasonCodeDuplicateItemCollapsed) {
+		t.Fatalf("items = %#v, warnings = %#v; want type identity and duplicate warning", result.Value.Items, result.Warnings)
+	}
+}
+
+func TestNormalizeAppliesSafeAliasProposalAndPreservesCurrency(t *testing.T) {
+	doc := semanticDocument()
+	input := dnd.ItemRegistry{Items: []dnd.Item{
+		{Name: "Star Compass", SourceRefs: []source.SourceRef{{SourceID: doc.ID, StartUnitID: 10, EndUnitID: 10}}},
+		{Name: "Compass of the Stars", SourceRefs: []source.SourceRef{{SourceID: doc.ID, StartUnitID: 20, EndUnitID: 20}}},
+		{Name: "Gold Pieces", SourceRefs: []source.SourceRef{{SourceID: doc.ID, StartUnitID: 30, EndUnitID: 30}}},
+	}}
+	client := &recordingNormalizerClient{response: `{"duplicate_groups":[{"members":["candidate-000001","candidate-000002"],"canonical":"candidate-000002"}]}`}
+	result, err := newNormalizer(t, client).Normalize(context.Background(), normalizeRequestWithSource(input, doc))
+	if err != nil || result.Retry != nil || len(result.Value.Items) != 2 {
+		t.Fatalf("Normalize() = %#v, %v", result, err)
+	}
+	merged := result.Value.Items[0]
+	if merged.Name != "Compass of the Stars" || merged.ID != identity.DeriveID(merged.Name) || len(merged.SourceRefs) != 2 || !hasWarning(result.Warnings, ReasonCodeDuplicateItemCollapsed) {
+		t.Fatalf("merged item = %#v, warnings = %#v", merged, result.Warnings)
+	}
+	encoded := string(client.requests[0].Inputs["candidates"].Content) + string(client.requests[0].Inputs["transcript"].Content)
+	if strings.Contains(encoded, doc.ID) || !strings.Contains(encoded, "candidate-000001") || strings.Contains(encoded, merged.ID) {
+		t.Fatalf("private inputs = %s", encoded)
+	}
+	currencyClient := &recordingNormalizerClient{response: `{"duplicate_groups":[{"members":["candidate-000001","candidate-000002"],"canonical":"candidate-000001"}]}`}
+	currency := dnd.ItemRegistry{Items: []dnd.Item{{Name: "Gold Pieces", SourceRefs: []source.SourceRef{{SourceID: doc.ID, StartUnitID: 10, EndUnitID: 10}}}, {Name: "Silver Pieces", SourceRefs: []source.SourceRef{{SourceID: doc.ID, StartUnitID: 20, EndUnitID: 20}}}}}
+	result, err = newNormalizer(t, currencyClient).Normalize(context.Background(), normalizeRequestWithSource(currency, doc))
+	if err != nil || result.Retry == nil || len(result.Value.Items) != 2 || !hasWarning(result.Warnings, ReasonCodeItemSemanticProposalInvalid) {
+		t.Fatalf("currency result = %#v, %v; want denominations retained", result, err)
+	}
+}
+
+func TestNormalizePreservesCandidatesForUnsafeProposalGroups(t *testing.T) {
+	doc := semanticDocument()
+	input := dnd.ItemRegistry{Items: []dnd.Item{
+		{Name: "Star Compass", SourceRefs: []source.SourceRef{{SourceID: doc.ID, StartUnitID: 10, EndUnitID: 10}}},
+		{Name: "Compass", SourceRefs: []source.SourceRef{{SourceID: doc.ID, StartUnitID: 20, EndUnitID: 20}}},
+		{Name: "Rope", SourceRefs: []source.SourceRef{{SourceID: doc.ID, StartUnitID: 30, EndUnitID: 30}}},
+	}}
+	client := &recordingNormalizerClient{response: `{"duplicate_groups":[{"members":["candidate-000001","candidate-000002"],"canonical":"candidate-000001"},{"members":["candidate-000002","candidate-000003"],"canonical":"candidate-000003"},{"members":["candidate-000001","candidate-000003"],"canonical":"candidate-000099"}]}`}
+	result, err := newNormalizer(t, client).Normalize(context.Background(), normalizeRequestWithSource(input, doc))
+	if err != nil || result.Retry == nil || len(result.Value.Items) != 3 || !strings.Contains(result.Retry.Message, "overlapping_member") || !strings.Contains(result.Retry.Message, "canonical_unknown") || strings.Contains(result.Retry.Message, "Star Compass") || len(result.Retry.Message) > 4096 {
+		t.Fatalf("Normalize() = %#v, %v; want deterministic retry fallback", result, err)
+	}
+}
+
+func TestNormalizeRetryFallbackErrorsWarningsAndIdempotence(t *testing.T) {
+	doc := semanticDocument()
+	input := dnd.ItemRegistry{Items: []dnd.Item{{Name: "Star Compass", SourceRefs: []source.SourceRef{{SourceID: doc.ID, StartUnitID: 10, EndUnitID: 10}}}, {Name: "Compass", SourceRefs: []source.SourceRef{{SourceID: doc.ID, StartUnitID: 20, EndUnitID: 20}}}}}
+	invalid, err := newNormalizer(t, &recordingNormalizerClient{err: contracts.ErrInvalidStructuredOutput}).Normalize(context.Background(), normalizeRequestWithSource(input, doc))
+	if err != nil || invalid.Retry == nil || invalid.Retry.ReasonCode != ReasonCodeItemSemanticProposalInvalid {
+		t.Fatalf("invalid result = %#v, %v", invalid, err)
+	}
+	_, err = newNormalizer(t, &recordingNormalizerClient{err: errors.New("provider unavailable")}).Normalize(context.Background(), normalizeRequestWithSource(input, doc))
+	if err == nil || !strings.Contains(err.Error(), "provider unavailable") {
+		t.Fatalf("provider error = %v", err)
+	}
+	if bounded := limitWarningsForRetry(make([]contracts.Warning, diagnostics.MaxWarnings+5)); len(bounded) != diagnostics.MaxWarnings-1 || bounded[len(bounded)-1].ReasonCode != ReasonCodeItemNormalizationWarningsOmitted {
+		t.Fatalf("retry warning limit = %#v", bounded)
+	}
+	first, err := newNormalizer(t, &recordingNormalizerClient{}).Normalize(context.Background(), normalizeRequestWithSource(input, doc))
+	second, secondErr := newNormalizer(t, &recordingNormalizerClient{}).Normalize(context.Background(), normalizeRequestWithSource(first.Value, doc))
+	if err != nil || secondErr != nil || !reflect.DeepEqual(first.Value, second.Value) {
+		t.Fatalf("normalization must be idempotent: first=%#v err=%v second=%#v err=%v", first, err, second, secondErr)
+	}
+}
+
+func TestRegisterPromptAssetsPreparesItemNormalizationPrompt(t *testing.T) {
+	registry := llm.NewAssetRegistry()
+	if err := entityreconcile.RegisterSchemaAssets(registry); err != nil {
+		t.Fatal(err)
+	}
+	if err := RegisterPromptAssets(registry); err != nil {
+		t.Fatal(err)
+	}
+	options, err := registry.PromptKitOptions()
+	if err != nil {
+		t.Fatal(err)
+	}
+	options = append(options, promptkit.WithProfiles(promptkit.OpenAICompatibleProfile(promptkit.OpenAICompatibleProfileConfig{ID: "item-normalize-test", Endpoint: "http://127.0.0.1:1/v1", Model: "test"})))
+	engine, err := promptkit.NewEngine(promptkit.Config{Timeout: time.Second}, options...)
+	if err != nil {
+		t.Fatal(err)
+	}
+	prepared, err := engine.Prepare(context.Background(), promptkit.RunRequest{PromptID: PromptID, PromptVersion: entityreconcile.SchemaVersion, ProfileID: "item-normalize-test", Inputs: map[string]promptkit.ArtifactRef{"candidates": promptkit.Inline(`{"candidates":[{"key":"candidate-000001","name":"Rope","source_refs":[]}]}`), "transcript": promptkit.Inline(`{"windows":[{"units":[]}]}`)}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if prepared.OutputContract.SchemaPath != "dnd_entity_reconcile_llm.v1.json" || !strings.Contains(prepared.Messages[1].Content, "currency denominations") || !strings.Contains(prepared.Messages[1].Content, "materially different item types") {
+		t.Fatalf("prepared prompt = %#v", prepared)
+	}
+}
+
+type recordingNormalizerClient struct {
+	response string
+	err      error
+	requests []contracts.StructuredCompletionRequest
+}
+
+func (c *recordingNormalizerClient) CompleteStructured(_ context.Context, request contracts.StructuredCompletionRequest, output any) (contracts.StructuredCompletionResponse, error) {
+	c.requests = append(c.requests, request)
+	if c.err != nil {
+		return contracts.StructuredCompletionResponse{}, c.err
+	}
+	response := c.response
+	if response == "" {
+		response = `{"duplicate_groups":[]}`
+	}
+	if err := json.Unmarshal([]byte(response), output); err != nil {
+		return contracts.StructuredCompletionResponse{}, err
+	}
+	return contracts.StructuredCompletionResponse{Content: json.RawMessage(response)}, nil
+}
+
+func newNormalizer(t *testing.T, client contracts.StructuredLLMClient) *Normalizer {
+	t.Helper()
+	normalizer, err := New(client, Options{})
+	if err != nil {
+		t.Fatalf("New() error = %v", err)
+	}
+	return normalizer
+}
+func normalizeRequest(value dnd.ItemRegistry) contracts.TypedNormalizeRequest[dnd.ItemRegistry] {
+	return contracts.TypedNormalizeRequest[dnd.ItemRegistry]{MergeOutput: contracts.MergeArtifact[dnd.ItemRegistry]{Value: value}}
+}
+func normalizeRequestWithSource(value dnd.ItemRegistry, doc *source.SourceDocument) contracts.TypedNormalizeRequest[dnd.ItemRegistry] {
+	request := normalizeRequest(value)
+	request.Source = doc
+	return request
+}
+func semanticDocument() *source.SourceDocument {
+	return &source.SourceDocument{ID: "item-session", Units: []source.SourceUnit{{ID: 10, Text: "The Star Compass points north."}, {ID: 20, Text: "The compass of the stars glows."}, {ID: 30, Text: "The chest holds gold pieces."}}}
+}
+func hasWarning(warnings []contracts.Warning, reason string) bool {
+	for _, warning := range warnings {
+		if warning.ReasonCode == reason {
+			return true
+		}
+	}
+	return false
+}
--- a/internal/modules/dnd/normalize/itemregistry/prompt_assets.go
+++ b/internal/modules/dnd/normalize/itemregistry/prompt_assets.go
@@ -0,0 +1,62 @@
+package itemregistry
+
+import (
+	"fmt"
+	"io/fs"
+	"sync"
+
+	rootassets "gitea.maximumdirect.net/eric/notarius/assets"
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/llm"
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/promptfs"
+	"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/shared"
+)
+
+const promptAssetRoot = "assets/prompts"
+
+var promptAssetManifest = shared.PromptAssetManifest{
+	ModuleDir: PromptID,
+	ModuleFiles: []promptfs.ModulePromptFile{
+		{Name: "prompt.yaml", Path: "prompts/prompt.yaml"},
+		{Name: "instructions.md", Path: "prompts/instructions.md"},
+		{Name: "candidates.md", Path: "prompts/candidates.md"},
+	},
+	SharedFiles: []string{"common-dnd-system.md", "common-dnd-entity-reconciliation.md", "common-dnd-transcript-windows.md"},
+}
+
+func moduleAssetFS() (fs.FS, error) {
+	assets, err := fs.Sub(rootassets.FS(), "dnd/item-registry/normalize")
+	if err != nil {
+		return nil, fmt.Errorf("scope item normalization assets: %w", err)
+	}
+	return assets, nil
+}
+
+func RegisterPromptAssets(registry *llm.AssetRegistry) error {
+	assets, err := moduleAssetFS()
+	if err != nil {
+		return err
+	}
+	promptFS, err := promptAssetManifest.PromptFS(assets)
+	if err != nil {
+		return fmt.Errorf("prepare item normalization prompt assets: %w", err)
+	}
+	return registry.RegisterPromptFS(promptFS, promptAssetRoot)
+}
+
+func promptAssetMetadata() (string, error) {
+	promptAssetHashOnce.Do(func() {
+		assets, err := moduleAssetFS()
+		if err != nil {
+			promptAssetHashErr = err
+			return
+		}
+		promptAssetHash, promptAssetHashErr = promptAssetManifest.Hash(assets)
+	})
+	return promptAssetHash, promptAssetHashErr
+}
+
+var (
+	promptAssetHashOnce sync.Once
+	promptAssetHash     string
+	promptAssetHashErr  error
+)
--- a/internal/modules/dnd/normalize/itemregistry/reconciliation.go
+++ b/internal/modules/dnd/normalize/itemregistry/reconciliation.go
@@ -0,0 +1,153 @@
+package itemregistry
+
+import (
+	"fmt"
+
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
+	"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"
+)
+
+type safeReconciliationGroup struct {
+	members   []int
+	canonical int
+}
+
+func reconciliationCandidates(records []normalizedRecord) []entityreconcile.Candidate {
+	candidates := make([]entityreconcile.Candidate, len(records))
+	for index, record := range records {
+		candidates[index] = entityreconcile.Candidate{Name: record.item.Name, SourceRefs: cloneSourceRefs(record.item.SourceRefs)}
+	}
+	return candidates
+}
+
+func reconciliationGroups(assessment entityreconcile.Assessment, candidateKeys []string) []safeReconciliationGroup {
+	positions := make(map[string]int, len(candidateKeys))
+	for index, key := range candidateKeys {
+		positions[key] = index
+	}
+	safeGroups := assessment.SafeGroups()
+	groups := make([]safeReconciliationGroup, 0, len(safeGroups))
+	for _, group := range safeGroups {
+		members := group.Members()
+		memberPositions := make([]int, len(members))
+		valid := true
+		for index, key := range members {
+			position, ok := positions[key]
+			if !ok {
+				valid = false
+				break
+			}
+			memberPositions[index] = position
+		}
+		canonical, ok := positions[group.Canonical()]
+		if valid && ok {
+			groups = append(groups, safeReconciliationGroup{members: memberPositions, canonical: canonical})
+		}
+	}
+	return groups
+}
+
+func reconciliationIssues(assessment entityreconcile.Assessment) []string {
+	issues := assessment.Issues()
+	details := make([]string, len(issues))
+	for index, issue := range issues {
+		details[index] = fmt.Sprintf("group %d: %s", issue.GroupIndex, issue.Category)
+	}
+	return details
+}
+
+func applySafeGroups(records []normalizedRecord, groups []safeReconciliationGroup, order shared.SourceRefOrder) ([]normalizedRecord, []contracts.Warning, int) {
+	byMember := make(map[int]safeReconciliationGroup, len(groups)*2)
+	for _, group := range groups {
+		for _, member := range group.members {
+			byMember[member] = group
+		}
+	}
+	output := make([]normalizedRecord, 0, len(records)-len(groups))
+	warnings := make([]contracts.Warning, 0, len(groups))
+	rejectedGroups := 0
+	for index, record := range records {
+		group, grouped := byMember[index]
+		if !grouped {
+			output = append(output, cloneRecord(record))
+			continue
+		}
+		if group.members[0] != index {
+			continue
+		}
+		if !canConsolidate(records, group) {
+			for _, member := range group.members {
+				output = append(output, cloneRecord(records[member]))
+			}
+			warnings = append(warnings, contracts.Warning{Scope: itemScope(records[group.members[0]].earliest), ReasonCode: ReasonCodeItemSemanticProposalInvalid, Message: "proposal group preserved because currency denominations must remain distinct"})
+			rejectedGroups++
+			continue
+		}
+		consolidated := consolidateSemanticGroup(records, group, order)
+		output = append(output, consolidated)
+		warnings = append(warnings, semanticDuplicateWarning(consolidated, records[group.canonical]))
+	}
+	return output, warnings, rejectedGroups
+}
+
+func canConsolidate(records []normalizedRecord, group safeReconciliationGroup) bool {
+	denomination := ""
+	for _, member := range group.members {
+		current := currencyDenomination(records[member].item.Name)
+		if current == "" {
+			continue
+		}
+		if denomination != "" && denomination != current {
+			return false
+		}
+		denomination = current
+	}
+	return true
+}
+
+func currencyDenomination(name string) string {
+	switch identity.ComparisonKey(name) {
+	case "copper piece", "copper pieces", "cp":
+		return "copper"
+	case "silver piece", "silver pieces", "sp":
+		return "silver"
+	case "electrum piece", "electrum pieces", "ep":
+		return "electrum"
+	case "gold piece", "gold pieces", "gp":
+		return "gold"
+	case "platinum piece", "platinum pieces", "pp":
+		return "platinum"
+	default:
+		return ""
+	}
+}
+
+func consolidateSemanticGroup(records []normalizedRecord, group safeReconciliationGroup, order shared.SourceRefOrder) normalizedRecord {
+	output := cloneRecord(records[group.members[0]])
+	output.item.Name = records[group.canonical].item.Name
+	for _, member := range group.members[1:] {
+		output.item.SourceRefs = append(output.item.SourceRefs, records[member].item.SourceRefs...)
+		output.inputIndexes = append(output.inputIndexes, records[member].inputIndexes...)
+		if records[member].earliest < output.earliest {
+			output.earliest = records[member].earliest
+		}
+	}
+	output.inputIndexes = sortedUniqueIndexes(output.inputIndexes)
+	output.item.SourceRefs = order.Canonicalize(output.item.SourceRefs)
+	output.item.ID = identity.DeriveID(output.item.Name)
+	return output
+}
+
+func semanticDuplicateWarning(record normalizedRecord, canonical normalizedRecord) contracts.Warning {
+	details := make([]string, 0, len(record.inputIndexes)+1)
+	for _, inputIndex := range record.inputIndexes {
+		details = append(details, fmt.Sprintf("input index %d", inputIndex))
+	}
+	if canonical.earliest != record.earliest {
+		details = append(details, fmt.Sprintf("canonical display name from input index %d", canonical.earliest))
+	}
+	return contracts.Warning{Scope: itemScope(record.earliest), ReasonCode: ReasonCodeDuplicateItemCollapsed, Message: diagnostics.Aggregate("semantic duplicate consolidation", details)}
+}