Integrate NPC semantic normalization

2026-07-26 01:41:35 +00:00
parent d1c48db4bc
commit 8d9a496935
4 changed files with 234 additions and 16 deletions
--- a/internal/modules/dnd/register/register_test.go
+++ b/internal/modules/dnd/register/register_test.go
@@ -32,6 +32,20 @@ func TestRegisterAddsDNDFamily(t *testing.T) {
 	if err := Register(registries, assets); err != nil {
 		t.Fatalf("Register() error = %v, want nil", err)
 	}
+	promptFS, err := assets.PromptFS()
+	if err != nil {
+		t.Fatalf("PromptFS() error = %v", err)
+	}
+	if _, err := fs.ReadFile(promptFS, "dnd.npcs.normalize/dnd.npcs.normalize.yaml"); err != nil {
+		t.Fatalf("normalization prompt asset = %v, want registered private prompt", err)
+	}
+	schemaFS, err := assets.SchemaFS()
+	if err != nil {
+		t.Fatalf("SchemaFS() error = %v", err)
+	}
+	if _, err := fs.ReadFile(schemaFS, "dnd_npcs_normalize_llm.v1.json"); err != nil {
+		t.Fatalf("normalization schema asset = %v, want registered private schema", err)
+	}
 	assertContainsKeys(t, "chunkers", registries.Chunkers.RegisteredKeys(), []string{"dnd/scenes"})
 	assertContainsKeys(t, "extractors", registries.Extractors.RegisteredKeys(), []string{"dnd/spells", npcextract.Key, combatextract.Key, itemeventextract.Key, interactionextract.Key, scenedescriptionextract.Key})
 	assertContainsKeys(t, "normalizers", registries.Normalizers.RegisteredKeys(), []string{spellnormalize.Key, npcnormalize.Key, combatnormalize.Key, itemeventnormalize.Key, interactionnormalize.Key, scenedescriptionnormalize.Key, pipeline.DefaultNormalizeModule})
--- a/internal/modules/integration/dnd_npc_grounded_test.go
+++ b/internal/modules/integration/dnd_npc_grounded_test.go
@@ -24,6 +24,7 @@ import (
 	"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/extract/npcs"
 	sceneextract "gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/extract/scenedescriptions"
 	"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/extract/spells"
+	npcnormalize "gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/normalize/npcs"
 )

 func TestNPCOutputGroundsSpellAndCombatConsumersThroughOneOperation(t *testing.T) {
@@ -47,6 +48,9 @@ func TestNPCOutputGroundsSpellAndCombatConsumersThroughOneOperation(t *testing.T
 	}
 	for name, value := range map[string]string{
 		"extract:npcs:dnd/npcs:mapping_policy":              "dnd.npcs.extract_mapping.v2",
+		"normalize:npcs:dnd/npcs:identity_policy":           "dnd.npcs.identity.v1",
+		"normalize:npcs:dnd/npcs:normalization_policy":      "dnd.npcs.normalize.v3",
+		"normalize:npcs:dnd/npcs:semantic_context_policy":   "dnd.npcs.semantic_context.v1:2",
 		"extract:spells:dnd/spells:mapping_policy":          "dnd.spells.extract_mapping.v2",
 		"extract:combat:dnd/combat-turns:scene_gate_policy": "dnd.combat_turns.scene_gate.v1",
 	} {
@@ -55,6 +59,8 @@ func TestNPCOutputGroundsSpellAndCombatConsumersThroughOneOperation(t *testing.T
 	for _, name := range []string{
 		"extract:npcs:dnd/npcs:prompt",
 		"extract:npcs:dnd/npcs:response_schema",
+		"normalize:npcs:dnd/npcs:prompt",
+		"normalize:npcs:dnd/npcs:response_schema",
 		"extract:spells:dnd/spells:prompt",
 		"extract:spells:dnd/spells:response_schema",
 		"extract:spells:dnd/spells:npc_registry",
@@ -455,6 +461,8 @@ func (client *groundedDNDLLMClient) CompleteStructured(ctx context.Context, requ
 				"name": "Hooded Guard", "source_refs": []any{map[string]int{"start_unit_id": 3, "end_unit_id": 3}},
 			},
 		}}
+	case npcnormalize.PromptID:
+		payload = map[string]any{"duplicate_groups": []any{}}
 	case sceneextract.PromptID:
 		kind := client.sceneKind
 		if kind == "" {
--- a/internal/modules/integration/dnd_npc_interactions_runner_test.go
+++ b/internal/modules/integration/dnd_npc_interactions_runner_test.go
@@ -5,6 +5,8 @@ import (
 	"encoding/json"
 	"fmt"
 	"os"
+	"reflect"
+	"strings"
 	"sync"
 	"testing"

@@ -16,6 +18,8 @@ import (
 	npccodec "gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/codec/npcs"
 	interactionextract "gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/extract/npcinteractions"
 	"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/extract/npcs"
+	npcnormalize "gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/normalize/npcs"
+	"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/npcs/identity"
 )

 func TestNPCInteractionPipelineUsesAcceptedRegistryAndCurrentEvidence(t *testing.T) {
@@ -70,6 +74,59 @@ func TestNPCInteractionPipelineUsesAcceptedRegistryAndCurrentEvidence(t *testing
 	}
 }

+func TestSemanticNPCNormalizationCrossesOrderedRegistryHandoff(t *testing.T) {
+	registries := productionNPCRegistries(t)
+	cfg := loadNPCInteractionPipelineConfig(t)
+	profile := cfg.Pipelines["dnd-npc-interactions-fixture"]
+	profile.Chunk = pipeline.ModuleBinding{Module: "generic", Options: map[string]any{"max_units": 1}}
+	cfg.Pipelines["dnd-npc-interactions-fixture"] = profile
+	effective, err := cfg.Resolve(config.ResolveInput{PipelineID: "dnd-npc-interactions-fixture", Catalog: moduleCatalog(registries)})
+	if err != nil {
+		t.Fatal(err)
+	}
+	resolved, warnings, err := pipeline.MaterializeReferences(effective.ResolvedPipeline, moduleCatalog(registries), pipeline.ReferenceMaterializationOptions{})
+	if err != nil || len(warnings) != 0 {
+		t.Fatalf("MaterializeReferences() error = %v warnings = %#v", err, warnings)
+	}
+	client := &semanticNPCInteractionClient{}
+	raw := []byte(`{"metadata":{"id":"semantic-session","title":"Semantic NPC session"},"segments":[{"id":1,"start":0,"end":1,"speaker":"DM","text":"Mira Thorn enters."},{"id":2,"start":1,"end":2,"speaker":"DM","text":"Mira Thorn, the Greencloak, waves."}]}`)
+	output, err := runPreparedPipeline(t, registries, resolved, client, pipeline.RunInput{RawInput: raw})
+	if err != nil {
+		t.Fatalf("Run() error = %v", err)
+	}
+	if client.requestCount(npcs.PromptID) != 2 || client.requestCount(npcnormalize.PromptID) != 1 {
+		t.Fatalf("prompt requests = %#v, want two extraction calls and one document normalization call", client.requests)
+	}
+	npcOutput := normalizedLane(t, output, "npcs")
+	npcsValue, err := npccodec.New().Decode(npcOutput.Artifact.Content)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if npcOutput.StepID != "identify-npcs" || len(npcsValue.NPCs) != 1 || npcsValue.NPCs[0].Name != "Mira Thorn" || npcsValue.NPCs[0].ID != identity.DeriveID("Mira Thorn") {
+		t.Fatalf("NPC output = %#v / %#v, want canonical ordered producer artifact", npcOutput, npcsValue)
+	}
+	if refs := npcsValue.NPCs[0].SourceRefs; !reflect.DeepEqual(refs, []source.SourceRef{{SourceID: "semantic-session", StartUnitID: 1, EndUnitID: 1}, {SourceID: "semantic-session", StartUnitID: 2, EndUnitID: 2}}) {
+		t.Fatalf("NPC evidence = %#v, want original extraction evidence union", refs)
+	}
+	interactionOutput := normalizedLane(t, output, "interactions")
+	if interactionOutput.StepID != "extract-interactions" {
+		t.Fatalf("interaction output step = %q, want ordered downstream step", interactionOutput.StepID)
+	}
+	registryRequest := client.requestFor(t, interactionextract.PromptID)
+	if got := string(registryRequest.Inputs["npcs"].Content); got != `{"npcs":[{"name":"Mira Thorn"}]}` {
+		t.Fatalf("downstream registry = %s, want one canonical names-only NPC", got)
+	}
+	manifestContent, err := json.Marshal(output.Manifest)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, forbidden := range []string{"npc:sha256:", "Mira Thorn enters.", "Mira Thorn, the Greencloak, waves."} {
+		if strings.Contains(string(manifestContent), forbidden) {
+			t.Fatalf("manifest leaked private identity or transcript content %q: %s", forbidden, manifestContent)
+		}
+	}
+}
+
 func TestNPCInteractionPipelineSkipsConsumerWhenNPCProducerIsRejected(t *testing.T) {
 	registries := productionNPCRegistries(t)
 	resolved := resolveNPCInteractionPipeline(t, registries)
@@ -168,6 +225,8 @@ func (client *npcInteractionLLMClient) CompleteStructured(ctx context.Context, r
 				map[string]any{"name": "Hooded Guard", "source_refs": []any{map[string]int{"start_unit_id": 3, "end_unit_id": 3}}},
 			}}
 		}
+	case npcnormalize.PromptID:
+		payload = map[string]any{"duplicate_groups": []any{}}
 	case interactionextract.PromptID:
 		payload = map[string]any{"interactions": []any{
 			map[string]any{"name": "Hooded Guard", "kind": "noncombat_presence", "source_refs": []any{map[string]int{"start_unit_id": 3, "end_unit_id": 3}}},
@@ -212,3 +271,59 @@ func (client *npcInteractionLLMClient) requestCount(promptID string) int {
 }

 var _ contracts.StructuredLLMClient = (*npcInteractionLLMClient)(nil)
+
+type semanticNPCInteractionClient struct {
+	requests []contracts.StructuredCompletionRequest
+	npcCalls int
+}
+
+func (client *semanticNPCInteractionClient) CompleteStructured(_ context.Context, request contracts.StructuredCompletionRequest, out any) (contracts.StructuredCompletionResponse, error) {
+	client.requests = append(client.requests, cloneStructuredCompletionRequest(request))
+	var payload any
+	switch request.PromptID {
+	case npcs.PromptID:
+		client.npcCalls++
+		name := "Mira Thorn"
+		if client.npcCalls == 2 {
+			name = "Mira Thorn, the Greencloak"
+		}
+		payload = map[string]any{"npcs": []any{map[string]any{"name": name, "source_refs": []any{map[string]int{"start_unit_id": client.npcCalls, "end_unit_id": client.npcCalls}}}}}
+	case npcnormalize.PromptID:
+		payload = map[string]any{"duplicate_groups": []any{map[string]any{"members": []string{"Mira Thorn", "Mira Thorn, the Greencloak"}, "canonical_name": "Mira Thorn"}}}
+	case interactionextract.PromptID:
+		payload = map[string]any{"interactions": []any{map[string]any{"name": "Mira Thorn", "kind": "dialogue", "source_refs": []any{map[string]int{"start_unit_id": 1, "end_unit_id": 1}}}}}
+	default:
+		return contracts.StructuredCompletionResponse{}, fmt.Errorf("unexpected semantic pipeline prompt %q", request.PromptID)
+	}
+	content, err := json.Marshal(payload)
+	if err != nil {
+		return contracts.StructuredCompletionResponse{}, err
+	}
+	if err := json.Unmarshal(content, out); err != nil {
+		return contracts.StructuredCompletionResponse{}, err
+	}
+	return contracts.StructuredCompletionResponse{Content: content}, nil
+}
+
+func (client *semanticNPCInteractionClient) requestCount(promptID string) int {
+	count := 0
+	for _, request := range client.requests {
+		if request.PromptID == promptID {
+			count++
+		}
+	}
+	return count
+}
+
+func (client *semanticNPCInteractionClient) requestFor(t *testing.T, promptID string) contracts.StructuredCompletionRequest {
+	t.Helper()
+	for _, request := range client.requests {
+		if request.PromptID == promptID {
+			return request
+		}
+	}
+	t.Fatalf("requests = %#v, missing %q", client.requests, promptID)
+	return contracts.StructuredCompletionRequest{}
+}
+
+var _ contracts.StructuredLLMClient = (*semanticNPCInteractionClient)(nil)
--- a/internal/modules/integration/dnd_npcs_runner_test.go
+++ b/internal/modules/integration/dnd_npcs_runner_test.go
@@ -14,6 +14,7 @@ import (
 	"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
 	npccodec "gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/codec/npcs"
 	"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/extract/npcs"
+	npcnormalize "gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/normalize/npcs"
 	"gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/npcs/identity"
 	dndregister "gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/register"
 	npcshape "gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/validate/npcs/shape"
@@ -95,7 +96,7 @@ func TestRunnerProcessesSeriatimInputWithProductionDNDNPCPipeline(t *testing.T)
 		t.Fatalf("manifest lane = %#v, want NPC production composition", lane)
 	}
 	normalizerMetadata, ok := lane.Metadata["normalizer"].(map[string]any)
-	if !ok || normalizerMetadata["identity_policy"] != identity.Policy || normalizerMetadata["normalization_policy"] != "dnd.npcs.normalize.v2" {
+	if !ok || normalizerMetadata["identity_policy"] != identity.Policy || normalizerMetadata["normalization_policy"] != npcnormalize.NormalizationPolicy || normalizerMetadata["prompt_id"] != npcnormalize.PromptID || normalizerMetadata["response_schema_id"] != npcnormalize.ResponseSchemaID {
 		t.Fatalf("normalizer metadata = %#v, want identity and normalization policies", lane.Metadata)
 	}
 	var npcOutputFile *contracts.OutputFile
@@ -108,8 +109,8 @@ func TestRunnerProcessesSeriatimInputWithProductionDNDNPCPipeline(t *testing.T)
 	if npcOutputFile == nil || npcOutputFile.ContentType != npccodec.MediaType {
 		t.Fatalf("output files = %#v, want JSON NPC lane file", output.OutputFiles)
 	}
-	if len(client.requests) != 1 || client.requests[0].PromptID != npcs.PromptID {
-		t.Fatalf("LLM requests = %#v, want one NPC prompt request", client.requests)
+	if len(client.requests) != 2 || client.requests[0].PromptID != npcs.PromptID || client.requests[1].PromptID != npcnormalize.PromptID {
+		t.Fatalf("LLM requests = %#v, want extraction and normalization prompt requests", client.requests)
 	}
 }

@@ -170,6 +171,52 @@ func TestProductionNPCPipelineRoutesSemanticCandidatesToDeterministicValidators(
 	}
 }

+func TestProductionNPCNormalizationRetryUsesFinalSafeProposal(t *testing.T) {
+	registries := productionNPCRegistries(t)
+	baseResponse := npcProductionResponse{NPCs: []npcProductionRecord{
+		{Name: "Mira", SourceRefs: []npcProductionSourceRef{{StartUnitID: 1, EndUnitID: 1}}},
+		{Name: "Mira Thorn", SourceRefs: []npcProductionSourceRef{{StartUnitID: 2, EndUnitID: 2}}},
+		{Name: "Hooded Guard", SourceRefs: []npcProductionSourceRef{{StartUnitID: 3, EndUnitID: 3}}},
+	}}
+	partial := []byte(`{"duplicate_groups":[{"members":["Mira","Mira Thorn"],"canonical_name":"Mira Thorn"},{"members":["Hooded Guard","Unknown"],"canonical_name":"Hooded Guard"}]}`)
+	safe := []byte(`{"duplicate_groups":[{"members":["Mira","Mira Thorn"],"canonical_name":"Mira Thorn"}]}`)
+
+	for _, test := range []struct {
+		name           string
+		retries        int
+		responses      [][]byte
+		wantCalls      int
+		wantExhaustion bool
+	}{
+		{name: "default retry budget", responses: [][]byte{partial}, wantCalls: 1, wantExhaustion: true},
+		{name: "later complete proposal", retries: 1, responses: [][]byte{partial, safe}, wantCalls: 2},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			cfg := loadNPCPipelineConfig(t)
+			profile := cfg.Pipelines["dnd-npcs-fixture"]
+			lane := profile.Artifacts["npcs"]
+			lane.Normalize.Retries = test.retries
+			profile.Artifacts["npcs"] = lane
+			cfg.Pipelines["dnd-npcs-fixture"] = profile
+			effective, err := cfg.Resolve(config.ResolveInput{PipelineID: "dnd-npcs-fixture", Catalog: moduleCatalog(registries)})
+			if err != nil {
+				t.Fatal(err)
+			}
+			client := &fakeNPCProductionLLMClient{response: baseResponse, normalizeResponses: test.responses}
+			output, err := runPreparedPipeline(t, registries, effective.ResolvedPipeline, client, pipeline.RunInput{RawInput: readNPCFixture(t)})
+			if err != nil {
+				t.Fatalf("Run() error = %v", err)
+			}
+			if client.requestCount(npcnormalize.PromptID) != test.wantCalls || len(output.NormalizeOutputs) != 1 {
+				t.Fatalf("normalization calls = %d output = %#v, want %d accepted result", client.requestCount(npcnormalize.PromptID), output.NormalizeOutputs, test.wantCalls)
+			}
+			if got := hasWarningReason(output.Warnings, npcnormalize.ReasonCodeNPCSemanticReconciliationExhausted); got != test.wantExhaustion {
+				t.Fatalf("warnings = %#v, exhaustion = %t, want %t", output.Warnings, got, test.wantExhaustion)
+			}
+		})
+	}
+}
+
 type npcProductionResponse struct {
 	NPCs []npcProductionRecord `json:"npcs"`
 }
@@ -185,26 +232,41 @@ type npcProductionSourceRef struct {
 }

 type fakeNPCProductionLLMClient struct {
-	response     npcProductionResponse
-	rawResponses [][]byte
-	requests     []contracts.StructuredCompletionRequest
+	response           npcProductionResponse
+	rawResponses       [][]byte
+	normalizeResponses [][]byte
+	requests           []contracts.StructuredCompletionRequest
 }

 func (client *fakeNPCProductionLLMClient) CompleteStructured(_ context.Context, req contracts.StructuredCompletionRequest, out any) (contracts.StructuredCompletionResponse, error) {
 	client.requests = append(client.requests, req)
 	var content []byte
-	if client.rawResponses != nil {
-		index := len(client.requests) - 1
-		if index >= len(client.rawResponses) {
-			return contracts.StructuredCompletionResponse{}, fmt.Errorf("missing fake NPC response %d", index)
+	switch req.PromptID {
+	case npcs.PromptID:
+		if client.rawResponses != nil {
+			index := client.requestCount(npcs.PromptID) - 1
+			if index >= len(client.rawResponses) {
+				return contracts.StructuredCompletionResponse{}, fmt.Errorf("missing fake NPC response %d", index)
+			}
+			content = append([]byte(nil), client.rawResponses[index]...)
+		} else {
+			var err error
+			content, err = json.Marshal(client.response)
+			if err != nil {
+				return contracts.StructuredCompletionResponse{}, err
+			}
 		}
-		content = append([]byte(nil), client.rawResponses[index]...)
-	} else {
-		var err error
-		content, err = json.Marshal(client.response)
-		if err != nil {
-			return contracts.StructuredCompletionResponse{}, err
+	case npcnormalize.PromptID:
+		content = []byte(`{"duplicate_groups":[]}`)
+		if client.normalizeResponses != nil {
+			index := client.requestCount(npcnormalize.PromptID) - 1
+			if index >= len(client.normalizeResponses) {
+				return contracts.StructuredCompletionResponse{}, fmt.Errorf("missing fake NPC normalization response %d", index)
+			}
+			content = append([]byte(nil), client.normalizeResponses[index]...)
 		}
+	default:
+		return contracts.StructuredCompletionResponse{}, fmt.Errorf("unexpected fake NPC prompt %q", req.PromptID)
 	}
 	if err := json.Unmarshal(content, out); err != nil {
 		return contracts.StructuredCompletionResponse{}, fmt.Errorf("populate NPC structured target: %w", err)
@@ -212,6 +274,16 @@ func (client *fakeNPCProductionLLMClient) CompleteStructured(_ context.Context,
 	return contracts.StructuredCompletionResponse{Content: content}, nil
 }

+func (client *fakeNPCProductionLLMClient) requestCount(promptID string) int {
+	count := 0
+	for _, request := range client.requests {
+		if request.PromptID == promptID {
+			count++
+		}
+	}
+	return count
+}
+
 func productionNPCRegistries(t *testing.T) pipeline.Registries {
 	t.Helper()
 	registries := pipeline.Registries{
@@ -283,3 +355,12 @@ func hasNPCWarning(warnings []contracts.Warning, reason string) bool {
 	}
 	return false
 }
+
+func hasWarningReason(warnings []contracts.Warning, reason string) bool {
+	for _, warning := range warnings {
+		if warning.ReasonCode == reason {
+			return true
+		}
+	}
+	return false
+}