diff --git a/examples/dnd-npc-grounded.config.yml b/examples/dnd-npc-grounded.config.yml new file mode 100644 index 0000000..f95cb7c --- /dev/null +++ b/examples/dnd-npc-grounded.config.yml @@ -0,0 +1,37 @@ +version: 3 +output: + directory: ./notarius-output +cache: + chunk_plans: + mode: bypass + checkpoints: + enabled: false + directory: "" +debug: + directory: ./notarius-debug +pipelines: + dnd-npc-grounded: + input: seriatim + steps: + - id: identify-npcs + artifacts: + npcs: + extract: + module: dnd/npcs + retries: 2 + normalize: dnd/npcs + - id: grounded-events + references: + npcs: + artifact: + step: identify-npcs + lane: npcs + artifacts: + spells: + extract: dnd/spells + normalize: dnd/spells + combat: + extract: + module: dnd/combat-turns + retries: 2 + normalize: dnd/combat-turns diff --git a/internal/cli/production_contract_test.go b/internal/cli/production_contract_test.go index 70d3aef..29c98c2 100644 --- a/internal/cli/production_contract_test.go +++ b/internal/cli/production_contract_test.go @@ -507,6 +507,7 @@ func maintainedExampleFiles(t *testing.T) []maintainedExample { {name: "sequential", path: repositoryPath("examples", "dnd-npc-spell-sequential.config.yml"), pipelineIDs: []string{"dnd-npcs", "dnd-spells"}}, {name: "combat", path: repositoryPath("examples", "dnd-combat-turns.config.yml"), pipelineIDs: []string{"dnd-combat"}}, {name: "npc-combat-sequential", path: repositoryPath("examples", "dnd-npc-combat-sequential.config.yml"), pipelineIDs: []string{"dnd-combat", "dnd-npcs"}}, + {name: "npc-grounded", path: repositoryPath("examples", "dnd-npc-grounded.config.yml"), pipelineIDs: []string{"dnd-npc-grounded"}}, } } diff --git a/internal/modules/dnd/extract/combatturns/extractor.go b/internal/modules/dnd/extract/combatturns/extractor.go index 80d9bbd..2502272 100644 --- a/internal/modules/dnd/extract/combatturns/extractor.go +++ b/internal/modules/dnd/extract/combatturns/extractor.go @@ -60,7 +60,7 @@ type Options struct{} type Extractor struct { llm contracts.StructuredLLMClient - npcRegistry *npcregistry.Registry + npcResolver *npcregistry.Resolver promptSHA string responseSchemaSHA string } @@ -76,7 +76,7 @@ func New(llmClient contracts.StructuredLLMClient, _ Options, references ...contr if len(references) == 1 { referenceSet = references[0] } - npcRegistry, err := npcregistry.Resolve(referenceSet) + npcResolver, err := npcregistry.NewResolver(referenceSet) if err != nil { return nil, extractorErrorf("prepare NPC registry prompt input: %w", err) } @@ -90,7 +90,7 @@ func New(llmClient contracts.StructuredLLMClient, _ Options, references ...contr } return &Extractor{ llm: llmClient, - npcRegistry: npcRegistry, + npcResolver: npcResolver, promptSHA: promptSHA, responseSchemaSHA: responseSchema.SHA256, }, nil @@ -115,9 +115,10 @@ func (e *Extractor) ManifestMetadata() map[string]any { "response_schema_version": SchemaVersion, "response_schema_sha256": e.responseSchemaSHA, } - if e.npcRegistry.Bound() { - metadata["npc_registry_digest"] = e.npcRegistry.Digest() - metadata["npc_count"] = e.npcRegistry.Count() + seeded := e.npcResolver.Seeded() + if seeded.Bound() { + metadata["npc_registry_digest"] = seeded.Digest() + metadata["npc_count"] = seeded.Count() } return metadata } @@ -131,8 +132,9 @@ func (e *Extractor) CheckpointFingerprints() []pipeline.CheckpointFingerprint { {Name: "response_schema", Value: e.responseSchemaSHA}, {Name: "mapping_policy", Value: mappingPolicy}, } - if e.npcRegistry.Bound() { - fingerprints = append(fingerprints, pipeline.CheckpointFingerprint{Name: "npc_registry", Value: e.npcRegistry.Digest()}) + seeded := e.npcResolver.Seeded() + if seeded.Bound() { + fingerprints = append(fingerprints, pipeline.CheckpointFingerprint{Name: "npc_registry", Value: seeded.Digest()}) } return fingerprints } @@ -163,10 +165,14 @@ func (e *Extractor) Extract(ctx context.Context, req contracts.TypedExtractionRe if err != nil { return contracts.TypedExtractionResult[dnd.CombatTurnList]{}, extractorErrorf("%w", err) } + npcRegistry, err := e.npcResolver.Resolve(req.References) + if err != nil { + return contracts.TypedExtractionResult[dnd.CombatTurnList]{}, extractorErrorf("resolve NPC registry: %w", err) + } var response extractionResponse inputs := shared.PromptInputs(sourceInput, req.References) - inputs[NPCRegistryReferenceSlot] = e.npcRegistry.PromptInput() + inputs[NPCRegistryReferenceSlot] = npcRegistry.PromptInput() if _, err := e.llm.CompleteStructured(ctx, contracts.StructuredCompletionRequest{ StageName: Key, PromptID: PromptID, diff --git a/internal/modules/dnd/extract/combatturns/extractor_test.go b/internal/modules/dnd/extract/combatturns/extractor_test.go index 6898277..73668c2 100644 --- a/internal/modules/dnd/extract/combatturns/extractor_test.go +++ b/internal/modules/dnd/extract/combatturns/extractor_test.go @@ -160,6 +160,29 @@ func TestExtractUnboundRegistryUsesExactEmptyPromptAndOmitsIdentity(t *testing.T } } +func TestExtractorResolvesOperationNPCOverrideWithoutSingletonMetadata(t *testing.T) { + content := npcRegistryJSON(t) + client := &fakeCombatTurnsLLMClient{response: extractionResponse{CombatTurns: []combatTurnResponse{}}} + extractor := newExtractor(t, client) + request := extractionRequest() + request.References = contracts.ReferenceSet{Slots: map[string]contracts.ResolvedReferenceSlot{ + NPCRegistryReferenceSlot: { + Slot: contracts.ReferenceSlot{Name: NPCRegistryReferenceSlot}, + Items: []contracts.ReferenceItem{{SlotName: NPCRegistryReferenceSlot, MediaType: npccodec.MediaType, Content: append([]byte(nil), content...), Origin: contracts.ReferenceOrigin{Type: "generated"}}}, + }, + }} + if _, err := extractor.Extract(context.Background(), request); err != nil { + t.Fatalf("Extract() error = %v", err) + } + input := client.requests[0].Inputs[NPCRegistryReferenceSlot] + if input.Digest == "" || string(input.Content) != string(content) || input.OriginURI != "" { + t.Fatalf("operation NPC input = %#v, want generated canonical grounding without provenance", input) + } + if metadata := extractor.ManifestMetadata(); metadata["npc_registry_digest"] != nil || metadata["npc_count"] != nil { + t.Fatalf("singleton metadata = %#v, want no operation-varying NPC identity", metadata) + } +} + func TestNewRejectsMalformedNPCRegistryBeforeLLMCallWithoutContent(t *testing.T) { client := &fakeCombatTurnsLLMClient{} references := contracts.ReferenceSet{Slots: map[string]contracts.ResolvedReferenceSlot{ diff --git a/internal/modules/dnd/extract/spells/extractor.go b/internal/modules/dnd/extract/spells/extractor.go index 483cdea..a5a2773 100644 --- a/internal/modules/dnd/extract/spells/extractor.go +++ b/internal/modules/dnd/extract/spells/extractor.go @@ -65,7 +65,7 @@ type Extractor struct { llm contracts.StructuredLLMClient effectiveCatalog spellcatalog.EffectiveCatalog catalogPromptInput contracts.LLMInputMaterial - npcRegistry *npcregistry.Registry + npcResolver *npcregistry.Resolver promptSHA string responseSchemaSHA string } @@ -89,7 +89,7 @@ func New(llmClient contracts.StructuredLLMClient, _ Options, references ...contr if err != nil { return nil, extractorErrorf("prepare spell catalog prompt input: %w", err) } - npcRegistry, err := npcregistry.Resolve(referenceSet) + npcResolver, err := npcregistry.NewResolver(referenceSet) if err != nil { return nil, extractorErrorf("prepare NPC registry prompt input: %w", err) } @@ -105,7 +105,7 @@ func New(llmClient contracts.StructuredLLMClient, _ Options, references ...contr llm: llmClient, effectiveCatalog: effectiveCatalog, catalogPromptInput: catalogPromptInput, - npcRegistry: npcRegistry, + npcResolver: npcResolver, promptSHA: promptSHA, responseSchemaSHA: responseSchema.SHA256, }, nil @@ -133,9 +133,10 @@ func (e *Extractor) ManifestMetadata() map[string]any { "response_schema_version": SchemaVersion, "response_schema_sha256": e.responseSchemaSHA, } - if e.npcRegistry.Bound() { - metadata["npc_registry_digest"] = e.npcRegistry.Digest() - metadata["npc_count"] = e.npcRegistry.Count() + seeded := e.npcResolver.Seeded() + if seeded.Bound() { + metadata["npc_registry_digest"] = seeded.Digest() + metadata["npc_count"] = seeded.Count() } return metadata } @@ -149,8 +150,9 @@ func (e *Extractor) CheckpointFingerprints() []pipeline.CheckpointFingerprint { {Name: "prompt", Value: e.promptSHA}, {Name: "response_schema", Value: e.responseSchemaSHA}, } - if e.npcRegistry.Bound() { - fingerprints = append(fingerprints, pipeline.CheckpointFingerprint{Name: "npc_registry", Value: e.npcRegistry.Digest()}) + seeded := e.npcResolver.Seeded() + if seeded.Bound() { + fingerprints = append(fingerprints, pipeline.CheckpointFingerprint{Name: "npc_registry", Value: seeded.Digest()}) } return fingerprints } @@ -181,11 +183,15 @@ func (e *Extractor) Extract(ctx context.Context, req contracts.TypedExtractionRe if err != nil { return contracts.TypedExtractionResult[dnd.SpellList]{}, extractorErrorf("%w", err) } + npcRegistry, err := e.npcResolver.Resolve(req.References) + if err != nil { + return contracts.TypedExtractionResult[dnd.SpellList]{}, extractorErrorf("resolve NPC registry: %w", err) + } var response extractionResponse inputs := shared.PromptInputs(sourceInput, req.References) inputs[spellcatalog.SpellCatalogReferenceSlot] = e.catalogPromptInput.Clone() - inputs[NPCRegistryReferenceSlot] = e.npcRegistry.PromptInput() + inputs[NPCRegistryReferenceSlot] = npcRegistry.PromptInput() if _, err := e.llm.CompleteStructured(ctx, contracts.StructuredCompletionRequest{ StageName: Key, PromptID: PromptID, diff --git a/internal/modules/dnd/extract/spells/npc_registry_wiring_test.go b/internal/modules/dnd/extract/spells/npc_registry_wiring_test.go index e1dc967..a6f8119 100644 --- a/internal/modules/dnd/extract/spells/npc_registry_wiring_test.go +++ b/internal/modules/dnd/extract/spells/npc_registry_wiring_test.go @@ -84,6 +84,32 @@ func TestSpellExtractorPreservesSemanticNPCRegistryFingerprintAndPromptWiring(t } } +func TestSpellExtractorResolvesOperationNPCOverrideWithoutSingletonMetadata(t *testing.T) { + canonical, err := npccodec.New().Encode(registryFixture()) + if err != nil { + t.Fatalf("encode NPC registry: %v", err) + } + client := &fakeSpellsLLMClient{response: extractionResponse{SpellCasts: []spellCastResponse{}}} + extractor := newExtractor(t, client) + request := extractionRequest() + request.References = spellNPCRegistryReference(canonical, "file:///generated.json") + if _, err := extractor.Extract(context.Background(), request); err != nil { + t.Fatalf("Extract() error = %v", err) + } + input := client.requests[0].Inputs[NPCRegistryReferenceSlot] + if input.Digest == "" || string(input.Content) != string(canonical) || input.OriginURI != "" { + t.Fatalf("operation NPC input = %#v, want generated canonical grounding without provenance", input) + } + if metadata := extractor.ManifestMetadata(); metadata["npc_registry_digest"] != nil || metadata["npc_count"] != nil { + t.Fatalf("singleton metadata = %#v, want no operation-varying NPC identity", metadata) + } + for _, fingerprint := range extractor.CheckpointFingerprints() { + if fingerprint.Name == "npc_registry" { + t.Fatalf("singleton fingerprints = %#v, want no operation-varying NPC identity", extractor.CheckpointFingerprints()) + } + } +} + func registryFixture() dnd.NPCList { return dnd.NPCList{NPCs: []dnd.NPC{{ ID: identity.DeriveID("Mira Thorn"), diff --git a/internal/modules/dnd/normalize/combatturns/normalizer.go b/internal/modules/dnd/normalize/combatturns/normalizer.go index 512ec5b..7e452f1 100644 --- a/internal/modules/dnd/normalize/combatturns/normalizer.go +++ b/internal/modules/dnd/normalize/combatturns/normalizer.go @@ -45,7 +45,7 @@ var _ pipeline.CheckpointFingerprintProvider = (*Normalizer)(nil) type Options struct{} type Normalizer struct { - npcRegistry *npcregistry.Registry + npcResolver *npcregistry.Resolver } func New(_ Options, references ...contracts.ReferenceSet) (*Normalizer, error) { @@ -56,11 +56,11 @@ func New(_ Options, references ...contracts.ReferenceSet) (*Normalizer, error) { if len(references) == 1 { referenceSet = references[0] } - npcRegistry, err := npcregistry.Resolve(referenceSet) + npcResolver, err := npcregistry.NewResolver(referenceSet) if err != nil { return nil, normalizerErrorf("prepare NPC registry: %w", err) } - return &Normalizer{npcRegistry: npcRegistry}, nil + return &Normalizer{npcResolver: npcResolver}, nil } func (n *Normalizer) Key() string { return Key } @@ -75,9 +75,10 @@ func (n *Normalizer) ManifestMetadata() map[string]any { "normalization_policy": normalizationPolicy, "identity_policy": identity.Policy, } - if n.npcRegistry.Bound() { - metadata["npc_registry_digest"] = n.npcRegistry.Digest() - metadata["npc_count"] = n.npcRegistry.Count() + seeded := n.npcResolver.Seeded() + if seeded.Bound() { + metadata["npc_registry_digest"] = seeded.Digest() + metadata["npc_count"] = seeded.Count() } return metadata } @@ -90,8 +91,9 @@ func (n *Normalizer) CheckpointFingerprints() []pipeline.CheckpointFingerprint { {Name: "normalization_policy", Value: normalizationPolicy}, {Name: "identity_policy", Value: identity.Policy}, } - if n.npcRegistry.Bound() { - fingerprints = append(fingerprints, pipeline.CheckpointFingerprint{Name: "npc_registry", Value: n.npcRegistry.Digest()}) + seeded := n.npcResolver.Seeded() + if seeded.Bound() { + fingerprints = append(fingerprints, pipeline.CheckpointFingerprint{Name: "npc_registry", Value: seeded.Digest()}) } return fingerprints } @@ -107,7 +109,11 @@ func (n *Normalizer) Normalize(ctx context.Context, req contracts.TypedNormalize return contracts.TypedNormalizeResult[dnd.CombatTurnList]{}, normalizerErrorf("context error before normalize: %w", err) } - value, warnings := normalizeList(req.MergeOutput.Value, req.Source, n.npcRegistry) + npcRegistry, err := n.npcResolver.Resolve(req.References) + if err != nil { + return contracts.TypedNormalizeResult[dnd.CombatTurnList]{}, normalizerErrorf("resolve NPC registry: %w", err) + } + value, warnings := normalizeList(req.MergeOutput.Value, req.Source, npcRegistry) return contracts.TypedNormalizeResult[dnd.CombatTurnList]{Value: value, Warnings: warnings}, nil } diff --git a/internal/modules/dnd/normalize/combatturns/normalizer_test.go b/internal/modules/dnd/normalize/combatturns/normalizer_test.go index 6030d1d..5230267 100644 --- a/internal/modules/dnd/normalize/combatturns/normalizer_test.go +++ b/internal/modules/dnd/normalize/combatturns/normalizer_test.go @@ -70,6 +70,36 @@ func TestNormalizeCanonicalizesFieldsAndRegistryIdentities(t *testing.T) { } } +func TestNormalizeResolvesOperationNPCOverrideWithoutSingletonMetadata(t *testing.T) { + doc := testDocument() + normalizer, err := New(Options{}) + if err != nil { + t.Fatalf("New() error = %v", err) + } + input := dnd.CombatTurnList{CombatTurns: []dnd.CombatTurn{{ + Actor: "Storm", + TurnKind: dnd.CombatTurnKindTurn, + Actions: []dnd.CombatAction{{Category: dnd.CombatActionCategoryAttack, Declaration: "attacks", Targets: []string{"Minion"}}}, + Summary: "Storm attacks", + SourceRefs: []source.SourceRef{{SourceID: doc.ID, StartUnitID: 10, EndUnitID: 10}}, + }}} + result, err := normalizer.Normalize(context.Background(), contracts.TypedNormalizeRequest[dnd.CombatTurnList]{ + Source: doc, + MergeOutput: contracts.MergeArtifact[dnd.CombatTurnList]{Value: input}, + References: npcReferences(t), + }) + if err != nil { + t.Fatalf("Normalize() error = %v", err) + } + turn := result.Value.CombatTurns[0] + if turn.Actor != "Aria" || turn.Actions[0].Targets[0] != "Goblin" { + t.Fatalf("operation-normalized turn = %#v, want Aria/Goblin", turn) + } + if metadata := normalizer.ManifestMetadata(); metadata["npc_registry_digest"] != nil || metadata["npc_count"] != nil { + t.Fatalf("singleton metadata = %#v, want no operation-varying NPC identity", metadata) + } +} + func TestNormalizeOrdersBySourcePositionAndCollapsesExactDuplicates(t *testing.T) { doc := &source.SourceDocument{ID: "session", Units: []source.SourceUnit{{ID: 50}, {ID: 10}, {ID: 90}}} first := validTurn("Aria", source.SourceRef{SourceID: doc.ID, StartUnitID: 50, EndUnitID: 50}) diff --git a/internal/modules/dnd/npcs/registry/registry.go b/internal/modules/dnd/npcs/registry/registry.go index 96ff62c..307f659 100644 --- a/internal/modules/dnd/npcs/registry/registry.go +++ b/internal/modules/dnd/npcs/registry/registry.go @@ -8,6 +8,7 @@ import ( "fmt" "mime" "strings" + "sync" "gitea.maximumdirect.net/eric/notarius/internal/core/source" "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" @@ -34,6 +35,106 @@ type Registry struct { lookupByKey map[string]int } +// Resolver retains only the validated construction-time registry and immutable +// canonical registries keyed by their semantic digest. Operation references +// are resolved on demand; caller-owned reference bytes are never retained. +type Resolver struct { + seeded *Registry + + mu sync.Mutex + cache map[string]*Registry + rawCache map[string]*Registry +} + +// NewResolver validates the optional construction-time NPC reference and +// prepares the operation-time registry cache. A malformed static reference +// therefore fails before any operation starts. +func NewResolver(references contracts.ReferenceSet) (*Resolver, error) { + seeded, err := Resolve(constructionReferences(references)) + if err != nil { + return nil, err + } + return &Resolver{seeded: seeded, cache: make(map[string]*Registry), rawCache: make(map[string]*Registry)}, nil +} + +func constructionReferences(references contracts.ReferenceSet) contracts.ReferenceSet { + slot, ok := references.Slots[ReferenceSlot] + if !ok || len(slot.Items) > 0 { + return references + } + cloned := contracts.ReferenceSet{Slots: make(map[string]contracts.ResolvedReferenceSlot, len(references.Slots))} + for name, value := range references.Slots { + cloned.Slots[name] = value + } + delete(cloned.Slots, ReferenceSlot) + return cloned +} + +// Seeded returns the immutable construction-time registry. Its accessors are +// defensive, so callers may safely use the returned view for static metadata. +func (r *Resolver) Seeded() *Registry { + if r == nil { + return nil + } + return r.seeded +} + +// Resolve returns the effective registry for one operation. An operation +// without an NPC item uses the construction-time registry. A canonical item +// matching that registry reuses it; other canonical registries are cached by +// digest for concurrent chunk operations. +func (r *Resolver) Resolve(references contracts.ReferenceSet) (*Registry, error) { + if r == nil { + return Resolve(references) + } + if _, ok := references.Slots[ReferenceSlot]; !ok { + return r.seeded, nil + } + + slot := references.Slots[ReferenceSlot] + rawKey := "" + if len(slot.Items) == 1 { + rawKey = strings.ToLower(strings.TrimSpace(slot.Items[0].MediaType)) + "\x00" + semanticDigest(slot.Items[0].Content) + } + + r.mu.Lock() + defer r.mu.Unlock() + if rawKey != "" { + if cached, ok := r.rawCache[rawKey]; ok { + return cached, nil + } + } + resolved, err := Resolve(references) + if err != nil { + return nil, err + } + if sameRegistryIdentity(r.seeded, resolved) { + if rawKey != "" { + r.rawCache[rawKey] = r.seeded + } + return r.seeded, nil + } + + if cached, ok := r.cache[resolved.Digest()]; ok { + if rawKey != "" { + r.rawCache[rawKey] = cached + } + return cached, nil + } + r.cache[resolved.Digest()] = resolved + if rawKey != "" { + r.rawCache[rawKey] = resolved + } + return resolved, nil +} + +func sameRegistryIdentity(first, second *Registry) bool { + if first == nil || second == nil { + return first == second + } + return first.bound == second.bound && first.digest == second.digest +} + // Resolve prepares the optional NPC registry reference. An absent slot // produces the exact empty prompt input and no semantic registry identity. func Resolve(references contracts.ReferenceSet) (*Registry, error) { diff --git a/internal/modules/dnd/npcs/registry/registry_test.go b/internal/modules/dnd/npcs/registry/registry_test.go index 77226ef..5880b04 100644 --- a/internal/modules/dnd/npcs/registry/registry_test.go +++ b/internal/modules/dnd/npcs/registry/registry_test.go @@ -5,6 +5,7 @@ import ( "encoding/json" "fmt" "strings" + "sync" "testing" "unicode/utf8" @@ -168,6 +169,95 @@ func TestRegistryAccessorsAndLookupAreDefensive(t *testing.T) { } } +func TestResolverUsesSeededFallbackAndCachesGeneratedCanonicalRegistry(t *testing.T) { + staticContent := encodeRegistry(t, validRegistryList()) + resolver, err := NewResolver(registryReference(staticContent, "file:///static.json")) + if err != nil { + t.Fatalf("NewResolver() error = %v", err) + } + if got, err := resolver.Resolve(contracts.ReferenceSet{}); err != nil || got != resolver.Seeded() { + t.Fatalf("Resolve(absent) = %p, %v, want seeded %p", got, err, resolver.Seeded()) + } + + generated := registryReference(append([]byte("\n"), staticContent...), "file:///generated.json") + first, err := resolver.Resolve(generated) + if err != nil { + t.Fatalf("Resolve(generated) error = %v", err) + } + second, err := resolver.Resolve(generated) + if err != nil { + t.Fatalf("Resolve(generated second) error = %v", err) + } + if first != resolver.Seeded() || second != first { + t.Fatalf("resolved registries = %p, %p, seeded %p; want seeded reuse", first, second, resolver.Seeded()) + } + + changed := validRegistryList() + changed.NPCs[0].Description = "A changed generated description." + changedContent := encodeRegistry(t, changed) + changedReferences := registryReference(changedContent, "file:///changed.json") + resolved, err := resolver.Resolve(changedReferences) + if err != nil { + t.Fatalf("Resolve(changed) error = %v", err) + } + changedReferences.Slots[ReferenceSlot].Items[0].Content[0] = 'X' + if resolved == first || string(resolved.CanonicalBytes()) != string(changedContent) { + t.Fatalf("changed registry = %p/%s, want independent canonical cache entry", resolved, resolved.CanonicalBytes()) + } +} + +func TestResolverSharesOneCachedRegistryAcrossConcurrentOperations(t *testing.T) { + content := encodeRegistry(t, validRegistryList()) + resolver, err := NewResolver(contracts.ReferenceSet{}) + if err != nil { + t.Fatalf("NewResolver() error = %v", err) + } + references := registryReference(content, "file:///generated.json") + const callers = 32 + results := make(chan *Registry, callers) + errors := make(chan error, callers) + var wait sync.WaitGroup + for index := 0; index < callers; index++ { + wait.Add(1) + go func() { + defer wait.Done() + resolved, resolveErr := resolver.Resolve(references) + if resolveErr != nil { + errors <- resolveErr + return + } + results <- resolved + }() + } + wait.Wait() + close(results) + close(errors) + for err := range errors { + t.Fatalf("concurrent Resolve() error = %v", err) + } + var first *Registry + for resolved := range results { + if first == nil { + first = resolved + } else if resolved != first { + t.Fatalf("concurrent resolved registry %p differs from cached %p", resolved, first) + } + } +} + +func TestNewResolverAllowsGeneratedDeclarationButRejectsMalformedStaticItem(t *testing.T) { + placeholder := contracts.ReferenceSet{Slots: map[string]contracts.ResolvedReferenceSlot{ + ReferenceSlot: {Slot: contracts.ReferenceSlot{Name: ReferenceSlot, AcceptedArtifactKinds: []contracts.ArtifactKind{dnd.NPCListKind}}}, + }} + if _, err := NewResolver(placeholder); err != nil { + t.Fatalf("NewResolver(generated declaration) error = %v, want nil", err) + } + malformed := registryReference([]byte(`{"npcs":[`), "file:///runtime.json") + if _, err := NewResolver(malformed); err == nil || !strings.Contains(err.Error(), "invalid approved NPC JSON") { + t.Fatalf("NewResolver(malformed) error = %v, want bounded decode failure", err) + } +} + func validRegistryList() dnd.NPCList { return dnd.NPCList{NPCs: []dnd.NPC{{ ID: identity.DeriveID("Mira Thorn"), diff --git a/internal/modules/integration/dnd_npc_grounded_test.go b/internal/modules/integration/dnd_npc_grounded_test.go new file mode 100644 index 0000000..0ded177 --- /dev/null +++ b/internal/modules/integration/dnd_npc_grounded_test.go @@ -0,0 +1,222 @@ +package integration_test + +import ( + "context" + "crypto/sha256" + "encoding/json" + "fmt" + "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" + "gitea.maximumdirect.net/eric/notarius/internal/modules/dnd" + combatcodec "gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/codec/combatturns" + npccodec "gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/codec/npcs" + combatextract "gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/extract/combatturns" + "gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/extract/npcs" + "gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/extract/spells" + "gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/shared" +) + +func TestNPCOutputGroundsSpellAndCombatConsumersThroughOneOperation(t *testing.T) { + registries := productionNPCRegistries(t) + catalog := moduleCatalog(registries) + configValue := loadGroundedPipelineConfig(t) + effective, err := configValue.Resolve(config.ResolveInput{PipelineID: "dnd-npc-grounded", Catalog: catalog}) + if err != nil { + t.Fatalf("Resolve() error = %v", err) + } + materialized, warnings, err := pipeline.MaterializeReferences(effective.ResolvedPipeline, catalog, pipeline.ReferenceMaterializationOptions{}) + if err != nil || len(warnings) != 0 { + t.Fatalf("MaterializeReferences() error = %v warnings = %#v", err, warnings) + } + + client := &groundedDNDLLMClient{} + output, err := runPreparedPipeline(t, registries, materialized, client, pipeline.RunInput{ + RawInput: readNPCFixture(t), + ExtractWorkers: 1, + }) + if err != nil { + t.Fatalf("Run() error = %v", err) + } + if len(output.Rejected) != 0 || len(output.NormalizeOutputs) != 3 { + t.Fatalf("run outputs = %#v rejected = %#v, want NPC, spell, and combat outputs", output.NormalizeOutputs, output.Rejected) + } + + var npcPayload []byte + for _, serialized := range output.NormalizeOutputs { + if serialized.LaneID == "npcs" { + npcPayload = append([]byte(nil), serialized.Artifact.Content...) + break + } + } + if len(npcPayload) == 0 { + t.Fatal("NPC producer did not publish a canonical payload") + } + npcValue, err := npccodec.New().Decode(npcPayload) + if err != nil { + t.Fatalf("decode NPC producer payload: %v", err) + } + npcPayload, err = npccodec.New().Encode(npcValue) + if err != nil { + t.Fatalf("encode canonical NPC producer payload: %v", err) + } + canonicalDigest := digestBytes(npcPayload) + seenConsumers := map[string]bool{} + for _, request := range client.requestsSnapshot() { + if request.PromptID != spells.PromptID && request.PromptID != combatextract.PromptID { + continue + } + input := request.Inputs["npcs"] + if input.MediaType != npccodec.MediaType || input.Digest != canonicalDigest || string(input.Content) != string(npcPayload) || input.OriginURI != "" { + t.Fatalf("%s NPC prompt input = %#v, want canonical generated registry without provenance", request.PromptID, input) + } + seenConsumers[request.PromptID] = true + } + if !seenConsumers[spells.PromptID] || !seenConsumers[combatextract.PromptID] { + t.Fatalf("consumer prompt IDs = %#v, want spell and combat requests", seenConsumers) + } + + provenanceCount := 0 + for _, reference := range output.Manifest.References { + if reference.SlotName != "npcs" { + continue + } + provenanceCount++ + if reference.Digest != canonicalDigest || reference.OriginType != "generated" { + t.Fatalf("NPC generated provenance = %#v, want canonical digest and generated origin", reference) + } + } + if provenanceCount != 3 { + t.Fatalf("NPC generated provenance count = %d, want spell extract plus combat extract/normalize", provenanceCount) + } + for _, lane := range output.Manifest.ArtifactLanes { + for _, component := range []string{"extractor", "normalizer"} { + metadata, ok := lane.Metadata[component].(map[string]any) + if ok && metadata["npc_registry_digest"] != nil { + t.Fatalf("%s %s metadata = %#v, want generated identity only in framework provenance", lane.ID, component, metadata) + } + } + } + + var combatValue dnd.CombatTurnList + for _, serialized := range output.NormalizeOutputs { + switch serialized.LaneID { + case "spells": + spellValue := decodeRunnerSpellResponse(t, serialized.Artifact.Content) + if len(spellValue.SpellCasts) != 1 || spellValue.SpellCasts[0].Caster != "The Greencloak" { + t.Fatalf("spell output = %#v, want one registry-grounded-context spell", spellValue) + } + assertSpellEvidence(t, spellValue.SpellCasts[0].SourceRefs) + case "combat": + decoded, decodeErr := combatcodec.New().Decode(serialized.Artifact.Content) + if decodeErr != nil { + t.Fatalf("decode combat output: %v", decodeErr) + } + combatValue = decoded + } + } + if len(combatValue.CombatTurns) != 1 || combatValue.CombatTurns[0].Actor != "Mira Thorn" || combatValue.CombatTurns[0].Actions[0].Targets[0] != "Hooded Guard" { + t.Fatalf("combat output = %#v, want registry-normalized actor and target", combatValue) + } + assertCurrentEvidence(t, combatValue.CombatTurns[0].SourceRefs) +} + +func assertSpellEvidence(t *testing.T, references []shared.SourceRefResponse) { + t.Helper() + for _, reference := range references { + if reference.SourceID != "npc-session" { + t.Fatalf("spell evidence reference = %#v, want current source only", reference) + } + } +} + +func assertCurrentEvidence(t *testing.T, references []source.SourceRef) { + t.Helper() + for _, reference := range references { + if reference.SourceID != "npc-session" { + t.Fatalf("evidence reference = %#v, want current source only", reference) + } + } +} + +func loadGroundedPipelineConfig(t *testing.T) config.Config { + t.Helper() + fileConfig, err := config.LoadFileConfig(repositoryPathForIntegration("internal", "modules", "integration", "testdata", "dnd_npc_grounded_pipeline.yml")) + if err != nil { + t.Fatalf("LoadFileConfig() error = %v", err) + } + cfg := config.Default() + if err := cfg.ApplyFileConfig(fileConfig); err != nil { + t.Fatalf("ApplyFileConfig() error = %v", err) + } + return cfg +} + +type groundedDNDLLMClient struct { + mu sync.Mutex + requests []contracts.StructuredCompletionRequest +} + +func (client *groundedDNDLLMClient) 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: + payload = map[string]any{"npcs": []any{ + map[string]any{ + "name": "Mira Thorn", "aliases": []string{"The Greencloak"}, "description": "A guarded ranger.", + "relationships": []any{}, "source_refs": []any{map[string]int{"start_unit_id": 1, "end_unit_id": 1}}, + }, + map[string]any{ + "name": "Hooded Guard", "aliases": []string{}, "description": "A sentry.", + "relationships": []any{}, "source_refs": []any{map[string]int{"start_unit_id": 3, "end_unit_id": 3}}, + }, + }} + case spells.PromptID: + payload = map[string]any{"spell_casts": []any{map[string]any{ + "caster": "The Greencloak", "spell": "Cure Wounds", "effect": "Restores an ally.", + "narrative_description": "The Greencloak restores an ally.", + "source_refs": []any{map[string]int{"start_unit_id": 1, "end_unit_id": 1}}, + }}} + case combatextract.PromptID: + payload = map[string]any{"combat_turns": []any{map[string]any{ + "actor": "The Greencloak", "turn_kind": "turn", "round": 1, + "actions": []any{map[string]any{"category": "attack", "declaration": "watches", "targets": []string{"Hooded Guard"}, "resolution": "observed"}}, + "summary": "The Greencloak watches the gate.", + "source_refs": []any{map[string]int{"start_unit_id": 1, "end_unit_id": 1}}, + }}} + default: + return contracts.StructuredCompletionResponse{}, fmt.Errorf("unexpected grounded 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 grounded response: %w", err) + } + return contracts.StructuredCompletionResponse{Content: content, Provider: "test", Model: "grounded-fake"}, nil +} + +func (client *groundedDNDLLMClient) requestsSnapshot() []contracts.StructuredCompletionRequest { + client.mu.Lock() + defer client.mu.Unlock() + return append([]contracts.StructuredCompletionRequest(nil), client.requests...) +} + +func digestBytes(content []byte) string { + sum := sha256.Sum256(content) + return fmt.Sprintf("sha256:%x", sum[:]) +} + +var _ contracts.StructuredLLMClient = (*groundedDNDLLMClient)(nil) diff --git a/internal/modules/integration/testdata/dnd_npc_grounded_pipeline.yml b/internal/modules/integration/testdata/dnd_npc_grounded_pipeline.yml new file mode 100644 index 0000000..125fb4a --- /dev/null +++ b/internal/modules/integration/testdata/dnd_npc_grounded_pipeline.yml @@ -0,0 +1,31 @@ +version: 3 +output: + directory: ./notarius-output +cache: + chunk_plans: + mode: bypass + checkpoints: {} +debug: + directory: ./notarius-debug +pipelines: + dnd-npc-grounded: + input: seriatim + steps: + - id: identify-npcs + artifacts: + npcs: + extract: dnd/npcs + normalize: dnd/npcs + - id: grounded-events + references: + npcs: + artifact: + step: identify-npcs + lane: npcs + artifacts: + spells: + extract: dnd/spells + normalize: dnd/spells + combat: + extract: dnd/combat-turns + normalize: dnd/combat-turns