From ae97adb8b01a3461676462378dfadc2c5b51a89b Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Mon, 20 Jul 2026 21:16:08 +0000 Subject: [PATCH] Add assembled spell pipeline and checkpoint coverage --- .../assembled_spell_pipeline_contract_test.go | 231 ++++++++++++++++++ .../spell_catalog_identity_contract_test.go | 79 +++++- 2 files changed, 303 insertions(+), 7 deletions(-) create mode 100644 internal/cli/assembled_spell_pipeline_contract_test.go diff --git a/internal/cli/assembled_spell_pipeline_contract_test.go b/internal/cli/assembled_spell_pipeline_contract_test.go new file mode 100644 index 0000000..999d50c --- /dev/null +++ b/internal/cli/assembled_spell_pipeline_contract_test.go @@ -0,0 +1,231 @@ +package cli + +import ( + "context" + "encoding/json" + "fmt" + "reflect" + "sort" + "strings" + "sync" + "testing" + + "gitea.maximumdirect.net/eric/notarius/internal/core/artifacts" + "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" + spellnormalize "gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/normalize/spells" + spellcatalog "gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/spells/catalog" +) + +const assembledSpellExtractorKey = "test/dnd/spell-casts" + +func TestAssembledSpellPipelineNormalizesMergedCasts(t *testing.T) { + registries, resolved, extractor := assembledSpellPipeline(t, false) + prepared, err := pipeline.Prepare(resolved, registries, pipeline.ModuleDependencies{}) + if err != nil { + t.Fatalf("Prepare() error = %v, want nil", err) + } + + output, err := pipeline.New().Run(context.Background(), pipeline.RunInput{ + Prepared: prepared, + RawInput: readRepositoryFile(t, "examples", "seriatim-minimal-transcript.json"), + ChunkCacheMode: pipeline.ChunkCacheBypass, + }) + if err != nil { + t.Fatalf("Run() error = %v, want nil", err) + } + + chunkIndexes := extractor.chunkIndexesSnapshot() + sort.Ints(chunkIndexes) + if !reflect.DeepEqual(chunkIndexes, []int{0, 1}) { + t.Fatalf("extractor chunk indexes = %#v, want two chunk-boundary calls", chunkIndexes) + } + if output.Manifest.ValidationStatus != "approved" || len(output.Rejected) != 0 || len(output.NormalizeOutputs) != 1 { + t.Fatalf("run output = %#v, want approved normalized output without rejections", output) + } + if output.NormalizeOutputs[0].NormalizerKey != spellnormalize.Key { + t.Fatalf("normalized output module = %q, want %q", output.NormalizeOutputs[0].NormalizerKey, spellnormalize.Key) + } + + var normalized dnd.SpellList + if err := json.Unmarshal(output.NormalizeOutputs[0].Artifact.Content, &normalized); err != nil { + t.Fatalf("decode normalized output: %v", err) + } + if len(normalized.SpellCasts) != 2 { + t.Fatalf("normalized casts = %#v, want collapsed duplicate plus distinct evidence", normalized.SpellCasts) + } + first, distinct := normalized.SpellCasts[0], normalized.SpellCasts[1] + if first.Spell != "Cure Wounds" || first.Caster != " Aria \t" || first.Effect != "first occurrence" || first.NarrativeDescription != "first narrative" { + t.Fatalf("retained cast = %#v, want canonical spell with first occurrence fields", first) + } + if !reflect.DeepEqual(first.SourceRefs, []source.SourceRef{{SourceID: "session-alpha", StartUnitID: 1, EndUnitID: 1}, {SourceID: "session-alpha", StartUnitID: 2, EndUnitID: 2}}) { + t.Fatalf("retained refs = %#v, want sorted complete evidence", first.SourceRefs) + } + if distinct.Spell != "Cure Wounds" || distinct.Caster != "aria" || !reflect.DeepEqual(distinct.SourceRefs, []source.SourceRef{{SourceID: "session-alpha", StartUnitID: 2, EndUnitID: 2}}) { + t.Fatalf("distinct cast = %#v, want separate evidence event", distinct) + } + + wantWarningReasons := []string{ + spellnormalize.ReasonCodeSpellNameCanonicalized, + spellnormalize.ReasonCodeSourceReferencesNormalized, + spellnormalize.ReasonCodeDuplicateSpellCastCollapsed, + "spell_not_near_source", + } + gotWarningReasons := make([]string, len(output.Warnings)) + for index, warning := range output.Warnings { + gotWarningReasons[index] = warning.ReasonCode + } + if !reflect.DeepEqual(gotWarningReasons, wantWarningReasons) { + t.Fatalf("warnings = %#v, want deterministic normalize and validation warnings", output.Warnings) + } + if output.Warnings[2].Scope != "spell_casts[0]" || !strings.Contains(output.Warnings[2].Message, "retained input index 0") || !strings.Contains(output.Warnings[2].Message, "removed input indices [1]") { + t.Fatalf("duplicate warning = %#v, want retained and removed merged indices", output.Warnings[2]) + } + + warningsFile := decodeAssembledOutput[struct { + Warnings []contracts.Warning `json:"warnings"` + }](t, output.OutputFiles, "warnings.json") + if !reflect.DeepEqual(warningsFile.Warnings, output.Warnings) { + t.Fatalf("warnings file = %#v, run warnings = %#v, want manifest output path to preserve warnings", warningsFile.Warnings, output.Warnings) + } + manifest := decodeAssembledOutput[artifacts.RunManifest](t, output.OutputFiles, "manifest.json") + if len(manifest.ArtifactLanes) != 1 || manifest.ArtifactLanes[0].Normalizer != spellnormalize.Key { + t.Fatalf("manifest lanes = %#v, want assembled spell normalizer", manifest.ArtifactLanes) + } + normalizerMetadata, ok := manifest.ArtifactLanes[0].Metadata["normalizer"].(map[string]any) + _, hasOverlayIDs := normalizerMetadata["catalog_overlay_ids"] + if !ok || normalizerMetadata["catalog_base_id"] != spellcatalog.SRD5E2014ID || !strings.HasPrefix(stringValue(normalizerMetadata["catalog_digest"]), "sha256:") || !hasOverlayIDs { + t.Fatalf("normalizer manifest metadata = %#v, want base ID, digest, and overlay IDs", manifest.ArtifactLanes[0].Metadata) + } +} + +func TestAssembledSpellPipelineHonorsNormalizeValidatorOverride(t *testing.T) { + registries, resolved, _ := assembledSpellPipeline(t, true) + var normalizeChain *pipeline.ResolvedValidatorChain + for index := range resolved.ValidatorChains { + chain := &resolved.ValidatorChains[index] + if chain.Stage == pipeline.StageNormalize && chain.ModuleKey == spellnormalize.Key && chain.LaneID == "spells" { + normalizeChain = chain + break + } + } + if normalizeChain == nil || len(normalizeChain.Validators) != 1 || normalizeChain.Validators[0].Binding.Module != "generic/always_accept" { + t.Fatalf("normalize validator chain = %#v, want explicit always-accept override", normalizeChain) + } + + prepared, err := pipeline.Prepare(resolved, registries, pipeline.ModuleDependencies{}) + if err != nil { + t.Fatalf("Prepare() error = %v, want nil", err) + } + output, err := pipeline.New().Run(context.Background(), pipeline.RunInput{ + Prepared: prepared, + RawInput: readRepositoryFile(t, "examples", "seriatim-minimal-transcript.json"), + ChunkCacheMode: pipeline.ChunkCacheBypass, + }) + if err != nil || output.Manifest.ValidationStatus != "approved" || len(output.Rejected) != 0 || len(output.NormalizeOutputs) != 1 { + t.Fatalf("Run() error = %v output = %#v, want approved override run", err, output) + } + for _, warning := range output.Warnings { + if warning.ReasonCode == "spell_not_near_source" { + t.Fatalf("warnings = %#v, want explicit validator override to replace default relatedness chain", output.Warnings) + } + } +} + +func assembledSpellPipeline(t *testing.T, override bool) (pipeline.Registries, pipeline.ResolvedPipeline, *assembledSpellExtractor) { + t.Helper() + components := productionTestComponents(t) + extractor := &assembledSpellExtractor{} + if err := pipeline.RegisterExtractor[dnd.SpellList](components.registries.Extractors, pipeline.ModuleSpec{ + Key: assembledSpellExtractorKey, + Stage: pipeline.StageExtract, + Requires: []string{"chunks", "source.transcript"}, + Provides: []string{"dnd.spell_casts"}, + ArtifactKind: dnd.SpellListKind, + }, func() (contracts.Extractor[dnd.SpellList], error) { + return extractor, nil + }); err != nil { + t.Fatalf("register assembled extractor: %v", err) + } + + normalize := pipeline.Binding(spellnormalize.Key) + if override { + normalize.Validators = pipeline.ValidatorOverride{ + Set: true, + Validators: []pipeline.ModuleBinding{pipeline.Binding("generic/always_accept")}, + } + } + resolved, err := pipeline.ResolvePipeline(pipeline.PipelineProfile{ + ID: "assembled-dnd-spells", + Input: pipeline.Binding("seriatim"), + Chunk: pipeline.ModuleBinding{Module: "generic", Options: map[string]any{"max_units": 1}}, + Artifacts: map[string]pipeline.ArtifactLaneProfile{ + "spells": {Extract: pipeline.Binding(assembledSpellExtractorKey), Normalize: normalize}, + }, + Output: pipeline.Binding("json"), + }, pipeline.ResolveOptions{}, catalogFromRegistries(components.registries)) + if err != nil { + t.Fatalf("ResolvePipeline() error = %v, want nil", err) + } + return components.registries, resolved, extractor +} + +type assembledSpellExtractor struct { + mu sync.Mutex + chunkIndexes []int +} + +func (e *assembledSpellExtractor) Key() string { return assembledSpellExtractorKey } + +func (*assembledSpellExtractor) ReferenceSlots() []contracts.ReferenceSlot { return nil } + +func (e *assembledSpellExtractor) Extract(ctx context.Context, req contracts.TypedExtractionRequest) (contracts.TypedExtractionResult[dnd.SpellList], error) { + if err := ctx.Err(); err != nil { + return contracts.TypedExtractionResult[dnd.SpellList]{}, err + } + if req.Chunk == nil || req.Source == nil { + return contracts.TypedExtractionResult[dnd.SpellList]{}, fmt.Errorf("assembled extractor requires source and chunk") + } + e.mu.Lock() + e.chunkIndexes = append(e.chunkIndexes, req.Chunk.Index) + e.mu.Unlock() + refOne := source.SourceRef{SourceID: req.Source.ID, StartUnitID: 1, EndUnitID: 1} + refTwo := source.SourceRef{SourceID: req.Source.ID, StartUnitID: 2, EndUnitID: 2} + switch req.Chunk.Index { + case 0: + return contracts.TypedExtractionResult[dnd.SpellList]{Value: dnd.SpellList{SpellCasts: []dnd.SpellCast{{ + Caster: " Aria \t", Spell: " cure wounds ", Effect: "first occurrence", NarrativeDescription: "first narrative", SourceRefs: []source.SourceRef{refTwo, refOne}, + }}}}, nil + case 1: + return contracts.TypedExtractionResult[dnd.SpellList]{Value: dnd.SpellList{SpellCasts: []dnd.SpellCast{ + {Caster: "aria", Spell: "Cure Wounds", Effect: "removed occurrence", NarrativeDescription: "removed narrative", SourceRefs: []source.SourceRef{refOne, refTwo}}, + {Caster: "aria", Spell: "Cure Wounds", Effect: "different evidence", NarrativeDescription: "different narrative", SourceRefs: []source.SourceRef{refTwo}}, + }}}, nil + default: + return contracts.TypedExtractionResult[dnd.SpellList]{}, fmt.Errorf("unexpected assembled chunk index %d", req.Chunk.Index) + } +} + +func (e *assembledSpellExtractor) chunkIndexesSnapshot() []int { + e.mu.Lock() + defer e.mu.Unlock() + return append([]int(nil), e.chunkIndexes...) +} + +func decodeAssembledOutput[T any](t *testing.T, files []contracts.OutputFile, name string) T { + t.Helper() + for _, file := range files { + if file.Name != name { + continue + } + var value T + if err := json.Unmarshal(file.Bytes, &value); err != nil { + t.Fatalf("decode %s: %v", name, err) + } + return value + } + t.Fatalf("output files = %#v, want %q", files, name) + return *new(T) +} diff --git a/internal/cli/spell_catalog_identity_contract_test.go b/internal/cli/spell_catalog_identity_contract_test.go index 73959a9..e376870 100644 --- a/internal/cli/spell_catalog_identity_contract_test.go +++ b/internal/cli/spell_catalog_identity_contract_test.go @@ -14,6 +14,7 @@ import ( "gitea.maximumdirect.net/eric/notarius/internal/core/config" "gitea.maximumdirect.net/eric/notarius/internal/core/source" "gitea.maximumdirect.net/eric/notarius/internal/framework/checkpoint" + "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" "gitea.maximumdirect.net/eric/notarius/internal/modules/dnd" "gitea.maximumdirect.net/eric/notarius/internal/modules/dnd/extract/spells" @@ -42,6 +43,18 @@ func TestSpellCatalogBytesAffectCheckpointIdentityButNotSemanticDigest(t *testin t.Fatalf("spell catalog bindings = %#v, want catalog binding", bindings) } resolved.ArtifactLanes[0].ExtractReferences.Bindings[catalogBindingIndex].Source = overlayPath + normalizeBindings := resolved.ArtifactLanes[0].NormalizeReferences.Bindings + normalizeCatalogBindingIndex := -1 + for index, binding := range normalizeBindings { + if binding.SlotName == spellcatalog.SpellCatalogReferenceSlot { + normalizeCatalogBindingIndex = index + break + } + } + if normalizeCatalogBindingIndex < 0 { + t.Fatalf("normalize spell catalog bindings = %#v, want catalog binding", normalizeBindings) + } + resolved.ArtifactLanes[0].NormalizeReferences.Bindings[normalizeCatalogBindingIndex].Source = overlayPath if err := os.WriteFile(overlayPath, []byte(reorderedOverlayA), 0o600); err != nil { t.Fatal(err) @@ -52,6 +65,7 @@ func TestSpellCatalogBytesAffectCheckpointIdentityButNotSemanticDigest(t *testin } identityA := catalogCheckpointIdentity(t, materializedA) metadataA := catalogExtractorMetadata(t, materializedA) + normalizerMetadataA := catalogNormalizerMetadata(t, materializedA) referenceA := catalogReference(t, materializedA) if err := os.WriteFile(overlayPath, []byte(reorderedOverlayB), 0o600); err != nil { @@ -63,6 +77,7 @@ func TestSpellCatalogBytesAffectCheckpointIdentityButNotSemanticDigest(t *testin } identityB := catalogCheckpointIdentity(t, materializedB) metadataB := catalogExtractorMetadata(t, materializedB) + normalizerMetadataB := catalogNormalizerMetadata(t, materializedB) referenceB := catalogReference(t, materializedB) if identityA.Digest == identityB.Digest { @@ -82,6 +97,14 @@ func TestSpellCatalogBytesAffectCheckpointIdentityButNotSemanticDigest(t *testin if got, want := metadataA["catalog_overlay_ids"], []string{"campaign.a", "campaign.b"}; !reflect.DeepEqual(got, want) || !reflect.DeepEqual(metadataB["catalog_overlay_ids"], want) { t.Fatalf("extractor overlay IDs = %#v and %#v, want %#v", got, metadataB["catalog_overlay_ids"], want) } + normalizerDigestA, ok := normalizerMetadataA["catalog_digest"].(string) + normalizerDigestB, okB := normalizerMetadataB["catalog_digest"].(string) + if !ok || !okB || normalizerDigestA != digestA || normalizerDigestB != digestB { + t.Fatalf("normalizer catalog digests = %#v and %#v, want extractor semantic digests %q and %q", normalizerMetadataA["catalog_digest"], normalizerMetadataB["catalog_digest"], digestA, digestB) + } + if got, want := normalizerMetadataA["catalog_overlay_ids"], []string{"campaign.a", "campaign.b"}; !reflect.DeepEqual(got, want) || !reflect.DeepEqual(normalizerMetadataB["catalog_overlay_ids"], want) { + t.Fatalf("normalizer overlay IDs = %#v and %#v, want %#v", got, normalizerMetadataB["catalog_overlay_ids"], want) + } } func TestConfiguredSpellCatalogBindingChangesResolvedPipelineIdentity(t *testing.T) { @@ -205,6 +228,18 @@ func TestChangedSemanticSpellCatalogFingerprintCannotResumeRecordedCheckpoint(t if err := recorder.SourceSucceeded(materialized.Input.Module, &doc); err != nil { t.Fatal(err) } + normalizeDependencies := []pipeline.CheckpointFingerprint{{Name: "artifact[0]", Value: "sha256:merged-artifact"}} + normalizeSchema := contracts.ArtifactSchema{ID: "notarius.dnd.spells", Name: "notarius_dnd_spells", Version: "v1"} + normalizeArtifact := pipeline.CheckpointArtifact{ + LaneID: "spells", ModuleKey: spellnormalize.Key, SourceID: doc.ID, + SchemaDigest: contracts.DigestArtifactSchema(normalizeSchema), + Artifact: contracts.SerializedArtifact{ + Kind: dnd.SpellListKind, Schema: normalizeSchema, MediaType: "application/json", Content: []byte(`{"spell_casts":[]}`), + }, + } + if err := recorder.NormalizeSucceeded("spells", spellnormalize.Key, normalizeDependencies, normalizeArtifact, nil); err != nil { + t.Fatal(err) + } _, sameLoader, err := checkpointHandlersForRun(settings, Options{}, materialized, fingerprints, []byte("same input"), nil, nil, "", "", true) if err != nil { @@ -213,6 +248,9 @@ func TestChangedSemanticSpellCatalogFingerprintCannotResumeRecordedCheckpoint(t if _, decision := sameLoader.Source(materialized.Input.Module); !decision.Reused { t.Fatalf("same fingerprint decision = %#v, want reuse", decision) } + if restored, decision := sameLoader.Normalize("spells", spellnormalize.Key, normalizeDependencies); !decision.Reused || string(restored.Output.Artifact.Content) != `{"spell_casts":[]}` { + t.Fatalf("same normalize checkpoint = %#v, decision=%#v, want reuse", restored, decision) + } changed := append([]pipeline.CheckpointFingerprint(nil), fingerprints...) changed[0].Value = "sha256:changed-effective-catalog" _, changedLoader, err := checkpointHandlersForRun(settings, Options{}, materialized, changed, []byte("same input"), nil, nil, "", "", true) @@ -222,6 +260,9 @@ func TestChangedSemanticSpellCatalogFingerprintCannotResumeRecordedCheckpoint(t if _, decision := changedLoader.Source(materialized.Input.Module); decision.Reused { t.Fatalf("changed fingerprint decision = %#v, want cold miss", decision) } + if _, decision := changedLoader.Normalize("spells", spellnormalize.Key, normalizeDependencies); decision.Reused { + t.Fatalf("changed normalize fingerprint decision = %#v, want normalize checkpoint cold miss", decision) + } } func TestMaintainedProductionOverlayRunAlignsGroundingValidationAndProvenance(t *testing.T) { @@ -255,21 +296,35 @@ func TestMaintainedProductionOverlayRunAlignsGroundingValidationAndProvenance(t if got := stringValues(extractorMetadata["catalog_overlay_ids"]); !reflect.DeepEqual(got, []string{"notarius.example-campaign"}) { t.Fatalf("catalog overlay IDs = %#v, want maintained overlay", got) } + normalizerMetadata, ok := lane.Metadata["normalizer"].(map[string]any) + if !ok { + t.Fatalf("lane metadata = %#v, want normalizer metadata", lane.Metadata) + } + if normalizerMetadata["catalog_base_id"] != spellcatalog.SRD5E2014ID || !strings.HasPrefix(stringValue(normalizerMetadata["catalog_digest"]), "sha256:") || !reflect.DeepEqual(stringValues(normalizerMetadata["catalog_overlay_ids"]), []string{"notarius.example-campaign"}) { + t.Fatalf("normalizer catalog metadata = %#v, want base ID, semantic digest, and overlay IDs", normalizerMetadata) + } + if normalizerMetadata["catalog_digest"] != extractorMetadata["catalog_digest"] || !reflect.DeepEqual(stringValues(normalizerMetadata["catalog_overlay_ids"]), stringValues(extractorMetadata["catalog_overlay_ids"])) { + t.Fatalf("extractor metadata = %#v, normalizer metadata = %#v, want shared catalog identity", extractorMetadata, normalizerMetadata) + } - var catalogProvenance *artifacts.ReferenceProvenance + var catalogProvenances []artifacts.ReferenceProvenance for index := range manifest.References { reference := &manifest.References[index] if reference.SlotName == spellcatalog.SpellCatalogReferenceSlot { - catalogProvenance = reference - break + catalogProvenances = append(catalogProvenances, *reference) } } - if catalogProvenance == nil { - t.Fatalf("manifest references = %#v, want spell catalog provenance", manifest.References) + if len(catalogProvenances) != 2 { + t.Fatalf("manifest references = %#v, want independently materialized extract and normalize catalog provenance", manifest.References) } overlayBytes := readRepositoryFile(t, "examples", "dnd-spells-catalog.json") - if catalogProvenance.Stage != "extract" || catalogProvenance.LaneID != "spells" || catalogProvenance.OriginType != "file" || catalogProvenance.MediaType != "application/json" || catalogProvenance.SizeBytes != int64(len(overlayBytes)) || catalogProvenance.Digest != digestBytes(overlayBytes) || !strings.Contains(catalogProvenance.OriginURI, "dnd-spells-catalog.json") { - t.Fatalf("catalog provenance = %#v, want extract origin, media, size, and raw digest", catalogProvenance) + for _, catalogProvenance := range catalogProvenances { + if catalogProvenance.Stage != "extract" && catalogProvenance.Stage != "normalize" { + t.Fatalf("catalog provenance = %#v, want extract or normalize scope", catalogProvenance) + } + if catalogProvenance.LaneID != "spells" || catalogProvenance.OriginType != "file" || catalogProvenance.MediaType != "application/json" || catalogProvenance.SizeBytes != int64(len(overlayBytes)) || catalogProvenance.Digest != digestBytes(overlayBytes) || !strings.Contains(catalogProvenance.OriginURI, "dnd-spells-catalog.json") { + t.Fatalf("catalog provenance = %#v, want raw overlay provenance in both scopes", catalogProvenance) + } } manifestBytes, err := json.Marshal(manifest) if err != nil { @@ -325,6 +380,16 @@ func catalogExtractorMetadata(t *testing.T, resolved pipeline.ResolvedPipeline) return extractor.ManifestMetadata() } +func catalogNormalizerMetadata(t *testing.T, resolved pipeline.ResolvedPipeline) map[string]any { + t.Helper() + lane := resolved.ArtifactLanes[0] + normalizer, err := spellnormalize.New(spellnormalize.Options{}, lane.NormalizeReferences.ReferenceSet) + if err != nil { + t.Fatalf("construct normalizer: %v", err) + } + return normalizer.ManifestMetadata() +} + func catalogReference(t *testing.T, resolved pipeline.ResolvedPipeline) artifacts.ReferenceProvenance { t.Helper() for _, reference := range pipeline.ReferenceProvenance(resolved) {