package integration_test import ( "context" "encoding/json" "fmt" "os" "reflect" "strings" "sync" "testing" "gitea.maximumdirect.net/eric/notarius/internal/core/config" "gitea.maximumdirect.net/eric/notarius/internal/core/source" "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" interactioncodec "gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/codec/npcinteractions" 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) { registries := productionNPCRegistries(t) resolved := resolveNPCInteractionPipeline(t, registries) client := &npcInteractionLLMClient{} output, err := runPreparedPipeline(t, registries, resolved, client, pipeline.RunInput{RawInput: readNPCFixture(t)}) if err != nil { t.Fatalf("Run() error = %v", err) } if len(output.Rejected) != 0 || len(output.NormalizeOutputs) != 2 { t.Fatalf("run outputs = %#v rejected = %#v, want NPC and interaction artifacts", output.NormalizeOutputs, output.Rejected) } request := client.requestFor(t, interactionextract.PromptID) wantRegistry := `{"npcs":[{"name":"Mira Thorn"},{"name":"Hooded Guard"}]}` if got := string(request.Inputs["npcs"].Content); got != wantRegistry { t.Fatalf("interaction registry input = %s, want names-only projection %s", got, wantRegistry) } if request.Inputs["npcs"].MediaType != npccodec.MediaType { t.Fatalf("interaction registry media type = %q, want %q", request.Inputs["npcs"].MediaType, npccodec.MediaType) } serialized := normalizedLane(t, output, "interactions") if serialized.Artifact.Schema.ID != interactioncodec.SchemaID || serialized.Artifact.Schema.Version != interactioncodec.SchemaVersion { t.Fatalf("interaction artifact schema = %#v", serialized.Artifact.Schema) } interactions, err := interactioncodec.New().Decode(serialized.Artifact.Content) if err != nil { t.Fatalf("Decode(interaction output) error = %v", err) } if len(interactions.Interactions) != 2 { t.Fatalf("interactions = %#v, want two occurrences", interactions) } first, second := interactions.Interactions[0], interactions.Interactions[1] if first.Name != "Mira Thorn" || string(first.Kind) != "dialogue" || second.Name != "Hooded Guard" || string(second.Kind) != "noncombat_presence" { t.Fatalf("interactions = %#v, want canonical names, kinds, and source chronology", interactions) } assertInteractionEvidence(t, first.SourceRefs) assertInteractionEvidence(t, second.SourceRefs) if first.SourceRefs[0].StartUnitID >= second.SourceRefs[0].StartUnitID { t.Fatalf("interaction chronology = %#v, want source order", interactions.Interactions) } var durable map[string]json.RawMessage if err := json.Unmarshal(serialized.Artifact.Content, &durable); err != nil { t.Fatalf("unmarshal durable interaction payload: %v", err) } if len(durable) != 1 || durable["interactions"] == nil { t.Fatalf("durable interaction payload = %#v, want only interactions", durable) } } 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) client := &npcInteractionLLMClient{rejectNPCs: true} output, err := runPreparedPipeline(t, registries, resolved, client, pipeline.RunInput{RawInput: readNPCFixture(t)}) if err == nil { t.Fatalf("Run() output = %#v, want missing generated NPC producer error", output) } if client.requestCount(npcs.PromptID) != 1 { t.Fatalf("NPC requests = %d, want rejected producer", client.requestCount(npcs.PromptID)) } if client.requestCount(interactionextract.PromptID) != 0 { t.Fatalf("interaction requests = %d, want none after rejected producer", client.requestCount(interactionextract.PromptID)) } } func resolveNPCInteractionPipeline(t *testing.T, registries pipeline.Registries) pipeline.ResolvedPipeline { t.Helper() configValue := loadNPCInteractionPipelineConfig(t) effective, err := configValue.Resolve(config.ResolveInput{PipelineID: "dnd-npc-interactions-fixture", Catalog: moduleCatalog(registries)}) if err != nil { t.Fatalf("Resolve() error = %v", 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) } return resolved } func loadNPCInteractionPipelineConfig(t *testing.T) config.Config { t.Helper() data, err := os.ReadFile("testdata/dnd_npc_interactions_pipeline.yml") if err != nil { t.Fatalf("ReadFile(dnd_npc_interactions_pipeline.yml) error = %v", err) } fileConfig, err := config.ParseFileConfigYAML(data) if err != nil { t.Fatalf("ParseFileConfigYAML() error = %v", err) } result := config.Default() if err := result.ApplyFileConfig(fileConfig); err != nil { t.Fatalf("ApplyFileConfig() error = %v", err) } return result } func normalizedLane(t *testing.T, output pipeline.RunOutput, laneID string) contracts.SerializedOutput { t.Helper() for _, serialized := range output.NormalizeOutputs { if serialized.LaneID == laneID { return serialized } } t.Fatalf("normalized lanes = %#v, missing %q", output.NormalizeOutputs, laneID) return contracts.SerializedOutput{} } func assertInteractionEvidence(t *testing.T, references []source.SourceRef) { t.Helper() if len(references) == 0 { t.Fatal("interaction has no current-source evidence") } for _, reference := range references { if reference.SourceID != "npc-session" { t.Fatalf("interaction evidence = %#v, want current source only", reference) } } } type npcInteractionLLMClient struct { mu sync.Mutex requests []contracts.StructuredCompletionRequest rejectNPCs bool } func (client *npcInteractionLLMClient) CompleteStructured(ctx context.Context, request contracts.StructuredCompletionRequest, out any) (contracts.StructuredCompletionResponse, error) { if err := ctx.Err(); err != nil { return contracts.StructuredCompletionResponse{}, err } client.mu.Lock() client.requests = append(client.requests, cloneStructuredCompletionRequest(request)) client.mu.Unlock() var payload any switch request.PromptID { case npcs.PromptID: if client.rejectNPCs { payload = map[string]any{"npcs": []any{map[string]any{ "name": "", "source_refs": []any{map[string]int{"start_unit_id": 1, "end_unit_id": 1}}, }}} } else { payload = map[string]any{"npcs": []any{ map[string]any{"name": "Mira Thorn", "source_refs": []any{map[string]int{"start_unit_id": 1, "end_unit_id": 1}}}, 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}}}, map[string]any{"name": " mira thorn ", "kind": "dialogue", "source_refs": []any{map[string]int{"start_unit_id": 2, "end_unit_id": 2}}}, }} default: return contracts.StructuredCompletionResponse{}, fmt.Errorf("unexpected interaction 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{}, fmt.Errorf("populate interaction response: %w", err) } return contracts.StructuredCompletionResponse{Content: content, Provider: "test", Model: "interaction-fake"}, nil } func (client *npcInteractionLLMClient) requestFor(t *testing.T, promptID string) contracts.StructuredCompletionRequest { t.Helper() client.mu.Lock() defer client.mu.Unlock() for _, request := range client.requests { if request.PromptID == promptID { return request } } t.Fatalf("requests = %#v, missing prompt %q", client.requests, promptID) return contracts.StructuredCompletionRequest{} } func (client *npcInteractionLLMClient) requestCount(promptID string) int { client.mu.Lock() defer client.mu.Unlock() count := 0 for _, request := range client.requests { if request.PromptID == promptID { count++ } } return count } 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)