Implement operation-time D&D NPC artifact handoff

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

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"}},
 	}
 }
 

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() {
+	seeded := e.npcResolver.Seeded()
-		metadata["npc_registry_digest"] = e.npcRegistry.Digest()
+	if seeded.Bound() {
-		metadata["npc_count"] = e.npcRegistry.Count()
+		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() {
+	seeded := e.npcResolver.Seeded()
-		fingerprints = append(fingerprints, pipeline.CheckpointFingerprint{Name: "npc_registry", Value: e.npcRegistry.Digest()})
+	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,

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{

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() {
+	seeded := e.npcResolver.Seeded()
-		metadata["npc_registry_digest"] = e.npcRegistry.Digest()
+	if seeded.Bound() {
-		metadata["npc_count"] = e.npcRegistry.Count()
+		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() {
+	seeded := e.npcResolver.Seeded()
-		fingerprints = append(fingerprints, pipeline.CheckpointFingerprint{Name: "npc_registry", Value: e.npcRegistry.Digest()})
+	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,

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"),

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() {
+	seeded := n.npcResolver.Seeded()
-		metadata["npc_registry_digest"] = n.npcRegistry.Digest()
+	if seeded.Bound() {
-		metadata["npc_count"] = n.npcRegistry.Count()
+		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() {
+	seeded := n.npcResolver.Seeded()
-		fingerprints = append(fingerprints, pipeline.CheckpointFingerprint{Name: "npc_registry", Value: n.npcRegistry.Digest()})
+	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
 }
 

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})

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) {

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"),

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)

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