diff --git a/internal/modules/dnd/normalize/locations/assets.go b/internal/modules/dnd/normalize/locations/assets.go new file mode 100644 index 0000000..4283674 --- /dev/null +++ b/internal/modules/dnd/normalize/locations/assets.go @@ -0,0 +1,6 @@ +package locations + +import "embed" + +//go:embed assets/prompts/*.yaml assets/prompts/*.md +var embeddedAssets embed.FS diff --git a/internal/modules/dnd/normalize/locations/assets/prompts/candidates.md b/internal/modules/dnd/normalize/locations/assets/prompts/candidates.md new file mode 100644 index 0000000..61c17ca --- /dev/null +++ b/internal/modules/dnd/normalize/locations/assets/prompts/candidates.md @@ -0,0 +1,2 @@ +Location candidates: +{{ input "candidates" }} diff --git a/internal/modules/dnd/normalize/locations/assets/prompts/dnd.locations.normalize.yaml b/internal/modules/dnd/normalize/locations/assets/prompts/dnd.locations.normalize.yaml new file mode 100644 index 0000000..c0ecb1d --- /dev/null +++ b/internal/modules/dnd/normalize/locations/assets/prompts/dnd.locations.normalize.yaml @@ -0,0 +1,30 @@ +id: dnd.locations.normalize +version: "v1" +default_profile: dnd-extraction +inputs: + - name: candidates + required: true + content_type: application/json + - name: transcript + required: true + content_type: application/json +messages: + - role: system + content_file: ./sharedassets/common-dnd-system.md + - role: user + content_file: ./task.md + - role: user + content_file: ./sharedassets/common-dnd-entity-reconciliation.md + cache_control: + type: ephemeral + - role: user + content_file: ./candidates.md + - role: user + content_file: ./sharedassets/common-dnd-transcript.md + cache_control: + type: ephemeral +output: + format: json + validation_mode: json_schema + schema_path: dnd_entity_reconcile_llm.v1.json + repair_attempts: 0 diff --git a/internal/modules/dnd/normalize/locations/assets/prompts/task.md b/internal/modules/dnd/normalize/locations/assets/prompts/task.md new file mode 100644 index 0000000..b691e8e --- /dev/null +++ b/internal/modules/dnd/normalize/locations/assets/prompts/task.md @@ -0,0 +1,8 @@ +Review location candidates and cited transcript context. Group candidates only +when the evidence clearly identifies one physical place. + +Do not group candidates solely because their names match, their evidence is +nearby, one place is nested inside another, or their labels are generic. Keep +parent and child places, similarly named places, and uncertain aliases +separate. For an accepted group, select the supplied candidate with the +clearest established display name as canonical. diff --git a/internal/modules/dnd/normalize/locations/normalizer.go b/internal/modules/dnd/normalize/locations/normalizer.go new file mode 100644 index 0000000..3b08c4e --- /dev/null +++ b/internal/modules/dnd/normalize/locations/normalizer.go @@ -0,0 +1,343 @@ +// Package locations normalizes merged D&D location candidates conservatively. +package locations + +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/locations/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/locations" + PromptID = "dnd.locations.normalize" + normalizationPolicy = "dnd.locations.normalize.v1" + semanticContextPolicy = "dnd.entity_reconcile.context.v1" + semanticContextRadius = 2 + NormalizationPolicy = normalizationPolicy + + ReasonCodeLocationFieldsNormalized = "location_fields_normalized" + ReasonCodeLocationIDRecomputed = "location_id_recomputed" + ReasonCodeSourceReferencesNormalized = "source_references_normalized" + ReasonCodeDuplicateLocationCollapsed = "duplicate_location_collapsed" + ReasonCodeLocationSemanticProposalInvalid = "location_semantic_proposal_invalid" + ReasonCodeLocationSemanticReconciliationExhausted = "location_semantic_reconciliation_exhausted" + ReasonCodeLocationNormalizationWarningsOmitted = "location_normalization_warnings_omitted" +) + +var requiredCapabilities = []string{"merged"} +var providedCapabilities = []string{"normalized"} + +var _ contracts.Normalizer[dnd.LocationList] = (*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.LocationList]) (contracts.TypedNormalizeResult[dnd.LocationList], error) { + if n == nil { + return contracts.TypedNormalizeResult[dnd.LocationList]{}, normalizerErrorf("normalizer must not be nil") + } + if n.llm == nil { + return contracts.TypedNormalizeResult[dnd.LocationList]{}, normalizerErrorf("LLM client must not be nil") + } + if ctx == nil { + return contracts.TypedNormalizeResult[dnd.LocationList]{}, normalizerErrorf("context must not be nil") + } + if err := ctx.Err(); err != nil { + return contracts.TypedNormalizeResult[dnd.LocationList]{}, 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.LocationList]{}, normalizerErrorf("build semantic context: %w", err) + } + if !ready { + return contracts.TypedNormalizeResult[dnd.LocationList]{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.LocationList]{}, normalizerErrorf("complete structured output: %w", err) + } + assessment := materials.Assess(response) + applied, semanticWarnings := applySafeGroups(records, reconciliationGroups(assessment, materials.CandidateKeys()), order) + warnings = append(warnings, semanticWarnings...) + if assessment.DiscardedGroups() == 0 { + return contracts.TypedNormalizeResult[dnd.LocationList]{Value: recordList(applied), Warnings: limitWarnings(warnings)}, nil + } + return retryResult(recordList(applied), warnings, assessment), nil +} + +func (n *Normalizer) invalidStructuredResult(value dnd.LocationList, warnings []contracts.Warning) contracts.TypedNormalizeResult[dnd.LocationList] { + return contracts.TypedNormalizeResult[dnd.LocationList]{Value: value, Warnings: limitWarningsForRetry(warnings), Retry: &contracts.NormalizeRetry{ + ReasonCode: ReasonCodeLocationSemanticProposalInvalid, Message: "semantic proposal requires retry: invalid structured output", + FallbackWarnings: []contracts.Warning{semanticFallbackWarning(-1)}, + }} +} + +func retryResult(value dnd.LocationList, warnings []contracts.Warning, assessment entityreconcile.Assessment) contracts.TypedNormalizeResult[dnd.LocationList] { + return contracts.TypedNormalizeResult[dnd.LocationList]{Value: value, Warnings: limitWarningsForRetry(warnings), Retry: &contracts.NormalizeRetry{ + ReasonCode: ReasonCodeLocationSemanticProposalInvalid, Message: diagnostics.Aggregate("semantic proposal requires retry", reconciliationIssues(assessment)), + FallbackWarnings: []contracts.Warning{semanticFallbackWarning(assessment.DiscardedGroups())}, + }} +} + +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: "locations", ReasonCode: ReasonCodeLocationSemanticReconciliationExhausted, Message: message} +} + +func limitWarnings(warnings []contracts.Warning) []contracts.Warning { + return diagnostics.LimitWarnings(warnings, "locations", ReasonCodeLocationNormalizationWarningsOmitted) +} + +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: "locations", ReasonCode: ReasonCodeLocationNormalizationWarningsOmitted, Message: fmt.Sprintf("%d additional warning(s) omitted", len(warnings)-displayed)}) +} + +type normalizedRecord struct { + location dnd.Location + inputIndexes []int + earliest int +} + +func preprocessRecords(input dnd.LocationList, order shared.SourceRefOrder) ([]normalizedRecord, []contracts.Warning) { + if input.Locations == nil { + return nil, nil + } + records := make([]normalizedRecord, len(input.Locations)) + warnings := make([]contracts.Warning, 0) + for index, inputLocation := range input.Locations { + location, fieldsChanged, refsChanged := normalizeRecord(inputLocation, order) + records[index] = normalizedRecord{location: location, inputIndexes: []int{index}, earliest: index} + if fieldsChanged { + warnings = append(warnings, contracts.Warning{Scope: locationScope(index), ReasonCode: ReasonCodeLocationFieldsNormalized, Message: fmt.Sprintf("input index %d: location name normalized for %s", index, diagnostics.Quote(inputLocation.Name))}) + } + if refsChanged { + warnings = append(warnings, contracts.Warning{Scope: locationScope(index), ReasonCode: ReasonCodeSourceReferencesNormalized, Message: fmt.Sprintf("input index %d: source references normalized (original count %d, final count %d)", index, len(inputLocation.SourceRefs), len(location.SourceRefs))}) + } + if inputLocation.ID != location.ID { + warnings = append(warnings, contracts.Warning{Scope: locationScope(index), ReasonCode: ReasonCodeLocationIDRecomputed, Message: fmt.Sprintf("input index %d: location ID recomputed from %s", index, diagnostics.Quote(location.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.Location, order shared.SourceRefOrder) (dnd.Location, bool, bool) { + output := cloneLocation(input) + output.Name = identity.NormalizeDisplay(input.Name) + output.SourceRefs = order.Canonicalize(input.SourceRefs) + output.ID = identity.DeriveID(output.Name, output.SourceRefs) + 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.location.Name) + found := false + for groupIndex, members := range groups { + first := records[members[0]] + if identity.ComparisonKey(first.location.Name) == key && reflect.DeepEqual(first.location.SourceRefs, record.location.SourceRefs) { + groups[groupIndex] = append(groups[groupIndex], index) + found = true + break + } + } + if !found { + groups = append(groups, []int{index}) + } + } + return groups +} + +func cloneLocation(input dnd.Location) dnd.Location { + input.SourceRefs = cloneSourceRefs(input.SourceRefs) + return input +} +func cloneRecord(input normalizedRecord) normalizedRecord { + input.location = cloneLocation(input.location) + input.inputIndexes = append([]int(nil), input.inputIndexes...) + return input +} +func cloneSourceRefs(input []source.SourceRef) []source.SourceRef { + if input == nil { + return nil + } + 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.Location { + if records == nil { + return nil + } + values := make([]dnd.Location, len(records)) + for index, record := range records { + values[index] = cloneLocation(record.location) + } + return values +} +func recordList(records []normalizedRecord) dnd.LocationList { + if records == nil { + return dnd.LocationList{} + } + return dnd.LocationList{Locations: 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: locationScope(retainedIndex), ReasonCode: ReasonCodeDuplicateLocationCollapsed, Message: message} +} +func locationScope(index int) string { return fmt.Sprintf("locations[%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.LocationListKind} +} +func Register(registry *pipeline.NormalizerRegistry) error { + return pipeline.RegisterNormalizerBuilder(registry, ModuleSpec(), validateOptions, func(request pipeline.BuildRequest) (contracts.Normalizer[dnd.LocationList], 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 locations normalizer: "+format, args...) +} diff --git a/internal/modules/dnd/normalize/locations/normalizer_test.go b/internal/modules/dnd/normalize/locations/normalizer_test.go new file mode 100644 index 0000000..fd1dff4 --- /dev/null +++ b/internal/modules/dnd/normalize/locations/normalizer_test.go @@ -0,0 +1,136 @@ +package locations + +import ( + "context" + "errors" + "reflect" + "strings" + "testing" + + "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/locations/identity" + "gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/shared/entityreconcile" +) + +func TestModuleContractAndMetadata(t *testing.T) { + want := pipeline.ModuleSpec{Key: Key, Stage: pipeline.StageNormalize, ExecutionClass: contracts.ExecutionClassLLMBacked, Requires: []string{"merged"}, Provides: []string{"normalized"}, ArtifactKind: dnd.LocationListKind} + 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") + } + normalizer := newNormalizer(t, &recordingLocationNormalizerClient{}) + metadata := normalizer.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) + } + if got := normalizer.CheckpointFingerprints(); len(got) != 5 || got[2].Value != identity.Policy || got[3].Value != normalizationPolicy || got[4].Value != semanticContextPolicy+":2" { + t.Fatalf("fingerprints = %#v", got) + } +} + +func TestNormalizePreparesOnlyExactDuplicatesAndRetainsSameNameAndNestedPlaces(t *testing.T) { + input := dnd.LocationList{Locations: []dnd.Location{ + {Name: " The Tavern ", SourceRefs: []source.SourceRef{{SourceID: "session", StartUnitID: 1, EndUnitID: 1}}}, + {Name: "the tavern", SourceRefs: []source.SourceRef{{SourceID: "session", StartUnitID: 1, EndUnitID: 1}}}, + {Name: "The Tavern", SourceRefs: []source.SourceRef{{SourceID: "session", StartUnitID: 2, EndUnitID: 2}}}, + {Name: "The Tavern Cellar", SourceRefs: []source.SourceRef{{SourceID: "session", StartUnitID: 2, EndUnitID: 2}}}, + }} + before := dnd.LocationList{Locations: append([]dnd.Location(nil), input.Locations...)} + result, err := newNormalizer(t, &recordingLocationNormalizerClient{}).Normalize(context.Background(), normalizeRequest(input)) + if err != nil || len(result.Value.Locations) != 3 { + t.Fatalf("Normalize() = %#v, %v; want one exact duplicate removed", result, err) + } + if !reflect.DeepEqual(input, before) { + t.Fatalf("Normalize() mutated input: %#v", input) + } + if got := []string{result.Value.Locations[0].Name, result.Value.Locations[1].Name, result.Value.Locations[2].Name}; !reflect.DeepEqual(got, []string{"The Tavern", "The Tavern", "The Tavern Cellar"}) { + t.Fatalf("locations = %#v, want same names and nested place retained", got) + } + if result.Value.Locations[0].ID == result.Value.Locations[1].ID || !hasWarning(result.Warnings, ReasonCodeDuplicateLocationCollapsed) { + t.Fatalf("result = %#v, want evidence-anchored IDs and exact duplicate warning", result) + } +} + +func TestNormalizeAppliesSafeAliasGroupAndUsesOpaqueInputs(t *testing.T) { + client := &recordingLocationNormalizerClient{response: `{"duplicate_groups":[{"members":["candidate-000001","candidate-000002"],"canonical":"candidate-000002"}]}`} + doc := semanticDocument() + input := dnd.LocationList{Locations: []dnd.Location{ + {Name: "Old Mill", SourceRefs: []source.SourceRef{{SourceID: doc.ID, StartUnitID: 10, EndUnitID: 10}}}, + {Name: "the Greencloak's refuge", SourceRefs: []source.SourceRef{{SourceID: doc.ID, StartUnitID: 20, EndUnitID: 20}}}, + {Name: "The Tavern", SourceRefs: []source.SourceRef{{SourceID: doc.ID, StartUnitID: 30, EndUnitID: 30}}}, + }} + result, err := newNormalizer(t, client).Normalize(context.Background(), normalizeRequestWithSource(input, doc)) + if err != nil || result.Retry != nil || len(result.Value.Locations) != 2 { + t.Fatalf("Normalize() = %#v, %v", result, err) + } + merged := result.Value.Locations[0] + if merged.Name != "the Greencloak's refuge" || merged.ID != identity.DeriveID(merged.Name, merged.SourceRefs) || len(merged.SourceRefs) != 2 || !hasWarning(result.Warnings, ReasonCodeDuplicateLocationCollapsed) { + t.Fatalf("merged location = %#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) + } +} + +func TestNormalizeRejectsUnsafeAndOverlappingGroupsWithoutLosingCandidates(t *testing.T) { + doc := semanticDocument() + input := dnd.LocationList{Locations: []dnd.Location{ + {Name: "Old Mill", SourceRefs: []source.SourceRef{{SourceID: doc.ID, StartUnitID: 10, EndUnitID: 10}}}, + {Name: "Mill", SourceRefs: []source.SourceRef{{SourceID: doc.ID, StartUnitID: 20, EndUnitID: 20}}}, + {Name: "Tavern", SourceRefs: []source.SourceRef{{SourceID: doc.ID, StartUnitID: 30, EndUnitID: 30}}}, + }} + client := &recordingLocationNormalizerClient{response: `{"duplicate_groups":[{"members":["candidate-000001","candidate-000002"],"canonical":"candidate-000001"},{"members":["candidate-000002","candidate-000003"],"canonical":"candidate-000003"}]}`} + result, err := newNormalizer(t, client).Normalize(context.Background(), normalizeRequestWithSource(input, doc)) + if err != nil || result.Retry == nil || len(result.Value.Locations) != 3 || !strings.Contains(result.Retry.Message, "overlapping_member") { + t.Fatalf("Normalize() = %#v, %v; want safe retry fallback", result, err) + } + if len(result.Retry.FallbackWarnings) != 1 || !strings.Contains(result.Retry.FallbackWarnings[0].Message, "2 proposal group") { + t.Fatalf("fallback warnings = %#v", result.Retry.FallbackWarnings) + } +} + +func TestNormalizeHandlesRetryFallbackAndErrors(t *testing.T) { + doc := semanticDocument() + input := dnd.LocationList{Locations: []dnd.Location{{Name: "Old Mill", SourceRefs: []source.SourceRef{{SourceID: doc.ID, StartUnitID: 10, EndUnitID: 10}}}, {Name: "Mill", SourceRefs: []source.SourceRef{{SourceID: doc.ID, StartUnitID: 20, EndUnitID: 20}}}}} + invalid, err := newNormalizer(t, &recordingLocationNormalizerClient{err: contracts.ErrInvalidStructuredOutput}).Normalize(context.Background(), normalizeRequestWithSource(input, doc)) + if err != nil || invalid.Retry == nil || invalid.Retry.ReasonCode != ReasonCodeLocationSemanticProposalInvalid { + t.Fatalf("invalid result = %#v, %v", invalid, err) + } + _, err = newNormalizer(t, &recordingLocationNormalizerClient{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) + } + warnings := make([]contracts.Warning, 25) + bounded := limitWarningsForRetry(warnings) + if len(bounded) != 19 || bounded[len(bounded)-1].ReasonCode != ReasonCodeLocationNormalizationWarningsOmitted { + t.Fatalf("retry warning limit = %#v", bounded) + } +} + +func TestNormalizeOrdersEvidenceAndIsIdempotent(t *testing.T) { + doc := &source.SourceDocument{ID: "ordered", Units: []source.SourceUnit{{ID: 30}, {ID: 10}}} + input := dnd.LocationList{Locations: []dnd.Location{{Name: "Old Mill", SourceRefs: []source.SourceRef{ + {SourceID: doc.ID, StartUnitID: 10, EndUnitID: 10}, + {SourceID: doc.ID, StartUnitID: 30, EndUnitID: 30}, + {SourceID: doc.ID, StartUnitID: 30, EndUnitID: 30}, + }}}} + normalizer := newNormalizer(t, &recordingLocationNormalizerClient{}) + first, err := normalizer.Normalize(context.Background(), normalizeRequestWithSource(input, doc)) + if err != nil || !reflect.DeepEqual([]int{first.Value.Locations[0].SourceRefs[0].StartUnitID, first.Value.Locations[0].SourceRefs[1].StartUnitID}, []int{30, 10}) { + t.Fatalf("first Normalize() = %#v, %v; want document-ordered evidence", first, err) + } + second, err := normalizer.Normalize(context.Background(), normalizeRequestWithSource(first.Value, doc)) + if err != nil || !reflect.DeepEqual(second.Value, first.Value) { + t.Fatalf("second Normalize() = %#v, %v; want idempotent value %#v", second, err, first.Value) + } +} diff --git a/internal/modules/dnd/normalize/locations/prompt_assets.go b/internal/modules/dnd/normalize/locations/prompt_assets.go new file mode 100644 index 0000000..2f7e40e --- /dev/null +++ b/internal/modules/dnd/normalize/locations/prompt_assets.go @@ -0,0 +1,41 @@ +package locations + +import ( + "fmt" + "sync" + + "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: "dnd.locations.normalize.yaml", Path: "assets/prompts/dnd.locations.normalize.yaml"}, + {Name: "task.md", Path: "assets/prompts/task.md"}, + {Name: "candidates.md", Path: "assets/prompts/candidates.md"}, + }, + SharedFiles: []string{"common-dnd-system.md", "common-dnd-entity-reconciliation.md", "common-dnd-transcript.md"}, +} + +func RegisterPromptAssets(registry *llm.AssetRegistry) error { + promptFS, err := promptAssetManifest.PromptFS(embeddedAssets) + if err != nil { + return fmt.Errorf("prepare location normalization prompt assets: %w", err) + } + return registry.RegisterPromptFS(promptFS, promptAssetRoot) +} + +func promptAssetMetadata() (string, error) { + promptAssetHashOnce.Do(func() { promptAssetHash, promptAssetHashErr = promptAssetManifest.Hash(embeddedAssets) }) + return promptAssetHash, promptAssetHashErr +} + +var ( + promptAssetHashOnce sync.Once + promptAssetHash string + promptAssetHashErr error +) diff --git a/internal/modules/dnd/normalize/locations/prompt_assets_test.go b/internal/modules/dnd/normalize/locations/prompt_assets_test.go new file mode 100644 index 0000000..2a93f1a --- /dev/null +++ b/internal/modules/dnd/normalize/locations/prompt_assets_test.go @@ -0,0 +1,46 @@ +package locations + +import ( + "context" + "strings" + "testing" + "time" + + "gitea.maximumdirect.net/eric/notarius/internal/framework/llm" + "gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/shared/entityreconcile" + "gitea.maximumdirect.net/eric/promptkit" +) + +func TestRegisterPromptAssetsPreparesLocationNormalizationPrompt(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: "location-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: "location-normalize-test", Inputs: map[string]promptkit.ArtifactRef{"candidates": promptkit.Inline(`{"candidates":[{"key":"candidate-000001","name":"The Tavern","source_refs":[]}]}`), "transcript": promptkit.Inline(`{"windows":[{"units":[]}]}`)}}) + if err != nil { + t.Fatal(err) + } + if prepared.OutputContract.SchemaPath != "dnd_entity_reconcile_llm.v1.json" || len(prepared.Messages) != 5 { + t.Fatalf("prepared prompt = %#v", prepared) + } + for _, index := range []int{2, 4} { + if prepared.Messages[index].CacheControl == nil { + t.Fatalf("message %d cache = %#v", index, prepared.Messages[index].CacheControl) + } + } + if !strings.Contains(prepared.Messages[3].Content, "candidate-000001") || strings.Contains(prepared.Messages[3].Content, `"windows"`) { + t.Fatalf("candidate message = %q", prepared.Messages[3].Content) + } +} diff --git a/internal/modules/dnd/normalize/locations/reconciliation.go b/internal/modules/dnd/normalize/locations/reconciliation.go new file mode 100644 index 0000000..0a73664 --- /dev/null +++ b/internal/modules/dnd/normalize/locations/reconciliation.go @@ -0,0 +1,112 @@ +package locations + +import ( + "fmt" + + "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" + "gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/locations/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.location.Name, SourceRefs: cloneSourceRefs(record.location.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) { + 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)) + for index, record := range records { + group, grouped := byMember[index] + if !grouped { + output = append(output, cloneRecord(record)) + continue + } + if group.members[0] != index { + continue + } + consolidated := consolidateSemanticGroup(records, group, order) + output = append(output, consolidated) + warnings = append(warnings, semanticDuplicateWarning(consolidated, records[group.canonical])) + } + return output, warnings +} + +func consolidateSemanticGroup(records []normalizedRecord, group safeReconciliationGroup, order shared.SourceRefOrder) normalizedRecord { + output := cloneRecord(records[group.members[0]]) + output.location.Name = records[group.canonical].location.Name + for _, member := range group.members[1:] { + output.location.SourceRefs = append(output.location.SourceRefs, records[member].location.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.location.SourceRefs = order.Canonicalize(output.location.SourceRefs) + output.location.ID = identity.DeriveID(output.location.Name, output.location.SourceRefs) + 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: locationScope(record.earliest), ReasonCode: ReasonCodeDuplicateLocationCollapsed, Message: diagnostics.Aggregate("semantic duplicate consolidation", details)} +} diff --git a/internal/modules/dnd/normalize/locations/test_helpers_test.go b/internal/modules/dnd/normalize/locations/test_helpers_test.go new file mode 100644 index 0000000..aed0a49 --- /dev/null +++ b/internal/modules/dnd/normalize/locations/test_helpers_test.go @@ -0,0 +1,60 @@ +package locations + +import ( + "context" + "encoding/json" + "testing" + + "gitea.maximumdirect.net/eric/notarius/internal/core/source" + "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" + "gitea.maximumdirect.net/eric/notarius/internal/modules/dnd" +) + +type recordingLocationNormalizerClient struct { + response string + err error + requests []contracts.StructuredCompletionRequest +} + +func (c *recordingLocationNormalizerClient) 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.LocationList) contracts.TypedNormalizeRequest[dnd.LocationList] { + return contracts.TypedNormalizeRequest[dnd.LocationList]{MergeOutput: contracts.MergeArtifact[dnd.LocationList]{Value: value}} +} +func normalizeRequestWithSource(value dnd.LocationList, doc *source.SourceDocument) contracts.TypedNormalizeRequest[dnd.LocationList] { + request := normalizeRequest(value) + request.Source = doc + return request +} +func semanticDocument() *source.SourceDocument { + return &source.SourceDocument{ID: "location-session", Units: []source.SourceUnit{{ID: 10, Kind: "speech", Text: "The old mill is the Greencloak's refuge."}, {ID: 20, Kind: "speech", Text: "The mill stands on the northern road."}, {ID: 30, Kind: "speech", Text: "The tavern is beside the mill."}, {ID: 40, Kind: "speech", Text: "The mill's cellar is flooded."}}} +} +func hasWarning(warnings []contracts.Warning, reason string) bool { + for _, warning := range warnings { + if warning.ReasonCode == reason { + return true + } + } + return false +}