diff --git a/docs/internal/pipeline.md b/docs/internal/pipeline.md index 36bda2a..31fd08f 100644 --- a/docs/internal/pipeline.md +++ b/docs/internal/pipeline.md @@ -82,6 +82,11 @@ separate target namespaces; serialized registrations declare whether they support chunks, artifacts, or both. Duplicate variants and exact Go-type mismatches are rejected deterministically. +Lane-sensitive merger and normalizer spec discovery always supplies the +extractor's artifact kind, so variants under one reusable key may declare +different capabilities and reference slots. Kind-neutral registry inspection +selects the first registered artifact kind in sorted order. + Production composition registers the D&D spell-list codec and typed extractor, matching typed merge, normalize, and semantic-validator variants, and serialized JSON validators. Every artifact lane resolves through the typed diff --git a/internal/cli/run.go b/internal/cli/run.go index 49d580f..19cab56 100644 --- a/internal/cli/run.go +++ b/internal/cli/run.go @@ -1117,6 +1117,10 @@ func selectedReferenceTargets(cfg config.Config, pipelineID string, only []strin if err != nil { return nil, fmt.Errorf("pipeline %q lane %q extract module %q: %w", strings.TrimSpace(pipelineID), laneID, extractModule, err) } + artifactKind := extractSpec.ArtifactKind + if artifactKind == "" { + return nil, fmt.Errorf("pipeline %q lane %q extract module %q does not declare an artifact kind", strings.TrimSpace(pipelineID), laneID, extractModule) + } targets = append(targets, selectedReferenceTarget{ laneID: laneID, stage: pipeline.StageExtract, @@ -1128,7 +1132,7 @@ func selectedReferenceTargets(cfg config.Config, pipelineID string, only []strin if mergeModule == "" { mergeModule = pipeline.DefaultMergeModule } - mergeSpec, err := cliReferenceMergerSpec(catalog, mergeModule) + mergeSpec, err := cliReferenceMergerSpec(catalog, mergeModule, artifactKind) if err != nil { return nil, fmt.Errorf("pipeline %q lane %q merge module %q: %w", strings.TrimSpace(pipelineID), laneID, mergeModule, err) } @@ -1143,7 +1147,7 @@ func selectedReferenceTargets(cfg config.Config, pipelineID string, only []strin if normalizeModule == "" { normalizeModule = pipeline.DefaultNormalizeModule } - normalizeSpec, err := cliReferenceNormalizerSpec(catalog, normalizeModule) + normalizeSpec, err := cliReferenceNormalizerSpec(catalog, normalizeModule, artifactKind) if err != nil { return nil, fmt.Errorf("pipeline %q lane %q normalize module %q: %w", strings.TrimSpace(pipelineID), laneID, normalizeModule, err) } @@ -1189,28 +1193,47 @@ func cliReferenceExtractorSpec(catalog pipeline.ModuleCatalog, module string) (p return spec, nil } -func cliReferenceMergerSpec(catalog pipeline.ModuleCatalog, module string) (pipeline.ModuleSpec, error) { +func cliReferenceMergerSpec(catalog pipeline.ModuleCatalog, module string, kind contracts.ArtifactKind) (pipeline.ModuleSpec, error) { if catalog.Mergers == nil { return pipeline.ModuleSpec{}, fmt.Errorf("module %q is not registered", module) } - spec, ok := catalog.Mergers.Spec(module) - if !ok { + registered := catalog.Mergers.RegisteredArtifactKinds(module) + if len(registered) == 0 { return pipeline.ModuleSpec{}, fmt.Errorf("module %q is not registered", module) } + spec, ok := catalog.Mergers.SpecForArtifact(module, kind) + if !ok { + return pipeline.ModuleSpec{}, cliReferenceArtifactVariantError("merger", module, kind, registered) + } return spec, nil } -func cliReferenceNormalizerSpec(catalog pipeline.ModuleCatalog, module string) (pipeline.ModuleSpec, error) { +func cliReferenceNormalizerSpec(catalog pipeline.ModuleCatalog, module string, kind contracts.ArtifactKind) (pipeline.ModuleSpec, error) { if catalog.Normalizers == nil { return pipeline.ModuleSpec{}, fmt.Errorf("module %q is not registered", module) } - spec, ok := catalog.Normalizers.Spec(module) - if !ok { + registered := catalog.Normalizers.RegisteredArtifactKinds(module) + if len(registered) == 0 { return pipeline.ModuleSpec{}, fmt.Errorf("module %q is not registered", module) } + spec, ok := catalog.Normalizers.SpecForArtifact(module, kind) + if !ok { + return pipeline.ModuleSpec{}, cliReferenceArtifactVariantError("normalizer", module, kind, registered) + } return spec, nil } +func cliReferenceArtifactVariantError(moduleType string, module string, kind contracts.ArtifactKind, registered []contracts.ArtifactKind) error { + values := make([]string, len(registered)) + for i, value := range registered { + values[i] = string(value) + } + if len(values) == 0 { + return fmt.Errorf("%s %q has no typed variant for artifact kind %q", moduleType, module, kind) + } + return fmt.Errorf("%s %q has no typed variant for artifact kind %q; registered kinds: %s", moduleType, module, kind, strings.Join(values, ", ")) +} + func referenceSlotSet(slots []contracts.ReferenceSlot) map[string]struct{} { slotSet := make(map[string]struct{}, len(slots)) for _, slot := range slots { diff --git a/internal/cli/run_test.go b/internal/cli/run_test.go index adf7a3a..8cef149 100644 --- a/internal/cli/run_test.go +++ b/internal/cli/run_test.go @@ -1268,6 +1268,111 @@ func TestRunPipelineReferenceFlagBindsUnambiguousSlot(t *testing.T) { } } +func TestCLIReferenceDiscoveryUsesLaneArtifactVariant(t *testing.T) { + catalog := referenceVariantCatalog(t) + cfg := config.Config{Pipelines: map[string]pipeline.PipelineProfile{ + "variants": { + ID: "variants", + Chunk: pipeline.Binding("generic"), + Artifacts: map[string]pipeline.ArtifactLaneProfile{ + "alpha": { + Extract: pipeline.Binding("extract/alpha"), + Merge: pipeline.Binding("shared/merge"), + Normalize: pipeline.Binding("shared/normalize"), + }, + "beta": { + Extract: pipeline.Binding("extract/beta"), + Merge: pipeline.Binding("shared/merge"), + Normalize: pipeline.Binding("shared/normalize"), + }, + }, + }, + }} + targets, err := selectedReferenceTargets(cfg, "variants", nil, catalog) + if err != nil { + t.Fatalf("selectedReferenceTargets() error = %v, want nil", err) + } + + tests := []struct { + name string + selector cliReferenceSelector + laneID string + stage pipeline.ModuleStage + }{ + { + name: "qualified merger", + selector: cliReferenceSelector{LaneID: "alpha", Stage: pipeline.StageMerge, SlotName: "alpha_merge"}, + laneID: "alpha", + stage: pipeline.StageMerge, + }, + { + name: "unqualified merger", + selector: cliReferenceSelector{Stage: pipeline.StageMerge, SlotName: "beta_merge"}, + laneID: "beta", + stage: pipeline.StageMerge, + }, + { + name: "flat normalizer", + selector: cliReferenceSelector{SlotName: "alpha_normalize"}, + laneID: "alpha", + stage: pipeline.StageNormalize, + }, + { + name: "lane normalizer", + selector: cliReferenceSelector{LaneID: "beta", SlotName: "beta_normalize"}, + laneID: "beta", + stage: pipeline.StageNormalize, + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + target, err := resolveCLIReferenceTarget(targets, tc.selector) + if err != nil { + t.Fatalf("resolveCLIReferenceTarget() error = %v, want nil", err) + } + if target.laneID != tc.laneID || target.stage != tc.stage { + t.Fatalf("target = %#v, want lane %q stage %q", target, tc.laneID, tc.stage) + } + }) + } + + _, err = resolveCLIReferenceTarget(targets, cliReferenceSelector{LaneID: "beta", Stage: pipeline.StageMerge, SlotName: "alpha_merge"}) + if err == nil || !strings.Contains(err.Error(), "not declared") { + t.Fatalf("beta alpha-variant reference error = %v, want slot rejection", err) + } +} + +func TestCLIReferenceDiscoveryReportsMissingArtifactVariant(t *testing.T) { + catalog := referenceVariantCatalog(t) + if err := pipeline.RegisterExtractor(catalog.Extractors, pipeline.ModuleSpec{Key: "extract/missing", Stage: pipeline.StageExtract, ArtifactKind: "test/missing"}, func() (contracts.Extractor[fakeRunArtifact], error) { + return referenceVariantExtractor{key: "extract/missing"}, nil + }); err != nil { + t.Fatalf("register missing-kind extractor: %v", err) + } + cfg := config.Config{Pipelines: map[string]pipeline.PipelineProfile{ + "variants": { + ID: "variants", + Artifacts: map[string]pipeline.ArtifactLaneProfile{ + "missing": { + Extract: pipeline.Binding("extract/missing"), + Merge: pipeline.Binding("shared/merge"), + Normalize: pipeline.Binding("shared/normalize"), + }, + }, + }, + }} + _, err := selectedReferenceTargets(cfg, "variants", nil, catalog) + want := []string{"pipeline \"variants\"", "lane \"missing\"", "merge module \"shared/merge\"", "artifact kind \"test/missing\"", "registered kinds: test/alpha, test/beta"} + if err == nil { + t.Fatal("selectedReferenceTargets() error = nil, want missing variant error") + } + for _, value := range want { + if !strings.Contains(err.Error(), value) { + t.Fatalf("selectedReferenceTargets() error = %q, want %q", err, value) + } + } +} + func TestRunPipelineReferenceFlagBindsLaneQualifiedSlot(t *testing.T) { configPath := writeTestConfig(t, testConfigYAML("example", "events", "notes")) inputPath := filepath.Join(t.TempDir(), "missing.json") @@ -3652,6 +3757,30 @@ const fakeRunArtifactKind contracts.ArtifactKind = "test/fake" type fakeRunArtifact struct { Value bool `json:"value"` } + +type referenceVariantExtractor struct{ key string } + +func (e referenceVariantExtractor) Key() string { return e.key } +func (referenceVariantExtractor) ReferenceSlots() []contracts.ReferenceSlot { return nil } +func (referenceVariantExtractor) Extract(context.Context, contracts.TypedExtractionRequest) (contracts.TypedExtractionResult[fakeRunArtifact], error) { + return contracts.TypedExtractionResult[fakeRunArtifact]{}, nil +} + +type referenceVariantMerger struct{ key string } + +func (m referenceVariantMerger) Key() string { return m.key } +func (referenceVariantMerger) Merge(context.Context, contracts.TypedMergeRequest[fakeRunArtifact]) (contracts.TypedMergeResult[fakeRunArtifact], error) { + return contracts.TypedMergeResult[fakeRunArtifact]{}, nil +} + +type referenceVariantNormalizer struct{ key string } + +func (n referenceVariantNormalizer) Key() string { return n.key } +func (referenceVariantNormalizer) ReferenceSlots() []contracts.ReferenceSlot { return nil } +func (referenceVariantNormalizer) Normalize(context.Context, contracts.TypedNormalizeRequest[fakeRunArtifact]) (contracts.TypedNormalizeResult[fakeRunArtifact], error) { + return contracts.TypedNormalizeResult[fakeRunArtifact]{}, nil +} + type fakeRunCodec struct{} func (fakeRunCodec) Kind() contracts.ArtifactKind { return fakeRunArtifactKind } @@ -3994,6 +4123,52 @@ func fakeCatalog(t *testing.T, overrides ...pipeline.ModuleSpec) pipeline.Module } } +func referenceVariantCatalog(t *testing.T) pipeline.ModuleCatalog { + t.Helper() + chunkers := pipeline.NewChunkerRegistry() + extractors := pipeline.NewExtractorRegistry() + mergers := pipeline.NewMergerRegistry() + normalizers := pipeline.NewNormalizerRegistry() + mustRegisterChunker(t, chunkers, pipeline.ModuleSpec{Key: "generic", Stage: pipeline.StageChunk}) + + for _, item := range []struct { + key string + kind contracts.ArtifactKind + }{ + {key: "extract/alpha", kind: "test/alpha"}, + {key: "extract/beta", kind: "test/beta"}, + } { + item := item + if err := pipeline.RegisterExtractor(extractors, pipeline.ModuleSpec{Key: item.key, Stage: pipeline.StageExtract, ArtifactKind: item.kind}, func() (contracts.Extractor[fakeRunArtifact], error) { + return referenceVariantExtractor{key: item.key}, nil + }); err != nil { + t.Fatalf("register extractor %q: %v", item.key, err) + } + } + for _, item := range []struct { + kind contracts.ArtifactKind + mergeSlot string + normalizeSlot string + }{ + {kind: "test/beta", mergeSlot: "beta_merge", normalizeSlot: "beta_normalize"}, + {kind: "test/alpha", mergeSlot: "alpha_merge", normalizeSlot: "alpha_normalize"}, + } { + mergeSpec := pipeline.ModuleSpec{Key: "shared/merge", Stage: pipeline.StageMerge, ArtifactKind: item.kind, ReferenceSlots: []contracts.ReferenceSlot{{Name: item.mergeSlot}}} + if err := pipeline.RegisterMerger(mergers, mergeSpec, func() (contracts.Merger[fakeRunArtifact], error) { + return referenceVariantMerger{key: "shared/merge"}, nil + }); err != nil { + t.Fatalf("register merger %q: %v", item.kind, err) + } + normalizeSpec := pipeline.ModuleSpec{Key: "shared/normalize", Stage: pipeline.StageNormalize, ArtifactKind: item.kind, ReferenceSlots: []contracts.ReferenceSlot{{Name: item.normalizeSlot}}} + if err := pipeline.RegisterNormalizer(normalizers, normalizeSpec, func() (contracts.Normalizer[fakeRunArtifact], error) { + return referenceVariantNormalizer{key: "shared/normalize"}, nil + }); err != nil { + t.Fatalf("register normalizer %q: %v", item.kind, err) + } + } + return pipeline.ModuleCatalog{Chunkers: chunkers, Extractors: extractors, Mergers: mergers, Normalizers: normalizers} +} + func mustRegisterInput(t *testing.T, registry *pipeline.InputAdapterRegistry, spec pipeline.ModuleSpec) { t.Helper() if err := registry.RegisterWithSpec(spec, func() (contracts.InputAdapter, error) { return fakeRunInputAdapter{}, nil }); err != nil { diff --git a/internal/framework/pipeline/merger_registry.go b/internal/framework/pipeline/merger_registry.go index 5d4a730..f1f4bf1 100644 --- a/internal/framework/pipeline/merger_registry.go +++ b/internal/framework/pipeline/merger_registry.go @@ -108,16 +108,28 @@ func (r *MergerRegistry) validateOptions(key string, kind contracts.ArtifactKind } func (r *MergerRegistry) Spec(key string) (ModuleSpec, bool) { - if r == nil { + // Spec is for kind-neutral catalog inspection. Behavior-sensitive callers + // must use SpecForArtifact so they select the lane's exact typed variant. + kinds := r.registeredKinds(key) + if len(kinds) == 0 { return ModuleSpec{}, false } - module := strings.TrimSpace(key) - for variant, entry := range r.typedEntries { - if variant.module == module { - return cloneModuleSpec(entry.spec), true - } + return r.SpecForArtifact(key, kinds[0]) +} + +// SpecForArtifact returns the merger spec registered for an exact artifact kind. +func (r *MergerRegistry) SpecForArtifact(key string, kind contracts.ArtifactKind) (ModuleSpec, bool) { + entry, ok := r.typedEntry(key, kind) + if !ok { + return ModuleSpec{}, false } - return ModuleSpec{}, false + return cloneModuleSpec(entry.spec), true +} + +// RegisteredArtifactKinds returns the sorted artifact kinds registered for a +// merger key. +func (r *MergerRegistry) RegisteredArtifactKinds(key string) []contracts.ArtifactKind { + return r.registeredKinds(key) } func (r *MergerRegistry) typedEntry(key string, kind contracts.ArtifactKind) (typedMergerEntry, bool) { diff --git a/internal/framework/pipeline/normalizer_registry.go b/internal/framework/pipeline/normalizer_registry.go index ea324e9..afd7ff3 100644 --- a/internal/framework/pipeline/normalizer_registry.go +++ b/internal/framework/pipeline/normalizer_registry.go @@ -99,16 +99,28 @@ func (r *NormalizerRegistry) validateOptions(key string, kind contracts.Artifact } func (r *NormalizerRegistry) Spec(key string) (ModuleSpec, bool) { - if r == nil { + // Spec is for kind-neutral catalog inspection. Behavior-sensitive callers + // must use SpecForArtifact so they select the lane's exact typed variant. + kinds := r.registeredKinds(key) + if len(kinds) == 0 { return ModuleSpec{}, false } - module := strings.TrimSpace(key) - for variant, entry := range r.typedEntries { - if variant.module == module { - return cloneModuleSpec(entry.spec), true - } + return r.SpecForArtifact(key, kinds[0]) +} + +// SpecForArtifact returns the normalizer spec registered for an exact artifact kind. +func (r *NormalizerRegistry) SpecForArtifact(key string, kind contracts.ArtifactKind) (ModuleSpec, bool) { + entry, ok := r.typedEntry(key, kind) + if !ok { + return ModuleSpec{}, false } - return ModuleSpec{}, false + return cloneModuleSpec(entry.spec), true +} + +// RegisteredArtifactKinds returns the sorted artifact kinds registered for a +// normalizer key. +func (r *NormalizerRegistry) RegisteredArtifactKinds(key string) []contracts.ArtifactKind { + return r.registeredKinds(key) } func (r *NormalizerRegistry) typedEntry(key string, kind contracts.ArtifactKind) (typedNormalizerEntry, bool) { diff --git a/internal/framework/pipeline/profile.go b/internal/framework/pipeline/profile.go index 7204599..9a30d74 100644 --- a/internal/framework/pipeline/profile.go +++ b/internal/framework/pipeline/profile.go @@ -417,7 +417,7 @@ func resolveArtifactIdentity(pipelineID, laneID string, lane *ResolvedArtifactLa func mergerSpecForArtifact(catalog ModuleCatalog, key string, kind contracts.ArtifactKind, expectedType reflect.Type) (ModuleSpec, error) { if kind == "" { - return mergerSpec(catalog, key) + return ModuleSpec{}, fmt.Errorf("merger %q cannot be resolved without an artifact kind", key) } if catalog.Mergers == nil { return ModuleSpec{}, fmt.Errorf("module %q is not registered", key) @@ -434,7 +434,7 @@ func mergerSpecForArtifact(catalog ModuleCatalog, key string, kind contracts.Art func normalizerSpecForArtifact(catalog ModuleCatalog, key string, kind contracts.ArtifactKind, expectedType reflect.Type) (ModuleSpec, error) { if kind == "" { - return normalizerSpec(catalog, key) + return ModuleSpec{}, fmt.Errorf("normalizer %q cannot be resolved without an artifact kind", key) } if catalog.Normalizers == nil { return ModuleSpec{}, fmt.Errorf("module %q is not registered", key) @@ -1042,14 +1042,6 @@ func extractorSpec(catalog ModuleCatalog, key string) (ModuleSpec, error) { return registrySpec(catalog.Extractors, key) } -func mergerSpec(catalog ModuleCatalog, key string) (ModuleSpec, error) { - return registrySpec(catalog.Mergers, key) -} - -func normalizerSpec(catalog ModuleCatalog, key string) (ModuleSpec, error) { - return registrySpec(catalog.Normalizers, key) -} - func outputSpec(catalog ModuleCatalog, key string) (ModuleSpec, error) { return registrySpec(catalog.Outputs, key) } diff --git a/internal/framework/pipeline/references.go b/internal/framework/pipeline/references.go index 9f7d49a..2a0fbe7 100644 --- a/internal/framework/pipeline/references.go +++ b/internal/framework/pipeline/references.go @@ -30,7 +30,7 @@ type ReferenceMaterializationOptions struct { func MaterializeReferences(resolved ResolvedPipeline, catalog ModuleCatalog, options ReferenceMaterializationOptions) (ResolvedPipeline, []contracts.Warning, error) { out := resolved out.ChunkReferences = CloneReferenceTarget(resolved.ChunkReferences) - chunkReferenceSet, chunkWarnings, err := materializeReferenceTarget(resolved.ID, resolved.ChunkReferences, catalog, options) + chunkReferenceSet, chunkWarnings, err := materializeReferenceTarget(resolved.ID, resolved.ChunkReferences, "", catalog, options) if err != nil { return ResolvedPipeline{}, nil, err } @@ -46,21 +46,21 @@ func MaterializeReferences(resolved ResolvedPipeline, catalog ModuleCatalog, opt materializedLane.ExtractReferences = CloneReferenceTarget(lane.ExtractReferences) materializedLane.MergeReferences = CloneReferenceTarget(lane.MergeReferences) materializedLane.NormalizeReferences = CloneReferenceTarget(lane.NormalizeReferences) - extractReferenceSet, laneWarnings, err := materializeReferenceTarget(resolved.ID, lane.ExtractReferences, catalog, options) + extractReferenceSet, laneWarnings, err := materializeReferenceTarget(resolved.ID, lane.ExtractReferences, lane.ArtifactKind, catalog, options) if err != nil { return ResolvedPipeline{}, nil, err } materializedLane.ExtractReferences.ReferenceSet = extractReferenceSet warnings = append(warnings, laneWarnings...) - mergeReferenceSet, laneWarnings, err := materializeReferenceTarget(resolved.ID, lane.MergeReferences, catalog, options) + mergeReferenceSet, laneWarnings, err := materializeReferenceTarget(resolved.ID, lane.MergeReferences, lane.ArtifactKind, catalog, options) if err != nil { return ResolvedPipeline{}, nil, err } materializedLane.MergeReferences.ReferenceSet = mergeReferenceSet warnings = append(warnings, laneWarnings...) - normalizeReferenceSet, laneWarnings, err := materializeReferenceTarget(resolved.ID, lane.NormalizeReferences, catalog, options) + normalizeReferenceSet, laneWarnings, err := materializeReferenceTarget(resolved.ID, lane.NormalizeReferences, lane.ArtifactKind, catalog, options) if err != nil { return ResolvedPipeline{}, nil, err } @@ -74,13 +74,14 @@ func MaterializeReferences(resolved ResolvedPipeline, catalog ModuleCatalog, opt func materializeReferenceTarget( pipelineID string, target ResolvedReferenceTarget, + artifactKind contracts.ArtifactKind, catalog ModuleCatalog, options ReferenceMaterializationOptions, ) (contracts.ReferenceSet, []contracts.Warning, error) { if len(target.Bindings) == 0 { return contracts.ReferenceSet{}, nil, nil } - spec, err := referenceTargetSpec(target, catalog) + spec, err := referenceTargetSpec(target, artifactKind, catalog) if err != nil { return contracts.ReferenceSet{}, nil, fmt.Errorf("%s: %w", referenceTargetContext(pipelineID, target), err) } @@ -142,16 +143,28 @@ func materializeReferenceTarget( return set, warnings, nil } -func referenceTargetSpec(target ResolvedReferenceTarget, catalog ModuleCatalog) (ModuleSpec, error) { +func referenceTargetSpec(target ResolvedReferenceTarget, artifactKind contracts.ArtifactKind, catalog ModuleCatalog) (ModuleSpec, error) { switch target.Stage { case StageChunk: return registrySpec(catalog.Chunkers, target.Module) case StageExtract: return registrySpec(catalog.Extractors, target.Module) case StageMerge: - return registrySpec(catalog.Mergers, target.Module) + if catalog.Mergers == nil { + return ModuleSpec{}, fmt.Errorf("module %q is not registered", target.Module) + } + if spec, ok := catalog.Mergers.SpecForArtifact(target.Module, artifactKind); ok { + return spec, nil + } + return ModuleSpec{}, missingArtifactVariantError("merger", target.Module, artifactKind, catalog.Mergers.registeredKinds(target.Module)) case StageNormalize: - return registrySpec(catalog.Normalizers, target.Module) + if catalog.Normalizers == nil { + return ModuleSpec{}, fmt.Errorf("module %q is not registered", target.Module) + } + if spec, ok := catalog.Normalizers.SpecForArtifact(target.Module, artifactKind); ok { + return spec, nil + } + return ModuleSpec{}, missingArtifactVariantError("normalizer", target.Module, artifactKind, catalog.Normalizers.registeredKinds(target.Module)) default: return ModuleSpec{}, fmt.Errorf("reference target stage %q is not supported", target.Stage) } diff --git a/internal/framework/pipeline/references_test.go b/internal/framework/pipeline/references_test.go index b045a4f..01190d4 100644 --- a/internal/framework/pipeline/references_test.go +++ b/internal/framework/pipeline/references_test.go @@ -175,6 +175,57 @@ func TestMaterializeReferencesStoresSetsAndProvenanceForAllTargets(t *testing.T) } } +func TestMaterializeReferencesUsesLaneArtifactVariant(t *testing.T) { + configDir := t.TempDir() + writeReferenceFile(t, filepath.Join(configDir, "merge.txt"), []byte("merge alpha")) + writeReferenceFile(t, filepath.Join(configDir, "normalize.txt"), []byte("normalize alpha")) + mergers := NewMergerRegistry() + normalizers := NewNormalizerRegistry() + for _, item := range []struct { + kind contracts.ArtifactKind + mergeSlot string + normalizeSlot string + }{ + {kind: "test/beta", mergeSlot: "beta_merge", normalizeSlot: "beta_normalize"}, + {kind: "test/alpha", mergeSlot: "alpha_merge", normalizeSlot: "alpha_normalize"}, + } { + if err := RegisterMerger(mergers, ModuleSpec{Key: "shared/merge", Stage: StageMerge, ArtifactKind: item.kind, ReferenceSlots: []contracts.ReferenceSlot{{Name: item.mergeSlot}}}, func() (contracts.Merger[codecNotes], error) { + return typedTestMerger[codecNotes]{key: "shared/merge"}, nil + }); err != nil { + t.Fatalf("RegisterMerger(%s): %v", item.kind, err) + } + if err := RegisterNormalizer(normalizers, ModuleSpec{Key: "shared/normalize", Stage: StageNormalize, ArtifactKind: item.kind, ReferenceSlots: []contracts.ReferenceSlot{{Name: item.normalizeSlot}}}, func() (contracts.Normalizer[codecNotes], error) { + return typedTestNormalizer[codecNotes]{key: "shared/normalize"}, nil + }); err != nil { + t.Fatalf("RegisterNormalizer(%s): %v", item.kind, err) + } + } + resolved := ResolvedPipeline{ + ID: "variants", + ArtifactLanes: []ResolvedArtifactLane{{ + ID: "alpha", + ArtifactKind: "test/alpha", + MergeReferences: referenceTarget(StageMerge, "alpha", "shared/merge", []ReferenceBinding{{ + Stage: StageMerge, LaneID: "alpha", SlotName: "alpha_merge", Source: "merge.txt", BindingSource: contracts.ReferenceBindingSourceConfig, + }}), + NormalizeReferences: referenceTarget(StageNormalize, "alpha", "shared/normalize", []ReferenceBinding{{ + Stage: StageNormalize, LaneID: "alpha", SlotName: "alpha_normalize", Source: "normalize.txt", BindingSource: contracts.ReferenceBindingSourceConfig, + }}), + }}, + } + materialized, _, err := MaterializeReferences(resolved, ModuleCatalog{Mergers: mergers, Normalizers: normalizers}, ReferenceMaterializationOptions{ConfigPath: filepath.Join(configDir, "notarius.yml")}) + if err != nil { + t.Fatalf("MaterializeReferences() error = %v, want nil", err) + } + lane := materialized.ArtifactLanes[0] + if got := string(lane.MergeReferences.ReferenceSet.Slots["alpha_merge"].Items[0].Content); got != "merge alpha" { + t.Fatalf("merge reference = %q", got) + } + if got := string(lane.NormalizeReferences.ReferenceSet.Slots["alpha_normalize"].Items[0].Content); got != "normalize alpha" { + t.Fatalf("normalize reference = %q", got) + } +} + func TestMaterializeReferencesRejectsNonUTF8Content(t *testing.T) { configDir := t.TempDir() path := filepath.Join(configDir, "bad.txt") diff --git a/internal/framework/pipeline/typed_resolution_test.go b/internal/framework/pipeline/typed_resolution_test.go index 2b743ae..a87510c 100644 --- a/internal/framework/pipeline/typed_resolution_test.go +++ b/internal/framework/pipeline/typed_resolution_test.go @@ -2,6 +2,7 @@ package pipeline import ( "context" + "fmt" "reflect" "strings" "testing" @@ -267,6 +268,101 @@ func TestConstructorRegistrationsRejectUnconfiguredOptions(t *testing.T) { } } +func TestTypedRegistrySpecLookupUsesArtifactKindAndStableCatalogOrder(t *testing.T) { + type registration struct { + kind contracts.ArtifactKind + slot string + } + orders := [][]registration{ + {{kind: "test/score", slot: "score_notes"}, {kind: "test/notes", slot: "note_notes"}}, + {{kind: "test/notes", slot: "note_notes"}, {kind: "test/score", slot: "score_notes"}}, + } + + for orderIndex, order := range orders { + t.Run(fmt.Sprintf("registration order %d", orderIndex+1), func(t *testing.T) { + mergers := NewMergerRegistry() + normalizers := NewNormalizerRegistry() + for _, item := range order { + base := ModuleSpec{Key: "typed/shared", ArtifactKind: item.kind, ReferenceSlots: []contracts.ReferenceSlot{{Name: item.slot}}} + mergeSpec := base + mergeSpec.Stage = StageMerge + normalizeSpec := base + normalizeSpec.Stage = StageNormalize + switch item.kind { + case "test/notes": + if err := RegisterMerger(mergers, mergeSpec, func() (contracts.Merger[codecNotes], error) { + return typedTestMerger[codecNotes]{key: "typed/shared"}, nil + }); err != nil { + t.Fatalf("RegisterMerger(notes): %v", err) + } + if err := RegisterNormalizer(normalizers, normalizeSpec, func() (contracts.Normalizer[codecNotes], error) { + return typedTestNormalizer[codecNotes]{key: "typed/shared"}, nil + }); err != nil { + t.Fatalf("RegisterNormalizer(notes): %v", err) + } + case "test/score": + if err := RegisterMerger(mergers, mergeSpec, func() (contracts.Merger[codecScore], error) { + return typedTestMerger[codecScore]{key: "typed/shared"}, nil + }); err != nil { + t.Fatalf("RegisterMerger(score): %v", err) + } + if err := RegisterNormalizer(normalizers, normalizeSpec, func() (contracts.Normalizer[codecScore], error) { + return typedTestNormalizer[codecScore]{key: "typed/shared"}, nil + }); err != nil { + t.Fatalf("RegisterNormalizer(score): %v", err) + } + } + } + + for name, lookup := range map[string]func(contracts.ArtifactKind) (ModuleSpec, bool){ + "merger": func(kind contracts.ArtifactKind) (ModuleSpec, bool) { + return mergers.SpecForArtifact(" typed/shared ", kind) + }, + "normalizer": func(kind contracts.ArtifactKind) (ModuleSpec, bool) { + return normalizers.SpecForArtifact(" typed/shared ", kind) + }, + } { + t.Run(name, func(t *testing.T) { + notes, ok := lookup(" test/notes ") + if !ok || len(notes.ReferenceSlots) != 1 || notes.ReferenceSlots[0].Name != "note_notes" { + t.Fatalf("notes spec = %#v, ok = %v", notes, ok) + } + score, ok := lookup("test/score") + if !ok || len(score.ReferenceSlots) != 1 || score.ReferenceSlots[0].Name != "score_notes" { + t.Fatalf("score spec = %#v, ok = %v", score, ok) + } + notes.ReferenceSlots[0].Name = "mutated" + again, _ := lookup("test/notes") + if again.ReferenceSlots[0].Name != "note_notes" { + t.Fatalf("registry spec changed after caller mutation: %#v", again) + } + }) + } + + for name, lookup := range map[string]func() (ModuleSpec, bool){ + "merger": func() (ModuleSpec, bool) { return mergers.Spec("typed/shared") }, + "normalizer": func() (ModuleSpec, bool) { return normalizers.Spec("typed/shared") }, + } { + t.Run(name+" catalog", func(t *testing.T) { + for attempt := 0; attempt < 10; attempt++ { + spec, ok := lookup() + if !ok || spec.ArtifactKind != "test/notes" || spec.ReferenceSlots[0].Name != "note_notes" { + t.Fatalf("Spec() = %#v, ok = %v; want lexicographically first artifact kind", spec, ok) + } + } + }) + } + + if got := mergers.RegisteredArtifactKinds("typed/shared"); !reflect.DeepEqual(got, []contracts.ArtifactKind{"test/notes", "test/score"}) { + t.Fatalf("merger kinds = %#v", got) + } + if got := normalizers.RegisteredArtifactKinds("typed/shared"); !reflect.DeepEqual(got, []contracts.ArtifactKind{"test/notes", "test/score"}) { + t.Fatalf("normalizer kinds = %#v", got) + } + }) + } +} + func TestResolvedPipelineDigestIncludesArtifactSchemaIdentity(t *testing.T) { baseOptions := completeTypedCatalogOptions() base, err := ResolvePipeline(typedResolutionProfile(), ResolveOptions{}, typedResolutionCatalog(t, baseOptions))